From b5394ebea7d0d0b923994bd96ed8d4e63e32c381 Mon Sep 17 00:00:00 2001 From: "prath.shenoy" Date: Mon, 17 Aug 2026 14:25:22 +0000 Subject: [PATCH 1/3] feat(hook): Add contract for lifecycle events --- Makefile | 2 +- api/base/hook/BUILD.bazel | 31 +++++ api/base/hook/README.md | 49 +++++++ api/base/hook/event.go | 72 ++++++++++ api/base/hook/hook.go | 62 +++++++++ api/base/hook/hook_test.go | 184 ++++++++++++++++++++++++++ api/base/hook/proto/BUILD.bazel | 4 + api/base/hook/proto/hook.proto | 58 +++++++++ api/base/hook/protopb/BUILD.bazel | 14 ++ api/base/hook/protopb/hook.pb.go | 210 ++++++++++++++++++++++++++++++ api/base/hook/topics.go | 26 ++++ tool/proto/BUILD.bazel | 11 ++ 12 files changed, 722 insertions(+), 1 deletion(-) create mode 100644 api/base/hook/BUILD.bazel create mode 100644 api/base/hook/README.md create mode 100644 api/base/hook/event.go create mode 100644 api/base/hook/hook.go create mode 100644 api/base/hook/hook_test.go create mode 100644 api/base/hook/proto/BUILD.bazel create mode 100644 api/base/hook/proto/hook.proto create mode 100644 api/base/hook/protopb/BUILD.bazel create mode 100644 api/base/hook/protopb/hook.pb.go create mode 100644 api/base/hook/topics.go diff --git a/Makefile b/Makefile index ff8fe2832..e393ba872 100644 --- a/Makefile +++ b/Makefile @@ -36,7 +36,7 @@ GOIMPORTS_VERSION ?= v0.33.0 # (the out_dir convention in tool/proto/BUILD.bazel) and copied back here. A # package may hold multiple .proto files (e.g. an RPC contract plus messagequeue # contracts); all generated stubs land in the same protopb/ dir. -PROTO_PACKAGES = api/base/change api/base/mergestrategy api/base/messagequeue api/runway/messagequeue api/runway api/submitqueue/gateway api/submitqueue/orchestrator api/stovepipe stovepipe/core/messagequeue +PROTO_PACKAGES = api/base/change api/base/hook api/base/mergestrategy api/base/messagequeue api/runway/messagequeue api/runway api/submitqueue/gateway api/submitqueue/orchestrator api/stovepipe stovepipe/core/messagequeue # Set REPO_ROOT for docker-compose export REPO_ROOT := $(shell pwd) diff --git a/api/base/hook/BUILD.bazel b/api/base/hook/BUILD.bazel new file mode 100644 index 000000000..6b15b0874 --- /dev/null +++ b/api/base/hook/BUILD.bazel @@ -0,0 +1,31 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = [ + "event.go", + "hook.go", + "topics.go", + ], + importpath = "github.com/uber/submitqueue/api/base/hook", + visibility = ["//visibility:public"], + deps = [ + "//api/base/hook/protopb:go_default_library", + "//api/base/messagequeue/protopb:go_default_library", + "//platform/consumer:go_default_library", + "@org_golang_google_protobuf//encoding/protojson:go_default_library", + "@org_golang_google_protobuf//proto:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["hook_test.go"], + embed = [":go_default_library"], + deps = [ + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + "@org_golang_google_protobuf//proto:go_default_library", + "@org_golang_google_protobuf//types/known/structpb:go_default_library", + ], +) diff --git a/api/base/hook/README.md b/api/base/hook/README.md new file mode 100644 index 000000000..f48317645 --- /dev/null +++ b/api/base/hook/README.md @@ -0,0 +1,49 @@ +# Hook event contract + +The published, language-neutral contract for hook events: fire-and-forget lifecycle notifications that let integrations react to a pipeline transition without being able to stall or fail the pipeline. See [the hooks framework RFC](../../../doc/rfc/hook-framework.md) for the design and [the message queue contract RFC](../../../doc/rfc/messagequeue-contract.md) for the conventions it follows. + +It lives under `api/base/` rather than `api/{domain}/` because no domain owns it. Every domain publishes this same shape to its own hook topic, so a sink consuming several domains reads one schema rather than one per producer. + +Payloads are defined as proto3 messages in [`proto/hook.proto`](proto/hook.proto) and generated into [`protopb/`](protopb); the proto is the authority and a non-Go client compiles against it directly. On the wire, payloads are serialized as protobuf JSON (`protojson`), so the queue keeps storing self-describing JSON. The Go helpers here are generic `protojson` glue — `Marshal(m)` and `Unmarshal[T](b, m)` — plus the two rules that must be identical across producers: how an id is minted and what makes an event well-formed. Field names stay snake_case (`UseProtoNames`) and `int64` fields serialize as JSON strings. + +## The envelope + +`HookEvent` carries `id`, `source`, `type`, `timestamp_ms`, `version`, and `payload`. The envelope holds only what every consumer keys on uniformly; everything specific to what happened lives in the payload. + +`source` and `type` are open strings rather than enums, and `payload` is a `google.protobuf.Struct` rather than a `oneof`. That is the central trade: a producer adds a new event type by publishing it, instead of by changing the wire contract and redeploying every consumer. protojson rejects unknown *enum* values, so an enum here would break existing consumers on every addition. + +Subject, queue, and error are deliberately **not** on the envelope. They are facts about a particular occurrence, so they belong in the payload — no major event platform carries a top-level error either. + +## Identity and idempotency + +`id` is derived from the transition, not random: `source`, `type`, the subject's id, and the subject's post-transition `version`, joined. `NewEventID` mints it. Replaying the delivery that caused the transition therefore mints the *same* id, which is what lets the queue dedupe the redelivery and lets a hook stay idempotent by keying on it. That derivation is why the framework needs no transactional outbox: the publish rides inside the delivery that performed the state write, and a crash before the ack replays both halves safely. + +When a transition is not a versioned write there is no version to distinguish occurrences, so the id of the message that caused it stands in, plus an ordinal when one cause publishes several same-typed events. `NewUnversionedEventID` mints that form. + +Consumers never parse an id. It is a dedupe and idempotency key, not a structured field. + +## Staleness + +`version` is the subject's optimistic-locking version immediately after the transition, and `0` when the transition was not a versioned write. Delivery is at-least-once, so a hook can receive an event describing a transition that has since been superseded; comparing this version against the subject's current version in the store is how it tells the two apart. Timestamps cannot answer that, because the clocks belong to different machines. + +A domain with no versioned entities (Runway holds no durable state of its own) publishes `0` throughout. That is the normal mode for such a producer, not a degenerate case. + +## Payload + +Shaped per `type` by the domain that publishes it, add-only, and documented by that domain. It must carry the subject's id, and it must carry any fact recorded nowhere else — merge step outcomes, build failure detail — because for those the event is the only durable record. + +It must **not** be an entity snapshot. A snapshot is stale the moment it is redelivered, it competes with the store as a source of truth, and it drags a domain's schema into a contract shared by every domain. Hooks resolve entities from their stores. + +## Topic keys + +The binding between a topic key and its payload lives in the message's `topic_keys` option (defined in `api/base/messagequeue`); `TopicKeys` reads it back by reflection. A topic key is a stable logical name, not a concrete wire topic — each implementer maps the key to whatever topic name its broker/queue requires, via `consumer.TopicRegistry` in our Go wiring. + +| Message | Direction | Topic key | +|---|---|---| +| `HookEvent` | producing domain → hook dispatcher | `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. + +## Evolution + +Contract changes are additive-only: add new fields; never remove, rename, repurpose, or retype an existing field, and never reuse a field number. protojson ignores unknown fields on read and omits zero-valued fields on write, so a new optional field is backward-compatible in both directions. New event types and new payload keys are not contract changes at all — that is the point of the open envelope. diff --git a/api/base/hook/event.go b/api/base/hook/event.go new file mode 100644 index 000000000..5d85868db --- /dev/null +++ b/api/base/hook/event.go @@ -0,0 +1,72 @@ +// 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 ( + "fmt" + "strconv" + "strings" +) + +// Consumers never parse an id, so this is a minting convention rather than a +// wire format. All it must guarantee is that two different transitions cannot +// join to the same string. +const idSeparator = "/" + +// NewEventID mints the id of an event describing a versioned state write. +// +// Deriving the id rather than randomizing it is what makes replay safe: the same +// transition mints the same id, so the queue dedupes a redelivery and a hook +// stays idempotent without a publisher-side outbox. version is the subject's +// version immediately after the write, which is what separates two transitions +// of the same subject. +func NewEventID(source, eventType, subjectID string, version int32) string { + return strings.Join([]string{source, eventType, subjectID, strconv.Itoa(int(version))}, idSeparator) +} + +// NewUnversionedEventID mints the id of an event whose transition was not a +// versioned write, so no version distinguishes one occurrence from the next. +// +// The causing message's id stands in for the version, being stable across +// redeliveries for the same reason a version is. ordinal separates several +// same-typed events published for one cause; pass 0 when there is only one. +func NewUnversionedEventID(source, eventType, subjectID, causeID string, ordinal int) string { + return strings.Join( + []string{source, eventType, subjectID, causeID, strconv.Itoa(ordinal)}, + idSeparator, + ) +} + +// Validate reports whether e carries the three envelope fields every consumer +// keys on. The rest cannot be checked generically: version is legitimately 0 for +// an unversioned transition and payload is shaped per type. +// +// Both sides call it — a publisher to catch a malformed event before it reaches +// the queue, a consumer because the producer may not have. +func Validate(e *HookEvent) error { + if e == nil { + return fmt.Errorf("hook event is nil") + } + if e.GetId() == "" { + return fmt.Errorf("hook event has no id") + } + if e.GetSource() == "" { + return fmt.Errorf("hook event %q has no source", e.GetId()) + } + if e.GetType() == "" { + return fmt.Errorf("hook event %q has no type", e.GetId()) + } + return nil +} diff --git a/api/base/hook/hook.go b/api/base/hook/hook.go new file mode 100644 index 000000000..c38023733 --- /dev/null +++ b/api/base/hook/hook.go @@ -0,0 +1,62 @@ +// 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 hook event contract: the wire payload every domain +// publishes to its own hook topic for fire-and-forget lifecycle side effects. +package hook + +import ( + "google.golang.org/protobuf/encoding/protojson" + "google.golang.org/protobuf/proto" + + "github.com/uber/submitqueue/api/base/hook/protopb" + basemqpb "github.com/uber/submitqueue/api/base/messagequeue/protopb" +) + +// HookEvent aliases the generated binding so callers reference the contract +// through this package rather than protopb. +type HookEvent = protopb.HookEvent + +// UseProtoNames keeps JSON field names snake_case, matching the declared +// contract rather than protojson's default lowerCamelCase. +var marshalOpts = protojson.MarshalOptions{UseProtoNames: true} + +// DiscardUnknown makes an additive contract change backward-compatible: a field +// this consumer does not know yet is ignored rather than rejected. +var unmarshalOpts = protojson.UnmarshalOptions{DiscardUnknown: true} + +// Marshal serializes a contract message to protojson bytes for the queue payload. +func Marshal(m proto.Message) ([]byte, error) { + return marshalOpts.Marshal(m) +} + +// Unmarshal deserializes protojson bytes into the contract message m. +func Unmarshal[T proto.Message](b []byte, m T) error { + return unmarshalOpts.Unmarshal(b, m) +} + +// TopicKeys returns the logical topic keys bound to a message via the +// topic_keys proto option, or nil if it declares none. These are not wire topic +// names; a caller maps each key to its backend's topic. +func TopicKeys(m proto.Message) []string { + opts := m.ProtoReflect().Descriptor().Options() + if opts == nil { + return nil + } + keys, ok := proto.GetExtension(opts, basemqpb.E_TopicKeys).([]string) + if !ok { + return nil + } + return keys +} diff --git a/api/base/hook/hook_test.go b/api/base/hook/hook_test.go new file mode 100644 index 000000000..f3a3b58c3 --- /dev/null +++ b/api/base/hook/hook_test.go @@ -0,0 +1,184 @@ +// 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 ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/structpb" +) + +func mustStruct(t *testing.T, fields map[string]any) *structpb.Struct { + t.Helper() + s, err := structpb.NewStruct(fields) + require.NoError(t, err) + return s +} + +func TestHookEventRoundTrip(t *testing.T) { + cases := map[string]*HookEvent{ + "versioned with nested payload": { + Id: "submitqueue/batch.failed/batch-778/4", + Source: "submitqueue", + Type: "batch.failed", + TimestampMs: 1722800012345, + Version: 4, + Payload: mustStruct(t, map[string]any{ + "batch_id": "batch-778", + "queue": "go-monorepo", + "error": "merge conflict", + "failed_step": "sq-12346", + "conflict_paths": []any{"foo/bar.go"}, + }), + }, + "unversioned": { + Id: "runway/merge.completed/queue-a-42/msg-9/0", + Source: "runway", + Type: "merge.completed", + TimestampMs: 1722800012345, + Payload: mustStruct(t, map[string]any{"request_id": "queue-a/42"}), + }, + "envelope only": { + Id: "stovepipe/commit.green/git-abc/1", + Source: "stovepipe", + Type: "commit.green", + TimestampMs: 1722800012345, + Version: 1, + }, + } + + for name, event := range cases { + t.Run(name, func(t *testing.T) { + data, err := Marshal(event) + require.NoError(t, err) + + got := &HookEvent{} + require.NoError(t, Unmarshal(data, got)) + assert.True(t, proto.Equal(event, got), "round-tripped HookEvent should equal the original") + }) + } +} + +func TestWireFormat(t *testing.T) { + data, err := Marshal(&HookEvent{ + Id: "submitqueue/batch.failed/batch-778/4", + Source: "submitqueue", + Type: "batch.failed", + TimestampMs: 1722800012345, + Version: 4, + }) + require.NoError(t, err) + + assert.Contains(t, string(data), `"timestamp_ms"`, "fields must serialize as snake_case") + assert.Contains(t, string(data), `"1722800012345"`, "int64 must serialize as a JSON string") +} + +func TestUnmarshalDiscardsUnknownFields(t *testing.T) { + data := []byte(`{"id":"a/b/c/1","source":"a","type":"b","field_from_the_future":7}`) + + got := &HookEvent{} + require.NoError(t, Unmarshal(data, got)) + assert.Equal(t, "a/b/c/1", got.GetId()) +} + +func TestTopicKeysBindEveryTopicKey(t *testing.T) { + bound := map[string]int{} + for _, m := range []proto.Message{&HookEvent{}} { + keys := TopicKeys(m) + require.NotEmpty(t, keys, "message must declare a non-empty topic_keys option") + for _, key := range keys { + bound[key]++ + } + } + + keys := []TopicKey{TopicKeyHook} + + valid := map[string]bool{} + for _, k := range keys { + valid[k.String()] = true + assert.Equalf(t, 1, bound[k.String()], "topic key %q must be bound to exactly one message via the topic_keys option", k) + } + for key := range bound { + assert.Truef(t, valid[key], "topic_keys option names unknown key %q", key) + } +} + +func TestEventIDIsDerived(t *testing.T) { + t.Run("same transition mints the same id", func(t *testing.T) { + assert.Equal(t, + NewEventID("submitqueue", "batch.failed", "batch-778", 4), + NewEventID("submitqueue", "batch.failed", "batch-778", 4), + ) + }) + + t.Run("distinct transitions mint distinct ids", func(t *testing.T) { + ids := map[string]string{ + "baseline": NewEventID("submitqueue", "batch.failed", "batch-778", 4), + "later version": NewEventID("submitqueue", "batch.failed", "batch-778", 5), + "other subject": NewEventID("submitqueue", "batch.failed", "batch-779", 4), + "other type": NewEventID("submitqueue", "batch.succeeded", "batch-778", 4), + "other source": NewEventID("stovepipe", "batch.failed", "batch-778", 4), + "unversioned": NewUnversionedEventID("submitqueue", "batch.failed", "batch-778", "msg-1", 0), + "second ordinal": NewUnversionedEventID("submitqueue", "batch.failed", "batch-778", "msg-1", 1), + "other cause": NewUnversionedEventID("submitqueue", "batch.failed", "batch-778", "msg-2", 0), + "slashed subject": NewEventID("stovepipe", "commit.green", "request/monorepo/main/42", 4), + } + + seen := map[string]string{} + for name, id := range ids { + if other, dup := seen[id]; dup { + t.Errorf("%q and %q both mint id %q", name, other, id) + } + seen[id] = name + } + }) +} + +func TestValidate(t *testing.T) { + valid := func() *HookEvent { + return &HookEvent{Id: "submitqueue/batch.failed/batch-778/4", Source: "submitqueue", Type: "batch.failed"} + } + + t.Run("well-formed event", func(t *testing.T) { + require.NoError(t, Validate(valid())) + }) + + t.Run("unversioned event is well-formed", func(t *testing.T) { + event := valid() + event.Version = 0 + require.NoError(t, Validate(event)) + }) + + t.Run("event with no payload is well-formed", func(t *testing.T) { + event := valid() + event.Payload = nil + require.NoError(t, Validate(event)) + }) + + malformed := map[string]*HookEvent{ + "nil": nil, + "no id": {Source: "submitqueue", Type: "batch.failed"}, + "no source": {Id: "submitqueue/batch.failed/batch-778/4", Type: "batch.failed"}, + "no type": {Id: "submitqueue/batch.failed/batch-778/4", Source: "submitqueue"}, + } + for name, event := range malformed { + t.Run(name, func(t *testing.T) { + require.Error(t, Validate(event)) + }) + } +} diff --git a/api/base/hook/proto/BUILD.bazel b/api/base/hook/proto/BUILD.bazel new file mode 100644 index 000000000..189731ca7 --- /dev/null +++ b/api/base/hook/proto/BUILD.bazel @@ -0,0 +1,4 @@ +exports_files( + ["hook.proto"], + visibility = ["//tool/proto:__pkg__"], +) diff --git a/api/base/hook/proto/hook.proto b/api/base/hook/proto/hook.proto new file mode 100644 index 000000000..26270f233 --- /dev/null +++ b/api/base/hook/proto/hook.proto @@ -0,0 +1,58 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +syntax = "proto3"; + +package uber.base.hook; + +import "google/protobuf/struct.proto"; + +import "api/base/messagequeue/proto/messagequeue.proto"; + +option go_package = "github.com/uber/submitqueue/api/base/hook/protopb"; +option java_multiple_files = true; +option java_outer_classname = "HookProto"; +option java_package = "com.uber.submitqueue.base.hook"; + +// HookEvent is one fire-and-forget lifecycle event. Every domain publishes this +// same shape to its own hook topic, so a sink that consumes several domains +// reads one schema rather than one per producer. +message HookEvent { + option (uber.base.messagequeue.topic_keys) = "hook"; + + // id is the opaque identity of this occurrence, derived from the transition + // it describes so that replaying the transition mints the same id. It is + // the queue's dedupe key and a hook's idempotency key, and is never parsed. + string id = 1; + // source is the domain that produced the event. An open string rather than + // an enum so a new producer does not break existing consumers. + string source = 2; + // type is what happened, as one dotted open string: "request.landed", + // "batch.failed", etc. It is the only dimension a consumer filters on, and + // open for the same reason as source. + string type = 3; + // timestamp_ms is when the occurrence happened, in milliseconds since the + // Unix epoch, on the publisher's clock. + int64 timestamp_ms = 4; + // version is the subject's optimistic-locking version immediately after the + // transition, and 0 when the transition was not a versioned write. A hook + // compares it against the subject's current version to tell a current event + // from one that has since been superseded. + int32 version = 5; + // payload carries the facts specific to type, always including the + // subject's id, and must carry any fact recorded nowhere else because the + // event is that fact's only durable record. Add-only, and never an entity + // snapshot: hooks resolve entities from their stores. + google.protobuf.Struct payload = 6; +} diff --git a/api/base/hook/protopb/BUILD.bazel b/api/base/hook/protopb/BUILD.bazel new file mode 100644 index 000000000..574ecdd32 --- /dev/null +++ b/api/base/hook/protopb/BUILD.bazel @@ -0,0 +1,14 @@ +load("@rules_go//go:def.bzl", "go_library") + +go_library( + name = "go_default_library", + srcs = ["hook.pb.go"], + importpath = "github.com/uber/submitqueue/api/base/hook/protopb", + visibility = ["//visibility:public"], + deps = [ + "//api/base/messagequeue/protopb:go_default_library", + "@org_golang_google_protobuf//reflect/protoreflect:go_default_library", + "@org_golang_google_protobuf//runtime/protoimpl:go_default_library", + "@org_golang_google_protobuf//types/known/structpb:go_default_library", + ], +) diff --git a/api/base/hook/protopb/hook.pb.go b/api/base/hook/protopb/hook.pb.go new file mode 100644 index 000000000..cc0046c49 --- /dev/null +++ b/api/base/hook/protopb/hook.pb.go @@ -0,0 +1,210 @@ +// 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. + +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.36.10 +// protoc v5.29.3 +// source: hook.proto + +package protopb + +import ( + reflect "reflect" + sync "sync" + unsafe "unsafe" + + _ "github.com/uber/submitqueue/api/base/messagequeue/protopb" + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + structpb "google.golang.org/protobuf/types/known/structpb" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +// HookEvent is one fire-and-forget lifecycle event. Every domain publishes this +// same shape to its own hook topic, so a sink that consumes several domains +// reads one schema rather than one per producer. See api/base/hook/README.md. +type HookEvent struct { + state protoimpl.MessageState `protogen:"open.v1"` + // id is the opaque identity of this occurrence, derived from the transition + // it describes so that replaying the transition mints the same id. It is + // the queue's dedupe key and a hook's idempotency key, and is never parsed. + Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"` + // source is the domain that produced the event: "submitqueue", + // "stovepipe", ... An open string rather than an enum so a new producer + // does not break existing consumers. + Source string `protobuf:"bytes,2,opt,name=source,proto3" json:"source,omitempty"` + // type is what happened, as one dotted open string: "request.landed", + // "batch.failed", ... It is the only dimension a consumer filters on, and + // open for the same reason as source. + Type string `protobuf:"bytes,3,opt,name=type,proto3" json:"type,omitempty"` + // timestamp_ms is when the occurrence happened, in milliseconds since the + // Unix epoch, on the publisher's clock. + TimestampMs int64 `protobuf:"varint,4,opt,name=timestamp_ms,json=timestampMs,proto3" json:"timestamp_ms,omitempty"` + // version is the subject's optimistic-locking version immediately after the + // transition, and 0 when the transition was not a versioned write. A hook + // compares it against the subject's current version to tell a current event + // from one that has since been superseded. + Version int32 `protobuf:"varint,5,opt,name=version,proto3" json:"version,omitempty"` + // payload carries the facts specific to type, always including the + // subject's id, and must carry any fact recorded nowhere else because the + // event is that fact's only durable record. Add-only, and never an entity + // snapshot: hooks resolve entities from their stores. + Payload *structpb.Struct `protobuf:"bytes,6,opt,name=payload,proto3" json:"payload,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *HookEvent) Reset() { + *x = HookEvent{} + mi := &file_hook_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *HookEvent) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*HookEvent) ProtoMessage() {} + +func (x *HookEvent) ProtoReflect() protoreflect.Message { + mi := &file_hook_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use HookEvent.ProtoReflect.Descriptor instead. +func (*HookEvent) Descriptor() ([]byte, []int) { + return file_hook_proto_rawDescGZIP(), []int{0} +} + +func (x *HookEvent) GetId() string { + if x != nil { + return x.Id + } + return "" +} + +func (x *HookEvent) GetSource() string { + if x != nil { + return x.Source + } + return "" +} + +func (x *HookEvent) GetType() string { + if x != nil { + return x.Type + } + return "" +} + +func (x *HookEvent) GetTimestampMs() int64 { + if x != nil { + return x.TimestampMs + } + return 0 +} + +func (x *HookEvent) GetVersion() int32 { + if x != nil { + return x.Version + } + return 0 +} + +func (x *HookEvent) GetPayload() *structpb.Struct { + if x != nil { + return x.Payload + } + return nil +} + +var File_hook_proto protoreflect.FileDescriptor + +const file_hook_proto_rawDesc = "" + + "\n" + + "\n" + + "hook.proto\x12\x0euber.base.hook\x1a\x1cgoogle/protobuf/struct.proto\x1a.api/base/messagequeue/proto/messagequeue.proto\"\xc1\x01\n" + + "\tHookEvent\x12\x0e\n" + + "\x02id\x18\x01 \x01(\tR\x02id\x12\x16\n" + + "\x06source\x18\x02 \x01(\tR\x06source\x12\x12\n" + + "\x04type\x18\x03 \x01(\tR\x04type\x12!\n" + + "\ftimestamp_ms\x18\x04 \x01(\x03R\vtimestampMs\x12\x18\n" + + "\aversion\x18\x05 \x01(\x05R\aversion\x121\n" + + "\apayload\x18\x06 \x01(\v2\x17.google.protobuf.StructR\apayload:\b\x8a\xb5\x18\x04hookB`\n" + + "\x1ecom.uber.submitqueue.base.hookB\tHookProtoP\x01Z1github.com/uber/submitqueue/api/base/hook/protopbb\x06proto3" + +var ( + file_hook_proto_rawDescOnce sync.Once + file_hook_proto_rawDescData []byte +) + +func file_hook_proto_rawDescGZIP() []byte { + file_hook_proto_rawDescOnce.Do(func() { + file_hook_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_hook_proto_rawDesc), len(file_hook_proto_rawDesc))) + }) + return file_hook_proto_rawDescData +} + +var file_hook_proto_msgTypes = make([]protoimpl.MessageInfo, 1) +var file_hook_proto_goTypes = []any{ + (*HookEvent)(nil), // 0: uber.base.hook.HookEvent + (*structpb.Struct)(nil), // 1: google.protobuf.Struct +} +var file_hook_proto_depIdxs = []int32{ + 1, // 0: uber.base.hook.HookEvent.payload:type_name -> google.protobuf.Struct + 1, // [1:1] is the sub-list for method output_type + 1, // [1:1] is the sub-list for method input_type + 1, // [1:1] is the sub-list for extension type_name + 1, // [1:1] is the sub-list for extension extendee + 0, // [0:1] is the sub-list for field type_name +} + +func init() { file_hook_proto_init() } +func file_hook_proto_init() { + if File_hook_proto != nil { + return + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_hook_proto_rawDesc), len(file_hook_proto_rawDesc)), + NumEnums: 0, + NumMessages: 1, + NumExtensions: 0, + NumServices: 0, + }, + GoTypes: file_hook_proto_goTypes, + DependencyIndexes: file_hook_proto_depIdxs, + MessageInfos: file_hook_proto_msgTypes, + }.Build() + File_hook_proto = out.File + file_hook_proto_goTypes = nil + file_hook_proto_depIdxs = nil +} diff --git a/api/base/hook/topics.go b/api/base/hook/topics.go new file mode 100644 index 000000000..8640df720 --- /dev/null +++ b/api/base/hook/topics.go @@ -0,0 +1,26 @@ +// 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 "github.com/uber/submitqueue/platform/consumer" + +// TopicKey looks up a queue backend, topic name, and subscription config in a +// consumer.TopicRegistry. +type TopicKey = consumer.TopicKey + +// TopicKeyHook carries hook events. The key is per-host, not global: each domain +// runs its own hook topic, so two domains sharing a queue backend must map this +// to distinct topic names. +const TopicKeyHook TopicKey = "hook" diff --git a/tool/proto/BUILD.bazel b/tool/proto/BUILD.bazel index 5b9ebca18..16bc485cf 100644 --- a/tool/proto/BUILD.bazel +++ b/tool/proto/BUILD.bazel @@ -23,6 +23,16 @@ go_proto_generated_files( out_dir = "api_base_messagequeue", ) +go_proto_generated_files( + name = "api_base_hook", + srcs = ["//api/base/hook/proto:hook.proto"], + gen_services = False, + imports = [ + "//api/base/messagequeue/proto:messagequeue.proto", + ], + out_dir = "api_base_hook", +) + go_proto_generated_files( name = "api_runway_messagequeue", srcs = ["//api/runway/messagequeue/proto:merge.proto"], @@ -83,6 +93,7 @@ filegroup( name = "generated", srcs = [ ":api_base_change", + ":api_base_hook", ":api_base_mergestrategy", ":api_base_messagequeue", ":api_runway", From a09dcaba2f51328554ba6a89d5eb155e9b65d209 Mon Sep 17 00:00:00 2001 From: "prath.shenoy" Date: Mon, 17 Aug 2026 16:32:34 +0000 Subject: [PATCH 2/3] 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 e393ba872..3c8c451f3 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 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..d3946462b --- /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 000000000..8af13d405 --- /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 000000000..b26e0bab5 --- /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 000000000..0c12f8bc6 --- /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 000000000..14bae496a --- /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 000000000..bc4f417ab --- /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 000000000..7c61154cc --- /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 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..0e9063711 --- /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 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..4281248a3 --- /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 000000000..3b56c78ee --- /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 000000000..5f4762704 --- /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 000000000..598ee54f2 --- /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 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()) +} From 480e8ccb1bae1a495e960b55997769cb77399792 Mon Sep 17 00:00:00 2001 From: "prath.shenoy" Date: Tue, 18 Aug 2026 00:21:58 +0000 Subject: [PATCH 3/3] feat(orchestrator): Consume hook events --- .../orchestrator/server/BUILD.bazel | 1 + .../submitqueue/orchestrator/server/main.go | 2 + submitqueue/orchestrator/BUILD.bazel | 20 ++++- submitqueue/orchestrator/pipeline.go | 25 +++++++ submitqueue/orchestrator/pipeline_test.go | 73 +++++++++++++++++++ 5 files changed, 120 insertions(+), 1 deletion(-) create mode 100644 submitqueue/orchestrator/pipeline_test.go diff --git a/service/submitqueue/orchestrator/server/BUILD.bazel b/service/submitqueue/orchestrator/server/BUILD.bazel index c11a1f67e..186bc4cda 100644 --- a/service/submitqueue/orchestrator/server/BUILD.bazel +++ b/service/submitqueue/orchestrator/server/BUILD.bazel @@ -24,6 +24,7 @@ go_library( "//platform/extension/consumergate/noop:go_default_library", "//platform/extension/counter:go_default_library", "//platform/extension/counter/mysql:go_default_library", + "//platform/extension/hook/noop:go_default_library", "//platform/extension/messagequeue/mysql:go_default_library", "//platform/githubactions:go_default_library", "//platform/http:go_default_library", diff --git a/service/submitqueue/orchestrator/server/main.go b/service/submitqueue/orchestrator/server/main.go index a3a1d21ad..13861a43b 100644 --- a/service/submitqueue/orchestrator/server/main.go +++ b/service/submitqueue/orchestrator/server/main.go @@ -37,6 +37,7 @@ import ( consumergatenoop "github.com/uber/submitqueue/platform/extension/consumergate/noop" "github.com/uber/submitqueue/platform/extension/counter" mysqlcounter "github.com/uber/submitqueue/platform/extension/counter/mysql" + hooknoop "github.com/uber/submitqueue/platform/extension/hook/noop" queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" "github.com/uber/submitqueue/platform/pipeline" "github.com/uber/submitqueue/submitqueue/core/changeset" @@ -198,6 +199,7 @@ func run() error { Analyzer: profiles.AnalyzerFactory(), Speculator: profiles.SpeculatorFactory(), Validator: validatorFactory{}, + Hook: hooknoop.New(), } // Assemble the pipeline: one call builds the topic registry, creates diff --git a/submitqueue/orchestrator/BUILD.bazel b/submitqueue/orchestrator/BUILD.bazel index 82e60a9dc..dab1e92e3 100644 --- a/submitqueue/orchestrator/BUILD.bazel +++ b/submitqueue/orchestrator/BUILD.bazel @@ -1,4 +1,4 @@ -load("@rules_go//go:def.bzl", "go_library") +load("@rules_go//go:def.bzl", "go_library", "go_test") go_library( name = "go_default_library", @@ -6,9 +6,12 @@ go_library( importpath = "github.com/uber/submitqueue/submitqueue/orchestrator", visibility = ["//visibility:public"], deps = [ + "//api/base/hook:go_default_library", "//api/runway/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/extension/counter:go_default_library", + "//platform/extension/hook:go_default_library", + "//platform/hook:go_default_library", "//platform/pipeline:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/extension/buildrunner:go_default_library", @@ -34,3 +37,18 @@ go_library( "@org_uber_go_zap//:go_default_library", ], ) + +go_test( + name = "go_default_test", + srcs = ["pipeline_test.go"], + embed = [":go_default_library"], + deps = [ + "//api/base/hook:go_default_library", + "//platform/extension/hook/noop:go_default_library", + "//platform/pipeline: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_uber_go_zap//zaptest:go_default_library", + ], +) diff --git a/submitqueue/orchestrator/pipeline.go b/submitqueue/orchestrator/pipeline.go index 60cfef5e2..326fc8b69 100644 --- a/submitqueue/orchestrator/pipeline.go +++ b/submitqueue/orchestrator/pipeline.go @@ -18,10 +18,15 @@ package orchestrator import ( + "fmt" + "github.com/uber-go/tally" + basehook "github.com/uber/submitqueue/api/base/hook" runwaymq "github.com/uber/submitqueue/api/runway/messagequeue" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/extension/counter" + hookext "github.com/uber/submitqueue/platform/extension/hook" + platformhook "github.com/uber/submitqueue/platform/hook" "github.com/uber/submitqueue/platform/pipeline" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/extension/buildrunner" @@ -77,6 +82,10 @@ type Deps struct { // Validator resolves the validator for each queue. Validator validator.Factory + + // Hook receives lifecycle events for fire-and-forget side effects. Wire + // noop when the deployment has no integrations. + Hook hookext.Hook } // Stages is the orchestrator's pipeline topology as a typed table. @@ -211,6 +220,22 @@ var Stages = []pipeline.Stage[Deps]{ return dlq.NewDLQBatchController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil }, }, + // Any stage can publish here and this one publishes nothing onward, so the + // row sits outside the flow the rest of the table is ordered by. + { + Key: basehook.TopicKeyHook, + Name: "submitqueue-hook", + ConsumerGroup: "orchestrator", + New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + if d.Hook == nil { + return nil, fmt.Errorf("hook is required; wire noop when the deployment has no integrations") + } + return platformhook.NewDispatcher(d.Logger, d.Scope, d.Hook, sc.TopicKey, sc.ConsumerGroup), nil + }, + DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return platformhook.NewDLQController(d.Logger, d.Scope, sc.TopicKey, sc.ConsumerGroup), nil + }, + }, } // PublishOnlyTopics declares topics the orchestrator publishes to but does diff --git a/submitqueue/orchestrator/pipeline_test.go b/submitqueue/orchestrator/pipeline_test.go new file mode 100644 index 000000000..e523fb562 --- /dev/null +++ b/submitqueue/orchestrator/pipeline_test.go @@ -0,0 +1,73 @@ +// 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 orchestrator + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + basehook "github.com/uber/submitqueue/api/base/hook" + hooknoop "github.com/uber/submitqueue/platform/extension/hook/noop" + "github.com/uber/submitqueue/platform/pipeline" + "go.uber.org/zap/zaptest" +) + +func hookStage(t *testing.T) pipeline.Stage[Deps] { + t.Helper() + for _, s := range Stages { + if s.Key == basehook.TopicKeyHook { + return s + } + } + require.FailNow(t, "no stage is registered for the hook topic key") + return pipeline.Stage[Deps]{} +} + +func TestHookStage(t *testing.T) { + deps := func(t *testing.T) Deps { + return Deps{ + Logger: zaptest.NewLogger(t).Sugar(), + Scope: tally.NoopScope, + Hook: hooknoop.New(), + } + } + stageContext := func(key string) pipeline.StageContext { + return pipeline.StageContext{ + TopicKey: basehook.TopicKey(key), + ConsumerGroup: "orchestrator", + } + } + + t.Run("dispatcher subscribes to the key the engine assigns", func(t *testing.T) { + controller, err := hookStage(t).New(deps(t), stageContext("hook")) + require.NoError(t, err) + assert.Equal(t, basehook.TopicKeyHook, controller.TopicKey()) + }) + + t.Run("reconciler subscribes to the dead-letter key the engine derives", func(t *testing.T) { + controller, err := hookStage(t).DLQ(deps(t), stageContext("hook_dlq")) + require.NoError(t, err) + assert.Equal(t, basehook.TopicKey("hook_dlq"), controller.TopicKey()) + }) + + t.Run("a host that leaves the hook unwired fails to construct", func(t *testing.T) { + d := deps(t) + d.Hook = nil + _, err := hookStage(t).New(d, stageContext("hook")) + require.Error(t, err) + }) +}