Skip to content

Commit bb096db

Browse files
mnoah1mnoah1
committed
feat(stovepipe): add request state log recorder
Summary: Intent: - Provide one idempotent path for retaining durable request state transitions. - Keep the initial rollout limited to state entries while lifecycle events and metadata conventions remain deferred. Changes: - Derive stable occurrence identities from the queue, request, and request version. - Treat identical duplicate writes as success and surface conflicting retained content. - Report bounded recorder metrics and cover validation, retries, and conflicts. --- <sub>Generated by the 🪄 [pr-create](https://sg.uberinternal.com/code.uber.internal/uber-code/devexp-agent-marketplace/-/blob/claude-code/plugins/dev/uber-dev/skills/pr-create/SKILL.md) skill in devexp-agent-marketplace</sub>
1 parent e17ca6c commit bb096db

8 files changed

Lines changed: 437 additions & 8 deletions

File tree

‎Makefile‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -579,7 +579,7 @@ local-stovepipe-stop: ## Stop the Stovepipe service
579579

580580
mocks: ## Generate mock files using mockgen
581581
@echo "Generating mocks..."
582-
@$(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/...
582+
@$(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/...
583583
@echo "Mocks generated successfully!"
584584

585585
proto: ## Generate protobuf files from .proto definitions

‎doc/rfc/stovepipe/request-log.md‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -79,7 +79,7 @@ type RequestLog struct {
7979
Queue string
8080
// RequestID identifies the request whose log contains this record.
8181
RequestID string
82-
// TimestampMs is the durable occurrence time in Unix milliseconds.
82+
// TimestampMs is when the occurrence was first retained, in Unix milliseconds.
8383
TimestampMs int64
8484
// State is the durable request state recorded by a state entry. It is unset on an event entry.
8585
State RequestState
@@ -154,7 +154,7 @@ Terminal entries retain domain reasons rather than transport mechanisms. Initial
154154
| Build finished | Request ID, event kind, and build ID |
155155
| Validation fact recorded | Request ID, event kind, and whole-repository fact identity |
156156

157-
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.
157+
The controller passes the recorder the `RequestLogStore` from the same queue-scoped storage aggregate used for the source write. The recorder assigns the current time immediately before the first insertion attempt and calls `Create`. If the ID already exists, it loads the stored record and compares every domain field. The first successfully retained timestamp is authoritative and is not compared with a later retry's newly sampled time. Identical domain content is idempotent success; conflicting content is an internal consistency error, and the stored record is never overwritten.
158158

159159
## Storage Contract
160160

@@ -188,11 +188,11 @@ Request-log durability is part of completing a pipeline transition. The source w
188188

189189
For a Request transition, the controller:
190190

191-
1. builds an immutable updated copy with transition context and `StateChangedAtMs`;
191+
1. builds an immutable updated copy for the state transition;
192192
2. computes `newVersion = oldVersion + 1`;
193193
3. calls `RequestStore.Update(updated, oldVersion, newVersion)`;
194194
4. assigns the in-memory version only after the store succeeds;
195-
5. asks the recorder to create the log record from durable Request data;
195+
5. asks the recorder to create the log record from the durable Request and the bounded context still owned by that stage;
196196
6. publishes the downstream handoff.
197197

198198
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 +224,7 @@ This is rollout work, not deferred cleanup: a mandatory request log without a du
224224

225225
Rollout therefore:
226226

227-
1. deploys source timestamp and provenance fields, request-log storage, recorder, and readers;
227+
1. deploys request-log storage, the recorder, and readers;
228228
2. enables writers and verifies every repair path stage by stage;
229229
3. enables the public API after every writer and repair path is active.
230230

@@ -246,7 +246,7 @@ Identifiers, outcome reasons, and reasonable per-request build counts have expli
246246

247247
Contract tests cover required-field validation, stable IDs, idempotent create/reload, conflict detection, queue binding, deterministic ordering, equal-timestamp tie breaking, and empty histories.
248248

249-
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.
249+
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.
250250

251251
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.
252252

‎stovepipe/core/requestlog/BUILD.bazel‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
load("@rules_go//go:def.bzl", "go_library", "go_test")
2+
3+
go_library(
4+
name = "go_default_library",
5+
srcs = ["recorder.go"],
6+
importpath = "github.com/uber/submitqueue/stovepipe/core/requestlog",
7+
visibility = ["//visibility:public"],
8+
deps = [
9+
"//platform/metrics:go_default_library",
10+
"//stovepipe/entity:go_default_library",
11+
"//stovepipe/extension/storage:go_default_library",
12+
"@com_github_uber_go_tally//:go_default_library",
13+
],
14+
)
15+
16+
go_test(
17+
name = "go_default_test",
18+
srcs = ["recorder_test.go"],
19+
embed = [":go_default_library"],
20+
deps = [
21+
"//stovepipe/entity:go_default_library",
22+
"//stovepipe/extension/storage:go_default_library",
23+
"//stovepipe/extension/storage/mock:go_default_library",
24+
"@com_github_stretchr_testify//assert:go_default_library",
25+
"@com_github_stretchr_testify//require:go_default_library",
26+
"@com_github_uber_go_tally//:go_default_library",
27+
"@org_uber_go_mock//gomock:go_default_library",
28+
],
29+
)

‎stovepipe/core/requestlog/mock/BUILD.bazel‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,13 @@
1+
load("@rules_go//go:def.bzl", "go_library")
2+
3+
go_library(
4+
name = "go_default_library",
5+
srcs = ["recorder_mock.go"],
6+
importpath = "github.com/uber/submitqueue/stovepipe/core/requestlog/mock",
7+
visibility = ["//visibility:public"],
8+
deps = [
9+
"//stovepipe/entity:go_default_library",
10+
"//stovepipe/extension/storage:go_default_library",
11+
"@org_uber_go_mock//gomock:go_default_library",
12+
],
13+
)

‎stovepipe/core/requestlog/mock/recorder_mock.go‎

Lines changed: 57 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎stovepipe/core/requestlog/recorder.go‎

Lines changed: 140 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,140 @@
1+
// Copyright (c) 2025 Uber Technologies, Inc.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
// Package requestlog retains the request occurrences exposed by Stovepipe's history API.
16+
package requestlog
17+
18+
//go:generate mockgen -source=recorder.go -destination=mock/recorder_mock.go -package=mock
19+
20+
import (
21+
"context"
22+
"crypto/sha256"
23+
"encoding/binary"
24+
"encoding/hex"
25+
"errors"
26+
"fmt"
27+
"maps"
28+
"reflect"
29+
"strconv"
30+
"time"
31+
32+
"github.com/uber-go/tally"
33+
34+
"github.com/uber/submitqueue/platform/metrics"
35+
"github.com/uber/submitqueue/stovepipe/entity"
36+
"github.com/uber/submitqueue/stovepipe/extension/storage"
37+
)
38+
39+
const (
40+
_occurrenceKindState = "state"
41+
)
42+
43+
// Recorder retains idempotent request-state occurrences.
44+
type Recorder interface {
45+
// RecordRequestState retains the request's current durable state and version.
46+
RecordRequestState(context.Context, storage.RequestLogStore, entity.Request, entity.RequestOutcomeReason) error
47+
}
48+
49+
type recorder struct {
50+
scope tally.Scope
51+
now func() time.Time
52+
}
53+
54+
// NewRecorder creates a request-log recorder.
55+
func NewRecorder(scope tally.Scope) Recorder {
56+
return &recorder{
57+
scope: scope.SubScope("request_log_recorder"),
58+
now: time.Now,
59+
}
60+
}
61+
62+
func (r *recorder) RecordRequestState(ctx context.Context, store storage.RequestLogStore, request entity.Request, outcomeReason entity.RequestOutcomeReason) error {
63+
log := entity.RequestLog{
64+
ID: occurrenceID(request.Queue, request.ID, _occurrenceKindState, strconv.FormatInt(int64(request.Version), 10)),
65+
Queue: request.Queue,
66+
RequestID: request.ID,
67+
State: request.State,
68+
RequestVersion: request.Version,
69+
OutcomeReason: outcomeReason,
70+
}
71+
return r.record(ctx, store, log)
72+
}
73+
74+
func (r *recorder) record(ctx context.Context, store storage.RequestLogStore, log entity.RequestLog) error {
75+
log.TimestampMs = r.now().UnixMilli()
76+
tag := occurrenceTag(log)
77+
78+
if err := log.Validate(); err != nil {
79+
metrics.NamedCounter(r.scope, "record", "validation_failure", 1, tag)
80+
return fmt.Errorf("invalid request log occurrence: %w", err)
81+
}
82+
83+
if err := store.Create(ctx, log); err == nil {
84+
metrics.NamedCounter(r.scope, "record", "created", 1, tag)
85+
return nil
86+
} else if !errors.Is(err, storage.ErrAlreadyExists) {
87+
metrics.NamedCounter(r.scope, "record", "storage_failure", 1, tag)
88+
return fmt.Errorf("failed to create request log request_id=%q log_id=%q: %w", log.RequestID, log.ID, err)
89+
}
90+
91+
stored, err := store.Get(ctx, log.RequestID, log.ID)
92+
if err != nil {
93+
metrics.NamedCounter(r.scope, "record", "storage_failure", 1, tag)
94+
return fmt.Errorf("failed to reconcile request log request_id=%q log_id=%q: %w", log.RequestID, log.ID, err)
95+
}
96+
if !sameOccurrence(stored, log) {
97+
metrics.NamedCounter(r.scope, "record", "conflict", 1, tag)
98+
return fmt.Errorf("request log conflicts with retained occurrence request_id=%q log_id=%q", log.RequestID, log.ID)
99+
}
100+
101+
metrics.NamedCounter(r.scope, "record", "identical_existing", 1, tag)
102+
return nil
103+
}
104+
105+
func occurrenceID(queue, requestID string, identity ...string) string {
106+
hash := sha256.New()
107+
parts := append([]string{queue, requestID}, identity...)
108+
var size [8]byte
109+
for _, part := range parts {
110+
binary.BigEndian.PutUint64(size[:], uint64(len(part)))
111+
_, _ = hash.Write(size[:])
112+
_, _ = hash.Write([]byte(part))
113+
}
114+
return "log/" + hex.EncodeToString(hash.Sum(nil))
115+
}
116+
117+
func sameOccurrence(stored, candidate entity.RequestLog) bool {
118+
// The first successful insert owns display time; retries compare only the occurrence's domain content.
119+
storedMetadata := stored.Metadata
120+
candidateMetadata := candidate.Metadata
121+
stored.Metadata = nil
122+
candidate.Metadata = nil
123+
stored.TimestampMs = 0
124+
candidate.TimestampMs = 0
125+
return reflect.DeepEqual(stored, candidate) && maps.Equal(storedMetadata, candidateMetadata)
126+
}
127+
128+
func occurrenceTag(log entity.RequestLog) metrics.Tag {
129+
value := "invalid"
130+
switch log.State {
131+
case entity.RequestStateAccepted,
132+
entity.RequestStateProcessing,
133+
entity.RequestStateSuperseded,
134+
entity.RequestStateSucceeded,
135+
entity.RequestStateFailed,
136+
entity.RequestStateCancelled:
137+
value = string(log.State)
138+
}
139+
return metrics.NewTag("occurrence", value)
140+
}

0 commit comments

Comments
 (0)