waitForAcknowledged() never resolves for frames replayed from the store-and-forward journal #70

Summary

After a process restart, a sender that reopens a store-and-forward journal replays the unacknowledged frames, and QuestDB acknowledges them. But sender.waitForAcknowledged(sender.publishedSequence, timeoutMs) never resolves for those replayed frames. It always rejects with QwpIngressAckTimeoutError, and the error itself reports the frame as acknowledged:

QwpIngressAckTimeoutError: timed out waiting for QWP ACK watermark [targetSequence=0, acknowledgedSequence=0, timeoutMs=10000]

Any application that waits for the ACK watermark after reopening a journal hits this, for example to commit source offsets after a restart. It waits for the full timeout and then wrongly concludes that the batch was not acknowledged.

Environment

• @questdb/nodejs-client 5.0.0, built from main at a658130
• QuestDB 10.0.1 (questdb/questdb:10.0.1 Docker image)
• Node.js v24.3.0, macOS 26.5.1 (arm64)
• Reproduced 5 out of 5 times

Steps to reproduce

These are the store-and-forward and read-after-write examples from the Node.js client docs (questdb/documentation#569, before they were changed to avoid this bug). Each script runs as a separate process, so the second one recovers the frame from the journal.

step1-write.mts: write one row while QuestDB is down, so the row is only journaled:

import { connectQwpNodeClient } from "@questdb/nodejs-client";

// Persist this directory outside the container in production.
const db = await connectQwpNodeClient(
  "ws::addr=localhost:9000;sf_dir=./questdb-sf;sender_id=trades;" +
    "sf_max_segment_bytes=1m;lazy_connect=on;",
);
try {
  const sender = await db.borrowSender();
  try {
    await sender
      .table("trades_sf")
      .stringColumn("trade_id", "trade-12345")
      .symbol("symbol", "ETH-USD")
      .doubleColumn("price", 2615.54)
      .at(1723000000000, "ms");
    await sender.flush(); // persisted locally; not necessarily acknowledged
  } finally {
    await sender.close();
  }
} finally {
  await db.close();
}

step2-wait.mts: once QuestDB is back, reopen the same journal, wait for the replayed frame's ACK, then poll for the row:

import { connectQwpNodeClient } from "@questdb/nodejs-client";

const db = await connectQwpNodeClient(
  "ws::addr=localhost:9000;sf_dir=./questdb-sf;sender_id=trades;" +
    "sf_max_segment_bytes=1m;failover=off;",
);
try {
  const sender = await db.borrowSender();
  try {
    await sender.waitForAcknowledged(sender.publishedSequence, 10_000);
  } finally {
    await sender.close();
  }

  const lease = await db.borrowQuery();
  try {
    const deadline = Date.now() + 10_000;
    let visible = false;
    while (!visible && Date.now() < deadline) {
      const query = await lease.query(
        "SELECT trade_id FROM trades_sf WHERE trade_id = $1 LIMIT 1",
        {
          binds: (binds) => binds.setVarchar(0, "trade-12345"),
          timeoutMs: Math.max(1, deadline - Date.now()),
        },
      );
      for await (const batch of query) visible ||= batch.rowCount > 0;
      await query.completion;
      if (!visible) await new Promise((resolve) => setTimeout(resolve, 100));
    }
    if (!visible) throw new Error("trade not visible in time");
  } finally {
    await lease.close();
  }
} finally {
  await db.close();
}

Run them with nothing listening on port 9000 at first:

# 1. QuestDB is down: the row goes to the journal (exits 0 after the 5 s close timeout)
npx tsx step1-write.mts

# 2. Start QuestDB and create the table
docker run -d --name qdb -p 9000:9000 questdb/questdb:10.0.1
curl -G http://localhost:9000/exec --data-urlencode "query=CREATE TABLE trades_sf (timestamp TIMESTAMP, trade_id VARCHAR, symbol SYMBOL, price DOUBLE) TIMESTAMP(timestamp) PARTITION BY DAY DEDUP UPSERT KEYS(timestamp, trade_id);"

# 3. Reopen the journal and wait for the ACK
npx tsx step2-wait.mts

Expected

The pool's sender reopens slot trades-0 and replays frame 0. waitForAcknowledged(0n, 10_000) resolves as soon as QuestDB acknowledges the frame, the poll finds trade-12345, and the script exits 0.

Actual

The replay succeeds and QuestDB acknowledges the frame: the row is in the table, and acknowledgedSequence reaches 0. But waitForAcknowledged() is never woken, so the script fails after the full 10 s:

$ npx tsx step2-wait.mts
QWP ingress reported an asynchronous failure [message=timed out waiting for QWP ACK watermark [targetSequence=0, acknowledgedSequence=0, timeoutMs=10000]]
file:///.../node_modules/@questdb/nodejs-client/dist/es/index.mjs:11195
                const error = new QwpIngressAckTimeoutError(targetSequence, this.acknowledgedFrameSequence, timeoutMs);
                              ^

QwpIngressAckTimeoutError: timed out waiting for QWP ACK watermark [targetSequence=0, acknowledgedSequence=0, timeoutMs=10000]
    at Timeout._onTimeout (file:///.../node_modules/@questdb/nodejs-client/dist/es/index.mjs:11195:31)
    at listOnTimeout (node:internal/timers:608:17)
    at process.processTimers (node:internal/timers:543:7) {
  targetSequence: 0n,
  acknowledgedSequence: 0n,
  timeoutMs: 10000
}
# exit code 1, after 10 s

$ curl -sG http://localhost:9000/exec --data-urlencode "query=SELECT * FROM trades_sf"
... "dataset":[["2024-08-07T03:06:40.000000Z","trade-12345","ETH-USD",2615.54]],"count":1}

Root cause

Links are to a658130.

1. QwpIngressSession.acknowledgedFrameSequence reads the transport's watermark (ingress-session.ts#L814-L816).
2. waitForAcknowledged() resolves at once only if that watermark already covers the target. Otherwise it registers a waiter (ingress-session.ts#L1462-L1491).
3. Waiters are re-checked only by resolveAcknowledgedSequenceWaiters() (L1931). That runs once when the waiter registers, and afterwards only when the session receives a response (L1700) or a durable ACK (L1881).
4. Frames recovered from the journal are tracked with ackDelivered: true (reconnecting-ingress-connection.ts#L691-L697).
5. When their OK arrives, acknowledgeThrough() advances the transport watermark (L2142-L2153). But hasUndelivered() is false, so the OK is never forwarded to the session (L1856-L1872).

So the watermark moves past the target, but nothing re-evaluates the pending waiter. It can only time out. The comment at L1809-L1813 describes the same kind of failure for awaitServerAck()/awaitDurableAck() when an OK is treated as already delivered.

Workaround

Poll the watermark instead of calling waitForAcknowledged(). Replacing the wait in step2-wait.mts with this makes the same scenario exit 0 in under a second:

// Workaround: poll the watermark instead of waitForAcknowledged().
const target = sender.publishedSequence;
const ackDeadline = Date.now() + 10_000;
while (sender.acknowledgedSequence < target) {
  if (Date.now() > ackDeadline) throw new Error("journal not acknowledged in time");
  await new Promise((resolve) => setTimeout(resolve, 50));
}

Suggested fix

Notify the session whenever the transport's acknowledged watermark advances, including for frames whose OK is not forwarded, so it calls resolveAcknowledgedSequenceWaiters(). A regression test could reopen a journal containing unacknowledged frames, call waitForAcknowledged(sender.publishedSequence) before the replay's ACK arrives, and assert that it resolves once the ACK is received.