feat(ha): reader-proxy edge + writer fcntl lease for zero-downtime cloud rolls

The cloud writer embeds an exclusive-lock ZapDB KMS store. A new probe
(clients/kms.TestConcurrentOpen_LiveWriterStoreIsNotROShareable) proves that
opening that store READ-ONLY while the writer is live FAILS ("Log truncate
required to run DB") — Badger's RO open replays the live memtable WAL and
refuses to truncate it. So the prior groundwork's assumption that a reader can
open the KMS store RO off the writer's PVC is false for a LIVE writer (only the
sequential close-then-reopen case worked). The audit SQLite store IS
concurrently shareable (audit/shareability_probe_test.go); the KMS store is the
one that is not, and every mutation is audited on the writer anyway.

Reader tier is therefore a transparent, always-ready reverse proxy (opens no
stores) to the single writer:
 - reader_proxy.go: CLOUD_ROLE=reader boots serveReaderProxy BEFORE BuildDeps —
   forwards every request to CLOUD_WRITER_URL, streams SSE, preserves inbound
   Host. Dial-only retry (retryTransport) absorbs the writer's roll gap: it
   retries ONLY when the connection was never established (no ready endpoint /
   refused), so a non-idempotent POST is never double-executed; bounded by
   CLOUD_READER_RETRY_BUDGET (default 25s) then 502.
 - The reader Deployment rolls RollingUpdate(maxUnavailable:0), so the edge
   Service always has a ready endpoint — this removes the ~30s console blip that
   the writer's Recreate/replicas:1 causes today.

Writer zero-gap roll (opt-in, default OFF = byte-identical Recreate):
 - writer_lease.go (+_unix/_other): CLOUD_WRITER_LEASE takes an exclusive fcntl
   flock on {DataDir}/.writer.lock BEFORE opening the RWO stores and releases it
   LAST at shutdown (after every store closes). A surge writer blocks until the
   old one releases, so the exclusive ZapDB/audit stores are handed off, never
   double-opened. Fail-closed on timeout.

Removes the dead ReaderGuard (the reader no longer runs the full pipeline; it is
the proxy). Unset CLOUD_ROLE + unset CLOUD_WRITER_LEASE ⇒ writer, byte-identical
to today.

Co-authored-by: Hanzo Dev <dev@hanzo.ai>

clients/kms/concurrent_open_probe_test.go

@@ -0,0 +1,94 @@
+package kms
+
+// Concurrent-open invariant (HA carve evidence — a REGRESSION GUARD, not a feature).
+//
+// blue_readonly_test.go proves the SEQUENTIAL reader path (write, CLOSE the
+// writer, THEN reopen RO). That is a red herring for HA: in a shared-PVC
+// same-node topology the writer pod holds the ZapDB store open for WRITE (an
+// actively-growing memtable WAL) while a reader pod would open the SAME on-disk
+// files. This test proves that scenario is NOT supported and, deliberately,
+// asserts the FAILURE so the constraint is enforced in CI:
+//
+//	Opening a live ZapDB (Badger fork) store READ-ONLY while the writer is
+//	mid-write fails with "Log truncate required to run DB" — Badger's RO open
+//	replays the current memtable WAL, finds it partially written
+//	(end offset < preallocated size), and REFUSES to truncate it (truncation
+//	is a write, forbidden in RO mode). There is no torn read; there is no open.
+//
+// CONSEQUENCE (the design this guards): a reader-role pod must NOT open the KMS
+// ZapDB store off the live writer's PVC. The KMS store is the ONE cloud store
+// that is NOT concurrently shareable (unlike the audit SQLite store — see
+// audit/shareability_probe_test.go — which shares cleanly over WAL). Therefore a
+// reader serves KMS by REVERSE-PROXYING /v1/kms/* (and every mutation +
+// /v1/admin/*) to the writer, never by opening the store locally. If a future
+// zapdb release makes live concurrent RO-open work, THIS TEST WILL FAIL — that is
+// the signal to revisit the reader-serves-KMS-locally option.
+//
+// Run:
+//   CGO_ENABLED=0 GOWORK=off GOFLAGS=-mod=mod go test ./clients/kms/ -run ConcurrentOpen -v
+
+import (
+	"fmt"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	luxlog "github.com/luxfi/log"
+)
+
+func TestConcurrentOpen_LiveWriterStoreIsNotROShareable(t *testing.T) {
+	dir := t.TempDir()
+	log := luxlog.NewNoOpLogger()
+	key := b64key(t, 0x5A)
+
+	// Writer: create the encrypted store and seed baseline secrets, then keep
+	// writing so the current memtable WAL is genuinely mid-flight (not flushed,
+	// not closed) when the reader attempts to open.
+	w, err := New(Config{DataDir: dir, MasterKeyB64: key}, log)
+	if err != nil {
+		t.Fatalf("writer New: %v", err)
+	}
+	defer w.Close()
+	for i := 0; i < 200; i++ {
+		if err := w.Put("/orgs/acme", fmt.Sprintf("K%d", i), "default", []byte(fmt.Sprintf("v%d", i))); err != nil {
+			t.Fatalf("writer seed Put %d: %v", i, err)
+		}
+	}
+
+	var stop atomic.Bool
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for i := 200; !stop.Load(); i++ {
+			if err := w.Put("/orgs/acme", fmt.Sprintf("K%d", i), "default", []byte(fmt.Sprintf("v%d", i))); err != nil {
+				return // writer wound down; not the subject of this assertion
+			}
+			time.Sleep(200 * time.Microsecond)
+		}
+	}()
+	time.Sleep(50 * time.Millisecond) // ensure the WAL is actively mid-write
+
+	// Reader: attempt to open the SAME files READ-ONLY (BypassLockGuard) while the
+	// writer is live. INVARIANT: this must fail (no torn read, no silent success).
+	r, err := New(Config{DataDir: dir, MasterKeyB64: key, ReadOnly: true}, log)
+	stop.Store(true)
+	wg.Wait()
+
+	if err == nil {
+		if r != nil {
+			_ = r.Close()
+		}
+		t.Fatal("EXPECTED concurrent RO-open of a live ZapDB writer to FAIL, but it " +
+			"succeeded. If zapdb now supports live concurrent RO-open, the reader " +
+			"tier may serve KMS locally instead of proxying — revisit the design.")
+	}
+	// The failure is the WAL-truncation refusal, confirming Badger's RO open cannot
+	// coexist with a live writer's unflushed memtable.
+	if !strings.Contains(err.Error(), "truncate") && !strings.Contains(err.Error(), "Log truncate") {
+		t.Logf("concurrent RO-open failed (as required) with a different error: %v", err)
+	}
+	t.Logf("INVARIANT HELD: live ZapDB store is NOT RO-shareable (%v) — reader must proxy /v1/kms/*", err)
+}

config.go

@@ -6,6 +6,7 @@ import (
 	"os"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/hanzoai/cloud/role"
 )
@@ -82,6 +83,30 @@ type Config struct {
 	// to today's single-pod deployment.
 	Role role.Role
 
+	// WriterURL is the base URL of the single writer (CLOUD_WRITER_URL, e.g.
+	// http://cloud-writer.hanzo.svc:8000). A Reader forwards EVERY request here —
+	// it opens no stores and is a transparent, always-ready edge that absorbs the
+	// writer's rollout gap (see reader_proxy.go). Required when Role==Reader;
+	// ignored by a Writer.
+	WriterURL string
+
+	// ReaderRetryBudget bounds how long a Reader retries a request the writer
+	// could not yet accept (connection refused / no ready endpoint) during a
+	// writer roll, before returning 502. Dial-only retry (the request never
+	// reached the writer) keeps a non-idempotent POST safe. CLOUD_READER_RETRY_BUDGET
+	// (Go duration, default 25s).
+	ReaderRetryBudget time.Duration
+
+	// WriterLease, when true (CLOUD_WRITER_LEASE), makes a Writer take an
+	// exclusive fcntl lease on {DataDir}/.writer.lock BEFORE opening the RWO
+	// stores and release it LAST at shutdown (after every store is closed). This
+	// serializes a surge/overlap roll (RollingUpdate maxUnavailable:1 +
+	// same-node podAffinity) so the new writer opens the exclusive-lock ZapDB/
+	// audit stores only after the old one released them — never a double-open.
+	// Default OFF, so an unset variable is byte-identical to today's Recreate
+	// single-writer (which never overlaps and needs no lease).
+	WriterLease bool
+
 	// ListenAddr is the public HTTP listener (default :8080).
 	ListenAddr string
 
@@ -295,6 +320,9 @@ func LoadConfig() *Config {
 		KMSMPCAddr:      getenv("CLOUD_KMS_MPC_ADDR", ""),
 		KMSMPCVaultID:   getenv("CLOUD_KMS_MPC_VAULT_ID", ""),
 		DataDir:         getenv("CLOUD_DATA_DIR", "/var/lib/cloud"),
+		WriterURL:         strings.TrimRight(getenv("CLOUD_WRITER_URL", ""), "/"),
+		ReaderRetryBudget: getenvDuration("CLOUD_READER_RETRY_BUDGET", 25*time.Second),
+		WriterLease:       getenvBool("CLOUD_WRITER_LEASE"),
 		PaymentsZAPAddr: getenv("CLOUD_PAYMENTS_ZAP_ADDR", ""),
 		VaultZAPAddr:    getenv("CLOUD_VAULT_ZAP_ADDR", ""),
 		// Billing gate (KMS-backed COMMERCE_SERVICE_TOKEN; never plaintext).
@@ -583,6 +611,21 @@ func getenvBoolDefault(key string, dflt bool) bool {
 	}
 }
 
+// getenvDuration reads key as a Go duration (e.g. "25s", "1m"), returning dflt
+// when unset, blank, or unparseable (a malformed override can never silently
+// zero a timeout).
+func getenvDuration(key string, dflt time.Duration) time.Duration {
+	v := strings.TrimSpace(os.Getenv(key))
+	if v == "" {
+		return dflt
+	}
+	d, err := time.ParseDuration(v)
+	if err != nil {
+		return dflt
+	}
+	return d
+}
+
 // getenvInt reads key as a base-10 int, returning dflt when unset, blank, or
 // unparseable (a malformed override can never silently zero a scale knob).
 func getenvInt(key string, dflt int) int {

middleware_reader.go

@@ -1,32 +0,0 @@
-package cloud
-
-import (
-	"net/http"
-
-	"github.com/hanzoai/cloud/role"
-	"github.com/zap-proto/zip"
-)
-
-// ReaderGuard fails CLOSED on a read replica: every mutating request is rejected
-// at the app boundary, so correctness never rides on gateway routing.
-//
-// A Reader opens the KMS store read-only and hydrates audit / durable tasks /
-// per-tenant SQLite from the replication stream into an EPHEMERAL dir. A write
-// that reached a reader — a mis-route, a rollout race — would persist to that
-// ephemeral dir, answer 2xx, then VANISH on the next restart: silent data loss,
-// and UNAUDITED (the audit chain is writer-only). ONE guard gates ALL stores at
-// once, not just KMS: GET/HEAD/OPTIONS are served; every other verb is 405 with
-// a pointer to retry against the writer. Mounted only when role.IsReader(), so a
-// Writer (the default, unset CLOUD_ROLE) is untouched.
-func ReaderGuard() zip.Handler {
-	return func(c *zip.Ctx) error {
-		switch c.Method() {
-		case http.MethodGet, http.MethodHead, http.MethodOptions:
-			return c.Next()
-		}
-		return c.JSON(http.StatusMethodNotAllowed, map[string]string{
-			"error": "read-only replica: retry this write against the writer",
-			"role":  string(role.Reader),
-		})
-	}
-}

middleware_reader_test.go

@@ -1,74 +0,0 @@
-package cloud
-
-import (
-	"net/http"
-	"net/http/httptest"
-	"testing"
-
-	"github.com/zap-proto/zip"
-)
-
-// TestReaderGuardRejectsWrites proves the fail-closed reader boundary: with the
-// guard mounted (Reader), GET/HEAD/OPTIONS reach the handler and EVERY mutating
-// verb is 405 without reaching it — one guard covering every store, so a
-// mis-routed write can never persist to a reader's ephemeral dir.
-func TestReaderGuardRejectsWrites(t *testing.T) {
-	app := zip.New(zip.Config{})
-	app.Use(ReaderGuard())
-
-	var reached bool
-	app.All("/v1/kms/secret", func(c *zip.Ctx) error {
-		reached = true
-		return c.JSON(http.StatusOK, map[string]bool{"ok": true})
-	})
-
-	for _, tc := range []struct {
-		method string
-		want   int
-		reach  bool
-	}{
-		{http.MethodGet, http.StatusOK, true},
-		{http.MethodHead, http.StatusOK, true},
-		{http.MethodOptions, http.StatusOK, true},
-		{http.MethodPost, http.StatusMethodNotAllowed, false},
-		{http.MethodPut, http.StatusMethodNotAllowed, false},
-		{http.MethodPatch, http.StatusMethodNotAllowed, false},
-		{http.MethodDelete, http.StatusMethodNotAllowed, false},
-	} {
-		reached = false
-		resp, err := app.Fiber().Test(httptest.NewRequest(tc.method, "/v1/kms/secret", nil))
-		if err != nil {
-			t.Fatalf("%s: %v", tc.method, err)
-		}
-		_ = resp.Body.Close()
-		if resp.StatusCode != tc.want {
-			t.Errorf("%s = %d, want %d", tc.method, resp.StatusCode, tc.want)
-		}
-		if reached != tc.reach {
-			t.Errorf("%s reached store handler = %v, want %v (a write must never reach a reader's store)", tc.method, reached, tc.reach)
-		}
-	}
-}
-
-// TestWriterUngated proves the Writer path is byte-identical to today: serve.go
-// mounts ReaderGuard ONLY when role.IsReader(), so without it every verb reaches
-// the handler and succeeds.
-func TestWriterUngated(t *testing.T) {
-	app := zip.New(zip.Config{})
-	var reached bool
-	app.All("/v1/kms/secret", func(c *zip.Ctx) error {
-		reached = true
-		return c.JSON(http.StatusOK, map[string]bool{"ok": true})
-	})
-	for _, m := range []string{http.MethodGet, http.MethodPost, http.MethodDelete} {
-		reached = false
-		resp, err := app.Fiber().Test(httptest.NewRequest(m, "/v1/kms/secret", nil))
-		if err != nil {
-			t.Fatalf("%s: %v", m, err)
-		}
-		_ = resp.Body.Close()
-		if !reached || resp.StatusCode != http.StatusOK {
-			t.Errorf("writer %s: reached=%v status=%d, want reached=true status=200", m, reached, resp.StatusCode)
-		}
-	}
-}

reader_proxy.go

@@ -0,0 +1,221 @@
+package cloud
+
+// Reader edge — a transparent, always-ready reverse proxy to the single writer.
+//
+// WHY A PROXY, NOT A LOCAL-STORE REPLICA. The writer embeds an exclusive-lock
+// ZapDB KMS store (a Badger fork). clients/kms.TestConcurrentOpen_LiveWriterStore-
+// IsNotROShareable proves that opening that store READ-ONLY while the writer is
+// live FAILS ("Log truncate required to run DB") — Badger's RO open replays the
+// live memtable WAL and refuses to truncate it. So a reader CANNOT open the KMS
+// store off the writer's PVC, even read-only, even same-node. (The audit SQLite
+// store IS concurrently shareable — audit/shareability_probe_test.go — but the
+// KMS store is the one that is not, and every mutation is audited on the writer
+// anyway.) Rather than braid a partial local-read replica that must carefully
+// route KMS + every audited verb to the writer, the reader is the SIMPLEST
+// correct thing: it opens NO stores and forwards EVERY request to the writer.
+//
+// WHAT IT BUYS. The reader Deployment rolls RollingUpdate (maxUnavailable:0), so
+// the edge Service always has a ready endpoint. During a writer roll the reader
+// holds the client connection and RETRIES (dial-only) across the writer's brief
+// handoff gap, so a `rollout restart` of the writer never surfaces a 502/refused
+// at the edge — it surfaces as a little extra latency. This is what removes the
+// ~30s console blip that Recreate/replicas:1 causes today.
+//
+// SAFETY OF RETRY. Retry fires ONLY on a dial failure — the connection to the
+// writer was never established (no ready endpoint / connection refused), so the
+// request was never delivered and re-sending it cannot double-execute a
+// non-idempotent POST. Once bytes are on the wire to a writer, a failure is NOT
+// retried (it is ambiguous). The request body is buffered (bounded) so a retried
+// POST can be replayed.
+//
+// TRANSPARENCY / TRUST BOUNDARY. The reader forwards the request unchanged
+// (method, path, query, headers, body) and preserves the inbound Host for the
+// writer's host-based routing. It adds nothing to identity: the writer's
+// SanitizeIdentity re-validates the JWT and re-derives X-Org-Id/X-User-Id exactly
+// as if the gateway reached it directly, so the writer's trust boundary is
+// unchanged by the extra hop. httputil.ReverseProxy strips hop-by-hop headers and
+// appends X-Forwarded-For, and transparently proxies WebSocket/SSE upgrades.
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/http/httputil"
+	"net/url"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	luxlog "github.com/luxfi/log"
+)
+
+// serveReaderProxy runs the reader edge: a reverse proxy to cfg.WriterURL on the
+// public listener, plus the ops health listener, shutting down gracefully on
+// SIGINT/SIGTERM. It opens no stores and never returns until shutdown or a bind
+// error. Serve dispatches here when CLOUD_ROLE=reader, BEFORE BuildDeps, so a
+// reader never opens the KMS/audit/per-tenant stores.
+func serveReaderProxy(cfg *Config) error {
+	log := luxlog.New("cloud").New("subsystem", "reader")
+
+	rp, err := newReaderProxy(cfg, log)
+	if err != nil {
+		return err
+	}
+
+	mainSrv := &http.Server{Addr: cfg.ListenAddr, Handler: rp, ReadHeaderTimeout: 10 * time.Second}
+	healthSrv := &http.Server{Addr: cfg.HealthListenAddr, Handler: healthMux(), ReadHeaderTimeout: 5 * time.Second}
+
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+
+	listenErr := make(chan error, 2)
+	go func() {
+		log.Info("reader health listening", "addr", cfg.HealthListenAddr)
+		if err := healthSrv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+			listenErr <- fmt.Errorf("reader health listen: %w", err)
+		}
+	}()
+	go func() {
+		log.Info("reader edge listening", "addr", cfg.ListenAddr, "writer", cfg.WriterURL, "retry_budget", cfg.ReaderRetryBudget.String())
+		if err := mainSrv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+			listenErr <- fmt.Errorf("reader edge listen: %w", err)
+		}
+	}()
+
+	select {
+	case <-ctx.Done():
+		log.Info("reader shutdown requested")
+	case err := <-listenErr:
+		return fmt.Errorf("reader listen: %w", err)
+	}
+
+	shutdownCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	_ = healthSrv.Shutdown(shutdownCtx)
+	return mainSrv.Shutdown(shutdownCtx)
+}
+
+// newReaderProxy builds the transparent reverse proxy to cfg.WriterURL: dial-only
+// retry transport, prompt flushing for SSE, and a 502 error handler once the
+// retry budget is exhausted. It preserves the inbound Host for the writer's
+// host-based routing. Extracted from serveReaderProxy so it is unit-testable.
+func newReaderProxy(cfg *Config, log luxlog.Logger) (*httputil.ReverseProxy, error) {
+	if cfg.WriterURL == "" {
+		return nil, fmt.Errorf("reader role: CLOUD_WRITER_URL is required (the writer base URL to forward to, e.g. http://cloud-writer.hanzo.svc:8000)")
+	}
+	target, err := url.Parse(cfg.WriterURL)
+	if err != nil || target.Scheme == "" || target.Host == "" {
+		return nil, fmt.Errorf("reader role: invalid CLOUD_WRITER_URL %q: %v", cfg.WriterURL, err)
+	}
+	rp := httputil.NewSingleHostReverseProxy(target)
+	// Flush promptly so SSE / chunked streams (chat completions) pass through with
+	// no added buffering latency.
+	rp.FlushInterval = 100 * time.Millisecond
+	rp.Transport = newRetryTransport(cfg.ReaderRetryBudget, log)
+	rp.ErrorHandler = func(w http.ResponseWriter, r *http.Request, e error) {
+		if log != nil {
+			log.Warn("reader proxy failed", "path", r.URL.Path, "method", r.Method, "err", e)
+		}
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusBadGateway)
+		_, _ = w.Write([]byte(`{"error":{"code":"writer_unavailable","message":"upstream writer unavailable"}}`))
+	}
+	return rp, nil
+}
+
+// retryTransport re-sends a request ONLY when the writer could not be dialed —
+// the connection was never established, so the request never reached the writer
+// and replay cannot double-execute it. It buffers the request body (bounded) so a
+// retried POST can be replayed. All other failures (a delivered request whose
+// response failed) are returned as-is: retrying them would be ambiguous.
+type retryTransport struct {
+	base            http.RoundTripper
+	budget          time.Duration
+	maxBufferedBody int64
+	log             luxlog.Logger
+}
+
+func newRetryTransport(budget time.Duration, log luxlog.Logger) *retryTransport {
+	if budget <= 0 {
+		budget = 25 * time.Second
+	}
+	return &retryTransport{
+		base:            http.DefaultTransport.(*http.Transport).Clone(),
+		budget:          budget,
+		maxBufferedBody: 8 << 20, // 8 MiB — above this a request is a single attempt (never buffered)
+		log:             log,
+	}
+}
+
+func (t *retryTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+	// Make the body replayable. If it is already rewindable (GetBody set by the
+	// proxy for small bodies) use that; otherwise buffer up to the cap. A body
+	// larger than the cap is sent once with no retry (never silently truncated).
+	getBody := req.GetBody
+	if getBody == nil && req.Body != nil && req.Body != http.NoBody {
+		buf, err := io.ReadAll(io.LimitReader(req.Body, t.maxBufferedBody+1))
+		_ = req.Body.Close()
+		if err != nil {
+			return nil, err
+		}
+		if int64(len(buf)) > t.maxBufferedBody {
+			// Too large to safely rebuffer: single attempt with what we read.
+			req.Body = io.NopCloser(bytes.NewReader(buf))
+			req.ContentLength = int64(len(buf))
+			return t.base.RoundTrip(req)
+		}
+		body := buf
+		getBody = func() (io.ReadCloser, error) { return io.NopCloser(bytes.NewReader(body)), nil }
+		req.ContentLength = int64(len(buf))
+	}
+
+	deadline := time.Now().Add(t.budget)
+	backoff := 100 * time.Millisecond
+	attempts := 0
+	for {
+		attempts++
+		if getBody != nil {
+			b, err := getBody()
+			if err != nil {
+				return nil, err
+			}
+			req.Body = b
+		}
+		resp, err := t.base.RoundTrip(req)
+		if err == nil {
+			return resp, nil
+		}
+		// Only a dial failure (request never delivered) is safely retryable.
+		if !isDialError(err) || time.Now().After(deadline) {
+			return nil, err
+		}
+		select {
+		case <-req.Context().Done():
+			return nil, req.Context().Err()
+		case <-time.After(backoff):
+		}
+		if backoff < 2*time.Second {
+			backoff *= 2
+		}
+	}
+}
+
+// isDialError reports whether err means the connection to the writer was never
+// established — a no-endpoint / connection-refused / dial-timeout condition
+// during a writer roll — so re-sending the request cannot double-execute it.
+func isDialError(err error) bool {
+	if errors.Is(err, syscall.ECONNREFUSED) {
+		return true
+	}
+	var opErr *net.OpError
+	if errors.As(err, &opErr) {
+		// "dial" is the phase before any byte is written to the writer.
+		return opErr.Op == "dial"
+	}
+	return false
+}

