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/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,6 +18,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"time"
 
 	"github.com/uber-go/tally"
 	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
@@ -90,12 +91,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 +114,48 @@ 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.store.GetRequestSummaryStore().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. Later pipeline events repair
+		// the accepting projection even if this accepted log is not persisted. If
+		// the client retries after losing the response, the orchestrator rejects
+		// the new sqid as a duplicate while the original request continues.
+		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 +166,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,

submitqueue/gateway/controller/land_test.go

@@ -17,6 +17,7 @@ package controller
 import (
 	"context"
 	"fmt"
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -64,14 +65,9 @@ func newTestRegistryWithNoopPublisher(t *testing.T, ctrl *gomock.Controller) con
 	return registry
 }
 
-// noopStorage returns a storage.Storage whose RequestLogStore.Insert
-// succeeds silently for any entityqueue.
+// noopStorage returns stateful request storage whose writes succeed.
 func noopStorage(ctrl *gomock.Controller) storage.Storage {
-	logStore := storagemock.NewMockRequestLogStore(ctrl)
-	logStore.EXPECT().Insert(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
-	store := storagemock.NewMockStorage(ctrl)
-	store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes()
-	return store
+	return newControllerStorageFixture(ctrl).storage
 }
 
 // noopQueueConfigStore returns a mock queueconfig.Store that always reports
@@ -162,6 +158,48 @@ func TestLand_ReturnsErrorOnEmptyQueue(t *testing.T) {
 	assert.True(t, IsInvalidRequest(err))
 }
 
+func TestLand_ValidatesQueueLengthBeforeAllocatingSqid(t *testing.T) {
+	tests := []struct {
+		name      string
+		queue     string
+		wantError bool
+	}{
+		{name: "maximum length", queue: strings.Repeat("q", maxQueueIdentifierBytes)},
+		{name: "over maximum", queue: strings.Repeat("q", maxQueueIdentifierBytes+1), wantError: true},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+			cnt := countermock.NewMockCounter(ctrl)
+			if !tt.wantError {
+				cnt.EXPECT().Next(gomock.Any(), gomock.Any()).Return(int64(1), nil)
+			}
+			controller := NewLandController(
+				zap.NewNop().Sugar(),
+				tally.NoopScope,
+				cnt,
+				noopStorage(ctrl),
+				noopQueueConfigStore(ctrl),
+				newTestRegistryWithNoopPublisher(t, ctrl),
+			)
+
+			result, err := controller.Land(context.Background(), entity.LandRequest{
+				Queue:  tt.queue,
+				Change: change.Change{URIs: []string{"uri"}},
+			})
+
+			if tt.wantError {
+				require.Error(t, err)
+				assert.True(t, IsInvalidRequest(err))
+				return
+			}
+			require.NoError(t, err)
+			assert.Equal(t, tt.queue+"/1", result.ID)
+		})
+	}
+}
+
 func TestLand_ReturnsErrorOnEmptyChangeUri(t *testing.T) {
 	ctrl := gomock.NewController(t)
 
@@ -179,6 +217,39 @@ func TestLand_ReturnsErrorOnEmptyChangeUri(t *testing.T) {
 	assert.True(t, IsInvalidRequest(err))
 }
 
+func TestLand_ReturnsErrorOnInvalidChangeURIs(t *testing.T) {
+	tests := []struct {
+		name string
+		uris []string
+	}{
+		{name: "empty URI element", uris: []string{""}},
+		{name: "URI exceeds storage limit", uris: []string{strings.Repeat("x", maxStorageIdentifierBytes+1)}},
+		{name: "duplicate exact URI", uris: []string{"uri", "uri"}},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+			controller := NewLandController(
+				zap.NewNop().Sugar(),
+				tally.NoopScope,
+				countermock.NewMockCounter(ctrl),
+				noopStorage(ctrl),
+				noopQueueConfigStore(ctrl),
+				newTestRegistryWithNoopPublisher(t, ctrl),
+			)
+
+			_, err := controller.Land(context.Background(), entity.LandRequest{
+				Queue:  "test-queue",
+				Change: change.Change{URIs: tt.uris},
+			})
+
+			require.Error(t, err)
+			assert.True(t, IsInvalidRequest(err))
+		})
+	}
+}
+
 func TestLand_ReturnsErrorOnZeroValueChange(t *testing.T) {
 	ctrl := gomock.NewController(t)
 
@@ -238,22 +309,44 @@ func TestLand_PropagatesQueueConfigStoreError(t *testing.T) {
 func TestLand_PublishesToQueue(t *testing.T) {
 	var publishedTopic string
 	var publishedMessage entityqueue.Message
+	var persistedSummary entity.RequestSummary
+	var persistedLog entity.RequestLog
 
 	ctrl := gomock.NewController(t)
 
 	cnt := countermock.NewMockCounter(ctrl)
 	cnt.EXPECT().Next(gomock.Any(), gomock.Any()).Return(int64(123), nil)
 
+	store := storagemock.NewMockStorage(ctrl)
+	summaryStore := storagemock.NewMockRequestSummaryStore(ctrl)
+	logStore := storagemock.NewMockRequestLogStore(ctrl)
+	store.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes()
+	store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes()
+
 	registry, publisher := newTestRegistry(t, ctrl)
-	publisher.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
-		func(ctx context.Context, topic string, msg entityqueue.Message) error {
-			publishedTopic = topic
-			publishedMessage = msg
-			return nil
-		},
+	gomock.InOrder(
+		summaryStore.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(
+			func(_ context.Context, summary entity.RequestSummary) error {
+				persistedSummary = summary
+				return nil
+			},
+		),
+		publisher.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
+			func(_ context.Context, topic string, msg entityqueue.Message) error {
+				publishedTopic = topic
+				publishedMessage = msg
+				return nil
+			},
+		),
+		logStore.EXPECT().Insert(gomock.Any(), gomock.Any()).DoAndReturn(
+			func(_ context.Context, log entity.RequestLog) error {
+				persistedLog = log
+				return nil
+			},
+		),
 	)
 
-	controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, cnt, noopStorage(ctrl), noopQueueConfigStore(ctrl), registry)
+	controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, cnt, store, noopQueueConfigStore(ctrl), registry)
 	ctx := context.Background()
 
 	req := entity.LandRequest{
@@ -266,6 +359,24 @@ func TestLand_PublishesToQueue(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, "test-queue/123", result.ID)
 
+	assert.Equal(t, entity.RequestSummary{
+		RequestID:         "test-queue/123",
+		Queue:             "test-queue",
+		ChangeURIs:        []string{"github://github.example.com/uber/backend/pull/456/fedcba9876543210fedcba9876543210fedcba98"},
+		ReceivedAtMs:      persistedSummary.ReceivedAtMs,
+		Status:            entity.RequestStatusAccepting,
+		StatusTimestampMs: persistedSummary.ReceivedAtMs,
+		Version:           1,
+		Metadata:          map[string]string{},
+	}, persistedSummary)
+	assert.Positive(t, persistedSummary.ReceivedAtMs)
+	assert.Equal(t, entity.RequestLog{
+		RequestID:   "test-queue/123",
+		TimestampMs: persistedSummary.ReceivedAtMs,
+		Status:      entity.RequestStatusAccepted,
+		Metadata:    map[string]string{},
+	}, persistedLog)
+
 	// Verify message was published to the topic registered under TopicKeyStart
 	assert.Equal(t, "start", publishedTopic)
 	assert.Equal(t, "test-queue/123", publishedMessage.ID)
@@ -280,20 +391,48 @@ func TestLand_PublishesToQueue(t *testing.T) {
 	assert.Equal(t, mergestrategy.MergeStrategyRebase, deserializedReq.LandStrategy)
 }
 
-func TestLand_ContinuesWhenPublishFails(t *testing.T) {
+func TestLand_ReturnsErrorWhenPublishFails(t *testing.T) {
 	ctrl := gomock.NewController(t)
 
 	cnt := countermock.NewMockCounter(ctrl)
 	cnt.EXPECT().Next(gomock.Any(), gomock.Any()).Return(int64(999), nil)
 
+	store := storagemock.NewMockStorage(ctrl)
+	summaryStore := storagemock.NewMockRequestSummaryStore(ctrl)
+	store.EXPECT().GetRequestSummaryStore().Return(summaryStore)
+	summaryStore.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, summary entity.RequestSummary) error {
+		assert.Equal(t, entity.RequestStatusAccepting, summary.Status)
+		return nil
+	})
+
 	registry, publisher := newTestRegistry(t, ctrl)
 	publisher.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).Return(fmt.Errorf("queue unavailable"))
 
-	controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, cnt, noopStorage(ctrl), noopQueueConfigStore(ctrl), registry)
+	controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, cnt, store, noopQueueConfigStore(ctrl), registry)
 	ctx := context.Background()
 
 	_, err := controller.Land(ctx, testLandRequest("test-queue"))
 
-	// Should fail if publish fails
 	require.Error(t, err)
 }
+
+func TestLand_ReturnsSqidWhenAcceptedLogFailsAfterPublish(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	cnt := countermock.NewMockCounter(ctrl)
+	cnt.EXPECT().Next(gomock.Any(), gomock.Any()).Return(int64(999), nil)
+	fixture := newControllerStorageFixture(ctrl)
+	fixture.setLogInsertError(fmt.Errorf("log unavailable"))
+
+	controller := NewLandController(
+		zap.NewNop().Sugar(),
+		tally.NoopScope,
+		cnt,
+		fixture.storage,
+		noopQueueConfigStore(ctrl),
+		newTestRegistryWithNoopPublisher(t, ctrl),
+	)
+	result, err := controller.Land(context.Background(), testLandRequest("test-queue"))
+
+	require.NoError(t, err)
+	assert.Equal(t, "test-queue/999", result.ID)
+}