use std::fs;
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::thread;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
struct ComposePostgres {
file: PathBuf,
project: String,
}
impl ComposePostgres {
fn start() -> Option<Self> {
if !Command::new("docker")
.arg("info")
.stdout(Stdio::null())
.stderr(Stdio::null())
.status()
.is_ok_and(|status| status.success())
{
eprintln!(
"WO-24 Postgres queue integration: SKIP (Docker daemon unavailable; concurrent claim, retry, dead-letter, and row-drift paths were not exercised)"
);
return None;
}
if !Command::new("docker")
.args(["compose", "version"])
.stdout(Stdio::null())
.stderr(Stdio::null())
.status()
.is_ok_and(|status| status.success())
{
eprintln!(
"WO-24 Postgres queue integration: SKIP (Docker Compose unavailable; concurrent claim, retry, dead-letter, and row-drift paths were not exercised)"
);
return None;
}
let nonce = SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("clock after epoch")
.as_nanos();
let postgres = Self {
file: Path::new(env!("CARGO_MANIFEST_DIR"))
.join("tests/fixtures/wo19-postgres/compose.yaml"),
project: format!("noxidwo24{}{}", std::process::id(), nonce),
};
let output = postgres
.command()
.args(["up", "--detach"])
.output()
.expect("start queue Postgres");
assert!(
output.status.success(),
"Docker was available but queue Postgres failed to start: {}",
String::from_utf8_lossy(&output.stderr)
);
for _ in 0..480 {
if postgres
.command()
.args([
"exec",
"--no-TTY",
"postgres",
"pg_isready",
"-U",
"noxid_test",
"-d",
"noxid_test",
])
.stdout(Stdio::null())
.stderr(Stdio::null())
.status()
.expect("probe queue Postgres")
.success()
{
return Some(postgres);
}
thread::sleep(Duration::from_millis(250));
}
panic!("queue Postgres did not become ready within 120 seconds");
}
fn command(&self) -> Command {
let mut command = Command::new("docker");
command
.args(["compose", "-f"])
.arg(&self.file)
.args(["--project-name", &self.project]);
command
}
fn database_url(&self) -> String {
let output = self
.command()
.args(["port", "postgres", "5432"])
.output()
.expect("resolve queue Postgres port");
assert!(output.status.success());
let mapping = String::from_utf8(output.stdout).expect("UTF-8 port mapping");
let port = mapping
.trim()
.rsplit_once(':')
.map(|(_, port)| port)
.expect("mapped port");
format!("postgres://noxid_test:noxid_test@127.0.0.1:{port}/noxid_test")
}
}
impl Drop for ComposePostgres {
fn drop(&mut self) {
let _ = self
.command()
.args(["down", "--volumes", "--remove-orphans"])
.stdout(Stdio::null())
.stderr(Stdio::null())
.status();
}
}
#[test]
fn postgres_queue_claim_retry_dead_letter_and_drift_execute() {
let Some(postgres) = ComposePostgres::start() else {
return;
};
let root = std::env::temp_dir().join(format!("noxid-wo24-postgres-{}", std::process::id()));
let _ = fs::remove_dir_all(&root);
fs::create_dir_all(root.join("server/queues")).expect("create queue project");
fs::write(
root.join("Noxid.toml"),
"[app]\ntitle = \"Queue postgres\"\n",
)
.expect("write config");
fs::write(root.join("package.json"), "{\"type\":\"module\"}\n").expect("write package");
fs::write(
root.join("server/queues/Work.nox"),
"queue Work { payload { label: String failUntil: Int } retry: 2 backoff: 1s }\n",
)
.expect("write retry queue");
fs::write(
root.join("server/queues/AlwaysFail.nox"),
"queue AlwaysFail { payload { label: String } retry: 1 backoff: 1s }\n",
)
.expect("write dead-letter queue");
fs::write(
root.join("server/host.js"),
r#"export const queues = Object.freeze({
"queue:Work": async (payload, context) => {
if (context.attempts <= payload.failUntil) throw new Error(`retry-${context.attempts}`);
return payload.label;
},
"queue:AlwaysFail": async () => { throw new Error("always"); },
});
"#,
)
.expect("write queue host");
let build = Command::new(env!("CARGO_BIN_EXE_noxid"))
.args(["build", ".", "--out-dir", "dist"])
.current_dir(&root)
.output()
.expect("build queue project");
assert!(
build.status.success(),
"{}",
String::from_utf8_lossy(&build.stderr)
);
#[cfg(unix)]
{
let repository_modules = Path::new(env!("CARGO_MANIFEST_DIR")).join("../../node_modules");
if !repository_modules.join("postgres").exists() {
eprintln!(
"WO-24 Postgres queue integration: SKIP (workspace postgres Node driver unavailable)"
);
let _ = fs::remove_dir_all(&root);
return;
}
std::os::unix::fs::symlink(&repository_modules, root.join("node_modules"))
.expect("link admitted postgres driver");
}
fs::write(
root.join("dist/exercise.mjs"),
r#"import postgres from "postgres";
import { enqueue, workQueueOnce, queueStatus, closeQueueDatabase } from "./server/handler.js";
const first = await enqueue("Work", { label: "a", failUntil: 0 }, { runAt: "2026-08-29T12:00:00Z" });
const second = await enqueue("Work", { label: "b", failUntil: 0 }, { runAt: "2026-08-29T12:00:00Z" });
const concurrent = await Promise.all([
workQueueOnce({ queue: "Work", worker: "left", now: "2026-08-29T12:00:00Z" }),
workQueueOnce({ queue: "Work", worker: "right", now: "2026-08-29T12:00:00Z" }),
]);
if (new Set(concurrent.map((job) => job?.id)).size !== 2 || !concurrent.every((job) => job.state === "completed")) throw new Error(`concurrent claim failed: ${JSON.stringify(concurrent)}`);
const enqueued = new Set([first.id, second.id]);
if (!concurrent.every((job) => enqueued.has(job.id))) throw new Error("workers claimed an unknown job");
await enqueue("Work", { label: "retry", failUntil: 2 }, { runAt: "2026-08-29T13:00:00Z" });
let job = await workQueueOnce({ queue: "Work", worker: "retry", now: "2026-08-29T13:00:00Z" });
if (job.state !== "pending" || job.attempts !== 1 || job.runAt !== "2026-08-29T13:00:01.000Z") throw new Error(`first retry changed: ${JSON.stringify(job)}`);
if (await workQueueOnce({ queue: "Work", worker: "early", now: "2026-08-29T13:00:00.999Z" }) !== null) throw new Error("backoff was not honored");
job = await workQueueOnce({ queue: "Work", worker: "retry", now: "2026-08-29T13:00:01Z" });
if (job.state !== "pending" || job.attempts !== 2 || job.runAt !== "2026-08-29T13:00:02.000Z") throw new Error(`second retry changed: ${JSON.stringify(job)}`);
job = await workQueueOnce({ queue: "Work", worker: "retry", now: "2026-08-29T13:00:02Z" });
if (job.state !== "completed" || job.attempts !== 3 || job.value !== "retry") throw new Error(`retry completion changed: ${JSON.stringify(job)}`);
await enqueue("AlwaysFail", { label: "dead" }, { runAt: "2026-08-29T14:00:00Z" });
job = await workQueueOnce({ queue: "AlwaysFail", now: "2026-08-29T14:00:00Z" });
if (job.state !== "pending" || job.attempts !== 1) throw new Error(`dead retry changed: ${JSON.stringify(job)}`);
job = await workQueueOnce({ queue: "AlwaysFail", now: "2026-08-29T14:00:01Z" });
if (job.state !== "dead-letter" || job.attempts !== 2) throw new Error(`dead letter changed: ${JSON.stringify(job)}`);
const sql = postgres(process.env.DATABASE_URL, { max: 1 });
await sql`INSERT INTO _noxid_jobs (id, queue, payload, state, attempts, run_at) VALUES ('drift', 'Work', ${{ label: "drift" }}, 'pending', 0, ${new Date("2026-08-29T15:00:00Z")})`;
let drift = null;
try { await workQueueOnce({ queue: "Work", now: "2026-08-29T15:00:00Z" }); } catch (error) { drift = error.code; }
if (drift !== "QUEUE_PAYLOAD_DRIFT") throw new Error(`row drift was not refused: ${drift}`);
const rows = await sql`SELECT state FROM _noxid_jobs WHERE id = 'drift'`;
if (rows[0]?.state !== "dead-letter") throw new Error(`drift did not dead-letter: ${JSON.stringify(rows)}`);
const status = await queueStatus("AlwaysFail");
if (!status.some((entry) => entry.state === "dead-letter" && entry.count === 1)) throw new Error(`status omitted dead letter: ${JSON.stringify(status)}`);
await sql.end({ timeout: 1 });
await closeQueueDatabase();
"#,
)
.expect("write queue exercise");
let node = Command::new("node")
.arg("exercise.mjs")
.current_dir(root.join("dist"))
.env("DATABASE_URL", postgres.database_url())
.output()
.expect("run Postgres queue contract");
assert!(
node.status.success(),
"stdout:\n{}\nstderr:\n{}",
String::from_utf8_lossy(&node.stdout),
String::from_utf8_lossy(&node.stderr)
);
let _ = fs::remove_dir_all(root);
}