feat: forward writes from passive regions to active #691

docs/CROSS_REGION_WRITE_FORWARDING.md

@@ -0,0 +1,48 @@
+# Cross-Region Write Forwarding
+
+When running an active–passive multi-region deployment, the **passive** region
+must not accept mutating requests against a local writable database. Instead it
+forwards (or redirects) writes to the **active** region so failover data-loss
+windows stay small.
+
+## Configuration
+
+| Variable | Default | Description |
+| --- | --- | --- |
+| `REGION_ROLE` | `active` | `active` serves writes locally; `passive` forwards/redirects writes |
+| `ACTIVE_REGION_URL` | _(empty)_ | Base URL of the active region (required when `REGION_ROLE=passive`) |
+| `REGION_FORWARD_MODE` | `proxy` | `proxy` reverse-proxies the write; `redirect` returns HTTP 302 |
+| `REGION_FORWARD_AUTH_TOKEN` | _(empty)_ | Optional Bearer token attached when proxying to the active region |
+
+## Behaviour
+
+- **Read methods** (`GET`, `HEAD`, `OPTIONS`) always pass through locally.
+- **Write methods** (`POST`, `PUT`, `PATCH`, `DELETE`) on a passive region:
+  - **proxy**: reverse-proxy to `ACTIVE_REGION_URL` + path/query, stamp
+    `X-Region-Hop: 1`, and return the upstream status/body.
+  - **redirect**: respond with `302 Found` and a `Location` pointing at the
+    active URL.
+- **Loop prevention**: if an inbound write already carries `X-Region-Hop: 1`,
+  the middleware returns `508 Loop Detected` and does not forward again.
+- **Active unreachable**: proxy mode returns `503 Service Unavailable` with
+  `error: active_region_unavailable` when the active region cannot be reached
+  or `ACTIVE_REGION_URL` is unset.
+
+## Tracing
+
+Forwarded writes create a span `region.write_forward` with attributes:
+
+- `region.role`
+- `region.forward_mode`
+- `region.forwarded=true`
+- `region.active_url`
+
+## Example
+
+```bash
+# Passive region
+REGION_ROLE=passive
+ACTIVE_REGION_URL=https://api-active.example.com
+REGION_FORWARD_MODE=proxy
+REGION_FORWARD_AUTH_TOKEN=shared-tunnel-secret
+```

internal/config/config.go

