From a09dcaba2f51328554ba6a89d5eb155e9b65d209 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 +- platform/extension/hook/BUILD.bazel | 9 + platform/extension/hook/README.md | 36 ++++ platform/extension/hook/composite/BUILD.bazel | 24 +++ platform/extension/hook/composite/hook.go | 80 +++++++++ .../extension/hook/composite/hook_test.go | 87 ++++++++++ platform/extension/hook/hook.go | 65 ++++++++ platform/extension/hook/mock/BUILD.bazel | 12 ++ platform/extension/hook/mock/hook_mock.go | 70 ++++++++ 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 | 40 +++++ platform/hook/README.md | 50 ++++++ platform/hook/dispatcher.go | 136 ++++++++++++++++ platform/hook/dispatcher_test.go | 154 ++++++++++++++++++ platform/hook/dlq.go | 130 +++++++++++++++ platform/hook/dlq_test.go | 115 +++++++++++++ 18 files changed, 1112 insertions(+), 1 deletion(-) create mode 100644 platform/extension/hook/BUILD.bazel create mode 100644 platform/extension/hook/README.md create mode 100644 platform/extension/hook/composite/BUILD.bazel create mode 100644 platform/extension/hook/composite/hook.go create mode 100644 platform/extension/hook/composite/hook_test.go 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/dispatcher.go create mode 100644 platform/hook/dispatcher_test.go create mode 100644 platform/hook/dlq.go create mode 100644 platform/hook/dlq_test.go diff --git a/Makefile b/Makefile index e393ba87..3c8c451f 100644 --- a/Makefile +++ b/Makefile @@ -377,7 +377,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/platform/extension/hook/BUILD.bazel b/platform/extension/hook/BUILD.bazel new file mode 100644 index 00000000..bc6ba8b9 --- /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 00000000..d3946462 --- /dev/null +++ b/platform/extension/hook/README.md @@ -0,0 +1,36 @@ +# 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. + +## Wiring + +A hook is wired **once per host**, not resolved per queue, so this package has no `Config` and no `Factory`. What an integration does is a property of the deployment rather than of the queue an event came from; a hook that genuinely needs per-queue behavior resolves the queue from the event payload. + +The host constructs its hook and hands it to the dispatcher in [`platform/hook`](../../hook), which owns the consumer side: decode, validate, invoke. + +## Implementations + +- **`noop/`** — accepts every event and does nothing. The default before a host has any integration, so the seam behaves identically whether or not hooks are configured. +- **`composite/`** — fans an event out to several children, runs all of them even after one fails, and joins the failures with the name of each failing child. Read its package doc before wiring more than one child: they share a single retry budget, so one chronically failing integration eventually dead-letters events the others handled fine. + +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. Wire it into the host's dispatcher — inside a `composite` if the host has more than one. diff --git a/platform/extension/hook/composite/BUILD.bazel b/platform/extension/hook/composite/BUILD.bazel new file mode 100644 index 00000000..8af13d40 --- /dev/null +++ b/platform/extension/hook/composite/BUILD.bazel @@ -0,0 +1,24 @@ +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/composite", + 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", + "//platform/extension/hook:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + ], +) diff --git a/platform/extension/hook/composite/hook.go b/platform/extension/hook/composite/hook.go new file mode 100644 index 00000000..b26e0bab --- /dev/null +++ b/platform/extension/hook/composite/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 composite provides a hook.Hook that fans one event out to several +// children. It is how a host wires more than one integration, since the +// dispatcher takes a single hook. +// +// Every child runs on every event, even after one fails, so a broken +// integration cannot stop the others from seeing the event. Failures are +// collected and joined, each wrapped with the name of the child that raised it, +// so the error reaching the dispatcher says which integration failed rather than +// just that something did. +// +// # Children share one retry budget +// +// The composite is a single consumer, so a retry re-delivers the event to every +// child, including the ones that already succeeded. Two consequences: children +// must be idempotent on the event id (the hook contract requires this anyway), +// and one persistently failing child spends the budget for all of them, so the +// event eventually dead-letters even though the others were fine. +// +// The fix is a consumer group per hook on the shared hook topic, which the queue +// cannot express today: the registry admits one consumer group per topic key, +// and a rejection moves the shared message row to the DLQ for every group rather +// than only the one that rejected it. Until both change, prefer wiring children +// whose failure modes are independent and short-lived, and treat a chronically +// failing integration as something to remove from the composite rather than to +// absorb. +package composite + +import ( + "context" + "errors" + "fmt" + + 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 fans an event out to every child hook. +type Hook struct { + // children are the hooks the event is handed to, in wiring order. + children []hook.Hook +} + +// New returns a Hook that hands each event to every child in the order given. +// With no children it accepts every event and does nothing. +func New(children ...hook.Hook) Hook { + return Hook{children: children} +} + +// Handle implements hook.Hook. It runs every child and returns the joined +// failures, each attributed to the child that raised it, or nil when all +// succeeded. +func (h Hook) Handle(ctx context.Context, event *basehook.HookEvent) error { + var failures []error + for _, child := range h.children { + if err := child.Handle(ctx, event); err != nil { + failures = append(failures, fmt.Errorf("hook %s: %w", child.Name(), err)) + } + } + return errors.Join(failures...) +} + +// Name implements hook.Hook. +func (Hook) Name() string { return "composite" } diff --git a/platform/extension/hook/composite/hook_test.go b/platform/extension/hook/composite/hook_test.go new file mode 100644 index 00000000..0c12f8bc --- /dev/null +++ b/platform/extension/hook/composite/hook_test.go @@ -0,0 +1,87 @@ +// 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 composite + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + basehook "github.com/uber/submitqueue/api/base/hook" + "github.com/uber/submitqueue/platform/extension/hook" +) + +// recordingHook records the events it saw and fails with a fixed error. +type recordingHook struct { + name string + err error + seen []string +} + +var _ hook.Hook = (*recordingHook)(nil) + +func (h *recordingHook) Handle(_ context.Context, event *basehook.HookEvent) error { + h.seen = append(h.seen, event.GetId()) + return h.err +} + +func (h *recordingHook) Name() string { return h.name } + +func event() *basehook.HookEvent { + return &basehook.HookEvent{Id: "submitqueue/batch.failed/batch-778/4", Source: "submitqueue", Type: "batch.failed"} +} + +func TestHandle(t *testing.T) { + t.Run("no children", func(t *testing.T) { + require.NoError(t, New().Handle(context.Background(), event())) + }) + + t.Run("every child sees the event", func(t *testing.T) { + first := &recordingHook{name: "first"} + second := &recordingHook{name: "second"} + + require.NoError(t, New(first, second).Handle(context.Background(), event())) + assert.Equal(t, []string{event().GetId()}, first.seen) + assert.Equal(t, []string{event().GetId()}, second.seen) + }) + + t.Run("a failing child does not stop the others", func(t *testing.T) { + boom := errors.New("boom") + failing := &recordingHook{name: "failing", err: boom} + healthy := &recordingHook{name: "healthy"} + + err := New(failing, healthy).Handle(context.Background(), event()) + + require.Error(t, err) + assert.ErrorIs(t, err, boom) + assert.Equal(t, []string{event().GetId()}, healthy.seen, "the healthy child runs after the failing one") + }) + + t.Run("every failure survives the join", func(t *testing.T) { + first := errors.New("first failure") + second := errors.New("second failure") + + err := New( + &recordingHook{name: "first", err: first}, + &recordingHook{name: "second", err: second}, + ).Handle(context.Background(), event()) + + require.Error(t, err) + assert.ErrorIs(t, err, first) + assert.ErrorIs(t, err, second) + }) +} diff --git a/platform/extension/hook/hook.go b/platform/extension/hook/hook.go new file mode 100644 index 00000000..14bae496 --- /dev/null +++ b/platform/extension/hook/hook.go @@ -0,0 +1,65 @@ +// 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. +// +// A hook is wired once per host rather than resolved per queue, because what an +// integration does — post a comment, write a row — is a property of the +// deployment, not of the queue the event came from. There is therefore no Config +// and no Factory here: the host constructs its hook directly and hands it to the +// dispatcher. A hook that genuinely needs per-queue behavior resolves the queue +// from the event payload. +// +// 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 a + // composite reports. Stable and unique among the hooks a host wires. + Name() string +} diff --git a/platform/extension/hook/mock/BUILD.bazel b/platform/extension/hook/mock/BUILD.bazel new file mode 100644 index 00000000..bc4f417a --- /dev/null +++ b/platform/extension/hook/mock/BUILD.bazel @@ -0,0 +1,12 @@ +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", + "@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 00000000..7c61154c --- /dev/null +++ b/platform/extension/hook/mock/hook_mock.go @@ -0,0 +1,70 @@ +// 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" + 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)) +} diff --git a/platform/extension/hook/noop/BUILD.bazel b/platform/extension/hook/noop/BUILD.bazel new file mode 100644 index 00000000..5015ffb2 --- /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 00000000..0e906371 --- /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 the default a host wires before it has any integration, which keeps the +// dispatcher's behavior identical whether or not hooks are configured: 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 00000000..2350def5 --- /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 00000000..4281248a --- /dev/null +++ b/platform/hook/BUILD.bazel @@ -0,0 +1,40 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = [ + "dispatcher.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 = [ + "dispatcher_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", + "@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 00000000..3b56c78e --- /dev/null +++ b/platform/hook/README.md @@ -0,0 +1,50 @@ +# 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 dispatcher + +Decode, validate, 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 hook it wires. + +A host with no integrations wires the `noop` hook rather than skipping 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 | +|---|---| +| Hook returns nil | Ack. Also how a hook ignores an event — there is no filter API. | +| 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 `NewDispatcher` 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 dispatcher takes a single hook, so a host with several integrations wires a `composite` and they share one consumer, one retry budget, and one dead-letter fate. 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 composite's shared budget can be replaced. diff --git a/platform/hook/dispatcher.go b/platform/hook/dispatcher.go new file mode 100644 index 00000000..5f476270 --- /dev/null +++ b/platform/hook/dispatcher.go @@ -0,0 +1,136 @@ +// 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 dispatcher +// that turns hook events on a queue into hook.Hook calls, and the reconciler for +// the events that never made it. +// +// The dispatcher 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, invoke. +// The domain-specific parts are the topic name the host maps the key to, and the +// hook it wires. +// +// The contract this stage consumes is api/base/hook; the hooks it invokes +// implement platform/extension/hook. +package hook + +import ( + "context" + "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" + +// Dispatcher consumes hook events and hands each to the host's hook. +type Dispatcher struct { + logger *zap.SugaredLogger + metricsScope tally.Scope + hook hookext.Hook + topicKey consumer.TopicKey + consumerGroup string +} + +var _ consumer.Controller = (*Dispatcher)(nil) + +// NewDispatcher builds the hook dispatcher for a host. A host with no +// integrations wires the noop hook rather than skipping the stage, so that "off" +// and "lost" stay distinguishable. +func NewDispatcher( + logger *zap.SugaredLogger, + scope tally.Scope, + h hookext.Hook, + topicKey consumer.TopicKey, + consumerGroup string, +) *Dispatcher { + name := string(topicKey) + "_controller" + return &Dispatcher{ + logger: logger.Named(name), + metricsScope: scope.SubScope(name), + hook: h, + topicKey: topicKey, + consumerGroup: consumerGroup, + } +} + +// Process decodes the delivery's hook event, validates it, and invokes the +// host's hook. Returns nil to ack, or an error to nack (retry) / reject (DLQ). +// +// A hook that does not care about this event returns nil, so an ack here means +// "no hook still has work to do with it", not "something acted on it". +func (d *Dispatcher) 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(d.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(d.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) + } + + tags := []metrics.Tag{ + metrics.NewTag("source", event.GetSource()), + metrics.NewTag("event_type", event.GetType()), + } + + if err := d.hook.Handle(ctx, event); err != nil { + metrics.NamedCounter(d.metricsScope, dispatchOp, "hook_errors", 1, tags...) + return fmt.Errorf("hook %s failed to handle event %s: %w", d.hook.Name(), event.GetId(), err) + } + + metrics.NamedCounter(d.metricsScope, dispatchOp, "handled", 1, tags...) + d.logger.Debugw("dispatched hook event", + "event_id", event.GetId(), + "source", event.GetSource(), + "event_type", event.GetType(), + "version", event.GetVersion(), + "hook", d.hook.Name(), + ) + return nil +} + +// Name returns the controller name for logging and metrics. +func (d *Dispatcher) Name() string { + return string(d.topicKey) +} + +// TopicKey returns the topic key this controller subscribes to. +func (d *Dispatcher) TopicKey() consumer.TopicKey { + return d.topicKey +} + +// ConsumerGroup returns the consumer group for offset tracking. +func (d *Dispatcher) ConsumerGroup() string { + return d.consumerGroup +} diff --git a/platform/hook/dispatcher_test.go b/platform/hook/dispatcher_test.go new file mode 100644 index 00000000..598ee54f --- /dev/null +++ b/platform/hook/dispatcher_test.go @@ -0,0 +1,154 @@ +// 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" + "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 { + err error + seen []*basehook.HookEvent +} + +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 "stub" } + +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 newDispatcher(h *stubHook) *Dispatcher { + return NewDispatcher(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), h, testTopicKey, testGroup) +} + +func TestDispatcherProcess(t *testing.T) { + t.Run("hands a well-formed event to the hook", func(t *testing.T) { + ctrl := gomock.NewController(t) + h := &stubHook{} + event := validEvent(t) + + require.NoError(t, newDispatcher(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 := &stubHook{} + event := validEvent(t) + event.Version = 0 + + require.NoError(t, newDispatcher(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 := &stubHook{err: boom} + + err := newDispatcher(h).Process(context.Background(), dispatcherDelivery(ctrl, hookPayload(t, validEvent(t)))) + + require.Error(t, err) + assert.ErrorIs(t, err, boom) + }) +} + +// 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 TestDispatcherRejectsMalformedEvents(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 := &stubHook{} + + require.Error(t, newDispatcher(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 TestDispatcherToleratesUnknownFields(t *testing.T) { + ctrl := gomock.NewController(t) + h := &stubHook{} + payload := []byte(`{"id":"a/b/c/1","source":"a","type":"b","field_from_the_future":7}`) + + require.NoError(t, newDispatcher(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 00000000..b34b1556 --- /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 00000000..e7b50b26 --- /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()) +}