diff --git a/doc/rfc/stovepipe/steps/record.md b/doc/rfc/stovepipe/steps/record.md index e822c9ac6..90e913005 100644 --- a/doc/rfc/stovepipe/steps/record.md +++ b/doc/rfc/stovepipe/steps/record.md @@ -2,10 +2,10 @@ `record` turns a terminal build outcome into a durable validation fact. -- Phase 1 records whole-repository greenness. On green it also advances the Queue's last-green bookmark and promotes the commit onto the Queue's promotion ref. This is implemented: see [stovepipe/controller/record/record.go](../../../../stovepipe/controller/record/record.go). +- Phase 1 records whole-repository greenness. On green it also advances the Queue's last-green bookmark and promotes the commit onto the Queue's promotion ref, then announces the outcome as a hook event. This is implemented: see [stovepipe/controller/record/record.go](../../../../stovepipe/controller/record/record.go). - Phase 2 records greenness per project instead of per repository. Sketched here, to be expanded before implementation. -Notifying downstream systems is **not** implemented in either phase. It will ride the cross-domain hook framework instead of a Stovepipe-specific extension; see [Hooks](#hooks). +Notifying downstream systems rides the cross-domain hook framework rather than a Stovepipe-specific extension; see [Hooks](#hooks). See [workflow.md](../workflow.md) for the whole pipeline, [build.md](build.md) for how builds are created, and [buildsignal.md](buildsignal.md) for the terminal-only handoff into this stage. @@ -34,8 +34,8 @@ For a delivery carrying a `Record` payload: 5. Inspect R.State. - succeeded / failed -> continue. buildsignal stamps the outcome before it publishes here, and both values are verdicts about the code. - - cancelled -> ack, no fact: the build decided nothing about the commit (see - "When to record an outcome"). + - cancelled -> publish a validation.repository.cancelled HookEvent and ack, no + fact: the build decided nothing about the commit (see "When to record an outcome"). - superseded -> ack, no fact. Unreachable in practice. - accepted / processing / anything else -> return a non-retryable invariant error. @@ -44,8 +44,9 @@ For a delivery carrying a `Record` payload: - ErrAlreadyExists -> load and reconcile the existing immutable fact. - other store error -> return raw. -7. If the persisted fact is not green, ack. Report how long the break went undetected - first, but only if step 6 is the write that created the fact (see "Observability"). +7. If the persisted fact is not green, skip to step 10: a break moves neither the + bookmark nor the ref, but is still announced. Report how long it went undetected + first, and only if step 6 is the write that created the fact (see "Observability"). 8. Advance the bookmark to (R.URI, R.ID) in a CAS retry loop, which also reports whether R holds the bookmark afterwards: @@ -56,15 +57,19 @@ For a delivery carrying a `Record` payload: - ErrVersionMismatch -> reload and re-evaluate. 9. If R holds the bookmark, ask SourceControl to point the promotion ref at R.URI. - Otherwise ack: whichever commit holds the bookmark owns the ref. - - ErrNotFound -> count and ack. A rewritten history dropped the commit from the + Otherwise skip: whichever commit holds the bookmark owns the ref. + - ErrNotFound -> count and skip. A rewritten history dropped the commit from the ref, and no retry can promote it. - other error -> return raw. -10. ack. +10. Publish a validation.repository.recorded HookEvent naming the request whose fact + was persisted. + - publish failure -> return raw; non-retryable, so it dead-letters. + +11. ack. ``` -Every decision after step 6 uses the persisted fact, not the outcome read from this delivery's Request. The first immutable fact controls the bookmark and the ref, and will control the hook event. Once hooks land, the publish becomes a new step between 9 and 10. +Every decision after step 6 uses the persisted fact, not the outcome read from this delivery's Request. The first immutable fact controls the bookmark, the ref, and the hook event. ## Validation facts @@ -99,7 +104,7 @@ Absence is still distinct from degree `0`. Callers gating deployments must treat A fact is written only when the Request reaches `succeeded` or `failed`. A `cancelled` build is acked with no fact. -The fail-closed path is subtler than "no fact". A DLQ reconciler forces the Request to `failed` and does not itself publish here, so reconciliation on its own records nothing. That is not the same as nothing being recorded: a `buildsignal` delivery still in flight can reach this stage afterwards, and it will write `DegreeBroken` from the forced state. See [What fail-closed actually guarantees](#what-fail-closed-actually-guarantees). +The fail-closed path is subtler than "no fact". A DLQ reconciler forces the Request to `failed` and does not itself publish here, so reconciliation on its own records nothing. That is not the same as nothing being recorded: a `buildsignal` delivery still in flight can reach this stage afterwards, and it will write `DegreeBroken` from the forced state. See [DLQ and fail-closed behavior](#dlq-and-fail-closed-behavior). ### Phase 1 degree mapping @@ -156,7 +161,7 @@ Recording a fact is when the rest of the company can learn "this URI is now gree The mechanics are already settled in [hook-framework.md](../../hook-framework.md) — the envelope, the delivery promise, the per-domain dispatcher stage, the `hook_dlq`, and the reasoning behind each. This section covers only what Stovepipe has to decide for validation facts. The earlier design here, a Stovepipe `Hooks` extension called with `Notify(ctx, ValidationFactRef{…})` as the last algorithm step, is rejected there on both halves: inline calls couple pipeline latency to third-party integrations and drop the notification on a crash between the state write and the call, and a per-domain contract multiplies schemas and sinks for no gain. -None of it is built. It needs the shared `HookEvent` contract at `api/base/hook/` and the hook extension at `platform/extension/hook/`, plus Stovepipe's own share: a `hook` topic key, a dispatcher stage, a `hook_dlq` reconciler, and the wiring for all three. +The shared `HookEvent` contract at `api/base/hook/`, the hook extension at `platform/extension/hook/`, and Stovepipe's dispatcher stage and `hook_dlq` reconciler are all in place, and this stage publishes to them. What no deployment has yet is a hook that does something with the event: the resolver in `service/stovepipe/server/main.go` returns `noop`. ### Where the publish belongs @@ -166,35 +171,38 @@ Last thing before the ack, after the fact write and after both caches derived fr create fact → advance bookmark (green only) → promote (bookmark holder only) → publish HookEvent → [Phase 2: publish to analyze] → ack ``` -Last because the payload carries no entity snapshot and hooks resolve entities from stores: a hook reacting to "URI is green" by reading `LastGreenURI`, or by fetching the promotion ref, must not find either still pointing at the previous commit. Inside the delivery rather than after it, because that is what makes the event lossless without an outbox — the state writes are recognize-and-skip on redelivery, so a crash before the ack replays the whole chain. +Last because the payload names the Request rather than snapshotting it, and hooks resolve entities from stores: a hook reacting to "URI is green" by reading `LastGreenURI`, or by fetching the promotion ref, must not find either still pointing at the previous commit. Inside the delivery rather than after it, because that is what makes the event lossless without an outbox — the state writes are recognize-and-skip on redelivery, so a crash before the ack replays the whole chain. ### Event shape -| Envelope field | Value for a validation fact | -| -------------- | --------------------------------------------------------------- | -| `source` | `stovepipe` | -| `type` | `validation.repository.recorded` (see below) | -| `version` | `0` — a fact is create-only and has no version to report | -| `timestamp_ms` | Publish time; the fact's own `CreatedAt` travels in the payload | -| `id` | `source` / `type` / request id (see below) | +| Envelope field | Value for a validation fact | +| -------------- | --------------------------------------------------------------------------------- | +| `source` | `stovepipe` | +| `type` | `validation.repository.recorded` or `validation.repository.cancelled` (see below) | +| `version` | `0` — a fact is create-only and has no version to report | +| `timestamp_ms` | Publish time | +| `id` | `source` / `type` / request id / `0` | The **subject** is the Request — a payload fact rather than an envelope field, but it is what `id` is minted from and what the event partitions on. The Request over the URI keeps partitioning identical to the `record` topic's own, so per-request ordering carries through the seam, and it hands a consumer a way back into the pipeline. The two are near-interchangeable anyway: ingest dedups on `(queue, uri)`, so one Request means one URI. -The payload carries the fact's identity and value: `queue`, `uri`, `project`, `degree`, `request_id`. `queue` has to be there because neither the envelope nor the fact entity carries one, so the event is the only place a cross-queue consumer sees it. `degree` is there even though a hook could read it from the store: the fact is immutable, so the staleness objection behind the no-snapshots rule does not apply, and it is what a consumer branches on to tell green from broken. Build failure detail stays off, since the payload is reserved for facts persisted nowhere else and a failed build's detail is durable on the `Build` row. +Both types carry identity and nothing else: `queue` and `request_id`. A hook resolves the fact, its degree, and the bookmark from the stores. `queue` has to be there because neither the envelope nor the fact entity carries one, and a hook needs it to resolve per-queue storage before it can load anything by `request_id`. Identity rather than a snapshot because a snapshot is never fresher than its publish moment, so a consumer acting on one may be reasoning about state that has since moved; the cost is a store read per hook. + +So a hook reads state as it is when the hook runs, not as it was when the event was published. That is the same ordering the step list depends on, and a consumer needing the values a green commit was recorded with reads the immutable fact rather than the bookmark or the ref. ### What the `type` carries Data belongs in `type` when consumers need to avoid receiving the event, and in the payload when they need to interpret it. The type names the scope, `validation.repository.recorded`, with `validation.project.recorded` beside it in Phase 2. +`validation.repository.cancelled` is a separate type rather than a `state` on the recorded one, because the two differ in what a consumer can do with them: a recorded event carries a verdict to act on, a cancelled event only says to stop waiting. A sink that gates on greenness subscribes to one and ignores the other, and naming a cancellation "recorded" would claim a fact that was never written. ### What a consumer can and cannot assume Ordering is per-subject only and the subject is the Request, so events for *different* Requests can arrive out of order. A consumer must not infer "the newest green commit" from arrival order; it should compare request ids by ingest order (`entity.CompareRequestID`) or read the bookmark, which is monotonic by construction. -Absence of an event is not a signal: a cancelled build records nothing, and a Request abandoned before any build went terminal never reaches this stage, so a consumer waiting for one event per ingested commit waits forever on those. Gating keeps treating "no recorded fact" as not green. The converse holds too — an event is not proof the code was tested, since a fail-closed Request can produce a broken fact without a build having failed. +Absence of an event is not a signal. A Request abandoned before any build went terminal never reaches this stage, and a superseded one publishes nothing, so a consumer waiting for one event per ingested commit waits forever on those. Gating keeps treating "no recorded fact" as not green. The converse holds too — an event is not proof the code was tested, since a fail-closed Request can produce a broken fact without a build having failed. -Hooks here must be idempotent on `id`, as everywhere. "Fire-and-forget" describes downstream consumption, not the publish: `record` never waits for a hook, but a failed *publish* fails the delivery. Per `[platform/errs](../../../../platform/errs/README.md)` rule 4 it is not wrapped retryable just because replaying it is convenient, so it dead-letters, which is where the missing `record_dlq` reconciler stops being theoretical: the fact is durable and only the notification is lost. +Hooks here must be idempotent on `id`, as everywhere. "Fire-and-forget" describes downstream consumption, not the publish: `record` never waits for a hook, but a failed *publish* fails the delivery. Per `[platform/errs](../../../../platform/errs/README.md)` rule 4 it is not wrapped retryable just because replaying it is convenient, so it dead-letters. The `record_dlq` consumer is this same controller on the dead-letter topic, so it re-runs this identical idempotent algorithm and the republish is its own recovery path: the fact is already durable, and only the notification was outstanding. ## Request lifecycle @@ -246,7 +254,7 @@ Partitioning by request id keeps completion bookkeeping single-writer per Reques ## Edge cases and idempotency -Every effect is recognize-and-skip, so a redelivery after a complete run re-runs each step as a no-op. Once hooks land the publish re-fires, and the framework's dedupe on `id` absorbs it. +Every effect is recognize-and-skip, so a redelivery after a complete run re-runs each step as a no-op. The publish re-fires, and the queue's dedupe on the derived `id` absorbs it. - **Request not visible.** A storage defect rather than lag, since the publish follows the committed outcome write. Non-retryable. - **Fact already created.** Load it and continue from the stored fact. A fact from a *different* Request, or a Request carrying no build outcome, is an invariant violation rather than an expected outcome. @@ -255,7 +263,7 @@ Every effect is recognize-and-skip, so a redelivery after a complete run re-runs - **Green fact recorded out of order across Requests.** An older green commit can reach this stage after a newer one. Its fact is written as usual, since facts are per-URI and independent, but the bookmark guard skips it, and because it does not hold the bookmark it does not promote either. Neither cache moves backwards. - **Cancelled build.** Stovepipe never initiates cancellation, but a backend may still report it. No fact is written; the slot was already released and the Request already stamped `cancelled`. The fact identity stays unclaimed, and nothing can claim it today, since `cancelled` is terminal and re-validation does not exist. Recovery in practice is the next commit; the unclaimed identity only matters to a future re-run mechanism (see [Supporting re-run of the same URI](#supporting-re-run-of-the-same-uri)). - **History rewrite while a build runs.** The Request keeps the strategy and URI pinned at admission, and record stores the fact about that immutable URI. A later head is handled independently by `process`. If the rewrite dropped the commit from the ref, the fact and the bookmark still stand, since they describe a commit and not a ref, and only the promotion is skipped. -- **A fail-closed terminal outranks a build that passed.** The degree derives from `R.State`, so a Request forced to `failed` by DLQ reconciliation records `DegreeBroken` even when one of its builds reports success afterwards. Reachable today, and permanent once written; see [What fail-closed actually guarantees](#what-fail-closed-actually-guarantees). +- **A fail-closed terminal outranks a build that passed.** The degree derives from `R.State`, so a Request forced to `failed` by DLQ reconciliation records `DegreeBroken` even when one of its builds reports success afterwards. Reachable today, and permanent once written; see [DLQ and fail-closed behavior](#dlq-and-fail-closed-behavior). - **Crash between the fact write and the bookmark advance.** Redelivery reloads the existing fact and re-applies the idempotent guard. - **Crash between the bookmark advance and the promotion.** Redelivery finds its own id on the bookmark, reports that it holds it, and retries the idempotent promotion. @@ -267,7 +275,7 @@ Every effect is recognize-and-skip, so a redelivery after a complete run re-runs Two different things put a message there, and only one is a poison payload. A delivery that fails with its retry budget spent is dead-lettered by the nack itself, carrying the reason it actually failed. A delivery that never reaches a nack, because it crashed or because its **ack failed** and the visibility timeout redelivered it, is dead-lettered by the poll loop once `retry_count` reaches `MaxAttempts` (3 by default), without the controller running on that final attempt and with only a generic reason recorded. So a missing reconciler exposes more than malformed messages: a fact can be lost to a storage failure that would have succeeded on a later retry, or to an ack that never landed even though the write did. -Gating stays safe, because everything this stage can lose reads as not-green: a Request with no fact is indistinguishable from one not yet validated. What is lost is the *fact*. A green build whose fact write permanently failed leaves the URI looking unvalidated, which costs the queue an incremental baseline and forces a full build at the next head. Once hooks land, a lost notification joins that list, and unlike the fact it gets no second chance from a later commit. +Gating stays safe, because everything this stage can lose reads as not-green: a Request with no fact is indistinguishable from one not yet validated. What is lost is the *fact*. A green build whose fact write permanently failed leaves the URI looking unvalidated, which costs the queue an incremental baseline and forces a full build at the next head. A lost notification joins that list, and unlike the fact it gets no second chance from a later commit. This is the same failure shape [buildsignal.md](buildsignal.md#what-it-costs-when-a-backend-does-not-classify-status-errors) describes for a deployment that registers primary consumers without their reconciler. When the reconciler is built it should re-run this same idempotent algorithm from the request id, under `errs.AlwaysRetryableProcessor`: write and publish the immutable fact as usual if the Request carries a build outcome, keep retrying if Request storage is temporarily unavailable, and treat a malformed payload or a permanently missing Request as poison, which needs an operational alert rather than more retries. diff --git a/doc/rfc/stovepipe/workflow.md b/doc/rfc/stovepipe/workflow.md index 73bba9045..603a50339 100644 --- a/doc/rfc/stovepipe/workflow.md +++ b/doc/rfc/stovepipe/workflow.md @@ -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. Not implemented yet; it will be the shared cross-domain hook seam rather than a Stovepipe-specific extension. See [hook-framework.md](../hook-framework.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). | | **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. Neither half exists yet; see [record.md](steps/record.md#hooks) for the fact-to-event mapping and its open questions. +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. ## Workflow @@ -163,7 +163,7 @@ Per-stage design detail lives under `steps/` so this doc stays a pipeline overvi - [process.md](steps/process.md) — build-strategy decision, concurrency gate, backlog coalescing, [concurrency lifecycle](steps/process.md#concurrency-lifecycle), entity changes, [waiting for a slot](steps/process.md#waiting-for-a-slot) - [build.md](steps/build.md) — trigger-only stage: reads the decided scope off the Request, triggers the build-runner, hands off to buildsignal; the stovepipe `BuildRunner` contract and why it differs from SubmitQueue's - [buildsignal.md](steps/buildsignal.md) — the poll loop: hold-based re-poll cadence, target-graph return, per-build partitioning, and the fail-closed handoff to record -- [record.md](steps/record.md) — turning a terminal build outcome into an immutable validation fact, monotonic last-green advancement and ref promotion, and the deferred hook and analyze handoffs +- [record.md](steps/record.md) — turning a terminal build outcome into an immutable validation fact, monotonic last-green advancement and ref promotion, the hook event announcing the outcome, and the deferred analyze handoff ## Dedup, idempotency, and history rewrites diff --git a/platform/hook/BUILD.bazel b/platform/hook/BUILD.bazel index 1998e1da7..7809c529c 100644 --- a/platform/hook/BUILD.bazel +++ b/platform/hook/BUILD.bazel @@ -5,6 +5,7 @@ go_library( srcs = [ "controller.go", "dlq.go", + "publisher.go", ], importpath = "github.com/uber/submitqueue/platform/hook", visibility = ["//visibility:public"], @@ -14,6 +15,7 @@ go_library( "//platform/errs:go_default_library", "//platform/extension/hook:go_default_library", "//platform/metrics:go_default_library", + "//platform/publish:go_default_library", "@com_github_uber_go_tally//:go_default_library", "@org_uber_go_zap//:go_default_library", ], @@ -24,14 +26,17 @@ go_test( srcs = [ "controller_test.go", "dlq_test.go", + "publisher_test.go", ], embed = [":go_default_library"], deps = [ "//api/base/hook:go_default_library", "//platform/base/failure:go_default_library", "//platform/base/messagequeue:go_default_library", + "//platform/consumer:go_default_library", "//platform/consumer/mock:go_default_library", "//platform/extension/hook:go_default_library", + "//platform/extension/messagequeue/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", diff --git a/platform/hook/controller.go b/platform/hook/controller.go index c46388c9f..ff4283e8b 100644 --- a/platform/hook/controller.go +++ b/platform/hook/controller.go @@ -12,9 +12,10 @@ // See the License for the specific language governing permissions and // limitations under the License. -// Package hook holds the consumer side of the hooks framework: the controller -// that turns hook events on a queue into hook.Hook calls, and the reconciler for -// the events that never made it. +// Package hook holds the domain-neutral mechanics of the hooks framework: the +// controller that turns hook events on a queue into hook.Hook calls, the +// reconciler for the events that never made it, and the helper a producer +// publishes an event through. // // The controller is domain-neutral. Each domain runs its own hook topic and its // own instance of this stage — "per-domain" is about the topic and the wiring, diff --git a/platform/hook/publisher.go b/platform/hook/publisher.go new file mode 100644 index 000000000..07dc3b4bd --- /dev/null +++ b/platform/hook/publisher.go @@ -0,0 +1,58 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package hook + +import ( + "context" + "fmt" + + basehook "github.com/uber/submitqueue/api/base/hook" + "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/platform/publish" +) + +// Publish sends one hook event to the domain's hook topic, partitioned by +// partitionKey. The topic key is not a parameter: a domain runs a single hook +// topic, and the caller's registry is what binds that key to a wire topic. +// +// The event id is the message id, so a redelivery republishing the same event +// dedups into the original message instead of enqueuing a second one. Callers +// pass the partition key their own topic partitions on, carrying that ordering +// across the seam. +// +// Errors are returned unclassified, leaving retryability to the caller's +// classifier: a malformed event is the caller's bug, not a transient fault. +func Publish( + ctx context.Context, + registry consumer.TopicRegistry, + event *basehook.HookEvent, + partitionKey string, +) error { + if err := basehook.Validate(event); err != nil { + return fmt.Errorf("refusing to publish a malformed hook event: %w", err) + } + + body, err := basehook.Marshal(event) + if err != nil { + return fmt.Errorf("failed to serialize hook event %s: %w", event.GetId(), err) + } + + if err := publish.Message( + ctx, registry, basehook.TopicKeyHook, publish.IntentID(event.GetId()), body, partitionKey, + ); err != nil { + return fmt.Errorf("failed to publish hook event %s: %w", event.GetId(), err) + } + return nil +} diff --git a/platform/hook/publisher_test.go b/platform/hook/publisher_test.go new file mode 100644 index 000000000..c9db7b0ef --- /dev/null +++ b/platform/hook/publisher_test.go @@ -0,0 +1,125 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package hook + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + basehook "github.com/uber/submitqueue/api/base/hook" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + mqmock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" + "go.uber.org/mock/gomock" +) + +const ( + testEventID = "stovepipe/validation.repository.recorded/request/7/0" + testPartitionKey = "request/7" +) + +func testEvent() *basehook.HookEvent { + return &basehook.HookEvent{ + Id: testEventID, + Source: "stovepipe", + Type: "validation.repository.recorded", + TimestampMs: 1756327200000, + } +} + +// registryWithHookTopic returns a registry whose hook topic captures whatever is +// published to it, and the slot the captured message lands in. +func registryWithHookTopic(t *testing.T, ctrl *gomock.Controller, publishErr error) (consumer.TopicRegistry, *entityqueue.Message) { + t.Helper() + + var published entityqueue.Message + publisher := mqmock.NewMockPublisher(ctrl) + publisher.EXPECT().Publish(gomock.Any(), "domain-hook", gomock.Any()). + DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message) error { + published = msg + return publishErr + }).AnyTimes() + + queue := mqmock.NewMockQueue(ctrl) + queue.EXPECT().Publisher().Return(publisher).AnyTimes() + + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ + {Key: basehook.TopicKeyHook, Name: "domain-hook", Queue: queue}, + }) + require.NoError(t, err) + return registry, &published +} + +func TestPublish(t *testing.T) { + ctrl := gomock.NewController(t) + registry, published := registryWithHookTopic(t, ctrl, nil) + + require.NoError(t, Publish(context.Background(), registry, testEvent(), testPartitionKey)) + + // The event id is the message id, so a redelivery republishing the same + // event dedups into the original message. + assert.Equal(t, testEventID, published.ID) + assert.Equal(t, testPartitionKey, published.PartitionKey) + + decoded := &basehook.HookEvent{} + require.NoError(t, basehook.Unmarshal(published.Payload, decoded)) + assert.Equal(t, testEventID, decoded.GetId()) + assert.Equal(t, "validation.repository.recorded", decoded.GetType()) +} + +func TestPublish_RejectsMalformedEvent(t *testing.T) { + tests := []struct { + name string + event *basehook.HookEvent + }{ + {name: "nil", event: nil}, + {name: "no id", event: &basehook.HookEvent{Source: "stovepipe", Type: "t"}}, + {name: "no source", event: &basehook.HookEvent{Id: testEventID, Type: "t"}}, + {name: "no type", event: &basehook.HookEvent{Id: testEventID, Source: "stovepipe"}}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + publisher := mqmock.NewMockPublisher(ctrl) + queue := mqmock.NewMockQueue(ctrl) + queue.EXPECT().Publisher().Return(publisher).AnyTimes() + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ + {Key: basehook.TopicKeyHook, Name: "domain-hook", Queue: queue}, + }) + require.NoError(t, err) + + // No Publish expectation: a malformed event must not reach the queue. + require.Error(t, Publish(context.Background(), registry, tt.event, testPartitionKey)) + }) + } +} + +func TestPublish_PropagatesPublishFailure(t *testing.T) { + ctrl := gomock.NewController(t) + registry, _ := registryWithHookTopic(t, ctrl, errors.New("boom")) + + require.Error(t, Publish(context.Background(), registry, testEvent(), testPartitionKey)) +} + +func TestPublish_FailsWhenHookTopicIsUnregistered(t *testing.T) { + registry, err := consumer.NewTopicRegistry(nil) + require.NoError(t, err) + + require.Error(t, Publish(context.Background(), registry, testEvent(), testPartitionKey)) +} diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index ba33fabf9..d56e0b5ba 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -447,7 +447,7 @@ func registerPrimaryControllers( } count++ - recordController := record.NewController(logger, scope, store, sourceControl, stovepipemq.TopicKeyRecord, "stovepipe-record") + recordController := record.NewController(logger, scope, store, sourceControl, registry, stovepipemq.TopicKeyRecord, "stovepipe-record") if err := c.Register(recordController); err != nil { return count, fmt.Errorf("failed to register record controller: %w", err) } @@ -492,7 +492,7 @@ func registerDLQControllers( } count++ - recordDLQController := record.NewController(logger, scope, store, sourceControl, dlq.TopicKey(stovepipemq.TopicKeyRecord), "stovepipe-record-dlq") + recordDLQController := record.NewController(logger, scope, store, sourceControl, registry, dlq.TopicKey(stovepipemq.TopicKeyRecord), "stovepipe-record-dlq") if err := c.Register(recordDLQController); err != nil { return count, fmt.Errorf("failed to register record dlq controller: %w", err) } diff --git a/stovepipe/controller/record/BUILD.bazel b/stovepipe/controller/record/BUILD.bazel index 03b8cbc07..378ebfe32 100644 --- a/stovepipe/controller/record/BUILD.bazel +++ b/stovepipe/controller/record/BUILD.bazel @@ -6,8 +6,11 @@ go_library( importpath = "github.com/uber/submitqueue/stovepipe/controller/record", visibility = ["//visibility:public"], deps = [ + "//api/base/hook:go_default_library", "//platform/consumer:go_default_library", + "//platform/hook:go_default_library", "//platform/metrics: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", @@ -23,10 +26,13 @@ go_test( srcs = ["record_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/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/sourcecontrol:go_default_library", diff --git a/stovepipe/controller/record/record.go b/stovepipe/controller/record/record.go index f41b49afe..93366aec7 100644 --- a/stovepipe/controller/record/record.go +++ b/stovepipe/controller/record/record.go @@ -19,8 +19,12 @@ // The durable state is a ValidationFact per validated commit, plus the queue's // last-green bookmark, which process reads to choose an incremental build // baseline. A green commit is also promoted, moving the queue's promotion ref so -// downstream systems can pull the latest green commit by name. Downstream hooks -// are not implemented yet. +// downstream systems can pull the latest green commit by name. +// +// The outcome is then announced as a hook event, so integrations outside the +// pipeline learn that a commit is green or broken. See +// doc/rfc/stovepipe/steps/record.md for the event shape and +// doc/rfc/hook-framework.md for the delivery promise. package record import ( @@ -30,8 +34,11 @@ import ( "time" "github.com/uber-go/tally" + basehook "github.com/uber/submitqueue/api/base/hook" "github.com/uber/submitqueue/platform/consumer" + platformhook "github.com/uber/submitqueue/platform/hook" "github.com/uber/submitqueue/platform/metrics" + "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" @@ -48,6 +55,7 @@ type Controller struct { metricsScope tally.Scope stores storage.Factory sourceControl sourcecontrol.Factory + registry consumer.TopicRegistry topicKey consumer.TopicKey consumerGroup string } @@ -69,6 +77,7 @@ func NewController( scope tally.Scope, stores storage.Factory, sourceControl sourcecontrol.Factory, + registry consumer.TopicRegistry, topicKey consumer.TopicKey, consumerGroup string, ) *Controller { @@ -78,15 +87,16 @@ func NewController( metricsScope: scope.SubScope(name), stores: stores, sourceControl: sourceControl, + registry: registry, topicKey: topicKey, consumerGroup: consumerGroup, } } // Process loads the request referenced by the delivery, records its validation -// fact and, when that fact is green, advances the queue's last-green bookmark and -// promotes the commit. Returns nil to ack (success) or an error to nack (retry) / -// reject (DLQ). +// fact, applies that fact to the caches derived from it, and announces the +// outcome as a hook event. Returns nil to ack (success) or an error to nack +// (retry) / reject (DLQ). // // buildsignal stamps the outcome on the request before publishing here, so a // request without a build outcome is a producer invariant violation rather @@ -126,34 +136,16 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er if err != nil { return err } - if !fact.IsGreen() { - metrics.NamedCounter(c.metricsScope, _opName, "not_green", 1, metrics.TagsFromContext(ctx)...) - // Only the writer of the fact reports the latency: a redelivery adopts - // the stored fact instead, and a second sample would count one break - // twice in the distribution. - if created { - c.reportFailureDetectionLatency(ctx, request) - } - return nil - } - holdsBookmark, err := c.advanceLastGreen(ctx, store, request) - if err != nil { - metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1, metrics.TagsFromContext(ctx)...) + if err := c.applyFactToDerivedCaches(ctx, store, request, fact, created); err != nil { return err } - if !holdsBookmark { - // A later green commit already holds the bookmark, so it also owns - // the promotion ref: promoting this older commit would move the ref - // backwards. - return nil - } - return c.promote(ctx, request) + return c.publishHookEvent(ctx, request, hookevent.NewValidationRepositoryRecorded(request)) case entity.RequestStateCancelled: // A cancelled build decided nothing about the commit, so it establishes // no fact. The identity stays unclaimed; the next commit re-validates. metrics.NamedCounter(c.metricsScope, _opName, "cancelled", 1, metrics.TagsFromContext(ctx)...) - return nil + return c.publishHookEvent(ctx, request, hookevent.NewValidationRepositoryCancelled(request)) case entity.RequestStateSuperseded: // Terminal without a build outcome. buildsignal never publishes for a @@ -170,6 +162,41 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er } } +// applyFactToDerivedCaches moves the two caches that follow the persisted fact: a +// green fact advances the queue's bookmark and, when this request ends up holding +// it, promotes the commit. A broken fact moves neither, and instead reports how +// long the break it names went undetected. +func (c *Controller) applyFactToDerivedCaches( + ctx context.Context, + store storage.Storage, + request entity.Request, + fact entity.ValidationFact, + createdFact bool, +) error { + if !fact.IsGreen() { + metrics.NamedCounter(c.metricsScope, _opName, "not_green", 1, metrics.TagsFromContext(ctx)...) + // Only the writer of the fact reports the latency: a redelivery adopts + // the stored fact instead, and a second sample would count one break + // twice in the distribution. + if createdFact { + c.reportFailureDetectionLatency(ctx, request) + } + return nil + } + + holdsBookmark, err := c.advanceLastGreen(ctx, store, request) + if err != nil { + metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1, metrics.TagsFromContext(ctx)...) + return err + } + if !holdsBookmark { + // A later green commit already holds the bookmark, so it also owns the + // promotion ref: promoting this older commit would move the ref backwards. + return nil + } + return c.promote(ctx, request) +} + // recordFact writes the request's outcome as an immutable whole-repository fact and // returns the fact that is actually stored, which is not always the one just built: // facts are first-writer-wins, so an identity already claimed by this same request — @@ -452,6 +479,34 @@ func (c *Controller) promote(ctx context.Context, request entity.Request) error return nil } +// publishHookEvent sends one event about request to the domain's hook topic. +// +// Called last, after the fact write and after both caches derived from it have +// moved: the payload names the request rather than snapshotting it, so a hook +// reacting to a green commit by reading the bookmark or resolving the promotion +// ref must not find either still pointing at the previous commit. +// +// Partitioning by request id matches the record 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.Infow("announced validation outcome", + "queue", request.Queue, + "request_id", request.ID, + "uri", request.URI, + "event_type", event.GetType(), + "event_id", event.GetId(), + ) + return nil +} + // compareToBookmark orders candidate against the request id currently holding the // bookmark, by ingest order, using the sign convention of entity.CompareRequestID. // An empty current means the bookmark has never been set, so any candidate is newer. diff --git a/stovepipe/controller/record/record_test.go b/stovepipe/controller/record/record_test.go index bcbbee602..e4eee5b87 100644 --- a/stovepipe/controller/record/record_test.go +++ b/stovepipe/controller/record/record_test.go @@ -23,10 +23,13 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/uber-go/tally" + basehook "github.com/uber/submitqueue/api/base/hook" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" consumermock "github.com/uber/submitqueue/platform/consumer/mock" + mqmock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/stovepipe/core/hookevent" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol" @@ -66,9 +69,33 @@ type recordMocks struct { queueStore *storagemock.MockQueueStore factStore *storagemock.MockValidationFactStore sourceControl *sourcecontrolmock.MockSourceControl + hooks *hookRecorder metricsScope tally.TestScope } +// hookRecorder stands in for the hook topic, decoding whatever the controller +// publishes so a case can assert on the announcement rather than on the queue +// plumbing carrying it. Setting err makes the publish fail. +type hookRecorder struct { + events []*basehook.HookEvent + err error +} + +// only returns the single event the controller published, failing the case when +// it published any other number. +func (r *hookRecorder) only(t *testing.T) *basehook.HookEvent { + t.Helper() + require.Len(t, r.events, 1) + return r.events[0] +} + +// payload returns the event's payload as a plain map. +func payload(t *testing.T, event *basehook.HookEvent) map[string]any { + t.Helper() + require.NotNil(t, event.GetPayload()) + return event.GetPayload().AsMap() +} + // expectFactCreated wires a successful fact write and captures it, so a case can // assert on the recorded degree without pinning the wall-clock CreatedAt. func (m recordMocks) expectFactCreated(captured *entity.ValidationFact) { @@ -113,6 +140,7 @@ func newControllerForTopic(t *testing.T, ctrl *gomock.Controller, topicKey consu queueStore: storagemock.NewMockQueueStore(ctrl), factStore: storagemock.NewMockValidationFactStore(ctrl), sourceControl: sourcecontrolmock.NewMockSourceControl(ctrl), + hooks: &hookRecorder{}, metricsScope: scope, } @@ -121,11 +149,35 @@ func newControllerForTopic(t *testing.T, ctrl *gomock.Controller, topicKey consu store.EXPECT().GetQueueStore().Return(m.queueStore).AnyTimes() store.EXPECT().GetValidationFactStore().Return(m.factStore).AnyTimes() + publisher := mqmock.NewMockPublisher(ctrl) + publisher.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()). + DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message) error { + if m.hooks.err != nil { + return m.hooks.err + } + event := &basehook.HookEvent{} + if err := basehook.Unmarshal(msg.Payload, event); err != nil { + return err + } + m.hooks.events = append(m.hooks.events, event) + return nil + }).AnyTimes() + + queue := mqmock.NewMockQueue(ctrl) + queue.EXPECT().Publisher().Return(publisher).AnyTimes() + + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ + {Key: stovepipemq.TopicKeyRecord, Name: "record", Queue: queue}, + {Key: basehook.TopicKeyHook, Name: "stovepipe-hook", Queue: queue}, + }) + require.NoError(t, err) + c := NewController( zap.NewNop().Sugar(), scope, staticStorageFactory{store: store}, staticSourceControlFactory{sourceControl: m.sourceControl}, + registry, topicKey, consumerGroup, ) @@ -623,11 +675,19 @@ func TestProcess_PromotionErrorsPropagate(t *testing.T) { func TestProcess_TerminalWithoutFactDoesNotTouchStores(t *testing.T) { tests := []struct { - name string - state entity.RequestState + name string + state entity.RequestState + wantEventTypes []string }{ - {name: "cancelled", state: entity.RequestStateCancelled}, - {name: "superseded", state: entity.RequestStateSuperseded}, + { + name: "cancelled announces that no verdict is coming", + state: entity.RequestStateCancelled, + wantEventTypes: []string{string(hookevent.TypeValidationRepositoryCancelled)}, + }, + { + name: "superseded announces nothing", + state: entity.RequestStateSuperseded, + }, } for _, tt := range tests { @@ -639,10 +699,147 @@ func TestProcess_TerminalWithoutFactDoesNotTouchStores(t *testing.T) { // Neither a fact nor a queue write: these outcomes decide nothing. require.NoError(t, c.Process(queueContext(), delivery(t, ctrl, recordPayload(t, testID)))) + + var got []string + for _, event := range m.hooks.events { + got = append(got, event.GetType()) + } + assert.Equal(t, tt.wantEventTypes, got) }) } } +func TestProcess_AnnouncesRecordedValidation(t *testing.T) { + tests := []struct { + name string + state entity.RequestState + wantDegree float64 + }{ + {name: "green", state: entity.RequestStateSucceeded, wantDegree: entity.DegreeGreen}, + {name: "broken", state: entity.RequestStateFailed, wantDegree: entity.DegreeBroken}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + + request := requestWithState(tt.state) + request.BaseURI = testBaseURI + request.BuildStrategy = entity.BuildStrategyIncrementalSinceGreen + + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(request, nil) + var fact entity.ValidationFact + m.expectFactCreated(&fact) + + if tt.wantDegree == entity.DegreeGreen { + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(queueRow("", "", 1), nil) + m.queueStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(nil) + m.sourceControl.EXPECT().ChangeInfo(gomock.Any(), testURI). + Return(sourcecontrol.ChangeInfo{CreatedAt: testChangeTime.UnixMilli()}, nil) + m.sourceControl.EXPECT().Promote(gomock.Any(), testURI).Return(nil) + } else { + m.sourceControl.EXPECT().ChangeInfo(gomock.Any(), testBaseURI). + Return(sourcecontrol.ChangeInfo{CreatedAt: testChangeTime.UnixMilli()}, nil) + } + + require.NoError(t, c.Process(queueContext(), delivery(t, ctrl, recordPayload(t, testID)))) + + event := m.hooks.only(t) + assert.Equal(t, hookevent.Source, event.GetSource()) + assert.Equal(t, string(hookevent.TypeValidationRepositoryRecorded), event.GetType()) + assert.Positive(t, event.GetTimestampMs()) + + assert.Equal(t, map[string]any{"queue": testQueue, "request_id": testID}, payload(t, event)) + // The degree the event does not carry is on the fact a hook reads instead. + assert.Equal(t, tt.wantDegree, fact.Degree) + }) + } +} + +func TestProcess_AnnouncesCancelledValidation(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + + m.reqStore.EXPECT().Get(gomock.Any(), testID). + Return(requestWithState(entity.RequestStateCancelled), nil) + + require.NoError(t, c.Process(queueContext(), delivery(t, ctrl, recordPayload(t, testID)))) + + event := m.hooks.only(t) + assert.Equal(t, string(hookevent.TypeValidationRepositoryCancelled), event.GetType()) + assert.Equal(t, map[string]any{"queue": testQueue, "request_id": testID}, payload(t, event)) +} + +// The announcement is the last thing the stage does, so a hook reacting to a green +// commit by reading the bookmark or resolving the promotion ref cannot find either +// still pointing at the commit this event supersedes. +func TestProcess_AnnouncesOnlyAfterDerivedCachesMove(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + + m.reqStore.EXPECT().Get(gomock.Any(), testID). + Return(requestWithState(entity.RequestStateSucceeded), nil) + var fact entity.ValidationFact + m.expectFactCreated(&fact) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(queueRow("", "", 1), nil) + m.queueStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)). + DoAndReturn(func(context.Context, entity.Queue, int32, int32) error { + assert.Empty(t, m.hooks.events, "the bookmark must advance before the announcement") + return nil + }) + m.sourceControl.EXPECT().ChangeInfo(gomock.Any(), testURI). + Return(sourcecontrol.ChangeInfo{CreatedAt: testChangeTime.UnixMilli()}, nil) + m.sourceControl.EXPECT().Promote(gomock.Any(), testURI). + DoAndReturn(func(context.Context, string) error { + assert.Empty(t, m.hooks.events, "the promotion must land before the announcement") + return nil + }) + + require.NoError(t, c.Process(queueContext(), delivery(t, ctrl, recordPayload(t, testID)))) + assert.Len(t, m.hooks.events, 1) +} + +// The id is derived from the transition rather than the clock, which is what lets +// the queue dedupe a redelivery and a hook stay idempotent without an outbox. +func TestProcess_RedeliveryAnnouncesTheSameEventID(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(failedRequest(), nil).Times(2) + var fact entity.ValidationFact + m.expectFactCreated(&fact) + m.sourceControl.EXPECT().ChangeInfo(gomock.Any(), testBaseURI). + Return(sourcecontrol.ChangeInfo{CreatedAt: testChangeTime.UnixMilli()}, nil) + + m.factStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(storage.ErrAlreadyExists) + m.factStore.EXPECT().Get(gomock.Any(), testURI, wholeRepositoryProject). + Return(entity.ValidationFact{URI: testURI, Degree: entity.DegreeBroken, RequestID: testID}, nil) + + require.NoError(t, c.Process(queueContext(), delivery(t, ctrl, recordPayload(t, testID)))) + require.NoError(t, c.Process(queueContext(), delivery(t, ctrl, recordPayload(t, testID)))) + + require.Len(t, m.hooks.events, 2) + assert.Equal(t, m.hooks.events[0].GetId(), m.hooks.events[1].GetId()) +} + +// A failed announcement fails the delivery: the fact is already durable, so the +// retry re-runs an idempotent chain, and dead-lettering keeps the loss visible +// rather than acking an outcome nothing downstream heard about. +func TestProcess_AnnouncementFailureFailsRecord(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + m.hooks.err = errors.New("boom") + + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(failedRequest(), nil) + var fact entity.ValidationFact + m.expectFactCreated(&fact) + m.sourceControl.EXPECT().ChangeInfo(gomock.Any(), testBaseURI). + Return(sourcecontrol.ChangeInfo{CreatedAt: testChangeTime.UnixMilli()}, nil) + + require.Error(t, c.Process(queueContext(), delivery(t, ctrl, recordPayload(t, testID)))) +} + func TestProcess_NonTerminalRequestFails(t *testing.T) { tests := []struct { name string diff --git a/stovepipe/core/hookevent/BUILD.bazel b/stovepipe/core/hookevent/BUILD.bazel new file mode 100644 index 000000000..6ebdc813b --- /dev/null +++ b/stovepipe/core/hookevent/BUILD.bazel @@ -0,0 +1,25 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["hookevent.go"], + importpath = "github.com/uber/submitqueue/stovepipe/core/hookevent", + visibility = ["//visibility:public"], + deps = [ + "//api/base/hook:go_default_library", + "//stovepipe/entity:go_default_library", + "@org_golang_google_protobuf//types/known/structpb:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["hookevent_test.go"], + embed = [":go_default_library"], + deps = [ + "//api/base/hook:go_default_library", + "//stovepipe/entity:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + ], +) diff --git a/stovepipe/core/hookevent/hookevent.go b/stovepipe/core/hookevent/hookevent.go new file mode 100644 index 000000000..e5fa5baf9 --- /dev/null +++ b/stovepipe/core/hookevent/hookevent.go @@ -0,0 +1,83 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package hookevent holds stovepipe's hook events: the source it reports, the +// types it publishes, and a constructor per type. Centralizing them keeps the +// wire contract in one place, so a controller names an event rather than +// assembling one. +// +// Every payload carries identity only — the queue and the request id — and a +// hook resolves what it needs from storage. A richer payload would be a +// snapshot taken at publish time and read later, so it could describe state +// that has since moved on; identity cannot go stale. The cost is a store read +// per hook. See doc/rfc/stovepipe/steps/record.md. +package hookevent + +import ( + "time" + + basehook "github.com/uber/submitqueue/api/base/hook" + "github.com/uber/submitqueue/stovepipe/entity" + "google.golang.org/protobuf/types/known/structpb" +) + +// Source is reported on every hook event stovepipe publishes. A consumer +// serving several domains matches on it to recognize stovepipe's events. +const Source = "stovepipe" + +// Type is the event type a consumer filters on. +type Type string + +const ( + // TypeValidationRepositoryRecorded announces a durable whole-repository + // validation fact for the request named in the payload. + TypeValidationRepositoryRecorded Type = "validation.repository.recorded" + // TypeValidationRepositoryCancelled announces a validation that ended + // without establishing a fact, so a consumer can stop waiting on the commit. + TypeValidationRepositoryCancelled Type = "validation.repository.cancelled" +) + +// unversioned is the version reported on these events. Neither describes a +// versioned write, so nothing separates one occurrence of a type from the next +// beyond the request it names. +const unversioned int32 = 0 + +// NewValidationRepositoryRecorded builds the event announcing that request's +// validation fact is durable. +func NewValidationRepositoryRecorded(request entity.Request) *basehook.HookEvent { + return newEvent(TypeValidationRepositoryRecorded, request) +} + +// NewValidationRepositoryCancelled builds the event announcing that request's +// validation ended without establishing a fact. +func NewValidationRepositoryCancelled(request entity.Request) *basehook.HookEvent { + return newEvent(TypeValidationRepositoryCancelled, request) +} + +// newEvent is the shared body of the constructors above. The payload keys are +// the wire field names a consumer in another repository mirrors, and this is +// the only place they are written. +func newEvent(eventType Type, request entity.Request) *basehook.HookEvent { + return &basehook.HookEvent{ + Id: basehook.NewEventID(Source, string(eventType), request.ID, unversioned), + Source: Source, + Type: string(eventType), + TimestampMs: time.Now().UnixMilli(), + Version: unversioned, + Payload: &structpb.Struct{Fields: map[string]*structpb.Value{ + "queue": structpb.NewStringValue(request.Queue), + "request_id": structpb.NewStringValue(request.ID), + }}, + } +} diff --git a/stovepipe/core/hookevent/hookevent_test.go b/stovepipe/core/hookevent/hookevent_test.go new file mode 100644 index 000000000..22a1ae2bd --- /dev/null +++ b/stovepipe/core/hookevent/hookevent_test.go @@ -0,0 +1,92 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package hookevent + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + basehook "github.com/uber/submitqueue/api/base/hook" + "github.com/uber/submitqueue/stovepipe/entity" +) + +const ( + testQueue = "monorepo/main" + testID = "request/monorepo/main/7" +) + +func testRequest() entity.Request { + return entity.Request{ + ID: testID, + Queue: testQueue, + URI: "git://remote/monorepo/main/abc123", + BaseURI: "git://remote/monorepo/main/def456", + } +} + +func TestConstructors(t *testing.T) { + tests := []struct { + name string + build func(entity.Request) *basehook.HookEvent + wantType Type + }{ + { + name: "recorded", + build: NewValidationRepositoryRecorded, + wantType: TypeValidationRepositoryRecorded, + }, + { + name: "cancelled", + build: NewValidationRepositoryCancelled, + wantType: TypeValidationRepositoryCancelled, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + event := tt.build(testRequest()) + require.NoError(t, basehook.Validate(event)) + + assert.Equal(t, Source, event.GetSource()) + assert.Equal(t, string(tt.wantType), event.GetType()) + assert.Positive(t, event.GetTimestampMs()) + + // The wire field names a consumer in another repository reads. The + // request's uri and base uri are deliberately absent: a hook + // resolves those from storage. + assert.Equal(t, map[string]any{ + "queue": testQueue, + "request_id": testID, + }, event.GetPayload().AsMap()) + }) + } +} + +// The id is derived from the transition rather than the clock, which is what +// lets the queue dedupe a redelivery. +func TestNewValidationRepositoryRecorded_IDIsStable(t *testing.T) { + first := NewValidationRepositoryRecorded(testRequest()) + second := NewValidationRepositoryRecorded(testRequest()) + + assert.Equal(t, first.GetId(), second.GetId()) +} + +func TestConstructors_IDsDifferByType(t *testing.T) { + recorded := NewValidationRepositoryRecorded(testRequest()) + cancelled := NewValidationRepositoryCancelled(testRequest()) + + assert.NotEqual(t, recorded.GetId(), cancelled.GetId()) +}