Ed: solved with the help of the async_stream crate.

I’m struggling with the borrow checker!

My problem: I’m using actix-web and rusqlite. I want to return an unlimited number of records from an rusqlite query, and actix provides a Stream trait for that kind of thing. You just impl the trait and return your records from a poll_next() fn.

On the rusqlite side, there’s this query_map that returns an iterator of records from a query. All I have to do is smush these two features together.

So the plan is to put the iterator returned by query_map into a struct that impls Stream. Problem is the lifetime of a var used by query_map. How to make the var have the same lifetime as the iterator??

So here’s the code:

pub struct ZkNoteStream<'a, T> {
  rec_iter: Box<dyn Iterator<Item = T> + 'a>,
}

// impl of Stream just calls next() on the iterator.  This compiles fine.
impl<'a> Stream for ZkNoteStream<'a, serde_json::Value> {
  type Item = serde_json::Value;

  fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
    Poll::Ready(self.rec_iter.next())
  }
}

// init function to set up the ZkNoteStream.
impl<'a> ZkNoteStream<'a, Result<ZkListNote, rusqlite::Error>> {
  pub fn init(
    conn: &'a Connection,
    user: i64,
    search: &ZkNoteSearch,
  ) -> Result<Self, Box<dyn Error>> {
    let (sql, args) = build_sql(&conn, user, search.clone())?;

    let sysid = user_id(&conn, "system")?;
    let mut pstmt = conn.prepare(sql.as_str())?;

    // Here's the problem!  Borrowing pstmt.
    let rec_iter = pstmt.query_map(rusqlite::params_from_iter(args.iter()), move |row| {
      let id = row.get(0)?;
      let sysids = get_sysids(&conn, sysid, id)?;
      Ok(ZkListNote {
        id: id,
        title: row.get(1)?,
        is_file: {
          let wat: Option<i64> = row.get(2)?;
          wat.is_some()
        },
        user: row.get(3)?,
        createdate: row.get(4)?,
        changeddate: row.get(5)?,
        sysids: sysids,
      })
    })?;

    Ok(ZkNoteStream::<Result<ZkListNote, rusqlite::Error>> {
      rec_iter: Box::new(rec_iter),
    })
  }
}

And here’s the error:

error[E0515]: cannot return value referencing local variable `pstmt`
   --> server-lib/src/search.rs:170:5
    |
153 |       let rec_iter = pstmt.query_map(rusqlite::params_from_iter(args.iter()), move |row| {
    |                      ----- `pstmt` is borrowed here
...
170 | /     Ok(ZkNoteStream::<Result<ZkListNote, rusqlite::Error>> {
171 | |       rec_iter: Box::new(rec_iter),
172 | |     })
    | |______^ returns a value referencing data owned by the current function

So basically it boils down to pstmt getting borrowed in the query_map call. It needs to have the same lifetime as the closure. How do I ensure that?

  •  hallettj   ( @hallettj@beehaw.org ) English
    1•7 months ago

    Well I’m curious to know what solutions there are to this. But I did reduce your problem to a smaller, self-contained example:

    struct ZKNoteStream&lt;'a, T> {
    rec_iter: Box + 'a>,
}

struct BorrowedThing(Vec);

impl BorrowedThing {
    fn prepare(&amp;self) -> IterProducer&lt;'_> {
        IterProducer(&amp;self.0)
    }
}

struct IterProducer&lt;'a>(&amp;'a Vec);

impl&lt;'a> IterProducer&lt;'a> {
    fn query_map(&amp;self) -> impl Iterator {
        self.0.into_iter()
    }
}

