Difficulty with trait function lifetimes

I'm using rust 1.75 and trying to refactor my code to extract some functions to async traits.

One function in particular is causing me lifetime problems and I can't understand why. Its particularity is that it returns a Stream.

A simplified version of the code without using traits is:

Playground: https://play.rust-lang.org/?version=stable&mode=debug&edition=2021&gist=3a56b296ee15402f7fe5e8c74019e585

use futures::{stream, Stream, StreamExt, TryStream};
use std::{convert::Infallible, sync::Arc, time::Duration};

#[tokio::main]
async fn main() {
    // My real scenario uses axum crate and share the object using Arc
    // Here is just to simulate something similar
    let db = Arc::new(MyDatabase {});
    let _ = consume_stream(db.clone()).await;
}

pub struct MyDatabase {}

impl MyDatabase {
    async fn load_events(&self) -> impl Stream<Item = String> {
        // Do some database call here
        tokio::time::sleep(Duration::from_millis(100)).await;

        stream::unfold(0, |count| async move {
            // There's no reference to the database object inside returned the stream
            if count == 10 {
                return None;
            }

            Some((count.to_string(), count + 1))
        })
    }
}

pub struct Consumer<S> {
    stream: S,
}

pub type BoxError = Box<dyn std::error::Error + Send + Sync>;

impl<S> Consumer<S> {
    // My real scenario uses axum Sse
    // This is the same implementation
    pub fn new(stream: S) -> Self
    where
        S: TryStream<Ok = String> + Send + 'static,
        S::Error: Into<BoxError>,
    {
        Consumer { stream }
    }
}

async fn consume_stream(
    db: Arc<MyDatabase>,
) -> Consumer<impl Stream<Item = Result<String, Infallible>>> {
    let stream = db
        .load_events()
        .await
        .map(|e| -> Result<String, Infallible> { Ok(e) });

    Consumer::new(stream)
}

Without using traits the code works fine.
But after extracting the function to a trait, the code no longer compiles with lifetime problems.

Playground: https://play.rust-lang.org/?version=stable&mode=debug&edition=2021&gist=28cc7fb5990aa00a9879dad1a79b2832

use futures::{stream, Stream, StreamExt, TryStream};
use std::{convert::Infallible, sync::Arc, time::Duration};

#[tokio::main]
async fn main() {
    // My real scenario uses axum crate and share the object using Arc
    // Here is just to simulate something similar
    let db = Arc::new(MyDatabase {});
    let _ = consume_stream(db.clone()).await;
}

trait Database {
    async fn load_events(&self) -> impl Stream<Item = String>;
}

pub struct MyDatabase {}

impl Database for MyDatabase {
    async fn load_events(&self) -> impl Stream<Item = String> {
        // Do some database call here
        tokio::time::sleep(Duration::from_millis(100)).await;

        stream::unfold(0, |count| async move {
            // There's no reference to the database object inside returned the stream
            if count == 10 {
                return None;
            }

            Some((count.to_string(), count + 1))
        })
    }
}

pub struct Consumer<S> {
    stream: S,
}

pub type BoxError = Box<dyn std::error::Error + Send + Sync>;

impl<S> Consumer<S> {
    // My real scenario uses axum Sse
    // This is the same implementation
    pub fn new(stream: S) -> Self
    where
        S: TryStream<Ok = String> + Send + 'static,
        S::Error: Into<BoxError>,
    {
        Consumer { stream }
    }
}

async fn consume_stream(
    db: Arc<MyDatabase>,
) -> Consumer<impl Stream<Item = Result<String, Infallible>>> {
    /*
error[E0597]: `db` does not live long enough
  --> src/main.rs:73:18
   |
53 |       db: Arc<MyDatabase>,
   |       -- binding `db` declared here
...
73 |       let stream = db
   |                    -^
   |                    |
   |  __________________borrowed value does not live long enough
   | |
74 | |         .load_events()
   | |______________________- argument requires that `db` is borrowed for `'static`
...
79 |   }
   |   - `db` dropped here while still borrowed
    */
    let stream = db
        .load_events()
        .await
        .map(|e| -> Result<String, Infallible> { Ok(e) });

    Consumer::new(stream)
}

I'm a beginner in rust and I'm having difficulties with lifetimes. Could you help me understand what happens after moving the function to a trait, and how to solve it?

Thanks!

async fn and RPITIT (return position impl trait in traits ) capture input lifetimes of the function/method, but RPIT outside of traits does not. On stable, you can do this, with some cost/loss of ergonomics/loss of auto-trait flexibility:

trait Database {
    async fn load_events(&self) -> Pin<Box<dyn Stream<Item = String> + Send>>;
}

On nightly, you can use TAITIT or whatever they're calling it instead.


The current plan is for RPIT outside of traits to start capturing lifetimes in the next edition, which will break your current non-trait code. An analogous workaround is also planned, which is to be more explicit about what is captured using a TAIT (type alias impl trait). Here is your non-trait code with a TAIT on edition 2024. The Send bound was also required, so the fix is even less trivial than typical. I suggest mentioning your use case in the tracking issue for the lifetime capture change.

1 Like

Thanks for the explanation and solutions!
It was exactly what I was looking for.