Make offline replica mutations atomic

package.json

@@ -32,6 +32,7 @@
     "test:watch": "ng test",
     "test:actions": "node --test .github/actions/classify-mobile-release/classify-mobile-release.spec.mjs && vitest run --config .github/actions/vitest.config.mjs",
     "test:offline-types:nonstrict": "npm run prebuild:kit && tsc -p projects/kit/offline/tsconfig.nonstrict-null.json",
+    "test:sqlite-concurrency": "node --test projects/kit/offline/src/lib/sqlite-concurrency.node.test.mjs",
     "e2e": "playwright test",
     "e2e:ui": "playwright test --ui"
   },

projects/kit/offline/src/lib/offline-aggregate-intent-projector.spec.ts

@@ -239,7 +239,6 @@ describe('OfflineAggregateIntentProjector', () => {
         getReplicaRowByRemoteId: vi.fn(async () => null),
         getReplicaRowByRemoteIdentity: vi.fn(async () => null),
         getReplicaCursor: vi.fn(async () => null),
-        getReconciliationScopes: vi.fn(async () => []),
         getPullAttentions: vi.fn(async () => []),
         transactReplica,
       } as unknown as OfflineRepository;

projects/kit/offline/src/lib/offline-replica-mutation-coordinator.spec.ts

@@ -1,6 +1,8 @@
 import { TestBed } from '@angular/core/testing';
 import { describe, expect, it, vi } from 'vitest';
 import { OfflineReplicaMutationCoordinator } from './offline-replica-mutation-coordinator';
+import { OFFLINE_REPOSITORY_ATOMIC_MUTATION } from './offline-repository-concurrency';
+import { OFFLINE_REPOSITORY } from './offline-repository';
 
 describe('OfflineReplicaMutationCoordinator', () => {
   it('serializes local apply sections and releases the lane after failure', async () => {
@@ -28,4 +30,18 @@ describe('OfflineReplicaMutationCoordinator', () => {
 
     expect(order).toEqual(['first:start', 'first:end', 'second', 'third']);
   });
+
+  it('uses the repository atomic-mutation capability without retrying the product operation', async () => {
+    const atomicMutation = vi.fn(async (operation: () => Promise<string>) => operation());
+    TestBed.configureTestingModule({
+      providers: [{ provide: OFFLINE_REPOSITORY, useValue: { [OFFLINE_REPOSITORY_ATOMIC_MUTATION]: atomicMutation } }],
+    });
+    const coordinator = TestBed.inject(OfflineReplicaMutationCoordinator);
+    const operation = vi.fn(async () => 'done');
+
+    await expect(coordinator.run(operation)).resolves.toBe('done');
+
+    expect(atomicMutation).toHaveBeenCalledOnce();
+    expect(operation).toHaveBeenCalledOnce();
+  });
 });

projects/kit/offline/src/lib/offline-replica-mutation-coordinator.ts

@@ -8,7 +8,9 @@ import {
 } from './offline-aggregate-intent-projector';
 import { canonicalOfflineReplicaIdentity, commandIdentityMatchesReplicaRow, type OfflineCommandIdentity } from './offline-identity';
 import { OFFLINE_KIT_OPTIONS } from './offline-kit-options';
+import { OFFLINE_REPOSITORY_ATOMIC_MUTATION } from './offline-repository-concurrency';
 import {
+  OFFLINE_REPOSITORY,
   canonicalOfflineReplicaRowKey,
   type OfflineCommand,
   type OfflineReplicaRow,
@@ -28,13 +30,17 @@ import type { OfflineReplicaEntitySchema } from './offline-replica-schema';
  */
 @Injectable({ providedIn: 'root' })
 export class OfflineReplicaMutationCoordinator {
+  readonly #repository = inject(OFFLINE_REPOSITORY, { optional: true });
   readonly #projector = inject(OFFLINE_AGGREGATE_INTENT_PROJECTOR, { optional: true });
   readonly #options = inject(OFFLINE_KIT_OPTIONS, { optional: true });
   #tail: Promise<void> = Promise.resolve();
 
   /** Enqueues one local replica critical section behind any in-flight mutation. */
   run<T>(operation: () => Promise<T>): Promise<T> {
-    const mutation = this.#tail.then(operation);
+    const mutation = this.#tail.then(() => {
+      const atomicMutation = this.#repository?.[OFFLINE_REPOSITORY_ATOMIC_MUTATION];
+      return atomicMutation ? (atomicMutation.call(this.#repository, operation) as Promise<T>) : operation();
+    });
     this.#tail = mutation.then(
       () => undefined,
       () => undefined,

projects/kit/offline/src/lib/offline-repository-concurrency.ts

@@ -0,0 +1,7 @@
+/**
+ * Internal repository capability used to detect commits made through another
+ * native SQLite connection during a local read/derive/write operation.
+ *
+ * This symbol is intentionally not re-exported from the package entry point.
+ */
+export const OFFLINE_REPOSITORY_ATOMIC_MUTATION: unique symbol = Symbol('OFFLINE_REPOSITORY_ATOMIC_MUTATION');

projects/kit/offline/src/lib/offline-repository.ts

@@ -11,6 +11,7 @@ import {
   type OfflineReplicaIdentity,
   type OfflinePrincipalId,
 } from './offline-identity';
+import { OFFLINE_REPOSITORY_ATOMIC_MUTATION } from './offline-repository-concurrency';
 import { OFFLINE_KIT_OPTIONS } from './offline-kit-options';
 import {
   assertOfflineReplicaGeneratedRemoteId,
@@ -164,10 +165,6 @@ export interface OfflineReplicaTransaction {
   putCommands?: readonly OfflineCommand[];
   removeCommandIds?: readonly string[];
   putCursors?: readonly OfflineReplicaCursor[];
-  /** Scopes whose acknowledged server changes still require an authoritative pull. */
-  putReconciliationScopes?: readonly OfflineScope[];
-  /** Scopes whose authoritative post-acknowledgement pull completed successfully. */
-  removeReconciliationScopes?: readonly OfflineScope[];
   /** Durable fatal-pull attentions to upsert for user+scope. */
   putPullAttentions?: readonly OfflinePullAttention[];
   /** Scopes whose fatal-pull attentions should be removed after a successful pull. */
@@ -205,7 +202,6 @@ export interface OfflineRepositoryReader {
     identity: OfflineReplicaRemoteIdentity,
   ): Promise<OfflineReplicaRow<TValues> | null>;
   getReplicaCursor(scope: OfflineScope): Promise<OfflineReplicaCursor | null>;
-  getReconciliationScopes?(userId: OfflinePrincipalId): Promise<OfflineScope[]>;
   getPullAttentions?(userId: OfflinePrincipalId): Promise<OfflinePullAttention[]>;
   getCommands(scope: OfflineScope): Promise<OfflineCommand[]>;
   getCommandsForUser?(userId: OfflinePrincipalId): Promise<OfflineCommand[]>;
@@ -241,7 +237,6 @@ export interface OfflineRepository {
     identity: OfflineReplicaRemoteIdentity,
   ): Promise<OfflineReplicaRow<TValues> | null>;
   getReplicaCursor(scope: OfflineScope): Promise<OfflineReplicaCursor | null>;
-  getReconciliationScopes?(userId: OfflinePrincipalId): Promise<OfflineScope[]>;
   /** Durable fatal-pull attentions for the principal, ordered by scope id. */
   getPullAttentions?(userId: OfflinePrincipalId): Promise<OfflinePullAttention[]>;
   getCommands(scope: OfflineScope): Promise<OfflineCommand[]>;
@@ -266,6 +261,8 @@ export interface OfflineRepository {
    * must not be called from `read`.
    */
   runReadSnapshot<T>(read: (reader: OfflineRepositoryReader) => Promise<T>): Promise<T>;
+  /** @internal Runs one optimistic read/derive/write operation with platform-specific concurrency validation. */
+  [OFFLINE_REPOSITORY_ATOMIC_MUTATION]?<T>(operation: () => Promise<T>): Promise<T>;
 }
 
 /** DI token for the selected platform repository. */
@@ -310,6 +307,8 @@ const CURSORS_KEY = 'offline:replica:cursors';
 const OUTBOX_KEY = 'offline:outbox:commands';
 const REPLICA_TRANSACTION_KEY = 'offline:replica:transaction';
 const REPLICA_SCHEMA_MIGRATION_KEY = 'offline:replica:schema-migration';
+// Legacy marker storage is no longer read or written. Keep cleanup support for
+// databases created by released versions that persisted this key.
 const RECONCILIATION_SCOPES_KEY = 'offline:replica:reconciliation-scopes';
 const PULL_ATTENTIONS_KEY = 'offline:replica:pull-attentions';
 
@@ -403,10 +402,6 @@ export class IonicOfflineRepository implements OfflineRepository {
     return this.#withCommittedRead(() => this.#readReplicaCursor(scope));
   }
 
-  async getReconciliationScopes(userId: OfflinePrincipalId): Promise<OfflineScope[]> {
-    return this.#withCommittedRead(() => this.#readReconciliationScopes(userId));
-  }
-
   async getPullAttentions(userId: OfflinePrincipalId): Promise<OfflinePullAttention[]> {
     return this.#withCommittedRead(() => this.#readPullAttentions(userId));
   }
@@ -489,11 +484,6 @@ export class IonicOfflineRepository implements OfflineRepository {
     return cursor === undefined ? null : { ...scope, cursor };
   }
 
-  async #readReconciliationScopes(userId: OfflinePrincipalId): Promise<OfflineScope[]> {
-    const scopes = await this.#readRecord<OfflineScope>(RECONCILIATION_SCOPES_KEY);
-    return Object.values(scopes).filter((scope) => scope.userId === userId);
-  }
-
   async #readPullAttentions(userId: OfflinePrincipalId): Promise<OfflinePullAttention[]> {
     const attentions = await this.#readRecord<OfflinePullAttention>(PULL_ATTENTIONS_KEY);
     return Object.values(attentions)
@@ -530,7 +520,6 @@ export class IonicOfflineRepository implements OfflineRepository {
       },
       getReplicaRowByRemoteIdentity: (scope, sourceKey, identity) => this.#readReplicaRowByRemoteIdentity(scope, sourceKey, identity),
       getReplicaCursor: (scope) => this.#readReplicaCursor(scope),
-      getReconciliationScopes: (userId) => this.#readReconciliationScopes(userId),
       getPullAttentions: (userId) => this.#readPullAttentions(userId),
       getCommands: (scope) => this.#readCommands(scope),
       getCommandsForUser: (userId) => this.#readCommandsForUser(userId),
@@ -835,11 +824,10 @@ export class IonicOfflineRepository implements OfflineRepository {
   async #applyReplicaTransaction(transaction: OfflineReplicaTransaction, journal: boolean): Promise<void> {
     await this.#assertReplicaSchemaLocked();
     for (const row of transaction.putRows ?? []) this.#validateReplicaRow(row);
-    const [rows, commands, cursors, reconciliationScopes, pullAttentions] = await Promise.all([
+    const [rows, commands, cursors, pullAttentions] = await Promise.all([
       this.#readRecord<OfflineReplicaRow>(ROWS_KEY),
       this.#readRecord<OfflineCommand>(OUTBOX_KEY),
       this.#readRecord<string>(CURSORS_KEY),
-      this.#readRecord<OfflineScope>(RECONCILIATION_SCOPES_KEY),
       this.#readRecord<OfflinePullAttention>(PULL_ATTENTIONS_KEY),
     ]);
     const identityCheckRows = { ...rows };
@@ -890,12 +878,6 @@ export class IonicOfflineRepository implements OfflineRepository {
     for (const cursor of transaction.putCursors ?? []) {
       cursors[this.#cursorKey(cursor)] = cursor.cursor;
     }
-    for (const scope of transaction.putReconciliationScopes ?? []) {
-      reconciliationScopes[this.#cursorKey(scope)] = scope;
-    }
-    for (const scope of transaction.removeReconciliationScopes ?? []) {
-      delete reconciliationScopes[this.#cursorKey(scope)];
-    }
     for (const attention of transaction.putPullAttentions ?? []) {
       pullAttentions[this.#cursorKey(attention)] = attention;
     }
@@ -906,7 +888,6 @@ export class IonicOfflineRepository implements OfflineRepository {
       this.#storage.set(ROWS_KEY, rows),
       this.#storage.set(OUTBOX_KEY, commands),
       this.#storage.set(CURSORS_KEY, cursors),
-      this.#storage.set(RECONCILIATION_SCOPES_KEY, reconciliationScopes),
       this.#storage.set(PULL_ATTENTIONS_KEY, pullAttentions),
     ]);
     await this.#writeAffectedRowPartitions(rows, transaction);