Documentation
¶
Overview ¶
Package queue turns things that happened into work someone has to do.
Why the state lives in the store ¶
A reservations queue is not a message broker topic, and the difference decides the architecture. A broker queue is a transport: a message is delivered, acknowledged and gone. A reservations queue is a worklist that has to be listed, counted, filtered and re-read, and whose items survive being worked because "who cleared this and when" is the question asked after an interline dispute. Those are database semantics, not transport semantics, so queue state is held in store.QueueStore alongside the records it refers to.
Where an external system plugs in ¶
What an external queueing system is genuinely good at is the other half: telling something that work has arrived. Publisher is that seam. A placement is written to the store first and published second, in the same order and for the same reason as capture-before-parse on the inbound path: if the publish fails the work still exists and will be found by the next reader, whereas a publish that succeeded before the write could announce work nobody can look up.
So: Postgres today, another store.QueueStore tomorrow if the worklist wants to live somewhere else, and a Publisher whenever a robot or an external broker needs to be woken rather than to poll.
Index ¶
Constants ¶
const ( DefaultPendingAfter = 6 * time.Hour DefaultTicketingLead = 24 * time.Hour DefaultSweepLimit = 500 )
Sweep defaults.
Variables ¶
This section is empty.
Functions ¶
func ForStatus ¶
ForStatus returns the queue a segment status belongs on and a stable reason code, or ok false when the status needs no attention.
This is where a partner's answer becomes work. A confirmation has to reach whoever holds the other end of the itinerary; a refusal has to be rebooked by someone; a waitlist has to be watched. Requests we ourselves originated are silent, because waiting for an answer is not yet a problem -- that is the Sweeper's job once enough time has passed.
Statuses are read after a message has been applied, so a partner's reply code has usually already been folded into the holding code it implies: KK becomes HK, US becomes HL. Refusals keep the reply code, since nothing is held.
Types ¶
type Canceller ¶
type Canceller interface {
// CancelExpired withdraws every live segment and notifies the carriers.
// It returns the carriers that could not be told, which is the part the
// caller must not treat as success.
CancelExpired(ctx context.Context, locator, reason string) (unreachable []string, err error)
}
Canceller cancels a booking whose time limit has passed.
type Manager ¶
type Manager struct {
Store store.QueueStore
// Publish is optional. Nil means nothing external is notified.
Publish Publisher
Log *slog.Logger
// Now is overridable for tests.
Now func() time.Time
// Notify is called after a placement lands, for the console event bus.
Notify func(item *store.QueueItem)
}
Manager places records on queues.
func (*Manager) Place ¶
Place puts a record on a queue, reporting whether it was newly placed.
A repeat placement for the same reason while the item is still pending is not an error: it is a sweeper doing its job on a schedule. It reports false so a caller can tell "I created work" from "the work was already there".
type Publisher ¶
Publisher is an optional sink notified after a placement is durable.
Implementations may be an external broker, a webhook, or an in-process channel feeding a robot. A Publisher must not be relied on for durability: the item is already stored before it is called, and an error from it is logged rather than propagated, because failing the placement would discard work that has already been recorded.
type PublisherFunc ¶
PublisherFunc adapts a function to Publisher.
type Sweeper ¶
type Sweeper struct {
Records store.Store
Queues *Manager
Log *slog.Logger
// PendingAfter is how long a requested segment may go unanswered before it
// becomes someone's problem. Zero uses DefaultPendingAfter.
PendingAfter time.Duration
// TicketingLead is how far ahead of a ticketing deadline to raise it. Zero
// uses DefaultTicketingLead.
TicketingLead time.Duration
// Limit bounds records handled per pass. The store returns the most
// overdue first, so this drops the least urgent work rather than
// concealing all of it. Zero uses DefaultSweepLimit.
Limit int
// Cancel, when set, cancels a booking whose ticketing time limit has
// passed. Nil leaves the record alone and only raises it on a queue.
//
// It is an interface because the package that implements it imports this
// one, and it is optional because auto-cancel is a real cancellation: it
// gives seats back and tells the carriers. A deployment should have to ask
// for that rather than discover it.
Cancel Canceller
Now func() time.Time
}
Sweeper places records on queues for things that happen because time passed rather than because a message arrived.
It is the half of queueing that nothing else can do. A partner that answers puts work on a queue by answering; a partner that never answers puts work on no queue at all unless something notices the silence. Same for a ticketing time limit: the deadline passing is not an event anyone sends.
The due-date predicates are in the query, not in this loop, and that is a correctness property rather than an optimisation. A pass used to read the most recently updated records and look for stale ones among them, which is inverted -- the freshest records are by definition not the stale ones. Above a few hundred records a ticketing time limit could never fire and an unanswered segment could never be raised, with no error and no log line.
Source Files