diff --git a/docs/app/styles/docs.css b/docs/app/styles/docs.css index 34efdd10..d29be2ee 100644 --- a/docs/app/styles/docs.css +++ b/docs/app/styles/docs.css @@ -449,19 +449,19 @@ body { /* Ordered lists: Tailwind Preflight resets `ol` to `list-style: none`, so the numbers disappear. Restore decimal markers + the indent they need, and tint them. (Kept off `display: flex`, which would drop the ::marker.) */ -.article ol { +.article ol:not(.flowsteps) { margin: 14px 0; padding-left: 26px; list-style: decimal; } -.article ol li { +.article ol:not(.flowsteps) li { font-size: 15px; line-height: 1.6; color: var(--txt2); padding-left: 6px; margin: 9px 0; } -.article ol li::marker { +.article ol:not(.flowsteps) li::marker { color: var(--indigo); font-weight: 600; } diff --git a/docs/content/docs/shared/guides/core/predicates.mdx b/docs/content/docs/shared/guides/core/predicates.mdx index f4cd4fcc..e8e2f98b 100644 --- a/docs/content/docs/shared/guides/core/predicates.mdx +++ b/docs/content/docs/shared/guides/core/predicates.mdx @@ -452,7 +452,9 @@ skipped for the current job — job dispatch is unaffected. Gates run once, on the **producer** side, at enqueue time — there's -no worker-dispatch phase. +no worker-dispatch phase. Gates are in-process functions only: there's no +serializable predicate DSL (no JSON/string form, no config-file or dashboard +round-trip). @@ -460,7 +462,8 @@ no worker-dispatch phase. Predicates and gates run once, synchronously, on the **producer** thread at enqueue time — there's no worker-dispatch phase. Keep them -fast and side-effect-free. +fast and side-effect-free. Gates are in-process lambdas — there's no +serializable predicate DSL (no JSON or string form). @@ -502,10 +505,10 @@ payloads that need skip/defer semantics. - - ## Persistence & inspection + + Every registered predicate is serialized once at decoration time. The JSON is reachable through: @@ -533,8 +536,27 @@ restored = Predicate.from_dict(blob) # restored.evaluate(ctx) — re-evaluate against a context ``` + + + + +Node has no predicate inspection API — gates are opaque closures held on the +queue; there's no `listPredicates()` equivalent and the dashboard shows no +"gated by" schema for Node. + + + + + +Java exposes no predicate inspection API — registered gates are opaque and +there is no `listPredicates()` equivalent. + + + ## Metrics & events + + ```python queue.predicate_stats() # {"allowed": 412, "denied": 3, "deferred": 8, "cancelled": 1, "errors": 0} @@ -562,8 +584,27 @@ for event in ( queue.on_event(event, on_predicate_event) ``` + + + + +Node emits no predicate metrics or events. The event bus covers job lifecycle +only (`job.completed`, `job.retrying`, `job.dead`, `job.cancelled`) — there is +no `predicateStats()` and no deferred/rejected event. + + + + + +Java emits no predicate metrics or events — no `predicateStats()`, and the +event enum carries no deferred/rejected predicate signals. + + + ## Examples + + ### Business-hours-only report with urgent override ```python @@ -637,3 +678,68 @@ async def index_document(tenant: str, doc_id: str): ... ``` + + + +### Tenant allowlist (deny rejects the enqueue) + +```ts +const allowed = new Set(["acme", "globex"]); + +queue.gate("reindexSearch", ({ args }) => allowed.has(args[0] as string)); + +queue.enqueue("reindexSearch", ["acme"]); // ok +queue.enqueue("reindexSearch", ["evilcorp"]); // throws PredicateRejectedError +``` + +### Business-hours + urgent override (booleans, no defer) + +```ts +import { anyOf } from "@byteveda/taskito"; + +const businessHours = () => { + const h = new Date().getHours(); + return h >= 9 && h < 17; +}; +const urgent = ({ args }: { args: readonly unknown[] }) => args[1] === true; + +queue.gate("sendDailyReport", anyOf(businessHours, urgent)); + +queue.enqueue("sendDailyReport", ["alpha"]); // outside 09–17: throws (rejected, not deferred) +queue.enqueue("sendDailyReport", ["alpha", true]); // urgent bypass: always enqueues +``` + +Unlike Python's `is_business_hours` (which *defers* to opening time), a Node +gate that returns `false` **rejects** the enqueue — Node has no defer outcome. + + + + + +### Business-hours + urgent override (recipe defers, urgent allows now) + +```java +EnqueueGate hours = Recipes.businessHours(ZoneId.of("America/Los_Angeles")); + +taskito.gate("sendDailyReport", ctx -> + ((Report) ctx.payload()).urgent() ? EnqueueDecision.allow() : hours.decide(ctx)); + +taskito.enqueue(sendReport, new Report("alpha", false)); // defers to next 09:00 PT +taskito.enqueue(sendReport, new Report("alpha", true)); // runs now +``` + +### Tenant allowlist (skip vs reject, gate-aware `tryEnqueue`) + +```java +Set allowed = Set.of("acme", "globex"); + +taskito.gate("reindexSearch", ctx -> + allowed.contains(((Reindex) ctx.payload()).tenant()) + ? EnqueueDecision.allow() + : EnqueueDecision.reject("tenant not allowed")); + +Optional id = taskito.tryEnqueue(reindex, new Reindex("acme")); // present +taskito.enqueue(reindex, new Reindex("evilcorp")); // throws PredicateRejectedException +``` + + diff --git a/docs/content/docs/shared/guides/core/queues.mdx b/docs/content/docs/shared/guides/core/queues.mdx index 2e2f3ae8..805d699d 100644 --- a/docs/content/docs/shared/guides/core/queues.mdx +++ b/docs/content/docs/shared/guides/core/queues.mdx @@ -369,10 +369,10 @@ Map all = taskito.statsAllQueues(); - - ## Queue-level defaults + + Configure defaults for every task at the `Queue` level; individual `@queue.task()` decorators override them: @@ -386,3 +386,34 @@ queue = Queue( ``` + + + +There's no queue-level defaults object. Retry and timeout defaults are set +**per task** on `queue.task(...)` via `maxRetries` and `timeoutMs`; a per-call +`enqueue()` override still wins. `priority` has no default — it's enqueue-only +(see [Default priority](#default-priority) above). `configureQueue()` only sets +a queue's `rateLimit` / `maxConcurrent`. + +```ts +queue.task("resize", (id: string) => { /* ... */ }, { + maxRetries: 3, // default retry budget for this task + timeoutMs: 300_000, // default per-job timeout (ms) +}); +``` + + + + + +Java has no queue-level defaults object — defaults live on the **task +descriptor**, and a per-enqueue `EnqueueOptions` still overrides them: + +```java +Task resize = Task.of("resize", ResizePayload.class) + .priority(0) // default priority + .maxRetries(3) // default retry budget + .timeout(Duration.ofSeconds(300)); // default per-job timeout +``` + + diff --git a/docs/content/docs/shared/guides/core/tasks.mdx b/docs/content/docs/shared/guides/core/tasks.mdx index a65bce48..e1038e36 100644 --- a/docs/content/docs/shared/guides/core/tasks.mdx +++ b/docs/content/docs/shared/guides/core/tasks.mdx @@ -17,24 +17,46 @@ from taskito import Queue queue = Queue(db_path="myapp.db") -@queue.task() -def process_data(data: dict) -> str: - # Your logic here - return "done" +@queue.task(queue="emails", max_retries=5) +def send_email(to: str, subject: str, body: str) -> str: + smtp.send(to, subject, body) + return "sent" ``` ```ts -queue.task("add", (a: number, b: number) => a + b); +import { Queue } from "@byteveda/taskito"; + +const queue = new Queue({ dbPath: "myapp.db" }); + +// The name is the contract; the handler runs on a worker and its return is the result. +queue.task( + "sendEmail", + async (to: string, subject: string, body: string) => { + await smtp.send(to, subject, body); + return "sent"; + }, + { maxRetries: 5 }, // queue is chosen per-enqueue, not here +); ``` ```java -Task sendEmail = Task.of("send_email", EmailPayload.class); +record EmailPayload(String to, String subject, String body) {} + +// Producer side: a typed descriptor (name + payload type + defaults). +Task sendEmail = Task.of("send_email", EmailPayload.class) + .queue("emails") + .maxRetries(5); + +// Worker side: bind the logic separately as a TaskFunction. +Worker worker = taskito.worker() + .handle(sendEmail, payload -> { smtp.send(payload); return "sent"; }) + .start(); ``` For generic payloads that a `Class` token can't express, use a Jackson @@ -497,10 +519,10 @@ the limit, or serialize the critical section with a - - ## Per-task middleware + + Apply middleware to a specific task only, in addition to queue-level middleware: @@ -512,8 +534,28 @@ def important_task(): ... ``` + + + + +Node middleware is registered queue-wide with `queue.use(...)`; there's no +per-task `middleware` option. Scope logic to a task inside the middleware via +`ctx.taskName` in `before` / `after`, or toggle a global middleware per task +from the dashboard (`queue.disableMiddlewareForTask(task, name)`). + + + + + +Java `Middleware` is registered globally with `taskito.use(...)`; there's no +per-task attachment. Branch inside the middleware on `ctx.taskName()`. + + + ## Per-task serializer + + Override the queue-level serializer for a specific task — the full round-trip: arguments are serialized with it at enqueue time and deserialized with it on the worker before the task function is called. Both the sync @@ -532,8 +574,42 @@ Useful when a task needs a different format (e.g., human-readable JSON for audit tasks) or when the payload isn't picklable. See Serializers. + + + + +The serializer is a queue-level choice (`new Queue({ serializer })`); it can't +be swapped per task. For per-task payload handling, register named `codecs` +(compression / encryption / signing) applied on top of the queue serializer: + +```ts +const queue = new Queue({ dbPath: "myapp.db", codecs: { gzip: new GzipCodec() } }); +queue.task("audit_event", handleAudit, { codecs: ["gzip"] }); // transform, not a format swap +``` + +See Serializers. + + + + + +The serializer is a builder-level choice (`Taskito.builder().serializer(...)`); +it can't be swapped per task. For per-task payload handling, use +`Task.codecs(...)` — the same layered-transform model (compression / encryption +/ signing) applied on top of the builder serializer: + +```java +Task audit = Task.of("audit_event", AuditEvent.class).codecs("gzip"); +``` + +See Serializers. + + + ## Job expiration + + `expires` skips a job that wasn't started within the deadline. Set a default on the task, override it per call, or pass it only at enqueue time: @@ -546,8 +622,30 @@ time_sensitive.delay() # uses the 300s default time_sensitive.apply_async(args=(), expires=60) # tighter window for this call ``` + + + + +The Node SDK has no job-expiration deadline. Job options are `delayMs` +(earliest start) and `timeoutMs` (max run time), but a queued job is never +discarded for sitting too long. Gate stale work inside the handler, or cancel +it with `queue.cancelJob(id)`. + + + + + +The Java SDK has no job-expiration deadline. `EnqueueOptions` offers `delay` +(earliest start) and `timeout` (max run time), but a queued job is never +discarded for sitting too long. Gate stale work inside the handler, or cancel +it with `taskito.cancelJob(id)`. + + + ## Task naming + + By default, tasks are named using `module.qualname`: ```python @@ -567,6 +665,34 @@ def process(): # Named: my-custom-name + + +Node task names are always explicit — the required first argument of +`queue.task(name, handler)`. Unlike Python's `module.qualname` default, there is +no auto-derived name (JS function names are unreliable after bundling). + +```ts +queue.task("emails.send", (to: string, subject: string) => send(to, subject)); +queue.enqueue("emails.send", ["user@example.com", "Welcome"]); +``` + + + + + +`Task.of(name, type)` takes an explicit name. With the `@TaskHandler` +annotation, an empty `value()` defaults the task name to the method name: + +```java +@TaskHandler("emails.send") // explicit name +String send(EmailPayload p) { ... } + +@TaskHandler // name defaults to the method name: "report" +Report report(List m) { ... } +``` + + + ## Enqueuing jobs @@ -647,10 +773,10 @@ String id = taskito.enqueue(sendEmail, payload, EnqueueOptions.builder() - - ### Direct call + + Calling a task directly runs it synchronously, bypassing the queue — useful in tests: @@ -660,6 +786,35 @@ result = send_email("user@example.com", "Hello", "World") # Runs immediately + + +`queue.task()` returns the queue, not a task object, so there's nothing to +"call directly." Keep a reference to your handler function and invoke it +directly in tests; only `enqueue()` routes through the queue. + +```ts +const sendEmail = async (to: string, subject: string) => { /* ... */ return "sent"; }; +queue.task("sendEmail", sendEmail); + +const result = await sendEmail("user@example.com", "Hi"); // runs now, bypasses the queue +``` + + + + + +A `Task` carries no logic to call. Invoke the `TaskFunction` / handler +method directly for a synchronous run; `taskito.enqueue(task, payload)` is the +only path that goes through the queue. + +```java +TaskFunction send = payload -> smtp.deliver(payload); + +String result = send.apply(payload); // runs now, bypasses the queue +``` + + + ## Batch enqueue Enqueue many jobs of one task in a single storage round-trip: diff --git a/docs/content/docs/shared/guides/core/workers.mdx b/docs/content/docs/shared/guides/core/workers.mdx index 1a507af2..709a81c6 100644 --- a/docs/content/docs/shared/guides/core/workers.mdx +++ b/docs/content/docs/shared/guides/core/workers.mdx @@ -179,10 +179,10 @@ taskito.worker() - - ## Worker specialization + + Advertise capability tags when starting a worker: ```python @@ -197,6 +197,30 @@ the right machines with [dedicated queues](#selecting-queues) instead. + + +Node workers can't advertise capability tags — `runWorker()` has no `tags` +option. Route capability-specific work with [dedicated queues](#selecting-queues) +instead: give GPU or heavy jobs their own queue and start a worker with +`runWorker({ queues: ["gpu"] })`. + +`listWorkers()` rows include a `tags` field, but only SDKs that can set it +populate it — it stays empty for Node-started workers. + + + + + +`taskito.worker()` has no `tags(...)` builder method — Java workers can't +advertise capability tags. Route capability-specific work with +[dedicated queues](#selecting-queues) instead +(`taskito.worker().queues("gpu").start()`). + +`WorkerInfo.tags` exists on `listWorkers()` rows, but only SDKs that support +tag advertisement set it; it's null for Java-started workers. + + + ## Worker concurrency How many jobs one worker process runs at once follows the SDK's own @@ -348,6 +372,14 @@ The heartbeat fires every 5 seconds and carries the current resource-health snapshot, so worker resources going unhealthy shows up in `listWorkers()` too. +### Lifecycle events + +Node has no worker-lifecycle events. `queue.on(event, handler)` fires only the +four job-outcome events (`job.completed`, `job.retrying`, `job.dead`, +`job.cancelled`) — there's no `WORKER_ONLINE` / `OFFLINE` / `UNHEALTHY`. Observe +worker presence and health by polling `listWorkers()`, whose rows carry +`status` and a per-resource `resourceHealth` snapshot from the last heartbeat. + @@ -355,6 +387,14 @@ resources going unhealthy shows up in `listWorkers()` too. Heartbeating is managed natively by the core — there's no user-facing interval to configure. +### Lifecycle events + +Java has no worker-lifecycle events. `.on(EventName, ...)` fires only the four +job-outcome events (`SUCCESS`, `RETRY`, `DEAD`, `CANCELLED`) — there's no +`WORKER_ONLINE` / `OFFLINE` / `UNHEALTHY`. Observe worker presence and health by +polling `taskito.listWorkers()`, whose `WorkerInfo` rows carry `status` and a +`resourceHealth` snapshot from the last heartbeat. + ## Graceful shutdown diff --git a/docs/content/docs/shared/guides/extensibility/serializers.mdx b/docs/content/docs/shared/guides/extensibility/serializers.mdx index 2333b34f..a12001e0 100644 --- a/docs/content/docs/shared/guides/extensibility/serializers.mdx +++ b/docs/content/docs/shared/guides/extensibility/serializers.mdx @@ -505,6 +505,44 @@ overview and what gets serialized where. + + +## Choosing a serializer + +| Serializer | When to use | Cross-SDK? | +|---|---|---| +| `JsonSerializer` | The default — human-readable, debuggable payloads for Node↔Node work. Leave it unless you have a reason to switch. | No — portable format, same-SDK payloads | +| `MsgpackSerializer` | Node↔Node work where you want compact binary payloads instead of JSON text. | No — portable format, same-SDK payloads | +| `CborSerializer` | A task is produced or consumed by another Taskito SDK. The only serializer whose wire format (the `0x02` envelope tag) is part of the cross-SDK contract; round-trips big integers as `BigInt`. | Yes — cross-SDK wire format | +| `SignedSerializer(secret, inner?)` | Payloads must be tamper-evident but not hidden — prefixes an HMAC-SHA256 tag and rejects mismatches. Authentication, not encryption. | Same-SDK by default (wraps `JsonSerializer`) | +| `EncryptedSerializer(secret, inner?)` | Tasks carry sensitive data that must not be readable in the database — AES-256-GCM confidentiality **and** integrity. | Same-SDK by default (wraps `JsonSerializer`) | + +**Rule of thumb**: keep the default `JsonSerializer` unless you have a specific +reason to switch. Use `MsgpackSerializer` for more compact Node↔Node payloads, +`CborSerializer` when a task crosses SDK boundaries, and `EncryptedSerializer` +when payloads must not be readable in storage. + + + + + +## Choosing a serializer + +| Serializer | When to use | Cross-SDK? | +|---|---|---| +| `JsonSerializer` | The default — JSON via Jackson, accepts a custom `ObjectMapper`. Human-readable and debuggable for Java↔Java work. | No — portable format, same-SDK payloads | +| `MsgpackSerializer` | Java↔Java work where you want compact binary payloads. Its `jackson-dataformat-msgpack` dependency is `compileOnly` — add it to your build. | No — portable format, same-SDK payloads | +| `CborSerializer` | A task is produced or consumed by another Taskito SDK. The only serializer whose wire format (the `0x02` envelope tag) is part of the cross-SDK contract. `jackson-dataformat-cbor` is `compileOnly`. | Yes — cross-SDK wire format | +| `SignedSerializer(delegate, key)` | Payloads must be tamper-evident but not hidden — prefixes an HMAC-SHA256 tag with constant-time verification. Authentication, not encryption. | Same-SDK by default (depends on `delegate`) | +| `EncryptedSerializer(delegate, key)` | Tasks carry sensitive data that must not be readable in the database — AES-GCM confidentiality **and** integrity. Key must be 16, 24, or 32 bytes. | Same-SDK by default (depends on `delegate`) | + +**Rule of thumb**: keep the default `JsonSerializer` unless you have a specific +reason to switch. Use `MsgpackSerializer` for more compact Java↔Java payloads, +`CborSerializer` when a task crosses SDK boundaries, and `EncryptedSerializer` +when payloads must not be readable in storage. + + + ## Payload codecs A `PayloadCodec` is a reversible byte-to-byte transform layered *around* a diff --git a/docs/content/docs/shared/guides/integrations/prometheus.mdx b/docs/content/docs/shared/guides/integrations/prometheus.mdx index 942a81c3..624b33eb 100644 --- a/docs/content/docs/shared/guides/integrations/prometheus.mdx +++ b/docs/content/docs/shared/guides/integrations/prometheus.mdx @@ -7,14 +7,58 @@ Taskito exports Prometheus metrics through two pieces — a middleware that records per-task execution counters and a duration histogram, and a stats collector that polls queue depth and dead-letter size onto gauges on an interval. Both write into the same -prometheus-client} node={prom-client} /> +prometheus-client} node={prom-client} java={Micrometer} /> registry, so one HTTP endpoint can serve everything. -Java has no Prometheus contrib — metrics come from -[Micrometer](/java/guides/integrations/micrometer) instead, which yields a -timer (and a trace span) from one instrumentation. +Java has no Prometheus contrib. Metrics come from Micrometer: the +`TaskitoObservation` middleware wraps each task execution in a Micrometer +`Observation`, which a `PrometheusMeterRegistry` turns into a timer (and, if you +add a tracing bridge, a span). See the dedicated +[Micrometer guide](/java/guides/integrations/micrometer) for the full API. + +## Setup + +The SDK compiles against `io.micrometer:micrometer-observation` as `compileOnly`, +so add it plus a Prometheus registry to your build: + +- `io.micrometer:micrometer-observation` +- `io.micrometer:micrometer-registry-prometheus` (brings `micrometer-core`, which + provides `DefaultMeterObservationHandler` and `PrometheusMeterRegistry`) + +```java +import io.micrometer.core.instrument.observation.DefaultMeterObservationHandler; +import io.micrometer.observation.ObservationRegistry; +import io.micrometer.prometheusmetrics.PrometheusConfig; +import io.micrometer.prometheusmetrics.PrometheusMeterRegistry; +import org.byteveda.taskito.contrib.TaskitoObservation; + +PrometheusMeterRegistry prometheus = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT); +ObservationRegistry observations = ObservationRegistry.create(); +observations.observationConfig() + .observationHandler(new DefaultMeterObservationHandler(prometheus)); + +taskito.use(new TaskitoObservation(observations)); +``` + +## What's emitted + +Each execution attempt starts an observation named `taskito.task` (customizable +via `new TaskitoObservation(registry, name, taskFilter)`) tagged with +`taskito.task` = the task name. Micrometer's meter handler converts it into a +timer on the Prometheus registry; the exposed series name and suffixes follow +Micrometer's Prometheus naming convention (dots become underscores, a `_seconds` +suffix is added), not a fixed `taskito_*` name from the contrib. Retries are +separate attempts, each producing its own observation. + +## Exposing metrics + +Serve `prometheus.scrape()` from any HTTP endpoint, or use Spring Boot Actuator's +`/actuator/prometheus`, which wires the registry for you. The `taskito_jobs_total` +/ `taskito_dlq_size` names and the polled queue-depth gauges below are Python/Node +only — build Java dashboards and alerts on the timer series your +Micrometer→Prometheus registry emits. diff --git a/docs/content/docs/shared/guides/reliability/guarantees.mdx b/docs/content/docs/shared/guides/reliability/guarantees.mdx index 5dee2900..a6a37a34 100644 --- a/docs/content/docs/shared/guides/reliability/guarantees.mdx +++ b/docs/content/docs/shared/guides/reliability/guarantees.mdx @@ -133,8 +133,14 @@ queue.task("charge", async (orderId: string) => { ```java +// Java handlers receive only the payload, so idempotency comes from a +// natural key in it — here order.id(). Worker worker = queue.worker() - .handle(CHARGE, order -> payments.charge(order.total(), order.id())) + .handle(CHARGE, order -> { + if (charges.exists(order.id())) return; // already processed on a prior attempt + payments.charge(order.total(), order.id()); // key also dedupes at the provider + charges.record(order.id(), order.total()); + }) .start(); ``` diff --git a/docs/content/docs/shared/guides/reliability/retries.mdx b/docs/content/docs/shared/guides/reliability/retries.mdx index d8bf6f6d..3cc46413 100644 --- a/docs/content/docs/shared/guides/reliability/retries.mdx +++ b/docs/content/docs/shared/guides/reliability/retries.mdx @@ -149,6 +149,20 @@ retries fire immediately (no exponential fallback). + + +Node has no per-attempt delay list — tune the exponential curve instead via +`retryBackoff` (`baseMs` is `B`, `maxMs` is `M` in the backoff formula above): + +```ts +queue.task("flakyApiCall", handleCall, { + maxRetries: 5, + retryBackoff: { baseMs: 2000, maxMs: 300_000 }, // 2s base, 5min cap +}); +``` + + + ## Exception filtering diff --git a/docs/content/docs/shared/guides/workflows/canvas.mdx b/docs/content/docs/shared/guides/workflows/canvas.mdx index 72f4f707..f7dfdefd 100644 --- a/docs/content/docs/shared/guides/workflows/canvas.mdx +++ b/docs/content/docs/shared/guides/workflows/canvas.mdx @@ -540,10 +540,10 @@ return a plain `Workflow`, and `Workflow.named(...).step(...)` exposes a - - ## chunks / starmap + + ```python from taskito import chunks, starmap @@ -560,8 +560,58 @@ result = chord( results = starmap(add, [(1, 2), (3, 4), (5, 6)]).apply(queue) ``` + + + + +Node has no `chunks` / `starmap` helper — both are just a `group` of pre-sized +steps. Build the batches in JS and hand them to `.group()`: + +```ts +// chunks: split items into batches of 100 +const size = 100; +const batches = Array.from({ length: Math.ceil(items.length / size) }, (_, i) => + items.slice(i * size, (i + 1) * size), +); +queue.workflows + .define("batch") + .group(batches.map((b, i) => ({ name: `chunk-${i}`, task: "processBatch", args: [b] }))) + .submit(); + +// starmap: one step per arg tuple +queue.workflows + .define("pairs") + .group([[1, 2], [3, 4]].map(([a, b], i) => ({ name: `add-${i}`, task: "add", args: [a, b] }))) + .submit(); +``` + +For dynamic batching driven by an upstream result, use `fanOut` / `fanIn`. + + + + + +Java has no `chunks` / `starmap` shortcut — they are a `group` of fixed steps. +Slice in Java and pass the links to `Canvas.group`: + +```java +List links = new ArrayList<>(); +for (int i = 0; i < items.size(); i += 100) { + List batch = items.subList(i, Math.min(i + 100, items.size())); + links.add(Canvas.link("chunk-" + (i / 100), processBatch, batch)); +} +Workflow wf = Canvas.group("batch", links.toArray(new Canvas.Link[0])); +queue.submitWorkflow(wf); +``` + +For dynamic per-item fan-out, use `Step.Builder.fanOut(...)` / `.fanIn(...)`. + + + ## Canvas vs DAG workflows + + | Feature | Canvas | DAG Workflows | |---------|--------|---------------| | Setup | No imports needed | `from taskito.workflows import Workflow` | @@ -582,3 +632,20 @@ production pipelines that need monitoring, conditions, or complex topologies. + + + +Canvas is not a separate engine in Node — `chain` / `group` / `chord` are +methods on `WorkflowBuilder` that produce an ordinary DAG workflow. Reach for +conditions, gates, or sub-workflows on the same builder when you need them; see +Workflows. + + + + + +`Canvas.chain` / `group` / `chord` return a plain `Workflow` you can keep +extending with the builder (conditions, gates, sub-workflows) — canvas and DAG +are the same object. See Workflows. + + diff --git a/docs/content/docs/shared/guides/workflows/gates.mdx b/docs/content/docs/shared/guides/workflows/gates.mdx index 8ec7ccdd..f3974429 100644 --- a/docs/content/docs/shared/guides/workflows/gates.mdx +++ b/docs/content/docs/shared/guides/workflows/gates.mdx @@ -320,10 +320,10 @@ is cancelled if the gate is resolved manually first. - - ## Gate conditions + + Gates respect the same `condition` DSL as regular steps, evaluated against the gate's predecessors *before* it's entered: @@ -336,10 +336,24 @@ wf.step("deploy", deploy_task, after="approve") If `test` fails, the gate is skipped (its condition isn't met) — and `deploy`, the gate's successor, is skipped too. - - Node's `.gate()` step options and Java's `GateConfig`/`Workflow.gate()` - don't expose a condition on the gate node itself — only Python's - `Workflow.gate()` takes one directly. + + + + + + A Node gate node has no entry condition — `.gate()` options cover only the + predecessor(s), `timeoutMs`, `onTimeout`, and `message`. To make a gate + conditional, put the `condition` on the step(s) around it instead. + + + + + + + + A Java gate node has no entry condition — `GateConfig` covers only `timeout`, + `onTimeout`, and `message`. To make a gate conditional, put the condition on + the step(s) around it instead. @@ -381,10 +395,10 @@ rejection branch. - - ## Events + + When a gate enters `WAITING_APPROVAL`, a `WORKFLOW_GATE_REACHED` event fires in the process that reached it — the same process whose tracker owns the run (see the note under "Resolving a gate"). Subscribe with `queue.on_event()` @@ -405,6 +419,29 @@ The payload carries `run_id`, `node_name`, and `message` (the gate's + + + + There is no gate/workflow event in Node — `queue.on(...)` delivers only + job-outcome events (`job.completed`, `job.retrying`, `job.dead`, + `job.cancelled`). To react to a gate being reached, notify from the resolver + side (the endpoint that calls `approveGate` / `rejectGate`), or poll node + status for `waiting_approval` (see "Resolving a gate"). + + + + + + + + There is no gate/workflow event in Java — the worker's event listeners + deliver only job outcomes (`SUCCESS`, `RETRY`, `DEAD`, `CANCELLED`). To react + to a gate being reached, notify from the resolver side, or poll node status + for `WAITING_APPROVAL` (see "Resolving a gate"). + + + + ## Webhooks and external triggers A common pattern is an HTTP endpoint that receives a webhook and resolves the diff --git a/docs/content/docs/shared/guides/workflows/saga.mdx b/docs/content/docs/shared/guides/workflows/saga.mdx index 4349ccd9..ea8d7fa1 100644 --- a/docs/content/docs/shared/guides/workflows/saga.mdx +++ b/docs/content/docs/shared/guides/workflows/saga.mdx @@ -324,8 +324,12 @@ Workflow partial = Workflow.named("partial-saga") on a cancelled run) and `COMPLETED_WITH_FAILURES`, covered next. + + ## Compensating on continue-mode workflows + + By default, `on_failure="continue"` workflows never trigger compensation — they run every step regardless of individual failures and always reach `COMPLETED_WITH_FAILURES` if any step failed. Opt in with @@ -353,6 +357,29 @@ runs. + + + + The Node builder has no run-level failure policy, so there is no + `compensateOnContinue` equivalent. `on_failure` here is only a per-step + `condition` (`"on_success"` | `"on_failure"` | `"always"`), not a + continue-vs-abort mode for the whole run. Saga compensation is declared + per step with `compensate` and runs only when the run itself fails. + + + + + + + + The `Workflow` builder has no run-level failure policy and no + `compensateOnContinue`. `onFailure()` is only a per-step `condition` + shorthand. Compensation is declared per step with + `Step.Builder.compensate(...)` and rolls back only when the run fails. + + + + ## Idempotency Every compensation job is enqueued with a deterministic idempotency key — @@ -418,10 +445,10 @@ Workflow safeCheckout = Workflow.named("safe-checkout") - - ## Events + + Saga lifecycle emits dedicated events on the queue's event bus, subscribed via `queue.on_event(event_type, callback)`: @@ -437,6 +464,28 @@ via `queue.on_event(event_type, callback)`: + + + + The Node SDK emits only the four job-lifecycle events — `job.completed`, + `job.retrying`, `job.dead`, `job.cancelled` — via `queue.on(...)`. There + are no `WORKFLOW_COMPENSATING` / `NODE_COMPENSATED`-style saga events. Track + compensation progress by polling the run handle: `handle.status()` and + `handle.nodes()`. + + + + + + + + Java's `EventName` enum has only the four job outcomes — `SUCCESS`, `RETRY`, + `DEAD`, `CANCELLED`. No workflow/saga transition events are emitted. Poll + `WorkflowRun` / `status.nodes` for compensation progress. + + + + ## See also - Canvas primitives for