Skip to content
Draft
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
2 changes: 1 addition & 1 deletion Makefile
Original file line number Diff line number Diff line change
Expand Up @@ -579,7 +579,7 @@ local-stovepipe-stop: ## Stop the Stovepipe service

mocks: ## Generate mock files using mockgen
@echo "Generating mocks..."
@$(BAZEL) run @rules_go//go -- generate ./submitqueue/extension/storage/... ./submitqueue/extension/buildrunner/... ./submitqueue/extension/changeprovider/... ./platform/extension/counter/... ./platform/extension/consumergate/... ./platform/extension/messagequeue/... ./submitqueue/extension/queueconfig/... ./submitqueue/extension/mergechecker/... ./submitqueue/extension/scorer/... ./submitqueue/extension/conflict/... ./submitqueue/extension/speculation/... ./submitqueue/extension/validator/... ./platform/consumer/... ./stovepipe/extension/storage/... ./stovepipe/extension/sourcecontrol/...
@$(BAZEL) run @rules_go//go -- generate ./submitqueue/extension/storage/... ./submitqueue/extension/buildrunner/... ./submitqueue/extension/changeprovider/... ./platform/extension/counter/... ./platform/extension/consumergate/... ./platform/extension/hook/... ./platform/extension/messagequeue/... ./submitqueue/extension/queueconfig/... ./submitqueue/extension/mergechecker/... ./submitqueue/extension/scorer/... ./submitqueue/extension/conflict/... ./submitqueue/extension/speculation/... ./submitqueue/extension/validator/... ./platform/consumer/... ./stovepipe/extension/storage/... ./stovepipe/extension/sourcecontrol/...
@echo "Mocks generated successfully!"

proto: ## Generate protobuf files from .proto definitions
Expand Down
4 changes: 2 additions & 2 deletions api/base/hook/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -40,9 +40,9 @@ The binding between a topic key and its payload lives in the message's `topic_ke

| Message | Direction | Topic key |
|---|---|---|
| `HookEvent` | producing domain → hook dispatcher | `hook` |
| `HookEvent` | producing domain → hook stage | `hook` |

The key is per-host: each domain runs its own hook topic and its own dispatcher, so two domains sharing one queue backend must map `hook` to distinct topic names.
The key is per-host: each domain runs its own hook topic and its own hook controller, so two domains sharing one queue backend must map `hook` to distinct topic names.

## Evolution

Expand Down
19 changes: 10 additions & 9 deletions doc/rfc/hook-framework.md
Original file line number Diff line number Diff line change
Expand Up @@ -10,13 +10,13 @@ Two requirements: side effects must never stall or fail the pipeline, and "fire

## Proposal

When a controller performs a transition, it also publishes a **hook event** to a durable `hook` topic. A thin per-domain dispatcher stage consumes it and hands each event to the **hooks** the host wired — no-op by default, real integrations as they arrive.
When a controller performs a transition, it also publishes a **hook event** to a durable `hook` topic. A thin per-domain hook stage consumes it, asks the host's **hooks resolver** which integrations that event belongs to, and runs them — none by default, real integrations as they arrive.

```
pipeline controller dispatcher stage (per domain)
state write → hook publish → downstream ──▶ [hook topic] ──▶ decode → validate → hook.Handle
├─ noop (default)
│ retries exhausted └─ composite ─▶ warehouse, code host, …
pipeline controller hook stage (per domain)
state write → hook publish → downstream ──▶ [hook topic] ──▶ decode → validate → Hooks.For(event)
└─▶ warehouse, code host, …
│ retries exhausted
[hook_dlq] ──▶ log full event + page; manual republish
```
Expand Down Expand Up @@ -58,12 +58,13 @@ Delivery promise:

### Hooks and dispatch

