From 966e391a91ebafbb2ee9f94fcabbe28f4097554c Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Sat, 9 May 2026 15:47:19 -0500 Subject: [PATCH] fix some connection issues with jetstream causing a crashloop --- .goreleaser.yaml | 1 + pkg/appview/jetstream/worker.go | 96 ++++++++++++++++++++++----------- 2 files changed, 65 insertions(+), 32 deletions(-) diff --git a/.goreleaser.yaml b/.goreleaser.yaml index 442a8dd..b12a2e4 100644 --- a/.goreleaser.yaml +++ b/.goreleaser.yaml @@ -50,6 +50,7 @@ brews: ids: - credential-helper repository: + # https://tangled.org/did:plc:e3kzdezk5gsirzh7eoqplc64/ # Tap lives inside the main repo under Formula/. DID-based URL keeps # this stable across any future Tangled handle/domain rename. owner: 'did:plc:pddp4xt5lgnv2qsegbzzs4xg' diff --git a/pkg/appview/jetstream/worker.go b/pkg/appview/jetstream/worker.go index b8d1c54..d608db1 100644 --- a/pkg/appview/jetstream/worker.go +++ b/pkg/appview/jetstream/worker.go @@ -46,6 +46,13 @@ type Worker struct { lastPongTime time.Time pongMutex sync.Mutex + // Last-event tracking — used as a liveness signal alongside pongs, since + // some upstream Jetstream endpoints don't reliably reply to client pings. + // The connection is only considered dead when BOTH pongs and events have + // been silent past the timeout. + lastEventTime time.Time + lastEventMutex sync.Mutex + // In-memory cursor tracking for reconnects lastCursor int64 cursorMutex sync.RWMutex @@ -131,6 +138,11 @@ func (w *Worker) Start(ctx context.Context) error { w.lastPongTime = time.Now() w.pongMutex.Unlock() + // Seed last-event time so the watchdog has a baseline before any frames arrive. + w.lastEventMutex.Lock() + w.lastEventTime = time.Now() + w.lastEventMutex.Unlock() + // Set up pong handler - called when server responds to our ping conn.SetPongHandler(func(appData string) error { w.pongMutex.Lock() @@ -184,9 +196,21 @@ func (w *Worker) Start(ctx context.Context) error { pongsTotal := w.pongsReceived w.pongMutex.Unlock() - // If no pong for 60 seconds, connection is likely dead - if timeSinceLastPong > 60*time.Second { - slog.Info("Jetstream no pong received, closing connection", "time_since_last_pong", timeSinceLastPong, "pings_sent", pingsTotal, "pongs_received", pongsTotal) + // Connection is only "dead" when BOTH pongs and events have been + // silent for 60s. Some upstream endpoints don't reply to client + // pings even when the firehose is happily streaming data, so + // using pong silence alone causes spurious reconnects (and a + // rewind/replay loop on the cursor). + w.lastEventMutex.Lock() + timeSinceLastEvent := time.Since(w.lastEventTime) + w.lastEventMutex.Unlock() + + if timeSinceLastPong > 60*time.Second && timeSinceLastEvent > 60*time.Second { + slog.Info("Jetstream no pong AND no events, closing connection", + "time_since_last_pong", timeSinceLastPong, + "time_since_last_event", timeSinceLastEvent, + "pings_sent", pingsTotal, + "pongs_received", pongsTotal) conn.Close() return } @@ -261,6 +285,16 @@ func (w *Worker) Start(ctx context.Context) error { return fmt.Errorf("failed to read message: %w", err) } + // Any frame counts as liveness — refresh the deadline and the + // last-event marker so the watchdog and the read deadline don't + // fire while data is flowing, even if the server isn't ponging. + w.lastEventMutex.Lock() + w.lastEventTime = time.Now() + w.lastEventMutex.Unlock() + if err := conn.SetReadDeadline(time.Now().Add(90 * time.Second)); err != nil { + slog.Warn("Jetstream failed to refresh read deadline", "error", err) + } + // For now, process uncompressed messages // TODO: Re-enable compression once debugging is complete _ = decoder // Keep decoder to avoid unused variable error @@ -318,15 +352,21 @@ func (w *Worker) StartWithFailover(ctx context.Context) { currentURL := w.endpoints.Current() w.jetstreamURL = currentURL - slog.Info("Jetstream connecting", "url", currentURL) + // Resume from the most recent processed event before each (re)connect, + // otherwise we'd open the WS with the cursor from initial bootstrap + // and replay the same window every time. + if latest := w.GetLastCursor(); latest > 0 { + w.cursorMutex.Lock() + w.startCursor = latest + w.cursorMutex.Unlock() + } + + slog.Info("Jetstream connecting", "url", currentURL, "cursor", w.startCursor) err := w.Start(ctx) if ctx.Err() != nil { return // Context cancelled, clean shutdown } - // Capture cursor at disconnect time for rewind calculation - disconnectCursor := w.GetLastCursor() - // Retry same endpoint with escalating delays recovered := false for i, delay := range retryDelays { @@ -341,6 +381,16 @@ func (w *Worker) StartWithFailover(ctx context.Context) { return } + // Resume from the most recent processed event, not the cursor + // the connection was originally opened with. Without this, every + // reconnect re-uses the bootstrap cursor and replays the same + // historical window forever. + if latest := w.GetLastCursor(); latest > 0 { + w.cursorMutex.Lock() + w.startCursor = latest + w.cursorMutex.Unlock() + } + w.jetstreamURL = currentURL err = w.Start(ctx) if ctx.Err() != nil { @@ -350,39 +400,21 @@ func (w *Worker) StartWithFailover(ctx context.Context) { recovered = true break } - // Update disconnect cursor if we got further - if latest := w.GetLastCursor(); latest > disconnectCursor { - disconnectCursor = latest - } } if recovered { continue } - // All retries failed — failover to next endpoint + // All retries failed — failover to next endpoint. The outer loop will + // pick up startCursor from lastCursor on the next iteration, so no + // rewind is needed here: lastCursor is exactly where we stopped + // processing. failedURL := currentURL nextURL := w.endpoints.Next() - - // Rewind cursor 30 seconds (30M microseconds) to avoid gaps - if disconnectCursor > 0 { - rewound := disconnectCursor - 30_000_000 - if rewound < 0 { - rewound = 0 - } - w.cursorMutex.Lock() - w.lastCursor = rewound - w.startCursor = rewound - w.cursorMutex.Unlock() - slog.Warn("Jetstream failing over to next endpoint", - "failed_url", failedURL, - "next_url", nextURL, - "cursor_rewound_by", "30s") - } else { - slog.Warn("Jetstream failing over to next endpoint", - "failed_url", failedURL, - "next_url", nextURL) - } + slog.Warn("Jetstream failing over to next endpoint", + "failed_url", failedURL, + "next_url", nextURL) } }