From 761f93fc0cc12e6a39be99be6a9b505c715a7e00 Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Tue, 14 Jul 2026 19:19:26 -0700 Subject: [PATCH 1/3] atproto: actually resume from where you left off the cursor --- go.mod | 2 +- go.sum | 14 +----- pkg/atproto/firehose.go | 40 +++++++++++++++- pkg/atproto/firehose_cursor.go | 67 +++++++++++++++++++-------- pkg/atproto/firehose_cursor_test.go | 71 +++++++++++++++++++++++++++++ pkg/atproto/firehose_moq.go | 36 +++++++++++++-- pkg/atproto/firehose_moq_test.go | 2 +- 7 files changed, 193 insertions(+), 39 deletions(-) diff --git a/go.mod b/go.mod index 92a6a48f..0254e7e5 100644 --- a/go.mod +++ b/go.mod @@ -67,7 +67,7 @@ require ( github.com/skip2/go-qrcode v0.0.0-20200617195104-da1b6568686e github.com/slok/go-http-metrics v0.13.0 github.com/starttoaster/prometheus-exporter-scraper v0.0.1 - github.com/streamplace/atmoq-go v0.0.2 + github.com/streamplace/atmoq/go v0.0.4-0.20260701223355-13757de4ae08 github.com/streamplace/atproto-oauth-golang v0.0.0-20260413212710-98956064d06c github.com/streamplace/glex v0.0.0-20260714202640-f3b7557a2bdf github.com/streamplace/muxl/go v0.3.4 diff --git a/go.sum b/go.sum index 02386d8a..a023beed 100644 --- a/go.sum +++ b/go.sum @@ -1368,20 +1368,10 @@ github.com/starttoaster/prometheus-exporter-scraper v0.0.1 h1:E3Zl3ho3v3Yt2Yk7i9 github.com/starttoaster/prometheus-exporter-scraper v0.0.1/go.mod h1:qNccVOSo0np2jYi6DolRyUICFcMkhH5k1Gs7CYxav1w= github.com/stbenjam/no-sprintf-host-port v0.2.0 h1:i8pxvGrt1+4G0czLr/WnmyH7zbZ8Bg8etvARQ1rpyl4= github.com/stbenjam/no-sprintf-host-port v0.2.0/go.mod h1:eL0bQ9PasS0hsyTyfTjjG+E80QIyPnBVQbYZyv20Jfk= -github.com/streamplace/atmoq-go v0.0.2 h1:lLfJ9R88wzkugqMMrRIcy2Sz4s3fp6TLesf3iwG4/Vo= -github.com/streamplace/atmoq-go v0.0.2/go.mod h1:r4KV7lW5KWfAJa2OgNXA4fgDHHcF3pNa9vGEUj9z4e0= +github.com/streamplace/atmoq/go v0.0.4-0.20260701223355-13757de4ae08 h1:NiTRz8AXme+VPAVOuw3h8g3Ya0O0he+F02rSERU6FsY= +github.com/streamplace/atmoq/go v0.0.4-0.20260701223355-13757de4ae08/go.mod h1:3P8eSwKAGH7uh3SX5z1jlt/JgPTilJTUZngQJKhWY5s= github.com/streamplace/atproto-oauth-golang v0.0.0-20260413212710-98956064d06c h1:IzEPU2O4iL58Nb7aw+7lB9ttnesEwOVVE5oV9NEXemM= github.com/streamplace/atproto-oauth-golang v0.0.0-20260413212710-98956064d06c/go.mod h1:9LlKkqciiO5lRfbX0n4Wn5KNY9nvFb4R3by8FdW2TWc= -github.com/streamplace/atproto-oauth-golang v0.0.0-20250619231223-a9c04fb888ac h1:heVM4CGox3kfJclSmagsI3hvKk7W52EegnFqqs9CSYk= -github.com/streamplace/atproto-oauth-golang v0.0.0-20250619231223-a9c04fb888ac/go.mod h1:9LlKkqciiO5lRfbX0n4Wn5KNY9nvFb4R3by8FdW2TWc= -github.com/streamplace/glex v0.0.0-20260712184244-6c74782635ae h1:raCkXf/Wz4T19du2H0zLz+GLk2/efSgUWPaS5iEL7q8= -github.com/streamplace/glex v0.0.0-20260712184244-6c74782635ae/go.mod h1:LRaoeSMvSgOrhFX8s7ygjRlyka7wXdDa1s7JJ9o1IzY= -github.com/streamplace/glex v0.0.0-20260712200712-847c1b5963c1 h1:beael8HJivy8Cs++X7gzZS/71LTeEJr5Qzu3OQxaFDM= -github.com/streamplace/glex v0.0.0-20260712200712-847c1b5963c1/go.mod h1:LRaoeSMvSgOrhFX8s7ygjRlyka7wXdDa1s7JJ9o1IzY= -github.com/streamplace/glex v0.0.0-20260712210145-b0354b3b167c h1:OBII3YaKbiCN42YZBbUIT7fAnq5HJUrpO7dO9DvAZD8= -github.com/streamplace/glex v0.0.0-20260712210145-b0354b3b167c/go.mod h1:LRaoeSMvSgOrhFX8s7ygjRlyka7wXdDa1s7JJ9o1IzY= -github.com/streamplace/glex v0.0.0-20260713025044-48835ef7879b h1:cp+BGMUJEHK9srFoEaqXCv2Q030bnIjxDQUt9Mhb9SA= -github.com/streamplace/glex v0.0.0-20260713025044-48835ef7879b/go.mod h1:LRaoeSMvSgOrhFX8s7ygjRlyka7wXdDa1s7JJ9o1IzY= github.com/streamplace/glex v0.0.0-20260714202640-f3b7557a2bdf h1:KQLi0Ht1trBdr1lo+4VD4bsk7NbTGtjwZr7k9gME3zE= github.com/streamplace/glex v0.0.0-20260714202640-f3b7557a2bdf/go.mod h1:LRaoeSMvSgOrhFX8s7ygjRlyka7wXdDa1s7JJ9o1IzY= github.com/streamplace/go-dpop v0.0.0-20250510031900-c897158a8ad4 h1:L1fS4HJSaAyNnkwfuZubgfeZy8rkWmA0cMtH5Z0HqNc= diff --git a/pkg/atproto/firehose.go b/pkg/atproto/firehose.go index 2479d281..eafcbe59 100644 --- a/pkg/atproto/firehose.go +++ b/pkg/atproto/firehose.go @@ -57,6 +57,8 @@ type ATProtoSynchronizer struct { // (unix nanos). lastSeen atomic.Int64 lastEvent atomic.Int64 + // events seen across all relays, pre-dedup (for the periodic in-sync ping). + seenEvents atomic.Int64 // cross-relay dedup, shared by every relay consumer. Initialized at the // top of StartFirehose. @@ -66,6 +68,7 @@ type ATProtoSynchronizer struct { func (atsync *ATProtoSynchronizer) markSeen() { atsync.lastSeen.Store(time.Now().UnixNano()) + atsync.seenEvents.Add(1) } func (atsync *ATProtoSynchronizer) markEvent(t time.Time) { @@ -311,7 +314,20 @@ func (atsync *ATProtoSynchronizer) repoStreamCallbacks(ctx context.Context, rela // observeSeq advances the cursor and republishes the per-relay high-water // seq gauge; both relays carry the upstream's seq, so the gauge's cross-relay // difference measures how far apart the relays are. + // + // On a moq relay only the gauge gets the upstream seq: there the resume + // cursor holds MoQ group sequences (observeGroup in connectRelayMoq), and + // upstream at-seqs are orders of magnitude larger, so feeding them into the + // same max-wins cursor would bury the group cursor — the next SubscribeFrom + // would then wait forever on a group the relay will never produce. + moq := protocol == "moq" + var gaugeHigh highWater // upstream-seq high-water for the moq gauge only observeSeq := func(seq int64) { + if moq { + gaugeHigh.observe(seq) + spmetrics.FirehoseRelayHighSeq.WithLabelValues(relay, protocol).Set(float64(gaugeHigh.get())) + return + } cursor.observe(seq) spmetrics.FirehoseRelayHighSeq.WithLabelValues(relay, protocol).Set(float64(cursor.highSeq())) } @@ -359,9 +375,16 @@ func identityDedupKey(evt *indigoatproto.SyncSubscribeRepos_Identity) string { return evt.Did + "\x00" + evt.Time + "\x00" + handle } +// firehoseSyncPingInterval is how often monitorFirehose emits its Info-level +// "in sync" heartbeat while healthy. Trouble is still reported within the 5s +// tick; this only spaces out the all-is-well line so it doesn't spam. +const firehoseSyncPingInterval = 60 * time.Second + func (atsync *ATProtoSynchronizer) monitorFirehose(ctx context.Context) { ticker := time.NewTicker(5 * time.Second) defer ticker.Stop() + lastPing := time.Now() + lastPingSeen := atsync.seenEvents.Load() for { select { case <-ctx.Done(): @@ -374,9 +397,23 @@ func (atsync *ATProtoSynchronizer) monitorFirehose(ctx context.Context) { } else { log.Debug(ctx, fmt.Sprintf("firehose is %s behind real time", since), "goroutines", goroutines) } - if dry := sinceNanos(atsync.lastSeen.Load()); dry > 10*time.Second { + dry := sinceNanos(atsync.lastSeen.Load()) + if dry > 10*time.Second { log.Warn(ctx, fmt.Sprintf("firehose dry; no new events for %s", dry)) } + // Periodic in-sync ping: healthy operation is otherwise Debug-only, + // so surface a heartbeat at Info once a minute. An unhealthy stretch + // (warns above) just delays the next ping; the ping never lies. + if since <= 10*time.Second && dry <= 10*time.Second && + time.Since(lastPing) >= firehoseSyncPingInterval { + seen := atsync.seenEvents.Load() + log.Log(ctx, "firehose in sync", + "behind", since, + "events", seen-lastPingSeen, + "goroutines", goroutines) + lastPing = time.Now() + lastPingSeen = seen + } } } } @@ -411,6 +448,7 @@ func (atsync *ATProtoSynchronizer) handleCommitEventOps(ctx context.Context, evt return } ctx := log.WithLogValues(ctx, "eventKind", op.Action, "collection", collection.String(), "rkey", rkey.String()) + log.Warn(ctx, "got event") if len(CollectionFilter) > 0 { keep := slices.Contains(CollectionFilter, collection.String()) diff --git a/pkg/atproto/firehose_cursor.go b/pkg/atproto/firehose_cursor.go index d76a82a7..1cd3d347 100644 --- a/pkg/atproto/firehose_cursor.go +++ b/pkg/atproto/firehose_cursor.go @@ -15,6 +15,28 @@ import ( // in-process reconnects. const cursorFlushInterval = 5 * time.Second +// highWater is a concurrency-safe monotonically-increasing maximum. +type highWater struct{ v atomic.Int64 } + +// observe raises the mark to seq if it is higher. Safe for concurrent callers. +func (h *highWater) observe(seq int64) { + for { + cur := h.v.Load() + if seq <= cur { + return + } + if h.v.CompareAndSwap(cur, seq) { + return + } + } +} + +func (h *highWater) get() int64 { return h.v.Load() } + +// set overwrites the mark unconditionally, including downward. Only for +// initialization and reset; everything else goes through observe. +func (h *highWater) set(seq int64) { h.v.Store(seq) } + // relayCursor tracks how far we've consumed one relay's firehose so we can // resume after a disconnect or restart instead of re-tailing from live (which // would leave a gap). It keeps the high-water sequence number in memory, @@ -32,12 +54,16 @@ type relayCursor struct { // latest is the high-water cursor: the upstream at-sequence for a WebSocket // relay, or the high-water MoQ group sequence for a moqt:// relay (used to // resume replay via SubscribeFrom — see connectRelayMoq). A host is one - // transport or the other, so a single value covers both. 0 = nothing seen - // yet (tail from live). Persisted periodically so a Streamplace restart - // resumes from here — the relay assigns durable ids across its own restarts, - // so a stored cursor stays valid (it just ages out of the relay's replay - // window if we are down too long, which is the gap PDS re-sync covers). - latest atomic.Int64 + // transport or the other, so a single value covers both — but the two + // sequence spaces must never mix: upstream at-seqs are orders of magnitude + // larger than group sequences, so on a moq host only observeGroup may feed + // this (repoStreamCallbacks keeps the upstream seq out of it and in the + // metrics gauge only). 0 = nothing seen yet (tail from live). Persisted + // periodically so a Streamplace restart resumes from here — the relay + // assigns durable ids across its own restarts, so a stored cursor stays + // valid (it just ages out of the relay's replay window if we are down too + // long, which is the gap PDS re-sync covers). + latest highWater flushed int64 // last persisted value; only the flush loop touches it } @@ -49,7 +75,7 @@ func (atsync *ATProtoSynchronizer) newRelayCursor(ctx context.Context, host stri return rc } if stored != nil { - rc.latest.Store(stored.Cursor) + rc.latest.set(stored.Cursor) rc.flushed = stored.Cursor log.Log(ctx, "resuming relay from stored cursor", "cursor", stored.Cursor) } @@ -59,27 +85,19 @@ func (atsync *ATProtoSynchronizer) newRelayCursor(ctx context.Context, host stri // observe advances the high-water mark. Safe for concurrent callers (the // scheduler runs several event workers). func (rc *relayCursor) observe(seq int64) { - for { - cur := rc.latest.Load() - if seq <= cur { - return - } - if rc.latest.CompareAndSwap(cur, seq) { - return - } - } + rc.latest.observe(seq) } // param returns the cursor to dial with and whether to send one at all. With no // progress yet we send none, so a fresh external relay tails from live instead // of backfilling its entire history. func (rc *relayCursor) param() (int64, bool) { - v := rc.latest.Load() + v := rc.latest.get() return v, v > 0 } // highSeq returns the high-water upstream sequence number observed so far. -func (rc *relayCursor) highSeq() int64 { return rc.latest.Load() } +func (rc *relayCursor) highSeq() int64 { return rc.latest.get() } // observeGroup advances the high-water cursor from a MoQ group sequence (the // moqt:// transport's flavour of a cursor). Called on every frame received from @@ -96,17 +114,26 @@ func (rc *relayCursor) observeGroup(seq uint64) { // and self-healing, and in practice relay group ids are large (seeded for // durability across restarts). func (rc *relayCursor) groupStart() (uint64, bool) { - v := rc.latest.Load() + v := rc.latest.get() if v <= 0 { return 0, false } return uint64(v), true } +// reset abandons the cursor so the next connect tails the live edge. For when +// a resume proves the stored value can't be trusted — e.g. a row poisoned by +// upstream at-seqs before the group/at-seq split in repoStreamCallbacks — +// which observe()'s max-wins semantics could otherwise never walk back. The +// flush loop persists the zero, healing the stored row too. +func (rc *relayCursor) reset() { + rc.latest.set(0) +} + // flush persists the high-water mark if it has advanced since the last write. // Only ever called from the single flush goroutine, so flushed is unsynchronized. func (rc *relayCursor) flush(ctx context.Context) { - v := rc.latest.Load() + v := rc.latest.get() if v == rc.flushed { return } diff --git a/pkg/atproto/firehose_cursor_test.go b/pkg/atproto/firehose_cursor_test.go index 2cf7b8ab..20e4ea61 100644 --- a/pkg/atproto/firehose_cursor_test.go +++ b/pkg/atproto/firehose_cursor_test.go @@ -4,6 +4,8 @@ import ( "context" "testing" + indigoatproto "github.com/bluesky-social/indigo/api/atproto" + lexutil "github.com/bluesky-social/indigo/lex/util" "github.com/stretchr/testify/require" "stream.place/streamplace/pkg/model" ) @@ -95,3 +97,72 @@ func TestRelayCursorGroupResume(t *testing.T) { require.True(t, ok) require.Equal(t, uint64(640), g) } + +// Regression: connectRelayMoq shares repoStreamCallbacks with the WebSocket +// path, and those callbacks observe the upstream at-sequence on every event. +// On a moq relay that seq (billions) must stay out of the group cursor +// (thousands): observe() is max-wins, so one poisoned observation would bury +// the group cursor and the next SubscribeFrom would wait forever, silently, on +// a group the relay will never produce. +func TestMoqCallbacksDontPoisonGroupCursor(t *testing.T) { + atsync := &ATProtoSynchronizer{ + commitDedup: newFirehoseDeduper(dedupWindow), + identityDedup: newFirehoseDeduper(dedupWindow), + } + commit := &indigoatproto.SyncSubscribeRepos_Commit{ + Seq: 5_000_000_000, // an upstream at-seq, far above any group seq + Repo: "did:plc:abc123", + Commit: lexutil.LexLink(mustCID(t)), + } + // Pre-claim the commit in the deduper so the callback stops after its + // cursor/metrics bookkeeping instead of spawning the indexing handler. + atsync.commitDedup.seen(commit.Commit.String()) + + // moq relay: the group cursor must survive the callback untouched. + moqCursor := &relayCursor{host: "moqt://relay.example"} + moqCursor.observeGroup(640) + rsc := atsync.repoStreamCallbacks(context.Background(), "moqt://relay.example", moqCursor, func() {}) + require.NoError(t, rsc.RepoCommit(commit)) + g, ok := moqCursor.groupStart() + require.True(t, ok) + require.Equal(t, uint64(640), g, "upstream at-seq must not bury the group cursor") + + // websocket relay: the same callback is what advances the cursor. + wsCursor := &relayCursor{host: "wss://relay.example"} + rsc = atsync.repoStreamCallbacks(context.Background(), "wss://relay.example", wsCursor, func() {}) + require.NoError(t, rsc.RepoCommit(commit)) + v, ok := wsCursor.param() + require.True(t, ok) + require.Equal(t, int64(5_000_000_000), v) +} + +func TestRelayCursorReset(t *testing.T) { + mod, err := model.MakeDB(":memory:") + require.NoError(t, err) + atsync := &ATProtoSynchronizer{Model: mod} + ctx := context.Background() + const host = "moqt://relay.example" + + rc := atsync.newRelayCursor(ctx, host) + rc.observeGroup(640) + rc.flush(ctx) + + // A cursor a resume proved untrustworthy is abandoned wholesale: reset + // drops it below max-wins observe()'s reach, and the next flush heals the + // stored row too. + rc.reset() + if _, ok := rc.groupStart(); ok { + t.Fatal("reset cursor should tail live") + } + rc.flush(ctx) + stored, err := mod.GetRelayCursor(host) + require.NoError(t, err) + require.NotNil(t, stored) + require.Equal(t, int64(0), stored.Cursor) + + // A restart after the reset flush also tails live. + resumed := atsync.newRelayCursor(ctx, host) + if _, ok := resumed.groupStart(); ok { + t.Fatal("restarted cursor should tail live after reset") + } +} diff --git a/pkg/atproto/firehose_moq.go b/pkg/atproto/firehose_moq.go index 66901086..b524d5c7 100644 --- a/pkg/atproto/firehose_moq.go +++ b/pkg/atproto/firehose_moq.go @@ -3,16 +3,24 @@ package atproto import ( "bytes" "context" + "errors" "fmt" + "time" indigoatproto "github.com/bluesky-social/indigo/api/atproto" "github.com/bluesky-social/indigo/events" "github.com/bluesky-social/indigo/events/schedulers/parallel" - atmoq "github.com/streamplace/atmoq-go" + atmoq "github.com/streamplace/atmoq/go" "stream.place/streamplace/pkg/log" "stream.place/streamplace/pkg/spmetrics" ) +// moqResumeProbeTimeout bounds how long a resumed subscription may stay silent +// before we conclude the cursor is bad and re-tail live. The firehose delivers +// many events per second, so a healthy resume answers near-instantly; this only +// needs to be generous enough to cover a slow disk backfill starting up. +const moqResumeProbeTimeout = 30 * time.Second + // connectRelayMoq consumes one relay's atproto firehose over MoQ transport // (moqt:// and its aliases) instead of WebSocket, and pumps it until the // connection drops or ctx is cancelled. atmoq-go subscribes at the live edge @@ -40,8 +48,9 @@ func (atsync *ATProtoSynchronizer) connectRelayMoq(ctx context.Context, relay st // relay jumps forward to the oldest it still retains, leaving a gap we accept // here (deep recovery is a PDS re-sync, tracked separately). var sub *atmoq.Subscription - if g, ok := cursor.groupStart(); ok { - sub, err = sess.SubscribeFrom(streamCtx, atmoq.DefaultBroadcast, atmoq.DefaultTrack, g) + resumedFrom, resumed := cursor.groupStart() + if resumed { + sub, err = sess.SubscribeFrom(streamCtx, atmoq.DefaultBroadcast, atmoq.DefaultTrack, resumedFrom) } else { sub, err = sess.Subscribe(streamCtx, atmoq.DefaultBroadcast, atmoq.DefaultTrack) } @@ -60,13 +69,32 @@ func (atsync *ATProtoSynchronizer) connectRelayMoq(ctx context.Context, relay st log.Log(ctx, "connected to relay firehose (moq)", "version", sess.Version()) for { - raw, group, err := sub.ReadFrame(streamCtx) + // The relay accepts a SubscribeFrom for any group, including one past + // its live edge that it will never produce (a stale or corrupted + // cursor), and just serves silence. So probe the first frame of a + // resume: if replay yields nothing, abandon the cursor and reconnect + // at the live edge — the skipped replay is a gap of the kind we + // already accept (idempotent handlers + cross-relay redelivery). + rctx := streamCtx + var cancelProbe context.CancelFunc + if resumed { + rctx, cancelProbe = context.WithTimeout(streamCtx, moqResumeProbeTimeout) + } + raw, group, err := sub.ReadFrame(rctx) + if cancelProbe != nil { + cancelProbe() + } if err != nil { if streamCtx.Err() != nil { return streamCtx.Err() } + if resumed && errors.Is(err, context.DeadlineExceeded) { + cursor.reset() + return fmt.Errorf("moq resume from group %d served nothing for %s; cursor reset, tailing live on reconnect", resumedFrom, moqResumeProbeTimeout) + } return fmt.Errorf("moq firehose read: %w", err) } + resumed = false // Track the group on every frame (before dedup) so a reconnect resumes // from here; replayed frames are absorbed by the commit-CID deduper. cursor.observeGroup(group) diff --git a/pkg/atproto/firehose_moq_test.go b/pkg/atproto/firehose_moq_test.go index 59e60042..b424eec7 100644 --- a/pkg/atproto/firehose_moq_test.go +++ b/pkg/atproto/firehose_moq_test.go @@ -10,7 +10,7 @@ import ( "github.com/bluesky-social/indigo/events" "github.com/bluesky-social/indigo/events/schedulers/parallel" "github.com/ipfs/go-cid" - atmoq "github.com/streamplace/atmoq-go" + atmoq "github.com/streamplace/atmoq/go" glex "github.com/streamplace/glex/runtime" "github.com/stretchr/testify/require" "stream.place/streamplace/pkg/comatproto" -- 2.51.2 From 966867f28d55698e5e6f07a05049957b2d16f624 Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Wed, 15 Jul 2026 15:23:17 -0700 Subject: [PATCH 2/3] atproto: remove accidental log line --- pkg/atproto/firehose.go | 1 - 1 file changed, 1 deletion(-) diff --git a/pkg/atproto/firehose.go b/pkg/atproto/firehose.go index eafcbe59..a54c3cf5 100644 --- a/pkg/atproto/firehose.go +++ b/pkg/atproto/firehose.go @@ -448,7 +448,6 @@ func (atsync *ATProtoSynchronizer) handleCommitEventOps(ctx context.Context, evt return } ctx := log.WithLogValues(ctx, "eventKind", op.Action, "collection", collection.String(), "rkey", rkey.String()) - log.Warn(ctx, "got event") if len(CollectionFilter) > 0 { keep := slices.Contains(CollectionFilter, collection.String()) -- 2.51.2 From 279d0d8bcd82a3fc59553c5ba2c151131d94f77c Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Wed, 15 Jul 2026 18:05:56 -0700 Subject: [PATCH 3/3] moqt: handle bad data more gracefully --- pkg/atproto/firehose_moq.go | 63 ++++++++++++++++++++++++++++---- pkg/atproto/firehose_moq_test.go | 8 ++-- pkg/spmetrics/spmetrics.go | 11 ++++++ 3 files changed, 71 insertions(+), 11 deletions(-) diff --git a/pkg/atproto/firehose_moq.go b/pkg/atproto/firehose_moq.go index b524d5c7..3b5265d9 100644 --- a/pkg/atproto/firehose_moq.go +++ b/pkg/atproto/firehose_moq.go @@ -10,6 +10,7 @@ import ( indigoatproto "github.com/bluesky-social/indigo/api/atproto" "github.com/bluesky-social/indigo/events" "github.com/bluesky-social/indigo/events/schedulers/parallel" + "github.com/fxamacker/cbor/v2" atmoq "github.com/streamplace/atmoq/go" "stream.place/streamplace/pkg/log" "stream.place/streamplace/pkg/spmetrics" @@ -98,7 +99,7 @@ func (atsync *ATProtoSynchronizer) connectRelayMoq(ctx context.Context, relay st // Track the group on every frame (before dedup) so a reconnect resumes // from here; replayed frames are absorbed by the commit-CID deduper. cursor.observeGroup(group) - if err := atsync.dispatchMoqFrame(streamCtx, raw, scheduler); err != nil { + if err := atsync.dispatchMoqFrame(streamCtx, raw, group, relay, protocol, scheduler); err != nil { return err } } @@ -111,12 +112,27 @@ func (atsync *ATProtoSynchronizer) connectRelayMoq(ctx context.Context, relay st // message types whose callbacks repoStreamCallbacks registers; other types // (#sync/#account/#info/#labels) are decoded-but-ignored just as they are over // WebSocket (no matching callback => dropped). -func (atsync *ATProtoSynchronizer) dispatchMoqFrame(ctx context.Context, raw []byte, scheduler *parallel.Scheduler) error { +// +// A frame that fails strict decoding is logged (with whatever DID/seq context +// a lenient decode can salvage) and SKIPPED, not returned as an error: one +// undecodable event must not take down the whole relay connection, and — per +// the 2026-07-15 investigation — such frames have so far been relay-side +// serving corruption, not user data, so there is nothing to do but drop them +// and let cross-relay redelivery / PDS re-sync repair the gap. Only an +// explicit protocol-level ErrorFrame (or a scheduler/ctx failure) is fatal. +func (atsync *ATProtoSynchronizer) dispatchMoqFrame(ctx context.Context, raw []byte, group uint64, relay, protocol string, scheduler *parallel.Scheduler) error { + skip := func(stage string, err error) error { + spmetrics.FirehoseBadFramesTotal.WithLabelValues(relay, protocol).Inc() + fields := append([]any{"stage", stage, "err", err, "group", group, "len", len(raw)}, badFrameContext(raw)...) + log.Warn(ctx, "skipping undecodable firehose frame", fields...) + return nil + } + r := bytes.NewReader(raw) var header events.EventHeader if err := header.UnmarshalCBOR(r); err != nil { - return fmt.Errorf("reading moq frame header: %w", err) + return skip("header", err) } switch header.Op { @@ -125,13 +141,13 @@ func (atsync *ATProtoSynchronizer) dispatchMoqFrame(ctx context.Context, raw []b case "#commit": var evt indigoatproto.SyncSubscribeRepos_Commit if err := evt.UnmarshalCBOR(r); err != nil { - return fmt.Errorf("reading moq commit event: %w", err) + return skip("commit", err) } return scheduler.AddWork(ctx, evt.Repo, &events.XRPCStreamEvent{RepoCommit: &evt}) case "#identity": var evt indigoatproto.SyncSubscribeRepos_Identity if err := evt.UnmarshalCBOR(r); err != nil { - return fmt.Errorf("reading moq identity event: %w", err) + return skip("identity", err) } return scheduler.AddWork(ctx, evt.Did, &events.XRPCStreamEvent{RepoIdentity: &evt}) default: @@ -140,10 +156,43 @@ func (atsync *ATProtoSynchronizer) dispatchMoqFrame(ctx context.Context, raw []b case events.EvtKindErrorFrame: var errframe events.ErrorFrame if err := errframe.UnmarshalCBOR(r); err != nil { - return fmt.Errorf("reading moq error frame: %w", err) + return skip("errorframe", err) } return fmt.Errorf("moq firehose error frame: %s: %s", errframe.Error, errframe.Message) default: - return fmt.Errorf("unrecognized moq event op: %d", header.Op) + return skip("op", fmt.Errorf("unrecognized moq event op: %d", header.Op)) + } +} + +// badFrameContext best-effort-decodes a frame that failed strict decoding so +// the skip log can say whose event it was. Lenient on purpose (invalid UTF-8, +// unknown simple values, and truncation all still yield whatever fields sit in +// the valid prefix); returns log key/value pairs, possibly none. +func badFrameContext(raw []byte) []any { + dm, err := cbor.DecOptions{UTF8: cbor.UTF8DecodeInvalid}.DecMode() + if err != nil { + return nil + } + dec := dm.NewDecoder(bytes.NewReader(raw)) + var hdr, pay map[string]any + _ = dec.Decode(&hdr) + _ = dec.Decode(&pay) + var fields []any + if t, ok := hdr["t"].(string); ok { + fields = append(fields, "msgType", t) + } + // #commit events carry the repo in "repo"; #identity/#account in "did". + for _, k := range []string{"repo", "did"} { + if v, ok := pay[k].(string); ok { + fields = append(fields, "did", v) + break + } + } + if seq, ok := pay["seq"]; ok { + fields = append(fields, "seq", seq) + } + if rev, ok := pay["rev"].(string); ok { + fields = append(fields, "rev", rev) } + return fields } diff --git a/pkg/atproto/firehose_moq_test.go b/pkg/atproto/firehose_moq_test.go index b424eec7..23555f24 100644 --- a/pkg/atproto/firehose_moq_test.go +++ b/pkg/atproto/firehose_moq_test.go @@ -59,7 +59,7 @@ func TestDispatchMoqFrameCommit(t *testing.T) { sched, out := collectScheduler(t) atsync := &ATProtoSynchronizer{} - require.NoError(t, atsync.dispatchMoqFrame(context.Background(), buf.Bytes(), sched)) + require.NoError(t, atsync.dispatchMoqFrame(context.Background(), buf.Bytes(), 1, "moqt://test", "moq", sched)) select { case ev := <-out: @@ -86,7 +86,7 @@ func TestDispatchMoqFrameIdentity(t *testing.T) { sched, out := collectScheduler(t) atsync := &ATProtoSynchronizer{} - require.NoError(t, atsync.dispatchMoqFrame(context.Background(), buf.Bytes(), sched)) + require.NoError(t, atsync.dispatchMoqFrame(context.Background(), buf.Bytes(), 1, "moqt://test", "moq", sched)) select { case ev := <-out: @@ -113,7 +113,7 @@ func TestDispatchMoqFrameIgnoredType(t *testing.T) { sched, out := collectScheduler(t) atsync := &ATProtoSynchronizer{} - require.NoError(t, atsync.dispatchMoqFrame(context.Background(), buf.Bytes(), sched)) + require.NoError(t, atsync.dispatchMoqFrame(context.Background(), buf.Bytes(), 1, "moqt://test", "moq", sched)) select { case ev := <-out: @@ -148,7 +148,7 @@ func TestFirehoseMoqLive(t *testing.T) { for commits < 5 { raw, _, err := sub.ReadFrame(ctx) require.NoError(t, err, "reading %d live frames", commits) - require.NoError(t, atsync.dispatchMoqFrame(ctx, raw, sched)) + require.NoError(t, atsync.dispatchMoqFrame(ctx, raw, 1, "moqt://test", "moq", sched)) select { case ev := <-out: if ev.RepoCommit != nil { diff --git a/pkg/spmetrics/spmetrics.go b/pkg/spmetrics/spmetrics.go index af784f62..4cf13bc7 100644 --- a/pkg/spmetrics/spmetrics.go +++ b/pkg/spmetrics/spmetrics.go @@ -132,6 +132,17 @@ var FirehoseEventsDedupedTotal = promauto.NewCounterVec(prometheus.CounterOpts{ Help: "firehose events dropped as cross-relay duplicates, by relay/protocol/kind", }, []string{"relay", "protocol", "kind"}) +// FirehoseBadFramesTotal counts firehose frames that failed strict decoding +// and were skipped (the connection stays up; only the frame is dropped), +// labeled by relay and protocol. A nonzero rate means a relay is delivering +// bytes that violate the atproto event schema — either corruption in the +// relay's serving path or genuinely invalid upstream data; the accompanying +// warn log carries the decoded context (DID, seq) for investigation. +var FirehoseBadFramesTotal = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "streamplace_firehose_bad_frames_total", + Help: "firehose frames skipped because they failed decoding, by relay/protocol", +}, []string{"relay", "protocol"}) + // FirehoseRelayHighSeq is the highest upstream sequence number seen from each // relay. A WebSocket bsky relay and a MoQ relay that bridges the same upstream // both carry bsky's original seq, so the difference between two relays' high_seq -- 2.51.2