[2/N] feat(gateway): persist request receipts on Land

submitqueue/core/request/BUILD.bazel

@@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library", "go_test")
 go_library(
     name = "go_default_library",
     srcs = [
+        "admission.go",
         "log.go",
         "request.go",
     ],
@@ -20,6 +21,7 @@ go_library(
 go_test(
     name = "go_default_test",
     srcs = [
+        "admission_test.go",
         "log_test.go",
         "request_test.go",
     ],

submitqueue/core/request/admission.go

@@ -0,0 +1,81 @@
+// 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 request
+
+import (
+	"context"
+	"fmt"
+	"maps"
+	"slices"
+
+	"github.com/uber/submitqueue/submitqueue/entity"
+	"github.com/uber/submitqueue/submitqueue/extension/storage"
+)
+
+// AdmissionWriter creates immutable request context and initial read-model projections.
+type AdmissionWriter struct {
+	store storage.Storage
+}
+
+// NewAdmissionWriter creates a request receipt projection writer.
+func NewAdmissionWriter(store storage.Storage) *AdmissionWriter {
+	return &AdmissionWriter{store: store}
+}
+
+// Create writes immutable request context and initial accepted projections.
+// Writes are independent and stop on the first error; successful earlier writes are not rolled back.
+func (m *AdmissionWriter) Create(ctx context.Context, summary entity.RequestSummary) error {
+	if err := m.store.GetRequestSummaryStore().Create(ctx, summary); err != nil {
+		return fmt.Errorf("failed to create request summary request_id=%s: %w", summary.RequestID, err)
+	}
+
+	for _, changeURI := range summary.ChangeURIs {
+		mapping := entity.RequestURI{
+			ChangeURI:    changeURI,
+			ReceivedAtMs: summary.ReceivedAtMs,
+			RequestID:    summary.RequestID,
+		}
+		if err := m.store.GetRequestURIStore().Create(ctx, mapping); err != nil {
+			return fmt.Errorf("failed to create request URI mapping request_id=%s change_uri=%s: %w", summary.RequestID, changeURI, err)
+		}
+	}
+
+	queueSummary := queueSummaryFromSummary(summary)
+	if err := m.store.GetRequestQueueSummaryStore().Create(ctx, queueSummary); err != nil {
+		return fmt.Errorf("failed to create queue summary request_id=%s: %w", summary.RequestID, err)
+	}
+
+	return nil
+}
+
+func queueSummaryFromSummary(summary entity.RequestSummary) entity.RequestQueueSummary {
+	return entity.RequestQueueSummary{
+		RequestID:    summary.RequestID,
+		Queue:        summary.Queue,
+		ChangeURIs:   slices.Clone(summary.ChangeURIs),
+		ReceivedAtMs: summary.ReceivedAtMs,
+		Status:       summary.Status,
+		Version:      summary.Version,
+		LastError:    summary.LastError,
+		Metadata:     cloneMetadata(summary.Metadata),
+	}
+}
+
+func cloneMetadata(metadata map[string]string) map[string]string {
+	if metadata == nil {
+		return map[string]string{}
+	}
+	return maps.Clone(metadata)
+}

submitqueue/core/request/admission_test.go

@@ -0,0 +1,109 @@
+// 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 request
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/uber/submitqueue/submitqueue/entity"
+	"github.com/uber/submitqueue/submitqueue/extension/storage"
+	storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock"
+	"go.uber.org/mock/gomock"
+)
+
+func TestAdmissionWriter_Create(t *testing.T) {
+	summary := testRequestSummary()
+	tests := []struct {
+		name      string
+		setup     func(*gomock.Controller, *storagemock.MockStorage)
+		wantError bool
+	}{
+		{
+			name: "creates all projections",
+			setup: func(ctrl *gomock.Controller, store *storagemock.MockStorage) {
+				summaryStore := storagemock.NewMockRequestSummaryStore(ctrl)
+				uriStore := storagemock.NewMockRequestURIStore(ctrl)
+				queueStore := storagemock.NewMockRequestQueueSummaryStore(ctrl)
+				store.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes()
+				store.EXPECT().GetRequestURIStore().Return(uriStore).AnyTimes()
+				store.EXPECT().GetRequestQueueSummaryStore().Return(queueStore).AnyTimes()
+				summaryStore.EXPECT().Create(gomock.Any(), summary).Return(nil)
+				uriStore.EXPECT().Create(gomock.Any(), entity.RequestURI{ChangeURI: "uri/1", ReceivedAtMs: 10, RequestID: "q/1"}).Return(nil)
+				uriStore.EXPECT().Create(gomock.Any(), entity.RequestURI{ChangeURI: "uri/2", ReceivedAtMs: 10, RequestID: "q/1"}).Return(nil)
+				queueStore.EXPECT().Create(gomock.Any(), queueSummaryFromSummary(summary)).Return(nil)
+			},
+		},
+		{
+			name: "summary failure stops remaining writes",
+			setup: func(ctrl *gomock.Controller, store *storagemock.MockStorage) {
+				summaryStore := storagemock.NewMockRequestSummaryStore(ctrl)
+				store.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes()
+				summaryStore.EXPECT().Create(gomock.Any(), summary).Return(storage.ErrAlreadyExists)
+			},
+			wantError: true,
+		},
+		{
+			name: "URI failure stops remaining writes",
+			setup: func(ctrl *gomock.Controller, store *storagemock.MockStorage) {
+				summaryStore := storagemock.NewMockRequestSummaryStore(ctrl)
+				uriStore := storagemock.NewMockRequestURIStore(ctrl)
+				store.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes()
+				store.EXPECT().GetRequestURIStore().Return(uriStore).AnyTimes()
+				summaryStore.EXPECT().Create(gomock.Any(), summary).Return(nil)
+				uriStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(fmt.Errorf("URI down"))
+			},
+			wantError: true,
+		},
+		{
+			name: "queue projection failure is returned",
+			setup: func(ctrl *gomock.Controller, store *storagemock.MockStorage) {
+				summaryStore := storagemock.NewMockRequestSummaryStore(ctrl)
+				uriStore := storagemock.NewMockRequestURIStore(ctrl)
+				queueStore := storagemock.NewMockRequestQueueSummaryStore(ctrl)
+				store.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes()
+				store.EXPECT().GetRequestURIStore().Return(uriStore).AnyTimes()
+				store.EXPECT().GetRequestQueueSummaryStore().Return(queueStore).AnyTimes()
+				summaryStore.EXPECT().Create(gomock.Any(), summary).Return(nil)
+				uriStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).Times(2)
+				queueStore.EXPECT().Create(gomock.Any(), queueSummaryFromSummary(summary)).Return(fmt.Errorf("queue projection down"))
+			},
+			wantError: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+			store := storagemock.NewMockStorage(ctrl)
+			tt.setup(ctrl, store)
+			err := NewAdmissionWriter(store).Create(context.Background(), summary)
+			if tt.wantError {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+			}
+		})
+	}
+}
+
+func testRequestSummary() entity.RequestSummary {
+	return entity.RequestSummary{
+		RequestID: "q/1", Queue: "q", ChangeURIs: []string{"uri/1", "uri/2"}, ReceivedAtMs: 10,
+		Status: entity.RequestStatusAccepted, StatusTimestampMs: 10, Version: 1, Metadata: map[string]string{},
+	}
+}

