mirror of
https://github.com/johannesjo/super-productivity.git
synced 2026-07-17 16:37:43 +00:00
* 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.
This commit is contained in:
Johannes Millan
GitHub
parent
7445f1cdcd
commit
d30fd5f434
13 changed files with 511 additions and 47 deletions
|
|
@ -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`:
|
||||
|
|
|
|||
|
|
@ -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 <constant> 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 '{}';
|
||||
|
|
@ -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");
|
||||
|
|
@ -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")
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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<LatestEntityOperationRow[]>`
|
||||
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<string, LatestEntityOperationRow>();
|
||||
|
|
@ -173,20 +185,28 @@ export const detectConflictForEntity = async (
|
|||
entityId: string,
|
||||
tx: Prisma.TransactionClient,
|
||||
): Promise<ConflictResult> => {
|
||||
// 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<LatestBatchEntityOperationRow[]>`
|
||||
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) {
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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<string, any>();
|
||||
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');
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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<string, number>;
|
||||
};
|
||||
|
||||
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>): 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']);
|
||||
});
|
||||
});
|
||||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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<string>();
|
||||
for (let i = 0; i < touchedParams.length; i += 2) {
|
||||
touchedPairs.add(`${touchedParams[i]}\u0000${touchedParams[i + 1]}`);
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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)', () => {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue