diff --git a/NODE_SDK_TODO.md b/NODE_SDK_TODO.md new file mode 100644 index 00000000..6d1fef83 --- /dev/null +++ b/NODE_SDK_TODO.md @@ -0,0 +1,269 @@ +# Node SDK — Parity TODO & Status + +Living plan for the **Node.js SDK** (`crates/taskito-node` napi-rs crate + `sdks/node` +TS package). Goal: a fully standalone Node SDK with feature parity to the Python +SDK, zero Python dependency, over the shared Rust core. + +- **Branch:** landing as a sequence of focused PRs off `master` (see `tasks/branch-split-plan.md`). +- **Engine crates stay binding-agnostic:** `taskito-core`, `taskito-workflows`, + `taskito-mesh` must never depend on `pyo3` or `napi`. Enforced by a `cargo tree` + tripwire in CI. The napi shell lives only in `crates/taskito-node`. +- **Tooling:** always `pnpm` (never npm/npx). Rust via `cargo`. Pre-commit hooks + (`cargo fmt`, `cargo clippy`, biome) must pass — fix, don't bypass. + +--- + +## Done + +| Area | Commit(s) | Notes | +|------|-----------|-------| +| Production core | (earlier) | sqlite/postgres/redis backends, enqueue + idempotency, per-task retry/timeout/concurrency/rate-limit, cancellation, inspection/management, JSON + msgpack serializers | +| Worker lifecycle | `c8a9246`, `a77c592` | register + heartbeat (5s) + unregister; `list_workers`; dashboard workers panel | +| Outcomes → events | `5bf451e`, `3922a3e` | result-drain surfaces `ResultOutcome` to a JS callback; `Emitter` + `Middleware` | +| Events + middleware | `70d176e`, `704b4d8` | `job.completed/retrying/dead/cancelled`; before/after/onError + outcome hooks | +| Typed task registry | `7b4a057` | `Queue`, chainable `task()`, `enqueue` infers arg types; `RateLimit` template type | +| Settings KV | `f54f523` | get/set/delete/list settings | +| Webhooks | `1e09a08`, `174590c`, `755f66b`, `c3732e1`, `fc8a67e` | KV-persisted store, HMAC-SHA256 signed delivery + retries, `WebhookManager`, dashboard endpoints | +| CLI | (earlier) | standalone `taskito` bin: enqueue/stats/jobs/dlq/queues/cancel/run/dashboard | +| Dashboard | (earlier) | serves the **existing React SPA** + `/api/*` snake_case contract + open-auth stub; metrics + workers panels | +| **Mesh** | `bb20625` | `mesh` cargo feature; `MeshWorkerConfig` on `runWorker`; `run_mesh_bridge` in `worker.rs` | +| **Workflows (DAG v1)** | `4927a03` | submit/advance/query; core sequences via `depends_on`; TS builder DSL + `queue.workflows` | +| **Logger** | `228d16b` | `src/utils/logger.ts` — leveled, lazy thunks, namespaced children, pluggable sink | +| **Dashboard workflows panel** | `1be346b` | 4 GET endpoints serving the SPA workflows page | +| **Distributed locks** | `0655349` | `queue.lock` / `queue.withLock`, auto-extend, `Symbol.dispose` | +| **Periodic + circuit-breaker** | `a4cc9a4` | `queue.registerPeriodic`; per-task `circuitBreaker` config | +| **Workflows: fan-out / fan-in** | (this branch) | TS `WorkflowTracker` brain (`src/workflows/tracker.ts`) driven by the outcome callback; `.fanOut()` / `.fanIn()` builder steps; napi primitives `expandFanOut`/`checkFanOutCompletion`/`createDeferredJob`/`finalizeRunIfTerminal`/`getWorkflowRunPlan`/`workflowNodeForJob`/`cascadeSkipPending`; deferred-node submit. Storage-reconstructable (no submit-time registration) | +| **Dashboard DAG completeness** | (this branch) | `JsWorkflowNode` carries `fanOutCount` + `compensation_*`; `/dag` handler enriches the raw graph with per-node `deps[]` + live `status` + job-id `id` so the SPA visualizer renders edges/colours/links | + +**Verify everything green:** +```bash +cargo clippy -p taskito-node --features postgres,redis,mesh,workflows -- -D warnings +cargo check -p taskito-node # default build (mesh/workflows off) +cd sdks/node && pnpm run build:native && pnpm typecheck && pnpm lint && pnpm test # full Node suite +``` +`build:native` ships `--features postgres,redis,mesh,workflows`. + +--- + +## Left to do + +Ordered roughly by value / tractability. + +### 1. Resources / proxies / interception (Node-native equivalents) — LARGE + +Python has a worker dependency-injection runtime (`resources/`), transparent +non-serializable-object handling (`proxies/`), and argument interception +(`interception/`). These are **Python-idiom-specific** (decorators, cloudpickle, +context managers) — do **not** port 1:1. Design Node-native equivalents: + +- **Resources / DI:** a `queue.resource(name, factory, { scope })` registry with + scopes (worker / task / request); inject into handlers via a context accessor + (extend `context.ts` / `AsyncLocalStorage`) or an explicit `deps` argument. +- **Proxies:** likely unnecessary in Node — JS has no pickle problem. Handlers + close over their own resources. Skip unless a concrete need appears. +- **Interception:** argument validation / redaction hook before serialization in + `Queue.enqueue`. Could be a lightweight middleware-style `beforeEnqueue` hook. + +**Where:** new `sdks/node/src/resources/`; touch `worker.ts` (inject), `context.ts`. +**Effort:** large (design-heavy). **Recommendation:** scope a minimal resource DI +first (worker-scoped singletons + task-scoped factories); defer proxies entirely. + +### 2. Contrib integrations — MEDIUM + +Python ships `contrib/` (OpenTelemetry, Sentry, Prometheus, Flask, Django, +FastAPI). Node equivalents: + +- **Observability:** OpenTelemetry (`@opentelemetry/api`), Prometheus + (`prom-client`) — implement as middleware over the existing events/middleware + layer (`src/middleware.ts`). Each = a `Middleware` that records spans/metrics. +- **Web frameworks:** Express / Fastify / Nest helpers (enqueue from a request, + mount the dashboard). Python's Flask/Django/FastAPI map to these. + +**Where:** new `sdks/node/src/contrib/` (one file per integration, optional peer +deps). **Effort:** medium. Each integration is small and independent → **one +commit each**. + +### 3. Advanced workflow features — LARGE (fan-out DONE; gates / sub-workflows / saga remain) + +DAG/linear + **fan-out / fan-in** work. The tracker-brain foundation is built: +a TS `WorkflowTracker` (`src/workflows/tracker.ts`) reacts to the worker outcome +callback, reconstructs the run plan from storage (DAG + step metadata — no +submit-time registration), and drives on-demand orchestration via napi +primitives (`expandFanOut`, `checkFanOutCompletion`, `createDeferredJob`, +`finalizeRunIfTerminal`, `getWorkflowRunPlan`, `workflowNodeForJob`, +`cascadeSkipPending`). `submit_workflow` takes `deferredNodeNames` (fan-out / +fan-in ∪ descendants get a node but no static job). Failures are fail-fast. + +**Still unbound: gates/conditions, sub-workflows, saga compensation.** They build +on this same foundation (the Python references are +`crates/taskito-python/src/py_queue/workflow_ops/{gates,saga}.rs` + the Python +`WorkflowTracker`): +- **gates/conditions** — a deferred node enters `waiting_approval`; resolve via a + new `resolveWorkflowGate` napi method + a JS-side timer (`setTimeout`) for + timeouts. Conditions = a `should_execute` check in `evaluateSuccessors`. +- **sub-workflows** — extend `submit_workflow` with `parent_run_id` / + `parent_node_name`; the tracker submits a child run for a sub-workflow node and + resolves the parent node when the child finalizes (populates `/children`). +- **saga** — reverse-topo compensation waves driven by the tracker; needs the + `setWorkflowNodeCompensation*` + run-state napi setters bound (storage methods + already exist on the `WorkflowStorage` trait). + +**Where:** `crates/taskito-node/src/queue/workflows.rs` + `src/workflows/tracker.ts`. +**Effort:** medium each now the brain exists. Each is a separate feature/commit. + +### 4. Dashboard workflows DAG panel completeness — DONE + +`JsWorkflowNode` now carries `fanOutCount` + `compensation_*`; `contract.ts` maps +them (saga fields stay null until saga is bound). The `/runs/:id/dag` handler +(`dashboard/handlers.ts::enrichDag`) rewrites the raw `SerializableGraph` into the +shape the SPA visualizer actually consumes — per-node `deps[]` (from edges), live +`status` (merged from the nodes list), and `id` = job id for correct `/jobs/$id` +links — so the DAG tab renders real edges, colours, and links. `/runs/:id/children` +stays `[]` until sub-workflows exist (#3). + +### 5. Prebuilt platform matrix + npm publish — MEDIUM (infra) + +Today `build:native` is **host-only** (builds for the current platform). For npm: + +- Build prebuilt `.node` binaries per platform/arch (linux x64/arm64, macOS + x64/arm64, win x64) via `@napi-rs/cli` + CI matrix (GitHub Actions). +- napi-rs `optionalDependencies` per-platform package pattern, OR bundle prebuilds. +- npm publish workflow (mirror the Python `publish.yml`; OIDC / token). +- Decide package name + scope. + +**Where:** `.github/workflows/`, `sdks/node/package.json` (`napi` config, `files`, +`optionalDependencies`). **Effort:** medium, mostly CI plumbing. + +### 6. Python⇄Node interop — OPTIONAL (deferred) + +Cross-language jobs over **shared storage** (Python enqueues → Node worker runs, +and vice versa). **Not required for the standalone goal** — only matters for +mixed-language deployments on one DB. + +**Already partially free:** core + storage + metadata/notes/workflow tables are +shared. A Python task with **positional-only args + JSON/msgpack** serializer is +runnable by a Node worker today (if the task name is registered both sides) — +**unverified**. + +**Blockers:** +- Python default serializer = **cloudpickle** → opaque to Node. Interop forces + JSON/msgpack on both sides. +- **kwargs mismatch:** Python args = `(args tuple, kwargs dict)`; Node = positional + array, no kwargs. Need a shared payload envelope (e.g. msgpack `{args, kwargs}`). + +**Three options (decision pending):** +- `defer` — skip until a mixed deployment is actually needed (recommended). +- `verify-doc-only` — write interop tests for the simple JSON-positional case, + document which codecs/arg-shapes interop, ship that as the contract. +- `full-interop` — define + implement the cross-lang msgpack envelope (args + + kwargs), Node (de)serializer for it, interop tests across all 3 backends, docs. + +**Effort:** verify-doc-only = small; full-interop = medium. + +--- + +## Docs site — decision: **hybrid** (not yet built) + +Current docs (`docs/`, Fumadocs / Next.js 16) are **100% Python**, reference stale +`py_src` paths (51 files), have no `sdks/` section and no Node mention. + +**Chosen structure (hybrid):** +- `concepts/` — language-neutral (Rust core, storage, scheduler, mesh, workflow + engine, capabilities, glossary). Cross-cutting code samples use **tabbed + Python/Node snippets** (``). +- `python/` — full Python SDK tree (relocate current pages; fix the 51 stale refs). +- `node/` — full Node SDK tree (author from `sdks/node/README.md` + feature pages: + tasks, workers, cancellation, serializers, events+middleware, webhooks, locks, + workflows, mesh, periodic, cli, dashboard). +- `more/` — changelog, upgrading, contributing. + +**Single URL:** one Fumadocs deploy. SDK = path segment (`/docs/python/...`, +`/docs/node/...`). Fumadocs **root-folder sidebar tabs** (`"root": true` in each +SDK folder's `meta.json`) give a Python↔Node switch. No subdomains, no separate +sites. Landing page: pick Python | Node. + +**Why hybrid (not pure A/B):** the two SDK APIs diverge hard (Python decorators / +resources / proxies / interception vs Node typed registry / chainable builder / +`using` locks), so per-SDK trees avoid litter of "X only" callouts; neutral +concepts written once with tabs where code is genuinely parallel. + +**Migration steps:** (1) `concepts/` ← move `architecture/` + `capabilities`; +(2) `python/` ← relocate current pages, fix stale `py_src` refs; (3) `node/` ← +author new; (4) root `meta.json` order + landing SDK pick. + +--- + +## Key technical facts / gotchas + +- **DAG bytes = plain JSON** of `dagron_core::SerializableGraph`: + `{nodes:[{name}], edges:[{from,to,weight}]}`. Built directly in TS — no + DAG-builder binding needed. `taskito_workflows::topological_order` is `pub`. +- **Workflow advancement** rides the worker **outcome callback** + (`queue.markWorkflowNodeResult` on success/dead). For DAGs the **core scheduler + sequences nodes via `depends_on`** — no Python-style tracker for the happy path. + Fan-out/gates/saga need a real tracker (see #3). +- **Cron = `cron` crate format: 6/7-field, seconds first** — + `sec min hour day-of-month month day-of-week [year]`. NOT 5-field. e.g. hourly = + `0 0 * * * *`. +- **Periodic dispatch already runs** in the worker maintenance loop (so a Node + worker fires periodic tasks automatically); only registration is bound. +- **Lock auto-extend** uses an unref'd `setInterval` at `ttlMs/3`; `Lock` + implements `Symbol.dispose` so `using lock = queue.lock(...)` works. +- **napi cross-dir build:** `napi build --platform --release --cargo-cwd + ../../crates/taskito-node --features ... native`. Generated `native/index.js` is + CJS (nested `native/package.json` = `{"type":"commonjs"}`); loaded via runtime + `createRequire` + `new URL` so tsup leaves it external. +- **Feature gating:** `workflows` + `mesh` are cargo features (optional deps). + `postgres`/`redis` wire workflow-crate features via weak deps + (`taskito-workflows?/postgres`). Locks/periodic/circuit-breaker are **ungated** + (core storage). +- **Dashboard serves the existing React SPA** (`dashboard/`), built to + `sdks/node/static/dashboard` via `pnpm build:dashboard`. SPA expects + **snake_case** + Unix-ms; `dashboard/contract.ts` maps camelCase → snake_case. + +--- + +## Build / verify reference + +```bash +# Rust +cargo clippy -p taskito-node --features postgres,redis,mesh,workflows -- -D warnings +cargo check -p taskito-node # default (no mesh/workflows) +cargo check -p taskito-node --features workflows # weak-dep sanity +# crate purity tripwire (must print nothing): +cargo tree -p taskito-workflows -e normal | grep -iE 'pyo3|napi' +cargo tree -p taskito-mesh -e normal | grep -iE 'pyo3|napi' + +# Node (from sdks/node/) +pnpm run build:native # napi build, all features +pnpm run build:ts # tsup dual ESM/CJS + .d.ts +pnpm typecheck # tsc --noEmit (includes test/) +pnpm lint # biome +pnpm test # vitest (48 tests) +pnpm run build:dashboard # build the React SPA into static/dashboard +``` + +--- + +## File map (Node SDK) + +```text +crates/taskito-node/src/ + lib.rs backend.rs dispatcher.rs worker.rs error.rs config.rs + queue/{mod,inspect,admin,locks,periodic,workflows}.rs + convert/{mod,job,stats,outcome,task_config,lock,workflow}.rs + +sdks/node/src/ + index.ts native.ts queue.ts worker.ts context.ts errors.ts types.ts + utils/{logger,index}.ts + locks/{lock,types,index}.ts + workflows/{builder,manager,tracker,plan,types,index}.ts + webhooks/{store,deliverer,manager,types,index}.ts + serializers/{serializer,json,msgpack,index}.ts + dashboard/{server,routes,handlers,contract,static,metrics,api,index}.ts + cli/{index,connect,output,commands/*}.ts +sdks/node/test/*.test.ts # grouped by feature area +``` + +Memory: see `.claude/memory/session-history.md` (Node SDK section) for the running +log of decisions. diff --git a/crates/taskito-node/Cargo.toml b/crates/taskito-node/Cargo.toml index da6e637c..d391414f 100644 --- a/crates/taskito-node/Cargo.toml +++ b/crates/taskito-node/Cargo.toml @@ -11,8 +11,10 @@ crate-type = ["cdylib"] [features] default = [] -postgres = ["taskito-core/postgres"] -redis = ["taskito-core/redis"] +postgres = ["taskito-core/postgres", "taskito-workflows?/postgres"] +redis = ["taskito-core/redis", "taskito-workflows?/redis"] +mesh = ["dep:taskito-mesh"] +workflows = ["dep:taskito-workflows"] [dependencies] taskito-core = { path = "../taskito-core" } @@ -23,7 +25,10 @@ crossbeam-channel = { workspace = true } async-trait = { workspace = true } log = { workspace = true } uuid = { workspace = true } +serde_json = { workspace = true } gethostname = "1" +taskito-mesh = { path = "../taskito-mesh", optional = true } +taskito-workflows = { path = "../taskito-workflows", optional = true } [build-dependencies] napi-build = "2" diff --git a/crates/taskito-node/src/config.rs b/crates/taskito-node/src/config.rs index a74d42c3..2b315c31 100644 --- a/crates/taskito-node/src/config.rs +++ b/crates/taskito-node/src/config.rs @@ -53,6 +53,19 @@ pub struct JobFilter { pub offset: Option, } +/// Circuit-breaker config for a task: trip after `threshold` failures within +/// `windowMs`, stay open for `cooldownMs`, then probe before fully closing. +#[napi(object)] +pub struct CircuitBreakerInput { + pub threshold: i32, + pub window_ms: i64, + pub cooldown_ms: i64, + /// Probe requests allowed while half-open (default 5). + pub half_open_max_probes: Option, + /// Success rate (0.0–1.0) required to close from half-open (default 0.8). + pub half_open_success_rate: Option, +} + /// Per-task configuration registered on the scheduler before the worker runs. #[napi(object)] pub struct TaskConfigInput { @@ -63,6 +76,7 @@ pub struct TaskConfigInput { pub max_concurrent: Option, /// Rate-limit spec like `"100/m"`, `"50/s"`, `"3600/h"`. pub rate_limit: Option, + pub circuit_breaker: Option, } /// Per-queue configuration registered on the scheduler before the worker runs. @@ -73,6 +87,38 @@ pub struct QueueConfigInput { pub rate_limit: Option, } +/// Opt-in mesh overlay for a worker: decentralized peer discovery (SWIM gossip) +/// plus work-stealing across nodes. Ignored unless the addon is built with the +/// `mesh` cargo feature. The DB stays the source of truth — mesh only optimizes +/// dispatch locality. +#[napi(object)] +pub struct MeshWorkerConfig { + /// UDP gossip port. The TCP work-stealing server binds `port + 1`. + pub port: u32, + /// Bind address for both servers (default `"0.0.0.0"`). + pub bind_addr: Option, + /// Seed peers to join, as `"host:gossip_port"` strings. + pub seeds: Option>, + /// Enable work-stealing from busier peers (default true). + pub steal: Option, + /// Affinity weight for consistent-hash placement (default from core). + pub affinity_weight: Option, + /// Local deque capacity before back-pressure. + pub local_buffer: Option, + /// Max jobs pulled in one steal. + pub steal_batch: Option, + /// Deque depth at or below which this node tries to steal. + pub steal_threshold: Option, + /// Virtual nodes per peer on the hash ring. + pub virtual_nodes: Option, + /// Address advertised to peers when behind NAT (`"host:port"`). + pub advertise_addr: Option, + /// Shared key enabling XOR gossip encryption (must match across the mesh). + pub encryption_key: Option, + /// Per-peer steal rate limit (steals per second). + pub steal_rate_limit: Option, +} + /// Options for a running worker. `queues` defaults to `["default"]`. #[napi(object)] #[derive(Default)] @@ -83,4 +129,6 @@ pub struct WorkerOptions { pub batch_size: Option, pub task_configs: Option>, pub queue_configs: Option>, + /// Opt-in decentralized mesh overlay (requires the `mesh` build feature). + pub mesh: Option, } diff --git a/crates/taskito-node/src/convert/lock.rs b/crates/taskito-node/src/convert/lock.rs new file mode 100644 index 00000000..d6f15d88 --- /dev/null +++ b/crates/taskito-node/src/convert/lock.rs @@ -0,0 +1,22 @@ +//! JS-facing shape for distributed lock info. Timestamps are Unix milliseconds. + +use napi_derive::napi; +use taskito_core::storage::models::LockInfoRow; + +/// JS-facing view of a held distributed lock. +#[napi(object)] +pub struct JsLockInfo { + pub lock_name: String, + pub owner_id: String, + pub acquired_at: i64, + pub expires_at: i64, +} + +pub fn lock_info_to_js(row: LockInfoRow) -> JsLockInfo { + JsLockInfo { + lock_name: row.lock_name, + owner_id: row.owner_id, + acquired_at: row.acquired_at, + expires_at: row.expires_at, + } +} diff --git a/crates/taskito-node/src/convert/mod.rs b/crates/taskito-node/src/convert/mod.rs index 96da4b0a..b3720d9b 100644 --- a/crates/taskito-node/src/convert/mod.rs +++ b/crates/taskito-node/src/convert/mod.rs @@ -2,14 +2,23 @@ //! concern; kept out of the logic modules so they read as intent, not plumbing. mod job; +mod lock; mod outcome; mod stats; mod task_config; +#[cfg(feature = "workflows")] +mod workflow; pub use job::{build_new_job, job_to_js, JsJob, JsTaskInvocation}; +pub use lock::{lock_info_to_js, JsLockInfo}; pub use outcome::{outcome_to_js, JsOutcome}; pub use stats::{ dead_job_to_js, job_error_to_js, metric_to_js, stats_to_js, status_code, worker_to_js, JsDeadJob, JsJobError, JsMetric, JsStats, JsWorkerRow, }; pub use task_config::{queue_config, task_config}; +#[cfg(feature = "workflows")] +pub use workflow::{ + node_to_js, run_to_js, JsFanOutCompletion, JsWorkflowAdvance, JsWorkflowNode, + JsWorkflowNodeRef, JsWorkflowRun, JsWorkflowRunPlan, +}; diff --git a/crates/taskito-node/src/convert/task_config.rs b/crates/taskito-node/src/convert/task_config.rs index 9961dac6..5dc627f7 100644 --- a/crates/taskito-node/src/convert/task_config.rs +++ b/crates/taskito-node/src/convert/task_config.rs @@ -1,16 +1,19 @@ //! Build core task/queue configuration from JS option inputs. use napi::bindgen_prelude::Result; +use taskito_core::resilience::circuit_breaker::CircuitBreakerConfig; use taskito_core::resilience::rate_limiter::RateLimitConfig; use taskito_core::resilience::retry::RetryPolicy; use taskito_core::scheduler::{QueueConfig, TaskConfig}; -use crate::config::{QueueConfigInput, TaskConfigInput}; +use crate::config::{CircuitBreakerInput, QueueConfigInput, TaskConfigInput}; use crate::error::invalid_arg; const DEFAULT_RETRY_BASE_MS: i64 = 1_000; const DEFAULT_RETRY_MAX_MS: i64 = 300_000; const DEFAULT_MAX_RETRIES: i32 = 3; +const DEFAULT_HALF_OPEN_PROBES: i32 = 5; +const DEFAULT_HALF_OPEN_SUCCESS_RATE: f64 = 0.8; /// Parse an optional rate-limit spec, failing fast on a malformed value rather /// than silently disabling throttling (a misconfigured limit is a config error). @@ -33,11 +36,51 @@ pub fn task_config(input: &TaskConfigInput) -> Result { custom_delays_ms: None, }, rate_limit: parse_rate_limit(input.rate_limit.as_deref())?, - circuit_breaker: None, + circuit_breaker: input + .circuit_breaker + .as_ref() + .map(circuit_breaker_config) + .transpose()?, max_concurrent: input.max_concurrent, }) } +/// Build a [`CircuitBreakerConfig`] from JS input, validating bounds and filling +/// half-open defaults. Rejects non-positive/out-of-range values that would make +/// the breaker misbehave. +fn circuit_breaker_config(input: &CircuitBreakerInput) -> Result { + if input.threshold <= 0 { + return Err(invalid_arg("circuitBreaker.threshold must be > 0")); + } + if input.window_ms <= 0 { + return Err(invalid_arg("circuitBreaker.windowMs must be > 0")); + } + if input.cooldown_ms <= 0 { + return Err(invalid_arg("circuitBreaker.cooldownMs must be > 0")); + } + let half_open_max_probes = input + .half_open_max_probes + .unwrap_or(DEFAULT_HALF_OPEN_PROBES); + if half_open_max_probes <= 0 { + return Err(invalid_arg("circuitBreaker.halfOpenMaxProbes must be > 0")); + } + let half_open_success_rate = input + .half_open_success_rate + .unwrap_or(DEFAULT_HALF_OPEN_SUCCESS_RATE); + if !(0.0..=1.0).contains(&half_open_success_rate) { + return Err(invalid_arg( + "circuitBreaker.halfOpenSuccessRate must be in 0.0..=1.0", + )); + } + Ok(CircuitBreakerConfig { + threshold: input.threshold, + window_ms: input.window_ms, + cooldown_ms: input.cooldown_ms, + half_open_max_probes, + half_open_success_rate, + }) +} + /// Build a [`QueueConfig`] (rate limit, concurrency cap) from JS input. pub fn queue_config(input: &QueueConfigInput) -> Result { Ok(QueueConfig { diff --git a/crates/taskito-node/src/convert/workflow.rs b/crates/taskito-node/src/convert/workflow.rs new file mode 100644 index 00000000..d4c07e68 --- /dev/null +++ b/crates/taskito-node/src/convert/workflow.rs @@ -0,0 +1,116 @@ +//! JS-facing shapes for workflow runs and nodes. State enums are surfaced as +//! their lowercase string names (matching the rest of the SDK and the dashboard +//! contract). + +use napi_derive::napi; +use taskito_workflows::{WorkflowNode, WorkflowRun}; + +/// JS-facing view of a [`WorkflowRun`]. +#[napi(object)] +pub struct JsWorkflowRun { + pub id: String, + pub definition_id: String, + /// Lowercase run state: `pending`, `running`, `completed`, `failed`, + /// `cancelled`, `completed_with_failures`, … + pub state: String, + pub params: Option, + pub error: Option, + pub started_at: Option, + pub completed_at: Option, + pub created_at: i64, + pub parent_run_id: Option, + pub parent_node_name: Option, +} + +/// JS-facing view of a [`WorkflowNode`] (one step of a run). +#[napi(object)] +pub struct JsWorkflowNode { + pub run_id: String, + pub node_name: String, + pub job_id: Option, + /// Lowercase node status: `pending`, `ready`, `running`, `completed`, + /// `failed`, `skipped`, … + pub status: String, + pub error: Option, + pub result_hash: Option, + /// Number of children a fan-out node expanded into (set on the parent). + pub fan_out_count: Option, + pub started_at: Option, + pub completed_at: Option, + /// Saga: job id of the compensation enqueued for this node, if any. + pub compensation_job_id: Option, + pub compensation_started_at: Option, + pub compensation_completed_at: Option, + pub compensation_error: Option, +} + +/// Result of advancing a workflow node — returned by `markWorkflowNodeResult`. +/// `finalState` is set only when the whole run reached a terminal state. +#[napi(object)] +pub struct JsWorkflowAdvance { + pub run_id: String, + pub node_name: String, + pub final_state: Option, +} + +/// Identifies the workflow run + node a job belongs to. Returned by +/// `workflowNodeForJob` so the tracker can classify an outcome without an +/// in-memory job→run map. +#[napi(object)] +pub struct JsWorkflowNodeRef { + pub run_id: String, + pub node_name: String, +} + +/// Outcome of a fan-out completion check. Returned by `checkFanOutCompletion` +/// only when this caller performed the parent's terminal transition. +#[napi(object)] +pub struct JsFanOutCompletion { + /// Whether every child completed successfully. + pub succeeded: bool, + /// Job ids of the children, in storage order. + pub child_job_ids: Vec, +} + +/// The DAG + step metadata backing a run, so the tracker can reconstruct the +/// workflow structure from storage (no submit-time registration needed). +#[napi(object)] +pub struct JsWorkflowRunPlan { + /// `SerializableGraph` JSON (`{nodes:[{name}], edges:[{from,to,weight}]}`). + pub dag: String, + /// `HashMap` JSON, keyed by node name. + pub step_metadata: String, +} + +pub fn run_to_js(run: WorkflowRun) -> JsWorkflowRun { + JsWorkflowRun { + id: run.id, + definition_id: run.definition_id, + state: run.state.as_str().to_string(), + params: run.params, + error: run.error, + started_at: run.started_at, + completed_at: run.completed_at, + created_at: run.created_at, + parent_run_id: run.parent_run_id, + parent_node_name: run.parent_node_name, + } +} + +pub fn node_to_js(node: WorkflowNode) -> JsWorkflowNode { + JsWorkflowNode { + run_id: node.run_id, + node_name: node.node_name, + job_id: node.job_id, + status: node.status.as_str().to_string(), + error: node.error, + result_hash: node.result_hash, + fan_out_count: node.fan_out_count, + started_at: node.started_at, + completed_at: node.completed_at, + compensation_job_id: node.compensation_job_id, + compensation_started_at: node.compensation_started_at, + compensation_completed_at: node.compensation_completed_at, + compensation_error: node.compensation_error, + } +} diff --git a/crates/taskito-node/src/queue/locks.rs b/crates/taskito-node/src/queue/locks.rs new file mode 100644 index 00000000..69989d0f --- /dev/null +++ b/crates/taskito-node/src/queue/locks.rs @@ -0,0 +1,54 @@ +//! Distributed locking over the core storage. Locks are advisory, TTL-bounded, +//! and owner-scoped; expired locks are reaped by the worker's maintenance loop. + +use napi::bindgen_prelude::Result; +use napi_derive::napi; +use taskito_core::Storage; + +use super::JsQueue; +use crate::convert::{lock_info_to_js, JsLockInfo}; +use crate::error::{invalid_arg, to_napi_err}; + +#[napi] +impl JsQueue { + /// Acquire `name` for `ownerId`, expiring after `ttlMs`. Returns false when + /// another owner already holds a live lock. + #[napi] + pub fn acquire_lock(&self, name: String, owner_id: String, ttl_ms: i64) -> Result { + if ttl_ms <= 0 { + return Err(invalid_arg(format!("ttlMs must be > 0 (got {ttl_ms})"))); + } + self.storage + .acquire_lock(&name, &owner_id, ttl_ms) + .map_err(to_napi_err) + } + + /// Release `name` if held by `ownerId`. Returns false if not the holder. + #[napi] + pub fn release_lock(&self, name: String, owner_id: String) -> Result { + self.storage + .release_lock(&name, &owner_id) + .map_err(to_napi_err) + } + + /// Extend `name`'s TTL to `ttlMs` if held by `ownerId`. Returns false otherwise. + #[napi] + pub fn extend_lock(&self, name: String, owner_id: String, ttl_ms: i64) -> Result { + if ttl_ms <= 0 { + return Err(invalid_arg(format!("ttlMs must be > 0 (got {ttl_ms})"))); + } + self.storage + .extend_lock(&name, &owner_id, ttl_ms) + .map_err(to_napi_err) + } + + /// Current holder info for `name`, or `null` if free. + #[napi] + pub fn get_lock_info(&self, name: String) -> Result> { + Ok(self + .storage + .get_lock_info(&name) + .map_err(to_napi_err)? + .map(lock_info_to_js)) + } +} diff --git a/crates/taskito-node/src/queue/mod.rs b/crates/taskito-node/src/queue/mod.rs index 67e98208..181b540d 100644 --- a/crates/taskito-node/src/queue/mod.rs +++ b/crates/taskito-node/src/queue/mod.rs @@ -12,12 +12,20 @@ use crate::worker::{start_worker, JsWorker}; mod admin; mod inspect; +mod locks; +mod periodic; +#[cfg(feature = "workflows")] +mod workflows; /// A Taskito queue handle (SQLite/Postgres/Redis) exposed to JavaScript. #[napi] pub struct JsQueue { storage: StorageBackend, namespace: Option, + /// Workflow storage, lazily initialized on first workflow call so the + /// workflow migrations only run when workflows are actually used. + #[cfg(feature = "workflows")] + workflow_storage: std::sync::OnceLock, } #[napi] @@ -27,7 +35,12 @@ impl JsQueue { pub fn open(options: OpenOptions) -> Result { let namespace = options.namespace.clone(); let storage = crate::backend::open(&options)?; - Ok(Self { storage, namespace }) + Ok(Self { + storage, + namespace, + #[cfg(feature = "workflows")] + workflow_storage: std::sync::OnceLock::new(), + }) } /// Enqueue `task_name` with an opaque serialized `payload`. Returns the job diff --git a/crates/taskito-node/src/queue/periodic.rs b/crates/taskito-node/src/queue/periodic.rs new file mode 100644 index 00000000..13991559 --- /dev/null +++ b/crates/taskito-node/src/queue/periodic.rs @@ -0,0 +1,57 @@ +//! Periodic (cron) task registration. The worker's maintenance loop enqueues +//! due tasks; this binds only registration. Re-registering an existing name +//! replaces it (upsert). + +use napi::bindgen_prelude::{Buffer, Result}; +use napi_derive::napi; +use taskito_core::job::now_millis; +use taskito_core::periodic::{next_cron_time, next_cron_time_tz}; +use taskito_core::storage::models::NewPeriodicTaskRow; +use taskito_core::Storage; + +use super::JsQueue; +use crate::error::to_napi_err; + +const DEFAULT_QUEUE: &str = "default"; + +#[napi] +impl JsQueue { + /// Register (or replace) a cron-scheduled task. `args` is the opaque, + /// already-serialized payload handed to the task when it fires. Returns the + /// next fire time (Unix ms); the worker maintenance loop enqueues the job + /// once due. Rejects an invalid cron expression. + #[napi] + #[allow(clippy::too_many_arguments)] + pub fn register_periodic( + &self, + name: String, + task_name: String, + cron_expr: String, + args: Option, + queue: Option, + timezone: Option, + enabled: Option, + ) -> Result { + let now = now_millis(); + let next_run = match &timezone { + Some(tz) => next_cron_time_tz(&cron_expr, now, tz), + None => next_cron_time(&cron_expr, now), + } + .map_err(to_napi_err)?; + + let args_bytes = args.map(|b| b.to_vec()); + let row = NewPeriodicTaskRow { + name: &name, + task_name: &task_name, + cron_expr: &cron_expr, + args: args_bytes.as_deref(), + kwargs: None, + queue: queue.as_deref().unwrap_or(DEFAULT_QUEUE), + enabled: enabled.unwrap_or(true), + next_run, + timezone: timezone.as_deref(), + }; + self.storage.register_periodic(&row).map_err(to_napi_err)?; + Ok(next_run) + } +} diff --git a/crates/taskito-node/src/queue/workflows.rs b/crates/taskito-node/src/queue/workflows.rs new file mode 100644 index 00000000..93556469 --- /dev/null +++ b/crates/taskito-node/src/queue/workflows.rs @@ -0,0 +1,700 @@ +//! Workflow submission, advancement, and queries. +//! +//! Static DAG steps are pre-enqueued with a `depends_on` chain so the core +//! scheduler runs them in topological order. Fan-out / fan-in steps are +//! deferred — submitted without a job — and the worker-side tracker (see +//! `sdks/node/src/workflows/tracker.ts`) expands and enqueues them at runtime +//! via the primitives below. Gates, sub-workflows, and saga compensation are +//! not yet bound. + +use std::collections::{HashMap, HashSet}; + +use napi::bindgen_prelude::{Buffer, Error, Result}; +use napi_derive::napi; +use taskito_core::job::{now_millis, NewJob}; +use taskito_core::{Storage, StorageBackend}; +use taskito_workflows::{ + topological_order, StepMetadata, WorkflowDefinition, WorkflowNode, WorkflowNodeStatus, + WorkflowRun, WorkflowSqliteStorage, WorkflowState, WorkflowStorage, WorkflowStorageBackend, +}; + +use super::JsQueue; +use crate::convert::{ + node_to_js, run_to_js, JsFanOutCompletion, JsWorkflowAdvance, JsWorkflowNode, + JsWorkflowNodeRef, JsWorkflowRun, JsWorkflowRunPlan, +}; +use crate::error::{invalid_arg, to_napi_err}; + +/// Largest number of children a single fan-out may expand into — guards against +/// a producer returning an enormous list and flooding storage in one batch. +const MAX_FAN_OUT: usize = 10_000; + +const DEFAULT_QUEUE: &str = "default"; +const DEFAULT_MAX_RETRIES: i32 = 3; +const DEFAULT_TIMEOUT_MS: i64 = 300_000; +const DEFAULT_LIST_LIMIT: i64 = 50; + +#[napi] +impl JsQueue { + /// Submit a workflow run from a serialized DAG plus per-step metadata and + /// payloads. Pre-enqueues a job per step with a `depends_on` chain and + /// records a `WorkflowRun` + a `WorkflowNode` per step. Returns the run id. + /// + /// `deferred_node_names` lists steps the tracker enqueues on demand at + /// runtime (fan-out parents, fan-in collectors, and anything downstream of + /// them). A deferred node gets a `Pending` node row but **no job**, and is + /// excluded from its successors' `depends_on` so the static scheduler never + /// blocks on a job that will not exist until expansion. + #[napi] + #[allow(clippy::too_many_arguments)] + pub fn submit_workflow( + &self, + name: String, + version: i32, + dag_bytes: Buffer, + step_metadata_json: String, + node_payloads: HashMap, + queue_default: Option, + params_json: Option, + deferred_node_names: Option>, + ) -> Result { + let wf = self.workflow_store()?; + let dag = dag_bytes.to_vec(); + let step_meta: HashMap = + serde_json::from_str(&step_metadata_json).map_err(|e| reason(e.to_string()))?; + let ordered = topological_order(&dag).map_err(to_napi_err)?; + let queue_default = queue_default.unwrap_or_else(|| DEFAULT_QUEUE.to_string()); + let deferred: HashSet = deferred_node_names + .unwrap_or_default() + .into_iter() + .collect(); + + let definition_id = match wf + .get_workflow_definition(&name, Some(version)) + .map_err(to_napi_err)? + { + // Reuse the stored id only when the DAG matches; otherwise the run + // would execute one topology while `definition_id` points at another, + // breaking run-plan reconstruction. Force a version bump instead. + Some(existing) => { + if existing.dag_data != dag { + return Err(invalid_arg(format!( + "workflow '{name}' v{version} already exists with a different DAG; bump the version" + ))); + } + existing.id + } + None => { + let def = WorkflowDefinition { + id: uuid::Uuid::now_v7().to_string(), + name: name.clone(), + version, + dag_data: dag.clone(), + step_metadata: step_meta.clone(), + created_at: now_millis(), + }; + let id = def.id.clone(); + wf.create_workflow_definition(&def).map_err(to_napi_err)?; + id + } + }; + + let run_id = uuid::Uuid::now_v7().to_string(); + let now = now_millis(); + let run = WorkflowRun { + id: run_id.clone(), + definition_id, + params: params_json, + state: WorkflowState::Pending, + started_at: Some(now), + completed_at: None, + error: None, + parent_run_id: None, + parent_node_name: None, + created_at: now, + }; + wf.create_workflow_run(&run).map_err(to_napi_err)?; + + let mut job_ids: HashMap = HashMap::new(); + for topo in &ordered { + let meta = step_meta.get(&topo.name).ok_or_else(|| { + reason(format!("step '{}' missing from step_metadata", topo.name)) + })?; + + // Deferred node: tracked but not enqueued. The tracker creates its + // job (or expands it) once its predecessors settle. + if deferred.contains(&topo.name) { + wf.create_workflow_node(&new_workflow_node(&run_id, &topo.name, None)) + .map_err(to_napi_err)?; + continue; + } + + let payload = node_payloads + .get(&topo.name) + .map(|b| b.to_vec()) + .ok_or_else(|| { + reason(format!("step '{}' missing from node_payloads", topo.name)) + })?; + // Only static predecessors gate this job; deferred ones have no job + // to wait on, so the tracker is responsible for enqueuing nodes that + // sit downstream of them. + let depends_on = topo + .predecessors + .iter() + .filter(|p| !deferred.contains(*p)) + .map(|p| { + job_ids.get(p).cloned().ok_or_else(|| { + reason(format!( + "predecessor '{}' of step '{}' has no job id", + p, topo.name + )) + }) + }) + .collect::>>()?; + + let new_job = NewJob { + queue: meta.queue.clone().unwrap_or_else(|| queue_default.clone()), + task_name: meta.task_name.clone(), + payload, + priority: meta.priority.unwrap_or(0), + scheduled_at: now, + max_retries: meta.max_retries.unwrap_or(DEFAULT_MAX_RETRIES), + timeout_ms: meta.timeout_ms.unwrap_or(DEFAULT_TIMEOUT_MS), + unique_key: None, + metadata: Some(workflow_metadata_json(&run_id, &topo.name)), + notes: None, + depends_on, + expires_at: None, + result_ttl_ms: None, + namespace: self.namespace.clone(), + }; + let job = self.storage.enqueue(new_job).map_err(to_napi_err)?; + job_ids.insert(topo.name.clone(), job.id.clone()); + + wf.create_workflow_node(&new_workflow_node(&run_id, &topo.name, Some(job.id))) + .map_err(to_napi_err)?; + } + + wf.update_workflow_run_state(&run_id, WorkflowState::Running, None) + .map_err(to_napi_err)?; + Ok(run_id) + } + + /// Record the terminal outcome of a workflow node's job. A no-op (returns + /// `null`) for non-workflow jobs. On failure, cascades a fail-fast skip to + /// the run's remaining pending nodes. When the run reaches a terminal state, + /// the returned advance carries `finalState`. + /// + /// `skip_cascade` is set by the tracker for runs it manages (those with + /// deferred/fan-out nodes): the node's status is recorded but cascade and + /// run finalization are left to the tracker, which drives them after + /// expanding/aggregating. For plain DAGs it is false and behaviour is + /// unchanged. + #[napi] + pub fn mark_workflow_node_result( + &self, + job_id: String, + succeeded: bool, + error: Option, + skip_cascade: Option, + ) -> Result> { + let skip_cascade = skip_cascade.unwrap_or(false); + let wf = self.workflow_store()?; + let job = match self.storage.get_job(&job_id).map_err(to_napi_err)? { + Some(j) => j, + None => return Ok(None), + }; + let (run_id, node_name) = match parse_workflow_metadata(job.metadata.as_deref()) { + Some(pair) => pair, + None => return Ok(None), + }; + + let now = now_millis(); + if succeeded { + wf.set_workflow_node_completed(&run_id, &node_name, now, None) + .map_err(to_napi_err)?; + } else { + let msg = error.clone().unwrap_or_else(|| "failed".to_string()); + wf.set_workflow_node_error(&run_id, &node_name, &msg) + .map_err(to_napi_err)?; + if !skip_cascade { + let nodes = wf.get_workflow_nodes(&run_id).map_err(to_napi_err)?; + cascade_skip_pending(&self.storage, &wf, &run_id, &nodes); + } + } + + // Tracker-managed run: it owns cascade + finalization (it may still need + // to expand a fan-out or enqueue a deferred successor first). + if skip_cascade { + return Ok(Some(JsWorkflowAdvance { + run_id, + node_name, + final_state: None, + })); + } + + let nodes = wf.get_workflow_nodes(&run_id).map_err(to_napi_err)?; + if !nodes.iter().all(|n| n.status.is_terminal()) { + return Ok(Some(JsWorkflowAdvance { + run_id, + node_name, + final_state: None, + })); + } + + let any_failed = nodes.iter().any(|n| n.status == WorkflowNodeStatus::Failed); + let final_state = if any_failed || !succeeded { + WorkflowState::Failed + } else { + WorkflowState::Completed + }; + wf.update_workflow_run_state( + &run_id, + final_state, + if final_state == WorkflowState::Failed { + error.as_deref() + } else { + None + }, + ) + .map_err(to_napi_err)?; + wf.set_workflow_run_completed(&run_id, now) + .map_err(to_napi_err)?; + + Ok(Some(JsWorkflowAdvance { + run_id, + node_name, + final_state: Some(final_state.as_str().to_string()), + })) + } + + /// Fetch a workflow run by id, or `null` if no such run exists. + #[napi] + pub fn get_workflow_run(&self, run_id: String) -> Result> { + let wf = self.workflow_store()?; + Ok(wf + .get_workflow_run(&run_id) + .map_err(to_napi_err)? + .map(run_to_js)) + } + + /// List the nodes (steps) of a workflow run. + #[napi] + pub fn get_workflow_nodes(&self, run_id: String) -> Result> { + let wf = self.workflow_store()?; + Ok(wf + .get_workflow_nodes(&run_id) + .map_err(to_napi_err)? + .into_iter() + .map(node_to_js) + .collect()) + } + + /// List workflow runs, optionally filtered by definition name and/or state. + #[napi] + pub fn list_workflow_runs( + &self, + definition_name: Option, + state: Option, + limit: Option, + offset: Option, + ) -> Result> { + let wf = self.workflow_store()?; + let state = match state { + Some(s) => Some( + WorkflowState::from_str_val(&s) + .ok_or_else(|| reason(format!("unknown workflow state '{s}'")))?, + ), + None => None, + }; + let runs = wf + .list_workflow_runs( + definition_name.as_deref(), + state, + limit.unwrap_or(DEFAULT_LIST_LIMIT), + offset.unwrap_or(0), + ) + .map_err(to_napi_err)?; + Ok(runs.into_iter().map(run_to_js).collect()) + } + + /// The serialized DAG (a `SerializableGraph` JSON string) backing a run, or + /// `null` if the run or its definition no longer exists. + #[napi] + pub fn get_workflow_dag(&self, run_id: String) -> Result> { + let wf = self.workflow_store()?; + let Some(run) = wf.get_workflow_run(&run_id).map_err(to_napi_err)? else { + return Ok(None); + }; + let Some(def) = wf + .get_workflow_definition_by_id(&run.definition_id) + .map_err(to_napi_err)? + else { + return Ok(None); + }; + String::from_utf8(def.dag_data) + .map(Some) + .map_err(|e| reason(e.to_string())) + } + + /// Child (sub-workflow) runs spawned by a run. Always empty for runs + /// submitted by the Node SDK, which does not yet create sub-workflows. + #[napi] + pub fn get_workflow_children(&self, run_id: String) -> Result> { + let wf = self.workflow_store()?; + Ok(wf + .get_child_workflow_runs(&run_id) + .map_err(to_napi_err)? + .into_iter() + .map(run_to_js) + .collect()) + } + + /// The run + node a job belongs to, or `null` for a non-workflow job. Lets + /// the tracker classify any worker outcome without an in-memory job→run map. + #[napi] + pub fn workflow_node_for_job(&self, job_id: String) -> Result> { + let Some(job) = self.storage.get_job(&job_id).map_err(to_napi_err)? else { + return Ok(None); + }; + Ok(parse_workflow_metadata(job.metadata.as_deref()) + .map(|(run_id, node_name)| JsWorkflowNodeRef { run_id, node_name })) + } + + /// The DAG + step metadata backing a run, so the tracker can reconstruct + /// the workflow structure from storage. `null` if the run or its definition + /// no longer exists. + #[napi] + pub fn get_workflow_run_plan(&self, run_id: String) -> Result> { + let wf = self.workflow_store()?; + let Some(run) = wf.get_workflow_run(&run_id).map_err(to_napi_err)? else { + return Ok(None); + }; + let Some(def) = wf + .get_workflow_definition_by_id(&run.definition_id) + .map_err(to_napi_err)? + else { + return Ok(None); + }; + let dag = String::from_utf8(def.dag_data).map_err(|e| reason(e.to_string()))?; + let step_metadata = + serde_json::to_string(&def.step_metadata).map_err(|e| reason(e.to_string()))?; + Ok(Some(JsWorkflowRunPlan { dag, step_metadata })) + } + + /// Expand a fan-out parent into one child node + job per item. Sets the + /// parent's `fan_out_count` and transitions it to `Running`; an empty item + /// list completes the parent immediately. Returns the child job ids. + #[napi] + #[allow(clippy::too_many_arguments)] + pub fn expand_fan_out( + &self, + run_id: String, + parent_node_name: String, + child_names: Vec, + child_payloads: Vec, + task_name: String, + queue: String, + max_retries: i32, + timeout_ms: i64, + priority: i32, + ) -> Result> { + if child_names.len() != child_payloads.len() { + return Err(reason( + "child_names and child_payloads must have equal length", + )); + } + if child_names.len() > MAX_FAN_OUT { + return Err(reason(format!( + "fan-out of {} children exceeds the limit of {MAX_FAN_OUT}", + child_names.len() + ))); + } + let wf = self.workflow_store()?; + let now = now_millis(); + let count = child_names.len() as i32; + + if count == 0 { + wf.set_workflow_node_fan_out_count(&run_id, &parent_node_name, 0) + .map_err(to_napi_err)?; + wf.set_workflow_node_completed(&run_id, &parent_node_name, now, None) + .map_err(to_napi_err)?; + return Ok(Vec::new()); + } + + // Enqueue every child job first, then batch-insert their nodes in one + // transaction so a crash partway through never leaves half-tracked + // children. + let mut child_job_ids = Vec::with_capacity(child_names.len()); + let mut nodes = Vec::with_capacity(child_names.len()); + for (child_name, payload) in child_names.iter().zip(child_payloads) { + let new_job = NewJob { + queue: queue.clone(), + task_name: task_name.clone(), + payload: payload.to_vec(), + priority, + scheduled_at: now, + max_retries, + timeout_ms, + unique_key: None, + metadata: Some(workflow_metadata_json(&run_id, child_name)), + notes: None, + depends_on: vec![], + expires_at: None, + result_ttl_ms: None, + namespace: self.namespace.clone(), + }; + let job = self.storage.enqueue(new_job).map_err(to_napi_err)?; + child_job_ids.push(job.id.clone()); + nodes.push(new_workflow_node(&run_id, child_name, Some(job.id))); + } + + // Child jobs are already enqueued; if node creation fails, cancel them + // so they don't run untracked (orphaned) outside the workflow. + if let Err(err) = wf.create_workflow_nodes_batch(&nodes).map_err(to_napi_err) { + for id in &child_job_ids { + let _ = self.storage.cancel_job(id); + } + return Err(err); + } + wf.set_workflow_node_fan_out_count(&run_id, &parent_node_name, count) + .map_err(to_napi_err)?; + Ok(child_job_ids) + } + + /// Enqueue a job for a deferred node and bind it to that node. Used for the + /// fan-in collector and for static nodes downstream of a deferred one whose + /// predecessors are now all terminal. Returns the new job id. + #[napi] + #[allow(clippy::too_many_arguments)] + pub fn create_deferred_job( + &self, + run_id: String, + node_name: String, + payload: Buffer, + task_name: String, + queue: String, + max_retries: i32, + timeout_ms: i64, + priority: i32, + ) -> Result { + let wf = self.workflow_store()?; + let new_job = NewJob { + queue, + task_name, + payload: payload.to_vec(), + priority, + scheduled_at: now_millis(), + max_retries, + timeout_ms, + unique_key: None, + metadata: Some(workflow_metadata_json(&run_id, &node_name)), + notes: None, + depends_on: vec![], + expires_at: None, + result_ttl_ms: None, + namespace: self.namespace.clone(), + }; + let job = self.storage.enqueue(new_job).map_err(to_napi_err)?; + // Cancel the job if we can't bind it to its node, so it doesn't run + // unbound (the tracker would never see its outcome). + if let Err(err) = wf + .set_workflow_node_job(&run_id, &node_name, &job.id) + .map_err(to_napi_err) + { + let _ = self.storage.cancel_job(&job.id); + return Err(err); + } + Ok(job.id) + } + + /// Check whether all fan-out children of a parent are terminal. When they + /// are, atomically finalizes the parent exactly once (compare-and-swap) and + /// returns the outcome; otherwise — or if another caller already finalized — + /// returns `null`. + #[napi] + pub fn check_fan_out_completion( + &self, + run_id: String, + parent_node_name: String, + ) -> Result> { + let wf = self.workflow_store()?; + let prefix = format!("{parent_node_name}["); + let children = wf + .get_workflow_nodes_by_prefix(&run_id, &prefix) + .map_err(to_napi_err)?; + + if children.is_empty() || !children.iter().all(|n| n.status.is_terminal()) { + return Ok(None); + } + + let any_failed = children + .iter() + .any(|n| n.status == WorkflowNodeStatus::Failed); + let child_job_ids: Vec = children.iter().filter_map(|n| n.job_id.clone()).collect(); + + let transitioned = wf + .finalize_fan_out_parent( + &run_id, + &parent_node_name, + !any_failed, + if any_failed { + Some("fan-out child failed") + } else { + None + }, + now_millis(), + ) + .map_err(to_napi_err)?; + if !transitioned { + return Ok(None); + } + Ok(Some(JsFanOutCompletion { + succeeded: !any_failed, + child_job_ids, + })) + } + + /// Fail-fast cascade: skip every still-pending node of a run and cancel its + /// job. Called by the tracker when a managed run's node fails. + #[napi] + pub fn cascade_skip_pending(&self, run_id: String) -> Result<()> { + let wf = self.workflow_store()?; + let nodes = wf.get_workflow_nodes(&run_id).map_err(to_napi_err)?; + cascade_skip_pending(&self.storage, &wf, &run_id, &nodes); + Ok(()) + } + + /// Finalize a run if every node has reached a terminal state: set it + /// `Completed`, or `Failed` if any node failed. Returns the final state, or + /// `null` if the run still has work in flight. + #[napi] + pub fn finalize_run_if_terminal(&self, run_id: String) -> Result> { + let wf = self.workflow_store()?; + let nodes = wf.get_workflow_nodes(&run_id).map_err(to_napi_err)?; + if nodes.is_empty() || !nodes.iter().all(|n| n.status.is_terminal()) { + return Ok(None); + } + let any_failed = nodes.iter().any(|n| n.status == WorkflowNodeStatus::Failed); + let final_state = if any_failed { + WorkflowState::Failed + } else { + WorkflowState::Completed + }; + wf.update_workflow_run_state( + &run_id, + final_state, + if any_failed { + Some("a workflow node failed") + } else { + None + }, + ) + .map_err(to_napi_err)?; + wf.set_workflow_run_completed(&run_id, now_millis()) + .map_err(to_napi_err)?; + Ok(Some(final_state.as_str().to_string())) + } + + /// Lazily build the workflow storage from this queue's backend (runs the + /// workflow migrations on first use). Not exported to JS. + fn workflow_store(&self) -> Result { + if let Some(wf) = self.workflow_storage.get() { + return Ok(wf.clone()); + } + let wf = build_workflow_storage(&self.storage)?; + // If another thread raced us, either handle wraps the same pool. + let _ = self.workflow_storage.set(wf.clone()); + Ok(wf) + } +} + +/// Construct the workflow storage matching the queue's core backend. +fn build_workflow_storage(storage: &StorageBackend) -> Result { + let wf = match storage { + StorageBackend::Sqlite(s) => WorkflowSqliteStorage::new(s.clone()) + .map(WorkflowStorageBackend::Sqlite) + .map_err(to_napi_err)?, + #[cfg(feature = "postgres")] + StorageBackend::Postgres(s) => taskito_workflows::WorkflowPostgresStorage::new(s.clone()) + .map(WorkflowStorageBackend::Postgres) + .map_err(to_napi_err)?, + #[cfg(feature = "redis")] + StorageBackend::Redis(s) => taskito_workflows::WorkflowRedisStorage::new(s.clone()) + .map(WorkflowStorageBackend::Redis) + .map_err(to_napi_err)?, + }; + Ok(wf) +} + +/// Build a fresh `Pending` workflow node. `job_id` is `None` for deferred nodes +/// (fan-out/fan-in/downstream) that the tracker enqueues later. +fn new_workflow_node(run_id: &str, node_name: &str, job_id: Option) -> WorkflowNode { + WorkflowNode { + id: uuid::Uuid::now_v7().to_string(), + run_id: run_id.to_string(), + node_name: node_name.to_string(), + job_id, + status: WorkflowNodeStatus::Pending, + result_hash: None, + fan_out_count: None, + fan_in_data: None, + started_at: None, + completed_at: None, + error: None, + compensation_job_id: None, + compensation_started_at: None, + compensation_completed_at: None, + compensation_error: None, + } +} + +/// Job-metadata blob that links a job back to its workflow node. +fn workflow_metadata_json(run_id: &str, node_name: &str) -> String { + serde_json::json!({ + "workflow_run_id": run_id, + "workflow_node_name": node_name, + }) + .to_string() +} + +/// Parse `{workflow_run_id, workflow_node_name}` from a job's metadata blob. +fn parse_workflow_metadata(metadata: Option<&str>) -> Option<(String, String)> { + let parsed: serde_json::Value = serde_json::from_str(metadata?).ok()?; + let run_id = parsed.get("workflow_run_id")?.as_str()?.to_string(); + let node_name = parsed.get("workflow_node_name")?.as_str()?.to_string(); + Some((run_id, node_name)) +} + +/// Fail-fast: mark every still-pending node skipped and cancel its job. +/// Best-effort — per-node failures are logged, not propagated. +fn cascade_skip_pending( + storage: &StorageBackend, + wf: &WorkflowStorageBackend, + run_id: &str, + nodes: &[WorkflowNode], +) { + for node in nodes { + if !matches!( + node.status, + WorkflowNodeStatus::Pending | WorkflowNodeStatus::Ready + ) { + continue; + } + if let Some(job_id) = &node.job_id { + if let Err(e) = storage.cancel_job(job_id) { + log::warn!("[taskito-node] cancel_job({job_id}) during workflow cascade: {e}"); + } + } + if let Err(e) = + wf.update_workflow_node_status(run_id, &node.node_name, WorkflowNodeStatus::Skipped) + { + log::warn!("[taskito-node] skip node '{}' failed: {e}", node.node_name); + } + } +} + +/// A JS error carrying a plain message (for validation / serde failures). +fn reason(message: impl Into) -> Error { + Error::new(napi::Status::GenericFailure, message.into()) +} diff --git a/crates/taskito-node/src/worker.rs b/crates/taskito-node/src/worker.rs index bd27572e..8044a7a4 100644 --- a/crates/taskito-node/src/worker.rs +++ b/crates/taskito-node/src/worker.rs @@ -14,6 +14,8 @@ use tokio::sync::Notify; use crate::config::WorkerOptions; use crate::convert::{outcome_to_js, JsOutcome, JsTaskInvocation}; use crate::dispatcher::NodeDispatcher; +#[cfg(feature = "mesh")] +use crate::error::invalid_arg; const DEFAULT_QUEUE: &str = "default"; const DEFAULT_CHANNEL_CAPACITY: usize = 128; @@ -25,18 +27,23 @@ const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5); pub struct JsWorker { shutdown: Arc, heartbeat_stop: Arc, + #[cfg(feature = "mesh")] + mesh_shutdown: Arc, } #[napi] impl JsWorker { /// Stop the worker: the scheduler stops dispatching, the heartbeat loop ends - /// and unregisters, and the background tasks exit once in-flight results drain. + /// and unregisters, mesh gossip/steal tasks shut down, and the background + /// tasks exit once in-flight results drain. #[napi] pub fn stop(&self) { // `notify_one` stores a permit if no waiter is parked yet, so the signal // is never lost between loop iterations. self.shutdown.notify_one(); self.heartbeat_stop.notify_one(); + #[cfg(feature = "mesh")] + self.mesh_shutdown.notify_one(); } } @@ -52,6 +59,19 @@ pub fn start_worker( let queues = options .queues .unwrap_or_else(|| vec![DEFAULT_QUEUE.to_string()]); + // Validate the mesh port up front: the steal server binds `port + 1`, and + // both narrow to u16, so anything outside 1..=65534 wraps to a wrong port. + #[cfg(feature = "mesh")] + if let Some(mesh_cfg) = options.mesh.as_ref() { + if mesh_cfg.port < 1 || mesh_cfg.port > 65534 { + return Err(invalid_arg(format!( + "mesh port must be in 1..=65534 (got {})", + mesh_cfg.port + ))); + } + } + #[cfg(feature = "mesh")] + let mesh_shutdown = Arc::new(Notify::new()); // Clamp to >= 1: a zero-capacity Tokio/crossbeam channel panics on creation. let capacity = options .channel_capacity @@ -68,6 +88,11 @@ pub fn start_worker( let dispatcher_storage = storage.clone(); let lifecycle_storage = storage.clone(); let queues_csv = queues.join(","); + let worker_id = format!("node-{}", uuid::Uuid::now_v7()); + // Mesh gossip advertises the served queues; capture them before `queues` + // moves into the scheduler. + #[cfg(feature = "mesh")] + let mesh_queues = queues.clone(); // Per-task/queue config must be registered before the scheduler is shared // (register_* take &mut self). @@ -88,16 +113,56 @@ pub fn start_worker( let heartbeat_stop = Arc::new(Notify::new()); spawn_worker_lifecycle( lifecycle_storage, + worker_id.clone(), queues_csv, capacity, heartbeat_stop.clone(), ); - // Scheduler loop: poll storage, dispatch ready jobs onto `job_tx`. - let scheduler_run = scheduler.clone(); - spawn(async move { - scheduler_run.run(job_tx).await; - }); + // Scheduler loop: poll storage, dispatch ready jobs onto `job_tx`. With the + // mesh feature built and a mesh config supplied, route the scheduler's output + // through the mesh bridge (affinity-sorted local deque + work-stealing); the + // DB stays the source of truth, so the plain path is otherwise identical. + #[cfg(feature = "mesh")] + { + if let Some(mesh_cfg) = options.mesh.as_ref() { + let mesh_node = Arc::new(taskito_mesh::MeshNode::new( + worker_id.clone(), + build_mesh_config(mesh_cfg), + )); + let gossip = mesh_node.spawn_gossip(mesh_queues, capacity as u16); + let steal_server = mesh_node.spawn_steal_server(); + let bridge_scheduler = scheduler.clone(); + let bridge_node = mesh_node.clone(); + spawn(async move { + run_mesh_bridge(bridge_scheduler, bridge_node, job_tx).await; + }); + // On stop, signal the mesh node so gossip + steal-server tasks exit + // instead of leaking for the process lifetime. + let mesh_stop = mesh_shutdown.clone(); + let stop_node = mesh_node.clone(); + spawn(async move { + mesh_stop.notified().await; + stop_node.request_shutdown(); + }); + // Keep the gossip + steal-server tasks alive for the worker's lifetime. + spawn(async move { + let _ = tokio::join!(gossip, steal_server); + }); + } else { + let scheduler_run = scheduler.clone(); + spawn(async move { + scheduler_run.run(job_tx).await; + }); + } + } + #[cfg(not(feature = "mesh"))] + { + let scheduler_run = scheduler.clone(); + spawn(async move { + scheduler_run.run(job_tx).await; + }); + } // Dispatcher loop: execute each job in JS, report results on `result_tx`. let dispatcher = NodeDispatcher::new(callback, dispatcher_storage); @@ -128,17 +193,19 @@ pub fn start_worker( Ok(JsWorker { shutdown, heartbeat_stop, + #[cfg(feature = "mesh")] + mesh_shutdown, }) } /// Register this worker and heartbeat until `stop` is signalled, then unregister. fn spawn_worker_lifecycle( storage: StorageBackend, + worker_id: String, queues_csv: String, capacity: usize, stop: Arc, ) { - let worker_id = format!("node-{}", uuid::Uuid::now_v7()); let hostname = gethostname::gethostname().to_string_lossy().to_string(); let pid = std::process::id() as i32; // Log lifecycle failures: a worker that can't register/heartbeat goes @@ -173,3 +240,105 @@ fn spawn_worker_lifecycle( } }); } + +/// Mesh-aware scheduler bridge: the scheduler emits jobs into an intermediate +/// channel, which this loop pushes into the mesh local deque (affinity-sorted), +/// then drains the deque to the dispatcher channel and steals from peers when idle. +#[cfg(feature = "mesh")] +async fn run_mesh_bridge( + scheduler: Arc, + mesh_node: Arc, + job_tx: tokio::sync::mpsc::Sender, +) { + let (mesh_tx, mut mesh_rx) = tokio::sync::mpsc::channel::(64); + + let sched = scheduler.clone(); + let sched_task = spawn(async move { + sched.run(mesh_tx).await; + }); + + loop { + // Drain the local deque to the dispatcher first. + while let Some(job) = mesh_node.pop_local() { + if job_tx.send(job).await.is_err() { + let _ = sched_task.await; + return; + } + } + + // Steal from busier peers when the deque is low and stealing is enabled. + if mesh_node.should_steal() { + mesh_node.try_steal().await; + while let Some(job) = mesh_node.pop_local() { + if job_tx.send(job).await.is_err() { + let _ = sched_task.await; + return; + } + } + } + + // Wait for the scheduler to produce more jobs, batching what's ready. + match mesh_rx.recv().await { + Some(job) => { + let mut batch = vec![job]; + while let Ok(extra) = mesh_rx.try_recv() { + batch.push(extra); + } + mesh_node.prefetch(batch); + } + None => break, + } + } + + // Scheduler stopped — drain whatever remains in the deque. + while let Some(job) = mesh_node.pop_local() { + if job_tx.send(job).await.is_err() { + break; + } + } + let _ = sched_task.await; +} + +/// Translate the JS-facing [`crate::config::MeshWorkerConfig`] into the core +/// mesh config, leaving unspecified fields at their crate defaults. +#[cfg(feature = "mesh")] +#[allow(clippy::field_reassign_with_default)] +fn build_mesh_config(input: &crate::config::MeshWorkerConfig) -> taskito_mesh::MeshConfig { + let mut config = taskito_mesh::MeshConfig::default(); + config.gossip_port = input.port as u16; + config.steal_port = input.port.saturating_add(1) as u16; + if let Some(ref addr) = input.bind_addr { + config.bind_addr = addr.clone(); + } + if let Some(ref seeds) = input.seeds { + config.seeds = seeds.clone(); + } + if let Some(enabled) = input.steal { + config.enable_stealing = enabled; + } + if let Some(weight) = input.affinity_weight { + config.affinity_weight = weight; + } + if let Some(buffer) = input.local_buffer { + config.local_buffer_capacity = buffer as usize; + } + if let Some(batch) = input.steal_batch { + config.max_steal_batch = batch as usize; + } + if let Some(threshold) = input.steal_threshold { + config.steal_threshold = threshold as usize; + } + if let Some(vnodes) = input.virtual_nodes { + config.virtual_nodes = vnodes as usize; + } + if let Some(ref addr) = input.advertise_addr { + config.advertise_addr = Some(addr.clone()); + } + if let Some(ref key) = input.encryption_key { + config.encryption_key = Some(key.clone()); + } + if let Some(rate) = input.steal_rate_limit { + config.steal_rate_limit = rate; + } + config +} diff --git a/docs/content/docs/meta.json b/docs/content/docs/meta.json index 0caaa854..69c58e0d 100644 --- a/docs/content/docs/meta.json +++ b/docs/content/docs/meta.json @@ -3,6 +3,7 @@ "pages": [ "capabilities", "getting-started", + "node", "guides", "architecture", "api-reference", diff --git a/docs/content/docs/node/index.mdx b/docs/content/docs/node/index.mdx new file mode 100644 index 00000000..24bb8227 --- /dev/null +++ b/docs/content/docs/node/index.mdx @@ -0,0 +1,76 @@ +--- +title: Node.js SDK +description: "Rust-powered task queue for Node.js — no broker, no Python." +--- + +import { Cards, Card } from "fumadocs-ui/components/card"; +import { Callout } from "fumadocs-ui/components/callout"; + +A thin [napi-rs](https://napi.rs) shell over the Taskito Rust core, peer to +the Python SDK. Enqueue work and run workers in the same process or across +processes that share storage. Zero Python dependency. + +## Install + +```bash +pnpm add taskito +``` + +Requires Node.js >= 18. Ships as dual ESM + CommonJS. The native addon is +currently built from source (requires Rust + the napi-rs CLI) — prebuilt +per-platform binaries are not yet published: + +```bash +pnpm build:native +``` + +## Quickstart + +```ts +import { Queue } from "taskito"; + +const queue = new Queue({ dbPath: "taskito.db" }); + +// 1. Register a task. +queue.task("add", (a: number, b: number) => a + b, { + maxRetries: 3, + timeoutMs: 30_000, +}); + +// 2. Enqueue a job. +const id = queue.enqueue("add", [2, 3]); + +// 3. Start a worker (same process or a separate one). +const worker = queue.runWorker(); + +// 4. Await the result. +const result = await queue.result(id); // 5 +worker.stop(); +``` + +## Backends + +```ts +new Queue({ dbPath: "taskito.db" }); // SQLite (default) +new Queue({ backend: "postgres", dsn: process.env.PG_URL }); // PostgreSQL +new Queue({ backend: "redis", dsn: "redis://localhost" }); // Redis +``` + +## What's available + + + + + +For the full API surface — serializers, cancellation, events, middleware, +webhooks, distributed locks, periodic tasks, the CLI, and the dashboard — see +the [Node SDK README](https://github.com/pratyushsharma/taskito/blob/master/sdks/node/README.md). + + + Gates, sub-workflows, and saga compensation are not yet available in the Node + SDK. Use the [Python SDK](/guides/workflows/gates) in the meantime. + diff --git a/docs/content/docs/node/meta.json b/docs/content/docs/node/meta.json new file mode 100644 index 00000000..ecc9dcb8 --- /dev/null +++ b/docs/content/docs/node/meta.json @@ -0,0 +1,5 @@ +{ + "title": "Node SDK", + "root": true, + "pages": ["index", "workflows"] +} diff --git a/docs/content/docs/node/workflows/fan-out.mdx b/docs/content/docs/node/workflows/fan-out.mdx new file mode 100644 index 00000000..40e44e80 --- /dev/null +++ b/docs/content/docs/node/workflows/fan-out.mdx @@ -0,0 +1,171 @@ +--- +title: Fan-Out & Fan-In +description: "Split a step's result into parallel children, collect results into a downstream step." +--- + +import { Callout } from "fumadocs-ui/components/callout"; + +Split a step's result into parallel child jobs, then collect all results into +a downstream step. + +```mermaid +graph LR + fetch:::source --> process_0["process[0]"] + fetch --> process_1["process[1]"] + fetch --> process_2["process[2]"] + process_0 --> aggregate:::sink + process_1 --> aggregate + process_2 --> aggregate + + classDef source fill:#bbf7d0,color:#14532d,stroke:#16a34a + classDef sink fill:#bfdbfe,color:#1e3a8a,stroke:#3b82f6 +``` + +## Basic fan-out and fan-in + +Use `.fanOut()` to expand a predecessor's array result into one child job per +item, and `.fanIn()` to collect the children's results into a combiner task. + +```ts +queue.task("listFiles", () => ["a.csv", "b.csv", "c.csv"]); // returns the items array +queue.task("processFile", (file: string) => file.length); // runs once per item +queue.task("summarize", (sizes: number[]) => sizes.reduce((a, b) => a + b, 0)); + +const handle = queue.workflows + .define("batch") + .step("list", "listFiles") + .fanOut("process", { after: "list", task: "processFile", itemsFrom: "list" }) + .fanIn("collect", { after: "process", task: "summarize" }) + .submit(); + +queue.runWorker(); // advances workflow nodes by default + +const run = await handle.wait(); +console.log(run.state); // "completed" +``` + +`itemsFrom` names the predecessor whose result is the item array. It defaults +to the sole predecessor when omitted. Each item is passed to the child task as +its single positional argument. + +Child nodes are named `process[0]`, `process[1]`, `process[2]` and appear in +status queries. The parent node carries `fanOutCount` (the number of children +spawned). + +```ts +const nodes = handle.nodes(); +// nodes.find(n => n.nodeName === "process")?.fanOutCount === 3 +// nodes.find(n => n.nodeName === "process[0]")?.status === "completed" +``` + +## How it works + +1. `list` completes — the tracker reads its return value (must be an array) +2. The tracker calls `expandFanOut`, creating N child nodes and N jobs — each + receives one item from the array; children are ready immediately (no + `depends_on`) +3. Children execute in parallel; their statuses surface as `process[0]`, + `process[1]`, … +4. As each child settles, the tracker calls `checkFanOutCompletion` +5. When all children complete, the tracker collects results in index order and + calls `createDeferredJob` for the fan-in step +6. `summarize` runs with the results array as its single argument + +```mermaid +sequenceDiagram + participant L as list job + participant T as WorkflowTracker + participant R as Rust Engine + + L->>T: outcome(list, success) + T->>R: getWorkflowRunPlan(runId) + T->>R: expandFanOut(3 children) + R-->>T: [childJobIds] + Note over R: Children execute in parallel + R->>T: outcome(process[0], success) + R->>T: outcome(process[1], success) + R->>T: outcome(process[2], success) + T->>R: checkFanOutCompletion → all done + T->>R: createDeferredJob(collect) +``` + +The `WorkflowTracker` is driven entirely by the worker outcome stream and +reconstructs the run plan from storage on each event, so submission and +execution may run in different processes — no submit-time coordinator is needed. + +## Empty fan-out + +If the predecessor returns an empty array, the fan-out parent is marked +completed immediately with `fanOutCount` 0 and the fan-in runs with `[]`: + +```ts +queue.task("listFiles", () => []); // nothing to process +queue.task("summarize", (sizes: number[]) => sizes.reduce((a, b) => a + b, 0)); + +const handle = queue.workflows + .define("batch") + .step("list", "listFiles") + .fanOut("process", { after: "list", task: "processFile" }) + .fanIn("collect", { after: "process", task: "summarize" }) + .submit(); + +// summarize receives [] → result is 0 +``` + +## Downstream steps + +Steps after the fan-in work normally — declare them with `after` pointing to +the fan-in node: + +```ts +queue.task("notify", (total: number) => sendSlack(`Processed ${total} bytes`)); + +const handle = queue.workflows + .define("full-pipeline") + .step("list", "listFiles") + .fanOut("process", { after: "list", task: "processFile", itemsFrom: "list" }) + .fanIn("collect", { after: "process", task: "summarize" }) + .step("notify", "notify", { after: "collect" }) // runs after summarize + .submit(); +``` + +## Per-step options + +`.fanOut()` accepts the same per-step options as `.step()`: + +| Option | Description | +|--------|-------------| +| `after` | Predecessor node name (required) | +| `task` | Registered task name for each child (required) | +| `itemsFrom` | Node whose result is the items array (defaults to `after`) | +| `queue` | Queue name to run children on | +| `maxRetries` | Retry limit for each child | +| `timeoutMs` | Timeout for each child | +| `priority` | Priority for each child | + +Children and the combiner each inherit the fan-out step's `queue`, `maxRetries`, +`timeoutMs`, and `priority` unless overridden. + +## Failure handling + +Fan-out is **fail-fast**: if any child dead-letters: + +- Remaining pending children are cancelled +- The fan-out parent is marked `failed` +- The fan-in and downstream steps are `skipped` +- The workflow run transitions to `failed` + +```ts +const run = await handle.wait(); +if (run.state === "failed") { + const failed = handle.nodes().filter(n => n.status === "failed"); + console.error("failed nodes:", failed.map(n => n.nodeName)); +} +``` + + + Gates, sub-workflows, and saga compensation are not yet available in the Node + SDK. For those, use the Python SDK — see [gates](/guides/workflows/gates), + [composition](/guides/workflows/composition), and + [sagas](/guides/workflows/sagas). + diff --git a/docs/content/docs/node/workflows/index.mdx b/docs/content/docs/node/workflows/index.mdx new file mode 100644 index 00000000..857e4138 --- /dev/null +++ b/docs/content/docs/node/workflows/index.mdx @@ -0,0 +1,44 @@ +--- +title: Workflows +description: "Orchestrate multi-step DAGs in Node.js with the Taskito workflow builder." +--- + +import { Cards, Card } from "fumadocs-ui/components/card"; +import { Callout } from "fumadocs-ui/components/callout"; + +Orchestrate multi-step DAGs where each step is a registered task and `after` +declares its dependencies. The Rust core schedules steps in topological order; +a worker advances the run as each step settles. + +```ts +const handle = queue.workflows + .define("etl") + .step("extract", "extractTask", { args: ["s3://bucket/in"] }) + .step("transform", "transformTask", { after: "extract" }) + .step("load", "loadTask", { after: "transform", maxRetries: 5 }) + .submit(); + +queue.runWorker(); + +const run = await handle.wait(); // resolves when terminal +console.log(run.state); // "completed" | "failed" | ... +handle.nodes(); // per-step status snapshot +``` + +Requires the addon built with the `workflows` cargo feature (enabled by +`pnpm build:native`). + + + + + + + Gates, sub-workflows, and saga compensation are not yet available in the Node + SDK. See the Python SDK for [gates](/guides/workflows/gates), + [composition](/guides/workflows/composition), and + [sagas](/guides/workflows/sagas). + diff --git a/docs/content/docs/node/workflows/meta.json b/docs/content/docs/node/workflows/meta.json new file mode 100644 index 00000000..bf59b7cc --- /dev/null +++ b/docs/content/docs/node/workflows/meta.json @@ -0,0 +1,4 @@ +{ + "title": "Workflows", + "pages": ["index", "fan-out"] +} diff --git a/sdks/node/README.md b/sdks/node/README.md index e99532d6..74cff6c4 100644 --- a/sdks/node/README.md +++ b/sdks/node/README.md @@ -26,6 +26,7 @@ queue.task("add", (a: number, b: number) => a + b, { retryBackoff: { baseMs: 1000, maxMs: 60_000 }, timeoutMs: 30_000, maxConcurrent: 4, + circuitBreaker: { threshold: 5, windowMs: 60_000, cooldownMs: 30_000 }, }); // Producer. @@ -145,6 +146,52 @@ Events: `job.completed`, `job.retrying`, `job.dead`, `job.cancelled`. `before`/ `after`/`onError` wrap execution (awaited); the outcome hooks fire after the core decides the result. +## Distributed locks + +TTL-bounded, owner-scoped locks backed by the queue's storage — coordinate +across processes without a separate lock server. A held lock auto-extends at +`ttlMs / 3` so a slow section never loses it. + +```ts +// Scoped helper — acquires, runs, releases (throws if held elsewhere). +await queue.withLock("report:2026-06", async () => { + await rebuildReport(); +}); + +// Manual handle with explicit release. +const lock = queue.lock("resource", { ttlMs: 30_000 }); +if (lock.acquire()) { + try { + // ... critical section + } finally { + lock.release(); + } +} +``` + +`Lock` also implements `Symbol.dispose`, so on Node 20.4+ you can use `using lock = +queue.lock("resource")` for automatic release at block exit. + +`lock.extend(ms)`, `lock.info()`, and `lock.ownerId` round out the API. Expired +locks are reaped by the worker's maintenance loop. + +## Periodic (cron) tasks + +Schedule a registered task on a cron expression. A running worker enqueues it +when due (the scheduler's maintenance loop drives this). + +```ts +queue.task("digest", (date: string) => sendDigest(date)); + +// cron is 6/7-field, seconds first: sec min hour day-of-month month day-of-week +queue.registerPeriodic("daily-digest", "digest", "0 0 9 * * *", { + args: ["2026-06-16"], + timezone: "America/New_York", +}); +``` + +Returns the next fire time (Unix ms). Re-registering the same name replaces it. + ## Webhooks Deliver job events to HTTP endpoints — HMAC-SHA256 signed, retried with backoff, @@ -165,6 +212,70 @@ queue.webhooks.delete(hook.id); Deliveries fire from the worker process (where events originate). The dashboard exposes `/api/webhooks` for managing them. +## Workflows + +Orchestrate multi-step DAGs. Each step is a registered task; `after` declares +dependencies. Steps are pre-enqueued with `depends_on` chains, so the core +scheduler runs them in topological order — and a worker advances the run as each +step settles. + +```ts +const handle = queue.workflows + .define("etl") + .step("extract", "extractTask", { args: ["s3://bucket/in"] }) + .step("transform", "transformTask", { after: "extract" }) + .step("load", "loadTask", { after: "transform", maxRetries: 5 }) + .submit(); + +queue.runWorker(); // advances workflow nodes by default + +const run = await handle.wait(); // resolves when terminal +console.log(run.state); // "completed" | "failed" | ... +handle.nodes(); // per-step status +queue.workflows.list({ state: "running" }); +``` + +If a step dead-letters, the run fails and remaining steps are skipped +(fail-fast). Per-step options: `after`, `args`, `queue`, `maxRetries`, +`timeoutMs`, `priority`. Workers that never process workflow steps can skip the +per-job bookkeeping with `runWorker({ advanceWorkflows: false })`. Requires the +addon built with the `workflows` cargo feature (enabled by `build:native`). + +### Fan-out / fan-in + +A `fanOut` step expands at runtime into one child job per item, each running the +same task. The items come from the array result of a predecessor (`itemsFrom`, +defaulting to the sole predecessor) — each item is passed to the child task as +its single argument. An optional `fanIn` step collects the children's results +into an array and runs a combiner task over it. + +```ts +queue.task("listFiles", () => ["a.csv", "b.csv", "c.csv"]); // returns the items +queue.task("processFile", (file: string) => file.length); // runs once per item +queue.task("summarize", (sizes: number[]) => sizes.reduce((a, b) => a + b, 0)); + +const handle = queue.workflows + .define("batch") + .step("list", "listFiles") + .fanOut("process", { after: "list", task: "processFile", itemsFrom: "list" }) + .fanIn("collect", { after: "process", task: "summarize" }) + .submit(); + +queue.runWorker(); +const run = await handle.wait(); +// handle.nodes(): "process" carries fanOutCount; children are "process[0]", "process[1]", … +``` + +The worker-side tracker drives expansion from the outcome stream and +reconstructs the plan from storage, so submission and execution may run in +different processes. Fan-out is fail-fast: if any child dead-letters, the parent +and run fail and downstream steps are skipped. An empty item list completes the +fan-out immediately and runs the fan-in with `[]`. Children and the combiner each +run on the fan-out step's `queue` / `maxRetries` / `timeoutMs` / `priority`. + +Gates, sub-workflows, and saga compensation are not yet bound — for those, use +the Python SDK. + ## Dashboard A web dashboard (the same React UI the Python SDK serves) runs over the queue — @@ -186,9 +297,32 @@ const server = serveDashboard(queue, { port: 8787 }); ``` It serves the SPA plus the `/api/*` REST contract (stats, jobs, dead-letters, -queues, metrics, workers, webhooks, cancel/retry/pause/resume) over the queue. -Auth runs open (localhost); the metrics and workers panels populate from live -job history and running workers. +queues, metrics, workers, webhooks, workflow runs, cancel/retry/pause/resume) +over the queue. Auth runs open (localhost); the metrics and workers panels +populate from live job history and running workers. + +## Mesh (work-stealing overlay) + +Workers can form a decentralized mesh — SWIM gossip for peer discovery plus +consistent-hash placement and TCP work-stealing — so idle nodes pull work from +busy ones. The database stays the source of truth; the mesh only optimizes +dispatch locality. Requires the addon built with the `mesh` cargo feature +(`build:native` enables it). + +```ts +queue.runWorker({ + queues: ["default"], + mesh: { + port: 7946, // UDP gossip; TCP steal binds port + 1 + seeds: ["10.0.0.2:7946"], // peers to join (empty = standalone) + steal: true, + encryptionKey: process.env.MESH_KEY, // optional XOR-encrypt gossip + }, +}); +``` + +Other tunables: `bindAddr`, `advertiseAddr` (NAT), `affinityWeight`, +`localBuffer`, `stealBatch`, `stealThreshold`, `virtualNodes`, `stealRateLimit`. ## Development @@ -206,6 +340,7 @@ compiled in via `--features postgres,redis`. ## Not yet covered -Workflows, mesh, middleware/events, distributed locks, periodic/cron tasks, -prebuilt platform binaries + npm publish (host-only build for now), and -Python⇄Node cross-language interop. +Advanced workflow features (gates, sub-workflows, saga compensation — fan-out is +covered above), resources/proxies/interception, contrib integrations, prebuilt +platform binaries + npm publish (host-only build for now), and Python⇄Node +cross-language interop. diff --git a/sdks/node/package.json b/sdks/node/package.json index cdf5fa5b..19837be4 100644 --- a/sdks/node/package.json +++ b/sdks/node/package.json @@ -48,7 +48,7 @@ }, "scripts": { "build": "pnpm run build:native && pnpm run build:ts", - "build:native": "napi build --platform --release --cargo-cwd ../../crates/taskito-node --features postgres,redis native", + "build:native": "napi build --platform --release --cargo-cwd ../../crates/taskito-node --features postgres,redis,mesh,workflows native", "build:ts": "tsup", "build:dashboard": "pnpm -C ../../dashboard exec vite build --outDir ../sdks/node/static/dashboard --emptyOutDir", "prepack": "pnpm run build:dashboard", diff --git a/sdks/node/src/dashboard/contract.ts b/sdks/node/src/dashboard/contract.ts index b2942872..c5521c4d 100644 --- a/sdks/node/src/dashboard/contract.ts +++ b/sdks/node/src/dashboard/contract.ts @@ -3,6 +3,7 @@ import type { DeadJob, Job, WorkerInfo } from "../types"; import type { Webhook } from "../webhooks"; +import type { WorkflowNode, WorkflowRun } from "../workflows"; /** Replace header values with a mask so outbound credentials aren't exposed. */ function maskHeaderValues(headers: Record): Record { @@ -67,6 +68,40 @@ export function workerToContract(worker: WorkerInfo) { }; } +export function workflowRunToContract(run: WorkflowRun) { + return { + id: run.id, + definition_id: run.definitionId, + state: run.state, + params: run.params ?? null, + started_at: run.startedAt ?? null, + completed_at: run.completedAt ?? null, + error: run.error ?? null, + parent_run_id: run.parentRunId ?? null, + parent_node_name: run.parentNodeName ?? null, + created_at: run.createdAt, + }; +} + +export function workflowNodeToContract(node: WorkflowNode) { + // saga fields stay null until saga compensation is bound; fan_out_count is + // real once a node has expanded. + return { + node_name: node.nodeName, + status: node.status, + job_id: node.jobId ?? null, + result_hash: node.resultHash ?? null, + fan_out_count: node.fanOutCount ?? null, + started_at: node.startedAt ?? null, + completed_at: node.completedAt ?? null, + error: node.error ?? null, + compensation_job_id: node.compensationJobId ?? null, + compensation_started_at: node.compensationStartedAt ?? null, + compensation_completed_at: node.compensationCompletedAt ?? null, + compensation_error: node.compensationError ?? null, + }; +} + export function deadToContract(dead: DeadJob) { return { id: dead.id, diff --git a/sdks/node/src/dashboard/handlers.ts b/sdks/node/src/dashboard/handlers.ts index caa745e5..c5f9acdf 100644 --- a/sdks/node/src/dashboard/handlers.ts +++ b/sdks/node/src/dashboard/handlers.ts @@ -5,7 +5,15 @@ import { randomBytes } from "node:crypto"; import type { EventName } from "../events"; import type { Queue } from "../index"; import type { WebhookInput } from "../webhooks"; -import { deadToContract, jobToContract, webhookToContract, workerToContract } from "./contract"; +import type { WorkflowNode } from "../workflows"; +import { + deadToContract, + jobToContract, + webhookToContract, + workerToContract, + workflowNodeToContract, + workflowRunToContract, +} from "./contract"; import { aggregateByTask, bucketTimeseries } from "./metrics"; /** Finite, non-negative number from a query string, or `undefined`. */ @@ -88,6 +96,69 @@ export function eventTypes(queue: Queue) { return queue.webhooks.eventTypes(); } +export function workflowRuns(queue: Queue, url: URL) { + const limit = num(url, "limit") ?? 50; + const offset = num(url, "offset") ?? 0; + const runs = queue.workflows.list({ + state: url.searchParams.get("state") ?? undefined, + definitionName: url.searchParams.get("definition_name") ?? undefined, + limit, + offset, + }); + return { runs: runs.map(workflowRunToContract), limit, offset }; +} + +export function workflowRun(queue: Queue, id: string) { + const run = queue.workflows.run(id); + if (!run) { + return undefined; + } + return { + run: workflowRunToContract(run), + nodes: queue.workflows.nodes(id).map(workflowNodeToContract), + }; +} + +export function workflowDag(queue: Queue, id: string) { + const dag = queue.workflows.dag(id); + return dag === undefined ? undefined : { dag: enrichDag(dag, queue.workflows.nodes(id)) }; +} + +export function workflowChildren(queue: Queue, id: string) { + // Always empty until sub-workflows are bound (the Node SDK creates no child runs). + return { children: queue.workflows.children(id).map(workflowRunToContract) }; +} + +/** + * Rewrite the raw `SerializableGraph` into the shape the SPA's DAG visualizer + * consumes: it builds edges from each node's `deps[]` (ignoring top-level + * `edges`), colours by per-node `status`, and links via `id`. We fold the live + * node statuses + job ids in so the DAG tab renders real edges, live colours, + * and correct `/jobs/$id` links. Returns a JSON **string** (the SPA `JSON.parse`s it). + */ +function enrichDag(dagJson: string, nodes: readonly WorkflowNode[]): string { + let graph: { nodes?: Array<{ name?: string }>; edges?: Array<{ from: string; to: string }> }; + try { + graph = JSON.parse(dagJson); + } catch { + return dagJson; // not our JSON — pass through untouched + } + const byName = new Map(nodes.map((n) => [n.nodeName, n])); + const edges = graph.edges ?? []; + const enriched = (graph.nodes ?? []).map((raw) => { + const name = raw.name ?? ""; + const node = byName.get(name); + return { + name, + node_name: name, + status: node?.status ?? "pending", + id: node?.jobId ?? name, + deps: edges.filter((edge) => edge.to === name).map((edge) => edge.from), + }; + }); + return JSON.stringify({ nodes: enriched, edges }); +} + export function webhooks(queue: Queue) { return queue.webhooks.list().map(webhookToContract); } diff --git a/sdks/node/src/dashboard/routes.ts b/sdks/node/src/dashboard/routes.ts index 983f3582..48262758 100644 --- a/sdks/node/src/dashboard/routes.ts +++ b/sdks/node/src/dashboard/routes.ts @@ -47,6 +47,26 @@ export const routes: Route[] = [ handle: (q, _url, p) => h.resume(q, id(p)), }, { method: "GET", pattern: /^\/api\/event-types$/, handle: (q) => h.eventTypes(q) }, + { + method: "GET", + pattern: /^\/api\/workflows\/runs$/, + handle: (q, url) => h.workflowRuns(q, url), + }, + { + method: "GET", + pattern: /^\/api\/workflows\/runs\/([^/]+)\/dag$/, + handle: (q, _url, p) => h.workflowDag(q, id(p)), + }, + { + method: "GET", + pattern: /^\/api\/workflows\/runs\/([^/]+)\/children$/, + handle: (q, _url, p) => h.workflowChildren(q, id(p)), + }, + { + method: "GET", + pattern: /^\/api\/workflows\/runs\/([^/]+)$/, + handle: (q, _url, p) => h.workflowRun(q, id(p)), + }, { method: "GET", pattern: /^\/api\/webhooks$/, handle: (q) => h.webhooks(q) }, { method: "POST", diff --git a/sdks/node/src/dashboard/server.ts b/sdks/node/src/dashboard/server.ts index 75379994..7d0a5b8f 100644 --- a/sdks/node/src/dashboard/server.ts +++ b/sdks/node/src/dashboard/server.ts @@ -2,10 +2,13 @@ import { createServer, type IncomingMessage, type Server, type ServerResponse } from "node:http"; import type { Queue } from "../index"; +import { createLogger } from "../utils"; import { WebhookValidationError } from "../webhooks"; import { routes } from "./routes"; import { StaticAssets } from "./static"; +const log = createLogger("dashboard"); + /** Max request body size (1 MiB) — reject larger payloads to bound memory. */ const MAX_BODY_BYTES = 1024 * 1024; @@ -14,7 +17,7 @@ export function createDashboardServer(queue: Queue, staticDir: string): Server { const assets = new StaticAssets(staticDir); return createServer((req, res) => { void dispatch(queue, assets, req, res).catch((error) => { - console.error("[taskito] dashboard dispatch failed:", error); + log.error(() => "dashboard dispatch failed", error); if (!res.headersSent) { sendJson(res, 500, { error: "internal server error" }); } @@ -65,7 +68,7 @@ async function dispatch( return; } // Log internally but never leak error/stack details to the HTTP client. - console.error("[taskito] dashboard request failed:", error); + log.error(() => `${req.method} ${path} failed`, error); sendJson(res, 500, { error: "internal server error" }); } return; diff --git a/sdks/node/src/errors.ts b/sdks/node/src/errors.ts index 7763053b..b25140ee 100644 --- a/sdks/node/src/errors.ts +++ b/sdks/node/src/errors.ts @@ -32,3 +32,19 @@ export class JobCancelledError extends TaskitoError { this.name = "JobCancelledError"; } } + +/** Thrown by {@link Queue.withLock} when the lock is held by another owner. */ +export class LockNotAcquiredError extends TaskitoError { + constructor(readonly lockName: string) { + super(`Could not acquire lock "${lockName}"`); + this.name = "LockNotAcquiredError"; + } +} + +/** Thrown when a held lock's lease was lost before the guarded section finished. */ +export class LockLostError extends TaskitoError { + constructor(readonly lockName: string) { + super(`Lost lock "${lockName}" before the guarded section completed`); + this.name = "LockLostError"; + } +} diff --git a/sdks/node/src/index.ts b/sdks/node/src/index.ts index 771c46f4..9733a129 100644 --- a/sdks/node/src/index.ts +++ b/sdks/node/src/index.ts @@ -3,21 +3,27 @@ export { type DashboardOptions, serveDashboard } from "./dashboard"; export { JobCancelledError, JobFailedError, + LockLostError, + LockNotAcquiredError, TaskitoError, TaskNotRegisteredError, } from "./errors"; export type { EventHandler, EventName, OutcomeEvent } from "./events"; +export { Lock, type LockInfo, type LockOptions } from "./locks"; export type { Middleware, TaskContext } from "./middleware"; export { Queue, type QueueOptions } from "./queue"; export { JsonSerializer, MsgpackSerializer, type Serializer } from "./serializers"; export type { AnyHandler, + CircuitBreakerOptions, DeadJob, EnqueueOptions, Job, JobError, JobFilter, + MeshWorkerConfig, Metric, + PeriodicOptions, QueueLimits, RateLimit, ResultOptions, @@ -28,6 +34,28 @@ export type { WorkerInfo, WorkerRunOptions, } from "./types"; +export { + createLogger, + type Logger, + type LogLevel, + type LogMessage, + type LogSink, + logger, + setLogLevel, + setLogSink, +} from "./utils"; export type { Delivery, Webhook, WebhookInput } from "./webhooks"; export { WebhookManager, WebhookValidationError } from "./webhooks"; export { Worker } from "./worker"; +export type { + WorkflowAdvance, + WorkflowHandle, + WorkflowNode, + WorkflowRun, + WorkflowRunState, + WorkflowSpec, + WorkflowStepOptions, + WorkflowSubmitOptions, + WorkflowWaitOptions, +} from "./workflows"; +export { WorkflowBuilder, WorkflowManager } from "./workflows"; diff --git a/sdks/node/src/locks/index.ts b/sdks/node/src/locks/index.ts new file mode 100644 index 00000000..efc49ff6 --- /dev/null +++ b/sdks/node/src/locks/index.ts @@ -0,0 +1,2 @@ +export { Lock } from "./lock"; +export type { LockInfo, LockOptions } from "./types"; diff --git a/sdks/node/src/locks/lock.ts b/sdks/node/src/locks/lock.ts new file mode 100644 index 00000000..2e049bc6 --- /dev/null +++ b/sdks/node/src/locks/lock.ts @@ -0,0 +1,117 @@ +import { randomUUID } from "node:crypto"; +import type { NativeQueue } from "../native"; +import { createLogger } from "../utils"; +import type { LockInfo, LockOptions } from "./types"; + +const DEFAULT_TTL_MS = 30_000; +const log = createLogger("locks"); + +/** + * A TTL-bounded, owner-scoped distributed lock backed by the queue's storage. + * + * While held it auto-extends at `ttlMs / 3` so a slow critical section doesn't + * lose the lock; if extension ever fails (the lock expired or was stolen) the + * renewal stops. Implements `Symbol.dispose`, so it also works with `using`: + * + * ```ts + * using lock = queue.lock("report"); + * if (lock.acquire()) { + * // ... critical section; released automatically at block exit + * } + * ``` + */ +export class Lock { + readonly name: string; + readonly ownerId: string; + private readonly ttlMs: number; + private readonly autoExtend: boolean; + private timer?: ReturnType; + private held = false; + + constructor( + private readonly native: NativeQueue, + name: string, + options: LockOptions = {}, + ) { + this.name = name; + const ttl = options.ttlMs ?? DEFAULT_TTL_MS; + if (!Number.isFinite(ttl) || ttl <= 0) { + throw new RangeError(`Lock ttlMs must be a positive finite number, got ${ttl}`); + } + this.ttlMs = Math.floor(ttl); + this.ownerId = options.ownerId ?? randomUUID(); + this.autoExtend = options.autoExtend ?? true; + } + + /** Whether this handle currently believes it holds the lock. */ + get locked(): boolean { + return this.held; + } + + /** Try to acquire. Returns false if another owner holds a live lock. */ + acquire(): boolean { + if (this.held) { + // Re-validate the lease rather than trusting local state indefinitely: + // with autoExtend off the remote lock may have already expired. `extend` + // succeeds only if we still hold it; otherwise fall through to re-acquire. + if (this.extend()) { + return true; + } + this.held = false; + this.stopAutoExtend(); + } + const ok = this.native.acquireLock(this.name, this.ownerId, this.ttlMs); + if (ok) { + this.held = true; + this.startAutoExtend(); + } + return ok; + } + + /** Release if held. Returns false if this handle was not the holder. */ + release(): boolean { + this.stopAutoExtend(); + if (!this.held) { + return false; + } + this.held = false; + return this.native.releaseLock(this.name, this.ownerId); + } + + /** Extend the TTL (defaults to the original `ttlMs`). Returns false if not held. */ + extend(ttlMs: number = this.ttlMs): boolean { + return this.native.extendLock(this.name, this.ownerId, ttlMs); + } + + /** Current holder info, or `undefined` if the lock is free. */ + info(): LockInfo | undefined { + return this.native.getLockInfo(this.name) ?? undefined; + } + + [Symbol.dispose](): void { + this.release(); + } + + private startAutoExtend(): void { + if (!this.autoExtend) { + return; + } + const interval = Math.max(1, Math.floor(this.ttlMs / 3)); + this.timer = setInterval(() => { + if (!this.extend()) { + // Lost the lock (expired or stolen) — stop renewing. + log.warn(() => `lost lock '${this.name}' during auto-extend`); + this.held = false; + this.stopAutoExtend(); + } + }, interval); + this.timer.unref(); + } + + private stopAutoExtend(): void { + if (this.timer) { + clearInterval(this.timer); + this.timer = undefined; + } + } +} diff --git a/sdks/node/src/locks/types.ts b/sdks/node/src/locks/types.ts new file mode 100644 index 00000000..300d2245 --- /dev/null +++ b/sdks/node/src/locks/types.ts @@ -0,0 +1,14 @@ +import type { JsLockInfo } from "../native"; + +/** Holder info for a distributed lock. Timestamps are Unix milliseconds. */ +export type LockInfo = JsLockInfo; + +/** Options for {@link Queue.lock}. */ +export interface LockOptions { + /** Time-to-live before the lock auto-expires (ms). Default 30000. */ + ttlMs?: number; + /** Owner token. Defaults to a random UUID; only the owner can release/extend. */ + ownerId?: string; + /** Renew the lock at `ttlMs / 3` while held (default true). */ + autoExtend?: boolean; +} diff --git a/sdks/node/src/native.ts b/sdks/node/src/native.ts index ea1280dc..22c2f963 100644 --- a/sdks/node/src/native.ts +++ b/sdks/node/src/native.ts @@ -16,16 +16,22 @@ export type NativeQueue = InstanceType; export type NativeWorker = InstanceType; export type { + CircuitBreakerInput, EnqueueOptions, JobFilter, JsDeadJob, JsJob, JsJobError, + JsLockInfo, JsMetric, JsOutcome, JsStats, JsTaskInvocation, JsWorkerRow, + JsWorkflowAdvance, + JsWorkflowNode, + JsWorkflowRun, + MeshWorkerConfig, OpenOptions, QueueConfigInput, TaskConfigInput, diff --git a/sdks/node/src/queue.ts b/sdks/node/src/queue.ts index 62613f09..1d413af8 100644 --- a/sdks/node/src/queue.ts +++ b/sdks/node/src/queue.ts @@ -1,5 +1,12 @@ -import { JobCancelledError, JobFailedError, TaskitoError } from "./errors"; +import { + JobCancelledError, + JobFailedError, + LockLostError, + LockNotAcquiredError, + TaskitoError, +} from "./errors"; import { Emitter, type EventHandler, type EventName } from "./events"; +import { Lock, type LockOptions } from "./locks"; import type { Middleware } from "./middleware"; import { JsQueue, type NativeQueue, type OpenOptions } from "./native"; import { JsonSerializer, type Serializer } from "./serializers"; @@ -11,6 +18,7 @@ import type { JobError, JobFilter, Metric, + PeriodicOptions, QueueLimits, RegisteredTask, ResultOptions, @@ -22,6 +30,7 @@ import type { } from "./types"; import { WebhookManager } from "./webhooks"; import { Worker } from "./worker"; +import { WorkflowManager } from "./workflows"; /** Construction options for a {@link Queue}. */ export interface QueueOptions { @@ -55,6 +64,8 @@ export class Queue { private readonly middleware: Middleware[] = []; private readonly emitter = new Emitter(); private readonly webhookManager: WebhookManager; + /** Built lazily — its constructor throws on addons lacking the `workflows` feature. */ + private workflowManager?: WorkflowManager; constructor(options: QueueOptions) { this.native = JsQueue.open(toOpenOptions(options)); @@ -67,6 +78,66 @@ export class Queue { return this.webhookManager; } + /** Workflow definitions and runs — DAG/linear orchestration over the queue. */ + get workflows(): WorkflowManager { + if (!this.workflowManager) { + this.workflowManager = new WorkflowManager(this.native, this.serializer); + } + return this.workflowManager; + } + + /** Create a distributed lock handle (not yet acquired). */ + lock(name: string, options?: LockOptions): Lock { + return new Lock(this.native, name, options); + } + + /** + * Run `fn` while holding the named lock, releasing it afterwards. Rejects with + * {@link LockNotAcquiredError} if another owner holds the lock. + */ + async withLock(name: string, fn: () => T | Promise, options?: LockOptions): Promise { + const lock = this.lock(name, options); + if (!lock.acquire()) { + throw new LockNotAcquiredError(name); + } + let result: T; + try { + result = await fn(); + } catch (error) { + lock.release(); + throw error; + } + // `release()` returns false when the lease was lost mid-run — surface that + // rather than pretending the critical section ran under a held lock. + if (!lock.release()) { + throw new LockLostError(name); + } + return result; + } + + /** + * Register (or replace) a cron-scheduled task. A running worker enqueues + * `taskName` with the serialized `args` each time the schedule fires. Returns + * the next fire time (Unix ms); throws on an invalid cron expression. + */ + registerPeriodic( + name: string, + taskName: string, + cronExpr: string, + options?: PeriodicOptions, + ): number { + const args = Buffer.from(this.serializer.serialize(options?.args ?? [])); + return this.native.registerPeriodic( + name, + taskName, + cronExpr, + args, + options?.queue, + options?.timezone, + options?.enabled, + ); + } + /** * Register a task handler under `name`. Chain calls to build a typed registry — * {@link Queue.enqueue} then infers each task's argument types. diff --git a/sdks/node/src/types.ts b/sdks/node/src/types.ts index 168d42e0..ca1e098e 100644 --- a/sdks/node/src/types.ts +++ b/sdks/node/src/types.ts @@ -1,4 +1,7 @@ +import type { CircuitBreakerInput, MeshWorkerConfig } from "./native"; + export type { + CircuitBreakerInput as CircuitBreakerOptions, EnqueueOptions, JobFilter, JsDeadJob as DeadJob, @@ -7,6 +10,7 @@ export type { JsMetric as Metric, JsStats as Stats, JsWorkerRow as WorkerInfo, + MeshWorkerConfig, } from "./native"; /** Options for {@link Queue.result}. */ @@ -47,6 +51,20 @@ export interface TaskOptions { maxConcurrent?: number; /** Rate-limit spec like `"100/m"`, `"50/s"`, `"3600/h"`. */ rateLimit?: RateLimit; + /** Trip the task's circuit breaker after repeated failures. */ + circuitBreaker?: CircuitBreakerInput; +} + +/** Options for {@link Queue.registerPeriodic}. */ +export interface PeriodicOptions { + /** Positional args passed to the task each time it fires. */ + args?: unknown[]; + /** Queue the periodic job runs on (default `"default"`). */ + queue?: string; + /** IANA timezone for the cron schedule (default UTC). */ + timezone?: string; + /** Register disabled (won't fire until re-registered enabled). Default true. */ + enabled?: boolean; } /** Per-queue resilience config. */ @@ -69,4 +87,15 @@ export interface WorkerRunOptions { channelCapacity?: number; /** Jobs claimed per scheduler poll (default 1). */ batchSize?: number; + /** + * Opt-in decentralized mesh overlay (peer gossip + work-stealing). Requires + * the native addon to be built with the `mesh` cargo feature; ignored otherwise. + */ + mesh?: MeshWorkerConfig; + /** + * Advance workflow runs as this worker's node-jobs settle (default true when + * the addon supports workflows). Adds one job lookup per terminal job; set + * false on workers that never process workflow steps. + */ + advanceWorkflows?: boolean; } diff --git a/sdks/node/src/utils/index.ts b/sdks/node/src/utils/index.ts new file mode 100644 index 00000000..2f4221f8 --- /dev/null +++ b/sdks/node/src/utils/index.ts @@ -0,0 +1,2 @@ +export type { Logger, LogLevel, LogMessage, LogSink } from "./logger"; +export { createLogger, logger, setLogLevel, setLogSink } from "./logger"; diff --git a/sdks/node/src/utils/logger.ts b/sdks/node/src/utils/logger.ts new file mode 100644 index 00000000..cacbeb14 --- /dev/null +++ b/sdks/node/src/utils/logger.ts @@ -0,0 +1,114 @@ +// Tiny zero-dependency leveled logger. Writes to stderr by default so it never +// pollutes stdout (the CLI's `--json` output and piped data stay clean). Level +// is read once from `TASKITO_LOG_LEVEL` and is overridable at runtime; the sink +// is pluggable for tests or custom transports. + +/** Severity, from most to least verbose. `silent` disables all output. */ +export type LogLevel = "debug" | "info" | "warn" | "error" | "silent"; + +/** A message, or a thunk evaluated only when the level passes the threshold. */ +export type LogMessage = string | (() => string); + +/** Receives every formatted line that clears the level threshold. */ +export type LogSink = (level: Exclude, line: string) => void; + +export interface Logger { + debug(message: LogMessage, ...meta: unknown[]): void; + info(message: LogMessage, ...meta: unknown[]): void; + warn(message: LogMessage, ...meta: unknown[]): void; + error(message: LogMessage, ...meta: unknown[]): void; + /** Derive a namespaced child logger (`[taskito:webhooks]`). */ + child(namespace: string): Logger; +} + +const SEVERITY: Record = { + debug: 10, + info: 20, + warn: 30, + error: 40, + silent: Number.POSITIVE_INFINITY, +}; + +const defaultSink: LogSink = (_level, line) => { + process.stderr.write(`${line}\n`); +}; + +function isLogLevel(value: unknown): value is LogLevel { + // `Object.hasOwn` avoids matching inherited props (e.g. `TASKITO_LOG_LEVEL=toString`). + return typeof value === "string" && Object.hasOwn(SEVERITY, value); +} + +function levelFromEnv(): LogLevel { + const raw = process.env.TASKITO_LOG_LEVEL?.toLowerCase(); + return isLogLevel(raw) ? raw : "warn"; +} + +// Shared mutable config: every logger (root + children) reads through it, so +// setLogLevel / setLogSink take effect globally and immediately. +const config = { level: levelFromEnv(), sink: defaultSink }; + +/** Set the global threshold; messages below it are dropped (and never built). */ +export function setLogLevel(level: LogLevel): void { + if (!isLogLevel(level)) { + throw new TypeError(`invalid log level: ${String(level)}`); + } + config.level = level; +} + +/** Replace the output sink (default: stderr). Useful for capture in tests. */ +export function setLogSink(sink: LogSink): void { + if (typeof sink !== "function") { + throw new TypeError("log sink must be a function"); + } + config.sink = sink; +} + +function formatMeta(value: unknown): string { + if (value instanceof Error) { + return value.stack ?? `${value.name}: ${value.message}`; + } + if (typeof value === "object" && value !== null) { + try { + return JSON.stringify(value); + } catch { + return String(value); + } + } + return String(value); +} + +function emit( + namespace: string | undefined, + level: Exclude, + message: LogMessage, + meta: unknown[], +): void { + if (SEVERITY[level] < SEVERITY[config.level]) { + return; + } + const tag = namespace ? `taskito:${namespace}` : "taskito"; + const text = typeof message === "function" ? message() : message; + const parts = [new Date().toISOString(), `[${tag}]`, level.toUpperCase(), text]; + for (const item of meta) { + parts.push(formatMeta(item)); + } + config.sink(level, parts.join(" ")); +} + +function make(namespace?: string): Logger { + return { + debug: (message, ...meta) => emit(namespace, "debug", message, meta), + info: (message, ...meta) => emit(namespace, "info", message, meta), + warn: (message, ...meta) => emit(namespace, "warn", message, meta), + error: (message, ...meta) => emit(namespace, "error", message, meta), + child: (child) => make(namespace ? `${namespace}:${child}` : child), + }; +} + +/** The root logger. Prefer {@link createLogger} for a namespaced one. */ +export const logger: Logger = make(); + +/** Create a namespaced logger (`createLogger("worker")` → `[taskito:worker]`). */ +export function createLogger(namespace: string): Logger { + return make(namespace); +} diff --git a/sdks/node/src/webhooks/deliverer.ts b/sdks/node/src/webhooks/deliverer.ts index 87c507f1..97a6eaf6 100644 --- a/sdks/node/src/webhooks/deliverer.ts +++ b/sdks/node/src/webhooks/deliverer.ts @@ -1,10 +1,13 @@ import { createHmac, randomUUID } from "node:crypto"; import type { EventName, OutcomeEvent } from "../events"; +import { createLogger } from "../utils"; import type { Delivery, Webhook } from "./types"; const MAX_RECENT = 100; const MAX_BACKOFF_MS = 30_000; +const log = createLogger("webhooks"); + /** Signs and POSTs webhook payloads with retries; keeps a recent-delivery log. */ export class Deliverer { private readonly recent: Delivery[] = []; @@ -65,10 +68,29 @@ export class Deliverer { if (this.recent.length > MAX_RECENT) { this.recent.shift(); } + if (!delivery.ok) { + log.warn( + () => + `delivery of ${event} to ${redactUrl(webhook.url)} failed after ${attempts} attempt(s): ${error}`, + ); + } return delivery; } } +/** Strip credentials and query string from a URL before logging it. */ +function redactUrl(raw: string): string { + try { + const url = new URL(raw); + url.username = ""; + url.password = ""; + url.search = ""; + return url.toString(); + } catch { + return ""; + } +} + function sleep(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); } diff --git a/sdks/node/src/worker.ts b/sdks/node/src/worker.ts index 20a7bb20..bd4f05c0 100644 --- a/sdks/node/src/worker.ts +++ b/sdks/node/src/worker.ts @@ -13,6 +13,10 @@ import type { } from "./native"; import type { Serializer } from "./serializers"; import type { QueueLimits, RegisteredTask, WorkerRunOptions } from "./types"; +import { createLogger } from "./utils"; +import { WorkflowTracker } from "./workflows"; + +const log = createLogger("worker"); /** How often a running job polls the storage cancel flag. */ const CANCEL_POLL_INTERVAL_MS = 200; @@ -48,6 +52,12 @@ export class Worker { static start(queue: NativeQueue, params: WorkerStartParams): Worker { const { tasks, queueLimits, serializer, middleware, emitter, run } = params; + // Advance workflow runs as node-jobs settle, unless disabled or unsupported. + const tracker = + (run?.advanceWorkflows ?? true) && typeof queue.markWorkflowNodeResult === "function" + ? new WorkflowTracker(queue, serializer) + : null; + const taskCallback = async (invocation: JsTaskInvocation): Promise => { const task = tasks.get(invocation.taskName); if (!task) { @@ -68,8 +78,9 @@ export class Worker { if (queue.isCancelRequested(invocation.id)) { controller.abort(); } - } catch { + } catch (error) { // transient storage error — retry on the next tick + log.debug(() => `cancel poll for ${invocation.id} failed`, error); } }, CANCEL_POLL_INTERVAL_MS); poller.unref(); @@ -115,10 +126,12 @@ export class Worker { const hook = mw[mapping.hook] as ((e: OutcomeEvent) => void) | undefined; try { hook?.(event); - } catch { + } catch (error) { // outcome hook errors must not break the worker + log.debug(() => `${mapping.hook} middleware hook threw for ${outcome.jobId}`, error); } } + tracker?.onOutcome(outcome); }; const nativeOptions: NativeWorkerOptions = { @@ -127,6 +140,7 @@ export class Worker { batchSize: run?.batchSize, taskConfigs: buildTaskConfigs(tasks), queueConfigs: buildQueueConfigs(queueLimits), + mesh: run?.mesh, }; return new Worker(queue.runWorker(taskCallback, outcomeCallback, nativeOptions)); } @@ -147,7 +161,8 @@ function buildTaskConfigs(tasks: ReadonlyMap): TaskConfi (options.maxRetries === undefined && options.retryBackoff === undefined && options.maxConcurrent === undefined && - options.rateLimit === undefined) + options.rateLimit === undefined && + options.circuitBreaker === undefined) ) { continue; } @@ -158,6 +173,7 @@ function buildTaskConfigs(tasks: ReadonlyMap): TaskConfi retryMaxDelayMs: options.retryBackoff?.maxMs, maxConcurrent: options.maxConcurrent, rateLimit: options.rateLimit, + circuitBreaker: options.circuitBreaker, }); } return configs; diff --git a/sdks/node/src/workflows/builder.ts b/sdks/node/src/workflows/builder.ts new file mode 100644 index 00000000..20bf578f --- /dev/null +++ b/sdks/node/src/workflows/builder.ts @@ -0,0 +1,137 @@ +import { successorsOf, transitiveDeferred } from "./plan"; +import type { + FanInStepOptions, + FanOutStepOptions, + StepMetadataJson, + WorkflowHandle, + WorkflowSpec, + WorkflowStepOptions, +} from "./types"; + +/** + * Fluent builder for a workflow DAG. Each {@link WorkflowBuilder.step} adds a + * node; `after` declares edges. Steps may reference predecessors that are added + * later — the DAG is validated (and topologically ordered) at submit time. + * + * {@link WorkflowBuilder.fanOut} / {@link WorkflowBuilder.fanIn} add dynamic + * steps the tracker expands at runtime (see `tracker.ts`). + * + * Obtain one via `queue.workflows.define(name)`; finish with `.submit()`. + */ +export class WorkflowBuilder { + private readonly nodes: string[] = []; + private readonly edges: Array<{ from: string; to: string }> = []; + private readonly stepMetadata: Record = {}; + private readonly stepArgs: Record = {}; + private readonly seen = new Set(); + /** Nodes that run no static job — expanded/enqueued by the tracker. */ + private readonly deferredSeeds = new Set(); + + constructor( + private readonly name: string, + private readonly version: number, + private readonly submitFn: (spec: WorkflowSpec) => WorkflowHandle, + ) {} + + /** Add a step `name` that runs the registered task `taskName`. */ + step(name: string, taskName: string, options: WorkflowStepOptions = {}): this { + this.addNode(name, options.after); + this.stepMetadata[name] = { + task_name: taskName, + queue: options.queue, + max_retries: options.maxRetries, + timeout_ms: options.timeoutMs, + priority: options.priority, + }; + this.stepArgs[name] = options.args ?? []; + return this; + } + + /** + * Add a fan-out step. At runtime the tracker reads the array result of + * `itemsFrom` and expands one child per item, each running `options.task`. + */ + fanOut(name: string, options: FanOutStepOptions): this { + this.addNode(name, options.after); + this.stepMetadata[name] = { + task_name: options.task, + queue: options.queue, + max_retries: options.maxRetries, + timeout_ms: options.timeoutMs, + priority: options.priority, + fan_out: JSON.stringify({ itemsFrom: options.itemsFrom ?? null }), + }; + this.stepArgs[name] = []; + this.deferredSeeds.add(name); + return this; + } + + /** + * Add a fan-in step that collects the children of fan-out `options.after` + * and runs `options.task` with `[childResult, …]` as its single argument. + */ + fanIn(name: string, options: FanInStepOptions): this { + this.addNode(name, options.after); + this.stepMetadata[name] = { + task_name: options.task, + queue: options.queue, + max_retries: options.maxRetries, + timeout_ms: options.timeoutMs, + priority: options.priority, + fan_in: JSON.stringify({ from: options.after }), + }; + this.stepArgs[name] = []; + this.deferredSeeds.add(name); + return this; + } + + /** Materialize the validated spec without submitting. */ + build(): WorkflowSpec { + for (const edge of this.edges) { + if (!this.seen.has(edge.from)) { + throw new Error(`step '${edge.to}' depends on unknown step '${edge.from}'`); + } + } + // A fan-in node is only ever enqueued by its fan-out's completion, so its + // `after` must point at a fan-out step — otherwise the run would hang. + for (const [name, meta] of Object.entries(this.stepMetadata)) { + if (!meta.fan_in) { + continue; + } + const { from } = JSON.parse(meta.fan_in) as { from: string }; + if (!this.stepMetadata[from]?.fan_out) { + throw new Error( + `fan-in step '${name}' must target a fan-out step, but '${from}' is not one`, + ); + } + } + const deferred = transitiveDeferred(this.deferredSeeds, successorsOf(this.edges)); + return { + name: this.name, + version: this.version, + nodes: [...this.nodes], + edges: [...this.edges], + stepMetadata: { ...this.stepMetadata }, + stepArgs: { ...this.stepArgs }, + deferredNodeNames: [...deferred], + }; + } + + /** Build, then submit the run. Returns a handle carrying the run id. */ + submit(): WorkflowHandle { + return this.submitFn(this.build()); + } + + /** Register a node + its incoming edges, rejecting duplicate names. */ + private addNode(name: string, after: string | string[] | undefined): void { + if (this.seen.has(name)) { + throw new Error(`duplicate workflow step '${name}'`); + } + this.seen.add(name); + this.nodes.push(name); + const predecessors = after === undefined ? [] : Array.isArray(after) ? after : [after]; + for (const from of predecessors) { + this.edges.push({ from, to: name }); + } + } +} diff --git a/sdks/node/src/workflows/index.ts b/sdks/node/src/workflows/index.ts new file mode 100644 index 00000000..546b1507 --- /dev/null +++ b/sdks/node/src/workflows/index.ts @@ -0,0 +1,16 @@ +export { WorkflowBuilder } from "./builder"; +export { WorkflowManager } from "./manager"; +export { WorkflowTracker } from "./tracker"; +export type { + FanInStepOptions, + FanOutStepOptions, + WorkflowAdvance, + WorkflowHandle, + WorkflowNode, + WorkflowRun, + WorkflowRunState, + WorkflowSpec, + WorkflowStepOptions, + WorkflowSubmitOptions, + WorkflowWaitOptions, +} from "./types"; diff --git a/sdks/node/src/workflows/manager.ts b/sdks/node/src/workflows/manager.ts new file mode 100644 index 00000000..f3dcb519 --- /dev/null +++ b/sdks/node/src/workflows/manager.ts @@ -0,0 +1,158 @@ +import type { NativeQueue } from "../native"; +import type { Serializer } from "../serializers"; +import { WorkflowBuilder } from "./builder"; +import type { + WorkflowHandle, + WorkflowNode, + WorkflowRun, + WorkflowSpec, + WorkflowSubmitOptions, + WorkflowWaitOptions, +} from "./types"; + +/** Run states with no further transitions. */ +const TERMINAL_STATES = new Set([ + "completed", + "completed_with_failures", + "failed", + "cancelled", + "compensated", + "compensation_failed", +]); + +const DEFAULT_WAIT_TIMEOUT_MS = 30_000; +const DEFAULT_WAIT_POLL_MS = 100; + +/** + * The `queue.workflows` facade: define and submit workflows, then query runs. + * + * Supports DAG / linear workflows (steps pre-enqueued with `depends_on` chains, + * sequenced by the core scheduler) plus fan-out / fan-in steps the worker-side + * tracker expands at runtime (see `tracker.ts`). Gates, sub-workflows, and saga + * compensation are not yet available in the Node SDK. + */ +export class WorkflowManager { + constructor( + private readonly native: NativeQueue, + private readonly serializer: Serializer, + ) { + if (typeof this.native.submitWorkflow !== "function") { + throw new Error("the native addon was built without the 'workflows' feature"); + } + } + + /** Start defining a workflow. Chain `.step(...)` then `.submit()`. */ + define(name: string, version = 1): WorkflowBuilder { + return new WorkflowBuilder(name, version, (spec) => this.submitSpec(spec)); + } + + /** Submit a pre-built builder (alternative to `builder.submit()`). */ + submit(builder: WorkflowBuilder, options?: WorkflowSubmitOptions): WorkflowHandle { + return this.submitSpec(builder.build(), options); + } + + /** Fetch a run by id, or `undefined` if it doesn't exist. */ + run(runId: string): WorkflowRun | undefined { + return this.native.getWorkflowRun(runId) ?? undefined; + } + + /** The step nodes of a run. */ + nodes(runId: string): WorkflowNode[] { + return this.native.getWorkflowNodes(runId); + } + + /** The serialized DAG (graph JSON) for a run, or `undefined` if unknown. */ + dag(runId: string): string | undefined { + return this.native.getWorkflowDag(runId) ?? undefined; + } + + /** Sub-workflow runs spawned by a run (empty for Node-submitted runs). */ + children(runId: string): WorkflowRun[] { + return this.native.getWorkflowChildren(runId); + } + + /** List runs, optionally filtered by definition name and/or state. */ + list(options?: { + definitionName?: string; + state?: string; + limit?: number; + offset?: number; + }): WorkflowRun[] { + return this.native.listWorkflowRuns( + options?.definitionName ?? null, + options?.state ?? null, + options?.limit ?? null, + options?.offset ?? null, + ); + } + + private submitSpec(spec: WorkflowSpec, options?: WorkflowSubmitOptions): WorkflowHandle { + const dag = { + nodes: spec.nodes.map((name) => ({ name })), + edges: spec.edges.map((edge) => ({ from: edge.from, to: edge.to, weight: 1.0 })), + }; + const dagBytes = Buffer.from(JSON.stringify(dag)); + + // Static nodes are enqueued by `submitWorkflow` from these payloads. For + // deferred nodes the tracker enqueues later, so we persist their args in + // `args_template` (base64) — the only storage-reconstructable channel. + const deferred = new Set(spec.deferredNodeNames); + const stepMetadata = { ...spec.stepMetadata }; + const nodePayloads: Record = {}; + for (const name of spec.nodes) { + const payload = Buffer.from(this.serializer.serialize(spec.stepArgs[name] ?? [])); + nodePayloads[name] = payload; + const meta = stepMetadata[name]; + if (deferred.has(name) && meta) { + stepMetadata[name] = { ...meta, args_template: payload.toString("base64") }; + } + } + + const paramsJson = options?.params === undefined ? null : JSON.stringify(options.params); + const runId = this.native.submitWorkflow( + spec.name, + spec.version, + dagBytes, + JSON.stringify(stepMetadata), + nodePayloads, + options?.queueDefault ?? null, + paramsJson, + spec.deferredNodeNames, + ); + return this.makeHandle(runId); + } + + private makeHandle(runId: string): WorkflowHandle { + return { + runId, + status: () => this.run(runId), + nodes: () => this.nodes(runId), + wait: (waitOptions) => this.waitForRun(runId, waitOptions), + }; + } + + private async waitForRun(runId: string, options?: WorkflowWaitOptions): Promise { + const timeoutMs = options?.timeoutMs ?? DEFAULT_WAIT_TIMEOUT_MS; + const pollMs = options?.pollMs ?? DEFAULT_WAIT_POLL_MS; + if (!Number.isFinite(timeoutMs) || timeoutMs <= 0) { + throw new Error(`workflow wait timeoutMs must be a positive number, got ${timeoutMs}`); + } + if (!Number.isFinite(pollMs) || pollMs <= 0) { + throw new Error(`workflow wait pollMs must be a positive number, got ${pollMs}`); + } + const deadline = Date.now() + timeoutMs; + for (;;) { + const run = this.native.getWorkflowRun(runId); + if (!run) { + throw new Error(`workflow run '${runId}' not found`); + } + if (TERMINAL_STATES.has(run.state)) { + return run; + } + if (Date.now() >= deadline) { + throw new Error(`workflow run '${runId}' did not finish within ${timeoutMs}ms`); + } + await new Promise((resolve) => setTimeout(resolve, pollMs)); + } + } +} diff --git a/sdks/node/src/workflows/plan.ts b/sdks/node/src/workflows/plan.ts new file mode 100644 index 00000000..2f4cc861 --- /dev/null +++ b/sdks/node/src/workflows/plan.ts @@ -0,0 +1,60 @@ +// Shared DAG-structure helpers used by both the build-time {@link WorkflowBuilder} +// and the runtime {@link WorkflowTracker}. Pure functions — no I/O. + +/** A directed edge `from → to` in a workflow DAG. */ +export interface DagEdge { + from: string; + to: string; +} + +/** Map each node to the nodes that depend on it (its direct successors). */ +export function successorsOf(edges: readonly DagEdge[]): Map { + return adjacency(edges, (e) => [e.from, e.to]); +} + +/** Map each node to its direct predecessors (the nodes it depends on). */ +export function predecessorsOf(edges: readonly DagEdge[]): Map { + return adjacency(edges, (e) => [e.to, e.from]); +} + +/** Build an adjacency map, picking the (key, value) node from each edge. */ +function adjacency( + edges: readonly DagEdge[], + pick: (edge: DagEdge) => [string, string], +): Map { + const map = new Map(); + for (const edge of edges) { + const [key, value] = pick(edge); + const list = map.get(key); + if (list) { + list.push(value); + } else { + map.set(key, [value]); + } + } + return map; +} + +/** + * The set of deferred nodes: the `seeds` (fan-out / fan-in nodes the tracker + * enqueues on demand) plus every node reachable from them. A node downstream of + * a deferred one has no static job to depend on, so it too must be enqueued by + * the tracker once its predecessors settle. + */ +export function transitiveDeferred( + seeds: Iterable, + successors: Map, +): Set { + const deferred = new Set(); + const stack = [...seeds]; + for (let node = stack.pop(); node !== undefined; node = stack.pop()) { + if (deferred.has(node)) { + continue; + } + deferred.add(node); + for (const succ of successors.get(node) ?? []) { + stack.push(succ); + } + } + return deferred; +} diff --git a/sdks/node/src/workflows/tracker.ts b/sdks/node/src/workflows/tracker.ts new file mode 100644 index 00000000..41ca5d81 --- /dev/null +++ b/sdks/node/src/workflows/tracker.ts @@ -0,0 +1,349 @@ +import type { JsOutcome, NativeQueue } from "../native"; +import type { Serializer } from "../serializers"; +import { createLogger } from "../utils"; +import { predecessorsOf, successorsOf, transitiveDeferred } from "./plan"; + +const log = createLogger("workflow-tracker"); + +const DEFAULT_QUEUE = "default"; +const DEFAULT_MAX_RETRIES = 3; +const DEFAULT_TIMEOUT_MS = 300_000; + +/** Node statuses with no further transitions (mirrors `WorkflowNodeStatus::is_terminal`). */ +const TERMINAL_NODE_STATUS = new Set([ + "completed", + "failed", + "skipped", + "cache_hit", + "compensated", + "compensation_failed", +]); + +/** A fan-out child node name (`parent[3]`) → its parent (`parent`), else null. */ +const FAN_OUT_CHILD = /^(.+)\[\d+\]$/; + +/** snake_case step metadata as stored in the workflow definition. */ +interface StepMeta { + task_name: string; + queue?: string; + max_retries?: number; + timeout_ms?: number; + priority?: number; + args_template?: string; + fan_out?: string; + fan_in?: string; +} + +/** The reconstructed structure + config of a workflow run. */ +interface RunPlan { + successors: Map; + predecessors: Map; + deferred: Set; + meta: Map; +} + +/** + * The runtime workflow brain. Driven from the worker's outcome callback, it + * advances a run as each node-job settles. For plain DAGs the core scheduler + * already sequences nodes via `depends_on`, so the tracker only records status. + * For runs with deferred (fan-out / fan-in) nodes it owns the on-demand + * orchestration: expand fan-outs, collect fan-ins, enqueue downstream nodes, + * cascade fail-fast, and finalize the run. + * + * The run plan is reconstructed from storage (DAG + step metadata), so the + * tracker works regardless of which process submitted the run. + * + * Runs entirely synchronously inside the (sequential) outcome callback: job + * results are committed before the callback fires, so reads never race. + */ +export class WorkflowTracker { + private readonly plans = new Map(); + + constructor( + private readonly native: NativeQueue, + private readonly serializer: Serializer, + ) {} + + /** Handle one worker outcome. A no-op for non-workflow jobs. */ + onOutcome(outcome: JsOutcome): void { + let succeeded: boolean; + if (outcome.kind === "success") { + succeeded = true; + } else if (outcome.kind === "dead") { + succeeded = false; + } else { + return; // retry / cancelled are not terminal for workflow advancement + } + + try { + const ref = this.native.workflowNodeForJob(outcome.jobId); + if (!ref) { + return; // not a workflow job + } + this.advance(outcome.jobId, ref.runId, ref.nodeName, succeeded, outcome.error ?? null); + } catch (error) { + // Workflow bookkeeping must never break the worker loop. + log.debug(() => `workflow advance for ${outcome.jobId} failed`, error); + } + } + + private advance( + jobId: string, + runId: string, + nodeName: string, + succeeded: boolean, + error: string | null, + ): void { + const plan = this.plan(runId); + if (!plan) { + return; + } + const managed = plan.deferred.size > 0; + + // Record the node's outcome. For plain DAGs the core also cascades + finalizes. + const advance = this.native.markWorkflowNodeResult(jobId, succeeded, error, managed); + if (!managed) { + this.evictIfTerminal(runId, advance?.finalState ?? null); + return; + } + + const parent = fanOutParentOf(nodeName); + if (parent !== null && isFanOut(plan, parent)) { + this.onFanOutChild(runId, parent, plan); + } else if (succeeded) { + this.evaluateSuccessors(runId, nodeName, plan); + } else { + this.native.cascadeSkipPending(runId); // fail-fast + } + + this.evictIfTerminal(runId, this.native.finalizeRunIfTerminal(runId)); + } + + /** Drop a finished run's cached plan so the cache doesn't grow unbounded. */ + private evictIfTerminal(runId: string, finalState: string | null): void { + if (finalState !== null) { + this.plans.delete(runId); + } + } + + /** A fan-out child settled — finalize the parent once all children are terminal. */ + private onFanOutChild(runId: string, parent: string, plan: RunPlan): void { + const completion = this.native.checkFanOutCompletion(runId, parent); + if (!completion) { + return; // children still in flight, or another caller already finalized + } + if (completion.succeeded) { + this.onFanOutParentDone(runId, parent, completion.childJobIds, plan); + } else { + this.native.cascadeSkipPending(runId); // a child failed → fail-fast + } + } + + /** A fan-out parent completed successfully — collect into fan-in or move on. */ + private onFanOutParentDone( + runId: string, + parent: string, + childJobIds: string[], + plan: RunPlan, + ): void { + const fanInNode = this.fanInNodeFor(parent, plan); + if (fanInNode) { + this.createFanInJob(runId, fanInNode, childJobIds, plan); + } else { + this.evaluateSuccessors(runId, parent, plan); + } + } + + /** Read `itemsFrom`'s array result and expand the fan-out into one child per item. */ + private expandFanOut(runId: string, node: string, plan: RunPlan): void { + const meta = plan.meta.get(node); + if (!meta?.fan_out) { + return; + } + const itemsFrom = + (JSON.parse(meta.fan_out).itemsFrom as string | null) ?? singlePred(plan, node); + const items = this.readArrayResult(runId, itemsFrom); + const childNames = items.map((_, i) => `${node}[${i}]`); + const childPayloads = items.map((item) => Buffer.from(this.serializer.serialize([item]))); + + this.native.expandFanOut( + runId, + node, + childNames, + childPayloads, + meta.task_name, + meta.queue ?? DEFAULT_QUEUE, + meta.max_retries ?? DEFAULT_MAX_RETRIES, + meta.timeout_ms ?? DEFAULT_TIMEOUT_MS, + meta.priority ?? 0, + ); + + // Empty fan-out completes the parent immediately, so no child outcome will + // fire to carry the run forward — do it inline. + if (items.length === 0) { + this.onFanOutParentDone(runId, node, [], plan); + } + } + + /** Collect the children's results and enqueue the combiner job. */ + private createFanInJob( + runId: string, + fanInNode: string, + childJobIds: string[], + plan: RunPlan, + ): void { + const results = childJobIds.map((id) => { + const job = this.native.getJob(id); + if (!job?.result) { + throw new Error(`fan-in: child job '${id}' has no result`); + } + return this.serializer.deserialize(job.result); + }); + const meta = plan.meta.get(fanInNode); + if (!meta) { + return; + } + // The combiner receives the whole list as its single positional argument. + const payload = Buffer.from(this.serializer.serialize([results])); + this.native.createDeferredJob( + runId, + fanInNode, + payload, + meta.task_name, + meta.queue ?? DEFAULT_QUEUE, + meta.max_retries ?? DEFAULT_MAX_RETRIES, + meta.timeout_ms ?? DEFAULT_TIMEOUT_MS, + meta.priority ?? 0, + ); + } + + /** Enqueue/expand any deferred successor whose predecessors are now all terminal. */ + private evaluateSuccessors(runId: string, nodeName: string, plan: RunPlan): void { + for (const succ of plan.successors.get(nodeName) ?? []) { + if (!plan.deferred.has(succ) || !this.allPredecessorsTerminal(runId, succ, plan)) { + continue; + } + const meta = plan.meta.get(succ); + if (meta?.fan_in) { + continue; // fan-in jobs are created from the fan-out completion path + } + if (meta?.fan_out) { + this.expandFanOut(runId, succ, plan); + } else { + this.createDeferredJob(runId, succ, plan); + } + } + } + + /** Enqueue a plain deferred node using its persisted args. */ + private createDeferredJob(runId: string, node: string, plan: RunPlan): void { + const meta = plan.meta.get(node); + if (!meta) { + return; + } + const payload = meta.args_template + ? Buffer.from(meta.args_template, "base64") + : Buffer.from(this.serializer.serialize([])); + this.native.createDeferredJob( + runId, + node, + payload, + meta.task_name, + meta.queue ?? DEFAULT_QUEUE, + meta.max_retries ?? DEFAULT_MAX_RETRIES, + meta.timeout_ms ?? DEFAULT_TIMEOUT_MS, + meta.priority ?? 0, + ); + } + + private allPredecessorsTerminal(runId: string, node: string, plan: RunPlan): boolean { + const preds = plan.predecessors.get(node) ?? []; + if (preds.length === 0) { + return true; + } + const status = this.nodeStatuses(runId); + return preds.every((p) => TERMINAL_NODE_STATUS.has(status.get(p) ?? "")); + } + + /** The combiner-input node that collects a given fan-out, if any. */ + private fanInNodeFor(parent: string, plan: RunPlan): string | null { + for (const succ of plan.successors.get(parent) ?? []) { + const meta = plan.meta.get(succ); + if (meta?.fan_in) { + const from = JSON.parse(meta.fan_in).from as string | undefined; + if (!from || from === parent) { + return succ; + } + } + } + return null; + } + + /** Deserialize the array result of a node's job (its sole positional arg shape). */ + private readArrayResult(runId: string, nodeName: string): unknown[] { + const node = this.native.getWorkflowNodes(runId).find((n) => n.nodeName === nodeName); + if (!node?.jobId) { + throw new Error(`fan-out source '${nodeName}' has no job`); + } + const job = this.native.getJob(node.jobId); + if (!job?.result) { + throw new Error(`fan-out source '${nodeName}' has no result`); + } + const value = this.serializer.deserialize(job.result); + if (!Array.isArray(value)) { + throw new Error(`fan-out source '${nodeName}' must return an array`); + } + return value; + } + + private nodeStatuses(runId: string): Map { + return new Map(this.native.getWorkflowNodes(runId).map((n) => [n.nodeName, n.status])); + } + + /** Load + cache the run plan from storage. Returns null if the run is gone. */ + private plan(runId: string): RunPlan | null { + const cached = this.plans.get(runId); + if (cached) { + return cached; + } + const raw = this.native.getWorkflowRunPlan(runId); + if (!raw) { + return null; + } + const graph = JSON.parse(raw.dag) as { edges?: Array<{ from: string; to: string }> }; + const edges = graph.edges ?? []; + const meta = new Map( + Object.entries(JSON.parse(raw.stepMetadata) as Record), + ); + const successors = successorsOf(edges); + const seeds = [...meta].filter(([, m]) => m.fan_out || m.fan_in).map(([name]) => name); + const plan: RunPlan = { + successors, + predecessors: predecessorsOf(edges), + deferred: transitiveDeferred(seeds, successors), + meta, + }; + this.plans.set(runId, plan); + return plan; + } +} + +/** `parent[3]` → `parent`; a non-child name → null. */ +function fanOutParentOf(nodeName: string): string | null { + return FAN_OUT_CHILD.exec(nodeName)?.[1] ?? null; +} + +function isFanOut(plan: RunPlan, node: string): boolean { + return Boolean(plan.meta.get(node)?.fan_out); +} + +function singlePred(plan: RunPlan, node: string): string { + const preds = plan.predecessors.get(node) ?? []; + const [only] = preds; + if (preds.length !== 1 || only === undefined) { + throw new Error( + `fan-out '${node}' needs exactly one predecessor or an explicit itemsFrom (has ${preds.length})`, + ); + } + return only; +} diff --git a/sdks/node/src/workflows/types.ts b/sdks/node/src/workflows/types.ts new file mode 100644 index 00000000..cec02768 --- /dev/null +++ b/sdks/node/src/workflows/types.ts @@ -0,0 +1,122 @@ +import type { JsWorkflowAdvance, JsWorkflowNode, JsWorkflowRun } from "../native"; + +/** A workflow run row (state, timestamps, lineage). */ +export type WorkflowRun = JsWorkflowRun; +/** One step of a run, linked to its underlying job. */ +export type WorkflowNode = JsWorkflowNode; +/** Result of advancing a node — `finalState` set once the run is terminal. */ +export type WorkflowAdvance = JsWorkflowAdvance; + +/** Lowercase workflow run states (mirrors the Rust `WorkflowState`). */ +export type WorkflowRunState = + | "pending" + | "running" + | "paused" + | "completed" + | "completed_with_failures" + | "failed" + | "cancelled" + | "compensating" + | "compensated" + | "compensation_failed"; + +/** Per-step options when building a workflow. */ +export interface WorkflowStepOptions { + /** Predecessor step name(s) this step runs after. */ + after?: string | string[]; + /** Positional args passed to the step's task handler. */ + args?: unknown[]; + /** Queue to run this step on (defaults to the submit-time `queueDefault`). */ + queue?: string; + maxRetries?: number; + timeoutMs?: number; + priority?: number; +} + +/** Common per-step task config shared by every step kind. */ +interface StepTaskOptions { + queue?: string; + maxRetries?: number; + timeoutMs?: number; + priority?: number; +} + +/** + * Options for a fan-out step. The step itself runs no job; at runtime the + * tracker reads the array result of `itemsFrom` (its sole predecessor by + * default) and expands one child per item, each running `task` with that item + * as its single argument. + */ +export interface FanOutStepOptions extends StepTaskOptions { + /** Predecessor step name(s). At least one is required. */ + after: string | string[]; + /** Task each child runs, once per item. */ + task: string; + /** Predecessor whose array result supplies the items. Defaults to the sole predecessor. */ + itemsFrom?: string; +} + +/** + * Options for a fan-in step. Collects the results of a fan-out's children into + * an array and runs `task` with that array as its single argument. + */ +export interface FanInStepOptions extends StepTaskOptions { + /** The fan-out step to collect. */ + after: string; + /** Combiner task; receives `[childResult, …]` as its single argument. */ + task: string; +} + +/** snake_case step metadata, matching the Rust `StepMetadata` shape. */ +export interface StepMetadataJson { + task_name: string; + queue?: string; + max_retries?: number; + timeout_ms?: number; + priority?: number; + /** Base64 of the serialized args, persisted so the tracker can enqueue deferred nodes. */ + args_template?: string; + /** JSON `{itemsFrom}` marking a fan-out node. */ + fan_out?: string; + /** JSON `{from}` marking a fan-in node (the fan-out it collects). */ + fan_in?: string; +} + +/** A built workflow definition, ready to submit. */ +export interface WorkflowSpec { + name: string; + version: number; + nodes: string[]; + edges: Array<{ from: string; to: string }>; + stepMetadata: Record; + stepArgs: Record; + /** Nodes the tracker enqueues on demand (fan-out/fan-in ∪ their descendants). */ + deferredNodeNames: string[]; +} + +/** Submit-time options. */ +export interface WorkflowSubmitOptions { + /** Default queue for steps that don't set their own. */ + queueDefault?: string; + /** Arbitrary params recorded on the run (JSON-encoded). */ + params?: unknown; +} + +/** Options for {@link WorkflowHandle.wait}. */ +export interface WorkflowWaitOptions { + /** Max time to wait for a terminal state (ms). Default 30000. */ + timeoutMs?: number; + /** Poll interval (ms). Default 100. */ + pollMs?: number; +} + +/** Handle to a submitted run: query status/nodes or await completion. */ +export interface WorkflowHandle { + readonly runId: string; + /** Current run row, or `undefined` if it has been purged. */ + status(): WorkflowRun | undefined; + /** The run's step nodes. */ + nodes(): WorkflowNode[]; + /** Resolve once the run reaches a terminal state (or reject on timeout). */ + wait(options?: WorkflowWaitOptions): Promise; +} diff --git a/sdks/node/test/dashboard.test.ts b/sdks/node/test/dashboard.test.ts index 673eb6fa..8f3f00c8 100644 --- a/sdks/node/test/dashboard.test.ts +++ b/sdks/node/test/dashboard.test.ts @@ -147,3 +147,58 @@ it("aggregates metrics after a job completes", async () => { srv.close(); } }); + +it("serves workflow runs, detail, dag, and children", async () => { + const db = join(mkdtempSync(join(tmpdir(), "taskito-dash-")), "q.db"); + const queue = new Queue({ dbPath: db }); + queue.task("a", () => 1); + queue.task("b", () => 2); + const handle = queue.workflows + .define("etl") + .step("a", "a") + .step("b", "b", { after: "a" }) + .submit(); + const worker: Worker = queue.runWorker(); + const srv = serveDashboard(queue, { port: 0, staticDir }); + await once(srv, "listening"); + const { port } = srv.address() as AddressInfo; + const root = `http://127.0.0.1:${port}/api/workflows/runs`; + + try { + await handle.wait({ timeoutMs: 8000 }); + + const list = (await (await fetch(root)).json()) as { + runs: Array<{ id: string; state: string }>; + }; + expect(list.runs.find((r) => r.id === handle.runId)?.state).toBe("completed"); + + const detail = (await (await fetch(`${root}/${handle.runId}`)).json()) as { + run: { state: string }; + nodes: Array<{ node_name: string; status: string }>; + }; + expect(detail.run.state).toBe("completed"); + expect(detail.nodes.map((n) => n.node_name).sort()).toEqual(["a", "b"]); + + const dag = (await (await fetch(`${root}/${handle.runId}/dag`)).json()) as { dag: string }; + const parsed = JSON.parse(dag.dag) as { + edges: Array<{ from: string; to: string; weight: number }>; + nodes: Array<{ name: string; node_name: string; status: string; id: string; deps: string[] }>; + }; + expect(parsed.edges).toEqual([{ from: "a", to: "b", weight: 1.0 }]); + // The SPA's DAG visualizer reads per-node deps / status / id, so the handler + // enriches the raw graph with live status, edges-as-deps, and job-id links. + const b = parsed.nodes.find((n) => n.name === "b"); + expect(b?.deps).toEqual(["a"]); + expect(b?.status).toBe("completed"); + expect(b?.id).not.toBe("b"); // resolved to the job id, not the node name + expect(parsed.nodes.find((n) => n.name === "a")?.deps).toEqual([]); + + const children = (await (await fetch(`${root}/${handle.runId}/children`)).json()) as { + children: unknown[]; + }; + expect(children.children).toEqual([]); + } finally { + worker.stop(); + srv.close(); + } +}); diff --git a/sdks/node/test/locks.test.ts b/sdks/node/test/locks.test.ts new file mode 100644 index 00000000..0f9b3875 --- /dev/null +++ b/sdks/node/test/locks.test.ts @@ -0,0 +1,82 @@ +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { expect, it } from "vitest"; +import { LockNotAcquiredError, Queue } from "../src/index"; + +function freshQueue(): Queue { + const dbPath = join(mkdtempSync(join(tmpdir(), "taskito-node-lock-")), "queue.db"); + return new Queue({ dbPath }); +} + +it("grants the lock to one owner at a time", () => { + const queue = freshQueue(); + const a = queue.lock("resource", { autoExtend: false }); + const b = queue.lock("resource", { autoExtend: false }); + + expect(a.acquire()).toBe(true); + expect(b.acquire()).toBe(false); + expect(queue.lock("resource").info()?.ownerId).toBe(a.ownerId); + + expect(a.release()).toBe(true); + expect(b.acquire()).toBe(true); + b.release(); +}); + +it("runs withLock then releases", async () => { + const queue = freshQueue(); + let ran = false; + const result = await queue.withLock( + "job", + () => { + ran = true; + expect(queue.lock("job").info()).toBeDefined(); + return 42; + }, + { autoExtend: false }, + ); + + expect(ran).toBe(true); + expect(result).toBe(42); + expect(queue.lock("job").info()).toBeUndefined(); +}); + +it("rejects withLock when the lock is held", async () => { + const queue = freshQueue(); + const holder = queue.lock("busy", { autoExtend: false }); + holder.acquire(); + + await expect(queue.withLock("busy", () => 1, { autoExtend: false })).rejects.toBeInstanceOf( + LockNotAcquiredError, + ); + holder.release(); +}); + +it("extends the expiry", () => { + const queue = freshQueue(); + const lock = queue.lock("ext", { ttlMs: 1000, autoExtend: false }); + lock.acquire(); + + const before = queue.lock("ext").info()?.expiresAt ?? 0; + expect(lock.extend(60_000)).toBe(true); + const after = queue.lock("ext").info()?.expiresAt ?? 0; + expect(after).toBeGreaterThan(before); + lock.release(); +}); + +it("releases via Symbol.dispose", () => { + const queue = freshQueue(); + const lock = queue.lock("disp", { autoExtend: false }); + lock.acquire(); + expect(queue.lock("disp").info()).toBeDefined(); + + lock[Symbol.dispose](); + expect(queue.lock("disp").info()).toBeUndefined(); +}); + +it("rejects a non-positive or non-finite ttl at construction", () => { + const queue = freshQueue(); + expect(() => queue.lock("bad", { ttlMs: 0 })).toThrow(RangeError); + expect(() => queue.lock("bad", { ttlMs: -5 })).toThrow(RangeError); + expect(() => queue.lock("bad", { ttlMs: Number.NaN })).toThrow(RangeError); +}); diff --git a/sdks/node/test/logger.test.ts b/sdks/node/test/logger.test.ts new file mode 100644 index 00000000..d5078d42 --- /dev/null +++ b/sdks/node/test/logger.test.ts @@ -0,0 +1,45 @@ +import { beforeEach, expect, it } from "vitest"; +import { createLogger, setLogLevel, setLogSink } from "../src/index"; + +let lines: string[]; + +beforeEach(() => { + lines = []; + setLogSink((_level, line) => lines.push(line)); + setLogLevel("debug"); +}); + +it("emits namespaced, leveled lines", () => { + createLogger("worker").info("hello"); + expect(lines).toHaveLength(1); + expect(lines[0]).toContain("[taskito:worker]"); + expect(lines[0]).toContain("INFO"); + expect(lines[0]).toContain("hello"); +}); + +it("drops messages below the threshold without evaluating the thunk", () => { + setLogLevel("warn"); + let built = false; + createLogger("x").debug(() => { + built = true; + return "expensive"; + }); + expect(built).toBe(false); + expect(lines).toHaveLength(0); +}); + +it("nests child namespaces", () => { + createLogger("a").child("b").error("boom"); + expect(lines[0]).toContain("[taskito:a:b]"); +}); + +it("serializes Error meta", () => { + createLogger("e").warn("failed", new Error("kaboom")); + expect(lines[0]).toContain("kaboom"); +}); + +it("silences all output at the silent level", () => { + setLogLevel("silent"); + createLogger("s").error("nope"); + expect(lines).toHaveLength(0); +}); diff --git a/sdks/node/test/mesh.test.ts b/sdks/node/test/mesh.test.ts new file mode 100644 index 00000000..155fcf3d --- /dev/null +++ b/sdks/node/test/mesh.test.ts @@ -0,0 +1,30 @@ +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, expect, it } from "vitest"; +import { Queue, type Worker } from "../src/index"; + +let worker: Worker | undefined; + +afterEach(() => { + worker?.stop(); + worker = undefined; +}); + +// A standalone mesh node (no seeds) still routes its own jobs through the bridge: +// scheduler -> affinity-sorted local deque -> dispatcher. Stealing is a no-op +// without peers, so a job must complete exactly as on the plain path. +it("processes jobs through the mesh bridge on a single node", async () => { + const dbPath = join(mkdtempSync(join(tmpdir(), "taskito-node-mesh-")), "queue.db"); + const queue = new Queue({ dbPath }); + queue.task("add", (a: number, b: number) => a + b); + + const id = queue.enqueue("add", [2, 3]); + worker = queue.runWorker({ + queues: ["default"], + mesh: { port: 17_946, seeds: [], steal: true }, + }); + + const result = await queue.result(id, { timeoutMs: 8000 }); + expect(result).toBe(5); +}); diff --git a/sdks/node/test/periodic.test.ts b/sdks/node/test/periodic.test.ts new file mode 100644 index 00000000..abafe584 --- /dev/null +++ b/sdks/node/test/periodic.test.ts @@ -0,0 +1,48 @@ +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, expect, it } from "vitest"; +import { Queue, type Worker } from "../src/index"; + +let worker: Worker | undefined; + +afterEach(() => { + worker?.stop(); + worker = undefined; +}); + +function freshQueue(): Queue { + const dbPath = join(mkdtempSync(join(tmpdir(), "taskito-node-cron-")), "queue.db"); + return new Queue({ dbPath }); +} + +// The Rust `cron` crate uses 6/7-field expressions (seconds first): +// `sec min hour day-of-month month day-of-week [year]`. +it("registers a cron task and returns its next fire time", () => { + const queue = freshQueue(); + queue.task("beat", () => null); + + const now = Date.now(); + const next = queue.registerPeriodic("heartbeat", "beat", "0 */5 * * * *", { args: [] }); + expect(next).toBeGreaterThan(now); + + // Re-register (upsert) the same name with a different schedule. + const next2 = queue.registerPeriodic("heartbeat", "beat", "0 0 * * * *"); + expect(next2).toBeGreaterThan(now); +}); + +it("rejects an invalid cron expression", () => { + const queue = freshQueue(); + queue.task("beat", () => null); + expect(() => queue.registerPeriodic("bad", "beat", "not a cron")).toThrow(); +}); + +it("accepts a circuit-breaker config and still runs the task", async () => { + const queue = freshQueue(); + queue.task("guarded", (x: number) => x * 2, { + circuitBreaker: { threshold: 3, windowMs: 60_000, cooldownMs: 60_000 }, + }); + const id = queue.enqueue("guarded", [21]); + worker = queue.runWorker(); + expect(await queue.result(id, { timeoutMs: 8000 })).toBe(42); +}); diff --git a/sdks/node/test/workflows-fanout.test.ts b/sdks/node/test/workflows-fanout.test.ts new file mode 100644 index 00000000..7764dfc2 --- /dev/null +++ b/sdks/node/test/workflows-fanout.test.ts @@ -0,0 +1,154 @@ +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, expect, it } from "vitest"; +import { Queue, type Worker } from "../src/index"; + +let worker: Worker | undefined; + +afterEach(() => { + worker?.stop(); + worker = undefined; +}); + +function freshQueue(): Queue { + const dbPath = join(mkdtempSync(join(tmpdir(), "taskito-node-fanout-")), "queue.db"); + return new Queue({ dbPath }); +} + +/** Map a finished run's nodes by name for compact assertions. */ +function nodesByName(handle: { + nodes: () => T[]; +}): Record { + return Object.fromEntries(handle.nodes().map((n) => [n.nodeName, n])); +} + +it("fans out over a producer's items and collects them in a fan-in", async () => { + const queue = freshQueue(); + const processed: number[] = []; + let combined: number[] | undefined; + + queue.task("source", () => [1, 2, 3]); + queue.task("double", (n: number) => { + processed.push(n); + return n * 2; + }); + queue.task("sum", (results: number[]) => { + combined = results; + return results.reduce((acc, x) => acc + x, 0); + }); + + const handle = queue.workflows + .define("fanout-etl") + .step("source", "source") + .fanOut("process", { after: "source", task: "double", itemsFrom: "source" }) + .fanIn("collect", { after: "process", task: "sum" }) + .submit(); + + worker = queue.runWorker({ queues: ["default"] }); + const run = await handle.wait({ timeoutMs: 10_000 }); + + expect(run.state).toBe("completed"); + expect(processed.slice().sort((a, b) => a - b)).toEqual([1, 2, 3]); + expect(combined?.slice().sort((a, b) => a - b)).toEqual([2, 4, 6]); + + const nodes = nodesByName(handle); + expect(nodes.process?.status).toBe("completed"); + expect(nodes.process?.fanOutCount).toBe(3); + expect(nodes.collect?.status).toBe("completed"); + // One child node was created per item. + expect( + ["process[0]", "process[1]", "process[2]"].every((n) => nodes[n]?.status === "completed"), + ).toBe(true); +}); + +it("fails the run and skips the fan-in when a fan-out child dead-letters", async () => { + const queue = freshQueue(); + queue.task("source", () => [1, 2, 3]); + queue.task("maybeBoom", (n: number) => { + if (n === 2) { + throw new Error("boom"); + } + return n; + }); + queue.task("collect", (r: number[]) => r); + + const handle = queue.workflows + .define("fanout-fail") + .step("source", "source") + .fanOut("process", { after: "source", task: "maybeBoom", itemsFrom: "source", maxRetries: 0 }) + .fanIn("collect", { after: "process", task: "collect" }) + .submit(); + + worker = queue.runWorker({ queues: ["default"] }); + const run = await handle.wait({ timeoutMs: 10_000 }); + + expect(run.state).toBe("failed"); + const nodes = nodesByName(handle); + expect(nodes.process?.status).toBe("failed"); + expect(nodes.collect?.status).toBe("skipped"); +}); + +it("completes a fan-out over an empty list and runs the fan-in with []", async () => { + const queue = freshQueue(); + let combined: unknown = "unset"; + + queue.task("source", () => []); + queue.task("noop", (n: number) => n); + queue.task("collect", (r: unknown[]) => { + combined = r; + return r.length; + }); + + const handle = queue.workflows + .define("fanout-empty") + .step("source", "source") + .fanOut("process", { after: "source", task: "noop", itemsFrom: "source" }) + .fanIn("collect", { after: "process", task: "collect" }) + .submit(); + + worker = queue.runWorker({ queues: ["default"] }); + const run = await handle.wait({ timeoutMs: 8_000 }); + + expect(run.state).toBe("completed"); + expect(combined).toEqual([]); + const nodes = nodesByName(handle); + expect(nodes.process?.status).toBe("completed"); + expect(nodes.process?.fanOutCount).toBe(0); + expect(nodes.collect?.status).toBe("completed"); +}); + +it("runs a fan-out without a fan-in collector", async () => { + const queue = freshQueue(); + const seen: number[] = []; + + queue.task("source", () => [10, 20]); + queue.task("handle", (n: number) => { + seen.push(n); + return n; + }); + + const handle = queue.workflows + .define("fanout-no-collect") + .step("source", "source") + .fanOut("process", { after: "source", task: "handle", itemsFrom: "source" }) + .submit(); + + worker = queue.runWorker({ queues: ["default"] }); + const run = await handle.wait({ timeoutMs: 8_000 }); + + expect(run.state).toBe("completed"); + expect(seen.slice().sort((a, b) => a - b)).toEqual([10, 20]); + expect(nodesByName(handle).process?.fanOutCount).toBe(2); +}); + +it("rejects a fan-in whose target is not a fan-out step", () => { + const queue = freshQueue(); + expect(() => + queue.workflows + .define("bad-fanin") + .step("plain", "noop") + .fanIn("collect", { after: "plain", task: "sum" }) + .build(), + ).toThrow(/must target a fan-out step/); +}); diff --git a/sdks/node/test/workflows.test.ts b/sdks/node/test/workflows.test.ts new file mode 100644 index 00000000..13266b73 --- /dev/null +++ b/sdks/node/test/workflows.test.ts @@ -0,0 +1,81 @@ +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, expect, it } from "vitest"; +import { Queue, type Worker } from "../src/index"; + +let worker: Worker | undefined; + +afterEach(() => { + worker?.stop(); + worker = undefined; +}); + +function freshQueue(): Queue { + const dbPath = join(mkdtempSync(join(tmpdir(), "taskito-node-wf-")), "queue.db"); + return new Queue({ dbPath }); +} + +it("runs a linear DAG to completion in dependency order", async () => { + const queue = freshQueue(); + const ran: string[] = []; + queue.task("extract", () => { + ran.push("extract"); + return { rows: 3 }; + }); + queue.task("transform", () => { + ran.push("transform"); + return "done"; + }); + + const handle = queue.workflows + .define("etl") + .step("extract", "extract") + .step("transform", "transform", { after: "extract" }) + .submit(); + + worker = queue.runWorker({ queues: ["default"] }); + const run = await handle.wait({ timeoutMs: 8000 }); + + expect(run.state).toBe("completed"); + expect(ran).toEqual(["extract", "transform"]); + const status = Object.fromEntries(handle.nodes().map((n) => [n.nodeName, n.status])); + expect(status).toEqual({ extract: "completed", transform: "completed" }); +}); + +it("fails the run and skips downstream steps when a step dead-letters", async () => { + const queue = freshQueue(); + queue.task("ok", () => 1); + queue.task("boom", () => { + throw new Error("kaboom"); + }); + queue.task("after", () => 2); + + const handle = queue.workflows + .define("failing") + .step("a", "ok") + .step("b", "boom", { after: "a", maxRetries: 0 }) + .step("c", "after", { after: "b" }) + .submit(); + + worker = queue.runWorker(); + const run = await handle.wait({ timeoutMs: 8000 }); + + expect(run.state).toBe("failed"); + const status = Object.fromEntries(handle.nodes().map((n) => [n.nodeName, n.status])); + expect(status.a).toBe("completed"); + expect(status.b).toBe("failed"); + expect(status.c).toBe("skipped"); +}); + +it("lists submitted runs and filters by state", async () => { + const queue = freshQueue(); + queue.task("noop", () => null); + + const handle = queue.workflows.define("listme").step("only", "noop").submit(); + worker = queue.runWorker(); + await handle.wait({ timeoutMs: 8000 }); + + const completed = queue.workflows.list({ state: "completed" }); + expect(completed.map((r) => r.id)).toContain(handle.runId); +});