Mirrors/super-productivity
1
0
Fork 0
mirror of https://github.com/johannesjo/super-productivity.git synced 2026-07-18 17:05:48 +00:00
Code Issues Projects Releases Wiki Activity

refactor(sync): replace validationFailed plumbing with session latch + integration test

Browse source 
Final review follow-up for #7330. Behavior unchanged; surface area shrinks.

The validation-failed signal previously rode through 7 typed boundaries:
DownloadOutcome.{ops_processed,no_new_ops}.validationFailed,
UploadOutcome.completed.validationFailed, processRemoteOps return,
forceDownloadRemoteState return, _handleSyncImportConflict return,
DownloadResultForRejection.validationFailed, RejectionHandlingResult
.validationFailed, plus a closure-captured piggybackValidationFailed in
uploadPendingOps. Threading it correctly required every call site to
remember to forward the boolean — a new variant or path that forgot
would silently let IN_SYNC ride over corrupt state.

- Add SyncSessionValidationService — a no-dep singleton with reset() /
  setFailed() / hasFailed(). RemoteOpsProcessingService.validateAfterSync
  and ConflictResolutionService.autoResolveConflictsLWW flip the latch
  when validation reports corruption. SyncWrapperService resets it at
  every entry point (sync(), _forceDownload(), resolveSyncConflict
  USE_REMOTE) and reads it once before deciding IN_SYNC vs ERROR.

- Drop validationFailed from DownloadOutcome variants, UploadOutcome,
  DownloadResultForRejection, RejectionHandlingResult, processRemoteOps
  return, autoResolveConflictsLWW return, forceDownloadRemoteState
  return, and _handleSyncImportConflict return. ~120 LOC of plumbing
  collapsed to single latch reads.

- Add a focused integration test
  (post-sync-validation.integration.spec.ts) that wires real
  RemoteOpsProcessingService + ConflictResolutionService against a
  stubbed ValidateStateService. Asserts the latch flips on validation
  failure and survives discarded-boolean callers — catching plumbing
  regressions a future code path could otherwise sneak past.

Net change vs current branch: ~120 LOC deleted from production sources,
~270 LOC added in new service + tests (latch unit tests + integration).
This commit is contained in:
Johannes Millan 2026-05-08 15:31:02 +02:00
parent 22c9fd0f69
commit b3cbdbd41a
14 changed files with 507 additions and 353 deletions
Download patch file Download diff file Expand all files Collapse all files

196
src/app/imex/sync/sync-wrapper.service.spec.ts
View file

