From 48b40891ae323b861d95e55c7acc85044e5feebf Mon Sep 17 00:00:00 2001 From: rdlabo Date: Fri, 14 Aug 2026 00:07:04 +0900 Subject: [PATCH] Fix offline SQLite atomic ownership --- ...fline-replica-mutation-coordinator.spec.ts | 15 +- .../offline-replica-mutation-coordinator.ts | 6 +- .../src/lib/offline-replica-pull.service.ts | 10 +- .../kit/offline/src/lib/offline-repository.ts | 2 +- .../offline/src/lib/offline-sync.service.ts | 71 ++++---- .../src/lib/sqlite-offline-repository.spec.ts | 108 +++++++++++-- .../src/lib/sqlite-offline-repository.ts | 152 +++++++++++++----- 7 files changed, 269 insertions(+), 95 deletions(-) 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 cbd61fa..012cc43 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 @@ -2,10 +2,13 @@ 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'; +import { OFFLINE_REPOSITORY, type OfflineRepository } from './offline-repository'; describe('OfflineReplicaMutationCoordinator', () => { it('serializes local apply sections and releases the lane after failure', async () => { + TestBed.configureTestingModule({ + providers: [{ provide: OFFLINE_REPOSITORY, useValue: {} }], + }); const coordinator = TestBed.inject(OfflineReplicaMutationCoordinator); let release!: () => void; const gate = new Promise((resolve) => (release = resolve)); @@ -32,14 +35,16 @@ describe('OfflineReplicaMutationCoordinator', () => { }); it('uses the repository atomic-mutation capability without retrying the product operation', async () => { - const atomicMutation = vi.fn(async (operation: () => Promise) => operation()); + const repository = { getCommands: vi.fn() }; + const atomicMutation = vi.fn(async (operation: (owner: typeof repository) => Promise) => operation(repository)); TestBed.configureTestingModule({ - providers: [{ provide: OFFLINE_REPOSITORY, useValue: { [OFFLINE_REPOSITORY_ATOMIC_MUTATION]: atomicMutation } }], + providers: [{ provide: OFFLINE_REPOSITORY, useValue: { ...repository, [OFFLINE_REPOSITORY_ATOMIC_MUTATION]: atomicMutation } }], }); const coordinator = TestBed.inject(OfflineReplicaMutationCoordinator); - const operation = vi.fn(async () => 'done'); + const operation = vi.fn(async (_owner: OfflineRepository) => 'done'); - await expect(coordinator.run(operation)).resolves.toBe('done'); + await expect(coordinator.run((owner) => operation(owner))).resolves.toBe('done'); + expect(operation).toHaveBeenCalledWith(repository); 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 67817f0..4a1ed0f 100644 --- a/projects/kit/offline/src/lib/offline-replica-mutation-coordinator.ts +++ b/projects/kit/offline/src/lib/offline-replica-mutation-coordinator.ts @@ -15,6 +15,7 @@ import { type OfflineCommand, type OfflineReplicaRow, type OfflineReplicaRowKey, + type OfflineRepository, type OfflineScope, } from './offline-repository'; import type { OfflineReplicaEntitySchema } from './offline-replica-schema'; @@ -36,10 +37,11 @@ export class OfflineReplicaMutationCoordinator { #tail: Promise = Promise.resolve(); /** Enqueues one local replica critical section behind any in-flight mutation. */ - run(operation: () => Promise): Promise { + run(operation: (repository: OfflineRepository) => Promise): Promise { const mutation = this.#tail.then(() => { + if (!this.#repository) throw new Error('Offline repository is not configured.'); const atomicMutation = this.#repository?.[OFFLINE_REPOSITORY_ATOMIC_MUTATION]; - return atomicMutation ? (atomicMutation.call(this.#repository, operation) as Promise) : operation(); + return atomicMutation ? (atomicMutation.call(this.#repository, operation) as Promise) : operation(this.#repository); }); this.#tail = mutation.then( () => undefined, diff --git a/projects/kit/offline/src/lib/offline-replica-pull.service.ts b/projects/kit/offline/src/lib/offline-replica-pull.service.ts index 644ef7f..e378bd5 100644 --- a/projects/kit/offline/src/lib/offline-replica-pull.service.ts +++ b/projects/kit/offline/src/lib/offline-replica-pull.service.ts @@ -114,11 +114,11 @@ export class OfflineReplicaPullService { continue; } - const applied = await this.#replicaMutations.run(async () => { - const currentCursor = (await this.#repository.getReplicaCursor(scope))?.cursor ?? ''; + const applied = await this.#replicaMutations.run(async (repository) => { + const currentCursor = (await repository.getReplicaCursor(scope))?.cursor ?? ''; if (currentCursor !== persistedCursor) return currentCursor; - const scopeCommands = await this.#repository.getCommands(scope); - const userCommands = this.#repository.getCommandsForUser ? await this.#repository.getCommandsForUser(scope.userId) : scopeCommands; + const scopeCommands = await repository.getCommands(scope); + const userCommands = repository.getCommandsForUser ? await repository.getCommandsForUser(scope.userId) : scopeCommands; const changes = this.#collapseChanges(page.changes); const projection = await this.#projector?.project({ scope, @@ -279,7 +279,7 @@ export class OfflineReplicaPullService { scopeCommands.find((command) => command.commandId === commandId), ) .filter((command): command is OfflineCommand => command != null); - await this.#repository.transactReplica({ + await repository.transactReplica({ putRows: finalRows.putRows, removeRows: finalRows.removeRows, putCommands: [...putCommands.values()], diff --git a/projects/kit/offline/src/lib/offline-repository.ts b/projects/kit/offline/src/lib/offline-repository.ts index e41aaf2..ed4b67e 100644 --- a/projects/kit/offline/src/lib/offline-repository.ts +++ b/projects/kit/offline/src/lib/offline-repository.ts @@ -262,7 +262,7 @@ export interface OfflineRepository { */ 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; + [OFFLINE_REPOSITORY_ATOMIC_MUTATION]?(operation: (repository: OfflineRepository) => Promise): Promise; } /** DI token for the selected platform repository. */ diff --git a/projects/kit/offline/src/lib/offline-sync.service.ts b/projects/kit/offline/src/lib/offline-sync.service.ts index 0d38dd8..e998f7c 100644 --- a/projects/kit/offline/src/lib/offline-sync.service.ts +++ b/projects/kit/offline/src/lib/offline-sync.service.ts @@ -271,7 +271,7 @@ export class OfflineSyncService { enqueue(request: EnqueueOfflineCommand, options: { flush?: boolean } = {}): Promise { const generation = this.#generation; - return this.#serializeReplicaMutation(() => this.#enqueue(request, options, generation)); + return this.#serializeReplicaMutation((repository) => this.#enqueue(request, options, generation, undefined, repository)); } /** @@ -283,11 +283,11 @@ export class OfflineSyncService { options: { flush?: boolean } = {}, ): Promise { const generation = this.#generation; - return this.#serializeReplicaMutation(async () => { + return this.#serializeReplicaMutation(async (repository) => { await this.initialize(); if (!this.#isCurrent(generation)) throw new Error('Offline session changed before prepared enqueue.'); - const prepared = await prepare(this.#repository); - return this.#enqueue(prepared.request, options, generation); + const prepared = await prepare(repository); + return this.#enqueue(prepared.request, options, generation, undefined, repository); }); } @@ -304,11 +304,11 @@ export class OfflineSyncService { options: PreparedOfflineBatchOptions = {}, ): Promise { const generation = this.#generation; - return this.#serializeReplicaMutation(async () => { + return this.#serializeReplicaMutation(async (repository) => { await this.initialize(); if (!this.#isCurrent(generation)) throw new Error('Offline session changed before prepared batch enqueue.'); - const prepared = await prepare(this.#repository); - return this.#enqueuePreparedBatch(prepared, options, generation); + const prepared = await prepare(repository); + return this.#enqueuePreparedBatch(prepared, options, generation, repository); }); } @@ -325,7 +325,7 @@ export class OfflineSyncService { options: { flush?: boolean } = {}, ): Promise { const generation = this.#generation; - return this.#serializeReplicaMutation(async () => { + return this.#serializeReplicaMutation(async (repository) => { await this.initialize(); if (!this.#isCurrent(generation)) throw new Error('Offline session changed before prepared replacement.'); const knownCommands = await this.#readKnownCommands(); @@ -337,8 +337,8 @@ export class OfflineSyncService { ) { throw new Error('Offline replacement requires the command to be the only pending intent for its aggregate.'); } - const prepared = await prepare(this.#repository); - return this.#enqueue(prepared.request, options, generation, replaced); + const prepared = await prepare(repository); + return this.#enqueue(prepared.request, options, generation, replaced, repository); }); } @@ -355,7 +355,7 @@ export class OfflineSyncService { options: PreparedOfflineBatchOptions = {}, ): Promise { const generation = this.#generation; - return this.#serializeReplicaMutation(async () => { + return this.#serializeReplicaMutation(async (repository) => { await this.initialize(); if (!this.#isCurrent(generation)) throw new Error('Offline session changed before prepared aggregate replacement.'); const knownCommands = await this.#readKnownCommands(); @@ -364,7 +364,7 @@ export class OfflineSyncService { const aggregateKey = this.#aggregateKey(selected); const replaced = knownCommands.filter((command) => this.#aggregateKey(command) === aggregateKey); this.#assertDiscardable(replaced); - const prepared = await prepare(this.#repository, replaced); + const prepared = await prepare(repository, replaced); if (prepared.length !== replaced.length) { throw new Error('Offline aggregate replacement must preserve the ordered intent count.'); } @@ -388,6 +388,7 @@ export class OfflineSyncService { generation, options, replaced.map((command) => command.commandId), + repository, ); return materializations.map((item) => item.command.commandId); }); @@ -411,16 +412,16 @@ export class OfflineSyncService { */ runSerializedReplicaMutation(operation: (repository: OfflineRepository) => Promise): Promise { const generation = this.#generation; - return this.#serializeReplicaMutation(async () => { + return this.#serializeReplicaMutation(async (repository) => { await this.initialize(); if (!this.#isCurrent(generation)) throw new Error('Offline session changed before serialized replica mutation.'); - const result = await operation(this.#repository); + const result = await operation(repository); if (this.#isCurrent(generation)) await this.#refreshState(generation); return result; }); } - #serializeReplicaMutation(operation: () => Promise): Promise { + #serializeReplicaMutation(operation: (repository: OfflineRepository) => Promise): Promise { return this.#replicaMutations.run(operation); } @@ -429,6 +430,7 @@ export class OfflineSyncService { options: { flush?: boolean }, generation: number, replaced?: OfflineCommand, + repository: OfflineRepository = this.#repository, ): Promise { const session = await this.#beginEnqueueSession(generation); this.#assertEnqueueScope(session, request.scopeId); @@ -442,7 +444,7 @@ export class OfflineSyncService { replaced ? [replaced.commandId] : undefined, currentCommands, ); - await this.#commitMaterializedEnqueues([materialization], generation, options, replaced ? [replaced.commandId] : undefined); + await this.#commitMaterializedEnqueues([materialization], generation, options, replaced ? [replaced.commandId] : undefined, repository); return materialization.command.commandId; } @@ -450,6 +452,7 @@ export class OfflineSyncService { prepared: readonly PreparedOfflineCommand[], options: PreparedOfflineBatchOptions, generation: number, + repository: OfflineRepository, ): Promise { if (prepared.length === 0) { throw new Error('Prepared offline batch must contain at least one command.'); @@ -472,7 +475,7 @@ export class OfflineSyncService { currentCommands, ); options.assertCurrent?.(); - await this.#commitMaterializedEnqueues(materializations, generation, options); + await this.#commitMaterializedEnqueues(materializations, generation, options, undefined, repository); return materializations.map((item) => item.command.commandId); } @@ -681,6 +684,7 @@ export class OfflineSyncService { generation: number, options: { flush?: boolean }, removeCommandIds?: readonly string[], + repository: OfflineRepository = this.#repository, ): Promise { if (generation !== this.#generation) { throw new Error('Offline session changed before the command could be persisted'); @@ -697,7 +701,7 @@ export class OfflineSyncService { if (entry.seedBaseRow !== undefined) seeds.set(this.#aggregateKey(entry.command), entry.seedBaseRow); } const rematerialized = await this.#rematerializeAffectedAggregates(affected, remaining, seeds); - await this.#repository.transactReplica({ + await repository.transactReplica({ putRows: rematerialized.putRows, removeRows: rematerialized.removeRows, putCommands: entries.map((entry) => entry.command), @@ -797,12 +801,12 @@ export class OfflineSyncService { // and project the discard inside the same local mutation lane used by // enqueue, ACK, and pull application so an old before-image cannot replace // a newer authoritative row after its cursor has advanced. - const command = await this.#replicaMutations.run(async () => { + const command = await this.#replicaMutations.run(async (repository) => { const current = (await this.#readKnownCommands()).find((item) => item.commandId === commandId); if (!current) return null; this.#assertDiscardable([current]); this.#invalidateFlush(); - await this.#discardCommands([current]); + await this.#discardCommands([current], repository); return current; }); if (!command) return; @@ -840,11 +844,11 @@ export class OfflineSyncService { async discardAllPending(): Promise { await this.initialize(); - const commands = await this.#replicaMutations.run(async () => { + const commands = await this.#replicaMutations.run(async (repository) => { const current = await this.#readKnownCommands(); this.#assertDiscardable(current); this.#invalidateFlush(); - await this.#discardCommands(current); + await this.#discardCommands(current, repository); return current; }); await this.#refreshState(); @@ -1193,7 +1197,7 @@ export class OfflineSyncService { result: OfflineCommandResult, generation: number, ): Promise { - return this.#serializeReplicaMutation(() => this.#completeCommandLocked(commands, command, result, generation)); + return this.#serializeReplicaMutation((repository) => this.#completeCommandLocked(commands, command, result, generation, repository)); } async #completeCommandLocked( @@ -1201,6 +1205,7 @@ export class OfflineSyncService { command: OfflineCommand, result: OfflineCommandResult, generation: number, + repository: OfflineRepository, ): Promise { if (result.clearRemoteId === true && result.remoteId !== undefined) { throw new Error('Offline command cannot return remoteId and clearRemoteId together.'); @@ -1265,7 +1270,7 @@ export class OfflineSyncService { }; const rematerialized = await this.#rematerializeAggregate(awaitingPull, latestCommands, identityUpdatedBase); if (!this.#isCurrent(generation)) return; - await this.#repository.transactReplica({ + await repository.transactReplica({ putRows: rematerialized.putRows, releaseRemoteIds: result.clearRemoteId === true && current.identity.kind === 'generated' && current.identity.remoteId !== null @@ -1315,14 +1320,14 @@ export class OfflineSyncService { return schema; } - async #discardCommands(discarded: readonly OfflineCommand[]): Promise { + async #discardCommands(discarded: readonly OfflineCommand[], repository: OfflineRepository): Promise { const all = await this.#readKnownCommands(); const discardedIds = new Set(discarded.map((command) => command.commandId)); const affected = new Map(); for (const command of discarded) affected.set(this.#aggregateKey(command), command); const remaining = all.filter((item) => !discardedIds.has(item.commandId)); const rematerialized = await this.#rematerializeAffectedAggregates(affected, remaining); - await this.#repository.transactReplica({ + await repository.transactReplica({ putRows: rematerialized.putRows, removeRows: rematerialized.removeRows, removeCommandIds: [...discardedIds], @@ -1426,16 +1431,16 @@ export class OfflineSyncService { serverCommitUnknown = true, ): Promise { const failed = this.#failedCommand(command, error, serverCommitUnknown); - await this.#serializeReplicaMutation(async () => { + await this.#serializeReplicaMutation(async (repository) => { const current = row === undefined ? await this.#rowForCommand(command) : row; if (!this.#isCurrent(generation)) return; if (current) { - await this.#repository.transactReplica({ + await repository.transactReplica({ putRows: [{ ...current, syncState: this.#replicaState(failed.state) }], putCommands: [failed], }); } else { - await this.#repository.putCommand(failed); + await repository.putCommand(failed); } }); if (!this.#isCurrent(generation)) return; @@ -1667,7 +1672,7 @@ export class OfflineSyncService { } #claimSendingCommand(command: OfflineCommand, generation: number): Promise { - const transition = this.#serializeReplicaMutation(async () => { + const transition = this.#serializeReplicaMutation(async (repository) => { if (!this.#isCurrent(generation)) return null; const scope = { userId: command.userId, scopeId: command.scopeId }; const current = (await this.#repository.getCommands(scope)).find((candidate) => candidate.commandId === command.commandId); @@ -1679,7 +1684,7 @@ export class OfflineSyncService { retryAt: null, lastErrorCode: null, }; - await this.#repository.putCommand(sending); + await repository.putCommand(sending); return sending; }); this.#sendingTransitions.add(transition); @@ -1691,13 +1696,13 @@ export class OfflineSyncService { } #markTransportStarted(command: OfflineCommand, generation: number): Promise { - return this.#serializeReplicaMutation(async () => { + return this.#serializeReplicaMutation(async (repository) => { if (!this.#isCurrent(generation)) return null; const scope = { userId: command.userId, scopeId: command.scopeId }; const current = (await this.#repository.getCommands(scope)).find((candidate) => candidate.commandId === command.commandId); if (!current || current.state !== 'sending') return null; const transportCommand = { ...current, serverCommitUnknown: true }; - await this.#repository.putCommand(transportCommand); + await repository.putCommand(transportCommand); return transportCommand; }); } 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 1600c89..10e895d 100644 --- a/projects/kit/offline/src/lib/sqlite-offline-repository.spec.ts +++ b/projects/kit/offline/src/lib/sqlite-offline-repository.spec.ts @@ -13,7 +13,7 @@ import { text, type OfflineReplicaSchemaBundle, } from './offline-replica-schema'; -import { canonicalOfflinePrincipalId, type OfflineCommand, type OfflineReplicaRow } from './offline-repository'; +import { canonicalOfflinePrincipalId, type OfflineCommand, type OfflineReplicaRow, type OfflineRepository } from './offline-repository'; import { OFFLINE_REPOSITORY_ATOMIC_MUTATION } from './offline-repository-concurrency'; import { generatedCommandIdentity, generatedReplicaIdentity, naturalReplicaIdentity } from './offline-test-helpers'; import { @@ -308,10 +308,10 @@ describe('SqliteOfflineRepository community sqlite driver', () => { return { rows: [] }; }); const repository = createRepository(); - const operation = vi.fn(async () => { - await repository.getCommands({ userId: 1, scopeId: '10' }); + const operation = vi.fn(async (owner: OfflineRepository) => { + await owner.getCommands({ userId: 1, scopeId: '10' }); dataVersion = 2; - await repository.transactReplica({ putCursors: [{ userId: 1, scopeId: '10', cursor: 'stale' }] }); + await owner.transactReplica({ putCursors: [{ userId: 1, scopeId: '10', cursor: 'stale' }] }); }); await expect(repository[OFFLINE_REPOSITORY_ATOMIC_MUTATION]!(operation)).rejects.toThrow('changed through another SQLite connection'); @@ -339,8 +339,8 @@ describe('SqliteOfflineRepository community sqlite driver', () => { }); const repository = createRepository(); - await repository[OFFLINE_REPOSITORY_ATOMIC_MUTATION]!(async () => { - await repository.transactReplica({ putCursors: [{ userId: 1, scopeId: '10', cursor: 'fresh' }] }); + await repository[OFFLINE_REPOSITORY_ATOMIC_MUTATION]!(async (owner) => { + await owner.transactReplica({ putCursors: [{ userId: 1, scopeId: '10', cursor: 'fresh' }] }); }); const lockCall = plugin.execute.mock.calls.find(([options]) => @@ -380,8 +380,8 @@ describe('SqliteOfflineRepository community sqlite driver', () => { const repository = createRepository(); await expect( - repository[OFFLINE_REPOSITORY_ATOMIC_MUTATION]!(async () => { - await repository.transactReplica({ putCursors: [{ userId: 1, scopeId: '10', cursor: 'committed' }] }); + repository[OFFLINE_REPOSITORY_ATOMIC_MUTATION]!(async (owner) => { + await owner.transactReplica({ putCursors: [{ userId: 1, scopeId: '10', cursor: 'committed' }] }); dataVersion = 2; }), ).resolves.toBeUndefined(); @@ -407,10 +407,10 @@ describe('SqliteOfflineRepository community sqlite driver', () => { const repository = createRepository(); await expect( - repository[OFFLINE_REPOSITORY_ATOMIC_MUTATION]!(async () => { - await repository.transactReplica({ putCursors: [{ userId: 1, scopeId: '10', cursor: 'first' }] }); + repository[OFFLINE_REPOSITORY_ATOMIC_MUTATION]!(async (owner) => { + await owner.transactReplica({ putCursors: [{ userId: 1, scopeId: '10', cursor: 'first' }] }); dataVersion = 2; - await repository.transactReplica({ putCursors: [{ userId: 1, scopeId: '10', cursor: 'stale-second' }] }); + await owner.transactReplica({ putCursors: [{ userId: 1, scopeId: '10', cursor: 'stale-second' }] }); }), ).rejects.toThrow('changed through another SQLite connection'); @@ -418,6 +418,85 @@ describe('SqliteOfflineRepository community sqlite driver', () => { expect(plugin.rollbackTransaction).toHaveBeenCalledOnce(); }); + it('atomic owner以外のwriteをatomic operation完了後まで待機させる', 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.initialize(); + plugin.execute.mockClear(); + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + let ownerCommitted = false; + let externalCommitted = false; + let external: Promise = Promise.resolve(); + + const atomic = repository[OFFLINE_REPOSITORY_ATOMIC_MUTATION]!(async (owner) => { + external = repository + .putCommand({ + userId: 1, + scopeId: '10', + commandId: 'external', + aggregateType: 'test_items', + sourceKey: 'test_items', + identity: { kind: 'generated', localId: '019d-external' }, + operation: 'test_items.update', + payload: {}, + baseRevision: null, + state: 'pending', + attempts: 0, + retryAt: null, + createdAt: 2, + lastErrorCode: null, + }) + .then(() => { + externalCommitted = true; + }); + await owner.transactReplica({ putCursors: [{ userId: 1, scopeId: '10', cursor: 'owner' }] }); + ownerCommitted = true; + await gate; + }); + + await vi.waitFor(() => expect(ownerCommitted).toBe(true)); + expect(externalCommitted).toBe(false); + release(); + await atomic; + await external; + expect(externalCommitted).toBe(true); + }); + + it('snapshot callbackからrepository本体を再入しても同じsnapshotで完了する', 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.initialize(); + + await expect( + repository[OFFLINE_REPOSITORY_ATOMIC_MUTATION]!(async (owner) => + owner.runReadSnapshot(() => repository.getCommands({ userId: 1, scopeId: '10' })), + ), + ).resolves.toEqual([]); + }); + it('pull attentionをput/getしtransactionでupsertする', async () => { const repository = createRepository(); await repository.initialize(); @@ -2072,6 +2151,13 @@ describe('SqliteOfflineRepository replica rows', () => { } describe('runReadSnapshot', () => { + it('callbackからrepository本体を再入しても同じsnapshotで完了する', async () => { + const repository = createRepository(); + await repository.initialize(); + + await expect(repository.runReadSnapshot(() => repository.getCommands({ userId: 1, scopeId: '10' }))).resolves.toEqual([]); + }); + it('open snapshot中はwriteが待機し、readerはcommit前の状態だけを見る', async () => { const repository = createRepository(); await repository.initialize(); diff --git a/projects/kit/offline/src/lib/sqlite-offline-repository.ts b/projects/kit/offline/src/lib/sqlite-offline-repository.ts index 5eccf26..85c04d7 100644 --- a/projects/kit/offline/src/lib/sqlite-offline-repository.ts +++ b/projects/kit/offline/src/lib/sqlite-offline-repository.ts @@ -215,6 +215,9 @@ export class SqliteOfflineRepository implements OfflineRepository { #atomicMutationRevision: number | null = null; #atomicMutationCommitted = false; #atomicOperations: Promise = Promise.resolve(); + #readSnapshotActive = false; + #atomicIdle: Promise = Promise.resolve(); + #resolveAtomicIdle: (() => void) | null = null; initialize(): Promise { this.#initialization ??= this.#open(); @@ -306,12 +309,19 @@ export class SqliteOfflineRepository implements OfflineRepository { async runReadSnapshot(read: (reader: OfflineRepositoryReader) => Promise): Promise { if (this.#atomicMutationRevision !== null) { - return this.#queueAtomicOperation(async () => this.#nativeTransaction(await this.#databaseConnection(), () => read(this.#reader()))); + throw new Error('Use the repository passed to an atomic mutation for snapshot reads.'); } - return this.#transaction(() => read(this.#reader())); + return this.#transaction(async () => { + this.#readSnapshotActive = true; + try { + return await read(this.#reader()); + } finally { + this.#readSnapshotActive = false; + } + }); } - async [OFFLINE_REPOSITORY_ATOMIC_MUTATION](operation: () => Promise): Promise { + async [OFFLINE_REPOSITORY_ATOMIC_MUTATION](operation: (repository: OfflineRepository) => Promise): Promise { await this.initialize(); await this.#writes; if (this.#atomicMutationRevision !== null) { @@ -321,9 +331,12 @@ export class SqliteOfflineRepository implements OfflineRepository { 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 result = await operation(); + const result = await operation(this.#atomicRepository()); await this.#atomicOperations; if (!this.#atomicMutationCommitted) { await this.#queueAtomicOperation(() => this.#atomicTransaction(databaseId, async () => undefined, false)); @@ -332,6 +345,9 @@ export class SqliteOfflineRepository implements OfflineRepository { } finally { this.#atomicMutationRevision = null; this.#endReaders(); + this.#resolveAtomicIdle?.(); + this.#resolveAtomicIdle = null; + this.#atomicIdle = Promise.resolve(); } } @@ -368,42 +384,46 @@ export class SqliteOfflineRepository implements OfflineRepository { } async clearUser(userId: OfflinePrincipalId): Promise { - await this.#transaction(async (database) => { - const principal = canonicalOfflinePrincipalId(userId); - await this.#execute(database, 'DELETE FROM offline_session_manifests WHERE user_id = ?', [principal]); - await this.#execute(database, 'DELETE FROM offline_sync_commands WHERE user_id = ?', [principal]); - await this.#execute(database, 'DELETE FROM offline_replica_cursors WHERE user_id = ?', [principal]); - await this.#execute(database, 'DELETE FROM offline_reconciliation_scopes WHERE user_id = ?', [principal]); - await this.#execute(database, 'DELETE FROM offline_pull_attentions WHERE user_id = ?', [principal]); - for (const entity of this.#options.replicaSchema.entities) { - await this.#execute(database, `DELETE FROM ${entity.tableName} WHERE _offline_user_id = ?`, [principal]); - } - await this.#execute(database, 'UPDATE offline_metadata SET last_user_id = NULL WHERE id = 1 AND last_user_id = ?', [principal]); - }); + await this.#transaction((database) => this.#clearUser(database, userId)); + } + + async #clearUser(database: string, userId: OfflinePrincipalId): Promise { + const principal = canonicalOfflinePrincipalId(userId); + await this.#execute(database, 'DELETE FROM offline_session_manifests WHERE user_id = ?', [principal]); + await this.#execute(database, 'DELETE FROM offline_sync_commands WHERE user_id = ?', [principal]); + await this.#execute(database, 'DELETE FROM offline_replica_cursors WHERE user_id = ?', [principal]); + await this.#execute(database, 'DELETE FROM offline_reconciliation_scopes WHERE user_id = ?', [principal]); + await this.#execute(database, 'DELETE FROM offline_pull_attentions WHERE user_id = ?', [principal]); + for (const entity of this.#options.replicaSchema.entities) { + await this.#execute(database, `DELETE FROM ${entity.tableName} WHERE _offline_user_id = ?`, [principal]); + } + await this.#execute(database, 'UPDATE offline_metadata SET last_user_id = NULL WHERE id = 1 AND last_user_id = ?', [principal]); } async clearScope(scope: OfflineScope): Promise { - await this.#transaction(async (database) => { - const values = [canonicalOfflinePrincipalId(scope.userId), scope.scopeId]; - const partitionSourceKeys = this.#options.replicaSchema.entities - .filter((entity) => entity.scope === 'partition') - .map((entity) => entity.sourceKey); - if (partitionSourceKeys.length > 0) { - await this.#execute( - database, - `DELETE FROM offline_sync_commands + await this.#transaction((database) => this.#clearScope(database, scope)); + } + + async #clearScope(database: string, scope: OfflineScope): Promise { + const values = [canonicalOfflinePrincipalId(scope.userId), scope.scopeId]; + const partitionSourceKeys = this.#options.replicaSchema.entities + .filter((entity) => entity.scope === 'partition') + .map((entity) => entity.sourceKey); + if (partitionSourceKeys.length > 0) { + await this.#execute( + database, + `DELETE FROM offline_sync_commands WHERE user_id = ? AND scope_id = ? AND source_key IN (${partitionSourceKeys.map(() => '?').join(', ')})`, - [...values, ...partitionSourceKeys], - ); - } - await this.#execute(database, 'DELETE FROM offline_replica_cursors WHERE user_id = ? AND scope_id = ?', values); - await this.#execute(database, 'DELETE FROM offline_reconciliation_scopes WHERE user_id = ? AND scope_id = ?', values); - await this.#execute(database, 'DELETE FROM offline_pull_attentions WHERE user_id = ? AND scope_id = ?', values); - for (const entity of this.#options.replicaSchema.entities) { - if (entity.scope !== 'partition') continue; - await this.#execute(database, `DELETE FROM ${entity.tableName} WHERE _offline_user_id = ? AND _offline_scope_id = ?`, values); - } - }); + [...values, ...partitionSourceKeys], + ); + } + await this.#execute(database, 'DELETE FROM offline_replica_cursors WHERE user_id = ? AND scope_id = ?', values); + await this.#execute(database, 'DELETE FROM offline_reconciliation_scopes WHERE user_id = ? AND scope_id = ?', values); + await this.#execute(database, 'DELETE FROM offline_pull_attentions WHERE user_id = ? AND scope_id = ?', values); + for (const entity of this.#options.replicaSchema.entities) { + if (entity.scope !== 'partition') continue; + await this.#execute(database, `DELETE FROM ${entity.tableName} WHERE _offline_user_id = ? AND _offline_scope_id = ?`, values); + } } async transactReplica(transaction: OfflineReplicaTransaction): Promise { @@ -751,8 +771,64 @@ export class SqliteOfflineRepository implements OfflineRepository { }; } + #atomicRepository(): OfflineRepository { + const reader = this.#reader(); + const atomicTransaction = (run: (databaseId: string) => Promise, marksCommit = true): Promise => + this.#queueAtomicOperation(() => this.#atomicTransaction(this.#databaseId!, run, marksCommit)); + return { + initialize: () => Promise.resolve(), + ...reader, + runReadSnapshot: (read) => + this.#queueAtomicOperation(() => + this.#nativeTransaction(this.#databaseId!, async () => { + this.#readSnapshotActive = true; + try { + return await read(reader); + } finally { + this.#readSnapshotActive = false; + } + }), + ), + putCommand: (command) => atomicTransaction((databaseId) => this.#putCommand(databaseId, command)), + replaceCommand: (command) => atomicTransaction((databaseId) => this.#putCommand(databaseId, command)), + removeCommand: (commandId) => + atomicTransaction((databaseId) => this.#execute(databaseId, 'DELETE FROM offline_sync_commands WHERE command_id = ?', [commandId])), + putPullAttention: (attention) => + atomicTransaction((databaseId) => this.#applyReplicaTransaction(databaseId, { putPullAttentions: [attention] })), + removePullAttention: (scope) => + atomicTransaction((databaseId) => this.#applyReplicaTransaction(databaseId, { removePullAttentions: [scope] })), + setLastUserId: (userId) => + atomicTransaction((databaseId) => + this.#execute( + databaseId, + `INSERT INTO offline_metadata (id, schema_version, last_user_id) VALUES (1, ?, ?) + ON CONFLICT(id) DO UPDATE SET schema_version = excluded.schema_version, last_user_id = excluded.last_user_id`, + [OFFLINE_SCHEMA_VERSION, canonicalOfflinePrincipalId(userId)], + ), + ), + putSessionManifest: (userId, value) => + atomicTransaction((databaseId) => + this.#execute( + databaseId, + `INSERT INTO offline_session_manifests (user_id, value_json) VALUES (?, ?) + ON CONFLICT(user_id) DO UPDATE SET value_json = excluded.value_json`, + [canonicalOfflinePrincipalId(userId), JSON.stringify(value)], + ), + ), + clearUser: (userId) => atomicTransaction((databaseId) => this.#clearUser(databaseId, userId)), + clearScope: (scope) => atomicTransaction((databaseId) => this.#clearScope(databaseId, scope)), + transactReplica: (transaction) => { + for (const row of transaction.putRows ?? []) this.#validateReplicaRow(row); + return atomicTransaction((databaseId) => this.#applyReplicaTransaction(databaseId, transaction)); + }, + }; + } + async #withCommittedRead(operation: () => Promise): Promise { await this.initialize(); + if (this.#readSnapshotActive) { + return operation(); + } if (this.#atomicMutationRevision !== null) { return this.#queueAtomicOperation(operation); } @@ -793,7 +869,7 @@ 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)); + return this.#atomicIdle.then(() => this.#queueWrite(run)); } const write = this.#writes.then(async (): Promise => { if (this.#activeReaders > 0) await this.#readersIdle; @@ -805,7 +881,7 @@ export class SqliteOfflineRepository implements OfflineRepository { #transaction(run: (databaseId: string) => Promise): Promise { if (this.#atomicMutationRevision !== null) { - return this.#queueAtomicOperation(async () => this.#atomicTransaction(await this.#databaseConnection(), run)); + return this.#atomicIdle.then(() => this.#transaction(run)); } const transaction = this.#writes.then(async (): Promise => { if (this.#activeReaders > 0) await this.#readersIdle;