From babf53a35745c853cf4b89e8a8ffdef50401a72f Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Tue, 18 Aug 2026 18:33:36 -0700 Subject: [PATCH] feat(demo): add BURST mode for a simultaneous enqueue of independent changes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add -burst (Makefile BURST=false) which, for independent changes, creates every change first and only then fires all the Land calls together, so the requests reach the queue in one burst instead of being enqueued as each is created. It works for every provider: burst runs each source through the same two phases, so fake creates instantly, github creates concurrently, git creates serially on its single work tree — and in all three the enqueues arrive together once creation is done. Burst does not make creation faster; with the git source it is still serialized on one work tree. It only separates creation from enqueuing so the publishes are not spread across the creation phase. The default is unchanged — each change lands the moment it exists, so the queue starts working during creation. A small lander seam lets the two-phase ordering be tested without a live gateway. --- Makefile | 6 +- service/submitqueue/demo/requests/BUILD.bazel | 2 + service/submitqueue/demo/requests/main.go | 78 ++++++++++++++++++- .../submitqueue/demo/requests/main_test.go | 64 +++++++++++++++ 4 files changed, 146 insertions(+), 4 deletions(-) diff --git a/Makefile b/Makefile index 46ffc33b..1cfad738 100644 --- a/Makefile +++ b/Makefile @@ -78,6 +78,9 @@ FOLDERS ?= 0 FILES ?= 3 CONCURRENCY ?= 5 STACKED ?= false +# BURST=true creates every change first, then enqueues them all at once instead +# of as each is created. Independent changes only; a stack always lands as one. +BURST ?= false SINCE ?= 1h LIMIT ?= 50 LAND ?= true @@ -240,7 +243,7 @@ clean-proto: ## Clean generated proto files @rm -f $(foreach p,$(PROTO_PACKAGES),$(p)/protopb/*.pb.go $(p)/protopb/*.pb.yarpc.go) @echo "Proto clean complete!" -demo-requests: ## Create N changes, enqueue each as it is created, and watch (PROVIDER=fake|git|github COUNT=3 FOLDERS=0 FILES=3 CONCURRENCY=5) +demo-requests: ## Create N changes, enqueue each as it is created, and watch (PROVIDER=fake|git|github COUNT=3 FOLDERS=0 FILES=3 CONCURRENCY=5 BURST=false) @set -e; $(resolve_gateway_addr); $(resolve_provider); \ $(BAZEL) run //service/submitqueue/demo/requests -- \ -provider $$provider \ @@ -251,6 +254,7 @@ demo-requests: ## Create N changes, enqueue each as it is created, and watch (PR -files $(FILES) \ -concurrency $(CONCURRENCY) \ -stacked=$(STACKED) \ + -burst=$(BURST) \ -addr $$addr \ -queue $(QUEUE) \ -strategy $(STRATEGY) \ diff --git a/service/submitqueue/demo/requests/BUILD.bazel b/service/submitqueue/demo/requests/BUILD.bazel index e5ffc347..3fdedc7e 100644 --- a/service/submitqueue/demo/requests/BUILD.bazel +++ b/service/submitqueue/demo/requests/BUILD.bazel @@ -46,10 +46,12 @@ go_test( "SUBMITQUEUE_TEST_GIT": "$(location @git//:git)", }, deps = [ + "//api/base/mergestrategy/protopb:go_default_library", "//platform/base/change/git:go_default_library", "//platform/fakemarker:go_default_library", "//platform/gitexec:go_default_library", "//platform/gitexec/gitexectest:go_default_library", + "//submitqueue/client:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", "@com_github_stretchr_testify//require:go_default_library", ], diff --git a/service/submitqueue/demo/requests/main.go b/service/submitqueue/demo/requests/main.go index 6f87226c..fd542765 100644 --- a/service/submitqueue/demo/requests/main.go +++ b/service/submitqueue/demo/requests/main.go @@ -92,6 +92,7 @@ type config struct { files int concurrency int stacked bool + burst bool prefix string land bool watch bool @@ -120,6 +121,8 @@ func parseFlags() config { flag.IntVar(&c.concurrency, "concurrency", 5, "how many changes to create at once; a stack ignores it, being sequential by nature, and -provider git serializes its git commands") flag.BoolVar(&c.stacked, "stacked", false, "chain the changes and enqueue them as one stack") + flag.BoolVar(&c.burst, "burst", false, + "create every change first, then enqueue them all at once instead of as each is created; independent changes only") flag.StringVar(&c.prefix, "prefix", "demo", "branch name prefix") flag.BoolVar(&c.land, "land", true, "enqueue each change as it is created") flag.BoolVar(&c.watch, "watch", true, "watch the requests until they all settle") @@ -242,6 +245,9 @@ func shape(cfg config) string { if cfg.stacked { return "stacked, enqueued as one request once the chain exists" } + if cfg.burst && cfg.land { + return fmt.Sprintf("independent, created %d at a time, then all enqueued at once", cfg.concurrency) + } if cfg.concurrency > 1 { return fmt.Sprintf("independent, %d at a time, each enqueued as soon as it is created", cfg.concurrency) } @@ -432,6 +438,12 @@ func changeFileCount(tag string, change, min int) int { return min + int(sum[0]%4) } +// lander enqueues a request. *client.Client is the real one; a test supplies +// its own to observe when each change is enqueued relative to when it is created. +type lander interface { + Land(ctx context.Context, queue string, uris []string, strategy mergestrategypb.Strategy) (string, error) +} + // createAndEnqueue creates the changes and puts them on the queue, filling // in the tracker's rows as it goes and reporting each step beneath the table. // @@ -451,7 +463,7 @@ func changeFileCount(tag string, change, min int) int { func createAndEnqueue( ctx context.Context, src changeSource, - sq *client.Client, + sq lander, cfg config, strategy mergestrategypb.Strategy, tag, baseSHA string, @@ -480,12 +492,16 @@ func createAndEnqueue( func createIndependent( ctx context.Context, src changeSource, - sq *client.Client, + sq lander, cfg config, strategy mergestrategypb.Strategy, tag, baseSHA string, t *client.Tracker, ) ([]change, error) { + if cfg.burst && cfg.land { + return createBurst(ctx, src, sq, cfg, strategy, tag, baseSHA, t) + } + rows := t.Rows() // Indexed rather than appended: the workers finish in whatever order the // provider answers them, and the caller still wants the run's own order. @@ -521,6 +537,62 @@ func createIndependent( return created, nil } +// createBurst creates every change first and only then enqueues them, firing +// all the Land calls together so the requests arrive at the queue in one burst. +// +// The default path lands each change the moment it exists, so the queue starts +// working while later changes are still being created. Burst trades that early +// overlap for a simultaneous arrival: useful for watching the queue admit a +// large batch at once. It does not make creation faster — with -provider git the +// creation phase is still serialized on a single work tree — it only separates +// creation from enqueuing so the enqueues are not spread across it. +func createBurst( + ctx context.Context, + src changeSource, + sq lander, + cfg config, + strategy mergestrategypb.Strategy, + tag, baseSHA string, + t *client.Tracker, +) ([]change, error) { + rows := t.Rows() + created := make([]change, cfg.count) + + create, createCtx := errgroup.WithContext(ctx) + create.SetLimit(cfg.concurrency) + for i := 1; i <= cfg.count; i++ { + create.Go(func() error { + c, err := createOne(createCtx, src, cfg, tag, baseSHA, cfg.base, i, t, rows[i-1]) + if err != nil { + return err + } + created[i-1] = c + return nil + }) + } + if err := create.Wait(); err != nil { + return nil, err + } + + t.Note("enqueuing %d changes at once", cfg.count) + land, landCtx := errgroup.WithContext(ctx) + land.SetLimit(cfg.concurrency) + for i := 1; i <= cfg.count; i++ { + land.Go(func() error { + sqid, err := sq.Land(landCtx, cfg.queue, urisOf([]change{created[i-1]}), strategy) + if err != nil { + return err + } + t.Update(func() { rows[i-1].SQID, rows[i-1].Submitted = sqid, time.Now() }) + return nil + }) + } + if err := land.Wait(); err != nil { + return nil, err + } + return created, nil +} + // createStack creates the changes one after another, each based on the one // before it, and submits the whole chain as a single request. // @@ -530,7 +602,7 @@ func createIndependent( func createStack( ctx context.Context, src changeSource, - sq *client.Client, + sq lander, cfg config, strategy mergestrategypb.Strategy, tag, baseSHA string, diff --git a/service/submitqueue/demo/requests/main_test.go b/service/submitqueue/demo/requests/main_test.go index 48a85615..7d0a36c8 100644 --- a/service/submitqueue/demo/requests/main_test.go +++ b/service/submitqueue/demo/requests/main_test.go @@ -15,12 +15,18 @@ package main import ( + "context" "fmt" "strings" + "sync" + "sync/atomic" "testing" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + + mergestrategypb "github.com/uber/submitqueue/api/base/mergestrategy/protopb" + "github.com/uber/submitqueue/submitqueue/client" ) func TestChangeFilePath_IsUniquePerFileAcrossChangesAndRuns(t *testing.T) { @@ -206,4 +212,62 @@ func TestShapeReportsConcurrency(t *testing.T) { "one at a time is just sequential; saying so adds nothing") assert.Contains(t, shape(config{count: 10, concurrency: 5, stacked: true}), "stacked", "a stack is sequential whatever the limit says") + assert.Contains(t, shape(config{count: 10, concurrency: 5, land: true, burst: true}), "all enqueued at once") + assert.NotContains(t, shape(config{count: 10, concurrency: 5, burst: true, land: false}), "at once", + "burst is a landing decision; with nothing to land there is no burst") +} + +// recordingSource counts how many changes have been created, so a test can +// observe creation relative to enqueuing. It creates through fakeSource, so it +// stands in for any provider — the two-phase orchestration under test is the +// same whichever one is wired. +type recordingSource struct { + created *int64 +} + +func (recordingSource) baseSHA(context.Context, string) (string, error) { return "base", nil } + +func (r recordingSource) open(ctx context.Context, spec changeSpec) (openedChange, error) { + atomic.AddInt64(r.created, 1) + return fakeSource{}.open(ctx, spec) +} + +// landerFunc adapts a function to the lander interface. +type landerFunc func(context.Context, string, []string, mergestrategypb.Strategy) (string, error) + +func (f landerFunc) Land(ctx context.Context, queue string, uris []string, s mergestrategypb.Strategy) (string, error) { + return f(ctx, queue, uris, s) +} + +// TestCreateBurst_EnqueuesOnlyAfterEveryChangeIsCreated is the property -burst +// exists for: no request reaches the queue until the last change has been +// created, so they all arrive together. It is provider-independent — burst runs +// every source through the same two phases. +func TestCreateBurst_EnqueuesOnlyAfterEveryChangeIsCreated(t *testing.T) { + const count = 8 + cfg := config{count: count, concurrency: 4, files: 3, folders: 3, land: true, burst: true, queue: "q", prefix: "demo"} + + var created int64 + src := recordingSource{created: &created} + + var mu sync.Mutex + createdWhenLanded := make([]int64, 0, count) + lander := landerFunc(func(context.Context, string, []string, mergestrategypb.Strategy) (string, error) { + mu.Lock() + createdWhenLanded = append(createdWhenLanded, atomic.LoadInt64(&created)) + mu.Unlock() + return "sqid", nil + }) + + tracker := client.NewTracker(client.NewRows(count)) + got, err := createBurst(context.Background(), src, lander, cfg, + mergestrategypb.Strategy_SQUASH_REBASE, "run", "base", tracker) + require.NoError(t, err) + require.Len(t, got, count) + + require.Len(t, createdWhenLanded, count, "every change is enqueued exactly once") + for _, seen := range createdWhenLanded { + assert.Equal(t, int64(count), seen, + "burst enqueues nothing until every change has been created") + } }