llm-d · wseaton · Jul 8, 2026 · Jul 8, 2026 · Jul 8, 2026 · Jul 8, 2026
@@ -19,6 +19,22 @@ jobs:
     env:
       FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: true
 
+    # Real Postgres for the exchange integration tests (they skip without it).
+    services:
+      postgres:
+        image: postgres:16-alpine
+        env:
+          POSTGRES_USER: it
+          POSTGRES_PASSWORD: it
+          POSTGRES_DB: it
+        ports:
+          - 15432:5432
+        options: >-
+          --health-cmd "pg_isready -U it"
+          --health-interval 5s
+          --health-timeout 5s
+          --health-retries 10
+
     steps:
       - uses: actions/checkout@v7
 
@@ -43,6 +59,8 @@ jobs:
         uses: pre-commit/action@v3.0.1
 
       - name: Run integration tests
+        env:
+          POSTGRES_TEST_URL: postgres://it:it@localhost:15432/it?sslmode=disable
         run: make test-integration
 
       - name: Lint Helm chart

@@ -1,4 +1,4 @@
-.PHONY: help build build-apiserver build-processor build-gc run-apiserver run-processor run-gc run-apiserver-dev run-processor-dev run-gc-dev build-release package-release publish-helm-chart generate-release test test-coverage test-coverage-func clean lint fmt vet tidy install-tools deps-get deps-verify bench check check-container-tool ci image-build image-build-apiserver image-build-processor image-build-gc test-regression test-integration test-all test-e2e test-helm dev-deploy dev-clean dev-rm-cluster pre-commit benchmark-local benchmark-local-teardown benchmark-gpu benchmark-gpu-teardown
+.PHONY: help build build-apiserver build-processor build-gc run-apiserver run-processor run-gc run-apiserver-dev run-processor-dev run-gc-dev build-release package-release publish-helm-chart generate-release test test-coverage test-coverage-func clean lint fmt vet tidy install-tools deps-get deps-verify bench check check-container-tool ci image-build image-build-apiserver image-build-processor image-build-gc test-regression test-integration test-integration-postgres test-all test-e2e test-helm dev-deploy dev-clean dev-rm-cluster pre-commit benchmark-local benchmark-local-teardown benchmark-gpu benchmark-gpu-teardown
 
 SHELL := /usr/bin/env bash
 
@@ -330,6 +330,23 @@ test-integration:
 		(echo "\n❌ Integration tests failed" && exit 1)
 	@echo "\n✅ Integration tests passed!"
 
+## test-integration-postgres: Run Postgres exchange integration tests against a throwaway local Postgres container
+test-integration-postgres: check-container-tool
+	@echo "Starting throwaway Postgres container..."
+	@$(CONTAINER_TOOL) rm -f batch-gateway-it-postgres >/dev/null 2>&1 || true
+	@$(CONTAINER_TOOL) run -d --name batch-gateway-it-postgres \
+		-e POSTGRES_USER=it -e POSTGRES_PASSWORD=it -e POSTGRES_DB=it \
+		-p 15432:5432 postgres:16-alpine >/dev/null
+	@until $(CONTAINER_TOOL) exec batch-gateway-it-postgres pg_isready -U it >/dev/null 2>&1; do sleep 0.5; done
+	@sleep 1
+	@echo "Running Postgres exchange integration tests..."
+	@POSTGRES_TEST_URL="postgres://it:it@localhost:15432/it?sslmode=disable" \
+		$(GO) test -v -tags=integration -count=1 ./test/integration/ -run TestPostgresExchange; \
+		rc=$$?; \
+		$(CONTAINER_TOOL) rm -f batch-gateway-it-postgres >/dev/null 2>&1; \
+		if [ $$rc -ne 0 ]; then echo "\n❌ Postgres exchange integration tests failed" && exit $$rc; fi
+	@echo "\n✅ Postgres exchange integration tests passed!"
+
 ## test-all: Run all tests (unit + regression + integration)
 test-all: test test-regression test-integration
 

