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-qa6-{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 6\"\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_deferred_fake_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")) {
globalThis.__qaSelectStarted();
return globalThis.__qaSelectPromise;
}
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_round6_one_shot_and_loop_option_surfaces_remain_strictly_separate() {
let fixture = Fixture::new("separate-option-surfaces");
assert_success(&fixture.build(), "build option-surface fixture");
let node = fixture.run_node(
"separate-option-surfaces",
r#"import { startQueueWorker, workQueueOnce } from "./server/handler.js";
async function onceCode(options) {
try { await workQueueOnce(options); } catch (error) { return error?.code ?? error?.name; }
return null;
}
const onceCases = [
["setTimeout", { queue: "Probe", setTimeout() {} }],
["clearTimeout", { queue: "Probe", clearTimeout() {} }],
["pollIntervalMs", { queue: "Probe", pollIntervalMs: 5 }],
["onError", { queue: "Probe", onError() {} }],
["extra", { queue: "Probe", runAt: "2028-02-29T12:00:00Z" }],
];
for (const [label, options] of onceCases) {
const code = await onceCode(options);
if (code !== "QUEUE_CLOCK_INVALID") throw new Error(`workQueueOnce accepted loop-only ${label}: ${code}`);
}
let timerCalls = 0;
let errorCalls = 0;
for (const [label, options] of [
["enqueue runAt", { queue: "Probe", runAt: "2028-02-29T12:00:00Z" }],
["undeclared field", { queue: "Probe", authority: true }],
]) {
let code = null;
try {
startQueueWorker({
...options,
setTimeout() { timerCalls += 1; },
clearTimeout() {},
onError() { errorCalls += 1; },
});
} catch (error) { code = error?.code; }
if (code !== "QUEUE_CLOCK_INVALID") throw new Error(`startQueueWorker accepted ${label}: ${code}`);
}
if (timerCalls !== 0 || errorCalls !== 0) throw new Error(`invalid option acquired ownership: timers=${timerCalls}, errors=${errorCalls}`);
"#,
);
assert_success(&node, "keep worker API option sets disjoint");
}
#[test]
fn qa_round6_startup_validates_every_operational_value_synchronously() {
let fixture = Fixture::new("synchronous-validation");
assert_success(&fixture.build(), "build synchronous-validation fixture");
let node = fixture.run_node(
"synchronous-validation",
r#"import { startQueueWorker } from "./server/handler.js";
const quiet = () => {};
const timer = () => 1;
const clear = () => {};
const cases = [
["numeric queue", { queue: 7, onError: quiet }],
["object worker", { queue: "Probe", worker: {}, onError: quiet }],
["impossible clock", { queue: "Probe", now: "2026-02-31T12:00:00Z", onError: quiet }],
["Date impostor", { queue: "Probe", now: Object.create(new Date("2028-02-29T12:00:00Z")), onError: quiet }],
["set hook", { setTimeout: 1, clearTimeout: clear, onError: quiet }],
["clear hook", { setTimeout: timer, clearTimeout: {}, onError: quiet }],
["error hook", { onError: "quiet" }],
["zero poll", { pollIntervalMs: 0, onError: quiet }],
["negative poll", { pollIntervalMs: -1, onError: quiet }],
["fractional poll", { pollIntervalMs: 1.5, onError: quiet }],
["infinite poll", { pollIntervalMs: Infinity, onError: quiet }],
["string poll", { pollIntervalMs: "7", onError: quiet }],
];
for (const [label, options] of cases) {
let returned = false;
let code = null;
try { startQueueWorker(options); returned = true; } catch (error) { code = error?.code; }
if (returned || code !== "QUEUE_CLOCK_INVALID") throw new Error(`${label} was not synchronously refused: returned=${returned}, code=${code}`);
}
"#,
);
assert_success(
&node,
"validate identities, clock, hooks, and polling before ownership",
);
}
#[test]
fn qa_round6_async_on_error_can_self_stop_and_stop_remains_idempotent() {
let fixture = Fixture::new("async-self-stop");
assert_success(&fixture.build(), "build async self-stop fixture");
let node = fixture.run_node(
"async-self-stop",
r#"import { startQueueWorker } from "./server/handler.js";
const timers = new Map();
let nextTimer = 0;
let worker;
let selfStopSettled = false;
worker = startQueueWorker({
pollIntervalMs: 13,
setTimeout(callback, delay) { const id = ++nextTimer; timers.set(id, { callback, delay }); return id; },
clearTimeout(id) { timers.delete(id); },
async onError(error) {
if (error?.code !== "QUEUE_DATABASE_URL_REQUIRED") throw new Error(`unexpected error ${error?.code}`);
await worker.stop();
await worker.stop();
selfStopSettled = true;
},
});
for (let turn = 0; turn < 12 && !selfStopSettled; turn += 1) await new Promise((resolve) => setImmediate(resolve));
if (!selfStopSettled) throw new Error("async onError self-stop did not settle");
if (timers.size !== 0) throw new Error(`self-stop leaked ${timers.size} timer(s)`);
await Promise.all([worker.stop(), worker.stop(), worker.stop()]);
"#,
);
assert_success(&node, "self-stop without a promise cycle or timer leak");
}
#[test]
fn qa_round6_external_stop_joins_an_in_flight_database_attempt_without_repolling() {
let fixture = Fixture::new("stop-in-flight-attempt");
assert_success(&fixture.build(), "build in-flight-attempt fixture");
fixture.install_deferred_fake_postgres();
let node = fixture.run_node_with_database(
"stop-in-flight-attempt",
r#"let resolveSelect;
let selectStarted = false;
globalThis.__qaSelectPromise = new Promise((resolve) => { resolveSelect = resolve; });
globalThis.__qaSelectStarted = () => { selectStarted = true; };
const { startQueueWorker, closeQueueDatabase } = await import("./server/handler.js");
const timers = new Map();
let nextTimer = 0;
const worker = startQueueWorker({
queue: "Probe",
worker: "qa",
now: "2028-02-29T12:00:00Z",
pollIntervalMs: 19,
setTimeout(callback, delay) { const id = ++nextTimer; timers.set(id, { callback, delay }); return id; },
clearTimeout(id) { timers.delete(id); },
onError(error) { throw error; },
});
for (let turn = 0; turn < 12 && !selectStarted; turn += 1) await new Promise((resolve) => setImmediate(resolve));
if (!selectStarted) throw new Error("worker never reached the deferred claim");
let stopped = false;
const firstStop = worker.stop().then(() => { stopped = true; });
await new Promise((resolve) => setImmediate(resolve));
if (stopped) throw new Error("stop returned before the in-flight attempt settled");
const secondStop = worker.stop();
resolveSelect([]);
await Promise.all([firstStop, secondStop]);
if (timers.size !== 0) throw new Error(`in-flight stop scheduled ${timers.size} replacement timer(s)`);
await worker.stop();
await closeQueueDatabase();
"#,
);
assert_success(
&node,
"join an in-flight claim and suppress replacement polling",
);
}
#[test]
fn qa_round6_external_stop_during_error_notification_does_not_wait_or_repoll() {
let fixture = Fixture::new("stop-during-notification");
assert_success(&fixture.build(), "build notification-race fixture");
let node = fixture.run_node(
"stop-during-notification",
r#"import { startQueueWorker } from "./server/handler.js";
let resolveNotification;
let notificationStarted = false;
const notification = new Promise((resolve) => { resolveNotification = resolve; });
const timers = new Map();
let nextTimer = 0;
const worker = startQueueWorker({
pollIntervalMs: 23,
setTimeout(callback, delay) { const id = ++nextTimer; timers.set(id, { callback, delay }); return id; },
clearTimeout(id) { timers.delete(id); },
onError(error) {
if (error?.code !== "QUEUE_DATABASE_URL_REQUIRED") throw new Error(`unexpected error ${error?.code}`);
notificationStarted = true;
return notification;
},
});
for (let turn = 0; turn < 12 && !notificationStarted; turn += 1) await new Promise((resolve) => setImmediate(resolve));
if (!notificationStarted) throw new Error("error notification never started");
let stopped = false;
await Promise.race([
worker.stop().then(() => { stopped = true; }),
new Promise((_, reject) => setTimeout(() => reject(new Error("stop waited for error notification")), 100)),
]);
if (!stopped) throw new Error("external stop did not settle");
resolveNotification();
for (let turn = 0; turn < 4; turn += 1) await new Promise((resolve) => setImmediate(resolve));
if (timers.size !== 0) throw new Error(`settled notification scheduled ${timers.size} replacement timer(s)`);
await worker.stop();
"#,
);
assert_success(
&node,
"stop independently of an in-flight error notification",
);
}
#[test]
fn qa_round6_thrown_and_rejected_error_hooks_do_not_leak_timers_or_rejections() {
let fixture = Fixture::new("error-hook-failures");
assert_success(&fixture.build(), "build error-hook failure fixture");
let node = fixture.run_node(
"error-hook-failures",
r#"import { startQueueWorker } from "./server/handler.js";
const unhandled = [];
process.on("unhandledRejection", (reason) => { unhandled.push(reason); });
function harness(onError) {
const timers = new Map();
let nextTimer = 0;
const worker = startQueueWorker({
pollIntervalMs: 29,
setTimeout(callback, delay) { const id = ++nextTimer; timers.set(id, { callback, delay }); return id; },
clearTimeout(id) { timers.delete(id); },
onError,
});
return { worker, timers };
}
const thrown = harness(() => { throw new Error("sync notification failure"); });
const rejected = harness(async () => { throw new Error("async notification failure"); });
for (let turn = 0; turn < 12 && (thrown.timers.size === 0 || rejected.timers.size === 0); turn += 1) {
await new Promise((resolve) => setImmediate(resolve));
}
if (thrown.timers.size !== 1 || rejected.timers.size !== 1) throw new Error(`error hooks did not leave exactly one poll each: ${thrown.timers.size}/${rejected.timers.size}`);
await Promise.all([thrown.worker.stop(), rejected.worker.stop()]);
for (let turn = 0; turn < 3; turn += 1) await new Promise((resolve) => setImmediate(resolve));
if (thrown.timers.size !== 0 || rejected.timers.size !== 0) throw new Error(`error hook stop leaked timers: ${thrown.timers.size}/${rejected.timers.size}`);
if (unhandled.length !== 0) throw new Error(`error hooks leaked ${unhandled.length} unhandled rejection(s)`);
"#,
);
assert_success(
&node,
"contain synchronous and asynchronous notification failures",
);
}
#[test]
fn qa_round6_thrown_and_rejected_timer_hooks_do_not_leak_unhandled_rejections() {
let fixture = Fixture::new("throwing-timer-hook");
assert_success(&fixture.build(), "build throwing-timer fixture");
let node = fixture.run_node(
"throwing-timer-hook",
r#"import { startQueueWorker } from "./server/handler.js";
const unhandled = [];
process.on("unhandledRejection", (reason) => { unhandled.push(reason); });
let throwingCalls = 0;
let rejectingCalls = 0;
const throwingWorker = startQueueWorker({
pollIntervalMs: 31,
setTimeout() { throwingCalls += 1; throw new Error("timer hook threw"); },
clearTimeout() {},
onError() {},
});
const rejectingWorker = startQueueWorker({
pollIntervalMs: 31,
setTimeout() { rejectingCalls += 1; return Promise.reject(new Error("timer hook rejected")); },
clearTimeout() {},
onError() {},
});
const clearTimers = new Map();
let nextTimer = 0;
const rejectingClearWorker = startQueueWorker({
pollIntervalMs: 31,
setTimeout(callback, delay) { const id = ++nextTimer; clearTimers.set(id, { callback, delay }); return id; },
clearTimeout(id) { clearTimers.delete(id); return Promise.reject(new Error("clear hook rejected")); },
onError() {},
});
for (let turn = 0; turn < 12 && (throwingCalls === 0 || rejectingCalls === 0 || clearTimers.size === 0); turn += 1) await new Promise((resolve) => setImmediate(resolve));
await Promise.all([throwingWorker.stop(), rejectingWorker.stop(), rejectingClearWorker.stop()]);
for (let turn = 0; turn < 4; turn += 1) await new Promise((resolve) => setImmediate(resolve));
if (throwingCalls !== 1 || rejectingCalls !== 1) throw new Error(`timer hook calls changed: ${throwingCalls}/${rejectingCalls}`);
if (clearTimers.size !== 0) throw new Error(`rejecting clear hook leaked ${clearTimers.size} timer(s)`);
if (unhandled.length !== 0) throw new Error(`timer hooks leaked unhandled rejections: ${unhandled.map((reason) => reason?.message).join(", ")}`);
"#,
);
assert_success(&node, "contain a throwing poll scheduler hook");
}
#[test]
fn qa_round6_repeated_stop_clears_a_scheduled_timer_only_once() {
let fixture = Fixture::new("repeated-stop");
assert_success(&fixture.build(), "build repeated-stop fixture");
let node = fixture.run_node(
"repeated-stop",
r#"import { startQueueWorker } from "./server/handler.js";
const timers = new Map();
let nextTimer = 0;
let clearCalls = 0;
const worker = startQueueWorker({
pollIntervalMs: 37,
setTimeout(callback, delay) { const id = ++nextTimer; timers.set(id, { callback, delay }); return id; },
clearTimeout(id) { clearCalls += 1; timers.delete(id); },
onError() {},
});
for (let turn = 0; turn < 12 && timers.size === 0; turn += 1) await new Promise((resolve) => setImmediate(resolve));
if (timers.size !== 1) throw new Error(`worker did not schedule exactly one poll: ${timers.size}`);
await Promise.all([worker.stop(), worker.stop(), worker.stop()]);
await worker.stop();
if (timers.size !== 0) throw new Error(`repeated stop leaked ${timers.size} timer(s)`);
if (clearCalls !== 1) throw new Error(`idempotent stop invoked clearTimeout ${clearCalls} times`);
"#,
);
assert_success(&node, "make repeated stop externally idempotent");
}