Actor Model: Using channels to access shared state

The central idea is surprisingly simple:

Don't communicate by sharing memory ; share memory by communicating.

Instead of allowing multiple parts of your application to directly access and modify shared state, give one independent task exclusive ownership of that state. Other parts of the application communicate with it by sending messages.

Synchronization

Other popular synchronization primitives like Mutex, RwLock, or atomic variables, instead the Actor Model offers another approach.

1. Traditional shared state approach:

2. Actor-based approach:

Fundamental Concepts:

An Actor consists of two parts:

1. The Actor Task
2. The Actor Handle

Actuall Rust Code:

1. Define the Actor

use tokio::sync::{mpsc, oneshot};

struct MyActor {
    // The message that it receives from the Actor Handle through the mpsc channel
    receiver: mpsc::Receiver<ActorMessage>,
    // Shared state that needs exclusive ownership
    next_id: u32,
}

2. Define the message

enum ActorMessage {
    GetUniqueId {
        // Channel through which the actor can return the generated ID.
        respond_to: oneshot::Sender<u32>,
    },
}

3. Message Handling

impl MyActor {
    fn new(receiver: mpsc::Receiver<ActorMessage>) -> Self {
        MyActor {
            receiver,
            next_id: 0,
        }
    }

    fn handle_message(&mut self, msg: ActorMessage) {
        match msg {
            ActorMessage::GetUniqueId { respond_to } => {
                self.next_id += 1;

                let _ = respond_to.send(self.next_id);
            }
        }
    }
}

4. Run the Actor continuously

Actor's main event loop:

async fn run_my_actor(mut actor: MyActor) {
    // wait for the next message
    while let Some(msg) = actor.receiver.recv().await {
        // continue as long as the messages are available & the channel remains open.
        actor.handle_message(msg);
    }
}

The Actor Handle: How other worker tasks communicate

#[derive(Clone)]
pub struct MyActorHandle {
    sender: mpsc::Sender<ActorMessage>,
}

Now creating the Actor:

impl MyActorHandle {
    pub fn new() -> Self {
        // Multiple Producers ; Single Consumer bounded channel(8 bounds) 
        let (sender, receiver) = mpsc::channel(8);

        // Now the actor owns the receiver & can process the incoming message
        let actor = MyActor::new(receiver);

        // Actor as an independent tokio task
        tokio::spawn(run_my_actor(actor));

        // Now caller can communicate with the actor without having access to its internal state
        Self { sender }
    }
}

Implement the request method

This method is where everything comes together.

pub async fn get_unique_id(&self) -> u32 {
    // send: The actor will use this to return the result.
    // recv: The caller will use this to receive the result. 
    let (send, recv) = oneshot::channel();

    let msg = ActorMessage::GetUniqueId {
        respond_to: send,
    };

    // .await -> Because this is a bounded channel, sending can wait if the channel is full.
    let _ = self.sender.send(msg).await;

    // The caller waits until the actor sends the ID.
    recv.await.expect("Actor task has been killed")
}

Finally: Actor Communication Sequence

An important distinction: the callers can execute concurrently, but this actor handles its messages one at a time in this implementation. This is why the counter doesn't need a mutex.

Rust Ownership and Borrowing

You might be tempted to put everything into one struct:

struct MyActor {
    receiver: mpsc::Receiver<ActorMessage>,
    sender: mpsc::Sender<ActorMessage>,
    next_id: u32,
}

And then spawn the actor from a method that borrows self. However, this creates ownership and lifetime problems.

impl MyActor {
    fn run(&mut self) {
        tokio::spawn(async move {
            while let Some(msg) = self.receiver.recv().await {
                // ...
            }
        });
    }
}

Why is this problematic?

1. Borrowed reference escape into an independently running task.

• tokio::spawn requires the spawned future to be 'static( means that the spawned task must own the data it needs )
• But &mut self is a borrow.

2. Exposing exclusive Actor fields

• To build multiple Handles, we have to clone the MyActor struct, risk exposing fields that are exclusive to the Actor.

Separating ownership:

Clean ownership boundary

Actor

• Owns next_id
• Owns receiver
• Processes messages
• Runs independently

Handle

• Owns message sender
• Can be cloned
• Sends messages
• Used by application code

Backpressure

Instead of allowing the queue to grow indefinitely, you push the pressure back to the producers with the help of a bounded queue. "If 100,000 tasks are waiting on .send().await, haven't we just moved the memory problem somewhere else?" A bounded channel limits the messages stored in the channel, but it does not automatically limit the number of tasks waiting to send.

The bounded channel is not where the memory problem is solved.
It is where the pressure is created. Something upstream has to absorb it.

Same problem, moved: if producers are unbounded (one spawned task per request, each parked on send().await), 100,000 waiting tasks hold 100,000 futures, and each future holds the message it built before calling send. The queue is small, the heap is not.

