DEV Community: Wesley Conde

Scheduled and delayed messages:Routing with TTL and dead-letter exchange in RabbitMQ

Wesley Conde — Tue, 30 Jun 2026 17:35:00 +0000

The challenge of scheduling events without plugins

Often, we need to delay message delivery in our messaging architectures: for example, sending a reminder email 2 hours after account creation, or retrying a temporarily failed transaction after a planned 10-minute delay.

While RabbitMQ offers an official delayed messages plugin (rabbitmq-delayed-message-exchange), it is not always available or permitted in managed cloud environments (like AWS Amazon MQ).

The onkai-unified-bus bypasses this limitation by designing a native topology based on message Time-To-Live (TTL) expiration directed toward a Dead-Letter Exchange (DLX).

The delayed message routing flow

To delay a message, we create an indirect routing flow:

Publish the event to a temporary queue that has no active consumers.
Configure a x-message-ttl limit on this queue equivalent to the desired delay.
Configure a Dead-Letter Exchange (DLX) pointing to the final target exchange.
When the TTL expires, RabbitMQ automatically evicts the message from the temporary queue and routes it to the exchange configured in the DLX.

package main

import (
    "context"
    "fmt"
    "github.com/rabbitmq/amqp091-go"
)

// SetupDelayedTopology declares a consumer-less queue with TTL and DLX arguments
func SetupDelayedTopology(ch *amqp091_go.Channel, delayMs int) (string, error) {
    delayQueueName := fmt.Sprintf("delay-queue-%dms", delayMs)

    // Declares the queue with specific TTL and DLX configurations
    _, err := ch.QueueDeclare(
        delayQueueName,
        true,  // durable
        false, // auto-delete
        false, // exclusive
        false, // no-wait
        amqp091_go.Table{
            "x-message-ttl":             int32(delayMs),       // Retention period
            "x-dead-letter-exchange":    "main-event-exchange", // Destination post expiration
            "x-dead-letter-routing-key": "notifications.send",
        },
    )
    return delayQueueName, err
}

Benefits of the native approach

Using this routing pattern:

Eliminates dependencies on third-party broker plugins.
Leverages the native persistence and reliability guarantees of the RabbitMQ engine.
Supports varied delay durations by declaring queues with specific TTLs on the fly.

Technical terms demystified

TTL (Time-To-Live): The maximum period of time a message can remain in a queue before being discarded or moved to a dead-letter exchange.
Dead-Letter Exchange (DLX): A normal RabbitMQ exchange where expired, rejected, or failed messages are routed automatically.
Delayed Topology: An arrangement of exchanges and queues configured to temporarily hold messages for a controlled time.

link: https://github.com/wesleyskap/onkai-unified-bus

Avoiding duplicate processing: The inbox pattern and idempotent consumers

Wesley Conde — Fri, 15 May 2026 19:39:00 +0000

The at-least-once delivery problem

In high-scale distributed systems, guaranteeing that a message is delivered exactly once (Exactly-Once) is a highly complex and network-inefficient task. For this reason, most message brokers (like RabbitMQ and Kafka) guarantee At-Least-Once delivery.

This means that during network hiccups or server restarts, the same event may be delivered to your consumer more than once. If your consumer processes this duplicate message (for example, approving a payment again or deducting balance twice), it will cause severe data inconsistency.

To ensure each event is processed exactly once by the business logic, we implement the Inbox Pattern.

The inbox pattern

The Inbox Pattern stores the IDs of all successfully processed messages in a persistent, transactional database table. Before initiating any business logic, the consumer checks if the incoming message ID already exists in the inbox history:

package main

import (
    "context"
    "database/sql"
)

type InboxStore interface {
    HasProcessed(ctx context.Context, messageID string) (bool, error)
    MarkAsProcessed(ctx context.Context, messageID string) error
}

type SQLInboxStore struct {
    db *sql.DB
}

func (store *SQLInboxStore) HasProcessed(ctx context.Context, messageID string) (bool, error) {
    var exists bool
    query := "SELECT EXISTS(SELECT 1 FROM inbox_messages WHERE message_id = ?)"
    err := store.db.QueryRowContext(ctx, query, messageID).Scan(&exists)
    return exists, err
}

func (store *SQLInboxStore) MarkAsProcessed(ctx context.Context, messageID string) error {
    _, err := store.db.ExecContext(ctx, 
        "INSERT INTO inbox_messages (message_id, processed_at) VALUES (?, NOW())", 
        messageID,
    )
    return err
}

