Skip to content

Commit 83256b5

Browse files
committed
fix(corrosion): handle 404 errors during resubscription and stop retrying
1 parent 35191c0 commit 83256b5

2 files changed

Lines changed: 75 additions & 0 deletions

File tree

‎internal/corrosion/subscribe.go‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,10 @@ var (
2222
ChangeTypeDelete ChangeType = "delete"
2323
)
2424

25+
// ErrSubscriptionNotFound is returned when resubscribing to a subscription that Corrosion
26+
// no longer knows about (HTTP 404).
27+
var ErrSubscriptionNotFound = errors.New("subscription not found")
28+
2529
type ChangeEvent struct {
2630
Type ChangeType
2731
RowID uint64
@@ -289,6 +293,13 @@ func (c *APIClient) resubscribeWithBackoffFn(id string) func(context.Context, ui
289293
return backoff.RetryWithData(func() (*Subscription, error) {
290294
sub, err := c.ResubscribeContext(ctx, id, fromChange)
291295
if err != nil {
296+
// A gone subscription can never be resubscribed, so stop retrying immediately and let the caller
297+
// recover by creating a fresh subscription.
298+
if errors.Is(err, ErrSubscriptionNotFound) {
299+
slog.Debug("Corrosion subscription no longer exists, giving up resubscribing.",
300+
"id", id, "from_change", fromChange)
301+
return nil, backoff.Permanent(fmt.Errorf("resubscribe to %s: %w", id, err))
302+
}
292303
slog.Debug("Failed to resubscribe to Corrosion query. Retrying with backoff.",
293304
"id", id, "from_change", fromChange, "err", err)
294305
}
@@ -316,11 +327,17 @@ func (c *APIClient) ResubscribeContext(ctx context.Context, id string, fromChang
316327
}
317328

318329
if resp.StatusCode != http.StatusOK {
330+
if resp.StatusCode == http.StatusNotFound {
331+
resp.Body.Close()
332+
return nil, ErrSubscriptionNotFound
333+
}
334+
319335
respBody, err := io.ReadAll(resp.Body)
320336
resp.Body.Close()
321337
if err != nil {
322338
return nil, fmt.Errorf("read response body: %w", err)
323339
}
340+
324341
return nil, fmt.Errorf("unexpected status code %d: %s", resp.StatusCode, respBody)
325342
}
326343

‎internal/corrosion/subscribe_test.go‎

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ import (
77
"net/http/httptest"
88
"net/url"
99
"strings"
10+
"sync/atomic"
1011
"testing"
1112
"time"
1213

@@ -156,3 +157,60 @@ func TestSubscription_ResubscribeFromNonZeroSkipsSnapshot(t *testing.T) {
156157
require.FailNow(t, "timed out waiting for a change after resubscribe", sub.Err())
157158
}
158159
}
160+
161+
// TestSubscription_ResubscribeNotFoundFailsFast verifies that when Corrosion returns 404 to a resubscription
162+
// (the subscription no longer exists, e.g. after a restart that dropped it), the change handler stops retrying
163+
// immediately and closes the changes channel with ErrSubscriptionNotFound, instead of retrying for the full
164+
// backoff window.
165+
func TestSubscription_ResubscribeNotFoundFailsFast(t *testing.T) {
166+
t.Parallel()
167+
168+
var getCount atomic.Int32
169+
170+
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
171+
switch r.Method {
172+
case http.MethodPost:
173+
// Initial subscription, then end the stream to force a resubscribe.
174+
w.Header().Set("corro-query-id", "test-sub")
175+
w.WriteHeader(http.StatusOK)
176+
flushString(t, w, `{"columns":["id","info"]}`+"\n"+
177+
`{"row":[1,["a","b"]]}`+"\n"+
178+
`{"eoq":{"time":1e-7,"change_id":5}}`+"\n")
179+
case http.MethodGet:
180+
// Resubscription: Corrosion no longer knows this subscription.
181+
getCount.Add(1)
182+
http.Error(w, "", http.StatusNotFound)
183+
default:
184+
require.FailNow(t, fmt.Sprintf("unexpected request method: %s", r.Method))
185+
}
186+
}))
187+
defer srv.Close()
188+
189+
client := newTestClient(t, srv)
190+
191+
ctx, cancel := context.WithCancel(context.Background())
192+
defer cancel()
193+
194+
sub, err := client.SubscribeContext(ctx, "SELECT id, info FROM machines", nil, false)
195+
require.NoError(t, err)
196+
197+
rows := sub.Rows()
198+
for rows.Next() {
199+
}
200+
require.NoError(t, rows.Err())
201+
202+
changes, err := sub.Changes()
203+
require.NoError(t, err)
204+
205+
select {
206+
case change := <-changes:
207+
// The channel must close (nil change) rather than deliver anything.
208+
require.Nil(t, change, "expected the changes channel to close on a gone subscription")
209+
case <-time.After(5 * time.Second):
210+
require.FailNow(t, "timed out waiting for the changes channel to close")
211+
}
212+
213+
require.ErrorIs(t, sub.Err(), ErrSubscriptionNotFound)
214+
// The 404 must not be retried: exactly one GET resubscription request was made.
215+
assert.Equal(t, int32(1), getCount.Load())
216+
}

0 commit comments

Comments
 (0)