Documentation
¶
Overview ¶
Package queue is the engine's persistent queue.
The queue lives in Postgres, not in an in-memory channel: "never depend exclusively on an in-memory channel for critical jobs". A process that dies with items in a channel loses work; one that dies with items in a table does not.
The claim uses `FOR UPDATE SKIP LOCKED`, the standard pattern for a queue in Postgres: several dispatchers compete over the same table without blocking each other and without handing out the same item twice. The alternative -- a SELECT followed by an UPDATE -- has a race between the two statements.
Index ¶
- type Item
- type Queue
- func (q *Queue) Claim(ctx context.Context, worker string, limite int) ([]Item, error)
- func (q *Queue) Done(ctx context.Context, it Item) error
- func (q *Queue) Enqueue(ctx context.Context, runID uuid.UUID, priority int, availableAt time.Time) error
- func (q *Queue) Recover(ctx context.Context, limite time.Duration) ([]Item, error)
- func (q *Queue) Release(ctx context.Context, it Item, atraso time.Duration) error
- func (q *Queue) Renew(ctx context.Context, it Item) (held bool, err error)
- func (q *Queue) Size(ctx context.Context) (pending, claimed int, err error)
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Item ¶
type Item struct {
ID int64
RunID uuid.UUID
Priority int
AvailableAt time.Time
// ClaimedAt is the claim this holder knows: reivindicado_em as the claim
// wrote it. It is the fence on everything the holder does afterwards --
// Renew, Done, Release -- because a reclaim always writes a new one, so the
// pair (ID, ClaimedAt) names one claim and not merely one item.
//
// Zero means "no claim", which is what the sweep hands back: it leaves the
// item unclaimed, and the orphan path's own Done and Release are fenced on
// it staying that way.
ClaimedAt time.Time
}
Item is one entry in the queue.
type Queue ¶
type Queue struct {
// contains filtered or unexported fields
}
Queue operates on queue_items.
func (*Queue) Claim ¶
Claim claims up to `limite` items for this worker.
The limit is how concurrency is enforced: the dispatcher asks only for the slots it has free. There is no path in which more items leave the queue than concurrency allows, because whoever counts the slots is whoever asks.
Claims are serialized across processes by an advisory lock (see claimLock).
func (*Queue) Done ¶
Done removes the item: the work finished and does not come back.
Fenced on the claim (#121). A holder whose claim the sweep took and somebody else claimed again would otherwise delete the item that other attempt is running -- and the queue would forget a run still in flight.
func (*Queue) Enqueue ¶
func (q *Queue) Enqueue(ctx context.Context, runID uuid.UUID, priority int, availableAt time.Time) error
Enqueue puts a run in the queue. `disponivelEm` zero means NOW, measured by the DATABASE's clock.
`ON CONFLICT DO NOTHING` on run_id's unique: enqueueing the same run twice is a no-op, not an error. That is the behaviour §29 asks for -- the operation tolerates repetition.
The clock difference matters: the process's clock can be a few milliseconds ahead of Postgres's, and an item written with the application's `time.Now()` stays invisible until the database catches up. Nothing is lost -- the next cycle picks it up -- but it is unexplainable latency, and it is what made a concurrency test hand out 4 items where 5 were ready.
func (*Queue) Recover ¶
Recuperar returns to the queue the items whose holder has been silent longer than `limite`: no renewal for that long, or none at all and a claim that old. Measuring the claim alone took runs from live holders (#121).
The items come back with no claim (a zero ClaimedAt), because that is what this leaves behind.
It is the safety net against a dead worker: without it, an item claimed by a process that crashed would stay stuck forever. That was exactly the failure mode of the zombie runs that jammed pipelines for 33 days in the previous system.
func (*Queue) Release ¶
Release returns the item to the queue, available after `atraso`.
Used on retry and when a dispatcher is interrupted before finishing: the item becomes free again instead of staying stuck on a worker that died.
Fenced on the claim, as Done is, so a stale holder cannot hand somebody else's run back to the queue.
func (*Queue) Renew ¶
Renew says the claim is still held. It is fenced on the claim itself, not on the item: a reclaim writes a new reivindicado_em, so a stale holder's renewal matches nothing -- and nothing is the answer that tells it it lost.
`held` false with no error is that answer. An error is the database not answering, which says nothing about the claim either way.
Source Files