diff --git a/doc/rfc/stovepipe/queue-state.md b/doc/rfc/stovepipe/queue-state.md index c42df864a..1091bc7a2 100644 --- a/doc/rfc/stovepipe/queue-state.md +++ b/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. diff --git a/service/stovepipe/README.md b/service/stovepipe/README.md index 7e67d287d..b447b64fa 100644 --- a/service/stovepipe/README.md +++ b/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). diff --git a/service/stovepipe/server/BUILD.bazel b/service/stovepipe/server/BUILD.bazel index 73bfdd181..7c896fff6 100644 --- a/service/stovepipe/server/BUILD.bazel +++ b/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", diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index b4b2dc174..94d6d2167 100644 --- a/service/stovepipe/server/main.go +++ b/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) diff --git a/stovepipe/controller/ingest.go b/stovepipe/controller/ingest.go index 18a044af1..743267119 100644 --- a/stovepipe/controller/ingest.go +++ b/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) diff --git a/stovepipe/controller/ingest_test.go b/stovepipe/controller/ingest_test.go index 835e5dfe2..93edd2623 100644 --- a/stovepipe/controller/ingest_test.go +++ b/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) diff --git a/test/e2e/stovepipe/issuetrack_test.go b/test/e2e/stovepipe/issuetrack_test.go index 41ac01f85..780e23b02 100644 --- a/test/e2e/stovepipe/issuetrack_test.go +++ b/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}, diff --git a/test/e2e/stovepipe/queue_status_test.go b/test/e2e/stovepipe/queue_status_test.go index bee057b7a..343997769 100644 --- a/test/e2e/stovepipe/queue_status_test.go +++ b/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) diff --git a/test/e2e/stovepipe/suite_test.go b/test/e2e/stovepipe/suite_test.go index 37be8583f..4f09f0f58 100644 --- a/test/e2e/stovepipe/suite_test.go +++ b/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"))