From cc2e5652cf809e606e96d8a07aa13a4198d44d90 Mon Sep 17 00:00:00 2001 From: rdlabo Date: Wed, 12 Aug 2026 15:50:54 +0900 Subject: [PATCH 1/5] feat(offline): support product-defined conflict rebasing --- .../src/lib/offline-command-executor.ts | 22 ++++- .../lib/offline-replica-pull.service.spec.ts | 90 ++++++++++++++++++- .../src/lib/offline-replica-pull.service.ts | 48 +++++++++- 3 files changed, 156 insertions(+), 4 deletions(-) diff --git a/projects/kit/offline/src/lib/offline-command-executor.ts b/projects/kit/offline/src/lib/offline-command-executor.ts index fcdd1b2..8c11d7b 100644 --- a/projects/kit/offline/src/lib/offline-command-executor.ts +++ b/projects/kit/offline/src/lib/offline-command-executor.ts @@ -1,5 +1,5 @@ import { InjectionToken } from '@angular/core'; -import type { OfflineCommand, OfflineScope } from './offline-repository'; +import type { OfflineCommand, OfflineReplicaRow, OfflineReplicaRowKey, OfflineScope } from './offline-repository'; import type { OfflineCommandIdentity, OfflinePrincipalId, OfflineReplicaIdentity } from './offline-identity'; import type { OfflineGeneratedRemoteId, OfflineNaturalKey } from './offline-replica-schema'; @@ -31,6 +31,18 @@ export interface OfflineCommandExecutor { /** Sends the command using `command.commandId` as its durable server-side idempotency key. */ execute(command: OfflineCommand, target: OfflineCommandTarget): Promise; withServerRevision(command: OfflineCommand, revision: string | number): OfflineCommand; + /** + * Reapplies a complete aggregate's pending intents to a newer confirmed + * value. Return null when any intent is revision-sensitive. The returned + * commands must preserve identity and order while updating baseRevision and + * optimisticValue. Without this hook, revision changes conflict by default. + */ + rebasePendingCommands?( + commands: readonly OfflineCommand[], + confirmedValues: unknown, + revision: string | number, + companionRows: readonly OfflineReplicaRow[], + ): OfflinePendingRebase | null | Promise; /** * Whether this transport error authoritatively proves that this idempotency * key did not commit. Returning true may clear an ambiguity retained from an @@ -44,6 +56,14 @@ export interface OfflineCommandExecutor { withoutServerRevision?(command: OfflineCommand): OfflineCommand; } +export interface OfflinePendingRebase { + /** Commands rebased in their original durable FIFO order. */ + commands: readonly OfflineCommand[]; + /** Product-owned companion rows rematerialized from the new confirmed value. */ + putRows?: readonly OfflineReplicaRow[]; + removeRows?: readonly OfflineReplicaRowKey[]; +} + /** DI token for the product-specific command transport adapter. */ export const OFFLINE_COMMAND_EXECUTOR = new InjectionToken('OFFLINE_COMMAND_EXECUTOR'); diff --git a/projects/kit/offline/src/lib/offline-replica-pull.service.spec.ts b/projects/kit/offline/src/lib/offline-replica-pull.service.spec.ts index 90ecf02..1b4b00d 100644 --- a/projects/kit/offline/src/lib/offline-replica-pull.service.spec.ts +++ b/projects/kit/offline/src/lib/offline-replica-pull.service.spec.ts @@ -19,6 +19,7 @@ import { defineReplicaEntity, integer, generatedId, + localOnly, sha256OfflineReplicaSchema, text, } from './offline-replica-schema'; @@ -44,9 +45,17 @@ const testItemEntity = defineReplicaEntity()({ }, }); +const testViewEntity = defineReplicaEntity<{ title: string }>()({ + table: 'test_views', + sourceKey: 'test_views', + scope: 'user', + identity: localOnly(), + fields: { title: text() }, +}); + const replicaSchema = defineOfflineReplicaSchema({ version: 1, - entities: [testItemEntity], + entities: [testItemEntity, testViewEntity], migrations: [], }); @@ -149,6 +158,7 @@ describe('OfflineReplicaPullService', () => { provide: OFFLINE_COMMAND_EXECUTOR, useValue: { execute: vi.fn(), + rebasePendingCommands: vi.fn(() => null), withServerRevision: (command: OfflineCommand, revision: string | number) => ({ ...command, baseRevision: revision, @@ -640,6 +650,84 @@ describe('OfflineReplicaPullService', () => { ]); }); + it('rebase policyは他端末revisionへintentを移しconflictにしない', async () => { + const executor = TestBed.inject(OFFLINE_COMMAND_EXECUTOR); + const rebase = vi.spyOn(executor, 'rebasePendingCommands').mockImplementation((commands, confirmed, revision) => ({ + commands: commands.map((command) => ({ + ...command, + baseRevision: revision, + optimisticValue: { ...(confirmed as Record), title: 'Rebased delta' }, + })), + putRows: [ + { + ...scope, + sourceKey: 'test_views', + identity: { kind: 'local', localId: 'view-42' }, + values: { title: 'Rebased view' }, + confirmedValues: { title: 'Remote view' }, + serverRevision: null, + fetchedAt: 9, + syncState: 'pending', + }, + ], + })); + await repository.transactReplica({ + putRows: [ + { + ...scope, + sourceKey: 'test_items', + identity: { kind: 'generated', localId: '019d-rebase', remoteId: 42 }, + values: { id: 42, title: 'Local delta' }, + confirmedValues: { id: 42, title: 'Old confirmed' }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'pending', + }, + ], + putCommands: [ + { + ...scope, + commandId: 'cmd-rebase', + aggregateType: 'test_items', + sourceKey: 'test_items', + identity: { kind: 'generated', localId: '019d-rebase' }, + operation: 'test_items.delta', + payload: { delta: 1 }, + optimisticValue: { id: 42, title: 'Local delta' }, + payloadHash: 'hash', + baseRevision: 1, + state: 'pending', + attempts: 0, + retryAt: null, + createdAt: 1, + lastErrorCode: null, + }, + ], + }); + pull.mockResolvedValueOnce(page([itemChange(42, 'Remote truth', { serverRevision: 9 })], { nextCursor: 'cursor-v1' })); + + await service.pull(scope); + + expect(rebase).toHaveBeenCalledWith( + [expect.objectContaining({ commandId: 'cmd-rebase' })], + expect.objectContaining({ title: 'Remote truth' }), + 9, + [], + ); + await expect(repository.getReplicaRow(scope, 'test_items', generatedCommandIdentity('019d-rebase'))).resolves.toMatchObject({ + values: { title: 'Rebased delta' }, + confirmedValues: { title: 'Remote truth' }, + serverRevision: 9, + syncState: 'pending', + }); + await expect(repository.getCommands(scope)).resolves.toEqual([ + expect.objectContaining({ commandId: 'cmd-rebase', baseRevision: 9, state: 'pending', lastErrorCode: null }), + ]); + await expect( + repository.getReplicaRow(scope, 'test_views', { kind: 'local', localId: 'view-42' }), + ).resolves.toMatchObject({ values: { title: 'Rebased view' }, confirmedValues: { title: 'Remote view' } }); + }); + it('remote tombstone conflictはpending commandをremote_deleted conflictへ遷移する', async () => { await repository.transactReplica({ putRows: [ 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 6499284..91d0eca 100644 --- a/projects/kit/offline/src/lib/offline-replica-pull.service.ts +++ b/projects/kit/offline/src/lib/offline-replica-pull.service.ts @@ -170,10 +170,23 @@ export class OfflineReplicaPullService { continue; } - const conflicted = related.some((command) => command.baseRevision !== change.serverRevision); + const revisionChanged = related.some((command) => command.baseRevision !== change.serverRevision); + const rebase = revisionChanged + ? ((await this.#executor.rebasePendingCommands?.( + related, + confirmedValues, + change.serverRevision, + await this.#currentCompanionRows(scope, related), + )) ?? null) + : null; + this.#assertRebasedCommands(related, rebase?.commands ?? null); + const conflicted = revisionChanged && rebase === null; + for (const command of rebase?.commands ?? []) putCommands.set(command.commandId, command); + putRows.push(...(rebase?.putRows ?? [])); + removeRows.push(...(rebase?.removeRows ?? [])); putRows.push({ ...existing, - values: hasPending ? existing.values : confirmedValues, + values: rebase?.commands.at(-1)?.optimisticValue ?? (hasPending ? existing.values : confirmedValues), confirmedValues, serverRevision: change.serverRevision, fetchedAt: Date.now(), @@ -224,6 +237,37 @@ export class OfflineReplicaPullService { } } + #assertRebasedCommands( + original: readonly OfflineCommand[], + rebased: readonly OfflineCommand[] | null, + ): void { + if (rebased === null) return; + if ( + rebased.length !== original.length || + rebased.some((command, index) => command.commandId !== original[index]?.commandId) + ) { + throw new Error('Rebased offline commands must preserve aggregate command identity and order.'); + } + } + + async #currentCompanionRows( + scope: OfflineScope, + commands: readonly OfflineCommand[], + ): Promise { + const keys = new Map( + commands.flatMap((command) => + (command.optimisticCompanions ?? []).map((companion) => [ + `${companion.key.sourceKey}:${JSON.stringify(companion.key.identity)}`, + companion.key, + ] as const), + ), + ); + const rows = await Promise.all( + [...keys.values()].map((key) => this.#repository.getReplicaRow(scope, key.sourceKey, key.identity)), + ); + return rows.filter((row): row is OfflineReplicaRow => row !== null); + } + #assertPullChange(change: unknown, index: number): void { const label = `Offline replica pull page changes[${index}]`; if (!isPlainObject(change)) { From a8723aaa6f02de55b055236a9ab7a46593aeb324 Mon Sep 17 00:00:00 2001 From: rdlabo Date: Wed, 12 Aug 2026 15:54:53 +0900 Subject: [PATCH 2/5] refactor(offline): keep rebased command payload immutable --- .../src/lib/offline-command-executor.ts | 9 ++++--- .../lib/offline-replica-pull.service.spec.ts | 16 ++++++++---- .../src/lib/offline-replica-pull.service.ts | 26 +++++++++++-------- 3 files changed, 31 insertions(+), 20 deletions(-) diff --git a/projects/kit/offline/src/lib/offline-command-executor.ts b/projects/kit/offline/src/lib/offline-command-executor.ts index 8c11d7b..370e76c 100644 --- a/projects/kit/offline/src/lib/offline-command-executor.ts +++ b/projects/kit/offline/src/lib/offline-command-executor.ts @@ -34,8 +34,9 @@ export interface OfflineCommandExecutor { /** * Reapplies a complete aggregate's pending intents to a newer confirmed * value. Return null when any intent is revision-sensitive. The returned - * commands must preserve identity and order while updating baseRevision and - * optimisticValue. Without this hook, revision changes conflict by default. + * values correspond to the original FIFO command order. Kit alone updates + * command metadata; payload and idempotency identity remain immutable. + * Without this hook, revision changes conflict by default. */ rebasePendingCommands?( commands: readonly OfflineCommand[], @@ -57,8 +58,8 @@ export interface OfflineCommandExecutor { } export interface OfflinePendingRebase { - /** Commands rebased in their original durable FIFO order. */ - commands: readonly OfflineCommand[]; + /** Recomputed optimistic values in the original durable FIFO order. */ + optimisticValues: readonly unknown[]; /** Product-owned companion rows rematerialized from the new confirmed value. */ putRows?: readonly OfflineReplicaRow[]; removeRows?: readonly OfflineReplicaRowKey[]; diff --git a/projects/kit/offline/src/lib/offline-replica-pull.service.spec.ts b/projects/kit/offline/src/lib/offline-replica-pull.service.spec.ts index 1b4b00d..7fea112 100644 --- a/projects/kit/offline/src/lib/offline-replica-pull.service.spec.ts +++ b/projects/kit/offline/src/lib/offline-replica-pull.service.spec.ts @@ -653,10 +653,9 @@ describe('OfflineReplicaPullService', () => { it('rebase policyは他端末revisionへintentを移しconflictにしない', async () => { const executor = TestBed.inject(OFFLINE_COMMAND_EXECUTOR); const rebase = vi.spyOn(executor, 'rebasePendingCommands').mockImplementation((commands, confirmed, revision) => ({ - commands: commands.map((command) => ({ - ...command, - baseRevision: revision, - optimisticValue: { ...(confirmed as Record), title: 'Rebased delta' }, + optimisticValues: commands.map(() => ({ + ...(confirmed as Record), + title: 'Rebased delta', })), putRows: [ { @@ -721,7 +720,14 @@ describe('OfflineReplicaPullService', () => { syncState: 'pending', }); await expect(repository.getCommands(scope)).resolves.toEqual([ - expect.objectContaining({ commandId: 'cmd-rebase', baseRevision: 9, state: 'pending', lastErrorCode: null }), + expect.objectContaining({ + commandId: 'cmd-rebase', + payload: { delta: 1 }, + payloadHash: 'hash', + baseRevision: 9, + state: 'pending', + lastErrorCode: null, + }), ]); await expect( repository.getReplicaRow(scope, 'test_views', { kind: 'local', localId: 'view-42' }), 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 91d0eca..d801f24 100644 --- a/projects/kit/offline/src/lib/offline-replica-pull.service.ts +++ b/projects/kit/offline/src/lib/offline-replica-pull.service.ts @@ -179,14 +179,21 @@ export class OfflineReplicaPullService { await this.#currentCompanionRows(scope, related), )) ?? null) : null; - this.#assertRebasedCommands(related, rebase?.commands ?? null); + this.#assertRebasedValues(related, rebase?.optimisticValues ?? null); const conflicted = revisionChanged && rebase === null; - for (const command of rebase?.commands ?? []) putCommands.set(command.commandId, command); + const rebasedCommands = rebase + ? related.map((command, index) => ({ + ...command, + baseRevision: change.serverRevision, + optimisticValue: rebase.optimisticValues[index], + })) + : []; + for (const command of rebasedCommands) putCommands.set(command.commandId, command); putRows.push(...(rebase?.putRows ?? [])); removeRows.push(...(rebase?.removeRows ?? [])); putRows.push({ ...existing, - values: rebase?.commands.at(-1)?.optimisticValue ?? (hasPending ? existing.values : confirmedValues), + values: rebasedCommands.at(-1)?.optimisticValue ?? (hasPending ? existing.values : confirmedValues), confirmedValues, serverRevision: change.serverRevision, fetchedAt: Date.now(), @@ -237,16 +244,13 @@ export class OfflineReplicaPullService { } } - #assertRebasedCommands( + #assertRebasedValues( original: readonly OfflineCommand[], - rebased: readonly OfflineCommand[] | null, + optimisticValues: readonly unknown[] | null, ): void { - if (rebased === null) return; - if ( - rebased.length !== original.length || - rebased.some((command, index) => command.commandId !== original[index]?.commandId) - ) { - throw new Error('Rebased offline commands must preserve aggregate command identity and order.'); + if (optimisticValues === null) return; + if (optimisticValues.length !== original.length) { + throw new Error('Rebased offline values must match the aggregate command count.'); } } From d1f033996def002d7d7fab4636602eed2db18838 Mon Sep 17 00:00:00 2001 From: rdlabo Date: Wed, 12 Aug 2026 16:01:21 +0900 Subject: [PATCH 3/5] fix(offline): rebase companion history atomically --- .../src/lib/offline-command-executor.ts | 15 +++--- .../lib/offline-replica-pull.service.spec.ts | 26 +++++++---- .../src/lib/offline-replica-pull.service.ts | 46 +++++++++++++++---- 3 files changed, 65 insertions(+), 22 deletions(-) diff --git a/projects/kit/offline/src/lib/offline-command-executor.ts b/projects/kit/offline/src/lib/offline-command-executor.ts index 370e76c..418b4e7 100644 --- a/projects/kit/offline/src/lib/offline-command-executor.ts +++ b/projects/kit/offline/src/lib/offline-command-executor.ts @@ -1,5 +1,5 @@ import { InjectionToken } from '@angular/core'; -import type { OfflineCommand, OfflineReplicaRow, OfflineReplicaRowKey, OfflineScope } from './offline-repository'; +import type { OfflineCommand, OfflineOptimisticReplicaCompanion, OfflineReplicaRow, OfflineScope } from './offline-repository'; import type { OfflineCommandIdentity, OfflinePrincipalId, OfflineReplicaIdentity } from './offline-identity'; import type { OfflineGeneratedRemoteId, OfflineNaturalKey } from './offline-replica-schema'; @@ -58,11 +58,14 @@ export interface OfflineCommandExecutor { } export interface OfflinePendingRebase { - /** Recomputed optimistic values in the original durable FIFO order. */ - optimisticValues: readonly unknown[]; - /** Product-owned companion rows rematerialized from the new confirmed value. */ - putRows?: readonly OfflineReplicaRow[]; - removeRows?: readonly OfflineReplicaRowKey[]; + /** Recomputed projections in the original durable FIFO order. */ + steps: readonly OfflinePendingRebaseStep[]; +} + +export interface OfflinePendingRebaseStep { + optimisticValue: unknown; + /** Same footprint as the original command, rematerialized from the new confirmed value. */ + optimisticCompanions?: readonly OfflineOptimisticReplicaCompanion[]; } /** DI token for the product-specific command transport adapter. */ diff --git a/projects/kit/offline/src/lib/offline-replica-pull.service.spec.ts b/projects/kit/offline/src/lib/offline-replica-pull.service.spec.ts index 7fea112..f634acf 100644 --- a/projects/kit/offline/src/lib/offline-replica-pull.service.spec.ts +++ b/projects/kit/offline/src/lib/offline-replica-pull.service.spec.ts @@ -653,12 +653,16 @@ describe('OfflineReplicaPullService', () => { it('rebase policyは他端末revisionへintentを移しconflictにしない', async () => { const executor = TestBed.inject(OFFLINE_COMMAND_EXECUTOR); const rebase = vi.spyOn(executor, 'rebasePendingCommands').mockImplementation((commands, confirmed, revision) => ({ - optimisticValues: commands.map(() => ({ - ...(confirmed as Record), - title: 'Rebased delta', - })), - putRows: [ - { + steps: commands.map(() => ({ + optimisticValue: { ...(confirmed as Record), title: 'Rebased delta' }, + optimisticCompanions: [{ + key: { + ...scope, + sourceKey: 'test_views', + identity: { kind: 'local', localId: 'view-42' }, + }, + before: null, + after: { ...scope, sourceKey: 'test_views', identity: { kind: 'local', localId: 'view-42' }, @@ -667,8 +671,9 @@ describe('OfflineReplicaPullService', () => { serverRevision: null, fetchedAt: 9, syncState: 'pending', - }, - ], + }, + }], + })), })); await repository.transactReplica({ putRows: [ @@ -693,6 +698,11 @@ describe('OfflineReplicaPullService', () => { operation: 'test_items.delta', payload: { delta: 1 }, optimisticValue: { id: 42, title: 'Local delta' }, + optimisticCompanions: [{ + key: { ...scope, sourceKey: 'test_views', identity: { kind: 'local', localId: 'view-42' } }, + before: null, + after: null, + }], payloadHash: 'hash', baseRevision: 1, state: 'pending', 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 d801f24..93ab279 100644 --- a/projects/kit/offline/src/lib/offline-replica-pull.service.ts +++ b/projects/kit/offline/src/lib/offline-replica-pull.service.ts @@ -179,18 +179,20 @@ export class OfflineReplicaPullService { await this.#currentCompanionRows(scope, related), )) ?? null) : null; - this.#assertRebasedValues(related, rebase?.optimisticValues ?? null); + this.#assertRebasedSteps(related, rebase?.steps ?? null); const conflicted = revisionChanged && rebase === null; const rebasedCommands = rebase ? related.map((command, index) => ({ ...command, baseRevision: change.serverRevision, - optimisticValue: rebase.optimisticValues[index], + optimisticValue: rebase.steps[index]!.optimisticValue, + ...(rebase.steps[index]!.optimisticCompanions === undefined + ? {} + : { optimisticCompanions: rebase.steps[index]!.optimisticCompanions }), })) : []; for (const command of rebasedCommands) putCommands.set(command.commandId, command); - putRows.push(...(rebase?.putRows ?? [])); - removeRows.push(...(rebase?.removeRows ?? [])); + this.#appendRebasedCompanionRows(rebasedCommands, putRows, removeRows); putRows.push({ ...existing, values: rebasedCommands.at(-1)?.optimisticValue ?? (hasPending ? existing.values : confirmedValues), @@ -244,14 +246,42 @@ export class OfflineReplicaPullService { } } - #assertRebasedValues( + #assertRebasedSteps( original: readonly OfflineCommand[], - optimisticValues: readonly unknown[] | null, + steps: readonly { optimisticCompanions?: readonly { key: OfflineReplicaRowKey }[] }[] | null, ): void { - if (optimisticValues === null) return; - if (optimisticValues.length !== original.length) { + if (steps === null) return; + if (steps.length !== original.length) { throw new Error('Rebased offline values must match the aggregate command count.'); } + for (const [index, step] of steps.entries()) { + const originalKeys = (original[index]?.optimisticCompanions ?? []).map((item) => this.#rowKey(item.key)).sort(); + const rebasedKeys = (step.optimisticCompanions ?? []).map((item) => this.#rowKey(item.key)).sort(); + if (JSON.stringify(originalKeys) !== JSON.stringify(rebasedKeys)) { + throw new Error('Rebased offline companions must preserve each command footprint.'); + } + } + } + + #appendRebasedCompanionRows( + commands: readonly OfflineCommand[], + putRows: OfflineReplicaRow[], + removeRows: OfflineReplicaRowKey[], + ): void { + const latest = new Map(); + for (const command of commands) { + for (const companion of command.optimisticCompanions ?? []) { + latest.set(this.#rowKey(companion.key), { key: companion.key, after: companion.after }); + } + } + for (const { key, after } of latest.values()) { + if (after) putRows.push(after); + else removeRows.push(key); + } + } + + #rowKey(key: OfflineReplicaRowKey): string { + return `${key.userId}:${key.scopeId}:${key.sourceKey}:${JSON.stringify(key.identity)}`; } async #currentCompanionRows( From 416ffc888d80619a5e2f8d174f31ef47d0b91c02 Mon Sep 17 00:00:00 2001 From: rdlabo Date: Wed, 12 Aug 2026 16:04:45 +0900 Subject: [PATCH 4/5] fix(offline): constrain revision rebase eligibility --- .../src/lib/offline-command-executor.ts | 3 + .../lib/offline-replica-pull.service.spec.ts | 114 ++++++++++++++++++ .../src/lib/offline-replica-pull.service.ts | 21 ++-- 3 files changed, 131 insertions(+), 7 deletions(-) diff --git a/projects/kit/offline/src/lib/offline-command-executor.ts b/projects/kit/offline/src/lib/offline-command-executor.ts index 418b4e7..bcd86bd 100644 --- a/projects/kit/offline/src/lib/offline-command-executor.ts +++ b/projects/kit/offline/src/lib/offline-command-executor.ts @@ -57,12 +57,15 @@ export interface OfflineCommandExecutor { withoutServerRevision?(command: OfflineCommand): OfflineCommand; } +/** Product projection result after safely replaying pending intents onto a newer confirmed revision. */ export interface OfflinePendingRebase { /** Recomputed projections in the original durable FIFO order. */ steps: readonly OfflinePendingRebaseStep[]; } +/** Projection state produced for one immutable durable command in FIFO order. */ export interface OfflinePendingRebaseStep { + /** Recomputed full aggregate value after this command's intent is applied. */ optimisticValue: unknown; /** Same footprint as the original command, rematerialized from the new confirmed value. */ optimisticCompanions?: readonly OfflineOptimisticReplicaCompanion[]; diff --git a/projects/kit/offline/src/lib/offline-replica-pull.service.spec.ts b/projects/kit/offline/src/lib/offline-replica-pull.service.spec.ts index f634acf..442b162 100644 --- a/projects/kit/offline/src/lib/offline-replica-pull.service.spec.ts +++ b/projects/kit/offline/src/lib/offline-replica-pull.service.spec.ts @@ -744,6 +744,120 @@ describe('OfflineReplicaPullService', () => { ).resolves.toMatchObject({ values: { title: 'Rebased view' }, confirmedValues: { title: 'Remote view' } }); }); + it.each(['conflict', 'rejected', 'blocked_auth'] as const)( + '%s commandは自動rebaseせずattention stateを維持する', + async (state) => { + const executor = TestBed.inject(OFFLINE_COMMAND_EXECUTOR); + const rebase = vi.spyOn(executor, 'rebasePendingCommands'); + await repository.transactReplica({ + putRows: [ + { + ...scope, + sourceKey: 'test_items', + identity: { kind: 'generated', localId: `019d-${state}`, remoteId: 42 }, + values: { id: 42, title: 'Local' }, + confirmedValues: { id: 42, title: 'Old' }, + serverRevision: 1, + fetchedAt: 1, + syncState: state, + }, + ], + putCommands: [ + { + ...scope, + commandId: `cmd-${state}`, + aggregateType: 'test_items', + sourceKey: 'test_items', + identity: { kind: 'generated', localId: `019d-${state}` }, + operation: 'test_items.delta', + payload: { delta: 1 }, + optimisticValue: { id: 42, title: 'Local' }, + payloadHash: 'hash', + baseRevision: 1, + state, + attempts: 1, + retryAt: null, + createdAt: 1, + lastErrorCode: state, + }, + ], + }); + pull.mockResolvedValueOnce(page([itemChange(42, 'Remote', { serverRevision: 9 })], { nextCursor: 'cursor-v1' })); + + await service.pull(scope); + + expect(rebase).not.toHaveBeenCalled(); + await expect(repository.getCommands(scope)).resolves.toEqual([ + expect.objectContaining({ commandId: `cmd-${state}`, state: 'conflict', lastErrorCode: 'remote_revision' }), + ]); + await expect(repository.getReplicaRow(scope, 'test_items', generatedCommandIdentity(`019d-${state}`))).resolves.toMatchObject({ + syncState: 'conflict', + }); + }, + ); + + it('rebase reducer receives a pending-delete companion from its own scope', async () => { + const otherScope = { userId: scope.userId, scopeId: '20' }; + const companion: OfflineReplicaRow = { + ...otherScope, + sourceKey: 'test_views', + identity: { kind: 'local', localId: 'hidden-view' }, + values: { title: 'Hidden' }, + confirmedValues: { title: 'Confirmed hidden' }, + serverRevision: null, + fetchedAt: 1, + syncState: 'pending', + visibility: 'pending_delete', + }; + const executor = TestBed.inject(OFFLINE_COMMAND_EXECUTOR); + const rebase = vi.spyOn(executor, 'rebasePendingCommands').mockReturnValue(null); + await repository.transactReplica({ + putRows: [ + { + ...scope, + sourceKey: 'test_items', + identity: { kind: 'generated', localId: '019d-cross-scope', remoteId: 42 }, + values: { id: 42, title: 'Local' }, + confirmedValues: { id: 42, title: 'Old' }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'pending', + }, + companion, + ], + putCommands: [ + { + ...scope, + commandId: 'cmd-cross-scope', + aggregateType: 'test_items', + sourceKey: 'test_items', + identity: { kind: 'generated', localId: '019d-cross-scope' }, + operation: 'test_items.delta', + payload: { delta: 1 }, + optimisticValue: { id: 42, title: 'Local' }, + optimisticCompanions: [{ key: companion, before: null, after: companion }], + payloadHash: 'hash', + baseRevision: 1, + state: 'pending', + attempts: 0, + retryAt: null, + createdAt: 1, + lastErrorCode: null, + }, + ], + }); + pull.mockResolvedValueOnce(page([itemChange(42, 'Remote', { serverRevision: 9 })], { nextCursor: 'cursor-v1' })); + + await service.pull(scope); + + expect(rebase).toHaveBeenCalledWith( + expect.any(Array), + expect.anything(), + 9, + [expect.objectContaining({ scopeId: '20', visibility: 'pending_delete' })], + ); + }); + it('remote tombstone conflictはpending commandをremote_deleted conflictへ遷移する', async () => { await repository.transactReplica({ putRows: [ 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 93ab279..61e530e 100644 --- a/projects/kit/offline/src/lib/offline-replica-pull.service.ts +++ b/projects/kit/offline/src/lib/offline-replica-pull.service.ts @@ -171,12 +171,16 @@ export class OfflineReplicaPullService { } const revisionChanged = related.some((command) => command.baseRevision !== change.serverRevision); - const rebase = revisionChanged + const rebaseEligible = related.every( + (command) => + (command.state === 'pending' || command.state === 'retry_wait') && command.serverCommitUnknown !== true, + ); + const rebase = revisionChanged && rebaseEligible ? ((await this.#executor.rebasePendingCommands?.( related, confirmedValues, change.serverRevision, - await this.#currentCompanionRows(scope, related), + await this.#currentCompanionRows(related), )) ?? null) : null; this.#assertRebasedSteps(related, rebase?.steps ?? null); @@ -284,10 +288,7 @@ export class OfflineReplicaPullService { return `${key.userId}:${key.scopeId}:${key.sourceKey}:${JSON.stringify(key.identity)}`; } - async #currentCompanionRows( - scope: OfflineScope, - commands: readonly OfflineCommand[], - ): Promise { + async #currentCompanionRows(commands: readonly OfflineCommand[]): Promise { const keys = new Map( commands.flatMap((command) => (command.optimisticCompanions ?? []).map((companion) => [ @@ -297,7 +298,13 @@ export class OfflineReplicaPullService { ), ); const rows = await Promise.all( - [...keys.values()].map((key) => this.#repository.getReplicaRow(scope, key.sourceKey, key.identity)), + [...keys.values()].map((key) => { + const companionScope = { userId: key.userId, scopeId: key.scopeId }; + return ( + this.#repository.getReplicaRowIncludingPendingDelete?.(companionScope, key.sourceKey, key.identity) ?? + this.#repository.getReplicaRow(companionScope, key.sourceKey, key.identity) + ); + }), ); return rows.filter((row): row is OfflineReplicaRow => row !== null); } From ed0c694038b0a2757e46baa6529fd48e434344f2 Mon Sep 17 00:00:00 2001 From: rdlabo Date: Wed, 12 Aug 2026 16:06:44 +0900 Subject: [PATCH 5/5] fix(offline): dedupe companions by scoped key --- projects/kit/offline/src/lib/offline-replica-pull.service.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 61e530e..42727b4 100644 --- a/projects/kit/offline/src/lib/offline-replica-pull.service.ts +++ b/projects/kit/offline/src/lib/offline-replica-pull.service.ts @@ -292,7 +292,7 @@ export class OfflineReplicaPullService { const keys = new Map( commands.flatMap((command) => (command.optimisticCompanions ?? []).map((companion) => [ - `${companion.key.sourceKey}:${JSON.stringify(companion.key.identity)}`, + this.#rowKey(companion.key), companion.key, ] as const), ),