queue

Documentation

Overview

Package queue is the lead queue: dispatch, leases, and recovery.

Leads are LEASED rather than marked running (§9.4). The difference only shows up when something goes wrong, which is when it matters: a worker that dies mid-lead leaves a `running` row nothing will ever revisit, and rev 1 had no way out of that. A lease expires, and an expired lease is a lead the next sweep puts back.

The single property everything else rests on: a lead is dispatched to at most one worker. Two workers running the same lead pay for it twice, and the ledger cannot detect that — both charges are real, both are correctly recorded, and the budget simply drains faster than the work justifies.

Index

• Constants
• Variables
• type Lease
• type Queue
  • func New(st store.Store, ttl time.Duration) *Queue
  • func (q *Queue) Complete(ctx context.Context, l *Lease, status core.LeadStatus) error
  • func (q *Queue) LeaseNext(ctx context.Context, sessionID, owner string) (*Lease, error)
  • func (q *Queue) Push(ctx context.Context, leads []core.Lead) error
  • func (q *Queue) Release(ctx context.Context, l *Lease) error
  • func (q *Queue) Renew(ctx context.Context, l *Lease) (bool, error)
  • func (q *Queue) SetClock(now func() time.Time)
  • func (q *Queue) Stats(ctx context.Context, sessionID string) (Stats, error)
  • func (q *Queue) Sweep(ctx context.Context, sessionID string) (int, error)
  • func (q *Queue) TTL() time.Duration
• type Stats
  • func (s Stats) Pending() int

Constants

const DefaultLeaseTTL = 5 * time.Minute

DefaultLeaseTTL is how long a worker may hold a lead without a heartbeat.

Long enough that a slow fetch plus a slow model call does not lose the lease, short enough that a crash does not strand work for the length of a session.

Variables

var ErrLeaseLost = errors.New("queue: lease lost")

ErrLeaseLost means the lease was taken before the operation could complete.

Types

type Lease

type Lease struct {
	Lead  *core.Lead
	Owner string
}

Lease is a worker's claim on one lead.

type Queue

type Queue struct {
	// contains filtered or unexported fields
}

Queue dispatches leads for one session.

func New

func New(st store.Store, ttl time.Duration) *Queue

New creates a queue. A zero ttl uses DefaultLeaseTTL.

func (*Queue) Complete

func (q *Queue) Complete(ctx context.Context, l *Lease, status core.LeadStatus) error

Complete marks a lead finished.

func (*Queue) LeaseNext

func (q *Queue) LeaseNext(ctx context.Context, sessionID, owner string) (*Lease, error)

LeaseNext claims the highest-priority queued lead, or returns nil when the queue is empty.

func (*Queue) Push

func (q *Queue) Push(ctx context.Context, leads []core.Lead) error

Push adds leads to the queue.

Depth is carried from the caller. Note that the depth CAP is enforced by the executor's own round counter, not by walking the tree, so Lead.ParentID is currently informational and is not set — the tree is for the trace view, and M4's lineage guard (§11.4) is what will need it populated.

func (*Queue) Release

func (q *Queue) Release(ctx context.Context, l *Lease) error

Release returns a lead to the queue without completing it.

func (*Queue) Renew

func (q *Queue) Renew(ctx context.Context, l *Lease) (bool, error)

Renew extends a lease.

Reports false when the lease is gone — swept as expired, and possibly already re-leased to another worker. A worker that keeps going after that is racing the one that now owns the lead, so the caller must stop rather than finish.

func (*Queue) SetClock

func (q *Queue) SetClock(now func() time.Time)

SetClock overrides time, for tests.

func (*Queue) Stats

func (q *Queue) Stats(ctx context.Context, sessionID string) (Stats, error)

Stats counts the session's leads by status.

func (*Queue) Sweep

func (q *Queue) Sweep(ctx context.Context, sessionID string) (int, error)

Sweep requeues leads whose lease has expired.

sessionID scopes it. A running executor must pass its own: an unscoped sweep requeues another live process's in-flight leases, and since nothing stops that process finishing its lead, both end up running it and both settle a charge. Confirmed reachable with a single worker — the database has no exclusive lock.

Empty sessionID sweeps every running session, which is what boot recovery needs (§9.4): after a crash every lead the dead process held is leased with an expiry in the past, and without this they stay that way.

func (*Queue) TTL

func (q *Queue) TTL() time.Duration

TTL is the lease duration in force, so a caller can size a heartbeat from it.

type Stats

type Stats struct {
	Queued, Leased, Done, Failed, Cached int
}

Stats reports the lead counts by status.

func (Stats) Pending

func (s Stats) Pending() int

Pending reports work that is not finished — queued plus in flight.

The loop does NOT use this to decide when it is done: it stops when LeaseNext returns nil, because a lead another worker holds is not work this one can take. Pending is for reporting.