`requestSnapshot` skips changes that are in flight when it pauses the stream

Abierto
  • ¿Algo no está bien?
#4,847 0 comentarios 0 reacciones 0 asignados Ver en GitHub

Nadie ha tomado este issue todavía.

Evaluación

Dificultad
4/5
Tiempo estimado
3-5 días
Aptitud para principiantes
52/100
Tipo de issue
Error
Claridad
Bien especificado
Estado de actividad
Activo
Stack tecnológico
postgresql, typescript
Área
backend, databases

Línea de trabajo

Start in the client code that implements ShapeStream.requestSnapshot, especially the pause handling and handleResponseMetadata path that applies responseOffset; the payload does not name source or test files. Trace how the aborted live request's offset and the snapshot tracker (xmin/xmax/xip_list) affect resumption. Reproduce with the supplied script and verify both inserts arrive after a subset snapshot.

Escrito por el modelo de indexación a partir del texto del issue.

Descripción

triage

Versions

  • @electric-sql/client 1.5.28
  • electricsql/electric 1.6.0
  • Postgres 16

Summary

With log: "changes_only", calling stream.requestSnapshot({ where }) while a stream response that carries a change is
still in flight makes the stream lose that change for good, when the changed row does not match the snapshot's where.

requestSnapshot acquires the pause lock, which aborts the in-flight request (abort(PAUSE_STREAM)). When the snapshot
comes back, its electric-offset header is applied to the stream state (handleResponseMetadata with
responseOffset), so the stream resumes from the log head at snapshot time. The changes between the stream's previous
offset and that head are never fetched again. The snapshot only contains rows matching its where, so a change to any
other row is dropped.

In practice this hits TanStack DB syncMode: "on-demand" collections: every query that needs a new subset calls
requestSnapshot. A row inserted by another client while a page is loading its subsets never reaches the collection,
and live queries over it never update.

Reproduction

Show repro.mjs 
import { ShapeStream, isChangeMessage, isControlMessage } from "@electric-sql/client";
import pg from "pg";

const ELECTRIC_URL = process.env.ELECTRIC_URL ?? "http://localhost:30019/v1/shape";
const DATABASE_URL = process.env.DATABASE_URL ?? "******localhost:55439/electric";
const LIVE_LATENCY_MS = 1000;

const sleep = (ms, signal) =>
  new Promise((resolve, reject) => {
    const timer = setTimeout(resolve, ms);
    signal?.addEventListener("abort", () => {
      clearTimeout(timer);
      reject(signal.reason ?? new DOMException("Aborted", "AbortError"));
    });
  });

// Simulates network latency on live long-polls, so a committed change is still in flight
// when the app asks for a subset snapshot.
const fetchWithLiveLatency = async (input, init) => {
  const url = new URL(input instanceof Request ? input.url : String(input));
  const response = await fetch(input, init);
  if (process.env.DEBUG) {
    console.log("    ", init?.method ?? "GET", url.search, "->", response.status, "electric-offset", response.headers.get("electric-offset"));
  }
  if (url.searchParams.get("live") === "true") {
    const body = await response.arrayBuffer();
    await sleep(LIVE_LATENCY_MS, init?.signal);
    return new Response(body, { status: response.status, statusText: response.statusText, headers: response.headers });
  }
  return response;
};

