From 63b4e01da19d6fac80f1a696db1b08ab4c5d16d9 Mon Sep 17 00:00:00 2001 From: "prath.shenoy" Date: Mon, 17 Aug 2026 16:32:34 +0000 Subject: [PATCH] feat(hook): Deliver events to integrations --- Makefile | 2 +- api/base/hook/README.md | 4 +- doc/rfc/hook-framework.md | 19 +- platform/extension/hook/BUILD.bazel | 9 + platform/extension/hook/README.md | 43 +++++ platform/extension/hook/hook.go | 80 ++++++++ platform/extension/hook/mock/BUILD.bazel | 13 ++ platform/extension/hook/mock/hook_mock.go | 109 +++++++++++ platform/extension/hook/noop/BUILD.bazel | 22 +++ platform/extension/hook/noop/hook.go | 44 +++++ platform/extension/hook/noop/hook_test.go | 37 ++++ platform/hook/BUILD.bazel | 41 +++++ platform/hook/README.md | 54 ++++++ platform/hook/controller.go | 153 ++++++++++++++++ platform/hook/controller_test.go | 214 ++++++++++++++++++++++ platform/hook/dlq.go | 130 +++++++++++++ platform/hook/dlq_test.go | 115 ++++++++++++ 17 files changed, 1077 insertions(+), 12 deletions(-) create mode 100644 platform/extension/hook/BUILD.bazel create mode 100644 platform/extension/hook/README.md create mode 100644 platform/extension/hook/hook.go create mode 100644 platform/extension/hook/mock/BUILD.bazel create mode 100644 platform/extension/hook/mock/hook_mock.go create mode 100644 platform/extension/hook/noop/BUILD.bazel create mode 100644 platform/extension/hook/noop/hook.go create mode 100644 platform/extension/hook/noop/hook_test.go create mode 100644 platform/hook/BUILD.bazel create mode 100644 platform/hook/README.md create mode 100644 platform/hook/controller.go create mode 100644 platform/hook/controller_test.go create mode 100644 platform/hook/dlq.go create mode 100644 platform/hook/dlq_test.go diff --git a/Makefile b/Makefile index f06d0aea6..3bca99adb 100644 --- a/Makefile +++ b/Makefile @@ -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 diff --git a/api/base/hook/README.md b/api/base/hook/README.md index f48317645..461042eff 100644 --- a/api/base/hook/README.md +++ b/api/base/hook/README.md @@ -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 diff --git a/doc/rfc/hook-framework.md b/doc/rfc/hook-framework.md index 77305ae52..7ba3199d2 100644 --- a/doc/rfc/hook-framework.md +++ b/doc/rfc/hook-framework.md @@ -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 ``` @@ -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 diff --git a/platform/extension/hook/BUILD.bazel b/platform/extension/hook/BUILD.bazel new file mode 100644 index 000000000..bc6ba8b9c --- /dev/null +++ b/platform/extension/hook/BUILD.bazel @@ -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"], +) diff --git a/platform/extension/hook/README.md b/platform/extension/hook/README.md new file mode 100644 index 000000000..3e5213e44 --- /dev/null +++ b/platform/extension/hook/README.md @@ -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. diff --git a/platform/extension/hook/hook.go b/platform/extension/hook/hook.go new file mode 100644 index 000000000..8785c126a --- /dev/null +++ b/platform/extension/hook/hook.go @@ -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 { + // 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 +} diff --git a/platform/extension/hook/mock/BUILD.bazel b/platform/extension/hook/mock/BUILD.bazel new file mode 100644 index 000000000..667788250 --- /dev/null +++ b/platform/extension/hook/mock/BUILD.bazel @@ -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", + ], +) diff --git a/platform/extension/hook/mock/hook_mock.go b/platform/extension/hook/mock/hook_mock.go new file mode 100644 index 000000000..33da1fa99 --- /dev/null +++ b/platform/extension/hook/mock/hook_mock.go @@ -0,0 +1,109 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: hook.go +// +// Generated by this command: +// +// mockgen -source=hook.go -destination=mock/hook_mock.go -package=mock +// + +// Package mock is a generated GoMock package. +package mock + +import ( + context "context" + reflect "reflect" + + hook "github.com/uber/submitqueue/api/base/hook" + hook0 "github.com/uber/submitqueue/platform/extension/hook" + gomock "go.uber.org/mock/gomock" +) + +// MockHook is a mock of Hook interface. +type MockHook struct { + ctrl *gomock.Controller + recorder *MockHookMockRecorder + isgomock struct{} +} + +// MockHookMockRecorder is the mock recorder for MockHook. +type MockHookMockRecorder struct { + mock *MockHook +} + +// NewMockHook creates a new mock instance. +func NewMockHook(ctrl *gomock.Controller) *MockHook { + mock := &MockHook{ctrl: ctrl} + mock.recorder = &MockHookMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockHook) EXPECT() *MockHookMockRecorder { + return m.recorder +} + +// Handle mocks base method. +func (m *MockHook) Handle(ctx context.Context, event *hook.HookEvent) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Handle", ctx, event) + ret0, _ := ret[0].(error) + return ret0 +} + +// Handle indicates an expected call of Handle. +func (mr *MockHookMockRecorder) Handle(ctx, event any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Handle", reflect.TypeOf((*MockHook)(nil).Handle), ctx, event) +} + +// Name mocks base method. +func (m *MockHook) Name() string { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Name") + ret0, _ := ret[0].(string) + return ret0 +} + +// Name indicates an expected call of Name. +func (mr *MockHookMockRecorder) Name() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Name", reflect.TypeOf((*MockHook)(nil).Name)) +} + +// MockHooks is a mock of Hooks interface. +type MockHooks struct { + ctrl *gomock.Controller + recorder *MockHooksMockRecorder + isgomock struct{} +} + +// MockHooksMockRecorder is the mock recorder for MockHooks. +type MockHooksMockRecorder struct { + mock *MockHooks +} + +// NewMockHooks creates a new mock instance. +func NewMockHooks(ctrl *gomock.Controller) *MockHooks { + mock := &MockHooks{ctrl: ctrl} + mock.recorder = &MockHooksMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockHooks) EXPECT() *MockHooksMockRecorder { + return m.recorder +} + +// For mocks base method. +func (m *MockHooks) For(event *hook.HookEvent) []hook0.Hook { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "For", event) + ret0, _ := ret[0].([]hook0.Hook) + return ret0 +} + +// For indicates an expected call of For. +func (mr *MockHooksMockRecorder) For(event any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "For", reflect.TypeOf((*MockHooks)(nil).For), event) +} diff --git a/platform/extension/hook/noop/BUILD.bazel b/platform/extension/hook/noop/BUILD.bazel new file mode 100644 index 000000000..5015ffb24 --- /dev/null +++ b/platform/extension/hook/noop/BUILD.bazel @@ -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", + ], +) diff --git a/platform/extension/hook/noop/hook.go b/platform/extension/hook/noop/hook.go new file mode 100644 index 000000000..b26e378e7 --- /dev/null +++ b/platform/extension/hook/noop/hook.go @@ -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" } diff --git a/platform/extension/hook/noop/hook_test.go b/platform/extension/hook/noop/hook_test.go new file mode 100644 index 000000000..2350def51 --- /dev/null +++ b/platform/extension/hook/noop/hook_test.go @@ -0,0 +1,37 @@ +// 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 + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" + basehook "github.com/uber/submitqueue/api/base/hook" +) + +func TestHandleAcceptsEveryEvent(t *testing.T) { + events := map[string]*basehook.HookEvent{ + "well-formed": {Id: "submitqueue/batch.failed/batch-778/4", Source: "submitqueue", Type: "batch.failed"}, + "empty": {}, + "nil": nil, + } + + for name, event := range events { + t.Run(name, func(t *testing.T) { + require.NoError(t, New().Handle(context.Background(), event)) + }) + } +} diff --git a/platform/hook/BUILD.bazel b/platform/hook/BUILD.bazel new file mode 100644 index 000000000..110c0ddb6 --- /dev/null +++ b/platform/hook/BUILD.bazel @@ -0,0 +1,41 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = [ + "controller.go", + "dlq.go", + ], + importpath = "github.com/uber/submitqueue/platform/hook", + visibility = ["//visibility:public"], + deps = [ + "//api/base/hook:go_default_library", + "//platform/consumer:go_default_library", + "//platform/extension/hook:go_default_library", + "//platform/metrics:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + "@org_uber_go_zap//:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = [ + "controller_test.go", + "dlq_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/mock:go_default_library", + "//platform/extension/hook:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + "@org_golang_google_protobuf//types/known/structpb:go_default_library", + "@org_uber_go_mock//gomock:go_default_library", + "@org_uber_go_zap//:go_default_library", + ], +) diff --git a/platform/hook/README.md b/platform/hook/README.md new file mode 100644 index 000000000..1fe81ba67 --- /dev/null +++ b/platform/hook/README.md @@ -0,0 +1,54 @@ +# Hook dispatch + +The consumer side of the hooks framework: the stage that turns hook events on a queue into `hook.Hook` calls, and the reconciler for the events that never made it. See [the hooks framework RFC](../../doc/rfc/hook-framework.md) for the design, [`api/base/hook`](../../api/base/hook) for the event contract, and [`platform/extension/hook`](../extension/hook) for the hooks it invokes. + +## Why a stage at all + +Side effects must never stall or fail the pipeline, and "fire and forget" must not mean lossy — a merge-failure comment that silently never posts is a support ticket. A durable queue between the two resolves the tension: the producer's obligation ends once the event is enqueued, which is fast and local, and everything after that gets real retry semantics and a dead-letter queue. + +Calling hooks inline would give up both halves. It couples pipeline latency to whatever an integration talks to, and a crash between the state write and the call drops the notification with nothing to replay. + +## The controller + +Decode, validate, resolve, invoke. That is the whole stage, and it is the same in every domain — "per-domain" in the RFC is about the topic and the wiring, not the logic. The domain-specific parts are the topic name the host maps `hook` to and the hooks its resolver returns. + +Resolution is a `hook.Hooks` call per event, so two queues in one host can run different integrations without running different consumers. The controller itself has no notion of a queue: it hands the resolver the event and runs whatever comes back, which keeps the routing decision in the wiring layer that knows the queue topology. + +Every resolved hook runs even after one of them fails, so a broken integration cannot stop the others from seeing the event. Failures are attributed to the hook that raised them and joined, so the classifier weighs each one on its own merits — a transient failure alongside a permanent one still retries. + +A host with no integrations still registers the stage. Opting in is a topic-key registration, and a registered host never skips an event, so "hooks are off" and "an event was lost" stay distinguishable. + +Outcomes: + +| Situation | Result | +|---|---| +| Every resolved hook returns nil, or none resolve | Ack. Also how a hook ignores an event — there is no filter API. | +| Any resolved hook returns an error | Nack, retry, and dead-letter past the budget. | +| Payload does not decode, or the envelope is missing `id`/`source`/`type` | Non-retryable, so it dead-letters rather than being silently acked. | + +Ordering is per subject only, since the partition key is the subject id. Hook outcomes never write pipeline state. + +## The DLQ reconciler + +Every other DLQ reconciler in the repo repairs something: a stuck request driven to a terminal `failed`, a batch failed and fanned out. This one repairs nothing, because there is nothing it may touch. A hook never writes pipeline state, so an undelivered hook event leaves no half-finished transition behind. What is lost is the side effect itself, and only a person can decide how to recover it. + +So it makes the loss impossible to miss and hands it over: it logs the complete event (the raw protojson, which survives even when the event is here *because* it would not decode) along with the failure attribution, counts it on `reconcile.events_dropped`, and acks so the event does not sit in the DLQ unnoticed. Republishing the logged event recovers it. + +`reconcile.events_dropped` is the metric to alert on — it is the only signal that a side effect was lost, since nothing else in the system notices a comment that never posted. That is a deliberate step up from the log topic's DLQ, which warns and moves on: dropping an observability row costs a gap in a read model, which the next write repairs. + +The reconciler never returns an error. A DLQ consumer has no DLQ of its own and treats everything as retryable, so anything but an ack loops forever. + +## Wiring a host + +Register two topics and two controllers: + +- the primary `hook` topic, mapped to a topic name unique to this domain if the queue backend is shared, with `NewController` on the regular consumer; +- the derived `hook_dlq` topic, with `NewDLQController` on the DLQ consumer (`DLQSubscriptionConfig` plus `errs.AlwaysRetryableProcessor`, like every other DLQ consumer). + +A service assembled by `platform/pipeline` gets the pairing, the derived DLQ key, and the retry configuration from the stage table; Stovepipe and Runway wire their consumers by hand and register both controllers directly. + +## Known limit: one retry budget for all hooks + +The hooks an event resolves to run under one consumer, so they share one retry budget and one dead-letter fate. A retry re-delivers the event to all of them, including the ones that already succeeded — which the hook contract's idempotency requirement covers — and one chronically failing integration eventually dead-letters events the others handled fine. + +Per-hook isolation wants a consumer group per hook on the shared topic, which the queue cannot express today: `NewTopicRegistry` rejects a duplicate topic key and `Consumer.Register` admits one controller per key, so a second group fails at construction; and a rejection moves the shared `queue_messages` row to the DLQ for every group rather than only the one that rejected it. Both have to change before the shared budget can be replaced. diff --git a/platform/hook/controller.go b/platform/hook/controller.go new file mode 100644 index 000000000..69dc6dd2c --- /dev/null +++ b/platform/hook/controller.go @@ -0,0 +1,153 @@ +// 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 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. +// +// 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, +// not about the logic, which is the same everywhere: decode, validate, resolve, +// invoke. The domain-specific parts are the topic name the host maps the key to +// and the hooks its resolver returns. +// +// The contract this stage consumes is api/base/hook; the hooks it invokes +// implement platform/extension/hook. +package hook + +import ( + "context" + "errors" + "fmt" + + "github.com/uber-go/tally" + basehook "github.com/uber/submitqueue/api/base/hook" + "github.com/uber/submitqueue/platform/consumer" + hookext "github.com/uber/submitqueue/platform/extension/hook" + "github.com/uber/submitqueue/platform/metrics" + "go.uber.org/zap" +) + +// dispatchOp is the metric operation name shared by every emit in this file. +const dispatchOp = "dispatch" + +// unknownTagValue stands in for an envelope field that could not be read, so a +// metric series exists for events that failed before they could be attributed. +const unknownTagValue = "unknown" + +// Controller consumes hook events and runs the hooks each one resolves to. +type Controller struct { + logger *zap.SugaredLogger + metricsScope tally.Scope + hooks hookext.Hooks + topicKey consumer.TopicKey + consumerGroup string +} + +var _ consumer.Controller = (*Controller)(nil) + +// NewController builds the hook controller for a host. A host registers the +// stage even when it has no integrations yet — a resolver that returns no hooks +// still acks, so "off" and "lost" stay distinguishable. +func NewController( + logger *zap.SugaredLogger, + scope tally.Scope, + hooks hookext.Hooks, + topicKey consumer.TopicKey, + consumerGroup string, +) *Controller { + name := string(topicKey) + "_controller" + return &Controller{ + logger: logger.Named(name), + metricsScope: scope.SubScope(name), + hooks: hooks, + topicKey: topicKey, + consumerGroup: consumerGroup, + } +} + +// Process decodes the delivery's hook event, validates it, and runs every hook +// the resolver returns for it. Returns nil to ack, or an error to nack (retry) / +// reject (DLQ). +// +// An ack means "no hook still has work to do with this event", not "something +// acted on it": a hook that does not care returns nil, and an event no hook +// resolves to is acked unhandled. +func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { + msg := delivery.Message() + + event := &basehook.HookEvent{} + if err := basehook.Unmarshal(msg.Payload, event); err != nil { + metrics.NamedCounter(c.metricsScope, dispatchOp, "deserialize_errors", 1) + // Non-retryable: bytes that are not a hook event will not become one. + return fmt.Errorf("failed to deserialize hook event: %w", err) + } + + if err := basehook.Validate(event); err != nil { + metrics.NamedCounter(c.metricsScope, dispatchOp, "invalid_events", 1) + // Non-retryable: nothing downstream can supply an envelope field the + // publisher omitted. Dead-lettering it is what keeps a malformed event + // visible instead of silently acked. + return fmt.Errorf("refusing to dispatch malformed hook event: %w", err) + } + + hooks := c.hooks.For(event) + + // Every hook runs even after one fails, so a broken integration cannot stop + // the others from seeing the event. Failures carry the name of the hook that + // raised them and are joined, so the classifier weighs each one and the + // error says which integrations failed rather than that something did. + var failures []error + for _, h := range hooks { + if err := h.Handle(ctx, event); err != nil { + metrics.NamedCounter(c.metricsScope, dispatchOp, "hook_errors", 1, + metrics.NewTag("source", event.GetSource()), + metrics.NewTag("event_type", event.GetType()), + metrics.NewTag("hook", h.Name()), + ) + failures = append(failures, fmt.Errorf("hook %s: %w", h.Name(), err)) + } + } + if err := errors.Join(failures...); err != nil { + return fmt.Errorf("failed to dispatch hook event %s: %w", event.GetId(), err) + } + + metrics.NamedCounter(c.metricsScope, dispatchOp, "handled", 1, + metrics.NewTag("source", event.GetSource()), + metrics.NewTag("event_type", event.GetType()), + ) + c.logger.Debugw("dispatched hook event", + "event_id", event.GetId(), + "source", event.GetSource(), + "event_type", event.GetType(), + "version", event.GetVersion(), + "hooks", len(hooks), + ) + return nil +} + +// Name returns the controller name for logging and metrics. +func (c *Controller) Name() string { + return string(c.topicKey) +} + +// TopicKey returns the topic key this controller subscribes to. +func (c *Controller) TopicKey() consumer.TopicKey { + return c.topicKey +} + +// ConsumerGroup returns the consumer group for offset tracking. +func (c *Controller) ConsumerGroup() string { + return c.consumerGroup +} diff --git a/platform/hook/controller_test.go b/platform/hook/controller_test.go new file mode 100644 index 000000000..28ce2322f --- /dev/null +++ b/platform/hook/controller_test.go @@ -0,0 +1,214 @@ +// 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" + "github.com/uber-go/tally" + basehook "github.com/uber/submitqueue/api/base/hook" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" + hookext "github.com/uber/submitqueue/platform/extension/hook" + "go.uber.org/mock/gomock" + "go.uber.org/zap" + "google.golang.org/protobuf/types/known/structpb" +) + +const ( + testTopicKey = "hook" + testGroup = "submitqueue-hook" +) + +// stubHook records the events it was handed and returns a fixed error. +type stubHook struct { + name string + err error + seen []*basehook.HookEvent +} + +func newStubHook(name string, err error) *stubHook { + return &stubHook{name: name, err: err} +} + +func (h *stubHook) Handle(_ context.Context, event *basehook.HookEvent) error { + h.seen = append(h.seen, event) + return h.err +} + +func (h *stubHook) Name() string { return h.name } + +// stubHooks is a hookext.Hooks whose resolution is the function itself. +type stubHooks func(event *basehook.HookEvent) []hookext.Hook + +func (f stubHooks) For(event *basehook.HookEvent) []hookext.Hook { return f(event) } + +// fixedHooks resolves every event to the same hooks. +func fixedHooks(hooks ...hookext.Hook) stubHooks { + return func(*basehook.HookEvent) []hookext.Hook { return hooks } +} + +func validEvent(t *testing.T) *basehook.HookEvent { + t.Helper() + payload, err := structpb.NewStruct(map[string]any{"batch_id": "batch-778"}) + require.NoError(t, err) + return &basehook.HookEvent{ + Id: "submitqueue/batch.failed/batch-778/4", + Source: "submitqueue", + Type: "batch.failed", + TimestampMs: 1722800012345, + Version: 4, + Payload: payload, + } +} + +func hookPayload(t *testing.T, event *basehook.HookEvent) []byte { + t.Helper() + b, err := basehook.Marshal(event) + require.NoError(t, err) + return b +} + +func dispatcherDelivery(ctrl *gomock.Controller, payload []byte) *consumermock.MockDelivery { + d := consumermock.NewMockDelivery(ctrl) + d.EXPECT().Message().Return(entityqueue.NewMessage("msg-1", payload, "batch-778", nil)).AnyTimes() + d.EXPECT().Attempt().Return(1).AnyTimes() + return d +} + +func newController(hooks hookext.Hooks) *Controller { + return NewController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), hooks, testTopicKey, testGroup) +} + +func TestControllerProcess(t *testing.T) { + t.Run("hands a well-formed event to the resolved hook", func(t *testing.T) { + ctrl := gomock.NewController(t) + h := newStubHook("stub", nil) + event := validEvent(t) + + require.NoError(t, newController(fixedHooks(h)).Process(context.Background(), dispatcherDelivery(ctrl, hookPayload(t, event)))) + + require.Len(t, h.seen, 1) + assert.Equal(t, event.GetId(), h.seen[0].GetId()) + assert.Equal(t, event.GetVersion(), h.seen[0].GetVersion()) + assert.Equal(t, "batch-778", h.seen[0].GetPayload().GetFields()["batch_id"].GetStringValue()) + }) + + t.Run("an unversioned event reaches the hook unchanged", func(t *testing.T) { + ctrl := gomock.NewController(t) + h := newStubHook("stub", nil) + event := validEvent(t) + event.Version = 0 + + require.NoError(t, newController(fixedHooks(h)).Process(context.Background(), dispatcherDelivery(ctrl, hookPayload(t, event)))) + + require.Len(t, h.seen, 1) + assert.Zero(t, h.seen[0].GetVersion()) + }) + + t.Run("a hook failure fails the delivery", func(t *testing.T) { + ctrl := gomock.NewController(t) + boom := errors.New("boom") + h := newStubHook("stub", boom) + + err := newController(fixedHooks(h)).Process(context.Background(), dispatcherDelivery(ctrl, hookPayload(t, validEvent(t)))) + + require.Error(t, err) + assert.ErrorIs(t, err, boom) + }) + + t.Run("an event no hook resolves to is acked", func(t *testing.T) { + ctrl := gomock.NewController(t) + + require.NoError(t, newController(fixedHooks()).Process(context.Background(), dispatcherDelivery(ctrl, hookPayload(t, validEvent(t))))) + }) +} + +// One failing hook must not keep the others from seeing the event, and the +// error must say which hooks failed so a mixed outcome is attributable. +func TestControllerRunsEveryResolvedHook(t *testing.T) { + ctrl := gomock.NewController(t) + exportBoom := errors.New("export boom") + commentBoom := errors.New("comment boom") + export := newStubHook("warehouse", exportBoom) + comment := newStubHook("code-host", commentBoom) + audit := newStubHook("audit", nil) + + err := newController(fixedHooks(export, comment, audit)). + Process(context.Background(), dispatcherDelivery(ctrl, hookPayload(t, validEvent(t)))) + + require.Error(t, err) + assert.ErrorIs(t, err, exportBoom) + assert.ErrorIs(t, err, commentBoom) + for _, h := range []*stubHook{export, comment, audit} { + assert.Len(t, h.seen, 1, "hook %s should have seen the event", h.Name()) + } +} + +func TestControllerRunsOnlyTheHooksResolvedForTheEvent(t *testing.T) { + ctrl := gomock.NewController(t) + batchHook := newStubHook("batch", nil) + requestHook := newStubHook("request", nil) + hooks := stubHooks(func(event *basehook.HookEvent) []hookext.Hook { + if event.GetType() == "batch.failed" { + return []hookext.Hook{batchHook} + } + return []hookext.Hook{requestHook} + }) + + require.NoError(t, newController(hooks).Process(context.Background(), dispatcherDelivery(ctrl, hookPayload(t, validEvent(t))))) + + assert.Len(t, batchHook.seen, 1) + assert.Empty(t, requestHook.seen) +} + +// A malformed event must fail rather than ack: dead-lettering is what keeps the +// loss visible, and no hook should see an event the contract rejects. +func TestControllerRejectsMalformedEvents(t *testing.T) { + valid := validEvent(t) + + cases := map[string][]byte{ + "not json": []byte("{definitely not json"), + "empty payload": {}, + "no id": hookPayload(t, &basehook.HookEvent{Source: valid.Source, Type: valid.Type}), + "no source": hookPayload(t, &basehook.HookEvent{Id: valid.Id, Type: valid.Type}), + "no type": hookPayload(t, &basehook.HookEvent{Id: valid.Id, Source: valid.Source}), + } + + for name, payload := range cases { + t.Run(name, func(t *testing.T) { + ctrl := gomock.NewController(t) + h := newStubHook("stub", nil) + + require.Error(t, newController(fixedHooks(h)).Process(context.Background(), dispatcherDelivery(ctrl, payload))) + assert.Empty(t, h.seen, "a malformed event must never reach the hook") + }) + } +} + +// An event carrying a field this build does not know about must still dispatch: +// producers add fields without waiting for consumers. +func TestControllerToleratesUnknownFields(t *testing.T) { + ctrl := gomock.NewController(t) + h := newStubHook("stub", nil) + payload := []byte(`{"id":"a/b/c/1","source":"a","type":"b","field_from_the_future":7}`) + + require.NoError(t, newController(fixedHooks(h)).Process(context.Background(), dispatcherDelivery(ctrl, payload))) + require.Len(t, h.seen, 1) +} diff --git a/platform/hook/dlq.go b/platform/hook/dlq.go new file mode 100644 index 000000000..b34b1556a --- /dev/null +++ b/platform/hook/dlq.go @@ -0,0 +1,130 @@ +// 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" + + "github.com/uber-go/tally" + basehook "github.com/uber/submitqueue/api/base/hook" + "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/platform/metrics" + "go.uber.org/zap" +) + +// reconcileOp is the metric operation name shared by every emit in this file. +const reconcileOp = "reconcile" + +// DLQController is the reconciler for the hook topic's dead-letter queue. +// Implements consumer.Controller. +// +// It reconciles nothing, because there is nothing it may touch: a hook never +// writes pipeline state, so a hook event that could not be delivered leaves no +// half-finished transition behind. What is lost is the side effect — a comment +// not posted, a row not exported — which only a person can decide how to +// recover. So this controller makes the loss impossible to miss and hands it +// over: it records the complete event and why it failed, counts it on a metric +// meant to page, and acks so the event does not sit in the DLQ unnoticed. +// Republishing the logged event recovers it. +// +// That is a deliberate step up from the log topic's DLQ, which warns and moves +// on. Dropping an observability row costs a gap in a read model; dropping a +// merge-failure comment costs a support ticket, and nothing else in the system +// will notice it is missing. +type DLQController struct { + logger *zap.SugaredLogger + metricsScope tally.Scope + topicKey consumer.TopicKey + consumerGroup string +} + +var _ consumer.Controller = (*DLQController)(nil) + +// NewDLQController builds the DLQ reconciler for a host's hook topic. topicKey +// is the dead-letter key (the hook topic key plus the queue's DLQ suffix), not +// the primary one. +func NewDLQController( + logger *zap.SugaredLogger, + scope tally.Scope, + topicKey consumer.TopicKey, + consumerGroup string, +) *DLQController { + name := string(topicKey) + "_controller" + return &DLQController{ + logger: logger.Named(name), + metricsScope: scope.SubScope(name), + topicKey: topicKey, + consumerGroup: consumerGroup, + } +} + +// Process records a dropped hook event and acks it. +// +// It never returns an error: a failure here would re-deliver the message +// forever, since the DLQ consumer has no DLQ of its own and treats everything as +// retryable. The record is the outcome, so the only way to fail is not to write +// one. +func (c *DLQController) Process(_ context.Context, delivery consumer.Delivery) error { + msg := delivery.Message() + + // Decoding is best-effort: the event may be here precisely because it could + // not be decoded. The raw payload is protojson, so logging it verbatim + // preserves the whole event either way; the decoded fields only add + // dimensions worth filtering and alerting on. + source, eventType, eventID := unknownTagValue, unknownTagValue, "" + event := &basehook.HookEvent{} + if err := basehook.Unmarshal(msg.Payload, event); err == nil { + source, eventType, eventID = event.GetSource(), event.GetType(), event.GetId() + } + + metrics.NamedCounter(c.metricsScope, reconcileOp, "events_dropped", 1, + metrics.NewTag("source", source), + metrics.NewTag("event_type", eventType), + ) + + dmeta := delivery.Metadata() + fields := []any{ + "message_id", msg.ID, + "event_id", eventID, + "source", source, + "event_type", eventType, + "event", string(msg.Payload), + "attempt", delivery.Attempt(), + "dlq_original_topic", dmeta["dlq.original_topic"], + "dlq_failure_count", dmeta["dlq.failure_count"], + "dlq_last_error", dmeta["dlq.last_error"], + } + if f, ok := delivery.Failure(); ok { + fields = append(fields, "failure", f.Message, "failure_subjects", f.Subjects, "failure_detail", f.Detail) + } + + c.logger.Errorw("hook event dropped to dlq; republish the logged event to recover", fields...) + return nil +} + +// Name returns the controller name for logging and metrics. +func (c *DLQController) Name() string { + return string(c.topicKey) +} + +// TopicKey returns the topic key this controller subscribes to. +func (c *DLQController) TopicKey() consumer.TopicKey { + return c.topicKey +} + +// ConsumerGroup returns the consumer group for offset tracking. +func (c *DLQController) ConsumerGroup() string { + return c.consumerGroup +} diff --git a/platform/hook/dlq_test.go b/platform/hook/dlq_test.go new file mode 100644 index 000000000..e7b50b26d --- /dev/null +++ b/platform/hook/dlq_test.go @@ -0,0 +1,115 @@ +// 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" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + basehook "github.com/uber/submitqueue/api/base/hook" + "github.com/uber/submitqueue/platform/base/failure" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" + "go.uber.org/mock/gomock" + "go.uber.org/zap" +) + +const testDLQTopicKey = "hook_dlq" + +func dlqDelivery(ctrl *gomock.Controller, payload []byte, f *failure.Failure) *consumermock.MockDelivery { + d := consumermock.NewMockDelivery(ctrl) + d.EXPECT().Message().Return(entityqueue.NewMessage("msg-1", payload, "batch-778", nil)).AnyTimes() + d.EXPECT().Attempt().Return(4).AnyTimes() + d.EXPECT().Metadata().Return(map[string]string{ + "dlq.original_topic": "hook", + "dlq.failure_count": "3", + "dlq.last_error": "boom", + }).AnyTimes() + if f == nil { + d.EXPECT().Failure().Return(failure.Failure{}, false).AnyTimes() + } else { + d.EXPECT().Failure().Return(*f, true).AnyTimes() + } + return d +} + +func newDLQController(scope tally.Scope) *DLQController { + return NewDLQController(zap.NewNop().Sugar(), scope, testDLQTopicKey, "submitqueue-hook-dlq") +} + +// The DLQ consumer has no DLQ of its own and treats every error as retryable, so +// anything but an ack loops the message forever. Whatever the payload, the +// reconciler must ack. +func TestDLQControllerAlwaysAcks(t *testing.T) { + attributed := failure.New("hook boom", failure.Subject{Type: "batch", ID: "batch-778"}) + + cases := map[string]struct { + payload []byte + failure *failure.Failure + }{ + "decodable event with attribution": {payload: hookPayload(t, validEvent(t)), failure: &attributed}, + "decodable event unattributed": {payload: hookPayload(t, validEvent(t))}, + "undecodable payload": {payload: []byte("{definitely not json"), failure: &attributed}, + "empty payload": {payload: []byte{}}, + } + + for name, tt := range cases { + t.Run(name, func(t *testing.T) { + ctrl := gomock.NewController(t) + c := newDLQController(tally.NewTestScope("test", nil)) + + require.NoError(t, c.Process(context.Background(), dlqDelivery(ctrl, tt.payload, tt.failure))) + }) + } +} + +// The dropped-event counter is the only signal that a side effect was lost — +// nothing else in the system notices a comment that never posted — so it is the +// reconciler's actual output, tagged for attribution. +func TestDLQControllerCountsDroppedEvents(t *testing.T) { + t.Run("attributed to the decoded envelope", func(t *testing.T) { + ctrl := gomock.NewController(t) + scope := tally.NewTestScope("test", nil) + + require.NoError(t, newDLQController(scope).Process( + context.Background(), dlqDelivery(ctrl, hookPayload(t, validEvent(t)), nil))) + + counter, ok := scope.Snapshot().Counters()["test.hook_dlq_controller.reconcile.events_dropped+event_type=batch.failed,source=submitqueue"] + require.True(t, ok) + assert.Equal(t, int64(1), counter.Value()) + }) + + t.Run("counted even when the envelope cannot be read", func(t *testing.T) { + ctrl := gomock.NewController(t) + scope := tally.NewTestScope("test", nil) + + require.NoError(t, newDLQController(scope).Process( + context.Background(), dlqDelivery(ctrl, []byte("{definitely not json"), nil))) + + counter, ok := scope.Snapshot().Counters()["test.hook_dlq_controller.reconcile.events_dropped+event_type=unknown,source=unknown"] + require.True(t, ok) + assert.Equal(t, int64(1), counter.Value()) + }) +} + +func TestDLQControllerIdentity(t *testing.T) { + c := newDLQController(tally.NewTestScope("test", nil)) + + assert.Equal(t, basehook.TopicKey(testDLQTopicKey), c.TopicKey()) + assert.Equal(t, "submitqueue-hook-dlq", c.ConsumerGroup()) +}