use std::fs;
use std::path::PathBuf;
use std::process::{Command, Output};
use std::sync::atomic::{AtomicU64, Ordering};
static NEXT_FIXTURE: AtomicU64 = AtomicU64::new(0);
struct Fixture {
root: PathBuf,
}
impl Fixture {
fn new(label: &str) -> Self {
let ordinal = NEXT_FIXTURE.fetch_add(1, Ordering::Relaxed);
let root = std::env::temp_dir().join(format!(
"noxid-wo24-qa7-{label}-{}-{ordinal}",
std::process::id()
));
fs::create_dir_all(root.join("server/queues")).expect("create queue fixture");
fs::write(
root.join("Noxid.toml"),
"[app]\ntitle = \"WO-24 QA round 7\"\n",
)
.expect("write project manifest");
fs::write(root.join("package.json"), "{\"type\":\"module\"}\n").expect("write ESM marker");
fs::write(
root.join("server/host.js"),
"export const queues = Object.freeze({});\n",
)
.expect("write inert host");
fs::write(
root.join("server/queues/Probe.nox"),
"queue Probe { payload {} retry: 0 backoff: 1s }\n",
)
.expect("write probe queue");
Self { root }
}
fn write(&self, relative: &str, contents: &str) {
let path = self.root.join(relative);
if let Some(parent) = path.parent() {
fs::create_dir_all(parent).expect("create fixture parent");
}
fs::write(path, contents).expect("write fixture file");
}
fn build(&self) -> Output {
Command::new(env!("CARGO_BIN_EXE_noxid"))
.args(["build", ".", "--out-dir", "dist"])
.current_dir(&self.root)
.output()
.expect("build queue fixture")
}
fn run_node(&self, name: &str, source: &str) -> Output {
self.write(&format!("dist/{name}.mjs"), source);
Command::new("node")
.arg(format!("{name}.mjs"))
.current_dir(self.root.join("dist"))
.env_remove("DATABASE_URL")
.output()
.expect("execute generated queue runtime")
}
fn run_node_with_database(&self, name: &str, source: &str) -> Output {
self.write(&format!("dist/{name}.mjs"), source);
Command::new("node")
.arg(format!("{name}.mjs"))
.current_dir(self.root.join("dist"))
.env("DATABASE_URL", "postgres://qa.invalid/noxid")
.output()
.expect("execute generated queue runtime with fake database")
}
fn install_controlled_postgres(&self) {
self.write(
"dist/node_modules/postgres/package.json",
"{\"type\":\"module\",\"exports\":\"./index.js\"}\n",
);
self.write(
"dist/node_modules/postgres/index.js",
r#"export default function postgres() {
const sql = async (strings) => {
const query = strings.join("?");
if (query.includes("SELECT id, queue, payload")) return globalThis.__qaSelect();
return [];
};
sql.unsafe = async () => [];
sql.begin = async (callback) => callback(sql);
sql.end = async () => {};
sql.json = (value) => value;
return sql;
}
"#,
);
}
}
impl Drop for Fixture {
fn drop(&mut self) {
let _ = fs::remove_dir_all(&self.root);
}
}
fn output_text(output: &Output) -> String {
format!(
"stdout:\n{}\nstderr:\n{}",
String::from_utf8_lossy(&output.stdout),
String::from_utf8_lossy(&output.stderr)
)
}
fn assert_success(output: &Output, context: &str) {
assert!(
output.status.success(),
"{context}:\n{}",
output_text(output)
);
}
#[test]
fn qa_round7_falsy_object_and_thenable_timer_handles_are_owned_exactly_once() {
let fixture = Fixture::new("timer-handle-identities");
assert_success(&fixture.build(), "build timer-handle fixture");
let node = fixture.run_node(
"timer-handle-identities",
r#"import { startQueueWorker } from "./server/handler.js";
const unhandled = [];
process.on("unhandledRejection", (reason) => { unhandled.push(reason); });
const objectHandle = Object.freeze({ kind: "object-handle" });
const thenableHandle = Object.freeze({
then(_resolve, reject) { queueMicrotask(() => reject(new Error("thenable timer handle rejected"))); },
});
const handles = [0, null, objectHandle, thenableHandle];
const clears = handles.map(() => []);
let scheduled = 0;
const workers = handles.map((handle, index) => startQueueWorker({
pollIntervalMs: 41,
setTimeout() { scheduled += 1; return handle; },
clearTimeout(observed) { clears[index].push(observed); },
onError() {},
}));
for (let turn = 0; turn < 16 && scheduled !== handles.length; turn += 1) await new Promise((resolve) => setImmediate(resolve));
if (scheduled !== handles.length) throw new Error(`only ${scheduled} timer handles were scheduled`);
await Promise.all(workers.flatMap((worker) => [worker.stop(), worker.stop(), worker.stop()]));
for (let turn = 0; turn < 4; turn += 1) await new Promise((resolve) => setImmediate(resolve));
for (let index = 0; index < handles.length; index += 1) {
if (clears[index].length !== 1 || clears[index][0] !== handles[index]) {
throw new Error(`handle ${index} was not cleared exactly once by identity: ${clears[index].length}`);
}
}
if (unhandled.length !== 0) throw new Error(`timer handles leaked ${unhandled.length} rejection(s)`);
"#,
);
assert_success(&node, "own falsy, object, and thenable timer handles");
}
#[test]
fn qa_round7_set_timeout_can_reentrantly_stop_its_owner_without_leaking_the_handle() {
let fixture = Fixture::new("set-hook-reentrant-stop");
assert_success(&fixture.build(), "build reentrant-stop fixture");
let node = fixture.run_node(
"set-hook-reentrant-stop",
r#"import { startQueueWorker } from "./server/handler.js";
const handle = Object.freeze({ id: "reentrant" });
const cleared = [];
let setCalls = 0;
let stopFromHook;
let worker;
worker = startQueueWorker({
pollIntervalMs: 43,
setTimeout() {
setCalls += 1;
stopFromHook = Promise.all([worker.stop(), worker.stop(), worker.stop()]);
return handle;
},
clearTimeout(observed) { cleared.push(observed); },
onError() {},
});
for (let turn = 0; turn < 16 && stopFromHook === undefined; turn += 1) await new Promise((resolve) => setImmediate(resolve));
if (stopFromHook === undefined) throw new Error("setTimeout hook never attempted reentrant stop");
await stopFromHook;
await worker.stop();
if (setCalls !== 1) throw new Error(`reentrant stop allowed ${setCalls} schedules`);
if (cleared.length !== 1 || cleared[0] !== handle) throw new Error(`reentrant schedule cleared ${cleared.length} handle(s)`);
"#,
);
assert_success(
&node,
"allow setTimeout to reentrantly dispose the worker owner",
);
}
#[test]
fn qa_round7_throwing_and_rejected_clear_hooks_remain_terminal_under_concurrent_stop() {
let fixture = Fixture::new("clear-hook-failures");
assert_success(&fixture.build(), "build clear-hook failure fixture");
let node = fixture.run_node(
"clear-hook-failures",
r#"import { startQueueWorker } from "./server/handler.js";
const unhandled = [];
process.on("unhandledRejection", (reason) => { unhandled.push(reason); });
function harness(clearTimeout) {
let scheduled = false;
let clears = 0;
const worker = startQueueWorker({
pollIntervalMs: 47,
setTimeout() { scheduled = true; return Promise.reject(new Error("rejected timer handle")); },
clearTimeout(handle) { clears += 1; return clearTimeout(handle); },
onError() {},
});
return { worker, isScheduled: () => scheduled, clearCount: () => clears };
}
const thrown = harness(() => { throw new Error("clear hook threw"); });
const rejected = harness(() => Promise.reject(new Error("clear hook rejected")));
for (let turn = 0; turn < 16 && (!thrown.isScheduled() || !rejected.isScheduled()); turn += 1) await new Promise((resolve) => setImmediate(resolve));
if (!thrown.isScheduled() || !rejected.isScheduled()) throw new Error("workers did not reach timer ownership");
await Promise.all([
thrown.worker.stop(), thrown.worker.stop(), thrown.worker.stop(),
rejected.worker.stop(), rejected.worker.stop(), rejected.worker.stop(),
]);
await Promise.all([thrown.worker.stop(), rejected.worker.stop()]);
for (let turn = 0; turn < 4; turn += 1) await new Promise((resolve) => setImmediate(resolve));
if (thrown.clearCount() !== 1 || rejected.clearCount() !== 1) throw new Error(`clear hooks repeated: ${thrown.clearCount()}/${rejected.clearCount()}`);
if (unhandled.length !== 0) throw new Error(`hook failures leaked ${unhandled.length} rejection(s)`);
"#,
);
assert_success(
&node,
"contain failing clear hooks across repeated concurrent stop",
);
}
#[test]
fn qa_round7_callbacks_fired_after_stop_are_inert_and_cannot_repoll() {
let fixture = Fixture::new("callback-after-stop");
assert_success(&fixture.build(), "build callback-after-stop fixture");
fixture.install_controlled_postgres();
let node = fixture.run_node_with_database(
"callback-after-stop",
r#"let selectCalls = 0;
globalThis.__qaSelect = async () => { selectCalls += 1; return []; };
const { startQueueWorker, closeQueueDatabase } = await import("./server/handler.js");
let callback;
let clearCalls = 0;
const worker = startQueueWorker({
queue: "Probe",
worker: "qa",
now: "2028-02-29T12:00:00Z",
pollIntervalMs: 53,
setTimeout(next) { callback = next; return 0; },
clearTimeout(handle) { if (handle !== 0) throw new Error(`wrong handle ${handle}`); clearCalls += 1; },
onError(error) { throw error; },
});
for (let turn = 0; turn < 16 && callback === undefined; turn += 1) await new Promise((resolve) => setImmediate(resolve));
if (callback === undefined || selectCalls !== 1) throw new Error(`worker did not reach its first timer: callback=${callback !== undefined}, selects=${selectCalls}`);
await Promise.all([worker.stop(), worker.stop(), worker.stop()]);
callback();
callback();
callback();
for (let turn = 0; turn < 4; turn += 1) await new Promise((resolve) => setImmediate(resolve));
if (selectCalls !== 1) throw new Error(`late callbacks started ${selectCalls - 1} claim(s)`);
if (clearCalls !== 1) throw new Error(`late callbacks changed clear count to ${clearCalls}`);
await worker.stop();
await closeQueueDatabase();
"#,
);
assert_success(&node, "ignore timer callbacks fired after owner stop");
}
#[test]
fn qa_round7_synchronously_fired_timer_is_consumed_before_stop_joins_the_attempt() {
let fixture = Fixture::new("synchronous-fire");
assert_success(&fixture.build(), "build synchronous-fire fixture");
fixture.install_controlled_postgres();
let node = fixture.run_node_with_database(
"synchronous-fire",
r#"const resolvers = new Map();
let selectCalls = 0;
globalThis.__qaSelect = () => {
selectCalls += 1;
const index = selectCalls;
return new Promise((resolve) => { resolvers.set(index, resolve); });
};
const { startQueueWorker, closeQueueDatabase } = await import("./server/handler.js");
const firedHandle = Object.freeze({ id: "already-fired" });
const cleared = [];
let setCalls = 0;
const worker = startQueueWorker({
queue: "Probe",
worker: "qa",
now: "2028-02-29T12:00:00Z",
pollIntervalMs: 59,
setTimeout(callback) {
setCalls += 1;
if (setCalls === 1) callback();
else throw new Error("stopped owner attempted a replacement schedule");
return firedHandle;
},
clearTimeout(handle) { cleared.push(handle); },
onError(error) { throw error; },
});
for (let turn = 0; turn < 16 && !resolvers.has(1); turn += 1) await new Promise((resolve) => setImmediate(resolve));
resolvers.get(1)([]);
for (let turn = 0; turn < 16 && !resolvers.has(2); turn += 1) await new Promise((resolve) => setImmediate(resolve));
if (!resolvers.has(2)) throw new Error(`synchronous timer did not begin its attempt: selects=${selectCalls}`);
let stopped = false;
const stopping = worker.stop().then(() => { stopped = true; });
await new Promise((resolve) => setImmediate(resolve));
if (stopped) throw new Error("stop did not join the synchronously-fired attempt");
resolvers.get(2)([]);
await stopping;
if (cleared.length !== 0) throw new Error(`stop cleared ${cleared.length} already-fired handle(s)`);
await closeQueueDatabase();
"#,
);
assert_success(
&node,
"consume a synchronously-fired timer before joining its claim",
);
}
#[test]
fn qa_round7_stale_callback_cannot_steal_a_successor_timers_ownership() {
let fixture = Fixture::new("stale-callback-successor");
assert_success(&fixture.build(), "build stale-callback fixture");
fixture.install_controlled_postgres();
let node = fixture.run_node_with_database(
"stale-callback-successor",
r#"const resolvers = new Map();
let selectCalls = 0;
globalThis.__qaSelect = () => {
selectCalls += 1;
const index = selectCalls;
return new Promise((resolve) => { resolvers.set(index, resolve); });
};
const { startQueueWorker, closeQueueDatabase } = await import("./server/handler.js");
const timers = new Map();
const cleared = [];
let nextHandle = 0;
const worker = startQueueWorker({
queue: "Probe",
worker: "qa",
now: "2028-02-29T12:00:00Z",
pollIntervalMs: 61,
setTimeout(callback) { const handle = ++nextHandle; timers.set(handle, callback); return handle; },
clearTimeout(handle) { cleared.push(handle); timers.delete(handle); },
onError(error) { throw error; },
});
for (let turn = 0; turn < 16 && !resolvers.has(1); turn += 1) await new Promise((resolve) => setImmediate(resolve));
resolvers.get(1)([]);
for (let turn = 0; turn < 16 && !timers.has(1); turn += 1) await new Promise((resolve) => setImmediate(resolve));
const stale = timers.get(1);
timers.delete(1);
stale();
for (let turn = 0; turn < 16 && !resolvers.has(2); turn += 1) await new Promise((resolve) => setImmediate(resolve));
resolvers.get(2)([]);
for (let turn = 0; turn < 16 && !timers.has(2); turn += 1) await new Promise((resolve) => setImmediate(resolve));
if (!timers.has(2)) throw new Error("successor timer was not scheduled");
stale();
for (let turn = 0; turn < 4; turn += 1) await new Promise((resolve) => setImmediate(resolve));
const stopping = worker.stop();
if (resolvers.has(3)) resolvers.get(3)([]);
await stopping;
if (selectCalls !== 2) throw new Error(`stale callback started claim ${selectCalls}`);
if (timers.size !== 0 || cleared.length !== 1 || cleared[0] !== 2) {
throw new Error(`successor ownership was lost: timers=${[...timers.keys()]}, cleared=${cleared}`);
}
await closeQueueDatabase();
"#,
);
assert_success(&node, "keep successor timer ownership from stale callbacks");
}
#[test]
fn qa_round7_duplicate_timer_fire_cannot_replace_the_attempt_stop_must_join() {
let fixture = Fixture::new("duplicate-fire-attempt-join");
assert_success(&fixture.build(), "build duplicate-fire fixture");
fixture.install_controlled_postgres();
let node = fixture.run_node_with_database(
"duplicate-fire-attempt-join",
r#"const resolvers = new Map();
let selectCalls = 0;
globalThis.__qaSelect = () => {
selectCalls += 1;
const index = selectCalls;
return new Promise((resolve) => { resolvers.set(index, resolve); });
};
const { startQueueWorker, closeQueueDatabase } = await import("./server/handler.js");
let callback;
const worker = startQueueWorker({
queue: "Probe",
worker: "qa",
now: "2028-02-29T12:00:00Z",
pollIntervalMs: 67,
setTimeout(next) { callback = next; return 0; },
clearTimeout() {},
onError(error) { throw error; },
});
for (let turn = 0; turn < 16 && !resolvers.has(1); turn += 1) await new Promise((resolve) => setImmediate(resolve));
resolvers.get(1)([]);
for (let turn = 0; turn < 16 && callback === undefined; turn += 1) await new Promise((resolve) => setImmediate(resolve));
callback();
callback();
for (let turn = 0; turn < 8 && !resolvers.has(3); turn += 1) await new Promise((resolve) => setImmediate(resolve));
let stopSettled = false;
const stopping = worker.stop().then(() => { stopSettled = true; });
if (resolvers.has(3)) {
resolvers.get(3)([]);
await new Promise((resolve) => setImmediate(resolve));
}
const settledAfterOnlyDuplicate = stopSettled;
resolvers.get(2)([]);
await stopping;
if (selectCalls !== 2) throw new Error(`one timer callback started ${selectCalls - 1} claims`);
if (settledAfterOnlyDuplicate) throw new Error("stop joined a replacement attempt but abandoned the first in-flight claim");
await closeQueueDatabase();
"#,
);
assert_success(
&node,
"make a timer callback one-shot and preserve attempt join ownership",
);
}