From cc2e5652cf809e606e96d8a07aa13a4198d44d90 Mon Sep 17 00:00:00 2001 From: rdlabo Date: Wed, 12 Aug 2026 15:50:54 +0900 Subject: [PATCH 01/10] 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 02/10] 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 03/10] 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 04/10] 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 05/10] 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), ), From 80c75f39e502af00ec3698276da8a637988035c2 Mon Sep 17 00:00:00 2001 From: rdlabo Date: Wed, 12 Aug 2026 16:26:13 +0900 Subject: [PATCH 06/10] feat(offline): atomically apply rebaseline projections --- .../kit/offline/src/lib/offline-provider.ts | 9 +- .../lib/offline-replica-pull.service.spec.ts | 100 +++++++++++++++++- .../src/lib/offline-replica-pull.service.ts | 61 ++++++++++- .../offline/src/lib/offline-replica-puller.ts | 23 +++- 4 files changed, 186 insertions(+), 7 deletions(-) diff --git a/projects/kit/offline/src/lib/offline-provider.ts b/projects/kit/offline/src/lib/offline-provider.ts index 206e52f..d991539 100644 --- a/projects/kit/offline/src/lib/offline-provider.ts +++ b/projects/kit/offline/src/lib/offline-provider.ts @@ -11,8 +11,8 @@ import { OfflineCoordinatorService } from './offline-coordinator.service'; import { IonicOfflineRepository, OFFLINE_REPOSITORY, selectOfflineRepository } from './offline-repository'; import type { OfflineMutationRequestPolicy, OfflineRequestPolicy } from './offline-request-policy'; import { provideOfflineMutationRequestPolicy, provideOfflineRequestPolicy } from './offline-request-policy'; -import type { OfflineReplicaPuller } from './offline-replica-puller'; -import { OFFLINE_REPLICA_PULLER } from './offline-replica-puller'; +import type { OfflineReplicaProjector, OfflineReplicaPuller } from './offline-replica-puller'; +import { OFFLINE_REPLICA_PROJECTOR, OFFLINE_REPLICA_PULLER } from './offline-replica-puller'; import { OfflineSessionService } from './offline-session.service'; import { COMMUNITY_SQLITE, @@ -29,6 +29,8 @@ interface ProvideOfflineOptionsBase extends OfflineKitOptions { commandHooks?: Type; /** Optional additional providers required by product adapters. */ providers?: readonly Provider[]; + /** Optional pure adapter for product-owned local-only projections applied inside the Kit pull transaction. */ + replicaProjector?: Type; /** Application-installed `@capacitor-community/sqlite` connection. Required only on iOS and Android. */ sqliteConnection?: CommunitySqliteConnection; } @@ -109,6 +111,9 @@ export function provideOffline(options: ProvideOfflineOptions): EnvironmentProvi ? { provide: OFFLINE_REPLICA_PULLER, useExisting: options.replicaPuller } : { provide: OFFLINE_REPLICA_PULLER, useValue: READ_CACHE_ONLY_REPLICA_PULLER }, ...(options.commandHooks ? [options.commandHooks, { provide: OFFLINE_COMMAND_HOOKS, useExisting: options.commandHooks }] : []), + ...(options.replicaProjector + ? [options.replicaProjector, { provide: OFFLINE_REPLICA_PROJECTOR, useExisting: options.replicaProjector }] + : []), ...options.requestPolicies.flatMap((policy) => provideOfflineRequestPolicy(policy)), ...(options.mutationPolicies ?? []).flatMap((policy) => provideOfflineMutationRequestPolicy(policy)), ...(options.providers ?? []), 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..8f64ba2 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 @@ -6,6 +6,7 @@ import { OFFLINE_COMMAND_EXECUTOR } from './offline-command-executor'; import { OFFLINE_COMMAND_HOOKS } from './offline-command-hooks'; import { OFFLINE_KIT_OPTIONS } from './offline-kit-options'; import { + OFFLINE_REPLICA_PROJECTOR, OFFLINE_REPLICA_PULLER, type OfflineReplicaChange, type OfflineReplicaPullPage, @@ -19,6 +20,7 @@ import { defineReplicaEntity, integer, generatedId, + localOnly, sha256OfflineReplicaSchema, text, } from './offline-replica-schema'; @@ -44,9 +46,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: [], }); @@ -91,6 +101,7 @@ describe('OfflineReplicaPullService', () => { let storage: MemoryStorage; let schemaHash: string; let pull: ReturnType Promise>>; + let projector: { project: ReturnType }; function page( changes: readonly OfflineReplicaChange[], @@ -99,6 +110,7 @@ describe('OfflineReplicaPullService', () => { hasMore?: boolean; schemaVersion?: number; schemaHash?: string; + rebaselineRequired?: boolean; } = {}, ): OfflineReplicaPullPage { return { @@ -107,6 +119,7 @@ describe('OfflineReplicaPullService', () => { changes, nextCursor: options.nextCursor ?? 'cursor-v1', hasMore: options.hasMore ?? false, + ...(options.rebaselineRequired ? { rebaselineRequired: true } : {}), }; } @@ -141,6 +154,7 @@ describe('OfflineReplicaPullService', () => { { provide: OFFLINE_KIT_OPTIONS, useValue: { databaseName: 'test-offline', replicaSchema } }, { provide: OFFLINE_REPOSITORY, useExisting: IonicOfflineRepository }, { provide: OFFLINE_REPLICA_PULLER, useValue: { pull } }, + { provide: OFFLINE_REPLICA_PROJECTOR, useValue: projector }, { provide: OFFLINE_COMMAND_HOOKS, useValue: { entityType: (command: Pick) => command.aggregateType }, @@ -164,6 +178,7 @@ describe('OfflineReplicaPullService', () => { beforeEach(async () => { storage = new MemoryStorage(); pull = vi.fn(async () => page([])); + projector = { project: vi.fn(async () => ({})) }; await seedReplicaMetadata(); configureTestBed(); await repository.initialize(); @@ -183,6 +198,78 @@ describe('OfflineReplicaPullService', () => { }); }); + it('applies rebaseline reset, derived projection, base rows, and cursor in one transaction', async () => { + await repository.transactReplica({ + putRows: [ + { + ...scope, + sourceKey: 'test_items', + identity: { kind: 'generated', localId: 'stale', remoteId: 10 }, + values: { id: 10, title: 'Stale' }, + confirmedValues: { id: 10, title: 'Stale' }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'confirmed', + }, + ], + putCursors: [{ ...scope, cursor: 'expired' }], + }); + projector.project.mockResolvedValue({ + putRows: [ + { + ...scope, + sourceKey: 'test_views', + identity: { kind: 'local', localId: 'view-42' }, + values: { title: 'Derived' }, + confirmedValues: { title: 'Derived' }, + serverRevision: null, + fetchedAt: 2, + syncState: 'confirmed', + }, + ], + }); + const transact = vi.spyOn(repository, 'transactReplica'); + pull.mockResolvedValueOnce( + page([itemChange(42, 'Fresh')], { nextCursor: 'snapshot-1', rebaselineRequired: true }), + ); + + await service.pull(scope); + + expect(transact).toHaveBeenCalledTimes(1); + await expect(repository.getReplicaRowByRemoteId(scope, 'test_items', 10)).resolves.toBeNull(); + await expect(repository.getReplicaRowByRemoteId(scope, 'test_items', 42)).resolves.toMatchObject({ + values: { title: 'Fresh' }, + }); + await expect(repository.getReplicaRow(scope, 'test_views', { kind: 'local', localId: 'view-42' })).resolves.toMatchObject({ + values: { title: 'Derived' }, + }); + await expect(repository.getReplicaCursor(scope)).resolves.toEqual({ ...scope, cursor: 'snapshot-1' }); + }); + + it('rejects projector attempts to mutate synchronized base rows without advancing the cursor', async () => { + await repository.transactReplica({ putCursors: [{ ...scope, cursor: 'cursor-v0' }] }); + projector.project.mockResolvedValue({ + putRows: [ + { + ...scope, + sourceKey: 'test_items', + identity: { kind: 'generated', localId: 'illegal', remoteId: 42 }, + values: { id: 42, title: 'Illegal' }, + confirmedValues: { id: 42, title: 'Illegal' }, + serverRevision: 2, + fetchedAt: 2, + syncState: 'confirmed', + }, + ], + }); + pull.mockResolvedValueOnce(page([itemChange(42, 'Fresh')], { nextCursor: 'cursor-v1' })); + + await expect(service.pull(scope)).rejects.toThrow( + 'Offline replica projector may only mutate localOnly source "test_items".', + ); + await expect(repository.getReplicaCursor(scope)).resolves.toEqual({ ...scope, cursor: 'cursor-v0' }); + }); + it('does not hold the mutation lane during transport and preserves an enqueue completed before stale page apply', async () => { let releasePull!: (value: OfflineReplicaPullPage) => void; pull.mockImplementationOnce(() => new Promise((resolve) => (releasePull = resolve))); @@ -387,6 +474,17 @@ describe('OfflineReplicaPullService', () => { ); }); + it('rejects a malformed rebaseline marker without advancing the cursor', async () => { + await expectPullRejectsPreservingCursor( + () => + pull.mockResolvedValueOnce({ + ...page([], { nextCursor: 'snapshot-1' }), + rebaselineRequired: 'yes', + } as unknown as OfflineReplicaPullPage), + 'Offline replica pull page rebaselineRequired must be a boolean when present.', + ); + }); + it('malformed hasMoreはrejectしcursorを進めない', async () => { await expectPullRejectsPreservingCursor( () => 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..d3095c6 100644 --- a/projects/kit/offline/src/lib/offline-replica-pull.service.ts +++ b/projects/kit/offline/src/lib/offline-replica-pull.service.ts @@ -2,7 +2,12 @@ import { inject, Injectable } from '@angular/core'; import { OFFLINE_KIT_OPTIONS } from './offline-kit-options'; import { OFFLINE_COMMAND_EXECUTOR } from './offline-command-executor'; import { OfflineReplicaMutationCoordinator } from './offline-replica-mutation-coordinator'; -import { OFFLINE_REPLICA_PULLER, type OfflineReplicaChange, type OfflineReplicaPullPage } from './offline-replica-puller'; +import { + OFFLINE_REPLICA_PROJECTOR, + OFFLINE_REPLICA_PULLER, + type OfflineReplicaChange, + type OfflineReplicaPullPage, +} from './offline-replica-puller'; import { canonicalOfflineCommandIdentity, commandIdentityFromReplicaIdentity, @@ -40,6 +45,7 @@ export class OfflineReplicaPullService { readonly #repository = inject(OFFLINE_REPOSITORY); readonly #options = inject(OFFLINE_KIT_OPTIONS); readonly #puller = inject(OFFLINE_REPLICA_PULLER); + readonly #projector = inject(OFFLINE_REPLICA_PROJECTOR, { optional: true }); readonly #executor = inject(OFFLINE_COMMAND_EXECUTOR); readonly #replicaMutations = inject(OfflineReplicaMutationCoordinator); #schemaHash: Promise | null = null; @@ -75,6 +81,9 @@ export class OfflineReplicaPullService { const removeRows: OfflineReplicaRowKey[] = []; const putCommands = new Map(); const removeCommandIds = new Set(); + if (page.rebaselineRequired) { + removeRows.push(...(await this.#confirmedRowsForRebaseline(scope, scopeCommands))); + } for (const change of changes) { const schema = this.#entitySchema(change.sourceKey); @@ -191,9 +200,16 @@ export class OfflineReplicaPullService { } } + const projection = await this.#projector?.project({ + scope, + changes, + commands: scopeCommands, + repository: this.#repository, + }); + this.#assertProjection(scope, projection); await this.#repository.transactReplica({ - putRows, - removeRows, + putRows: [...putRows, ...(projection?.putRows ?? [])], + removeRows: [...removeRows, ...(projection?.removeRows ?? [])], putCommands: [...putCommands.values()], removeCommandIds: [...removeCommandIds], putCursors: [{ ...scope, cursor: page.nextCursor }], @@ -209,6 +225,42 @@ export class OfflineReplicaPullService { } } + async #confirmedRowsForRebaseline( + scope: OfflineScope, + commands: readonly OfflineCommand[], + ): Promise { + const preserved = new Set( + commands.flatMap((command) => + (command.optimisticCompanions ?? []).map((companion) => this.#rowKey(companion.key)), + ), + ); + const rows = ( + await Promise.all( + this.#options.replicaSchema.entities.map((entity) => this.#repository.getReplicaRows(scope, entity.sourceKey)), + ) + ).flat(); + return rows.filter((row) => row.syncState === 'confirmed' && !preserved.has(this.#rowKey(row))); + } + + #rowKey(row: OfflineReplicaRowKey): string { + return `${row.userId}:${row.scopeId}:${row.sourceKey}:${JSON.stringify(row.identity)}`; + } + + #assertProjection( + scope: OfflineScope, + projection: { putRows?: readonly OfflineReplicaRow[]; removeRows?: readonly OfflineReplicaRowKey[] } | undefined, + ): void { + for (const row of [...(projection?.putRows ?? []), ...(projection?.removeRows ?? [])]) { + const schema = this.#entitySchema(row.sourceKey); + if (schema.identity.kind !== 'localOnly') { + throw new Error(`Offline replica projector may only mutate localOnly source "${row.sourceKey}".`); + } + if (row.userId !== scope.userId || row.scopeId !== scope.scopeId || row.identity.kind !== 'local') { + throw new Error('Offline replica projector rows must use the current scope and local identity.'); + } + } + } + #assertPullPage(page: OfflineReplicaPullPage): void { if (typeof page.nextCursor !== 'string') { throw new Error('Offline replica pull page nextCursor must be a string.'); @@ -219,6 +271,9 @@ export class OfflineReplicaPullService { if (!Array.isArray(page.changes)) { throw new Error('Offline replica pull page changes must be an array.'); } + if (page.rebaselineRequired !== undefined && typeof page.rebaselineRequired !== 'boolean') { + throw new Error('Offline replica pull page rebaselineRequired must be a boolean when present.'); + } for (const [index, change] of page.changes.entries()) { this.#assertPullChange(change, index); } diff --git a/projects/kit/offline/src/lib/offline-replica-puller.ts b/projects/kit/offline/src/lib/offline-replica-puller.ts index 8439332..0df7027 100644 --- a/projects/kit/offline/src/lib/offline-replica-puller.ts +++ b/projects/kit/offline/src/lib/offline-replica-puller.ts @@ -1,5 +1,5 @@ import { InjectionToken } from '@angular/core'; -import type { OfflineScope } from './offline-repository'; +import type { OfflineReplicaRow, OfflineReplicaRowKey, OfflineScope } from './offline-repository'; import type { OfflineGeneratedRemoteId, OfflineNaturalKey, @@ -53,6 +53,24 @@ export interface OfflineReplicaPullPage { changes: readonly OfflineReplicaChange[]; nextCursor: string; hasMore: boolean; + /** The server can no longer continue this cursor and requires a confirmed-state snapshot rebuild. */ + rebaselineRequired?: boolean; +} + +/** Product projection derived from one collapsed authoritative pull page. */ +export interface OfflineReplicaPullProjection { + putRows?: readonly OfflineReplicaRow[]; + removeRows?: readonly OfflineReplicaRowKey[]; +} + +/** Pure product adapter for local-only projections derived from server replica changes. */ +export interface OfflineReplicaProjector { + project(input: { + scope: OfflineScope; + changes: readonly OfflineReplicaChange[]; + commands: readonly import('./offline-repository').OfflineCommand[]; + repository: import('./offline-repository').OfflineRepository; + }): Promise; } /** Backend response accepted by the shared pull-page normalizer. */ @@ -92,3 +110,6 @@ export interface OfflineReplicaPuller { /** DI token for the application-provided explicit replica pull transport. */ export const OFFLINE_REPLICA_PULLER = new InjectionToken('OFFLINE_REPLICA_PULLER'); + +/** Optional product adapter for local-only replica projections. */ +export const OFFLINE_REPLICA_PROJECTOR = new InjectionToken('OFFLINE_REPLICA_PROJECTOR'); From 2369bbe7af6c22771c82005efe6e7164d492cfba Mon Sep 17 00:00:00 2001 From: rdlabo Date: Wed, 12 Aug 2026 17:04:14 +0900 Subject: [PATCH 07/10] Keep rebaseline reset atomic with snapshot --- .../lib/offline-replica-pull.service.spec.ts | 70 +++++++++++++++++++ .../src/lib/offline-replica-pull.service.ts | 30 +++++--- 2 files changed, 92 insertions(+), 8 deletions(-) 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 8f64ba2..a59bb1d 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 @@ -246,6 +246,76 @@ describe('OfflineReplicaPullService', () => { await expect(repository.getReplicaCursor(scope)).resolves.toEqual({ ...scope, cursor: 'snapshot-1' }); }); + it('keeps the previous replica durable until the first rebaseline snapshot can commit atomically', async () => { + await repository.transactReplica({ + putRows: [ + { + ...scope, + sourceKey: 'test_items', + identity: { kind: 'generated', localId: 'stale', remoteId: 10 }, + values: { id: 10, title: 'Stale' }, + confirmedValues: { id: 10, title: 'Stale' }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'confirmed', + }, + ], + putCursors: [{ ...scope, cursor: 'expired' }], + }); + const transact = vi.spyOn(repository, 'transactReplica'); + transact.mockClear(); + pull + .mockResolvedValueOnce(page([], { nextCursor: '', hasMore: true, rebaselineRequired: true })) + .mockResolvedValueOnce(page([itemChange(42, 'Fresh')], { nextCursor: 'snapshot-1' })); + + await service.pull(scope); + + expect(pull.mock.calls.map(([request]) => request.cursor)).toEqual(['expired', '']); + expect(transact).toHaveBeenCalledTimes(1); + await expect(repository.getReplicaRowByRemoteId(scope, 'test_items', 10)).resolves.toBeNull(); + await expect(repository.getReplicaRowByRemoteId(scope, 'test_items', 42)).resolves.toMatchObject({ + values: { title: 'Fresh' }, + }); + await expect(repository.getReplicaCursor(scope)).resolves.toEqual({ ...scope, cursor: 'snapshot-1' }); + }); + + it('preserves the previous replica and cursor when the first rebaseline snapshot fails', async () => { + await repository.transactReplica({ + putRows: [ + { + ...scope, + sourceKey: 'test_items', + identity: { kind: 'generated', localId: 'stale', remoteId: 10 }, + values: { id: 10, title: 'Stale' }, + confirmedValues: { id: 10, title: 'Stale' }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'confirmed', + }, + ], + putCursors: [{ ...scope, cursor: 'expired' }], + }); + pull + .mockResolvedValueOnce(page([], { nextCursor: '', hasMore: true, rebaselineRequired: true })) + .mockRejectedValueOnce(new Error('snapshot unavailable')); + + await expect(service.pull(scope)).rejects.toThrow('snapshot unavailable'); + + await expect(repository.getReplicaRowByRemoteId(scope, 'test_items', 10)).resolves.toMatchObject({ + values: { title: 'Stale' }, + }); + await expect(repository.getReplicaCursor(scope)).resolves.toEqual({ ...scope, cursor: 'expired' }); + }); + + it('rejects a terminal rebaseline marker without clearing the previous replica', async () => { + await repository.transactReplica({ putCursors: [{ ...scope, cursor: 'expired' }] }); + pull.mockResolvedValueOnce(page([], { nextCursor: '', rebaselineRequired: true })); + + await expect(service.pull(scope)).rejects.toThrow('Offline replica rebaseline marker must lead to a snapshot page.'); + + await expect(repository.getReplicaCursor(scope)).resolves.toEqual({ ...scope, cursor: 'expired' }); + }); + it('rejects projector attempts to mutate synchronized base rows without advancing the cursor', async () => { await repository.transactReplica({ putCursors: [{ ...scope, cursor: 'cursor-v0' }] }); projector.project.mockResolvedValue({ 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 d3095c6..72b51e4 100644 --- a/projects/kit/offline/src/lib/offline-replica-pull.service.ts +++ b/projects/kit/offline/src/lib/offline-replica-pull.service.ts @@ -53,27 +53,37 @@ export class OfflineReplicaPullService { async pull(scope: OfflineScope): Promise { if (this.#options.mode === 'readCacheOnly') return; const schemaHash = await (this.#schemaHash ??= sha256OfflineReplicaSchema(this.#options.replicaSchema)); - let cursor = (await this.#repository.getReplicaCursor(scope))?.cursor ?? ''; + let persistedCursor = (await this.#repository.getReplicaCursor(scope))?.cursor ?? ''; + let requestCursor = persistedCursor; + let rebaselinePending = false; for (;;) { const page = await this.#puller.pull({ scope, - cursor, + cursor: requestCursor, schemaVersion: this.#options.replicaSchema.version, schemaHash, }); this.#assertPullPage(page); this.#assertHandshake(page.schemaVersion, page.schemaHash, schemaHash); - if (page.hasMore && page.nextCursor === cursor) { + if (page.hasMore && page.nextCursor === requestCursor) { throw new Error(`Offline replica pull cursor did not advance for scope ${scope.userId}:${scope.scopeId}.`); } - if (page.changes.length === 0 && !page.hasMore && page.nextCursor === cursor) { + if (page.changes.length === 0 && !page.hasMore && page.nextCursor === requestCursor && !rebaselinePending) { return; } + if (page.rebaselineRequired && page.changes.length === 0 && !page.hasMore) { + throw new Error('Offline replica rebaseline marker must lead to a snapshot page.'); + } + if (page.rebaselineRequired && page.changes.length === 0 && page.hasMore) { + rebaselinePending = true; + requestCursor = page.nextCursor; + continue; + } const applied = await this.#replicaMutations.run(async () => { const currentCursor = (await this.#repository.getReplicaCursor(scope))?.cursor ?? ''; - if (currentCursor !== cursor) return currentCursor; + 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 changes = this.#collapseChanges(page.changes); @@ -81,7 +91,7 @@ export class OfflineReplicaPullService { const removeRows: OfflineReplicaRowKey[] = []; const putCommands = new Map(); const removeCommandIds = new Set(); - if (page.rebaselineRequired) { + if (rebaselinePending || page.rebaselineRequired) { removeRows.push(...(await this.#confirmedRowsForRebaseline(scope, scopeCommands))); } @@ -217,10 +227,14 @@ export class OfflineReplicaPullService { return page.nextCursor; }); if (applied !== page.nextCursor) { - cursor = applied; + persistedCursor = applied; + requestCursor = applied; + rebaselinePending = false; continue; } - cursor = page.nextCursor; + persistedCursor = page.nextCursor; + requestCursor = page.nextCursor; + rebaselinePending = false; if (!page.hasMore) return; } } From a50c58a1030999850b841755da5b2498e18d5e01 Mon Sep 17 00:00:00 2001 From: rdlabo Date: Wed, 12 Aug 2026 17:15:36 +0900 Subject: [PATCH 08/10] fix(offline): preserve rebaseline projections atomically --- .../offline/src/lib/offline-contract.spec.ts | 9 +- .../lib/offline-replica-pull.service.spec.ts | 109 ++++++++++++++++++ .../src/lib/offline-replica-pull.service.ts | 14 ++- .../offline/src/lib/offline-replica-puller.ts | 2 + 4 files changed, 126 insertions(+), 8 deletions(-) diff --git a/projects/kit/offline/src/lib/offline-contract.spec.ts b/projects/kit/offline/src/lib/offline-contract.spec.ts index 1493022..2d71920 100644 --- a/projects/kit/offline/src/lib/offline-contract.spec.ts +++ b/projects/kit/offline/src/lib/offline-contract.spec.ts @@ -7,8 +7,7 @@ import { offlineSessionManifestAllows } from './offline-session.service'; describe('shared offline boundary contracts', () => { it('normalizes database serverId exactly once at the pull transport boundary', () => { - expect( - normalizeOfflineReplicaPullPage({ + const normalized = normalizeOfflineReplicaPullPage({ schemaVersion: 1, schemaHash: 'hash', changes: [ @@ -29,8 +28,9 @@ describe('shared offline boundary contracts', () => { ], nextCursor: '2', hasMore: false, - }).changes, - ).toEqual([ + rebaselineRequired: true, + }); + expect(normalized.changes).toEqual([ { sourceKey: 'items', remoteId: 42, @@ -46,6 +46,7 @@ describe('shared offline boundary contracts', () => { deleted: true, }, ]); + expect(normalized.rebaselineRequired).toBe(true); }); it('narrows positive database ids without accepting coercion or zero', () => { 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 a59bb1d..4aea6da 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 @@ -246,6 +246,115 @@ describe('OfflineReplicaPullService', () => { await expect(repository.getReplicaCursor(scope)).resolves.toEqual({ ...scope, cursor: 'snapshot-1' }); }); + it('keeps snapshot and derived rows whose keys overlap the rebaseline removal set', async () => { + await repository.transactReplica({ + putRows: [ + { + ...scope, + sourceKey: 'test_items', + identity: { kind: 'generated', localId: 'existing', remoteId: 42 }, + values: { id: 42, title: 'Stale' }, + confirmedValues: { id: 42, title: 'Stale' }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'confirmed', + }, + { + ...scope, + sourceKey: 'test_views', + identity: { kind: 'local', localId: 'view-42' }, + values: { title: 'Stale derived' }, + confirmedValues: { title: 'Stale derived' }, + serverRevision: null, + fetchedAt: 1, + syncState: 'confirmed', + }, + ], + putCursors: [{ ...scope, cursor: 'expired' }], + }); + projector.project.mockResolvedValue({ + putRows: [ + { + ...scope, + sourceKey: 'test_views', + identity: { kind: 'local', localId: 'view-42' }, + values: { title: 'Fresh derived' }, + confirmedValues: { title: 'Fresh derived' }, + serverRevision: null, + fetchedAt: 2, + syncState: 'confirmed', + }, + ], + }); + pull.mockResolvedValueOnce(page([itemChange(42, 'Fresh')], { nextCursor: 'snapshot-1', rebaselineRequired: true })); + + await service.pull(scope); + + await expect(repository.getReplicaRowByRemoteId(scope, 'test_items', 42)).resolves.toMatchObject({ + values: { title: 'Fresh' }, + }); + await expect(repository.getReplicaRow(scope, 'test_views', { kind: 'local', localId: 'view-42' })).resolves.toMatchObject({ + values: { title: 'Fresh derived' }, + }); + }); + + it('preserves pending companions by canonical identity across user-scoped partitions', async () => { + const otherScope = { userId: scope.userId, scopeId: '20' }; + const companion: OfflineReplicaRow = { + ...scope, + sourceKey: 'test_items', + identity: { kind: 'generated', localId: 'shared', remoteId: 42 }, + values: { id: 42, title: 'Optimistic' }, + confirmedValues: { id: 42, title: 'Confirmed' }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'confirmed', + }; + await repository.transactReplica({ + putRows: [companion], + putCommands: [ + { + ...otherScope, + commandId: 'other-scope-command', + aggregateType: 'test_items', + sourceKey: 'test_items', + identity: { kind: 'generated', localId: 'other' }, + operation: 'test_items.update', + payload: {}, + optimisticValue: {}, + optimisticCompanions: [ + { + key: { + ...scope, + sourceKey: 'test_items', + identity: { kind: 'generated', localId: 'shared', remoteId: null }, + }, + before: companion, + after: companion, + }, + ], + payloadHash: 'hash', + baseRevision: 1, + state: 'pending', + attempts: 0, + retryAt: null, + createdAt: 1, + lastErrorCode: null, + }, + ], + putCursors: [{ ...scope, cursor: 'expired' }], + }); + pull + .mockResolvedValueOnce(page([], { nextCursor: 'snapshot-1', rebaselineRequired: true, hasMore: true })) + .mockResolvedValueOnce(page([], { nextCursor: 'snapshot-1' })); + + await service.pull(scope); + + await expect(repository.getReplicaRowByRemoteId(scope, 'test_items', 42)).resolves.toMatchObject({ + values: { title: 'Optimistic' }, + }); + }); + it('keeps the previous replica durable until the first rebaseline snapshot can commit atomically', 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 72b51e4..d3c0316 100644 --- a/projects/kit/offline/src/lib/offline-replica-pull.service.ts +++ b/projects/kit/offline/src/lib/offline-replica-pull.service.ts @@ -9,7 +9,9 @@ import { type OfflineReplicaPullPage, } from './offline-replica-puller'; import { + canonicalOfflinePrincipalId, canonicalOfflineCommandIdentity, + canonicalOfflineReplicaIdentity, commandIdentityFromReplicaIdentity, commandIdentityMatchesReplicaRow, offlineGeneratedReplicaIdentity, @@ -92,7 +94,7 @@ export class OfflineReplicaPullService { const putCommands = new Map(); const removeCommandIds = new Set(); if (rebaselinePending || page.rebaselineRequired) { - removeRows.push(...(await this.#confirmedRowsForRebaseline(scope, scopeCommands))); + removeRows.push(...(await this.#confirmedRowsForRebaseline(scope, userCommands))); } for (const change of changes) { @@ -217,9 +219,13 @@ export class OfflineReplicaPullService { repository: this.#repository, }); this.#assertProjection(scope, projection); + const finalPutRows = [...putRows, ...(projection?.putRows ?? [])]; + const finalPutKeys = new Set(finalPutRows.map((row) => this.#rowKey(row))); await this.#repository.transactReplica({ - putRows: [...putRows, ...(projection?.putRows ?? [])], - removeRows: [...removeRows, ...(projection?.removeRows ?? [])], + putRows: finalPutRows, + removeRows: [...removeRows, ...(projection?.removeRows ?? [])].filter( + (row) => !finalPutKeys.has(this.#rowKey(row)), + ), putCommands: [...putCommands.values()], removeCommandIds: [...removeCommandIds], putCursors: [{ ...scope, cursor: page.nextCursor }], @@ -257,7 +263,7 @@ export class OfflineReplicaPullService { } #rowKey(row: OfflineReplicaRowKey): string { - return `${row.userId}:${row.scopeId}:${row.sourceKey}:${JSON.stringify(row.identity)}`; + return `${canonicalOfflinePrincipalId(row.userId)}:${row.scopeId}:${row.sourceKey}:${canonicalOfflineReplicaIdentity(row.identity)}`; } #assertProjection( diff --git a/projects/kit/offline/src/lib/offline-replica-puller.ts b/projects/kit/offline/src/lib/offline-replica-puller.ts index 0df7027..591da8d 100644 --- a/projects/kit/offline/src/lib/offline-replica-puller.ts +++ b/projects/kit/offline/src/lib/offline-replica-puller.ts @@ -80,6 +80,8 @@ export interface OfflineReplicaWirePullPage { changes: readonly OfflineReplicaWireChange[]; nextCursor: string; hasMore: boolean; + /** The server can no longer continue this cursor and requires a confirmed-state snapshot rebuild. */ + rebaselineRequired?: boolean; } /** From 1c90aad28aa9e49415748526801baf218dd84076 Mon Sep 17 00:00:00 2001 From: rdlabo Date: Wed, 12 Aug 2026 17:22:59 +0900 Subject: [PATCH 09/10] refactor(offline): share canonical pull row identity --- projects/kit/offline/src/lib/offline-replica-pull.service.ts | 4 ---- 1 file changed, 4 deletions(-) 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 233f93d..a65b550 100644 --- a/projects/kit/offline/src/lib/offline-replica-pull.service.ts +++ b/projects/kit/offline/src/lib/offline-replica-pull.service.ts @@ -359,10 +359,6 @@ export class OfflineReplicaPullService { } } - #rowKey(key: OfflineReplicaRowKey): string { - return `${key.userId}:${key.scopeId}:${key.sourceKey}:${JSON.stringify(key.identity)}`; - } - async #currentCompanionRows(commands: readonly OfflineCommand[]): Promise { const keys = new Map( commands.flatMap((command) => From 15bc1062030d92a075f3e727dfd5439f92bec37f Mon Sep 17 00:00:00 2001 From: rdlabo Date: Wed, 12 Aug 2026 17:57:54 +0900 Subject: [PATCH 10/10] fix(offline): normalize pull row mutations --- .../lib/offline-replica-pull.service.spec.ts | 268 ++++++++++-------- .../src/lib/offline-replica-pull.service.ts | 86 +++--- .../kit/offline/src/lib/offline-repository.ts | 12 +- .../src/lib/sqlite-offline-repository.ts | 3 +- 4 files changed, 201 insertions(+), 168 deletions(-) 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 dcddce8..ae99880 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 @@ -230,9 +230,7 @@ describe('OfflineReplicaPullService', () => { ], }); const transact = vi.spyOn(repository, 'transactReplica'); - pull.mockResolvedValueOnce( - page([itemChange(42, 'Fresh')], { nextCursor: 'snapshot-1', rebaselineRequired: true }), - ); + pull.mockResolvedValueOnce(page([itemChange(42, 'Fresh')], { nextCursor: 'snapshot-1', rebaselineRequired: true })); await service.pull(scope); @@ -326,7 +324,7 @@ describe('OfflineReplicaPullService', () => { optimisticCompanions: [ { key: { - ...scope, + ...otherScope, sourceKey: 'test_items', identity: { kind: 'generated', localId: 'shared', remoteId: null }, }, @@ -444,9 +442,7 @@ describe('OfflineReplicaPullService', () => { }); pull.mockResolvedValueOnce(page([itemChange(42, 'Fresh')], { nextCursor: 'cursor-v1' })); - await expect(service.pull(scope)).rejects.toThrow( - 'Offline replica projector may only mutate localOnly source "test_items".', - ); + await expect(service.pull(scope)).rejects.toThrow('Offline replica projector may only mutate localOnly source "test_items".'); await expect(repository.getReplicaCursor(scope)).resolves.toEqual({ ...scope, cursor: 'cursor-v0' }); }); @@ -918,152 +914,175 @@ 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) => ({ - 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' }, - 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' }, - optimisticCompanions: [{ - key: { ...scope, sourceKey: 'test_views', identity: { kind: 'local', localId: 'view-42' } }, - before: null, - after: null, - }], - 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', - payload: { delta: 1 }, - payloadHash: 'hash', - 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.each(['conflict', 'rejected', 'blocked_auth'] as const)( - '%s commandは自動rebaseせずattention stateを維持する', - async (state) => { + it.each(['stale put', 'remove'] as const)( + 'rebase overlay wins over projector %s for the same canonical row', + async (projectionMutation) => { const executor = TestBed.inject(OFFLINE_COMMAND_EXECUTOR); - const rebase = vi.spyOn(executor, 'rebasePendingCommands'); + const otherScope = { userId: scope.userId, scopeId: '20' }; + const oldView: OfflineReplicaRow = { + ...scope, + sourceKey: 'test_views', + identity: { kind: 'local', localId: 'view-42' }, + values: { title: 'Old optimistic view' }, + confirmedValues: { title: 'Old confirmed view' }, + serverRevision: null, + fetchedAt: 1, + syncState: 'pending', + }; + const rebase = vi.spyOn(executor, 'rebasePendingCommands').mockImplementation((commands, confirmed, revision) => ({ + steps: commands.map(() => ({ + optimisticValue: { ...(confirmed as Record), title: 'Rebased delta' }, + optimisticCompanions: [ + { + key: { + ...otherScope, + sourceKey: 'test_views', + identity: { kind: 'local', localId: 'view-42' }, + }, + before: null, + after: { + ...otherScope, + 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-${state}`, remoteId: 42 }, - values: { id: 42, title: 'Local' }, - confirmedValues: { id: 42, title: 'Old' }, + 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: state, + syncState: 'pending', }, + oldView, ], putCommands: [ { ...scope, - commandId: `cmd-${state}`, + commandId: 'cmd-rebase', aggregateType: 'test_items', sourceKey: 'test_items', - identity: { kind: 'generated', localId: `019d-${state}` }, + identity: { kind: 'generated', localId: '019d-rebase' }, operation: 'test_items.delta', payload: { delta: 1 }, - optimisticValue: { id: 42, title: 'Local' }, + optimisticValue: { id: 42, title: 'Local delta' }, + optimisticCompanions: [ + { + key: { ...otherScope, sourceKey: 'test_views', identity: { kind: 'local', localId: 'view-42' } }, + before: null, + after: oldView, + }, + ], payloadHash: 'hash', baseRevision: 1, - state, - attempts: 1, + state: 'pending', + attempts: 0, retryAt: null, createdAt: 1, - lastErrorCode: state, + lastErrorCode: null, }, ], }); - pull.mockResolvedValueOnce(page([itemChange(42, 'Remote', { serverRevision: 9 })], { nextCursor: 'cursor-v1' })); + projector.project.mockResolvedValue( + projectionMutation === 'stale put' + ? { putRows: [oldView] } + : { removeRows: [{ ...scope, sourceKey: 'test_views', identity: { kind: 'local', localId: 'view-42' } }] }, + ); + pull.mockResolvedValueOnce(page([itemChange(42, 'Remote truth', { serverRevision: 9 })], { nextCursor: 'cursor-v1' })); await service.pull(scope); - expect(rebase).not.toHaveBeenCalled(); + expect(rebase).toHaveBeenCalledWith( + [expect.objectContaining({ commandId: 'cmd-rebase' })], + expect.objectContaining({ title: 'Remote truth' }), + 9, + [expect.objectContaining({ values: { title: 'Old optimistic view' } })], + ); + 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-${state}`, state: 'conflict', lastErrorCode: 'remote_revision' }), + expect.objectContaining({ + commandId: 'cmd-rebase', + payload: { delta: 1 }, + payloadHash: 'hash', + baseRevision: 9, + state: 'pending', + lastErrorCode: null, + }), ]); - await expect(repository.getReplicaRow(scope, 'test_items', generatedCommandIdentity(`019d-${state}`))).resolves.toMatchObject({ - syncState: 'conflict', + await expect(repository.getReplicaRow(scope, 'test_views', { kind: 'local', localId: 'view-42' })).resolves.toMatchObject({ + values: { title: 'Rebased view' }, + confirmedValues: { title: 'Remote view' }, }); + await expect(repository.getReplicaCursor(scope)).resolves.toEqual({ ...scope, cursor: 'cursor-v1' }); }, ); + 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 = { @@ -1118,12 +1137,9 @@ describe('OfflineReplicaPullService', () => { await service.pull(scope); - expect(rebase).toHaveBeenCalledWith( - expect.any(Array), - expect.anything(), - 9, - [expect.objectContaining({ scopeId: '20', visibility: 'pending_delete' })], - ); + 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 () => { 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 a65b550..7e23ed7 100644 --- a/projects/kit/offline/src/lib/offline-replica-pull.service.ts +++ b/projects/kit/offline/src/lib/offline-replica-pull.service.ts @@ -9,9 +9,7 @@ import { type OfflineReplicaPullPage, } from './offline-replica-puller'; import { - canonicalOfflinePrincipalId, canonicalOfflineCommandIdentity, - canonicalOfflineReplicaIdentity, commandIdentityFromReplicaIdentity, commandIdentityMatchesReplicaRow, offlineGeneratedReplicaIdentity, @@ -28,6 +26,7 @@ import { } from './offline-replica-schema'; import { OFFLINE_REPOSITORY, + canonicalOfflineReplicaRowKey, type OfflineCommand, type OfflineReplicaRow, type OfflineReplicaRowKey, @@ -91,6 +90,8 @@ export class OfflineReplicaPullService { const changes = this.#collapseChanges(page.changes); const putRows: OfflineReplicaRow[] = []; const removeRows: OfflineReplicaRowKey[] = []; + const rebasedPutRows: OfflineReplicaRow[] = []; + const rebasedRemoveRows: OfflineReplicaRowKey[] = []; const putCommands = new Map(); const removeCommandIds = new Set(); if (rebaselinePending || page.rebaselineRequired) { @@ -193,17 +194,17 @@ export class OfflineReplicaPullService { const revisionChanged = related.some((command) => command.baseRevision !== change.serverRevision); const rebaseEligible = related.every( - (command) => - (command.state === 'pending' || command.state === 'retry_wait') && command.serverCommitUnknown !== true, + (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(related), - )) ?? null) - : null; + const rebase = + revisionChanged && rebaseEligible + ? ((await this.#executor.rebasePendingCommands?.( + related, + confirmedValues, + change.serverRevision, + await this.#currentCompanionRows(related), + )) ?? null) + : null; this.#assertRebasedSteps(related, rebase?.steps ?? null); const conflicted = revisionChanged && rebase === null; const rebasedCommands = rebase @@ -217,7 +218,7 @@ export class OfflineReplicaPullService { })) : []; for (const command of rebasedCommands) putCommands.set(command.commandId, command); - this.#appendRebasedCompanionRows(rebasedCommands, putRows, removeRows); + this.#appendRebasedCompanionRows(rebasedCommands, rebasedPutRows, rebasedRemoveRows); putRows.push({ ...existing, values: rebasedCommands.at(-1)?.optimisticValue ?? (hasPending ? existing.values : confirmedValues), @@ -245,13 +246,14 @@ export class OfflineReplicaPullService { repository: this.#repository, }); this.#assertProjection(scope, projection); - const finalPutRows = [...putRows, ...(projection?.putRows ?? [])]; - const finalPutKeys = new Set(finalPutRows.map((row) => this.#rowKey(row))); + const finalRows = this.#mergeRowMutations([ + { putRows, removeRows }, + { putRows: projection?.putRows ?? [], removeRows: projection?.removeRows ?? [] }, + { putRows: rebasedPutRows, removeRows: rebasedRemoveRows }, + ]); await this.#repository.transactReplica({ - putRows: finalPutRows, - removeRows: [...removeRows, ...(projection?.removeRows ?? [])].filter( - (row) => !finalPutKeys.has(this.#rowKey(row)), - ), + putRows: finalRows.putRows, + removeRows: finalRows.removeRows, putCommands: [...putCommands.values()], removeCommandIds: [...removeCommandIds], putCursors: [{ ...scope, cursor: page.nextCursor }], @@ -271,25 +273,38 @@ export class OfflineReplicaPullService { } } - async #confirmedRowsForRebaseline( - scope: OfflineScope, - commands: readonly OfflineCommand[], - ): Promise { + async #confirmedRowsForRebaseline(scope: OfflineScope, commands: readonly OfflineCommand[]): Promise { const preserved = new Set( - commands.flatMap((command) => - (command.optimisticCompanions ?? []).map((companion) => this.#rowKey(companion.key)), - ), + commands.flatMap((command) => (command.optimisticCompanions ?? []).map((companion) => this.#rowKey(companion.key))), ); const rows = ( - await Promise.all( - this.#options.replicaSchema.entities.map((entity) => this.#repository.getReplicaRows(scope, entity.sourceKey)), - ) + await Promise.all(this.#options.replicaSchema.entities.map((entity) => this.#repository.getReplicaRows(scope, entity.sourceKey))) ).flat(); return rows.filter((row) => row.syncState === 'confirmed' && !preserved.has(this.#rowKey(row))); } #rowKey(row: OfflineReplicaRowKey): string { - return `${canonicalOfflinePrincipalId(row.userId)}:${row.scopeId}:${row.sourceKey}:${canonicalOfflineReplicaIdentity(row.identity)}`; + return canonicalOfflineReplicaRowKey(this.#entitySchema(row.sourceKey), row); + } + + #mergeRowMutations( + layers: readonly { + putRows: readonly OfflineReplicaRow[]; + removeRows: readonly OfflineReplicaRowKey[]; + }[], + ): { putRows: OfflineReplicaRow[]; removeRows: OfflineReplicaRowKey[] } { + const mutations = new Map(); + for (const layer of layers) { + for (const row of layer.removeRows) mutations.set(this.#rowKey(row), { kind: 'remove', row }); + for (const row of layer.putRows) mutations.set(this.#rowKey(row), { kind: 'put', row }); + } + const putRows: OfflineReplicaRow[] = []; + const removeRows: OfflineReplicaRowKey[] = []; + for (const mutation of mutations.values()) { + if (mutation.kind === 'put') putRows.push(mutation.row); + else removeRows.push(mutation.row); + } + return { putRows, removeRows }; } #assertProjection( @@ -342,11 +357,7 @@ export class OfflineReplicaPullService { } } - #appendRebasedCompanionRows( - commands: readonly OfflineCommand[], - putRows: OfflineReplicaRow[], - removeRows: OfflineReplicaRowKey[], - ): void { + #appendRebasedCompanionRows(commands: readonly OfflineCommand[], putRows: OfflineReplicaRow[], removeRows: OfflineReplicaRowKey[]): void { const latest = new Map(); for (const command of commands) { for (const companion of command.optimisticCompanions ?? []) { @@ -362,10 +373,7 @@ export class OfflineReplicaPullService { async #currentCompanionRows(commands: readonly OfflineCommand[]): Promise { const keys = new Map( commands.flatMap((command) => - (command.optimisticCompanions ?? []).map((companion) => [ - this.#rowKey(companion.key), - companion.key, - ] as const), + (command.optimisticCompanions ?? []).map((companion) => [this.#rowKey(companion.key), companion.key] as const), ), ); const rows = await Promise.all( diff --git a/projects/kit/offline/src/lib/offline-repository.ts b/projects/kit/offline/src/lib/offline-repository.ts index f811bc9..8634059 100644 --- a/projects/kit/offline/src/lib/offline-repository.ts +++ b/projects/kit/offline/src/lib/offline-repository.ts @@ -120,6 +120,15 @@ export interface OfflineReplicaRowKey extends OfflineScope { identity: OfflineReplicaIdentity; } +/** Canonical physical key shared by every repository and replica transaction coordinator. */ +export function canonicalOfflineReplicaRowKey( + schema: Pick>, 'scope'>, + row: OfflineReplicaRowKey, +): string { + const partition = schema.scope === 'user' ? 'user' : String(row.scopeId); + return `${canonicalOfflinePrincipalId(row.userId)}:${partition}:${row.sourceKey}:${canonicalOfflineReplicaIdentity(row.identity)}`; +} + /** Explicit one-way release of a generated remote id during a replica delete acknowledgement. */ export interface OfflineReplicaRemoteIdRelease extends OfflineReplicaRowKey { /** The current remote id being released; the matching put row must set `remoteId` to null. */ @@ -820,8 +829,7 @@ export class IonicOfflineRepository implements OfflineRepository { #rowKey(row: OfflineScope & { sourceKey: string; identity: OfflineReplicaIdentity }): string { const schema = this.#resolveReplicaEntitySchema(row.sourceKey); - const partition = schema.scope === 'user' ? 'user' : String(row.scopeId); - return `${canonicalOfflinePrincipalId(row.userId)}:${partition}:${row.sourceKey}:${canonicalOfflineReplicaIdentity(row.identity)}`; + return canonicalOfflineReplicaRowKey(schema, row); } #findRowByAddress( diff --git a/projects/kit/offline/src/lib/sqlite-offline-repository.ts b/projects/kit/offline/src/lib/sqlite-offline-repository.ts index c3b4e95..6a87da0 100644 --- a/projects/kit/offline/src/lib/sqlite-offline-repository.ts +++ b/projects/kit/offline/src/lib/sqlite-offline-repository.ts @@ -28,6 +28,7 @@ import { } from './offline-replica-schema'; import { OFFLINE_SCHEMA_VERSION, + canonicalOfflineReplicaRowKey, type OfflineCommand, type OfflineCommandIdentity, type OfflinePrincipalId, @@ -905,7 +906,7 @@ export class SqliteOfflineRepository implements OfflineRepository { #replicaRowKey(row: OfflineReplicaRowKey): string { const schema = this.#resolveReplicaEntitySchema(row.sourceKey); - return `${canonicalOfflinePrincipalId(row.userId)}:${schema.scope === 'user' ? 'user' : row.scopeId}:${row.sourceKey}:${canonicalOfflineReplicaIdentity(row.identity)}`; + return canonicalOfflineReplicaRowKey(schema, row); } #assertReplicaIdentityAssignment(