diff --git a/src/app/op-log/persistence/operation-log-compaction.service.spec.ts b/src/app/op-log/persistence/operation-log-compaction.service.spec.ts index 140493cb41..34c9c6aefb 100644 --- a/src/app/op-log/persistence/operation-log-compaction.service.spec.ts +++ b/src/app/op-log/persistence/operation-log-compaction.service.spec.ts @@ -14,11 +14,25 @@ import { OperationLogEntry } from '../core/operation.types'; import { OpLog } from '../../core/log'; import { MODEL_CONFIGS } from '../model/model-config'; import { CLIENT_ID_PROVIDER, ClientIdProvider } from '../util/client-id.provider'; +import { OperationCaptureService } from '../capture/operation-capture.service'; +import { OperationWriteFlushService } from '../sync/operation-write-flush.service'; +import { + bufferDeferredAction, + clearDeferredActions, +} from '../capture/operation-capture.meta-reducer'; const MS_PER_DAY = 24 * 60 * 60 * 1000; +// Minimal persistent action for driving the real capture-service pending +// counter in the quiesce tests (#8469). +const FAKE_PENDING_ACTION = { + type: '[Task] Test pending action', + meta: { entityType: 'TASK', entityId: 't-pending' }, +} as any; + describe('OperationLogCompactionService', () => { let service: OperationLogCompactionService; + let captureService: OperationCaptureService; let mockOpLogStore: jasmine.SpyObj; let mockLockService: jasmine.SpyObj; let mockStateSnapshot: jasmine.SpyObj; @@ -78,6 +92,10 @@ describe('OperationLogCompactionService', () => { TestBed.configureTestingModule({ providers: [ OperationLogCompactionService, + // Real quiesce pipeline (flush + capture counter) on top of the mocked + // lock, so the #8469 tests exercise the actual drain behavior. + OperationWriteFlushService, + OperationCaptureService, { provide: OperationLogStoreService, useValue: mockOpLogStore }, { provide: LockService, useValue: mockLockService }, { provide: StateSnapshotService, useValue: mockStateSnapshot }, @@ -87,6 +105,11 @@ describe('OperationLogCompactionService', () => { }); service = TestBed.inject(OperationLogCompactionService); + captureService = TestBed.inject(OperationCaptureService); + // The compaction bail reads the MODULE-LEVEL deferred buffer (#8469); + // start clean so a leak from another spec can't fail these tests + // order-dependently under jasmine's random order. + clearDeferredActions(); }); describe('compact', () => { @@ -135,7 +158,12 @@ describe('OperationLogCompactionService', () => { it('should not log metrics if compaction is fast', async () => { await service.compact(); - expect(OpLog.normal).not.toHaveBeenCalled(); + // The flush pre-pass logs its own progress lines; only the compaction + // metrics line must be absent. + expect(OpLog.normal).not.toHaveBeenCalledWith( + 'OperationLogCompactionService: Compaction completed', + jasmine.anything(), + ); }); it('should get current vector clock', async () => { @@ -447,9 +475,10 @@ describe('OperationLogCompactionService', () => { await service.compact(); - // All operations should happen between lock-start and lock-end - const lockStartIndex = callOrder.indexOf('lock-start'); - const lockEndIndex = callOrder.indexOf('lock-end'); + // The flush pre-pass acquires/releases the lock once on its own; all + // compaction steps must happen inside the LAST lock window. + const lockStartIndex = callOrder.lastIndexOf('lock-start'); + const lockEndIndex = callOrder.lastIndexOf('lock-end'); expect(callOrder.indexOf('getState')).toBeGreaterThan(lockStartIndex); expect(callOrder.indexOf('getState')).toBeLessThan(lockEndIndex); @@ -608,6 +637,87 @@ describe('OperationLogCompactionService', () => { }); }); + describe('quiesced capture (#8469)', () => { + it('compact() should wait for in-flight op writes to drain before capturing', async () => { + // An action whose reducer has already run but whose op write is still + // queued must end up covered by the saved lastAppliedOpSeq — otherwise + // the next boot's tail replay re-applies an op whose effect is already + // baked into the cached state (double-applying non-idempotent reducers). + captureService.incrementPending(FAKE_PENDING_ACTION); + let opWriteDurable = false; + mockOpLogStore.getLastSeq.and.callFake(async () => (opWriteDurable ? 101 : 100)); + + // Simulate the persist effect completing the queued write while the + // compaction's flush pre-pass is polling. + setTimeout(() => { + opWriteDurable = true; + captureService.decrementPending(); + }, 30); + + const result = await service.compact(); + + expect(result).toBeTrue(); + expect(mockOpLogStore.saveStateCache).toHaveBeenCalledWith( + jasmine.objectContaining({ lastAppliedOpSeq: 101 }), + ); + }); + + it('compact() should bail without saving when a local write lands mid-capture', async () => { + // A dispatch landing during the locked body's awaits (the pending + // remote-ops read) would be baked into the captured state while its op + // is still unsequenced — the cache would be tagged behind an op it + // already contains. The re-check at the state-read cutoff must bail; + // compaction re-triggers on the next threshold. + mockOpLogStore.getPendingRemoteOps.and.callFake(async () => { + captureService.incrementPending(FAKE_PENDING_ACTION); + return []; + }); + + const result = await service.compact(); + + expect(result).toBeFalse(); + expect(mockOpLogStore.saveStateCache).not.toHaveBeenCalled(); + expect(mockOpLogStore.deleteOpsWhere).not.toHaveBeenCalled(); + }); + + it('compact() should bail without saving when deferred actions are pending durable write', async () => { + // Deferred actions (buffered during a sync window, kept across windows + // after a failed drain) have their reducer effects in state but no seq + // yet and are invisible to the pending counter — capturing would tag + // the cache behind their future seqs. + bufferDeferredAction(FAKE_PENDING_ACTION); + try { + const result = await service.compact(); + + expect(result).toBeFalse(); + expect(mockOpLogStore.saveStateCache).not.toHaveBeenCalled(); + expect(mockOpLogStore.deleteOpsWhere).not.toHaveBeenCalled(); + } finally { + // Module-level buffer persists across TestBed resets — always clean up. + clearDeferredActions(); + } + }); + + it('emergencyCompact() should NOT wait for pending writes (quota recovery)', async () => { + // Emergency compaction is invoked from the failing write's own call + // stack, where that write's pending-counter entry is still elevated. + // Draining the capture pipeline there would wait on ourselves until the + // flush timeout and break quota recovery — the emergency path skips the + // quiesce (pending counter AND deferred buffer) and accepts the + // residual re-replay window (#8469). + captureService.incrementPending(FAKE_PENDING_ACTION); // never drained + bufferDeferredAction(FAKE_PENDING_ACTION); + try { + const result = await service.emergencyCompact(); + + expect(result).toBeTrue(); + expect(mockOpLogStore.saveStateCache).toHaveBeenCalled(); + } finally { + clearDeferredActions(); + } + }); + }); + // ========================================================================= // Snapshot entity keys extraction tests // ========================================================================= @@ -911,8 +1021,9 @@ describe('OperationLogCompactionService', () => { await service.compact(); await service.compact(); - // Lock should have been requested 3 times - expect(lockRequests.length).toBe(3); + // At least one acquisition per compact(); the flush pre-pass (#8469 + // quiesce) may add more — that's an internal detail, not the contract. + expect(lockRequests.length).toBeGreaterThanOrEqual(3); }); it('should always request lock with correct name', async () => { @@ -990,8 +1101,12 @@ describe('OperationLogCompactionService', () => { expect(regularResult).toBe('regular-done'); expect(emergencyResult).toBe('emergency-done'); - // Lock should have been acquired twice (serialized) - expect(callOrder.filter((c) => c === 'lock-start').length).toBe(2); + // Both paths acquire the lock (serialized): regular compact at least + // once plus the flush pre-pass (#8469 quiesce), emergency exactly once + // (bare lock — pinned separately by the quiesced-capture tests). + expect(callOrder.filter((c) => c === 'lock-start').length).toBeGreaterThanOrEqual( + 2, + ); }); it('should not delete operations that arrive during compaction', async () => { @@ -1096,22 +1211,24 @@ describe('OperationLogCompactionService', () => { // to prevent data corruption from lock expiration. describe('compaction timeout handling', () => { + // NOTE: keep the capture-service pending counter at 0 in these tests — + // with the clock frozen by the Date.now spy, the flush pre-pass's poll + // loop would never trip its MAX_WAIT_TIME and would spin on real timers + // until the jasmine timeout (a confusing hang, not a clean failure). it('should throw error when compaction exceeds timeout', async () => { - // Simulate slow state retrieval that exceeds the 25s timeout + // Simulate slow pending-remote-ops retrieval that exceeds the 25s + // timeout. The jump is keyed off getPendingRemoteOps (the first step + // inside the compaction body) rather than a Date.now call count, so the + // flush pre-pass (#8469) reading the clock first doesn't skew it. const originalDateNow = Date.now; - let callCount = 0; + let timeJumped = false; - // Mock Date.now to simulate time passing during compaction - spyOn(Date, 'now').and.callFake(() => { - callCount++; - // First call is startTime, subsequent calls simulate 26s elapsed - if (callCount === 1) { - return 0; - } - return 26000; // 26 seconds - exceeds 25s timeout + spyOn(Date, 'now').and.callFake(() => (timeJumped ? 26000 : 0)); + mockOpLogStore.getPendingRemoteOps.and.callFake(async () => { + timeJumped = true; // 26 seconds elapse "during" the read + return []; }); - // Make state retrieval trigger a timeout check mockStateSnapshot.getStateSnapshot.and.returnValue(mockState); await expectAsync(service.compact()).toBeRejectedWithError( @@ -1130,14 +1247,13 @@ describe('OperationLogCompactionService', () => { }); it('should include phase info in timeout error message', async () => { - let callCount = 0; + // Same clock-jump keying as above — see the first timeout test. + let timeJumped = false; - spyOn(Date, 'now').and.callFake(() => { - callCount++; - if (callCount === 1) { - return 0; - } - return 26000; + spyOn(Date, 'now').and.callFake(() => (timeJumped ? 26000 : 0)); + mockOpLogStore.getPendingRemoteOps.and.callFake(async () => { + timeJumped = true; + return []; }); mockStateSnapshot.getStateSnapshot.and.returnValue(mockState); diff --git a/src/app/op-log/persistence/operation-log-compaction.service.ts b/src/app/op-log/persistence/operation-log-compaction.service.ts index 218d22ebb9..0814346c63 100644 --- a/src/app/op-log/persistence/operation-log-compaction.service.ts +++ b/src/app/op-log/persistence/operation-log-compaction.service.ts @@ -16,6 +16,9 @@ import { extractEntityKeysFromState } from './extract-entity-keys'; import { CLIENT_ID_PROVIDER, ClientIdProvider } from '../util/client-id.provider'; import { limitVectorClockSize } from '../../core/util/vector-clock'; import { hasMeaningfulStateData } from '../validation/has-meaningful-state-data.util'; +import { OperationCaptureService } from '../capture/operation-capture.service'; +import { OperationWriteFlushService } from '../sync/operation-write-flush.service'; +import { getDeferredActions } from '../capture/operation-capture.meta-reducer'; /** * Manages the compaction (garbage collection) of the operation log. @@ -32,6 +35,8 @@ export class OperationLogCompactionService { private stateSnapshot = inject(StateSnapshotService); private vectorClockService = inject(VectorClockService); private clientIdProvider: ClientIdProvider = inject(CLIENT_ID_PROVIDER); + private operationCapture = inject(OperationCaptureService); + private writeFlushService = inject(OperationWriteFlushService); async compact(): Promise { return this._doCompact(COMPACTION_RETENTION_MS, false); @@ -57,7 +62,7 @@ export class OperationLogCompactionService { * @param isEmergency - Whether this is an emergency compaction (for logging) */ private async _doCompact(retentionMs: number, isEmergency: boolean): Promise { - return this.lockService.request(LOCK_NAMES.OPERATION_LOG, async () => { + const compactExclusively = async (): Promise => { const startTime = Date.now(); const label = isEmergency ? 'emergency ' : ''; @@ -73,6 +78,28 @@ export class OperationLogCompactionService { return false; } + // #8469: the await above may have let a local dispatch land. Its reducer + // effect would be captured in the state below while its op is still + // unsequenced, so the cache would be tagged behind an op it already + // contains and the next boot would re-apply it (double-applying + // non-idempotent reducers). Deferred actions (buffered during a sync + // window, kept across windows after a failed drain) are in the same + // position but invisible to the pending counter, so check both. The + // check and the state read run in one synchronous block — no dispatch + // can interleave, and new deferrals cannot start while we hold the lock + // (the sync window that buffers them requires it). Bail; compaction + // re-triggers on the next threshold. The emergency path skips this: the + // failing write itself keeps the counter elevated (see below). + if ( + !isEmergency && + (this.operationCapture.getPendingCount() > 0 || getDeferredActions().length > 0) + ) { + OpLog.warn( + 'OperationLogCompactionService: Skipping compaction — local writes are pending (in-flight or deferred)', + ); + return false; + } + // 1. Get current state from NgRx store const currentState = this.stateSnapshot.getStateSnapshotForOperationLog(); this.checkCompactionTimeout(startTime, `${label}state snapshot`); @@ -161,7 +188,20 @@ export class OperationLogCompactionService { } return true; - }); + }; + + // #8469: drain the capture pipeline before capturing so no action can be + // dispatched-but-unsequenced at the state read — otherwise its effect is + // baked into the cache while its seq lands after lastAppliedOpSeq, and the + // next boot's tail replay double-applies it. Emergency compaction is + // invoked from the failing write's own call stack (quota handling), where + // that write's pending-counter entry is still elevated — flushing there + // would wait on ourselves until the flush timeout and break quota + // recovery, so it keeps the bare lock and accepts the residual re-replay + // window. + return isEmergency + ? this.lockService.request(LOCK_NAMES.OPERATION_LOG, compactExclusively) + : this.writeFlushService.flushThenRunExclusive(compactExclusively); } /** diff --git a/src/app/op-log/persistence/operation-log-snapshot.service.spec.ts b/src/app/op-log/persistence/operation-log-snapshot.service.spec.ts index cb550556ef..3eae048231 100644 --- a/src/app/op-log/persistence/operation-log-snapshot.service.spec.ts +++ b/src/app/op-log/persistence/operation-log-snapshot.service.spec.ts @@ -12,6 +12,12 @@ import { CLIENT_ID_PROVIDER, ClientIdProvider } from '../util/client-id.provider import { ValidateStateService } from '../validation/validate-state.service'; import { LockService } from '../sync/lock.service'; import { LOCK_NAMES } from '../core/operation-log.const'; +import { OperationCaptureService } from '../capture/operation-capture.service'; +import { OperationWriteFlushService } from '../sync/operation-write-flush.service'; +import { + bufferDeferredAction, + clearDeferredActions, +} from '../capture/operation-capture.meta-reducer'; import { MAX_VECTOR_CLOCK_SIZE } from '@sp/sync-core'; // Meaningful state (contains a task) so saveCurrentStateAsSnapshot proceeds past @@ -22,8 +28,16 @@ const MEANINGFUL_SNAPSHOT_STATE = { task: { ids: ['t1'], entities: { t1: { id: 't1' } } }, } as unknown; +// Minimal persistent action for driving the real capture-service pending +// counter in the quiesce tests (#8469). +const FAKE_PENDING_ACTION = { + type: '[Task] Test pending action', + meta: { entityType: 'TASK', entityId: 't-pending' }, +} as any; + describe('OperationLogSnapshotService', () => { let service: OperationLogSnapshotService; + let captureService: OperationCaptureService; let mockOpLogStore: jasmine.SpyObj; let mockVectorClockService: jasmine.SpyObj; let mockStateSnapshotService: jasmine.SpyObj; @@ -71,6 +85,10 @@ describe('OperationLogSnapshotService', () => { TestBed.configureTestingModule({ providers: [ OperationLogSnapshotService, + // Real quiesce pipeline (flush + capture counter) on top of the mocked + // lock, so the #8469 tests exercise the actual drain behavior. + OperationWriteFlushService, + OperationCaptureService, { provide: OperationLogStoreService, useValue: mockOpLogStore }, { provide: VectorClockService, useValue: mockVectorClockService }, { provide: StateSnapshotService, useValue: mockStateSnapshotService }, @@ -81,6 +99,11 @@ describe('OperationLogSnapshotService', () => { ], }); service = TestBed.inject(OperationLogSnapshotService); + captureService = TestBed.inject(OperationCaptureService); + // The save bails on a non-empty MODULE-LEVEL deferred buffer (#8469); + // start clean so a leak from another spec can't fail these tests + // order-dependently under jasmine's random order. + clearDeferredActions(); }); describe('isValidSnapshot', () => { @@ -348,7 +371,7 @@ describe('OperationLogSnapshotService', () => { ); }); - it('should read lastSeq BEFORE getStateSnapshot inside the lock', async () => { + it('should read state BEFORE lastSeq inside the lock (quiesced capture, #8469)', async () => { const callOrder: string[] = []; mockLockService.request.and.callFake( @@ -375,17 +398,24 @@ describe('OperationLogSnapshotService', () => { await service.saveCurrentStateAsSnapshot(); - // All operations should happen between lock-start and lock-end - const lockStartIndex = callOrder.indexOf('lock-start'); - const lockEndIndex = callOrder.indexOf('lock-end'); + // The flush pre-pass acquires/releases the lock once on its own; the + // capture body runs inside the LAST lock window. + const lockStartIndex = callOrder.lastIndexOf('lock-start'); + const lockEndIndex = callOrder.lastIndexOf('lock-end'); expect(callOrder.indexOf('getLastSeq')).toBeGreaterThan(lockStartIndex); expect(callOrder.indexOf('getLastSeq')).toBeLessThan(lockEndIndex); expect(callOrder.indexOf('getStateSnapshot')).toBeGreaterThan(lockStartIndex); expect(callOrder.indexOf('getStateSnapshot')).toBeLessThan(lockEndIndex); - // Key invariant: lastSeq is read BEFORE state snapshot - expect(callOrder.indexOf('getLastSeq')).toBeLessThan( - callOrder.indexOf('getStateSnapshot'), + // Key invariant (#8469): with the capture pipeline drained and the lock + // held, no op can gain a seq during the body. State is read first + // (synchronously, before any await can let a dispatch interleave) and + // lastSeq after — so every op with seq <= lastAppliedOpSeq has its + // effect in the captured state, and every later dispatch is absent from + // it and replays cleanly. (Pre-quiesce the order was inverted to bias + // the race toward re-replay instead of op-loss.) + expect(callOrder.indexOf('getStateSnapshot')).toBeLessThan( + callOrder.indexOf('getLastSeq'), ); }); @@ -397,6 +427,71 @@ describe('OperationLogSnapshotService', () => { }); }); + describe('saveCurrentStateAsSnapshot — quiesced capture (#8469)', () => { + beforeEach(() => { + mockStateSnapshotService.getStateSnapshot.and.returnValue( + MEANINGFUL_SNAPSHOT_STATE as any, + ); + mockVectorClockService.getCurrentVectorClock.and.resolveTo({}); + mockOpLogStore.saveStateCache.and.resolveTo(undefined); + }); + + it('should wait for in-flight op writes to drain before capturing', async () => { + // An action whose reducer has already run but whose op write is still + // queued must end up covered by the saved lastAppliedOpSeq — otherwise + // the next boot's tail replay re-applies an op whose effect is already + // baked into the snapshot state (double-applying non-idempotent + // reducers: accumulating time/metric deltas, plain-append branches). + captureService.incrementPending(FAKE_PENDING_ACTION); + let opWriteDurable = false; + mockOpLogStore.getLastSeq.and.callFake(async () => (opWriteDurable ? 11 : 10)); + + // Simulate the persist effect completing the queued write while the + // snapshot's flush pre-pass is polling. + setTimeout(() => { + opWriteDurable = true; + captureService.decrementPending(); + }, 30); + + await service.saveCurrentStateAsSnapshot(); + + expect(mockOpLogStore.saveStateCache).toHaveBeenCalledWith( + jasmine.objectContaining({ lastAppliedOpSeq: 11 }), + ); + }); + + it('should skip the save without throwing when the pipeline cannot quiesce', async () => { + const writeFlushService = TestBed.inject(OperationWriteFlushService); + spyOn(writeFlushService, 'flushThenRunExclusive').and.rejectWith( + new Error('Operation write cutoff not reached'), + ); + mockOpLogStore.getLastSeq.and.resolveTo(1); + + // Skipping is always correctness-safe: the snapshot is only a boot-time + // cache and the op-log stays the source of truth. + await expectAsync(service.saveCurrentStateAsSnapshot()).toBeResolved(); + expect(mockOpLogStore.saveStateCache).not.toHaveBeenCalled(); + }); + + it('should skip the save when deferred actions are pending durable write', async () => { + // Deferred actions (buffered during a sync window, kept across windows + // after a failed drain) have their reducer effects in state but no seq + // yet and are invisible to the pending counter — capturing would tag + // the snapshot behind their future seqs. + bufferDeferredAction(FAKE_PENDING_ACTION); + try { + mockOpLogStore.getLastSeq.and.resolveTo(1); + + await service.saveCurrentStateAsSnapshot(); + + expect(mockOpLogStore.saveStateCache).not.toHaveBeenCalled(); + } finally { + // Module-level buffer persists across TestBed resets — always clean up. + clearDeferredActions(); + } + }); + }); + describe('migrateSnapshotWithBackup', () => { const createSnapshot = (): MigratableStateCache => ({ state: { task: {}, project: {}, globalConfig: {} }, diff --git a/src/app/op-log/persistence/operation-log-snapshot.service.ts b/src/app/op-log/persistence/operation-log-snapshot.service.ts index 4dac9eb706..76bb519bef 100644 --- a/src/app/op-log/persistence/operation-log-snapshot.service.ts +++ b/src/app/op-log/persistence/operation-log-snapshot.service.ts @@ -13,8 +13,8 @@ import { CLIENT_ID_PROVIDER, ClientIdProvider } from '../util/client-id.provider import { limitVectorClockSize } from '../../core/util/vector-clock'; import { ValidateStateService } from '../validation/validate-state.service'; import { hasMeaningfulStateData } from '../validation/has-meaningful-state-data.util'; -import { LockService } from '../sync/lock.service'; -import { LOCK_NAMES } from '../core/operation-log.const'; +import { OperationWriteFlushService } from '../sync/operation-write-flush.service'; +import { getDeferredActions } from '../capture/operation-capture.meta-reducer'; type StateCache = MigratableStateCache; @@ -37,7 +37,7 @@ export class OperationLogSnapshotService { private schemaMigrationService = inject(SchemaMigrationService); private validateStateService = inject(ValidateStateService); private clientIdProvider: ClientIdProvider = inject(CLIENT_ID_PROVIDER); - private lockService = inject(LockService); + private writeFlushService = inject(OperationWriteFlushService); /** * Validates that a snapshot has the expected structure and data. @@ -82,29 +82,54 @@ export class OperationLogSnapshotService { * Saves the current NgRx state as a snapshot for faster future loads. * Called after replaying many operations to optimize next startup. * - * Wrapped in OPERATION_LOG lock to prevent a lost-update window: without - * the lock, an op appended between reading NgRx state and reading lastSeq - * would get a seq <= lastAppliedOpSeq but its effect would be absent from - * the snapshot. On next hydration the tail replay would start after that - * seq, silently skipping the op forever. + * Runs via flushThenRunExclusive (#8469): the capture pipeline is drained + * BEFORE the OPERATION_LOG lock is taken and the pending counter re-checked + * once inside it, so no counted action can be dispatched-but-unsequenced at + * the capture point. NgRx state mutates synchronously at dispatch while the + * op's seq is only assigned later in the persist effect — without the + * quiesce, an op whose reducer had already run but whose seq-write was + * still queued behind the lock would be baked into the snapshot state AND + * tail-replayed on the next boot, double-applying non-idempotent reducers + * (accumulating time/metric deltas, plain-append list branches). * - * Additionally, lastSeq is read BEFORE the state snapshot so the worst - * interleaving degrades to re-replay rather than a missed op. Most op types - * are idempotent on re-replay (entity-adapter CRUD/move/reorder), but some - * persistent ops accumulate onto current state and double-apply on re-replay - * (e.g. time-tracking/counter deltas and the plain-append branches of project - * task moves). This edge is unlikely (the save runs only during hydration) and - * strictly better than op-loss. The audited op list and a truly-lossless - * capture (quiesce the op-write queue before reading) are tracked in #8469. + * Deferred actions (buffered during a sync window, kept across windows + * after a failed drain) are NOT covered by the pending counter; they are + * handled by an explicit skip inside the body. + * + * Inside the lock, state is read first — synchronously, before any await + * can let a dispatch interleave — and lastSeq after. While the lock is held + * no new seq can be assigned, so every op with seq <= lastAppliedOpSeq has + * its effect in the captured state, and any later dispatch is absent from + * it and replays cleanly. This order relies on EVERY seq-assigning writer + * holding the OPERATION_LOG lock (persist effect, remote apply, repair, + * server migration — all do): a writer bypassing the lock would degrade + * this window to silent op-LOSS, strictly worse than the pre-quiesce + * re-replay bias. + * + * Failure (flush timeout, persistent dispatch activity) only skips the + * save: the snapshot is a boot-time cache and the op-log stays the source + * of truth. */ async saveCurrentStateAsSnapshot(): Promise { try { - await this.lockService.request(LOCK_NAMES.OPERATION_LOG, async () => { - // Read lastSeq BEFORE state snapshot — see JSDoc above. - // NOTE: compaction reads in the opposite order (state, then lastSeq); - // its failure mode if the lock is bypassed is missed-op data loss. - const lastSeq = await this.opLogStore.getLastSeq(); + await this.writeFlushService.flushThenRunExclusive(async () => { + // Deferred actions' reducer effects are already in NgRx state while + // their ops get seqs only on a future drain — capturing now would + // bake those effects into a snapshot tagged behind their future seqs + // (same double-apply as #8469). New deferrals cannot start while we + // hold the lock (the sync window that buffers them requires it), so + // this synchronous check at the capture cutoff is race-free. + if (getDeferredActions().length > 0) { + OpLog.warn( + 'OperationLogSnapshotService: Skipping snapshot save — deferred actions are pending durable write', + ); + return; + } + + // Read state synchronously at the quiesce cutoff (no await before it) + // and lastSeq after — see JSDoc above. const currentState = this.stateSnapshotService.getStateSnapshotForOperationLog(); + const lastSeq = await this.opLogStore.getLastSeq(); // GUARD (#7892): never cache an empty/degraded state over a good one. // The snapshot is only a load-time cache — the op-log is the source of diff --git a/src/app/op-log/testing/integration/compaction.integration.spec.ts b/src/app/op-log/testing/integration/compaction.integration.spec.ts index 462d70c979..b7347188e9 100644 --- a/src/app/op-log/testing/integration/compaction.integration.spec.ts +++ b/src/app/op-log/testing/integration/compaction.integration.spec.ts @@ -5,6 +5,7 @@ import { VectorClockService } from '../../sync/vector-clock.service'; import { OpType, VectorClock } from '../../core/operation.types'; import { CURRENT_SCHEMA_VERSION } from '../../persistence/schema-migration.service'; import { TestClient, resetTestUuidCounter } from './helpers/test-client.helper'; +import { clearDeferredActions } from '../../capture/operation-capture.meta-reducer'; import { createTaskOperation, createMinimalTaskPayload, @@ -70,6 +71,10 @@ describe('Compaction Integration', () => { await storeService.init(); await storeService._clearAllDataForTesting(); resetTestUuidCounter(); + // The real compaction service bails when the MODULE-LEVEL deferred buffer + // is non-empty (#8469). Another spec leaking into it would make these + // tests fail order-dependently under jasmine's random order — start clean. + clearDeferredActions(); }); describe('Snapshot creation', () => { diff --git a/src/app/op-log/testing/integration/meta-reducer-ordering.integration.spec.ts b/src/app/op-log/testing/integration/meta-reducer-ordering.integration.spec.ts index 74b2aadcb4..a6cf14aa1f 100644 --- a/src/app/op-log/testing/integration/meta-reducer-ordering.integration.spec.ts +++ b/src/app/op-log/testing/integration/meta-reducer-ordering.integration.spec.ts @@ -14,6 +14,7 @@ import { operationCaptureMetaReducer, setOperationCaptureService, setIsApplyingRemoteOps, + clearDeferredActions, } from '../../capture/operation-capture.meta-reducer'; import { OperationCaptureService } from '../../capture/operation-capture.service'; import { EntityType, OpType } from '../../core/operation.types'; @@ -118,6 +119,11 @@ describe('Meta-reducer ordering integration', () => { captureService.clear(); // Ensure sync state is reset after each test setIsApplyingRemoteOps(false); + // The sync-window tests buffer persistent actions into the MODULE-LEVEL + // deferred buffer; without this, they leak into every later spec in the + // bundle (order-dependent under jasmine's random order — see #8469's + // compaction bail, which reads that buffer). + clearDeferredActions(); }); describe('action capture with meta-reducer chain', () => { diff --git a/src/app/op-log/testing/integration/performance.integration.spec.ts b/src/app/op-log/testing/integration/performance.integration.spec.ts index dcb89a0855..1714e37ca9 100644 --- a/src/app/op-log/testing/integration/performance.integration.spec.ts +++ b/src/app/op-log/testing/integration/performance.integration.spec.ts @@ -5,6 +5,7 @@ import { VectorClockService } from '../../sync/vector-clock.service'; import { OpType } from '../../core/operation.types'; import { CURRENT_SCHEMA_VERSION } from '../../persistence/schema-migration.service'; import { TestClient, resetTestUuidCounter } from './helpers/test-client.helper'; +import { clearDeferredActions } from '../../capture/operation-capture.meta-reducer'; import { createTaskOperation, createMinimalTaskPayload, @@ -66,6 +67,10 @@ describe('Performance Integration', () => { await storeService.init(); await storeService._clearAllDataForTesting(); resetTestUuidCounter(); + // The real compaction service bails when the MODULE-LEVEL deferred buffer + // is non-empty (#8469). Another spec leaking into it would make these + // tests fail order-dependently under jasmine's random order — start clean. + clearDeferredActions(); }); describe('Large operation log handling', () => {