diff --git a/internal/microcosm/replay.go b/internal/microcosm/replay.go index bb430c6..3bb4e20 100644 --- a/internal/microcosm/replay.go +++ b/internal/microcosm/replay.go @@ -107,7 +107,8 @@ func (r *Replay) ConsumeReplay(ctx context.Context, cursor int64, wanted func(co } httpClient := *r.client.http httpClient.Transport = &replayTransport{base: httpClient.Transport, observe: observe} - options := []jetstream.Option{jetstream.WithAPIKey(r.key), jetstream.WithAfterSeq(uint64(cursor)), jetstream.WithCollections(collections), jetstream.WithHTTPClient(&httpClient), jetstream.WithBatchSize(replayBatchSize), jetstream.WithDownloadConcurrency(r.workers), jetstream.WithSegmentStripes(1), jetstream.WithMaxDownloadAttempts(upstream.MaxAttempts)} + // Keep the complete default retry policy. The SDK's attempt override omits required backoff fields and panics on retries. + options := []jetstream.Option{jetstream.WithAPIKey(r.key), jetstream.WithAfterSeq(uint64(cursor)), jetstream.WithCollections(collections), jetstream.WithHTTPClient(&httpClient), jetstream.WithBatchSize(replayBatchSize), jetstream.WithDownloadConcurrency(r.workers), jetstream.WithSegmentStripes(1)} if r.rangeSeq.Before > 0 { options = append(options, jetstream.WithBeforeSeq(uint64(r.rangeSeq.Before)), jetstream.WithSnapshotOnly()) } diff --git a/internal/microcosm/replay_test.go b/internal/microcosm/replay_test.go index 1f95faf..621624b 100644 --- a/internal/microcosm/replay_test.go +++ b/internal/microcosm/replay_test.go @@ -1,6 +1,7 @@ package microcosm import ( + "bytes" "context" "encoding/json" "errors" @@ -127,6 +128,48 @@ func TestReplayAuthentication(t *testing.T) { } } +func TestReplayTransientBlockRetry(t *testing.T) { + archive := archiveFixture(t) + reader, err := segment.OpenReaderAt(bytes.NewReader(archive), int64(len(archive)), segment.ReaderOptions{}) + if err != nil { + t.Fatal(err) + } + defer reader.Close() + const blockLengthPrefixBytes = 8 + info := reader.Blocks()[0] + start := info.Offset + blockLengthPrefixBytes + frame := archive[start : start+uint64(info.CompressedSize)] + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + queue, err := upstream.New(ctx, 1) + if err != nil { + t.Fatal(err) + } + defer queue.Close() + var blocks atomic.Int64 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/xrpc/network.bsky.jetstream.planSnapshot" { + w.Write([]byte(`{"sealedTipSeq":1,"plannedThroughSeq":1,"segments":[{"name":"seg_0000000000.jss","index":0,"minSeq":1,"maxSeq":1,"mode":"blocks","blocks":[{"first":0,"last":0}]}],"stats":{}}`)) + return + } + if blocks.Add(1) == 1 { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + w.Write(frame) + })) + defer server.Close() + replay, err := NewReplayWithRange(server.URL, "test-key", domain.SoftDisabled, queue, 1, domain.SequenceRange{Before: 1}) + if err != nil { + t.Fatal(err) + } + var events int + err = replay.ConsumeReplay(ctx, 0, nil, func(_ context.Context, batch []domain.Change, _ int64) error { events += len(batch); return nil }, nil) + if err != nil || events != 1 || blocks.Load() != 2 { + t.Fatalf("transient replay: events %d blocks %d error %v", events, blocks.Load(), err) + } +} + func TestReplayArchiveRedirect(t *testing.T) { archive := archiveFixture(t) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)