Reset cached statements after using them

Author: simolus3

.github/workflows/tests.yml

@@ -58,5 +58,5 @@ jobs:
         run: |
           cd dart
           dart pub get
-          dart test
+          dart --enable-experiment=native-assets test
           dart analyze

Cargo.lock

@@ -22,7 +22,6 @@ serde = { version = "1.0", default-features = false, features = ["alloc", "deriv
 const_format = "0.2.34"
 futures-lite = { version = "2.6.0", default-features = false, features = ["alloc"] }
 rustc-hash = { version = "2.1", default-features = false }
-streaming-iterator = { version = "0.1.9", default-features = false, features = ["alloc"] }
 
 [dependencies.uuid]
 version = "1.4.1"

crates/core/Cargo.toml

@@ -3,7 +3,6 @@ use core::{assert_matches::debug_assert_matches, fmt::Display};
 use alloc::{string::ToString, vec::Vec};
 use serde::Serialize;
 use sqlite_nostd::{self as sqlite, Connection, ManagedStmt, ResultCode};
-use streaming_iterator::StreamingIterator;
 
 use crate::{
     error::SQLiteError,
@@ -25,7 +24,7 @@ use super::{
 /// used frequently as an optimization, but we're not taking advantage of that yet.
 pub struct StorageAdapter {
     pub db: *mut sqlite::sqlite3,
-    progress_stmt: ManagedStmt,
+    pub progress_stmt: ManagedStmt,
     time_stmt: ManagedStmt,
 }
 
@@ -78,37 +77,22 @@ impl StorageAdapter {
         Ok(())
     }
 
-    pub fn local_progress(
-        &self,
-    ) -> Result<
-        impl StreamingIterator<Item = Result<PersistedBucketProgress, ResultCode>>,
-        ResultCode,
-    > {
-        self.progress_stmt.reset()?;
-
-        fn step(stmt: &ManagedStmt) -> Result<Option<PersistedBucketProgress>, ResultCode> {
-            if stmt.step()? == ResultCode::ROW {
-                let bucket = stmt.column_text(0)?;
-                let count_at_last = stmt.column_int64(1);
-                let count_since_last = stmt.column_int64(2);
-
-                return Ok(Some(PersistedBucketProgress {
-                    bucket,
-                    count_at_last,
-                    count_since_last,
-                }));
-            }
-
+    pub fn step_progress(&self) -> Result<Option<PersistedBucketProgress>, ResultCode> {
+        if self.progress_stmt.step()? == ResultCode::ROW {
+            let bucket = self.progress_stmt.column_text(0)?;
+            let count_at_last = self.progress_stmt.column_int64(1);
+            let count_since_last = self.progress_stmt.column_int64(2);
+
+            Ok(Some(PersistedBucketProgress {
+                bucket,
+                count_at_last,
+                count_since_last,
+            }))
+        } else {
+            // Done
+            self.progress_stmt.reset()?;
             Ok(None)
         }
-
-        Ok(streaming_iterator::from_fn(|| {
-            match step(&self.progress_stmt) {
-                Err(e) => Some(Err(e)),
-                Ok(Some(other)) => Some(Ok(other)),
-                Ok(None) => None,
-            }
-        }))
     }
 
     pub fn reset_progress(&self) -> Result<(), ResultCode> {
@@ -239,10 +223,11 @@ impl StorageAdapter {
     }
 
     pub fn now(&self) -> Result<Timestamp, ResultCode> {
-        self.time_stmt.reset()?;
         self.time_stmt.step()?;
+        let res = Timestamp(self.time_stmt.column_int64(0));
+        self.time_stmt.reset()?;
 
-        Ok(Timestamp(self.time_stmt.column_int64(0)))
+        Ok(res)
     }
 }
 

crates/core/src/sync/storage_adapter.rs

@@ -425,11 +425,10 @@ impl StreamingSyncIteration {
         &self,
         checkpoint: &OwnedCheckpoint,
     ) -> Result<SyncDownloadProgress, SQLiteError> {
-        let local_progress = self.adapter.local_progress()?;
         let SyncProgressFromCheckpoint {
             progress,
             needs_counter_reset,
-        } = SyncDownloadProgress::for_checkpoint(checkpoint, local_progress)?;
+        } = SyncDownloadProgress::for_checkpoint(checkpoint, &self.adapter)?;
 
         if needs_counter_reset {
             self.adapter.reset_progress()?;

crates/core/src/sync/streaming_sync.rs

@@ -3,11 +3,12 @@ use core::{cell::RefCell, hash::BuildHasher};
 use rustc_hash::FxBuildHasher;
 use serde::Serialize;
 use sqlite_nostd::ResultCode;
-use streaming_iterator::StreamingIterator;
+
+use crate::sync::storage_adapter::StorageAdapter;
 
 use super::{
     bucket_priority::BucketPriority, interface::Instruction, line::DataLine,
-    storage_adapter::PersistedBucketProgress, streaming_sync::OwnedCheckpoint,
+    streaming_sync::OwnedCheckpoint,
 };
 
 /// Information about a progressing download.
@@ -187,9 +188,7 @@ pub struct SyncProgressFromCheckpoint {
 impl SyncDownloadProgress {
     pub fn for_checkpoint<'a>(
         checkpoint: &OwnedCheckpoint,
-        mut local_progress: impl StreamingIterator<
-            Item = Result<PersistedBucketProgress<'a>, ResultCode>,
-        >,
+        adapter: &StorageAdapter,
     ) -> Result<SyncProgressFromCheckpoint, ResultCode> {
         let mut buckets = BTreeMap::<String, BucketProgress>::new();
         let mut needs_reset = false;
@@ -206,12 +205,11 @@ impl SyncDownloadProgress {
             );
         }
 
-        while let Some(row) = local_progress.next() {
-            let row = match row {
-                Ok(row) => row,
-                Err(e) => return Err(*e),
-            };
+        // Go through local bucket states to detect pending progress from previous sync iterations
+        // that may have been interrupted.
+        adapter.progress_stmt.reset()?;
 
+        while let Some(row) = adapter.step_progress()? {
             let Some(progress) = buckets.get_mut(row.bucket) else {
                 continue;
             };
@@ -232,6 +230,8 @@ impl SyncDownloadProgress {
             }
         }
 
+        adapter.progress_stmt.reset()?;
+
         Ok(SyncProgressFromCheckpoint {
             progress: Self { buckets },
             needs_counter_reset: needs_reset,

crates/core/src/sync/sync_status.rs

@@ -13,10 +13,18 @@ packages:
     dependency: transitive
     description:
       name: analyzer
-      sha256: "13c1e6c6fd460522ea840abec3f677cc226f5fec7872c04ad7b425517ccf54f7"
+      sha256: "904ae5bb474d32c38fb9482e2d925d5454cda04ddd0e55d2e6826bc72f6ba8c0"
       url: "https://pub.dev"
     source: hosted
-    version: "7.4.4"
+    version: "7.4.5"
+  archive:
+    dependency: transitive
+    description:
+      name: archive
+      sha256: "2fde1607386ab523f7a36bb3e7edb43bd58e6edaf2ffb29d8a6d578b297fdbbd"
+      url: "https://pub.dev"
+    source: hosted
+    version: "4.0.7"
   args:
     dependency: transitive
     description:
@@ -45,10 +53,10 @@ packages:
     dependency: "direct main"
     description:
       name: bson
-      sha256: "9b761248a3494fea594aecf5d6f369b5f04d7b082aa2b8c06579ade77f1a7e47"
+      sha256: f8c80be7a62a88f4add7c48cc83567c36a77532de107224df8328ef71f125045
       url: "https://pub.dev"
     source: hosted
-    version: "5.0.6"
+    version: "5.0.7"
   cli_config:
     dependency: transitive
     description:
@@ -85,10 +93,10 @@ packages:
     dependency: transitive
     description:
       name: coverage
-      sha256: "9086475ef2da7102a0c0a4e37e1e30707e7fb7b6d28c209f559a9c5f8ce42016"
+      sha256: aa07dbe5f2294c827b7edb9a87bba44a9c15a3cc81bc8da2ca19b37322d30080
       url: "https://pub.dev"
     source: hosted
-    version: "1.12.0"
+    version: "1.14.1"
   crypto:
     dependency: transitive
     description:
@@ -101,10 +109,10 @@ packages:
     dependency: transitive
     description:
       name: decimal
-      sha256: "28239b8b929c1bd8618702e6dbc96e2618cf99770bbe9cb040d6cf56a11e4ec3"
+      sha256: "6c2041df7caefc9393ae0b0dcc4abc700831014a2c252dd10e3952499673f0b2"
       url: "https://pub.dev"
     source: hosted
-    version: "3.2.1"
+    version: "3.2.2"
   fake_async:
     dependency: "direct dev"
     description:
@@ -153,6 +161,14 @@ packages:
       url: "https://pub.dev"
     source: hosted
     version: "2.1.3"
+  http:
+    dependency: transitive
+    description:
+      name: http
+      sha256: "2c11f3f94c687ee9bad77c171151672986360b2b001d109814ee7140b2cf261b"
+      url: "https://pub.dev"
+    source: hosted
+    version: "1.4.0"
   http_multi_server:
     dependency: transitive
     description:
@@ -213,10 +229,10 @@ packages:
     dependency: "direct dev"
     description:
       name: meta
-      sha256: e3641ec5d63ebf0d9b41bd43201a66e3fc79a65db5f61fc181f04cd27aab950c
+      sha256: "23f08335362185a5ea2ad3a4e597f1375e78bce8a040df5c600c8d3552ef2394"
       url: "https://pub.dev"
     source: hosted
-    version: "1.16.0"
+    version: "1.17.0"
   mime:
     dependency: transitive
     description:
@@ -225,6 +241,22 @@ packages:
       url: "https://pub.dev"
     source: hosted
     version: "2.0.0"
+  native_assets_cli:
+    dependency: transitive
+    description:
+      name: native_assets_cli
+      sha256: "298ba641e47b44f039601141ab26f6865186b5d05b7f313d45797513d7205b3d"
+      url: "https://pub.dev"
+    source: hosted
+    version: "0.14.0"
+  native_toolchain_c:
+    dependency: transitive
+    description:
+      name: native_toolchain_c
+      sha256: e5b14abd4309f820817a48701dfd40d1fbd80a1d5fa2ce6838b758e952ea46d9
+      url: "https://pub.dev"
+    source: hosted
+    version: "0.11.0"
   node_preamble:
     dependency: transitive
     description:
@@ -265,6 +297,14 @@ packages:
       url: "https://pub.dev"
     source: hosted
     version: "1.5.1"
+  posix:
+    dependency: transitive
+    description:
+      name: posix
+      sha256: f0d7856b6ca1887cfa6d1d394056a296ae33489db914e365e2044fdada449e62
+      url: "https://pub.dev"
+    source: hosted
+    version: "6.0.2"
   power_extensions:
     dependency: transitive
     description:
@@ -361,6 +401,14 @@ packages:
       url: "https://pub.dev"
     source: hosted
     version: "2.7.6"
+  sqlite3_native_assets:
+    dependency: "direct main"
+    description:
+      name: sqlite3_native_assets
+      sha256: "0b379f37a642b20905e86fba43940bb429e44bed1d0fc6149076090815d450a7"
+      url: "https://pub.dev"
+    source: hosted
+    version: "0.0.4"
   sqlite3_test:
     dependency: "direct dev"
     description:
@@ -405,26 +453,26 @@ packages:
     dependency: "direct dev"
     description:
       name: test
-      sha256: "301b213cd241ca982e9ba50266bd3f5bd1ea33f1455554c5abb85d1be0e2d87e"
+      sha256: "65e29d831719be0591f7b3b1a32a3cda258ec98c58c7b25f7b84241bc31215bb"
       url: "https://pub.dev"
     source: hosted
-    version: "1.25.15"
+    version: "1.26.2"
   test_api:
     dependency: transitive
     description:
       name: test_api
-      sha256: fb31f383e2ee25fbbfe06b40fe21e1e458d14080e3c67e7ba0acfde4df4e0bbd
+      sha256: "522f00f556e73044315fa4585ec3270f1808a4b186c936e612cab0b565ff1e00"
       url: "https://pub.dev"
     source: hosted
-    version: "0.7.4"
+    version: "0.7.6"
   test_core:
     dependency: transitive
     description:
       name: test_core
-      sha256: "84d17c3486c8dfdbe5e12a50c8ae176d15e2a771b96909a9442b40173649ccaa"
+      sha256: "80bf5a02b60af04b09e14f6fe68b921aad119493e26e490deaca5993fef1b05a"
       url: "https://pub.dev"
     source: hosted
-    version: "0.6.8"
+    version: "0.6.11"
   typed_data:
     dependency: transitive
     description:
@@ -445,18 +493,18 @@ packages:
     dependency: transitive
     description:
       name: vm_service
-      sha256: ddfa8d30d89985b96407efce8acbdd124701f96741f2d981ca860662f1c0dc02
+      sha256: "45caa6c5917fa127b5dbcfbd1fa60b14e583afdc08bfc96dda38886ca252eb60"
       url: "https://pub.dev"
     source: hosted
-    version: "15.0.0"
+    version: "15.0.2"
   watcher:
     dependency: transitive
     description:
       name: watcher
-      sha256: "69da27e49efa56a15f8afe8f4438c4ec02eff0a117df1b22ea4aad194fe1c104"
+      sha256: "0b7fd4a0bbc4b92641dbf20adfd7e3fd1398fe17102d94b674234563e110088a"
       url: "https://pub.dev"
     source: hosted
-    version: "1.1.1"
+    version: "1.1.2"
   web:
     dependency: transitive
     description:
@@ -469,10 +517,10 @@ packages:
     dependency: transitive
     description:
       name: web_socket
-      sha256: bfe6f435f6ec49cb6c01da1e275ae4228719e59a6b067048c51e72d9d63bcc4b
+      sha256: "34d64019aa8e36bf9842ac014bb5d2f5586ca73df5e4d9bf5c936975cae6982c"
       url: "https://pub.dev"
     source: hosted
-    version: "1.0.0"
+    version: "1.0.1"
   web_socket_channel:
     dependency: transitive
     description: