feat: add native worker with task registry and dispatcher

crates/taskito-core/examples/hello.rs

@@ -0,0 +1,83 @@
+//! Minimal end-to-end run: enqueue two jobs, execute them on a native worker.
+//!
+//! ```sh
+//! cargo run --example hello -p taskito-core
+//! ```
+
+use std::time::{Duration, Instant};
+
+use taskito_core::{
+    now_millis, Job, JobStatus, NewJob, SqliteStorage, Storage, StorageBackend, TaskError, Worker,
+};
+
+fn new_job(task_name: &str, payload: &[u8]) -> NewJob {
+    NewJob {
+        queue: "default".to_string(),
+        task_name: task_name.to_string(),
+        payload: payload.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,
+    }
+}
+
+fn wait_until_completed(storage: &StorageBackend, job_id: &str) -> Job {
+    let deadline = Instant::now() + Duration::from_secs(10);
+    loop {
+        let job = storage
+            .get_job(job_id)
+            .expect("get_job")
+            .expect("job exists");
+        if job.status == JobStatus::Complete {
+            return job;
+        }
+        assert!(Instant::now() < deadline, "job did not complete in time");
+        std::thread::sleep(Duration::from_millis(20));
+    }
+}
+
+fn main() {
+    let storage = StorageBackend::Sqlite(SqliteStorage::new("hello_taskito.db").expect("storage"));
+
+    let handle = Worker::new(storage.clone())
+        .num_workers(2)
+        .register("greet", |job: &Job| {
+            let name = String::from_utf8(job.payload.clone())
+                .map_err(|invalid| TaskError::fatal(format!("payload not utf-8: {invalid}")))?;
+            println!("hello, {name}!");
+            Ok(Some(format!("greeted {name}").into_bytes()))
+        })
+        .register_async("greet_async", |job: Job| async move {
+            let name = String::from_utf8(job.payload).unwrap_or_else(|_| "?".to_string());
+            println!("hello (async), {name}!");
+            Ok(None)
+        })
+        .spawn()
+        .expect("worker spawn");
+
+    let sync_job = storage
+        .enqueue(new_job("greet", b"world"))
+        .expect("enqueue");
+    let async_job = storage
+        .enqueue(new_job("greet_async", b"tokio"))
+        .expect("enqueue");
+
+    let done = wait_until_completed(&storage, &sync_job.id);
+    println!(
+        "sync result: {}",
+        String::from_utf8_lossy(done.result.as_deref().unwrap_or_default())
+    );
+    wait_until_completed(&storage, &async_job.id);
+
+    handle.shutdown().expect("clean shutdown");
+    println!("both jobs completed.");
+    let _ = std::fs::remove_file("hello_taskito.db");
+}

crates/taskito-core/src/lib.rs

@@ -32,4 +32,6 @@ pub use storage::sqlite::SqliteStorage;
 pub use storage::Storage;
 pub use storage::StorageBackend;
 pub use storage::{DeadJob, QueueStats, SubscriptionBacklogStats};
-pub use worker::WorkerDispatcher;
+pub use worker::{
+    NativeDispatcher, TaskError, TaskRegistry, TaskResult, Worker, WorkerDispatcher, WorkerHandle,
+};

crates/taskito-core/src/scheduler/mod.rs

@@ -199,6 +199,20 @@ pub struct TaskConfig {
     pub max_in_flight_per_task: Option<usize>,
 }
 
+impl Default for TaskConfig {
+    /// Default retry policy, no rate limit, breaker, retry budget, or caps.
+    fn default() -> Self {
+        Self {
+            retry_policy: RetryPolicy::default(),
+            rate_limit: None,
+            circuit_breaker: None,
+            retry_budget: None,
+            max_concurrent: None,
+            max_in_flight_per_task: None,
+        }
+    }
+}
+
 /// In-flight bookkeeping behind the dispatch caps: which jobs are out, and how many
 /// of each task. Per-task counts are maintained alongside rather than derived so the
 /// per-task gate stays O(1) — the poller consults it on every dispatch, and the map

crates/taskito-core/src/worker/dispatcher.rs

@@ -0,0 +1,121 @@
+//! Reference [`WorkerDispatcher`]: executes registered Rust handlers.
+//!
+//! Mirrors the language bindings' pools — bounded concurrency via a semaphore,
+//! blocking handlers on `spawn_blocking`, async handlers on the runtime — with
+//! a [`TaskRegistry`] in place of a foreign-object registry. Task timeouts are
+//! detected server-side by the scheduler's stale-job reap; cancellation relies
+//! on the storage cancel flag, so `notify_cancel` is a no-op here.
+
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+use std::time::Instant;
+
+use async_trait::async_trait;
+use crossbeam_channel::Sender;
+use tokio::sync::Semaphore;
+
+use super::registry::{TaskHandler, TaskRegistry, TaskResult};
+use super::WorkerDispatcher;
+use crate::job::Job;
+use crate::scheduler::JobResult;
+
+/// Dispatches dequeued jobs to handlers from a [`TaskRegistry`], at most
+/// `num_workers` concurrently.
+pub struct NativeDispatcher {
+    registry: Arc<TaskRegistry>,
+    num_workers: usize,
+    shutdown: AtomicBool,
+}
+
+impl NativeDispatcher {
+    pub fn new(registry: TaskRegistry, num_workers: usize) -> Self {
+        Self {
+            registry: Arc::new(registry),
+            num_workers: num_workers.max(1),
+            shutdown: AtomicBool::new(false),
+        }
+    }
+}
+
+/// Build the [`JobResult`] for a finished handler invocation.
+fn job_result(job: &Job, outcome: TaskResult, started: Instant) -> JobResult {
+    let wall_time_ns: i64 = started.elapsed().as_nanos().try_into().unwrap_or(i64::MAX);
+    match outcome {
+        Ok(result) => JobResult::Success {
+            job_id: job.id.clone(),
+            result,
+            task_name: job.task_name.clone(),
+            wall_time_ns,
+        },
+        Err(task_error) => JobResult::Failure {
+            job_id: job.id.clone(),
+            error: task_error.message,
+            retry_count: job.retry_count,
+            max_retries: job.max_retries,
+            task_name: job.task_name.clone(),
+            wall_time_ns,
+            should_retry: task_error.retryable,
+            // Timeouts are synthesized server-side by the stale-job reap.
+            timed_out: false,
+        },
+    }
+}
+
+#[async_trait]
+impl WorkerDispatcher for NativeDispatcher {
+    async fn run(
+        &self,
+        mut job_rx: tokio::sync::mpsc::Receiver<Job>,
+        result_tx: Sender<JobResult>,
+    ) {
+        let semaphore = Arc::new(Semaphore::new(self.num_workers));
+
+        while let Some(job) = job_rx.recv().await {
+            if self.shutdown.load(Ordering::Relaxed) {
+                break;
+            }
+            let permit = match semaphore.clone().acquire_owned().await {
+                Ok(permit) => permit,
+                Err(_) => break, // Semaphore closed.
+            };
+
+            let handler = match self.registry.get(&job.task_name) {
+                Some(handler) => handler.clone(),
+                None => {
+                    // Fatal, not retryable: no amount of retrying makes an
+                    // unregistered task runnable on this worker.
+                    let error = super::registry::TaskError::fatal(format!(
+                        "task not registered: {}",
+                        job.task_name
+                    ));
+                    let _ = result_tx.send(job_result(&job, Err(error), Instant::now()));
+                    continue;
+                }
+            };
+
+            let tx = result_tx.clone();
+            match handler {
+                TaskHandler::Sync(run) => {
+                    tokio::task::spawn_blocking(move || {
+                        let _permit = permit; // Hold the slot until done.
+                        let started = Instant::now();
+                        let outcome = run(&job);
+                        let _ = tx.send(job_result(&job, outcome, started));
+                    });
+                }
+                TaskHandler::Async(make_future) => {
+                    tokio::spawn(async move {
+                        let _permit = permit;
+                        let started = Instant::now();
+                        let outcome = make_future(job.clone()).await;
+                        let _ = tx.send(job_result(&job, outcome, started));
+                    });
+                }
+            }
+        }
+    }
+
+    fn shutdown(&self) {
+        self.shutdown.store(true, Ordering::Relaxed);
+    }
+}

crates/taskito-core/src/worker.rs → crates/taskito-core/src/worker/mod.rs

@@ -1,11 +1,20 @@
+pub mod dispatcher;
+pub mod registry;
+pub mod runner;
+
+pub use dispatcher::NativeDispatcher;
+pub use registry::{TaskError, TaskHandler, TaskRegistry, TaskResult};
+pub use runner::{Worker, WorkerHandle};
+
 use async_trait::async_trait;
 use crossbeam_channel::Sender;
 
 use crate::job::Job;
 use crate::scheduler::JobResult;
 
 /// Abstraction for worker pool implementations.
