From b53474f1d77796d856e9c194f95f27501b6e994e Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Tue, 13 Oct 2026 23:32:08 +0800 Subject: [PATCH 1/3] Add a durable SpanOp channel with independent route heads --- bt-daemon/Cargo.lock | 74 ++++- bt-daemon/Cargo.toml | 1 + bt-daemon/src/pipeline/journal.rs | 504 ++++++++++++++++++++++++++++++ bt-daemon/src/pipeline/mod.rs | 4 + 4 files changed, 580 insertions(+), 3 deletions(-) create mode 100644 bt-daemon/src/pipeline/journal.rs diff --git a/bt-daemon/Cargo.lock b/bt-daemon/Cargo.lock index fe6c4914..33c8c740 100644 --- a/bt-daemon/Cargo.lock +++ b/bt-daemon/Cargo.lock @@ -283,6 +283,7 @@ dependencies = [ "reqwest", "rquickjs", "rquickjs-serde", + "rusqlite", "semver", "serde", "serde_json", @@ -595,6 +596,18 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "fallible-iterator" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2acce4a10f12dc2fb14a218589d4f1f62ef011b2d0cc4b3cb1bba8e94da14649" + +[[package]] +name = "fallible-streaming-iterator" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7360491ce676a36bf9bb3c56c1aa791658183a54d2744120f27285738d90465a" + [[package]] name = "fastrand" version = "2.5.0" @@ -613,6 +626,12 @@ version = "1.0.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" +[[package]] +name = "foldhash" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" + [[package]] name = "foldhash" version = "0.2.0" @@ -782,6 +801,15 @@ dependencies = [ "tracing", ] +[[package]] +name = "hashbrown" +version = "0.15.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" +dependencies = [ + "foldhash 0.1.5", +] + [[package]] name = "hashbrown" version = "0.17.1" @@ -790,7 +818,16 @@ checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" dependencies = [ "allocator-api2", "equivalent", - "foldhash", + "foldhash 0.2.0", +] + +[[package]] +name = "hashlink" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7382cf6263419f2d8df38c55d7da83da5c18aef87fc7a7fc1fb1e344edfe14c1" +dependencies = [ + "hashbrown 0.15.5", ] [[package]] @@ -1051,7 +1088,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9" dependencies = [ "equivalent", - "hashbrown", + "hashbrown 0.17.1", ] [[package]] @@ -1114,6 +1151,17 @@ version = "0.2.189" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" +[[package]] +name = "libsqlite3-sys" +version = "0.35.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "133c182a6a2c87864fe97778797e46c7e999672690dc9fa3ee8e241aa4a9c13f" +dependencies = [ + "cc", + "pkg-config", + "vcpkg", +] + [[package]] name = "linux-raw-sys" version = "0.12.1" @@ -1573,7 +1621,7 @@ version = "0.12.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "16e4f499ac5b943d97ee6dbc44f23c2c10426f420f7d2f1793d6318911b6608c" dependencies = [ - "hashbrown", + "hashbrown 0.17.1", "relative-path", "rquickjs-sys", ] @@ -1597,6 +1645,20 @@ dependencies = [ "cc", ] +[[package]] +name = "rusqlite" +version = "0.37.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "165ca6e57b20e1351573e3729b958bc62f0e48025386970b6e4d29e7a7e71f3f" +dependencies = [ + "bitflags", + "fallible-iterator", + "fallible-streaming-iterator", + "hashlink", + "libsqlite3-sys", + "smallvec", +] + [[package]] name = "rustc-hash" version = "2.1.3" @@ -2252,6 +2314,12 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" +[[package]] +name = "vcpkg" +version = "0.2.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426" + [[package]] name = "version_check" version = "0.9.5" diff --git a/bt-daemon/Cargo.toml b/bt-daemon/Cargo.toml index b058b2bf..73902695 100644 --- a/bt-daemon/Cargo.toml +++ b/bt-daemon/Cargo.toml @@ -44,6 +44,7 @@ tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "io-uti tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } uuid = { version = "1", features = ["v4", "v5"] } +rusqlite = { version = "0.37", features = ["bundled"] } [target.'cfg(windows)'.dependencies] windows-sys = { version = "0.61", features = ["Win32_Foundation", "Win32_Security", "Win32_Security_Authorization", "Win32_System_Threading"] } diff --git a/bt-daemon/src/pipeline/journal.rs b/bt-daemon/src/pipeline/journal.rs new file mode 100644 index 00000000..f20fefae --- /dev/null +++ b/bt-daemon/src/pipeline/journal.rs @@ -0,0 +1,504 @@ +//! Concrete durable channel: one ordered SpanOp log and a head per route. +use super::{Completion, Delivery}; +use crate::{ + translate::{OriginSnapshot, SpanOp}, + wire::SessionRoute, +}; +use rusqlite::{params, Connection, OptionalExtension}; +use serde::{Deserialize, Serialize}; +use std::{ + path::Path, + sync::{Arc, Mutex}, +}; +use tokio::sync::watch; + +#[derive(Clone)] +pub(crate) struct SpanJournal { + connection: Arc>, + changed: watch::Sender, +} + +/// Explicit codec preserves fields intentionally omitted from wire/plugin JSON. +#[derive(Serialize, Deserialize)] +struct StoredOp { + version: u32, + op: SpanOp, + late_merge_key: Option, + turn_span_id: Option, + is_turn: bool, +} + +impl StoredOp { + fn encode(op: &SpanOp) -> anyhow::Result> { + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op; + Ok(serde_json::to_vec(&Self { + version: 1, + op: op.clone(), + late_merge_key: row.late_merge_key.clone(), + turn_span_id: row.turn_span_id.clone(), + is_turn: row.is_turn, + })?) + } + fn decode(bytes: &[u8]) -> anyhow::Result { + let mut stored: Self = serde_json::from_slice(bytes)?; + anyhow::ensure!( + stored.version == 1, + "unsupported SpanOp codec {}", + stored.version + ); + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = &mut stored.op; + row.late_merge_key = stored.late_merge_key; + row.turn_span_id = stored.turn_span_id; + row.is_turn = stored.is_turn; + Ok(stored.op) + } +} + +impl SpanJournal { + pub async fn open(path: &Path) -> anyhow::Result { + let path = path.to_owned(); + let connection = tokio::task::spawn_blocking(move || -> anyhow::Result { + std::fs::create_dir_all(path.parent().expect("journal parent"))?; + let db = Connection::open(path)?; + db.busy_timeout(std::time::Duration::from_secs(5))?; + let version: u32 = db.query_row("PRAGMA user_version", [], |row| row.get(0))?; + anyhow::ensure!(version <= 1, "unsupported SpanOp journal schema {version}"); + db.execute_batch( + "PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL; + CREATE TABLE IF NOT EXISTS batches ( + input INTEGER PRIMARY KEY, committed INTEGER NOT NULL DEFAULT 0); + CREATE TABLE IF NOT EXISTS ops ( + position INTEGER PRIMARY KEY AUTOINCREMENT, + input INTEGER NOT NULL, value BLOB NOT NULL); + CREATE INDEX IF NOT EXISTS ops_input ON ops(input); + CREATE TABLE IF NOT EXISTS heads ( + route TEXT PRIMARY KEY, position INTEGER NOT NULL DEFAULT 0, + failure TEXT); + PRAGMA user_version=1;", + )?; + Ok(db) + }) + .await??; + let (changed, _) = watch::channel(0); + Ok(Self { + connection: Arc::new(Mutex::new(connection)), + changed, + }) + } + + async fn with_db( + &self, + operation: impl FnOnce(&mut Connection) -> anyhow::Result + Send + 'static, + ) -> anyhow::Result { + let db = self.connection.clone(); + tokio::task::spawn_blocking(move || { + let mut db = db + .lock() + .map_err(|_| anyhow::anyhow!("SpanOp storage lock poisoned"))?; + operation(&mut db) + }) + .await? + } + + pub async fn translated_through(&self) -> anyhow::Result { + self.with_db(|db| { + Ok(db.query_row( + "SELECT coalesce(max(input),0) FROM batches WHERE committed=1", + [], + |row| row.get(0), + )?) + }) + .await + } + + /// Restart abandons only invisible chunks; a committed identity is never rewritten. + pub async fn begin(&self, input: u64) -> anyhow::Result { + self.with_db(move |db| { + let tx = db.transaction()?; + let done: Option = tx + .query_row( + "SELECT committed FROM batches WHERE input=?", + [input], + |row| row.get(0), + ) + .optional()?; + if done == Some(true) { + return Ok(false); + } + tx.execute("DELETE FROM ops WHERE input=?", [input])?; + tx.execute( + "INSERT OR REPLACE INTO batches(input,committed) VALUES (?,0)", + [input], + )?; + tx.commit()?; + Ok(true) + }) + .await + } + + pub async fn append(&self, input: u64, ops: &[SpanOp]) -> anyhow::Result<()> { + let encoded = ops + .iter() + .map(StoredOp::encode) + .collect::>>()?; + self.with_db(move |db| { + let tx = db.transaction()?; + let done: bool = tx.query_row( + "SELECT committed FROM batches WHERE input=?", + [input], + |r| r.get(0), + )?; + anyhow::ensure!(!done, "cannot change a committed translation input"); + for value in encoded { + tx.execute( + "INSERT INTO ops(input,value) VALUES (?,?)", + params![input, value], + )?; + } + tx.commit()?; + Ok(()) + }) + .await + } + + pub async fn commit(&self, input: u64) -> anyhow::Result { + let tail = self + .with_db(move |db| { + let tx = db.transaction()?; + let previous: u64 = tx.query_row( + "SELECT coalesce(max(input),0) FROM batches WHERE committed=1", + [], + |r| r.get(0), + )?; + anyhow::ensure!(input >= previous, "translation commits must be ordered"); + anyhow::ensure!( + tx.execute("UPDATE batches SET committed=1 WHERE input=?", [input])? == 1, + "missing translation batch" + ); + let tail = tx.query_row( + "SELECT coalesce(max(position),0) FROM ops WHERE input<=?", + [input], + |r| r.get(0), + )?; + tx.commit()?; + Ok(tail) + }) + .await?; + self.changed + .send_modify(|version| *version = version.wrapping_add(1)); + Ok(tail) + } + + pub async fn subscribe( + &self, + route: &SessionRoute, + legacy_through: u64, + ) -> anyhow::Result { + let route = serde_json::to_string(route)?; + let key = route.clone(); + let position = self + .with_db(move |db| { + let tx = db.transaction()?; + // Seed only a new head, using released event checkpoints where provable. + let legacy: u64 = tx.query_row( + "SELECT coalesce(max(position),0) FROM ops JOIN batches USING(input) + WHERE input<=? AND committed=1", + [legacy_through], + |r| r.get(0), + )?; + tx.execute( + "INSERT OR IGNORE INTO heads(route,position) VALUES (?,?)", + params![key, legacy], + )?; + let position = + tx.query_row("SELECT position FROM heads WHERE route=?", [key], |r| { + r.get(0) + })?; + tx.commit()?; + Ok(position) + }) + .await?; + Ok(Cursor { + journal: self.clone(), + route, + read: position, + completed: position, + changes: self.changed.subscribe(), + }) + } + + pub async fn tail(&self) -> anyhow::Result { + self.with_db(|db| { + Ok(db.query_row( + "SELECT coalesce(max(position),0) FROM ops JOIN batches USING(input) + WHERE committed=1", + [], + |r| r.get(0), + )?) + }) + .await + } + + pub async fn failure(&self, route: &str) -> anyhow::Result> { + let route = route.to_owned(); + self.with_db(move |db| { + Ok( + db.query_row("SELECT failure FROM heads WHERE route=?", [route], |r| { + r.get(0) + })?, + ) + }) + .await + } + + pub async fn set_failure(&self, route: &str, failure: Option) -> anyhow::Result<()> { + let route = route.to_owned(); + self.with_db(move |db| { + db.execute( + "UPDATE heads SET failure=? WHERE route=?", + params![failure, route], + )?; + Ok(()) + }) + .await + } +} + +pub(crate) struct Cursor { + journal: SpanJournal, + pub route: String, + read: u64, + completed: u64, + changes: watch::Receiver, +} + +impl Cursor { + pub async fn next_batch( + &mut self, + limit: usize, + ) -> anyhow::Result>>> { + anyhow::ensure!(limit > 0, "cursor batch limit must be nonzero"); + let after = self.read; + let rows = self + .journal + .with_db(move |db| { + let mut statement = db.prepare( + "SELECT position,value FROM ops JOIN batches USING(input) + WHERE committed=1 AND position>? ORDER BY position LIMIT ?", + )?; + let rows = statement.query_map(params![after, limit as u64], |r| { + Ok((r.get::<_, u64>(0)?, r.get::<_, Vec>(1)?)) + })?; + Ok(rows.collect::, _>>()?) + }) + .await?; + if rows.is_empty() { + return Ok(None); + } + let through = rows.last().expect("nonempty rows").0; + let ops = rows + .into_iter() + .map(|(_, bytes)| StoredOp::decode(&bytes)) + .collect::>>()?; + self.read = through; + Ok(Some(Delivery::new( + ops, + HeadCompletion { + journal: self.journal.clone(), + route: self.route.clone(), + expected: after, + through, + }, + ))) + } + + pub async fn rewind(&mut self) -> anyhow::Result<()> { + let key = self.route.clone(); + self.completed = self + .journal + .with_db(move |db| { + Ok( + db.query_row("SELECT position FROM heads WHERE route=?", [key], |r| { + r.get(0) + })?, + ) + }) + .await?; + self.read = self.completed; + Ok(()) + } + + pub async fn changed(&mut self) { + let _ = self.changes.changed().await; + } +} + +struct HeadCompletion { + journal: SpanJournal, + route: String, + expected: u64, + through: u64, +} +#[async_trait::async_trait] +impl Completion for HeadCompletion { + async fn complete(&mut self) -> anyhow::Result<()> { + let (route, expected, through) = (self.route.clone(), self.expected, self.through); + self.journal + .with_db(move |db| { + let tx = db.transaction()?; + let head: u64 = + tx.query_row("SELECT position FROM heads WHERE route=?", [&route], |r| { + r.get(0) + })?; + if head >= through { + return Ok(()); + } + anyhow::ensure!(head == expected, "completion cannot skip unresolved input"); + tx.execute( + "UPDATE heads SET position=?, failure=NULL WHERE route=?", + params![through, route], + )?; + tx.commit()?; + Ok(()) + }) + .await + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::translate::SpanRow; + + fn op(name: &str) -> SpanOp { + SpanOp::Insert(SpanRow { + span_id: name.into(), + name: name.into(), + late_merge_key: Some("late".into()), + turn_span_id: Some("turn".into()), + is_turn: true, + origin: Some(OriginSnapshot { + plugin_version: Some("capture".into()), + source_version: None, + bt_version: "test".into(), + }), + ..SpanRow::default() + }) + } + fn route(name: &str) -> SessionRoute { + SessionRoute { + tags: vec![name.into()], + ..SessionRoute::default() + } + } + + #[tokio::test] + async fn invisible_chunks_empty_commit_and_restart_preserve_the_boundary() { + let tmp = tempfile::tempdir().unwrap(); + let path = tmp.path().join("spans.sqlite"); + let journal = SpanJournal::open(&path).await.unwrap(); + journal.begin(10).await.unwrap(); + journal.append(10, &[op("abandoned")]).await.unwrap(); + let mut reader = journal.subscribe(&route("a"), 0).await.unwrap(); + assert!(reader.next_batch(8).await.unwrap().is_none()); + drop(reader); + drop(journal); + let journal = SpanJournal::open(&path).await.unwrap(); + assert_eq!(journal.translated_through().await.unwrap(), 0); + journal.begin(10).await.unwrap(); + journal.append(10, &[op("retained")]).await.unwrap(); + journal.commit(10).await.unwrap(); + assert!(!journal.begin(10).await.unwrap()); + journal.begin(20).await.unwrap(); + journal.commit(20).await.unwrap(); + assert_eq!(journal.translated_through().await.unwrap(), 20); + let mut reader = journal.subscribe(&route("a"), 0).await.unwrap(); + let message = reader.next_batch(8).await.unwrap().unwrap(); + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = &message.value()[0]; + assert_eq!(row.name, "retained"); + assert_eq!(row.late_merge_key.as_deref(), Some("late")); + assert!(row.is_turn); + assert_eq!( + row.origin.as_ref().unwrap().plugin_version.as_deref(), + Some("capture") + ); + } + + #[tokio::test] + async fn independent_heads_and_dropped_deliveries_resume_without_skipping() { + let tmp = tempfile::tempdir().unwrap(); + let path = tmp.path().join("spans.sqlite"); + let journal = SpanJournal::open(&path).await.unwrap(); + journal.begin(10).await.unwrap(); + journal.append(10, &[op("one"), op("two")]).await.unwrap(); + journal.commit(10).await.unwrap(); + let mut a = journal.subscribe(&route("a"), 0).await.unwrap(); + let mut b = journal.subscribe(&route("b"), 0).await.unwrap(); + a.next_batch(2) + .await + .unwrap() + .unwrap() + .into_completed() + .await + .ok() + .unwrap(); + drop(b.next_batch(2).await.unwrap().unwrap()); + a.rewind().await.unwrap(); + b.rewind().await.unwrap(); + assert!(a.next_batch(2).await.unwrap().is_none()); + assert_eq!(b.next_batch(2).await.unwrap().unwrap().value().len(), 2); + let mut c = journal.subscribe(&route("c"), 0).await.unwrap(); + let first = c.next_batch(1).await.unwrap().unwrap(); + let second = c.next_batch(1).await.unwrap().unwrap(); + assert!( + second.into_completed().await.is_err(), + "later work cannot release an earlier obligation" + ); + first.into_completed().await.ok().unwrap(); + c.rewind().await.unwrap(); + assert_eq!(c.next_batch(2).await.unwrap().unwrap().value().len(), 1); + } + #[tokio::test] + async fn head_write_failure_retains_work_and_commit_wakes_readers() { + let tmp = tempfile::tempdir().unwrap(); + let journal = SpanJournal::open(&tmp.path().join("spans.sqlite")) + .await + .unwrap(); + let mut reader = journal.subscribe(&route("a"), 0).await.unwrap(); + journal.begin(10).await.unwrap(); + journal.append(10, &[op("one")]).await.unwrap(); + journal.commit(10).await.unwrap(); + tokio::time::timeout(std::time::Duration::from_secs(1), reader.changed()) + .await + .unwrap(); + assert_eq!(journal.tail().await.unwrap(), 1); + journal + .set_failure(&reader.route, Some("held".into())) + .await + .unwrap(); + journal.with_db(|db| { + db.execute_batch("CREATE TRIGGER fail_head BEFORE UPDATE OF position ON heads BEGIN SELECT RAISE(FAIL,'injected head failure'); END;")?; + Ok(()) + }).await.unwrap(); + let pending = reader.next_batch(8).await.unwrap().unwrap(); + let failed = pending.into_completed().await.err().unwrap(); + assert!(failed.error.to_string().contains("injected head failure")); + assert_eq!( + journal.failure(&reader.route).await.unwrap().as_deref(), + Some("held") + ); + reader.rewind().await.unwrap(); + assert!(reader.next_batch(8).await.unwrap().is_some()); + journal + .with_db(|db| { + db.execute_batch("DROP TRIGGER fail_head;")?; + Ok(()) + }) + .await + .unwrap(); + failed.delivery.into_completed().await.ok().unwrap(); + assert!(journal.failure(&reader.route).await.unwrap().is_none()); + reader.rewind().await.unwrap(); + assert!(reader.next_batch(8).await.unwrap().is_none()); + } +} diff --git a/bt-daemon/src/pipeline/mod.rs b/bt-daemon/src/pipeline/mod.rs index 76d8ab52..e1e8579a 100644 --- a/bt-daemon/src/pipeline/mod.rs +++ b/bt-daemon/src/pipeline/mod.rs @@ -16,3 +16,7 @@ pub(crate) trait Stage: Send { async fn process(&mut self, input: &Self::Input) -> Self::Outcome; } + +// Storage is activated by the next stack layer after its crash contract is tested. +#[cfg(test)] +mod journal; From 01ca61e5f2b6a75623443c524c5e45e81367dec0 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Wed, 14 Oct 2026 01:29:54 +0800 Subject: [PATCH 2/3] Allow cold Windows console probe compilation to finish --- scripts/tests/windows-background-process.mjs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scripts/tests/windows-background-process.mjs b/scripts/tests/windows-background-process.mjs index 7f9b5fa7..59e38bd8 100755 --- a/scripts/tests/windows-background-process.mjs +++ b/scripts/tests/windows-background-process.mjs @@ -154,7 +154,7 @@ async function run() { const probe = join(directory, "console probe.exe") await execFileAsync("rustc", [ "--edition=2021", join(repoRoot, "scripts/tests/windows-console-probe.rs"), "-o", probe, - ], { windowsHide: true, timeout: 60_000 }) + ], { windowsHide: true, timeout: 180_000 }) let failure try { await execFileAsync(probe, ["--launch", directory, process.execPath, script, "--exercise", directory], { From a8f34a92a2b82162d9a9361bbd24d41687b013e5 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Fri, 16 Oct 2026 22:40:34 +0800 Subject: [PATCH 3/3] Remove journal path and cursor panic assumptions --- bt-daemon/src/pipeline/journal.rs | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/bt-daemon/src/pipeline/journal.rs b/bt-daemon/src/pipeline/journal.rs index f20fefae..7b48ab39 100644 --- a/bt-daemon/src/pipeline/journal.rs +++ b/bt-daemon/src/pipeline/journal.rs @@ -58,7 +58,10 @@ impl SpanJournal { pub async fn open(path: &Path) -> anyhow::Result { let path = path.to_owned(); let connection = tokio::task::spawn_blocking(move || -> anyhow::Result { - std::fs::create_dir_all(path.parent().expect("journal parent"))?; + let parent = path + .parent() + .ok_or_else(|| anyhow::anyhow!("SpanOp journal path has no parent"))?; + std::fs::create_dir_all(parent)?; let db = Connection::open(path)?; db.busy_timeout(std::time::Duration::from_secs(5))?; let version: u32 = db.query_row("PRAGMA user_version", [], |row| row.get(0))?; @@ -292,10 +295,9 @@ impl Cursor { Ok(rows.collect::, _>>()?) }) .await?; - if rows.is_empty() { + let Some(&(through, _)) = rows.last() else { return Ok(None); - } - let through = rows.last().expect("nonempty rows").0; + }; let ops = rows .into_iter() .map(|(_, bytes)| StoredOp::decode(&bytes))