diff --git a/stovepipe/controller/ingest.go b/stovepipe/controller/ingest.go index 697432c9..c9c817b4 100644 --- a/stovepipe/controller/ingest.go +++ b/stovepipe/controller/ingest.go @@ -126,6 +126,10 @@ func (c *IngestController) Ingest(ctx context.Context, req *pb.IngestRequest) (r return nil, err } + if err := c.advanceQueueLatestRequestID(ctx, queue, id); err != nil { + return nil, err + } + // Publish while the request is still pre-pipeline (Accepted). The process consumer is // idempotent (keyed on the request id, at-least-once), so re-publishing on a retry or a // duplicate report is safe and closes the "request created but publish failed" gap. Once @@ -211,6 +215,66 @@ func (c *IngestController) ensureRequest(ctx context.Context, id, queue, uri str return request, nil } +// ensureQueue returns the queue row for name, creating it if it does not yet exist. +// A concurrent creator (ErrAlreadyExists) is resolved by re-reading the canonical row. +func (c *IngestController) ensureQueue(ctx context.Context, name string) (entity.Queue, error) { + queueStore := c.store.GetQueueStore() + + got, err := queueStore.Get(ctx, name) + if err == nil { + return got, nil + } + if !errors.Is(err, storage.ErrNotFound) { + return entity.Queue{}, fmt.Errorf("IngestController failed to load queue %s: %w", name, err) + } + + queue := entity.Queue{ + Name: name, + Version: 1, + } + if err := queueStore.Create(ctx, queue); err != nil { + if !errors.Is(err, storage.ErrAlreadyExists) { + return entity.Queue{}, fmt.Errorf("IngestController failed to persist queue %s: %w", name, err) + } + // Raced with a concurrent creator; read the canonical row. + return queueStore.Get(ctx, name) + } + return queue, nil +} + +// advanceQueueLatestRequestID CAS-updates queue.latest_request_id to id when id is newer. +// Retries on optimistic-lock conflicts so concurrent ingests converge. +func (c *IngestController) advanceQueueLatestRequestID(ctx context.Context, queue, id string) error { + queueStore := c.store.GetQueueStore() + + for { + queueRow, err := c.ensureQueue(ctx, queue) + if err != nil { + return err + } + if queueRow.LatestRequestID != "" { + cmp, err := entity.CompareRequestID(queue, id, queueRow.LatestRequestID) + if err != nil { + return fmt.Errorf("IngestController failed to compare request ids for queue %s: %w", queue, err) + } + if cmp <= 0 { + return nil + } + } + + updated := queueRow + updated.LatestRequestID = id + newVersion := queueRow.Version + 1 + if err := queueStore.Update(ctx, updated, queueRow.Version, newVersion); err != nil { + if errors.Is(err, storage.ErrVersionMismatch) { + continue + } + return fmt.Errorf("IngestController failed to update queue %s latest_request_id: %w", queue, err) + } + return nil + } +} + // publishProcess publishes the request ID to the process stage, partitioned by queue so a // queue's requests stay ordered. func (c *IngestController) publishProcess(ctx context.Context, id, queue string) error { diff --git a/stovepipe/controller/ingest_test.go b/stovepipe/controller/ingest_test.go index 484fbee1..89b03fd0 100644 --- a/stovepipe/controller/ingest_test.go +++ b/stovepipe/controller/ingest_test.go @@ -43,29 +43,32 @@ const ( // ingestMocks bundles the mocks an Ingest test case wires expectations on. type ingestMocks struct { - counter *countermock.MockCounter - factory *scmock.MockFactory - sc *scmock.MockSourceControl - reqStore *storagemock.MockRequestStore - uriStore *storagemock.MockRequestURIStore - publisher *mqmock.MockPublisher + counter *countermock.MockCounter + factory *scmock.MockFactory + sc *scmock.MockSourceControl + reqStore *storagemock.MockRequestStore + uriStore *storagemock.MockRequestURIStore + queueStore *storagemock.MockQueueStore + publisher *mqmock.MockPublisher } func newIngestController(t *testing.T, ctrl *gomock.Controller) (*IngestController, ingestMocks) { t.Helper() m := ingestMocks{ - counter: countermock.NewMockCounter(ctrl), - factory: scmock.NewMockFactory(ctrl), - sc: scmock.NewMockSourceControl(ctrl), - reqStore: storagemock.NewMockRequestStore(ctrl), - uriStore: storagemock.NewMockRequestURIStore(ctrl), - publisher: mqmock.NewMockPublisher(ctrl), + counter: countermock.NewMockCounter(ctrl), + factory: scmock.NewMockFactory(ctrl), + sc: scmock.NewMockSourceControl(ctrl), + reqStore: storagemock.NewMockRequestStore(ctrl), + uriStore: storagemock.NewMockRequestURIStore(ctrl), + queueStore: storagemock.NewMockQueueStore(ctrl), + publisher: mqmock.NewMockPublisher(ctrl), } store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetRequestStore().Return(m.reqStore).AnyTimes() store.EXPECT().GetRequestURIStore().Return(m.uriStore).AnyTimes() + store.EXPECT().GetQueueStore().Return(m.queueStore).AnyTimes() queue := mqmock.NewMockQueue(ctrl) queue.EXPECT().Publisher().Return(m.publisher).AnyTimes() @@ -85,6 +88,23 @@ func expectResolve(m ingestMocks) { m.sc.EXPECT().Latest(gomock.Any()).Return(testURI, nil) } +// expectAdvanceLatestRequestID wires Get + Create + Update for queue.latest_request_id. +func expectAdvanceLatestRequestID(m ingestMocks, queue, id string) { + m.queueStore.EXPECT().Get(gomock.Any(), queue).Return(entity.Queue{}, storage.ErrNotFound) + m.queueStore.EXPECT().Create(gomock.Any(), entity.Queue{Name: queue, Version: 1}).Return(nil) + updated := entity.Queue{Name: queue, LatestRequestID: id, Version: 1} + m.queueStore.EXPECT().Update(gomock.Any(), updated, int32(1), int32(2)).Return(nil) +} + +// expectAdvanceLatestRequestIDNoOp wires Get when latest_request_id is already at id. +func expectAdvanceLatestRequestIDNoOp(m ingestMocks, queue, id string) { + m.queueStore.EXPECT().Get(gomock.Any(), queue).Return(entity.Queue{ + Name: queue, + LatestRequestID: id, + Version: 1, + }, nil) +} + func TestIngestController_Ingest(t *testing.T) { tests := []struct { name string @@ -104,6 +124,7 @@ func TestIngestController_Ingest(t *testing.T) { m.uriStore.EXPECT().Create(gomock.Any(), testQueue, testURI, "request/monorepo/main/7").Return(nil) m.reqStore.EXPECT().Get(gomock.Any(), "request/monorepo/main/7").Return(entity.Request{}, storage.ErrNotFound) m.reqStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil) + expectAdvanceLatestRequestID(m, testQueue, "request/monorepo/main/7") m.publisher.EXPECT().Publish(gomock.Any(), "process", gomock.Any()).Return(nil) }, wantID: "request/monorepo/main/7", @@ -115,6 +136,7 @@ func TestIngestController_Ingest(t *testing.T) { expectResolve(m) m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testQueue, testURI).Return("request/monorepo/main/3", nil) m.reqStore.EXPECT().Get(gomock.Any(), "request/monorepo/main/3").Return(entity.Request{ID: "request/monorepo/main/3", State: entity.RequestStateAccepted}, nil) + expectAdvanceLatestRequestIDNoOp(m, testQueue, "request/monorepo/main/3") m.publisher.EXPECT().Publish(gomock.Any(), "process", gomock.Any()).Return(nil) }, wantID: "request/monorepo/main/3", @@ -128,6 +150,7 @@ func TestIngestController_Ingest(t *testing.T) { m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testQueue, testURI).Return("request/monorepo/main/3", nil) m.reqStore.EXPECT().Get(gomock.Any(), "request/monorepo/main/3").Return(entity.Request{}, storage.ErrNotFound) m.reqStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil) + expectAdvanceLatestRequestID(m, testQueue, "request/monorepo/main/3") m.publisher.EXPECT().Publish(gomock.Any(), "process", gomock.Any()).Return(nil) }, wantID: "request/monorepo/main/3", @@ -142,6 +165,7 @@ func TestIngestController_Ingest(t *testing.T) { m.uriStore.EXPECT().Create(gomock.Any(), testQueue, testURI, "request/monorepo/main/7").Return(storage.ErrAlreadyExists) m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testQueue, testURI).Return("request/monorepo/main/3", nil) m.reqStore.EXPECT().Get(gomock.Any(), "request/monorepo/main/3").Return(entity.Request{ID: "request/monorepo/main/3", State: entity.RequestStateAccepted}, nil) + expectAdvanceLatestRequestIDNoOp(m, testQueue, "request/monorepo/main/3") m.publisher.EXPECT().Publish(gomock.Any(), "process", gomock.Any()).Return(nil) }, wantID: "request/monorepo/main/3", @@ -205,6 +229,7 @@ func TestIngestController_Ingest(t *testing.T) { m.uriStore.EXPECT().Create(gomock.Any(), testQueue, testURI, gomock.Any()).Return(nil) m.reqStore.EXPECT().Get(gomock.Any(), gomock.Any()).Return(entity.Request{}, storage.ErrNotFound) m.reqStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil) + expectAdvanceLatestRequestID(m, testQueue, "request/monorepo/main/7") m.publisher.EXPECT().Publish(gomock.Any(), "process", gomock.Any()).Return(errors.New("queue down")) }, wantErr: true, diff --git a/test/integration/stovepipe/suite_test.go b/test/integration/stovepipe/suite_test.go index d9f4e0e8..3f7495eb 100644 --- a/test/integration/stovepipe/suite_test.go +++ b/test/integration/stovepipe/suite_test.go @@ -136,6 +136,11 @@ func (s *StovepipeIntegrationSuite) TestIngestAPI() { resp2, err := s.client.Ingest(s.ctx, &pb.IngestRequest{Queue: queue}) require.NoError(t, err, "second Ingest failed") assert.Equal(t, id, resp2.Id, "re-ingest of the same head should dedup to the same id") + + // latest_request_id on the queue row. + var latestRequestID string + require.NoError(t, s.db.QueryRow("SELECT latest_request_id FROM queue WHERE name = ?", queue).Scan(&latestRequestID)) + assert.Equal(t, id, latestRequestID, "queue latest_request_id should point at the minted request") } // TestIngestEmptyQueue verifies the request-validation error surfaces over gRPC.