Skip to content

Commit 1f6af7e

Browse files
authored
feat(stovepipe): store queue state and read models (#855)
## Why? Queue enablement needs durable internal state and separate read models so orchestration updates do not alter enablement history. ## What? Extend the existing queue snapshot, add immutable QueueLog and derived QueueSummary records, and implement key-oriented stores with conditional writes. Add queue and read-model schemas, with backend tests for byte-exact queue identity and independent projection versions. ## Test Plan Run the Stovepipe storage unit and MySQL integration tests. ## Revert Plan Revert this layer together with any dependent layers in the stack. ## Issue Links None — this is a standalone queue-status proposal. ## Stack Review in this order; each PR targets the preceding branch. 1. [API and RFC](#854) 2. [Storage](#855) (this PR) 3. [Transitions and materialization](#856) 4. [Lookup](#857)
1 parent d0e2c42 commit 1f6af7e

28 files changed

Lines changed: 1047 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: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,25 @@ 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+
func (q Queue) EnablementState() QueueState {
65+
return QueueState{Queue: q.Name, State: q.Enablement, Version: q.EnablementVersion,
66+
EffectiveFromCommitURI: q.EffectiveFromCommitURI, ChangedAtMs: q.EnablementChangedAtMs}
67+
}

‎stovepipe/entity/queue_log.go‎

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
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+
func (l QueueLog) Snapshot() QueueState {
41+
return QueueState{Queue: l.Queue, State: l.State, Version: l.QueueStateVersion,
42+
EffectiveFromCommitURI: l.EffectiveFromCommitURI, ChangedAtMs: l.TimestampMs}
43+
}
44+
45+
func (l QueueLog) Validate() error {
46+
if l.ID != QueueLogID(l.QueueStateVersion) {
47+
return fmt.Errorf("queue state log identity differs from its state version")
48+
}
49+
return l.Snapshot().Validate()
50+
}

‎stovepipe/entity/queue_state.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 "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 QueueEnablement = "enabled"
26+
QueueEnablementDisabled QueueEnablement = "disabled"
27+
)
28+
29+
// QueueState describes applied enablement, independent of storage row versions.
30+
type QueueState struct {
31+
// Queue is the stable repo-and-ref namespace, at most 255 bytes.
32+
Queue string
33+
// State is enabled or disabled.
34+
State QueueEnablement
35+
// EffectiveFromCommitURI is the inclusive immutable boundary, at most 255 bytes; empty only at version 1.
36+
EffectiveFromCommitURI string
37+
// ChangedAtMs is the application timestamp in Unix milliseconds.
38+
ChangedAtMs int64
39+
// Version is the positive queue-scoped enablement sequence.
40+
Version int64
41+
}
42+
43+
func (s QueueState) Validate() error {
44+
if s.Queue == "" || len(s.Queue) > 255 || s.Version <= 0 || s.ChangedAtMs <= 0 {
45+
return fmt.Errorf("invalid queue state identity, version or timestamp")
46+
}
47+
if s.State != QueueEnablementEnabled && s.State != QueueEnablementDisabled {
48+
return fmt.Errorf("invalid queue enablement %q", s.State)
49+
}
50+
if s.Version == 1 {
51+
if s.State != QueueEnablementDisabled || s.EffectiveFromCommitURI != "" {
52+
return fmt.Errorf("initial queue state must be disabled without a boundary")
53+
}
54+
} else if s.EffectiveFromCommitURI == "" || len(s.EffectiveFromCommitURI) > 255 {
55+
return fmt.Errorf("invalid queue state commit boundary")
56+
}
57+
return nil
58+
}
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: 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 "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+
func (s QueueSummary) Snapshot() QueueState {
36+
return QueueState{Queue: s.Queue, State: s.State, Version: s.QueueStateVersion,
37+
EffectiveFromCommitURI: s.EffectiveFromCommitURI, ChangedAtMs: s.StateTimestampMs}
38+
}
39+
40+
func (s QueueSummary) Validate() error {
41+
if s.Version <= 0 {
42+
return fmt.Errorf("invalid queue summary version")
43+
}
44+
return s.Snapshot().Validate()
45+
}

‎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)