diff --git a/.github/workflows/ci-rust.yml b/.github/workflows/ci-rust.yml index bfb7a317..6c190519 100644 --- a/.github/workflows/ci-rust.yml +++ b/.github/workflows/ci-rust.yml @@ -128,3 +128,71 @@ jobs: run: LD_LIBRARY_PATH="$pythonLocation/lib" cargo test --workspace --features redis,workflows env: TASKITO_REDIS_TEST_URL: redis://localhost:6379/15 + + # Everything a crates.io release of taskito-core depends on: the declared + # MSRV actually builds, the rustdoc is warning-free, the package passes a + # publish dry-run, and (once a baseline exists on crates.io) the public API + # hasn't broken semver. + publish-readiness: + name: Publish Readiness (taskito-core) + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v7 + with: + persist-credentials: false + + - name: Read MSRV from workspace manifest + id: msrv + run: | + msrv="$(grep -m1 '^rust-version' Cargo.toml | sed 's/.*"\(.*\)"/\1/')" + echo "version=$msrv" >> "$GITHUB_OUTPUT" + + - name: Install MSRV toolchain + uses: dtolnay/rust-toolchain@master + with: + toolchain: ${{ steps.msrv.outputs.version }} + + - name: Check core builds at the MSRV + run: cargo +${{ steps.msrv.outputs.version }} check -p taskito-core --all-features + + - name: Install stable toolchain + uses: dtolnay/rust-toolchain@stable + + - name: Rustdoc is warning-free + run: RUSTDOCFLAGS="-D warnings" cargo doc -p taskito-core --no-deps --all-features + + - name: Package passes a publish dry-run + run: cargo publish --dry-run -p taskito-core + + # No baseline exists before the first publish; the check is skipped until + # then, and a breaking change fails CI once one does. + - name: Check for a crates.io baseline + id: baseline + run: | + # Probe the sparse index (index.crates.io), not the REST API: it is + # the registry's sanctioned programmatic path — a static CDN in + # Cargo's own index format, no rate limits, no User-Agent policy + # (the API 403s UA-less requests). Index path scheme for names of + # 4+ chars: /{first-two}/{next-two}/{name}, lowercase. + crate="taskito-core" + prefix="${crate:0:2}/${crate:2:2}" + # Fail closed: only an explicit 404 means "no baseline" — a transient + # network error must not silently skip the semver check. Retries + # absorb CDN blips before that verdict. + if ! status="$(curl -sS -o /dev/null -w '%{http_code}' \ + --retry 3 --retry-connrefused \ + "https://index.crates.io/${prefix}/${crate}")"; then + echo "Failed to query the crates.io sparse index" >&2 + exit 1 + fi + case "$status" in + 200) echo "exists=true" >> "$GITHUB_OUTPUT" ;; + 404) echo "exists=false" >> "$GITHUB_OUTPUT" ;; + *) echo "Unexpected sparse-index response: HTTP $status" >&2; exit 1 ;; + esac + + - name: Semver check against crates.io baseline + if: steps.baseline.outputs.exists == 'true' + uses: obi1kenobi/cargo-semver-checks-action@v2 + with: + package: taskito-core diff --git a/Cargo.toml b/Cargo.toml index faa5c2e1..5cfd9dde 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,6 +2,16 @@ members = ["crates/taskito-core", "crates/taskito-python", "crates/taskito-node", "crates/taskito-java", "crates/taskito-workflows", "crates/taskito-mesh", "crates/taskito-tui"] resolver = "2" +# Shared crate metadata — every member inherits via `x.workspace = true`, so a +# release bumps `version` in exactly one place. `rust-version` is verified by +# the CI msrv job; sea-query 1.x pins the floor at 1.88. +[workspace.package] +version = "0.20.0" +edition = "2021" +license = "MIT" +repository = "https://github.com/ByteVeda/taskito" +rust-version = "1.88" + [workspace.dependencies] diesel = { version = "2.2", features = ["sqlite", "r2d2", "returning_clauses_for_sqlite_3_35"] } libsqlite3-sys = { version = "0.37", features = ["bundled"] } diff --git a/crates/taskito-core/Cargo.toml b/crates/taskito-core/Cargo.toml index c7340e1e..64dea84c 100644 --- a/crates/taskito-core/Cargo.toml +++ b/crates/taskito-core/Cargo.toml @@ -1,7 +1,14 @@ [package] name = "taskito-core" -version = "0.20.0" -edition = "2021" +version.workspace = true +edition.workspace = true +license.workspace = true +repository.workspace = true +rust-version.workspace = true +description = "Embeddable task queue: SQLite/Postgres/Redis storage, scheduler with retries, rate limits, and workers" +keywords = ["task-queue", "jobs", "scheduler", "background", "worker"] +categories = ["asynchronous", "database"] +readme = "README.md" [features] default = [] diff --git a/crates/taskito-core/LICENSE b/crates/taskito-core/LICENSE new file mode 100644 index 00000000..f75a1e74 --- /dev/null +++ b/crates/taskito-core/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Pratyush Sharma + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/crates/taskito-core/README.md b/crates/taskito-core/README.md new file mode 100644 index 00000000..881ee724 --- /dev/null +++ b/crates/taskito-core/README.md @@ -0,0 +1,83 @@ +# taskito-core + +Embeddable task queue for Rust: durable jobs on SQLite (default), PostgreSQL, +or Redis, a scheduler with retries, rate limits, circuit breakers, cron +periodics, and pub/sub fan-out — and a native worker that runs your handlers. + +This crate is the engine behind the Taskito SDKs; it works standalone in any +Rust program. + +## Quick start + +```rust,no_run +use taskito_core::{ + now_millis, Job, NewJob, SqliteStorage, Storage, StorageBackend, Worker, +}; + +fn main() -> taskito_core::Result<()> { + let storage = StorageBackend::Sqlite(SqliteStorage::new("taskito.db")?); + + // A worker executes registered handlers for dequeued jobs. + let handle = Worker::new(storage.clone()) + .num_workers(4) + .register("greet", |job: &Job| { + println!("hello, {}!", String::from_utf8_lossy(&job.payload)); + Ok(None) + }) + .spawn()?; + + // Producers enqueue jobs — from this process or any other. + storage.enqueue(NewJob { + queue: "default".to_string(), + task_name: "greet".to_string(), + payload: b"world".to_vec(), + priority: 0, + scheduled_at: now_millis(), + max_retries: 3, + timeout_ms: 30_000, + unique_key: None, + metadata: None, + notes: None, + depends_on: vec![], + expires_at: None, + result_ttl_ms: None, + namespace: None, + })?; + + std::thread::sleep(std::time::Duration::from_millis(500)); + handle.shutdown() +} +``` + +Async handlers work too — `register_async` spawns the future on the worker's +runtime. See `examples/hello.rs` for a runnable version. + +## Storage backends + +| Backend | Feature | Notes | +|---|---|---| +| SQLite | (default) | Zero-config, single file, bundled `libsqlite3` | +| PostgreSQL | `postgres` | `SKIP LOCKED` dequeue, `LISTEN/NOTIFY` push dispatch | +| Redis | `redis` | JSON blobs + sorted-set indexes, Lua-atomic claims | + +All three pass one shared contract test suite; behavior is identical up to +backend-documented differences. + +## What's in the box + +- **Jobs**: priorities, delays, per-job timeouts, uniqueness/idempotency keys, + dependencies, expiry, result TTLs, namespaces. +- **Scheduler**: adaptive polling (or event-driven wakeups via the + `push-dispatch` feature), batch claiming, bounded in-flight dispatch, + stale-job reaping, retention with per-table windows. +- **Resilience**: exponential-backoff retries, per-task rate limits and retry + budgets, circuit breakers, a dead-letter queue with replay. +- **Workers**: `Worker` builder + `TaskRegistry` for sync and async Rust + handlers, heartbeat/registry with elected cluster reaps. +- **Pub/sub**: topics with durable/ephemeral subscriptions and fan-out on + publish. +- **Periodic tasks**: cron expressions with optional time zones. + +## License + +MIT diff --git a/crates/taskito-core/src/error.rs b/crates/taskito-core/src/error.rs index 41c7b43b..2b1e0143 100644 --- a/crates/taskito-core/src/error.rs +++ b/crates/taskito-core/src/error.rs @@ -1,52 +1,69 @@ use thiserror::Error; +/// Every error the queue can produce. #[derive(Error, Debug)] pub enum QueueError { + /// A Diesel query or transaction failed. #[error("storage error: {0}")] Storage(#[from] diesel::result::Error), + /// The r2d2 connection pool could not hand out a connection. #[error("connection pool error: {0}")] Pool(#[from] diesel::r2d2::PoolError), + /// A Redis command or connection failed. #[cfg(feature = "redis")] #[error("redis error: {0}")] Redis(#[from] redis::RedisError), + /// JSON encoding or decoding failed. #[error("json error: {0}")] Json(#[from] serde_json::Error), + /// No job exists with the given id. #[error("job not found: {0}")] JobNotFound(String), + /// A job referenced a task name with no registered handler. #[error("task not registered: {0}")] TaskNotRegistered(String), + /// A payload or result could not be serialized/deserialized. #[error("serialization error: {0}")] Serialization(String), + /// A worker-side failure (spawn, dispatch, or pool error). #[error("worker error: {0}")] Worker(String), + /// A scheduler-side failure (dispatch, maintenance, or config). #[error("scheduler error: {0}")] Scheduler(String), + /// A queue/task rate limit rejected the operation. #[error("rate limit exceeded for: {0}")] RateLimitExceeded(String), + /// A job exceeded its execution timeout. #[error("job timed out: {0}")] Timeout(String), + /// Invalid or inconsistent configuration. #[error("config error: {0}")] Config(String), + /// A job dependency id does not exist. #[error("dependency not found: {0}")] DependencyNotFound(String), + /// A distributed lock was already held by another owner. #[error("lock not acquired: {0}")] LockNotAcquired(String), + /// Any other failure that fits no specific variant. #[error("{0}")] Other(String), } +/// Crate-wide result alias over [`QueueError`]. pub type Result = std::result::Result; diff --git a/crates/taskito-core/src/job.rs b/crates/taskito-core/src/job.rs index cbeacce9..27aa7df7 100644 --- a/crates/taskito-core/src/job.rs +++ b/crates/taskito-core/src/job.rs @@ -4,18 +4,26 @@ use uuid::Uuid; use crate::storage::models::{ArchivedJobRow, JobRow, NarrowArchivedJobRow, NarrowJobRow}; +/// Lifecycle state of a job. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[repr(i32)] pub enum JobStatus { + /// Waiting to be dequeued (or scheduled for the future). Pending = 0, + /// Claimed by a worker and executing. Running = 1, + /// Finished successfully. Complete = 2, + /// Failed; may still be retried. Failed = 3, + /// Exhausted retries and moved to the dead-letter queue. Dead = 4, + /// Cancelled before or during execution. Cancelled = 5, } impl JobStatus { + /// Decode the `#[repr(i32)]` discriminant; `None` for unknown values. pub fn from_i32(v: i32) -> Option { match v { 0 => Some(Self::Pending), @@ -28,6 +36,7 @@ impl JobStatus { } } + /// Lowercase display name (e.g. `"pending"`), as shown in listings. pub fn as_str(&self) -> &'static str { match self { Self::Pending => "pending", @@ -61,23 +70,41 @@ impl JobStatus { /// A job stored in the queue. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct Job { + /// Unique job id (UUIDv7). pub id: String, + /// Queue the job belongs to. pub queue: String, + /// Registered task the job runs. pub task_name: String, + /// Serialized task arguments (wire-envelope bytes). pub payload: Vec, + /// Current lifecycle state. pub status: JobStatus, + /// Dispatch priority; higher runs first. pub priority: i32, + /// Unix-millisecond time the job was enqueued. pub created_at: i64, + /// Unix-millisecond time the job becomes eligible to run. pub scheduled_at: i64, + /// Unix-millisecond time execution started, unset until dequeued. pub started_at: Option, + /// Unix-millisecond time the job reached a terminal state. pub completed_at: Option, + /// Retries attempted so far. pub retry_count: i32, + /// Maximum retries before dead-lettering. pub max_retries: i32, + /// Serialized result of a successful run. pub result: Option>, + /// Error message of the last failure (canonical JSON `TaskError` when structured). pub error: Option, + /// Execution timeout in milliseconds. pub timeout_ms: i64, + /// Deduplication key; at most one non-terminal job per key. pub unique_key: Option, + /// Task-reported progress percentage (0-100). pub progress: Option, + /// Pre-encoded JSON of free-form caller metadata, if any. pub metadata: Option, /// Structured, user-readable annotations attached to the job (canonical /// JSON object, ≤ 15 top-level fields). Validated at the binding @@ -85,9 +112,13 @@ pub struct Job { /// shell); stored as the already-encoded JSON string here. #[serde(default)] pub notes: Option, + /// True when cancellation was requested; workers poll this flag. pub cancel_requested: bool, + /// Unix-millisecond time after which a still-pending job expires. pub expires_at: Option, + /// How long the archived result is kept, in milliseconds. pub result_ttl_ms: Option, + /// Tenant namespace the job is scoped to. `None` = default namespace. pub namespace: Option, /// True when the job was enqueued with at least one dependency. Lets the /// scheduler skip the dependency lookup entirely for the common case. @@ -159,7 +190,7 @@ impl From for Job { } impl Job { - /// Assemble a [`Job`] from a blob-free [`NarrowJobRow`] plus `payload`/ + /// Assemble a [`Job`] from a blob-free `NarrowJobRow` plus `payload`/ /// `result` supplied by the caller. The narrow row carries every non-blob /// column; reap/listing paths that don't need the blobs pass empty ones. pub fn from_narrow(narrow: NarrowJobRow, payload: Vec, result: Option>) -> Self { @@ -191,7 +222,7 @@ impl Job { } } - /// Assemble a terminal [`Job`] from a blob-free [`NarrowArchivedJobRow`]. + /// Assemble a terminal [`Job`] from a blob-free `NarrowArchivedJobRow`. /// Listing paths use this so paging the archive never loads `payload`/ /// `result`; both come back empty (fetch the full job via `get_job`). /// Archived jobs are terminal and never re-dequeued, so `has_deps` is false. @@ -231,32 +262,51 @@ impl Job { /// /// [`Storage::complete_batch`]: crate::storage::Storage::complete_batch pub struct JobCompletion { + /// Id of the completed job. pub job_id: String, + /// Serialized result to archive. pub result: Option>, + /// Task that ran, recorded on the metric. pub task_name: String, + /// Wall-clock execution time in nanoseconds. pub wall_time_ns: i64, } /// Parameters for creating a new job. pub struct NewJob { + /// Queue to enqueue into. pub queue: String, + /// Registered task the job will run. pub task_name: String, + /// Serialized task arguments (wire-envelope bytes). pub payload: Vec, + /// Dispatch priority; higher runs first. pub priority: i32, + /// Unix-millisecond time the job becomes eligible to run. pub scheduled_at: i64, + /// Maximum retries before dead-lettering. pub max_retries: i32, + /// Execution timeout in milliseconds. pub timeout_ms: i64, + /// Deduplication key; at most one non-terminal job per key. pub unique_key: Option, + /// Pre-encoded JSON of free-form caller metadata, if any. pub metadata: Option, /// Pre-encoded canonical JSON object (≤ 15 fields). See [`Job::notes`]. pub notes: Option, + /// Ids of jobs that must complete before this one runs. pub depends_on: Vec, + /// Unix-millisecond time after which a still-pending job expires. pub expires_at: Option, + /// How long the archived result is kept, in milliseconds. pub result_ttl_ms: Option, + /// Tenant namespace the job is scoped to. `None` = default namespace. pub namespace: Option, } impl NewJob { + /// Materialize a [`Job`] with a fresh UUIDv7 id, `Pending` status, and + /// `created_at` = now. pub fn into_job(self) -> Job { let now = now_millis(); let has_deps = !self.depends_on.is_empty(); @@ -289,6 +339,7 @@ impl NewJob { } } +/// Current wall-clock time as Unix milliseconds. pub fn now_millis() -> i64 { SystemTime::now() .duration_since(UNIX_EPOCH) diff --git a/crates/taskito-core/src/lib.rs b/crates/taskito-core/src/lib.rs index dee2cab5..c91785e9 100644 --- a/crates/taskito-core/src/lib.rs +++ b/crates/taskito-core/src/lib.rs @@ -1,10 +1,20 @@ +#![doc = include_str!("../README.md")] +#![warn(missing_docs)] + +/// Error types: [`QueueError`] and the crate-wide [`Result`] alias. pub mod error; +/// Core job model: [`Job`], [`JobStatus`], [`NewJob`], [`JobCompletion`]. pub mod job; +/// Periodic (cron) task scheduling helpers. pub mod periodic; pub mod pubsub; +/// Resilience primitives: retry policies, rate limiting, circuit breakers, DLQ. pub mod resilience; +/// The [`Scheduler`]: job dispatch, retries, maintenance, retention. pub mod scheduler; +/// The [`Storage`] trait, backend implementations, and shared records. pub mod storage; +/// Native worker: task registry, dispatcher trait, worker runner. pub mod worker; // Primary public API — the types most consumers need. The crate root is the diff --git a/crates/taskito-core/src/pubsub.rs b/crates/taskito-core/src/pubsub.rs index 20bb5cbc..b58e7417 100644 --- a/crates/taskito-core/src/pubsub.rs +++ b/crates/taskito-core/src/pubsub.rs @@ -13,8 +13,11 @@ use crate::storage::Storage; /// nor the subscription row specifies a value. #[derive(Clone, Copy)] pub struct DeliveryDefaults { + /// Default dispatch priority for deliveries. pub priority: i32, + /// Default retry cap for deliveries. pub max_retries: i32, + /// Default execution timeout in milliseconds. pub timeout_ms: i64, } @@ -23,6 +26,7 @@ pub struct DeliveryDefaults { /// registry; delivery settings resolve per field as explicit publish override, /// then the subscription row's persisted setting, then the queue default. pub struct PublishRequest { + /// Topic to fan out on. pub topic: String, /// Wire-envelope payload bytes; every subscriber receives the same body. pub payload: Vec, @@ -31,17 +35,26 @@ pub struct PublishRequest { /// job `unique_key`: the jobs table's unique index is global, so reusing /// the raw key across the fan-out would dedup away all but one delivery. pub idempotency_key: Option, + /// Pre-encoded JSON of free-form caller metadata, copied to every delivery. pub metadata: Option, /// Pre-validated canonical notes JSON object (see `Job::notes`); the /// topic and subscription name are stamped in per delivery. pub notes: Option, + /// Priority override for every delivery. `None` = subscription, then default. pub priority: Option, + /// Unix-millisecond time the deliveries become eligible to run. pub scheduled_at: i64, + /// Retry-cap override for every delivery. `None` = subscription, then default. pub max_retries: Option, + /// Timeout override in milliseconds. `None` = subscription, then default. pub timeout_ms: Option, + /// Unix-millisecond expiry for still-pending deliveries. pub expires_at: Option, + /// How long each delivery's archived result is kept, in milliseconds. pub result_ttl_ms: Option, + /// Tenant namespace the deliveries are scoped to. `None` = default namespace. pub namespace: Option, + /// Queue-level fallback delivery settings. pub queue_defaults: DeliveryDefaults, } @@ -116,7 +129,7 @@ fn delivery_job(request: &PublishRequest, sub: &Subscription) -> NewJob { } } -/// Inverse of [`delivery_notes`]: pull `topic`/`subscription` back out of a +/// Inverse of `delivery_notes`: pull `topic`/`subscription` back out of a /// job's notes JSON when both are present. Feeds the indexed /// `jobs.topic`/`jobs.subscription_name` columns (and their `dead_letter` /// mirrors) at insert time, so backlog/lag aggregation runs off an index diff --git a/crates/taskito-core/src/resilience/circuit_breaker.rs b/crates/taskito-core/src/resilience/circuit_breaker.rs index 3794ad1e..c85d7ea4 100644 --- a/crates/taskito-core/src/resilience/circuit_breaker.rs +++ b/crates/taskito-core/src/resilience/circuit_breaker.rs @@ -7,12 +7,16 @@ use crate::storage::{Storage, StorageBackend}; #[derive(Debug, Clone, Copy, PartialEq, Eq)] #[repr(i32)] pub enum CircuitState { + /// Healthy: executions flow normally. Closed = 0, + /// Tripped: executions are rejected until the cooldown elapses. Open = 1, + /// Probing: a limited number of executions test recovery. HalfOpen = 2, } impl CircuitState { + /// Decode the `#[repr(i32)]` discriminant; unknown values map to `Closed`. pub fn from_i32(v: i32) -> Self { match v { 1 => Self::Open, @@ -21,6 +25,7 @@ impl CircuitState { } } + /// Lowercase display name (`"closed"`/`"open"`/`"half_open"`). pub fn as_str(&self) -> &'static str { match self { Self::Closed => "closed", @@ -33,8 +38,11 @@ impl CircuitState { /// Configuration for a task's circuit breaker. #[derive(Debug, Clone)] pub struct CircuitBreakerConfig { + /// Failure count that trips the breaker open. pub threshold: i32, + /// Failure-counting window in milliseconds. pub window_ms: i64, + /// Open-state cooldown in milliseconds before probing. pub cooldown_ms: i64, /// Number of probe requests allowed in HalfOpen state (default: 5). pub half_open_max_probes: i32, @@ -48,6 +56,7 @@ pub struct CircuitBreaker { } impl CircuitBreaker { + /// Build a breaker manager over `storage`. pub fn new(storage: StorageBackend) -> Self { Self { storage } } diff --git a/crates/taskito-core/src/resilience/dlq.rs b/crates/taskito-core/src/resilience/dlq.rs index 2331498a..ebe2c483 100644 --- a/crates/taskito-core/src/resilience/dlq.rs +++ b/crates/taskito-core/src/resilience/dlq.rs @@ -8,6 +8,7 @@ pub struct DeadLetterQueue { } impl DeadLetterQueue { + /// Build a DLQ manager over `storage`. pub fn new(storage: StorageBackend) -> Self { Self { storage } } diff --git a/crates/taskito-core/src/resilience/mod.rs b/crates/taskito-core/src/resilience/mod.rs index 4ed67714..a29ddede 100644 --- a/crates/taskito-core/src/resilience/mod.rs +++ b/crates/taskito-core/src/resilience/mod.rs @@ -1,4 +1,8 @@ +/// Per-task circuit breaker with half-open probing. pub mod circuit_breaker; +/// Dead-letter queue manager. pub mod dlq; +/// Storage-backed token-bucket rate limiter. pub mod rate_limiter; +/// Retry policies (exponential backoff, custom delays). pub mod retry; diff --git a/crates/taskito-core/src/resilience/rate_limiter.rs b/crates/taskito-core/src/resilience/rate_limiter.rs index 34df21b2..a5c7bf6d 100644 --- a/crates/taskito-core/src/resilience/rate_limiter.rs +++ b/crates/taskito-core/src/resilience/rate_limiter.rs @@ -9,8 +9,10 @@ pub struct RateLimiter { /// Parsed rate limit configuration (e.g., "100/m" → 100 tokens, refill 1.667/s). #[derive(Debug, Clone)] pub struct RateLimitConfig { + /// Bucket capacity. pub max_tokens: f64, - pub refill_rate: f64, // tokens per second + /// Tokens added per second. + pub refill_rate: f64, } impl RateLimitConfig { @@ -37,6 +39,7 @@ impl RateLimitConfig { } impl RateLimiter { + /// Build a rate limiter over `storage`. pub fn new(storage: StorageBackend) -> Self { Self { storage } } diff --git a/crates/taskito-core/src/scheduler/codel.rs b/crates/taskito-core/src/scheduler/codel.rs index 5b03e17a..7307569f 100644 --- a/crates/taskito-core/src/scheduler/codel.rs +++ b/crates/taskito-core/src/scheduler/codel.rs @@ -16,7 +16,9 @@ /// shedding begins. #[derive(Debug, Clone, Copy)] pub struct CodelConfig { + /// Acceptable steady-state sojourn in milliseconds. pub target_ms: i64, + /// Milliseconds the sojourn must stay above target before shedding. pub interval_ms: i64, } diff --git a/crates/taskito-core/src/scheduler/mod.rs b/crates/taskito-core/src/scheduler/mod.rs index be71a933..15931a7c 100644 --- a/crates/taskito-core/src/scheduler/mod.rs +++ b/crates/taskito-core/src/scheduler/mod.rs @@ -115,25 +115,43 @@ impl SchedulerConfig { /// Result of executing a job (sent back from worker threads). pub enum JobResult { + /// The task ran to completion. Success { + /// Id of the completed job. job_id: String, + /// Serialized result to archive. result: Option>, + /// Task that ran. task_name: String, + /// Wall-clock execution time in nanoseconds. wall_time_ns: i64, }, + /// The task raised an error or timed out. Failure { + /// Id of the failed job. job_id: String, + /// Error message (canonical JSON `TaskError` when structured). error: String, + /// Retries already attempted before this failure. retry_count: i32, + /// The job's retry cap. max_retries: i32, + /// Task that ran. task_name: String, + /// Wall-clock execution time in nanoseconds. wall_time_ns: i64, + /// Whether the failure is retryable (subject to the retry policy). should_retry: bool, + /// True when the failure was an execution timeout. timed_out: bool, }, + /// The task was cancelled while executing. Cancelled { + /// Id of the cancelled job. job_id: String, + /// Task that ran. task_name: String, + /// Wall-clock execution time in nanoseconds. wall_time_ns: i64, }, } @@ -154,28 +172,47 @@ impl JobResult { #[derive(Debug, Clone)] pub enum ResultOutcome { /// Task completed successfully. - Success { job_id: String, task_name: String }, + Success { + /// Id of the completed job. + job_id: String, + /// Task that ran. + task_name: String, + }, /// Task failed and will be retried. Retry { + /// Id of the failed job. job_id: String, + /// Task that ran. task_name: String, + /// Queue the job belongs to. queue: String, + /// Error message of this failure. error: String, + /// Retry count after this failure (1-based). retry_count: i32, + /// True when the failure was an execution timeout. timed_out: bool, }, /// Task exhausted retries and moved to the dead-letter queue. DeadLettered { + /// Id of the dead-lettered job. job_id: String, + /// Task that ran. task_name: String, + /// Queue the job belonged to. queue: String, + /// Final error message. error: String, + /// True when the final failure was an execution timeout. timed_out: bool, }, /// Task was cancelled during execution. Cancelled { + /// Id of the cancelled job. job_id: String, + /// Task that ran. task_name: String, + /// Queue the job belonged to. queue: String, }, } @@ -183,14 +220,18 @@ pub enum ResultOutcome { /// Per-task configuration for retry, rate limiting, and circuit breaker. #[derive(Debug, Clone)] pub struct TaskConfig { + /// Retry delays and cap applied to failed runs. pub retry_policy: RetryPolicy, + /// Dispatch rate limit for this task. `None` = unlimited. pub rate_limit: Option, + /// Circuit-breaker settings for this task. `None` = no breaker. pub circuit_breaker: Option, /// Cap on how fast this task may *retry*, across every job of the task. /// Nothing else catches sustained retry storms: the circuit breaker trips on /// hard failure, not on aggregate retry rate, and per-job `max_retries` is a /// per-job budget, not a rate. Exhausting it dead-letters instead of retrying. pub retry_budget: Option, + /// Cluster-wide cap on concurrently running jobs of this task. pub max_concurrent: Option, /// Cap on this task's share of *this worker's* in-flight slots, so one slow /// task cannot occupy the whole pool and starve every other task. Complements @@ -260,7 +301,9 @@ impl InFlight { /// Per-queue configuration for rate limiting and concurrency caps. #[derive(Debug, Clone)] pub struct QueueConfig { + /// Dispatch rate limit for the whole queue. `None` = unlimited. pub rate_limit: Option, + /// Cap on concurrently running jobs across the queue. `None` = uncapped. pub max_concurrent: Option, } @@ -323,6 +366,8 @@ struct TickCounters { } impl Scheduler { + /// Build a scheduler over `storage`, dispatching from `queues`, optionally + /// scoped to a tenant `namespace`. pub fn new( storage: StorageBackend, queues: Vec, @@ -362,6 +407,7 @@ impl Scheduler { } } + /// The storage backend this scheduler runs on. pub fn storage(&self) -> &StorageBackend { &self.storage } @@ -473,6 +519,7 @@ impl Scheduler { } } + /// Handle that stops [`Scheduler::run`] when notified. pub fn shutdown_handle(&self) -> Arc { self.shutdown.clone() } @@ -492,16 +539,18 @@ impl Scheduler { == 0 } + /// Set the rate-limit/concurrency policy for a queue. pub fn register_queue_config(&mut self, queue_name: String, config: QueueConfig) { self.queue_configs.insert(queue_name, config); } /// Enable opt-in CoDel load shedding on a queue. Orthogonal to - /// [`register_queue_config`] so the common (no-CoDel) path never pays for it. + /// [`Scheduler::register_queue_config`] so the common (no-CoDel) path never pays for it. pub fn register_queue_codel(&mut self, queue_name: String, config: codel::CodelConfig) { self.codel_configs.insert(queue_name, config); } + /// Set a task's resilience policy and register its circuit breaker, if any. pub fn register_task(&mut self, task_name: String, config: TaskConfig) { if let Some(ref cb_config) = config.circuit_breaker { if let Err(e) = self.circuit_breaker.register(&task_name, cb_config) { @@ -511,6 +560,7 @@ impl Scheduler { self.task_configs.insert(task_name, config); } + /// The circuit-breaker manager shared with this scheduler's storage. pub fn circuit_breaker(&self) -> &CircuitBreaker { &self.circuit_breaker } diff --git a/crates/taskito-core/src/scheduler/retention.rs b/crates/taskito-core/src/scheduler/retention.rs index 9a9985c7..76924c76 100644 --- a/crates/taskito-core/src/scheduler/retention.rs +++ b/crates/taskito-core/src/scheduler/retention.rs @@ -13,9 +13,13 @@ const DAY_MS: i64 = 86_400_000; /// logs churn hardest and are worth least (shortest); the DLQ is the only copy /// of a payload a human must act on, so it is deleted last (longest). pub const DEFAULT_ARCHIVED_JOBS_TTL_MS: i64 = 7 * DAY_MS; +/// Default `dead_letter` window: 30 days, in milliseconds. pub const DEFAULT_DEAD_LETTER_TTL_MS: i64 = 30 * DAY_MS; +/// Default `task_logs` window: 3 days, in milliseconds. pub const DEFAULT_TASK_LOGS_TTL_MS: i64 = 3 * DAY_MS; +/// Default `task_metrics` window: 7 days, in milliseconds. pub const DEFAULT_TASK_METRICS_TTL_MS: i64 = 7 * DAY_MS; +/// Default `job_errors` window: 7 days, in milliseconds. pub const DEFAULT_JOB_ERRORS_TTL_MS: i64 = 7 * DAY_MS; /// Retention window per history table, in milliseconds. `None` keeps a table diff --git a/crates/taskito-core/src/storage/cursor.rs b/crates/taskito-core/src/storage/cursor.rs index bc861f30..c6937f82 100644 --- a/crates/taskito-core/src/storage/cursor.rs +++ b/crates/taskito-core/src/storage/cursor.rs @@ -52,7 +52,9 @@ pub fn next_cursor(rows: &[T], limit: i64, key: impl Fn(&T) -> (i64, &str)) - /// next page (`None` when this page is the last). #[derive(Debug, Clone)] pub struct Page { + /// Rows in this page. pub items: Vec, + /// Opaque cursor for the next page; `None` when this page is the last. pub next_cursor: Option, } diff --git a/crates/taskito-core/src/storage/diesel_common/jobs.rs b/crates/taskito-core/src/storage/diesel_common/jobs.rs index de6de1e5..27d54b38 100644 --- a/crates/taskito-core/src/storage/diesel_common/jobs.rs +++ b/crates/taskito-core/src/storage/diesel_common/jobs.rs @@ -428,7 +428,7 @@ macro_rules! impl_diesel_job_ops { )) } - /// Batch variant of [`enqueue_unique`]: dedupe-insert many jobs in a + /// Batch variant of `enqueue_unique`: dedupe-insert many jobs in a /// single transaction instead of one transaction per job. Used by /// pub/sub keyed fan-out, where a publish creates one uniquely-keyed /// job per subscriber — previously N separate write transactions @@ -1114,7 +1114,7 @@ macro_rules! impl_diesel_job_ops { ) } - /// Keyset-paginated `list_jobs` — see [`list_jobs_filtered_after`]. + /// Keyset-paginated `list_jobs` — see `list_jobs_filtered_after`. pub fn list_jobs_after( &self, status: Option, diff --git a/crates/taskito-core/src/storage/migrate.rs b/crates/taskito-core/src/storage/migrate.rs index 519fff08..633078e1 100644 --- a/crates/taskito-core/src/storage/migrate.rs +++ b/crates/taskito-core/src/storage/migrate.rs @@ -31,7 +31,9 @@ use crate::job::now_millis; /// The SQL dialect a migration renders for. #[derive(Clone, Copy, PartialEq, Eq, Debug)] pub enum Backend { + /// SQLite dialect. Sqlite, + /// PostgreSQL dialect. Postgres, } diff --git a/crates/taskito-core/src/storage/mod.rs b/crates/taskito-core/src/storage/mod.rs index 8b69316d..198eca75 100644 --- a/crates/taskito-core/src/storage/mod.rs +++ b/crates/taskito-core/src/storage/mod.rs @@ -8,13 +8,17 @@ pub(crate) mod migrations; pub(crate) mod models; #[cfg(feature = "push-dispatch")] pub(crate) mod notify; +/// PostgreSQL storage backend (feature `postgres`). #[cfg(feature = "postgres")] pub mod postgres; pub mod records; +/// Redis storage backend (feature `redis`). #[cfg(feature = "redis")] pub mod redis_backend; pub(crate) mod schema; +/// SQLite storage backend (the default). pub mod sqlite; +/// The [`Storage`] trait every backend implements. pub mod traits; pub use traits::Storage; @@ -131,13 +135,20 @@ pub fn sweep_ephemeral_subscriptions( // ── Shared helper types ──────────────────────────────────────────────── +/// Per-status job counts for a queue (or the whole cluster). #[derive(Debug, Clone, Default)] pub struct QueueStats { + /// Jobs waiting to run. pub pending: i64, + /// Jobs currently executing. pub running: i64, + /// Jobs that finished successfully. pub completed: i64, + /// Jobs that failed terminally. pub failed: i64, + /// Jobs moved to the dead-letter queue. pub dead: i64, + /// Jobs that were cancelled. pub cancelled: i64, } @@ -148,14 +159,23 @@ pub struct QueueStats { /// indexes; they can never drift the way a maintained counter would. #[derive(Debug, Clone)] pub struct SubscriptionBacklogStats { + /// Topic the subscription listens on. pub topic: String, + /// Subscription name, unique per topic. pub subscription_name: String, + /// Task enqueued for each published message. pub task_name: String, + /// Queue deliveries are enqueued into. pub queue: String, + /// Whether deliveries are currently enabled. pub active: bool, + /// True for durable subscriptions that outlive their creator. pub durable: bool, + /// Deliveries waiting to run. pub pending: i64, + /// Deliveries currently executing. pub running: i64, + /// Deliveries in the dead-letter queue. pub dead: i64, /// Milliseconds since the oldest still-pending delivery was created. /// `None` when the subscription currently has no pending backlog. @@ -223,23 +243,41 @@ pub(crate) fn merge_backlog_stats( by_key.into_values().collect() } +/// A dead-lettered job: a copy of the original with its failure context, +/// enough to re-enqueue via `retry_dead`. #[derive(Debug, Clone)] pub struct DeadJob { + /// Unique id of the dead-letter entry. pub id: String, + /// Id of the job that dead-lettered. pub original_job_id: String, + /// Queue the job ran on. pub queue: String, + /// Task name of the job. pub task_name: String, + /// Serialized task arguments. pub payload: Vec, + /// Final error message (canonical JSON `TaskError` when structured). pub error: Option, + /// Retries the job had consumed when it dead-lettered. pub retry_count: i32, + /// Unix-millisecond time the job dead-lettered. pub failed_at: i64, + /// Pre-encoded JSON metadata blob, if any. pub metadata: Option, + /// Pre-encoded JSON structured notes, if any. pub notes: Option, + /// Job priority (higher runs first). pub priority: i32, + /// Retry cap the job was enqueued with. pub max_retries: i32, + /// Execution timeout in milliseconds. pub timeout_ms: i64, + /// Result retention in milliseconds. `None` = global default. pub result_ttl_ms: Option, + /// Namespace of the job, if any. pub namespace: Option, + /// Times this entry was auto-retried out of the DLQ. pub dlq_retry_count: i32, } @@ -267,7 +305,7 @@ impl From for DeadJob { } impl DeadJob { - /// Build a [`DeadJob`] from a blob-free [`NarrowDeadLetterRow`]. Listing + /// Build a [`DeadJob`] from a blob-free `NarrowDeadLetterRow`. Listing /// paths use this so paging the DLQ never loads the `payload` blob; it /// comes back empty and is only read when a single entry is requeued by id. pub fn from_narrow(row: models::NarrowDeadLetterRow) -> Self { @@ -975,9 +1013,12 @@ pub(crate) use impl_storage; /// Storage backend enum that dispatches to either SQLite or PostgreSQL. #[derive(Clone)] pub enum StorageBackend { + /// SQLite backend (the default). Sqlite(sqlite::SqliteStorage), + /// PostgreSQL backend (feature `postgres`). #[cfg(feature = "postgres")] Postgres(postgres::PostgresStorage), + /// Redis backend (feature `redis`). #[cfg(feature = "redis")] Redis(redis_backend::RedisStorage), } diff --git a/crates/taskito-core/src/storage/postgres/archival.rs b/crates/taskito-core/src/storage/postgres/archival.rs index 8116d964..0c9273a2 100644 --- a/crates/taskito-core/src/storage/postgres/archival.rs +++ b/crates/taskito-core/src/storage/postgres/archival.rs @@ -9,6 +9,10 @@ use crate::job::{Job, JobStatus}; crate::storage::diesel_common::impl_diesel_archival_ops!(PostgresStorage); impl PostgresStorage { + /// Move `Complete`/`Dead`/`Cancelled` jobs completed before `cutoff_ms` + /// (Unix milliseconds) from `jobs` into `archived_jobs`. Returns the count + /// archived. `Failed` jobs never appear here — `fail()` archives them + /// immediately. pub fn archive_old_jobs(&self, cutoff_ms: i64) -> Result { let mut conn = self.conn()?; diff --git a/crates/taskito-core/src/storage/postgres/dashboard_settings.rs b/crates/taskito-core/src/storage/postgres/dashboard_settings.rs index 15027fee..9a6d071c 100644 --- a/crates/taskito-core/src/storage/postgres/dashboard_settings.rs +++ b/crates/taskito-core/src/storage/postgres/dashboard_settings.rs @@ -9,6 +9,7 @@ use crate::error::Result; use crate::job::now_millis; impl PostgresStorage { + /// Fetch a single setting value by key, or `None` if unset. pub fn get_setting(&self, key: &str) -> Result> { let mut conn = self.conn()?; let row: Option = dashboard_settings::table @@ -18,6 +19,7 @@ impl PostgresStorage { Ok(row.map(|r| r.value)) } + /// Insert or update a setting. pub fn set_setting(&self, key: &str, value: &str) -> Result<()> { let mut conn = self.conn()?; let now = now_millis(); @@ -38,6 +40,7 @@ impl PostgresStorage { Ok(()) } + /// Delete a setting. Returns `true` if a row was removed. pub fn delete_setting(&self, key: &str) -> Result { let mut conn = self.conn()?; let deleted = @@ -46,6 +49,7 @@ impl PostgresStorage { Ok(deleted > 0) } + /// All settings as a key-to-value map. pub fn list_settings(&self) -> Result> { let mut conn = self.conn()?; let rows: Vec = dashboard_settings::table diff --git a/crates/taskito-core/src/storage/postgres/mod.rs b/crates/taskito-core/src/storage/postgres/mod.rs index 4632037c..d21921fd 100644 --- a/crates/taskito-core/src/storage/postgres/mod.rs +++ b/crates/taskito-core/src/storage/postgres/mod.rs @@ -116,6 +116,7 @@ impl PostgresStorage { &self.database_url } + /// Pooled connection with `search_path` set to this storage's schema. pub fn conn(&self) -> Result>> { let mut conn = self.pool.get()?; diesel::sql_query(format!( diff --git a/crates/taskito-core/src/storage/postgres/queue_state.rs b/crates/taskito-core/src/storage/postgres/queue_state.rs index 174fdcc7..81910cb2 100644 --- a/crates/taskito-core/src/storage/postgres/queue_state.rs +++ b/crates/taskito-core/src/storage/postgres/queue_state.rs @@ -7,6 +7,7 @@ use crate::error::Result; use crate::job::now_millis; impl PostgresStorage { + /// Pause a queue so no new jobs are dispatched from it. pub fn pause_queue(&self, queue_name: &str) -> Result<()> { let mut conn = self.conn()?; let now = now_millis(); @@ -27,6 +28,7 @@ impl PostgresStorage { Ok(()) } + /// Resume a paused queue. pub fn resume_queue(&self, queue_name: &str) -> Result<()> { let mut conn = self.conn()?; @@ -36,6 +38,7 @@ impl PostgresStorage { Ok(()) } + /// Names of all currently paused queues. pub fn list_paused_queues(&self) -> Result> { let mut conn = self.conn()?; diff --git a/crates/taskito-core/src/storage/postgres/rate_limits.rs b/crates/taskito-core/src/storage/postgres/rate_limits.rs index 64368e4a..ef25f7be 100644 --- a/crates/taskito-core/src/storage/postgres/rate_limits.rs +++ b/crates/taskito-core/src/storage/postgres/rate_limits.rs @@ -8,6 +8,7 @@ use crate::error::Result; use crate::job::now_millis; impl PostgresStorage { + /// Token-bucket state for a rate-limit key, if one exists. pub fn get_rate_limit(&self, key: &str) -> Result> { let mut conn = self.conn()?; @@ -20,6 +21,7 @@ impl PostgresStorage { Ok(row.map(Into::into)) } + /// Insert or replace a token-bucket state row. pub fn upsert_rate_limit(&self, state: &RateLimitState) -> Result<()> { let mut conn = self.conn()?; diff --git a/crates/taskito-core/src/storage/records.rs b/crates/taskito-core/src/storage/records.rs index c14c9670..5c82d6cc 100644 --- a/crates/taskito-core/src/storage/records.rs +++ b/crates/taskito-core/src/storage/records.rs @@ -11,49 +11,78 @@ use serde::{Deserialize, Serialize}; /// One recorded failure attempt for a job. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct JobError { + /// Unique id of this error record. pub id: String, + /// Id of the job that failed. pub job_id: String, + /// 1-based attempt number that produced this error. pub attempt: i32, + /// Error message (canonical JSON `TaskError` when structured). pub error: String, + /// Unix-millisecond time of the failure. pub failed_at: i64, } /// Token-bucket state for a rate-limit key. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct RateLimitState { + /// Rate-limit key (task or queue scoped). pub key: String, + /// Tokens currently available. pub tokens: f64, + /// Bucket capacity. pub max_tokens: f64, + /// Tokens added per second. pub refill_rate: f64, + /// Unix-millisecond time of the last refill. pub last_refill: i64, } /// A registered periodic (cron) task. #[derive(Debug, Clone)] pub struct PeriodicTask { + /// Unique schedule name. pub name: String, + /// Task to enqueue on each firing. pub task_name: String, + /// Cron expression driving the schedule. pub cron_expr: String, + /// Serialized positional arguments. pub args: Option>, + /// Serialized keyword arguments. pub kwargs: Option>, + /// Queue to enqueue into. pub queue: String, + /// Whether the schedule is active (false = paused). pub enabled: bool, + /// Unix-millisecond time of the last firing, unset until first run. pub last_run: Option, + /// Unix-millisecond time of the next firing. pub next_run: i64, + /// IANA timezone for cron evaluation. `None` = UTC. pub timezone: Option, } /// Registration payload for a periodic task. `last_run` starts unset. #[derive(Debug, Clone)] pub struct NewPeriodicTask { + /// Unique schedule name. pub name: String, + /// Task to enqueue on each firing. pub task_name: String, + /// Cron expression driving the schedule. pub cron_expr: String, + /// Serialized positional arguments. pub args: Option>, + /// Serialized keyword arguments. pub kwargs: Option>, + /// Queue to enqueue into. pub queue: String, + /// Whether the schedule starts active. pub enabled: bool, + /// Unix-millisecond time of the first firing. pub next_run: i64, + /// IANA timezone for cron evaluation. `None` = UTC. pub timezone: Option, } @@ -65,94 +94,151 @@ pub struct NewPeriodicTask { /// when its worker dies. #[derive(Debug, Clone)] pub struct Subscription { + /// Topic the subscription listens on. pub topic: String, + /// Subscription name, unique per topic. pub subscription_name: String, + /// Task enqueued for each published message. pub task_name: String, + /// Queue deliveries are enqueued into. pub queue: String, + /// Whether deliveries are currently enabled (false = paused). pub active: bool, + /// True for durable subscriptions that outlive their creator. pub durable: bool, + /// Owning worker id for ephemeral subscriptions; `None` = durable. pub owner_worker_id: Option, + /// Unix-millisecond registration time. pub created_at: i64, /// Per-subscription delivery settings persisted at registration so /// `publish_to_topic` applies them cross-process. `None` = queue default. pub priority: Option, + /// Per-delivery retry cap. `None` = queue default. pub max_retries: Option, + /// Per-delivery timeout in milliseconds. `None` = queue default. pub timeout_ms: Option, } /// Registration payload for a topic subscription. #[derive(Debug, Clone)] pub struct NewSubscription { + /// Topic the subscription listens on. pub topic: String, + /// Subscription name, unique per topic. pub subscription_name: String, + /// Task enqueued for each published message. pub task_name: String, + /// Queue deliveries are enqueued into. pub queue: String, + /// Whether deliveries start enabled. pub active: bool, + /// True for durable subscriptions that outlive their creator. pub durable: bool, + /// Owning worker id for ephemeral subscriptions; `None` = durable. pub owner_worker_id: Option, + /// Unix-millisecond registration time. pub created_at: i64, + /// Per-delivery priority override. `None` = queue default. pub priority: Option, + /// Per-delivery retry cap. `None` = queue default. pub max_retries: Option, + /// Per-delivery timeout in milliseconds. `None` = queue default. pub timeout_ms: Option, } /// One execution measurement for a task. #[derive(Debug, Clone)] pub struct TaskMetric { + /// Unique id of this metric record. pub id: String, + /// Task that was executed. pub task_name: String, + /// Job the measurement belongs to. pub job_id: String, + /// Wall-clock execution time in nanoseconds. pub wall_time_ns: i64, + /// Peak memory delta in bytes. pub memory_bytes: i64, + /// Whether the execution succeeded. pub succeeded: bool, + /// Unix-millisecond time the metric was recorded. pub recorded_at: i64, } /// One replay of a completed job, pairing original and replay outcomes. #[derive(Debug, Clone)] pub struct ReplayEntry { + /// Unique id of this replay record. pub id: String, + /// Id of the job that was replayed. pub original_job_id: String, + /// Id of the replay job. pub replay_job_id: String, + /// Unix-millisecond time of the replay. pub replayed_at: i64, + /// Serialized result of the original run. pub original_result: Option>, + /// Serialized result of the replay run. pub replay_result: Option>, + /// Error message of the original run, if it failed. pub original_error: Option, + /// Error message of the replay run, if it failed. pub replay_error: Option, } /// One structured log line emitted during task execution. #[derive(Debug, Clone)] pub struct TaskLogEntry { + /// Unique id of this log line (UUIDv7, doubles as a stream cursor). pub id: String, + /// Job the log line belongs to. pub job_id: String, + /// Task that emitted the line. pub task_name: String, + /// Log level (`debug`/`info`/`warning`/`error`). pub level: String, + /// Log message text. pub message: String, + /// Pre-encoded JSON of structured extra fields, if any. pub extra: Option, + /// Unix-millisecond time the line was logged. pub logged_at: i64, } /// Persisted circuit-breaker state for a task. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct CircuitBreakerState { + /// Task the breaker guards. pub task_name: String, + /// Current state: 0 = closed, 1 = open, 2 = half-open. pub state: i32, + /// Failures observed in the current window. pub failure_count: i32, + /// Unix-millisecond time of the most recent failure. pub last_failure_at: Option, + /// Unix-millisecond time the breaker opened. pub opened_at: Option, + /// Unix-millisecond time the breaker entered half-open. pub half_open_at: Option, + /// Failure count that trips the breaker open. pub threshold: i32, + /// Failure-counting window in milliseconds. pub window_ms: i64, + /// Open-state cooldown in milliseconds before probing. pub cooldown_ms: i64, + /// Maximum probe executions allowed while half-open. #[serde(default = "default_max_probes")] pub half_open_max_probes: i32, + /// Probe success ratio (0.0-1.0) required to close. #[serde(default = "default_success_rate")] pub half_open_success_rate: f64, + /// Probes dispatched in the current half-open round. #[serde(default)] pub half_open_probe_count: i32, + /// Probes that succeeded in the current half-open round. #[serde(default)] pub half_open_success_count: i32, + /// Probes that failed in the current half-open round. #[serde(default)] pub half_open_failure_count: i32, } @@ -168,25 +254,41 @@ fn default_success_rate() -> f64 { /// A registered worker as seen by the cluster registry. #[derive(Debug, Clone)] pub struct WorkerInfo { + /// Unique worker id. pub worker_id: String, + /// Unix-millisecond time of the last heartbeat. pub last_heartbeat: i64, + /// Comma-separated queue names the worker consumes. pub queues: String, + /// Worker status string (e.g. `active`, `offline`). pub status: String, + /// Pre-encoded JSON list of worker tags, if any. pub tags: Option, + /// Pre-encoded JSON list of resource names the worker provides. pub resources: Option, + /// Pre-encoded JSON of per-resource health, refreshed each heartbeat. pub resource_health: Option, + /// Worker thread count. pub threads: i32, + /// Unix-millisecond time the worker started. pub started_at: Option, + /// Host the worker runs on. pub hostname: Option, + /// OS process id of the worker. pub pid: Option, + /// Execution pool type (e.g. `thread`, `prefork`). pub pool_type: Option, } /// Holder and expiry of a distributed lock. #[derive(Debug, Clone)] pub struct LockInfo { + /// Lock name. pub lock_name: String, + /// Current holder's owner id. pub owner_id: String, + /// Unix-millisecond time the lock was acquired. pub acquired_at: i64, + /// Unix-millisecond time the lock expires. pub expires_at: i64, } diff --git a/crates/taskito-core/src/storage/redis_backend/archival.rs b/crates/taskito-core/src/storage/redis_backend/archival.rs index 8f8e5ff9..cf6de012 100644 --- a/crates/taskito-core/src/storage/redis_backend/archival.rs +++ b/crates/taskito-core/src/storage/redis_backend/archival.rs @@ -5,6 +5,10 @@ use crate::error::Result; use crate::job::{Job, JobStatus}; impl RedisStorage { + /// Move `Complete`/`Dead`/`Cancelled` jobs completed before `cutoff_ms` + /// (Unix milliseconds) out of every live index and into the archive. + /// Returns the count archived. `Failed` jobs never appear here — `fail()` + /// archives them immediately. pub fn archive_old_jobs(&self, cutoff_ms: i64) -> Result { let mut conn = self.conn()?; let mut count = 0u64; @@ -35,6 +39,7 @@ impl RedisStorage { Ok(count) } + /// Archived jobs, newest first, paginated. Rows are blob-free. pub fn list_archived(&self, limit: i64, offset: i64) -> Result> { let mut conn = self.conn()?; let archived_all = self.key(&["archived", "all"]); diff --git a/crates/taskito-core/src/storage/redis_backend/circuit_breakers.rs b/crates/taskito-core/src/storage/redis_backend/circuit_breakers.rs index 1479737b..afb20251 100644 --- a/crates/taskito-core/src/storage/redis_backend/circuit_breakers.rs +++ b/crates/taskito-core/src/storage/redis_backend/circuit_breakers.rs @@ -5,6 +5,7 @@ use crate::error::Result; use crate::storage::records::CircuitBreakerState; impl RedisStorage { + /// Persisted circuit-breaker state for a task, if one exists. pub fn get_circuit_breaker(&self, task_name: &str) -> Result> { let mut conn = self.conn()?; let cb_key = self.key(&["cb", task_name]); @@ -19,6 +20,7 @@ impl RedisStorage { } } + /// Insert or replace a task's circuit-breaker state. pub fn upsert_circuit_breaker(&self, row: &CircuitBreakerState) -> Result<()> { let mut conn = self.conn()?; let cb_key = self.key(&["cb", &row.task_name]); @@ -34,6 +36,7 @@ impl RedisStorage { Ok(()) } + /// Every persisted circuit-breaker state. pub fn list_circuit_breakers(&self) -> Result> { let mut conn = self.conn()?; let cb_all = self.key(&["cb", "all"]); diff --git a/crates/taskito-core/src/storage/redis_backend/dashboard_settings.rs b/crates/taskito-core/src/storage/redis_backend/dashboard_settings.rs index c2385078..81e7a25c 100644 --- a/crates/taskito-core/src/storage/redis_backend/dashboard_settings.rs +++ b/crates/taskito-core/src/storage/redis_backend/dashboard_settings.rs @@ -12,12 +12,14 @@ fn settings_key(storage: &RedisStorage) -> String { } impl RedisStorage { + /// Fetch a single setting value by key, or `None` if unset. pub fn get_setting(&self, key: &str) -> Result> { let mut conn = self.conn()?; let value: Option = conn.hget(settings_key(self), key).map_err(map_err)?; Ok(value) } + /// Insert or update a setting. pub fn set_setting(&self, key: &str, value: &str) -> Result<()> { let mut conn = self.conn()?; conn.hset::<_, _, _, ()>(settings_key(self), key, value) @@ -25,12 +27,14 @@ impl RedisStorage { Ok(()) } + /// Delete a setting. Returns `true` if an entry was removed. pub fn delete_setting(&self, key: &str) -> Result { let mut conn = self.conn()?; let removed: i64 = conn.hdel(settings_key(self), key).map_err(map_err)?; Ok(removed > 0) } + /// All settings as a key-to-value map. pub fn list_settings(&self) -> Result> { let mut conn = self.conn()?; let map: HashMap = conn.hgetall(settings_key(self)).map_err(map_err)?; diff --git a/crates/taskito-core/src/storage/redis_backend/dead_letter.rs b/crates/taskito-core/src/storage/redis_backend/dead_letter.rs index ddef9096..ff3728f4 100644 --- a/crates/taskito-core/src/storage/redis_backend/dead_letter.rs +++ b/crates/taskito-core/src/storage/redis_backend/dead_letter.rs @@ -53,6 +53,7 @@ impl From for DeadJob { } impl RedisStorage { + /// Move a job to the dead-letter queue and cascade-cancel its dependents. pub fn move_to_dlq(&self, job: &Job, error: &str, metadata: Option<&str>) -> Result<()> { let now = now_millis(); let dlq_id = uuid::Uuid::now_v7().to_string(); @@ -128,6 +129,7 @@ impl RedisStorage { Ok(()) } + /// Dead-letter entries, newest first, paginated. pub fn list_dead(&self, limit: i64, offset: i64) -> Result> { let mut conn = self.conn()?; let dlq_all = self.key(&["dlq", "all"]); @@ -180,6 +182,7 @@ impl RedisStorage { Ok(results) } + /// Dead-letter entries for one task, newest first, paginated. pub fn list_dead_by_task( &self, task_name: &str, @@ -223,6 +226,7 @@ impl RedisStorage { Ok(matches.into_iter().skip(offset).take(limit).collect()) } + /// Delete every dead-letter entry for a task. Returns the number removed. pub fn purge_dead_by_task(&self, task_name: &str) -> Result { let mut conn = self.conn()?; let dlq_all = self.key(&["dlq", "all"]); @@ -261,6 +265,8 @@ impl RedisStorage { Ok(to_delete.len() as u64) } + /// Re-enqueue a dead-letter entry as a fresh job, deleting the entry. + /// Returns the new job's id; `JobNotFound` if the entry is absent. pub fn retry_dead(&self, dead_id: &str) -> Result { let mut conn = self.conn()?; let dlq_key = self.key(&["dlq", dead_id]); @@ -327,6 +333,8 @@ impl RedisStorage { Ok(job.id) } + /// Purge dead-letter entries older than the cutoff. Returns the count + /// removed. pub fn purge_dead(&self, older_than_ms: i64) -> Result { let mut conn = self.conn()?; let dlq_all = self.key(&["dlq", "all"]); @@ -374,6 +382,7 @@ impl RedisStorage { Ok(total) } + /// Delete one dead-letter entry. Returns `false` when none matched. pub fn delete_dead(&self, dead_id: &str) -> Result { let mut conn = self.conn()?; let dlq_key = self.key(&["dlq", dead_id]); @@ -395,6 +404,8 @@ impl RedisStorage { Ok(true) } + /// Purge dead-letter entries by the global/per-entry TTL. Returns the + /// count removed. pub fn purge_dead_with_ttl(&self, global_cutoff_ms: Option) -> Result { let mut conn = self.conn()?; let dlq_all = self.key(&["dlq", "all"]); @@ -457,6 +468,7 @@ impl RedisStorage { Ok(total) } + /// Dead-letter entries eligible for automatic retry, bounded by `limit`. pub fn list_dead_for_retry( &self, cutoff_ms: i64, diff --git a/crates/taskito-core/src/storage/redis_backend/jobs/dequeue.rs b/crates/taskito-core/src/storage/redis_backend/jobs/dequeue.rs index a95174b4..96b37bde 100644 --- a/crates/taskito-core/src/storage/redis_backend/jobs/dequeue.rs +++ b/crates/taskito-core/src/storage/redis_backend/jobs/dequeue.rs @@ -54,6 +54,8 @@ impl RedisStorage { Ok(claimed == 1) } + /// Atomically claim the highest-priority ready job, moving it to `Running`. + /// Scans the queue's sorted set by score; each candidate is claimed Lua-atomically. pub fn dequeue( &self, queue_name: &str, @@ -163,6 +165,7 @@ impl RedisStorage { Ok(None) } + /// [`dequeue`](Self::dequeue) across several queues, checked in order. pub fn dequeue_from( &self, queues: &[String], diff --git a/crates/taskito-core/src/storage/redis_backend/jobs/enqueue.rs b/crates/taskito-core/src/storage/redis_backend/jobs/enqueue.rs index 1aeb7d49..363dec2d 100644 --- a/crates/taskito-core/src/storage/redis_backend/jobs/enqueue.rs +++ b/crates/taskito-core/src/storage/redis_backend/jobs/enqueue.rs @@ -43,6 +43,8 @@ impl RedisStorage { Ok(()) } + /// Insert a new job and return it, writing its JSON plus the status, + /// queue, and task index entries. pub fn enqueue(&self, new_job: NewJob) -> Result { let depends_on = new_job.depends_on.clone(); let job = new_job.into_job(); @@ -84,6 +86,7 @@ impl RedisStorage { Ok(job) } + /// Batch [`enqueue`](Self::enqueue): insert multiple jobs at once. pub fn enqueue_batch(&self, new_jobs: Vec) -> Result> { // Collect dependency lists before consuming new_jobs let dep_lists: Vec> = new_jobs.iter().map(|nj| nj.depends_on.clone()).collect(); @@ -133,7 +136,7 @@ impl RedisStorage { Ok(jobs) } - /// Batch variant of [`enqueue_unique`]. Redis has no database-wide write + /// Batch variant of `enqueue_unique`. Redis has no database-wide write /// lock (each op is atomic and cheap), so looping the single-delivery path /// is correct and avoids a bespoke multi-delivery Lua script — the batch /// API's real win is on the Diesel backends. Salted keys are distinct @@ -145,6 +148,8 @@ impl RedisStorage { .collect() } + /// Enqueue with `unique_key` deduplication: a Lua script atomically returns + /// the existing active job when a duplicate is found instead of inserting. pub fn enqueue_unique(&self, new_job: NewJob) -> Result { let mut conn = self.conn()?; diff --git a/crates/taskito-core/src/storage/redis_backend/jobs/errors.rs b/crates/taskito-core/src/storage/redis_backend/jobs/errors.rs index 708ee8c5..1c43818f 100644 --- a/crates/taskito-core/src/storage/redis_backend/jobs/errors.rs +++ b/crates/taskito-core/src/storage/redis_backend/jobs/errors.rs @@ -8,6 +8,7 @@ use crate::storage::records::JobError; use crate::storage::redis_backend::{map_err, RedisStorage}; impl RedisStorage { + /// Record one failed attempt's error for a job. pub fn record_error(&self, job_id: &str, attempt: i32, error: &str) -> Result<()> { let mut conn = self.conn()?; let id = uuid::Uuid::now_v7().to_string(); @@ -29,6 +30,7 @@ impl RedisStorage { Ok(()) } + /// All recorded errors for a job, ordered by attempt. pub fn get_job_errors(&self, job_id: &str) -> Result> { let mut conn = self.conn()?; let errors_key = self.key(&["job_errors", job_id]); @@ -43,6 +45,7 @@ impl RedisStorage { Ok(rows) } + /// Purge error records older than the cutoff. Returns the count removed. pub fn purge_job_errors(&self, older_than_ms: i64) -> Result { let mut conn = self.conn()?; // Scan for all job_errors keys diff --git a/crates/taskito-core/src/storage/redis_backend/jobs/maintenance.rs b/crates/taskito-core/src/storage/redis_backend/jobs/maintenance.rs index 10990b87..42826e65 100644 --- a/crates/taskito-core/src/storage/redis_backend/jobs/maintenance.rs +++ b/crates/taskito-core/src/storage/redis_backend/jobs/maintenance.rs @@ -17,6 +17,8 @@ impl RedisStorage { ]) } + /// Purge archived completed jobs older than the cutoff. Returns the count + /// removed. pub fn purge_completed(&self, older_than_ms: i64) -> Result { self.purge_completed_scan(|job| { job.completed_at @@ -234,6 +236,8 @@ impl RedisStorage { Ok(count) } + /// Running jobs that exceeded their timeout, for the scheduler to fail or + /// retry. pub fn reap_stale_jobs(&self, now: i64) -> Result> { let mut conn = self.conn()?; let status_key = self.key(&["jobs", "status", &(JobStatus::Running as i32).to_string()]); @@ -299,6 +303,8 @@ impl RedisStorage { Ok(orphaned) } + /// Cancel pending jobs whose `expires_at` has passed, archiving them as + /// `Cancelled`. Returns the count expired. pub fn expire_pending_jobs(&self, now: i64) -> Result { let mut conn = self.conn()?; let status_key = self.key(&["jobs", "status", &(JobStatus::Pending as i32).to_string()]); @@ -327,6 +333,7 @@ impl RedisStorage { Ok(count) } + /// Cancel every pending job in a queue. Returns the count cancelled. pub fn cancel_pending_by_queue(&self, queue: &str) -> Result { let mut conn = self.conn()?; let by_queue_key = self.key(&["jobs", "by_queue", queue]); @@ -354,6 +361,7 @@ impl RedisStorage { Ok(count) } + /// Cancel every pending job for a task. Returns the count cancelled. pub fn cancel_pending_by_task(&self, task_name: &str) -> Result { let mut conn = self.conn()?; let by_task_key = self.key(&["jobs", "by_task", task_name]); diff --git a/crates/taskito-core/src/storage/redis_backend/jobs/query.rs b/crates/taskito-core/src/storage/redis_backend/jobs/query.rs index 57b3366d..bcfaa025 100644 --- a/crates/taskito-core/src/storage/redis_backend/jobs/query.rs +++ b/crates/taskito-core/src/storage/redis_backend/jobs/query.rs @@ -8,6 +8,8 @@ use crate::storage::redis_backend::{map_err, strip_list_blobs, RedisStorage}; use crate::storage::QueueStats; impl RedisStorage { + /// List jobs by filter. Rows are blob-free — fetch the full job with + /// [`get_job`](Self::get_job). pub fn list_jobs( &self, status: Option, @@ -152,6 +154,7 @@ impl RedisStorage { Ok(jobs) } + /// Fetch a job by id, blobs included — live jobs first, then the archive. pub fn get_job(&self, id: &str) -> Result> { let mut conn = self.conn()?; match self.load_job(&mut conn, id)? { @@ -160,6 +163,8 @@ impl RedisStorage { } } + /// Global queue statistics: live counts plus terminal counts from the + /// archive. pub fn stats(&self) -> Result { let mut conn = self.conn()?; let mut stats = QueueStats::default(); @@ -261,11 +266,13 @@ impl RedisStorage { self.count_in_status(&mut conn, &by_queue_key, JobStatus::Pending) } + /// Statistics for one queue: live counts plus archived terminal counts. pub fn stats_by_queue(&self, queue_name: &str) -> Result { let mut conn = self.conn()?; self.queue_stats(&mut conn, queue_name) } + /// Statistics broken down per queue name. pub fn stats_all_queues(&self) -> Result> { let mut conn = self.conn()?; @@ -312,6 +319,8 @@ impl RedisStorage { } #[allow(clippy::too_many_arguments)] + /// `list_jobs` with extra filters (metadata/error substring, created-at + /// range). Rows are blob-free like every listing. pub fn list_jobs_filtered( &self, status: Option, diff --git a/crates/taskito-core/src/storage/redis_backend/jobs/state.rs b/crates/taskito-core/src/storage/redis_backend/jobs/state.rs index 237ccc11..93b3dad9 100644 --- a/crates/taskito-core/src/storage/redis_backend/jobs/state.rs +++ b/crates/taskito-core/src/storage/redis_backend/jobs/state.rs @@ -47,6 +47,7 @@ const REQUEUE_STUCK_IF_RUNNING: &str = r#" "#; impl RedisStorage { + /// Mark a job completed with its result, moving it into the archive. pub fn complete(&self, id: &str, result_bytes: Option>) -> Result<()> { let mut conn = self.conn()?; let mut job = self.get_job_required(id)?; @@ -85,6 +86,7 @@ impl RedisStorage { Ok(()) } + /// Mark a job terminally failed, moving it into the archive. pub fn fail(&self, id: &str, error: &str) -> Result<()> { let mut conn = self.conn()?; let mut job = self.get_job_required(id)?; @@ -102,6 +104,8 @@ impl RedisStorage { Ok(()) } + /// Re-schedule a job for retry at `next_scheduled_at` (Unix milliseconds), + /// incrementing its `retry_count`. pub fn retry(&self, id: &str, next_scheduled_at: i64) -> Result<()> { let mut conn = self.conn()?; let mut job = self.get_job_required(id)?; @@ -194,6 +198,8 @@ impl RedisStorage { Ok(applied == 1) } + /// Cancel a `Pending` job (archived as `Cancelled`) and cascade-cancel its + /// dependents. Returns `false` when the job is missing or not pending. pub fn cancel_job(&self, id: &str) -> Result { let mut conn = self.conn()?; let job = match self.get_job(id)? { @@ -221,6 +227,8 @@ impl RedisStorage { Ok(true) } + /// Set the cancel-requested flag on a `Running` job — the task must poll + /// for it. Returns `false` when no running job matched. pub fn request_cancel(&self, id: &str) -> Result { let mut conn = self.conn()?; let mut job = match self.get_job(id)? { @@ -253,6 +261,7 @@ impl RedisStorage { Ok(applied == 1) } + /// Whether cancellation has been requested for a job. pub fn is_cancel_requested(&self, id: &str) -> Result { let mut conn = self.conn()?; let cancel_set = self.key(&["jobs", "cancel_requested"]); @@ -260,6 +269,7 @@ impl RedisStorage { Ok(is_member) } + /// Archive a live job as `Cancelled` after a cancel request was observed. pub fn mark_cancelled(&self, id: &str) -> Result<()> { let mut conn = self.conn()?; let mut job = self.get_job_required(id)?; @@ -277,6 +287,8 @@ impl RedisStorage { Ok(()) } + /// Cancel every pending job that depends, directly or transitively, on + /// `failed_job_id`. pub fn cascade_cancel(&self, failed_job_id: &str, reason: &str) -> Result<()> { let now = now_millis(); @@ -322,6 +334,7 @@ impl RedisStorage { Ok(()) } + /// Ids of the jobs `job_id` depends on. pub fn get_dependencies(&self, job_id: &str) -> Result> { let mut conn = self.conn()?; let key = self.key(&["job", job_id, "depends_on"]); @@ -329,6 +342,7 @@ impl RedisStorage { Ok(ids) } + /// Ids of the jobs that depend on `job_id`. pub fn get_dependents(&self, job_id: &str) -> Result> { let mut conn = self.conn()?; let key = self.key(&["job", job_id, "dependents"]); @@ -336,6 +350,7 @@ impl RedisStorage { Ok(ids) } + /// Update a live (not yet archived) job's progress (0-100). pub fn update_progress(&self, id: &str, progress: i32) -> Result<()> { if !(0..=100).contains(&progress) { return Err(QueueError::Other( diff --git a/crates/taskito-core/src/storage/redis_backend/locks.rs b/crates/taskito-core/src/storage/redis_backend/locks.rs index ab2835a4..05fe75ab 100644 --- a/crates/taskito-core/src/storage/redis_backend/locks.rs +++ b/crates/taskito-core/src/storage/redis_backend/locks.rs @@ -78,6 +78,8 @@ const REAP_LOCK_SCRIPT: &str = r#" "#; impl RedisStorage { + /// Try to take a distributed lock for `ttl_ms` (milliseconds) via a + /// Lua-atomic check-and-set. Returns `false` while another holder's lock is unexpired. pub fn acquire_lock(&self, lock_name: &str, owner_id: &str, ttl_ms: i64) -> Result { let mut conn = self.conn()?; let now = now_millis(); @@ -97,6 +99,7 @@ impl RedisStorage { Ok(result == 1) } + /// Release a lock. Lua-checked: returns `true` only if `owner_id` held it. pub fn release_lock(&self, lock_name: &str, owner_id: &str) -> Result { let mut conn = self.conn()?; let lkey = self.key(&["lock", lock_name]); @@ -110,6 +113,8 @@ impl RedisStorage { Ok(result == 1) } + /// Reset a lock's expiry to `ttl_ms` milliseconds from now (not additive). + /// Returns `true` only if `owner_id` held it. pub fn extend_lock(&self, lock_name: &str, owner_id: &str, ttl_ms: i64) -> Result { let mut conn = self.conn()?; let now = now_millis(); @@ -126,6 +131,7 @@ impl RedisStorage { Ok(result == 1) } + /// Holder and expiry of a lock, if it exists. pub fn get_lock_info(&self, lock_name: &str) -> Result> { let mut conn = self.conn()?; let lkey = self.key(&["lock", lock_name]); @@ -154,6 +160,8 @@ impl RedisStorage { })) } + /// Remove locks expired before `now` (Unix milliseconds); a Lua re-check + /// closes the scan-then-delete race. Returns the count removed. pub fn reap_expired_locks(&self, now: i64) -> Result { let mut conn = self.conn()?; let pattern = self.key(&["lock", "*"]); @@ -185,6 +193,8 @@ impl RedisStorage { Ok(count) } + /// Claim exclusive execution of a job via `SET NX` with a 24-hour expiry. + /// Returns `false` when a claim already exists. pub fn claim_execution(&self, job_id: &str, worker_id: &str) -> Result { let mut conn = self.conn()?; let now = now_millis(); @@ -258,6 +268,7 @@ impl RedisStorage { Ok(claimed) } + /// Remove the execution claim of a finished job. pub fn complete_execution(&self, job_id: &str) -> Result<()> { let mut conn = self.conn()?; let ckey = self.key(&["exec_claim", job_id]); @@ -298,6 +309,8 @@ impl RedisStorage { Ok(result == 1) } + /// Purge execution claims older than the cutoff via the time-indexed sorted + /// set. Returns the count removed. pub fn purge_execution_claims(&self, older_than_ms: i64) -> Result { let mut conn = self.conn()?; let index_key = self.key(&["exec_claims", "by_time"]); diff --git a/crates/taskito-core/src/storage/redis_backend/logs.rs b/crates/taskito-core/src/storage/redis_backend/logs.rs index e7801511..37119e90 100644 --- a/crates/taskito-core/src/storage/redis_backend/logs.rs +++ b/crates/taskito-core/src/storage/redis_backend/logs.rs @@ -32,6 +32,7 @@ impl From for TaskLogEntry { } impl RedisStorage { + /// Write one structured log line for a job. `extra` is pre-encoded JSON. pub fn write_task_log( &self, job_id: &str, @@ -69,6 +70,8 @@ impl RedisStorage { Ok(()) } + /// All log lines for a job, in chronological (UUIDv7 id) order; + /// same-millisecond lines have no guaranteed emission order. pub fn get_task_logs(&self, job_id: &str) -> Result> { let mut conn = self.conn()?; let by_job_key = self.key(&["logs", "by_job", job_id]); @@ -125,6 +128,8 @@ impl RedisStorage { Ok(rows) } + /// Log lines across jobs, filtered by task name and/or level, newest since + /// `since_ms` (Unix milliseconds), bounded by `limit`. pub fn query_task_logs( &self, task_name: Option<&str>, @@ -132,6 +137,11 @@ impl RedisStorage { since_ms: i64, limit: i64, ) -> Result> { + // A zero limit is an empty page; without this the filtered walk below + // never hits its `rows.len() == limit` stop and scans the whole range. + if limit == 0 { + return Ok(Vec::new()); + } let mut conn = self.conn()?; let all_key = self.key(&["logs", "all"]); @@ -209,6 +219,8 @@ impl RedisStorage { Ok(rows) } + /// Purge log lines logged at or before the cutoff (inclusive). Returns + /// the count removed. pub fn purge_task_logs(&self, older_than_ms: i64) -> Result { let mut conn = self.conn()?; let all_key = self.key(&["logs", "all"]); diff --git a/crates/taskito-core/src/storage/redis_backend/metrics.rs b/crates/taskito-core/src/storage/redis_backend/metrics.rs index 108659b0..02fe9249 100644 --- a/crates/taskito-core/src/storage/redis_backend/metrics.rs +++ b/crates/taskito-core/src/storage/redis_backend/metrics.rs @@ -59,6 +59,7 @@ impl From for ReplayEntry { } impl RedisStorage { + /// Record one execution measurement for a task. pub fn record_metric( &self, task_name: &str, @@ -96,6 +97,8 @@ impl RedisStorage { Ok(()) } + /// Metrics recorded since `since_ms` (Unix milliseconds) for one task, or + /// all tasks when `name` is `None`. pub fn get_metrics(&self, name: Option<&str>, since_ms: i64) -> Result> { let mut conn = self.conn()?; @@ -124,6 +127,8 @@ impl RedisStorage { Ok(rows) } + /// Purge metric records recorded at or before the cutoff (inclusive). + /// Returns the count removed. pub fn purge_metrics(&self, older_than_ms: i64) -> Result { let mut conn = self.conn()?; let all_key = self.key(&["metrics", "all"]); @@ -155,6 +160,8 @@ impl RedisStorage { Ok(ids.len() as u64) } + /// Record a replay of a completed job, pairing original and replay + /// outcomes. pub fn record_replay( &self, original_job_id: &str, @@ -192,6 +199,7 @@ impl RedisStorage { Ok(()) } + /// All replays recorded against `original_job_id`. pub fn get_replay_history(&self, original_job_id: &str) -> Result> { let mut conn = self.conn()?; let by_original = self.key(&["replay", "by_original", original_job_id]); diff --git a/crates/taskito-core/src/storage/redis_backend/periodic.rs b/crates/taskito-core/src/storage/redis_backend/periodic.rs index 5f54f602..3685cd92 100644 --- a/crates/taskito-core/src/storage/redis_backend/periodic.rs +++ b/crates/taskito-core/src/storage/redis_backend/periodic.rs @@ -37,6 +37,7 @@ impl From for PeriodicTask { } impl RedisStorage { + /// Register or update a periodic task by name. pub fn register_periodic(&self, task: &NewPeriodicTask) -> Result<()> { let mut conn = self.conn()?; @@ -68,6 +69,8 @@ impl RedisStorage { Ok(()) } + /// Enabled periodic tasks whose `next_run` is due at `now` (Unix + /// milliseconds). pub fn get_due_periodic(&self, now: i64) -> Result> { let mut conn = self.conn()?; let due_key = self.key(&["periodic", "due"]); @@ -91,6 +94,7 @@ impl RedisStorage { Ok(rows) } + /// Advance a periodic task's schedule after it fires. pub fn update_periodic_schedule(&self, name: &str, last_run: i64, next_run: i64) -> Result<()> { let mut conn = self.conn()?; let pkey = self.key(&["periodic", name]); diff --git a/crates/taskito-core/src/storage/redis_backend/queue_state.rs b/crates/taskito-core/src/storage/redis_backend/queue_state.rs index efce541e..78057f12 100644 --- a/crates/taskito-core/src/storage/redis_backend/queue_state.rs +++ b/crates/taskito-core/src/storage/redis_backend/queue_state.rs @@ -4,6 +4,7 @@ use super::{map_err, RedisStorage}; use crate::error::Result; impl RedisStorage { + /// Pause a queue so no new jobs are dispatched from it. pub fn pause_queue(&self, queue_name: &str) -> Result<()> { let mut conn = self.conn()?; let paused_key = self.key(&["queues", "paused"]); @@ -12,6 +13,7 @@ impl RedisStorage { Ok(()) } + /// Resume a paused queue. pub fn resume_queue(&self, queue_name: &str) -> Result<()> { let mut conn = self.conn()?; let paused_key = self.key(&["queues", "paused"]); @@ -20,6 +22,7 @@ impl RedisStorage { Ok(()) } + /// Names of all currently paused queues. pub fn list_paused_queues(&self) -> Result> { let mut conn = self.conn()?; let paused_key = self.key(&["queues", "paused"]); diff --git a/crates/taskito-core/src/storage/redis_backend/rate_limits.rs b/crates/taskito-core/src/storage/redis_backend/rate_limits.rs index 8c0a984d..305d7836 100644 --- a/crates/taskito-core/src/storage/redis_backend/rate_limits.rs +++ b/crates/taskito-core/src/storage/redis_backend/rate_limits.rs @@ -6,6 +6,7 @@ use crate::job::now_millis; use crate::storage::records::RateLimitState; impl RedisStorage { + /// Token-bucket state for a rate-limit key, if one exists. pub fn get_rate_limit(&self, key: &str) -> Result> { let mut conn = self.conn()?; let rkey = self.key(&["rate_limit", key]); @@ -20,6 +21,7 @@ impl RedisStorage { } } + /// Insert or replace a token-bucket state entry. pub fn upsert_rate_limit(&self, row: &RateLimitState) -> Result<()> { let mut conn = self.conn()?; let rkey = self.key(&["rate_limit", &row.key]); @@ -28,6 +30,8 @@ impl RedisStorage { Ok(()) } + /// Atomically refill and consume one token via a Lua script. Returns + /// `false` when the bucket is empty. pub fn try_acquire_token(&self, key: &str, max_tokens: f64, refill_rate: f64) -> Result { let mut conn = self.conn()?; let now = now_millis(); diff --git a/crates/taskito-core/src/storage/redis_backend/workers.rs b/crates/taskito-core/src/storage/redis_backend/workers.rs index 27ef431c..ba821a0b 100644 --- a/crates/taskito-core/src/storage/redis_backend/workers.rs +++ b/crates/taskito-core/src/storage/redis_backend/workers.rs @@ -7,6 +7,8 @@ use crate::storage::records::WorkerInfo; impl RedisStorage { #[allow(clippy::too_many_arguments)] + /// Register a worker in the cluster registry, or update it if the id + /// already exists. `tags`/`resources` are pre-encoded JSON. pub fn register_worker( &self, worker_id: &str, @@ -42,6 +44,8 @@ impl RedisStorage { Ok(()) } + /// Refresh a worker's heartbeat timestamp and overwrite its + /// resource-health JSON (`None` clears any previous value). pub fn heartbeat(&self, worker_id: &str, resource_health: Option<&str>) -> Result<()> { let mut conn = self.conn()?; let now = now_millis(); @@ -55,6 +59,7 @@ impl RedisStorage { Ok(()) } + /// Set a worker's status string. pub fn update_worker_status(&self, worker_id: &str, status: &str) -> Result<()> { let mut conn = self.conn()?; let wkey = self.key(&["worker", worker_id]); @@ -65,6 +70,7 @@ impl RedisStorage { Ok(()) } + /// Every registered worker with its heartbeat state. pub fn list_workers(&self) -> Result> { let mut conn = self.conn()?; let wall = self.key(&["workers", "all"]); @@ -143,6 +149,8 @@ impl RedisStorage { .collect()) } + /// Remove workers whose heartbeat is stale past the dead-worker threshold. + /// Returns the reaped worker ids. pub fn reap_dead_workers(&self) -> Result> { let mut conn = self.conn()?; let cutoff = crate::storage::dead_worker_cutoff(now_millis()); @@ -189,6 +197,7 @@ impl RedisStorage { pipe.query(conn).map_err(map_err) } + /// Remove a worker from the registry (called on shutdown). pub fn unregister_worker(&self, worker_id: &str) -> Result<()> { let mut conn = self.conn()?; let wkey = self.key(&["worker", worker_id]); @@ -202,6 +211,7 @@ impl RedisStorage { Ok(()) } + /// Job ids currently execution-claimed by a worker. pub fn list_claims_by_worker(&self, worker_id: &str) -> Result> { let mut conn = self.conn()?; let pattern = self.key(&["exec_claim", "*"]); diff --git a/crates/taskito-core/src/storage/sqlite/dashboard_settings.rs b/crates/taskito-core/src/storage/sqlite/dashboard_settings.rs index bc98930f..83b07ecb 100644 --- a/crates/taskito-core/src/storage/sqlite/dashboard_settings.rs +++ b/crates/taskito-core/src/storage/sqlite/dashboard_settings.rs @@ -9,6 +9,7 @@ use crate::error::Result; use crate::job::now_millis; impl SqliteStorage { + /// Fetch a single setting value by key, or `None` if unset. pub fn get_setting(&self, key: &str) -> Result> { let mut conn = self.conn()?; let row: Option = dashboard_settings::table @@ -18,6 +19,7 @@ impl SqliteStorage { Ok(row.map(|r| r.value)) } + /// Insert or update a setting. pub fn set_setting(&self, key: &str, value: &str) -> Result<()> { let mut conn = self.conn()?; let row = DashboardSettingRow { @@ -31,6 +33,7 @@ impl SqliteStorage { Ok(()) } + /// Delete a setting. Returns `true` if a row was removed. pub fn delete_setting(&self, key: &str) -> Result { let mut conn = self.conn()?; let deleted = @@ -39,6 +42,7 @@ impl SqliteStorage { Ok(deleted > 0) } + /// Return all settings as a key-to-value map. pub fn list_settings(&self) -> Result> { let mut conn = self.conn()?; let rows: Vec = dashboard_settings::table diff --git a/crates/taskito-core/src/storage/sqlite/mod.rs b/crates/taskito-core/src/storage/sqlite/mod.rs index c35144cd..2ea2f91c 100644 --- a/crates/taskito-core/src/storage/sqlite/mod.rs +++ b/crates/taskito-core/src/storage/sqlite/mod.rs @@ -98,7 +98,7 @@ impl SqliteStorage { } /// Replace the in-process wake handle so it is shared with the scheduler's - /// [`WakeSource`]. Only meaningful under `push-dispatch`. + /// `WakeSource`. Only meaningful under `push-dispatch`. #[cfg(feature = "push-dispatch")] pub fn set_notify_handle(&mut self, notify: std::sync::Arc) { self.notify = notify; @@ -111,6 +111,7 @@ impl SqliteStorage { &self.notify } + /// Check a pooled SQLite connection out of the r2d2 pool. pub fn conn( &self, ) -> Result>> { diff --git a/crates/taskito-core/src/storage/sqlite/rate_limits.rs b/crates/taskito-core/src/storage/sqlite/rate_limits.rs index f379992c..eec729aa 100644 --- a/crates/taskito-core/src/storage/sqlite/rate_limits.rs +++ b/crates/taskito-core/src/storage/sqlite/rate_limits.rs @@ -8,6 +8,7 @@ use crate::error::Result; use crate::job::now_millis; impl SqliteStorage { + /// Token-bucket state for a rate-limit key, if one exists. pub fn get_rate_limit(&self, key: &str) -> Result> { let mut conn = self.conn()?; @@ -20,6 +21,7 @@ impl SqliteStorage { Ok(row.map(Into::into)) } + /// Insert or replace a token-bucket state row. pub fn upsert_rate_limit(&self, state: &RateLimitState) -> Result<()> { let mut conn = self.conn()?; @@ -34,7 +36,7 @@ impl SqliteStorage { /// Atomically try to acquire a rate limit token. /// Does the read-refill-consume-write in a single write transaction to /// prevent race conditions between concurrent workers. Uses - /// [`SqliteStorage::write_transaction`] (BEGIN IMMEDIATE) so the read-then- + /// `SqliteStorage::write_transaction` (BEGIN IMMEDIATE) so the read-then- /// write can't hit the deferred-lock-upgrade `SQLITE_BUSY` deadlock. pub fn try_acquire_token(&self, key: &str, max_tokens: f64, refill_rate: f64) -> Result { let now = now_millis(); diff --git a/crates/taskito-core/src/storage/traits.rs b/crates/taskito-core/src/storage/traits.rs index 59a6773e..d98e92a1 100644 --- a/crates/taskito-core/src/storage/traits.rs +++ b/crates/taskito-core/src/storage/traits.rs @@ -14,11 +14,20 @@ use crate::storage::{DeadJob, QueueStats, SubscriptionBacklogStats}; pub trait Storage: Send + Sync + Clone { // ── Job operations ────────────────────────────────────────────── + /// Insert a new job and return it. fn enqueue(&self, new_job: NewJob) -> Result; + /// Insert multiple jobs in a single transaction. fn enqueue_batch(&self, new_jobs: Vec) -> Result>; + /// Enqueue with `unique_key` deduplication: returns the existing active + /// job when a duplicate is found instead of inserting. fn enqueue_unique(&self, new_job: NewJob) -> Result; + /// Batch variant of [`enqueue_unique`](Self::enqueue_unique), one transaction. fn enqueue_unique_batch(&self, new_jobs: Vec) -> Result>; + /// Atomically claim the highest-priority ready job from a queue, moving it + /// to `Running`. `None` when no job is eligible. `namespace = None` + /// matches only namespace-less jobs. fn dequeue(&self, queue_name: &str, now: i64, namespace: Option<&str>) -> Result>; + /// [`dequeue`](Self::dequeue) across several queues, checked in order. fn dequeue_from( &self, queues: &[String], @@ -44,6 +53,8 @@ pub trait Storage: Send + Sync + Clone { namespace: Option<&str>, max: usize, ) -> Result>; + /// Mark a job completed with its result, moving it from `jobs` into + /// `archived_jobs` in one transaction. fn complete(&self, id: &str, result_bytes: Option>) -> Result<()>; /// Persist many successful completions at once. Each entry archives the @@ -52,7 +63,10 @@ pub trait Storage: Send + Sync + Clone { /// /// [`JobCompletion`]: crate::job::JobCompletion fn complete_batch(&self, completions: &[crate::job::JobCompletion]) -> Result<()>; + /// Mark a job terminally failed, moving it from `jobs` into `archived_jobs`. fn fail(&self, id: &str, error: &str) -> Result<()>; + /// Re-schedule a job for retry at `next_scheduled_at`, incrementing its + /// `retry_count`. fn retry(&self, id: &str, next_scheduled_at: i64) -> Result<()>; /// Re-schedule a job back to `Pending` **without** consuming its retry /// budget. Used for soft-gate reschedules (rate limit, circuit breaker, @@ -66,13 +80,24 @@ pub trait Storage: Send + Sync + Clone { /// missing or not `Running`. Only for confirmed-dead/hung workers: a /// still-alive owner may finish the old attempt, double-executing the job. fn requeue_stuck(&self, id: &str, now: i64) -> Result; + /// Cancel a `Pending` job (archived as `Cancelled`) and cascade-cancel its + /// dependents. Returns `false` when the job is missing or not pending. fn cancel_job(&self, id: &str) -> Result; + /// Set the cancel-requested flag on a `Running` job — the task must poll + /// for it. Returns `false` when no running job matched. fn request_cancel(&self, id: &str) -> Result; + /// Whether cancellation has been requested for a job. fn is_cancel_requested(&self, id: &str) -> Result; + /// Archive a running job as `Cancelled` after it observed a cancel request. fn mark_cancelled(&self, id: &str) -> Result<()>; + /// Cancel every pending job that depends, directly or transitively, on + /// `failed_job_id`. fn cascade_cancel(&self, failed_job_id: &str, reason: &str) -> Result<()>; + /// Ids of the jobs `job_id` depends on. fn get_dependencies(&self, job_id: &str) -> Result>; + /// Ids of the jobs that depend on `job_id`. fn get_dependents(&self, job_id: &str) -> Result>; + /// Update a running job's progress (0-100). fn update_progress(&self, id: &str, progress: i32) -> Result<()>; /// List jobs by filter. Rows are **blob-free** on every backend: the /// `payload`/`result` blobs come back empty (Diesel selects a narrow @@ -110,25 +135,38 @@ pub trait Storage: Send + Sync + Clone { after: Option<(i64, &str)>, namespace: Option<&str>, ) -> Result>; + /// Fetch a job by id, blobs included — live `jobs` first, then + /// `archived_jobs`. fn get_job(&self, id: &str) -> Result>; + /// Global queue statistics: live counts from `jobs`, terminal counts from + /// `archived_jobs`. fn stats(&self) -> Result; + /// Purge archived completed jobs older than the cutoff. Returns the count + /// removed. fn purge_completed(&self, older_than_ms: i64) -> Result; /// Purge archived jobs by the global/per-entry TTL, covering every terminal /// status on all backends. fn purge_completed_with_ttl(&self, global_cutoff_ms: Option) -> Result; + /// Running jobs that exceeded their timeout, for the scheduler to fail or + /// retry. fn reap_stale_jobs(&self, now: i64) -> Result>; /// Running jobs whose execution-claim owner is not in `live_owner_ids` (the /// worker that claimed them has died). Read-only — paired with the dead /// owner so the caller can atomically reclaim before requeuing. fn reap_orphaned_jobs(&self, live_owner_ids: &[String], now: i64) -> Result>; + /// Record one failed attempt's error for a job. fn record_error(&self, job_id: &str, attempt: i32, error: &str) -> Result<()>; + /// All recorded errors for a job, ordered by attempt. fn get_job_errors(&self, job_id: &str) -> Result>; + /// Purge error records older than the cutoff. Returns the count removed. fn purge_job_errors(&self, older_than_ms: i64) -> Result; // ── Dead letter operations ────────────────────────────────────── + /// Move a job to the dead-letter queue and cascade-cancel its dependents. fn move_to_dlq(&self, job: &Job, error: &str, metadata: Option<&str>) -> Result<()>; + /// Dead-letter entries, newest first, paginated. fn list_dead(&self, limit: i64, offset: i64) -> Result>; /// Keyset-paginated `list_dead`, ordered by `(failed_at, id)` descending. /// See [`Storage::list_jobs_after`] for the cursor contract. @@ -137,10 +175,18 @@ pub trait Storage: Send + Sync + Clone { fn list_dead_by_task(&self, task_name: &str, limit: i64, offset: i64) -> Result>; /// Delete every dead-letter entry for a task. Returns the number removed. fn purge_dead_by_task(&self, task_name: &str) -> Result; + /// Re-enqueue a dead-letter entry as a fresh job, deleting the entry. + /// Returns the new job's id; `JobNotFound` if the entry is absent. fn retry_dead(&self, dead_id: &str) -> Result; + /// Purge dead-letter entries older than the cutoff. Returns the count + /// removed. fn purge_dead(&self, older_than_ms: i64) -> Result; + /// Delete one dead-letter entry. Returns `false` when no row matched. fn delete_dead(&self, dead_id: &str) -> Result; + /// Purge dead-letter entries by the global/per-entry TTL. Returns the + /// count removed. fn purge_dead_with_ttl(&self, global_cutoff_ms: Option) -> Result; + /// Dead-letter entries eligible for automatic retry, bounded by `limit`. fn list_dead_for_retry( &self, cutoff_ms: i64, @@ -152,14 +198,21 @@ pub trait Storage: Send + Sync + Clone { // ── Rate limit operations ─────────────────────────────────────── + /// Token-bucket state for a rate-limit key, if one exists. fn get_rate_limit(&self, key: &str) -> Result>; + /// Insert or replace a token-bucket state row. fn upsert_rate_limit(&self, row: &RateLimitState) -> Result<()>; + /// Atomically refill and consume one token. Returns `false` when the + /// bucket is empty. fn try_acquire_token(&self, key: &str, max_tokens: f64, refill_rate: f64) -> Result; // ── Periodic task operations ──────────────────────────────────── + /// Register or update a periodic task by name. fn register_periodic(&self, task: &NewPeriodicTask) -> Result<()>; + /// Enabled periodic tasks whose `next_run` is due at `now`. fn get_due_periodic(&self, now: i64) -> Result>; + /// Advance a periodic task's schedule after it fires. fn update_periodic_schedule(&self, name: &str, last_run: i64, next_run: i64) -> Result<()>; /// All registered periodic tasks, enabled or paused. fn list_periodic(&self) -> Result>; @@ -198,6 +251,7 @@ pub trait Storage: Send + Sync + Clone { // ── Metrics operations ────────────────────────────────────────── + /// Record one execution measurement for a task. fn record_metric( &self, task_name: &str, @@ -206,8 +260,13 @@ pub trait Storage: Send + Sync + Clone { memory_bytes: i64, succeeded: bool, ) -> Result<()>; + /// Metrics recorded since `since_ms` for one task, or all tasks when + /// `name` is `None`. fn get_metrics(&self, name: Option<&str>, since_ms: i64) -> Result>; + /// Purge metric records older than the cutoff. Returns the count removed. fn purge_metrics(&self, older_than_ms: i64) -> Result; + /// Record a replay of a completed job, pairing original and replay + /// outcomes. fn record_replay( &self, original_job_id: &str, @@ -217,10 +276,12 @@ pub trait Storage: Send + Sync + Clone { original_error: Option<&str>, replay_error: Option<&str>, ) -> Result<()>; + /// All replays recorded against `original_job_id`. fn get_replay_history(&self, original_job_id: &str) -> Result>; // ── Log operations ────────────────────────────────────────────── + /// Write one structured log line for a job. `extra` is pre-encoded JSON. fn write_task_log( &self, job_id: &str, @@ -229,6 +290,7 @@ pub trait Storage: Send + Sync + Clone { message: &str, extra: Option<&str>, ) -> Result<()>; + /// All log lines for a job, in emission order. fn get_task_logs(&self, job_id: &str) -> Result>; /// Logs for a job with id strictly after `after_id` (UUIDv7 ids are /// time-ordered, so the id doubles as a stream cursor). `None` = all. @@ -237,6 +299,8 @@ pub trait Storage: Send + Sync + Clone { job_id: &str, after_id: Option<&str>, ) -> Result>; + /// Log lines across jobs, filtered by task name and/or level, newest + /// since `since_ms`, bounded by `limit`. fn query_task_logs( &self, task_name: Option<&str>, @@ -244,16 +308,22 @@ pub trait Storage: Send + Sync + Clone { since_ms: i64, limit: i64, ) -> Result>; + /// Purge log lines older than the cutoff. Returns the count removed. fn purge_task_logs(&self, older_than_ms: i64) -> Result; // ── Circuit breaker operations ────────────────────────────────── + /// Persisted circuit-breaker state for a task, if one exists. fn get_circuit_breaker(&self, task_name: &str) -> Result>; + /// Insert or replace a task's circuit-breaker state. fn upsert_circuit_breaker(&self, row: &CircuitBreakerState) -> Result<()>; + /// Every persisted circuit-breaker state. fn list_circuit_breakers(&self) -> Result>; // ── Worker operations ─────────────────────────────────────────── + /// Register a worker in the cluster registry, or update it if the id + /// already exists. `tags`/`resources` are pre-encoded JSON. #[allow(clippy::too_many_arguments)] fn register_worker( &self, @@ -267,35 +337,54 @@ pub trait Storage: Send + Sync + Clone { pid: Option, pool_type: Option<&str>, ) -> Result<()>; + /// Refresh a worker's heartbeat timestamp, optionally updating its + /// resource-health JSON. fn heartbeat(&self, worker_id: &str, resource_health: Option<&str>) -> Result<()>; + /// Set a worker's status string. fn update_worker_status(&self, worker_id: &str, status: &str) -> Result<()>; + /// Every registered worker with its heartbeat state. fn list_workers(&self) -> Result>; /// Ids of workers whose heartbeat is at or after `cutoff_ms`. A narrow /// projection of [`Self::list_workers`] for callers that only need the live /// set and must not pay to load every worker's `resource_health` blob. fn list_live_worker_ids(&self, cutoff_ms: i64) -> Result>; + /// Remove workers whose heartbeat is stale past the dead-worker threshold. + /// Returns the reaped worker ids. fn reap_dead_workers(&self) -> Result>; + /// Remove a worker from the registry (called on shutdown). fn unregister_worker(&self, worker_id: &str) -> Result<()>; + /// Job ids currently execution-claimed by a worker. fn list_claims_by_worker(&self, worker_id: &str) -> Result>; // ── Queue pause/resume ─────────────────────────────────────── + /// Pause a queue so no new jobs are dispatched from it. fn pause_queue(&self, queue_name: &str) -> Result<()>; + /// Resume a paused queue. fn resume_queue(&self, queue_name: &str) -> Result<()>; + /// Names of all currently paused queues. fn list_paused_queues(&self) -> Result>; // ── Job expiry ─────────────────────────────────────────────── + /// Fail pending jobs whose `expires_at` has passed. Returns the count + /// expired. fn expire_pending_jobs(&self, now: i64) -> Result; // ── Job revocation ─────────────────────────────────────────── + /// Cancel every pending job in a queue. Returns the count cancelled. fn cancel_pending_by_queue(&self, queue: &str) -> Result; + /// Cancel every pending job for a task. Returns the count cancelled. fn cancel_pending_by_task(&self, task_name: &str) -> Result; // ── Job archival ───────────────────────────────────────────── + /// Move `Complete`/`Dead`/`Cancelled` jobs older than the cutoff from + /// `jobs` into `archived_jobs`. Returns the count archived. `Failed` + /// jobs are archived immediately by `fail()`, never by this sweep. fn archive_old_jobs(&self, cutoff_ms: i64) -> Result; + /// Archived jobs, newest first, paginated. Rows are blob-free. fn list_archived(&self, limit: i64, offset: i64) -> Result>; /// Keyset-paginated `list_archived`, ordered by `(completed_at, id)` /// descending. See [`Storage::list_jobs_after`] for the cursor contract. @@ -303,21 +392,32 @@ pub trait Storage: Send + Sync + Clone { // ── Distributed locking ──────────────────────────────────── + /// Try to take a distributed lock for `ttl_ms`. Returns `false` when + /// another owner (or this one) still holds an unexpired lock. fn acquire_lock(&self, lock_name: &str, owner_id: &str, ttl_ms: i64) -> Result; + /// Release a lock. Returns `true` only if `owner_id` held it. fn release_lock(&self, lock_name: &str, owner_id: &str) -> Result; + /// Extend a lock's TTL. Returns `true` only if `owner_id` held it. fn extend_lock(&self, lock_name: &str, owner_id: &str, ttl_ms: i64) -> Result; + /// Holder and expiry of a lock, if it exists. fn get_lock_info(&self, lock_name: &str) -> Result>; + /// Remove locks that expired before `now`. Returns the count removed. fn reap_expired_locks(&self, now: i64) -> Result; // ── Execution claims (exactly-once) ──────────────────────── + /// Claim exclusive execution of a job for `worker_id`. Returns `false` + /// when a claim already exists. fn claim_execution(&self, job_id: &str, worker_id: &str) -> Result; /// Batch variant of [`Storage::claim_execution`]: attempt to claim every /// `job_id` for `worker_id` in as few round trips as the backend allows. /// Returns one flag per input id, in order — `true` if this worker won the /// claim, `false` if a claim already existed. fn claim_execution_batch(&self, job_ids: &[&str], worker_id: &str) -> Result>; + /// Remove the execution claim of a finished job. fn complete_execution(&self, job_id: &str) -> Result<()>; + /// Purge execution claims older than the cutoff. Returns the count + /// removed. fn purge_execution_claims(&self, older_than_ms: i64) -> Result; /// Atomically transfer an existing claim from `expected_owner` to /// `new_owner`. Returns `true` only if the claim was held by @@ -332,6 +432,7 @@ pub trait Storage: Send + Sync + Clone { // ── Per-task concurrency ────────────────────────────────────── + /// Running-job count for a task — the per-task concurrency-cap primitive. fn count_running_by_task(&self, task_name: &str) -> Result; // ── Per-queue stats ────────────────────────────────────────── @@ -340,11 +441,16 @@ pub trait Storage: Send + Sync + Clone { /// Single-status, unlike the full-breakdown `stats_by_queue`. fn count_pending_by_queue(&self, queue_name: &str) -> Result; + /// Statistics for one queue: live counts from `jobs`, terminal counts + /// from `archived_jobs`. fn stats_by_queue(&self, queue_name: &str) -> Result; + /// Statistics broken down per queue name. fn stats_all_queues(&self) -> Result>; // ── Filtered job listing ───────────────────────────────────── + /// `list_jobs` with extra filters (metadata/error substring, created-at + /// range). Rows are blob-free like every listing. #[allow(clippy::too_many_arguments)] fn list_jobs_filtered( &self, diff --git a/crates/taskito-core/src/worker/dispatcher.rs b/crates/taskito-core/src/worker/dispatcher.rs index 06213c33..9b77c9f2 100644 --- a/crates/taskito-core/src/worker/dispatcher.rs +++ b/crates/taskito-core/src/worker/dispatcher.rs @@ -28,6 +28,8 @@ pub struct NativeDispatcher { } impl NativeDispatcher { + /// Build a dispatcher over `registry`, running at most `num_workers` + /// handlers concurrently (minimum 1). pub fn new(registry: TaskRegistry, num_workers: usize) -> Self { Self { registry: Arc::new(registry), diff --git a/crates/taskito-core/src/worker/registry.rs b/crates/taskito-core/src/worker/registry.rs index 75a49d65..17cf1960 100644 --- a/crates/taskito-core/src/worker/registry.rs +++ b/crates/taskito-core/src/worker/registry.rs @@ -23,7 +23,9 @@ pub type TaskFuture = Pin + Send + 'static>> /// straight to the dead-letter queue. #[derive(Debug, Clone)] pub struct TaskError { + /// Human-readable failure message. pub message: String, + /// Whether the scheduler may retry the job. pub retryable: bool, } @@ -57,7 +59,9 @@ impl std::error::Error for TaskError {} /// (spawned on the worker's tokio runtime). #[derive(Clone)] pub enum TaskHandler { + /// Blocking handler, run on a blocking thread. Sync(Arc TaskResult + Send + Sync>), + /// Async handler, spawned on the worker's tokio runtime. Async(Arc TaskFuture + Send + Sync>), } @@ -68,6 +72,7 @@ pub struct TaskRegistry { } impl TaskRegistry { + /// An empty registry. pub fn new() -> Self { Self::default() } @@ -96,6 +101,7 @@ impl TaskRegistry { ); } + /// The handler registered for `task_name`, if any. pub fn get(&self, task_name: &str) -> Option<&TaskHandler> { self.handlers.get(task_name) } @@ -105,6 +111,7 @@ impl TaskRegistry { self.handlers.keys().map(String::as_str) } + /// True when no handlers are registered. pub fn is_empty(&self) -> bool { self.handlers.is_empty() } diff --git a/crates/taskito-core/src/worker/runner.rs b/crates/taskito-core/src/worker/runner.rs index 6a0623b2..4021120e 100644 --- a/crates/taskito-core/src/worker/runner.rs +++ b/crates/taskito-core/src/worker/runner.rs @@ -45,6 +45,7 @@ pub struct Worker { } impl Worker { + /// A worker builder over `storage` with default settings. pub fn new(storage: StorageBackend) -> Self { Self { storage, @@ -72,6 +73,7 @@ impl Worker { self } + /// Tenant namespace this worker is scoped to (default: none). pub fn namespace(mut self, namespace: impl Into) -> Self { self.namespace = Some(namespace.into()); self @@ -304,6 +306,7 @@ pub struct WorkerHandle { } impl WorkerHandle { + /// Id this worker registered under. pub fn worker_id(&self) -> &str { &self.worker_id } diff --git a/crates/taskito-core/tests/rust/storage_tests.rs b/crates/taskito-core/tests/rust/storage_tests.rs index 35c32fe0..af5b4355 100644 --- a/crates/taskito-core/tests/rust/storage_tests.rs +++ b/crates/taskito-core/tests/rust/storage_tests.rs @@ -1563,6 +1563,10 @@ fn test_task_logs_after_cursor(s: &impl Storage) { ); let after_last = s.get_task_logs_after(&job.id, Some(&all[2].id)).unwrap(); assert!(after_last.is_empty()); + + // A zero limit is an empty page, even on the filtered (unindexed) path. + let zero = s.query_task_logs(Some("log_task"), None, 0, 0).unwrap(); + assert!(zero.is_empty()); } fn test_rate_limit_token_exhaustion(s: &impl Storage) { diff --git a/crates/taskito-java/Cargo.toml b/crates/taskito-java/Cargo.toml index 53135ffd..a59c3ce4 100644 --- a/crates/taskito-java/Cargo.toml +++ b/crates/taskito-java/Cargo.toml @@ -1,9 +1,11 @@ [package] name = "taskito-java" -version = "0.20.0" -edition = "2021" +version.workspace = true +edition.workspace = true +license.workspace = true +repository.workspace = true +rust-version.workspace = true description = "Java (JNI) bindings for the Taskito task-queue core" -license = "MIT" publish = false [lib] diff --git a/crates/taskito-mesh/Cargo.toml b/crates/taskito-mesh/Cargo.toml index 08b76bb0..7712dd39 100644 --- a/crates/taskito-mesh/Cargo.toml +++ b/crates/taskito-mesh/Cargo.toml @@ -1,7 +1,11 @@ [package] name = "taskito-mesh" -version = "0.20.0" -edition = "2021" +version.workspace = true +edition.workspace = true +license.workspace = true +repository.workspace = true +rust-version.workspace = true +description = "Decentralized mesh scheduling overlay for the Taskito task queue" [dependencies] taskito-core = { path = "../taskito-core" } diff --git a/crates/taskito-node/Cargo.toml b/crates/taskito-node/Cargo.toml index ba4817ce..eee70baf 100644 --- a/crates/taskito-node/Cargo.toml +++ b/crates/taskito-node/Cargo.toml @@ -1,9 +1,11 @@ [package] name = "taskito-node" -version = "0.20.0" -edition = "2021" +version.workspace = true +edition.workspace = true +license.workspace = true +repository.workspace = true +rust-version.workspace = true description = "Node.js (napi-rs) bindings for the Taskito task-queue core" -license = "MIT" publish = false [lib] diff --git a/crates/taskito-python/Cargo.toml b/crates/taskito-python/Cargo.toml index b9ec638d..03666feb 100644 --- a/crates/taskito-python/Cargo.toml +++ b/crates/taskito-python/Cargo.toml @@ -1,7 +1,12 @@ [package] name = "taskito-python" -version = "0.20.0" -edition = "2021" +version.workspace = true +edition.workspace = true +license.workspace = true +repository.workspace = true +rust-version.workspace = true +description = "Python (PyO3) bindings for the Taskito task-queue core" +publish = false [features] default = [] diff --git a/crates/taskito-tui/Cargo.toml b/crates/taskito-tui/Cargo.toml index 70948d7b..9be51249 100644 --- a/crates/taskito-tui/Cargo.toml +++ b/crates/taskito-tui/Cargo.toml @@ -1,7 +1,10 @@ [package] name = "taskito-tui" -version = "0.20.0" -edition = "2021" +version.workspace = true +edition.workspace = true +license.workspace = true +repository.workspace = true +rust-version.workspace = true description = "Headless terminal UI for inspecting and operating a Taskito queue" [[bin]] diff --git a/crates/taskito-workflows/Cargo.toml b/crates/taskito-workflows/Cargo.toml index bddd958a..feea6147 100644 --- a/crates/taskito-workflows/Cargo.toml +++ b/crates/taskito-workflows/Cargo.toml @@ -1,7 +1,13 @@ [package] name = "taskito-workflows" -version = "0.20.0" -edition = "2021" +version.workspace = true +edition.workspace = true +license.workspace = true +repository.workspace = true +rust-version.workspace = true +description = "Workflow DAG engine for the Taskito task queue" +# Blocked from crates.io by the dagron-core git dependency; flip once it publishes. +publish = false [features] default = []