feat(gateway): persist request receipts on Land

Create authoritative request summaries, exact change URI mappings, and queue receipt projections before publishing accepted Land requests.

Validation: make fmt && make build && make test && make e2e-test

Author: albertywu

submitqueue/core/request/BUILD.bazel

@@ -4,6 +4,7 @@ go_library(
     name = "go_default_library",
     srcs = [
         "log.go",
+        "receipt.go",
         "request.go",
     ],
     importpath = "github.com/uber/submitqueue/submitqueue/core/request",
@@ -21,6 +22,7 @@ go_test(
     name = "go_default_test",
     srcs = [
         "log_test.go",
+        "receipt_test.go",
         "request_test.go",
     ],
     embed = [":go_default_library"],

submitqueue/core/request/receipt.go

@@ -0,0 +1,41 @@
+// 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"
+
+	"github.com/uber/submitqueue/submitqueue/entity"
+	"github.com/uber/submitqueue/submitqueue/extension/storage"
+)
+
+// ReceiptWriter stores an internal request receipt before pipeline publication.
+type ReceiptWriter struct {
+	store storage.Storage
+}
+
+// NewReceiptWriter creates a request receipt writer.
+func NewReceiptWriter(store storage.Storage) *ReceiptWriter {
+	return &ReceiptWriter{store: store}
+}
+
+// Create writes the authoritative request receipt.
+func (w *ReceiptWriter) Create(ctx context.Context, summary entity.RequestSummary) error {
+	if err := w.store.GetRequestSummaryStore().Create(ctx, summary); err != nil {
+		return fmt.Errorf("failed to create request summary request_id=%s: %w", summary.RequestID, err)
+	}
+	return nil
+}

submitqueue/core/request/receipt_test.go

@@ -0,0 +1,74 @@
+// 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"
+	"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 TestReceiptWriter_Create(t *testing.T) {
+	summary := testRequestSummary()
+	tests := []struct {
+		name      string
+		setup     func(*gomock.Controller, *storagemock.MockStorage)
+		wantError bool
+	}{
+		{
+			name: "creates receipt",
+			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(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,
+		},
+	}
+
+	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 := NewReceiptWriter(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.RequestStatusAccepting, StatusTimestampMs: 10, Version: 1, Metadata: map[string]string{},
+	}
+}

submitqueue/entity/request_log.go

@@ -31,7 +31,11 @@ const (
 	// RequestStatusUnknown is the unknown sentinel status. It is set by default when the structure is initialized. It should never be seen in the system.
 	RequestStatusUnknown RequestStatus = ""
 
-	// RequestStatusAccepted indicates that the request has been accepted by the system. Typically a gateway service will set this status when the land request is received and persisted to the logging database.
+	// RequestStatusAccepting is the internal status of a persisted Land receipt that has not yet been published to the processing pipeline.
+	// Public read APIs must not expose requests that remain in this status.
+	RequestStatusAccepting RequestStatus = "accepting"
+
+	// RequestStatusAccepted indicates that the request has been published to the processing pipeline.
 	RequestStatusAccepted RequestStatus = "accepted"
 
 	// RequestStatusStarted is the initial status of a request. It corresponds to the RequestStateStarted state and typically set by the orchestrator service when the request is received and persisted to the operating database.

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",
@@ -34,6 +35,7 @@ go_test(
         "land_test.go",
         "ping_test.go",
         "status_test.go",
+        "storage_fixture_test.go",
     ],
     embed = [":go_default_library"],
     deps = [

submitqueue/gateway/controller/land.go

@@ -18,13 +18,15 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"time"
 
 	"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"
+	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"
@@ -61,25 +63,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
+	receiptWriter *requestcore.ReceiptWriter
+	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,
+		receiptWriter: requestcore.NewReceiptWriter(store),
+		queueConfigs:  queueConfigs,
+		registry:      registry,
 	}
 }
 
@@ -90,12 +94,12 @@ func (c *LandController) Land(ctx context.Context, req entity.LandRequest) (resu
 	op := metrics.Begin(c.metricsScope, opName)
 	defer func() { op.Complete(retErr) }()
 
-	// Validate required fields.
-	if req.Queue == "" {
-		return entity.LandResult{}, 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 := validateQueueIdentifier(req.Queue); err != nil {
+		return entity.LandResult{}, fmt.Errorf("LandController invalid queue: %w", err)
 	}
-	if len(req.Change.URIs) == 0 {
-		return entity.LandResult{}, 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 entity.LandResult{}, fmt.Errorf("LandController invalid change URIs: %w", err)
 	}
 
 	queue := req.Queue
@@ -113,13 +117,46 @@ func (c *LandController) Land(ctx context.Context, req entity.LandRequest) (resu
 		return entity.LandResult{}, fmt.Errorf("LandController failed to generate request ID for queue=%s: %w", queue, err)
 	}
 	req.ID = fmt.Sprintf("%s/%d", queue, seq)
+	if err := validateStoredIdentifier("generated sqid", req.ID); err != nil {
+		return entity.LandResult{}, fmt.Errorf("LandController generated invalid request ID for queue=%s: %w", queue, err)
+	}
+
+	receivedAtMs := time.Now().UnixMilli()
+	summary := entity.RequestSummary{
+		RequestID:         req.ID,
+		Queue:             req.Queue,
+		ChangeURIs:        append([]string{}, req.Change.URIs...),
+		ReceivedAtMs:      receivedAtMs,
+		Status:            entity.RequestStatusAccepting,
+		StatusTimestampMs: receivedAtMs,
+		Version:           1,
+		Metadata:          map[string]string{},
+	}
+	if err := c.receiptWriter.Create(ctx, summary); err != nil {
+		return entity.LandResult{}, fmt.Errorf("LandController failed to create request receipt sqid=%s: %w", req.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(req.ID, entity.RequestStatusAccepted, 0, "", nil)
+	// Publish before exposing the request as accepted. A failed publish leaves an
+	// internal accepting receipt that public read APIs do not expose.
+	if err := c.publishToQueue(ctx, req); err != nil {
+		return entity.LandResult{}, fmt.Errorf("LandController failed to publish request to queue: %w", err)
+	}
+
+	logEntry := entity.RequestLog{
+		RequestID:   req.ID,
+		TimestampMs: receivedAtMs,
+		Status:      entity.RequestStatusAccepted,
+		Metadata:    map[string]string{},
+	}
 	if err := c.store.GetRequestLogStore().Insert(ctx, logEntry); err != nil {
-		return entity.LandResult{}, fmt.Errorf("LandController failed to insert request log for sqid=%s: %w", req.ID, err)
+		// Publication is the Land success boundary. Returning an error here would
+		// encourage the caller to submit a duplicate request that is already queued.
+		c.logger.Errorw("failed to record accepted status after publishing request",
+			"queue", req.Queue,
+			"sqid", req.ID,
+			"error", err,
+		)
+		metrics.NamedCounter(c.metricsScope, opName, "accepted_log_failure", 1)
 	}
 
 	c.logger.Debugw("land request created",
@@ -130,11 +167,6 @@ func (c *LandController) Land(ctx context.Context, req entity.LandRequest) (resu
 		"strategy", string(req.LandStrategy),
 	)
 
-	// Publish to queue for async processing
-	if err := c.publishToQueue(ctx, req); err != nil {
-		return entity.LandResult{}, fmt.Errorf("LandController failed to publish request to queue: %w", err)
-	}
-
 	c.logger.Infow("request published to queue",
 		"queue", req.Queue,
 		"sqid", req.ID,