danabra.mov / slices
forked from slices.network/slices
Highly ambitious ATProtocol AppView service and sdks
fork atom

fix jetstream reconnect logic

chadtmiller.com e4e8789d 7b9f4d02

+6 -2
unified split
+6 -2
api/src/main.rs
··· 187 187 if now.duration_since(window_start) >= RECONNECT_WINDOW { 188 188 reconnect_count = 0; 189 189 window_start = now; 190 + retry_delay = tokio::time::Duration::from_secs(5); // Reset delay after window passes 190 191 } 191 192 192 193 // Check rate limit ··· 202 203 } 203 204 204 205 reconnect_count += 1; 206 + tracing::info!("Jetstream connection attempt #{} (retry delay: {:?})", reconnect_count, retry_delay); 205 207 206 208 // Read cursor position from database 207 209 let initial_cursor = ··· 265 267 let cancellation_token = atproto_jetstream::CancellationToken::new(); 266 268 match consumer_arc.start_consuming(cancellation_token).await { 267 269 Ok(_) => { 268 - tracing::info!("Jetstream consumer shut down normally"); 270 + tracing::info!("Jetstream consumer shut down normally - reconnecting in {:?}", retry_delay); 269 271 jetstream_connected_clone 270 272 .store(false, std::sync::atomic::Ordering::Relaxed); 273 + tokio::time::sleep(retry_delay).await; 274 + retry_delay = std::cmp::min(retry_delay * 2, MAX_RETRY_DELAY); 271 275 } 272 276 Err(e) => { 273 - tracing::error!("Jetstream consumer failed: {} - will reconnect", e); 277 + tracing::error!("Jetstream consumer failed: {} - reconnecting in {:?}", e, retry_delay); 274 278 jetstream_connected_clone 275 279 .store(false, std::sync::atomic::Ordering::Relaxed); 276 280 tokio::time::sleep(retry_delay).await;