Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 6 additions & 8 deletions doc/rfc/stovepipe/steps/buildsignal.md
Original file line number Diff line number Diff line change
Expand Up @@ -20,14 +20,12 @@ For a delivery carrying build id `B`:

```
1. Load Build B from the build store.
- ErrNotFound -> retryable (build's Create not visible yet; redelivery converges).
- ErrNotFound -> return raw; non-retryable (storage is read-after-write consistent; see [storage README](stovepipe/extension/storage/README.md)).
- other store error -> return raw; classifier decides.

2. Load Request R = store.Get(Build.RequestID) — needed for R.Queue to resolve the build-runner.
- ErrNotFound -> retryable, like step 1: the Build's existence proves the Request write is older,
so a miss here is almost certainly a lagging read; redelivery converges. A genuinely orphaned
Build (integrity fault) still dead-letters at MaxAttempts — the same terminal outcome, without
rejecting straight to DLQ on a stale read.
- ErrNotFound -> return raw; non-retryable, same as step 1 — the Build's existence proves the
Request write is already committed, so a miss here is a storage defect, not a lagging read.

3. If R.State is terminal (superseded / recorded-green / recorded-not-green): ack and return.
- the Request is done (record already ran, or the head was superseded); stop polling.
Expand Down Expand Up @@ -102,19 +100,19 @@ Per `platform/errs`'s non-retryable-by-default rule (see [platform/errs/README.m

| Failure | Disposition | Why |
|---|---|---|
| `Build` not found | retryable (`errs.NewRetryableError`) | `build`'s `Create` not visible yet; redelivery converges. |
| `Request` not found | retryable (`errs.NewRetryableError`) | The Build's existence proves the Request write is older, so a miss is a stale read; a genuine orphan still dead-letters at `MaxAttempts`. |
| `Status` call | raw error; classifier decides | Deliberately left open rather than fixed either way — runner timeout/connection is transient, "runner not deployed for this queue" is not, and only a backend classifier can tell them apart. |
| `Update` CAS conflict (`ErrVersionMismatch`) | retryable | A concurrent (redelivered) writer moved the row; reload and re-check converges. |
| `PublishAfter` re-poll | retryable | The poll heartbeat; it runs only after status/persist/record all succeeded, so a transient enqueue blip is worth replaying to `MaxAttempts` before dead-lettering. |

`Build`/`Request` not found (`storage.ErrNotFound`) are **not** in this table: storage is required to be read-after-write consistent (see [storage README](stovepipe/extension/storage/README.md)), so a miss here is already the correct default (non-retryable, straight to DLQ) rather than a departure worth overriding.

Everything else — factory lookup, an `Update` store error other than a CAS conflict, and the publish to `record` — is returned raw with no override, because the default is already correct: a queue with no registered runner is a config error, and storage/queue failures dead-letter and let DLQ reconciliation recover.

## Idempotency

Every branch is safe under at-least-once redelivery:

- **Build not found** — retryable; converges as the row becomes visible.
- **Build not found** — non-retryable; storage's read-after-write guarantee means a miss here is a storage defect, not a lag condition to retry through.
- **Status already persisted** — a redelivery re-runs the whole algorithm from step 1, including a redundant `Status` poll (harmless — the runner reports the same thing); step 6 no-ops on the unchanged status, and the delivery proceeds to re-schedule the poll (non-terminal) or republish to `record` (terminal, idempotent). No corruption.
- **Terminal already published** — a redelivery reloads, re-polls, no-ops at step 6, republishes the same terminal signal to `record` (idempotent), and acks. Harmless.
- **`PublishAfter` failed, then retried** — the nacked delivery re-runs from step 1; there is no way to resume mid-algorithm, so it re-polls the runner too, but the row already carries the non-terminal status and step 6 no-ops. Only the final enqueue does new work.
Expand Down
1 change: 1 addition & 0 deletions service/stovepipe/server/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ go_library(
"//service/stovepipe/server/mapper:go_default_library",
"//stovepipe/controller:go_default_library",
"//stovepipe/controller/build:go_default_library",
"//stovepipe/controller/buildsignal:go_default_library",
"//stovepipe/controller/dlq:go_default_library",
"//stovepipe/controller/process:go_default_library",
"//stovepipe/core/messagequeue:go_default_library",
Expand Down
26 changes: 24 additions & 2 deletions service/stovepipe/server/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@ import (
"github.com/uber/submitqueue/service/stovepipe/server/mapper"
"github.com/uber/submitqueue/stovepipe/controller"
"github.com/uber/submitqueue/stovepipe/controller/build"
"github.com/uber/submitqueue/stovepipe/controller/buildsignal"
"github.com/uber/submitqueue/stovepipe/controller/dlq"
"github.com/uber/submitqueue/stovepipe/controller/process"
stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
Expand Down Expand Up @@ -382,6 +383,12 @@ func registerPrimaryControllers(
}
count++

buildSignalController := buildsignal.NewController(logger, scope, store, brf, registry, stovepipemq.TopicKeyBuildSignal, "stovepipe-buildsignal")
if err := c.Register(buildSignalController); err != nil {
return count, fmt.Errorf("failed to register buildsignal controller: %w", err)
}
count++

return count, nil
}

Expand All @@ -407,8 +414,10 @@ func registerDLQControllers(

// newTopicRegistry builds the TopicRegistry for Stovepipe's internal pipeline queues. ingest
// publishes to the process topic and the process consumer subscribes to it; process publishes
// to the build topic and the build consumer subscribes to it. The buildsignal topic is added
// once the buildsignal controller lands to consume it.
// to the build topic and the build consumer subscribes to it; build publishes to the buildsignal
// topic and the buildsignal consumer subscribes to it, and also republishes to itself while
// polling. buildsignal publishes to the record topic once a build reaches a terminal status; it
// has no Subscription yet since no consumer for it exists until the record stage lands.
func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRegistry, error) {
return consumer.NewTopicRegistry([]consumer.TopicConfig{
{
Expand All @@ -427,6 +436,19 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
subscriberName, "stovepipe-build",
),
},
{
Key: stovepipemq.TopicKeyBuildSignal,
Name: "buildsignal",
Queue: q,
Subscription: extqueue.DefaultSubscriptionConfig(
subscriberName, "stovepipe-buildsignal",
),
},
{
Key: stovepipemq.TopicKeyRecord,
Name: "record",
Queue: q,
},
{
Key: dlq.TopicKey(stovepipemq.TopicKeyProcess),
Name: "process_dlq",
Expand Down
1 change: 1 addition & 0 deletions stovepipe/controller/build/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ go_library(
"//platform/consumer:go_default_library",
"//platform/errs:go_default_library",
"//platform/metrics:go_default_library",
"//stovepipe/core/loader:go_default_library",
"//stovepipe/core/messagequeue:go_default_library",
"//stovepipe/entity:go_default_library",
"//stovepipe/extension/buildrunner:go_default_library",
Expand Down
7 changes: 2 additions & 5 deletions stovepipe/controller/build/build.go
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@ import (
"github.com/uber/submitqueue/platform/consumer"
"github.com/uber/submitqueue/platform/errs"
"github.com/uber/submitqueue/platform/metrics"
"github.com/uber/submitqueue/stovepipe/core/loader"
stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
"github.com/uber/submitqueue/stovepipe/entity"
"github.com/uber/submitqueue/stovepipe/extension/buildrunner"
Expand Down Expand Up @@ -149,11 +150,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r

// loadRequest returns the request for id.
func (c *Controller) loadRequest(ctx context.Context, id string) (entity.Request, error) {
got, err := c.store.GetRequestStore().Get(ctx, id)
if err != nil {
return entity.Request{}, fmt.Errorf("BuildController failed to load request %s: %w", id, err)
}
return got, nil
return loader.ByID(ctx, id, c.store.GetRequestStore().Get, "BuildController", "request")
}

// publishBuildSignal publishes buildID to the buildsignal stage, partitioned by
Expand Down
44 changes: 44 additions & 0 deletions stovepipe/controller/buildsignal/BUILD.bazel
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
load("@rules_go//go:def.bzl", "go_library", "go_test")

go_library(
name = "go_default_library",
srcs = ["buildsignal.go"],
importpath = "github.com/uber/submitqueue/stovepipe/controller/buildsignal",
visibility = ["//visibility:public"],
deps = [
"//platform/base/messagequeue:go_default_library",
"//platform/consumer:go_default_library",
"//platform/errs:go_default_library",
"//platform/metrics:go_default_library",
"//stovepipe/core/loader:go_default_library",
"//stovepipe/core/messagequeue:go_default_library",
"//stovepipe/entity:go_default_library",
"//stovepipe/extension/buildrunner:go_default_library",
"//stovepipe/extension/storage:go_default_library",
"@com_github_uber_go_tally//:go_default_library",
"@org_uber_go_zap//:go_default_library",
],
)

go_test(
name = "go_default_test",
srcs = ["buildsignal_test.go"],
embed = [":go_default_library"],
deps = [
"//platform/base/messagequeue:go_default_library",
"//platform/consumer:go_default_library",
"//platform/errs:go_default_library",
"//platform/extension/messagequeue/mock:go_default_library",
"//stovepipe/core/messagequeue:go_default_library",
"//stovepipe/entity:go_default_library",
"//stovepipe/extension/buildrunner:go_default_library",
"//stovepipe/extension/buildrunner/mock:go_default_library",
"//stovepipe/extension/storage:go_default_library",
"//stovepipe/extension/storage/mock:go_default_library",
"@com_github_stretchr_testify//assert:go_default_library",
"@com_github_stretchr_testify//require:go_default_library",
"@com_github_uber_go_tally//:go_default_library",
"@org_uber_go_mock//gomock:go_default_library",
"@org_uber_go_zap//:go_default_library",
],
)
Loading
Loading