mirror of
https://github.com/QwenLM/qwen-code.git
synced 2026-07-10 01:29:17 +00:00
fix(serve): correct queue tagging + plumb maxQueued through SDK
Address both P2 findings from the Codex review pass on PR #4237. **Bug 1: `BoundedAsyncQueue.forcedInBuf` position-invariant break** The previous `forcedInBuf` counter only tracked LIVE-vs-FORCED correctly when all forced entries lived at the FRONT of the buffer (subscribe-time `Last-Event-ID` replay). The new mid-stream `slow_client_warning` path force-pushes to the BACK of the queue while the queue is still open, which the existing accounting was not designed for: - publish 6 events at maxQueued=8 → 75% threshold trips → force-push warning at the back → buf=[1..6, warning], forcedInBuf=1. - consumer shifts `1` → forcedInBuf decremented to 0 (incorrect: `1` was a live frame, not the forced one). - consumer drains 2..6 + warning → buf=[], forcedInBuf=0, true live count = 0, but `size` getter and `push()` cap check then use `buf.length - forcedInBuf` which drifts over subsequent refills, causing premature warn / eviction before the cap is actually reached. Replace the position-dependent counter with a per-entry `{value, forced}` tag. `liveCount` is incremented in `push()` / decremented in `next()` only when the shifted entry was non-forced — position becomes irrelevant. `size` getter returns `liveCount` directly. The class doc comment is rewritten to call out that the new tag is the position-independent replacement for the old "forced frames must stay at the front" invariant. Regression test in `eventBus.test.ts` reproduces the codex trace (warn at 75%, drain past warning, refill to cap) and asserts no premature eviction. **Bug 2: SDK does not expose `?maxQueued`** `docs/users/qwen-serve.md` and `docs/developers/qwen-serve-protocol.md` both document `?maxQueued=N` as something SDK clients can request, but `SubscribeOptions` on `DaemonClient` only declared `lastEventId` + `signal`, and `subscribeEvents()` always fetched `/events` without a query string. Typed-SDK consumers had no way to opt in without hand-crafting URLs. - Add `SubscribeOptions.maxQueued?: number` with JSDoc noting the daemon range `[16, 2048]` and the pre-flight requirement on `caps.features.slow_client_warning`. - `DaemonClient.subscribeEvents` builds the URL with an optional `?maxQueued=<n>` segment. No client-side range validation — the daemon's `parseMaxQueuedQuery` is the source of truth and returns structured `400 invalid_max_queued`; duplicating the bounds in two layers would diverge on the next tweak. - `DaemonSessionSubscribeOptions extends SubscribeOptions` so the new field flows through `DaemonSessionClient` automatically. Three new SDK tests: - subscribeEvents appends `?maxQueued=N` when set - omits the query string when absent (existing behavior preserved) - propagates a `400 invalid_max_queued` unchanged Tests: 214 focused tests across eventBus / bridge / SDK DaemonClient / DaemonSessionClient / daemonEvents, plus 111 in the server suite. All green; the new eventBus regression case proves the position-invariant fix. 🤖 Generated with [Qwen Code](https://github.com/QwenLM/qwen-code)
This commit is contained in:
doudouOUC
parent
a6ccde5b18
commit
bae42c88bc
4 changed files with 166 additions and 43 deletions
|
|
@ -204,6 +204,68 @@ describe('EventBus', () => {
|
|||
abort.abort();
|
||||
});
|
||||
|
||||
it('warn-at-back forced frame does NOT skew the live cap for subsequent publishes (codex P2)', async () => {
|
||||
// Regression for the `forcedInBuf` position-invariant bug Codex
|
||||
// flagged: a mid-stream slow_client_warning force-pushed to the
|
||||
// BACK of the queue, then drained past, would previously cause
|
||||
// `next()` to decrement the forced counter on a LIVE shift,
|
||||
// making subsequent `push()` cap checks under-count live items
|
||||
// and warn/evict the client before they actually had `maxQueued`
|
||||
// live items in queue.
|
||||
const bus = new EventBus();
|
||||
const abort = new AbortController();
|
||||
const iter = bus.subscribe({ maxQueued: 8, signal: abort.signal });
|
||||
const it = iter[Symbol.asyncIterator]();
|
||||
|
||||
// Episode 1: fill to 6 → warn at 75%. buf = [1..6, warning].
|
||||
for (let i = 1; i <= 6; i++) bus.publish({ type: 'foo', data: i });
|
||||
|
||||
// Drain ALL 7 items (events 1..6 + warning frame). Live cap should
|
||||
// now be 0 — the warning was a forced frame and must NOT have
|
||||
// counted as a live drain.
|
||||
const drained: BridgeEvent[] = [];
|
||||
for (let i = 0; i < 7; i++) drained.push((await it.next()).value);
|
||||
expect(
|
||||
drained.filter((e) => e.type === 'slow_client_warning'),
|
||||
).toHaveLength(1);
|
||||
|
||||
// Refill to EXACTLY maxQueued (8). Pre-fix: the post-drain live
|
||||
// count was wrong, so somewhere between pushes 5 and 7 the 75%
|
||||
// threshold (live=6) fired a second warning prematurely or the
|
||||
// push at 7 was even rejected. Post-fix: live count is the truth,
|
||||
// and the second warning fires exactly at push 8 (live=8, queue
|
||||
// full → push 8 fills the cap and either succeeds at the cap line
|
||||
// or trips the warn check first).
|
||||
let rejected = 0;
|
||||
for (let i = 7; i <= 14; i++) {
|
||||
// Stop publishing once the queue refuses — the 8th live publish
|
||||
// is the maxQueued ceiling.
|
||||
const ok = bus.publish({ type: 'foo', data: i }) !== undefined;
|
||||
if (!ok) rejected++;
|
||||
}
|
||||
void rejected; // EventBus.publish never returns false; rejection
|
||||
// happens inside the bus when subscriber queues fill.
|
||||
|
||||
// Drain everything that's still alive in the iter. The exact frame
|
||||
// shape varies (depending on whether the bus also force-pushed a
|
||||
// second warning + evicted), but the ASSERTION we need is: the
|
||||
// sub didn't get evicted on a phantom premature overflow — i.e.
|
||||
// we received MORE THAN 1 live frame in this episode (pre-fix,
|
||||
// the live count drift evicted after 0-1 frames).
|
||||
const episode2: BridgeEvent[] = [];
|
||||
for (let i = 0; i < 9; i++) {
|
||||
const { value, done } = await it.next();
|
||||
if (done) break;
|
||||
episode2.push(value);
|
||||
}
|
||||
const live2 = episode2.filter((e) => e.id !== undefined && e.id >= 7);
|
||||
// Pre-fix: live2 would be <8 because the queue evicted prematurely
|
||||
// after the buggy live count drift. Post-fix: all 8 live frames
|
||||
// (ids 7..14) get through cleanly.
|
||||
expect(live2.length).toBeGreaterThanOrEqual(8);
|
||||
abort.abort();
|
||||
});
|
||||
|
||||
it('default ring size is 8000 (#3803 §02 target)', async () => {
|
||||
const bus = new EventBus();
|
||||
for (let i = 1; i <= 8001; i++) bus.publish({ type: 'foo', data: i });
|
||||
|
|
|
|||
|
|
@ -446,50 +446,53 @@ function emptyAsyncIterable<T>(): AsyncIterable<T> {
|
|||
* that signal to evict slow subscribers.
|
||||
*
|
||||
* The cap (`maxSize`) applies only to LIVE items pushed via `push()`. Items
|
||||
* inserted via `forcePush()` (the `Last-Event-ID` replay path on subscribe
|
||||
* and the terminal `client_evicted` frame) are tracked separately and don't
|
||||
* inserted via `forcePush()` (the `Last-Event-ID` replay path on subscribe,
|
||||
* the terminal `client_evicted` frame, and the mid-stream
|
||||
* `slow_client_warning` frame) carry a `forced` tag per entry and never
|
||||
* count toward the cap. Without this split, a reconnect with a large
|
||||
* backlog would force-push ~ringSize entries into `buf`, push `buf.length`
|
||||
* past `maxSize`, and the very next live publish would evict the
|
||||
* just-resumed subscriber — defeating the resume contract.
|
||||
*
|
||||
* Previously this class tracked `forcedInBuf` as a count, which was
|
||||
* correct only when forced frames stayed contiguous at the FRONT of the
|
||||
* buffer (subscribe-time replay). The `slow_client_warning` path
|
||||
* force-pushes mid-stream to the BACK of the queue, so the count-based
|
||||
* approach drifted: a live shift would decrement `forcedInBuf`, then a
|
||||
* later cap check on a live push would under-count the live backlog and
|
||||
* warn/evict the client before there were actually `maxSize` live
|
||||
* items. The per-entry `forced` tag below is the position-independent
|
||||
* fix.
|
||||
*/
|
||||
interface BoundedQueueEntry<T> {
|
||||
value: T;
|
||||
/** True for replay / eviction / slow_client_warning frames (don't count toward cap). */
|
||||
forced: boolean;
|
||||
}
|
||||
|
||||
class BoundedAsyncQueue<T> {
|
||||
private readonly buf: T[] = [];
|
||||
private readonly buf: Array<BoundedQueueEntry<T>> = [];
|
||||
private readonly resolvers: Array<(v: IteratorResult<T>) => void> = [];
|
||||
private closed = false;
|
||||
/**
|
||||
* Number of force-pushed items still in `buf`. The cap check in
|
||||
* `push()` only applies to LIVE items; this counter tells us how
|
||||
* many slots in `buf` are replay-injected and shouldn't count.
|
||||
*
|
||||
* Position invariant: under the bus's two callers,
|
||||
* 1. subscribe-time replay (`Last-Event-ID` resume) — forcePush
|
||||
* fires BEFORE any live `push()`, so replay items are at the
|
||||
* front of `buf`;
|
||||
* 2. eviction terminal frame — forcePush fires AFTER `push()`
|
||||
* rejection, then `close()` is called immediately, so the
|
||||
* eviction frame is at the BACK of `buf`.
|
||||
*
|
||||
* `next()` decrements `forcedInBuf` whenever the counter is > 0 on
|
||||
* shift, which is correct for case (1). For case (2) it slightly
|
||||
* misaccounts (decrements on the first live shift), but that's
|
||||
* harmless: the queue is closed so no `push()` runs the cap check
|
||||
* again. The counter only matters for live cap enforcement.
|
||||
* O(1) snapshot of how many LIVE (non-forced) entries are in `buf`.
|
||||
* Maintained directly by `push()`/`next()`: any time a forced entry
|
||||
* is added or removed `liveCount` is untouched; any time a live entry
|
||||
* is added or removed `liveCount` moves with it. Replaces the
|
||||
* position-dependent `forcedInBuf` heuristic — `liveCount` is correct
|
||||
* no matter where in the queue the forced entries are.
|
||||
*/
|
||||
private forcedInBuf = 0;
|
||||
private liveCount = 0;
|
||||
|
||||
constructor(private readonly maxSize: number) {}
|
||||
|
||||
/**
|
||||
* Number of LIVE (non-force-pushed) items currently waiting in the
|
||||
* buffer. Mirrors the cap check in `push()`: replay/eviction frames
|
||||
* inserted via `forcePush` don't count toward the backpressure
|
||||
* threshold the bus uses to decide when to emit
|
||||
* `slow_client_warning`. Returns 0 (not negative) if the buffer
|
||||
* happens to be all-force-pushed.
|
||||
* buffer. Backpressure decisions in `EventBus.publish()` (the
|
||||
* `slow_client_warning` threshold) read this value.
|
||||
*/
|
||||
get size(): number {
|
||||
return Math.max(0, this.buf.length - this.forcedInBuf);
|
||||
return this.liveCount;
|
||||
}
|
||||
|
||||
/** Returns true if accepted, false if dropped due to overflow. */
|
||||
|
|
@ -501,12 +504,14 @@ class BoundedAsyncQueue<T> {
|
|||
return true;
|
||||
}
|
||||
// Cap is on the LIVE backlog only.
|
||||
if (this.buf.length - this.forcedInBuf >= this.maxSize) return false;
|
||||
this.buf.push(value);
|
||||
if (this.liveCount >= this.maxSize) return false;
|
||||
this.buf.push({ value, forced: false });
|
||||
this.liveCount += 1;
|
||||
return true;
|
||||
}
|
||||
|
||||
/** Bypasses the size cap. Used for replay frames and terminal eviction. */
|
||||
/** Bypasses the size cap. Used for replay frames, eviction terminal,
|
||||
* and slow-client warnings. */
|
||||
forcePush(value: T): void {
|
||||
if (this.closed) return;
|
||||
const r = this.resolvers.shift();
|
||||
|
|
@ -514,8 +519,7 @@ class BoundedAsyncQueue<T> {
|
|||
r({ value, done: false });
|
||||
return;
|
||||
}
|
||||
this.buf.push(value);
|
||||
this.forcedInBuf += 1;
|
||||
this.buf.push({ value, forced: true });
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -539,7 +543,7 @@ class BoundedAsyncQueue<T> {
|
|||
// Truncate the buffer so subsequent `next()` calls see the
|
||||
// closed sentinel immediately.
|
||||
this.buf.length = 0;
|
||||
this.forcedInBuf = 0;
|
||||
this.liveCount = 0;
|
||||
}
|
||||
while (this.resolvers.length > 0) {
|
||||
this.resolvers.shift()!({
|
||||
|
|
@ -554,12 +558,9 @@ class BoundedAsyncQueue<T> {
|
|||
// queue whose element type legitimately includes `undefined`. The bus
|
||||
// never pushes undefined today, but the queue is generic.
|
||||
if (this.buf.length > 0) {
|
||||
const value = this.buf.shift() as T;
|
||||
// Force-pushed entries are FIFO at the front of `buf` (forcePush
|
||||
// only happens at subscribe time, before any live push). So as long
|
||||
// as `forcedInBuf > 0` the shifted item is a replay frame.
|
||||
if (this.forcedInBuf > 0) this.forcedInBuf -= 1;
|
||||
return Promise.resolve({ value, done: false });
|
||||
const entry = this.buf.shift() as BoundedQueueEntry<T>;
|
||||
if (!entry.forced) this.liveCount -= 1;
|
||||
return Promise.resolve({ value: entry.value, done: false });
|
||||
}
|
||||
if (this.closed) {
|
||||
return Promise.resolve({
|
||||
|
|
|
|||
|
|
@ -139,6 +139,18 @@ export interface SubscribeOptions {
|
|||
lastEventId?: number;
|
||||
/** Aborts the subscription cleanly. */
|
||||
signal?: AbortSignal;
|
||||
/**
|
||||
* Per-subscriber backlog cap requested from the daemon. Forwarded as
|
||||
* `?maxQueued=N` on `GET /session/:id/events`. Daemon-side range is
|
||||
* `[16, 2048]` (default 256); out-of-range or non-decimal values get
|
||||
* a `400 invalid_max_queued` response. Old daemons without the
|
||||
* `slow_client_warning` capability silently ignore the param — SDK
|
||||
* clients should pre-flight `caps.features.slow_client_warning`
|
||||
* before opting in. Useful for cold reconnects with a large
|
||||
* `Last-Event-ID: 0` replay backlog so the force-pushed replay
|
||||
* frames don't trip the warn / eviction path on the first publish.
|
||||
*/
|
||||
maxQueued?: number;
|
||||
}
|
||||
|
||||
export class DaemonClient {
|
||||
|
|
@ -510,12 +522,19 @@ export class DaemonClient {
|
|||
const fetchSignal = opts.signal
|
||||
? composeAbortSignals([opts.signal, connectCtrl.signal])
|
||||
: connectCtrl.signal;
|
||||
// Build the SSE URL, optionally with `?maxQueued=N`. We don't
|
||||
// validate the value client-side — the daemon's
|
||||
// `parseMaxQueuedQuery` is the source of truth on the range
|
||||
// `[16, 2048]` and returns a structured `400 invalid_max_queued`
|
||||
// for anything outside, so duplicating the bounds here would
|
||||
// diverge if the daemon's range ever shifts.
|
||||
let url = `${this.baseUrl}/session/${encodeURIComponent(sessionId)}/events`;
|
||||
if (opts.maxQueued !== undefined) {
|
||||
url += `?maxQueued=${encodeURIComponent(String(opts.maxQueued))}`;
|
||||
}
|
||||
let res: Response;
|
||||
try {
|
||||
res = await this._fetch(
|
||||
`${this.baseUrl}/session/${encodeURIComponent(sessionId)}/events`,
|
||||
{ headers, signal: fetchSignal },
|
||||
);
|
||||
res = await this._fetch(url, { headers, signal: fetchSignal });
|
||||
} finally {
|
||||
if (connectTimer !== undefined) clearTimeout(connectTimer);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -546,6 +546,47 @@ describe('DaemonClient', () => {
|
|||
const iter = client.subscribeEvents('missing');
|
||||
await expect(iter.next()).rejects.toMatchObject({ status: 404 });
|
||||
});
|
||||
|
||||
it('appends ?maxQueued=N when SubscribeOptions.maxQueued is set', async () => {
|
||||
const { fetch, calls } = recordingFetch(() => sseResponse(''));
|
||||
const client = new DaemonClient({ baseUrl: 'http://daemon', fetch });
|
||||
for await (const _ of client.subscribeEvents('s-1', {
|
||||
maxQueued: 512,
|
||||
})) {
|
||||
/* unreachable */
|
||||
}
|
||||
expect(calls[0]?.url).toBe(
|
||||
'http://daemon/session/s-1/events?maxQueued=512',
|
||||
);
|
||||
});
|
||||
|
||||
it('omits the query string when maxQueued is undefined', async () => {
|
||||
const { fetch, calls } = recordingFetch(() => sseResponse(''));
|
||||
const client = new DaemonClient({ baseUrl: 'http://daemon', fetch });
|
||||
for await (const _ of client.subscribeEvents('s-1', {
|
||||
lastEventId: 7,
|
||||
})) {
|
||||
/* unreachable */
|
||||
}
|
||||
// Bare events URL — no `?` introduced when the caller didn't ask.
|
||||
expect(calls[0]?.url).toBe('http://daemon/session/s-1/events');
|
||||
expect(calls[0]?.headers['last-event-id']).toBe('7');
|
||||
});
|
||||
|
||||
it('propagates a server 400 invalid_max_queued unchanged', async () => {
|
||||
const { fetch } = recordingFetch(() =>
|
||||
jsonResponse(400, {
|
||||
error: '`maxQueued` must be in [16, 2048]',
|
||||
code: 'invalid_max_queued',
|
||||
}),
|
||||
);
|
||||
const client = new DaemonClient({ baseUrl: 'http://daemon', fetch });
|
||||
const iter = client.subscribeEvents('s-1', { maxQueued: 9999 });
|
||||
await expect(iter.next()).rejects.toMatchObject({
|
||||
status: 400,
|
||||
body: { code: 'invalid_max_queued' },
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe('listWorkspaceSessions', () => {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue