diff --git a/submitqueue/entity/BUILD.bazel b/submitqueue/entity/BUILD.bazel index 2ffad326..964b7ee9 100644 --- a/submitqueue/entity/BUILD.bazel +++ b/submitqueue/entity/BUILD.bazel @@ -20,6 +20,7 @@ go_library( "request_history.go", "request_log.go", "request_summary.go", + "speculation.go", ], importpath = "github.com/uber/submitqueue/submitqueue/entity", visibility = ["//visibility:public"], @@ -38,6 +39,7 @@ go_test( "land_test.go", "request_log_test.go", "request_test.go", + "speculation_test.go", ], embed = [":go_default_library"], deps = [ diff --git a/submitqueue/entity/speculation.go b/submitqueue/entity/speculation.go new file mode 100644 index 00000000..5f55d77b --- /dev/null +++ b/submitqueue/entity/speculation.go @@ -0,0 +1,229 @@ +// 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 entity + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "strings" +) + +// DependencyBetType is how a path treats one of its head's dependencies. +type DependencyBetType string + +const ( + // BetUnknown is the zero-value sentinel; it is never a valid bet. + BetUnknown DependencyBetType = "" + // BetIncluded bets the dependency lands: the head is built on top of it, and + // the path is invalidated if the dependency ultimately fails. + BetIncluded DependencyBetType = "included" + // BetExcluded bets the dependency does not land: the head is built without it, + // and the path is invalidated if the dependency ultimately lands. + BetExcluded DependencyBetType = "excluded" + // BetDropped marks a dependency ignored by conflict relaxation: whether it + // lands or fails never affects the path — it neither gates the merge nor + // invalidates the path. + BetDropped DependencyBetType = "dropped" +) + +// DependencyBet is a path's bet on one dependency of its head. +type DependencyBet struct { + // Batch is the dependency batch ID this bet is about. + Batch string + // Bet is how the path treats the dependency (included, excluded, or dropped). + Bet DependencyBetType +} + +// SpeculationPath is one guess at how a batch's dependencies resolve: a head +// batch plus a bet on each of its dependencies. Every dependency of the head +// appears exactly once, in queue order, so the path is self-describing — its +// full meaning can be read without consulting any external relaxed set or +// dependency list. +type SpeculationPath struct { + // Head is the batch being built along this path. + Head string + // Bets is one bet per dependency of Head, in queue order. + Bets []DependencyBet +} + +// ID returns the path's stable identity: a hex-encoded SHA-256 over the head +// and its bets in order. Two paths with the same head and the same ordered +// bets share an ID; any difference in head, dependency, or bet yields a +// different ID. +func (p SpeculationPath) ID() string { + var b strings.Builder + b.WriteString(p.Head) + for _, bet := range p.Bets { + b.WriteByte('\n') + b.WriteString(bet.Batch) + b.WriteByte('=') + b.WriteString(string(bet.Bet)) + } + sum := sha256.Sum256([]byte(b.String())) + return hex.EncodeToString(sum[:]) +} + +// ToBytes serializes the SpeculationPath to JSON bytes for queue message payload. +func (p SpeculationPath) ToBytes() ([]byte, error) { + return json.Marshal(p) +} + +// SpeculationPathFromBytes deserializes a SpeculationPath from JSON bytes. +func SpeculationPathFromBytes(data []byte) (SpeculationPath, error) { + var path SpeculationPath + err := json.Unmarshal(data, &path) + return path, err +} + +// SpeculationPathStatus is the lifecycle status of one speculation path's +// current build attempt. +type SpeculationPathStatus string + +const ( + // SpeculationPathStatusUnknown is the zero-value sentinel; it should never be + // seen in the system. + SpeculationPathStatusUnknown SpeculationPathStatus = "" + // SpeculationPathStatusPending indicates the path is funded under the build + // budget but its build has not started yet. + SpeculationPathStatusPending SpeculationPathStatus = "pending" + // SpeculationPathStatusBuilding indicates the path's build is running in the + // build system. + SpeculationPathStatusBuilding SpeculationPathStatus = "building" + // SpeculationPathStatusPassed indicates the path's build completed + // successfully. This is a terminal state. + SpeculationPathStatusPassed SpeculationPathStatus = "passed" + // SpeculationPathStatusFailed indicates the path's build did not complete + // successfully. This is a terminal state. + SpeculationPathStatusFailed SpeculationPathStatus = "failed" + // SpeculationPathStatusCancelling is the non-terminal intent state set when + // the path's build is being cancelled. The build holds its slot until it + // reaches a terminal state. + SpeculationPathStatusCancelling SpeculationPathStatus = "cancelling" + // SpeculationPathStatusCancelled indicates the path's build was cancelled. + // This is a terminal state. + SpeculationPathStatusCancelled SpeculationPathStatus = "cancelled" +) + +// IsTerminal returns true if the status represents a final state +// (Passed, Failed, or Cancelled). Cancelling is intentionally excluded: it is +// a non-terminal intent, and the build may still reach Passed or Failed before +// it reaches Cancelled. +func (s SpeculationPathStatus) IsTerminal() bool { + switch s { + case SpeculationPathStatusPassed, SpeculationPathStatusFailed, SpeculationPathStatusCancelled: + return true + default: + return false + } +} + +// SpeculationPathEntry is the stored record of one chosen speculation path, +// keyed by the hash of its content. It holds no build reference (that lives on +// the separate execution record, keyed by (ID, Attempt)) and no score (a score +// is meaningful only within a single speculation run). +type SpeculationPathEntry struct { + // ID is the primary key: the hash of the path's content (head plus its + // bets). It equals Path.ID(). + ID string + // Path is the head plus one bet per dependency, in queue order. + Path SpeculationPath + // Status is the lifecycle status of the current build attempt. + Status SpeculationPathStatus + // Attempt is the build attempt number for this path, starting at 1. A path + // can be built more than once (e.g. after a prior build is cancelled to free + // budget, or fails and is retried), so ID alone does not identify an + // execution — (ID, Attempt) does. It increments with each new build. + Attempt int + // 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 + // CreatedAtMs is the creation time in Unix epoch milliseconds. + CreatedAtMs int64 + // UpdatedAtMs is the last-update time in Unix epoch milliseconds. + UpdatedAtMs int64 +} + +// ToBytes serializes the SpeculationPathEntry to JSON bytes for queue message payload. +func (e SpeculationPathEntry) ToBytes() ([]byte, error) { + return json.Marshal(e) +} + +// SpeculationPathEntryFromBytes deserializes a SpeculationPathEntry from JSON bytes. +func SpeculationPathEntryFromBytes(data []byte) (SpeculationPathEntry, error) { + var entry SpeculationPathEntry + err := json.Unmarshal(data, &entry) + return entry, err +} + +// SpeculationPathSet is one head's chosen speculation paths under a single +// version. It holds both live paths and recently finished ones — finished +// entries linger briefly so that a re-run cannot collide with an old build. +// Every path in the set shares the same head and bets over the same ordered +// dependency list. +type SpeculationPathSet struct { + // BatchID is the primary key: the head batch these paths speculate on. + BatchID string + // Paths is the head's chosen paths, live and recently finished. + Paths []SpeculationPathEntry + // 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 +} + +// ToBytes serializes the SpeculationPathSet to JSON bytes for queue message payload. +func (s SpeculationPathSet) ToBytes() ([]byte, error) { + return json.Marshal(s) +} + +// SpeculationPathSetFromBytes deserializes a SpeculationPathSet from JSON bytes. +func SpeculationPathSetFromBytes(data []byte) (SpeculationPathSet, error) { + var set SpeculationPathSet + err := json.Unmarshal(data, &set) + return set, err +} + +// PathAction is an action proposed on a speculation path. The set is limited to +// build and cancel; there is no merge or fail action, because a batch's verdict +// is a controller-owned fact, not a proposed action. +type PathAction string + +const ( + // PathActionUnknown is the zero-value sentinel; it is never a valid action. + PathActionUnknown PathAction = "" + // PathActionBuild proposes starting (or resurrecting) a build for the path. + PathActionBuild PathAction = "build" + // PathActionCancel proposes preempting an in-flight path to free build budget. + PathActionCancel PathAction = "cancel" +) + +// Speculation is one proposed action on one path. A path left as-is has no +// Speculation. +type Speculation struct { + // Path is the path the action applies to; its ID hashes the head and its bets. + Path SpeculationPath + // Action is the proposed action (build or cancel). + Action PathAction +} + +// CandidatePath is a path paired with the transient ranking score assigned to it +// within a single speculation run. The ranking score orders candidates for that +// run only and is never stored — rankings go stale across runs. +type CandidatePath struct { + // Path is the candidate: a head plus one bet per dependency. + Path SpeculationPath + // RankingScore is the ordering score assigned this run; higher sorts first. + RankingScore float64 +} diff --git a/submitqueue/entity/speculation_test.go b/submitqueue/entity/speculation_test.go new file mode 100644 index 00000000..1789e0b5 --- /dev/null +++ b/submitqueue/entity/speculation_test.go @@ -0,0 +1,321 @@ +// 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 entity + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestSpeculationPath_ID_Deterministic(t *testing.T) { + path := SpeculationPath{ + Head: "queueA/batch/3", + Bets: []DependencyBet{ + {Batch: "queueA/batch/1", Bet: BetIncluded}, + {Batch: "queueA/batch/2", Bet: BetExcluded}, + }, + } + + // Same content produces the same ID across separate value instances. + same := SpeculationPath{ + Head: "queueA/batch/3", + Bets: []DependencyBet{ + {Batch: "queueA/batch/1", Bet: BetIncluded}, + {Batch: "queueA/batch/2", Bet: BetExcluded}, + }, + } + + assert.Equal(t, path.ID(), same.ID()) + assert.NotEmpty(t, path.ID()) +} + +func TestSpeculationPath_ID_Sensitivity(t *testing.T) { + base := SpeculationPath{ + Head: "queueA/batch/3", + Bets: []DependencyBet{ + {Batch: "queueA/batch/1", Bet: BetIncluded}, + {Batch: "queueA/batch/2", Bet: BetIncluded}, + }, + } + + tests := []struct { + name string + path SpeculationPath + }{ + { + name: "different head", + path: SpeculationPath{ + Head: "queueA/batch/4", + Bets: base.Bets, + }, + }, + { + name: "different bet type", + path: SpeculationPath{ + Head: base.Head, + Bets: []DependencyBet{ + {Batch: "queueA/batch/1", Bet: BetIncluded}, + {Batch: "queueA/batch/2", Bet: BetExcluded}, + }, + }, + }, + { + name: "different dependency", + path: SpeculationPath{ + Head: base.Head, + Bets: []DependencyBet{ + {Batch: "queueA/batch/1", Bet: BetIncluded}, + {Batch: "queueA/batch/9", Bet: BetIncluded}, + }, + }, + }, + { + name: "different bet order", + path: SpeculationPath{ + Head: base.Head, + Bets: []DependencyBet{ + {Batch: "queueA/batch/2", Bet: BetIncluded}, + {Batch: "queueA/batch/1", Bet: BetIncluded}, + }, + }, + }, + { + name: "no bets", + path: SpeculationPath{ + Head: base.Head, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.NotEqual(t, base.ID(), tt.path.ID()) + }) + } +} + +func TestSpeculationPath_SerializationRoundTrip(t *testing.T) { + tests := []struct { + name string + path SpeculationPath + }{ + { + name: "path with mixed bets", + path: SpeculationPath{ + Head: "queueA/batch/3", + Bets: []DependencyBet{ + {Batch: "queueA/batch/1", Bet: BetIncluded}, + {Batch: "queueA/batch/2", Bet: BetDropped}, + }, + }, + }, + { + name: "head with no dependencies", + path: SpeculationPath{ + Head: "queueA/batch/1", + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + data, err := tt.path.ToBytes() + require.NoError(t, err) + + deserialized, err := SpeculationPathFromBytes(data) + require.NoError(t, err) + + assert.Equal(t, tt.path, deserialized) + }) + } +} + +func TestSpeculationPathFromBytes_InvalidJSON(t *testing.T) { + _, err := SpeculationPathFromBytes([]byte(`{"invalid": json"}`)) + assert.Error(t, err) +} + +func TestSpeculationPathFromBytes_EmptyData(t *testing.T) { + path, err := SpeculationPathFromBytes([]byte(`{}`)) + require.NoError(t, err) + + assert.Empty(t, path.Head) + assert.Empty(t, path.Bets) +} + +func TestSpeculationPathStatus_IsTerminal(t *testing.T) { + tests := []struct { + name string + status SpeculationPathStatus + expected bool + }{ + {name: "passed is terminal", status: SpeculationPathStatusPassed, expected: true}, + {name: "failed is terminal", status: SpeculationPathStatusFailed, expected: true}, + {name: "cancelled is terminal", status: SpeculationPathStatusCancelled, expected: true}, + {name: "pending is not terminal", status: SpeculationPathStatusPending, expected: false}, + {name: "building is not terminal", status: SpeculationPathStatusBuilding, expected: false}, + {name: "cancelling is not terminal", status: SpeculationPathStatusCancelling, expected: false}, + {name: "unknown is not terminal", status: SpeculationPathStatusUnknown, expected: false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.expected, tt.status.IsTerminal()) + }) + } +} + +func TestSpeculationPathEntry_IDDerivesFromPath(t *testing.T) { + path := SpeculationPath{ + Head: "queueA/batch/2", + Bets: []DependencyBet{{Batch: "queueA/batch/1", Bet: BetIncluded}}, + } + + // A well-formed entry keys itself by its path's content hash. + entry := SpeculationPathEntry{ID: path.ID(), Path: path} + assert.Equal(t, entry.Path.ID(), entry.ID) + + // A different path (same head, different bet) is a different entry. + other := SpeculationPath{ + Head: path.Head, + Bets: []DependencyBet{{Batch: "queueA/batch/1", Bet: BetExcluded}}, + } + assert.NotEqual(t, entry.ID, other.ID()) +} + +func TestSpeculationPathEntry_SerializationRoundTrip(t *testing.T) { + tests := []struct { + name string + entry SpeculationPathEntry + }{ + { + name: "pending entry", + entry: SpeculationPathEntry{ + ID: "abc123", + Path: SpeculationPath{ + Head: "queueA/batch/2", + Bets: []DependencyBet{{Batch: "queueA/batch/1", Bet: BetIncluded}}, + }, + Status: SpeculationPathStatusPending, + Attempt: 1, + Version: 1, + CreatedAtMs: 1000, + UpdatedAtMs: 2000, + }, + }, + { + name: "passed entry with no bets", + entry: SpeculationPathEntry{ + ID: "def456", + Path: SpeculationPath{Head: "queueA/batch/1"}, + Status: SpeculationPathStatusPassed, + Attempt: 3, + Version: 5, + CreatedAtMs: 10, + UpdatedAtMs: 20, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + data, err := tt.entry.ToBytes() + require.NoError(t, err) + + deserialized, err := SpeculationPathEntryFromBytes(data) + require.NoError(t, err) + + assert.Equal(t, tt.entry, deserialized) + }) + } +} + +func TestSpeculationPathEntryFromBytes_InvalidJSON(t *testing.T) { + _, err := SpeculationPathEntryFromBytes([]byte(`{"invalid": json"}`)) + assert.Error(t, err) +} + +func TestSpeculationPathEntryFromBytes_EmptyData(t *testing.T) { + entry, err := SpeculationPathEntryFromBytes([]byte(`{}`)) + require.NoError(t, err) + + assert.Empty(t, entry.ID) + assert.Equal(t, SpeculationPathStatusUnknown, entry.Status) + assert.Zero(t, entry.Attempt) + assert.Zero(t, entry.Version) +} + +func TestSpeculationPathSet_SerializationRoundTrip(t *testing.T) { + tests := []struct { + name string + set SpeculationPathSet + }{ + { + name: "set with paths", + set: SpeculationPathSet{ + BatchID: "queueA/batch/2", + Paths: []SpeculationPathEntry{ + { + ID: "abc123", + Path: SpeculationPath{Head: "queueA/batch/2", Bets: []DependencyBet{{Batch: "queueA/batch/1", Bet: BetIncluded}}}, + Status: SpeculationPathStatusPending, + Attempt: 1, + Version: 1, + }, + { + ID: "def456", + Path: SpeculationPath{Head: "queueA/batch/2", Bets: []DependencyBet{{Batch: "queueA/batch/1", Bet: BetExcluded}}}, + Status: SpeculationPathStatusBuilding, + Attempt: 1, + Version: 2, + }, + }, + Version: 3, + }, + }, + { + name: "empty set", + set: SpeculationPathSet{BatchID: "queueA/batch/1", Version: 1}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + data, err := tt.set.ToBytes() + require.NoError(t, err) + + deserialized, err := SpeculationPathSetFromBytes(data) + require.NoError(t, err) + + assert.Equal(t, tt.set, deserialized) + }) + } +} + +func TestSpeculationPathSetFromBytes_InvalidJSON(t *testing.T) { + _, err := SpeculationPathSetFromBytes([]byte(`{"invalid": json"}`)) + assert.Error(t, err) +} + +func TestSpeculationPathSetFromBytes_EmptyData(t *testing.T) { + set, err := SpeculationPathSetFromBytes([]byte(`{}`)) + require.NoError(t, err) + + assert.Empty(t, set.BatchID) + assert.Empty(t, set.Paths) + assert.Zero(t, set.Version) +}