From e1ab350cb7b3a8fa3be874cdcbc2db003d6a86dd Mon Sep 17 00:00:00 2001
From: dawn <90008@gaze.systems>
Date: Thu, 2 Apr 2026 21:20:00 +0300
Subject: [PATCH] [firehose] only reset backoff to 0 if the stream has been
 healthy for over a minute

---
 src/ingest/firehose.rs | 43 +++++++++++++++++++++---------------------
 1 file changed, 22 insertions(+), 21 deletions(-)

diff --git a/src/ingest/firehose.rs b/src/ingest/firehose.rs
index 16e6c58..4bdecf3 100644
--- a/src/ingest/firehose.rs
+++ b/src/ingest/firehose.rs
@@ -157,6 +157,7 @@ impl FirehoseIngestor {
 
             self.throttle.record_success();
             info!("firehose connected");
+            let connected_at = tokio::time::Instant::now();
 
             let res = loop {
                 tokio::select! {
@@ -189,23 +190,23 @@ impl FirehoseIngestor {
                                 }
                                 self.handle_message(msg).await
                             },
-                            Err(e) => {
-                                match e {
-                                    // dont disconnect on unknown op or type
-                                    FirehoseError::UnknownOp(op) => {
-                                        warn!(op = %op, "unknown frame op");
-                                        continue;
-                                    },
-                                    FirehoseError::UnknownType(t) => {
-                                        warn!(ty = %t, "unknown frame type");
-                                        continue;
-                                    },
-                                    // everything else is a hard error
-                                    e => break Err(e),
-                                }
-                            }
+                            Err(e) => match e {
+                                // dont disconnect on unknown op or type
+                                FirehoseError::UnknownOp(op) => {
+                                    warn!(op = %op, "unknown frame op");
+                                    continue;
+                                },
+                                FirehoseError::UnknownType(t) => {
+                                    warn!(ty = %t, "unknown frame type");
+                                    continue;
+                                },
+                                // everything else is a hard error
+                                e => break Err(e),
+                            },
+                        }
+                        if connected_at.elapsed() > Duration::from_secs(60) {
+                            backoff = Duration::from_secs(0);
                         }
-                        backoff = Duration::from_secs(0);
                     }
                     _ = self.enabled.changed() => {
                         if !*self.enabled.borrow() {
@@ -217,11 +218,11 @@ impl FirehoseIngestor {
             };
 
             if let Err(e) = res {
-                // todo: investigate why this happens on test server further
-                // also idk if this is even a good idea to do but whatever
-                if let FirehoseError::TcpDropped = e {
-                    debug!(err = %e, "tcp connection dropped!!!");
-                    tokio::time::sleep(rng.add_jitter(Duration::from_secs(10))).await;
+                if let FirehoseError::StreamClosed { code, reason } = &e
+                    && *code == 1001
+                {
+                    debug!(reason = %reason, "host gone away");
+                    tokio::time::sleep(Duration::from_secs(1)).await;
                     continue;
                 }
                 if let FirehoseError::RelayError { error, message } = e {
-- 
2.51.2