From cb549b4f773ba26025d1bfc5502b8df950d9a46d Mon Sep 17 00:00:00 2001 From: rdlabo Date: Fri, 14 Aug 2026 22:59:41 +0900 Subject: [PATCH] Make offline upgrades lossless --- .../lib/offline-replica-pull.service.spec.ts | 59 ++- .../src/lib/offline-replica-pull.service.ts | 430 +++++++++++------- .../src/lib/offline-repository.spec.ts | 50 ++ .../kit/offline/src/lib/offline-repository.ts | 57 ++- .../src/lib/sqlite-offline-repository.spec.ts | 95 +++- .../src/lib/sqlite-offline-repository.ts | 96 +++- 6 files changed, 600 insertions(+), 187 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 d47565a..ff994ed 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 @@ -145,14 +145,14 @@ describe('OfflineReplicaPullService', () => { storage.values.set('offline:replica:cursors', {}); } - function configureTestBed(): void { + function configureTestBed(mode: 'synchronized' | 'readCacheOnly' = 'synchronized'): void { TestBed.resetTestingModule(); TestBed.configureTestingModule({ providers: [ OfflineReplicaPullService, IonicOfflineRepository, { provide: KitStorageService, useValue: storage }, - { provide: OFFLINE_KIT_OPTIONS, useValue: { databaseName: 'test-offline', replicaSchema } }, + { provide: OFFLINE_KIT_OPTIONS, useValue: { databaseName: 'test-offline', replicaSchema, mode } }, { provide: OFFLINE_REPOSITORY, useExisting: IonicOfflineRepository }, { provide: OFFLINE_REPLICA_PULLER, useValue: { pull } }, { provide: OFFLINE_REPLICA_PROJECTOR, useValue: projector }, @@ -201,6 +201,20 @@ describe('OfflineReplicaPullService', () => { }); }); + it('readCacheOnlyでもprefetched authoritative changesは永続cacheへ適用する', async () => { + configureTestBed('readCacheOnly'); + repository = TestBed.inject(OFFLINE_REPOSITORY); + service = TestBed.inject(OfflineReplicaPullService); + await repository.initialize(); + + await service.applyChanges(scope, [itemChange(42, 'Web foreground')]); + + await expect(repository.getReplicaRowByRemoteId(scope, 'test_items', 42)).resolves.toMatchObject({ + confirmedValues: { title: 'Web foreground' }, + }); + expect(pull).not.toHaveBeenCalled(); + }); + it('applies rebaseline reset, derived projection, base rows, and cursor in one transaction', async () => { await repository.transactReplica({ putRows: [ @@ -1108,6 +1122,47 @@ describe('OfflineReplicaPullService', () => { expect(await repository.getReplicaRows(scope, 'test_items')).toHaveLength(1); }); + it('prefetched authoritative changeもpullと同じ境界でACKしcursorを変更しない', async () => { + await seedPendingCreate(); + await repository.transactReplica({ putCursors: [{ ...scope, cursor: 'cursor-before-prefetch' }] }); + + await service.applyChanges(scope, async (transactionRepository) => { + await expect(transactionRepository.getReplicaCursor(scope)).resolves.toEqual({ + ...scope, + cursor: 'cursor-before-prefetch', + }); + return [ + itemChange(42, 'Created from foreground response', { + serverRevision: 1, + acknowledgedCommandIds: ['cmd-create'], + }), + ]; + }); + + expect(pull).not.toHaveBeenCalled(); + await expect(repository.getReplicaCursor(scope)).resolves.toEqual({ ...scope, cursor: 'cursor-before-prefetch' }); + await expect(repository.getCommands(scope)).resolves.toEqual([]); + await expect(repository.getReplicaRow(scope, 'test_items', generatedCommandIdentity('019d-create'))).resolves.toMatchObject({ + identity: { kind: 'generated', localId: '019d-create', remoteId: 42 }, + confirmedValues: { title: 'Created from foreground response' }, + syncState: 'confirmed', + }); + }); + + it('prefetched pageはsource単位にbulk preloadしchange単位lookupを行わない', async () => { + const bulkRead = vi.spyOn(repository, 'getReplicaRowsIncludingPendingDelete'); + const remoteLookup = vi.spyOn(repository, 'getReplicaRowByRemoteIdentity'); + + await service.applyChanges( + scope, + Array.from({ length: 25 }, (_, index) => itemChange(index + 1, `Item ${index + 1}`)), + ); + + expect(bulkRead).toHaveBeenCalledOnce(); + expect(remoteLookup).not.toHaveBeenCalled(); + await expect(repository.getReplicaRows(scope, 'test_items')).resolves.toHaveLength(25); + }); + it('journal retention後のrebaselineでもawaiting-pull targetをhydrateしACK changeと同じtransactionで除去する', async () => { await seedPendingCreate(); const current = (await repository.getCommands(scope))[0]!; 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 e378bd5..ddd6034 100644 --- a/projects/kit/offline/src/lib/offline-replica-pull.service.ts +++ b/projects/kit/offline/src/lib/offline-replica-pull.service.ts @@ -31,6 +31,7 @@ import { OFFLINE_REPOSITORY, canonicalOfflineReplicaRowKey, type OfflineCommand, + type OfflineRepository, type OfflineReplicaRow, type OfflineReplicaRowKey, type OfflineScope, @@ -43,6 +44,10 @@ type CollapsedOfflineReplicaChange = OfflineReplicaChange & { collapsedOrdinal: number; }; +export type OfflineReplicaChangeBatch = + | readonly OfflineReplicaChange[] + | ((repository: OfflineRepository) => Promise); + /** * Pull handshake reported a replica schema version/hash that does not match the local Kit schema. * @@ -114,201 +119,284 @@ export class OfflineReplicaPullService { continue; } - const applied = await this.#replicaMutations.run(async (repository) => { - const currentCursor = (await repository.getReplicaCursor(scope))?.cursor ?? ''; - if (currentCursor !== persistedCursor) return currentCursor; - const scopeCommands = await repository.getCommands(scope); - const userCommands = repository.getCommandsForUser ? await repository.getCommandsForUser(scope.userId) : scopeCommands; - const changes = this.#collapseChanges(page.changes); - const projection = await this.#projector?.project({ - scope, - changes, - }); - this.#assertProjection(scope, projection); - const putRows: OfflineReplicaRow[] = []; - const removeRows: OfflineReplicaRowKey[] = []; - const putCommands = new Map(); - const removeCommandIds = new Set(); - const rematerializeAfter: OfflineCommand[] = []; - if (rebaselinePending || page.rebaselineRequired) { - removeRows.push(...(await this.#confirmedRowsForRebaseline(scope, userCommands))); - } + const applied = await this.#applyAuthoritativeChanges(scope, page.changes, { + expectedCursor: persistedCursor, + nextCursor: page.nextCursor, + rebaseline: rebaselinePending || page.rebaselineRequired === true, + reconciliationTargetsById, + }); + if (applied !== page.nextCursor) { + persistedCursor = applied; + requestCursor = applied; + rebaselinePending = false; + continue; + } + persistedCursor = page.nextCursor; + requestCursor = page.nextCursor; + rebaselinePending = false; + if (!page.hasMore) return; + } + } - for (const change of changes) { - const schema = this.#entitySchema(change.sourceKey); - const commands = schema.scope === 'user' ? userCommands : scopeCommands; - const acknowledged = (change.acknowledgedCommandIds ?? []) - .map((commandId) => { - const command = commands.find((candidate) => candidate.commandId === commandId); - if (!command) return null; - if (command.sourceKey !== change.sourceKey) { - throw new Error(`Acknowledged command "${commandId}" does not target "${change.sourceKey}".`); + /** + * Applies an already-fetched authoritative server batch through the same + * acknowledgement, conflict, and aggregate-rematerialization boundary as pull. + */ + async applyChanges(scope: OfflineScope, changes: OfflineReplicaChangeBatch): Promise { + const reconciliationTargets = await this.#reconciliationTargets(scope); + await this.#applyAuthoritativeChanges(scope, changes, { + expectedCursor: null, + nextCursor: null, + rebaseline: false, + reconciliationTargetsById: new Map(reconciliationTargets.map((target) => [target.commandId, target] as const)), + }); + } + + async #applyAuthoritativeChanges( + scope: OfflineScope, + incomingChanges: OfflineReplicaChangeBatch, + options: { + readonly expectedCursor: string | null; + readonly nextCursor: string | null; + readonly rebaseline: boolean; + readonly reconciliationTargetsById: ReadonlyMap; + }, + ): Promise { + return this.#replicaMutations.run(async (repository) => { + const currentCursor = (await repository.getReplicaCursor(scope))?.cursor ?? ''; + if (options.expectedCursor !== null && currentCursor !== options.expectedCursor) return currentCursor; + const scopeCommands = await repository.getCommands(scope); + const userCommands = repository.getCommandsForUser ? await repository.getCommandsForUser(scope.userId) : scopeCommands; + const changes = this.#collapseChanges(typeof incomingChanges === 'function' ? await incomingChanges(repository) : incomingChanges); + const existingRows = await this.#preloadRows(repository, scope, changes); + const projection = await this.#projector?.project({ + scope, + changes, + }); + this.#assertProjection(scope, projection); + const putRows: OfflineReplicaRow[] = []; + const removeRows: OfflineReplicaRowKey[] = []; + const putCommands = new Map(); + const removeCommandIds = new Set(); + const rematerializeAfter: OfflineCommand[] = []; + if (options.rebaseline) { + removeRows.push(...(await this.#confirmedRowsForRebaseline(scope, userCommands))); + } + + for (const change of changes) { + const schema = this.#entitySchema(change.sourceKey); + const commands = schema.scope === 'user' ? userCommands : scopeCommands; + const acknowledged = (change.acknowledgedCommandIds ?? []) + .map((commandId) => { + const command = commands.find((candidate) => candidate.commandId === commandId); + if (!command) return null; + if (command.sourceKey !== change.sourceKey) { + throw new Error(`Acknowledged command "${commandId}" does not target "${change.sourceKey}".`); + } + const requested = options.reconciliationTargetsById.get(commandId); + if (command.state === 'awaiting_pull') { + if (!requested) { + throw new Error(`Acknowledged command "${commandId}" was not requested for reconciliation on this pull page.`); } - const requested = reconciliationTargetsById.get(commandId); - if (command.state === 'awaiting_pull') { - if (!requested) { - throw new Error(`Acknowledged command "${commandId}" was not requested for reconciliation on this pull page.`); - } - const schema = this.#entitySchema(change.sourceKey); - if ( - canonicalOfflineRemoteIdentity(schema, requested.identity) !== - canonicalOfflineRemoteIdentity(schema, this.#identity(change)) - ) { - throw new Error(`Acknowledged command "${commandId}" does not match the requested remote identity.`); - } + const schema = this.#entitySchema(change.sourceKey); + if ( + canonicalOfflineRemoteIdentity(schema, requested.identity) !== + canonicalOfflineRemoteIdentity(schema, this.#identity(change)) + ) { + throw new Error(`Acknowledged command "${commandId}" does not match the requested remote identity.`); } - return command; - }) - .filter((command): command is OfflineCommand => command !== null); - const acknowledgedIdentities = new Set(acknowledged.map((command) => canonicalOfflineCommandIdentity(command.identity))); - if (acknowledgedIdentities.size > 1) { - throw new Error(`Acknowledged commands for "${change.sourceKey}" target multiple replica identities.`); - } - const acknowledgedCommand = acknowledged[0]; - const acknowledgedScope = acknowledgedCommand - ? { userId: acknowledgedCommand.userId, scopeId: acknowledgedCommand.scopeId } - : scope; - const acknowledgedRow = acknowledgedCommand - ? await (this.#repository.getReplicaRowIncludingPendingDelete?.( - acknowledgedScope, - change.sourceKey, - acknowledgedCommand.identity, - ) ?? this.#repository.getReplicaRow(acknowledgedScope, change.sourceKey, acknowledgedCommand.identity)) - : null; - if (acknowledgedCommand && !acknowledgedRow) { - throw new Error(`Acknowledged command "${acknowledgedCommand.commandId}" has no local replica row.`); - } - const identity = this.#identity(change); - const serverRow = await this.#repository.getReplicaRowByRemoteIdentity(scope, change.sourceKey, identity); - if ( - acknowledgedRow && - serverRow && - !commandIdentityMatchesReplicaRow(schema, acknowledgedRow, commandIdentityFromReplicaIdentity(serverRow.identity)) - ) { - if (identity.remoteId !== undefined) { - throw new Error(`Server id ${String(identity.remoteId)} is already mapped to another local replica row.`); } - throw new Error(`Remote identity for "${change.sourceKey}" is already mapped to another local replica row.`); - } - const existing = acknowledgedRow ?? serverRow; - const related = existing - ? commands.filter( - (command) => command.sourceKey === change.sourceKey && commandIdentityMatchesReplicaRow(schema, existing, command.identity), - ) - : []; - const hasPending = related.length > 0; - - if (acknowledgedCommand) { - this.#assertIdentityAssignment(schema, existing!, identity); - this.#applyAcknowledgement(change, existing!, related, putRows, removeRows, putCommands, removeCommandIds, rematerializeAfter); - continue; + return command; + }) + .filter((command): command is OfflineCommand => command !== null); + const acknowledgedIdentities = new Set(acknowledged.map((command) => canonicalOfflineCommandIdentity(command.identity))); + if (acknowledgedIdentities.size > 1) { + throw new Error(`Acknowledged commands for "${change.sourceKey}" target multiple replica identities.`); + } + const acknowledgedCommand = acknowledged[0]; + const acknowledgedScope = acknowledgedCommand + ? { userId: acknowledgedCommand.userId, scopeId: acknowledgedCommand.scopeId } + : scope; + const acknowledgedRow = acknowledgedCommand + ? (existingRows.byCommand.get(this.#commandRowKey(change.sourceKey, acknowledgedScope, acknowledgedCommand.identity)) ?? null) + : null; + if (acknowledgedCommand && !acknowledgedRow) { + throw new Error(`Acknowledged command "${acknowledgedCommand.commandId}" has no local replica row.`); + } + const identity = this.#identity(change); + const serverRow = existingRows.byRemote.get(this.#remoteRowKey(change.sourceKey, scope, identity)) ?? null; + if ( + acknowledgedRow && + serverRow && + !commandIdentityMatchesReplicaRow(schema, acknowledgedRow, commandIdentityFromReplicaIdentity(serverRow.identity)) + ) { + if (identity.remoteId !== undefined) { + throw new Error(`Server id ${String(identity.remoteId)} is already mapped to another local replica row.`); } + throw new Error(`Remote identity for "${change.sourceKey}" is already mapped to another local replica row.`); + } + const existing = acknowledgedRow ?? serverRow; + const related = existing + ? commands.filter( + (command) => command.sourceKey === change.sourceKey && commandIdentityMatchesReplicaRow(schema, existing, command.identity), + ) + : []; + const hasPending = related.length > 0; + + if (acknowledgedCommand) { + this.#assertIdentityAssignment(schema, existing!, identity); + this.#applyAcknowledgement(change, existing!, related, putRows, removeRows, putCommands, removeCommandIds, rematerializeAfter); + continue; + } - if (change.deleted) { - if (!existing) continue; - if (!hasPending) { - removeRows.push({ ...existing, identity: existing.identity }); - continue; - } - putRows.push({ - ...existing, - confirmedValues: null, - serverRevision: change.serverRevision, - syncState: 'conflict', - fetchedAt: Date.now(), - }); - for (const command of related) { - putCommands.set(command.commandId, { ...command, state: 'conflict', retryAt: null, lastErrorCode: 'remote_deleted' }); - } - rematerializeAfter.push(related[0]!); + if (change.deleted) { + if (!existing) continue; + if (!hasPending) { + removeRows.push({ ...existing, identity: existing.identity }); continue; } - - const confirmedValues = this.#validatedValues(schema, change); - if (!existing) { - putRows.push({ - ...scope, - sourceKey: change.sourceKey, - identity: - schema.identity.kind === 'naturalKey' - ? offlineNaturalReplicaIdentity(schema, confirmedValues) - : offlineGeneratedReplicaIdentity(crypto.randomUUID(), identity.remoteId ?? null), - values: confirmedValues, - confirmedValues, - serverRevision: change.serverRevision, - fetchedAt: Date.now(), - syncState: 'confirmed', - }); - continue; + putRows.push({ + ...existing, + confirmedValues: null, + serverRevision: change.serverRevision, + syncState: 'conflict', + fetchedAt: Date.now(), + }); + for (const command of related) { + putCommands.set(command.commandId, { ...command, state: 'conflict', retryAt: null, lastErrorCode: 'remote_deleted' }); } + rematerializeAfter.push(related[0]!); + continue; + } - const remaining = related.filter((command) => !removeCommandIds.has(command.commandId)); - const confirmedRow: OfflineReplicaRow = { - ...existing, + const confirmedValues = this.#validatedValues(schema, change); + if (!existing) { + putRows.push({ + ...scope, + sourceKey: change.sourceKey, + identity: + schema.identity.kind === 'naturalKey' + ? offlineNaturalReplicaIdentity(schema, confirmedValues) + : offlineGeneratedReplicaIdentity(crypto.randomUUID(), identity.remoteId ?? null), + values: confirmedValues, confirmedValues, serverRevision: change.serverRevision, fetchedAt: Date.now(), - values: remaining.length > 0 ? existing.values : confirmedValues, - syncState: remaining.length > 0 ? 'pending' : 'confirmed', - }; - putRows.push(confirmedRow); - if (remaining.length > 0) rematerializeAfter.push(remaining[0]!); - for (const command of remaining) { - putCommands.set(command.commandId, offlineCommandWithBaseRevision(command, change.serverRevision)); - } + syncState: 'confirmed', + }); + continue; } - const confirmedAndProjected = this.#mergeRowMutations([ - { putRows, removeRows }, - { putRows: projection?.putRows ?? [], removeRows: projection?.removeRows ?? [] }, - ]); - const rematerialized = await this.#rematerializePendingAggregates( - scope, - userCommands, - scopeCommands, - confirmedAndProjected, - putCommands, - removeCommandIds, - rematerializeAfter, - ); - const finalRows = this.#mergeRowMutations([confirmedAndProjected, rematerialized]); - const removedCommands = [...removeCommandIds] - .map( - (commandId) => - userCommands.find((command) => command.commandId === commandId) ?? - scopeCommands.find((command) => command.commandId === commandId), - ) - .filter((command): command is OfflineCommand => command != null); - await repository.transactReplica({ - putRows: finalRows.putRows, - removeRows: finalRows.removeRows, - putCommands: [...putCommands.values()], - removeCommandIds: [...removeCommandIds], - putCursors: [{ ...scope, cursor: page.nextCursor }], - }); - await Promise.all(removedCommands.map((command) => this.#hooks.onCommandRemoved?.(command).catch(() => undefined))); - return page.nextCursor; + const remaining = related.filter((command) => !removeCommandIds.has(command.commandId)); + const confirmedRow: OfflineReplicaRow = { + ...existing, + confirmedValues, + serverRevision: change.serverRevision, + fetchedAt: Date.now(), + values: remaining.length > 0 ? existing.values : confirmedValues, + syncState: remaining.length > 0 ? 'pending' : 'confirmed', + }; + putRows.push(confirmedRow); + if (remaining.length > 0) rematerializeAfter.push(remaining[0]!); + for (const command of remaining) { + putCommands.set(command.commandId, offlineCommandWithBaseRevision(command, change.serverRevision)); + } + } + + const confirmedAndProjected = this.#mergeRowMutations([ + { putRows, removeRows }, + { putRows: projection?.putRows ?? [], removeRows: projection?.removeRows ?? [] }, + ]); + const rematerialized = await this.#rematerializePendingAggregates( + scope, + userCommands, + scopeCommands, + confirmedAndProjected, + putCommands, + removeCommandIds, + rematerializeAfter, + ); + const finalRows = this.#mergeRowMutations([confirmedAndProjected, rematerialized]); + const removedCommands = [...removeCommandIds] + .map( + (commandId) => + userCommands.find((command) => command.commandId === commandId) ?? + scopeCommands.find((command) => command.commandId === commandId), + ) + .filter((command): command is OfflineCommand => command != null); + await repository.transactReplica({ + putRows: finalRows.putRows, + removeRows: finalRows.removeRows, + putCommands: [...putCommands.values()], + removeCommandIds: [...removeCommandIds], + ...(options.nextCursor === null ? {} : { putCursors: [{ ...scope, cursor: options.nextCursor }] }), }); - if (applied !== page.nextCursor) { - persistedCursor = applied; - requestCursor = applied; - rebaselinePending = false; - continue; + await Promise.all(removedCommands.map((command) => this.#hooks.onCommandRemoved?.(command).catch(() => undefined))); + return options.nextCursor ?? currentCursor; + }); + } + + async #preloadRows( + repository: OfflineRepository, + scope: OfflineScope, + changes: readonly OfflineReplicaChange[], + ): Promise<{ + byCommand: ReadonlyMap; + byRemote: ReadonlyMap; + }> { + const sourceKeys = [...new Set(changes.map((change) => change.sourceKey))]; + const partitions = await Promise.all( + sourceKeys.map(async (sourceKey) => ({ + sourceKey, + rows: await (repository.getReplicaRowsIncludingPendingDelete?.(scope, sourceKey) ?? repository.getReplicaRows(scope, sourceKey)), + })), + ); + const byCommand = new Map(); + const byRemote = new Map(); + for (const { sourceKey, rows } of partitions) { + const schema = this.#entitySchema(sourceKey); + for (const row of rows) { + if (row.identity.kind !== 'local') { + byCommand.set(this.#commandRowKey(sourceKey, row, commandIdentityFromReplicaIdentity(row.identity)), row); + } + if (row.identity.kind === 'generated' && row.identity.remoteId !== null) { + byRemote.set(this.#remoteRowKey(sourceKey, row, { remoteId: row.identity.remoteId }), row); + } else if (row.identity.kind === 'natural') { + byRemote.set(this.#remoteRowKey(sourceKey, row, { naturalKey: row.identity.naturalKey }), row); + } } - persistedCursor = page.nextCursor; - requestCursor = page.nextCursor; - rebaselinePending = false; - if (!page.hasMore) return; } + return { byCommand, byRemote }; + } + + #commandRowKey(sourceKey: string, scope: OfflineScope, identity: OfflineCommand['identity']): string { + const partition = this.#entitySchema(sourceKey).scope === 'user' ? '' : scope.scopeId; + return `${sourceKey}:${scope.userId}:${partition}:${canonicalOfflineCommandIdentity(identity)}`; + } + + #remoteRowKey(sourceKey: string, scope: OfflineScope, identity: OfflineReplicaRemoteIdentity): string { + const schema = this.#entitySchema(sourceKey); + const partition = schema.scope === 'user' ? '' : scope.scopeId; + return `${sourceKey}:${scope.userId}:${partition}:${canonicalOfflineRemoteIdentity(schema, identity)}`; } async #reconciliationTargets(scope: OfflineScope): Promise { const commands = (await this.#repository.getCommands(scope)).filter((command) => command.state === 'awaiting_pull'); + const rowsByCommand = new Map(); + await Promise.all( + [...new Set(commands.map((command) => command.sourceKey))].map(async (sourceKey) => { + const rows = await (this.#repository.getReplicaRowsIncludingPendingDelete?.(scope, sourceKey) ?? + this.#repository.getReplicaRows(scope, sourceKey)); + for (const row of rows) { + if (row.identity.kind !== 'local') { + rowsByCommand.set(this.#commandRowKey(sourceKey, row, commandIdentityFromReplicaIdentity(row.identity)), row); + } + } + }), + ); const targets: OfflineReplicaReconciliationTarget[] = []; for (const command of commands) { - const row = - (await this.#repository.getReplicaRowIncludingPendingDelete?.(scope, command.sourceKey, command.identity)) ?? - (await this.#repository.getReplicaRow(scope, command.sourceKey, command.identity)); + const row = rowsByCommand.get(this.#commandRowKey(command.sourceKey, command, command.identity)); if (!row) throw new Error(`Awaiting-pull command "${command.commandId}" has no replica row.`); const identity = command.reconciliationIdentity ?? diff --git a/projects/kit/offline/src/lib/offline-repository.spec.ts b/projects/kit/offline/src/lib/offline-repository.spec.ts index bf6e5a8..dfef13a 100644 --- a/projects/kit/offline/src/lib/offline-repository.spec.ts +++ b/projects/kit/offline/src/lib/offline-repository.spec.ts @@ -1021,6 +1021,56 @@ describe('IonicOfflineRepository', () => { expect(storage.values.get('firebaseToken')).toEqual({ token: 'keep' }); }); + it('released core schema v1をOutboxの状態とoptimistic imageを保ってv2へ上げる', async () => { + const legacyCommand = { + userId: 1, + scopeId: '10', + commandId: 'legacy-create', + aggregateType: 'test_items', + sourceKey: 'test_items', + identity: { kind: 'generated', localId: '019d-legacy' }, + operation: 'test_items.create', + payload: { encryptedBody: 'ciphertext' }, + optimisticValue: { id: 0, title: 'Legacy optimistic' }, + optimisticCompanions: [{ key: 'legacy-companion' }], + payloadHash: 'legacy-hash', + baseRevision: null, + state: 'retry_wait', + attempts: 2, + retryAt: 123, + createdAt: 1, + lastErrorCode: 'network', + serverCommitUnknown: true, + }; + storage.values.set('offline:metadata', { + schemaVersion: 1, + lastUserId: 1, + replicaSchemaVersion: replicaSchemaV1.version, + replicaSchemaHash: await sha256OfflineReplicaSchema(replicaSchemaV1), + }); + storage.values.set('offline:outbox:commands', { [legacyCommand.commandId]: legacyCommand }); + + await repository.initialize(); + + await expect(repository.getCommands({ userId: 1, scopeId: '10' })).resolves.toEqual([ + expect.objectContaining({ + commandId: 'legacy-create', + payload: { encryptedBody: 'ciphertext' }, + legacyOptimisticValue: { id: 0, title: 'Legacy optimistic' }, + legacyOptimisticCompanions: [{ key: 'legacy-companion' }], + legacyPayloadHash: 'legacy-hash', + state: 'retry_wait', + attempts: 2, + retryAt: 123, + serverCommitUnknown: true, + }), + ]); + expect(storage.values.get('offline:metadata')).toMatchObject({ schemaVersion: OFFLINE_SCHEMA_VERSION, lastUserId: 1 }); + expect(storage.values.get('offline:outbox:commands')).not.toHaveProperty('legacy-create.optimisticValue'); + expect(storage.values.get('offline:outbox:commands')).not.toHaveProperty('legacy-create.optimisticCompanions'); + expect(storage.values.get('offline:outbox:commands')).not.toHaveProperty('legacy-create.payloadHash'); + }); + it('local replica fallbackは通信不能だけを対象にする', () => { expect(isOfflineFallbackError({ status: 0 })).toBe(true); expect(isOfflineFallbackError({ status: 403 })).toBe(false); diff --git a/projects/kit/offline/src/lib/offline-repository.ts b/projects/kit/offline/src/lib/offline-repository.ts index 75cfec6..e0eccd4 100644 --- a/projects/kit/offline/src/lib/offline-repository.ts +++ b/projects/kit/offline/src/lib/offline-repository.ts @@ -73,6 +73,16 @@ interface OfflineCommandBase extends OfflineScope { operation: string; /** Opaque product payload. Kit never mutates this after enqueue. */ payload: T; + /** + * Read-only compatibility image retained when upgrading schema-v1 commands. + * New commands never write this field; products may use it only to normalize + * a released legacy payload into their current versioned intent contract. + */ + readonly legacyOptimisticValue?: unknown; + /** Released schema-v1 companion images retained for product-owned normalization. */ + readonly legacyOptimisticCompanions?: unknown; + /** Released schema-v1 integrity value retained while a legacy command remains durable. */ + readonly legacyPayloadHash?: string; /** * Declared localOnly projection rows this intent may create, update, or remove. * Kit persists keys only; before/after images are not durable truth. @@ -193,6 +203,8 @@ export interface OfflineRepositoryReader { identity: OfflineReplicaAddress, ): Promise | null>; getReplicaRows(scope: OfflineScope, sourceKey: string): Promise[]>; + /** Internal synchronization bulk read that also returns pending-delete tombstones. */ + getReplicaRowsIncludingPendingDelete?(scope: OfflineScope, sourceKey: string): Promise[]>; getReplicaRowByRemoteId( scope: OfflineScope, sourceKey: string, @@ -228,6 +240,8 @@ export interface OfflineRepository { identity: OfflineReplicaAddress, ): Promise | null>; getReplicaRows(scope: OfflineScope, sourceKey: string): Promise[]>; + /** Internal synchronization bulk read that also returns pending-delete tombstones. */ + getReplicaRowsIncludingPendingDelete?(scope: OfflineScope, sourceKey: string): Promise[]>; getReplicaRowByRemoteId( scope: OfflineScope, sourceKey: string, @@ -383,6 +397,13 @@ export class IonicOfflineRepository implements OfflineRepository { return this.#withCommittedRead(() => this.#readReplicaRows(scope, sourceKey)); } + async getReplicaRowsIncludingPendingDelete( + scope: OfflineScope, + sourceKey: string, + ): Promise[]> { + return this.#withCommittedRead(() => this.#readReplicaRows(scope, sourceKey, true)); + } + async getReplicaRowByRemoteId( scope: OfflineScope, sourceKey: string, @@ -450,13 +471,17 @@ export class IonicOfflineRepository implements OfflineRepository { return this.#rowForScope(row, schema, scope) as OfflineReplicaRow; } - async #readReplicaRows(scope: OfflineScope, sourceKey: string): Promise[]> { + async #readReplicaRows( + scope: OfflineScope, + sourceKey: string, + includePendingDelete = false, + ): Promise[]> { const schema = this.#resolveReplicaEntitySchema(sourceKey); const rows = await this.#readRowPartition(scope, sourceKey, schema); return Object.values(rows) .filter((row) => { if (row.sourceKey !== sourceKey || row.userId !== scope.userId) return false; - if ((row.visibility ?? 'present') === 'pending_delete') return false; + if (!includePendingDelete && (row.visibility ?? 'present') === 'pending_delete') return false; return schema.scope === 'partition' ? row.scopeId === scope.scopeId : true; }) .map((row) => this.#rowForScope(row, schema, scope)) @@ -516,6 +541,7 @@ export class IonicOfflineRepository implements OfflineRepository { getReplicaRow: (scope, sourceKey, identity) => this.#readReplicaRow(scope, sourceKey, identity, false), getReplicaRowIncludingPendingDelete: (scope, sourceKey, identity) => this.#readReplicaRow(scope, sourceKey, identity, true), getReplicaRows: (scope, sourceKey) => this.#readReplicaRows(scope, sourceKey), + getReplicaRowsIncludingPendingDelete: (scope, sourceKey) => this.#readReplicaRows(scope, sourceKey, true), getReplicaRowByRemoteId: async (scope, sourceKey, remoteId) => { if (this.#resolveReplicaEntitySchema(sourceKey).identity.kind !== 'generated') return null; return this.#readReplicaRowByRemoteIdentity(scope, sourceKey, { remoteId }); @@ -647,7 +673,9 @@ export class IonicOfflineRepository implements OfflineRepository { async #migrate(): Promise { const metadata = await this.#storage.get>(METADATA_KEY); - if (metadata?.schemaVersion !== undefined && metadata.schemaVersion !== OFFLINE_SCHEMA_VERSION) { + if (metadata?.schemaVersion === 1) { + await this.#migrateCoreSchemaV1(); + } else if (metadata?.schemaVersion !== undefined && metadata.schemaVersion !== OFFLINE_SCHEMA_VERSION) { throw new Error( `Unsupported offline storage schema version ${metadata.schemaVersion}; expected ${OFFLINE_SCHEMA_VERSION}. ` + 'A lossless core schema migration is required before this database can be opened.', @@ -666,6 +694,29 @@ export class IonicOfflineRepository implements OfflineRepository { if (interrupted) await this.#applyReplicaTransaction(interrupted, false); } + async #migrateCoreSchemaV1(): Promise { + const commands = await this.#readRecord< + OfflineCommand & { optimisticValue?: unknown; optimisticCompanions?: unknown; payloadHash?: string } + >(OUTBOX_KEY); + const migratedCommands = Object.fromEntries( + Object.entries(commands).map(([key, command]) => { + const { optimisticValue, optimisticCompanions, payloadHash, ...current } = command; + return [ + key, + { + ...current, + ...(optimisticValue === undefined ? {} : { legacyOptimisticValue: structuredClone(optimisticValue) }), + ...(optimisticCompanions === undefined ? {} : { legacyOptimisticCompanions: structuredClone(optimisticCompanions) }), + ...(payloadHash === undefined ? {} : { legacyPayloadHash: payloadHash }), + }, + ]; + }), + ); + const metadata = await this.#metadata(); + await this.#storage.set(OUTBOX_KEY, migratedCommands); + await this.#storage.set(METADATA_KEY, { ...metadata, schemaVersion: OFFLINE_SCHEMA_VERSION }); + } + async #initializeReplicaSchema(storedVersion: number | null, storedHash: string | null): Promise { const bundle = this.#options.replicaSchema; const targetVersion = bundle.version; 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 1018fd2..14f56cc 100644 --- a/projects/kit/offline/src/lib/sqlite-offline-repository.spec.ts +++ b/projects/kit/offline/src/lib/sqlite-offline-repository.spec.ts @@ -276,6 +276,72 @@ describe('SqliteOfflineRepository community sqlite driver', () => { await expect(options.createEncryptionKey?.()).resolves.toBe('first-install-secret'); }); + it('released core schema v1を同一transactionでv2へ上げoptimistic imageを保持する', async () => { + plugin.query.mockImplementation(async ({ statement }: { statement: string }) => { + if (statement.includes('offline_replica_schema_metadata')) { + return { columns: ['version', 'schema_hash'], rows: [[replicaSchemaV1.version, replicaSchemaV1Hash]] }; + } + if (statement.includes('offline_metadata') && statement.includes('schema_version')) { + return { rows: [{ schema_version: 1 }] }; + } + if (statement === 'PRAGMA table_info(offline_sync_commands)') { + return { + rows: [{ name: 'optimistic_value_json' }, { name: 'server_commit_unknown' }], + }; + } + return { rows: [] }; + }); + const repository = createRepository(); + + await repository.initialize(); + + const statements = plugin.execute.mock.calls.map(([options]) => options as { statement: string; values?: unknown[] }); + expect(statements.map(({ statement }) => statement)).toEqual( + expect.arrayContaining([ + 'ALTER TABLE offline_sync_commands ADD COLUMN legacy_optimistic_value_json TEXT', + 'ALTER TABLE offline_sync_commands ADD COLUMN local_only_footprint_json TEXT', + 'ALTER TABLE offline_sync_commands ADD COLUMN reconciliation_identity_json TEXT', + 'UPDATE offline_sync_commands SET legacy_optimistic_value_json = optimistic_value_json WHERE legacy_optimistic_value_json IS NULL', + ]), + ); + expect(statements).toContainEqual( + expect.objectContaining({ + statement: 'UPDATE offline_metadata SET schema_version = ? WHERE id = 1', + values: [OFFLINE_SCHEMA_VERSION], + }), + ); + expect(plugin.beginTransaction).toHaveBeenCalledOnce(); + expect(plugin.commitTransaction).toHaveBeenCalledOnce(); + expect(plugin.rollbackTransaction).not.toHaveBeenCalled(); + }); + + it('v1 tableのNOT NULL optimistic columnを保ったまま新形式commandを書ける', async () => { + const repository = createRepository(); + await repository.initialize(); + await repository.putCommand({ + userId: 1, + scopeId: '10', + commandId: 'new-command', + aggregateType: 'test_items', + sourceKey: 'test_items', + identity: { kind: 'generated', localId: 'new-local' }, + operation: 'test_items.create', + payload: { title: 'New' }, + baseRevision: null, + state: 'pending', + attempts: 0, + retryAt: null, + createdAt: 1, + lastErrorCode: null, + }); + + const insert = plugin.execute.mock.calls + .map(([options]) => options as { statement: string; values?: unknown[] }) + .find(({ statement }) => statement.startsWith('INSERT INTO offline_sync_commands')); + expect(insert?.statement).toContain('optimistic_value_json'); + expect(insert?.values).toContain('null'); + }); + it('partition scopeのcursorだけを単一transactionで削除しuser-scoped outboxを保持する', async () => { const repository = createRepository(); await repository.initialize(); @@ -685,7 +751,7 @@ describe('SqliteOfflineRepository community sqlite driver', () => { expect(insert?.statement).toContain('replica_mutation'); expect(insert?.values).toContain('delete'); expect(insert?.statement).toContain('payload_hash'); - expect(insert?.values?.[10]).toBe(''); + expect(insert?.values?.[13]).toBe(''); }); it('persists and restores declared localOnly footprint keys', async () => { @@ -779,6 +845,33 @@ describe('SqliteOfflineRepository community sqlite driver', () => { expect(plugin.commitTransaction).toHaveBeenCalledOnce(); }); + it('同一partitionの複数putはexisting rowsを一度だけbulk preloadする', async () => { + const repository = createRepository(); + await repository.initialize(); + plugin.query.mockClear(); + + await repository.transactReplica({ + putRows: [1, 2].map( + (id): OfflineReplicaRow => ({ + userId: 1, + scopeId: '10', + sourceKey: 'test_items', + identity: { kind: 'generated', localId: `local-${id}`, remoteId: id }, + values: { id, title: `Item ${id}` }, + confirmedValues: { id, title: `Item ${id}` }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'confirmed', + }), + ), + }); + + const tableReads = plugin.query.mock.calls.filter(([options]) => + (options as { statement: string }).statement.includes('SELECT * FROM test_items'), + ); + expect(tableReads).toHaveLength(1); + }); + it('transaction中の書き込み失敗をrollbackして握りつぶさない', async () => { const error = new Error('disk full'); const repository = createRepository(); diff --git a/projects/kit/offline/src/lib/sqlite-offline-repository.ts b/projects/kit/offline/src/lib/sqlite-offline-repository.ts index 4b8f327..aeb9aac 100644 --- a/projects/kit/offline/src/lib/sqlite-offline-repository.ts +++ b/projects/kit/offline/src/lib/sqlite-offline-repository.ts @@ -7,7 +7,6 @@ import { offlineNaturalReplicaIdentity, parseOfflineCommandIdentity, parseOfflinePrincipalId, - replicaAddressFromIdentity, serializeOfflineCommandIdentity, } from './offline-identity'; import { OFFLINE_KIT_OPTIONS } from './offline-kit-options'; @@ -176,6 +175,9 @@ const SCHEMA = [ identity_json TEXT NOT NULL, operation TEXT NOT NULL, payload_json TEXT NOT NULL, + optimistic_value_json TEXT NOT NULL DEFAULT 'null', + optimistic_companions_json TEXT, + legacy_optimistic_value_json TEXT, local_only_footprint_json TEXT, replica_mutation TEXT NOT NULL DEFAULT 'upsert', payload_hash TEXT NOT NULL, @@ -288,6 +290,13 @@ export class SqliteOfflineRepository implements OfflineRepository { return this.#withCommittedRead(() => this.#readReplicaRows(scope, sourceKey)); } + async getReplicaRowsIncludingPendingDelete( + scope: OfflineScope, + sourceKey: string, + ): Promise[]> { + return this.#withCommittedRead(() => this.#readReplicaRows(scope, sourceKey, true)); + } + async getReplicaRowByRemoteId( scope: OfflineScope, sourceKey: string, @@ -457,10 +466,28 @@ export class SqliteOfflineRepository implements OfflineRepository { releases.set(key, release); } const consumedReleases = new Set(); + const existingRows = new Map(); + const partitions = new Map(); + for (const row of transaction.putRows ?? []) { + const schema = this.#resolveReplicaEntitySchema(row.sourceKey); + const partitionScopeId = schema.scope === 'partition' ? row.scopeId : ''; + partitions.set(`${canonicalOfflinePrincipalId(row.userId)}:${partitionScopeId}:${row.sourceKey}`, { + scope: row, + sourceKey: row.sourceKey, + }); + } + await Promise.all( + [...partitions.values()].map(async ({ scope, sourceKey }) => { + for (const row of await this.#readReplicaRows(scope, sourceKey, true)) { + existingRows.set(this.#replicaRowKey(row), row); + } + }), + ); for (const row of transaction.putRows ?? []) { const key = this.#replicaRowKey(row); const release = releases.get(key); - await this.#putReplicaRow(databaseId, row, release); + await this.#putReplicaRow(databaseId, row, release, existingRows.get(key) ?? null); + existingRows.set(key, row); if (release) consumedReleases.add(key); } if (consumedReleases.size !== releases.size) { @@ -516,7 +543,9 @@ export class SqliteOfflineRepository implements OfflineRepository { ]); } else { const storedVersion = this.#number(metadata[0]!['schema_version']); - if (storedVersion !== OFFLINE_SCHEMA_VERSION) { + if (storedVersion === 1) { + await this.#migrateCoreSchemaV1(databaseId); + } else if (storedVersion !== OFFLINE_SCHEMA_VERSION) { throw new OfflineStorageUnavailableError( 'core_schema_incompatible', `Unsupported offline storage schema version ${storedVersion}; expected ${OFFLINE_SCHEMA_VERSION}. ` + @@ -527,6 +556,29 @@ export class SqliteOfflineRepository implements OfflineRepository { await this.#initializeReplicaSchema(databaseId); } + async #migrateCoreSchemaV1(databaseId: string): Promise { + const columns = await this.#queryDatabase(databaseId, 'PRAGMA table_info(offline_sync_commands)'); + const names = new Set(columns.map((row) => this.#string(row['name']))); + await this.#nativeTransaction(databaseId, async () => { + if (!names.has('legacy_optimistic_value_json')) { + await this.#execute(databaseId, 'ALTER TABLE offline_sync_commands ADD COLUMN legacy_optimistic_value_json TEXT'); + } + if (!names.has('local_only_footprint_json')) { + await this.#execute(databaseId, 'ALTER TABLE offline_sync_commands ADD COLUMN local_only_footprint_json TEXT'); + } + if (!names.has('reconciliation_identity_json')) { + await this.#execute(databaseId, 'ALTER TABLE offline_sync_commands ADD COLUMN reconciliation_identity_json TEXT'); + } + if (names.has('optimistic_value_json')) { + await this.#execute( + databaseId, + 'UPDATE offline_sync_commands SET legacy_optimistic_value_json = optimistic_value_json WHERE legacy_optimistic_value_json IS NULL', + ); + } + await this.#execute(databaseId, 'UPDATE offline_metadata SET schema_version = ? WHERE id = 1', [OFFLINE_SCHEMA_VERSION]); + }); + } + async #initializeReplicaSchema(databaseId: string): Promise { const bundle = this.#options.replicaSchema; const targetVersion = bundle.version; @@ -673,7 +725,11 @@ export class SqliteOfflineRepository implements OfflineRepository { return row ? this.#parse(row['value_json']) : null; } - async #readReplicaRows(scope: OfflineScope, sourceKey: string): Promise[]> { + async #readReplicaRows( + scope: OfflineScope, + sourceKey: string, + includePendingDelete = false, + ): Promise[]> { const schema = this.#resolveReplicaEntitySchema(sourceKey); const predicates = ['_offline_user_id = ?']; const values: SQLiteValue[] = [canonicalOfflinePrincipalId(scope.userId)]; @@ -687,7 +743,7 @@ export class SqliteOfflineRepository implements OfflineRepository { : 'local_id ASC'; const rows = await this.#query(`SELECT * FROM ${schema.tableName} WHERE ${predicates.join(' AND ')} ORDER BY ${orderBy}`, values); return rows - .filter((row) => (row['_offline_visibility'] ?? 'present') !== 'pending_delete') + .filter((row) => includePendingDelete || (row['_offline_visibility'] ?? 'present') !== 'pending_delete') .map((row) => this.#replicaRowFromSqliteRow(schema, scope, sourceKey, row)); } @@ -784,6 +840,7 @@ export class SqliteOfflineRepository implements OfflineRepository { ): Promise | null> => (await this.#queryReplicaRow(scope, sourceKey, identity, true)) as OfflineReplicaRow | null, getReplicaRows: (scope, sourceKey) => this.#readReplicaRows(scope, sourceKey), + getReplicaRowsIncludingPendingDelete: (scope, sourceKey) => this.#readReplicaRows(scope, sourceKey, true), getReplicaRowByRemoteId: async (scope, sourceKey, remoteId) => { if (this.#resolveReplicaEntitySchema(sourceKey).identity.kind !== 'generated') return null; return this.#readReplicaRowByRemoteIdentity(scope, sourceKey, { remoteId }); @@ -956,6 +1013,15 @@ export class SqliteOfflineRepository implements OfflineRepository { identity: parseOfflineCommandIdentity(this.#parse(row['identity_json'])), operation: this.#string(row['operation']), payload: this.#parse(row['payload_json']), + ...(row['legacy_optimistic_value_json'] === null || row['legacy_optimistic_value_json'] === undefined + ? {} + : { legacyOptimisticValue: this.#parse(row['legacy_optimistic_value_json']) }), + ...(row['optimistic_companions_json'] === null || row['optimistic_companions_json'] === undefined + ? {} + : { legacyOptimisticCompanions: this.#parse(row['optimistic_companions_json']) }), + ...(row['payload_hash'] === null || row['payload_hash'] === undefined || row['payload_hash'] === '' + ? {} + : { legacyPayloadHash: this.#string(row['payload_hash']) }), ...(row['local_only_footprint_json'] === null || row['local_only_footprint_json'] === undefined ? {} : { localOnlyFootprint: this.#parse(row['local_only_footprint_json']) }), @@ -978,13 +1044,16 @@ 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, - local_only_footprint_json, replica_mutation, payload_hash, base_revision_json, state, attempts, retry_at, created_at, last_error_code, + optimistic_value_json, optimistic_companions_json, legacy_optimistic_value_json, local_only_footprint_json, replica_mutation, payload_hash, base_revision_json, state, attempts, retry_at, created_at, last_error_code, server_commit_unknown, reconciliation_identity_json) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + 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, identity_json = excluded.identity_json, operation = excluded.operation, payload_json = excluded.payload_json, + optimistic_value_json = excluded.optimistic_value_json, + optimistic_companions_json = excluded.optimistic_companions_json, + legacy_optimistic_value_json = excluded.legacy_optimistic_value_json, local_only_footprint_json = excluded.local_only_footprint_json, replica_mutation = excluded.replica_mutation, payload_hash = excluded.payload_hash, @@ -1001,9 +1070,12 @@ export class SqliteOfflineRepository implements OfflineRepository { serializeOfflineCommandIdentity(command.identity), command.operation, JSON.stringify(command.payload), + JSON.stringify(command.legacyOptimisticValue ?? null), + command.legacyOptimisticCompanions === undefined ? null : JSON.stringify(command.legacyOptimisticCompanions), + command.legacyOptimisticValue === undefined ? null : JSON.stringify(command.legacyOptimisticValue), command.localOnlyFootprint === undefined ? null : JSON.stringify(command.localOnlyFootprint), command.replicaMutation ?? 'upsert', - '', + command.legacyPayloadHash ?? '', this.#stringifyNullable(command.baseRevision), command.state, command.attempts, @@ -1052,7 +1124,12 @@ export class SqliteOfflineRepository implements OfflineRepository { return this.#replicaRowFromSqliteRow(schema, scope, sourceKey, row); } - async #putReplicaRow(databaseId: string, row: OfflineReplicaRow, release: OfflineReplicaRemoteIdRelease | undefined): Promise { + async #putReplicaRow( + databaseId: string, + row: OfflineReplicaRow, + release: OfflineReplicaRemoteIdRelease | undefined, + existing: OfflineReplicaRow | null, + ): Promise { const schema = this.#resolveReplicaEntitySchema(row.sourceKey); if (row.identity.kind === 'generated') { assertOfflineReplicaGeneratedRemoteId(schema, row.identity.remoteId); @@ -1072,7 +1149,6 @@ export class SqliteOfflineRepository implements OfflineRepository { const encoded = encodeOfflineReplicaValues(schema, row.values); if (row.confirmedValues !== null) encodeOfflineReplicaValues(schema, row.confirmedValues); assertOfflineReplicaNaturalKeyBaseline(schema, row.values, row.confirmedValues); - const existing = await this.#queryReplicaRow(row, row.sourceKey, replicaAddressFromIdentity(row.identity), true); if (!existing && release) { throw new Error( `Offline replica remoteId release requires an existing row for ${row.sourceKey}/${canonicalOfflineReplicaIdentity(row.identity)}.`,