From 8f57127212bb5bdfa09ca4ddd28ffd33d4ddbed2 Mon Sep 17 00:00:00 2001 From: rdlabo Date: Fri, 14 Aug 2026 00:33:17 +0900 Subject: [PATCH] Fix offline atomic startup cleanup --- .../src/lib/sqlite-offline-repository.spec.ts | 43 +++++++++++++++++++ .../src/lib/sqlite-offline-repository.ts | 14 +++--- 2 files changed, 50 insertions(+), 7 deletions(-) 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 10e895d..33fd1d9 100644 --- a/projects/kit/offline/src/lib/sqlite-offline-repository.spec.ts +++ b/projects/kit/offline/src/lib/sqlite-offline-repository.spec.ts @@ -325,6 +325,49 @@ describe('SqliteOfflineRepository community sqlite driver', () => { ).toBe(false); }); + it('initial revision read失敗後もreader leaseとwrite gateを解放する', async () => { + const repository = createRepository(); + await repository.initialize(); + const operation = vi.fn(async () => undefined); + plugin.query.mockImplementation(async ({ statement }: { statement: string }) => { + if (statement === 'PRAGMA data_version') throw new Error('temporary SQLite read failure'); + 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: [] }; + }); + + await expect(repository[OFFLINE_REPOSITORY_ATOMIC_MUTATION]!(operation)).rejects.toThrow('temporary SQLite read failure'); + expect(operation).not.toHaveBeenCalled(); + + plugin.query.mockImplementation(async ({ statement }: { statement: string }) => { + if (statement === 'PRAGMA data_version') return { columns: ['data_version'], rows: [[1]] }; + return { rows: [] }; + }); + await expect( + repository.putCommand({ + userId: 1, + scopeId: '10', + commandId: 'after-startup-failure', + aggregateType: 'test_items', + sourceKey: 'test_items', + identity: { kind: 'generated', localId: '019d-after-failure' }, + operation: 'test_items.update', + payload: {}, + baseRevision: null, + state: 'pending', + attempts: 0, + retryAt: null, + createdAt: 1, + lastErrorCode: null, + }), + ).resolves.toBeUndefined(); + }); + 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]] }; diff --git a/projects/kit/offline/src/lib/sqlite-offline-repository.ts b/projects/kit/offline/src/lib/sqlite-offline-repository.ts index 85c04d7..a7f6e1e 100644 --- a/projects/kit/offline/src/lib/sqlite-offline-repository.ts +++ b/projects/kit/offline/src/lib/sqlite-offline-repository.ts @@ -328,14 +328,14 @@ export class SqliteOfflineRepository implements OfflineRepository { 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.#atomicIdle = new Promise((resolve) => { - this.#resolveAtomicIdle = resolve; - }); - this.#atomicMutationRevision = await this.#nativeTransaction(databaseId, () => this.#dataVersion(databaseId)); try { + const databaseId = await this.#databaseConnection(); + this.#atomicOperations = Promise.resolve(); + this.#atomicMutationCommitted = false; + this.#atomicIdle = new Promise((resolve) => { + this.#resolveAtomicIdle = resolve; + }); + this.#atomicMutationRevision = await this.#nativeTransaction(databaseId, () => this.#dataVersion(databaseId)); const result = await operation(this.#atomicRepository()); await this.#atomicOperations; if (!this.#atomicMutationCommitted) {