diff --git a/.github/actions/provision-gradle/action.yml b/.github/actions/provision-gradle/action.yml new file mode 100644 index 000000000..876b8c366 --- /dev/null +++ b/.github/actions/provision-gradle/action.yml @@ -0,0 +1,24 @@ +name: Provision Gradle +description: >- + Downloads the Gradle distribution the wrapper pins, retrying transient network + failures. The wrapper fetches it on first use and gives up on the first error, + so a reset connection to services.gradle.org fails the build before any code + has run. Doing it in its own step keeps that retry off the real invocation, + which would otherwise re-run the build to recover a download. + +runs: + using: composite + steps: + - name: Download the pinned Gradle distribution + working-directory: sdks/java + shell: bash + run: | + for attempt in 1 2 3; do + if ./gradlew --version --no-daemon; then + exit 0 + fi + echo "::warning::Gradle provisioning attempt ${attempt} failed; retrying" + sleep $((attempt * 10)) + done + echo "::error::could not download the Gradle distribution after 3 attempts" + exit 1 diff --git a/.github/workflows/ci-java.yml b/.github/workflows/ci-java.yml index 8fcf3d652..b94c93385 100644 --- a/.github/workflows/ci-java.yml +++ b/.github/workflows/ci-java.yml @@ -55,6 +55,9 @@ jobs: with: java-version: ${{ matrix.java }} + - name: Provision Gradle + uses: ./.github/actions/provision-gradle + # Stage the prebuilt .so where copyNative expects it, so no Rust is needed. - name: Download the cdylib uses: actions/download-artifact@v8 @@ -100,6 +103,9 @@ jobs: - name: Set up Java + Gradle uses: ./.github/actions/setup-java + - name: Provision Gradle + uses: ./.github/actions/provision-gradle + # bash so `./gradlew` runs uniformly on the Windows runner too. - name: Build and test the Java SDK working-directory: sdks/java @@ -128,6 +134,9 @@ jobs: distribution: graalvm github-token: ${{ secrets.GITHUB_TOKEN }} + - name: Provision Gradle + uses: ./.github/actions/provision-gradle + - name: Build and run the native image from shipped metadata working-directory: sdks/java run: | diff --git a/crates/taskito-node/src/executor.rs b/crates/taskito-node/src/executor.rs index 198972a22..47e00b816 100644 --- a/crates/taskito-node/src/executor.rs +++ b/crates/taskito-node/src/executor.rs @@ -17,7 +17,7 @@ use napi::threadsafe_function::ThreadsafeFunction; use napi_derive::napi; use taskito_core::worker::{ AttachAddress, CancelSignals, ExecutorClient, ExecutorConfig, ExecutorError, ExecutorHandle, - ExecutorSession, WorkerDispatcher, + ExecutorSession, ExecutorSideChannel, WorkerDispatcher, }; use crate::convert::{JsTaskInvocation, JsTaskOutcome}; @@ -60,6 +60,9 @@ pub struct JsExecutor { /// Cancels delivered as protocol frames. The JS side polls this instead of /// a storage flag, which a detached executor does not have. cancels: Arc, + /// Progress, task logs and toggles — the storage-shaped operations the + /// scheduler performs on this executor's behalf. + side_channel: ExecutorSideChannel, scheduler_id: String, executor_id: String, peer: String, @@ -100,6 +103,52 @@ impl JsExecutor { self.cancels.is_cancelled(&job_id) } + /// Whether the scheduler applies progress and task logs on our behalf. + /// + /// False against a scheduler with no storage configured for it, or one + /// built before the side-channel existed. The methods below are no-ops + /// either way; this exists so the shell can say so once rather than + /// silently dropping a task's progress bar. + #[napi] + pub fn supports_side_channel(&self) -> bool { + self.side_channel.is_supported() + } + + /// Report a running job's progress (0-100). + /// + /// An executor has no storage of its own, so this travels to the scheduler + /// instead. Fire-and-forget: it never blocks the calling task and never + /// fails its job. + #[napi] + pub fn report_progress(&self, job_id: String, progress: i32) { + self.side_channel.report_progress(&job_id, progress); + } + + /// Write one structured log line for a running job. A published partial is + /// this at level `result`, with the value as `extra`. + #[napi] + pub fn write_task_log( + &self, + job_id: String, + task_name: String, + level: String, + message: String, + extra: Option, + ) { + self.side_channel + .write_task_log(&job_id, &task_name, &level, &message, extra.as_deref()); + } + + /// Middleware the operator has disabled for a running job's task. + /// + /// Resolved by the scheduler at dispatch and carried on the job frame, so + /// a dashboard toggle is honoured without the settings read this process + /// has no storage to perform. + #[napi] + pub fn disabled_middleware(&self, job_id: String) -> Vec { + self.side_channel.disabled_middleware(&job_id) + } + /// Resolve once the scheduler ends the session — a `shutdown` frame, or the /// connection dropping. Does not drain; call `shutdown()` for that. #[napi] @@ -217,6 +266,7 @@ pub async fn start_executor( Ok(JsExecutor { executor_id: handle.executor_id().to_string(), session: handle.session(), + side_channel: handle.side_channel(), cancels, handle: Arc::new(Mutex::new(Some(handle))), scheduler_id, diff --git a/sdks/node/src/executor.ts b/sdks/node/src/executor.ts index 006d1fd78..742e076a7 100644 --- a/sdks/node/src/executor.ts +++ b/sdks/node/src/executor.ts @@ -43,7 +43,14 @@ export interface ExecutorStartParams { tasks: ReadonlyMap; serializer: Serializer; codecs?: ReadonlyMap; - middlewareFor: (taskName: string) => readonly Middleware[]; + /** + * The middleware chain for a task, minus `disabled`. + * + * Takes the disable list rather than reading one: an executor has no + * settings store, so the scheduler resolves it and attaches it to each + * dispatch. + */ + middlewareFor: (taskName: string, disabled: readonly string[]) => readonly Middleware[]; emitter: Emitter; resources: ResourceRuntime; run?: ExecutorRunOptions; @@ -87,20 +94,41 @@ export class Executor { // The executor does not exist yet, and the callback it needs must already // be able to reach it — a cancel frame lands in native state that a running // handler polls. Resolved through this holder, assigned once the attach - // succeeds; until then nothing is running, so nothing can be cancelled. + // succeeds. + // + // The native attach starts its job loop before this promise resolves, so + // the scheduler can dispatch into that window. An invocation waits for the + // holder to be filled rather than reading it empty, which would run a + // middleware the dispatch said was disabled and drop the job's progress. let attached: NativeExecutor | undefined; + let markAttached: () => void = () => {}; + const attachedReady = new Promise((resolve) => { + markAttached = resolve; + }); - const taskCallback = createTaskCallback({ + const invoke = createTaskCallback({ tasks, serializer, codecs, - middlewareFor, + // Every one of these reaches for the executor rather than for storage, + // which this process deliberately has none of: the scheduler holds the + // connection and does the work on its behalf. + middlewareFor: (taskName, jobId) => + middlewareFor(taskName, attached?.disabledMiddleware(jobId) ?? []), emitter, resources, queue, isCancelled: (jobId) => attached?.isCancelRequested(jobId) ?? false, + setProgress: (jobId, progress) => attached?.reportProgress(jobId, progress), + writeTaskLog: (jobId, taskName, level, message, extra) => + attached?.writeTaskLog(jobId, taskName, level, message, extra), }); + const taskCallback: typeof invoke = async (invocation) => { + await attachedReady; + return invoke(invocation); + }; + const native = await startNativeExecutor(taskCallback, { address, tasks: advertised, @@ -115,6 +143,7 @@ export class Executor { }); attached = native; + markAttached(); try { // Only lease the resource runtime once the attach actually succeeded, so a // refused handshake leaks nothing. diff --git a/sdks/node/src/queue.ts b/sdks/node/src/queue.ts index 043747942..dbc2b9dd0 100644 --- a/sdks/node/src/queue.ts +++ b/sdks/node/src/queue.ts @@ -1658,18 +1658,17 @@ export class Queue { * without polling storage itself. Hold the returned {@link Executor}. */ async runExecutor(options?: ExecutorRunOptions): Promise { - const disables = new MiddlewareDisableStore(this.native); const executor: Executor = await Executor.start(this.native, { onStopped: () => this.liveExecutors.delete(executor), tasks: this.tasks, serializer: this.serializer, codecs: this.codecs, - middlewareFor: (taskName) => { - const disabled = disables.getFor(taskName); - return disabled.length === 0 + // The list is supplied rather than read: an executor opens no storage, + // so the scheduler resolves the toggles and sends them with the job. + middlewareFor: (_taskName, disabled) => + disabled.length === 0 ? this.middleware - : this.middleware.filter((mw, index) => !disabled.includes(middlewareKey(mw, index))); - }, + : this.middleware.filter((mw, index) => !disabled.includes(middlewareKey(mw, index))), emitter: this.emitter, resources: this.resources, run: options, diff --git a/sdks/node/src/task-callback.ts b/sdks/node/src/task-callback.ts index 809b6dfc1..4cef111f7 100644 --- a/sdks/node/src/task-callback.ts +++ b/sdks/node/src/task-callback.ts @@ -21,8 +21,15 @@ export interface TaskCallbackDeps { serializer: Serializer; /** Named codec registry for per-task payload decode (see `TaskOptions.codecs`). */ codecs?: ReadonlyMap; - /** The middleware chain for a task, after dashboard disables are applied. */ - middlewareFor: (taskName: string) => readonly Middleware[]; + /** + * The middleware chain for a task, after dashboard disables are applied. + * + * Takes the job id because an attached executor resolves disables per + * dispatch — the scheduler attaches the list to the job frame, since the + * executor has no settings store to read it from. A worker ignores it and + * reads storage by task name. + */ + middlewareFor: (taskName: string, jobId: string) => readonly Middleware[]; emitter: Emitter; resources: ResourceRuntime; /** Backs progress, published partials, and the cancel-flag poll. */ @@ -35,6 +42,21 @@ export interface TaskCallbackDeps { * the native state those land in. */ isCancelled?: (jobId: string) => boolean; + /** + * Overrides where a task's progress goes. + * + * Same reason as {@link isCancelled}: an executor has no storage, so it + * sends progress to the scheduler, which applies it. + */ + setProgress?: (jobId: string, progress: number) => void; + /** Overrides where a task's log lines and published partials go. */ + writeTaskLog?: ( + jobId: string, + taskName: string, + level: string, + message: string, + extra?: string, + ) => void; } /** @@ -49,6 +71,13 @@ export function createTaskCallback( ): (invocation: JsTaskInvocation) => Promise { const { tasks, serializer, codecs, middlewareFor, emitter, resources, queue } = deps; const isCancelled = deps.isCancelled ?? ((jobId: string) => queue.isCancelRequested(jobId)); + const setProgress = + deps.setProgress ?? + ((jobId: string, progress: number) => queue.updateProgress(jobId, progress)); + const writeTaskLog = + deps.writeTaskLog ?? + ((jobId: string, taskName: string, level: string, message: string, extra?: string) => + queue.writeTaskLog(jobId, taskName, level, message, extra)); return async (invocation: JsTaskInvocation): Promise => { // Built-in workflow cache-return: echo the single (cached) arg as the result. @@ -74,16 +103,16 @@ export function createTaskCallback( // Resolve the middleware chain BEFORE allocating the cancel poller and // task scope — it reads storage and may throw, and nothing would clean // those up yet. - const chain = middlewareFor(invocation.taskName); + const chain = middlewareFor(invocation.taskName, invocation.id); // Cooperative cancel signal + job context exposed to the handler. const controller = new AbortController(); const context: JobContext = { jobId: invocation.id, signal: controller.signal, - setProgress: (progress) => queue.updateProgress(invocation.id, progress), + setProgress: (progress) => setProgress(invocation.id, progress), publish: (value) => - queue.writeTaskLog(invocation.id, invocation.taskName, "result", "", JSON.stringify(value)), + writeTaskLog(invocation.id, invocation.taskName, "result", "", JSON.stringify(value)), }; const poller = setInterval(() => { try { diff --git a/sdks/node/src/worker.ts b/sdks/node/src/worker.ts index f40ebbc2c..9a69611a6 100644 --- a/sdks/node/src/worker.ts +++ b/sdks/node/src/worker.ts @@ -103,6 +103,8 @@ export class Worker { // every invocation (live toggles); task/queue overrides apply here, at // worker startup. const disables = new MiddlewareDisableStore(queue); + // The job id is unused here: a worker has storage, so it reads the live + // toggle list by task name rather than taking one off the dispatch. const middlewareFor = (taskName: string): readonly Middleware[] => { const disabled = disables.getFor(taskName); if (disabled.length === 0) { diff --git a/sdks/node/test/integrations/cli.test.ts b/sdks/node/test/integrations/cli.test.ts index 74817d32c..a53f15c2f 100644 --- a/sdks/node/test/integrations/cli.test.ts +++ b/sdks/node/test/integrations/cli.test.ts @@ -76,11 +76,14 @@ describe("taskito CLI", () => { stdio: "ignore", env: { ...process.env, TASKITO_INDEX: indexUrl, TASKITO_DB: db, TASKITO_MARKER: marker }, }); - expect(await waitForFile(marker, 6000)).toBe(true); + expect(await waitForFile(marker, 25_000)).toBe(true); } finally { child?.kill("SIGTERM"); } - }); + // Spawning a second Node, importing the SDK and opening SQLite before the + // job can run costs more than the file's default budget on a cold Windows + // runner, where this timed out at just over six seconds. + }, 30_000); }); async function waitForFile(path: string, timeoutMs: number): Promise { diff --git a/sdks/node/test/worker/executorAttach.test.ts b/sdks/node/test/worker/executorAttach.test.ts index 29fb63005..892f24d02 100644 --- a/sdks/node/test/worker/executorAttach.test.ts +++ b/sdks/node/test/worker/executorAttach.test.ts @@ -54,6 +54,10 @@ class FakeScheduler { hello?: Record; /** Set when the handshake should be refused rather than acked. */ refuse = false; + /** Optional behaviours this scheduler advertises in its `hello_ack`. */ + capabilities: readonly string[] = []; + /** Run immediately after the ack, to dispatch into the attach window. */ + afterAck?: () => void; private constructor(server: Server, port: number, connected: Promise) { this.server = server; @@ -117,7 +121,12 @@ class FakeScheduler { schedulerId: undefined, scheduler_id: "fake-scheduler", protocol_version: PROTOCOL_VERSION, + // Whatever this scheduler promises to do on the executor's behalf. + // Empty by default: that is a scheduler built before the + // side-channel existed, and the case worth defaulting to. + capabilities: this.capabilities, }); + this.afterAck?.(); } else { this.socket?.destroy(); } @@ -142,7 +151,12 @@ class FakeScheduler { id: string, taskName: string, payload: Buffer, - options?: { retryCount?: number; maxRetries?: number; timeoutMs?: number }, + options?: { + retryCount?: number; + maxRetries?: number; + timeoutMs?: number; + disabledMiddleware?: readonly string[]; + }, ): void { this.send( { @@ -155,11 +169,33 @@ class FakeScheduler { queue: "default", timeout_ms: options?.timeoutMs ?? 30_000, namespace: null, + // Resolved by the scheduler, because an executor has no settings store + // of its own to read the toggle list from. + disabled_middleware: options?.disabledMiddleware ?? [], + metadata: null, }, payload, ); } + /** + * Every side-channel frame a job produced, plus its result. + * + * The result is ordered behind them on one connection, so its arrival is + * what proves the collection is complete rather than merely early. + */ + async collectUntilResult(): Promise<{ result: Frame; sideChannel: Frame[] }> { + const sideChannel: Frame[] = []; + for (;;) { + const frame = await this.nextResult(); + if (frame.header.type === "progress" || frame.header.type === "task_log") { + sideChannel.push(frame); + continue; + } + return { result: frame, sideChannel }; + } + } + /** Wait for the executor's `hello` to arrive. */ async attached(): Promise> { await this.connected; @@ -224,6 +260,11 @@ function declaredPayloadLength(header: Record): number { if (header.type === "success") { return typeof header.result_len === "number" ? header.result_len : 0; } + if (header.type === "task_log") { + // A published partial can be arbitrarily large, so `extra` rides as the + // frame's blob rather than inside the header. + return typeof header.extra_len === "number" ? header.extra_len : 0; + } return 0; } @@ -235,6 +276,15 @@ function newQueue(): Queue { return new Queue({ dbPath: join(mkdtempSync(join(tmpdir(), "taskito-exec-")), "q.db") }); } +/** Uses the job-scoped conveniences that need storage in a worker. */ +async function reportingHandler(): Promise { + const job = currentJob(); + job?.setProgress(50); + job?.publish({ stage: "halfway" }); + job?.setProgress(100); + return "reported"; +} + /** Encode a call the way the enqueue path does. */ function payloadFor(queue: Queue, args: unknown[]): Buffer { // biome-ignore lint/complexity/useLiteralKeys: reaching the internal serializer @@ -539,27 +589,112 @@ it("opens no storage", async () => { }); it("degrades progress and publish rather than failing the job", async () => { - // Losing the progress bar is a degradation; failing the job over it would be - // a regression for anyone moving a worker to an executor. + // This scheduler advertises no side-channel — an older `taskito-server` — + // so the executor sends nothing it could not parse. Losing the progress bar + // is a degradation; failing the job over it would be a regression for anyone + // moving a worker to an executor. scheduler = await FakeScheduler.listen(); process.env[DETACHED_ENV] = "1"; try { const queue = new Queue({ backend: "postgres", dsn: "postgres://x:y@127.0.0.1:1/absent" }); - queue.task("reports", async () => { - const job = currentJob(); - job?.setProgress(50); - job?.publish({ stage: "halfway" }); - job?.setProgress(100); - return "reported"; - }); + queue.task("reports", reportingHandler); executor = await queue.runExecutor({ attach: `127.0.0.1:${scheduler.port}` }); await scheduler.attached(); scheduler.sendJob("job-1", "reports", payloadFor(queue, [])); - const frame = await scheduler.nextResult(); - expect(frame.header.type).toBe("success"); - expect(frame.header.job_id).toBe("job-1"); + const { result, sideChannel } = await scheduler.collectUntilResult(); + expect(result.header.type).toBe("success"); + expect(result.header.job_id).toBe("job-1"); + expect(sideChannel).toEqual([]); + } finally { + delete process.env[DETACHED_ENV]; + } +}); + +it("sends progress and logs to a scheduler that advertised the side-channel", async () => { + // The whole point of #589: a task on an executor is not silently poorer than + // the same task on an in-process worker. + scheduler = await FakeScheduler.listen(); + scheduler.capabilities = ["side_channel"]; + process.env[DETACHED_ENV] = "1"; + try { + const queue = new Queue({ backend: "postgres", dsn: "postgres://x:y@127.0.0.1:1/absent" }); + queue.task("reports", reportingHandler); + + executor = await queue.runExecutor({ attach: `127.0.0.1:${scheduler.port}` }); + await scheduler.attached(); + scheduler.sendJob("job-1", "reports", payloadFor(queue, [])); + + const { result, sideChannel } = await scheduler.collectUntilResult(); + expect(result.header.type).toBe("success"); + + const progress = sideChannel + .filter((frame) => frame.header.type === "progress") + .map((frame) => frame.header.progress); + expect(progress.at(-1)).toBe(100); + + const partial = sideChannel.find((frame) => frame.header.level === "result"); + expect(partial).toBeDefined(); + expect(partial?.header.job_id).toBe("job-1"); + expect(partial?.header.task_name).toBe("reports"); + expect(JSON.parse(partial?.payload.toString() ?? "")).toEqual({ stage: "halfway" }); + } finally { + delete process.env[DETACHED_ENV]; + } +}); + +it("skips a middleware the dispatch says is disabled", async () => { + // A dashboard toggle has to reach a process that cannot read settings, so it + // rides the job frame instead. + scheduler = await FakeScheduler.listen(); + process.env[DETACHED_ENV] = "1"; + try { + const ran: string[] = []; + const queue = new Queue({ backend: "postgres", dsn: "postgres://x:y@127.0.0.1:1/absent" }); + queue.use({ name: "recorder", before: () => void ran.push("recorder") }); + queue.task("echo", (value: string) => `echo:${value}`); + + executor = await queue.runExecutor({ attach: `127.0.0.1:${scheduler.port}` }); + await scheduler.attached(); + + scheduler.sendJob("job-1", "echo", payloadFor(queue, ["a"])); + expect((await scheduler.nextResult()).header.type).toBe("success"); + expect(ran).toEqual(["recorder"]); + + scheduler.sendJob("job-2", "echo", payloadFor(queue, ["b"]), { + disabledMiddleware: ["recorder"], + }); + expect((await scheduler.nextResult()).header.type).toBe("success"); + expect(ran).toEqual(["recorder"]); + } finally { + delete process.env[DETACHED_ENV]; + } +}); + +it("honours the toggles on a job dispatched in the same tick as the ack", async () => { + // The earliest a scheduler can dispatch: the job frame follows the ack with + // nothing in between. The native attach starts its job loop before + // `runExecutor` resolves, so an invocation here can outrun the holder the + // callback reads the executor from — and reading it empty would run a + // middleware the dispatch said was disabled. + scheduler = await FakeScheduler.listen(); + process.env[DETACHED_ENV] = "1"; + try { + const ran: string[] = []; + const queue = new Queue({ backend: "postgres", dsn: "postgres://x:y@127.0.0.1:1/absent" }); + queue.use({ name: "recorder", before: () => void ran.push("recorder") }); + queue.task("echo", (value: string) => `echo:${value}`); + + const payload = payloadFor(queue, ["a"]); + scheduler.afterAck = () => { + scheduler?.sendJob("job-1", "echo", payload, { disabledMiddleware: ["recorder"] }); + }; + + executor = await queue.runExecutor({ attach: `127.0.0.1:${scheduler.port}` }); + + expect((await scheduler.nextResult()).header.type).toBe("success"); + expect(ran).toEqual([]); } finally { delete process.env[DETACHED_ENV]; } diff --git a/sdks/node/test/worker/executorAttachServer.test.ts b/sdks/node/test/worker/executorAttachServer.test.ts index 0cf654316..f9d6644a8 100644 --- a/sdks/node/test/worker/executorAttachServer.test.ts +++ b/sdks/node/test/worker/executorAttachServer.test.ts @@ -16,7 +16,7 @@ import { createConnection, createServer } from "node:net"; import { tmpdir } from "node:os"; import { join, resolve } from "node:path"; import { afterEach, describe, expect, it } from "vitest"; -import { type Executor, Queue } from "../../src/index"; +import { currentJob, type Executor, Queue } from "../../src/index"; const SERVER_BIN = process.env.TASKITO_SERVER_BIN; const SETTLE_MS = 60_000; @@ -141,6 +141,65 @@ describe.skipIf(!SERVER_BIN)("against a real taskito-server", () => { ); }, 120_000); + it("puts progress and published partials into storage via the scheduler", async () => { + // The done-when for #589, against the binary an operator actually runs: + // this process holds no database credentials, so these rows can only have + // reached storage through the scheduler. + const { port, dbPath } = await startScheduler(); + const queue = new Queue({ dbPath }); + queue.task("reports", async () => { + const job = currentJob(); + job?.setProgress(50); + job?.publish({ stage: "halfway" }); + job?.setProgress(100); + return "reported"; + }); + + executor = await queue.runExecutor({ attach: `127.0.0.1:${port}` }); + await sleep(1000); + const jobId = String(queue.enqueue("reports", [])); + + // The side-channel is fire-and-forget, so the last progress value and the + // partial may land just after the result. Making the job's completion a + // barrier for them would put a database write between a task and its own + // result, which is exactly what an executor exists to avoid. + await waitFor(async () => queue.getJob(jobId)?.progress === 100, "progress never reached 100"); + await waitFor( + async () => queue.taskLogs(jobId).some((entry) => entry.level === "result"), + "the published partial never reached storage", + ); + + const partial = queue.taskLogs(jobId).find((entry) => entry.level === "result"); + expect(JSON.parse(String(partial?.extra))).toEqual({ stage: "halfway" }); + }, 120_000); + + it("honours a dashboard middleware toggle on an attached executor", async () => { + const { port, dbPath } = await startScheduler(); + const queue = new Queue({ dbPath }); + const ran: string[] = []; + queue.use({ name: "recorder", before: () => void ran.push("recorder") }); + queue.task("echo", (value: string) => `echo:${value}`); + + executor = await queue.runExecutor({ attach: `127.0.0.1:${port}` }); + await sleep(1000); + + const first = String(queue.enqueue("echo", ["a"])); + await waitFor(async () => queue.getJob(first)?.status === "complete", "the first job hung"); + expect(ran).toEqual(["recorder"]); + + queue.disableMiddlewareForTask("echo", "recorder"); + // The scheduler resolves the list per dispatch behind a short cache, so a + // toggle takes effect within it rather than instantly. Each probe is judged + // against the count before it ran, since an earlier probe that predates the + // cache expiry legitimately still fires the middleware. + await waitFor(async () => { + const before = ran.length; + const jobId = String(queue.enqueue("echo", ["b"])); + await waitFor(async () => queue.getJob(jobId)?.status === "complete", "a probe job hung"); + return ran.length === before; + }, "a middleware disabled in the dashboard still ran on the executor"); + }, 120_000); + it("refuses an attach with the wrong token", async () => { const { port, dbPath } = await startScheduler({ token: "correct-token-0123456789" }); const queue = new Queue({ dbPath }); diff --git a/sdks/python/tests/core/test_shutdown.py b/sdks/python/tests/core/test_shutdown.py index 0b2dd820f..cd9d7ee8a 100644 --- a/sdks/python/tests/core/test_shutdown.py +++ b/sdks/python/tests/core/test_shutdown.py @@ -44,21 +44,35 @@ def slow_task() -> str: assert fetched.status == "complete" -def test_shutdown_stops_worker(queue: Queue) -> None: +def test_shutdown_stops_worker(queue: Queue, poll_until: PollUntil) -> None: """queue.shutdown() causes run_worker to return.""" + started = threading.Event() @queue.task() def noop() -> None: - pass + started.set() + + job = noop.delay() worker_thread = threading.Thread(target=queue.run_worker, daemon=True) worker_thread.start() - # Tiny grace window so the worker reaches its poll loop before shutdown. - time.sleep(0.1) + # A job the worker actually ran, rather than a fixed grace window: reaching + # the poll loop takes far longer than 100ms on a loaded runner, and a + # shutdown that lands before it does leaves this waiting on a worker that + # never saw the request. + poll_until( + lambda: ( + started.is_set() + and (j := queue.get_job(job.id)) is not None + and j.status == "complete" + ), + timeout=30, + message="the worker never reached its poll loop", + ) queue.shutdown() - worker_thread.join(timeout=10) + worker_thread.join(timeout=30) assert not worker_thread.is_alive()