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

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/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",
     ],

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