Why it still helps: a parked future is a few hundred bytes. A message can be megabytes. The channel bounds the expensive thing. And if the number of producers is itself bounded (one rendering thread, a fixed pool), then the number of waiters is bounded by construction.

Reserve before you produce: Sender::reserve().await hands you a permit before you build the payload. Do the heavy work only once you have a slot, so nothing large waits in memory.

Push the pressure to the edge. Backpressure only works if it reaches something that can slow down or say no:

• Bound the producers: a semaphore before spawning, a fixed worker pool, or no spawn at all so the wait propagates to the socket and TCP flow control slows the client.
• Load shed: try_send and return 429/503 when full, instead of parking.
• Push the queue out of the process: a Kafka consumer polls only when it has capacity, and the backlog lives on the broker's disk, not in your heap.

For a robust system, you might also use:

• A semaphore to limit concurrent producers.
• A fixed-size worker pool.
• Request rate limiting.
• Timeouts and cancellation.
• Load shedding (rejecting excess requests).

Graceful shutdown

Actor's event loop:

while let Some(msg) = actor.receiver.recv().await {
    actor.handle_message(msg);
}

What happens when the actor should stop?

• When all sender handles are dropped, the receiver eventually returns None.The loop exits, and the actor task finishes. This is a particularly nice property of this design: the actor's lifetime is connected to the lifetime of its handles.

• One caveat: if another actor holds a handle to this actor, that sender also keeps the channel alive. Circular actor references can therefore prevent graceful shutdown. The article discusses these shutdown cycles and possible solutions.

Real World Implementation:

A text-extraction service renders every page of a PDF to an image and extracts its text layout, then uploads the image to S3 and writes the layout to Postgres. The first version opened the document from a pool of workers, one page per task.

The problem: PDFium is not reentrant. It has process-wide global state and allows one call at a time. Opening the document once per page from a pool of workers parsed a 1500-page file 1500 times and raced that global state until every later open failed with FormatError. A Mutex around pdfium would have fixed the race, but every worker would then spend most of its time queued on the lock, and the JPEG encoding, the S3 upload, and the Postgres write, which need no lock at all, would crawl along at pdfium's pace.

The per-page work has two halves. For each page:

1. Render the bitmap and extract text. This touches pdfium.
2. Encode the bitmap to JPEG, upload it to S3, write the layout to Postgres.
   This never touches pdfium.

Half 1 must be one at a time, because pdfium is not reentrant. Half 2 can run
on every core at once, because it is plain CPU work and network I/O with no
shared state.

The obvious fix with a mutex: keep the worker pool, wrap pdfium in a mutex,
and each worker does:

let page = {
    let pdfium = PDFIUM.lock();      // wait for the lock
    render_page(&pdfium, n)          // half 1
};                                   // lock released
encode_and_store(page).await;        // half 2

That does fix the race. But look at what happens in practice with, say, eight
workers. All eight want the lock. One holds it and renders. Seven sit waiting.
When the first one releases the lock and goes off to encode and upload, the
next one grabs it. So at any moment, one worker is rendering and the others
are mostly queued on the lock. Half 2 only ever overlaps a little, because
every worker spends most of its life waiting for the single pdfium slot.

The mutex does not just serialize pdfium. It turns the whole pool into a line
that moves at the speed of pdfium, and the parallelisable half rides in that
line too.

There is also a second cost: with a mutex, the document is still opened per
worker or per page. The thing you actually want is one open for the whole
document, and a mutex does not give you that.

The actor version separates the halves physically. One thread owns pdfium and
does nothing but half 1, page after page, with no lock to take. It pushes each
rendered page into the channel. Eight workers pull from the channel and do
only half 2. Now pdfium runs flat out on one core, and encoding and uploads
run flat out on the other seven. Nobody ever waits for the lock, because there
is no lock. The only waiting is the channel's backpressure when the workers
fall behind, which is exactly the waiting you want.

The fix is the actor pattern from above: pdfium gets one owner. One blocking thread opens the document once, renders pages in order, and sends each rendered page down a bounded channel. A JoinSet of workers receives them and does everything that does not touch pdfium.

The actor: one thread owns pdfium

fn render_pages_on_owner_thread(
    temp_path: PathBuf,
    total_pages: i32,
    tx: mpsc::Sender<RenderAttempt>,
) -> anyhow::Result<()> {
    let pdfium = pdfium::pdfium()?;
    // Opened once, for the whole document.
    let document = pdfium.load_pdf_from_file(&temp_path, None)?;

    for page_number in 1..=total_pages {
        let result = render_page(&document, page_number);
        // Bounded channel: when the workers fall behind, this blocks
        // instead of racing ahead and filling memory with bitmaps.
        if tx.blocking_send(RenderAttempt { page_number, result }).is_err() {
            // The receiving side has already failed and gone; stop rendering.
            break;
        }
    }
    Ok(())
}

A page that fails to render is sent as Err and stays one failed page. Only a document that cannot be opened ends the run.

The handle side: bounded channel and a worker pool

