Documentation
¶
Overview ¶
Package queue implements the persistent failed-message (dead-letter) queue on top of the sqlitedb KV store.
Each queued message is stored as its own row under the key
<queue name> || 0x00 || <8-byte big-endian time> || <8-byte big-endian seq>
so that enqueueing is O(1) (no read-modify-write of an aggregate blob) and fetching never destroys data: FetchFailedMsgs only reads, and the caller removes each row with Dequeue after the message has been processed. A crash mid-cycle therefore re-delivers instead of losing messages (at-least-once). Older releases stored the whole queue as a single CBOR list under the bare queue name; FetchFailedMsgs migrates such rows to the per-message layout on first encounter.
Index ¶
Constants ¶
const MaxQueueAge = 30 * 24 * time.Hour
MaxQueueAge caps how long a failed message is retried before being dropped at fetch.
Variables ¶
var ErrQueueing = errors.New("problem with persistent queue")
Functions ¶
func Dequeue ¶ added in v0.34.0
Dequeue removes a processed row returned by FetchFailedMsgs. Call it only after the outcome is durable (delivered, or re-queued as a fresh row): a crash before Dequeue re-delivers rather than loses. The delete survives ctx cancel so a shutdown mid-drain doesn't duplicate the row next run.
Types ¶
type Queued ¶ added in v0.34.0
Queued couples a fetched message with the row key it was read from so the caller can Dequeue exactly that row once the message has been processed.
func FetchFailedMsgs ¶
FetchFailedMsgs returns the failed messages queued under queueKey, oldest first, without removing them: rows stay in the database until the caller confirms processing via Dequeue, so a crash mid-resend re-delivers instead of losing messages.
Messages older than MaxQueueAge are dropped (their rows deleted) here. Legacy aggregate-blob queues written by older releases are transparently split into per-message rows on first encounter.
If any of the operations fail, the function logs an error and returns whatever could be read.
Source Files