Feature: Queueing System
Relevant source files
The following files were used as context for generating this wiki page:The queueing system is a core component of the Kodus Runtime, designed for robust, asynchronous, and prioritized event processing. It provides the infrastructure for decoupling event production from consumption, enabling features like durable persistence, retries with exponential backoff, and a Dead-Letter Queue (DLQ) for handling unprocessable events. This system is fundamental to higher-level abstractions like the Workflow Engine.
The system is built on tokio::sync::mpsc channels and features a typed event structure with rich metadata for tracking and correlation. It is configurable via a central config.json file and is exposed to both internal Rust components and external Node.js applications through a native addon.
Sources: apps/runtime/crates/core/src/queue.rs, README.md
Core Data Structures
The queueing system revolves around the Event structure, which encapsulates the data being processed and its associated metadata.
Event Structure
An Event is the fundamental unit of data that flows through the system. It contains the payload, metadata for routing and context, a timestamp, priority level, and a retry counter.
This diagram shows the composition of an Event, which includes EventMetadata and a Priority level.
Sources: apps/runtime/crates/core/src/queue.rs:6-21
Event Fields
| Field | Type | Description |
|---|---|---|
id | Uuid | A unique identifier for the event, generated upon creation. |
payload | Vec<u8> | The raw data of the event, as a byte vector. |
metadata | EventMetadata | A struct containing contextual information about the event. |
timestamp | DateTime<Utc> | The UTC timestamp when the event was created. |
priority | Priority | The processing priority of the event. |
retry_count | u32 | A counter for how many times processing has been retried. |
Sources: apps/runtime/crates/core/src/queue.rs:7-14
EventMetadata Fields
| Field | Type | Description |
|---|---|---|
source | String | The origin service or component that produced the event. |
correlation_id | Option<Uuid> | An optional ID to correlate multiple events across a workflow. |
user_id | Option<String> | An optional identifier for the user associated with the event. |
tags | HashMap<String, String> | A map of key-value pairs for custom tagging and filtering. |
Sources: apps/runtime/crates/core/src/queue.rs:17-23
Priority Levels
Events can be assigned one of four priority levels, which determines the order in which they are dequeued for processing.
| Level | Value | Description |
|---|---|---|
Critical | 3 | Highest priority, processed before all others. |
High | 2 | High priority. |
Normal | 1 | The default priority level. |
Low | 0 | Lowest priority, processed only when no higher-priority events are present. |
Sources: apps/runtime/crates/core/src/queue.rs:26-33
Queue Architecture
The system implements a priority queue not as a single data structure but as a set of four distinct tokio::sync::mpsc channels, one for each priority level. This design simplifies the implementation and leverages Tokio's efficient, asynchronous channels.
Sources: apps/runtime/crates/core/src/queue.rs:90-100
Priority Queue Implementation
A PrioritySender directs an incoming event to the appropriate channel based on its priority field. The PriorityReceiver then polls these channels in a biased manner, ensuring higher-priority events are always processed first.
This diagram illustrates how an event is routed by the PrioritySender to one of four priority-specific channels, and how the PriorityReceiver selects events for consumption, prioritizing Critical over Low.
Sources: apps/runtime/crates/core/src/queue.rs:90-148
The PriorityReceiver::recv method uses a tokio::select! macro with the biased keyword to check for messages, starting from the critical channel down to the low channel.
// Source: apps/runtime/crates/core/src/queue.rs:133-142
pub async fn recv(&mut self) -> Option<Event> {
tokio::select! {
biased;
Some(e) = self.critical.recv() => Some(e),
Some(e) = self.high.recv() => Some(e),
Some(e) = self.normal.recv() => Some(e),
Some(e) = self.low.recv() => Some(e),
else => None,
}
}
This implementation guarantees that as long as there is an event in a higher-priority queue, it will be dequeued before any event from a lower-priority queue.
Sources: apps/runtime/crates/core/tests/queue.rs:4-19
Error Handling and Resilience
The queueing system includes built-in mechanisms for handling processing failures gracefully, including a configurable retry policy and a Dead-Letter Queue (DLQ).
Retry Policy
The RetryPolicy struct defines the behavior for retrying a failed event. It uses an exponential backoff strategy to calculate the delay between retries, preventing downstream systems from being overwhelmed.
| Field | Type | Default | Description |
|---|---|---|---|
max_retries | u32 | 3 | The maximum number of times to retry a failed event. |
initial_delay_ms | u64 | 1000 | The initial delay in milliseconds before the first retry. |
max_delay_ms | u64 | 30000 | The maximum possible delay between retries. |
backoff_multiplier | f64 | 2.0 | The factor by which the delay increases with each retry. |
Sources: apps/runtime/crates/core/src/queue.rs:200-216
The delay is calculated as: delay = initial_delay * (multiplier ^ retry_count), capped at max_delay_ms.
// Source: apps/runtime/crates/core/src/queue.rs:218-222
pub fn calculate_delay(&self, retry_count: u32) -> u64 {
let delay = self.initial_delay_ms as f64 * self.backoff_multiplier.powi(retry_count as i32);
delay.min(self.max_delay_ms as f64) as u64
}
Dead-Letter Queue (DLQ)
If an event fails processing even after all retries are exhausted, it is considered a "dead letter". If the DLQ is enabled for the workflow, the event is wrapped in a DeadLetterEvent and sent to a separate queue for manual inspection or offline processing.
The DeadLetterEvent structure captures the original event, the reason for failure, and other context.
| Field | Type | Description |
|---|---|---|
original_event | Event | The event that failed processing. |
failure_reason | String | A message describing why the processing failed. |
failed_at | DateTime<Utc> | The timestamp of the final failure. |
total_retries | u32 | The number of retries attempted before moving to the DLQ. |
Sources: apps/runtime/crates/core/src/queue.rs:182-197
The following diagram shows the complete lifecycle of an event that fails, is retried, and eventually sent to the DLQ.
This flow is demonstrated in the execute_workflow_with_retry function and the retry_dlq.rs example.
Sources: apps/runtime/crates/core/src/workflow.rs:88-121, apps/runtime/crates/core/examples/retry_dlq.rs:80-123
Integration and Usage
Configuration
The queueing system's behavior can be configured in the project's config.json file.
{
"queue": {
"capacity": 10000,
"priority_enabled": true,
"dead_letter_queue_enabled": true,
"retry_policy": {
"max_retries": 3,
"initial_delay_ms": 1000,
"max_delay_ms": 30000,
"backoff_multiplier": 2.0
}
}
}
This configuration allows operators to tune queue capacity, enable/disable features, and define the global default retry policy.
Sources: README.md:213-228
Node.js Addon
The core queueing and event features are exposed to Node.js via the ts-addon crate. This allows JavaScript and TypeScript applications to create and persist events with full metadata.
The persistEventWithMetadata function in the KodusRuntime N-API class maps JavaScript types to the Rust Event structure.
// Example from apps/api/examples/nodejs-usage.js
const eventId2 = await runtime.persistEventWithMetadata(
Buffer.from('{"payment": {"amount": 100.50, "currency": "USD"}}'),
'payment-service',
2, // Priority: High
'user-123',
'correlation-456',
[['environment', 'production'], ['region', 'us-east-1']]
);
This demonstrates how a Node.js application can create a high-priority event with a user ID, correlation ID, and custom tags, which are then handled by the underlying Rust runtime.
Sources: apps/runtime/crates/ts-addon/src/lib.rs:49-83, apps/api/examples/nodejs-usage.js:16-25
Summary
The Kodus Runtime queueing system is a sophisticated and resilient foundation for event-driven architecture. Its key features include a strict priority-based processing model, a rich and extensible event data structure, and robust error handling through configurable retries and a Dead-Letter Queue. By providing these capabilities at the core, it enables the development of complex, reliable, and observable systems, such as the Workflow Engine, which builds directly upon this event-processing infrastructure.
Sources: apps/runtime/crates/core/src/queue.rs, apps/runtime/crates/core/src/workflow.rs