Documentation
¶
Overview ¶
Package worker drives the Postgres-backed job queue introduced in S14. The pool dispatches one Job per goroutine, claiming rows via FOR UPDATE SKIP LOCKED so concurrent workers don't double-process.
Job kinds, their payload schema, and their handlers live alongside this package in sub-packages (jobs/<kind>.go). The pool itself is kind-agnostic: a Handler is just a func that takes a payload and returns nil-or-error.
Index ¶
- Constants
- Variables
- func Backoff(attempts int, jitter func() float64) time.Duration
- func Enqueue(ctx context.Context, db DBTX, kind Kind, payload any, opts EnqueueOptions) (int64, error)
- func Notify(ctx context.Context, db DBTX) error
- func PoisonError(cause error) error
- type DBTX
- type EnqueueOptions
- type Handler
- type Kind
- type Pool
- type PoolConfig
Constants ¶
const NotifyChannel = "shithub_jobs"
NotifyChannel is the Postgres LISTEN/NOTIFY channel the pool subscribes to so it wakes up immediately when a job is enqueued, instead of polling. Callers wrapping enqueue in a tx must NOTIFY inside the same tx so the notification only fires on commit.
Variables ¶
var ErrPoison = errors.New("worker: poison job")
ErrPoison wraps a handler error that should NOT be retried. The pool jumps the job straight to MarkJobFailed instead of rescheduling.
Functions ¶
func Backoff ¶
Backoff returns the delay before retrying a job that is about to be rescheduled. The formula is `30s * 2^attempts` capped at 1 hour, with ±20% jitter so a fleet doesn't synchronize retries on a sibling dependency outage.
`attempts` is the number of attempts already made (1-indexed: the just-failed attempt counts as 1).
func Enqueue ¶
func Enqueue(ctx context.Context, db DBTX, kind Kind, payload any, opts EnqueueOptions) (int64, error)
Enqueue inserts a job row and returns its id. Callers running inside a transaction should pass the tx as db so the enqueue is rolled back alongside any related state changes; same goes for the NOTIFY (issued separately by the caller via Notify).
func Notify ¶
Notify wakes any LISTENing workers. Safe to call after a successful commit; if called inside a tx, the NOTIFY only delivers when the tx commits (Postgres semantics). Errors are non-fatal — workers also poll at a slow interval as a backstop.
func PoisonError ¶
PoisonError wraps cause as a poison error. The cause is preserved in last_error for operator inspection.
Types ¶
type DBTX ¶
DBTX matches the pgx interface that sqlc-generated methods accept (anything providing Exec/Query/QueryRow). The pool, a tx, and the helpers in dbtest all satisfy it.
type EnqueueOptions ¶
type EnqueueOptions struct {
RunAt pgtype.Timestamptz
MaxAttempts int32
}
EnqueueOptions tunes a single enqueue. RunAt zero means "now"; pass a future time to delay the first run. MaxAttempts zero means use the table default (5).
type Handler ¶
type Handler func(ctx context.Context, payload json.RawMessage) error
Handler runs one job. The framework supplies the raw JSON payload; handlers Unmarshal into their own typed schema. A nil error reports success and the job is marked completed; any non-nil error triggers the backoff/retry path. ErrPoison is the explicit "do not retry" signal — useful when the input is malformed and retrying can't help.
type Kind ¶
type Kind string
Kind is the canonical name of a job. Use lowercase letters and colon-separated namespaces (e.g. "push:process", "repo:size_recalc"). Kind doubles as the dispatch index — workers query `WHERE kind = $1` so adding new kinds doesn't disturb existing ones.
const ( KindPushProcess Kind = "push:process" KindRepoSizeRecalc Kind = "repo:size_recalc" KindJobsPurge Kind = "jobs:purge_completed" )
Built-in kinds shipped in S14.
S22 pull-request kinds. Synchronize refreshes commits + files after a head-side push; mergeability runs the merge-tree probe.
(No async-merge kind: the POST .../merge handler executes the merge synchronously so the redirect lands on the merged state. If we add async merging later — for very large repos — re-introduce a KindPRMerge here and a matching jobs.PRMerge handler.)
const ( KindRepoIndexCode Kind = "repo:index_code" KindRepoIndexReconcile Kind = "repo:index_reconcile" )
S28 code-search kinds. index_code re-indexes a repo's default branch (paths + content). Enqueued by push:process when the default branch advances; also by index_reconcile when drift between default_branch_oid and last_indexed_oid is detected.
const (
KindLifecycleSweep Kind = "lifecycle:sweep"
)
S16 lifecycle housekeeping kind.
const (
KindNotifyFanout Kind = "notify:fanout"
)
S29 notification kinds. notify:fanout drains the domain_events table past the persisted cursor and materializes inbox rows + (optional) emails. Self-throttling per FanoutBatch; loop-friendly (the handler returns the count so the pool can re-enqueue when a drain didn't catch up).
const (
KindRepoForkClone Kind = "repo:fork_clone"
)
S27 fork kinds. fork_clone runs `git clone --bare --shared` for a freshly created fork shell; the fork's init_status flips from init_pending → initialized (or init_failed) on success/failure.
type Pool ¶
type Pool struct {
// contains filtered or unexported fields
}
Pool dispatches jobs from the queue. Construct via NewPool, register handlers via Register, run via Run.
func NewPool ¶
func NewPool(db *pgxpool.Pool, cfg PoolConfig) *Pool
NewPool wires a pool against an open pgx pool. Callers register handlers before calling Run.
type PoolConfig ¶
type PoolConfig struct {
Workers int // default 4
IdlePoll time.Duration // default 5s — backstop when LISTEN drops a wake
JobTimeout time.Duration // default 5min, applied per-job via context
InstanceID string // default "<hostname>:<pid>"
Logger *slog.Logger // default discards
}
PoolConfig configures Pool. Leave fields zero for the documented defaults.
Source Files
Directories