Documentation
¶
Overview ¶
Package queue runs the River job queue on the control-plane SQLite database, with the workers embedded in platformd (E1.3; golden rule 7: no Redis, no separate worker process).
River's SQLite driver runs its internal services (producers, completer, leader election, maintenance) concurrently, and SQLite serialises writers; the driver therefore recommends a single-connection pool so River never races itself into SQLITE_BUSY. The store's pool keeps eight connections for request handling, so the queue opens its own one-connection pool on the same file, with the same pragmas (WAL, foreign keys, busy timeout). Application transactions still come from the store's pool: EnqueueDeployTx / EnqueueTx insert through the caller's *sql.Tx, so a rolled-back transaction enqueues nothing.
Job args carry database ids only, never secret values (golden rule 3). Worker errors are logged by River and persisted in river_job.errors, so with WithRedactor every error a worker returns is redacted before River sees it.
Index ¶
- Constants
- func AddWorker[T river.JobArgs](w *river.Workers, worker river.Worker[T])
- func NewWorkers() *river.Workers
- type BackupArgs
- type Client
- func (c *Client) Close() error
- func (c *Client) Enqueue(ctx context.Context, args river.JobArgs, opts *river.InsertOpts) (int64, error)
- func (c *Client) EnqueueDeployTx(ctx context.Context, tx *sql.Tx, args DeployArgs) (int64, error)
- func (c *Client) EnqueueTx(ctx context.Context, tx *sql.Tx, args river.JobArgs, opts *river.InsertOpts) (int64, error)
- func (c *Client) Start(ctx context.Context) error
- func (c *Client) Stop(ctx context.Context) error
- type DeployArgs
- type Option
- type ScanArgs
Constants ¶
const ( QueueDeploy = "deploy" QueueDefault = river.QueueDefault )
Queue names. Deploy jobs run on their own queue so that build concurrency can be capped independently of everything else (plan §9: one build at a time on small hosts).
const ( KindDeploy = "deploy" KindBackup = "backup" KindScan = "scan" )
Job kinds. Kinds are persisted in the river_job table, so they are stable identifiers.
const ( DefaultDeployConcurrency = 1 DefaultDefaultConcurrency = 2 DefaultJobTimeout = 30 * time.Minute DefaultMaxAttempts = 3 // DefaultStopTimeout is how long Stop waits for in-flight jobs before cancelling them. DefaultStopTimeout = 30 * time.Second )
Defaults. Deploy concurrency is 1 (one build at a time on a small host, plan §9); the default queue takes two workers. A deploy (clone + build + start + health check) gets 30 minutes before its context is cancelled. Jobs are retried at most twice (3 attempts) — a build that fails deterministically should be surfaced, not hammered.
Variables ¶
This section is empty.
Functions ¶
func AddWorker ¶
AddWorker registers worker for the job kind of T. It panics if a worker for that kind is already registered, which is a programming error caught at startup.
func NewWorkers ¶
NewWorkers returns an empty worker bundle to register workers on before calling New.
Types ¶
type BackupArgs ¶
type BackupArgs struct {
BackupID string `json:"backup_id"`
}
BackupArgs takes one addon backup (Phase 2). The worker is registered later.
type Client ¶
type Client struct {
// contains filtered or unexported fields
}
Client is the embedded River client. It is safe for concurrent use.
func New ¶
func New(ctx context.Context, st *store.Store, workers *river.Workers, logger *slog.Logger, opts ...Option) (*Client, error)
New opens River's connection pool on the store's database file, applies River's migrations (idempotent), and builds the client with the "deploy" and "default" queues. Workers must hold every worker that will be registered for this process; jobs of unregistered kinds are refused at insert time. Start must be called before jobs are worked; Stop releases the pool.
func (*Client) Close ¶
Close releases River's connection pool without waiting for jobs. Use Stop for a graceful shutdown; Close exists for callers that never started the client. Idempotent.
func (*Client) Enqueue ¶
func (c *Client) Enqueue(ctx context.Context, args river.JobArgs, opts *river.InsertOpts) (int64, error)
Enqueue inserts a job outside any application transaction. Prefer EnqueueTx whenever the job belongs with other writes, so that they commit or roll back together. Validation is the same as EnqueueTx's.
func (*Client) EnqueueDeployTx ¶
EnqueueDeployTx inserts a deploy job inside the caller's transaction. The job becomes visible to workers only when tx commits; a rollback enqueues nothing. Returns the River job id.
func (*Client) EnqueueTx ¶
func (c *Client) EnqueueTx(ctx context.Context, tx *sql.Tx, args river.JobArgs, opts *river.InsertOpts) (int64, error)
EnqueueTx inserts any job inside the caller's transaction (see EnqueueDeployTx). opts may be nil. Args are validated (see validator) and opts are checked against the queue's invariants (see insertOptsFor): args that pin a queue (DeployArgs) cannot be moved to another one and MaxAttempts cannot exceed DefaultMaxAttempts.
func (*Client) Start ¶
Start begins fetching and working jobs. Cancelling ctx initiates the same graceful stop as Stop; callers should still call Stop (or Close) to release the pool.
func (*Client) Stop ¶
Stop stops fetching new jobs and waits for in-flight jobs to finish. Jobs still running after the stop timeout (default 30s) have their contexts cancelled and are waited for once more. If ctx ends first, River's services are still torn down on a short detached context (running jobs get their contexts cancelled) and Stop returns ctx's error. Stop is safe to call whether or not Start was called, and always releases River's connection pool before returning.
type DeployArgs ¶
type DeployArgs struct {
DeploymentID string `json:"deployment_id"`
}
DeployArgs runs the deployment state machine (internal/deploy) for one deployment.
func (DeployArgs) InsertOpts ¶
func (DeployArgs) InsertOpts() river.InsertOpts
InsertOpts pins deploy jobs to the serialised deploy queue. River itself lets a per-call *river.InsertOpts override these defaults, so EnqueueTx / Enqueue additionally refuse any caller options that name a different queue (see insertOptsFor); together they keep every deploy on QueueDeploy, whose worker count is the build-concurrency cap.
type Option ¶
type Option func(*options)
Option configures New.
func WithDeployConcurrency ¶
WithDeployConcurrency sets the number of deploy jobs that may run at once (default 1).
func WithRedactor ¶
WithRedactor redacts every error returned by a worker — the text River logs and persists in river_job.errors — with r before River sees it (golden rule 3). The redactor is consulted at work time, so values added to it after New are covered too. A nil redactor redacts nothing.
func WithStopTimeout ¶
WithStopTimeout sets how long Stop waits for running jobs before cancelling their contexts (default 30s).
Source Files