Got this solved over on the rust-lang discourse. The solution was to use the async_stream crate. Ends up being a small amount of code too.

I’ve been looking into it a bit - not pinning per se but self referential. There’s a library called ouroboros that looks helpful. There’s even an example on github where someone uses rusqlite and ouroboros together.

So it seems like this should work:

#[self_referencing]
pub struct ZkNoteStream {
  conn: Connection,
  #[borrows(conn)]
  pstmt: rusqlite::Statement<'this>,
  #[borrows(mut pstmt)]
  #[covariant]
  rec_iter: rusqlite::Rows<'this>,
}

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

    Ok(
      ZkNoteStreamTryBuilder {
        conn: conn,
        pstmt_builder: |conn: &Connection| conn.prepare(sql.as_str()),
        rec_iter_builder: |pstmt: &mut rusqlite::Statement<'_>| {
          pstmt.query(rusqlite::params_from_iter(args.iter()))
        },
      }
      .try_build()?,
    )
  }
}

Unfortunately I get this:

error[E0597]: `pstmt` does not live long enough
   --> server-lib/src/search.rs:880:1
    |
880 | #[self_referencing]
    | ^^^^^^^^^^^^^^^^^^-
    | |                 |
    | |                 `pstmt` dropped here while still borrowed
    | |                 borrow might be used here, when `pstmt` is dropped and runs the `Drop` code for type `Statement`
    | borrowed value does not live long enough
    |
    = note: this error originates in the attribute macro `self_referencing` (in Nightly builds, run with -Z macro-backtrace for more info)

So close! But no cigar so far. No idea why its complaining.

I’m not quite ready to give up, but its not looking good. One of the rusqlite maintainers issued this haiku-like missive:

yeah you just have to collect
you can't return that as an iterator
it needs to borrow from the statement

Got to wondering how Vec does this interator-with-internal-state thing, and its with unsafe.

update from the maintainer on discord:

fundamentally you're asking for a self-referential type. e.g. one field borrows from another field of the same struct. cant  be done without  unsafe
very easy to have soundness holes even if you use unsafe

The problem is that the return from the actix handler has to be HttpResult::Ok().stream(my_znsstream). Anything else in the function will get dropped before the HttpResult. So either the pstmt is somehow embedded in my_znsstream, or it doesn’t compile… : (

Yep, gave that a try (I think!). Here’s that version.

pub struct ZkNoteStream<'a> {
  rec_iter: Box + 'a>,
}

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

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

    let bytes_iter = {
      let mut pstmt = conn.prepare(sql.as_str())?;
      let rec_iter = pstmt.query_map(rusqlite::params_from_iter(args.iter()), move |row| {
        let id = row.get(0)?;
        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: Vec::new(),
        })
      })?;

      let val_iter = rec_iter
        .filter_map(|x| x.ok())
        .map(|x| serde_json::to_value(x).map_err(|e| e.into()));

      val_iter
        .filter_map(|x: Result| x.ok())
        .map(|x| Bytes::from(x.to_string()))
    };

    Ok(ZkNoteStream {
      rec_iter: Box::new(bytes_iter),
    })
  }
}

impl<'a> Stream for ZkNoteStream<'a> {
  type Item = Bytes;

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

This gets two errors, one for the conn and one for the pstmt:

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

error[E0515]: cannot return value referencing function parameter `conn`
   --> server-lib/src/search.rs:181:5
    |
152 |         let mut pstmt = conn.prepare(sql.as_str())?;
    |                         ---- `conn` is borrowed here
...
181 | /     Ok(ZkNoteStream {
182 | |       rec_iter: Box::new(bytes_iter),
183 | |     })
    | |______^ returns a value referencing data owned by the current function

So close! The problem is that query_map doesn’t consume pstmt, it only references it.

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

(bytes_iter is rec_iter run through some map()s)

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<'a, T> {
  rec_iter: Box + 'a>,
}

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

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

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

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

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

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

  pub fn make_stream(
    &'a mut self,
    conn: &'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(&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::<'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: &Config,
  uid: i64,
  msg: &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(&conn, uid, &search)?;
      {
        // borrowed value of znsm doesn't live long enough!  wat do?
        let znsstream = &znsm.make_stream(&conn)?;
      }
      Err("wat".into())
    }
    wat => Err(format!("invalid 'what' code:'{}'", wat).into()),
  }
}