Documentation
¶
Overview ¶
Package queue is a durable, per-destination delivery queue.
Without it, a destination that is down for two minutes loses everything sent during those two minutes: retries happen in memory, and when they run out the message is recorded as failed and gone. With it, the message is on disk and keeps its place until the destination comes back.
Two design decisions shape everything else.
**Order is preserved per destination, and that means a queued destination stops taking the fast path.** HL7 is a stream of events about the same patients: an A01 admits, an A03 discharges. If a failed A01 goes to the queue while the next A03 is delivered directly, the receiving system sees a discharge for a patient it never admitted. So once anything is queued for a destination, everything for that destination queues behind it until the queue drains. That costs throughput during an outage and it is not optional, because the alternative is silent clinical nonsense. It is also the specific thing that goes wrong in other engines when somebody turns on concurrent queue threads to make a backlog drain faster.
**A queued message is an accepted message.** The sender is told AA, because the bytes are on disk and fsynced before the acknowledgement goes out. Saying AE would make a correctly functioning store-and-forward queue look like a fault and invite the sender to resend what we already hold.
Index ¶
- Constants
- type DepthRow
- type Item
- type Observer
- type Policy
- type Query
- type Sender
- type State
- type Store
- func (s *Store) Blocked(ctx context.Context, channel, destination string) (bool, error)
- func (s *Store) Depth(ctx context.Context) ([]DepthRow, error)
- func (s *Store) Drain(ctx context.Context, channel, destination, who string) (int64, error)
- func (s *Store) Enqueue(ctx context.Context, item Item) (int64, error)
- func (s *Store) Erase(ctx context.Context, ids []int64) (int64, error)
- func (s *Store) GaveUp(ctx context.Context, id int64, attempts int, cause string) error
- func (s *Store) List(ctx context.Context, q Query) ([]Item, int, error)
- func (s *Store) Next(ctx context.Context, channel, destination string, now time.Time) (*Item, time.Duration, error)
- func (s *Store) Payload(ctx context.Context, id int64) ([]byte, error)
- func (s *Store) PendingDestinations(ctx context.Context) ([]DepthRow, error)
- func (s *Store) Purge(ctx context.Context, before time.Time) (int64, error)
- func (s *Store) Reader() *sql.DB
- func (s *Store) Remove(ctx context.Context, id int64) (int64, error)
- func (s *Store) Retry(ctx context.Context, id int64, attempts int, next time.Time, cause string) error
- func (s *Store) RetryNow(ctx context.Context, id int64, channel, destination string) (int64, error)
- func (s *Store) ScanPayloads(ctx context.Context, ...) error
- func (s *Store) Skip(ctx context.Context, id int64, who string) (int64, error)
- func (s *Store) Succeeded(ctx context.Context, id int64, attempts int) error
- type Worker
Constants ¶
const ( DefaultBackoff = 5 * time.Second DefaultMaxBackoff = time.Minute DefaultTimeout = 30 * time.Second )
Defaults for anything left unset. The backoff ceiling is a minute rather than something larger because a receiver that comes back after an outage should be found quickly; waiting an hour to notice would turn a two-minute blip into an hour-long backlog.
const DefaultIdle = 2 * time.Second
DefaultIdle is the poll interval when the queue is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type DepthRow ¶
type DepthRow struct {
Channel string `json:"channel"`
Destination string `json:"destination"`
Pending int `json:"pending"`
Failed int `json:"failed"`
Oldest time.Time `json:"oldest,omitzero"`
// OldestSeconds is the age of the oldest pending item. This is the number to
// alert on: depth alone cannot distinguish a busy queue that is draining from
// a small one that has been stuck since Tuesday.
OldestSeconds float64 `json:"oldestSeconds"`
// MaxAttempts is the highest attempt count among pending items, which is what
// says "this is being retried and not progressing".
MaxAttempts int `json:"maxAttempts"`
}
DepthRow is the per-destination summary the dashboard shows.
type Item ¶
type Item struct {
ID int64 `json:"id"`
Channel string `json:"channel"`
Destination string `json:"destination"`
ControlID string `json:"controlId"`
MessageType string `json:"messageType"`
Raw []byte `json:"-"`
Size int `json:"size"`
State State `json:"state"`
Attempts int `json:"attempts"`
EnqueuedAt time.Time `json:"enqueuedAt"`
NextAttempt time.Time `json:"nextAttempt"`
LastAttempt time.Time `json:"lastAttempt,omitzero"`
LastError string `json:"lastError,omitempty"`
// Reason is why it entered the queue, kept separate from LastError so the
// original cause survives a series of different later failures.
Reason string `json:"reason,omitempty"`
}
Item is one queued delivery.
type Observer ¶
type Observer interface {
QueueDelivered(channel, destination string, attempts int, waited time.Duration)
QueueRetried(channel, destination string, attempts int, cause string)
QueueGaveUp(channel, destination string, attempts int, cause string)
}
Observer is told what the worker did, so metrics and alerts can be driven without the queue importing either.
type Policy ¶
type Policy struct {
// MaxAttempts is how many times the queue tries before giving up. Zero means
// keep trying, which is a legitimate choice for a feed that must not lose
// anything, but it has to be chosen rather than inherited.
MaxAttempts int
// Backoff is the first delay, doubling up to MaxBackoff.
Backoff time.Duration
MaxBackoff time.Duration
// Timeout bounds one attempt.
Timeout time.Duration
}
Policy bounds how hard a queued message is retried.
type Query ¶
type Query struct {
Channel string
Destination string
State State
// Limit defaults to 100 and is capped at 1000, because this feeds a page and
// an unbounded query against a large backlog would be a denial of service on
// ourselves.
Limit int
Offset int
}
Query selects items for the interface.
type Sender ¶
Sender delivers one message. It is the same contract the engine uses, so a destination's real sender drains the queue rather than a reimplementation of it — otherwise the retry path could succeed where the live path fails, or worse, the other way round.
type State ¶
type State string
State is where a queued message is in its life.
const ( // Pending is waiting for its next attempt. Pending State = "pending" // Delivered means it eventually got there. Kept briefly so somebody watching // a backlog drain can see it happening. Delivered State = "delivered" // Failed means attempts were exhausted, or an operator gave up on it. It stays // until removed, because a message that left the queue without arriving is // the one somebody needs to look at. Failed State = "failed" // Skipped means an operator deliberately abandoned it. Distinct from failed: // one is the system giving up and the other is a person deciding, and // conflating them loses who is accountable. Skipped State = "skipped" )
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store persists the queue.
func NewStore ¶
NewStore prepares the schema.
Pass the write handle and, separately, a read handle. Queue depth is polled by the dashboard, and that poll must not sit in front of a delivery.
func (*Store) Blocked ¶
Blocked reports whether anything is waiting for a destination, regardless of whether its next attempt is due.
This is what makes ordering work. A destination with a pending item must not take the direct path, or a later message would overtake an earlier one. Note it deliberately ignores next_attempt: an item backing off is still ahead in the queue.
func (*Store) Drain ¶
Drain skips every pending item for a destination. Used when a receiver has been down long enough that the backlog is no longer clinically useful — a week of stale admissions replayed at once can do more harm than never sending them.
func (*Store) Enqueue ¶
Enqueue adds a message, returning its id.
The write is committed before this returns, so a caller may acknowledge the sender once it has. That ordering is the entire promise of the queue and it must not be relaxed for throughput.
func (*Store) Erase ¶ added in v0.1.18
Erase deletes items whatever their state, for a data protection erasure: a pending delivery of an erased person's message is not sent. A worker that already took one finds it gone when it reports back, which changes nothing.
func (*Store) List ¶
List returns matching items, newest first, and the total count.
Payloads are not included. A queue page showing a hundred messages does not need a hundred payloads, and shipping them would put clinical content into a response that only needed counts.
func (*Store) Next ¶
func (s *Store) Next(ctx context.Context, channel, destination string, now time.Time) (*Item, time.Duration, error)
Next returns the head of a destination's queue when it is ready to be sent.
When the head exists but is still backing off, it returns a nil item and how long to wait. That distinction is the whole ordering guarantee, and getting it wrong is subtle: an earlier version selected the oldest item *whose next attempt was due*, which reads as reasonable and is badly broken. A message backing off after a failure is not due, so the query skipped it and returned a later message that had never been tried — and the later message went out first. The queue preserved order right up until the first failure, which is precisely when order starts to matter.
So the head is the head. If it is waiting, everything behind it waits too. That is the cost of ordering and it is not negotiable.
It returns one item, not a batch, because delivering in order means delivering one at a time. A batch would only help if they could be sent concurrently, which is the thing that reorders them.
func (*Store) PendingDestinations ¶
PendingDestinations lists every destination with work waiting, so a worker starting up knows what to drain without being told.
This is what makes the queue survive a restart rather than merely persist: the rows would be there either way, but nothing would look at them.
func (*Store) Reader ¶
Reader exposes the handle reports use, for callers in this package that build their own queries.
func (*Store) Remove ¶
Remove deletes items that have finished, one by one or in bulk.
Pending items are never deleted here. Deleting a message that has not been delivered and has not been explicitly abandoned would lose it silently, so it has to be skipped first — which records that somebody chose to.
func (*Store) Retry ¶
func (s *Store) Retry(ctx context.Context, id int64, attempts int, next time.Time, cause string) error
Retry records a failed attempt and schedules the next one.
func (*Store) RetryNow ¶
RetryNow brings forward the next attempt for one item, or for every pending item on a destination when id is zero.
It resets the attempt counter, because an operator retrying by hand has usually just fixed something, and counting the failures from before the fix against the new attempt budget would abandon the message immediately.
func (*Store) ScanPayloads ¶ added in v0.1.18
func (s *Store) ScanPayloads(ctx context.Context, fn func(id int64, channel, destination, state string, raw []byte) error) error
ScanPayloads calls fn for every item still held (pending, failed or given up), with its payload. It is for finding a data subject's queued messages, which have no identity index: the queue is short-lived, so reading it whole is acceptable.
type Worker ¶
type Worker struct {
Store *Store
Channel string
Destination string
Sender Sender
Policy Policy
Log *slog.Logger
Observer Observer
// Idle is how long to wait before looking again when there was nothing to do.
// Short enough that a recovered destination is noticed quickly, long enough
// that an empty queue is not a busy loop.
Idle time.Duration
// contains filtered or unexported fields
}
Worker drains one destination's queue, in order, one message at a time.
One worker per destination is the whole point. Running several would drain faster and reorder the messages, and an A03 arriving before its A01 is a receiving system being told about a discharge for a patient it has never heard of.
Source Files