diff --git a/projects/kit/offline/src/lib/offline-coordinator.service.ts b/projects/kit/offline/src/lib/offline-coordinator.service.ts index e289911..ca27c92 100644 --- a/projects/kit/offline/src/lib/offline-coordinator.service.ts +++ b/projects/kit/offline/src/lib/offline-coordinator.service.ts @@ -56,6 +56,7 @@ export class OfflineCoordinatorService { readonly networkState = this.#network.state; readonly syncState = this.#sync.syncState; + readonly pendingCommands = this.#sync.pendingCommands; readonly pendingCount = this.#sync.pendingCount; readonly conflicts = this.#sync.conflicts; /** Device-local control for accepting new durable Outbox mutations. */ diff --git a/projects/kit/offline/src/lib/offline-request-policy.ts b/projects/kit/offline/src/lib/offline-request-policy.ts index 7ff7bab..3b96c7a 100644 --- a/projects/kit/offline/src/lib/offline-request-policy.ts +++ b/projects/kit/offline/src/lib/offline-request-policy.ts @@ -43,7 +43,7 @@ export function shouldCommitOfflineCollection(emission: OfflineReadEmission { expect(rows).toEqual([]); }); + it('uses a caller-owned idempotency key once and rejects a duplicate command id before persistence', async () => { + const request = { + commandId: 'caller-owned-idempotency-key', + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated' as const, localId: 'caller-owned-1' }, + operation: 'documents.create', + payload: { title: 'first' }, + }; + + await expect(service.enqueue(request, { flush: false })).resolves.toBe(request.commandId); + await expect( + service.enqueue( + { + ...request, + identity: { kind: 'generated', localId: 'caller-owned-2' }, + payload: { title: 'second' }, + }, + { flush: false }, + ), + ).rejects.toThrow('is already in use'); + + expect(commands).toHaveLength(1); + expect(rows).toHaveLength(1); + }); + it('rejects every new command entry point after mutation admission closes', async () => { await TestBed.inject(OfflineMutationAdmissionService).close(); const request = { @@ -2289,6 +2316,548 @@ describe('OfflineSyncService', () => { expect(commands.every((command) => !('remoteId' in command))).toBe(true); }); + it('新規generated aggregateはpre-pullを待たずdurable commandの同じ冪等IDで即時送信する', async () => { + execute.mockImplementationOnce(async (command, target) => { + expect(pull).not.toHaveBeenCalled(); + expect(target).toEqual({ kind: 'generated', localId: 'network-first-create', remoteId: null }); + return { remoteId: 38143, response: 38143 }; + }); + const commandId = await service.enqueue( + { + commandId: 'network-first-idempotency', + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'network-first-create' }, + operation: 'documents.create', + payload: { title: 'Created' }, + }, + { flush: false }, + ); + expect(commandId).toBe('network-first-idempotency'); + networkConnected = () => true; + + await expect( + service.sendGeneratedCommandNow(commandId, { + scopeId: '10', + sourceKey: 'documents', + localId: 'network-first-create', + }), + ).resolves.toBe(38143); + + expect(execute).toHaveBeenCalledOnce(); + expect(execute.mock.calls[0]?.[0].commandId).toBe(commandId); + expect(rows[0]).toMatchObject({ + identity: { kind: 'generated', localId: 'network-first-create', remoteId: 38143 }, + syncState: 'pending', + }); + expect(commands[0]).toMatchObject({ + commandId, + state: 'awaiting_pull', + reconciliationIdentity: { remoteId: 38143 }, + }); + }); + + it('unrelatedな先行flushのpre-pullを待たずcreateを送信してserver idを返す', async () => { + networkConnected = () => true; + let releasePrePull!: () => void; + pull + .mockImplementationOnce( + () => + new Promise((resolve) => { + releasePrePull = resolve; + }), + ) + .mockImplementationOnce(async () => { + commands = []; + }); + execute.mockResolvedValueOnce({ remoteId: 38144, response: 38144 }); + const activeFlush = service.flush(); + await vi.waitFor(() => expect(pull).toHaveBeenCalledOnce()); + const commandId = await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'create-during-active-flush' }, + operation: 'documents.create', + payload: { title: 'Concurrent' }, + }, + { flush: false }, + ); + const delivery = service.sendGeneratedCommandNow(commandId, { + scopeId: '10', + sourceKey: 'documents', + localId: 'create-during-active-flush', + }); + + await expect(delivery).resolves.toBe(38144); + expect(execute).toHaveBeenCalledOnce(); + expect(pull).toHaveBeenCalledOnce(); + + releasePrePull(); + await activeFlush; + + expect(execute).toHaveBeenCalledOnce(); + await vi.waitFor(() => expect(commands).toEqual([])); + expect(rows[0]).toMatchObject({ + identity: { kind: 'generated', localId: 'create-during-active-flush', remoteId: 38144 }, + }); + }); + + it('create A後のbackground pullを待たずcreate Bを即時送信する', async () => { + networkConnected = () => true; + let releaseBackgroundPull!: () => void; + pull.mockImplementationOnce( + () => + new Promise((resolve) => { + releaseBackgroundPull = resolve; + }), + ); + execute.mockResolvedValueOnce({ remoteId: 38145, response: 38145 }).mockResolvedValueOnce({ remoteId: 38146, response: 38146 }); + const commandA = await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'network-first-a' }, + operation: 'documents.create', + payload: { title: 'A' }, + }, + { flush: false }, + ); + + await expect( + service.sendGeneratedCommandNow(commandA, { + scopeId: '10', + sourceKey: 'documents', + localId: 'network-first-a', + }), + ).resolves.toBe(38145); + await vi.waitFor(() => expect(pull).toHaveBeenCalledOnce()); + + const commandB = await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'network-first-b' }, + operation: 'documents.create', + payload: { title: 'B' }, + }, + { flush: false }, + ); + await expect( + service.sendGeneratedCommandNow(commandB, { + scopeId: '10', + sourceKey: 'documents', + localId: 'network-first-b', + }), + ).resolves.toBe(38146); + + expect(execute).toHaveBeenCalledTimes(2); + expect(pull).toHaveBeenCalledOnce(); + releaseBackgroundPull(); + }); + + it('通常flushとfast pathが同じcommandを競合取得してもtransportは1回だけ実行する', async () => { + networkConnected = () => true; + let releasePrePull!: () => void; + let releaseExecute!: () => void; + pull.mockImplementationOnce( + () => + new Promise((resolve) => { + releasePrePull = resolve; + }), + ); + execute.mockImplementationOnce( + () => + new Promise((resolve) => { + releaseExecute = () => resolve({ remoteId: 38147, response: 38147 }); + }), + ); + const commandId = await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'same-command-race' }, + operation: 'documents.create', + payload: { title: 'Race' }, + }, + { flush: false }, + ); + const flush = service.flush(); + await vi.waitFor(() => expect(pull).toHaveBeenCalledOnce()); + const delivery = service.sendGeneratedCommandNow(commandId, { + scopeId: '10', + sourceKey: 'documents', + localId: 'same-command-race', + }); + releasePrePull(); + await vi.waitFor(() => expect(execute).toHaveBeenCalledOnce()); + releaseExecute(); + + await expect(delivery).resolves.toBe(38147); + await flush; + expect(execute).toHaveBeenCalledOnce(); + }); + + it('通常flushが送信中にした同じcommandの結果をfast pathが待ってserver idを返す', async () => { + networkConnected = () => true; + let releaseExecute!: () => void; + execute.mockImplementationOnce( + () => + new Promise((resolve) => { + releaseExecute = () => resolve({ remoteId: 38151, response: 38151 }); + }), + ); + const commandId = await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'already-sending-race' }, + operation: 'documents.create', + payload: { title: 'Already sending' }, + }, + { flush: false }, + ); + const flush = service.flush(); + await vi.waitFor(() => expect(execute).toHaveBeenCalledOnce()); + expect(commands[0]).toMatchObject({ commandId, state: 'sending' }); + + const delivery = service.sendGeneratedCommandNow(commandId, { + scopeId: '10', + sourceKey: 'documents', + localId: 'already-sending-race', + }); + releaseExecute(); + + await expect(delivery).resolves.toBe(38151); + await flush; + expect(execute).toHaveBeenCalledOnce(); + }); + + it('sendingの読取直後にactive transitionが消えてもdurableなserver idを再読込する', async () => { + networkConnected = () => true; + const commandId = await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'completed-before-map-lookup' }, + operation: 'documents.create', + payload: { title: 'Completed concurrently' }, + }, + { flush: false }, + ); + commands = commands.map((command) => + command.commandId === commandId ? { ...command, state: 'sending', attempts: 1, serverCommitUnknown: true } : command, + ); + const repository = TestBed.inject(OFFLINE_REPOSITORY); + const getCommandsForUser = vi.mocked(repository.getCommandsForUser!); + let readCount = 0; + getCommandsForUser.mockImplementation(async (userId) => { + const snapshot = commands.filter((command) => command.userId === userId).map((command) => structuredClone(command)); + readCount += 1; + if (readCount === 1) { + commands = commands.map((command) => + command.commandId === commandId + ? { + ...command, + state: 'awaiting_pull', + serverCommitUnknown: false, + reconciliationIdentity: { remoteId: 38152 }, + } + : command, + ); + rows = rows.map((row) => + row.identity.kind === 'generated' && row.identity.localId === 'completed-before-map-lookup' + ? { ...row, identity: { ...row.identity, remoteId: 38152 } } + : row, + ); + } + return snapshot; + }); + + await expect( + service.sendGeneratedCommandNow(commandId, { + scopeId: '10', + sourceKey: 'documents', + localId: 'completed-before-map-lookup', + }), + ).resolves.toBe(38152); + expect(readCount).toBeGreaterThanOrEqual(2); + expect(execute).not.toHaveBeenCalled(); + }); + + it('新規generated aggregateの即時送信で応答を失っても同じdurable commandをretry_waitに残す', async () => { + execute.mockRejectedValueOnce({ status: 0, message: 'response lost' }); + const commandId = await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'network-first-response-loss' }, + operation: 'documents.create', + payload: { title: 'Created once' }, + }, + { flush: false }, + ); + networkConnected = () => true; + + await expect( + service.sendGeneratedCommandNow(commandId, { + scopeId: '10', + sourceKey: 'documents', + localId: 'network-first-response-loss', + }), + ).resolves.toBeNull(); + + expect(execute).toHaveBeenCalledOnce(); + expect(execute.mock.calls[0]?.[0].commandId).toBe(commandId); + expect(commands[0]).toMatchObject({ + commandId, + state: 'retry_wait', + serverCommitUnknown: true, + lastErrorCode: 'network', + }); + }); + + it('新規generated aggregateの即時送信で確定4xxをoptimistic成功として隠さない', async () => { + execute.mockRejectedValueOnce({ status: 422, message: 'invalid create' }); + const commandId = await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'network-first-rejected' }, + operation: 'documents.create', + payload: { title: 'Invalid' }, + }, + { flush: false }, + ); + networkConnected = () => true; + + const delivery = service.sendGeneratedCommandNow(commandId, { + scopeId: '10', + sourceKey: 'documents', + localId: 'network-first-rejected', + }); + await expect(delivery).rejects.toBeInstanceOf(OfflineImmediateCommandRejectedError); + await expect(delivery).rejects.toMatchObject({ + name: 'OfflineImmediateCommandRejectedError', + state: 'rejected', + code: '422', + }); + + expect(commands[0]).toMatchObject({ commandId, state: 'rejected', lastErrorCode: '422' }); + }); + + it('confirmed aggregateはfast pathを拒否し通常flushのpre-pull後だけ送信する', async () => { + rows.push({ + userId: 1, + scopeId: '10', + sourceKey: 'documents', + identity: { kind: 'generated', localId: 'confirmed-document', remoteId: 55 }, + values: { id: 55, title: 'Confirmed' }, + confirmedValues: { id: 55, title: 'Confirmed' }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'confirmed', + }); + const commandId = await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'confirmed-document' }, + operation: 'documents.update', + payload: { title: 'Updated' }, + baseRevision: 1, + }, + { flush: false }, + ); + networkConnected = () => true; + + await expect( + service.sendGeneratedCommandNow(commandId, { + scopeId: '10', + sourceKey: 'documents', + localId: 'confirmed-document', + }), + ).rejects.toThrow('requires a new unconfirmed aggregate'); + expect(execute).not.toHaveBeenCalled(); + + let prePullCompleted = false; + pull.mockImplementationOnce(async () => { + expect(execute).not.toHaveBeenCalled(); + prePullCompleted = true; + }); + execute.mockImplementationOnce(async () => { + expect(prePullCompleted).toBe(true); + return { serverRevision: 2, response: null }; + }); + await service.flush(); + expect(execute).toHaveBeenCalledOnce(); + }); + + it('durable pull attention中はgenerated fast pathのtransportを開始しない', async () => { + pullAttentions.push({ userId: 1, scopeId: '10', reason: 'authorization_required', status: 401 }); + const commandId = await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'attention-blocked-create' }, + operation: 'documents.create', + payload: { title: 'Blocked' }, + }, + { flush: false }, + ); + networkConnected = () => true; + + await expect( + service.sendGeneratedCommandNow(commandId, { + scopeId: '10', + sourceKey: 'documents', + localId: 'attention-blocked-create', + }), + ).resolves.toBeNull(); + expect(execute).not.toHaveBeenCalled(); + }); + + it('同一aggregateの先行intentを追い越してgenerated fast pathを送信しない', async () => { + await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'ordered-create' }, + operation: 'documents.create', + payload: { title: 'First' }, + }, + { flush: false }, + ); + const secondId = await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'ordered-create' }, + operation: 'documents.update', + payload: { title: 'Second' }, + }, + { flush: false }, + ); + networkConnected = () => true; + + await expect( + service.sendGeneratedCommandNow(secondId, { + scopeId: '10', + sourceKey: 'documents', + localId: 'ordered-create', + }), + ).rejects.toThrow('requires the first aggregate intent'); + expect(execute).not.toHaveBeenCalled(); + }); + + it('session generation切替後は新principalの同一locator remote idを旧callerへ返さない', async () => { + let releaseExecute!: () => void; + execute.mockImplementationOnce( + () => + new Promise((resolve) => { + releaseExecute = () => resolve({ remoteId: 38148, response: 38148 }); + }), + ); + const commandId = await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'shared-locator' }, + operation: 'documents.create', + payload: { title: 'Old principal' }, + }, + { flush: false }, + ); + networkConnected = () => true; + const delivery = service.sendGeneratedCommandNow(commandId, { + scopeId: '10', + sourceKey: 'documents', + localId: 'shared-locator', + }); + await vi.waitFor(() => expect(execute).toHaveBeenCalledOnce()); + + service.revokeSession(); + session = { userId: 2, scopes: [{ userId: 2, scopeId: '10' }] }; + rows.push({ + userId: 2, + scopeId: '10', + sourceKey: 'documents', + identity: { kind: 'generated', localId: 'shared-locator', remoteId: 999 }, + values: { id: 999, title: 'New principal' }, + confirmedValues: { id: 999, title: 'New principal' }, + serverRevision: 1, + fetchedAt: 2, + syncState: 'confirmed', + }); + networkConnected = () => false; + await service.refreshSession(); + releaseExecute(); + + await expect(delivery).resolves.toBeNull(); + }); + + it('旧principalのtransport中でも新principalの同一command idは独立して即時送信する', async () => { + let releaseOldExecute!: () => void; + execute + .mockImplementationOnce( + () => + new Promise((resolve) => { + releaseOldExecute = () => resolve({ remoteId: 38149, response: 38149 }); + }), + ) + .mockResolvedValueOnce({ remoteId: 38150, response: 38150 }); + const sharedCommandId = 'principal-scoped-idempotency-key'; + await service.enqueue( + { + commandId: sharedCommandId, + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'old-principal-create' }, + operation: 'documents.create', + payload: { title: 'Old principal' }, + }, + { flush: false }, + ); + networkConnected = () => true; + const oldDelivery = service.sendGeneratedCommandNow(sharedCommandId, { + scopeId: '10', + sourceKey: 'documents', + localId: 'old-principal-create', + }); + await vi.waitFor(() => expect(execute).toHaveBeenCalledOnce()); + + service.revokeSession(); + commands = commands.filter((command) => command.userId !== 1); + rows = rows.filter((row) => row.userId !== 1); + session = { userId: 2, scopes: [{ userId: 2, scopeId: '10' }] }; + networkConnected = () => false; + await service.refreshSession(); + await service.enqueue( + { + commandId: sharedCommandId, + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'new-principal-create' }, + operation: 'documents.create', + payload: { title: 'New principal' }, + }, + { flush: false }, + ); + networkConnected = () => true; + + await expect( + service.sendGeneratedCommandNow(sharedCommandId, { + scopeId: '10', + sourceKey: 'documents', + localId: 'new-principal-create', + }), + ).resolves.toBe(38150); + expect(execute).toHaveBeenCalledTimes(2); + + releaseOldExecute(); + await expect(oldDelivery).resolves.toBeNull(); + }); + it('session scope発見後に前回起動のsending commandをpendingへ復旧する', async () => { session = null; rows.push({ @@ -2811,6 +3380,47 @@ describe('OfflineSyncService', () => { expect(rows[0]).toMatchObject({ values: { id: 13, title: 'new local' }, syncState: 'pending' }); }); + it('同じcommand idのreplacementを新commandの削除なしで原子的に保持する', async () => { + const commandId = await service.enqueue( + { + commandId: 'stable-idempotency-key', + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'replace-stable-command' }, + operation: 'documents.update', + payload: { title: 'rejected draft' }, + }, + { flush: false }, + ); + commands[0] = { ...commands[0]!, state: 'rejected', lastErrorCode: '422' }; + const originalCreatedAt = commands[0]!.createdAt; + + const replacementId = await service.replacePrepared( + commandId, + async () => ({ + request: { + commandId, + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'replace-stable-command' }, + operation: 'documents.update', + payload: { title: 'corrected draft' }, + }, + }), + { flush: false }, + ); + + expect(replacementId).toBe(commandId); + expect(commands).toHaveLength(1); + expect(commands[0]).toMatchObject({ + commandId, + state: 'pending', + createdAt: originalCreatedAt, + payload: { title: 'corrected draft' }, + }); + expect(rows).toContainEqual(expect.objectContaining({ values: { id: 0, title: 'corrected draft' } })); + }); + it('同じaggregateに後続commandがあるreplacementを元状態のまま拒否する', async () => { const oldCommandId = 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 04b2ae9..1b0ba72 100644 --- a/projects/kit/offline/src/lib/offline-sync.service.ts +++ b/projects/kit/offline/src/lib/offline-sync.service.ts @@ -51,6 +51,8 @@ export type OfflineSyncState = 'idle' | 'pending' | 'syncing' | 'attention'; /** Mutation appended atomically to the outbox; Kit rematerializes optimistic replica state. */ export interface EnqueueOfflineCommand { + /** Stable caller-owned idempotency key; generated by Kit when omitted. */ + commandId?: string; scopeId: string; aggregateType: string; identity: EnqueueOfflineCommandIdentity; @@ -81,6 +83,13 @@ export interface PreparedOfflineBatchOptions { assertCurrent?: () => void; } +/** Stable lookup retained by a caller while a generated command may be reconciled and removed. */ +export interface OfflineGeneratedCommandLocator { + readonly scopeId: string; + readonly sourceKey: string; + readonly localId: string; +} + /** Validated Outbox command ready for a rematerialized replica commit. */ interface MaterializedOfflineEnqueue { command: OfflineCommand; @@ -96,6 +105,17 @@ export class OfflinePayloadValidationError extends Error { } } +/** A foreground durable-command send reached a terminal server-side state. */ +export class OfflineImmediateCommandRejectedError extends Error { + constructor( + readonly state: 'blocked_auth' | 'rejected' | 'conflict', + readonly code: string | null, + ) { + super(`Immediate offline command delivery ended in ${state}${code ? ` (${code})` : ''}.`); + this.name = 'OfflineImmediateCommandRejectedError'; + } +} + /** Raised before persistence when retaining another command would exceed the configured durable Outbox capacity. */ export class OfflineOutboxCapacityError extends Error { constructor( @@ -176,6 +196,7 @@ export class OfflineSyncService { readonly #flushTransitions = new Set>(); #generation = 0; readonly #sendingTransitions = new Set>(); + readonly #commandSendTransitions = new Map>(); #retryTimer: ReturnType | null = null; /** When non-null, automatic flushes pull only foreground scopes plus Outbox scopes. */ #foregroundScopePolicy: readonly string[] | null = null; @@ -542,7 +563,10 @@ export class OfflineSyncService { this.noteScope(scope); const commandIdentity = offlineCommandLookupIdentity(request.identity); const normalized = await this.#normalizeEnqueueRequest(scope, request, commandIdentity, repository); - const commandId = crypto.randomUUID(); + const commandId = request.commandId ?? crypto.randomUUID(); + if (typeof commandId !== 'string' || commandId.length === 0 || commandId.length > 255) { + throw new Error('Offline command id must contain between 1 and 255 characters.'); + } const sourceKey = this.#hooks.entityType(request); const localOnlyFootprint = this.#normalizedLocalOnlyFootprint(scope, request.localOnlyFootprint); if (replaced) this.#assertReplacementFootprint(replaced, localOnlyFootprint); @@ -678,12 +702,17 @@ export class OfflineSyncService { ): void { const aggregates = new Set(); const replicaKeys = new Set(); + const commandIds = new Set(existingCommands.map((command) => command.commandId)); const existingFootprints = new Map(); for (const command of existingCommands) { const aggregate = this.#aggregateKey(command); for (const key of this.#commandFootprintKeys(command)) existingFootprints.set(key, aggregate); } for (const entry of entries) { + if (commandIds.has(entry.command.commandId)) { + throw new Error(`Offline command id ${entry.command.commandId} is already in use.`); + } + commandIds.add(entry.command.commandId); const aggregate = this.#aggregateKey(entry.command); if (aggregates.has(aggregate) && !allowOneAggregate) { throw new Error('Prepared offline batch contains overlapping aggregate intents.'); @@ -721,6 +750,8 @@ export class OfflineSyncService { throw new Error('Offline session changed before the command could be persisted'); } const known = await this.#readKnownCommands(repository); + const putCommandIds = new Set(entries.map((entry) => entry.command.commandId)); + const effectiveRemoveCommandIds = removeCommandIds?.filter((commandId) => !putCommandIds.has(commandId)); const remaining = [ ...known.filter((command) => !(removeCommandIds ?? []).includes(command.commandId)), ...entries.map((entry) => entry.command), @@ -736,7 +767,7 @@ export class OfflineSyncService { putRows: rematerialized.putRows, removeRows: rematerialized.removeRows, putCommands: entries.map((entry) => entry.command), - removeCommandIds, + removeCommandIds: effectiveRemoveCommandIds, }); await this.#refreshState(generation, repository).catch((error) => this.#reportError(error)); if (options.flush !== false && this.#canSynchronize()) this.#flushInBackground(); @@ -907,11 +938,162 @@ export class OfflineSyncService { return this.#flushAfterInitialization(generation); } + /** + * Sends one newly-created generated aggregate immediately after its durable + * enqueue, without waiting for a pre-send pull. + * + * This fast path is deliberately narrow: the command must still be pending, + * have no remote id or confirmed baseline, and be the first intent for its + * aggregate. Existing aggregates continue through {@link flush}, where the + * pre-pull conflict barrier remains mandatory. Authoritative pull + * reconciliation is scheduled in the background after the local transport + * acknowledgement has been committed. + * + * @returns The generated remote id when transport completed locally, or + * `null` when transport is unavailable or did not produce a confirmed id. + */ + async sendGeneratedCommandNow(commandId: string, locator: OfflineGeneratedCommandLocator): Promise { + const generation = this.#generation; + if (!(await this.#restoreCurrentGeneration(generation))) return null; + const userId = this.#activeUserId; + if (userId === null) return null; + return this.#beginGeneratedCommandSend(commandId, locator, generation, userId); + } + async #flushAfterInitialization(generation: number): Promise { if (!(await this.#restoreCurrentGeneration(generation))) return; return this.#beginFlush(true); } + #beginGeneratedCommandSend( + commandId: string, + locator: OfflineGeneratedCommandLocator, + generation: number, + userId: OfflinePrincipalId, + ): Promise { + const result = this.#runGeneratedCommandSend(commandId, locator, generation, userId); + let requiresReconciliation = false; + const transition = result + .then( + (remoteId) => { + requiresReconciliation = remoteId !== null; + }, + () => undefined, + ) + .finally(() => { + this.#flushTransitions.delete(transition); + if (requiresReconciliation && this.#canSynchronize() && this.#isCurrentUser(generation, userId)) { + // A normal flush may already be in its post-pull phase. Ensure the + // newly acknowledged command receives a later reconciliation pass. + if (this.#flushPromise) this.#resumeAfterFlush = true; + queueMicrotask(() => this.#flushInBackground()); + } + }); + this.#flushTransitions.add(transition); + return result; + } + + async #runGeneratedCommandSend( + commandId: string, + locator: OfflineGeneratedCommandLocator, + generation: number, + userId: OfflinePrincipalId, + ): Promise { + if (!this.#isCurrentUser(generation, userId)) return null; + if (!this.#canSynchronize()) { + await this.#refreshState(generation); + return null; + } + if (this.#pullAttentions().length > 0) return null; + if (!(await this.#discoverScopes(generation))) return null; + if (!this.#isCurrentUser(generation, userId)) return null; + const known = await this.#readKnownCommands(); + if (!this.#isCurrentUser(generation, userId)) return null; + const command = known.find((candidate) => candidate.commandId === commandId); + if (!command) return this.#generatedRemoteId(locator, userId, generation); + if (command.userId !== userId) return null; + if ( + command.scopeId !== locator.scopeId || + command.sourceKey !== locator.sourceKey || + command.identity.kind !== 'generated' || + command.identity.localId !== locator.localId + ) { + throw new Error('Immediate generated command locator does not match the durable command.'); + } + if (command.state === 'awaiting_pull') { + return this.#commandReconciliationRemoteId(command) ?? this.#generatedRemoteId(locator, userId, generation); + } + if (command.state === 'sending') { + const activeSend = this.#commandSendTransitions.get(this.#commandSendKey(command, generation)); + if (activeSend) await activeSend; + return this.#generatedCommandOutcome(commandId, locator, generation, userId); + } + if (command.state !== 'pending') return null; + const aggregate = known.filter((candidate) => this.#aggregateKey(candidate) === this.#aggregateKey(command)); + if (aggregate[0]?.commandId !== command.commandId) { + throw new Error('Immediate generated command delivery requires the first aggregate intent.'); + } + const row = await this.#rowForCommand(command); + if ( + !row || + row.identity.kind !== 'generated' || + row.identity.remoteId !== null || + row.confirmedValues !== null || + command.identity.kind !== 'generated' || + command.baseRevision !== null || + command.replicaMutation === 'delete' + ) { + throw new Error('Immediate generated command delivery requires a new unconfirmed aggregate.'); + } + const scope = { userId: command.userId, scopeId: command.scopeId }; + const scopeKey = this.#scopeKey(scope); + if (!this.#knownScopes.has(scopeKey)) return null; + const dirtyScopes = new Map(); + await this.#sendAggregate([command], generation, dirtyScopes, new Set([scopeKey])); + if (!this.#isCurrentUser(generation, userId)) return null; + for (const dirtyScope of dirtyScopes.values()) { + this.#pendingPullScopes.set(this.#scopeKey(dirtyScope), dirtyScope); + } + return this.#generatedCommandOutcome(commandId, locator, generation, userId); + } + + async #generatedCommandOutcome( + commandId: string, + locator: OfflineGeneratedCommandLocator, + generation: number, + userId: OfflinePrincipalId, + ): Promise { + if (!this.#isCurrentUser(generation, userId)) return null; + await this.#refreshState(generation); + if (!this.#isCurrentUser(generation, userId)) return null; + const completed = (await this.#readKnownCommands()).find((candidate) => candidate.commandId === commandId); + if (!this.#isCurrentUser(generation, userId)) return null; + if (completed && ['blocked_auth', 'rejected', 'conflict'].includes(completed.state)) { + throw new OfflineImmediateCommandRejectedError(completed.state as 'blocked_auth' | 'rejected' | 'conflict', completed.lastErrorCode); + } + return completed + ? (this.#commandReconciliationRemoteId(completed) ?? this.#generatedRemoteId(locator, userId, generation)) + : this.#generatedRemoteId(locator, userId, generation); + } + + async #generatedRemoteId( + locator: OfflineGeneratedCommandLocator, + userId: OfflinePrincipalId, + generation: number, + ): Promise { + if (!this.#isCurrentUser(generation, userId)) return null; + const row = await this.#repository.getReplicaRow({ userId, scopeId: locator.scopeId }, locator.sourceKey, { + kind: 'generated', + localId: locator.localId, + }); + return this.#isCurrentUser(generation, userId) && row?.identity.kind === 'generated' ? row.identity.remoteId : null; + } + + #commandReconciliationRemoteId(command: OfflineCommand): OfflineGeneratedRemoteId | null { + const identity = command.reconciliationIdentity; + return identity && 'remoteId' in identity ? (identity.remoteId ?? null) : null; + } + #beginFlush(explicitFull: boolean): Promise { const lifecycleRevision = this.#lifecycleRevision(); const isPartial = !explicitFull && this.#foregroundScopePolicy !== null; @@ -1200,73 +1382,106 @@ export class OfflineSyncService { pulledScopeKeys: ReadonlySet, ): Promise { for (const command of commands) { - if (!this.#isCurrent(generation)) return; - if (command.state === 'awaiting_pull') continue; - 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; - await this.#refreshState(generation); - if (!this.#isCurrent(generation)) return; - const claimedCommand = sending; - const readRow = async (): Promise => this.#rowForCommand(claimedCommand); - const rowResult = await readRow().then( - (row) => ({ status: 'fulfilled' as const, row }), - (error: unknown) => ({ status: 'rejected' as const, error }), - ); - if (rowResult.status === 'rejected') { - if (!this.#isCurrent(generation)) return; - await this.#persistFailedCommand(sending, rowResult.error, generation, null, sending.serverCommitUnknown === true); - throw rowResult.error; - } - const row = rowResult.row; - if (!row) { - const error = new Error( - `Offline replica row not found: ${sending.aggregateType}/${canonicalOfflineCommandIdentity(sending.identity)}`, - ); - await this.#persistFailedCommand(sending, error, generation, null, sending.serverCommitUnknown === true); - throw error; - } - const priorCommitUnknown = sending.serverCommitUnknown === true; - const transportCommand = await this.#markTransportStarted(sending, generation); - if (!transportCommand) return; - sending = transportCommand; - const executeCommand = async (): Promise => - this.#executor.execute(sending, offlineCommandTargetFromReplicaRow(row)); - const execution = await executeCommand().then( - (result) => ({ status: 'fulfilled' as const, result }), - (error: unknown) => ({ status: 'rejected' as const, error }), - ); - if (execution.status === 'rejected') { - if (!this.#isCurrent(generation)) return; - const commitUnknown = this.#executor.provesCommandNotCommitted?.(execution.error, sending) - ? false - : priorCommitUnknown || this.#serverCommitCouldBeUnknown(execution.error); - await this.#persistFailedCommand(sending, execution.error, generation, row, commitUnknown); - if (!this.#isClassifiableTransportError(execution.error)) throw execution.error; - break; + if (!(await this.#sendCommandOnce(commands, command, generation, dirtyScopes, pulledScopeKeys))) break; + } + } + + #sendCommandOnce( + aggregateCommands: OfflineCommand[], + command: OfflineCommand, + generation: number, + dirtyScopes: Map, + pulledScopeKeys: ReadonlySet, + ): Promise { + const transitionKey = this.#commandSendKey(command, generation); + const active = this.#commandSendTransitions.get(transitionKey); + if (active) return active; + const transition = this.#sendCommand(aggregateCommands, command, generation, dirtyScopes, pulledScopeKeys).finally(() => { + if (this.#commandSendTransitions.get(transitionKey) === transition) { + this.#commandSendTransitions.delete(transitionKey); } - const result = execution.result; - if (!this.#isCurrent(generation)) return; - const completeCommand = async (): Promise => this.#completeCommandWithRetry(commands, sending, result, generation); - const completion = await completeCommand().then( - () => ({ status: 'fulfilled' as const }), - (error: unknown) => ({ status: 'rejected' as const, error }), + }); + this.#commandSendTransitions.set(transitionKey, transition); + return transition; + } + + #commandSendKey(command: OfflineCommand, generation: number): string { + return JSON.stringify([generation, canonicalOfflinePrincipalId(command.userId), command.commandId]); + } + + async #sendCommand( + aggregateCommands: OfflineCommand[], + command: OfflineCommand, + generation: number, + dirtyScopes: Map, + pulledScopeKeys: ReadonlySet, + ): Promise { + if (!this.#isCurrent(generation)) return false; + if (command.state === 'awaiting_pull') return true; + if (command.state === 'retry_wait' && (command.retryAt ?? 0) > Date.now()) return false; + if (!['pending', 'retry_wait'].includes(command.state)) return false; + // 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 }))) return false; + let sending = await this.#claimSendingCommand(command, generation); + if (!sending || !this.#isCurrent(generation)) return false; + await this.#refreshState(generation); + if (!this.#isCurrent(generation)) return false; + const claimedCommand = sending; + const readRow = async (): Promise => this.#rowForCommand(claimedCommand); + const rowResult = await readRow().then( + (row) => ({ status: 'fulfilled' as const, row }), + (error: unknown) => ({ status: 'rejected' as const, error }), + ); + if (rowResult.status === 'rejected') { + if (!this.#isCurrent(generation)) return false; + await this.#persistFailedCommand(sending, rowResult.error, generation, null, sending.serverCommitUnknown === true); + throw rowResult.error; + } + const row = rowResult.row; + if (!row) { + const error = new Error( + `Offline replica row not found: ${sending.aggregateType}/${canonicalOfflineCommandIdentity(sending.identity)}`, ); - if (completion.status === 'rejected') { - if (!this.#isCurrent(generation)) return; - await this.#persistFailedCommand(sending, completion.error, generation, row, true); - throw completion.error; - } - if (this.#isCurrent(generation)) { - const scope = { userId: sending.userId, scopeId: sending.scopeId }; - dirtyScopes.set(this.#scopeKey(scope), scope); - } + await this.#persistFailedCommand(sending, error, generation, null, sending.serverCommitUnknown === true); + throw error; + } + const priorCommitUnknown = sending.serverCommitUnknown === true; + const transportCommand = await this.#markTransportStarted(sending, generation); + if (!transportCommand) return false; + sending = transportCommand; + const executeCommand = async (): Promise => + this.#executor.execute(sending, offlineCommandTargetFromReplicaRow(row)); + const execution = await executeCommand().then( + (result) => ({ status: 'fulfilled' as const, result }), + (error: unknown) => ({ status: 'rejected' as const, error }), + ); + if (execution.status === 'rejected') { + if (!this.#isCurrent(generation)) return false; + const commitUnknown = this.#executor.provesCommandNotCommitted?.(execution.error, sending) + ? false + : priorCommitUnknown || this.#serverCommitCouldBeUnknown(execution.error); + await this.#persistFailedCommand(sending, execution.error, generation, row, commitUnknown); + if (!this.#isClassifiableTransportError(execution.error)) throw execution.error; + return false; } + const result = execution.result; + if (!this.#isCurrent(generation)) return false; + const completeCommand = async (): Promise => this.#completeCommandWithRetry(aggregateCommands, sending, result, generation); + const completion = await completeCommand().then( + () => ({ status: 'fulfilled' as const }), + (error: unknown) => ({ status: 'rejected' as const, error }), + ); + if (completion.status === 'rejected') { + if (!this.#isCurrent(generation)) return false; + await this.#persistFailedCommand(sending, completion.error, generation, row, true); + throw completion.error; + } + if (this.#isCurrent(generation)) { + const scope = { userId: sending.userId, scopeId: sending.scopeId }; + dirtyScopes.set(this.#scopeKey(scope), scope); + } + return true; } async #normalizeEnqueueRequest( @@ -1796,6 +2011,10 @@ export class OfflineSyncService { return generation === this.#generation; } + #isCurrentUser(generation: number, userId: OfflinePrincipalId): boolean { + return this.#isCurrent(generation) && this.#activeUserId === userId; + } + async #restoreInterruptedCommands(): Promise { const commands = await this.#readKnownCommands(); await Promise.all( @@ -1880,7 +2099,7 @@ export class OfflineSyncService { } async #waitForSendingTransitions(): Promise { - await Promise.allSettled([...this.#sendingTransitions]); + await Promise.allSettled([...this.#sendingTransitions, ...this.#commandSendTransitions.values()]); } #canonicalJson(value: unknown): string { diff --git a/projects/kit/offline/src/lib/offline.interceptor.spec.ts b/projects/kit/offline/src/lib/offline.interceptor.spec.ts index d7261b7..571c35f 100644 --- a/projects/kit/offline/src/lib/offline.interceptor.spec.ts +++ b/projects/kit/offline/src/lib/offline.interceptor.spec.ts @@ -236,6 +236,22 @@ describe('offlineInterceptor', () => { expect(readLocal).not.toHaveBeenCalled(); }); + it('local-only readはremote transportを開始せずlocal responseだけを返す', async () => { + const local = new HttpResponse({ body: [{ id: 'provisional' }], status: 200 }); + const readLocal = vi.fn(async () => local); + resolve.mockReturnValue({ kind: 'read', readStrategy: 'local-only', readLocal }); + const next = vi.fn(() => of(new HttpResponse({ body: [{ id: 'remote' }], status: 200 }))); + + const response = await firstValueFrom(run(new HttpRequest('GET', '/documents/provisional'), next)); + + expect(next).not.toHaveBeenCalled(); + expect(readLocal).toHaveBeenCalledOnce(); + expect(response instanceof HttpResponse && response.body).toEqual([{ id: 'provisional' }]); + expect(response instanceof HttpResponse && response.headers.get(OFFLINE_RESPONSE_HEADER)).toBe('local'); + expect(markApiSuccess).not.toHaveBeenCalled(); + expect(markApiFailure).not.toHaveBeenCalled(); + }); + it('未登録POSTはread policyを解決せずtransportへ渡しreachabilityを更新する', async () => { resolve.mockReturnValue({ kind: 'read', readLocal: vi.fn() }); const request = new HttpRequest('POST', '/groups/1/documents', {}); diff --git a/projects/kit/offline/src/lib/offline.interceptor.ts b/projects/kit/offline/src/lib/offline.interceptor.ts index d5631a4..c7f4d39 100644 --- a/projects/kit/offline/src/lib/offline.interceptor.ts +++ b/projects/kit/offline/src/lib/offline.interceptor.ts @@ -48,6 +48,9 @@ export const offlineInterceptor: HttpInterceptorFn = (request, next) => { const fallback = inject(OfflineRequestFallbackService); const plan = registry.resolve(request); if (!plan) return transport(); + if (plan.readStrategy === 'local-only') { + return readLocalOnly(plan, inject(ErrorHandler), inject(OfflineReplicaMutationCoordinator)); + } if (plan.readStrategy === 'local-first') { return readLocalFirst(request, plan, transport, fallback, inject(ErrorHandler), inject(OfflineReplicaMutationCoordinator)); } @@ -68,6 +71,15 @@ export const offlineInterceptor: HttpInterceptorFn = (request, next) => { return transport(); }; +/** Resolves a provisional/local identity without starting remote transport. */ +function readLocalOnly( + plan: OfflineReadRequestPlan, + errorHandler: ErrorHandler, + replicaMutations: OfflineReplicaMutationCoordinator, +): Observable> { + return resolveLocalAttempt(plan, errorHandler, replicaMutations).pipe(concatMap((response) => (response ? of(response) : EMPTY))); +} + function readNetworkFirst( request: HttpRequest, plan: OfflineReadRequestPlan,