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

オープン
  • 何か問題がありますか？
#4,847 コメント 0 件 リアクション 0 件 担当者 0 名 GitHub で見る

まだ誰も着手していません。

評価

難易度
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 はありません

環境構築

はじめの一歩

  1. issue を最後まで読み、次にプロジェクトのコントリビューションガイドを読みます。
  2. 着手することを issue にコメントします — 二人が同じ作業をするのを防げます。
  3. リポジトリをフォークし、ブランチを切って変更します。
  4. issue 番号を参照したプルリクエストを送ります。

electric-sql/electric のほかの issue

electric-sql/electric の issue をすべて見る

似ている issue

TypeScript の issue をもっと見る

新しい issue をメールで受け取る

初心者向けの GitHub issue を短くまとめたダイジェスト。