Build a Supervised Job Queue in Rust
Build a self-healing actor system with supervised workers using atomr.
Introduction
30 Second Summary
When your app runs dozens of background tasks at once, a single crash can bring everything down. The question is whether your system can heal itself automatically or whether you have to manually restart things every time something goes wrong.
In this project, you will build a supervised job queue in Rust using the atomr actor framework. A manager actor distributes compute jobs across a pool of workers, and when a worker crashes, the supervisor automatically restarts it without disrupting the rest of the system.
What You'll Build
A self-healing job processing system where a Manager actor supervises Worker actors, distributes math computations via round-robin routing, and recovers from deliberate crashes through automatic restarts.
By the end of this project, you'll have:
- A Manager actor that spawns and supervises a pool of Worker actors using a OneForOne restart strategy.
- A working request/reply pattern that returns computed results from workers back to the caller via typed messages.
- Structured terminal output showing the full actor lifecycle: pre_start, handle, pre_restart, post_restart, and post_stop.
- Secret Mission: a dead-letter observer that logs messages sent to stopped actors.
Are there any prerequisites?
This project assumes intermediate Rust knowledge. You should be comfortable with async/await, enums, traits, and basic Tokio usage before starting.
Before We Start
Before We Start...
Set Up the Project
To build a system where actors can crash and automatically recover, you first need a Rust project with the right foundation.
In this step, you'll scaffold a new Rust binary project and wire up three crates: atomr for the actor framework, tokio for the async runtime, and async-trait for async trait methods.
In this step, get ready to:
- Create a new Rust binary project.
- Add the atomr, tokio, and async-trait dependencies.
- Verify the project compiles cleanly.
Create a new Rust binary project
- Open your terminal.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your project with the 2021 edition by running this command:
cargo new --edition 2021 atomr-job-queueYou should see a message that says Creating binary (application) `atomr-job-queue` package.
- Move into the new project folder by running:
cd atomr-job-queue- Confirm the project files exist by running:
lsYou should see Cargo.toml and a src folder in the output.
Don't see the project files?
Make sure you ran cargo new from your Desktop. If the command wasn't found, your Rust toolchain may not be on your PATH.
Try running rustc --version to confirm Rust is installed and accessible.
help me troubleshoot why cargo new isn't working on my machine
Add dependencies to Cargo.toml
The Cargo.toml file is your project's manifest. It declares metadata and lists all dependencies that cargo downloads and compiles for you.
- Open the atomr-job-queue folder in your editor.
- Open Cargo.toml and replace its entire contents with the following:
[package]
name = "atomr-job-queue"
version = "0.1.0"
edition = "2021"
[dependencies]
atomr = { version = "0.9", features = ["macros"] }
async-trait = "0.1"
tokio = { version = "1.39", features = ["full"] }What do these dependencies do?
- atomr is the actor framework. The macros feature enables derive macros for actor boilerplate.
- async-trait lets you write async fn in trait implementations. The Actor trait requires this.
- tokio is the async runtime that powers the actor system under the hood. The full feature enables all runtime components (timers, I/O, channels).
- Save Cargo.toml.
Seeing a red underline in your editor?
Some editors show errors until the dependencies are actually downloaded. This is normal. The next substep runs cargo build which fetches and compiles everything.
help me fix a Cargo.toml error in my atomr project
Verify the project compiles
Running cargo build downloads every dependency, compiles them, and links your binary. A successful build proves the toolchain and all three crates are working together.
- In your terminal, make sure you're inside the atomr-job-queue folder, then build the project by running:
cargo buildYou should see several Compiling lines as each crate is fetched and compiled, ending with a Finished line. This first build may take a minute or two.
Build failed with an error?
Double check that your Cargo.toml matches the code above exactly. A missing quote or bracket will cause a parse error.
If you see a network error, make sure you have an internet connection. cargo needs to download crates from crates.io on the first build.
help me fix my cargo build error for the atomr-job-queue project
✔️ Awesome, I've got everything!
Great. Double check you've saved Cargo.toml and your build finished without errors.
ⓧ I'd like to double check the full code
Here's what your Cargo.toml should look like:
[package]
name = "atomr-job-queue"
version = "0.1.0"
edition = "2021"
[dependencies]
atomr = { version = "0.9", features = ["macros"] }
async-trait = "0.1"
tokio = { version = "1.39", features = ["full"] }Your project skeleton is ready and all dependencies compile. Next up, you'll define the message types that actors use to communicate with each other.
Define Your Message Types
Your project compiles and is ready for code. Now it's time to define how your actors will communicate.
In the actor model, actors never share memory directly. They talk to each other exclusively through typed messages. In this step, you'll define the message enums that your Worker and Manager actors will use to exchange work requests and results.
In this step, get ready to:
- Define a WorkerMsg enum for compute requests and crash simulation.
- Define a ManagerMsg enum for submitting jobs to the manager.
Replace the hello-world stub with imports and WorkerMsg
Actors in atomr receive messages through a typed mailbox. Each actor declares an associated Msg type that defines exactly which messages it can handle. You'll use Rust enums to model these messages, giving you compile-time guarantees that actors only receive valid inputs.
- Open src/main.rs in your editor.
- Delete all the existing hello-world code (the fn main() with println!).
- Replace it with the following imports and WorkerMsg enum:
use std::time::Duration;
use async_trait::async_trait;
use atomr::prelude::*;
use tokio::sync::oneshot;
// ---------- Message types ----------
enum WorkerMsg {
Compute {
value: u64,
reply: oneshot::Sender<u64>,
},
Poison,
}What does this code do?
- use atomr::prelude::* imports core actor types like Actor, ActorRef, Props, Context, and supervision strategies.
- use tokio::sync::oneshot gives you a one-shot channel for request/reply patterns. The sender is embedded in the message so the worker can reply directly.
- WorkerMsg::Compute carries a value to compute and a reply channel for sending the result back.
- WorkerMsg::Poison is a "poison pill" that will deliberately crash the worker, letting you observe supervision in action later.
- Save src/main.rs.
- Run cargo build in your terminal. The build succeeds, though you'll see warnings about unused imports. That's expected since nothing uses them yet.
Seeing a compile error instead of warnings?
- Make sure you deleted all the original hello-world code. There should be no fn main() in the file yet.
- Check that your Cargo.toml lists atomr, async-trait, and tokio as dependencies. Without them, the use statements won't resolve.
- Verify the enum syntax has a comma after the Poison variant's closing brace and before the closing } of the enum.
Why won't my Rust project compile after adding the message types?
Add the ManagerMsg enum
The Manager actor needs its own message type to accept job submissions from the outside world. It follows the same request/reply pattern as WorkerMsg.
- In src/main.rs, add the following enum below the WorkerMsg definition:
enum ManagerMsg {
Submit {
value: u64,
reply: oneshot::Sender<u64>,
},
}What does this code do?
ManagerMsg::Submit is the message external code sends to the Manager to request a computation. It carries the value to compute and a reply channel. The Manager will forward this work to one of its child Workers using round-robin routing.
Why do the message types need Send + 'static?
atomr requires type Msg: Send + 'static on every actor's associated message type. This ensures messages can safely cross async task boundaries.
Your enums satisfy these bounds automatically because all their fields (u64 and oneshot::Sender<u64>) are themselves Send + 'static. No explicit impl is needed.
- Save src/main.rs.
- Confirm it compiles by running:
cargo build- You should see warnings about unused imports and dead code, but no errors. The build completes successfully.
Seeing an error about missing fields?
- Make sure ManagerMsg uses struct-style variants with curly braces { value: u64, reply: ... }, not tuple-style (u64, ...).
- Confirm the ManagerMsg enum appears after the WorkerMsg enum, not inside it.
Why is cargo build showing errors for my ManagerMsg enum?
✔️ Awesome, I've got everything!
Great. Double check you've saved src/main.rs and move on.
ⓧ I'd like to double check the full code
Here is the complete src/main.rs at this point:
use std::time::Duration;
use async_trait::async_trait;
use atomr::prelude::*;
use tokio::sync::oneshot;
// ---------- Message types ----------
enum WorkerMsg {
Compute {
value: u64,
reply: oneshot::Sender<u64>,
},
Poison,
}
enum ManagerMsg {
Submit {
value: u64,
reply: oneshot::Sender<u64>,
},
}Your message types are locked in. Next up, you'll build the Worker actor that receives these messages and actually does the work.
Build the Worker Actor
Your message types are defined. Now it's time to bring those messages to life by building the actor that receives them.
In atomr, an actor is a struct that implements the Actor trait. The trait gives you lifecycle hooks that fire at key moments: when the actor starts, when it receives a message, when it crashes, and when it restarts. You'll implement all of these for the Worker so you can observe exactly what happens at each stage.
In this step, get ready to:
- Create the Worker struct and implement the Actor trait with message handling.
- Add lifecycle hooks that log each stage of the actor's life.
- Write a helper function that workers use for computation.
Create the Worker struct
Every actor in atomr needs a struct to hold its internal state. The Worker only needs an id so it can identify itself in log output.
- In src/main.rs, find the closing brace of the ManagerMsg enum. Add the following code below it:
// ---------- Worker actor ----------
struct Worker {
id: usize,
}
impl Worker {
fn new(id: usize) -> Self {
Self { id }
}
}What does this code do?
The Worker struct holds a single field: id. This lets each worker print which instance is handling work or crashing.
The new() constructor is important because atomr's Props::create takes a closure that builds a fresh actor instance. Every time the supervisor restarts a crashed worker, it calls this closure again to get a clean Worker.
Implement the Actor trait for Worker
The Actor trait is where your actor's behavior lives. You define what message type it accepts, how it handles each message, and what it does at each lifecycle stage.
- Directly below the impl Worker block you just added, paste this Actor implementation:
#[async_trait]
impl Actor for Worker {
type Msg = WorkerMsg;
async fn pre_start(&mut self, _ctx: &mut Context<Self>) {
println!("[Worker {}] pre_start: ready for work", self.id);
}
async fn handle(&mut self, _ctx: &mut Context<Self>, msg: WorkerMsg) {
match msg {
WorkerMsg::Compute { value, reply } => {
println!("[Worker {}] computing fib({})", self.id, value);
let result = fib(value);
let _ = reply.send(result);
}
WorkerMsg::Poison => {
println!("[Worker {}] received poison pill, crashing!", self.id);
panic!("deliberate crash");
}
}
}
async fn pre_restart(&mut self, _ctx: &mut Context<Self>, err: &str) {
println!("[Worker {}] pre_restart: failed with '{}'", self.id, err);
}
async fn post_restart(&mut self, _ctx: &mut Context<Self>, _err: &str) {
println!("[Worker {}] post_restart: back online", self.id);
}
async fn post_stop(&mut self, _ctx: &mut Context<Self>) {
println!("[Worker {}] post_stop: shutting down", self.id);
}
}What does this code do?
- type Msg = WorkerMsg tells atomr this actor's mailbox accepts WorkerMsg values only. Messages are statically typed.
- pre_start fires once when the actor is first spawned. You'll see this in the logs for each worker.
- handle is the main message loop. For Compute, it calculates a Fibonacci number and sends the result back through the oneshot channel. For Poison, it panics deliberately to simulate a crash.
- pre_restart fires after a crash but before the supervisor creates a fresh instance. The err parameter contains the panic message.
- post_restart fires after the fresh instance is ready. This confirms recovery succeeded.
- post_stop fires when the actor shuts down permanently during system termination.
Why panic instead of returning an error?
In atomr, a panic inside handle is caught by the runtime and forwarded to the supervisor. The supervisor then decides what to do: restart, stop, or escalate.
This is the same pattern used in production actor systems to model unexpected failures that should not bring down the entire application.
Add the Fibonacci helper function
Workers need something to compute. A simple iterative Fibonacci function gives visible work without external dependencies.
- Below the closing brace of the impl Actor for Worker block, add this helper function:
// ---------- Helper ----------
fn fib(n: u64) -> u64 {
match n {
0 => 0,
1 => 1,
_ => {
let mut a = 0u64;
let mut b = 1u64;
for _ in 2..=n {
let tmp = a + b;
a = b;
b = tmp;
}
b
}
}
}What does this code do?
This is a straightforward iterative Fibonacci calculator. It avoids recursion so it won't blow the stack for larger inputs. The function takes a u64 and returns the nth Fibonacci number by accumulating values in a loop.
- Save src/main.rs.
- Verify the project still compiles by running:
cargo buildYou should see a successful compilation with no errors. You may see warnings about unused imports or dead code. That's expected because you haven't wired the Manager or main function yet.
Seeing a compilation error?
- Check that #[async_trait] appears directly above impl Actor for Worker. Without it, the async methods won't compile.
- Make sure fib is defined outside the impl Actor for Worker block. It's a standalone function, not a method on Worker.
- Confirm your WorkerMsg enum matches exactly what you defined in the previous step, including the reply: oneshot::Sender<u64> field.
Help me debug a compilation error in my Worker actor implementation.
✔️ Awesome, I've got everything!
Great. Double check you've saved src/main.rs.
ⓧ I'd like to double check the full code
Here's the complete src/main.rs at this point:
use std::time::Duration;
use async_trait::async_trait;
use atomr::prelude::*;
use tokio::sync::oneshot;
// ---------- Message types ----------
enum WorkerMsg {
Compute {
value: u64,
reply: oneshot::Sender<u64>,
},
Poison,
}
enum ManagerMsg {
Submit {
value: u64,
reply: oneshot::Sender<u64>,
},
}
// ---------- Worker actor ----------
struct Worker {
id: usize,
}
impl Worker {
fn new(id: usize) -> Self {
Self { id }
}
}
#[async_trait]
impl Actor for Worker {
type Msg = WorkerMsg;
async fn pre_start(&mut self, _ctx: &mut Context<Self>) {
println!("[Worker {}] pre_start: ready for work", self.id);
}
async fn handle(&mut self, _ctx: &mut Context<Self>, msg: WorkerMsg) {
match msg {
WorkerMsg::Compute { value, reply } => {
println!("[Worker {}] computing fib({})", self.id, value);
let result = fib(value);
let _ = reply.send(result);
}
WorkerMsg::Poison => {
println!("[Worker {}] received poison pill, crashing!", self.id);
panic!("deliberate crash");
}
}
}
async fn pre_restart(&mut self, _ctx: &mut Context<Self>, err: &str) {
println!("[Worker {}] pre_restart: failed with '{}'", self.id, err);
}
async fn post_restart(&mut self, _ctx: &mut Context<Self>, _err: &str) {
println!("[Worker {}] post_restart: back online", self.id);
}
async fn post_stop(&mut self, _ctx: &mut Context<Self>) {
println!("[Worker {}] post_stop: shutting down", self.id);
}
}
// ---------- Helper ----------
fn fib(n: u64) -> u64 {
match n {
0 => 0,
1 => 1,
_ => {
let mut a = 0u64;
let mut b = 1u64;
for _ in 2..=n {
let tmp = a + b;
a = b;
b = tmp;
}
b
}
}
}Your Worker actor is complete. It can receive compute jobs, reply with results, crash on command, and log every stage of its life. Next up, you'll build the Manager actor that spawns these workers and supervises them.
Build the Manager Actor with Supervision
You have workers that can process jobs and crash gracefully. But a pile of independent workers is useless without someone coordinating them.
In the actor model, a supervisor is an actor whose sole job is to watch its children, route work to them, and restart them when they fail. Your Manager actor fills this role. It spawns a pool of Workers on startup, distributes incoming jobs using round-robin, and defines a restart strategy so crashed workers come back automatically.
In this step, get ready to:
- Define the Manager struct that tracks its worker pool and routing state.
- Implement the Actor trait so the Manager spawns workers, routes messages, and supervises failures.
Define the Manager struct and constructor
The Manager needs to track three things: the pool of worker references it supervises, the index for round-robin routing, and the desired pool size.
- In src/main.rs, find the // ---------- Helper ---------- comment.
- Add the following code above that comment to define the Manager struct and its constructor:
// ---------- Manager actor ----------
struct Manager {
workers: Vec<ActorRef<WorkerMsg>>,
next: usize,
pool_size: usize,
}
impl Manager {
fn new(pool_size: usize) -> Self {
Self {
workers: Vec::new(),
next: 0,
pool_size,
}
}
}What does this code do?
- workers holds ActorRef<WorkerMsg> handles for each child Worker. You use these to send messages to specific workers.
- next is the round-robin counter. Each time a job arrives, you pick workers[next % len] and increment.
- pool_size stores how many workers to spawn. The constructor starts with an empty Vec because workers are spawned later in the pre_start lifecycle hook.
Why spawn in pre_start instead of new()?
Spawning actors requires a Context, which only exists once the actor is running inside the system. The constructor runs before the actor is registered, so it has no context to spawn children with. The pre_start hook is the earliest point where the actor has a live context.
Implement the Actor trait for Manager
The Actor impl ties together spawning, message routing, and supervision in one block. This is where the Manager comes alive.
- Below the Manager::new constructor, add the full Actor trait implementation:
#[async_trait]
impl Actor for Manager {
type Msg = ManagerMsg;
async fn pre_start(&mut self, ctx: &mut Context<Self>) {
println!("[Manager] pre_start: spawning {} workers", self.pool_size);
for i in 0..self.pool_size {
let id = i;
let worker = ctx
.spawn(Props::create(move || Worker::new(id)), &format!("worker-{}", i))
.expect("failed to spawn worker");
self.workers.push(worker);
}
println!("[Manager] all workers spawned");
}
async fn handle(&mut self, _ctx: &mut Context<Self>, msg: ManagerMsg) {
match msg {
ManagerMsg::Submit { value, reply } => {
let worker = &self.workers[self.next % self.workers.len()];
self.next += 1;
worker.tell(WorkerMsg::Compute { value, reply });
}
}
}
fn supervisor_strategy(&self) -> SupervisorStrategy {
OneForOneStrategy::new()
.with_max_retries(3)
.with_within(Duration::from_secs(60))
.with_decider(|err| {
println!("[Supervisor] child failed: '{}' -> Restart", err);
Directive::Restart
})
.into()
}
}What does this code do?
This impl has three responsibilities:
- pre_start loops from 0 to pool_size, calling ctx.spawn() to create each Worker as a supervised child. Props::create wraps a closure factory. The supervisor calls this factory again whenever it restarts a crashed child, producing a fresh Worker instance.
- handle receives ManagerMsg::Submit and forwards it to the next worker in round-robin order using tell. Fire-and-forget: the Manager does not wait for the worker to finish.
- supervisor_strategy returns a OneForOneStrategy that restarts only the failed child (not the entire pool). It allows up to 3 retries within 60 seconds before giving up. The decider closure logs the failure reason and returns Directive::Restart.
What is OneForOne supervision?
OneForOne means each child is supervised independently. If worker-0 panics, only worker-0 restarts. Workers 1 and 2 continue processing jobs without interruption.
The alternative is AllForOne, where a single child failing triggers a restart of the entire pool. OneForOne is the right choice here because your workers are independent and share no state.
- Save src/main.rs.
- Verify the project still compiles by running:
cargo buildYou should see a successful compilation with no errors. You may see unused-variable warnings for the Manager since nothing uses it yet.
Seeing a compilation error?
- If you see a type mismatch on ActorRef<WorkerMsg>, make sure your WorkerMsg enum is defined above the Manager struct in the file.
- If ctx.spawn is not found, confirm your import line reads use atomr::prelude::*; at the top of the file.
- If Duration is unresolved, check that use std::time::Duration; is present.
Help me debug a compilation error in my Manager Actor impl
✔️ Awesome, I've got everything!
Great. Double check you have saved src/main.rs and the build succeeded.
ⓧ I'd like to double check the full code
use std::time::Duration;
use async_trait::async_trait;
use atomr::prelude::*;
use tokio::sync::oneshot;
// ---------- Message types ----------
enum WorkerMsg {
Compute {
value: u64,
reply: oneshot::Sender<u64>,
},
Poison,
}
enum ManagerMsg {
Submit {
value: u64,
reply: oneshot::Sender<u64>,
},
}
// ---------- Worker actor ----------
struct Worker {
id: usize,
}
impl Worker {
fn new(id: usize) -> Self {
Self { id }
}
}
#[async_trait]
impl Actor for Worker {
type Msg = WorkerMsg;
async fn pre_start(&mut self, _ctx: &mut Context<Self>) {
println!("[Worker {}] pre_start: ready for work", self.id);
}
async fn handle(&mut self, _ctx: &mut Context<Self>, msg: WorkerMsg) {
match msg {
WorkerMsg::Compute { value, reply } => {
println!("[Worker {}] computing fib({})", self.id, value);
let result = fib(value);
let _ = reply.send(result);
}
WorkerMsg::Poison => {
println!("[Worker {}] received poison pill, crashing!", self.id);
panic!("deliberate crash");
}
}
}
async fn pre_restart(&mut self, _ctx: &mut Context<Self>, err: &str) {
println!("[Worker {}] pre_restart: failed with '{}'", self.id, err);
}
async fn post_restart(&mut self, _ctx: &mut Context<Self>, _err: &str) {
println!("[Worker {}] post_restart: back online", self.id);
}
async fn post_stop(&mut self, _ctx: &mut Context<Self>) {
println!("[Worker {}] post_stop: shutting down", self.id);
}
}
// ---------- Manager actor ----------
struct Manager {
workers: Vec<ActorRef<WorkerMsg>>,
next: usize,
pool_size: usize,
}
impl Manager {
fn new(pool_size: usize) -> Self {
Self {
workers: Vec::new(),
next: 0,
pool_size,
}
}
}
#[async_trait]
impl Actor for Manager {
type Msg = ManagerMsg;
async fn pre_start(&mut self, ctx: &mut Context<Self>) {
println!("[Manager] pre_start: spawning {} workers", self.pool_size);
for i in 0..self.pool_size {
let id = i;
let worker = ctx
.spawn(Props::create(move || Worker::new(id)), &format!("worker-{}", i))
.expect("failed to spawn worker");
self.workers.push(worker);
}
println!("[Manager] all workers spawned");
}
async fn handle(&mut self, _ctx: &mut Context<Self>, msg: ManagerMsg) {
match msg {
ManagerMsg::Submit { value, reply } => {
let worker = &self.workers[self.next % self.workers.len()];
self.next += 1;
worker.tell(WorkerMsg::Compute { value, reply });
}
}
}
fn supervisor_strategy(&self) -> SupervisorStrategy {
OneForOneStrategy::new()
.with_max_retries(3)
.with_within(Duration::from_secs(60))
.with_decider(|err| {
println!("[Supervisor] child failed: '{}' -> Restart", err);
Directive::Restart
})
.into()
}
}
// ---------- Helper ----------
fn fib(n: u64) -> u64 {
match n {
0 => 0,
1 => 1,
_ => {
let mut a = 0u64;
let mut b = 1u64;
for _ in 2..=n {
let tmp = a + b;
a = b;
b = tmp;
}
b
}
}
}Your Manager is ready to supervise. Next up, you will wire everything together in main and watch the full lifecycle play out in your terminal.
Wire Up main and Observe the System
Your Manager and Worker actors are fully implemented with supervision logic. Now you need an entry point that boots the entire system and puts it through its paces.
In this step, you will write the main function that creates an ActorSystem, spawns the Manager, sends compute jobs, and shuts everything down cleanly.
In this step, get ready to:
- Create an ActorSystem and spawn the Manager as a top-level actor.
- Send compute jobs via oneshot channels and print results.
- Run the system and observe the full actor lifecycle in your terminal.
Create the ActorSystem and spawn the Manager
The ActorSystem is the runtime container for all your actors. It manages mailboxes, dispatching, and lifecycle. You will create one, then spawn the Manager as a top-level actor inside it.
- In src/main.rs, scroll to the bottom of the file (below the closing brace of the fib function) and add the main function by pasting the following code:
// ---------- Main ----------
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let system = ActorSystem::create("job-queue", Config::empty()).await?;
let pool_size = 3;
let manager = system.actor_of(
Props::create(move || Manager::new(pool_size)),
"manager",
)?;
// Give actors time to start
tokio::time::sleep(Duration::from_millis(100)).await;What does this code do?
ActorSystem::create("job-queue", Config::empty()) boots a new actor runtime named job-queue with default configuration.
system.actor_of(Props::create(...), "manager") spawns the Manager as a top-level actor. The Props::create closure is a factory that produces a fresh Manager instance on start and on every supervisor-triggered restart.
The 100ms sleep gives the Manager's pre_start hook time to spawn all three Worker children before you start sending messages.
Send compute jobs and print results
To get results back from actors, you embed a oneshot channel sender inside each message. The worker sends its answer through that channel, and your main function awaits the receiver.
- Directly below the sleep line you just added, paste the first batch of compute jobs:
// Send compute jobs
println!("\n--- Sending compute jobs ---");
for val in [10, 20, 30, 15, 25] {
let (tx, rx) = oneshot::channel();
manager.tell(ManagerMsg::Submit { value: val, reply: tx });
match tokio::time::timeout(Duration::from_secs(2), rx).await {
Ok(Ok(result)) => println!("fib({}) = {}", val, result),
Ok(Err(_)) => println!("fib({}) = <sender dropped>", val),
Err(_) => println!("fib({}) = <timeout>", val),
}
}What does this code do?
- oneshot::channel() creates a one-shot sender/receiver pair. You embed tx inside the message so the Worker can reply.
- manager.tell(...) is fire-and-forget delivery. The Manager receives the message and routes it to the next Worker via round-robin.
- tokio::time::timeout wraps the rx future with a 2-second deadline so you never hang if a worker crashes before replying.
- The three match arms handle success, a dropped sender (worker panicked), and a timeout.
Complete the main function and run the system
After the first batch, you will add a second batch of jobs to demonstrate that the system remains healthy. Then you will shut everything down.
- Below the first for loop you just pasted, add the second batch and the shutdown sequence:
tokio::time::sleep(Duration::from_millis(200)).await;
// Send more jobs after the restart
println!("\n--- Sending more jobs (post-restart) ---");
for val in [5, 8, 12] {
let (tx, rx) = oneshot::channel();
manager.tell(ManagerMsg::Submit { value: val, reply: tx });
match tokio::time::timeout(Duration::from_secs(2), rx).await {
Ok(Ok(result)) => println!("fib({}) = {}", val, result),
Ok(Err(_)) => println!("fib({}) = <sender dropped>", val),
Err(_) => println!("fib({}) = <timeout>", val),
}
}
println!("\n--- Shutting down ---");
system.terminate().await;
Ok(())
}What does this code do?
- The 200ms sleep provides a buffer for any in-flight supervisor restarts to complete before the second batch arrives.
- The second loop sends three more fibonacci jobs. If a worker crashed and restarted in the interim, these jobs prove the system recovered.
- system.terminate().await triggers a coordinated shutdown. Every actor's post_stop hook fires, and you will see the lifecycle logs in your terminal.
- Save src/main.rs.
- Run the system by executing this command in your terminal:
cargo runYou should see structured lifecycle output in your terminal. The Manager's pre_start fires first, spawning three workers. Each Worker logs its own pre_start. Then the fibonacci results print. Finally, post_stop logs appear during shutdown.
Not seeing any output?
- Make sure you saved src/main.rs before running cargo run.
- Check that the #[tokio::main] attribute is directly above async fn main() with no blank lines between them.
- If you see a compile error about oneshot, confirm your imports at the top of the file include use tokio::sync::oneshot;.
Help me debug my cargo run output.
✔️ Awesome, I've got everything!
Great. Double check you have saved src/main.rs and that cargo run produces lifecycle logs and fibonacci results.
ⓧ I'd like to double check the full code
Here is the complete src/main.rs file for comparison:
use std::time::Duration;
use async_trait::async_trait;
use atomr::prelude::*;
use tokio::sync::oneshot;
// ---------- Message types ----------
enum WorkerMsg {
Compute {
value: u64,
reply: oneshot::Sender<u64>,
},
Poison,
}
enum ManagerMsg {
Submit {
value: u64,
reply: oneshot::Sender<u64>,
},
}
// ---------- Worker actor ----------
struct Worker {
id: usize,
}
impl Worker {
fn new(id: usize) -> Self {
Self { id }
}
}
#[async_trait]
impl Actor for Worker {
type Msg = WorkerMsg;
async fn pre_start(&mut self, _ctx: &mut Context<Self>) {
println!("[Worker {}] pre_start: ready for work", self.id);
}
async fn handle(&mut self, _ctx: &mut Context<Self>, msg: WorkerMsg) {
match msg {
WorkerMsg::Compute { value, reply } => {
println!("[Worker {}] computing fib({})", self.id, value);
let result = fib(value);
let _ = reply.send(result);
}
WorkerMsg::Poison => {
println!("[Worker {}] received poison pill, crashing!", self.id);
panic!("deliberate crash");
}
}
}
async fn pre_restart(&mut self, _ctx: &mut Context<Self>, err: &str) {
println!("[Worker {}] pre_restart: failed with '{}'", self.id, err);
}
async fn post_restart(&mut self, _ctx: &mut Context<Self>, _err: &str) {
println!("[Worker {}] post_restart: back online", self.id);
}
async fn post_stop(&mut self, _ctx: &mut Context<Self>) {
println!("[Worker {}] post_stop: shutting down", self.id);
}
}
// ---------- Manager actor ----------
struct Manager {
workers: Vec<ActorRef<WorkerMsg>>,
next: usize,
pool_size: usize,
}
impl Manager {
fn new(pool_size: usize) -> Self {
Self {
workers: Vec::new(),
next: 0,
pool_size,
}
}
}
#[async_trait]
impl Actor for Manager {
type Msg = ManagerMsg;
async fn pre_start(&mut self, ctx: &mut Context<Self>) {
println!("[Manager] pre_start: spawning {} workers", self.pool_size);
for i in 0..self.pool_size {
let id = i;
let worker = ctx
.spawn(Props::create(move || Worker::new(id)), &format!("worker-{}", i))
.expect("failed to spawn worker");
self.workers.push(worker);
}
println!("[Manager] all workers spawned");
}
async fn handle(&mut self, _ctx: &mut Context<Self>, msg: ManagerMsg) {
match msg {
ManagerMsg::Submit { value, reply } => {
let worker = &self.workers[self.next % self.workers.len()];
self.next += 1;
worker.tell(WorkerMsg::Compute { value, reply });
}
}
}
fn supervisor_strategy(&self) -> SupervisorStrategy {
OneForOneStrategy::new()
.with_max_retries(3)
.with_within(Duration::from_secs(60))
.with_decider(|err| {
println!("[Supervisor] child failed: '{}' -> Restart", err);
Directive::Restart
})
.into()
}
}
// ---------- Helper ----------
fn fib(n: u64) -> u64 {
match n {
0 => 0,
1 => 1,
_ => {
let mut a = 0u64;
let mut b = 1u64;
for _ in 2..=n {
let tmp = a + b;
a = b;
b = tmp;
}
b
}
}
}
// ---------- Main ----------
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let system = ActorSystem::create("job-queue", Config::empty()).await?;
let pool_size = 3;
let manager = system.actor_of(
Props::create(move || Manager::new(pool_size)),
"manager",
)?;
// Give actors time to start
tokio::time::sleep(Duration::from_millis(100)).await;
// Send compute jobs
println!("\n--- Sending compute jobs ---");
for val in [10, 20, 30, 15, 25] {
let (tx, rx) = oneshot::channel();
manager.tell(ManagerMsg::Submit { value: val, reply: tx });
match tokio::time::timeout(Duration::from_secs(2), rx).await {
Ok(Ok(result)) => println!("fib({}) = {}", val, result),
Ok(Err(_)) => println!("fib({}) = <sender dropped>", val),
Err(_) => println!("fib({}) = <timeout>", val),
}
}
tokio::time::sleep(Duration::from_millis(200)).await;
// Send more jobs after the restart
println!("\n--- Sending more jobs (post-restart) ---");
for val in [5, 8, 12] {
let (tx, rx) = oneshot::channel();
manager.tell(ManagerMsg::Submit { value: val, reply: tx });
match tokio::time::timeout(Duration::from_secs(2), rx).await {
Ok(Ok(result)) => println!("fib({}) = {}", val, result),
Ok(Err(_)) => println!("fib({}) = <sender dropped>", val),
Err(_) => println!("fib({}) = <timeout>", val),
}
}
println!("\n--- Shutting down ---");
system.terminate().await;
Ok(())
}Your actor system is alive. You have seen the Manager spawn workers, distribute jobs via round-robin, collect results through oneshot channels, and shut down cleanly with full lifecycle output. In the Secret Mission, you will push this further by triggering a deliberate crash and observing the supervisor restart a worker in real time.
Secret mission
Add a Dead-Letter Observer
In a real system, messages can arrive after an actor has already stopped. Without visibility into these lost messages, debugging becomes a nightmare. Your mission: implement a DeadLetterObserver that catches and logs every message sent to a dead actor.
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your project files for future reference or delete them entirely.
Resources you used:
- The atomr-job-queue project folder on your Desktop (includes source code and compiled artifacts in target/).
Keep everything running
No action needed. Choose this if you want to keep experimenting with the actor system or extend the project further.
- Your project folder stays on your Desktop with all source code and compiled binaries intact.
- You can run cargo run anytime to re-execute the actor system.
Pause - I'll come back to this later
Free up disk space by removing compiled artifacts while keeping your source code.
- Delete the target/ directory inside your project folder to reclaim disk space. Run this from inside the atomr-job-queue folder:
cargo clean- Your source files (Cargo.toml and src/main.rs) remain untouched. Run cargo build when you return to recompile.
Delete - I don't want to use this again
Remove all project files from your machine.
- Delete the entire project folder by running the command for your operating system.
macOS
cd ~/Desktop
rm -rf atomr-job-queueWindows
cd ~\Desktop
rmdir /s /q atomr-job-queue- Confirm the folder is gone by running ls (macOS) or dir (Windows) on your Desktop. You should no longer see atomr-job-queue listed.
Nice Work!
Nice Work!
You just built a self-healing actor system in Rust. Your Manager spawns workers, distributes jobs, and automatically recovers from crashes without dropping the rest of the system.
You've learned how to:
- Built a typed actor system with atomr where actors communicate via enum messages and reply through oneshot channels.
- Configured supervision with a OneForOneStrategy to automatically restart crashed actors without affecting healthy siblings.
- Observed the full actor lifecycle (pre_start, handle, pre_restart, post_restart, post_stop) through structured terminal logging.
- Secret Mission: Implemented a DeadLetterObserver to catch messages sent to stopped actors.
Ready to quiz yourself?