diff --git a/docs/sync-and-op-log/vector-clocks.md b/docs/sync-and-op-log/vector-clocks.md index fcef56bdda..d709b72c6b 100644 --- a/docs/sync-and-op-log/vector-clocks.md +++ b/docs/sync-and-op-log/vector-clocks.md @@ -168,7 +168,7 @@ With MAX=20, a user needs 21+ unique client IDs before pruning triggers. Both si ### Server-Side Flow -1. Server finds the latest operation for the same entity (`findFirst` by `entityType + entityId`, ordered by `serverSeq desc`) +1. Server finds the latest operation for the same entity — **two separately-indexed lookups**, a scalar `findFirst` plus a raw-SQL `MATERIALIZED` CTE over `entity_ids`, taking whichever has the higher `serverSeq`. Deliberately NOT one combined filter; see the multi-entity section below for why that caused an outage. 2. Compares incoming clock vs existing clock using the **full unpruned** incoming clock 3. Possible outcomes: - `GREATER_THAN` → **accept** (incoming op causally succeeds existing) @@ -237,10 +237,51 @@ To make that symmetric, the `operations` row stores: 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. +- `detectConflictForEntity` (single) — **two separately-indexed lookups, never one combined filter.** A scalar `findFirst` on `{ userId, entityType, entityId }` ordered by `server_seq` (served end to end by the `(user_id, entity_type, entity_id, server_seq)` btree), plus a raw-SQL `MATERIALIZED` CTE taking `MAX(server_seq)` over `entity_ids @> ARRAY[id]`, with the winning row then fetched by the `(user_id, server_seq)` unique key. -**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. + > ⚠️ **Do not "simplify" this back into one query.** It used to be + > `where: { OR: [{ entityId }, { entityIds: { has: entityId } }] }` + `orderBy: { serverSeq: 'desc' }`, + > and on 2026-07-20 that caused a total sync outage — 47 stuck backends, longest 75 minutes, + > 61/66 connections consumed. The `OR` spans two different indexes and GIN cannot supply + > `server_seq` ordering, so the planner abandons **both** index paths and walks the user's + > history. Nothing bounds that walk when the entity has no matching rows — i.e. the + > first-ever op for a new task, the most common upload there is. Op-log pruning does **not** + > bound it; that assumption is what this paragraph used to assert, and it was wrong. + > + > The obvious escalations are broken too. Two ordered `LIMIT 1` lookups still leave the + > array side unable to order on GIN. Measured under generic planning on a 40k-row seed, the + > outage query, the naive array-only `LIMIT 1`, the flat `MAX`, Prisma's `aggregate({ _max })` + > and the CTE with `MATERIALIZED` dropped **all** read the user's whole entity-type slice, + > against 143 blocks and 0 discarded for the shipped form. The **816 blocks / 2500 rows + > discarded** figure is the outage query specifically, pinned by the `CANARY` case in + > `conflict-entity-lookup-plan.pglite.spec.ts`. The other four are not unguarded: that + > spec rebuilds the array branch from the live tagged template, so dropping `MATERIALIZED` + > or flattening the `MAX` blows the block budget and fails there (verified by mutation). + > What is _not_ pinned is their individual historical block counts. + > + > Measure any change here with `SET plan_cache_mode = force_generic_plan`. Prisma sends + > parameterized prepared statements; under `auto` Postgres plans the first ~5 executions as + > custom, then compares the generic cost against the average custom cost and **may** switch + > to a generic plan — a cost comparison, not an automatic switch, so some statements stay on + > custom plans indefinitely. This one was observed going generic on production, and a + > generic plan cannot see the parameter values. `EXPLAIN` with literal constants is + > different again and makes every one of those broken shapes look perfect. See + > `packages/super-sync-server/tests/conflict-entity-lookup-plan.pglite.spec.ts` and the note + > at `detectConflictForEntity` in `packages/super-sync-server/src/sync/conflict.ts`. + +- `detectConflictForEntities` / `prefetchLatestEntityOpsForBatch` (batch) — raw SQL unnesting the **union** of both columns, `entity_ids || CASE WHEN entity_id IS NULL THEN '{}' ELSE ARRAY[entity_id] END`, deduped by `DISTINCT ON`, with an `entity_ids && ... OR entity_id = ANY(...)` prefilter so the `GIN(entity_ids)` index (migration `20260613000001`) and the existing `entity_id` btree stay usable. + + > ⚠️ It must be a **union**, not the mutually exclusive + > `CASE WHEN cardinality(entity_ids) > 0 THEN entity_ids ELSE ARRAY[entity_id] END` + > this section used to document. The server does **not** enforce + > `entity_id === entityIds[0]`, so a multi-entity op can carry a scalar that is not a + > member of its own `entity_ids` (see `getStoredEntityIds`). The exclusive form drops + > that scalar whenever the array is non-empty, making the entity invisible to conflict + > lookups — a later concurrent write to it is wrongly accepted, which is **silent data + > loss**. That was the #8334 bug; the divergent-scalar case is the decisive test in + > `tests/integration/conflict-detection-sql.integration.spec.ts`. + +**Forward-only by design:** rows written before migration `20260613000000` have an empty `entity_ids` array, so they are reached only by their scalar `entity_id` (= first entity) — via the scalar arm of the batch union above, or the scalar branch of the single-entity lookup. (Not via the exclusive `CASE` form: that is the removed #8334 bug documented in the warning above, not the current shape.) 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 diff --git a/packages/super-sync-server/prisma/schema.prisma b/packages/super-sync-server/prisma/schema.prisma index 8b78994d2d..e7d69faf55 100644 --- a/packages/super-sync-server/prisma/schema.prisma +++ b/packages/super-sync-server/prisma/schema.prisma @@ -11,31 +11,31 @@ datasource db { } model User { - id Int @id @default(autoincrement()) - email String @unique - passwordHash String? @map("password_hash") // Nullable for passkey-only users - isVerified Int @default(0) @map("is_verified") // 0 or 1 - verificationToken String? @map("verification_token") - verificationTokenExpiresAt BigInt? @map("verification_token_expires_at") - verificationResendCount Int @default(0) @map("verification_resend_count") - resetPasswordToken String? @map("reset_password_token") - resetPasswordTokenExpiresAt BigInt? @map("reset_password_token_expires_at") - passkeyRecoveryToken String? @map("passkey_recovery_token") - passkeyRecoveryTokenExpiresAt BigInt? @map("passkey_recovery_token_expires_at") - loginToken String? @map("login_token") - loginTokenExpiresAt BigInt? @map("login_token_expires_at") - failedLoginAttempts Int @default(0) @map("failed_login_attempts") - lockedUntil BigInt? @map("locked_until") - tokenVersion Int @default(0) @map("token_version") - termsAcceptedAt BigInt? @map("terms_accepted_at") - createdAt DateTime @default(now()) @map("created_at") - storageQuotaBytes BigInt @default(104857600) @map("storage_quota_bytes") // 100MB default - storageUsedBytes BigInt @default(0) @map("storage_used_bytes") + id Int @id @default(autoincrement()) + email String @unique + passwordHash String? @map("password_hash") // Nullable for passkey-only users + isVerified Int @default(0) @map("is_verified") // 0 or 1 + verificationToken String? @map("verification_token") + verificationTokenExpiresAt BigInt? @map("verification_token_expires_at") + verificationResendCount Int @default(0) @map("verification_resend_count") + resetPasswordToken String? @map("reset_password_token") + resetPasswordTokenExpiresAt BigInt? @map("reset_password_token_expires_at") + passkeyRecoveryToken String? @map("passkey_recovery_token") + passkeyRecoveryTokenExpiresAt BigInt? @map("passkey_recovery_token_expires_at") + loginToken String? @map("login_token") + loginTokenExpiresAt BigInt? @map("login_token_expires_at") + failedLoginAttempts Int @default(0) @map("failed_login_attempts") + lockedUntil BigInt? @map("locked_until") + tokenVersion Int @default(0) @map("token_version") + termsAcceptedAt BigInt? @map("terms_accepted_at") + createdAt DateTime @default(now()) @map("created_at") + storageQuotaBytes BigInt @default(104857600) @map("storage_quota_bytes") // 100MB default + storageUsedBytes BigInt @default(0) @map("storage_used_bytes") - operations Operation[] - syncState UserSyncState? - devices SyncDevice[] - passkeys Passkey[] + operations Operation[] + syncState UserSyncState? + devices SyncDevice[] + passkeys Passkey[] pendingPasskeyRegistrations PendingPasskeyRegistration[] @@index([verificationToken]) @@ -46,16 +46,16 @@ model User { } model Passkey { - id String @id @default(cuid()) - credentialId Bytes @unique @map("credential_id") - publicKey Bytes @map("public_key") - counter BigInt @default(0) - transports String? // JSON array of transport types - createdAt DateTime @default(now()) @map("created_at") - lastUsedAt DateTime? @map("last_used_at") + id String @id @default(cuid()) + credentialId Bytes @unique @map("credential_id") + publicKey Bytes @map("public_key") + counter BigInt @default(0) + transports String? // JSON array of transport types + createdAt DateTime @default(now()) @map("created_at") + lastUsedAt DateTime? @map("last_used_at") - userId Int @map("user_id") - user User @relation(fields: [userId], references: [id], onDelete: Cascade) + userId Int @map("user_id") + user User @relation(fields: [userId], references: [id], onDelete: Cascade) @@index([userId]) @@map("passkeys") @@ -79,14 +79,14 @@ model PendingPasskeyRegistration { } model Operation { - id String @id - userId Int @map("user_id") - clientId String @map("client_id") - serverSeq Int @map("server_seq") - actionType String @map("action_type") - opType String @map("op_type") - entityType String @map("entity_type") - entityId String? @map("entity_id") + id String @id + userId Int @map("user_id") + clientId String @map("client_id") + serverSeq Int @map("server_seq") + actionType String @map("action_type") + 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 @@ -94,40 +94,48 @@ model Operation { // 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") - schemaVersion Int @map("schema_version") - clientTimestamp BigInt @map("client_timestamp") - receivedAt BigInt @map("received_at") - isPayloadEncrypted Boolean @default(false) @map("is_payload_encrypted") - syncImportReason String? @map("sync_import_reason") - repairBaseServerSeq Int? @map("repair_base_server_seq") + entityIds String[] @default([]) @map("entity_ids") + payload Json + payloadBytes BigInt @default(0) @map("payload_bytes") + vectorClock Json @map("vector_clock") + schemaVersion Int @map("schema_version") + clientTimestamp BigInt @map("client_timestamp") + receivedAt BigInt @map("received_at") + isPayloadEncrypted Boolean @default(false) @map("is_payload_encrypted") + syncImportReason String? @map("sync_import_reason") + repairBaseServerSeq Int? @map("repair_base_server_seq") - user User @relation(fields: [userId], references: [id], onDelete: Cascade) + user User @relation(fields: [userId], references: [id], onDelete: Cascade) @@unique([userId, serverSeq]) @@index([userId, entityType, entityId, serverSeq]) @@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). + // Multi-entity conflict lookups (#8334) probe entity_ids with @> / &&. This MUST stay + // modelled here, not only in the 20260613000001 raw migration: `prisma db push` builds + // from this file alone, so while the GIN existed only as raw SQL the CI, E2E and + // documented manual-setup databases had NO index at all — and detectConflictForEntity's + // MATERIALIZED CTE carries no user_id predicate, so without it every probe Seq Scans + // EVERY tenant's rows, twice per accepted op. Worse than the outage this replaced. + // The map name matches the raw migration, so migrate-based deployments see no drift. + @@index([entityIds], type: Gin, map: "operations_entity_ids_gin") + // Restore-point opType lookups use a raw PARTIAL index (20260512000000). That one + // genuinely cannot be modelled — Prisma has no partial-index syntax — so db-push + // databases still lack it. Same trap, different index; tracked in #9192. @@map("operations") } model UserSyncState { - userId Int @id @map("user_id") - lastSeq Int @default(0) @map("last_seq") - lastSnapshotSeq Int? @map("last_snapshot_seq") - snapshotData Bytes? @map("snapshot_data") - snapshotAt BigInt? @map("snapshot_at") - snapshotSchemaVersion Int? @default(1) @map("snapshot_schema_version") - latestFullStateSeq Int? @map("latest_full_state_seq") - latestFullStateVectorClock Json? @map("latest_full_state_vector_clock") + userId Int @id @map("user_id") + lastSeq Int @default(0) @map("last_seq") + lastSnapshotSeq Int? @map("last_snapshot_seq") + snapshotData Bytes? @map("snapshot_data") + snapshotAt BigInt? @map("snapshot_at") + snapshotSchemaVersion Int? @default(1) @map("snapshot_schema_version") + latestFullStateSeq Int? @map("latest_full_state_seq") + latestFullStateVectorClock Json? @map("latest_full_state_vector_clock") - user User @relation(fields: [userId], references: [id], onDelete: Cascade) + user User @relation(fields: [userId], references: [id], onDelete: Cascade) @@map("user_sync_state") } @@ -141,7 +149,7 @@ model SyncDevice { lastAckedSeq Int @default(0) @map("last_acked_seq") createdAt BigInt @map("created_at") - user User @relation(fields: [userId], references: [id], onDelete: Cascade) + user User @relation(fields: [userId], references: [id], onDelete: Cascade) @@id([userId, clientId]) @@map("sync_devices") diff --git a/packages/super-sync-server/src/sync/conflict.ts b/packages/super-sync-server/src/sync/conflict.ts index 101f26323e..7600a4e00a 100644 --- a/packages/super-sync-server/src/sync/conflict.ts +++ b/packages/super-sync-server/src/sync/conflict.ts @@ -204,9 +204,11 @@ export const resolveConflictForExistingOp = ( }; /** - * Checks conflicts for the common single-entity upload path using Prisma's - * typed model API. Multi-entity operations use the batched raw-SQL path above - * to avoid one round trip per entity. + * Checks conflicts for the common single-entity upload path. TWO separately-indexed + * lookups: a typed `findFirst` on the scalar entity_id, plus a raw-SQL MATERIALIZED CTE + * over entity_ids — see the PERF note below, the combined filter caused an outage. + * Multi-entity operations use the batched raw-SQL path above instead, to avoid one round + * trip per entity. */ export const detectConflictForEntity = async ( userId: number, @@ -219,25 +221,152 @@ export const detectConflictForEntity = async ( // 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, - OR: [{ entityId }, { entityIds: { has: entityId } }], - }, + // PERF — this must stay TWO separately-indexed lookups. It was one Prisma filter + // (`OR: [{ entityId }, { entityIds: { has: entityId } }]` + + // `orderBy: { serverSeq: 'desc' }`), and that took production down. The OR spans + // two different indexes — the (user_id, entity_type, entity_id, server_seq) btree + // and the entity_ids GIN — and GIN cannot supply server_seq ordering, so the + // planner abandons BOTH index paths, filters the (user_id, entity_type) slice and + // sorts it (or walks (user_id, server_seq) backwards, betting `LIMIT 1` resolves + // early). For an entity with no matching rows — the first-ever op for a new task, + // the single most common upload there is — nothing bounds that work and it reads + // the user's whole slice. The batch unnest paths (detectConflictForEntities / + // prefetchLatestEntityOpsForBatch) cannot make that EARLY-EXIT bet — no LIMIT, and + // DISTINCT ON forces full evaluation. They carry the same two-index OR, though, so + // the SLICE-SCAN degeneracy is not excluded for them, and nothing EXPLAINs either + // batch query today (#9205). + // + // Scalar branch: the (user_id, entity_type, entity_id, server_seq) btree covers all + // three equality columns PLUS the sort column, so this is a direct index seek and a + // one-step backward walk. No trap — the ORDER BY is served by the index itself. + const scalarOp = await tx.operation.findFirst({ + where: { userId, entityType: op.entityType, entityId }, select: { actionType: true, clientId: true, vectorClock: true, serverSeq: true }, orderBy: { serverSeq: 'desc' }, }); + // Array branch — raw SQL, and the MATERIALIZED CTE is load-bearing. + // + // What the CTE removes structurally is the COMPETING BTREE: inside it the only + // predicate is `entity_ids @> ...`, so the composite btree has no usable leading + // column and GIN is the only INDEX available at any cost estimate. MATERIALIZED is + // what stops the outer user_id / entity_type predicates being pushed down, which + // would hand the btree back. + // + // That is NOT a guarantee that GIN is chosen. A sequential scan is always still + // available, and wins when the probed id is unselective — both plans have been + // reproduced on PG16 depending on row shape. GIN winning on production-shaped data is a MEASURED + // outcome, not a structural one. Re-measure after any change rather than trusting + // this paragraph; a confidently-worded comment asserting what the planner "will" do + // is what preceded the outage. + // Every simpler form reads the whole (user_id, entity_type) slice instead: the + // array-only `findFirst` + `orderBy`, Prisma's `aggregate({ _max })`, and the flat + // `SELECT MAX(server_seq) ... AND @>`. (The outage itself was the combined OR + // described above, not any of these.) + // + // Do NOT "simplify" this without measuring under `plan_cache_mode = + // force_generic_plan`. Prisma sends parameterized prepared statements; under the + // default `auto` Postgres plans the first ~5 executions as CUSTOM, then compares the + // generic cost against the average custom cost and MAY switch to a generic plan. That + // is a cost comparison, not an automatic switch — a statement can stay on custom plans + // indefinitely. THIS statement was observed going generic on production + // (pg_prepared_statements: custom_plans=5, generic_plans=15), and a generic plan cannot + // see the parameter values. `EXPLAIN` with literal constants is different again, and + // every broken form above looks perfect that way. + // conflict-entity-lookup-plan.pglite.spec.ts measures the generic mode correctly and + // fails on a block budget; it does NOT cover custom plans. + // + // Adding `server_seq > ` to narrow the CTE was evaluated and REJECTED: + // under generic planning the bound is invisible, so it lands as a post-GIN Filter + // and buys nothing, and inside the CTE it lets a custom plan bitmap-scan + // (user_id, server_seq) on `server_seq > $4` with NO leading-column bound — a + // full-index scan across every user's history. + // + // Isolation: the CTE matches by entity id across ALL users and the outer WHERE + // enforces the user boundary, so this is CORRECT but NOT cost-bounded per user. Some + // entity ids are byte-identical across every tenant: the bulk `sortBoards` action + // (boards.actions.ts) stores the hard-coded 'EISENHOWER_MATRIX' / 'KANBAN_DEFAULT' + // ids (boards.const.ts) in entity_ids, so probing one walks every tenant's matching + // rows and the cost scales with total server population, bounded by nothing. + // Single-entity writes against a shared id — updateTag({ id: 'TODAY' }), + // updateBoard({ id: 'KANBAN_DEFAULT' }) — do not POPULATE the GIN under that id: + // getStoredEntityIds persists '{}' for them. They are still a PROBE vector, though. + // Each one routes through detectConflictForEntity and probes that shared literal, so + // it walks every tenant's matching rows without contributing any of its own. + // + // The fix is a GIN index on the expression (ARRAY['u:' || user_id] || entity_ids), + // which makes the probe flat. It needs no btree_gin extension (the operand is text[], + // served by the built-in array_ops) and is measurable in PGlite. Not done here + // because it is a real tradeoff, not a free win: this predicate must be rewritten to + // match the expression or the index is simply ignored. Make it PARTIAL + // (WHERE entity_ids <> '{}') and it covers only the multi-entity minority instead of + // every row. That is lossless: a single-entity op stores '{}', so its indexed + // expression is just ARRAY['u:'] and can never contain the real entity id a probe + // carries. That losslessness is a claim about the DATA and holds by inspection: + // getStoredEntityIds collapses single-entity sets to [], entity_ids is NOT NULL + // DEFAULT '{}', and '{}' @> ARRAY[] is false. + // + // UNMEASURED, and do not build on it until you have: whether the PARTIAL form is + // usable at all is a claim about the PLANNER, and nobody has run it. The query would + // have to carry a matching `AND entity_ids <> '{}'` and Postgres would have to prove + // that implies the index predicate — for an array `<>`, which is not something this + // comment has any evidence about. EXPLAIN it under force_generic_plan first (see the + // plan-cache note above). A comment asserting unmeasured planner behaviour is what + // preceded the outage; this paragraph is a lead to chase, not a design to trust. + // + // The outer user_id predicate stays load-bearing either way — the 'u:' prefix is a + // namespace, not a security boundary. + // + // Sequential, never Promise.all: `tx` is a single-connection interactive transaction + // client and concurrent queries on it are unsafe. + const arrayBranchRows = await tx.$queryRaw>` + WITH cand AS MATERIALIZED ( + SELECT user_id, entity_type, server_seq + FROM operations + WHERE entity_ids @> ARRAY[${entityId}]::text[] + ) + SELECT MAX(server_seq)::int AS "maxSeq" + FROM cand + WHERE user_id = ${userId} AND entity_type = ${op.entityType} + `; + // INVARIANT: an aggregate with no GROUP BY returns exactly one row, so the `?.` + // fold below is unreachable and `maxSeq` is null only when nothing matched. + // `GROUP BY user_id` would NOT break that: zero groups arise only when zero rows + // matched, which folds to null and correctly reads as "no prior op" — the same + // outcome as `MAX` over no rows. What DOES break it is any grouping that can return + // more than one row (`GROUP BY server_seq`, say), because `[0]` then takes an + // arbitrary group instead of the maximum and can under-report the latest op — silent + // acceptance of a conflicting write, not an error. No runtime guard here on purpose + // (it could never fire today); if you change the shape of this query, change this + // fold with it. + const arrayBranchMaxSeq = arrayBranchRows[0]?.maxSeq ?? null; + + // Fetch the array-branch row only when it actually beats the scalar branch. + // (user_id, server_seq) is UNIQUE, so this is a single indexed point lookup, and + // ties need no tie-break: an equal server_seq IS the same row. + const arrayWins = + arrayBranchMaxSeq !== null && (!scalarOp || arrayBranchMaxSeq > scalarOp.serverSeq); + const arrayOp = arrayWins + ? await tx.operation.findUnique({ + where: { userId_serverSeq: { userId, serverSeq: arrayBranchMaxSeq } }, + select: { + actionType: true, + clientId: true, + vectorClock: true, + serverSeq: true, + }, + }) + : null; + + // This `??` carries TWO meanings: "the array branch did not win" and "the array + // branch won but its row was not there". Only the first is reachable — the MAX came + // from a row in this transaction's snapshot and (user_id, server_seq) is unique, so + // at RepeatableRead the row cannot vanish under us. If the isolation level is ever + // lowered, the second case silently falls back to the STALE scalar row and accepts a + // write that should have conflicted. Retiring the separate findUnique (see the + // row-returning CTE, #9197) removes this ambiguity rather than guarding it. + const existingOp = arrayOp ?? scalarOp; + // Histories written before schema v2 persist migrated task settings under // the raw `GLOBAL_CONFIG:misc` key. Consult that key as an alias when the // incoming write targets `tasks`; no backfill (and no payload decryption) is @@ -458,10 +587,15 @@ const isLegacyMiscConfigOperation = (op: Operation): boolean => /** * 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. + * already covered by the scalar entity_id store an empty array; any other set is + * stored in full. + * + * This makes the entity_ids GIN index CHEAP, not absent. Postgres indexes an empty + * array as one degenerate GIN_CAT_EMPTY_ITEM key, so single-entity ops DO have index + * entries, the index grows with row count (measured on an all-empty column: 10k rows + * → 5 pages, 30k → 7, 60k → 11) and every insert still touches it. The win is one + * key per single-entity op instead of one key per member, which keeps + * `entity_ids @> ARRAY[id]` probes bounded by genuine multi-entity matches. * * 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 diff --git a/packages/super-sync-server/tests/conflict-detection.spec.ts b/packages/super-sync-server/tests/conflict-detection.spec.ts index 0541039fd6..8ec1c38f19 100644 --- a/packages/super-sync-server/tests/conflict-detection.spec.ts +++ b/packages/super-sync-server/tests/conflict-detection.spec.ts @@ -8,6 +8,8 @@ vi.mock('../src/db', async () => { const { applyOperationSelect, hasOperationUniqueConflict, + isEntityArrayBranchQuery, + entityArrayBranchRows, testState: state, } = await import('./sync.service.test-state'); const { Prisma: PrismaModule } = await import('@prisma/client'); @@ -63,27 +65,10 @@ 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; - } + // Single-entity conflict lookup, scalar branch: where { userId, entityType, + // entityId }. The entity_ids half is a separate $queryRaw call below — the + // two were one OR filter until it degenerated into a full history scan in + // production (see the PERF note in conflict.ts detectConflictForEntity). // Scalar-only lookup (other callers): where { userId, entityType, entityId }. if (args.where?.entityId && args.where?.entityType) { state.entityConflictFindFirstCount++; @@ -120,6 +105,15 @@ vi.mock('../src/db', async () => { .slice(0, args.take || 500); }), findUnique: vi.fn().mockImplementation(async (args: any) => { + // (user_id, server_seq) compound unique — fetches the array branch's winner. + const compound = args.where?.userId_serverSeq; + if (compound) { + const match = Array.from(state.operations.values()).find( + (op: any) => + op.userId === compound.userId && op.serverSeq === compound.serverSeq, + ); + return applyOperationSelect(match, args.select) || null; + } if (args.where?.id) { return ( applyOperationSelect(state.operations.get(args.where.id), args.select) || null @@ -187,6 +181,12 @@ vi.mock('../src/db', async () => { $executeRaw: vi.fn().mockResolvedValue(0), $queryRaw: vi.fn().mockImplementation(async (strings: any, ...params: unknown[]) => { const sql = Array.isArray(strings) ? strings.join('') : String(strings); + // Array branch of the single-entity conflict lookup: MAX(server_seq) over + // `entity_ids @> ARRAY[id]`, kept separate from the scalar findFirst above. + if (isEntityArrayBranchQuery(strings)) { + state.entityConflictArrayQueryCount++; + return entityArrayBranchRows(state.operations, params); + } // Full-state op uploads aggregate prior vector clocks via $queryRaw. if (sql.includes('jsonb_each_text(vector_clock)')) { const [txUserId, beforeServerSeq] = params as [number, number]; @@ -215,6 +215,12 @@ vi.mock('../src/db', async () => { return [{ lastSeq: state.userSyncStates.get(txUserId)?.lastSeq ?? 0 }]; } + // Anything left must be the batched multi-entity conflict lookup. Assert that + // rather than assuming it: falling through and reinterpreting an unrelated + // query as this one is how a mock silently answers a call it never modelled. + if (!sql.includes('DISTINCT ON')) { + throw new Error(`Unmocked raw query in tx: ${sql}`); + } const [userId, entityType, entityIdsSql] = params as [number, string, Prisma.Sql]; state.batchConflictQueryCount++; if (!Array.isArray(entityIdsSql.values)) { diff --git a/packages/super-sync-server/tests/conflict-entity-lookup-plan.pglite.spec.ts b/packages/super-sync-server/tests/conflict-entity-lookup-plan.pglite.spec.ts new file mode 100644 index 0000000000..9e7631026c --- /dev/null +++ b/packages/super-sync-server/tests/conflict-entity-lookup-plan.pglite.spec.ts @@ -0,0 +1,882 @@ +import { describe, it, expect, beforeAll, afterAll } from 'vitest'; +import { PGlite } from '@electric-sql/pglite'; +import { detectConflictForEntity } from '../src/sync/conflict'; +import { isEntityArrayBranchQuery } from './sync.service.test-state'; +import type { Operation } from '../src/sync/sync.types'; + +/** + * Production incident regression: the single-entity conflict lookup read a user's + * entire (user_id, entity_type) slice on every upload of a not-yet-seen entity. + * The mechanism, the rejected alternatives and the isolation caveat are documented + * once, at detectConflictForEntity in src/sync/conflict.ts — not repeated here. + * + * This spec runs the REAL detectConflictForEntity against in-process Postgres + * (PGlite — no Docker, no DATABASE_URL) through a tx shim that renders each Prisma + * call as the SQL Prisma emits and EXPLAINs it. The array branch is NOT rebuilt from + * a constant: the shim reconstructs it from the actual tagged-template text, so the + * SQL under test is byte-for-byte what conflict.ts sends. That is what lets a change + * to the aggregate (MAX -> MIN), the fence (dropping MATERIALIZED) or the CTE shape + * fail here instead of passing against a stale copy. + * + * MEASURE WITH `force_generic_plan`, NEVER WITH LITERALS. Prisma sends parameterized + * prepared statements; under the default `auto` Postgres plans the first ~5 executions as + * CUSTOM, then compares the generic cost against the average custom cost and MAY switch to + * a generic plan — a cost comparison, not an automatic switch, so a statement can stay on + * custom plans indefinitely. THIS one was observed going generic on production, and a + * generic plan cannot see parameter values, so that is the mode this file covers. + * Production also serves custom plans and this file does NOT cover them (a + * custom-plan-only regression is possible; see the rejected `server_seq >` narrowing in + * conflict.ts). `EXPLAIN` with literal constants is a third thing again, and is the trap: + * this file once tested that way and the blind spot passed two designs that were + * catastrophic in production. EVERYTHING here — including the shim — goes through + * explainGeneric. If you add a shape, use explainGeneric. + * + * WHY THE SEED SHAPE IS WHAT IT IS — do not "simplify" it: + * + * - entity_ids stays '{}' on EVERY row. The planner only mis-plans when it has no + * array-element statistics for entity_ids, falling back to a default `@>` + * selectivity. That is the DEPLOYED state: entity_ids was added by migration + * 20260613000000 with no backfill, and getStoredEntityIds stores [] for every + * single-entity op. Populating the column here silently disarms the regression. + * - MANY users and MANY entity types. This is the load-bearing part. With one user + * and one entity_type the GIN estimate (which scales with the whole table) and the + * btree-slice estimate (N / (users x entity_types)) cover the SAME rows, so GIN + * always wins on cost and no regression is detectable. That degenerate shape is + * why this suite could not catch the outage. At 20k rows for the probed user plus + * 20k spread over ~20k other users across 8 entity types, PGlite reproduces the + * SHAPE of the production mis-plan (same nodes, same discarded-row signature) and + * the regression lands ~2.7x over the block budget — 816 against MAX_BLOCKS 300. + * Node-for-node identity with production is NOT claimed; see the fidelity limit + * below. + * - THE INDEXES ARE CREATED BEFORE THE ROWS. This is not cosmetic ordering. Building + * the GIN after a bulk load produces a compact, pending-list-free index, and the + * array branch then measures 2 blocks — a number production only sees right after a + * vacuum. In production the index pre-exists and rows arrive one op at a time, so + * inserts land in the GIN pending list (fastupdate defaults to on) which `@>` must + * scan linearly: same data, same query, 140 blocks and a 14x larger index (1120 kB + * vs 72 kB). VACUUM flushes the pending list and restores 2 blocks, so the real cost + * OSCILLATES across the autovacuum cycle. This suite therefore seeds in production + * order and never vacuums, measuring the dirty end of that cycle rather than the + * freshly-vacuumed end — but 140 is NOT a ceiling: the pending list is bounded by + * gin_pending_list_limit (4MB default), not by anything in this seed, so a + * production write burst can be worse. The budget below is calibrated to catch the + * mis-plan, NOT to certify a maximum. Setting fastupdate=off stops new entries + * queueing — it does not flush what is already pending, which still needs a VACUUM — + * and so removes the oscillation going forward. + * + * REMAINING FIDELITY LIMIT: PGlite is not the production cluster — different major + * version, and it reports every block as a cache hit, so these counts cannot model + * cold-cache I/O. + */ + +const OWN_OPS = 20_000; +const OTHER_OPS = 20_000; +const USER_ID = 1; +/** Entity types are spread across the seed so the btree slice is N/(users x types). */ +const ENTITY_TYPES = [ + 'TASK', + 'PROJECT', + 'TAG', + 'NOTE', + 'BOARD', + 'GLOBAL_CONFIG', + 'SIMPLE_COUNTER', + 'TASK_REPEAT_CFG', +]; +/** seq % ENTITY_TYPES.length === 0 => 'TASK', so this row is in the probed slice. */ +const DEEP_ENTITY_SEQ = 16; + +const CREATE_TABLE = ` + CREATE TABLE operations ( + id text PRIMARY KEY, + user_id integer NOT NULL, + client_id text NOT NULL, + -- integer, NOT bigint: production maps serverSeq as Prisma Int, so this matches the + -- deployed column type. It is a FIDELITY fix and nothing more — it does not make + -- dropping the production ::int cast catchable, because over an integer column MAX() + -- already returns integer and the cast is a no-op either way. + server_seq integer NOT NULL, + action_type text NOT NULL, + entity_type text NOT NULL, + entity_id text, + entity_ids text[] NOT NULL DEFAULT '{}', + schema_version integer NOT NULL DEFAULT 1, + vector_clock jsonb NOT NULL + ); +`; + +// A deliberate SUBSET of prisma/schema.prisma + the migrations — the three indexes this +// lookup can actually ride: 0_init (the (user_id, server_seq) unique the backward walk +// rides on), 20260511000000 (the entity btree) and 20260613000001 (the entity_ids GIN). +// Production also has a PK on id plus (user_id, client_id) and (user_id, received_at) +// btrees; they are left out because no predicate here can use them. Add them if a new +// shape could. +const CREATE_INDEXES = ` + CREATE UNIQUE INDEX operations_user_id_server_seq_key + ON operations (user_id, server_seq); + CREATE INDEX operations_user_id_entity_type_entity_id_server_seq_idx + ON operations (user_id, entity_type, entity_id, server_seq); + CREATE INDEX operations_entity_ids_gin ON operations USING GIN (entity_ids); +`; + +const INSERT_COLS = + 'id,user_id,client_id,server_seq,action_type,entity_type,entity_id,entity_ids,' + + 'schema_version,vector_clock'; + +type PlanStats = { + blocks: number; + rowsFiltered: number; + sql: string[]; + rawSql: string[]; + /** Plan node types + index names, per measured query, parallel to `sql`. */ + nodes: string[]; +}; +type PlanNode = Record; +type Measured = { blocks: number; rowsFiltered: number; nodes: string }; + +const newStats = (): PlanStats => ({ + blocks: 0, + rowsFiltered: 0, + sql: [], + rawSql: [], + nodes: [], +}); + +/** + * Walks the plan tree for node names and filtered-row counts. + * + * Blocks are deliberately NOT summed here: `Shared Hit/Read Blocks` are CUMULATIVE, + * so a parent already includes everything its children read. Summing every node + * double-counts the same buffers once per level of nesting, inflating deep plans + * (the CTE form nests one level deeper than the flat one) and biasing the budgets + * against the new code. The ROOT node's value is the true total. + */ +const accumulatePlan = (node: PlanNode, stats: PlanStats, nodes: string[]): void => { + stats.rowsFiltered += (node['Rows Removed by Filter'] as number) ?? 0; + nodes.push( + `${node['Node Type']}${node['Scan Direction'] ? ' ' + node['Scan Direction'] : ''}` + + `${node['Index Name'] ? ' on ' + node['Index Name'] : ''}`, + ); + for (const child of (node.Plans as PlanNode[]) ?? []) { + accumulatePlan(child, stats, nodes); + } +}; + +const rootBlocks = (node: PlanNode): number => + ((node['Shared Hit Blocks'] as number) ?? 0) + + ((node['Shared Read Blocks'] as number) ?? 0); + +const toSqlLiteral = (value: unknown): string => { + if (value === null || value === undefined) return 'NULL'; + if (typeof value === 'number' || typeof value === 'bigint') return String(value); + if (Array.isArray(value)) { + return `ARRAY[${value.map(toSqlLiteral).join(',')}]::text[]`; + } + return `'${String(value).replace(/'/g, "''")}'`; +}; + +/** + * EXPLAIN through PREPARE/EXECUTE under `force_generic_plan` — the ONLY faithful way + * to see what production gets. The params are rendered as literals for EXECUTE, but + * the PLAN is built at PREPARE time with the values invisible, which is exactly the + * situation Prisma puts Postgres in. + */ +let preparedCounterId = 0; +const explainGeneric = async ( + db: PGlite, + sql: string, + params: readonly unknown[], +): Promise => { + const name = `plan_probe_${preparedCounterId++}`; + const args = params.map(toSqlLiteral).join(', '); + await db.exec(`SET plan_cache_mode = force_generic_plan`); + await db.exec(`PREPARE ${name} AS ${sql}`); + try { + const res = await db.query>( + `EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) EXECUTE ${name}${args ? `(${args})` : ''}`, + ); + const plan = (res.rows[0]['QUERY PLAN'] as PlanNode[])[0].Plan as PlanNode; + const stats = newStats(); + const nodes: string[] = []; + accumulatePlan(plan, stats, nodes); + return { + blocks: rootBlocks(plan), + rowsFiltered: stats.rowsFiltered, + nodes: nodes.join(' -> '), + }; + } finally { + await db.exec(`DEALLOCATE ${name}`); + await db.exec(`SET plan_cache_mode = auto`); + } +}; + +const incomingOp = ( + entityId: string, + overrides: Partial> = {}, +): Operation => + ({ + id: 'op-incoming', + clientId: 'uploader', + actionType: '[Task] Update', + opType: 'UPD', + entityType: 'TASK', + entityId, + vectorClock: { uploader: 1 }, + timestamp: 1, + schemaVersion: 1, + ...overrides, + }) as unknown as Operation; + +/** Column list per Prisma `select` key. An unmapped key must fail loudly, not vanish. */ +const COLUMN_SQL: Record = { + actionType: 'action_type AS "actionType"', + clientId: 'client_id AS "clientId"', + vectorClock: 'vector_clock AS "vectorClock"', + serverSeq: 'server_seq AS "serverSeq"', + entityId: 'entity_id AS "entityId"', + entityType: 'entity_type AS "entityType"', +}; + +/** + * Honouring `select` is load-bearing, not cosmetic: conflict.ts reads + * existingOp.actionType to let concurrent time-tracking deltas merge. A shim that + * always returned every column would keep passing if actionType were dropped from + * the array-branch select, which is a silent rejection of tracked time. + */ +const selectCols = (select?: Record): string => { + if (!select) return Object.values(COLUMN_SQL).join(', '); + const cols = Object.entries(select) + .filter(([, isSelected]) => isSelected) + .map(([key]) => { + const col = COLUMN_SQL[key]; + if (!col) throw new Error(`Shim has no column mapping for select key "${key}"`); + return col; + }); + if (cols.length === 0) throw new Error('Shim received an empty select'); + return cols.join(', '); +}; + +/** + * Renders the Prisma calls detectConflictForEntity makes as the SQL Prisma emits, + * EXPLAIN-ing each one into `stats`. It deliberately also renders the OLD combined + * OR filter: reverting the fix must fail this spec on the BUDGET (proving the plan + * degenerated), not on an unsupported-shape error. + */ +const makeMeasuringTx = (db: PGlite, stats: PlanStats): unknown => { + // Prisma renders `entityIds: { has: x }` as `entity_ids @> ARRAY[x]`. + const renderConditions = (where: Record, params: unknown[]): string[] => { + const push = (value: unknown): string => `$${params.push(value)}`; + const conds: string[] = []; + if (where.userId !== undefined) conds.push(`user_id = ${push(where.userId)}`); + if (where.entityType !== undefined) { + conds.push(`entity_type = ${push(where.entityType)}`); + } + if (where.entityId !== undefined) conds.push(`entity_id = ${push(where.entityId)}`); + if (where.entityIds?.has !== undefined) { + conds.push(`entity_ids @> ARRAY[${push(where.entityIds.has)}]::text[]`); + } + if (where.schemaVersion?.lt !== undefined) { + conds.push(`schema_version < ${push(where.schemaVersion.lt)}`); + } + if (Array.isArray(where.OR)) { + const alternatives = where.OR.map( + (alt: Record) => + renderConditions(alt, params).join(' AND ') || 'TRUE', + ); + conds.push(`(${alternatives.join(' OR ')})`); + } + return conds; + }; + + const runMeasured = async ( + sql: string, + params: unknown[], + ): Promise[]> => { + const measured = await explainGeneric(db, sql, params); + stats.blocks += measured.blocks; + stats.rowsFiltered += measured.rowsFiltered; + stats.sql.push(sql); + stats.nodes.push(measured.nodes); + return (await db.query>(sql, params)).rows; + }; + + const normalize = (row?: Record): Record | null => { + if (!row) return null; + return 'serverSeq' in row ? { ...row, serverSeq: Number(row.serverSeq) } : row; + }; + + return { + operation: { + findFirst: async (args: Record) => { + const params: unknown[] = []; + const conds = renderConditions(args.where, params); + const order = + args.orderBy?.serverSeq === 'desc' + ? 'ORDER BY server_seq DESC' + : args.orderBy?.serverSeq === 'asc' + ? 'ORDER BY server_seq ASC' + : ''; + const rows = await runMeasured( + `SELECT ${selectCols(args.select)} FROM operations` + + ` WHERE ${conds.join(' AND ')} ${order} LIMIT 1`, + params, + ); + return normalize(rows[0]); + }, + findUnique: async (args: Record) => { + const { userId, serverSeq } = args.where.userId_serverSeq; + const rows = await runMeasured( + `SELECT ${selectCols(args.select)} FROM operations` + + ` WHERE user_id = $1 AND server_seq = $2 LIMIT 1`, + [userId, serverSeq], + ); + return normalize(rows[0]); + }, + }, + // Array branch. Rebuilt from the REAL tagged template — the literal text + // conflict.ts sends, with `$n` substituted in template order — so the aggregate, + // the MATERIALIZED fence and the CTE structure are all under test here rather + // than compared against a copy that can drift. + $queryRaw: async (strings: TemplateStringsArray, ...values: unknown[]) => { + if (!isEntityArrayBranchQuery(strings)) { + throw new Error(`Unexpected raw query: ${strings.join('?')}`); + } + const sql = strings.reduce( + (acc, part, i) => acc + part + (i < values.length ? `$${i + 1}` : ''), + '', + ); + stats.rawSql.push(sql); + const rows = await runMeasured(sql, values); + const max = rows[0]?.maxSeq; + return [{ maxSeq: max === null || max === undefined ? null : Number(max) }]; + }, + }; +}; + +// Post-fix both branches are index lookups bounded by actually-matching rows. +// Measured on this seed, seeded in production order: the array branch reads 140 blocks +// (GIN pending list — see the header) and the scalar 3, both filtering NOTHING. The +// regression form pinned by the CANARY below — the combined OR that caused the outage — +// reads 816 blocks and filters 2500, the probed user's whole TASK slice. (The other +// broken shapes named in conflict.ts degrade the same way, but only this one is +// measured here.) +// +// `rowsFiltered === 0` is the load-bearing assertion; the budget is the backstop. The +// filtered count is the regression's actual signature — "read the user's history and +// threw it away" — and it is scale-free, so it keeps its meaning if the seed changes. +// The block budget is NOT scale-free: it sits ~2x above the measured 143 and only ~2.7x +// below the regression's 816, a margin that holds only for THIS seed's +// user/entity-type ratio. (The wider ~5.7x is regression-to-measured, not +// regression-to-budget — the real headroom is the smaller number.) If +// you change the seed, re-derive it — and the canary test below exists to fail loudly if +// the seed ever stops reproducing the mis-plan at all. +const MAX_BLOCKS = 300; + +const expectWithinBudget = (measured: { blocks: number; rowsFiltered: number }): void => { + expect(measured.rowsFiltered).toBe(0); + expect(measured.blocks).toBeLessThan(MAX_BLOCKS); +}; + +describe('detectConflictForEntity does not scan the history (PGlite)', () => { + let db: PGlite; + + beforeAll(async () => { + db = new PGlite(); + await db.waitReady; + await db.exec(CREATE_TABLE); + // BEFORE the rows, as in production — see the header note. Building the GIN after + // the load yields a pending-list-free index that measures 2 blocks instead of 140. + await db.exec(CREATE_INDEXES); + + // entity_ids stays '{}' on EVERY row — see the header note. Populating it here + // gives the planner array statistics and disarms the regression. + let rows: string[] = []; + const flush = async (): Promise => { + if (rows.length === 0) return; + await db.exec(`INSERT INTO operations (${INSERT_COLS}) VALUES ${rows.join(',')}`); + rows = []; + }; + const entityTypeFor = (n: number): string => ENTITY_TYPES[n % ENTITY_TYPES.length]; + + for (let seq = 1; seq <= OWN_OPS; seq++) { + rows.push( + `('op-${seq}', ${USER_ID}, 'seed-client', ${seq}, '[Task] Update',` + + ` '${entityTypeFor(seq)}', 'task-${seq}', '{}', 1, '{"seed-client":${seq}}')`, + ); + if (rows.length === 1000) await flush(); + } + // A second population of comparable size spread over ~20k OTHER users, so the + // per-user btree slice is a small fraction of the table the GIN estimate sees. + for (let i = 1; i <= OTHER_OPS; i++) { + rows.push( + `('other-${i}', ${1000 + i}, 'seed-other', ${i}, '[Task] Update',` + + ` '${entityTypeFor(i)}', 'otask-${i}', '{}', 1, '{"seed-other":${i}}')`, + ); + if (rows.length === 1000) await flush(); + } + await flush(); + + // ANALYZE so the planner works from real statistics rather than defaults on an + // unanalyzed table. Deliberately NOT VACUUM: that would flush the GIN pending list + // and measure the freshly-vacuumed best case instead of the steady state. + await db.exec('ANALYZE operations'); + }, 120_000); + + afterAll(async () => { + await db.close(); + }); + + it('reads a bounded amount for a BRAND-NEW entity (the incident case)', async () => { + const stats = newStats(); + + // The worst case and the common case at once: the first-ever op for a new task. + // Nothing matches, so a LIMIT-1 backward walk never finds its early exit. + const result = await detectConflictForEntity( + USER_ID, + incomingOp('task-brand-new'), + 'task-brand-new', + makeMeasuringTx(db, stats) as never, + ); + + expect(result.hasConflict).toBe(false); + expectWithinBudget(stats); + + // Asserted on the REAL template rather than a copy of it. Inside the CTE the only + // predicate is `entity_ids @> ...`, so the composite btree has no usable leading + // column and GIN is the only INDEX available at any cost estimate — that much is + // structural, and it is why every regression form (inlining the CTE, flattening it, + // reinstating the OR) reaches the btree instead and is caught here even if a future + // seed stops blowing the budget. + // + // It does NOT prove GIN is forced: a sequential scan remains available at any time + // and wins for an unselective id (a globally shared entity id does exactly that). + // This pins the MEASURED plan for this seed, not a guarantee. + expect(stats.rawSql).toHaveLength(1); + const arrayBranchPlan = stats.nodes[stats.sql.indexOf(stats.rawSql[0])]; + expect(arrayBranchPlan).toContain('operations_entity_ids_gin'); + expect(arrayBranchPlan).not.toContain( + 'operations_user_id_entity_type_entity_id_server_seq_idx', + ); + expect(arrayBranchPlan).not.toContain('Backward'); + }); + + it('reads a bounded amount for an entity deep in the history', async () => { + const stats = newStats(); + + await detectConflictForEntity( + USER_ID, + incomingOp(`task-${DEEP_ENTITY_SEQ}`), + `task-${DEEP_ENTITY_SEQ}`, + makeMeasuringTx(db, stats) as never, + ); + + expectWithinBudget(stats); + }); + + /** + * A CANARY, not a test of the source. It EXPLAINs a hardcoded copy of the OLD outage + * query — a string that exists only here — so it responds to no change in conflict.ts + * and proves nothing about the shipped code. Its one job is to prove that THIS SEED + * still reproduces the mis-plan, which is what gives the budget above its meaning. + * + * That job is real: the budget is an absolute, and its detection power depends on the + * seed's user/entity-type ratio. Shrink the seed, or collapse it toward one user and + * one entity type, and the regression's cost falls below MAX_BLOCKS while every other + * test in this file keeps passing — a suite that has silently stopped being able to + * fail. This fails instead. + * + * It replaced five near-identical shapes (the inlined CTE, the flat MAX, Prisma's + * aggregate form, the naive array-only fix, and this one). They were the same + * hardcoded-string assertion five times over and responded to no source mutation, + * while the shipped query's own budget and structural plan assertions above cover + * those regressions where it counts — on the real SQL. Verified by mutation: inlining + * the CTE in conflict.ts fails the two end-to-end budget tests, not this block. + */ + it('CANARY: the seed still reproduces the mis-plan the budget is calibrated against', async () => { + const regressed = await explainGeneric( + db, + `SELECT server_seq FROM operations + WHERE user_id = $1 AND entity_type = $2 + AND (entity_id = $3 OR entity_ids @> ARRAY[$3]::text[]) + ORDER BY server_seq DESC LIMIT 1`, + [USER_ID, 'TASK', 'task-brand-new'], + ); + + expect(regressed.blocks).toBeGreaterThan(MAX_BLOCKS); + // It read and discarded the probed user's whole entity_type slice. + expect(regressed.rowsFiltered).toBe(OWN_OPS / ENTITY_TYPES.length); + expect(regressed.nodes).toContain( + 'operations_user_id_entity_type_entity_id_server_seq_idx', + ); + }); +}); + +describe('detectConflictForEntity behaviour is unchanged by the query split (PGlite)', () => { + let db: PGlite; + + const TIME_DELTA_ACTION = '[TimeTracking] Sync time spent'; + const OTHER_USER_ID = 7; + /** Own tenant, so the legacy-misc row seeded for USER_ID cannot mask the gate. */ + const POST_SPLIT_USER_ID = 8; + + const seed = async (op: { + id: string; + serverSeq: number; + clientId: string; + entityId: string | null; + entityIds?: string[]; + entityType?: string; + actionType?: string; + schemaVersion?: number; + userId?: number; + vectorClock?: Record; + }): Promise => { + await db.query( + `INSERT INTO operations (${INSERT_COLS}) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)`, + [ + op.id, + op.userId ?? USER_ID, + op.clientId, + op.serverSeq, + op.actionType ?? '[Task] Update', + op.entityType ?? 'TASK', + op.entityId, + op.entityIds ?? [], + op.schemaVersion ?? 1, + JSON.stringify(op.vectorClock ?? { [op.clientId]: 1 }), + ], + ); + }; + + const detect = ( + entityId: string, + opOverrides: Partial> = {}, + userId: number = USER_ID, + ): Promise<{ hasConflict: boolean }> => + detectConflictForEntity( + userId, + incomingOp(entityId, opOverrides), + entityId, + makeMeasuringTx(db, newStats()) as never, + ); + + beforeAll(async () => { + db = new PGlite(); + await db.waitReady; + await db.exec(CREATE_TABLE); + await db.exec(CREATE_INDEXES); + }); + + afterAll(async () => { + await db.close(); + }); + + it('reports no conflict for an entity nothing has touched', async () => { + expect((await detect('never-seen-entity')).hasConflict).toBe(false); + }); + + it('finds a stored multi-entity op via its NON-FIRST entity (#8334)', async () => { + await seed({ + id: 'op-multi', + serverSeq: 1, + clientId: 'other', + entityId: 'conflict-first', + entityIds: ['conflict-first', 'conflict-second'], + }); + + // Concurrent clocks ({other:1} vs {uploader:1}). conflict-second is reachable + // only through entity_ids, i.e. only via the array branch. + expect((await detect('conflict-second')).hasConflict).toBe(true); + expect((await detect('conflict-first')).hasConflict).toBe(true); + }); + + it('finds an op via its DIVERGENT scalar (not a member of its own entity_ids)', async () => { + await seed({ + id: 'op-divergent', + serverSeq: 2, + clientId: 'other', + entityId: 'divergent-scalar', + entityIds: ['divergent-member'], + }); + + expect((await detect('divergent-scalar')).hasConflict).toBe(true); + expect((await detect('divergent-member')).hasConflict).toBe(true); + }); + + it('does not re-fetch when both branches TIE on the same row', async () => { + // A multi-entity op whose scalar entity_id is also a member of its entity_ids — + // a real stored shape (getStoredEntityIds keeps the full set once length > 1). + // Both branches then return the SAME server_seq, and because + // @@unique([userId, serverSeq]) makes an equal server_seq the same row, the + // findUnique would be pure waste. Pins "fetch only when it BEATS the scalar": + // relaxing `>` to `>=` still returns the right answer, so only the round-trip + // count can catch it — and this lookup runs twice per uploaded op. + await seed({ + id: 'op-tie', + serverSeq: 20, + clientId: 'other', + entityId: 'tie-entity', + entityIds: ['tie-entity', 'tie-sibling'], + vectorClock: { other: 1 }, + }); + + const stats = newStats(); + const result = await detectConflictForEntity( + USER_ID, + incomingOp('tie-entity'), + 'tie-entity', + makeMeasuringTx(db, stats) as never, + ); + + expect(result.hasConflict).toBe(true); + // Scalar findFirst + array CTE. A third query means the tie triggered a fetch. + expect(stats.sql).toHaveLength(2); + }); + + it('picks the ARRAY row when it has the higher server_seq', async () => { + // Scalar row first, then a NEWER multi-entity row covering the same entity. + // Pins the merge and the winning-row fetch: against the scalar row alone an + // incoming {uploader:1} is EQUAL from the SAME client (a retry, no conflict), + // so only picking the newer array row can produce a conflict here. + await seed({ + id: 'op-older-scalar', + serverSeq: 3, + clientId: 'uploader', + entityId: 'merge-entity', + vectorClock: { uploader: 1 }, + }); + await seed({ + id: 'op-newer-array', + serverSeq: 4, + clientId: 'other', + entityId: 'merge-other', + entityIds: ['merge-other', 'merge-entity'], + vectorClock: { other: 1 }, + }); + + expect((await detect('merge-entity')).hasConflict).toBe(true); + }); + + it('keeps the SCALAR row when it has the higher server_seq', async () => { + await seed({ + id: 'op-older-array', + serverSeq: 5, + clientId: 'other', + entityId: 'reverse-other', + entityIds: ['reverse-other', 'reverse-entity'], + vectorClock: { other: 1 }, + }); + await seed({ + id: 'op-newer-scalar', + serverSeq: 6, + clientId: 'uploader', + entityId: 'reverse-entity', + vectorClock: { uploader: 1 }, + }); + + // Newer scalar row is an EQUAL clock from the SAME client (a retry) → accepted. + // Picking the older array row instead would wrongly report a conflict. + expect((await detect('reverse-entity')).hasConflict).toBe(false); + }); + + it('takes the NEWEST of several SCALAR matches, not the oldest', async () => { + // Every other case here gives an entity at most one scalar row, which makes the + // scalar branch's `orderBy: { serverSeq: 'desc' }` unobservable: asc and desc return + // the same row. Flipping it to 'asc' passed the whole server suite. That is a silent + // data-loss bug, not a style issue — conflict detection would compare the incoming + // clock against a STALE one, so an op that is a clean successor of the OLD state but + // CONCURRENT with the current one is accepted and overwrites a remote edit. + // + // Two scalar rows for one entity, chosen so the verdict differs by row: + // vs seq 31 {cC:5} -> CONCURRENT (incoming has cA/cB, stored has cC) -> conflict + // vs seq 30 {cB:1} -> GREATER_THAN (incoming is a clean successor) -> accepted + await seed({ + id: 'op-scalar-older', + serverSeq: 30, + clientId: 'cB', + entityId: 'scalar-order-entity', + vectorClock: { cB: 1 }, + }); + await seed({ + id: 'op-scalar-newer', + serverSeq: 31, + clientId: 'cC', + entityId: 'scalar-order-entity', + vectorClock: { cC: 5 }, + }); + + expect( + (await detect('scalar-order-entity', { vectorClock: { cA: 1, cB: 1 } })) + .hasConflict, + ).toBe(true); + }); + + it('takes the NEWEST of several array-branch matches, not the oldest', async () => { + // Two stored ops mention the same entity via entity_ids. The aggregate must be + // MAX: against the newer row the incoming clock is CONCURRENT (conflict), against + // the older one it is GREATER_THAN (clean successor). MIN therefore accepts an op + // that overwrites a concurrent remote edit — silently, with no error anywhere. + await seed({ + id: 'op-max-older', + serverSeq: 12, + clientId: 'cB', + entityId: 'max-primary', + entityIds: ['max-primary', 'max-target'], + vectorClock: { cB: 1 }, + }); + await seed({ + id: 'op-max-newer', + serverSeq: 13, + clientId: 'cC', + entityId: 'max-primary', + entityIds: ['max-primary', 'max-target'], + vectorClock: { cC: 7 }, + }); + + expect( + (await detect('max-target', { vectorClock: { cA: 4, cB: 1 } })).hasConflict, + ).toBe(true); + }); + + it('carries actionType from the ARRAY branch so concurrent time deltas still merge', async () => { + // Timer deltas are additive and commute, so two CONCURRENT deltas must NOT be + // reported as a conflict — resolveConflictForExistingOp only reaches that rule if + // the stored row's actionType survives the array-branch select. Dropping + // actionType there silently rejects tracked time, and only a delta routed through + // the ARRAY branch (reachable via entity_ids, not the scalar) exercises it. + await seed({ + id: 'op-delta-remote', + serverSeq: 14, + clientId: 'cB', + actionType: TIME_DELTA_ACTION, + entityId: 'delta-primary', + entityIds: ['delta-primary', 'delta-target'], + vectorClock: { cB: 5 }, + }); + + const result = await detect('delta-target', { + actionType: TIME_DELTA_ACTION, + vectorClock: { cA: 3 }, + }); + + expect(result.hasConflict).toBe(false); + }); + + it('scopes the array-branch row fetch to the REQUESTING user', async () => { + // Every other case here runs as user 1, so a findUnique that ignored its userId + // argument would be invisible. Under a different user the winning row is only + // reachable when the point lookup is scoped correctly; otherwise it returns null, + // the conflict disappears, and a concurrent remote edit is overwritten. + await seed({ + userId: OTHER_USER_ID, + id: 'op-other-user', + serverSeq: 42, + clientId: 'other', + entityId: 'scoped-primary', + entityIds: ['scoped-primary', 'scoped-target'], + vectorClock: { other: 1 }, + }); + + expect((await detect('scoped-target', {}, OTHER_USER_ID)).hasConflict).toBe(true); + }); + + // The CTE matches entity_ids across ALL users and types; only the OUTER user_id / + // entity_type predicates restore isolation. Both were uncovered — replacing them with + // typed tautologies left all 915 tests green. The failure is silent rather than empty + // because server_seq is per-user: a leaked MAX still resolves to a REAL row of the + // requesting user through the (user_id, server_seq) point lookup, so an unrelated op + // becomes the conflict basis. Each case below seeds exactly that collision. + it('does not take the array-branch MAX from ANOTHER user', async () => { + await seed({ + userId: OTHER_USER_ID, + id: 'op-cross-tenant', + serverSeq: 9001, + clientId: 'tenant-a', + entityId: 'tenant-a-primary', + entityIds: ['tenant-a-primary', 'cross-tenant-entity'], + vectorClock: { 'tenant-a': 1 }, + }); + // Same server_seq under the REQUESTING user, unrelated entity, concurrent clock. + // Reachable only if the CTE leaks the other tenant's sequence. + await seed({ + id: 'op-decoy-same-seq', + serverSeq: 9001, + clientId: 'decoy', + entityId: 'unrelated-to-the-probe', + vectorClock: { decoy: 5 }, + }); + + // USER_ID has never touched cross-tenant-entity, so nothing can conflict. + expect((await detect('cross-tenant-entity')).hasConflict).toBe(false); + }); + + it('does not take the array-branch MAX from another ENTITY TYPE', async () => { + // Same user and same entity id, but the only op carrying it is a PROJECT op while + // the incoming op is a TASK. Dropping the entity_type predicate fetches this very + // row (it belongs to USER_ID), and its concurrent clock invents a conflict. + await seed({ + id: 'op-cross-type', + serverSeq: 9002, + clientId: 'other-type', + entityType: 'PROJECT', + entityId: 'proj-primary', + entityIds: ['proj-primary', 'cross-type-entity'], + vectorClock: { 'other-type': 9 }, + }); + + expect((await detect('cross-type-entity')).hasConflict).toBe(false); + }); + + it('ignores a full-state op (entity_id NULL, entity_ids {}) without erroring', async () => { + await seed({ id: 'op-full', serverSeq: 8, clientId: 'other', entityId: null }); + + expect((await detect('some-entity-after-full-state')).hasConflict).toBe(false); + }); + + it('still consults the legacy GLOBAL_CONFIG:misc alias for tasks', async () => { + // Pre-split (schema_version < 2) misc writes also carried what became + // GLOBAL_CONFIG:tasks; that alias lookup must survive the query split. + await seed({ + id: 'op-legacy-misc', + serverSeq: 10, + clientId: 'other', + entityId: 'misc', + entityType: 'GLOBAL_CONFIG', + schemaVersion: 1, + vectorClock: { other: 1 }, + }); + + expect((await detect('tasks', { entityType: 'GLOBAL_CONFIG' })).hasConflict).toBe( + true, + ); + }); + + it('does not alias a POST-split misc write onto tasks', async () => { + // The alias is gated on the fixed v1→v2 split boundary; a v2+ misc write is + // disjoint from tasks and must not fabricate a conflict. + // + // MUST probe exactly 'tasks'. detectConflictForEntity enters the legacy-misc branch + // only for entityId === 'tasks', so probing any other id (this once read + // 'tasks-v2-only') never reaches the gate and passes no matter what the gate does — + // verified: breaking it to `lte` left all 915 tests green. + // + // Runs as its OWN user because the preceding test leaves a schema_version 1 misc row + // for USER_ID in the shared table, which would legitimately alias and mask this. + await seed({ + userId: POST_SPLIT_USER_ID, + id: 'op-modern-misc', + serverSeq: 11, + clientId: 'other', + entityId: 'misc', + entityType: 'GLOBAL_CONFIG', + schemaVersion: 2, + vectorClock: { other: 2 }, + }); + + expect( + (await detect('tasks', { entityType: 'GLOBAL_CONFIG' }, POST_SPLIT_USER_ID)) + .hasConflict, + ).toBe(false); + }); +}); diff --git a/packages/super-sync-server/tests/duplicate-operation-precheck.spec.ts b/packages/super-sync-server/tests/duplicate-operation-precheck.spec.ts index 4b8cfd51ab..3cc4078643 100644 --- a/packages/super-sync-server/tests/duplicate-operation-precheck.spec.ts +++ b/packages/super-sync-server/tests/duplicate-operation-precheck.spec.ts @@ -468,6 +468,10 @@ describe('Duplicate Operation Pre-check', () => { user: { update: vi.fn(), }, + // entity_ids branch of the conflict lookup (raw SQL): no multi-entity op + // stored, so no max — keeps the array branch from consuming a findUnique + // mock slot. + $queryRaw: vi.fn().mockResolvedValue([{ maxSeq: null }]), }; vi.mocked(prisma.$transaction).mockImplementationOnce(async (callback: any) => @@ -545,6 +549,10 @@ describe('Duplicate Operation Pre-check', () => { user: { update: vi.fn(), }, + // entity_ids branch of the conflict lookup (raw SQL): no multi-entity op + // stored, so no max — keeps the array branch from consuming a findUnique + // mock slot. + $queryRaw: vi.fn().mockResolvedValue([{ maxSeq: null }]), }; vi.mocked(prisma.$transaction).mockImplementationOnce(async (callback: any) => @@ -590,6 +598,10 @@ describe('Duplicate Operation Pre-check', () => { user: { update: vi.fn(), }, + // entity_ids branch of the conflict lookup (raw SQL): no multi-entity op + // stored, so no max — keeps the array branch from consuming a findUnique + // mock slot. + $queryRaw: vi.fn().mockResolvedValue([{ maxSeq: null }]), }; vi.mocked(prisma.$transaction).mockImplementationOnce(async (callback: any) => @@ -656,6 +668,10 @@ describe('Duplicate Operation Pre-check', () => { user: { update: vi.fn(), }, + // entity_ids branch of the conflict lookup (raw SQL): no multi-entity op + // stored, so no max — keeps the array branch from consuming a findUnique + // mock slot. + $queryRaw: vi.fn().mockResolvedValue([{ maxSeq: null }]), }; vi.mocked(prisma.$transaction).mockImplementationOnce(async (callback: any) => diff --git a/packages/super-sync-server/tests/entity-ids-conflict.pglite.spec.ts b/packages/super-sync-server/tests/entity-ids-conflict.pglite.spec.ts index 50330e1e59..559c0a0484 100644 --- a/packages/super-sync-server/tests/entity-ids-conflict.pglite.spec.ts +++ b/packages/super-sync-server/tests/entity-ids-conflict.pglite.spec.ts @@ -105,28 +105,6 @@ describe('#8334 multi-entity conflict SQL (PGlite)', () => { return res.rows; }; - // detectConflictForEntity() — single entity, Prisma `OR: [{entityId}, {entityIds:{has}}]` - // which Prisma renders as `entity_id = $1 OR entity_ids @> ARRAY[$1]`. - const detectForEntity = async ( - entityType: string, - id: string, - ): Promise => { - const res = await db.query( - `SELECT o.entity_id AS "entityId", - o.client_id AS "clientId", - o.server_seq AS "serverSeq", - o.vector_clock AS "vectorClock" - FROM operations o - WHERE o.user_id = 1 - AND o.entity_type = $1 - AND (o.entity_id = $2 OR o.entity_ids @> ARRAY[$2]::text[]) - ORDER BY o.server_seq DESC - LIMIT 1`, - [entityType, id], - ); - return res.rows[0] ?? null; - }; - // prefetchLatestEntityOpsForBatch() — multi-entity-TYPE batch via a JOIN over // (entity_type, entity_id) pairs. const prefetchForPairs = async ( @@ -295,35 +273,6 @@ describe('#8334 multi-entity conflict SQL (PGlite)', () => { }); }); - describe('detectConflictForEntity (Prisma OR / entity_ids @> ARRAY[id])', () => { - it('finds a multi-entity op via its non-first entity', async () => { - await insertOp({ - id: 'opA', - serverSeq: 1, - clientId: 'A', - entityId: 'task-1', - entityIds: ['task-1', 'task-2'], - }); - - const row = await detectForEntity('TASK', 'task-2'); - - expect(row).not.toBeNull(); - expect(row?.clientId).toBe('A'); - }); - - it('returns null for an unrelated entity', async () => { - await insertOp({ - id: 'opA', - serverSeq: 1, - clientId: 'A', - entityId: 'task-1', - entityIds: ['task-1', 'task-2'], - }); - - expect(await detectForEntity('TASK', 'task-9')).toBeNull(); - }); - }); - describe('prefetchLatestEntityOpsForBatch (JOIN-over-pairs unnest SQL)', () => { it('matches each (type,id) pair against the op entity set without crossing types', async () => { await insertOp({ diff --git a/packages/super-sync-server/tests/gap-detection.spec.ts b/packages/super-sync-server/tests/gap-detection.spec.ts index 69a70fd431..8906597fb2 100644 --- a/packages/super-sync-server/tests/gap-detection.spec.ts +++ b/packages/super-sync-server/tests/gap-detection.spec.ts @@ -8,6 +8,8 @@ vi.mock('../src/db', async () => { const { applyOperationSelect, hasOperationUniqueConflict, + isEntityArrayBranchQuery, + entityArrayBranchRows, testState: state, } = await import('./sync.service.test-state'); const { Prisma: PrismaModule } = await import('@prisma/client'); @@ -186,6 +188,14 @@ vi.mock('../src/db', async () => { return state.users.get(args.where.id) || null; }), }, + // Array branch of the single-entity conflict lookup: MAX(server_seq) over + // `entity_ids @> ARRAY[id]`, issued as raw SQL since the full-history-scan fix. + $queryRaw: vi.fn().mockImplementation(async (strings: any, ...params: unknown[]) => { + if (!isEntityArrayBranchQuery(strings)) { + throw new Error(`Unexpected raw query: ${String(strings)}`); + } + return entityArrayBranchRows(state.operations, params); + }), // Upload transaction writes the storage counter atomically via $executeRaw. $executeRaw: vi.fn().mockResolvedValue(0), }); 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 index 81587a5c88..a2935d00a6 100644 --- a/packages/super-sync-server/tests/issue-8334-detect-conflict.spec.ts +++ b/packages/super-sync-server/tests/issue-8334-detect-conflict.spec.ts @@ -1,5 +1,6 @@ import { describe, expect, it } from 'vitest'; import { detectConflict, getStoredEntityIds } from '../src/sync/conflict'; +import { isEntityArrayBranchQuery } from './sync.service.test-state'; import type { Operation } from '../src/sync/sync.types'; /** @@ -7,8 +8,8 @@ import type { Operation } from '../src/sync/sync.types'; * (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. + * matches an entity with a scalar `entityId` lookup plus a separate raw-SQL + * max-serverSeq lookup over `entity_ids`, which the mock below reproduces. * * The batch lookup paths (raw unnest SQL) are validated separately against real * Postgres semantics and in conflict-detection.spec.ts. @@ -21,28 +22,77 @@ type StoredRow = { clientId: string; serverSeq: number; vectorClock: Record; + /** + * REQUIRED, not optional, on purpose. detectConflictForEntity reads + * existingOp.actionType to let two CONCURRENT time-tracking deltas merge instead of + * conflicting. A row shape that can omit it makes actionType silently `undefined` on + * the array branch, which is exactly how that merge gets lost without a red test. + * The merge itself is covered against real Postgres in + * conflict-entity-lookup-plan.pglite.spec.ts; this only keeps the mock honest. + */ + actionType: string; }; -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 - ); +// detectConflictForEntity issues the scalar and array halves as two separately +// indexed queries (a single OR + ORDER BY ... LIMIT 1 degenerated into a full +// history scan and took production down — see the PERF note in conflict.ts). +// This mock mirrors that three-call shape and throws on the old OR filter so a +// silent revert cannot pass here. +const makeTx = (rows: StoredRow[]): any => { + const scoped = (where: any): StoredRow[] => + rows.filter((r) => r.userId === where.userId && r.entityType === where.entityType); + return { + operation: { + // Scalar branch: where { userId, entityType, entityId }, orderBy serverSeq desc. + findFirst: async ({ where }: any) => { + if (where.OR) { + throw new Error('detectConflictForEntity must not use a combined OR filter'); + } + return ( + scoped(where) + .filter((r) => r.entityId === where.entityId) + .sort((a, b) => b.serverSeq - a.serverSeq)[0] ?? null + ); + }, + // Winning array-branch row, fetched by the (user_id, server_seq) unique key. + // Honours `select` so that dropping a column from the production query — most + // consequentially actionType, see StoredRow — actually changes what this returns. + findUnique: async ({ where, select }: any) => { + const { userId, serverSeq } = where.userId_serverSeq; + const row = rows.find((r) => r.userId === userId && r.serverSeq === serverSeq); + if (!row) return null; + if (!select) return row; + return Object.fromEntries( + Object.entries(select) + .filter(([, isSelected]) => isSelected) + .map(([key]) => { + if (!(key in row)) { + throw new Error(`Mock row has no column "${key}"`); + } + return [key, row[key as keyof StoredRow]]; + }), + ); + }, }, - }, -}); + // Array branch: MAX(server_seq) over `entity_ids @> ARRAY[id]`, issued as raw + // SQL (a MATERIALIZED CTE) so the planner is forced onto the GIN index. + $queryRaw: async (strings: TemplateStringsArray, ...params: unknown[]) => { + if (!isEntityArrayBranchQuery(strings)) { + throw new Error(`Unexpected raw query: ${strings.join('?')}`); + } + const [entityId, userId, entityType] = params as [string, number, string]; + const seqs = rows + .filter( + (r) => + r.userId === userId && + r.entityType === entityType && + r.entityIds.includes(entityId), + ) + .map((r) => r.serverSeq); + return [{ maxSeq: seqs.length ? Math.max(...seqs) : null }]; + }, + }; +}; const staleOp = (entityId: string): Operation => ({ @@ -60,6 +110,7 @@ const staleOp = (entityId: string): Operation => const multiEntityRow: StoredRow = { userId: 1, entityType: 'TASK', + actionType: 'UPDATE_TASK', entityId: 'task-1', entityIds: ['task-1', 'task-2'], clientId: 'A', @@ -82,6 +133,7 @@ describe('#8334 detectConflict single-entity path', () => { const oldRow: StoredRow = { userId: 1, entityType: 'TASK', + actionType: 'UPDATE_TASK', entityId: 'task-3', entityIds: [], // pre-migration: empty array, only scalar persisted clientId: 'A', diff --git a/packages/super-sync-server/tests/setup.ts b/packages/super-sync-server/tests/setup.ts index d3548f983e..50ceedb941 100644 --- a/packages/super-sync-server/tests/setup.ts +++ b/packages/super-sync-server/tests/setup.ts @@ -5,6 +5,10 @@ * test infrastructure after the migration to Prisma. */ import { vi, beforeEach } from 'vitest'; +import { + isEntityArrayBranchQuery, + entityArrayBranchRows, +} from './sync.service.test-state'; // In-memory storage for test data interface TestData { @@ -119,6 +123,16 @@ vi.mock('../src/db', () => { return false; } if (where.entityId !== undefined && op.entityId !== where.entityId) return false; + // entity_ids @> ARRAY[id]. No production caller uses this via the typed API any + // more — detectConflictForEntity's array branch is raw SQL (see the $queryRaw + // mock below) — but keep the matcher generic so this shim stays a faithful + // stand-in for Prisma's filter semantics. + if ( + where.entityIds?.has !== undefined && + !(Array.isArray(op.entityIds) && op.entityIds.includes(where.entityIds.has)) + ) { + return false; + } if (where.clientId !== undefined) { if (typeof where.clientId === 'object' && where.clientId !== null) { if (where.clientId.not !== undefined && op.clientId === where.clientId.not) { @@ -221,6 +235,16 @@ vi.mock('../src/db', () => { return { count }; }), findUnique: vi.fn().mockImplementation(async (args: any) => { + // (user_id, server_seq) compound unique — used by the conflict lookup's + // array branch to fetch the winning row once its max serverSeq is known. + const compound = args.where?.userId_serverSeq; + if (compound) { + const match = Array.from(testData.operations.values()).find( + (op: any) => + op.userId === compound.userId && op.serverSeq === compound.serverSeq, + ); + return applySelect(match, args.select) || null; + } // Check if operation with given ID exists return ( applySelect(testData.operations.get(args.where?.id), args.select) || null @@ -300,7 +324,22 @@ vi.mock('../src/db', () => { }), update: vi.fn().mockResolvedValue({}), }, - $queryRaw: vi.fn().mockResolvedValue([{ total: BigInt(0) }]), + // Every raw query issued inside the transaction must be recognised here and + // anything unknown must THROW. A tolerant default is how the array branch + // stayed silently stubbed out: conflict.ts reads an unrecognised row via + // `arrayBranchRows[0]?.maxSeq ?? null`, i.e. as "no match", so the branch + // disappears instead of failing. + $queryRaw: vi + .fn() + .mockImplementation(async (strings: any, ...params: unknown[]) => { + // Single-entity conflict lookup, array branch (raw SQL since the fix for + // the full-history scan). + if (isEntityArrayBranchQuery(strings)) { + return entityArrayBranchRows(testData.operations, params); + } + const sql = Array.isArray(strings) ? strings.join('') : String(strings); + throw new Error(`Unmocked raw query in tx: ${sql}`); + }), // The upload transaction writes the storage counter atomically via // $executeRaw to keep the data write and the counter delta in a single // commit. Default mock is a no-op; specs that care about counter diff --git a/packages/super-sync-server/tests/sync-fixes.spec.ts b/packages/super-sync-server/tests/sync-fixes.spec.ts index 96a1a65469..afad7438a5 100644 --- a/packages/super-sync-server/tests/sync-fixes.spec.ts +++ b/packages/super-sync-server/tests/sync-fixes.spec.ts @@ -170,7 +170,9 @@ vi.mock('../src/db', () => { upsert: vi.fn().mockResolvedValue({}), count: vi.fn().mockResolvedValue(1), }, - $queryRaw: vi.fn().mockResolvedValue([]), + // maxSeq null mirrors the conflict lookup's entity_ids branch, matching the + // findFirst above, which finds no prior op for an entity either. + $queryRaw: vi.fn().mockResolvedValue([{ maxSeq: null }]), // Upload transaction writes the storage counter atomically via $executeRaw. $executeRaw: vi.fn().mockResolvedValue(0), }; diff --git a/packages/super-sync-server/tests/sync-operations.spec.ts b/packages/super-sync-server/tests/sync-operations.spec.ts index b8b8b40070..4c7c82038b 100644 --- a/packages/super-sync-server/tests/sync-operations.spec.ts +++ b/packages/super-sync-server/tests/sync-operations.spec.ts @@ -10,6 +10,8 @@ vi.mock('../src/db', async () => { const { applyOperationSelect, hasOperationUniqueConflict, + isEntityArrayBranchQuery, + entityArrayBranchRows, testState: state, } = await import('./sync.service.test-state'); const { Prisma: PrismaModule } = await import('@prisma/client'); @@ -290,11 +292,19 @@ vi.mock('../src/db', async () => { }, // Upload transaction writes the storage counter atomically via $executeRaw. $executeRaw: vi.fn().mockResolvedValue(0), - // Full-state op uploads aggregate prior vector clocks via $queryRaw inside - // the same transaction. Dispatch based on the SQL text so other $queryRaw - // callers (storage counter, etc.) keep working. + // Raw queries issued inside the upload transaction. Every shape must be + // recognised explicitly and anything else must THROW: this mock used to fall + // through to a `total` row, which conflict.ts reads via + // `arrayBranchRows[0]?.maxSeq ?? null` as "no array-branch match". That silently + // disabled the array branch for every conflict assertion in this file. $queryRaw: vi.fn().mockImplementation(async (strings: any, ...params: any[]) => { const sql = Array.isArray(strings) ? strings.join('') : String(strings); + // Array branch of the single-entity conflict lookup: MAX(server_seq) over + // `entity_ids @> ARRAY[id]`, scoped to ONE entity. + if (isEntityArrayBranchQuery(strings)) { + return entityArrayBranchRows(state.operations, params); + } + // Full-state op uploads aggregate prior vector clocks in the same transaction. if (sql.includes('jsonb_each_text(vector_clock)')) { const [userId, beforeServerSeq] = params; const aggregate = new Map(); @@ -316,7 +326,7 @@ vi.mock('../src/db', async () => { max_counter: BigInt(max_counter), })); } - return [{ total: BigInt(0) }]; + throw new Error(`Unmocked raw query in upload tx: ${sql}`); }), }); diff --git a/packages/super-sync-server/tests/sync.service.spec.ts b/packages/super-sync-server/tests/sync.service.spec.ts index 7d8a60e268..0ba51c0a9b 100644 --- a/packages/super-sync-server/tests/sync.service.spec.ts +++ b/packages/super-sync-server/tests/sync.service.spec.ts @@ -10,6 +10,8 @@ vi.mock('../src/db', async () => { const { applyOperationSelect, hasOperationUniqueConflict, + isEntityArrayBranchQuery, + entityArrayBranchRows, testState: state, } = await import('./sync.service.test-state'); const { Prisma: PrismaModule } = await import('@prisma/client'); @@ -141,23 +143,10 @@ vi.mock('../src/db', async () => { .sort((a: any, b: any) => b.serverSeq - a.serverSeq); return applyOperationSelect(ops[0], args.select) || null; } - if (args.where?.entityType && Array.isArray(args.where?.OR)) { - const targetEntityId = - args.where.OR.find((condition: any) => condition.entityId !== undefined) - ?.entityId ?? - args.where.OR.find((condition: any) => condition.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 === targetEntityId || - op.entityIds?.includes(targetEntityId)), - ) - .sort((a: any, b: any) => b.serverSeq - a.serverSeq); - return applyOperationSelect(ops[0], args.select) || null; - } + // Scalar branch of the single-entity conflict lookup. The entity_ids half is + // a separate $queryRaw call; the two were one OR + ORDER BY ... LIMIT 1 + // until that degenerated into a full history scan in production (see the + // PERF note in conflict.ts detectConflictForEntity). if (args.where?.entityId && args.where?.entityType) { const ops = Array.from(state.operations.values()) .filter( @@ -281,6 +270,16 @@ vi.mock('../src/db', async () => { return count; }), findUnique: vi.fn().mockImplementation(async (args: any) => { + // (user_id, server_seq) compound unique — fetches the conflict lookup's + // array-branch winner once its max serverSeq is known. + const compound = args.where?.userId_serverSeq; + if (compound) { + const match = Array.from(state.operations.values()).find( + (op: any) => + op.userId === compound.userId && op.serverSeq === compound.serverSeq, + ); + return applyOperationSelect(match, args.select) || null; + } if (args.where?.id) { return ( applyOperationSelect(state.operations.get(args.where.id), args.select) || null @@ -419,6 +418,12 @@ vi.mock('../src/db', async () => { // returning their existing default shape. $queryRaw: vi.fn().mockImplementation(async (strings: any, ...params: any[]) => { const sql = Array.isArray(strings) ? strings.join('') : String(strings); + // Array branch of the single-entity conflict lookup: MAX(server_seq) over + // `entity_ids @> ARRAY[id]`, scoped to ONE entity — not a user-wide max + // (see conflict.ts detectConflictForEntity). + if (isEntityArrayBranchQuery(strings)) { + return entityArrayBranchRows(state.operations, params); + } if (sql.includes('FROM user_sync_state') && sql.includes('FOR UPDATE')) { const [txUserId] = params as [number]; return [ @@ -497,7 +502,11 @@ vi.mock('../src/db', async () => { max_counter: BigInt(max_counter), })); } - return [{ total: BigInt(0) }]; + // Unrecognised raw queries must THROW, never return a plausible-looking row. + // conflict.ts reads an unknown shape via `arrayBranchRows[0]?.maxSeq ?? null` + // as "no array-branch match", so a tolerant default silently deletes the + // branch under test instead of failing. + throw new Error(`Unmocked raw query in tx: ${sql}`); }), }); diff --git a/packages/super-sync-server/tests/sync.service.test-state.ts b/packages/super-sync-server/tests/sync.service.test-state.ts index 9b582ea64c..4fc909ba8e 100644 --- a/packages/super-sync-server/tests/sync.service.test-state.ts +++ b/packages/super-sync-server/tests/sync.service.test-state.ts @@ -1,5 +1,5 @@ /** - * Shared test state for sync.service.spec.ts + * Shared test state and Prisma-mock helpers for the sync service specs. * * Separated into its own file to avoid circular import issues with vitest mock hoisting. */ @@ -12,6 +12,7 @@ export const testState = { serverSeqCounter: 0, batchConflictQueryCount: 0, entityConflictFindFirstCount: 0, + entityConflictArrayQueryCount: 0, fullStateAuthorLookupCount: 0, }; @@ -23,9 +24,44 @@ export function resetTestState(): void { testState.serverSeqCounter = 0; testState.batchConflictQueryCount = 0; testState.entityConflictFindFirstCount = 0; + testState.entityConflictArrayQueryCount = 0; testState.fullStateAuthorLookupCount = 0; } +/** + * detectConflictForEntity's array branch is raw SQL — a MATERIALIZED CTE over the + * entity_ids GIN index — so every tx mock must answer it via $queryRaw rather than + * the typed model API. `AS "maxSeq"` is the discriminator: the only other raw query + * against `operations` (prefetchLatestEntityOpsForBatch) is a DISTINCT ON with no + * such alias, so the two never collide. + */ +export function isEntityArrayBranchQuery(strings: unknown): boolean { + const sql = Array.isArray(strings) ? strings.join('') : String(strings); + return sql.includes('AS "maxSeq"') && sql.includes('entity_ids @>'); +} + +/** + * Answers that query from in-memory ops. Parameter order follows the tagged + * template in conflict.ts: entityId (inside the CTE), then userId, then entityType. + * Returns the single-row shape the caller destructures. + */ +export function entityArrayBranchRows( + operations: Map, + params: unknown[], +): Array<{ maxSeq: number | null }> { + const [entityId, userId, entityType] = params as [string, number, string]; + const seqs = Array.from(operations.values()) + .filter( + (op: any) => + op.userId === userId && + op.entityType === entityType && + Array.isArray(op.entityIds) && + op.entityIds.includes(entityId), + ) + .map((op: any) => op.serverSeq); + return [{ maxSeq: seqs.length ? Math.max(...seqs) : null }]; +} + export function applyOperationSelect(op: any, select?: Record): any { if (!op || !select) { return op; 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 d0d559b45a..63a5993279 100644 --- a/packages/super-sync-server/tests/time-tracking-operations.spec.ts +++ b/packages/super-sync-server/tests/time-tracking-operations.spec.ts @@ -26,6 +26,8 @@ vi.mock('../src/db', async () => { const { applyOperationSelect, hasOperationUniqueConflict, + isEntityArrayBranchQuery, + entityArrayBranchRows, testState: state, } = await import('./sync.service.test-state'); const { Prisma: PrismaModule } = await import('@prisma/client'); @@ -81,24 +83,10 @@ 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; - } + // Scalar branch of the single-entity conflict lookup (#8334). The entity_ids + // half is a separate $queryRaw call; the two were one OR + ORDER BY ... + // LIMIT 1 until that degenerated into a full history scan in production + // (see the PERF note in conflict.ts detectConflictForEntity). if (args.where?.entityId && args.where?.entityType) { const ops = Array.from(state.operations.values()) .filter( @@ -194,6 +182,15 @@ vi.mock('../src/db', async () => { }), deleteMany: vi.fn().mockImplementation(async () => ({ count: 0 })), findUnique: vi.fn().mockImplementation(async (args: any) => { + // (user_id, server_seq) compound unique — fetches the array-branch winner. + const compound = args.where?.userId_serverSeq; + if (compound) { + const match = Array.from(state.operations.values()).find( + (op: any) => + op.userId === compound.userId && op.serverSeq === compound.serverSeq, + ); + return applyOperationSelect(match, args.select) || null; + } if (args.where?.id) { return ( applyOperationSelect(state.operations.get(args.where.id), args.select) || null @@ -289,6 +286,15 @@ vi.mock('../src/db', async () => { return state.users.get(args.where.id) || null; }), }, + // Array branch of the single-entity conflict lookup: MAX(server_seq) over + // `entity_ids @> ARRAY[id]`, scoped to ONE entity — not a user-wide max. + // Raw SQL since the full-history-scan fix. + $queryRaw: vi.fn().mockImplementation(async (strings: any, ...params: unknown[]) => { + if (!isEntityArrayBranchQuery(strings)) { + throw new Error(`Unexpected raw query: ${String(strings)}`); + } + return entityArrayBranchRows(state.operations, params); + }), // Upload transaction writes the storage counter atomically via $executeRaw. $executeRaw: vi.fn().mockResolvedValue(0), });