async function run(table, { requestSubsetSnapshot }) {
  const db = new pg.Client({ connectionString: DATABASE_URL });
  await db.connect();
  await db.query(`DROP TABLE IF EXISTS ${table}`);
  await db.query(`CREATE TABLE ${table} (id int PRIMARY KEY, name text NOT NULL)`);
  await db.query(`INSERT INTO ${table} VALUES (1, 'one'), (2, 'two')`);

  const stream = new ShapeStream({
    url: ELECTRIC_URL,
    params: { table },
    log: "changes_only",
    offset: "now",
    subsetMethod: process.env.SUBSET_METHOD ?? "GET",
    fetchClient: fetchWithLiveLatency,
    onError: (error) => console.log("  stream error:", error?.message ?? error),
  });

  const received = [];
  let resolveUpToDate;
  const upToDate = new Promise((resolve) => (resolveUpToDate = resolve));
  const unsubscribe = stream.subscribe((messages) => {
    for (const message of messages) {
      if (isChangeMessage(message)) received.push(`${message.headers.operation} ${message.value.id}`);
      if (isControlMessage(message) && message.headers.control === "up-to-date") resolveUpToDate();
    }
  });

  await upToDate;
  // Let the stream settle into a live long-poll
  await sleep(500);

  await db.query(`INSERT INTO ${table} VALUES (3, 'three')`);
  // Give Electric time to add the insert to the shape log, while its live response is still on the wire
  await sleep(150);
  if (requestSubsetSnapshot) {
    // A snapshot of an unrelated subset, as an on-demand collection does when a query needs row 1
    const { data } = await stream.requestSnapshot({ where: "id = 1" });
    console.log(`  subset snapshot (id = 1) returned rows: [${data.map((m) => m.value.id).join(", ")}]`);
  }
  await db.query(`INSERT INTO ${table} VALUES (4, 'four')`);

  await sleep(3000);
  unsubscribe();
  stream.unsubscribeAll?.();
  await db.end();
  return received;
}

const expected = ["insert 3", "insert 4"];
const check = (label, received) => {
  const changes = received.filter((change) => change === "insert 3" || change === "insert 4");
  const ok = expected.every((change) => changes.includes(change));
  console.log(`${ok ? "OK  " : "BUG "} ${label}: received [${changes.join(", ")}], expected [${expected.join(", ")}]`);
  return ok;
};

console.log("Control: no snapshot request between the two inserts");
const control = check("control", await run(`items_control_${Date.now()}`, { requestSubsetSnapshot: false }));
console.log("Repro: subset snapshot requested right after the first insert");
const repro = check("repro", await run(`items_repro_${Date.now()}`, { requestSubsetSnapshot: true }));

process.exit(control && repro ? 0 : 1);

docker compose up -d --wait
npm install
node repro.mjs          # SUBSET_METHOD=POST node repro.mjs gives the same result

The script:

  1. Creates a table with rows 1 and 2, and opens a ShapeStream with log: "changes_only" and offset: "now".
  2. Waits for up-to-date and lets the stream sit in a live long-poll. A fetchClient wrapper adds 1s of latency to
    live responses, as a slow network would, and honours the abort signal.
  3. Inserts row 3, waits 150ms for Electric to add it to the shape log, then calls
    stream.requestSnapshot({ where: "id = 1" }) while the live response with row 3 is still on the wire.
  4. Inserts row 4 and waits 3s.

A control run does the same without step 3's snapshot.

Expected

Both runs receive insert 3 and insert 4.

Actual

OK   control: received [insert 3, insert 4], expected [insert 3, insert 4]
BUG  repro: received [insert 4], expected [insert 3, insert 4]

5 of 5 runs fail the same way. With DEBUG=1 the requests show the jump:

GET  ?live=true&offset=0_inf            -> 200 electric-offset 27046064_0   (carries insert 3, aborted by the pause)
POST ?offset=0_inf  (snapshot id = 1)   -> 200 electric-offset 27046064_0   (snapshot data: row 1 only)
GET  ?offset=27046064_0                 -> resumes after insert 3, which is never delivered

Possible fix

A subset snapshot should not move the stream's offset forward past changes the stream has not processed. For
example, only adopt the snapshot's offset when the stream has no offset yet (the offset: "now" handshake). Otherwise,
resume from the offset of the request the pause aborted, and rely on the snapshot tracker (xmin/xmax/xip_list) to
drop the duplicates the snapshot already contains.

Lenguaje dominante
TypeScript
Estrellas
10.4k
Forks
399
Sin PR fusionados en 30 d

Preparar el entorno

Primeros pasos

  1. Lee el issue completo y luego la guía de contribución del proyecto.
  2. Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
  3. Haz un fork del repositorio y trabaja en una rama.
  4. Abre un pull request que haga referencia al número del issue.

Más de electric-sql/electric

Todos los issues de electric-sql/electric

Issues similares

Más issues de TypeScript

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.