[POSTPONED post-v0.2.0] fix(clients/typescript): wake consumer on LISTEN/NOTIFY (#147)

[NikolayS]

Summary

• RCA: The TS Consumer only used setTimeout-based polling (default 30 s). pgque.ticker() already emits pg_notify('pgque_<queue>', tick_id::text) but the TS driver never listened, causing up to 30 s of unnecessary latency.
• Design: A dedicated pg.Client (separate from the shared pool) is created by Client.newConsumer via a factory closure and issues LISTEN pgque_<queue>. A notification event handler resolves a Promise that races against sleep(pollInterval), waking the poll loop immediately on tick. Polling is fully retained as a safety-net fallback (missed notifications, reconnect windows). The LISTEN connection auto-reconnects on drop with 1 s → 2 s → 5 s backoff capped at pollInterval. On AbortSignal abort, UNLISTEN and client.end() are called before start() resolves.
• README updated with a "LISTEN/NOTIFY wakeup" section in the Consumer section.

Red/green proof

Red commit 6ef4279: unit test NOTIFY wakes the poll loop before the long sleep elapses (stub pg.Client) fails immediately — listenCalled is false because the old Consumer never opened a LISTEN connection.

Green commit c02bfe8: same test passes in <50 ms. With pollInterval: 60_000, the notification fires and the poll loop calls receive() within ~10 ms of the event — well under the 500 ms assertion window. A live-PG integration test (gated on PGQUE_TEST_DSN) additionally asserts delivery within 2 s with a 60 s poll interval.

Test plan

• bun run check — TypeScript types clean
• bun run build — compiles without errors
• bun run test — unit/in-memory mock tests pass (no DB required)
• PGQUE_TEST_DSN=... bun run test — integration tests including NOTIFY wakeup test pass in <2 s with pollInterval: 60_000

Closes #147

[NikolayS]

REV review — PR #194

Verdict

Has potential issues — no hard blockers, but two correctness concerns and two test coverage gaps warrant fixes before merge.

Summary

The LISTEN/NOTIFY design is solid: a dedicated pg.Client races notifications against sleep(pollInterval), polling is preserved as a fallback, and quoteIdentifier correctly escapes channel names. The SQL side enforces ≤57-byte queue names so the 63-byte pg_notify limit is respected on both sides. CI is all green (PG 14–18, TS, Python, Go). Two medium-confidence correctness issues exist in the reconnect/teardown flow, and two important test paths are absent.

CI status

All 9 check runs completed successfully: test (14/15/16/17/18) ✓, TypeScript client tests ✓, Python client tests ✓, Go client tests ✓, verify ✓.

Findings

Security

• [NON-BLOCKING] 7/10 — clients/typescript/src/consumer.ts (channel construction) — Channel name is built as `pgque_${this.queue}` and then passed through quoteIdentifier, which correctly doubles internal double-quotes and wraps in double-quotes. The SQL enforcement in create_queue() additionally rejects queue names >57 bytes (issue Bug: queue names longer than 57 bytes fail via pg_notify channel name in ticker/create_queue #109 is handled). No injection vector found.

  • Evidence: quoteIdentifier implementation: `"${name.replace(/"/g, '""')}"` — correct standard SQL identifier escaping.
  • Fix: none required. Optionally document that _listenClientFactory-path callers bypass the create_queue guard and should validate externally, but not a security risk in practice.

• [NON-BLOCKING] 5/10 — clients/typescript/src/client.ts:338 — DSN is stored as a field on Client and passed to Client_({ connectionString: dsn }). If a connection error is thrown and the caller logs err.message, the DSN (with credentials) could appear in logs. The existing mapPgError strips DSN from pool errors but the new pg.Client connect path (client.connect()) is outside that wrapper.

  • Evidence: await client.connect() in connectListen — errors caught by the bare catch (err) block are passed to this.logger.error(... formatErr(err) ...). formatErr returns err.message verbatim; pg sometimes includes the connection string in ECONNREFUSED messages.
  • Fix: Wrap err.message in connectListen's error path with the same DSN-redaction regex already used in mapPgError, or use a helper like sanitizeDsn(formatErr(err)).

Bugs / correctness

• [NON-BLOCKING] 7/10 — clients/typescript/src/consumer.ts:~103 — Notification during receive() in-flight is queued, but the re-arm after the race is unconditional. After await Promise.race([sleep(...), currentNotifyPromise]), the code always does currentNotifyPromise = makeNotifyPromise(). If a second notification arrives while receive() is processing the first batch (i.e., while the poll loop body is running and currentNotifyPromise has already resolved and been re-armed), the second notification resolves the new promise. When the loop returns to the top and calls receive() again it will get the batch — correct. However, if onNotification() fires while notifyResolve is null (between resolve() and the makeNotifyPromise() call), it takes the early-return branch (if (resolve) is false) and the notification is silently dropped, requiring the poll fallback to eventually wake the loop.

  • Evidence: onNotification sets notifyResolve = null then calls resolve() then immediately does currentNotifyPromise = makeNotifyPromise() — but that last line runs inside the handler, which is fine. The narrow window is: notifyResolve is null between the re-arm at the bottom of the race block and the re-assignment inside onNotification. Re-reading: makeNotifyPromise sets notifyResolve synchronously in the Promise constructor, so the window is effectively zero in a single-threaded event loop. Downgrade to POTENTIAL — no real race in Node's single-threaded model, but worth a comment.
  • Fix: Add a code comment explaining why the single-threaded event loop makes this safe, so future readers don't introduce a real race when porting.

• [POTENTIAL] 6/10 — clients/typescript/src/consumer.ts:~120 — 'end' event vs 'error' event ordering on abrupt disconnect. When the Postgres server closes the connection ungracefully (e.g. pg_terminate_backend), pg emits 'error' before 'end'. The code attaches an 'error' suppressor via client.on('error', ...) but waits for 'end' to unblock the inner promise. If 'error' fires but 'end' never fires (which can happen with certain connection resets), the await new Promise block hangs until signal is aborted.

  • Evidence: The inner wait-promise only resolves on 'end' or onAbort. If the socket is destroyed without emitting 'end', the promise never resolves and the reconnect loop stalls indefinitely (until AbortSignal fires).
  • Fix: Also resolve the inner promise on 'error': client.once('error', onEnd) alongside client.once('end', onEnd), or listen for client.once('end', onEnd) and know that pg always emits 'end' after 'error' — verify this with the pg source/docs and add a comment if guaranteed.

• [NON-BLOCKING] 5/10 — clients/typescript/src/consumer.ts:backoff — When pollInterval < 1000 ms (non-default but valid), Math.min(LISTEN_BACKOFF_MS[0], pollInterval) produces a backoff shorter than 1 s, which is fine. But the intent of "capped at pollInterval" means the first reconnect could fire in <1 s, potentially causing rapid reconnect storms on flapping connections. Minor edge case with unusually short poll intervals.

  • Evidence: LISTEN_BACKOFF_MS = [1_000, 2_000, 5_000] but Math.min(..., this.pollIntervalMs) — if pollIntervalMs = 100, backoff is 100 ms.
  • Fix: Use Math.max(Math.min(...), 500) to floor the backoff at 500 ms regardless of pollInterval.

Tests

• [NON-BLOCKING] 8/10 — clients/typescript/test/consumer.test.ts — No reconnect test. The entire reconnect loop (backoff, re-LISTEN, backoff cap) is exercised by zero unit tests. The stub fakePgClient in the NOTIFY unit test never emits 'end', so the reconnect path is completely untested.

  • Evidence: Searching the diff: no test constructs a factory that emits 'end' after first connect and verifies a second connect() call is made with backoff.
  • Fix: Add a unit test that returns a stub client whose once('end', cb) fires after 10 ms; verify connect is called twice and the second LISTEN fires.

• [NON-BLOCKING] 7/10 — clients/typescript/test/consumer.test.ts — No AbortSignal teardown test. The existing NOTIFY unit test does abort and await startPromise, and the diff shows unlistenCalled is asserted true — this is a positive signal. However, it only covers the happy path. There is no test for abort-during-reconnect-backoff (i.e., signal fires while sleep(delay, signal) is running inside connectListen) to verify the listen loop exits promptly rather than sleeping through the backoff.

  • Evidence: Teardown test only reaches cleanup via the happy-path (connected client, LISTEN succeeded). Abort-during-backoff path is untested.
  • Fix: Add a test where the factory throws on the first call, then assert that aborting the signal causes start() to resolve within 100 ms (not after the full backoff delay).

• [NON-BLOCKING] 5/10 — clients/typescript/test/consumer.test.ts:109 — Integration test uses advanceQueue to trigger the ticker. If advanceQueue calls the ticker synchronously and the LISTEN connection's 200 ms startup window is tight, there is a risk of a NOTIFY arriving before LISTEN is issued, requiring the 60 s poll fallback and causing a test timeout. The 200 ms sleep before send mitigates but does not eliminate this on slow CI runners.

  • Evidence: await sleep(200) is a fixed delay, not a readiness check. On a cold CI runner the pg.Client.connect() + LISTEN round-trip may exceed 200 ms.
  • Fix: Replace the fixed sleep with a poll on a listenReady signal (e.g., resolve a promise from inside _listenClientFactory after LISTEN completes), or increase the sleep to 500 ms as a pragmatic guard.

Guidelines (CLAUDE.md)

• [NON-BLOCKING] 9/10 — Commit subject: fix(clients/typescript): wake consumer on LISTEN/NOTIFY (#147) — 61 characters; CLAUDE.md requires subject < 50 chars.

  • Evidence: echo -n "fix(clients/typescript): wake consumer on LISTEN/NOTIFY (#147)" | wc -c → 65.
  • Fix: Shorten to e.g. fix(ts): LISTEN/NOTIFY wakeup for consumer (43 chars).

• [NON-BLOCKING] 4/10 — clients/typescript/src/types.ts:_listenClientFactory — The field is marked @internal in JSDoc, which is correct. However, it is a public property on a public interface (ConsumerOptions) with an underscore prefix as the only guard. No @ts-ignore or @deprecated discourages external use; a determined user can rely on it.

  • Evidence: _listenClientFactory?: () => Promise<import('pg').Client>; — public interface field.
  • Fix: Move to a separate internal interface (_ConsumerInternals) and intersect only inside the module, or document explicitly: "Subject to removal without semver notice."

Docs

• [NON-BLOCKING] 4/10 — clients/typescript/README.md — The new "LISTEN/NOTIFY wakeup" section accurately describes the feature. The Python driver uses LISTEN pgque_<queue> via psycopg.sql.Identifier (correctly quoted), same channel format. The TS driver now matches. No cross-driver parity matrix exists yet (Go is still poll-only); this is a pre-existing gap, not introduced by this PR.

  • Fix (optional): Add a one-line note: "Go client is poll-only; TS and Python use LISTEN/NOTIFY." Can be deferred to a docs PR.

• [NON-BLOCKING] 3/10 — README hero section likely mentions latency. If there was a claim like "up to 30 s latency," it may now be stale. Not verifiable without reading the full README, but worth a pass.

  • Fix: Search for latency claims in clients/typescript/README.md and update any "up to 30 s" → "typically single-digit ms (up to 30 s fallback)".

Recommended next step

has-potential — fix the 'error'-without-'end' hang risk (POTENTIAL/6) and add the reconnect unit test (NON-BLOCKING/8) before merging; shorten the commit subject to satisfy CLAUDE.md; remaining items are advisory.

---

REV-style review. SOC2 checks intentionally omitted per project policy.

---

Generated by Claude Code

[NikolayS]

Lifecycle status — ready for your final look

• CI: 9/9 green (PG 14–18 + verify + Go/Python/TS client jobs).
• REV: no blockers. Potential-only items:
  • Security: DSN credential leak risk in LISTEN error path (medium, non-blocking)
  • Bugs: 'error'-without-'end' on abrupt disconnect could hang the reconnect loop indefinitely (POTENTIAL/6); reconnect backoff sub-500 ms when pollInterval < 1000 ms
  • Tests: no reconnect-path unit test (NON-BLOCKING/8 — highest-priority follow-up); abort-during-backoff not covered
  • Guidelines: commit subject 65 chars vs CLAUDE.md 50-char limit
  • Docs: clean; channel format matches Python driver
• Tests: unit tests pass (bun run test 6 passed / 26 skipped without DSN). Integration LISTEN→receive chain is gated on PGQUE_TEST_DSN and exercised in CI's TypeScript client tests job.

Not merging per your "I'll look myself" instruction. Branch ready when you are.

---

Generated by Claude Code

[NikolayS]

REV review

The fix correctly races a notification Promise against the existing poll-interval setTimeout, opens a dedicated pg.Client for LISTEN pgque_<queue> (channel name matches sql/pgque.sql lines 688/793), and reconnects with capped backoff (1s/2s/5s, capped at pollInterval). AbortSignal is honored in the outer reconnect loop, in sleep(delay, signal), and in the inner "wait for end" promise via onAbort. UNLISTEN is issued before client.end() on the abort path, and start() awaits listenDone before resolving so no orphan promise warnings. Channel quoting via quoteIdentifier is safe; the 57-byte queue-name limit enforced server-side keeps the prefixed channel inside the 63-byte pg_notify limit. Tests cover both an env-gated end-to-end NOTIFY wakeup and an in-memory stub asserting LISTEN/UNLISTEN are issued and that receive() is re-invoked on notification. No SECURITY DEFINER, no SQL changes, no docs/version-tag policy violations.

Blocking (must fix before merge)

None.

Non-blocking

FINDING (bug-hunter, confidence 6, MEDIUM)

• file: clients/typescript/src/consumer.ts
• lines: ~95-105 (the inner await new Promise<void>((resolve) => { ... }) block)
• issue: signal.addEventListener('abort', onAbort, { once: true }) is registered every reconnect cycle, but if the connection emits 'end' first and the promise resolves via onEnd, onAbort is never removed from the AbortSignal. With repeated reconnects this accumulates abort listeners on the same signal (one per cycle) until abort actually fires. Bounded in practice but flagged by Node's MaxListenersExceededWarning at 11+.
• fix: capture the listener and remove it in both branches, e.g. signal?.removeEventListener('abort', onAbort) after the await resolves; or have onEnd itself remove onAbort.

FINDING (bug-hunter, confidence 5, LOW)

• file: clients/typescript/src/consumer.ts
• line: ~155 (poll loop, non-empty msgs branch)
• issue: when receive() returns messages, the loop processes them without re-arming currentNotifyPromise. If a NOTIFY arrives during message processing, onNotification resolves the existing promise and pre-arms a new one — fine — but if two NOTIFYs land back-to-back during a slow handler, the second one finds notifyResolve === null and is silently dropped. Polling is the safety net so messages still flow, but the "wake immediately" guarantee weakens under load.
• fix: track a pendingNotify boolean set true when notifyResolve is null, and have the next race short-circuit if it is set.

FINDING (test-analyzer, confidence 6, MEDIUM)

• file: clients/typescript/test/consumer.test.ts
• issue: no test exercises the reconnect path. The backoff-on-'end' and the "abort during reconnect sleep" branches are uncovered. The client.on('error', ...) swallow is also untested.
• fix: add a stub-pg.Client test that emits 'end' after LISTEN, asserts the warn log fires, asserts a second connect()/LISTEN happens, then aborts and asserts start() resolves.

FINDING (test-analyzer, confidence 5, LOW)

• file: clients/typescript/test/consumer.test.ts
• issue: the env-gated test relies on a fixed 200 ms / 2 s sleep. Flake risk under loaded CI; consider polling for listenCalled via a hook instead of a fixed delay (the in-memory test already does this — mirror it).

Skipped (SOC2)

Skipped per repo policy.

Verdict: APPROVE

---

Generated by Claude Code

[NikolayS]

Step 3 — real testing

Walked the user-facing claim: consumer with pollInterval: 60_000 still receives a fresh message in ~ms after a tick — because LISTEN/NOTIFY wakes the poll loop instead of waiting the full 60s.

Setup: PG 16, pgque from this branch's sql/pgque.sql, bun install --frozen-lockfile clean.

Script:

import { connect } from '/home/user/pgque/.claude/worktrees/agent-aa65362714010e20b/clients/typescript/src/index.js';

const dsn = 'postgresql://postgres:pgque_test@localhost/pgque_test';
const queue = 'walkthrough_194';
const consumer = 'c194';

const producerClient = await connect(dsn);
const consumerClient = await connect(dsn);

const ac = new AbortController();
const received: { type: string; receivedAt: number }[] = [];

const c = consumerClient.newConsumer(queue, consumer, {
  pollInterval: 60_000,  // 60 SECONDS — polling alone would never fire in time.
});

c.handle('latency.test', async (msg) => {
  received.push({ type: msg.type, receivedAt: Date.now() });
});

const startedAt = Date.now();
const consumerStartPromise = c.start(ac.signal);

// Give the LISTEN connection time to attach BEFORE we tick.
await new Promise(r => setTimeout(r, 500));

// Send + tick. On a vanilla 30s poll-only consumer, this would wait 30+ seconds before delivery.
const sendStart = Date.now();
await producerClient.send(queue, { type: 'latency.test', payload: { i: 1 } });
await producerClient.forceTick(queue);
await producerClient.ticker(queue);
const tickAt = Date.now();

// Wait up to 5 seconds for delivery — well under pollInterval=60s.
const deadline = Date.now() + 5_000;
while (received.length === 0 && Date.now() < deadline) {
  await new Promise(r => setTimeout(r, 50));
}

const elapsed = received.length > 0 ? received[0].receivedAt - tickAt : -1;
console.log(`tick → receive elapsed: ${elapsed}ms (received ${received.length} msg)`);
console.log(`pollInterval: 60000ms (sleeping consumer would not have fired)`);

ac.abort();
try { await consumerStartPromise; } catch {}
await producerClient.close();
await consumerClient.close();

if (received.length === 0) {
  console.error('FAIL: no message received within 5s — LISTEN/NOTIFY did NOT wake consumer');
  process.exit(1);
}
if (elapsed > 5_000) {
  console.error(`FAIL: delivery latency ${elapsed}ms exceeds the 5s ceiling (would have to be polling)`);
  process.exit(1);
}
console.log(`PASS: delivery in ${elapsed}ms — well under pollInterval=60000ms, proves LISTEN/NOTIFY wakeup`);

Output:

tick → receive elapsed: 7ms (received 1 msg)
pollInterval: 60000ms (sleeping consumer would not have fired)
PASS: delivery in 7ms — well under pollInterval=60000ms, proves LISTEN/NOTIFY wakeup

Verdict: ✓ Notify-driven wakeup works end-to-end. Latency was 7 ms with a 60_000 ms poll interval — polling alone would not have fired. The README's low-latency claim now holds for the TS driver.

---

Generated by Claude Code

[NikolayS]

Rebased onto origin/main (now includes #188 + #189). New head: 774e88b.

What conflicted: Only clients/typescript/README.md had a marker conflict — HEAD had updated the bigint sentence to mention ticker(queue) and forceTick(queue) return values (#188); the PR added the Consumer options table, LISTEN/NOTIFY wakeup section, and payload coercion section. Resolved by keeping HEAD's bigint sentence and all of the PR's new sections.

clients/typescript/src/client.ts auto-merged cleanly: git correctly combined #189's pgqueTypes/Omit<..., 'types'>/scoped-types approach with the PR's DSN storage, Client_ alias, and newConsumer factory injection. ticker/tickerAll/forceTick return-value signatures from #188 are preserved.

Local test results: 55/55 passed (bun run check && bun run build && bun run test against PG 16). LISTEN/NOTIFY wakeup test (stub factory, pollInterval: 60_000) completed in ~313 ms.

---

Generated by Claude Code

[NikolayS]

Moved out of scope of v0.2.0 and back to draft.

Per issue #201 the LISTEN/NOTIFY tick-wakeup semantics still need:

• A missed-notification fallback test (kill the LISTEN connection mid-flight, send + tick, assert delivery via the poll fallback).
• A SQL-side notification-overhead benchmark under high tick rate / many queues / many consumers.

Both are appropriate for v0.2.x rather than blocking the v0.2.0 stamp. The SQL change (tick-level pg_notify) is already on main and works without this client-side wakeup; consumers continue to fall back to polling at pollInterval.

Will resume on this branch (or a fresh one) once #201's verification work is staged.

---

Generated by Claude Code