🥮 mooncakes.io
    Sign in

    relay

    Distributed message queue with V3 reliability and Valkey support

    Download zip
    Author
    Version
    0.4.0
    License
    Apache-2.0
    Repository
    Last updated
    4 months ago
    Downloads
    133

    Metalymph/relay does not have a README file

    QueueError

    source
    pub suberror QueueError {
      OperationFailed(String)
    } derive(ToJson, 
    Debug    )

    QueueError represents errors related to queue lifecycle and operations.

    AdminReport

    source
    pub struct AdminReport {
      status : SystemStatus
      requires_attention : Bool
      queue_size : Int?
      dlq_size : Int?
      pending_count : Int64?
      summary : String
    }

    AdminReport represents a high-level operational report.

    AdminReport::has_incidents

    source
    fn AdminReport::has_incidents(self : AdminReport) -> Bool

    has_incidents returns true when operator attention is required.

    AdminReport::is_healthy

    source
    fn AdminReport::is_healthy(self : AdminReport) -> Bool

    is_healthy returns true when the report status is healthy.

    BackendCapabilities

    source
    pub struct BackendCapabilities {
      supports_ack : Bool
      supports_dlq : Bool
      supports_pending : Bool
      supports_consumer_groups : Bool
      supports_health_checks : Bool
      supports_telemetry : Bool
    }

    BackendCapabilities describes the features supported by a queue backend.

    BackendCapabilities::in_memory

    source

    BackendCapabilities::redis_lists

    source

    BackendCapabilities::redis_streams

    source

    BackendHealth

    source
    pub struct BackendHealth {
      healthy : Bool
      message : String?
    }

    Dashboard

    source
    pub struct Dashboard[T] {
      queue : RelayQueue[T]
      backend : InMemoryBackend[T]
    }

    Dashboard provides a visual representation of the RelayQueue state.

    Dashboard::new

    source
    fn[T] Dashboard::new(queue : RelayQueue[T], backend : InMemoryBackend[T]) -> Dashboard[T]

    new creates a new dashboard for the provided queue and backend.

    Dashboard::render

    source
    fn[T] Dashboard::render(self : Dashboard[T]) -> Unit

    render prints an ANSI color-coded snapshot of the current queue state.

    Dashboard::start_auto_render

    source
    fn[T] Dashboard::start_auto_render(self : Dashboard[T]) -> Unit

    start_auto_render starts an infinite loop of rendering.

    InMemoryBackend

    source
    pub struct InMemoryBackend[T] {
      queue : 
    Queue    [Message[T]]
      dlq : 
    Queue    [Message[T]]
      in_flight : 
    HashMap    [String, Message[T]]
      policy : RetryPolicy
      next_id : Int
      count : Int
      dlq_count : Int
    }

    InMemoryBackend uses @aqueue.Queue from moonbitlang/async for local, non-blocking storage.

    InMemoryBackend::ack

    source
    fn[T] InMemoryBackend::ack(self : InMemoryBackend[T], id : String) -> Unit

    ack confirms successful processing and removes from in-flight.

    InMemoryBackend::capabilities

    source
    fn[T] InMemoryBackend::capabilities(_ : InMemoryBackend[T]) -> BackendCapabilities

    InMemoryBackend::dlq_size

    source
    fn[T] InMemoryBackend::dlq_size(self : InMemoryBackend[T]) -> Int

    dlq_size returns the count of messages in the Dead Letter Queue.

    InMemoryBackend::health

    source

    InMemoryBackend::nack

    source
    async fn[T] InMemoryBackend::nack(self : InMemoryBackend[T], id : String) -> NackResult

    nack re-inserts the message into the queue or moves to DLQ if max attempts reached.

    InMemoryBackend::new

    source
    fn[T] InMemoryBackend::new(capacity : Int, policy? : RetryPolicy) -> InMemoryBackend[T]

    new creates a new InMemoryBackend with a fixed capacity and optional policy.

    InMemoryBackend::pop

    source
    async fn[T] InMemoryBackend::pop(self : InMemoryBackend[T]) -> Message[T]

    pop retrieves a message and marks it as in-flight.

    InMemoryBackend::push

    source
    async fn[T] InMemoryBackend::push(self : InMemoryBackend[T], v : T) -> Unit

    push adds a message to the internal queue.

    InMemoryBackend::queue_metrics

    source
    fn[T] InMemoryBackend::queue_metrics(self : InMemoryBackend[T]) -> QueueMetrics

    queue_metrics returns the current queue metrics.

    InMemoryBackend::size

    source
    fn[T] InMemoryBackend::size(self : InMemoryBackend[T]) -> Int

    size returns the cached count of messages in the main queue.

    InMemoryBackend::to_relay_queue

    source
    fn[T] InMemoryBackend::to_relay_queue(self : InMemoryBackend[T]) -> RelayQueue[T]

    to_relay_queue converts the backend into a generic RelayQueue interface.

    Incident

    source
    pub enum Incident {
      BackendUnavailable
      WorkerFailures
      DeadLetterMessages
      PendingMessages
    } derive(Eq, ToJson, 
    Debug    )

    Incident represents a specific operational problem in Relay.

    Incident::severity

    source
    fn Incident::severity(self : Incident) -> IncidentSeverity

    severity returns the operational severity of an incident.

    IncidentSeverity

    source
    pub enum IncidentSeverity {
      Info
      Warning
      Critical
    } derive(Eq, ToJson, 
    Debug    )

    IncidentSeverity describes the operational severity of an incident.

    IncidentSeverity::to_string

    source
    fn IncidentSeverity::to_string(self : IncidentSeverity) -> String

    Message

    source
    pub struct Message[T] {
      id : String
      payload : T
      attempt_count : Int
    }

    Message[T] represents a wrapper for data sent through the queue.

    Message::new

    source
    fn[T] Message::new(id : String, payload : T, attempt_count? : Int) -> Message[T]

    new creates a new message.

    NackResult

    source
    pub enum NackResult {
      Requeued
      SentToDlq
    }

    Result of a negative acknowledgement.

    OtelAttribute

    source
    pub struct OtelAttribute {
      key : String
      value : String
    }

    OtelAttribute represents a simple OpenTelemetry-compatible attribute.

    OtelRecord

    source
    pub struct OtelRecord {
      name : String
      timestamp_ms : Int64
      attributes : Array[OtelAttribute]
    }

    OtelRecord represents a normalized OpenTelemetry-compatible event record.

    PrometheusMetric

    source
    pub struct PrometheusMetric {
      name : String
      value : Float
    }

    PrometheusMetric represents a Prometheus-compatible metric sample.

    QueueInspection

    source
    pub struct QueueInspection {
      status : SystemStatus
      queue_size : Int?
      dlq_size : Int?
      pending_count : Int64?
      backend_healthy : Bool
    }

    QueueInspection represents an operational view of a queue.

    QueueInspection::has_dlq

    source
    fn QueueInspection::has_dlq(self : QueueInspection) -> Bool

    has_dlq returns true when DLQ messages are present.

    QueueInspection::has_pending

    source
    fn QueueInspection::has_pending(self : QueueInspection) -> Bool

    has_pending returns true when pending messages are present.

    QueueInspection::is_empty

    source
    fn QueueInspection::is_empty(self : QueueInspection) -> Bool

    is_empty returns true when the queue size is known and equal to zero.

    QueueInspection::summary

    source
    fn QueueInspection::summary(self : QueueInspection) -> String

    summary returns a compact operational summary.

    QueueMetrics

    source
    pub struct QueueMetrics {
      size : Int?
      dlq_size : Int?
      pending_count : Int64?
    }

    QueueMetrics is a backend-agnostic snapshot of queue state.

    RedisBackend

    source
    pub struct RedisBackend[T] {
      client : 
    Client
      key : String
      serialize : (T) -> String
      deserialize : (String) -> T
      next_id : Int
    }

    RedisBackend implements the storage interface for Relay using Redis/Valkey.

    RedisBackend::ack

    source
    async fn[T] RedisBackend::ack(_self : RedisBackend[T], _id : String) -> Unit

    ack confirms message processing.

    Current Valkey backend pop semantics remove the message from the list when BLPOP returns, so ack is a no-op until visibility-timeout tracking is added.

    RedisBackend::capabilities

    source
    fn[T] RedisBackend::capabilities(self : RedisBackend[T]) -> BackendCapabilities

    RedisBackend::health

    source
    async fn[T] RedisBackend::health(self : RedisBackend[T]) -> BackendHealth

    RedisBackend::nack

    source
    async fn[T] RedisBackend::nack(_self : RedisBackend[T], _id : String) -> NackResult

    nack is a no-op for the list-based Valkey backend.

    BLPOP removes the message immediately, so the backend cannot requeue by id.

    RedisBackend::new

    source
    fn[T] RedisBackend::new(client : 
    Client    , key : String, serialize~ : (T) -> String, deserialize~ : (String) -> T) -> RedisBackend[T]

    new creates a RedisBackend instance with the provided client and serializers.

    RedisBackend::pop

    source
    async fn[T] RedisBackend::pop(self : RedisBackend[T]) -> Message[T]

    pop extracts a message from the Redis queue in a blocking manner using BLPOP.

    RedisBackend::pop_nowait

    source
    async fn[T] RedisBackend::pop_nowait(self : RedisBackend[T]) -> Message[T]?

    pop_nowait extracts a message without blocking, returning None when empty.

    RedisBackend::push

    source
    async fn[T] RedisBackend::push(self : RedisBackend[T], v : T) -> Unit

    push sends a message to the Redis queue using RPUSH.

    RedisBackend::queue_metrics

    source
    fn[T] RedisBackend::queue_metrics(_self : RedisBackend[T]) -> QueueMetrics

    queue_metrics returns the current queue metrics.

    RedisBackend::queue_metrics_async

    source
    async fn[T] RedisBackend::queue_metrics_async(self : RedisBackend[T]) -> QueueMetrics

    RedisBackend::size

    source
    async fn[T] RedisBackend::size(self : RedisBackend[T]) -> Int

    size returns the current number of queued messages.

    RedisBackend::to_relay_queue

    source
    fn[T] RedisBackend::to_relay_queue(self : RedisBackend[T]) -> RelayQueue[T]

    to_relay_queue exports the backend as a generic RelayQueue.

    RedisStreamBackend

    source
    pub struct RedisStreamBackend[T] {
      client : 
    Client
      stream : String
      dlq_stream : String
      group : String
      consumer : String
      serialize : (T) -> String
      deserialize : (String) -> T
    }

    RedisStreamBackend::ack

    source
    async fn[T] RedisStreamBackend::ack(self : RedisStreamBackend[T], id : String) -> Unit

    ack acknowledges a stream message by id.

    RedisStreamBackend::capabilities

    source

    RedisStreamBackend::dead_letter

    source
    async fn[T] RedisStreamBackend::dead_letter(self : RedisStreamBackend[T], value : T) -> Unit

    dead_letter writes a payload to the configured DLQ stream.

    RedisStreamBackend::health

    source
    async fn[T] RedisStreamBackend::health(self : RedisStreamBackend[T]) -> BackendHealth

    RedisStreamBackend::init

    source
    async fn[T] RedisStreamBackend::init(self : RedisStreamBackend[T]) -> Unit

    RedisStreamBackend::nack

    source
    async fn[T] RedisStreamBackend::nack(_self : RedisStreamBackend[T], _id : String) -> NackResult

    nack leaves the stream message pending when only an id is available.

    Use nack_message when the payload is available and should be moved to DLQ.

    RedisStreamBackend::nack_message

    source
    async fn[T] RedisStreamBackend::nack_message(self : RedisStreamBackend[T], msg : Message[T]) -> Unit

    nack_message moves the message payload to the DLQ and acknowledges the original stream entry.

    RedisStreamBackend::new

    source
    fn[T] RedisStreamBackend::new(client : 
    Client    , stream : String, group : String, consumer : String, serialize~ : (T) -> String, deserialize~ : (String) -> T) -> RedisStreamBackend[T]

    RedisStreamBackend::new_with_dlq

    source
    fn[T] RedisStreamBackend::new_with_dlq(client : 
    Client    , stream : String, dlq_stream : String, group : String, consumer : String, serialize~ : (T) -> String, deserialize~ : (String) -> T) -> RedisStreamBackend[T]

    RedisStreamBackend::pending_count

    source
    async fn[T] RedisStreamBackend::pending_count(self : RedisStreamBackend[T]) -> Int64

    pending_count returns the number of unacknowledged messages.

    RedisStreamBackend::pop

    source
    async fn[T] RedisStreamBackend::pop(self : RedisStreamBackend[T]) -> Message[T]

    pop reads one message from the stream consumer group.

    RedisStreamBackend::pop_block

    source
    async fn[T] RedisStreamBackend::pop_block(self : RedisStreamBackend[T], block_ms : Int) -> Message[T]

    pop_block waits for a message for up to block_ms milliseconds.

    RedisStreamBackend::push

    source
    async fn[T] RedisStreamBackend::push(self : RedisStreamBackend[T], value : T) -> Unit

    RedisStreamBackend::queue_metrics

    source
    fn[T] RedisStreamBackend::queue_metrics(self : RedisStreamBackend[T]) -> QueueMetrics

    RedisStreamBackend::queue_metrics_async

    source
    async fn[T] RedisStreamBackend::queue_metrics_async(self : RedisStreamBackend[T]) -> QueueMetrics

    RedisStreamBackend::recover_or_pop

    source
    async fn[T] RedisStreamBackend::recover_or_pop(self : RedisStreamBackend[T], min_idle_ms : Int, block_ms : Int) -> Message[T]

    recover_or_pop first tries to recover a pending message, then falls back to blocking reads for new stream entries.

    RedisStreamBackend::recover_pending

    source
    async fn[T] RedisStreamBackend::recover_pending(self : RedisStreamBackend[T], min_idle_ms : Int) -> Message[T]?

    recover_pending claims one pending message idle for at least min_idle_ms.

    RedisStreamBackend::to_relay_queue

    source
    fn[T] RedisStreamBackend::to_relay_queue(self : RedisStreamBackend[T]) -> RelayQueue[T]

    to_relay_queue exports the stream backend as a generic RelayQueue.

    RedisStreamBackend::to_relay_queue_with_recovery

    source
    fn[T] RedisStreamBackend::to_relay_queue_with_recovery(self : RedisStreamBackend[T], min_idle_ms : Int, block_ms : Int) -> RelayQueue[T]

    to_relay_queue_with_recovery exports the stream backend as a RelayQueue whose pop path first recovers pending entries before blocking for new ones.

    RelayMetrics

    source
    pub struct RelayMetrics {
      worker : WorkerMetrics?
      queue : QueueMetrics?
      health : BackendHealth?
    }

    RelayMetrics combines queue and worker telemetry.

    RelayQueue

    source
    pub struct RelayQueue[T] {
      push_fn : async (T) -> Unit
      pop_fn : async () -> Message[T]
      ack_fn : async (String) -> Unit
      nack_fn : async (String) -> NackResult
      size_fn : () -> Int
    }

    RelayQueue is the main entry point for interacting with the message queue. It uses a closure-based backend abstraction to support multiple storage engines (Memory, Redis, etc.).

    RelayQueue::ack

    source
    async fn[T] RelayQueue::ack(self : RelayQueue[T], id : String) -> Unit

    ack confirms that a message has been processed successfully. The backend will permanently remove the message from its tracking system.

    RelayQueue::nack

    source
    async fn[T] RelayQueue::nack(self : RelayQueue[T], id : String) -> NackResult

    nack (negative acknowledgement) informs the queue that processing failed. The backend will typically re-queue the message for another attempt, unless it exceeds the RetryPolicy.

    RelayQueue::new

    source
    fn[T] RelayQueue::new(push~ : async (T) -> Unit, pop~ : async () -> Message[T], ack~ : async (String) -> Unit, nack~ : async (String) -> NackResult, size~ : () -> Int) -> RelayQueue[T]

    new creates a new RelayQueue with the provided backend functions.

    RelayQueue::pop

    source
    async fn[T] RelayQueue::pop(self : RelayQueue[T]) -> Message[T]

    pop removes and returns a value from the queue. Suspends if the queue is empty. In V3, a popped message is considered "In Flight" until acknowledged.

    RelayQueue::push

    source
    async fn[T] RelayQueue::push(self : RelayQueue[T], value : T) -> Unit

    push adds a value to the queue. Asynchronous and thread-safe via the underlying backend.

    RelayQueue::size

    source
    fn[T] RelayQueue::size(self : RelayQueue[T]) -> Int

    size returns the current number of items in the queue.

    RelaySnapshot

    source
    pub struct RelaySnapshot {
      collected_at_ms : Int64
      metrics : RelayMetrics
    }

    RelaySnapshot represents a metrics snapshot collected at a specific time.

    RelaySnapshot::dlq_size

    source
    fn RelaySnapshot::dlq_size(self : RelaySnapshot) -> Int?

    dlq_size returns the known DLQ size, if available.

    RelaySnapshot::has_failures

    source
    fn RelaySnapshot::has_failures(self : RelaySnapshot) -> Bool

    has_failures returns true when worker failures or ack/nack failures are present.

    RelaySnapshot::is_healthy

    source
    fn RelaySnapshot::is_healthy(self : RelaySnapshot) -> Bool

    is_healthy returns true when health is present and healthy.

    RelaySnapshot::new

    source

    new creates a timestamped RelaySnapshot.

    RelaySnapshot::pending_count

    source
    fn RelaySnapshot::pending_count(self : RelaySnapshot) -> Int64?

    pending_count returns the known pending message count, if available.

    RelaySnapshot::queue_size

    source
    fn RelaySnapshot::queue_size(self : RelaySnapshot) -> Int?

    queue_size returns the known queue size, if available.

    RelaySnapshot::requires_attention

    source
    fn RelaySnapshot::requires_attention(self : RelaySnapshot) -> Bool

    requires_attention returns true when the snapshot is degraded or unhealthy.

    RelaySnapshot::status

    source

    status returns a high-level health status for the snapshot.

    RelaySnapshot::status_message

    source
    fn RelaySnapshot::status_message(self : RelaySnapshot) -> String

    status_message returns a short human-readable status summary.

    RetryPolicy

    source
    pub struct RetryPolicy {
      max_attempts : Int
    }

    RetryPolicy defines how the queue handles failed message processing.

    RetryPolicy::default

    source
    fn RetryPolicy::default() -> RetryPolicy

    default_policy provides a standard policy with 3 max attempts.

    RetryPolicy::new

    source
    fn RetryPolicy::new(max_attempts : Int) -> RetryPolicy

    new_policy creates a custom retry policy.

    SystemStatus

    source
    pub enum SystemStatus {
      Healthy
      Degraded
      Unhealthy
    } derive(Eq, ToJson, 
    Debug    )

    SystemStatus summarizes the current observable state of Relay.

    SystemStatus::to_string

    source
    fn SystemStatus::to_string(self : SystemStatus) -> String

    TelemetryEvent

    source
    pub struct TelemetryEvent {
      timestamp_ms : Int64
      status : SystemStatus
      severity : IncidentSeverity?
      incident_count : Int
      summary : String
    }

    TelemetryEvent is a normalized event that can later be exported to OpenTelemetry, Prometheus, JSON, logs, or other observability sinks.

    TelemetryEvent::to_otel_record

    source
    fn TelemetryEvent::to_otel_record(self : TelemetryEvent) -> OtelRecord

    to_otel_record converts a Relay telemetry event into an OpenTelemetry-compatible record.

    TelemetryEvent::to_prometheus_metrics

    source
    fn TelemetryEvent::to_prometheus_metrics(self : TelemetryEvent) -> Array[PrometheusMetric]

    to_prometheus_metrics converts a telemetry event into Prometheus-compatible metrics.

    WorkerMetrics

    source
    pub struct WorkerMetrics {
      processed : Int
      succeeded : Int
      failed : Int
      retries : Int
      dlq_routed : Int
      ack_failed : Int
      nack_failed : Int
      uptime_ms : Int64
      last_success_at_ms : Int64?
      last_failure_at_ms : Int64?
    }

    WorkerMetrics is a backend-agnostic snapshot of worker activity.

    WorkerPool

    source
    pub struct WorkerPool[T] {
      queue : RelayQueue[T]
      concurrency : Int
      stats : WorkerStats
    }

    WorkerPool manages a group of concurrent workers that process messages from the queue.

    WorkerPool::metrics

    source
    fn[T] WorkerPool::metrics(self : WorkerPool[T]) -> WorkerMetrics

    metrics returns the current worker processing counters.

    WorkerPool::new

    source
    fn[T] WorkerPool::new(queue : RelayQueue[T], concurrency~ : Int) -> WorkerPool[T]

    new creates a new worker pool with the specified concurrency.

    WorkerPool::run

    source
    async fn[T] WorkerPool::run(self : WorkerPool[T], handler : async (T) -> Unit) -> Unit

    run starts the pool. Each worker is launched as a background task within a TaskGroup.

    WorkerPool::stats

    source
    fn[T] WorkerPool::stats(self : WorkerPool[T]) -> WorkerStats

    stats returns the current worker processing counters.

    WorkerStats

    source
    pub struct WorkerStats {
      processed : Int
      succeeded : Int
      failed : Int
      retries : Int
      dlq_routed : Int
      ack_failed : Int
      nack_failed : Int
      started_at_ms : Int64
      last_success_at_ms : Int64?
      last_failure_at_ms : Int64?
    }

    WorkerStats tracks basic worker processing counters.

    WorkerStats::has_failures

    source
    fn WorkerStats::has_failures(self : WorkerStats) -> Bool

    WorkerStats::new

    source

    new creates empty worker stats.

    WorkerStats::uptime_ms

    source
    fn WorkerStats::uptime_ms(self : WorkerStats) -> Int64

    collect

    source
    fn collect(worker : WorkerMetrics?, queue : QueueMetrics?, health : BackendHealth?) -> RelayMetrics

    collect combines worker, queue, and backend health telemetry into a single snapshot.

    collect_queue

    source
    fn collect_queue(queue : QueueMetrics, health : BackendHealth) -> RelayMetrics

    collect_queue creates a telemetry snapshot containing only queue metrics.

    collect_snapshot

    source
    fn collect_snapshot(worker : WorkerMetrics?, queue : QueueMetrics?, health : BackendHealth?) -> RelaySnapshot

    collect_snapshot combines telemetry and wraps it in a timestamped snapshot.

    collect_worker

    source
    fn collect_worker(worker : WorkerMetrics, health : BackendHealth) -> RelayMetrics

    collect_worker creates a telemetry snapshot containing only worker metrics.

    detect_incidents

    source
    fn detect_incidents(report : AdminReport) -> Array[Incident]

    detect_incidents analyzes an administrative report.

    detect_incidents_with_capabilities

    source
    fn detect_incidents_with_capabilities(report : AdminReport, capabilities : BackendCapabilities) -> Array[Incident]

    from_report

    source
    fn from_report(report : AdminReport, incidents : Array[Incident]) -> TelemetryEvent

    from_report converts an administrative report and detected incidents into a normalized telemetry event.

    has_incidents

    source
    fn has_incidents(report : AdminReport) -> Bool

    has_incidents returns true when incidents are detected.

    highest_severity

    source
    fn highest_severity(incidents : Array[Incident]) -> IncidentSeverity?

    highest_severity returns the highest severity among incidents.

    incident_count

    source
    fn incident_count(report : AdminReport) -> Int

    incident_count returns the number of detected incidents.

    inspect

    source
    fn inspect(snapshot : RelaySnapshot) -> QueueInspection

    inspect builds an operational queue inspection from a telemetry snapshot.

    report

    source
    fn report(inspection : QueueInspection) -> AdminReport

    report builds an administrative report from a queue inspection.

    start_worker

    source
    async fn[T] start_worker(queue : RelayQueue[T], handler : (Message[T]) -> Unit) -> Unit

    start_worker starts an infinite loop that pops messages from the queue and passes them to a handler.

    Source Files

    Powered by MoonBit

    Site sourceReport issuePackagesBuild queueSkillsStatistics

    © 2026 mooncakes.io