diff --git a/internal/corrosion/subscribe.go b/internal/corrosion/subscribe.go index 2e317cb7..c29f6876 100644 --- a/internal/corrosion/subscribe.go +++ b/internal/corrosion/subscribe.go @@ -22,6 +22,10 @@ var ( ChangeTypeDelete ChangeType = "delete" ) +// ErrSubscriptionNotFound is returned when resubscribing to a subscription that Corrosion +// no longer knows about (HTTP 404). +var ErrSubscriptionNotFound = errors.New("subscription not found") + type ChangeEvent struct { Type ChangeType RowID uint64 @@ -289,6 +293,13 @@ func (c *APIClient) resubscribeWithBackoffFn(id string) func(context.Context, ui return backoff.RetryWithData(func() (*Subscription, error) { sub, err := c.ResubscribeContext(ctx, id, fromChange) if err != nil { + // A gone subscription can never be resubscribed, so stop retrying immediately and let the caller + // recover by creating a fresh subscription. + if errors.Is(err, ErrSubscriptionNotFound) { + slog.Debug("Corrosion subscription no longer exists, giving up resubscribing.", + "id", id, "from_change", fromChange) + return nil, backoff.Permanent(fmt.Errorf("resubscribe to %s: %w", id, err)) + } slog.Debug("Failed to resubscribe to Corrosion query. Retrying with backoff.", "id", id, "from_change", fromChange, "err", err) } @@ -316,11 +327,17 @@ func (c *APIClient) ResubscribeContext(ctx context.Context, id string, fromChang } if resp.StatusCode != http.StatusOK { + if resp.StatusCode == http.StatusNotFound { + resp.Body.Close() + return nil, ErrSubscriptionNotFound + } + respBody, err := io.ReadAll(resp.Body) resp.Body.Close() if err != nil { return nil, fmt.Errorf("read response body: %w", err) } + return nil, fmt.Errorf("unexpected status code %d: %s", resp.StatusCode, respBody) } diff --git a/internal/corrosion/subscribe_test.go b/internal/corrosion/subscribe_test.go index 3d11c841..74dd1fb1 100644 --- a/internal/corrosion/subscribe_test.go +++ b/internal/corrosion/subscribe_test.go @@ -7,6 +7,7 @@ import ( "net/http/httptest" "net/url" "strings" + "sync/atomic" "testing" "time" @@ -156,3 +157,60 @@ func TestSubscription_ResubscribeFromNonZeroSkipsSnapshot(t *testing.T) { require.FailNow(t, "timed out waiting for a change after resubscribe", sub.Err()) } } + +// TestSubscription_ResubscribeNotFoundFailsFast verifies that when Corrosion returns 404 to a resubscription +// (the subscription no longer exists, e.g. after a restart that dropped it), the change handler stops retrying +// immediately and closes the changes channel with ErrSubscriptionNotFound, instead of retrying for the full +// backoff window. +func TestSubscription_ResubscribeNotFoundFailsFast(t *testing.T) { + t.Parallel() + + var getCount atomic.Int32 + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.Method { + case http.MethodPost: + // Initial subscription, then end the stream to force a resubscribe. + w.Header().Set("corro-query-id", "test-sub") + w.WriteHeader(http.StatusOK) + flushString(t, w, `{"columns":["id","info"]}`+"\n"+ + `{"row":[1,["a","b"]]}`+"\n"+ + `{"eoq":{"time":1e-7,"change_id":5}}`+"\n") + case http.MethodGet: + // Resubscription: Corrosion no longer knows this subscription. + getCount.Add(1) + http.Error(w, "", http.StatusNotFound) + default: + require.FailNow(t, fmt.Sprintf("unexpected request method: %s", r.Method)) + } + })) + defer srv.Close() + + client := newTestClient(t, srv) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + sub, err := client.SubscribeContext(ctx, "SELECT id, info FROM machines", nil, false) + require.NoError(t, err) + + rows := sub.Rows() + for rows.Next() { + } + require.NoError(t, rows.Err()) + + changes, err := sub.Changes() + require.NoError(t, err) + + select { + case change := <-changes: + // The channel must close (nil change) rather than deliver anything. + require.Nil(t, change, "expected the changes channel to close on a gone subscription") + case <-time.After(5 * time.Second): + require.FailNow(t, "timed out waiting for the changes channel to close") + } + + require.ErrorIs(t, sub.Err(), ErrSubscriptionNotFound) + // The 404 must not be retried: exactly one GET resubscription request was made. + assert.Equal(t, int32(1), getCount.Load()) +}