seed_checkpoint_request_id should not be a sync event

Author: stevensJourney

crates/core/src/sync/interface.rs

@@ -90,10 +90,6 @@ pub enum SyncEvent<'a> {
     ///
     /// In response, we'll stop the current iteration to begin another one with the new token.
     DidRefreshToken,
-    /// Seeds the checkpoint request counter from service state.
-    SeedCheckpointRequestId {
-        request_id: i64,
-    },
     /// Notifies the sync client that the current CRUD upload (for which the client SDK is
     /// responsible) has finished.
     ///
@@ -249,6 +245,29 @@ pub fn register(db: *mut sqlite::sqlite3, state: Rc<DatabaseState>) -> 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();
@@ -316,15 +335,6 @@ pub fn register(db: *mut sqlite::sqlite3, state: Rc<DatabaseState>) -> Result<()
                     },
                 }),
                 "refreshed_token" => SyncControlRequest::SyncEvent(SyncEvent::DidRefreshToken),
-                "seed_checkpoint_request_id" => {
-                    SyncControlRequest::SyncEvent(SyncEvent::SeedCheckpointRequestId {
-                        request_id: parse_positive_i64_payload(
-                            *payload,
-                            "checkpoint request id",
-                            "checkpoint request id must be an integer or integer string",
-                        )?,
-                    })
-                }
                 "completed_upload" => SyncControlRequest::SyncEvent(SyncEvent::UploadFinished),
                 "update_subscriptions" => {
                     SyncControlRequest::SyncEvent(SyncEvent::DidUpdateSubscriptions {

crates/core/src/sync/streaming_sync.rs

@@ -562,10 +562,6 @@ impl StreamingSyncIteration {
                         .update(|s| s.disconnect(), &mut event.instructions);
                     break false;
                 }
-                SyncEvent::SeedCheckpointRequestId { request_id } => {
-                    self.adapter.seed_checkpoint_request_id(request_id)?;
-                    continue;
-                }
                 SyncEvent::TextLine { data } => SyncLineWithSource::from_text(data)?,
                 SyncEvent::BinaryLine { data } => SyncLineWithSource::from_binary(data)?,
                 SyncEvent::UploadFinished => {

dart/test/goldens/simple_iteration.json

@@ -33,11 +33,6 @@
       }
     ]
   },
-  {
-    "operation": "seed_checkpoint_request_id",
-    "data": 1,
-    "output": []
-  },
   {
     "operation": "line_text",
     "data": {

dart/test/goldens/starting_stream.json

@@ -38,10 +38,5 @@
         }
       }
     ]
-  },
-  {
-    "operation": "seed_checkpoint_request_id",
-    "data": 1,
-    "output": []
   }
 ]

dart/test/sync_stream_test.dart

