From d901c5b1e3937bee64bc3c198b0960b3e53d0893 Mon Sep 17 00:00:00 2001 From: rdlabo Date: Tue, 11 Aug 2026 07:51:01 +0900 Subject: [PATCH] fix offline synchronization invariants --- .../src/lib/offline-repository.spec.ts | 42 ++ .../kit/offline/src/lib/offline-repository.ts | 37 +- .../src/lib/offline-sync.service.spec.ts | 412 +++++++++++++++++- .../offline/src/lib/offline-sync.service.ts | 244 +++++++++-- .../src/lib/sqlite-offline-repository.spec.ts | 19 +- .../src/lib/sqlite-offline-repository.ts | 50 ++- 6 files changed, 736 insertions(+), 68 deletions(-) diff --git a/projects/kit/offline/src/lib/offline-repository.spec.ts b/projects/kit/offline/src/lib/offline-repository.spec.ts index 6604691..ef7840f 100644 --- a/projects/kit/offline/src/lib/offline-repository.spec.ts +++ b/projects/kit/offline/src/lib/offline-repository.spec.ts @@ -842,6 +842,48 @@ describe('IonicOfflineRepository', () => { expect((await repository.getCommandsForUser!(1)).map((item) => item.commandId)).toEqual(['cmd-a', 'cmd-m', 'cmd-z']); }); + it('legacy web outboxの送信中と複数回試行済みの最終失敗をcommit不明として安全側へnormalizeする', async () => { + const base: OfflineCommand = { + userId: 1, + scopeId: '10', + commandId: 'legacy-pending', + aggregateType: 'test_items', + sourceKey: 'test_items', + identity: { kind: 'generated', localId: 'legacy' }, + operation: 'test_items.update', + payload: {}, + optimisticValue: {}, + payloadHash: 'hash', + baseRevision: null, + state: 'pending', + attempts: 0, + retryAt: null, + createdAt: 1, + lastErrorCode: null, + }; + storage.values.set('offline:outbox:commands', { + pending: base, + sending: { ...base, commandId: 'legacy-sending', state: 'sending' }, + retry: { ...base, commandId: 'legacy-retry', state: 'retry_wait' }, + conflict: { ...base, commandId: 'legacy-conflict', state: 'conflict', attempts: 2 }, + rejected: { ...base, commandId: 'legacy-rejected', state: 'rejected', attempts: 2 }, + firstRejected: { ...base, commandId: 'legacy-first-rejected', state: 'rejected', attempts: 1 }, + explicitSafe: { ...base, commandId: 'new-pretransport', state: 'retry_wait', serverCommitUnknown: false }, + }); + + const restored = await repository.getCommands({ userId: 1, scopeId: '10' }); + expect(restored).toEqual( + expect.arrayContaining([ + expect.objectContaining({ commandId: 'legacy-sending', serverCommitUnknown: true }), + expect.objectContaining({ commandId: 'legacy-retry', serverCommitUnknown: true }), + expect.objectContaining({ commandId: 'legacy-conflict', serverCommitUnknown: true }), + expect.objectContaining({ commandId: 'legacy-rejected', serverCommitUnknown: true }), + expect.objectContaining({ commandId: 'new-pretransport', serverCommitUnknown: false }), + ]), + ); + expect(restored.find(({ commandId }) => commandId === 'legacy-first-rejected')).not.toHaveProperty('serverCommitUnknown'); + }); + it('outboxを作成順で保持し、scope削除時もuser-scoped commandを保持する', async () => { const base: Omit = { userId: 1, diff --git a/projects/kit/offline/src/lib/offline-repository.ts b/projects/kit/offline/src/lib/offline-repository.ts index 6db18de..9f0bea7 100644 --- a/projects/kit/offline/src/lib/offline-repository.ts +++ b/projects/kit/offline/src/lib/offline-repository.ts @@ -82,6 +82,8 @@ interface OfflineCommandBase extends OfflineScope { retryAt: number | null; createdAt: number; lastErrorCode: string | null; + /** True when transport started but the client cannot prove whether the server committed. */ + serverCommitUnknown?: boolean; } /** Durable before/after image used to reconcile product-owned derived rows. */ @@ -141,6 +143,10 @@ export interface OfflineReplicaTransaction { putCommands?: readonly OfflineCommand[]; removeCommandIds?: readonly string[]; putCursors?: readonly OfflineReplicaCursor[]; + /** Scopes whose acknowledged server changes still require an authoritative pull. */ + putReconciliationScopes?: readonly OfflineScope[]; + /** Scopes whose authoritative post-acknowledgement pull completed successfully. */ + removeReconciliationScopes?: readonly OfflineScope[]; } /** Durable local replica and outbox persistence contract. */ @@ -173,6 +179,7 @@ export interface OfflineRepository { identity: OfflineReplicaRemoteIdentity, ): Promise | null>; getReplicaCursor(scope: OfflineScope): Promise; + getReconciliationScopes?(userId: OfflinePrincipalId): Promise; getCommands(scope: OfflineScope): Promise; getCommandsForUser?(userId: OfflinePrincipalId): Promise; putCommand(command: OfflineCommand): Promise; @@ -218,6 +225,7 @@ const CURSORS_KEY = 'offline:replica:cursors'; const OUTBOX_KEY = 'offline:outbox:commands'; const REPLICA_TRANSACTION_KEY = 'offline:replica:transaction'; const REPLICA_SCHEMA_MIGRATION_KEY = 'offline:replica:schema-migration'; +const RECONCILIATION_SCOPES_KEY = 'offline:replica:reconciliation-scopes'; function compareOfflineCommands(left: OfflineCommand, right: OfflineCommand): number { return left.createdAt - right.createdAt || (left.commandId < right.commandId ? -1 : left.commandId > right.commandId ? 1 : 0); @@ -344,12 +352,20 @@ export class IonicOfflineRepository implements OfflineRepository { return cursor === undefined ? null : { ...scope, cursor }; } + async getReconciliationScopes(userId: OfflinePrincipalId): Promise { + await this.initialize(); + await this.#writes; + const scopes = await this.#readRecord(RECONCILIATION_SCOPES_KEY); + return Object.values(scopes).filter((scope) => scope.userId === userId); + } + async getCommands(scope: OfflineScope): Promise { await this.initialize(); await this.#writes; const commands = await this.#readRecord(OUTBOX_KEY); return Object.values(commands) .filter((command) => command.userId === scope.userId && command.scopeId === scope.scopeId) + .map((command) => this.#normalizeCommand(command)) .sort(compareOfflineCommands); } @@ -359,9 +375,18 @@ export class IonicOfflineRepository implements OfflineRepository { const commands = await this.#readRecord(OUTBOX_KEY); return Object.values(commands) .filter((command) => command.userId === userId) + .map((command) => this.#normalizeCommand(command)) .sort(compareOfflineCommands); } + #normalizeCommand(command: OfflineCommand): OfflineCommand { + if (command.serverCommitUnknown !== undefined) return command; + const legacyAmbiguousFailure = command.attempts >= 2 && ['blocked_auth', 'conflict', 'rejected'].includes(command.state); + return command.state === 'sending' || command.state === 'retry_wait' || legacyAmbiguousFailure + ? { ...command, serverCommitUnknown: true } + : command; + } + async putCommand(command: OfflineCommand): Promise { await this.initialize(); await this.#assertReplicaSchemaLocked(); @@ -393,6 +418,7 @@ export class IonicOfflineRepository implements OfflineRepository { this.#filterRecord(ROWS_KEY, (value) => value.userId !== userId), this.#filterRecord(OUTBOX_KEY, (value) => value.userId !== userId), this.#filterRecord(CURSORS_KEY, (_value, key) => !key.startsWith(`${canonicalOfflinePrincipalId(userId)}:`)), + this.#filterRecord(RECONCILIATION_SCOPES_KEY, (value) => value.userId !== userId), ]); const metadata = await this.#metadata(); if (metadata.lastUserId === userId) { @@ -413,6 +439,7 @@ export class IonicOfflineRepository implements OfflineRepository { return schema.scope === 'user' || !belongsToGroup(value); }), this.#filterRecord(CURSORS_KEY, (_value, key) => key !== this.#cursorKey(scope)), + this.#filterRecord(RECONCILIATION_SCOPES_KEY, (value) => !belongsToGroup(value)), ]); } @@ -601,10 +628,11 @@ export class IonicOfflineRepository implements OfflineRepository { async #applyReplicaTransaction(transaction: OfflineReplicaTransaction, journal: boolean): Promise { await this.#assertReplicaSchemaLocked(); for (const row of transaction.putRows ?? []) this.#validateReplicaRow(row); - const [rows, commands, cursors] = await Promise.all([ + const [rows, commands, cursors, reconciliationScopes] = await Promise.all([ this.#readRecord(ROWS_KEY), this.#readRecord(OUTBOX_KEY), this.#readRecord(CURSORS_KEY), + this.#readRecord(RECONCILIATION_SCOPES_KEY), ]); const identityCheckRows = { ...rows }; const releases = new Map(); @@ -654,10 +682,17 @@ export class IonicOfflineRepository implements OfflineRepository { for (const cursor of transaction.putCursors ?? []) { cursors[this.#cursorKey(cursor)] = cursor.cursor; } + for (const scope of transaction.putReconciliationScopes ?? []) { + reconciliationScopes[this.#cursorKey(scope)] = scope; + } + for (const scope of transaction.removeReconciliationScopes ?? []) { + delete reconciliationScopes[this.#cursorKey(scope)]; + } await Promise.all([ this.#storage.set(ROWS_KEY, rows), this.#storage.set(OUTBOX_KEY, commands), this.#storage.set(CURSORS_KEY, cursors), + this.#storage.set(RECONCILIATION_SCOPES_KEY, reconciliationScopes), ]); await this.#storage.remove(REPLICA_TRANSACTION_KEY); } diff --git a/projects/kit/offline/src/lib/offline-sync.service.spec.ts b/projects/kit/offline/src/lib/offline-sync.service.spec.ts index 6e68b2a..2c1db8f 100644 --- a/projects/kit/offline/src/lib/offline-sync.service.spec.ts +++ b/projects/kit/offline/src/lib/offline-sync.service.spec.ts @@ -7,6 +7,7 @@ import { type OfflineCommandResult, type OfflineCommandTarget, } from './offline-command-executor'; +import { OFFLINE_COMMAND_HOOKS } from './offline-command-hooks'; import { OFFLINE_KIT_OPTIONS, type OfflineKitOptions } from './offline-kit-options'; import { OfflineNetworkService } from './offline-network.service'; import { OfflineReplicaPullService } from './offline-replica-pull.service'; @@ -23,7 +24,7 @@ import { type OfflineScope, } from './offline-repository'; import { generatedCommandIdentity } from './offline-test-helpers'; -import { OfflinePayloadValidationError, OfflineSyncService } from './offline-sync.service'; +import { OfflineCommandInFlightError, OfflinePayloadValidationError, OfflineSyncService } from './offline-sync.service'; const replicaSchema = defineOfflineReplicaSchema({ version: 1, @@ -72,6 +73,7 @@ describe('OfflineSyncService', () => { let service: OfflineSyncService; let commands: OfflineCommand[]; let rows: OfflineReplicaRow[]; + let reconciliationScopes: OfflineScope[]; let connected: ReturnType>; let session: { userId: number; scopes: OfflineScope[] } | null; let localSession: { userId: number; scopes: OfflineScope[] } | null | undefined; @@ -80,6 +82,7 @@ describe('OfflineSyncService', () => { let beforeGetReplicaRow: (() => Promise) | null; let pull: ReturnType Promise>>; let handleError: ReturnType void>>; + let onCommandRemoved: ReturnType Promise>>; let options: OfflineKitOptions; const execute = vi.fn( async (_command: OfflineCommand, _target: OfflineCommandTarget): Promise => ({ response: null }), @@ -88,6 +91,7 @@ describe('OfflineSyncService', () => { beforeEach(() => { commands = []; rows = []; + reconciliationScopes = []; connected = signal(false); session = { userId: 1, scopes: [{ userId: 1, scopeId: '10' }] }; localSession = undefined; @@ -96,6 +100,7 @@ describe('OfflineSyncService', () => { beforeGetReplicaRow = null; pull = vi.fn(async () => undefined); handleError = vi.fn(); + onCommandRemoved = vi.fn(async () => undefined); options = { databaseName: 'test-offline', replicaSchema }; execute.mockReset(); execute.mockResolvedValue({ response: null }); @@ -185,6 +190,9 @@ describe('OfflineSyncService', () => { ); }), getReplicaCursor: vi.fn(async () => null), + getReconciliationScopes: vi.fn(async (userId: number) => + reconciliationScopes.filter((scope) => scope.userId === userId).map((scope) => ({ ...scope })), + ), transactReplica: vi.fn(async (transaction) => { for (const row of transaction.putRows ?? []) { rows = rows.filter( @@ -210,6 +218,17 @@ describe('OfflineSyncService', () => { commands.push(structuredClone(command)); } commands = commands.filter((command) => !(transaction.removeCommandIds ?? []).includes(command.commandId)); + for (const scope of transaction.putReconciliationScopes ?? []) { + reconciliationScopes = reconciliationScopes.filter( + (candidate) => candidate.userId !== scope.userId || candidate.scopeId !== scope.scopeId, + ); + reconciliationScopes.push({ ...scope }); + } + for (const scope of transaction.removeReconciliationScopes ?? []) { + reconciliationScopes = reconciliationScopes.filter( + (candidate) => candidate.userId !== scope.userId || candidate.scopeId !== scope.scopeId, + ); + } commands.sort((left, right) => left.createdAt - right.createdAt); }), } as unknown as OfflineRepository; @@ -221,6 +240,10 @@ describe('OfflineSyncService', () => { { provide: OFFLINE_KIT_OPTIONS, useValue: options }, { provide: OfflineReplicaPullService, useValue: { pull } }, { provide: ErrorHandler, useValue: { handleError } }, + { + provide: OFFLINE_COMMAND_HOOKS, + useValue: { entityType: (command: OfflineCommand) => command.aggregateType, onCommandRemoved }, + }, { provide: OFFLINE_SYNC_CONTEXT, useValue: { @@ -644,6 +667,7 @@ describe('OfflineSyncService', () => { expect(execute.mock.calls.map(([command]) => (command as OfflineCommand<{ seq: number }>).payload.seq)).toEqual([1, 2]); expect(service.pendingCount()).toBe(0); expect(pull).toHaveBeenCalledTimes(2); + expect(onCommandRemoved).toHaveBeenCalledTimes(2); }); it('送信成功後は同一scopeの複数aggregateを一度だけ再pullする', async () => { @@ -759,6 +783,96 @@ describe('OfflineSyncService', () => { expect(service.pendingCount()).toBe(0); }); + it('partial flushのACK後pull失敗scopeをOutbox削除後もreconnectで再pullする', async () => { + session = { + userId: 1, + scopes: [ + { userId: 1, scopeId: '10' }, + { userId: 1, scopeId: '20' }, + ], + }; + let scope20Pulls = 0; + const postPullError = new Error('scope 20 post-send pull failed'); + pull.mockImplementation(async (scope) => { + if (scope.scopeId === '20' && ++scope20Pulls === 2) throw postPullError; + }); + await service.refreshSession(['10']); + await service.enqueue( + { + scopeId: '20', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'partial-post-pull-failure' }, + operation: 'documents.create', + payload: { title: 'one' }, + optimisticValue: { id: 0, title: 'one' }, + }, + { flush: false }, + ); + + connected.set(true); + await vi.waitFor(() => expect(handleError).toHaveBeenCalledWith(postPullError)); + expect(commands).toEqual([]); + expect(execute).toHaveBeenCalledOnce(); + + connected.set(false); + connected.set(true); + await vi.waitFor(() => expect(scope20Pulls).toBe(3)); + expect(execute).toHaveBeenCalledOnce(); + }); + + it('ACK後pull失敗scopeをreset後もdurable markerから復元しcommandを再送しない', async () => { + session = { + userId: 1, + scopes: [ + { userId: 1, scopeId: '10' }, + { userId: 1, scopeId: '20' }, + ], + }; + let scope20Pulls = 0; + const postPullError = new Error('scope 20 post-send pull failed before restart'); + pull.mockImplementation(async (scope) => { + if (scope.scopeId === '20' && ++scope20Pulls === 2) throw postPullError; + }); + await service.refreshSession(['10']); + await service.enqueue( + { + scopeId: '20', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'durable-post-pull-failure' }, + operation: 'documents.create', + payload: { title: 'one' }, + optimisticValue: { id: 0, title: 'one' }, + }, + { flush: false }, + ); + + connected.set(true); + await vi.waitFor(() => expect(handleError).toHaveBeenCalledWith(postPullError)); + expect(commands).toEqual([]); + expect(reconciliationScopes).toEqual([{ userId: 1, scopeId: '20' }]); + + connected.set(false); + await service.resetSession(); + await service.refreshSession(['10']); + connected.set(true); + await vi.waitFor(() => expect(scope20Pulls).toBe(3)); + + expect(execute).toHaveBeenCalledOnce(); + expect(reconciliationScopes).toEqual([]); + }); + + it('所属から外れたdurable reconciliation scopeをsession discoveryで破棄する', async () => { + reconciliationScopes = [{ userId: 1, scopeId: '20' }]; + session = { userId: 1, scopes: [{ userId: 1, scopeId: '10' }] }; + + await service.refreshSession(['10']); + expect(reconciliationScopes).toEqual([]); + + connected.set(true); + await service.flush(); + expect(pull).not.toHaveBeenCalledWith({ userId: 1, scopeId: '20' }); + }); + it('local_idを不変主キーにして送信直前に最新server_idへ解決する', async () => { execute.mockResolvedValueOnce({ remoteId: 38142, @@ -853,7 +967,7 @@ describe('OfflineSyncService', () => { await service.initialize(); session = { userId: 1, scopes: [{ userId: 1, scopeId: '10' }] }; await service.refreshSession(); - expect(service.pendingCommands()[0]?.state).toBe('pending'); + expect(service.pendingCommands()[0]).toMatchObject({ state: 'pending', serverCommitUnknown: true }); }); it('未同期createを破棄するとoutboxと未確定replica rowを同時に除く', async () => { @@ -1116,7 +1230,7 @@ describe('OfflineSyncService', () => { expect(commands[0]).toMatchObject({ replicaMutation: 'delete', state }); }); - it('flush中の一括discard後に旧commandを送信・復活させない', async () => { + it('executor送信中のdiscardAllPendingを拒否してserver結果の確定を待つ', async () => { let resolveExecute!: (value: { response: null; serverRevision?: number }) => void; execute.mockImplementationOnce(() => new Promise((resolve) => (resolveExecute = resolve))); await service.enqueue( @@ -1144,14 +1258,106 @@ describe('OfflineSyncService', () => { connected.set(true); const flush = service.flush(); await vi.waitFor(() => expect(execute).toHaveBeenCalledOnce()); - await service.discardAllPending(); + await expect(service.discardAllPending()).rejects.toBeInstanceOf(OfflineCommandInFlightError); + expect(commands[0]?.state).toBe('sending'); resolveExecute({ response: null, serverRevision: 2 }); await flush; - expect(execute).toHaveBeenCalledOnce(); + expect(execute).toHaveBeenCalledTimes(2); expect(commands).toEqual([]); expect(service.pendingCount()).toBe(0); }); + it('executor送信中のsingle discardを拒否してoptimistic rowとcommandを保持する', async () => { + let resolveExecute!: (value: { response: null }) => void; + execute.mockImplementationOnce(() => new Promise((resolve) => (resolveExecute = resolve))); + const commandId = await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'discard-in-flight' }, + operation: 'documents.upsert', + payload: {}, + optimisticValue: { title: 'pending' }, + }, + { flush: false }, + ); + connected.set(true); + const flush = service.flush(); + await vi.waitFor(() => expect(execute).toHaveBeenCalledOnce()); + + await expect(service.discard(commandId, { flush: false })).rejects.toBeInstanceOf(OfflineCommandInFlightError); + expect(commands).toEqual([expect.objectContaining({ commandId, state: 'sending' })]); + expect(rows).toEqual([expect.objectContaining({ values: { title: 'pending' } })]); + + resolveExecute({ response: null }); + await flush; + }); + + it('response-lossでretry_waitのcommandはserver commit不明のためdiscardを拒否する', async () => { + execute.mockRejectedValueOnce({ status: 0 }); + const commandId = await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'discard-response-loss' }, + operation: 'documents.upsert', + payload: {}, + optimisticValue: { title: 'pending' }, + }, + { flush: false }, + ); + connected.set(true); + await service.flush(); + expect(commands[0]?.state).toBe('retry_wait'); + + await expect(service.discard(commandId, { flush: false })).rejects.toBeInstanceOf(OfflineCommandInFlightError); + await expect(service.discardAllPending()).rejects.toBeInstanceOf(OfflineCommandInFlightError); + expect(commands).toEqual([expect.objectContaining({ commandId, state: 'retry_wait' })]); + + connected.set(false); + await service.retryNow(commandId); + expect(commands).toEqual([expect.objectContaining({ commandId, state: 'pending', serverCommitUnknown: true })]); + await expect(service.discard(commandId, { flush: false })).rejects.toBeInstanceOf(OfflineCommandInFlightError); + await expect(service.discardAllPending()).rejects.toBeInstanceOf(OfflineCommandInFlightError); + + execute.mockRejectedValueOnce({ status: 401 }); + connected.set(true); + await vi.waitFor(() => expect(service.pendingCommands()[0]?.state).toBe('blocked_auth')); + expect(service.pendingCommands()[0]).toMatchObject({ serverCommitUnknown: true }); + await expect(service.discard(commandId, { flush: false })).rejects.toBeInstanceOf(OfflineCommandInFlightError); + }); + + it.each([409, 422])('response-loss後にHTTP %sへ分類されても同じkeyで再確認してACKへ収束する', async (status) => { + execute.mockRejectedValueOnce({ status: 0 }).mockRejectedValueOnce({ status }).mockResolvedValueOnce({ response: null }); + const commandId = await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: `ambiguous-${status}` }, + operation: 'documents.upsert', + payload: {}, + optimisticValue: {}, + }, + { flush: false }, + ); + connected.set(true); + await service.flush(); + await service.retryNow(commandId); + + expect(service.pendingCommands()[0]).toMatchObject({ + commandId, + state: status === 409 ? 'conflict' : 'rejected', + serverCommitUnknown: true, + }); + await expect(service.discard(commandId, { flush: false })).rejects.toBeInstanceOf(OfflineCommandInFlightError); + + await service.retryNow(commandId); + + expect(execute).toHaveBeenCalledTimes(3); + expect(execute.mock.calls.map(([command]) => command.commandId)).toEqual([commandId, commandId, commandId]); + expect(service.pendingCommands()).toEqual([]); + }); + it('flush中のsession切替後に旧user commandを新sessionへ復活させない', async () => { let resolveExecute!: (value: { response: null; serverRevision?: number }) => void; execute.mockImplementationOnce(() => new Promise((resolve) => (resolveExecute = resolve))); @@ -1270,7 +1476,7 @@ describe('OfflineSyncService', () => { expect(service.pendingCount()).toBe(0); }); - it('local replica row lookup failureはrejectしbackground flushはErrorHandlerへ渡す', async () => { + it('local replica row lookup rejectionでもsendingに残さずretry_waitへ戻す', async () => { const repository = TestBed.inject(OFFLINE_REPOSITORY) as OfflineRepository; await service.enqueue( { @@ -1283,18 +1489,167 @@ describe('OfflineSyncService', () => { }, { flush: false }, ); - vi.mocked(repository.getReplicaRow).mockResolvedValue(null); - vi.mocked(repository.getReplicaRowIncludingPendingDelete!).mockResolvedValue(null); + const lookupError = new Error('replica lookup failed'); + vi.mocked(repository.getReplicaRow).mockRejectedValue(lookupError); + vi.mocked(repository.getReplicaRowIncludingPendingDelete!).mockRejectedValue(lookupError); connected.set(true); await service.refreshSession(); - await vi.waitFor(() => - expect(handleError).toHaveBeenCalledWith( - expect.objectContaining({ message: 'Offline replica row not found: documents/generated:1' }), - ), - ); + await vi.waitFor(() => expect(handleError).toHaveBeenCalledWith(lookupError)); await service.refreshSession(); - await expect(service.flush()).rejects.toThrow('Offline replica row not found'); + await expect(service.flush()).resolves.toBeUndefined(); + expect(service.pendingCommands()[0]).toMatchObject({ + state: 'retry_wait', + lastErrorCode: 'network', + serverCommitUnknown: false, + }); + expect(execute).not.toHaveBeenCalled(); + + vi.mocked(repository.getReplicaRow).mockResolvedValue(rows[0] ?? null); + vi.mocked(repository.getReplicaRowIncludingPendingDelete!).mockResolvedValue(rows[0] ?? null); + await service.discard(service.pendingCommands()[0]!.commandId, { flush: false }); + expect(service.pendingCount()).toBe(0); + expect(commands).toEqual([]); + }); + + it('pre-transport retry_waitはdiscardAllPendingで回復できる', async () => { + const commandId = await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'pretransport-discard-all' }, + operation: 'documents.upsert', + payload: {}, + optimisticValue: {}, + }, + { flush: false }, + ); + commands = commands.map((command) => + command.commandId === commandId + ? { ...command, state: 'retry_wait', retryAt: Date.now() + 1_000, serverCommitUnknown: false } + : command, + ); + await service.reloadPendingCommands(); + + await service.discardAllPending(); + expect(commands).toEqual([]); + expect(rows).toEqual([]); + expect(service.pendingCount()).toBe(0); + }); + + it('single discardのpostcommit hook失敗は報告のみでrepositoryとsignalを空へ収束させる', async () => { + const hookError = new Error('media cleanup failed'); + onCommandRemoved.mockRejectedValueOnce(hookError); + const commandId = await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'discard-hook-failure' }, + operation: 'documents.upsert', + payload: {}, + optimisticValue: {}, + }, + { flush: false }, + ); + + await expect(service.discard(commandId, { flush: false })).resolves.toBeUndefined(); + expect(commands).toEqual([]); + expect(service.pendingCount()).toBe(0); + await vi.waitFor(() => expect(handleError).toHaveBeenCalledWith(hookError)); + }); + + it('discardAllPendingのpostcommit hook失敗も報告のみでrepositoryとsignalを空へ収束させる', async () => { + const hookError = new Error('bulk media cleanup failed'); + onCommandRemoved.mockRejectedValueOnce(hookError); + await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'discard-all-hook-failure' }, + operation: 'documents.upsert', + payload: {}, + optimisticValue: {}, + }, + { flush: false }, + ); + + await expect(service.discardAllPending()).resolves.toBeUndefined(); + expect(commands).toEqual([]); + expect(service.pendingCount()).toBe(0); + await vi.waitFor(() => expect(handleError).toHaveBeenCalledWith(hookError)); + }); + + it('sending claimとcommand取消を同じmutation laneで直列化する', async () => { + await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'claim-race' }, + operation: 'documents.upsert', + payload: {}, + optimisticValue: {}, + }, + { flush: false }, + ); + let sendingStarted!: () => void; + const started = new Promise((resolve) => (sendingStarted = resolve)); + let releaseSending!: () => void; + const sendingBarrier = new Promise((resolve) => (releaseSending = resolve)); + beforePutCommand = async (command) => { + if (command.state !== 'sending') return; + sendingStarted(); + await sendingBarrier; + }; + + connected.set(true); + const flush = service.flush(); + await started; + let cancellationEntered = false; + const cancellation = service.runSerializedReplicaMutation(async (repository) => { + cancellationEntered = true; + const current = (await repository.getCommands({ userId: 1, scopeId: '10' })).find( + (command) => command.commandId === commands[0]?.commandId, + ); + expect(current?.state).toBe('sending'); + }); + await Promise.resolve(); + expect(cancellationEntered).toBe(false); + + releaseSending(); + await cancellation; + await flush; + expect(execute).toHaveBeenCalledOnce(); + }); + + it('取消が先にmutation laneを確保した場合はtransport claimがcommandを復活させない', async () => { + await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'cancel-before-claim' }, + operation: 'documents.upsert', + payload: {}, + optimisticValue: {}, + }, + { flush: false }, + ); + let cancellationStarted!: () => void; + const started = new Promise((resolve) => (cancellationStarted = resolve)); + let releaseCancellation!: () => void; + const cancellationBarrier = new Promise((resolve) => (releaseCancellation = resolve)); + const cancellation = service.runSerializedReplicaMutation(async (repository) => { + cancellationStarted(); + await cancellationBarrier; + await repository.transactReplica({ removeCommandIds: [commands[0]!.commandId] }); + }); + await started; + connected.set(true); + const flush = service.flush(); + releaseCancellation(); + await cancellation; + await flush; + + expect(commands).toEqual([]); expect(execute).not.toHaveBeenCalled(); }); @@ -1323,10 +1678,11 @@ describe('OfflineSyncService', () => { await vi.waitFor(() => expect(handleError).toHaveBeenCalledWith(expect.objectContaining({ message: 'transaction failed' }))); await service.refreshSession(); - await expect(service.flush()).rejects.toThrow('transaction failed'); + await expect(service.flush()).resolves.toBeUndefined(); + expect(service.pendingCommands()[0]).toMatchObject({ state: 'retry_wait', lastErrorCode: 'network' }); }); - it('executor error without integer statusはclassifyせずrejectする', async () => { + it('executor error without integer statusもsendingに残さずretry_waitへ戻す', async () => { execute.mockRejectedValueOnce(new Error('programming failure')); await service.enqueue( { @@ -1341,11 +1697,10 @@ describe('OfflineSyncService', () => { ); connected.set(true); await expect(service.flush()).rejects.toThrow('programming failure'); - expect(service.pendingCommands()[0]?.state).toBe('sending'); - expect(handleError).not.toHaveBeenCalled(); + expect(service.pendingCommands()[0]).toMatchObject({ state: 'retry_wait', lastErrorCode: 'network' }); }); - it('executor error with negative statusはclassifyせずrejectする', async () => { + it('executor error with negative statusもsendingに残さずretry_waitへ戻す', async () => { execute.mockRejectedValueOnce({ status: -1 }); await service.enqueue( { @@ -1360,7 +1715,7 @@ describe('OfflineSyncService', () => { ); connected.set(true); await expect(service.flush()).rejects.toThrow(); - expect(service.pendingCommands()[0]?.state).toBe('sending'); + expect(service.pendingCommands()[0]).toMatchObject({ state: 'retry_wait', lastErrorCode: 'network' }); }); it('invalid remoteIdはhard failする', async () => { @@ -2146,7 +2501,7 @@ describe('OfflineSyncService', () => { { flush: false }, ); execute.mockResolvedValueOnce({ removeReplica: true, response: null }); - execute.mockRejectedValueOnce({ status: 0 }); + execute.mockRejectedValueOnce({ status: 422 }); connected.set(true); await service.flush(); expect(rows[0]).toMatchObject({ confirmedValues: null, visibility: 'present' }); @@ -2362,6 +2717,7 @@ describe('OfflineSyncService', () => { TestBed.resetTestingModule(); commands = []; rows = []; + reconciliationScopes = []; connected = signal(false); session = multiScopeSession; beforePutCommand = null; @@ -2421,6 +2777,9 @@ describe('OfflineSyncService', () => { return row ? projectReplicaRow(row, scope) : null; }), getReplicaCursor: vi.fn(async () => null), + getReconciliationScopes: vi.fn(async (userId: number) => + reconciliationScopes.filter((scope) => scope.userId === userId).map((scope) => ({ ...scope })), + ), transactReplica: vi.fn(async (transaction) => { for (const row of transaction.putRows ?? []) { const existing = findReplicaRow(row, row.sourceKey, row.identity); @@ -2447,6 +2806,17 @@ describe('OfflineSyncService', () => { commands.push(structuredClone(command)); } commands = commands.filter((command) => !(transaction.removeCommandIds ?? []).includes(command.commandId)); + for (const scope of transaction.putReconciliationScopes ?? []) { + reconciliationScopes = reconciliationScopes.filter( + (candidate) => candidate.userId !== scope.userId || candidate.scopeId !== scope.scopeId, + ); + reconciliationScopes.push({ ...scope }); + } + for (const scope of transaction.removeReconciliationScopes ?? []) { + reconciliationScopes = reconciliationScopes.filter( + (candidate) => candidate.userId !== scope.userId || candidate.scopeId !== scope.scopeId, + ); + } commands.sort(compareCommands); }), } as unknown as OfflineRepository; diff --git a/projects/kit/offline/src/lib/offline-sync.service.ts b/projects/kit/offline/src/lib/offline-sync.service.ts index f527dca..0625af5 100644 --- a/projects/kit/offline/src/lib/offline-sync.service.ts +++ b/projects/kit/offline/src/lib/offline-sync.service.ts @@ -97,6 +97,14 @@ export class OfflineOutboxCapacityError extends Error { } } +/** Raised when a discard could race a request whose server commit is unknown. */ +export class OfflineCommandInFlightError extends Error { + constructor(readonly commandIds: readonly string[]) { + super('Offline commands are being synchronized and cannot be discarded until their server result is known'); + this.name = 'OfflineCommandInFlightError'; + } +} + const MAX_PARALLEL_AGGREGATES = 3; const MAX_BACKOFF_MS = 60 * 60 * 1000; const POST_SEND_PULL_RETRY_MS = 1_000; @@ -117,18 +125,21 @@ export class OfflineSyncService { readonly #errorHandler = inject(ErrorHandler); readonly #commands = signal([]); readonly #knownScopes = new Map(); + /** ACKed scopes whose authoritative post-send pull has not completed yet. */ + readonly #pendingPullScopes = new Map(); #activeUserId: OfflinePrincipalId | null = null; #flushPromise: Promise | null = null; #partialFlushInFlight = false; #chainedFullFlush: Promise | null = null; readonly #flushTransitions = new Set>(); #generation = 0; - readonly #sendingTransitions = new Set>(); + readonly #sendingTransitions = new Set>(); #retryTimer: ReturnType | null = null; /** When non-null, automatic flushes pull only foreground scopes plus Outbox scopes. */ #foregroundScopePolicy: readonly string[] | null = null; #initialized = false; #lastCommandCreatedAt = 0; + #coldReconciliationRequired = this.#repository.getReconciliationScopes === undefined; readonly pendingCommands = this.#commands.asReadonly(); readonly pendingCount = computed(() => this.pendingCommands().length); @@ -155,7 +166,13 @@ export class OfflineSyncService { await Promise.all( commands .filter((command) => command.state === 'sending') - .map((command) => this.#repository.putCommand({ ...command, state: 'pending' })), + .map((command) => + this.#repository.putCommand({ + ...command, + state: 'pending', + serverCommitUnknown: command.serverCommitUnknown ?? true, + }), + ), ); this.#initialized = true; await this.#refreshState(); @@ -195,6 +212,7 @@ export class OfflineSyncService { await this.#restoreInterruptedCommands(); this.#activeUserId = null; this.#knownScopes.clear(); + this.#pendingPullScopes.clear(); this.#commands.set([]); this.#scheduleRetry(null); } @@ -482,8 +500,6 @@ export class OfflineSyncService { async discard(commandId: string, options: { flush?: boolean } = {}): Promise { await this.initialize(); - this.#invalidateFlush(); - await this.#waitForSendingTransitions(); // A pull may have completed while transport was being cancelled. Re-read // and project the discard inside the same local mutation lane used by // enqueue, ACK, and pull application so an old before-image cannot replace @@ -491,13 +507,15 @@ export class OfflineSyncService { const command = await this.#replicaMutations.run(async () => { const current = (await this.#readKnownCommands()).find((item) => item.commandId === commandId); if (!current) return null; + this.#assertDiscardable([current]); + this.#invalidateFlush(); await this.#discardCommands([current]); return current; }); if (!command) return; - await this.#hooks.onCommandRemoved?.(command); await this.#restoreInterruptedCommands(); await this.#refreshState(); + await this.#hooks.onCommandRemoved?.(command).catch((error) => this.#reportError(error)); if (options.flush !== false && this.#network.connected()) this.#flushInBackground(); } @@ -513,10 +531,15 @@ export class OfflineSyncService { // resurrect such a command from an object captured before invalidation. const current = (await this.#readKnownCommands()).find((item) => item.commandId === commandId); if (!current) return false; - if (current.state !== 'retry_wait' && current.state !== 'blocked_auth') { + if (current.state !== 'retry_wait' && current.state !== 'blocked_auth' && current.serverCommitUnknown !== true) { throw new Error(`Offline command ${commandId} is not waiting for retry or reauthentication.`); } - await repository.putCommand({ ...current, state: 'pending', retryAt: null, lastErrorCode: null }); + await repository.putCommand({ + ...current, + state: 'pending', + retryAt: null, + lastErrorCode: null, + }); return true; }); if (retried && this.#network.connected()) await this.flush(); @@ -524,15 +547,22 @@ export class OfflineSyncService { async discardAllPending(): Promise { await this.initialize(); - this.#invalidateFlush(); - await this.#waitForSendingTransitions(); const commands = await this.#replicaMutations.run(async () => { const current = await this.#readKnownCommands(); + this.#assertDiscardable(current); + this.#invalidateFlush(); await this.#discardCommands(current); return current; }); - await Promise.all(commands.map((command) => this.#hooks.onCommandRemoved?.(command))); await this.#refreshState(); + await Promise.all(commands.map((command) => this.#hooks.onCommandRemoved?.(command).catch((error) => this.#reportError(error)))); + } + + #assertDiscardable(commands: readonly OfflineCommand[]): void { + const ambiguous = commands.filter((command) => command.state === 'sending' || command.serverCommitUnknown === true); + if (ambiguous.length > 0) { + throw new OfflineCommandInFlightError(ambiguous.map((command) => command.commandId)); + } } #flushInBackground(): void { @@ -544,7 +574,7 @@ export class OfflineSyncService { } #beginFlush(explicitFull: boolean): Promise { - const isPartial = !explicitFull && this.#foregroundScopePolicy !== null; + const isPartial = !explicitFull && !this.#coldReconciliationRequired && this.#foregroundScopePolicy !== null; if (this.#flushPromise) { if (explicitFull && this.#partialFlushInFlight) { if (!this.#chainedFullFlush) { @@ -595,7 +625,15 @@ export class OfflineSyncService { : [...this.#knownScopes.values()]; for (const scope of pullScopes) { if (!this.#isCurrent(generation) || !this.#network.connected()) return; - await this.#pull.pull(scope); + try { + await this.#pull.pull(scope); + await this.#markScopeReconciled(scope, generation); + } catch (error) { + if (this.#pendingPullScopes.has(this.#scopeKey(scope))) { + this.#scheduleRetry(Date.now() + POST_SEND_PULL_RETRY_MS); + } + throw error; + } } const dirtyScopes = new Map(); while (this.#network.connected() && this.#isCurrent(generation)) { @@ -611,14 +649,18 @@ export class OfflineSyncService { }); await Promise.all(workers); } - const postPullFailures: unknown[] = []; for (const scope of dirtyScopes.values()) { + this.#pendingPullScopes.set(this.#scopeKey(scope), scope); + } + const postPullFailures: unknown[] = []; + for (const scope of this.#pendingPullScopes.values()) { if (!this.#isCurrent(generation) || !this.#network.connected()) break; try { // A command response may contain only the aggregate's base row. Pull // once per dirty scope so sibling-table journal entries are visible // before the completed Outbox state is exposed to product UI. await this.#pull.pull(scope); + await this.#markScopeReconciled(scope, generation); } catch (error) { postPullFailures.push(error); } @@ -631,6 +673,7 @@ export class OfflineSyncService { this.#scheduleRetry(Date.now() + POST_SEND_PULL_RETRY_MS); throw postPullFailures[0]; } + if (this.#isCurrent(generation)) this.#coldReconciliationRequired = false; } #setForegroundScopePolicy(foregroundScopeIds?: readonly string[]): void { @@ -641,7 +684,10 @@ export class OfflineSyncService { const foregroundScopeSet = new Set(foregroundScopeIds); const outboxScopeKeys = new Set(commands.map((command) => this.#scopeKey({ userId: command.userId, scopeId: command.scopeId }))); return [...this.#knownScopes.values()].filter( - (scope) => foregroundScopeSet.has(scope.scopeId) || outboxScopeKeys.has(this.#scopeKey(scope)), + (scope) => + foregroundScopeSet.has(scope.scopeId) || + outboxScopeKeys.has(this.#scopeKey(scope)) || + this.#pendingPullScopes.has(this.#scopeKey(scope)), ); } @@ -665,37 +711,49 @@ export class OfflineSyncService { if (!this.#isCurrent(generation)) return; if (command.state === 'retry_wait' && (command.retryAt ?? 0) > Date.now()) break; if (!['pending', 'retry_wait'].includes(command.state)) break; - const sending: OfflineCommand = { - ...command, - state: 'sending', - attempts: command.attempts + 1, - retryAt: null, - lastErrorCode: null, - }; - await this.#putSendingCommand(sending); + let sending = await this.#claimSendingCommand(command, generation); + if (!sending) return; if (!this.#isCurrent(generation)) return; await this.#refreshState(generation); if (!this.#isCurrent(generation)) return; - const row = await this.#rowForCommand(sending); - if (!row) - throw new Error(`Offline replica row not found: ${sending.aggregateType}/${canonicalOfflineCommandIdentity(sending.identity)}`); + let row: OfflineReplicaRow | null; + try { + row = await this.#rowForCommand(sending); + } catch (error) { + if (!this.#isCurrent(generation)) return; + await this.#persistFailedCommand(sending, error, generation, null, sending.serverCommitUnknown === true); + throw error; + } + if (!row) { + const error = new Error( + `Offline replica row not found: ${sending.aggregateType}/${canonicalOfflineCommandIdentity(sending.identity)}`, + ); + await this.#persistFailedCommand(sending, error, generation, null, sending.serverCommitUnknown === true); + throw error; + } + const priorCommitUnknown = sending.serverCommitUnknown === true; + const transportCommand = await this.#markTransportStarted(sending, generation); + if (!transportCommand) return; + sending = transportCommand; let result: OfflineCommandResult; try { result = await this.#executor.execute(sending, offlineCommandTargetFromReplicaRow(row)); } catch (error) { if (!this.#isCurrent(generation)) return; + await this.#persistFailedCommand(sending, error, generation, row, priorCommitUnknown || this.#serverCommitCouldBeUnknown(error)); if (!this.#isClassifiableTransportError(error)) throw error; - const failed = this.#failedCommand(sending, error); - await this.#repository.transactReplica({ - putRows: [{ ...row, syncState: this.#replicaState(failed.state) }], - putCommands: [failed], - }); - if (failed.state === 'retry_wait') this.#scheduleRetry(failed.retryAt); break; } if (!this.#isCurrent(generation)) return; - await this.#completeCommand(commands, sending, result, generation); + try { + await this.#completeCommand(commands, sending, result, generation); + } catch (error) { + if (!this.#isCurrent(generation)) return; + await this.#persistFailedCommand(sending, error, generation, row, true); + throw error; + } if (this.#isCurrent(generation)) { + await this.#hooks.onCommandRemoved?.(sending).catch((error) => this.#reportError(error)); const scope = { userId: sending.userId, scopeId: sending.scopeId }; dirtyScopes.set(this.#scopeKey(scope), scope); } @@ -817,7 +875,10 @@ export class OfflineSyncService { removeRows: [...(removesReplica && rebased.length === 0 ? [current] : []), ...(companionTransaction.removeRows ?? [])], putCommands: rebased, removeCommandIds: [command.commandId], + putReconciliationScopes: [{ userId: command.userId, scopeId: command.scopeId }], }); + const scope = { userId: command.userId, scopeId: command.scopeId }; + this.#pendingPullScopes.set(this.#scopeKey(scope), scope); commands.splice(0, commands.length, ...latestCommands); } @@ -964,15 +1025,48 @@ export class OfflineSyncService { return `${canonicalOfflinePrincipalId(command.userId)}:${partition}:${sourceKey}:${canonicalOfflineCommandIdentity(command.identity)}`; } - #failedCommand(command: OfflineCommand, error: unknown): OfflineCommand { + #failedCommand(command: OfflineCommand, error: unknown, serverCommitUnknown: boolean): OfflineCommand { const status = this.#errorStatus(error); - if (status === 401 || status === 403) return { ...command, state: 'blocked_auth', lastErrorCode: String(status) }; - if (status === 409 || status === 412) return { ...command, state: 'conflict', lastErrorCode: String(status) }; + if (status === 401 || status === 403) { + return { ...command, state: 'blocked_auth', lastErrorCode: String(status), serverCommitUnknown }; + } + if (status === 409 || status === 412) { + return { ...command, state: 'conflict', lastErrorCode: String(status), serverCommitUnknown }; + } if (status >= 400 && status < 500 && status !== 429) { - return { ...command, state: 'rejected', lastErrorCode: String(status) }; + return { ...command, state: 'rejected', lastErrorCode: String(status), serverCommitUnknown }; } const retryAt = Date.now() + Math.min(MAX_BACKOFF_MS, 1000 * 2 ** Math.max(0, command.attempts - 1)); - return { ...command, state: 'retry_wait', retryAt, lastErrorCode: status > 0 ? String(status) : 'network' }; + return { + ...command, + state: 'retry_wait', + retryAt, + lastErrorCode: status > 0 ? String(status) : 'network', + serverCommitUnknown, + }; + } + + async #persistFailedCommand( + command: OfflineCommand, + error: unknown, + generation: number, + row?: OfflineReplicaRow | null, + serverCommitUnknown = true, + ): Promise { + const failed = this.#failedCommand(command, error, serverCommitUnknown); + const current = row === undefined ? await this.#rowForCommand(command) : row; + if (!this.#isCurrent(generation)) return; + if (current) { + await this.#repository.transactReplica({ + putRows: [{ ...current, syncState: this.#replicaState(failed.state) }], + putCommands: [failed], + }); + } else { + await this.#repository.putCommand(failed); + } + if (!this.#isCurrent(generation)) return; + if (failed.state === 'retry_wait') this.#scheduleRetry(failed.retryAt); + await this.#refreshState(generation); } #errorStatus(error: unknown): number { @@ -987,6 +1081,17 @@ export class OfflineSyncService { return typeof status === 'number' && Number.isInteger(status) && status >= 0; } + #serverCommitCouldBeUnknown(error: unknown): boolean { + const status = this.#errorStatus(error); + return status === 0 || status === 429 || status >= 500; + } + + #reportError(error: unknown): void { + void Promise.resolve() + .then(() => this.#errorHandler.handleError(error)) + .catch(() => undefined); + } + #assertServerRevision(revision: string | number | undefined): void { if (typeof revision === 'number' && !Number.isFinite(revision)) { throw new Error(`Offline command returned invalid serverRevision ${String(revision)}.`); @@ -1070,6 +1175,7 @@ export class OfflineSyncService { this.#setActiveUser(session.userId); this.#knownScopes.clear(); for (const scope of session.scopes) this.#knownScopes.set(this.#scopeKey(scope), scope); + await this.#restorePendingPullScopes(session.userId, generation); return true; } @@ -1085,6 +1191,7 @@ export class OfflineSyncService { this.#setActiveUser(session.userId); this.#knownScopes.clear(); for (const scope of session.scopes) this.#knownScopes.set(this.#scopeKey(scope), scope); + await this.#restorePendingPullScopes(session.userId, generation); return true; } @@ -1095,6 +1202,7 @@ export class OfflineSyncService { #setActiveUser(userId: OfflinePrincipalId): void { if (this.#activeUserId === userId) return; this.#knownScopes.clear(); + this.#pendingPullScopes.clear(); this.#activeUserId = userId; this.#lastCommandCreatedAt = 0; } @@ -1170,12 +1278,32 @@ export class OfflineSyncService { await Promise.all( commands .filter((command) => command.state === 'sending') - .map((command) => this.#repository.putCommand({ ...command, state: 'pending' })), + .map((command) => + this.#repository.putCommand({ + ...command, + state: 'pending', + serverCommitUnknown: command.serverCommitUnknown ?? true, + }), + ), ); } - #putSendingCommand(command: OfflineCommand): Promise { - const transition = this.#repository.putCommand(command); + #claimSendingCommand(command: OfflineCommand, generation: number): Promise { + const transition = this.#serializeReplicaMutation(async () => { + if (!this.#isCurrent(generation)) return null; + const scope = { userId: command.userId, scopeId: command.scopeId }; + const current = (await this.#repository.getCommands(scope)).find((candidate) => candidate.commandId === command.commandId); + if (!current || !['pending', 'retry_wait'].includes(current.state)) return null; + const sending: OfflineCommand = { + ...current, + state: 'sending', + attempts: current.attempts + 1, + retryAt: null, + lastErrorCode: null, + }; + await this.#repository.putCommand(sending); + return sending; + }); this.#sendingTransitions.add(transition); void transition.then( () => this.#sendingTransitions.delete(transition), @@ -1184,6 +1312,42 @@ export class OfflineSyncService { return transition; } + #markTransportStarted(command: OfflineCommand, generation: number): Promise { + return this.#serializeReplicaMutation(async () => { + if (!this.#isCurrent(generation)) return null; + const scope = { userId: command.userId, scopeId: command.scopeId }; + const current = (await this.#repository.getCommands(scope)).find((candidate) => candidate.commandId === command.commandId); + if (!current || current.state !== 'sending') return null; + const transportCommand = { ...current, serverCommitUnknown: true }; + await this.#repository.putCommand(transportCommand); + return transportCommand; + }); + } + + async #restorePendingPullScopes(userId: OfflinePrincipalId, generation: number): Promise { + if (!this.#repository.getReconciliationScopes) return; + const durableScopes = await this.#repository.getReconciliationScopes(userId); + if (!this.#isCurrent(generation) || this.#activeUserId !== userId) return; + const currentKeys = new Set(this.#knownScopes.keys()); + this.#pendingPullScopes.clear(); + const revoked: OfflineScope[] = []; + for (const scope of durableScopes) { + const key = this.#scopeKey(scope); + if (scope.userId === userId && currentKeys.has(key)) this.#pendingPullScopes.set(key, scope); + else revoked.push(scope); + } + if (revoked.length > 0) { + await this.#repository.transactReplica({ removeReconciliationScopes: revoked }); + } + } + + async #markScopeReconciled(scope: OfflineScope, generation: number): Promise { + if (!this.#isCurrent(generation)) return; + await this.#repository.transactReplica({ removeReconciliationScopes: [scope] }); + if (!this.#isCurrent(generation)) return; + this.#pendingPullScopes.delete(this.#scopeKey(scope)); + } + async #waitForSendingTransitions(): Promise { await Promise.allSettled([...this.#sendingTransitions]); } diff --git a/projects/kit/offline/src/lib/sqlite-offline-repository.spec.ts b/projects/kit/offline/src/lib/sqlite-offline-repository.spec.ts index 2b7730c..b320c04 100644 --- a/projects/kit/offline/src/lib/sqlite-offline-repository.spec.ts +++ b/projects/kit/offline/src/lib/sqlite-offline-repository.spec.ts @@ -261,6 +261,18 @@ describe('SqliteOfflineRepository community sqlite driver', () => { }); }); + it('legacy SQLite outboxの送信中と複数回試行済みの最終失敗をcolumn追加時にcommit不明へbackfillする', async () => { + const repository = createRepository(); + + await repository.initialize(); + + expect(plugin.execute).toHaveBeenCalledWith( + expect.objectContaining({ + statement: expect.stringContaining("OR (attempts >= 2 AND state IN ('blocked_auth', 'conflict', 'rejected'))"), + }), + ); + }); + it('暗号鍵の生成関数をcommunity driverへ渡す', async () => { const createEncryptionKey = vi.fn(async () => 'first-install-secret'); const repository = createRepository(createEncryptionKey); @@ -276,8 +288,11 @@ describe('SqliteOfflineRepository community sqlite driver', () => { const deletes = plugin.execute.mock.calls .map(([options]) => options as { statement: string; values?: unknown[] }) .filter(({ statement }) => statement.startsWith('DELETE FROM')); - expect(deletes).toHaveLength(1); - expect(deletes[0]?.values).toEqual([canonicalOfflinePrincipalId(7), '8']); + expect(deletes).toHaveLength(2); + expect(deletes.map(({ values }) => values)).toEqual([ + [canonicalOfflinePrincipalId(7), '8'], + [canonicalOfflinePrincipalId(7), '8'], + ]); expect(plugin.beginTransaction).toHaveBeenCalledOnce(); expect(plugin.commitTransaction).toHaveBeenCalledOnce(); expect(plugin.rollbackTransaction).not.toHaveBeenCalled(); diff --git a/projects/kit/offline/src/lib/sqlite-offline-repository.ts b/projects/kit/offline/src/lib/sqlite-offline-repository.ts index 4f37d97..c3b4e95 100644 --- a/projects/kit/offline/src/lib/sqlite-offline-repository.ts +++ b/projects/kit/offline/src/lib/sqlite-offline-repository.ts @@ -165,7 +165,8 @@ const SCHEMA = [ attempts INTEGER NOT NULL, retry_at INTEGER, created_at INTEGER NOT NULL, - last_error_code TEXT + last_error_code TEXT, + server_commit_unknown INTEGER NOT NULL DEFAULT 0 )`, `CREATE INDEX IF NOT EXISTS offline_sync_commands_scope_created ON offline_sync_commands (user_id, scope_id, created_at)`, @@ -179,6 +180,11 @@ const SCHEMA = [ scope_id TEXT NOT NULL, cursor TEXT NOT NULL, PRIMARY KEY (user_id, scope_id) + )`, + `CREATE TABLE IF NOT EXISTS offline_reconciliation_scopes ( + user_id TEXT NOT NULL, + scope_id TEXT NOT NULL, + PRIMARY KEY (user_id, scope_id) )`, ]; @@ -316,6 +322,13 @@ export class SqliteOfflineRepository implements OfflineRepository { return { ...scope, cursor: this.#string(row['cursor']) }; } + async getReconciliationScopes(userId: OfflinePrincipalId): Promise { + const rows = await this.#query('SELECT scope_id FROM offline_reconciliation_scopes WHERE user_id = ? ORDER BY scope_id', [ + canonicalOfflinePrincipalId(userId), + ]); + return rows.map((row) => ({ userId, scopeId: this.#string(row['scope_id']) })); + } + async getCommands(scope: OfflineScope): Promise { const rows = await this.#query( 'SELECT * FROM offline_sync_commands WHERE user_id = ? AND scope_id = ? ORDER BY created_at ASC, command_id ASC', @@ -349,6 +362,7 @@ export class SqliteOfflineRepository implements OfflineRepository { await this.#execute(database, 'DELETE FROM offline_session_manifests WHERE user_id = ?', [principal]); await this.#execute(database, 'DELETE FROM offline_sync_commands WHERE user_id = ?', [principal]); await this.#execute(database, 'DELETE FROM offline_replica_cursors WHERE user_id = ?', [principal]); + await this.#execute(database, 'DELETE FROM offline_reconciliation_scopes WHERE user_id = ?', [principal]); for (const entity of this.#options.replicaSchema.entities) { await this.#execute(database, `DELETE FROM ${entity.tableName} WHERE _offline_user_id = ?`, [principal]); } @@ -371,6 +385,7 @@ export class SqliteOfflineRepository implements OfflineRepository { ); } await this.#execute(database, 'DELETE FROM offline_replica_cursors WHERE user_id = ? AND scope_id = ?', values); + await this.#execute(database, 'DELETE FROM offline_reconciliation_scopes WHERE user_id = ? AND scope_id = ?', values); for (const entity of this.#options.replicaSchema.entities) { if (entity.scope !== 'partition') continue; await this.#execute(database, `DELETE FROM ${entity.tableName} WHERE _offline_user_id = ? AND _offline_scope_id = ?`, values); @@ -408,6 +423,20 @@ export class SqliteOfflineRepository implements OfflineRepository { await this.#execute(databaseId, 'DELETE FROM offline_sync_commands WHERE command_id = ?', [commandId]); } for (const cursor of transaction.putCursors ?? []) await this.#putReplicaCursor(databaseId, cursor); + for (const scope of transaction.putReconciliationScopes ?? []) { + await this.#execute( + databaseId, + `INSERT INTO offline_reconciliation_scopes (user_id, scope_id) VALUES (?, ?) + ON CONFLICT(user_id, scope_id) DO NOTHING`, + [canonicalOfflinePrincipalId(scope.userId), scope.scopeId], + ); + } + for (const scope of transaction.removeReconciliationScopes ?? []) { + await this.#execute(databaseId, 'DELETE FROM offline_reconciliation_scopes WHERE user_id = ? AND scope_id = ?', [ + canonicalOfflinePrincipalId(scope.userId), + scope.scopeId, + ]); + } }); } @@ -423,6 +452,15 @@ export class SqliteOfflineRepository implements OfflineRepository { if (!commandColumns.some((row) => row['name'] === 'optimistic_companions_json')) { await this.#execute(databaseId, 'ALTER TABLE offline_sync_commands ADD COLUMN optimistic_companions_json TEXT'); } + if (!commandColumns.some((row) => row['name'] === 'server_commit_unknown')) { + await this.#execute(databaseId, 'ALTER TABLE offline_sync_commands ADD COLUMN server_commit_unknown INTEGER NOT NULL DEFAULT 0'); + await this.#execute( + databaseId, + `UPDATE offline_sync_commands SET server_commit_unknown = 1 + WHERE state IN ('sending', 'retry_wait') + OR (attempts >= 2 AND state IN ('blocked_auth', 'conflict', 'rejected'))`, + ); + } const metadata = await this.#queryDatabase(databaseId, 'SELECT schema_version FROM offline_metadata WHERE id = 1'); if (metadata.length === 0) { await this.#execute(databaseId, 'INSERT INTO offline_metadata (id, schema_version, last_user_id) VALUES (1, ?, NULL)', [ @@ -573,6 +611,7 @@ export class SqliteOfflineRepository implements OfflineRepository { retryAt: this.#numberOrNull(row['retry_at']), createdAt: this.#number(row['created_at']), lastErrorCode: this.#stringOrNull(row['last_error_code']), + serverCommitUnknown: this.#numberOrNull(row['server_commit_unknown']) === 1, }; } @@ -581,8 +620,9 @@ export class SqliteOfflineRepository implements OfflineRepository { databaseId, `INSERT INTO offline_sync_commands (command_id, user_id, scope_id, aggregate_type, source_key, identity_json, operation, payload_json, optimistic_value_json, - optimistic_companions_json, replica_mutation, payload_hash, base_revision_json, state, attempts, retry_at, created_at, last_error_code) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + optimistic_companions_json, replica_mutation, payload_hash, base_revision_json, state, attempts, retry_at, created_at, last_error_code, + server_commit_unknown) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(command_id) DO UPDATE SET user_id = excluded.user_id, scope_id = excluded.scope_id, aggregate_type = excluded.aggregate_type, source_key = excluded.source_key, @@ -591,7 +631,8 @@ export class SqliteOfflineRepository implements OfflineRepository { replica_mutation = excluded.replica_mutation, payload_hash = excluded.payload_hash, base_revision_json = excluded.base_revision_json, state = excluded.state, attempts = excluded.attempts, - retry_at = excluded.retry_at, created_at = excluded.created_at, last_error_code = excluded.last_error_code`, + retry_at = excluded.retry_at, created_at = excluded.created_at, last_error_code = excluded.last_error_code, + server_commit_unknown = excluded.server_commit_unknown`, [ command.commandId, canonicalOfflinePrincipalId(command.userId), @@ -611,6 +652,7 @@ export class SqliteOfflineRepository implements OfflineRepository { command.retryAt, command.createdAt, command.lastErrorCode, + command.serverCommitUnknown === true ? 1 : 0, ], ); }