-/// Core defines the interface; language-specific crates provide implementations.
+/// Core defines the interface; the [`NativeDispatcher`] runs registered Rust
+/// handlers, and the language bindings provide their own pools.
 #[async_trait]
 pub trait WorkerDispatcher: Send + Sync {
     /// Start dispatching jobs from the receiver. Runs until channel closes.
@@ -23,3 +32,6 @@ pub trait WorkerDispatcher: Send + Sync {
     /// storage.
     fn notify_cancel(&self, _job_id: &str) {}
 }
+
+#[cfg(test)]
+mod tests;

crates/taskito-core/src/worker/registry.rs

@@ -0,0 +1,111 @@
+//! Task registry: maps a job's `task_name` to the Rust handler that runs it.
+//!
+//! The language bindings keep their own registries (a Python dict, a JS map);
+//! this one exists so a pure-Rust consumer has a first-class way to say
+//! "`send_email` means this function".
+
+use std::collections::HashMap;
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::Arc;
+
+use crate::job::Job;
+
+/// What a handler produced: optional result bytes (stored on the job and
+/// readable via `get_job`), or a [`TaskError`].
+pub type TaskResult = std::result::Result<Option<Vec<u8>>, TaskError>;
+
+/// A boxed future returned by async handlers.
+pub type TaskFuture = Pin<Box<dyn Future<Output = TaskResult> + Send + 'static>>;
+
+/// A failed task execution. `retryable` feeds the scheduler's retry decision:
+/// a retryable failure follows the job's retry policy; a fatal one goes
+/// straight to the dead-letter queue.
+#[derive(Debug, Clone)]
+pub struct TaskError {
+    pub message: String,
+    pub retryable: bool,
+}
+
+impl TaskError {
+    /// A failure the scheduler may retry (subject to the job's retry policy).
+    pub fn retryable(message: impl Into<String>) -> Self {
+        TaskError {
+            message: message.into(),
+            retryable: true,
+        }
+    }
+
+    /// A failure that must not be retried — the job dead-letters immediately.
+    pub fn fatal(message: impl Into<String>) -> Self {
+        TaskError {
+            message: message.into(),
+            retryable: false,
+        }
+    }
+}
+
+impl std::fmt::Display for TaskError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(&self.message)
+    }
+}
+
+impl std::error::Error for TaskError {}
+
+/// A registered handler: blocking (run on a blocking thread) or async
+/// (spawned on the worker's tokio runtime).
+#[derive(Clone)]
+pub enum TaskHandler {
+    Sync(Arc<dyn Fn(&Job) -> TaskResult + Send + Sync>),
+    Async(Arc<dyn Fn(Job) -> TaskFuture + Send + Sync>),
+}
+
+/// Name → handler map consulted by the dispatcher for every dequeued job.
+#[derive(Clone, Default)]
+pub struct TaskRegistry {
+    handlers: HashMap<String, TaskHandler>,
+}
+
+impl TaskRegistry {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Register a blocking handler for `task_name`. It runs on a dedicated
+    /// blocking thread, so it may block freely (I/O, CPU work).
+    pub fn register(
+        &mut self,
+        task_name: impl Into<String>,
+        handler: impl Fn(&Job) -> TaskResult + Send + Sync + 'static,
+    ) {
+        self.handlers
+            .insert(task_name.into(), TaskHandler::Sync(Arc::new(handler)));
+    }
+
+    /// Register an async handler for `task_name`. The future is spawned on
+    /// the worker's runtime, so it must not block the executor.
+    pub fn register_async<F, Fut>(&mut self, task_name: impl Into<String>, handler: F)
+    where
+        F: Fn(Job) -> Fut + Send + Sync + 'static,
+        Fut: Future<Output = TaskResult> + Send + 'static,
+    {
+        self.handlers.insert(
+            task_name.into(),
+            TaskHandler::Async(Arc::new(move |job| Box::pin(handler(job)))),
+        );
+    }
+
+    pub fn get(&self, task_name: &str) -> Option<&TaskHandler> {
+        self.handlers.get(task_name)
+    }
+
+    /// Names of every registered task.
+    pub fn task_names(&self) -> impl Iterator<Item = &str> {
+        self.handlers.keys().map(String::as_str)
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.handlers.is_empty()
+    }
+}