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

Đang mở
  • Có gì không ổn?
#4,847 0 bình luận 0 reaction 0 người được giao Xem trên GitHub

Chưa có ai nhận issue này.

Đánh giá

Độ khó
4/5
Thời gian dự kiến
3-5 ngày
Mức phù hợp với người mới
52/100
Loại issue
Lỗi
Độ rõ ràng
Đặc tả rõ ràng
Mức độ hoạt động
Sôi nổi
Công nghệ
postgresql, typescript
Lĩnh vực
backend, databases

Hướng nghiên cứu

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.

Do mô hình lập chỉ mục viết ra từ nội dung của issue.

Mô tả

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.

Ngôn ngữ chính
TypeScript
Star
10.4k
Fork
399
Không có pull request nào được merge trong 30 ngày

Chuẩn bị môi trường

Bắt đầu từ đâu

  1. Đọc hết issue, rồi đọc hướng dẫn đóng góp của dự án.
  2. Bình luận trên issue rằng bạn sẽ nhận — tránh hai người làm cùng một việc.
  3. Fork repository và làm thay đổi trên một nhánh.
  4. Mở pull request có tham chiếu số hiệu của issue.

Issue khác của electric-sql/electric

Tất cả issue của electric-sql/electric

Issue tương tự

Thêm issue về TypeScript

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.