From d1cd9999c04afbd38e9e2c38e9416e16d6b5d41b Mon Sep 17 00:00:00 2001
From: crashkeys.dev <dawn@crashkeys.dev>
Date: Wed, 22 Apr 2026 17:52:58 +0000
Subject: [PATCH] actor redis: restore broken connections to server.

Before, whenever the connection to the Redis server closed,
the Redis quote storage actor would panic and stop working altogether.
Now, it instead attempts to open a new connection to the same server.
---
 src/lib.rs | 12 +-----------
 src/sink.rs | 1 -
 src/storage.rs | 74 ++++++++++++++++++++++++++++++++++++++++++++++----------------------------
 3 file(s) changed, 47 insertion(s)(+), 40 deletion(s)(-)

diff --git a/src/lib.rs b/src/lib.rs
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -52,17 +52,7 @@ };
 
 		let use_redis = std::env::var("USE_REDIS").unwrap_or("0".to_string()) == "1";
 		if use_redis {
-			let redis = redis::Client::open(
-				std::env::var("REDIS_URL").unwrap_or("redis://localhost".to_string()),
-			)?;
-			let con = redis.get_multiplexed_async_connection().await?;
-
-			/// The default queue key we use to fetch and store Redis quote.
-			// TODO(redis): Parameterize the quote queue key.
-			const DEFAULT_KEY: &str = "queue:default";
-
-			let queue =
-				RedisQueueStorage::spawn(RedisQueueStorage::new(con, DEFAULT_KEY.to_string()));
+			let queue = RedisQueueStorage::spawn(RedisQueueStorage::new_from_env().unwrap());
 			run_cycle(schedule, source, queue, stdout, bsky).await
 		} else {
 			let queue = MemoryQueueStorage::spawn(MemoryQueueStorage::new());
diff --git a/src/sink.rs b/src/sink.rs
--- a/src/sink.rs
+++ b/src/sink.rs
@@ -229,7 +229,6 @@
 		let messages = ["First test!", "Second test.", "Third..."];
 		for msg in messages {
 			manager.tell(PostQuote(msg.into())).await.unwrap();
-			tokio::time::sleep(std::time::Duration::from_secs(5)).await;
 		}
 
 		// Hopefully we don't crash...!
diff --git a/src/storage.rs b/src/storage.rs
--- a/src/storage.rs
+++ b/src/storage.rs
@@ -190,6 +190,7 @@ }
 }
 
 pub mod queue {
+	use redis::AsyncCommands;
 	use std::collections::VecDeque;
 
 	use super::*;
@@ -273,34 +274,46 @@ }
 	}
 
 	/// Implementation of persistent quote queue storage through a Redis server.
-	#[derive(Actor, Default)]
-	pub struct RedisQueueStorage<R>
-	where
-		R: redis::aio::ConnectionLike + redis::AsyncCommands + 'static,
-	{
+	#[derive(Actor)]
+	pub struct RedisQueueStorage {
 		/// Key that points to the Redis queue to be used for fetching and storing quotes.
 		queue_key: String,
 
-		/// Underlying connection representing the Redis client; must be boxed to derive [Actor].
-		con: Box<R>,
+		/// Underlying client from which to instantiate Redis connections.
+		client: redis::Client,
 	}
 
-	impl<R> RedisQueueStorage<R>
-	where
-		R: redis::aio::ConnectionLike + redis::AsyncCommands + 'static,
-	{
-		pub fn new(con: R, queue_key: String) -> Self {
-			Self {
-				queue_key,
-				con: Box::new(con),
-			}
+	impl RedisQueueStorage {
+		/// The default queue key we use to fetch and store Redis quote.
+		const DEFAULT_KEY: &str = "queue:default";
+
+		pub fn new(client: redis::Client, queue_key: String) -> Self {
+			Self { queue_key, client }
+		}
+
+		pub fn new_from_env() -> Result<Self, ()> {
+			let client = redis::Client::open(
+				std::env::var("REDIS_URL").unwrap_or("redis://localhost".to_string()),
+			)
+			.map_err(|_| ())?;
+
+			let queue_key = std::env::var("REDIS_KEY").unwrap_or(Self::DEFAULT_KEY.to_string());
+
+			Ok(Self { queue_key, client })
+		}
+
+		async fn con(&self) -> Result<redis::aio::MultiplexedConnection, ()> {
+			self.client
+				.get_multiplexed_async_connection()
+				.await
+				.map_err(|e| {
+					tracing::error!(error = e.to_string(), "Could not connect to Redis server");
+					()
+				})
 		}
 	}
 
-	impl<R> Message<EnqueueQuotes> for RedisQueueStorage<R>
-	where
-		R: redis::aio::ConnectionLike + redis::AsyncCommands + 'static,
-	{
+	impl Message<EnqueueQuotes> for RedisQueueStorage {
 		type Reply = EnqueueReply;
 
 		#[tracing::instrument(name = "redis::enqueue_many", skip(self, _ctx, msg))]
@@ -315,9 +328,13 @@ "Enqueuing {} quotes...",
 				msg.0.len()
 			);
 
+			// We retrieve the connection object ahead of time to error-out early
+			// in case of connection issues and avoid the overhead of fetching
+			// a connection every iteration of the loop.
+			let mut con = self.con().await.map_err(|_| 0usize)?;
+
 			for (idx, q) in msg.0.iter().enumerate() {
-				let _: () = self
-					.con
+				let _: () = con
 					.rpush(&self.queue_key, q.get())
 					.await
 					// If something goes wrong, we return the number of quotes
@@ -337,10 +354,7 @@ Ok(())
 		}
 	}
 
-	impl<R> Message<DequeueQuote> for RedisQueueStorage<R>
-	where
-		R: redis::aio::ConnectionLike + redis::AsyncCommands + 'static,
-	{
+	impl Message<DequeueQuote> for RedisQueueStorage {
 		type Reply = DequeueReply;
 
 		#[tracing::instrument(name = "redis::dequeue_one", skip(self, _ctx, _msg))]
@@ -350,8 +364,12 @@ _msg: DequeueQuote,
 			_ctx: &mut Context<Self, Self::Reply>,
 		) -> Self::Reply {
 			tracing::debug!("Dequeuing quote...");
-			let quote: Result<Option<String>, ()> =
-				self.con.lpop(&self.queue_key, None).await.map_err(|e| {
+			let quote: Result<Option<String>, ()> = self
+				.con()
+				.await?
+				.lpop(&self.queue_key, None)
+				.await
+				.map_err(|e| {
 					tracing::error!(error = e.to_string(), "Could not dequeue quote from Redis");
 					()
 				});

--
tangled.sh