Skip to content

Commit dde2b9a

Browse files
prathshenoyprathshenoy
committed
feat(stovepipe): Add project ordering index
1 parent 6ea7eb7 commit dde2b9a

27 files changed

Lines changed: 846 additions & 21 deletions

‎doc/rfc/stovepipe/project-validation-index.md‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,17 @@
1+
# Project validation ordering index
2+
3+
The project validation index is a rebuildable mapping from `(queue, URI, degree_order, project)` to the immutable fact keyed by `(queue, URI, project)`. It contains no duplicate validation payload. Its primary key supports bounded range reads in descending breakage degree and ascending bytewise project ID, without sorting a non-key attribute or scanning every fact.
4+
5+
`degree_order` reverses nonnegative IEEE-754 degree bit patterns into an ascending integer key without rounding. Positive and negative zero share one key; invalid degrees and whole-repository facts cannot produce mappings. Binary identifiers preserve bytewise tie order.
6+
7+
## Persistence and repair
8+
9+
The record controller persists the winning immutable fact before ensuring its mapping. Redelivery repairs missing mappings even when the fact already exists, and derives their keys from the stored degree rather than a losing candidate. Project-recording logs and downstream terminal notifications follow successful mapping writes. No cross-store transaction is required.
10+
11+
`stovepipe/core/projectindex.Backfill` rebuilds one queue-scoped URI using bounded project-key scans and idempotent mapping writes. It is restartable and does not load all facts into memory. Schema deployment and historical backfill are operator-controlled, not performed by an RPC or service startup.
12+
13+
## Rollout
14+
15+
The ordering table is additive; the original fact identity and primary key are unchanged. Install the table before deploying index writers. Fully deploy writers before running historical backfill so facts recorded concurrently are also indexed. Populate and verify the historical mappings before adopting readers that rely exclusively on this projection; creating the table alone does not make the projection complete.
16+
17+
Keep the additive table when reverting application adoption. If index writers are disabled, backfill any facts recorded during that interval before adopting index-based readers again.

‎stovepipe/controller/record/BUILD.bazel‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ go_library(
1717
"//stovepipe/core/hookevent:go_default_library",
1818
"//stovepipe/core/loader:go_default_library",
1919
"//stovepipe/core/messagequeue:go_default_library",
20+
"//stovepipe/core/projectindex:go_default_library",
2021
"//stovepipe/core/requestlog:go_default_library",
2122
"//stovepipe/entity:go_default_library",
2223
"//stovepipe/extension/projectstatus:go_default_library",

‎stovepipe/controller/record/record.go‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,7 @@ import (
4646
"github.com/uber/submitqueue/stovepipe/core/hookevent"
4747
"github.com/uber/submitqueue/stovepipe/core/loader"
4848
stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
49+
"github.com/uber/submitqueue/stovepipe/core/projectindex"
4950
"github.com/uber/submitqueue/stovepipe/core/requestlog"
5051
"github.com/uber/submitqueue/stovepipe/entity"
5152
"github.com/uber/submitqueue/stovepipe/extension/projectstatus"
@@ -197,7 +198,7 @@ func (c *Controller) recordProjectFacts(ctx context.Context, store storage.Stora
197198
return err
198199
}
199200
for _, result := range results {
200-
_, _, err := c.recordValidationFact(ctx, store.GetValidationFactStore(), entity.ValidationFact{
201+
fact, _, err := c.recordValidationFact(ctx, store.GetValidationFactStore(), entity.ValidationFact{
201202
URI: request.URI,
202203
Project: result.Project,
203204
Degree: result.Degree,
@@ -207,6 +208,10 @@ func (c *Controller) recordProjectFacts(ctx context.Context, store storage.Stora
207208
if err != nil {
208209
return err
209210
}
211+
212+
if err := projectindex.Ensure(ctx, store.GetProjectValidationIndexStore(), fact); err != nil {
213+
return fmt.Errorf("failed to index project fact for request %q and project %q: %w", request.ID, fact.Project, err)
214+
}
210215
}
211216
if len(results) == 0 {
212217
return nil

‎stovepipe/controller/record/record_test.go‎

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -74,6 +74,7 @@ type recordMocks struct {
7474
reqStore *storagemock.MockRequestStore
7575
queueStore *storagemock.MockQueueStore
7676
factStore *storagemock.MockValidationFactStore
77+
indexStore *storagemock.MockProjectValidationIndexStore
7778
store *storagemock.MockStorage
7879
materializer *requestlogmock.MockMaterializer
7980
sourceControl *sourcecontrolmock.MockSourceControl
@@ -148,6 +149,7 @@ func newControllerForTopic(t *testing.T, ctrl *gomock.Controller, topicKey consu
148149
reqStore: storagemock.NewMockRequestStore(ctrl),
149150
queueStore: storagemock.NewMockQueueStore(ctrl),
150151
factStore: storagemock.NewMockValidationFactStore(ctrl),
152+
indexStore: storagemock.NewMockProjectValidationIndexStore(ctrl),
151153
store: storagemock.NewMockStorage(ctrl),
152154
materializer: requestlogmock.NewMockMaterializer(ctrl),
153155
sourceControl: sourcecontrolmock.NewMockSourceControl(ctrl),
@@ -158,6 +160,7 @@ func newControllerForTopic(t *testing.T, ctrl *gomock.Controller, topicKey consu
158160
m.store.EXPECT().GetRequestStore().Return(m.reqStore).AnyTimes()
159161
m.store.EXPECT().GetQueueStore().Return(m.queueStore).AnyTimes()
160162
m.store.EXPECT().GetValidationFactStore().Return(m.factStore).AnyTimes()
163+
m.store.EXPECT().GetProjectValidationIndexStore().Return(m.indexStore).AnyTimes()
161164
m.materializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).DoAndReturn(
162165
func(_ context.Context, _ storage.Storage, log entity.RequestLog) error {
163166
assert.Equal(t, entity.RequestEventValidationFactRecorded, log.Event)
@@ -402,6 +405,13 @@ func TestProcess_RecordsNamedProjectStatusResults(t *testing.T) {
402405
}).
403406
Times(3)
404407

408+
var indexes []entity.ProjectValidationIndex
409+
m.indexStore.EXPECT().Create(gomock.Any(), gomock.Any()).
410+
DoAndReturn(func(_ context.Context, entry entity.ProjectValidationIndex) error {
411+
indexes = append(indexes, entry)
412+
return nil
413+
}).Times(2)
414+
405415
var logs []entity.RequestLog
406416
eventMaterializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).
407417
DoAndReturn(func(_ context.Context, _ storage.Storage, log entity.RequestLog) error {
@@ -412,6 +422,12 @@ func TestProcess_RecordsNamedProjectStatusResults(t *testing.T) {
412422

413423
require.NoError(t, c.Process(queueContext(), delivery(t, ctrl, recordPayload(t, testID))))
414424
require.Len(t, facts, 3)
425+
require.Len(t, indexes, 2)
426+
for i, fact := range facts[1:] {
427+
entry, err := entity.NewProjectValidationIndex(fact)
428+
require.NoError(t, err)
429+
assert.Equal(t, entry, indexes[i])
430+
}
415431
assert.Equal(t, []string{"", "project-a", "project-b"}, []string{facts[0].Project, facts[1].Project, facts[2].Project})
416432
for _, fact := range facts {
417433
assert.Equal(t, testURI, fact.URI)
@@ -1217,3 +1233,43 @@ func TestProcess_QueueMismatchFails(t *testing.T) {
12171233

12181234
require.Error(t, c.Process(queueContext(), d))
12191235
}
1236+
1237+
func TestRecordProjectFactsRepairsIndexOnRetry(t *testing.T) {
1238+
for _, retryErr := range []error{nil, storage.ErrAlreadyExists} {
1239+
name := "index repaired"
1240+
if retryErr != nil {
1241+
name = "index already repaired"
1242+
}
1243+
t.Run(name, func(t *testing.T) {
1244+
ctrl := gomock.NewController(t)
1245+
c, m := newController(t, ctrl)
1246+
request := requestWithState(entity.RequestStateFailed)
1247+
resolverFactory := projectstatusmock.NewMockResolverFactory(ctrl)
1248+
resolver := projectstatusmock.NewMockResolver(ctrl)
1249+
c.projectStatusResolverFactory = resolverFactory
1250+
resolverFactory.EXPECT().For(projectstatus.ResolverConfig{QueueName: testQueue}).Return(resolver, nil).Times(2)
1251+
stored := entity.ValidationFact{URI: request.URI, Project: "p", Degree: 0.25, RequestID: request.ID, CreatedAt: 10}
1252+
entry, err := entity.NewProjectValidationIndex(stored)
1253+
require.NoError(t, err)
1254+
writeErr := errors.New("index write unavailable")
1255+
materializer := requestlogmock.NewMockMaterializer(ctrl)
1256+
c.materializer = materializer
1257+
1258+
gomock.InOrder(
1259+
resolver.EXPECT().Resolve(gomock.Any(), request, request.TerminalBuildID).Return([]projectstatus.Result{{Project: "p", Degree: 0.25}}, nil),
1260+
m.factStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil),
1261+
m.indexStore.EXPECT().Create(gomock.Any(), entry).Return(writeErr),
1262+
resolver.EXPECT().Resolve(gomock.Any(), request, request.TerminalBuildID).Return([]projectstatus.Result{{Project: "p", Degree: 0.75}}, nil),
1263+
m.factStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(storage.ErrAlreadyExists),
1264+
m.factStore.EXPECT().Get(gomock.Any(), request.URI, "p").Return(stored, nil),
1265+
m.indexStore.EXPECT().Create(gomock.Any(), entry).Return(retryErr),
1266+
materializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).DoAndReturn(func(_ context.Context, _ storage.Storage, log entity.RequestLog) error {
1267+
assert.Equal(t, entity.RequestEventProjectFactsRecorded, log.Event)
1268+
return nil
1269+
}),
1270+
)
1271+
require.ErrorIs(t, c.recordProjectFacts(queueContext(), m.store, request), writeErr)
1272+
require.NoError(t, c.recordProjectFacts(queueContext(), m.store, request))
1273+
})
1274+
}
1275+
}

‎stovepipe/core/projectindex/BUILD.bazel‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
load("@rules_go//go:def.bzl", "go_library", "go_test")
2+
3+
go_library(
4+
name = "go_default_library",
5+
srcs = ["projectindex.go"],
6+
importpath = "github.com/uber/submitqueue/stovepipe/core/projectindex",
7+
visibility = ["//visibility:public"],
8+
deps = [
9+
"//stovepipe/entity:go_default_library",
10+
"//stovepipe/extension/storage:go_default_library",
11+
],
12+
)
13+
14+
go_test(
15+
name = "go_default_test",
16+
srcs = ["projectindex_test.go"],
17+
embed = [":go_default_library"],
18+
deps = [
19+
"//stovepipe/entity:go_default_library",
20+
"//stovepipe/extension/storage:go_default_library",
21+
"//stovepipe/extension/storage/mock:go_default_library",
22+
"@com_github_stretchr_testify//require:go_default_library",
23+
"@org_uber_go_mock//gomock:go_default_library",
24+
],
25+
)

‎stovepipe/core/projectindex/projectindex.go‎

Lines changed: 72 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,72 @@
1+
// Copyright (c) 2026 Uber Technologies, Inc.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
// Package projectindex maintains the primary-key ordering projection of immutable project facts.
16+
package projectindex
17+
18+
import (
19+
"context"
20+
"errors"
21+
"fmt"
22+
23+
"github.com/uber/submitqueue/stovepipe/entity"
24+
"github.com/uber/submitqueue/stovepipe/extension/storage"
25+
)
26+
27+
// Ensure idempotently maps a persisted named-project fact into its degree-ordered key.
28+
// fact must be the authoritative stored value, not a candidate that lost a create race.
29+
func Ensure(ctx context.Context, index storage.ProjectValidationIndexStore, fact entity.ValidationFact) error {
30+
entry, err := entity.NewProjectValidationIndex(fact)
31+
if err != nil {
32+
return err
33+
}
34+
35+
if err := index.Create(ctx, entry); err != nil && !errors.Is(err, storage.ErrAlreadyExists) {
36+
return fmt.Errorf("failed to index project %q at URI %q: %w", fact.Project, fact.URI, err)
37+
}
38+
39+
return nil
40+
}
41+
42+
// Backfill rebuilds one URI's ordering mappings through bounded project-key scans.
43+
// It is restartable and does not delete entries or load every fact into memory.
44+
func Backfill(ctx context.Context, facts storage.ValidationFactStore, index storage.ProjectValidationIndexStore, uri string, batchSize int) error {
45+
if uri == "" || batchSize <= 0 {
46+
return fmt.Errorf("project index backfill requires a URI and positive batch size")
47+
}
48+
49+
after := ""
50+
for {
51+
page, err := facts.ListProjects(ctx, uri, after, batchSize)
52+
if err != nil {
53+
return fmt.Errorf("failed to list facts for project index backfill: %w", err)
54+
}
55+
56+
for _, fact := range page {
57+
if fact.URI != uri {
58+
return fmt.Errorf("project index backfill fact does not match URI %q", uri)
59+
}
60+
61+
if err := Ensure(ctx, index, fact); err != nil {
62+
return err
63+
}
64+
}
65+
66+
if len(page) < batchSize {
67+
return nil
68+
}
69+
70+
after = page[len(page)-1].Project
71+
}
72+
}

‎stovepipe/core/projectindex/projectindex_test.go‎

Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,75 @@
1+
// Copyright (c) 2026 Uber Technologies, Inc.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package projectindex
16+
17+
import (
18+
"context"
19+
"errors"
20+
"testing"
21+
22+
"github.com/stretchr/testify/require"
23+
"github.com/uber/submitqueue/stovepipe/entity"
24+
"github.com/uber/submitqueue/stovepipe/extension/storage"
25+
storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock"
26+
"go.uber.org/mock/gomock"
27+
)
28+
29+
func TestBackfillRepairsPartialProgress(t *testing.T) {
30+
ctrl := gomock.NewController(t)
31+
facts := storagemock.NewMockValidationFactStore(ctrl)
32+
index := storagemock.NewMockProjectValidationIndexStore(ctrl)
33+
first := entity.ValidationFact{URI: "commit", Project: "a", Degree: 0.25}
34+
second := entity.ValidationFact{URI: "commit", Project: "b", Degree: 1}
35+
firstEntry, err := entity.NewProjectValidationIndex(first)
36+
require.NoError(t, err)
37+
secondEntry, err := entity.NewProjectValidationIndex(second)
38+
require.NoError(t, err)
39+
writeErr := errors.New("index unavailable")
40+
gomock.InOrder(
41+
facts.EXPECT().ListProjects(gomock.Any(), "commit", "", 1).Return([]entity.ValidationFact{first}, nil),
42+
index.EXPECT().Create(gomock.Any(), firstEntry).Return(nil),
43+
facts.EXPECT().ListProjects(gomock.Any(), "commit", "a", 1).Return([]entity.ValidationFact{second}, nil),
44+
index.EXPECT().Create(gomock.Any(), secondEntry).Return(writeErr),
45+
facts.EXPECT().ListProjects(gomock.Any(), "commit", "", 1).Return([]entity.ValidationFact{first}, nil),
46+
index.EXPECT().Create(gomock.Any(), firstEntry).Return(storage.ErrAlreadyExists),
47+
facts.EXPECT().ListProjects(gomock.Any(), "commit", "a", 1).Return([]entity.ValidationFact{second}, nil),
48+
index.EXPECT().Create(gomock.Any(), secondEntry).Return(nil),
49+
facts.EXPECT().ListProjects(gomock.Any(), "commit", "b", 1).Return(nil, nil),
50+
)
51+
require.ErrorIs(t, Backfill(context.Background(), facts, index, "commit", 1), writeErr)
52+
require.NoError(t, Backfill(context.Background(), facts, index, "commit", 1))
53+
}
54+
55+
func TestBackfillRejectsInvalidFactsAndReads(t *testing.T) {
56+
for _, tt := range []struct {
57+
name string
58+
facts []entity.ValidationFact
59+
readErr error
60+
}{
61+
{name: "read failure", readErr: errors.New("read unavailable")},
62+
{name: "wrong URI", facts: []entity.ValidationFact{{URI: "other", Project: "a", Degree: 0.5}}},
63+
{name: "repository fact", facts: []entity.ValidationFact{{URI: "commit", Degree: 0.5}}},
64+
} {
65+
t.Run(tt.name, func(t *testing.T) {
66+
ctrl := gomock.NewController(t)
67+
facts := storagemock.NewMockValidationFactStore(ctrl)
68+
index := storagemock.NewMockProjectValidationIndexStore(ctrl)
69+
facts.EXPECT().ListProjects(gomock.Any(), "commit", "", 2).Return(tt.facts, tt.readErr)
70+
require.Error(t, Backfill(context.Background(), facts, index, "commit", 2))
71+
})
72+
}
73+
require.Error(t, Backfill(context.Background(), nil, nil, "", 1))
74+
require.Error(t, Backfill(context.Background(), nil, nil, "commit", 0))
75+
}

‎stovepipe/entity/BUILD.bazel‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ go_library(
88
"ingest.go",
99
"list.go",
1010
"project_status.go",
11+
"project_validation_index.go",
1112
"queue.go",
1213
"queue_config.go",
1314
"queue_log.go",
@@ -31,6 +32,7 @@ go_test(
3132
name = "go_default_test",
3233
srcs = [
3334
"build_test.go",
35+
"project_validation_index_test.go",
3436
"queue_state_test.go",
3537
"request_id_test.go",
3638
"request_log_test.go",

‎stovepipe/entity/project_validation_index.go‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
// Copyright (c) 2026 Uber Technologies, Inc.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package entity
16+
17+
import (
18+
"fmt"
19+
"math"
20+
)
21+
22+
// ProjectValidationIndex maps an ordered key to an immutable project fact within a queue.
23+
type ProjectValidationIndex struct {
24+
// URI is the commit containing the project result.
25+
URI string
26+
// DegreeOrder is the ascending integer key for descending breakage degree.
27+
DegreeOrder int64
28+
// Project identifies the fact and breaks equal-degree ties in ascending bytewise order.
29+
Project string
30+
}
31+
32+
// NewProjectValidationIndex derives an exact ordering key from a named project's fact.
33+
// Invalid degrees, empty URIs and whole-repository facts return an error.
34+
func NewProjectValidationIndex(fact ValidationFact) (ProjectValidationIndex, error) {
35+
if fact.URI == "" || fact.Project == "" || math.IsNaN(fact.Degree) || fact.Degree < DegreeGreen || fact.Degree > DegreeBroken {
36+
return ProjectValidationIndex{}, fmt.Errorf("project validation index requires a URI, project and degree in [0, 1]")
37+
}
38+
degree := fact.Degree
39+
if degree == 0 {
40+
degree = 0 // Canonicalize negative zero to the same key as positive zero.
41+
}
42+
// Nonnegative IEEE-754 bit patterns increase with degree; integer subtraction reverses that order without rounding.
43+
order := int64(math.Float64bits(DegreeBroken) - math.Float64bits(degree))
44+
return ProjectValidationIndex{URI: fact.URI, DegreeOrder: order, Project: fact.Project}, nil
45+
}

0 commit comments

Comments
 (0)