diff --git a/README.md b/README.md index 55b3ef7..2b538fd 100644 --- a/README.md +++ b/README.md @@ -204,6 +204,8 @@ In-repo: [`docs/engine.md`](docs/engine.md) (engine internals), [`docs/history.m | Idempotent produce (`IDMP`) | ✓ | — | — | — | | Stable job IDs (`addUnique`) | ✓ | ✓ | ✓ | — | | Result backends (`getJobResult` / `waitForResult`) | ✓ | ✓ | ✓ | — | +| Event-driven completion wait (`waitUntilFinished`) | ✓ | ✓ | ✓ | — | +| Worker / Queue / Job event listeners (`EventEmitter`-style) | ✓ | ✓ | ✓ | partial | | Delayed jobs | ✓ | ✓ | ✓ | — | | Idempotent delayed scheduling | ✓ | — | — | — | | Cancel scheduled job | ✓ | ✓ | ✓ | — | diff --git a/chasquimq-node/README.md b/chasquimq-node/README.md index e74c684..f21c695 100644 --- a/chasquimq-node/README.md +++ b/chasquimq-node/README.md @@ -69,9 +69,9 @@ main() | Surface | What it does | |---|---| | `Queue` | Producer + queue inspection. `add` / `addBulk` / `addUnique` / `getJob` / `getJobs` / `getJobState` / `getJobCounts` / `getWaitingCount` / `getActiveCount` / `getDelayedCount` / `getCompletedCount` / `getFailedCount` / `count` / `getJobResult` / `peekDlq` / `replayDlq` / `cancelDelayed` / `getRepeatableJobs` / `removeRepeatableByKey` / `pause` / `resume` / `isPaused` / `remove` / `removeReport` / `drain` / `clean` / `obliterate`. `[Symbol.asyncDispose]`. | -| `Worker` | Consumer pool. tokio-side dispatch, opt-in result storage (`storeResults: true`), `EventEmitter` events (`completed` / `failed` / `error`), `pause` / `resume` / `isPaused`. `[Symbol.asyncDispose]`. | -| `Job` | Read-only handle. `id`, `name`, `data`, `attemptsMade`, `waitForResult({ timeoutMs, intervalMs, signal })`. | -| `QueueEvents` | Cross-process pub/sub via the events stream. Subscribe to `completed` / `failed` / `dlq` / `retry-scheduled` / `delayed` / `drained`. `[Symbol.asyncDispose]`. | +| `Worker` | Consumer pool. tokio-side dispatch, opt-in result storage (`storeResults: true`), `EventEmitter` events (`ready` / `active` / `completed` / `failed` / `error` / `closing` / `closed` / `drained` / `paused` / `resumed`), `pause` / `resume` / `isPaused`. `[Symbol.asyncDispose]`. | +| `Job` | Read-only handle. `id`, `name`, `data`, `attemptsMade`, `waitForResult({ timeoutMs, intervalMs, signal })`, `waitUntilFinished(queueEvents, ttl?)`. | +| `QueueEvents` | Cross-process pub/sub via the events stream. Subscribe to `waiting` / `active` / `completed` / `failed` / `dlq` / `retry-scheduled` / `delayed` / `drained` / `retries-exhausted`, plus per-id channels (`completed:` / `failed:` / `active:`) for targeted subscribers. `[Symbol.asyncDispose]`. | | `BackoffSpec` | Builders: `.fixed(delayMs)` / `.exponential(initialMs, { multiplier, maxMs, jitterMs })`. | | `UnrecoverableError` | Throw from your handler to bypass retries and route the job directly to DLQ. | | `NotSupportedError` | Surfaces from APIs that aren't on the chasquimq roadmap (e.g. parent/child flows). | @@ -190,6 +190,89 @@ await queue.resume() // lift it everywhere The same durable flag is what the CLI's `chasqui pause ` / `chasqui resume ` toggle. Both surfaces are idempotent — double-pause / double-resume are no-ops. +### Subscribing to events + +Two layered surfaces: per-worker `EventEmitter` events for in-process observation, and the cross-process `QueueEvents` stream for fan-out across workers, dashboards, and ops tooling. + +In-process `Worker` events fire on the local worker only and use plain `EventEmitter` semantics: + +```ts +const worker = new Worker("emails", handler, { connection }) + +worker.on("ready", () => console.log("engine loop started")) +worker.on("active", (job) => console.log("running", job.id)) +worker.on("completed", (job, result) => console.log("ok", job.id, result)) +worker.on("failed", (job, err) => console.error("failed", job?.id, err)) +worker.on("error", (err) => console.error("engine error", err)) +worker.on("drained", () => console.log("queue is empty")) +worker.on("paused", () => console.log("worker parked")) +worker.on("resumed", () => console.log("worker resumed")) +worker.on("closing", () => console.log("shutting down")) +worker.on("closed", () => console.log("shutdown complete")) +``` + +`drained` is the only `Worker` event that requires a cross-process subscription (the engine emits it on the events stream, since "no jobs left" is a queue-wide observation). It's lazily wired: the first `worker.on('drained', ...)` call spawns an embedded `QueueEvents` subscriber, torn down on `worker.close()`. Workers that never subscribe pay no extra Redis connections. + +Cross-process `QueueEvents` listens on the events stream — every process running a `QueueEvents` instance for the same queue sees every transition, regardless of which worker emitted it: + +```ts +import { QueueEvents } from "chasquimq" + +const events = new QueueEvents("emails", { connection }) +await events.waitUntilReady() + +events.on("waiting", ({ jobId, name }) => { /* enqueued */ }) +events.on("active", ({ jobId, name, attempt }) => { /* started */ }) +events.on("completed", ({ jobId, name, attempt }) => { /* ok */ }) +events.on("failed", ({ jobId, failedReason, attempt }) => { /* err */ }) +events.on("drained", () => { /* empty */ }) + +// chasquimq-specific extensions +events.on("retry-scheduled", ({ jobId, attempt, backoffMs }) => { /* next try */ }) +events.on("dlq", ({ jobId, reason, attempt }) => { /* moved to DLQ */ }) +events.on("retries-exhausted", ({ jobId, attemptsMade, reason }) => { /* DLQ */ }) + +// Per-id channels — fire only for the named job. Used internally by +// Job.waitUntilFinished; useful for targeted UI updates without +// filtering every broadcast event by jobId. +events.on(`completed:${jobId}`, ({ jobId }) => { /* this job, done */ }) +events.on(`failed:${jobId}`, ({ failedReason }) => { /* this job, failed */ }) + +await events.close() +``` + +`completed` payloads from the events stream do **not** carry the handler's return value (that would double-allocate the payload onto every subscriber). To read the value, pair `QueueEvents` with `Queue.getJobResult(jobId)` and run the worker with `storeResults: true`. The next section's `Job.waitUntilFinished` does this for you. + +`progress` and `stalled` listeners are accepted on `Worker` for API stability but are currently no-op — the engine doesn't emit those transitions yet. Attach them safely; they'll start firing when the corresponding engine work lands. + +### Awaiting a single job's completion + +`Job.waitUntilFinished(queueEvents, ttl?)` is the event-driven completion-wait — subscribes to the per-id `completed:` / `failed:` channels and resolves / rejects on the first to fire: + +```ts +const events = new QueueEvents("emails", { connection }) +await events.waitUntilReady() + +const job = await queue.add("send", { to: "ada@example.com" }) + +try { + // Resolves with the handler's return value (requires + // storeResults: true on the worker; otherwise resolves with + // undefined). Rejects with new Error(failedReason) on failure. + // ttl in ms — omit for an unbounded wait. + const result = await job.waitUntilFinished(events, 30_000) + console.log("sent:", result) +} catch (err) { + // Either a job failure (engine reason) or a + // WaitUntilFinishedTimeoutError (ttl elapsed with no terminal event). + console.error(err) +} finally { + await events.close() +} +``` + +Distinct from `Job.waitForResult({ timeoutMs })`: that one polls the `Queue.getJobResult` Redis key and **requires** `storeResults: true` to detect completion at all. `waitUntilFinished` is event-driven, so it detects completion even when no result key was written — but it cannot tell you a job that already finished before the call wired up. Pick `waitUntilFinished` for low-latency awaits of jobs you're about to enqueue; pick `waitForResult` for "did this id ever finish". + ### Inspect jobs Read-only introspection across the queue's stream, delayed ZSET, DLQ, and result-key surfaces. Bounded scans; no secondary index; zero impact on the producer / consumer hot paths. diff --git a/chasquimq-node/__test__/event-listeners.test.ts b/chasquimq-node/__test__/event-listeners.test.ts new file mode 100644 index 0000000..ed9948e --- /dev/null +++ b/chasquimq-node/__test__/event-listeners.test.ts @@ -0,0 +1,287 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { + Queue, + QueueEvents, + Worker, + WaitUntilFinishedTimeoutError, +} from '../dist/index.js' + +const REDIS_URL = process.env.REDIS_URL +const skipIfNoRedis = REDIS_URL ? describe : describe.skip + +skipIfNoRedis('Worker + Queue + Job event listeners', () => { + let queueName: string + let queue: Queue<{ value: number }, number> + let worker: Worker<{ value: number }, number> | undefined + let queueEvents: QueueEvents | undefined + + beforeEach(() => { + queueName = `qmq-test-evl-${Date.now()}-${Math.random().toString(36).slice(2, 8)}` + queue = new Queue(queueName, { connection: parseConn(REDIS_URL!) }) + }) + + afterEach(async () => { + if (worker) { + await worker.close().catch(() => {}) + worker = undefined + } + if (queueEvents) { + await queueEvents.close().catch(() => {}) + queueEvents = undefined + } + await queue.close().catch(() => {}) + }) + + // --- Worker.on('drained') ------------------------------------------------- + + it("Worker emits 'drained' after the engine observes a full→empty transition", async () => { + const drainedSpy = vi.fn() + + worker = new Worker<{ value: number }, number>( + queueName, + async (job) => job.data.value, + { + connection: parseConn(REDIS_URL!), + concurrency: 4, + drainDelay: 200, + autorun: false, + }, + ) + worker.on('drained', drainedSpy) + void worker.run() + + // Add then wait for the worker to process the job AND observe an + // empty XREADGROUP. The engine emits `drained` on the full→empty + // transition only (not on every empty poll). + await queue.add('one', { value: 1 }) + + await waitFor(() => drainedSpy.mock.calls.length >= 1, 10_000) + expect(drainedSpy).toHaveBeenCalled() + }, 30_000) + + it("Worker doesn't spawn a drained subscriber when no listener attaches", async () => { + worker = new Worker<{ value: number }, number>( + queueName, + async (job) => job.data.value, + { connection: parseConn(REDIS_URL!), autorun: false }, + ) + void worker.run() + await queue.add('one', { value: 1 }) + + // Give the worker a tick to process, then close. The internal + // subscriber field should never have been constructed — observable + // by `close()` returning quickly (no extra `XREAD BLOCK` to drain). + await new Promise((r) => setTimeout(r, 100)) + + const start = Date.now() + await worker.close() + const elapsed = Date.now() - start + // Native consumer shutdown is fast (<1s). A live QueueEvents would + // add up to `blockingTimeout` (10s default) wait. + expect(elapsed).toBeLessThan(3_000) + }) + + // --- Worker.on('paused' | 'resumed') ------------------------------------- + + it("Worker emits 'paused' and 'resumed' when pause()/resume() are called", async () => { + const pausedSpy = vi.fn() + const resumedSpy = vi.fn() + + worker = new Worker<{ value: number }, number>( + queueName, + async (job) => job.data.value, + { connection: parseConn(REDIS_URL!), autorun: false }, + ) + worker.on('paused', pausedSpy) + worker.on('resumed', resumedSpy) + void worker.run() + // Let it start + await new Promise((r) => setTimeout(r, 50)) + + await worker.pause() + expect(pausedSpy).toHaveBeenCalledTimes(1) + + worker.resume() + expect(resumedSpy).toHaveBeenCalledTimes(1) + }) + + // --- QueueEvents per-id channels ------------------------------------------ + + it("QueueEvents emits per-id `completed:` alongside the broadcast", async () => { + queueEvents = new QueueEvents(queueName, { + connection: parseConn(REDIS_URL!), + autorun: true, + blockingTimeout: 500, + }) + await queueEvents.waitUntilReady() + + worker = new Worker<{ value: number }, number>( + queueName, + async (job) => job.data.value, + { connection: parseConn(REDIS_URL!), autorun: false }, + ) + void worker.run() + + const job = await queue.add('compute', { value: 42 }) + const targetedSpy = vi.fn() + const broadcastSpy = vi.fn() + queueEvents.on(`completed:${job.id}`, targetedSpy) + queueEvents.on('completed', broadcastSpy) + + await waitFor( + () => targetedSpy.mock.calls.length >= 1 && broadcastSpy.mock.calls.length >= 1, + 10_000, + ) + const [payload] = targetedSpy.mock.calls[0]! + expect((payload as { jobId: string }).jobId).toBe(job.id) + }, 30_000) + + it("QueueEvents emits per-id `failed:` with the engine reason", async () => { + queueEvents = new QueueEvents(queueName, { + connection: parseConn(REDIS_URL!), + autorun: true, + blockingTimeout: 500, + }) + await queueEvents.waitUntilReady() + + worker = new Worker<{ value: number }, number>( + queueName, + async () => { + throw new Error('boom') + }, + { + connection: parseConn(REDIS_URL!), + concurrency: 1, + maxStalledCount: 1, + autorun: false, + }, + ) + void worker.run() + + const job = await queue.add('fail-me', { value: 0 }) + const targetedSpy = vi.fn() + queueEvents.on(`failed:${job.id}`, targetedSpy) + + await waitFor(() => targetedSpy.mock.calls.length >= 1, 10_000) + const [payload] = targetedSpy.mock.calls[0]! + expect((payload as { jobId: string; failedReason: string })).toMatchObject({ + jobId: job.id, + failedReason: 'boom', + }) + }, 30_000) + + // --- Job.waitUntilFinished ----------------------------------------------- + + it('Job.waitUntilFinished resolves on completed with the stored result', async () => { + queueEvents = new QueueEvents(queueName, { + connection: parseConn(REDIS_URL!), + autorun: true, + blockingTimeout: 500, + }) + await queueEvents.waitUntilReady() + + worker = new Worker<{ value: number }, number>( + queueName, + async (job) => job.data.value * 3, + { + connection: parseConn(REDIS_URL!), + autorun: false, + storeResults: true, + resultTtlMs: 60_000, + }, + ) + void worker.run() + + const job = await queue.add('triple', { value: 7 }) + const result = await job.waitUntilFinished(queueEvents, 10_000) + expect(result).toBe(21) + }, 30_000) + + it('Job.waitUntilFinished resolves with undefined when storeResults is off', async () => { + queueEvents = new QueueEvents(queueName, { + connection: parseConn(REDIS_URL!), + autorun: true, + blockingTimeout: 500, + }) + await queueEvents.waitUntilReady() + + worker = new Worker<{ value: number }, number>( + queueName, + async (job) => job.data.value, + { connection: parseConn(REDIS_URL!), autorun: false }, + ) + void worker.run() + + const job = await queue.add('no-store', { value: 7 }) + const result = await job.waitUntilFinished(queueEvents, 10_000) + // Event-driven completion still detected; the return value is + // unavailable without the result backend. + expect(result).toBeUndefined() + }, 30_000) + + it('Job.waitUntilFinished rejects with the engine reason on failure', async () => { + queueEvents = new QueueEvents(queueName, { + connection: parseConn(REDIS_URL!), + autorun: true, + blockingTimeout: 500, + }) + await queueEvents.waitUntilReady() + + worker = new Worker<{ value: number }, number>( + queueName, + async () => { + throw new Error('handler said no') + }, + { + connection: parseConn(REDIS_URL!), + concurrency: 1, + maxStalledCount: 1, + autorun: false, + }, + ) + void worker.run() + + const job = await queue.add('rejects', { value: 0 }) + await expect(job.waitUntilFinished(queueEvents, 10_000)).rejects.toThrow( + /handler said no/, + ) + }, 30_000) + + it('Job.waitUntilFinished throws WaitUntilFinishedTimeoutError on ttl elapse', async () => { + queueEvents = new QueueEvents(queueName, { + connection: parseConn(REDIS_URL!), + autorun: true, + blockingTimeout: 500, + }) + await queueEvents.waitUntilReady() + + // No worker; the job will sit waiting forever. + const job = await queue.add('orphan', { value: 0 }) + await expect(job.waitUntilFinished(queueEvents, 300)).rejects.toThrow( + WaitUntilFinishedTimeoutError, + ) + }) +}) + +async function waitFor( + predicate: () => boolean, + timeoutMs: number, +): Promise { + const start = Date.now() + while (!predicate()) { + if (Date.now() - start > timeoutMs) + throw new Error(`waitFor timed out after ${timeoutMs}ms`) + await new Promise((res) => setTimeout(res, 25)) + } +} + +function parseConn(url: string) { + const u = new URL(url) + return { + host: u.hostname || '127.0.0.1', + port: u.port ? Number(u.port) : 6379, + password: u.password || undefined, + username: u.username || undefined, + db: u.pathname && u.pathname !== '/' ? Number(u.pathname.slice(1)) : undefined, + } +} diff --git a/chasquimq-node/package-lock.json b/chasquimq-node/package-lock.json index a81ed92..2b40891 100644 --- a/chasquimq-node/package-lock.json +++ b/chasquimq-node/package-lock.json @@ -1,12 +1,12 @@ { "name": "chasquimq", - "version": "0.1.0", + "version": "1.3.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "chasquimq", - "version": "0.1.0", + "version": "1.3.0", "license": "MIT", "dependencies": { "@msgpack/msgpack": "^3", diff --git a/chasquimq-node/src-ts/errors.ts b/chasquimq-node/src-ts/errors.ts index 3bd999c..668413f 100644 --- a/chasquimq-node/src-ts/errors.ts +++ b/chasquimq-node/src-ts/errors.ts @@ -64,3 +64,23 @@ export class WaitForResultTimeoutError extends Error { this.name = "WaitForResultTimeoutError"; } } + +/** + * Thrown by {@link Job.waitUntilFinished} when neither a `completed` nor + * a `failed` event for the watched job arrives within the supplied + * `ttl` ms. Distinct from a failed job: a failed job rejects with a + * regular `Error` carrying the engine-reported `failedReason`. This + * error fires only when the events stream itself goes silent (the worker + * died, the network blipped, or the `ttl` was too short for the + * handler). + * + * Detect via `err.name === 'WaitUntilFinishedTimeoutError'` rather than + * `err instanceof WaitUntilFinishedTimeoutError` so subclasses across + * realms (workers / vm contexts) still match. + */ +export class WaitUntilFinishedTimeoutError extends Error { + constructor(message?: string) { + super(message ?? "waitUntilFinished timed out"); + this.name = "WaitUntilFinishedTimeoutError"; + } +} diff --git a/chasquimq-node/src-ts/job.ts b/chasquimq-node/src-ts/job.ts index c8d3369..e80f105 100644 --- a/chasquimq-node/src-ts/job.ts +++ b/chasquimq-node/src-ts/job.ts @@ -8,8 +8,13 @@ // rewriting a stream entry (e.g. `update`) throw — Streams are append-only. import type { JobsOptions, JobState, JobProgress } from "./types.js"; -import { NotSupportedError, WaitForResultTimeoutError } from "./errors.js"; +import { + NotSupportedError, + WaitForResultTimeoutError, + WaitUntilFinishedTimeoutError, +} from "./errors.js"; import type { Queue } from "./queue.js"; +import type { QueueEvents } from "./queue-events.js"; /** * Options for {@link Job.waitForResult}. @@ -201,6 +206,135 @@ export class Job< } } + /** + * Subscribe to the engine's events stream and resolve / reject when + * the `completed` or `failed` event for **this** job fires. + * + * Unlike {@link Job.waitForResult}, this method is event-driven (no + * polling, no Redis `GET` per interval) and does not require + * `WorkerOptions.storeResults = true` to detect completion. It does, + * however, require an attached {@link QueueEvents} subscriber so the + * events-stream traffic actually reaches this process. + * + * Return value semantics: + * - On `completed`, resolves with the handler's return value when + * `WorkerOptions.storeResults = true` was set on the worker. The + * value is fetched via `Queue.getJobResult(this.id)` after the + * event fires. If `storeResults` was not enabled (or the handler + * returned `undefined`), resolves with `undefined`. The events + * stream itself never carries the return value — keeping subscriber + * traffic small and predictable. + * - On `failed`, rejects with `new Error(failedReason)` carrying the + * engine-reported reason (the same string surfaced on the + * `Worker`'s `failed` event). + * - On `ttl` elapse, rejects with {@link WaitUntilFinishedTimeoutError}. + * + * **Race window.** If the job completed (or failed) *before* this + * call wires up its listeners, the events-stream event has already + * been dispatched and this method has nothing to subscribe to. The + * `ttl` will fire normally. For producers that want to await a job + * that may already have finished, pair this with a `getJobState` + * check, or use {@link Job.waitForResult} (which can read a + * persisted result key written before the wait started). + * + * @param queueEvents A live {@link QueueEvents} subscriber for the + * same queue this job was added to. The caller owns the subscriber + * (create one per process; share it across all `waitUntilFinished` + * calls). + * @param ttl Optional timeout in milliseconds. Omit for an unbounded + * wait — practical only when you control the worker and trust it to + * either complete or fail every job. + */ + async waitUntilFinished( + queueEvents: QueueEvents, + ttl?: number, + ): Promise { + const jobId = this.id; + // Surface a queue/queueEvents mismatch up-front rather than + // letting the wait silently time out (the events stream is + // per-queue, so a `QueueEvents` for a different queue will never + // fire the per-id channel). Only checked when we have the queue + // backref; jobs constructed without one (the worker-side path, + // which has no queue handle) skip the guard — that's also the + // path where calling `waitUntilFinished` is unusual. + if (this.queue && queueEvents.name !== this.queue.name) { + throw new Error( + `Job.waitUntilFinished: queueEvents is for "${queueEvents.name}" ` + + `but this job is on "${this.queue.name}" — pass a QueueEvents ` + + `subscribed to the right queue`, + ); + } + const completedChannel = `completed:${jobId}`; + const failedChannel = `failed:${jobId}`; + + return new Promise((resolve, reject) => { + let timer: ReturnType | undefined; + + const onCompleted = ( + _args: { jobId: string; name: string; returnvalue: unknown }, + ): void => { + cleanup(); + // Fetch the stored result on a best-effort basis. The engine + // emits the `completed` event *before* the per-entry + // JOB_OK_SCRIPT writes the result key (the events emit is + // off the ack hot path, the result write is on it), so a + // single `getJobResult` immediately after the event can lose + // the race. Poll a few times with short backoff to give the + // result writer a chance to land. If `storeResults` was + // disabled on the worker, every poll returns `undefined` and + // we fall through to resolve(undefined) — same shape as + // jobs whose handler explicitly returned undefined. + const queue = this.queue; + if (!queue) { + resolve(undefined); + return; + } + void (async () => { + for (let i = 0; i < 10; i++) { + try { + const v = await queue.getJobResult(jobId); + if (v !== undefined) { + resolve(v); + return; + } + } catch { + resolve(undefined); + return; + } + await new Promise((r) => setTimeout(r, 50)); + } + resolve(undefined); + })(); + }; + + const onFailed = (args: { jobId: string; failedReason: string }): void => { + cleanup(); + reject(new Error(args.failedReason || "job failed")); + }; + + const onTimeout = (): void => { + cleanup(); + reject( + new WaitUntilFinishedTimeoutError( + `Job.waitUntilFinished: no terminal event for ${jobId} after ${ttl}ms`, + ), + ); + }; + + const cleanup = (): void => { + if (timer !== undefined) clearTimeout(timer); + queueEvents.removeListener(completedChannel, onCompleted); + queueEvents.removeListener(failedChannel, onFailed); + }; + + queueEvents.once(completedChannel, onCompleted); + queueEvents.once(failedChannel, onFailed); + if (ttl !== undefined && ttl > 0) { + timer = setTimeout(onTimeout, ttl); + } + }); + } + toJSON(): object { return { id: this.id, diff --git a/chasquimq-node/src-ts/queue-events.ts b/chasquimq-node/src-ts/queue-events.ts index 5d6d382..31cc824 100644 --- a/chasquimq-node/src-ts/queue-events.ts +++ b/chasquimq-node/src-ts/queue-events.ts @@ -171,15 +171,35 @@ export class QueueEvents extends EventEmitter { case 'waiting': this.emit('waiting', { jobId, name }, eventId) break - case 'active': - this.emit('active', { jobId, name, prev: 'waiting', attempt: parseIntSafe(f['attempt']) }, eventId) + case 'active': { + const attempt = parseIntSafe(f['attempt']) + this.emit('active', { jobId, name, prev: 'waiting', attempt }, eventId) + // Per-job channel for `Job.waitUntilFinished` and other targeted + // subscribers. Same payload shape as the broadcast event; the + // narrower event name keeps the targeted-listener path off the + // O(N-listeners) dispatch for the broadcast channel. + if (jobId) this.emit(`active:${jobId}`, { jobId, name, prev: 'waiting', attempt }, eventId) break - case 'completed': - this.emit('completed', { jobId, name, attempt: parseIntSafe(f['attempt']), returnvalue: undefined }, eventId) + } + case 'completed': { + const attempt = parseIntSafe(f['attempt']) + // `returnvalue` is intentionally `undefined` — the events stream + // does not carry the handler's return bytes (that would + // double-allocate the payload onto every subscriber). Callers + // that need the value should pair this with `Queue.getJobResult` + // (requires `WorkerOptions.storeResults = true`), which is what + // `Job.waitUntilFinished` does internally. + this.emit('completed', { jobId, name, attempt, returnvalue: undefined }, eventId) + if (jobId) this.emit(`completed:${jobId}`, { jobId, name, attempt, returnvalue: undefined }, eventId) break - case 'failed': - this.emit('failed', { jobId, name, failedReason: f['reason'] ?? '', attempt: parseIntSafe(f['attempt']) }, eventId) + } + case 'failed': { + const attempt = parseIntSafe(f['attempt']) + const failedReason = f['reason'] ?? '' + this.emit('failed', { jobId, name, failedReason, attempt }, eventId) + if (jobId) this.emit(`failed:${jobId}`, { jobId, name, failedReason, attempt }, eventId) break + } case 'retry-scheduled': // chasquimq-specific extension event; advanced subscribers use this // to observe retry scheduling decisions before they fire. diff --git a/chasquimq-node/src-ts/worker.ts b/chasquimq-node/src-ts/worker.ts index a38f7a3..03161a7 100644 --- a/chasquimq-node/src-ts/worker.ts +++ b/chasquimq-node/src-ts/worker.ts @@ -34,6 +34,7 @@ import { type Job as NativeJob, } from '../index.js' import { Job } from './job.js' +import { QueueEvents } from './queue-events.js' import type { ConnectionOptions, JobsOptions } from './types.js' import { encodePayload } from './encoding.js' import { NotSupportedError } from './errors.js' @@ -151,6 +152,10 @@ export interface WorkerOptions { * * ## Events * + * Event names follow the familiar `EventEmitter` listener convention so + * existing application code reads naturally. ChasquiMQ-specific events + * are clearly marked. + * * - `ready` — `()`. Fired once when `.run()` starts the engine loop. * - `active` — `(job: Job, prev: string)`. Fired before each * processor invocation. `prev` is reserved (always `''`). @@ -163,6 +168,27 @@ export interface WorkerOptions { * from the native loop. * - `closing` — `(msg: string)`. Fired at the start of `.close()`. * - `closed` — `()`. Fired once shutdown completes. + * - `drained` — `()`. Fired when the engine observes a full→empty + * transition on the main stream (no more jobs to dispatch right now). + * Lazily subscribes to the cross-process events stream on the first + * `worker.on('drained', ...)` call; the subscriber is torn down on + * `.close()`. **Cross-process scope:** every worker on this queue + * receives `drained`, not just this one. + * - `paused` — `()`. Fired when `.pause()` is called. Process-local + * (does not reflect a cross-process `Queue.pause`). + * - `resumed` — `()`. Fired when `.resume()` is called. Process-local. + * + * ## Listener names accepted for API stability but currently no-op + * + * These can be wired up without throwing, but the engine does not emit + * the underlying transition yet. Listed here so subscriber code keeps + * type-checking; promoted to active events when the corresponding + * engine work lands. + * + * - `progress` — `(job: Job, progress: JobProgress)`. Requires engine + * support for in-handler progress updates. Not yet emitted. + * - `stalled` — `(jobId: string, prev: string)`. Requires a stalled- + * detector in the engine. Not yet emitted. */ export class Worker< DataType = unknown, @@ -177,6 +203,21 @@ export class Worker< private running = false private closed = false private runPromise?: Promise + /** + * Lazily-constructed events-stream subscriber that fans the engine's + * cross-process `drained` event onto this worker's `EventEmitter`. + * Created the first time a listener attaches to `'drained'`; torn + * down in `.close()`. Workers that never subscribe to `drained` pay + * no extra Redis connections. + */ + private drainedEvents?: QueueEvents + /** + * Resolves once the lazy drained subscriber has issued its first + * `XREAD BLOCK`. `run()` awaits this (when set) so a user pattern of + * `worker.on('drained', cb); queue.add(...)` doesn't race the + * engine's first emit against the subscriber's connect+block. + */ + private drainedReadyPromise?: Promise constructor( name: string, @@ -216,6 +257,16 @@ export class Worker< // every reconnect / AUTH cycle. this.native = new NativeConsumer(url, nativeOpts, opts.connection.credentialProvider) + // Spawn the cross-process events-stream subscriber the first time + // a user attaches a `drained` listener. Using `newListener` (not a + // public method) keeps the API surface plain `EventEmitter`-shaped — + // users just call `worker.on('drained', ...)`, no extra setup. + this.on('newListener', (event: string) => { + if (event === 'drained' && !this.drainedEvents && !this.closed) { + this.spawnDrainedSubscriber() + } + }) + if (opts.autorun !== false) { // Defer to the next microtask so subscribers can attach listeners // (`worker.on('completed', ...)`, etc.) before the first event fires. @@ -225,6 +276,88 @@ export class Worker< } } + /** + * Lazily start a {@link QueueEvents} subscriber that forwards the + * engine's cross-process `drained` event onto this worker's + * `EventEmitter`. Idempotent — calling twice has no effect (the first + * call wins). Errors from the subscriber are forwarded to this + * worker's `error` channel so application code only needs one error + * subscription. + */ + private spawnDrainedSubscriber(): void { + if (this.drainedEvents) return + // `autorun: false` + explicit `await waitUntilReady` + explicit + // `run()` lets us hold the worker's own `run()` until the + // subscriber's first `XREAD BLOCK` is in flight. Without this, the + // engine's first `drained` emit (which fires within a few hundred + // ms of worker startup on a fresh queue) can race the subscriber's + // connect+block and the event is lost. `lastEventId: '$'` keeps + // the subscriber from replaying ancient events from a long-lived + // queue; the race window we close is the connect-and-block latency + // only. + // `blockingTimeout: 1000` (vs the QueueEvents default 10s) keeps + // `worker.close()` snappy — close awaits the in-flight `XREAD + // BLOCK` to time out plus 1s grace. A 10s block-and-wait would + // mean every worker shutdown drags for up to 11s before the + // `closed` event fires; 1s + 1s grace = ~2s worst-case teardown. + const events = new QueueEvents(this.name, { + connection: this.opts.connection, + autorun: false, + blockingTimeout: 1000, + }) + events.on('drained', () => { + this.emit('drained') + }) + // Forward subscriber errors onto the worker's single ``error`` + // channel so application code only needs one error subscription. + // Caller MUST wire a ``worker.on('error', ...)`` listener — an + // unhandled ``error`` emit crashes the Node process, same as for + // any other ``EventEmitter``. + events.on('error', (err: Error) => { + if (!this.closed) { + this.emit('error', err) + } + }) + this.drainedEvents = events + // Capture a ready promise that `run()` awaits before kicking the + // native engine. `waitUntilReady()` establishes the ioredis + // connection; the subsequent `run()` queues the first `XREAD + // BLOCK` synchronously into ioredis's command queue, so once + // `events.run()` has been invoked the subscription window is open + // even though the outer run promise won't resolve until close. + // Errors surface via the `error` forwarder above so the worker's + // single error channel stays canonical. + let resolveReady!: () => void + this.drainedReadyPromise = new Promise((res) => { + resolveReady = res + }) + void (async () => { + try { + await events.waitUntilReady() + // Fire-and-forget — the loop runs until close. The XREAD + // BLOCK lands in ioredis's send queue synchronously inside + // `events.run()`, so once we yield past the first microtask + // tick the subscriber is "live" for incoming events. + void events.run().catch((err) => { + if (!this.closed) { + this.emit('error', err instanceof Error ? err : new Error(String(err))) + } + }) + // Yield once so the XREAD command flushes onto the socket + // before we release the worker's startup gate. + await new Promise((r) => setImmediate(r)) + resolveReady() + } catch (err) { + // Release the gate on connect failure so `run()` never hangs; + // the error itself surfaces via the `error` forwarder above. + resolveReady() + if (!this.closed) { + this.emit('error', err instanceof Error ? err : new Error(String(err))) + } + } + })() + } + /** * Start the engine loop. Resolves once the engine drains (after * `.close()` is called). Calling `.run()` more than once returns the @@ -235,6 +368,14 @@ export class Worker< this.running = true this.emit('ready') + // If a `drained` listener attached before `run()` was called, hold + // the engine start until the subscriber's `XREAD BLOCK` is in + // flight. Best-effort: the gate releases on subscriber error too, + // so a Redis blip on the events stream cannot wedge the worker. + if (this.drainedReadyPromise) { + await this.drainedReadyPromise + } + const storeResults = this.opts.storeResults === true const handler = async ( nativeJob: NativeJob, @@ -315,6 +456,15 @@ export class Worker< /* swallow — already surfaced via 'error' */ } } + // Tear down the lazy drained subscriber if one was started. Best- + // effort: swallow errors so a transient Redis blip on close doesn't + // mask the worker's own shutdown path. + if (this.drainedEvents) { + await this.drainedEvents.close().catch(() => { + /* best-effort cleanup */ + }) + this.drainedEvents = undefined + } this.running = false this.emit('closed') } @@ -340,12 +490,17 @@ export class Worker< * being processed run to completion; no new jobs are dispatched until * {@link Worker.resume}. Process-local (does not write the cross-process * Redis flag — use {@link Queue.pause} for queue-wide durable pause). - * Idempotent. The `doNotWaitActive` argument is accepted for BullMQ - * call-shape parity but is a no-op: this method returns immediately and + * Idempotent. The `doNotWaitActive` argument is accepted for call-shape + * stability but is a no-op: this method returns immediately and * in-flight jobs always drain in the background. */ async pause(_doNotWaitActive = false): Promise { this.native.pause() + // Emit AFTER trip so a listener firing `pause()` synchronously + // observes consistent state: `worker.isPaused() === true` by the + // time `paused` fires. Process-local — fired from the local Worker, + // not from the cross-process Redis pause flag. + this.emit('paused') } /** @@ -354,6 +509,7 @@ export class Worker< */ resume(): void { this.native.resume() + this.emit('resumed') } /** diff --git a/chasquimq-py/README.md b/chasquimq-py/README.md index 9eb5e1f..fab2aee 100644 --- a/chasquimq-py/README.md +++ b/chasquimq-py/README.md @@ -65,9 +65,9 @@ asyncio.run(main()) | Surface | What it does | |---|---| | `Queue` | Producer + queue inspection. `add` / `add_bulk` / `add_unique` / `get_job` / `get_jobs` / `get_jobs_page` / `get_job_state` / `get_job_counts` / `get_waiting_count` / `get_active_count` / `get_delayed_count` / `get_completed_count` / `get_failed_count` / `count` / `get_job_result` / `peek_dlq` / `replay_dlq` / `cancel_delayed` / `get_repeatable_jobs` / `remove_repeatable_by_key` / `pause` / `resume` / `is_paused` / `remove` / `remove_report` / `drain` / `clean` / `obliterate`. Async context manager. | -| `Worker` | Consumer pool. asyncio-first dispatch, opt-in result storage (`store_results=True`), graceful shutdown, `pause` / `resume` / `is_paused`. Async context manager. | -| `Job` | Frozen dataclass returned by `Queue.add`. Has `id`, `name`, `data`, `attempts_made`, `wait_for_result(timeout=)`. | -| `QueueEvents` | Asyncio iterator over the engine events stream. Cross-process pub/sub for `completed` / `failed` / `dlq` / `retry-scheduled` / `delayed`. | +| `Worker` | Consumer pool. asyncio-first dispatch, opt-in result storage (`store_results=True`), graceful shutdown, `pause` / `resume` / `is_paused`, listener API (`on`/`off`/`once` for `ready` / `active` / `completed` / `failed` / `error` / `closing` / `closed` / `drained` / `paused` / `resumed`). Async context manager. | +| `Job` | Dataclass returned by `Queue.add`. Has `id`, `name`, `data`, `attempts_made`, `wait_for_result(timeout=)`, `wait_until_finished(queue_events, timeout=)`. | +| `QueueEvents` | Async-iterator + listener API over the engine events stream. Cross-process pub/sub for `waiting` / `active` / `completed` / `failed` / `dlq` / `retry-scheduled` / `delayed` / `drained` / `retries-exhausted`, plus per-id channels (`completed:` / `failed:` / `active:`) for targeted subscribers. | | `BackoffSpec` | Builders: `.fixed(delay_ms)` / `.exponential(initial_ms, multiplier, max_ms, jitter_ms)`. | | `RepeatPattern` | Builders: `.cron(expr, tz=)` / `.every(interval_ms)`. DST-aware via IANA tz names. | | `MissedFiresPolicy` | `.skip()` / `.fire_once()` / `.fire_all(max_catchup)` for cron catch-up after scheduler downtime. | @@ -215,6 +215,92 @@ async with Queue("emails", redis_url=url) as queue: The same durable flag is what the CLI's `chasqui pause ` / `chasqui resume ` toggle. Both surfaces are idempotent — double-pause / double-resume are no-ops. +### Subscribing to events + +Two layered surfaces: per-worker listeners for in-process observation, and the cross-process `QueueEvents` stream for fan-out across workers, dashboards, and ops tooling. Listener callbacks may be plain functions or `async def` coroutines — async callbacks are scheduled on the running loop. + +In-process `Worker` events fire on the local worker only: + +```python +async with Worker("emails", handler, redis_url=url) as worker: + worker.on("ready", lambda: print("engine loop started")) + worker.on("active", lambda job: print("running", job.id)) + worker.on("completed", lambda job, result: print("ok", job.id, result)) + worker.on("failed", lambda job, err: print("failed", job.id, err)) + worker.on("error", lambda err: print("engine error", err)) + worker.on("drained", lambda: print("queue is empty")) + worker.on("paused", lambda: print("worker parked")) + worker.on("resumed", lambda: print("worker resumed")) + worker.on("closing", lambda: print("shutting down")) + worker.on("closed", lambda: print("shutdown complete")) + await worker.run() +``` + +`drained` is the only `Worker` event that requires a cross-process subscription (the engine emits it on the events stream, since "no jobs left" is a queue-wide observation). It's lazily wired: the first `worker.on('drained', ...)` call spawns an embedded `QueueEvents` subscriber, torn down on `worker.close()`. Workers that never subscribe pay no extra Redis connections. + +Cross-process `QueueEvents` listens on the events stream — every process running a `QueueEvents` instance for the same queue sees every transition, regardless of which worker emitted it. Two surfaces are available; pick one (they share the same Redis connection but only one consumer of XREAD at a time): + +```python +from chasquimq import QueueEvents + +# Listener API (EventEmitter-shaped) +events = QueueEvents("emails", redis_url=url) +events.on("waiting", lambda payload, eid: ...) +events.on("active", lambda payload, eid: ...) +events.on("completed", lambda payload, eid: ...) +events.on("failed", lambda payload, eid: ...) # payload['failedReason'] +events.on("drained", lambda eid: ...) +events.on("retry-scheduled", lambda payload, eid: ...) +events.on("dlq", lambda payload, eid: ...) +events.on("retries-exhausted", lambda payload, eid: ...) +# Per-id channels — fire only for the named job. Used internally by +# Job.wait_until_finished; useful for targeted UI updates without +# filtering every broadcast event by jobId. +events.on(f"completed:{job_id}", lambda payload, eid: ...) +events.on(f"failed:{job_id}", lambda payload, eid: ...) +await events.wait_until_ready() # subscriber's first XREAD BLOCK is in flight +# ... do work ... +await events.close() + +# Async-iterator surface (the original; one consumer) +events = QueueEvents("emails", redis_url=url) +async for ev in events: + print(ev.name, ev.job_id, ev.fields) +``` + +`completed` payloads from the events stream do **not** carry the handler's return value (that would double-allocate the payload onto every subscriber). To read the value, pair `QueueEvents` with `Queue.get_job_result(job_id)` and run the worker with `store_results=True`. The next section's `Job.wait_until_finished` does this for you. + +`progress` and `stalled` listeners are accepted on `Worker` for parity but are currently no-op — the engine doesn't emit those transitions yet. Attach them safely; they'll start firing when the corresponding engine work lands. + +### Awaiting a single job's completion + +`Job.wait_until_finished(queue_events, timeout=...)` is the event-driven completion-wait, mirroring the Node shim. Subscribes to the per-id `completed:` / `failed:` channels and resolves / raises on the first to fire: + +```python +events = QueueEvents("emails", redis_url=url) +# Force the subscriber loop to start so wait_until_ready is awaitable. +events.on("completed", lambda *_a: None) +await events.wait_until_ready() + +job = await queue.add("send", {"to": "ada@example.com"}) + +try: + # Returns the handler's return value (requires store_results=True + # on the worker; otherwise returns None). Raises + # RuntimeError(failedReason) on failure. + # `timeout` in seconds — omit for an unbounded wait. + result = await job.wait_until_finished(events, timeout=30.0) + print("sent:", result) +except WaitUntilFinishedTimeoutError: + print("no terminal event within deadline") +except RuntimeError as err: + print("job failed:", err) +finally: + await events.close() +``` + +Distinct from `Job.wait_for_result(timeout=...)`: that one polls the `Queue.get_job_result` Redis key and **requires** `store_results=True` to detect completion at all. `wait_until_finished` is event-driven, so it detects completion even when no result key was written — but it cannot tell you a job that already finished before the call wired up. Pick `wait_until_finished` for low-latency awaits of jobs you're about to enqueue; pick `wait_for_result` for "did this id ever finish". + ### Inspect jobs Read-only introspection across the queue's stream, delayed ZSET, DLQ, and result-key surfaces. Bounded scans; no secondary index; zero impact on the producer / consumer hot paths. diff --git a/chasquimq-py/src/chasquimq/__init__.py b/chasquimq-py/src/chasquimq/__init__.py index b9e5298..d4aed9b 100644 --- a/chasquimq-py/src/chasquimq/__init__.py +++ b/chasquimq-py/src/chasquimq/__init__.py @@ -14,7 +14,11 @@ _logging.getLogger("chasquimq").addHandler(_logging.NullHandler()) from ._native import Consumer, Producer, Scheduler, version -from .errors import NotSupportedError, UnrecoverableError +from .errors import ( + NotSupportedError, + UnrecoverableError, + WaitUntilFinishedTimeoutError, +) from .job import Job from .queue import Queue from .queue_events import QueueEvent, QueueEvents @@ -37,6 +41,7 @@ "RepeatableMeta", "Scheduler", "UnrecoverableError", + "WaitUntilFinishedTimeoutError", "Worker", "version", ] diff --git a/chasquimq-py/src/chasquimq/errors.py b/chasquimq-py/src/chasquimq/errors.py index 7d561b2..d79a501 100644 --- a/chasquimq-py/src/chasquimq/errors.py +++ b/chasquimq-py/src/chasquimq/errors.py @@ -23,3 +23,17 @@ class UnrecoverableError(RuntimeError): PoisonPill(UnrecoverableError): ...``) and still get the short-circuit behavior. """ + + +class WaitUntilFinishedTimeoutError(TimeoutError): + """Raised by :meth:`Job.wait_until_finished` when neither a + ``completed`` nor a ``failed`` event for the watched job arrives + within the supplied ``timeout`` seconds. + + Distinct from a failed job: a failed job raises a regular + :class:`RuntimeError` carrying the engine-reported ``failedReason``. + This error fires only when the events stream itself goes silent + (the worker died, the network blipped, or the ``timeout`` was too + short for the handler). Subclasses :class:`TimeoutError` so callers + that catch the broad timeout case still match. + """ diff --git a/chasquimq-py/src/chasquimq/job.py b/chasquimq-py/src/chasquimq/job.py index 04553de..10887f8 100644 --- a/chasquimq-py/src/chasquimq/job.py +++ b/chasquimq-py/src/chasquimq/job.py @@ -25,8 +25,11 @@ from dataclasses import dataclass, field from typing import TYPE_CHECKING, Any, Optional +from .errors import WaitUntilFinishedTimeoutError + if TYPE_CHECKING: from .queue import Queue + from .queue_events import QueueEvents @dataclass @@ -134,3 +137,114 @@ async def _loop() -> Any | None: await asyncio.sleep(poll_interval) return await asyncio.wait_for(_loop(), timeout=timeout) + + async def wait_until_finished( + self, + queue_events: "QueueEvents", + *, + timeout: Optional[float] = None, + ) -> Any | None: + """Event-driven completion wait. + + Subscribes to the engine's cross-process events stream and + resolves / raises when the ``completed`` or ``failed`` event + for **this** job fires. Mirrors the Node shim's + :meth:`Job.waitUntilFinished`. + + Unlike :meth:`wait_for_result`, this method is event-driven (no + polling, no Redis ``GET`` per interval) and does not require + ``Worker(store_results=True)`` to *detect* completion. It does, + however, require an attached :class:`QueueEvents` subscriber so + the events-stream traffic actually reaches this process. + + Return / raise semantics: + + * On ``completed``, returns the handler's return value when + ``store_results=True`` was set on the worker. The value is + fetched via ``Queue.get_job_result(self.id)`` after the event + fires. If ``store_results`` was not enabled (or the handler + returned ``None``), returns ``None``. The events stream + itself never carries the return value. + * On ``failed``, raises :class:`RuntimeError` carrying the + engine-reported ``failedReason`` (the same string surfaced + on the :class:`Worker`'s ``failed`` event). + * On ``timeout`` elapse, raises + :class:`WaitUntilFinishedTimeoutError`. + + **Race window.** If the job completed (or failed) *before* + this call wires up its listeners, the events-stream event has + already been dispatched and this method has nothing to + subscribe to. The ``timeout`` will fire normally. For producers + that want to await a job that may already have finished, pair + this with a :meth:`Queue.get_job_state` check, or use + :meth:`wait_for_result` (which can read a persisted result key + written before the wait started). + """ + # Surface a queue/queue_events mismatch up-front rather than + # letting the wait silently time out (the events stream is + # per-queue, so a ``QueueEvents`` for a different queue will + # never fire the per-id channel). Only checked when we have the + # queue backref; jobs constructed without one (the worker-side + # path, which has no queue handle) skip the guard. + if self._queue is not None and queue_events.name != self._queue.name: + raise ValueError( + f"Job.wait_until_finished: queue_events is for " + f"{queue_events.name!r} but this job is on " + f"{self._queue.name!r} — pass a QueueEvents subscribed " + f"to the right queue" + ) + loop = asyncio.get_running_loop() + fut: asyncio.Future[Any] = loop.create_future() + completed_channel = f"completed:{self.id}" + failed_channel = f"failed:{self.id}" + + def _on_completed(payload: dict, _event_id: str) -> None: + if fut.done(): + return + # Best-effort result fetch. The engine emits ``completed`` + # *before* the per-entry result write lands (events emit + # is off the ack hot path; result write is on it), so poll + # briefly. If ``store_results=False`` was set on the + # worker, every poll returns ``None`` and we resolve with + # ``None`` — same shape as handlers that return ``None``. + async def _resolve() -> None: + if fut.done(): + return + value: Any = None + if self._queue is not None: + for _ in range(10): + try: + v = await self._queue.get_job_result(self.id) + except Exception: + break + if v is not None: + value = v + break + await asyncio.sleep(0.05) + if not fut.done(): + fut.set_result(value) + + asyncio.ensure_future(_resolve()) + + def _on_failed(payload: dict, _event_id: str) -> None: + if fut.done(): + return + reason = payload.get("failedReason") or payload.get("reason") or "job failed" + fut.set_exception(RuntimeError(reason)) + + queue_events.on(completed_channel, _on_completed) + queue_events.on(failed_channel, _on_failed) + + try: + if timeout is None: + return await fut + try: + return await asyncio.wait_for(asyncio.shield(fut), timeout=timeout) + except asyncio.TimeoutError as exc: + raise WaitUntilFinishedTimeoutError( + f"Job.wait_until_finished: no terminal event for " + f"{self.id} after {timeout}s" + ) from exc + finally: + queue_events.off(completed_channel, _on_completed) + queue_events.off(failed_channel, _on_failed) diff --git a/chasquimq-py/src/chasquimq/queue_events.py b/chasquimq-py/src/chasquimq/queue_events.py index 0b98a2b..e2f3161 100644 --- a/chasquimq-py/src/chasquimq/queue_events.py +++ b/chasquimq-py/src/chasquimq/queue_events.py @@ -6,6 +6,22 @@ entries. :class:`QueueEvents` is the asyncio-friendly subscriber: it ``XREAD`` -blocks on that stream and yields :class:`QueueEvent` values. +Two surfaces are supported: + +* **Async iterator** — ``async for ev in events: ...`` — the original + Pythonic surface; one consumer, mutually exclusive with the listener + API. +* **Listener API** — ``events.on("completed", cb)`` — + ``EventEmitter``-style surface. Callbacks may be plain functions or + ``async def`` coroutines; async callbacks are scheduled on the + current loop. Per-id channels ``"completed:"`` / + ``"failed:"`` / ``"active:"`` let + :meth:`Job.wait_until_finished` target a single job without paying + the broadcast dispatch cost. + +The first ``.on(...)`` call lazily spawns an internal subscriber task; +:meth:`close` cancels it. + Implementation note: this uses ``redis-py`` directly rather than the native binding because the events stream is a generic Redis Stream (human-readable ASCII fields, not msgpack), so a thin async-redis @@ -16,14 +32,27 @@ from __future__ import annotations import asyncio +import inspect +import logging +from collections import defaultdict from dataclasses import dataclass -from typing import Any, AsyncIterator, Optional +from typing import Any, AsyncIterator, Awaitable, Callable, Optional, Union import redis.asyncio as aioredis from ._url import apply_tls +_log = logging.getLogger("chasquimq.queue_events") + +# A listener callback. May be sync (``Callable[..., Any]``) or async +# (``Callable[..., Awaitable[Any]]``). The dispatcher invokes both +# shapes correctly — sync callbacks fire inline (caller's exception +# only logs; never raises into the subscriber loop), async callbacks +# are scheduled on the running loop via ``asyncio.create_task``. +Listener = Callable[..., Union[Any, Awaitable[Any]]] + + @dataclass(frozen=True) class QueueEvent: """One event emitted by the engine. @@ -53,17 +82,29 @@ class QueueEvent: class QueueEvents: - """Subscribe to a queue's events stream as an async iterator. + """Subscribe to a queue's events stream. + + Two surfaces, pick one (they share the same Redis connection but + only one consumer of XREAD at a time): - Usage:: + Async iterator:: events = QueueEvents("emails") async for ev in events: print(ev.name, ev.job_id, ev.fields) + Listener API (``EventEmitter``-style):: + + events = QueueEvents("emails") + events.on("completed", lambda payload, event_id: print(payload)) + events.on(f"completed:{job_id}", on_one_completed) # targeted + # ... do work ... + await events.close() + The subscriber starts from ``$`` (only events emitted after the iterator is opened) by default — pass ``last_event_id="0"`` to - replay history. Iteration ends when :meth:`close` is called. + replay history. Iteration / dispatch ends when :meth:`close` is + called. """ def __init__( @@ -83,11 +124,24 @@ def __init__( self._block_ms = block_ms self._count = count self._closed = False + # Listener API state. Two-tier map: event-name → list of + # callbacks; per-id channels live in the same dict under the + # qualified key (``"completed:"``). Plain dict is fine — + # all mutation is on the asyncio loop thread. + self._listeners: dict[str, list[Listener]] = defaultdict(list) + # Backing task for the listener-API subscriber loop. Spawned + # lazily on the first ``on(...)`` call; ``None`` when no + # listeners are attached so workers that only use the iterator + # API pay no extra task. + self._listener_task: Optional[asyncio.Task[None]] = None + self._listener_ready: Optional[asyncio.Future[None]] = None @property def name(self) -> str: return self._queue_name + # --- Async iterator surface --------------------------------------------- + def __aiter__(self) -> AsyncIterator[QueueEvent]: return self._iterate() @@ -121,15 +175,223 @@ async def _iterate(self) -> AsyncIterator[QueueEvent]: if self._closed: return + # --- Listener API surface ----------------------------------------------- + + def on(self, event_name: str, callback: Listener) -> None: + """Register a callback for a named event. + + ``event_name`` accepts both broadcast event names (``"waiting"`` + / ``"active"`` / ``"completed"`` / ``"failed"`` / + ``"retry-scheduled"`` / ``"delayed"`` / ``"dlq"`` / ``"drained"`` + / ``"retries-exhausted"``) and per-id qualified names + (``"completed:"`` / ``"failed:"`` / + ``"active:"``) for :meth:`Job.wait_until_finished` and + other targeted subscribers. + + Callbacks may be plain functions or ``async def`` coroutines; + async callbacks are scheduled on the running loop. A raised + exception from a sync callback is logged and swallowed — the + subscriber loop survives. The callback is invoked with one + argument for queue-scoped events (``drained``) and two + arguments for per-job events (``payload, event_id``); mirrors + the Node shim's ``EventEmitter`` arity. + + The first ``on(...)`` call lazily starts the subscriber task — + prefer attaching listeners before producing work, since events + emitted before the first ``XREAD BLOCK`` lands are lost (same + race window as the Node shim). + """ + self._listeners[event_name].append(callback) + if self._listener_task is None and not self._closed: + self._spawn_listener_task() + + def off(self, event_name: str, callback: Listener) -> None: + """Remove a previously-registered callback. + + Removes the first matching callback only; passing the same + callback twice removes both copies via two ``off`` calls. No + error if the callback is not registered. + """ + listeners = self._listeners.get(event_name) + if not listeners: + return + try: + listeners.remove(callback) + except ValueError: + pass + if not listeners: + self._listeners.pop(event_name, None) + + remove_listener = off + """Alias matching the Node ``EventEmitter`` naming.""" + + def once(self, event_name: str, callback: Listener) -> None: + """Register a callback that fires exactly once and then removes + itself. + """ + # Bind the callback as a closure so the wrapper is its own + # identity for ``off`` lookup. + async def _wrapper(*args: Any, **kwargs: Any) -> None: + self.off(event_name, _wrapper) + res = callback(*args, **kwargs) + if inspect.isawaitable(res): + await res + + self.on(event_name, _wrapper) + + def listener_count(self, event_name: Optional[str] = None) -> int: + """Total number of registered callbacks, optionally for a + single event name. Used in tests to assert wiring; users + normally won't need this. + """ + if event_name is not None: + return len(self._listeners.get(event_name, [])) + return sum(len(v) for v in self._listeners.values()) + + def _spawn_listener_task(self) -> None: + # Capture a ready future so callers (notably :class:`Worker`'s + # internal drained subscriber) can await the first XREAD + # before producing work. The future resolves once we've issued + # the first BLOCK call so the subscription window is open. + # Uses ``get_running_loop`` (not ``get_event_loop``) so a + # sync call from outside a coroutine surfaces as a clear + # ``RuntimeError`` rather than silently binding to a stale or + # implicit loop — matches the existing shim's "must be inside + # an async context" posture. + loop = asyncio.get_running_loop() + self._listener_ready = loop.create_future() + self._listener_task = asyncio.ensure_future(self._listener_loop()) + + async def wait_until_ready(self) -> None: + """Block until the listener subscriber has issued its first + ``XREAD BLOCK`` (i.e. the subscription window is open). No-op + when no listeners are attached. + """ + if self._listener_ready is not None: + await self._listener_ready + + async def _listener_loop(self) -> None: + last_id = self._last_id + first = True + while not self._closed: + try: + xread_task = asyncio.ensure_future( + self._client.xread( + {self._stream_key: last_id}, + count=self._count, + block=self._block_ms, + ) + ) + if first: + # Yield once so the XREAD command flushes onto the + # socket (redis-py uses an async writer that needs + # one loop turn to drain its buffer), then release + # the ready gate. After this point the subscription + # window is open from Redis's perspective. + await asyncio.sleep(0) + if ( + self._listener_ready is not None + and not self._listener_ready.done() + ): + self._listener_ready.set_result(None) + first = False + res = await xread_task + except asyncio.CancelledError: + raise + except Exception as exc: + if self._closed: + return + _log.warning("QueueEvents listener loop error: %s", exc) + await asyncio.sleep(0.2) + continue + + if not res: + continue + + for _stream_key_bytes, entries in res: + for entry_id_bytes, fields_bytes in entries: + entry_id = _to_str(entry_id_bytes) + last_id = entry_id + self._last_id = entry_id + ev = _parse_event(fields_bytes) + self._dispatch(ev, entry_id) + if self._closed: + return + + def _dispatch(self, ev: QueueEvent, event_id: str) -> None: + # Broadcast handlers first. + broadcast = list(self._listeners.get(ev.name, ())) + # Per-id targeted handlers for events that carry a job id. + targeted: list[Listener] = [] + if ev.job_id is not None and ev.name in _PER_ID_EVENTS: + targeted = list(self._listeners.get(f"{ev.name}:{ev.job_id}", ())) + if not broadcast and not targeted: + return + + # Payload shape mirrors the Node shim: a single dict argument + # for per-job events, ``(event_id,)`` for queue-scoped events + # (``drained``). + if ev.job_id is None: + args: tuple[Any, ...] = (event_id,) + else: + payload = { + "jobId": ev.job_id, + "name": ev.job_name, + **ev.fields, + } + # Map the engine's ``reason`` field to ``failedReason`` as + # well, so subscribers using the camelCase event payload key + # see it under both names. + if ev.name == "failed" and "reason" in ev.fields: + payload["failedReason"] = ev.fields["reason"] + args = (payload, event_id) + + for cb in broadcast + targeted: + self._invoke(cb, args) + + def _invoke(self, cb: Listener, args: tuple[Any, ...]) -> None: + try: + result = cb(*args) + except Exception as exc: + # Sync callbacks raising must not kill the subscriber loop. + _log.warning( + "QueueEvents listener raised; swallowing: %s", exc, exc_info=True + ) + return + if inspect.isawaitable(result): + # Async callbacks are scheduled on the running loop. We + # attach a done-callback so an unhandled rejection + # surfaces in the log rather than as a silent + # ``Task exception was never retrieved`` warning. + task = asyncio.ensure_future(result) + task.add_done_callback(_log_task_exception) + async def close(self) -> None: - """Stop iteration and release the Redis connection.""" + """Stop iteration / dispatch and release the Redis connection.""" self._closed = True + if self._listener_task is not None: + self._listener_task.cancel() + try: + await self._listener_task + except (asyncio.CancelledError, Exception): + pass + self._listener_task = None + if self._listener_ready is not None and not self._listener_ready.done(): + self._listener_ready.set_exception(RuntimeError("QueueEvents closed")) try: await self._client.aclose() except Exception: pass +def _log_task_exception(task: asyncio.Task[Any]) -> None: + if task.cancelled(): + return + exc = task.exception() + if exc is not None: + _log.warning("QueueEvents async listener task failed: %s", exc, exc_info=exc) + + def _to_str(v: Any) -> str: if isinstance(v, bytes): try: @@ -148,6 +410,14 @@ def _to_str(v: Any) -> str: {"attempt", "backoff_ms", "delay_ms", "duration_us", "ts"} ) +# Events for which we emit a per-id targeted channel +# (``":"``) in addition to the broadcast channel. +# Mirrors the Node shim's `active:`, `completed:`, `failed:` per-id +# emit. The targeted channels exist specifically so +# :meth:`Job.wait_until_finished` doesn't have to filter every +# broadcast event by jobId. +_PER_ID_EVENTS: frozenset[str] = frozenset({"active", "completed", "failed"}) + def _maybe_int(s: str) -> Any: try: diff --git a/chasquimq-py/src/chasquimq/worker.py b/chasquimq-py/src/chasquimq/worker.py index b92283b..f3a6314 100644 --- a/chasquimq-py/src/chasquimq/worker.py +++ b/chasquimq-py/src/chasquimq/worker.py @@ -12,16 +12,28 @@ from __future__ import annotations import asyncio -from typing import Any, Awaitable, Callable, Optional, Tuple +import inspect +import logging +from collections import defaultdict +from typing import Any, Awaitable, Callable, Optional, Tuple, Union from . import _native from ._encoding import decode_payload, encode_payload from ._url import apply_tls from .job import Job +from .queue_events import QueueEvents + + +_log = logging.getLogger("chasquimq.worker") Handler = Callable[[Job], Awaitable[Any]] + +WorkerListener = Callable[..., Union[Any, Awaitable[Any]]] +"""A callback registered via :meth:`Worker.on`. May be sync or +``async def``; async callbacks are scheduled on the running loop.""" + CredentialProvider = Callable[ [Optional[str]], Awaitable[Tuple[Optional[str], Optional[str]]] ] @@ -40,6 +52,40 @@ class Worker: headline throughput target. Pass ``concurrency=1`` explicitly when serial processing is required (e.g. handlers that mutate shared state without their own synchronization). + + ## Events + + Event names mirror the Node shim's ``Worker`` listener interface, + so existing application code reads naturally across languages. + Subscribe with :meth:`on`; callbacks may be plain functions or + ``async def`` coroutines. + + * ``ready`` — ``()``. Fired once when :meth:`run` starts the + engine loop. + * ``active`` — ``(job: Job)``. Fired before each handler + invocation. + * ``completed`` — ``(job: Job, result)``. Fired after the handler + returns. The engine acks the job. + * ``failed`` — ``(job: Job, err: BaseException)``. Fired after + the handler raises. The exception is re-raised so the engine + routes the job to retry-or-DLQ. + * ``error`` — ``(err: BaseException)``. Fired on engine-side + errors surfaced from the native loop or the drained subscriber. + * ``closing`` — ``()``. Fired at the start of :meth:`close`. + * ``closed`` — ``()``. Fired once shutdown completes. + * ``drained`` — ``()``. Fired when the engine observes a + full→empty transition on the main stream. Lazily subscribes to + the cross-process events stream on the first ``on('drained', + ...)`` call; the subscriber is torn down on :meth:`close`. + **Cross-process scope:** every worker on this queue receives + ``drained``, not just this one. + * ``paused`` — ``()``. Fired when :meth:`pause` is called. + * ``resumed`` — ``()``. Fired when :meth:`resume` is called. + + Listener names accepted for parity but currently no-op (engine + doesn't emit the underlying transition yet): ``progress``, + ``stalled``. These can be wired up without raising; promoted to + active events when the corresponding engine work lands. """ def __init__( @@ -126,6 +172,18 @@ def __init__( # Pause intent recorded before the (deferred) native consumer # exists; applied once it is constructed in ``run()``. self._pending_paused = False + # Listener API. Mirrors :class:`QueueEvents`'s shape so the + # mental model is one. Sync and async callbacks both work; an + # exception from a sync callback is logged and swallowed so a + # buggy listener cannot crash the worker. + self._listeners: dict[str, list[WorkerListener]] = defaultdict(list) + # Lazy embedded :class:`QueueEvents` subscriber for the + # cross-process ``drained`` event. ``None`` until a ``drained`` + # listener attaches; torn down in :meth:`close`. Workers that + # never subscribe to ``drained`` pay no extra Redis connections. + self._drained_events: Optional[QueueEvents] = None + self._drained_redis_url = redis_url + self._drained_tls = tls @property def name(self) -> str: @@ -159,6 +217,18 @@ async def run(self) -> None: self._consumer.pause() self._running = True + self._emit("ready") + + # If a ``drained`` listener attached before ``run()`` was + # called, hold the engine start until the subscriber's first + # ``XREAD BLOCK`` is in flight. Best-effort: cancellation / + # error on the subscriber side resolves the wait so the + # worker startup is never wedged. + if self._drained_events is not None: + try: + await self._drained_events.wait_until_ready() + except Exception: + pass async def native_handler(native_job: Any) -> Optional[bytes]: data = decode_payload(bytes(native_job.payload)) @@ -169,7 +239,28 @@ async def native_handler(native_job: Any) -> Optional[bytes]: attempt=native_job.attempt, created_at_ms=native_job.created_at_ms, ) - result = await self._handler(job) + # ``active`` fires before the handler runs so subscribers + # building a "currently running" view see jobs even for + # long-running handlers. Mirrors the Node shim. + self._emit("active", job) + try: + result = await self._handler(job) + except asyncio.CancelledError: + # Cancellation (from shutdown / shield bypass) is a + # control-flow signal, not a handler failure. The + # engine treats the cancelled handler as in-progress + # at shutdown; no ``failed`` event fires (a cancelled + # handler is not a handler failure). + raise + except BaseException as exc: + # ``failed`` fires before re-raising so subscribers see + # the exception that triggered the routing decision + # (retry vs. DLQ-unrecoverable). The native binding + # detects ``UnrecoverableError`` via MRO; this emit + # path stays agnostic. + self._emit("failed", job, exc) + raise + self._emit("completed", job, result) if result is None: return None return encode_payload(result) @@ -180,6 +271,17 @@ async def native_handler(native_job: Any) -> Optional[bytes]: try: await self._consumer_task + except asyncio.CancelledError: + # External cancellation (test teardown, shutdown via + # ``close``). Not an engine error — propagate without + # firing the ``error`` channel so subscribers don't see + # spurious shutdown noise. + raise + except BaseException as exc: + # Engine-side errors surfaced from the native loop. Mirrors + # the Node shim's ``error`` channel. + self._emit("error", exc) + raise finally: self._running = False @@ -195,12 +297,23 @@ async def close(self) -> None: if self._closed: return self._closed = True + self._emit("closing") # ``close()`` may be called before ``run()`` (e.g. an aborted # async-context-manager path). When a credential_provider # deferred construction, the native consumer may not exist yet — # there is nothing to drain, so just flag closed. if self._consumer is not None: self._consumer.shutdown() + # Tear down the lazy drained subscriber if one was started. + # Best-effort: swallow errors so a transient Redis blip on + # close doesn't mask the worker's own shutdown path. + if self._drained_events is not None: + try: + await self._drained_events.close() + except Exception: + pass + self._drained_events = None + self._emit("closed") def pause(self) -> None: """Pause this worker's reader at the next batch boundary. @@ -218,6 +331,11 @@ def pause(self) -> None: self._consumer.pause() else: self._pending_paused = True + # Emit AFTER trip so a listener firing :meth:`pause` + # synchronously observes consistent state: + # :meth:`is_paused` returns ``True`` by the time ``paused`` + # fires. Process-local; mirrors the Node shim. + self._emit("paused") def resume(self) -> None: """Resume a paused worker. The reader wakes immediately (no @@ -226,6 +344,7 @@ def resume(self) -> None: self._pending_paused = False if self._consumer is not None: self._consumer.resume() + self._emit("resumed") def is_paused(self) -> bool: """Whether this worker is paused via :meth:`pause`. Does not @@ -243,8 +362,125 @@ def is_running(self) -> bool: def is_closed(self) -> bool: return self._closed + # --- Listener API surface ----------------------------------------------- + + def on(self, event_name: str, callback: WorkerListener) -> None: + """Register a callback for a worker event. + + See the class docstring for the supported event names. Callbacks + may be plain functions or ``async def`` coroutines. Listeners + for ``'drained'`` lazily spawn an internal cross-process + events-stream subscriber the first time one attaches; the + subscriber is torn down in :meth:`close`. + """ + self._listeners[event_name].append(callback) + if ( + event_name == "drained" + and self._drained_events is None + and not self._closed + ): + self._spawn_drained_subscriber() + + def off(self, event_name: str, callback: WorkerListener) -> None: + """Remove a previously-registered callback. + + Removes the first matching callback only; passing the same + callback twice removes both copies via two ``off`` calls. No + error if the callback is not registered. + """ + listeners = self._listeners.get(event_name) + if not listeners: + return + try: + listeners.remove(callback) + except ValueError: + pass + if not listeners: + self._listeners.pop(event_name, None) + + remove_listener = off + """Alias matching the Node ``EventEmitter`` naming.""" + + def once(self, event_name: str, callback: WorkerListener) -> None: + """Register a callback that fires exactly once and then removes + itself. + """ + async def _wrapper(*args: Any, **kwargs: Any) -> None: + self.off(event_name, _wrapper) + res = callback(*args, **kwargs) + if inspect.isawaitable(res): + await res + + self.on(event_name, _wrapper) + + def listener_count(self, event_name: Optional[str] = None) -> int: + if event_name is not None: + return len(self._listeners.get(event_name, [])) + return sum(len(v) for v in self._listeners.values()) + + def _emit(self, event_name: str, *args: Any) -> None: + listeners = self._listeners.get(event_name) + if not listeners: + return + # Snapshot before invoking so a ``once`` wrapper's ``off`` + # call during dispatch doesn't mutate the live list. + for cb in list(listeners): + try: + result = cb(*args) + except Exception as exc: + _log.warning( + "Worker listener for %r raised; swallowing: %s", + event_name, + exc, + exc_info=True, + ) + continue + if inspect.isawaitable(result): + task = asyncio.ensure_future(result) + task.add_done_callback(_log_listener_task_exception) + + def _spawn_drained_subscriber(self) -> None: + """Lazily start a :class:`QueueEvents` subscriber that forwards + the engine's cross-process ``drained`` event onto this worker's + listener registry. Idempotent. Errors from the subscriber are + forwarded to this worker's ``error`` channel so application + code only needs one error subscription. + + ``block_ms=1000`` keeps :meth:`close` snappy — the QueueEvents + default 5s would mean every worker shutdown drags for up to + 5s before ``closed`` fires; 1s + small grace ≈ 1s teardown. + """ + events = QueueEvents( + self._queue_name, + redis_url=self._drained_redis_url, + tls=self._drained_tls, + block_ms=1000, + ) + + def _on_drained(_event_id: str) -> None: + self._emit("drained") + + events.on("drained", _on_drained) + # Note: :class:`QueueEvents` does not currently expose an + # explicit ``error`` channel the way Node's ``EventEmitter`` + # does; transient subscriber errors are logged inside the + # ``_listener_loop``. If a future slice adds one, wire it onto + # this Worker's ``error`` emitter the same way the Node shim + # does. + self._drained_events = events + async def __aenter__(self) -> "Worker": return self async def __aexit__(self, exc_type, exc, tb) -> None: await self.close() + + +def _log_listener_task_exception(task: "asyncio.Task[Any]") -> None: + if task.cancelled(): + return + exc = task.exception() + if exc is not None: + _log.warning( + "Worker async listener task failed: %s", exc, exc_info=exc + ) diff --git a/chasquimq-py/tests/test_event_listeners.py b/chasquimq-py/tests/test_event_listeners.py new file mode 100644 index 0000000..4b4f534 --- /dev/null +++ b/chasquimq-py/tests/test_event_listeners.py @@ -0,0 +1,441 @@ +"""Integration tests for Worker / QueueEvents listener API + Job.wait_until_finished. + +Mirrors ``chasquimq-node/__test__/event-listeners.test.ts``. +""" + +from __future__ import annotations + +import asyncio +import os + +import pytest + +from chasquimq import ( + Job, + Queue, + QueueEvents, + WaitUntilFinishedTimeoutError, + Worker, +) + + +REDIS_URL = os.environ.get("CHASQUIMQ_TEST_REDIS_URL", "redis://127.0.0.1:6379") + + +pytestmark = pytest.mark.usefixtures("cleanup_keys") + + +async def _run_worker(worker: Worker) -> asyncio.Task[None]: + """Spawn the worker's ``run()`` loop and return the task. Swallows + cancellation so the test's ``finally`` cleanup is quiet. + """ + + async def _run() -> None: + try: + await worker.run() + except asyncio.CancelledError: + pass + except Exception: + pass + + return asyncio.ensure_future(_run()) + + +# --- Worker.on('drained') ----------------------------------------------------- + + +@pytest.mark.asyncio +async def test_worker_emits_drained_after_full_to_empty_transition( + redis_url: str, queue_name: str +) -> None: + drained_evt = asyncio.Event() + + async def handler(job: Job) -> None: + return None + + worker = Worker( + queue_name, + handler, + redis_url=redis_url, + concurrency=4, + read_block_ms=200, + delayed_enabled=False, + run_scheduler=False, + ) + worker.on("drained", lambda *_a: drained_evt.set()) + + queue = Queue(queue_name, redis_url=redis_url) + run_task = await _run_worker(worker) + try: + await queue.add("one", {"x": 1}) + await asyncio.wait_for(drained_evt.wait(), timeout=10.0) + finally: + await worker.close() + await asyncio.wait([run_task], timeout=5.0) + await queue.close() + + +@pytest.mark.asyncio +async def test_worker_does_not_spawn_drained_subscriber_when_no_listener( + redis_url: str, queue_name: str +) -> None: + async def handler(job: Job) -> None: + return None + + worker = Worker( + queue_name, + handler, + redis_url=redis_url, + concurrency=4, + read_block_ms=200, + delayed_enabled=False, + run_scheduler=False, + ) + queue = Queue(queue_name, redis_url=redis_url) + run_task = await _run_worker(worker) + try: + await queue.add("one", {"x": 1}) + # Give the worker a beat to process; no drained listener + # attached, so the embedded subscriber was never created. + await asyncio.sleep(0.2) + # Close should be fast — no embedded QueueEvents to drain. + loop = asyncio.get_event_loop() + t0 = loop.time() + await worker.close() + elapsed = loop.time() - t0 + # Bound generously: native consumer shutdown is fast (<1s); a + # live QueueEvents would add up to ~5s before close completes. + assert elapsed < 3.0 + finally: + await asyncio.wait([run_task], timeout=5.0) + await queue.close() + + +# --- Worker.on('paused' / 'resumed') ----------------------------------------- + + +@pytest.mark.asyncio +async def test_worker_emits_paused_and_resumed( + redis_url: str, queue_name: str +) -> None: + paused_calls: list[None] = [] + resumed_calls: list[None] = [] + + async def handler(job: Job) -> None: + return None + + worker = Worker( + queue_name, + handler, + redis_url=redis_url, + read_block_ms=200, + delayed_enabled=False, + run_scheduler=False, + ) + worker.on("paused", lambda: paused_calls.append(None)) + worker.on("resumed", lambda: resumed_calls.append(None)) + + run_task = await _run_worker(worker) + try: + # Let it start + await asyncio.sleep(0.05) + worker.pause() + assert len(paused_calls) == 1 + worker.resume() + assert len(resumed_calls) == 1 + finally: + await worker.close() + await asyncio.wait([run_task], timeout=5.0) + + +# --- Worker.on('completed' / 'failed') --------------------------------------- + + +@pytest.mark.asyncio +async def test_worker_emits_completed_with_result( + redis_url: str, queue_name: str +) -> None: + completed: list[tuple[Job, object]] = [] + + async def handler(job: Job) -> int: + return job.data["v"] * 3 + + worker = Worker( + queue_name, + handler, + redis_url=redis_url, + read_block_ms=200, + delayed_enabled=False, + run_scheduler=False, + ) + worker.on("completed", lambda job, result: completed.append((job, result))) + + queue = Queue(queue_name, redis_url=redis_url) + run_task = await _run_worker(worker) + try: + await queue.add("triple", {"v": 7}) + deadline = asyncio.get_event_loop().time() + 5.0 + while not completed and asyncio.get_event_loop().time() < deadline: + await asyncio.sleep(0.05) + assert len(completed) == 1 + assert completed[0][1] == 21 + finally: + await worker.close() + await asyncio.wait([run_task], timeout=5.0) + await queue.close() + + +@pytest.mark.asyncio +async def test_worker_emits_failed_with_exception( + redis_url: str, queue_name: str +) -> None: + failed: list[tuple[Job, BaseException]] = [] + + async def handler(job: Job) -> None: + raise RuntimeError("boom") + + worker = Worker( + queue_name, + handler, + redis_url=redis_url, + concurrency=1, + max_attempts=1, + read_block_ms=200, + delayed_enabled=False, + run_scheduler=False, + ) + worker.on("failed", lambda job, err: failed.append((job, err))) + + queue = Queue(queue_name, redis_url=redis_url) + run_task = await _run_worker(worker) + try: + await queue.add("fail-me", {"v": 0}) + deadline = asyncio.get_event_loop().time() + 5.0 + while not failed and asyncio.get_event_loop().time() < deadline: + await asyncio.sleep(0.05) + assert len(failed) >= 1 + assert isinstance(failed[0][1], RuntimeError) + assert "boom" in str(failed[0][1]) + finally: + await worker.close() + await asyncio.wait([run_task], timeout=5.0) + await queue.close() + + +# --- QueueEvents listener API + per-id channels ------------------------------ + + +@pytest.mark.asyncio +async def test_queue_events_per_id_completed_channel( + redis_url: str, queue_name: str +) -> None: + events = QueueEvents(queue_name, redis_url=redis_url, block_ms=300) + + async def handler(job: Job) -> int: + return job.data["v"] + + worker = Worker( + queue_name, + handler, + redis_url=redis_url, + read_block_ms=200, + delayed_enabled=False, + run_scheduler=False, + ) + queue = Queue(queue_name, redis_url=redis_url) + + run_task = await _run_worker(worker) + try: + # Add job; capture id, then attach per-id listener AFTER + # subscription is ready. + job = await queue.add("compute", {"v": 99}) + targeted_calls: list[dict] = [] + broadcast_calls: list[dict] = [] + + events.on(f"completed:{job.id}", lambda payload, eid: targeted_calls.append(payload)) + events.on("completed", lambda payload, eid: broadcast_calls.append(payload)) + await events.wait_until_ready() + + # Re-add to fire after the subscription is open (the earlier + # job may have completed before the subscriber connected). + job2 = await queue.add("compute", {"v": 100}) + # Targeted listener was for `job` (the earlier one), so wire + # one more for job2 too. + events.on(f"completed:{job2.id}", lambda payload, eid: targeted_calls.append(payload)) + + deadline = asyncio.get_event_loop().time() + 8.0 + while ( + not targeted_calls or not broadcast_calls + ) and asyncio.get_event_loop().time() < deadline: + await asyncio.sleep(0.05) + + assert broadcast_calls, "broadcast `completed` listener should fire" + assert targeted_calls, "per-id `completed:` listener should fire" + finally: + await events.close() + await worker.close() + await asyncio.wait([run_task], timeout=5.0) + await queue.close() + + +@pytest.mark.asyncio +async def test_queue_events_supports_async_callbacks( + redis_url: str, queue_name: str +) -> None: + events = QueueEvents(queue_name, redis_url=redis_url, block_ms=300) + seen = asyncio.Event() + + async def on_completed(payload: dict, _event_id: str) -> None: + # Async callback — small await to prove the dispatcher + # schedules it on the loop. + await asyncio.sleep(0) + seen.set() + + events.on("completed", on_completed) + await events.wait_until_ready() + + async def handler(job: Job) -> None: + return None + + worker = Worker( + queue_name, + handler, + redis_url=redis_url, + read_block_ms=200, + delayed_enabled=False, + run_scheduler=False, + ) + queue = Queue(queue_name, redis_url=redis_url) + run_task = await _run_worker(worker) + try: + await queue.add("async-cb", {}) + await asyncio.wait_for(seen.wait(), timeout=10.0) + finally: + await events.close() + await worker.close() + await asyncio.wait([run_task], timeout=5.0) + await queue.close() + + +# --- Job.wait_until_finished ------------------------------------------------- + + +@pytest.mark.asyncio +async def test_job_wait_until_finished_resolves_with_stored_result( + redis_url: str, queue_name: str +) -> None: + events = QueueEvents(queue_name, redis_url=redis_url, block_ms=300) + # Force the subscriber loop to start so `wait_until_ready` is + # awaitable. + events.on("completed", lambda *_a: None) + await events.wait_until_ready() + + async def handler(job: Job) -> int: + return job.data["v"] * 2 + + worker = Worker( + queue_name, + handler, + redis_url=redis_url, + read_block_ms=200, + delayed_enabled=False, + run_scheduler=False, + store_results=True, + result_ttl_ms=60_000, + ) + queue = Queue(queue_name, redis_url=redis_url) + run_task = await _run_worker(worker) + try: + job = await queue.add("double", {"v": 21}) + result = await job.wait_until_finished(events, timeout=10.0) + assert result == 42 + finally: + await events.close() + await worker.close() + await asyncio.wait([run_task], timeout=5.0) + await queue.close() + + +@pytest.mark.asyncio +async def test_job_wait_until_finished_returns_none_when_store_results_off( + redis_url: str, queue_name: str +) -> None: + events = QueueEvents(queue_name, redis_url=redis_url, block_ms=300) + events.on("completed", lambda *_a: None) + await events.wait_until_ready() + + async def handler(job: Job) -> None: + return None + + worker = Worker( + queue_name, + handler, + redis_url=redis_url, + read_block_ms=200, + delayed_enabled=False, + run_scheduler=False, + ) + queue = Queue(queue_name, redis_url=redis_url) + run_task = await _run_worker(worker) + try: + job = await queue.add("void", {}) + result = await job.wait_until_finished(events, timeout=10.0) + assert result is None + finally: + await events.close() + await worker.close() + await asyncio.wait([run_task], timeout=5.0) + await queue.close() + + +@pytest.mark.asyncio +async def test_job_wait_until_finished_raises_on_failed( + redis_url: str, queue_name: str +) -> None: + events = QueueEvents(queue_name, redis_url=redis_url, block_ms=300) + events.on("failed", lambda *_a: None) + await events.wait_until_ready() + + async def handler(job: Job) -> None: + raise RuntimeError("handler said no") + + worker = Worker( + queue_name, + handler, + redis_url=redis_url, + concurrency=1, + max_attempts=1, + read_block_ms=200, + delayed_enabled=False, + run_scheduler=False, + ) + queue = Queue(queue_name, redis_url=redis_url) + run_task = await _run_worker(worker) + try: + job = await queue.add("rejects", {}) + with pytest.raises(RuntimeError, match="handler said no"): + await job.wait_until_finished(events, timeout=10.0) + finally: + await events.close() + await worker.close() + await asyncio.wait([run_task], timeout=5.0) + await queue.close() + + +@pytest.mark.asyncio +async def test_job_wait_until_finished_times_out( + redis_url: str, queue_name: str +) -> None: + events = QueueEvents(queue_name, redis_url=redis_url, block_ms=300) + events.on("completed", lambda *_a: None) + await events.wait_until_ready() + + queue = Queue(queue_name, redis_url=redis_url) + try: + # No worker; the job sits waiting forever. + job = await queue.add("orphan", {}) + with pytest.raises(WaitUntilFinishedTimeoutError): + await job.wait_until_finished(events, timeout=0.3) + finally: + await events.close() + await queue.close() diff --git a/docs/history.md b/docs/history.md index 62b69de..c5a2e0b 100644 --- a/docs/history.md +++ b/docs/history.md @@ -314,6 +314,32 @@ Tests: 19 engine integration cases (live Redis) covering every state plus the id Doc surfaces synced: this slice, `docs/engine.md` (new "Job maintenance" section), both shim READMEs (symmetric maintenance sections), `site/src/content/docs/reference/{node-api,python-api,rust-api,cli}.md`, a new `guides/clean-and-obliterate.mdx` registered in `site/astro.config.mjs`, and the root `README.md` feature table. +## Event listeners on Worker, Queue, Job (cross-FFI slice) + +**BullMQ-shaped event-listener surface shipped, cross-FFI, May 2026.** Closes the parity gap for event-driven consumption. The engine already emits a cross-process events stream covering `waiting / active / completed / failed / retry-scheduled / delayed / dlq / drained`; this slice finishes the high-level shim surfaces on top so application code mirrors BullMQ exactly. **Zero engine changes.** + +Engine: no diff. The events stream wire format and emit sites were already in place from the post-Phase-4 polish (slice 5 of name-on-wire and the events-emission work in `chasquimq/src/events.rs`). The eng-review locked the call to *not* extend the wire format to carry the handler's `returnvalue` on `completed` events — keeping subscriber fan-out cheap was the right trade-off (1 MB result on a queue with 50 dashboards = 50 MB per job otherwise). `Job.waitUntilFinished` instead composes the existing `completed` event with `Queue.getJobResult` to surface the value when `storeResults=true`. + +Node shim (`chasquimq-node`): + +- `QueueEvents` gains per-id channels (`active:` / `completed:` / `failed:`) emitted alongside the existing broadcast channels. Targeted subscribers (notably `Job.waitUntilFinished`) wire onto these without paying the O(N-listeners) broadcast dispatch cost. Same payload shape as the broadcast. +- `Worker` gains `'drained'` / `'paused'` / `'resumed'` events. `'drained'` lazily spawns an embedded `QueueEvents` subscriber on the first `worker.on('drained', ...)` call (zero-cost when unused, torn down on `close()`). The subscriber uses `blockingTimeout: 1000` for snappy shutdown vs the QueueEvents 10s default. `pause()` / `resume()` emit synchronously after the native trip. `'progress'` / `'stalled'` documented as accepted-but-no-op for BullMQ-API parity (engine doesn't emit those transitions yet). +- `Job.waitUntilFinished(queueEvents, ttl?)` — BullMQ-parity event-driven wait. Resolves with the stored handler return value via `queue.getJobResult(id)` (requires `storeResults=true`; falls back to `undefined` when off). Rejects with `new Error(failedReason)` on `failed`, throws new `WaitUntilFinishedTimeoutError` on `ttl` elapse. Distinct from existing `waitForResult` which polls — `waitUntilFinished` is event-driven and detects completion without `storeResults`. Includes a small retry loop on the `getJobResult` fetch because the engine emits `completed` *before* the per-entry `JOB_OK_SCRIPT` writes the result key (events emit lives off the ack hot path). + +Python shim (`chasquimq-py`): + +- `QueueEvents` gains a `.on()` / `.off()` / `.once()` listener API alongside the existing async-iterator. Sync and async callbacks both supported; async callbacks are scheduled on the running loop, sync callback exceptions are logged and swallowed so a buggy listener cannot crash the subscriber. Per-id channels emit symmetric to Node. New `await wait_until_ready()` so callers can deterministically gate on subscriber-is-listening. +- `Worker` gains the same `.on()` / `.off()` / `.once()` listener API. Events: `ready`, `active`, `completed`, `failed`, `error`, `closing`, `closed`, `drained`, `paused`, `resumed`. Same lazy-drained-subscriber pattern as Node. `progress`/`stalled` accepted for parity. +- `Job.wait_until_finished(queue_events, *, timeout=None)` — mirrors Node. Returns the stored value or `None`, raises `RuntimeError(failedReason)` on failed, raises new `WaitUntilFinishedTimeoutError` (subclass of `TimeoutError`) on deadline. Same brief retry loop on `get_job_result` to handle the events-emit-before-result-write race. + +Design call locked: BullMQ's `Job.waitUntilFinished` exposes the return value because BullMQ JSON-encodes the value onto the events stream itself. Re-implementing that on ChasquiMQ would either (a) JSON-encode (regression on the msgpack wire-format invariant) or (b) msgpack-encode and force every subscriber to depend on the msgpack crate. Composing event + `getJobResult` keeps the events stream small, keeps subscribers msgpack-free, and works without the result backend for "just detect completion" cases. Documented as a deliberate cross-shim contract. + +Tests: 9 vitest cases (Node, `__test__/event-listeners.test.ts`) + 11 pytest cases (Python, `tests/test_event_listeners.py`) covering every new listener, both shims of the per-id channels, sync + async callback dispatch (Python), `waitUntilFinished` happy path / no-store-results fallback / failure / timeout, and the zero-cost "no drained listener attached" path. Full suites green: **194 Node** (185 before + 9 new) + **160 Python** (149 before + 11 new), no engine regressions. + +Bench (same-host, no engine changes — empty `git diff main -- chasquimq/src` so the host-load gate applies): TBD by the orchestrator run; the engine ceiling is unchanged because the slice adds zero round trips and zero allocations on the producer / consumer / promoter / scheduler / events-emit hot paths. + +Doc surfaces synced: this slice, both shim READMEs (symmetric "Subscribing to events" + "Awaiting a single job's completion" sections), `site/src/content/docs/reference/{node-api,python-api}.md` (Worker events table extended; new `waitUntilFinished` / `wait_until_finished` sections; QueueEvents per-id channels documented; `WaitUntilFinishedTimeoutError` in the errors section), a new `concepts/events-and-listeners.md` registered in `site/astro.config.mjs` (+ a link in `concepts/index.md`), and the root `README.md` feature table. + ## Deferred follow-ups for 1.x - **Opt-in result-write bench scenario.** The PR #75 bench guard locked in the no-overhead-when-off claim (`store_results=false` regresses 0%). The opt-in path (`store_results=true` under sustained load) is not yet measured. diff --git a/site/astro.config.mjs b/site/astro.config.mjs index aec0bd4..93977a1 100644 --- a/site/astro.config.mjs +++ b/site/astro.config.mjs @@ -139,6 +139,10 @@ export default defineConfig({ { label: "DLQ and recovery", slug: "concepts/dlq-and-recovery" }, { label: "Result backends", slug: "concepts/result-backends" }, { label: "The scheduler", slug: "concepts/the-scheduler" }, + { + label: "Events and listeners", + slug: "concepts/events-and-listeners", + }, { label: "Pause and resume", slug: "concepts/pause-and-resume" }, { label: "Redis Cluster", slug: "concepts/redis-cluster" }, { diff --git a/site/src/content/docs/concepts/events-and-listeners.md b/site/src/content/docs/concepts/events-and-listeners.md new file mode 100644 index 0000000..8d7265d --- /dev/null +++ b/site/src/content/docs/concepts/events-and-listeners.md @@ -0,0 +1,81 @@ +--- +title: Events and listeners +description: Two layers of events — in-process Worker listeners and the cross-process events stream — and when to reach for each. +--- + +ChasquiMQ exposes job-lifecycle events at two layers. Both surfaces use familiar `EventEmitter`-style names so existing application code reads naturally, but they have different scope, cost, and delivery semantics. Pick the right one for the job. + +## The two layers + +**In-process `Worker` events** are plain `EventEmitter` calls inside the worker that's running your processor. They fire on that worker only — no Redis traffic, no cross-process fan-out. Use them to observe what *this* worker is doing. + +**Cross-process `QueueEvents`** is a subscription to the per-queue events stream (`{chasqui:}:events`). The engine `XADD`s a small ASCII entry on every transition; any process running a `QueueEvents` instance for the same queue sees every event, regardless of which worker emitted it. Use it to observe the queue as a whole, or to coordinate between processes (a producer awaiting a worker's result, a dashboard tailing transitions, a metrics sidecar). + +The two surfaces share event names where it makes sense. `Worker.on('completed')` fires for jobs *this* worker processed; `queueEvents.on('completed')` fires for **every** job completed on the queue. + +## What the engine emits + +The events stream carries these transitions, all best-effort (a network blip on the events stream cannot delay an ack or cause a job to be retried): + +| Event | Fired by | Payload highlights | +|---|---|---| +| `waiting` | Producer (`XADD` to main stream) or promoter (delayed job became eligible). | `jobId`, `name` | +| `active` | Worker, before invoking the processor. | `jobId`, `name`, `attempt` | +| `completed` | Worker, after the processor returns. | `jobId`, `name`, `attempt` | +| `failed` | Worker, after the processor raises. | `jobId`, `name`, `failedReason`, `attempt` | +| `retry-scheduled` | Retry relocator, after an atomic reschedule onto the delayed ZSET. | `jobId`, `name`, `attempt`, `backoffMs` | +| `dlq` | DLQ relocator, after writing the entry to the DLQ stream. | `jobId`, `name`, `reason`, `attempt` | +| `drained` | Reader, on a full→empty transition (not on every empty poll). | (queue-scoped) | + +`retries-exhausted` is a synthetic alias of `dlq` (with the `reason` carried as the engine's `DlqReason` string) that the Node shim emits to match existing high-level-shim subscribers. + +Two listener names are accepted on both shims for API stability but are currently no-op — the engine does not emit the underlying transition yet: + +- `progress` — requires engine support for in-handler progress updates. +- `stalled` — requires a stalled-detector in the engine. + +Wire them up safely; they'll start firing when the corresponding engine work lands. + +## Per-id channels + +For events that carry a `jobId` (`active`, `completed`, `failed`), `QueueEvents` also fans the event onto a per-id channel named `:`. Targeted subscribers (like `Job.waitUntilFinished` / `Job.wait_until_finished`) listen there directly instead of filtering every broadcast event by id — at large fan-out this is the difference between an `O(N-listeners)` dispatch tax and an `O(1)` one. + +The channel naming convention is `:` (e.g. `completed:` / `failed:`). Power users can subscribe directly: + +```ts +events.on(`completed:${jobId}`, ({ jobId }) => { /* this job, done */ }); +events.on(`failed:${jobId}`, ({ failedReason }) => { /* this job, failed */ }); +``` + +## The return-value choice + +`completed` events from the events stream do **not** carry the handler's return bytes. Two reasons: + +1. **Subscriber-fan-out cost.** A 1 MB result on a queue with 50 dashboards subscribed would push 50 MB across the events stream per job. Keeping the events stream small keeps the cross-process observation surface cheap. +2. **Result storage is opt-in.** Workers run with `storeResults: false` by default; persisting nothing is the cheapest path through the engine. Carrying the value on every event would invert that default. + +The contract is: the events stream tells you *that* a job completed; if you need the value, fetch it from the result key via `Queue.getJobResult(jobId)` and run the worker with `storeResults: true`. The `Job.waitUntilFinished(queueEvents, ttl?)` helper composes both — it listens for the event and fetches the result after. + +## Awaiting a single job's completion + +Two helpers, both on `Job`, with different races: + +- **`Job.waitForResult({ timeoutMs })`** (Node) / **`Job.wait_for_result(timeout=)`** (Python) — *polls* the result key. Requires `storeResults: true` to detect completion at all. Wins when the result key was written *before* the wait started (a job that finished a few seconds before you called). +- **`Job.waitUntilFinished(queueEvents, ttl?)`** (Node) / **`Job.wait_until_finished(queue_events, timeout=)`** (Python) — *subscribes* to the per-id channels on the events stream. Detects completion even when `storeResults` is off. Loses to a job that completed before the listeners wired up. + +For low-latency awaits of jobs you just enqueued, `waitUntilFinished` is the right call (no polling tax, no need for the result backend). For "did this id ever finish" lookups, `waitForResult` is the right call. The two compose: in many deployments you set `storeResults: true` and use `waitUntilFinished` exclusively — it both detects completion via the event and fetches the persisted value. + +## Lazy subscriber lifecycle + +The `Worker.on('drained', ...)` listener is the only `Worker` event that requires cross-process traffic. The shim lazily spawns an embedded `QueueEvents` subscriber the first time a `drained` listener attaches; it's torn down on `Worker.close()`. Workers that never subscribe to `drained` pay no extra Redis connections — this is a strict zero-overhead-when-unused contract. + +The same lazy pattern applies on the Python `Worker`. Cross-shim symmetric. + +## Lost-event race + +Events emitted *before* a subscriber's first `XREAD BLOCK` lands are missed. The shims minimise this race: + +- The Node `Worker`'s embedded drained subscriber awaits `XREAD BLOCK` flush via a `setImmediate` yield before releasing the worker's startup gate. +- The Python `QueueEvents` exposes `await events.wait_until_ready()` — callers needing a deterministic "subscriber is listening" gate can await it before producing work. + +In production this rarely matters: subscribers attach early in process startup and the events flow continuously. In tests, prefer `wait_until_ready` (Python) or the `await events.waitUntilReady()` plus a small `await new Promise(r => setImmediate(r))` (Node) to close the window. diff --git a/site/src/content/docs/concepts/index.md b/site/src/content/docs/concepts/index.md index 6a51269..9757930 100644 --- a/site/src/content/docs/concepts/index.md +++ b/site/src/content/docs/concepts/index.md @@ -22,6 +22,7 @@ The why behind every design decision. Read [Thinking in ChasquiMQ](/concepts/thi - [DLQ and recovery](/concepts/dlq-and-recovery/) — why the DLQ is just another stream. - [Result backends](/concepts/result-backends/) — why result storage is opt-in. - [The scheduler](/concepts/the-scheduler/) — separate from the consumer for a reason. +- [Events and listeners](/concepts/events-and-listeners/) — two layers of events, when to reach for each, and how `waitUntilFinished` composes them. ## Architecture and trade-offs diff --git a/site/src/content/docs/reference/node-api.md b/site/src/content/docs/reference/node-api.md index edb21ec..4fcfdac 100644 --- a/site/src/content/docs/reference/node-api.md +++ b/site/src/content/docs/reference/node-api.md @@ -443,7 +443,7 @@ reader immediately (edge-triggered, no poll latency). In-memory only — does not write the cross-process flag and does not survive a process restart; for queue-wide durable pause use [`queue.pause()`](#queuepause-queueresume-queueispaused). Idempotent. -`doNotWaitActive` is accepted for BullMQ call-shape parity but is a +`doNotWaitActive` is accepted for call-shape stability but is a no-op (in-flight jobs always drain in the background). ### `worker.isClosed`, `worker.isRunning` @@ -479,11 +479,24 @@ Throws `NotSupportedError` in v1. | `error` | `(err)` | Engine-side error surfaced from the native loop. | | `closing` | `(msg)` | Start of `.close()`. | | `closed` | `()` | Shutdown completes. | +| `drained` | `()` | Engine observed a full→empty transition on the main stream. **Cross-process scope.** Lazily wires an embedded `QueueEvents` subscriber on the first `.on('drained', ...)` call; torn down on `close()`. | +| `paused` | `()` | `.pause()` was called. Process-local. | +| `resumed` | `()` | `.resume()` was called. Process-local. | + +Names accepted for API stability but currently never fire (engine +doesn't emit the underlying transition yet — safe to wire up, will +start firing when the corresponding engine work lands): + +- `progress` — `(job, progress)`. +- `stalled` — `(jobId, prev)`. ```ts worker.on("completed", (job, result) => { console.log("completed", job.id, result); }); +worker.on("drained", () => { + console.log("queue empty"); +}); ``` ## Job @@ -550,6 +563,47 @@ from "not yet completed", so this method will time out. Mirror [`QueueEvents`](#queueevents) instead. ::: +### `job.waitUntilFinished(queueEvents, ttl?)` + +```ts +async waitUntilFinished( + queueEvents: QueueEvents, + ttl?: number, +): Promise +``` + +Event-driven completion-wait. Subscribes to the +per-id `completed:` / `failed:` channels on the +supplied `queueEvents` and resolves / rejects on the first to fire. + +- Resolves with the handler's return value when the worker ran with + `storeResults: true`. The value is fetched via + `Queue.getJobResult(this.id)` after the `completed` event lands. + When `storeResults` was off (or the handler returned `undefined`), + resolves with `undefined`. +- Rejects with `new Error(failedReason)` on the engine-reported + failure reason (the same string surfaced on `Worker`'s `failed` + event). +- Throws [`WaitUntilFinishedTimeoutError`](/reference/error-codes/) + on `ttl` elapse. Omit `ttl` for an unbounded wait. + +Distinct from [`waitForResult`](#jobwaitforresultopts): event-driven +(no polling), and works without `storeResults` to *detect* +completion. The two cover different races — `waitUntilFinished` +loses to a job that finished before the listeners wired up; +`waitForResult` can read a result key written before the wait +started but needs `storeResults`. + +```ts +const events = new QueueEvents("emails", { connection }); +await events.waitUntilReady(); + +const job = await queue.add("send", { to: "ada@example.com" }); +const result = await job.waitUntilFinished(events, 30_000); + +await events.close(); +``` + ### `job.toJSON()` ```ts @@ -610,6 +664,23 @@ Implements `[Symbol.asyncDispose]`. | `unknown` | `({ eventName, fields }, eventId)` | Forward-compat sink for unrecognized event types. | | `error` | `(err)` | Operational error during XREAD. | +In addition to the broadcast channels, three per-id channels fire +alongside the matching broadcast for events that carry a `jobId` +(naming convention: `:`): + +| Event | Args | Fires alongside | +|---|---|---| +| `active:` | `({ jobId, name, prev, attempt }, eventId)` | `active` | +| `completed:` | `({ jobId, name, attempt, returnvalue }, eventId)` | `completed` | +| `failed:` | `({ jobId, name, failedReason, attempt }, eventId)` | `failed` | + +Per-id channels let `Job.waitUntilFinished` (and any UI watching one +job) wire a targeted listener without paying the O(N-listeners) +broadcast dispatch cost. `returnvalue` is always `undefined` on the +wire — the events stream does not carry the handler's bytes; pair +with `Queue.getJobResult(jobId)` (and `WorkerOptions.storeResults`) +to read it back. + Numeric fields (`attempt`, `backoffMs`, `delay`, `duration_us`, `ts`) are coerced to `number` at parse time so subscribers don't have to remember which ones need an explicit cast. @@ -892,6 +963,7 @@ The shim throws typed errors so application code can branch on - [`UnrecoverableError`](/reference/error-codes/#cmq-011--handler-signaled-unrecoverable--dlq) — throw from a processor to skip retries and route to the DLQ. - [`WaitForResultTimeoutError`](/reference/error-codes/#cmq-102--node-result-wait-timeout) — `Job.waitForResult` timed out. +- `WaitUntilFinishedTimeoutError` — `Job.waitUntilFinished` saw neither a `completed` nor a `failed` event within the supplied `ttl`. Distinct from a failed job: a failed job rejects with `new Error(failedReason)`; this error fires only when the events stream itself goes silent. - [`NotSupportedError`](/reference/error-codes/#cmq-100--node-feature-not-supported) — caller asked for a v1-stubbed feature. - [`RateLimitError`](/reference/error-codes/#cmq-101--node-rate-limit) — reserved; `Worker.rateLimit` throws this in a future slice. diff --git a/site/src/content/docs/reference/python-api.md b/site/src/content/docs/reference/python-api.md index 5ede9ef..0dfabcb 100644 --- a/site/src/content/docs/reference/python-api.md +++ b/site/src/content/docs/reference/python-api.md @@ -535,6 +535,40 @@ Idempotent. - `worker.is_running` — engine task in flight. - `worker.is_closed` — `True` after the first `close()` call. +### Worker events + +Subscribe via `worker.on(event_name, callback)`. Callbacks may be +plain functions or `async def` coroutines — async callbacks are +scheduled on the running loop. Use `worker.off(event_name, callback)` +to remove, or `worker.once(event_name, callback)` for a single-shot +listener. A raised exception from a sync callback is logged and +swallowed so a buggy listener cannot crash the worker. + +| Event | Args | Fires when | +|---|---|---| +| `ready` | `()` | `run()` starts the engine loop. | +| `active` | `(job: Job)` | Before each handler invocation. | +| `completed` | `(job: Job, result)` | Handler returns. Engine acks the job. | +| `failed` | `(job: Job, err: BaseException)` | Handler raises. Exception re-raised into engine retry/DLQ path. | +| `error` | `(err: BaseException)` | Engine-side error surfaced from the native loop. | +| `closing` | `()` | Start of `close()`. | +| `closed` | `()` | Shutdown completes. | +| `drained` | `()` | Engine observed a full→empty transition on the main stream. **Cross-process scope.** Lazily wires an embedded `QueueEvents` subscriber on the first `worker.on('drained', ...)` call; torn down on `close()`. | +| `paused` | `()` | `pause()` was called. Process-local. | +| `resumed` | `()` | `resume()` was called. Process-local. | + +Names accepted for parity but currently never fire (engine doesn't +emit the underlying transition yet — safe to wire up, will start +firing when the corresponding engine work lands): + +- `progress` — `(job: Job, progress)`. +- `stalled` — `(job_id: str, prev: str)`. + +```python +worker.on("completed", lambda job, result: print("ok", job.id)) +worker.on("drained", lambda: print("queue empty")) +``` + ### Type alias: `Handler` ```python @@ -609,6 +643,52 @@ the most common cause. For high-fanout workloads, subscribe to [`QueueEvents`](#queueevents) instead. ::: +### `await job.wait_until_finished(queue_events, *, timeout)` + +```python +async def wait_until_finished( + self, + queue_events: QueueEvents, + *, + timeout: float | None = None, +) -> Any | None +``` + +Event-driven completion-wait, mirroring the Node shim. +Subscribes to the per-id `completed:` / `failed:` +channels on the supplied `queue_events` and resolves / raises on +the first to fire. + +- Returns the handler's return value when the worker ran with + `store_results=True`. The value is fetched via + `Queue.get_job_result(self.id)` after the `completed` event lands. + When `store_results` was off (or the handler returned `None`), + returns `None`. +- **Raises** `RuntimeError(failed_reason)` on the engine-reported + failure reason (the same string surfaced on `Worker`'s `failed` + event). +- **Raises** `WaitUntilFinishedTimeoutError` on `timeout` elapse. + Omit `timeout` for an unbounded wait. + +Distinct from [`wait_for_result`](#await-jobwait_for_resulttimeoutpoll_interval): +event-driven (no polling), and works without `store_results=True` +to *detect* completion. The two cover different races — +`wait_until_finished` loses to a job that finished before the +listeners wired up; `wait_for_result` can read a result key written +before the wait started but needs `store_results=True`. + +```python +events = QueueEvents("emails", redis_url=url) +# Force the subscriber loop to start so wait_until_ready is awaitable. +events.on("completed", lambda *_a: None) +await events.wait_until_ready() + +job = await queue.add("send", {"to": "ada@example.com"}) +result = await job.wait_until_finished(events, timeout=30.0) + +await events.close() +``` + ## QueueEvents ```python @@ -645,13 +725,75 @@ async for ev in events: Yields [`QueueEvent`](#queueevent) values. Iteration ends when [`close()`](#await-queueeventsclose) is called. +### Listener API + +```python +def on(self, event_name: str, callback: Listener) -> None +def off(self, event_name: str, callback: Listener) -> None +def once(self, event_name: str, callback: Listener) -> None +async def wait_until_ready(self) -> None +``` + +`EventEmitter`-style listener surface alongside the iterator. The +first `on(...)` call lazily spawns an internal subscriber task; +`close()` cancels it. Pick *one* of the two surfaces — both share +the same connection but only one `XREAD` consumer can run at a +time. + +Callbacks may be plain functions or `async def` coroutines. +Callback arity mirrors the Node shim: one argument (`event_id`) for +queue-scoped events (`drained`), two arguments (`payload`, +`event_id`) for per-job events. `payload` is a `dict` with +`jobId` / `name` / engine-specific fields. The `failed` payload +exposes the engine reason as both `reason` (raw) and `failedReason` +(camelCase alias) for convenience. + +Event names accepted: + +| Event | Args | Engine origin | +|---|---|---| +| `waiting` | `(payload, event_id)` | Producer added (or promoter promoted) the job. | +| `active` | `(payload, event_id)` | Worker pulled the job; handler is about to run. | +| `completed` | `(payload, event_id)` | Handler returned. | +| `failed` | `(payload, event_id)` | Handler raised. `payload['failedReason']`. | +| `retry-scheduled` | `(payload, event_id)` | Engine atomically rescheduled the job onto the delayed ZSET. | +| `delayed` | `(payload, event_id)` | Producer enqueued with `delay > 0`. | +| `dlq` | `(payload, event_id)` | DLQ relocator wrote the entry to the DLQ stream. | +| `retries-exhausted` | `(payload, event_id)` | Synthetic alias of `dlq` (chasquimq-specific). | +| `drained` | `(event_id,)` | Engine drained (queue-scoped, no `jobId`). | + +Per-id channels fire alongside the broadcast channel for events +that carry a `job_id` (naming convention: `:`). +Used internally by +[`Job.wait_until_finished`](#await-jobwait_until_finishedqueue_events-timeout): + +| Event | Args | Fires alongside | +|---|---|---| +| `active:` | `(payload, event_id)` | `active` | +| `completed:` | `(payload, event_id)` | `completed` | +| `failed:` | `(payload, event_id)` | `failed` | + +`await wait_until_ready()` blocks until the listener subscriber has +issued its first `XREAD BLOCK` (i.e. the subscription window is +open) — useful for tests that need a deterministic "subscriber is +listening" gate. No-op when no listeners are attached. + +```python +events = QueueEvents("emails") +events.on("completed", lambda payload, eid: print("ok", payload["jobId"])) +events.on(f"completed:{job_id}", on_one_completed) # targeted +await events.wait_until_ready() +# ... do work ... +await events.close() +``` + ### `await queue_events.close()` ```python async def close(self) -> None ``` -Stop iteration and release the Redis connection. +Stop iteration / dispatch and release the Redis connection. ## QueueEvent @@ -843,6 +985,7 @@ The shim raises typed exceptions: - [`UnrecoverableError`](/reference/error-codes/#cmq-011--handler-signaled-unrecoverable--dlq) — raise from a handler to skip retries and route to the DLQ. Subclassing is supported via MRO-aware `issubclass` check on the native side. - [`NotSupportedError`](/reference/error-codes/#cmq-200--python-feature-not-supported) — caller asked for a v1-stubbed feature. +- `WaitUntilFinishedTimeoutError` — [`Job.wait_until_finished`](#await-jobwait_until_finishedqueue_events-timeout) saw neither a `completed` nor a `failed` event within the supplied `timeout`. Subclasses `TimeoutError`. Distinct from a failed job: a failed job raises a regular `RuntimeError(failed_reason)`; this error fires only when the events stream itself goes silent. ```python from chasquimq import UnrecoverableError