Streaming partial results

A long-running task can publish intermediate results as it works; a consumer streams them as they arrive — useful for ETL, ML training steps, or batch progress.

Publish

Call currentJob().publish(value) from inside the task. The value must be JSON-serializable.

import { currentJob } from "@byteveda/flexiq";

queue.task("train", async (epochs: number) => {
  const job = currentJob();
  for (let epoch = 1; epoch <= epochs; epoch++) {
    const loss = await trainEpoch(epoch);
    job?.publish({ epoch, loss });
  }
  return "trained";
});

Consume

queue.stream(jobId) is an async iterator that yields each published value in order, then ends when the job reaches a terminal state.

const id = queue.enqueue("train", [10]);
queue.runWorker(); // a worker must be consuming the queue for `train` to run

for await (const update of queue.stream(id)) {
  console.log(`epoch ${update.epoch}: loss ${update.loss}`);
}

const result = await queue.result(id); // "trained"

stream(jobId, { timeoutMs, pollMs }) bounds the wait (default 60 s) and poll interval (default 200 ms). It works from any process sharing the storage, and a late consumer still drains every value already published.

Raw logs

queue.taskLogs(jobId) returns the raw log entries (published partials have level: "result").