From bb30051711126f7cab70cea59bacdf0bafb2b262 Mon Sep 17 00:00:00 2001 From: Albert Wu Date: Mon, 13 Jul 2026 14:52:31 -0700 Subject: [PATCH] feat(request): materialize request log status Add projection winner selection, optimistic-concurrency convergence, and queue-summary repair without activating production call sites yet. Validation: make fmt && make build && make test && make e2e-test --- submitqueue/core/request/BUILD.bazel | 2 + submitqueue/core/request/materializer.go | 184 ++++++++++++ submitqueue/core/request/materializer_test.go | 263 ++++++++++++++++++ 3 files changed, 449 insertions(+) create mode 100644 submitqueue/core/request/materializer.go create mode 100644 submitqueue/core/request/materializer_test.go diff --git a/submitqueue/core/request/BUILD.bazel b/submitqueue/core/request/BUILD.bazel index 65abf76c..05906ec7 100644 --- a/submitqueue/core/request/BUILD.bazel +++ b/submitqueue/core/request/BUILD.bazel @@ -4,6 +4,7 @@ go_library( name = "go_default_library", srcs = [ "log.go", + "materializer.go", "receipt.go", "request.go", ], @@ -22,6 +23,7 @@ go_test( name = "go_default_test", srcs = [ "log_test.go", + "materializer_test.go", "receipt_test.go", "request_test.go", ], diff --git a/submitqueue/core/request/materializer.go b/submitqueue/core/request/materializer.go new file mode 100644 index 00000000..909d1a9b --- /dev/null +++ b/submitqueue/core/request/materializer.go @@ -0,0 +1,184 @@ +// 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" +) + +// Materializer appends request logs and projects the winning public request state. +// It owns winner selection, optimistic concurrency, and public projection repair. +type Materializer struct { + store storage.Storage +} + +// NewMaterializer creates a request read-model materializer. +func NewMaterializer(store storage.Storage) *Materializer { + return &Materializer{store: store} +} + +// PersistLog appends one audit log and materializes its winning state. +// Projection errors are returned so queue deliveries are retried rather than silently dropping the side write. +// Because the append happens first, retrying after a projection failure may retain another copy of the event in History. +func (m *Materializer) PersistLog(ctx context.Context, log entity.RequestLog) error { + if err := m.store.GetRequestLogStore().Insert(ctx, log); err != nil { + return fmt.Errorf("failed to insert request log request_id=%s: %w", log.RequestID, err) + } + + for { + summary, err := m.store.GetRequestSummaryStore().Get(ctx, log.RequestID) + if err != nil { + return fmt.Errorf("failed to get request summary request_id=%s: %w", log.RequestID, err) + } + + if logWins(log, summary) { + oldVersion := summary.Version + newVersion := oldVersion + 1 + updated := summary + updated.Status = log.Status + updated.RequestVersion = log.RequestVersion + updated.StatusTimestampMs = log.TimestampMs + updated.LastError = log.LastError + updated.Metadata = cloneMetadata(log.Metadata) + + if err := m.store.GetRequestSummaryStore().Update(ctx, updated, oldVersion, newVersion); err != nil { + if errors.Is(err, storage.ErrVersionMismatch) { + continue + } + return fmt.Errorf("failed to update request summary request_id=%s: %w", log.RequestID, err) + } + updated.Version = newVersion + summary = updated + } + + if err := m.repairPublicProjections(ctx, summary); err != nil { + return err + } + return nil + } +} + +// repairPublicProjections activates and repairs the public query projections. +// URI mappings are created before the queue summary, which acts as the marker that activation completed. +func (m *Materializer) repairPublicProjections(ctx context.Context, authoritative entity.RequestSummary) error { + desired := queueSummaryFromSummary(authoritative) + for { + current, err := m.store.GetRequestQueueSummaryStore().Get(ctx, desired.Queue, desired.ReceivedAtMs, desired.RequestID) + if errors.Is(err, storage.ErrNotFound) { + if err := m.createURIMappings(ctx, authoritative); err != nil { + return err + } + if err := m.store.GetRequestQueueSummaryStore().Create(ctx, desired); err != nil { + if errors.Is(err, storage.ErrAlreadyExists) { + continue + } + return fmt.Errorf("failed to recreate queue summary request_id=%s: %w", desired.RequestID, err) + } + return nil + } + if err != nil { + return fmt.Errorf("failed to get queue summary request_id=%s: %w", desired.RequestID, err) + } + if current.Version == desired.Version { + return nil + } + if current.Version > desired.Version { + // Another materializer already projected a newer authoritative snapshot. + return nil + } + if err := m.store.GetRequestQueueSummaryStore().Update(ctx, desired, current.Version, desired.Version); err != nil { + if errors.Is(err, storage.ErrVersionMismatch) { + continue + } + return fmt.Errorf("failed to update queue summary request_id=%s: %w", desired.RequestID, err) + } + return nil + } +} + +func (m *Materializer) createURIMappings(ctx context.Context, summary entity.RequestSummary) error { + 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) + } + } + return nil +} + +// logWins keeps accepting internal, treats accepted as the lowest public state, +// then applies versioned-terminal precedence and timestamp ordering. +func logWins(log entity.RequestLog, summary entity.RequestSummary) bool { + if summary.Status == entity.RequestStatusAccepting { + return log.Status != entity.RequestStatusAccepting + } + if log.Status == entity.RequestStatusAccepting { + return false + } + if log.Status == entity.RequestStatusAccepted && summary.Status != entity.RequestStatusAccepted { + return false + } + + incomingTerminal := isVersionedTerminal(log) + currentTerminal := isVersionedTerminalSummary(summary) + if incomingTerminal != currentTerminal { + return incomingTerminal + } + if incomingTerminal { + if log.RequestVersion != summary.RequestVersion { + return log.RequestVersion > summary.RequestVersion + } + } + return log.TimestampMs > summary.StatusTimestampMs +} + +func isVersionedTerminalSummary(summary entity.RequestSummary) bool { + return summary.RequestVersion > 0 && entity.IsRequestStateTerminal(entity.RequestState(summary.Status)) +} + +func isVersionedTerminal(log entity.RequestLog) bool { + return log.RequestVersion > 0 && entity.IsRequestStateTerminal(entity.RequestState(log.Status)) +} + +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 cloneMetadata(metadata map[string]string) map[string]string { + if metadata == nil { + return map[string]string{} + } + return maps.Clone(metadata) +} diff --git a/submitqueue/core/request/materializer_test.go b/submitqueue/core/request/materializer_test.go new file mode 100644 index 00000000..be42bef8 --- /dev/null +++ b/submitqueue/core/request/materializer_test.go @@ -0,0 +1,263 @@ +// 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" + "testing" + + "github.com/stretchr/testify/assert" + "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 TestMaterializer_PersistLog(t *testing.T) { + base := testRequestSummary() + log := entity.RequestLog{RequestID: "q/1", TimestampMs: 20, Status: entity.RequestStatusLanded, RequestVersion: 2, Metadata: map[string]string{}} + t.Run("winning log updates both projections", func(t *testing.T) { + ctrl := gomock.NewController(t) + store, summaryStore, queueStore, _, logStore := materializerStores(ctrl) + logStore.EXPECT().Insert(gomock.Any(), log).Return(nil) + summaryStore.EXPECT().Get(gomock.Any(), "q/1").Return(base, nil) + summaryStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).DoAndReturn(func(_ context.Context, updated entity.RequestSummary, _, _ int32) error { + assert.Equal(t, entity.RequestStatusLanded, updated.Status) + assert.Equal(t, int32(2), updated.RequestVersion) + return nil + }) + queueStore.EXPECT().Get(gomock.Any(), "q", int64(10), "q/1").Return(queueSummaryFromSummary(base), nil) + queueStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(nil) + require.NoError(t, NewMaterializer(store).PersistLog(context.Background(), log)) + }) + + t.Run("unversioned terminal status does not receive terminal precedence", func(t *testing.T) { + ctrl := gomock.NewController(t) + store, summaryStore, queueStore, _, logStore := materializerStores(ctrl) + current := base + current.Status = entity.RequestStatusLanded + current.RequestVersion = 0 + incoming := entity.RequestLog{RequestID: "q/1", TimestampMs: 20, Status: entity.RequestStatusProcessing, RequestVersion: 0, Metadata: map[string]string{}} + logStore.EXPECT().Insert(gomock.Any(), incoming).Return(nil) + summaryStore.EXPECT().Get(gomock.Any(), "q/1").Return(current, nil) + summaryStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).DoAndReturn(func(_ context.Context, updated entity.RequestSummary, _, _ int32) error { + assert.Equal(t, entity.RequestStatusProcessing, updated.Status) + return nil + }) + queueStore.EXPECT().Get(gomock.Any(), "q", int64(10), "q/1").Return(queueSummaryFromSummary(current), nil) + queueStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(nil) + require.NoError(t, NewMaterializer(store).PersistLog(context.Background(), incoming)) + }) + + t.Run("CAS conflict reloads and repairs winner", func(t *testing.T) { + ctrl := gomock.NewController(t) + store, summaryStore, queueStore, _, logStore := materializerStores(ctrl) + logStore.EXPECT().Insert(gomock.Any(), log).Return(nil) + summaryStore.EXPECT().Get(gomock.Any(), "q/1").Return(base, nil) + summaryStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(storage.ErrVersionMismatch) + advanced := base + advanced.Status = entity.RequestStatusLanded + advanced.RequestVersion = 2 + advanced.StatusTimestampMs = 20 + advanced.Version = 2 + summaryStore.EXPECT().Get(gomock.Any(), "q/1").Return(advanced, nil) + queueStore.EXPECT().Get(gomock.Any(), "q", int64(10), "q/1").Return(queueSummaryFromSummary(base), nil) + queueStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(nil) + require.NoError(t, NewMaterializer(store).PersistLog(context.Background(), log)) + }) + + t.Run("non-winning redelivery repairs stale queue projection", func(t *testing.T) { + ctrl := gomock.NewController(t) + store, summaryStore, queueStore, _, logStore := materializerStores(ctrl) + logStore.EXPECT().Insert(gomock.Any(), log).Return(nil) + advanced := base + advanced.Status = entity.RequestStatusLanded + advanced.RequestVersion = 2 + advanced.StatusTimestampMs = 20 + advanced.Version = 2 + summaryStore.EXPECT().Get(gomock.Any(), "q/1").Return(advanced, nil) + queueStore.EXPECT().Get(gomock.Any(), "q", int64(10), "q/1").Return(queueSummaryFromSummary(base), nil) + queueStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(nil) + require.NoError(t, NewMaterializer(store).PersistLog(context.Background(), log)) + }) + + t.Run("first public event activates URI and queue projections", func(t *testing.T) { + ctrl := gomock.NewController(t) + store, summaryStore, queueStore, uriStore, logStore := materializerStores(ctrl) + logStore.EXPECT().Insert(gomock.Any(), log).Return(nil) + summaryStore.EXPECT().Get(gomock.Any(), "q/1").Return(base, nil) + summaryStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(nil) + activated := base + activated.Status = entity.RequestStatusLanded + activated.RequestVersion = 2 + activated.StatusTimestampMs = 20 + activated.Version = 2 + queueStore.EXPECT().Get(gomock.Any(), "q", int64(10), "q/1").Return(entity.RequestQueueSummary{}, storage.ErrNotFound) + 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(activated)).Return(nil) + require.NoError(t, NewMaterializer(store).PersistLog(context.Background(), log)) + }) + + t.Run("retry after projection failure appends another audit row", func(t *testing.T) { + ctrl := gomock.NewController(t) + store, summaryStore, queueStore, _, logStore := materializerStores(ctrl) + materializer := NewMaterializer(store) + advanced := base + advanced.Status = entity.RequestStatusLanded + advanced.RequestVersion = 2 + advanced.StatusTimestampMs = 20 + advanced.Version = 2 + logStore.EXPECT().Insert(gomock.Any(), log).Return(nil).Times(2) + summaryStore.EXPECT().Get(gomock.Any(), "q/1").Return(advanced, nil).Times(2) + queueStore.EXPECT().Get(gomock.Any(), "q", int64(10), "q/1").Return(entity.RequestQueueSummary{}, errors.New("queue store down")) + queueStore.EXPECT().Get(gomock.Any(), "q", int64(10), "q/1").Return(queueSummaryFromSummary(advanced), nil) + + require.Error(t, materializer.PersistLog(context.Background(), log)) + require.NoError(t, materializer.PersistLog(context.Background(), log)) + }) + + t.Run("missing authoritative summary fails", func(t *testing.T) { + ctrl := gomock.NewController(t) + store, summaryStore, _, _, logStore := materializerStores(ctrl) + logStore.EXPECT().Insert(gomock.Any(), log).Return(nil) + summaryStore.EXPECT().Get(gomock.Any(), "q/1").Return(entity.RequestSummary{}, storage.ErrNotFound) + require.Error(t, NewMaterializer(store).PersistLog(context.Background(), log)) + }) + + t.Run("queue projection already ahead succeeds", func(t *testing.T) { + ctrl := gomock.NewController(t) + store, summaryStore, queueStore, _, logStore := materializerStores(ctrl) + logStore.EXPECT().Insert(gomock.Any(), log).Return(nil) + advanced := base + advanced.Status = entity.RequestStatusLanded + advanced.RequestVersion = 2 + advanced.StatusTimestampMs = 20 + advanced.Version = 2 + summaryStore.EXPECT().Get(gomock.Any(), "q/1").Return(advanced, nil) + queueAhead := queueSummaryFromSummary(advanced) + queueAhead.Version = 3 + queueStore.EXPECT().Get(gomock.Any(), "q", int64(10), "q/1").Return(queueAhead, nil) + require.NoError(t, NewMaterializer(store).PersistLog(context.Background(), log)) + }) +} + +func TestLogWins(t *testing.T) { + base := testRequestSummary() + tests := []struct { + name string + current entity.RequestSummary + incoming entity.RequestLog + want bool + }{ + { + name: "accepted activates accepting receipt", + current: entity.RequestSummary{Status: entity.RequestStatusAccepting, StatusTimestampMs: 200}, + incoming: entity.RequestLog{ + Status: entity.RequestStatusAccepted, TimestampMs: 100, + }, + want: true, + }, + { + name: "started activates accepting receipt despite older timestamp", + current: entity.RequestSummary{Status: entity.RequestStatusAccepting, StatusTimestampMs: 200}, + incoming: entity.RequestLog{ + Status: entity.RequestStatusStarted, TimestampMs: 100, + }, + want: true, + }, + { + name: "late accepted does not replace started", + current: entity.RequestSummary{Status: entity.RequestStatusStarted, StatusTimestampMs: 100}, + incoming: entity.RequestLog{ + Status: entity.RequestStatusAccepted, TimestampMs: 200, + }, + }, + { + name: "versioned terminal beats newer unversioned status", + current: entity.RequestSummary{Status: entity.RequestStatusProcessing, StatusTimestampMs: 200}, + incoming: entity.RequestLog{ + Status: entity.RequestStatusLanded, RequestVersion: 1, TimestampMs: 100, + }, + want: true, + }, + { + name: "nonterminal cannot replace versioned terminal", + current: entity.RequestSummary{Status: entity.RequestStatusLanded, RequestVersion: 1, StatusTimestampMs: 100}, + incoming: entity.RequestLog{ + Status: entity.RequestStatusProcessing, TimestampMs: 200, + }, + }, + { + name: "higher terminal request version wins", + current: entity.RequestSummary{Status: entity.RequestStatusError, RequestVersion: 1, StatusTimestampMs: 200}, + incoming: entity.RequestLog{ + Status: entity.RequestStatusLanded, RequestVersion: 2, TimestampMs: 100, + }, + want: true, + }, + { + name: "lower terminal request version loses", + current: entity.RequestSummary{Status: entity.RequestStatusLanded, RequestVersion: 2, StatusTimestampMs: 100}, + incoming: entity.RequestLog{ + Status: entity.RequestStatusError, RequestVersion: 1, TimestampMs: 200, + }, + }, + { + name: "equal terminal version uses later timestamp", + current: entity.RequestSummary{Status: entity.RequestStatusError, RequestVersion: 2, StatusTimestampMs: 100}, + incoming: entity.RequestLog{ + Status: entity.RequestStatusLanded, RequestVersion: 2, TimestampMs: 200, + }, + want: true, + }, + { + name: "without terminal winner later timestamp wins", + current: base, + incoming: entity.RequestLog{ + Status: entity.RequestStatusStarted, TimestampMs: base.StatusTimestampMs + 1, + }, + want: true, + }, + { + name: "exact version and timestamp tie keeps current winner", + current: entity.RequestSummary{Status: entity.RequestStatusError, RequestVersion: 2, StatusTimestampMs: 200}, + incoming: entity.RequestLog{ + Status: entity.RequestStatusLanded, RequestVersion: 2, TimestampMs: 200, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, logWins(tt.incoming, tt.current)) + }) + } +} + +func materializerStores(ctrl *gomock.Controller) (*storagemock.MockStorage, *storagemock.MockRequestSummaryStore, *storagemock.MockRequestQueueSummaryStore, *storagemock.MockRequestURIStore, *storagemock.MockRequestLogStore) { + store := storagemock.NewMockStorage(ctrl) + summaryStore := storagemock.NewMockRequestSummaryStore(ctrl) + queueStore := storagemock.NewMockRequestQueueSummaryStore(ctrl) + uriStore := storagemock.NewMockRequestURIStore(ctrl) + logStore := storagemock.NewMockRequestLogStore(ctrl) + store.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes() + store.EXPECT().GetRequestQueueSummaryStore().Return(queueStore).AnyTimes() + store.EXPECT().GetRequestURIStore().Return(uriStore).AnyTimes() + store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes() + return store, summaryStore, queueStore, uriStore, logStore +}