Documentation
¶
Overview ¶
Package queue implements merge queue operations on top of the PostgreSQL store. All state lives in the database — there is no in-memory copy. Multi-step operations use transactions to prevent races.
Index ¶
- type DequeueResult
- type EnqueueResult
- type Service
- func (s *Service) Advance(ctx context.Context, repoID int64, targetBranch string) (*pg.QueueEntry, error)
- func (s *Service) CancelLiveBatches(ctx context.Context, repoID int64) error
- func (s *Service) ClearCheckStatuses(ctx context.Context, ids []int64) error
- func (s *Service) ClearMergeBranch(ctx context.Context, ids []int64) error
- func (s *Service) Dequeue(ctx context.Context, repoID, prNumber int64) (*DequeueResult, error)
- func (s *Service) DequeueAll(ctx context.Context, repoID int64) error
- func (s *Service) Enqueue(ctx context.Context, repoID, prNumber int64, prHeadSHA, targetBranch string) (*EnqueueResult, error)
- func (s *Service) FormBatch(ctx context.Context, repoID int64, targetBranch string, max int) (*pg.Batch, error)
- func (s *Service) GetBatch(ctx context.Context, id int64) (*pg.Batch, error)
- func (s *Service) GetCheckStatuses(ctx context.Context, entryID int64) ([]pg.CheckStatus, error)
- func (s *Service) GetEntriesByIDs(ctx context.Context, ids []int64) ([]pg.QueueEntry, error)
- func (s *Service) GetEntry(ctx context.Context, repoID, prNumber int64) (*pg.QueueEntry, error)
- func (s *Service) GetLiveBatch(ctx context.Context, repoID int64, targetBranch string) (*pg.Batch, error)
- func (s *Service) GetOrCreateRepo(ctx context.Context, forge, owner, name string) (pg.Repo, error)
- func (s *Service) Head(ctx context.Context, repoID int64, targetBranch string) (*pg.QueueEntry, error)
- func (s *Service) List(ctx context.Context, repoID int64, targetBranch string) ([]pg.QueueEntry, error)
- func (s *Service) ListActiveEntries(ctx context.Context, repoID int64) ([]pg.QueueEntry, error)
- func (s *Service) ListLiveBatches(ctx context.Context, repoID int64) ([]pg.Batch, error)
- func (s *Service) LoadActiveQueues(ctx context.Context) ([]pg.LoadActiveQueuesRow, error)
- func (s *Service) Position(ctx context.Context, repoID int64, targetBranch string, prNumber int64) (int64, error)
- func (s *Service) SaveBatch(ctx context.Context, b *pg.Batch, removeEntries ...int64) error
- func (s *Service) SaveCheckStatus(ctx context.Context, entryID int64, checkContext string, state pg.CheckState, ...) error
- func (s *Service) SetError(ctx context.Context, repoID, prNumber int64, msg string) error
- func (s *Service) SetMergeBranch(ctx context.Context, repoID, prNumber int64, branchName, branchSHA string) error
- func (s *Service) UpdateState(ctx context.Context, repoID, prNumber int64, state pg.EntryState) error
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type DequeueResult ¶
type DequeueResult struct {
WasHead bool // true if the removed entry was head-of-queue
Found bool // true if the PR was in the queue
Entry pg.QueueEntry
}
DequeueResult describes what happened when dequeueing a PR.
type EnqueueResult ¶
type EnqueueResult struct {
Position int64 // 1-based position in queue
IsNew bool
Entry pg.QueueEntry
}
EnqueueResult describes what happened when enqueueing a PR.
type Service ¶
type Service struct {
// contains filtered or unexported fields
}
Service provides merge queue operations backed by the database.
func NewService ¶
NewService creates a new queue service.
func (*Service) Advance ¶
func (s *Service) Advance(ctx context.Context, repoID int64, targetBranch string) (*pg.QueueEntry, error)
Advance removes the head-of-queue and returns the new head (or nil). Runs in a transaction so the delete and new-head lookup are atomic.
func (*Service) CancelLiveBatches ¶
CancelLiveBatches marks every live batch for a repo cancelled. Used on repo removal so the unique-live index does not block a future re-add.
func (*Service) ClearCheckStatuses ¶
ClearCheckStatuses deletes recorded check statuses for the given entry IDs. Called on each batch rebuild so stale results from a previous build cannot satisfy or fail the next one.
func (*Service) ClearMergeBranch ¶
ClearMergeBranch nulls merge_branch_{name,sha} for the given entry IDs.
func (*Service) Dequeue ¶
Dequeue removes a PR from the queue. Returns whether it was found and whether it was head-of-queue. Runs in a transaction so the head check and delete are atomic.
func (*Service) DequeueAll ¶
DequeueAll removes all queue entries for a repo. Used when a repo is removed from the registry to avoid leaving orphaned entries in the DB.
func (*Service) Enqueue ¶
func (s *Service) Enqueue(ctx context.Context, repoID, prNumber int64, prHeadSHA, targetBranch string) (*EnqueueResult, error)
Enqueue adds a PR to the tail of its repo+branch queue. If the PR is already queued, it is a no-op and returns the existing position. Runs in a transaction so the insert and position count are atomic.
func (*Service) FormBatch ¶
func (s *Service) FormBatch(ctx context.Context, repoID int64, targetBranch string, max int) (*pg.Batch, error)
FormBatch atomically takes up to max queued entries for (repo, branch), creates a batch row owning them, and marks each entry testing with active_batch_id set. Returns nil when there is nothing queued. max <= 0 means "everything currently queued".
func (*Service) GetCheckStatuses ¶
GetCheckStatuses returns all check statuses for a queue entry.
func (*Service) GetEntriesByIDs ¶
GetEntriesByIDs loads queue entries by primary key, preserving the input order so callers can rely on FIFO member ordering.
func (*Service) GetEntry ¶
GetEntry returns a queue entry by repo and PR number, or nil if not found.
func (*Service) GetLiveBatch ¶
func (s *Service) GetLiveBatch(ctx context.Context, repoID int64, targetBranch string) (*pg.Batch, error)
GetLiveBatch returns the live (forming/testing) batch for a queue, or nil.
func (*Service) GetOrCreateRepo ¶
GetOrCreateRepo ensures a repo row exists and returns it.
func (*Service) Head ¶
func (s *Service) Head(ctx context.Context, repoID int64, targetBranch string) (*pg.QueueEntry, error)
Head returns the head-of-queue entry for a (repo, branch), or nil if empty.
func (*Service) List ¶
func (s *Service) List(ctx context.Context, repoID int64, targetBranch string) ([]pg.QueueEntry, error)
List returns all entries in a (repo, branch) queue in FIFO order.
func (*Service) ListActiveEntries ¶
ListActiveEntries returns all non-terminal entries for a repo.
func (*Service) ListLiveBatches ¶
ListLiveBatches returns all live batches for a repo (across target branches).
func (*Service) LoadActiveQueues ¶
LoadActiveQueues returns all non-terminal entries across all repos for startup recovery.
func (*Service) Position ¶
func (s *Service) Position(ctx context.Context, repoID int64, targetBranch string, prNumber int64) (int64, error)
Position returns the PR's 1-based position among active entries for its target branch, or 0 when the PR is not queued.
func (*Service) SaveCheckStatus ¶
func (s *Service) SaveCheckStatus(ctx context.Context, entryID int64, checkContext string, state pg.CheckState, targetURL string) error
SaveCheckStatus records or updates a check status for an entry.
func (*Service) SetMergeBranch ¶
func (s *Service) SetMergeBranch(ctx context.Context, repoID, prNumber int64, branchName, branchSHA string) error
SetMergeBranch records the merge branch name and SHA for an entry.
func (*Service) UpdateState ¶
func (s *Service) UpdateState(ctx context.Context, repoID, prNumber int64, state pg.EntryState) error
UpdateState transitions a queue entry to a new state.
Source Files