Transactional idempotent consumption

We couple the Inbox verification and recording inside the same database transaction that executes the business logic. If the message is already present, the flow terminates gracefully without executing the business logic again:

func (c *OrderCompletedConsumer) Consume(ctx context.Context, msg Message) error {
    tx, err := c.db.BeginTx(ctx, nil)
    if err != nil {
        return err
    }
    defer tx.Rollback()

    // 1. Checks duplicate state using the InboxStore within the transaction
    alreadyProcessed, err := c.inboxStore.HasProcessed(ctx, msg.ID)
    if err != nil || alreadyProcessed {
        // Already processed! Exit silently to avoid processing again
        return nil
    }

    // 2. Executes the core business logic
    err = c.executeOrderBusinessLogic(tx, msg.Payload)
    if err != nil {
        return err
    }

    // 3. Marks the message as processed in the Inbox
    err = c.inboxStore.MarkAsProcessed(ctx, msg.ID)
    if err != nil {
        return err
    }

    return tx.Commit()
}

Technical terms demystified

Idempotency: The property of certain operations where they can be applied multiple times without changing the result beyond the initial application.
Inbox Pattern: A pattern where incoming messages are logged into a local table/database to check and filter out duplicates before executing business logic.
Silent Ignore: The practice of ignoring a duplicate request without returning a failure, avoiding unnecessary error logs or retries.

link: https://github.com/wesleyskap/onkai-unified-bus

Zero-allocation dispatcher: Routing millions of events without heap allocations in C#

Wesley Conde — Mon, 20 Apr 2026 17:09:00 +0000

The bottleneck of dynamic allocations in high-throughput event buses

In high-performance event-driven architectures, every microsecond counts. When implementing a local in-memory Event Bus in .NET, the most common pattern is allocating new message containers and wrapping delegates dynamically. While simple, this approach creates a large volume of short-lived objects on the heap. As a consequence, the Garbage Collector (GC) runs aggressively, causing gen-0/gen-1 collection overhead that degrades overall latency.

To solve this cleanly, onkai-unified-bus adopts a virtual zero-allocation strategy. By utilizing a high-performance, structured dispatch engine, it processes events via internally pooled message envelopes and concurrent channels, giving .NET developers maximum throughput with minimal memory footprint.

Bootstrapping the event bus

The library exposes a fluent registration API for Microsoft Dependency Injection. Under the hood, this sets up the zero-allocation transport pipelines, pooled resources, and background workers automatically.

using Microsoft.Extensions.DependencyInjection;
using Onkai.EventBus.Core.Extensions;
using Onkai.EventBus.RabbitMQ.Extensions;

var builder = WebApplication.CreateBuilder(args);

// Register Core EventBus infrastructure
builder.Services.AddEventBus()
                .UseRabbitMq(config =>
                {
                    config.HostName = "localhost";
                    config.UserName = "guest";
                    config.Password = "guest";
                });

High-speed event publishing

Once configured, developers simply inject the thread-safe IEventPublisher interface to dispatch events. The framework takes care of acquiring a pooled wrapper envelope, serializing the event payload, and scheduling delivery via optimized pipelines:

using Onkai.EventBus.Abstractions;

public sealed class OrderService
{
    private readonly IEventPublisher _publisher;

    public OrderService(IEventPublisher publisher)
    {
        _publisher = publisher;
    }

    public async Task CheckoutAsync(Guid orderId, CancellationToken cancellationToken)
    {
        var orderEvent = new OrderCreatedEvent(orderId, 199.99m, "customer@example.com");

        // Dispatches the event through onkai-unified-bus high-performance engine
        await _publisher.PublishAsync(orderEvent, cancellationToken: cancellationToken);
    }
}

Technical terms demystified

Heap: The region of system memory where dynamically allocated data lives at runtime. Managing the heap and performing Garbage Collection is far more expensive than using the Stack.
ObjectPool: A design pattern and .NET class that caches temporary objects for future reuse, reducing allocations.
System.Threading.Channels: A set of APIs in .NET that provides a high-performance, zero-allocation producer-consumer queue for concurrent architectures.

link: https://github.com/wesleyskap/onkai-unified-bus

Distributed transactions with sagas: Orchestrating steps and compensations

Wesley Conde — Sat, 18 Apr 2026 18:01:00 +0000

Distributed transactions in microservices