- Extension at `platform/extension/hook/`, singleton shape (counter precedent), wired once per host; no per-queue factory.
- Extension at `platform/extension/hook/`: the `Hook` contract plus a `Hooks` resolver the host builds in wiring. No `Config` and no `Factory` — selection is the resolver's, and only wiring knows the queue topology.
- Hook contract: at-least-once, idempotent by `id`, plain errors, never writes pipeline state; ignore an event by returning nil (no filter API).
- Ships `noop` (default) and `composite` (runs all children, joins failures, names failing children). A cross-domain sink is the same impl wired into each domain.
- Dispatcher: decode, validate (`id`/`source`/`type` non-empty), invoke. Malformed events dead-letter, never silently acked; hook errors retry then dead-letter, with errs classifiers fast-pathing permanent failures.
- `Hooks.For(event)` keys on the event, not a queue name: the envelope carries no queue, and which scope selects hooks (queue, source, type) differs per domain. Resolving to none is ordinary. Ships `noop` for a host that wants an explicit placeholder. A cross-domain sink is the same impl wired into each domain.
- Controller (`platform/hook`, wired by each service): decode, validate (`id`/`source`/`type` non-empty), resolve, invoke all. Malformed events dead-letter, never silently acked; hook errors retry then dead-letter, with errs classifiers fast-pathing permanent failures.
- Mixed outcomes: every resolved hook runs even after one fails, and the failures are attributed and joined. `errs` weighs each branch of a joined error, so a transient failure alongside a permanent one still retries.
- DLQ reconciler: log the full event with its failure attribution, page (new metric — the log DLQ only warns), then ack. Manual republish recovers; pipeline state is never touched.
- Per-hook retry isolation later: consumer groups on the same `hook` topic key, once the registry supports multiple groups per key and rejection becomes group-local (today it moves the shared row). Until then the composite's shared budget is accepted.
- Per-hook retry isolation later: consumer groups on the same `hook` topic key, once the registry supports multiple groups per key and rejection becomes group-local (today it moves the shared row). Until then one shared budget for all of an event's hooks is accepted.

## Example

Expand Down
9 changes: 9 additions & 0 deletions platform/extension/hook/BUILD.bazel
Original file line number Diff line number Diff line change
@@ -0,0 +1,9 @@
load("@rules_go//go:def.bzl", "go_library")

go_library(
name = "go_default_library",
srcs = ["hook.go"],
importpath = "github.com/uber/submitqueue/platform/extension/hook",
visibility = ["//visibility:public"],
deps = ["//api/base/hook:go_default_library"],
)
43 changes: 43 additions & 0 deletions platform/extension/hook/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,43 @@
# Hook

Vendor-agnostic interface for fire-and-forget side effects run in response to pipeline lifecycle events: warehouse exports, code-host comments, notifications, audit trails. See [the hooks framework RFC](../../../doc/rfc/hook-framework.md) for the design and [`api/base/hook`](../../../api/base/hook) for the event contract.

## Interface

### Hook

Handles one lifecycle event. `Name` identifies it in logs, metrics, and failure attribution.

Four obligations, all of them consequences of running behind an at-least-once queue:

- **Idempotent on the event id.** The same event may arrive more than once, including after a successful `Handle`. The id is derived from the transition, so a redelivery carries the id the first delivery did.
- **Return nil to ignore an event.** There is no filter or subscription API. A hook that does not care about a type returns nil and costs nothing; routing can become a wiring decorator if it ever pays for itself.
- **Return plain errors.** Classification is the consumer's job. An error must mean the side effect did not happen — reporting failure for work that succeeded turns at-least-once delivery into repeated duplicate effects.
- **Never write pipeline state.** A hook's outcome is invisible to the pipeline, which is exactly what makes it unable to affect the transition that triggered it.

### Hooks

Resolves the hooks that run for an event. The controller in [`platform/hook`](../../hook) asks it once per delivery and runs everything it returns; returning none is ordinary and means nothing this host wired cares about the event.

`For` takes the event rather than a queue name because the envelope carries no queue. Which scope selects hooks differs per domain — queue, source, event type — and only the host that publishes the payload can read a queue out of it, so the choice belongs to the resolver. Resolution runs on every delivery and cannot fail: an integration that cannot be reached is a `Handle` error, not an absent hook.

## Wiring

There is no `Config` and no `Factory` here. Selection is the resolver's job, and the resolver is built in the wiring layer — the only place that knows the full set of queues and the integrations wired for each. The host constructs its `Hooks` and hands it to the controller in [`platform/hook`](../../hook), which owns the consumer side: decode, validate, resolve, invoke.

Two queues in one host can point at different providers and want different integrations, which is why hooks are resolved per event rather than fixed per deployment.

## Implementations

- **`noop/`** — accepts every event and does nothing. A placeholder for a host that wants the stage registered before it has any integration; a resolver that returns no hooks does the same thing.

A sink that serves several domains is one implementation wired into each domain's host, not one implementation per domain.

## Implementing a Hook

1. Create `platform/extension/hook/{name}/` for a hook reusable across domains, or `{domain}/extension/hook/{name}/` for one that is domain-specific.
2. Implement `Handle` and `Name`, keying any deduplication on `event.GetId()`.
3. Decide per event `type` what to do, and return nil for the types you ignore.
4. Return it from the host's `Hooks` resolver for the events it should run on.

