From a86482778a15cf01b35821d963d3b09df7157083 Mon Sep 17 00:00:00 2001 From: rdlabo Date: Wed, 12 Aug 2026 01:12:10 +0900 Subject: [PATCH 1/2] fix offline repository invariants --- .../src/lib/offline-command-executor.ts | 6 + .../src/lib/offline-repository.spec.ts | 125 ++++++++++++ .../kit/offline/src/lib/offline-repository.ts | 180 ++++++++++++++---- .../src/lib/offline-sync.service.spec.ts | 115 +++++++++++ .../offline/src/lib/offline-sync.service.ts | 50 ++++- 5 files changed, 430 insertions(+), 46 deletions(-) diff --git a/projects/kit/offline/src/lib/offline-command-executor.ts b/projects/kit/offline/src/lib/offline-command-executor.ts index 222ff37..fcdd1b2 100644 --- a/projects/kit/offline/src/lib/offline-command-executor.ts +++ b/projects/kit/offline/src/lib/offline-command-executor.ts @@ -31,6 +31,12 @@ 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; + /** + * Whether this transport error authoritatively proves that this idempotency + * key did not commit. Returning true may clear an ambiguity retained from an + * earlier response-loss attempt and expose normal conflict resolution. + */ + provesCommandNotCommitted?(error: unknown, command: OfflineCommand): boolean; /** * Removes the deleted remote row's revision from a queued recreate. * Required only when `clearRemoteId` completes while later commands remain. diff --git a/projects/kit/offline/src/lib/offline-repository.spec.ts b/projects/kit/offline/src/lib/offline-repository.spec.ts index ef7840f..17d3748 100644 --- a/projects/kit/offline/src/lib/offline-repository.spec.ts +++ b/projects/kit/offline/src/lib/offline-repository.spec.ts @@ -1711,6 +1711,131 @@ describe('IonicOfflineRepository', () => { expect(rows).toHaveLength(1); expect(rows[0]?.identity).toEqual(generatedReplicaIdentity('019d-aaaa', null)); }); + + it('legacy単一recordを一度だけ走査して全scope/source indexを構築する', async () => { + const userRow: OfflineReplicaRow = { + ...baseRow, + userId: 1, + scopeId: '10', + identity: generatedReplicaIdentity('019d-user-index', 41), + values: { id: 41, title: 'Indexed user row' }, + }; + const groupRow: OfflineReplicaRow = { + userId: 1, + scopeId: '20', + sourceKey: 'test_group_items', + identity: generatedReplicaIdentity('019d-group-index', 42), + values: { id: 42, name: 'Indexed group row' }, + confirmedValues: null, + serverRevision: null, + fetchedAt: 1, + syncState: 'pending', + }; + repository = await createSeededRepository(replicaSchemaV1, async () => { + await seedReplicaMetadata(replicaSchemaV1, { + '1:user:test_items:generated:019d-user-index': userRow, + '1:20:test_group_items:generated:019d-group-index': groupRow, + }); + }); + const get = vi.spyOn(storage, 'get'); + + await expect(repository.getReplicaRows({ userId: 1, scopeId: '10' }, 'test_items')).resolves.toHaveLength(1); + await expect(repository.getReplicaRows({ userId: 1, scopeId: '20' }, 'test_group_items')).resolves.toHaveLength(1); + + expect(get.mock.calls.filter(([key]) => key === 'offline:replica:rows')).toHaveLength(1); + }); + + it('replica transactionと同じcommit境界で既存partition indexを更新する', async () => { + const scope = { userId: 1, scopeId: '10' }; + const initial: OfflineReplicaRow = { + ...baseRow, + ...scope, + identity: generatedReplicaIdentity('019d-index-update', 44), + values: { id: 44, title: 'Before' }, + }; + await repository.transactReplica({ putRows: [initial] }); + await repository.getReplicaRows(scope, 'test_items'); + + await repository.transactReplica({ putRows: [{ ...initial, values: { id: 44, title: 'After' } }] }); + + await expect(repository.getReplicaRows(scope, 'test_items')).resolves.toMatchObject([{ values: { title: 'After' } }]); + }); + + it('初回partition構築と並行するtransactionを同じwrite laneで直列化する', async () => { + const scope = { userId: 1, scopeId: '10' }; + const initial: OfflineReplicaRow = { + ...baseRow, + ...scope, + identity: generatedReplicaIdentity('019d-index-race', 45), + values: { id: 45, title: 'Before' }, + }; + await repository.transactReplica({ putRows: [initial] }); + for (const key of [...storage.values.keys()]) { + if (key.startsWith('offline:replica:rows:index:v1:')) storage.values.delete(key); + } + let releaseRowsRead!: () => void; + const rowsRead = new Promise((resolve) => { + releaseRowsRead = resolve; + }); + const originalGet = storage.get.bind(storage); + vi.spyOn(storage, 'get').mockImplementation(async (key: string): Promise => { + if (key === 'offline:replica:rows') await rowsRead; + return originalGet(key); + }); + + const build = repository.getReplicaRows(scope, 'test_items'); + const update = repository.transactReplica({ + putRows: [{ ...initial, values: { id: 45, title: 'After' } }], + }); + releaseRowsRead(); + + await Promise.all([build, update]); + await expect(repository.getReplicaRows(scope, 'test_items')).resolves.toMatchObject([{ values: { title: 'After' } }]); + }); + + it('進行中transactionの後にclearScopeを同じwrite laneで確定する', async () => { + const scope = { userId: 1, scopeId: '10' }; + const initial: OfflineReplicaRow = { + userId: 1, + scopeId: '10', + sourceKey: 'test_group_items', + identity: generatedReplicaIdentity('019d-clear-race', 46), + values: { id: 46, name: 'Before clear' }, + confirmedValues: null, + serverRevision: null, + fetchedAt: 1, + syncState: 'pending', + }; + await repository.transactReplica({ putRows: [initial] }); + let releaseRowsWrite!: () => void; + let announceRowsWrite!: () => void; + const rowsWriteStarted = new Promise((resolve) => { + announceRowsWrite = resolve; + }); + const rowsWrite = new Promise((resolve) => { + releaseRowsWrite = resolve; + }); + const originalSet = storage.set.bind(storage); + let deferNextRowsWrite = true; + vi.spyOn(storage, 'set').mockImplementation(async (key: string, value: T): Promise => { + if (key === 'offline:replica:rows' && deferNextRowsWrite) { + deferNextRowsWrite = false; + announceRowsWrite(); + await rowsWrite; + } + return originalSet(key, value); + }); + + const update = repository.transactReplica({ + putRows: [{ ...initial, values: { id: 46, name: 'Concurrent update' } }], + }); + await rowsWriteStarted; + const clear = repository.clearScope(scope); + releaseRowsWrite(); + + await Promise.all([update, clear]); + await expect(repository.getReplicaRows(scope, 'test_group_items')).resolves.toEqual([]); + }); }); }); diff --git a/projects/kit/offline/src/lib/offline-repository.ts b/projects/kit/offline/src/lib/offline-repository.ts index 9f0bea7..128caa4 100644 --- a/projects/kit/offline/src/lib/offline-repository.ts +++ b/projects/kit/offline/src/lib/offline-repository.ts @@ -221,6 +221,8 @@ interface OfflineReplicaSchemaMigrationJournal { const METADATA_KEY = 'offline:metadata'; const SESSION_MANIFESTS_KEY = 'offline:session:manifests'; const ROWS_KEY = 'offline:replica:rows'; +const ROW_PARTITION_PREFIX = 'offline:replica:rows:index:v1:'; +const ROW_PARTITION_READY_KEY = 'offline:replica:rows:index:v1:ready'; const CURSORS_KEY = 'offline:replica:cursors'; const OUTBOX_KEY = 'offline:outbox:commands'; const REPLICA_TRANSACTION_KEY = 'offline:replica:transaction'; @@ -238,6 +240,7 @@ export class IonicOfflineRepository implements OfflineRepository { readonly #options = inject(OFFLINE_KIT_OPTIONS); #initialization: Promise | null = null; #writes: Promise = Promise.resolve(); + #rowIndexBuild: Promise | null = null; initialize(): Promise { if (!this.#initialization) { @@ -282,7 +285,7 @@ export class IonicOfflineRepository implements OfflineRepository { await this.initialize(); await this.#writes; const schema = this.#resolveReplicaEntitySchema(sourceKey); - const rows = await this.#readRecord>(ROWS_KEY); + const rows = await this.#readRowPartition(scope, sourceKey, schema); const row = this.#findRowByAddress(rows, scope, sourceKey, schema, identity); if (!row || (row.visibility ?? 'present') === 'pending_delete') return null; return this.#rowForScope(row, schema, scope) as OfflineReplicaRow; @@ -296,7 +299,7 @@ export class IonicOfflineRepository implements OfflineRepository { await this.initialize(); await this.#writes; const schema = this.#resolveReplicaEntitySchema(sourceKey); - const rows = await this.#readRecord>(ROWS_KEY); + const rows = await this.#readRowPartition(scope, sourceKey, schema); const row = this.#findRowByAddress(rows, scope, sourceKey, schema, identity); return row ? (this.#rowForScope(row, schema, scope) as OfflineReplicaRow) : null; } @@ -305,7 +308,7 @@ export class IonicOfflineRepository implements OfflineRepository { await this.initialize(); await this.#writes; const schema = this.#resolveReplicaEntitySchema(sourceKey); - const rows = await this.#readRecord>(ROWS_KEY); + const rows = await this.#readRowPartition(scope, sourceKey, schema); return Object.values(rows) .filter((row) => { if (row.sourceKey !== sourceKey || row.userId !== scope.userId) return false; @@ -334,7 +337,7 @@ export class IonicOfflineRepository implements OfflineRepository { await this.#writes; const schema = this.#resolveReplicaEntitySchema(sourceKey); const canonical = canonicalOfflineRemoteIdentity(schema, identity); - const rows = await this.#readRecord>(ROWS_KEY); + const rows = await this.#readRowPartition(scope, sourceKey, schema); const row = Object.values(rows).find((candidate) => { if (candidate.sourceKey !== sourceKey || candidate.userId !== scope.userId) return false; if (schema.scope === 'partition' && candidate.scopeId !== scope.scopeId) return false; @@ -410,44 +413,51 @@ export class IonicOfflineRepository implements OfflineRepository { async clearUser(userId: OfflinePrincipalId): Promise { await this.initialize(); - await Promise.all([ - this.#mutateRecord(SESSION_MANIFESTS_KEY, (manifests) => { - delete manifests[canonicalOfflinePrincipalId(userId)]; - return manifests; - }), - 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) { - await this.#storage.set(METADATA_KEY, { ...metadata, lastUserId: null }); - } + await this.#enqueueWrite(async () => { + await this.#removeRowPartitions((key) => key.startsWith(`${ROW_PARTITION_PREFIX}${canonicalOfflinePrincipalId(userId)}:`)); + await Promise.all([ + this.#mutateRecordNow(SESSION_MANIFESTS_KEY, (manifests) => { + delete manifests[canonicalOfflinePrincipalId(userId)]; + return manifests; + }), + this.#filterRecordNow(ROWS_KEY, (value) => value.userId !== userId), + this.#filterRecordNow(OUTBOX_KEY, (value) => value.userId !== userId), + this.#filterRecordNow(CURSORS_KEY, (_value, key) => !key.startsWith(`${canonicalOfflinePrincipalId(userId)}:`)), + this.#filterRecordNow(RECONCILIATION_SCOPES_KEY, (value) => value.userId !== userId), + ]); + const metadata = await this.#metadata(); + if (metadata.lastUserId === userId) { + await this.#storage.set(METADATA_KEY, { ...metadata, lastUserId: null }); + } + }); } async clearScope(scope: OfflineScope): Promise { await this.initialize(); const belongsToGroup = (value: OfflineScope) => value.userId === scope.userId && value.scopeId === scope.scopeId; - await Promise.all([ - this.#filterRecord(ROWS_KEY, (value) => { - const schema = this.#resolveReplicaEntitySchema(value.sourceKey); - return schema.scope === 'user' || !belongsToGroup(value); - }), - this.#filterRecord(OUTBOX_KEY, (value) => { - const schema = this.#resolveReplicaEntitySchema(value.sourceKey); - return schema.scope === 'user' || !belongsToGroup(value); - }), - this.#filterRecord(CURSORS_KEY, (_value, key) => key !== this.#cursorKey(scope)), - this.#filterRecord(RECONCILIATION_SCOPES_KEY, (value) => !belongsToGroup(value)), - ]); + await this.#enqueueWrite(async () => { + await this.#removeRowPartitions((key) => { + if (!key.startsWith(`${ROW_PARTITION_PREFIX}${canonicalOfflinePrincipalId(scope.userId)}:`)) return false; + return key.includes(`:partition:${encodeURIComponent(scope.scopeId)}:`); + }); + await Promise.all([ + this.#filterRecordNow(ROWS_KEY, (value) => { + const schema = this.#resolveReplicaEntitySchema(value.sourceKey); + return schema.scope === 'user' || !belongsToGroup(value); + }), + this.#filterRecordNow(OUTBOX_KEY, (value) => { + const schema = this.#resolveReplicaEntitySchema(value.sourceKey); + return schema.scope === 'user' || !belongsToGroup(value); + }), + this.#filterRecordNow(CURSORS_KEY, (_value, key) => key !== this.#cursorKey(scope)), + this.#filterRecordNow(RECONCILIATION_SCOPES_KEY, (value) => !belongsToGroup(value)), + ]); + }); } async transactReplica(transaction: OfflineReplicaTransaction): Promise { await this.initialize(); - const write = this.#writes.then(() => this.#applyReplicaTransaction(transaction, true)); - this.#writes = write.catch((): void => undefined); - return write; + return this.#enqueueWrite(() => this.#applyReplicaTransaction(transaction, true)); } async #migrate(): Promise { @@ -510,6 +520,7 @@ export class IonicOfflineRepository implements OfflineRepository { } async #recoverReplicaSchemaMigration(journal: OfflineReplicaSchemaMigrationJournal): Promise { + await this.#removeRowPartitions(() => true); await this.#storage.set(ROWS_KEY, journal.originalRows); const metadata = await this.#metadata(); await this.#storage.set(METADATA_KEY, { @@ -528,7 +539,7 @@ export class IonicOfflineRepository implements OfflineRepository { } } - const write = this.#writes.then(async (): Promise => { + return this.#enqueueWrite(async (): Promise => { const rows = await this.#readRecord(ROWS_KEY); const originalRows = structuredClone(rows); await this.#storage.set(REPLICA_SCHEMA_MIGRATION_KEY, { @@ -577,6 +588,7 @@ export class IonicOfflineRepository implements OfflineRepository { const metadata = await this.#metadata(); await this.#storage.set(ROWS_KEY, transformedRows); + await this.#removeRowPartitions(() => true); await this.#storage.set(METADATA_KEY, { ...metadata, replicaSchemaVersion: targetVersion, @@ -594,8 +606,6 @@ export class IonicOfflineRepository implements OfflineRepository { throw error; } }); - this.#writes = write.catch((): void => undefined); - return write; } #toWebMigrationRow(row: OfflineReplicaRow): OfflineReplicaWebMigrationRow { @@ -694,9 +704,82 @@ export class IonicOfflineRepository implements OfflineRepository { this.#storage.set(CURSORS_KEY, cursors), this.#storage.set(RECONCILIATION_SCOPES_KEY, reconciliationScopes), ]); + await this.#writeAffectedRowPartitions(rows, transaction); await this.#storage.remove(REPLICA_TRANSACTION_KEY); } + async #readRowPartition( + scope: OfflineScope, + sourceKey: string, + schema: OfflineReplicaEntitySchema>, + ): Promise>> { + const key = this.#rowPartitionKey(scope, sourceKey, schema); + const cached = await this.#storage.get>>(key); + if (cached !== null) return cached; + if (await this.#storage.get(ROW_PARTITION_READY_KEY)) return {}; + await this.#buildRowPartitions(); + const built = await this.#storage.get>>(key); + return built ?? {}; + } + + #buildRowPartitions(): Promise { + if (!this.#rowIndexBuild) { + const build = this.#enqueueWrite(async () => { + if (await this.#storage.get(ROW_PARTITION_READY_KEY)) return; + const rows = await this.#readRecord(ROWS_KEY); + const partitions = new Map>(); + for (const [rowKey, row] of Object.entries(rows)) { + const schema = this.#resolveReplicaEntitySchema(row.sourceKey); + const key = this.#rowPartitionKey(row, row.sourceKey, schema); + const partition = partitions.get(key) ?? {}; + partition[rowKey] = row; + partitions.set(key, partition); + } + await Promise.all([...partitions].map(([key, partition]) => this.#storage.set(key, partition))); + await this.#storage.set(ROW_PARTITION_READY_KEY, true); + }); + this.#rowIndexBuild = build.finally(() => { + this.#rowIndexBuild = null; + }); + } + return this.#rowIndexBuild; + } + + async #writeAffectedRowPartitions(rows: Record, transaction: OfflineReplicaTransaction): Promise { + const affected = new Map< + string, + { scope: OfflineScope; sourceKey: string; schema: OfflineReplicaEntitySchema> } + >(); + for (const row of [...(transaction.putRows ?? []), ...(transaction.removeRows ?? [])]) { + const schema = this.#resolveReplicaEntitySchema(row.sourceKey); + const key = this.#rowPartitionKey(row, row.sourceKey, schema); + affected.set(key, { scope: row, sourceKey: row.sourceKey, schema }); + } + await Promise.all( + [...affected].map(([key, { scope, sourceKey, schema }]) => + this.#storage.set( + key, + Object.fromEntries( + Object.entries(rows).filter(([, row]) => { + if (row.userId !== scope.userId || row.sourceKey !== sourceKey) return false; + return schema.scope === 'user' || row.scopeId === scope.scopeId; + }), + ), + ), + ), + ); + } + + #rowPartitionKey(scope: OfflineScope, sourceKey: string, schema: OfflineReplicaEntitySchema>): string { + const partition = schema.scope === 'user' ? 'user' : `partition:${encodeURIComponent(scope.scopeId)}`; + return `${ROW_PARTITION_PREFIX}${canonicalOfflinePrincipalId(scope.userId)}:${partition}:${encodeURIComponent(sourceKey)}`; + } + + async #removeRowPartitions(matches: (key: string) => boolean): Promise { + const keys = (await this.#storage.keys()).filter((key) => key.startsWith(ROW_PARTITION_PREFIX) && matches(key)); + await Promise.all(keys.map((key) => this.#storage.remove(key))); + } + async #metadata(): Promise { const metadata = await this.#storage.get>(METADATA_KEY); return { @@ -712,17 +795,30 @@ export class IonicOfflineRepository implements OfflineRepository { } async #filterRecord(key: string, predicate: (value: T, recordKey: string) => boolean): Promise { - await this.#mutateRecord(key, (record) => + await this.#enqueueWrite(() => this.#filterRecordNow(key, predicate)); + } + + #filterRecordNow(key: string, predicate: (value: T, recordKey: string) => boolean): Promise { + return this.#mutateRecordNow(key, (record) => Object.fromEntries(Object.entries(record).filter(([recordKey, value]) => predicate(value, recordKey))), ); } #mutateRecord(key: string, mutate: (record: Record) => Record): Promise { - const write = this.#writes.then(async (): Promise => { - const record = await this.#readRecord(key); - await this.#storage.set(key, mutate(record)); - }); - this.#writes = write.catch((): void => undefined); + return this.#enqueueWrite(() => this.#mutateRecordNow(key, mutate)); + } + + async #mutateRecordNow(key: string, mutate: (record: Record) => Record): Promise { + const record = await this.#readRecord(key); + await this.#storage.set(key, mutate(record)); + } + + #enqueueWrite(operation: () => Promise): Promise { + const write = this.#writes.then(operation); + this.#writes = write.then( + () => undefined, + () => undefined, + ); return write; } 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 2c1db8f..477cc76 100644 --- a/projects/kit/offline/src/lib/offline-sync.service.spec.ts +++ b/projects/kit/offline/src/lib/offline-sync.service.spec.ts @@ -87,6 +87,7 @@ describe('OfflineSyncService', () => { const execute = vi.fn( async (_command: OfflineCommand, _target: OfflineCommandTarget): Promise => ({ response: null }), ); + const provesCommandNotCommitted = vi.fn((_error: unknown, _command: OfflineCommand) => false); beforeEach(() => { commands = []; @@ -104,6 +105,8 @@ describe('OfflineSyncService', () => { options = { databaseName: 'test-offline', replicaSchema }; execute.mockReset(); execute.mockResolvedValue({ response: null }); + provesCommandNotCommitted.mockReset(); + provesCommandNotCommitted.mockReturnValue(false); const repository = { initialize: vi.fn(async () => undefined), getCommands: vi.fn(async (scope: OfflineScope) => { @@ -255,6 +258,7 @@ describe('OfflineSyncService', () => { provide: OFFLINE_COMMAND_EXECUTOR, useValue: { execute, + provesCommandNotCommitted, withServerRevision: (command: OfflineCommand) => command, withoutServerRevision: (command: OfflineCommand) => ({ ...command, baseRevision: null }), }, @@ -1021,6 +1025,91 @@ describe('OfflineSyncService', () => { }); }); + it('resolved conflictはreplacement準備成功まで元commandとoptimistic rowを保持する', async () => { + rows.push({ + userId: 1, + scopeId: '10', + sourceKey: 'documents', + identity: { kind: 'generated', localId: 'replace-failure', remoteId: 12 }, + values: { id: 12, title: 'local conflict' }, + confirmedValues: { id: 12, title: 'server' }, + serverRevision: 3, + fetchedAt: 1, + syncState: 'conflict', + }); + const commandId = await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'replace-failure' }, + operation: 'documents.update', + payload: { title: 'local conflict' }, + optimisticValue: { id: 12, title: 'local conflict' }, + baseRevision: 3, + }, + { flush: false }, + ); + commands[0] = { ...commands[0]!, state: 'conflict' }; + + await expect( + service.replacePrepared(commandId, async () => { + throw new Error('replacement preparation failed'); + }), + ).rejects.toThrow('replacement preparation failed'); + + expect(commands).toHaveLength(1); + expect(commands[0]?.commandId).toBe(commandId); + expect(rows[0]?.values).toEqual({ id: 12, title: 'local conflict' }); + }); + + it('resolved conflictをreplacement commandとoptimistic rowへ一transactionで置換する', async () => { + rows.push({ + userId: 1, + scopeId: '10', + sourceKey: 'documents', + identity: { kind: 'generated', localId: 'replace-success', remoteId: 13 }, + values: { id: 13, title: 'old local' }, + confirmedValues: { id: 13, title: 'server' }, + serverRevision: 4, + fetchedAt: 1, + syncState: 'conflict', + }); + const oldCommandId = await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'replace-success' }, + operation: 'documents.update', + payload: { title: 'old local' }, + optimisticValue: { id: 13, title: 'old local' }, + baseRevision: 4, + }, + { flush: false }, + ); + commands[0] = { ...commands[0]!, state: 'conflict' }; + + const newCommandId = await service.replacePrepared( + oldCommandId, + async () => ({ + request: { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'replace-success' }, + operation: 'documents.update', + payload: { title: 'new local' }, + optimisticValue: { id: 13, title: 'new local' }, + baseRevision: 4, + }, + }), + { flush: false }, + ); + + expect(newCommandId).not.toBe(oldCommandId); + expect(commands).toHaveLength(1); + expect(commands[0]).toMatchObject({ commandId: newCommandId, state: 'pending' }); + expect(rows[0]).toMatchObject({ values: { id: 13, title: 'new local' }, syncState: 'pending' }); + }); + it('同一ミリ秒のDate.nowでもcreatedAtは単調増加で保存する', async () => { const nowSpy = vi.spyOn(Date, 'now').mockReturnValue(1_700_000_000_000); await service.enqueue( @@ -1358,6 +1447,32 @@ describe('OfflineSyncService', () => { expect(service.pendingCommands()).toEqual([]); }); + it('executorが同じkeyの未commitを証明した競合はambiguityを解除して通常解決へ渡す', async () => { + execute.mockRejectedValueOnce({ status: 0 }).mockRejectedValueOnce({ status: 412 }); + provesCommandNotCommitted.mockImplementation((error) => (error as { status?: number }).status === 412); + const commandId = await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'authoritative-no-commit' }, + operation: 'documents.upsert', + payload: {}, + optimisticValue: {}, + }, + { flush: false }, + ); + connected.set(true); + await service.flush(); + await service.retryNow(commandId); + + expect(service.pendingCommands()[0]).toMatchObject({ + commandId, + state: 'conflict', + serverCommitUnknown: false, + }); + await expect(service.discard(commandId, { flush: false })).resolves.toBeUndefined(); + }); + it('flush中のsession切替後に旧user commandを新sessionへ復活させない', async () => { let resolveExecute!: (value: { response: null; serverRevision?: number }) => void; execute.mockImplementationOnce(() => new Promise((resolve) => (resolveExecute = resolve))); diff --git a/projects/kit/offline/src/lib/offline-sync.service.ts b/projects/kit/offline/src/lib/offline-sync.service.ts index 0625af5..73258e0 100644 --- a/projects/kit/offline/src/lib/offline-sync.service.ts +++ b/projects/kit/offline/src/lib/offline-sync.service.ts @@ -245,6 +245,30 @@ export class OfflineSyncService { }); } + /** + * Atomically replaces a resolved durable command with a newly prepared command. + * + * Use this after an authoritative conflict read. The old intent remains durable + * until validation, capacity checks, optimistic projection and replacement + * command preparation have all succeeded. + */ + replacePrepared( + commandId: string, + prepare: (repository: OfflineRepository) => Promise>, + options: { flush?: boolean } = {}, + ): Promise { + const generation = this.#generation; + return this.#serializeReplicaMutation(async () => { + await this.initialize(); + if (!this.#isCurrent(generation)) throw new Error('Offline session changed before prepared replacement.'); + const replaced = (await this.#readKnownCommands()).find((command) => command.commandId === commandId); + if (!replaced) throw new Error(`Offline command ${commandId} no longer exists.`); + this.#assertDiscardable([replaced]); + const prepared = await prepare(this.#repository); + return this.#enqueue(prepared.request, options, generation, prepared.replicaTransaction, replaced); + }); + } + /** * Serializes a product-owned replica projection with enqueue and command ACK * reconciliation. For read/derive/write cache updates, prefer @@ -281,6 +305,7 @@ export class OfflineSyncService { options: { flush?: boolean }, generation: number, replicaTransaction?: Pick, + replaced?: OfflineCommand, ): Promise { await this.initialize(); if (this.#options.mode === 'readCacheOnly') { @@ -319,6 +344,16 @@ export class OfflineSyncService { createdAt: await this.#nextCommandCreatedAt(userId), lastErrorCode: null, }; + if ( + replaced && + (replaced.userId !== command.userId || + replaced.scopeId !== command.scopeId || + replaced.aggregateType !== command.aggregateType || + replaced.sourceKey !== command.sourceKey || + canonicalOfflineCommandIdentity(replaced.identity) !== canonicalOfflineCommandIdentity(command.identity)) + ) { + throw new Error('Offline replacement command must address the same aggregate and replica identity.'); + } const entityType = command.sourceKey; const schema = this.#entitySchema(entityType); if (schema.identity.kind === 'localOnly') { @@ -400,7 +435,7 @@ export class OfflineSyncService { const optimisticCompanions = await this.#prepareOptimisticCompanions(scope, optimisticRow, replicaTransaction); if (replicaTransaction) this.#canonicalJson(replicaTransaction); if (optimisticCompanions.length > 0) command = { ...command, optimisticCompanions }; - await this.#assertOutboxCapacity(userId, command); + await this.#assertOutboxCapacity(userId, command, replaced?.commandId); if (generation !== this.#generation) { throw new Error('Offline session changed before the command could be persisted'); } @@ -408,6 +443,7 @@ export class OfflineSyncService { putRows: [optimisticRow, ...(replicaTransaction?.putRows ?? [])], removeRows: replicaTransaction?.removeRows, putCommands: [command], + removeCommandIds: replaced ? [replaced.commandId] : undefined, }); await this.#refreshState(); if (options.flush !== false && this.#network.connected()) this.#flushInBackground(); @@ -474,14 +510,17 @@ export class OfflineSyncService { return `${canonicalOfflinePrincipalId(key.userId)}:${key.scopeId}:${key.sourceKey}:${canonicalOfflineReplicaIdentity(key.identity)}`; } - async #assertOutboxCapacity(userId: OfflinePrincipalId, command: OfflineCommand): Promise { - const commands = this.#repository.getCommandsForUser + async #assertOutboxCapacity(userId: OfflinePrincipalId, command: OfflineCommand, excludingCommandId?: string): Promise { + const currentCommands = this.#repository.getCommandsForUser ? await this.#repository.getCommandsForUser(userId) : ( await Promise.all( [...this.#knownScopes.values()].filter((scope) => scope.userId === userId).map((scope) => this.#repository.getCommands(scope)), ) ).flat(); + const commands = excludingCommandId + ? currentCommands.filter((candidate) => candidate.commandId !== excludingCommandId) + : currentCommands; const maxCommands = this.#options.outboxLimits?.maxCommandsPerUser ?? DEFAULT_MAX_OUTBOX_COMMANDS_PER_USER; const maxBytes = this.#options.outboxLimits?.maxBytesPerUser ?? DEFAULT_MAX_OUTBOX_BYTES_PER_USER; const currentBytes = this.#serializedOutboxBytes(commands); @@ -740,7 +779,10 @@ export class OfflineSyncService { 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)); + const commitUnknown = this.#executor.provesCommandNotCommitted?.(error, sending) + ? false + : priorCommitUnknown || this.#serverCommitCouldBeUnknown(error); + await this.#persistFailedCommand(sending, error, generation, row, commitUnknown); if (!this.#isClassifiableTransportError(error)) throw error; break; } From a9b40888f2ff51e1223e69ceae4ab8cc8c7cf9ec Mon Sep 17 00:00:00 2001 From: rdlabo Date: Wed, 12 Aug 2026 01:15:17 +0900 Subject: [PATCH 2/2] remove obsolete repository helper --- projects/kit/offline/src/lib/offline-repository.ts | 4 ---- 1 file changed, 4 deletions(-) diff --git a/projects/kit/offline/src/lib/offline-repository.ts b/projects/kit/offline/src/lib/offline-repository.ts index 128caa4..f811bc9 100644 --- a/projects/kit/offline/src/lib/offline-repository.ts +++ b/projects/kit/offline/src/lib/offline-repository.ts @@ -794,10 +794,6 @@ export class IonicOfflineRepository implements OfflineRepository { return (await this.#storage.get>(key)) ?? {}; } - async #filterRecord(key: string, predicate: (value: T, recordKey: string) => boolean): Promise { - await this.#enqueueWrite(() => this.#filterRecordNow(key, predicate)); - } - #filterRecordNow(key: string, predicate: (value: T, recordKey: string) => boolean): Promise { return this.#mutateRecordNow(key, (record) => Object.fromEntries(Object.entries(record).filter(([recordKey, value]) => predicate(value, recordKey))),