In a monolithic application, maintaining data consistency is straightforward: open a database transaction, modify the tables, and commit the changes. If any step fails, the database rolls back everything.

However, in a microservices architecture, each service owns its database. We cannot execute a single database transaction across the Order service, Inventory service, and Billing service.

To maintain consistency across multiple microservice boundaries without slow distributed locks (like two-phase commits - 2PC), we implement the Orchestrated Saga pattern.

The saga pattern and compensations

A Saga is a sequence of local transactions executed by individual microservices. If a local transaction fails, the orchestrator triggers Compensation transactions in reverse order to undo the side effects of previous steps. Here is how SagaOrchestrator<TState> is built in onkai-unified-bus:

using System.Collections.Concurrent;
using Onkai.EventBus.Abstractions;

namespace Onkai.EventBus.Core.Sagas;

public sealed class SagaOrchestrator<TState>
    where TState : class, new()
{
    private readonly ISagaStateStore<TState> _store;
    private readonly ConcurrentDictionary<string, Func<SagaContext<TState>, CancellationToken, Task>> _compensations = new();

    public SagaOrchestrator(ISagaStateStore<TState> store)
    {
        _store = store;
    }

    // Registers a compensation callback for a given step name
    public void RegisterCompensation(string stepName, Func<SagaContext<TState>, CancellationToken, Task> compensation)
    {
        _compensations[stepName] = compensation;
    }
}

Running saga steps safely

As business steps execute, the orchestrator updates the saga state persistently. If a step fails, the orchestrator automatically triggers the rollback flow, invoking registered compensations in reverse chronological order:

public async Task ExecuteStepAsync<TEvent>(
    string sagaId,
    TEvent @event,
    Func<SagaContext<TState>, TEvent, CancellationToken, Task> stepAction,
    CancellationToken cancellationToken)
    where TEvent : IEvent
{
    var context = await _store.GetAsync(sagaId, cancellationToken) ?? new SagaContext<TState> { SagaId = sagaId };

    if (context.Status is "Failed" or "Compensated")
    {
        return;
    }

    try
    {
        await stepAction(context, @event, cancellationToken);
        context.CompletedSteps.Add(typeof(TEvent).Name);
        await _store.SaveAsync(context, cancellationToken);
    }
    catch
    {
        await RollbackSagaAsync(context, typeof(TEvent).Name, cancellationToken);
    }
}

private async Task RollbackSagaAsync(SagaContext<TState> context, string failingStep, CancellationToken cancellationToken)
{
    context.Status = "Failed";
    await _store.SaveAsync(context, cancellationToken);

    // Compensate the failing step and all previously completed steps
    await CompensateStepAsync(context, failingStep, cancellationToken);
    for (var i = context.CompletedSteps.Count - 1; i >= 0; i--)
    {
        var completedStep = context.CompletedSteps[i];
        await CompensateStepAsync(context, completedStep, cancellationToken);
    }

    context.Status = "Compensated";
    await _store.SaveAsync(context, cancellationToken);
}

Technical terms demystified

Saga Pattern: A design pattern to coordinate distributed local transactions using structured compensations to restore consistency.
Compensation: A logical transaction that rolls back or undoes the changes made by a previously successful operation.
Saga State Store: A persistent or in-memory database that tracks the active steps and context of ongoing sagas.

link: https://github.com/wesleyskap/onkai-unified-bus

Reflection-free dispatcher: Optimizing performance and supporting native AOT

Wesley Conde — Thu, 16 Apr 2026 17:41:00 +0000

The costs of dynamic dispatch and AOT compilation

In traditional messaging frameworks, when a new event arrives, the router must dynamically find which class or struct handles that message type. In most runtimes, this is done using Reflection to inspect types at runtime.

While simple and flexible, reflection introduces severe issues:

Performance Overhead: Dynamic inspections and method invocations allocate metadata on the heap, adding load to the garbage collector.
Incompatibility with Native AOT (Ahead-of-Time): AOT compilers prune unused code and metadata during build time to produce tiny binaries. Invoking code via dynamic reflection often fails or crashes in production because the compiler cannot predict what will be dynamically inspected.

The onkai-unified-bus solves this by replacing dynamic invocations with a statically-typed dispatcher using a concurrent cache of typed executors.

Statically-typed consumer executors

Instead of using reflection to find and invoke the consumer method on every message, the framework registers generic executor objects during startup. The dispatcher delegates execution through static interfaces:

using Onkai.EventBus.Abstractions;

