Skip to content

Commit 154ecef

Browse files
committed
feat(offline): replace aggregate intent chains atomically
1 parent f8e53af commit 154ecef

2 files changed

Lines changed: 164 additions & 7 deletions

File tree

‎projects/kit/offline/src/lib/offline-sync.service.spec.ts‎

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1414,6 +1414,99 @@ describe('OfflineSyncService', () => {
14141414
});
14151415
});
14161416

1417+
it('同じaggregateの競合commandと後続intentを一transactionで再materializeする', async () => {
1418+
const firstId = await service.enqueue(
1419+
{
1420+
scopeId: '10',
1421+
aggregateType: 'documents',
1422+
identity: { kind: 'generated', localId: 'replace-chain' },
1423+
operation: 'documents.update',
1424+
payload: { title: 'stocktake' },
1425+
optimisticValue: { id: 15, title: 'stocktake' },
1426+
baseRevision: 1,
1427+
},
1428+
{ flush: false },
1429+
);
1430+
await service.enqueue(
1431+
{
1432+
scopeId: '10',
1433+
aggregateType: 'documents',
1434+
identity: { kind: 'generated', localId: 'replace-chain' },
1435+
operation: 'documents.update',
1436+
payload: { title: 'later delta' },
1437+
optimisticValue: { id: 15, title: 'later delta' },
1438+
baseRevision: 1,
1439+
},
1440+
{ flush: false },
1441+
);
1442+
commands[0] = { ...commands[0]!, state: 'conflict' };
1443+
const originalIds = commands.map((command) => command.commandId);
1444+
1445+
const replacementIds = await service.replacePreparedAggregate(
1446+
firstId,
1447+
async (_repository, chain) =>
1448+
chain.map((command, index) => ({
1449+
request: {
1450+
scopeId: command.scopeId,
1451+
aggregateType: command.aggregateType,
1452+
identity: command.identity,
1453+
operation: command.operation,
1454+
payload: command.payload,
1455+
optimisticValue: { id: 15, title: index === 0 ? 'new stocktake' : 'new stocktake plus delta' },
1456+
baseRevision: 2,
1457+
},
1458+
})),
1459+
{ flush: false },
1460+
);
1461+
1462+
expect(replacementIds).toHaveLength(2);
1463+
expect(replacementIds).not.toEqual(originalIds);
1464+
expect(commands.map((command) => command.commandId)).toEqual(replacementIds);
1465+
expect(commands.map((command) => command.state)).toEqual(['pending', 'pending']);
1466+
expect(commands.map((command) => command.createdAt)).toEqual([commands[0]!.createdAt, commands[1]!.createdAt]);
1467+
expect(rows.find((row) => row.identity.kind === 'generated' && row.identity.localId === 'replace-chain')?.values).toEqual({
1468+
id: 15,
1469+
title: 'new stocktake plus delta',
1470+
});
1471+
});
1472+
1473+
it('aggregate chainの準備失敗では元commandとprojectionを一切変更しない', async () => {
1474+
const firstId = await service.enqueue(
1475+
{
1476+
scopeId: '10',
1477+
aggregateType: 'documents',
1478+
identity: { kind: 'generated', localId: 'replace-chain-failure' },
1479+
operation: 'documents.update',
1480+
payload: { title: 'first' },
1481+
optimisticValue: { id: 16, title: 'first' },
1482+
},
1483+
{ flush: false },
1484+
);
1485+
await service.enqueue(
1486+
{
1487+
scopeId: '10',
1488+
aggregateType: 'documents',
1489+
identity: { kind: 'generated', localId: 'replace-chain-failure' },
1490+
operation: 'documents.update',
1491+
payload: { title: 'second' },
1492+
optimisticValue: { id: 16, title: 'second' },
1493+
},
1494+
{ flush: false },
1495+
);
1496+
commands[0] = { ...commands[0]!, state: 'conflict' };
1497+
const beforeCommands = structuredClone(commands);
1498+
const beforeRows = structuredClone(rows);
1499+
1500+
await expect(
1501+
service.replacePreparedAggregate(firstId, async () => {
1502+
throw new Error('chain preparation failed');
1503+
}),
1504+
).rejects.toThrow('chain preparation failed');
1505+
1506+
expect(commands).toEqual(beforeCommands);
1507+
expect(rows).toEqual(beforeRows);
1508+
});
1509+
14171510
it('companionの対象集合を変えるreplacementを元commandと楽観値を残して拒否する', async () => {
14181511
const companion: OfflineReplicaRow = {
14191512
userId: 1,

‎projects/kit/offline/src/lib/offline-sync.service.ts‎

Lines changed: 71 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -308,6 +308,66 @@ export class OfflineSyncService {
308308
});
309309
}
310310

311+
/**
312+
* Atomically rematerializes every unresolved intent for one aggregate.
313+
*
314+
* This is the conflict-recovery boundary for ordered intent chains: the old
315+
* chain remains durable until every replacement has been prepared and the
316+
* complete chain can be committed in one replica transaction.
317+
*/
318+
replacePreparedAggregate<T>(
319+
commandId: string,
320+
prepare: (
321+
repository: OfflineRepository,
322+
commands: readonly OfflineCommand[],
323+
) => Promise<readonly PreparedOfflineCommand<T>[]>,
324+
options: { flush?: boolean } = {},
325+
): Promise<readonly string[]> {
326+
const generation = this.#generation;
327+
return this.#serializeReplicaMutation(async () => {
328+
await this.initialize();
329+
if (!this.#isCurrent(generation)) throw new Error('Offline session changed before prepared aggregate replacement.');
330+
const knownCommands = await this.#readKnownCommands();
331+
const selected = knownCommands.find((command) => command.commandId === commandId);
332+
if (!selected) throw new Error(`Offline command ${commandId} no longer exists.`);
333+
const aggregateKey = this.#aggregateKey(selected);
334+
const replaced = knownCommands.filter((command) => this.#aggregateKey(command) === aggregateKey);
335+
this.#assertDiscardable(replaced);
336+
const prepared = await prepare(this.#repository, replaced);
337+
if (prepared.length !== replaced.length) {
338+
throw new Error('Offline aggregate replacement must preserve the ordered intent count.');
339+
}
340+
const session = await this.#beginEnqueueSession(generation);
341+
const materializations: MaterializedOfflineEnqueue[] = [];
342+
for (const [index, entry] of prepared.entries()) {
343+
this.#assertEnqueueScope(session, entry.request.scopeId);
344+
materializations.push(
345+
await this.#materializeEnqueue(
346+
session.userId,
347+
entry.request,
348+
entry.replicaTransaction,
349+
replaced[index],
350+
),
351+
);
352+
}
353+
const retained = knownCommands.filter((command) => !replaced.some((item) => item.commandId === command.commandId));
354+
this.#assertDistinctBatchFootprints(materializations, retained, true);
355+
await this.#assertOutboxCapacity(
356+
session.userId,
357+
materializations.map((item) => item.command),
358+
replaced.map((command) => command.commandId),
359+
knownCommands,
360+
);
361+
await this.#commitMaterializedEnqueues(
362+
materializations,
363+
generation,
364+
options,
365+
replaced.map((command) => command.commandId),
366+
);
367+
return materializations.map((item) => item.command.commandId);
368+
});
369+
}
370+
311371
/**
312372
* Serializes a product-owned replica projection with enqueue and command ACK
313373
* reconciliation. For read/derive/write cache updates, prefer
@@ -352,7 +412,7 @@ export class OfflineSyncService {
352412
const currentCommands = await this.#commandsForUser(session.userId);
353413
const retainedCommands = replaced ? currentCommands.filter((command) => command.commandId !== replaced.commandId) : currentCommands;
354414
this.#assertDistinctBatchFootprints([materialization], retainedCommands);
355-
await this.#assertOutboxCapacity(session.userId, [materialization.command], replaced?.commandId, currentCommands);
415+
await this.#assertOutboxCapacity(session.userId, [materialization.command], replaced ? [replaced.commandId] : undefined, currentCommands);
356416
await this.#commitMaterializedEnqueues([materialization], generation, options, replaced ? [replaced.commandId] : undefined);
357417
return materialization.command.commandId;
358418
}
@@ -538,7 +598,11 @@ export class OfflineSyncService {
538598
return { command, optimisticRow, optimisticCompanions };
539599
}
540600

541-
#assertDistinctBatchFootprints(entries: readonly MaterializedOfflineEnqueue[], existingCommands: readonly OfflineCommand[]): void {
601+
#assertDistinctBatchFootprints(
602+
entries: readonly MaterializedOfflineEnqueue[],
603+
existingCommands: readonly OfflineCommand[],
604+
allowOneAggregate = false,
605+
): void {
542606
const aggregates = new Set<string>();
543607
const replicaKeys = new Set<string>();
544608
const existingFootprints = new Map<string, string>();
@@ -548,15 +612,15 @@ export class OfflineSyncService {
548612
}
549613
for (const entry of entries) {
550614
const aggregate = this.#aggregateKey(entry.command);
551-
if (aggregates.has(aggregate)) {
615+
if (aggregates.has(aggregate) && !allowOneAggregate) {
552616
throw new Error('Prepared offline batch contains overlapping aggregate intents.');
553617
}
554618
aggregates.add(aggregate);
555619
for (const key of [
556620
this.#replicaRowKey(entry.optimisticRow),
557621
...entry.optimisticCompanions.map((companion) => this.#replicaRowKey(companion.key)),
558622
]) {
559-
if (replicaKeys.has(key)) {
623+
if (replicaKeys.has(key) && !allowOneAggregate) {
560624
throw new Error('Prepared offline batch contains overlapping replica footprints.');
561625
}
562626
const existingAggregate = existingFootprints.get(key);
@@ -700,12 +764,12 @@ export class OfflineSyncService {
700764
async #assertOutboxCapacity(
701765
userId: OfflinePrincipalId,
702766
newCommands: readonly OfflineCommand[],
703-
excludingCommandId?: string,
767+
excludingCommandIds?: readonly string[],
704768
knownCommands?: readonly OfflineCommand[],
705769
): Promise<void> {
706770
const currentCommands = knownCommands ?? (await this.#commandsForUser(userId));
707-
const commands = excludingCommandId
708-
? currentCommands.filter((candidate) => candidate.commandId !== excludingCommandId)
771+
const commands = excludingCommandIds
772+
? currentCommands.filter((candidate) => !excludingCommandIds.includes(candidate.commandId))
709773
: currentCommands;
710774
const maxCommands = this.#options.outboxLimits?.maxCommandsPerUser ?? DEFAULT_MAX_OUTBOX_COMMANDS_PER_USER;
711775
const maxBytes = this.#options.outboxLimits?.maxBytesPerUser ?? DEFAULT_MAX_OUTBOX_BYTES_PER_USER;

0 commit comments

Comments
 (0)