From 651cd5b966a277d1470981638815e3a42f868d91 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Wed, 2 Sep 2026 23:29:06 +0000 Subject: [PATCH 1/2] feat(stovepipe): read request history by URI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Summary: Intent: - Support history lookup from an exact commit URI while preserving log-only authority. - Keep the plural response model ready for a future multi-attempt URI index. Changes: - Resolve the current URI mapping and load its retained request log. - Return one grouped history under the insert-once mapping contract. - Cover selector validation, missing history, infrastructure errors, and metric tags. This PR builds on #670, which adds request-ID history reads. --- 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 --- stovepipe/controller/read_errors.go | 4 + stovepipe/controller/request_history.go | 53 +++++++- stovepipe/controller/request_history_test.go | 133 +++++++++++++++++-- 3 files changed, 178 insertions(+), 12 deletions(-) diff --git a/stovepipe/controller/read_errors.go b/stovepipe/controller/read_errors.go index 1bbeff6b..d81153c1 100644 --- a/stovepipe/controller/read_errors.go +++ b/stovepipe/controller/read_errors.go @@ -44,10 +44,14 @@ func validateHistoryIdentifier(name, value string) error { // RequestHistoryNotFoundError indicates that no retained history exists for a selector. type RequestHistoryNotFoundError struct { RequestID string + URI string } // Error implements error. func (e *RequestHistoryNotFoundError) Error() string { + if e.RequestID == "" { + return fmt.Sprintf("request history not found for URI %q", e.URI) + } return fmt.Sprintf("request history not found for request ID %q", e.RequestID) } diff --git a/stovepipe/controller/request_history.go b/stovepipe/controller/request_history.go index 70f49cf0..c81db249 100644 --- a/stovepipe/controller/request_history.go +++ b/stovepipe/controller/request_history.go @@ -29,6 +29,7 @@ import ( // RequestHistoryController handles retained request-history lookups. type RequestHistoryController interface { GetRequestHistoryByID(ctx context.Context, req entity.GetRequestHistoryByIDRequest) ([]entity.RequestLog, error) + GetRequestHistoryByURI(ctx context.Context, req entity.GetRequestHistoryByURIRequest) ([]entity.RequestHistory, error) } var _ RequestHistoryController = (*requestHistoryController)(nil) @@ -78,13 +79,57 @@ func (c *requestHistoryController) readHistoryByID(ctx context.Context, req enti return nil, fmt.Errorf("GetRequestHistoryByID failed to resolve storage for queue %q: %w", req.Queue, err) } - logs, err := stores.GetRequestLogStore().List(ctx, req.ID) + logs, err := loadRequestLogs(ctx, stores.GetRequestLogStore(), req.ID, &RequestHistoryNotFoundError{RequestID: req.ID}) if err != nil { - if storage.IsNotFound(err) { - return nil, errs.NewUserError(&RequestHistoryNotFoundError{RequestID: req.ID}) - } return nil, fmt.Errorf("GetRequestHistoryByID failed to list request logs request_id=%s: %w", req.ID, err) } return logs, nil } + +// GetRequestHistoryByURI returns the retained history mapped to an exact commit URI. +func (c *requestHistoryController) GetRequestHistoryByURI(ctx context.Context, req entity.GetRequestHistoryByURIRequest) (histories []entity.RequestHistory, retErr error) { + op := metrics.Begin(c.metricsScope, "get_by_uri", metrics.StorageLatencyBuckets, metrics.TagsFromContext(ctx)...) + defer func() { op.Complete(retErr) }() + + if err := validateHistoryIdentifier("queue", req.Queue); err != nil { + return nil, fmt.Errorf("GetRequestHistoryByURI invalid queue: %w", err) + } + if err := validateHistoryIdentifier("URI", req.URI); err != nil { + return nil, fmt.Errorf("GetRequestHistoryByURI invalid request: %w", err) + } + + stores, err := c.stores.For(storage.Config{QueueName: req.Queue}) + if err != nil { + return nil, fmt.Errorf("GetRequestHistoryByURI failed to resolve storage for queue %q: %w", req.Queue, err) + } + + requestID, err := stores.GetRequestURIStore().GetIDByURI(ctx, req.URI) + if err != nil { + if storage.IsNotFound(err) { + return nil, errs.NewUserError(&RequestHistoryNotFoundError{URI: req.URI}) + } + return nil, fmt.Errorf("GetRequestHistoryByURI failed to resolve request URI %s: %w", req.URI, err) + } + + logs, err := loadRequestLogs(ctx, stores.GetRequestLogStore(), requestID, &RequestHistoryNotFoundError{URI: req.URI}) + if err != nil { + return nil, fmt.Errorf("GetRequestHistoryByURI failed to list request logs uri=%s request_id=%s: %w", req.URI, requestID, err) + } + + c.logger.Debugw("request history retrieved by URI", + "uri", req.URI, + "request_id", requestID, + "queue", req.Queue, + "event_count", len(logs), + ) + return []entity.RequestHistory{{RequestID: requestID, Events: logs}}, nil +} + +func loadRequestLogs(ctx context.Context, store storage.RequestLogStore, requestID string, notFound *RequestHistoryNotFoundError) ([]entity.RequestLog, error) { + logs, err := store.List(ctx, requestID) + if storage.IsNotFound(err) { + return nil, errs.NewUserError(notFound) + } + return logs, err +} diff --git a/stovepipe/controller/request_history_test.go b/stovepipe/controller/request_history_test.go index bd57e66b..dc37d4d0 100644 --- a/stovepipe/controller/request_history_test.go +++ b/stovepipe/controller/request_history_test.go @@ -125,29 +125,146 @@ func TestGetRequestHistoryByID(t *testing.T) { start, ok := snapshot.Counters()["test.request_history_controller.get_by_id.start+queue=context-queue"] require.True(t, ok) assert.EqualValues(t, 1, start.Value()) - assertOperationFinishIncludesContextTag(t, snapshot, err == nil) + assertOperationFinishIncludesContextTag(t, snapshot, "get_by_id", err == nil) + }) + } +} + +func TestGetRequestHistoryByURI(t *testing.T) { + const ( + queue = "monorepo/main" + uri = "git://example.com/repo.git/commit/deadbeef" + requestID = "request/monorepo/main/42" + ) + backendErr := errors.New("backend unavailable") + logs := []entity.RequestLog{ + {ID: "state/1", RequestID: requestID, TimestampMs: 10, State: entity.RequestStateAccepted}, + {ID: "event/a", RequestID: requestID, TimestampMs: 20, Event: entity.RequestEventBuildTriggered}, + {ID: "event/a", RequestID: requestID, TimestampMs: 20, Event: entity.RequestEventBuildTriggered}, + } + wantHistory := []entity.RequestHistory{{RequestID: requestID, Events: logs}} + + tests := []struct { + name string + req entity.GetRequestHistoryByURIRequest + mappedID string + factoryErr error + mappingErr error + listErr error + want []entity.RequestHistory + wantInvalid bool + wantNotFound bool + wantCause error + wantLog bool + }{ + {name: "singleton history preserves log order and duplicates", req: entity.GetRequestHistoryByURIRequest{Queue: queue, URI: uri}, mappedID: requestID, want: wantHistory, wantLog: true}, + {name: "empty queue", req: entity.GetRequestHistoryByURIRequest{URI: uri}, wantInvalid: true}, + {name: "oversized queue", req: entity.GetRequestHistoryByURIRequest{Queue: strings.Repeat("q", maxHistoryIdentifierBytes+1), URI: uri}, wantInvalid: true}, + {name: "empty URI", req: entity.GetRequestHistoryByURIRequest{Queue: queue}, wantInvalid: true}, + {name: "oversized URI", req: entity.GetRequestHistoryByURIRequest{Queue: queue, URI: strings.Repeat("u", maxHistoryIdentifierBytes+1)}, wantInvalid: true}, + {name: "storage factory failure", req: entity.GetRequestHistoryByURIRequest{Queue: queue, URI: uri}, factoryErr: backendErr, wantCause: backendErr}, + {name: "URI mapping not found", req: entity.GetRequestHistoryByURIRequest{Queue: queue, URI: uri}, mappingErr: fmt.Errorf("lookup: %w", storage.ErrNotFound), wantNotFound: true}, + {name: "URI store failure", req: entity.GetRequestHistoryByURIRequest{Queue: queue, URI: uri}, mappingErr: backendErr, wantCause: backendErr}, + {name: "mapped history not found", req: entity.GetRequestHistoryByURIRequest{Queue: queue, URI: uri}, mappedID: requestID, listErr: fmt.Errorf("query: %w", storage.ErrNotFound), wantNotFound: true}, + {name: "log store failure", req: entity.GetRequestHistoryByURIRequest{Queue: queue, URI: uri}, mappedID: requestID, listErr: backendErr, wantCause: backendErr}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + mockCtrl := gomock.NewController(t) + factory := storagemock.NewMockFactory(mockCtrl) + stores := storagemock.NewMockStorage(mockCtrl) + uriStore := storagemock.NewMockRequestURIStore(mockCtrl) + logStore := storagemock.NewMockRequestLogStore(mockCtrl) + if !tt.wantInvalid { + factory.EXPECT().For(storage.Config{QueueName: tt.req.Queue}).Return(stores, tt.factoryErr) + if tt.factoryErr == nil { + stores.EXPECT().GetRequestURIStore().Return(uriStore) + uriStore.EXPECT().GetIDByURI(gomock.Any(), tt.req.URI).Return(tt.mappedID, tt.mappingErr) + if tt.mappingErr == nil { + stores.EXPECT().GetRequestLogStore().Return(logStore) + logStore.EXPECT().List(gomock.Any(), tt.mappedID).Return(logs, tt.listErr) + } + } + } + + core, observed := observer.New(zap.DebugLevel) + scope := tally.NewTestScope("test", nil) + controller := NewRequestHistoryController(zap.New(core).Sugar(), scope, factory) + ctx := metrics.WithContextTags(context.Background(), metrics.NewTag("queue", "context-queue")) + + got, err := controller.GetRequestHistoryByURI(ctx, tt.req) + + assert.Equal(t, tt.want, got) + if tt.wantInvalid { + assert.True(t, IsInvalidRequest(err)) + } + assert.Equal(t, tt.wantNotFound, IsRequestHistoryNotFound(err)) + assert.Equal(t, tt.wantInvalid || tt.wantNotFound, errs.IsUserError(err)) + if tt.wantCause != nil { + assert.ErrorIs(t, err, tt.wantCause) + } + if tt.want != nil { + require.NoError(t, err) + } else { + require.Error(t, err) + } + if tt.wantNotFound { + var notFound *RequestHistoryNotFoundError + require.ErrorAs(t, err, ¬Found) + assert.Empty(t, notFound.RequestID) + assert.Equal(t, uri, notFound.URI) + } + + entries := observed.FilterMessage("request history retrieved by URI").All() + if tt.wantLog { + require.Len(t, entries, 1) + assert.Equal(t, uri, entries[0].ContextMap()["uri"]) + assert.Equal(t, requestID, entries[0].ContextMap()["request_id"]) + assert.Equal(t, queue, entries[0].ContextMap()["queue"]) + assert.Equal(t, int64(len(logs)), entries[0].ContextMap()["event_count"]) + } else { + assert.Empty(t, entries) + } + + snapshot := scope.Snapshot() + start, ok := snapshot.Counters()["test.request_history_controller.get_by_uri.start+queue=context-queue"] + require.True(t, ok) + assert.EqualValues(t, 1, start.Value()) + assertOperationFinishIncludesContextTag(t, snapshot, "get_by_uri", err == nil) }) } } func TestRequestHistoryNotFoundError(t *testing.T) { - err := fmt.Errorf("lookup failed: %w", &RequestHistoryNotFoundError{RequestID: "request/queue/1"}) + tests := []struct { + name string + err error + want RequestHistoryNotFoundError + }{ + {name: "request ID", err: fmt.Errorf("lookup failed: %w", &RequestHistoryNotFoundError{RequestID: "request/queue/1"}), want: RequestHistoryNotFoundError{RequestID: "request/queue/1"}}, + {name: "URI", err: fmt.Errorf("lookup failed: %w", &RequestHistoryNotFoundError{URI: "git://repo/commit/1"}), want: RequestHistoryNotFoundError{URI: "git://repo/commit/1"}}, + } - assert.True(t, IsRequestHistoryNotFound(err)) + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.True(t, IsRequestHistoryNotFound(tt.err)) + var notFound *RequestHistoryNotFoundError + require.ErrorAs(t, tt.err, ¬Found) + assert.Equal(t, tt.want, *notFound) + }) + } assert.False(t, IsRequestHistoryNotFound(errors.New("other"))) - var notFound *RequestHistoryNotFoundError - require.ErrorAs(t, err, ¬Found) - assert.Equal(t, "request/queue/1", notFound.RequestID) } -func assertOperationFinishIncludesContextTag(t *testing.T, snapshot tally.Snapshot, success bool) { +func assertOperationFinishIncludesContextTag(t *testing.T, snapshot tally.Snapshot, operation string, success bool) { t.Helper() wantResult := "error" if success { wantResult = "success" } for _, histogram := range snapshot.Histograms() { - if histogram.Name() == "test.request_history_controller.get_by_id.finish" { + if histogram.Name() == "test.request_history_controller."+operation+".finish" { assert.Equal(t, "context-queue", histogram.Tags()["queue"]) assert.Equal(t, wantResult, histogram.Tags()["result"]) return From 93f522ba790b22a2622e1d6a6203da0aed57ae5d Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Thu, 3 Sep 2026 00:46:13 +0000 Subject: [PATCH 2/2] refactor(stovepipe): separate history URI read --- stovepipe/controller/request_history.go | 34 +++++++++++++++---------- 1 file changed, 21 insertions(+), 13 deletions(-) diff --git a/stovepipe/controller/request_history.go b/stovepipe/controller/request_history.go index c81db249..50537df8 100644 --- a/stovepipe/controller/request_history.go +++ b/stovepipe/controller/request_history.go @@ -92,38 +92,46 @@ func (c *requestHistoryController) GetRequestHistoryByURI(ctx context.Context, r op := metrics.Begin(c.metricsScope, "get_by_uri", metrics.StorageLatencyBuckets, metrics.TagsFromContext(ctx)...) defer func() { op.Complete(retErr) }() + history, retErr := c.readHistoryByURI(ctx, req) + if retErr != nil { + return nil, retErr + } + c.logger.Debugw("request history retrieved by URI", + "uri", req.URI, + "request_id", history.RequestID, + "queue", req.Queue, + "event_count", len(history.Events), + ) + return []entity.RequestHistory{history}, nil +} + +func (c *requestHistoryController) readHistoryByURI(ctx context.Context, req entity.GetRequestHistoryByURIRequest) (entity.RequestHistory, error) { if err := validateHistoryIdentifier("queue", req.Queue); err != nil { - return nil, fmt.Errorf("GetRequestHistoryByURI invalid queue: %w", err) + return entity.RequestHistory{}, fmt.Errorf("GetRequestHistoryByURI invalid queue: %w", err) } if err := validateHistoryIdentifier("URI", req.URI); err != nil { - return nil, fmt.Errorf("GetRequestHistoryByURI invalid request: %w", err) + return entity.RequestHistory{}, fmt.Errorf("GetRequestHistoryByURI invalid request: %w", err) } stores, err := c.stores.For(storage.Config{QueueName: req.Queue}) if err != nil { - return nil, fmt.Errorf("GetRequestHistoryByURI failed to resolve storage for queue %q: %w", req.Queue, err) + return entity.RequestHistory{}, fmt.Errorf("GetRequestHistoryByURI failed to resolve storage for queue %q: %w", req.Queue, err) } requestID, err := stores.GetRequestURIStore().GetIDByURI(ctx, req.URI) if err != nil { if storage.IsNotFound(err) { - return nil, errs.NewUserError(&RequestHistoryNotFoundError{URI: req.URI}) + return entity.RequestHistory{}, errs.NewUserError(&RequestHistoryNotFoundError{URI: req.URI}) } - return nil, fmt.Errorf("GetRequestHistoryByURI failed to resolve request URI %s: %w", req.URI, err) + return entity.RequestHistory{}, fmt.Errorf("GetRequestHistoryByURI failed to resolve request URI %s: %w", req.URI, err) } logs, err := loadRequestLogs(ctx, stores.GetRequestLogStore(), requestID, &RequestHistoryNotFoundError{URI: req.URI}) if err != nil { - return nil, fmt.Errorf("GetRequestHistoryByURI failed to list request logs uri=%s request_id=%s: %w", req.URI, requestID, err) + return entity.RequestHistory{}, fmt.Errorf("GetRequestHistoryByURI failed to list request logs uri=%s request_id=%s: %w", req.URI, requestID, err) } - c.logger.Debugw("request history retrieved by URI", - "uri", req.URI, - "request_id", requestID, - "queue", req.Queue, - "event_count", len(logs), - ) - return []entity.RequestHistory{{RequestID: requestID, Events: logs}}, nil + return entity.RequestHistory{RequestID: requestID, Events: logs}, nil } func loadRequestLogs(ctx context.Context, store storage.RequestLogStore, requestID string, notFound *RequestHistoryNotFoundError) ([]entity.RequestLog, error) {