From 98ec0e0fec9b8b6535d1ce34ae2cd14bccec4c92 Mon Sep 17 00:00:00 2001
From: afterlifepro <vielle.dev@proton.me>
Date: Wed, 28 Jan 2026 21:05:20 +0000
Subject: [PATCH] better reconnection logic

---
 src/config/mod.rs   |  5 ++++-
 src/ingest/queue.rs | 27 ++++++++++++++++++---------
 2 files changed, 22 insertions(+), 10 deletions(-)

diff --git a/src/config/mod.rs b/src/config/mod.rs
index 9cde13a..0495ea0 100644
--- a/src/config/mod.rs
+++ b/src/config/mod.rs
@@ -11,6 +11,7 @@ use jacquard::types::string::Did;
 use std::env;
 
 pub const DB_MAX_REQ: usize = 65535;
+pub const MAX_RECONNECT_ATTEMPTS: usize = 30;
 
 mod config_value;
 
@@ -80,7 +81,9 @@ pub static USER_PDS_URL: ConfigValue<String> = ConfigValue::new_async(|| {
             resolver
                 .pds_for_did(&self::USER_DID)
                 .await
-                .expect("The did document for a user should contain a #atproto_pds service with a url")
+                .expect(
+                    "The did document for a user should contain a #atproto_pds service with a url",
+                )
                 .domain()
                 .expect("A users pds should have a domain.")
                 .to_string()
diff --git a/src/ingest/queue.rs b/src/ingest/queue.rs
index 15c91fe..291f18a 100644
--- a/src/ingest/queue.rs
+++ b/src/ingest/queue.rs
@@ -47,18 +47,27 @@ pub async fn queue() -> (
                                 // if it failed, try reconnect 10 times, waiting a second between each attempt
                                 Err(_) => {
                                     let mut new_messages = None;
-                                    for _ in 0..10 {
-                                        let Ok(stream) =
-                                            client.subscribe(&SubscribeRepos::new().build()).await
-                                        else {
-                                            // wait a second
-                                            thread::sleep(Duration::from_secs(1));
-                                            continue;
-                                        };
-                                        new_messages = Some(stream.into_stream().1)
+                                    for i in 0..config::MAX_RECONNECT_ATTEMPTS {
+                                        new_messages = match client
+                                            .subscribe(&SubscribeRepos::new().build())
+                                            .await
+                                        {
+                                            Ok(val) => Some(val.into_stream().1),
+                                            Err(err) => {
+                                                eprintln!(
+                                                    "Warning: Error: {} ({}/{})",
+                                                    err,
+                                                    i + 1,
+                                                    config::MAX_RECONNECT_ATTEMPTS
+                                                );
+                                                thread::sleep(Duration::from_secs(1));
+                                                continue;
+                                            }
+                                        }
                                     }
 
                                     if let Some(new_messages) = new_messages {
+                                        println!("Reconnected.");
                                         new_messages
                                     } else {
                                         // could not reconnect so just die lmao
-- 
2.51.2