From 8f53ed161262d012bd6b68e09f2a3e6e19e37390 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Wed, 2 Sep 2026 23:20:30 +0000 Subject: [PATCH 1/2] feat(stovepipe): read request history by ID MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Summary: Intent: - Serve retained history by request ID without consulting operational request data. - Preserve domain-level invalid and missing-history classifications for future transports. Changes: - Add the request-history controller and bounded selector validation. - Read only the queue-scoped request log and preserve context metric tags. - Cover successful, invalid, missing, and infrastructure outcomes. This PR builds on #669, which defines the retained-history read model. --- 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/BUILD.bazel | 6 + stovepipe/controller/ingest.go | 10 -- stovepipe/controller/read_errors.go | 58 +++++++ stovepipe/controller/request_history.go | 82 ++++++++++ stovepipe/controller/request_history_test.go | 157 +++++++++++++++++++ 5 files changed, 303 insertions(+), 10 deletions(-) create mode 100644 stovepipe/controller/read_errors.go create mode 100644 stovepipe/controller/request_history.go create mode 100644 stovepipe/controller/request_history_test.go diff --git a/stovepipe/controller/BUILD.bazel b/stovepipe/controller/BUILD.bazel index d56dcd40..f01593dd 100644 --- a/stovepipe/controller/BUILD.bazel +++ b/stovepipe/controller/BUILD.bazel @@ -5,6 +5,8 @@ go_library( srcs = [ "ingest.go", "ping.go", + "read_errors.go", + "request_history.go", ], importpath = "github.com/uber/submitqueue/stovepipe/controller", visibility = ["//visibility:public"], @@ -31,15 +33,18 @@ go_test( srcs = [ "ingest_test.go", "ping_test.go", + "request_history_test.go", ], embed = [":go_default_library"], deps = [ "//api/stovepipe/protopb:go_default_library", "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", + "//platform/errs:go_default_library", "//platform/extension/counter:go_default_library", "//platform/extension/counter/mock:go_default_library", "//platform/extension/messagequeue/mock:go_default_library", + "//platform/metrics:go_default_library", "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/core/requestlog:go_default_library", "//stovepipe/core/requestlog/mock:go_default_library", @@ -53,5 +58,6 @@ go_test( "@com_github_uber_go_tally//:go_default_library", "@org_uber_go_mock//gomock:go_default_library", "@org_uber_go_zap//:go_default_library", + "@org_uber_go_zap//zaptest/observer:go_default_library", ], ) diff --git a/stovepipe/controller/ingest.go b/stovepipe/controller/ingest.go index 1c9022f0..8dc492ac 100644 --- a/stovepipe/controller/ingest.go +++ b/stovepipe/controller/ingest.go @@ -22,7 +22,6 @@ import ( "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" - "github.com/uber/submitqueue/platform/errs" "github.com/uber/submitqueue/platform/extension/counter" "github.com/uber/submitqueue/platform/metrics" "github.com/uber/submitqueue/platform/publish" @@ -34,20 +33,11 @@ import ( "go.uber.org/zap" ) -// ErrInvalidRequest is returned when the request fails validation. -// This error should be mapped to codes.InvalidArgument at the gRPC layer. -var ErrInvalidRequest = errs.NewUserError(errors.New("invalid request")) - // counterDomainRequest names the per-queue sequence that mints request IDs. It also // happens to be the leading segment of the ID, but the two are written independently // (see resolveID) so they cannot drift into each other. const counterDomainRequest = "request" -// IsInvalidRequest returns true if any error in the error chain is ErrInvalidRequest. -func IsInvalidRequest(err error) bool { - return errors.Is(err, ErrInvalidRequest) -} - // IngestController handles ingest business logic for stovepipe: it admits a queue's newly // observed commit into the validation pipeline. // diff --git a/stovepipe/controller/read_errors.go b/stovepipe/controller/read_errors.go new file mode 100644 index 00000000..1bbeff6b --- /dev/null +++ b/stovepipe/controller/read_errors.go @@ -0,0 +1,58 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package controller + +import ( + "errors" + "fmt" + + "github.com/uber/submitqueue/platform/errs" +) + +const maxHistoryIdentifierBytes = 255 + +// ErrInvalidRequest is returned when a request fails validation. +var ErrInvalidRequest = errs.NewUserError(errors.New("invalid request")) + +// IsInvalidRequest reports whether err contains an invalid request classification. +func IsInvalidRequest(err error) bool { + return errors.Is(err, ErrInvalidRequest) +} + +func validateHistoryIdentifier(name, value string) error { + if value == "" { + return fmt.Errorf("%s must be non-empty: %w", name, ErrInvalidRequest) + } + if len(value) > maxHistoryIdentifierBytes { + return fmt.Errorf("%s exceeds %d bytes: %w", name, maxHistoryIdentifierBytes, ErrInvalidRequest) + } + return nil +} + +// RequestHistoryNotFoundError indicates that no retained history exists for a selector. +type RequestHistoryNotFoundError struct { + RequestID string +} + +// Error implements error. +func (e *RequestHistoryNotFoundError) Error() string { + return fmt.Sprintf("request history not found for request ID %q", e.RequestID) +} + +// IsRequestHistoryNotFound reports whether err contains a retained-history absence. +func IsRequestHistoryNotFound(err error) bool { + var target *RequestHistoryNotFoundError + return errors.As(err, &target) +} diff --git a/stovepipe/controller/request_history.go b/stovepipe/controller/request_history.go new file mode 100644 index 00000000..64166aaf --- /dev/null +++ b/stovepipe/controller/request_history.go @@ -0,0 +1,82 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package controller + +import ( + "context" + "fmt" + + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/errs" + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" + "go.uber.org/zap" +) + +// RequestHistoryController handles retained request-history lookups. +type RequestHistoryController interface { + GetRequestHistoryByID(ctx context.Context, req entity.GetRequestHistoryByIDRequest) ([]entity.RequestLog, error) +} + +var _ RequestHistoryController = (*requestHistoryController)(nil) + +type requestHistoryController struct { + logger *zap.SugaredLogger + metricsScope tally.Scope + stores storage.Factory +} + +// NewRequestHistoryController creates a request-history controller. +func NewRequestHistoryController(logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory) RequestHistoryController { + return &requestHistoryController{ + logger: logger, + metricsScope: scope.SubScope("request_history_controller"), + stores: stores, + } +} + +// GetRequestHistoryByID returns every retained log event for one request ID. +func (c *requestHistoryController) GetRequestHistoryByID(ctx context.Context, req entity.GetRequestHistoryByIDRequest) (logs []entity.RequestLog, retErr error) { + op := metrics.Begin(c.metricsScope, "get_by_id", metrics.StorageLatencyBuckets, metrics.TagsFromContext(ctx)...) + defer func() { op.Complete(retErr) }() + + if err := validateHistoryIdentifier("queue", req.Queue); err != nil { + return nil, fmt.Errorf("GetRequestHistoryByID invalid queue: %w", err) + } + if err := validateHistoryIdentifier("request ID", req.ID); err != nil { + return nil, fmt.Errorf("GetRequestHistoryByID invalid request: %w", err) + } + + stores, err := c.stores.For(storage.Config{QueueName: req.Queue}) + if err != nil { + return nil, fmt.Errorf("GetRequestHistoryByID failed to resolve storage for queue %q: %w", req.Queue, err) + } + + logs, err = stores.GetRequestLogStore().List(ctx, 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) + } + + c.logger.Debugw("request history retrieved", + "request_id", req.ID, + "queue", req.Queue, + "event_count", len(logs), + ) + return logs, nil +} diff --git a/stovepipe/controller/request_history_test.go b/stovepipe/controller/request_history_test.go new file mode 100644 index 00000000..bd57e66b --- /dev/null +++ b/stovepipe/controller/request_history_test.go @@ -0,0 +1,157 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package controller + +import ( + "context" + "errors" + "fmt" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/errs" + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" + storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock" + "go.uber.org/mock/gomock" + "go.uber.org/zap" + "go.uber.org/zap/zaptest/observer" +) + +func TestGetRequestHistoryByID(t *testing.T) { + const ( + queue = "monorepo/main" + requestID = "request/monorepo/main/42" + ) + backendErr := errors.New("backend unavailable") + ordered := []entity.RequestLog{ + {ID: "state/1", RequestID: requestID, TimestampMs: 10, State: entity.RequestStateAccepted}, + {ID: "state/2", RequestID: requestID, TimestampMs: 20, State: entity.RequestStateProcessing}, + } + equalTimestamp := []entity.RequestLog{ + {ID: "event/a", RequestID: requestID, TimestampMs: 20, Event: entity.RequestEventBuildTriggered}, + {ID: "event/b", RequestID: requestID, TimestampMs: 20, Event: entity.RequestEventBuildFinished}, + } + duplicate := entity.RequestLog{ID: "event/a", RequestID: requestID, TimestampMs: 20, Event: entity.RequestEventBuildTriggered} + + tests := []struct { + name string + req entity.GetRequestHistoryByIDRequest + logs []entity.RequestLog + factoryErr error + listErr error + wantLogs []entity.RequestLog + wantInvalid bool + wantNotFound bool + wantUser bool + wantCause error + wantLog bool + }{ + {name: "ordered passthrough", req: entity.GetRequestHistoryByIDRequest{Queue: queue, ID: requestID}, logs: ordered, wantLogs: ordered, wantLog: true}, + {name: "equal timestamp order preserved", req: entity.GetRequestHistoryByIDRequest{Queue: queue, ID: requestID}, logs: equalTimestamp, wantLogs: equalTimestamp, wantLog: true}, + {name: "duplicates preserved", req: entity.GetRequestHistoryByIDRequest{Queue: queue, ID: requestID}, logs: []entity.RequestLog{duplicate, duplicate}, wantLogs: []entity.RequestLog{duplicate, duplicate}, wantLog: true}, + {name: "empty queue", req: entity.GetRequestHistoryByIDRequest{ID: requestID}, wantInvalid: true, wantUser: true}, + {name: "oversized queue", req: entity.GetRequestHistoryByIDRequest{Queue: strings.Repeat("q", maxHistoryIdentifierBytes+1), ID: requestID}, wantInvalid: true, wantUser: true}, + {name: "empty request ID", req: entity.GetRequestHistoryByIDRequest{Queue: queue}, wantInvalid: true, wantUser: true}, + {name: "oversized request ID", req: entity.GetRequestHistoryByIDRequest{Queue: queue, ID: strings.Repeat("r", maxHistoryIdentifierBytes+1)}, wantInvalid: true, wantUser: true}, + {name: "storage factory failure", req: entity.GetRequestHistoryByIDRequest{Queue: queue, ID: requestID}, factoryErr: backendErr, wantCause: backendErr}, + {name: "history not found", req: entity.GetRequestHistoryByIDRequest{Queue: queue, ID: requestID}, listErr: fmt.Errorf("query: %w", storage.ErrNotFound), wantNotFound: true, wantUser: true}, + {name: "log store failure", req: entity.GetRequestHistoryByIDRequest{Queue: queue, ID: 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) + 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().GetRequestLogStore().Return(logStore) + logStore.EXPECT().List(gomock.Any(), tt.req.ID).Return(tt.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.GetRequestHistoryByID(ctx, tt.req) + + assert.Equal(t, tt.wantLogs, got) + if tt.wantInvalid { + assert.True(t, IsInvalidRequest(err)) + } + assert.Equal(t, tt.wantNotFound, IsRequestHistoryNotFound(err)) + assert.Equal(t, tt.wantUser, errs.IsUserError(err)) + if tt.wantCause != nil { + assert.ErrorIs(t, err, tt.wantCause) + } + if tt.wantLogs != nil { + require.NoError(t, err) + } else { + require.Error(t, err) + } + + entries := observed.FilterMessage("request history retrieved").All() + if tt.wantLog { + require.Len(t, entries, 1) + assert.Equal(t, requestID, entries[0].ContextMap()["request_id"]) + assert.Equal(t, queue, entries[0].ContextMap()["queue"]) + assert.Equal(t, int64(len(tt.wantLogs)), entries[0].ContextMap()["event_count"]) + } else { + assert.Empty(t, entries) + } + + snapshot := scope.Snapshot() + 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) + }) + } +} + +func TestRequestHistoryNotFoundError(t *testing.T) { + err := fmt.Errorf("lookup failed: %w", &RequestHistoryNotFoundError{RequestID: "request/queue/1"}) + + assert.True(t, IsRequestHistoryNotFound(err)) + 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) { + t.Helper() + wantResult := "error" + if success { + wantResult = "success" + } + for _, histogram := range snapshot.Histograms() { + if histogram.Name() == "test.request_history_controller.get_by_id.finish" { + assert.Equal(t, "context-queue", histogram.Tags()["queue"]) + assert.Equal(t, wantResult, histogram.Tags()["result"]) + return + } + } + require.Fail(t, "operation finish histogram not found") +} From 57591bf0ca75e0bbc443853af00ed79f8dc85764 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Thu, 3 Sep 2026 00:44:13 +0000 Subject: [PATCH 2/2] refactor(stovepipe): separate history ID read --- stovepipe/controller/request_history.go | 20 ++++++++++++++------ 1 file changed, 14 insertions(+), 6 deletions(-) diff --git a/stovepipe/controller/request_history.go b/stovepipe/controller/request_history.go index 64166aaf..70f49cf0 100644 --- a/stovepipe/controller/request_history.go +++ b/stovepipe/controller/request_history.go @@ -53,6 +53,19 @@ func (c *requestHistoryController) GetRequestHistoryByID(ctx context.Context, re op := metrics.Begin(c.metricsScope, "get_by_id", metrics.StorageLatencyBuckets, metrics.TagsFromContext(ctx)...) defer func() { op.Complete(retErr) }() + logs, retErr = c.readHistoryByID(ctx, req) + if retErr != nil { + return nil, retErr + } + c.logger.Debugw("request history retrieved", + "request_id", req.ID, + "queue", req.Queue, + "event_count", len(logs), + ) + return logs, nil +} + +func (c *requestHistoryController) readHistoryByID(ctx context.Context, req entity.GetRequestHistoryByIDRequest) ([]entity.RequestLog, error) { if err := validateHistoryIdentifier("queue", req.Queue); err != nil { return nil, fmt.Errorf("GetRequestHistoryByID invalid queue: %w", err) } @@ -65,7 +78,7 @@ func (c *requestHistoryController) GetRequestHistoryByID(ctx context.Context, re 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 := stores.GetRequestLogStore().List(ctx, req.ID) if err != nil { if storage.IsNotFound(err) { return nil, errs.NewUserError(&RequestHistoryNotFoundError{RequestID: req.ID}) @@ -73,10 +86,5 @@ func (c *requestHistoryController) GetRequestHistoryByID(ctx context.Context, re return nil, fmt.Errorf("GetRequestHistoryByID failed to list request logs request_id=%s: %w", req.ID, err) } - c.logger.Debugw("request history retrieved", - "request_id", req.ID, - "queue", req.Queue, - "event_count", len(logs), - ) return logs, nil }