feat(stovepipe): offer newly opened issues to every registered fixer

doc/rfc/stovepipe/request-log.md

@@ -143,7 +143,7 @@ State entries retain the selected baseline URI when known, including an empty UR
 
 Build running and unchanged polls are not retained. Trigger and terminal result explain the request outcome without turning polling into an unbounded log. Individual project facts remain outside the vocabulary; one batch-completion event summarizes their durable recording. Its `project_fact_count` and `broken_project_fact_count` metadata contain the total and the number with degree above zero, counted from persisted facts after their indexes are durable. Project completion projection requires both counts and assigns them together in `RequestSummary` from the retained event.
 
-`build_analysis_completed`, `issues_reconciled`, `fix_submission_skipped`, and `fix_submission_abandoned` use the build ID as their occurrence key. `fix_submitted` uses a digest of the executor fix ID to fit the storage key limit, retaining the verbatim fix ID in metadata. Existing issue-tracking abandonment uses a digest of the failed notification ID to fit the storage key limit. The materializer prefixes each key with its event kind, and request identity scopes all entries. Replays retain the first stored timestamp.
+`build_analysis_completed` and `issues_reconciled` use the build ID as their occurrence key. `fix_submission_abandoned` uses the build ID. Per-item `fix_submitted` and `fix_submission_skipped` events use a digest of opening build, task, unit, and executor identity, retaining the verbatim fix ID or skip reason and executor ID in metadata. Existing issue-tracking abandonment uses a digest of the failed notification ID to fit the storage key limit. The materializer prefixes each key with its event kind, and request identity scopes all entries. Replays retain the first stored timestamp.
 
 Analysis and repair events do not advance Request state or alter validation facts. Analysis completion is retained only when the Request records the analyzed terminal build as its winning outcome, immediately before the final issue-tracking handoff. Its issue count and timestamp are an immutable snapshot of the first retained completion, including zero for an empty result. Retries still persist detected issues and publish the handoff, but reuse that snapshot if later detector results differ; count discrepancies are logged rather than treated as history conflicts. Other semantic conflicts remain errors. `issues_reconciled` records closure mode `reconciled`, `suppressed_cancelled`, or `suppressed_task_wide_issue`; suppression does not mean reconciliation failed. Skipped repair uses reason `no_eligible_issue`; abandoned repair uses `acceptance_unknown`. Raw dependency errors remain in operational logs.
 
@@ -238,7 +238,7 @@ The `project_facts_recorded` and `record_abandoned` events project completion of
 | Build analysis DLQ | Retain `build_analysis_abandoned` without rerunning detection or changing validation state. | One stable event per build deduplicates recovery deliveries; only state reads and history retention retry. |
 | Issue tracking DLQ | Protect the current cursor from delayed writes, retain its issue history and `issue_tracking_abandoned`, then mark the pass abandoned. | A stable hash of the notification ID deduplicates the event. Recovery retains accepted issue changes and never processes the remaining range. |
 | Issue tracking | Finalize the durable checkpoint, then retain `issues_reconciled`. | Reconstruct history from the finalized checkpoint before it can be replaced by another pass. |
-| Fix | Associate the returned fix ID, then retain `fix_submitted`; retain `fix_submission_skipped` when no issue is eligible. | Ensure submitted history from any existing association before skipping executor work. |
+| Fix | Retain each executor disposition, then retain its submitted or skipped event. | Restore every durable disposition event before skipping that item for that executor. |
 | Fix DLQ | Repair submitted or skipped history from current durable state, or retain `fix_submission_abandoned` when acceptance is unknown. | Only read sources and retain history; never call the executor or change issue state. |
 | Reconciler | CAS an unrecoverable non-terminal Request to failed, then retain failed. | An existing terminal Request is repaired from its persisted outcome without relabeling it. |
 

service/stovepipe/README.md

