Skip to content

Commit 2fc21ab

Browse files
mnoah1mnoah1
committed
feat(stovepipe): [DRAFT] store queue state and read models
Summary: Queue enablement needs durable internal state and separate read models so orchestration updates do not alter enablement history. Extend the existing queue snapshot, add immutable QueueLog and derived QueueSummary records, and implement key-oriented stores with conditional writes. Include the one-time migration for existing queue tables and backend tests for byte-exact queue identity and independent projection versions. Part 2 of 4, built on the API and RFC. This layer supplies storage for transitions and lookup. Test Plan: Run the Stovepipe storage unit and MySQL integration tests. Migration coverage verifies existing operational values survive. Before deploying to an existing database, apply queue_enablement.sql once and create the queue log and summary tables.
1 parent 97b3f3e commit 2fc21ab

30 files changed

Lines changed: 1091 additions & 16 deletions

‎stovepipe/entity/BUILD.bazel‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,9 @@ go_library(
99
"project_status.go",
1010
"queue.go",
1111
"queue_config.go",
12+
"queue_log.go",
13+
"queue_state.go",
14+
"queue_summary.go",
1215
"request.go",
1316
"request_acceptance.go",
1417
"request_history.go",
@@ -26,6 +29,7 @@ go_test(
2629
name = "go_default_test",
2730
srcs = [
2831
"build_test.go",
32+
"queue_state_test.go",
2933
"request_id_test.go",
3034
"request_log_test.go",
3135
"request_test.go",

‎stovepipe/entity/queue.go‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,26 @@ type Queue struct {
4343
// Empty until the first request is created. Coalescing compares IDs via CompareRequestID.
4444
LatestRequestID string `json:"latest_request_id"`
4545

46+
// Enablement is the applied enabled/disabled state; empty means unregistered.
47+
Enablement QueueEnablement `json:"enablement"`
48+
49+
// EnablementVersion is the queue-scoped enablement sequence; zero means unregistered.
50+
// Operational changes do not advance this sequence.
51+
EnablementVersion int64 `json:"enablement_version"`
52+
53+
// EffectiveFromCommitURI is the inclusive immutable enablement boundary; empty for initial registration.
54+
EffectiveFromCommitURI string `json:"effective_from_commit_uri"`
55+
56+
// EnablementChangedAtMs is the enablement application timestamp in Unix milliseconds.
57+
EnablementChangedAtMs int64 `json:"enablement_changed_at_ms"`
58+
4659
// Version is the version of the object. It is used for optimistic locking.
4760
// Versioning starts at 1 and is incremented for each change to the object.
4861
Version int32 `json:"version"`
4962
}
63+
64+
// EnablementState returns the applied occurrence independently of operational row version.
65+
func (q Queue) EnablementState() QueueState {
66+
return QueueState{Queue: q.Name, State: q.Enablement, Version: q.EnablementVersion,
67+
EffectiveFromCommitURI: q.EffectiveFromCommitURI, ChangedAtMs: q.EnablementChangedAtMs}
68+
}

‎stovepipe/entity/queue_log.go‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,52 @@
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 "fmt"
18+
19+
// QueueLog is one immutable occurrence of a committed enablement state.
20+
type QueueLog struct {
21+
// ID is the stable queue-scoped occurrence identity derived from QueueStateVersion.
22+
ID string
23+
// Queue is the stable repo-and-ref namespace.
24+
Queue string
25+
// TimestampMs is the application's timestamp in Unix milliseconds.
26+
TimestampMs int64
27+
// State is the committed enabled or disabled value.
28+
State QueueEnablement
29+
// QueueStateVersion is the positive version identifying the applied state.
30+
QueueStateVersion int64
31+
// EffectiveFromCommitURI is the inclusive immutable boundary; empty only at version 1.
32+
EffectiveFromCommitURI string
33+
}
34+
35+
// QueueLogID returns the stable occurrence identity for a state version.
36+
func QueueLogID(version int64) string {
37+
return fmt.Sprintf("state/%d", version)
38+
}
39+
40+
// Snapshot returns the applied state represented by the occurrence.
41+
func (l QueueLog) Snapshot() QueueState {
42+
return QueueState{Queue: l.Queue, State: l.State, Version: l.QueueStateVersion,
43+
EffectiveFromCommitURI: l.EffectiveFromCommitURI, ChangedAtMs: l.TimestampMs}
44+
}
45+
46+
// Validate verifies a retained occurrence's identity and snapshot invariants.
47+
func (l QueueLog) Validate() error {
48+
if l.ID != QueueLogID(l.QueueStateVersion) {
49+
return fmt.Errorf("queue state log identity differs from its state version")
50+
}
51+
return l.Snapshot().Validate()
52+
}

‎stovepipe/entity/queue_state.go‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,61 @@
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 "fmt"
18+
19+
// QueueEnablement is applied enablement, independent of execution controls and verdicts.
20+
type QueueEnablement string
21+
22+
const (
23+
// QueueEnablementUnknown is unset and never an explicit disablement.
24+
QueueEnablementUnknown QueueEnablement = ""
25+
// QueueEnablementEnabled identifies enabled queue coverage.
26+
QueueEnablementEnabled QueueEnablement = "enabled"
27+
// QueueEnablementDisabled identifies disabled queue coverage.
28+
QueueEnablementDisabled QueueEnablement = "disabled"
29+
)
30+
31+
// QueueState is one applied enablement occurrence, independent of storage row versions.
32+
type QueueState struct {
33+
// Queue is the stable repo-and-ref namespace, at most 255 bytes.
34+
Queue string
35+
// State is enabled or disabled.
36+
State QueueEnablement
37+
// EffectiveFromCommitURI is the inclusive immutable boundary, at most 255 bytes; empty only at version 1.
38+
EffectiveFromCommitURI string
39+
// ChangedAtMs is the application timestamp in Unix milliseconds.
40+
ChangedAtMs int64
41+
// Version is the positive queue-scoped state version.
42+
Version int64
43+
}
44+
45+
// Validate verifies the invariants of a persisted state snapshot.
46+
func (s QueueState) Validate() error {
47+
if s.Queue == "" || len(s.Queue) > 255 || s.Version <= 0 || s.ChangedAtMs <= 0 {
48+
return fmt.Errorf("invalid queue state identity, version or timestamp")
49+
}
50+
if s.State != QueueEnablementEnabled && s.State != QueueEnablementDisabled {
51+
return fmt.Errorf("invalid queue enablement %q", s.State)
52+
}
53+
if s.Version == 1 {
54+
if s.State != QueueEnablementDisabled || s.EffectiveFromCommitURI != "" {
55+
return fmt.Errorf("initial queue state must be disabled without a boundary")
56+
}
57+
} else if s.EffectiveFromCommitURI == "" || len(s.EffectiveFromCommitURI) > 255 {
58+
return fmt.Errorf("invalid queue state commit boundary")
59+
}
60+
return nil
61+
}

‎stovepipe/entity/queue_state_test.go‎

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,58 @@
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+
"strings"
19+
"testing"
20+
21+
"github.com/stretchr/testify/require"
22+
)
23+
24+
func TestQueueStateInvariants(t *testing.T) {
25+
initial := QueueState{Queue: "repo/main", State: QueueEnablementDisabled, ChangedAtMs: 1000, Version: 1}
26+
for _, tc := range []struct {
27+
name string
28+
change func(*QueueState)
29+
valid bool
30+
}{
31+
{"initial", func(*QueueState) {}, true},
32+
{"enabled occurrence", func(s *QueueState) { s.Version = 2; s.State = QueueEnablementEnabled; s.EffectiveFromCommitURI = "A" }, true},
33+
{"missing queue", func(s *QueueState) { s.Queue = "" }, false},
34+
{"oversized queue", func(s *QueueState) { s.Queue = strings.Repeat("x", 256) }, false},
35+
{"unknown state", func(s *QueueState) { s.State = QueueEnablementUnknown }, false},
36+
{"invalid version", func(s *QueueState) { s.Version = 0 }, false},
37+
{"missing timestamp", func(s *QueueState) { s.ChangedAtMs = 0 }, false},
38+
{"enabled initial", func(s *QueueState) { s.State = QueueEnablementEnabled }, false},
39+
{"initial boundary", func(s *QueueState) { s.EffectiveFromCommitURI = "A" }, false},
40+
{"missing later boundary", func(s *QueueState) { s.Version = 2 }, false},
41+
{"oversized boundary", func(s *QueueState) { s.Version = 2; s.EffectiveFromCommitURI = strings.Repeat("x", 256) }, false},
42+
} {
43+
t.Run(tc.name, func(t *testing.T) {
44+
state := initial
45+
tc.change(&state)
46+
if tc.valid {
47+
require.NoError(t, state.Validate())
48+
} else {
49+
require.Error(t, state.Validate())
50+
}
51+
})
52+
}
53+
log := QueueLog{ID: QueueLogID(1), Queue: initial.Queue, TimestampMs: initial.ChangedAtMs, State: initial.State, QueueStateVersion: 1}
54+
require.NoError(t, log.Validate())
55+
require.Equal(t, initial, log.Snapshot())
56+
log.ID = "other"
57+
require.Error(t, log.Validate())
58+
}

‎stovepipe/entity/queue_summary.go‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,47 @@
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 "fmt"
18+
19+
// QueueSummary is the current read model derived from retained queue history.
20+
type QueueSummary struct {
21+
// Queue is the stable repo-and-ref namespace, at most 255 bytes.
22+
Queue string
23+
// State is the latest materialized enabled/disabled value.
24+
State QueueEnablement
25+
// QueueStateVersion identifies the applied enablement occurrence represented here.
26+
QueueStateVersion int64
27+
// EffectiveFromCommitURI is the inclusive immutable boundary; empty at state version 1.
28+
EffectiveFromCommitURI string
29+
// StateTimestampMs is the occurrence timestamp in Unix milliseconds.
30+
StateTimestampMs int64
31+
// Version is the positive optimistic-lock version of this projection.
32+
Version int64
33+
}
34+
35+
// Snapshot returns the enablement occurrence represented by this summary.
36+
func (s QueueSummary) Snapshot() QueueState {
37+
return QueueState{Queue: s.Queue, State: s.State, Version: s.QueueStateVersion,
38+
EffectiveFromCommitURI: s.EffectiveFromCommitURI, ChangedAtMs: s.StateTimestampMs}
39+
}
40+
41+
// Validate verifies projection identity, version and represented state.
42+
func (s QueueSummary) Validate() error {
43+
if s.Version <= 0 {
44+
return fmt.Errorf("invalid queue summary version")
45+
}
46+
return s.Snapshot().Validate()
47+
}

‎stovepipe/extension/storage/BUILD.bazel‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,9 @@ go_library(
44
name = "go_default_library",
55
srcs = [
66
"build_store.go",
7+
"queue_log_store.go",
78
"queue_store.go",
9+
"queue_summary_store.go",
810
"request_acceptance_store.go",
911
"request_log_store.go",
1012
"request_store.go",

‎stovepipe/extension/storage/README.md‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,3 +21,5 @@ A `Get` immediately following a successful write (`Create`/`Update`) — by the
2121
Same design space as [`submitqueue/extension/storage`](../../../submitqueue/extension/storage/README.md#key-value-contract): every method must be satisfiable by a plain key-value backend as cheaply as by MySQL — get/put/conditional-update by primary key only, no query-by-attribute or server-side filtering. `RequestURIStore` is the reverse-lookup example here (which request owns a given commit URI), kept as its own store rather than a secondary index on `RequestStore`.
2222

2323
`RequestAcceptanceStore` is the ordered lookup mapping for List: immutable `(queue, accepted_at_ms, request_id)` keys with bounded primary-key range scans, not a secondary index or a second status projection. The request-log materializer ensures each known acceptance has a mapping after summary persistence, including retries where the summary is unchanged. Unknown acceptance times are not mapped; historical backfill is separate.
24+
25+
`QueueStore` owns internal enablement alongside pipeline coordination. `QueueSummaryStore` persists the current read model projected from immutable `QueueLogStore` occurrences. Enablement sequences, operational row versions and projection versions are distinct. These stores use primary-key operations; version arithmetic, duplicate reconciliation and projection repair belong to business logic. See [queue state](../../../doc/rfc/stovepipe/queue-state.md).

‎stovepipe/extension/storage/mock/BUILD.bazel‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,9 @@ go_library(
44
name = "go_default_library",
55
srcs = [
66
"build_store_mock.go",
7+
"queue_log_store_mock.go",
78
"queue_store_mock.go",
9+
"queue_summary_store_mock.go",
810
"request_acceptance_store_mock.go",
911
"request_log_store_mock.go",
1012
"request_store_mock.go",

‎stovepipe/extension/storage/mock/queue_log_store_mock.go‎

Lines changed: 71 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

0 commit comments

Comments
 (0)