Documentation
¶
Overview ¶
Package prioritize implements the queue-wide reconcile stage that rations a shared build budget across every in-flight batch of a queue.
Unlike every other pipeline stage, prioritize is not batch-scoped: its queue message carries only a queue name (entity.QueueID), and each invocation re-evaluates every Speculating batch's speculation tree for that queue together. It gathers the queue-wide set of candidate paths (Selected, Prioritized, or Building), hands them to the queue's prioritizer.Prioritizer, and applies the returned decisions:
- Promote on a Selected path clears it to run (-> Prioritized).
- Cancel on a Building path asks the build runner to stop it and marks it Cancelling; the build stage's own signal loop confirms the stop.
- Cancel on a Prioritized path (no build yet) drops it straight to Cancelled.
Each affected tree is persisted under its own optimistic lock, so a version conflict only nacks and re-derives that tree's part of the round on redelivery — the whole computation is a pure function of freshly read state, so recomputing it is always safe. After applying decisions, the controller republishes to the build topic for every batch whose tree has at least one Prioritized path with no build yet, not just newly promoted ones — this heals a build message dropped by a prior crash and is itself idempotent, since the build stage dedups on batch ID.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Controller ¶
type Controller struct {
// contains filtered or unexported fields
}
Controller consumes queue-wide prioritize messages, ranks every candidate speculation path in the queue against its build budget, applies the resulting decisions to each affected speculation tree, and republishes to build for paths cleared to run.
func NewController ¶
func NewController( logger *zap.SugaredLogger, scope tally.Scope, store storage.Storage, prioritizers prioritizer.Factory, registry consumer.TopicRegistry, topicKey consumer.TopicKey, consumerGroup string, ) *Controller
NewController creates a new prioritize controller for the orchestrator.
func (*Controller) ConsumerGroup ¶
func (c *Controller) ConsumerGroup() string
ConsumerGroup returns the consumer group for offset tracking.
func (*Controller) Name ¶
func (c *Controller) Name() string
Name returns the controller name for logging and metrics.
func (*Controller) Process ¶
Process re-evaluates the build budget for one queue: it loads every Speculating batch's speculation tree, ranks the queue-wide candidate paths through the queue's prioritizer, applies the returned decisions, persists the affected trees, and republishes to build for any path now cleared to run. Returns nil to ack (success), or error to nack (retry) — the whole round is a pure function of freshly read state, so redelivery simply recomputes it.
func (*Controller) TopicKey ¶
func (c *Controller) TopicKey() consumer.TopicKey
TopicKey returns the topic key this controller subscribes to.
Source Files