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
25 changes: 21 additions & 4 deletions doc/rfc/stovepipe/steps/process.md
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ For a delivery carrying request id `R`:
2. If R.State is terminal (superseded / succeeded / failed / cancelled):
- ack and return (idempotent no-op).
3. If R.State is processing (strategy already recorded):
- re-publish R to build (the prior publish may have failed), ack, return.
- re-announce validation start and re-publish R to build (either prior publish may have failed), ack, return.
4. R.State is accepted. Load the Queue row Q.
5. Coalesce: if CompareRequestID(R.Queue, R.ID, Q.latest_request_id) < 0:
- a newer head exists -> mark R superseded, ack, return. (No slot consumed.)
Expand All @@ -31,8 +31,9 @@ For a delivery carrying request id `R`:
a. Derive build strategy + baseline (see "Build-strategy decision").
b. CAS the Queue row: in_flight_count += 1.
c. CAS the Request: accepted -> processing, persist build_strategy + base_uri.
d. Publish R to build.
e. ack.
d. Announce validation start on the hook topic (see "Hooks").
e. Publish R to build.
f. ack.
```

Step 5 runs regardless of the gate: an intermediate head is superseded on sight (even mid-validation), because superseding consumes no slot.
Expand Down Expand Up @@ -138,12 +139,28 @@ A, D, F each get a full cycle; B, C, E end `superseded`. No intermediate is vali
- A newer head does **not** preempt an in-flight validation.
- Deferred messages are **not** failed or dead-lettered — they wait for the gate (see [Waiting for a slot](#waiting-for-a-slot)).

## Hooks

Admitting a request is when the rest of the company can learn "validation of this commit has begun". `process` publishes that as a `HookEvent` on Stovepipe's durable `hook` topic — the same seam `record` uses to announce the outcome. The mechanics (envelope, delivery promise, per-domain dispatcher stage, `hook_dlq`) are settled in [hook-framework.md](../../hook-framework.md); this section covers only what admitting has to decide.

The event type is `validation.repository.started`. Its payload names the Queue and the Request and nothing else, exactly as the terminal events in [record.md](record.md#hooks) do: a hook resolves the commit, the chosen strategy, and the baseline from the request store rather than reading a snapshot off the wire.

Published after the admit CAS and before the publish to `build`:

```
CAS accepted -> processing → publish HookEvent → publish to build → ack
```

After the CAS because the payload names the Request rather than snapshotting it, and the two facts a start event exists to carry — the scope it chose and the baseline it builds on — are written by that very CAS. A hook that reloads the Request must not find it still `accepted` with neither set. Before the build publish because the announce is the cheaper of the two to retry: a failed announce leaves nothing downstream to undo, whereas announcing after the build publish would make a failed announce force the redelivery to re-publish a build that was already accepted.

Only an admit announces. A Request that coalescing supersedes never reaches step 7, so it produces no start event — and a start event is not a promise that a verdict follows, since an admitted Request can still be cancelled or driven to a fail-closed outcome. Consumers pairing a start with an end must tolerate a start that never gets one.

## Idempotency and at-least-once delivery

Every branch is safe under redelivery:

- **accepted, no strategy** → full admit path. On a crash after incrementing `in_flight_count` but before persisting `processing`, redelivery re-reads `accepted` and re-runs; the increment re-applies only if the count CAS hasn't already moved (see integrity below).
- **processing** → re-publish to `build` and ack. The `build` consumer is keyed on the request id and idempotent, so a duplicate publish is harmless.
- **processing** → re-announce the start event, re-publish to `build`, ack. The `build` consumer is keyed on the request id and idempotent, so a duplicate publish is harmless, and the start event's id is derived from the transition rather than the clock, so a re-announce carries the id the first attempt would have and consumers dedupe on it. Re-announcing here is what makes the event at-least-once rather than at-most-once: this is the only branch a redelivery takes once `processing` is durable, so an admit that failed after the state write would otherwise lose the event for good.
- **terminal** (superseded / recorded) → ack, no-op.
- **deferred (waiting for slot)** → no state or count change; pure deferral (re-enters when the held delivery comes due).

Expand Down
14 changes: 7 additions & 7 deletions doc/rfc/stovepipe/workflow.md
Original file line number Diff line number Diff line change
Expand Up @@ -55,10 +55,10 @@ The ref is a *cache* of the last-green URI, not a second record of greenness. It
|---|---|
| **SourceControl** | Resolve a Queue name to its current head URI; answer ancestry/comparison questions between two URIs (is the new head a fast-forward descendant of the last green, or was history rewritten?); enumerate commits in a range; advance the Queue's **promotion ref** to a commit. The sole owner of URI semantics, including which refs a Queue name resolves to. |
| **build-runner** | Build a scope at a URI (optionally relative to a baseline URI), returning pass/fail and the target graph. See [build-runner.md](../submitqueue/build-runner.md). |
| **Hooks** | Deliver Stovepipe's greenness events to downstream systems — "this URI / this project is now green (or not green)". Fire-and-forget notification, decoupled so Stovepipe does not know or care who consumes the event. The shared cross-domain hook seam rather than a Stovepipe-specific extension. See [hook-framework.md](../hook-framework.md). |
| **Hooks** | Deliver Stovepipe's validation events to downstream systems — "validation of this URI has begun", "this URI / this project is now green (or not green)". Fire-and-forget notification, decoupled so Stovepipe does not know or care who consumes the event. The shared cross-domain hook seam rather than a Stovepipe-specific extension. See [hook-framework.md](../hook-framework.md). |
| **Storage** | Persist Queues (incl. last-green URI), Requests, build records, and per-URI / per-project greenness. Key/value-shaped per the extension-design rules in [AGENTS.md](../../../AGENTS.md). |

Hooks are the notification boundary. When a validation fact is recorded — whole-repo green/not-green, or later a project green/not-green — the event reaches deployment systems, dashboards, and developer tooling without any of them polling Stovepipe's store, and each environment can route it to its own downstream (a deploy gate, a Slack notifier, an event bus) without changing the pipeline. The mechanism is the cross-domain hook framework rather than a call out of the recording stage: `record` publishes a `HookEvent` to Stovepipe's `hook` topic, and a dispatcher stage consumes it and invokes the wired hooks, so a slow or failing downstream cannot add latency to the pipeline. Both halves exist; what a deployment supplies is the hooks themselves, since the example server resolves every event to `noop`. See [record.md](steps/record.md#hooks) for the fact-to-event mapping.
Hooks are the notification boundary. When validation of a commit begins, and when a validation fact is recorded — whole-repo green/not-green, or later a project green/not-green — the event reaches deployment systems, dashboards, and developer tooling without any of them polling Stovepipe's store, and each environment can route it to its own downstream (a deploy gate, a Slack notifier, an event bus) without changing the pipeline. The mechanism is the cross-domain hook framework rather than a call out of the pipeline stages: `process` and `record` publish a `HookEvent` to Stovepipe's `hook` topic, and a dispatcher stage consumes it and invokes the wired hooks, so a slow or failing downstream cannot add latency to the pipeline. Both halves exist; what a deployment supplies is the hooks themselves, since the example server resolves every event to `noop`. See [process.md](steps/process.md#hooks) for the start event and [record.md](steps/record.md#hooks) for the fact-to-event mapping.

## Workflow

Expand All @@ -74,9 +74,9 @@ The pipeline runs in two phases against the same Request. **Phase 1** establishe
└───────────────┬──────────────┘
│ RequestID
┌──────────────────────────────┐
│ process │
│ Ask SourceControl: is head a │
┌──────────────────────────────┐ Hooks
│ process │┄┄┄┄┄► "validation
│ Ask SourceControl: is head a │ started"
│ descendant of last-green? │
│ → incremental since green │
│ else (history rewrite) │
Expand Down Expand Up @@ -132,7 +132,7 @@ The pipeline runs in two phases against the same Request. **Phase 1** establishe
### Phase 1 — whole-repo greenness

1. **ingest** — invoked by the external poller with a **Queue name**. It asks `SourceControl` for that Queue's current head URI, mints a Request namespaced by the Queue, persists it with no recorded greenness yet, and dedups on `(Queue, head URI)` so a re-reported head is processed once. It publishes the RequestID onward.
2. **process** — decides build strategy (incremental since last-green vs full monorepo), gates concurrent work per Queue, coalesces backlog to the latest head, and publishes to `build`. See [process.md](steps/process.md).
2. **process** — decides build strategy (incremental since last-green vs full monorepo), gates concurrent work per Queue, coalesces backlog to the latest head, publishes a **hook event** announcing that validation of the commit has begun, and publishes to `build`. See [process.md](steps/process.md).
3. **build** — runs the build-runner for the chosen scope. A flag derived from `process` decides whether to build relative to the last-green **baseline URI** (incremental) or from scratch (full). It records a build and publishes the BuildID.
4. **buildsignal** — records the build's status and target graph when the build completes, then releases the Queue's `in_flight_count` slot, projects the terminal status onto the Request (`succeeded` / `failed` / `cancelled`), and publishes the RequestID to `record`.
5. **record** — writes the whole-repo greenness for the head URI (`0` green / `1` broken to start), derived from the Request's build outcome. On green it advances the Queue's **last-green URI** so the next `process` can build incrementally from here, and asks `SourceControl` to advance the Queue's **promotion ref** to the same commit (see [Promotion ref](#promotion-ref--the-last-green-commit-by-name)). It publishes a **hook event** for the green/not-green transition, then fans out into Phase 2. The Queue's `in_flight_count` was already released by `buildsignal` when the build went terminal.
Expand All @@ -150,7 +150,7 @@ The pipeline runs in two phases against the same Request. **Phase 1** establishe
| Controller | In | Out | One-line role |
|---|---|---|---|
| **ingest** | Queue name (from poller) | process | Resolve head URI via SourceControl, mint Request, persist (no greenness), dedup on `(Queue, head URI)` |
| **process** | RequestID | build | Build strategy, concurrency gate, backlog coalescing → [process.md](steps/process.md) |
| **process** | RequestID | build, hook topic | Build strategy, concurrency gate, backlog coalescing; announce validation start on admit → [process.md](steps/process.md) |
| **build** | RequestID | buildsignal | Run the build-runner for the chosen scope; baseline = last-green URI iff incremental |
| **buildsignal** | BuildID | record (P1), record (P2) | Record build status + target graph; release `in_flight_count`; project the outcome onto the Request; signal completion |
| **record** | RequestID | analyze (P1→P2), hook topic | Write greenness; on whole-repo green advance last-green URI and the promotion ref; publish the hook event |
Expand Down
5 changes: 5 additions & 0 deletions stovepipe/controller/process/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -6,10 +6,13 @@ go_library(
importpath = "github.com/uber/submitqueue/stovepipe/controller/process",
visibility = ["//visibility:public"],
deps = [
"//api/base/hook:go_default_library",
"//platform/consumer:go_default_library",
"//platform/errs:go_default_library",
"//platform/hook:go_default_library",
"//platform/metrics:go_default_library",
"//platform/publish:go_default_library",
"//stovepipe/core/hookevent:go_default_library",
"//stovepipe/core/loader:go_default_library",
"//stovepipe/core/messagequeue:go_default_library",
"//stovepipe/entity:go_default_library",
Expand All @@ -26,12 +29,14 @@ go_test(
srcs = ["process_test.go"],
embed = [":go_default_library"],
deps = [
"//api/base/hook:go_default_library",
"//platform/base/messagequeue:go_default_library",
"//platform/consumer:go_default_library",
"//platform/consumer/mock:go_default_library",
"//platform/errs:go_default_library",
"//platform/extension/messagequeue/mock:go_default_library",
"//platform/metrics:go_default_library",
"//stovepipe/core/hookevent:go_default_library",
"//stovepipe/core/messagequeue:go_default_library",
"//stovepipe/entity:go_default_library",
"//stovepipe/extension/queueconfig/default:go_default_library",
Expand Down
42 changes: 42 additions & 0 deletions stovepipe/controller/process/process.go
Original file line number Diff line number Diff line change
Expand Up @@ -24,10 +24,13 @@ import (
"fmt"

"github.com/uber-go/tally"
basehook "github.com/uber/submitqueue/api/base/hook"
"github.com/uber/submitqueue/platform/consumer"
"github.com/uber/submitqueue/platform/errs"
platformhook "github.com/uber/submitqueue/platform/hook"
"github.com/uber/submitqueue/platform/metrics"
"github.com/uber/submitqueue/platform/publish"
"github.com/uber/submitqueue/stovepipe/core/hookevent"
"github.com/uber/submitqueue/stovepipe/core/loader"
stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
"github.com/uber/submitqueue/stovepipe/entity"
Expand Down Expand Up @@ -113,6 +116,14 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er

switch request.State {
case entity.RequestStateProcessing:
// Announce here as well as at admit: this is the only path a redelivery
// takes once the transition is durable, so an admit that failed after
// persisting would otherwise lose the start event for good. The event id
// is derived from the transition, so a repeat carries the id the first
// attempt would have and a consumer deduplicates on it.
if err := c.publishHookEvent(ctx, request, hookevent.NewValidationRepositoryStarted(request)); err != nil {
return err
}
if err := c.publishBuild(ctx, request.ID, request.Queue); err != nil {
metrics.NamedCounter(c.metricsScope, _opName, "publish_errors", 1, metrics.TagsFromContext(ctx)...)
return fmt.Errorf("failed to publish request %s to build: %w", request.ID, err)
Expand Down Expand Up @@ -252,6 +263,10 @@ func (c *Controller) admitLatestHead(ctx context.Context, store storage.Storage,
return nil
}

if err := c.publishHookEvent(ctx, request, hookevent.NewValidationRepositoryStarted(request)); err != nil {
return err
}

if err := c.publishBuild(ctx, request.ID, request.Queue); err != nil {
metrics.NamedCounter(c.metricsScope, _opName, "publish_errors", 1, metrics.TagsFromContext(ctx)...)
return fmt.Errorf("failed to publish request %s to build: %w", request.ID, err)
Expand Down Expand Up @@ -474,6 +489,33 @@ func (c *Controller) publishBuild(ctx context.Context, id, queue string) error {
return nil
}

// publishHookEvent announces a lifecycle transition on the hook topic.
//
// Published only once the transition is durable: the payload names the request
// rather than snapshotting it, so a hook that reloads it must not find a request
// whose strategy and baseline are still unwritten.
//
// Partitioning by request id matches the process topic's own, carrying
// per-request ordering across the seam.
func (c *Controller) publishHookEvent(ctx context.Context, request entity.Request, event *basehook.HookEvent) error {
if err := platformhook.Publish(ctx, c.registry, event, request.ID); err != nil {
metrics.NamedCounter(c.metricsScope, _opName, "hook_errors", 1, metrics.TagsFromContext(ctx)...)
return fmt.Errorf("failed to announce %s for request %s: %w", event.GetType(), request.ID, err)
}

metrics.NamedCounter(c.metricsScope, _opName, "hook_events_published", 1,
metrics.TagsFromContext(ctx, metrics.NewTag("event_type", event.GetType()))...,
)
c.logger.Debugw("announced validation event",
"queue", request.Queue,
"request_id", request.ID,
"uri", request.URI,
"event_type", event.GetType(),
"event_id", event.GetId(),
)
return nil
}

// Name returns the controller name for logging and metrics.
func (c *Controller) Name() string {
return "process"
Expand Down
Loading
Loading