From 7a82ea5fe519a352b033203a19307e922a1ac926 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Tue, 20 Oct 2026 15:48:55 +0000 Subject: [PATCH] feat(stovepipe): define repair coordination contracts and RFC --- Makefile | 1 + doc/rfc/stovepipe/fixer.md | 136 +++++++++++++ doc/rfc/stovepipe/workflow.md | 2 + stovepipe/core/messagequeue/README.md | 9 + stovepipe/core/messagequeue/messagequeue.go | 6 + .../core/messagequeue/messagequeue_test.go | 43 +++- stovepipe/core/messagequeue/proto/BUILD.bazel | 2 + stovepipe/core/messagequeue/proto/fix.proto | 37 ++++ .../core/messagequeue/proto/fixsignal.proto | 35 ++++ .../core/messagequeue/protopb/BUILD.bazel | 2 + stovepipe/core/messagequeue/protopb/fix.pb.go | 166 +++++++++++++++ .../core/messagequeue/protopb/fixsignal.pb.go | 153 ++++++++++++++ stovepipe/core/messagequeue/topics.go | 8 + stovepipe/entity/BUILD.bazel | 1 + stovepipe/entity/fix.go | 60 ++++++ stovepipe/entity/queue.go | 3 + stovepipe/extension/fixer/BUILD.bazel | 9 + stovepipe/extension/fixer/fixer.go | 96 +++++++++ stovepipe/extension/fixer/mock/BUILD.bazel | 13 ++ stovepipe/extension/fixer/mock/fixer_mock.go | 190 ++++++++++++++++++ stovepipe/extension/storage/BUILD.bazel | 1 + stovepipe/extension/storage/README.md | 10 + stovepipe/extension/storage/fix_store.go | 32 +++ stovepipe/extension/storage/mock/BUILD.bazel | 1 + .../extension/storage/mock/fix_store_mock.go | 85 ++++++++ tool/proto/BUILD.bazel | 2 + 26 files changed, 1102 insertions(+), 1 deletion(-) create mode 100644 doc/rfc/stovepipe/fixer.md create mode 100644 stovepipe/core/messagequeue/proto/fix.proto create mode 100644 stovepipe/core/messagequeue/proto/fixsignal.proto create mode 100644 stovepipe/core/messagequeue/protopb/fix.pb.go create mode 100644 stovepipe/core/messagequeue/protopb/fixsignal.pb.go create mode 100644 stovepipe/entity/fix.go create mode 100644 stovepipe/extension/fixer/BUILD.bazel create mode 100644 stovepipe/extension/fixer/fixer.go create mode 100644 stovepipe/extension/fixer/mock/BUILD.bazel create mode 100644 stovepipe/extension/fixer/mock/fixer_mock.go create mode 100644 stovepipe/extension/storage/fix_store.go create mode 100644 stovepipe/extension/storage/mock/fix_store_mock.go diff --git a/Makefile b/Makefile index 6699fbfe8..0cd9c83f5 100644 --- a/Makefile +++ b/Makefile @@ -629,6 +629,7 @@ GO_GENERATE_PACKAGES := \ ./stovepipe/controller/... \ ./stovepipe/core/requestlog/... \ ./stovepipe/extension/buildrunner/... \ + ./stovepipe/extension/fixer/... \ ./stovepipe/extension/projectstatus/... \ ./stovepipe/extension/queueconfig/... \ ./stovepipe/extension/sourcecontrol/... \ diff --git a/doc/rfc/stovepipe/fixer.md b/doc/rfc/stovepipe/fixer.md new file mode 100644 index 000000000..39498dfd4 --- /dev/null +++ b/doc/rfc/stovepipe/fixer.md @@ -0,0 +1,136 @@ +# Stovepipe breakage repair coordination + +Stovepipe accepts a broken check, decides whether to start a repair or attach it to existing work, and tracks progress until validation confirms recovery. A repair can cover several checks. The [workflow map](workflow.md#workflow) shows this work running alongside validation. + +`buildanalysis.Controller.Process` persists observations and manages failure episodes and recovery evidence. Each episode also stores its authoritative `FixID` assignment, updated by `fix`. The `fix` queue stage owns repair decisions, associations, and submission; `fixsignal` observes execution and resolves repairs. Both stages implement [`consumer.Controller`](../../../platform/consumer/controller.go) through `Process(context.Context, consumer.Delivery) error`. [`fixer.RepairEvaluator`](../../../stovepipe/extension/fixer/fixer.go) recommends Start, Attach, or Defer; `fixer.RepairExecutor` submits and observes work through a host-supplied implementation. + +**Implementation status:** the [queue messages and topic keys](../../../stovepipe/core/messagequeue/messagequeue.go), [extension interfaces](../../../stovepipe/extension/fixer/fixer.go), [repair entity](../../../stovepipe/entity/fix.go), and [FixStore contract](../../../stovepipe/extension/storage/fix_store.go) are defined. [`Queue.ActiveFixIDs`](../../../stovepipe/entity/queue.go) is part of the existing queue snapshot. Persistence implementations follow in the storage layer. `buildanalysis` and both fixer runtime controllers remain to be implemented. The episode requirements below define the integration needed from build analysis. + +## 1. Persist the failure episode in buildanalysis + +The controller reads available reports through `buildanalysis.FailedTargetReader`, whose contract is defined in [#898](https://github.com/uber/submitqueue/pull/898). A reader `Report.CheckID` identifies a CI check or shard within a build; `FailedTargets` contains the individual failed items. Map each item to a stable queue-scoped `BreakageRef.Key` including its validation predicate and relevant configuration. Retain the BuildID, CheckID, and AttemptID on persisted observations. The controller assigns episode generations; build attempts do not define new generations. + +After persisting observations, `buildanalysis.Controller.Process` creates or updates an episode identified by `(queue, check key, generation)` in the same delivery, then publishes [`FixRequest`](../../../stovepipe/core/messagequeue/proto/fix.proto) with that identity to `TopicKeyFix`. The same handoff applies when recovery or new evidence changes the episode. Persist before publishing; retry failed publication with the same message identity. + +### Required failure-episode contract + +The episode entity and store are part of the build-analysis implementation. They must support the following data and behavior; the field names below describe the required contract rather than an implemented Go type. + +| Field | Required content | +| --- | --- | +| `Queue` | Validation queue containing the check and its repairs. | +| `Key` | Stable, opaque check identity, including its predicate and configuration. Preserve case and trailing spaces. | +| `Generation` | Positive episode number. Recovery followed by another failure creates a new generation. Retain previous generations. | +| `Failure.BuildID` | Build containing the first positive failure evidence for this episode. Preserve this initial reference as later observations arrive. | +| `Failure.ObservationID` | Optional reference to persisted diagnostics within that build, when supplied by build analysis. | +| `Failure.URI` | Optional retained revision for provenance or ordering after build archival. | +| `Recovery` | Optional verified recovery evidence with a required validating `BuildID` and optional observation ID and revision URI. Absence means recovery is unverified. It can be recorded before a fix is assigned. | +| `FixID` | Current assigned logical repair ID, scoped to the same queue. Empty means unassigned. This is the authoritative repair assignment. | +| `Version` | Optimistic concurrency version covering evidence, recovery, and assignment. | + +The episode store must provide a primary-key read for any retained `(queue, key, generation)` and conditional updates that preserve unrelated fields. Build-analysis updates preserve `FixID`; fixer updates preserve failure and recovery evidence. Competing assignments use an expected version and expected current owner. Stale observations preserve verified recovery; recurrence creates a new generation. Diagnosis and root cause can emerge later and are resolved through extension-owned context. + +The episode reference matches [`entity.BreakageRef`](../../../stovepipe/entity/fix.go). The fixer stores that reference in `Fix.Breakages`; the build-analysis episode stores the assignment and evidence. This uses one episode record for both health and repair association. + +For failure-only reports, absence establishes recovery only after complete, comparable validation covering that item. Use the job configuration to establish coverage; the CI artifact can continue listing only failures. Reader completeness describes report availability, so retain coverage and configuration provenance alongside observations. A missing or partial report leaves recovery unverified. + +Persist current snapshots and reject superseded observations. Build and attempt provenance supports deduplication and freshness checks; opaque attempt IDs distinguish executions but do not establish their ordering. + +For provenance, follow `Fix.Breakages` to the retained episode, then `Failure.BuildID` to [`Build.RequestID`](../../../stovepipe/entity/build.go) and [`Request.URI`](../../../stovepipe/entity/request.go). If builds or requests can be archived before episodes and repairs, the episode must retain the original request ID and revision needed for that lookup. + +Apply queue policy to the reader's automatic retry disposition when deciding when to request repair. An available failed report may request repair before other checks complete; a pending automatic retry may defer that request. Persist positive failure evidence independently of that decision. + +Publish an updated `FixRequest` after recovery, assignment changes requiring follow-up, or evidence changes that warrant reevaluation. Reconcile missed or dead-lettered handoffs and revisit broken episodes that still need repair, so interrupted processing eventually resumes. + +## 2. Evaluate the episode against active repairs + +`fix.Controller.Process` reloads the failure episode. An unassigned, unrecovered episode proceeds to evaluation. An assigned episode resumes or signals its existing repair. Recovery before assignment resolves the pending request. + +Load the queue through [`QueueStore.Get`](../../../stovepipe/extension/storage/queue_store.go). Read each repair in `Queue.ActiveFixIDs` through [`FixStore.Get`](../../../stovepipe/extension/storage/fix_store.go), and identify compatible candidates. Compatibility covers the queue, tracked ref, configuration, and whether the repair can still cover another episode. A candidate-loading error leaves the request pending for normal retry. + +Call `fixer.RepairEvaluator.Evaluate` with the queue, episode reference, and candidate IDs. The implementation resolves diagnostics, investigation progress, and changes through injected dependencies, following the [`conflict.Analyzer`](../../../submitqueue/extension/conflict/conflict.go) pattern. + +| Decision | Meaning | Controller action | +| --- | --- | --- | +| `Start` | The evidence supports a new repair. | Persist a repair and its episode association, then submit it. | +| `Attach` | An existing repair is expected to cover the episode. | Revalidate the returned fix ID and associate the episode with it. | +| `Defer` | More evidence or investigation is needed. | Keep the request pending with `delivery.Hold(...)`. | + +For example, several checks can fail because an entity is missing one generated field. The evaluator can attach them to a repair adding that field, or defer while a plausible overlapping repair investigates. A bounded waiting policy escalates episodes that remain undecidable or permits independent work. + +## 3. Persist the repair and assignment + +Recheck the episode before applying a decision. An existing assignment takes precedence over a replayed evaluation; verified recovery resolves the pending request. + +For `Start`: + +1. Persist a queued `Fix` containing its initial `BreakageRef`. +2. Add its ID to `Queue.ActiveFixIDs` through `QueueStore.Update`. +3. Conditionally assign the episode's `FixID`. +4. Confirm that the episode remains assigned to this repair and still needs repair before submission. + +Queue updates use the existing `Queue.Version`. On a version conflict, reload the queue and merge the membership change while preserving pipeline and enablement fields. Copy slices before changing membership. The version guards the queue update, not the entire evaluation. Concurrent evaluations can start overlapping repairs; occasional duplicate repair attempts are an accepted tradeoff. + +For `Attach`, validate the selected repair's state and eligibility, persist the episode reference on the repair, and conditionally assign the episode to that fix. If the selected repair closes or the episode changes, reevaluate. If a worker needs new evidence, use the host's membership-refresh protocol before relying on coverage. + +The episode's `FixID` is authoritative. `Fix.Breakages` supplies navigation from a repair to its episodes and may temporarily contain a candidate association that did not win. Reload episode assignments when observing a repair and finish interrupted writes through replay. If another assignment wins, resolve unused queued work and remove its active ID. A crash before active membership is persisted can leave an unused, unsubmitted repair record; build-analysis reconciliation retries the still-unassigned episode. Once the repair is listed, reconciliation can discover it by queue key and finish or resolve its initial association. + +## 4. Trigger the repair + +`fix.Controller.Process` calls `fixer.RepairExecutor.Submit` after persisting the repair, active membership, and episode assignment. This follows the trigger-and-monitor split used by `build` and `buildsignal`. Service wiring selects the evaluator and executor for each queue. + +The executor returns an opaque execution ID and an error. A successful call returns a nonempty ID that `Observe` can use; an unsuccessful call follows normal error handling. + +1. Recheck episode ownership and recovery, then persist `FixSubmitting` before calling the executor. Clear the previous execution ID when starting a retry. Advance `Fix.Version` for each outbound attempt, including a retry already in the submitting state, and retain that version for the result write. +2. Call `RepairExecutor.Submit`. On success, conditionally persist the returned ID in `Fix.ExecutionRef` and transition to `FixRunning`, using the version recorded before the call. An empty ID with no error violates the executor contract and is handled as an error. +3. On error, ignore the returned ID and apply normal error classification and retry policy. A timeout or interrupted delivery can leave accepted work without a persisted ID, so that execution may continue without monitoring. Retrying can start duplicate work; occasional overlap is an accepted tradeoff. +4. After the execution ID is durable, publish [`FixSignal`](../../../stovepipe/core/messagequeue/proto/fixsignal.proto). A replay with an existing execution ID resumes monitoring and retries the handoff. + +Conditional writes protect newer state from delayed submission results. If the result write conflicts, reload the repair and resume from its current state. Do not merge the delayed result into the newer snapshot: it may belong to a superseded attempt. The existing lifecycle, execution ID, and snapshot version provide the durable state for this flow. + +## 5. Observe execution and check recovery + +`fixsignal.Controller.Process` reloads the repair and the failure episodes referenced by `Fix.Breakages`. When `Fix.ExecutionRef` is present, it calls `fixer.RepairExecutor.Observe` and persists execution progress. It uses each currently assigned episode's recovery evidence to determine repair resolution. Use `delivery.Hold(...)` while execution or verification is pending. + +`Observe` uses the persisted `Fix.ExecutionRef`. If `Snapshot.Found` is false, hold and poll according to the observation policy; absence alone is not a reason to submit another execution. The snapshot supplies execution state and optional metadata, including change URIs, native run IDs, landed revisions, diagnostics, and code-review status. Provider success advances execution progress; `buildanalysis` establishes recovery. A failed or ineffective execution can be retried within the same episode. Apply observations against the loaded repair version and recheck the execution ID after conflicts, so results from an older execution cannot overwrite its replacement. + +`buildanalysis` records recovery for the exact episode and publishes its `FixRequest`. The `fix` stage resolves an unassigned request or signals its assigned repair. Health recovery and execution completion remain separate: after all associated episodes recover, continue reconciliation until any known execution is settled or stopped. Retain the active ID while either verification or a known execution remains unresolved. Persist terminal repair state before removing the ID from `Queue.ActiveFixIDs`; a stale terminal entry is filtered during candidate discovery and removed by replay. + +## 6. Handle overlap discovered during execution + +A worker can discover overlap after repairs start. Persist a reassociation intent, then publish the affected episode's `FixRequest`. The `fix` stage validates expected assignment and repair versions, updates membership and the episode's `FixID`, and signals affected repairs. Preserve prior repair records and transfer only covered episodes when overlap is partial. + +Define how the worker acknowledges supersession and how publication ownership is enforced. Updating an episode assignment alone does not stop an external worker. Keep this protocol in the host integration. + +## Queue delivery and wiring + +| Topic | Message | Partition | Processing | +| --- | --- | --- | --- | +| `TopicKeyFix` (`fix`) | `FixRequest`: queue, check key, generation | Episode identity | Reload the failure episode; evaluate, persist assignment, and submit or signal a repair. | +| `TopicKeyFixSignal` (`fixsignal`) | `FixSignal`: queue, fix ID | Fix ID | Reload the repair and failure episodes; observe execution and check recovery. | + +Register both controllers in [service wiring](../../../service/stovepipe/server/main.go), following [`build`](../../../stovepipe/controller/build/build.go) and [`buildsignal`](../../../stovepipe/controller/buildsignal/buildsignal.go). Each exposes `Process`, `Name`, `TopicKey`, and `ConsumerGroup` through `consumer.Controller`. + +Return `nil` on completion, propagate failures through normal error classification, and call `Hold` then return `nil` while waiting. Reload persisted state on every delivery. Use stable message identities for each durable transition; recovery and new attempts need distinct identities. Persist state before publishing a dependent message, and reconcile failed or dead-lettered handoffs. + +## Data and storage + +| Record | Primary key | Contents | +| --- | --- | --- | +| `Fix`, through `FixStore` | `(queue, fix ID)` | Episode references, execution ID, lifecycle, metadata, version, and timestamps. | +| Existing `Queue`, through `QueueStore` | `(name)` | Existing pipeline coordination plus `ActiveFixIDs`, sharing `Queue.Version`. | +| Failure episode, managed by buildanalysis | `(queue, check key, generation)` | Initial failure provenance, verified recovery, assigned `FixID`, and version. | + +All access uses primary keys. Candidate discovery reads `Queue.ActiveFixIDs` and then individual fixes. Keep the active set bounded; detailed history remains in repair and failure-episode records. + +`fix.metadata` follows the existing [request-log metadata pattern](../../../stovepipe/entity/request_log.go). Nil and empty metadata maps are equivalent and persist as `{}`. Nil and empty `Queue.ActiveFixIDs` both persist as `[]` and read as nil. Conditional updates replace complete snapshots. Every queue writer preserves the active IDs alongside existing coordination fields. Queue updates can conflict with ingestion or validation updates and retry using the latest full snapshot. + +For an existing database, add and backfill `queue.active_fix_ids` with an empty JSON array before running the new queue-store code. The storage implementation supplies schemas for fresh databases. + +## Implementation + +1. **Implement persistence and episode tracking.** Implement `FixStore` and persist `Queue.ActiveFixIDs`. In buildanalysis, supply the episode data and primary-key operations listed above, preserve historical generations and provenance, and publish/reconcile `FixRequest` handoffs. +2. **Implement submission transitions.** Persist the submitting state before dispatch and the execution ID after success; use snapshot versions to reject stale results. Define retry policy and replayable reassociation before enabling execution. +3. **Implement the queue stages.** Add `fix.Controller.Process` and `fixsignal.Controller.Process`, queue membership updates, episode assignment, bounded deferral, and reconciliation; register both consumers. +4. **Implement extensions.** Supply `RepairEvaluator` and `RepairExecutor`, including evidence refresh, observation correlation, and overlap handling. Keep transport adapters in host integration. +5. **Verify the lifecycle.** Cover concurrent membership updates, duplicate deliveries, interrupted associations, recovery before submission, candidate closure, ambiguous submission, provider retries, late overlap, partial recovery, and repeated failure episodes. diff --git a/doc/rfc/stovepipe/workflow.md b/doc/rfc/stovepipe/workflow.md index a6fcbc518..7e39f7af1 100644 --- a/doc/rfc/stovepipe/workflow.md +++ b/doc/rfc/stovepipe/workflow.md @@ -189,6 +189,8 @@ Per-stage design detail lives under `steps/` so this doc stays a pipeline overvi - [buildsignal.md](steps/buildsignal.md) — the poll loop: hold-based re-poll cadence, per-build partitioning, and the fail-closed handoff to record - [record.md](steps/record.md) — turning a terminal build outcome into an immutable validation fact, monotonic last-green advancement and ref promotion, and the hook event announcing the outcome. Project analysis already runs inline through `projectstatus.Resolver` and may later use results calculated by build analysis +For repair decisions, execution, and recovery, see the [repair RFC](fixer.md). + ## Dedup, idempotency, and history rewrites Ingestion is idempotent on `(Queue, head URI)`, so duplicate poller reports — and any future webhook producer racing the poller — converge on one Request. The pipeline persists the Queue's last-green URI and per-URI greenness, so it must tolerate a **history rewrite**: when `SourceControl` reports that the last-green URI is no longer an ancestor of the current head, `process` falls back to a full-monorepo build rather than trusting a baseline that no longer exists on the branch. The system converges to a correct greenness rather than wedging on a stale pointer. diff --git a/stovepipe/core/messagequeue/README.md b/stovepipe/core/messagequeue/README.md index e0c437c7a..b36fd735e 100644 --- a/stovepipe/core/messagequeue/README.md +++ b/stovepipe/core/messagequeue/README.md @@ -10,3 +10,12 @@ Payloads are defined in proto3 (`proto/`, generated into `protopb/`) and seriali - **build** (`TopicKeyBuild`, `BuildRequest`) — process publishes the request id here once its build scope (`BuildStrategy`/`BaseURI`) is decided; the build controller reloads the `Request`, triggers the build, and persists the resulting `Build`. Partitioned by request id. - **buildsignal** (`TopicKeyBuildSignal`, `BuildSignal`) — build publishes the build id here after triggering; buildsignal holds the delivery between polls until the build reaches a terminal status. Partitioned by build id, so each build's poll loop is an independent partition. - **record** (`TopicKeyRecord`, `Record`) — buildsignal publishes the request id after persisting the terminal outcome; record reloads the request and records its validation fact. Partitioned by request id. + +## Repair stage contracts + +The repair messages and topic keys are defined for the [repair design](../../../doc/rfc/stovepipe/fixer.md). Controller implementations and runtime registration follow in the implementation layers. Both stages use the existing `consumer.Controller` interface. + +- **fix** (`TopicKeyFix`, [`FixRequest`](proto/fix.proto)) — `buildanalysis.Controller.Process` persists an episode's failure or recovery evidence, then publishes its queue, check key, and generation. Processing reloads the episode, evaluates Start/Attach/Defer, persists repair membership and episode assignment, and submits new work through `RepairExecutor.Submit` or signals the existing repair. Partitioned by episode identity. +- **fixsignal** (`TopicKeyFixSignal`, [`FixSignal`](proto/fixsignal.proto)) — fix processing publishes the repair ID after persisting the returned execution ID, or receiving recovery evidence for an assigned episode. Processing reloads the repair and its failure episodes to call `RepairExecutor.Observe` and check recovery. Partitioned by repair ID, with held deliveries between observations. + +Each durable episode update has a distinct publication intent; retries reuse that intent. A recovery update re-enters through `fix`, so processing can resolve either a pending request or an already assigned repair using the current stored ownership. diff --git a/stovepipe/core/messagequeue/messagequeue.go b/stovepipe/core/messagequeue/messagequeue.go index 867eb3aaa..293e4717e 100644 --- a/stovepipe/core/messagequeue/messagequeue.go +++ b/stovepipe/core/messagequeue/messagequeue.go @@ -53,6 +53,12 @@ type ( // Record is the payload buildsignal publishes to the record stage once a // build reaches a terminal status: the build id to record. Record = protopb.Record + + // FixRequest identifies a persisted failure episode whose assignment needs processing. + FixRequest = protopb.FixRequest + + // FixSignal identifies a persisted repair whose execution and recovery need processing. + FixSignal = protopb.FixSignal ) // marshalOpts keeps the JSON field names identical to the proto field names diff --git a/stovepipe/core/messagequeue/messagequeue_test.go b/stovepipe/core/messagequeue/messagequeue_test.go index e0d0aa88d..b83efb51b 100644 --- a/stovepipe/core/messagequeue/messagequeue_test.go +++ b/stovepipe/core/messagequeue/messagequeue_test.go @@ -66,6 +66,45 @@ func TestRecordRoundTrip(t *testing.T) { assert.True(t, proto.Equal(rec, got), "round-tripped Record should equal the original") } +func TestFixMessagesRoundTrip(t *testing.T) { + for _, tt := range []struct { + name string + message proto.Message + }{ + { + name: "episode request", + message: &FixRequest{ + QueueName: "repo/main", + Key: "checks/package:compile", + Generation: 9007199254740993, + }, + }, + { + name: "repair signal", + message: &FixSignal{QueueName: "repo/main", Id: "repair-42"}, + }, + } { + t.Run(tt.name, func(t *testing.T) { + data, err := Marshal(tt.message) + require.NoError(t, err) + got := tt.message.ProtoReflect().New().Interface() + require.NoError(t, Unmarshal(data, got)) + assert.True(t, proto.Equal(tt.message, got)) + }) + } +} + +func TestFixRequestWireFormat(t *testing.T) { + message := &FixRequest{QueueName: "repo/main", Key: "check-a", Generation: 9007199254740993} + data, err := Marshal(message) + require.NoError(t, err) + assert.JSONEq(t, `{"queue_name":"repo/main","key":"check-a","generation":"9007199254740993"}`, string(data)) + + var got FixRequest + require.NoError(t, Unmarshal([]byte(`{"queue_name":"repo/main","key":"check-a","generation":"9007199254740993","future_field":true}`), &got)) + assert.True(t, proto.Equal(message, &got)) +} + // TestWireFormat locks the protojson encoding decision the contract relies on: // snake_case field names (UseProtoNames). func TestWireFormat(t *testing.T) { @@ -80,7 +119,7 @@ func TestWireFormat(t *testing.T) { // no topic_keys option names an unknown key. func TestTopicKeysBindEveryTopicKey(t *testing.T) { bound := map[string]int{} - for _, m := range []proto.Message{&ProcessRequest{}, &BuildRequest{}, &BuildSignal{}, &Record{}} { + for _, m := range []proto.Message{&ProcessRequest{}, &BuildRequest{}, &BuildSignal{}, &Record{}, &FixRequest{}, &FixSignal{}} { keys := TopicKeys(m) require.NotEmpty(t, keys, "message must declare a non-empty topic_keys option") for _, key := range keys { @@ -93,6 +132,8 @@ func TestTopicKeysBindEveryTopicKey(t *testing.T) { TopicKeyBuild, TopicKeyBuildSignal, TopicKeyRecord, + TopicKeyFix, + TopicKeyFixSignal, } valid := map[string]bool{} diff --git a/stovepipe/core/messagequeue/proto/BUILD.bazel b/stovepipe/core/messagequeue/proto/BUILD.bazel index 5f2390a8a..fc7484e98 100644 --- a/stovepipe/core/messagequeue/proto/BUILD.bazel +++ b/stovepipe/core/messagequeue/proto/BUILD.bazel @@ -2,6 +2,8 @@ exports_files( [ "build.proto", "buildsignal.proto", + "fix.proto", + "fixsignal.proto", "process.proto", "record.proto", ], diff --git a/stovepipe/core/messagequeue/proto/fix.proto b/stovepipe/core/messagequeue/proto/fix.proto new file mode 100644 index 000000000..6431a2f55 --- /dev/null +++ b/stovepipe/core/messagequeue/proto/fix.proto @@ -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. + +syntax = "proto3"; + +package uber.stovepipe.messagequeue; + +import "api/base/messagequeue/proto/messagequeue.proto"; + +option go_package = "github.com/uber/submitqueue/stovepipe/core/messagequeue/protopb"; +option java_multiple_files = true; +option java_outer_classname = "FixRequestProto"; +option java_package = "com.uber.submitqueue.stovepipe.messagequeue"; + +// FixRequest references a durable episode request; evidence and assignment are loaded from storage. +// Publishing requires the episode state to have been persisted first. +message FixRequest { + option (uber.base.messagequeue.topic_keys) = "fix"; + + // queue_name identifies the validation queue containing the episode. + string queue_name = 1; + // key is the opaque check identity within the queue. + string key = 2; + // generation distinguishes a recurrence after recovery from the previous failure episode. + int64 generation = 3; +} diff --git a/stovepipe/core/messagequeue/proto/fixsignal.proto b/stovepipe/core/messagequeue/proto/fixsignal.proto new file mode 100644 index 000000000..1e15ee984 --- /dev/null +++ b/stovepipe/core/messagequeue/proto/fixsignal.proto @@ -0,0 +1,35 @@ +// 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.stovepipe.messagequeue; + +import "api/base/messagequeue/proto/messagequeue.proto"; + +option go_package = "github.com/uber/submitqueue/stovepipe/core/messagequeue/protopb"; +option java_multiple_files = true; +option java_outer_classname = "FixSignalProto"; +option java_package = "com.uber.submitqueue.stovepipe.messagequeue"; + +// FixSignal references a durable repair; execution and episode recovery are loaded from storage. +// Publishing requires the repair and its episode associations to have been persisted first. +message FixSignal { + option (uber.base.messagequeue.topic_keys) = "fixsignal"; + + // id is the queue-scoped repair identifier. + string id = 1; + // queue_name identifies the validation queue containing the repair. + string queue_name = 2; +} diff --git a/stovepipe/core/messagequeue/protopb/BUILD.bazel b/stovepipe/core/messagequeue/protopb/BUILD.bazel index 7f61fb89a..d609129e7 100644 --- a/stovepipe/core/messagequeue/protopb/BUILD.bazel +++ b/stovepipe/core/messagequeue/protopb/BUILD.bazel @@ -5,6 +5,8 @@ go_library( srcs = [ "build.pb.go", "buildsignal.pb.go", + "fix.pb.go", + "fixsignal.pb.go", "process.pb.go", "record.pb.go", ], diff --git a/stovepipe/core/messagequeue/protopb/fix.pb.go b/stovepipe/core/messagequeue/protopb/fix.pb.go new file mode 100644 index 000000000..7cd27a98e --- /dev/null +++ b/stovepipe/core/messagequeue/protopb/fix.pb.go @@ -0,0 +1,166 @@ +// 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.11 +// protoc v5.29.3 +// source: fix.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" +) + +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) +) + +// FixRequest references a durable episode request; evidence and assignment are loaded from storage. +// Publishing requires the episode state to have been persisted first. +type FixRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + // queue_name identifies the validation queue containing the episode. + QueueName string `protobuf:"bytes,1,opt,name=queue_name,json=queueName,proto3" json:"queue_name,omitempty"` + // key is the opaque check identity within the queue. + Key string `protobuf:"bytes,2,opt,name=key,proto3" json:"key,omitempty"` + // generation distinguishes a recurrence after recovery from the previous failure episode. + Generation int64 `protobuf:"varint,3,opt,name=generation,proto3" json:"generation,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *FixRequest) Reset() { + *x = FixRequest{} + mi := &file_fix_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *FixRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*FixRequest) ProtoMessage() {} + +func (x *FixRequest) ProtoReflect() protoreflect.Message { + mi := &file_fix_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 FixRequest.ProtoReflect.Descriptor instead. +func (*FixRequest) Descriptor() ([]byte, []int) { + return file_fix_proto_rawDescGZIP(), []int{0} +} + +func (x *FixRequest) GetQueueName() string { + if x != nil { + return x.QueueName + } + return "" +} + +func (x *FixRequest) GetKey() string { + if x != nil { + return x.Key + } + return "" +} + +func (x *FixRequest) GetGeneration() int64 { + if x != nil { + return x.Generation + } + return 0 +} + +var File_fix_proto protoreflect.FileDescriptor + +const file_fix_proto_rawDesc = "" + + "\n" + + "\tfix.proto\x12\x1buber.stovepipe.messagequeue\x1a.api/base/messagequeue/proto/messagequeue.proto\"f\n" + + "\n" + + "FixRequest\x12\x1d\n" + + "\n" + + "queue_name\x18\x01 \x01(\tR\tqueueName\x12\x10\n" + + "\x03key\x18\x02 \x01(\tR\x03key\x12\x1e\n" + + "\n" + + "generation\x18\x03 \x01(\x03R\n" + + "generation:\a\x8a\xb5\x18\x03fixB\x81\x01\n" + + "+com.uber.submitqueue.stovepipe.messagequeueB\x0fFixRequestProtoP\x01Z?github.com/uber/submitqueue/stovepipe/core/messagequeue/protopbb\x06proto3" + +var ( + file_fix_proto_rawDescOnce sync.Once + file_fix_proto_rawDescData []byte +) + +func file_fix_proto_rawDescGZIP() []byte { + file_fix_proto_rawDescOnce.Do(func() { + file_fix_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_fix_proto_rawDesc), len(file_fix_proto_rawDesc))) + }) + return file_fix_proto_rawDescData +} + +var file_fix_proto_msgTypes = make([]protoimpl.MessageInfo, 1) +var file_fix_proto_goTypes = []any{ + (*FixRequest)(nil), // 0: uber.stovepipe.messagequeue.FixRequest +} +var file_fix_proto_depIdxs = []int32{ + 0, // [0:0] is the sub-list for method output_type + 0, // [0:0] is the sub-list for method input_type + 0, // [0:0] is the sub-list for extension type_name + 0, // [0:0] is the sub-list for extension extendee + 0, // [0:0] is the sub-list for field type_name +} + +func init() { file_fix_proto_init() } +func file_fix_proto_init() { + if File_fix_proto != nil { + return + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_fix_proto_rawDesc), len(file_fix_proto_rawDesc)), + NumEnums: 0, + NumMessages: 1, + NumExtensions: 0, + NumServices: 0, + }, + GoTypes: file_fix_proto_goTypes, + DependencyIndexes: file_fix_proto_depIdxs, + MessageInfos: file_fix_proto_msgTypes, + }.Build() + File_fix_proto = out.File + file_fix_proto_goTypes = nil + file_fix_proto_depIdxs = nil +} diff --git a/stovepipe/core/messagequeue/protopb/fixsignal.pb.go b/stovepipe/core/messagequeue/protopb/fixsignal.pb.go new file mode 100644 index 000000000..4cbdf7484 --- /dev/null +++ b/stovepipe/core/messagequeue/protopb/fixsignal.pb.go @@ -0,0 +1,153 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.36.11 +// protoc v5.29.3 +// source: fixsignal.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" +) + +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) +) + +// FixSignal references a durable repair; execution and episode recovery are loaded from storage. +// Publishing requires the repair and its episode associations to have been persisted first. +type FixSignal struct { + state protoimpl.MessageState `protogen:"open.v1"` + // id is the queue-scoped repair identifier. + Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"` + // queue_name identifies the validation queue containing the repair. + QueueName string `protobuf:"bytes,2,opt,name=queue_name,json=queueName,proto3" json:"queue_name,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *FixSignal) Reset() { + *x = FixSignal{} + mi := &file_fixsignal_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *FixSignal) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*FixSignal) ProtoMessage() {} + +func (x *FixSignal) ProtoReflect() protoreflect.Message { + mi := &file_fixsignal_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 FixSignal.ProtoReflect.Descriptor instead. +func (*FixSignal) Descriptor() ([]byte, []int) { + return file_fixsignal_proto_rawDescGZIP(), []int{0} +} + +func (x *FixSignal) GetId() string { + if x != nil { + return x.Id + } + return "" +} + +func (x *FixSignal) GetQueueName() string { + if x != nil { + return x.QueueName + } + return "" +} + +var File_fixsignal_proto protoreflect.FileDescriptor + +const file_fixsignal_proto_rawDesc = "" + + "\n" + + "\x0ffixsignal.proto\x12\x1buber.stovepipe.messagequeue\x1a.api/base/messagequeue/proto/messagequeue.proto\"I\n" + + "\tFixSignal\x12\x0e\n" + + "\x02id\x18\x01 \x01(\tR\x02id\x12\x1d\n" + + "\n" + + "queue_name\x18\x02 \x01(\tR\tqueueName:\r\x8a\xb5\x18\tfixsignalB\x80\x01\n" + + "+com.uber.submitqueue.stovepipe.messagequeueB\x0eFixSignalProtoP\x01Z?github.com/uber/submitqueue/stovepipe/core/messagequeue/protopbb\x06proto3" + +var ( + file_fixsignal_proto_rawDescOnce sync.Once + file_fixsignal_proto_rawDescData []byte +) + +func file_fixsignal_proto_rawDescGZIP() []byte { + file_fixsignal_proto_rawDescOnce.Do(func() { + file_fixsignal_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_fixsignal_proto_rawDesc), len(file_fixsignal_proto_rawDesc))) + }) + return file_fixsignal_proto_rawDescData +} + +var file_fixsignal_proto_msgTypes = make([]protoimpl.MessageInfo, 1) +var file_fixsignal_proto_goTypes = []any{ + (*FixSignal)(nil), // 0: uber.stovepipe.messagequeue.FixSignal +} +var file_fixsignal_proto_depIdxs = []int32{ + 0, // [0:0] is the sub-list for method output_type + 0, // [0:0] is the sub-list for method input_type + 0, // [0:0] is the sub-list for extension type_name + 0, // [0:0] is the sub-list for extension extendee + 0, // [0:0] is the sub-list for field type_name +} + +func init() { file_fixsignal_proto_init() } +func file_fixsignal_proto_init() { + if File_fixsignal_proto != nil { + return + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_fixsignal_proto_rawDesc), len(file_fixsignal_proto_rawDesc)), + NumEnums: 0, + NumMessages: 1, + NumExtensions: 0, + NumServices: 0, + }, + GoTypes: file_fixsignal_proto_goTypes, + DependencyIndexes: file_fixsignal_proto_depIdxs, + MessageInfos: file_fixsignal_proto_msgTypes, + }.Build() + File_fixsignal_proto = out.File + file_fixsignal_proto_goTypes = nil + file_fixsignal_proto_depIdxs = nil +} diff --git a/stovepipe/core/messagequeue/topics.go b/stovepipe/core/messagequeue/topics.go index bac3f71d9..df4418832 100644 --- a/stovepipe/core/messagequeue/topics.go +++ b/stovepipe/core/messagequeue/topics.go @@ -48,4 +48,12 @@ const ( // terminal status; non-terminal polls never publish here. Partitioned by // request id. TopicKeyRecord TopicKey = "record" + + // TopicKeyFix carries episode references for repair assignment and recovery handoff. + // Its consumer reloads the episode; deliveries are partitioned by episode identity. + TopicKeyFix TopicKey = "fix" + + // TopicKeyFixSignal carries repair IDs for execution and recovery reconciliation. + // Deliveries are partitioned by fix ID and held between observations while work is active. + TopicKeyFixSignal TopicKey = "fixsignal" ) diff --git a/stovepipe/entity/BUILD.bazel b/stovepipe/entity/BUILD.bazel index 3b255e96b..e32f0117d 100644 --- a/stovepipe/entity/BUILD.bazel +++ b/stovepipe/entity/BUILD.bazel @@ -4,6 +4,7 @@ go_library( name = "go_default_library", srcs = [ "build.go", + "fix.go", "ingest.go", "list.go", "project_status.go", diff --git a/stovepipe/entity/fix.go b/stovepipe/entity/fix.go new file mode 100644 index 000000000..8f16e1f97 --- /dev/null +++ b/stovepipe/entity/fix.go @@ -0,0 +1,60 @@ +// Copyright (c) 2025 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 entity + +// FixState is a repair's durable lifecycle, independent of check health. +type FixState string + +const ( + FixStateUnknown FixState = "" + FixQueued FixState = "queued" + FixSubmitting FixState = "submitting" + FixRunning FixState = "running" + FixAwaitingVerification FixState = "awaiting_verification" + FixCompleted FixState = "completed" + FixFailed FixState = "failed" +) + +func (s FixState) Terminal() bool { return s == FixCompleted || s == FixFailed } + +// BreakageRef identifies a breakage episode within a queue. +type BreakageRef struct { + // Key is the stable check identity. + Key string `json:"key"` + // Generation identifies the breakage episode. + Generation int64 `json:"generation"` +} + +// Fix is a versioned logical repair covering one or more breakages. +type Fix struct { + // ID is the queue-scoped repair identifier. + ID string `json:"id"` + // Queue is the owning validation queue. + Queue string `json:"queue"` + // State is the repair lifecycle. + State FixState `json:"state"` + // Breakages contains episode references; a reference alone does not establish current assignment. + Breakages []BreakageRef `json:"breakages"` + // ExecutionRef is the opaque execution ID returned by submission, empty until known. + ExecutionRef string `json:"execution_ref"` + // Metadata contains optional change references and provider context; nil and empty maps are equivalent. + Metadata map[string]string `json:"metadata"` + // Version is the optimistic storage version. + Version int32 `json:"version"` + // CreatedAt is the UTC creation time in milliseconds since the epoch. + CreatedAt int64 `json:"created_at"` + // UpdatedAt is the UTC update time in milliseconds since the epoch. + UpdatedAt int64 `json:"updated_at"` +} diff --git a/stovepipe/entity/queue.go b/stovepipe/entity/queue.go index 9ca1c8fcc..2c19e070a 100644 --- a/stovepipe/entity/queue.go +++ b/stovepipe/entity/queue.go @@ -43,6 +43,9 @@ type Queue struct { // Empty until the first request is created. Coalescing compares IDs via CompareRequestID. LatestRequestID string `json:"latest_request_id"` + // ActiveFixIDs contains unique IDs of active repairs within the queue; nil and empty are equivalent. + ActiveFixIDs []string `json:"active_fix_ids"` + // Enablement is the applied enabled/disabled state; empty means unregistered. Enablement QueueEnablement `json:"enablement"` diff --git a/stovepipe/extension/fixer/BUILD.bazel b/stovepipe/extension/fixer/BUILD.bazel new file mode 100644 index 000000000..a6af87006 --- /dev/null +++ b/stovepipe/extension/fixer/BUILD.bazel @@ -0,0 +1,9 @@ +load("@rules_go//go:def.bzl", "go_library") + +go_library( + name = "go_default_library", + srcs = ["fixer.go"], + importpath = "github.com/uber/submitqueue/stovepipe/extension/fixer", + visibility = ["//visibility:public"], + deps = ["//stovepipe/entity:go_default_library"], +) diff --git a/stovepipe/extension/fixer/fixer.go b/stovepipe/extension/fixer/fixer.go new file mode 100644 index 000000000..d6eeff8d5 --- /dev/null +++ b/stovepipe/extension/fixer/fixer.go @@ -0,0 +1,96 @@ +// Copyright (c) 2025 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 fixer defines repair evaluation and asynchronous execution. +package fixer + +//go:generate mockgen -source=fixer.go -destination=mock/fixer_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/stovepipe/entity" +) + +// Action is an evaluator's recommendation for a failure episode. +type Action string + +const ( + ActionUnknown Action = "" + Start Action = "start" + Attach Action = "attach" + Defer Action = "defer" +) + +// RepairEvaluator recommends how to handle an unassigned failure episode. +// Implementations resolve diagnostics and investigation progress through injected dependencies. +type RepairEvaluator interface { + Evaluate(context.Context, EvaluationRequest) (Decision, error) +} + +// EvaluationRequest identifies an episode and the active repairs considered for attachment. +type EvaluationRequest struct { + // Queue is the validation queue containing the episode and candidate repairs. + Queue string + // Breakage identifies the persisted failure episode. + Breakage entity.BreakageRef + // CandidateFixIDs contains active repair IDs from the queue snapshot. + CandidateFixIDs []string +} + +// Decision is a recommendation; durable assignment requires a conditional state update. +type Decision struct { + // Action is Start, Attach, or Defer. + Action Action + // FixID is a candidate repair ID required for Attach and empty for other actions. + FixID string + // Detail is an optional explanation of the recommendation. + Detail string +} + +// EvaluatorFactory resolves an evaluator for a validation queue. +type EvaluatorFactory interface { + For(Config) (RepairEvaluator, error) +} + +// RepairExecutor submits repairs and observes their execution and change lifecycle. +// Submit returns a nonempty execution ID on success; its ID is ignored on error. +// An error may follow acceptance, so retrying can start duplicate work. +// Observe requires Fix.ExecutionRef to contain a previously returned execution ID. +type RepairExecutor interface { + Submit(context.Context, entity.Fix) (string, error) + Observe(context.Context, entity.Fix) (Snapshot, error) +} + +// Config identifies the queue whose repair extension is being resolved. +type Config struct { + // QueueName is the validation queue. + QueueName string +} + +// Factory resolves configured repair executors. +type Factory interface { + For(Config) (RepairExecutor, error) +} + +// Snapshot is provider evidence about a repair. +type Snapshot struct { + // Found reports whether Fix.ExecutionRef is visible; false leaves execution unresolved. + Found bool + // State is running, awaiting_verification, or failed. + // Completion is established separately by trusted recovery evidence. + State entity.FixState + // Metadata contains optional change references, provider details, and diagnostics. + Metadata map[string]string +} diff --git a/stovepipe/extension/fixer/mock/BUILD.bazel b/stovepipe/extension/fixer/mock/BUILD.bazel new file mode 100644 index 000000000..5727675f3 --- /dev/null +++ b/stovepipe/extension/fixer/mock/BUILD.bazel @@ -0,0 +1,13 @@ +load("@rules_go//go:def.bzl", "go_library") + +go_library( + name = "go_default_library", + srcs = ["fixer_mock.go"], + importpath = "github.com/uber/submitqueue/stovepipe/extension/fixer/mock", + visibility = ["//visibility:public"], + deps = [ + "//stovepipe/entity:go_default_library", + "//stovepipe/extension/fixer:go_default_library", + "@org_uber_go_mock//gomock:go_default_library", + ], +) diff --git a/stovepipe/extension/fixer/mock/fixer_mock.go b/stovepipe/extension/fixer/mock/fixer_mock.go new file mode 100644 index 000000000..42feeaea2 --- /dev/null +++ b/stovepipe/extension/fixer/mock/fixer_mock.go @@ -0,0 +1,190 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: fixer.go +// +// Generated by this command: +// +// mockgen -source=fixer.go -destination=mock/fixer_mock.go -package=mock +// + +// Package mock is a generated GoMock package. +package mock + +import ( + context "context" + reflect "reflect" + + entity "github.com/uber/submitqueue/stovepipe/entity" + fixer "github.com/uber/submitqueue/stovepipe/extension/fixer" + gomock "go.uber.org/mock/gomock" +) + +// MockRepairEvaluator is a mock of RepairEvaluator interface. +type MockRepairEvaluator struct { + ctrl *gomock.Controller + recorder *MockRepairEvaluatorMockRecorder + isgomock struct{} +} + +// MockRepairEvaluatorMockRecorder is the mock recorder for MockRepairEvaluator. +type MockRepairEvaluatorMockRecorder struct { + mock *MockRepairEvaluator +} + +// NewMockRepairEvaluator creates a new mock instance. +func NewMockRepairEvaluator(ctrl *gomock.Controller) *MockRepairEvaluator { + mock := &MockRepairEvaluator{ctrl: ctrl} + mock.recorder = &MockRepairEvaluatorMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockRepairEvaluator) EXPECT() *MockRepairEvaluatorMockRecorder { + return m.recorder +} + +// Evaluate mocks base method. +func (m *MockRepairEvaluator) Evaluate(arg0 context.Context, arg1 fixer.EvaluationRequest) (fixer.Decision, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Evaluate", arg0, arg1) + ret0, _ := ret[0].(fixer.Decision) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Evaluate indicates an expected call of Evaluate. +func (mr *MockRepairEvaluatorMockRecorder) Evaluate(arg0, arg1 any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Evaluate", reflect.TypeOf((*MockRepairEvaluator)(nil).Evaluate), arg0, arg1) +} + +// MockEvaluatorFactory is a mock of EvaluatorFactory interface. +type MockEvaluatorFactory struct { + ctrl *gomock.Controller + recorder *MockEvaluatorFactoryMockRecorder + isgomock struct{} +} + +// MockEvaluatorFactoryMockRecorder is the mock recorder for MockEvaluatorFactory. +type MockEvaluatorFactoryMockRecorder struct { + mock *MockEvaluatorFactory +} + +// NewMockEvaluatorFactory creates a new mock instance. +func NewMockEvaluatorFactory(ctrl *gomock.Controller) *MockEvaluatorFactory { + mock := &MockEvaluatorFactory{ctrl: ctrl} + mock.recorder = &MockEvaluatorFactoryMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockEvaluatorFactory) EXPECT() *MockEvaluatorFactoryMockRecorder { + return m.recorder +} + +// For mocks base method. +func (m *MockEvaluatorFactory) For(arg0 fixer.Config) (fixer.RepairEvaluator, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "For", arg0) + ret0, _ := ret[0].(fixer.RepairEvaluator) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// For indicates an expected call of For. +func (mr *MockEvaluatorFactoryMockRecorder) For(arg0 any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "For", reflect.TypeOf((*MockEvaluatorFactory)(nil).For), arg0) +} + +// MockRepairExecutor is a mock of RepairExecutor interface. +type MockRepairExecutor struct { + ctrl *gomock.Controller + recorder *MockRepairExecutorMockRecorder + isgomock struct{} +} + +// MockRepairExecutorMockRecorder is the mock recorder for MockRepairExecutor. +type MockRepairExecutorMockRecorder struct { + mock *MockRepairExecutor +} + +// NewMockRepairExecutor creates a new mock instance. +func NewMockRepairExecutor(ctrl *gomock.Controller) *MockRepairExecutor { + mock := &MockRepairExecutor{ctrl: ctrl} + mock.recorder = &MockRepairExecutorMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockRepairExecutor) EXPECT() *MockRepairExecutorMockRecorder { + return m.recorder +} + +// Observe mocks base method. +func (m *MockRepairExecutor) Observe(arg0 context.Context, arg1 entity.Fix) (fixer.Snapshot, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Observe", arg0, arg1) + ret0, _ := ret[0].(fixer.Snapshot) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Observe indicates an expected call of Observe. +func (mr *MockRepairExecutorMockRecorder) Observe(arg0, arg1 any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Observe", reflect.TypeOf((*MockRepairExecutor)(nil).Observe), arg0, arg1) +} + +// Submit mocks base method. +func (m *MockRepairExecutor) Submit(arg0 context.Context, arg1 entity.Fix) (string, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Submit", arg0, arg1) + ret0, _ := ret[0].(string) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Submit indicates an expected call of Submit. +func (mr *MockRepairExecutorMockRecorder) Submit(arg0, arg1 any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Submit", reflect.TypeOf((*MockRepairExecutor)(nil).Submit), arg0, arg1) +} + +// MockFactory is a mock of Factory interface. +type MockFactory struct { + ctrl *gomock.Controller + recorder *MockFactoryMockRecorder + isgomock struct{} +} + +// MockFactoryMockRecorder is the mock recorder for MockFactory. +type MockFactoryMockRecorder struct { + mock *MockFactory +} + +// NewMockFactory creates a new mock instance. +func NewMockFactory(ctrl *gomock.Controller) *MockFactory { + mock := &MockFactory{ctrl: ctrl} + mock.recorder = &MockFactoryMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockFactory) EXPECT() *MockFactoryMockRecorder { + return m.recorder +} + +// For mocks base method. +func (m *MockFactory) For(arg0 fixer.Config) (fixer.RepairExecutor, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "For", arg0) + ret0, _ := ret[0].(fixer.RepairExecutor) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// For indicates an expected call of For. +func (mr *MockFactoryMockRecorder) For(arg0 any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "For", reflect.TypeOf((*MockFactory)(nil).For), arg0) +} diff --git a/stovepipe/extension/storage/BUILD.bazel b/stovepipe/extension/storage/BUILD.bazel index fc12fdf84..7b56efa05 100644 --- a/stovepipe/extension/storage/BUILD.bazel +++ b/stovepipe/extension/storage/BUILD.bazel @@ -4,6 +4,7 @@ go_library( name = "go_default_library", srcs = [ "build_store.go", + "fix_store.go", "queue_log_store.go", "queue_store.go", "queue_summary_store.go", diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md index de79c8959..642695aa5 100644 --- a/stovepipe/extension/storage/README.md +++ b/stovepipe/extension/storage/README.md @@ -23,3 +23,13 @@ Same design space as [`submitqueue/extension/storage`](../../../submitqueue/exte `RequestAcceptanceStore` is the ordered lookup mapping for List: immutable `(queue, accepted_at_ms, request_id)` keys with bounded primary-key range scans, not a secondary index or a second status projection. The request-log materializer ensures each known acceptance has a mapping after summary persistence, including retries where the summary is unchanged. Unknown acceptance times are not mapped; historical backfill is separate. `QueueStore` owns internal enablement alongside pipeline coordination. `QueueSummaryStore` persists the current read model projected from immutable `QueueLogStore` occurrences. Enablement sequences, operational row versions and projection versions are distinct. These stores use primary-key operations; version arithmetic, duplicate reconciliation and projection repair belong to business logic. See [queue state](../../../doc/rfc/stovepipe/queue-state.md). + +## Repair contracts + +`FixStore` defines queue-scoped, primary-key operations on logical repairs. Each repair retains episode references, lifecycle state, an execution ID, optional metadata, timestamps, and a version. Conditional updates replace the complete snapshot; callers own version arithmetic. Nil and empty metadata maps are equivalent. + +`Queue.ActiveFixIDs` holds the queue's active repair IDs and shares `Queue.Version`. Queue writers preserve this membership when updating unrelated fields. Version checks protect membership updates; concurrent evaluations can still start overlapping repairs. + +Build analysis manages failure episodes, failure and recovery evidence, and the authoritative `FixID` assignment. Fixer updates preserve episode evidence, and build-analysis updates preserve assignments. The [required failure-episode contract](../../../doc/rfc/stovepipe/fixer.md#required-failure-episode-contract) defines the integration; its entity and persistence implementation belong to buildanalysis. + +The [dispatch sequence](../../../doc/rfc/stovepipe/fixer.md#4-trigger-the-repair) uses `Fix.State`, `Fix.ExecutionRef`, and `Fix.Version` to persist intent before submission and reject stale results. Persistence implementations follow in the storage layer. diff --git a/stovepipe/extension/storage/fix_store.go b/stovepipe/extension/storage/fix_store.go new file mode 100644 index 000000000..be25421e0 --- /dev/null +++ b/stovepipe/extension/storage/fix_store.go @@ -0,0 +1,32 @@ +// Copyright (c) 2025 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 storage + +//go:generate mockgen -source=fix_store.go -destination=mock/fix_store_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/stovepipe/entity" +) + +// FixStore persists queue-scoped versioned logical repairs by ID. +// Create/Get use ErrAlreadyExists/ErrNotFound; Update uses ErrVersionMismatch. +// Callers supply both the guard version and replacement version. +type FixStore interface { + Create(context.Context, entity.Fix) error + Get(context.Context, string) (entity.Fix, error) + Update(context.Context, entity.Fix, int32, int32) error +} diff --git a/stovepipe/extension/storage/mock/BUILD.bazel b/stovepipe/extension/storage/mock/BUILD.bazel index dab7cef72..88ddedf11 100644 --- a/stovepipe/extension/storage/mock/BUILD.bazel +++ b/stovepipe/extension/storage/mock/BUILD.bazel @@ -4,6 +4,7 @@ go_library( name = "go_default_library", srcs = [ "build_store_mock.go", + "fix_store_mock.go", "queue_log_store_mock.go", "queue_store_mock.go", "queue_summary_store_mock.go", diff --git a/stovepipe/extension/storage/mock/fix_store_mock.go b/stovepipe/extension/storage/mock/fix_store_mock.go new file mode 100644 index 000000000..6b6919c4f --- /dev/null +++ b/stovepipe/extension/storage/mock/fix_store_mock.go @@ -0,0 +1,85 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: fix_store.go +// +// Generated by this command: +// +// mockgen -source=fix_store.go -destination=mock/fix_store_mock.go -package=mock +// + +// Package mock is a generated GoMock package. +package mock + +import ( + context "context" + reflect "reflect" + + entity "github.com/uber/submitqueue/stovepipe/entity" + gomock "go.uber.org/mock/gomock" +) + +// MockFixStore is a mock of FixStore interface. +type MockFixStore struct { + ctrl *gomock.Controller + recorder *MockFixStoreMockRecorder + isgomock struct{} +} + +// MockFixStoreMockRecorder is the mock recorder for MockFixStore. +type MockFixStoreMockRecorder struct { + mock *MockFixStore +} + +// NewMockFixStore creates a new mock instance. +func NewMockFixStore(ctrl *gomock.Controller) *MockFixStore { + mock := &MockFixStore{ctrl: ctrl} + mock.recorder = &MockFixStoreMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockFixStore) EXPECT() *MockFixStoreMockRecorder { + return m.recorder +} + +// Create mocks base method. +func (m *MockFixStore) Create(arg0 context.Context, arg1 entity.Fix) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Create", arg0, arg1) + ret0, _ := ret[0].(error) + return ret0 +} + +// Create indicates an expected call of Create. +func (mr *MockFixStoreMockRecorder) Create(arg0, arg1 any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockFixStore)(nil).Create), arg0, arg1) +} + +// Get mocks base method. +func (m *MockFixStore) Get(arg0 context.Context, arg1 string) (entity.Fix, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Get", arg0, arg1) + ret0, _ := ret[0].(entity.Fix) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Get indicates an expected call of Get. +func (mr *MockFixStoreMockRecorder) Get(arg0, arg1 any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockFixStore)(nil).Get), arg0, arg1) +} + +// Update mocks base method. +func (m *MockFixStore) Update(arg0 context.Context, arg1 entity.Fix, arg2, arg3 int32) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Update", arg0, arg1, arg2, arg3) + ret0, _ := ret[0].(error) + return ret0 +} + +// Update indicates an expected call of Update. +func (mr *MockFixStoreMockRecorder) Update(arg0, arg1, arg2, arg3 any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Update", reflect.TypeOf((*MockFixStore)(nil).Update), arg0, arg1, arg2, arg3) +} diff --git a/tool/proto/BUILD.bazel b/tool/proto/BUILD.bazel index 8c6037e11..51aa992aa 100644 --- a/tool/proto/BUILD.bazel +++ b/tool/proto/BUILD.bazel @@ -104,6 +104,8 @@ go_proto_generated_files( srcs = [ "//stovepipe/core/messagequeue/proto:build.proto", "//stovepipe/core/messagequeue/proto:buildsignal.proto", + "//stovepipe/core/messagequeue/proto:fix.proto", + "//stovepipe/core/messagequeue/proto:fixsignal.proto", "//stovepipe/core/messagequeue/proto:process.proto", "//stovepipe/core/messagequeue/proto:record.proto", ],