Skip to content
Open
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
2 changes: 1 addition & 1 deletion submitqueue/extension/conflict/tango/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -20,4 +20,4 @@ Each signature is computed against trunk at its own analysis time, without the o

## Injected pieces

The analyzer owns no connection. A deployment supplies the Tango client adapter, which collects the response and resolves each target's label, and a function that returns the commit SHA of the target branch tip a new batch is analyzed against. The changes of a batch come from the changeset resolver, the signature store is bound to one queue, and one signature cache can be shared by every queue's analyzer. The MySQL implementations of the signature and dictionary stores are in `mysql/`.
The analyzer owns no connection. A deployment supplies the Tango client adapter, which collects the response and resolves each target's label, and a function that returns the commit SHA of the target branch tip a new batch is analyzed against. `baserevision.FromGit` builds that function over any `gitworkspace.Factory`. The changes of a batch come from the changeset resolver, the signature store is bound to one queue, and one signature cache can be shared by every queue's analyzer. The MySQL implementations of the signature and dictionary stores are in `mysql/`.
22 changes: 22 additions & 0 deletions submitqueue/extension/conflict/tango/baserevision/BUILD.bazel
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
load("@rules_go//go:def.bzl", "go_library", "go_test")

go_library(
name = "go_default_library",
srcs = ["baserevision.go"],
importpath = "github.com/uber/submitqueue/submitqueue/extension/conflict/tango/baserevision",
visibility = ["//visibility:public"],
deps = ["//platform/extension/gitworkspace:go_default_library"],
)

go_test(
name = "go_default_test",
srcs = ["baserevision_test.go"],
embed = [":go_default_library"],
deps = [
"//platform/extension/gitworkspace:go_default_library",
"//platform/extension/gitworkspace/mock:go_default_library",
"@com_github_stretchr_testify//assert:go_default_library",
"@com_github_stretchr_testify//require:go_default_library",
"@org_uber_go_mock//gomock:go_default_library",
],
)
102 changes: 102 additions & 0 deletions submitqueue/extension/conflict/tango/baserevision/baserevision.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,102 @@
// 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 baserevision resolves the commit a new batch is analyzed against by
// asking a git workspace for the tip of the queue's target branch. Any
// gitworkspace.Factory works, so where git runs is the deployment's choice.
package baserevision

import (
"context"
"fmt"
"regexp"
"strings"
"sync"

"github.com/uber/submitqueue/platform/extension/gitworkspace"
)

const defaultRemote = "origin"

// fullCommitSHA matches the SHA-1 and SHA-256 object formats. Tango needs the
// full SHA, so an abbreviated one is rejected.
var fullCommitSHA = regexp.MustCompile(`^([0-9a-f]{40}|[0-9a-f]{64})$`)

// Target locates the branch whose tip is a queue's base revision.
type Target struct {
// Repo identifies the repository to the workspace factory.
Repo string

// Remote is the git remote the branch is fetched from. It defaults to
// origin.
Remote string

// Branch is the target branch.
Branch string
}