namespace Onkai.EventBus.Core.Subscription;

// Defines a contract to dispatch events to untyped consumer instances without using reflection invocation.
internal interface IEventConsumerExecutor
{
    Task ExecuteAsync(object consumer, IEvent @event, ConsumeContext context, CancellationToken cancellationToken);
}

// A generic helper that casts untyped inputs and executes the strongly-typed ConsumeAsync method directly.
internal sealed class EventConsumerExecutor<TEvent> : IEventConsumerExecutor
    where TEvent : IEvent
{
    public Task ExecuteAsync(object consumer, IEvent @event, ConsumeContext context, CancellationToken cancellationToken)
    {
        if (consumer == null) throw new ArgumentNullException(nameof(consumer));
        if (@event == null) throw new ArgumentNullException(nameof(@event));

        var typedConsumer = (IEventConsumer<TEvent>)consumer;
        var typedEvent = (TEvent)@event;

        return typedConsumer.ConsumeAsync(typedEvent, context, cancellationToken);
    }
}

The reflection-free dispatch engine

Upon receiving a message, the dispatcher loads the matching executor from a thread-safe map (ConcurrentDictionary) cached by event type, executing the interface call directly in nanoseconds:

using System.Collections.Concurrent;
using Onkai.EventBus.Abstractions;

public sealed class RabbitMqConsumer
{
    private readonly ConcurrentDictionary<Type, IEventConsumerExecutor> _executors = new();

    private async Task ExecuteConsumerAsync(Type eventType, object consumerInstance, IEvent eventData, ConsumeContext context, CancellationToken token)
    {
        // Get or add the cached compiled executor statically without reflection lookup
        var executor = _executors.GetOrAdd(eventType, t =>
        {
            var executorType = typeof(EventConsumerExecutor<>).MakeGenericType(t);
            return (IEventConsumerExecutor)Activator.CreateInstance(executorType)!;
        });

        await executor.ExecuteAsync(consumerInstance, eventData, context, token);
    }
}

Technical terms demystified

Native AOT (Ahead-of-Time): A compilation technology that compiles source code directly to native machine code at build time, bypassing JIT compilers or runtime interpreters.
Reflection-Free Dispatcher: A routing pattern that uses static interfaces or pre-compiled delegates to invoke handlers without inspecting objects at runtime.
Type Casting: Explicitly converting a generic interface or variable to its concrete structural type.

link: https://github.com/wesleyskap/onkai-unified-bus

Reliable messaging:Guaranteeing at-least-once with transactional outbox pattern

Wesley Conde — Sun, 12 Apr 2026 19:46:00 +0000

The two-phase distributed transaction dilemma

Consider this classic microservice scenario: your payment service completes a payment and saves the record in its local database. Next, it publishes a PaymentApprovedEvent to your message broker to notify the shipping service.

What happens if the database write fails but the message is already published? The customer didn't pay, but the product will be shipped. And what if the database write succeeds, but the network drops before publishing the message to the broker? The customer paid, but the product will never be dispatched.

The Transactional Outbox pattern resolves this data consistency dilemma, ensuring At-Least-Once message delivery.

Outbox pattern design

Instead of publishing the event directly to the broker, we save the message in a special table called outbox within the same database, using the same database transaction that writes the business data. In .NET, this is naturally achieved using Entity Framework Core's DbContext within a transaction scope:

using System.Text.Json;
using Microsoft.EntityFrameworkCore;
using Onkai.EventBus.Abstractions;

public sealed class OrderService
{
    private readonly DbContext _dbContext;
    private readonly IOutboxStore _outboxStore;

    public OrderService(DbContext dbContext, IOutboxStore outboxStore)
    {
        _dbContext = dbContext;
        _outboxStore = outboxStore;
    }

    public async Task CreateOrderAsync(Guid orderId, decimal amount, CancellationToken token)
    {
        // Execute both business write and outbox write inside the same atomic database transaction
        using var transaction = await _dbContext.Database.BeginTransactionAsync(token);
        try
        {
            // 1. Writes business data
            var order = new Order { Id = orderId, Amount = amount };
            _dbContext.Add(order);
            await _dbContext.SaveChangesAsync(token);

            // 2. Serializes and registers event in the outbox table via the outbox store within the transaction
            var orderEvent = new OrderCreatedEvent(orderId, amount);
            await _outboxStore.SaveMessageAsync(orderEvent, token);

            // Commits the transaction. If it fails, everything is rolled back.
            await transaction.CommitAsync(token);
        }
        catch
        {
            await transaction.RollbackAsync(token);
            throw;
        }
    }
}

The outbox processor (relay)

A background hosted service (OutboxProcessor) periodically polls unpublished messages from the outbox store, attempts to publish them to the broker transport, and marks them as published upon receiving confirmations (ACKs):

using Microsoft.Extensions.Hosting;
using Onkai.EventBus.Core.Transport;

public sealed class OutboxProcessor : BackgroundService
{
    private readonly IServiceProvider _serviceProvider;

    public OutboxProcessor(IServiceProvider serviceProvider)
    {
        _serviceProvider = serviceProvider;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            using var scope = _serviceProvider.CreateScope();
            var store = scope.ServiceProvider.GetRequiredService<IOutboxStore>();
            var transport = scope.ServiceProvider.GetRequiredService<IMessageTransport>();

            var pendingMessages = await store.GetUnpublishedMessagesAsync(stoppingToken);
            foreach (var message in pendingMessages)
            {
                var envelope = new TransportEnvelope
                {
                    EventId = message.EventId,
                    EventName = message.EventName,
                    Body = message.Body,
                    CorrelationId = message.CorrelationId
                };

                // Publishes the envelope to the broker
                await transport.SendAsync(envelope, stoppingToken);

                // Updates status in the outbox table
                await store.MarkAsPublishedAsync(message.Id, stoppingToken);
            }

            await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken);
        }
    }
}

Technical terms demystified

At-Least-Once Delivery: A guarantee that all messages will be delivered to their destination at least once, accepting the possibility of duplicates.
Database Transaction: A set of operations executed as a single, atomic logical unit of work (all or nothing).
Outbox Relay: The component responsible for reading the database outbox table and reliably publishing the records to the network.

link: https://github.com/wesleyskap/onkai-unified-eventbus

Background message consumption: Designing a resilient hosted service

Wesley Conde — Wed, 08 Apr 2026 14:42:00 +0000

The importance of isolated asynchronous processing

In modern microservices, web servers handle fast, short-lived HTTP requests. However, consuming messages from queues or topics (such as RabbitMQ, Kafka, or NATS) requires a different runtime model: continuous and asynchronous listening. This message consumption loop cannot run on the same thread as web requests, otherwise, it would freeze the main web server.

To solve this, onkai-unified-bus implements a background execution engine (Worker/Background Service) that manages the message consumption lifecycle in an isolated and resilient manner.

Implementing a background consumption worker

The background worker encapsulates the infinite listening loop of the transport broker. It boots alongside the application container startup and ensures correct handling of termination signals using cancellation contexts (context.Context):

package main

import (
    "context"
    "log"
    "sync"
    "time"
)

type MessageConsumer struct {
    driver    Driver
    topic     string
    stopChan  chan struct{}
    waitGroup sync.WaitGroup
}

func NewMessageConsumer(driver Driver, topic string) *MessageConsumer {
    return &MessageConsumer{
        driver:   driver,
        topic:    topic,
        stopChan: make(chan struct{}),
    }
}

// Start spawns the goroutine that consumes messages asynchronously
func (mc *MessageConsumer) Start(ctx context.Context) {
    mc.waitGroup.Add(1)
    go func() {
        defer mc.waitGroup.Done()
        log.Printf("[Consumer] Starting subscription on topic: %s", mc.topic)

        err := mc.driver.Subscribe(ctx, mc.topic, func(msg Message) error {
            // Simulates message processing logic
            log.Printf("[Consumer] Processing message: %s", msg.ID)
            return nil
        })

        if err != nil {
            log.Printf("[Consumer] Topic subscription error: %v", err)
        }
    }()
}

// Stop gracefully stops the consumer, draining active processing execution
func (mc *MessageConsumer) Stop() {
    close(mc.stopChan)
    mc.waitGroup.Wait()
    log.Println("[Consumer] Graceful shutdown completed.")
}

Failure handling and connection lifecycle

A production background worker must not crash the application if the network connection with the broker flickers. The consumption layer integrates with the driver to automatically re-establish the subscription after disconnections:

Continuous physical connection monitoring.
On-demand topology declaration (re-declaring queues if deleted).
Cooperative cancellation with context.Context during machine shutdowns.

Technical terms demystified

Background Service: A process executed in the background that performs continuous tasks (like reading a queue) without direct client interaction.
Graceful Drain: The process of stopping the intake of new events and waiting for active messages to complete before terminating the process.
Cooperative Cancellation: A pattern where parallel routines monitor a centralized signal (like a context or channel) to immediately stop execution.

link: https://github.com/wesleyskap/onkai-unified-bus

Asynchronous tracing: Propagating telemetry context across event boundaries

Wesley Conde — Sat, 28 Mar 2026 11:15:00 +0000

The challenge of tracing asynchronous pipelines

In traditional synchronous microservice architectures (like HTTP or gRPC), propagating trace contexts to assemble call trees is straightforward. Requests carry trace data directly inside standard headers.

However, in event-driven systems, this execution flow is decoupled. A producer publishes a message to a broker and immediately finishes its task. Seconds or minutes later, a consumer pulls the message and processes it. How do you correlate the telemetry of the consumer with the original transaction initiated by the producer?

onkai-unified-bus solves this by injecting and extracting W3C Trace Context headers transparently inside the physical metadata headers of every message.

Injecting and extracting trace context

To ensure compatibility with industry-leading observability tools (such as OpenTelemetry, Jaeger, and Zipkin), the bus exposes injection and extraction adapters conforming to the W3C standard.

import (
    "context"
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/propagation"
)

// InjectTrace injects active trace context data into the message headers
func InjectTrace(ctx context.Context, msg *Message) {
    propagator := otel.GetTextMapPropagator()
    carrier := propagation.MapCarrier(msg.Headers)
    propagator.Inject(ctx, carrier)
}

// ExtractTrace extracts trace context data from the message headers and returns a new context.Context
func ExtractTrace(ctx context.Context, msg Message) context.Context {
    propagator := otel.GetTextMapPropagator()
    carrier := propagation.MapCarrier(msg.Headers)
    return propagator.Extract(ctx, carrier)
}

End-to-end visual correlation

When a producer creates an event, the active Span ID is serialized into the message's traceparent header. When the event reaches the transport driver (e.g. RabbitMQ), the consumer's middleware intercepts it, extracts the context using ExtractTrace, and spawns a child Span under the original parent trace.

Thanks to this mechanism, even if an event is deferred and processed long after it was generated, your APM dashboards will show the exact correlation graph across all distributed systems.

Technical terms demystified

W3C Trace Context: A unified standard defining mandatory headers (traceparent, tracestate) to guarantee tracing interoperability across heterogeneous services.
Span ID: A unique identifier representing a single logical unit of work (such as a database query or message handling block).
Map Carrier: A utility adapter that maps OpenTelemetry fields into standard string-to-string dictionary headers.

link: https://github.com/wesleyskap/onkai-unified-bus

Error state machines:Designing dead-letter queues (DLQ) and jittered retries

Wesley Conde — Fri, 27 Mar 2026 17:19:00 +0000

Transient vs. permanent errors

When a consumer fails to process an event, the worst action is retrying it immediately and infinitely. If the error is transient (e.g., temporary network blip), rapid retries will overwhelm the failing downstream service. If the error is permanent (e.g., corrupted payload or schema validation failure), repeating the operation wastes CPU cycles and stalls progress.

To handle this onkai-unified-bus implements a robust error state machine combining Exponential Backoff with Jitter retries and a Dead-Letter Queue (DLQ).

Exponential backoff with Jitter

To prevent dozens of disconnected consumers from retrying or reconnecting at the exact same millisecond (the "thundering herd" effect), we inject random delay variance (Jitter) combined with an exponentially increasing interval between retries.

func CalculateBackoff(attempt int, baseDelay, maxDelay time.Duration) time.Duration {
    // Exponential calculation: base * 2^attempt
    backoff := float64(baseDelay) * math.Pow(2, float64(attempt))
    duration := time.Duration(backoff)
    if duration > maxDelay {
        duration = maxDelay
    }
    // Add +/- 10% Jitter (random noise) to spread out retry load
    jitter := rand.Float64() * 0.2 - 0.1
    duration = time.Duration(float64(duration) * (1.0 + jitter))
    return duration
}

Isolation via dead-letter queue (DLQ)

If an event fails after reaching the maximum number of retry attempts (e.g., 5 attempts), it is classified as a permanent logical failure. To prevent stalling the rest of the queue, the event bus intercepts the offending message, attaches the error's stack trace to the message headers, and forwards it to a designated isolation queue called the Dead-Letter Queue (DLQ).

From the DLQ, faulty messages can be audited manually or processed through administrative tools once the underlying business logic bug is fixed.

Technical terms demystified

Dead-Letter Queue (DLQ): A secondary queue dedicated to holding messages that couldn't be delivered or processed successfully by consumers.
Thundering Herd Effect: A situation where many concurrent processes wake up at the same time to handle a single event, overwhelming system resources.
Jitter: The deliberate introduction of small, random time variations to prevent processes from synchronizing their retries.

link: https://github.com/wesleyskap/onkai-unified-bus

Decoupled drivers: Abstracting NATS, RabbitMQ, and Kafka with a unified interface

Wesley Conde — Sun, 22 Mar 2026 17:05:00 +0000

The problem of broker coupling

Many engineering teams start their projects by coupling their business logic directly to a specific messaging client library (such as RabbitMQ's amqp driver or Kafka's sarama client). As the ecosystem grows, needs shift: perhaps Kafka is overkill for small microservices and you want NATS, or you need disk durability that simple NATS setups lack.

Migrating when your business logic is heavily coupled with the specific types and SDK methods of a particular broker is a tedious, error-prone refactoring nightmare.

To prevent this, onkai-unified-bus introduces a structured driver abstraction layer behind clean Go interfaces.

Defining the unified contract

The key to transport decoupling is establishing cohesive behavioral contracts. The unified interface exposes the minimum common denominator of requirements for asynchronous messaging:

type Message struct {
    ID      string
    Payload []byte
    Headers map[string]string
}

type Driver interface {
    Connect(ctx context.Context) error
    Publish(ctx context.Context, topic string, msg Message) error
    Subscribe(ctx context.Context, topic string, handler func(msg Message) error) error
    Close() error
}

Managing physical connections and reconnections

Each concrete driver (e.g. RabbitMQDriver, NATSDriver) implements the interface above and takes full responsibility for managing its physical network connections.

For instance, the RabbitMQ driver listens to channel closed events (amqp.Channel.NotifyClose) in a background monitoring routine to automatically attempt reconnections on network dropouts, without exposing any of these internal mechanics to the application layers.

func (d *RabbitMQDriver) handleReconnect(ctx context.Context) {
    for {
        select {
        case <-d.closeChan:
            return
        case err := <-d.conn.NotifyClose(make(chan *amqp.Error)):
            if err != nil {
                log.Println("RabbitMQ connection lost, attempting reconnect...")
                d.reconnect(ctx)
            }
        }
    }
}

Technical terms demystified

Go Interfaces: Types that define method signatures (behaviors). Any struct that implements these methods implicitly satisfies the interface.
Message Broker: A middleware agent that translates and routes messages across distributed components (e.g., RabbitMQ, NATS, Kafka).
Decoupling: The design practice of ensuring software components have little or no direct dependency on one another.

link: https://github.com/wesleyskap/onkai-unified-bus

Flow control and backpressure: Preventing memory saturation in event consumption

Wesley Conde — Sat, 21 Mar 2026 12:23:00 +0000

The phenomenon of consumer saturation

In any distributed or event-driven architecture, mismatching speeds between producers and consumers is inevitable. If your producer service generates 50,000 messages per second, but your downstream database or API integrated into the consumer can only handle 5,000 requests per second, system memory will quickly become a point of failure. Without structured flow control, local buffers will accumulate data indefinitely until an Out Of Memory (OOM) crash occurs.

To shield applications against these traffic surges, onkai-unified-bus implements robust Backpressure policies and non-blocking circular buffers.

Control policies: Blocking vs. dropping

When an event reception channel reaches its maximum capacity (saturation), the event bus offers three configurable policies:

Block: The producer goroutine is temporarily paused when attempting to publish new events. This naturally throttles the production rate until consumers drain the queue.
Drop Oldest: The bus discards the oldest message in the queue to insert the new one, prioritizing fresh events and avoiding unbounded latency.
Drop Newest: The current event is discarded immediately, returning a saturation error to the caller.

type BackpressurePolicy int

const (
    PolicyBlock BackpressurePolicy = iota
    PolicyDropOldest
    PolicyDropNewest
)

func (b *Bus) Publish(topic string, data []byte, policy BackpressurePolicy) error {
    select {
    case b.inbox <- data:
        return nil
    default:
        switch policy {
        case PolicyBlock:
            b.inbox <- data // Blocks the producer goroutine until space is freed
            return nil
        case PolicyDropOldest:
            select {
            case <-b.inbox: // Discards the oldest item in the queue
            default:
            }
            b.inbox <- data // Insert the new data
            return nil
        case PolicyDropNewest:
            return ErrQueueSaturated
        }
    }
    return nil
}

Channel-based circular buffers

By leveraging Go's native buffered channels, the event bus performs these checks extremely efficiently without complex locks or slow custom queue structures. Using a simple select block without a default channel allows backpressure decisions to occur at the CPU level in nanoseconds.

Technical terms demystified

Backpressure: The resistance or force exerted in opposition to data flow, forcing the sending system to throttle its transmission rate.
Out Of Memory (OOM): A critical state where the operating system terminates a process because it has consumed all available physical memory.
Buffered Channels: Native Go queues with pre-allocated capacity that allow sending messages without blocking the sender until the buffer is full.

link: https://github.com/wesleyskap/onkai-unified-bus

From custom to industry standard: Histogram percentiles and the OpenTelemetry semantic bridge

Wesley Conde — Sat, 14 Mar 2026 12:05:00 +0000

The statistical illusion of the mean and the mathematics of percentiles

Metrics calculated solely on the arithmetic mean lie heavily to reliability engineers. A web server can display an excellent average latency of 15ms, masking that 1% of its users face catastrophic delays of 5 seconds due to Garbage Collector pauses. To map long-tail quality of service, we must compute precise statistical percentiles (p50, p90, p99).

Computing exact percentiles over time would require storing all physical measurements indefinitely in RAM, leading to memory leaks and resource exhaustion in production.

To solve this problem, Orkai utilizes a memory-bounded structure called a latencyReservoir. It stores a fixed sliding window of samples (capped at 2000 float64 entries) in a concurrency-safe manner. To compute a percentile (e.g. p99), we isolate the internal slice using a thread-safe copy (copy(sorted, r.samples)) to avoid race conditions, sort the copied slice, and extract the value at the proportional index.

func (r *latencyReservoir) extractPercentile(p float64) float64 {
    r.mu.Lock()
    defer r.mu.Unlock()

    n := len(r.samples)
    if n == 0 {
        return 0.0
    }
    // Create an isolated copy for safe sorting without corrupting concurrent memory
    sorted := make([]float64, n)
    copy(sorted, r.samples)
    sort.Float64s(sorted)

    idx := int(math.Ceil(p*float64(n))) - 1
    if idx < 0 {
        idx = 0
    } else if idx >= n {
        idx = n - 1
    }
    return sorted[idx]
}

Semantic bridges and double routing in openTelemetry

Most modern companies have standardized their observability stacks using the CNCF's OpenTelemetry (OTel) specification. However, refactoring every single legacy application to interface with the verbose OTel APIs is economically and technically unviable.

In Orkai, we solved this by developing lightweight semantic adapters (NewOTelTracer and NewOTelMetrics). They cleanly translate calls from our lightweight unified facades into native OpenTelemetry SDK structures.

Furthermore, we introduced Double Routing: when a latency metric is registered, our bridge forwards it simultaneously to both the remote OpenTelemetry Collector and the local InMemoryMetrics engine. This design allows you to keep scraping traditional /metrics via Prometheus for fast, low-overhead local dashboards, without degrading CPU performance or doubling processing latency.

[DIAGRAM_DOUBLE_ROUTING]

type otelMetrics struct {
    mu          sync.Mutex
    meter       metric.Meter
    instruments map[string]interface{}
    localEngine *InMemoryMetrics // Primary local Prometheus collector
}

func (o *otelMetrics) RecordLatency(name string, val float64, labels map[string]string) {
    o.mu.Lock()
    // 1. Primary local routing
    o.localEngine.RecordLatency(name, val, labels)
    histogram := o.getOrCreateHistogram(name)
    o.mu.Unlock()

    // 2. Concurrent propagation to the native OpenTelemetry SDK
    ctx := context.Background()
    histogram.Record(ctx, val, metric.WithAttributes(convertLabels(labels)...))
}

Technical terms explained

Outliers: Experimental measurements or telemetry records that deviate drastically from the typical operational envelope (e.g. a 10s database query in a cluster of 2ms requests).
Percentile: A statistical measure representing the threshold below which a given percentage of data falls. For instance, a p95 latency of 200ms means 95% of requests completed under 200ms.
Double Routing: The technique of multiplexing a single stream of telemetry data to multiple independent endpoints concurrently without blocking the main execution path.

link: https://github.com/wesleyskap/orkai-observability