From 0b4f7892e38fd374926716caf4d9b87a79f6cbdc Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Tue, 28 Jul 2026 14:57:33 -0700 Subject: [PATCH] fix(stovepipe): mint a distinct message id for each buildsignal re-poll MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? `buildsignal` schedules its next poll by re-publishing to its own topic, and it reused the build id as the message id — byte-identical to the message `build` published to start the loop. The MySQL queue dedups on the `(topic, partition_key, id)` unique key and `InsertDelayed` swallows the collision with `ON DUPLICATE KEY UPDATE topic = topic`, returning success. So the reschedule was accepted and silently discarded. This is deterministic, not a race. The re-publish happens before the delivery is acked, and GC only collects up to the minimum *acked* offset on idle ticks, so the colliding row is always still present. The effect: any build that is not terminal on its first poll is never polled again. `Build.Status` freezes at `accepted`/`running`, the request never leaves `processing`, nothing is published to `record`, and the queue's `in_flight_count` slot is never released — so after `MaxConcurrent` such builds the queue stops admitting work entirely. Nothing caught it because the fake build runner could not report a non-terminal status until the previous commit, and the unit tests matched the published message with `gomock.Any()`. ### What? `publishBuildSignal` now mints `{buildID}/poll/{generation}`, where the generation is read off the id of the delivery being processed and incremented — so a chain runs `B` -> `B/poll/1` -> `B/poll/2` -> … The partition key stays the build id, so each build's poll loop keeps its own partition. The generation advances deterministically rather than randomly, which matters for the case a random suffix handles badly. The next id is a pure function of the delivery, so a redelivery racing the original computes the *same* id and dedup collapses the two into one message: the build keeps a single poll chain. A random suffix would instead fork a second chain, doubling the poll rate and racing the first chain's status CAS for no benefit. A stable id is not an option in the other direction either — that is the bug itself. Even a fixed-but-different id like `B/poll` only survives one extra tick, because the second tick's re-poll then collides with the message being processed. ## Test Plan ✅ `bazel test //stovepipe/...` — new unit test asserts successive re-polls mint distinct ids and never reuse the build id. Verified it genuinely regresses: reverting just the id line fails it with `"map[bk-1:{}]" should have 3 item(s), but has 1`. ✅ `bazel test //test/e2e/stovepipe/...` — new e2e ingests a queue carrying the `build-slow` marker and waits for the build row to reach `succeeded`, which requires more than one poll tick. The service log shows three distinct ids on the `buildsignal` topic for one build. --- doc/rfc/stovepipe/steps/buildsignal.md | 4 + .../controller/buildsignal/buildsignal.go | 41 +++++++++- .../buildsignal/buildsignal_test.go | 74 +++++++++++++++++++ test/e2e/stovepipe/harness_test.go | 18 +++++ test/e2e/stovepipe/suite_test.go | 35 ++++++++- 5 files changed, 165 insertions(+), 7 deletions(-) diff --git a/doc/rfc/stovepipe/steps/buildsignal.md b/doc/rfc/stovepipe/steps/buildsignal.md index 3a3c9c4df..e574c35e6 100644 --- a/doc/rfc/stovepipe/steps/buildsignal.md +++ b/doc/rfc/stovepipe/steps/buildsignal.md @@ -61,6 +61,10 @@ For a delivery carrying build id `B`: 8. Else PublishAfter(B -> buildsignal, delayMs), partitioned by build id: - delayMs = pollDelay(status): shorter while running, longer while accepted. - a fresh message (retry_count resets to 0), not a nack — polling is not failure. + - the message id must be unique per tick. The queue dedups on (topic, partition_key, id) + and the delivery being processed is still un-acked, so its row is present: reusing B as + the message id makes every re-poll collide with the message that scheduled it and be + silently discarded, ending the poll loop after one tick. - ack. ``` diff --git a/stovepipe/controller/buildsignal/buildsignal.go b/stovepipe/controller/buildsignal/buildsignal.go index eb65d43bb..4b3092e46 100644 --- a/stovepipe/controller/buildsignal/buildsignal.go +++ b/stovepipe/controller/buildsignal/buildsignal.go @@ -22,6 +22,8 @@ package buildsignal import ( "context" "fmt" + "strconv" + "strings" "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" @@ -151,7 +153,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er } delayMs := pollDelay(effective) - if err := c.publishBuildSignal(ctx, build.ID, delayMs); err != nil { + if err := c.publishBuildSignal(ctx, build.ID, msg.ID, delayMs); err != nil { return errs.NewRetryableError(fmt.Errorf("failed to reschedule poll for build %s: %w", build.ID, err)) } c.logger.Debugw("rescheduled build status poll", @@ -218,15 +220,48 @@ func (c *Controller) publishRecord(ctx context.Context, buildID, requestID strin return c.publish(ctx, stovepipemq.TopicKeyRecord, msg, 0) } +// pollIDInfix separates a build id from its poll generation in a re-poll +// message id: "/poll/". +const pollIDInfix = "/poll/" + +// nextPollMessageID returns the message id for the next poll of buildID, one +// generation past the delivery that scheduled it. currentMsgID is the id of the +// message being processed — either the initial publish from build (no +// generation, so the next is 1) or a previous re-poll. +// +// The generation has to advance because the queue dedups on +// (topic, partition_key, id) and the delivery being processed has not been acked +// yet, so its row is still present: a re-poll reusing the current id would +// collide with the message that scheduled it and be silently discarded, ending +// the poll loop after one tick. +// +// It advances deterministically rather than randomly so the id stays a pure +// function of the delivery. A redelivery racing the original computes the same +// next id, so dedup collapses the two into one message and the build keeps a +// single poll chain — where a random suffix would fork a second chain that +// doubles the poll rate and races the first one's status CAS. +func nextPollMessageID(buildID, currentMsgID string) string { + generation := 0 + if rest, found := strings.CutPrefix(currentMsgID, buildID+pollIDInfix); found { + if n, err := strconv.Atoi(rest); err == nil && n > 0 { + generation = n + } + } + return fmt.Sprintf("%s%s%d", buildID, pollIDInfix, generation+1) +} + // publishBuildSignal re-publishes buildID to buildsignal after delayMs, // partitioned by build id so each build's poll loop runs in its own // partition. A fresh message, not a nack — polling is not failure. -func (c *Controller) publishBuildSignal(ctx context.Context, buildID string, delayMs int64) error { +// +// currentMsgID is the id of the delivery being processed; see nextPollMessageID +// for why the new id is derived from it rather than reused or randomized. +func (c *Controller) publishBuildSignal(ctx context.Context, buildID, currentMsgID string, delayMs int64) error { payload, err := stovepipemq.Marshal(&stovepipemq.BuildSignal{Id: buildID}) if err != nil { return fmt.Errorf("failed to serialize build signal: %w", err) } - msg := entityqueue.NewMessage(buildID, payload, buildID, nil) + msg := entityqueue.NewMessage(nextPollMessageID(buildID, currentMsgID), payload, buildID, nil) return c.publish(ctx, stovepipemq.TopicKeyBuildSignal, msg, delayMs) } diff --git a/stovepipe/controller/buildsignal/buildsignal_test.go b/stovepipe/controller/buildsignal/buildsignal_test.go index 4b14f02f4..3e06be15c 100644 --- a/stovepipe/controller/buildsignal/buildsignal_test.go +++ b/stovepipe/controller/buildsignal/buildsignal_test.go @@ -326,3 +326,77 @@ func TestProcess(t *testing.T) { }) } } + +// TestPublishBuildSignalAdvancesPollGeneration is the regression test for the poll +// loop stalling. The queue dedups on (topic, partition_key, id) and the delivery that +// scheduled a re-poll is still un-acked when the re-poll is published, so a reused +// message id makes the reschedule a silent no-op and the build is never polled again. +// The generation therefore has to advance each tick. The partition key must stay the +// build id so each poll loop keeps its own partition. +func TestPublishBuildSignalAdvancesPollGeneration(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + + var ids []string + m.publisher.EXPECT().PublishAfter(gomock.Any(), "buildsignal", gomock.Any(), PollDelayRunningMs). + DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message, _ int64) error { + ids = append(ids, msg.ID) + assert.Equal(t, testBuildID, msg.PartitionKey) + return nil + }).Times(3) + + // Walk a chain: each publish is scheduled by the message the previous one minted. + current := testBuildID + for range 3 { + require.NoError(t, c.publishBuildSignal(context.Background(), testBuildID, current, PollDelayRunningMs)) + current = ids[len(ids)-1] + } + + assert.Equal(t, []string{ + testBuildID + "/poll/1", + testBuildID + "/poll/2", + testBuildID + "/poll/3", + }, ids, "each tick must mint a fresh id, or the reschedule dedups against the message that scheduled it") +} + +// TestPublishBuildSignalIsIdempotentPerDelivery pins the other half of the contract: +// the next id is a pure function of the delivery being processed. A redelivery racing +// the original computes the same id, so dedup collapses them and the build keeps a +// single poll chain rather than forking a second one that doubles the poll rate and +// races the first one's status CAS. +func TestPublishBuildSignalIsIdempotentPerDelivery(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + + var ids []string + m.publisher.EXPECT().PublishAfter(gomock.Any(), "buildsignal", gomock.Any(), PollDelayRunningMs). + DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message, _ int64) error { + ids = append(ids, msg.ID) + return nil + }).Times(2) + + scheduledBy := testBuildID + "/poll/7" + require.NoError(t, c.publishBuildSignal(context.Background(), testBuildID, scheduledBy, PollDelayRunningMs)) + require.NoError(t, c.publishBuildSignal(context.Background(), testBuildID, scheduledBy, PollDelayRunningMs)) + + assert.Equal(t, ids[0], ids[1], "a redelivery of the same message must republish the same id so dedup collapses it") + assert.Equal(t, testBuildID+"/poll/8", ids[0]) +} + +func TestNextPollMessageID(t *testing.T) { + tests := []struct { + name string + current string + expected string + }{ + {name: "initial publish from build starts at one", current: testBuildID, expected: testBuildID + "/poll/1"}, + {name: "advances the generation", current: testBuildID + "/poll/4", expected: testBuildID + "/poll/5"}, + {name: "unparsable generation restarts at one", current: testBuildID + "/poll/x", expected: testBuildID + "/poll/1"}, + {name: "another build's id is not a prefix match", current: "other/poll/9", expected: testBuildID + "/poll/1"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.expected, nextPollMessageID(testBuildID, tt.current)) + }) + } +} diff --git a/test/e2e/stovepipe/harness_test.go b/test/e2e/stovepipe/harness_test.go index 8eb517c58..70c8604f1 100644 --- a/test/e2e/stovepipe/harness_test.go +++ b/test/e2e/stovepipe/harness_test.go @@ -130,3 +130,21 @@ func (s *StovepipeE2ESuite) assertIngestPersisted(queue, id string) { assert.Equal(t, id, s.uriMapping(queue), "URI mapping should point at the minted request id") assert.Equal(t, 1, s.publishedMessageCount(id), "should have published one process message for %s", id) } + +// awaitBuildStatus blocks until the build row for a request reaches want. The +// pipeline runs inside the stovepipe-service container, so the build's own status +// column is the durable, black-box signal that the poll loop converged: buildsignal +// is its only writer after build creates the row. +func (s *StovepipeE2ESuite) awaitBuildStatus(requestID, want string) { + pollUntil(processPollInterval, func() bool { + var status string + err := s.db.QueryRow("SELECT status FROM build WHERE request_id = ?", requestID).Scan(&status) + if err != nil { + // sql.ErrNoRows means the build row is not created yet. + s.log.Logf("build for %s not readable yet: %v", requestID, err) + return false + } + s.log.Logf("build for %s status = %q (want %q)", requestID, status, want) + return status == want + }) +} diff --git a/test/e2e/stovepipe/suite_test.go b/test/e2e/stovepipe/suite_test.go index 365cb808f..072ae70eb 100644 --- a/test/e2e/stovepipe/suite_test.go +++ b/test/e2e/stovepipe/suite_test.go @@ -30,10 +30,11 @@ package e2e_test // bazel test //test/e2e/stovepipe:stovepipe_test // // The stack runs the Stovepipe gRPC service plus a storage MySQL (request, -// request_uri) and a queue MySQL (the process stage). Unlike the integration -// suite (test/integration/stovepipe), which asserts only that Ingest *publishes* -// a process message, this suite additionally drives the asynchronous process -// consumer to completion — proving the ingest→process pipeline runs end-to-end. +// request_uri, queue, build) and a queue MySQL (the pipeline stages). Unlike the +// integration suite (test/integration/stovepipe), which asserts only that Ingest +// *publishes* a process message, this suite additionally drives the asynchronous +// consumers to completion — proving the ingest→process→build→buildsignal pipeline +// runs end-to-end. import ( "context" @@ -157,3 +158,29 @@ func (s *StovepipeE2ESuite) TestIngest_Idempotent() { id2 := s.ingest(queue) assert.Equal(s.T(), id, id2, "re-ingest of the same head should dedup to the same id") } + +// TestIngest_SlowBuild_PollsToCompletion drives a build that is not terminal on its +// first poll, which is the only path that exercises buildsignal's reschedule. +// +// The queue name carries a fake-buildrunner marker: the fake SourceControl resolves a +// queue to "git:///HEAD", so the marker rides into the head URI and the fake +// BuildRunner reports running for a while before succeeding. Reaching a terminal build +// status therefore requires the poll loop to tick more than once. +// +// This is the regression test for the loop stalling: buildsignal re-publishes to its +// own topic to schedule the next poll, and the queue dedups on +// (topic, partition_key, id). While the delivery being processed is still un-acked its +// row is present, so a re-poll that reuses the build id as the message id is silently +// discarded and the build is never polled again — the build would sit at `running` +// forever. +func (s *StovepipeE2ESuite) TestIngest_SlowBuild_PollsToCompletion() { + const queue = "monorepo/slow?buildrunner-fake=build-slow" + + id := s.ingest(queue) + s.log.Logf("Ingest succeeded: id=%s; waiting for the poll loop to reach a terminal build", id) + + s.assertIngestPersisted(queue, id) + + // Getting here at all means the reschedule produced a deliverable message. + s.awaitBuildStatus(id, "succeeded") +}