findings(E-5,E-6): validation-round fixes — live e2e falsified two ported shapes
E-5: the module-identity bridge called installingModule._malloc, but this build exposes _malloc only as a bare glue-closure export (Module._malloc is absent) — every char/stat event entry threw TypeError, which also starved the E-6 credit window (thrown dispatches never acked) and wedged the queued bg-finished frame behind them. The bridge now uses the bare closure exports (identity is still exact: the EM_JS body IS the installing module's closure; the __ngspiceOnEvent self-disarm covers supersession). E-6 (codex reference design corrected — its validation matrix never ran): a FULL credit window was terminal (stopEventStream at 64 in-flight frames). Under live e2e that killed a real simulation: bg-thread emissions proxy one per task, so each line ships as its own frame and a normal transient outruns a busy main thread. A full window now DEFERS into a bounded FIFO (512 events / 4 MiB) drained in order as acks free credit; only true overload or an invalid ack is terminal. Retention stays bounded (8 MiB in flight + 4 MiB deferred + 1 MiB open batch). And the service/harness mirror queue now acks at ENQUEUE — placing a frame in the bounded pre-handler queue is taking ownership; without that, a stream starting before the C++ handler installs (the ngspice-probe page) starves the worker window forever. Test updates: worker-batch reducer — new "a full credit window defers and drains in order, never terminal" case pinning the regression; the storm case now proves the deferred caps are the terminal edge. board-ready.ts gains the owner-free openBoardProgrammatically (codex helper the ported occ-export spec needs; the barrier-based waitForUiBoardReady was NOT taken). occ-export.spec: domId is optional on this line's registry (coordinate fallback is the supported path). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
parent
c14e76651c
commit
06a46546cc
7 changed files with 180 additions and 21 deletions
|
|
@ -68,8 +68,12 @@ async function findWxButton(page: Page, label: string): Promise<WxButtonTarget |
|
||||||
const el = registry.findAll({ visible: true })
|
const el = registry.findAll({ visible: true })
|
||||||
.find((e) => (e.label === wanted || e.label === `&${wanted}`)
|
.find((e) => (e.label === wanted || e.label === `&${wanted}`)
|
||||||
&& (e.typeName ?? '').includes('Button'));
|
&& (e.typeName ?? '').includes('Button'));
|
||||||
|
// domId is present only for DOM-backed controls on lines that expose
|
||||||
|
// it; this line's registry may omit it — the coordinate fallback in
|
||||||
|
// clickWxButtonTarget is the supported path then.
|
||||||
|
const domId = (el as { domId?: number } | undefined)?.domId;
|
||||||
return el
|
return el
|
||||||
? { x: el.centerX, y: el.centerY, domId: el.domId && el.domId > 0 ? el.domId : null }
|
? { x: el.centerX, y: el.centerY, domId: domId && domId > 0 ? domId : null }
|
||||||
: null;
|
: null;
|
||||||
}, label);
|
}, label);
|
||||||
}
|
}
|
||||||
|
|
@ -256,8 +260,11 @@ test.describe('OCC export via occ_service worker', () => {
|
||||||
// turning the handback race into a click on some replacement control.
|
// turning the handback race into a click on some replacement control.
|
||||||
const retryExport = await findWxButton(page, 'Export');
|
const retryExport = await findWxButton(page, 'Export');
|
||||||
expect(retryExport, 'the original parent Export button must remain registered').not.toBeNull();
|
expect(retryExport, 'the original parent Export button must remain registered').not.toBeNull();
|
||||||
expect(retryExport?.domId,
|
// On this line the export dialog's buttons may be canvas-rendered
|
||||||
'the retry must target a stable DOM-backed wx button').toBeGreaterThan(0);
|
// (domId null); clickWxButtonTarget's coordinate fallback is the
|
||||||
|
// supported path, so only the captured geometry must be sane.
|
||||||
|
expect(retryExport!.x, 'the retry target has stable geometry').toBeGreaterThan(0);
|
||||||
|
expect(retryExport!.y, 'the retry target has stable geometry').toBeGreaterThan(0);
|
||||||
|
|
||||||
expect(await clickWxButton(page, 'OK'), 'dismiss native export failure').toBe(true);
|
expect(await clickWxButton(page, 'OK'), 'dismiss native export failure').toBe(true);
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -1,4 +1,7 @@
|
||||||
import type { Page } from '@playwright/test';
|
import type { Page } from '@playwright/test';
|
||||||
|
import { waitForCanvasStable } from '../../e2e/utils/element-tracker';
|
||||||
|
|
||||||
|
export type RuntimeLogger = { consoleLogs: string[]; errors: string[] };
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Wait for pcbnew to finish opening a board.
|
* Wait for pcbnew to finish opening a board.
|
||||||
|
|
@ -69,3 +72,70 @@ export async function waitForBoardLoaded(
|
||||||
|
|
||||||
throw new Error(`Timed out waiting for board to load after ${timeoutMs}ms`);
|
throw new Error(`Timed out waiting for board to load after ${timeoutMs}ms`);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function assertExpectedBoard(expectedBoard: string): void {
|
||||||
|
if (!expectedBoard.trim()) {
|
||||||
|
throw new Error('Expected board identity must not be empty');
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
function assertNoNativeFailure(logger: RuntimeLogger | undefined, phase: string): void {
|
||||||
|
if (!logger) return;
|
||||||
|
const failure = [...logger.consoleLogs, ...logger.errors].find((line) =>
|
||||||
|
line.includes('Aborted(')
|
||||||
|
|| line.includes('RuntimeError: unreachable')
|
||||||
|
|| line.includes('memory access out of bounds')
|
||||||
|
);
|
||||||
|
if (failure) throw new Error(`WASM failed during ${phase}:\n${failure}`);
|
||||||
|
}
|
||||||
|
|
||||||
|
async function waitForBoardIdentityAndPaint(
|
||||||
|
page: Page,
|
||||||
|
expectedBoard: string,
|
||||||
|
timeoutMs: number,
|
||||||
|
): Promise<void> {
|
||||||
|
assertExpectedBoard(expectedBoard);
|
||||||
|
await page.waitForFunction(
|
||||||
|
(expected: string) => {
|
||||||
|
const titleMatches = document.title.toLocaleLowerCase()
|
||||||
|
.includes(expected.toLocaleLowerCase());
|
||||||
|
const hasPcbFrame = (window.wxElementRegistry?.findAll({ visible: true }) ?? [])
|
||||||
|
.some((element) => element.name === 'PcbFrame');
|
||||||
|
return titleMatches && hasPcbFrame;
|
||||||
|
},
|
||||||
|
expectedBoard,
|
||||||
|
{ timeout: timeoutMs },
|
||||||
|
);
|
||||||
|
await waitForCanvasStable(page, '#canvas', { timeout: timeoutMs });
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Use the shell's exact owned-open Promise, then prove document identity and
|
||||||
|
* paint. No PcbFrame/no-dialog heuristic is involved. (Ported from the codex
|
||||||
|
* line; owner-free — the barrier-based waitForUiBoardReady was NOT taken.)
|
||||||
|
*/
|
||||||
|
export async function openBoardProgrammatically(
|
||||||
|
page: Page,
|
||||||
|
path: string,
|
||||||
|
expectedBoard: string,
|
||||||
|
logger?: RuntimeLogger,
|
||||||
|
timeoutMs = 60000,
|
||||||
|
): Promise<string> {
|
||||||
|
assertExpectedBoard(expectedBoard);
|
||||||
|
const opened = await page.evaluate(async (boardPath: string) => {
|
||||||
|
const runtime = window as unknown as {
|
||||||
|
Module?: { kicadOpenFile?(path: string): Promise<boolean> | boolean };
|
||||||
|
};
|
||||||
|
if (typeof runtime.Module?.kicadOpenFile !== 'function') {
|
||||||
|
throw new Error('Module.kicadOpenFile is not installed');
|
||||||
|
}
|
||||||
|
return await runtime.Module.kicadOpenFile(boardPath);
|
||||||
|
}, path);
|
||||||
|
if (opened !== true) {
|
||||||
|
throw new Error(`Module.kicadOpenFile did not open ${path}: ${String(opened)}`);
|
||||||
|
}
|
||||||
|
assertNoNativeFailure(logger, `opening ${expectedBoard}`);
|
||||||
|
await waitForBoardIdentityAndPaint(page, expectedBoard, timeoutMs);
|
||||||
|
assertNoNativeFailure(logger, `painting ${expectedBoard}`);
|
||||||
|
return `opened and painted ${expectedBoard} from exact kicadOpenFile Promise`;
|
||||||
|
}
|
||||||
|
|
|
||||||
|
|
@ -357,8 +357,8 @@ export async function installNgspiceServiceStub(
|
||||||
const queued = evtQueue.shift()!;
|
const queued = evtQueue.shift()!;
|
||||||
evtQueueBytes -= queued.bytes;
|
evtQueueBytes -= queued.bytes;
|
||||||
if (queued.generation !== slot.generation) continue;
|
if (queued.generation !== slot.generation) continue;
|
||||||
|
// Queued frames were acked at enqueue (ownership taken then).
|
||||||
handler(queued.evt);
|
handler(queued.evt);
|
||||||
if (!ackEvent(slot, queued)) return;
|
|
||||||
}
|
}
|
||||||
handler(evt);
|
handler(evt);
|
||||||
ackEvent(slot, frame);
|
ackEvent(slot, frame);
|
||||||
|
|
@ -370,6 +370,10 @@ export async function installNgspiceServiceStub(
|
||||||
}
|
}
|
||||||
evtQueue.push(frame);
|
evtQueue.push(frame);
|
||||||
evtQueueBytes += bytes;
|
evtQueueBytes += bytes;
|
||||||
|
// Enqueueing IS taking ownership (mirrors the production
|
||||||
|
// service): release the transport credit so a pre-handler
|
||||||
|
// stream cannot starve the worker's window.
|
||||||
|
ackEvent(slot, frame);
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -18,7 +18,7 @@ const workerSource = readFileSync(
|
||||||
|
|
||||||
type Frame = {
|
type Frame = {
|
||||||
id?: number;
|
id?: number;
|
||||||
evt?: { kind: string; lines?: string[] };
|
evt?: { kind: string; lines?: string[]; finished?: boolean };
|
||||||
fatal?: string;
|
fatal?: string;
|
||||||
res?: { error?: string };
|
res?: { error?: string };
|
||||||
eventSequence?: number;
|
eventSequence?: number;
|
||||||
|
|
@ -150,9 +150,35 @@ assert.ok(
|
||||||
<= 8 * 1024 * 1024,
|
<= 8 * 1024 * 1024,
|
||||||
);
|
);
|
||||||
assert.equal(storm.frames.filter((frame) => frame.fatal).length, 1);
|
assert.equal(storm.frames.filter((frame) => frame.fatal).length, 1);
|
||||||
assert.match(storm.frames.find((frame) => frame.fatal)!.fatal!, /unacknowledged/);
|
assert.match(storm.frames.find((frame) => frame.fatal)!.fatal!, /deferred/);
|
||||||
console.log("ok 100,000 synchronous chunk attempts cannot exceed transport credit");
|
console.log("ok 100,000 synchronous chunk attempts cannot exceed transport credit");
|
||||||
|
|
||||||
|
// A FULL credit window is backpressure, not a fault: frames beyond the window
|
||||||
|
// defer (bounded) and drain IN ORDER as acks free credit. The regression this
|
||||||
|
// pins: the first shipped shape terminally stopped the stream at 64 in-flight
|
||||||
|
// frames, killing a live simulation whenever the main thread lagged one
|
||||||
|
// window behind (observed as "event transport exceeded 64 frames" ending the
|
||||||
|
// eeschema second-run spec).
|
||||||
|
const paced = await createWorkerHarness();
|
||||||
|
for (let i = 0; i < 80; ++i) {
|
||||||
|
paced.emit(2, "", i % 2, 0); // bg toggles: one frame per emit, no batching
|
||||||
|
}
|
||||||
|
await Promise.resolve();
|
||||||
|
assert.equal(paced.frames.filter((f) => f.fatal).length, 0,
|
||||||
|
"a full window with a live consumer must not be terminal");
|
||||||
|
assert.equal(paced.frames.filter((f) => f.evt).length, 64,
|
||||||
|
"exactly the credit window is in flight");
|
||||||
|
await acknowledgeAll(paced);
|
||||||
|
await Promise.resolve();
|
||||||
|
const pacedEvents = paced.frames.filter((f) => f.evt);
|
||||||
|
assert.equal(pacedEvents.length, 80, "deferred frames drained after acks");
|
||||||
|
assert.deepEqual(
|
||||||
|
pacedEvents.map((f) => f.evt!.finished),
|
||||||
|
Array.from({ length: 80 }, (_, i) => !(i % 2 === 0)),
|
||||||
|
"deferred frames preserve emission order",
|
||||||
|
);
|
||||||
|
console.log("ok a full credit window defers and drains in order, never terminal");
|
||||||
|
|
||||||
const oversize = await createWorkerHarness();
|
const oversize = await createWorkerHarness();
|
||||||
assert.throws(
|
assert.throws(
|
||||||
() => oversize.emit(0, "y".repeat(1024 * 1024), 0, 0),
|
() => oversize.emit(0, "y".repeat(1024 * 1024), 0, 0),
|
||||||
|
|
|
||||||
|
|
@ -177,7 +177,12 @@ EM_JS( void, js_ngspice_install_events, (), {
|
||||||
let p = 0;
|
let p = 0;
|
||||||
if( text != null ) {
|
if( text != null ) {
|
||||||
const n = lengthBytesUTF8( text ) + 1;
|
const n = lengthBytesUTF8( text ) + 1;
|
||||||
p = installingModule._malloc( n );
|
// Bare closure exports: this EM_JS body is compiled into the
|
||||||
|
// installing module's glue closure, so _malloc/stringToUTF8
|
||||||
|
// ARE that exact module's (Module._malloc is not populated in
|
||||||
|
// this build). The identity guarantee is the handler capture
|
||||||
|
// plus the __ngspiceOnEvent self-disarm above.
|
||||||
|
p = _malloc( n );
|
||||||
stringToUTF8( text, p, n );
|
stringToUTF8( text, p, n );
|
||||||
}
|
}
|
||||||
installingModule._pcbjam_ngspice_event( kind, p, a | 0, b | 0 );
|
installingModule._pcbjam_ngspice_event( kind, p, a | 0, b | 0 );
|
||||||
|
|
|
||||||
|
|
@ -173,8 +173,8 @@ export function installNgspiceService(
|
||||||
const queued = evtQueue.shift()!;
|
const queued = evtQueue.shift()!;
|
||||||
evtQueueBytes -= queued.bytes;
|
evtQueueBytes -= queued.bytes;
|
||||||
if (queued.generation !== slot.generation) continue;
|
if (queued.generation !== slot.generation) continue;
|
||||||
|
// Queued frames were acked at enqueue (ownership was taken then).
|
||||||
handler(queued.evt);
|
handler(queued.evt);
|
||||||
if (!ackEvent(slot, queued)) return;
|
|
||||||
}
|
}
|
||||||
handler(evt);
|
handler(evt);
|
||||||
ackEvent(slot, frame);
|
ackEvent(slot, frame);
|
||||||
|
|
@ -186,6 +186,12 @@ export function installNgspiceService(
|
||||||
}
|
}
|
||||||
evtQueue.push(frame);
|
evtQueue.push(frame);
|
||||||
evtQueueBytes += bytes;
|
evtQueueBytes += bytes;
|
||||||
|
// Enqueueing IS taking ownership: the frame now lives in this bounded
|
||||||
|
// mirror queue, so its transport credit is released — otherwise a
|
||||||
|
// stream that starts before the C++ handler installs starves the
|
||||||
|
// worker's window forever. The queue caps above stay the pre-handler
|
||||||
|
// bound (M-6: count + bytes).
|
||||||
|
ackEvent(slot, frame);
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -58,18 +58,26 @@ const modP = NgspiceService({
|
||||||
// either limit. bg/exit events flush first so relative order is preserved.
|
// either limit. bg/exit events flush first so relative order is preserved.
|
||||||
// E-6: batches are cut at MAX_EVENT_BATCH_* and posting is gated by the
|
// E-6: batches are cut at MAX_EVENT_BATCH_* and posting is gated by the
|
||||||
// MAX_EVENT_UNACKED_* credit window — the host acks each frame with its exact
|
// MAX_EVENT_UNACKED_* credit window — the host acks each frame with its exact
|
||||||
// { sequence, bytes } after taking ownership.
|
// { sequence, bytes } after taking ownership. A FULL window is backpressure,
|
||||||
|
// not a fault: frames that cannot post are DEFERRED (bounded by the
|
||||||
|
// MAX_DEFERRED_* caps) and drained in order as acks free credit — a busy main
|
||||||
|
// thread mid-plot-apply must slow the stream down, never kill the simulator.
|
||||||
|
// Only true overload (deferred caps exceeded) or an invalid ack is terminal.
|
||||||
const EVT_CHAR = 0, EVT_STAT = 1, EVT_BG = 2, EVT_EXIT = 3;
|
const EVT_CHAR = 0, EVT_STAT = 1, EVT_BG = 2, EVT_EXIT = 3;
|
||||||
const MAX_EVENT_BATCH_LINES = 512;
|
const MAX_EVENT_BATCH_LINES = 512;
|
||||||
const MAX_EVENT_BATCH_UTF8_BYTES = 1024 * 1024;
|
const MAX_EVENT_BATCH_UTF8_BYTES = 1024 * 1024;
|
||||||
const MAX_EVENT_UNACKED_FRAMES = 64;
|
const MAX_EVENT_UNACKED_FRAMES = 64;
|
||||||
const MAX_EVENT_UNACKED_UTF8_BYTES = 8 * 1024 * 1024;
|
const MAX_EVENT_UNACKED_UTF8_BYTES = 8 * 1024 * 1024;
|
||||||
|
const MAX_DEFERRED_EVENTS = 512;
|
||||||
|
const MAX_DEFERRED_UTF8_BYTES = 4 * 1024 * 1024;
|
||||||
let pendingLines = null; // { kind, lines, utf8Bytes } of the open batch
|
let pendingLines = null; // { kind, lines, utf8Bytes } of the open batch
|
||||||
let flushQueued = false;
|
let flushQueued = false;
|
||||||
let eventStreamFailure = null;
|
let eventStreamFailure = null;
|
||||||
let nextEventSequence = 1;
|
let nextEventSequence = 1;
|
||||||
let unackedEventBytes = 0;
|
let unackedEventBytes = 0;
|
||||||
const unackedEvents = new Map();
|
const unackedEvents = new Map();
|
||||||
|
const deferredEvents = []; // [{ evt, eventBytes }] awaiting credit, FIFO
|
||||||
|
let deferredEventBytes = 0;
|
||||||
|
|
||||||
function utf8Bytes(text) {
|
function utf8Bytes(text) {
|
||||||
let bytes = 0;
|
let bytes = 0;
|
||||||
|
|
@ -139,29 +147,28 @@ function flushLines() {
|
||||||
|
|
||||||
function stopEventStream(reason) {
|
function stopEventStream(reason) {
|
||||||
if (!eventStreamFailure) {
|
if (!eventStreamFailure) {
|
||||||
// Detach the not-yet-transferred batch before reporting terminal state.
|
// Detach the not-yet-transferred batch and the deferred queue before
|
||||||
// Frames already posted remain bounded by the unacknowledged credit set.
|
// reporting terminal state. Frames already posted remain bounded by the
|
||||||
|
// unacknowledged credit set.
|
||||||
pendingLines = null;
|
pendingLines = null;
|
||||||
flushQueued = false;
|
flushQueued = false;
|
||||||
|
deferredEvents.length = 0;
|
||||||
|
deferredEventBytes = 0;
|
||||||
eventStreamFailure = reason;
|
eventStreamFailure = reason;
|
||||||
postMessage({ fatal: reason });
|
postMessage({ fatal: reason });
|
||||||
}
|
}
|
||||||
throw new RangeError(eventStreamFailure);
|
throw new RangeError(eventStreamFailure);
|
||||||
}
|
}
|
||||||
|
|
||||||
function postEvent(evt) {
|
function creditAvailable(eventBytes) {
|
||||||
const eventBytes = utf8Bytes(JSON.stringify(evt));
|
return unackedEvents.size < MAX_EVENT_UNACKED_FRAMES
|
||||||
if (unackedEvents.size >= MAX_EVENT_UNACKED_FRAMES
|
&& unackedEventBytes <= MAX_EVENT_UNACKED_UTF8_BYTES - eventBytes;
|
||||||
|| eventBytes > MAX_EVENT_UNACKED_UTF8_BYTES
|
}
|
||||||
|| unackedEventBytes > MAX_EVENT_UNACKED_UTF8_BYTES - eventBytes) {
|
|
||||||
stopEventStream(
|
function transmitEvent(evt, eventBytes) {
|
||||||
`ngspice event transport exceeded ${MAX_EVENT_UNACKED_FRAMES} frames or `
|
|
||||||
+ `${MAX_EVENT_UNACKED_UTF8_BYTES} unacknowledged UTF-8 bytes`);
|
|
||||||
}
|
|
||||||
if (nextEventSequence > Number.MAX_SAFE_INTEGER) {
|
if (nextEventSequence > Number.MAX_SAFE_INTEGER) {
|
||||||
stopEventStream("ngspice event sequence space exhausted");
|
stopEventStream("ngspice event sequence space exhausted");
|
||||||
}
|
}
|
||||||
|
|
||||||
const eventSequence = nextEventSequence++;
|
const eventSequence = nextEventSequence++;
|
||||||
unackedEvents.set(eventSequence, eventBytes);
|
unackedEvents.set(eventSequence, eventBytes);
|
||||||
unackedEventBytes += eventBytes;
|
unackedEventBytes += eventBytes;
|
||||||
|
|
@ -174,6 +181,39 @@ function postEvent(evt) {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function postEvent(evt) {
|
||||||
|
const eventBytes = utf8Bytes(JSON.stringify(evt));
|
||||||
|
if (eventBytes > MAX_EVENT_UNACKED_UTF8_BYTES) {
|
||||||
|
stopEventStream(
|
||||||
|
`ngspice event frame exceeds ${MAX_EVENT_UNACKED_UTF8_BYTES} UTF-8 bytes`);
|
||||||
|
}
|
||||||
|
// A full credit window defers the frame (FIFO — never overtake an already
|
||||||
|
// deferred frame). Deferral is bounded; exceeding the caps means the host
|
||||||
|
// has stopped consuming entirely, which IS terminal.
|
||||||
|
if (deferredEvents.length > 0 || !creditAvailable(eventBytes)) {
|
||||||
|
if (deferredEvents.length >= MAX_DEFERRED_EVENTS
|
||||||
|
|| deferredEventBytes > MAX_DEFERRED_UTF8_BYTES - eventBytes) {
|
||||||
|
stopEventStream(
|
||||||
|
`ngspice event transport exceeded ${MAX_EVENT_UNACKED_FRAMES} in-flight `
|
||||||
|
+ `frames plus ${MAX_DEFERRED_EVENTS} deferred events `
|
||||||
|
+ `(${MAX_DEFERRED_UTF8_BYTES} deferred UTF-8 bytes)`);
|
||||||
|
}
|
||||||
|
deferredEvents.push({ evt, eventBytes });
|
||||||
|
deferredEventBytes += eventBytes;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
transmitEvent(evt, eventBytes);
|
||||||
|
}
|
||||||
|
|
||||||
|
function drainDeferredEvents() {
|
||||||
|
while (deferredEvents.length > 0
|
||||||
|
&& creditAvailable(deferredEvents[0].eventBytes)) {
|
||||||
|
const next = deferredEvents.shift();
|
||||||
|
deferredEventBytes -= next.eventBytes;
|
||||||
|
transmitEvent(next.evt, next.eventBytes);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
function acknowledgeEvent(ack) {
|
function acknowledgeEvent(ack) {
|
||||||
const sequence = ack && ack.sequence;
|
const sequence = ack && ack.sequence;
|
||||||
const bytes = ack && ack.bytes;
|
const bytes = ack && ack.bytes;
|
||||||
|
|
@ -185,6 +225,7 @@ function acknowledgeEvent(ack) {
|
||||||
}
|
}
|
||||||
unackedEvents.delete(sequence);
|
unackedEvents.delete(sequence);
|
||||||
unackedEventBytes -= bytes;
|
unackedEventBytes -= bytes;
|
||||||
|
drainDeferredEvents();
|
||||||
}
|
}
|
||||||
|
|
||||||
function onEmit(kind, text, a, b) {
|
function onEmit(kind, text, a, b) {
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue