Requests hang indefinitely when tokio task dies

[banool]

When you build a pool you have to call init, which returns a JoinHandle. I await that task alongside all the other tasks of my service. When I run the service, I frequently have an issue where the tokio task dies with an error like this:

Failed to run cache worker: One of the tasks that weren't meant to end ended unexpectedly: Ok(
    Err(
        Error {
            context: "Redis pool task ended unexpectedly",
            source: Redis Error - kind: IO, details: Os { code: 61, kind: ConnectionRefused, message: "Connection refused" },
        },
    ),
)

Generally this only happens when I run my integration tests, which is a bit weird because I have limited them to run one at a time and I make sure to abort the Redis pool task at the end of each test.

This is all just context, the main problem is that when this happens, any operation involving the pool / a connection from the pool after that point just hangs forever. I'd expect it to instead return an error saying that the pool is gone.

I'm building my pool like this:

pub type MyRedisPool = WithOptions<RedisPool>;

#[derive(Clone, Debug, Deserialize)]
pub struct MyRedisConfig {
    /// Even though we support clustered Redis, we only accept a single URL because we
    /// assume that the caller will provide a single node in the cluster and the client
    /// will discover the rest.
    pub redis_url: RedisUrl,

    /// The size of the pool to create.
    pub redis_pool_size: usize,

    /// Set this to true if we're running against a Redis Cluster.
    pub redis_cluster: bool,
}

impl MyRedisConfig {
    pub fn from_url(redis_url: RedisUrl) -> Self {
        Self {
            redis_url,
            ..Default::default()
        }
    }

    /// Build a Redis pool. We call `init` here rather than letting the caller do it to make
    /// the client safer to use, since it won't work if you don't call init.
    ///
    /// We do not use `prefer_connected` on the advice of the creator of the crate:
    /// https://github.com/aembke/fred.rs/discussions/254
    ///
    /// Prefer not to use `.next` with the pool that comes from this, just use the pool
    /// directly, e.g. `pool.ping().await` is better than `pool.next().ping().await`.
    ///
    /// This also returns the JoinHandle for the task used by the client in the background.
    /// You should not throw this out, instead wait on it along with all other tasks.
    ///
    /// TODO: Resolve this: https://github.com/aembke/fred.rs/discussions/255
    pub async fn build_pool(&self) -> Result<(MyRedisPool, JoinHandle<Result<(), RedisError>>)> {
        let redis_url_str = self.redis_url.as_str();
        let config = if self.redis_cluster {
            RedisConfig::from_url_clustered(redis_url_str)
        } else {
            RedisConfig::from_url_centralized(redis_url_str)
        };
        let config = config.context("Failed to build Redis config")?;
        let pool = Builder::from_config(config)
            // The default ReconnectPolicy is to try to reconnect forever. We set out own
            // policy here to prevent that behavior.
            .set_policy(ReconnectPolicy::new_constant(5, 1500))
            // This configuration affects what happens for each connection / connection
            // attempt.
            .with_connection_config(|config| {
                config.connection_timeout = Duration::from_secs(5);
                config.unresponsive = UnresponsiveConfig {
                    max_timeout: Some(Duration::from_secs(5)),
                    ..Default::default()
                };
                config.replica = ReplicaConfig {
                    lazy_connections: true,
                    primary_fallback: true,
                    ..Default::default()
                };
            })
            // This enables otel tracing.
            .with_config(|config| {
                config.tracing.enabled = true;
                config.tracing.default_tracing_level = tracing::Level::INFO;
            })
            .build_pool(self.redis_pool_size)
            .context("Failed to create Redis connection pool")?;

        // These options apply to all requests made using a connection from the pool.
        // Setting with_options here is equivalent to doing it for each request.
        let options = Options {
            max_attempts: Some(3),
            timeout: Some(Duration::from_secs(4)),
            fail_fast: true,
            ..Default::default()
        };

        info!("Initializing and pinging Redis pool...");

        // Init the pool. You have to do this before you can use it. We use tokio::timeout
        // because `init` can hang (e.g. if Redis isn't there), and we don't want it to
        // just hang forever.
        let handle = tokio::time::timeout(Duration::from_secs(15), pool.init())
            .await
            .context("Failed to init Redis pool")??;

        // Ping to confirm that the pool is working.
        pool.ping().await.context("Failed to ping Redis")?;

        info!("Initialized and pinged Redis pool");

        Ok((pool.with_options(&options), handle))
    }
}

Unfortunately the project is closed source right now so I can't share it as-is, but I could try and put together a full repro.

I'm sort of losing my marbles a bit hahah, I was really looking forward to this library but so far I've just had a lot of problems with it, especially around hanging, even after setting all of these extra configs for connection / request timeouts. Managing the tokio task is also an extra pain point compared to deadpool, but that's an aside.

---

I dug a bit deeper and found the problem. In short, if I kill the Redis container and then try to do a read, it just hangs:

        println!("TRYING TO READ");
        let value: Option<String> = self
            .redis_pool
            // We don't care if the data from the cache reader is a bit stale, it is
            // a good candidate for using replicas instead. If there are no replicas,
            // the client will just use a normal node.
            .replicas()
            .get(redis_key)
            .await
            .inspect_err(|e| error!(error = ?e, key=redacted_key, "Failed to get key from Redis"))
            .with_context(|| format!("Failed to get key {}", redacted_key))?;
        println!("READ SUCCESS");

I see this order of output:

TRYING TO READ
thread 'tokio-runtime-worker' panicked at tests/common/mod.rs:201:32:
Failed to run cache worker: One of the tasks that weren't meant to end ended unexpectedly: Ok(
    Err(
        Error {
            context: "Redis pool task ended unexpectedly",
            source: Redis Error - kind: IO, details: Os { code: 61, kind: ConnectionRefused, message: "Connection refused" },
        },
    ),
)
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace
Gateway finished, did an error occur?

But we never see READ SUCCESS.

I'm building my pool using the above code. Given I set options like fail_fast, timeout, max_attempts, etc. I feel like it shouldn't hang. Better yet, I see that the tokio task for the pool dies in the background after Redis goes away, so I would've thought that getting a connection from the pool would instantly fail at that point.

[banool]

Ah I figured it out, when you use .replicas(), WithOptions is not transferred over. I opened an issue: #256.