replica-routing.integration.test.ts

replica-routing.integration.test.tstests/integration/replica-routing.integration.test.tsraw ← integration8,487 B← Back to the app
1/**
2 * TASK-0135 (Phase 15, pooling + replica) — LIVE replica-routing integration
3 * suite.
4 *
5 * Proves, against the real NestJS API (Testcontainers Postgres) with a
6 * DISTINCT read-replica wired via `DATABASE_REPLICA_URL`, the spec's
7 * read/write split and staleness contract:
8 *
9 *   • The block TREE read (`db.read()` — a spec-named replica read) is served
10 *     from the READER (replica). With a zero-copy replica the tree read is
11 *     STALE: it returns the replica's (empty) blocks, not the primary's.
12 *
13 *   • The write-then-read-your-write invariant (spec: "a write-then-read-
14 *     your-write on the same request always hits primary") — the realtime
15 *     block FLUSH (`POST /internal/blocks/flush`) reads `pages.version` for
16 *     its `baseVersion` CAS and writes in one request. Against the empty
17 *     replica that read would 404 (no page) — it succeeds and bumps the
18 *     PRIMARY version, proving the version check read the PRIMARY, not the
19 *     replica. The optimistic-concurrency version check is never a replica
20 *     read.
21 *
22 *   • The replica stays DIVERGED from the primary (replica.version stays 1
23 *     while the primary bumps to 2) — the documented lag; the reader is a
24 *     distinct client, not the primary.
25 *
26 * The replica is a SEPARATE, schema-migrated but EMPTY database (zero-copy =
27 * zero lag — every read the primary has is missing on the replica, so a
28 * mis-route to the replica is always observable). PERF_CACHE is forced OFF so
29 * the tree read is always a live reader read (no cache-hit masking).
30 */
31import { afterAll, beforeAll, describe, expect, it } from "vitest";
32import { PrismaClient } from "@prisma/client";
33import { createTestApp, deploySchema, type TestAppHandle } from "./helpers/api";
34import { getDatabaseUrl } from "./helpers/containers";
35import { createUser, createWorkspace, authHeaders } from "./helpers/seed";
36 
37const REPLICA_DB = "factory_replica";
38 
39let app: TestAppHandle;
40let replicaUrl = "";
41let replicaPrisma: PrismaClient | null = null;
42 
43/** Build the replica URL = the primary URL with the database swapped. */
44function buildReplicaUrl(primaryUrl: string): string {
45  const u = new URL(primaryUrl);
46  u.pathname = `/${REPLICA_DB}`;
47  return u.toString();
48}
49 
50/** A standalone client to the primary (for schema + direct assertions). */
51async function primaryClient(primaryUrl: string): Promise<PrismaClient> {
52  const c = new PrismaClient({ datasources: { db: { url: primaryUrl } } });
53  await c.$connect();
54  return c;
55}
56 
57beforeAll(async () => {
58  const primaryUrl = await getDatabaseUrl();
59  const p = await primaryClient(primaryUrl);
60 
61  // 1) Create the replica DB (idempotent) + apply the schema (no data = lag).
62  try {
63    await p.$executeRawUnsafe(`CREATE DATABASE ${REPLICA_DB}`);
64  } catch (e) {
65    if (!String(e).includes("already exists")) throw e;
66  }
67  replicaUrl = buildReplicaUrl(primaryUrl);
68  await deploySchema(replicaUrl);
69  await p.$disconnect();
70 
71  // 2) Wire the replica BEFORE the app boots (PrismaService reads it on
72  //    construction). PERF_CACHE off → the tree read is always a live reader
73  //    read (no cache-hit masking the replica route).
74  process.env.DATABASE_REPLICA_URL = replicaUrl;
75  process.env.PERF_CACHE = "off";
76 
77  app = await createTestApp();
78  replicaPrisma = new PrismaClient({ datasources: { db: { url: replicaUrl } } });
79  await replicaPrisma.$connect();
80}, 240_000);
81 
82afterAll(async () => {
83  await app?.close();
84  try {
85    await replicaPrisma?.$disconnect();
86  } catch {
87    /* ignore */
88  }
89  // Drop the replica for cleanliness (after the app — its reader disconnects
90  // in app.close(); an open reader would make the DROP wait for a session).
91  try {
92    const p = await primaryClient(await getDatabaseUrl());
93    await p.$executeRawUnsafe(`DROP DATABASE IF EXISTS ${REPLICA_DB}`);
94    await p.$disconnect();
95  } catch {
96    /* best-effort */
97  }
98});
99 
100describe("replica routing (live, DATABASE_REPLICA_URL set)", () => {
101  let headers: Record<string, string>;
102  let workspaceId: string;
103  let pageId: string;
104  let primaryBlocks: { id: string; type: string; position: string; content: Record<string, unknown>; properties: Record<string, unknown>; parentId: string | null }[];
105 
106  beforeAll(async () => {
107    const owner = await createUser(app.request, { name: "Replica Owner" });
108    headers = authHeaders(owner.accessToken!);
109    const ws = await createWorkspace(app.request, headers, `Replica WS ${Date.now()}`);
110    workspaceId = ws.workspaceId;
111    const page = await app.request("POST", `/api/v1/workspaces/${workspaceId}/pages`, {
112      headers,
113      body: { title: "Replica Page" }
114    });
115    expect(page.status).toBeLessThan(300);
116    pageId = ((await page.json()) as { id: string }).id;
117 
118    // Two blocks, written to the PRIMARY (the public create path — `db.write`).
119    for (const i of [1, 2]) {
120      const b = await app.request("POST", `/api/v1/pages/${pageId}/blocks`, {
121        headers,
122        body: { type: "paragraph", content: { text: `replica-${i}` }, properties: {} }
123      });
124      expect(b.status).toBeLessThan(300);
125    }
126    // Sanity: the PRIMARY has the seeded title block + the 2 created
127    // paragraphs (page create seeds a default title block). Capture ALL of
128    // them so the flush below passes the full set back (a no-op diff).
129    const rows = await app.prisma.block.findMany({ where: { pageId, deletedAt: null } });
130    expect(rows.length).toBeGreaterThanOrEqual(3);
131    primaryBlocks = rows.map((b) => ({
132      id: b.id,
133      type: b.type as string,
134      position: b.position,
135      content: b.content as Record<string, unknown>,
136      properties: b.properties as Record<string, unknown>,
137      parentId: b.parentId
138    }));
139  });
140 
141  it("the block tree read (db.read) is served from the REPLICA → stale (empty)", async () => {
142    // The tree read is the spec-named replica read. With the cache off and a
143    // zero-copy replica, the page lookup + permission check run on the
144    // primary (they succeed — the page exists there), but the BLOCK loader
145    // (`this.prisma.read()`) runs on the replica → ZERO blocks. If the read
146    // were mis-routed to the primary it would return the 2 blocks — a hard
147    // failure of this assertion.
148    const res = await app.request("GET", `/api/v1/pages/${pageId}/blocks?depth=all`, { headers });
149    expect(res.status).toBe(200);
150    const body = (await res.json()) as { data: unknown[] };
151    expect(body.data).toHaveLength(0); // stale: the replica has no blocks
152  });
153 
154  it("write-then-read-your-write: the flush's version CAS hits the PRIMARY (not the replica)", async () => {
155    // The flush reads pages.version for its baseVersion CAS and writes in the
156    // SAME request. The primary page is at version 1; a replica read would
157    // 404 (the page is absent on the empty replica) — it succeeds and bumps
158    // the PRIMARY to 2, proving the version check read the primary, never the
159    // replica (the optimistic-concurrency check is never a replica read). The
160    // two blocks are passed back verbatim (a no-op diff — no soft-delete).
161    const flush = await app.request("POST", "/internal/blocks/flush", {
162      headers: { Authorization: `Bearer ${app.internalServiceToken}` },
163      body: {
164        pageId,
165        baseVersion: 1,
166        blocks: primaryBlocks.map((b) => ({
167          id: b.id,
168          parentId: b.parentId ?? null,
169          type: b.type.toLowerCase(),
170          position: b.position,
171          content: b.content,
172          properties: b.properties
173        }))
174      }
175    });
176    expect(flush.status).toBe(200);
177    const body = (await flush.json()) as { version?: number; conflict?: unknown };
178    expect(body.conflict).toBeUndefined(); // no CAS conflict
179    expect(body.version).toBe(2); // the primary bumped 1 → 2
180  });
181 
182  it("the replica stays DIVERGED from the primary (the documented lag)", async () => {
183    // After the flush the primary page is at version 2; the empty replica
184    // never received the page (no write path routes to it) — the strongest
185    // possible divergence proof. A mis-routed write would have created one.
186    const primaryPage = await app.prisma.page.findUnique({
187      where: { id: pageId },
188      select: { version: true }
189    });
190    expect(primaryPage?.version).toBe(2);
191    const replicaPage = await replicaPrisma!.page.findUnique({
192      where: { id: pageId },
193      select: { version: true }
194    });
195    expect(replicaPage).toBeNull();
196  });
197});
198