// FromGit returns a function for tango.Params.BaseRevision that fetches the
// target branch of the queue and returns its tip as a full commit SHA, as of
// the call. A queue without a target, or any git failure, is an error. Calls
// that share a Repo are serialized, because concurrent fetches into one clone
// can fail on a ref lock and overwrite each other's FETCH_HEAD.
func FromGit(factory gitworkspace.Factory, targets map[string]Target) func(ctx context.Context, queue string) (string, error) {
locks := make(map[string]*sync.Mutex, len(targets))
for _, target := range targets {
locks[target.Repo] = &sync.Mutex{}
}

return func(ctx context.Context, queue string) (string, error) {
target, ok := targets[queue]
if !ok {
return "", fmt.Errorf("no target branch configured for queue %s", queue)
}
remote := target.Remote
if remote == "" {
remote = defaultRemote
}

lock := locks[target.Repo]
lock.Lock()
defer lock.Unlock()

workspace, err := factory.For(ctx, target.Repo)
if err != nil {
return "", fmt.Errorf("failed to open a workspace for repo %s: %w", target.Repo, err)
}
defer workspace.Close()

outputs, err := workspace.Exec([]gitworkspace.Command{
{Alias: "fetch", Bin: "git", Args: []string{"fetch", remote, target.Branch}},
{Alias: "tip", Bin: "git", Args: []string{"rev-parse", "FETCH_HEAD^{commit}"}},
})
if err != nil {
return "", fmt.Errorf("failed to run git in repo %s: %w", target.Repo, err)
}
for _, out := range outputs {
if out.ExitCode != 0 {
return "", fmt.Errorf("git %s failed with exit code %d: %s", out.Alias, out.ExitCode, strings.TrimSpace(out.Stderr))
}
}
if len(outputs) != 2 {
return "", fmt.Errorf("expected 2 git outputs, got %d", len(outputs))
}

sha := strings.TrimSpace(outputs[1].Stdout)
if !fullCommitSHA.MatchString(sha) {
return "", fmt.Errorf("git returned %q, not a full commit SHA", sha)
}
return sha, nil
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,228 @@
// 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 baserevision

import (
"context"
"errors"
"runtime"
"sync"
"sync/atomic"
"testing"

"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"go.uber.org/mock/gomock"

"github.com/uber/submitqueue/platform/extension/gitworkspace"
"github.com/uber/submitqueue/platform/extension/gitworkspace/mock"
)

const (
testQueue = "monorepo"
testRepo = "org/repo"
testSHA = "9f2c1d0e3a4b5c6d7e8f9a0b1c2d3e4f5a6b7c8d"
)

var errTestFailure = errors.New("dependency down")

func TestFromGit(t *testing.T) {
tests := []struct {
name string
queue string
target Target
factoryOK bool
execOut []gitworkspace.Output
execErr error
wantArgs [][]string
want string
wantErr bool
}{
{
name: "returns the fetched tip",
queue: testQueue,
target: Target{Repo: testRepo, Remote: "upstream", Branch: "main"},
factoryOK: true,
execOut: []gitworkspace.Output{{Alias: "fetch"}, {Alias: "tip", Stdout: testSHA + "\n"}},
wantArgs: [][]string{{"fetch", "upstream", "main"}, {"rev-parse", "FETCH_HEAD^{commit}"}},
want: testSHA,
},
{
name: "defaults the remote to origin",
queue: testQueue,
target: Target{Repo: testRepo, Branch: "main"},
factoryOK: true,
execOut: []gitworkspace.Output{{Alias: "fetch"}, {Alias: "tip", Stdout: testSHA}},
wantArgs: [][]string{{"fetch", "origin", "main"}, {"rev-parse", "FETCH_HEAD^{commit}"}},
want: testSHA,
},
{
name: "queue without a target fails",
queue: "other",
target: Target{Repo: testRepo, Branch: "main"},
wantErr: true,
},
{
name: "workspace that cannot be opened fails",
queue: testQueue,
target: Target{Repo: testRepo, Branch: "main"},
wantErr: true,
},
{
name: "batch that cannot run fails",
queue: testQueue,
target: Target{Repo: testRepo, Branch: "main"},
factoryOK: true,
execErr: errTestFailure,
wantArgs: [][]string{{"fetch", "origin", "main"}, {"rev-parse", "FETCH_HEAD^{commit}"}},
wantErr: true,
},
{
name: "failed fetch fails",
queue: testQueue,
target: Target{Repo: testRepo, Branch: "main"},
factoryOK: true,
execOut: []gitworkspace.Output{{Alias: "fetch", ExitCode: 128, Stderr: "auth"}, {Alias: "tip", Skipped: true, ExitCode: -1}},
wantArgs: [][]string{{"fetch", "origin", "main"}, {"rev-parse", "FETCH_HEAD^{commit}"}},
wantErr: true,
},
{
name: "abbreviated SHA fails",
queue: testQueue,
target: Target{Repo: testRepo, Branch: "main"},
factoryOK: true,
execOut: []gitworkspace.Output{{Alias: "fetch"}, {Alias: "tip", Stdout: "9f2c1d0\n"}},
wantArgs: [][]string{{"fetch", "origin", "main"}, {"rev-parse", "FETCH_HEAD^{commit}"}},
wantErr: true,
},
{
name: "missing output fails",
queue: testQueue,
target: Target{Repo: testRepo, Branch: "main"},
factoryOK: true,
execOut: []gitworkspace.Output{{Alias: "fetch"}},
wantArgs: [][]string{{"fetch", "origin", "main"}, {"rev-parse", "FETCH_HEAD^{commit}"}},
wantErr: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
ctrl := gomock.NewController(t)
factory := mock.NewMockFactory(ctrl)
if tt.queue == testQueue {
if tt.factoryOK {
workspace := mock.NewMockWorkspace(ctrl)
factory.EXPECT().For(gomock.Any(), testRepo).Return(workspace, nil)
workspace.EXPECT().Exec(gomock.Cond(func(commands []gitworkspace.Command) bool {
if len(commands) != len(tt.wantArgs) {
return false
}
for i, command := range commands {
if command.Bin != "git" || !assert.ObjectsAreEqual(tt.wantArgs[i], command.Args) {
return false
}
}
return true
})).Return(tt.execOut, tt.execErr)
workspace.EXPECT().Close().Return(nil)
} else {
factory.EXPECT().For(gomock.Any(), testRepo).Return(nil, errTestFailure)
}
}

got, err := FromGit(factory, map[string]Target{testQueue: tt.target})(context.Background(), tt.queue)

if tt.wantErr {
require.Error(t, err)
return
}
require.NoError(t, err)
assert.Equal(t, tt.want, got)
})
}
}

func TestFromGit_RepoLocks(t *testing.T) {
validOutputs := []gitworkspace.Output{{Alias: "fetch"}, {Alias: "tip", Stdout: testSHA}}

t.Run("calls on one repo never overlap", func(t *testing.T) {
ctrl := gomock.NewController(t)
factory := mock.NewMockFactory(ctrl)
workspace := mock.NewMockWorkspace(ctrl)
factory.EXPECT().For(gomock.Any(), testRepo).Return(workspace, nil).AnyTimes()
workspace.EXPECT().Close().Return(nil).AnyTimes()
var inFlight, maxInFlight atomic.Int32
workspace.EXPECT().Exec(gomock.Any()).DoAndReturn(func([]gitworkspace.Command) ([]gitworkspace.Output, error) {
n := inFlight.Add(1)
for m := maxInFlight.Load(); n > m && !maxInFlight.CompareAndSwap(m, n); m = maxInFlight.Load() {
}
runtime.Gosched()
inFlight.Add(-1)
return validOutputs, nil
}).AnyTimes()
baseRevision := FromGit(factory, map[string]Target{
"main-queue": {Repo: testRepo, Branch: "main"},
"release-queue": {Repo: testRepo, Branch: "release"},
})

var wg sync.WaitGroup
for i := 0; i < 32; i++ {
queue := "main-queue"
if i%2 == 1 {
queue = "release-queue"
}
wg.Add(1)
go func() {
defer wg.Done()
_, err := baseRevision(context.Background(), queue)
assert.NoError(t, err)
}()
}
wg.Wait()

assert.Equal(t, int32(1), maxInFlight.Load())
})

t.Run("calls on different repos do not block each other", func(t *testing.T) {
ctrl := gomock.NewController(t)
factory := mock.NewMockFactory(ctrl)
entered := map[string]chan struct{}{"repo-x": make(chan struct{}), "repo-y": make(chan struct{})}
for repo, other := range map[string]string{"repo-x": "repo-y", "repo-y": "repo-x"} {
workspace := mock.NewMockWorkspace(ctrl)
factory.EXPECT().For(gomock.Any(), repo).Return(workspace, nil)
workspace.EXPECT().Exec(gomock.Any()).DoAndReturn(func([]gitworkspace.Command) ([]gitworkspace.Output, error) {
close(entered[repo])
<-entered[other]
return validOutputs, nil
})
workspace.EXPECT().Close().Return(nil)
}
baseRevision := FromGit(factory, map[string]Target{
"queue-x": {Repo: "repo-x", Branch: "main"},
"queue-y": {Repo: "repo-y", Branch: "main"},
})

var wg sync.WaitGroup
for _, queue := range []string{"queue-x", "queue-y"} {
wg.Add(1)
go func() {
defer wg.Done()
_, err := baseRevision(context.Background(), queue)
assert.NoError(t, err)
}()
}
wg.Wait()
})
}
Loading