Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
- API
  
  Reference documentation for the pkg.go.dev API
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

queue

package

Go to main page

Versions in this module

v0

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