Help us shape the future of Go! Take the 2026 Go Developer Survey

sqlqueue

package
v0.0.0-...-d88842b Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 29, 2026 License: Apache-2.0 Imports: 18 Imported by: 0

Documentation

Overview

Package sqlqueue is the Postgres queue behind the llm-d-async sql transport. Requests hash into Partitions partitions per queue, and each dispatcher leases a share of them. A request is pending while its dispatch_epoch is 0; dispatch stamps the partition epoch, and a new owner bumps the epoch and resets older stamps, which redelivers whatever the previous owner had in flight.

Index

Constants

View Source 
const DefaultMaxConns = 32

DefaultMaxConns is how many connections a Store opens and keeps when Open is not given WithMaxConns.

View Source 
const Partitions = 64

Variables

View Source 
var ErrQuotaHolderLapsed = errors.New("sqlqueue: quota holder lease lapsed")

ErrQuotaHolderLapsed means the holder's lease expired; its slots no longer count and it must register again under a new name.

Functions

This section is empty.

Types

type Completion 

type Completion struct {
	Key
	Attempt   int64
	Route     string
	Payload   string
	ExpiresAt int64
}

type Consumer 

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

Consumer is one dispatcher's partitions and in-flight requests on a queue. 

owned ──(above share)──> draining ──(nothing in flight, or drainTimeout)──> released
  ^                          │
  └───────(below share)──────┘

func NewConsumer 

func NewConsumer(store *Store, queue, owner string, leaseTTL, drainTimeout time.Duration) *Consumer

func (*Consumer) Abandon 

func (c *Consumer) Abandon(stamps ...Stamp)

Abandon hands requests whose outcome could not be written to the next Rebalance.

func (*Consumer) Ack 

func (c *Consumer) Ack(ctx context.Context, completions []Completion) ([]bool, error)

func (*Consumer) Close 

func (c *Consumer) Close(ctx context.Context) error

func (*Consumer) Defer 

func (c *Consumer) Defer(ctx context.Context, stamps []Stamp, notBefore int64) error

Defer is Undispatch for requests that must not dispatch again before notBefore.

func (*Consumer) Poll 

func (c *Consumer) Poll(ctx context.Context, now time.Time, limit int) ([]Request, error)

Poll marks the consumer for reconciliation on error, since the dispatch may have committed.

func (*Consumer) Rebalance 

func (c *Consumer) Rebalance(ctx context.Context, now time.Time) error

func (*Consumer) Retry 

func (c *Consumer) Retry(ctx context.Context, stamp Stamp, notBefore int64, envelope string) (bool, error)

func (*Consumer) Stop 

func (c *Consumer) Stop()

func (*Consumer) Undispatch 

func (c *Consumer) Undispatch(ctx context.Context, stamps []Stamp) error

type Key 

type Key struct {
	ID    string
	Token string
}

type Lease 

type Lease struct {
	Partition int
	Epoch     int64
	Draining  bool
}

type OpenOption 

type OpenOption func(*openOptions)

OpenOption configures Open.

func WithMaxConns 

func WithMaxConns(n int) OpenOption

WithMaxConns caps the Store's open connections and keeps that many idle; n <= 0 means DefaultMaxConns.

func WithTracerProvider 

func WithTracerProvider(tp trace.TracerProvider) OpenOption

WithTracerProvider records a span per database statement.

type Request 

type Request struct {
	ID        string
	Token     string
	Queue     string
	Partition int
	Epoch     int64
	Attempt   int64
	Deadline  int64
	Envelope  string
	Payload   []byte
	Cancelled bool
	// contains filtered or unexported fields
}

func (Request) Key 

func (r Request) Key() Key

func (Request) Stamp 

func (r Request) Stamp() Stamp

type Result 

type Result struct {
	Seq     int64
	Route   string
	ID      string
	Token   string
	Payload string
}

type Stamp 

type Stamp struct {
	Key
	Attempt int64
}

Stamp names one dispatch of a request; every dispatch draws a fresh attempt.

type Store 

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

func Open 

func Open(ctx context.Context, dsn string, opts ...OpenOption) (*Store, error)

func (*Store) Ack 

func (s *Store) Ack(ctx context.Context, owner string, completions []Completion) ([]bool, error)

Ack deletes each request and records its result in one transaction. Fenced, missing, and duplicate completions report false.

func (*Store) AcquirePartitions 

func (s *Store) AcquirePartitions(ctx context.Context, queue, owner string, now time.Time, ttl time.Duration, limit int) ([]Lease, error)

AcquirePartitions takes unowned or lapsed partitions and bumps their epoch.

func (*Store) AcquireQuotaSlots 

func (s *Store) AcquireQuotaSlots(ctx context.Context, holder, key string, n, limit int) (int, error)

AcquireQuotaSlots takes up to n of limit concurrent slots for key on behalf of holder and returns how many it took.

func (*Store) AdmitQuota 

func (s *Store) AdmitQuota(ctx context.Context, key string, n, limit int, window time.Duration) (int, error)

AdmitQuota admits up to n requests for key while fewer than limit were admitted in the trailing window, and returns how many it admitted. Each window keeps its own log, so gates with different windows on one key count only their own admissions.

func (*Store) Backlog 

func (s *Store) Backlog(ctx context.Context, queue string, now int64, windows []int64) (depth int64, expiring []int64, err error)

func (*Store) Cancel 

func (s *Store) Cancel(ctx context.Context, ids []string) error

func (*Store) CancelledKeys 

func (s *Store) CancelledKeys(ctx context.Context, keys []Key) (map[Key]bool, error)

CancelledKeys returns the subset of keys marked cancelled. A key with no row is absent from the result: an unknown request is not cancelled.

func (*Store) Close 

func (s *Store) Close() error

func (*Store) Defer 

func (s *Store) Defer(ctx context.Context, owner string, stamps []Stamp, notBefore int64) error

Defer returns dispatched requests to pending, not dispatched again before notBefore.

func (*Store) DeleteQuotaHolder 

func (s *Store) DeleteQuotaHolder(ctx context.Context, holder string) error

DeleteQuotaHolder deletes a holder and its slots, freeing everything it still counts.

func (*Store) Dispatch 

func (s *Store) Dispatch(ctx context.Context, queue, owner string, now time.Time, limit int) ([]Request, error)

Dispatch stamps up to limit pending requests in partitions owner holds and is not draining.

func (*Store) Enqueue 

func (s *Store) Enqueue(ctx context.Context, reqs ...Request) error

Enqueue inserts all requests or none. Equal deadlines dispatch in submission order.

func (*Store) EnsurePartitions 

func (s *Store) EnsurePartitions(ctx context.Context, queue string) error

func (*Store) ExpireResults 

func (s *Store) ExpireResults(ctx context.Context, now int64, limit int) (int64, error)

ExpireResults deletes up to limit results on any route whose TTL ended before now and returns how many it deleted.

func (*Store) HasRequests 

func (s *Store) HasRequests(ctx context.Context, queue string) (bool, error)

func (*Store) Heartbeat 

func (s *Store) Heartbeat(ctx context.Context, queue, owner string, now time.Time, ttl time.Duration, member bool) ([]Lease, int, error)

Heartbeat renews owner's leases and reports live dispatchers. A non-member renews without counting toward the share, so peers take over its partitions while it drains.

func (*Store) InFlight 

func (s *Store) InFlight(ctx context.Context, queue, owner string) ([]Stamp, error)

func (*Store) Leave 

func (s *Store) Leave(ctx context.Context, queue, owner string) error

func (*Store) Migrate 

func (s *Store) Migrate(ctx context.Context) error

Migrate creates the schema in a database that has none and refuses one that another build created. It is safe to run concurrently; a current schema takes no DDL locks.

func (*Store) Now 

func (s *Store) Now(ctx context.Context) (time.Time, error)

Now reads the database clock so every dispatcher times leases alike.

func (*Store) Ping 

func (s *Store) Ping(ctx context.Context) error

func (*Store) PopResults 

func (s *Store) PopResults(ctx context.Context, route string, now int64, limit int) ([]Result, error)

PopResults deletes the results it returns.

func (*Store) ReapQuota 

func (s *Store) ReapQuota(ctx context.Context, limit int) (int, error)

ReapQuota deletes up to limit rows of each kind that no longer count: expired rate admissions, windows and slots that hold nothing, and keys no slot names. It returns how many windows and rows it changed.

func (*Store) ReapQuotaHolders 

func (s *Store) ReapQuotaHolders(ctx context.Context, grace time.Duration) error

ReapQuotaHolders deletes holders whose lease lapsed more than grace ago, along with their slots.

func (*Store) RegisterQuotaHolder 

func (s *Store) RegisterQuotaHolder(ctx context.Context, holder string, ttl time.Duration) error

RegisterQuotaHolder creates a holder whose concurrency slots count until its lease lapses.

func (*Store) ReleasePartitions 

func (s *Store) ReleasePartitions(ctx context.Context, queue, owner string, partitions []int) error

func (*Store) ReleaseQuotaSlots 

func (s *Store) ReleaseQuotaSlots(ctx context.Context, holder string, keys []string, counts []int) error

ReleaseQuotaSlots returns counts[i] of holder's slots for keys[i].

func (*Store) RenewQuotaHolder 

func (s *Store) RenewQuotaHolder(ctx context.Context, holder string, ttl time.Duration) (bool, error)

RenewQuotaHolder extends a live holder's lease. It returns false once the lease has lapsed, and a lapsed holder is never renewed.

func (*Store) ResetStale 

func (s *Store) ResetStale(ctx context.Context, queue, owner string, partitions []int) error

ResetStale returns requests stamped by an older epoch of owner's partitions to pending, limited to the given partitions unless nil. Rows another transaction holds are skipped and left for a later call.

func (*Store) ResultDepth 

func (s *Store) ResultDepth(ctx context.Context, route string) (int64, error)

func (*Store) Retry 

func (s *Store) Retry(ctx context.Context, owner string, stamp Stamp, notBefore int64, envelope string) (bool, error)

Retry parks a dispatched request until notBefore with an updated envelope; its payload is unchanged.

func (*Store) SetDraining 

func (s *Store) SetDraining(ctx context.Context, queue, owner string, partitions []int, draining bool) error

A draining partition stays leased for acks but dispatches nothing new.

func (*Store) TruncateForTest 

func (s *Store) TruncateForTest(ctx context.Context) error

func (*Store) Undispatch 

func (s *Store) Undispatch(ctx context.Context, owner string, stamps []Stamp) error

Undispatch, Defer and Retry only touch a request still carrying the given stamp in a partition owner holds.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.