@@ -68,10 +68,7 @@ void main() {
   List<Object?> control(String operation, Object? data) {
     final result = controlRaw(operation, data);
     if (operation == 'start' && establishesSyncStream(result)) {
-      return [
-        ...result,
-        ...controlRaw('seed_checkpoint_request_id', 1),
-      ];
+      controlRaw('seed_checkpoint_request_id', 1);
     }
     return result;
   }

dart/test/sync_test.dart

@@ -60,7 +60,12 @@ void _syncTests<T>({
 
     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) {
@@ -83,6 +88,10 @@ void _syncTests<T>({
     return row.columnAt(0);
   }
 
+  int seedCheckpointRequestId(Object? requestId) {
+    return invokeControlScalar('seed_checkpoint_request_id', requestId) as int;
+  }
+
   bool establishesSyncStream(List<Object?> instructions) {
     return instructions.any((instruction) =>
         instruction is Map && instruction.containsKey('EstablishSyncStream'));
@@ -98,10 +107,7 @@ void _syncTests<T>({
     }
 
     if (operation == 'start' && establishesSyncStream(result)) {
-      final seedResult = matcher.enabled
-          ? matcher.invoke('seed_checkpoint_request_id', 1)
-          : invokeControlRaw('seed_checkpoint_request_id', 1);
-      return [...result, ...seedResult];
+      seedCheckpointRequestId(1);
     }
 
     return result;
@@ -246,15 +252,12 @@ void _syncTests<T>({
   });
 
   syncTest('app_metadata is passed to EstablishSyncStream request', (_) {
-    final startInstructions = [
-      ...invokeControlRaw(
-        'start',
-        json.encode({
-          'app_metadata': {'key1': 'value1', 'key2': 'value2'}
-        }),
-      ),
-      ...invokeControlRaw('seed_checkpoint_request_id', 1),
-    ];
+    final startInstructions = invokeControlRaw(
+      'start',
+      json.encode({
+        'app_metadata': {'key1': 'value1', 'key2': 'value2'}
+      }),
+    );
 
     expect(
       startInstructions,
@@ -470,7 +473,7 @@ void _syncTests<T>({
     expect(currentCheckpointRequestId(), isNull);
 
     invokeControlRaw('start', null);
-    invokeControlRaw('seed_checkpoint_request_id', 1);
+    expect(seedCheckpointRequestId(1), 1);
     expect(currentCheckpointRequestId(), 1);
 
     expect(currentCheckpointRequestId(), 1);
@@ -494,7 +497,7 @@ void _syncTests<T>({
   syncTest('seeds requested checkpoint request ids from service state', (_) {
     final startInstructions = invokeControlRaw('start', null);
     expect(streamLastCheckpointRequestId(startInstructions), isNull);
-    invokeControlRaw('seed_checkpoint_request_id', 41);
+    expect(seedCheckpointRequestId(41), 41);
 
     expect(nextCheckpointRequestId(), 42);
     expect(lastRequestedCheckpointRequestId(), 42);
@@ -505,21 +508,21 @@ void _syncTests<T>({
       contains(containsPair('EstablishSyncStream', anything)),
     );
     expect(streamLastCheckpointRequestId(restartInstructions), 42);
-    expect(invokeControlRaw('seed_checkpoint_request_id', 100), isEmpty);
+    expect(seedCheckpointRequestId(100), 100);
 
     expect(nextCheckpointRequestId(), 101);
     expect(lastRequestedCheckpointRequestId(), 101);
   });
 
   syncTest('stores seeded checkpoint request ids verbatim', (_) {
     invokeControlRaw('start', null);
-    invokeControlRaw('seed_checkpoint_request_id', 41);
+    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).
-    invokeControlRaw('seed_checkpoint_request_id', 5);
+    expect(seedCheckpointRequestId(5), 5);
     expect(lastRequestedCheckpointRequestId(), 5);
     expect(nextCheckpointRequestId(), 6);
   });
@@ -552,12 +555,23 @@ void _syncTests<T>({
 
   syncTest('accepts text checkpoint request ids when seeding', (_) {
     invokeControlRaw('start', null);
-    invokeControlRaw('seed_checkpoint_request_id', '41');
+    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);
@@ -572,7 +586,9 @@ void _syncTests<T>({
     expect(probeTargetCheckpointRequestId(), isNull);
   });
 
-  syncTest('probes and updates target checkpoint request id without sync iteration', (_) {
+  syncTest(
+      'probes and updates target checkpoint request id without sync iteration',
+      (_) {
     expect(probeTargetCheckpointRequestId(), isNull);
     expect(probeTargetCheckpointRequestId(1), isNull);
     expect(lastRequestedCheckpointRequestId(), isNull);
@@ -583,7 +599,9 @@ void _syncTests<T>({
     expect(probeTargetCheckpointRequestId(), 2);
   });
 
-  syncTest('accepts text checkpoint request ids for target checkpoint request id', (_) {
+  syncTest(
+      'accepts text checkpoint request ids for target checkpoint request id',
+      (_) {
     expect(probeTargetCheckpointRequestId('1'), isNull);
     expect(lastRequestedCheckpointRequestId(), isNull);
     expect(probeTargetCheckpointRequestId(), 1);
@@ -599,9 +617,10 @@ void _syncTests<T>({
     );
   });
 
-  syncTest('target checkpoint request id does not update checkpoint request id', (_) {
+  syncTest('target checkpoint request id does not update checkpoint request id',
+      (_) {
     invokeControlRaw('start', null);
-    invokeControlRaw('seed_checkpoint_request_id', 10);
+    expect(seedCheckpointRequestId(10), 10);
 
     expect(lastRequestedCheckpointRequestId(), 10);
     expect(probeTargetCheckpointRequestId(7), isNull);

docs/sync.md

@@ -45,9 +45,10 @@ The following commands are supported:
     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. After receiving
-    `EstablishSyncStream`, SDKs should reconcile the local hint with service-side
-    checkpoint-request state, then seed core with the accepted positive id.
+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.
 
 ## Checkpoint Request Expectations
 
@@ -57,8 +58,8 @@ checkpoint waiters (through `DidCompleteSync.applied_checkpoint_request_id` or t
 sync-status field) are documented in `write-checkpoint-requests.md`.
 
 Most `powersync_control` commands return a JSON-encoded array of instructions for the client.
-`next_checkpoint_request_id`, `current_checkpoint_request_id` and `target_checkpoint_request_id` return values
-directly.
+`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 }

docs/write-checkpoint-requests.md

@@ -146,7 +146,9 @@ checkpoint state.
 `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)`