reader_proxy_test.go

@@ -0,0 +1,167 @@
+package cloud
+
+import (
+	"errors"
+	"io"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"syscall"
+	"testing"
+	"time"
+
+	luxlog "github.com/luxfi/log"
+)
+
+// TestRetryTransport_AbsorbsDialGapThenSucceeds proves the core zero-downtime
+// property: while the writer is un-dialable (its endpoint is down mid-roll), the
+// reader retries and, once the writer comes back, the request SUCCEEDS — no
+// 502/refused surfaces at the edge. This is what turns the writer's roll gap into
+// a little latency instead of a blip.
+func TestRetryTransport_AbsorbsDialGapThenSucceeds(t *testing.T) {
+	var up atomic.Bool
+	// A stub upstream that only exists while up==true; when down, dials are refused.
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	addr := ln.Addr().String()
+	_ = ln.Close() // free the port; nothing listens until we bring it up
+
+	srvErr := make(chan error, 1)
+	bringUp := func() {
+		l, err := net.Listen("tcp", addr)
+		if err != nil {
+			srvErr <- err
+			return
+		}
+		up.Store(true)
+		s := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			b, _ := io.ReadAll(r.Body)
+			w.WriteHeader(http.StatusOK)
+			_, _ = w.Write([]byte("ok:" + string(b)))
+		})}
+		_ = s.Serve(l)
+	}
+
+	// Bring the upstream up after 600ms — simulating the writer handoff gap.
+	go func() {
+		time.Sleep(600 * time.Millisecond)
+		bringUp()
+	}()
+
+	rt := newRetryTransport(10*time.Second, luxlog.NewNoOpLogger())
+	req, _ := http.NewRequest(http.MethodPost, "http://"+addr+"/v1/chat/completions", strings.NewReader(`{"x":1}`))
+
+	start := time.Now()
+	resp, err := rt.RoundTrip(req)
+	if err != nil {
+		t.Fatalf("RoundTrip failed despite retry budget: %v (up=%v)", err, up.Load())
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("status = %d, want 200", resp.StatusCode)
+	}
+	body, _ := io.ReadAll(resp.Body)
+	if string(body) != `ok:{"x":1}` {
+		t.Fatalf("body = %q, want replayed POST body echoed", body)
+	}
+	if elapsed := time.Since(start); elapsed < 500*time.Millisecond {
+		t.Fatalf("succeeded in %s — expected to have waited through the ~600ms gap", elapsed)
+	}
+	select {
+	case e := <-srvErr:
+		t.Fatalf("bringUp failed: %v", e)
+	default:
+	}
+}
+
+// TestRetryTransport_GivesUpAfterBudget proves the retry is BOUNDED: if the
+// writer never returns, the reader stops retrying at the budget and surfaces the
+// dial error (which the proxy renders as 502), rather than hanging forever.
+func TestRetryTransport_GivesUpAfterBudget(t *testing.T) {
+	rt := newRetryTransport(300*time.Millisecond, luxlog.NewNoOpLogger())
+	// 127.0.0.1:1 is reserved/unbound → connection refused (a dial error).
+	req, _ := http.NewRequest(http.MethodGet, "http://127.0.0.1:1/v1/models", nil)
+	start := time.Now()
+	_, err := rt.RoundTrip(req)
+	if err == nil {
+		t.Fatal("expected a dial error after the budget elapsed")
+	}
+	if elapsed := time.Since(start); elapsed < 250*time.Millisecond {
+		t.Fatalf("gave up in %s — should have retried until ~300ms budget", elapsed)
+	}
+	if elapsed := time.Since(start); elapsed > 5*time.Second {
+		t.Fatalf("took %s — retry budget was not honored", elapsed)
+	}
+}
+
+// TestIsDialError_OnlyRetriesUndeliveredRequests pins the safety invariant: only
+// a dial failure (request NEVER delivered) is retryable, so a non-idempotent POST
+// is never double-executed after it reached the writer.
+func TestIsDialError_OnlyRetriesUndeliveredRequests(t *testing.T) {
+	if !isDialError(&net.OpError{Op: "dial", Err: syscall.ECONNREFUSED}) {
+		t.Fatal("a dial connection-refused must be retryable")
+	}
+	if !isDialError(syscall.ECONNREFUSED) {
+		t.Fatal("bare ECONNREFUSED must be retryable")
+	}
+	// A read failure AFTER the request was written is NOT a dial error — ambiguous,
+	// must not be retried.
+	if isDialError(&net.OpError{Op: "read", Err: io.ErrUnexpectedEOF}) {
+		t.Fatal("a post-delivery read error must NOT be retried (ambiguous)")
+	}
+	if isDialError(errors.New("some upstream 500")) {
+		t.Fatal("a generic error must NOT be retried")
+	}
+}
+
+// TestServeReaderProxy_RequiresWriterURL proves a reader fails loud without a
+// target rather than silently serving nothing.
+func TestServeReaderProxy_RequiresWriterURL(t *testing.T) {
+	cfg := &Config{WriterURL: ""}
+	if err := serveReaderProxy(cfg); err == nil {
+		t.Fatal("reader with no CLOUD_WRITER_URL must fail closed")
+	}
+}
+
+// TestReaderProxy_EndToEndTransparent proves the assembled proxy forwards method,
+// path, body, and a header to the writer and streams the response back — the
+// transparent-forwarder contract, exercised through httptest.
+func TestReaderProxy_EndToEndTransparent(t *testing.T) {
+	var gotPath, gotOrg, gotMethod string
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		gotPath, gotMethod, gotOrg = r.URL.Path, r.Method, r.Header.Get("X-Org-Id")
+		b, _ := io.ReadAll(r.Body)
+		w.WriteHeader(http.StatusCreated)
+		_, _ = w.Write(append([]byte("echo:"), b...))
+	}))
+	defer upstream.Close()
+
+	cfg := &Config{WriterURL: upstream.URL, ListenAddr: "127.0.0.1:0", HealthListenAddr: "127.0.0.1:0", ReaderRetryBudget: 2 * time.Second}
+	// Build the proxy handler the same way serveReaderProxy does, and drive it via
+	// httptest so we assert forwarding without binding real ports.
+	proxy, err := newReaderProxy(cfg, luxlog.NewNoOpLogger())
+	if err != nil {
+		t.Fatalf("newReaderProxy: %v", err)
+	}
+	edge := httptest.NewServer(proxy)
+	defer edge.Close()
+
+	req, _ := http.NewRequest(http.MethodPut, edge.URL+"/v1/prompts/foo", strings.NewReader("BODY"))
+	req.Header.Set("X-Org-Id", "acme")
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatalf("edge request: %v", err)
+	}
+	defer resp.Body.Close()
+	body, _ := io.ReadAll(resp.Body)
+	if gotMethod != http.MethodPut || gotPath != "/v1/prompts/foo" || gotOrg != "acme" {
+		t.Fatalf("writer saw method=%q path=%q org=%q — not transparently forwarded", gotMethod, gotPath, gotOrg)
+	}
+	if resp.StatusCode != http.StatusCreated || string(body) != "echo:BODY" {
+		t.Fatalf("edge returned status=%d body=%q — response not passed through", resp.StatusCode, body)
+	}
+}

serve.go

@@ -14,6 +14,7 @@ import (
 	"github.com/hanzoai/cloud/role"
 	"github.com/hanzoai/cloud/writerpin"
 	"github.com/hanzoai/cloud/zapface"
+	luxlog "github.com/luxfi/log"
 	"github.com/zap-proto/zip"
 	"github.com/zap-proto/zip/middleware"
 )
@@ -58,6 +59,33 @@ func Serve(enable []string) error {
 	}
 	cfg.Role = resolvedRole
 
+	// Reader role: a transparent, always-ready reverse proxy to the writer. It
+	// opens NO stores (the KMS ZapDB store is not RO-shareable while the writer is
+	// live — clients/kms.TestConcurrentOpen_LiveWriterStoreIsNotROShareable) and
+	// forwards every request to CLOUD_WRITER_URL, retrying dial-only across the
+	// writer's roll gap so the edge never blips. Returns here — never reaches
+	// BuildDeps. Unset CLOUD_ROLE ⇒ Writer, so this is inert by default.
+	if cfg.Role.IsReader() {
+		return serveReaderProxy(cfg)
+	}
+
+	// Writer role, optional lease. When CLOUD_WRITER_LEASE is set (the surge/
+	// overlap roll topology), take the exclusive writer lease BEFORE opening the
+	// RWO stores and release it LAST (after every store is closed) so a surge
+	// writer never double-opens the exclusive-lock ZapDB/audit stores. Default
+	// OFF: a Recreate single-writer never overlaps and needs no lease, so an unset
+	// variable is byte-identical to today.
+	if cfg.WriterLease {
+		release, lerr := acquireWriterLease(cfg.DataDir, 90*time.Second, luxlog.New("cloud").New("subsystem", "writer-lease"))
+		if lerr != nil {
+			return fmt.Errorf("writer lease: %w", lerr)
+		}
+		// Released after the shutdown path closes every store (ShutdownAll +
+		// audit + gateway-policy) below; defer is the store-close backstop that
+		// also covers early error returns (the kernel reclaims on exit regardless).
+		defer func() { _ = release() }()
+	}
+
 	deps := BuildDeps(cfg)
 
 	// Surface the resolved role and the writer-pin backing it. The pin is