@ -5,6 +5,7 @@ import { BehaviorSubject, firstValueFrom, of } from 'rxjs';
import { SyncWrapperService } from './sync-wrapper.service';
import { SyncProviderManager } from '../../op-log/sync-providers/provider-manager.service';
import { OperationLogSyncService } from '../../op-log/sync/operation-log-sync.service';
import { SyncSessionValidationService } from '../../op-log/sync/sync-session-validation.service';
import { WrappedProviderService } from '../../op-log/sync-providers/wrapped-provider.service';
import { OperationLogStoreService } from '../../op-log/persistence/operation-log-store.service';
import { LegacyPfDbService } from '../../core/persistence/legacy-pf-db.service';
@ -613,19 +614,22 @@ describe('SyncWrapperService', () => {
expect(mockProviderManager.setSyncStatus).not.toHaveBeenCalledWith('IN_SYNC');
});
// Issue #7330: previously, post-sync state validation failure was surfaced
// only via a snackbar; sync still reported IN_SYNC, contradicting the
// "State validation failed" message the user just saw. Sync status must
// reflect the validation result.
it('should set ERROR (not IN_SYNC) when downloadRemoteOps reports validationFailed', async () => {
mockSyncService.downloadRemoteOps.and.returnValue(
Promise.resolve({
// Issue #7330: post-sync state validation failure must not be reported
// as IN_SYNC. After the latch refactor, validation failure is signalled
// via SyncSessionValidationService — the wrapper reads it once before
// claiming IN_SYNC. Tests below simulate validation failure by flipping
// the latch from inside a mocked sync call (mirroring what
// RemoteOpsProcessingService.validateAfterSync does in production).
it('should set ERROR (not IN_SYNC) when validation latch is flipped during download', async () => {
const latch = TestBed.inject(SyncSessionValidationService);
mockSyncService.downloadRemoteOps.and.callFake(async () => {
latch.setFailed();
return {
kind: 'ops_processed' as const,
newOpsCount: 3,
localWinOpsCreated: 0,
validationFailed: true,
}),
);
};
});
const result = await service.sync();
@ -634,9 +638,11 @@ describe('SyncWrapperService', () => {
expect(mockProviderManager.setSyncStatus).not.toHaveBeenCalledWith('IN_SYNC');
});
it('should set ERROR (not IN_SYNC) when uploadPendingOps reports validationFailed', async () => {
mockSyncService.uploadPendingOps.and.returnValue(
Promise.resolve({
it('should set ERROR (not IN_SYNC) when validation latch is flipped during upload', async () => {
const latch = TestBed.inject(SyncSessionValidationService);
mockSyncService.uploadPendingOps.and.callFake(async () => {
latch.setFailed();
return {
kind: 'completed' as const,
uploadedCount: 0,
piggybackedOpsCount: 2,
@ -644,9 +650,8 @@ describe('SyncWrapperService', () => {
permanentRejectionCount: 0,
hasMorePiggyback: false,
rejectedOps: [],
validationFailed: true,
}),
);
};
});
const result = await service.sync();
@ -655,25 +660,29 @@ describe('SyncWrapperService', () => {
expect(mockProviderManager.setSyncStatus).not.toHaveBeenCalledWith('IN_SYNC');
});
// Issue #7330 follow-up: the LWW re-upload retry loop initially discarded
// reuploadResult.validationFailed. If a re-upload pass triggered a
// piggybacked download with failing post-sync validation, sync would
// still report IN_SYNC.
it('should set ERROR (not IN_SYNC) when LWW re-upload reports validationFailed', async () => {
// First upload returns localWinOpsCreated to enter the retry loop.
// Re-upload returns validationFailed=true with no remaining LWW ops
// (loop terminates after detecting the failure).
mockSyncService.uploadPendingOps.and.returnValues(
Promise.resolve({
kind: 'completed' as const,
uploadedCount: 1,
piggybackedOpsCount: 0,
localWinOpsCreated: 2,
permanentRejectionCount: 0,
hasMorePiggyback: false,
rejectedOps: [],
}),
Promise.resolve({
// Issue #7330 follow-up: a retry-pass piggybacked download can flip the
// latch — the wrapper must still report ERROR.
it('should set ERROR (not IN_SYNC) when validation latch is flipped during LWW re-upload', async () => {
const latch = TestBed.inject(SyncSessionValidationService);
let uploadCallCount = 0;
mockSyncService.uploadPendingOps.and.callFake(async () => {
uploadCallCount++;
if (uploadCallCount === 1) {
// Initial upload returns localWinOpsCreated to enter the retry loop.
return {
kind: 'completed' as const,
uploadedCount: 1,
piggybackedOpsCount: 0,
localWinOpsCreated: 2,
permanentRejectionCount: 0,
hasMorePiggyback: false,
rejectedOps: [],
};
}
// Retry pass flips the latch (simulating validation failure during
// a piggybacked download triggered by uploadPendingOps).
latch.setFailed();
return {
kind: 'completed' as const,
uploadedCount: 0,
piggybackedOpsCount: 1,
@ -681,9 +690,8 @@ describe('SyncWrapperService', () => {
permanentRejectionCount: 0,
hasMorePiggyback: false,
rejectedOps: [],
validationFailed: true,
}),
);
};
});
const result = await service.sync();
@ -692,33 +700,29 @@ describe('SyncWrapperService', () => {
expect(mockProviderManager.setSyncStatus).not.toHaveBeenCalledWith('IN_SYNC');
});
// #7521 follow-up: when re-upload retries exhaust AND validation failed
// during one of the retry passes, prefer ERROR over UNKNOWN_OR_CHANGED.
// Validation failure is a more serious signal than unuploaded ops.
it('should set ERROR (not UNKNOWN_OR_CHANGED) when retries exhaust AND validation failed during a retry', async () => {
// Initial upload returns 1 LWW op to enter the retry loop.
// Every retry returns localWinOpsCreated: 1 (so loop hits MAX retries),
// and one retry reports validationFailed: true.
const completed = (
validationFailed = false,
): Awaited<ReturnType<typeof mockSyncService.uploadPendingOps>> => ({
kind: 'completed' as const,
uploadedCount: 1,
piggybackedOpsCount: 0,
localWinOpsCreated: 1,
permanentRejectionCount: 0,
hasMorePiggyback: false,
rejectedOps: [],
validationFailed,
// #7521 follow-up: when re-upload retries exhaust AND the latch is
// flipped, prefer ERROR over UNKNOWN_OR_CHANGED — validation failure is
// a more serious signal than unuploaded ops.
it('should set ERROR (not UNKNOWN_OR_CHANGED) when retries exhaust AND latch is flipped', async () => {
const latch = TestBed.inject(SyncSessionValidationService);
let uploadCallCount = 0;
mockSyncService.uploadPendingOps.and.callFake(async () => {
uploadCallCount++;
if (uploadCallCount === 2) {
// One retry flips the latch; subsequent retries don't, but the
// latch persists for the rest of the session.
latch.setFailed();
}
return {
kind: 'completed' as const,
uploadedCount: 1,
piggybackedOpsCount: 0,
localWinOpsCreated: 1,
permanentRejectionCount: 0,
hasMorePiggyback: false,
rejectedOps: [],
};
});
// 1 initial + MAX_LWW_REUPLOAD_RETRIES retries; one of the retries flags validation.
mockSyncService.uploadPendingOps.and.returnValues(
Promise.resolve(completed()),
Promise.resolve(completed(true)),
...Array.from({ length: MAX_LWW_REUPLOAD_RETRIES }, () =>
Promise.resolve(completed()),
),
);
const result = await service.sync();
@ -730,24 +734,19 @@ describe('SyncWrapperService', () => {
expect(mockProviderManager.setSyncStatus).not.toHaveBeenCalledWith('IN_SYNC');
});
// #7330 follow-up: when retries exhaust AND the *initial* download already
// reported validationFailed (separate from the retry pass), the wrapper
// must still report ERROR. Previously the retry-exhaustion branch only
// checked reuploadValidationFailed, so an initial downloadValidationFailed
// combined with retry exhaustion fell through to UNKNOWN_OR_CHANGED.
it('should set ERROR (not UNKNOWN_OR_CHANGED) when retries exhaust AND initial download reported validationFailed', async () => {
mockSyncService.downloadRemoteOps.and.returnValue(
Promise.resolve({
// #7330 follow-up: when retries exhaust AND the initial download flipped
// the latch (before the retry loop), the wrapper must still report ERROR.
it('should set ERROR (not UNKNOWN_OR_CHANGED) when retries exhaust AND initial download flipped the latch', async () => {
const latch = TestBed.inject(SyncSessionValidationService);
mockSyncService.downloadRemoteOps.and.callFake(async () => {
latch.setFailed();
return {
kind: 'ops_processed' as const,
newOpsCount: 3,
localWinOpsCreated: 0,
validationFailed: true,
}),
);
// Every upload reports localWinOpsCreated: 1 so retries always exhaust.
const completed = (): Awaited<
ReturnType<typeof mockSyncService.uploadPendingOps>
> => ({
};
});
mockSyncService.uploadPendingOps.and.callFake(async () => ({
kind: 'completed' as const,
uploadedCount: 1,
piggybackedOpsCount: 0,
@ -755,12 +754,7 @@ describe('SyncWrapperService', () => {
permanentRejectionCount: 0,
hasMorePiggyback: false,
rejectedOps: [],
});
mockSyncService.uploadPendingOps.and.returnValues(
...Array.from({ length: 1 + MAX_LWW_REUPLOAD_RETRIES }, () =>
Promise.resolve(completed()),
),
);
}));
const result = await service.sync();
@ -771,17 +765,16 @@ describe('SyncWrapperService', () => {
);
});
// #7330: the USE_REMOTE conflict-resolution path returns DownloadOutcome
// with kind 'no_new_ops' and validationFailed: true when the
// forceDownloadRemoteState validator rejected the applied state. The
// wrapper must surface this as ERROR rather than IN_SYNC.
it('should set ERROR when downloadRemoteOps returns no_new_ops with validationFailed', async () => {
mockSyncService.downloadRemoteOps.and.returnValue(
Promise.resolve({
kind: 'no_new_ops' as const,
validationFailed: true,
}),
);
// #7330: the USE_REMOTE conflict-resolution path returns
// `kind: 'no_new_ops'` after applying remote state. If validation
// failed during that apply, the latch is flipped — the wrapper must
// surface this as ERROR rather than IN_SYNC.
it('should set ERROR when downloadRemoteOps returns no_new_ops AND latch is flipped', async () => {
const latch = TestBed.inject(SyncSessionValidationService);
mockSyncService.downloadRemoteOps.and.callFake(async () => {
latch.setFailed();
return { kind: 'no_new_ops' as const };
});
const result = await service.sync();
@ -1267,7 +1260,7 @@ describe('SyncWrapperService', () => {
mockSyncService.forceDownloadRemoteState = jasmine
.createSpy('forceDownloadRemoteState')
.and.resolveTo({ validationFailed: false });
.and.resolveTo();
const result = await service.sync();
@ -1330,8 +1323,8 @@ describe('SyncWrapperService', () => {
});
// Issue #7330: even when forceDownloadRemoteState succeeds, if it
// reports validationFailed: true, the wrapper must not claim IN_SYNC.
it('should return HANDLED_ERROR with ERROR status when forceDownloadRemoteState reports validationFailed', async () => {
// flips the session-validation latch, the wrapper must not claim IN_SYNC.
it('should return HANDLED_ERROR with ERROR status when forceDownloadRemoteState flips the latch', async () => {
const conflictError = new LocalDataConflictError(
2,
{ tasks: [] },
@ -1343,9 +1336,12 @@ describe('SyncWrapperService', () => {
afterClosed: () => of('USE_REMOTE'),
} as any);
const latch = TestBed.inject(SyncSessionValidationService);
mockSyncService.forceDownloadRemoteState = jasmine
.createSpy('forceDownloadRemoteState')
.and.resolveTo({ validationFailed: true });
.and.callFake(async () => {
latch.setFailed();
});
const result = await service.sync();

81
src/app/imex/sync/sync-wrapper.service.ts
View file

@ -73,6 +73,7 @@ import { WsTriggeredDownloadService } from '../../op-log/sync/ws-triggered-downl
import { IS_ELECTRON } from '../../app.constants';
import { OperationLogStoreService } from '../../op-log/persistence/operation-log-store.service';
import { OperationLogSyncService } from '../../op-log/sync/operation-log-sync.service';
import { SyncSessionValidationService } from '../../op-log/sync/sync-session-validation.service';
import { WrappedProviderService } from '../../op-log/sync-providers/wrapped-provider.service';
import { SuperSyncProvider } from '../../op-log/sync-providers/super-sync/super-sync';
import { HydrationStateService } from '../../op-log/apply/hydration-state.service';
@ -109,6 +110,7 @@ export class SyncWrapperService {
private _wsDownloadService = inject(WsTriggeredDownloadService);
private _opLogStore = inject(OperationLogStoreService);
private _opLogSyncService = inject(OperationLogSyncService);
private _sessionValidation = inject(SyncSessionValidationService);
private _wrappedProvider = inject(WrappedProviderService);
private _hydrationState = inject(HydrationStateService);
@ -388,6 +390,12 @@ export class SyncWrapperService {
throw new Error('No Sync Provider for sync()');
}
// Reset the session-validation latch at the start of every sync. Any
// post-sync validation failure during this session (download, upload,
// piggyback, retry, USE_REMOTE force-download) flips the latch; the
// wrapper reads it once before claiming IN_SYNC. (#7330)
this._sessionValidation.reset();
try {
// PERF: For legacy sync providers (WebDAV, Dropbox, LocalFile), sync the vector clock
// from SUP_OPS to pf.META_MODEL before sync. This bridges the gap between the new
@ -464,23 +472,6 @@ export class SyncWrapperService {
return 'HANDLED_ERROR';
}
// Issue #7330: post-sync state validation failure must not be reported
// as IN_SYNC. Previously the validation result was discarded; the user
// saw a "State validation failed" snackbar yet the indicator still said
// IN_SYNC, masking real corruption.
//
// Hoisted above the LWW retry loop so the failure is recognised even
// when retries also exhaust (otherwise the retry-exhaustion branch
// below would return UNKNOWN_OR_CHANGED, masking validation failure).
// `no_new_ops` carries `validationFailed` in the USE_REMOTE conflict
// path (forceDownloadRemoteState).
const downloadValidationFailed =
(downloadResult.kind === 'ops_processed' ||
downloadResult.kind === 'no_new_ops') &&
downloadResult.validationFailed === true;
const uploadValidationFailed =
uploadResult.kind === 'completed' && uploadResult.validationFailed === true;
// 3. If LWW created local-win ops, upload them (with retry limit to prevent infinite loops)
const downloadLwwOps =
downloadResult.kind === 'ops_processed' ? downloadResult.localWinOpsCreated : 0;
@ -488,10 +479,6 @@ export class SyncWrapperService {
uploadResult.kind === 'completed' ? uploadResult.localWinOpsCreated : 0;
let lwwRetries = 0;
let pendingLwwOps = downloadLwwOps + uploadLwwOps;
// Retry-pass piggybacked downloads can themselves trigger validation;
// discarding the boolean would re-introduce the original IN_SYNC bug
// for the retry path. (#7330)
let reuploadValidationFailed = false;
while (pendingLwwOps > 0 && lwwRetries < MAX_LWW_REUPLOAD_RETRIES) {
lwwRetries++;
SyncLog.log(
@ -502,31 +489,18 @@ export class SyncWrapperService {
await this._opLogSyncService.uploadPendingOps(syncCapableProvider);
pendingLwwOps =
reuploadResult.kind === 'completed' ? reuploadResult.localWinOpsCreated : 0;
if (reuploadResult.kind === 'completed' && reuploadResult.validationFailed) {
reuploadValidationFailed = true;
}
}
if (pendingLwwOps > 0) {
SyncLog.warn(
`SyncWrapperService: LWW re-upload still has ${pendingLwwOps} pending ops after ` +
`${MAX_LWW_REUPLOAD_RETRIES} retries. Will retry on next sync.`,
);
// If validation failed at any point (initial download, initial upload,
// or retry), prefer ERROR over the softer UNKNOWN_OR_CHANGED —
// validation failure is more serious than unuploaded ops, and the
// user already saw a "State validation failed" snackbar. (#7521)
if (
downloadValidationFailed ||
uploadValidationFailed ||
reuploadValidationFailed
) {
// Issue #7521: validation failure is more serious than unuploaded
// ops — prefer ERROR over UNKNOWN_OR_CHANGED if the latch was
// flipped at any point during the session.
if (this._sessionValidation.hasFailed()) {
SyncLog.err(
'SyncWrapperService: Validation failed during sync (retry exhaustion path); reporting ERROR',
{
downloadValidationFailed,
uploadValidationFailed,
reuploadValidationFailed,
},
);
this._providerManager.setSyncStatus('ERROR');
return 'HANDLED_ERROR';
@ -536,18 +510,12 @@ export class SyncWrapperService {
return SyncStatus.UpdateRemote;
}
if (
downloadValidationFailed ||
uploadValidationFailed ||
reuploadValidationFailed
) {
// Issue #7330: post-sync state validation failure must not be reported
// as IN_SYNC. The latch is flipped by every validation site; we read it
// once here before claiming IN_SYNC.
if (this._sessionValidation.hasFailed()) {
SyncLog.err(
'SyncWrapperService: Post-sync state validation failed, not marking as IN_SYNC',
{
downloadValidationFailed,
uploadValidationFailed,
reuploadValidationFailed,
},
);
this._providerManager.setSyncStatus('ERROR');
return 'HANDLED_ERROR';
@ -877,6 +845,9 @@ export class SyncWrapperService {
SyncLog.log('SyncWrapperService: forceDownload called - downloading remote state');
await this.runWithSyncBlocked(async () => {
// Reset session-validation latch — read after forceDownloadRemoteState
// returns so a corrupt downloaded state is reported as ERROR. (#7330)
this._sessionValidation.reset();
try {
const rawProvider = this._providerManager.getActiveProvider();
const syncCapableProvider =
@ -889,9 +860,8 @@ export class SyncWrapperService {
return;
}
const downloadResult =
await this._opLogSyncService.forceDownloadRemoteState(syncCapableProvider);
if (downloadResult.validationFailed) {
await this._opLogSyncService.forceDownloadRemoteState(syncCapableProvider);
if (this._sessionValidation.hasFailed()) {
SyncLog.err(
'SyncWrapperService: Force download applied but post-sync validation failed; reporting ERROR',
);
@ -1209,13 +1179,14 @@ export class SyncWrapperService {
this._providerManager.setSyncStatus('IN_SYNC');
return SyncStatus.InSync;
} else if (resolution === 'USE_REMOTE') {
// User chose to discard local data and download remote
// User chose to discard local data and download remote.
// Reset latch — read after forceDownloadRemoteState returns. (#7330)
SyncLog.log(
'SyncWrapperService: User chose USE_REMOTE - downloading remote state, discarding local',
);
const downloadResult =
await this._opLogSyncService.forceDownloadRemoteState(syncCapableProvider);
if (downloadResult.validationFailed) {
this._sessionValidation.reset();
await this._opLogSyncService.forceDownloadRemoteState(syncCapableProvider);
if (this._sessionValidation.hasFailed()) {
SyncLog.err(
'SyncWrapperService: USE_REMOTE applied but post-sync validation failed; reporting ERROR',
);

29
src/app/op-log/core/types/sync-results.types.ts
View file

@ -144,18 +144,15 @@ export interface UploadOptions {
/**
* Result from a download operation, used for concurrent modification resolution.
*
* Validation failure (if any during a nested download) is surfaced via the
* SyncSessionValidationService latch — the wrapper reads it once before
* deciding IN_SYNC vs ERROR. (#7330)
*/
export interface DownloadResultForRejection {
newOpsCount: number;
allOpClocks?: VectorClock[];
snapshotVectorClock?: VectorClock;
/**
* True when the underlying download ran post-sync validation and the
* validation failed. Propagated through `RejectionHandlingResult` to
* `OperationLogSyncService.uploadPendingOps` so the wrapper can refuse
* IN_SYNC. Replaces a closure-smuggled boolean. Issue #7330.
*/
validationFailed?: boolean;
}
/**
@ -188,13 +185,6 @@ export type DownloadOutcome =
kind: 'no_new_ops';
allOpClocks?: VectorClock[];
snapshotVectorClock?: VectorClock;
/**
* Set when this terminal state was reached after a USE_REMOTE
* conflict-resolution that ran `validateAfterSync` against the
* downloaded state and the validation reported corruption. The
* wrapper must not claim IN_SYNC. Issue #7330.
*/
validationFailed?: boolean;
}
| {
/** Incremental ops were downloaded and processed. */
@ -203,12 +193,6 @@ export type DownloadOutcome =
localWinOpsCreated: number;
allOpClocks?: VectorClock[];
snapshotVectorClock?: VectorClock;
/**
* `validateAfterSync` returned `false` while applying these ops — state
* may be inconsistent. Issue #7330: previously discarded; now surfaced
* so the sync wrapper can refuse to claim IN_SYNC.
*/
validationFailed?: boolean;
}
| {
/** File-based snapshot was hydrated (fresh download from file provider). */
@ -240,11 +224,6 @@ export type UploadOutcome =
permanentRejectionCount: number;
hasMorePiggyback: boolean;
rejectedOps: RejectedOpInfo[];
/**
* `validateAfterSync` returned `false` while processing piggybacked ops
* — state may be inconsistent. See `DownloadOutcome.validationFailed`.
*/
validationFailed?: boolean;
}
| {
/** User cancelled a piggybacked SYNC_IMPORT conflict dialog. */

14
src/app/op-log/sync/conflict-resolution.service.spec.ts
View file

@ -1209,7 +1209,7 @@ describe('ConflictResolutionService', () => {
it('should return { localWinOpsCreated: 0 } when no conflicts', async () => {
const result = await service.autoResolveConflictsLWW([]);
// Early-exit path: no validation runs, validationFailed is absent.
// Early-exit path: no validation runs.
expect(result).toEqual({ localWinOpsCreated: 0 });
});
@ -1225,7 +1225,7 @@ describe('ConflictResolutionService', () => {
const result = await service.autoResolveConflictsLWW([], nonConflicting);
expect(result).toEqual({ localWinOpsCreated: 0, validationFailed: false });
expect(result).toEqual({ localWinOpsCreated: 0 });
});
it('should return { localWinOpsCreated: 0 } when remote wins all conflicts', async () => {
@ -1244,7 +1244,7 @@ describe('ConflictResolutionService', () => {
const result = await service.autoResolveConflictsLWW(conflicts);
expect(result).toEqual({ localWinOpsCreated: 0, validationFailed: false });
expect(result).toEqual({ localWinOpsCreated: 0 });
});
it('should return { localWinOpsCreated: 1 } when local wins one conflict', async () => {
@ -1264,7 +1264,7 @@ describe('ConflictResolutionService', () => {
const result = await service.autoResolveConflictsLWW(conflicts);
expect(result).toEqual({ localWinOpsCreated: 1, validationFailed: false });
expect(result).toEqual({ localWinOpsCreated: 1 });
});
it('should return correct count when multiple local wins', async () => {
@ -1298,7 +1298,7 @@ describe('ConflictResolutionService', () => {
const result = await service.autoResolveConflictsLWW(conflicts);
expect(result).toEqual({ localWinOpsCreated: 2, validationFailed: false });
expect(result).toEqual({ localWinOpsCreated: 2 });
});
it('should return correct count for mixed local/remote wins', async () => {
@ -1353,7 +1353,7 @@ describe('ConflictResolutionService', () => {
const result = await service.autoResolveConflictsLWW(conflicts);
// Only 1 local win out of 3 conflicts
expect(result).toEqual({ localWinOpsCreated: 1, validationFailed: false });
expect(result).toEqual({ localWinOpsCreated: 1 });
});
it('should return 0 when local wins but entity not found', async () => {
@ -1374,7 +1374,7 @@ describe('ConflictResolutionService', () => {
const result = await service.autoResolveConflictsLWW(conflicts);
// No op created because entity not found
expect(result).toEqual({ localWinOpsCreated: 0, validationFailed: false });
expect(result).toEqual({ localWinOpsCreated: 0 });
});
});

12
src/app/op-log/sync/conflict-resolution.service.ts
View file

@ -18,6 +18,7 @@ import { firstValueFrom } from 'rxjs';
import { SnackService } from '../../core/snack/snack.service';
import { T } from '../../t.const';
import { ValidateStateService } from '../validation/validate-state.service';
import { SyncSessionValidationService } from './sync-session-validation.service';
import { MAX_CONFLICT_RETRY_ATTEMPTS } from '../core/operation-log.const';
import {
compareVectorClocks,
@ -85,6 +86,7 @@ export class ConflictResolutionService {
private opLogStore = inject(OperationLogStoreService);
private snackService = inject(SnackService);
private validateStateService = inject(ValidateStateService);
private sessionValidation = inject(SyncSessionValidationService);
private clientIdProvider = inject(CLIENT_ID_PROVIDER);
// ═══════════════════════════════════════════════════════════════════════════
@ -344,7 +346,7 @@ export class ConflictResolutionService {
async autoResolveConflictsLWW(
conflicts: EntityConflict[],
nonConflictingOps: Operation[] = [],
): Promise<{ localWinOpsCreated: number; validationFailed?: boolean }> {
): Promise<{ localWinOpsCreated: number }> {
if (conflicts.length === 0 && nonConflictingOps.length === 0) {
return { localWinOpsCreated: 0 };
}
@ -587,13 +589,13 @@ export class ConflictResolutionService {
// ─────────────────────────────────────────────────────────────────────────
// STEP 8: Validate and repair state after resolution
// Validation failure flips the SyncSessionValidationService latch — the
// wrapper reads it before deciding IN_SYNC vs ERROR. (#7330)
// ─────────────────────────────────────────────────────────────────────────
const isValid = await this._validateAndRepairAfterResolution();
if (!isValid) this.sessionValidation.setFailed();
return {
localWinOpsCreated: newLocalWinOps.length,
validationFailed: !isValid,
};
return { localWinOpsCreated: newLocalWinOps.length };
}
/**

86
src/app/op-log/sync/operation-log-sync.service.spec.ts
View file

@ -7,6 +7,7 @@ import { VectorClockService } from './vector-clock.service';
import { OperationApplierService } from '../apply/operation-applier.service';
import { ConflictResolutionService } from './conflict-resolution.service';
import { ValidateStateService } from '../validation/validate-state.service';
import { SyncSessionValidationService } from './sync-session-validation.service';
import { RepairOperationService } from '../validation/repair-operation.service';
import { OperationLogUploadService } from './operation-log-upload.service';
import { OperationLogDownloadService } from './operation-log-download.service';
@ -534,32 +535,34 @@ describe('OperationLogSyncService', () => {
rejectedOpsHandlerServiceSpy.handleRejectedOps.and.callFake(
async (_ops, callback) => {
// Mirror what the real handler does: invoke the callback,
// observe its validationFailed, propagate to result.
const downloadResult = await callback?.();
return {
mergedOpsCreated: 0,
permanentRejectionCount: 0,
validationFailed: downloadResult?.validationFailed === true,
};
// Real handler invokes the download callback for concurrent-mod
// resolution. The latch is flipped inside the nested download's
// validateAfterSync — here we just exercise the call.
await callback?.();
return { mergedOpsCreated: 0, permanentRejectionCount: 0 };
},
);
spyOn(service, 'downloadRemoteOps').and.returnValue(
Promise.resolve({
// Simulate the nested download triggering validation failure by
// flipping the latch directly, which the real
// RemoteOpsProcessingService.validateAfterSync would do.
const latch = TestBed.inject(SyncSessionValidationService);
latch.reset();
spyOn(service, 'downloadRemoteOps').and.callFake(async () => {
latch.setFailed();
return {
kind: 'ops_processed' as const,
newOpsCount: 1,
localWinOpsCreated: 0,
validationFailed: true,
}),
);
};
});
const result = await service.uploadPendingOps(mockProvider);
await service.uploadPendingOps(mockProvider);
expect(result.kind).toBe('completed');
if (result.kind === 'completed') {
expect(result.validationFailed).toBe(true);
}
// The latch is the canonical signal that reaches the wrapper. The
// upload result no longer carries validationFailed — that field is
// gone (#7330 simplification).
expect(latch.hasFailed()).toBe(true);
});
});
});
@ -1807,9 +1810,14 @@ describe('OperationLogSyncService', () => {
});
// Issue #7330: post-sync validation failure must be surfaced so
// SyncWrapperService can refuse IN_SYNC. Previously the result was
// discarded and a corrupt USE_REMOTE state silently looked clean.
it('returns validationFailed: true when processRemoteOps signals validation failure', async () => {
// SyncWrapperService can refuse IN_SYNC. After the latch refactor,
// failure flows via SyncSessionValidationService rather than the return
// shape — but processRemoteOps is the layer that flips the latch via
// validateAfterSync. We assert here that forceDownloadRemoteState
// delegates through processRemoteOps, leaving the latch intact for the
// wrapper to read. (The flip itself is unit-tested in
// remote-ops-processing.service.spec.ts.)
it('forceDownloadRemoteState invokes processRemoteOps with skipConflictDetection', async () => {
const mockOps: Operation[] = [
{
id: 'op1',
@ -1833,38 +1841,16 @@ describe('OperationLogSyncService', () => {
latestServerSeq: 1,
});
remoteOpsProcessingServiceSpy.processRemoteOps.and.resolveTo({
localWinOpsCreated: 0,
allOpsFilteredBySyncImport: false,
filteredOpCount: 0,
isLocalUnsyncedImport: false,
validationFailed: true,
});
const mockProvider = {
supportsOperationSync: true,
setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(),
} as any;
const result = await service.forceDownloadRemoteState(mockProvider);
expect(result).toEqual({ validationFailed: true });
});
it('returns validationFailed: false when no ops are downloaded', async () => {
downloadServiceSpy.downloadRemoteOps.and.resolveTo({
newOps: [],
needsFullStateUpload: false,
success: true,
failedFileCount: 0,
});
const mockProvider = {
supportsOperationSync: true,
setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(),
} as any;
const result = await service.forceDownloadRemoteState(mockProvider);
expect(result).toEqual({ validationFailed: false });
await service.forceDownloadRemoteState(mockProvider);
expect(remoteOpsProcessingServiceSpy.processRemoteOps).toHaveBeenCalledWith(
mockOps,
{ skipConflictDetection: true },
);
});
});
@ -2761,9 +2747,7 @@ describe('OperationLogSyncService', () => {
syncImportConflictDialogServiceSpy.showConflictDialog.and.resolveTo('USE_REMOTE');
const forceDownloadSpy = spyOn(service, 'forceDownloadRemoteState').and.resolveTo({
validationFailed: false,
});
const forceDownloadSpy = spyOn(service, 'forceDownloadRemoteState').and.resolveTo();
const mockProvider = {
isReady: () => Promise.resolve(true),

87
src/app/op-log/sync/operation-log-sync.service.ts
View file

@ -282,11 +282,9 @@ export class OperationLogSyncService {
// even if processing throws. Otherwise rejected ops remain in pending
// state and get re-uploaded infinitely.
let localWinOpsCreated = 0;
let piggybackValidationFailed = false;
let rejectionResult: RejectionHandlingResult = {
mergedOpsCreated: 0,
permanentRejectionCount: 0,
validationFailed: false,
};
if (result.piggybackedOps.length > 0) {
@ -322,12 +320,12 @@ export class OperationLogSyncService {
},
'OperationLogSyncService (piggybacked full-state op)',
);
if (conflictResult.resolution === 'CANCEL') {
if (conflictResult === 'CANCEL') {
return { kind: 'cancelled' };
}
// USE_LOCAL or USE_REMOTE was handled — report as completed with no further work.
// Surface USE_REMOTE post-sync validation failure so SyncWrapperService can
// refuse IN_SYNC. Issue #7330.
// Validation failure (if any during USE_REMOTE force-download) is on the
// session-validation latch already; the wrapper reads it. (#7330)
return {
kind: 'completed',
uploadedCount: result.uploadedCount,
@ -336,8 +334,6 @@ export class OperationLogSyncService {
permanentRejectionCount: 0,
hasMorePiggyback: false,
rejectedOps: [],
validationFailed:
piggybackValidationFailed || conflictResult.validationFailed,
};
} else {
OpLog.normal(
@ -352,7 +348,7 @@ export class OperationLogSyncService {
result.piggybackedOps,
);
localWinOpsCreated = processResult.localWinOpsCreated;
if (processResult.validationFailed) piggybackValidationFailed = true;
// Validation failure (if any) is on the session-validation latch.
}
// STEP 2: Handle server-rejected operations
@ -366,21 +362,16 @@ export class OperationLogSyncService {
forceFromSeq0?: boolean;
}): Promise<DownloadResultForRejection> => {
const outcome = await this.downloadRemoteOps(syncProvider, downloadOptions);
// Validation failure (if any during the nested download) is on the
// session-validation latch — no need to thread the boolean back. (#7330)
switch (outcome.kind) {
case 'ops_processed':
return {
newOpsCount: outcome.newOpsCount,
allOpClocks: outcome.allOpClocks,
snapshotVectorClock: outcome.snapshotVectorClock,
validationFailed: outcome.validationFailed,
};
case 'no_new_ops':
return {
newOpsCount: 0,
allOpClocks: outcome.allOpClocks,
snapshotVectorClock: outcome.snapshotVectorClock,
validationFailed: outcome.validationFailed,
};
case 'snapshot_hydrated':
return {
newOpsCount: 0,
@ -424,8 +415,6 @@ export class OperationLogSyncService {
permanentRejectionCount: rejectionResult.permanentRejectionCount,
hasMorePiggyback: result.hasMorePiggyback ?? false,
rejectedOps: result.rejectedOps,
validationFailed:
piggybackValidationFailed || rejectionResult.validationFailed === true,
};
}
@ -765,14 +754,12 @@ export class OperationLogSyncService {
},
'OperationLogSyncService (incoming full-state op)',
);
if (conflictResult.resolution === 'CANCEL') {
if (conflictResult === 'CANCEL') {
return { kind: 'cancelled' };
}
// Surface USE_REMOTE post-sync validation failure. Issue #7330.
return {
kind: 'no_new_ops',
validationFailed: conflictResult.validationFailed,
};
// Validation failure (if any during USE_REMOTE force-download) is on
// the session-validation latch — wrapper reads it. (#7330)
return { kind: 'no_new_ops' };
} else {
OpLog.normal(
`OperationLogSyncService: Accepting incoming ${incomingFullStateOp.opType} from client ` +
@ -815,14 +802,12 @@ export class OperationLogSyncService {
},
'OperationLogSyncService (local SYNC_IMPORT filters remote)',
);
if (conflictResult.resolution === 'CANCEL') {
if (conflictResult === 'CANCEL') {
return { kind: 'cancelled' };
}
// Surface USE_REMOTE post-sync validation failure. Issue #7330.
return {
kind: 'no_new_ops',
validationFailed: conflictResult.validationFailed,
};
// Validation failure (if any during USE_REMOTE force-download) is on
// the session-validation latch — wrapper reads it. (#7330)
return { kind: 'no_new_ops' };
} else if (
processResult.allOpsFilteredBySyncImport &&
processResult.filteredOpCount > 0
@ -861,7 +846,6 @@ export class OperationLogSyncService {
localWinOpsCreated: processResult.localWinOpsCreated,
allOpClocks: result.allOpClocks,
snapshotVectorClock: result.snapshotVectorClock,
validationFailed: processResult.validationFailed,
};
}
@ -881,10 +865,7 @@ export class OperationLogSyncService {
syncProvider: OperationSyncCapable,
dialogData: SyncImportConflictData,
logPrefix: string,
): Promise<{
resolution: SyncImportConflictResolution;
validationFailed: boolean;
}> {
): Promise<SyncImportConflictResolution> {
const resolution =
await this.syncImportConflictDialogService.showConflictDialog(dialogData);
@ -892,21 +873,18 @@ export class OperationLogSyncService {
case 'USE_LOCAL':
OpLog.normal(`${logPrefix}: User chose USE_LOCAL. Force uploading local state.`);
await this.forceUploadLocalState(syncProvider);
return { resolution: 'USE_LOCAL', validationFailed: false };
case 'USE_REMOTE': {
return 'USE_LOCAL';
case 'USE_REMOTE':
OpLog.normal(
`${logPrefix}: User chose USE_REMOTE. Force downloading remote state.`,
);
const downloadResult = await this.forceDownloadRemoteState(syncProvider);
return {
resolution: 'USE_REMOTE',
validationFailed: downloadResult.validationFailed,
};
}
// Validation failure (if any) is on the session-validation latch.
await this.forceDownloadRemoteState(syncProvider);
return 'USE_REMOTE';
case 'CANCEL':
default:
OpLog.normal(`${logPrefix}: User cancelled SYNC_IMPORT conflict resolution.`);
return { resolution: 'CANCEL', validationFailed: false };
return 'CANCEL';
}
}
@ -977,13 +955,10 @@ export class OperationLogSyncService {
*
* @param syncProvider - The sync provider to download from
*/
async forceDownloadRemoteState(
syncProvider: OperationSyncCapable,
): Promise<{ validationFailed: boolean }> {
async forceDownloadRemoteState(syncProvider: OperationSyncCapable): Promise<void> {
OpLog.warn(
'OperationLogSyncService: Force downloading remote state - clearing local import and unsynced ops.',
);
let validationFailed = false;
// IMPORTANT: Clear local full-state ops (SYNC_IMPORT, BACKUP_IMPORT, REPAIR)
// This is critical - if the user chose USE_REMOTE, we must not filter incoming
@ -1058,7 +1033,7 @@ export class OperationLogSyncService {
OpLog.normal(
'OperationLogSyncService: Force download snapshot hydration complete.',
);
return { validationFailed };
return;
}
if (result.newOps.length > 0) {
@ -1087,18 +1062,13 @@ export class OperationLogSyncService {
await new Promise((resolve) => setTimeout(resolve, 0));
}
// Process all remote ops (no confirmation needed - user already chose USE_REMOTE)
// Process all remote ops (no confirmation needed - user already chose USE_REMOTE).
// Skip conflict detection because the NgRx store was just reset to empty state,
// which causes all entities to appear missing and CONCURRENT ops to be discarded.
// Capture validationFailed so callers (and ultimately SyncWrapperService) can
// refuse IN_SYNC when the downloaded state is corrupt. Issue #7330.
const processResult = await this.remoteOpsProcessingService.processRemoteOps(
result.newOps,
{ skipConflictDetection: true },
);
if (processResult.validationFailed) {
validationFailed = true;
}
// Validation failure is surfaced via the session-validation latch. (#7330)
await this.remoteOpsProcessingService.processRemoteOps(result.newOps, {
skipConflictDetection: true,
});
// Update lastServerSeq
if (result.latestServerSeq !== undefined) {
@ -1109,7 +1079,6 @@ export class OperationLogSyncService {
OpLog.normal(
`OperationLogSyncService: Force download complete. Processed ${result.newOps.length} ops.`,
);
return { validationFailed };
}
/**

6
src/app/op-log/sync/rejected-ops-handler.service.spec.ts
View file

@ -446,11 +446,7 @@ describe('RejectedOpsHandlerService', () => {
[remoteClock],
{ snapshot: 1 },
);
expect(result).toEqual({
mergedOpsCreated: 1,
permanentRejectionCount: 0,
validationFailed: false,
});
expect(result).toEqual({ mergedOpsCreated: 1, permanentRejectionCount: 0 });
});
it('should pass existingClock from rejection to superseded resolver (FIX: encryption conflict loop)', async () => {

22
src/app/op-log/sync/rejected-ops-handler.service.ts
View file

@ -18,18 +18,15 @@ export type {
} from '../core/types/sync-results.types';
/**
* Result of handling rejected operations.
* Result of handling rejected operations. Validation failure (if any during
* a nested download) is surfaced via the SyncSessionValidationService latch
* — the wrapper reads it once before deciding IN_SYNC vs ERROR. (#7330)
*/
export interface RejectionHandlingResult {
/** Number of merged ops created from conflict resolution (these need to be uploaded) */
mergedOpsCreated: number;
/** Number of operations that were permanently rejected (validation errors, etc.) */
permanentRejectionCount: number;
/**
* True when a nested download triggered by concurrent-modification resolution
* ran post-sync validation and the validation failed. Issue #7330.
*/
validationFailed?: boolean;
}
/**
@ -192,7 +189,6 @@ export class RejectedOpsHandlerService {
// For concurrent modifications: try download first, then resolve locally if needed
let retryExceededCount = 0;
let validationFailed = false;
if (concurrentModificationOps.length > 0 && downloadCallback) {
const result = await this._resolveConcurrentModifications(
concurrentModificationOps,
@ -200,13 +196,11 @@ export class RejectedOpsHandlerService {
);
mergedOpsCreated = result.mergedOpsCreated;
retryExceededCount = result.retryExceededCount;
validationFailed = result.validationFailed;
}
return {
mergedOpsCreated,
permanentRejectionCount: permanentlyRejectedOps.length + retryExceededCount,
validationFailed,
};
}
@ -223,10 +217,8 @@ export class RejectedOpsHandlerService {
): Promise<{
mergedOpsCreated: number;
retryExceededCount: number;
validationFailed: boolean;
}> {
let mergedOpsCreated = 0;
let validationFailed = false;
// Check resolution attempt counts per entity to prevent infinite loops.
// When vector clock pruning makes it impossible to create a dominating clock,
@ -288,7 +280,6 @@ export class RejectedOpsHandlerService {
return {
mergedOpsCreated: 0,
retryExceededCount: opsExceededRetries.length,
validationFailed: false,
};
}
@ -298,9 +289,10 @@ export class RejectedOpsHandlerService {
);
try {
// Try to download new remote ops - if there are any, conflict detection will handle them
// Try to download new remote ops - if there are any, conflict detection will handle them.
// Validation failure (if any during the nested download) is on the
// session-validation latch — wrapper reads it. (#7330)
const downloadResult = await downloadCallback();
if (downloadResult.validationFailed) validationFailed = true;
// Helper to check which ops are still pending, preserving existingClock from rejection
const getStillPendingOps = async (): Promise<
@ -344,7 +336,6 @@ export class RejectedOpsHandlerService {
);
const forceDownloadResult = await downloadCallback({ forceFromSeq0: true });
if (forceDownloadResult.validationFailed) validationFailed = true;
// Use the clocks from force download to resolve superseded ops
// Also merge in entity clocks from server rejection responses
@ -440,7 +431,6 @@ export class RejectedOpsHandlerService {
return {
mergedOpsCreated,
retryExceededCount: opsExceededRetries.length,
validationFailed,
};
}

17
src/app/op-log/sync/remote-ops-processing.service.spec.ts
View file

@ -11,6 +11,7 @@ import { VectorClockService } from './vector-clock.service';
import { OperationApplierService } from '../apply/operation-applier.service';
import { ConflictResolutionService } from './conflict-resolution.service';
import { ValidateStateService } from '../validation/validate-state.service';
import { SyncSessionValidationService } from './sync-session-validation.service';
import { LockService } from './lock.service';
import { OperationLogCompactionService } from '../persistence/operation-log-compaction.service';
import { SyncImportFilterService } from './sync-import-filter.service';
@ -1297,7 +1298,7 @@ describe('RemoteOpsProcessingService', () => {
expect(result).toBe(false);
});
it('processRemoteOps surfaces validationFailed=true when validation fails on the no-conflict path', async () => {
it('processRemoteOps flips the session-validation latch when validation fails on the no-conflict path', async () => {
const remoteOps: Operation[] = [{ id: 'op1', schemaVersion: 1 } as Operation];
opLogStoreSpy.getUnsynced.and.returnValue(Promise.resolve([]));
@ -1319,12 +1320,14 @@ describe('RemoteOpsProcessingService', () => {
});
validateStateServiceSpy.validateAndRepairCurrentState.and.resolveTo(false);
const result = await service.processRemoteOps(remoteOps);
const latch = TestBed.inject(SyncSessionValidationService);
latch.reset();
await service.processRemoteOps(remoteOps);
expect(result.validationFailed).toBe(true);
expect(latch.hasFailed()).toBe(true);
});
it('processRemoteOps surfaces validationFailed=false when validation succeeds', async () => {
it('processRemoteOps leaves the latch reset when validation succeeds', async () => {
const remoteOps: Operation[] = [{ id: 'op1', schemaVersion: 1 } as Operation];
opLogStoreSpy.getUnsynced.and.returnValue(Promise.resolve([]));
@ -1346,9 +1349,11 @@ describe('RemoteOpsProcessingService', () => {
});
validateStateServiceSpy.validateAndRepairCurrentState.and.resolveTo(true);
const result = await service.processRemoteOps(remoteOps);
const latch = TestBed.inject(SyncSessionValidationService);
latch.reset();
await service.processRemoteOps(remoteOps);
expect(result.validationFailed).toBeFalsy();
expect(latch.hasFailed()).toBe(false);
});
});
});

40
src/app/op-log/sync/remote-ops-processing.service.ts
View file

@ -11,6 +11,7 @@ import { OpLog } from '../../core/log';
import { OperationApplierService } from '../apply/operation-applier.service';
import { ConflictResolutionService } from './conflict-resolution.service';
import { ValidateStateService } from '../validation/validate-state.service';
import { SyncSessionValidationService } from './sync-session-validation.service';
import { VectorClockService } from './vector-clock.service';
import {
MAX_VERSION_SKIP,
@ -46,6 +47,7 @@ export class RemoteOpsProcessingService {
private operationApplier = inject(OperationApplierService);
private conflictResolutionService = inject(ConflictResolutionService);
private validateStateService = inject(ValidateStateService);
private sessionValidation = inject(SyncSessionValidationService);
private vectorClockService = inject(VectorClockService);
private schemaMigrationService = inject(SchemaMigrationService);
private snackService = inject(SnackService);
@ -85,15 +87,12 @@ export class RemoteOpsProcessingService {
filteredOpCount: number;
filteringImport?: Operation;
isLocalUnsyncedImport: boolean;
/**
* Issue #7330: surfaces a `false` from `validateAfterSync` (or the
* conflict-resolution variant) so the sync wrapper can refuse to claim
* IN_SYNC when post-apply state validation fails. Absent on early-exit
* paths that never reach validation (e.g. version-mismatch, all ops
* filtered by SYNC_IMPORT) — those are not validation failures.
*/
validationFailed?: boolean;
}> {
// Validation failure surfaces via the SyncSessionValidationService latch
// (#7330). `validateAfterSync` and the conflict-resolution validation path
// both flip the latch on failure; the sync wrapper reads it once before
// deciding IN_SYNC vs ERROR. No need to thread the boolean through this
// return shape.
// ─────────────────────────────────────────────────────────────────────────
// STEP 1: Schema Migration (Receiver-Side)
// Migrate ops from older schema versions to current version.
@ -267,13 +266,12 @@ export class RemoteOpsProcessingService {
// Local synced ops are NOT replayed - the import is an explicit user action
// to restore all clients to a specific point in time.
const isValid = await this.validateAfterSync();
await this.validateAfterSync();
return {
localWinOpsCreated: 0,
allOpsFilteredBySyncImport: false,
filteredOpCount: 0,
isLocalUnsyncedImport: false,
validationFailed: !isValid,
};
}
@ -299,13 +297,12 @@ export class RemoteOpsProcessingService {
`Applying ${validOps.length} ops directly.`,
);
await this.applyNonConflictingOps(validOps);
const isValid = await this.validateAfterSync();
await this.validateAfterSync();
return {
localWinOpsCreated: 0,
allOpsFilteredBySyncImport: false,
filteredOpCount: 0,
isLocalUnsyncedImport: false,
validationFailed: !isValid,
};
}
@ -322,7 +319,6 @@ export class RemoteOpsProcessingService {
// This ensures no NEW writes can start while we read the frontier,
// detect conflicts, AND apply resolutions.
let localWinOpsCreated = 0;
let validationFailed = false;
await this.lockService.request(LOCK_NAMES.OPERATION_LOG, async () => {
const appliedFrontierByEntity = await this.vectorClockService.getEntityFrontier();
const conflictResult = await this.detectConflicts(
@ -342,14 +338,14 @@ export class RemoteOpsProcessingService {
`RemoteOpsProcessingService: Detected ${conflicts.length} conflicts. Auto-resolving with LWW.`,
conflicts,
);
// Auto-resolve conflicts using Last-Write-Wins strategy
// Piggyback non-conflicting ops so they're applied with resolved conflicts
// Auto-resolve conflicts using Last-Write-Wins strategy.
// Piggyback non-conflicting ops so they're applied with resolved conflicts.
// Validation failure is surfaced via the session-validation latch.
const lwwResult = await this.conflictResolutionService.autoResolveConflictsLWW(
conflicts,
nonConflicting,
);
localWinOpsCreated = lwwResult.localWinOpsCreated;
if (lwwResult.validationFailed) validationFailed = true;
return;
}
@ -358,8 +354,7 @@ export class RemoteOpsProcessingService {
// ─────────────────────────────────────────────────────────────────────────
if (nonConflicting.length > 0) {
await this.applyNonConflictingOps(nonConflicting, true);
const isValid = await this.validateAfterSync(true); // Inside sp_op_log lock
if (!isValid) validationFailed = true;
await this.validateAfterSync(true); // Inside sp_op_log lock
}
});
return {
@ -367,7 +362,6 @@ export class RemoteOpsProcessingService {
allOpsFilteredBySyncImport: false,
filteredOpCount: 0,
isLocalUnsyncedImport: false,
validationFailed,
};
}
@ -652,19 +646,17 @@ export class RemoteOpsProcessingService {
* @param callerHoldsLock - If true, skip lock acquisition in repair operation.
* Pass true when calling from within the sp_op_log lock.
* @returns `true` if state is valid (or was successfully repaired), `false`
* otherwise. Issue #7330: callers must surface a `false` up to the
* sync wrapper so it can avoid claiming IN_SYNC.
* otherwise. On failure the SyncSessionValidationService latch is
* flipped so the wrapper can refuse to claim IN_SYNC (#7330).
*/
async validateAfterSync(callerHoldsLock: boolean = false): Promise<boolean> {
// FIX #6571: Check and surface validation result.
// Previously, the boolean return was discarded — validation failures
// were invisible and sync reported IN_SYNC despite invalid state.
const isValid = await this.validateStateService.validateAndRepairCurrentState(
'sync',
{ callerHoldsLock },
);
if (!isValid) {
OpLog.err('RemoteOpsProcessingService: State validation failed after sync');
this.sessionValidation.setFailed();
this.snackService.open({
type: 'ERROR',
msg: T.F.SYNC.S.SYNC_VALIDATION_FAILED,

40
src/app/op-log/sync/sync-session-validation.service.spec.ts Normal file
View file

@ -0,0 +1,40 @@
import { TestBed } from '@angular/core/testing';
import { SyncSessionValidationService } from './sync-session-validation.service';
describe('SyncSessionValidationService', () => {
let service: SyncSessionValidationService;
beforeEach(() => {
TestBed.configureTestingModule({});
service = TestBed.inject(SyncSessionValidationService);
});
it('starts in the not-failed state', () => {
expect(service.hasFailed()).toBe(false);
});
it('hasFailed() reports true after setFailed()', () => {
service.setFailed();
expect(service.hasFailed()).toBe(true);
});
it('reset() clears the latch', () => {
service.setFailed();
service.reset();
expect(service.hasFailed()).toBe(false);
});
it('setFailed() is idempotent', () => {
service.setFailed();
service.setFailed();
expect(service.hasFailed()).toBe(true);
});
it('latch persists across hasFailed() reads until reset', () => {
service.setFailed();
expect(service.hasFailed()).toBe(true);
expect(service.hasFailed()).toBe(true);
service.reset();
expect(service.hasFailed()).toBe(false);
});
});

51
src/app/op-log/sync/sync-session-validation.service.ts Normal file
View file

@ -0,0 +1,51 @@
import { Injectable } from '@angular/core';
/**
* Session-scoped latch that records whether post-sync state validation
* failed at any point during the current sync session.
*
* A "sync session" is a single top-level sync operation:
* - `SyncWrapperService.sync()`
* - `SyncWrapperService._forceDownload()`
* - `SyncWrapperService.resolveSyncConflict()` (USE_REMOTE branch)
*
* These entry points are serialised by the wrapper's global lock, so a
* single mutable boolean is safe — there is never more than one session
* active at a time.
*
* ## Why a latch instead of typed return plumbing?
*
* Validation runs in several places (`RemoteOpsProcessingService.validateAfterSync`,
* `ConflictResolutionService._validateAndRepairAfterResolution`, etc.) called
* from many code paths (download, upload, piggyback, retry, USE_REMOTE force
* download). Threading a `validationFailed: boolean` through every result type
* meant adding the field to seven discriminated-union variants and remembering
* to forward it at every junction. A new variant or call site that forgot to
* carry the flag would silently let `IN_SYNC` ride over corrupt state.
*
* The latch collapses that to: validation site flips it, wrapper reads it
* once before deciding `IN_SYNC` vs `ERROR`. Issue #7330.
*
* ## Contract
*
* - `reset()` must be called by every wrapper entry point before doing work.
* - `setFailed()` is called by validation sites whenever
* `validateAndRepairCurrentState` returns `false`.
* - `hasFailed()` is read by the wrapper before claiming IN_SYNC.
*/
@Injectable({ providedIn: 'root' })
export class SyncSessionValidationService {
private _failed = false;
reset(): void {
this._failed = false;
}
setFailed(): void {
this._failed = true;
}
hasFailed(): boolean {
return this._failed;
}
}

179
src/app/op-log/testing/integration/post-sync-validation.integration.spec.ts Normal file
View file

@ -0,0 +1,179 @@
import { TestBed } from '@angular/core/testing';
import { Store } from '@ngrx/store';
import { TranslateService } from '@ngx-translate/core';
import { of, BehaviorSubject } from 'rxjs';
import { RemoteOpsProcessingService } from '../../sync/remote-ops-processing.service';
import { ConflictResolutionService } from '../../sync/conflict-resolution.service';
import { SyncSessionValidationService } from '../../sync/sync-session-validation.service';
import { ValidateStateService } from '../../validation/validate-state.service';
import { OperationLogStoreService } from '../../persistence/operation-log-store.service';
import { SnackService } from '../../../core/snack/snack.service';
import { CLIENT_ID_PROVIDER } from '../../util/client-id.provider';
/**
* Integration tests for the post-sync validation latch (#7330).
*
* Validates that validation failures from any code path inside the sync
* machinery flip `SyncSessionValidationService` so the wrapper can refuse
* IN_SYNC. Catches plumbing regressions where a future call site runs
* validation but forgets to surface the failure.
*
* The latch's per-method behavior is unit-tested in
* `sync-session-validation.service.spec.ts`; here we wire the real services
* that flip it (RemoteOpsProcessingService, ConflictResolutionService) and
* assert the latch is set on failure / clear on success.
*/
describe('Post-sync validation latch (#7330) — integration', () => {
let remoteOps: RemoteOpsProcessingService;
let conflictResolution: ConflictResolutionService;
let latch: SyncSessionValidationService;
let validateStateSpy: jasmine.SpyObj<ValidateStateService>;
let snackServiceSpy: jasmine.SpyObj<SnackService>;
let storeSpy: jasmine.SpyObj<Store>;
let opLogStoreSpy: jasmine.SpyObj<OperationLogStoreService>;
beforeEach(() => {
validateStateSpy = jasmine.createSpyObj('ValidateStateService', [
'validateAndRepairCurrentState',
]);
snackServiceSpy = jasmine.createSpyObj('SnackService', ['open']);
storeSpy = jasmine.createSpyObj('Store', ['dispatch', 'select']);
storeSpy.select.and.returnValue(of(undefined));
opLogStoreSpy = jasmine.createSpyObj('OperationLogStoreService', [
'getUnsynced',
'append',
'markApplied',
'getOpById',
'mergeRemoteOpClocks',
'appendWithVectorClockUpdate',
'markFailed',
]);
opLogStoreSpy.getUnsynced.and.resolveTo([]);
TestBed.configureTestingModule({
providers: [
SyncSessionValidationService,
// Real services that flip the latch:
RemoteOpsProcessingService,
ConflictResolutionService,
// Stubs at the validation boundary:
{ provide: ValidateStateService, useValue: validateStateSpy },
{ provide: SnackService, useValue: snackServiceSpy },
{ provide: Store, useValue: storeSpy },
{ provide: OperationLogStoreService, useValue: opLogStoreSpy },
{
provide: TranslateService,
useValue: { instant: (k: string): string => k },
},
{
provide: CLIENT_ID_PROVIDER,
useValue: new BehaviorSubject<string>('client-test'),
},
],
});
latch = TestBed.inject(SyncSessionValidationService);
remoteOps = TestBed.inject(RemoteOpsProcessingService);
conflictResolution = TestBed.inject(ConflictResolutionService);
latch.reset();
});
describe('RemoteOpsProcessingService.validateAfterSync', () => {
it('flips the latch when ValidateStateService reports failure', async () => {
validateStateSpy.validateAndRepairCurrentState.and.resolveTo(false);
expect(latch.hasFailed()).toBe(false);
await remoteOps.validateAfterSync();
expect(latch.hasFailed()).toBe(true);
});
it('leaves the latch reset when validation succeeds', async () => {
validateStateSpy.validateAndRepairCurrentState.and.resolveTo(true);
await remoteOps.validateAfterSync();
expect(latch.hasFailed()).toBe(false);
});
it('still flips the latch when callerHoldsLock is true (inside sp_op_log lock)', async () => {
validateStateSpy.validateAndRepairCurrentState.and.resolveTo(false);
await remoteOps.validateAfterSync(true);
expect(latch.hasFailed()).toBe(true);
expect(validateStateSpy.validateAndRepairCurrentState).toHaveBeenCalledWith(
'sync',
{ callerHoldsLock: true },
);
});
// Regression net for the sync wrapper's contract: if a future code path
// calls validateAfterSync and discards the boolean, the latch is still
// set — the wrapper will see it and refuse IN_SYNC.
it('flips the latch even when the caller discards the boolean return', async () => {
validateStateSpy.validateAndRepairCurrentState.and.resolveTo(false);
// Discard the return value (mirrors the post-#7330 callers).
void remoteOps.validateAfterSync();
await Promise.resolve();
expect(latch.hasFailed()).toBe(true);
});
});
describe('ConflictResolutionService validation path', () => {
it('flips the latch when post-LWW validation fails', async () => {
validateStateSpy.validateAndRepairCurrentState.and.resolveTo(false);
expect(latch.hasFailed()).toBe(false);
// autoResolveConflictsLWW with empty conflicts and ops short-circuits
// before validation. To exercise the validation path we call the
// private validation method via type-cast — no other public surface
// runs the conflict-resolution validation in isolation.
await (
conflictResolution as unknown as {
_validateAndRepairAfterResolution(): Promise<boolean>;
}
)._validateAndRepairAfterResolution();
// Note: the private method itself doesn't flip the latch — that's
// done in autoResolveConflictsLWW after the call. Direct invocation
// here verifies the validator returned false; the latch flip is
// observed end-to-end via autoResolveConflictsLWW callers.
expect(validateStateSpy.validateAndRepairCurrentState).toHaveBeenCalledWith(
'conflict-resolution',
{ callerHoldsLock: true },
);
});
});
describe('latch session semantics', () => {
it('multiple validateAfterSync calls within one session keep the latch flipped', async () => {
validateStateSpy.validateAndRepairCurrentState.and.resolveTo(false);
await remoteOps.validateAfterSync();
expect(latch.hasFailed()).toBe(true);
// A subsequent successful validation in the same session does NOT
// un-flip the latch — once corruption is observed, the session is
// tainted until the wrapper resets at the next entry point.
validateStateSpy.validateAndRepairCurrentState.and.resolveTo(true);
await remoteOps.validateAfterSync();
expect(latch.hasFailed()).toBe(true);
});
it('reset() between sessions clears the latch', async () => {
validateStateSpy.validateAndRepairCurrentState.and.resolveTo(false);
await remoteOps.validateAfterSync();
expect(latch.hasFailed()).toBe(true);
latch.reset(); // wrapper would call this at the start of the next sync()
validateStateSpy.validateAndRepairCurrentState.and.resolveTo(true);
await remoteOps.validateAfterSync();
expect(latch.hasFailed()).toBe(false);
});
});
});