submitqueue/gateway/controller/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
         "cancel.go",
         "land.go",
         "ping.go",
+        "read_errors.go",
         "status.go",
     ],
     importpath = "github.com/uber/submitqueue/submitqueue/gateway/controller",
@@ -37,6 +38,7 @@ go_test(
         "land_test.go",
         "ping_test.go",
         "status_test.go",
+        "storage_test.go",
     ],
     embed = [":go_default_library"],
     deps = [

submitqueue/gateway/controller/land.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"time"
 
 	"github.com/uber-go/tally"
 	mergestrategypb "github.com/uber/submitqueue/api/base/mergestrategy/protopb"
@@ -29,20 +30,23 @@ import (
 	"github.com/uber/submitqueue/platform/errs"
 	"github.com/uber/submitqueue/platform/extension/counter"
 	"github.com/uber/submitqueue/platform/metrics"
+	requestcore "github.com/uber/submitqueue/submitqueue/core/request"
 	"github.com/uber/submitqueue/submitqueue/core/topickey"
 	"github.com/uber/submitqueue/submitqueue/entity"
 	"github.com/uber/submitqueue/submitqueue/extension/queueconfig"
 	"github.com/uber/submitqueue/submitqueue/extension/storage"
 	"go.uber.org/zap"
 )
 
+var errInvalidRequest = errors.New("invalid request")
+
 // 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"))
+var ErrInvalidRequest = errs.NewUserError(errInvalidRequest)
 
 // IsInvalidRequest returns true if any error in the error chain is ErrInvalidRequest.
 func IsInvalidRequest(err error) bool {
-	return errors.Is(err, ErrInvalidRequest)
+	return errors.Is(err, errInvalidRequest)
 }
 
 // UnrecognizedQueueError indicates the request named a queue that is not
@@ -65,25 +69,27 @@ func IsUnrecognizedQueue(err error) bool {
 
 // LandController handles land business logic for the gateway
 type LandController struct {
-	logger       *zap.SugaredLogger
-	metricsScope tally.Scope
-	counter      counter.Counter
-	store        storage.Storage
-	queueConfigs queueconfig.Store
-	registry     consumer.TopicRegistry
+	logger          *zap.SugaredLogger
+	metricsScope    tally.Scope
+	counter         counter.Counter
+	store           storage.Storage
+	admissionWriter *requestcore.AdmissionWriter
+	queueConfigs    queueconfig.Store
+	registry        consumer.TopicRegistry
 }
 
 // NewLandController creates a new instance of the gateway land controller.
 // The controller publishes land requests to the topic registered under
 // topickey.TopicKeyStart in the registry.
 func NewLandController(logger *zap.SugaredLogger, scope tally.Scope, counter counter.Counter, store storage.Storage, queueConfigs queueconfig.Store, registry consumer.TopicRegistry) *LandController {
 	return &LandController{
-		logger:       logger,
-		metricsScope: scope.SubScope("land_controller"),
-		counter:      counter,
-		store:        store,
-		queueConfigs: queueConfigs,
-		registry:     registry,
+		logger:          logger,
+		metricsScope:    scope.SubScope("land_controller"),
+		counter:         counter,
+		store:           store,
+		admissionWriter: requestcore.NewAdmissionWriter(store),
+		queueConfigs:    queueConfigs,
+		registry:        registry,
 	}
 }
 
@@ -94,13 +100,16 @@ func (c *LandController) Land(ctx context.Context, req *pb.LandRequest) (resp *p
 	op := metrics.Begin(c.metricsScope, opName)
 	defer func() { op.Complete(retErr) }()
 
-	// Validate required fields.
-	if req.Queue == "" {
-		return nil, fmt.Errorf("LandController requires the request to have a queue name specified: %w", ErrInvalidRequest)
+	// Validate provider-agnostic request constraints before allocating an sqid.
+	if err := validateStoredIdentifier("queue", req.Queue); err != nil {
+		return nil, fmt.Errorf("LandController invalid queue: %w", err)
 	}
-	if req.Change == nil || len(req.Change.Uris) == 0 {
+	if req.Change == nil {
 		return nil, fmt.Errorf("LandController requires the request to have at least one change URI specified: %w", ErrInvalidRequest)
 	}
+	if err := validateChangeURIs(req.Change.Uris); err != nil {
+		return nil, fmt.Errorf("LandController invalid change URIs: %w", err)
+	}
 
 	change := change.Change{
 		URIs: req.Change.GetUris(),
@@ -132,11 +141,33 @@ func (c *LandController) Land(ctx context.Context, req *pb.LandRequest) (resp *p
 		Change:       change,
 		LandStrategy: strategy,
 	}
+	if err := validateStoredIdentifier("generated sqid", landRequest.ID); err != nil {
+		return nil, fmt.Errorf("LandController generated invalid request ID for queue=%s: %w", queue, err)
+	}
+	receivedAtMs := time.Now().UnixMilli()
+	summary := entity.RequestSummary{
+		RequestID:         landRequest.ID,
+		Queue:             landRequest.Queue,
+		ChangeURIs:        append([]string{}, landRequest.Change.URIs...),
+		ReceivedAtMs:      receivedAtMs,
+		Status:            entity.RequestStatusAccepted,
+		StatusTimestampMs: receivedAtMs,
+		Version:           1,
+		Metadata:          map[string]string{},
+	}
+	if err := c.admissionWriter.Create(ctx, summary); err != nil {
+		return nil, fmt.Errorf("LandController failed to create request receipt sqid=%s: %w", landRequest.ID, err)
+	}
 
 	// Record the accepted status in the request log for reconciliation. Once the request materializes as a Request entity, the status might be updated to "new".
 	// It is important to record the status before publishing to the queue for processing. It is important to publish straight to the database and not via a entityqueue.
 	// Gateway has to stay consistent with the request log.
-	logEntry := entity.NewRequestLog(landRequest.ID, entity.RequestStatusAccepted, 0, "", nil)
+	logEntry := entity.RequestLog{
+		RequestID:   landRequest.ID,
+		TimestampMs: receivedAtMs,
+		Status:      entity.RequestStatusAccepted,
+		Metadata:    map[string]string{},
+	}
 	if err := c.store.GetRequestLogStore().Insert(ctx, logEntry); err != nil {
 		return nil, fmt.Errorf("LandController failed to insert request log for sqid=%s: %w", landRequest.ID, err)
 	}