From 56e091ca244ebefe0d6b45dd3bc146e7d0b31ea6 Mon Sep 17 00:00:00 2001 From: rdlabo Date: Wed, 12 Aug 2026 13:38:19 +0900 Subject: [PATCH] fix(offline): keep prepared batches linear --- .../src/lib/offline-sync.service.spec.ts | 23 ++++++++ .../offline/src/lib/offline-sync.service.ts | 55 ++++++++++++++----- 2 files changed, 63 insertions(+), 15 deletions(-) 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 3857980..9f6fa06 100644 --- a/projects/kit/offline/src/lib/offline-sync.service.spec.ts +++ b/projects/kit/offline/src/lib/offline-sync.service.spec.ts @@ -553,6 +553,8 @@ describe('OfflineSyncService', () => { it('2件の成功は1回のtransactReplicaでFIFO createdAtを永続化する', async () => { const repository = TestBed.inject(OFFLINE_REPOSITORY); const transactReplica = vi.mocked(repository.transactReplica); + const getCommandsForUser = vi.mocked(repository.getCommandsForUser!); + getCommandsForUser.mockClear(); const commandIds = await service.enqueuePreparedBatch(async () => [prepared('batch-a', 'A'), prepared('batch-b', 'B')], { flush: false, @@ -560,6 +562,7 @@ describe('OfflineSyncService', () => { expect(commandIds).toHaveLength(2); expect(transactReplica).toHaveBeenCalledTimes(1); + expect(getCommandsForUser).toHaveBeenCalledTimes(1); expect(commands).toHaveLength(2); expect(commands.map((command) => (command.identity.kind === 'generated' ? command.identity.localId : ''))).toEqual([ 'batch-a', @@ -581,6 +584,26 @@ describe('OfflineSyncService', () => { ); }); + it('product lease失効時は全prepare後もcommit直前にbatch全体を拒否する', async () => { + const repository = TestBed.inject(OFFLINE_REPOSITORY); + const transactReplica = vi.mocked(repository.transactReplica); + const assertCurrent = vi.fn(() => { + throw new Error('product principal changed'); + }); + + await expect( + service.enqueuePreparedBatch(async () => [prepared('lease-a', 'A'), prepared('lease-b', 'B')], { + flush: false, + assertCurrent, + }), + ).rejects.toThrow('product principal changed'); + + expect(assertCurrent).toHaveBeenCalledOnce(); + expect(transactReplica).not.toHaveBeenCalled(); + expect(commands).toEqual([]); + expect(rows).toEqual([]); + }); + it('同一principalの複数scopeを1回のtransactionで受け付ける', async () => { localSession = { userId: 1, diff --git a/projects/kit/offline/src/lib/offline-sync.service.ts b/projects/kit/offline/src/lib/offline-sync.service.ts index 598636a..851f902 100644 --- a/projects/kit/offline/src/lib/offline-sync.service.ts +++ b/projects/kit/offline/src/lib/offline-sync.service.ts @@ -73,6 +73,12 @@ export interface PreparedOfflineCommand { replicaTransaction?: Pick; } +export interface PreparedOfflineBatchOptions { + flush?: boolean; + /** Product identity/scope lease asserted after all async preparation and immediately before the durable commit. */ + assertCurrent?: () => void; +} + /** Validated optimistic projection ready for a single Outbox commit. */ interface MaterializedOfflineEnqueue { command: OfflineCommand; @@ -262,7 +268,7 @@ export class OfflineSyncService { */ enqueuePreparedBatch( prepare: (repository: OfflineRepository) => Promise[]>, - options: { flush?: boolean } = {}, + options: PreparedOfflineBatchOptions = {}, ): Promise { const generation = this.#generation; return this.#serializeReplicaMutation(async () => { @@ -351,23 +357,38 @@ export class OfflineSyncService { async #enqueuePreparedBatch( prepared: readonly PreparedOfflineCommand[], - options: { flush?: boolean }, + options: PreparedOfflineBatchOptions, generation: number, ): Promise { if (prepared.length === 0) { throw new Error('Prepared offline batch must contain at least one command.'); } const session = await this.#beginEnqueueSession(generation); + const currentCommands = await this.#commandsForUser(session.userId); + this.#rememberCreatedAt(currentCommands); + const firstCreatedAt = Math.max(Date.now(), this.#lastCommandCreatedAt + 1); + this.#lastCommandCreatedAt = firstCreatedAt + prepared.length - 1; const materializations: MaterializedOfflineEnqueue[] = []; - for (const entry of prepared) { + for (const [index, entry] of prepared.entries()) { this.#assertEnqueueScope(session, entry.request.scopeId); - materializations.push(await this.#materializeEnqueue(session.userId, entry.request, entry.replicaTransaction)); + materializations.push( + await this.#materializeEnqueue( + session.userId, + entry.request, + entry.replicaTransaction, + undefined, + firstCreatedAt + index, + ), + ); } this.#assertDistinctBatchFootprints(materializations); await this.#assertOutboxCapacity( session.userId, materializations.map((item) => item.command), + undefined, + currentCommands, ); + options.assertCurrent?.(); await this.#commitMaterializedEnqueues(materializations, generation, options); return materializations.map((item) => item.command.commandId); } @@ -399,6 +420,7 @@ export class OfflineSyncService { request: EnqueueOfflineCommand, replicaTransaction?: Pick, replaced?: OfflineCommand, + createdAt?: number, ): Promise { const scope = { userId, scopeId: request.scopeId }; this.noteScope(scope); @@ -422,7 +444,7 @@ export class OfflineSyncService { state: 'pending', attempts: 0, retryAt: null, - createdAt: replaced?.createdAt ?? (await this.#nextCommandCreatedAt(userId)), + createdAt: replaced?.createdAt ?? createdAt ?? (await this.#nextCommandCreatedAt(userId)), lastErrorCode: null, }; if ( @@ -663,14 +685,9 @@ export class OfflineSyncService { userId: OfflinePrincipalId, newCommands: readonly OfflineCommand[], excludingCommandId?: string, + knownCommands?: readonly OfflineCommand[], ): Promise { - const currentCommands = this.#repository.getCommandsForUser - ? await this.#repository.getCommandsForUser(userId) - : ( - await Promise.all( - [...this.#knownScopes.values()].filter((scope) => scope.userId === userId).map((scope) => this.#repository.getCommands(scope)), - ) - ).flat(); + const currentCommands = knownCommands ?? (await this.#commandsForUser(userId)); const commands = excludingCommandId ? currentCommands.filter((candidate) => candidate.commandId !== excludingCommandId) : currentCommands; @@ -686,6 +703,16 @@ export class OfflineSyncService { } } + async #commandsForUser(userId: OfflinePrincipalId): Promise { + return this.#repository.getCommandsForUser + ? this.#repository.getCommandsForUser(userId) + : ( + await Promise.all( + [...this.#knownScopes.values()].filter((scope) => scope.userId === userId).map((scope) => this.#repository.getCommands(scope)), + ) + ).flat(); + } + #serializedOutboxBytes(commands: readonly OfflineCommand[]): number { return new TextEncoder().encode(JSON.stringify(commands)).byteLength; } @@ -1420,9 +1447,7 @@ export class OfflineSyncService { } async #nextCommandCreatedAt(userId: OfflinePrincipalId): Promise { - const commands = this.#repository.getCommandsForUser - ? await this.#repository.getCommandsForUser(userId) - : await this.#readKnownCommands(); + const commands = await this.#commandsForUser(userId); this.#rememberCreatedAt(commands); const createdAt = Math.max(Date.now(), this.#lastCommandCreatedAt + 1); this.#lastCommandCreatedAt = createdAt;