feat(demo): add BURST mode for a simultaneous enqueue of independent changes (#615)

## Summary
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.

## Test Plan

✅ New unit test asserts nothing is enqueued until every change is
created, provider-agnostically (a `lander` seam + a recording source
stand in for any provider); shape banner covered; existing
`//service/submitqueue/demo/requests` tests green.

## Stack
1. @ #615
1. #616

Author: behinddwalls

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) \

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/git/exec:go_default_library",
         "//platform/git/exectest:go_default_library",
+        "//submitqueue/client:go_default_library",
         "@com_github_stretchr_testify//assert:go_default_library",
         "@com_github_stretchr_testify//require:go_default_library",
     ],

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,

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")
+	}
 }