refactor(request-log): gateway is sole writer of request log

## Summary

### Why?

The request log had two persistence paths: the gateway wrote some entries
directly, while the orchestrator ran the `log`-topic consumer that wrote all
downstream entries to storage. Having the orchestrator persist the request log
blurs ownership — the orchestrator is a pipeline that should only emit events,
not own the request-log table. Concentrating all writes in the gateway gives a
single owner for the request log and keeps the orchestrator free of request-log
storage writes.

### What?

The request-log persistence consumer moves from the orchestrator to the gateway:

- Move `submitqueue/orchestrator/controller/log/` → `submitqueue/gateway/controller/log/`
  (importpath, doc comment, and default consumer group `orchestrator-log` →
  `gateway-log`). Logic is unchanged.
- Orchestrator: `TopicKeyLog` becomes publish-only (subscription dropped), the
  log controller registration and import are removed, controller count 11 → 10.
  It still publishes via `submitqueue/core/request.PublishLog`.
- Gateway: builds a consumer (generic + mysql classifiers), registers the moved
  log controller on `TopicKeyLog` with a subscription (group `gateway-log`),
  starts it, and drains it with `Stop(30000)` on shutdown — preserving the
  128+SIGTERM graceful-exit contract.
- Add `HOSTNAME=gateway-dev` to both gateway compose files for a stable
  subscriber name; update the workflow RFC and gateway README.
- Tests: add a gateway integration test that publishes to the log topic (as the
  orchestrator does) and asserts the gateway consumer persists it, and an e2e
  test that lands a request and asserts Status advances to `started` —
  exercising the publish→consume→persist path across both services.

The gateway keeps its two synchronous direct writes (`accepted` on Land,
`cancelling` on Cancel) for read-your-write visibility at RPC return. Both are
gateway writes, so the invariant holds: only the gateway persists the request
log; the orchestrator only publishes. This works because gateway and
orchestrator already share the same queue and app databases.

## Test Plan

- ✅ `bazel build` of both servers + the moved package
- ✅ `make test` — unit tests pass (incl. the moved log_test)
- ✅ `make check-gazelle`, `make check-tidy`, `make lint` (fmt + license)
- ✅ `make integration-test-submitqueue-gateway` — new `TestRequestLogConsumer`
  verifies the gateway consumer persists a log entry published to the log topic
- ✅ `make e2e-test` — new `TestLandRequest_PersistsStartedLogViaGatewayConsumer`
  verifies an orchestrator-published `started` log is persisted by the gateway
  and readable via Status; both services still exit 128+SIGTERM on shutdown

Author: behinddwalls

doc/rfc/submitqueue/workflow.md

@@ -2,7 +2,7 @@
 
 The orchestrator processes land requests through a queue-driven pipeline of small, single-purpose controllers. The gateway accepts a request over RPC and hands it off asynchronously; from there each controller consumes one topic, advances the request or batch, and publishes to the next topic. Most hops carry only an ID — the controller fetches the entity from storage — while a few entry points (`start`, `buildsignal`, `log`) carry the full payload because there is no row to fetch yet.
 
-The pipeline has two cycles: `speculate → build → buildsignal → speculate` (CI feedback loop) and `merge → speculate` (advance the next batch). `conclude` is the only stage that transitions a request to a terminal state; `log` is an append-only sink that any controller can publish to via `submitqueue/core/request.PublishLog`.
+The pipeline has two cycles: `speculate → build → buildsignal → speculate` (CI feedback loop) and `merge → speculate` (advance the next batch). `conclude` is the only stage that transitions a request to a terminal state; `log` is an append-only sink that any controller can publish to via `submitqueue/core/request.PublishLog`. The orchestrator only *publishes* request log events — the **gateway** is the sole consumer of the `log` topic and the only service that persists the request log to storage.
 
 ## Diagram
 
@@ -14,8 +14,8 @@ The pipeline has two cycles: `speculate → build → buildsignal → speculate`
                                      │ LandRequest
                                      ▼
    ┌──────────────────────┐    ┌──────────────────────────────────┐
-   │ log  (terminal sink) │◄───│ start                            │
-   │ Append RequestLog    │    │ Persist Request, emit Started    │
+   │ log  (gateway sink)  │◄───│ start                            │
+   │ Persist RequestLog   │    │ Persist Request, emit Started    │
    └──────────────────────┘    └────────────────┬─────────────────┘
               ▲                                 │ RequestID
               │                                 ▼
