From da40d3539c0bbbf32e9ebede7490fe280b35a868 Mon Sep 17 00:00:00 2001 From: Timothy Quilling Date: Mon, 13 Oct 2025 01:33:52 -0400 Subject: [PATCH] fix: handle more websocket errors --- consumer/src/indexer/mod.rs | 24 ++++++++++++++++++++---- 1 file changed, 20 insertions(+), 4 deletions(-) diff --git a/consumer/src/indexer/mod.rs b/consumer/src/indexer/mod.rs index 1ae455bd..700316d3 100644 --- a/consumer/src/indexer/mod.rs +++ b/consumer/src/indexer/mod.rs @@ -725,16 +725,24 @@ impl RelayIndexer { let error_str = err.to_string(); let is_websocket_eof = error_str.contains("WebSocket error") && error_str.contains("unexpected EOF"); + let is_websocket_reset = error_str.contains("WebSocket error") && + error_str.contains("Connection reset without closing handshake"); + let is_websocket_error = error_str.contains("WebSocket error"); let is_parsing_error = error_str.contains("Failed to parse Jetstream event"); - // Handle either parsing errors or WebSocket EOF errors - let should_reconnect = is_websocket_eof || is_parsing_error; + // Handle websocket errors or parsing errors + let should_reconnect = is_websocket_eof || is_websocket_reset || is_websocket_error || is_parsing_error; if should_reconnect { if let Some(config) = &self.opts.config { if config.use_jetstream { if let Some(jetstream_url) = &config.jetstream_source { - tracing::info!("Attempting to reconnect to Jetstream after error: {}", error_str); + tracing::info!("Attempting to reconnect to Jetstream after error: {} (reconnecting for {})", + error_str, + if is_websocket_eof { "unexpected EOF" } + else if is_websocket_reset { "connection reset" } + else if is_websocket_error { "websocket error" } + else { "parsing error" }); let seq = Some(self.consumer.current_seq()); let ua = "Parakeet Indexer"; @@ -781,11 +789,19 @@ impl RelayIndexer { } } - // If we couldn't reconnect, continue rather than exit for parsing errors + // If we couldn't reconnect but the error is non-fatal, we can continue if is_parsing_error { tracing::warn!("Continuing after Jetstream parsing error"); counter!("jetstream_events.parse_error_continued").increment(1); continue 'outer; + } else if is_websocket_error { + // For websocket errors, we've already tried to reconnect but it failed + // Log this but continue trying rather than completely exiting + tracing::warn!("Continuing after failed reconnection attempt for WebSocket error"); + counter!("jetstream_events.websocket_error_continued").increment(1); + // Sleep briefly before retrying + tokio::time::sleep(std::time::Duration::from_secs(2)).await; + continue 'outer; } // If we couldn't reconnect or it wasn't a reconnectable error, exit -- 2.51.2