Skip to content

Commit bbabbb7

Browse files
committed
feat(stovepipe): expose request history RPCs
Summary: Intent: - Make retained Stovepipe request history available through both published selectors. - Keep request-log presence authoritative without consulting operational request rows. Changes: - Wire the shared request-history controller into thin gRPC handlers. - Cover transport delegation and MySQL-backed history behavior for both selectors. - Document runnable grpcurl examples and retained-history discovery semantics. This PR builds on #671, which adds history lookup by URI. --- <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 0303c73 commit bbabbb7

5 files changed

Lines changed: 290 additions & 5 deletions

File tree

service/stovepipe/README.md

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,11 @@
11
# Stovepipe Service
22

3-
Runnable wiring for the **Stovepipe** domain — a single-service domain (the domain *is* the service). The server exposes two RPCs and runs the internal pipeline stages as queue consumers:
3+
Runnable wiring for the **Stovepipe** domain — a single-service domain (the domain *is* the service). The server exposes four RPCs and runs the internal pipeline stages as queue consumers:
44

55
- **`Ping`** — health check.
66
- **`Ingest`** — resolves a queue's head commit, persists a `Request` (and its head URI) to storage, and publishes the request to the **process** stage.
7+
- **`GetRequestHistoryByID`** — returns the retained request log for one request ID.
8+
- **`GetRequestHistoryByURI`** — returns retained histories selected by an exact commit URI.
79
- **process consumer** (`TopicKeyProcess`) — reloads the persisted `Request` from storage and runs the process stage (`stovepipe/controller/process`).
810
- **build consumer** (`TopicKeyBuild`) — reloads the persisted `Request` and triggers the build-runner, then publishes to `buildsignal`.
911
- **buildsignal consumer** (`TopicKeyBuildSignal`) — polls/records the build's terminal status and releases the queue's in-flight slot, then publishes to `record`.
@@ -70,8 +72,16 @@ Attach with `.vscode/launch.json` (**Debug: attach (dlv in docker)**), then send
7072
```bash
7173
# Ingest example
7274
grpcurl -plaintext -d '{"queue":"monorepo/main"}' localhost:PORT uber.submitqueue.stovepipe.Stovepipe/Ingest
75+
76+
# Retained history by request ID
77+
grpcurl -plaintext -d '{"queue":"monorepo/main","request_id":"request/monorepo/main/1"}' localhost:PORT uber.submitqueue.stovepipe.Stovepipe/GetRequestHistoryByID
78+
79+
# Retained history by exact commit URI
80+
grpcurl -plaintext -d '{"queue":"monorepo/main","uri":"git://monorepo/main/HEAD"}' localhost:PORT uber.submitqueue.stovepipe.Stovepipe/GetRequestHistoryByURI
7381
```
7482

83+
History lookup is defined by retained `request_log` rows. A request with no retained rows is not discoverable through these RPCs, even if operational request data still exists.
84+
7585
### Bazel / Go
7686

