mirror of
https://github.com/johannesjo/super-productivity.git
synced 2026-07-17 16:37:43 +00:00
The REPAIR paths built their vector clock from the per-tab in-memory clock cache and then REPLACED the durable clock with it. A stale cache (another tab advanced the clock) regressed the durable clock, letting subsequent captures reuse counters already shipped — silently corrupting cross-device dominance comparisons. - createRepairOperation: route through appendMixedSourceBatchSkipDuplicates; the in-transaction rebase makes regression unrepresentable. State cache stores the clock actually written. - replaceRejectedRepair: rebase the replacement clock onto the durable clock inside its transaction (shared rebaseLocalClockOnDurable helper). - Drop client-side clock pruning from repair op building: inert under the rebase, and it dropped client IDs the server still tracks (false CONCURRENT). Server prunes after conflict detection. - Rename appendWithVectorClockUpdate -> appendWithVectorClockOverwrite and document the derivation invariant; the capture path (its only remaining production caller) already satisfies it.
This commit is contained in:
Johannes Millan
GitHub
parent
5fcb1cb2d7
commit
d1ff7963d0
21 changed files with 489 additions and 248 deletions
|
|
@ -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).
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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.**
|
||||
|
||||
|
|
|
|||
|
|
@ -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',
|
||||
|
|
|
|||
|
|
@ -81,11 +81,11 @@ describe('regression #8306: persistOperation$ stream survives write failures', (
|
|||
actions$ = new ReplaySubject<Action>(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');
|
||||
|
||||
|
|
|
|||
|
|
@ -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<number>((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(
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
|
|
|
|||
|
|
@ -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<O
|
|||
/**
|
||||
* Builds a StoredOperationLogEntry (minus auto-incremented seq) from an
|
||||
* Operation, encoding it to compact format. Shared by append/appendBatch/
|
||||
* appendBatchSkipDuplicates/appendWithVectorClockUpdate.
|
||||
* appendBatchSkipDuplicates/appendWithVectorClockOverwrite.
|
||||
*/
|
||||
private _buildStoredEntry(
|
||||
op: Operation,
|
||||
|
|
@ -1118,13 +1140,10 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort<O
|
|||
'Cannot append a local operation for a non-current client ID.',
|
||||
);
|
||||
}
|
||||
const mergedClock: VectorClock = { ...runningClock };
|
||||
for (const [clientId, counter] of Object.entries(proposedOp.vectorClock)) {
|
||||
mergedClock[clientId] = Math.max(mergedClock[clientId] ?? 0, counter);
|
||||
}
|
||||
mergedClock[currentClientId] = Math.max(
|
||||
(runningClock[currentClientId] ?? 0) + 1,
|
||||
proposedOp.vectorClock[currentClientId] ?? 0,
|
||||
const mergedClock = rebaseLocalClockOnDurable(
|
||||
runningClock,
|
||||
proposedOp.vectorClock,
|
||||
currentClientId,
|
||||
);
|
||||
runningClock = mergedClock;
|
||||
op = { ...proposedOp, vectorClock: mergedClock };
|
||||
|
|
@ -2607,23 +2626,29 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort<O
|
|||
}
|
||||
|
||||
/**
|
||||
* Appends an operation AND updates the vector clock in a SINGLE atomic transaction.
|
||||
* Appends an operation AND OVERWRITES the durable vector clock with
|
||||
* `op.vectorClock` as-is, in a single atomic transaction. No rebase, no
|
||||
* merge — whatever clock the caller built replaces the durable one.
|
||||
*
|
||||
* INVARIANT (#8939): the caller MUST derive `op.vectorClock` from the
|
||||
* durable clock inside the same sp_op_log lock hold, after
|
||||
* `clearVectorClockCache()` — the capture path in OperationLogEffects is the
|
||||
* only production caller and does exactly that. Any other derivation (e.g.
|
||||
* from the per-tab in-memory cache) can regress the durable clock and reuse
|
||||
* counters. New writers must use `appendMixedSourceBatchSkipDuplicates`,
|
||||
* whose in-transaction rebase makes regression unrepresentable.
|
||||
*
|
||||
* PERFORMANCE: This is the key optimization for mobile devices. Previously, each action
|
||||
* required two separate IndexedDB transactions (one to SUP_OPS, one to pf.META_MODEL).
|
||||
* By consolidating the vector clock into SUP_OPS, we can write both in a single transaction,
|
||||
* reducing disk I/O by ~50%.
|
||||
*
|
||||
* NOTE: The operation's vectorClock field should already contain the incremented clock
|
||||
* (incremented by the caller). This method stores that clock as the current vector clock,
|
||||
* it does NOT increment again.
|
||||
*
|
||||
* @param op The operation to append (with vectorClock already set)
|
||||
* @param source Whether this is a local or remote operation
|
||||
* @param options Additional options (e.g., pendingApply for remote ops)
|
||||
* @returns The sequence number of the appended operation
|
||||
*/
|
||||
async appendWithVectorClockUpdate(
|
||||
async appendWithVectorClockOverwrite(
|
||||
op: Operation,
|
||||
source: 'local' | 'remote' = 'local',
|
||||
options?: { pendingApply?: boolean },
|
||||
|
|
@ -2675,6 +2700,7 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort<O
|
|||
await this._ensureInit();
|
||||
|
||||
const { staleRepairOpId, replacementOp, repairedState } = opts;
|
||||
let committedClock: VectorClock | undefined;
|
||||
const seq = await this._adapter.transaction(
|
||||
[
|
||||
STORE_NAMES.OPS,
|
||||
|
|
@ -2696,28 +2722,42 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort<O
|
|||
staleEntry.rejectedAt = Date.now();
|
||||
await tx.put(STORE_NAMES.OPS, staleEntry);
|
||||
|
||||
const replacementEntry = this._buildStoredEntry(replacementOp, 'local');
|
||||
// Rebase onto the durable clock read in this same transaction — the
|
||||
// caller-built clock may come from a stale in-memory cache (#8939).
|
||||
const currentClockEntry = await tx.get<VectorClockEntry>(
|
||||
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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
|
|
|
|||
|
|
@ -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<string, number> };
|
||||
|
||||
// The merged remote entries must be present.
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -50,7 +50,7 @@ describe('Post-sync validation latch (#7330) — integration', () => {
|
|||
'markApplied',
|
||||
'getOpById',
|
||||
'mergeRemoteOpClocks',
|
||||
'appendWithVectorClockUpdate',
|
||||
'appendWithVectorClockOverwrite',
|
||||
'markFailed',
|
||||
]);
|
||||
opLogStoreSpy.getUnsynced.and.resolveTo([]);
|
||||
|
|
|
|||
|
|
@ -101,7 +101,7 @@ describe('done task operation replay', () => {
|
|||
const actions$: Observable<Action> = of(action);
|
||||
const mockOpLogStore = jasmine.createSpyObj<OperationLogStoreService>(
|
||||
'OperationLogStoreService',
|
||||
['appendWithVectorClockUpdate', 'getCompactionCounter', 'clearVectorClockCache'],
|
||||
['appendWithVectorClockOverwrite', 'getCompactionCounter', 'clearVectorClockCache'],
|
||||
);
|
||||
const mockLockService = jasmine.createSpyObj<LockService>('LockService', ['request']);
|
||||
const mockVectorClockService = jasmine.createSpyObj<VectorClockService>(
|
||||
|
|
@ -132,7 +132,7 @@ describe('done task operation replay', () => {
|
|||
mockLockService.request.and.callFake(async <T>(_name: string, fn: () => Promise<T>) =>
|
||||
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;
|
||||
};
|
||||
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
|
|
|
|||
|
|
@ -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<string, number>,
|
||||
): Promise<void> => {
|
||||
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,
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
@ -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 <T>(_name: string, fn: () => Promise<T>) =>
|
||||
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);
|
||||
|
|
|
|||
|
|
@ -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(),
|
||||
|
|
|
|||
|
|
@ -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 }),
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue