queue

package
v0.5.2 Latest Latest
Warning

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

 Go to latest Published: Aug 7, 2026 License: MIT Imports: 19 Imported by: 0

Documentation

Overview

Package queue wires River with isolated interactive, event, bulk, and sweep work lanes (SYNC_ENGINE C-B3).

Index

Constants

View Source 
const (
	// QueueInteractive serves latency-sensitive user-requested refreshes.
	QueueInteractive = "interactive"
	// QueueEvent serves webhook-originated refreshes.
	QueueEvent = "event"
	// QueueBulk serves bounded branch-reconciliation pages without consuming
	// direct-event or C-R1 sweep workers.
	QueueBulk = "bulk"
	// QueueSweep serves bounded-staleness refreshes.
	QueueSweep = "sweep"
	// QueueReconcile serves sweep state-machine and gap-healing work.
	QueueReconcile = "reconcile"
	// QueueDrift serves sampled semantic-drift detection.
	QueueDrift = "drift"
	// QueuePruner serves retention deletion work.
	QueuePruner = "pruner"
)
View Source 
const (
	// KindRefreshPR refreshes one pull request.
	KindRefreshPR = "refresh_pr"
	// KindRefreshRepository refreshes one repository.
	KindRefreshRepository = "refresh_repository"
	// KindRefreshRepoRules refreshes repository rules.
	KindRefreshRepoRules = "refresh_repo_rules"
	// KindRefreshStack refreshes one stack.
	KindRefreshStack = "refresh_stack"
	// KindRefreshChecks refreshes check runs for one head SHA.
	KindRefreshChecks = "refresh_checks"
	// KindRefreshBranch refreshes branch-dependent entities.
	KindRefreshBranch = "refresh_branch"
	// KindReconcileBranchPage authoritatively refreshes one bounded subset of
	// branch-dependent entities on the background lane.
	KindReconcileBranchPage = "reconcile_branch_page"
	// KindResolveStackMembership resolves a pull request's stack ownership.
	KindResolveStackMembership = "resolve_stack_membership"
	// KindBackfillRepoPage continues one repository backfill.
	KindBackfillRepoPage = "backfill_repo_page"
	// KindBackfillInstallation continues installation repository discovery.
	KindBackfillInstallation = "backfill_installation_page"
)

Variables

View Source 
var ErrRefreshGenerationContention = errors.New("refresh generation contention")

ErrRefreshGenerationContention reports that a nonblocking generation-lock attempt must be rolled back before it can be retried.

Functions

func InsertRefreshesTx 

func InsertRefreshesTx(
	ctx context.Context,
	tx pgx.Tx,
	client *river.Client[pgx.Tx],
	specs []RefreshSpec,
	queueName string,
) error

InsertRefreshesTx atomically advances durable generations and inserts follow-up pointers. Stack diffs, branch fan-out, and backfill all use this path so running-state coalescing keeps the same meaning everywhere.

func LockRefreshIntentGenerationsTx added in v0.4.1 

func LockRefreshIntentGenerationsTx(
	ctx context.Context,
	tx pgx.Tx,
	keys []RefreshGenerationKey,
) error

LockRefreshIntentGenerationsTx waits for a batch without retaining locks from a failed attempt. Each try runs inside a savepoint, whose rollback releases any advisory and generation-row locks acquired before contention. Successful locks remain owned by the caller's transaction after the savepoint is released.

func NewBackfillInsertOpts 

func NewBackfillInsertOpts() *river.InsertOpts

NewBackfillInsertOpts returns interactive-queue uniqueness for backfills.

func NewBackfillInsertOptsForQueue 

func NewBackfillInsertOptsForQueue(queueName string) *river.InsertOpts

NewBackfillInsertOptsForQueue returns backfill uniqueness for queueName.

func NewClient 

func NewClient(
	pool *pgxpool.Pool,
	options ...ClientOption,
) (*river.Client[pgx.Tx], error)

NewClient builds a River client for the selected component queues and worker registrars. Without WithQueues it owns the four fetch queues: interactive, event, bulk, and sweep.

func NewRefreshInsertOpts 

func NewRefreshInsertOpts(scheduledAt time.Time) *river.InsertOpts

NewRefreshInsertOpts is the one supported uniqueness definition for refresh work. River requires running in this mask; a durable generation handles signals that coalesce while a job runs.

func NewRefreshInsertOptsForQueue 

func NewRefreshInsertOptsForQueue(
	queueName string,
	scheduledAt time.Time,
) *river.InsertOpts

NewRefreshInsertOptsForQueue returns refresh uniqueness for queueName.

func TryLockRefreshIntentGenerationsTx added in v0.4.1 

func TryLockRefreshIntentGenerationsTx(
	ctx context.Context,
	tx pgx.Tx,
	keys []RefreshGenerationKey,
) error

TryLockRefreshIntentGenerationsTx gates a batch before the generation upsert. Advisory and existing-row locks are tried in deterministic order. On contention, callers must roll back the transaction or containing savepoint so every lock acquired by the attempt is released before retrying.

Types

type BackfillInstallationPageArgs 

type BackfillInstallationPageArgs struct {
	InstallationID int64  `json:"installation_id"`
	Phase          string `json:"phase"`
	Page           int    `json:"page"`
}

BackfillInstallationPageArgs identifies one repository-discovery page.

func NewBackfillInstallationPageArgs 

func NewBackfillInstallationPageArgs(
	installationID int64,
	phase string,
	page int,
) BackfillInstallationPageArgs

NewBackfillInstallationPageArgs constructs a discovery backfill pointer.

func (BackfillInstallationPageArgs) Kind 

func (BackfillInstallationPageArgs) Kind() string

Kind returns the River job kind.

type BackfillRepoPageArgs 

type BackfillRepoPageArgs struct {
	InstallationID int64  `json:"installation_id"`
	RepoFullName   string `json:"repo"`
	Phase          string `json:"phase"`
	Page           int    `json:"page"`
}

BackfillRepoPageArgs identifies one resumable repository backfill page.

func NewBackfillRepoPageArgs 

func NewBackfillRepoPageArgs(
	installationID int64,
	repoFullName string,
	phase string,
	page int,
) BackfillRepoPageArgs

NewBackfillRepoPageArgs constructs a resumable repository backfill pointer.

func (BackfillRepoPageArgs) Kind 

func (BackfillRepoPageArgs) Kind() string

Kind returns the River job kind.

type BranchReconcilePageArgs added in v0.5.0 

type BranchReconcilePageArgs struct {
	RepoID       int64                   `json:"repo_id"`
	RepoFullName string                  `json:"repo"`
	Branch       string                  `json:"branch"`
	Generation   int64                   `json:"generation"`
	Page         int                     `json:"page"`
	Targets      []BranchReconcileTarget `json:"targets"`
}

BranchReconcilePageArgs identifies one bounded failure/retry unit. Targets contain pointers and generation fences only, never entity data.

func NewBranchReconcilePageArgs added in v0.5.0 

func NewBranchReconcilePageArgs(
	repoID int64,
	repoFullName string,
	branch string,
	generation int64,
	page int,
	targets []BranchReconcileTarget,
) BranchReconcilePageArgs

NewBranchReconcilePageArgs constructs one bounded branch page pointer.

func (BranchReconcilePageArgs) Kind added in v0.5.0 

func (BranchReconcilePageArgs) Kind() string

Kind returns the River job kind.

type BranchReconcileTarget added in v0.5.0 

type BranchReconcileTarget struct {
	Kind       string `json:"kind"`
	Key        string `json:"key"`
	EntityKey  string `json:"entity_key"`
	Generation int64  `json:"generation"`
}

BranchReconcileTarget is an entity pointer plus the direct-refresh generation observed by the atomic branch bulk transaction.

type ClientOption 

type ClientOption func(*clientOptions)

ClientOption customizes River workers, queues, schedules, and observers.

func WithDeadlineObserver 

func WithDeadlineObserver(observer DeadlineObserver) ClientOption

WithDeadlineObserver installs refresh-deadline instrumentation.

func WithNow 

func WithNow(now func() time.Time) ClientOption

WithNow supplies worker time for deterministic testing.

func WithPeriodicJobs 

func WithPeriodicJobs(jobs ...*river.PeriodicJob) ClientOption

WithPeriodicJobs installs leader-elected River schedules supplied by the enabled serve roles.

func WithPlugins 

func WithPlugins(plugins ...rivertype.Plugin) ClientOption

WithPlugins installs River plugins on inserts, workers, and internal hooks.

func WithQueueMaxWorkers 

func WithQueueMaxWorkers(queueName string, maxWorkers int) ClientOption

WithQueueMaxWorkers is primarily a deterministic test seam. Production keeps the reserved defaults below.

func WithQueues 

func WithQueues(names ...string) ClientOption

WithQueues limits a started client to the component queues owned by its enabled roles. Producer-only clients may pass no names and are never started, while combined roles pass the union of their queue families.

func WithRefreshHandler 

func WithRefreshHandler(handler RefreshHandler) ClientOption

WithRefreshHandler installs the authoritative refresh implementation.

func WithRefreshObserver 

func WithRefreshObserver(observer RefreshObserver) ClientOption

WithRefreshObserver installs authoritative-fetch instrumentation.

func WithWorkerRegistrar 

func WithWorkerRegistrar(registrar func(*river.Workers)) ClientOption

WithWorkerRegistrar lets milestone packages own their typed River args and workers without making queue import sweep/drift and creating a package cycle.

func WithoutRefreshWorkers 

func WithoutRefreshWorkers() ClientOption

WithoutRefreshWorkers is for a worker-only role such as the retention pruner. Producer clients keep the default registrations so River can validate refresh args at insertion time.

type DeadlineObserver 

type DeadlineObserver interface {
	RefreshDeadlineMissed(
		context.Context,
		string,
		string,
		time.Time,
		time.Time,
	)
}

DeadlineObserver receives completed refreshes that missed reconciliation deadlines.

type DeadlineObservers 

type DeadlineObservers []DeadlineObserver

DeadlineObservers fans deadline misses out in declaration order.

func (DeadlineObservers) RefreshDeadlineMissed 

func (observers DeadlineObservers) RefreshDeadlineMissed(
	ctx context.Context,
	kind string,
	key string,
	deadline time.Time,
	completedAt time.Time,
)

type LogDeadlineObserver 

type LogDeadlineObserver struct{}

LogDeadlineObserver logs reconciliation deadline misses.

func (LogDeadlineObserver) RefreshDeadlineMissed 

func (LogDeadlineObserver) RefreshDeadlineMissed(
	_ context.Context,
	kind string,
	key string,
	deadline time.Time,
	completedAt time.Time,
)

type OrphanReconcileResult added in v0.5.2 

type OrphanReconcileResult struct {
	Scanned int
	Delayed int
	Orphans []OrphanedRefreshPointer
}

OrphanReconcileResult reports one reconciliation pass. Delayed counts outstanding generations whose unique insert deduplicated against a live pointer; they are merely late, not orphaned.

func ReconcileOrphanedRefreshPointers added in v0.5.2 

func ReconcileOrphanedRefreshPointers(
	ctx context.Context,
	pool *pgxpool.Pool,
	client *river.Client[pgx.Tx],
	staleBefore time.Time,
	batchLimit int32,
	queueName string,
) (OrphanReconcileResult, error)

ReconcileOrphanedRefreshPointers restores the invariant that every outstanding refresh generation has live work capable of completing it. A pointer job that River moved to a terminal state (for example discarded after exhausting retries) leaves its generation permanently outstanding because generation bumps only insert jobs and completion only runs inside a finishing worker.

For each stale outstanding generation the reconciler re-executes the producer's idempotent unique insert for the current generation, under the same single-key advisory lock the completion path takes. A live pointer in any queue deduplicates the insert, so a merely delayed generation is left untouched, concurrent reconcilers cannot create duplicate active work, and newer producer signals keep coalescing onto the replacement exactly as they would onto an original pointer. The generation itself is never completed here: only a successful refresh may advance completed_generation.

type OrphanedRefreshPointer added in v0.5.2 

type OrphanedRefreshPointer struct {
	Kind       string
	Key        string
	Generation int64
	// TerminalState is the most recent terminal River job state recorded for
	// the refresh identity, or "unknown" when River already pruned it.
	TerminalState string
}

OrphanedRefreshPointer is one outstanding generation that had no live pointer job left and was re-enqueued by reconciliation (issue #61).

type RefreshArgs 

type RefreshArgs struct {
	PointerKind              string `json:"kind"`
	Key                      string `json:"key"`
	ObserveDefaultBranchHead bool   `json:"observe_default_branch_head,omitempty"`
}

RefreshArgs is the complete durable job pointer. It intentionally contains no webhook or entity payload (SYNC_ENGINE §8 and C-I4).

type RefreshBranchArgs 

type RefreshBranchArgs struct{ RefreshArgs }

RefreshBranchArgs points at one repository branch.

func NewRefreshBranchArgs 

func NewRefreshBranchArgs(key string) RefreshBranchArgs

NewRefreshBranchArgs constructs a branch refresh pointer.

func (RefreshBranchArgs) Kind 

func (RefreshBranchArgs) Kind() string

Kind returns the River job kind.

type RefreshChecksArgs 

type RefreshChecksArgs struct{ RefreshArgs }

RefreshChecksArgs points at one repository head SHA.

func NewRefreshChecksArgs 

func NewRefreshChecksArgs(key string) RefreshChecksArgs

NewRefreshChecksArgs constructs a checks refresh pointer.

func (RefreshChecksArgs) Kind 

func (RefreshChecksArgs) Kind() string

Kind returns the River job kind.

type RefreshGeneration 

type RefreshGeneration struct {
	Spec       RefreshSpec
	Generation int64
}

RefreshGeneration pairs a deduplicated refresh spec with its durable generation after insertion.

func InsertRefreshesTxReturning 

func InsertRefreshesTxReturning(
	ctx context.Context,
	tx pgx.Tx,
	client *river.Client[pgx.Tx],
	specs []RefreshSpec,
	queueName string,
) ([]RefreshGeneration, error)

InsertRefreshesTxReturning performs InsertRefreshesTx and returns each deduplicated spec's new durable generation.

type RefreshGenerationKey added in v0.4.1 

type RefreshGenerationKey struct {
	Kind string `json:"kind"`
	Key  string `json:"refresh_key"`
}

RefreshGenerationKey identifies one durable generation row.

type RefreshHandler 

type RefreshHandler interface {
	RefreshPR(context.Context, RefreshRequest) error
	RefreshRepository(context.Context, RefreshRequest) error
	RefreshRepoRules(context.Context, RefreshRequest) error
	RefreshStack(context.Context, RefreshRequest) error
	RefreshChecks(context.Context, RefreshRequest) error
	RefreshBranch(context.Context, RefreshRequest) error
	ReconcileBranchPage(context.Context, *BranchReconcilePageArgs) error
	ResolveStackMembership(context.Context, RefreshRequest) error
	BackfillRepoPage(context.Context, BackfillRepoPageArgs) error
	BackfillInstallationPage(
		context.Context,
		BackfillInstallationPageArgs,
	) error
}

RefreshHandler is implemented by internal/fetch. Keeping this interface in queue avoids coupling durable job definitions to GitHub/cache internals.

type RefreshObservation 

type RefreshObservation struct {
	Kind             string
	Queue            string
	EventReceivedAt  time.Time
	CacheCommittedAt time.Time
	StartedAt        time.Time
	CompletedAt      time.Time
	Superseded       bool
	Err              error
}

RefreshObservation describes one authoritative fetch attempt. C-Q2's event-to-cache latency is populated only for event-originated generations.

type RefreshObserver 

type RefreshObserver interface {
	RefreshFinished(context.Context, *RefreshObservation)
}

RefreshObserver receives completed authoritative-fetch observations.

type RefreshPRArgs 

type RefreshPRArgs struct{ RefreshArgs }

RefreshPRArgs points at one pull request.

func NewRefreshPRArgs 

func NewRefreshPRArgs(key string) RefreshPRArgs

NewRefreshPRArgs constructs a pull-request refresh pointer.

func (RefreshPRArgs) Kind 

func (RefreshPRArgs) Kind() string

Kind returns the River job kind.

type RefreshRepoRulesArgs 

type RefreshRepoRulesArgs struct{ RefreshArgs }

RefreshRepoRulesArgs points at one repository ruleset.

func NewRefreshRepoRulesArgs 

func NewRefreshRepoRulesArgs(key string) RefreshRepoRulesArgs

NewRefreshRepoRulesArgs constructs a repository-rules refresh pointer.

func (RefreshRepoRulesArgs) Kind 

func (RefreshRepoRulesArgs) Kind() string

Kind returns the River job kind.

type RefreshRepositoryArgs 

type RefreshRepositoryArgs struct{ RefreshArgs }

RefreshRepositoryArgs points at one repository.

func NewRefreshRepositoryArgs 

func NewRefreshRepositoryArgs(key string) RefreshRepositoryArgs

NewRefreshRepositoryArgs constructs a repository refresh pointer.

func NewRefreshRepositoryHeadArgs added in v0.5.0 

func NewRefreshRepositoryHeadArgs(key string) RefreshRepositoryArgs

NewRefreshRepositoryHeadArgs constructs the constant-size repository observation used to close a default-branch push's partial-order gap.

func (RefreshRepositoryArgs) Kind 

func (RefreshRepositoryArgs) Kind() string

Kind returns the River job kind.

type RefreshRequest 

type RefreshRequest struct {
	Args  RefreshArgs
	Queue string
}

RefreshRequest gives a handler its durable pointer and owning queue.

type RefreshSpec 

type RefreshSpec struct {
	Kind            string
	Key             string
	ScheduledAt     time.Time
	Deadline        time.Time
	EventReceivedAt time.Time
}

RefreshSpec describes one desired refresh generation and optional timing metadata.

type RefreshStackArgs 

type RefreshStackArgs struct{ RefreshArgs }

RefreshStackArgs points at one stack.

func NewRefreshStackArgs 

func NewRefreshStackArgs(key string) RefreshStackArgs

NewRefreshStackArgs constructs a stack refresh pointer.

func (RefreshStackArgs) Kind 

func (RefreshStackArgs) Kind() string

Kind returns the River job kind.

type ResolveStackMembershipArgs 

type ResolveStackMembershipArgs struct{ RefreshArgs }

ResolveStackMembershipArgs points at a pull request requiring resolution.

func NewResolveStackMembershipArgs 

func NewResolveStackMembershipArgs(key string) ResolveStackMembershipArgs

NewResolveStackMembershipArgs constructs a membership-resolution pointer.

func (ResolveStackMembershipArgs) Kind 

func (ResolveStackMembershipArgs) Kind() string

Kind returns the River job kind.

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.