async fn process_pages_progressively(/* ... */) -> anyhow::Result<PageProcessingSummary> {
    // Two pages per worker keeps the workers fed while bounding memory:
    // a rendered A4 page at 150 DPI is ~9 MB of RGBA.
    let (tx, mut rx) = mpsc::channel::<RenderAttempt>(worker_count * 2);

    let renderer = tokio::task::spawn_blocking(move || {
        render_pages_on_owner_thread(temp_path, total_pages, tx)
    });

    let mut join_set = JoinSet::new();
    let mut summary = PageProcessingSummary::default();

    while let Some(attempt) = rx.recv().await {
        // Fixed-size pool: wait for a worker before spawning another.
        if join_set.len() >= worker_count {
            if let Some(result) = join_set.join_next().await {
                summary.record(result??);
            }
        }
        spawn_page_storage(&mut join_set, attempt, /* ... */);
    }
    while let Some(result) = join_set.join_next().await {
        summary.record(result??);
    }

    // The channel closes when the renderer returns.
    renderer.await??;
    Ok(summary)
}

Each storage worker encodes the JPEG on spawn_blocking, then runs the S3 upload and the Postgres write concurrently with tokio::join!. None of that touches pdfium, so any number of them may run at once.

Retries: three layers, three places

The useful question is not how to retry but where. A failure can be retried at three layers, and each belongs somewhere different.

1. Inside the worker, per operation. The S3 upload is the one that fails transiently: a network blip, throttling, a 503. Three attempts with exponential backoff, then give up and mark the page failed.

async fn upload_image_with_retry(
    storage_service: &StorageService,
    key: &str,
    data: &[u8],
) -> anyhow::Result<()> {
    const MAX_ATTEMPTS: usize = 3;

    for attempt in 1..=MAX_ATTEMPTS {
        match storage_service.upload_image(key, data).await {
            Ok(()) => return Ok(()),
            Err(error) if attempt < MAX_ATTEMPTS => {
                warn!("Upload attempt {}/{} failed for {}: {:#}", attempt, MAX_ATTEMPTS, key, error);
                // 250ms, 500ms, 1000ms
                tokio::time::sleep(Duration::from_millis(250 * (1 << (attempt - 1)))).await;
            }
            Err(error) => return Err(error),
        }
    }

    unreachable!("upload retry loop always returns")
}

This is safe because a PUT to the same key is idempotent: retrying twice leaves the same object. The rules for this layer:

• Retry only transient errors. A 403 or a malformed key will fail the same way forever.
• Bounded attempts and exponential backoff, ideally with jitter so a herd of workers does not retry in lockstep.
• Only idempotent operations. If a retry could double-write, fix the operation first.

2. Per page: do not. Rendering is deterministic. A page pdfium cannot render will fail again, and a retry only burns time on the one thread that is the bottleneck. It is recorded as a failed page and the document fails. Do not send a page back to the renderer for a second go either: that needs a second channel pointing back at the actor, which is the circular-reference shutdown problem from above.

3. Per document: let the queue do it. If the step fails, the Kafka message is not acked, the broker redelivers it, and extract runs again from the top. The processor is built so that rerun is safe:

• It checks the document status first and skips if already completed, so a duplicate delivery is a no-op.
• Page rows, S3 keys and text results are all upserts to deterministic keys, so re-running overwrites rather than duplicates.

Keep the actor dumb and fire-and-forget. Do not make the actor resumable.
Make the work idempotent and let the queue retry it.

Redelivery resumes, it does not restart. If 1499 of 1500 pages succeeded before the step failed, re-rendering all of them is pdfium time wasted on work that is already in S3 and Postgres. The page rows carry the status of every page, so a redelivered request reads them first and hands the renderer the set it may skip:

// Pages whose image is uploaded and whose text is stored need no second pass.
let completed: HashSet<i32> = page_repo.completed_page_numbers(document_id).await?;

for page_number in 1..=total_pages {
    if completed.contains(&page_number) {
        continue;
    }
    let result = render_page(&document, page_number);
    if tx.blocking_send(RenderAttempt { page_number, result }).is_err() {
        break;
    }
}

The first delivery sees an empty set and renders everything. A redelivery renders only the pages that failed or never ran, and the final check compares the pages received against total_pages - completed.len() instead of total_pages. The actor is still dumb: it does not know why a page is in the set, only that it is.

Shutdown, both directions

This is the graceful shutdown from the previous section, in both directions:

• The renderer finishes its loop and drops tx. rx.recv() returns None, the while let ends, and the remaining workers are drained.
• If the receiving side fails and drops rx, the next blocking_send returns Err and the renderer stops. No orphaned thread keeps rendering a document nobody will store.

What it bought: one parse per document instead of one per page, no more FormatError under load, and the CPU-bound and I/O-bound work spread across all cores while pdfium stays single-threaded.

References:

• ryhl.io
• tokio/spawning
• tokio/channels