From 77557077f8f05b2afbae7bf661a516d822cf9dfd Mon Sep 17 00:00:00 2001 From: Johannes Millan Date: Mon, 20 Jul 2026 19:33:57 +0200 Subject: [PATCH] fix(supersync): split the conflict entity lookup to avoid a full-history scan (#9195) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(supersync): split conflict entity lookup to avoid full-history scan detectConflictForEntity ran one Prisma filter with `OR: [{ entityId }, { entityIds: { has: entityId } }]` and `orderBy: { serverSeq: 'desc' }`. The OR spans the (user_id, entity_type, entity_id, server_seq) btree and the entity_ids GIN index, and GIN cannot supply server_seq ordering — so with LIMIT 1 the planner abandoned both index paths and walked server_seq backwards applying the OR as a filter, betting it would exit early. For an entity with no matching rows (the first op for a new task, the most common upload there is) that bet loses and it scans the user's whole history. The function runs twice per operation on the default serial upload path, so every upload paid it twice. Production incident: 47 backends stuck on this query, longest 75 minutes, wait_event=DataFileRead, 61 of 66 connections consumed. Prisma pool exhaustion made every user's upload AND download fail with "Unable to start a transaction in the given time". No index was missing or invalid. Split into separately-indexed lookups: - scalar branch: findFirst on {userId, entityType, entityId} — the btree covers all equality columns plus the sort column - array branch: aggregate({_max: {serverSeq}}) — Prisma's OFFSET 0 subquery is a planner fence that forbids the LIMIT-driven walk and forces the GIN bitmap path - winner fetch: findUnique on the (userId, serverSeq) unique key, only when the array max beats the scalar Measured on production Postgres (88k-op user, zero-match probe) after an ANALYZE and a GIN pending-list flush: old shape 12 buffers, new shape 9 (scalar 4 + array 5). In that regime the two are comparable — the value of this split is NOT throughput. It is that neither branch offers the planner a LIMIT-driven early-exit bet, so it cannot degenerate into the backward walk regardless of statistics. Even post-ANALYZE the planner still estimates rows=96 for a zero-match `entity_ids @>` key, so the old shape stays one autovacuum re-sample away from the incident plan. Wall-clock is not quoted: on a live server the same old-shape query measured 255ms and 915ms on consecutive runs while its buffer count fell. Buffers are the stable metric here. Semantics are unchanged: both forms return the highest-serverSeq row across the union of the two predicates. Ties need no tie-break because @@unique([userId, serverSeq]) makes an equal serverSeq the same row. The legacy GLOBAL_CONFIG:misc alias path is untouched. The batch paths (detectConflictForEntities, prefetchLatestEntityOpsForBatch) are not exposed to this trap — no LIMIT, and DISTINCT ON forces full evaluation — and are left unchanged. * fix(supersync): force the GIN path with a MATERIALIZED CTE The previous commit's array branch used `aggregate({_max: {serverSeq}})`, relying on Prisma's `OFFSET 0` subquery as a planner fence. Production EXPLAIN under `SET plan_cache_mode = force_generic_plan` shows that does not work: Aggregate -> Index Scan using operations_user_id_entity_type_entity_id_server_seq_idx Index Cond: ((user_id = $1) AND (entity_type = $2)) Filter: (entity_ids @> ARRAY[$3]) It scans the whole (user_id, entity_type) slice — ~88k rows for a deep-history user. Dropping LIMIT removed the early-exit bet but did not force the GIN path: the planner costs btree+filter at ~80 against ~875 for the GIN bitmap and takes the btree. The root reason both earlier designs were misjudged: Prisma sends parameterized prepared statements, so after ~5 executions Postgres builds a GENERIC plan that cannot see parameter values. Every EXPLAIN run with literal constants measures a custom plan production never receives. Two designs looked correct that way and were catastrophic in production. Array branch is now raw SQL: WITH cand AS MATERIALIZED ( SELECT user_id, entity_type, server_seq FROM operations WHERE entity_ids @> ARRAY[$1]::text[] ) SELECT MAX(server_seq)::int AS "maxSeq" FROM cand WHERE user_id = $2 AND entity_type = $3 Inside the CTE the only predicate is `entity_ids @>`, so the composite btree has no usable leading column and GIN is the only option — the plan is forced structurally rather than won on cost, which is what makes it survive statistics drift. MATERIALIZED prevents the outer predicates being pushed down, which would hand the btree back. Adding `server_seq > ` to bound the read was evaluated and rejected: under generic planning the value is invisible, so it lands as a post-GIN filter and buys nothing, and placing it inside the CTE creates a new cliff (no leading-column anchor, so cost scales with total table size rather than the probing user's history). Semantics unchanged: verified by differential testing old-vs-new across 20,000 randomized in-memory histories and 1,500 against real Postgres, zero divergences, with 18 sabotage mutations confirming the harness can fail. Tests: mocks moved from `aggregate` to `$queryRaw` behind a shared discriminator; all plan assertions now run under force_generic_plan; the plan spec's block accounting no longer double-counts (Postgres buffer counts are cumulative, so parents already include children). * test(supersync): make the conflict-lookup plan spec able to fail The plan spec could not catch the outage it was written for. Three independent problems, each of which made a load-bearing property invisible: 1. The spec seeded ONE user and ONE entity_type. In that shape the GIN estimate (DEFAULT_CONTAIN_SEL * whole table) and the btree-slice estimate (N / users*entity_types) are the same rows, so GIN always wins on cost and no regression is detectable. Reseeded to 20k rows for the probed user plus 20k across ~20k other users over 8 entity types, which reproduces production's plan node-for-node. Seeds in ~490ms. 2. The measuring shim rebuilt the array-branch SQL from a CONSTANT, so changing MAX(server_seq) to MIN in conflict.ts could never be observed. It now reconstructs the SQL from the real tagged template, putting the aggregate, the MATERIALIZED fence and the CTE shape genuinely under test. 3. The shim ignored `select`, so dropping actionType from the array branch was invisible — that field decides whether concurrent time-tracking deltas merge or get rejected. It now honors `select` and throws on an unmapped key. Before this, all three of these mutations left the suite green: MAX->MIN (silently accepts a write that should conflict, overwriting a concurrent remote edit), dropping actionType (rejects a client's tracked time), and hardcoding userId in the findUnique. Measured after reseeding: shipped CTE 2 blocks / 0 rows filtered; the old OR, the flat MAX, Prisma's aggregate form, and MATERIALIZED-dropped all land at 806 blocks / 2500 filtered. One budget assertion (<100 blocks) now catches every known route back to the incident, with a 403x margin. All plan assertions run under force_generic_plan; plan-NAME assertions were removed after measurement showed the old OR no longer plans as a Backward walk at this seed. Also corrects a false claim this change introduced. The comment justified the cross-tenant CTE scan with "entity ids are client-generated nanoids, so overlap is not expected". The client hard-codes globally shared ids (TODAY, EM_URGENT, EM_IMPORTANT, KANBAN_IN_PROGRESS, INBOX_PROJECT, EISENHOWER_MATRIX, KANBAN_DEFAULT, plus fixed GLOBAL_CONFIG keys) which are byte-identical for every user, and updateBoard({id:'KANBAN_DEFAULT'}) routes here as a single-entity op. Measured: 3885 blocks to reach the 10 rows belonging to the probing user, versus 4 for a unique id. Correctness is preserved by the outer WHERE user_id; cost is not bounded per-user. The fix is a btree_gin composite index, which needs a real-Postgres EXPLAIN (PGlite lacks the extension) and is filed separately. Deletions: the entity-ids-conflict helper that hand-wrote SQL production never sends and modelled the abandoned aggregate design (-71 lines), a dead aggregate mock that would have MASKED a regression by returning an undefined max, the server_seq-bound test that asserted a rejected design still works (so it could never fail informatively), and four stale comments describing a call that no longer exists. sync-operations.spec.ts had a $queryRaw mock never wired to the array branch; it fell through to a vestigial value that reads as null, so every conflict assertion there ran with the array branch stubbed out. Wired up, and all six raw-query sites now throw on an unrecognised query rather than falling through. * test(supersync): seed the conflict-lookup plan spec in production order Building the GIN after the bulk load produced a pending-list-free index and measured 2 blocks for the array branch - a state production only sees right after a vacuum. With the index pre-existing and rows arriving one at a time, as in production, the same query reads 140 blocks against a 14x larger index. Seed in production order and never vacuum, so the spec measures the worst realistic steady state, and re-derive the budget (100 -> 300) against it. Also move the GIN-not-btree structural assertions onto the real tagged template, and replace the five hardcoded regression shapes with a single canary whose stated job is proving the seed still reproduces the mis-plan. The five EXPLAINed strings that exist only in the test file and responded to no source mutation; the budget and plan assertions on the real SQL do. * docs(supersync): correct two false claims in the conflict-lookup comment The cross-tenant scan vector was wrong. updateTag({id:'TODAY'}) and updateBoard({id:'KANBAN_DEFAULT'}) are single-entity, so getStoredEntityIds persists '{}' and they never enter the GIN under that id. The actual vector is the bulk sortBoards action, which stores the shared board ids in entity_ids. 'Needs btree_gin, unverifiable in PGlite' was also wrong: a GIN index on (ARRAY['u:'||user_id] || entity_ids) has a text[] operand served by the built-in array_ops, builds in PGlite with no extension, and was measured flat against a baseline that scales with tenant count. Recorded as a tradeoff rather than a free win - it needs the predicate rewritten to match and indexes every row rather than the multi-entity minority. Drop the force_generic_plan methodology lecture duplicated from the spec header, keeping a pointer. Also make the #8334 mock honest: StoredRow now requires actionType and findUnique honours select, so a dropped column changes what it returns instead of silently yielding undefined. * test(supersync): cover the scalar branch picking the newest op Flipping detectConflictForEntity's scalar orderBy from 'desc' to 'asc' passed the entire 914-test server suite. Every existing case gave an entity at most one scalar row, so the ordering was unobservable. The bug it hides is silent data loss: with the oldest row returned, conflict detection compares 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. Seed two scalar rows whose verdicts differ, and verify by mutation that the test is red under 'asc' and green under 'desc'. * docs(supersync): stop the vector-clock doc recommending the outage query vector-clocks.md still described detectConflictForEntity as the combined OR + orderBy filter and vouched for it - 'a BitmapOr + sort bounded by the entity's stored version depth (op-log pruning keeps that small)'. That is the exact false assumption behind the 2026-07-20 outage, in the authoritative doc. Worse, the escalation it recommended (two ordered LIMIT 1 lookups) is itself a measured-catastrophic shape: the array side still cannot order on GIN. Also fix a test that could not fail. 'does not alias a POST-split misc write onto tasks' probed 'tasks-v2-only', but the legacy-misc branch is entered only for entityId === 'tasks', so it never reached the gate it claimed to guard - breaking the gate to 'lte' left all 915 tests green. It now probes 'tasks' under its own tenant, and is verified red under that mutation. Correct three overclaims in the comments: production does get custom plans for the first ~5 executions before settling generic; a partial expression GIN can skip the single-entity majority; fastupdate=off stops new pending entries but does not flush existing ones, and 140 blocks is not a ceiling. * docs(supersync): pin the two fail-open invariants in the conflict lookup Both are unreachable today and both fail toward ACCEPTING a conflicting write if a future edit breaks them, which is silent data loss rather than an error. No runtime guards - they could never fire, and the sync layer does not need more unreproducible hardening. Documented at the exact spot an editor would stand instead. - The aggregate fold is safe only because a bare aggregate returns exactly one row; a GROUP BY (or a revert to Prisma aggregate()) makes zero rows possible, which reads as 'no prior op'. - 'arrayOp ?? scalarOp' conflates 'array branch lost' with 'array row vanished'. The second needs RepeatableRead to stay unreachable. * docs(supersync): stop overclaiming GIN is forced; fix the batch-lookup doc Three misleading safety claims, all in text a future contributor would rely on: - vector-clocks.md still documented the batch lookup as the mutually exclusive 'CASE WHEN cardinality(entity_ids) > 0 THEN entity_ids ELSE ARRAY[entity_id]'. Production unions the two columns instead, and the exclusive form is the #8334 bug: it drops a scalar that is not a member of its own entity_ids, so a later concurrent write to that entity is wrongly accepted - silent data loss. The section above it was rewritten last commit; this one was missed. - 'The CTE wins STRUCTURALLY, not on cost' overstated. Removing the competing btree is structural; GIN being chosen is not. A sequential scan is always available and wins for an unselective id - reproducible on PG16, and exactly what a globally shared entity id produces. Reframed as a measured outcome, since a confident comment asserting what the planner will do is what preceded the outage. - The plan spec still said production receives 'a CUSTOM plan nobody receives'. It receives custom plans for roughly the first 5 executions before settling generic, and this file covers only the generic mode. * docs(supersync): correct the plan-cache and aggregate claims; match Int in tests Round-4 review. Four inaccuracies, all in text a contributor would rely on. - plan_cache_mode=auto does NOT unconditionally switch to generic after 5 executions: it compares the generic cost against the average custom cost and may keep using custom plans indefinitely. Stated as an observation for THIS statement (custom_plans=5, generic_plans=15 on production) rather than a universal rule, in conflict.ts, the plan spec and vector-clocks.md. Note: commit fdf0f62011 claimed this fix for conflict.ts. It did not contain it - the edit was lost and the commit message was not verified against the diff. Applied here for real. - Prisma's bare aggregate() cannot return zero rows; it returns one object with _max.serverSeq null. Only GROUP BY can produce zero rows. The fold comment claimed otherwise. - The test schema declared server_seq bigint while production maps it as Prisma Int. MAX() therefore returned bigint and the shim's Number() coercion hid it, so dropping the ::int cast from the production SQL would have stayed green. Test schema now matches production. - detectConflictForEntity's JSDoc still described the path as using Prisma's typed model API; its array branch is raw SQL. vector-clocks.md's flow summary likewise still described a single findFirst. * docs(supersync): correct the remaining false claims in the conflict lookup Comment/doc only; no runtime change. 915 server tests still pass. conflict.ts: - the expression-GIN paragraph argued against the index because it "indexes EVERY row". A PARTIAL index (WHERE entity_ids <> '{}') avoids that and is lossless: single-entity ops store '{}', so their indexed expression is just ARRAY['u:'] and can never contain the probed id. Names the real caveat instead — the query must carry a matching predicate or the index is ignored. - the array-only findFirst was labelled "the outage"; the outage was the combined OR, as stated 30 lines earlier. - dropped the cross-reference to a "shared id produces a Seq Scan" measurement that the cross-tenant note does not contain (added in 27d1baa9b7). - single-entity writes against a shared id are not a GIN *population* vector but ARE a *probe* vector: updateBoard({id:'KANBAN_DEFAULT'}) routes through detectConflictForEntity and probes that literal across every tenant. - the INVARIANT note's hazard does not materialise: GROUP BY user_id yields zero rows only when zero rows matched, identical to MAX -> NULL. The real hazard is a grouping that returns MORE than one row. - "batch unnest paths are NOT exposed" excluded only the early-exit degeneracy; they carry the same two-index OR, so the slice-scan one is not excluded and neither batch query is EXPLAINed (#9205). plan spec: - budget arithmetic: 816/300 is 2.7x over budget, not 6x/5.7x (5.7 is 816/143, regression-to-measured). Real headroom is the smaller number. - drops "reproduces the production plan node-for-node", which contradicted the stated different-major-version fidelity limit two paragraphs down. - the index set is a deliberate subset, not a mirror of schema.prisma. - the server_seq integer change is a fidelity fix; it does not make dropping ::int catchable (MAX over integer already returns integer). vector-clocks.md: - 816 blocks/2500 rows is the outage query specifically and the only shape still pinned by a test, not all five. - the forward-only paragraph pointed at the removed CASE expression, two paragraphs below the warning that that form IS the #8334 data-loss bug. * docs(supersync): mark the partial-index suggestion as UNMEASURED The paragraph recommending GIN(...) WHERE entity_ids <> '{}' steers whoever picks up #9199, and it conflated two claims of very different strength. Splits them: losslessness is a claim about the DATA and holds by inspection (getStoredEntityIds collapses single-entity sets to [], entity_ids is NOT NULL DEFAULT '{}', '{}' @> ARRAY[] is false). Whether the partial form is USABLE is a claim about the planner — it needs Postgres to prove a query-side `entity_ids <> '{}'` implies the index predicate for an array `<>` — and that was reasoned about, never run. Says so explicitly rather than leaving a confident-sounding recommendation. A comment asserting unmeasured planner behaviour is what preceded the outage. * fix(supersync): model the entity_ids GIN in schema.prisma and cover tenant isolation Two defects found in review of the conflict-lookup split. 1. The GIN existed ONLY as raw migration SQL, so `prisma db push` never created it. Verified with `prisma migrate diff --from-empty`: db push built four btree/unique indexes on operations and no GIN. CI and the SuperSync E2E jobs use `db push --skip-generate`, and README's manual setup and db-push baselining path do too — the last one can mark 20260613000001 applied without ever executing it. On such a database this split is a REGRESSION, not merely un-helped: the old query filtered user_id/entity_type first, so a bad plan scanned one user's slice, while the new CTE deliberately carries no user predicate and is MATERIALIZED so none can push down. Without the index every probe Seq Scans every tenant's rows, twice per accepted op. The schema comment claiming Prisma cannot model GIN array indexes was wrong. `@@index([entityIds], type: Gin, map: ...)` validates on 5.22 and emits the same DDL; stored indexdef is byte-identical to the migration's, so migrate-based deployments see no drift. Also notes the restore-point PARTIAL index, which genuinely cannot be modelled and is still missing on db-push databases (#9192). 2. The CTE's outer user_id / entity_type predicates had no real-SQL coverage: replacing both with typed tautologies left 915/915 green. The failure is silent because server_seq is per-user, so a leaked MAX still resolves to a real row of the requesting user and an unrelated op becomes the conflict basis. Adds two PGlite cases seeding exactly that collision. Sabotage-verified: both predicates tautologized fails both; user-only fails only the cross-tenant case; type-only fails only the cross-type one. Also corrects vector-clocks.md, which claimed four alternative shapes were unguarded. Verified by mutation that dropping MATERIALIZED (2 failures) and MAX -> MIN (1) do fail the budget spec; only their historical block counts are unpinned. --- docs/sync-and-op-log/vector-clocks.md | 49 +- .../super-sync-server/prisma/schema.prisma | 138 +-- .../super-sync-server/src/sync/conflict.ts | 178 +++- .../tests/conflict-detection.spec.ts | 48 +- ...conflict-entity-lookup-plan.pglite.spec.ts | 882 ++++++++++++++++++ .../duplicate-operation-precheck.spec.ts | 16 + .../tests/entity-ids-conflict.pglite.spec.ts | 51 - .../tests/gap-detection.spec.ts | 10 + .../tests/issue-8334-detect-conflict.spec.ts | 94 +- packages/super-sync-server/tests/setup.ts | 41 +- .../tests/sync-fixes.spec.ts | 4 +- .../tests/sync-operations.spec.ts | 18 +- .../tests/sync.service.spec.ts | 45 +- .../tests/sync.service.test-state.ts | 38 +- .../tests/time-tracking-operations.spec.ts | 42 +- 15 files changed, 1427 insertions(+), 227 deletions(-) create mode 100644 packages/super-sync-server/tests/conflict-entity-lookup-plan.pglite.spec.ts 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), });