From d30fd5f434ae6f78cea7f67d96deb49f0663241b Mon Sep 17 00:00:00 2001 From: Johannes Millan Date: Sat, 13 Jun 2026 15:40:10 +0200 Subject: [PATCH] fix(sync): conflict-check all entities of multi-entity ops (#8334) (#8377) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(sync): conflict-check all entities of multi-entity ops #8334 Multi-entity ops (deleteTasks, moveToArchive, __updateMultipleTaskSimple, round-time-spent, batch board/issue-provider actions) carry entityIds[], but the server operations table only persisted the scalar entity_id (= entityIds[0]). Once such an op was stored, only its first entity took part in future conflict lookups, so a later stale write to a non-first entity found no prior writer and was wrongly accepted instead of rejected as CONFLICT_SUPERSEDED/CONCURRENT. - Add an entity_ids text[] column (populated for multi-entity ops only via getStoredEntityIds; single-entity ops store [] and use the scalar) + a GIN index. Migration is metadata-only with no backfill: pre-migration rows fall back to entity_id, so the fix is forward-only (entities 2..n of already-stored ops were never persisted and stay unrecoverable). - detectConflictForEntity now runs two ordered LIMIT-1 lookups (scalar btree + entity_ids GIN) and takes the higher server_seq, preserving the fast ordered hot path instead of an OR's BitmapOr+sort. - detectConflictForEntities / prefetchLatestEntityOpsForBatch match an entity as the scalar entity_id OR a member of entity_ids (unnest CASE + && / = ANY). - Harden validateOp to bound entityIds (length + per-element), mirroring entityId. Raw SQL validated against Postgres (PGlite) incl. GIN usage and two-query correctness; single path + validation + migrations covered by unit tests. The full conflict-detection.spec needs a generated Prisma client (CI), and the hot-path round-trip tradeoff should be confirmed with a real-PG EXPLAIN. * fix(sync): store entity_ids when a batch op dedups off the scalar #8334 Multi-review follow-up. getStoredEntityIds gated on `length > 1`, so a batch op whose entityIds dedup to a single value that differs from entityId (the server does not enforce entity_id === entityIds[0]) stored [] and that entity became invisible to conflict lookups — reintroducing #8334 for it. Gate on "is the set exactly [entity_id]?" instead, and cover it with unit tests. Also: correct the array-branch comment/doc — a GIN(entity_ids) lookup has no server_seq so it match-all-then-sorts (cheap only because multi-entity ops are rare), it is not an ordered walk; drop a stale "OR filter" test docstring; add a counter-note on getConflictEntityIds vs getStoredEntityIds to prevent swapping. * refactor(sync): single OR lookup for entity conflict detection #8334 Third multi-review follow-up. Revert detectConflictForEntity from the two ordered findFirst lookups back to a single Prisma OR [{entityId}, {entityIds:{has}}]. The two-query split optimized the OR's BitmapOr+sort, but that is bounded by op-log pruning (sub-ms in practice) while the split added a guaranteed extra round-trip on the common single-entity path (doubled by the FIX-1.5 re-check) — a net-negative for the median. The OR is simpler, fully typed/testable, and likely faster in aggregate; a code comment documents the split as the escalation if a real-PG EXPLAIN ever shows the OR is a problem. Review polish (no behaviour change): correct the array-branch/getStoredEntityIds comments (a GIN @> is match-all-then-sort, not an ordered walk; multi-entity-only storage's win is GIN size + keeping single-entity inserts off the GIN, not sort depth); note the batch unnest paths as the first EXPLAIN candidates under load; keep the two batch queries' CASE/prefilter SQL inline (a shared fragment would shift the positional params the conflict-detection.spec mock relies on) with a keep-in-sync note; replace a non-ASCII <= in a client-facing validation error string. * test(sync): update db mocks for OR + prefetch entity_ids lookups #8334 Running the full super-sync-server suite (with a generated Prisma client) surfaced two specs whose inline db mocks hadn't tracked the new conflict-detection SQL: - time-tracking-operations.spec: findFirst now models the single-entity lookup's OR: [{entityId}, {entityIds:{has}}] shape (it previously only matched the scalar entityId, so a concurrent single-entity update was wrongly "accepted"). - sync.service.spec: the prefetch $queryRaw mock parsed userId as the last param and flattened all params into the touched pairs; the #8334 prefilter adds idArray params after userId, so it now finds userId by type and reads the touched pairs from the VALUES join fragment only. Production code unchanged; these are test-mock fidelity fixes. Full suite: 800 passing. --- docs/sync-and-op-log/vector-clocks.md | 16 +++ .../migration.sql | 15 ++ .../migration.sql | 20 +++ .../super-sync-server/prisma/schema.prisma | 10 ++ .../super-sync-server/src/sync/conflict.ts | 109 +++++++++++---- .../sync/services/operation-upload.service.ts | 9 ++ .../src/sync/services/validation.service.ts | 28 +++- .../tests/conflict-detection.spec.ts | 118 ++++++++++++++-- .../tests/issue-8334-detect-conflict.spec.ts | 128 ++++++++++++++++++ .../tests/migration-sql.spec.ts | 32 +++++ .../tests/sync.service.spec.ts | 24 ++-- .../tests/time-tracking-operations.spec.ts | 18 +++ .../tests/validation.service.spec.ts | 31 +++++ 13 files changed, 511 insertions(+), 47 deletions(-) create mode 100644 packages/super-sync-server/prisma/migrations/20260613000000_add_operation_entity_ids/migration.sql create mode 100644 packages/super-sync-server/prisma/migrations/20260613000001_add_operation_entity_ids_gin_index/migration.sql create mode 100644 packages/super-sync-server/tests/issue-8334-detect-conflict.spec.ts diff --git a/docs/sync-and-op-log/vector-clocks.md b/docs/sync-and-op-log/vector-clocks.md index 8d5112986b..32ea655b7b 100644 --- a/docs/sync-and-op-log/vector-clocks.md +++ b/docs/sync-and-op-log/vector-clocks.md @@ -227,6 +227,22 @@ An import is an explicit user action to restore **all clients** to a specific st In `detectConflict()`, operations with `opType` of `SYNC_IMPORT`, `BACKUP_IMPORT`, or `REPAIR` return `{ hasConflict: false }` immediately. These operations replace entire state and don't operate on individual entities. +### Multi-Entity Ops and Server-Side Conflict Detection (issue #8334) + +An op may carry `entityIds: string[]` (batch actions: `deleteTasks`, `moveToArchive`, `__updateMultipleTaskSimple`, round-time-spent, task-repeat-cfg/board/issue-provider batches). `detectConflict()` checks the **incoming** op against **all** of its `entityIds`. The op must also be **looked up** by all of its entities once stored — otherwise a later stale write to a non-first entity would find no prior writer and be wrongly accepted as non-conflicting. + +To make that symmetric, the `operations` row stores: + +- `entity_id` — the client-supplied scalar. For batch ops the client sets it to `entityIds[0]` (`operation-log.effects.ts`), but the **server does not enforce** `entity_id === entityIds[0]`. It is the lookup key for single-entity ops and the first entity of multi-entity ops, and is also used by duplicate detection. +- `entity_ids` — the entity set for **multi-entity ops only** (a `text[]` column; populated via `getStoredEntityIds(op)`, which returns `[]` for single-entity ops). Keeping single-entity rows out of this column keeps the `GIN(entity_ids)` index small and the array-branch lookup cheap. + +The lookups in `conflict.ts` match a requested entity as the scalar `entity_id` **or** a member of `entity_ids`: + +- `detectConflictForEntity` (single) — Prisma `where: { OR: [{ entityId }, { entityIds: { has: entityId } }] }`, ordered by `server_seq`. The `OR` spans the `entity_id` btree and the `entity_ids` GIN, so the planner uses a `BitmapOr` + sort bounded by the entity's stored version depth (op-log pruning keeps that small). If a real-Postgres `EXPLAIN` on a deep-history entity ever shows this hot path is a problem, the escalation is two ordered `LIMIT 1` lookups (scalar btree + `entity_ids` GIN) taking the higher `server_seq` — the array side stays small because the column is multi-entity-only. +- `detectConflictForEntities` / `prefetchLatestEntityOpsForBatch` (batch) — raw SQL unnesting `CASE WHEN cardinality(entity_ids) > 0 THEN entity_ids ELSE ARRAY[entity_id] END`, with a `entity_ids && ... OR entity_id = ANY(...)` prefilter so the `GIN(entity_ids)` index (migration `20260613000001`) and the existing `entity_id` btree stay usable. + +**Forward-only by design:** rows written before migration `20260613000000` have an empty `entity_ids` array and fall back to the scalar `entity_id` (= first entity) in the `CASE` expression / scalar branch above — there is no `UPDATE` backfill. Entities 2..n of already-stored multi-entity ops were never persisted and are unrecoverable, so they remain invisible to conflict detection until that entity gets a fresh write. This residual is bounded: client-side LWW is unaffected (the client persists the full op and `VectorClockService.getEntityFrontier()` fans each op out to **every** entity), and the server only builds an authoritative snapshot from non-encrypted ops (`replayOpsToState()` throws on encrypted ops), so the pre-fix gap could only surface a stale value to a fresh client on non-encrypted self-hosted servers. + ### The `SyncImportFilterService` Algorithm Implemented in `src/app/op-log/sync/sync-import-filter.service.ts`: diff --git a/packages/super-sync-server/prisma/migrations/20260613000000_add_operation_entity_ids/migration.sql b/packages/super-sync-server/prisma/migrations/20260613000000_add_operation_entity_ids/migration.sql new file mode 100644 index 0000000000..d612bde180 --- /dev/null +++ b/packages/super-sync-server/prisma/migrations/20260613000000_add_operation_entity_ids/migration.sql @@ -0,0 +1,15 @@ +-- Add the entity_ids array column backing #8334 multi-entity conflict detection. +-- +-- Ops can touch multiple entities (deleteTasks, moveToArchive, __updateMultipleTaskSimple, +-- round-time-spent, batch board/issue-provider actions). Until now only the scalar +-- entity_id (= entityIds[0]) was persisted, so once a multi-entity op was stored only its +-- first entity participated in future conflict lookups; a later stale write to entity 2..n +-- found no prior writer and was wrongly accepted. entity_ids carries the full set so the +-- conflict lookups (src/sync/conflict.ts) can match any touched entity across uploads. +-- +-- ADD COLUMN ... NOT NULL DEFAULT is a metadata-only operation on PostgreSQL 11+ +-- (the default is stored in the catalog, no table rewrite, no long lock). There is no data +-- backfill on purpose: pre-migration rows keep an empty array and fall back to entity_id in +-- the conflict lookups, so the fix is forward-only (entities 2..n of already-stored ops were +-- never persisted and are unrecoverable). +ALTER TABLE "operations" ADD COLUMN "entity_ids" TEXT[] NOT NULL DEFAULT '{}'; diff --git a/packages/super-sync-server/prisma/migrations/20260613000001_add_operation_entity_ids_gin_index/migration.sql b/packages/super-sync-server/prisma/migrations/20260613000001_add_operation_entity_ids_gin_index/migration.sql new file mode 100644 index 0000000000..46efe94028 --- /dev/null +++ b/packages/super-sync-server/prisma/migrations/20260613000001_add_operation_entity_ids_gin_index/migration.sql @@ -0,0 +1,20 @@ +-- GIN index for #8334 multi-entity conflict detection. +-- +-- The latest-writer lookups match a requested entity id against entity_ids +-- (entity_ids @> ARRAY[id] / entity_ids && ARRAY[...]). A GIN index on the array +-- lets those match without scanning a user's operation history. +-- +-- Single-statement CREATE INDEX CONCURRENTLY: `prisma migrate deploy` applies it +-- natively because PostgreSQL does not wrap a single statement in an implicit +-- transaction (CONCURRENTLY is forbidden inside a transaction). See +-- prisma/migrations/README.md. +-- +-- Avoid name-only idempotency: an interrupted concurrent build leaves an INVALID +-- index with this name, which must fail loudly and be rebuilt rather than be +-- silently skipped (matching the 20260511000000 precedent). Pre-migration rows +-- have an empty entity_ids array; the conflict lookups fall back to the scalar +-- entity_id (served by the existing +-- operations_user_id_entity_type_entity_id_server_seq index), so no data backfill +-- is needed for this index to be correct. +CREATE INDEX CONCURRENTLY "operations_entity_ids_gin" + ON "operations" USING GIN ("entity_ids"); diff --git a/packages/super-sync-server/prisma/schema.prisma b/packages/super-sync-server/prisma/schema.prisma index d2bbc7e361..5e0e33b0ab 100644 --- a/packages/super-sync-server/prisma/schema.prisma +++ b/packages/super-sync-server/prisma/schema.prisma @@ -69,6 +69,14 @@ model Operation { opType String @map("op_type") entityType String @map("entity_type") entityId String? @map("entity_id") + // Entity set for multi-entity (batch) ops only — single-entity ops store [] and + // are matched via the scalar `entityId` (the client sets `entityId` to entityIds[0] + // but the server does not enforce that). Conflict detection consults this array so + // writes to a non-first entity are not invisible across uploads (#8334). + // Pre-migration rows have an empty array and fall back to `entityId` in the + // conflict lookups, so the backfill is forward-only (older entities 2..n are + // unrecoverable — they were never persisted). + entityIds String[] @default([]) @map("entity_ids") payload Json payloadBytes BigInt @default(0) @map("payload_bytes") vectorClock Json @map("vector_clock") @@ -85,6 +93,8 @@ model Operation { @@index([userId, clientId]) @@index([userId, receivedAt]) // Restore-point opType lookups use a raw partial index in the 20260512000000 migration. + // Multi-entity conflict lookups (#8334) use a raw GIN index on entity_ids in the + // 20260613000001 migration (Prisma does not model GIN array indexes here). @@map("operations") } diff --git a/packages/super-sync-server/src/sync/conflict.ts b/packages/super-sync-server/src/sync/conflict.ts index 44ff53a92f..8ad673fdab 100644 --- a/packages/super-sync-server/src/sync/conflict.ts +++ b/packages/super-sync-server/src/sync/conflict.ts @@ -68,16 +68,28 @@ export const detectConflictForEntities = async ( start, start + CONFLICT_DETECTION_ENTITY_BATCH_SIZE, ); + // Match each requested id against the op's full entity set: entity_ids when + // present, else the scalar entity_id (pre-migration rows). The `&&`/`= ANY` + // prefilter keeps the GIN(entity_ids) + entity_id indexes usable; the unnest + // + `eid = ANY` keeps only the requested entities, then DISTINCT ON picks the + // latest op per entity. Kept inline (not a shared fragment) so the positional + // params stay stable for the conflict-detection.spec mock; the same shape is + // duplicated in prefetchLatestEntityOpsForBatch — keep them in sync. (#8334) + const idArray = Prisma.sql`ARRAY[${Prisma.join(batchEntityIds)}]::text[]`; const latestOps = await tx.$queryRaw` - SELECT DISTINCT ON (entity_id) - entity_id AS "entityId", - client_id AS "clientId", - vector_clock AS "vectorClock" - FROM operations - WHERE user_id = ${userId} - AND entity_type = ${op.entityType} - AND entity_id IN (${Prisma.join(batchEntityIds)}) - ORDER BY entity_id, server_seq DESC + SELECT DISTINCT ON (eid) + eid AS "entityId", + o.client_id AS "clientId", + o.vector_clock AS "vectorClock" + FROM operations o + CROSS JOIN LATERAL unnest( + CASE WHEN cardinality(o.entity_ids) > 0 THEN o.entity_ids ELSE ARRAY[o.entity_id] END + ) AS eid + WHERE o.user_id = ${userId} + AND o.entity_type = ${op.entityType} + AND (o.entity_ids && ${idArray} OR o.entity_id = ANY(${idArray})) + AND eid = ANY(${idArray}) + ORDER BY eid, o.server_seq DESC `; const latestOpByEntityId = new Map(); @@ -173,20 +185,28 @@ export const detectConflictForEntity = async ( entityId: string, tx: Prisma.TransactionClient, ): Promise => { - // Get the latest operation for this entity + // Get the latest op that touched this entity. A multi-entity (batch) op stores + // its full set in entity_ids, so match the entity as the scalar entity_id OR a + // member of entity_ids. Single-entity ops store an empty array and are matched + // via the scalar; pre-migration rows likewise fall back to the scalar (#8334). + // + // PERF: the OR spans the entity_id btree + the entity_ids GIN, so the planner + // uses a BitmapOr + sort rather than an ordered LIMIT-1 walk. Bounded by one + // entity's stored version depth (op-log pruning keeps that small, so the sort is + // sub-ms in practice). If a real-Postgres EXPLAIN on a deep-history entity ever + // shows this hot path (run per single-entity upload) is a problem, split into two + // ordered LIMIT-1 lookups (scalar btree + entity_ids GIN) and take the higher + // server_seq — the array branch stays small because entity_ids is multi-entity-only. + // The batch unnest paths (detectConflictForEntities / prefetchLatestEntityOpsForBatch) + // carry the larger sort, so EXPLAIN those first under heavy-user latency. const existingOp = await tx.operation.findFirst({ where: { userId, entityType: op.entityType, - entityId, - }, - select: { - clientId: true, - vectorClock: true, - }, - orderBy: { - serverSeq: 'desc', + OR: [{ entityId }, { entityIds: { has: entityId } }], }, + select: { clientId: true, vectorClock: true }, + orderBy: { serverSeq: 'desc' }, }); // No existing operation = no conflict @@ -287,6 +307,11 @@ export const toStableJsonValue = (value: unknown): unknown => { return value; }; +/** + * All entities an op touches (full set, including single-entity ops). Use for the + * incoming-op conflict check and the in-memory in-batch map — NOT for the persisted + * `entity_ids` column, which uses {@link getStoredEntityIds} (multi-entity only). + */ export const getConflictEntityIds = (op: Operation): string[] => { const rawEntityIds = op.entityIds?.length ? op.entityIds @@ -296,6 +321,26 @@ export const getConflictEntityIds = (op: Operation): string[] => { return Array.from(new Set(rawEntityIds)); }; +/** + * The entity_ids array to persist with an op. Ops whose touched-entity set is + * already covered by the scalar entity_id store an empty array, so single-entity + * ops (the vast majority) stay out of the entity_ids GIN index — keeping it small + * and off their insert write path (Postgres GIN indexes no keys for an empty + * array). Any other set is stored in full. + * + * The gate is "is the set exactly [entity_id]?", NOT "length > 1": a batch op + * whose ids dedup to a single value that differs from entity_id (the server does + * not enforce entity_id === entityIds[0]) must still be stored, or that entity + * would be invisible to conflict lookups (#8334). + */ +export const getStoredEntityIds = (op: Operation): string[] => { + const ids = getConflictEntityIds(op); + if (ids.length <= 1 && ids[0] === op.entityId) { + return []; + } + return ids; +}; + export const getEntityConflictKey = (entityType: string, entityId: string): string => { return `${entityType}\u0000${entityId}`; }; @@ -332,22 +377,36 @@ export const prefetchLatestEntityOpsForBatch = async ( start < entityPairs.length; start += CONFLICT_DETECTION_ENTITY_BATCH_SIZE ) { - const touchedRows = entityPairs - .slice(start, start + CONFLICT_DETECTION_ENTITY_BATCH_SIZE) - .map(({ entityType, entityId }) => Prisma.sql`(${entityType}, ${entityId})`); + const batchPairs = entityPairs.slice( + start, + start + CONFLICT_DETECTION_ENTITY_BATCH_SIZE, + ); + const touchedRows = batchPairs.map( + ({ entityType, entityId }) => Prisma.sql`(${entityType}, ${entityId})`, + ); + // Prefilter array (all requested ids) so the JOIN below can match a requested + // id inside a stored op's entity_ids set, not just its scalar entity_id, while + // keeping the GIN(entity_ids) + entity_id indexes usable. (#8334) + const idArray = Prisma.sql`ARRAY[${Prisma.join( + batchPairs.map((pair) => pair.entityId), + )}]::text[]`; const latestOps = await tx.$queryRaw` - SELECT DISTINCT ON (o.entity_type, o.entity_id) + SELECT DISTINCT ON (o.entity_type, eid) o.entity_type AS "entityType", - o.entity_id AS "entityId", + eid AS "entityId", o.client_id AS "clientId", o.vector_clock AS "vectorClock" FROM operations o + CROSS JOIN LATERAL unnest( + CASE WHEN cardinality(o.entity_ids) > 0 THEN o.entity_ids ELSE ARRAY[o.entity_id] END + ) AS eid JOIN (VALUES ${Prisma.join(touchedRows)}) AS touched(entity_type, entity_id) ON touched.entity_type = o.entity_type - AND touched.entity_id = o.entity_id + AND touched.entity_id = eid WHERE o.user_id = ${userId} - ORDER BY o.entity_type, o.entity_id, o.server_seq DESC + AND (o.entity_ids && ${idArray} OR o.entity_id = ANY(${idArray})) + ORDER BY o.entity_type, eid, o.server_seq DESC `; for (const latestOp of latestOps) { diff --git a/packages/super-sync-server/src/sync/services/operation-upload.service.ts b/packages/super-sync-server/src/sync/services/operation-upload.service.ts index 48d0620832..00c69cc7b4 100644 --- a/packages/super-sync-server/src/sync/services/operation-upload.service.ts +++ b/packages/super-sync-server/src/sync/services/operation-upload.service.ts @@ -21,6 +21,7 @@ import { detectConflict, getBatchConflictEntityPairs, getConflictEntityIds, + getStoredEntityIds, getEntityConflictKey, isSameDuplicateOperation, prefetchLatestEntityOpsForBatch, @@ -544,6 +545,10 @@ export class OperationUploadService { opType: candidate.op.opType, entityType: candidate.op.entityType, entityId: candidate.op.entityId ?? null, + // Persist the full entity set for multi-entity ops so conflict detection + // can match a write to any touched entity across uploads, not just + // entityIds[0]; single-entity ops store [] and use the scalar (#8334). + entityIds: getStoredEntityIds(candidate.op), payload: candidate.op.payload as Prisma.InputJsonValue, payloadBytes: BigInt(candidate.storageBytes), vectorClock: candidate.op.vectorClock as Prisma.InputJsonValue, @@ -800,6 +805,10 @@ export class OperationUploadService { opType: op.opType, entityType: op.entityType, entityId: op.entityId ?? null, + // Persist the full entity set for multi-entity ops so conflict detection + // can match a write to any touched entity across uploads, not just + // entityIds[0]; single-entity ops store [] and use the scalar (#8334). + entityIds: getStoredEntityIds(op), payload: op.payload as Prisma.InputJsonValue, payloadBytes: BigInt(sized.bytes), vectorClock: op.vectorClock as Prisma.InputJsonValue, diff --git a/packages/super-sync-server/src/sync/services/validation.service.ts b/packages/super-sync-server/src/sync/services/validation.service.ts index 87102c6dc0..382467ee12 100644 --- a/packages/super-sync-server/src/sync/services/validation.service.ts +++ b/packages/super-sync-server/src/sync/services/validation.service.ts @@ -13,7 +13,7 @@ import { sanitizeVectorClock, validatePayload, } from '../sync.types'; -import { ENTITY_TYPES } from '@sp/shared-schema'; +import { ENTITY_TYPES, SUPER_SYNC_MAX_ENTITY_IDS_PER_OP } from '@sp/shared-schema'; import { Logger } from '../../logger'; /** @@ -106,6 +106,32 @@ export class ValidationService { } } + // Validate the multi-entity set the same way as the scalar entityId. The HTTP + // contract Zod schema also bounds this, but enforcing it here keeps the invariant + // with the op (defense-in-depth for any non-HTTP caller) since entityIds is now + // persisted and consulted by conflict detection (#8334). + if (op.entityIds !== undefined && op.entityIds !== null) { + if ( + !Array.isArray(op.entityIds) || + op.entityIds.length > SUPER_SYNC_MAX_ENTITY_IDS_PER_OP + ) { + return { + valid: false, + error: 'Invalid entityIds: not an array or too many entries', + errorCode: SYNC_ERROR_CODES.INVALID_ENTITY_ID, + }; + } + for (const id of op.entityIds) { + if (typeof id !== 'string' || id.length > 255 || id.trim().length === 0) { + return { + valid: false, + error: 'Invalid entityIds element: must be a non-empty string <= 255 chars', + errorCode: SYNC_ERROR_CODES.INVALID_ENTITY_ID, + }; + } + } + } + // Require entityId for regular entity operations. // Full-state operations (SYNC_IMPORT, BACKUP_IMPORT, REPAIR) and bulk entity types // (ALL, RECOVERY) legitimately don't have entityId. diff --git a/packages/super-sync-server/tests/conflict-detection.spec.ts b/packages/super-sync-server/tests/conflict-detection.spec.ts index 46d4645d40..475efcef40 100644 --- a/packages/super-sync-server/tests/conflict-detection.spec.ts +++ b/packages/super-sync-server/tests/conflict-detection.spec.ts @@ -63,6 +63,28 @@ vi.mock('../src/db', async () => { applyOperationSelect(state.operations.get(args.where.id), args.select) || null ); } + // Single-entity conflict lookup: where { userId, entityType, + // OR: [{ entityId: X }, { entityIds: { has: X } }] } — match X as the + // scalar entity_id OR a member of the entity_ids array (#8334). + if (Array.isArray(args.where?.OR) && args.where?.entityType) { + state.entityConflictFindFirstCount++; + const scalarClause = args.where.OR.find((c: any) => 'entityId' in c); + const hasClause = args.where.OR.find( + (c: any) => c.entityIds?.has !== undefined, + ); + const targetId = scalarClause?.entityId ?? hasClause?.entityIds?.has; + const ops = Array.from(state.operations.values()) + .filter( + (op: any) => + op.userId === args.where.userId && + op.entityType === args.where.entityType && + (op.entityId === targetId || + (Array.isArray(op.entityIds) && op.entityIds.includes(targetId))), + ) + .sort((a: any, b: any) => b.serverSeq - a.serverSeq); + return applyOperationSelect(ops[0], args.select) || null; + } + // Scalar-only lookup (other callers): where { userId, entityType, entityId }. if (args.where?.entityId && args.where?.entityType) { state.entityConflictFindFirstCount++; const ops = Array.from(state.operations.values()) @@ -200,23 +222,29 @@ vi.mock('../src/db', async () => { (entityId): entityId is string => typeof entityId === 'string', ), ); + // An op covers an entity via its full entity_ids set (multi-entity ops), + // falling back to the scalar entity_id for pre-migration rows (#8334). + const coveredEntityIds = (op: any): string[] => + Array.isArray(op.entityIds) && op.entityIds.length > 0 + ? op.entityIds + : op.entityId != null + ? [op.entityId] + : []; const latestByEntityId = new Map(); const ops = Array.from(state.operations.values()) - .filter( - (op: any) => - op.userId === userId && - op.entityType === entityType && - batchEntityIds.has(op.entityId), - ) + .filter((op: any) => op.userId === userId && op.entityType === entityType) .sort((a: any, b: any) => b.serverSeq - a.serverSeq); for (const op of ops) { - if (!op.entityId || latestByEntityId.has(op.entityId)) continue; - latestByEntityId.set(op.entityId, op); + for (const eid of coveredEntityIds(op)) { + if (batchEntityIds.has(eid) && !latestByEntityId.has(eid)) { + latestByEntityId.set(eid, op); + } + } } - return Array.from(latestByEntityId.values()).map((op: any) => ({ - entityId: op.entityId, + return Array.from(latestByEntityId.entries()).map(([eid, op]) => ({ + entityId: eid, clientId: op.clientId, vectorClock: op.vectorClock, })); @@ -969,5 +997,75 @@ describe('Conflict Detection', () => { }); expect(result[0].error).toContain('TASK:task-2'); }); + + // Regression for #8334. The test above proves an *incoming* multi-entity op + // is checked against all its ids. These two prove the reverse — a *stored* + // multi-entity op exposes every entity to later conflict lookups, not just + // entityIds[0] — across both the single-entity and batch lookup paths. + it('#8334 single path: stale write to a non-first stored entity conflicts', async () => { + const service = getSyncService(); + + // clientA stores a multi-entity op over [task-1, task-2] + // (persisted scalar entity_id = task-1, entity_ids = both). + const stored = await service.uploadOps(userId, clientA, [ + createOp({ + entityId: 'task-1', + entityIds: ['task-1', 'task-2'], + clientId: clientA, + vectorClock: { [clientA]: 1 }, + }), + ]); + expect(stored[0].accepted).toBe(true); + + // clientB sends a stale single-entity op for task-2 (the SECOND entity). + // {clientB:1} is CONCURRENT with {clientA:1}. + const result = await service.uploadOps(userId, clientB, [ + createOp({ + entityId: 'task-2', + clientId: clientB, + vectorClock: { [clientB]: 1 }, + }), + ]); + + expect(result[0]).toMatchObject({ + accepted: false, + errorCode: SYNC_ERROR_CODES.CONFLICT_CONCURRENT, + }); + }); + + // NOTE: with the default (non-batch) upload config this exercises + // `detectConflictForEntities`. The `prefetchLatestEntityOpsForBatch` variant + // (batchUpload=true) shares the same entity_ids matching SQL but is not driven + // here; its raw query is validated separately against real Postgres. + it('#8334 batch path: incoming multi-entity op hits a non-first stored entity', async () => { + const service = getSyncService(); + + const stored = await service.uploadOps(userId, clientA, [ + createOp({ + entityId: 'task-1', + entityIds: ['task-1', 'task-2'], + clientId: clientA, + vectorClock: { [clientA]: 1 }, + }), + ]); + expect(stored[0].accepted).toBe(true); + + // Incoming *multi-entity* op (→ batch lookup path) touching task-2. + const result = await service.uploadOps(userId, clientB, [ + createOp({ + entityId: 'task-2', + entityIds: ['task-2', 'task-9'], + clientId: clientB, + vectorClock: { [clientB]: 1 }, + }), + ]); + + expect(testState.batchConflictQueryCount).toBeGreaterThan(0); + expect(result[0]).toMatchObject({ + accepted: false, + errorCode: SYNC_ERROR_CODES.CONFLICT_CONCURRENT, + }); + expect(result[0].error).toContain('TASK:task-2'); + }); }); }); diff --git a/packages/super-sync-server/tests/issue-8334-detect-conflict.spec.ts b/packages/super-sync-server/tests/issue-8334-detect-conflict.spec.ts new file mode 100644 index 0000000000..81587a5c88 --- /dev/null +++ b/packages/super-sync-server/tests/issue-8334-detect-conflict.spec.ts @@ -0,0 +1,128 @@ +import { describe, expect, it } from 'vitest'; +import { detectConflict, getStoredEntityIds } from '../src/sync/conflict'; +import type { Operation } from '../src/sync/sync.types'; + +/** + * Unit-level regression for #8334's single-entity lookup path + * (detectConflict → detectConflictForEntity). It exercises the REAL function + * against a tx mock modelling how production persists ops: multi-entity ops carry + * the full entity_ids array, single-entity ops store []. detectConflictForEntity + * matches an entity via a Prisma `OR: [{entityId}, {entityIds:{has}}]` filter, + * which the mock's findFirst reproduces. + * + * The batch lookup paths (raw unnest SQL) are validated separately against real + * Postgres semantics and in conflict-detection.spec.ts. + */ +type StoredRow = { + userId: number; + entityType: string; + entityId: string | null; + entityIds: string[]; + clientId: string; + serverSeq: number; + vectorClock: Record; +}; + +const makeTx = (rows: StoredRow[]): any => ({ + operation: { + // Mirrors: where { userId, entityType, OR: [{entityId:X}, {entityIds:{has:X}}] } + findFirst: async ({ where }: any) => { + const target = + where.OR.find((c: any) => 'entityId' in c)?.entityId ?? + where.OR.find((c: any) => c.entityIds?.has !== undefined)?.entityIds?.has; + return ( + rows + .filter( + (r) => + r.userId === where.userId && + r.entityType === where.entityType && + (r.entityId === target || r.entityIds.includes(target)), + ) + .sort((a, b) => b.serverSeq - a.serverSeq)[0] ?? null + ); + }, + }, +}); + +const staleOp = (entityId: string): Operation => + ({ + id: 'op-b', + clientId: 'B', + actionType: 'UPDATE_TASK', + opType: 'UPD', + entityType: 'TASK', + entityId, + vectorClock: { B: 1 }, + timestamp: 1, + schemaVersion: 1, + }) as unknown as Operation; + +const multiEntityRow: StoredRow = { + userId: 1, + entityType: 'TASK', + entityId: 'task-1', + entityIds: ['task-1', 'task-2'], + clientId: 'A', + serverSeq: 1, + vectorClock: { A: 1 }, +}; + +describe('#8334 detectConflict single-entity path', () => { + it('rejects a stale write to a NON-FIRST entity of a stored multi-entity op', async () => { + const result = await detectConflict(1, staleOp('task-2'), makeTx([multiEntityRow])); + expect(result.hasConflict).toBe(true); + }); + + it('still rejects a stale write to the first/scalar entity', async () => { + const result = await detectConflict(1, staleOp('task-1'), makeTx([multiEntityRow])); + expect(result.hasConflict).toBe(true); + }); + + it('finds a pre-migration single-entity row via the scalar fallback', async () => { + const oldRow: StoredRow = { + userId: 1, + entityType: 'TASK', + entityId: 'task-3', + entityIds: [], // pre-migration: empty array, only scalar persisted + clientId: 'A', + serverSeq: 1, + vectorClock: { A: 1 }, + }; + const result = await detectConflict(1, staleOp('task-3'), makeTx([oldRow])); + expect(result.hasConflict).toBe(true); + }); + + it('does not flag an unrelated entity', async () => { + const result = await detectConflict(1, staleOp('task-9'), makeTx([multiEntityRow])); + expect(result.hasConflict).toBe(false); + }); +}); + +describe('#8334 getStoredEntityIds', () => { + const op = (over: Partial): Operation => over as unknown as Operation; + + it('stores [] for a single-entity op (covered by the scalar)', () => { + expect(getStoredEntityIds(op({ entityId: 'task-1' }))).toEqual([]); + expect(getStoredEntityIds(op({ entityId: 'task-1', entityIds: ['task-1'] }))).toEqual( + [], + ); + }); + + it('stores the full set for a multi-entity op', () => { + expect( + getStoredEntityIds(op({ entityId: 'task-1', entityIds: ['task-1', 'task-2'] })), + ).toEqual(['task-1', 'task-2']); + }); + + it('stores [] for an entity-less op', () => { + expect(getStoredEntityIds(op({}))).toEqual([]); + }); + + it('stores the id when a batch op dedups to one value that differs from entityId', () => { + // The server does not enforce entityId === entityIds[0]; without this the + // entity would be invisible to conflict lookups (#8334). + expect( + getStoredEntityIds(op({ entityId: 'task-Z', entityIds: ['task-A', 'task-A'] })), + ).toEqual(['task-A']); + }); +}); diff --git a/packages/super-sync-server/tests/migration-sql.spec.ts b/packages/super-sync-server/tests/migration-sql.spec.ts index b48779036b..64e6645c17 100644 --- a/packages/super-sync-server/tests/migration-sql.spec.ts +++ b/packages/super-sync-server/tests/migration-sql.spec.ts @@ -133,6 +133,38 @@ describe('performance migrations', () => { expect(migrationSql).not.toMatch(/\bBEGIN\b|\bCOMMIT\b/i); }); + it('adds the operation entity_ids column as a metadata-only column (no table rewrite)', () => { + const migrationSql = readMigration('20260613000000_add_operation_entity_ids'); + + // Same fast-path guards as payload_bytes: ADD COLUMN with a constant default is + // metadata-only on PG 11+. A future edit to an expression default or a separate + // UPDATE backfill would rewrite/lock a 100M-row table — #8334 is forward-only by + // design (pre-migration rows fall back to the scalar entity_id), so no backfill. + expect(migrationSql).toMatch( + /ALTER TABLE "operations"\s+ADD COLUMN "entity_ids" TEXT\[\] NOT NULL DEFAULT '\{\}'/i, + ); + expect(migrationSql).not.toMatch(/\bUPDATE\b/i); + expect(migrationSql).not.toMatch(/\bUSING\b/i); + expect(migrationSql).not.toMatch(/\bDROP\s+TABLE\b/i); + expect(migrationSql).not.toMatch(/\bBEGIN\b|\bCOMMIT\b/i); + }); + + it('adds the entity_ids GIN index concurrently as a single native-apply statement', () => { + const migrationSql = readMigration( + '20260613000001_add_operation_entity_ids_gin_index', + ); + + expect(migrationSql).toContain('CREATE INDEX CONCURRENTLY'); + expect(migrationSql).toContain('"operations_entity_ids_gin"'); + expect(migrationSql).toContain('USING GIN ("entity_ids")'); + // Bare CREATE (no IF NOT EXISTS / no drop-then-create): an interrupted concurrent + // build must fail loudly, matching the 20260511000000 precedent. + expect(migrationSql).not.toMatch(/\bIF\s+NOT\s+EXISTS\b/i); + expect(migrationSql).not.toMatch(/\bALTER\s+TABLE\b/i); + expect(migrationSql).not.toMatch(/\bDROP\s+TABLE\b/i); + expect(migrationSql).not.toMatch(/\bBEGIN\b|\bCOMMIT\b/i); + }); + it('runs migrations before replacing the app during compose deploys', () => { const deployScript = readFileSync(join(currentDir, '../scripts/deploy.sh'), 'utf8'); const runtimeMigrateScript = readFileSync( diff --git a/packages/super-sync-server/tests/sync.service.spec.ts b/packages/super-sync-server/tests/sync.service.spec.ts index 326afb0a2f..9c7448f827 100644 --- a/packages/super-sync-server/tests/sync.service.spec.ts +++ b/packages/super-sync-server/tests/sync.service.spec.ts @@ -325,17 +325,19 @@ vi.mock('../src/db', async () => { return [{ lastSeq }]; } if (sql.includes('JOIN (VALUES')) { - const txUserId = params[params.length - 1] as number; - const touchedParams = params.slice(0, -1).flatMap((param: unknown): unknown[] => { - if ( - param && - typeof param === 'object' && - Array.isArray((param as { values?: unknown[] }).values) - ) { - return (param as { values: unknown[] }).values; - } - return [param]; - }); + // Params are [touchedRows (VALUES join), userId, idArray, idArray]: userId + // is the only number; the VALUES join is the first Sql fragment, whose + // .values hold the flattened (entity_type, entity_id) touched pairs. The + // idArray prefilter params (#8334) are not needed by the mock. (Previously + // userId was the last param and the join the only fragment.) + const txUserId = params.find((p: unknown) => typeof p === 'number') as number; + const valuesParam = params.find( + (p: unknown) => + !!p && + typeof p === 'object' && + Array.isArray((p as { values?: unknown[] }).values), + ) as { values: unknown[] } | undefined; + const touchedParams = valuesParam?.values ?? []; const touchedPairs = new Set(); for (let i = 0; i < touchedParams.length; i += 2) { touchedPairs.add(`${touchedParams[i]}\u0000${touchedParams[i + 1]}`); diff --git a/packages/super-sync-server/tests/time-tracking-operations.spec.ts b/packages/super-sync-server/tests/time-tracking-operations.spec.ts index 3476d167e3..fea681c0e5 100644 --- a/packages/super-sync-server/tests/time-tracking-operations.spec.ts +++ b/packages/super-sync-server/tests/time-tracking-operations.spec.ts @@ -81,6 +81,24 @@ vi.mock('../src/db', async () => { applyOperationSelect(state.operations.get(args.where.id), args.select) || null ); } + // Single-entity conflict lookup: where { userId, entityType, + // OR: [{ entityId: X }, { entityIds: { has: X } }] } (#8334). + if (Array.isArray(args.where?.OR) && args.where?.entityType) { + const targetId = + args.where.OR.find((c: any) => 'entityId' in c)?.entityId ?? + args.where.OR.find((c: any) => c.entityIds?.has !== undefined)?.entityIds + ?.has; + const ops = Array.from(state.operations.values()) + .filter( + (op: any) => + op.userId === args.where.userId && + op.entityType === args.where.entityType && + (op.entityId === targetId || + (Array.isArray(op.entityIds) && op.entityIds.includes(targetId))), + ) + .sort((a: any, b: any) => b.serverSeq - a.serverSeq); + return applyOperationSelect(ops[0], args.select) || null; + } if (args.where?.entityId && args.where?.entityType) { const ops = Array.from(state.operations.values()) .filter( diff --git a/packages/super-sync-server/tests/validation.service.spec.ts b/packages/super-sync-server/tests/validation.service.spec.ts index 78efc3f676..b2e65c4c31 100644 --- a/packages/super-sync-server/tests/validation.service.spec.ts +++ b/packages/super-sync-server/tests/validation.service.spec.ts @@ -114,6 +114,37 @@ describe('ValidationService', () => { expect(result.errorCode).toBe(SYNC_ERROR_CODES.INVALID_ENTITY_ID); }); + // === entityIds (multi-entity batch ops) validation (#8334) === + + it('should accept a valid entityIds array', () => { + const op = createValidOp({ entityIds: ['task-1', 'task-2'] }); + expect(validationService.validateOp(op, clientId).valid).toBe(true); + }); + + it('should reject an entityIds element longer than 255 characters', () => { + const op = createValidOp({ entityIds: ['ok', 'x'.repeat(256)] }); + const result = validationService.validateOp(op, clientId); + expect(result.valid).toBe(false); + expect(result.errorCode).toBe(SYNC_ERROR_CODES.INVALID_ENTITY_ID); + }); + + it('should reject a non-string / empty entityIds element', () => { + expect( + validationService.validateOp(createValidOp({ entityIds: [123] }), clientId).valid, + ).toBe(false); + expect( + validationService.validateOp(createValidOp({ entityIds: [' '] }), clientId) + .valid, + ).toBe(false); + }); + + it('should reject more than SUPER_SYNC_MAX_ENTITY_IDS_PER_OP entries', () => { + const op = createValidOp({ entityIds: new Array(1001).fill('id') }); + const result = validationService.validateOp(op, clientId); + expect(result.valid).toBe(false); + expect(result.errorCode).toBe(SYNC_ERROR_CODES.INVALID_ENTITY_ID); + }); + // === entityId validation for regular entity types === it('should reject null entityId for regular entity type (TASK)', () => {