Every hook the resolver returns for an event shares one consumer and therefore one retry budget: one chronically failing integration eventually dead-letters events the others handled fine. See [`platform/hook`](../../hook) before wiring several.
80 changes: 80 additions & 0 deletions platform/extension/hook/hook.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,80 @@
// 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 defines the contract for a hook: a pluggable side effect run in
// response to a pipeline lifecycle event. Warehouse exports, code-host comments,
// notifications, and audit trails are all hooks.
//
// Which hooks run is a property of the event rather than of the deployment: two
// queues in one host can point at different providers and want different
// integrations. A host therefore supplies a Hooks resolver, and the controller
// in platform/hook asks it once per event.
//
// Hooks run behind a durable queue, never inline in the pipeline, so a slow or
// failing integration cannot stall or fail the work that triggered it.
package hook

//go:generate mockgen -source=hook.go -destination=mock/hook_mock.go -package=mock

import (
"context"

basehook "github.com/uber/submitqueue/api/base/hook"
)

// Hook performs a side effect in response to a lifecycle event.
type Hook interface {

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

my general question here is why is it an extension vs a plain consumer who gets hooks events from wire and implements consumer controller? why it needs to be an interface?

// Handle performs the side effect for event.
//
// Delivery is at-least-once, so the same event — identical id — may arrive
// more than once, including after a successful Handle. Implementations must
// be idempotent on the event id.
//
// Returning nil means "done with this event", which is also how a hook
// ignores one: there is no filter or subscription API, because a hook that
// does not care about a type simply returns nil, and routing can be added as
// a wiring decorator if it ever pays for itself.
//
// Returning an error retries the event and, past the retry budget,
// dead-letters it. Return plain errors; classification is the consumer's
// job. An error must mean the side effect did not happen — reporting failure
// for work that succeeded turns at-least-once into repeated duplicate
// effects.
//
// A hook must never write pipeline state. Its outcome is invisible to the
// pipeline by design: that is what makes the side effect unable to affect
// the transition that triggered it.
Handle(ctx context.Context, event *basehook.HookEvent) error

// Name identifies the hook in logs, metrics, and the failure attribution
// the controller reports. Stable and unique among the hooks a host wires.
Name() string
}

// Hooks resolves the hooks that run for an event.
type Hooks interface {
// For returns the hooks to run for event, in the order they should run.
// Returning none is an ordinary outcome: it means nothing this host wired
// is interested in the event.
//
// It takes the event rather than a queue name because the envelope carries
// no queue. Which scope selects hooks differs per domain — queue, source,
// event type — and only the host that publishes the payload can read a
// queue out of it, so the choice belongs to the resolver.
//
// Called on every delivery, so resolution must be cheap and must not fail:
// an integration that cannot be reached is a Handle error, not an absent
// hook.
For(event *basehook.HookEvent) []Hook
}
13 changes: 13 additions & 0 deletions platform/extension/hook/mock/BUILD.bazel
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
load("@rules_go//go:def.bzl", "go_library")

go_library(
name = "go_default_library",
srcs = ["hook_mock.go"],
importpath = "github.com/uber/submitqueue/platform/extension/hook/mock",
visibility = ["//visibility:public"],
deps = [
"//api/base/hook:go_default_library",
"//platform/extension/hook:go_default_library",
"@org_uber_go_mock//gomock:go_default_library",
],
)
109 changes: 109 additions & 0 deletions platform/extension/hook/mock/hook_mock.go

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

22 changes: 22 additions & 0 deletions platform/extension/hook/noop/BUILD.bazel
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
load("@rules_go//go:def.bzl", "go_library", "go_test")

go_library(
name = "go_default_library",
srcs = ["hook.go"],
importpath = "github.com/uber/submitqueue/platform/extension/hook/noop",
visibility = ["//visibility:public"],
deps = [
"//api/base/hook:go_default_library",
"//platform/extension/hook:go_default_library",
],
)

go_test(
name = "go_default_test",
srcs = ["hook_test.go"],
embed = [":go_default_library"],
deps = [
"//api/base/hook:go_default_library",
"@com_github_stretchr_testify//require:go_default_library",
],
)
44 changes: 44 additions & 0 deletions platform/extension/hook/noop/hook.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
// 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 noop provides a hook.Hook that accepts every event and does nothing.
// It is a placeholder for a host that wants the stage registered before it has
// any integration — a resolver returning no hooks does the same thing. Either
// way events are still published, consumed, and acked, so turning a real hook on
// later changes only what happens to the event, not whether the seam works.
package noop

import (
"context"

basehook "github.com/uber/submitqueue/api/base/hook"
"github.com/uber/submitqueue/platform/extension/hook"
)

// Verify interface compliance at compile time.
var _ hook.Hook = Hook{}

// Hook is a hook that discards every event.
type Hook struct{}

// New returns a no-op Hook.
func New() Hook {
return Hook{}
}

// Handle implements hook.Hook. The event is discarded.
func (Hook) Handle(context.Context, *basehook.HookEvent) error { return nil }

// Name implements hook.Hook.
func (Hook) Name() string { return "noop" }
Loading