feat(stovepipe): request history api - by request URI

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)
 }
 

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,65 @@ 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) }()
+
+	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 entity.RequestHistory{}, fmt.Errorf("GetRequestHistoryByURI invalid queue: %w", err)
+	}
+	if err := validateHistoryIdentifier("URI", req.URI); err != nil {
+		return entity.RequestHistory{}, fmt.Errorf("GetRequestHistoryByURI invalid request: %w", err)
+	}
+
+	stores, err := c.stores.For(storage.Config{QueueName: req.Queue})
+	if err != nil {
+		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 entity.RequestHistory{}, errs.NewUserError(&RequestHistoryNotFoundError{URI: req.URI})
+		}
+		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 entity.RequestHistory{}, fmt.Errorf("GetRequestHistoryByURI failed to list request logs uri=%s request_id=%s: %w", req.URI, requestID, err)
+	}
+
+	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
+}

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, &notFound)
+				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, &notFound)
+			assert.Equal(t, tt.want, *notFound)
+		})
+	}
 	assert.False(t, IsRequestHistoryNotFound(errors.New("other")))
-	var notFound *RequestHistoryNotFoundError
-	require.ErrorAs(t, err, &notFound)
-	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