queue

package

Go to main page

Versions in this module

v0

v0.5.1

Oct 4, 2026

v0.5.0

Oct 4, 2026

Changes in this version

+ type HeartbeatMismatchError struct

+ Expected string

+ Item *Work

+ TTLSeconds int64

+ func (e *HeartbeatMismatchError) Error() string

+ func (e *HeartbeatMismatchError) Unwrap() error

type LeaseKeeper

+ func (k *LeaseKeeper) Renew(ctx context.Context) error

type Queue

+ func (q *Queue) RequeueAfter(ctx context.Context, db DB, item *Item, delay time.Duration) error

v0.4.0

Sep 17, 2026

Changes in this version

+ const WindDown

+ type AbandonedWork struct

+ ID domain.ID

+ SessionID domain.ID

type Item

+ Chain int

+ ChainGrading bool

+ ThreadID domain.ID

type Queue

+ func (q *Queue) CancelThread(ctx context.Context, db DB, sessionID, threadID domain.ID) error

+ func (q *Queue) EnqueueOutputsHarvest(ctx context.Context, db DB, envID, sessionID domain.ID, chainGrading bool) (bool, error)

+ func (q *Queue) EnqueueThread(ctx context.Context, db DB, envID, sessionID, threadID domain.ID, kind Kind) (bool, error)

+ func (q *Queue) FinalizeAbandoned(ctx context.Context, db DB, envID, workID domain.ID) (bool, error)

+ func (q *Queue) HasUndrainedWork(ctx context.Context, envID domain.ID) (bool, error)

+ func (q *Queue) ListAbandoned(ctx context.Context, envID domain.ID) ([]AbandonedWork, error)

+ func (q *Queue) PollOn(ctx context.Context, db DB, envID domain.ID, reclaim time.Duration) (*Work, error)

+ func (q *Queue) RequeueAsGrading(ctx context.Context, db DB, item *Item) error

+ func (q *Queue) RequeueSettlement(ctx context.Context, db DB, item *Item, chain int) error

+ func (q *Queue) StopWith(ctx context.Context, db DB, envID, workID domain.ID, force bool) (*Work, error)

v0.3.0

Aug 16, 2026

Changes in this version

+ var ErrWorkStalled = errors.New("queue: work item stalled")

type Kind

+ const MCPExec

type LeaseKeeper

+ func (k *LeaseKeeper) Progress()

v0.2.0

Aug 7, 2026

Changes in this version

+ const MetricQueueDepth

+ const MetricQueuePending

+ const MetricQueueWorkersPolling

type Item

+ TraceContext map[string]string

type Kind

+ const OutputsHarvest

+ const WebExec

+ type LeaseKeeper struct

+ func (k *LeaseKeeper) Close() error

type Queue

+ func (q *Queue) CancelSession(ctx context.Context, db DB, sessionID domain.ID) error

+ func (q *Queue) KeepLease(ctx context.Context, item *Item, ttl time.Duration) (context.Context, *LeaseKeeper)

+ func (q *Queue) RegisterMetrics() (metric.Registration, error)

v0.1.0

Jul 16, 2026

Changes in this version

+ const NoHeartbeat

+ var ErrHeartbeatMismatch = errors.New("queue: heartbeat precondition failed")

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

+ var ErrWorkConflict = errors.New("queue: work item is in a conflicting state")

+ var ErrWorkNotFound = errors.New("queue: work item not found")

+ type DB interface

+ Exec func(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)

+ QueryRow func(ctx context.Context, sql string, args ...any) pgx.Row

+ type HeartbeatResult struct

+ LastHeartbeat time.Time

+ LeaseExtended bool

+ State string

+ TTLSeconds int64

+ type Item struct

+ EnvironmentID domain.ID

+ ID domain.ID

+ Kind Kind

+ Lease time.Time

+ Reclaimed bool

+ SessionID domain.ID

+ type Kind string

+ const ModelTurn

+ const ToolExec

+ type Queue struct

+ func New(pool *pgxpool.Pool) *Queue

+ func (q *Queue) Ack(ctx context.Context, envID, workID domain.ID) (*Work, error)

+ func (q *Queue) Assert(ctx context.Context, db DB, item *Item) error

+ func (q *Queue) Claim(ctx context.Context, kind Kind, ttl time.Duration) (*Item, error)

+ func (q *Queue) Complete(ctx context.Context, db DB, item *Item) error

+ func (q *Queue) Enqueue(ctx context.Context, db DB, envID, sessionID domain.ID, kind Kind) (bool, error)

+ func (q *Queue) Extend(ctx context.Context, item *Item, ttl time.Duration) error

+ func (q *Queue) GetWork(ctx context.Context, envID, workID domain.ID) (*Work, error)

+ func (q *Queue) Heartbeat(ctx context.Context, envID, workID domain.ID, expected string, ...) (*HeartbeatResult, error)

+ func (q *Queue) ListWork(ctx context.Context, envID domain.ID, after bool, afterT time.Time, ...) ([]*Work, error)

+ func (q *Queue) Poll(ctx context.Context, envID domain.ID, reclaim time.Duration) (*Work, error)

+ func (q *Queue) RecordPoll(ctx context.Context, envID domain.ID, workerID string) error

+ func (q *Queue) Requeue(ctx context.Context, db DB, item *Item) error

+ func (q *Queue) Stats(ctx context.Context, envID domain.ID) (*WorkStats, error)

+ func (q *Queue) Stop(ctx context.Context, envID, workID domain.ID, force bool) (*Work, error)

+ func (q *Queue) UpdateMetadata(ctx context.Context, envID, workID domain.ID, upserts map[string]string, ...) (*Work, error)

+ type Work struct

+ AcknowledgedAt *time.Time

+ CreatedAt time.Time

+ EnvironmentID domain.ID

+ ID domain.ID

+ LastHeartbeat *time.Time

+ Metadata map[string]string

+ SessionID domain.ID

+ StartedAt *time.Time

+ State string

+ StopRequestedAt *time.Time

+ StoppedAt *time.Time

+ TraceContext map[string]string

+ type WorkStats struct

+ Depth int64

+ OldestQueuedAt *time.Time

+ Pending int64

+ WorkersPolling int64

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL