From 59e2a1791c6749febf007c7eed8ae57d719a31ec Mon Sep 17 00:00:00 2001 From: Johannes Millan Date: Sat, 13 Jun 2026 16:14:48 +0200 Subject: [PATCH] fix(sync): prevent lock-timeout from wedging op capture (#8306, #8318) (#8383) A LockAcquisitionTimeoutError during op capture errored the whole persistOperation$ stream: concatMap tore down and silently dropped every buffered action, the positional capture FIFO leaked an entry so flushPendingWrites() could never reach 0 (every sync then failed after 30s), and after NgRx's 10-resubscribe cap the effect died until reload. Fix, bundled with the #8318 cleanup: - Replace the positional FIFO queue with a pending counter. The meta-reducer increments it; the effect decrements it in a `finally` (writeOperationFromEffect), so a thrown write can never leak the flush signal. The decrement runs after the write commits + lock releases, preserving the flush commit-ordering invariant. - The effect catches per write so one failure never tears down the shared stream (the resubscribe-death and silent-drop fixes). - entityChanges is now computed in the write path via the pure extractEntityChanges(); the `[]` field is still emitted (Android reads it; isMultiEntityPayload requires it). - writeOperation keeps its throw for the #7700 deferred retry loop (that path bypasses the wrapper and is not counted). This also structurally removes the #8307 double-dequeue. Adds operation-log-effect-stream-survival.regression.spec.ts (stream survives lock timeout, counter drains on always-fail, survives >10 failures) and updates the capture/flush/integration specs + sync docs. --- .../operation-log-architecture.md | 30 +- docs/sync-and-op-log/operation-rules.md | 2 +- .../operation-capture.meta-reducer.spec.ts | 46 ++-- .../capture/operation-capture.meta-reducer.ts | 7 +- .../capture/operation-capture.service.spec.ts | 186 ++++++------- .../capture/operation-capture.service.ts | 146 +++++----- ...-effect-stream-survival.regression.spec.ts | 259 ++++++++++++++++++ .../capture/operation-log.effects.spec.ts | 27 +- .../op-log/capture/operation-log.effects.ts | 128 ++++++--- ...ration-log-lock-reentry.regression.spec.ts | 5 +- .../sync/operation-write-flush.service.ts | 59 ++-- .../meta-reducer-ordering.integration.spec.ts | 10 +- ...multi-entity-atomicity.integration.spec.ts | 23 +- .../race-conditions.integration.spec.ts | 54 ++-- .../task-done-replay.integration.spec.ts | 4 +- 15 files changed, 631 insertions(+), 355 deletions(-) create mode 100644 src/app/op-log/capture/operation-log-effect-stream-survival.regression.spec.ts diff --git a/docs/sync-and-op-log/operation-log-architecture.md b/docs/sync-and-op-log/operation-log-architecture.md index 9e943492ee..a5e89242f2 100644 --- a/docs/sync-and-op-log/operation-log-architecture.md +++ b/docs/sync-and-op-log/operation-log-architecture.md @@ -2086,19 +2086,21 @@ Meta-reducers intercept actions before they reach feature reducers and can modif | `plannerSharedMetaReducer` | Planner day management | | `taskRepeatCfgSharedMetaReducer` | Repeat config deletion with task cleanup | | `issueProviderSharedMetaReducer` | Issue provider updates | -| `operationCaptureMetaReducer` | Captures before/after state, enqueues entity changes | +| `operationCaptureMetaReducer` | Marks the action as pending capture (increments counter) | ## F.3 Multi-Entity Operation Capture -The `OperationCaptureService` and `operation-capture.meta-reducer` work together using a **simple FIFO queue** to capture actions: +The `OperationCaptureService` and `operation-capture.meta-reducer` work together using a **pending counter** to track captures (no positional queue — see the note below): -1. **After action**: Meta-reducer calls `OperationCaptureService.enqueue()` with the action -2. **Effect processes**: Effect calls `OperationCaptureService.dequeue()` to get entity changes +1. **After action**: Meta-reducer calls `OperationCaptureService.incrementPending()` with the action +2. **Effect processes**: Effect computes `entityChanges` via `OperationCaptureService.extractEntityChanges()`, writes the operation, then decrements the counter in a `finally` 3. **Result**: Single operation with action payload and optional `entityChanges[]` array -The FIFO queue works because NgRx reducers process actions sequentially, and effects use `concatMap` for sequential processing. Order is preserved between enqueue and dequeue. +`flushPendingWrites()` polls `getPendingCount()` to know when every dispatched action has been written. NgRx reducers process actions sequentially and the effect uses `concatMap`, so writes stay ordered. -**Note**: Most actions return empty `entityChanges[]` - the action payload is sufficient for replay. Only TIME_TRACKING and TASK time sync actions have special handling to extract entity changes from the action payload. +**Why a counter, not a positional FIFO queue (#8306 / #8318)**: the old design queued an `EntityChange[]` per action and correlated meta-reducer `push` with effect `shift` purely by position. If a write threw before its `dequeue` ran (e.g. a `LockAcquisitionTimeoutError`), the entry leaked and `flushPendingWrites()` could never reach 0 — every later sync then failed after its 30s timeout. A counter decremented in a `finally` cannot leak. `entityChanges` is now computed in the write path from the action (a pure function), so there is nothing to keep positionally aligned. + +**Note**: Most actions return empty `entityChanges[]` - the action payload is sufficient for replay. Only TIME_TRACKING and TASK time sync actions have special handling to extract entity changes from the action payload. The field is still emitted (even as `[]`) because the Android background provider reads it and the `isMultiEntityPayload` guard requires it. ``` User Action (e.g., Delete Tag) @@ -2112,14 +2114,14 @@ Feature Reducers │ ▼ operation-capture.meta-reducer - ├──► Call OperationCaptureService.enqueue(action) - │ └──► Extracts entity changes from action payload (for special cases) - │ └──► Pushes to FIFO queue + ├──► Call OperationCaptureService.incrementPending(action) + │ └──► Increments the pending counter │ ▼ -OperationLogEffects - ├──► Call OperationCaptureService.dequeue() to get entity changes - └──► Create single Operation with action payload +OperationLogEffects (per-action wrapper: writeOperationFromEffect) + ├──► Call OperationCaptureService.extractEntityChanges(action) + ├──► Create + persist single Operation with action payload + └──► finally: OperationCaptureService.decrementPending() ``` ## F.4 When to Use Meta-Reducers vs Effects @@ -2293,7 +2295,7 @@ When adding new entities or relationships: > - **End-to-End Encryption**: AES-256-GCM payload encryption with Argon2id key derivation via `OperationEncryptionService` > - **Server Security Hardening**: Audit logging, structured error codes, request deduplication, transaction isolation, input validation, rate limiting > - **Unified Archive Handling**: `ArchiveOperationHandler` is now the single source of truth for all archive operations, used by both local effects and remote operation application -> - **Simplified OperationCaptureService**: Refactored to FIFO queue with reference equality optimization for detecting changed feature states +> - **Simplified OperationCaptureService**: Tracks pending captures with a counter (no positional FIFO queue); `entityChanges` is computed in the write path from the action > - **Simplified OperationApplierService**: Refactored to fail-fast approach - throws `SyncStateCorruptedError` on missing hard deps (no retry queues) > - **Tag sanitization**: Remove subtask IDs from tags when parent deleted, filter non-existent taskIds on sync > - **Anchor-based move operations**: All task drag-drop moves now use `afterTaskId` instead of full list replacement (including subtask moves) @@ -2346,7 +2348,7 @@ src/app/op-log/ │ └── operation-sync.util.ts # Sync helper utilities ├── processing/ │ ├── operation-applier.service.ts # Apply ops with fail-fast dependency handling -│ ├── operation-capture.service.ts # FIFO queue for capturing entity changes +│ ├── operation-capture.service.ts # Pending-capture counter + entityChanges extractor │ ├── operation-capture.meta-reducer.ts # Meta-reducer for before/after state capture │ ├── hydration-state.service.ts # Track hydration/remote ops application state │ ├── archive-operation-handler.service.ts # Unified handler for archive side effects diff --git a/docs/sync-and-op-log/operation-rules.md b/docs/sync-and-op-log/operation-rules.md index 0abd382abf..ea837e8dce 100644 --- a/docs/sync-and-op-log/operation-rules.md +++ b/docs/sync-and-op-log/operation-rules.md @@ -186,7 +186,7 @@ export class MyEffects { ### 5.2 Capture Multi-Entity Changes - **Rule:** The `OperationCaptureService` automatically captures all entity changes from a single action. -- **Implementation:** The `operation-capture.meta-reducer` calls `OperationCaptureService.enqueue()` with the action. +- **Implementation:** The `operation-capture.meta-reducer` calls `OperationCaptureService.incrementPending()` with the action; the persist effect computes the changes via `OperationCaptureService.extractEntityChanges()` and decrements the pending counter in a `finally`. - **Result:** Single operation with `entityChanges[]` array containing all affected entities. ## 6. Configuration Constants diff --git a/src/app/op-log/capture/operation-capture.meta-reducer.spec.ts b/src/app/op-log/capture/operation-capture.meta-reducer.spec.ts index 7789abd6e3..0f0adab79d 100644 --- a/src/app/op-log/capture/operation-capture.meta-reducer.spec.ts +++ b/src/app/op-log/capture/operation-capture.meta-reducer.spec.ts @@ -52,9 +52,9 @@ describe('operationCaptureMetaReducer', () => { beforeEach(() => { mockCaptureService = jasmine.createSpyObj('OperationCaptureService', [ - 'enqueue', - 'dequeue', - 'getQueueSize', + 'incrementPending', + 'decrementPending', + 'getPendingCount', 'clear', ]); @@ -75,7 +75,7 @@ describe('operationCaptureMetaReducer', () => { describe('setOperationCaptureService', () => { it('should set the capture service instance', () => { const newCaptureService = jasmine.createSpyObj('OperationCaptureService', [ - 'enqueue', + 'incrementPending', ]); setOperationCaptureService(newCaptureService); @@ -108,14 +108,14 @@ describe('operationCaptureMetaReducer', () => { expect(result).toBe(mockModifiedState); }); - it('should enqueue action for persistent local actions', () => { + it('should capture action for persistent local actions', () => { const wrappedReducer = operationCaptureMetaReducer(mockReducer); const action = createMockAction(); wrappedReducer(mockState, action); - // Should call enqueue with just the action (no state params) - expect(mockCaptureService.enqueue).toHaveBeenCalledWith(action); + // Should increment the pending counter with just the action (no state params) + expect(mockCaptureService.incrementPending).toHaveBeenCalledWith(action); }); it('should NOT process remote actions', () => { @@ -132,7 +132,7 @@ describe('operationCaptureMetaReducer', () => { wrappedReducer(mockState, action); - expect(mockCaptureService.enqueue).not.toHaveBeenCalled(); + expect(mockCaptureService.incrementPending).not.toHaveBeenCalled(); }); it('should NOT process non-persistent actions', () => { @@ -141,7 +141,7 @@ describe('operationCaptureMetaReducer', () => { wrappedReducer(mockState, action); - expect(mockCaptureService.enqueue).not.toHaveBeenCalled(); + expect(mockCaptureService.incrementPending).not.toHaveBeenCalled(); }); it('should process even when state is undefined (initial state)', () => { @@ -151,7 +151,7 @@ describe('operationCaptureMetaReducer', () => { wrappedReducer(undefined, action); - expect(mockCaptureService.enqueue).toHaveBeenCalledWith(action); + expect(mockCaptureService.incrementPending).toHaveBeenCalledWith(action); }); it('should work without service (graceful degradation)', () => { @@ -164,7 +164,7 @@ describe('operationCaptureMetaReducer', () => { }); it('should handle errors in capture service gracefully', () => { - mockCaptureService.enqueue.and.throwError('Test error'); + mockCaptureService.incrementPending.and.throwError('Test error'); const wrappedReducer = operationCaptureMetaReducer(mockReducer); const action = createMockAction(); @@ -175,8 +175,8 @@ describe('operationCaptureMetaReducer', () => { }); }); - describe('enqueue ordering', () => { - it('should call reducer before enqueuing action', () => { + describe('capture ordering', () => { + it('should call reducer before capturing action', () => { const callOrder: string[] = []; mockReducer.and.callFake(() => { @@ -184,15 +184,15 @@ describe('operationCaptureMetaReducer', () => { return mockModifiedState; }); - mockCaptureService.enqueue.and.callFake(() => { - callOrder.push('enqueue'); + mockCaptureService.incrementPending.and.callFake(() => { + callOrder.push('capture'); }); const wrappedReducer = operationCaptureMetaReducer(mockReducer); wrappedReducer(mockState, createMockAction()); - // Reducer should be called first, then enqueue - expect(callOrder).toEqual(['reducer', 'enqueue']); + // Reducer should be called first, then capture + expect(callOrder).toEqual(['reducer', 'capture']); }); }); @@ -208,10 +208,10 @@ describe('operationCaptureMetaReducer', () => { ]; actionTypes.forEach((type) => { - mockCaptureService.enqueue.calls.reset(); + mockCaptureService.incrementPending.calls.reset(); const action = createMockAction({ type }); wrappedReducer(mockState, action); - expect(mockCaptureService.enqueue).toHaveBeenCalled(); + expect(mockCaptureService.incrementPending).toHaveBeenCalled(); }); }); }); @@ -297,7 +297,7 @@ describe('operationCaptureMetaReducer', () => { wrappedReducer(mockState, action); // Should NOT immediately capture - sync is in progress - expect(mockCaptureService.enqueue).not.toHaveBeenCalled(); + expect(mockCaptureService.incrementPending).not.toHaveBeenCalled(); // But action should be buffered for later processing const buffered = getDeferredActions(); @@ -314,7 +314,7 @@ describe('operationCaptureMetaReducer', () => { wrappedReducer(mockState, action1); wrappedReducer(mockState, action2); - expect(mockCaptureService.enqueue).not.toHaveBeenCalled(); + expect(mockCaptureService.incrementPending).not.toHaveBeenCalled(); const buffered = getDeferredActions(); expect(buffered).toEqual([action1, action2]); @@ -328,12 +328,12 @@ describe('operationCaptureMetaReducer', () => { // During sync - action is buffered setIsApplyingRemoteOps(true); wrappedReducer(mockState, syncAction); - expect(mockCaptureService.enqueue).not.toHaveBeenCalled(); + expect(mockCaptureService.incrementPending).not.toHaveBeenCalled(); // After sync - actions are captured immediately setIsApplyingRemoteOps(false); wrappedReducer(mockState, normalAction); - expect(mockCaptureService.enqueue).toHaveBeenCalledWith(normalAction); + expect(mockCaptureService.incrementPending).toHaveBeenCalledWith(normalAction); // Verify the sync action was buffered const buffered = getDeferredActions(); diff --git a/src/app/op-log/capture/operation-capture.meta-reducer.ts b/src/app/op-log/capture/operation-capture.meta-reducer.ts index f2618bc4ca..54cfaf89c0 100644 --- a/src/app/op-log/capture/operation-capture.meta-reducer.ts +++ b/src/app/op-log/capture/operation-capture.meta-reducer.ts @@ -242,14 +242,15 @@ export const operationCaptureMetaReducer = ( try { const persistentAction = action as PersistentAction; - // Enqueue action for effect processing (no state diffing needed) - operationCaptureService.enqueue(persistentAction); + // Record the action as pending; the persist effect decrements the + // counter after writing it. No state diffing needed. + operationCaptureService.incrementPending(persistentAction); // Reset failure counter on success consecutiveCaptureFailures = 0; OpLog.verbose( - 'operationCaptureMetaReducer: Enqueued action for operation capture', + 'operationCaptureMetaReducer: Captured action for operation persistence', { actionType: persistentAction.type, }, diff --git a/src/app/op-log/capture/operation-capture.service.spec.ts b/src/app/op-log/capture/operation-capture.service.spec.ts index b3777f9030..747282dff2 100644 --- a/src/app/op-log/capture/operation-capture.service.spec.ts +++ b/src/app/op-log/capture/operation-capture.service.spec.ts @@ -31,8 +31,63 @@ describe('OperationCaptureService', () => { service.clear(); }); - describe('enqueue and dequeue', () => { - it('should enqueue and dequeue empty entityChanges for regular actions', () => { + describe('pending counter', () => { + it('should start at zero', () => { + expect(service.getPendingCount()).toBe(0); + }); + + it('should increment on capture and decrement on processing', () => { + const action = createPersistentAction('[Task] Update Task', 'TASK', 'task-1'); + + service.incrementPending(action); + expect(service.getPendingCount()).toBe(1); + + service.decrementPending(); + expect(service.getPendingCount()).toBe(0); + }); + + it('should track multiple pending operations', () => { + const action = createPersistentAction('[Task] Update', 'TASK'); + + service.incrementPending(action); + service.incrementPending(action); + service.incrementPending(action); + expect(service.getPendingCount()).toBe(3); + + service.decrementPending(); + expect(service.getPendingCount()).toBe(2); + + service.decrementPending(); + service.decrementPending(); + expect(service.getPendingCount()).toBe(0); + }); + + it('should clamp at zero on underflow (decrement without matching increment)', () => { + // Degenerate window: a decrement arrives with nothing pending. The counter + // must stay at 0 so the flush signal never goes negative. + service.decrementPending(); + expect(service.getPendingCount()).toBe(0); + + const action = createPersistentAction('[Task] Update', 'TASK'); + service.incrementPending(action); + service.decrementPending(); + service.decrementPending(); + expect(service.getPendingCount()).toBe(0); + }); + + it('should reset the counter on clear', () => { + const action = createPersistentAction('[Task] Update', 'TASK'); + service.incrementPending(action); + service.incrementPending(action); + expect(service.getPendingCount()).toBe(2); + + service.clear(); + expect(service.getPendingCount()).toBe(0); + }); + }); + + describe('extractEntityChanges', () => { + it('should return empty entityChanges for regular actions', () => { const action = createPersistentAction( '[Task] Update Task', 'TASK', @@ -40,53 +95,27 @@ describe('OperationCaptureService', () => { OpType.Update, ); - service.enqueue(action); - expect(service.getQueueSize()).toBe(1); - - const changes = service.dequeue(); - expect(changes).toEqual([]); - expect(service.getQueueSize()).toBe(0); + expect(service.extractEntityChanges(action)).toEqual([]); }); - it('should maintain FIFO order', () => { - const action1 = createPersistentAction( - '[Task] Add Task', - 'TASK', - 'task-1', - OpType.Create, - ); - const action2 = createPersistentAction( - '[Task] Update Task', - 'TASK', - 'task-2', - OpType.Update, - ); - const action3 = createPersistentAction( - '[Task] Delete Task', - 'TASK', - 'task-3', - OpType.Delete, - ); + it('should be a pure function (idempotent across repeated calls)', () => { + const action = { + type: '[TimeTracking] Sync Time Tracking', + contextType: 'PROJECT', + contextId: 'project-1', + date: '2024-01-15', + data: { workedMs: 3600000 }, + meta: { + isPersistent: true, + entityType: 'TIME_TRACKING' as EntityType, + entityId: 'time-1', + opType: OpType.Update, + }, + } as PersistentAction; - service.enqueue(action1); - service.enqueue(action2); - service.enqueue(action3); - - expect(service.getQueueSize()).toBe(3); - - service.dequeue(); - expect(service.getQueueSize()).toBe(2); - - service.dequeue(); - expect(service.getQueueSize()).toBe(1); - - service.dequeue(); - expect(service.getQueueSize()).toBe(0); - }); - - it('should return empty array when dequeue from empty queue', () => { - const changes = service.dequeue(); - expect(changes).toEqual([]); + const first = service.extractEntityChanges(action); + const second = service.extractEntityChanges(action); + expect(first).toEqual(second); }); }); @@ -106,8 +135,7 @@ describe('OperationCaptureService', () => { }, } as PersistentAction; - service.enqueue(action); - const changes = service.dequeue(); + const changes = service.extractEntityChanges(action); expect(changes.length).toBe(1); expect(changes[0].entityType).toBe('TIME_TRACKING'); @@ -135,8 +163,7 @@ describe('OperationCaptureService', () => { }, } as PersistentAction; - service.enqueue(action); - const changes = service.dequeue(); + const changes = service.extractEntityChanges(action); expect(changes.length).toBe(1); expect(changes[0].entityType).toBe('TIME_TRACKING'); @@ -156,8 +183,7 @@ describe('OperationCaptureService', () => { }, } as PersistentAction; - service.enqueue(action); - const changes = service.dequeue(); + const changes = service.extractEntityChanges(action); expect(changes).toEqual([]); }); @@ -179,8 +205,7 @@ describe('OperationCaptureService', () => { }, } as PersistentAction; - service.enqueue(action); - const changes = service.dequeue(); + const changes = service.extractEntityChanges(action); expect(changes.length).toBe(1); expect(changes[0].entityType).toBe('TASK'); @@ -209,61 +234,10 @@ describe('OperationCaptureService', () => { }, } as PersistentAction; - service.enqueue(action); - const changes = service.dequeue(); + const changes = service.extractEntityChanges(action); // Should return empty - not captured as syncTimeSpent expect(changes).toEqual([]); }); }); - - describe('queue management', () => { - it('should track queue size correctly', () => { - expect(service.getQueueSize()).toBe(0); - - const action = createPersistentAction('[Task] Update', 'TASK'); - service.enqueue(action); - expect(service.getQueueSize()).toBe(1); - - service.enqueue(action); - expect(service.getQueueSize()).toBe(2); - - service.dequeue(); - expect(service.getQueueSize()).toBe(1); - }); - - it('should clear all queued operations', () => { - const action = createPersistentAction('[Task] Update', 'TASK'); - service.enqueue(action); - service.enqueue(action); - expect(service.getQueueSize()).toBe(2); - - service.clear(); - expect(service.getQueueSize()).toBe(0); - }); - - it('should peek at pending operations without removing them', () => { - // TIME_TRACKING action to generate actual entity changes - const action = { - type: '[TimeTracking] Sync', - contextType: 'PROJECT', - contextId: 'p1', - date: '2024-01-15', - data: {}, - meta: { - isPersistent: true, - entityType: 'TIME_TRACKING' as EntityType, - entityId: 'time-1', - opType: OpType.Update, - }, - } as PersistentAction; - - service.enqueue(action); - expect(service.getQueueSize()).toBe(1); - - const pending = service.peekPendingOperations(); - expect(pending.length).toBe(1); - expect(service.getQueueSize()).toBe(1); // Still in queue - }); - }); }); diff --git a/src/app/op-log/capture/operation-capture.service.ts b/src/app/op-log/capture/operation-capture.service.ts index 5898ff0e2c..c4ff05963f 100644 --- a/src/app/op-log/capture/operation-capture.service.ts +++ b/src/app/op-log/capture/operation-capture.service.ts @@ -4,135 +4,139 @@ import { PersistentAction } from '../core/persistent-action.interface'; import { OpLog } from '../../core/log'; /** - * Captures action payloads and queues them for persistence using a simple FIFO queue. + * Tracks how many local actions have been captured but not yet written to the + * operation log, and computes the `entityChanges` payload for an action. * - * PERFORMANCE OPTIMIZATION: This service no longer performs expensive state diffing. - * Instead, it relies on action payloads and meta.entityId/entityIds for sync. + * ## Pending counter (replaces the former positional FIFO queue) * - * The entityChanges computed here are for backward compatibility with the operation - * log format. The actual replay uses action payloads (extractActionPayload), and - * conflict detection uses meta.entityId from the action. + * The meta-reducer increments the counter synchronously when it captures a + * local action; the persist effect decrements it in a `finally` after each + * write attempt completes (success OR failure). `flushPendingWrites()` polls + * `getPendingCount()` to know when every dispatched action has been processed. * - * Flow: - * 1. Meta-reducer calls `enqueue()` with the action - * 2. Service queues entity changes (empty for most actions, extracted for TIME_TRACKING) - * 3. Effect calls `dequeue()` to retrieve changes for persistence (FIFO order) + * The previous implementation kept a FIFO of `EntityChange[]` correlated with + * actions purely by position (meta-reducer `push`, effect `shift`). That + * positional contract leaked an entry whenever a write threw before its + * dequeue ran (e.g. a lock-acquisition timeout), permanently wedging the flush + * — issue #8306. A plain counter decremented in a `finally` cannot leak. * - * The FIFO queue works because: - * - NgRx reducers process actions sequentially - * - Effect uses concatMap for sequential processing - * - Order is preserved between enqueue and dequeue + * ## entityChanges extraction + * + * `extractEntityChanges()` is a pure function of the action — there is no state + * diffing. It returns `[]` for every action except `TIME_TRACKING` and the + * `[TimeTracking] Sync time spent` TASK action, whose reducers don't follow the + * standard entity-adapter pattern. The field is kept on the wire (even as `[]`) + * because the Android background provider reads `payload.entityChanges` for + * reminder scheduling and the `isMultiEntityPayload` guard requires it; replay + * and conflict detection use `actionPayload` / `meta.entityId` only. */ @Injectable({ providedIn: 'root', }) export class OperationCaptureService { /** - * Warning threshold for queue size. - * If effects fail to dequeue, we log a warning. + * Warning threshold for the pending counter. + * If the effect falls far behind the meta-reducer, log a warning. */ - private readonly QUEUE_WARNING_THRESHOLD = 100; + private readonly PENDING_WARNING_THRESHOLD = 100; /** - * FIFO queue of pending entity changes. + * Count of actions captured (incremented by the meta-reducer) but not yet + * processed by the persist effect (decremented after each write attempt). */ - private queue: EntityChange[][] = []; + private pendingCount = 0; /** - * Tracks if we've already warned about queue overflow to avoid log spam. + * Tracks if we've already warned about the pending counter growing large, + * to avoid log spam. */ - private hasWarnedAboutQueueSize = false; + private hasWarnedAboutPending = false; /** - * Enqueues entity changes for an action. + * Records that a local action has been captured and is awaiting persistence. * Called synchronously by the operation-capture meta-reducer. - * - * For most actions, this queues empty entityChanges (the action payload is sufficient). - * For TIME_TRACKING and TASK time sync, it extracts changes from the action payload. */ - enqueue(action: PersistentAction): void { - const entityChanges = this._extractEntityChanges(action); + incrementPending(action: PersistentAction): void { + this.pendingCount++; - // Warn if queue is growing large (indicates potential processing issue) + // Warn if the counter is growing large (indicates the effect is not keeping up) if ( - this.queue.length >= this.QUEUE_WARNING_THRESHOLD && - !this.hasWarnedAboutQueueSize + this.pendingCount >= this.PENDING_WARNING_THRESHOLD && + !this.hasWarnedAboutPending ) { OpLog.warn( - `OperationCaptureService: Queue size (${this.queue.length}) exceeds warning threshold ` + - `(${this.QUEUE_WARNING_THRESHOLD}). Effects may not be processing operations.`, + `OperationCaptureService: Pending count (${this.pendingCount}) exceeds warning threshold ` + + `(${this.PENDING_WARNING_THRESHOLD}). Effect may not be processing operations.`, ); - this.hasWarnedAboutQueueSize = true; + this.hasWarnedAboutPending = true; } - this.queue.push(entityChanges); - - OpLog.verbose('OperationCaptureService: Enqueued operation', { + OpLog.verbose('OperationCaptureService: Captured action', { actionType: action.type, - changeCount: entityChanges.length, - queueSize: this.queue.length, + pendingCount: this.pendingCount, }); } /** - * Dequeues the next batch of entity changes (FIFO). - * Called by the effect to retrieve pre-computed changes for persistence. + * Records that a captured action's write attempt has completed (success or + * failure). Called from the persist effect's `finally`, so a thrown write — + * e.g. a lock-acquisition timeout — can never leak a pending entry (#8306). */ - dequeue(): EntityChange[] { - const entityChanges = this.queue.shift(); - - if (entityChanges === undefined) { - OpLog.warn('OperationCaptureService: No queued operation found'); - return []; + decrementPending(): void { + if (this.pendingCount <= 0) { + // Underflow guard: a decrement with no matching increment. This is only + // reachable in the degenerate window before the meta-reducer service is + // wired (operations are dropped with a warning there). Clamp at 0 so the + // flush signal stays correct. + OpLog.warn( + 'OperationCaptureService: decrementPending called with no pending operations', + ); + this.pendingCount = 0; + return; } - // Reset warning flag when queue drains below threshold so we can warn again if it fills up + this.pendingCount--; + + // Reset warning flag once the backlog drains so we can warn again if it refills if ( - this.queue.length < this.QUEUE_WARNING_THRESHOLD && - this.hasWarnedAboutQueueSize + this.pendingCount < this.PENDING_WARNING_THRESHOLD && + this.hasWarnedAboutPending ) { - this.hasWarnedAboutQueueSize = false; + this.hasWarnedAboutPending = false; } - OpLog.verbose('OperationCaptureService: Dequeued operation', { - changeCount: entityChanges.length, - queueSize: this.queue.length, + OpLog.verbose('OperationCaptureService: Processed action', { + pendingCount: this.pendingCount, }); - - return entityChanges; } /** - * Gets the current queue size (for monitoring). + * Returns the number of captured actions still awaiting persistence. + * Used by `flushPendingWrites()` to know when all writes have completed. */ - getQueueSize(): number { - return this.queue.length; + getPendingCount(): number { + return this.pendingCount; } /** - * Peeks at pending operations without removing them (for diagnostics). - * Returns a flat list of all entity changes in the queue. - */ - peekPendingOperations(): EntityChange[] { - return this.queue.flat(); - } - - /** - * Clears all queued operations (for testing). + * Resets the pending counter (for testing and error recovery). */ clear(): void { - this.queue = []; - this.hasWarnedAboutQueueSize = false; + this.pendingCount = 0; + this.hasWarnedAboutPending = false; } /** - * Extracts entity changes from action payload. + * Extracts entity changes from an action payload. * * For most actions, returns empty array (action payload is sufficient for sync). * TIME_TRACKING and TASK time sync need special handling because their reducers * don't follow the standard entity adapter pattern. + * + * Pure function of the action — safe to call from the write path and idempotent + * across retries. */ - private _extractEntityChanges(action: PersistentAction): EntityChange[] { + extractEntityChanges(action: PersistentAction): EntityChange[] { // TIME_TRACKING: Extract from action payload if (action.meta.entityType === 'TIME_TRACKING') { return this._captureTimeTrackingFromAction(action); diff --git a/src/app/op-log/capture/operation-log-effect-stream-survival.regression.spec.ts b/src/app/op-log/capture/operation-log-effect-stream-survival.regression.spec.ts new file mode 100644 index 0000000000..99ab5c2d11 --- /dev/null +++ b/src/app/op-log/capture/operation-log-effect-stream-survival.regression.spec.ts @@ -0,0 +1,259 @@ +/** + * Regression test for issue #8306. + * + * Pre-fix bug: a single `LockAcquisitionTimeoutError` during op capture errored + * the whole `persistOperation$` effect stream. Because the inner `writeOperation` + * rethrew straight into `concatMap`: + * 1. concatMap tore down — every action buffered behind the failed one was + * silently dropped (no op written, no snackbar for them); + * 2. the positional capture queue leaked the failed action's entry, so + * `flushPendingWrites()` could never reach 0 and every subsequent sync + * failed after its 30s timeout (permanent wedge); + * 3. after NgRx's default 10 resubscribes the effect died silently until the + * app was reloaded. + * + * Fix (this PR, bundled with #8318): + * - the effect wraps each write in `writeOperationFromEffect`, which catches so + * one failed write can never tear down the shared stream; + * - the FIFO queue is replaced by a pending counter decremented in a `finally`, + * so a thrown write can never leak the flush signal. + * + * These tests drive the REAL `persistOperation$` effect and the REAL + * `OperationCaptureService` counter; only the LockService is mocked so a write + * can be forced to time out deterministically. + */ +import { TestBed } from '@angular/core/testing'; +import { provideMockActions } from '@ngrx/effects/testing'; +import { Action, Store } from '@ngrx/store'; +import { ReplaySubject } from 'rxjs'; +import { OperationLogEffects } from './operation-log.effects'; +import { OperationCaptureService } from './operation-capture.service'; +import { OperationLogStoreService } from '../persistence/operation-log-store.service'; +import { LockService } from '../sync/lock.service'; +import { VectorClockService } from '../sync/vector-clock.service'; +import { OperationLogCompactionService } from '../persistence/operation-log-compaction.service'; +import { SnackService } from '../../core/snack/snack.service'; +import { ImmediateUploadService } from '../sync/immediate-upload.service'; +import { ActionType, OpType, Operation } from '../core/operation.types'; +import { PersistentAction } from '../core/persistent-action.interface'; +import { ClientIdService } from '../../core/util/client-id.service'; +import { SuperSyncStatusService } from '../sync/super-sync-status.service'; +import { LockAcquisitionTimeoutError } from '../core/errors/sync-errors'; +import { LOCK_NAMES } from '../core/operation-log.const'; +import { T } from '../../t.const'; + +describe('regression #8306: persistOperation$ stream survives write failures', () => { + let effects: OperationLogEffects; + let captureService: OperationCaptureService; + let lockRequest: jasmine.Spy; + let opLogStoreSpy: jasmine.SpyObj; + let snackSpy: jasmine.SpyObj; + let actions$: ReplaySubject; + + const createAction = (entityId: string): PersistentAction => ({ + type: ActionType.TASK_SHARED_UPDATE, + meta: { + isPersistent: true, + isRemote: false, + opType: OpType.Update, + entityType: 'TASK', + entityId, + }, + }); + + /** Simulate the meta-reducer (increment) + dispatch (effect path). */ + const dispatch = (action: PersistentAction): void => { + captureService.incrementPending(action); + actions$.next(action); + }; + + const waitFor = async (predicate: () => boolean, timeoutMs = 2000): Promise => { + const start = Date.now(); + while (!predicate()) { + if (Date.now() - start > timeoutMs) { + throw new Error('waitFor timed out'); + } + await new Promise((r) => setTimeout(r, 5)); + } + }; + + beforeEach(() => { + actions$ = new ReplaySubject(1); + + opLogStoreSpy = jasmine.createSpyObj('OperationLogStoreService', [ + 'appendWithVectorClockUpdate', + 'getCompactionCounter', + 'clearVectorClockCache', + ]); + opLogStoreSpy.appendWithVectorClockUpdate.and.resolveTo(1); + opLogStoreSpy.getCompactionCounter.and.resolveTo(0); + + const vectorClockSpy = jasmine.createSpyObj('VectorClockService', [ + 'getCurrentVectorClock', + ]); + vectorClockSpy.getCurrentVectorClock.and.resolveTo({ testClient: 5 }); + + const compactionSpy = jasmine.createSpyObj('OperationLogCompactionService', [ + 'compact', + 'emergencyCompact', + ]); + compactionSpy.compact.and.resolveTo(); + compactionSpy.emergencyCompact.and.resolveTo(true); + + snackSpy = jasmine.createSpyObj('SnackService', ['open']); + + const storeSpy = jasmine.createSpyObj('Store', ['dispatch', 'select']); + + const immediateUploadSpy = jasmine.createSpyObj('ImmediateUploadService', [ + 'trigger', + ]); + const clientIdSpy = jasmine.createSpyObj('ClientIdService', [ + 'getOrGenerateClientId', + ]); + clientIdSpy.getOrGenerateClientId.and.resolveTo('testClient'); + + const lockServiceSpy = jasmine.createSpyObj('LockService', ['request']); + lockRequest = lockServiceSpy.request; + // Default: lock granted — run the callback. + lockRequest.and.callFake(async (_name: string, cb: () => Promise) => cb()); + + const superSyncStatusSpy = jasmine.createSpyObj('SuperSyncStatusService', [ + 'updatePendingOpsStatus', + ]); + + TestBed.configureTestingModule({ + providers: [ + OperationLogEffects, + provideMockActions(() => actions$), + { provide: OperationLogStoreService, useValue: opLogStoreSpy }, + { provide: LockService, useValue: lockServiceSpy }, + { provide: VectorClockService, useValue: vectorClockSpy }, + { provide: OperationLogCompactionService, useValue: compactionSpy }, + { provide: SnackService, useValue: snackSpy }, + { provide: Store, useValue: storeSpy }, + { provide: ImmediateUploadService, useValue: immediateUploadSpy }, + { provide: ClientIdService, useValue: clientIdSpy }, + { provide: SuperSyncStatusService, useValue: superSyncStatusSpy }, + // REAL capture service — we assert its pending counter directly. + ], + }); + + effects = TestBed.inject(OperationLogEffects); + captureService = TestBed.inject(OperationCaptureService); + captureService.clear(); + }); + + afterEach(() => captureService.clear()); + + /** + * Defect #1 + #3: one failed write must not tear down the stream — later + * actions still persist, and the effect never resubscribes/dies. + */ + it('keeps persisting later actions after a lock-timeout failure', async () => { + // First OPERATION_LOG request times out; all later ones succeed. + let opLogRequests = 0; + lockRequest.and.callFake(async (name: string, cb: () => Promise) => { + if (name === LOCK_NAMES.OPERATION_LOG) { + opLogRequests++; + if (opLogRequests === 1) { + throw new LockAcquisitionTimeoutError(name, 1000); + } + } + return cb(); + }); + + let streamErrored = false; + const sub = effects.persistOperation$.subscribe({ + error: () => (streamErrored = true), + }); + + dispatch(createAction('task-fail')); + dispatch(createAction('task-ok')); + + // Both processed once the counter drains back to 0. + await waitFor(() => captureService.getPendingCount() === 0); + + // The stream survived the first failure... + expect(streamErrored).toBe(false); + // ...the failed action was NOT written, the later one WAS. + expect(opLogStoreSpy.appendWithVectorClockUpdate).toHaveBeenCalledTimes(1); + const writtenOp = opLogStoreSpy.appendWithVectorClockUpdate.calls.mostRecent() + .args[0] as Operation; + expect(writtenOp.entityId).toBe('task-ok'); + // The user was told the failed write needs a reload. + expect(snackSpy.open).toHaveBeenCalledWith( + jasmine.objectContaining({ type: 'ERROR', msg: T.F.SYNC.S.PERSIST_FAILED }), + ); + + sub.unsubscribe(); + }); + + /** + * Defect #2: a thrown write must not leak the pending counter, otherwise + * flushPendingWrites() wedges forever. The `finally` decrement guarantees the + * counter returns to 0 even when every write times out. + */ + it('drains the pending counter even when the write always times out', async () => { + lockRequest.and.callFake(async (name: string, _cb: () => Promise) => { + if (name === LOCK_NAMES.OPERATION_LOG) { + throw new LockAcquisitionTimeoutError(name, 1000); + } + return _cb(); + }); + + const sub = effects.persistOperation$.subscribe(); + + dispatch(createAction('task-1')); + dispatch(createAction('task-2')); + dispatch(createAction('task-3')); + + // Despite every write throwing, the counter must reach 0 (no leak). + await waitFor(() => captureService.getPendingCount() === 0); + + expect(captureService.getPendingCount()).toBe(0); + expect(opLogStoreSpy.appendWithVectorClockUpdate).not.toHaveBeenCalled(); + + sub.unsubscribe(); + }); + + /** + * Defect #3 explicitly: more than NgRx's default 10 consecutive write errors + * must not kill the effect. Pre-fix, the 11th error exceeded the resubscribe + * limit and persistOperation$ died silently; a write dispatched afterwards + * was never persisted. With the per-write catch the stream never errors, so a + * write after 11 failures still lands. + */ + it('survives more than 10 consecutive failures (effect does not die)', async () => { + const FAILURES = 11; + let opLogRequests = 0; + lockRequest.and.callFake(async (name: string, cb: () => Promise) => { + if (name === LOCK_NAMES.OPERATION_LOG) { + opLogRequests++; + if (opLogRequests <= FAILURES) { + throw new LockAcquisitionTimeoutError(name, 1000); + } + } + return cb(); + }); + + let streamErrored = false; + const sub = effects.persistOperation$.subscribe({ + error: () => (streamErrored = true), + }); + + for (let i = 0; i < FAILURES; i++) { + dispatch(createAction(`fail-${i}`)); + } + dispatch(createAction('survivor')); + + await waitFor(() => captureService.getPendingCount() === 0); + + expect(streamErrored).toBe(false); + expect(opLogStoreSpy.appendWithVectorClockUpdate).toHaveBeenCalledTimes(1); + const writtenOp = opLogStoreSpy.appendWithVectorClockUpdate.calls.mostRecent() + .args[0] as Operation; + expect(writtenOp.entityId).toBe('survivor'); + + sub.unsubscribe(); + }); +}); diff --git a/src/app/op-log/capture/operation-log.effects.spec.ts b/src/app/op-log/capture/operation-log.effects.spec.ts index 1fddae0857..59a88e40c9 100644 --- a/src/app/op-log/capture/operation-log.effects.spec.ts +++ b/src/app/op-log/capture/operation-log.effects.spec.ts @@ -76,7 +76,8 @@ describe('OperationLogEffects', () => { 'getOrGenerateClientId', ]); mockOperationCaptureService = jasmine.createSpyObj('OperationCaptureService', [ - 'dequeue', + 'extractEntityChanges', + 'decrementPending', ]); // Default mock implementations @@ -95,7 +96,7 @@ describe('OperationLogEffects', () => { mockClientIdService.getOrGenerateClientId.and.returnValue( Promise.resolve('testClient'), ); - mockOperationCaptureService.dequeue.and.returnValue([]); + mockOperationCaptureService.extractEntityChanges.and.returnValue([]); TestBed.configureTestingModule({ providers: [ @@ -340,7 +341,7 @@ describe('OperationLogEffects', () => { effects.persistOperation$.subscribe({ complete: () => { - expect(mockOperationCaptureService.dequeue).toHaveBeenCalled(); + expect(mockOperationCaptureService.extractEntityChanges).toHaveBeenCalled(); expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledWith( jasmine.objectContaining({ actionType: ActionType.GLOBAL_CONFIG_UPDATE_SECTION, @@ -505,11 +506,12 @@ describe('OperationLogEffects', () => { expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledTimes(2); })); - it('should NOT dequeue a second time on the quota-exceeded retry (#8307)', fakeAsync(() => { - // Regression: the first attempt consumes this action's queue entry via - // dequeue() BEFORE appendWithVectorClockUpdate throws the quota error. The - // retry after emergency compaction must skip dequeue, otherwise it steals - // the NEXT pending action's entry. + it('re-extracts the SAME action on the quota-exceeded retry, never a second op (#8307)', fakeAsync(() => { + // Regression (#8307, now structural): the positional dequeue is gone. + // entityChanges is recomputed by the pure, idempotent extractEntityChanges() + // on every write, so the quota retry re-extracts THIS action's changes and + // can never steal the next pending action's entry the way the old + // double-dequeue did. const quotaError = new DOMException('Quota exceeded', 'QuotaExceededError'); let appendCount = 0; mockOpLogStore.appendWithVectorClockUpdate.and.callFake(() => { @@ -519,7 +521,6 @@ describe('OperationLogEffects', () => { } return Promise.resolve(1); }); - mockOperationCaptureService.dequeue.and.returnValue([]); const action = createPersistentAction(ActionType.TASK_SHARED_UPDATE); actions$ = of(action); @@ -527,9 +528,13 @@ describe('OperationLogEffects', () => { effects.persistOperation$.subscribe(); tick(100); - // Append ran twice (initial + retry) but dequeue ran exactly once. + // Append ran twice (initial + retry). Extraction ran once per write and + // always against the same action — no positional queue to mis-consume. expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledTimes(2); - expect(mockOperationCaptureService.dequeue).toHaveBeenCalledTimes(1); + expect(mockOperationCaptureService.extractEntityChanges).toHaveBeenCalledTimes(2); + expect(mockOperationCaptureService.extractEntityChanges).toHaveBeenCalledWith( + action, + ); })); it('should use updated clientId after backup import generates new one', (done) => { diff --git a/src/app/op-log/capture/operation-log.effects.ts b/src/app/op-log/capture/operation-log.effects.ts index 955fe5a192..e3e8fe06dd 100644 --- a/src/app/op-log/capture/operation-log.effects.ts +++ b/src/app/op-log/capture/operation-log.effects.ts @@ -87,11 +87,14 @@ export class OperationLogEffects implements DeferredLocalActionsPort { * 3. Deferred actions (buffered during sync, processed later by processDeferredActions) * * Note: We do NOT filter by `isApplyingRemoteOps()` here because of a race - * condition: the meta-reducer may enqueue an action before sync starts, but - * the effect processes it after sync starts. Filtering by isApplyingRemoteOps - * would skip the action while its queue entry remains, causing flushPendingWrites - * to time out. Instead, we use isDeferredAction() which precisely identifies - * actions that were buffered (not enqueued) by the meta-reducer. + * condition: the meta-reducer may capture (increment the pending counter for) + * an action before sync starts, but the effect processes it after sync starts. + * Filtering by isApplyingRemoteOps would skip the action while its pending + * count stays elevated, causing flushPendingWrites to time out. Instead, we use + * isDeferredAction() — a stable, per-action property decided at meta-reducer + * time — which precisely identifies actions that were buffered (not counted) by + * the meta-reducer. This keeps the increment set (meta-reducer) and the + * decrement set (this effect) identical, so the counter always balances. */ persistOperation$ = createEffect( () => @@ -102,15 +105,56 @@ export class OperationLogEffects implements DeferredLocalActionsPort { !action.meta.isRemote && !isDeferredAction(action), ), - // Use concatMap for sequential processing to maintain FIFO queue order - concatMap((action) => this.writeOperation(action)), + // concatMap for sequential, ordered processing (one write at a time). + concatMap((action) => this.writeOperationFromEffect(action)), ), { dispatch: false }, ); + /** + * Effect-path wrapper around {@link writeOperation}. + * + * #8306: a thrown write (e.g. a `LockAcquisitionTimeoutError`) must NOT error + * the `persistOperation$` stream. If it did, `concatMap` would tear down and + * silently drop every action buffered behind it, no snackbar would fire for + * them, and after NgRx's default 10 resubscribes the effect would die until + * reload. So we catch here — `writeOperation` already surfaced the failure to + * the user via a sticky snackbar — and the stream lives on. + * + * The `finally` decrements the pending counter exactly once per dispatched + * action, regardless of success, failure, or internal quota-retry recursion. + * This is what frees `flushPendingWrites()`: the counter cannot leak even when + * the write throws (the structural fix for the #8306 flush wedge). + * + * NOTE: the deferred-action path (`processDeferredActions` → `writeOperation`) + * deliberately does NOT go through here — it keeps `writeOperation`'s throw so + * its own retry loop can react (#7700), and its actions were never counted. + */ + private async writeOperationFromEffect(action: PersistentAction): Promise { + try { + await this.writeOperation(action); + } catch (e) { + // Already surfaced to the user inside writeOperation; swallow so the + // shared effect stream is never torn down by a single failed write. + OpLog.err('OperationLogEffects: persist failed (handled; stream preserved)', e); + } finally { + this.operationCaptureService.decrementPending(); + } + } + + /** + * Persists a single action as an operation. + * + * @param isDeferredWrite when true, the action was buffered during the sync + * window and is being flushed by `processDeferredActions`. Deferred writes + * emit `entityChanges: []` (matching the pre-counter behaviour — deferred + * actions were never run through the extractor) and are NOT tracked by the + * pending counter (they were never incremented). Throws on lock timeout so + * the deferred retry loop can react (#7700). + */ private async writeOperation( action: PersistentAction, - skipDequeue = false, + isDeferredWrite = false, options: WriteOperationOptions = {}, ): Promise { const operationTimestamp = Date.now(); @@ -130,10 +174,8 @@ export class OperationLogEffects implements DeferredLocalActionsPort { (id: unknown) => id && typeof id === 'string' && (id as string).trim().length > 0, ); if (!isBulkAllOperation && !hasValidEntityId && !hasValidEntityIds) { - // IMPORTANT: Dequeue first to prevent queue from getting stuck - if (!skipDequeue) { - this.operationCaptureService.dequeue(); - } + // No queue bookkeeping needed here: the effect wrapper's `finally` + // decrements the pending counter for this action even on early return. devError( `[OperationLogEffects] Action ${action.type} has invalid entityId/entityIds (${action.meta.entityId}) - skipping persistence`, ); @@ -175,12 +217,15 @@ export class OperationLogEffects implements DeferredLocalActionsPort { // cache, so Tab B's cache could be stale if Tab A wrote while Tab B was waiting. this.opLogStore.clearVectorClockCache(); - // Get entity changes from the FIFO queue (for TIME_TRACKING and TASK time sync) - // For most actions, this returns empty array since action payloads are sufficient. - // IMPORTANT: This MUST happen inside the lock to prevent race with flushPendingWrites. - // Deferred actions skip dequeue because the meta-reducer buffers them without - // enqueueing — there's no matching queue entry to dequeue. - const entityChanges = skipDequeue ? [] : this.operationCaptureService.dequeue(); + // Compute entity changes from the action (for TIME_TRACKING and TASK time + // sync; empty array for everything else, where the action payload suffices). + // extractEntityChanges is a pure function of the action, so it is safe to + // call here and idempotent across the quota-retry path. Deferred writes + // emit [] to preserve the pre-counter behaviour (they were buffered without + // being run through the extractor). + const entityChanges = isDeferredWrite + ? [] + : this.operationCaptureService.extractEntityChanges(action); const actionPayload = this.addReplayDateFieldsToActionPayload( action, @@ -294,14 +339,12 @@ export class OperationLogEffects implements DeferredLocalActionsPort { if (options.callerHoldsOperationLogLock) { await writeInsideOperationLogLock(); } else { - // CRITICAL: Acquire lock BEFORE dequeuing to prevent race condition with flushPendingWrites(). - // Previously, dequeue happened before lock acquisition, which caused a race: - // 1. dequeue() runs, queue becomes 0 - // 2. flushPendingWrites() polls, sees queue = 0, tries to acquire lock - // 3. If flush wins the lock, it returns before this effect writes to IndexedDB - // 4. Sync uploads with "No pending operations" even though operations were dispatched - // - // By acquiring the lock first, flushPendingWrites() must wait for us to finish writing. + // The pending counter is only decremented by the effect wrapper's + // `finally`, which runs AFTER this request resolves (i.e. after the + // IndexedDB write committed and the lock was released). So the count + // cannot reach 0 before the write is durable, and flushPendingWrites() + // (phase 1 polls the count, phase 2 re-acquires this lock) can never + // report "all flushed" while a write is still pending or in flight. await this.lockService.request( LOCK_NAMES.OPERATION_LOG, writeInsideOperationLogLock, @@ -313,11 +356,14 @@ export class OperationLogEffects implements DeferredLocalActionsPort { if (e instanceof LockAcquisitionTimeoutError) { // #7700: do NOT silently swallow lock timeouts. Pre-fix, a reentrant // sp_op_log timeout was caught here, the user got a snackbar, and the - // deferred action vanished from the op log. Re-throw after notifying so - // processDeferredActions's retry loop retries and persistOperation$'s - // concatMap surfaces the failure upstream. Defense in depth: even if a - // future caller forgets to pass callerHoldsOperationLogLock, the bug is + // deferred action vanished from the op log. Notify, then re-throw so + // processDeferredActions's retry loop retries. Defense in depth: even + // if a future caller forgets callerHoldsOperationLogLock, the bug is // loud instead of silent. + // + // #8306: the effect path catches this throw in writeOperationFromEffect + // — the shared persistOperation$ stream must NOT be torn down by one + // failed write — while still showing the sticky snackbar below. this.notifyUserAndTriggerRollback(); throw e; } @@ -329,7 +375,7 @@ export class OperationLogEffects implements DeferredLocalActionsPort { ); this.notifyUserAndTriggerRollback(); } else { - await this.handleQuotaExceeded(action, options); + await this.handleQuotaExceeded(action, isDeferredWrite, options); } } else { this.notifyUserAndTriggerRollback(); @@ -455,6 +501,7 @@ export class OperationLogEffects implements DeferredLocalActionsPort { */ private async handleQuotaExceeded( action: PersistentAction, + isDeferredWrite = false, options: WriteOperationOptions = {}, ): Promise { OpLog.err( @@ -487,16 +534,11 @@ export class OperationLogEffects implements DeferredLocalActionsPort { // Set circuit breaker before retry to prevent recursive handling this.isHandlingQuotaExceeded = true; // Retry the failed operation after compaction freed space. - // #8307: force skipDequeue=true. The first attempt already consumed - // this action's queue entry (dequeue runs at the top of the lock, - // BEFORE the appendWithVectorClockUpdate that threw the quota error), - // so re-passing the original skipDequeue=false would dequeue a SECOND - // time and steal the NEXT pending action's entityChanges. This mirrors - // the deferred path (processDeferredActions), which also hardcodes - // true. Trade-off: the first attempt's entityChanges are not - // re-captured, so the retried op carries empty entityChanges — - // acceptable for this rare quota-recovery edge case. - await this.writeOperation(action, true, options); + // #8307 (structural): there is no longer a positional dequeue to + // double-consume — entityChanges is recomputed by the pure, idempotent + // extractEntityChanges() inside writeOperation, so the retry simply + // re-extracts the same changes. Pass isDeferredWrite through unchanged. + await this.writeOperation(action, isDeferredWrite, options); this.snackService.open({ type: 'SUCCESS', msg: T.F.SYNC.S.STORAGE_RECOVERED_AFTER_COMPACTION, @@ -577,7 +619,9 @@ export class OperationLogEffects implements DeferredLocalActionsPort { for (let attempt = 0; attempt < MAX_RETRIES; attempt++) { try { - // skipDequeue=true: deferred actions were buffered, not enqueued + // isDeferredWrite=true: deferred actions were buffered (never counted), + // emit entityChanges: [], and keep writeOperation's throw so this retry + // loop can react. await this.writeOperation(action, true, options); success = true; break; diff --git a/src/app/op-log/sync/operation-log-lock-reentry.regression.spec.ts b/src/app/op-log/sync/operation-log-lock-reentry.regression.spec.ts index 0b33bd3a57..1fe578bbba 100644 --- a/src/app/op-log/sync/operation-log-lock-reentry.regression.spec.ts +++ b/src/app/op-log/sync/operation-log-lock-reentry.regression.spec.ts @@ -109,9 +109,10 @@ describe('regression #7700: operation-log lock reentry', () => { clientIdSpy.getOrGenerateClientId.and.resolveTo('testClient'); const operationCaptureSpy = jasmine.createSpyObj('OperationCaptureService', [ - 'dequeue', + 'extractEntityChanges', + 'decrementPending', ]); - operationCaptureSpy.dequeue.and.returnValue([]); + operationCaptureSpy.extractEntityChanges.and.returnValue([]); const superSyncStatusSpy = jasmine.createSpyObj('SuperSyncStatusService', [ 'updatePendingOpsStatus', diff --git a/src/app/op-log/sync/operation-write-flush.service.ts b/src/app/op-log/sync/operation-write-flush.service.ts index f01cf0f2f2..b36456380d 100644 --- a/src/app/op-log/sync/operation-write-flush.service.ts +++ b/src/app/op-log/sync/operation-write-flush.service.ts @@ -15,14 +15,15 @@ import { LOCK_NAMES } from '../core/operation-log.const'; * * ### Two-Phase Wait Strategy * - * **Phase 1: Wait for RxJS Queue to Drain** - * The NgRx effect uses `concatMap` for sequential processing. Actions are enqueued - * in the OperationCaptureService synchronously (in meta-reducer), then dequeued by - * the effect. We poll the queue size until it reaches 0, meaning all dispatched - * actions have been processed by the effect. + * **Phase 1: Wait for the pending counter to reach 0** + * The NgRx effect uses `concatMap` for sequential processing. The meta-reducer + * increments OperationCaptureService's pending counter synchronously when it + * captures an action; the effect decrements it (in a `finally`) after each write + * attempt completes. We poll the counter until it reaches 0, meaning all + * dispatched actions have been processed by the effect. * * **Phase 2: Acquire Write Lock** - * Once the RxJS queue is drained, we acquire the same lock used by `writeOperation()`. + * Once the counter is drained, we acquire the same lock used by `writeOperation()`. * This ensures the final write has completed its IndexedDB transaction. * * This two-phase approach handles the case where many actions are dispatched rapidly @@ -47,49 +48,43 @@ export class OperationWriteFlushService { * Waits for all pending operation writes to complete. * * This is a two-phase wait: - * 1. Poll the capture service queue until it's empty (all actions processed by effect) + * 1. Poll the capture service pending counter until it's 0 (all actions + * processed by the effect) * 2. Acquire the write lock to ensure the final IndexedDB transaction is complete * * @returns Promise that resolves when all pending writes are complete - * @throws Error if timeout is reached while waiting for queue to drain + * @throws Error if timeout is reached while waiting for the counter to drain */ async flushPendingWrites(): Promise { - // Phase 1: Wait for the capture service queue to drain - // This ensures all dispatched actions have been dequeued by the effect + // Phase 1: Wait for the capture service pending counter to drain + // This ensures all dispatched actions have been processed by the effect const startTime = Date.now(); - let lastLoggedSize = -1; - const initialQueueSize = this.captureService.getQueueSize(); + let lastLoggedCount = -1; + const initialPendingCount = this.captureService.getPendingCount(); OpLog.normal( - `OperationWriteFlushService: Starting flush. Initial queue size: ${initialQueueSize}`, + `OperationWriteFlushService: Starting flush. Initial pending count: ${initialPendingCount}`, ); - while (this.captureService.getQueueSize() > 0) { - const queueSize = this.captureService.getQueueSize(); + while (this.captureService.getPendingCount() > 0) { + const pendingCount = this.captureService.getPendingCount(); - // Log progress periodically (when queue size changes significantly) - if (queueSize !== lastLoggedSize && queueSize % 10 === 0) { + // Log progress periodically (when pending count changes significantly) + if (pendingCount !== lastLoggedCount && pendingCount % 10 === 0) { OpLog.verbose( - `OperationWriteFlushService: Waiting for queue to drain, current size: ${queueSize}`, + `OperationWriteFlushService: Waiting for writes to drain, pending: ${pendingCount}`, ); - lastLoggedSize = queueSize; + lastLoggedCount = pendingCount; } // Check for timeout if (Date.now() - startTime > this.MAX_WAIT_TIME) { - // Get diagnostic info about stuck operations - const pendingOps = this.captureService.peekPendingOperations(); - const sampleOps = pendingOps.slice(0, 5).map((op) => ({ - opType: op.opType, - entityType: op.entityType, - entityId: op.entityId, - })); OpLog.err( - `OperationWriteFlushService: Timeout waiting for queue to drain. ` + - `Queue still has ${queueSize} items after ${this.MAX_WAIT_TIME}ms.`, - { queueSize, sampleOps }, + `OperationWriteFlushService: Timeout waiting for writes to drain. ` + + `${pendingCount} operation(s) still pending after ${this.MAX_WAIT_TIME}ms.`, + { pendingCount }, ); throw new Error( - `Operation write flush timeout: queue still has ${queueSize} pending items. ` + + `Operation write flush timeout: ${pendingCount} pending operation(s). ` + `This may indicate a stuck effect. Try reloading the app.`, ); } @@ -106,10 +101,10 @@ export class OperationWriteFlushService { }); const totalWait = Date.now() - startTime; - const finalQueueSize = this.captureService.getQueueSize(); + const finalPendingCount = this.captureService.getPendingCount(); OpLog.normal( `OperationWriteFlushService: Flush complete in ${totalWait}ms. ` + - `Initial queue: ${initialQueueSize}, Final queue: ${finalQueueSize}`, + `Initial pending: ${initialPendingCount}, Final pending: ${finalPendingCount}`, ); } } 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 ddb3fce583..74b2aadcb4 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 @@ -105,7 +105,7 @@ describe('Meta-reducer ordering integration', () => { enqueuedActions = []; // Spy on enqueue to capture what actions are being passed - spyOn(captureService, 'enqueue').and.callFake((action: PersistentAction) => { + spyOn(captureService, 'incrementPending').and.callFake((action: PersistentAction) => { enqueuedActions.push(action); }); @@ -141,7 +141,7 @@ describe('Meta-reducer ordering integration', () => { expect(resultState[TASKS_FEATURE].entities['task-1']).toBeUndefined(); // Verify enqueue was called with the action - expect(captureService.enqueue).toHaveBeenCalledWith(action); + expect(captureService.incrementPending).toHaveBeenCalledWith(action); expect(enqueuedActions.length).toBe(1); expect(enqueuedActions[0].meta.entityId).toBe('task-1'); }); @@ -166,7 +166,7 @@ describe('Meta-reducer ordering integration', () => { expect(resultState[TASKS_FEATURE].ids).not.toContain('task-1'); // enqueue should still be called with the action - expect(captureService.enqueue).toHaveBeenCalledWith(action); + expect(captureService.incrementPending).toHaveBeenCalledWith(action); expect(enqueuedActions.length).toBe(1); }); @@ -219,7 +219,7 @@ describe('Meta-reducer ordering integration', () => { composedReducer(initialState, remoteAction); // Should NOT enqueue remote actions - expect(captureService.enqueue).not.toHaveBeenCalled(); + expect(captureService.incrementPending).not.toHaveBeenCalled(); expect(enqueuedActions.length).toBe(0); }); }); @@ -241,7 +241,7 @@ describe('Meta-reducer ordering integration', () => { composedReducer(initialState, action); // Should NOT enqueue when sync is in progress - expect(captureService.enqueue).not.toHaveBeenCalled(); + expect(captureService.incrementPending).not.toHaveBeenCalled(); }); it('should resume capturing after sync completes', () => { diff --git a/src/app/op-log/testing/integration/multi-entity-atomicity.integration.spec.ts b/src/app/op-log/testing/integration/multi-entity-atomicity.integration.spec.ts index 598ecc0757..e2f612a127 100644 --- a/src/app/op-log/testing/integration/multi-entity-atomicity.integration.spec.ts +++ b/src/app/op-log/testing/integration/multi-entity-atomicity.integration.spec.ts @@ -29,10 +29,8 @@ describe('Multi-Entity Atomicity Integration', () => { /** * Helper to compute entity changes using the service. */ - const computeChanges = (action: PersistentAction): EntityChange[] => { - service.enqueue(action); - return service.dequeue(); - }; + const computeChanges = (action: PersistentAction): EntityChange[] => + service.extractEntityChanges(action); beforeEach(() => { service = new OperationCaptureService(); @@ -213,9 +211,9 @@ describe('Multi-Entity Atomicity Integration', () => { }); }); - describe('queue behavior', () => { - it('should maintain FIFO order for multiple actions', () => { - // Enqueue multiple TIME_TRACKING actions + describe('entityChanges extraction for multiple actions', () => { + it('should extract independent entity changes per action', () => { + // Multiple TIME_TRACKING actions const action1: PersistentAction = { type: '[TimeTracking] Sync Time Tracking', contextType: 'PROJECT', @@ -244,18 +242,11 @@ describe('Multi-Entity Atomicity Integration', () => { } satisfies PersistentActionMeta, }; - service.enqueue(action1); - service.enqueue(action2); - - expect(service.getQueueSize()).toBe(2); - - const changes1 = service.dequeue(); + const changes1 = service.extractEntityChanges(action1); expect(changes1[0].entityId).toBe('PROJECT:p1:2024-01-15'); - expect(service.getQueueSize()).toBe(1); - const changes2 = service.dequeue(); + const changes2 = service.extractEntityChanges(action2); expect(changes2[0].entityId).toBe('TAG:t1:2024-01-16'); - expect(service.getQueueSize()).toBe(0); }); }); diff --git a/src/app/op-log/testing/integration/race-conditions.integration.spec.ts b/src/app/op-log/testing/integration/race-conditions.integration.spec.ts index a047e77b1e..73ab15baff 100644 --- a/src/app/op-log/testing/integration/race-conditions.integration.spec.ts +++ b/src/app/op-log/testing/integration/race-conditions.integration.spec.ts @@ -55,14 +55,14 @@ describe('Race Condition Integration Tests', () => { }); describe('Flush polling race conditions', () => { - it('should handle rapid queue operations during flush', async () => { - // Simulate the scenario where actions are enqueued rapidly - // while we're polling for the queue to drain - const enqueueCount = 10; + it('should handle rapid pending operations during flush', async () => { + // Simulate the scenario where actions are captured rapidly + // while we're polling for the pending counter to drain + const captureCount = 10; - // Start with some operations in the queue (simulating real capture) - for (let i = 0; i < enqueueCount; i++) { - captureService.enqueue({ + // Start with some operations pending (simulating real capture) + for (let i = 0; i < captureCount; i++) { + captureService.incrementPending({ type: `[Task] Test Action ${i}`, meta: { entityType: 'TASK' as any, @@ -71,42 +71,42 @@ describe('Race Condition Integration Tests', () => { } as any); } - // Verify initial queue state - expect(captureService.getQueueSize()).toBe(enqueueCount); + // Verify initial pending state + expect(captureService.getPendingCount()).toBe(captureCount); - // Drain the queue (simulating the effect processing) - for (let i = 0; i < enqueueCount; i++) { - captureService.dequeue(); + // Drain the counter (simulating the effect processing) + for (let i = 0; i < captureCount; i++) { + captureService.decrementPending(); } - // Queue should now be empty - expect(captureService.getQueueSize()).toBe(0); + // Counter should now be empty + expect(captureService.getPendingCount()).toBe(0); }); - it('should correctly report queue size during concurrent enqueue/dequeue', async () => { - // Test that queue size is consistent under concurrent operations + it('should correctly report pending count during concurrent capture/process', async () => { + // Test that the pending count is consistent under concurrent operations const iterations = 100; - let maxObservedSize = 0; + let maxObservedCount = 0; for (let i = 0; i < iterations; i++) { - // Enqueue - captureService.enqueue({ + // Capture + captureService.incrementPending({ type: `[Task] Concurrent Action ${i}`, meta: { entityType: 'TASK' as any, entityId: `task-${i}` }, } as any); - const size = captureService.getQueueSize(); - maxObservedSize = Math.max(maxObservedSize, size); + const count = captureService.getPendingCount(); + maxObservedCount = Math.max(maxObservedCount, count); - // Dequeue immediately (simulating rapid processing) - captureService.dequeue(); + // Process immediately (simulating rapid processing) + captureService.decrementPending(); } - // Queue should be empty at the end - expect(captureService.getQueueSize()).toBe(0); + // Counter should be empty at the end + expect(captureService.getPendingCount()).toBe(0); - // At some point we should have seen items in the queue - expect(maxObservedSize).toBeGreaterThan(0); + // At some point we should have seen pending items + expect(maxObservedCount).toBeGreaterThan(0); }); }); diff --git a/src/app/op-log/testing/integration/task-done-replay.integration.spec.ts b/src/app/op-log/testing/integration/task-done-replay.integration.spec.ts index 644b887c2b..83863ddf7e 100644 --- a/src/app/op-log/testing/integration/task-done-replay.integration.spec.ts +++ b/src/app/op-log/testing/integration/task-done-replay.integration.spec.ts @@ -114,7 +114,7 @@ describe('done task operation replay', () => { ); const mockOperationCaptureService = jasmine.createSpyObj( 'OperationCaptureService', - ['dequeue'], + ['extractEntityChanges', 'decrementPending'], ); const mockSnackService = jasmine.createSpyObj('SnackService', ['open']); const mockImmediateUploadService = jasmine.createSpyObj( @@ -139,7 +139,7 @@ describe('done task operation replay', () => { ); mockCompactionService.compact.and.returnValue(Promise.resolve()); mockCompactionService.emergencyCompact.and.returnValue(Promise.resolve(true)); - mockOperationCaptureService.dequeue.and.returnValue([]); + mockOperationCaptureService.extractEntityChanges.and.returnValue([]); mockClientIdService.getOrGenerateClientId.and.returnValue(Promise.resolve('clientA')); TestBed.configureTestingModule({