mirror of
https://github.com/johannesjo/super-productivity.git
synced 2026-07-18 00:46:45 +00:00
fix(sync): implement clean slate semantics for SYNC_IMPORT/BACKUP_IMPORT
SYNC_IMPORT and BACKUP_IMPORT now represent a complete fresh start: - All operations without knowledge of the import are dropped - CONCURRENT ops from "unknown clients" are no longer preserved - Local synced ops are NOT replayed after import This ensures a true "restore to point in time" semantic where all clients start fresh from the imported state, with no concurrent work preserved. Changes: - Simplify SyncImportFilterService to filter ALL CONCURRENT/LESS_THAN ops - Remove _replayLocalSyncedOpsAfterImport() method and related code - Remove _checkOperationEntitiesExist() helper method - Update tests to expect new behavior - Document semantics in CLAUDE.md
This commit is contained in:
Johannes Millan
parent
5138b46546
commit
9f6bb4e7ac
5 changed files with 71 additions and 1309 deletions
|
|
@ -106,6 +106,7 @@ The app uses NgRx (Redux pattern) for state management. Key state slices:
|
|||
9. **Atomic Multi-Entity Changes**: When one action affects multiple entities (e.g., deleting a tag removes it from tasks), use **meta-reducers** instead of effects to ensure all changes happen in a single reducer pass. This creates one operation in the sync log, preventing partial sync and state inconsistency. See `src/app/root-store/meta/task-shared-meta-reducers/` and Part F in the architecture docs.
|
||||
10. **TODAY_TAG is a Virtual Tag**: TODAY_TAG (ID: `'TODAY'`) must **NEVER** be added to `task.tagIds`. It's a "virtual tag" where membership is determined by `task.dueDay`, and `TODAY_TAG.taskIds` only stores ordering. This keeps move operations uniform across all tags. See `docs/ai/today-tag-architecture.md`.
|
||||
11. **Event Loop Yield After Bulk Dispatches**: When applying many operations to NgRx in rapid succession (e.g., during sync replay), add `await new Promise(resolve => setTimeout(resolve, 0))` after the dispatch loop. `store.dispatch()` is non-blocking and returns immediately. Without yielding, 50+ rapid dispatches can overwhelm the store and cause state updates to be lost. See `OperationApplierService.applyOperations()` for the reference implementation.
|
||||
12. **SYNC_IMPORT Semantics**: `SYNC_IMPORT` (and `BACKUP_IMPORT`) operations represent a **complete fresh start** - they replace the entire application state. All operations without knowledge of the import (CONCURRENT or LESS_THAN by vector clock) are dropped for all clients. See `SyncImportFilterService.filterOpsInvalidatedBySyncImport()`. This is correct behavior: the import is an explicit user action to restore to a specific state, and concurrent work is intentionally discarded.
|
||||
|
||||
## 🚫 Known Anti-Patterns to Avoid
|
||||
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load diff
|
|
@ -33,9 +33,6 @@ import {
|
|||
} from '../store/schema-migration.service';
|
||||
import { SnackService } from '../../../snack/snack.service';
|
||||
import { T } from '../../../../t.const';
|
||||
import { getEntityConfig, isAdapterEntity } from '../entity-registry';
|
||||
import { firstValueFrom } from 'rxjs';
|
||||
import { Dictionary } from '@ngrx/entity';
|
||||
import { PfapiService } from '../../../../pfapi/pfapi.service';
|
||||
import { PfapiStoreDelegateService } from '../../../../pfapi/pfapi-store-delegate.service';
|
||||
import { uuidv7 } from '../../../../util/uuid-v7';
|
||||
|
|
@ -748,232 +745,6 @@ export class OperationLogSyncService {
|
|||
return window.confirm(`${title}\n\n${message}`);
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks if the primary target entities of an operation exist in the current store.
|
||||
*
|
||||
* For UPDATE/DELETE operations, the target entities (entityId/entityIds) must exist
|
||||
* in the store for the operation to be applied successfully. This method checks
|
||||
* existence using the DependencyResolverService.
|
||||
*
|
||||
* ## Why This Is Needed
|
||||
*
|
||||
* Operations don't carry their own state - they're like Redux actions that describe
|
||||
* "what happened" but rely on the current store state to apply correctly.
|
||||
*
|
||||
* After a SYNC_IMPORT replaces the entire state, some local operations may reference
|
||||
* entities that no longer exist. For example:
|
||||
* - Local client has task T1 (created long ago, CREATE op compacted)
|
||||
* - Local client does `planTasksForToday({ taskIds: [T1] })` → operation O2
|
||||
* - SYNC_IMPORT from remote client arrives (doesn't include T1)
|
||||
* - SYNC_IMPORT replaces state → T1 is gone
|
||||
* - Replay tries to apply O2 → fails because T1 doesn't exist
|
||||
*
|
||||
* @param op - The operation to check
|
||||
* @returns Array of missing entity IDs (empty if all exist or check not applicable)
|
||||
*/
|
||||
private async _checkOperationEntitiesExist(op: Operation): Promise<string[]> {
|
||||
// Get entity IDs directly from the operation metadata.
|
||||
// Operations have: entityId (single entity) and entityIds (bulk operations)
|
||||
const entityIds: string[] = op.entityIds?.length
|
||||
? op.entityIds
|
||||
: op.entityId && op.entityId !== '*'
|
||||
? [op.entityId]
|
||||
: [];
|
||||
|
||||
if (entityIds.length === 0) {
|
||||
return []; // No specific entities to check (e.g., global config)
|
||||
}
|
||||
|
||||
// Skip check for CREATE operations - entities won't exist yet by definition
|
||||
if (op.opType === OpType.Create) {
|
||||
return [];
|
||||
}
|
||||
|
||||
// Check if entities exist in the store using entity registry
|
||||
const config = getEntityConfig(op.entityType);
|
||||
if (!config || !isAdapterEntity(config) || !config.selectEntities) {
|
||||
return []; // Non-adapter entities (singletons, etc.) - assume they exist
|
||||
}
|
||||
|
||||
try {
|
||||
const entities = await firstValueFrom(
|
||||
this.store.select(
|
||||
config.selectEntities as (state: object) => Dictionary<unknown>,
|
||||
),
|
||||
);
|
||||
|
||||
// If entities is undefined/null, assume all entities exist (graceful degradation)
|
||||
if (!entities) {
|
||||
return [];
|
||||
}
|
||||
|
||||
return entityIds.filter((id) => !entities[id]);
|
||||
} catch {
|
||||
// If selector throws (e.g., missing state slice), assume entities exist
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Re-applies local synced operations after a SYNC_IMPORT is applied.
|
||||
*
|
||||
* This handles the "late joiner" scenario where:
|
||||
* 1. Client B creates local tasks (B1, B2, B3)
|
||||
* 2. Client B uploads them to server (server accepts)
|
||||
* 3. Client B receives piggybacked SYNC_IMPORT from Client A
|
||||
* 4. SYNC_IMPORT replaces entire state (B's tasks disappear!)
|
||||
* 5. This method re-applies B's synced ops to restore them
|
||||
*
|
||||
* The key insight: piggybacked ops exclude the client's own ops,
|
||||
* so the SYNC_IMPORT doesn't include Client B's changes.
|
||||
* But those ops ARE on the server and SHOULD be in the final state.
|
||||
*
|
||||
* ## Entity Existence Check
|
||||
*
|
||||
* Before replaying, we verify that the operation's target entities still exist.
|
||||
* SYNC_IMPORT may have deleted entities that local operations reference.
|
||||
* Operations referencing deleted entities are skipped to prevent dangling references.
|
||||
*
|
||||
* @param appliedOps - The ops that were just applied (includes the SYNC_IMPORT)
|
||||
*/
|
||||
private async _replayLocalSyncedOpsAfterImport(appliedOps: Operation[]): Promise<void> {
|
||||
// Get the SYNC_IMPORT's vector clock - we need to replay ops that happened AFTER it
|
||||
const syncImportOp = appliedOps.find(
|
||||
(op) => op.opType === OpType.SyncImport || op.opType === OpType.BackupImport,
|
||||
);
|
||||
if (!syncImportOp) {
|
||||
return; // Shouldn't happen, but be safe
|
||||
}
|
||||
|
||||
// Get the current client ID
|
||||
const clientId = await this.clientIdProvider.loadClientId();
|
||||
if (!clientId) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Get all local ops that:
|
||||
// 1. Were created by THIS client (so they're not in the piggybacked ops)
|
||||
// 2. Are already synced (accepted by server)
|
||||
// 3. Were created AFTER the SYNC_IMPORT (by UUIDv7 timestamp comparison)
|
||||
const allEntries = await this.opLogStore.getOpsAfterSeq(0);
|
||||
const localSyncedOps = allEntries
|
||||
.filter((entry) => {
|
||||
// Must be created by this client
|
||||
if (entry.op.clientId !== clientId) return false;
|
||||
// Must be synced (accepted by server)
|
||||
if (!entry.syncedAt) return false;
|
||||
// Must NOT be a full-state op itself
|
||||
if (
|
||||
entry.op.opType === OpType.SyncImport ||
|
||||
entry.op.opType === OpType.BackupImport
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
// Must be created AFTER the SYNC_IMPORT/BACKUP_IMPORT.
|
||||
// UUIDv7 is time-ordered: first 48 bits = millisecond timestamp.
|
||||
// Ops created BEFORE the import should NOT be replayed - they reference
|
||||
// the old state that was replaced by the import.
|
||||
// NOTE: We use UUIDv7 comparison instead of vector clock comparison
|
||||
// because imports with isForceConflict=true create a fresh vector clock
|
||||
// with a new client ID, breaking vector clock causality detection.
|
||||
if (entry.op.id < syncImportOp.id) {
|
||||
OpLog.verbose(
|
||||
`OperationLogSyncService: Skipping op ${entry.op.id} - created before SYNC_IMPORT`,
|
||||
);
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
})
|
||||
.map((entry) => entry.op);
|
||||
|
||||
if (localSyncedOps.length === 0) {
|
||||
OpLog.normal(
|
||||
'OperationLogSyncService: No local synced ops to replay after SYNC_IMPORT.',
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────
|
||||
// Filter out operations whose target entities were deleted by SYNC_IMPORT
|
||||
// ─────────────────────────────────────────────────────────────────────────
|
||||
// SYNC_IMPORT replaces the entire state. Local operations may reference
|
||||
// entities that existed before but are now gone. We must skip these to
|
||||
// prevent creating dangling references (e.g., Today tag → non-existent task).
|
||||
//
|
||||
// IMPORTANT: We must track entities that will be CREATED by ops in this sequence.
|
||||
// If an ADD op creates entity X, then UPDATE/DELETE ops for X should NOT be skipped,
|
||||
// even if X doesn't exist in the SYNC_IMPORT state yet.
|
||||
const validOps: Operation[] = [];
|
||||
const skippedOps: { op: Operation; missingIds: string[] }[] = [];
|
||||
|
||||
// First pass: collect entity IDs that will be created by CREATE ops in this sequence
|
||||
const entitiesCreatedBySequence = new Set<string>();
|
||||
for (const op of localSyncedOps) {
|
||||
if (op.opType === OpType.Create && op.entityId) {
|
||||
entitiesCreatedBySequence.add(op.entityId);
|
||||
}
|
||||
}
|
||||
|
||||
for (const op of localSyncedOps) {
|
||||
const missingEntityIds = await this._checkOperationEntitiesExist(op);
|
||||
|
||||
// Filter out missing entities that will be created by earlier ops in this sequence
|
||||
const trulyMissingIds = missingEntityIds.filter(
|
||||
(id) => !entitiesCreatedBySequence.has(id),
|
||||
);
|
||||
|
||||
if (trulyMissingIds.length > 0) {
|
||||
OpLog.warn(
|
||||
`OperationLogSyncService: Skipping op ${op.id} (${op.actionType}) - ` +
|
||||
`target entities deleted by SYNC_IMPORT: ${trulyMissingIds.join(', ')}`,
|
||||
);
|
||||
skippedOps.push({ op, missingIds: trulyMissingIds });
|
||||
} else {
|
||||
validOps.push(op);
|
||||
}
|
||||
}
|
||||
|
||||
if (validOps.length === 0) {
|
||||
OpLog.normal(
|
||||
`OperationLogSyncService: All ${localSyncedOps.length} local ops skipped - ` +
|
||||
`target entities deleted by SYNC_IMPORT.`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
OpLog.normal(
|
||||
`OperationLogSyncService: Replaying ${validOps.length} local synced ops ` +
|
||||
`(${skippedOps.length} skipped) after SYNC_IMPORT.`,
|
||||
);
|
||||
|
||||
// Re-apply these ops to restore the local changes on top of the SYNC_IMPORT state
|
||||
// Operations are already in server sequence order, which is causally correct
|
||||
// Use applyOperations which handles action dispatching
|
||||
try {
|
||||
await this.operationApplier.applyOperations(validOps);
|
||||
} catch (replayError) {
|
||||
// If replay fails after SYNC_IMPORT succeeded, user loses their local synced operations.
|
||||
// Log the error, notify user, and trigger validation to detect/repair any inconsistencies.
|
||||
OpLog.err(
|
||||
`OperationLogSyncService: Failed to replay ${validOps.length} local ops after SYNC_IMPORT`,
|
||||
replayError,
|
||||
);
|
||||
|
||||
// Run validation to detect and repair any state inconsistencies
|
||||
await this.validateStateService.validateAndRepairCurrentState(
|
||||
'replay-local-ops-failed',
|
||||
);
|
||||
|
||||
// Notify user that some changes may be lost
|
||||
this.snackService.open({
|
||||
type: 'ERROR',
|
||||
msg: T.F.SYNC.S.REPLAY_LOCAL_OPS_FAILED,
|
||||
});
|
||||
|
||||
// Don't re-throw - the SYNC_IMPORT itself succeeded, so sync should continue
|
||||
}
|
||||
}
|
||||
|
||||
// ═══════════════════════════════════════════════════════════════════════════
|
||||
// REMOTE OPS PROCESSING (Core Pipeline)
|
||||
// ═══════════════════════════════════════════════════════════════════════════
|
||||
|
|
@ -1099,14 +870,9 @@ export class OperationLogSyncService {
|
|||
);
|
||||
await this._applyNonConflictingOps(validOps);
|
||||
|
||||
// IMPORTANT: After applying a SYNC_IMPORT, re-apply any local ops that were
|
||||
// already synced to the server. This handles the "late joiner" scenario where:
|
||||
// 1. Client B creates local tasks (B1, B2, B3)
|
||||
// 2. Client B uploads them (server accepts)
|
||||
// 3. Client B receives piggybacked SYNC_IMPORT from Client A
|
||||
// 4. SYNC_IMPORT replaces state (B's tasks lost!)
|
||||
// 5. We need to re-apply B's synced ops to restore them
|
||||
await this._replayLocalSyncedOpsAfterImport(validOps);
|
||||
// Clean Slate Semantics: SYNC_IMPORT/BACKUP_IMPORT replaces entire state.
|
||||
// Local synced ops are NOT replayed - the import is an explicit user action
|
||||
// to restore all clients to a specific point in time.
|
||||
|
||||
await this._validateAfterSync();
|
||||
return { localWinOpsCreated: 0 };
|
||||
|
|
|
|||
|
|
@ -244,11 +244,12 @@ describe('SyncImportFilterService', () => {
|
|||
|
||||
const result = await service.filterOpsInvalidatedBySyncImport(ops);
|
||||
|
||||
// Updated behavior: Client A was UNKNOWN to the latest import (no clientA entry)
|
||||
// So ALL of Client A's ops are kept (parallel branch of work)
|
||||
// Valid: all 5 ops (2 imports + 3 updates from unknown client)
|
||||
expect(result.validOps.length).toBe(5);
|
||||
expect(result.invalidatedOps.length).toBe(0);
|
||||
// Clean Slate Semantics: Only ops with knowledge of the latest import are kept.
|
||||
// - First two Client A ops are CONCURRENT (no knowledge of latest import) → filtered
|
||||
// - Third Client A op is GREATER_THAN (has latest import's clock) → kept
|
||||
// - Both SYNC_IMPORTs are kept
|
||||
expect(result.validOps.length).toBe(3); // 2 imports + 1 post-import op
|
||||
expect(result.invalidatedOps.length).toBe(2); // 2 concurrent ops from Client A
|
||||
});
|
||||
|
||||
it('should filter pre-import ops when SYNC_IMPORT was downloaded in a PREVIOUS sync cycle', async () => {
|
||||
|
|
@ -271,6 +272,7 @@ describe('SyncImportFilterService', () => {
|
|||
);
|
||||
|
||||
// These are OLD ops from Client A, created BEFORE the import
|
||||
// They are CONCURRENT with the import (no knowledge of it)
|
||||
const oldOpsFromClientA: Operation[] = [
|
||||
{
|
||||
id: '019afd60-0001-7000-0000-000000000000',
|
||||
|
|
@ -280,7 +282,7 @@ describe('SyncImportFilterService', () => {
|
|||
entityId: 'task-1',
|
||||
payload: { title: 'Old title' },
|
||||
clientId: 'client-A',
|
||||
vectorClock: { clientA: 5 },
|
||||
vectorClock: { clientA: 5 }, // CONCURRENT - no knowledge of import
|
||||
timestamp: Date.now(),
|
||||
schemaVersion: 1,
|
||||
},
|
||||
|
|
@ -292,7 +294,7 @@ describe('SyncImportFilterService', () => {
|
|||
entityId: 'task-2',
|
||||
payload: { title: 'Another old title' },
|
||||
clientId: 'client-A',
|
||||
vectorClock: { clientA: 6 },
|
||||
vectorClock: { clientA: 6 }, // CONCURRENT - no knowledge of import
|
||||
timestamp: Date.now(),
|
||||
schemaVersion: 1,
|
||||
},
|
||||
|
|
@ -300,10 +302,10 @@ describe('SyncImportFilterService', () => {
|
|||
|
||||
const result = await service.filterOpsInvalidatedBySyncImport(oldOpsFromClientA);
|
||||
|
||||
// Updated behavior: Client A was UNKNOWN to the import (no clientA entry)
|
||||
// So Client A's ops are kept (parallel branch of work)
|
||||
expect(result.validOps.length).toBe(2);
|
||||
expect(result.invalidatedOps.length).toBe(0);
|
||||
// Clean Slate Semantics: CONCURRENT ops are filtered, even from unknown clients.
|
||||
// These ops have no knowledge of the import, so they're invalidated.
|
||||
expect(result.validOps.length).toBe(0);
|
||||
expect(result.invalidatedOps.length).toBe(2);
|
||||
expect(opLogStoreSpy.getLatestFullStateOp).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
|
|
@ -379,10 +381,10 @@ describe('SyncImportFilterService', () => {
|
|||
|
||||
const result = await service.filterOpsInvalidatedBySyncImport(ops);
|
||||
|
||||
// Updated behavior: Client B was UNKNOWN to the import (no clientB entry)
|
||||
// So Client B's ops are kept (parallel branch of work)
|
||||
expect(result.validOps.length).toBe(2);
|
||||
expect(result.invalidatedOps.length).toBe(0);
|
||||
// Clean Slate Semantics: CONCURRENT ops are filtered, even from unknown clients.
|
||||
// Client B's op has no knowledge of the import, so it's invalidated.
|
||||
expect(result.validOps.length).toBe(1); // Only SYNC_IMPORT
|
||||
expect(result.invalidatedOps.length).toBe(1); // Client B's concurrent op
|
||||
});
|
||||
|
||||
it('should filter LESS_THAN ops (dominated by import)', async () => {
|
||||
|
|
@ -519,10 +521,11 @@ describe('SyncImportFilterService', () => {
|
|||
|
||||
const result = await service.filterOpsInvalidatedBySyncImport(ops);
|
||||
|
||||
// Updated behavior: Client B was UNKNOWN to the import (no clientB entry)
|
||||
// So Client B's ops are kept (parallel branch of work)
|
||||
expect(result.validOps.length).toBe(2);
|
||||
expect(result.invalidatedOps.length).toBe(0);
|
||||
// Clean Slate Semantics: CONCURRENT ops are filtered based on vector clock,
|
||||
// NOT UUIDv7 timestamp. Even though UUIDv7 is later, vector clock shows
|
||||
// no knowledge of import, so it's filtered.
|
||||
expect(result.validOps.length).toBe(1); // Only SYNC_IMPORT
|
||||
expect(result.invalidatedOps.length).toBe(1); // Client B's concurrent op
|
||||
});
|
||||
|
||||
it('should handle REPAIR operations the same as SYNC_IMPORT', async () => {
|
||||
|
|
@ -555,10 +558,10 @@ describe('SyncImportFilterService', () => {
|
|||
|
||||
const result = await service.filterOpsInvalidatedBySyncImport(ops);
|
||||
|
||||
// Updated behavior: Client B was UNKNOWN to the repair (no clientB entry)
|
||||
// So Client B's ops are kept (parallel branch of work)
|
||||
expect(result.validOps.length).toBe(2);
|
||||
expect(result.invalidatedOps.length).toBe(0);
|
||||
// Clean Slate Semantics: REPAIR ops are handled the same as SYNC_IMPORT.
|
||||
// CONCURRENT ops are filtered, even from unknown clients.
|
||||
expect(result.validOps.length).toBe(1); // Only REPAIR
|
||||
expect(result.invalidatedOps.length).toBe(1); // Client B's concurrent op
|
||||
});
|
||||
});
|
||||
|
||||
|
|
@ -624,15 +627,15 @@ describe('SyncImportFilterService', () => {
|
|||
expect(result.invalidatedOps.length).toBe(0);
|
||||
});
|
||||
|
||||
it('should correctly keep ops from unknown clients', async () => {
|
||||
// Updated behavior: ops from clients unknown to the SYNC_IMPORT are KEPT
|
||||
// Rationale: if the import didn't know about this client, the op is NOT
|
||||
// "pre-import state" - it's from a parallel branch of work.
|
||||
it('should filter ops from unknown clients (clean slate semantics)', async () => {
|
||||
// Clean Slate Semantics: ops from clients unknown to the SYNC_IMPORT are FILTERED.
|
||||
// Rationale: the import is an explicit user action to restore ALL clients to
|
||||
// a specific state. Concurrent work is intentionally discarded.
|
||||
//
|
||||
// Scenario:
|
||||
// 1. Client A creates SYNC_IMPORT with clock {clientA: 1}
|
||||
// 2. Client B (unknown to A) creates op with clock {clientB: 6}
|
||||
// 3. Client B's op should be KEPT (client B was unknown to import)
|
||||
// 3. Client B's op should be FILTERED (no knowledge of import)
|
||||
|
||||
const existingSyncImport: Operation = {
|
||||
id: '019afd68-0050-7000-0000-000000000000',
|
||||
|
|
@ -669,9 +672,9 @@ describe('SyncImportFilterService', () => {
|
|||
|
||||
const result = await service.filterOpsInvalidatedBySyncImport(unknownClientOp);
|
||||
|
||||
// Should be KEPT since client-B was unknown to the import
|
||||
expect(result.validOps.length).toBe(1);
|
||||
expect(result.invalidatedOps.length).toBe(0);
|
||||
// Should be FILTERED - no knowledge of import means it's pre-import state
|
||||
expect(result.validOps.length).toBe(0);
|
||||
expect(result.invalidatedOps.length).toBe(1);
|
||||
});
|
||||
|
||||
it('should handle multiple clients scenario correctly', async () => {
|
||||
|
|
|
|||
|
|
@ -21,11 +21,21 @@ import { OpLog } from '../../../log';
|
|||
* - Applying them causes "Task not found" errors
|
||||
* ```
|
||||
*
|
||||
* ## The Solution
|
||||
* Use VECTOR CLOCK comparison to determine if ops were created with knowledge
|
||||
* of the import. This is more reliable than UUIDv7 timestamps because vector
|
||||
* clocks track CAUSALITY (did the client know about the import?) rather than
|
||||
* wall-clock time (which can be affected by clock drift).
|
||||
* ## The Solution: Clean Slate Semantics
|
||||
* SYNC_IMPORT and BACKUP_IMPORT are explicit user actions to restore ALL clients
|
||||
* to a specific state. ALL operations without knowledge of the import are dropped:
|
||||
*
|
||||
* - **GREATER_THAN / EQUAL**: Op was created with knowledge of import → KEEP
|
||||
* - **CONCURRENT**: Op was created without knowledge of import → DROP
|
||||
* - **LESS_THAN**: Op is dominated by import → DROP
|
||||
*
|
||||
* This ensures a true "restore to point in time" semantic. Concurrent work from
|
||||
* other clients is intentionally discarded because the user explicitly chose to
|
||||
* reset all state to the imported snapshot.
|
||||
*
|
||||
* We use vector clock comparison (not UUIDv7 timestamps) because vector clocks
|
||||
* track CAUSALITY (did the client know about the import?) rather than wall-clock
|
||||
* time (which can be affected by clock drift).
|
||||
*/
|
||||
@Injectable({
|
||||
providedIn: 'root',
|
||||
|
|
@ -36,6 +46,10 @@ export class SyncImportFilterService {
|
|||
/**
|
||||
* Filters out operations invalidated by a SYNC_IMPORT, BACKUP_IMPORT, or REPAIR.
|
||||
*
|
||||
* ## Clean Slate Semantics
|
||||
* Imports are explicit user actions to restore all clients to a specific state.
|
||||
* ALL operations without knowledge of the import are dropped - no exceptions.
|
||||
*
|
||||
* ## Vector Clock Comparison Results
|
||||
* | Comparison | Meaning | Action |
|
||||
* |----------------|--------------------------------------|---------|
|
||||
|
|
@ -44,8 +58,8 @@ export class SyncImportFilterService {
|
|||
* | LESS_THAN | Op dominated by import | ❌ Filter|
|
||||
* | CONCURRENT | Op created without knowledge of import| ❌ Filter|
|
||||
*
|
||||
* Note: CONCURRENT ops are filtered because they reference pre-import state,
|
||||
* even if they were created "after" the import in wall-clock time.
|
||||
* CONCURRENT ops are filtered even if they come from a client the import
|
||||
* didn't know about. This ensures a true "restore to point in time" semantic.
|
||||
*
|
||||
* The import can be in the current batch OR in the local store from a
|
||||
* previous sync cycle. We check both to handle the case where old ops from
|
||||
|
|
@ -115,17 +129,15 @@ export class SyncImportFilterService {
|
|||
// Vector clocks track CAUSALITY ("did this client know about the import?")
|
||||
// rather than wall-clock time, making them immune to client clock drift.
|
||||
//
|
||||
// Comparison results:
|
||||
// Clean Slate Semantics:
|
||||
// - GREATER_THAN: Op was created by a client that SAW the import → KEEP
|
||||
// - EQUAL: Same causal history as import → KEEP
|
||||
// - LESS_THAN: Op is dominated by import (created before with less history) → FILTER
|
||||
// - CONCURRENT: Op created WITHOUT knowledge of import → handled below
|
||||
// - CONCURRENT: Op created WITHOUT knowledge of import → FILTER
|
||||
// - LESS_THAN: Op is dominated by import → FILTER
|
||||
//
|
||||
// SPECIAL CASE for CONCURRENT ops:
|
||||
// If the op's client is NOT in the SYNC_IMPORT's clock, this means the
|
||||
// SYNC_IMPORT was created without knowledge of that client. In this case,
|
||||
// the op is NOT "pre-import state" - it's from a client that the import
|
||||
// didn't know about. These ops should be KEPT and applied.
|
||||
// CONCURRENT ops are filtered even from "unknown" clients. The import is
|
||||
// an explicit user action to restore to a specific state - any concurrent
|
||||
// work is intentionally discarded to ensure a clean slate.
|
||||
const comparison = compareVectorClocks(op.vectorClock, latestImport.vectorClock);
|
||||
|
||||
if (
|
||||
|
|
@ -134,36 +146,9 @@ export class SyncImportFilterService {
|
|||
) {
|
||||
// Op was created by a client that had knowledge of the import
|
||||
validOps.push(op);
|
||||
} else if (comparison === VectorClockComparison.CONCURRENT) {
|
||||
// Check if the op's client was unknown when the SYNC_IMPORT was created
|
||||
const opClientValue = op.vectorClock[op.clientId] ?? 0;
|
||||
const importClientValue = latestImport.vectorClock[op.clientId] ?? 0;
|
||||
|
||||
if (importClientValue === 0) {
|
||||
// The SYNC_IMPORT didn't know about this client at all
|
||||
// This op is NOT "pre-import state" - it's from a parallel branch
|
||||
// that the import didn't include. Keep it and apply.
|
||||
OpLog.verbose(
|
||||
`SyncImportFilterService: Keeping CONCURRENT op ${op.id} - ` +
|
||||
`client ${op.clientId} was unknown to SYNC_IMPORT`,
|
||||
);
|
||||
validOps.push(op);
|
||||
} else if (opClientValue > importClientValue) {
|
||||
// The op's client counter is higher than what the import knew about
|
||||
// This means the op was created AFTER the import's knowledge of this client
|
||||
OpLog.verbose(
|
||||
`SyncImportFilterService: Keeping CONCURRENT op ${op.id} - ` +
|
||||
`client counter ${opClientValue} > import's ${importClientValue}`,
|
||||
);
|
||||
validOps.push(op);
|
||||
} else {
|
||||
// The op was created before or at the import's knowledge level for this client
|
||||
// AND the import has entries the op doesn't know about
|
||||
// This is truly pre-import state - filter it
|
||||
invalidatedOps.push(op);
|
||||
}
|
||||
} else {
|
||||
// LESS_THAN: Op is dominated by import - definitely pre-import state
|
||||
// CONCURRENT or LESS_THAN: Op was created without knowledge of import
|
||||
// Filter it to ensure clean slate semantics
|
||||
invalidatedOps.push(op);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue