From 4c30d2362efc7ad7d0eb4ede31a3364b5d83db10 Mon Sep 17 00:00:00 2001 From: rdlabo Date: Thu, 13 Aug 2026 16:53:05 +0900 Subject: [PATCH 1/2] Make offline replica mutations atomic --- package.json | 1 + ...offline-aggregate-intent-projector.spec.ts | 1 - ...fline-replica-mutation-coordinator.spec.ts | 16 ++ .../offline-replica-mutation-coordinator.ts | 8 +- .../src/lib/offline-repository-concurrency.ts | 7 + .../kit/offline/src/lib/offline-repository.ts | 31 +-- .../src/lib/offline-sync.service.spec.ts | 73 +----- .../offline/src/lib/offline-sync.service.ts | 25 +- .../src/lib/sqlite-concurrency.node.test.mjs | 55 +++++ .../src/lib/sqlite-offline-repository.spec.ts | 152 +++++++++++- .../src/lib/sqlite-offline-repository.ts | 216 +++++++++++------- 11 files changed, 381 insertions(+), 204 deletions(-) create mode 100644 projects/kit/offline/src/lib/offline-repository-concurrency.ts create mode 100644 projects/kit/offline/src/lib/sqlite-concurrency.node.test.mjs diff --git a/package.json b/package.json index 7e96c67..77b7f62 100644 --- a/package.json +++ b/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" }, diff --git a/projects/kit/offline/src/lib/offline-aggregate-intent-projector.spec.ts b/projects/kit/offline/src/lib/offline-aggregate-intent-projector.spec.ts index d4ce04f..daa0e55 100644 --- a/projects/kit/offline/src/lib/offline-aggregate-intent-projector.spec.ts +++ b/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; diff --git a/projects/kit/offline/src/lib/offline-replica-mutation-coordinator.spec.ts b/projects/kit/offline/src/lib/offline-replica-mutation-coordinator.spec.ts index 40a543c..cbd61fa 100644 --- a/projects/kit/offline/src/lib/offline-replica-mutation-coordinator.spec.ts +++ b/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) => 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(); + }); }); diff --git a/projects/kit/offline/src/lib/offline-replica-mutation-coordinator.ts b/projects/kit/offline/src/lib/offline-replica-mutation-coordinator.ts index b061c3a..67817f0 100644 --- a/projects/kit/offline/src/lib/offline-replica-mutation-coordinator.ts +++ b/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 = Promise.resolve(); /** Enqueues one local replica critical section behind any in-flight mutation. */ run(operation: () => Promise): Promise { - 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) : operation(); + }); this.#tail = mutation.then( () => undefined, () => undefined, diff --git a/projects/kit/offline/src/lib/offline-repository-concurrency.ts b/projects/kit/offline/src/lib/offline-repository-concurrency.ts new file mode 100644 index 0000000..4f477cb --- /dev/null +++ b/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'); diff --git a/projects/kit/offline/src/lib/offline-repository.ts b/projects/kit/offline/src/lib/offline-repository.ts index 425a9dc..e41aaf2 100644 --- a/projects/kit/offline/src/lib/offline-repository.ts +++ b/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 | null>; getReplicaCursor(scope: OfflineScope): Promise; - getReconciliationScopes?(userId: OfflinePrincipalId): Promise; getPullAttentions?(userId: OfflinePrincipalId): Promise; getCommands(scope: OfflineScope): Promise; getCommandsForUser?(userId: OfflinePrincipalId): Promise; @@ -241,7 +237,6 @@ export interface OfflineRepository { identity: OfflineReplicaRemoteIdentity, ): Promise | null>; getReplicaCursor(scope: OfflineScope): Promise; - getReconciliationScopes?(userId: OfflinePrincipalId): Promise; /** Durable fatal-pull attentions for the principal, ordered by scope id. */ getPullAttentions?(userId: OfflinePrincipalId): Promise; getCommands(scope: OfflineScope): Promise; @@ -266,6 +261,8 @@ export interface OfflineRepository { * must not be called from `read`. */ runReadSnapshot(read: (reader: OfflineRepositoryReader) => Promise): Promise; + /** @internal Runs one optimistic read/derive/write operation with platform-specific concurrency validation. */ + [OFFLINE_REPOSITORY_ATOMIC_MUTATION]?(operation: () => Promise): Promise; } /** 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 { - return this.#withCommittedRead(() => this.#readReconciliationScopes(userId)); - } - async getPullAttentions(userId: OfflinePrincipalId): Promise { 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 { - const scopes = await this.#readRecord(RECONCILIATION_SCOPES_KEY); - return Object.values(scopes).filter((scope) => scope.userId === userId); - } - async #readPullAttentions(userId: OfflinePrincipalId): Promise { const attentions = await this.#readRecord(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 { 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(ROWS_KEY), this.#readRecord(OUTBOX_KEY), this.#readRecord(CURSORS_KEY), - this.#readRecord(RECONCILIATION_SCOPES_KEY), this.#readRecord(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); diff --git a/projects/kit/offline/src/lib/offline-sync.service.spec.ts b/projects/kit/offline/src/lib/offline-sync.service.spec.ts index ff759c5..a22a985 100644 --- a/projects/kit/offline/src/lib/offline-sync.service.spec.ts +++ b/projects/kit/offline/src/lib/offline-sync.service.spec.ts @@ -128,7 +128,6 @@ describe('OfflineSyncService', () => { let service: OfflineSyncService; let commands: OfflineCommand[]; let rows: OfflineReplicaRow[]; - let reconciliationScopes: OfflineScope[]; let pullAttentions: OfflinePullAttention[]; let connected: ReturnType>; let session: { userId: number; scopes: OfflineScope[] } | null; @@ -140,9 +139,11 @@ describe('OfflineSyncService', () => { let handleError: ReturnType void>>; let onCommandRemoved: ReturnType Promise>>; let options: OfflineKitOptions; - const execute = vi.fn(async (_command: OfflineCommand, _target: OfflineCommandTarget): Promise => ({ - response: null, - })); + const execute = vi.fn( + async (_command: OfflineCommand, _target: OfflineCommandTarget): Promise => ({ + response: null, + }), + ); const provesCommandNotCommitted = vi.fn((_error: unknown, _command: OfflineCommand) => false); function expectAwaitingPull(count = commands.length): void { @@ -155,7 +156,6 @@ describe('OfflineSyncService', () => { beforeEach(() => { commands = []; rows = []; - reconciliationScopes = []; pullAttentions = []; connected = signal(false); session = { userId: 1, scopes: [{ userId: 1, scopeId: '10' }] }; @@ -257,9 +257,6 @@ describe('OfflineSyncService', () => { ); }), getReplicaCursor: vi.fn(async () => null), - getReconciliationScopes: vi.fn(async (userId: number) => - reconciliationScopes.filter((scope) => scope.userId === userId).map((scope) => ({ ...scope })), - ), getPullAttentions: vi.fn(async (userId: number) => pullAttentions.filter((attention) => attention.userId === userId).map((attention) => structuredClone(attention)), ), @@ -275,14 +272,10 @@ describe('OfflineSyncService', () => { clearUser: vi.fn(async (userId: number) => { commands = commands.filter((item) => item.userId !== userId); rows = rows.filter((item) => item.userId !== userId); - reconciliationScopes = reconciliationScopes.filter((scope) => scope.userId !== userId); pullAttentions = pullAttentions.filter((attention) => attention.userId !== userId); }), clearScope: vi.fn(async (scope: OfflineScope) => { commands = commands.filter((item) => item.userId !== scope.userId || item.scopeId !== scope.scopeId); - reconciliationScopes = reconciliationScopes.filter( - (candidate) => candidate.userId !== scope.userId || candidate.scopeId !== scope.scopeId, - ); pullAttentions = pullAttentions.filter((candidate) => candidate.userId !== scope.userId || candidate.scopeId !== scope.scopeId); }), transactReplica: vi.fn(async (transaction) => { @@ -310,17 +303,6 @@ describe('OfflineSyncService', () => { commands.push(structuredClone(command)); } commands = commands.filter((command) => !(transaction.removeCommandIds ?? []).includes(command.commandId)); - for (const scope of transaction.putReconciliationScopes ?? []) { - reconciliationScopes = reconciliationScopes.filter( - (candidate) => candidate.userId !== scope.userId || candidate.scopeId !== scope.scopeId, - ); - reconciliationScopes.push({ ...scope }); - } - for (const scope of transaction.removeReconciliationScopes ?? []) { - reconciliationScopes = reconciliationScopes.filter( - (candidate) => candidate.userId !== scope.userId || candidate.scopeId !== scope.scopeId, - ); - } for (const attention of transaction.putPullAttentions ?? []) { pullAttentions = pullAttentions.filter( (candidate) => candidate.userId !== attention.userId || candidate.scopeId !== attention.scopeId, @@ -1133,7 +1115,7 @@ describe('OfflineSyncService', () => { expect(execute).toHaveBeenCalledOnce(); }); - it('ACK後pull失敗scopeをreset後もdurable markerから復元しcommandを再送しない', async () => { + it('ACK後pull失敗scopeをreset後もdurable awaiting_pullから復元しcommandを再送しない', async () => { session = { userId: 1, scopes: [ @@ -1161,7 +1143,6 @@ describe('OfflineSyncService', () => { connected.set(true); await vi.waitFor(() => expect(handleError).toHaveBeenCalledWith(postPullError)); expectAwaitingPull(1); - expect(reconciliationScopes).toEqual([{ userId: 1, scopeId: '20' }]); connected.set(false); await service.resetSession(); @@ -1170,7 +1151,7 @@ describe('OfflineSyncService', () => { await vi.waitFor(() => expect(scope20Pulls).toBe(3)); expect(execute).toHaveBeenCalledOnce(); - expect(reconciliationScopes).toEqual([]); + expect(commands).toEqual([expect.objectContaining({ scopeId: '20', state: 'awaiting_pull' })]); }); it('pre-pull: 無関係なscope A失敗でも成功したscope Bのeligible aggregateは送信する', async () => { @@ -1454,7 +1435,6 @@ describe('OfflineSyncService', () => { getReplicaRowByRemoteId: vi.fn(async () => null), getReplicaRowByRemoteIdentity: vi.fn(async () => null), getReplicaCursor: vi.fn(async () => null), - getReconciliationScopes: vi.fn(async () => []), getPullAttentions: vi.fn(async (userId: number) => pullAttentions.filter((attention) => attention.userId === userId).map((attention) => structuredClone(attention)), ), @@ -1796,14 +1776,7 @@ describe('OfflineSyncService', () => { const postPullScopes = [...pullsByScope.entries()].filter(([, count]) => count >= 2).map(([scopeId]) => scopeId); expect(postPullScopes).toHaveLength(1); expect([...pullsByScope.values()].reduce((sum, count) => sum + count, 0)).toBe(3); - // Reconciliation markers remain for later auth/upgrade recovery. - expect(reconciliationScopes).toEqual( - expect.arrayContaining([ - { userId: 1, scopeId: '10' }, - { userId: 1, scopeId: '20' }, - ]), - ); - expect(reconciliationScopes).toHaveLength(2); + expect(new Set(commands.map((command) => command.scopeId))).toEqual(new Set(['10', '20'])); // Fatal must not arm the 1s automatic post-pull flush retry. expect(setTimeoutSpy.mock.calls.some(([, delay]) => delay === 1_000)).toBe(false); @@ -1868,7 +1841,7 @@ describe('OfflineSyncService', () => { expect(postPullAttempts).toBe(2); expect([...pullsByScope.values()].filter((count) => count >= 2)).toHaveLength(2); expect([...pullsByScope.values()].filter((count) => count === 1)).toHaveLength(1); - expect(reconciliationScopes).toHaveLength(3); + expect(new Set(commands.map((command) => command.scopeId))).toEqual(new Set(['10', '20', '30'])); expect(setTimeoutSpy.mock.calls.some(([, delay]) => delay === 1_000)).toBe(false); const pullsAfterFatal = pull.mock.calls.length; @@ -2076,18 +2049,6 @@ describe('OfflineSyncService', () => { expect(execute.mock.calls[0]?.[0]).toMatchObject({ identity: { localId: 'scope-b-lookalike' } }); }); - it('所属から外れたdurable reconciliation scopeをsession discoveryで破棄する', async () => { - reconciliationScopes = [{ userId: 1, scopeId: '20' }]; - session = { userId: 1, scopes: [{ userId: 1, scopeId: '10' }] }; - - await service.refreshSession(['10']); - expect(reconciliationScopes).toEqual([]); - - connected.set(true); - await service.flush(); - expect(pull).not.toHaveBeenCalledWith({ userId: 1, scopeId: '20' }); - }); - it('local_idを不変主キーにして送信直前に最新server_idへ解決する', async () => { execute.mockResolvedValueOnce({ remoteId: 38142, @@ -2944,7 +2905,6 @@ describe('OfflineSyncService', () => { getReplicaRowByRemoteId: vi.fn(async () => null), getReplicaRowByRemoteIdentity: vi.fn(async () => null), getReplicaCursor: vi.fn(async () => null), - getReconciliationScopes: vi.fn(async () => []), getPullAttentions: vi.fn(async () => []), transactReplica: vi.fn(async (transaction) => { for (const command of transaction.putCommands ?? []) { @@ -4590,7 +4550,6 @@ describe('OfflineSyncService', () => { TestBed.resetTestingModule(); commands = []; rows = []; - reconciliationScopes = []; pullAttentions = []; connected = signal(false); session = multiScopeSession; @@ -4651,9 +4610,6 @@ describe('OfflineSyncService', () => { return row ? projectReplicaRow(row, scope) : null; }), getReplicaCursor: vi.fn(async () => null), - getReconciliationScopes: vi.fn(async (userId: number) => - reconciliationScopes.filter((scope) => scope.userId === userId).map((scope) => ({ ...scope })), - ), getPullAttentions: vi.fn(async (userId: number) => pullAttentions.filter((attention) => attention.userId === userId).map((attention) => structuredClone(attention)), ), @@ -4683,17 +4639,6 @@ describe('OfflineSyncService', () => { commands.push(structuredClone(command)); } commands = commands.filter((command) => !(transaction.removeCommandIds ?? []).includes(command.commandId)); - for (const scope of transaction.putReconciliationScopes ?? []) { - reconciliationScopes = reconciliationScopes.filter( - (candidate) => candidate.userId !== scope.userId || candidate.scopeId !== scope.scopeId, - ); - reconciliationScopes.push({ ...scope }); - } - for (const scope of transaction.removeReconciliationScopes ?? []) { - reconciliationScopes = reconciliationScopes.filter( - (candidate) => candidate.userId !== scope.userId || candidate.scopeId !== scope.scopeId, - ); - } for (const attention of transaction.putPullAttentions ?? []) { pullAttentions = pullAttentions.filter( (candidate) => candidate.userId !== attention.userId || candidate.scopeId !== attention.scopeId, diff --git a/projects/kit/offline/src/lib/offline-sync.service.ts b/projects/kit/offline/src/lib/offline-sync.service.ts index de583df..b21576c 100644 --- a/projects/kit/offline/src/lib/offline-sync.service.ts +++ b/projects/kit/offline/src/lib/offline-sync.service.ts @@ -162,7 +162,7 @@ export class OfflineSyncService { readonly #commands = signal([]); readonly #pullAttentions = signal([]); readonly #knownScopes = new Map(); - /** ACKed scopes whose authoritative post-send pull has not completed yet. */ + /** In-memory scheduling cache for scopes whose authoritative post-send pull has not completed yet. */ readonly #pendingPullScopes = new Map(); #activeUserId: OfflinePrincipalId | null = null; #flushPromise: Promise | null = null; @@ -176,7 +176,6 @@ export class OfflineSyncService { #foregroundScopePolicy: readonly string[] | null = null; #initialized = false; #lastCommandCreatedAt = 0; - #coldReconciliationRequired = this.#repository.getReconciliationScopes === undefined; readonly pendingCommands = this.#commands.asReadonly(); readonly pendingCount = computed(() => this.pendingCommands().length); @@ -870,7 +869,7 @@ export class OfflineSyncService { } #beginFlush(explicitFull: boolean): Promise { - const isPartial = !explicitFull && !this.#coldReconciliationRequired && this.#foregroundScopePolicy !== null; + const isPartial = !explicitFull && this.#foregroundScopePolicy !== null; if (this.#flushPromise) { if (explicitFull && this.#partialFlushInFlight) { if (!this.#chainedFullFlush) { @@ -1002,7 +1001,7 @@ export class OfflineSyncService { await this.#refreshState(generation); if (fatalPullFailure !== null) { // Auth/upgrade recovery only — never arm the 1s automatic flush retry. - // Post-send ACK already removed the command; reconciliation markers remain for later recovery. + // Transported commands remain awaiting_pull for later auth/upgrade recovery. // Only the owning generation may clear the timer — a stale fatal must not disarm a // newer session's already-armed retry_wait / post-pull retry. if (this.#isCurrent(generation)) this.#scheduleRetry(null); @@ -1017,7 +1016,6 @@ export class OfflineSyncService { } throw failures[0]; } - if (this.#isCurrent(generation)) this.#coldReconciliationRequired = false; if (this.#isCurrent(generation)) { for (const command of await this.#readKnownCommands()) { if (command.state !== 'awaiting_pull') continue; @@ -1283,7 +1281,6 @@ export class OfflineSyncService { : undefined, removeRows: rematerialized.removeRows, putCommands: [awaitingPull, ...rebased], - putReconciliationScopes: [{ userId: command.userId, scopeId: command.scopeId }], }); const scope = { userId: command.userId, scopeId: command.scopeId }; this.#pendingPullScopes.set(this.#scopeKey(scope), scope); @@ -1704,19 +1701,14 @@ export class OfflineSyncService { } async #restorePendingPullScopes(userId: OfflinePrincipalId, generation: number): Promise { - if (!this.#repository.getReconciliationScopes) return; - const durableScopes = await this.#repository.getReconciliationScopes(userId); + const commands = await this.#readKnownCommands(); if (!this.#isCurrent(generation) || this.#activeUserId !== userId) return; - const currentKeys = new Set(this.#knownScopes.keys()); this.#pendingPullScopes.clear(); - const revoked: OfflineScope[] = []; - for (const scope of durableScopes) { + for (const command of commands) { + if (command.state !== 'awaiting_pull') continue; + const scope = { userId: command.userId, scopeId: command.scopeId }; const key = this.#scopeKey(scope); - if (scope.userId === userId && currentKeys.has(key)) this.#pendingPullScopes.set(key, scope); - else revoked.push(scope); - } - if (revoked.length > 0) { - await this.#repository.transactReplica({ removeReconciliationScopes: revoked }); + this.#pendingPullScopes.set(key, scope); } } @@ -1734,7 +1726,6 @@ export class OfflineSyncService { async #markScopeReconciled(scope: OfflineScope, generation: number): Promise { if (!this.#isCurrent(generation)) return; await this.#repository.transactReplica({ - removeReconciliationScopes: [scope], removePullAttentions: [scope], }); if (!this.#isCurrent(generation)) return; diff --git a/projects/kit/offline/src/lib/sqlite-concurrency.node.test.mjs b/projects/kit/offline/src/lib/sqlite-concurrency.node.test.mjs new file mode 100644 index 0000000..3797225 --- /dev/null +++ b/projects/kit/offline/src/lib/sqlite-concurrency.node.test.mjs @@ -0,0 +1,55 @@ +import assert from 'node:assert/strict'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import test from 'node:test'; +import { DatabaseSync } from 'node:sqlite'; + +test('SQLite data_version detects another connection and a reserved write lock closes the CAS window', () => { + const directory = mkdtempSync(join(tmpdir(), 'offline-sqlite-concurrency-')); + const path = join(directory, 'replica.sqlite'); + const first = new DatabaseSync(path); + let second; + try { + first.exec('PRAGMA journal_mode = WAL; CREATE TABLE replica (id INTEGER PRIMARY KEY, value TEXT NOT NULL)'); + second = new DatabaseSync(path); + second.exec('PRAGMA busy_timeout = 0'); + + const before = first.prepare('PRAGMA data_version').get().data_version; + second.exec("INSERT INTO replica (value) VALUES ('external')"); + const after = first.prepare('PRAGMA data_version').get().data_version; + assert.ok(after > before); + + first.exec('BEGIN; UPDATE replica SET value = value WHERE id = 1'); + const lockedRevision = first.prepare('PRAGMA data_version').get().data_version; + assert.equal(lockedRevision, after); + assert.throws(() => second.exec("INSERT INTO replica (value) VALUES ('racing')"), /locked/u); + first.exec('ROLLBACK'); + } finally { + second?.close(); + first.close(); + rmSync(directory, { recursive: true, force: true }); + } +}); + +test('a SQLite read transaction keeps both reads on one snapshot while another connection commits', () => { + const directory = mkdtempSync(join(tmpdir(), 'offline-sqlite-snapshot-')); + const path = join(directory, 'replica.sqlite'); + const first = new DatabaseSync(path); + let second; + try { + first.exec("PRAGMA journal_mode = WAL; CREATE TABLE replica (id INTEGER PRIMARY KEY, value TEXT NOT NULL); INSERT INTO replica (value) VALUES ('one')"); + second = new DatabaseSync(path); + + first.exec('BEGIN'); + assert.equal(first.prepare('SELECT count(*) AS count FROM replica').get().count, 1); + second.exec("INSERT INTO replica (value) VALUES ('two')"); + assert.equal(first.prepare('SELECT count(*) AS count FROM replica').get().count, 1); + first.exec('COMMIT'); + assert.equal(first.prepare('SELECT count(*) AS count FROM replica').get().count, 2); + } finally { + second?.close(); + first.close(); + rmSync(directory, { recursive: true, force: true }); + } +}); diff --git a/projects/kit/offline/src/lib/sqlite-offline-repository.spec.ts b/projects/kit/offline/src/lib/sqlite-offline-repository.spec.ts index c87fdbb..1600c89 100644 --- a/projects/kit/offline/src/lib/sqlite-offline-repository.spec.ts +++ b/projects/kit/offline/src/lib/sqlite-offline-repository.spec.ts @@ -14,6 +14,7 @@ import { type OfflineReplicaSchemaBundle, } from './offline-replica-schema'; import { canonicalOfflinePrincipalId, type OfflineCommand, type OfflineReplicaRow } from './offline-repository'; +import { OFFLINE_REPOSITORY_ATOMIC_MUTATION } from './offline-repository-concurrency'; import { generatedCommandIdentity, generatedReplicaIdentity, naturalReplicaIdentity } from './offline-test-helpers'; import { COMMUNITY_SQLITE, @@ -293,6 +294,130 @@ describe('SqliteOfflineRepository community sqlite driver', () => { expect(plugin.rollbackTransaction).not.toHaveBeenCalled(); }); + it('別SQLite connectionのcommitをwrite lock取得後に検出し、product callbackを再実行しない', async () => { + let dataVersion = 1; + plugin.query.mockImplementation(async ({ statement }: { statement: string }) => { + if (statement === 'PRAGMA data_version') return { columns: ['data_version'], rows: [[dataVersion]] }; + if (statement.includes('offline_replica_schema_metadata')) { + return { + columns: ['version', 'schema_hash'], + rows: [[storedReplicaMetadata!.version, storedReplicaMetadata!.schemaHash]], + }; + } + if (statement.startsWith('PRAGMA table_info')) return { rows: [{ name: 'next_local_id' }] }; + return { rows: [] }; + }); + const repository = createRepository(); + const operation = vi.fn(async () => { + await repository.getCommands({ userId: 1, scopeId: '10' }); + dataVersion = 2; + await repository.transactReplica({ putCursors: [{ userId: 1, scopeId: '10', cursor: 'stale' }] }); + }); + + await expect(repository[OFFLINE_REPOSITORY_ATOMIC_MUTATION]!(operation)).rejects.toThrow('changed through another SQLite connection'); + + expect(operation).toHaveBeenCalledOnce(); + expect(plugin.rollbackTransaction).toHaveBeenCalledOnce(); + expect( + plugin.execute.mock.calls.some(([options]) => + String((options as { statement: string }).statement).includes('INSERT INTO offline_replica_cursors'), + ), + ).toBe(false); + }); + + it('guarded commitはwrite lockをrevision確認より先に取得し、確認後にreplica mutationを適用する', async () => { + plugin.query.mockImplementation(async ({ statement }: { statement: string }) => { + if (statement === 'PRAGMA data_version') return { columns: ['data_version'], rows: [[1]] }; + if (statement.includes('offline_replica_schema_metadata')) { + return { + columns: ['version', 'schema_hash'], + rows: [[storedReplicaMetadata!.version, storedReplicaMetadata!.schemaHash]], + }; + } + if (statement.startsWith('PRAGMA table_info')) return { rows: [{ name: 'next_local_id' }] }; + return { rows: [] }; + }); + const repository = createRepository(); + + await repository[OFFLINE_REPOSITORY_ATOMIC_MUTATION]!(async () => { + await repository.transactReplica({ putCursors: [{ userId: 1, scopeId: '10', cursor: 'fresh' }] }); + }); + + const lockCall = plugin.execute.mock.calls.find(([options]) => + String((options as { statement: string }).statement).startsWith('UPDATE offline_metadata SET schema_version'), + ); + const revisionCall = plugin.query.mock.calls + .filter(([options]) => (options as { statement: string }).statement === 'PRAGMA data_version') + .at(1); + const cursorCall = plugin.execute.mock.calls.find(([options]) => + String((options as { statement: string }).statement).includes('INSERT INTO offline_replica_cursors'), + ); + expect(lockCall).toBeDefined(); + expect(revisionCall).toBeDefined(); + expect(cursorCall).toBeDefined(); + expect(lockCall![0]).toBeDefined(); + expect(plugin.execute.mock.invocationCallOrder[plugin.execute.mock.calls.indexOf(lockCall!)]).toBeLessThan( + plugin.query.mock.invocationCallOrder[plugin.query.mock.calls.indexOf(revisionCall!)]!, + ); + expect(plugin.query.mock.invocationCallOrder[plugin.query.mock.calls.indexOf(revisionCall!)]).toBeLessThan( + plugin.execute.mock.invocationCallOrder[plugin.execute.mock.calls.indexOf(cursorCall!)]!, + ); + }); + + it('guarded writeのcommit後にexternal revisionが進んでも完了済みproduct operationを失敗扱いにしない', async () => { + let dataVersion = 1; + plugin.query.mockImplementation(async ({ statement }: { statement: string }) => { + if (statement === 'PRAGMA data_version') return { columns: ['data_version'], rows: [[dataVersion]] }; + if (statement.includes('offline_replica_schema_metadata')) { + return { + columns: ['version', 'schema_hash'], + rows: [[storedReplicaMetadata!.version, storedReplicaMetadata!.schemaHash]], + }; + } + if (statement.startsWith('PRAGMA table_info')) return { rows: [{ name: 'next_local_id' }] }; + return { rows: [] }; + }); + const repository = createRepository(); + + await expect( + repository[OFFLINE_REPOSITORY_ATOMIC_MUTATION]!(async () => { + await repository.transactReplica({ putCursors: [{ userId: 1, scopeId: '10', cursor: 'committed' }] }); + dataVersion = 2; + }), + ).resolves.toBeUndefined(); + + expect( + plugin.query.mock.calls.filter(([options]) => (options as { statement: string }).statement === 'PRAGMA data_version'), + ).toHaveLength(2); + }); + + it('同一operation内の2回目のtransactionもexternal revisionを再確認する', async () => { + let dataVersion = 1; + plugin.query.mockImplementation(async ({ statement }: { statement: string }) => { + if (statement === 'PRAGMA data_version') return { columns: ['data_version'], rows: [[dataVersion]] }; + if (statement.includes('offline_replica_schema_metadata')) { + return { + columns: ['version', 'schema_hash'], + rows: [[storedReplicaMetadata!.version, storedReplicaMetadata!.schemaHash]], + }; + } + if (statement.startsWith('PRAGMA table_info')) return { rows: [{ name: 'next_local_id' }] }; + return { rows: [] }; + }); + const repository = createRepository(); + + await expect( + repository[OFFLINE_REPOSITORY_ATOMIC_MUTATION]!(async () => { + await repository.transactReplica({ putCursors: [{ userId: 1, scopeId: '10', cursor: 'first' }] }); + dataVersion = 2; + await repository.transactReplica({ putCursors: [{ userId: 1, scopeId: '10', cursor: 'stale-second' }] }); + }), + ).rejects.toThrow('changed through another SQLite connection'); + + expect(plugin.commitTransaction).toHaveBeenCalledTimes(2); + expect(plugin.rollbackTransaction).toHaveBeenCalledOnce(); + }); + it('pull attentionをput/getしtransactionでupsertする', async () => { const repository = createRepository(); await repository.initialize(); @@ -2030,7 +2155,7 @@ describe('SqliteOfflineRepository replica rows', () => { await expect(repository.runReadSnapshot((reader) => reader.getReplicaCursor({ userId: 1, scopeId: '10' }))).resolves.toBeNull(); }); - it('独立した並行snapshotはそれぞれreader leaseを持ち、writerは両方の完了を待つ', async () => { + it('同一native connectionのsnapshotを直列化し、後続writerは両snapshotの完了を待つ', async () => { const repository = createRepository(); await repository.initialize(); plugin.execute.mockClear(); @@ -2043,11 +2168,15 @@ describe('SqliteOfflineRepository replica rows', () => { const gateB = new Promise((resolve) => { releaseB = resolve; }); - let bothReadersReady: (() => void) | undefined; - const readersReady = new Promise((resolve) => { - bothReadersReady = resolve; + let snapshotAReady: (() => void) | undefined; + const readerAReady = new Promise((resolve) => { + snapshotAReady = resolve; }); - let readersHeld = 0; + let snapshotBReady: (() => void) | undefined; + const readerBReady = new Promise((resolve) => { + snapshotBReady = resolve; + }); + let snapshotBEntered = false; plugin.query.mockImplementation(async ({ statement }: { statement: string }) => { if (statement.includes('offline_replica_schema_metadata')) { @@ -2063,18 +2192,18 @@ describe('SqliteOfflineRepository replica rows', () => { const snapshotA = repository.runReadSnapshot(async (reader) => { await reader.getCommands({ userId: 1, scopeId: '10' }); - readersHeld += 1; - if (readersHeld === 2) bothReadersReady?.(); + snapshotAReady?.(); await gateA; }); const snapshotB = repository.runReadSnapshot(async (reader) => { + snapshotBEntered = true; + snapshotBReady?.(); await reader.getCommands({ userId: 1, scopeId: '10' }); - readersHeld += 1; - if (readersHeld === 2) bothReadersReady?.(); await gateB; }); - await readersReady; + await readerAReady; + expect(snapshotBEntered).toBe(false); let writeFinished = false; const write = repository .putCommand({ @@ -2111,7 +2240,8 @@ describe('SqliteOfflineRepository replica rows', () => { releaseA?.(); await snapshotA; - await Promise.resolve(); + await readerBReady; + expect(snapshotBEntered).toBe(true); expect(writeFinished).toBe(false); releaseB?.(); diff --git a/projects/kit/offline/src/lib/sqlite-offline-repository.ts b/projects/kit/offline/src/lib/sqlite-offline-repository.ts index 5ebcbd7..5eccf26 100644 --- a/projects/kit/offline/src/lib/sqlite-offline-repository.ts +++ b/projects/kit/offline/src/lib/sqlite-offline-repository.ts @@ -45,6 +45,7 @@ import { type OfflineReplicaTransaction, type OfflineScope, } from './offline-repository'; +import { OFFLINE_REPOSITORY_ATOMIC_MUTATION } from './offline-repository-concurrency'; import { OfflineStorageUnavailableError } from './offline-storage'; /** Minimal native SQLite driver surface required by the offline repository. */ @@ -211,6 +212,9 @@ export class SqliteOfflineRepository implements OfflineRepository { #activeReaders = 0; #readersIdle: Promise = Promise.resolve(); #resolveReadersIdle: (() => void) | null = null; + #atomicMutationRevision: number | null = null; + #atomicMutationCommitted = false; + #atomicOperations: Promise = Promise.resolve(); initialize(): Promise { this.#initialization ??= this.#open(); @@ -288,10 +292,6 @@ export class SqliteOfflineRepository implements OfflineRepository { return this.#withCommittedRead(() => this.#readReplicaCursor(scope)); } - async getReconciliationScopes(userId: OfflinePrincipalId): Promise { - return this.#withCommittedRead(() => this.#readReconciliationScopes(userId)); - } - async getPullAttentions(userId: OfflinePrincipalId): Promise { return this.#withCommittedRead(() => this.#readPullAttentions(userId)); } @@ -305,7 +305,34 @@ export class SqliteOfflineRepository implements OfflineRepository { } async runReadSnapshot(read: (reader: OfflineRepositoryReader) => Promise): Promise { - return this.#withCommittedRead(() => read(this.#reader())); + if (this.#atomicMutationRevision !== null) { + return this.#queueAtomicOperation(async () => this.#nativeTransaction(await this.#databaseConnection(), () => read(this.#reader()))); + } + return this.#transaction(() => read(this.#reader())); + } + + async [OFFLINE_REPOSITORY_ATOMIC_MUTATION](operation: () => Promise): Promise { + await this.initialize(); + await this.#writes; + if (this.#atomicMutationRevision !== null) { + throw new Error('Nested offline replica atomic mutations are not supported.'); + } + this.#beginReaders(); + const databaseId = await this.#databaseConnection(); + this.#atomicOperations = Promise.resolve(); + this.#atomicMutationCommitted = false; + this.#atomicMutationRevision = await this.#nativeTransaction(databaseId, () => this.#dataVersion(databaseId)); + try { + const result = await operation(); + await this.#atomicOperations; + if (!this.#atomicMutationCommitted) { + await this.#queueAtomicOperation(() => this.#atomicTransaction(databaseId, async () => undefined, false)); + } + return result; + } finally { + this.#atomicMutationRevision = null; + this.#endReaders(); + } } async putCommand(command: OfflineCommand): Promise { @@ -381,68 +408,57 @@ export class SqliteOfflineRepository implements OfflineRepository { async transactReplica(transaction: OfflineReplicaTransaction): Promise { for (const row of transaction.putRows ?? []) this.#validateReplicaRow(row); - await this.#transaction(async (databaseId) => { - const releases = new Map(); - for (const release of transaction.releaseRemoteIds ?? []) { - this.#assertValidReleaseRemoteId(release.remoteId); - const key = this.#replicaRowKey(release); - if (releases.has(key)) { - throw new Error( - `Offline replica remoteId release is duplicated for ${release.sourceKey}/${canonicalOfflineReplicaIdentity(release.identity)}.`, - ); - } - releases.set(key, release); - } - const consumedReleases = new Set(); - for (const row of transaction.putRows ?? []) { - const key = this.#replicaRowKey(row); - const release = releases.get(key); - await this.#putReplicaRow(databaseId, row, release); - if (release) consumedReleases.add(key); - } - if (consumedReleases.size !== releases.size) { - throw new Error('Offline replica remoteId release must match an existing row in putRows.'); - } - for (const row of transaction.removeRows ?? []) await this.#removeReplicaRow(databaseId, row); - for (const command of transaction.putCommands ?? []) await this.#putCommand(databaseId, command); - for (const commandId of transaction.removeCommandIds ?? []) { - await this.#execute(databaseId, 'DELETE FROM offline_sync_commands WHERE command_id = ?', [commandId]); - } - for (const cursor of transaction.putCursors ?? []) await this.#putReplicaCursor(databaseId, cursor); - for (const scope of transaction.putReconciliationScopes ?? []) { - await this.#execute( - databaseId, - `INSERT INTO offline_reconciliation_scopes (user_id, scope_id) VALUES (?, ?) - ON CONFLICT(user_id, scope_id) DO NOTHING`, - [canonicalOfflinePrincipalId(scope.userId), scope.scopeId], + const apply = (databaseId: string): Promise => this.#applyReplicaTransaction(databaseId, transaction); + await this.#transaction(apply); + } + + async #applyReplicaTransaction(databaseId: string, transaction: OfflineReplicaTransaction): Promise { + const releases = new Map(); + for (const release of transaction.releaseRemoteIds ?? []) { + this.#assertValidReleaseRemoteId(release.remoteId); + const key = this.#replicaRowKey(release); + if (releases.has(key)) { + throw new Error( + `Offline replica remoteId release is duplicated for ${release.sourceKey}/${canonicalOfflineReplicaIdentity(release.identity)}.`, ); } - for (const scope of transaction.removeReconciliationScopes ?? []) { - await this.#execute(databaseId, 'DELETE FROM offline_reconciliation_scopes WHERE user_id = ? AND scope_id = ?', [ - canonicalOfflinePrincipalId(scope.userId), - scope.scopeId, - ]); - } - for (const attention of transaction.putPullAttentions ?? []) { - await this.#execute( - databaseId, - `INSERT INTO offline_pull_attentions (user_id, scope_id, reason, status) VALUES (?, ?, ?, ?) + releases.set(key, release); + } + const consumedReleases = new Set(); + for (const row of transaction.putRows ?? []) { + const key = this.#replicaRowKey(row); + const release = releases.get(key); + await this.#putReplicaRow(databaseId, row, release); + if (release) consumedReleases.add(key); + } + if (consumedReleases.size !== releases.size) { + throw new Error('Offline replica remoteId release must match an existing row in putRows.'); + } + for (const row of transaction.removeRows ?? []) await this.#removeReplicaRow(databaseId, row); + for (const command of transaction.putCommands ?? []) await this.#putCommand(databaseId, command); + for (const commandId of transaction.removeCommandIds ?? []) { + await this.#execute(databaseId, 'DELETE FROM offline_sync_commands WHERE command_id = ?', [commandId]); + } + for (const cursor of transaction.putCursors ?? []) await this.#putReplicaCursor(databaseId, cursor); + for (const attention of transaction.putPullAttentions ?? []) { + await this.#execute( + databaseId, + `INSERT INTO offline_pull_attentions (user_id, scope_id, reason, status) VALUES (?, ?, ?, ?) ON CONFLICT(user_id, scope_id) DO UPDATE SET reason = excluded.reason, status = excluded.status`, - [ - canonicalOfflinePrincipalId(attention.userId), - attention.scopeId, - attention.reason, - attention.status === undefined ? null : attention.status, - ], - ); - } - for (const scope of transaction.removePullAttentions ?? []) { - await this.#execute(databaseId, 'DELETE FROM offline_pull_attentions WHERE user_id = ? AND scope_id = ?', [ - canonicalOfflinePrincipalId(scope.userId), - scope.scopeId, - ]); - } - }); + [ + canonicalOfflinePrincipalId(attention.userId), + attention.scopeId, + attention.reason, + attention.status === undefined ? null : attention.status, + ], + ); + } + for (const scope of transaction.removePullAttentions ?? []) { + await this.#execute(databaseId, 'DELETE FROM offline_pull_attentions WHERE user_id = ? AND scope_id = ?', [ + canonicalOfflinePrincipalId(scope.userId), + scope.scopeId, + ]); + } } async #open(): Promise { @@ -580,11 +596,12 @@ export class SqliteOfflineRepository implements OfflineRepository { ); } - async #nativeTransaction(databaseId: string, run: () => Promise): Promise { + async #nativeTransaction(databaseId: string, run: () => Promise): Promise { await this.#sqlite!.beginTransaction({ databaseId }); try { - await run(); + const result = await run(); await this.#sqlite!.commitTransaction({ databaseId }); + return result; } catch (error) { await this.#sqlite!.rollbackTransaction({ databaseId }); throw error; @@ -674,13 +691,6 @@ export class SqliteOfflineRepository implements OfflineRepository { return { ...scope, cursor: this.#string(row['cursor']) }; } - async #readReconciliationScopes(userId: OfflinePrincipalId): Promise { - const rows = await this.#query('SELECT scope_id FROM offline_reconciliation_scopes WHERE user_id = ? ORDER BY scope_id', [ - canonicalOfflinePrincipalId(userId), - ]); - return rows.map((row) => ({ userId, scopeId: this.#string(row['scope_id']) })); - } - async #readPullAttentions(userId: OfflinePrincipalId): Promise { const rows = await this.#query('SELECT scope_id, reason, status FROM offline_pull_attentions WHERE user_id = ? ORDER BY scope_id', [ canonicalOfflinePrincipalId(userId), @@ -735,7 +745,6 @@ export class SqliteOfflineRepository 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), @@ -744,6 +753,9 @@ export class SqliteOfflineRepository implements OfflineRepository { async #withCommittedRead(operation: () => Promise): Promise { await this.initialize(); + if (this.#atomicMutationRevision !== null) { + return this.#queueAtomicOperation(operation); + } await this.#writes; this.#beginReaders(); try { @@ -780,6 +792,9 @@ export class SqliteOfflineRepository implements OfflineRepository { } #queueWrite(run: (databaseId: string) => Promise): Promise { + if (this.#atomicMutationRevision !== null) { + return this.#queueAtomicOperation(async () => this.#atomicTransaction(await this.#databaseConnection(), run)); + } const write = this.#writes.then(async (): Promise => { if (this.#activeReaders > 0) await this.#readersIdle; await run(await this.#databaseConnection()); @@ -788,21 +803,52 @@ export class SqliteOfflineRepository implements OfflineRepository { return write; } - #transaction(run: (databaseId: string) => Promise): Promise { - const write = this.#writes.then(async (): Promise => { + #transaction(run: (databaseId: string) => Promise): Promise { + if (this.#atomicMutationRevision !== null) { + return this.#queueAtomicOperation(async () => this.#atomicTransaction(await this.#databaseConnection(), run)); + } + const transaction = this.#writes.then(async (): Promise => { if (this.#activeReaders > 0) await this.#readersIdle; const databaseId = await this.#databaseConnection(); - await this.#sqlite!.beginTransaction({ databaseId }); - try { - await run(databaseId); - await this.#sqlite!.commitTransaction({ databaseId }); - } catch (error) { - await this.#sqlite!.rollbackTransaction({ databaseId }); - throw error; + return this.#nativeTransaction(databaseId, () => run(databaseId)); + }); + this.#writes = transaction.then( + () => undefined, + () => undefined, + ); + return transaction; + } + + async #atomicTransaction(databaseId: string, run: (databaseId: string) => Promise, marksCommit = true): Promise { + const expected = this.#atomicMutationRevision; + if (expected === null) throw new Error('Offline replica atomic mutation is not active.'); + const result = await this.#nativeTransaction(databaseId, async () => { + // A write, even when it leaves the value unchanged, obtains SQLite's + // RESERVED lock before the revision check. No second connection can + // commit between this check and the transaction commit. + await this.#execute(databaseId, 'UPDATE offline_metadata SET schema_version = schema_version WHERE id = 1'); + const actual = await this.#dataVersion(databaseId); + if (actual !== expected) { + throw new Error('Offline replica changed through another SQLite connection; retry the operation from fresh state.'); } + return run(databaseId); }); - this.#writes = write.catch((): void => undefined); - return write; + if (marksCommit) this.#atomicMutationCommitted = true; + return result; + } + + #queueAtomicOperation(run: () => Promise): Promise { + const operation = this.#atomicOperations.then(run); + this.#atomicOperations = operation.then( + () => undefined, + () => undefined, + ); + return operation; + } + + async #dataVersion(databaseId: string): Promise { + const rows = await this.#queryDatabase(databaseId, 'PRAGMA data_version'); + return this.#number(rows[0]?.['data_version']); } #command(row: SQLiteRow): OfflineCommand { From d10ac6c9046d65d30d1fa81315e717e1d504268d Mon Sep 17 00:00:00 2001 From: rdlabo Date: Thu, 13 Aug 2026 18:47:05 +0900 Subject: [PATCH 2/2] Serialize failed command persistence --- .../offline/src/lib/offline-sync.service.ts | 22 ++++++++++--------- 1 file changed, 12 insertions(+), 10 deletions(-) diff --git a/projects/kit/offline/src/lib/offline-sync.service.ts b/projects/kit/offline/src/lib/offline-sync.service.ts index b21576c..0d38dd8 100644 --- a/projects/kit/offline/src/lib/offline-sync.service.ts +++ b/projects/kit/offline/src/lib/offline-sync.service.ts @@ -1426,16 +1426,18 @@ export class OfflineSyncService { serverCommitUnknown = true, ): Promise { const failed = this.#failedCommand(command, error, serverCommitUnknown); - const current = row === undefined ? await this.#rowForCommand(command) : row; - if (!this.#isCurrent(generation)) return; - if (current) { - await this.#repository.transactReplica({ - putRows: [{ ...current, syncState: this.#replicaState(failed.state) }], - putCommands: [failed], - }); - } else { - await this.#repository.putCommand(failed); - } + await this.#serializeReplicaMutation(async () => { + const current = row === undefined ? await this.#rowForCommand(command) : row; + if (!this.#isCurrent(generation)) return; + if (current) { + await this.#repository.transactReplica({ + putRows: [{ ...current, syncState: this.#replicaState(failed.state) }], + putCommands: [failed], + }); + } else { + await this.#repository.putCommand(failed); + } + }); if (!this.#isCurrent(generation)) return; if (failed.state === 'retry_wait') this.#scheduleRetry(failed.retryAt); await this.#refreshState(generation);