From 6033a68775b20e150c93e0bcaecdd9104c5b7b02 Mon Sep 17 00:00:00 2001 From: rdlabo Date: Thu, 13 Aug 2026 00:03:46 +0900 Subject: [PATCH] fix(offline): isolate synchronization failure boundaries --- .../lib/offline-replica-pull.service.spec.ts | 23 +- .../src/lib/offline-replica-pull.service.ts | 26 +- .../src/lib/offline-repository.spec.ts | 345 ++++++ .../kit/offline/src/lib/offline-repository.ts | 261 +++- .../src/lib/offline-sync.service.spec.ts | 1045 ++++++++++++++++- .../offline/src/lib/offline-sync.service.ts | 148 ++- .../src/lib/sqlite-offline-repository.spec.ts | 288 +++++ .../src/lib/sqlite-offline-repository.ts | 262 +++-- 8 files changed, 2211 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 8e01f8b..17cf942 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 @@ -12,7 +12,7 @@ import { type OfflineReplicaPullPage, type OfflineReplicaPullRequest, } from './offline-replica-puller'; -import { OfflineReplicaPullService } from './offline-replica-pull.service'; +import { OfflineReplicaPullService, OfflineReplicaSchemaMismatchError } from './offline-replica-pull.service'; import { OfflineReplicaMutationCoordinator } from './offline-replica-mutation-coordinator'; import { generatedCommandIdentity, generatedReplicaIdentity } from './offline-test-helpers'; import { @@ -762,11 +762,18 @@ describe('OfflineReplicaPullService', () => { await expect(repository.getReplicaCursor(scope)).resolves.toEqual({ ...scope, cursor: 'cursor-v0' }); }); - it('schema mismatchはrejectしcursorを進めない', async () => { + it('schema mismatchはtyped errorでrejectしcursorを進めない', async () => { await repository.transactReplica({ putCursors: [{ ...scope, cursor: 'cursor-v0' }] }); pull.mockResolvedValueOnce(page([itemChange(42, 'Created')], { nextCursor: 'cursor-v1', schemaVersion: 99, schemaHash: 'deadbeef' })); - await expect(service.pull(scope)).rejects.toThrow('Offline replica schema mismatch'); + const rejection = service.pull(scope); + await expect(rejection).rejects.toBeInstanceOf(OfflineReplicaSchemaMismatchError); + await expect(rejection).rejects.toMatchObject({ + code: OfflineReplicaSchemaMismatchError.code, + clientVersion: 1, + serverVersion: 99, + serverHash: 'deadbeef', + }); await expect(repository.getReplicaCursor(scope)).resolves.toEqual({ ...scope, cursor: 'cursor-v0' }); expect(await repository.getReplicaRows(scope, 'test_items')).toEqual([]); }); @@ -1115,13 +1122,9 @@ describe('OfflineReplicaPullService', () => { })), }); pull.mockResolvedValueOnce( - page( - [ - itemChange(42, 'Remote 42', { serverRevision: 9 }), - itemChange(43, 'Remote 43', { serverRevision: 10 }), - ], - { nextCursor: 'cursor-v2' }, - ), + page([itemChange(42, 'Remote 42', { serverRevision: 9 }), itemChange(43, 'Remote 43', { serverRevision: 10 })], { + nextCursor: 'cursor-v2', + }), ); await service.pull(scope); 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 d0c26a7..ba36c4c 100644 --- a/projects/kit/offline/src/lib/offline-replica-pull.service.ts +++ b/projects/kit/offline/src/lib/offline-replica-pull.service.ts @@ -40,6 +40,28 @@ type CollapsedOfflineReplicaChange = OfflineReplicaChange & { collapsedOrdinal: number; }; +/** + * Pull handshake reported a replica schema version/hash that does not match the local Kit schema. + * + * Prefer `instanceof` (or {@link OfflineReplicaSchemaMismatchError.code}) over English message text. + */ +export class OfflineReplicaSchemaMismatchError extends Error { + /** Stable machine-readable discriminator for fatal pull classification (pre- and post-send). */ + static readonly code = 'OFFLINE_REPLICA_SCHEMA_MISMATCH' as const; + + readonly code = OfflineReplicaSchemaMismatchError.code; + + constructor( + readonly clientVersion: number, + readonly clientHash: string, + readonly serverVersion: number, + readonly serverHash: string, + ) { + super(`Offline replica schema mismatch: client=${clientVersion}/${clientHash}, server=${serverVersion}/${serverHash}.`); + this.name = 'OfflineReplicaSchemaMismatchError'; + } +} + /** Pulls authoritative server deltas into one durable local replica partition. */ @Injectable({ providedIn: 'root' }) export class OfflineReplicaPullService { @@ -456,9 +478,7 @@ export class OfflineReplicaPullService { #assertHandshake(version: number, hash: string, expectedHash: string): void { if (version !== this.#options.replicaSchema.version || hash !== expectedHash) { - throw new Error( - `Offline replica schema mismatch: client=${this.#options.replicaSchema.version}/${expectedHash}, server=${version}/${hash}.`, - ); + throw new OfflineReplicaSchemaMismatchError(this.#options.replicaSchema.version, expectedHash, version, hash); } } diff --git a/projects/kit/offline/src/lib/offline-repository.spec.ts b/projects/kit/offline/src/lib/offline-repository.spec.ts index 17d3748..ef254dd 100644 --- a/projects/kit/offline/src/lib/offline-repository.spec.ts +++ b/projects/kit/offline/src/lib/offline-repository.spec.ts @@ -1836,6 +1836,351 @@ describe('IonicOfflineRepository', () => { await Promise.all([update, clear]); await expect(repository.getReplicaRows(scope, 'test_group_items')).resolves.toEqual([]); }); + + it('standalone getReplicaRowsはreader leaseを保持し、完了までwriterを開始せずin-flight writeを観測しない', async () => { + const scope = { userId: 1, scopeId: '10' }; + const initial: OfflineReplicaRow = { + ...baseRow, + ...scope, + identity: generatedReplicaIdentity('019d-lease-read', 50), + values: { id: 50, title: 'Before' }, + confirmedValues: { id: 50, title: 'Before' }, + serverRevision: 1, + syncState: 'confirmed', + }; + await repository.transactReplica({ putRows: [initial] }); + // Build indexes first so the deferred get is the leased partition read, not the write-lane build. + await repository.getReplicaRows(scope, 'test_items'); + + let releaseRead!: () => void; + const readGate = new Promise((resolve) => { + releaseRead = resolve; + }); + let announceRead!: () => void; + const readStarted = new Promise((resolve) => { + announceRead = resolve; + }); + const originalGet = storage.get.bind(storage); + let deferPartitionRead = true; + vi.spyOn(storage, 'get').mockImplementation(async (key: string): Promise => { + if (deferPartitionRead && key === 'offline:replica:rows:index:v1:n:1:user:test_items') { + deferPartitionRead = false; + announceRead(); + await readGate; + } + return originalGet(key); + }); + + const read = repository.getReplicaRows(scope, 'test_items'); + await readStarted; + + let writeFinished = false; + const write = repository + .transactReplica({ + putRows: [{ ...initial, values: { id: 50, title: 'After' }, confirmedValues: { id: 50, title: 'After' } }], + }) + .then(() => { + writeFinished = true; + }); + void write.then( + () => undefined, + () => undefined, + ); + await Promise.resolve(); + await Promise.resolve(); + expect(writeFinished).toBe(false); + const rowsBeforeRelease = storage.values.get('offline:replica:rows') as Record; + expect(Object.values(rowsBeforeRelease)).toEqual([expect.objectContaining({ values: expect.objectContaining({ title: 'Before' }) })]); + + releaseRead(); + await expect(read).resolves.toEqual([expect.objectContaining({ values: expect.objectContaining({ title: 'Before' }) })]); + await write; + expect(writeFinished).toBe(true); + + await expect( + repository.transactReplica({ + putRows: [ + { + ...initial, + identity: generatedReplicaIdentity('019d-lease-read', 51), + values: { id: 51, title: 'illegal-rebind' }, + }, + ], + }), + ).rejects.toThrow('Offline replica remoteId is immutable'); + await expect(repository.getReplicaRows(scope, 'test_items')).resolves.toEqual([ + expect.objectContaining({ + values: expect.objectContaining({ title: 'After' }), + identity: expect.objectContaining({ remoteId: 50 }), + }), + ]); + }); + + it('ready確認のawait中に開始したwriterを先に確定してからreaderを登録し、committed状態だけを観測する', async () => { + const scope = { userId: 1, scopeId: '10' }; + const initial: OfflineReplicaRow = { + ...baseRow, + ...scope, + identity: generatedReplicaIdentity('019d-admit-order', 60), + values: { id: 60, title: 'Before' }, + confirmedValues: { id: 60, title: 'Before' }, + serverRevision: 1, + syncState: 'confirmed', + }; + await repository.transactReplica({ putRows: [initial] }); + // Indexes must already be ready so ensure only does an async ready check (admission gap). + await repository.getReplicaRows(scope, 'test_items'); + + let releaseReadyCheck!: () => void; + const readyCheckGate = new Promise((resolve) => { + releaseReadyCheck = resolve; + }); + let announceReadyCheck!: () => void; + const readyCheckStarted = new Promise((resolve) => { + announceReadyCheck = resolve; + }); + const originalGet = storage.get.bind(storage); + let deferReadyCheck = true; + vi.spyOn(storage, 'get').mockImplementation(async (key: string): Promise => { + if (deferReadyCheck && key === 'offline:replica:rows:index:v1:ready') { + deferReadyCheck = false; + announceReadyCheck(); + await readyCheckGate; + } + return originalGet(key); + }); + + let releaseWrite!: () => void; + const writeGate = new Promise((resolve) => { + releaseWrite = resolve; + }); + let announceWrite!: () => void; + const writeStarted = new Promise((resolve) => { + announceWrite = resolve; + }); + const originalSet = storage.set.bind(storage); + let deferRowsWrite = true; + vi.spyOn(storage, 'set').mockImplementation(async (key: string, value: T): Promise => { + if (deferRowsWrite && key === 'offline:replica:rows') { + deferRowsWrite = false; + announceWrite(); + await writeGate; + } + return originalSet(key, value); + }); + + const read = repository.getReplicaRows(scope, 'test_items'); + await readyCheckStarted; + + let writeFinished = false; + const write = repository + .transactReplica({ + putRows: [{ ...initial, values: { id: 60, title: 'After' }, confirmedValues: { id: 60, title: 'After' } }], + }) + .then(() => { + writeFinished = true; + }); + void write.then( + () => undefined, + () => undefined, + ); + await writeStarted; + + releaseReadyCheck(); + await Promise.resolve(); + await Promise.resolve(); + expect(writeFinished).toBe(false); + // Reader must still be waiting on the write tail — not overlapping the in-flight write. + let readSettled = false; + void read.then( + () => { + readSettled = true; + }, + () => { + readSettled = true; + }, + ); + await Promise.resolve(); + expect(readSettled).toBe(false); + + releaseWrite(); + await write; + expect(writeFinished).toBe(true); + await expect(read).resolves.toEqual([expect.objectContaining({ values: expect.objectContaining({ title: 'After' }) })]); + }); + }); + + describe('runReadSnapshot', () => { + const scope = { userId: 1 as const, scopeId: '10' }; + const row: OfflineReplicaRow = { + ...scope, + sourceKey: 'test_items', + identity: { kind: 'generated', localId: '019d-snap', remoteId: 7 }, + values: { id: 7, title: 'before' }, + confirmedValues: { id: 7, title: 'before' }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'confirmed', + }; + const command: OfflineCommand = { + ...scope, + commandId: 'cmd-snap', + aggregateType: 'test_items', + sourceKey: 'test_items', + identity: generatedCommandIdentity('019d-snap'), + operation: 'test_items.update', + payload: {}, + optimisticValue: row.values, + payloadHash: 'hash', + baseRevision: 1, + state: 'pending', + attempts: 0, + retryAt: null, + createdAt: 1, + lastErrorCode: null, + }; + + it('複数readを同一snapshotで一貫させ、in-flight writeのtorn readを防ぐ', async () => { + await repository.transactReplica({ putRows: [row], putCommands: [command] }); + + let releaseSnapshot: (() => void) | undefined; + const snapshotGate = new Promise((resolve) => { + releaseSnapshot = resolve; + }); + let writeStarted: (() => void) | undefined; + const writeBegan = new Promise((resolve) => { + writeStarted = resolve; + }); + + const snapshot = repository.runReadSnapshot(async (reader) => { + const commandsBefore = await reader.getCommands(scope); + const write = repository.transactReplica({ + putRows: [{ ...row, values: { id: 7, title: 'after' }, confirmedValues: { id: 7, title: 'after' } }], + putCommands: [{ ...command, state: 'retry_wait', retryAt: 99 }], + }); + void write.then( + () => undefined, + () => undefined, + ); + writeStarted?.(); + await snapshotGate; + const rowsAfterWait = await reader.getReplicaRows(scope, 'test_items'); + const commandsAfterWait = await reader.getCommands(scope); + return { commandsBefore, rowsAfterWait, commandsAfterWait, write }; + }); + + await writeBegan; + await Promise.resolve(); + expect(storage.values.get('offline:outbox:commands')).toEqual( + expect.objectContaining({ + 'cmd-snap': expect.objectContaining({ state: 'pending' }), + }), + ); + + releaseSnapshot?.(); + const observed = await snapshot; + await observed.write; + + expect(observed.commandsBefore).toEqual([expect.objectContaining({ commandId: 'cmd-snap', state: 'pending' })]); + expect(observed.rowsAfterWait).toEqual([expect.objectContaining({ values: expect.objectContaining({ title: 'before' }) })]); + expect(observed.commandsAfterWait).toEqual([expect.objectContaining({ state: 'pending' })]); + await expect(repository.getReplicaRows(scope, 'test_items')).resolves.toEqual([ + expect.objectContaining({ values: expect.objectContaining({ title: 'after' }) }), + ]); + }); + + it('独立した並行snapshotはそれぞれreader leaseを持ち、writerは両方の完了を待つ', async () => { + await repository.transactReplica({ putRows: [row], putCommands: [command] }); + + let releaseA: (() => void) | undefined; + let releaseB: (() => void) | undefined; + const gateA = new Promise((resolve) => { + releaseA = resolve; + }); + const gateB = new Promise((resolve) => { + releaseB = resolve; + }); + let bothReadersReady: (() => void) | undefined; + const readersReady = new Promise((resolve) => { + bothReadersReady = resolve; + }); + let readersHeld = 0; + + const snapshotA = repository.runReadSnapshot(async (reader) => { + const before = await reader.getCommands(scope); + readersHeld += 1; + if (readersHeld === 2) bothReadersReady?.(); + await gateA; + return before; + }); + const snapshotB = repository.runReadSnapshot(async (reader) => { + const before = await reader.getReplicaRows(scope, 'test_items'); + readersHeld += 1; + if (readersHeld === 2) bothReadersReady?.(); + await gateB; + return before; + }); + + await readersReady; + let writeFinished = false; + const write = repository + .transactReplica({ + putRows: [{ ...row, values: { id: 7, title: 'after' }, confirmedValues: { id: 7, title: 'after' } }], + }) + .then(() => { + writeFinished = true; + }); + void write.then( + () => undefined, + () => undefined, + ); + await Promise.resolve(); + await Promise.resolve(); + expect(writeFinished).toBe(false); + + releaseA?.(); + await snapshotA; + await Promise.resolve(); + expect(writeFinished).toBe(false); + + releaseB?.(); + await expect(snapshotB).resolves.toEqual([expect.objectContaining({ values: expect.objectContaining({ title: 'before' }) })]); + await write; + expect(writeFinished).toBe(true); + await expect(repository.getReplicaRows(scope, 'test_items')).resolves.toEqual([ + expect.objectContaining({ values: expect.objectContaining({ title: 'after' }) }), + ]); + }); + + it('write開始前のvalidation失敗ではcommitted snapshotが変わらない', async () => { + await repository.transactReplica({ putRows: [row], putCommands: [command] }); + await expect( + repository.transactReplica({ + putRows: [ + { + ...row, + identity: { kind: 'generated', localId: '019d-snap', remoteId: 8 }, + values: { id: 8, title: 'illegal-rebind' }, + }, + ], + }), + ).rejects.toThrow('Offline replica remoteId is immutable'); + + await expect( + repository.runReadSnapshot(async (reader) => ({ + rows: await reader.getReplicaRows(scope, 'test_items'), + commands: await reader.getCommands(scope), + })), + ).resolves.toEqual({ + rows: [ + expect.objectContaining({ + values: expect.objectContaining({ title: 'before' }), + identity: expect.objectContaining({ remoteId: 7 }), + }), + ], + commands: [expect.objectContaining({ commandId: 'cmd-snap', state: 'pending' })], + }); + }); }); }); diff --git a/projects/kit/offline/src/lib/offline-repository.ts b/projects/kit/offline/src/lib/offline-repository.ts index 71d4762..33a25f8 100644 --- a/projects/kit/offline/src/lib/offline-repository.ts +++ b/projects/kit/offline/src/lib/offline-repository.ts @@ -158,6 +158,42 @@ export interface OfflineReplicaTransaction { removeReconciliationScopes?: readonly OfflineScope[]; } +/** + * Read-only view of durable offline state that cannot observe an in-flight write transaction. + * + * Obtain via {@link OfflineRepository.runReadSnapshot}. Compose multiple reads through this + * handle inside one callback; do not call {@link OfflineRepository.runReadSnapshot} recursively. + */ +export interface OfflineRepositoryReader { + getLastUserId(): Promise; + getSessionManifest(userId: OfflinePrincipalId): Promise; + getReplicaRow( + scope: OfflineScope, + sourceKey: string, + identity: OfflineReplicaAddress, + ): Promise | null>; + getReplicaRowIncludingPendingDelete?( + scope: OfflineScope, + sourceKey: string, + identity: OfflineReplicaAddress, + ): Promise | null>; + getReplicaRows(scope: OfflineScope, sourceKey: string): Promise[]>; + getReplicaRowByRemoteId( + scope: OfflineScope, + sourceKey: string, + remoteId: OfflineGeneratedRemoteId, + ): Promise | null>; + getReplicaRowByRemoteIdentity( + scope: OfflineScope, + sourceKey: string, + identity: OfflineReplicaRemoteIdentity, + ): Promise | null>; + getReplicaCursor(scope: OfflineScope): Promise; + getReconciliationScopes?(userId: OfflinePrincipalId): Promise; + getCommands(scope: OfflineScope): Promise; + getCommandsForUser?(userId: OfflinePrincipalId): Promise; +} + /** Durable local replica and outbox persistence contract. */ export interface OfflineRepository { initialize(): Promise; @@ -197,6 +233,16 @@ export interface OfflineRepository { clearUser(userId: OfflinePrincipalId): Promise; clearScope(scope: OfflineScope): Promise; transactReplica(transaction: OfflineReplicaTransaction): Promise; + /** + * Runs `read` against a committed snapshot that cannot observe an in-flight write. + * + * Each call independently waits for prior writes, then holds a reader lease until `read` + * settles so concurrent writers wait for all active readers. Compose multi-reads only via + * the provided {@link OfflineRepositoryReader}; callers must not invoke `runReadSnapshot` + * recursively from inside `read` (nested support is intentionally absent). Mutating APIs + * must not be called from `read`. + */ + runReadSnapshot(read: (reader: OfflineRepositoryReader) => Promise): Promise; } /** DI token for the selected platform repository. */ @@ -254,6 +300,9 @@ export class IonicOfflineRepository implements OfflineRepository { readonly #options = inject(OFFLINE_KIT_OPTIONS); #initialization: Promise | null = null; #writes: Promise = Promise.resolve(); + #activeReaders = 0; + #readersIdle: Promise = Promise.resolve(); + #resolveReadersIdle: (() => void) | null = null; #rowIndexBuild: Promise | null = null; initialize(): Promise { @@ -267,20 +316,18 @@ export class IonicOfflineRepository implements OfflineRepository { } async getLastUserId(): Promise { - await this.initialize(); - return (await this.#metadata()).lastUserId; + return this.#withCommittedRead(() => this.#readLastUserId()); } async setLastUserId(userId: OfflinePrincipalId): Promise { await this.initialize(); - await this.#storage.set(METADATA_KEY, { ...(await this.#metadata()), lastUserId: userId }); + await this.#enqueueWrite(async () => { + await this.#storage.set(METADATA_KEY, { ...(await this.#metadata()), lastUserId: userId }); + }); } async getSessionManifest(userId: OfflinePrincipalId): Promise { - await this.initialize(); - await this.#writes; - const manifests = await this.#readRecord(SESSION_MANIFESTS_KEY); - return manifests[canonicalOfflinePrincipalId(userId)] ?? null; + return this.#withCommittedRead(() => this.#readSessionManifest(userId)); } async putSessionManifest(userId: OfflinePrincipalId, value: T): Promise { @@ -296,13 +343,7 @@ export class IonicOfflineRepository implements OfflineRepository { sourceKey: string, identity: OfflineReplicaAddress, ): Promise | null> { - await this.initialize(); - await this.#writes; - const schema = this.#resolveReplicaEntitySchema(sourceKey); - 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; + return this.#withCommittedRead(() => this.#readReplicaRow(scope, sourceKey, identity, false)); } async getReplicaRowIncludingPendingDelete( @@ -310,17 +351,81 @@ export class IonicOfflineRepository implements OfflineRepository { sourceKey: string, identity: OfflineReplicaAddress, ): Promise | null> { - await this.initialize(); - await this.#writes; + return this.#withCommittedRead(() => this.#readReplicaRow(scope, sourceKey, identity, true)); + } + + async getReplicaRows(scope: OfflineScope, sourceKey: string): Promise[]> { + return this.#withCommittedRead(() => this.#readReplicaRows(scope, sourceKey)); + } + + async getReplicaRowByRemoteId( + scope: OfflineScope, + sourceKey: string, + remoteId: OfflineGeneratedRemoteId, + ): Promise | null> { + if (this.#resolveReplicaEntitySchema(sourceKey).identity.kind !== 'generated') return null; + return this.getReplicaRowByRemoteIdentity(scope, sourceKey, { remoteId }); + } + + async getReplicaRowByRemoteIdentity( + scope: OfflineScope, + sourceKey: string, + identity: OfflineReplicaRemoteIdentity, + ): Promise | null> { + return this.#withCommittedRead(() => this.#readReplicaRowByRemoteIdentity(scope, sourceKey, identity)); + } + + async getReplicaCursor(scope: OfflineScope): Promise { + return this.#withCommittedRead(() => this.#readReplicaCursor(scope)); + } + + async getReconciliationScopes(userId: OfflinePrincipalId): Promise { + return this.#withCommittedRead(() => this.#readReconciliationScopes(userId)); + } + + async getCommands(scope: OfflineScope): Promise { + return this.#withCommittedRead(() => this.#readCommands(scope)); + } + + async getCommandsForUser(userId: OfflinePrincipalId): Promise { + return this.#withCommittedRead(() => this.#readCommandsForUser(userId)); + } + + async runReadSnapshot(read: (reader: OfflineRepositoryReader) => Promise): Promise { + return this.#withCommittedRead(() => read(this.#reader())); + } + + #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 #readLastUserId(): Promise { + return (await this.#metadata()).lastUserId; + } + + async #readSessionManifest(userId: OfflinePrincipalId): Promise { + const manifests = await this.#readRecord(SESSION_MANIFESTS_KEY); + return manifests[canonicalOfflinePrincipalId(userId)] ?? null; + } + + async #readReplicaRow( + scope: OfflineScope, + sourceKey: string, + identity: OfflineReplicaAddress, + includePendingDelete: boolean, + ): Promise | null> { const schema = this.#resolveReplicaEntitySchema(sourceKey); 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; + if (!row || (!includePendingDelete && (row.visibility ?? 'present') === 'pending_delete')) return null; + return this.#rowForScope(row, schema, scope) as OfflineReplicaRow; } - async getReplicaRows(scope: OfflineScope, sourceKey: string): Promise[]> { - await this.initialize(); - await this.#writes; + async #readReplicaRows(scope: OfflineScope, sourceKey: string): Promise[]> { const schema = this.#resolveReplicaEntitySchema(sourceKey); const rows = await this.#readRowPartition(scope, sourceKey, schema); return Object.values(rows) @@ -333,22 +438,11 @@ export class IonicOfflineRepository implements OfflineRepository { .sort((left, right) => this.#compareReplicaIdentity(schema, left.identity, right.identity)); } - async getReplicaRowByRemoteId( - scope: OfflineScope, - sourceKey: string, - remoteId: OfflineGeneratedRemoteId, - ): Promise | null> { - if (this.#resolveReplicaEntitySchema(sourceKey).identity.kind !== 'generated') return null; - return this.getReplicaRowByRemoteIdentity(scope, sourceKey, { remoteId }); - } - - async getReplicaRowByRemoteIdentity( + async #readReplicaRowByRemoteIdentity( scope: OfflineScope, sourceKey: string, identity: OfflineReplicaRemoteIdentity, ): Promise | null> { - await this.initialize(); - await this.#writes; const schema = this.#resolveReplicaEntitySchema(sourceKey); const canonical = canonicalOfflineRemoteIdentity(schema, identity); const rows = await this.#readRowPartition(scope, sourceKey, schema); @@ -361,24 +455,18 @@ export class IonicOfflineRepository implements OfflineRepository { return row ? (this.#rowForScope(row, schema, scope) as OfflineReplicaRow) : null; } - async getReplicaCursor(scope: OfflineScope): Promise { - await this.initialize(); - await this.#writes; + async #readReplicaCursor(scope: OfflineScope): Promise { const cursors = await this.#readRecord(CURSORS_KEY); const cursor = cursors[this.#cursorKey(scope)]; return cursor === undefined ? null : { ...scope, cursor }; } - async getReconciliationScopes(userId: OfflinePrincipalId): Promise { - await this.initialize(); - await this.#writes; + async #readReconciliationScopes(userId: OfflinePrincipalId): Promise { 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; + async #readCommands(scope: OfflineScope): Promise { const commands = await this.#readRecord(OUTBOX_KEY); return Object.values(commands) .filter((command) => command.userId === scope.userId && command.scopeId === scope.scopeId) @@ -386,9 +474,7 @@ export class IonicOfflineRepository implements OfflineRepository { .sort(compareOfflineCommands); } - async getCommandsForUser(userId: OfflinePrincipalId): Promise { - await this.initialize(); - await this.#writes; + async #readCommandsForUser(userId: OfflinePrincipalId): Promise { const commands = await this.#readRecord(OUTBOX_KEY); return Object.values(commands) .filter((command) => command.userId === userId) @@ -396,12 +482,55 @@ export class IonicOfflineRepository implements OfflineRepository { .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; + #reader(): OfflineRepositoryReader { + return { + getLastUserId: () => this.#readLastUserId(), + getSessionManifest: (userId) => this.#readSessionManifest(userId), + 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), + getReplicaRowByRemoteId: async (scope, sourceKey, remoteId) => { + if (this.#resolveReplicaEntitySchema(sourceKey).identity.kind !== 'generated') return null; + return this.#readReplicaRowByRemoteIdentity(scope, sourceKey, { remoteId }); + }, + getReplicaRowByRemoteIdentity: (scope, sourceKey, identity) => this.#readReplicaRowByRemoteIdentity(scope, sourceKey, identity), + getReplicaCursor: (scope) => this.#readReplicaCursor(scope), + getReconciliationScopes: (userId) => this.#readReconciliationScopes(userId), + getCommands: (scope) => this.#readCommands(scope), + getCommandsForUser: (userId) => this.#readCommandsForUser(userId), + }; + } + + async #withCommittedRead(operation: () => Promise): Promise { + await this.initialize(); + // Finish index readiness first (may enqueue a write). Then await the latest write + // tail and register the reader synchronously — no await gap for a writer to start. + await this.#ensureRowPartitionsReady(); + await this.#writes; + this.#beginReaders(); + try { + return await operation(); + } finally { + this.#endReaders(); + } + } + + #beginReaders(): void { + if (this.#activeReaders === 0) { + this.#readersIdle = new Promise((resolve) => { + this.#resolveReadersIdle = resolve; + }); + } + this.#activeReaders += 1; + } + + #endReaders(): void { + this.#activeReaders -= 1; + if (this.#activeReaders === 0) { + this.#resolveReadersIdle?.(); + this.#resolveReadersIdle = null; + this.#readersIdle = Promise.resolve(); + } } async putCommand(command: OfflineCommand): Promise { @@ -731,14 +860,32 @@ export class IonicOfflineRepository implements OfflineRepository { const cached = await this.#storage.get>>(key); if (cached !== null) return cached; if (await this.#storage.get(ROW_PARTITION_READY_KEY)) return {}; + // Non-mutating fallback under a reader lease (indexes are normally built before the lease). + return this.#legacyRowPartition(scope, sourceKey, schema); + } + + async #legacyRowPartition( + scope: OfflineScope, + sourceKey: string, + schema: OfflineReplicaEntitySchema>, + ): Promise>> { + const rows = await this.#readRecord>(ROWS_KEY); + return 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; + }), + ); + } + + async #ensureRowPartitionsReady(): Promise { + 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 () => { + const run = async (): Promise => { if (await this.#storage.get(ROW_PARTITION_READY_KEY)) return; const rows = await this.#readRecord(ROWS_KEY); const partitions = new Map>(); @@ -751,7 +898,8 @@ export class IonicOfflineRepository implements OfflineRepository { } await Promise.all([...partitions].map(([key, partition]) => this.#storage.set(key, partition))); await this.#storage.set(ROW_PARTITION_READY_KEY, true); - }); + }; + const build = this.#enqueueWrite(run); this.#rowIndexBuild = build.finally(() => { this.#rowIndexBuild = null; }); @@ -824,7 +972,10 @@ export class IonicOfflineRepository implements OfflineRepository { } #enqueueWrite(operation: () => Promise): Promise { - const write = this.#writes.then(operation); + const write = this.#writes.then(async () => { + if (this.#activeReaders > 0) await this.#readersIdle; + return operation(); + }); this.#writes = write.then( () => undefined, () => undefined, 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 ac65c04..f2a5b37 100644 --- a/projects/kit/offline/src/lib/offline-sync.service.spec.ts +++ b/projects/kit/offline/src/lib/offline-sync.service.spec.ts @@ -10,7 +10,7 @@ import { 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'; +import { OfflineReplicaPullService, OfflineReplicaSchemaMismatchError } from './offline-replica-pull.service'; import { OfflineReplicaMutationCoordinator } from './offline-replica-mutation-coordinator'; import { defineOfflineReplicaSchema, @@ -36,6 +36,8 @@ import { OfflineCommandInFlightError, OfflinePayloadValidationError, OfflineSyncService, + OFFLINE_RETRY_RANDOM, + offlineRetryDelayMs, type PreparedOfflineCommand, } from './offline-sync.service'; @@ -283,6 +285,8 @@ describe('OfflineSyncService', () => { withoutServerRevision: (command: OfflineCommand) => ({ ...command, baseRevision: null }), }, }, + // Fixed sample so backoff stays deterministic (except dedicated jitter unit tests). + { provide: OFFLINE_RETRY_RANDOM, useValue: () => 0.5 }, ], }); service = TestBed.inject(OfflineSyncService); @@ -1123,6 +1127,655 @@ describe('OfflineSyncService', () => { expect(reconciliationScopes).toEqual([]); }); + it('pre-pull: 無関係なscope A失敗でも成功したscope Bのeligible aggregateは送信する', async () => { + session = { + userId: 1, + scopes: [ + { userId: 1, scopeId: '10' }, + { userId: 1, scopeId: '20' }, + ], + }; + const scopeAError = new Error('scope A pre-pull failed'); + pull.mockImplementation(async (scope) => { + if (scope.scopeId === '10') throw scopeAError; + }); + await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'scope-a' }, + operation: 'documents.create', + payload: { title: 'a' }, + optimisticValue: { id: 0, title: 'a' }, + }, + { flush: false }, + ); + await service.enqueue( + { + scopeId: '20', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'scope-b' }, + operation: 'documents.create', + payload: { title: 'b' }, + optimisticValue: { id: 0, title: 'b' }, + }, + { flush: false }, + ); + + connected.set(true); + await expect(service.flush()).rejects.toBe(scopeAError); + + expect(execute).toHaveBeenCalledOnce(); + expect(execute.mock.calls.map(([command]) => (command as OfflineCommand).identity)).toEqual([ + expect.objectContaining({ localId: 'scope-b' }), + ]); + expect(commands.map((command) => command.identity)).toEqual([expect.objectContaining({ localId: 'scope-a' })]); + }); + + it('pre-pull: 同一scopeのpull失敗ではそのscopeのcommandを送らない', async () => { + const scopeError = new Error('same scope pre-pull failed'); + pull.mockRejectedValue(scopeError); + await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'blocked-by-pull' }, + operation: 'documents.create', + payload: { title: 'blocked' }, + optimisticValue: { id: 0, title: 'blocked' }, + }, + { flush: false }, + ); + + connected.set(true); + await expect(service.flush()).rejects.toBe(scopeError); + expect(execute).not.toHaveBeenCalled(); + expect(commands).toEqual([expect.objectContaining({ identity: expect.objectContaining({ localId: 'blocked-by-pull' }) })]); + }); + + it('pre-pull失敗があってもsend workerが全てsettledしてからflushがrejectする', async () => { + session = { + userId: 1, + scopes: [ + { userId: 1, scopeId: '10' }, + { userId: 1, scopeId: '20' }, + ], + }; + const scopeAError = new Error('scope A pre-pull failed after workers'); + let releaseSend: (() => void) | undefined; + const sendGate = new Promise((resolve) => { + releaseSend = resolve; + }); + let sendStarted: (() => void) | undefined; + const started = new Promise((resolve) => { + sendStarted = resolve; + }); + pull.mockImplementation(async (scope) => { + if (scope.scopeId === '10') throw scopeAError; + }); + execute.mockImplementation(async () => { + sendStarted?.(); + await sendGate; + return { response: null }; + }); + await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'scope-a-wait' }, + operation: 'documents.create', + payload: { title: 'a' }, + optimisticValue: { id: 0, title: 'a' }, + }, + { flush: false }, + ); + await service.enqueue( + { + scopeId: '20', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'scope-b-wait' }, + operation: 'documents.create', + payload: { title: 'b' }, + optimisticValue: { id: 0, title: 'b' }, + }, + { flush: false }, + ); + + connected.set(true); + const flush = service.flush(); + const flushRejected = expect(flush).rejects.toBe(scopeAError); + await started; + expect(execute).toHaveBeenCalledOnce(); + releaseSend?.(); + await flushRejected; + expect(execute).toHaveBeenCalledOnce(); + expect(commands.map((command) => command.identity)).toEqual([expect.objectContaining({ localId: 'scope-a-wait' })]); + }); + + it('status無しworker失敗でも他workerのACK完了までflushをrejectしない', async () => { + session = { + userId: 1, + scopes: [ + { userId: 1, scopeId: '10' }, + { userId: 1, scopeId: '20' }, + ], + }; + const statuslessFailure = new Error('status-less transport failure'); + let releaseSuccess!: () => void; + const successGate = new Promise((resolve) => { + releaseSuccess = resolve; + }); + let successStarted!: () => void; + const started = new Promise((resolve) => { + successStarted = resolve; + }); + execute.mockImplementation(async (command) => { + if (command.identity.kind === 'generated' && command.identity.localId === 'fail-early') { + throw statuslessFailure; + } + successStarted(); + await successGate; + return { response: null, serverRevision: 2 }; + }); + await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'fail-early' }, + operation: 'documents.create', + payload: { title: 'fail' }, + optimisticValue: { id: 0, title: 'fail' }, + }, + { flush: false }, + ); + await service.enqueue( + { + scopeId: '20', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'succeed-deferred' }, + operation: 'documents.create', + payload: { title: 'ok' }, + optimisticValue: { id: 0, title: 'ok' }, + }, + { flush: false }, + ); + + connected.set(true); + const flush = service.flush(); + const flushRejected = expect(flush).rejects.toBe(statuslessFailure); + await started; + expect(commands.some((command) => command.identity.kind === 'generated' && command.identity.localId === 'succeed-deferred')).toBe(true); + releaseSuccess(); + await flushRejected; + expect(commands.map((command) => command.identity)).toEqual([expect.objectContaining({ localId: 'fail-early' })]); + expect(commands[0]).toMatchObject({ state: 'retry_wait', serverCommitUnknown: true }); + expect(service.syncState()).toBe('attention'); + expect(pull.mock.calls.some((call) => call[0]?.scopeId === '20')).toBe(true); + }); + + it('typed schema mismatchのpre-pull fatalでは残りscopeを止め成功scopeも送らない', async () => { + session = { + userId: 1, + scopes: [ + { userId: 1, scopeId: '10' }, + { userId: 1, scopeId: '20' }, + ], + }; + const schemaError = new OfflineReplicaSchemaMismatchError(1, 'abc', 2, 'def'); + pull.mockImplementation(async (scope) => { + if (scope.scopeId === '10') throw schemaError; + }); + await service.enqueue( + { + scopeId: '20', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'scope-b-fatal' }, + operation: 'documents.create', + payload: { title: 'b' }, + optimisticValue: { id: 0, title: 'b' }, + }, + { flush: false }, + ); + + connected.set(true); + await expect(service.flush()).rejects.toBe(schemaError); + expect(execute).not.toHaveBeenCalled(); + expect(pull.mock.calls.map((call) => call[0]?.scopeId)).toEqual(['10']); + }); + + it('pre-pull HTTP 409はschema mismatch fatalとして残りscopeを止め成功scopeも送らない', async () => { + session = { + userId: 1, + scopes: [ + { userId: 1, scopeId: '10' }, + { userId: 1, scopeId: '20' }, + ], + }; + const schemaConflict = { status: 409, message: 'Conflict' }; + pull.mockImplementation(async (scope) => { + if (scope.scopeId === '10') throw schemaConflict; + }); + await service.enqueue( + { + scopeId: '20', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'scope-b-http-409' }, + operation: 'documents.create', + payload: { title: 'b' }, + optimisticValue: { id: 0, title: 'b' }, + }, + { flush: false }, + ); + + connected.set(true); + await expect(service.flush()).rejects.toBe(schemaConflict); + expect(execute).not.toHaveBeenCalled(); + expect(pull.mock.calls.map((call) => call[0]?.scopeId)).toEqual(['10']); + }); + + it.each([ + ['OfflineReplicaSchemaMismatchError', () => new OfflineReplicaSchemaMismatchError(1, 'abc', 2, 'def')], + ['HTTP 401', () => ({ status: 401, message: 'Unauthorized' })], + ['HTTP 403', () => ({ status: 403, message: 'Forbidden' })], + ['HTTP 409', () => ({ status: 409, message: 'Conflict' })], + ] as const)('pre-pull fatal (%s) は1s自動retryせずpending post-pullもスキップする', async (_label, createError) => { + vi.useFakeTimers(); + try { + 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 fatal'); + const fatalError = createError(); + pull.mockImplementation(async (scope) => { + if (scope.scopeId === '20') { + scope20Pulls += 1; + // First full flush: pre-pull ok, post-send pull fails and leaves pending marker. + if (scope20Pulls === 2) throw postPullError; + } + }); + await service.enqueue( + { + scopeId: '20', + aggregateType: 'documents', + identity: { kind: 'generated', localId: `fatal-skip-post-${_label.replace(/\s+/g, '-')}` }, + operation: 'documents.create', + payload: { title: 'seed' }, + optimisticValue: { id: 0, title: 'seed' }, + }, + { flush: false }, + ); + connected.set(true); + await expect(service.flush()).rejects.toBe(postPullError); + expect(execute).toHaveBeenCalledOnce(); + expect(commands).toEqual([]); + // Drop the transient post-pull retry so this case only asserts fatal does not arm a new one. + vi.clearAllTimers(); + + pull.mockImplementation(async (scope) => { + if (scope.scopeId === '10') throw fatalError; + }); + const pullsBeforeFatal = pull.mock.calls.length; + await expect(service.flush()).rejects.toBe(fatalError); + expect(execute).toHaveBeenCalledOnce(); + expect(pull.mock.calls.slice(pullsBeforeFatal).map((call) => call[0]?.scopeId)).toEqual(['10']); + + const pullsAfterFatal = pull.mock.calls.length; + await vi.advanceTimersByTimeAsync(2_000); + expect(pull.mock.calls.length).toBe(pullsAfterFatal); + expect(handleError).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + } + }); + + it.each([ + ['OfflineReplicaSchemaMismatchError', () => new OfflineReplicaSchemaMismatchError(1, 'abc', 2, 'def')], + ['HTTP 401', () => ({ status: 401, message: 'Unauthorized' })], + ['HTTP 403', () => ({ status: 403, message: 'Forbidden' })], + ['HTTP 409', () => ({ status: 409, message: 'Conflict' })], + ] as const)('post-send pull fatal (%s) は残りpending post-pullを止めACK保持・1s自動retryなし', async (_label, createError) => { + vi.useFakeTimers(); + const setTimeoutSpy = vi.spyOn(globalThis, 'setTimeout'); + try { + session = { + userId: 1, + scopes: [ + { userId: 1, scopeId: '10' }, + { userId: 1, scopeId: '20' }, + ], + }; + const fatalError = createError(); + const pullsByScope = new Map(); + pull.mockImplementation(async (scope) => { + const count = (pullsByScope.get(scope.scopeId) ?? 0) + 1; + pullsByScope.set(scope.scopeId, count); + // Second pull for a scope is the post-send pull after ACK. + if (count >= 2) throw fatalError; + }); + await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: `post-fatal-a-${_label.replace(/\s+/g, '-')}` }, + operation: 'documents.create', + payload: { title: 'a' }, + optimisticValue: { id: 0, title: 'a' }, + }, + { flush: false }, + ); + await service.enqueue( + { + scopeId: '20', + aggregateType: 'documents', + identity: { kind: 'generated', localId: `post-fatal-b-${_label.replace(/\s+/g, '-')}` }, + operation: 'documents.create', + payload: { title: 'b' }, + optimisticValue: { id: 0, title: 'b' }, + }, + { flush: false }, + ); + + setTimeoutSpy.mockClear(); + connected.set(true); + await expect(service.flush()).rejects.toBe(fatalError); + + // Both commands ACKed (removed); no resend path. + expect(execute).toHaveBeenCalledTimes(2); + expect(commands).toEqual([]); + expect(service.pendingCount()).toBe(0); + // Two pending post-pull scopes: first fatal stops the second immediately. + expect(pullsByScope.get('10')).toBeGreaterThanOrEqual(1); + expect(pullsByScope.get('20')).toBeGreaterThanOrEqual(1); + const postPullScopes = [...pullsByScope.entries()].filter(([, count]) => count >= 2).map(([scopeId]) => scopeId); + expect(postPullScopes).toHaveLength(1); + expect([...pullsByScope.values()].reduce((sum, count) => sum + count, 0)).toBe(3); + // Reconciliation markers remain for later auth/upgrade recovery. + expect(reconciliationScopes).toEqual( + expect.arrayContaining([ + { userId: 1, scopeId: '10' }, + { userId: 1, scopeId: '20' }, + ]), + ); + expect(reconciliationScopes).toHaveLength(2); + // Fatal must not arm the 1s automatic post-pull flush retry. + expect(setTimeoutSpy.mock.calls.some(([, delay]) => delay === 1_000)).toBe(false); + + const pullsAfterFatal = pull.mock.calls.length; + // Drop unrelated scheduler/effect timers, then prove no 1s retry remained. + vi.clearAllTimers(); + await vi.advanceTimersByTimeAsync(2_000); + expect(pull.mock.calls.length).toBe(pullsAfterFatal); + expect(execute).toHaveBeenCalledTimes(2); + } finally { + setTimeoutSpy.mockRestore(); + vi.useRealTimers(); + } + }); + + it('post-send pull transientの後のfatalはfatalを優先して投げ残りpendingを止める', async () => { + vi.useFakeTimers(); + const setTimeoutSpy = vi.spyOn(globalThis, 'setTimeout'); + try { + session = { + userId: 1, + scopes: [ + { userId: 1, scopeId: '10' }, + { userId: 1, scopeId: '20' }, + { userId: 1, scopeId: '30' }, + ], + }; + const transient = new Error('post-send transient'); + const fatal = new OfflineReplicaSchemaMismatchError(1, 'abc', 2, 'def'); + const pullsByScope = new Map(); + let postPullAttempts = 0; + pull.mockImplementation(async (scope) => { + const count = (pullsByScope.get(scope.scopeId) ?? 0) + 1; + pullsByScope.set(scope.scopeId, count); + if (count < 2) return; + postPullAttempts += 1; + if (postPullAttempts === 1) throw transient; + throw fatal; + }); + for (const [scopeId, localId] of [ + ['10', 'post-prefer-a'], + ['20', 'post-prefer-b'], + ['30', 'post-prefer-c'], + ] as const) { + await service.enqueue( + { + scopeId, + aggregateType: 'documents', + identity: { kind: 'generated', localId }, + operation: 'documents.create', + payload: { title: localId }, + optimisticValue: { id: 0, title: localId }, + }, + { flush: false }, + ); + } + + setTimeoutSpy.mockClear(); + connected.set(true); + await expect(service.flush()).rejects.toBe(fatal); + expect(execute).toHaveBeenCalledTimes(3); + expect(commands).toEqual([]); + expect(postPullAttempts).toBe(2); + expect([...pullsByScope.values()].filter((count) => count >= 2)).toHaveLength(2); + expect([...pullsByScope.values()].filter((count) => count === 1)).toHaveLength(1); + expect(reconciliationScopes).toHaveLength(3); + expect(setTimeoutSpy.mock.calls.some(([, delay]) => delay === 1_000)).toBe(false); + + const pullsAfterFatal = pull.mock.calls.length; + vi.clearAllTimers(); + await vi.advanceTimersByTimeAsync(2_000); + expect(pull.mock.calls.length).toBe(pullsAfterFatal); + } finally { + setTimeoutSpy.mockRestore(); + vi.useRealTimers(); + } + }); + + it('遅延した旧世代のpost-pull fatalは新世代のretry_wait timerを消さない', async () => { + vi.useFakeTimers(); + try { + let releasePostPull!: (error: unknown) => void; + let postPullStarted!: () => void; + const postPullEntered = new Promise((resolve) => { + postPullStarted = resolve; + }); + const postPullGate = new Promise((_resolve, reject) => { + releasePostPull = (error) => reject(error); + }); + // Avoid unhandled rejection if the gate is abandoned mid-test. + void postPullGate.catch(() => undefined); + + const pullsByScope = new Map(); + pull.mockImplementation(async (scope) => { + const count = (pullsByScope.get(`${scope.userId}:${scope.scopeId}`) ?? 0) + 1; + pullsByScope.set(`${scope.userId}:${scope.scopeId}`, count); + // Session A post-send pull stays pending until we release the fatal. + if (scope.userId === 1 && scope.scopeId === '10' && count >= 2) { + postPullStarted(); + await postPullGate; + } + }); + + await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'stale-fatal-a' }, + operation: 'documents.create', + payload: { title: 'a' }, + optimisticValue: { id: 0, title: 'a' }, + }, + { flush: false }, + ); + connected.set(true); + const flushA = service.flush(); + await postPullEntered; + + // Transition generation without waiting for A's deferred post-pull to settle. + service.revokeSession(); + commands = commands.filter((command) => command.userId !== 1); + rows = rows.filter((row) => row.userId !== 1); + session = { userId: 2, scopes: [{ userId: 2, scopeId: '20' }] }; + // Stay offline during session B activation so refreshSession does not start a + // background flush that would swallow the subsequent explicit flush(). + connected.set(false); + await service.refreshSession(); + + execute.mockRejectedValueOnce({ status: 500 }).mockResolvedValue({ response: null }); + await service.enqueue( + { + scopeId: '20', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'stale-fatal-b' }, + operation: 'documents.create', + payload: { title: 'b' }, + optimisticValue: { id: 0, title: 'b' }, + }, + { flush: false }, + ); + connected.set(true); + await service.flush(); + expect(service.pendingCommands()[0]).toMatchObject({ + state: 'retry_wait', + retryAt: expect.any(Number), + identity: { kind: 'generated', localId: 'stale-fatal-b' }, + }); + const executesBeforeRetry = execute.mock.calls.length; + + const fatal = new OfflineReplicaSchemaMismatchError(1, 'abc', 2, 'def'); + releasePostPull(fatal); + await expect(flushA).rejects.toBe(fatal); + // Stale fatal must settle/reject without clearing B's armed retry timer. + expect(service.pendingCommands()[0]).toMatchObject({ + state: 'retry_wait', + identity: { kind: 'generated', localId: 'stale-fatal-b' }, + }); + + await vi.advanceTimersByTimeAsync(2_000); + await Promise.resolve(); + await Promise.resolve(); + expect(execute.mock.calls.length).toBeGreaterThan(executesBeforeRetry); + expect(service.pendingCount()).toBe(0); + expect(commands.some((command) => command.identity.kind === 'generated' && command.identity.localId === 'stale-fatal-b')).toBe(false); + } finally { + vi.useRealTimers(); + } + }); + + it('pre-pull transient失敗はscope隔離し1s自動retryをスケジュールする', async () => { + vi.useFakeTimers(); + try { + session = { + userId: 1, + scopes: [ + { userId: 1, scopeId: '10' }, + { userId: 1, scopeId: '20' }, + ], + }; + const transient = new Error('scope 10 transient pre-pull'); + let scope10Pulls = 0; + pull.mockImplementation(async (scope) => { + if (scope.scopeId === '10' && ++scope10Pulls === 1) throw transient; + }); + await service.enqueue( + { + scopeId: '20', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'transient-isolated' }, + operation: 'documents.create', + payload: { title: 'b' }, + optimisticValue: { id: 0, title: 'b' }, + }, + { flush: false }, + ); + + connected.set(true); + await expect(service.flush()).rejects.toBe(transient); + expect(execute).toHaveBeenCalledOnce(); + expect(pull.mock.calls.map((call) => call[0]?.scopeId)).toEqual(expect.arrayContaining(['10', '20'])); + + const pullsBeforeRetry = pull.mock.calls.length; + await vi.advanceTimersByTimeAsync(1_000); + await Promise.resolve(); + await Promise.resolve(); + expect(pull.mock.calls.length).toBeGreaterThan(pullsBeforeRetry); + } finally { + vi.useRealTimers(); + } + }); + + it('pre-pull transientの後のfatalはfatalを優先して投げ残りscopeを止める', async () => { + session = { + userId: 1, + scopes: [ + { userId: 1, scopeId: '10' }, + { userId: 1, scopeId: '20' }, + { userId: 1, scopeId: '30' }, + ], + }; + const transient = new Error('scope 10 transient'); + const fatal = new OfflineReplicaSchemaMismatchError(1, 'abc', 2, 'def'); + pull.mockImplementation(async (scope) => { + if (scope.scopeId === '10') throw transient; + if (scope.scopeId === '20') throw fatal; + }); + await service.enqueue( + { + scopeId: '30', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'prefer-fatal' }, + operation: 'documents.create', + payload: { title: 'c' }, + optimisticValue: { id: 0, title: 'c' }, + }, + { flush: false }, + ); + + connected.set(true); + await expect(service.flush()).rejects.toBe(fatal); + expect(execute).not.toHaveBeenCalled(); + expect(pull.mock.calls.map((call) => call[0]?.scopeId)).toEqual(['10', '20']); + }); + + it('英語messageだけのgeneric Errorはpre-pull fatalにしない', async () => { + session = { + userId: 1, + scopes: [ + { userId: 1, scopeId: '10' }, + { userId: 1, scopeId: '20' }, + ], + }; + const lookalike = new Error('Offline replica schema mismatch: client=1/abc, server=2/def.'); + pull.mockImplementation(async (scope) => { + if (scope.scopeId === '10') throw lookalike; + }); + await service.enqueue( + { + scopeId: '20', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'scope-b-lookalike' }, + operation: 'documents.create', + payload: { title: 'b' }, + optimisticValue: { id: 0, title: 'b' }, + }, + { flush: false }, + ); + + connected.set(true); + await expect(service.flush()).rejects.toBe(lookalike); + expect(execute).toHaveBeenCalledOnce(); + expect(execute.mock.calls[0]?.[0]).toMatchObject({ identity: { localId: 'scope-b-lookalike' } }); + }); + it('所属から外れたdurable reconciliation scopeをsession discoveryで破棄する', async () => { reconciliationScopes = [{ userId: 1, scopeId: '20' }]; session = { userId: 1, scopes: [{ userId: 1, scopeId: '10' }] }; @@ -1195,13 +1848,50 @@ describe('OfflineSyncService', () => { expect(commands.every((command) => !('remoteId' in command))).toBe(true); }); - it('session scope発見後に前回起動のsending commandをpendingへ復旧する', async () => { + it('session scope発見後に前回起動のsending commandをpendingへ復旧する', async () => { + session = null; + rows.push({ + userId: 1, + scopeId: '10', + sourceKey: 'documents', + identity: { kind: 'generated', localId: '019d-aaaa', remoteId: null }, + values: {}, + confirmedValues: null, + serverRevision: null, + fetchedAt: 1, + syncState: 'pending', + }); + commands.push({ + userId: 1, + scopeId: '10', + commandId: 'interrupted', + aggregateType: 'documents', + sourceKey: 'documents', + identity: { kind: 'generated', localId: '019d-aaaa' }, + operation: 'documents.create', + payload: {}, + optimisticValue: {}, + payloadHash: 'hash', + baseRevision: null, + state: 'sending', + attempts: 1, + retryAt: null, + createdAt: 1, + lastErrorCode: null, + }); + await service.initialize(); + session = { userId: 1, scopes: [{ userId: 1, scopeId: '10' }] }; + await service.refreshSession(); + expect(service.pendingCommands()[0]).toMatchObject({ state: 'pending', serverCommitUnknown: true }); + }); + + it('restart正規化のpending+serverCommitUnknownはattentionでdiscard禁止かつretry UI対象', async () => { session = null; rows.push({ userId: 1, scopeId: '10', sourceKey: 'documents', - identity: { kind: 'generated', localId: '019d-aaaa', remoteId: null }, + identity: { kind: 'generated', localId: '019d-restart-unknown', remoteId: null }, values: {}, confirmedValues: null, serverRevision: null, @@ -1211,10 +1901,10 @@ describe('OfflineSyncService', () => { commands.push({ userId: 1, scopeId: '10', - commandId: 'interrupted', + commandId: 'restart-unknown', aggregateType: 'documents', sourceKey: 'documents', - identity: { kind: 'generated', localId: '019d-aaaa' }, + identity: { kind: 'generated', localId: '019d-restart-unknown' }, operation: 'documents.create', payload: {}, optimisticValue: {}, @@ -1229,7 +1919,16 @@ describe('OfflineSyncService', () => { await service.initialize(); session = { userId: 1, scopes: [{ userId: 1, scopeId: '10' }] }; await service.refreshSession(); + expect(service.pendingCommands()[0]).toMatchObject({ state: 'pending', serverCommitUnknown: true }); + expect(service.syncState()).toBe('attention'); + await expect(service.discard('restart-unknown', { flush: false })).rejects.toBeInstanceOf(OfflineCommandInFlightError); + await expect(service.discardAllPending()).rejects.toBeInstanceOf(OfflineCommandInFlightError); + + execute.mockResolvedValueOnce({ response: null }); + connected.set(true); + await service.retryNow('restart-unknown'); + expect(execute).toHaveBeenCalledOnce(); }); it('未同期createを破棄するとoutboxと未確定replica rowを同時に除く', async () => { @@ -1921,6 +2620,196 @@ describe('OfflineSyncService', () => { expect(rows[0]?.syncState).toBe(rowSyncState); }); + it('retry_waitかつserverCommitUnknownのcommandはattentionとして見える', async () => { + execute.mockRejectedValueOnce({ status: 500 }); + await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'ambiguous-retry' }, + operation: 'documents.upsert', + payload: {}, + optimisticValue: {}, + }, + { flush: false }, + ); + connected.set(true); + await service.flush(); + expect(service.pendingCommands()[0]).toMatchObject({ state: 'retry_wait', serverCommitUnknown: true }); + expect(service.syncState()).toBe('attention'); + }); + + it('pendingかつserverCommitUnknownのcommandもattentionとして見える', async () => { + await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'pending-unknown' }, + operation: 'documents.upsert', + payload: {}, + optimisticValue: {}, + }, + { flush: false }, + ); + const current = commands[0]!; + commands[0] = { ...current, state: 'pending', serverCommitUnknown: true }; + await service.reloadPendingCommands(); + expect(service.syncState()).toBe('attention'); + await expect(service.discard(current.commandId, { flush: false })).rejects.toBeInstanceOf(OfflineCommandInFlightError); + }); + + it('serverCommitUnknownでないretry_waitはpendingのままattentionにしない', async () => { + await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'safe-retry' }, + operation: 'documents.upsert', + payload: {}, + optimisticValue: {}, + }, + { flush: false }, + ); + const current = commands[0]!; + commands[0] = { + ...current, + state: 'retry_wait', + retryAt: Date.now() + 60_000, + serverCommitUnknown: false, + }; + await service.reloadPendingCommands(); + expect(service.syncState()).toBe('pending'); + }); + + it('retry delayはequal jitterで注入可能な乱数を使う', async () => { + TestBed.resetTestingModule(); + const random = vi.fn(() => 0.25); + // Rebuild the standard providers from beforeEach with an injectable random source. + connected = signal(false); + session = { userId: 1, scopes: [{ userId: 1, scopeId: '10' }] }; + commands = []; + rows = []; + pull = vi.fn(async () => undefined); + execute.mockReset(); + execute.mockRejectedValueOnce({ status: 500 }); + const repository = { + initialize: vi.fn(async () => undefined), + getCommands: vi.fn(async (scope: OfflineScope) => + commands.filter((item) => item.userId === scope.userId && item.scopeId === scope.scopeId), + ), + getCommandsForUser: vi.fn(async (userId: number) => commands.filter((item) => item.userId === userId)), + putCommand: vi.fn(async (command: OfflineCommand) => { + commands = commands.filter((item) => item.commandId !== command.commandId); + commands.push(structuredClone(command)); + }), + replaceCommand: vi.fn(async (command: OfflineCommand) => { + commands = commands.filter((item) => item.commandId !== command.commandId); + commands.push(structuredClone(command)); + }), + removeCommand: vi.fn(async (commandId: string) => { + commands = commands.filter((item) => item.commandId !== commandId); + }), + getReplicaRow: vi.fn(async (scope: OfflineScope, sourceKey: string, identity: OfflineReplicaAddress) => { + return ( + rows.find((item) => { + if (item.userId !== scope.userId || item.scopeId !== scope.scopeId || item.sourceKey !== sourceKey) return false; + if (identity.kind === 'generated') { + return item.identity.kind === 'generated' && item.identity.localId === identity.localId; + } + return false; + }) ?? null + ); + }), + getReplicaRowIncludingPendingDelete: vi.fn(async (scope: OfflineScope, sourceKey: string, identity: OfflineReplicaAddress) => { + return ( + rows.find((item) => { + if (item.userId !== scope.userId || item.scopeId !== scope.scopeId || item.sourceKey !== sourceKey) return false; + if (identity.kind === 'generated') { + return item.identity.kind === 'generated' && item.identity.localId === identity.localId; + } + return false; + }) ?? null + ); + }), + getReplicaRowByRemoteId: vi.fn(async () => null), + getReplicaRowByRemoteIdentity: vi.fn(async () => null), + getReplicaCursor: vi.fn(async () => null), + getReconciliationScopes: vi.fn(async () => []), + transactReplica: vi.fn(async (transaction) => { + for (const command of transaction.putCommands ?? []) { + commands = commands.filter((item) => item.commandId !== command.commandId); + commands.push(structuredClone(command)); + } + for (const row of transaction.putRows ?? []) { + rows = rows.filter( + (item) => + item.userId !== row.userId || + item.scopeId !== row.scopeId || + item.sourceKey !== row.sourceKey || + canonicalOfflineReplicaIdentity(item.identity) !== canonicalOfflineReplicaIdentity(row.identity), + ); + rows.push(structuredClone(row)); + } + commands = commands.filter((command) => !(transaction.removeCommandIds ?? []).includes(command.commandId)); + }), + } as unknown as OfflineRepository; + TestBed.configureTestingModule({ + providers: [ + OfflineSyncService, + { provide: OFFLINE_REPOSITORY, useValue: repository }, + { provide: OfflineNetworkService, useValue: { connected } }, + { provide: OFFLINE_KIT_OPTIONS, useValue: { databaseName: 'test-offline', replicaSchema } }, + { provide: OfflineReplicaPullService, useValue: { pull } }, + { provide: ErrorHandler, useValue: { handleError: vi.fn() } }, + { provide: OFFLINE_RETRY_RANDOM, useValue: random }, + { + provide: OFFLINE_COMMAND_HOOKS, + useValue: { entityType: (command: OfflineCommand) => command.aggregateType }, + }, + { + provide: OFFLINE_SYNC_CONTEXT, + useValue: { + getLocalSession: vi.fn(async () => session), + getSession: vi.fn(async () => session), + }, + }, + { + provide: OFFLINE_COMMAND_EXECUTOR, + useValue: { + execute, + provesCommandNotCommitted: () => false, + withServerRevision: (command: OfflineCommand) => command, + }, + }, + ], + }); + service = TestBed.inject(OfflineSyncService); + await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'jitter' }, + operation: 'documents.upsert', + payload: {}, + optimisticValue: {}, + }, + { flush: false }, + ); + connected.set(true); + await service.flush(); + expect(random).toHaveBeenCalled(); + const retryAt = service.pendingCommands()[0]?.retryAt; + expect(retryAt).toEqual(expect.any(Number)); + expect(retryAt! - Date.now()).toBeLessThanOrEqual(offlineRetryDelayMs(1, () => 0.25)); + }); + + it('offlineRetryDelayMsは[⌊cap/2⌋, cap)のequal jitterを返す', () => { + expect(offlineRetryDelayMs(1, () => 0)).toBe(500); + expect(offlineRetryDelayMs(1, () => 0.999)).toBe(999); + expect(offlineRetryDelayMs(3, () => 0.5)).toBe(3000); + expect(() => offlineRetryDelayMs(1, () => 1)).toThrow('must return a number in [0, 1)'); + }); + it('retryNowは未来のbackoffを解除して選択したcommandを直ちに再送する', async () => { execute.mockRejectedValueOnce({ status: 500 }).mockResolvedValueOnce({ response: null }); const commandId = await service.enqueue( @@ -3679,6 +4568,7 @@ describe('OfflineSyncService', () => { }), }, }, + { provide: OFFLINE_RETRY_RANDOM, useValue: () => 0.5 }, ], }); service = TestBed.inject(OfflineSyncService); @@ -3755,6 +4645,151 @@ describe('OfflineSyncService', () => { expect(service.pendingCount()).toBe(0); }); + it('pre-pull: head scope成功/後続scope失敗の同一user-scoped aggregateは成功prefixだけ送る', async () => { + const scope11Error = new Error('scope 11 pre-pull failed'); + pull.mockImplementation(async (scope) => { + if (scope.scopeId === '11') throw scope11Error; + }); + await service.enqueue( + { + scopeId: '10', + aggregateType: 'test_items', + identity: { kind: 'generated', localId: '019d-user-item' }, + operation: 'test_items.update', + payload: { title: 'A' }, + optimisticValue: { id: 42, title: 'A' }, + baseRevision: 1, + }, + { flush: false }, + ); + await service.enqueue( + { + scopeId: '11', + aggregateType: 'test_items', + identity: { kind: 'generated', localId: '019d-user-item' }, + operation: 'test_items.update', + payload: { title: 'B' }, + optimisticValue: { id: 42, title: 'B' }, + baseRevision: 1, + }, + { flush: false }, + ); + + connected.set(true); + await expect(service.flush()).rejects.toBe(scope11Error); + + expect(execute).toHaveBeenCalledOnce(); + expect(execute.mock.calls[0]?.[0]).toMatchObject({ scopeId: '10', payload: { title: 'A' } }); + expect(commands).toHaveLength(1); + expect(commands[0]).toMatchObject({ scopeId: '11', optimisticValue: { title: 'B' }, state: 'pending' }); + }); + + it('pre-pull: head scope失敗/後続scope成功の同一user-scoped aggregateは一切送らない', async () => { + const scope10Error = new Error('scope 10 pre-pull failed'); + pull.mockImplementation(async (scope) => { + if (scope.scopeId === '10') throw scope10Error; + }); + await service.enqueue( + { + scopeId: '10', + aggregateType: 'test_items', + identity: { kind: 'generated', localId: '019d-user-item' }, + operation: 'test_items.update', + payload: { title: 'A' }, + optimisticValue: { id: 42, title: 'A' }, + baseRevision: 1, + }, + { flush: false }, + ); + await service.enqueue( + { + scopeId: '11', + aggregateType: 'test_items', + identity: { kind: 'generated', localId: '019d-user-item' }, + operation: 'test_items.update', + payload: { title: 'B' }, + optimisticValue: { id: 42, title: 'B' }, + baseRevision: 1, + }, + { flush: false }, + ); + + connected.set(true); + await expect(service.flush()).rejects.toBe(scope10Error); + + expect(execute).not.toHaveBeenCalled(); + expect(commands.map((command) => command.scopeId)).toEqual(['10', '11']); + }); + + it('pre-pull: 成功prefix送信後、両scope成功の次flushは残りのBだけ送る', async () => { + const scope11Error = new Error('scope 11 pre-pull failed once'); + pull.mockImplementation(async (scope) => { + if (scope.scopeId === '11') throw scope11Error; + }); + await service.enqueue( + { + scopeId: '10', + aggregateType: 'test_items', + identity: { kind: 'generated', localId: '019d-user-item' }, + operation: 'test_items.update', + payload: { title: 'A' }, + optimisticValue: { id: 42, title: 'A' }, + baseRevision: 1, + }, + { flush: false }, + ); + await service.enqueue( + { + scopeId: '11', + aggregateType: 'test_items', + identity: { kind: 'generated', localId: '019d-user-item' }, + operation: 'test_items.update', + payload: { title: 'B' }, + optimisticValue: { id: 42, title: 'B' }, + baseRevision: 1, + }, + { flush: false }, + ); + + connected.set(true); + await expect(service.flush()).rejects.toBe(scope11Error); + expect(execute).toHaveBeenCalledOnce(); + expect(commands).toHaveLength(1); + expect(commands[0]).toMatchObject({ scopeId: '11', payload: { title: 'B' } }); + + pull.mockResolvedValue(undefined); + execute.mockClear(); + await service.flush(); + + expect(execute).toHaveBeenCalledOnce(); + expect(execute.mock.calls[0]?.[0]).toMatchObject({ scopeId: '11', payload: { title: 'B' } }); + expect(service.pendingCount()).toBe(0); + }); + + it('pre-pull: user-scoped aggregateでもfatalは成功scopeを送らず残りscopeを止める', async () => { + const fatal = new OfflineReplicaSchemaMismatchError(1, 'abc', 2, 'def'); + pull.mockImplementation(async (scope) => { + if (scope.scopeId === '10') throw fatal; + }); + await service.enqueue( + { + scopeId: '11', + aggregateType: 'test_items', + identity: { kind: 'generated', localId: '019d-user-item' }, + operation: 'test_items.update', + payload: { title: 'B' }, + optimisticValue: { id: 42, title: 'B' }, + baseRevision: 1, + }, + { flush: false }, + ); + + connected.set(true); + await expect(service.flush()).rejects.toBe(fatal); + expect(execute).not.toHaveBeenCalled(); + expect(pull.mock.calls.map((call) => call[0]?.scopeId)).toEqual(['10']); + }); + it('cross-partition commandの一方discardでも他方のoptimistic valueを保持する', async () => { const firstId = await service.enqueue( { diff --git a/projects/kit/offline/src/lib/offline-sync.service.ts b/projects/kit/offline/src/lib/offline-sync.service.ts index 8e6d6a9..57b749e 100644 --- a/projects/kit/offline/src/lib/offline-sync.service.ts +++ b/projects/kit/offline/src/lib/offline-sync.service.ts @@ -1,4 +1,4 @@ -import { computed, effect, ErrorHandler, inject, Injectable, signal } from '@angular/core'; +import { computed, effect, ErrorHandler, inject, Injectable, InjectionToken, signal } from '@angular/core'; import { OFFLINE_COMMAND_EXECUTOR, OFFLINE_SYNC_CONTEXT, @@ -11,7 +11,7 @@ import { import { OFFLINE_COMMAND_HOOKS } from './offline-command-hooks'; import { OFFLINE_KIT_OPTIONS } from './offline-kit-options'; import { OfflineNetworkService } from './offline-network.service'; -import { OfflineReplicaPullService } from './offline-replica-pull.service'; +import { OfflineReplicaPullService, OfflineReplicaSchemaMismatchError } from './offline-replica-pull.service'; import { OfflineReplicaMutationCoordinator } from './offline-replica-mutation-coordinator'; import type { OfflineCommand, @@ -123,6 +123,28 @@ const POST_SEND_PULL_RETRY_MS = 1_000; const DEFAULT_MAX_OUTBOX_COMMANDS_PER_USER = 1_000; const DEFAULT_MAX_OUTBOX_BYTES_PER_USER = 10 * 1024 * 1024; +/** Injectable `[0, 1)` source used by equal-jitter retry delay. Defaults to `Math.random`. */ +export const OFFLINE_RETRY_RANDOM = new InjectionToken<() => number>('OFFLINE_RETRY_RANDOM', { + factory: () => Math.random, +}); + +/** + * Equal-jitter delay in `[⌊cap/2⌋, cap)` for exponential offline command backoff. + * Keeps a positive lower bound (half the exponential cap) while still desynchronizing clients. + * + * @param attempts - Attempt count already recorded on the failed command (`>= 1` after a send claim). + * @param random - Unit interval sample in `[0, 1)`. + */ +export function offlineRetryDelayMs(attempts: number, random: () => number = Math.random): number { + const cap = Math.min(MAX_BACKOFF_MS, 1000 * 2 ** Math.max(0, attempts - 1)); + const unit = random(); + if (!(unit >= 0 && unit < 1)) { + throw new Error('Offline retry random() must return a number in [0, 1).'); + } + const half = cap / 2; + return Math.floor(half + unit * half); +} + /** Maintains the optimistic local replica and synchronizes its durable outbox. */ @Injectable({ providedIn: 'root' }) export class OfflineSyncService { @@ -135,6 +157,7 @@ export class OfflineSyncService { readonly #pull = inject(OfflineReplicaPullService); readonly #replicaMutations = inject(OfflineReplicaMutationCoordinator); readonly #errorHandler = inject(ErrorHandler); + readonly #retryRandom = inject(OFFLINE_RETRY_RANDOM); readonly #commands = signal([]); readonly #knownScopes = new Map(); /** ACKed scopes whose authoritative post-send pull has not completed yet. */ @@ -159,6 +182,8 @@ export class OfflineSyncService { readonly syncState = computed(() => { const commands = this.#commands(); if (commands.some((command) => ['blocked_auth', 'rejected', 'conflict'].includes(command.state))) return 'attention'; + // Any non-sending ambiguous commit (including restart-normalized pending+unknown) needs attention. + if (commands.some((command) => command.state !== 'sending' && command.serverCommitUnknown === true)) return 'attention'; if (commands.some((command) => command.state === 'sending')) return 'syncing'; return commands.length > 0 ? 'pending' : 'idle'; }); @@ -317,10 +342,7 @@ export class OfflineSyncService { */ replacePreparedAggregate( commandId: string, - prepare: ( - repository: OfflineRepository, - commands: readonly OfflineCommand[], - ) => Promise[]>, + prepare: (repository: OfflineRepository, commands: readonly OfflineCommand[]) => Promise[]>, options: PreparedOfflineBatchOptions = {}, ): Promise { const generation = this.#generation; @@ -341,14 +363,7 @@ export class OfflineSyncService { const materializations: MaterializedOfflineEnqueue[] = []; for (const [index, entry] of prepared.entries()) { this.#assertEnqueueScope(session, entry.request.scopeId); - materializations.push( - await this.#materializeEnqueue( - session.userId, - entry.request, - entry.replicaTransaction, - replaced[index], - ), - ); + materializations.push(await this.#materializeEnqueue(session.userId, entry.request, entry.replicaTransaction, replaced[index])); } const retained = knownCommands.filter((command) => !replaced.some((item) => item.commandId === command.commandId)); this.#assertDistinctBatchFootprints(materializations, retained, true); @@ -413,7 +428,12 @@ export class OfflineSyncService { const currentCommands = await this.#commandsForUser(session.userId); const retainedCommands = replaced ? currentCommands.filter((command) => command.commandId !== replaced.commandId) : currentCommands; this.#assertDistinctBatchFootprints([materialization], retainedCommands); - await this.#assertOutboxCapacity(session.userId, [materialization.command], replaced ? [replaced.commandId] : undefined, currentCommands); + await this.#assertOutboxCapacity( + session.userId, + [materialization.command], + replaced ? [replaced.commandId] : undefined, + currentCommands, + ); await this.#commitMaterializedEnqueues([materialization], generation, options, replaced ? [replaced.commandId] : undefined); return materialization.command.commandId; } @@ -923,59 +943,103 @@ export class OfflineSyncService { const pullScopes = isPartial ? this.#scopesForPartialPull(this.#foregroundScopePolicy!, await this.#readKnownCommands()) : [...this.#knownScopes.values()]; + const prePullFailures: unknown[] = []; + let fatalPullFailure: unknown | null = null; + const pulledScopeKeys = new Set(); for (const scope of pullScopes) { if (!this.#isCurrent(generation) || !this.#network.connected()) return; try { await this.#pull.pull(scope); await this.#markScopeReconciled(scope, generation); + pulledScopeKeys.add(this.#scopeKey(scope)); } catch (error) { + prePullFailures.push(error); + if (this.#isFatalPullFailure(error)) { + // Auth/upgrade-driven recovery only: stop remaining scopes immediately. + fatalPullFailure = fatalPullFailure ?? error; + break; + } 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)) { - const groups = this.#eligibleAggregateGroups(await this.#readKnownCommands()); + const sendWorkerFailures: unknown[] = []; + while (this.#network.connected() && this.#isCurrent(generation) && fatalPullFailure === null) { + const groups = this.#eligibleAggregateGroups(await this.#readKnownCommands()).filter((group) => { + const head = group[0]; + return head !== undefined && pulledScopeKeys.has(this.#scopeKey(head)); + }); if (!this.#isCurrent(generation)) return; if (groups.length === 0) break; let cursor = 0; const workers = Array.from({ length: Math.min(MAX_PARALLEL_AGGREGATES, groups.length) }, async () => { while (cursor < groups.length) { const group = groups[cursor++]; - if (group && this.#isCurrent(generation)) await this.#sendAggregate(group, generation, dirtyScopes); + if (group && this.#isCurrent(generation)) await this.#sendAggregate(group, generation, dirtyScopes, pulledScopeKeys); } }); - await Promise.all(workers); + // Drain with allSettled so one rejecting worker cannot skip sibling settlement or post-send pull. + const settled = await Promise.allSettled(workers); + for (const result of settled) { + if (result.status === 'rejected') sendWorkerFailures.push(result.reason); + } + if (sendWorkerFailures.length > 0) break; } 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); + // Fatal pre-pull skips pending post-pulls; recovery is auth/upgrade-driven, not timer retry. + // Fatal post-send pull stops remaining pending scopes the same way (ACK preserved, no resend). + if (fatalPullFailure === null) { + 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) { + if (this.#isFatalPullFailure(error)) { + fatalPullFailure = fatalPullFailure ?? error; + break; + } + postPullFailures.push(error); + } } } await this.#refreshState(generation); - if (postPullFailures.length > 0 && this.#isCurrent(generation)) { - // The server command is already committed and acknowledged locally, so - // never resend it. Retry only the idempotent scope pull and surface the - // failure to explicit flush callers/ErrorHandler. - this.#scheduleRetry(Date.now() + POST_SEND_PULL_RETRY_MS); - throw postPullFailures[0]; + if (fatalPullFailure !== null) { + // Auth/upgrade recovery only — never arm the 1s automatic flush retry. + // Post-send ACK already removed the command; reconciliation markers remain for later recovery. + // Only the owning generation may clear the timer — a stale fatal must not disarm a + // newer session's already-armed retry_wait / post-pull retry. + if (this.#isCurrent(generation)) this.#scheduleRetry(null); + throw fatalPullFailure; + } + const failures = [...prePullFailures, ...sendWorkerFailures, ...postPullFailures]; + if (failures.length > 0) { + // Keep rejecting the flush promise after workers settle even when the session + // was revoked mid-flight, so callers and ErrorHandler still observe the failure. + if (this.#isCurrent(generation)) { + this.#scheduleRetry(Date.now() + POST_SEND_PULL_RETRY_MS); + } + throw failures[0]; } if (this.#isCurrent(generation)) this.#coldReconciliationRequired = false; } + #isFatalPullFailure(error: unknown): boolean { + const status = this.#errorStatus(error); + // Pull-protocol HTTP 409 is schema mismatch (distinct from command-send conflict + // classification elsewhere in this service). Same classifier for pre-pull and post-send pull. + if (status === 401 || status === 403 || status === 409) return true; + return error instanceof OfflineReplicaSchemaMismatchError; + } + #setForegroundScopePolicy(foregroundScopeIds?: readonly string[]): void { this.#foregroundScopePolicy = foregroundScopeIds !== undefined ? foregroundScopeIds : null; } @@ -1006,11 +1070,19 @@ export class OfflineSyncService { }); } - async #sendAggregate(commands: OfflineCommand[], generation: number, dirtyScopes: Map): Promise { + async #sendAggregate( + commands: OfflineCommand[], + generation: number, + dirtyScopes: Map, + pulledScopeKeys: ReadonlySet, + ): Promise { for (const command of commands) { if (!this.#isCurrent(generation)) return; if (command.state === 'retry_wait' && (command.retryAt ?? 0) > Date.now()) break; if (!['pending', 'retry_wait'].includes(command.state)) break; + // User-scoped aggregates ignore scopeId in the FIFO key, so later commands may + // belong to scopes that failed pre-pull even when the head was admitted. + if (!pulledScopeKeys.has(this.#scopeKey({ userId: command.userId, scopeId: command.scopeId }))) break; let sending = await this.#claimSendingCommand(command, generation); if (!sending) return; if (!this.#isCurrent(generation)) return; @@ -1339,7 +1411,7 @@ export class OfflineSyncService { if (status >= 400 && status < 500 && status !== 429) { 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)); + const retryAt = Date.now() + offlineRetryDelayMs(command.attempts, this.#retryRandom); return { ...command, state: 'retry_wait', 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 b320c04..08b99f3 100644 --- a/projects/kit/offline/src/lib/sqlite-offline-repository.spec.ts +++ b/projects/kit/offline/src/lib/sqlite-offline-repository.spec.ts @@ -1290,6 +1290,110 @@ describe('SqliteOfflineRepository replica rows', () => { repository.getReplicaRow({ userId: 2, scopeId: '10' }, 'test_items', generatedCommandIdentity(localId)), ).resolves.toMatchObject({ values: { title: 'User 2' } }); }); + + it('standalone getReplicaRowsはreader leaseを保持し、完了までwriterを開始せずrollbackを漏らさない', async () => { + const repository = createRepository(); + await repository.initialize(); + const scope = { userId: 1 as const, scopeId: '10' }; + const row: OfflineReplicaRow = { + ...scope, + sourceKey: 'test_items', + identity: generatedReplicaIdentity('019d-lease-read', 50), + values: { id: 50, title: 'Before' }, + confirmedValues: { id: 50, title: 'Before' }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'confirmed', + }; + await repository.transactReplica({ putRows: [row] }); + + let releaseRead!: () => void; + const readGate = new Promise((resolve) => { + releaseRead = resolve; + }); + let announceRead!: () => void; + const readStarted = new Promise((resolve) => { + announceRead = resolve; + }); + let deferRowsQuery = true; + const originalQuery = plugin.query.getMockImplementation() as (options: { + statement: string; + values?: unknown[]; + }) => Promise; + plugin.query.mockImplementation(async (options: { statement: string; values?: unknown[] }) => { + if (deferRowsQuery && options.statement.startsWith('SELECT * FROM test_items')) { + deferRowsQuery = false; + announceRead(); + await readGate; + } + return originalQuery(options); + }); + + plugin.execute.mockClear(); + const read = repository.getReplicaRows(scope, 'test_items'); + await readStarted; + + let writeFinished = false; + const write = repository + .putCommand({ + userId: 1, + scopeId: '10', + commandId: 'cmd-after-lease-read', + aggregateType: 'test_items', + sourceKey: 'test_items', + identity: { kind: 'generated', localId: '019d-lease-read' }, + operation: 'test_items.update', + payload: {}, + optimisticValue: {}, + payloadHash: 'hash', + baseRevision: null, + state: 'pending', + attempts: 0, + retryAt: null, + createdAt: 1, + lastErrorCode: null, + }) + .then(() => { + writeFinished = true; + }); + void write.then( + () => undefined, + () => undefined, + ); + await Promise.resolve(); + await Promise.resolve(); + expect(writeFinished).toBe(false); + expect( + plugin.execute.mock.calls.some(([options]) => + String((options as { statement: string }).statement).includes('INSERT INTO offline_sync_commands'), + ), + ).toBe(false); + + releaseRead(); + await expect(read).resolves.toEqual([expect.objectContaining({ values: expect.objectContaining({ title: 'Before' }) })]); + await write; + expect(writeFinished).toBe(true); + + const originalExecute = plugin.execute.getMockImplementation() as (options: { + statement: string; + values?: unknown[]; + }) => Promise; + plugin.execute.mockImplementation(async (options: { statement: string; values?: unknown[] }) => { + if (options.statement.includes('INSERT INTO offline_replica_cursors')) throw new Error('constraint failed'); + return originalExecute(options); + }); + await expect( + repository.transactReplica({ + putCursors: [{ userId: 1, scopeId: '10', cursor: 'c-rollback' }], + }), + ).rejects.toThrow('constraint failed'); + expect(plugin.rollbackTransaction).toHaveBeenCalled(); + plugin.execute.mockImplementation(originalExecute); + await expect(repository.getReplicaRows(scope, 'test_items')).resolves.toEqual([ + expect.objectContaining({ values: expect.objectContaining({ title: 'Before' }) }), + ]); + await expect(repository.getReplicaCursor(scope)).resolves.toBeNull(); + }); }); describe('replica pull persistence', () => { @@ -1659,4 +1763,188 @@ describe('SqliteOfflineRepository replica rows', () => { }); return TestBed.inject(SqliteOfflineRepository); } + + describe('runReadSnapshot', () => { + it('open snapshot中はwriteが待機し、readerはcommit前の状態だけを見る', async () => { + const repository = createRepository(); + await repository.initialize(); + plugin.execute.mockClear(); + + let releaseSnapshot: (() => void) | undefined; + const snapshotGate = new Promise((resolve) => { + releaseSnapshot = resolve; + }); + let write: Promise = Promise.resolve(); + + const snapshot = repository.runReadSnapshot(async (reader) => { + await reader.getCommands({ userId: 1, scopeId: '10' }); + write = repository.putCommand({ + userId: 1, + scopeId: '10', + commandId: 'cmd-after-snapshot', + aggregateType: 'test_items', + sourceKey: 'test_items', + identity: { kind: 'generated', localId: '019d-aaaa' }, + operation: 'test_items.update', + payload: {}, + optimisticValue: {}, + payloadHash: 'hash', + baseRevision: null, + state: 'pending', + attempts: 0, + retryAt: null, + createdAt: 1, + lastErrorCode: null, + }); + void write.then( + () => undefined, + () => undefined, + ); + await Promise.resolve(); + await Promise.resolve(); + expect( + plugin.execute.mock.calls.some(([options]) => + String((options as { statement: string }).statement).includes('INSERT INTO offline_sync_commands'), + ), + ).toBe(false); + await snapshotGate; + }); + + releaseSnapshot?.(); + await snapshot; + await write; + expect( + plugin.execute.mock.calls.some(([options]) => + String((options as { statement: string }).statement).includes('INSERT INTO offline_sync_commands'), + ), + ).toBe(true); + }); + + it('transaction rollback後にcommitted readへ失敗を漏らさない', async () => { + const repository = createRepository(); + await repository.initialize(); + plugin.execute.mockImplementation(async ({ statement }: { statement: string }) => { + if (statement.includes('INSERT INTO offline_replica_cursors')) throw new Error('constraint failed'); + return {}; + }); + + await expect( + repository.transactReplica({ + putCursors: [{ userId: 1, scopeId: '10', cursor: 'c1' }], + }), + ).rejects.toThrow('constraint failed'); + expect(plugin.rollbackTransaction).toHaveBeenCalledOnce(); + expect(plugin.commitTransaction).not.toHaveBeenCalled(); + + plugin.execute.mockImplementation(async () => ({})); + plugin.query.mockImplementation(async ({ statement }: { statement: string }) => { + if (statement.includes('offline_replica_schema_metadata')) { + return { + columns: ['version', 'schema_hash'], + rows: [[storedReplicaMetadata!.version, storedReplicaMetadata!.schemaHash]], + }; + } + if (statement.includes('offline_replica_cursors')) return { rows: [] }; + return { rows: [] }; + }); + await expect(repository.runReadSnapshot((reader) => reader.getReplicaCursor({ userId: 1, scopeId: '10' }))).resolves.toBeNull(); + }); + + it('独立した並行snapshotはそれぞれreader leaseを持ち、writerは両方の完了を待つ', async () => { + const repository = createRepository(); + await repository.initialize(); + plugin.execute.mockClear(); + + let releaseA: (() => void) | undefined; + let releaseB: (() => void) | undefined; + const gateA = new Promise((resolve) => { + releaseA = resolve; + }); + const gateB = new Promise((resolve) => { + releaseB = resolve; + }); + let bothReadersReady: (() => void) | undefined; + const readersReady = new Promise((resolve) => { + bothReadersReady = resolve; + }); + let readersHeld = 0; + + plugin.query.mockImplementation(async ({ statement }: { statement: string }) => { + if (statement.includes('offline_replica_schema_metadata')) { + return { + columns: ['version', 'schema_hash'], + rows: [[storedReplicaMetadata!.version, storedReplicaMetadata!.schemaHash]], + }; + } + if (statement.startsWith('PRAGMA table_info')) return { rows: [{ name: 'next_local_id' }] }; + if (statement.includes('offline_sync_commands')) return { rows: [] }; + return { rows: [] }; + }); + + const snapshotA = repository.runReadSnapshot(async (reader) => { + await reader.getCommands({ userId: 1, scopeId: '10' }); + readersHeld += 1; + if (readersHeld === 2) bothReadersReady?.(); + await gateA; + }); + const snapshotB = repository.runReadSnapshot(async (reader) => { + await reader.getCommands({ userId: 1, scopeId: '10' }); + readersHeld += 1; + if (readersHeld === 2) bothReadersReady?.(); + await gateB; + }); + + await readersReady; + let writeFinished = false; + const write = repository + .putCommand({ + userId: 1, + scopeId: '10', + commandId: 'cmd-after-concurrent-snapshots', + aggregateType: 'test_items', + sourceKey: 'test_items', + identity: { kind: 'generated', localId: '019d-aaaa' }, + operation: 'test_items.update', + payload: {}, + optimisticValue: {}, + payloadHash: 'hash', + baseRevision: null, + state: 'pending', + attempts: 0, + retryAt: null, + createdAt: 1, + lastErrorCode: null, + }) + .then(() => { + writeFinished = true; + }); + void write.then( + () => undefined, + () => undefined, + ); + await Promise.resolve(); + await Promise.resolve(); + expect(writeFinished).toBe(false); + expect( + plugin.execute.mock.calls.some(([options]) => + String((options as { statement: string }).statement).includes('INSERT INTO offline_sync_commands'), + ), + ).toBe(false); + + releaseA?.(); + await snapshotA; + await Promise.resolve(); + expect(writeFinished).toBe(false); + + releaseB?.(); + await snapshotB; + await write; + expect(writeFinished).toBe(true); + expect( + plugin.execute.mock.calls.some(([options]) => + String((options as { statement: string }).statement).includes('INSERT INTO offline_sync_commands'), + ), + ).toBe(true); + }); + }); }); diff --git a/projects/kit/offline/src/lib/sqlite-offline-repository.ts b/projects/kit/offline/src/lib/sqlite-offline-repository.ts index 6a87da0..fb2bf25 100644 --- a/projects/kit/offline/src/lib/sqlite-offline-repository.ts +++ b/projects/kit/offline/src/lib/sqlite-offline-repository.ts @@ -39,6 +39,7 @@ import { type OfflineReplicaRowKey, type OfflineReplicaRemoteIdRelease, type OfflineRepository, + type OfflineRepositoryReader, type OfflineReplicaTransaction, type OfflineScope, } from './offline-repository'; @@ -197,6 +198,9 @@ export class SqliteOfflineRepository implements OfflineRepository { #databaseId: string | null = null; #initialization: Promise | null = null; #writes: Promise = Promise.resolve(); + #activeReaders = 0; + #readersIdle: Promise = Promise.resolve(); + #resolveReadersIdle: (() => void) | null = null; initialize(): Promise { this.#initialization ??= this.#open(); @@ -204,9 +208,7 @@ export class SqliteOfflineRepository implements OfflineRepository { } async getLastUserId(): Promise { - const rows = await this.#query('SELECT last_user_id FROM offline_metadata WHERE id = 1'); - const value = this.#stringOrNull(rows[0]?.['last_user_id']); - return value === null ? null : parseOfflinePrincipalId(value); + return this.#withCommittedRead(() => this.#readLastUserId()); } async setLastUserId(userId: OfflinePrincipalId): Promise { @@ -218,11 +220,7 @@ export class SqliteOfflineRepository implements OfflineRepository { } async getSessionManifest(userId: OfflinePrincipalId): Promise { - const rows = await this.#query('SELECT value_json FROM offline_session_manifests WHERE user_id = ?', [ - canonicalOfflinePrincipalId(userId), - ]); - const row = rows[0]; - return row ? this.#parse(row['value_json']) : null; + return this.#withCommittedRead(() => this.#readSessionManifest(userId)); } async putSessionManifest(userId: OfflinePrincipalId, value: T): Promise { @@ -238,8 +236,10 @@ export class SqliteOfflineRepository implements OfflineRepository { sourceKey: string, identity: OfflineReplicaAddress, ): Promise | null> { - const row = await this.#queryReplicaRow(scope, sourceKey, identity, false); - return row as OfflineReplicaRow | null; + return this.#withCommittedRead(async () => { + const row = await this.#queryReplicaRow(scope, sourceKey, identity, false); + return row as OfflineReplicaRow | null; + }); } async getReplicaRowIncludingPendingDelete( @@ -247,26 +247,14 @@ export class SqliteOfflineRepository implements OfflineRepository { sourceKey: string, identity: OfflineReplicaAddress, ): Promise | null> { - const row = await this.#queryReplicaRow(scope, sourceKey, identity, true); - return row as OfflineReplicaRow | null; + return this.#withCommittedRead(async () => { + const row = await this.#queryReplicaRow(scope, sourceKey, identity, true); + return row as OfflineReplicaRow | null; + }); } async getReplicaRows(scope: OfflineScope, sourceKey: string): Promise[]> { - const schema = this.#resolveReplicaEntitySchema(sourceKey); - const predicates = ['_offline_user_id = ?']; - const values: SQLiteValue[] = [canonicalOfflinePrincipalId(scope.userId)]; - if (schema.scope === 'partition') { - predicates.push('_offline_scope_id = ?'); - values.push(scope.scopeId); - } - const orderBy = - schema.identity.kind === 'naturalKey' - ? schema.identity.sourceKeys.map((key) => schema.fields.find((field) => field.sourceKey === key)!.sqliteColumnName!).join(', ') - : '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') - .map((row) => this.#replicaRowFromSqliteRow(schema, scope, sourceKey, row)); + return this.#withCommittedRead(() => this.#readReplicaRows(scope, sourceKey)); } async getReplicaRowByRemoteId( @@ -283,66 +271,27 @@ export class SqliteOfflineRepository implements OfflineRepository { sourceKey: string, identity: OfflineReplicaRemoteIdentity, ): Promise | null> { - const schema = this.#resolveReplicaEntitySchema(sourceKey); - canonicalOfflineRemoteIdentity(schema, identity); - if (schema.identity.kind === 'generated') { - const predicates = ['server_id = ?', '_offline_user_id = ?']; - const values: SQLiteValue[] = [identity.remoteId!, canonicalOfflinePrincipalId(scope.userId)]; - if (schema.scope === 'partition') { - predicates.push('_offline_scope_id = ?'); - values.push(scope.scopeId); - } - const rows = await this.#query(`SELECT * FROM ${schema.tableName} WHERE ${predicates.join(' AND ')}`, values); - const row = rows[0]; - return row ? this.#replicaRowFromSqliteRow(schema, scope, sourceKey, row) : null; - } - const naturalKey = normalizeOfflineNaturalKey(schema, identity.naturalKey!); - const predicates = ['_offline_user_id = ?']; - const values: SQLiteValue[] = [canonicalOfflinePrincipalId(scope.userId)]; - if (schema.scope === 'partition') { - predicates.push('_offline_scope_id = ?'); - values.push(scope.scopeId); - } - for (const sourceKeyPart of schema.identity.sourceKeys) { - const field = schema.fields.find((candidate) => candidate.sourceKey === sourceKeyPart)!; - predicates.push(`${field.sqliteColumnName!} = ?`); - values.push(naturalKey[sourceKeyPart]!); - } - const rows = await this.#query(`SELECT * FROM ${schema.tableName} WHERE ${predicates.join(' AND ')}`, values); - const row = rows[0]; - return row ? this.#replicaRowFromSqliteRow(schema, scope, sourceKey, row) : null; + return this.#withCommittedRead(() => this.#readReplicaRowByRemoteIdentity(scope, sourceKey, identity)); } async getReplicaCursor(scope: OfflineScope): Promise { - const rows = await this.#query('SELECT cursor FROM offline_replica_cursors WHERE user_id = ? AND scope_id = ?', [ - canonicalOfflinePrincipalId(scope.userId), - scope.scopeId, - ]); - const row = rows[0]; - if (!row) return null; - return { ...scope, cursor: this.#string(row['cursor']) }; + return this.#withCommittedRead(() => this.#readReplicaCursor(scope)); } 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']) })); + return this.#withCommittedRead(() => this.#readReconciliationScopes(userId)); } 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', - [canonicalOfflinePrincipalId(scope.userId), scope.scopeId], - ); - return rows.map((row) => this.#command(row)); + return this.#withCommittedRead(() => this.#readCommands(scope)); } async getCommandsForUser(userId: OfflinePrincipalId): Promise { - const rows = await this.#query('SELECT * FROM offline_sync_commands WHERE user_id = ? ORDER BY created_at ASC, command_id ASC', [ - canonicalOfflinePrincipalId(userId), - ]); - return rows.map((row) => this.#command(row)); + return this.#withCommittedRead(() => this.#readCommandsForUser(userId)); + } + + async runReadSnapshot(read: (reader: OfflineRepositoryReader) => Promise): Promise { + return this.#withCommittedRead(() => read(this.#reader())); } async putCommand(command: OfflineCommand): Promise { @@ -560,6 +509,163 @@ export class SqliteOfflineRepository implements OfflineRepository { return this.#databaseId; } + async #readLastUserId(): Promise { + const rows = await this.#query('SELECT last_user_id FROM offline_metadata WHERE id = 1'); + const value = this.#stringOrNull(rows[0]?.['last_user_id']); + return value === null ? null : parseOfflinePrincipalId(value); + } + + async #readSessionManifest(userId: OfflinePrincipalId): Promise { + const rows = await this.#query('SELECT value_json FROM offline_session_manifests WHERE user_id = ?', [ + canonicalOfflinePrincipalId(userId), + ]); + const row = rows[0]; + return row ? this.#parse(row['value_json']) : null; + } + + async #readReplicaRows(scope: OfflineScope, sourceKey: string): Promise[]> { + const schema = this.#resolveReplicaEntitySchema(sourceKey); + const predicates = ['_offline_user_id = ?']; + const values: SQLiteValue[] = [canonicalOfflinePrincipalId(scope.userId)]; + if (schema.scope === 'partition') { + predicates.push('_offline_scope_id = ?'); + values.push(scope.scopeId); + } + const orderBy = + schema.identity.kind === 'naturalKey' + ? schema.identity.sourceKeys.map((key) => schema.fields.find((field) => field.sourceKey === key)!.sqliteColumnName!).join(', ') + : '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') + .map((row) => this.#replicaRowFromSqliteRow(schema, scope, sourceKey, row)); + } + + async #readReplicaRowByRemoteIdentity( + scope: OfflineScope, + sourceKey: string, + identity: OfflineReplicaRemoteIdentity, + ): Promise | null> { + const schema = this.#resolveReplicaEntitySchema(sourceKey); + canonicalOfflineRemoteIdentity(schema, identity); + if (schema.identity.kind === 'generated') { + const predicates = ['server_id = ?', '_offline_user_id = ?']; + const values: SQLiteValue[] = [identity.remoteId!, canonicalOfflinePrincipalId(scope.userId)]; + if (schema.scope === 'partition') { + predicates.push('_offline_scope_id = ?'); + values.push(scope.scopeId); + } + const rows = await this.#query(`SELECT * FROM ${schema.tableName} WHERE ${predicates.join(' AND ')}`, values); + const row = rows[0]; + return row ? this.#replicaRowFromSqliteRow(schema, scope, sourceKey, row) : null; + } + const naturalKey = normalizeOfflineNaturalKey(schema, identity.naturalKey!); + const predicates = ['_offline_user_id = ?']; + const values: SQLiteValue[] = [canonicalOfflinePrincipalId(scope.userId)]; + if (schema.scope === 'partition') { + predicates.push('_offline_scope_id = ?'); + values.push(scope.scopeId); + } + for (const sourceKeyPart of schema.identity.sourceKeys) { + const field = schema.fields.find((candidate) => candidate.sourceKey === sourceKeyPart)!; + predicates.push(`${field.sqliteColumnName!} = ?`); + values.push(naturalKey[sourceKeyPart]!); + } + const rows = await this.#query(`SELECT * FROM ${schema.tableName} WHERE ${predicates.join(' AND ')}`, values); + const row = rows[0]; + return row ? this.#replicaRowFromSqliteRow(schema, scope, sourceKey, row) : null; + } + + async #readReplicaCursor(scope: OfflineScope): Promise { + const rows = await this.#query('SELECT cursor FROM offline_replica_cursors WHERE user_id = ? AND scope_id = ?', [ + canonicalOfflinePrincipalId(scope.userId), + scope.scopeId, + ]); + const row = rows[0]; + if (!row) return null; + return { ...scope, cursor: this.#string(row['cursor']) }; + } + + async #readReconciliationScopes(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 #readCommands(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', + [canonicalOfflinePrincipalId(scope.userId), scope.scopeId], + ); + return rows.map((row) => this.#command(row)); + } + + async #readCommandsForUser(userId: OfflinePrincipalId): Promise { + const rows = await this.#query('SELECT * FROM offline_sync_commands WHERE user_id = ? ORDER BY created_at ASC, command_id ASC', [ + canonicalOfflinePrincipalId(userId), + ]); + return rows.map((row) => this.#command(row)); + } + + #reader(): OfflineRepositoryReader { + return { + getLastUserId: () => this.#readLastUserId(), + getSessionManifest: (userId) => this.#readSessionManifest(userId), + getReplicaRow: async ( + scope: OfflineScope, + sourceKey: string, + identity: OfflineReplicaAddress, + ): Promise | null> => + (await this.#queryReplicaRow(scope, sourceKey, identity, false)) as OfflineReplicaRow | null, + getReplicaRowIncludingPendingDelete: async ( + scope: OfflineScope, + sourceKey: string, + identity: OfflineReplicaAddress, + ): Promise | null> => + (await this.#queryReplicaRow(scope, sourceKey, identity, true)) as OfflineReplicaRow | null, + getReplicaRows: (scope, sourceKey) => this.#readReplicaRows(scope, sourceKey), + getReplicaRowByRemoteId: async (scope, sourceKey, remoteId) => { + if (this.#resolveReplicaEntitySchema(sourceKey).identity.kind !== 'generated') return null; + return this.#readReplicaRowByRemoteIdentity(scope, sourceKey, { remoteId }); + }, + getReplicaRowByRemoteIdentity: (scope, sourceKey, identity) => this.#readReplicaRowByRemoteIdentity(scope, sourceKey, identity), + getReplicaCursor: (scope) => this.#readReplicaCursor(scope), + getReconciliationScopes: (userId) => this.#readReconciliationScopes(userId), + getCommands: (scope) => this.#readCommands(scope), + getCommandsForUser: (userId) => this.#readCommandsForUser(userId), + }; + } + + async #withCommittedRead(operation: () => Promise): Promise { + await this.initialize(); + await this.#writes; + this.#beginReaders(); + try { + return await operation(); + } finally { + this.#endReaders(); + } + } + + #beginReaders(): void { + if (this.#activeReaders === 0) { + this.#readersIdle = new Promise((resolve) => { + this.#resolveReadersIdle = resolve; + }); + } + this.#activeReaders += 1; + } + + #endReaders(): void { + this.#activeReaders -= 1; + if (this.#activeReaders === 0) { + this.#resolveReadersIdle?.(); + this.#resolveReadersIdle = null; + this.#readersIdle = Promise.resolve(); + } + } + async #query(statement: string, values: SQLiteValue[] = []): Promise { return this.#queryDatabase(await this.#databaseConnection(), statement, values); } @@ -569,13 +675,17 @@ export class SqliteOfflineRepository implements OfflineRepository { } #queueWrite(run: (databaseId: string) => Promise): Promise { - const write = this.#writes.then(async (): Promise => run(await this.#databaseConnection())); + const write = this.#writes.then(async (): Promise => { + if (this.#activeReaders > 0) await this.#readersIdle; + await run(await this.#databaseConnection()); + }); this.#writes = write.catch((): void => undefined); return write; } #transaction(run: (databaseId: string) => Promise): Promise { const write = this.#writes.then(async (): Promise => { + if (this.#activeReaders > 0) await this.#readersIdle; const databaseId = await this.#databaseConnection(); await this.#sqlite!.beginTransaction({ databaseId }); try {