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
6 changes: 5 additions & 1 deletion stovepipe/entity/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -5,14 +5,18 @@ go_library(
srcs = [
"queue.go",
"request.go",
"request_id.go",
],
importpath = "github.com/uber/submitqueue/stovepipe/entity",
visibility = ["//visibility:public"],
)

go_test(
name = "go_default_test",
srcs = ["request_test.go"],
srcs = [
"request_id_test.go",
"request_test.go",
],
embed = [":go_default_library"],
deps = [
"@com_github_stretchr_testify//assert:go_default_library",
Expand Down
6 changes: 3 additions & 3 deletions stovepipe/entity/queue.go
Original file line number Diff line number Diff line change
Expand Up @@ -34,9 +34,9 @@ type Queue struct {
// InFlightCount is the number of trunk validations admitted by process but not yet terminal.
InFlightCount int32 `json:"in_flight_count"`

// LatestRequestSeq is the highest request sequence accepted for this queue. Used to coalesce
// backlog so intermediate heads are skipped without superseding the newest head.
LatestRequestSeq int64 `json:"latest_request_seq"`
// LatestRequestID is the request id of the newest head ingest accepted for this queue.
// Empty until the first request is created. Coalescing compares IDs via CompareRequestID.
LatestRequestID string `json:"latest_request_id"`

// 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.
Expand Down
31 changes: 29 additions & 2 deletions stovepipe/entity/request.go
Original file line number Diff line number Diff line change
Expand Up @@ -29,8 +29,25 @@ const (
RequestStateUnknown RequestState = ""
// RequestStateAccepted is the initial state of a request: a new commit has been observed
// for the queue and the request has been admitted into the pipeline, but no validation
// strategy has been chosen yet. Later stages (process, build, record) add their own states.
// strategy has been chosen yet.
RequestStateAccepted RequestState = "accepted"
// RequestStateProcessing means process admitted the request, recorded build strategy and
// baseline, and published to build.
RequestStateProcessing RequestState = "processing"
// RequestStateSuperseded means process skipped the request because a newer head exists.
RequestStateSuperseded RequestState = "superseded"
)

// BuildStrategy defines how build validates the request's commit.
type BuildStrategy string

const (
// BuildStrategyUnknown is the zero value before process chooses a strategy.
BuildStrategyUnknown BuildStrategy = ""
// BuildStrategyIncrementalSinceGreen validates only the delta since the pinned baseline URI.
BuildStrategyIncrementalSinceGreen BuildStrategy = "incremental_since_green"
// BuildStrategyFull validates the whole repo from scratch.
BuildStrategyFull BuildStrategy = "full"
)

// Request represents a single validation of a queue at a particular commit. The queue reports
Expand All @@ -48,9 +65,19 @@ type Request struct {
// and is the stable handle the ingest caller supplies.
Queue string `json:"queue"`
// URI is the opaque, VCS-agnostic locator of the commit under validation, as produced by the
// SourceControl extension. It is empty until SourceControl resolution is wired in.
// SourceControl extension.
URI string `json:"uri"`

// ****************
// Set once at process admit — empty until accepted→processing; never overwritten after
// ****************

// BuildStrategy is the validation scope process chose when admitting this request.
BuildStrategy BuildStrategy `json:"build_strategy"`
// BaseURI is the base URI to be used for this request when the strategy is incremental.
// Empty for full builds and cold start.
BaseURI string `json:"base_uri"`
Comment thread
mnoah1 marked this conversation as resolved.

// ****************
// Following fields could be changed throughout the lifecycle of the request
// ****************
Expand Down
60 changes: 60 additions & 0 deletions stovepipe/entity/request_id.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,60 @@
// 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 (
"fmt"
"strconv"
"strings"
)

// CompareRequestID compares ingest order of two request IDs in the same queue.
// Returns -1 if a is older than b, 0 if equal, 1 if a is newer than b.
// IDs must follow the format request/<queue>/<counter>.
func CompareRequestID(queue, a, b string) (int, error) {
if a == b {
return 0, nil
}
aCounter, err := requestCounter(a, queue)
if err != nil {
return 0, err
}
bCounter, err := requestCounter(b, queue)
if err != nil {
return 0, err
}
switch {
case aCounter < bCounter:
return -1, nil
case aCounter > bCounter:
return 1, nil
default:
return 0, nil
}
}

// requestIDPrefix returns the expected prefix for request ids in queue: "request/<queue>/".
func requestIDPrefix(queue string) string {
return "request/" + queue + "/"
}

// requestCounter extracts the per-queue counter suffix from a request id.
func requestCounter(id, queue string) (int64, error) {
prefix := requestIDPrefix(queue)
if !strings.HasPrefix(id, prefix) {
return 0, fmt.Errorf("request id %q does not match queue %q", id, queue)
}
return strconv.ParseInt(id[len(prefix):], 10, 64)
}
Comment thread
mnoah1 marked this conversation as resolved.
77 changes: 77 additions & 0 deletions stovepipe/entity/request_id_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,77 @@
// 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 TestCompareRequestID(t *testing.T) {
const queue = "monorepo/main"

tests := []struct {
name string
a string
b string
want int
wantErr bool
}{
{
name: "older vs newer",
a: "request/monorepo/main/7",
b: "request/monorepo/main/10",
want: -1,
},
{
name: "newer vs older",
a: "request/monorepo/main/10",
b: "request/monorepo/main/7",
want: 1,
},
{
name: "equal",
a: "request/monorepo/main/42",
b: "request/monorepo/main/42",
want: 0,
},
{
name: "numeric not lexicographic",
a: "request/monorepo/main/9",
b: "request/monorepo/main/10",
want: -1,
},
{
name: "wrong queue prefix",
a: "request/other/1",
b: "request/monorepo/main/2",
wantErr: true,
},
}

for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got, err := CompareRequestID(queue, tt.a, tt.b)
if tt.wantErr {
require.Error(t, err)
return
}
require.NoError(t, err)
assert.Equal(t, tt.want, got)
})
}
}
14 changes: 8 additions & 6 deletions stovepipe/entity/request_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -37,13 +37,15 @@ func TestRequest_SerializationRoundTrip(t *testing.T) {
},
},
{
name: "uri not yet resolved",
name: "processing with strategy and baseline",
req: Request{
ID: "request/monorepo/main/101",
Queue: "monorepo/main",
URI: "",
State: RequestStateAccepted,
Version: 1,
ID: "request/monorepo/main/101",
Queue: "monorepo/main",
URI: "git://remote/monorepo/main/bbbb2222",
State: RequestStateProcessing,
BuildStrategy: BuildStrategyIncrementalSinceGreen,
BaseURI: "git://remote/monorepo/main/green-aaaa",
Version: 2,
},
},
}
Expand Down
12 changes: 6 additions & 6 deletions stovepipe/extension/storage/mysql/queue_store.go
Original file line number Diff line number Diff line change
Expand Up @@ -42,12 +42,12 @@ func (q *queueStore) Create(ctx context.Context, queue entity.Queue) (retErr err
defer func() { op.Complete(retErr) }()

_, err := q.db.ExecContext(ctx,
`INSERT INTO queue (name, last_green_uri, in_flight_count, latest_request_seq, version)
`INSERT INTO queue (name, last_green_uri, in_flight_count, latest_request_id, version)
VALUES (?, ?, ?, ?, ?)`,
queue.Name,
queue.LastGreenURI,
queue.InFlightCount,
queue.LatestRequestSeq,
queue.LatestRequestID,
queue.Version,
)
if err != nil {
Expand All @@ -66,13 +66,13 @@ func (q *queueStore) Get(ctx context.Context, name string) (ret entity.Queue, re

var queue entity.Queue
err := q.db.QueryRowContext(ctx,
"SELECT name, last_green_uri, in_flight_count, latest_request_seq, version FROM queue WHERE name = ?",
"SELECT name, last_green_uri, in_flight_count, latest_request_id, version FROM queue WHERE name = ?",
name,
).Scan(
&queue.Name,
&queue.LastGreenURI,
&queue.InFlightCount,
&queue.LatestRequestSeq,
&queue.LatestRequestID,
&queue.Version,
)

Expand All @@ -94,11 +94,11 @@ func (q *queueStore) Update(ctx context.Context, queue entity.Queue, oldVersion,

result, err := q.db.ExecContext(ctx,
`UPDATE queue
SET last_green_uri = ?, in_flight_count = ?, latest_request_seq = ?, version = ?
SET last_green_uri = ?, in_flight_count = ?, latest_request_id = ?, version = ?
WHERE name = ? AND version = ?`,
queue.LastGreenURI,
queue.InFlightCount,
queue.LatestRequestSeq,
queue.LatestRequestID,
newVersion,
queue.Name,
oldVersion,
Expand Down
38 changes: 31 additions & 7 deletions stovepipe/extension/storage/mysql/request_store.go
Original file line number Diff line number Diff line change
Expand Up @@ -48,8 +48,15 @@ func (r *requestStore) Create(ctx context.Context, request entity.Request) (retE
defer func() { op.Complete(retErr) }()

_, err := r.db.ExecContext(ctx,
"INSERT INTO request (id, queue, uri, state, version) VALUES (?, ?, ?, ?, ?)",
request.ID, request.Queue, request.URI, request.State, request.Version,
`INSERT INTO request (id, queue, uri, state, build_strategy, base_uri, version)
VALUES (?, ?, ?, ?, ?, ?, ?)`,
request.ID,
request.Queue,
request.URI,
request.State,
request.BuildStrategy,
request.BaseURI,
request.Version,
)
if err != nil {
if isDuplicateEntry(err) {
Expand All @@ -68,9 +75,18 @@ func (r *requestStore) Get(ctx context.Context, id string) (ret entity.Request,

var req entity.Request
err := r.db.QueryRowContext(ctx,
"SELECT id, queue, uri, state, version FROM request WHERE id = ?",
`SELECT id, queue, uri, state, build_strategy, base_uri, version
FROM request WHERE id = ?`,
id,
).Scan(&req.ID, &req.Queue, &req.URI, &req.State, &req.Version)
).Scan(
&req.ID,
&req.Queue,
&req.URI,
&req.State,
&req.BuildStrategy,
&req.BaseURI,
&req.Version,
)

if errors.Is(err, sql.ErrNoRows) {
return entity.Request{}, storage.WrapNotFound(err)
Expand All @@ -82,7 +98,7 @@ func (r *requestStore) Get(ctx context.Context, id string) (ret entity.Request,
return req, nil
}

// Update persists the mutable fields of request (uri, state) if the stored version matches
// Update persists the mutable fields of request (uri, state, build_strategy, base_uri) if the
// oldVersion, writing newVersion. Returns ErrVersionMismatch if the stored version does not match
// (including when the request does not exist). This is a pure conditional write; the caller owns
// version arithmetic.
Expand All @@ -91,8 +107,16 @@ func (r *requestStore) Update(ctx context.Context, request entity.Request, oldVe
defer func() { op.Complete(retErr) }()

result, err := r.db.ExecContext(ctx,
"UPDATE request SET uri = ?, state = ?, version = ? WHERE id = ? AND version = ?",
request.URI, request.State, newVersion, request.ID, oldVersion,
`UPDATE request
SET uri = ?, state = ?, build_strategy = ?, base_uri = ?, version = ?
WHERE id = ? AND version = ?`,
request.URI,
request.State,
request.BuildStrategy,
request.BaseURI,
newVersion,
request.ID,
oldVersion,
)
if err != nil {
return fmt.Errorf(
Expand Down
4 changes: 2 additions & 2 deletions stovepipe/extension/storage/mysql/schema/queue.sql
Original file line number Diff line number Diff line change
@@ -1,10 +1,10 @@
-- queue holds per-queue coordination state for the validation pipeline: the last-green
-- bookmark, in-flight gate count, and latest-request sequence pointer.
-- bookmark, in-flight gate count, and latest-request id pointer.
CREATE TABLE IF NOT EXISTS queue (
name VARCHAR(255) NOT NULL,
last_green_uri VARCHAR(255) NOT NULL DEFAULT '',
in_flight_count INT NOT NULL DEFAULT 0,
latest_request_seq BIGINT NOT NULL DEFAULT 0,
latest_request_id VARCHAR(255) NOT NULL DEFAULT '',
version INT NOT NULL,
PRIMARY KEY (name)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
12 changes: 7 additions & 5 deletions stovepipe/extension/storage/mysql/schema/request.sql
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,12 @@
-- VCS-agnostic commit locator; it may be empty until SourceControl resolution is wired in.
-- No timestamps: created/updated times are not part of the Request entity.
CREATE TABLE IF NOT EXISTS request (
id VARCHAR(255) NOT NULL,
queue VARCHAR(255) NOT NULL,
uri VARCHAR(255) NOT NULL,
state VARCHAR(64) NOT NULL,
version INT NOT NULL,
id VARCHAR(255) NOT NULL,
queue VARCHAR(255) NOT NULL,
uri VARCHAR(255) NOT NULL,
state VARCHAR(64) NOT NULL,
build_strategy VARCHAR(64) NOT NULL DEFAULT '',
base_uri VARCHAR(255) NOT NULL DEFAULT '',
version INT NOT NULL,
PRIMARY KEY (id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
Loading
Loading