fix some connection issues with jetstream causing a crashloop

.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'

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)
 	}
 }