@@ -26,7 +26,7 @@ Stovepipe therefore needs two MySQL databases: a **storage** database (the `queu
 
 - **`inMemoryCounter`** — a process-local `counter.Counter` for sequence numbers; not durable. A real deployment uses a persistent implementation (e.g. `platform/extension/counter/mysql`).
 - **`fakeSourceControlFactory`** — seeds each queue with a deterministic single-commit history so ingest resolves a stable head URI (and re-ingesting the same queue exercises the dedup path). A real deployment supplies a VCS-backed `sourcecontrol.Factory`, which is also where a queue's promotion ref is resolved. The fake has no ref to move, so a promotion locally shows up only in the record consumer's logs.
-- **`noopRepairExecutorFactory`** — accepts repairs without launching work and returns `noop/<buildID>`. The fix controller stores that ID, consuming the build's one-repair allowance. Replacing this factory does not retry builds that already have a no-op association. The default no-op issue detector reports no failures, so the reference stack does not normally reach repair submission.
+- **`noopRepairExecutorFactory`** — accepts repairs without launching work and returns `noop/<buildID>`. The fix controller stores per-item submission decisions under registration ID `noop`. A different registration ID has independent eligibility. The default no-op issue detector reports no failures, so the reference stack does not normally reach repair submission.
 
 ## Layout
 

service/stovepipe/server/main.go

@@ -97,8 +97,8 @@ func (fakeBuildRunnerFactory) For(cfg buildrunner.Config) (buildrunner.BuildRunn
 
 type noopRepairExecutorFactory struct{}
 
-func (noopRepairExecutorFactory) For(fixer.Config) (fixer.RepairExecutor, error) {
-	return fixernoop.New(), nil
+func (noopRepairExecutorFactory) For(fixer.Config) ([]fixer.Registration, error) {
+	return []fixer.Registration{{ID: "noop", Executor: fixernoop.New()}}, nil
 }
 
 // hookResolver sends every event to the no-op hook. Which hooks an event goes to is host

service/stovepipe/server/main_test.go

@@ -136,7 +136,7 @@ func TestFixRegistration(t *testing.T) {
 	assert.Contains(t, topicKeys(deadLetter), dlq.TopicKey(stovepipemq.TopicKeyFix))
 }
 
-func TestRegisteredFixControllerRecordsNoopRepairOncePerBuild(t *testing.T) {
+func TestRegisteredFixControllerRetainsPerExecutorDisposition(t *testing.T) {
 	const queueName = "validation/main"
 	const buildID = "build/1"
 	ctrl := gomock.NewController(t)
@@ -181,14 +181,23 @@ func TestRegisteredFixControllerRecordsNoopRepairOncePerBuild(t *testing.T) {
 		TaskID: observation.TaskID, UnitID: observation.UnitID, Status: entity.IssueStatusOpen,
 		FirstSeenBuildID: buildID, LastProcessedBuildID: buildID, LastChangedBuildID: buildID,
 		ChangedAtMs: 1000, Version: 1,
-	}, nil)
-	observations.EXPECT().Update(gomock.Any(), buildID, gomock.Any(), int32(1), int32(2)).DoAndReturn(
-		func(_ context.Context, _ string, updated entity.BuildIssue, _, newVersion int32) error {
-			require.Equal(t, "noop/"+buildID, updated.FixID)
-			observation = updated
-			observation.Version = newVersion
-			return nil
-		})
+	}, nil).AnyTimes()
+	repairs := storagemock.NewMockIssueRepairStore(ctrl)
+	store.EXPECT().GetIssueRepairStore().Return(repairs).AnyTimes()
+	var repair entity.IssueRepair
+	repairs.EXPECT().Get(gomock.Any(), gomock.Any(), "noop").DoAndReturn(func(context.Context, entity.OpenIssue, string) (entity.IssueRepair, error) {
+		if repair.Version == 0 {
+			return entity.IssueRepair{}, storage.ErrNotFound
+		}
+		return repair, nil
+	}).Times(3)
+	repairs.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, value entity.IssueRepair) error { repair = value; return nil })
+	repairs.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).DoAndReturn(func(_ context.Context, value entity.IssueRepair, _, newVersion int32) error {
+		require.Empty(t, repair.FixID)
+		repair = value
+		repair.Version = newVersion
+		return nil
+	})
 
 	registry, err := newTopicRegistry(nil, "subscriber")
 	require.NoError(t, err)

stovepipe/controller/fix/BUILD.bazel

@@ -7,6 +7,7 @@ go_library(
         "dlq.go",
         "fix.go",
         "history.go",
+        "submission.go",
     ],
     importpath = "github.com/uber/submitqueue/stovepipe/controller/fix",
     visibility = ["//visibility:public"],
@@ -39,6 +40,7 @@ go_test(
         "//platform/consumer:go_default_library",
         "//platform/consumer/mock:go_default_library",
         "//platform/errs:go_default_library",
+        "//platform/errs/generic:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",
         "//stovepipe/core/requestlog:go_default_library",
         "//stovepipe/entity:go_default_library",
@@ -51,6 +53,5 @@ go_test(
         "@com_github_uber_go_tally//:go_default_library",
         "@org_uber_go_mock//gomock:go_default_library",
         "@org_uber_go_zap//:go_default_library",
-        "@org_uber_go_zap//zaptest/observer:go_default_library",
     ],
 )

stovepipe/controller/fix/README.md

@@ -1,13 +1,17 @@
 # Fix submission
 
-The controller consumes a `FixRequest` containing a queue name and build ID. It pages through a build's durable issue records and skips submission when any record has a fix ID, including recovered issues. Otherwise, it selects one currently open issue whose first-seen build matches this build and calls `RepairExecutor.Submit`.
+The controller consumes a build-scoped `FixRequest`. It reads every page of the build's observations, selects still-open issues whose opening build matches, and offers each registered executor all of its outstanding items in one `Submit` call. Different executors may choose different groupings. Their stable, queue-local registration IDs are part of the durable identity; an empty registry disables submission.
 
-A returned fix ID is associated with the issue through `BuildIssueStore.Update`, using the version read during selection. A concurrent association is preserved. A submission error without an ID leaves the issue eligible for normal queue retries; an ID returned alongside an error is retained.
+The executor returns per-item `Submission` values once the backend accepts work or the executor deliberately skips it. `FixID` identifies the backend request or run; it does not indicate repair completion or success. Multiple items and successive calls may share a fix ID. Each returned disposition is retained independently in `issue_repair`, keyed by queue, opening build, task, unit, and executor. Accepted submissions are retained even alongside an error. Omitted items remain pending; omitting items without an error violates the contract. Invalid, foreign, or duplicate submissions fail processing. One executor's error does not suppress another executor's submission.
 
-Issue tracking calls [`fixrequest.Publish`](../../core/fixrequest/publish.go) after completing a nonempty opening pass and persisting its issue state and history. Publications share a build partition, while the message identity includes the pass notification ID so a later pass can trigger another evaluation. A pass with no eligible issue completes without submitting a repair; a recorded fix ID prevents later passes from submitting another repair for the build.
+Before calling an executor, the controller ensures a pending `issue_repair` row exists for every offered opening and executor. Empty `FixID` and `SkipReason` mean pending; a row's existence does not establish backend acceptance. Returned submissions update those fields after acceptance or a deliberate skip. Existing pending rows are offered again, while submitted and skipped rows restore history without resubmission.
 
-The one-per-build limit is controller policy. Submission precedes persistence, so overlapping deliveries, crashes, and failed writes can still produce duplicate executor calls. Issue recovery remains separate from repair association.
+`IssueRepair.Metadata` is an executor-owned `map[string]string` stored in the JSON `metadata` column. Executors read and update it through injected storage; it is absent from `Submission`. String values may contain serialized JSON, and nil and empty maps are equivalent. Storage uses versioned replacement snapshots: both metadata writers and the controller must preserve other fields and reload before reapplying an edit after `ErrVersionMismatch`. The controller reloads the latest row before recording a receipt, preserving metadata written during submission. Recorded submission outcomes are never replaced by conflicting receipts.
 
-A durable association retains `fix_submitted` in the owning Request's history, with build, task, unit, and fix IDs. A digest of the fix ID identifies the occurrence and fits the history key limit; metadata retains the verbatim fix ID. An existing association repairs missing history before skipping submission, including when the issue has since recovered. Without an eligible issue, the controller retains `fix_submission_skipped` once per build with reason `no_eligible_issue`. These entries do not change Request outcome or establish issue recovery.
+There is no controller-wide repair cap or batch identity. Executor calls can overlap; pending rows are not exclusive dispatch claims. Implementations must deduplicate by issue opening even after a lost response or a failed local write. Build-level limits belong to host adapters. The legacy `build_issue.fix_id` field is preserved but does not gate this controller.
+
+Each durable disposition restores its request-history event before being skipped on replay, including after issue closure. Events retain build, task, unit, executor, and accepted fix ID or skip reason. A digest of the opening and executor identifies the occurrence. The legacy request-issue scalar fix field exposes the bytewise least accepted fix ID; complete associations remain in history and `issue_repair`. Repair submission does not establish issue recovery.
+
+Issue tracking publishes another nudge after each nonempty opening pass, so late issues are offered independently of earlier submissions. Queue redelivery rederives pending work and may offer a smaller list. Reopening in another build creates a new identity. A fix ID must reflect actual backend acceptance for every item carrying it; returning an old ID alone does not deliver new work.
 
 The DLQ controller reads the same durable decision without calling the executor. It repairs submitted or skipped history when possible; otherwise, an eligible issue without a fix ID retains `fix_submission_abandoned` once per build with reason `acceptance_unknown`. An executor error without an ID can follow acceptance, so this event does not claim rejection. Source and history errors retry. Unusable payloads, unresolvable queues, and missing or mismatched sources are logged and acknowledged. The reference host registers recovery with unlimited, always-retryable DLQ processing to preserve history repair through a storage outage.

stovepipe/controller/fix/delivery_test.go

@@ -24,6 +24,7 @@ import (
 	consumermock "github.com/uber/submitqueue/platform/consumer/mock"
 	"github.com/uber/submitqueue/platform/errs"
 	mq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
+	"github.com/uber/submitqueue/stovepipe/entity"
 	"go.uber.org/mock/gomock"
 )
 
@@ -46,8 +47,7 @@ func TestProcessSubmitsBuildIssue(t *testing.T) {
 	controller, m := newTestController(t)
 	issue := testIssue()
 	m.expectIssue(issue)
-	m.executor.EXPECT().Submit(gomock.Any(), issue).Return("attempt", nil)
-	m.observations.EXPECT().Update(gomock.Any(), testBuildID, issueWithFixID(issue, "attempt"), int32(0), int32(1)).Return(nil)
+	m.executor.EXPECT().Submit(gomock.Any(), []entity.OpenIssue{issue}).Return(accepted([]entity.OpenIssue{issue}, "attempt"), nil)
 
 	require.NoError(t, controller.Process(context.Background(), m.delivery(repairMessage(t))))
 }