| Safe Haskell | None |
|---|---|
| Language | GHC2021 |
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
| |
Instances
| Show MemoryQueueConfig Source # | |
Defined in Ecluse.Core.Queue.Memory Methods showsPrec :: Int -> MemoryQueueConfig -> ShowS # show :: MemoryQueueConfig -> String # showList :: [MemoryQueueConfig] -> ShowS # | |
| Eq MemoryQueueConfig Source # | |
Defined in Ecluse.Core.Queue.Memory Methods (==) :: MemoryQueueConfig -> MemoryQueueConfig -> Bool # (/=) :: MemoryQueueConfig -> MemoryQueueConfig -> Bool # | |
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 |
| -> 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
memQueueMaxDepthjobs; anenqueuethat would exceed the cap is rejected (the newest job is dropped) rather than growing memory without bound -- the load-bearing constraint, since a cold-cachenpm cienqueues thousands of jobs at once.enqueuenever 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 bymemoryQueueDropReportIntervalso a flood does not spam. - No redelivery;
ack/extendVisibility/deadLetterare no-ops. Unlike the cloud backends, there is no visibility-timeout in-flight tracking: areceiveremoves 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.