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

未关闭
  • 哪里不对？
#4,847 0 条评论 0 个 reaction 已指派 0 人 在 GitHub 查看

还没有人认领这个 Issue。

评估

难度
4/5
预计耗时
3-5 天
新手友好度
52/100
Issue 类型
缺陷
描述清晰度
描述清楚
活跃度
活跃
技术栈
postgresql, typescript
领域
backend, databases

调研方向

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.

由索引模型根据 Issue 内容生成。

描述

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.

主要语言
TypeScript
星标
10.4k
派生
399
30 天内没有已合并 PR

环境准备

  • 没有 Dockerfile 或 Docker Compose 文件
  • 没有 Pull Request 模板
  • 阅读贡献指南

从这里开始

  1. 先读完整个 Issue，再读项目的贡献指南。
  2. 在 Issue 下留言说明你要接手 —— 这能避免两个人做同样的事。
  3. Fork 仓库，在一个分支上完成修改。
  4. 提交 Pull Request，并在描述里引用这个 Issue 编号。

electric-sql/electric 的其他 Issue

查看 electric-sql/electric 的全部 Issue

相似的 Issue

更多 TypeScript Issue

把新 issue 发到你的邮箱

精选适合新手参与的 GitHub issue 摘要。