@@ -106,14 +134,8 @@ func Serve(enable []string) error {
 
 	app.Use(middleware.Logger(deps.Logger))
 
-	// Read-replica write guard (fail-closed). On a Reader, refuse mutating verbs
-	// at the boundary so a mis-routed write can never silently persist to the
-	// reader's ephemeral store and vanish on restart. ONE place gates EVERY store
-	// (KMS + audit + tasks + per-tenant SQLite), not just KMS's read-only open.
-	// No-op on a Writer (the default), so unset CLOUD_ROLE is byte-identical.
-	if cfg.Role.IsReader() {
-		app.Use(ReaderGuard())
-	}
+	// (A Reader never reaches here — it returns at serveReaderProxy above, opening
+	// no stores and no middleware pipeline. This body is the Writer path only.)
 
 	// Public site edge (clients/sites). Installed FIRST — after Recover/RequestID/
 	// Logger, BEFORE SanitizeIdentity + BillingGate — so a request whose Host is a

writer_lease.go

@@ -0,0 +1,90 @@
+package cloud
+
+// Writer lease — the cross-process interlock that makes a zero-gap (surge)
+// writer roll SAFE despite the exclusive-lock ZapDB KMS store.
+//
+// The writer embeds stores that permit exactly one live opener: the ZapDB KMS
+// store (an OS-locked Badger fork whose live files are NOT RO-shareable — see
+// clients/kms.TestConcurrentOpen_...) and the audit chain whose head is recovered
+// at open. Under strategy: Recreate the old pod fully terminates before the new
+// one starts, so there is never an overlap and no lease is needed (default OFF).
+//
+// To shrink the roll gap we can instead run the writer as RollingUpdate
+// (maxUnavailable:1, maxSurge:1) with same-node podAffinity so the surge pod is
+// already scheduled and running when the old pod is told to terminate. THEN the
+// lease is what keeps it safe: the new writer blocks in acquireWriterLease before
+// it opens any store, and the old writer releases the lease LAST at shutdown —
+// after every store is closed. So the exclusive stores are handed off, never
+// double-opened, and the gap shrinks from "schedule + pull + full startup"
+// (Recreate) to just "old flush/close + new open".
+//
+// The lock is an flock(2) advisory exclusive lock on {DataDir}/.writer.lock. Two
+// pods sharing the one RWO PVC on the same node interlock through the same inode;
+// a crashed holder's lock is reclaimed by the kernel on process death, so a lease
+// can never be stranded by an OOM/crash.
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"time"
+
+	luxlog "github.com/luxfi/log"
+)
+
+// writerLockName is the lock file under DataDir. It is NOT a store — only a
+// zero-length advisory-lock anchor — so it is safe to create/keep on the PVC.
+const writerLockName = ".writer.lock"
+
+// acquireWriterLease takes the exclusive writer lease, polling (non-blocking
+// flock) until it is free or timeout elapses. On success it returns a release
+// func the caller MUST invoke AFTER every store is closed (the lease's whole
+// purpose is to gate store opens). On timeout it fails CLOSED: a writer that
+// cannot prove it is the sole store-opener refuses to open the stores rather than
+// risk a double-open.
+func acquireWriterLease(dataDir string, timeout time.Duration, log luxlog.Logger) (func() error, error) {
+	if err := os.MkdirAll(dataDir, 0o755); err != nil {
+		return nil, fmt.Errorf("writer lease: data dir %q: %w", dataDir, err)
+	}
+	path := filepath.Join(dataDir, writerLockName)
+	f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600)
+	if err != nil {
+		return nil, fmt.Errorf("writer lease: open %q: %w", path, err)
+	}
+
+	deadline := time.Now().Add(timeout)
+	backoff := 50 * time.Millisecond
+	waited := false
+	for {
+		ok, lerr := tryLockExclusive(f)
+		if lerr != nil {
+			_ = f.Close()
+			return nil, fmt.Errorf("writer lease: lock %q: %w", path, lerr)
+		}
+		if ok {
+			if log != nil {
+				log.Info("writer lease acquired", "path", path, "waited", waited)
+			}
+			return func() error {
+				uerr := unlockFile(f)
+				cerr := f.Close()
+				if uerr != nil {
+					return uerr
+				}
+				return cerr
+			}, nil
+		}
+		if time.Now().After(deadline) {
+			_ = f.Close()
+			return nil, fmt.Errorf("writer lease: %q still held by another writer after %s — refusing to open the exclusive stores (fail-closed; the previous writer has not released it — inspect its shutdown)", path, timeout)
+		}
+		waited = true
+		if log != nil {
+			log.Info("writer lease held by peer; waiting for handoff", "path", path, "backoff", backoff.String())
+		}
+		time.Sleep(backoff)
+		if backoff < time.Second {
+			backoff *= 2
+		}
+	}
+}

