fix(stovepipe): initialize queue state within ingest

doc/rfc/stovepipe/queue-state.md

@@ -50,7 +50,7 @@ Suppose commits arrive in order A through G, with enablement changes at A, D and
 
 C and F need no validation requests. Their ancestry relative to the boundaries determines the answer.
 
-Registration starts a queue as disabled version 1, with no boundary; this covers commits before its first enablement. Missing state is an error, not disabled. If changes share a boundary, the highest version wins. Versions order changes; timestamps do not.
+The first `Ingest` call initializes a configured queue as disabled version 1, with no boundary, before applying an optional enablement decision. Initialization also adopts existing queue rows without enablement and preserves their pipeline fields. No separate startup registration is required. The initial disabled state covers commits before its first enablement. Missing state is an error, not disabled. If changes share a boundary, the highest version wins. Versions order changes; timestamps do not.
 
 `Ingest` accepts optional `ENABLED` or `DISABLED` enablement and returns the applied or observed status with the request ID. Omission or `UNKNOWN` preserves state; repeating the current setting preserves its boundary and version. A change records an inclusive boundary at the commit being ingested, without backfilling earlier requests. Enablement does not pause ingestion or processing. Deploy server support before callers send decisions, since older servers ignore the field.
 

service/stovepipe/README.md

@@ -110,4 +110,4 @@ The server handles `SIGINT` / `SIGTERM` gracefully: it drains in-flight RPCs, th
 
 ## Queue status
 
-Queue enablement is stored separately from validation results. Startup registers configured queues as disabled and preserves any registered state. `GetQueueStatus` returns the current summary, or the state covering a supplied commit through the existing source-control integration. History and summary writes can be retried after an interruption, so the summary may briefly lag internal state. Apply the schemas before starting the service; the local Make targets handle this order. See [queue status](../../doc/rfc/stovepipe/queue-state.md).
+Queue enablement is stored separately from validation results. The first `Ingest` call initializes a configured queue as disabled before applying any supplied enablement decision, and preserves existing state. `GetQueueStatus` returns the current summary, or the state covering a supplied commit through the existing source-control integration; it returns `NOT_FOUND` until initialization completes. History and summary writes can be retried after an interruption, so the summary may briefly lag internal state. Apply the schemas before starting the service; the local Make targets handle this order. See [queue status](../../doc/rfc/stovepipe/queue-state.md).

service/stovepipe/server/BUILD.bazel

@@ -33,8 +33,6 @@ go_library(
         "//stovepipe/controller/process:go_default_library",
         "//stovepipe/controller/record:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",
-        "//stovepipe/core/queuelog:go_default_library",
-        "//stovepipe/core/queuestate:go_default_library",
         "//stovepipe/core/requestlog:go_default_library",
         "//stovepipe/extension/buildissue:go_default_library",
         "//stovepipe/extension/buildissue/noop:go_default_library",

service/stovepipe/server/main.go

@@ -55,8 +55,6 @@ import (
 	"github.com/uber/submitqueue/stovepipe/controller/process"
 	"github.com/uber/submitqueue/stovepipe/controller/record"
 	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
-	"github.com/uber/submitqueue/stovepipe/core/queuelog"
-	"github.com/uber/submitqueue/stovepipe/core/queuestate"
 	"github.com/uber/submitqueue/stovepipe/core/requestlog"
 	"github.com/uber/submitqueue/stovepipe/extension/buildissue"
 	buildissuenoop "github.com/uber/submitqueue/stovepipe/extension/buildissue/noop"
@@ -252,12 +250,6 @@ func run() error {
 	brf := fakeBuildRunnerFactory{}
 
 	storageFty := storageFactory{backend: store}
-	queueEnablementController := queuestate.NewTransitionController(storageFty, sourceControl, queuelog.NewMaterializer(scope))
-	for _, queue := range tenants {
-		if _, err := queueEnablementController.Initialize(ctx, queue); err != nil {
-			return fmt.Errorf("initialize queue enablement %q: %w", queue, err)
-		}
-	}
 	materializer := requestlog.NewMaterializer(scope)
 	primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, storageFty, materializer, registry, sourceControl, brf, hookResolver{}, detectorFactory)
 	if err != nil {
@@ -296,7 +288,6 @@ func run() error {
 		materializer,
 		registry,
 		tenants,
-		queueEnablementController,
 	)
 	requestHistoryController := controller.NewRequestHistoryController(logger.Sugar(), scope, storageFty)
 	projectStatusController := controller.NewProjectStatusController(logger.Sugar(), scope, storageFty)

stovepipe/controller/ingest.go

@@ -26,6 +26,7 @@ import (
 	"github.com/uber/submitqueue/platform/metrics"
 	"github.com/uber/submitqueue/platform/publish"
 	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
+	"github.com/uber/submitqueue/stovepipe/core/queuelog"
 	"github.com/uber/submitqueue/stovepipe/core/queuestate"
 	"github.com/uber/submitqueue/stovepipe/core/requestlog"
 	"github.com/uber/submitqueue/stovepipe/entity"
@@ -58,6 +59,7 @@ type IngestController struct {
 
 // NewIngestController creates a new instance of the stovepipe ingest controller. It publishes
 // accepted requests to the topic registered under messagequeue.TopicKeyProcess in the registry.
+// Queue state is initialized by Ingest; construction performs no storage I/O.
 func NewIngestController(
 	logger *zap.SugaredLogger,
 	scope tally.Scope,
@@ -67,7 +69,6 @@ func NewIngestController(
 	materializer requestlog.Materializer,
 	registry consumer.TopicRegistry,
 	configuredQueues []string,
-	queueEnablementController queuestate.TransitionController,
 ) *IngestController {
 	queueNames := make(map[string]struct{}, len(configuredQueues))
 	for _, queue := range configuredQueues {
@@ -82,7 +83,7 @@ func NewIngestController(
 		materializer:              materializer,
 		registry:                  registry,
 		configuredQueues:          queueNames,
-		queueEnablementController: queueEnablementController,
+		queueEnablementController: queuestate.NewTransitionController(stores, sourceControl, queuelog.NewMaterializer(scope)),
 	}
 }
 
@@ -127,11 +128,12 @@ func (c *IngestController) Ingest(ctx context.Context, req entity.IngestRequest)
 		return entity.IngestResult{}, fmt.Errorf("failed to resolve head for queue=%s: %w", queue, err)
 	}
 
-	var state entity.QueueState
-	if req.HasEnablement {
+	if req.HasEnablement && req.Enablement != entity.QueueEnablementEnabled && req.Enablement != entity.QueueEnablementDisabled {
+		return entity.IngestResult{}, fmt.Errorf("unsupported enablement %q: %w", req.Enablement, queuestate.ErrInvalidChange)
+	}
+	state, err := c.queueEnablementController.Initialize(ctx, queue)
+	if err == nil && req.HasEnablement {
 		state, err = c.applyEnablementDecision(ctx, queue, req.Enablement, uri)
-	} else {
-		state, err = c.queueEnablementController.Initialize(ctx, queue)
 	}
 	if err != nil {
 		return entity.IngestResult{}, fmt.Errorf("resolve enablement for queue %q: %w", queue, err)

stovepipe/controller/ingest_test.go

@@ -103,7 +103,8 @@ func newIngestController(t *testing.T, ctrl *gomock.Controller) (*IngestControll
 	})
 	require.NoError(t, err)
 
-	c := NewIngestController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), staticCounterFactory{counter: m.counter}, m.factory, staticStorageFactory{store: store}, m.materializer, registry, []string{testQueue}, m.queueEnablementController)
+	c := NewIngestController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), staticCounterFactory{counter: m.counter}, m.factory, staticStorageFactory{store: store}, m.materializer, registry, []string{testQueue})
+	c.queueEnablementController = m.queueEnablementController
 	return c, m
 }
 
@@ -406,8 +407,18 @@ func newIngestEnablementFixture(t *testing.T) *ingestEnablementFixture {
 	}
 	f.logs[entity.QueueLogID(1)] = queuelog.NewQueueStateLog(f.queue)
 	m.queueStore.EXPECT().Get(gomock.Any(), testQueue).DoAndReturn(func(context.Context, string) (entity.Queue, error) {
+		if f.queue.Name == "" {
+			return entity.Queue{}, storage.ErrNotFound
+		}
 		return f.queue, nil
 	}).AnyTimes()
+	m.queueStore.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, row entity.Queue) error {
+		if f.queue.Name != "" {
+			return storage.ErrAlreadyExists
+		}
+		f.queue = row
+		return nil
+	}).AnyTimes()
 	m.queueStore.EXPECT().Update(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, row entity.Queue, oldVersion, newVersion int32) error {
 		if hook := f.beforeUpdate; hook != nil {
 			f.beforeUpdate = nil
@@ -442,8 +453,21 @@ func newIngestEnablementFixture(t *testing.T) *ingestEnablementFixture {
 	summaries := storagemock.NewMockQueueSummaryStore(ctrl)
 	m.store.EXPECT().GetQueueSummaryStore().Return(summaries).AnyTimes()
 	summaries.EXPECT().Get(gomock.Any()).DoAndReturn(func(context.Context) (entity.QueueSummary, error) {
+		if f.summary.Version == 0 {
+			return entity.QueueSummary{}, storage.ErrNotFound
+		}
 		return f.summary, nil
 	}).AnyTimes()
+	summaries.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, summary entity.QueueSummary) error {
+		if f.summaryErr != nil {
+			return f.summaryErr
+		}
+		if f.summary.Version != 0 {
+			return storage.ErrAlreadyExists
+		}
+		f.summary = summary
+		return nil
+	}).AnyTimes()
 	summaries.EXPECT().Update(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, summary entity.QueueSummary, oldVersion, newVersion int64) error {
 		if f.summaryErr != nil {
 			return f.summaryErr
@@ -459,7 +483,7 @@ func newIngestEnablementFixture(t *testing.T) *ingestEnablementFixture {
 	m.sc.EXPECT().Latest(gomock.Any()).DoAndReturn(func(context.Context) (string, error) { return f.head, nil }).AnyTimes()
 	history := scfake.New(sourcecontrol.Config{QueueName: testQueue}, []string{nextIngestURI, testURI})
 	m.sc.EXPECT().IsAncestor(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(history.IsAncestor).AnyTimes()
-	c.queueEnablementController = queuestate.NewTransitionController(staticStorageFactory{store: m.store}, m.factory, queuelog.NewMaterializer(tally.NoopScope))
+	f.controller = NewIngestController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: m.counter}, m.factory, staticStorageFactory{store: m.store}, m.materializer, c.registry, []string{testQueue})
 	return f
 }
 
@@ -520,29 +544,118 @@ func TestIngestEnablementDoesNotPauseWork(t *testing.T) {
 
 func TestIngestEnablementFailure(t *testing.T) {
 	for _, tc := range []struct {
-		name     string
-		explicit bool
-		err      error
+		name            string
+		explicit        bool
+		invalid         bool
+		initializeFails bool
+		err             error
 	}{
-		{"invalid decision", true, queuestate.ErrInvalidChange},
-		{"transition storage unavailable", true, errors.New("unavailable")},
-		{"observed state unavailable", false, errors.New("unavailable")},
+		{"invalid decision", true, true, false, queuestate.ErrInvalidChange},
+		{"transition storage unavailable", true, false, false, errors.New("unavailable")},
+		{"explicit initialization unavailable", true, false, true, errors.New("unavailable")},
+		{"observed state unavailable", false, false, true, errors.New("unavailable")},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			c, m := newIngestController(t, gomock.NewController(t))
 			m.factory.EXPECT().For(sourcecontrol.Config{QueueName: testQueue}).Return(m.sc, nil)
 			m.sc.EXPECT().Latest(gomock.Any()).Return(testURI, nil)
-			if tc.explicit {
-				m.queueEnablementController.EXPECT().Prepare(gomock.Any(), testQueue, entity.QueueEnablementUnknown, testURI).Return(queuestate.PreparedChange{}, tc.err)
-			} else {
+			desired := entity.QueueEnablementEnabled
+			if tc.invalid {
+				desired = entity.QueueEnablementUnknown
+			} else if tc.initializeFails {
 				m.queueEnablementController.EXPECT().Initialize(gomock.Any(), testQueue).Return(entity.QueueState{}, tc.err)
+			} else {
+				m.queueEnablementController.EXPECT().Initialize(gomock.Any(), testQueue).Return(entity.QueueState{Queue: testQueue, State: entity.QueueEnablementDisabled, Version: 1}, nil)
+				m.queueEnablementController.EXPECT().Prepare(gomock.Any(), testQueue, desired, testURI).Return(queuestate.PreparedChange{}, tc.err)
 			}
-			_, err := c.Ingest(context.Background(), entity.IngestRequest{Queue: testQueue, HasEnablement: tc.explicit})
+			_, err := c.Ingest(context.Background(), entity.IngestRequest{Queue: testQueue, HasEnablement: tc.explicit, Enablement: desired})
 			require.ErrorIs(t, err, tc.err)
 		})
 	}
 }
 
+func TestIngestInitializesQueueEnablement(t *testing.T) {
+	for _, existing := range []bool{false, true} {
+		for _, decision := range []struct {
+			name     string
+			explicit bool
+			state    entity.QueueEnablement
+			version  int64
+		}{
+			{"ordinary", false, entity.QueueEnablementDisabled, 1},
+			{"enable", true, entity.QueueEnablementEnabled, 2},
+			{"disable", true, entity.QueueEnablementDisabled, 1},
+		} {
+			name := "new/" + decision.name
+			if existing {
+				name = "existing empty/" + decision.name
+			}
+			t.Run(name, func(t *testing.T) {
+				f := newIngestEnablementFixture(t)
+				f.queue = entity.Queue{}
+				if existing {
+					f.queue = entity.Queue{Name: testQueue, Version: 5, InFlightCount: 3, LastGreenRequestID: "4"}
+				}
+				f.summary = entity.QueueSummary{}
+				clear(f.logs)
+				m := f.mocks
+				m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testURI).Return("7", nil).Times(2)
+				m.reqStore.EXPECT().Get(gomock.Any(), "7").Return(acceptedRequest("7"), nil).Times(2)
+				m.materializer.EXPECT().PersistLog(gomock.Any(), m.store, requestlog.NewRequestStateLog(acceptedRequest("7"), entity.RequestOutcomeReasonUnknown)).Return(nil).Times(2)
+				m.publisher.EXPECT().Publish(gomock.Any(), "process", gomock.Any()).Return(nil).Times(2)
+				request := entity.IngestRequest{Queue: testQueue, HasEnablement: decision.explicit, Enablement: decision.state}
+				result, err := f.controller.Ingest(context.Background(), request)
+				require.NoError(t, err)
+				require.Equal(t, decision.state, result.Enablement.State)
+				require.Equal(t, decision.version, result.Enablement.Version)
+				require.Positive(t, result.Enablement.ChangedAtMs)
+				require.Equal(t, result.Enablement, f.summary.Snapshot())
+				require.Len(t, f.logs, int(decision.version))
+				require.Equal(t, "7", f.queue.LatestRequestID)
+				if existing {
+					require.Equal(t, int32(3), f.queue.InFlightCount)
+					require.Equal(t, "4", f.queue.LastGreenRequestID)
+				}
+				retry, err := f.controller.Ingest(context.Background(), request)
+				require.NoError(t, err)
+				require.Equal(t, result, retry)
+			})
+		}
+	}
+}
+
+func TestIngestRetriesQueueInitialization(t *testing.T) {
+	for _, stage := range []string{"history", "summary"} {
+		t.Run(stage, func(t *testing.T) {
+			f := newIngestEnablementFixture(t)
+			f.queue, f.summary = entity.Queue{}, entity.QueueSummary{}
+			clear(f.logs)
+			if stage == "history" {
+				f.logErr = assert.AnError
+			} else {
+				f.summaryErr = assert.AnError
+			}
+			request := entity.IngestRequest{Queue: testQueue, HasEnablement: true, Enablement: entity.QueueEnablementEnabled}
+			_, err := f.controller.Ingest(context.Background(), request)
+			require.ErrorIs(t, err, assert.AnError)
+			initial := f.queue.EnablementState()
+			require.Equal(t, int64(1), initial.Version)
+			f.logErr, f.summaryErr = nil, nil
+			m := f.mocks
+			m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testURI).Return("7", nil)
+			m.reqStore.EXPECT().Get(gomock.Any(), "7").Return(acceptedRequest("7"), nil)
+			expectMaterializeAccepted(m, "7")
+			m.publisher.EXPECT().Publish(gomock.Any(), "process", gomock.Any()).Return(nil)
+			result, err := f.controller.Ingest(context.Background(), request)
+			require.NoError(t, err)
+			require.Equal(t, int64(2), result.Enablement.Version)
+			require.Equal(t, entity.QueueEnablementEnabled, result.Enablement.State)
+			require.Equal(t, initial, f.logs[entity.QueueLogID(1)].Snapshot())
+			require.Equal(t, result.Enablement, f.summary.Snapshot())
+		})
+	}
+}
+
 func TestIngestEnablementDecisionsPreserveBoundaries(t *testing.T) {
 	ctx := context.Background()
 	f := newIngestEnablementFixture(t)

test/e2e/stovepipe/issuetrack_test.go

@@ -80,10 +80,7 @@ func (s *StovepipeE2ESuite) TestIssueTracking_DLQAbandonsPrematureFinalNotificat
 	build := entity.Build{ID: "running-build", RequestID: request.ID, Status: entity.BuildStatusRunning, Version: 1}
 	require.NoError(t, store.GetRequestStore().Create(s.ctx, request))
 	require.NoError(t, store.GetBuildStore().Create(s.ctx, build))
-	queueState, err := store.GetQueueStore().Get(s.ctx, queue)
-	require.NoError(t, err)
-	queueState.InFlightCount = 1
-	require.NoError(t, store.GetQueueStore().Update(s.ctx, queueState, queueState.Version, queueState.Version+1))
+	require.NoError(t, store.GetQueueStore().Create(s.ctx, entity.Queue{Name: queue, InFlightCount: 1, Version: 1}))
 
 	mq, err := queuemysql.NewQueue(queuemysql.Params{
 		DB: s.queueDB, Logger: zap.NewNop(), MetricsScope: tally.NoopScope, Tenants: []string{queue},

test/e2e/stovepipe/queue_status_test.go

@@ -48,15 +48,20 @@ func (s *StovepipeE2ESuite) TestQueueStatusViaIngest() {
 	queue := queueStateTestQueue
 	uri := "git://" + queue + "/HEAD"
 	initial, err := s.client.GetQueueStatus(s.ctx, &pb.GetQueueStatusRequest{Queue: queue})
-	require.NoError(t, err)
-	require.NotNil(t, initial.Enablement)
-	require.Equal(t, pb.QueueEnablement_QUEUE_ENABLEMENT_DISABLED, initial.Enablement.State)
-	require.Equal(t, int64(1), initial.Enablement.Version)
-	require.Positive(t, initial.Enablement.ChangedAtMs)
+	require.Equal(t, codes.NotFound, status.Code(err))
+	require.Nil(t, initial)
 	ordinary, err := s.client.Ingest(s.ctx, &pb.IngestRequest{Queue: queue})
 	require.NoError(t, err)
 	require.NotEmpty(t, ordinary.Id)
-	require.Equal(t, initial.Enablement, ordinary.Enablement)
+	require.NotNil(t, ordinary.Enablement)
+	require.Equal(t, pb.QueueEnablement_QUEUE_ENABLEMENT_DISABLED, ordinary.Enablement.State)
+	require.Equal(t, int64(1), ordinary.Enablement.Version)
+	require.Positive(t, ordinary.Enablement.ChangedAtMs)
+	initial, err = s.client.GetQueueStatus(s.ctx, &pb.GetQueueStatusRequest{Queue: queue})
+	require.NoError(t, err)
+	require.Equal(t, ordinary.Enablement, initial.Enablement)
+	_, err = s.client.GetQueueStatus(s.ctx, &pb.GetQueueStatusRequest{Queue: queue, ChangeUri: "unknown"})
+	require.Equal(t, codes.FailedPrecondition, status.Code(err))
 	manager := queuestate.NewTransitionController(queueStateStorageFactory{backend: s.appStorage}, queueStateSourceFactory{}, queuelog.NewMaterializer(tally.NoopScope))
 	for _, tc := range []struct {
 		name    string
@@ -123,7 +128,6 @@ func (s *StovepipeE2ESuite) TestQueueStatusErrors() {
 	}{
 		{"missing queue", &pb.GetQueueStatusRequest{}, codes.InvalidArgument},
 		{"unregistered queue", &pb.GetQueueStatusRequest{Queue: "unregistered"}, codes.NotFound},
-		{"unknown commit", &pb.GetQueueStatusRequest{Queue: queueStateTestQueue, ChangeUri: "unknown"}, codes.FailedPrecondition},
 	} {
 		s.Run(tc.name, func() {
 			response, err := s.client.GetQueueStatus(s.ctx, tc.request)

test/e2e/stovepipe/suite_test.go

@@ -86,7 +86,6 @@ func (s *StovepipeE2ESuite) SetupSuite() {
 	s.queueDB, err = s.stack.ConnectMySQLService("mysql-queue")
 	require.NoError(t, err, "failed to connect to queue MySQL")
 
-	// Queue enablement is registered at service startup, so its schema must exist first.
 	testutil.ApplySchema(t, s.log, s.db, testutil.SchemaDir("platform/extension/counter/mysql/schema"))
 	testutil.ApplySchema(t, s.log, s.db, testutil.SchemaDir("stovepipe/extension/storage/mysql/schema"))
 	testutil.ApplySchema(t, s.log, s.queueDB, testutil.SchemaDir("platform/extension/messagequeue/mysql/schema"))