|
7 | 7 | "net/http/httptest" |
8 | 8 | "net/url" |
9 | 9 | "strings" |
| 10 | + "sync/atomic" |
10 | 11 | "testing" |
11 | 12 | "time" |
12 | 13 |
|
@@ -156,3 +157,60 @@ func TestSubscription_ResubscribeFromNonZeroSkipsSnapshot(t *testing.T) { |
156 | 157 | require.FailNow(t, "timed out waiting for a change after resubscribe", sub.Err()) |
157 | 158 | } |
158 | 159 | } |
| 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