@@ -0,0 +1,13 @@
+# Exercises: postgres-only mode (no Redis).
+# Both the persistent store (dbClient) and the runtime exchange (exchangeClient)
+# use PostgreSQL, so the deployment needs only the postgresql-url secret key — no
+# Redis/Valkey at all. Async dispatch is intentionally left unset (llm-d-async is
+# redis-only and rejected by the exchange validation in this mode).
+global:
+  dbClient:
+    type: "postgresql"
+  exchangeClient:
+    type: "postgresql"
+  fileClient:
+    fs:
+      pvcName: "batch-data"
@@ -69,6 +69,29 @@ Usage: {{ include "batch-gateway.validateTLS" (dict "tls" .Values.apiserver.tls
 {{- end -}}
 {{- end -}}
 
+{{/* ========== Exchange Client Validation Helper ========== */}}
+
+{{/*
+Validate exchange client configuration against the db client and dispatch mode.
+Mirrors the startup validation in internal/util/clientset/clientset.go so misconfig
+fails fast at `helm template`/`helm install` time instead of at pod startup.
+Rules (only enforced when exchangeClient.type is postgresql):
+  - dbClient.type must also be postgresql (the pg exchange reuses the db connection).
+  - async dispatch is unsupported (llm-d-async is redis-only).
+Usage: {{ include "batch-gateway.validateExchange" . }}
+*/}}
+{{- define "batch-gateway.validateExchange" -}}
+{{- $exchangeType := .Values.global.exchangeClient.type | default "redis" -}}
+{{- if eq $exchangeType "postgresql" -}}
+  {{- if ne .Values.global.dbClient.type "postgresql" -}}
+    {{- fail "global.exchangeClient.type=postgresql requires global.dbClient.type=postgresql" -}}
+  {{- end -}}
+  {{- if eq (.Values.processor.config.dispatchMode | default "") "async" -}}
+    {{- fail "global.exchangeClient.type=postgresql does not support async dispatch; use sync dispatch or a redis exchange" -}}
+  {{- end -}}
+{{- end -}}
+{{- end -}}
+
 {{/* ========== API Server Helpers ========== */}}
 
 {{/*

@@ -1,4 +1,5 @@
 {{- if .Values.apiserver.enabled -}}
+{{- include "batch-gateway.validateExchange" . -}}
 apiVersion: v1
 kind: ConfigMap
 metadata:
@@ -36,6 +37,8 @@ data:
         pool_timeout: {{ .Values.global.dbClient.redis.poolTimeout | quote }}
         conn_max_idle_time: {{ .Values.global.dbClient.redis.connMaxIdleTime | quote }}
         conn_max_lifetime: {{ .Values.global.dbClient.redis.connMaxLifetime | quote }}
+    exchange_client:
+      type: {{ .Values.global.exchangeClient.type | quote }}
     file_client:
       type: {{ .Values.global.fileClient.type | quote }}
       fs:

@@ -1,4 +1,5 @@
 {{- if .Values.gc.enabled -}}
+{{- include "batch-gateway.validateExchange" . -}}
 apiVersion: v1
 kind: ConfigMap
 metadata:
@@ -26,6 +27,8 @@ data:
         pool_timeout: {{ .Values.global.dbClient.redis.poolTimeout | quote }}
         conn_max_idle_time: {{ .Values.global.dbClient.redis.connMaxIdleTime | quote }}
         conn_max_lifetime: {{ .Values.global.dbClient.redis.connMaxLifetime | quote }}
+    exchange_client:
+      type: {{ .Values.global.exchangeClient.type | quote }}
     file_client:
       type: {{ .Values.global.fileClient.type | quote }}
       fs:

@@ -1,4 +1,5 @@
 {{- if .Values.processor.enabled -}}
+{{- include "batch-gateway.validateExchange" . -}}
 apiVersion: v1
 kind: ConfigMap
 metadata:
@@ -42,6 +43,9 @@ data:
         conn_max_idle_time: {{ .Values.global.dbClient.redis.connMaxIdleTime | quote }}
         conn_max_lifetime: {{ .Values.global.dbClient.redis.connMaxLifetime | quote }}
 
+    exchange_client:
+      type: {{ .Values.global.exchangeClient.type | quote }}
+
     {{- if .Values.processor.config.dispatchMode }}
     dispatch_mode: {{ .Values.processor.config.dispatchMode | quote }}
     {{- end }}

@@ -21,6 +21,26 @@ tests:
       - matchRegex:
           path: data["config.yaml"]
           pattern: 'db_client:\n\s+type: "postgresql"'
+      - matchRegex:
+          path: data["config.yaml"]
+          pattern: 'exchange_client:\n\s+type: "redis"'
+
+  - it: should render postgresql exchange_client when set
+    set:
+      global.dbClient.type: "postgresql"
+      global.exchangeClient.type: "postgresql"
+    asserts:
+      - matchRegex:
+          path: data["config.yaml"]
+          pattern: 'exchange_client:\n\s+type: "postgresql"'
+
+  - it: should fail when postgresql exchange is used with a non-postgresql db
+    set:
+      global.dbClient.type: "redis"
+      global.exchangeClient.type: "postgresql"
+    asserts:
+      - failedTemplate:
+          errorMessage: "global.exchangeClient.type=postgresql requires global.dbClient.type=postgresql"
 
   - it: should render inputHeaders when set
     asserts:

@@ -67,6 +67,29 @@ tests:
           path: data["config.yaml"]
           pattern: 'base_path: "/tmp/batch-gateway"'
 
+  - it: should render exchange_client default (redis)
+    asserts:
+      - matchRegex:
+          path: data["config.yaml"]
+          pattern: 'exchange_client:\n\s+type: "redis"'
+
+  - it: should render postgresql exchange_client when set
+    set:
+      global.dbClient.type: "postgresql"
+      global.exchangeClient.type: "postgresql"
+    asserts:
+      - matchRegex:
+          path: data["config.yaml"]
+          pattern: 'exchange_client:\n\s+type: "postgresql"'
+
+  - it: should fail when postgresql exchange is used with a non-postgresql db
+    set:
+      global.dbClient.type: "redis"
+      global.exchangeClient.type: "postgresql"
+    asserts:
+      - failedTemplate:
+          errorMessage: "global.exchangeClient.type=postgresql requires global.dbClient.type=postgresql"
+
   - it: should not render ConfigMap when gc is disabled
     set:
       gc.enabled: false

@@ -367,6 +367,30 @@ tests:
           path: data["config.yaml"]
           pattern: 'send_fairness_header: true'
 
+  - it: should render exchange_client default (redis)
+    asserts:
+      - matchRegex:
+          path: data["config.yaml"]
+          pattern: 'exchange_client:\n\s+type: "redis"'
+
+  - it: should render postgresql exchange_client when set
+    set:
+      global.dbClient.type: "postgresql"
+      global.exchangeClient.type: "postgresql"
+    asserts:
+      - matchRegex:
+          path: data["config.yaml"]
+          pattern: 'exchange_client:\n\s+type: "postgresql"'
+
+  - it: should fail when postgresql exchange is used with async dispatch
+    set:
+      global.dbClient.type: "postgresql"
+      global.exchangeClient.type: "postgresql"
+      processor.config.dispatchMode: "async"
+    asserts:
+      - failedTemplate:
+          errorMessage: "global.exchangeClient.type=postgresql does not support async dispatch; use sync dispatch or a redis exchange"
+
   - it: should not render ConfigMap when processor is disabled
     set:
       processor.enabled: false

@@ -19,8 +19,12 @@ global:
   # Name of the Kubernetes Secret containing sensitive values (e.g. database URLs, API keys).
   # Shared by all components; mounted read-only at /etc/.secrets/ in each container.
   # The secret must use these exact key names (hardcoded in the application):
-  #   redis-url             — Redis/Valkey connection URL (e.g. redis://host:6379/0)
-  #   postgresql-url        — PostgreSQL connection URL (e.g. postgresql://[user]:[pass]@[host]/[db])
+  #   redis-url             — Redis/Valkey connection URL (e.g. redis://host:6379/0).
+  #                           Required only when dbClient.type or exchangeClient.type is
+  #                           redis/valkey. A postgres-only deployment (both set to
+  #                           postgresql) does not need this key at all.
+  #   postgresql-url        — PostgreSQL connection URL (e.g. postgresql://[user]:[pass]@[host]/[db]).
+  #                           Required when dbClient.type or exchangeClient.type is postgresql.
   #   inference-api-key     — API key for the inference gateway (optional; omit if not required)
   #   s3-secret-access-key  — S3 secret access key (required when file_client.type is "s3")
   # Create with: kubectl create secret generic <name> --from-literal=redis-url=redis://...
@@ -46,6 +50,20 @@ global:
       connMaxIdleTime: "0s"
       connMaxLifetime: "0s"
 
+  # Exchange client configuration (shared by apiserver, processor, and gc).
+  # The exchange client backs the priority queue, in-flight tracking, volatile
+  # status store, and per-job event channels (the runtime coordination plane),
+  # separate from the persistent dbClient store.
+  exchangeClient:
+    # Exchange backend type: "redis" or "postgresql".
+    #   redis      — default; unchanged behavior for existing deployments. Requires
+    #                the redis-url secret key even when dbClient.type is postgresql.
+    #   postgresql — postgres-only mode. Reuses the postgresql-url connection from
+    #                dbClient (no separate connection settings). Requires
+    #                dbClient.type to also be "postgresql". Does NOT support async
+    #                dispatch (llm-d-async is redis-only); use sync dispatch.
+    type: "redis"
+
   # OpenTelemetry configuration
   otel:
     # OTLP gRPC endpoint for trace collection. Examples:

@@ -69,6 +69,13 @@ file_client:
 #   redis:
 #     url_secret_key: "redis-url"  # key name within /etc/.secrets/ (mounted from existingSecret)
 
+# Exchange client configuration (priority queue, events, status, in-flight).
+# exchange_client:
+#   # Type selects the exchange backend: "redis" (default) or "postgresql".
+#   # "postgresql" requires db_client.type=postgresql and does not support async dispatch.
+#   # Connection settings are reused from db_client, so nothing else is configured here.
+#   type: "redis"
+
 # Batch API configuration
 batch_api:
   # Batch event TTL in seconds (default: 30 days)

@@ -33,6 +33,13 @@ db_client:
   #   conn_max_idle_time: "0s"
   #   conn_max_lifetime: "0s"
 
+# Exchange client configuration (priority queue, events, status, in-flight).
+# exchange_client:
+#   # Type selects the exchange backend: "redis" (default) or "postgresql".
+#   # "postgresql" requires db_client.type=postgresql and does not support async dispatch.
+#   # Connection settings are reused from db_client, so nothing else is configured here.
+#   type: "redis"
+
 # File storage backend configuration.
 file_client:
   type: "fs"

@@ -84,7 +84,7 @@ func run() error {
 		clientset.WithFile(cfg.FileClientCfg),
 	}
 	if cfg.Reconciler.Enabled {
-		clientOpts = append(clientOpts, clientset.WithExchange(cfg.DBClientCfg.RedisCfg))
+		clientOpts = append(clientOpts, clientset.WithExchange(cfg.ExchangeClientCfg, cfg.DBClientCfg.RedisCfg, cfg.DBClientCfg.PostgreSQLCfg))
 	}
 
 	clients, err := clientset.NewClientset(ctx, ucom.ComponentGC, clientOpts...)
@@ -130,6 +130,11 @@ func run() error {
 			return fmt.Errorf("failed to create reconciler: %w", err)
 		}
 		g.Go(func() error { return rec.RunLoop(gCtx) })
+
+		if sweeper, ok := clients.Queue.(exchangeSweeper); ok {
+			logger.Info("Exchange sweep loop enabled", "interval", cfg.Reconciler.Interval)
+			g.Go(func() error { return runExchangeSweepLoop(gCtx, sweeper, cfg.Reconciler.Interval, logger) })
+		}
 	}
 
 	ready.Store(true)
@@ -143,6 +148,31 @@ func run() error {
 	return nil
 }
 
+// exchangeSweeper is implemented by exchange backends that lack native TTL
+// (PostgreSQL) and need expired rows physically reclaimed; Redis expires keys
+// itself and does not implement it.
+type exchangeSweeper interface {
+	SweepExpired(ctx context.Context) error
+}
+
+// runExchangeSweepLoop bounds table growth; correctness never depends on it
+// (reads filter expired rows inline). Returns nil on shutdown so it does not trip
+// the errgroup's failure path.
+func runExchangeSweepLoop(ctx context.Context, sweeper exchangeSweeper, interval time.Duration, logger logr.Logger) error {
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-ticker.C:
+			if err := sweeper.SweepExpired(ctx); err != nil {
+				logger.Error(err, "exchange sweep failed")
+			}
+		}
+	}
+}
+
 func startMetricsServer(ctx context.Context, addr string, logger logr.Logger, ready *atomic.Bool) (<-chan error, error) {
 	mux := http.NewServeMux()
 	mux.Handle("/metrics", promhttp.Handler())

@@ -90,6 +90,13 @@ default_output_expiration_seconds: 7776000  # 90 days
 # TTL for temporary progress updates in the status store (Redis), in seconds.
 progress_ttl_seconds: 86400  # 24 hours
 
+# Exchange client configuration (priority queue, events, status, in-flight).
+# exchange_client:
+#   # Type selects the exchange backend: "redis" (default) or "postgresql".
+#   # "postgresql" requires db_client.type=postgresql and does not support async dispatch.
+#   # Connection settings are reused from db_client, so nothing else is configured here.
+#   type: "redis"
+
 # Whether to send x-gateway-inference-fairness-id on inference requests.
 # false (default) omits the fairness header entirely.
 send_fairness_header: false
@@ -288,7 +288,7 @@ func buildProcessorClients(ctx context.Context, cfg *config.ProcessorConfig) (*c
 	opts := []clientset.Option{
 		clientset.WithDB(cfg.DBClientCfg),
 		clientset.WithFile(cfg.FileClientCfg),
-		clientset.WithExchange(cfg.DBClientCfg.RedisCfg),
+		clientset.WithExchange(cfg.ExchangeClientCfg, cfg.DBClientCfg.RedisCfg, cfg.DBClientCfg.PostgreSQLCfg),
 	}
 	if resolved.Global != nil {
 		opts = append(opts, clientset.WithGlobalInference(*resolved.Global))