@@ -41,23 +41,28 @@ func (e *ConfigError) Error() string {
 
 // Config holds all application configuration
 type Config struct {
-	Env                    string   `json:"env"`
-	Port                   int      `json:"port"`
-	DBConn                 string   `json:"db_conn" secret:"true"`
-	JWTSecret              string   `json:"jwt_secret" secret:"true"`
-	MaxHeaderBytes         int      `json:"max_header_bytes"`
-	ReadTimeout            int      `json:"read_timeout"`
-	WriteTimeout           int      `json:"write_timeout"`
-	IdleTimeout            int      `json:"idle_timeout"`
-	AllowedOrigins         string   `json:"allowed_origins"`
-	AdminToken             string   `json:"admin_token" secret:"true"`
-	DBReplicaConn          string   `json:"db_replica_conn" secret:"true"`
+	Env            string `json:"env"`
+	Port           int    `json:"port"`
+	DBConn         string `json:"db_conn" secret:"true"`
+	JWTSecret      string `json:"jwt_secret" secret:"true"`
+	MaxHeaderBytes int    `json:"max_header_bytes"`
+	ReadTimeout    int    `json:"read_timeout"`
+	WriteTimeout   int    `json:"write_timeout"`
+	IdleTimeout    int    `json:"idle_timeout"`
+	AllowedOrigins string `json:"allowed_origins"`
+	AdminToken     string `json:"admin_token" secret:"true"`
+	DBReplicaConn  string `json:"db_replica_conn" secret:"true"`
 	// Rate limiting configuration
 	RateLimitEnabled   bool     `json:"rate_limit_enabled"`
 	RateLimitMode      string   `json:"rate_limit_mode"`
 	RateLimitRPS       int      `json:"rate_limit_rps"`
 	RateLimitBurst     int      `json:"rate_limit_burst"`
 	RateLimitWhitelist []string `json:"rate_limit_whitelist"`
+	// Cross-region write forwarding (active–passive topology)
+	RegionRole             string `json:"region_role"`
+	ActiveRegionURL        string `json:"active_region_url"`
+	RegionForwardMode      string `json:"region_forward_mode"`
+	RegionForwardAuthToken string `json:"region_forward_auth_token" secret:"true"`
 	// Tracing configuration
 	TracingExporter        string
 	TracingServiceName     string
@@ -71,12 +76,12 @@ type Config struct {
 	CSPReportRPS int
 	// CSPReportBurst is the per-tenant burst size for /api/v1/csp-reports.
 	// Default: 10.
-	CSPReportBurst int
-	SpiffeSocketPath   string
-	SpiffeTrustDomain  string
-	MaxRequestSize         int64
-	MaxGzipUncompressed    int64
-	MaxGzipRatio           float64
+	CSPReportBurst      int
+	SpiffeSocketPath    string
+	SpiffeTrustDomain   string
+	MaxRequestSize      int64
+	MaxGzipUncompressed int64
+	MaxGzipRatio        float64
 	// RedisURL configures the Redis cache backend. When empty, an in-memory
 	// cache is used instead.
 	RedisURL string `json:"redis_url" secret:"true"`
@@ -123,11 +128,11 @@ type Config struct {
 	//   PGBOUNCER_MAX_CONN_IDLE_IN_TRANSACTION  (default 30) – idle-in-transaction
 	//                             server-side timeout forwarded into pgbouncer.ini
 	//                             as query_wait_timeout / idle_transaction_timeout.
-	PgBouncerEnabled        bool
-	PgBouncerHost           string
-	PgBouncerPort           int
-	DBStatementCacheMode    string // "prepare" | "describe" | "simple"
-	PgBouncerIdleInTxTimeout int   // seconds; written into pgbouncer.ini
+	PgBouncerEnabled         bool
+	PgBouncerHost            string
+	PgBouncerPort            int
+	DBStatementCacheMode     string // "prepare" | "describe" | "simple"
+	PgBouncerIdleInTxTimeout int    // seconds; written into pgbouncer.ini
 	// GracefulShutdownTimeout is the maximum seconds the server waits for
 	// in-flight requests to complete before forcing shutdown. Env:
 	// GRACEFUL_SHUTDOWN_TIMEOUT (default: DefaultGracefulShutdownTimeout).
@@ -186,8 +191,7 @@ const (
 	DefaultDBPoolMetricsInterval   = 15   // 15 s Prometheus scrape cadence
 
 	// Graceful shutdown defaults — coordinate with k8s terminationGracePeriodSeconds.
-	DefaultGracefulShutdownTimeout = 30   // 30 s to drain in-flight requests and pool
-
+	DefaultGracefulShutdownTimeout = 30 // 30 s to drain in-flight requests and pool
 
 	// Validation bounds
 	MinDBPoolMaxConns = 1
@@ -196,12 +200,12 @@ const (
 	MaxDBPoolTimeout  = 300 // seconds
 
 	// PgBouncer sidecar defaults.
-	DefaultPgBouncerHost           = "127.0.0.1"
-	DefaultPgBouncerPort           = 5432
-	DefaultDBStatementCacheMode    = "prepare"
+	DefaultPgBouncerHost            = "127.0.0.1"
+	DefaultPgBouncerPort            = 5432
+	DefaultDBStatementCacheMode     = "prepare"
 	DefaultPgBouncerIdleInTxTimeout = 30 // seconds
-	MinPgBouncerPort               = 1
-	MaxPgBouncerPort               = 65535
+	MinPgBouncerPort                = 1
+	MaxPgBouncerPort                = 65535
 
 	// Valid DB_STATEMENT_CACHE_MODE values.
 	StatementCacheModeDescribe = "describe"
@@ -272,9 +276,9 @@ func Load(opts ...Option) (Config, error) {
 		MaxGzipUncompressed:    getEnvInt64("MAX_GZIP_UNCOMPRESSED", 1024*1024*50), // 50MB
 		MaxGzipRatio:           getEnvFloat64("MAX_GZIP_RATIO", 10.0),
 		// DB pool — safe production defaults
-		DBReplicaConn:    getEnv("DB_REPLICA_URL", ""),
-		RedisURL:         getEnv("REDIS_URL", ""),
-		CacheTTL:         getEnvInt("CACHE_TTL", 60), // 60 second default
+		DBReplicaConn:           getEnv("DB_REPLICA_URL", ""),
+		RedisURL:                getEnv("REDIS_URL", ""),
+		CacheTTL:                getEnvInt("CACHE_TTL", 60), // 60 second default
 		DBPoolMaxConns:          DefaultDBPoolMaxConns,
 		DBPoolMinConns:          DefaultDBPoolMinConns,
 		DBPoolMaxConnLifetime:   DefaultDBPoolMaxConnLifetime,
@@ -290,6 +294,10 @@ func Load(opts ...Option) (Config, error) {
 		PgBouncerIdleInTxTimeout: DefaultPgBouncerIdleInTxTimeout,
 		GracefulShutdownTimeout:  DefaultGracefulShutdownTimeout,
 		ConcurrencyCapsPath:      getEnv("CONCURRENCY_CAPS_PATH", ""),
+		RegionRole:               getEnv("REGION_ROLE", "active"),
+		ActiveRegionURL:          getEnv("ACTIVE_REGION_URL", ""),
+		RegionForwardMode:        getEnv("REGION_FORWARD_MODE", "proxy"),
+		RegionForwardAuthToken:   getEnv("REGION_FORWARD_AUTH_TOKEN", ""),
 	}
 
 	// Resolve secrets through the provider

internal/middleware/region_router.go

@@ -0,0 +1,215 @@
+package middleware
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"strings"
+	"time"
+
+	"github.com/gin-gonic/gin"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
+	"go.opentelemetry.io/otel/trace"
+)
+
+const (
+	// RegionRoleActive serves writes locally.
+	RegionRoleActive = "active"
+	// RegionRolePassive forwards or redirects writes to the active region.
+	RegionRolePassive = "passive"
+
+	// RegionForwardModeProxy reverse-proxies the write to the active region.
+	RegionForwardModeProxy = "proxy"
+	// RegionForwardModeRedirect returns HTTP 302 with the active region URL.
+	RegionForwardModeRedirect = "redirect"
+
+	// RegionHopHeader prevents write-forward loops across regions.
+	RegionHopHeader = "X-Region-Hop"
+	// RegionHopValue is the value stamped on forwarded requests.
+	RegionHopValue = "1"
+)
+
+// RegionRouterConfig configures cross-region write forwarding.
+type RegionRouterConfig struct {
+	// Role is "active" (default) or "passive".
+	Role string
+	// ActiveRegionURL is the base URL of the active region (required when Role=passive).
+	ActiveRegionURL string
+	// ForwardMode is "proxy" (default) or "redirect".
+	ForwardMode string
+	// ForwardAuthToken is sent as Authorization: Bearer <token> when proxying.
+	ForwardAuthToken string
+	// HTTPClient is used for proxy forwards; if nil a default client is created.
+	HTTPClient *http.Client
+	// Timeout bounds the proxy round-trip (default 10s).
+	Timeout time.Duration
+}
+
+// RegionRouterMiddleware forwards write requests from a passive region to the
+// active region. Read methods (GET/HEAD/OPTIONS) always pass through.
+//
+// Loop prevention: requests that already carry X-Region-Hop: 1 are rejected
+// with 508 Loop Detected instead of being forwarded again.
+//
+// When the active region is unreachable in proxy mode, the middleware returns
+// 503 Service Unavailable with a helpful body.
+func RegionRouterMiddleware(cfg RegionRouterConfig) gin.HandlerFunc {
+	role := strings.ToLower(strings.TrimSpace(cfg.Role))
+	if role == "" {
+		role = RegionRoleActive
+	}
+	mode := strings.ToLower(strings.TrimSpace(cfg.ForwardMode))
+	if mode == "" {
+		mode = RegionForwardModeProxy
+	}
+	timeout := cfg.Timeout
+	if timeout <= 0 {
+		timeout = 10 * time.Second
+	}
+	client := cfg.HTTPClient
+	if client == nil {
+		client = &http.Client{Timeout: timeout}
+	}
+
+	var activeBase *url.URL
+	if role == RegionRolePassive && cfg.ActiveRegionURL != "" {
+		if u, err := url.Parse(cfg.ActiveRegionURL); err == nil && u.Scheme != "" && u.Host != "" {
+			activeBase = u
+		}
+	}
+
+	tracer := otel.Tracer("stellarbill/middleware/region_router")
+
+	return func(c *gin.Context) {
+		if role != RegionRolePassive {
+			c.Next()
+			return
+		}
+		if !isWriteMethod(c.Request.Method) {
+			c.Next()
+			return
+		}
+
+		ctx, span := tracer.Start(c.Request.Context(), "region.write_forward",
+			trace.WithAttributes(
+				attribute.String("region.role", role),
+				attribute.String("region.forward_mode", mode),
+				attribute.String("http.method", c.Request.Method),
+				attribute.String("http.target", c.Request.URL.Path),
+			),
+		)
+		defer span.End()
+		c.Request = c.Request.WithContext(ctx)
+
+		// Loop prevention — never forward a request that was already hopped.
+		if c.GetHeader(RegionHopHeader) == RegionHopValue {
+			span.SetStatus(codes.Error, "region hop loop detected")
+			c.AbortWithStatusJSON(http.StatusLoopDetected, gin.H{
+				"error":   "region_hop_loop",
+				"message": "write was already forwarded once; refusing to forward again",
+			})
+			return
+		}
+
+		if activeBase == nil {
+			span.SetStatus(codes.Error, "active region URL not configured")
+			c.AbortWithStatusJSON(http.StatusServiceUnavailable, gin.H{
+				"error":   "active_region_unavailable",
+				"message": "passive region cannot accept writes: ACTIVE_REGION_URL is not configured",
+			})
+			return
+		}
+
+		target := *activeBase
+		target.Path = singleJoinPath(activeBase.Path, c.Request.URL.Path)
+		target.RawQuery = c.Request.URL.RawQuery
+
+		span.SetAttributes(
+			attribute.Bool("region.forwarded", true),
+			attribute.String("region.active_url", target.String()),
+		)
+
+		if mode == RegionForwardModeRedirect {
+			c.Header(RegionHopHeader, RegionHopValue)
+			c.Redirect(http.StatusFound, target.String())
+			c.Abort()
+			return
+		}
+
+		// Proxy mode: reverse-proxy the write to the active region.
+		proxyReq, err := http.NewRequestWithContext(ctx, c.Request.Method, target.String(), c.Request.Body)
+		if err != nil {
+			span.RecordError(err)
+			span.SetStatus(codes.Error, "failed to build forward request")
+			c.AbortWithStatusJSON(http.StatusServiceUnavailable, gin.H{
+				"error":   "active_region_unavailable",
+				"message": "failed to build forward request to active region",
+			})
+			return
+		}
+		copyHeaders(c.Request.Header, proxyReq.Header)
+		proxyReq.Header.Set(RegionHopHeader, RegionHopValue)
+		if cfg.ForwardAuthToken != "" {
+			proxyReq.Header.Set("Authorization", "Bearer "+cfg.ForwardAuthToken)
+		}
+
+		resp, err := client.Do(proxyReq)
+		if err != nil {
+			span.RecordError(err)
+			span.SetStatus(codes.Error, "active region unreachable")
+			c.AbortWithStatusJSON(http.StatusServiceUnavailable, gin.H{
+				"error":   "active_region_unavailable",
+				"message": fmt.Sprintf("active region unreachable: %v", err),
+			})
+			return
+		}
+		defer resp.Body.Close()
+
+		for k, vals := range resp.Header {
+			for _, v := range vals {
+				c.Writer.Header().Add(k, v)
+			}
+		}
+		c.Writer.Header().Set(RegionHopHeader, RegionHopValue)
+		c.Status(resp.StatusCode)
+		_, _ = io.Copy(c.Writer, resp.Body)
+		c.Abort()
+	}
+}
+
+func isWriteMethod(method string) bool {
+	switch strings.ToUpper(method) {
+	case http.MethodPost, http.MethodPut, http.MethodPatch, http.MethodDelete:
+		return true
+	default:
+		return false
+	}
+}
+
+func singleJoinPath(base, rel string) string {
+	base = strings.TrimSuffix(base, "/")
+	if !strings.HasPrefix(rel, "/") {
+		rel = "/" + rel
+	}
+	if base == "" {
+		return rel
+	}
+	return base + rel
+}
+
+func copyHeaders(src, dst http.Header) {
+	for k, vals := range src {
+		// Hop-by-hop headers must not be forwarded.
+		switch strings.ToLower(k) {
+		case "connection", "keep-alive", "proxy-authenticate", "proxy-authorization",
+			"te", "trailers", "transfer-encoding", "upgrade", "content-length":
+			continue
+		}
+		for _, v := range vals {
+			dst.Add(k, v)
+		}
+	}
+}