7787
```bash

service/stovepipe/server/BUILD.bazel

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -79,8 +79,11 @@ go_test(
7979
embed = [":go_default_library"],
8080
deps = [
8181
"//api/base/hook:go_default_library",
82+
"//api/stovepipe/protopb:go_default_library",
8283
"//platform/consumer:go_default_library",
84+
"//stovepipe/controller:go_default_library",
8385
"//stovepipe/controller/dlq:go_default_library",
86+
"//stovepipe/entity:go_default_library",
8487
"@com_github_stretchr_testify//assert:go_default_library",
8588
"@com_github_stretchr_testify//require:go_default_library",
8689
"@com_github_uber_go_tally//:go_default_library",

service/stovepipe/server/main.go

Lines changed: 25 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -67,8 +67,9 @@ import (
6767
// StovepipeServer wraps the controllers and implements the gRPC service interface.
6868
type StovepipeServer struct {
6969
pb.UnimplementedStovepipeServer
70-
pingController *controller.PingController
71-
ingestController *controller.IngestController
70+
pingController *controller.PingController
71+
ingestController *controller.IngestController
72+
requestHistoryController controller.RequestHistoryController
7273
}
7374

7475
// Ping delegates to the controller.
@@ -86,6 +87,24 @@ func (s *StovepipeServer) Ingest(ctx context.Context, req *pb.IngestRequest) (*p
8687
return mapper.IngestResultToProto(result), nil
8788
}
8889

90+
// GetRequestHistoryByID returns retained history for one request ID.
91+
func (s *StovepipeServer) GetRequestHistoryByID(ctx context.Context, req *pb.GetRequestHistoryByIDRequest) (*pb.GetRequestHistoryByIDResponse, error) {
92+
events, err := s.requestHistoryController.GetRequestHistoryByID(ctx, mapper.ProtoToGetRequestHistoryByIDRequest(req))
93+
if err != nil {
94+
return nil, err
95+
}
96+
return &pb.GetRequestHistoryByIDResponse{Events: mapper.HistoryEventsToProto(events)}, nil
97+
}
98+
99+
// GetRequestHistoryByURI returns retained histories for one commit URI.
100+
func (s *StovepipeServer) GetRequestHistoryByURI(ctx context.Context, req *pb.GetRequestHistoryByURIRequest) (*pb.GetRequestHistoryByURIResponse, error) {
101+
histories, err := s.requestHistoryController.GetRequestHistoryByURI(ctx, mapper.ProtoToGetRequestHistoryByURIRequest(req))
102+
if err != nil {
103+
return nil, err
104+
}
105+
return &pb.GetRequestHistoryByURIResponse{Histories: mapper.RequestHistoriesToProto(histories)}, nil
106+
}
107+
89108
// inMemoryCounter is a minimal, process-local counter.Counter used to wire the example
90109
// server. It is not durable; a real deployment supplies a persistent implementation
91110
// (e.g. platform/extension/counter/mysql).
@@ -341,9 +360,11 @@ func run() error {
341360
materializer,
342361
registry,
343362
)
363+
requestHistoryController := controller.NewRequestHistoryController(logger.Sugar(), scope, storageFty)
344364
srv := &StovepipeServer{
345-
pingController: pingController,
346-
ingestController: ingestController,
365+
pingController: pingController,
366+
ingestController: ingestController,
367+
requestHistoryController: requestHistoryController,
347368
}
348369
pb.RegisterStovepipeServer(grpcServer, srv)
349370

service/stovepipe/server/main_test.go

Lines changed: 116 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,18 +16,134 @@ package main
1616

1717
import (
1818
"context"
19+
"errors"
1920
"strings"
2021
"testing"
2122

2223
"github.com/stretchr/testify/assert"
2324
"github.com/stretchr/testify/require"
2425
"github.com/uber-go/tally"
2526
basehook "github.com/uber/submitqueue/api/base/hook"
27+
pb "github.com/uber/submitqueue/api/stovepipe/protopb"
2628
"github.com/uber/submitqueue/platform/consumer"
29+
"github.com/uber/submitqueue/stovepipe/controller"
2730
"github.com/uber/submitqueue/stovepipe/controller/dlq"
31+
"github.com/uber/submitqueue/stovepipe/entity"
2832
"go.uber.org/zap/zaptest"
2933
)
3034

35+
type fakeRequestHistoryController struct {
36+
getByID func(context.Context, entity.GetRequestHistoryByIDRequest) ([]entity.RequestLog, error)
37+
getByURI func(context.Context, entity.GetRequestHistoryByURIRequest) ([]entity.RequestHistory, error)
38+
}
39+
40+
var _ controller.RequestHistoryController = (*fakeRequestHistoryController)(nil)
41+
42+
func (f *fakeRequestHistoryController) GetRequestHistoryByID(ctx context.Context, req entity.GetRequestHistoryByIDRequest) ([]entity.RequestLog, error) {
43+
return f.getByID(ctx, req)
44+
}
45+
46+
func (f *fakeRequestHistoryController) GetRequestHistoryByURI(ctx context.Context, req entity.GetRequestHistoryByURIRequest) ([]entity.RequestHistory, error) {
47+
return f.getByURI(ctx, req)
48+
}
49+
50+
func TestGetRequestHistoryByID(t *testing.T) {
51+
controllerErr := errors.New("controller failed")
52+
logs := []entity.RequestLog{
53+
{ID: "occurrence/1", State: entity.RequestStateAccepted, TimestampMs: 1000},
54+
{ID: "occurrence/2", Event: entity.RequestEventBuildTriggered, TimestampMs: 2000},
55+
}
56+
tests := []struct {
57+
name string
58+
logs []entity.RequestLog
59+
err error
60+
wantLogs int
61+
}{
62+
{name: "maps successful result", logs: logs, wantLogs: 2},
63+
{name: "maps empty result", logs: nil, wantLogs: 0},
64+
{name: "returns controller error unchanged", err: controllerErr},
65+
}
66+
67+
for _, tt := range tests {
68+
t.Run(tt.name, func(t *testing.T) {
69+
var gotReq entity.GetRequestHistoryByIDRequest
70+
fake := &fakeRequestHistoryController{
71+
getByID: func(_ context.Context, req entity.GetRequestHistoryByIDRequest) ([]entity.RequestLog, error) {
72+
gotReq = req
73+
return tt.logs, tt.err
74+
},
75+
}
76+
srv := &StovepipeServer{requestHistoryController: fake}
77+
78+
resp, err := srv.GetRequestHistoryByID(context.Background(), &pb.GetRequestHistoryByIDRequest{
79+
Queue: "monorepo/main", RequestId: "request/1",
80+
})
81+
82+
assert.Equal(t, entity.GetRequestHistoryByIDRequest{Queue: "monorepo/main", ID: "request/1"}, gotReq)
83+
if tt.err != nil {
84+
require.ErrorIs(t, err, tt.err)
85+
assert.Nil(t, resp)
86+
return
87+
}
88+
require.NoError(t, err)
89+
require.Len(t, resp.Events, tt.wantLogs)
90+
if tt.wantLogs > 0 {
91+
assert.Equal(t, "accepted", resp.Events[0].GetRequestState())
92+
assert.Equal(t, "build_triggered", resp.Events[1].GetEvent())
93+
}
94+
})
95+
}
96+
}
97+
98+
func TestGetRequestHistoryByURI(t *testing.T) {
99+
controllerErr := errors.New("controller failed")
100+
histories := []entity.RequestHistory{{
101+
RequestID: "request/1",
102+
Events: []entity.RequestLog{{ID: "occurrence/1", State: entity.RequestStateAccepted}},
103+
}}
104+
tests := []struct {
105+
name string
106+
histories []entity.RequestHistory
107+
err error
108+
wantHistories int
109+
}{
110+
{name: "maps successful result", histories: histories, wantHistories: 1},
111+
{name: "maps empty result", histories: nil, wantHistories: 0},
112+
{name: "returns controller error unchanged", err: controllerErr},
113+
}
114+
115+
for _, tt := range tests {
116+
t.Run(tt.name, func(t *testing.T) {
117+
var gotReq entity.GetRequestHistoryByURIRequest
118+
fake := &fakeRequestHistoryController{
119+
getByURI: func(_ context.Context, req entity.GetRequestHistoryByURIRequest) ([]entity.RequestHistory, error) {
120+
gotReq = req
121+
return tt.histories, tt.err
122+
},
123+
}
124+
srv := &StovepipeServer{requestHistoryController: fake}
125+
126+
resp, err := srv.GetRequestHistoryByURI(context.Background(), &pb.GetRequestHistoryByURIRequest{
127+
Queue: "monorepo/main", Uri: "git://monorepo/abc",
128+
})
129+
130+
assert.Equal(t, entity.GetRequestHistoryByURIRequest{Queue: "monorepo/main", URI: "git://monorepo/abc"}, gotReq)
131+
if tt.err != nil {
132+
require.ErrorIs(t, err, tt.err)
133+
assert.Nil(t, resp)
134+
return
135+
}
136+
require.NoError(t, err)
137+
require.Len(t, resp.Histories, tt.wantHistories)
138+
if tt.wantHistories > 0 {
139+
assert.Equal(t, "request/1", resp.Histories[0].RequestId)
140+
require.Len(t, resp.Histories[0].Events, 1)
141+
assert.Equal(t, "accepted", resp.Histories[0].Events[0].GetRequestState())
142+
}
143+
})
144+
}
145+
}
146+
31147
// recordingConsumer captures what the host registers instead of subscribing.
32148
type recordingConsumer struct {
33149
controllers []consumer.Controller

test/integration/stovepipe/suite_test.go

Lines changed: 135 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -153,3 +153,138 @@ func (s *StovepipeIntegrationSuite) TestIngestEmptyQueue() {
153153
_, err := s.client.Ingest(s.ctx, &pb.IngestRequest{Queue: ""})
154154
require.Error(t, err, "Ingest with empty queue should fail")
155155
}
156+
157+
func (s *StovepipeIntegrationSuite) TestRequestHistoryAPIs() {
158+
t := s.T()
159+
const (
160+
queue = "history-api/main"
161+
requestID = "request/history-api/main/7"
162+
uri = "git://history-api/main/abc123"
163+
)
164+
165+
_, err := s.db.Exec(
166+
"INSERT INTO request_uri (queue, uri, request_id, version) VALUES (?, ?, ?, ?)",
167+
queue, uri, requestID, 1,
168+
)
169+
require.NoError(t, err)
170+
_, err = s.db.Exec(
171+
`INSERT INTO request_log
172+
(queue, request_id, log_id, timestamp_ms, state, event, request_version, outcome_reason, metadata)
173+
VALUES
174+
(?, ?, ?, ?, ?, ?, ?, ?, ?),
175+
(?, ?, ?, ?, ?, ?, ?, ?, ?),
176+
(?, ?, ?, ?, ?, ?, ?, ?, ?)`,
177+
queue, requestID, "occurrence/001", 1000, "accepted", "", 1, "", `{}`,
178+
queue, requestID, "occurrence/002", 2000, "", "build_triggered", 0, "", `{}`,
179+
queue, requestID, "occurrence/003", 2000, "succeeded", "", 3, "build_succeeded", `{}`,
180+
)
181+
require.NoError(t, err)
182+
183+
var requestRows int
184+
require.NoError(t, s.db.QueryRow("SELECT COUNT(*) FROM request WHERE id = ?", requestID).Scan(&requestRows))
185+
require.Zero(t, requestRows)
186+
187+
byID, err := s.client.GetRequestHistoryByID(s.ctx, &pb.GetRequestHistoryByIDRequest{Queue: queue, RequestId: requestID})
188+
require.NoError(t, err)
189+
assertHistoryEvents(t, byID.Events)
190+
191+
byURI, err := s.client.GetRequestHistoryByURI(s.ctx, &pb.GetRequestHistoryByURIRequest{Queue: queue, Uri: uri})
192+
require.NoError(t, err)
193+
require.Len(t, byURI.Histories, 1)
194+
assert.Equal(t, requestID, byURI.Histories[0].RequestId)
195+
assertHistoryEvents(t, byURI.Histories[0].Events)
196+
}
197+
198+
func (s *StovepipeIntegrationSuite) TestRequestHistoryAbsence() {
199+
t := s.T()
200+
const (
201+
queue = "history-api-absence/main"
202+
mappedID = "request/history-api-absence/main/1"
203+
mappedURI = "git://history-api-absence/main/mapped"
204+
scopedID = "request/history-api-absence/main/2"
205+
scopedURI = "git://history-api-absence/main/scoped"
206+
missingID = "request/history-api-absence/main/missing"
207+
missingURI = "git://history-api-absence/main/missing"
208+
wrongQueue = "history-api-absence/other"
209+
)
210+
211+
_, err := s.db.Exec(
212+
"INSERT INTO request_uri (queue, uri, request_id, version) VALUES (?, ?, ?, ?), (?, ?, ?, ?)",
213+
queue, mappedURI, mappedID, 1, queue, scopedURI, scopedID, 1,
214+
)
215+
require.NoError(t, err)
216+
_, err = s.db.Exec(
217+
`INSERT INTO request_log
218+
(queue, request_id, log_id, timestamp_ms, state, event, request_version, outcome_reason, metadata)
219+
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
220+
queue, scopedID, "occurrence/001", 1000, "accepted", "", 1, "", `{}`,
221+
)
222+
require.NoError(t, err)
223+
224+
tests := []struct {
225+
name string
226+
call func() error
227+
}{
228+
{
229+
name: "missing request ID",
230+
call: func() error {
231+
_, err := s.client.GetRequestHistoryByID(s.ctx, &pb.GetRequestHistoryByIDRequest{Queue: queue, RequestId: missingID})
232+
return err
233+
},
234+
},
235+
{
236+
name: "missing URI mapping",
237+
call: func() error {
238+
_, err := s.client.GetRequestHistoryByURI(s.ctx, &pb.GetRequestHistoryByURIRequest{Queue: queue, Uri: missingURI})
239+
return err
240+
},
241+
},
242+
{
243+
name: "mapped URI without retained logs",
244+
call: func() error {
245+
_, err := s.client.GetRequestHistoryByURI(s.ctx, &pb.GetRequestHistoryByURIRequest{Queue: queue, Uri: mappedURI})
246+
return err
247+
},
248+
},
249+
{
250+
name: "request ID in wrong queue",
251+
call: func() error {
252+
_, err := s.client.GetRequestHistoryByID(s.ctx, &pb.GetRequestHistoryByIDRequest{Queue: wrongQueue, RequestId: scopedID})
253+
return err
254+
},
255+
},
256+
{
257+
name: "URI mapping in wrong queue",
258+
call: func() error {
259+
_, err := s.client.GetRequestHistoryByURI(s.ctx, &pb.GetRequestHistoryByURIRequest{Queue: wrongQueue, Uri: scopedURI})
260+
return err
261+
},
262+
},
263+
}
264+
265+
for _, tt := range tests {
266+
t.Run(tt.name, func(t *testing.T) {
267+
require.Error(t, tt.call())
268+
})
269+
}
270+
}
271+
272+
func assertHistoryEvents(t *testing.T, events []*pb.HistoryEvent) {
273+
t.Helper()
274+
require.Len(t, events, 3)
275+
276+
assert.Equal(t, "occurrence/001", events[0].EventId)
277+
assert.Equal(t, int64(1000), events[0].TimestampMs)
278+
assert.Equal(t, "accepted", events[0].GetRequestState())
279+
assert.Empty(t, events[0].GetEvent())
280+
281+
assert.Equal(t, "occurrence/002", events[1].EventId)
282+
assert.Equal(t, int64(2000), events[1].TimestampMs)
283+
assert.Equal(t, "build_triggered", events[1].GetEvent())
284+
assert.Empty(t, events[1].GetRequestState())
285+
286+
assert.Equal(t, "occurrence/003", events[2].EventId)
287+
assert.Equal(t, int64(2000), events[2].TimestampMs)
288+
assert.Equal(t, "succeeded", events[2].GetRequestState())
289+
assert.Equal(t, "build_succeeded", events[2].OutcomeReason)
290+
}

0 commit comments

Comments
 (0)