Lucas WintherandClaude Opus 5.5
7f33169487
Adds aclRepo to @stc/db for canView. Co-Authored-By: Claude Opus 5.5 <[email protected]>
62 lines
2.4 KiB
TypeScript
62 lines
2.4 KiB
TypeScript
import { expect } from "bun:test";
|
|
import { forEachDb } from "../../db/test/helpers";
|
|
import { jobsRepo } from "@stc/db";
|
|
import { createJobQueue } from "../src";
|
|
|
|
const until = async (f: () => Promise<boolean>) => {
|
|
for (let i = 0; i < 200; i++) { if (await f()) return; await Bun.sleep(10); }
|
|
throw new Error("timeout");
|
|
};
|
|
|
|
forEachDb("job runs to done with result and persisted progress", async (db) => {
|
|
const q = createJobQueue(db, {
|
|
ok: async (p, jc) => { await jc.reportProgress(0.5, "half"); return { echoed: p }; },
|
|
}, { pollMs: 10 });
|
|
q.start();
|
|
try {
|
|
const id = await q.enqueue("ok", { a: 1 });
|
|
await until(async () => (await q.get(id))?.status === "done");
|
|
const j = (await q.get(id))!;
|
|
expect(j.result).toEqual({ echoed: { a: 1 } });
|
|
expect(j.message).toBe("half");
|
|
expect(j.progress).toBe(1);
|
|
} finally { await q.stop(); }
|
|
});
|
|
|
|
forEachDb("throwing handler gives failed with message", async (db) => {
|
|
const q = createJobQueue(db, { bad: async () => { throw new Error("boom"); } }, { pollMs: 10 });
|
|
q.start();
|
|
try {
|
|
const id = await q.enqueue("bad", null);
|
|
await until(async () => (await q.get(id))?.status === "failed");
|
|
expect((await q.get(id))!.error).toBe("boom");
|
|
} finally { await q.stop(); }
|
|
});
|
|
|
|
forEachDb("progress updates persist while running", async (db) => {
|
|
let release!: () => void;
|
|
const gate = new Promise<void>((r) => (release = r));
|
|
const q = createJobQueue(db, { slow: async (_p, jc) => { await jc.reportProgress(0.3, "working"); await gate; return 1; } }, { pollMs: 10 });
|
|
q.start();
|
|
try {
|
|
const id = await q.enqueue("slow", null);
|
|
await until(async () => (await q.get(id))?.progress === 0.3);
|
|
expect((await q.get(id))!.status).toBe("running");
|
|
expect((await q.get(id))!.message).toBe("working");
|
|
release();
|
|
await until(async () => (await q.get(id))?.status === "done");
|
|
} finally { release(); await q.stop(); }
|
|
});
|
|
|
|
forEachDb("stale running job is requeued at start", async (db) => {
|
|
const id = await jobsRepo(db).insert("ok", { z: 1 });
|
|
await jobsRepo(db).claimNext(["ok"]);
|
|
expect((await jobsRepo(db).get(id))!.status).toBe("running");
|
|
const q = createJobQueue(db, { ok: async () => "again" }, { pollMs: 10 });
|
|
q.start();
|
|
try {
|
|
await until(async () => (await q.get(id))?.status === "done");
|
|
expect((await q.get(id))!.result).toBe("again");
|
|
} finally { await q.stop(); }
|
|
});
|