diff --git a/sqsim/adapter/buildrunner/BUILD.bazel b/sqsim/adapter/buildrunner/BUILD.bazel new file mode 100644 index 00000000..1f36a098 --- /dev/null +++ b/sqsim/adapter/buildrunner/BUILD.bazel @@ -0,0 +1,32 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["buildrunner.go"], + importpath = "github.com/uber/submitqueue/sqsim/adapter/buildrunner", + visibility = ["//visibility:public"], + deps = [ + "//sqsim/adapter/internal/operationlock:go_default_library", + "//sqsim/entity:go_default_library", + "//sqsim/model:go_default_library", + "//submitqueue/core/changeset:go_default_library", + "//submitqueue/entity:go_default_library", + "//submitqueue/extension/buildrunner:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["buildrunner_test.go"], + embed = [":go_default_library"], + deps = [ + "//platform/base/change:go_default_library", + "//sqsim:go_default_library", + "//sqsim/model:go_default_library", + "//submitqueue/core/changeset/fake:go_default_library", + "//submitqueue/entity:go_default_library", + "//submitqueue/extension/buildrunner:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + ], +) diff --git a/sqsim/adapter/buildrunner/buildrunner.go b/sqsim/adapter/buildrunner/buildrunner.go new file mode 100644 index 00000000..f9320663 --- /dev/null +++ b/sqsim/adapter/buildrunner/buildrunner.go @@ -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 buildrunner adapts SQSim Build Runner behavior to the production extension. +package buildrunner + +import ( + "context" + "crypto/sha256" + "fmt" + "reflect" + "strconv" + "strings" + "sync" + "time" + + "github.com/uber/submitqueue/sqsim/adapter/internal/operationlock" + simentity "github.com/uber/submitqueue/sqsim/entity" + "github.com/uber/submitqueue/sqsim/model" + "github.com/uber/submitqueue/submitqueue/core/changeset" + "github.com/uber/submitqueue/submitqueue/entity" + buildrunnerext "github.com/uber/submitqueue/submitqueue/extension/buildrunner" +) + +var _ buildrunnerext.BuildRunner = (*Runner)(nil) + +// Runner executes SQSim Build Runner behavior. +type Runner struct { + runtime *model.Runtime + resolver changeset.Resolver + queue string + operations *operationlock.Locker + triggered *model.State[entity.BuildID] + builds *model.State[*buildExecution] +} + +type buildExecution struct { + mu sync.Mutex + changeURI string + createdAt time.Time + execution simentity.BuildExecution + statusCalls int + cancelled bool +} + +// New returns an SQSim Build Runner bound to one queue. +func New(runtime *model.Runtime, resolver changeset.Resolver, queue string) *Runner { + return &Runner{ + runtime: runtime, + resolver: resolver, + queue: queue, + operations: operationlock.New(), + triggered: model.NewState[entity.BuildID](), + builds: model.NewState[*buildExecution](), + } +} + +// Trigger creates the modeled build selected by the head batch's SQSim URI. +func (r *Runner) Trigger(ctx context.Context, base []entity.Batch, head entity.Batch, _ entity.BuildMetadata) (entity.BuildID, error) { + if head.Queue != r.queue { + return entity.BuildID{}, fmt.Errorf("head batch queue %q does not match SQSim Build Runner queue %q", head.Queue, r.queue) + } + changeURI, err := r.resolveHeadBehavior(ctx, head) + if err != nil { + return entity.BuildID{}, err + } + key := triggerKey(r.queue, base, head) + unlock := r.operations.Lock(key) + defer unlock() + + if buildID, ok := r.triggered.Get(key); ok { + return buildID, nil + } + invocation, err := r.runtime.NextBuildTrigger(changeURI) + if err != nil { + return entity.BuildID{}, fmt.Errorf("next Build Runner Trigger invocation: %w", err) + } + if err := r.runtime.Clock().Wait(ctx, time.Duration(invocation.DelayMs)*time.Millisecond); err != nil { + return entity.BuildID{}, err + } + if invocation.Fault.Phase == simentity.FaultBeforeSideEffect { + return entity.BuildID{}, model.ErrorForFault(invocation.Fault) + } + + buildID := entity.BuildID{ID: buildIDFor(key)} + execution := &buildExecution{ + changeURI: changeURI, + createdAt: r.runtime.Clock().Now(), + execution: cloneBuildExecution(invocation.Outcome.Build), + } + r.builds.Set(buildID.ID, execution) + r.triggered.Set(key, buildID) + if err := model.ErrorForFault(invocation.Fault); err != nil { + return entity.BuildID{}, err + } + return buildID, nil +} + +// Status returns status from the build's real-time scenario timeline. +func (r *Runner) Status(_ context.Context, buildID entity.BuildID) (entity.BuildStatus, entity.BuildMetadata, error) { + execution, ok := r.builds.Get(buildID.ID) + if !ok { + return entity.BuildStatusUnknown, nil, fmt.Errorf("SQSim build %q does not exist", buildID.ID) + } + execution.mu.Lock() + defer execution.mu.Unlock() + + execution.statusCalls++ + for _, fault := range execution.execution.StatusFaults { + if fault.Call == execution.statusCalls { + return entity.BuildStatusUnknown, nil, model.ErrorForFault(fault.Fault) + } + } + + elapsed := r.runtime.Clock().Now().Sub(execution.createdAt) + status := toBuildStatus(model.BuildStatusAt(execution.execution.Timeline, elapsed)) + if execution.cancelled { + status = entity.BuildStatusCancelled + } + return status, entity.BuildMetadata{ + "sqsim.change_uri": execution.changeURI, + "sqsim.elapsed_ms": strconv.FormatInt(elapsed.Milliseconds(), 10), + }, nil +} + +// Cancel marks a non-terminal modeled build as cancelled. +func (r *Runner) Cancel(_ context.Context, buildID entity.BuildID) error { + execution, ok := r.builds.Get(buildID.ID) + if !ok { + return fmt.Errorf("SQSim build %q does not exist", buildID.ID) + } + execution.mu.Lock() + defer execution.mu.Unlock() + if execution.cancelled { + return nil + } + status := model.BuildStatusAt(execution.execution.Timeline, r.runtime.Clock().Now().Sub(execution.createdAt)) + if status.IsTerminal() { + return nil + } + execution.cancelled = true + return nil +} + +func (r *Runner) resolveHeadBehavior(ctx context.Context, head entity.Batch) (string, error) { + changes, err := r.resolver.ChangesForBatch(ctx, head) + if err != nil { + return "", fmt.Errorf("resolve head batch %s: %w", head.ID, err) + } + var selectedURI string + var selected simentity.BuildRunnerBehavior + for _, change := range changes { + for _, changeURI := range change.URIs { + land, err := r.runtime.Resolve(changeURI) + if err != nil { + return "", fmt.Errorf("resolve head change URI %q: %w", changeURI, err) + } + if land.Queue != r.queue { + return "", fmt.Errorf("land %q queue %q does not match SQSim Build Runner queue %q", land.Name, land.Queue, r.queue) + } + if selectedURI == "" { + selectedURI = changeURI + selected = land.Behavior.BuildRunner + continue + } + if !reflect.DeepEqual(selected, land.Behavior.BuildRunner) { + return "", fmt.Errorf("head batch %q contains incompatible SQSim Build Runner behavior", head.ID) + } + } + } + if selectedURI == "" { + return "", fmt.Errorf("head batch %q has no SQSim change URI", head.ID) + } + return selectedURI, nil +} + +func triggerKey(queue string, base []entity.Batch, head entity.Batch) string { + var key strings.Builder + appendKeyPart(&key, queue) + for _, batch := range base { + appendKeyPart(&key, batch.ID) + } + appendKeyPart(&key, head.ID) + return key.String() +} + +func appendKeyPart(key *strings.Builder, value string) { + fmt.Fprintf(key, "%d:%s", len(value), value) +} + +func buildIDFor(key string) string { + sum := sha256.Sum256([]byte(key)) + return fmt.Sprintf("sqsim-%x", sum[:]) +} + +func cloneBuildExecution(execution simentity.BuildExecution) simentity.BuildExecution { + execution.Timeline = append([]simentity.BuildStatusPoint(nil), execution.Timeline...) + execution.StatusFaults = append([]simentity.FaultOnCall(nil), execution.StatusFaults...) + return execution +} + +func toBuildStatus(status simentity.BuildStatus) entity.BuildStatus { + switch status { + case simentity.BuildAccepted: + return entity.BuildStatusAccepted + case simentity.BuildRunning: + return entity.BuildStatusRunning + case simentity.BuildSucceeded: + return entity.BuildStatusSucceeded + case simentity.BuildFailed: + return entity.BuildStatusFailed + case simentity.BuildCancelled: + return entity.BuildStatusCancelled + default: + return entity.BuildStatusUnknown + } +} diff --git a/sqsim/adapter/buildrunner/buildrunner_test.go b/sqsim/adapter/buildrunner/buildrunner_test.go new file mode 100644 index 00000000..df620f13 --- /dev/null +++ b/sqsim/adapter/buildrunner/buildrunner_test.go @@ -0,0 +1,146 @@ +// 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 buildrunner + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/platform/base/change" + "github.com/uber/submitqueue/sqsim" + "github.com/uber/submitqueue/sqsim/model" + changesetfake "github.com/uber/submitqueue/submitqueue/core/changeset/fake" + "github.com/uber/submitqueue/submitqueue/entity" + buildrunnerext "github.com/uber/submitqueue/submitqueue/extension/buildrunner" +) + +const ( + testQueue = "sqsim" + testURI = "sqsim://local/build-runner/l1" +) + +func TestRunnerImplementsBuildRunner(t *testing.T) { + var _ buildrunnerext.BuildRunner = (*Runner)(nil) +} + +func TestTriggerResponseLossReturnsSameBuildOnRetry(t *testing.T) { + runner, clock := newRunner(t, sqsim.NewBuildRunnerBehavior().Trigger( + sqsim.BuildCreatedWithFault( + sqsim.RetryableErrorAfterSideEffect(), + sqsim.StatusAt(0, sqsim.BuildAccepted), + sqsim.StatusAt(time.Second, sqsim.BuildSucceeded), + ), + )) + head := entity.Batch{ID: "sqsim/batch/1", Queue: testQueue} + + first, err := runner.Trigger(context.Background(), nil, head, nil) + require.Error(t, err) + assert.Empty(t, first.ID) + + second, err := runner.Trigger(context.Background(), nil, head, nil) + require.NoError(t, err) + assert.NotEmpty(t, second.ID) + + status, _, err := runner.Status(context.Background(), second) + require.NoError(t, err) + assert.Equal(t, entity.BuildStatusAccepted, status) + clock.Advance(time.Second) + status, _, err = runner.Status(context.Background(), second) + require.NoError(t, err) + assert.Equal(t, entity.BuildStatusSucceeded, status) +} + +func TestTriggerFailureBeforeSideEffectRecovers(t *testing.T) { + runner, _ := newRunner(t, sqsim.NewBuildRunnerBehavior().Trigger( + sqsim.BuildTriggerFault(sqsim.RetryableErrorBeforeSideEffect()), + sqsim.BuildCreated(sqsim.StatusAt(0, sqsim.BuildSucceeded)), + )) + head := entity.Batch{ID: "sqsim/batch/1", Queue: testQueue} + + _, err := runner.Trigger(context.Background(), nil, head, nil) + require.Error(t, err) + + buildID, err := runner.Trigger(context.Background(), nil, head, nil) + require.NoError(t, err) + status, _, err := runner.Status(context.Background(), buildID) + require.NoError(t, err) + assert.Equal(t, entity.BuildStatusSucceeded, status) +} + +func TestStatusFaultRecovers(t *testing.T) { + runner, _ := newRunner(t, sqsim.NewBuildRunnerBehavior(). + Trigger(sqsim.BuildCreated(sqsim.StatusAt(0, sqsim.BuildSucceeded))). + StatusFaultOnCall(1, sqsim.RetryableErrorBeforeSideEffect())) + head := entity.Batch{ID: "sqsim/batch/1", Queue: testQueue} + buildID, err := runner.Trigger(context.Background(), nil, head, nil) + require.NoError(t, err) + + _, _, err = runner.Status(context.Background(), buildID) + require.Error(t, err) + status, _, err := runner.Status(context.Background(), buildID) + require.NoError(t, err) + assert.Equal(t, entity.BuildStatusSucceeded, status) +} + +func newRunner(t *testing.T, buildRunner *sqsim.BuildRunnerBehaviorBuilder) (*Runner, *fakeClock) { + t.Helper() + behavior := sqsim.NewBehavior(). + BuildRunner(buildRunner). + MergeConflictCheck(sqsim.SuccessfulMergeConflictCheck()). + Merge(sqsim.SuccessfulMerge()) + scenario, err := sqsim.NewScenario(). + Timeout(time.Minute). + Land(sqsim.NewLand("l1").Queue(testQueue).Behavior(behavior).Expect(sqsim.RequestLanded)). + Build() + require.NoError(t, err) + profile, err := model.Compile("build-runner", scenario) + require.NoError(t, err) + clock := &fakeClock{now: time.Unix(0, 0)} + runtime, err := model.NewRuntime(profile, clock) + require.NoError(t, err) + resolver := changesetfake.New().Set("sqsim/batch/1", change.Change{URIs: []string{testURI}}) + return New(runtime, resolver, testQueue), clock +} + +type fakeClock struct { + mu sync.Mutex + now time.Time +} + +func (c *fakeClock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.now +} + +func (c *fakeClock) Wait(ctx context.Context, duration time.Duration) error { + select { + case <-ctx.Done(): + return ctx.Err() + default: + c.Advance(duration) + return nil + } +} + +func (c *fakeClock) Advance(duration time.Duration) { + c.mu.Lock() + defer c.mu.Unlock() + c.now = c.now.Add(duration) +} diff --git a/sqsim/adapter/internal/operationlock/BUILD.bazel b/sqsim/adapter/internal/operationlock/BUILD.bazel new file mode 100644 index 00000000..fecf261b --- /dev/null +++ b/sqsim/adapter/internal/operationlock/BUILD.bazel @@ -0,0 +1,15 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["operationlock.go"], + importpath = "github.com/uber/submitqueue/sqsim/adapter/internal/operationlock", + visibility = ["//sqsim/adapter:__subpackages__"], +) + +go_test( + name = "go_default_test", + srcs = ["operationlock_test.go"], + embed = [":go_default_library"], + deps = ["@com_github_stretchr_testify//assert:go_default_library"], +) diff --git a/sqsim/adapter/internal/operationlock/operationlock.go b/sqsim/adapter/internal/operationlock/operationlock.go new file mode 100644 index 00000000..ebce8adc --- /dev/null +++ b/sqsim/adapter/internal/operationlock/operationlock.go @@ -0,0 +1,43 @@ +// 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 operationlock serializes duplicate calls for one production operation. +package operationlock + +import "sync" + +// Locker supplies a stable mutex for each operation key. +type Locker struct { + mu sync.Mutex + locks map[string]*sync.Mutex +} + +// New returns an empty keyed locker. +func New() *Locker { + return &Locker{locks: make(map[string]*sync.Mutex)} +} + +// Lock locks key and returns its unlock function. +func (l *Locker) Lock(key string) func() { + l.mu.Lock() + lock, ok := l.locks[key] + if !ok { + lock = &sync.Mutex{} + l.locks[key] = lock + } + l.mu.Unlock() + + lock.Lock() + return lock.Unlock +} diff --git a/sqsim/adapter/internal/operationlock/operationlock_test.go b/sqsim/adapter/internal/operationlock/operationlock_test.go new file mode 100644 index 00000000..650d19d9 --- /dev/null +++ b/sqsim/adapter/internal/operationlock/operationlock_test.go @@ -0,0 +1,31 @@ +// 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 operationlock + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestLockerReturnsStableLockPerKey(t *testing.T) { + locker := New() + unlock := locker.Lock("operation") + unlock() + + unlock = locker.Lock("operation") + assert.NotNil(t, unlock) + unlock() +} diff --git a/sqsim/adapter/merger/BUILD.bazel b/sqsim/adapter/merger/BUILD.bazel new file mode 100644 index 00000000..539d0d15 --- /dev/null +++ b/sqsim/adapter/merger/BUILD.bazel @@ -0,0 +1,32 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["merger.go"], + importpath = "github.com/uber/submitqueue/sqsim/adapter/merger", + visibility = ["//visibility:public"], + deps = [ + "//api/runway/messagequeue:go_default_library", + "//api/runway/messagequeue/protopb:go_default_library", + "//runway/extension/merger:go_default_library", + "//sqsim/adapter/internal/operationlock:go_default_library", + "//sqsim/entity:go_default_library", + "//sqsim/model:go_default_library", + "@org_golang_google_protobuf//proto:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["merger_test.go"], + embed = [":go_default_library"], + deps = [ + "//api/base/change/protopb:go_default_library", + "//api/runway/messagequeue:go_default_library", + "//runway/extension/merger:go_default_library", + "//sqsim:go_default_library", + "//sqsim/model:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + ], +) diff --git a/sqsim/adapter/merger/merger.go b/sqsim/adapter/merger/merger.go new file mode 100644 index 00000000..5f4127ba --- /dev/null +++ b/sqsim/adapter/merger/merger.go @@ -0,0 +1,206 @@ +// 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 merger adapts SQSim merge behavior to the production Runway extension. +package merger + +import ( + "context" + "crypto/sha256" + "fmt" + "reflect" + "time" + + runwaymq "github.com/uber/submitqueue/api/runway/messagequeue" + runwaypb "github.com/uber/submitqueue/api/runway/messagequeue/protopb" + mergerext "github.com/uber/submitqueue/runway/extension/merger" + "github.com/uber/submitqueue/sqsim/adapter/internal/operationlock" + simentity "github.com/uber/submitqueue/sqsim/entity" + "github.com/uber/submitqueue/sqsim/model" + "google.golang.org/protobuf/proto" +) + +var _ mergerext.Merger = (*Merger)(nil) + +// Merger executes SQSim merge-conflict-check and committing merge behavior. +type Merger struct { + runtime *model.Runtime + queue string + operations *operationlock.Locker + checks *model.State[checkResult] + merges *model.State[*runwaymq.MergeResult] +} + +type checkResult struct { + result *runwaymq.MergeResult + conflict bool +} + +// New returns an SQSim Merger bound to one queue. +func New(runtime *model.Runtime, queue string) *Merger { + return &Merger{ + runtime: runtime, + queue: queue, + operations: operationlock.New(), + checks: model.NewState[checkResult](), + merges: model.NewState[*runwaymq.MergeResult](), + } +} + +// CheckMergeability executes the next modeled dry-run invocation. +func (m *Merger) CheckMergeability(ctx context.Context, req *runwaymq.MergeRequest) (*runwaymq.MergeResult, error) { + changeURI, err := m.resolveBehavior(req, true) + if err != nil { + return nil, err + } + key := "check:" + req.GetId() + unlock := m.operations.Lock(key) + defer unlock() + + if retained, ok := m.checks.Get(key); ok { + if retained.conflict { + return nil, mergerext.ErrConflict + } + return cloneResult(retained.result), nil + } + invocation, err := m.runtime.NextMergeConflictCheck(changeURI) + if err != nil { + return nil, fmt.Errorf("next merge-conflict-check invocation: %w", err) + } + if err := m.runtime.Clock().Wait(ctx, time.Duration(invocation.DelayMs)*time.Millisecond); err != nil { + return nil, err + } + if err := model.ErrorForFault(invocation.Fault); err != nil { + return nil, err + } + if invocation.Outcome == simentity.MergeConflict { + m.checks.Set(key, checkResult{conflict: true}) + return nil, mergerext.ErrConflict + } + result := successfulCheckResult(req) + m.checks.Set(key, checkResult{result: result}) + return cloneResult(result), nil +} + +// Merge executes the next modeled committing merge invocation. +func (m *Merger) Merge(ctx context.Context, req *runwaymq.MergeRequest) (*runwaymq.MergeResult, error) { + changeURI, err := m.resolveBehavior(req, false) + if err != nil { + return nil, err + } + key := "merge:" + req.GetId() + unlock := m.operations.Lock(key) + defer unlock() + + if retained, ok := m.merges.Get(key); ok { + return cloneResult(retained), nil + } + invocation, err := m.runtime.NextMerge(changeURI) + if err != nil { + return nil, fmt.Errorf("next Merge invocation: %w", err) + } + if err := m.runtime.Clock().Wait(ctx, time.Duration(invocation.DelayMs)*time.Millisecond); err != nil { + return nil, err + } + if invocation.Fault.Phase == simentity.FaultBeforeSideEffect { + return nil, model.ErrorForFault(invocation.Fault) + } + + result := successfulMergeResult(req) + m.merges.Set(key, result) + if err := model.ErrorForFault(invocation.Fault); err != nil { + return nil, err + } + return cloneResult(result), nil +} + +func (m *Merger) resolveBehavior(req *runwaymq.MergeRequest, check bool) (string, error) { + if req == nil { + return "", fmt.Errorf("merge request is required") + } + if req.GetId() == "" { + return "", fmt.Errorf("merge request ID is required") + } + if req.GetQueueName() != m.queue { + return "", fmt.Errorf("merge request queue %q does not match SQSim Merger queue %q", req.GetQueueName(), m.queue) + } + var selectedURI string + var selected any + for _, step := range req.GetSteps() { + for _, change := range step.GetChanges() { + for _, changeURI := range change.GetUris() { + land, err := m.runtime.Resolve(changeURI) + if err != nil { + return "", fmt.Errorf("resolve merge change URI %q: %w", changeURI, err) + } + if land.Queue != m.queue { + return "", fmt.Errorf("land %q queue %q does not match SQSim Merger queue %q", land.Name, land.Queue, m.queue) + } + var behavior any = land.Behavior.Merge + if check { + behavior = land.Behavior.MergeConflictCheck + } + if selectedURI == "" { + selectedURI = changeURI + selected = behavior + continue + } + if !reflect.DeepEqual(selected, behavior) { + return "", fmt.Errorf("merge request %q contains incompatible SQSim behavior", req.GetId()) + } + } + } + } + if selectedURI == "" { + return "", fmt.Errorf("merge request %q has no SQSim change URI", req.GetId()) + } + return selectedURI, nil +} + +func successfulCheckResult(req *runwaymq.MergeRequest) *runwaymq.MergeResult { + steps := make([]*runwaymq.StepResult, len(req.GetSteps())) + for i, step := range req.GetSteps() { + steps[i] = &runwaymq.StepResult{StepId: step.GetStepId()} + } + return &runwaymq.MergeResult{ + Id: req.GetId(), + Outcome: runwaypb.Outcome_SUCCEEDED, + Steps: steps, + } +} + +func successfulMergeResult(req *runwaymq.MergeRequest) *runwaymq.MergeResult { + steps := make([]*runwaymq.StepResult, len(req.GetSteps())) + for i, step := range req.GetSteps() { + sum := sha256.Sum256([]byte(fmt.Sprintf("%s\x00%s\x00%d", req.GetId(), step.GetStepId(), i))) + steps[i] = &runwaymq.StepResult{ + StepId: step.GetStepId(), + Outputs: []*runwaymq.StepOutput{{ + Id: fmt.Sprintf("%x", sum[:20]), + }}, + } + } + return &runwaymq.MergeResult{ + Id: req.GetId(), + Outcome: runwaypb.Outcome_SUCCEEDED, + Steps: steps, + } +} + +func cloneResult(result *runwaymq.MergeResult) *runwaymq.MergeResult { + if result == nil { + return nil + } + return proto.Clone(result).(*runwaymq.MergeResult) +} diff --git a/sqsim/adapter/merger/merger_test.go b/sqsim/adapter/merger/merger_test.go new file mode 100644 index 00000000..8f36c565 --- /dev/null +++ b/sqsim/adapter/merger/merger_test.go @@ -0,0 +1,135 @@ +// 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 merger + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + changepb "github.com/uber/submitqueue/api/base/change/protopb" + runwaymq "github.com/uber/submitqueue/api/runway/messagequeue" + mergerext "github.com/uber/submitqueue/runway/extension/merger" + "github.com/uber/submitqueue/sqsim" + "github.com/uber/submitqueue/sqsim/model" +) + +const testQueue = "sqsim" + +func TestMergerImplementsInterface(t *testing.T) { + var _ mergerext.Merger = (*Merger)(nil) +} + +func TestCheckMergeabilityFailureRecovers(t *testing.T) { + adapter := newMerger(t, + sqsim.NewMergeConflictCheckBehavior().Invoke( + sqsim.MergeConflictCheckFault(sqsim.RetryableErrorBeforeSideEffect()), + sqsim.MergeConflictCheckSucceeded(), + ), + sqsim.SuccessfulMerge(), + ) + req := mergeRequest("check/1") + + _, err := adapter.CheckMergeability(context.Background(), req) + require.Error(t, err) + result, err := adapter.CheckMergeability(context.Background(), req) + require.NoError(t, err) + assert.Equal(t, req.GetId(), result.GetId()) +} + +func TestCheckMergeabilityConflictIsRetained(t *testing.T) { + adapter := newMerger(t, sqsim.ConflictingMergeConflictCheck(), sqsim.SuccessfulMerge()) + req := mergeRequest("check/1") + + _, err := adapter.CheckMergeability(context.Background(), req) + require.ErrorIs(t, err, mergerext.ErrConflict) + _, err = adapter.CheckMergeability(context.Background(), req) + require.ErrorIs(t, err, mergerext.ErrConflict) +} + +func TestMergeResponseLossReturnsRetainedResult(t *testing.T) { + adapter := newMerger(t, + sqsim.SuccessfulMergeConflictCheck(), + sqsim.NewMergeBehavior().Invoke( + sqsim.MergeSucceededAfter(0).Fault(sqsim.RetryableErrorAfterSideEffect()), + ), + ) + req := mergeRequest("merge/1") + + _, err := adapter.Merge(context.Background(), req) + require.Error(t, err) + result, err := adapter.Merge(context.Background(), req) + require.NoError(t, err) + require.Len(t, result.GetSteps(), 1) + require.Len(t, result.GetSteps()[0].GetOutputs(), 1) + assert.Len(t, result.GetSteps()[0].GetOutputs()[0].GetId(), 40) +} + +func newMerger(t *testing.T, check *sqsim.MergeConflictCheckBehaviorBuilder, merge *sqsim.MergeBehaviorBuilder) *Merger { + t.Helper() + behavior := sqsim.NewBehavior(). + BuildRunner(sqsim.SuccessfulBuildRunner()). + MergeConflictCheck(check). + Merge(merge) + scenario, err := sqsim.NewScenario(). + Timeout(time.Minute). + Land(sqsim.NewLand("l1").Queue(testQueue).Behavior(behavior).Expect(sqsim.RequestLanded)). + Build() + require.NoError(t, err) + profile, err := model.Compile("merger", scenario) + require.NoError(t, err) + runtime, err := model.NewRuntime(profile, &fakeClock{now: time.Unix(0, 0)}) + require.NoError(t, err) + return New(runtime, testQueue) +} + +func mergeRequest(id string) *runwaymq.MergeRequest { + return &runwaymq.MergeRequest{ + Id: id, + QueueName: testQueue, + Steps: []*runwaymq.MergeStep{{ + StepId: "step/1", + Changes: []*changepb.Change{{ + Uris: []string{"sqsim://local/merger/l1"}, + }}, + }}, + } +} + +type fakeClock struct { + mu sync.Mutex + now time.Time +} + +func (c *fakeClock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.now +} + +func (c *fakeClock) Wait(ctx context.Context, duration time.Duration) error { + select { + case <-ctx.Done(): + return ctx.Err() + default: + c.mu.Lock() + c.now = c.now.Add(duration) + c.mu.Unlock() + return nil + } +}