1212// See the License for the specific language governing permissions and
1313// limitations under the License.
1414
15- // Package buildsignal implements the build poll loop. Each message carries
16- // a Build; the controller calls BuildRunner.Status, writes the latest
17- // status to the BuildStore, publishes the batch ID to TopicKeySpeculate
18- // so the state machine re-evaluates, and holds the delivery for the next
19- // poll when the build has not yet reached a terminal state. Each message
20- // partitions by batch ID, so slow polls on one batch's build do not block
21- // others. A webhook-capable backend can publish into this same topic — the
22- // controller cannot tell a poll-driven message from a push.
15+ // Package buildsignal implements the build poll loop. Each message names one
16+ // build by the runner's own ID; the controller calls BuildRunner.Status, writes
17+ // the latest status to that build's record, and wakes the speculate run for its
18+ // batch, holding its delivery between polls while the build is still in
19+ // flight.
20+ //
21+ // The poll loop is also where builds are stopped. It follows every build to a
22+ // terminal state anyway, so on each poll it checks whether anything still
23+ // wants the build running — the batch not halted, the path's current attempt,
24+ // with a status that is not a stop, linked to this very build — and asks the
25+ // runner to cancel when nothing does. That makes cancellation level-triggered:
26+ // the speculate run records intent in the path set and nothing more, no cancel
27+ // message exists to go stale, and a check that misses one poll is remade on
28+ // the next.
29+ //
30+ // The path set is read here as that kill list, and never written — the
31+ // speculate run stays its only writer, which is what lets a run hold one
32+ // version of a head's paths across its whole decision without a poll
33+ // invalidating it. The read must come from the primary: a stale replica read
34+ // could report a wanted build unwanted, and a cancel is irreversible.
35+ //
36+ // Each build partitions independently, so slow polls on one build do not block
37+ // another, and successive polls of one build stay ordered. A webhook-capable
38+ // backend can publish into this same topic — the controller cannot tell a
39+ // poll-driven message from a push.
2340package buildsignal
2441
2542import (
2643 "context"
44+ "errors"
2745 "fmt"
2846
2947 "github.com/uber-go/tally"
30- entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
3148 "github.com/uber/submitqueue/platform/consumer"
3249 "github.com/uber/submitqueue/platform/metrics"
50+ "github.com/uber/submitqueue/submitqueue/core/publish"
3351 "github.com/uber/submitqueue/submitqueue/core/topickey"
3452 "github.com/uber/submitqueue/submitqueue/entity"
3553 "github.com/uber/submitqueue/submitqueue/extension/buildrunner"
5270 PollDelayRunningMs int64 = 2000
5371)
5472
73+ // opName is the metric operation name shared by every emit in this file.
74+ const opName = "process"
75+
5576// Controller consumes build signal messages, polls BuildRunner.Status,
5677// persists the result, and drives the polling loop.
5778type Controller struct {
@@ -88,20 +109,30 @@ func NewController(
88109 }
89110}
90111
91- // Process polls the build 's current status, persists it, publishes the
92- // batch ID to speculate so the state machine re-evaluates, and holds the
93- // delivery for the next poll when the build is still in flight.
112+ // Process polls one attempt 's build status, stops the build if nothing wants
113+ // it running any more, persists the status, wakes the speculate run, and
114+ // holds the delivery for the next poll while the build is still in flight.
94115// Returns nil to ack (success), or error to nack/reject.
95116//
96- // Error classification: deserialize, Status, Update, and the speculate
97- // publish stay non-retryable — they reject straight to DLQ on the first
98- // failure, where the operational republish path is the recovery mechanism.
99- // The poll loop's continuation is a hold, not a publish: the framework
100- // postpones the delivery, and a failed postpone write lapses into a normal
101- // visibility-timeout redelivery, so the loop cannot stall on an enqueue.
117+ // There is deliberately no short-circuit for halted batches. A cancelling batch
118+ // reaches its terminal state only once its paths stop, and this loop is the
119+ // only thing watching them stop — and, now, the thing stopping them: speculate
120+ // marks a path cancelling, the next poll here asks the runner to cancel, and a
121+ // later poll observes CI actually stop and records it. Skipping the work —
122+ // including the hold — for a halted batch would leave every cancelled
123+ // batch stranded in Cancelling forever.
124+ //
125+ // Error classification: deserialize, Status, the kill-list reads, the
126+ // persistence writes, and the speculate publish stay non-retryable — they
127+ // reject straight to DLQ on the first failure, where the operational republish
128+ // path is the recovery mechanism. The Cancel call is best-effort instead:
129+ // failing the message for it would kill the poll chain that is the only thing
130+ // that will retry the cancel, so a failure is logged and the next poll remakes
131+ // the whole check. The poll loop's continuation is a hold, not a publish: the
132+ // framework postpones the delivery, and a failed postpone write lapses into a
133+ // normal visibility-timeout redelivery, so the loop cannot stall on an
134+ // enqueue.
102135func (c * Controller ) Process (ctx context.Context , delivery consumer.Delivery ) error {
103- const opName = "process"
104-
105136 msg := delivery .Message ()
106137
107138 buildID , err := entity .BuildIDFromBytes (msg .Payload )
@@ -111,30 +142,27 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
111142 return fmt .Errorf ("failed to deserialize build ID: %w" , err )
112143 }
113144
114- // Only the build ID travels on the queue; load the full Build from
115- // storage, which is the single source of truth for its BatchID and the
116- // snapshot the poll loop updates.
145+ // Only the build ID travels on the queue; the record is the source of truth
146+ // for which batch this build belongs to and what it last reported.
117147 build , err := c .store .GetBuildStore ().Get (ctx , buildID .ID )
118148 if err != nil {
119149 metrics .NamedCounter (c .metricsScope , opName , "storage_errors" , 1 )
120150 return fmt .Errorf ("failed to get build %s: %w" , buildID .ID , err )
121151 }
122152
123- c .logger .Debugw ("polling build status" ,
124- "build_id" , build .ID ,
125- "batch_id" , build .BatchID ,
126- "attempt" , delivery .Attempt (),
127- "partition_key" , msg .PartitionKey ,
128- )
129-
130- // Load the batch first: it gives us the queue (needed to build the right
131- // BuildRunner) and lets us short-circuit halted batches before polling.
132153 batch , err := c .store .GetBatchStore ().Get (ctx , build .BatchID )
133154 if err != nil {
134155 metrics .NamedCounter (c .metricsScope , opName , "storage_errors" , 1 )
135156 return fmt .Errorf ("failed to get batch %s: %w" , build .BatchID , err )
136157 }
137158
159+ c .logger .Debugw ("polling build status" ,
160+ "build_id" , build .ID ,
161+ "batch_id" , build .BatchID ,
162+ "delivery_attempt" , delivery .Attempt (),
163+ "partition_key" , msg .PartitionKey ,
164+ )
165+
138166 buildRunner , err := c .buildRunners .For (buildrunner.Config {QueueName : batch .Queue })
139167 if err != nil {
140168 metrics .NamedCounter (c .metricsScope , opName , "status_errors" , 1 )
@@ -147,39 +175,56 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
147175 return fmt .Errorf ("failed to get status for build %s: %w" , buildID .ID , err )
148176 }
149177
150- // Short-circuit if the batch is already halted (terminal OR cancelling).
151- // Speculate is already idempotent on terminal, but skipping the publish
152- // avoids noise. For Cancelling batches the cancel controller owns the
153- // terminal write and the downstream fan-out, so further pipeline work
154- // would race against it; silent ack is the only safe action.
155- if entity .IsBatchStateHalted (batch .State ) {
156- metrics .NamedCounter (c .metricsScope , opName , "skipped_halted" , 1 )
157- c .logger .Infow ("skipping buildsignal publish for halted batch" ,
158- "batch_id" , batch .ID ,
159- "state" , string (batch .State ),
160- )
161- return nil
178+ // Reconcile before recording: a build still running that nothing wants any
179+ // more is asked to stop. Best-effort by design — the reschedule below is
180+ // what guarantees the request is remade, so a failed Cancel must not fail
181+ // the message and take that reschedule with it.
182+ if ! status .IsTerminal () {
183+ stop , err := c .unwanted (ctx , batch , build )
184+ if err != nil {
185+ return err
186+ }
187+ if stop {
188+ if err := buildRunner .Cancel (ctx , buildID ); err != nil {
189+ metrics .NamedCounter (c .metricsScope , opName , "cancel_errors" , 1 )
190+ c .logger .Warnw ("failed to cancel an unwanted build; the next poll retries" ,
191+ "build_id" , build .ID ,
192+ "batch_id" , build .BatchID ,
193+ "error" , err ,
194+ )
195+ } else {
196+ metrics .NamedCounter (c .metricsScope , opName , "build_cancelled" , 1 )
197+ c .logger .Infow ("requested cancellation of a build nothing wants running" ,
198+ "build_id" , build .ID ,
199+ "batch_id" , build .BatchID ,
200+ "path_id" , build .PathID ,
201+ "attempt" , build .Attempt ,
202+ )
203+ }
204+ }
162205 }
163206
164- updatedBuild : =build
165- updatedBuild .Status = status
166-
167- if err := c . store . GetBuildStore (). Update ( ctx , updatedBuild ); err != nil {
168- metrics . NamedCounter ( c . metricsScope , opName , "storage_errors" , 1 )
169- return fmt . Errorf ( "failed to update status for build %s: %w" , build . ID , err )
207+ if status ! =build . Status {
208+ build .Status = status
209+ if err := c . store . GetBuildStore (). Update ( ctx , build ); err != nil {
210+ metrics . NamedCounter ( c . metricsScope , opName , "storage_errors" , 1 )
211+ return fmt . Errorf ( "failed to update status for build %s: %w" , build . ID , err )
212+ }
170213 }
171214
172- // Re-evaluate the batch state machine with the latest build status.
173- if err := c .publishBatchID (ctx , topickey .TopicKeySpeculate , updatedBuild .BatchID , msg .PartitionKey ); err != nil {
215+ // Wake the speculate run so it re-plans the queue with this result. It
216+ // reads the status from the record above rather than being told it, so a
217+ // duplicated or reordered signal costs nothing.
218+ if err := c .publishBatchID (ctx , topickey .TopicKeySpeculate , batch .ID , batch .Queue ); err != nil {
174219 metrics .NamedCounter (c .metricsScope , opName , "publish_errors" , 1 )
175220 return fmt .Errorf ("failed to publish to speculate: %w" , err )
176221 }
177222
178223 if status .IsTerminal () {
179224 metrics .NamedCounter (c .metricsScope , opName , "terminal" , 1 , metrics .NewTag ("status" , string (status )))
180225 c .logger .Infow ("build reached terminal status" ,
181- "build_id" , updatedBuild .ID ,
182- "batch_id" , updatedBuild .BatchID ,
226+ "build_id" , build .ID ,
227+ "batch_id" , build .BatchID ,
183228 "status" , string (status ),
184229 )
185230 return nil
@@ -192,7 +237,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
192237 delivery .Hold (delayMs )
193238
194239 c .logger .Debugw ("holding for next build status poll" ,
195- "build_id" , updatedBuild .ID ,
240+ "build_id" , build .ID ,
196241 "status" , string (status ),
197242 "delay_ms" , delayMs ,
198243 )
@@ -210,31 +255,105 @@ func pollDelay(status entity.BuildStatus) int64 {
210255 }
211256}
212257
213- // publishBatchID publishes a batch ID to the topic identified by key.
214- func (c * Controller ) publishBatchID (ctx context.Context , key consumer.TopicKey , batchID string , partitionKey string ) error {
215- bid := entity.BatchID {ID : batchID }
216- payload , err := bid .ToBytes ()
258+ // unwanted reports whether nothing wants this build running any more: its
259+ // batch has halted, its path was called off or has moved to another attempt,
260+ // or the attempt's link names a different build (this one lost a dispatch
261+ // race). Every one of those conditions is permanent once true, so a stale read
262+ // can only err toward keeping a build — never toward cancelling a wanted one.
263+ //
264+ // The two anomaly cases run the other way on purpose. A missing set or entry
265+ // cannot legitimately happen — the dispatch read the entry out of the set to
266+ // start this build, and entries are not removed — and a missing link cannot
267+ // either, because the signal that led here is published after the link. Both
268+ // therefore indicate store corruption, and since a cancel is irreversible, a
269+ // corrupt kill list keeps the build rather than killing it; a halted batch is
270+ // still caught by the first check, which needs none of those records.
271+ func (c * Controller ) unwanted (ctx context.Context , batch entity.Batch , build entity.Build ) (bool , error ) {
272+ if entity .IsBatchStateHalted (batch .State ) {
273+ return true , nil
274+ }
275+
276+ // A build without path coordinates predates per-path dispatch; the batch
277+ // state above is the only kill list it has.
278+ if build .PathID == "" {
279+ return false , nil
280+ }
281+
282+ set , err := c .store .GetSpeculationPathSetStore ().Get (ctx , batch .ID )
217283 if err != nil {
218- return fmt .Errorf ("failed to serialize batch ID: %w" , err )
284+ if errors .Is (err , storage .ErrNotFound ) {
285+ metrics .NamedCounter (c .metricsScope , opName , "kill_list_anomalies" , 1 )
286+ c .logger .Warnw ("build exists but its head has no path set; keeping the build" ,
287+ "build_id" , build .ID ,
288+ "batch_id" , batch .ID ,
289+ )
290+ return false , nil
291+ }
292+ metrics .NamedCounter (c .metricsScope , opName , "storage_errors" , 1 )
293+ return false , fmt .Errorf ("failed to get path set for batch %s: %w" , batch .ID , err )
294+ }
295+
296+ entry , found := findEntry (set , build .PathID )
297+ if ! found {
298+ metrics .NamedCounter (c .metricsScope , opName , "kill_list_anomalies" , 1 )
299+ c .logger .Warnw ("build exists but its path is gone from the set; keeping the build" ,
300+ "build_id" , build .ID ,
301+ "batch_id" , batch .ID ,
302+ "path_id" , build .PathID ,
303+ )
304+ return false , nil
219305 }
220306
221- msg := entityqueue .NewMessage (batchID , payload , partitionKey , nil )
307+ switch entry .Status {
308+ case entity .SpeculationPathStatusCancelling , entity .SpeculationPathStatusCancelled :
309+ return true , nil
310+ }
222311
223- q , ok := c .registry .Queue (key )
224- if ! ok {
225- return fmt .Errorf ("no queue registered for topic key %s" , key )
312+ // The path has moved on to a newer attempt; this build belongs to a
313+ // superseded one.
314+ if entry .Attempt != build .Attempt {
315+ return true , nil
226316 }
227317
228- topicName , ok := c .registry .TopicName (key )
229- if ! ok {
230- return fmt .Errorf ("no topic name registered for topic key %s" , key )
318+ link , err := c .store .GetPathBuildStore ().Get (ctx , build .PathID , build .Attempt )
319+ if err != nil {
320+ if errors .Is (err , storage .ErrNotFound ) {
321+ metrics .NamedCounter (c .metricsScope , opName , "kill_list_anomalies" , 1 )
322+ c .logger .Warnw ("build exists but its attempt has no link; keeping the build" ,
323+ "build_id" , build .ID ,
324+ "path_id" , build .PathID ,
325+ "attempt" , build .Attempt ,
326+ )
327+ return false , nil
328+ }
329+ metrics .NamedCounter (c .metricsScope , opName , "storage_errors" , 1 )
330+ return false , fmt .Errorf ("failed to look up build for path %s attempt %d: %w" , build .PathID , build .Attempt , err )
231331 }
232332
233- if err := q .Publisher ().Publish (ctx , topicName , msg ); err != nil {
234- return fmt .Errorf ("failed to publish message: %w" , err )
333+ // The attempt's build is a different one: this build lost the dispatch
334+ // race, and nothing downstream will ever look at it.
335+ return link .BuildID != build .ID , nil
336+ }
337+
338+ // findEntry returns the set's entry for a path ID.
339+ func findEntry (set entity.SpeculationPathSet , pathID string ) (entity.SpeculationPathEntry , bool ) {
340+ for _ , entry := range set .Paths {
341+ if entry .ID == pathID {
342+ return entry , true
343+ }
235344 }
345+ return entity.SpeculationPathEntry {}, false
346+ }
236347
237- return nil
348+ // publishBatchID publishes a batch ID to the topic identified by key, with a
349+ // distinct message ID per publish (publish.UniqueID) so a later wake-up for the
350+ // same batch is never deduplicated away.
351+ func (c * Controller ) publishBatchID (ctx context.Context , key consumer.TopicKey , batchID string , partitionKey string ) error {
352+ payload , err := entity.BatchID {ID : batchID }.ToBytes ()
353+ if err != nil {
354+ return fmt .Errorf ("failed to serialize batch ID: %w" , err )
355+ }
356+ return publish .Message (ctx , c .registry , key , publish .UniqueID (batchID ), payload , partitionKey )
238357}
239358
240359// Name returns the controller name for logging and metrics.
0 commit comments