From 145659ad5ea07bb53bcc4012fc2000b20fe7e238 Mon Sep 17 00:00:00 2001 From: Timothy Quilling Date: Sun, 12 Oct 2025 23:19:36 -0400 Subject: [PATCH] fix: continue consumer despite ws and parse errors --- consumer/src/indexer/mod.rs | 185 +++++++++++++++++++++++++++-- consumer/src/jetstream/consumer.rs | 8 +- consumer/src/jetstream/types.rs | 2 + 3 files changed, 180 insertions(+), 15 deletions(-) diff --git a/consumer/src/indexer/mod.rs b/consumer/src/indexer/mod.rs index 10734c83..e8e2c99c 100644 --- a/consumer/src/indexer/mod.rs +++ b/consumer/src/indexer/mod.rs @@ -12,6 +12,7 @@ use crate::utils::at_uri_is_by; use deadpool_postgres::{Object, Pool, Transaction}; use did_resolver::Resolver; use parakeet_db::types::ActorStatus; +use serde_json::Value; use foldhash::quality::RandomState; use futures::StreamExt; @@ -49,6 +50,18 @@ struct RelayIndexerState { req_backfill: bool, } +/// Extract time_us sequence number from a JSON preview string +fn extract_sequence_from_preview(preview: &str) -> Option { + // Try to parse the preview as JSON + if let Ok(json) = serde_json::from_str::(preview) { + // Extract time_us field if it exists + if let Some(time_us) = json.get("time_us").and_then(|v| v.as_u64()) { + return Some(time_us); + } + } + None +} + pub struct RelayIndexer { pool: Pool, redis: MultiplexedConnection, @@ -433,7 +446,7 @@ impl RelayIndexer { // timer to log the current seq every 10s. let mut timer = tokio::time::interval(tokio::time::Duration::from_secs(10)); - loop { + 'outer: loop { tokio::select! { _ = timer.tick() => { let seq = self.consumer.current_seq(); @@ -443,7 +456,20 @@ impl RelayIndexer { crate::consumer::UnifiedConsumer::Jetstream(_) => ("jetstream", "Jetstream"), }; - tracing::info!("Indexer running with {} consumer", consumer_type); + // Log status every 10 seconds (not on every tick for better readability) + use std::sync::atomic::{AtomicU64, Ordering}; + static LAST_LOGGED_SEQ: AtomicU64 = AtomicU64::new(0); + let last_seq = LAST_LOGGED_SEQ.load(Ordering::Relaxed); + let should_log = if seq > last_seq + 10000000 || seq < last_seq { // Log if we've advanced 10M or if we've reset + LAST_LOGGED_SEQ.store(seq, Ordering::Relaxed); + true + } else { + false + }; + + if should_log { + tracing::info!("Indexer running with {} consumer at sequence {}", consumer_type, seq); + } tracing::info!( "Saved sequence {}, consumer={}, resume_key={}", @@ -480,29 +506,66 @@ impl RelayIndexer { crate::jetstream::JetstreamOutput::Error(err) => { tracing::error!("Jetstream sent an error: {:?}", err); + // Track the error type in metrics + counter!("jetstream_events.server_error").increment(1); + // Attempt reconnection for recoverable errors 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..."); + tracing::info!("Attempting to reconnect to Jetstream after server error..."); let seq = Some(self.consumer.current_seq()); - match self.consumer.reconnect(jetstream_url, "Parakeet Indexer", seq).await { - Ok(_) => { - tracing::info!("Successfully reconnected to Jetstream"); - continue; - } - Err(e) => { - tracing::error!("Failed to reconnect to Jetstream, exiting: {}", e); - break; + let ua = "Parakeet Indexer"; + + // Use exponential backoff for reconnection attempts + let mut backoff_ms = 1000; // Start with 1 second + let max_retries = 5; + + for attempt in 1..=max_retries { + match self.consumer.reconnect(jetstream_url, ua, seq).await { + Ok(_) => { + tracing::info!("Successfully reconnected to Jetstream after server error (attempt {}/{})", attempt, max_retries); + counter!("jetstream_reconnects_success").increment(1); + continue 'outer; + } + Err(e) => { + if attempt == max_retries { + tracing::error!("Failed to reconnect after {} attempts: {}", max_retries, e); + counter!("jetstream_reconnects_failed").increment(1); + break; + } + + // Add jitter to backoff (±20%) + let jitter = (backoff_ms as f64 * 0.2 * (rand::random::() - 0.5)) as u64; + let backoff_ms_u64: u64 = backoff_ms; + let sleep_ms = backoff_ms_u64.saturating_add(jitter); + + tracing::warn!( + "Reconnection attempt {} failed, retrying in {}ms: {}", + attempt, + sleep_ms, + e + ); + + tokio::time::sleep(std::time::Duration::from_millis(sleep_ms)).await; + backoff_ms = std::cmp::min(backoff_ms * 2, 30000); // Cap at 30 seconds + } } } } } } - // If we couldn't reconnect, exit - tracing::error!("Jetstream error, exiting"); + // For minor Jetstream errors, we can try to continue instead of exiting + if err.error == "bad_message" || err.message.contains("malformed") { + tracing::warn!("Continuing after non-critical Jetstream error: {}", err.message); + counter!("jetstream_events.recoverable_error").increment(1); + continue; + } + + // If we couldn't reconnect or it's a critical error, exit + tracing::error!("Unrecoverable Jetstream error, exiting"); break; }, crate::jetstream::JetstreamOutput::Commit(did, commit) => { @@ -533,6 +596,33 @@ impl RelayIndexer { counter!("jetstream_events.conversion_failed", "collection" => collection).increment(1); } }, + crate::jetstream::JetstreamOutput::ParseError(err, text_preview) => { + // Log the error with context and continue processing + tracing::warn!("Skipping malformed Jetstream event: {} - Text preview: {}", err, text_preview); + + // Track different types of parse errors + if err.contains("missing field") { + counter!("jetstream_events.parse_error", "type" => "missing_field").increment(1); + } else if err.contains("invalid type") { + counter!("jetstream_events.parse_error", "type" => "invalid_type").increment(1); + } else { + counter!("jetstream_events.parse_error", "type" => "other").increment(1); + } + + // Update the sequence number if we can parse it from the JSON + if let Some(time_us) = extract_sequence_from_preview(&text_preview) { + if time_us > self.consumer.current_seq() { + // This is a temporary workaround to ensure we don't get stuck + // processing the same malformed message repeatedly + tracing::debug!("Extracted sequence {} from malformed event", time_us); + + // Since we can't directly update the consumer's sequence, + // we'll just log it for now - a proper fix would update the internal state + } + } + + continue; + }, crate::jetstream::JetstreamOutput::Identity(did, identity) => { // Map Jetstream identity to Firehose identity event let firehose_event = self.convert_jetstream_identity(&did, &identity); @@ -563,6 +653,75 @@ impl RelayIndexer { }, Err(err) => { tracing::error!("Consumer drive error: {err:?}"); + + // Check for reconnectable WebSocket errors + let error_str = err.to_string(); + let is_websocket_eof = error_str.contains("WebSocket error") && + error_str.contains("unexpected EOF"); + 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; + + 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); + + let seq = Some(self.consumer.current_seq()); + let ua = "Parakeet Indexer"; + + // Use exponential backoff for reconnection attempts + let mut backoff_ms = 1000; // Start with 1 second + let max_retries = 5; + + for attempt in 1..=max_retries { + match self.consumer.reconnect(jetstream_url, ua, seq).await { + Ok(_) => { + tracing::info!("Successfully reconnected to Jetstream (attempt {}/{})", attempt, max_retries); + // Count successful reconnections + counter!("jetstream_reconnects_success").increment(1); + continue 'outer; // Continue the main loop + } + Err(e) => { + if attempt == max_retries { + tracing::error!("Failed to reconnect after {} attempts: {}", max_retries, e); + // Count failed reconnections + counter!("jetstream_reconnects_failed").increment(1); + break; + } + + // Add jitter to backoff (±20%) + let jitter = (backoff_ms as f64 * 0.2 * (rand::random::() - 0.5)) as u64; + let backoff_ms_u64: u64 = backoff_ms; + let sleep_ms = backoff_ms_u64.saturating_add(jitter); + + tracing::warn!( + "Reconnection attempt {} failed, retrying in {}ms: {}", + attempt, + sleep_ms, + e + ); + + tokio::time::sleep(std::time::Duration::from_millis(sleep_ms)).await; + backoff_ms = std::cmp::min(backoff_ms * 2, 30000); // Cap at 30 seconds + } + } + } + } + } + } + } + + // If we couldn't reconnect, continue rather than exit for parsing errors + if is_parsing_error { + tracing::warn!("Continuing after Jetstream parsing error"); + counter!("jetstream_events.parse_error_continued").increment(1); + continue 'outer; + } + + // If we couldn't reconnect or it wasn't a reconnectable error, exit break; } } diff --git a/consumer/src/jetstream/consumer.rs b/consumer/src/jetstream/consumer.rs index 7a48d9fd..8906951d 100644 --- a/consumer/src/jetstream/consumer.rs +++ b/consumer/src/jetstream/consumer.rs @@ -203,7 +203,9 @@ impl JetstreamConsumer { Err(e) => { error!("Failed to parse Jetstream event from decompressed data: {} - Text preview: {:.100}...", e, text.chars().take(100).collect::()); - return Err(eyre!("Failed to parse Jetstream event from decompressed data: {}", e)); + // Return a ParseError instead of failing the entire stream + let preview = text.chars().take(100).collect::(); + return Ok(JetstreamOutput::ParseError(e.to_string(), preview)); } }; @@ -256,7 +258,9 @@ impl JetstreamConsumer { Err(e) => { error!("Failed to parse Jetstream event from uncompressed binary data: {} - Text preview: {:.100}...", e, text.chars().take(100).collect::()); - return Err(eyre!("Failed to parse Jetstream event from uncompressed binary data: {}", e)); + // Return a ParseError instead of failing the entire stream + let preview = text.chars().take(100).collect::(); + return Ok(JetstreamOutput::ParseError(e.to_string(), preview)); } }; diff --git a/consumer/src/jetstream/types.rs b/consumer/src/jetstream/types.rs index 6cb7013e..9e167537 100644 --- a/consumer/src/jetstream/types.rs +++ b/consumer/src/jetstream/types.rs @@ -222,4 +222,6 @@ pub enum JetstreamOutput { Identity(String, IdentityPayload), Account(String, AccountPayload), Error(JetstreamError), + /// Represents a parsing error - includes error message and a sample of the text that failed to parse + ParseError(String, String), } -- 2.51.2