mirror of
https://github.com/johannesjo/super-productivity.git
synced 2026-07-17 16:37:43 +00:00
fix(sync): remediate multi-agent review findings
- USE_REMOTE: fix guaranteed self-deadlock — flushPendingWrites() was called while already holding the non-reentrant sp_op_log lock its own Phase 2 re-acquires, so every "Use Server Data" hung 30s and aborted. New OperationWriteFlushService.flushThenRunExclusive owns the bounded flush->lock->recheck barrier; ServerMigrationService reuses it, which also bounds its previously unbounded snapshot-cutoff retry loop. - USE_REMOTE: make the rebuild crash-resumable. A raw_rebuild_incomplete META marker is set atomically with the baseline replacement and cleared after the replay commits; the next sync redoes the raw rebuild instead of the normal download (which excludes own ops server-side and would silently lose the un-replayed suffix). The resume keeps the first attempt's pre-replace backup instead of overwriting the single slot with the partial baseline. - Deferred actions: serialize overlapping drains (two concurrent drains could mint two ops for one user intent), stop the drain at the first transient failure instead of persisting successors out of order (the older edit would win LWW everywhere via clock inversion), abandon permanently invalid actions after one attempt (typed PermanentDeferredWriteError) instead of retrying + snacking on every sync forever, and latch the buffer-size warnings to once per stuck window (previously a blocking dev dialog per action past 100); drop the no-op 5000 "hard cap" tier. - writeOperation: post-append bookkeeping failures no longer propagate into the deferred retry loop, which re-appended the same action under a fresh op id (double-apply of additive payloads). - remote-apply: full-state cleanup retains every full-state op of the current batch, so an archive_pending import can no longer be deleted before its archive retry (startup replay would lose a change already visible at runtime). - Hydration: sanitize a malformed stored schemaVersion per-op instead of failing the whole boot into recovery; strict parsing (floor now 1, matching the server contract) stays on the receive/upload paths. - Server: request fingerprints are computed lazily — only when a dedup entry exists, and otherwise after the rate-limit/pre-quota gates but before uploadOps mutates the parsed ops. Fixes the pre-gate hashing of full payloads (authenticated DoS-cost regression) and repairs three sync-fixes retry tests to model true identical-body retries. - Delete dead code: failed-op-ids util (+spec) and MAX_VERSION_SKIP (the removed forward-compat band); fix the stale clock-merge parity comment. sync-core 210/210, shared-schema 50/50, super-sync-server 825/825, and all touched Angular specs pass.
This commit is contained in:
Johannes Millan
parent
d212740254
commit
eecf33a4e8
31 changed files with 1069 additions and 452 deletions
|
|
@ -1,9 +1,5 @@
|
|||
// Schema version constants
|
||||
export {
|
||||
CURRENT_SCHEMA_VERSION,
|
||||
MIN_SUPPORTED_SCHEMA_VERSION,
|
||||
MAX_VERSION_SKIP,
|
||||
} from './schema-version';
|
||||
export { CURRENT_SCHEMA_VERSION, MIN_SUPPORTED_SCHEMA_VERSION } from './schema-version';
|
||||
|
||||
// Types
|
||||
export type {
|
||||
|
|
|
|||
|
|
@ -10,9 +10,6 @@ export const CURRENT_SCHEMA_VERSION = 2;
|
|||
*/
|
||||
export const MIN_SUPPORTED_SCHEMA_VERSION = 1;
|
||||
|
||||
/**
|
||||
* Maximum version difference we tolerate before forcing an app update.
|
||||
* If remote data is more than MAX_VERSION_SKIP versions ahead,
|
||||
* the user must update their app.
|
||||
*/
|
||||
export const MAX_VERSION_SKIP = 3;
|
||||
// NOTE: there is deliberately NO forward-compat band: any op from a NEWER
|
||||
// schema version is blocked outright (the client cannot know how to interpret
|
||||
// it), and the user is prompted to update the app.
|
||||
|
|
|
|||
|
|
@ -63,13 +63,19 @@ export class RequestDeduplicationService {
|
|||
|
||||
/**
|
||||
* Check if a request has already been processed for this user + namespace.
|
||||
* @returns Cached results if found and not expired, null otherwise.
|
||||
*
|
||||
* @param getFingerprint lazy fingerprint supplier — hashing the full request
|
||||
* body is expensive (stable stringify + SHA-256 over up to multi-MB
|
||||
* payloads), so it must only run when an entry for this requestId actually
|
||||
* exists (genuine retries are rare) and never for first-time requests.
|
||||
* @returns Cached results if found, not expired, and fingerprint-matching;
|
||||
* null otherwise.
|
||||
*/
|
||||
checkDeduplication<N extends RequestDedupNamespace>(
|
||||
userId: number,
|
||||
namespace: N,
|
||||
requestId: string,
|
||||
fingerprint?: string,
|
||||
getFingerprint?: () => string,
|
||||
): DedupPayload<N> | null {
|
||||
const key = this._key(userId, namespace, requestId);
|
||||
const entry = this.cache.get(key);
|
||||
|
|
@ -80,7 +86,14 @@ export class RequestDeduplicationService {
|
|||
}
|
||||
// Defensive: keying already isolates namespaces, but verify before casting.
|
||||
if (entry.namespace !== namespace) return null;
|
||||
if (fingerprint !== undefined && entry.fingerprint !== fingerprint) return null;
|
||||
if (getFingerprint !== undefined) {
|
||||
// A pre-fingerprint (legacy) entry cannot prove body equality — treat as
|
||||
// a miss so the request is re-processed (safe: the server dedups ops by
|
||||
// id anyway).
|
||||
if (entry.fingerprint === undefined || entry.fingerprint !== getFingerprint()) {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
return entry.results as DedupPayload<N>;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -87,9 +87,15 @@ export const uploadOpsHandler = async (
|
|||
}
|
||||
|
||||
const { ops, clientId, lastKnownServerSeq, requestId } = parseResult.data;
|
||||
const requestFingerprint = requestId
|
||||
? createOpsRequestFingerprint(clientId, ops as unknown as Operation[])
|
||||
: undefined;
|
||||
// Lazy + memoized: hashing the full ops payload is expensive and must not
|
||||
// run before the rate-limit gate below, nor on first-time requests — the
|
||||
// dedup check only invokes it when an entry for this requestId exists.
|
||||
let memoizedFingerprint: string | undefined;
|
||||
const getRequestFingerprint = (): string =>
|
||||
(memoizedFingerprint ??= createOpsRequestFingerprint(
|
||||
clientId,
|
||||
ops as unknown as Operation[],
|
||||
));
|
||||
const syncService = getSyncService();
|
||||
|
||||
Logger.info(
|
||||
|
|
@ -119,7 +125,7 @@ export const uploadOpsHandler = async (
|
|||
const cachedResults = syncService.checkOpsRequestDedup(
|
||||
userId,
|
||||
requestId,
|
||||
requestFingerprint,
|
||||
getRequestFingerprint,
|
||||
);
|
||||
if (cachedResults) {
|
||||
Logger.info(`[user:${userId}] Returning cached results for request ${requestId}`);
|
||||
|
|
@ -170,6 +176,14 @@ export const uploadOpsHandler = async (
|
|||
}
|
||||
}
|
||||
|
||||
// Pin the fingerprint BEFORE processing: uploadOps mutates the parsed ops
|
||||
// (e.g. vector-clock pruning), so hashing after it would never match the
|
||||
// retry's pre-processing hash. Still after the rate-limit gate above, so a
|
||||
// rate-limited client cannot burn CPU on it.
|
||||
if (requestId) {
|
||||
getRequestFingerprint();
|
||||
}
|
||||
|
||||
const results = await syncService.runWithStorageUsageLock<UploadResult[] | null>(
|
||||
userId,
|
||||
async () => {
|
||||
|
|
@ -222,7 +236,12 @@ export const uploadOpsHandler = async (
|
|||
requestId &&
|
||||
!results.some((r) => r.errorCode === SYNC_ERROR_CODES.INTERNAL_ERROR)
|
||||
) {
|
||||
syncService.cacheOpsRequestResults(userId, requestId, results, requestFingerprint);
|
||||
syncService.cacheOpsRequestResults(
|
||||
userId,
|
||||
requestId,
|
||||
results,
|
||||
getRequestFingerprint(),
|
||||
);
|
||||
}
|
||||
|
||||
const accepted = results.filter((r) => r.accepted).length;
|
||||
|
|
|
|||
|
|
@ -100,26 +100,31 @@ export const uploadSnapshotHandler = async (
|
|||
requestId,
|
||||
} = snapshotRequest;
|
||||
const syncService = getSyncService();
|
||||
const requestFingerprint = requestId
|
||||
? createSnapshotRequestFingerprint({
|
||||
state,
|
||||
clientId,
|
||||
reason,
|
||||
vectorClock,
|
||||
schemaVersion,
|
||||
isPayloadEncrypted,
|
||||
syncImportReason,
|
||||
opId,
|
||||
isCleanSlate,
|
||||
snapshotOpType,
|
||||
})
|
||||
: undefined;
|
||||
// Lazy + memoized: fingerprinting stable-stringifies + SHA-256-hashes the
|
||||
// full (possibly multi-MB plaintext) snapshot state. It must not run
|
||||
// before the pre-quota gate below, and first-time requests never need it —
|
||||
// the dedup check only invokes it when an entry for this requestId exists
|
||||
// (a genuine retry).
|
||||
let memoizedFingerprint: string | undefined;
|
||||
const getRequestFingerprint = (): string =>
|
||||
(memoizedFingerprint ??= createSnapshotRequestFingerprint({
|
||||
state,
|
||||
clientId,
|
||||
reason,
|
||||
vectorClock,
|
||||
schemaVersion,
|
||||
isPayloadEncrypted,
|
||||
syncImportReason,
|
||||
opId,
|
||||
isCleanSlate,
|
||||
snapshotOpType,
|
||||
}));
|
||||
|
||||
if (requestId) {
|
||||
const cachedResponse = syncService.checkSnapshotRequestDedup(
|
||||
userId,
|
||||
requestId,
|
||||
requestFingerprint,
|
||||
getRequestFingerprint,
|
||||
);
|
||||
if (cachedResponse) {
|
||||
Logger.info(
|
||||
|
|
@ -156,6 +161,13 @@ export const uploadSnapshotHandler = async (
|
|||
}
|
||||
}
|
||||
|
||||
// Pin the fingerprint BEFORE processing mutates anything, but AFTER the
|
||||
// pre-quota gate above so quota-exhausted clients cannot burn CPU on the
|
||||
// full-state hash.
|
||||
if (requestId) {
|
||||
getRequestFingerprint();
|
||||
}
|
||||
|
||||
// Reject duplicate SYNC_IMPORT before we acquire the per-user lock — a
|
||||
// duplicate rejection is cheap (one indexed lookup) and skipping the
|
||||
// lock lets concurrent legitimate clients keep moving.
|
||||
|
|
@ -363,7 +375,7 @@ export const uploadSnapshotHandler = async (
|
|||
userId,
|
||||
requestId,
|
||||
responseBody,
|
||||
requestFingerprint,
|
||||
getRequestFingerprint(),
|
||||
);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -457,13 +457,13 @@ export class SyncService {
|
|||
checkOpsRequestDedup(
|
||||
userId: number,
|
||||
requestId: string,
|
||||
fingerprint?: string,
|
||||
getFingerprint?: () => string,
|
||||
): UploadResult[] | null {
|
||||
return this.requestDeduplicationService.checkDeduplication(
|
||||
userId,
|
||||
'ops',
|
||||
requestId,
|
||||
fingerprint,
|
||||
getFingerprint,
|
||||
);
|
||||
}
|
||||
|
||||
|
|
@ -485,13 +485,13 @@ export class SyncService {
|
|||
checkSnapshotRequestDedup(
|
||||
userId: number,
|
||||
requestId: string,
|
||||
fingerprint?: string,
|
||||
getFingerprint?: () => string,
|
||||
): SnapshotDedupResponse | null {
|
||||
return this.requestDeduplicationService.checkDeduplication(
|
||||
userId,
|
||||
'snapshot',
|
||||
requestId,
|
||||
fingerprint,
|
||||
getFingerprint,
|
||||
);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -44,13 +44,36 @@ describe('RequestDeduplicationService', () => {
|
|||
service.cacheResults(1, 'ops', 'request-123', mockResults, 'fingerprint-a');
|
||||
|
||||
expect(
|
||||
service.checkDeduplication(1, 'ops', 'request-123', 'fingerprint-a'),
|
||||
service.checkDeduplication(1, 'ops', 'request-123', () => 'fingerprint-a'),
|
||||
).toEqual(mockResults);
|
||||
expect(
|
||||
service.checkDeduplication(1, 'ops', 'request-123', 'fingerprint-b'),
|
||||
service.checkDeduplication(1, 'ops', 'request-123', () => 'fingerprint-b'),
|
||||
).toBeNull();
|
||||
});
|
||||
|
||||
it('should not compute the fingerprint when no entry exists for the requestId', () => {
|
||||
// The fingerprint hashes the full request body — first-time requests
|
||||
// (the overwhelming majority) must not pay that cost.
|
||||
const getFingerprint = vi.fn(() => 'fingerprint-a');
|
||||
|
||||
expect(service.checkDeduplication(1, 'ops', 'request-999', getFingerprint)).toBe(
|
||||
null,
|
||||
);
|
||||
expect(getFingerprint).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('should treat a legacy entry without fingerprint as a miss when a fingerprint is supplied', () => {
|
||||
const mockResults = createMockResults(1);
|
||||
service.cacheResults(1, 'ops', 'request-123', mockResults);
|
||||
|
||||
const getFingerprint = vi.fn(() => 'fingerprint-a');
|
||||
expect(
|
||||
service.checkDeduplication(1, 'ops', 'request-123', getFingerprint),
|
||||
).toBeNull();
|
||||
// No entry fingerprint to compare against — the hash is never computed.
|
||||
expect(getFingerprint).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('should return null for expired request', () => {
|
||||
const mockResults = createMockResults(1);
|
||||
service.cacheResults(1, 'ops', 'request-123', mockResults);
|
||||
|
|
|
|||
|
|
@ -810,7 +810,7 @@ describe('Sync compressed body routes', () => {
|
|||
expect(mocks.syncService.checkSnapshotRequestDedup).toHaveBeenCalledWith(
|
||||
1,
|
||||
'snapshot-v1-retry',
|
||||
expect.any(String),
|
||||
expect.any(Function),
|
||||
);
|
||||
expect(mocks.syncService.checkOpsRequestDedup).not.toHaveBeenCalled();
|
||||
expect(mocks.prisma.operation.findFirst).not.toHaveBeenCalled();
|
||||
|
|
|
|||
|
|
@ -303,6 +303,9 @@ describe('Sync System Fixes', () => {
|
|||
const clientA = 'client-a';
|
||||
const clientB = 'client-b';
|
||||
const requestId = 'req-123';
|
||||
// A true retry resends the IDENTICAL body — the request fingerprint
|
||||
// treats a same-requestId/different-body request as a cache miss.
|
||||
const opA = createOp(clientA, { entityId: 'task-a' });
|
||||
|
||||
// Step 1: Client A uploads with requestId
|
||||
const firstResponse = await app.inject({
|
||||
|
|
@ -310,7 +313,7 @@ describe('Sync System Fixes', () => {
|
|||
url: '/api/sync/ops',
|
||||
headers: { authorization: `Bearer ${authToken}` },
|
||||
payload: {
|
||||
ops: [createOp(clientA, { entityId: 'task-a' })],
|
||||
ops: [opA],
|
||||
clientId: clientA,
|
||||
lastKnownServerSeq: 0,
|
||||
requestId,
|
||||
|
|
@ -339,7 +342,7 @@ describe('Sync System Fixes', () => {
|
|||
url: '/api/sync/ops',
|
||||
headers: { authorization: `Bearer ${authToken}` },
|
||||
payload: {
|
||||
ops: [createOp(clientA, { entityId: 'task-a' })],
|
||||
ops: [opA],
|
||||
clientId: clientA,
|
||||
lastKnownServerSeq: clientASeq,
|
||||
requestId,
|
||||
|
|
@ -366,6 +369,7 @@ describe('Sync System Fixes', () => {
|
|||
const clientA = 'client-a';
|
||||
const clientB = 'client-b';
|
||||
const requestId = 'req-456';
|
||||
const opA = createOp(clientA, { entityId: 'task-1' });
|
||||
|
||||
// Client A uploads
|
||||
await app.inject({
|
||||
|
|
@ -373,7 +377,7 @@ describe('Sync System Fixes', () => {
|
|||
url: '/api/sync/ops',
|
||||
headers: { authorization: `Bearer ${authToken}` },
|
||||
payload: {
|
||||
ops: [createOp(clientA, { entityId: 'task-1' })],
|
||||
ops: [opA],
|
||||
clientId: clientA,
|
||||
lastKnownServerSeq: 0,
|
||||
requestId,
|
||||
|
|
@ -397,7 +401,7 @@ describe('Sync System Fixes', () => {
|
|||
url: '/api/sync/ops',
|
||||
headers: { authorization: `Bearer ${authToken}` },
|
||||
payload: {
|
||||
ops: [createOp(clientA, { entityId: 'task-1' })],
|
||||
ops: [opA],
|
||||
clientId: clientA,
|
||||
lastKnownServerSeq: 0,
|
||||
requestId,
|
||||
|
|
@ -488,6 +492,7 @@ describe('Sync System Fixes', () => {
|
|||
it('should return cached results even when quota would be exceeded on retry', async () => {
|
||||
const clientId = uuidv7();
|
||||
const requestId = uuidv7();
|
||||
const op = createOp(clientId, { entityId: 'task-1' });
|
||||
|
||||
// First request with requestId
|
||||
const firstResponse = await app.inject({
|
||||
|
|
@ -495,7 +500,7 @@ describe('Sync System Fixes', () => {
|
|||
url: '/api/sync/ops',
|
||||
headers: { authorization: `Bearer ${authToken}` },
|
||||
payload: {
|
||||
ops: [createOp(clientId, { entityId: 'task-1' })],
|
||||
ops: [op],
|
||||
clientId,
|
||||
lastKnownServerSeq: 0,
|
||||
requestId,
|
||||
|
|
@ -513,7 +518,7 @@ describe('Sync System Fixes', () => {
|
|||
url: '/api/sync/ops',
|
||||
headers: { authorization: `Bearer ${authToken}` },
|
||||
payload: {
|
||||
ops: [createOp(clientId, { entityId: 'task-1' })],
|
||||
ops: [op],
|
||||
clientId,
|
||||
lastKnownServerSeq: 0,
|
||||
requestId,
|
||||
|
|
|
|||
|
|
@ -74,7 +74,8 @@ const emptyRemoteApplyResult = <
|
|||
* 2. apply only the newly appended ops through the host applier;
|
||||
* 3. mark successfully applied seqs;
|
||||
* 4. merge applied remote vector clocks;
|
||||
* 5. retain only the newest applied full-state ops when configured;
|
||||
* 5. after applying a full-state op, clear older full-state ops while
|
||||
* retaining every full-state op of this batch (incl. archive_pending);
|
||||
* 6. mark the failed op and remaining ops as failed on partial error — their
|
||||
* reducer effects committed with the bulk dispatch; only their archive side
|
||||
* effects are outstanding (see ApplyOperationsResult.failedOp).
|
||||
|
|
@ -145,8 +146,16 @@ export const applyRemoteOperations = async <
|
|||
.map((op) => op.id);
|
||||
|
||||
if (appliedFullStateOpIds.length > 0) {
|
||||
clearedFullStateOpCount =
|
||||
await store.clearFullStateOpsExcept(appliedFullStateOpIds);
|
||||
// Exclude every full-state op of THIS batch, not only the applied ones:
|
||||
// a later full-state op whose reducers committed but whose archive
|
||||
// handling failed is still archive_pending/failed and must survive
|
||||
// cleanup so the retry path can finish it. Deleting it here would let
|
||||
// markFailed miss it and lose a change already visible at runtime on
|
||||
// the next startup replay.
|
||||
const batchFullStateOpIds = appendResult.writtenOps
|
||||
.filter(isFullStateOperation)
|
||||
.map((op) => op.id);
|
||||
clearedFullStateOpCount = await store.clearFullStateOpsExcept(batchFullStateOpIds);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -112,6 +112,38 @@ describe('applyRemoteOperations', () => {
|
|||
expect(result.clearedFullStateOpCount).toBe(3);
|
||||
});
|
||||
|
||||
it('cleanup preserves a batch full-state op whose archive handling failed (archive_pending)', async () => {
|
||||
// Both imports reducer-committed; the LATER one failed only its archive
|
||||
// side effects. Cleanup keyed on the applied one must not delete the
|
||||
// archive_pending entry, or markFailed misses it and the change is lost
|
||||
// from the next startup replay.
|
||||
const appliedImport = createOperation('sync-import-1', 'SYNC_IMPORT');
|
||||
const failedImport = createOperation('sync-import-2', 'SYNC_IMPORT');
|
||||
const error = new Error('archive failure');
|
||||
const store = createStore({
|
||||
seqs: [1, 2],
|
||||
writtenOps: [appliedImport, failedImport],
|
||||
skippedCount: 0,
|
||||
});
|
||||
const applier = createApplier({
|
||||
appliedOps: [appliedImport],
|
||||
failedOp: { op: failedImport, error },
|
||||
});
|
||||
|
||||
await applyRemoteOperations({
|
||||
ops: [appliedImport, failedImport],
|
||||
store,
|
||||
applier,
|
||||
isFullStateOperation: (op) => op.opType === 'SYNC_IMPORT',
|
||||
});
|
||||
|
||||
expect(store.clearFullStateOpsExcept).toHaveBeenCalledWith([
|
||||
'sync-import-1',
|
||||
'sync-import-2',
|
||||
]);
|
||||
expect(store.markFailed).toHaveBeenCalledWith(['sync-import-2']);
|
||||
});
|
||||
|
||||
it('checkpoints the whole reducer batch but charges only the attempted archive failure', async () => {
|
||||
const op1 = createOperation('op-1');
|
||||
const op2 = createOperation('op-2');
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue