ecluse:ecluse-core
Safe HaskellNone
LanguageGHC2021

Ecluse.Core.Queue.Memory

Description

The STM-backed in-memory MirrorQueue: the __bounded, best-effort production backend mirroring rolls over to when no @ECLUSE_QUEUEURL@ is set.

It honours the handle's contract (see Ecluse.Core.Queue for the enqueue / don't-ack-to-retry / no-nack conventions) and is built from the contract module's backend building blocks. See newBoundedInMemoryQueue for why it is correctness-safe (a dropped job is re-enqueued on the next demand) and why it deliberately does not redeliver.

Synopsis

Bounded in-memory production backend

data MemoryQueueConfig Source #

What the bounded in-memory backend needs: its depth cap and its idle-poll window. A record (like the SQS backend's SqsConfig) so each knob is named rather than a bare Int; build it with defaultMemoryQueueConfig for the production poll window.

Constructors

MemoryQueueConfig 

Fields

  • memQueueMaxDepth :: Int

    The maximum number of jobs the queue holds. A fresh enqueue past this cap is dropped-newest (the enqueue is rejected); a dropped job is safe, as it is re-enqueued on the next demand. Must be positive (the config layer enforces it).

  • memQueuePollWaitMicros :: Int

    The idle long-poll window in microseconds: how long a receive waits for a job before returning [] (an empty, healthy poll). Bounds the idle wait so the worker's liveness heartbeat keeps advancing -- see newBoundedInMemoryQueue.

defaultMemoryQueueConfig :: Int -> MemoryQueueConfig Source #

A MemoryQueueConfig for a given depth cap with the idle-poll window at its production default -- 20s, mirroring the SQS long-poll cadence (the SQS backend's defaultSqsConfig) and comfortably under the worker's heartbeat-staleness budget (workerHeartbeatStaleAfter), so an idle receive returns a healthy empty poll long before /livez would flag the loop stalled. The depth cap stays the operator-tunable knob; the poll window is a fixed cadence, exposed on the record only so a test can shorten it.

newBoundedInMemoryQueue Source #

Arguments

:: MemoryQueueConfig

The depth cap (and any future knobs).

-> (Int -> IO ())

Invoked on each report-worthy cap-overflow drop with the running total drops, so the composition root can log it (and, once the ecluse.mirror.* metric catalogue lands, increment a drop counter alongside).

-> IO MirrorQueue 

Build a bounded, best-effort in-memory MirrorQueue -- the production backend mirroring runs on when no ECLUSE_QUEUE__URL is set, a TBQueue shared between the serve path's enqueue and the worker's receive.

It is correctness-safe despite being lossy: mirroring is a demand-driven optimization over the always-available public upstream, so a job lost to the cap or to process teardown just means the package is served from public again and re-enqueued on the next pull -- a deferred performance win, never a correctness loss. That admits two deliberate departures from the cloud backends' contract:

  • Bounded, drop-newest on overflow. The queue holds at most memQueueMaxDepth jobs; an enqueue that would exceed the cap is rejected (the newest job is dropped) rather than growing memory without bound -- the load-bearing constraint, since a cold-cache npm ci enqueues thousands of jobs at once. enqueue never throws (it runs on the serve hot path), and each report-worthy drop invokes the injected drop callback with the running drop count, rate-limited by memoryQueueDropReportInterval so a flood does not spam.
  • No redelivery; ack / extendVisibility / deadLetter are no-ops. Unlike the cloud backends, there is no visibility-timeout in-flight tracking: a receive removes a job for good. A job whose processing fails is therefore not redelivered -- it is simply re-enqueued on the next demand. This bounds memory hardest (nothing is retained after delivery) and is admissible precisely because a lost job is safe. A terminal fault (deadLetter) is the same drop, since this backend has no dead-letter queue to route to; its observability is the worker's error log and metric, not a retained message.

receive is a bounded long-poll: it waits up to memQueuePollWaitMicros for a job, then drains up to memoryQueueBatchSize without blocking, or returns [] when the window lapses -- the in-process analogue of the cloud long-poll. The bound is load-bearing: on an idle queue the worker advances its liveness heartbeat only when receive returns (an empty poll is a healthy idle; a busy worker also beats after each completed job), so an idle receive that blocked forever would let the heartbeat go stale and /livez flag the loop stalled. The wait is the timeout-over-atomically idiom rather than registerDelay so it works on the non-threaded RTS too; an interrupted poll aborts the STM transaction, consuming nothing.

memoryQueueBatchSize :: Int Source #

The most jobs one receive delivers from the bounded in-memory backend. Held at the SQS batch cap so the worker -- which processes a batch sequentially -- sees the same bounded batch shape regardless of backend, rather than one poll returning a whole cold-cache burst. The worker advances its liveness heartbeat after each completed job (not once per poll), so this cap bounds per-poll work and memory, and is no longer the heartbeat's protection against a long batch.

memoryQueueDropReportInterval :: Int Source #

How many cap-overflow drops the bounded in-memory backend absorbs between warning reports. The first drop is always reported, then every multiple of this, so a sustained flood logs at most about one line per this many drops rather than one per dropped job.