refactor(request-log): address review on gateway sole-owner PR

- Make the log-consumer subscriber name unique per instance (hostname+PID)
  so co-located gateway processes don't contend for the same partition lease.
- Report the gRPC server error first in the shutdown errors.Join (it is the
  primary failure; consumer-stop is secondary cleanup).
- Clarify in the README that the gateway is the sole owner (writer and reader)
  of the request log; Status/Cancel read directly, orchestrator only publishes.
- Extract named poll constants (persistTimeout/persistPollInterval) in the
  gateway integration and e2e suites with a comment explaining that the
  in-container consumer is observed black-box via Status, so a bounded poll is
  used in lieu of an in-process channel/HookSignal wait.

Follow-ups split out: design doc (#211) and DLQ PublishLog() (#212).

Co-Authored-By: Oz <oz-agent@warp.dev>

Author: behinddwalls

example/submitqueue/gateway/server/main.go

@@ -178,12 +178,18 @@ func run() error {
 		zap.String("queue_dsn", queueDSN),
 	)
 
-	// 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())
+	// Subscriber name for the log-topic consumer. It must be unique per running
+	// instance: SubscriberName identifies a subscriber for partition leases, so
+	// two gateway processes on the same host (sharing HOSTNAME) would otherwise
+	// contend for the same lease. Append the PID to keep co-located instances
+	// distinct; the PID is stable for the life of the process. Offset tracking
+	// stays keyed on the shared ConsumerGroup ("gateway-log"), not this name.
+	// Falls back to a time-seeded name when HOSTNAME is unset (e.g. local runs).
+	hostname := os.Getenv("HOSTNAME")
+	if hostname == "" {
+		hostname = fmt.Sprintf("gateway-%d", time.Now().Unix())
 	}
+	subscriberName := fmt.Sprintf("%s-%d", hostname, os.Getpid())
 
 	// Build the topic registry. The gateway publishes to the start of the
 	// orchestrator pipeline (TopicKeyStart) and the cancel topic (TopicKeyCancel) —
@@ -337,8 +343,10 @@ func run() error {
 	}
 
 	if errStop != nil || serverErr != nil {
-		// Override context cancellation error with the shutdown error
-		err = errors.Join(errStop, serverErr)
+		// Override context cancellation error with the shutdown error. The server
+		// error is the primary/root failure, so it leads; the consumer-stop error
+		// is secondary cleanup.
+		err = errors.Join(serverErr, errStop)
 	}
 
 	return err

submitqueue/gateway/README.md

@@ -6,8 +6,9 @@ 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:
+The gateway is the **sole owner of the request log** — the only service that
+both writes and reads it. No other service persists or reads 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
@@ -16,5 +17,9 @@ request log entries:
   the `log` topic via `submitqueue/core/request.PublishLog`. The gateway runs a
   consumer that drains the `log` topic and persists each entry to storage.
 
+Reads are likewise gateway-only: the `Status` and `Cancel` RPCs read the request
+log directly from storage. The orchestrator only *publishes* log entries and
+never touches the request log store.
+
 This keeps a single service responsible for the request log while letting the
 orchestrator remain free of storage writes for it.

test/e2e/submitqueue/suite_test.go

@@ -56,6 +56,17 @@ func TestE2EIntegration(t *testing.T) {
 	suite.Run(t, new(E2EIntegrationSuite))
 }
 
+// The gateway log consumer runs inside the gateway-service container, so this
+// suite can only observe persistence black-box through the Status RPC — there is
+// no in-process channel/HookSignal to wait on across the container boundary. A
+// bounded poll is therefore the deterministic-enough analog: persistTimeout is a
+// safety net (a failure here means something is genuinely stuck, not a timing
+// race), and persistPollInterval bounds how often we re-query.
+const (
+	persistTimeout      = 30 * time.Second
+	persistPollInterval = 500 * time.Millisecond
+)
+
 func (s *E2EIntegrationSuite) SetupSuite() {
 	t := s.T()
 	s.ctx = context.Background()
@@ -203,7 +214,7 @@ func (s *E2EIntegrationSuite) TestLandRequest_PersistsStartedLogViaGatewayConsum
 		}
 		s.log.Logf("Status(%s) = %q", sqid, resp.Status)
 		return resp.Status == string(entity.RequestStatusStarted)
-	}, 30*time.Second, 500*time.Millisecond,
+	}, persistTimeout, persistPollInterval,
 		"request %s should reach status %q via the gateway log consumer", sqid, entity.RequestStatusStarted)
 
 	s.log.Logf("Gateway consumer persisted orchestrator-published 'started' log for sqid=%s", sqid)

test/integration/submitqueue/gateway/suite_test.go

@@ -60,6 +60,17 @@ func TestGatewayIntegration(t *testing.T) {
 	suite.Run(t, new(GatewayIntegrationSuite))
 }
 
+// The log consumer runs inside the gateway-service container, so this suite can
+// only observe persistence black-box through the Status RPC — there is no
+// in-process channel/HookSignal to wait on across the container boundary. A
+// bounded poll is therefore the deterministic-enough analog: persistTimeout is a
+// safety net (a failure here means something is genuinely stuck, not a timing
+// race), and persistPollInterval bounds how often we re-query.
+const (
+	persistTimeout      = 30 * time.Second
+	persistPollInterval = 500 * time.Millisecond
+)
+
 func (s *GatewayIntegrationSuite) SetupSuite() {
 	t := s.T()
 	s.ctx = context.Background()
@@ -190,7 +201,7 @@ func (s *GatewayIntegrationSuite) TestRequestLogConsumer() {
 			return false
 		}
 		return resp.Status == string(entity.RequestStatusStarted)
-	}, 30*time.Second, 500*time.Millisecond,
+	}, persistTimeout, persistPollInterval,
 		"gateway log consumer should persist the published request log for sqid=%s", sqid)
 
 	s.log.Logf("Request log consumer test passed: entry persisted and readable via Status")