fix(sync): quiesce op capture before snapshot and compaction (#8469) (#9083)

* fix(sync): quiesce op capture before snapshot and compaction (#8469)

NgRx state mutates synchronously at dispatch while an op's seq is only
assigned later in the persist effect under the OPERATION_LOG lock. The
snapshot save and threshold compaction captured state + lastSeq under a
bare lock, so an op dispatched-but-unsequenced could have its reducer
effect baked into the state cache while its seq landed after
lastAppliedOpSeq - the next boot's tail replay then re-applied it,
double-applying non-idempotent reducers (removeTimeSpent,
logFocusSession, addNote, addBoard, addTaskAttachment, plain-append
move/batch branches). Compaction made this a live-use window (every
COMPACTION_THRESHOLD ops), not just a hydration edge.

Both writers now run through flushThenRunExclusive: drain the capture
pipeline before acquiring the non-reentrant lock, re-check the pending
counter inside it. The snapshot body reads state first (synchronously at
the quiesce cutoff) and lastSeq after - exact while the lock blocks seq
assignment; compaction re-checks the counter AND the deferred-actions
buffer (uncounted, kept across windows after a failed drain) in the same
synchronous block as its state read and bails, retriggering on the next
threshold. Emergency compaction keeps the bare lock: it runs inside the
failing write's call stack during quota handling, where that write's
pending entry is still elevated - flushing would self-wait until timeout
and break quota recovery.

Known trade-off: a wedged persist effect now surfaces as a flush timeout
(skipped save / compaction failure counter) instead of a silently
tagged-behind cache.

* test(sync): isolate module-level deferred buffer across specs

The #8469 capture bail reads the module-level deferred-actions buffer,
which is shared across the whole karma bundle. The meta-reducer-ordering
integration spec buffers persistent actions during its simulated sync
window and never cleared them, so under jasmine's random spec order any
later spec using the real compaction/snapshot services could see a dirty
buffer and skip its capture - the CI-only failure in
performance.integration.spec.ts ("Expected null not to be null").

Clear the buffer in the leaking spec's afterEach, and defensively in the
beforeEach of every spec that exercises the real bail, so no other
leaker can fail them order-dependently.

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<OperationLogStoreService>;
   let mockLockService: jasmine.SpyObj<LockService>;
   let mockStateSnapshot: jasmine.SpyObj<StateSnapshotService>;
@@ -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
+      // The flush pre-pass acquires/releases the lock once on its own; all
-      const lockStartIndex = callOrder.indexOf('lock-start');
+      // compaction steps must happen inside the LAST lock window.
-      const lockEndIndex = callOrder.indexOf('lock-end');
+      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
+      // At least one acquisition per compact(); the flush pre-pass (#8469
-      expect(lockRequests.length).toBe(3);
+      // 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)
+      // Both paths acquire the lock (serialized): regular compact at least
-      expect(callOrder.filter((c) => c === 'lock-start').length).toBe(2);
+      // 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(() => (timeJumped ? 26000 : 0));
-      spyOn(Date, 'now').and.callFake(() => {
+      mockOpLogStore.getPendingRemoteOps.and.callFake(async () => {
-        callCount++;
+        timeJumped = true; // 26 seconds elapse "during" the read
-        // First call is startTime, subsequent calls simulate 26s elapsed
+        return [];
-        if (callCount === 1) {
-          return 0;
-        }
-        return 26000; // 26 seconds - exceeds 25s timeout
       });
 
-      // 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(() => {
+      spyOn(Date, 'now').and.callFake(() => (timeJumped ? 26000 : 0));
-        callCount++;
+      mockOpLogStore.getPendingRemoteOps.and.callFake(async () => {
-        if (callCount === 1) {
+        timeJumped = true;
-          return 0;
+        return [];
-        }
-        return 26000;
       });
 
       mockStateSnapshot.getStateSnapshot.and.returnValue(mockState);

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<boolean> {
     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<boolean> {
-    return this.lockService.request(LOCK_NAMES.OPERATION_LOG, async () => {
+    const compactExclusively = async (): Promise<boolean> => {
       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);
   }
 
   /**

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<OperationLogStoreService>;
   let mockVectorClockService: jasmine.SpyObj<VectorClockService>;
   let mockStateSnapshotService: jasmine.SpyObj<StateSnapshotService>;
@@ -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
+      // The flush pre-pass acquires/releases the lock once on its own; the
-      const lockStartIndex = callOrder.indexOf('lock-start');
+      // capture body runs inside the LAST lock window.
-      const lockEndIndex = callOrder.indexOf('lock-end');
+      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
+      // Key invariant (#8469): with the capture pipeline drained and the lock
-      expect(callOrder.indexOf('getLastSeq')).toBeLessThan(
+      // held, no op can gain a seq during the body. State is read first
-        callOrder.indexOf('getStateSnapshot'),
+      // (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: {} },

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 { OperationWriteFlushService } from '../sync/operation-write-flush.service';
-import { LOCK_NAMES } from '../core/operation-log.const';
+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
+   * Runs via flushThenRunExclusive (#8469): the capture pipeline is drained
-   * the lock, an op appended between reading NgRx state and reading lastSeq
+   * BEFORE the OPERATION_LOG lock is taken and the pending counter re-checked
-   * would get a seq <= lastAppliedOpSeq but its effect would be absent from
+   * once inside it, so no counted action can be dispatched-but-unsequenced at
-   * the snapshot. On next hydration the tail replay would start after that
+   * the capture point. NgRx state mutates synchronously at dispatch while the
-   * seq, silently skipping the op forever.
+   * 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
+   * Deferred actions (buffered during a sync window, kept across windows
-   * interleaving degrades to re-replay rather than a missed op. Most op types
+   * after a failed drain) are NOT covered by the pending counter; they are
-   * are idempotent on re-replay (entity-adapter CRUD/move/reorder), but some
+   * handled by an explicit skip inside the body.
-   * 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
+   * Inside the lock, state is read first — synchronously, before any await
-   * task moves). This edge is unlikely (the save runs only during hydration) and
+   * can let a dispatch interleave — and lastSeq after. While the lock is held
-   * strictly better than op-loss. The audited op list and a truly-lossless
+   * no new seq can be assigned, so every op with seq <= lastAppliedOpSeq has
-   * capture (quiesce the op-write queue before reading) are tracked in #8469.
+   * 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<void> {
     try {
-      await this.lockService.request(LOCK_NAMES.OPERATION_LOG, async () => {
+      await this.writeFlushService.flushThenRunExclusive(async () => {
-        // Read lastSeq BEFORE state snapshot — see JSDoc above.
+        // Deferred actions' reducer effects are already in NgRx state while
-        // NOTE: compaction reads in the opposite order (state, then lastSeq);
+        // their ops get seqs only on a future drain — capturing now would
-        // its failure mode if the lock is bypassed is missed-op data loss.
+        // bake those effects into a snapshot tagged behind their future seqs
-        const lastSeq = await this.opLogStore.getLastSeq();
+        // (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

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', () => {

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', () => {

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', () => {