feat(stovepipe): add request log materializer (#657)

## Summary
Intent:
- Provide one direct, idempotent materialization path for durable
request state logs.
- Keep the initial rollout limited to state entries while preserving a
boundary for future request summaries and other projections.

Changes:
- Add `Materializer.PersistLog` over the queue-scoped storage aggregate.
- Compose readable stable occurrence IDs with `publish.IntentID`.
- Deduplicate retained occurrences because Stovepipe writes directly;
SubmitQueue instead deduplicates the message carrying a log to its
materializer.
- Preserve caller-supplied timestamps, assign missing timestamps, and
reconcile duplicate writes by retained semantic content.
- Emit bounded metrics with context-derived tags.
- Document how future projections can be added within the materializer
without changing callers.

## Test Plan
- `./tool/bazel test //stovepipe/core/requestlog:go_default_test
--cache_test_results=no`
- `make check-gazelle`
- `make lint`

## Revert Plan
- Revert this PR. No production call site depends on the materializer
until the next PR in the stack lands.

## Issues


## Stack
1. @ #657
1. #658

Author: mnoah1

Makefile

@@ -579,7 +579,7 @@ local-stovepipe-stop: ## Stop the Stovepipe service
 
 mocks: ## Generate mock files using mockgen
 	@echo "Generating mocks..."
-	@$(BAZEL) run @rules_go//go -- generate ./submitqueue/extension/storage/... ./submitqueue/extension/buildrunner/... ./submitqueue/extension/changeprovider/... ./platform/extension/counter/... ./platform/extension/consumergate/... ./platform/extension/hook/... ./platform/extension/messagequeue/... ./submitqueue/extension/queueconfig/... ./submitqueue/extension/mergechecker/... ./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/conflict/... ./submitqueue/extension/speculation/... ./submitqueue/extension/validator/... ./platform/consumer/... ./stovepipe/core/requestlog/... ./stovepipe/extension/storage/... ./stovepipe/extension/sourcecontrol/...
 	@echo "Mocks generated successfully!"
 
 proto: ## Generate protobuf files from .proto definitions

doc/rfc/stovepipe/request-history-api.md

@@ -107,7 +107,9 @@ Event rows remain in SubmitQueue history but never participate in current-status
 
 Stovepipe has no equivalent ownership gap. The same service owns the queue-scoped `Request`, `ValidationFact`, request-URI mapping, and request-log store. Operational reads use their owning entities, while request history reads retained log records directly.
 
-Stovepipe therefore does not add `RequestSummary`, replay history to determine current state, or materialize another history table. The controller performs only an in-memory wire projection from stored log records to protobuf messages. This avoids a second winner-selection algorithm competing with Request CAS state.
+Stovepipe calls `requestlog.Materializer.PersistLog` directly, without an intermediate topic. Its initial materializer appends only the request log: it does not add `RequestSummary`, replay history to determine current state, or materialize another history table. The controller performs only an in-memory wire projection from stored log records to protobuf messages. This avoids a second winner-selection algorithm competing with Request CAS state.
+
+`PersistLog` receives the whole queue-scoped storage aggregate so a future current-status or queue-list API can add SubmitQueue-style summary and index projections behind the same write boundary without changing request-log producers. Such projections remain deferred until a concrete read path needs them; `Request` remains authoritative in the initial implementation.
 
 ## Ordering and Consistency
 

doc/rfc/stovepipe/request-log.md

@@ -2,7 +2,7 @@
 
 ## Summary
 
-Stovepipe retains an append-only request log for each validation request. Its internal `RequestLog` is the counterpart of SubmitQueue's `RequestLog`: both retain request status changes and explanatory lifecycle events, while Stovepipe persists records directly instead of sending them through a cross-service log topic and materializer. The public API presents these records as request history. The log records every durable `Request.State` transition plus three asynchronous milestones needed to explain those transitions and the public verdict:
+Stovepipe retains an append-only request log for each validation request. Its internal `RequestLog` is the counterpart of SubmitQueue's `RequestLog`: both retain request status changes and explanatory lifecycle events, while Stovepipe calls its materializer directly instead of sending records through a cross-service log topic. The public API presents these records as request history. The log records every durable `Request.State` transition plus three asynchronous milestones needed to explain those transitions and the public verdict:
 
 - `build_triggered`;
 - `build_finished`;
@@ -73,13 +73,13 @@ The retained unit is `entity.RequestLog`:
 
 ```go
 type RequestLog struct {
-    // ID is the stable identity of one logical occurrence within the request. It is opaque, stable across redelivery, and never derived from time or randomness.
+    // ID is the stable identity of one logical occurrence within the request. It is stable across redelivery and never derived from time or randomness.
     ID string
     // Queue is the logical queue containing the request and scopes RequestID.
     Queue string
     // RequestID identifies the request whose log contains this record.
     RequestID string
-    // TimestampMs is the durable occurrence time in Unix milliseconds.
+    // TimestampMs is when the occurrence was first retained, in Unix milliseconds.
     TimestampMs int64
     // State is the durable request state recorded by a state entry. It is unset on an event entry.
     State RequestState
@@ -145,7 +145,9 @@ Terminal entries retain domain reasons rather than transport mechanisms. Initial
 
 ## Stable IDs and Idempotency
 
-`stovepipe/core/requestlog.Recorder` constructs opaque IDs from durable identities:
+`stovepipe/core/requestlog` composes readable IDs from durable identities with the same `publish.IntentID` convention used by SubmitQueue before passing each record to `Materializer.PersistLog`. The surrounding `(queue, request_id)` storage key scopes the ID to one request, so a request state entry uses `state/<request-version>` without repeating the request identity:
+
+SubmitQueue applies that identity to the message carrying a log to its materializer, while its log store remains append-only and may retain a duplicate after a later materialization step fails. Stovepipe has no intermediate log topic, so it applies the identity to the retained row itself: retrying the direct call reloads the existing occurrence and succeeds only when its semantic content is compatible.
 
 | Entry | Stable identity inputs |
 |---|---|
@@ -154,7 +156,7 @@ Terminal entries retain domain reasons rather than transport mechanisms. Initial
 | Build finished | Request ID, event kind, and build ID |
 | Validation fact recorded | Request ID, event kind, and whole-repository fact identity |
 
-The recorder calls `RequestLogStore.Create`. If the ID already exists, it loads the stored record and compares every semantic field. Identical content is idempotent success; conflicting content is an internal consistency error, and the stored record is never overwritten.
+The controller passes the materializer the same queue-scoped storage aggregate used for the source write. The materializer preserves a supplied occurrence time or assigns the current time immediately before the first insertion attempt, then calls `RequestLogStore.Create`. If the ID already exists, it loads the stored record and compares the explicitly designated stable semantic fields. The first successfully retained timestamp is authoritative and is not compared with a later retry's newly sampled time. Metadata keys emitted by both records must agree, while a key present on only one record remains compatible so an additive metadata rollout does not turn retries of older occurrences into conflicts. Compatible content is idempotent success; conflicting content is an internal consistency error, and the stored record is never overwritten or enriched.
 
 ## Storage Contract
 
@@ -188,11 +190,11 @@ Request-log durability is part of completing a pipeline transition. The source w
 
 For a Request transition, the controller:
 
-1. builds an immutable updated copy with transition context and `StateChangedAtMs`;
+1. builds an immutable updated copy for the state transition;
 2. computes `newVersion = oldVersion + 1`;
 3. calls `RequestStore.Update(updated, oldVersion, newVersion)`;
 4. assigns the in-memory version only after the store succeeds;
-5. asks the recorder to create the log record from durable Request data;
+5. constructs the log record from the durable Request and bounded context still owned by that stage, then calls `Materializer.PersistLog`;
 6. publishes the downstream handoff.
 
 Request creation, Build changes, and fact creation use the same source-write, log-write, dependent-publish ordering. A request-log outage can leave a source update visible, but it cannot allow dependent processing to move past an unrecorded transition.
@@ -224,7 +226,7 @@ This is rollout work, not deferred cleanup: a mandatory request log without a du
 
 Rollout therefore:
 
-1. deploys source timestamp and provenance fields, request-log storage, recorder, and readers;
+1. deploys request-log storage, the materializer, and readers;
 2. enables writers and verifies every repair path stage by stage;
 3. enables the public API after every writer and repair path is active.
 
@@ -246,11 +248,11 @@ Identifiers, outcome reasons, and reasonable per-request build counts have expli
 
 Contract tests cover required-field validation, stable IDs, idempotent create/reload, conflict detection, queue binding, deterministic ordering, equal-timestamp tie breaking, and empty histories.
 
-Writer tests cover source success followed by log failure, redelivery with the log record absent or present, CAS loss, conflicting terminal writers, downstream publish failure, stable timestamps, and controller-owned version arithmetic. End-to-end tests cover successful, failed, cancelled, superseded, and fail-closed paths plus idempotent redelivery.
+Writer tests cover source success followed by log failure, redelivery with the log record absent or present, CAS loss, conflicting terminal writers, downstream publish failure, first-insert timestamp reuse, and controller-owned version arithmetic. End-to-end tests cover successful, failed, cancelled, superseded, and fail-closed paths plus idempotent redelivery.
 
 Tests reconstruct the latest Request state from state entries by request version and compare it with `RequestStore.Get`. They separately verify that only a durable fact produces green or broken.
 
-The recorder reports create, identical-existing, conflict, validation failure, and storage failure counters tagged only by bounded state or event. IDs, URIs, and errors remain structured log fields rather than metric tags. Alerts cover sustained repair gaps and content conflicts.
+The materializer reports create, identical-existing, conflict, validation failure, and storage failure counters tagged only by bounded state or event plus tags carried by the context. IDs, URIs, and errors remain structured log fields rather than metric tags. Alerts cover sustained repair gaps and content conflicts.
 
 ## Alternatives Considered
 

stovepipe/core/requestlog/BUILD.bazel

@@ -0,0 +1,31 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["materializer.go"],
+    importpath = "github.com/uber/submitqueue/stovepipe/core/requestlog",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//platform/metrics:go_default_library",
+        "//platform/publish:go_default_library",
+        "//stovepipe/entity:go_default_library",
+        "//stovepipe/extension/storage:go_default_library",
+        "@com_github_uber_go_tally//:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["materializer_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//platform/metrics:go_default_library",
+        "//stovepipe/entity:go_default_library",
+        "//stovepipe/extension/storage:go_default_library",
+        "//stovepipe/extension/storage/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_uber_go_mock//gomock:go_default_library",
+    ],
+)

stovepipe/core/requestlog/materializer.go

@@ -0,0 +1,133 @@
+// 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 requestlog retains the request occurrences exposed by Stovepipe's history API.
+package requestlog
+
+//go:generate mockgen -source=materializer.go -destination=mock/materializer_mock.go -package=mock
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/uber-go/tally"
+
+	"github.com/uber/submitqueue/platform/metrics"
+	"github.com/uber/submitqueue/platform/publish"
+	"github.com/uber/submitqueue/stovepipe/entity"
+	"github.com/uber/submitqueue/stovepipe/extension/storage"
+)
+
+const (
+	_occurrenceKindState = "state"
+)
+
+// Materializer persists request-log occurrences into their queue-scoped read model.
+type Materializer interface {
+	// PersistLog retains one request-log occurrence idempotently.
+	PersistLog(context.Context, storage.Storage, entity.RequestLog) error
+}
+
+type materializer struct {
+	scope tally.Scope
+	now   func() time.Time
+}
+
+var _ Materializer = (*materializer)(nil)
+
+// NewMaterializer creates a request-log materializer.
+func NewMaterializer(scope tally.Scope) Materializer {
+	return &materializer{
+		scope: scope.SubScope("request_log_materializer"),
+		now:   time.Now,
+	}
+}
+
+// NewRequestStateLog constructs the occurrence representing the request's current durable state.
+func NewRequestStateLog(request entity.Request, outcomeReason entity.RequestOutcomeReason) entity.RequestLog {
+	return entity.RequestLog{
+		ID:             publish.IntentID(_occurrenceKindState, strconv.FormatInt(int64(request.Version), 10)),
+		Queue:          request.Queue,
+		RequestID:      request.ID,
+		State:          request.State,
+		RequestVersion: request.Version,
+		OutcomeReason:  outcomeReason,
+	}
+}
+
+func (m *materializer) PersistLog(ctx context.Context, stores storage.Storage, log entity.RequestLog) error {
+	if log.TimestampMs == 0 {
+		log.TimestampMs = m.now().UnixMilli()
+	}
+
+	if err := log.Validate(); err != nil {
+		m.count(ctx, "validation_failure")
+		return fmt.Errorf("invalid request log occurrence: %w", err)
+	}
+
+	store := stores.GetRequestLogStore()
+	// SubmitQueue deduplicates the message that hands a log to its materializer. Stovepipe has no
+	// log topic, so the retained occurrence ID is the retry boundary instead.
+	if err := store.Create(ctx, log); err == nil {
+		m.count(ctx, "created")
+		return nil
+	} else if !errors.Is(err, storage.ErrAlreadyExists) {
+		m.count(ctx, "storage_failure")
+		return fmt.Errorf("failed to create request log request_id=%q log_id=%q: %w", log.RequestID, log.ID, err)
+	}
+
+	stored, err := store.Get(ctx, log.RequestID, log.ID)
+	if err != nil {
+		m.count(ctx, "storage_failure")
+		return fmt.Errorf("failed to reconcile request log request_id=%q log_id=%q: %w", log.RequestID, log.ID, err)
+	}
+	if !sameSemanticOccurrence(stored, log) {
+		m.count(ctx, "conflict")
+		return fmt.Errorf("request log conflicts with retained occurrence request_id=%q log_id=%q", log.RequestID, log.ID)
+	}
+
+	m.count(ctx, "identical_existing")
+	return nil
+}
+
+func (m *materializer) count(ctx context.Context, counter string) {
+	metrics.NamedCounter(m.scope, "persist", counter, 1, metrics.TagsFromContext(ctx)...)
+}
+
+func sameSemanticOccurrence(stored, candidate entity.RequestLog) bool {
+	// This list is the compatibility boundary for duplicate reconciliation. New entity fields do not
+	// become conflict-sensitive until they are deliberately added here.
+	return stored.ID == candidate.ID &&
+		stored.Queue == candidate.Queue &&
+		stored.RequestID == candidate.RequestID &&
+		stored.State == candidate.State &&
+		stored.Event == candidate.Event &&
+		stored.RequestVersion == candidate.RequestVersion &&
+		stored.OutcomeReason == candidate.OutcomeReason &&
+		metadataCompatible(stored.Metadata, candidate.Metadata)
+}
+
+func metadataCompatible(stored, candidate map[string]string) bool {
+	// One-sided keys permit additive metadata rollout without making a retry conflict with an older
+	// immutable row. Values emitted by both versions must still agree.
+	for key, storedValue := range stored {
+		if candidateValue, ok := candidate[key]; ok && candidateValue != storedValue {
+			return false
+		}
+	}
+	return true
+}