From bc37b88728ab19d450a04813b43ee247efb4a459 Mon Sep 17 00:00:00 2001 From: Jonathan Baldie Date: Thu, 27 Aug 2026 20:46:18 +0100 Subject: [PATCH] fix: enforce count limit, snapshot persist on load, JSON dequeue Enqueue now throws when queueCountLimit is reached and does not register a queue before depth/count checks, so depth 0 cannot leak an empty queue. load() rewrites persist.dat as a snapshot of remaining items instead of clearing the log. dequeue and peek always encode payloads as application/json so string and number values do not collide. --- README.md | 4 +- src/handler.ts | 9 ++--- src/manager.ts | 22 +++++------ tests/handler_test.ts | 83 ++++++++++++++++++++++++++++++--------- tests/manager_test.ts | 91 ++++++++++++++++++++++++++++++++++++++++++- 5 files changed, 167 insertions(+), 42 deletions(-) diff --git a/README.md b/README.md index 8f8b7cd..06ad956 100644 --- a/README.md +++ b/README.md @@ -52,7 +52,7 @@ To get the next payload from the `foo` queue, send a get request to `/dequeue/:q curl -X GET http://127.0.0.1:1991/dequeue/foo ``` -This returns the oldest added payload on queue `foo` and removes it, guaranteeing both the order and that each payload will only be read once. +This returns the oldest added payload on queue `foo` as JSON and removes it, guaranteeing both the order and that each payload will only be read once. Strings, numbers, booleans, arrays, and objects all use `application/json` so a string `"0"` is distinct from the number `0`. That's all you need to get started! 😎 @@ -113,7 +113,7 @@ To get persistency, simply add the `--persist` option when starting up the serve docker run -d -e PORT=1991 -e HOST=0.0.0.0 -e PERSIST=/mnt/ jonbaldie/queue /usr/bin/queue --persist ``` -If the server sees that the `persist.dat` file exists on startup, it will run the binary log from the beginning and then clear the file down. +If the server sees that the `persist.dat` file exists on startup, it will replay the binary log and then rewrite the file as a snapshot of remaining items. When using Docker, it might be useful to add `persist.dat` as a persistent volume to keep your binary logs safe. diff --git a/src/handler.ts b/src/handler.ts index 509b3b2..2f4ea10 100644 --- a/src/handler.ts +++ b/src/handler.ts @@ -67,12 +67,9 @@ function itemResponse(item: unknown): Response { if (item === undefined) { return new Response(null, { status: 204 }); } - if (typeof item === "object" && item !== null) { - return new Response(JSON.stringify(item), { - headers: { "Content-Type": "application/json" }, - }); - } - return new Response(String(item)); + return new Response(JSON.stringify(item), { + headers: { "Content-Type": "application/json" }, + }); } function dequeueHandler(mgr: QueueManager): RouteHandler { diff --git a/src/manager.ts b/src/manager.ts index 5380df9..9e81a5f 100644 --- a/src/manager.ts +++ b/src/manager.ts @@ -83,10 +83,6 @@ export default class Manager { } } - private registered(name: string): boolean { - return this.queues.has(name); - } - public canCreateQueue(): boolean { return this.queues.size < this.queueCountLimit; } @@ -95,10 +91,8 @@ export default class Manager { this.validateName(name); const queue = this.find(name); if (!queue) { - // Creating a new queue - check if we have room - return this.canCreateQueue(); + return this.canCreateQueue() && 0 < this.queueDepthLimit; } - // Existing queue - check if it has room return queue.length < this.queueDepthLimit; } @@ -108,15 +102,17 @@ export default class Manager { public enqueue(name: string, payload: T): Manager { this.validateName(name); - const queue = this.find(name) || new FIFOQueue(); - - if (this.registered(name) === false) { - this.register(name, queue); + const existing = this.find(name); + if (!existing && !this.canCreateQueue()) { + throw new Error("Queue count limit reached"); } - + const queue = existing || new FIFOQueue(); if (queue.length >= this.queueDepthLimit) { throw new Error("Queue depth limit reached"); } + if (!existing) { + this.register(name, queue); + } queue.push(payload); if (this.persistEnabled) { @@ -183,7 +179,7 @@ export default class Manager { for (const event of this.store.loadState()) { this.applyLoadedEvent(event); } - this.store.clear(); + this.save(); } private applyLoadedEvent(event: QueueEvent): void { diff --git a/tests/handler_test.ts b/tests/handler_test.ts index e0f71ab..043edf7 100644 --- a/tests/handler_test.ts +++ b/tests/handler_test.ts @@ -747,8 +747,8 @@ Deno.test("GET peek returns 200 with payload when queue has items", async () => }); const response = await handler(request); assertEquals(response.status, 200); - const body = await response.text(); - assertEquals(body, "peekable"); + assertEquals(response.headers.get("content-type"), "application/json"); + assertEquals(await response.json(), "peekable"); }); // Happy path: GET peek returns 204 when queue is empty @@ -806,7 +806,7 @@ Deno.test("dequeue returns application/json for array payload", async () => { assertEquals(body, [1, 2, 3]); }); -Deno.test("dequeue returns text/plain for number payload", async () => { +Deno.test("dequeue returns application/json for number payload", async () => { const mgr = new QueueManager(new Persistency.MemoryStore); const handler = createHandler(mgr, API_TOKEN); @@ -823,11 +823,11 @@ Deno.test("dequeue returns text/plain for number payload", async () => { }); const response = await handler(request); assertEquals(response.status, 200); - const body = await response.text(); - assertEquals(body, "42"); + assertEquals(response.headers.get("content-type"), "application/json"); + assertEquals(await response.json(), 42); }); -Deno.test("dequeue returns text/plain for boolean payload", async () => { +Deno.test("dequeue returns application/json for boolean payload", async () => { const mgr = new QueueManager(new Persistency.MemoryStore); const handler = createHandler(mgr, API_TOKEN); @@ -844,8 +844,8 @@ Deno.test("dequeue returns text/plain for boolean payload", async () => { }); const response = await handler(request); assertEquals(response.status, 200); - const body = await response.text(); - assertEquals(body, "true"); + assertEquals(response.headers.get("content-type"), "application/json"); + assertEquals(await response.json(), true); }); // queue-nyc: Missing payload key returns 400 @@ -1003,7 +1003,7 @@ Deno.test("body read error returns 413 not 400", async () => { assertEquals(response.status, 413); }); -Deno.test("dequeue returns text/plain for string payload", async () => { +Deno.test("dequeue returns application/json for string payload", async () => { const mgr = new QueueManager(new Persistency.MemoryStore); const handler = createHandler(mgr, API_TOKEN); @@ -1020,8 +1020,8 @@ Deno.test("dequeue returns text/plain for string payload", async () => { }); const response = await handler(request); assertEquals(response.status, 200); - const body = await response.text(); - assertEquals(body, "hello world"); + assertEquals(response.headers.get("content-type"), "application/json"); + assertEquals(await response.json(), "hello world"); }); Deno.test("Manager: enqueue followed by multiple dequeues (catches state corruption)", () => { @@ -1145,8 +1145,7 @@ Deno.test("API: enqueue stores payload and dequeue retrieves it (catches data lo }) ); assertEquals(deqRes.status, 200); - const retrieved = await deqRes.text(); - assertEquals(retrieved, payload); + assertEquals(await deqRes.json(), payload); }); Deno.test("API: dequeue empty queue returns 204 (catches crash)", async () => { @@ -1271,8 +1270,7 @@ Deno.test("API: FIFO order through HTTP (catches dequeue order mutation)", async headers: { "Authorization": `Bearer ${TEST_TOKEN}` }, }) ); - const actual = await res.text(); - assertEquals(actual, expected); + assertEquals(await res.json(), expected); } }); @@ -1297,13 +1295,13 @@ Deno.test("API: multiple queues isolated (catches queue mixing)", async () => { headers: { "Authorization": `Bearer ${TEST_TOKEN}` }, }) ); - assertEquals(await res1.text(), "q1-item"); + assertEquals(await res1.json(), "q1-item"); const res2 = await handler( new Request("http://localhost/dequeue/q2", { headers: { "Authorization": `Bearer ${TEST_TOKEN}` }, }) ); - assertEquals(await res2.text(), "q2-item"); + assertEquals(await res2.json(), "q2-item"); }); Deno.test("API: enqueue with empty payload (catches validation)", async () => { @@ -1347,8 +1345,7 @@ Deno.test("API: very long payload (catches buffer handling)", async () => { headers: { "Authorization": `Bearer ${TEST_TOKEN}` }, }) ); - const retrieved = await deqRes.text(); - assertEquals(retrieved, longPayload); + assertEquals(await deqRes.json(), longPayload); }); Deno.test("auth: no token on enqueue returns 401", async () => { @@ -1449,3 +1446,51 @@ Deno.test("auth: valid token on length returns 200", async () => { const res = await handler(req); assertEquals(200, res.status); }); + +Deno.test("dequeue distinguishes string zero from number zero", async () => { + const mgr = new QueueManager(new Persistency.MemoryStore); + const handler = createHandler(mgr, API_TOKEN); + + await handler(new Request("http://localhost:3000/enqueue/q", { + method: "POST", + body: JSON.stringify({ payload: "0" }), + headers: authHeaders, + })); + await handler(new Request("http://localhost:3000/enqueue/q", { + method: "POST", + body: JSON.stringify({ payload: 0 }), + headers: authHeaders, + })); + + const first = await handler(new Request("http://localhost:3000/dequeue/q", { + headers: authHeaders, + })); + assertEquals(first.status, 200); + assertEquals(first.headers.get("content-type"), "application/json"); + assertEquals(await first.json(), "0"); + + const second = await handler(new Request("http://localhost:3000/dequeue/q", { + headers: authHeaders, + })); + assertEquals(second.status, 200); + assertEquals(second.headers.get("content-type"), "application/json"); + assertEquals(await second.json(), 0); +}); + +Deno.test("peek returns JSON for a string payload", async () => { + const mgr = new QueueManager(new Persistency.MemoryStore); + const handler = createHandler(mgr, API_TOKEN); + + await handler(new Request("http://localhost:3000/enqueue/q", { + method: "POST", + body: JSON.stringify({ payload: "hello" }), + headers: authHeaders, + })); + + const response = await handler(new Request("http://localhost:3000/peek/q", { + headers: authHeaders, + })); + assertEquals(response.status, 200); + assertEquals(response.headers.get("content-type"), "application/json"); + assertEquals(await response.json(), "hello"); +}); diff --git a/tests/manager_test.ts b/tests/manager_test.ts index 2e5beb0..2e16b8a 100644 --- a/tests/manager_test.ts +++ b/tests/manager_test.ts @@ -194,7 +194,9 @@ Deno.test("manager persistency", () => { const mgr = new QueueManager(persist); mgr.load(); - assertEquals([], persist.loadState()); + const snapshot = persist.loadState(); + assertEquals(snapshot.length, 2); + assertEquals(snapshot.every((event) => event.enqueue === true), true); assertEquals(1, mgr.length("foo")); assertEquals("bar", mgr.dequeue("foo")); assertEquals(1, mgr.length("fee")); @@ -212,7 +214,10 @@ Deno.test("json persistency", () => { mgr.load(); - assertEquals([], persist.loadState()); + const snapshot = persist.loadState(); + assertEquals(snapshot.length, 1); + assertEquals(snapshot[0].payload, payload); + assertEquals(snapshot[0].enqueue, true); assertEquals(1, mgr.length("foo")); assertEquals(payload, mgr.dequeue("foo")); persist.close(); @@ -577,3 +582,85 @@ Deno.test("manager load skips events with neither enqueue nor dequeue", () => { assertEquals(mgr.length("q"), 1); assertEquals(mgr.dequeue("q"), "a"); }); + +Deno.test("enqueue throws when queue count limit reached", () => { + const mgr = new QueueManager(new Persistency.MemoryStore(), 10, 1); + mgr.enqueue("a", "1"); + assertThrows(() => mgr.enqueue("b", "2"), Error, "Queue count limit reached"); + assertEquals(mgr.listQueues(), ["a"]); + assertEquals(mgr.dequeue("a"), "1"); +}); + +Deno.test("enqueue on an existing queue is allowed at the count limit", () => { + const mgr = new QueueManager(new Persistency.MemoryStore(), 10, 1); + mgr.enqueue("a", "1"); + mgr.enqueue("a", "2"); + assertEquals(mgr.length("a"), 2); + assertEquals(mgr.dequeue("a"), "1"); + assertEquals(mgr.dequeue("a"), "2"); +}); + +Deno.test("canEnqueue false means enqueue does not create a queue", () => { + const mgr = new QueueManager(new Persistency.MemoryStore(), 10, 1); + mgr.enqueue("a", "1"); + assertEquals(mgr.canEnqueue("b"), false); + assertThrows(() => mgr.enqueue("b", "2"), Error, "Queue count limit reached"); + assertEquals(mgr.listQueues(), ["a"]); +}); + +Deno.test("depth limit 0 refuses enqueue without leaking an empty queue", () => { + const mgr = new QueueManager(new Persistency.MemoryStore(), 0, 2); + assertEquals(mgr.canEnqueue("q"), false); + assertThrows(() => mgr.enqueue("q", "x"), Error, "Queue depth limit reached"); + assertEquals(mgr.listQueues(), []); + assertEquals(mgr.length("q"), 0); +}); + +Deno.test("load snapshots remaining items so a second load restores them", () => { + const persist = new Persistency.MemoryStore(); + const mgr = new QueueManager(persist); + mgr.enqueue("jobs", "keep-me"); + + const boot = new Persistency.MemoryStore(); + for (const ev of persist.loadState()) { + boot.saveEvent(ev.queue, ev.payload, ev.enqueue); + } + const loaded = new QueueManager(boot); + loaded.load(); + assertEquals(loaded.peek("jobs"), "keep-me"); + + const boot2 = new Persistency.MemoryStore(); + for (const ev of boot.loadState()) { + boot2.saveEvent(ev.queue, ev.payload, ev.enqueue); + } + const loaded2 = new QueueManager(boot2); + loaded2.load(); + assertEquals(loaded2.dequeue("jobs"), "keep-me"); +}); + +Deno.test("FileStore load snapshots remaining items onto disk", () => { + const tmp = Deno.makeTempDirSync(); + try { + const store = new Persistency.FileStore(); + store.dir(tmp); + const mgr = new QueueManager(store); + mgr.enqueue("jobs", "keep-me"); + store.close(); + + const store2 = new Persistency.FileStore(); + store2.dir(tmp); + const loaded = new QueueManager(store2); + loaded.load(); + assertEquals(loaded.peek("jobs"), "keep-me"); + store2.close(); + + const store3 = new Persistency.FileStore(); + store3.dir(tmp); + const loaded2 = new QueueManager(store3); + loaded2.load(); + assertEquals(loaded2.dequeue("jobs"), "keep-me"); + store3.close(); + } finally { + Deno.removeSync(tmp, { recursive: true }); + } +});