mirror of
https://github.com/johannesjo/super-productivity.git
synced 2026-07-17 16:37:43 +00:00
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.
This commit is contained in:
Johannes Millan
parent
d865d21fa3
commit
b51f28f961
15 changed files with 615 additions and 339 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
|
|
|
|||
|
|
@ -242,14 +242,15 @@ export const operationCaptureMetaReducer = <S, A extends Action = Action>(
|
|||
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,
|
||||
},
|
||||
|
|
|
|||
|
|
@ -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
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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<OperationLogStoreService>;
|
||||
let snackSpy: jasmine.SpyObj<SnackService>;
|
||||
let actions$: ReplaySubject<Action>;
|
||||
|
||||
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<void> => {
|
||||
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<Action>(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 <U>(_name: string, cb: () => Promise<U>) => 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 <U>(name: string, cb: () => Promise<U>) => {
|
||||
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 <U>(name: string, _cb: () => Promise<U>) => {
|
||||
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 <U>(name: string, cb: () => Promise<U>) => {
|
||||
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();
|
||||
});
|
||||
});
|
||||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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<void> {
|
||||
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<void> {
|
||||
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, skipDequeue, options);
|
||||
await this.handleQuotaExceeded(action, isDeferredWrite, options);
|
||||
}
|
||||
} else {
|
||||
this.notifyUserAndTriggerRollback();
|
||||
|
|
@ -455,7 +501,7 @@ export class OperationLogEffects implements DeferredLocalActionsPort {
|
|||
*/
|
||||
private async handleQuotaExceeded(
|
||||
action: PersistentAction,
|
||||
skipDequeue = false,
|
||||
isDeferredWrite = false,
|
||||
options: WriteOperationOptions = {},
|
||||
): Promise<void> {
|
||||
OpLog.err(
|
||||
|
|
@ -488,7 +534,7 @@ 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
|
||||
await this.writeOperation(action, skipDequeue, options);
|
||||
await this.writeOperation(action, isDeferredWrite, options);
|
||||
this.snackService.open({
|
||||
type: 'SUCCESS',
|
||||
msg: T.F.SYNC.S.STORAGE_RECOVERED_AFTER_COMPACTION,
|
||||
|
|
@ -569,7 +615,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;
|
||||
|
|
|
|||
|
|
@ -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',
|
||||
|
|
|
|||
|
|
@ -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<void> {
|
||||
// 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}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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', () => {
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
});
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
});
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -114,7 +114,7 @@ describe('done task operation replay', () => {
|
|||
);
|
||||
const mockOperationCaptureService = jasmine.createSpyObj<OperationCaptureService>(
|
||||
'OperationCaptureService',
|
||||
['dequeue'],
|
||||
['extractEntityChanges', 'decrementPending'],
|
||||
);
|
||||
const mockSnackService = jasmine.createSpyObj<SnackService>('SnackService', ['open']);
|
||||
const mockImmediateUploadService = jasmine.createSpyObj<ImmediateUploadService>(
|
||||
|
|
@ -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({
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue