Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions stovepipe/entity/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,9 @@ go_library(
"project_status.go",
"queue.go",
"queue_config.go",
"queue_log.go",
"queue_state.go",
"queue_summary.go",
"request.go",
"request_acceptance.go",
"request_history.go",
Expand All @@ -26,6 +29,7 @@ go_test(
name = "go_default_test",
srcs = [
"build_test.go",
"queue_state_test.go",
"request_id_test.go",
"request_log_test.go",
"request_test.go",
Expand Down
18 changes: 18 additions & 0 deletions stovepipe/entity/queue.go
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,25 @@ type Queue struct {
// Empty until the first request is created. Coalescing compares IDs via CompareRequestID.
LatestRequestID string `json:"latest_request_id"`

// Enablement is the applied enabled/disabled state; empty means unregistered.
Enablement QueueEnablement `json:"enablement"`

// EnablementVersion is the queue-scoped enablement sequence; zero means unregistered.
// Operational changes do not advance this sequence.
EnablementVersion int64 `json:"enablement_version"`

// EffectiveFromCommitURI is the inclusive immutable enablement boundary; empty for initial registration.
EffectiveFromCommitURI string `json:"effective_from_commit_uri"`

// EnablementChangedAtMs is the enablement application timestamp in Unix milliseconds.
EnablementChangedAtMs int64 `json:"enablement_changed_at_ms"`

// Version is the version of the object. It is used for optimistic locking.
// Versioning starts at 1 and is incremented for each change to the object.
Version int32 `json:"version"`
}

func (q Queue) EnablementState() QueueState {
return QueueState{Queue: q.Name, State: q.Enablement, Version: q.EnablementVersion,
EffectiveFromCommitURI: q.EffectiveFromCommitURI, ChangedAtMs: q.EnablementChangedAtMs}
}
50 changes: 50 additions & 0 deletions stovepipe/entity/queue_log.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,50 @@
// Copyright (c) 2026 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 entity

import "fmt"

// QueueLog is one immutable occurrence of a committed enablement state.
type QueueLog struct {
// ID is the stable queue-scoped occurrence identity derived from QueueStateVersion.
ID string
// Queue is the stable repo-and-ref namespace.
Queue string
// TimestampMs is the application's timestamp in Unix milliseconds.
TimestampMs int64
// State is the committed enabled or disabled value.
State QueueEnablement
// QueueStateVersion is the positive version identifying the applied state.
QueueStateVersion int64
// EffectiveFromCommitURI is the inclusive immutable boundary; empty only at version 1.
EffectiveFromCommitURI string
}

// QueueLogID returns the stable occurrence identity for a state version.
func QueueLogID(version int64) string {
return fmt.Sprintf("state/%d", version)
}

func (l QueueLog) Snapshot() QueueState {
return QueueState{Queue: l.Queue, State: l.State, Version: l.QueueStateVersion,
EffectiveFromCommitURI: l.EffectiveFromCommitURI, ChangedAtMs: l.TimestampMs}
}

func (l QueueLog) Validate() error {
if l.ID != QueueLogID(l.QueueStateVersion) {
return fmt.Errorf("queue state log identity differs from its state version")
}
return l.Snapshot().Validate()
}
58 changes: 58 additions & 0 deletions stovepipe/entity/queue_state.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,58 @@
// Copyright (c) 2026 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 entity

import "fmt"

// QueueEnablement is applied enablement, independent of execution controls and verdicts.
type QueueEnablement string

const (
// QueueEnablementUnknown is unset and never an explicit disablement.
QueueEnablementUnknown QueueEnablement = ""
QueueEnablementEnabled QueueEnablement = "enabled"
QueueEnablementDisabled QueueEnablement = "disabled"
)

// QueueState describes applied enablement, independent of storage row versions.
type QueueState struct {
// Queue is the stable repo-and-ref namespace, at most 255 bytes.
Queue string
// State is enabled or disabled.
State QueueEnablement
// EffectiveFromCommitURI is the inclusive immutable boundary, at most 255 bytes; empty only at version 1.
EffectiveFromCommitURI string
// ChangedAtMs is the application timestamp in Unix milliseconds.
ChangedAtMs int64
// Version is the positive queue-scoped enablement sequence.
Version int64
}

func (s QueueState) Validate() error {
if s.Queue == "" || len(s.Queue) > 255 || s.Version <= 0 || s.ChangedAtMs <= 0 {
return fmt.Errorf("invalid queue state identity, version or timestamp")
}
if s.State != QueueEnablementEnabled && s.State != QueueEnablementDisabled {
return fmt.Errorf("invalid queue enablement %q", s.State)
}
if s.Version == 1 {
if s.State != QueueEnablementDisabled || s.EffectiveFromCommitURI != "" {
return fmt.Errorf("initial queue state must be disabled without a boundary")
}
} else if s.EffectiveFromCommitURI == "" || len(s.EffectiveFromCommitURI) > 255 {
return fmt.Errorf("invalid queue state commit boundary")
}
return nil
}
58 changes: 58 additions & 0 deletions stovepipe/entity/queue_state_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,58 @@
// Copyright (c) 2026 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 entity

import (
"strings"
"testing"

"github.com/stretchr/testify/require"
)

func TestQueueStateInvariants(t *testing.T) {
initial := QueueState{Queue: "repo/main", State: QueueEnablementDisabled, ChangedAtMs: 1000, Version: 1}
for _, tc := range []struct {
name string
change func(*QueueState)
valid bool
}{
{"initial", func(*QueueState) {}, true},
{"enabled occurrence", func(s *QueueState) { s.Version = 2; s.State = QueueEnablementEnabled; s.EffectiveFromCommitURI = "A" }, true},
{"missing queue", func(s *QueueState) { s.Queue = "" }, false},
{"oversized queue", func(s *QueueState) { s.Queue = strings.Repeat("x", 256) }, false},
{"unknown state", func(s *QueueState) { s.State = QueueEnablementUnknown }, false},
{"invalid version", func(s *QueueState) { s.Version = 0 }, false},
{"missing timestamp", func(s *QueueState) { s.ChangedAtMs = 0 }, false},
{"enabled initial", func(s *QueueState) { s.State = QueueEnablementEnabled }, false},
{"initial boundary", func(s *QueueState) { s.EffectiveFromCommitURI = "A" }, false},
{"missing later boundary", func(s *QueueState) { s.Version = 2 }, false},
{"oversized boundary", func(s *QueueState) { s.Version = 2; s.EffectiveFromCommitURI = strings.Repeat("x", 256) }, false},
} {
t.Run(tc.name, func(t *testing.T) {
state := initial
tc.change(&state)
if tc.valid {
require.NoError(t, state.Validate())
} else {
require.Error(t, state.Validate())
}
})
}
log := QueueLog{ID: QueueLogID(1), Queue: initial.Queue, TimestampMs: initial.ChangedAtMs, State: initial.State, QueueStateVersion: 1}
require.NoError(t, log.Validate())
require.Equal(t, initial, log.Snapshot())
log.ID = "other"
require.Error(t, log.Validate())
}
45 changes: 45 additions & 0 deletions stovepipe/entity/queue_summary.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,45 @@
// Copyright (c) 2026 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 entity

import "fmt"

// QueueSummary is the current read model derived from retained queue history.
type QueueSummary struct {
// Queue is the stable repo-and-ref namespace, at most 255 bytes.
Queue string
// State is the latest materialized enabled/disabled value.
State QueueEnablement
// QueueStateVersion identifies the applied enablement occurrence represented here.
QueueStateVersion int64
// EffectiveFromCommitURI is the inclusive immutable boundary; empty at state version 1.
EffectiveFromCommitURI string
// StateTimestampMs is the occurrence timestamp in Unix milliseconds.
StateTimestampMs int64
// Version is the positive optimistic-lock version of this projection.
Version int64
}

func (s QueueSummary) Snapshot() QueueState {
return QueueState{Queue: s.Queue, State: s.State, Version: s.QueueStateVersion,
EffectiveFromCommitURI: s.EffectiveFromCommitURI, ChangedAtMs: s.StateTimestampMs}
}

func (s QueueSummary) Validate() error {
if s.Version <= 0 {
return fmt.Errorf("invalid queue summary version")
}
return s.Snapshot().Validate()
}
2 changes: 2 additions & 0 deletions stovepipe/extension/storage/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,9 @@ go_library(
name = "go_default_library",
srcs = [
"build_store.go",
"queue_log_store.go",
"queue_store.go",
"queue_summary_store.go",
"request_acceptance_store.go",
"request_log_store.go",
"request_store.go",
Expand Down
2 changes: 2 additions & 0 deletions stovepipe/extension/storage/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -21,3 +21,5 @@ A `Get` immediately following a successful write (`Create`/`Update`) — by the
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`.

`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.

`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).
2 changes: 2 additions & 0 deletions stovepipe/extension/storage/mock/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,9 @@ go_library(
name = "go_default_library",
srcs = [
"build_store_mock.go",
"queue_log_store_mock.go",
"queue_store_mock.go",
"queue_summary_store_mock.go",
"request_acceptance_store_mock.go",
"request_log_store_mock.go",
"request_store_mock.go",
Expand Down
71 changes: 71 additions & 0 deletions stovepipe/extension/storage/mock/queue_log_store_mock.go

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Loading
Loading