@@ -79,4 +79,4 @@ The pipeline has two cycles: `speculate → build → buildsignal → speculate`
 | **buildsignal** | Build | speculate | Feed CI result back into speculation |
 | **merge** | BatchID | conclude, speculate | Merge the batch and advance the queue |
 | **conclude** | BatchID | — | Map terminal batch state to request state |
-| **log** | RequestLog | — | Append-only sink for request log events |
+| **log** | RequestLog | — | Gateway-owned sink: persists request log events to storage |

example/submitqueue/docker-compose.yml

@@ -55,6 +55,8 @@ services:
       - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true
       # Path to YAML queue configuration baked into the image
       - QUEUE_CONFIG_PATH=/root/queues.yaml
+      # Stable subscriber name for the request-log consumer
+      - HOSTNAME=gateway-dev
     depends_on:
       mysql-app:
         condition: service_healthy

example/submitqueue/gateway/server/BUILD.bazel

@@ -11,12 +11,16 @@ go_library(
     importpath = "github.com/uber/submitqueue/example/submitqueue/gateway/server",
     visibility = ["//visibility:private"],
     deps = [
+        "//core/errs/generic",
+        "//core/errs/mysql",
         "//extension/counter/mysql",
+        "//extension/messagequeue",
         "//extension/messagequeue/mysql",
         "//submitqueue/core/consumer",
         "//submitqueue/extension/queueconfig/yaml",
         "//submitqueue/extension/storage/mysql",
         "//submitqueue/gateway/controller",
+        "//submitqueue/gateway/controller/log",
         "//submitqueue/gateway/protopb",
         "@com_github_go_sql_driver_mysql//:mysql",
         "@com_github_uber_go_tally_v4//:tally",

example/submitqueue/gateway/server/docker-compose.yml

@@ -55,6 +55,8 @@ services:
       - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true
       # Path to YAML queue configuration baked into the image
       - QUEUE_CONFIG_PATH=/root/queues.yaml
+      # Stable subscriber name for the request-log consumer
+      - HOSTNAME=gateway-dev
     depends_on:
       mysql-app:
         condition: service_healthy

example/submitqueue/gateway/server/main.go

@@ -28,12 +28,16 @@ import (
 
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/uber-go/tally/v4"
+	genericerrs "github.com/uber/submitqueue/core/errs/generic"
+	mysqlerrs "github.com/uber/submitqueue/core/errs/mysql"
 	mysqlcounter "github.com/uber/submitqueue/extension/counter/mysql"
+	extqueue "github.com/uber/submitqueue/extension/messagequeue"
 	queueMySQL "github.com/uber/submitqueue/extension/messagequeue/mysql"
 	"github.com/uber/submitqueue/submitqueue/core/consumer"
 	yamlqueueconfig "github.com/uber/submitqueue/submitqueue/extension/queueconfig/yaml"
 	mysqlstorage "github.com/uber/submitqueue/submitqueue/extension/storage/mysql"
 	"github.com/uber/submitqueue/submitqueue/gateway/controller"
+	logctrl "github.com/uber/submitqueue/submitqueue/gateway/controller/log"
 	pb "github.com/uber/submitqueue/submitqueue/gateway/protopb"
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
@@ -174,12 +178,29 @@ func run() error {
 		zap.String("queue_dsn", queueDSN),
 	)
 
-	// Build a publish-only topic registry: gateway only feeds the start of the
-	// orchestrator pipeline (TopicKeyStart) and the cancel topic (TopicKeyCancel).
-	// No subscription is configured because the gateway never consumes from the queue.
+	// Stable subscriber name for the log-topic consumer. Falls back to a
+	// time-seeded name when HOSTNAME is unset (e.g. local runs).
+	subscriberName := os.Getenv("HOSTNAME")
+	if subscriberName == "" {
+		subscriberName = fmt.Sprintf("gateway-%d", time.Now().Unix())
+	}
+
+	// Build the topic registry. The gateway publishes to the start of the
+	// orchestrator pipeline (TopicKeyStart) and the cancel topic (TopicKeyCancel) —
+	// both publish-only. It additionally consumes the log topic (TopicKeyLog):
+	// the gateway is the sole writer of the request log, persisting entries that
+	// the orchestrator publishes there.
 	registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{
 		{Key: consumer.TopicKeyStart, Name: "start", Queue: mysqlQueue},
 		{Key: consumer.TopicKeyCancel, Name: "cancel", Queue: mysqlQueue},
+		{
+			Key:   consumer.TopicKeyLog,
+			Name:  "log",
+			Queue: mysqlQueue,
+			Subscription: extqueue.DefaultSubscriptionConfig(
+				subscriberName, "gateway-log",
+			),
+		},
 	})
 	if err != nil {
 		return fmt.Errorf("failed to create topic registry: %w", err)
@@ -201,7 +222,8 @@ func run() error {
 
 	// Initialize storage from the shared app database connection. The land
 	// controller writes to this store directly; cancel/status use the request
-	// log store directly.
+	// log store directly. The log consumer (registered below) is the sole
+	// persister of request log entries published by the orchestrator.
 	store, err := mysqlstorage.NewStorage(appDB, scope.SubScope("storage"))
 	if err != nil {
 		return fmt.Errorf("failed to create storage: %w", err)
@@ -236,6 +258,29 @@ func run() error {
 	// Register reflection service for debugging with grpcurl
 	reflection.Register(grpcServer)
 
+	// Create the queue consumer and register the log controller. The gateway is
+	// the sole persister of the request log: the orchestrator publishes entries
+	// to the log topic and this consumer writes them to storage.
+	logConsumer := consumer.New(logger.Sugar(), scope.SubScope("consumer"), registry,
+		// Storage (extension/storage/mysql) and queue (extension/messagequeue/mysql)
+		// both run on the same MySQL driver, so a single classifier covers
+		// errors surfaced from either backend.
+		genericerrs.Classifier,
+		mysqlerrs.Classifier,
+	)
+
+	logController := logctrl.NewController(logger.Sugar(), scope, store, consumer.TopicKeyLog, "gateway-log")
+	if err := logConsumer.Register(logController); err != nil {
+		return fmt.Errorf("failed to register log controller: %w", err)
+	}
+
+	if err := logConsumer.Start(ctx); err != nil {
+		// The error can also be a result of a context cancellation due to SIGINT or SIGTERM.
+		// This is expected, just propagate it.
+		return fmt.Errorf("failed to start log consumer: %w", err)
+	}
+	logger.Info("log consumer started")
+
 	// Listen on configurable port
 	port := os.Getenv("PORT")
 	if port == "" {
@@ -257,6 +302,8 @@ func run() error {
 
 	// Wait for interrupt signal or server critical error
 	// If interruption is signaled, gracefully stop the server
+	// If the server exits with an error, cancel the context to signal the consumer
+	// After this, stop the consumer
 	// If an error happens during shutdown, return the actual error, not the context cancellation error
 	var serverErr error
 	select {
@@ -273,10 +320,25 @@ func run() error {
 		serverErr = <-serverErrCh
 	case serverErr = <-serverErrCh:
 		fmt.Println("Shutting down gateway server due to critical GRPC server error...")
+
+		// Cancel the context to signal cancellation to the queue consumer
+		cancel()
 	}
 
 	if serverErr != nil {
-		err = fmt.Errorf("GRPC server exited with error: %w", serverErr)
+		serverErr = fmt.Errorf("GRPC server exited with error: %w", serverErr)
+	}
+
+	// Stop the consumer with a 30s timeout; by this time the context should be
+	// cancelled and the processing threads may already be exiting; recollect them.
+	errStop := logConsumer.Stop(30000)
+	if errStop != nil {
+		errStop = fmt.Errorf("failed to stop consumer: %w", errStop)
+	}
+
+	if errStop != nil || serverErr != nil {
+		// Override context cancellation error with the shutdown error
+		err = errors.Join(errStop, serverErr)
 	}
 
 	return err

example/submitqueue/orchestrator/server/BUILD.bazel

@@ -42,7 +42,6 @@ go_library(
         "//submitqueue/orchestrator/controller/buildsignal",
         "//submitqueue/orchestrator/controller/cancel",
         "//submitqueue/orchestrator/controller/conclude",
-        "//submitqueue/orchestrator/controller/log",
         "//submitqueue/orchestrator/controller/merge",
         "//submitqueue/orchestrator/controller/score",
         "//submitqueue/orchestrator/controller/speculate",

example/submitqueue/orchestrator/server/main.go

@@ -61,7 +61,6 @@ import (
 	"github.com/uber/submitqueue/submitqueue/orchestrator/controller/buildsignal"
 	"github.com/uber/submitqueue/submitqueue/orchestrator/controller/cancel"
 	"github.com/uber/submitqueue/submitqueue/orchestrator/controller/conclude"
-	logctrl "github.com/uber/submitqueue/submitqueue/orchestrator/controller/log"
 	"github.com/uber/submitqueue/submitqueue/orchestrator/controller/merge"
 	"github.com/uber/submitqueue/submitqueue/orchestrator/controller/score"
 	"github.com/uber/submitqueue/submitqueue/orchestrator/controller/speculate"
@@ -238,7 +237,7 @@ func run() error {
 		return err
 	}
 
-	logger.Info("controllers registered", zap.Int("count", 11))
+	logger.Info("controllers registered", zap.Int("count", 10))
 
 	// Start consumers
 	if err := c.Start(ctx); err != nil {
@@ -408,12 +407,12 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 			),
 		},
 		{
+			// Publish-only: the orchestrator emits request log entries to the
+			// log topic but never persists them. The gateway is the sole
+			// consumer that writes the request log to storage.
 			Key:   consumer.TopicKeyLog,
 			Name:  "log",
 			Queue: q,
-			Subscription: extqueue.DefaultSubscriptionConfig(
-				subscriberName, "orchestrator-log",
-			),
 		},
 	})
 }
@@ -605,17 +604,6 @@ func registerControllers(c consumer.Consumer, logger *zap.SugaredLogger, scope t
 		return fmt.Errorf("failed to register conclude controller: %w", err)
 	}
 
-	logController := logctrl.NewController(
-		logger,
-		scope,
-		store,
-		consumer.TopicKeyLog,
-		"orchestrator-log",
-	)
-	if err := c.Register(logController); err != nil {
-		return fmt.Errorf("failed to register log controller: %w", err)
-	}
-
 	return nil
 }
 

submitqueue/gateway/README.md

@@ -1 +1,20 @@
-SubmitQueue Gateway
+# SubmitQueue Gateway
+
+The gateway is the RPC entry point to SubmitQueue. It accepts `Land`, `Cancel`,
+`Status`, and `Ping` calls, validates them at the edge, and hands work off to the
+orchestrator pipeline asynchronously via the message queue.
+
+## Request log ownership
+
+The gateway is the **sole writer of the request log**. No other service persists
+request log entries:
+
+- For statuses it produces synchronously (`accepted` on `Land`, `cancelling` on
+  `Cancel`), the gateway writes directly to storage so the entry is visible the
+  moment the RPC returns.
+- For statuses produced downstream, the orchestrator only *publishes* entries to
+  the `log` topic via `submitqueue/core/request.PublishLog`. The gateway runs a
+  consumer that drains the `log` topic and persists each entry to storage.
+
+This keeps a single service responsible for the request log while letting the
+orchestrator remain free of storage writes for it.

…/orchestrator/controller/log/BUILD.bazel …queue/gateway/controller/log/BUILD.bazelsubmitqueue/orchestrator/controller/log/BUILD.bazel renamed to submitqueue/gateway/controller/log/BUILD.bazel submitqueue/orchestrator/controller/log/BUILD.bazel renamed to submitqueue/gateway/controller/log/BUILD.bazel

@@ -3,7 +3,7 @@ load("@rules_go//go:def.bzl", "go_library", "go_test")
 go_library(
     name = "log",
     srcs = ["log.go"],
-    importpath = "github.com/uber/submitqueue/submitqueue/orchestrator/controller/log",
+    importpath = "github.com/uber/submitqueue/submitqueue/gateway/controller/log",
     visibility = ["//visibility:public"],
     deps = [
         "//core/metrics",

…queue/orchestrator/controller/log/log.go submitqueue/gateway/controller/log/log.gosubmitqueue/orchestrator/controller/log/log.go renamed to submitqueue/gateway/controller/log/log.go submitqueue/orchestrator/controller/log/log.go renamed to submitqueue/gateway/controller/log/log.go

@@ -29,6 +29,10 @@ import (
 // Controller handles log queue messages.
 // It consumes request log entries and persists them to storage.
 // Implements consumer.Controller interface for integration with the consumer.
+//
+// The request log is written exclusively by the gateway: other services
+// (e.g. the orchestrator) only publish log entries to the log topic, and this
+// controller is the single consumer that persists them to storage.
 type Controller struct {
 	logger        *zap.SugaredLogger
 	metricsScope  tally.Scope
@@ -40,7 +44,7 @@ type Controller struct {
 // Verify Controller implements consumer.Controller interface at compile time.
 var _ consumer.Controller = (*Controller)(nil)
 
-// NewController creates a new log controller for the orchestrator.
+// NewController creates a new log controller for the gateway.
 func NewController(
 	logger *zap.SugaredLogger,
 	scope tally.Scope,