diff --git a/.github/actions/xcframework/action.yml b/.github/actions/xcframework/action.yml index 6ec6afbc..21207846 100644 --- a/.github/actions/xcframework/action.yml +++ b/.github/actions/xcframework/action.yml @@ -24,8 +24,7 @@ runs: - name: Set up XCode uses: maxim-lobanov/setup-xcode@ed7a3b1fda3918c0306d1b724322adc0b8cc0a90 # v1.7.0 with: - # TODO: Update to latest-stable once GH installs iOS 26 simulators - xcode-version: '^16.4.0' + xcode-version: latest-stable - name: List simulators shell: bash diff --git a/crates/core/src/crud_vtab.rs b/crates/core/src/crud_vtab.rs index 4c0f4070..80a79ebd 100644 --- a/crates/core/src/crud_vtab.rs +++ b/crates/core/src/crud_vtab.rs @@ -15,6 +15,10 @@ use crate::error::PowerSyncError; use crate::ext::SafeManagedStmt; use crate::schema::TableInfoFlags; use crate::state::DatabaseState; +use crate::sync::storage_adapter::{ + LAST_APPLIED_CHECKPOINT_REQUEST_ID_KEY, LAST_SEEN_CHECKPOINT_REQUEST_ID_KEY, + TARGET_CHECKPOINT_REQUEST_ID_KEY, +}; use crate::utils::MAX_OP_ID; use crate::vtab_util::*; @@ -29,7 +33,7 @@ const SIMPLE_NAME: &CStr = c"powersync_crud"; // ps_crud(tx_id, data). // The second form (without the trailing underscore) takes the data to insert as individual // components and constructs the data to insert into `ps_crud` internally. It will also update -// `ps_updated_rows` and the `$local` bucket. +// `ps_updated_rows` and the local write apply gate. // // Using a virtual table like this allows us to hook into xBegin, xCommit and xRollback to automatically // increment transaction ids. These are only called when powersync_crud_ is used as part of a transaction, @@ -166,7 +170,7 @@ impl VirtualTable { stmt.bind_text(2, &serialized, sqlite::Destructor::STATIC)?; stmt.exec()?; - // However, we also set ps_updated_rows and update the $local bucket + // However, we also set ps_updated_rows and mark the local write state. let set_updated_rows = simple.set_updated_rows_statement(db)?; set_updated_rows.bind_text(1, row_type, sqlite::Destructor::STATIC)?; set_updated_rows.bind_text(2, id, sqlite::Destructor::STATIC)?; @@ -247,7 +251,14 @@ impl SimpleCrudTransactionMode { fn record_local_write(&mut self, db: *mut sqlite::sqlite3) -> Result<(), ResultCode> { if !self.had_writes { - db.exec_safe(formatcp!("INSERT OR REPLACE INTO ps_buckets(name, last_op, target_op) VALUES('$local', 0, {MAX_OP_ID})"))?; + // Also clear the seen/applied high-water marks: checkpoint request ids observed before + // this write can't acknowledge it, and stale values may predate a request counter + // restart. Keeping them around could open the apply gate for a newly allocated target + // id that compares below a stale seen value. + db.exec_safe(formatcp!( + "INSERT OR REPLACE INTO ps_kv(key, value) VALUES('{TARGET_CHECKPOINT_REQUEST_ID_KEY}', {MAX_OP_ID}); +DELETE FROM ps_kv WHERE key IN ('{LAST_SEEN_CHECKPOINT_REQUEST_ID_KEY}', '{LAST_APPLIED_CHECKPOINT_REQUEST_ID_KEY}')" + ))?; self.had_writes = true; } diff --git a/crates/core/src/migrations.rs b/crates/core/src/migrations.rs index 06f22146..87c56cb9 100644 --- a/crates/core/src/migrations.rs +++ b/crates/core/src/migrations.rs @@ -16,7 +16,7 @@ use crate::fix_data::apply_v035_fix; use crate::schema::inspection::ExistingView; use crate::sync::BucketPriority; -pub const LATEST_VERSION: i32 = 13; +pub const LATEST_VERSION: i32 = 14; pub fn powersync_migrate( ctx: *mut sqlite::context, @@ -460,6 +460,148 @@ DROP TABLE ps_sync_state_old; track_migration.exec()?; } + if current_version < 14 && target_version >= 14 { + // Move the legacy `$local` checkpoint bookkeeping into ps_kv. + // + // In older databases, `$local.last_applied_op` represented the latest legacy write + // checkpoint that was actually applied, so it becomes the last applied checkpoint request. + // `$local.last_op` represented the latest legacy write checkpoint seen in the sync stream, + // so it becomes the last seen checkpoint request. + // + // `$local.target_op` can either be a concrete legacy write checkpoint id or a sentinel such + // as i64::MAX while local writes are pending. Store it separately as `target_checkpoint_request_id`. + // Seeding `last_requested_checkpoint_request_id` from a concrete target would be possible, + // but should be redundant because SDKs reconcile the request counter with service state on + // connect before advancing it through `next_checkpoint_request_id`. + // + // An absent local target can safely start client-created checkpoint requests from 1. The + // ambiguous case is an existing max-op local target without a concrete requested id: + // pending local writes may already be associated with legacy service-created write + // checkpoints, so SDKs should bridge once through the legacy endpoint before starting + // client-created checkpoint requests. + // + // This migration can also run on a database that was previously on version 14 and then + // downgraded: the down migration rebuilds the `$local` row from ps_kv but keeps the ps_kv + // keys around, and an older SDK may have advanced `$local` since. Clear the keys first so + // the `$local` row is the source of truth and the inserts below can't conflict. + // + // After copying, the `$local` row is deleted: version 14 tracks this state exclusively in + // ps_kv, so ps_buckets only contains real sync buckets. The down migration recreates the + // row from ps_kv when needed. + // + // DROP COLUMN requires SQLite 3.35+; the extension already refuses to load below + // MIN_SQLITE_VERSION_NUMBER (3.44), so this is safe in the up path. + let up = "\ +DELETE FROM ps_kv + WHERE key IN ( + 'last_applied_checkpoint_request_id', + 'last_seen_checkpoint_request_id', + 'target_checkpoint_request_id' + ); + +INSERT INTO ps_kv(key, value) +SELECT 'last_applied_checkpoint_request_id', last_applied_op + FROM ps_buckets + WHERE name = '$local' + AND last_applied_op > 0; + +INSERT INTO ps_kv(key, value) +SELECT 'last_seen_checkpoint_request_id', last_op + FROM ps_buckets + WHERE name = '$local' + AND last_op > 0; + +INSERT INTO ps_kv(key, value) +SELECT 'target_checkpoint_request_id', target_op + FROM ps_buckets + WHERE name = '$local' + AND target_op > 0; + +DELETE FROM ps_buckets WHERE name = '$local'; + +ALTER TABLE ps_buckets DROP COLUMN target_op; +"; + local_db.exec_safe(up).into_db_result(local_db)?; + + // Downgrading needs to rebuild the old `$local` row from the new ps_kv state so older SDKs + // can keep using their target-op based blocking behavior. In that model, `$local.last_op` + // tracked the latest seen legacy write checkpoint and was compared with `$local.target_op` + // to decide whether downloaded changes could be applied. The `$local.last_applied_op` + // value represented the checkpoint that had actually been applied locally. + // `$local.pending_delete = 1` marked this as a synthetic local-only bucket instead of a + // normal service bucket. Restore each old progress column from its matching ps_kv key. + // The 0 defaults cover a local target that exists before any checkpoint has been seen or + // applied. If `target_checkpoint_request_id` is absent, don't create a `$local` row: the old + // implementation also didn't have a `$local` bucket unless there was local target state to + // track. + const DOWN_STATEMENTS: &[&str] = &[ + "ALTER TABLE ps_buckets RENAME TO ps_buckets_14", + "DROP INDEX ps_buckets_name", + "CREATE TABLE ps_buckets( + id INTEGER PRIMARY KEY, + name TEXT NOT NULL, + last_applied_op INTEGER NOT NULL DEFAULT 0, + last_op INTEGER NOT NULL DEFAULT 0, + target_op INTEGER NOT NULL DEFAULT 0, + add_checksum INTEGER NOT NULL DEFAULT 0, + op_checksum INTEGER NOT NULL DEFAULT 0, + pending_delete INTEGER NOT NULL DEFAULT 0 +) STRICT", + "CREATE UNIQUE INDEX ps_buckets_name ON ps_buckets (name)", + "ALTER TABLE ps_buckets ADD COLUMN count_at_last INTEGER NOT NULL DEFAULT 0", + "ALTER TABLE ps_buckets ADD COLUMN count_since_last INTEGER NOT NULL DEFAULT 0", + "ALTER TABLE ps_buckets ADD COLUMN downloaded_size INTEGER NOT NULL DEFAULT 0", + "INSERT INTO ps_buckets( + id, + name, + last_applied_op, + last_op, + add_checksum, + op_checksum, + pending_delete, + count_at_last, + count_since_last, + downloaded_size +) +SELECT + id, + name, + last_applied_op, + last_op, + add_checksum, + op_checksum, + pending_delete, + count_at_last, + count_since_last, + downloaded_size +FROM ps_buckets_14", + "DROP TABLE ps_buckets_14", + "INSERT INTO ps_buckets(name, pending_delete, last_op, last_applied_op, target_op) +SELECT '$local', 1, seen, applied, target + FROM ( + SELECT + IFNULL((SELECT CAST(value AS INTEGER) FROM ps_kv WHERE key = 'last_seen_checkpoint_request_id'), 0) AS seen, + IFNULL((SELECT CAST(value AS INTEGER) FROM ps_kv WHERE key = 'last_applied_checkpoint_request_id'), 0) AS applied, + (SELECT CAST(value AS INTEGER) FROM ps_kv WHERE key = 'target_checkpoint_request_id') AS target + ) + WHERE EXISTS ( + SELECT 1 FROM ps_kv WHERE key = 'target_checkpoint_request_id' + ) +ON CONFLICT(name) DO UPDATE SET + pending_delete = excluded.pending_delete, + last_op = excluded.last_op, + last_applied_op = excluded.last_applied_op, + target_op = excluded.target_op", + "DELETE FROM ps_migration WHERE id >= 14", + ]; + let down = serialize_down_statements(DOWN_STATEMENTS)?; + let track_migration = + local_db.prepare_v2("INSERT INTO ps_migration(id, down_migrations) VALUES (?, ?)")?; + track_migration.bind_int(1, 14)?; + track_migration.bind_text(2, &down, Destructor::STATIC)?; + track_migration.exec()?; + } + Ok(()) } diff --git a/crates/core/src/schema/raw_table.rs b/crates/core/src/schema/raw_table.rs index ee06501b..23144ca7 100644 --- a/crates/core/src/schema/raw_table.rs +++ b/crates/core/src/schema/raw_table.rs @@ -267,8 +267,7 @@ pub fn generate_raw_table_trigger( // Prevent illegal writes to a table marked as insert-only by raising errors here. buffer.push_str("SELECT RAISE(FAIL, 'Unexpected update on insert-only table');\n"); } else { - // Write directly to powersync_crud_ to skip writing the $local bucket for insert-only - // tables. + // Insert-only tables use manual CRUD writes so they don't block incoming data. let fragment = table_columns_to_json_object("NEW", &as_schema_table)?; buffer.powersync_crud_manual_put(&table.name, &fragment); } diff --git a/crates/core/src/sync/interface.rs b/crates/core/src/sync/interface.rs index e09eba94..b85d0994 100644 --- a/crates/core/src/sync/interface.rs +++ b/crates/core/src/sync/interface.rs @@ -12,6 +12,7 @@ use crate::sync::diagnostics::{DiagnosticOptions, DiagnosticsEvent}; use crate::sync::subscriptions::{StreamKey, apply_subscriptions}; use alloc::borrow::Cow; use alloc::boxed::Box; +use alloc::format; use alloc::rc::Rc; use alloc::{string::String, vec::Vec}; use powersync_sqlite_nostd::bindings::SQLITE_RESULT_SUBTYPE; @@ -128,21 +129,33 @@ pub enum Instruction { }, /// Connect to the sync service using the [StreamingSyncRequest] created by the core extension, /// and then forward received lines via [SyncEvent::TextLine] and [SyncEvent::BinaryLine]. - EstablishSyncStream { request: StreamingSyncRequest }, + EstablishSyncStream { + request: StreamingSyncRequest, + /// The latest checkpoint request id known locally before opening this stream. + /// + /// This is simply the client's current counter state. SDKs use it on every connection + /// attempt to re-affirm checkpoint request state with the service, which may have deleted + /// its record. The re-affirmation works bidirectionally: it can restore the service-side + /// value from this hint or bump the local counter from the service's response, which is + /// reported back with `seed_checkpoint_request_id`. + last_checkpoint_request_id: Option, + }, FetchCredentials { /// Whether the credentials currently used have expired. /// /// If false, this is a pre-fetch. did_expire: bool, }, - // These are defined like this because deserializers in Kotlin can't support either an - // object or a literal value /// Close the websocket / HTTP stream to the sync service. CloseSyncStream(CloseSyncStream), /// Flush the file-system if it's non-durable (only applicable to the Dart SDK). FlushFileSystem {}, /// Notify that a sync has been completed, prompting client SDKs to clear earlier errors. - DidCompleteSync {}, + DidCompleteSync { + /// The checkpoint request id applied by this completed sync, if the checkpoint had one. + #[serde(skip_serializing_if = "Option::is_none")] + applied_checkpoint_request_id: Option, + }, /// Handle a diagnostic event. /// @@ -232,6 +245,77 @@ pub fn register(db: *mut sqlite::sqlite3, state: Rc) -> Result<() } }), "stop" => SyncControlRequest::StopSyncStream, + "seed_checkpoint_request_id" => { + let has_sync_iteration = { + let client = state.sync_client.borrow(); + client + .as_ref() + .map(|client| client.has_sync_iteration()) + .unwrap_or(false) + }; + + if !has_sync_iteration { + return Err(PowerSyncError::state_error("No iteration is active")); + } + + let request_id = parse_positive_i64_payload( + *payload, + "checkpoint request id", + "checkpoint request id must be an integer or integer string", + )?; + let adapter = state.storage_adapter(db)?; + adapter.seed_checkpoint_request_id(request_id)?; + ctx.result_int64(request_id); + return Ok(()); + } + "next_checkpoint_request_id" => { + let has_sync_iteration = { + let client = state.sync_client.borrow(); + client + .as_ref() + .map(|client| client.has_sync_iteration()) + .unwrap_or(false) + }; + + if !has_sync_iteration { + return Err(PowerSyncError::state_error("No iteration is active")); + } + + let adapter = state.storage_adapter(db)?; + if !adapter.has_checkpoint_request_id()? { + return Err(PowerSyncError::state_error( + "Checkpoint request state has not been seeded", + )); + } + + let request_id = adapter.next_checkpoint_request_id()?; + ctx.result_int64(request_id); + return Ok(()); + } + "target_checkpoint_request_id" => { + let target = parse_optional_i64_payload( + *payload, + "target checkpoint request id", + "target checkpoint request id must be an integer, integer string, or null", + )?; + let adapter = state.storage_adapter(db)?; + let previous_target = adapter.probe_target_checkpoint_request_id(target)?; + + match previous_target { + Some(target) => ctx.result_int64(target), + None => ctx.result_null(), + } + + return Ok(()); + } + "current_checkpoint_request_id" => { + let adapter = state.storage_adapter(db)?; + match adapter.last_checkpoint_request_id()? { + Some(request_id) => ctx.result_int64(request_id), + None => ctx.result_null(), + } + return Ok(()); + } "line_text" => SyncControlRequest::SyncEvent(SyncEvent::TextLine { data: if payload.value_type() == ColumnType::Text { payload.text() @@ -345,3 +429,47 @@ create_sqlite_text_fn!( powersync_offline_sync_status_impl, "powersync_offline_sync_status" ); + +fn parse_optional_i64_payload( + payload: *mut sqlite::value, + name: &'static str, + type_error: &'static str, +) -> Result, PowerSyncError> { + let value = match payload.value_type() { + ColumnType::Null => return Ok(None), + ColumnType::Integer => payload.int64(), + // Allow decimal strings as a fallback for JavaScript SQLite drivers that can't bind + // 64-bit integers as BigInt without losing precision through Number. + ColumnType::Text => payload + .text() + .parse::() + .map_err(|_| PowerSyncError::argument_error(type_error))?, + _ => return Err(PowerSyncError::argument_error(type_error)), + }; + + if value < 0 { + return Err(PowerSyncError::argument_error(format!( + "{name} must be a non-negative integer" + ))); + } + + Ok(Some(value)) +} + +fn parse_positive_i64_payload( + payload: *mut sqlite::value, + name: &'static str, + type_error: &'static str, +) -> Result { + let Some(value) = parse_optional_i64_payload(payload, name, type_error)? else { + return Err(PowerSyncError::argument_error(type_error)); + }; + + if value == 0 { + return Err(PowerSyncError::argument_error(format!( + "{name} must be a positive integer" + ))); + } + + Ok(value) +} diff --git a/crates/core/src/sync/storage_adapter.rs b/crates/core/src/sync/storage_adapter.rs index 7b5298f3..d78a30c8 100644 --- a/crates/core/src/sync/storage_adapter.rs +++ b/crates/core/src/sync/storage_adapter.rs @@ -26,6 +26,17 @@ use super::{ bucket_priority::BucketPriority, interface::BucketRequest, streaming_sync::OwnedCheckpoint, }; +pub(crate) const LAST_REQUESTED_CHECKPOINT_REQUEST_ID_KEY: &str = + "last_requested_checkpoint_request_id"; +pub(crate) const LAST_SEEN_CHECKPOINT_REQUEST_ID_KEY: &str = "last_seen_checkpoint_request_id"; +pub(crate) const LAST_APPLIED_CHECKPOINT_REQUEST_ID_KEY: &str = + "last_applied_checkpoint_request_id"; + +// Tracks the target used to block applying downloaded rows while local writes are outstanding. +// When present, this is normally either the max-op sentinel for pending local writes or a concrete +// checkpoint request id also stored in LAST_REQUESTED_CHECKPOINT_REQUEST_ID_KEY. +pub(crate) const TARGET_CHECKPOINT_REQUEST_ID_KEY: &str = "target_checkpoint_request_id"; + /// An adapter for storing sync state. /// /// This is used to encapsulate some SQL queries used for the sync implementation, making the code @@ -68,9 +79,10 @@ impl StorageAdapter { pub fn collect_bucket_requests(&self) -> Result, PowerSyncError> { // language=SQLite - let statement = self.db.prepare_v2( - "SELECT name, last_op FROM ps_buckets WHERE pending_delete = 0 AND name != '$local'", - ).into_db_result(self.db)?; + let statement = self + .db + .prepare_v2("SELECT name, last_op FROM ps_buckets WHERE pending_delete = 0") + .into_db_result(self.db)?; let mut requests = Vec::::new(); @@ -123,6 +135,8 @@ impl StorageAdapter { priority_status: priority_items, downloading: None, streams, + // Checkpoint requests should not be made or compared while offline. + internal_last_applied_checkpoint_request_id: None, }) } @@ -260,10 +274,8 @@ WHERE bucket = ?1", } } - if let (None, Some(write_checkpoint)) = (&priority, &checkpoint.write_checkpoint) { - update_bucket.bind_int64(1, *write_checkpoint)?; - update_bucket.bind_text(2, "$local", sqlite::Destructor::STATIC)?; - update_bucket.exec()?; + if let (None, Some(checkpoint_request_id)) = (&priority, &checkpoint.write_checkpoint) { + self.persist_last_seen_checkpoint_request_id(*checkpoint_request_id)?; } #[derive(Serialize)] @@ -499,23 +511,146 @@ WHERE bucket = ?1", Ok(()) } - pub fn local_state(&self) -> Result, PowerSyncError> { - let stmt = self + pub fn target_checkpoint_request_id(&self) -> Result, PowerSyncError> { + self.read_i64_kv(TARGET_CHECKPOINT_REQUEST_ID_KEY) + } + + /// Probes and optionally updates the target checkpoint request id used to block applying downloaded rows + /// while local writes are outstanding. + /// + /// In the write-checkpoint flow, callers allocate a checkpoint request id, post it to the + /// service, and then update this from the max-op sentinel to the concrete checkpoint request id + /// once the request succeeds. This is also used for older services where the SDK cannot create + /// checkpoint requests explicitly. + /// + /// The target can also be used internally as a sentinel value such as max op id while local + /// writes are pending, so it must not always be interpreted as a checkpoint request id. + /// + /// This only updates the apply gate. It does not allocate, seed or overwrite + /// `last_requested_checkpoint_request_id`, which is managed by `seed_checkpoint_request_id` and + /// `next_checkpoint_request_id`. + /// + /// Returns the target value from before this call. When `target` is `None`, this only + /// reads the current value. A `target` of zero clears the stored target, removing the apply + /// gate entirely; any other value overwrites it. + /// + /// Negative values are rejected when parsing the `powersync_control` payload, before this is + /// called. + pub fn probe_target_checkpoint_request_id( + &self, + target: Option, + ) -> Result, PowerSyncError> { + let previous_target = self.target_checkpoint_request_id()?; + + let Some(target) = target else { + return Ok(previous_target); + }; + + if target == 0 { + self.delete_kv(TARGET_CHECKPOINT_REQUEST_ID_KEY)?; + return Ok(previous_target); + } + + self.write_i64_kv(TARGET_CHECKPOINT_REQUEST_ID_KEY, target)?; + + Ok(previous_target) + } + + /// Persists the checkpoint request id observed in a complete sync checkpoint. + /// + /// This is used to decide whether downloaded data can be applied after local uploads complete. + pub fn persist_last_seen_checkpoint_request_id( + &self, + request_id: i64, + ) -> Result<(), PowerSyncError> { + self.write_i64_kv(LAST_SEEN_CHECKPOINT_REQUEST_ID_KEY, request_id) + } + + /// Persists the checkpoint request id that was applied locally. + /// + /// This is always the id from the last applied checkpoint as sent by the service - a plain + /// overwrite with no monotonicity enforced by core. External code owns consistency of these + /// ids; waiters should rely on `DidCompleteSync.applied_checkpoint_request_id` rather than + /// comparing this value across reconnects. + pub fn persist_last_applied_checkpoint_request_id( + &self, + request_id: i64, + ) -> Result<(), PowerSyncError> { + self.write_i64_kv(LAST_APPLIED_CHECKPOINT_REQUEST_ID_KEY, request_id) + } + + /// Increments, persists and returns the next client-created checkpoint request id. + pub fn next_checkpoint_request_id(&self) -> Result { + let statement = self.db.prepare_v2( + "INSERT INTO ps_kv(key, value) +VALUES(?1, 1) +ON CONFLICT(key) DO UPDATE SET value = CAST(value AS INTEGER) + 1 +RETURNING value", + )?; + statement.bind_text( + 1, + LAST_REQUESTED_CHECKPOINT_REQUEST_ID_KEY, + sqlite::Destructor::STATIC, + )?; + + if statement.step()? == ResultCode::ROW { + Ok(statement.column_int64(0)) + } else { + Err(PowerSyncError::unknown_internal()) + } + } + + /// Returns whether the local checkpoint request counter has been initialized. + pub fn has_checkpoint_request_id(&self) -> Result { + Ok(self.last_checkpoint_request_id()?.is_some()) + } + + /// Returns the latest checkpoint request id known locally. + pub fn last_checkpoint_request_id(&self) -> Result, PowerSyncError> { + self.read_i64_kv(LAST_REQUESTED_CHECKPOINT_REQUEST_ID_KEY) + } + + /// Seeds the local checkpoint request counter from service state. + /// + /// The value is stored verbatim: core does not enforce monotonicity here. SDKs are + /// responsible for reconciling their local hint with the service before seeding, and cannot + /// allocate new checkpoint request ids until that seeding has completed. + pub fn seed_checkpoint_request_id(&self, request_id: i64) -> Result<(), PowerSyncError> { + self.write_i64_kv(LAST_REQUESTED_CHECKPOINT_REQUEST_ID_KEY, request_id) + } + + fn read_i64_kv(&self, key: &'static str) -> Result, PowerSyncError> { + let statement = self .db - .prepare_v2("SELECT target_op FROM ps_buckets WHERE name = ?")?; - stmt.bind_text(1, "$local", sqlite::Destructor::STATIC)?; + .prepare_v2("SELECT value FROM ps_kv WHERE key = ?1") + .into_db_result(self.db)?; + statement.bind_text(1, key, sqlite::Destructor::STATIC)?; - Ok(if stmt.step()? == ResultCode::ROW { - let target_op = stmt.column_int64(0); - Some(LocalState { target_op }) + Ok(if statement.step()? == ResultCode::ROW { + Some(statement.column_int64(0)) } else { None }) } -} -pub struct LocalState { - pub target_op: i64, + fn write_i64_kv(&self, key: &'static str, value: i64) -> Result<(), PowerSyncError> { + let stmt = self.db.prepare_v2( + "INSERT INTO ps_kv(key, value) +VALUES(?1, ?2) +ON CONFLICT(key) DO UPDATE SET value = excluded.value", + )?; + stmt.bind_text(1, key, sqlite::Destructor::STATIC)?; + stmt.bind_int64(2, value)?; + stmt.exec()?; + Ok(()) + } + + fn delete_kv(&self, key: &'static str) -> Result<(), PowerSyncError> { + let stmt = self.db.prepare_v2("DELETE FROM ps_kv WHERE key = ?1")?; + stmt.bind_text(1, key, sqlite::Destructor::STATIC)?; + stmt.exec()?; + Ok(()) + } } pub struct BucketInfo { diff --git a/crates/core/src/sync/streaming_sync.rs b/crates/core/src/sync/streaming_sync.rs index 5c680ce3..381a097b 100644 --- a/crates/core/src/sync/streaming_sync.rs +++ b/crates/core/src/sync/streaming_sync.rs @@ -363,7 +363,16 @@ impl StreamingSyncIteration { line: "Validated and applied checkpoint".into(), }); event.instructions.push(Instruction::FlushFileSystem {}); + + // Persist here so that all database writes happen while preparing the + // transition, keeping apply_transition infallible. + if let Some(request_id) = target.write_checkpoint { + self.adapter + .persist_last_applied_checkpoint_request_id(request_id)?; + } + SyncStateMachineTransition::SyncLocalChangesApplied { + applied_checkpoint_request_id: target.write_checkpoint, partial: None, timestamp, } @@ -400,6 +409,9 @@ impl StreamingSyncIteration { } SyncLocalResult::ChangesApplied { timestamp } => { SyncStateMachineTransition::SyncLocalChangesApplied { + // A checkpoint request is only considered applied once the full + // checkpoint has been applied, not for partial completions. + applied_checkpoint_request_id: None, partial: Some(priority), timestamp, } @@ -497,7 +509,11 @@ impl StreamingSyncIteration { } => { self.validated_but_not_applied = Some(validated_but_not_applied); } - SyncStateMachineTransition::SyncLocalChangesApplied { partial, timestamp } => { + SyncStateMachineTransition::SyncLocalChangesApplied { + applied_checkpoint_request_id, + partial, + timestamp, + } => { if let Some(priority) = partial { self.status.update( |status| { @@ -506,7 +522,7 @@ impl StreamingSyncIteration { &mut event.instructions, ); } else { - self.handle_checkpoint_applied(event, timestamp); + self.handle_checkpoint_applied(event, timestamp, applied_checkpoint_request_id); } } SyncStateMachineTransition::Empty => {} @@ -631,7 +647,7 @@ impl StreamingSyncIteration { return Ok(()); }; - let target_write = self.adapter.local_state()?.map(|e| e.target_op); + let target_write = self.adapter.target_checkpoint_request_id()?; if checkpoint.write_checkpoint < target_write { // Note: None < Some(x). The pending checkpoint does not contain the write // checkpoint created during the upload, so we don't have to try applying it, it's @@ -647,7 +663,11 @@ impl StreamingSyncIteration { line: "Applied pending checkpoint after completed upload".into(), }); - self.handle_checkpoint_applied(event, timestamp); + if let Some(request_id) = checkpoint.write_checkpoint { + self.adapter + .persist_last_applied_checkpoint_request_id(request_id)?; + } + self.handle_checkpoint_applied(event, timestamp, checkpoint.write_checkpoint); } _ => { event.instructions.push(Instruction::LogLine { @@ -922,20 +942,40 @@ impl StreamingSyncIteration { app_metadata: self.options.app_metadata.take(), }; - event - .instructions - .push(Instruction::EstablishSyncStream { request }); + event.instructions.push(Instruction::EstablishSyncStream { + request, + last_checkpoint_request_id: self.adapter.last_checkpoint_request_id()?, + }); Ok(BeforeCheckpoint { local_buckets: local_bucket_names, stream_subscriptions: stream_subscriptions, }) } - fn handle_checkpoint_applied(&mut self, event: &mut ActiveEvent, timestamp: TimestampMicros) { - event.instructions.push(Instruction::DidCompleteSync {}); + /// Emits the instructions and status update for a fully applied checkpoint. + /// + /// The applied checkpoint request id must already have been persisted by the caller: this + /// runs while applying a state transition, which must stay infallible (see + /// [SyncStateMachineTransition]). + fn handle_checkpoint_applied( + &mut self, + event: &mut ActiveEvent, + timestamp: TimestampMicros, + applied_checkpoint_request_id: Option, + ) { + if let Some(request_id) = applied_checkpoint_request_id { + event.instructions.push(Instruction::LogLine { + severity: LogSeverity::DEBUG, + line: format!("Applied checkpoint request id {request_id}").into(), + }); + } + + event.instructions.push(Instruction::DidCompleteSync { + applied_checkpoint_request_id, + }); self.status.update( - |status| status.applied_checkpoint(timestamp), + |status| status.applied_checkpoint(timestamp, applied_checkpoint_request_id), &mut event.instructions, ); } @@ -1113,6 +1153,7 @@ enum SyncStateMachineTransition<'a> { validated_but_not_applied: OwnedCheckpoint, }, SyncLocalChangesApplied { + applied_checkpoint_request_id: Option, partial: Option, timestamp: TimestampMicros, }, diff --git a/crates/core/src/sync/sync_local.rs b/crates/core/src/sync/sync_local.rs index 1293ef33..d8770b91 100644 --- a/crates/core/src/sync/sync_local.rs +++ b/crates/core/src/sync/sync_local.rs @@ -12,8 +12,12 @@ use crate::schema::{ }; use crate::state::DatabaseState; use crate::sync::BucketPriority; +use crate::sync::storage_adapter::{ + LAST_SEEN_CHECKPOINT_REQUEST_ID_KEY, TARGET_CHECKPOINT_REQUEST_ID_KEY, +}; use crate::sync::sync_status::TimestampMicros; use crate::utils::SqlBuffer; +use const_format::formatcp; use powersync_sqlite_nostd::{self as sqlite, Destructor, ManagedStmt}; use powersync_sqlite_nostd::{Connection, ResultCode}; @@ -66,9 +70,13 @@ impl<'a> SyncOperation<'a> { if needs_check { // language=SQLite - let statement = self.db.prepare_v2( - "SELECT 1 FROM ps_buckets WHERE target_op > last_op AND name = '$local'", - )?; + let statement = self.db.prepare_v2(formatcp!( + "SELECT 1 +FROM ps_kv AS target +LEFT JOIN ps_kv AS seen ON seen.key = '{LAST_SEEN_CHECKPOINT_REQUEST_ID_KEY}' +WHERE target.key = '{TARGET_CHECKPOINT_REQUEST_ID_KEY}' + AND CAST(target.value AS INTEGER) > COALESCE(CAST(seen.value AS INTEGER), 0)" + ))?; if statement.step()? == ResultCode::ROW { return Ok(false); diff --git a/crates/core/src/sync/sync_status.rs b/crates/core/src/sync/sync_status.rs index eb7daeea..04568e47 100644 --- a/crates/core/src/sync/sync_status.rs +++ b/crates/core/src/sync/sync_status.rs @@ -53,6 +53,11 @@ pub struct DownloadSyncStatus { /// received), information about how far the download has progressed. pub downloading: Option, pub streams: Vec, + /// Runtime-only request id from the most recent applied checkpoint request. + /// + /// This is exposed in sync status for SDK internals, but it is not persisted and should not be + /// treated as user-facing download progress. + pub internal_last_applied_checkpoint_request_id: Option, } impl DownloadSyncStatus { @@ -67,12 +72,14 @@ impl DownloadSyncStatus { self.connected = false; self.connecting = false; self.downloading = None; + self.internal_last_applied_checkpoint_request_id = None; } pub fn start_connecting(&mut self) { self.connected = false; self.downloading = None; self.connecting = true; + self.internal_last_applied_checkpoint_request_id = None; self.debug_assert_priority_status_is_sorted(); } @@ -117,7 +124,11 @@ impl DownloadSyncStatus { self.debug_assert_priority_status_is_sorted(); } - pub fn applied_checkpoint(&mut self, now: TimestampMicros) { + pub fn applied_checkpoint( + &mut self, + now: TimestampMicros, + applied_checkpoint_request_id: Option, + ) { self.downloading = None; self.priority_status.clear(); @@ -126,6 +137,8 @@ impl DownloadSyncStatus { last_synced_at: Some(now), has_synced: Some(true), }); + + self.internal_last_applied_checkpoint_request_id = applied_checkpoint_request_id; } } @@ -137,6 +150,7 @@ impl Default for DownloadSyncStatus { downloading: None, priority_status: Vec::new(), streams: Vec::new(), + internal_last_applied_checkpoint_request_id: None, } } } @@ -180,12 +194,18 @@ impl Serialize for DownloadSyncStatus { } } - let mut serializer = serializer.serialize_struct("DownloadSyncStatus", 4)?; + let field_count = + 5 + usize::from(self.internal_last_applied_checkpoint_request_id.is_some()); + let mut serializer = serializer.serialize_struct("DownloadSyncStatus", field_count)?; serializer.serialize_field("connected", &self.connected)?; serializer.serialize_field("connecting", &self.connecting)?; serializer.serialize_field("priority_status", &self.priority_status)?; serializer.serialize_field("downloading", &self.downloading)?; serializer.serialize_field("streams", &SerializeStreamsWithProgress(self))?; + if let Some(request_id) = self.internal_last_applied_checkpoint_request_id { + serializer + .serialize_field("internal_last_applied_checkpoint_request_id", &request_id)?; + } serializer.end() } diff --git a/crates/core/src/view_admin.rs b/crates/core/src/view_admin.rs index 6b9496c5..3ef012b6 100644 --- a/crates/core/src/view_admin.rs +++ b/crates/core/src/view_admin.rs @@ -85,7 +85,6 @@ fn powersync_clear_impl( local_db.exec_safe("DELETE FROM ps_oplog; DELETE FROM ps_buckets")?; } else { trigger_resync(local_db, state)?; - local_db.exec_safe("DELETE FROM ps_buckets WHERE name = '$local'")?; } // language=SQLite @@ -164,7 +163,7 @@ fn trigger_resync(db: *mut sqlite::sqlite3, state: &DatabaseState) -> Result<(), } } - db.exec_safe("UPDATE ps_buckets SET last_applied_op = 0 WHERE name != '$local'") + db.exec_safe("UPDATE ps_buckets SET last_applied_op = 0") .into_db_result(db)?; Ok(Default::default()) } diff --git a/dart/benchmark/apply_lines.dart b/dart/benchmark/apply_lines.dart index ee0654b5..c06d361a 100644 --- a/dart/benchmark/apply_lines.dart +++ b/dart/benchmark/apply_lines.dart @@ -17,8 +17,10 @@ void main(List args) { final db = openTestDatabase(); db + ..execute('BEGIN') ..execute('select powersync_init()') - ..execute('select powersync_control(?, null)', ['start']); + ..execute('select powersync_control(?, null)', ['start']) + ..execute('COMMIT'); final stopwatch = Stopwatch()..start(); diff --git a/dart/test/crud_test.dart b/dart/test/crud_test.dart index 5d6d379b..74ca5743 100644 --- a/dart/test/crud_test.dart +++ b/dart/test/crud_test.dart @@ -249,7 +249,15 @@ void main() { }); }); - test('updates local bucket and updated rows', () { + test('updates target checkpoint request id and updated rows', () { + // Stale high-water marks (e.g. from before a request counter restart) must be cleared + // by a local write, so they can't open the apply gate for a smaller new target id. + db.execute(''' +INSERT INTO ps_kv(key, value) VALUES + ('last_seen_checkpoint_request_id', 6), + ('last_applied_checkpoint_request_id', 5); +'''); + db.execute( 'INSERT INTO powersync_crud (op, id, type, data) VALUES (?, ?, ?, ?)', [ @@ -262,12 +270,17 @@ void main() { expect(db.select('SELECT * FROM ps_updated_rows'), [ {'row_type': 'users', 'row_id': 'foo'} ]); - expect(db.select('SELECT * FROM ps_buckets'), [ - allOf( - containsPair('name', r'$local'), - containsPair('target_op', 9223372036854775807), - ) - ]); + expect(db.select(r"SELECT * FROM ps_buckets WHERE name = '$local'"), + isEmpty); + expect( + db.select( + "SELECT key, value FROM ps_kv WHERE key LIKE '%checkpoint_request_id'"), + [ + { + 'key': 'target_checkpoint_request_id', + 'value': 9223372036854775807, + } + ]); }); test('does not require data', () { diff --git a/dart/test/error_test.dart b/dart/test/error_test.dart index 74da34a6..7ac7c800 100644 --- a/dart/test/error_test.dart +++ b/dart/test/error_test.dart @@ -78,6 +78,7 @@ void main() { final control = db.prepare(stmt); control.execute(['start', null]); + control.execute(['seed_checkpoint_request_id', 1]); expect( () => control.execute(['line_text', 'invalid sync line']), throwsA(isSqliteException( diff --git a/dart/test/goldens/simple_iteration.json b/dart/test/goldens/simple_iteration.json index 32c1db89..d26c21bf 100644 --- a/dart/test/goldens/simple_iteration.json +++ b/dart/test/goldens/simple_iteration.json @@ -27,7 +27,8 @@ "include_defaults": true, "subscriptions": [] } - } + }, + "last_checkpoint_request_id": null } } ] @@ -173,4 +174,4 @@ } ] } -] \ No newline at end of file +] diff --git a/dart/test/goldens/starting_stream.json b/dart/test/goldens/starting_stream.json index 2549905a..987da69c 100644 --- a/dart/test/goldens/starting_stream.json +++ b/dart/test/goldens/starting_stream.json @@ -33,9 +33,10 @@ "include_defaults": true, "subscriptions": [] } - } + }, + "last_checkpoint_request_id": null } } ] } -] \ No newline at end of file +] diff --git a/dart/test/migration_test.dart b/dart/test/migration_test.dart index 9203f74f..ef9387cc 100644 --- a/dart/test/migration_test.dart +++ b/dart/test/migration_test.dart @@ -97,6 +97,144 @@ void main() { }); } + test('migrates local checkpoint state to ps_kv', () async { + db.execute(fixtures.expectedState[13]!); + db.execute(r''' +INSERT INTO ps_buckets(id, name, last_applied_op, last_op, target_op, add_checksum, op_checksum, pending_delete, count_at_last, count_since_last, downloaded_size) +VALUES(1, '$local', 5, 6, 7, 0, 0, 1, 0, 0, 0); +'''); + + db.executeInTx('select powersync_init()'); + + expect(db.select('SELECT key, value FROM ps_kv ORDER BY key'), [ + {'key': 'last_applied_checkpoint_request_id', 'value': 5}, + {'key': 'last_seen_checkpoint_request_id', 'value': 6}, + {'key': 'target_checkpoint_request_id', 'value': 7}, + ]); + expect(db.select(r"SELECT * FROM ps_buckets WHERE name = '$local'"), + isEmpty); + expect( + db.select("PRAGMA table_info('ps_buckets')").map((row) => row['name']), + isNot(contains('target_op')), + ); + expect( + () => db.execute( + r"UPDATE ps_buckets SET target_op = 8 WHERE name = '$local'"), + throwsA(isA()), + ); + }); + + test('does not migrate last applied op as requested checkpoint id', + () async { + db.execute(fixtures.expectedState[13]!); + // Simulate pending local writes during migration. The legacy target op is the max-op + // sentinel, while last_applied_op is the last write checkpoint applied locally. + db.execute(r''' +INSERT INTO ps_buckets(id, name, last_applied_op, last_op, target_op, add_checksum, op_checksum, pending_delete, count_at_last, count_since_last, downloaded_size) +VALUES(1, '$local', 5, 6, 9223372036854775807, 0, 0, 1, 0, 0, 0); +'''); + + db.executeInTx('select powersync_init()'); + + // last_applied_op becomes the applied checkpoint id, but it must not seed the requested + // checkpoint counter. The sentinel target is preserved for blocking, but target ops no longer + // seed last_requested_checkpoint_request_id. + expect(db.select('SELECT key, value FROM ps_kv ORDER BY key'), [ + {'key': 'last_applied_checkpoint_request_id', 'value': 5}, + {'key': 'last_seen_checkpoint_request_id', 'value': 6}, + {'key': 'target_checkpoint_request_id', 'value': 9223372036854775807}, + ]); + }); + + test('does not migrate sentinel target op as requested checkpoint id', + () async { + db.execute(fixtures.expectedState[13]!); + db.execute(r''' +INSERT INTO ps_buckets(id, name, last_applied_op, last_op, target_op, add_checksum, op_checksum, pending_delete, count_at_last, count_since_last, downloaded_size) +VALUES(1, '$local', 0, 0, 9223372036854775807, 0, 0, 1, 0, 0, 0); +'''); + + db.executeInTx('select powersync_init()'); + + // The max-op sentinel is valid local target state, but target ops no longer seed + // last_requested_checkpoint_request_id. + expect(db.select('SELECT key, value FROM ps_kv ORDER BY key'), [ + {'key': 'target_checkpoint_request_id', 'value': 9223372036854775807}, + ]); + }); + + test('restores local checkpoint state on downgrade', () async { + db.execute(fixtures.finalState); + db.execute(r''' +INSERT INTO ps_kv(key, value) VALUES + ('last_requested_checkpoint_request_id', 7), + ('last_seen_checkpoint_request_id', 6), + ('last_applied_checkpoint_request_id', 5), + ('target_checkpoint_request_id', 7); +'''); + + db.executeInTx('select powersync_test_migration(13)'); + + expect( + db.select( + r"SELECT pending_delete, last_op, last_applied_op, target_op FROM ps_buckets WHERE name = '$local'"), + [ + { + 'pending_delete': 1, + 'last_op': 6, + 'last_applied_op': 5, + 'target_op': 7, + } + ], + ); + }); + + test('re-upgrades after downgrade with checkpoint state', () async { + db.execute(fixtures.finalState); + db.execute(r''' +INSERT INTO ps_kv(key, value) VALUES + ('last_requested_checkpoint_request_id', 7), + ('last_seen_checkpoint_request_id', 6), + ('last_applied_checkpoint_request_id', 5), + ('target_checkpoint_request_id', 7); +'''); + + db.executeInTx('select powersync_test_migration(13)'); + + // Simulate an older SDK advancing the restored $local row while downgraded. + db.execute(r''' +UPDATE ps_buckets + SET last_op = 8, last_applied_op = 8, target_op = 9 + WHERE name = '$local' +'''); + + db.executeInTx('select powersync_init()'); + + // The $local row is the source of truth on re-upgrade; the request counter is unrelated to + // $local and survives the downgrade unchanged. + expect(db.select('SELECT key, value FROM ps_kv ORDER BY key'), [ + {'key': 'last_applied_checkpoint_request_id', 'value': 8}, + {'key': 'last_requested_checkpoint_request_id', 'value': 7}, + {'key': 'last_seen_checkpoint_request_id', 'value': 8}, + {'key': 'target_checkpoint_request_id', 'value': 9}, + ]); + expect(db.select(r"SELECT * FROM ps_buckets WHERE name = '$local'"), + isEmpty); + + final schema = '${getSchema(db)}\n${getMigrations(db)}'; + expect(schema, equals(fixtures.finalState.trim())); + }); + + test('does not restore local bucket without local target on downgrade', + () async { + db.execute(fixtures.finalState); + + db.executeInTx('select powersync_test_migration(13)'); + + expect(db.select(r"SELECT * FROM ps_buckets WHERE name = '$local'"), + isEmpty); + }); + /// Here we apply a developer schema _after_ migrating test('schema after migration', () async { db.execute(fixtures.expectedState[2]!); @@ -209,6 +347,7 @@ void main() { db.execute('begin'); control('start'); + control('seed_checkpoint_request_id', 1); control( 'line_text', json.encode(checkpoint(lastOpId: 3, buckets: [ diff --git a/dart/test/sync_local_performance_test.dart b/dart/test/sync_local_performance_test.dart index 92ba1c2b..ed5d5b5b 100644 --- a/dart/test/sync_local_performance_test.dart +++ b/dart/test/sync_local_performance_test.dart @@ -100,6 +100,7 @@ COMMIT; // Start a fake sync client to apply the changes we've already written to // ps_oplog control('start'); + control('seed_checkpoint_request_id', 1); final lastOpid = db.select('select max(op_id) from ps_oplog').single.columnAt(0) as int; final allBuckets = db diff --git a/dart/test/sync_stream_test.dart b/dart/test/sync_stream_test.dart index e5ba8ac1..b062d945 100644 --- a/dart/test/sync_stream_test.dart +++ b/dart/test/sync_stream_test.dart @@ -32,7 +32,7 @@ void main() { ['client_id', 'test-test-test-test']); }); - List control(String operation, Object? data) { + List controlRaw(String operation, Object? data) { db.execute('begin'); ResultSet result; @@ -60,6 +60,19 @@ void main() { } } + bool establishesSyncStream(List instructions) { + return instructions.any((instruction) => + instruction is Map && instruction.containsKey('EstablishSyncStream')); + } + + List control(String operation, Object? data) { + final result = controlRaw(operation, data); + if (operation == 'start' && establishesSyncStream(result)) { + controlRaw('seed_checkpoint_request_id', 1); + } + return result; + } + group('default streams', () { syncTest('are created on-demand', (_) { control('start', null); diff --git a/dart/test/sync_test.dart b/dart/test/sync_test.dart index 579ece80..3f3d0eb6 100644 --- a/dart/test/sync_test.dart +++ b/dart/test/sync_test.dart @@ -4,12 +4,12 @@ import 'dart:typed_data'; import 'package:bson/bson.dart'; import 'package:file/local.dart'; +import 'package:path/path.dart'; import 'package:sqlite3/common.dart'; import 'package:sqlite3/sqlite3.dart'; import 'package:sqlite3_test/sqlite3_test.dart'; import 'package:test/test.dart'; import 'package:test_descriptor/test_descriptor.dart' as d; -import 'package:path/path.dart'; import 'utils/native_test_utils.dart'; import 'utils/test_utils.dart'; @@ -60,16 +60,57 @@ void _syncTests({ db.execute('commit'); final [row] = result; - return jsonDecode(row.columnAt(0)); + final rawResult = row.columnAt(0); + if (rawResult is String) { + return jsonDecode(rawResult); + } else { + return const []; + } + } + + Object? invokeControlScalar(String operation, Object? data) { + db.execute('begin'); + ResultSet result; + + try { + result = db.select('SELECT powersync_control(?, ?)', [operation, data]); + + const statement = 'SELECT * FROM sqlite_stmt WHERE busy AND sql != ?;'; + final busy = db.select(statement, [statement]); + expect(busy, isEmpty); + } catch (e) { + db.execute('rollback'); + rethrow; + } + + db.execute('commit'); + final [row] = result; + return row.columnAt(0); + } + + int seedCheckpointRequestId(Object? requestId) { + return invokeControlScalar('seed_checkpoint_request_id', requestId) as int; + } + + bool establishesSyncStream(List instructions) { + return instructions.any((instruction) => + instruction is Map && instruction.containsKey('EstablishSyncStream')); } List invokeControl(String operation, Object? data) { + List result; if (matcher.enabled) { // Trace through golden matcher - return matcher.invoke(operation, data); + result = matcher.invoke(operation, data); } else { - return invokeControlRaw(operation, data); + result = invokeControlRaw(operation, data); } + + if (operation == 'start' && establishesSyncStream(result)) { + seedCheckpointRequestId(1); + } + + return result; } setUp(() async { @@ -135,6 +176,37 @@ void _syncTests({ return syncLine(checkpointComplete(priority: priority, lastOpId: lastOpId)); } + Object? lastRequestedCheckpointRequestId() { + final rows = db.select( + "SELECT value FROM ps_kv WHERE key = 'last_requested_checkpoint_request_id'"); + return rows.isEmpty ? null : rows.single.columnAt(0); + } + + Object? lastAppliedCheckpointRequestId() { + final rows = db.select( + "SELECT value FROM ps_kv WHERE key = 'last_applied_checkpoint_request_id'"); + return rows.isEmpty ? null : rows.single.columnAt(0); + } + + Object? streamLastCheckpointRequestId(List instructions) { + final instruction = instructions.whereType().firstWhere( + (instruction) => instruction.containsKey('EstablishSyncStream')); + final establish = instruction['EstablishSyncStream'] as Map; + return establish['last_checkpoint_request_id']; + } + + int nextCheckpointRequestId() { + return invokeControlScalar('next_checkpoint_request_id', null) as int; + } + + Object? probeTargetCheckpointRequestId([Object? opId]) { + return invokeControlScalar('target_checkpoint_request_id', opId); + } + + Object? currentCheckpointRequestId() { + return invokeControlScalar('current_checkpoint_request_id', null); + } + ResultSet fetchRows() { return db.select('select * from items'); } @@ -318,8 +390,16 @@ void _syncTests({ syncTest('remembers sync state', (controller) { invokeControl('start', null); - pushCheckpoint(buckets: priorityBuckets); - pushCheckpointComplete(); + pushCheckpoint(buckets: priorityBuckets, writeCheckpoint: '1'); + expect( + pushCheckpointComplete(), + contains( + containsPair( + 'DidCompleteSync', + {'applied_checkpoint_request_id': 1}, + ), + ), + ); controller.elapse(Duration(minutes: 10)); pushCheckpoint(buckets: priorityBuckets); @@ -329,28 +409,36 @@ void _syncTests({ final instructions = invokeControl('start', null); expect( instructions, - contains( - containsPair( + isNot(contains(containsPair( 'UpdateSyncStatus', containsPair( - 'status', + 'status', + containsPair( + 'internal_last_applied_checkpoint_request_id', anything))))), + ); + expect( + instructions, + contains( + containsPair( + 'UpdateSyncStatus', containsPair( - 'priority_status', - [ - { - 'priority': 2, - 'last_synced_at': timestamp(), - 'has_synced': true - }, - { - 'priority': 2147483647, - 'last_synced_at': timestamp(plusMinutes: -10), - 'has_synced': true - }, - ], - ), - ), - ), + 'status', + containsPair( + 'priority_status', + [ + { + 'priority': 2, + 'last_synced_at': timestamp(), + 'has_synced': true + }, + { + 'priority': 2147483647, + 'last_synced_at': timestamp(plusMinutes: -10), + 'has_synced': true + }, + ], + ), + )), ), ); @@ -371,6 +459,335 @@ void _syncTests({ }); }); + syncTest('allocates requested checkpoint request ids', (_) { + invokeControl('start', null); + + expect(nextCheckpointRequestId(), 2); + expect(lastRequestedCheckpointRequestId(), 2); + + expect(nextCheckpointRequestId(), 3); + expect(lastRequestedCheckpointRequestId(), 3); + }); + + syncTest('reports current checkpoint request id without incrementing', (_) { + expect(currentCheckpointRequestId(), isNull); + + invokeControlRaw('start', null); + expect(seedCheckpointRequestId(1), 1); + expect(currentCheckpointRequestId(), 1); + + expect(currentCheckpointRequestId(), 1); + expect(nextCheckpointRequestId(), 2); + expect(currentCheckpointRequestId(), 2); + expect(lastRequestedCheckpointRequestId(), 2); + + pushCheckpoint(buckets: priorityBuckets, writeCheckpoint: '2'); + pushCheckpointComplete(); + expect(lastAppliedCheckpointRequestId(), 2); + expect(currentCheckpointRequestId(), 2); + + db.execute("insert into items (id, col) values ('local', 'data');"); + expect(lastAppliedCheckpointRequestId(), isNull); + expect(currentCheckpointRequestId(), 2); + + expect(nextCheckpointRequestId(), 3); + expect(currentCheckpointRequestId(), 3); + }); + + syncTest('seeds requested checkpoint request ids from service state', (_) { + final startInstructions = invokeControlRaw('start', null); + expect(streamLastCheckpointRequestId(startInstructions), isNull); + expect(seedCheckpointRequestId(41), 41); + + expect(nextCheckpointRequestId(), 42); + expect(lastRequestedCheckpointRequestId(), 42); + + final restartInstructions = invokeControlRaw('start', null); + expect( + restartInstructions, + contains(containsPair('EstablishSyncStream', anything)), + ); + expect(streamLastCheckpointRequestId(restartInstructions), 42); + expect(seedCheckpointRequestId(100), 100); + + expect(nextCheckpointRequestId(), 101); + expect(lastRequestedCheckpointRequestId(), 101); + }); + + syncTest('stores seeded checkpoint request ids verbatim', (_) { + invokeControlRaw('start', null); + expect(seedCheckpointRequestId(41), 41); + expect(lastRequestedCheckpointRequestId(), 41); + + // Core does not enforce monotonicity when seeding. SDKs own reconciliation and seed the + // effective state accepted by the service, which may be below the local counter (e.g. after + // switching users). + expect(seedCheckpointRequestId(5), 5); + expect(lastRequestedCheckpointRequestId(), 5); + expect(nextCheckpointRequestId(), 6); + }); + + syncTest('rejects absent checkpoint request ids when seeding', (_) { + invokeControlRaw('start', null); + + expect( + () => invokeControlRaw('seed_checkpoint_request_id', null), + throwsA(isSqliteException( + 3091, + contains('checkpoint request id must be an integer or integer string'), + )), + ); + expect(lastRequestedCheckpointRequestId(), isNull); + }); + + syncTest('rejects zero checkpoint request ids when seeding', (_) { + invokeControlRaw('start', null); + + expect( + () => invokeControlRaw('seed_checkpoint_request_id', 0), + throwsA(isSqliteException( + 3091, + contains('checkpoint request id must be a positive integer'), + )), + ); + expect(lastRequestedCheckpointRequestId(), isNull); + }); + + syncTest('accepts text checkpoint request ids when seeding', (_) { + invokeControlRaw('start', null); + expect(seedCheckpointRequestId('41'), 41); + + expect(lastRequestedCheckpointRequestId(), 41); + expect(nextCheckpointRequestId(), 42); + }); + + syncTest('requires active sync iteration before seeding checkpoint ids', (_) { + expect( + () => seedCheckpointRequestId(1), + throwsA(isSqliteException( + 21, + contains('No iteration is active'), + )), + ); + expect(lastRequestedCheckpointRequestId(), isNull); + }); + + syncTest('requires checkpoint request state before allocating checkpoint ids', + (_) { + invokeControlRaw('start', null); + + expect( + () => invokeControlRaw('next_checkpoint_request_id', null), + throwsA(isSqliteException( + 21, + contains('Checkpoint request state has not been seeded'), + )), + ); + expect(probeTargetCheckpointRequestId(), isNull); + }); + + syncTest( + 'probes and updates target checkpoint request id without sync iteration', + (_) { + expect(probeTargetCheckpointRequestId(), isNull); + expect(probeTargetCheckpointRequestId(1), isNull); + expect(lastRequestedCheckpointRequestId(), isNull); + expect(probeTargetCheckpointRequestId(), 1); + + expect(probeTargetCheckpointRequestId(2), 1); + expect(lastRequestedCheckpointRequestId(), isNull); + expect(probeTargetCheckpointRequestId(), 2); + }); + + syncTest( + 'accepts text checkpoint request ids for target checkpoint request id', + (_) { + expect(probeTargetCheckpointRequestId('1'), isNull); + expect(lastRequestedCheckpointRequestId(), isNull); + expect(probeTargetCheckpointRequestId(), 1); + }); + + syncTest('rejects negative target checkpoint request ids', (_) { + expect( + () => invokeControlRaw('target_checkpoint_request_id', -1), + throwsA(isSqliteException( + 3091, + contains('target checkpoint request id must be a non-negative integer'), + )), + ); + }); + + syncTest('target checkpoint request id does not update checkpoint request id', + (_) { + invokeControlRaw('start', null); + expect(seedCheckpointRequestId(10), 10); + + expect(lastRequestedCheckpointRequestId(), 10); + expect(probeTargetCheckpointRequestId(7), isNull); + expect(probeTargetCheckpointRequestId(), 7); + expect(lastRequestedCheckpointRequestId(), 10); + }); + + syncTest('does not store target ops as checkpoint request id', (_) { + expect(probeTargetCheckpointRequestId(0), isNull); + expect(lastRequestedCheckpointRequestId(), isNull); + expect(probeTargetCheckpointRequestId(), isNull); + + expect(probeTargetCheckpointRequestId(1), isNull); + expect(lastRequestedCheckpointRequestId(), isNull); + expect(probeTargetCheckpointRequestId(), 1); + + expect(probeTargetCheckpointRequestId(0), 1); + expect(probeTargetCheckpointRequestId(), isNull); + + expect(probeTargetCheckpointRequestId(9223372036854775807), isNull); + expect(lastRequestedCheckpointRequestId(), isNull); + expect(probeTargetCheckpointRequestId(), 9223372036854775807); + }); + + syncTest('does not persist placeholder checkpoint request id', (_) { + db.execute("insert into items (id, col) values ('local', 'data');"); + + invokeControlRaw('start', null); + + expect(lastRequestedCheckpointRequestId(), isNull); + }); + + syncTest( + 'does not emit applied checkpoint request id for partial checkpoint', + (_) { + invokeControl('start', null); + + pushCheckpoint(buckets: priorityBuckets, writeCheckpoint: '1'); + final instructions = pushCheckpointComplete(priority: 2); + + expect( + instructions, + isNot(contains(containsPair('DidCompleteSync', + containsPair('applied_checkpoint_request_id', anything)))), + ); + expect( + instructions, + isNot(contains(containsPair( + 'UpdateSyncStatus', + containsPair( + 'status', + containsPair('internal_last_applied_checkpoint_request_id', + anything))))), + ); + expect(lastAppliedCheckpointRequestId(), isNull); + + final [row] = db.select('select powersync_offline_sync_status();'); + expect( + json.decode(row[0]), + isNot(containsPair('last_applied_checkpoint_request_id', anything)), + ); + }, + ); + + syncTest('emits applied checkpoint request id for full checkpoint', (_) { + invokeControl('start', null); + + pushCheckpoint(buckets: priorityBuckets, writeCheckpoint: '1'); + final appliedInstructions = pushCheckpointComplete(); + + expect( + appliedInstructions, + contains( + containsPair( + 'DidCompleteSync', + {'applied_checkpoint_request_id': 1}, + ), + ), + ); + expect( + appliedInstructions, + contains(containsPair('LogLine', + {'severity': 'DEBUG', 'line': 'Applied checkpoint request id 1'})), + ); + expect( + appliedInstructions, + contains(containsPair( + 'UpdateSyncStatus', + containsPair( + 'status', + containsPair('internal_last_applied_checkpoint_request_id', 1), + ), + )), + ); + expect(lastAppliedCheckpointRequestId(), 1); + + pushCheckpoint(buckets: priorityBuckets); + final instructions = pushCheckpointComplete(); + + expect( + instructions, + contains(containsPair('DidCompleteSync', {})), + ); + expect( + instructions, + isNot(contains(containsPair( + 'UpdateSyncStatus', + containsPair( + 'status', + containsPair( + 'internal_last_applied_checkpoint_request_id', anything))))), + ); + + final [row] = db.select('select powersync_offline_sync_status();'); + expect( + json.decode(row[0]), + isNot(containsPair('last_applied_checkpoint_request_id', anything)), + ); + + expect( + db.select(r"SELECT * FROM ps_buckets WHERE name = '$local'"), isEmpty); + }); + + syncTest('disconnect clears applied checkpoint request id from status', (_) { + invokeControl('start', null); + + pushCheckpoint(buckets: priorityBuckets, writeCheckpoint: '1'); + pushCheckpointComplete(); + + final instructions = invokeControl('stop', null); + expect( + instructions, + contains(containsPair( + 'UpdateSyncStatus', + containsPair( + 'status', + allOf( + containsPair('connected', false), + isNot(containsPair( + 'internal_last_applied_checkpoint_request_id', anything)), + ), + ), + )), + ); + }); + + syncTest('local writes clear checkpoint request high-water marks', (_) { + invokeControl('start', null); + + pushCheckpoint(buckets: priorityBuckets, writeCheckpoint: '5'); + pushCheckpointComplete(); + expect(lastAppliedCheckpointRequestId(), 5); + + // A local write can only be acknowledged by a checkpoint request id observed after it. Stale + // seen/applied values (which may predate a request counter restart) must not remain to open + // the apply gate for a smaller new target id. + db.execute("insert into items (id, col) values ('local', 'data');"); + + expect(lastAppliedCheckpointRequestId(), isNull); + expect( + db.select( + "SELECT 1 FROM ps_kv WHERE key = 'last_seen_checkpoint_request_id'"), + isEmpty, + ); + expect(probeTargetCheckpointRequestId(), 9223372036854775807); + }); + test('clearing database clears sync status', () { invokeControl('start', null); pushCheckpoint(buckets: priorityBuckets); @@ -405,15 +822,18 @@ void _syncTests({ final request = invokeControl('start', null); expect( request, - contains(containsPair( - 'EstablishSyncStream', - { - // Should request state from before clear - 'request': containsPair('buckets', [ - {'name': 'a', 'after': '1'} - ]), - }, - )), + contains( + containsPair( + 'EstablishSyncStream', + containsPair( + // Should request state from before clear + 'request', + containsPair('buckets', [ + {'name': 'a', 'after': '1'} + ]), + ), + ), + ), ); pushCheckpoint(buckets: [bucketDescription('a', count: 1)]); @@ -513,7 +933,7 @@ void _syncTests({ }); test('deletes old buckets', () { - for (final name in ['one', 'two', 'three', r'$local']) { + for (final name in ['one', 'two', 'three']) { db.execute('INSERT INTO ps_buckets (name) VALUES (?)', [name]); } @@ -545,7 +965,6 @@ void _syncTests({ // Should delete the old buckets two and three expect(db.select('select name from ps_buckets order by id'), [ {'name': 'one'}, - {'name': r'$local'} ]); }); @@ -816,8 +1235,14 @@ void _syncTests({ ]); // Now complete the upload process. - db.execute(r"UPDATE ps_buckets SET target_op = 1 WHERE name = '$local'"); - invokeControl('completed_upload', null); + probeTargetCheckpointRequestId(1); + final uploadCompleteInstructions = + invokeControl('completed_upload', null); + expect( + uploadCompleteInstructions, + contains(containsPair('LogLine', + {'severity': 'DEBUG', 'line': 'Applied checkpoint request id 1'})), + ); // This should apply the pending write checkpoint. expect(fetchRows(), [ @@ -832,8 +1257,9 @@ void _syncTests({ // Complete upload process db.execute('DELETE FROM ps_crud'); - db.execute(r"UPDATE ps_buckets SET target_op = 1 WHERE name = '$local'"); + probeTargetCheckpointRequestId(1); expect(invokeControl('completed_upload', null), isEmpty); + expect(lastRequestedCheckpointRequestId(), 1); // Sync afterwards containing data and write checkpoint. pushCheckpoint(buckets: priorityBuckets, writeCheckpoint: '1'); @@ -861,7 +1287,7 @@ void _syncTests({ ]); // Now the upload is complete and requests a write checkpoint - db.execute(r"UPDATE ps_buckets SET target_op = 1 WHERE name = '$local'"); + probeTargetCheckpointRequestId(1); expect(invokeControl('completed_upload', null), isEmpty); // Which triggers a new iteration @@ -898,7 +1324,7 @@ void _syncTests({ db.execute("insert into items (id, col) values ('local2', 'data2');"); // Now the upload is complete and requests a write checkpoint - db.execute(r"UPDATE ps_buckets SET target_op = 1 WHERE name = '$local'"); + probeTargetCheckpointRequestId(1); expect(invokeControl('completed_upload', null), [ containsPair('LogLine', { 'severity': 'WARNING', diff --git a/dart/test/utils/fix_035_fixtures.dart b/dart/test/utils/fix_035_fixtures.dart index fa912e89..0b600a90 100644 --- a/dart/test/utils/fix_035_fixtures.dart +++ b/dart/test/utils/fix_035_fixtures.dart @@ -18,9 +18,9 @@ const dataBroken = ''' /// Data after applying the migration fix, but before sync_local const dataMigrated = ''' -;INSERT INTO ps_buckets(id, name, last_applied_op, last_op, target_op, add_checksum, op_checksum, pending_delete, count_at_last, count_since_last, downloaded_size) VALUES - (1, 'b1', 0, 0, 0, 0, 120, 0, 0, 0, 0), - (2, 'b2', 0, 0, 0, 0, 3, 0, 0, 0, 0) +;INSERT INTO ps_buckets(id, name, last_applied_op, last_op, add_checksum, op_checksum, pending_delete, count_at_last, count_since_last, downloaded_size) VALUES + (1, 'b1', 0, 0, 0, 120, 0, 0, 0, 0), + (2, 'b2', 0, 0, 0, 3, 0, 0, 0, 0) ;INSERT INTO ps_oplog(bucket, op_id, row_type, row_id, key, data, hash) VALUES (1, 1, 'todos', 't1', '', '{}', 100), (1, 2, 'todos', 't2', '', '{}', 20), @@ -39,9 +39,9 @@ const dataMigrated = ''' /// Data after applying the migration fix and sync_local const dataFixed = ''' -;INSERT INTO ps_buckets(id, name, last_applied_op, last_op, target_op, add_checksum, op_checksum, pending_delete, count_at_last, count_since_last, downloaded_size) VALUES - (1, 'b1', 3, 3, 0, 0, 120, 0, 1, 0, 0), - (2, 'b2', 3, 3, 0, 0, 3, 0, 1, 0, 0) +;INSERT INTO ps_buckets(id, name, last_applied_op, last_op, add_checksum, op_checksum, pending_delete, count_at_last, count_since_last, downloaded_size) VALUES + (1, 'b1', 3, 3, 0, 120, 0, 1, 0, 0), + (2, 'b2', 3, 3, 0, 3, 0, 1, 0, 0) ;INSERT INTO ps_oplog(bucket, op_id, row_type, row_id, key, data, hash) VALUES (1, 1, 'todos', 't1', '', '{}', 100), (1, 2, 'todos', 't2', '', '{}', 20), diff --git a/dart/test/utils/migration_fixtures.dart b/dart/test/utils/migration_fixtures.dart index 9566a182..35c23b7a 100644 --- a/dart/test/utils/migration_fixtures.dart +++ b/dart/test/utils/migration_fixtures.dart @@ -1,9 +1,12 @@ /// The current database version -const databaseVersion = 13; +const databaseVersion = 14; /// This is the base database state that we expect at various schema versions. /// Generated by loading the specific library version, and exporting the schema. -const expectedState = { +final expectedState = _expectedState(); + +Map _expectedState() { + final state = { 2: r''' ;CREATE TABLE ps_buckets( name TEXT PRIMARY KEY, @@ -537,12 +540,19 @@ const expectedState = { ;INSERT INTO ps_migration(id, down_migrations) VALUES(11, '[{"sql":"DROP TABLE ps_stream_subscriptions"},{"sql":"DELETE FROM ps_migration WHERE id >= 11"}]') ;INSERT INTO ps_migration(id, down_migrations) VALUES(12, '[{"sql":"ALTER TABLE ps_buckets DROP COLUMN downloaded_size"},{"sql":"DELETE FROM ps_migration WHERE id >= 12"}]') ;INSERT INTO ps_migration(id, down_migrations) VALUES(13, '[{"sql":"UPDATE ps_stream_subscriptions SET expires_at = expires_at / 1000000, last_synced_at = last_synced_at / 1000000"},{"sql":"ALTER TABLE ps_sync_state RENAME TO ps_sync_state_new"},{"sql":"CREATE TABLE ps_sync_state (\n priority INTEGER NOT NULL PRIMARY KEY,\n last_synced_at TEXT NOT NULL\n) STRICT;"},{"sql":"INSERT INTO ps_sync_state (priority, last_synced_at) SELECT priority, datetime(last_synced_at / 1000000, ''unixepoch'') FROM ps_sync_state_new"},{"sql":"DROP TABLE ps_sync_state_new"},{"sql":"DELETE FROM ps_migration WHERE id >= 13"}]')''', -}; + }; + state[14] = '''${state[13]!.trim().replaceFirst(' target_op INTEGER NOT NULL DEFAULT 0,\n', '')} +;INSERT INTO ps_migration(id, down_migrations) VALUES(14, '[{"sql":"ALTER TABLE ps_buckets RENAME TO ps_buckets_14"},{"sql":"DROP INDEX ps_buckets_name"},{"sql":"CREATE TABLE ps_buckets(\\n id INTEGER PRIMARY KEY,\\n name TEXT NOT NULL,\\n last_applied_op INTEGER NOT NULL DEFAULT 0,\\n last_op INTEGER NOT NULL DEFAULT 0,\\n target_op INTEGER NOT NULL DEFAULT 0,\\n add_checksum INTEGER NOT NULL DEFAULT 0,\\n op_checksum INTEGER NOT NULL DEFAULT 0,\\n pending_delete INTEGER NOT NULL DEFAULT 0\\n) STRICT"},{"sql":"CREATE UNIQUE INDEX ps_buckets_name ON ps_buckets (name)"},{"sql":"ALTER TABLE ps_buckets ADD COLUMN count_at_last INTEGER NOT NULL DEFAULT 0"},{"sql":"ALTER TABLE ps_buckets ADD COLUMN count_since_last INTEGER NOT NULL DEFAULT 0"},{"sql":"ALTER TABLE ps_buckets ADD COLUMN downloaded_size INTEGER NOT NULL DEFAULT 0"},{"sql":"INSERT INTO ps_buckets(\\n id,\\n name,\\n last_applied_op,\\n last_op,\\n add_checksum,\\n op_checksum,\\n pending_delete,\\n count_at_last,\\n count_since_last,\\n downloaded_size\\n)\\nSELECT\\n id,\\n name,\\n last_applied_op,\\n last_op,\\n add_checksum,\\n op_checksum,\\n pending_delete,\\n count_at_last,\\n count_since_last,\\n downloaded_size\\nFROM ps_buckets_14"},{"sql":"DROP TABLE ps_buckets_14"},{"sql":"INSERT INTO ps_buckets(name, pending_delete, last_op, last_applied_op, target_op)\\nSELECT ''\$local'', 1, seen, applied, target\\n FROM (\\n SELECT\\n IFNULL((SELECT CAST(value AS INTEGER) FROM ps_kv WHERE key = ''last_seen_checkpoint_request_id''), 0) AS seen,\\n IFNULL((SELECT CAST(value AS INTEGER) FROM ps_kv WHERE key = ''last_applied_checkpoint_request_id''), 0) AS applied,\\n (SELECT CAST(value AS INTEGER) FROM ps_kv WHERE key = ''target_checkpoint_request_id'') AS target\\n )\\n WHERE EXISTS (\\n SELECT 1 FROM ps_kv WHERE key = ''target_checkpoint_request_id''\\n )\\nON CONFLICT(name) DO UPDATE SET\\n pending_delete = excluded.pending_delete,\\n last_op = excluded.last_op,\\n last_applied_op = excluded.last_applied_op,\\n target_op = excluded.target_op"},{"sql":"DELETE FROM ps_migration WHERE id >= 14"}]')'''; + return state; +} final finalState = expectedState[databaseVersion]!; /// data to test "up" migrations -const data1 = { +final data1 = _data1(); + +Map _data1() { + final data = { 2: r''' ;INSERT INTO ps_buckets(name, last_applied_op, last_op, target_op, add_checksum, pending_delete) VALUES ('b1', 0, 0, 0, 0, 0), @@ -672,7 +682,20 @@ const data1 = { ;INSERT INTO ps_updated_rows(row_type, row_id) VALUES ('lists', 'l2') ''' -}; + }; + data[14] = r''' +;INSERT INTO ps_buckets(id, name, last_applied_op, last_op, add_checksum, op_checksum, pending_delete, count_at_last, count_since_last, downloaded_size) VALUES + (1, 'b1', 0, 0, 0, 120, 0, 0, 0, 0), + (2, 'b2', 0, 0, 1005, 3, 0, 0, 0, 0) +;INSERT INTO ps_oplog(bucket, op_id, row_type, row_id, key, data, hash) VALUES + (1, 1, 'todos', 't1', '', '{}', 100), + (1, 2, 'todos', 't2', '', '{}', 20), + (2, 3, 'lists', 'l1', '', '{}', 3) +;INSERT INTO ps_updated_rows(row_type, row_id) VALUES + ('lists', 'l2') +'''; + return data; +} /// data to test "down" migrations /// This is slightly different from the above, @@ -719,6 +742,7 @@ final dataDown1 = { 10: data1[9]!, 11: data1[10]!, 12: data1[12]!, + 13: data1[13]!, }; final finalData1 = data1[databaseVersion]!; diff --git a/dart/test/utils/test_utils.dart b/dart/test/utils/test_utils.dart index 943f632d..8c6bd1fc 100644 --- a/dart/test/utils/test_utils.dart +++ b/dart/test/utils/test_utils.dart @@ -23,7 +23,10 @@ Object stream(String name, bool isDefault, {List errors = const []}) { } /// Creates a `checkpoint_complete` or `partial_checkpoint_complete` line. -Object checkpointComplete({int? priority, String lastOpId = '1'}) { +Object checkpointComplete({ + int? priority, + String lastOpId = '1', +}) { return { priority == null ? 'checkpoint_complete' : 'partial_checkpoint_complete': { 'last_op_id': lastOpId, diff --git a/docs/historic-write-checkpoints.md b/docs/historic-write-checkpoints.md new file mode 100644 index 00000000..dd697108 --- /dev/null +++ b/docs/historic-write-checkpoints.md @@ -0,0 +1,208 @@ +# Write Checkpointing + +The general flow for mutations is as follows. + +A client makes a write to a table/view. Triggers are used to populate the `ps_crud` table with an entry for the operation. Every local write marks the `$local` bucket in `ps_buckets` as having a `target_op` of the maximum i64 value - this effectively blocks incoming synced checkpoints from being applied. + +```sql +INSERT OR REPLACE INTO ps_buckets(name, last_op, target_op) VALUES('$local', 0, {MAX_OP_ID}) +``` + +A connected client SDK monitors the `ps_crud` table - or the sync state machine triggers CRUD uploads when ready. The user's `uploadData` gets CRUD transactions with `getNextCrudTransaction` or some equivalent method. Calling the `complete` method on a CRUD transaction-like object will: + +- Remove the entries from `ps_crud` +- Depending on the write checkpoint method used: + - Optionally apply a custom_write_checkpoint as the target_op - ONLY if the `ps_crud` queue is empty + - Ensure that the `target_op` is at the MAX_OP_ID + +```TypeScript + await tx.execute(`DELETE FROM ${PSInternalTable.CRUD} WHERE id <= ?`, [lastClientId]); + if (writeCheckpoint) { + const check = await tx.execute(`SELECT 1 FROM ${PSInternalTable.CRUD} LIMIT 1`); + if (!check.rows?.length) { + await tx.execute(`UPDATE ${PSInternalTable.BUCKETS} SET target_op = CAST(? as INTEGER) WHERE name='$local'`, [ + writeCheckpoint + ]); + } + } else { + await tx.execute(`UPDATE ${PSInternalTable.BUCKETS} SET target_op = CAST(? as INTEGER) WHERE name='$local'`, [ + this.bucketStorageAdapter.getMaxOpId() + ]); + } +``` + +Once all the uploads have completed, the Sync implementation will attempt to update the local target. + +```typescript +// private async _uploadAllCrud(signal: AbortSignal): Promise { + +// AbstractStreamingSyncImplementation.ts line 418 +// Uploading is completed +const neededUpdate = await this.options.adapter.updateLocalTarget(() => this.getWriteCheckpoint()); + +// ... +// } + +// SqliteBucketStorage.ts line 67 +// async updateLocalTarget(cb: () => Promise): Promise { +const rs1 = await this.db.getAll( + "SELECT target_op FROM ps_buckets WHERE name = '$local' AND target_op = CAST(? as INTEGER)", + [MAX_OP_ID] +); + +// If the target op is not the MAX_OP_ID (it's a concrete checkpoint ID) +// Then: Don't fetch a new write checkpoint from the service, leave it as is. +// This essentially caters for the custom write checkpoint case where a concrete `writeCheckpoint` +// is set after all items have been uploaded. +// In the managed checkpoint flow, the target_op should be the MAX_OP_ID here. +if (!rs1.length) { + // Nothing to update + return false; +} + +// The logic below tries to ensure that no uploads happened in-between async operations, +// like fetching a write-checkpoint from the PowerSync service +const rs = await this.db.getAll<{ seq: number }>("SELECT seq FROM main.sqlite_sequence WHERE name = 'ps_crud'"); +if (!rs.length) { + // Nothing to update + return false; +} + +const seqBefore: number = rs[0]['seq']; + +// This callback usually connects to the PowerSync service write-checkpoint2.json endpoint +const opId = await cb(); + +// Now we apply the target_op, only if no other CRUD items have dirtied the local state meanwhile. +return this.writeTransaction(async (tx) => { + const anyData = await tx.execute('SELECT 1 FROM ps_crud LIMIT 1'); + if (anyData.rows?.length) { + // if isNotEmpty + this.logger.debug(`New data uploaded since write checkpoint ${opId} - need new write checkpoint`); + return false; + } + + const rs = await tx.execute("SELECT seq FROM main.sqlite_sequence WHERE name = 'ps_crud'"); + if (!rs.rows?.length) { + // assert isNotEmpty + throw new Error('SQLite Sequence should not be empty'); + } + + const seqAfter: number = rs.rows?.item(0)['seq']; + if (seqAfter != seqBefore) { + this.logger.debug( + `New data uploaded since write checkpoint ${opId} - need new write checkpoint (sequence updated)` + ); + + // New crud data may have been uploaded since we got the checkpoint. Abort. + return false; + } + + this.logger.debug(`Updating target write checkpoint to ${opId}`); + await tx.execute("UPDATE ps_buckets SET target_op = CAST(? as INTEGER) WHERE name='$local'", [opId]); + return true; +}); +// } +``` + +Concurrently, the streaming sync implementation is reading checkpoints from the PowerSync service `/sync/stream/` endpoint. The PowerSync service reports which checkpoints are associated with a corresponding write checkpoint. + +The client does not publish guarded changes as soon as it sees that value. After a full checkpoint has completed and its checksums have been validated, `sync_local` stores the checkpoint's `write_checkpoint` as `$local.last_op`. Incoming changes can then be applied locally only when `$local.target_op <= $local.last_op` and the `ps_crud` queue is empty. Partial priority 0 syncs are the exception: they may publish while uploads are outstanding. + +```Rust +// sync_local.rs + +fn can_apply_sync_changes(&self) -> Result { + // Don't publish downloaded data until the upload queue is empty (except for downloaded data + // in priority 0, which is published earlier). + + let needs_check = match &self.partial { + Some(p) => !p.priority.may_publish_with_outstanding_uploads(), + None => true, + }; + + if needs_check { + // language=SQLite + let statement = self.db.prepare_v2( + "SELECT 1 FROM ps_buckets WHERE target_op > last_op AND name = '$local'", + )?; + + if statement.step()? == ResultCode::ROW { + return Ok(false); + } + + let statement = self.db.prepare_v2("SELECT 1 FROM ps_crud LIMIT 1")?; + if statement.step()? != ResultCode::DONE { + return Ok(false); + } + } + + Ok(true) + } + +// storage_adapter.rs + +pub fn sync_local( + // ... +) { + +// ... + if let (None, Some(write_checkpoint)) = (&priority, &checkpoint.write_checkpoint) { + update_bucket.bind_int64(1, *write_checkpoint)?; + update_bucket.bind_text(2, "$local", sqlite::Destructor::STATIC)?; + update_bucket.exec()?; + } + +// ... +} +``` + +## The $local bucket + +`$local` is a special row in `ps_buckets` used to track whether downloaded changes are safe to +publish while local writes are being uploaded. It is stored in the same table as real sync buckets, +but it is not sent to the sync service as a bucket request. + +- `target_op`: The write checkpoint that must be reached before guarded upstream changes may be + published locally. Local writes set this to `MAX_OP_ID`; custom write checkpoints or managed + write-checkpoint requests replace it with a concrete checkpoint id once the relevant upload has + completed. + + ```sql + -- Local writes block guarded publishes until a concrete write checkpoint is known. + INSERT OR REPLACE INTO ps_buckets(name, last_op, target_op) + VALUES('$local', 0, {MAX_OP_ID}); + + -- After upload completion, custom or managed checkpointing stores the target checkpoint. + UPDATE ps_buckets SET target_op = CAST(? AS INTEGER) WHERE name = '$local'; + ``` + +- `last_op`: The latest write checkpoint observed from the sync service. This is updated from + `checkpoint.write_checkpoint` when a full checkpoint is validated. + + ```rust + let update_bucket = self.db.prepare_v2("UPDATE ps_buckets SET last_op = ? WHERE name = ?")?; + + if let (None, Some(write_checkpoint)) = (&priority, &checkpoint.write_checkpoint) { + update_bucket.bind_int64(1, *write_checkpoint)?; + update_bucket.bind_text(2, "$local", sqlite::Destructor::STATIC)?; + update_bucket.exec()?; + } + ``` + +- `last_applied_op`: The latest write checkpoint whose guarded changes have actually been published + locally. This advances to `last_op` after a full `sync_local` apply succeeds. + + ```sql + UPDATE ps_buckets + SET last_applied_op = last_op + WHERE last_applied_op != last_op; + ``` + +The apply gate checks `$local.target_op > $local.last_op` before publishing full checkpoints and +non-priority-0 partial checkpoints. It also checks that `ps_crud` is empty. This means downloaded +changes remain buffered until the client has both uploaded local CRUD and seen the corresponding +write checkpoint in the sync stream. + +Clearing the database removes `$local`: a hard clear deletes all rows from `ps_buckets`, while a +soft clear deletes only the `$local` row and keeps reusable remote bucket state. diff --git a/docs/schema.md b/docs/schema.md index f7dfbde3..2140cb27 100644 --- a/docs/schema.md +++ b/docs/schema.md @@ -15,8 +15,8 @@ A bucket is instantiated for every row returned by a parameter query in a [bucke Clients create entries in `ps_buckets` when receiving a checkpoint message from the sync service, they are also responsible for removing buckets that are no longer relevant to the client. -There is also a special `$local` bucket representing pending -uploads. +Older schema versions also used a special `$local` bucket to represent pending uploads. Current +schema versions keep that local write gate in `ps_kv` instead. We store the following information in `ps_buckets`: @@ -24,12 +24,20 @@ We store the following information in `ps_buckets`: 2. `name`: The name of the bucket as received from the sync service. 3. `last_applied_op`: The last operation id that has been verified and published to views (meaning that it was part of a checkpoint and that we have validated its checksum). -4. `target_op`: Only used for `$local`. TODO: Document further. -5. `add_checksum`: TODO: Document further. -6. `op_checksum`: TODO: Document further. -7. `pending_delete`: TODO: Appears to be unused, document further. -8. `count_at_last`: The amount of operations in the bucket at the last verified checkpoint. -9. `count_since_last`: The amount of operations downloaded since the last verified checkpoint. +4. `add_checksum`: TODO: Document further. +5. `op_checksum`: TODO: Document further. +6. `pending_delete`: TODO: Appears to be unused, document further. +7. `count_at_last`: The amount of operations in the bucket at the last verified checkpoint. +8. `count_since_last`: The amount of operations downloaded since the last verified checkpoint. + +Schema version 14 removes the legacy `target_op` column after migrating `$local.target_op` to +`ps_kv.target_checkpoint_request_id`, and deletes the `$local` row so `ps_buckets` only contains real sync +buckets. This makes older SDKs fail with a hard SQLite error if they try to keep using the migrated +database without downgrading. That failure is deliberate — including for multi-process deployments +where processes with mixed SDK versions share one database — because an older SDK silently +maintaining `$local` state the new implementation no longer reads would be worse than a loud error. +The down migration restores `target_op` and recreates the `$local` row from `ps_kv` for older +schema versions. ## `ps_crud` diff --git a/docs/sync.md b/docs/sync.md index 8737cee0..4a51f125 100644 --- a/docs/sync.md +++ b/docs/sync.md @@ -10,7 +10,7 @@ The function should always be called in a transaction. The following commands are supported: 1. `start`: Payload is a JSON-encoded object. This requests the client to start a sync iteration. - The payload can either be `null` or an JSON object with: + The payload can either be `null` or a JSON object with: - An optional `parameters: Record` entry, specifying parameters to include in the request to the sync service. - A `schema: { tables: Table[], raw_tables: RawTable[] }` entry specifying the schema of the database to @@ -26,18 +26,40 @@ The following commands are supported: - The client will emit an instruction to stop the current stream, clients should restart by sending another `start` command. 6. `completed_upload`: Notify the sync implementation that all local changes have been uploaded. -7. `update_subscriptions`: Notify the sync implementation that subscriptions which are currently active in the app - have changed. Depending on the TTL of caches, this may cause it to request a reconnect. +7. `update_subscriptions`: Payload is a JSON-encoded array of + `{name: string, params: Record}`. Notify the sync implementation that subscriptions + which are currently active in the app have changed. Depending on the TTL of caches, this may + cause it to request a reconnect. 8. `connection`: Notify the sync implementation about the connection being opened (second parameter should be `established`) or the HTTP stream closing (second parameter should be `end`). This is used to set `connected` to true in the sync status without waiting for the first sync line. -9. `subscriptions`: Store a new sync steam subscription in the database or remove it. +9. `subscriptions`: Store a new sync stream subscription in the database or remove it. This command can run outside of a sync iteration and does not affect it. -10. `update_subscriptions`: Second parameter is a JSON-encoded array of `{name: string, params: Record}`. - If a new subscription is created, or when a subscription without a TTL has been removed, the client will ask to - restart the connection. +10. `next_checkpoint_request_id`: No payload. During an active sync iteration after checkpoint + request state exists locally, allocates and returns the next checkpoint request id as an + integer result. +11. `current_checkpoint_request_id`: No payload. Returns the current checkpoint request sequence + value as an integer result, or SQL `NULL` if absent. This command does not allocate a new id and + can run outside a sync iteration. +12. `target_checkpoint_request_id`: Payload is `null`, an integer, or an integer string. Probes, updates or + clears the target checkpoint request id and returns the previously-observed value as an integer result, or + SQL `NULL` if there was no target. This command can run outside of a sync iteration and does not + affect it. +13. `seed_checkpoint_request_id`: Payload is a positive integer or integer string. During an active + sync iteration, after receiving `EstablishSyncStream`, SDKs should reconcile the local hint with + service-side checkpoint-request state, then seed core with the accepted positive id. Returns the + seeded id as an integer result. -`powersync_control` returns a JSON-encoded array of instructions for the client: +## Checkpoint Request Expectations + +Checkpoint request state exists to protect local writes and to support explicit "wait until synced" +requests. The state model, the per-connection reconciliation and seeding flow, and how SDKs resolve +checkpoint waiters (through `DidCompleteSync.applied_checkpoint_request_id` or the equivalent +sync-status field) are documented in `write-checkpoint-requests.md`. + +Most `powersync_control` commands return a JSON-encoded array of instructions for the client. +`seed_checkpoint_request_id`, `next_checkpoint_request_id`, `current_checkpoint_request_id` and +`target_checkpoint_request_id` return values directly. ```typescript type Instruction = { LogLine: LogLine } @@ -49,8 +71,9 @@ type Instruction = { LogLine: LogLine } // For the Dart web client, flush the (otherwise non-durable) file system. | { FlushFileSystem: {} } // Notify clients that a checkpoint was completed. Clients can clear the - // download error state in response to this. - | { DidCompleteSync: {} } + // download error state in response to this. If a full checkpoint with a + // write_checkpoint was applied, applied_checkpoint_request_id is set. + | { DidCompleteSync: DidCompleteSync } interface LogLine { severity: 'DEBUG' | 'INFO' | 'WARNING', @@ -58,8 +81,14 @@ interface LogLine { } // Instructs client SDKs to open a connection to the sync service. +// last_checkpoint_request_id is the client's current counter state before this stream request. +// On every connect, SDKs use it to re-affirm checkpoint request state with the service (which may +// have deleted its record). The re-affirmation is bidirectional: the hint can restore the +// service-side value, or the service's response can bump the local counter via +// powersync_control('seed_checkpoint_request_id', response). interface EstablishSyncStream { request: any // The JSON-encoded StreamingSyncRequest to send to the sync service + last_checkpoint_request_id: null | number } // Instructs SDKS to update the downloading state of their SyncStatus. @@ -68,6 +97,12 @@ interface UpdateSyncStatus { connecting: boolean, priority_status: [], downloading: null | DownloadProgress, + streams: [], + internal_last_applied_checkpoint_request_id?: number, +} + +interface DidCompleteSync { + applied_checkpoint_request_id?: number, } // Instructs SDKs to refresh credentials from the backend connector. diff --git a/docs/write-checkpoint-requests.md b/docs/write-checkpoint-requests.md new file mode 100644 index 00000000..9106aa99 --- /dev/null +++ b/docs/write-checkpoint-requests.md @@ -0,0 +1,175 @@ +# Write Checkpoint State in `ps_kv` + +This document describes the checkpoint state used to keep downloaded rows behind local writes until +the service has observed those writes, while also supporting explicit "wait until synced" +checkpoint requests. + +The state is internal core/SDK bookkeeping. Apps should not read it as user-facing sync progress. + +## State Model + +| Key | Purpose | Who updates it | +| -------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------- | +| `target_checkpoint_request_id` | Apply gate for local writes. Downloaded full checkpoints can apply only after the stream has seen this id. `MAX_OP_ID` means "local writes exist, but no concrete checkpoint id is known yet". | Core CRUD triggers set the sentinel; SDK upload code stores the accepted concrete id through `powersync_control('target_checkpoint_request_id', id)`. | +| `last_requested_checkpoint_request_id` | Allocation counter for client-created checkpoint requests. | SDKs seed it after connection reconciliation, then core increments it through `powersync_control('next_checkpoint_request_id', NULL)`. | +| `last_seen_checkpoint_request_id` | Latest full checkpoint `write_checkpoint` observed in the stream since the last local write. | Core updates it when a full checkpoint validates. Local writes clear it. | +| `last_applied_checkpoint_request_id` | Latest full checkpoint `write_checkpoint` applied locally since the last local write. | Core updates it after a full checkpoint applies. Local writes clear it. | + +Why four keys? + +- `target_checkpoint_request_id` is the gate that protects local writes. +- `last_requested_checkpoint_request_id` is the counter used to create new requests. +- `last_seen_checkpoint_request_id` answers "has the stream reached the gate yet?" +- `last_applied_checkpoint_request_id` is persisted for consistency with the legacy + `$local.last_applied_op` bookkeeping and for diagnostics. SDK waiters should use + `DidCompleteSync.applied_checkpoint_request_id` or the equivalent sync-status field instead. + +## SDK Expectations + +SDKs should use `powersync_control` as the public API for this state: + +1. On each sync connection, read `EstablishSyncStream.last_checkpoint_request_id`. +2. Reconcile that local hint with service-side checkpoint-request state before allocating new ids. +3. Seed core with the reconciled positive id by calling + `powersync_control('seed_checkpoint_request_id', id)`. +4. Wait for seeding to complete before creating checkpoint requests. +5. When upload code needs a write checkpoint, allocate an id inside a transaction with + `powersync_control('next_checkpoint_request_id', NULL)`. +6. Post that id to the service checkpoint-request endpoint. +7. After the service accepts it, store it as the local write gate with + `powersync_control('target_checkpoint_request_id', id)`. +8. Resolve explicit waiters from `DidCompleteSync.applied_checkpoint_request_id`, not from `ps_kv`. + +`seed_checkpoint_request_id` stores the id verbatim and does not enforce monotonicity. SDKs own +reconciliation and must not seed stale service state. The recommended reconciliation pattern is: + +```text +effectiveId = max(localHint ?? 0, concreteLocalTarget ?? 0, 1) +acceptedId = postCheckpointRequestStateToService(effectiveId) +powersync_control('seed_checkpoint_request_id', acceptedId) +``` + +Posting at least `1` covers the no-record case and probes whether the service supports checkpoint +requests. Core rejects `NULL` and `0` seeds. + +The service returns the maximum of the client-provided id and its service-side record. If the client +lost local state, for example after `disconnectAndClear`, this response hydrates core's counter. If +the service lost its record, posting the local hint recreates the service-side state. + +`powersync_control('next_checkpoint_request_id', NULL)` requires an active sync iteration and a +seeded request counter. SDKs should wait for the connection reconciliation and seed step before +creating checkpoint requests. + +To retry an existing checkpoint request without advancing the counter, SDKs can read: + +```text +powersync_control('current_checkpoint_request_id', NULL) +``` + +Core returns the current sequence value as a SQLite integer, or SQL `NULL` when the counter has not +been seeded. SDKs can compare this with their runtime last-applied checkpoint request id and repost +the current id when the applied id is absent or lower. + +## Local Write Gate + +A local write records CRUD and sets: + +```sql +ps_kv['target_checkpoint_request_id'] = MAX_OP_ID +``` + +It also clears `last_seen_checkpoint_request_id` and `last_applied_checkpoint_request_id`, because +checkpoint ids observed before the write cannot acknowledge it. + +After uploaded CRUD is accepted by the backend, SDK code replaces `MAX_OP_ID` with a concrete +checkpoint id: + +```text +transaction { + requestId = powersync_control('next_checkpoint_request_id', NULL) +} + +POST /sync/checkpoint-request { + client_id, + checkpoint_request_id: requestId +} + +transaction { + previousTarget = powersync_control('target_checkpoint_request_id', NULL) + if previousTarget == MAX_OP_ID && ps_crud is still empty { + powersync_control('target_checkpoint_request_id', requestId) + } +} +``` + +`target_checkpoint_request_id` is intentionally separate from `last_requested_checkpoint_request_id`: allocating +a checkpoint request id does not mean it should block or unblock local writes. + +## Applying Downloaded Checkpoints + +The service reports accepted checkpoint request ids as `checkpoint.write_checkpoint`. For full +checkpoints, core stores that value as `last_seen_checkpoint_request_id`. +The gate uses "seen" rather than "applied" because this check runs before the current checkpoint +can be applied. + +Full checkpoints and non-priority-0 partial checkpoints can publish only when: + +- `ps_crud` is empty, and +- `target_checkpoint_request_id` is absent or less than or equal to `last_seen_checkpoint_request_id`. + +Priority 0 partial syncs may publish while uploads are outstanding. + +If a full checkpoint validates but cannot apply because local CRUD is pending, core keeps it as a +pending checkpoint. When the SDK later sends `completed_upload`, core retries it unless the +pending checkpoint is older than the current `target_checkpoint_request_id`. + +After a full checkpoint applies, core emits: + +```text +DidCompleteSync { applied_checkpoint_request_id?: checkpoint.write_checkpoint } +``` + +SDKs should resolve `requestCheckpoint()` / `waitForSync()` waiters when this value is greater than +or equal to the requested id. + +Core also includes the applied request id in +`UpdateSyncStatus.status.internal_last_applied_checkpoint_request_id` when a status update follows a +checkpoint apply with a `write_checkpoint`. Later status updates without an applied request id, +including disconnect updates, clear the field. SDKs don't need to memoize it: it is an event-style +signal, so react to the value when a status update carries it (for example, resolve waiters whose +requested id is less than or equal to the emitted value) rather than polling it as a high-water +mark. Treat it as internal, runtime-only state rather than app-visible progress or persisted +checkpoint state. + +## Control Commands + +`powersync_control('seed_checkpoint_request_id', id)` + +- Payload: positive integer or integer string. +- Returns: seeded checkpoint request id as a SQLite integer. +- Stores the reconciled checkpoint-request counter seed. +- Requires an active sync iteration. +- Must be called after connection reconciliation and before allocating new ids. + +`powersync_control('next_checkpoint_request_id', NULL)` + +- Returns: next checkpoint request id as a SQLite integer. +- Requires an active sync iteration and a seeded request counter. +- SDKs should call this only after the connection reconciliation and seed step has completed. +- Participates in the caller's transaction. If the transaction rolls back after the id was posted + to the service, retrying posts the same id again, which is safe because the service treats the + latest posted id as effective state. + +`powersync_control('current_checkpoint_request_id', NULL)` + +- Returns: current checkpoint request sequence value as a SQLite integer, or SQL `NULL` if absent. +- Does not allocate a new id. +- SDKs should compare this with their runtime last-applied checkpoint request id to decide whether + to repost the current id. + +`powersync_control('target_checkpoint_request_id', value)` + +- Payload `NULL`: return current target without changing it. +- Payload `0`: clear the target. +- Positive payload: store a concrete target. +- Returns the previous target as a SQLite integer, or SQL `NULL` if absent.