From 52dfbd5bf2b956a8b4d2ecd539ea6b19a205c6f9 Mon Sep 17 00:00:00 2001 From: rdlabo Date: Wed, 12 Aug 2026 21:10:46 +0900 Subject: [PATCH] feat(offline): replace aggregate intent chains atomically --- .../src/lib/offline-sync.service.spec.ts | 94 +++++++++++++++++++ .../offline/src/lib/offline-sync.service.ts | 79 ++++++++++++++-- 2 files changed, 166 insertions(+), 7 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 98d9f93..ac65c04 100644 --- a/projects/kit/offline/src/lib/offline-sync.service.spec.ts +++ b/projects/kit/offline/src/lib/offline-sync.service.spec.ts @@ -1414,6 +1414,100 @@ describe('OfflineSyncService', () => { }); }); + it('同じaggregateの競合commandと後続intentを一transactionで再materializeする', async () => { + const firstId = await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'replace-chain' }, + operation: 'documents.update', + payload: { title: 'stocktake' }, + optimisticValue: { id: 15, title: 'stocktake' }, + baseRevision: 1, + }, + { flush: false }, + ); + await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'replace-chain' }, + operation: 'documents.update', + payload: { title: 'later delta' }, + optimisticValue: { id: 15, title: 'later delta' }, + baseRevision: 1, + }, + { flush: false }, + ); + commands[0] = { ...commands[0]!, state: 'conflict' }; + const originalIds = commands.map((command) => command.commandId); + const originalCreatedAt = commands.map((command) => command.createdAt); + + const replacementIds = await service.replacePreparedAggregate( + firstId, + async (_repository, chain) => + chain.map((command, index) => ({ + request: { + scopeId: command.scopeId, + aggregateType: command.aggregateType, + identity: command.identity, + operation: command.operation, + payload: command.payload, + optimisticValue: { id: 15, title: index === 0 ? 'new stocktake' : 'new stocktake plus delta' }, + baseRevision: 2, + }, + })), + { flush: false }, + ); + + expect(replacementIds).toHaveLength(2); + expect(replacementIds).not.toEqual(originalIds); + expect(commands.map((command) => command.commandId)).toEqual(replacementIds); + expect(commands.map((command) => command.state)).toEqual(['pending', 'pending']); + expect(commands.map((command) => command.createdAt)).toEqual(originalCreatedAt); + expect(rows.find((row) => row.identity.kind === 'generated' && row.identity.localId === 'replace-chain')?.values).toEqual({ + id: 15, + title: 'new stocktake plus delta', + }); + }); + + it('aggregate chainの準備失敗では元commandとprojectionを一切変更しない', async () => { + const firstId = await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'replace-chain-failure' }, + operation: 'documents.update', + payload: { title: 'first' }, + optimisticValue: { id: 16, title: 'first' }, + }, + { flush: false }, + ); + await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'replace-chain-failure' }, + operation: 'documents.update', + payload: { title: 'second' }, + optimisticValue: { id: 16, title: 'second' }, + }, + { flush: false }, + ); + commands[0] = { ...commands[0]!, state: 'conflict' }; + const beforeCommands = structuredClone(commands); + const beforeRows = structuredClone(rows); + + await expect( + service.replacePreparedAggregate(firstId, async () => { + throw new Error('chain preparation failed'); + }), + ).rejects.toThrow('chain preparation failed'); + + expect(commands).toEqual(beforeCommands); + expect(rows).toEqual(beforeRows); + }); + it('companionの対象集合を変えるreplacementを元commandと楽観値を残して拒否する', async () => { const companion: OfflineReplicaRow = { 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 2d69169..8e6d6a9 100644 --- a/projects/kit/offline/src/lib/offline-sync.service.ts +++ b/projects/kit/offline/src/lib/offline-sync.service.ts @@ -308,6 +308,67 @@ export class OfflineSyncService { }); } + /** + * Atomically rematerializes every unresolved intent for one aggregate. + * + * This is the conflict-recovery boundary for ordered intent chains: the old + * chain remains durable until every replacement has been prepared and the + * complete chain can be committed in one replica transaction. + */ + replacePreparedAggregate( + commandId: string, + prepare: ( + repository: OfflineRepository, + commands: readonly OfflineCommand[], + ) => Promise[]>, + options: PreparedOfflineBatchOptions = {}, + ): Promise { + const generation = this.#generation; + return this.#serializeReplicaMutation(async () => { + await this.initialize(); + if (!this.#isCurrent(generation)) throw new Error('Offline session changed before prepared aggregate replacement.'); + const knownCommands = await this.#readKnownCommands(); + const selected = knownCommands.find((command) => command.commandId === commandId); + if (!selected) throw new Error(`Offline command ${commandId} no longer exists.`); + const aggregateKey = this.#aggregateKey(selected); + const replaced = knownCommands.filter((command) => this.#aggregateKey(command) === aggregateKey); + this.#assertDiscardable(replaced); + const prepared = await prepare(this.#repository, replaced); + if (prepared.length !== replaced.length) { + throw new Error('Offline aggregate replacement must preserve the ordered intent count.'); + } + const session = await this.#beginEnqueueSession(generation); + const materializations: MaterializedOfflineEnqueue[] = []; + for (const [index, entry] of prepared.entries()) { + this.#assertEnqueueScope(session, entry.request.scopeId); + materializations.push( + await this.#materializeEnqueue( + session.userId, + entry.request, + entry.replicaTransaction, + replaced[index], + ), + ); + } + const retained = knownCommands.filter((command) => !replaced.some((item) => item.commandId === command.commandId)); + this.#assertDistinctBatchFootprints(materializations, retained, true); + await this.#assertOutboxCapacity( + session.userId, + materializations.map((item) => item.command), + replaced.map((command) => command.commandId), + knownCommands, + ); + options.assertCurrent?.(); + await this.#commitMaterializedEnqueues( + materializations, + generation, + options, + replaced.map((command) => command.commandId), + ); + return materializations.map((item) => item.command.commandId); + }); + } + /** * Serializes a product-owned replica projection with enqueue and command ACK * reconciliation. For read/derive/write cache updates, prefer @@ -352,7 +413,7 @@ export class OfflineSyncService { const currentCommands = await this.#commandsForUser(session.userId); const retainedCommands = replaced ? currentCommands.filter((command) => command.commandId !== replaced.commandId) : currentCommands; this.#assertDistinctBatchFootprints([materialization], retainedCommands); - await this.#assertOutboxCapacity(session.userId, [materialization.command], replaced?.commandId, currentCommands); + await this.#assertOutboxCapacity(session.userId, [materialization.command], replaced ? [replaced.commandId] : undefined, currentCommands); await this.#commitMaterializedEnqueues([materialization], generation, options, replaced ? [replaced.commandId] : undefined); return materialization.command.commandId; } @@ -538,7 +599,11 @@ export class OfflineSyncService { return { command, optimisticRow, optimisticCompanions }; } - #assertDistinctBatchFootprints(entries: readonly MaterializedOfflineEnqueue[], existingCommands: readonly OfflineCommand[]): void { + #assertDistinctBatchFootprints( + entries: readonly MaterializedOfflineEnqueue[], + existingCommands: readonly OfflineCommand[], + allowOneAggregate = false, + ): void { const aggregates = new Set(); const replicaKeys = new Set(); const existingFootprints = new Map(); @@ -548,7 +613,7 @@ export class OfflineSyncService { } for (const entry of entries) { const aggregate = this.#aggregateKey(entry.command); - if (aggregates.has(aggregate)) { + if (aggregates.has(aggregate) && !allowOneAggregate) { throw new Error('Prepared offline batch contains overlapping aggregate intents.'); } aggregates.add(aggregate); @@ -556,7 +621,7 @@ export class OfflineSyncService { this.#replicaRowKey(entry.optimisticRow), ...entry.optimisticCompanions.map((companion) => this.#replicaRowKey(companion.key)), ]) { - if (replicaKeys.has(key)) { + if (replicaKeys.has(key) && !allowOneAggregate) { throw new Error('Prepared offline batch contains overlapping replica footprints.'); } const existingAggregate = existingFootprints.get(key); @@ -700,12 +765,12 @@ export class OfflineSyncService { async #assertOutboxCapacity( userId: OfflinePrincipalId, newCommands: readonly OfflineCommand[], - excludingCommandId?: string, + excludingCommandIds?: readonly string[], knownCommands?: readonly OfflineCommand[], ): Promise { const currentCommands = knownCommands ?? (await this.#commandsForUser(userId)); - const commands = excludingCommandId - ? currentCommands.filter((candidate) => candidate.commandId !== excludingCommandId) + const commands = excludingCommandIds + ? currentCommands.filter((candidate) => !excludingCommandIds.includes(candidate.commandId)) : currentCommands; const maxCommands = this.#options.outboxLimits?.maxCommandsPerUser ?? DEFAULT_MAX_OUTBOX_COMMANDS_PER_USER; const maxBytes = this.#options.outboxLimits?.maxBytesPerUser ?? DEFAULT_MAX_OUTBOX_BYTES_PER_USER;