Merge branch 'main' into fix/busy-partition-gc

Author: Jal-Bafana

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

platform/publish/publish.go

@@ -47,6 +47,13 @@ import (
 // which is what makes redelivery safe, while a new cause about the same entity
 // can never be swallowed by an older row.
 func Message(ctx context.Context, registry consumer.TopicRegistry, key consumer.TopicKey, msgID string, payload []byte, partitionKey string) error {
+	return MessageWithMetadata(ctx, registry, key, msgID, payload, partitionKey, nil)
+}
+
+// MessageWithMetadata is Message with side-band message metadata (headers/attributes)
+// attached to the delivery. Use it to carry diagnostic context that is not part of
+// the payload — the backend persists and redelivers metadata alongside the message.
+func MessageWithMetadata(ctx context.Context, registry consumer.TopicRegistry, key consumer.TopicKey, msgID string, payload []byte, partitionKey string, metadata map[string]string) error {
 	q, ok := registry.Queue(key)
 	if !ok {
 		return fmt.Errorf("no queue registered for topic key %s", key)
@@ -56,7 +63,7 @@ func Message(ctx context.Context, registry consumer.TopicRegistry, key consumer.
 		return fmt.Errorf("no topic name registered for topic key %s", key)
 	}
 
-	msg := entityqueue.NewMessage(msgID, payload, partitionKey, nil)
+	msg := entityqueue.NewMessage(msgID, payload, partitionKey, metadata)
 	return q.Publisher().Publish(ctx, topicName, msg)
 }
 

service/runway/server/BUILD.bazel

@@ -28,6 +28,7 @@ go_library(
         "//platform/extension/consumergate/noop:go_default_library",
         "//platform/extension/messagequeue:go_default_library",
         "//platform/extension/messagequeue/mysql:go_default_library",
+        "//platform/git/exec:go_default_library",
         "//runway/controller:go_default_library",
         "//runway/controller/dlq:go_default_library",
         "//runway/controller/merge:go_default_library",

service/runway/server/checkout.go

@@ -26,6 +26,7 @@ import (
 
 	"go.uber.org/zap"
 
+	gitexec "github.com/uber/submitqueue/platform/git/exec"
 	gitmerger "github.com/uber/submitqueue/runway/extension/merger/git"
 )
 
@@ -190,8 +191,9 @@ func setLocalConfig(checkoutPath, key, value string) error {
 }
 
 // runGit invokes the pinned git in dir with an environment scrubbed of ambient
-// configuration but retaining what is needed to reach a remote — the same split
-// the merger draws, so provisioning and merging authenticate identically.
+// configuration but retaining what is needed to reach a remote. It composes that
+// environment through gitexec.Env, the same source the merger uses, so
+// provisioning and merging authenticate — and behave — identically.
 func runGit(ctx context.Context, runtime gitmerger.GitRuntime, dir string, args ...string) ([]byte, error) {
 	full := append([]string{
 		"--exec-path=" + runtime.ExecPath,
@@ -200,27 +202,17 @@ func runGit(ctx context.Context, runtime gitmerger.GitRuntime, dir string, args
 
 	cmd := exec.CommandContext(ctx, runtime.Executable, full...)
 	cmd.Dir = dir
-	cmd.Env = []string{
-		"HOME=" + filepath.Join(dir, ".submitqueue-git-home"),
-		"GIT_CONFIG_NOSYSTEM=1",
-		"GIT_CONFIG_GLOBAL=" + os.DevNull,
-		"GIT_TERMINAL_PROMPT=0",
-		"GIT_EXEC_PATH=" + runtime.ExecPath,
-		"GIT_TEMPLATE_DIR=" + runtime.TemplateDir,
-		"LC_ALL=C",
-		"LANG=C",
-	}
-	for _, name := range []string{
-		"PATH", "SSH_AUTH_SOCK", "SSH_AGENT_PID",
-		"GIT_SSH", "GIT_SSH_COMMAND", "GIT_SSH_VARIANT",
-		"GIT_SSL_CAINFO", "GIT_SSL_CAPATH", "SSL_CERT_DIR", "SSL_CERT_FILE",
-		"HTTP_PROXY", "HTTPS_PROXY", "NO_PROXY",
-		"http_proxy", "https_proxy", "no_proxy",
-	} {
-		if v, ok := os.LookupEnv(name); ok {
-			cmd.Env = append(cmd.Env, name+"="+v)
-		}
-	}
+	cmd.Env = gitexec.Env(gitexec.EnvOptions{
+		Transport:   true,
+		Passthrough: runtime.PassthroughEnv,
+		Literal: []string{
+			"HOME=" + filepath.Join(dir, ".submitqueue-git-home"),
+			"GIT_EXEC_PATH=" + runtime.ExecPath,
+			"GIT_TEMPLATE_DIR=" + runtime.TemplateDir,
+			"LC_ALL=C",
+			"LANG=C",
+		},
+	})
 
 	var stdout, stderr bytes.Buffer
 	cmd.Stdout = &stdout

service/stovepipe/server/main.go

@@ -282,15 +282,15 @@ func run() error {
 	// Each factory is constructed once and threaded through every consumer of
 	// it, so a real (stateful) backend introduced later is shared rather than
 	// silently duplicated across controllers.
-	scf := fakeSourceControlFactory{}
+	sourceControl := fakeSourceControlFactory{}
 	brf := fakeBuildRunnerFactory{}
 
 	storageFty := storageFactory{backend: store}
-	primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, storageFty, registry, scf, brf)
+	primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, storageFty, registry, sourceControl, brf)
 	if err != nil {
 		return err
 	}
-	dlqCount, err := registerDLQControllers(dlqConsumer, logger.Sugar(), scope, storageFty, registry)
+	dlqCount, err := registerDLQControllers(dlqConsumer, logger.Sugar(), scope, storageFty, registry, sourceControl)
 	if err != nil {
 		return err
 	}
@@ -318,7 +318,7 @@ func run() error {
 		logger.Sugar(),
 		scope,
 		newInMemoryCounterFactory(),
-		scf,
+		sourceControl,
 		storageFty,
 		registry,
 	)
@@ -398,7 +398,7 @@ func registerPrimaryControllers(
 	scope tally.Scope,
 	store storage.Factory,
 	registry consumer.TopicRegistry,
-	scf sourcecontrol.Factory,
+	sourceControl sourcecontrol.Factory,
 	brf buildrunner.Factory,
 ) (int, error) {
 	var count int
@@ -408,7 +408,7 @@ func registerPrimaryControllers(
 		scope,
 		store,
 		queueconfigdefault.NewStore(),
-		scf,
+		sourceControl,
 		registry,
 		stovepipemq.TopicKeyProcess,
 		"stovepipe-process",
@@ -430,7 +430,7 @@ func registerPrimaryControllers(
 	}
 	count++
 
-	recordController := record.NewController(logger, scope, store, scf, stovepipemq.TopicKeyRecord, "stovepipe-record")
+	recordController := record.NewController(logger, scope, store, sourceControl, stovepipemq.TopicKeyRecord, "stovepipe-record")
 	if err := c.Register(recordController); err != nil {
 		return count, fmt.Errorf("failed to register record controller: %w", err)
 	}
@@ -447,21 +447,34 @@ func registerDLQControllers(
 	scope tally.Scope,
 	store storage.Factory,
 	registry consumer.TopicRegistry,
+	sourceControl sourcecontrol.Factory,
 ) (int, error) {
 	var count int
 
-	processDLQController := dlq.NewController(logger, scope, store, dlq.TopicKey(stovepipemq.TopicKeyProcess), "stovepipe-process-dlq")
+	processDLQController := dlq.NewDLQRequestController(logger, scope, store, dlq.TopicKey(stovepipemq.TopicKeyProcess), "stovepipe-process-dlq")
 	if err := c.Register(processDLQController); err != nil {
 		return count, fmt.Errorf("failed to register process dlq controller: %w", err)
 	}
 	count++
 
-	buildSignalDLQController := dlq.NewBuildSignalController(logger, scope, store, dlq.TopicKey(stovepipemq.TopicKeyBuildSignal), "stovepipe-buildsignal-dlq")
+	buildDLQController := dlq.NewDLQBuildController(logger, scope, store, dlq.TopicKey(stovepipemq.TopicKeyBuild), "stovepipe-build-dlq")
+	if err := c.Register(buildDLQController); err != nil {
+		return count, fmt.Errorf("failed to register build dlq controller: %w", err)
+	}
+	count++
+
+	buildSignalDLQController := dlq.NewDLQBuildSignalController(logger, scope, store, dlq.TopicKey(stovepipemq.TopicKeyBuildSignal), "stovepipe-buildsignal-dlq")
 	if err := c.Register(buildSignalDLQController); err != nil {
 		return count, fmt.Errorf("failed to register buildsignal dlq controller: %w", err)
 	}
 	count++
 
+	recordDLQController := record.NewController(logger, scope, store, sourceControl, dlq.TopicKey(stovepipemq.TopicKeyRecord), "stovepipe-record-dlq")
+	if err := c.Register(recordDLQController); err != nil {
+		return count, fmt.Errorf("failed to register record dlq controller: %w", err)
+	}
+	count++
+
 	return count, nil
 }
 
@@ -511,12 +524,24 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 			Queue:        q,
 			Subscription: extqueue.DLQSubscriptionConfig(subscriberName, "stovepipe-process-dlq"),
 		},
+		{
+			Key:          dlq.TopicKey(stovepipemq.TopicKeyBuild),
+			Name:         "build_dlq",
+			Queue:        q,
+			Subscription: extqueue.DLQSubscriptionConfig(subscriberName, "stovepipe-build-dlq"),
+		},
 		{
 			Key:          dlq.TopicKey(stovepipemq.TopicKeyBuildSignal),
 			Name:         "buildsignal_dlq",
 			Queue:        q,
 			Subscription: extqueue.DLQSubscriptionConfig(subscriberName, "stovepipe-buildsignal-dlq"),
 		},
+		{
+			Key:          dlq.TopicKey(stovepipemq.TopicKeyRecord),
+			Name:         "record_dlq",
+			Queue:        q,
+			Subscription: extqueue.DLQSubscriptionConfig(subscriberName, "stovepipe-record-dlq"),
+		},
 	})
 }
 

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,