Backpressure, AIMD concurrency, and store-and-forward for actors with only local resources.
TL;DR: The classic fallacies of distributed computing cover the network miss critical two fallacies: that you have unlimited capacity to serve requests, and that the other party is available (temporal dependence). Actor-based flow control handles both with bounded mailboxes, admission control, AIMD concurrency limits, and a DB-backed outbox.
1. Missing Fallacies
You must have heard of the famous seven fallacies from Peter Deutsch that give hard lessons for every distributed systems programmer in understanding the networks, latency, bandwidth, topology, transport cost, etc. However, they miss a couple of other key fallacies, i.e.,
1.1 The capacity fallacy: You have infinite capacity to serve all requests
This is similar to the infinite bandwidth fallacy but instead of network capacity, it focuses on the application level capacity. For example, assume a leader actor connected to 1 million IoT workers that all try to request the leader for some resource at the same time in a style of thundering herd. When working in a cloud or SaaS environment, the usual fixes assume shared infrastructure like rate limiting at a load balancer, queuing behind Kafka, shedding load at an API gateway (see Effective Load Shedding and Throttling Strategies for Managing Traffic Spikes and DDoS Attacks). But you can’t use these shared infrastructure when working in edge deployments, IoT fleets, or on-prem systems.
1.2 The availability fallacy: The other party is available when you send (Temporal Dependence)
Standard flow control assumes both sides are alive and reachable. I described CSP channels in my prior blog on Structured Concurrency with CSP and Actors, where CSP channels give you backpressure for free, but only while both parties are active. Actor mailboxes on the other hand, decouple sender and receiver in time. However, by default they can grow without bound when one side goes silent. A remote device or a worker may go offline for hours so leader can’t send it new messages. You have availability problems both on the producer side and on the consumer side where you may need to queue incoming/outgoing messages in-memory. When the worker reconnects, it may get 100,000 stale messages or the leader may simply lose messages because they were all in memory. WhatsApp built chat services with Erlang’s actor model where it stored messages on the server until the recipient came online. It used fire-and-forget for delivery receipts with a reliable fallback.
PlexSpaces framework that I built is based on actors model where every actor maintains private state and processes messages sequentially. It supports following fundamental communication patterns:
- Tell (asynchronous): The sender dispatches a message and moves on.
- Ask (request-reply): The sender dispatches a request and waits for a response with a timeout.
- Tell with Outbox fallback (asynchronous): The sender attempts to dispatch a message and falls back to an outbox when the other part is not available.
1.3 Why the usual tools don’t fit
| Solution | Breaks when… |
|---|---|
| Kafka / NATS | IoT devices can’t run a broker due to lack of shared infrastructure |
| Load balancer | Internal actor fan-out bypasses LB |
| API gateway | Peer-to-peer actor communication without centralized entry point |
| Shared DB queue | Worker is offline or disconnected due to single point of failure |
The actor model provides a flow control that works with local resources.
2. The Design: Seven Mechanisms, Three Layers
PlexSpaces implements flow control as three layers that uses local resources:
2.1 Bounded Mailboxes:
Every actor mailbox has a configurable capacity and an error strategy, which decides what happens when the mailbox is full:
ErrorreturnsMailboxError::Fullwith aretry_after_mshint.DropOldestdiscards the oldest queued message.DropNewestdiscards the incoming one.
This Error allows flow control instead of silent data loss so that the sender can route to the outbox instead of spinning. Unbounded queue are generally considered harmful because they hide problems, eat memory, and eventually crash.
2.2 Backpressure
PlexSpaces actors follow the Erlang model where each actor processes one message at a time. This means that there’s no concurrency between messages inside an actor. Erlang mailbox is unbounded by default, which can result in OOM crash but PlexSpaces framework supports backpressure at enqueue. When an actor can’t keep up, its mailbox fills up and tell() returns MailboxFull immediately that maps to DeliveryError::Capacity enum. The sender can recognize this error as a capacity signal and take proper action like use the outbox relay to retry later.
crates/outbox/src/outbox_relay.rs:
#[derive(Debug, thiserror::Error)]
pub enum DeliveryError {
/// Receiver at capacity (mailbox full). Triggers AIMD multiplicative decrease.
#[error(“RESOURCE_EXHAUSTED: {message} (retry_after_ms={retry_after_ms})”)]
Capacity { message: String, retry_after_ms: u64 },
/// Transient failure. AIMD limit unchanged — not a congestion signal.
#[error(“{0}”)]
Transient(String),
}
On the WASM path, MailboxFull is mapped in crates/application/src/wasm_message_sender.rs:
registry
.tell(&ctx, &to_id, msg)
.await
.map_err(|e| match e {
ActorRegistryError::MailboxFull { retry_after_ms, .. } => {
format!("RESOURCE_EXHAUSTED: mailbox full (retry_after_ms={retry_after_ms})")
}
other => other.to_string(),
})The AIMD loop in outbox_host.rs checks RESOURCE_EXHAUSTED error for retrying later.
2.3 AIMD – TCP Congestion Control
Instead of using static configuration for concurrency limit, which may result in wastes capacity or overload, PlexSpaces uses Additive Increase / Multiplicative Decrease (AIMD) algorithm, which is typically used for the TCP congestion control:
- On successful delivery:
limit += 1(additive increase) - On a capacity rejection (mailbox full):
limit /= 2(multiplicative decrease)
The implementation, from crates/outbox/src/concurrency_limiter.rs:
/// Acquire a permit. Returns None if at capacity (non-blocking).
pub fn try_acquire(&self) -> Option<Permit<'_>> {
loop {
let current = self.in_flight.load(Ordering::Acquire);
let limit = self.current_limit.load(Ordering::Acquire);
if current >= limit {
metrics::counter!("plexspaces_outbox_limiter_acquire_rejected_total").increment(1);
return None; // caller must back off
}
if self.in_flight.compare_exchange_weak(
current, current + 1, Ordering::AcqRel, Ordering::Acquire
).is_ok() {
return Some(Permit { limiter: self });
}
}
}
/// Delivery succeeded — additive increase: limit += 1, capped at max_limit.
pub fn on_success(&self, _permit: Permit<'_>) {
metrics::counter!("plexspaces_outbox_limiter_success_total").increment(1);
if matches!(self.strategy, ConcurrencyStrategy::ConcurrencyStrategyAimd) {
loop {
let current = self.current_limit.load(Ordering::Acquire);
let new = (current + 1).min(self.max_limit);
if self.current_limit.compare_exchange_weak(
current, new, Ordering::AcqRel, Ordering::Acquire
).is_ok() {
metrics::gauge!("plexspaces_outbox_limiter_current_limit").set(new as f64);
break;
}
}
}
}
/// Delivery rejected (RESOURCE_EXHAUSTED) — multiplicative decrease: limit /= 2, floored at min_limit.
pub fn on_rejection(&self, _permit: Permit<'_>) {
metrics::counter!("plexspaces_outbox_limiter_rejection_total").increment(1);
if matches!(self.strategy, ConcurrencyStrategy::ConcurrencyStrategyAimd) {
loop {
let current = self.current_limit.load(Ordering::Acquire);
let new = (current / 2).max(self.min_limit);
if self.current_limit.compare_exchange_weak(
current, new, Ordering::AcqRel, Ordering::Acquire
).is_ok() {
metrics::gauge!("plexspaces_outbox_limiter_current_limit").set(new as f64);
break;
}
}
}
}Both on_success and on_rejection are lock-free CAS loops too. Configuration (from examples/rust/embedded/job_queue/src/main.rs):
let outbox_config = OutboxConfig {
concurrency: Some(ConcurrencyConfig {
initial_limit: 10, // start here
min_limit: 1, // never go below 1 (AIMD floor)
max_limit: 50, // never exceed 50 (AIMD ceiling)
strategy: ConcurrencyStrategy::ConcurrencyStrategyAimd as i32,
}),
retry_policy: Some(default_retry_policy()),
relay_interval_ms: 5, // drain every 5ms
relay_batch_size: 20, // up to 20 messages per drain cycle
..Default::default()
};
let limiter = Arc::new(ConcurrencyLimiter::new(
outbox_config.concurrency.as_ref().unwrap(),
));Or use the SDK helper for defaults (initial=10, min=1, max=100):
use plexspaces_sdk::default_outbox_limiter;
let limiter = default_outbox_limiter(); // Arc<ConcurrencyLimiter>
2.4 Outbox Relay: Closing the Feedback Loop
The outbox relay is where the mailbox’s capacity signal and AIMD connect. Here’s the hot path, process_message, from crates/outbox/src/outbox_relay.rs:
async fn process_message(&self, msg: Message) {
let msg_id = msg.id.clone();
// 1. Acquire AIMD permit — non-blocking. If at limit, nack and skip.
let permit = match self.limiter.try_acquire() {
Some(p) => p,
None => {
self.channel.nack(&msg_id, true, None).await.ok(); // requeue, no delay
return;
}
};
let t0 = std::time::Instant::now();
match self.delivery.deliver(&msg).await {
Ok(()) => {
// 2a. Success ? additive increase: limit += 1
metrics::histogram!("plexspaces_outbox_relay_delivery_seconds")
.record(t0.elapsed().as_secs_f64());
metrics::counter!("plexspaces_outbox_relay_delivered_total").increment(1);
self.limiter.on_success(permit); // limit++
self.channel.ack(&msg_id).await.ok();
}
Err(err) => {
match &err {
DeliveryError::Capacity { .. } => {
// 2b. Mailbox full ? typed Capacity error ? multiplicative decrease: limit /= 2
metrics::counter!("plexspaces_outbox_relay_backpressure_total").increment(1);
self.limiter.on_rejection(permit); // limit /= 2
}
DeliveryError::Transient(_) => {
// 2c. Non-capacity error (network, timeout) ? don't penalize AIMD limit
metrics::counter!("plexspaces_outbox_relay_errors_total").increment(1);
self.limiter.on_success(permit); // preserve limit, not a congestion signal
}
}
let err_str = err.to_string();
let attempt = msg.delivery_count + 1;
if should_retry(&self.retry_policy, attempt, &err_str) {
// 3. Compute exponential backoff with full jitter
let backoff = compute_backoff(&self.retry_policy, attempt, &mut rand::thread_rng());
metrics::counter!("plexspaces_outbox_relay_retries_total").increment(1);
// 4. nack with delay ? sets send_at = now + backoff in the DB
// The dequeue query filters WHERE send_at <= now(), so the message
// is physically invisible to the relay until the backoff expires.
self.channel.nack(&msg_id, true, Some(backoff)).await.ok();
} else {
// 5. Max retries exceeded ? dead letter queue
metrics::counter!("plexspaces_outbox_relay_dead_lettered_total").increment(1);
self.channel.nack(&msg_id, false, None).await.ok();
}
}
}
}On a capacity rejection (DeliveryError::Capacity), the relay halves its concurrency limit and backs the message off in the DB so that it sends slower.
2.5 DB-Backed Store-and-Forward
The outbox is implemented with a database backend such as SQLite/Postgres (crates/channel/src/sqlite_backend.rs). The dequeue query enforces backoff at the storage layer, e.g.,
SELECT id, payload, headers_json, idempotency_key, conflict_resolution, delivery_count
FROM channel_messages
WHERE tenant_id = ? AND namespace = ? AND channel_name = ?
AND acked = 0 AND dead_lettered = 0
AND (send_at = 0 OR send_at <= ?) -- ? messages with future send_at are invisible
ORDER BY send_at ASC, created_at ASC
LIMIT ?The nack() implements the max-retries guard:
// crates/channel/src/sqlite_backend.rs
async fn nack(&self, message_id: &str, requeue: bool, delay: Option<Duration>) -> ChannelResult<()> {
let max_retries = crate::channel::max_retries_from_config(&self.config);
let current_delivery_count = /* fetch from DB */;
if requeue && current_delivery_count < max_retries {
// compute send_at timestamp: delay converts to Unix ms, 0 means immediate
let send_at_ms = delay
.map(|d| Self::system_time_to_unix_ms(SystemTime::now()) + d.as_millis() as i64)
.unwrap_or(0);
sqlx::query(&format!(
"UPDATE {} SET acked = 0, delivery_count = delivery_count + 1, send_at = ? \
WHERE id = ?",
self.table_name
))
.bind(send_at_ms) // ? real backoff enforced here, not just logged
.bind(message_id)
.execute(&self.pool).await?;
} else {
// max retries exceeded ? dead letter queue (same table, dead_lettered = 1)
sqlx::query(&format!(
"UPDATE {} SET dead_lettered = 1, delivery_count = delivery_count + 1 \
WHERE id = ?",
self.table_name
))
.bind(message_id)
.execute(&self.pool).await?;
}
}Above code implement the transactional outbox pattern, which is commonly used by common by standards such as DTN (Delay-Tolerant Networking, RFC 9171) and real life use cases like mobile or IoT workers that might get offline temporarily.
2.6 Partitioned Relay: Scale-Out Without a Broker
When dealing with millions of actors, a single relay becomes a bottleneck so PlexSpaces defines PartitionedOutboxRelay to split the work into N independent partitions:
- An FNV-1a hash of
receiver_idpicks the partition. - Each partition has its own channel, AIMD limiter, and relay routine.
- Messages for the same recipient always land in the same partition, so ordering is preserved.
2.7 OVERWRITE: Last-Write-Wins for Transient Data
The outbox pattern suits idempotent transactional messages that are processed in FIFO order but you may need some use cases for transient data such as heartbeats and status updates. In those cases, you may just need to keep latest request instead of processing stale request so PlexSpaces defines ConflictResolutionPolicy::OVERWRITE that replaces existing message if it matches the idempotency_key instead of appending. It keeps a sequence_number header to prevent out-of-order overwrites.
Heartbeat dispatch in Go (examples/go/apps/job_queue_outbox/job_queue_outbox.go):
// All heartbeats use the same idempotency key with OVERWRITE so only the latest
// heartbeat is queued if the worker hasn't processed the previous one yet.
hbPayload := `{"type":"heartbeat","seq":` + strconv.Itoa(seq) + `}`
ctx := `{"tenant":"default","namespace":"default"}`
result, err := host.Outbox().DispatchWithTargets(
ctx,
dispatcherID,
workers,
hbPayload,
"cast",
heartbeatIdemKey, // ? same key = last-write-wins, not append
"overwrite", // ? OVERWRITE conflict resolution
)The same pattern in TypeScript (examples/typescript/apps/job_queue_outbox/job_queue_outbox_actor.ts):
// 50 rapid heartbeats with the same idempotency key — only the last survives in queue
for (let i = 1; i <= count; i++) {
const payloadBytes = new TextEncoder().encode(
JSON.stringify({ type: "heartbeat", seq: i, cpu: 0.42 }));
const result = host.outbox.dispatch(
ctx, DISPATCHER_ID, workers, payloadBytes,
"cast",
HEARTBEAT_IDEM_KEY, // ? same key every iteration
"overwrite"
);
totalAccepted += result[0];
totalRejected += result[1];
}2.8 Dead Letter Queue: Poison Message Safety
The real world applications have to deal with poison messages that consistently fail due to malformed messages so you need a bounded retry limits so that it doesn’t blocks the head of the queue forever. After max_attempts failures, nack(requeue=false), poison messages are moved to the DLQ (in the same table with dead_lettered = 1). Each retry uses exponential backoff with full jitter:
delay = random(0, min(cap, base × 2^attempt))You can then review DLQ and replay/purge them if needed.
3. How the Flow Control Works
PlexSpaces uses composition to implement flow control, here’s what happens under load:
Step by step:
OutboxRelayacquires a permit fromConcurrencyLimiter, bounded by the current AIMD limit.- The relay calls
tell()to enqueue the message in the actor’s bounded mailbox. - Mailbox full:
tell()returnsMailboxFull, which becomesDeliveryError::Capacity. The relay callslimiter.on_rejection(), AIMD halves the limit, and the message gets asend_atdelay in the DB. - On success: the message is enqueued, the relay calls
limiter.on_success()and acks, and AIMD grows the limit by 1. - Non-capacity errors (network, timeout) return
DeliveryError::Transientand don’t shrink the limit.
And the whole system in one picture:
| Problem | Mechanism |
|---|---|
| Actor can’t keep up, messages pile up | Bounded mailbox (the admission gate) |
| Sender doesn’t know the right send rate | AIMD limiter |
| Receiver is offline or at capacity | DB-backed outbox with send_at backoff |
| One relay can’t keep up | Partitioned relay |
| Old status updates pile up | OVERWRITE |
| One bad message blocks the queue | Retry limit + DLQ |
4. Production Patterns
Following are a few real life scenarios with working examples in the PlexSpaces repository:
4.1 Thundering Herd Protection
Let’s say you have 100,000 IoT workers that reconnect after a network partition and then send heartbeat or other request at once. Without backpressure, all of them will hit the leader together. But with a outbox relay dispatch plus AIMD, the workers can act on the leader’s MailboxFull or DeliveryError::Capacity) and retry later. The relay will halve its concurrency and backs off without dropping messages.
Go (examples/go/apps/job_queue_outbox/job_queue_outbox.go):
func (j *JobQueueLeader) handleThunderingHerdTest(payloadJSON string) string {
targets := j.WorkerIDs // all workers at once — the herd
ctx := `{"tenant":"default","namespace":"default"}`
herdPayload := `{"task":"thundering_herd","priority":"high"}`
t0 := host.NowMs()
// DispatchWithTargets fans out to ALL targets via the outbox.
// The AIMD limiter throttles automatically when workers reject.
result, err := host.Outbox().DispatchWithTargets(
ctx, dispatcherID, targets, herdPayload, "cast", "", "")
wallMs := int(host.NowMs() - t0)
accepted, rejected := 0, 0
if err == nil && result != nil {
accepted = int(result.Accepted)
rejected = int(result.Rejected)
}
// rejected messages go back to the outbox with exponential backoff send_at
return marshal(map[string]any{
"targets": len(targets),
"accepted": accepted,
"rejected": rejected, // non-zero = AIMD kicked in, system is adapting
"wall_ms": wallMs,
})
}Rust embedded setup (examples/rust/embedded/job_queue/src/main.rs):
// Workers have bounded mailboxes — MailboxFull is the backpressure signal.
// No in-flight limit needed: the actor event loop is sequential, one message at a time.
// AIMD outbox: starts at 10, backs off when workers' mailboxes fill up
let outbox_config = OutboxConfig {
concurrency: Some(ConcurrencyConfig {
initial_limit: 10,
min_limit: 1,
max_limit: 50,
strategy: ConcurrencyStrategy::ConcurrencyStrategyAimd as i32,
}),
relay_interval_ms: 5,
relay_batch_size: 20,
..Default::default()
};4.2 Wave-Based Config Deployment
Let’s say you have 100,000 workers and need to rollout a firmware update. Config servers that serve firmware updates have bounded mailboxes and process one download at a time so deploying to all workers will fill mailboxes of the config servers. Instead of updating all workes simultaneously, you can update workers in waves of 2000. When a config server’s mailbox is full, MailboxFull becomes DeliveryError::Capacity, and the outbox relay halves its concurrency and backs off. The workers can then retry later with exponential backoff.
Go (examples/go/apps/job_queue_outbox/job_queue_outbox.go):
func (j *JobQueueLeader) handleConfigWaveDeploy(payloadJSON string) string {
waveSize := 20 // process 20 workers per wave
ctx := `{"tenant":"default","namespace":"default"}`
configPayload := `{"task":"config_download","version":"2.1.0"}`
totalAccepted, totalRejected := 0, 0
for start := 0; start < len(j.WorkerIDs); start += waveSize {
end := start + waveSize
if end > len(j.WorkerIDs) { end = len(j.WorkerIDs) }
targets := j.WorkerIDs[start:end]
t0 := host.NowMs()
result, err := host.Outbox().DispatchWithTargets(
ctx, dispatcherID, targets, configPayload, "cast", "", "")
wallMs := int(host.NowMs() - t0)
if err == nil && result != nil {
totalAccepted += int(result.Accepted)
totalRejected += int(result.Rejected)
}
// rejected messages are in the outbox with backoff — they'll retry automatically
waves = append(waves, waveResult{accepted: accepted, rejected: rejected, wallMs: wallMs})
}
}Config server with a bounded mailbox in app-config.toml:
[[supervisor.children]]
name = "config-server"
actor_type = "ConfigServer"
mailbox_capacity = 24.3 Heartbeat Coalescing
Again, let’s assume you have 100,000 workers that send heartbeats with some metrics and environment data every 5 seconds, but the leader only processes every 30 seconds. Without coalescing, 6 × 100,000 = 600,000 heartbeats queue up and 500,000 requests are stale by the time they’re processed. Instead, each sender can use one fixed idempotency_key for its heartbeats and OVERWRITE replaces the queued heartbeat if there is already pending request.
In Go, the key is a single constant reused across all heartbeat dispatches (examples/go/apps/job_queue_outbox/job_queue_outbox.go):
const heartbeatIdemKey = "hb-leader-worker-101" // same key for all heartbeats to leader
func (j *JobQueueLeader) handleSendHeartbeat(payloadJSON string) string {
seq := j.HeartbeatCount
j.HeartbeatCount++
hbPayload := `{"type":"heartbeat","seq":` + strconv.Itoa(seq) + `}`
ctx := `{"tenant":"default","namespace":"default"}`
result, err := host.Outbox().DispatchWithTargets(
ctx,
dispatcherID,
workers,
hbPayload,
"cast",
heartbeatIdemKey, // ? same key every time = last-write-wins
"overwrite", // ? OVERWRITE: replaces existing unacked message with same key
)
// result.Accepted = 1 (new message), result.Rejected = 0, OR
// if OVERWRITE replaced: result.Accepted = 1 (replace), previous message gone
}Python (examples/python/apps/job_queue_outbox/job_queue_outbox_actor.py):
HEARTBEAT_IDEM_KEY = "hb-leader-worker-101" # constant key = only latest survives in queue
def handle_send_heartbeat(self, payload: dict) -> dict:
seq = self.heartbeat_count
self.heartbeat_count += 1
hb_payload = json.dumps({"type": "heartbeat", "seq": seq})
result = host.outbox.dispatch(
ctx, DISPATCHER_ID, self.worker_ids,
hb_payload.encode(), "cast",
HEARTBEAT_IDEM_KEY, # ? constant idempotency key
"overwrite" # ? OVERWRITE resolution
)4.4 Store-and-Forward: Tell with Outbox Fallback
In this case workers send telemetry every second using fire-and-forget style. Assume, the collector is at capacity half the time and metrics must not be silently dropped. You can then try tell() (fire and forget) first, which delivers straight to the mailbox with zero overhead. If it fails for any reason (mailbox full, target offline, network error), fall back to the outbox for durable retry. This is similarl to earlier WhatsApp architecture where it used Erlang actors to deliver directly when you can but then store them on the server when you are offline.
It’s also a first-class API in the core ActorRegistry:
// crates/actor/src/actor_registry.rs
/// Try direct tell() first. On any failure, store in outbox for async retry.
/// Returns Ok(true) = delivered directly, Ok(false) = stored in outbox.
/// Returns Err only if BOTH paths fail.
pub async fn tell_reliable(
&self,
ctx: &RequestContext,
actor_id: &ActorId,
message: Message,
outbox_channel: &dyn plexspaces_channel::Channel,
) -> Result<bool, ActorRegistryError> {
let start = std::time::Instant::now();
match self.tell(ctx, actor_id, message.clone()).await {
Ok(()) => {
metrics::counter!("plexspaces_tell_reliable_total", "path" => "direct").increment(1);
Ok(true) // ? fast path: zero outbox overhead
}
Err(tell_err) => {
tracing::debug!(actor_id = %actor_id, error = %tell_err,
"tell_reliable: direct delivery failed, falling back to outbox");
let mut outbox_msg = message;
if outbox_msg.receiver_id.is_empty() {
outbox_msg.receiver_id = actor_id.to_string();
}
match outbox_channel.send(outbox_msg).await {
Ok(_msg_id) => {
metrics::counter!("plexspaces_tell_reliable_total", "path" => "outbox").increment(1);
Ok(false) // ? stored durably, OutboxRelay will deliver
}
Err(channel_err) => {
metrics::counter!("plexspaces_tell_reliable_total", "path" => "failed").increment(1);
Err(ActorRegistryError::SendFailed(format!(
"tell_reliable: direct failed ({}), outbox also failed ({})",
tell_err, channel_err
)))
}
}
}
}
}SDK wrapper (sdks/rust/plexspaces-sdk/src/lib.rs):
pub async fn tell_reliable(
ctx: &plexspaces_actor::RequestContext,
service_locator: &dyn ServiceLocator,
actor_id: &ActorId,
message: Message,
outbox_channel: &dyn plexspaces_channel::Channel,
) -> Result<bool, Box<dyn std::error::Error + Send + Sync>> {
let registry = service_locator.actor_registry().await
.ok_or("ActorRegistry not available")?;
registry.tell_reliable(ctx, actor_id, message, outbox_channel).await
.map_err(|e| Box::new(e) as _)
}WASM actors implement the same pattern by hand (examples/go/apps/job_queue_outbox/job_queue_outbox.go):
// examples/go/apps/job_queue_outbox/job_queue_outbox.go
func (j *JobQueueLeader) handleTelemetryTest(payloadJSON string) string {
fastPathOK, fallbackOK := 0, 0
for i := 0; i < count; i++ {
targetID := workers[i%len(workers)]
telPayload := `{"type":"telemetry","cpu":0.5,"mem":0.8,"seq":` + strconv.Itoa(i) + `}`
// Fast path: direct ask() — zero overhead if worker is up and has capacity
_, err := host.Ask(targetID, "process",
map[string]any{"type": "telemetry", "seq": i}, 2000)
if err == nil {
fastPathOK++
} else {
// Fallback: outbox — durable, will retry with AIMD + backoff
_, dispErr := host.Outbox().DispatchWithTargets(
ctx, dispatcherID, []string{targetID},
telPayload, "cast", "", "")
if dispErr == nil { fallbackOK++ }
}
}
return marshal(map[string]any{
"fast_path_success": fastPathOK,
"outbox_fallback": fallbackOK,
})
}Python (examples/python/apps/job_queue_outbox/job_queue_outbox_actor.py):
def handle_telemetry_test(self, payload: dict) -> dict:
fast_path, fallback = 0, 0
for i in range(count):
target = self.worker_ids[i % len(self.worker_ids)]
try:
# Direct ask — zero overhead when worker is available
host.ask(target, "process", {"type": "telemetry", "seq": i}, timeout_ms=2000)
fast_path += 1
except Exception:
# Outbox fallback — durable when worker is at capacity or offline
host.outbox.dispatch(ctx, DISPATCHER_ID, [target],
json.dumps({"type": "telemetry", "seq": i}).encode(),
"cast", "", "")
fallback += 1
return {"fast_path_success": fast_path, "outbox_fallback": fallback}4.5 Reliable Offline Delivery
In this scenario, workers go offline mid-deployment but the firmware update messages must survive. When workers reconnect they should get the messages they missed. This can be implemented using a SQLite-backed outbox channel. The delivery_count persists in the DB so retry counts survive restarts. The send_at enforces a backoff delays and the DLQ catches poison messages after max_attempts. The channel survives node restarts because we are using the DB as a queue.
Rust embedded SQLite setup (examples/rust/embedded/job_queue/src/main.rs):
let channel_config = ChannelConfig {
name: "job-outbox".to_string(),
provider: ChannelProvider::ChannelProviderSqlite as i32,
delivery: DeliveryGuarantee::DeliveryGuaranteeAtLeastOnce as i32,
ordering: OrderingGuarantee::OrderingGuaranteeFifo as i32,
storage_path: "./job-queue.db".to_string(), // survives restarts
..Default::default()
};
let channel = Arc::new(SqliteChannel::new(channel_config).await?);
let relay = Arc::new(OutboxRelay::new(
&outbox_config,
channel.clone(), // DB-backed: messages persist across restarts
limiter.clone(), // AIMD: adapts to receiver capacity
delivery.clone(), // callback: delivers to actor mailbox
));
// Start the relay loop — drains DB and delivers messages
let relay_handle = relay.clone();
tokio::spawn(async move { relay_handle.run().await });5. Safety Rules
Above Outbox pattern work best when the rest of the system follows a few widely used practices such as (see From Code to Production: A Checklist for Reliable, Scalable, and Secure Deployments):
- Idempotency. At-least-once delivery means duplicates will happen with retries so use
idempotency_keyto dedupe and design handlers to handle duplicate requests (see How Duplicate Detection Became the Dangerous Impostor of True Idempotency). - Retry at one layer. HTTP clients, gRPC, and SDKs often retry by default. This can amplify when retries happen at multiple layers. You can send the remaining time budget down the call chain. I discussed this in my earlier blog Robust Retry Strategies for Building Resilient Distributed Systems.
- Permanent errors. Capacity errors, timeouts, and network errors are worth retrying but a malformed payload is not retryable so send it to the DLQ.
- Retry with jitter and backoff. Add a random offset with exponential backoff on the client side when connecting and retrying to prevent the thundering herd attack.
- TTL. You can add a TTL to your messages so that recipient can discard stale requests.
- Bulkheads. Keep separate limits per downstream dependency or message class so that only one dependency’s limit can be adjusted as needed.
- Watch queue age and depth. Monitor both age of the oldest unacknowledged message and queue depth to track whether you’re keeping up. You can add other metrics like current limit, in-flight count, and rejections.
- Metastable failures. Keep bounded-queue size and retry budget small as a growing backlog with retries can keep a system overloaded after the original trigger is gone. See Effective Load Shedding and Throttling Strategies for Managing Traffic Spikes and DDoS Attacks.
- Use typed errors for control signals. Instead of matching on error strings, use proper error codes and enums for error types (e.g.,
DeliveryError::Capacity).
Related Posts
This post is part of a series on building distributed systems with actors:
- Building an Agent Harness and Eval Pipeline with Durable Actors
- Write a Redis Clone with Virtual Actors
- Building PlexSpaces: Decades of Distributed Systems Distilled
- Building Polyglot and Serverless Applications with WebAssembly
- Building Mini-OpenClaw: Secure AI Agents with Actors, WASM, and Supervision
- Building a Self-Improving AI Agent with Durable Actors — MiniHermes
- 20 Production Patterns for Distributed AI Agents Using Actors and TupleSpaces
- Multi-agent coordination patterns: Five approaches and when to use them
- Self-Improving AI Agent (MiniHermes)
- Redis Clone with Virtual Actors
- Agent Harness and Eval Pipeline
- Migrating off Cloudflare Durable Objects
- Structured Concurrency Part V: CSP, Actors, Linda, and async/await
- Effective Load Shedding and Throttling Strategies
- When Hundreds of AI Agents Self-Organized: Coordination Patterns
GitHub: https://github.com/bhatti/PlexSpaces
Clone the repo. Read examples/README.md. Run ./test.sh in any example directory.
