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, id int64) 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, id int64, atraso time.Duration) 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 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.
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 ¶ added in v0.7.0
Recuperar returns to the queue the items claimed longer ago than `limite`.
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.
Source Files