feat(stovepipe): request history api - controller wiring

service/stovepipe/README.md

@@ -1,9 +1,11 @@
 # Stovepipe Service
 
-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:
+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:
 
 - **`Ping`** — health check.
 - **`Ingest`** — resolves a queue's head commit, persists a `Request` (and its head URI) to storage, and publishes the request to the **process** stage.
+- **`GetRequestHistoryByID`** — returns the retained request log for one request ID.
+- **`GetRequestHistoryByURI`** — returns retained histories selected by an exact commit URI.
 - **process consumer** (`TopicKeyProcess`) — reloads the persisted `Request` from storage and runs the process stage (`stovepipe/controller/process`).
 - **build consumer** (`TopicKeyBuild`) — reloads the persisted `Request` and triggers the build-runner, then publishes to `buildsignal`.
 - **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
 ```bash
 # Ingest example
 grpcurl -plaintext -d '{"queue":"monorepo/main"}' localhost:PORT uber.submitqueue.stovepipe.Stovepipe/Ingest
+
+# Retained history by request ID
+grpcurl -plaintext -d '{"queue":"monorepo/main","request_id":"request/monorepo/main/1"}' localhost:PORT uber.submitqueue.stovepipe.Stovepipe/GetRequestHistoryByID
+
+# Retained history by exact commit URI
+grpcurl -plaintext -d '{"queue":"monorepo/main","uri":"git://monorepo/main/HEAD"}' localhost:PORT uber.submitqueue.stovepipe.Stovepipe/GetRequestHistoryByURI
 ```
 
+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.
+
 ### Bazel / Go
 
 ```bash

service/stovepipe/server/BUILD.bazel

@@ -79,8 +79,11 @@ go_test(
     embed = [":go_default_library"],
     deps = [
         "//api/base/hook:go_default_library",
+        "//api/stovepipe/protopb:go_default_library",
         "//platform/consumer:go_default_library",
+        "//stovepipe/controller:go_default_library",
         "//stovepipe/controller/dlq:go_default_library",
+        "//stovepipe/entity: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",

service/stovepipe/server/main.go

@@ -67,8 +67,9 @@ import (
 // StovepipeServer wraps the controllers and implements the gRPC service interface.
 type StovepipeServer struct {
 	pb.UnimplementedStovepipeServer
-	pingController   *controller.PingController
-	ingestController *controller.IngestController
+	pingController           *controller.PingController
+	ingestController         *controller.IngestController
+	requestHistoryController controller.RequestHistoryController
 }
 
 // Ping delegates to the controller.
@@ -86,6 +87,24 @@ func (s *StovepipeServer) Ingest(ctx context.Context, req *pb.IngestRequest) (*p
 	return mapper.IngestResultToProto(result), nil
 }
 
+// GetRequestHistoryByID returns retained history for one request ID.
+func (s *StovepipeServer) GetRequestHistoryByID(ctx context.Context, req *pb.GetRequestHistoryByIDRequest) (*pb.GetRequestHistoryByIDResponse, error) {
+	events, err := s.requestHistoryController.GetRequestHistoryByID(ctx, mapper.ProtoToGetRequestHistoryByIDRequest(req))
+	if err != nil {
+		return nil, err
+	}
+	return &pb.GetRequestHistoryByIDResponse{Events: mapper.HistoryEventsToProto(events)}, nil
+}
+
+// GetRequestHistoryByURI returns retained histories for one commit URI.
+func (s *StovepipeServer) GetRequestHistoryByURI(ctx context.Context, req *pb.GetRequestHistoryByURIRequest) (*pb.GetRequestHistoryByURIResponse, error) {
+	histories, err := s.requestHistoryController.GetRequestHistoryByURI(ctx, mapper.ProtoToGetRequestHistoryByURIRequest(req))
+	if err != nil {
+		return nil, err
+	}
+	return &pb.GetRequestHistoryByURIResponse{Histories: mapper.RequestHistoriesToProto(histories)}, nil
+}
+
 // inMemoryCounter is a minimal, process-local counter.Counter used to wire the example
 // server. It is not durable; a real deployment supplies a persistent implementation
 // (e.g. platform/extension/counter/mysql).
@@ -341,9 +360,11 @@ func run() error {
 		materializer,
 		registry,
 	)
+	requestHistoryController := controller.NewRequestHistoryController(logger.Sugar(), scope, storageFty)
 	srv := &StovepipeServer{
-		pingController:   pingController,
-		ingestController: ingestController,
+		pingController:           pingController,
+		ingestController:         ingestController,
+		requestHistoryController: requestHistoryController,
 	}
 	pb.RegisterStovepipeServer(grpcServer, srv)
 

service/stovepipe/server/main_test.go

@@ -16,18 +16,134 @@ package main
 
 import (
 	"context"
+	"errors"
 	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/uber-go/tally"
 	basehook "github.com/uber/submitqueue/api/base/hook"
+	pb "github.com/uber/submitqueue/api/stovepipe/protopb"
 	"github.com/uber/submitqueue/platform/consumer"
+	"github.com/uber/submitqueue/stovepipe/controller"
 	"github.com/uber/submitqueue/stovepipe/controller/dlq"
+	"github.com/uber/submitqueue/stovepipe/entity"
 	"go.uber.org/zap/zaptest"
 )
 
+type fakeRequestHistoryController struct {
+	getByID  func(context.Context, entity.GetRequestHistoryByIDRequest) ([]entity.RequestLog, error)
+	getByURI func(context.Context, entity.GetRequestHistoryByURIRequest) ([]entity.RequestHistory, error)
+}
+
+var _ controller.RequestHistoryController = (*fakeRequestHistoryController)(nil)
+
+func (f *fakeRequestHistoryController) GetRequestHistoryByID(ctx context.Context, req entity.GetRequestHistoryByIDRequest) ([]entity.RequestLog, error) {
+	return f.getByID(ctx, req)
+}
+
+func (f *fakeRequestHistoryController) GetRequestHistoryByURI(ctx context.Context, req entity.GetRequestHistoryByURIRequest) ([]entity.RequestHistory, error) {
+	return f.getByURI(ctx, req)
+}
+
+func TestGetRequestHistoryByID(t *testing.T) {
+	controllerErr := errors.New("controller failed")
+	logs := []entity.RequestLog{
+		{ID: "occurrence/1", State: entity.RequestStateAccepted, TimestampMs: 1000},
+		{ID: "occurrence/2", Event: entity.RequestEventBuildTriggered, TimestampMs: 2000},
+	}
+	tests := []struct {
+		name     string
+		logs     []entity.RequestLog
+		err      error
+		wantLogs int
+	}{
+		{name: "maps successful result", logs: logs, wantLogs: 2},
+		{name: "maps empty result", logs: nil, wantLogs: 0},
+		{name: "returns controller error unchanged", err: controllerErr},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var gotReq entity.GetRequestHistoryByIDRequest
+			fake := &fakeRequestHistoryController{
+				getByID: func(_ context.Context, req entity.GetRequestHistoryByIDRequest) ([]entity.RequestLog, error) {
+					gotReq = req
+					return tt.logs, tt.err
+				},
+			}
+			srv := &StovepipeServer{requestHistoryController: fake}
+
+			resp, err := srv.GetRequestHistoryByID(context.Background(), &pb.GetRequestHistoryByIDRequest{
+				Queue: "monorepo/main", RequestId: "request/1",
+			})
+
+			assert.Equal(t, entity.GetRequestHistoryByIDRequest{Queue: "monorepo/main", ID: "request/1"}, gotReq)
+			if tt.err != nil {
+				require.ErrorIs(t, err, tt.err)
+				assert.Nil(t, resp)
+				return
+			}
+			require.NoError(t, err)
+			require.Len(t, resp.Events, tt.wantLogs)
+			if tt.wantLogs > 0 {
+				assert.Equal(t, "accepted", resp.Events[0].GetRequestState())
+				assert.Equal(t, "build_triggered", resp.Events[1].GetEvent())
+			}
+		})
+	}
+}
+
+func TestGetRequestHistoryByURI(t *testing.T) {
+	controllerErr := errors.New("controller failed")
+	histories := []entity.RequestHistory{{
+		RequestID: "request/1",
+		Events:    []entity.RequestLog{{ID: "occurrence/1", State: entity.RequestStateAccepted}},
+	}}
+	tests := []struct {
+		name          string
+		histories     []entity.RequestHistory
+		err           error
+		wantHistories int
+	}{
+		{name: "maps successful result", histories: histories, wantHistories: 1},
+		{name: "maps empty result", histories: nil, wantHistories: 0},
+		{name: "returns controller error unchanged", err: controllerErr},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var gotReq entity.GetRequestHistoryByURIRequest
+			fake := &fakeRequestHistoryController{
+				getByURI: func(_ context.Context, req entity.GetRequestHistoryByURIRequest) ([]entity.RequestHistory, error) {
+					gotReq = req
+					return tt.histories, tt.err
+				},
+			}
+			srv := &StovepipeServer{requestHistoryController: fake}
+
+			resp, err := srv.GetRequestHistoryByURI(context.Background(), &pb.GetRequestHistoryByURIRequest{
+				Queue: "monorepo/main", Uri: "git://monorepo/abc",
+			})
+
+			assert.Equal(t, entity.GetRequestHistoryByURIRequest{Queue: "monorepo/main", URI: "git://monorepo/abc"}, gotReq)
+			if tt.err != nil {
+				require.ErrorIs(t, err, tt.err)
+				assert.Nil(t, resp)
+				return
+			}
+			require.NoError(t, err)
+			require.Len(t, resp.Histories, tt.wantHistories)
+			if tt.wantHistories > 0 {
+				assert.Equal(t, "request/1", resp.Histories[0].RequestId)
+				require.Len(t, resp.Histories[0].Events, 1)
+				assert.Equal(t, "accepted", resp.Histories[0].Events[0].GetRequestState())
+			}
+		})
+	}
+}
+
 // recordingConsumer captures what the host registers instead of subscribing.
 type recordingConsumer struct {
 	controllers []consumer.Controller

test/integration/stovepipe/suite_test.go

@@ -153,3 +153,138 @@ func (s *StovepipeIntegrationSuite) TestIngestEmptyQueue() {
 	_, err := s.client.Ingest(s.ctx, &pb.IngestRequest{Queue: ""})
 	require.Error(t, err, "Ingest with empty queue should fail")
 }
+
+func (s *StovepipeIntegrationSuite) TestRequestHistoryAPIs() {
+	t := s.T()
+	const (
+		queue     = "history-api/main"
+		requestID = "request/history-api/main/7"
+		uri       = "git://history-api/main/abc123"
+	)
+
+	_, err := s.db.Exec(
+		"INSERT INTO request_uri (queue, uri, request_id, version) VALUES (?, ?, ?, ?)",
+		queue, uri, requestID, 1,
+	)
+	require.NoError(t, err)
+	_, err = s.db.Exec(
+		`INSERT INTO request_log
+			(queue, request_id, log_id, timestamp_ms, state, event, request_version, outcome_reason, metadata)
+		 VALUES
+			(?, ?, ?, ?, ?, ?, ?, ?, ?),
+			(?, ?, ?, ?, ?, ?, ?, ?, ?),
+			(?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+		queue, requestID, "occurrence/001", 1000, "accepted", "", 1, "", `{}`,
+		queue, requestID, "occurrence/002", 2000, "", "build_triggered", 0, "", `{}`,
+		queue, requestID, "occurrence/003", 2000, "succeeded", "", 3, "build_succeeded", `{}`,
+	)
+	require.NoError(t, err)
+
+	var requestRows int
+	require.NoError(t, s.db.QueryRow("SELECT COUNT(*) FROM request WHERE id = ?", requestID).Scan(&requestRows))
+	require.Zero(t, requestRows)
+
+	byID, err := s.client.GetRequestHistoryByID(s.ctx, &pb.GetRequestHistoryByIDRequest{Queue: queue, RequestId: requestID})
+	require.NoError(t, err)
+	assertHistoryEvents(t, byID.Events)
+
+	byURI, err := s.client.GetRequestHistoryByURI(s.ctx, &pb.GetRequestHistoryByURIRequest{Queue: queue, Uri: uri})
+	require.NoError(t, err)
+	require.Len(t, byURI.Histories, 1)
+	assert.Equal(t, requestID, byURI.Histories[0].RequestId)
+	assertHistoryEvents(t, byURI.Histories[0].Events)
+}
+
+func (s *StovepipeIntegrationSuite) TestRequestHistoryAbsence() {
+	t := s.T()
+	const (
+		queue      = "history-api-absence/main"
+		mappedID   = "request/history-api-absence/main/1"
+		mappedURI  = "git://history-api-absence/main/mapped"
+		scopedID   = "request/history-api-absence/main/2"
+		scopedURI  = "git://history-api-absence/main/scoped"
+		missingID  = "request/history-api-absence/main/missing"
+		missingURI = "git://history-api-absence/main/missing"
+		wrongQueue = "history-api-absence/other"
+	)
+
+	_, err := s.db.Exec(
+		"INSERT INTO request_uri (queue, uri, request_id, version) VALUES (?, ?, ?, ?), (?, ?, ?, ?)",
+		queue, mappedURI, mappedID, 1, queue, scopedURI, scopedID, 1,
+	)
+	require.NoError(t, err)
+	_, err = s.db.Exec(
+		`INSERT INTO request_log
+			(queue, request_id, log_id, timestamp_ms, state, event, request_version, outcome_reason, metadata)
+		 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+		queue, scopedID, "occurrence/001", 1000, "accepted", "", 1, "", `{}`,
+	)
+	require.NoError(t, err)
+
+	tests := []struct {
+		name string
+		call func() error
+	}{
+		{
+			name: "missing request ID",
+			call: func() error {
+				_, err := s.client.GetRequestHistoryByID(s.ctx, &pb.GetRequestHistoryByIDRequest{Queue: queue, RequestId: missingID})
+				return err
+			},
+		},
+		{
+			name: "missing URI mapping",
+			call: func() error {
+				_, err := s.client.GetRequestHistoryByURI(s.ctx, &pb.GetRequestHistoryByURIRequest{Queue: queue, Uri: missingURI})
+				return err
+			},
+		},
+		{
+			name: "mapped URI without retained logs",
+			call: func() error {
+				_, err := s.client.GetRequestHistoryByURI(s.ctx, &pb.GetRequestHistoryByURIRequest{Queue: queue, Uri: mappedURI})
+				return err
+			},
+		},
+		{
+			name: "request ID in wrong queue",
+			call: func() error {
+				_, err := s.client.GetRequestHistoryByID(s.ctx, &pb.GetRequestHistoryByIDRequest{Queue: wrongQueue, RequestId: scopedID})
+				return err
+			},
+		},
+		{
+			name: "URI mapping in wrong queue",
+			call: func() error {
+				_, err := s.client.GetRequestHistoryByURI(s.ctx, &pb.GetRequestHistoryByURIRequest{Queue: wrongQueue, Uri: scopedURI})
+				return err
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			require.Error(t, tt.call())
+		})
+	}
+}
+
+func assertHistoryEvents(t *testing.T, events []*pb.HistoryEvent) {
+	t.Helper()
+	require.Len(t, events, 3)
+
+	assert.Equal(t, "occurrence/001", events[0].EventId)
+	assert.Equal(t, int64(1000), events[0].TimestampMs)
+	assert.Equal(t, "accepted", events[0].GetRequestState())
+	assert.Empty(t, events[0].GetEvent())
+
+	assert.Equal(t, "occurrence/002", events[1].EventId)
+	assert.Equal(t, int64(2000), events[1].TimestampMs)
+	assert.Equal(t, "build_triggered", events[1].GetEvent())
+	assert.Empty(t, events[1].GetRequestState())
+
+	assert.Equal(t, "occurrence/003", events[2].EventId)
+	assert.Equal(t, int64(2000), events[2].TimestampMs)
+	assert.Equal(t, "succeeded", events[2].GetRequestState())
+	assert.Equal(t, "build_succeeded", events[2].OutcomeReason)
+}