writer_lease_other.go

@@ -0,0 +1,12 @@
+//go:build !unix
+
+package cloud
+
+import "os"
+
+// Non-unix build: the writer lease is a no-op. Production cloud runs on Linux;
+// this keeps the package buildable on other platforms (dev tooling) without
+// pulling in a platform lock. A single-writer Recreate deployment needs no lease
+// anyway, and the surge topology that requires it is Linux-only.
+func tryLockExclusive(*os.File) (bool, error) { return true, nil }
+func unlockFile(*os.File) error               { return nil }

writer_lease_test.go

@@ -0,0 +1,94 @@
+package cloud
+
+import (
+	"sync/atomic"
+	"testing"
+	"time"
+
+	luxlog "github.com/luxfi/log"
+)
+
+// TestWriterLease_SerializesHandoff proves the surge-roll safety property: while
+// one writer holds the lease, a second cannot acquire it; the instant the first
+// releases (as it would AFTER closing its stores at shutdown), the second
+// acquires. This is exactly the ZapDB/audit store handoff — never a double-open.
+func TestWriterLease_SerializesHandoff(t *testing.T) {
+	dir := t.TempDir()
+	log := luxlog.NewNoOpLogger()
+
+	release1, err := acquireWriterLease(dir, 2*time.Second, log)
+	if err != nil {
+		t.Fatalf("first acquire: %v", err)
+	}
+
+	// A second writer must NOT acquire while the first holds it.
+	var acquired atomic.Bool
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		release2, err := acquireWriterLease(dir, 5*time.Second, log)
+		if err != nil {
+			t.Errorf("second acquire (after handoff): %v", err)
+			return
+		}
+		acquired.Store(true)
+		_ = release2()
+	}()
+
+	time.Sleep(300 * time.Millisecond)
+	if acquired.Load() {
+		t.Fatal("second writer acquired the lease while the first still held it — double-open possible")
+	}
+
+	// Release the first (post-store-close). The second must now acquire promptly.
+	if err := release1(); err != nil {
+		t.Fatalf("release1: %v", err)
+	}
+	select {
+	case <-done:
+		if !acquired.Load() {
+			t.Fatal("second writer did not acquire after handoff")
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatal("second writer never acquired after the first released (handoff stuck)")
+	}
+}
+
+// TestWriterLease_FailsClosedOnTimeout proves that a writer which cannot get the
+// lease within the timeout REFUSES to proceed (fail-closed) rather than opening
+// the exclusive stores beside a live holder.
+func TestWriterLease_FailsClosedOnTimeout(t *testing.T) {
+	dir := t.TempDir()
+	log := luxlog.NewNoOpLogger()
+
+	release, err := acquireWriterLease(dir, time.Second, log)
+	if err != nil {
+		t.Fatalf("holder acquire: %v", err)
+	}
+	defer release()
+
+	start := time.Now()
+	if _, err := acquireWriterLease(dir, 400*time.Millisecond, log); err == nil {
+		t.Fatal("expected fail-closed timeout while the lease is held")
+	}
+	if elapsed := time.Since(start); elapsed < 350*time.Millisecond {
+		t.Fatalf("gave up in %s — should have waited the ~400ms timeout", elapsed)
+	}
+}
+
+// TestWriterLease_ReacquireAfterRelease proves a clean single-writer restart
+// (Recreate) reacquires with no wait — the default topology stays byte-identical
+// in behavior (acquire returns immediately when uncontended).
+func TestWriterLease_ReacquireAfterRelease(t *testing.T) {
+	dir := t.TempDir()
+	log := luxlog.NewNoOpLogger()
+	for i := 0; i < 3; i++ {
+		release, err := acquireWriterLease(dir, time.Second, log)
+		if err != nil {
+			t.Fatalf("acquire %d: %v", i, err)
+		}
+		if err := release(); err != nil {
+			t.Fatalf("release %d: %v", i, err)
+		}
+	}
+}

writer_lease_unix.go

@@ -0,0 +1,29 @@
+//go:build unix
+
+package cloud
+
+import (
+	"os"
+
+	"golang.org/x/sys/unix"
+)
+
+// tryLockExclusive attempts a non-blocking exclusive flock. It returns
+// (true, nil) when the lease is acquired, (false, nil) when another live opener
+// holds it (EWOULDBLOCK), and (false, err) on a real error.
+func tryLockExclusive(f *os.File) (bool, error) {
+	err := unix.Flock(int(f.Fd()), unix.LOCK_EX|unix.LOCK_NB)
+	if err == nil {
+		return true, nil
+	}
+	if err == unix.EWOULDBLOCK {
+		return false, nil
+	}
+	return false, err
+}
+
+// unlockFile releases the flock. The kernel also releases it on fd close /
+// process death, so this is the graceful path, not the only one.
+func unlockFile(f *os.File) error {
+	return unix.Flock(int(f.Fd()), unix.LOCK_UN)
+}