bzuccaro
f98043c21b
On a long-lived SSE connection a missed remove leaves a stale request id in the tracker forever; llama-swap only sends a fresh snapshot on connect, so counts drifted up and stayed stuck. The tracker now timestamps each request id (upserts refresh it) and prunes ids not updated within 120s, on every inflight event and on a 30s runtime interval. Removes also accept numeric ids defensively.
107 lines
4.2 KiB
TypeScript
107 lines
4.2 KiB
TypeScript
import assert from "node:assert/strict";
|
|
import { test } from "node:test";
|
|
import { InflightTracker, type FeedEvent } from "../src/lib/inflight-tracker";
|
|
|
|
test("snapshot rebuilds counts for all in-flight requests", () => {
|
|
const tracker = new InflightTracker();
|
|
tracker.apply({
|
|
type: "inflight",
|
|
operation: "snapshot",
|
|
requests: [
|
|
{ model: "A", id: "1" },
|
|
{ model: "A", id: "2" },
|
|
{ model: "B", id: "3" },
|
|
],
|
|
});
|
|
assert.equal(tracker.count("A"), 2);
|
|
assert.equal(tracker.count("B"), 1);
|
|
assert.equal(tracker.count("C"), 0);
|
|
});
|
|
|
|
test("add (upsert) is idempotent per request id", () => {
|
|
const tracker = new InflightTracker();
|
|
tracker.apply({ type: "inflight", operation: "add", requests: [{ model: "A", id: "1" }] });
|
|
tracker.apply({ type: "inflight", operation: "add", requests: [{ model: "A", id: "1" }] });
|
|
tracker.apply({ type: "inflight", operation: "add", requests: [{ model: "A", id: "2" }] });
|
|
assert.equal(tracker.count("A"), 2);
|
|
});
|
|
|
|
test("remove is id-based and decrements the matching request", () => {
|
|
const tracker = new InflightTracker();
|
|
tracker.apply({ type: "inflight", operation: "add", requests: [{ model: "A", id: "1" }, { model: "B", id: "2" }] });
|
|
assert.equal(tracker.count("A"), 1);
|
|
assert.equal(tracker.count("B"), 1);
|
|
tracker.apply({ type: "inflight", operation: "remove", id: "1" });
|
|
assert.equal(tracker.count("A"), 0);
|
|
assert.equal(tracker.count("B"), 1);
|
|
tracker.apply({ type: "inflight", operation: "remove", id: "999" });
|
|
assert.equal(tracker.count("B"), 1);
|
|
});
|
|
|
|
test("snapshot replaces prior in-flight state", () => {
|
|
const tracker = new InflightTracker();
|
|
tracker.apply({ type: "inflight", operation: "add", requests: [{ model: "A", id: "1" }] });
|
|
tracker.apply({ type: "inflight", operation: "snapshot", requests: [{ model: "A", id: "2" }] });
|
|
assert.equal(tracker.count("A"), 1);
|
|
assert.equal(tracker.count("B"), 0);
|
|
});
|
|
|
|
test("modelStatus normalizes states", () => {
|
|
const tracker = new InflightTracker();
|
|
tracker.apply({
|
|
type: "modelStatus",
|
|
models: [
|
|
{ id: "A", state: "ready" },
|
|
{ id: "B", state: "loading" },
|
|
{ id: "C", state: "stopped" },
|
|
{ id: "D", state: "weird" },
|
|
],
|
|
});
|
|
assert.equal(tracker.state("A"), "ready");
|
|
assert.equal(tracker.state("B"), "loading");
|
|
assert.equal(tracker.state("C"), "stopped");
|
|
assert.equal(tracker.state("D"), "stopped");
|
|
assert.equal(tracker.state("missing"), undefined);
|
|
});
|
|
|
|
test("total sums in-flight requests across all models", () => {
|
|
const tracker = new InflightTracker();
|
|
assert.equal(tracker.total(), 0);
|
|
tracker.apply({ type: "inflight", operation: "add", requests: [{ model: "A", id: "1" }, { model: "B", id: "2" }] });
|
|
tracker.apply({ type: "inflight", operation: "add", requests: [{ model: "A", id: "3" }] });
|
|
assert.equal(tracker.total(), 3);
|
|
tracker.apply({ type: "inflight", operation: "remove", id: "1" });
|
|
assert.equal(tracker.total(), 2);
|
|
});
|
|
|
|
test("prune drops request ids that stopped receiving updates", () => {
|
|
let now = 0;
|
|
const tracker = new InflightTracker(() => now);
|
|
tracker.apply({ type: "inflight", operation: "add", requests: [{ model: "A", id: "1" }] });
|
|
now = 200_000;
|
|
tracker.apply({ type: "inflight", operation: "add", requests: [{ model: "A", id: "2" }] });
|
|
tracker.prune(60_000);
|
|
assert.equal(tracker.total(), 1);
|
|
assert.equal(tracker.count("A"), 1);
|
|
});
|
|
|
|
test("active requests refreshed by upserts survive pruning", () => {
|
|
let now = 0;
|
|
const tracker = new InflightTracker(() => now);
|
|
tracker.apply({ type: "inflight", operation: "add", requests: [{ model: "A", id: "1" }] });
|
|
now = 200_000;
|
|
tracker.apply({ type: "inflight", operation: "add", requests: [{ model: "A", id: "1" }] });
|
|
tracker.prune(60_000);
|
|
assert.equal(tracker.total(), 1);
|
|
});
|
|
|
|
test("prune is applied on every inflight event", () => {
|
|
let now = 0;
|
|
const tracker = new InflightTracker(() => now);
|
|
tracker.apply({ type: "inflight", operation: "add", requests: [{ model: "A", id: "stale" }] });
|
|
now = 200_000;
|
|
tracker.apply({ type: "inflight", operation: "add", requests: [{ model: "B", id: "live" }] });
|
|
assert.equal(tracker.total(), 1);
|
|
assert.equal(tracker.count("B"), 1);
|
|
});
|