fn test(conn: &amp;BorrowedThing) -> ZKNoteStream&lt;'_, &amp;String> {
    let pstmt = conn.prepare();
    let rec_iter = pstmt.query_map();

    ZKNoteStream { // cannot return value referencing local variable `pstmt`
        rec_iter: Box::new(rec_iter),
    }
}

    Edit: Wow, that code block came out mangled. Here is a Rust Playground link

    I ran into a similar problem yesterday, but it was a case where I had control over all the code. What I did was the equivalent of changing query_map so that instead of taking a reference to its receiver, it takes ownership.

    My thinking at the moment is that you may need to modify ZKNoteStream so that instead of containing the resulting iterator it contains pstmt, maybe paired with a function takes pstmt as an argument and returns the iterator. It seems like you should be able to create an FnOnce closure with a move keyword that moves ownership of pstmt into that closure, which you could then store in ZKNoteStream. But I’m not able to make that happen right now.

    •  pr06lefs   ( @pr06lefs@lemmy.ml ) OP
      1•7 months ago

      UPDATE!

      I sort of solved this part of it, or at least got it to compile. I’ve got a reddit post of this too! Someone there hinted that I should use another struct ‘above’ ZkNoteStream. I’m doing that in the code listing below.

      ZnsMaker has an init() fn, then you call make_stream() and it returns a ZkNoteStream. The intent is ZnsMaker should be managed so it lasts as long as the ZkNoteStream needs to last. All this bit compiles, great! But when I go to use it in my actix handler, I get a borrowing problem there instead. So I may have just kicked the can down the road.

      This part compiles. Wrong types still, should produce Bytes instead of ZkListNotes.

      pub struct ZkNoteStream&lt;'a, T> {
  rec_iter: Box + 'a>,
}

impl&lt;'a> Stream for ZkNoteStream&lt;'a, Result> {
  type Item = Result;

  fn poll_next(mut self: Pin&lt;&amp;mut Self>, cx: &amp;mut Context&lt;'_>) -> Poll> {
    Poll::Ready(self.rec_iter.next())
  }
}

pub struct ZnsMaker&lt;'a> {
  pstmt: rusqlite::Statement&lt;'a>,
  sysid: i64,
  args: Vec,
}

impl&lt;'a> ZnsMaker&lt;'a> {
  pub fn init(
    conn: &amp;'a Connection,
    user: i64,
    search: &amp;ZkNoteSearch,
  ) -> Result> {
    let (sql, args) = build_sql(&amp;conn, user, search.clone())?;

    let sysid = user_id(&amp;conn, "system")?;

    Ok(ZnsMaker {
      args: args,
      sysid: sysid,
      pstmt: conn.prepare(sql.as_str())?,
    })
  }

  pub fn make_stream(
    &amp;'a mut self,
    conn: &amp;'a Connection,  // have to pass the connection in here instead of storing in ZnsMaker, for Reasons.
  ) -> Result>, rusqlite::Error> {
    let sysid = self.sysid;
    let rec_iter =
      self
        .pstmt
        .query_map(rusqlite::params_from_iter(self.args.iter()), move |row| {
          let id = row.get(0)?;
          let sysids = get_sysids(&amp;conn, sysid, id)?;
          Ok(ZkListNote {
            id: id,
            title: row.get(1)?,
            is_file: {
              let wat: Option = row.get(2)?;
              wat.is_some()
            },
            user: row.get(3)?,
            createdate: row.get(4)?,
            changeddate: row.get(5)?,
            sysids: sysids,
          })
        })?;

    Ok(ZkNoteStream::&lt;'a, Result> {
      rec_iter: Box::new(rec_iter),
    })
  }
}

      Ok and here’s the handler function where I receive a query and make the ZnsMaker. But if I create a ZkNoteStream with it, I get a borrowing error. Maybe it would be ok if I immediately consumed it in an HttpResponse::Ok().streaming(znsstream). Got to fix the types first though.

      pub async fn zk_interface_loggedin_streaming(
  config: &amp;Config,
  uid: i64,
  msg: &amp;UserMessage,
) -> Result> {
  match msg.what.as_str() {
    "searchzknotesstream" => {
      let msgdata = Option::ok_or(msg.data.as_ref(), "malformed json data")?;
      let search: ZkNoteSearch = serde_json::from_value(msgdata.clone())?;
      let conn = sqldata::connection_open(config.orgauth_config.db.as_path())?;
      let mut znsm = ZnsMaker::init(&amp;conn, uid, &amp;search)?;
      {
        // borrowed value of znsm doesn't live long enough!  wat do?
        let znsstream = &amp;znsm.make_stream(&amp;conn)?;
      }
      Err("wat".into())
    }
    wat => Err(format!("invalid 'what' code:'{}'", wat).into()),
  }
}