diff --git a/docs/plans/2026-05-21-clean-slate-data-loss-prevention.md b/docs/plans/2026-05-21-clean-slate-data-loss-prevention.md index 833b886726..cccfc2eed2 100644 --- a/docs/plans/2026-05-21-clean-slate-data-loss-prevention.md +++ b/docs/plans/2026-05-21-clean-slate-data-loss-prevention.md @@ -79,7 +79,7 @@ Two callers refactor to use it: `CleanSlateService.createCleanSlate()` and `Back **Implementation: single multi-store readwrite transaction.** All writes (clear OPS, append the SYNC_IMPORT entry, write vector_clock, write state_cache, optionally write archive_young / archive_old) happen inside one `db.transaction(stores, 'readwrite')`. If any step rejects the IDB request, `tx.done` rejects, the engine auto-aborts, and no committed change to any of the touched stores survives. The catch block calls an explicit `tx.abort()` as well — that branch is unreachable in production (rejected IDB requests already abort the tx) but is load-bearing for the spy-based fault-injection seam used by the interrupt integration test, where the spy throws synchronously instead of rejecting an IDB request. -**Why a single multi-store tx, not snapshot-then-swap.** The plan's first revision proposed staging the new state to a `STATE_CACHE_STAGING_KEY` row outside the destructive tx, then swapping references inside a small cross-store tx, on the premise that "every existing `db.transaction(...)` call in `operation-log-store.service.ts` is single-store" and WebKit/Capacitor had no precedent for multi-store + multi-MB. That premise was wrong: `appendWithVectorClockUpdate` already runs a 2-store (`OPS` + `VECTOR_CLOCK`) readwrite tx on every local action, including on Capacitor iOS. Staging would have cost one extra full-state structured-clone, a boot-time staging-row reconciliation step on every cold start, a catch-block cleanup, two integration tests, and ~30 lines of JSDoc — in exchange for a crash-detection sentinel nothing would have acted on. The simpler design provides the same atomicity guarantee with none of the machinery. +**Why a single multi-store tx, not snapshot-then-swap.** The plan's first revision proposed staging the new state to a `STATE_CACHE_STAGING_KEY` row outside the destructive tx, then swapping references inside a small cross-store tx, on the premise that "every existing `db.transaction(...)` call in `operation-log-store.service.ts` is single-store" and WebKit/Capacitor had no precedent for multi-store + multi-MB. That premise was wrong: `appendWithVectorClockOverwrite` already runs a 2-store (`OPS` + `VECTOR_CLOCK`) readwrite tx on every local action, including on Capacitor iOS. Staging would have cost one extra full-state structured-clone, a boot-time staging-row reconciliation step on every cold start, a catch-block cleanup, two integration tests, and ~30 lines of JSDoc — in exchange for a crash-detection sentinel nothing would have acted on. The simpler design provides the same atomicity guarantee with none of the machinery. **Payload duplication trade-off.** Both `syncImportOp.payload` (full state) and `newState` (structurally identical) are persisted in the same tx: the payload via the OPS row's encoded operation, the state via the STATE_CACHE singleton. Both writes are required — OPS holds the payload the uploader sends in the snapshot endpoint; STATE_CACHE is what `isWhollyFreshClient` reads on next launch. Eliminating the duplication would require either lazy hydration of the OPS payload from STATE_CACHE at upload time (unsafe if compaction advances STATE_CACHE past this op's seq) or a dedicated payload-staging store. Neither pays back for an infrequent (password change / backup restore) operation. @@ -306,7 +306,7 @@ Code references verified at `feat/issue-7709-567f99` HEAD `c5158dd35b`: - Corrected Fix 3 — line 606 has no `snapshotState`; synthetic descriptor for op-streaming case; introduced `RemoteDataDescriptor` discriminated union. - Corrected Fix 1 caller graph — `createCleanSlate` has only one production caller; preflight wraps three independent call sites if it ships. - Cut preflight (Fix 1) from v1; gated on log evidence from Fix 4. - - Initially switched atomicity from multi-store-tx-with-large-payload to snapshot-then-swap; later reverted to single multi-store tx after verifying `appendWithVectorClockUpdate` already uses multi-store readwrite on every action (the staging row was over-engineering on a wrong premise). + - Initially switched atomicity from multi-store-tx-with-large-payload to snapshot-then-swap; later reverted to single multi-store tx after verifying `appendWithVectorClockOverwrite` already uses multi-store readwrite on every action (the staging row was over-engineering on a wrong premise). - Switched from inline IDB calls to `runDestructiveStateReplacement` helper (BackupService is a second caller, helper justified). - Acknowledged `pre-migration-backup.service.ts` is a placeholder; PR-A implements it for real. - Removed Q3 schema-migration cost claim (STATE_CACHE rows already support optional fields). diff --git a/docs/sync-and-op-log/sqlite-migration.md b/docs/sync-and-op-log/sqlite-migration.md index 42a943fc1a..3dca9bc578 100644 --- a/docs/sync-and-op-log/sqlite-migration.md +++ b/docs/sync-and-op-log/sqlite-migration.md @@ -13,7 +13,7 @@ > and no spec breakage. > - ✅ **`OperationLogStoreService` fully migrated** — every method routes > through the adapter, including the two flagship atomic flows -> (`appendWithVectorClockUpdate`, `runDestructiveStateReplacement`). No direct +> (`appendWithVectorClockOverwrite`, `runDestructiveStateReplacement`). No direct > `this.db` calls remain. > - ✅ **`ArchiveStoreService` fully migrated** (own adopted connection + > `_withRetryOnClose` re-adopt path). @@ -102,7 +102,7 @@ non-evictable; only a full _Clear storage_ or uninstall removes it. read-only-migration and are out of scope for the data-loss fix. - **Atomicity is the hard part**, not CRUD. Two methods need single-transaction multi-store writes that MUST stay atomic: - - `appendWithVectorClockUpdate()` — OPS + VECTOR_CLOCK in one tx. + - `appendWithVectorClockOverwrite()` — OPS + VECTOR_CLOCK in one tx. - `runDestructiveStateReplacement()` — OPS + STATE*CACHE + VECTOR_CLOCK + CLIENT_ID (+ ARCHIVE*\*) in one tx (crash-safety, issues #7709, #7732). Plus: auto-increment `seq` keypath, a **unique** `byId` index, and a compound diff --git a/docs/sync-and-op-log/vector-clocks.md b/docs/sync-and-op-log/vector-clocks.md index 9e40a48aab..7becd7b8c1 100644 --- a/docs/sync-and-op-log/vector-clocks.md +++ b/docs/sync-and-op-log/vector-clocks.md @@ -77,7 +77,7 @@ interface VectorClockEntry { } ``` -The global clock is the **single source of truth** for the client's current causal knowledge. During local operation capture, it is updated atomically with operation writes (single IndexedDB transaction via `appendWithVectorClockUpdate`). The remote merge path (`mergeRemoteOpClocks`) updates the clock in a separate write after reading the current state. +The global clock is the **single source of truth** for the client's current causal knowledge. During local operation capture, it is updated atomically with operation writes (single IndexedDB transaction via `appendWithVectorClockOverwrite`). The remote merge path (`mergeRemoteOpClocks`) updates the clock in a separate write after reading the current state. ### Snapshot Clock @@ -98,7 +98,7 @@ In `operation-log.effects.ts`: 1. `VectorClockService.getCurrentVectorClock()` reads the global clock from the `vector_clock` store 2. `incrementVectorClock(currentClock, clientId)` creates a new clock with the client's counter incremented 3. The operation is created with this **full, unpruned** clock -4. `appendWithVectorClockUpdate(op, 'local')` writes the operation AND updates the global clock in a **single atomic IndexedDB transaction** +4. `appendWithVectorClockOverwrite(op, 'local')` writes the operation AND updates the global clock in a **single atomic IndexedDB transaction** **Key invariant: Normal operations carry full (unpruned) vector clocks. No client-side pruning happens during capture.** diff --git a/e2e/tests/sync/supersync-import-other-client-ops.spec.ts b/e2e/tests/sync/supersync-import-other-client-ops.spec.ts index 34657d31e1..fc976dbff1 100644 --- a/e2e/tests/sync/supersync-import-other-client-ops.spec.ts +++ b/e2e/tests/sync/supersync-import-other-client-ops.spec.ts @@ -238,7 +238,7 @@ test.describe('@supersync @pruning Other client post-import ops sync correctly', // CRITICAL: OperationLogStoreService has an in-memory _vectorClockCache that // bypasses IndexedDB reads. Without reloading, the injected entries never reach // the ops' vector clocks — getVectorClock() returns the stale cached value, - // and appendWithVectorClockUpdate() overwrites the injected IDB data. + // and appendWithVectorClockOverwrite() overwrites the injected IDB data. // Reloading destroys the Angular service, forcing a fresh read from IndexedDB. console.log( '[Other-Client Import] Phase 6b: Reloading Client B to pick up injected clock', 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 index 99ab5c2d11..961c71b4e7 100644 --- 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 @@ -81,11 +81,11 @@ describe('regression #8306: persistOperation$ stream survives write failures', ( actions$ = new ReplaySubject(1); opLogStoreSpy = jasmine.createSpyObj('OperationLogStoreService', [ - 'appendWithVectorClockUpdate', + 'appendWithVectorClockOverwrite', 'getCompactionCounter', 'clearVectorClockCache', ]); - opLogStoreSpy.appendWithVectorClockUpdate.and.resolveTo(1); + opLogStoreSpy.appendWithVectorClockOverwrite.and.resolveTo(1); opLogStoreSpy.getCompactionCounter.and.resolveTo(0); const vectorClockSpy = jasmine.createSpyObj('VectorClockService', [ @@ -176,8 +176,8 @@ describe('regression #8306: persistOperation$ stream survives write failures', ( // 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() + expect(opLogStoreSpy.appendWithVectorClockOverwrite).toHaveBeenCalledTimes(1); + const writtenOp = opLogStoreSpy.appendWithVectorClockOverwrite.calls.mostRecent() .args[0] as Operation; expect(writtenOp.entityId).toBe('task-ok'); // The user was told the failed write needs a reload. @@ -211,7 +211,7 @@ describe('regression #8306: persistOperation$ stream survives write failures', ( await waitFor(() => captureService.getPendingCount() === 0); expect(captureService.getPendingCount()).toBe(0); - expect(opLogStoreSpy.appendWithVectorClockUpdate).not.toHaveBeenCalled(); + expect(opLogStoreSpy.appendWithVectorClockOverwrite).not.toHaveBeenCalled(); sub.unsubscribe(); }); @@ -249,8 +249,8 @@ describe('regression #8306: persistOperation$ stream survives write failures', ( await waitFor(() => captureService.getPendingCount() === 0); expect(streamErrored).toBe(false); - expect(opLogStoreSpy.appendWithVectorClockUpdate).toHaveBeenCalledTimes(1); - const writtenOp = opLogStoreSpy.appendWithVectorClockUpdate.calls.mostRecent() + expect(opLogStoreSpy.appendWithVectorClockOverwrite).toHaveBeenCalledTimes(1); + const writtenOp = opLogStoreSpy.appendWithVectorClockOverwrite.calls.mostRecent() .args[0] as Operation; expect(writtenOp.entityId).toBe('survivor'); 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 1140ba1045..bfd5297c86 100644 --- a/src/app/op-log/capture/operation-log.effects.spec.ts +++ b/src/app/op-log/capture/operation-log.effects.spec.ts @@ -56,7 +56,7 @@ describe('OperationLogEffects', () => { beforeEach(() => { mockOpLogStore = jasmine.createSpyObj('OperationLogStoreService', [ 'append', - 'appendWithVectorClockUpdate', + 'appendWithVectorClockOverwrite', 'getCompactionCounter', 'clearVectorClockCache', ]); @@ -86,7 +86,7 @@ describe('OperationLogEffects', () => { fn(), ); mockOpLogStore.append.and.returnValue(Promise.resolve(1)); - mockOpLogStore.appendWithVectorClockUpdate.and.returnValue(Promise.resolve(1)); + mockOpLogStore.appendWithVectorClockOverwrite.and.returnValue(Promise.resolve(1)); mockOpLogStore.getCompactionCounter.and.returnValue(Promise.resolve(0)); mockVectorClockService.getCurrentVectorClock.and.returnValue( Promise.resolve({ testClient: 5 }), @@ -133,7 +133,7 @@ describe('OperationLogEffects', () => { effects.persistOperation$.subscribe({ complete: () => { - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledWith( + expect(mockOpLogStore.appendWithVectorClockOverwrite).toHaveBeenCalledWith( jasmine.objectContaining({ actionType: ActionType.TASK_SHARED_UPDATE, opType: OpType.Update, @@ -153,7 +153,7 @@ describe('OperationLogEffects', () => { effects.persistOperation$.subscribe({ complete: () => { - expect(mockOpLogStore.appendWithVectorClockUpdate).not.toHaveBeenCalled(); + expect(mockOpLogStore.appendWithVectorClockOverwrite).not.toHaveBeenCalled(); done(); }, }); @@ -165,7 +165,7 @@ describe('OperationLogEffects', () => { effects.persistOperation$.subscribe({ complete: () => { - expect(mockOpLogStore.appendWithVectorClockUpdate).not.toHaveBeenCalled(); + expect(mockOpLogStore.appendWithVectorClockOverwrite).not.toHaveBeenCalled(); done(); }, }); @@ -203,7 +203,7 @@ describe('OperationLogEffects', () => { callOrder.push('clientId'); return 'testClient'; }); - mockOpLogStore.appendWithVectorClockUpdate.and.callFake(async () => { + mockOpLogStore.appendWithVectorClockOverwrite.and.callFake(async () => { callOrder.push('append'); return 1; }); @@ -244,13 +244,13 @@ describe('OperationLogEffects', () => { effects.persistOperation$.subscribe({ complete: () => { - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledWith( + expect(mockOpLogStore.appendWithVectorClockOverwrite).toHaveBeenCalledWith( jasmine.objectContaining({ clientId: 'newClient' }), 'local', ); // Negative assertion: if clientId were read before lock acquisition, // we'd see 'oldClient'. The fix prevents that. - expect(mockOpLogStore.appendWithVectorClockUpdate).not.toHaveBeenCalledWith( + expect(mockOpLogStore.appendWithVectorClockOverwrite).not.toHaveBeenCalledWith( jasmine.objectContaining({ clientId: 'oldClient' }), jasmine.anything(), ); @@ -267,7 +267,7 @@ describe('OperationLogEffects', () => { complete: () => { expect(mockVectorClockService.getCurrentVectorClock).toHaveBeenCalled(); const appendCall = - mockOpLogStore.appendWithVectorClockUpdate.calls.mostRecent(); + mockOpLogStore.appendWithVectorClockOverwrite.calls.mostRecent(); const operation = appendCall.args[0]; expect(operation.vectorClock['testClient']).toBe(6); // Incremented from 5 done(); @@ -276,7 +276,7 @@ describe('OperationLogEffects', () => { }); // Note: Tests for incrementVectorClockForLocalChange have been removed. - // Vector clock updates are now handled atomically within appendWithVectorClockUpdate. + // Vector clock updates are now handled atomically within appendWithVectorClockOverwrite. it('should trigger compaction when threshold reached', fakeAsync(() => { // Counter starts at threshold - 1, after increment it reaches threshold @@ -318,7 +318,7 @@ describe('OperationLogEffects', () => { effects.persistOperation$.subscribe({ complete: () => { const appendCall = - mockOpLogStore.appendWithVectorClockUpdate.calls.mostRecent(); + mockOpLogStore.appendWithVectorClockOverwrite.calls.mostRecent(); const operation = appendCall.args[0]; // Payload now uses MultiEntityPayload structure with actionPayload and entityChanges expect(operation.payload).toEqual({ @@ -343,7 +343,7 @@ describe('OperationLogEffects', () => { effects.persistOperation$.subscribe({ complete: () => { expect(mockOperationCaptureService.extractEntityChanges).toHaveBeenCalled(); - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledWith( + expect(mockOpLogStore.appendWithVectorClockOverwrite).toHaveBeenCalledWith( jasmine.objectContaining({ actionType: ActionType.GLOBAL_CONFIG_UPDATE_SECTION, payload: { @@ -379,7 +379,7 @@ describe('OperationLogEffects', () => { effects.persistOperation$.subscribe({ complete: () => { const operation = - mockOpLogStore.appendWithVectorClockUpdate.calls.mostRecent().args[0]; + mockOpLogStore.appendWithVectorClockOverwrite.calls.mostRecent().args[0]; expect(operation.actionType).toBe(ActionType.GLOBAL_CONFIG_UPDATE_SECTION); expect(operation.payload).toEqual({ @@ -409,7 +409,7 @@ describe('OperationLogEffects', () => { effects.persistOperation$.subscribe({ complete: () => { const operation = - mockOpLogStore.appendWithVectorClockUpdate.calls.mostRecent().args[0]; + mockOpLogStore.appendWithVectorClockOverwrite.calls.mostRecent().args[0]; expect(operation.actionType).toBe(ActionType.TASK_SHARED_PLAN_FOR_TODAY); expect(operation.payload).toEqual({ @@ -439,7 +439,7 @@ describe('OperationLogEffects', () => { complete: () => { try { const operation = - mockOpLogStore.appendWithVectorClockUpdate.calls.mostRecent().args[0]; + mockOpLogStore.appendWithVectorClockOverwrite.calls.mostRecent().args[0]; expect(operation.timestamp).toBe(now.getTime()); expect(operation.payload).toEqual({ @@ -467,7 +467,7 @@ describe('OperationLogEffects', () => { }); it('should notify user on persistence error', (done) => { - mockOpLogStore.appendWithVectorClockUpdate.and.rejectWith( + mockOpLogStore.appendWithVectorClockOverwrite.and.rejectWith( new Error('Write failed'), ); const action = createPersistentAction(ActionType.TASK_SHARED_UPDATE); @@ -489,7 +489,7 @@ describe('OperationLogEffects', () => { const quotaError = new DOMException('Quota exceeded', 'QuotaExceededError'); // First call fails with quota error, second call (retry) succeeds let callCount = 0; - mockOpLogStore.appendWithVectorClockUpdate.and.callFake(() => { + mockOpLogStore.appendWithVectorClockOverwrite.and.callFake(() => { callCount++; if (callCount === 1) { return Promise.reject(quotaError); @@ -504,7 +504,7 @@ describe('OperationLogEffects', () => { tick(100); expect(mockCompactionService.emergencyCompact).toHaveBeenCalled(); // Should have tried to append twice (initial + retry after compaction) - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledTimes(2); + expect(mockOpLogStore.appendWithVectorClockOverwrite).toHaveBeenCalledTimes(2); })); it('re-extracts the SAME action on the quota-exceeded retry, never a second op (#8307)', fakeAsync(() => { @@ -515,7 +515,7 @@ describe('OperationLogEffects', () => { // double-dequeue did. const quotaError = new DOMException('Quota exceeded', 'QuotaExceededError'); let appendCount = 0; - mockOpLogStore.appendWithVectorClockUpdate.and.callFake(() => { + mockOpLogStore.appendWithVectorClockOverwrite.and.callFake(() => { appendCount++; if (appendCount === 1) { return Promise.reject(quotaError); @@ -531,7 +531,7 @@ describe('OperationLogEffects', () => { tick(100); // 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(mockOpLogStore.appendWithVectorClockOverwrite).toHaveBeenCalledTimes(2); expect(mockOperationCaptureService.extractEntityChanges).toHaveBeenCalledTimes(2); expect(mockOperationCaptureService.extractEntityChanges).toHaveBeenCalledWith( action, @@ -547,7 +547,7 @@ describe('OperationLogEffects', () => { effects.persistOperation$.subscribe({ complete: () => { const firstOp = - mockOpLogStore.appendWithVectorClockUpdate.calls.mostRecent().args[0]; + mockOpLogStore.appendWithVectorClockOverwrite.calls.mostRecent().args[0]; expect(firstOp.clientId).toBe('testClient'); // Simulate a backup import rotating the client ID (BackupService's @@ -562,7 +562,7 @@ describe('OperationLogEffects', () => { effects.persistOperation$.subscribe({ complete: () => { const secondOp = - mockOpLogStore.appendWithVectorClockUpdate.calls.mostRecent().args[0]; + mockOpLogStore.appendWithVectorClockOverwrite.calls.mostRecent().args[0]; expect(secondOp.clientId).toBe('newImportClient'); done(); }, @@ -574,7 +574,7 @@ describe('OperationLogEffects', () => { it('should show error when retry after emergency compaction fails with quota error', fakeAsync(() => { const quotaError = new DOMException('Quota exceeded', 'QuotaExceededError'); // All attempts fail with quota error (nested quota failure) - mockOpLogStore.appendWithVectorClockUpdate.and.returnValue( + mockOpLogStore.appendWithVectorClockOverwrite.and.returnValue( Promise.reject(quotaError), ); const action = createPersistentAction(ActionType.TASK_SHARED_UPDATE); @@ -596,7 +596,7 @@ describe('OperationLogEffects', () => { it('should abort immediately when quota error during retry (circuit breaker)', fakeAsync(() => { const quotaError = new DOMException('Quota exceeded', 'QuotaExceededError'); // First call fails with quota, emergency compaction succeeds, retry also fails with quota - mockOpLogStore.appendWithVectorClockUpdate.and.callFake(() => { + mockOpLogStore.appendWithVectorClockOverwrite.and.callFake(() => { return Promise.reject(quotaError); }); const action = createPersistentAction(ActionType.TASK_SHARED_UPDATE); @@ -606,7 +606,7 @@ describe('OperationLogEffects', () => { tick(100); // Should have tried twice (initial + one retry after compaction) - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledTimes(2); + expect(mockOpLogStore.appendWithVectorClockOverwrite).toHaveBeenCalledTimes(2); // Should not trigger recursive compaction expect(mockCompactionService.emergencyCompact).toHaveBeenCalledTimes(1); // User should see error snackbar @@ -615,7 +615,7 @@ describe('OperationLogEffects', () => { it('should show error when emergency compaction itself fails', fakeAsync(() => { const quotaError = new DOMException('Quota exceeded', 'QuotaExceededError'); - mockOpLogStore.appendWithVectorClockUpdate.and.returnValue( + mockOpLogStore.appendWithVectorClockOverwrite.and.returnValue( Promise.reject(quotaError), ); // Emergency compaction fails @@ -628,7 +628,7 @@ describe('OperationLogEffects', () => { tick(100); expect(mockCompactionService.emergencyCompact).toHaveBeenCalled(); // No retry after failed compaction - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledTimes(1); + expect(mockOpLogStore.appendWithVectorClockOverwrite).toHaveBeenCalledTimes(1); // User should be notified expect(mockSnackService.open).toHaveBeenCalledWith( jasmine.objectContaining({ @@ -643,7 +643,7 @@ describe('OperationLogEffects', () => { 'NS_ERROR_DOM_QUOTA_REACHED', ); let callCount = 0; - mockOpLogStore.appendWithVectorClockUpdate.and.callFake(() => { + mockOpLogStore.appendWithVectorClockOverwrite.and.callFake(() => { callCount++; if (callCount === 1) { return Promise.reject(firefoxQuotaError); @@ -670,7 +670,7 @@ describe('OperationLogEffects', () => { }) as DOMException; let callCount = 0; - mockOpLogStore.appendWithVectorClockUpdate.and.callFake(() => { + mockOpLogStore.appendWithVectorClockOverwrite.and.callFake(() => { callCount++; if (callCount === 1) { return Promise.reject(safariQuotaError); @@ -689,7 +689,7 @@ describe('OperationLogEffects', () => { it('should not treat regular DOMException as quota error', fakeAsync(() => { const regularError = new DOMException('Read failed', 'NotReadableError'); - mockOpLogStore.appendWithVectorClockUpdate.and.returnValue( + mockOpLogStore.appendWithVectorClockOverwrite.and.returnValue( Promise.reject(regularError), ); const action = createPersistentAction(ActionType.TASK_SHARED_UPDATE); @@ -711,7 +711,7 @@ describe('OperationLogEffects', () => { it('should show success message after recovery from quota exceeded', fakeAsync(() => { const quotaError = new DOMException('Quota exceeded', 'QuotaExceededError'); let callCount = 0; - mockOpLogStore.appendWithVectorClockUpdate.and.callFake(() => { + mockOpLogStore.appendWithVectorClockOverwrite.and.callFake(() => { callCount++; if (callCount === 1) { return Promise.reject(quotaError); @@ -809,7 +809,7 @@ describe('OperationLogEffects', () => { await port.processDeferredActions(); - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledWith( + expect(mockOpLogStore.appendWithVectorClockOverwrite).toHaveBeenCalledWith( jasmine.objectContaining({ actionType: ActionType.TASK_SHARED_UPDATE, clientId: 'testClient', @@ -821,7 +821,7 @@ describe('OperationLogEffects', () => { it('should do nothing when no deferred actions are buffered', async () => { await expectAsync(effects.processDeferredActions()).toBeResolved(); - expect(mockOpLogStore.appendWithVectorClockUpdate).not.toHaveBeenCalled(); + expect(mockOpLogStore.appendWithVectorClockOverwrite).not.toHaveBeenCalled(); }); it('should process a single deferred action', async () => { @@ -830,7 +830,7 @@ describe('OperationLogEffects', () => { await effects.processDeferredActions(); - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledWith( + expect(mockOpLogStore.appendWithVectorClockOverwrite).toHaveBeenCalledWith( jasmine.objectContaining({ actionType: ActionType.TASK_SHARED_UPDATE, clientId: 'testClient', @@ -850,9 +850,9 @@ describe('OperationLogEffects', () => { await effects.processDeferredActions(); - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledTimes(3); + expect(mockOpLogStore.appendWithVectorClockOverwrite).toHaveBeenCalledTimes(3); - const calls = mockOpLogStore.appendWithVectorClockUpdate.calls.all(); + const calls = mockOpLogStore.appendWithVectorClockOverwrite.calls.all(); expect(calls[0].args[0].actionType).toBe(ActionType.TASK_SHARED_ADD); expect(calls[1].args[0].actionType).toBe(ActionType.TASK_SHARED_UPDATE); expect(calls[2].args[0].actionType).toBe(ActionType.TASK_SHARED_DELETE); @@ -865,10 +865,10 @@ describe('OperationLogEffects', () => { await effects.processDeferredActions(); // Call again - should not process anything (buffer cleared) - mockOpLogStore.appendWithVectorClockUpdate.calls.reset(); + mockOpLogStore.appendWithVectorClockOverwrite.calls.reset(); await effects.processDeferredActions(); - expect(mockOpLogStore.appendWithVectorClockUpdate).not.toHaveBeenCalled(); + expect(mockOpLogStore.appendWithVectorClockOverwrite).not.toHaveBeenCalled(); }); it('should continue processing remaining actions when one fails', async () => { @@ -880,7 +880,7 @@ describe('OperationLogEffects', () => { // First action fails, second succeeds let callCount = 0; - mockOpLogStore.appendWithVectorClockUpdate.and.callFake(() => { + mockOpLogStore.appendWithVectorClockOverwrite.and.callFake(() => { callCount++; if (callCount === 1) { return Promise.reject(new Error('First action failed')); @@ -892,7 +892,7 @@ describe('OperationLogEffects', () => { await expectAsync(effects.processDeferredActions()).toBeResolved(); // The failed write is retried once, then processing continues to action 2. - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledTimes(3); + expect(mockOpLogStore.appendWithVectorClockOverwrite).toHaveBeenCalledTimes(3); }); it('should stop the drain at an exhausted transient failure, keeping it AND its successors queued in order', async () => { @@ -903,7 +903,7 @@ describe('OperationLogEffects', () => { const successorAction = createPersistentAction(ActionType.TASK_SHARED_UPDATE); bufferDeferredAction(failedAction); bufferDeferredAction(successorAction); - mockOpLogStore.appendWithVectorClockUpdate.and.rejectWith( + mockOpLogStore.appendWithVectorClockOverwrite.and.rejectWith( new Error('transient failure'), ); @@ -911,7 +911,7 @@ describe('OperationLogEffects', () => { // Only the failed action was attempted (3 retries); the successor was // never written out of order and both remain buffered. - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledTimes(3); + expect(mockOpLogStore.appendWithVectorClockOverwrite).toHaveBeenCalledTimes(3); expect(getDeferredActions()).toEqual([failedAction, successorAction]); expect(mockSnackService.open).toHaveBeenCalledWith( jasmine.objectContaining({ @@ -920,12 +920,12 @@ describe('OperationLogEffects', () => { }), ); - mockOpLogStore.appendWithVectorClockUpdate.calls.reset(); - mockOpLogStore.appendWithVectorClockUpdate.and.resolveTo(2); + mockOpLogStore.appendWithVectorClockOverwrite.calls.reset(); + mockOpLogStore.appendWithVectorClockOverwrite.and.resolveTo(2); await effects.processDeferredActions(); // Next window drains both in the original order. - const calls = mockOpLogStore.appendWithVectorClockUpdate.calls.all(); + const calls = mockOpLogStore.appendWithVectorClockOverwrite.calls.all(); expect(calls.length).toBe(2); expect(calls[0].args[0].actionType).toBe(ActionType.TASK_SHARED_ADD); expect(calls[1].args[0].actionType).toBe(ActionType.TASK_SHARED_UPDATE); @@ -945,7 +945,7 @@ describe('OperationLogEffects', () => { // The invalid reducer action already changed live state. Neither it nor // its successor may be discarded or persisted out of order. - expect(mockOpLogStore.appendWithVectorClockUpdate).not.toHaveBeenCalled(); + expect(mockOpLogStore.appendWithVectorClockOverwrite).not.toHaveBeenCalled(); expect(getDeferredActions()).toEqual([invalidAction, validAction]); expect(mockSnackService.open).toHaveBeenCalledWith( jasmine.objectContaining({ @@ -964,7 +964,7 @@ describe('OperationLogEffects', () => { bufferDeferredAction(action); let resolveFirstWrite!: (seq: number) => void; - mockOpLogStore.appendWithVectorClockUpdate.and.returnValue( + mockOpLogStore.appendWithVectorClockOverwrite.and.returnValue( new Promise((resolve) => { resolveFirstWrite = resolve; }), @@ -977,12 +977,12 @@ describe('OperationLogEffects', () => { resolveFirstWrite(1); await Promise.all([firstDrain, secondDrain]); - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledTimes(1); + expect(mockOpLogStore.appendWithVectorClockOverwrite).toHaveBeenCalledTimes(1); expect(getDeferredActions()).toEqual([]); }); it('should not re-append a deferred action when post-append bookkeeping fails', async () => { - // After appendWithVectorClockUpdate commits, a bookkeeping throw (e.g. + // After appendWithVectorClockOverwrite commits, a bookkeeping throw (e.g. // getCompactionCounter) must not bubble into the retry loop — that would // append the same user action again under a fresh op id and double-apply // additive payloads on every client. @@ -994,7 +994,7 @@ describe('OperationLogEffects', () => { await effects.processDeferredActions(); - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledTimes(1); + expect(mockOpLogStore.appendWithVectorClockOverwrite).toHaveBeenCalledTimes(1); expect(getDeferredActions()).toEqual([]); expect(mockSnackService.open).not.toHaveBeenCalledWith( jasmine.objectContaining({ msg: T.F.SYNC.S.DEFERRED_ACTION_FAILED }), @@ -1012,7 +1012,7 @@ describe('OperationLogEffects', () => { await effects.processDeferredActions(); - const appendCall = mockOpLogStore.appendWithVectorClockUpdate.calls.mostRecent(); + const appendCall = mockOpLogStore.appendWithVectorClockOverwrite.calls.mostRecent(); const operation = appendCall.args[0]; // Vector clock should be incremented from current value (includes remote ops) @@ -1027,7 +1027,7 @@ describe('OperationLogEffects', () => { await effects.processDeferredActions({ callerHoldsOperationLogLock: true }); expect(mockLockService.request).not.toHaveBeenCalled(); - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledWith( + expect(mockOpLogStore.appendWithVectorClockOverwrite).toHaveBeenCalledWith( jasmine.objectContaining({ actionType: ActionType.TASK_SHARED_UPDATE, clientId: 'testClient', @@ -1039,7 +1039,7 @@ describe('OperationLogEffects', () => { it('should not run emergency compaction while caller already holds operation log lock', async () => { const action = createPersistentAction(ActionType.TASK_SHARED_UPDATE); bufferDeferredAction(action); - mockOpLogStore.appendWithVectorClockUpdate.and.rejectWith( + mockOpLogStore.appendWithVectorClockOverwrite.and.rejectWith( new DOMException('Quota exceeded', 'QuotaExceededError'), ); @@ -1074,7 +1074,7 @@ describe('OperationLogEffects', () => { it('should surface DEFERRED_ACTION_FAILED when quota fires under caller-holds-lock', async () => { const action = createPersistentAction(ActionType.TASK_SHARED_UPDATE); bufferDeferredAction(action); - mockOpLogStore.appendWithVectorClockUpdate.and.rejectWith( + mockOpLogStore.appendWithVectorClockOverwrite.and.rejectWith( new DOMException('Quota exceeded', 'QuotaExceededError'), ); @@ -1103,7 +1103,7 @@ describe('OperationLogEffects', () => { // 3. The retry loop saw the throw and actually retried — appendWith* // was attempted MAX_RETRIES=3 times, not once. Pre-fix the loop // would have broken on attempt #1 with success=true. - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledTimes(3); + expect(mockOpLogStore.appendWithVectorClockOverwrite).toHaveBeenCalledTimes(3); // 4. DEFERRED_ACTION_FAILED fires ONLY when the retry loop's // failedCount > 0 after all retries — the loud-fail outcome. expect(mockSnackService.open).toHaveBeenCalledWith( diff --git a/src/app/op-log/capture/operation-log.effects.ts b/src/app/op-log/capture/operation-log.effects.ts index a3ddc5d4ae..3fe6003bdd 100644 --- a/src/app/op-log/capture/operation-log.effects.ts +++ b/src/app/op-log/capture/operation-log.effects.ts @@ -334,7 +334,7 @@ export class OperationLogEffects implements DeferredLocalActionsPort { // (one to SUP_OPS, one to pf.META_MODEL) into a single atomic transaction, // reducing disk I/O by ~50% on mobile devices. // The op.vectorClock already contains the incremented clock (from newClock above). - await this.opLogStore.appendWithVectorClockUpdate(op, 'local'); + await this.opLogStore.appendWithVectorClockOverwrite(op, 'local'); // The op is durably committed past this point. Bookkeeping failures // below must NOT propagate: a throw would send the deferred retry loop diff --git a/src/app/op-log/persistence/op-log-db-adapter.ts b/src/app/op-log/persistence/op-log-db-adapter.ts index 01c6b2e012..f6a33fab31 100644 --- a/src/app/op-log/persistence/op-log-db-adapter.ts +++ b/src/app/op-log/persistence/op-log-db-adapter.ts @@ -13,7 +13,7 @@ * - Singleton stores (`vector_clock`, `client_id`, `state_cache`, * `import_backup`, archive) are keyed by a known string key. * - Two flows require atomic multi-store writes: - * `appendWithVectorClockUpdate` (OPS + VECTOR_CLOCK) and + * `appendWithVectorClockOverwrite` (OPS + VECTOR_CLOCK) and * `runDestructiveStateReplacement` (OPS + STATE_CACHE + VECTOR_CLOCK + * CLIENT_ID + archive). Hence the callback-based {@link transaction}: commit * on resolve, roll back on throw — the one shape both IDB auto-commit and diff --git a/src/app/op-log/persistence/operation-log-store.service.spec.ts b/src/app/op-log/persistence/operation-log-store.service.spec.ts index d924be4961..95a263ad5b 100644 --- a/src/app/op-log/persistence/operation-log-store.service.spec.ts +++ b/src/app/op-log/persistence/operation-log-store.service.spec.ts @@ -3430,13 +3430,13 @@ describe('OperationLogStoreService', () => { }); }); - describe('appendWithVectorClockUpdate', () => { + describe('appendWithVectorClockOverwrite', () => { it('should append operation and update vector clock atomically for local ops', async () => { const op = createTestOperation({ vectorClock: { testClient: 1 }, }); - const seq = await service.appendWithVectorClockUpdate(op, 'local'); + const seq = await service.appendWithVectorClockOverwrite(op, 'local'); // Operation should be stored const ops = await service.getOpsAfterSeq(0); @@ -3458,7 +3458,7 @@ describe('OperationLogStoreService', () => { vectorClock: { remoteClient: 10 }, }); - await service.appendWithVectorClockUpdate(remoteOp, 'remote'); + await service.appendWithVectorClockOverwrite(remoteOp, 'remote'); // Vector clock should NOT be updated (remote ops don't change local clock) const clock = await service.getVectorClock(); @@ -3479,15 +3479,15 @@ describe('OperationLogStoreService', () => { vectorClock: { testClient: 3, otherClient: 1 }, }); - await service.appendWithVectorClockUpdate(op1, 'local'); + await service.appendWithVectorClockOverwrite(op1, 'local'); let clock = await service.getVectorClock(); expect(clock).toEqual({ testClient: 1 }); - await service.appendWithVectorClockUpdate(op2, 'local'); + await service.appendWithVectorClockOverwrite(op2, 'local'); clock = await service.getVectorClock(); expect(clock).toEqual({ testClient: 2 }); - await service.appendWithVectorClockUpdate(op3, 'local'); + await service.appendWithVectorClockOverwrite(op3, 'local'); clock = await service.getVectorClock(); expect(clock).toEqual({ testClient: 3, otherClient: 1 }); }); @@ -3495,7 +3495,7 @@ describe('OperationLogStoreService', () => { it('should set applicationStatus to pending for remote ops with pendingApply', async () => { const op = createTestOperation(); - await service.appendWithVectorClockUpdate(op, 'remote', { pendingApply: true }); + await service.appendWithVectorClockOverwrite(op, 'remote', { pendingApply: true }); const ops = await service.getOpsAfterSeq(0); expect(ops[0].applicationStatus).toBe('pending'); @@ -3504,7 +3504,7 @@ describe('OperationLogStoreService', () => { it('should set applicationStatus to applied for remote ops without pendingApply', async () => { const op = createTestOperation(); - await service.appendWithVectorClockUpdate(op, 'remote'); + await service.appendWithVectorClockOverwrite(op, 'remote'); const ops = await service.getOpsAfterSeq(0); expect(ops[0].applicationStatus).toBe('applied'); @@ -3514,7 +3514,7 @@ describe('OperationLogStoreService', () => { const beforeTime = Date.now(); const op = createTestOperation(); - await service.appendWithVectorClockUpdate(op, 'remote'); + await service.appendWithVectorClockOverwrite(op, 'remote'); const afterTime = Date.now(); const ops = await service.getOpsAfterSeq(0); @@ -3526,7 +3526,7 @@ describe('OperationLogStoreService', () => { it('should NOT set syncedAt for local ops', async () => { const op = createTestOperation(); - await service.appendWithVectorClockUpdate(op, 'local'); + await service.appendWithVectorClockOverwrite(op, 'local'); const ops = await service.getOpsAfterSeq(0); expect(ops[0].syncedAt).toBeUndefined(); @@ -3542,7 +3542,7 @@ describe('OperationLogStoreService', () => { // Append concurrently await Promise.all( - ops.map((op) => service.appendWithVectorClockUpdate(op, 'local')), + ops.map((op) => service.appendWithVectorClockOverwrite(op, 'local')), ); const storedOps = await service.getOpsAfterSeq(0); @@ -3587,6 +3587,42 @@ describe('OperationLogStoreService', () => { expect((await service.getLatestFullStateOpEntry())?.op.id).toBe(replacement.id); }); + // #8939: the caller-built clock may be stale (derived from a lagging + // in-memory cache); the replacement must be rebased onto the durable + // clock inside the transaction so the clock can never regress. + it('should rebase the replacement clock past a durable clock that advanced since it was built', async () => { + const staleRepair = createTestOperation({ + id: 'stale-repair', + opType: OpType.Repair, + entityType: 'ALL', + entityId: undefined, + }); + await service.append(staleRepair); + await service.setVectorClock({ testClient: 9, otherClient: 4 }); + const repairedState = { task: { ids: [], entities: {} } }; + const replacement = createTestOperation({ + id: 'replacement-repair', + opType: OpType.Repair, + entityType: 'ALL', + entityId: undefined, + payload: { appDataComplete: repairedState }, + vectorClock: { testClient: 2 }, + }); + + await service.replaceRejectedRepair({ + staleRepairOpId: staleRepair.id, + replacementOp: replacement, + repairedState, + }); + + const expectedClock = { testClient: 10, otherClient: 4 }; + expect(await service.getVectorClock()).toEqual(expectedClock); + expect((await service.getOpById(replacement.id))?.op.vectorClock).toEqual( + expectedClock, + ); + expect((await service.loadStateCache())?.vectorClock).toEqual(expectedClock); + }); + it('should abort without appending when the stale repair is missing', async () => { const replacement = createTestOperation({ id: 'replacement-repair', @@ -3630,7 +3666,7 @@ describe('OperationLogStoreService', () => { const op = createTestOperation({ vectorClock: { snapshotClient: 51 }, }); - await service.append(op); // Using append, not appendWithVectorClockUpdate + await service.append(op); // Using append, not appendWithVectorClockOverwrite // VectorClockService should fall back to computing from snapshot+ops const clock = await vectorClockService.getCurrentVectorClock(); @@ -3795,7 +3831,7 @@ describe('OperationLogStoreService', () => { vectorClock: { localClient: 5 }, }); - await service.appendWithVectorClockUpdate(localOp, 'local'); + await service.appendWithVectorClockOverwrite(localOp, 'local'); // Cache should be updated with the new clock const clock = await service.getVectorClock(); @@ -3811,7 +3847,7 @@ describe('OperationLogStoreService', () => { clientId: 'remoteClient', vectorClock: { remoteClient: 99, localClient: 15 }, }); - await service.appendWithVectorClockUpdate(remoteOp, 'remote'); + await service.appendWithVectorClockOverwrite(remoteOp, 'remote'); // Cache should NOT be updated - still returns local clock const clock = await service.getVectorClock(); @@ -3824,7 +3860,7 @@ describe('OperationLogStoreService', () => { entityId: 'task1', vectorClock: { localClient: 1 }, }); - await service.appendWithVectorClockUpdate(localOp1, 'local'); + await service.appendWithVectorClockOverwrite(localOp1, 'local'); // Remote op (should not affect cache) const remoteOp = createTestOperation({ @@ -3832,14 +3868,14 @@ describe('OperationLogStoreService', () => { clientId: 'remoteClient', vectorClock: { remoteClient: 50, localClient: 5 }, }); - await service.appendWithVectorClockUpdate(remoteOp, 'remote'); + await service.appendWithVectorClockOverwrite(remoteOp, 'remote'); // Local op 2 const localOp2 = createTestOperation({ entityId: 'task3', vectorClock: { localClient: 2 }, }); - await service.appendWithVectorClockUpdate(localOp2, 'local'); + await service.appendWithVectorClockOverwrite(localOp2, 'local'); // Cache should reflect the last LOCAL op's clock const clock = await service.getVectorClock(); diff --git a/src/app/op-log/persistence/operation-log-store.service.ts b/src/app/op-log/persistence/operation-log-store.service.ts index 2545dd4cdb..880525356c 100644 --- a/src/app/op-log/persistence/operation-log-store.service.ts +++ b/src/app/op-log/persistence/operation-log-store.service.ts @@ -311,6 +311,28 @@ interface OpLogDB extends DBSchema { type OpLogStoreName = (typeof STORE_NAMES)[keyof typeof STORE_NAMES]; +/** + * Rebases a local operation's proposed clock onto the durable clock read in + * the same transaction: entry-wise max, with this client's counter bumped + * past the durable value. Makes counter reuse/regression unrepresentable + * regardless of what the caller derived its proposed clock from (#8939). + */ +const rebaseLocalClockOnDurable = ( + durableClock: VectorClock, + proposedClock: VectorClock, + clientId: string, +): VectorClock => { + const merged: VectorClock = { ...durableClock }; + for (const [id, counter] of Object.entries(proposedClock)) { + merged[id] = Math.max(merged[id] ?? 0, counter); + } + merged[clientId] = Math.max( + (durableClock[clientId] ?? 0) + 1, + proposedClock[clientId] ?? 0, + ); + return merged; +}; + /** * Manages the persistence of operations and state snapshots in IndexedDB. * It uses a dedicated IndexedDB database ('SUP_OPS') to store: @@ -486,7 +508,7 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort( + STORE_NAMES.VECTOR_CLOCK, + SINGLETON_KEY, + ); + const rebasedClock = rebaseLocalClockOnDurable( + currentClockEntry?.clock ?? {}, + replacementOp.vectorClock, + replacementOp.clientId, + ); + const rebasedOp: Operation = { ...replacementOp, vectorClock: rebasedClock }; + + const replacementEntry = this._buildStoredEntry(rebasedOp, 'local'); const replacementSeq = await tx.add(STORE_NAMES.OPS, replacementEntry); await this._recordFullStateOpInTx(tx, replacementEntry.op, replacementSeq); await tx.put( STORE_NAMES.VECTOR_CLOCK, - { clock: replacementOp.vectorClock, lastUpdate: Date.now() }, + { clock: rebasedClock, lastUpdate: Date.now() }, SINGLETON_KEY, ); await tx.put(STORE_NAMES.STATE_CACHE, { id: SINGLETON_KEY, state: repairedState, lastAppliedOpSeq: replacementSeq, - vectorClock: replacementOp.vectorClock, + vectorClock: rebasedClock, compactedAt: Date.now(), - schemaVersion: replacementOp.schemaVersion, + schemaVersion: rebasedOp.schemaVersion, }); + committedClock = rebasedClock; return replacementSeq; }, ); - this._vectorClockCache = replacementOp.vectorClock; + this._vectorClockCache = committedClock ?? null; this._invalidateUnsyncedCache(); return seq; } diff --git a/src/app/op-log/sync/conflict-journal-hook.integration.spec.ts b/src/app/op-log/sync/conflict-journal-hook.integration.spec.ts index bdcc943b2e..ab0aca9955 100644 --- a/src/app/op-log/sync/conflict-journal-hook.integration.spec.ts +++ b/src/app/op-log/sync/conflict-journal-hook.integration.spec.ts @@ -82,7 +82,7 @@ describe('ConflictResolution → ConflictJournal hook (integration)', () => { mockOpLogStore = jasmine.createSpyObj('OperationLogStoreService', [ 'appendBatchSkipDuplicates', 'appendMixedSourceBatchSkipDuplicates', - 'appendWithVectorClockUpdate', + 'appendWithVectorClockOverwrite', 'markApplied', 'markRejected', 'markFailed', @@ -103,7 +103,7 @@ describe('ConflictResolution → ConflictJournal hook (integration)', () => { skippedCount: 0, })); mockOpLogStore.getUnsyncedByEntity.and.resolveTo(new Map()); - mockOpLogStore.appendWithVectorClockUpdate.and.resolveTo(undefined); + mockOpLogStore.appendWithVectorClockOverwrite.and.resolveTo(undefined); mockOpLogStore.markRejected.and.resolveTo(undefined); mockOpLogStore.appendBatchSkipDuplicates.and.callFake((ops: Operation[]) => Promise.resolve({ @@ -269,19 +269,19 @@ describe('ConflictResolution → ConflictJournal hook (integration)', () => { // Control run: journaling works normally. await service.autoResolveConflictsLWW([buildConflict()]); const controlRejected = mockOpLogStore.markRejected.calls.allArgs(); - const controlAppended = mockOpLogStore.appendWithVectorClockUpdate.calls + const controlAppended = mockOpLogStore.appendWithVectorClockOverwrite.calls .allArgs() .map(([o]) => (o as Operation).entityId); mockOpLogStore.markRejected.calls.reset(); - mockOpLogStore.appendWithVectorClockUpdate.calls.reset(); + mockOpLogStore.appendWithVectorClockOverwrite.calls.reset(); // Sabotaged run: force record() to reject — resolution must be unaffected. spyOn(journal, 'record').and.rejectWith(new Error('journal boom')); await service.autoResolveConflictsLWW([buildConflict()]); const sabotagedRejected = mockOpLogStore.markRejected.calls.allArgs(); - const sabotagedAppended = mockOpLogStore.appendWithVectorClockUpdate.calls + const sabotagedAppended = mockOpLogStore.appendWithVectorClockOverwrite.calls .allArgs() .map(([o]) => (o as Operation).entityId); diff --git a/src/app/op-log/sync/conflict-resolution.disjoint-merge.spec.ts b/src/app/op-log/sync/conflict-resolution.disjoint-merge.spec.ts index dbd66870c3..3fbcf3223f 100644 --- a/src/app/op-log/sync/conflict-resolution.disjoint-merge.spec.ts +++ b/src/app/op-log/sync/conflict-resolution.disjoint-merge.spec.ts @@ -95,7 +95,7 @@ describe('ConflictResolutionService — SPAP-14 disjoint-field merge', () => { mockOpLogStore = jasmine.createSpyObj('OperationLogStoreService', [ 'appendBatchSkipDuplicates', 'appendMixedSourceBatchSkipDuplicates', - 'appendWithVectorClockUpdate', + 'appendWithVectorClockOverwrite', 'markApplied', 'markRejected', 'markFailed', @@ -118,7 +118,7 @@ describe('ConflictResolutionService — SPAP-14 disjoint-field merge', () => { mockOpLogStore.getUnsyncedByEntity.and.resolveTo(new Map()); mockOpLogStore.markRejected.and.resolveTo(undefined); mockOpLogStore.markApplied.and.resolveTo(undefined); - mockOpLogStore.appendWithVectorClockUpdate.and.resolveTo(1); + mockOpLogStore.appendWithVectorClockOverwrite.and.resolveTo(1); mockOpLogStore.appendBatchSkipDuplicates.and.callFake((ops: Operation[]) => Promise.resolve({ seqs: ops.map((_, i) => i + 1), @@ -253,9 +253,9 @@ describe('ConflictResolutionService — SPAP-14 disjoint-field merge', () => { await service.autoResolveConflictsLWW([conflictOf([localOp], [remoteOp])]); - // appendWithVectorClockUpdate REPLACES the durable clock with the caller's + // appendWithVectorClockOverwrite REPLACES the durable clock with the caller's // clock (built only from the conflict's ops) — the batch rebases instead. - expect(mockOpLogStore.appendWithVectorClockUpdate).not.toHaveBeenCalled(); + expect(mockOpLogStore.appendWithVectorClockOverwrite).not.toHaveBeenCalled(); const batches = mockOpLogStore.appendMixedSourceBatchSkipDuplicates.calls.mostRecent().args[0]; const remoteBatch = batches.find((b) => b.source === 'remote'); @@ -1359,7 +1359,7 @@ describe('ConflictResolutionService — SPAP-14 disjoint-field merge', () => { const opLogStore = jasmine.createSpyObj('OperationLogStoreService', [ 'appendBatchSkipDuplicates', 'appendMixedSourceBatchSkipDuplicates', - 'appendWithVectorClockUpdate', + 'appendWithVectorClockOverwrite', 'markApplied', 'markRejected', 'markFailed', @@ -1383,7 +1383,7 @@ describe('ConflictResolutionService — SPAP-14 disjoint-field merge', () => { opLogStore.markRejected.and.resolveTo(undefined); opLogStore.markApplied.and.resolveTo(undefined); opLogStore.markFailed.and.resolveTo(undefined); - opLogStore.appendWithVectorClockUpdate.and.resolveTo(1); + opLogStore.appendWithVectorClockOverwrite.and.resolveTo(1); opLogStore.appendBatchSkipDuplicates.and.callFake((ops: Operation[]) => Promise.resolve({ seqs: ops.map((_, i) => i + 1), 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 e7d0dd52c9..7f21b00016 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 @@ -77,11 +77,11 @@ describe('regression #7700: operation-log lock reentry', () => { beforeEach(() => { opLogStoreSpy = jasmine.createSpyObj('OperationLogStoreService', [ - 'appendWithVectorClockUpdate', + 'appendWithVectorClockOverwrite', 'getCompactionCounter', 'clearVectorClockCache', ]); - opLogStoreSpy.appendWithVectorClockUpdate.and.resolveTo(1); + opLogStoreSpy.appendWithVectorClockOverwrite.and.resolveTo(1); opLogStoreSpy.getCompactionCounter.and.resolveTo(0); vectorClockSpy = jasmine.createSpyObj('VectorClockService', [ @@ -198,7 +198,7 @@ describe('regression #7700: operation-log lock reentry', () => { const elapsed = Date.now() - start; expect(order).toEqual(['outer-start', 'outer-end']); - expect(opLogStoreSpy.appendWithVectorClockUpdate).toHaveBeenCalledTimes(1); + expect(opLogStoreSpy.appendWithVectorClockOverwrite).toHaveBeenCalledTimes(1); // Should be near-instant — orders of magnitude under the 30s lock timeout. expect(elapsed).toBeLessThan(2000); }, 10000); @@ -230,7 +230,7 @@ describe('regression #7700: operation-log lock reentry', () => { ).toBeRejected(); // Action was NOT written — no silent persistence. - expect(opLogStoreSpy.appendWithVectorClockUpdate).not.toHaveBeenCalled(); + expect(opLogStoreSpy.appendWithVectorClockOverwrite).not.toHaveBeenCalled(); // The differentiating assertion: pre-loud-fail, writeOperation // swallowed the timeout and returned success → the retry loop // exited on attempt #1 and DEFERRED_ACTION_FAILED never fired. @@ -281,8 +281,8 @@ describe('regression #7700: operation-log lock reentry', () => { await effects.processDeferredActions({ callerHoldsOperationLogLock: true }); }); - expect(opLogStoreSpy.appendWithVectorClockUpdate).toHaveBeenCalledTimes(1); - const writtenOp = opLogStoreSpy.appendWithVectorClockUpdate.calls.mostRecent() + expect(opLogStoreSpy.appendWithVectorClockOverwrite).toHaveBeenCalledTimes(1); + const writtenOp = opLogStoreSpy.appendWithVectorClockOverwrite.calls.mostRecent() .args[0] as { vectorClock: Record }; // The merged remote entries must be present. diff --git a/src/app/op-log/sync/superseded-operation-resolver.service.spec.ts b/src/app/op-log/sync/superseded-operation-resolver.service.spec.ts index 3eddfba67a..dbf553af87 100644 --- a/src/app/op-log/sync/superseded-operation-resolver.service.spec.ts +++ b/src/app/op-log/sync/superseded-operation-resolver.service.spec.ts @@ -51,7 +51,7 @@ describe('SupersededOperationResolverService', () => { mockOpLogStore = jasmine.createSpyObj('OperationLogStoreService', [ 'markRejected', 'appendMixedSourceBatchSkipDuplicates', - 'appendWithVectorClockUpdate', + 'appendWithVectorClockOverwrite', ]); mockVectorClockService = jasmine.createSpyObj('VectorClockService', [ 'getCurrentVectorClock', @@ -73,13 +73,13 @@ describe('SupersededOperationResolverService', () => { // Default mocks mockVectorClockService.getCurrentVectorClock.and.returnValue(Promise.resolve({})); mockOpLogStore.markRejected.and.returnValue(Promise.resolve()); - mockOpLogStore.appendWithVectorClockUpdate.and.returnValue(Promise.resolve(1)); + mockOpLogStore.appendWithVectorClockOverwrite.and.returnValue(Promise.resolve(1)); mockOpLogStore.appendMixedSourceBatchSkipDuplicates.and.callFake(async (batches) => { const written: MixedSourceWrittenOperation[] = []; let seq = 1; for (const batch of batches) { for (const op of batch.ops) { - await mockOpLogStore.appendWithVectorClockUpdate(op, batch.source); + await mockOpLogStore.appendWithVectorClockOverwrite(op, batch.source); written.push({ seq: seq++, op, source: batch.source }); } } @@ -184,8 +184,8 @@ describe('SupersededOperationResolverService', () => { mockOpLogStore.markRejected.and.callFake(async () => { callOrder.push('markRejected'); }); - mockOpLogStore.appendWithVectorClockUpdate.and.callFake(async () => { - callOrder.push('appendWithVectorClockUpdate'); + mockOpLogStore.appendWithVectorClockOverwrite.and.callFake(async () => { + callOrder.push('appendWithVectorClockOverwrite'); return 1; }); @@ -194,7 +194,7 @@ describe('SupersededOperationResolverService', () => { // Verify write operations happen inside the lock expect(callOrder).toEqual([ 'lock-start', - 'appendWithVectorClockUpdate', + 'appendWithVectorClockOverwrite', 'markRejected', 'lock-end', ]); @@ -224,7 +224,7 @@ describe('SupersededOperationResolverService', () => { expect(result).toBe(0); expect(mockOpLogStore.markRejected).not.toHaveBeenCalled(); - expect(mockOpLogStore.appendWithVectorClockUpdate).not.toHaveBeenCalled(); + expect(mockOpLogStore.appendWithVectorClockOverwrite).not.toHaveBeenCalled(); expect(mockSnackService.open).not.toHaveBeenCalled(); }); @@ -251,7 +251,7 @@ describe('SupersededOperationResolverService', () => { expect(result).toBe(0); expect(mockOpLogStore.markRejected).not.toHaveBeenCalled(); - expect(mockOpLogStore.appendWithVectorClockUpdate).not.toHaveBeenCalled(); + expect(mockOpLogStore.appendWithVectorClockOverwrite).not.toHaveBeenCalled(); }); it('should create LWW Update op for a single superseded op', async () => { @@ -286,9 +286,9 @@ describe('SupersededOperationResolverService', () => { expect(result).toBe(1); expect(mockOpLogStore.markRejected).toHaveBeenCalledWith(['op-1']); - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledTimes(1); + expect(mockOpLogStore.appendWithVectorClockOverwrite).toHaveBeenCalledTimes(1); - const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() + const appendedOp = mockOpLogStore.appendWithVectorClockOverwrite.calls.first() .args[0] as Operation; expect(appendedOp.actionType).toBe('[TASK] LWW Update'); expect(appendedOp.opType).toBe(OpType.Update); @@ -371,7 +371,7 @@ describe('SupersededOperationResolverService', () => { expect(mockOpLogStore.appendMixedSourceBatchSkipDuplicates).toHaveBeenCalledTimes( 1, ); - const appendedOps = mockOpLogStore.appendWithVectorClockUpdate.calls + const appendedOps = mockOpLogStore.appendWithVectorClockOverwrite.calls .allArgs() .map(([op]) => op); expect( @@ -401,7 +401,7 @@ describe('SupersededOperationResolverService', () => { await service.resolveSupersededLocalOps([{ opId: 'op-1', op: supersededOp }]); - const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() + const appendedOp = mockOpLogStore.appendWithVectorClockOverwrite.calls.first() .args[0] as Operation; expect(appendedOp.entityIds).toBeUndefined(); }); @@ -430,7 +430,7 @@ describe('SupersededOperationResolverService', () => { await service.resolveSupersededLocalOps([{ opId: 'op-1', op: supersededOp }]); - const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() + const appendedOp = mockOpLogStore.appendWithVectorClockOverwrite.calls.first() .args[0] as Operation; // The footprint is derived from the authenticated payload and passed to // createLWWUpdateOp (which embeds it in both entityIds and the payload; @@ -512,9 +512,9 @@ describe('SupersededOperationResolverService', () => { expect(result).toBe(1); // Only ONE new op for all 3 superseded ops expect(mockOpLogStore.markRejected).toHaveBeenCalledWith(['op-1', 'op-2', 'op-3']); - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledTimes(1); + expect(mockOpLogStore.appendWithVectorClockOverwrite).toHaveBeenCalledTimes(1); - const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() + const appendedOp = mockOpLogStore.appendWithVectorClockOverwrite.calls.first() .args[0] as Operation; // Timestamp should be max of all superseded ops (2000) expect(appendedOp.timestamp).toBe(2000); @@ -551,7 +551,7 @@ describe('SupersededOperationResolverService', () => { expect(result).toBe(2); expect(mockOpLogStore.markRejected).toHaveBeenCalledWith(['op-1', 'op-2']); - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledTimes(2); + expect(mockOpLogStore.appendWithVectorClockOverwrite).toHaveBeenCalledTimes(2); }); it('should mark ops as rejected but not create new op when entity not found', async () => { @@ -570,7 +570,7 @@ describe('SupersededOperationResolverService', () => { expect(result).toBe(0); expect(mockOpLogStore.markRejected).toHaveBeenCalledWith(['op-1']); - expect(mockOpLogStore.appendWithVectorClockUpdate).not.toHaveBeenCalled(); + expect(mockOpLogStore.appendWithVectorClockOverwrite).not.toHaveBeenCalled(); // Should notify user that local changes were discarded expect(mockSnackService.open).toHaveBeenCalledOnceWith( jasmine.objectContaining({ @@ -597,7 +597,7 @@ describe('SupersededOperationResolverService', () => { snapshotVectorClock, ); - const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() + const appendedOp = mockOpLogStore.appendWithVectorClockOverwrite.calls.first() .args[0] as Operation; // Clock should include entries from global clock, snapshot clock, superseded op clock, and be incremented expect(appendedOp.vectorClock['clientX']).toBe(100); @@ -621,7 +621,7 @@ describe('SupersededOperationResolverService', () => { extraClocks, ); - const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() + const appendedOp = mockOpLogStore.appendWithVectorClockOverwrite.calls.first() .args[0] as Operation; // Should have max of all merged clocks expect(appendedOp.vectorClock['clientP']).toBe(25); // max(20, 25) @@ -646,7 +646,7 @@ describe('SupersededOperationResolverService', () => { await service.resolveSupersededLocalOps([{ opId: 'op-1', op: supersededOp }]); - const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() + const appendedOp = mockOpLogStore.appendWithVectorClockOverwrite.calls.first() .args[0] as Operation; expect(appendedOp.timestamp).toBe(oldTimestamp); // Verify it's NOT a recent timestamp @@ -669,7 +669,7 @@ describe('SupersededOperationResolverService', () => { await service.resolveSupersededLocalOps([{ opId: 'op-1', op: supersededOp }]); - const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() + const appendedOp = mockOpLogStore.appendWithVectorClockOverwrite.calls.first() .args[0] as Operation; // New clock should be incremented and dominate all known clocks expect(appendedOp.vectorClock['clientA']).toBeGreaterThanOrEqual(10); // max of 10 and 8 @@ -691,7 +691,7 @@ describe('SupersededOperationResolverService', () => { await service.resolveSupersededLocalOps([{ opId: 'op-1', op: supersededOp }]); - const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() + const appendedOp = mockOpLogStore.appendWithVectorClockOverwrite.calls.first() .args[0] as Operation; expect(appendedOp.actionType).toBe('[PROJECT] LWW Update'); }); @@ -758,7 +758,7 @@ describe('SupersededOperationResolverService', () => { expect(result).toBe(2); // Only 2 new ops (for existing entities) expect(mockOpLogStore.markRejected).toHaveBeenCalledWith(['op-1', 'op-2', 'op-3']); // All 3 rejected - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledTimes(2); + expect(mockOpLogStore.appendWithVectorClockOverwrite).toHaveBeenCalledTimes(2); }); it('should append ops with local source', async () => { @@ -772,7 +772,7 @@ describe('SupersededOperationResolverService', () => { await service.resolveSupersededLocalOps([{ opId: 'op-1', op: supersededOp }]); - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledWith( + expect(mockOpLogStore.appendWithVectorClockOverwrite).toHaveBeenCalledWith( jasmine.any(Object), 'local', ); @@ -792,7 +792,7 @@ describe('SupersededOperationResolverService', () => { { opId: 'op-2', op: supersededOp2 }, ]); - const calls = mockOpLogStore.appendWithVectorClockUpdate.calls.all(); + const calls = mockOpLogStore.appendWithVectorClockOverwrite.calls.all(); const op1 = calls[0].args[0] as Operation; const op2 = calls[1].args[0] as Operation; @@ -844,9 +844,9 @@ describe('SupersededOperationResolverService', () => { expect(result).toBe(1); expect(mockOpLogStore.markRejected).toHaveBeenCalledWith(['op-archive-1']); - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledTimes(1); + expect(mockOpLogStore.appendWithVectorClockOverwrite).toHaveBeenCalledTimes(1); - const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() + const appendedOp = mockOpLogStore.appendWithVectorClockOverwrite.calls.first() .args[0] as Operation; expect(appendedOp.actionType).toBe(ActionType.TASK_SHARED_MOVE_TO_ARCHIVE); expect(appendedOp.opType).toBe(OpType.Update); @@ -867,7 +867,7 @@ describe('SupersededOperationResolverService', () => { { opId: 'op-archive-1', op: archiveOp }, ]); - const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() + const appendedOp = mockOpLogStore.appendWithVectorClockOverwrite.calls.first() .args[0] as Operation; expect(appendedOp.payload).toEqual(archiveOp.payload); }); @@ -886,7 +886,7 @@ describe('SupersededOperationResolverService', () => { { opId: 'op-archive-1', op: archiveOp }, ]); - const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() + const appendedOp = mockOpLogStore.appendWithVectorClockOverwrite.calls.first() .args[0] as Operation; expect(appendedOp.entityId).toBe('task-1'); expect(appendedOp.entityIds).toEqual(['task-1', 'task-2', 'task-3']); @@ -908,7 +908,7 @@ describe('SupersededOperationResolverService', () => { { opId: 'op-archive-1', op: archiveOp }, ]); - const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() + const appendedOp = mockOpLogStore.appendWithVectorClockOverwrite.calls.first() .args[0] as Operation; expect(appendedOp.vectorClock['clientA']).toBeGreaterThanOrEqual(10); expect(appendedOp.vectorClock['clientB']).toBeGreaterThanOrEqual(5); @@ -930,7 +930,7 @@ describe('SupersededOperationResolverService', () => { { opId: 'op-archive-1', op: archiveOp }, ]); - const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() + const appendedOp = mockOpLogStore.appendWithVectorClockOverwrite.calls.first() .args[0] as Operation; expect(appendedOp.timestamp).toBe(1609459200000); }); @@ -981,9 +981,9 @@ describe('SupersededOperationResolverService', () => { 'op-archive', 'op-regular', ]); - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledTimes(2); + expect(mockOpLogStore.appendWithVectorClockOverwrite).toHaveBeenCalledTimes(2); - const calls = mockOpLogStore.appendWithVectorClockUpdate.calls.all(); + const calls = mockOpLogStore.appendWithVectorClockOverwrite.calls.all(); const ops = calls.map((c) => c.args[0] as Operation); const archiveResult = ops.find( @@ -1015,7 +1015,7 @@ describe('SupersededOperationResolverService', () => { { opId: 'op-archive-1', op: archiveOp }, ]); - const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() + const appendedOp = mockOpLogStore.appendWithVectorClockOverwrite.calls.first() .args[0] as Operation; expect(appendedOp.clientId).toBe(TEST_CLIENT_ID); expect(appendedOp.clientId).not.toBe('original-client'); @@ -1039,7 +1039,7 @@ describe('SupersededOperationResolverService', () => { snapshotVectorClock, ); - const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() + const appendedOp = mockOpLogStore.appendWithVectorClockOverwrite.calls.first() .args[0] as Operation; // Clock should include entries from snapshot, extra, archive op, and be incremented expect(appendedOp.vectorClock['snapshot']).toBe(5); @@ -1059,7 +1059,7 @@ describe('SupersededOperationResolverService', () => { { opId: 'op-archive-1', op: archiveOp }, ]); - const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() + const appendedOp = mockOpLogStore.appendWithVectorClockOverwrite.calls.first() .args[0] as Operation; expect(appendedOp.schemaVersion).toBe(CURRENT_SCHEMA_VERSION); }); @@ -1090,9 +1090,9 @@ describe('SupersededOperationResolverService', () => { 'op-archive-1', 'op-archive-2', ]); - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledTimes(2); + expect(mockOpLogStore.appendWithVectorClockOverwrite).toHaveBeenCalledTimes(2); - const calls = mockOpLogStore.appendWithVectorClockUpdate.calls.all(); + const calls = mockOpLogStore.appendWithVectorClockOverwrite.calls.all(); const ops = calls.map((c) => c.args[0] as Operation); expect(ops[0].entityIds).toEqual(['task-1']); @@ -1146,9 +1146,9 @@ describe('SupersededOperationResolverService', () => { expect(result).toBe(1); expect(mockOpLogStore.markRejected).toHaveBeenCalledWith(['op-1']); - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledTimes(1); + expect(mockOpLogStore.appendWithVectorClockOverwrite).toHaveBeenCalledTimes(1); - const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() + const appendedOp = mockOpLogStore.appendWithVectorClockOverwrite.calls.first() .args[0] as Operation; expect(appendedOp.opType).toBe(OpType.Delete); expect(appendedOp.actionType).toBe('[TASK] Delete Task'); @@ -1190,9 +1190,9 @@ describe('SupersededOperationResolverService', () => { expect(result).toBe(1); // Only ONE new op for both superseded DELETE ops expect(mockOpLogStore.markRejected).toHaveBeenCalledWith(['op-1', 'op-2']); - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledTimes(1); + expect(mockOpLogStore.appendWithVectorClockOverwrite).toHaveBeenCalledTimes(1); - const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() + const appendedOp = mockOpLogStore.appendWithVectorClockOverwrite.calls.first() .args[0] as Operation; expect(appendedOp.opType).toBe(OpType.Delete); // Timestamp should be max of all superseded ops (2000) @@ -1224,7 +1224,7 @@ describe('SupersededOperationResolverService', () => { { opId: 'op-1', op: supersededDeleteOp }, ]); - const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() + const appendedOp = mockOpLogStore.appendWithVectorClockOverwrite.calls.first() .args[0] as Operation; expect(appendedOp.actionType).toBe('[TASK] Delete Task'); expect(appendedOp.payload).toEqual(customPayload); @@ -1244,7 +1244,7 @@ describe('SupersededOperationResolverService', () => { { opId: 'op-1', op: supersededDeleteOp }, ]); - const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() + const appendedOp = mockOpLogStore.appendWithVectorClockOverwrite.calls.first() .args[0] as Operation; // New clock should dominate all known clocks expect(appendedOp.vectorClock['clientA']).toBeGreaterThanOrEqual(10); @@ -1281,9 +1281,9 @@ describe('SupersededOperationResolverService', () => { expect(result).toBe(2); // One DELETE op + one UPDATE op expect(mockOpLogStore.markRejected).toHaveBeenCalledWith(['op-1', 'op-2']); - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledTimes(2); + expect(mockOpLogStore.appendWithVectorClockOverwrite).toHaveBeenCalledTimes(2); - const calls = mockOpLogStore.appendWithVectorClockUpdate.calls.all(); + const calls = mockOpLogStore.appendWithVectorClockOverwrite.calls.all(); const ops = calls.map((c) => c.args[0] as Operation); const deleteOp = ops.find((op) => op.entityId === 'deleted-task'); @@ -1342,7 +1342,7 @@ describe('SupersededOperationResolverService', () => { await service.resolveSupersededLocalOps([{ opId: 'op-1', op: supersededOp }]); - const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() + const appendedOp = mockOpLogStore.appendWithVectorClockOverwrite.calls.first() .args[0] as Operation; // Client sends unpruned merged clock; server prunes after conflict detection expect(Object.keys(appendedOp.vectorClock).length).toBeGreaterThan( @@ -1377,7 +1377,7 @@ describe('SupersededOperationResolverService', () => { await service.resolveSupersededLocalOps([{ opId: 'op-archive', op: archiveOp }]); - const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() + const appendedOp = mockOpLogStore.appendWithVectorClockOverwrite.calls.first() .args[0] as Operation; // Client sends unpruned merged clock; server prunes after conflict detection expect(Object.keys(appendedOp.vectorClock).length).toBeGreaterThan( @@ -1408,7 +1408,7 @@ describe('SupersededOperationResolverService', () => { await service.resolveSupersededLocalOps([{ opId: 'op-delete', op: deleteOp }]); - const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() + const appendedOp = mockOpLogStore.appendWithVectorClockOverwrite.calls.first() .args[0] as Operation; // Client sends unpruned merged clock; server prunes after conflict detection expect(Object.keys(appendedOp.vectorClock).length).toBeGreaterThan( @@ -1436,7 +1436,7 @@ describe('SupersededOperationResolverService', () => { await service.resolveSupersededLocalOps([{ opId: 'op-1', op: supersededOp }]); - const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() + const appendedOp = mockOpLogStore.appendWithVectorClockOverwrite.calls.first() .args[0] as Operation; expect(appendedOp.vectorClock[TEST_CLIENT_ID]).toBeDefined(); // All keys preserved — no client-side pruning (server handles it) @@ -1467,7 +1467,7 @@ describe('SupersededOperationResolverService', () => { await service.resolveSupersededLocalOps([{ opId: 'op-1', op: supersededOp }]); - const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() + const appendedOp = mockOpLogStore.appendWithVectorClockOverwrite.calls.first() .args[0] as Operation; expect(appendedOp.vectorClock[protectedId]).toBeDefined(); expect(appendedOp.vectorClock[TEST_CLIENT_ID]).toBeDefined(); @@ -1510,7 +1510,7 @@ describe('SupersededOperationResolverService', () => { { opId: 'op-1', op: supersededOp, existingClock }, ]); - const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() + const appendedOp = mockOpLogStore.appendWithVectorClockOverwrite.calls.first() .args[0] as Operation; // No client-side pruning — all keys preserved including server entity client expect(appendedOp.vectorClock['serverEntityClient']).toBe(7); @@ -1559,7 +1559,7 @@ describe('SupersededOperationResolverService', () => { { opId: 'op-archive', op: archiveOp, existingClock }, ]); - const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() + const appendedOp = mockOpLogStore.appendWithVectorClockOverwrite.calls.first() .args[0] as Operation; // No client-side pruning — all keys preserved including server entity client expect(appendedOp.vectorClock['serverEntityClient']).toBe(7); diff --git a/src/app/op-log/testing/integration/post-sync-validation.integration.spec.ts b/src/app/op-log/testing/integration/post-sync-validation.integration.spec.ts index c363f76661..196bac8ade 100644 --- a/src/app/op-log/testing/integration/post-sync-validation.integration.spec.ts +++ b/src/app/op-log/testing/integration/post-sync-validation.integration.spec.ts @@ -50,7 +50,7 @@ describe('Post-sync validation latch (#7330) — integration', () => { 'markApplied', 'getOpById', 'mergeRemoteOpClocks', - 'appendWithVectorClockUpdate', + 'appendWithVectorClockOverwrite', 'markFailed', ]); opLogStoreSpy.getUnsynced.and.resolveTo([]); 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 5949ce5f27..fb87d9c34a 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 @@ -101,7 +101,7 @@ describe('done task operation replay', () => { const actions$: Observable = of(action); const mockOpLogStore = jasmine.createSpyObj( 'OperationLogStoreService', - ['appendWithVectorClockUpdate', 'getCompactionCounter', 'clearVectorClockCache'], + ['appendWithVectorClockOverwrite', 'getCompactionCounter', 'clearVectorClockCache'], ); const mockLockService = jasmine.createSpyObj('LockService', ['request']); const mockVectorClockService = jasmine.createSpyObj( @@ -132,7 +132,7 @@ describe('done task operation replay', () => { mockLockService.request.and.callFake(async (_name: string, fn: () => Promise) => fn(), ); - mockOpLogStore.appendWithVectorClockUpdate.and.returnValue(Promise.resolve(1)); + mockOpLogStore.appendWithVectorClockOverwrite.and.returnValue(Promise.resolve(1)); mockOpLogStore.getCompactionCounter.and.returnValue(Promise.resolve(0)); mockVectorClockService.getCurrentVectorClock.and.returnValue( Promise.resolve({ clientA: 1 }), @@ -166,7 +166,7 @@ describe('done task operation replay', () => { }); }); - return mockOpLogStore.appendWithVectorClockUpdate.calls.mostRecent() + return mockOpLogStore.appendWithVectorClockOverwrite.calls.mostRecent() .args[0] as Operation; }; diff --git a/src/app/op-log/testing/integration/vector-clock-sync.integration.spec.ts b/src/app/op-log/testing/integration/vector-clock-sync.integration.spec.ts index 99a7bfb77a..d81b061ca4 100644 --- a/src/app/op-log/testing/integration/vector-clock-sync.integration.spec.ts +++ b/src/app/op-log/testing/integration/vector-clock-sync.integration.spec.ts @@ -69,7 +69,7 @@ describe('Vector Clock Sync Integration', () => { vectorClock: { clientA: 1 }, }); - await opLogStore.appendWithVectorClockUpdate(op, 'local'); + await opLogStore.appendWithVectorClockOverwrite(op, 'local'); // Both should be written const ops = await opLogStore.getOpsAfterSeq(0); @@ -86,7 +86,7 @@ describe('Vector Clock Sync Integration', () => { entityId: `task-${i}`, vectorClock: { clientA: i }, }); - await opLogStore.appendWithVectorClockUpdate(op, 'local'); + await opLogStore.appendWithVectorClockOverwrite(op, 'local'); } const ops = await opLogStore.getOpsAfterSeq(0); @@ -103,7 +103,7 @@ describe('Vector Clock Sync Integration', () => { entityId: 'task-1', vectorClock: { clientA: 1 }, }); - await opLogStore.appendWithVectorClockUpdate(opA, 'local'); + await opLogStore.appendWithVectorClockOverwrite(opA, 'local'); // Client A receives remote from B and creates new operation const opA2 = createTestOperation({ @@ -111,7 +111,7 @@ describe('Vector Clock Sync Integration', () => { entityId: 'task-2', vectorClock: { clientA: 2, clientB: 3 }, }); - await opLogStore.appendWithVectorClockUpdate(opA2, 'local'); + await opLogStore.appendWithVectorClockOverwrite(opA2, 'local'); const clock = await opLogStore.getVectorClock(); expect(clock).toEqual({ clientA: 2, clientB: 3 }); @@ -168,7 +168,7 @@ describe('Vector Clock Sync Integration', () => { compactedAt: Date.now(), }); - // Ops after snapshot (using regular append, not appendWithVectorClockUpdate) + // Ops after snapshot (using regular append, not appendWithVectorClockOverwrite) const op1 = createTestOperation({ vectorClock: { clientA: 11, clientB: 5 }, }); @@ -200,7 +200,7 @@ describe('Vector Clock Sync Integration', () => { clientId: 'remoteClient', vectorClock: { remoteClient: 50 }, }); - await opLogStore.appendWithVectorClockUpdate(remoteOp, 'remote'); + await opLogStore.appendWithVectorClockOverwrite(remoteOp, 'remote'); // Local clock should be unchanged const clock = await opLogStore.getVectorClock(); @@ -214,7 +214,7 @@ describe('Vector Clock Sync Integration', () => { entityId: 'task-local', vectorClock: { localClient: 1 }, }); - await opLogStore.appendWithVectorClockUpdate(localOp, 'local'); + await opLogStore.appendWithVectorClockOverwrite(localOp, 'local'); // Remote ops should be stored but not affect clock for (let i = 1; i <= 3; i++) { @@ -223,7 +223,7 @@ describe('Vector Clock Sync Integration', () => { entityId: `task-remote-${i}`, vectorClock: { remoteClient: i, localClient: 1 }, }); - await opLogStore.appendWithVectorClockUpdate(remoteOp, 'remote'); + await opLogStore.appendWithVectorClockOverwrite(remoteOp, 'remote'); } // All ops should be stored @@ -280,7 +280,7 @@ describe('Vector Clock Sync Integration', () => { entityId: `task-${i}`, vectorClock: { localClient: i }, }); - await opLogStore.appendWithVectorClockUpdate(op, 'local'); + await opLogStore.appendWithVectorClockOverwrite(op, 'local'); } // 2. Read clock for sync (simulates _syncVectorClockToPfapi reading) @@ -323,7 +323,7 @@ describe('Vector Clock Sync Integration', () => { clientId: 'localClient', vectorClock: { remoteClient: 10, localClient: 1 }, }); - await opLogStore.appendWithVectorClockUpdate(localOp, 'local'); + await opLogStore.appendWithVectorClockOverwrite(localOp, 'local'); // 3. Clock should now include both clients const clock = await opLogStore.getVectorClock(); diff --git a/src/app/op-log/validation/repair-operation.clock-derivation.integration.spec.ts b/src/app/op-log/validation/repair-operation.clock-derivation.integration.spec.ts new file mode 100644 index 0000000000..1ea267fa86 --- /dev/null +++ b/src/app/op-log/validation/repair-operation.clock-derivation.integration.spec.ts @@ -0,0 +1,118 @@ +import { TestBed } from '@angular/core/testing'; +import { RepairOperationService } from './repair-operation.service'; +import { OperationLogStoreService } from '../persistence/operation-log-store.service'; +import { VectorClockService } from '../sync/vector-clock.service'; +import { CLIENT_ID_PROVIDER, ClientIdProvider } from '../util/client-id.provider'; +import { OpLogDbAdapter } from '../persistence/op-log-db-adapter'; +import { SINGLETON_KEY, STORE_NAMES } from '../persistence/db-keys.const'; +import { StateSnapshotService } from '../backup/state-snapshot.service'; +import { SnackService } from '../../core/snack/snack.service'; +import { TranslateService } from '@ngx-translate/core'; +import { RepairSummary } from '../core/operation.types'; + +/** + * Regression tests for #8939: the REPAIR clock must be derived from the + * DURABLE vector clock at write time, not from the per-tab in-memory cache. + * + * The in-memory clock cache can be stale when another tab advanced the durable + * clock (Web Locks serialize writers across tabs, but each tab keeps its own + * cache). Before the fix, createRepairOperation incremented the stale cached + * clock and OVERWROTE the durable clock with it — regressing this client's + * counter and making already-used counters reusable by the next capture. + */ +describe('RepairOperationService clock derivation (#8939)', () => { + let service: RepairOperationService; + let opLogStore: OperationLogStoreService; + + const mockClientIdProvider: ClientIdProvider = { + loadClientId: () => Promise.resolve('testClient'), + getOrGenerateClientId: () => Promise.resolve('testClient'), + clearCache: () => {}, + }; + + const repairedState = { + task: { entities: {}, ids: [] }, + project: { entities: {}, ids: [] }, + }; + + const repairSummary: RepairSummary = { + entityStateFixed: 1, + orphanedEntitiesRestored: 0, + invalidReferencesRemoved: 0, + relationshipsFixed: 0, + structureRepaired: 0, + typeErrorsFixed: 0, + }; + + beforeEach(async () => { + TestBed.configureTestingModule({ + providers: [ + RepairOperationService, + OperationLogStoreService, + VectorClockService, + { provide: CLIENT_ID_PROVIDER, useValue: mockClientIdProvider }, + { + provide: TranslateService, + useValue: jasmine.createSpyObj('TranslateService', ['instant']), + }, + { + provide: StateSnapshotService, + useValue: jasmine.createSpyObj('StateSnapshotService', [ + 'getStateSnapshotAsync', + ]), + }, + { + provide: SnackService, + useValue: jasmine.createSpyObj('SnackService', ['open']), + }, + ], + }); + + service = TestBed.inject(RepairOperationService); + opLogStore = TestBed.inject(OperationLogStoreService); + await opLogStore.init(); + await opLogStore._clearAllDataForTesting(); + }); + + /** Writes the durable clock directly, bypassing this instance's cache — as another tab would. */ + const advanceDurableClockBehindCache = async ( + clock: Record, + ): Promise => { + const adapter = (opLogStore as unknown as { _adapter: OpLogDbAdapter })._adapter; + await adapter.transaction([STORE_NAMES.VECTOR_CLOCK], 'readwrite', async (tx) => { + await tx.put( + STORE_NAMES.VECTOR_CLOCK, + { clock, lastUpdate: Date.now() }, + SINGLETON_KEY, + ); + }); + }; + + it('derives the repair clock from the durable clock even when the in-memory cache is stale', async () => { + // Durable clock + this tab's cache both at {testClient: 3}. + await opLogStore.setVectorClock({ testClient: 3 }); + + // Another tab advances the durable clock; our cache still says 3. + await advanceDurableClockBehindCache({ testClient: 9, otherTab: 2 }); + + await service.createRepairOperation(repairedState, repairSummary, 'testClient'); + + opLogStore.clearVectorClockCache(); + const durable = await opLogStore.getVectorClock(); + // Pre-fix: repair derived {testClient: 4} from the stale cache and + // overwrote the durable clock with it — regressing 9 → 4, so the next + // capture would reuse counters 5–9 already shipped by the other tab. + expect(durable).toEqual({ testClient: 10, otherTab: 2 }); + + // The written REPAIR op must carry the rebased clock (it is what uploads). + const ops = await opLogStore.getOpsAfterSeq(0); + expect(ops.length).toBe(1); + expect(ops[0].op.vectorClock).toEqual({ testClient: 10, otherTab: 2 }); + + // The state cache must record the clock actually written, not the stale one. + expect((await opLogStore.loadStateCache())?.vectorClock).toEqual({ + testClient: 10, + otherTab: 2, + }); + }); +}); diff --git a/src/app/op-log/validation/repair-operation.service.spec.ts b/src/app/op-log/validation/repair-operation.service.spec.ts index e162423b8e..dd19dbced5 100644 --- a/src/app/op-log/validation/repair-operation.service.spec.ts +++ b/src/app/op-log/validation/repair-operation.service.spec.ts @@ -1,9 +1,12 @@ import { TestBed } from '@angular/core/testing'; import { RepairOperationService } from './repair-operation.service'; -import { OperationLogStoreService } from '../persistence/operation-log-store.service'; +import { + MixedSourceOperationBatch, + OperationLogStoreService, +} from '../persistence/operation-log-store.service'; import { LockService } from '../sync/lock.service'; import { VectorClockService } from '../sync/vector-clock.service'; -import { ActionType, RepairSummary, OpType } from '../core/operation.types'; +import { ActionType, Operation, OpType, RepairSummary } from '../core/operation.types'; import { CURRENT_SCHEMA_VERSION } from '../persistence/schema-migration.service'; import { TranslateService } from '@ngx-translate/core'; import { RepairSyncContextService } from './repair-sync-context.service'; @@ -40,9 +43,26 @@ describe('RepairOperationService', () => { ...overrides, }); + /** The operation passed to the mixed-source batch in the most recent call. */ + const getAppendedOp = (): Operation => + mockOpLogStore.appendMixedSourceBatchSkipDuplicates.calls.mostRecent().args[0][0] + .ops[0]; + + /** Makes the batch mock echo its input ops back as written, with the given seq. */ + const mockBatchAppendWithSeq = (seq: number): void => { + mockOpLogStore.appendMixedSourceBatchSkipDuplicates.and.callFake( + async (batches: readonly MixedSourceOperationBatch[]) => ({ + written: batches.flatMap((batch) => + batch.ops.map((op) => ({ seq, op, source: batch.source })), + ), + skippedCount: 0, + }), + ); + }; + beforeEach(() => { mockOpLogStore = jasmine.createSpyObj('OperationLogStoreService', [ - 'appendWithVectorClockUpdate', + 'appendMixedSourceBatchSkipDuplicates', 'replaceRejectedRepair', 'saveStateCache', ]); @@ -60,8 +80,7 @@ describe('RepairOperationService', () => { mockLockService.request.and.callFake(async (_name: string, fn: () => Promise) => fn(), ); - // appendWithVectorClockUpdate returns the sequence number - mockOpLogStore.appendWithVectorClockUpdate.and.returnValue(Promise.resolve(100)); + mockBatchAppendWithSeq(100); mockOpLogStore.saveStateCache.and.returnValue(Promise.resolve()); mockOpLogStore.replaceRejectedRepair.and.resolveTo(101); mockVectorClockService.getCurrentVectorClock.and.returnValue( @@ -146,26 +165,30 @@ describe('RepairOperationService', () => { await service.createRepairOperation(mockRepairedState, summary, 'test-client'); - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledWith( - jasmine.objectContaining({ - actionType: '[Repair] Auto Repair' as ActionType, - opType: OpType.Repair, - entityType: 'ALL', - clientId: 'test-client', - schemaVersion: CURRENT_SCHEMA_VERSION, - }), - 'local', - ); + expect(mockOpLogStore.appendMixedSourceBatchSkipDuplicates).toHaveBeenCalledWith([ + { + ops: [ + jasmine.objectContaining({ + actionType: '[Repair] Auto Repair' as ActionType, + opType: OpType.Repair, + entityType: 'ALL', + clientId: 'test-client', + schemaVersion: CURRENT_SCHEMA_VERSION, + }), + ], + source: 'local', + }, + ]); }); - it('should use appendWithVectorClockUpdate to ensure vector clock is updated atomically', async () => { + it('should use the mixed-source batch so the clock is rebased on the durable clock', async () => { const summary = createRepairSummary({ entityStateFixed: 1 }); await service.createRepairOperation(mockRepairedState, summary, 'test-client'); - // Verify appendWithVectorClockUpdate is called (not plain append) - // This ensures the vector clock store is updated atomically with the operation - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalled(); + // The batch's in-transaction rebase is what prevents a stale in-memory + // clock cache from regressing the durable clock (#8939). + expect(mockOpLogStore.appendMixedSourceBatchSkipDuplicates).toHaveBeenCalled(); }); it('should include repaired state and summary in payload', async () => { @@ -173,10 +196,7 @@ describe('RepairOperationService', () => { await service.createRepairOperation(mockRepairedState, summary, 'test-client'); - const appendCall = mockOpLogStore.appendWithVectorClockUpdate.calls.mostRecent(); - const operation = appendCall.args[0]; - - expect(operation.payload).toEqual({ + expect(getAppendedOp().payload).toEqual({ appDataComplete: mockRepairedState, repairSummary: summary, }); @@ -189,9 +209,7 @@ describe('RepairOperationService', () => { service.createRepairOperation(mockRepairedState, summary, 'test-client'), ); - const operation = - mockOpLogStore.appendWithVectorClockUpdate.calls.mostRecent().args[0]; - expect(operation.payload).toEqual( + expect(getAppendedOp().payload).toEqual( jasmine.objectContaining({ repairBaseServerSeq: 17 }), ); }); @@ -207,7 +225,7 @@ describe('RepairOperationService', () => { ); }); - it('should increment vector clock for the client', async () => { + it('should propose an incremented vector clock for the client', async () => { mockVectorClockService.getCurrentVectorClock.and.returnValue( Promise.resolve({ clientA: 10, clientB: 5 }), ); @@ -215,9 +233,7 @@ describe('RepairOperationService', () => { await service.createRepairOperation(mockRepairedState, summary, 'test-client'); - const appendCall = mockOpLogStore.appendWithVectorClockUpdate.calls.mostRecent(); - const operation = appendCall.args[0]; - + const operation = getAppendedOp(); // Should have incremented the test-client entry expect(operation.vectorClock['test-client']).toBe(1); // Should preserve existing entries @@ -225,9 +241,21 @@ describe('RepairOperationService', () => { expect(operation.vectorClock['clientB']).toBe(5); }); - it('should save state cache after appending operation', async () => { - // appendWithVectorClockUpdate returns the sequence number directly - mockOpLogStore.appendWithVectorClockUpdate.and.returnValue(Promise.resolve(42)); + it('should save state cache with the seq and clock that were actually written', async () => { + const rebasedClock = { rebasedClient: 9, otherClient: 3 }; + mockOpLogStore.appendMixedSourceBatchSkipDuplicates.and.callFake( + async (batches: readonly MixedSourceOperationBatch[]) => ({ + // Simulate the in-transaction rebase changing the proposed clock. + written: batches.flatMap((batch) => + batch.ops.map((op) => ({ + seq: 42, + op: { ...op, vectorClock: rebasedClock }, + source: batch.source, + })), + ), + skippedCount: 0, + }), + ); const summary = createRepairSummary(); await service.createRepairOperation(mockRepairedState, summary, 'test-client'); @@ -236,14 +264,14 @@ describe('RepairOperationService', () => { jasmine.objectContaining({ state: mockRepairedState, lastAppliedOpSeq: 42, + vectorClock: rebasedClock, schemaVersion: CURRENT_SCHEMA_VERSION, }), ); }); it('should return the sequence number of the created operation', async () => { - // appendWithVectorClockUpdate returns the sequence number directly - mockOpLogStore.appendWithVectorClockUpdate.and.returnValue(Promise.resolve(77)); + mockBatchAppendWithSeq(77); const summary = createRepairSummary(); const seq = await service.createRepairOperation( @@ -296,7 +324,7 @@ describe('RepairOperationService', () => { await service.createRepairOperation(mockRepairedState, summary, 'test-client'); - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalled(); + expect(mockOpLogStore.appendMixedSourceBatchSkipDuplicates).toHaveBeenCalled(); // translateService.instant + alert() are only reached by the blocking path. expect(mockTranslateService.instant).not.toHaveBeenCalled(); expect(alertSpy).not.toHaveBeenCalled(); @@ -330,14 +358,12 @@ describe('RepairOperationService', () => { const summary = createRepairSummary(); await service.createRepairOperation(mockRepairedState, summary, 'test-client'); - const firstCall = mockOpLogStore.appendWithVectorClockUpdate.calls.mostRecent(); - const firstId = firstCall.args[0].id; + const firstId = getAppendedOp().id; - mockOpLogStore.appendWithVectorClockUpdate.calls.reset(); + mockOpLogStore.appendMixedSourceBatchSkipDuplicates.calls.reset(); await service.createRepairOperation(mockRepairedState, summary, 'test-client'); - const secondCall = mockOpLogStore.appendWithVectorClockUpdate.calls.mostRecent(); - const secondId = secondCall.args[0].id; + const secondId = getAppendedOp().id; expect(firstId).not.toBe(secondId); }); @@ -349,8 +375,7 @@ describe('RepairOperationService', () => { await service.createRepairOperation(mockRepairedState, summary, 'test-client'); const afterTime = Date.now(); - const appendCall = mockOpLogStore.appendWithVectorClockUpdate.calls.mostRecent(); - const operation = appendCall.args[0]; + const operation = getAppendedOp(); expect(operation.timestamp).toBeGreaterThanOrEqual(beforeTime); expect(operation.timestamp).toBeLessThanOrEqual(afterTime); diff --git a/src/app/op-log/validation/repair-operation.service.ts b/src/app/op-log/validation/repair-operation.service.ts index 133c346a0c..622cfadf42 100644 --- a/src/app/op-log/validation/repair-operation.service.ts +++ b/src/app/op-log/validation/repair-operation.service.ts @@ -8,7 +8,7 @@ import { ActionType, } from '../core/operation.types'; import { uuidv7 } from '../../util/uuid-v7'; -import { incrementVectorClock, limitVectorClockSize } from '../../core/util/vector-clock'; +import { incrementVectorClock } from '../../core/util/vector-clock'; import { LockService } from '../sync/lock.service'; import { T } from '../../t.const'; import { OpLog } from '../../core/log'; @@ -84,15 +84,25 @@ export class RepairOperationService { this.repairSyncContext.baseServerSeq, ); - // 1. Append REPAIR operation to log and update vector clock atomically - seq = await this.opLogStore.appendWithVectorClockUpdate(op, 'local'); + // 1. Append via the mixed-source batch: its in-transaction rebase derives + // the final clock from the durable clock, so a stale in-memory clock + // cache (e.g. another tab advanced the clock since our last read) can + // never regress the durable clock or reuse counters (#8939). + const { written } = await this.opLogStore.appendMixedSourceBatchSkipDuplicates([ + { ops: [op], source: 'local' }, + ]); + const writtenOp = written[0]; + if (!writtenOp) { + throw new Error('REPAIR operation was not appended'); + } + seq = writtenOp.seq; - // 2. Save state cache with repaired state for fast hydration - // Note: vector clock is already updated in step 1, so we omit it here + // 2. Save state cache with repaired state for fast hydration. + // Use the rebased clock that was actually written, not the proposed one. await this.opLogStore.saveStateCache({ state: repairedState, lastAppliedOpSeq: seq, - vectorClock: op.vectorClock, + vectorClock: writtenOp.op.vectorClock, compactedAt: Date.now(), schemaVersion: CURRENT_SCHEMA_VERSION, }); @@ -160,11 +170,13 @@ export class RepairOperationService { repairSummary, ...(repairBaseServerSeq !== undefined ? { repairBaseServerSeq } : {}), }; + // Proposed clock only — both write paths rebase it onto the durable clock + // in-transaction (#8939). No client-side pruning: like capture ops, REPAIR + // ops ship the full clock; the server prunes AFTER conflict detection, and + // pruning here would drop client IDs the server still tracks, causing + // false CONCURRENT comparisons. const currentClock = await this.vectorClockService.getCurrentVectorClock(); - const newClock = limitVectorClockSize( - incrementVectorClock(currentClock, clientId), - clientId, - ); + const newClock = incrementVectorClock(currentClock, clientId); return { id: uuidv7(), diff --git a/src/app/op-log/validation/sync-repair-non-blocking.integration.spec.ts b/src/app/op-log/validation/sync-repair-non-blocking.integration.spec.ts index c10396776d..d570acdebe 100644 --- a/src/app/op-log/validation/sync-repair-non-blocking.integration.spec.ts +++ b/src/app/op-log/validation/sync-repair-non-blocking.integration.spec.ts @@ -3,7 +3,10 @@ import { provideMockStore } from '@ngrx/store/testing'; import { TranslateService } from '@ngx-translate/core'; import { ValidateStateService } from './validate-state.service'; import { StateSnapshotService } from '../backup/state-snapshot.service'; -import { OperationLogStoreService } from '../persistence/operation-log-store.service'; +import { + MixedSourceOperationBatch, + OperationLogStoreService, +} from '../persistence/operation-log-store.service'; import { LockService } from '../sync/lock.service'; import { VectorClockService } from '../sync/vector-clock.service'; import { CLIENT_ID_PROVIDER } from '../util/client-id.provider'; @@ -92,10 +95,17 @@ describe('#9026 background-sync repair is non-blocking (stitched)', () => { ); mockOpLogStore = jasmine.createSpyObj('OperationLogStoreService', [ - 'appendWithVectorClockUpdate', + 'appendMixedSourceBatchSkipDuplicates', 'saveStateCache', ]); - mockOpLogStore.appendWithVectorClockUpdate.and.resolveTo(1); + mockOpLogStore.appendMixedSourceBatchSkipDuplicates.and.callFake( + async (batches: readonly MixedSourceOperationBatch[]) => ({ + written: batches.flatMap((batch) => + batch.ops.map((op) => ({ seq: 1, op, source: batch.source })), + ), + skippedCount: 0, + }), + ); mockOpLogStore.saveStateCache.and.resolveTo(); const mockLockService = jasmine.createSpyObj('LockService', ['request']); @@ -174,7 +184,7 @@ describe('#9026 background-sync repair is non-blocking (stitched)', () => { expect(confirmSpy).not.toHaveBeenCalled(); expect(alertSpy).not.toHaveBeenCalled(); // A REPAIR op was actually created (real createRepairOperation ran). - expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalled(); + expect(mockOpLogStore.appendMixedSourceBatchSkipDuplicates).toHaveBeenCalled(); // ...and the non-blocking awareness snack was surfaced. expect(mockSnackService.open).toHaveBeenCalledWith( jasmine.objectContaining({ msg: T.F.SYNC.D_DATA_REPAIRED.MSG }),