diff --git a/docs/jetstream-replay.md b/docs/jetstream-replay.md index b9a1588..d5523da 100644 --- a/docs/jetstream-replay.md +++ b/docs/jetstream-replay.md @@ -130,6 +130,8 @@ Plans larger than the event-size limit still update the archive tip. Observation failures do not change the response that the SDK receives. Authentication, malformed events, storage errors, and unclassified upstream errors stop replay without a checkpoint advance for the failed batch. +The adapter logs SDK warnings for non-object record payloads and continues with valid records. +Those records do not enter the index. Other decode errors and mixed error groups still stop replay. Transient errors retain reconnect backoff, jitter, and shared cooldowns. To recover from a terminal error, repair its cause and restart the service. The restart uses the durable checkpoint. No endpoint skips a failed event. diff --git a/internal/microcosm/replay.go b/internal/microcosm/replay.go index 938d701..8c095d0 100644 --- a/internal/microcosm/replay.go +++ b/internal/microcosm/replay.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "io" + "log/slog" "math" "net/http" "net/url" @@ -16,6 +17,7 @@ import ( "provenance/internal/atproto" "provenance/internal/domain" + "provenance/internal/logging" "provenance/internal/upstream" ) @@ -121,6 +123,10 @@ func (r *Replay) ConsumeReplay(ctx context.Context, cursor int64, wanted func(co defer client.Close() for batch, err := range client.Events(ctx) { if err != nil { + if recordDecodeError(err) { + slog.Warn("malformed archive record skipped", "component", "replay", "source", r.Source(), "error", logging.Error(err)) + continue + } return replayFailure(err) } if batch.LastCursor() > math.MaxInt64 { diff --git a/internal/microcosm/replay_bad_record_test.go b/internal/microcosm/replay_bad_record_test.go new file mode 100644 index 0000000..c05bef2 --- /dev/null +++ b/internal/microcosm/replay_bad_record_test.go @@ -0,0 +1,128 @@ +package microcosm + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "testing" + "time" + + "github.com/bluesky-social/jetstream/segment" + + "provenance/internal/domain" + "provenance/internal/upstream" +) + +func TestReplayMalformedRecord(t *testing.T) { + for _, tail := range []segment.Kind{segment.KindDelete, segment.KindCreate} { + t.Run(fmt.Sprint(tail), func(t *testing.T) { + path := filepath.Join(t.TempDir(), "archive.jss") + writer, err := segment.New(segment.Config{Path: path}) + if err != nil { + t.Fatal(err) + } + defer writer.Close() + + // Separate blocks cover valid rows after a bad record and a malformed tail. + for i, kind := range []segment.Kind{segment.KindDelete, segment.KindCreate, tail} { + _, err := writer.Append(segment.Event{ + Seq: uint64(i + 1), WitnessedAt: int64(i + 100), Kind: kind, + DID: author, Collection: domain.BlockCollection, Rkey: "a", + Payload: []byte{0xf6}, + }) + if err != nil { + t.Fatal(err) + } + if err := writer.Flush(); err != nil { + t.Fatal(err) + } + } + if _, err := writer.Seal(); err != nil { + t.Fatal(err) + } + archive, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/xrpc/network.bsky.jetstream.planSnapshot": + json.NewEncoder(w).Encode(map[string]any{ + "sealedTipSeq": 3, "plannedThroughSeq": 3, + "segments": []map[string]any{{"name": "seg_0000000000.jss", "index": 0, "minSeq": 1, "maxSeq": 3, "mode": "segment"}}, + "stats": map[string]any{}, + }) + case "/xrpc/network.bsky.jetstream.getSegment": + w.Write(archive) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + 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() + replay, err := NewReplayWithRange(server.URL, "test-key", domain.SoftAllAuthors, queue, 1, domain.SequenceRange{Before: 3}) + if err != nil { + t.Fatal(err) + } + var changes []domain.Change + var checkpoint int64 + err = replay.ConsumeReplay(ctx, 0, nil, func(_ context.Context, batch []domain.Change, cursor int64) error { + changes = append(changes, batch...) + checkpoint = cursor + return nil + }, nil) + if err != nil { + t.Fatalf("malformed record stopped replay: %v", err) + } + want := 1 + if tail == segment.KindDelete { + want = 2 + } + if len(changes) != want || changes[0].Sequence != 1 || checkpoint != 3 { + t.Fatalf("changes %v, checkpoint %d", changes, checkpoint) + } + if want == 2 && changes[1].Sequence != 3 { + t.Fatalf("lost valid row after malformed record: %v", changes) + } + }) + } +} + +func TestRecordDecodeError(t *testing.T) { + badRecord := fmt.Errorf("jetstream: decode record (did=%s collection=%s rkey=a seq=2): %w", author, domain.BlockCollection, errors.New("cbor decode: record is not an object")) + badBlock := fmt.Errorf("jetstream: decode block 0 of archive: %w", errors.New("cbor decode: invalid frame")) + badCBOR := fmt.Errorf("jetstream: decode record (did=%s collection=%s rkey=a seq=2): %w", author, domain.BlockCollection, errors.New("cbor decode: unexpected EOF")) + for _, tc := range []struct { + name string + err error + want bool + }{ + {"record", badRecord, true}, + {"joined records", errors.Join(badRecord, badRecord), true}, + {"wrapped records", fmt.Errorf("archive: %w", errors.Join(badRecord)), true}, + {"block", badBlock, false}, + {"record corruption", badCBOR, false}, + {"mixed", errors.Join(badRecord, badBlock), false}, + {"transport", errors.Join(badRecord, context.DeadlineExceeded), false}, + {"unknown", errors.New("unknown archive failure"), false}, + {"untyped text", errors.New(badRecord.Error()), false}, + {"nil", nil, false}, + } { + t.Run(tc.name, func(t *testing.T) { + if got := recordDecodeError(tc.err); got != tc.want { + t.Fatalf("record warning = %t, want %t: %v", got, tc.want, tc.err) + } + }) + } +} diff --git a/internal/microcosm/replay_status.go b/internal/microcosm/replay_status.go index b536cbe..f844039 100644 --- a/internal/microcosm/replay_status.go +++ b/internal/microcosm/replay_status.go @@ -17,6 +17,36 @@ import ( ) const replayRedirectLimit = 10 +const recordDecodePrefix = "jetstream: decode record (" +const nonObjectRecordError = "cbor decode: record is not an object" + +// Continue only for SDK record warnings. Block, transport, and mixed errors remain fatal. +func recordDecodeError(err error) bool { + if err == nil { + return false + } + if joined, ok := err.(interface{ Unwrap() []error }); ok { + children := joined.Unwrap() + if len(children) == 0 { + return false + } + for _, child := range children { + if !recordDecodeError(child) { + return false + } + } + return true + } + child := errors.Unwrap(err) + if child == nil { + return false + } + // The SDK has no typed record error. Match its wrapper and CBOR cause together. + if strings.HasPrefix(err.Error(), recordDecodePrefix) && child.Error() == nonObjectRecordError { + return true + } + return recordDecodeError(child) +} // Follow archive download redirects without forwarding the key to another origin. func replayRedirect(req *http.Request, via []*http.Request) error {