feat(gateway): persist request context 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,122 @@
+// 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"
+	"errors"
+	"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.
+// Storage implementations remain mechanical; this type decides whether duplicate creates are identical retries or conflicts.
+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.
+// A duplicate for the same request ID is accepted only when its immutable context matches exactly.
+func (m *AdmissionWriter) Create(ctx context.Context, summary entity.RequestSummary) error {
+	if err := m.createRequestSummary(ctx, summary); err != nil {
+		return 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 && !errors.Is(err, storage.ErrAlreadyExists) {
+			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 {
+		if !errors.Is(err, storage.ErrAlreadyExists) {
+			return fmt.Errorf("failed to create queue summary request_id=%s: %w", summary.RequestID, err)
+		}
+		existing, getErr := m.store.GetRequestQueueSummaryStore().Get(ctx, summary.Queue, summary.ReceivedAtMs, summary.RequestID)
+		if getErr != nil {
+			return fmt.Errorf("failed to get duplicate queue summary request_id=%s: %w", summary.RequestID, getErr)
+		}
+		if !sameQueueSummaryIdentity(existing, queueSummary) {
+			return fmt.Errorf("conflicting queue summary request_id=%s: %w", summary.RequestID, storage.ErrAlreadyExists)
+		}
+	}
+
+	return nil
+}
+
+func (m *AdmissionWriter) createRequestSummary(ctx context.Context, summary entity.RequestSummary) error {
+	if err := m.store.GetRequestSummaryStore().Create(ctx, summary); err != nil {
+		if !errors.Is(err, storage.ErrAlreadyExists) {
+			return fmt.Errorf("failed to create request summary request_id=%s: %w", summary.RequestID, err)
+		}
+		existing, getErr := m.store.GetRequestSummaryStore().Get(ctx, summary.RequestID)
+		if getErr != nil {
+			return fmt.Errorf("failed to get duplicate request summary request_id=%s: %w", summary.RequestID, getErr)
+		}
+		if !sameRequestSummaryIdentity(existing, summary) {
+			return fmt.Errorf("conflicting request summary request_id=%s: %w", summary.RequestID, storage.ErrAlreadyExists)
+		}
+	}
+	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 sameRequestSummaryIdentity(left, right entity.RequestSummary) bool {
+	return left.RequestID == right.RequestID &&
+		left.Queue == right.Queue &&
+		left.ReceivedAtMs == right.ReceivedAtMs &&
+		slices.Equal(left.ChangeURIs, right.ChangeURIs)
+}
+
+func sameQueueSummaryIdentity(left, right entity.RequestQueueSummary) bool {
+	return left.RequestID == right.RequestID &&
+		left.Queue == right.Queue &&
+		left.ReceivedAtMs == right.ReceivedAtMs &&
+		slices.Equal(left.ChangeURIs, right.ChangeURIs)
+}
+
+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,114 @@
+// 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)
+		wantErr error
+	}{
+		{
+			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: "identical retry succeeds",
+			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(storage.ErrAlreadyExists)
+				summaryStore.EXPECT().Get(gomock.Any(), "q/1").Return(summary, nil)
+				uriStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(storage.ErrAlreadyExists).Times(2)
+				queueStore.EXPECT().Create(gomock.Any(), queueSummaryFromSummary(summary)).Return(storage.ErrAlreadyExists)
+				queueStore.EXPECT().Get(gomock.Any(), "q", int64(10), "q/1").Return(queueSummaryFromSummary(summary), nil)
+			},
+		},
+		{
+			name: "conflicting summary retry fails",
+			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)
+				conflict := summary
+				conflict.Queue = "other"
+				summaryStore.EXPECT().Get(gomock.Any(), "q/1").Return(conflict, nil)
+			},
+			wantErr: storage.ErrAlreadyExists,
+		},
+		{
+			name: "URI write failure stops queue projection",
+			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"))
+			},
+		},
+	}
+
+	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.wantErr != nil {
+				require.ErrorIs(t, err, tt.wantErr)
+			} else if tt.name == "URI write failure stops queue projection" {
+				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/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,6 +30,7 @@ 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"
@@ -65,25 +67,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,
 	}
 }
 
@@ -132,13 +136,32 @@ func (c *LandController) Land(ctx context.Context, req *pb.LandRequest) (resp *p
 		Change:       change,
 		LandStrategy: strategy,
 	}
+	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)
+		return nil, fmt.Errorf("LandController failed to insert accepted request log for sqid=%s: %w", landRequest.ID, err)
 	}
 
 	c.logger.Debugw("land request created",