amadan

carlos / platform Public

console: one policy writer per bucket, on the platform's lease (#328, #329)

Ruling 10: the app policy writer lease, internal/lease's lock at
control/app-policy-lease/lock.json in the deployment bucket. Lazy, then
long-lived; numbered holdings; policyAttempt binds every signing and
publication attempt to the same live holding inside its safe deadline
(Astra plan R3-1). A directory deployment is the sole writer; a bucket
deployment without the leaser refuses. cmd/console wires
ProductionPolicyLeaser from CARLOS_DEPLOYMENT_BUCKET and releases the
lease on SIGTERM or interrupt.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 389c85d7bc667e0726b036411c4bd641de642c8b parent 2133c5d
5 files changed, +677 −1
  • cmd/console/main.go +27 −0
  • cmd/console/wiring_test.go +27 −1
  • internal/console/policylease.go +249 −0
  • internal/console/policylease_test.go +359 −0
  • internal/console/server.go +15 −0
diff --git a/cmd/console/main.go b/cmd/console/main.go
index d78578a29..1c02fb51f 100644
--- a/cmd/console/main.go
+++ b/cmd/console/main.go
@@ -129,10 +129,12 @@ import (
"math/rand/v2"
"net"
"os"
+ "os/signal"
"path/filepath"
"sort"
"strconv"
"strings"
+ "syscall"
"time"
"unicode/utf8"
@@ -407,6 +409,10 @@ func serve() {
Bucket: os.Getenv("CARLOS_DEPLOYMENT_BUCKET"),
HomeRegion: strings.TrimSpace(os.Getenv("CONSOLE_HOME_REGION")),
RegionHA: regionHAEnabled(os.Getenv),
+ // The app policy writer lease (ruling 10), in the deployment bucket. A
+ // CARLOS_DEPLOYMENT_DIR deployment has none: its console is the sole
+ // writer.
+ PolicyLease: policyLeaseFor(os.Getenv),
// The deployment's AWS account id, the trust anchor of the Phase 1
// move door's box identity check (unset: no box verifies).
AWSAccount: strings.TrimSpace(os.Getenv("CARLOS_AWS_ACCOUNT")),
@@ -521,6 +527,17 @@ func serve() {
if err != nil {
log.Fatal(err)
}
+ // A graceful stop releases the app policy writer lease, so a rolled
+ // console's successor need not wait out its TTL (the housekeeper's
+ // "graceful stand-down must be reachable"). Nothing else here shuts down
+ // gracefully, so the handler exits straight after the release.
+ go func() {
+ sig := make(chan os.Signal, 1)
+ signal.Notify(sig, syscall.SIGTERM, os.Interrupt)
+ <-sig
+ srv.ReleasePolicyWriter(context.Background())
+ os.Exit(0)
+ }()
// The console mailer's identity ensure (consoleMailer above) — makes or
// re-reads the sending domain's SES/DKIM identity. Started here, not in
// NewServer, same reasoning as the alert watcher below: constructing a
@@ -1759,6 +1776,16 @@ func sweepMailBoundaries(ctx context.Context, claims bool, p mail.Provisioner, l
logf("console: mail boundary sweep: all %d mail users bounded (%d already, %d fixed)", res.Total, res.Bounded, len(res.Fixed))
}
+// policyLeaseFor is the app policy writer's leaser: the production one on a
+// deployment with a bucket (CARLOS_DEPLOYMENT_BUCKET), nil on a
+// CARLOS_DEPLOYMENT_DIR deployment, whose console is the sole writer.
+func policyLeaseFor(getenv func(string) string) console.PolicyLeaser {
+ if getenv("CARLOS_DEPLOYMENT_BUCKET") == "" {
+ return nil
+ }
+ return console.ProductionPolicyLeaser
+}
+
// regionHAEnabled is CARLOS_REGION_HA as strconv.ParseBool reads it: on
// for "1", "t", "true" in any case, off for "", "0", "false" and anything
// unparseable. The same rule as the box's (internal/host/boxrecord.go).
diff --git a/cmd/console/wiring_test.go b/cmd/console/wiring_test.go
index 415e59a8f..64bf748f5 100644
--- a/cmd/console/wiring_test.go
+++ b/cmd/console/wiring_test.go
@@ -1,6 +1,11 @@
package main
-import "testing"
+import (
+ "reflect"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/console"
+)
// The wiring resolver, whose refusals are otherwise a fatal path nobody can
// exercise.
@@ -96,3 +101,24 @@ func TestDBDoorWiringIsOffWithoutABucket(t *testing.T) {
t.Fatalf("bucket = %q, want the door off", bucket)
}
}
+
+// The policy writer lease (ruling 10) follows the deployment bucket: a
+// bucket deployment gets the production leaser, a CARLOS_DEPLOYMENT_DIR one
+// none (its console is the sole writer by construction). In this file rather
+// than main_test.go because that file is package main_test and cannot reach
+// policyLeaseFor.
+func TestPolicyLeaseForFollowsTheDeploymentBucket(t *testing.T) {
+ env := func(m map[string]string) func(string) string {
+ return func(k string) string { return m[k] }
+ }
+ got := policyLeaseFor(env(map[string]string{"CARLOS_DEPLOYMENT_BUCKET": "deploy"}))
+ if got == nil {
+ t.Fatal("a bucket deployment has no policy leaser")
+ }
+ if reflect.ValueOf(got).Pointer() != reflect.ValueOf(console.ProductionPolicyLeaser).Pointer() {
+ t.Fatal("a bucket deployment's policy leaser is not console.ProductionPolicyLeaser")
+ }
+ if got := policyLeaseFor(env(map[string]string{"CARLOS_DEPLOYMENT_DIR": "/srv/deploy"})); got != nil {
+ t.Fatal("a directory deployment has a policy leaser")
+ }
+}
diff --git a/internal/console/policylease.go b/internal/console/policylease.go
new file mode 100644
index 000000000..cf8d7b33a
--- /dev/null
+++ b/internal/console/policylease.go
@@ -0,0 +1,249 @@
+// The policy writer lease: one console signs app policies per deployment
+// bucket (spec 2026-10-21-instance-sizes-and-egress-design.md, ruling 10).
+// Two consoles signing the same app's policy would each build on what they
+// last saw, and the later write would silently undo the earlier one, a revoke
+// included. So nothing signs a policy, by door write or re-sign, without this
+// lease, and a second console refuses and names the holder.
+//
+// The mechanism is internal/lease, the platform's one election (litestream's
+// conditional-write lock.json), already reused by the housekeeper under the
+// rule "no second election pattern may be introduced" (internal/host/
+// housekeeper.go). There is no store-CAS record of our own and no bespoke
+// TTL: the lock is control/app-policy-lease/lock.json in the deployment
+// bucket, held for lease.LeaseTTL and renewed by internal/lease every TTL/3.
+//
+// It is lazy, then long-lived. Nothing is acquired at start, so a deployment
+// that never writes a policy or creates an app sends no lease requests at all.
+// The first need (a door PUT, a re-sign pass with something to sign, an app
+// create) acquires it, and it is kept for as long as renewal wins. While held
+// it costs one conditional PUT per renewal, 288 a day for the one writer
+// console, outside release.Store and independent of apps and boxes. A refused
+// console tries again only on its next need, never on a timer.
+//
+// A holding is numbered. Each successful acquisition is a new holding, one
+// above the last, and a new holding means "another console may have written
+// since I last did": the door (plan Task 9) uses the number to decide when to
+// rebuild on the bucket's verified record instead of its own last write.
+//
+// Every write is bound to its holding (Astra plan R3-1). A writer keeps the
+// holding policyWriter gave it and calls policyAttempt before each signing and
+// each publication attempt, retries included. A write that outlived its
+// holding could overwrite the next writer's content, a revoke among it, so the
+// attempt is refused unless the same holding is still current, its lease is
+// healthy and its safe deadline is at least policyAttemptWindow away, and the
+// attempt then runs under a context that ends at that window, at the safe
+// deadline, or the moment the lease reports Lost. The bound: SafeUntil is
+// lease.RenewMargin (5 s) before the lock's expiry and no other console can
+// acquire before that expiry, so an attempt either reaches the store while
+// this console still holds the lock or is cancelled. A request the network
+// holds back past its own cancellation is the residual; a later holder's
+// conditional write then sees a changed ETag (Task 9).
+//
+// Release is reachable: cmd/console calls ReleasePolicyWriter on SIGTERM, so
+// a rolled console's successor need not wait out the TTL. Where the console's
+// role is denied s3:DeleteObject on the bucket (the flagship's store IAM Deny;
+// plan 2026-10-21-instance-policy-1-foundation, "Boundary verification"), the
+// release fails, is logged, and the successor waits up to lease.LeaseTTL, 15
+// minutes.
+package console
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "log"
+ "net/http"
+ "time"
+
+ "github.com/carlosframework/platform/internal/lease"
+)
+
+// appPolicyLeasePrefix is where the lock lands in the deployment bucket:
+// <prefix>/lock.json.
+const appPolicyLeasePrefix = "control/app-policy-lease"
+
+// policyAttemptWindow is the longest one signing or publication attempt may
+// run, and the least time its holding must have left before its safe deadline.
+const policyAttemptWindow = 10 * time.Second
+
+// policyLeaseReleaseBudget bounds a release, as the housekeeper's does.
+const policyLeaseReleaseBudget = 5 * time.Second
+
+// Door copy (collected for review in plan Task 15).
+const (
+ policyLeaseHeldMsg = "another console (%s) is the policy writer for this deployment until %s. Make the change there, or try again after that"
+ policyLeaseUnavailableMsg = "could not take the policy writer lease. Try again"
+ policyLeaseMissingMsg = "this console is not set up to write app policies (no writer lease). Deploy a console built from current main"
+ policyLeaseLostMsg = "this console stopped being the policy writer during the change, so it may not have been published. Read the policy again before you retry"
+)
+
+// PolicyLease is the slice of *lease.Held the writer uses.
+type PolicyLease interface {
+ Err() error
+ Release(ctx context.Context) error
+ Lost() <-chan struct{}
+ SafeUntil() (time.Time, error)
+}
+
+var _ PolicyLease = (*lease.Held)(nil)
+
+// PolicyLeaser acquires the policy writer lease: Config.PolicyLease.
+type PolicyLeaser func(ctx context.Context, bucket, prefix string, ttl time.Duration) (PolicyLease, error)
+
+// ProductionPolicyLeaser is lease.AcquireTTL with the region left to the AWS
+// default chain, the same answer release.NewS3Store signs for against this
+// bucket (acquireDeploymentLease's reasoning). A func, not a method value,
+// because AcquireTTL returns a typed nil *lease.Held with its error, which
+// would arrive as a non-nil PolicyLease.
+func ProductionPolicyLeaser(ctx context.Context, bucket, prefix string, ttl time.Duration) (PolicyLease, error) {
+ h, err := lease.AcquireTTL(ctx, bucket, prefix, "", ttl)
+ if err != nil {
+ return nil, err
+ }
+ return h, nil
+}
+
+// policyHolding is one acquisition of the writer lease. n counts holdings
+// from 1; sole is a directory deployment's console, the only writer by
+// construction, whose lease is nil.
+type policyHolding struct {
+ n uint64
+ sole bool
+ lease PolicyLease
+}
+
+// policyWriter returns the current holding, acquiring the lease if there is
+// none or the last one was lost. The caller keeps the holding for its whole
+// write and checks it with policyAttempt before every signing and publication.
+func (s *Server) policyWriter(ctx context.Context) (policyHolding, *doorRefusal) {
+ if s.cfg.Bucket == "" {
+ return policyHolding{n: 1, sole: true}, nil
+ }
+ if s.cfg.PolicyLease == nil {
+ return policyHolding{}, &doorRefusal{http.StatusServiceUnavailable, policyLeaseMissingMsg}
+ }
+ s.policyLeaseMu.Lock()
+ defer s.policyLeaseMu.Unlock()
+ if cur := s.policyCur.Load(); cur != nil {
+ err := cur.lease.Err()
+ if err == nil {
+ return *cur, nil
+ }
+ s.endPolicyHolding(cur, err)
+ }
+ ttl := s.cfg.PolicyLeaseTTL
+ if ttl == 0 {
+ ttl = lease.LeaseTTL
+ }
+ l, err := s.cfg.PolicyLease(ctx, s.cfg.Bucket, appPolicyLeasePrefix, ttl)
+ if err == nil && l == nil {
+ err = errors.New("the leaser returned no lease")
+ }
+ if err != nil {
+ var held *lease.HeldError
+ if errors.As(err, &held) {
+ return policyHolding{}, &doorRefusal{http.StatusServiceUnavailable,
+ fmt.Sprintf(policyLeaseHeldMsg, held.Owner, held.ExpiresAt.UTC().Format(time.RFC3339))}
+ }
+ log.Printf("console: policy writer lease: %v", err)
+ return policyHolding{}, &doorRefusal{http.StatusServiceUnavailable, policyLeaseUnavailableMsg}
+ }
+ s.policyN++
+ h := &policyHolding{n: s.policyN, lease: l}
+ s.policyCur.Store(h)
+ return *h, nil
+}
+
+// policyAttempt is the check before each signing and each publication
+// attempt (Astra plan R3-1). It refuses unless h is still the current holding
+// and its lease is healthy with at least policyAttemptWindow left before its
+// safe deadline; otherwise it returns the context the attempt must run under,
+// which ends at the window, at the safe deadline, or when the lease is lost.
+// An attempt cut off by that context is this same refusal, not a store
+// failure. A sole holding is never refused and its context is ctx itself.
+func (s *Server) policyAttempt(ctx context.Context, h policyHolding) (context.Context, context.CancelFunc, *doorRefusal) {
+ if h.sole {
+ return ctx, func() {}, nil
+ }
+ lost := &doorRefusal{http.StatusServiceUnavailable, policyLeaseLostMsg}
+ cur := s.policyCur.Load()
+ if cur == nil || cur.n != h.n || h.lease == nil {
+ return ctx, func() {}, lost
+ }
+ if err := h.lease.Err(); err != nil {
+ s.endPolicyHolding(cur, err)
+ return ctx, func() {}, lost
+ }
+ select {
+ case <-h.lease.Lost():
+ s.endPolicyHolding(cur, errors.New("the lease reported lost"))
+ return ctx, func() {}, lost
+ default:
+ }
+ safe, err := h.lease.SafeUntil()
+ if err != nil {
+ s.endPolicyHolding(cur, err)
+ return ctx, func() {}, lost
+ }
+ now := time.Now()
+ if safe.Sub(now) < policyAttemptWindow {
+ s.endPolicyHolding(cur, fmt.Errorf("safe until %s, less than %s away", safe.UTC().Format(time.RFC3339), policyAttemptWindow))
+ return ctx, func() {}, lost
+ }
+ deadline := now.Add(policyAttemptWindow)
+ if safe.Before(deadline) {
+ deadline = safe
+ }
+ actx, cancel := context.WithDeadline(ctx, deadline)
+ go func() {
+ select {
+ case <-h.lease.Lost():
+ cancel()
+ case <-actx.Done():
+ }
+ }()
+ return actx, cancel, nil
+}
+
+// policyHeld reports, without blocking, whether this console holds the
+// writer lease and its renewal is healthy. A sole console always does.
+func (s *Server) policyHeld() bool {
+ if s.cfg.Bucket == "" {
+ return true
+ }
+ if s.cfg.PolicyLease == nil {
+ return false
+ }
+ cur := s.policyCur.Load()
+ return cur != nil && cur.lease.Err() == nil
+}
+
+// ReleasePolicyWriter ends the holding and releases its lease within
+// policyLeaseReleaseBudget, so a successor console need not wait out the TTL.
+// Nothing held is a no-op. It does not take policyLeaseMu, so a stand-down
+// never waits on an acquisition in flight.
+func (s *Server) ReleasePolicyWriter(ctx context.Context) {
+ cur := s.policyCur.Swap(nil)
+ if cur == nil {
+ return
+ }
+ releasePolicyLease(ctx, cur.lease)
+}
+
+// endPolicyHolding ends cur if it is still the current holding, logging why
+// and releasing its lease best-effort. Only the caller that ends it releases.
+func (s *Server) endPolicyHolding(cur *policyHolding, why error) {
+ if !s.policyCur.CompareAndSwap(cur, nil) {
+ return
+ }
+ log.Printf("console: policy writer lease lost: %v", why)
+ releasePolicyLease(context.Background(), cur.lease)
+}
+
+func releasePolicyLease(ctx context.Context, l PolicyLease) {
+ ctx, cancel := context.WithTimeout(ctx, policyLeaseReleaseBudget)
+ defer cancel()
+ if err := l.Release(ctx); err != nil {
+ log.Printf("console: policy writer lease release: %v", err)
+ }
+}
diff --git a/internal/console/policylease_test.go b/internal/console/policylease_test.go
new file mode 100644
index 000000000..563939912
--- /dev/null
+++ b/internal/console/policylease_test.go
@@ -0,0 +1,359 @@
+package console
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "net/http"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/lease"
+)
+
+// fakePolicyLease is a PolicyLease whose health the test sets.
+type fakePolicyLease struct {
+ mu sync.Mutex
+ err error
+ safe time.Time
+ safeErr error
+ lost chan struct{}
+ releases int
+}
+
+func newFakePolicyLease() *fakePolicyLease {
+ return &fakePolicyLease{safe: time.Now().Add(10 * time.Minute), lost: make(chan struct{})}
+}
+
+func (f *fakePolicyLease) Err() error {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return f.err
+}
+
+func (f *fakePolicyLease) Release(context.Context) error {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ f.releases++
+ return nil
+}
+
+func (f *fakePolicyLease) Lost() <-chan struct{} { return f.lost }
+
+func (f *fakePolicyLease) SafeUntil() (time.Time, error) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return f.safe, f.safeErr
+}
+
+func (f *fakePolicyLease) set(err error, safe time.Time, safeErr error) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ f.err, f.safe, f.safeErr = err, safe, safeErr
+}
+
+func (f *fakePolicyLease) released() int {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return f.releases
+}
+
+type policyLeaseCall struct {
+ bucket, prefix string
+ ttl time.Duration
+}
+
+// fakePolicyLeaser records its calls and hands out a fresh fake lease per
+// call, or fails with err when it is set.
+type fakePolicyLeaser struct {
+ mu sync.Mutex
+ calls []policyLeaseCall
+ leases []*fakePolicyLease
+ err error
+}
+
+func (f *fakePolicyLeaser) leaser(_ context.Context, bucket, prefix string, ttl time.Duration) (PolicyLease, error) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ f.calls = append(f.calls, policyLeaseCall{bucket, prefix, ttl})
+ if f.err != nil {
+ return nil, f.err
+ }
+ l := newFakePolicyLease()
+ f.leases = append(f.leases, l)
+ return l, nil
+}
+
+func (f *fakePolicyLeaser) callCount() int {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return len(f.calls)
+}
+
+func (f *fakePolicyLeaser) lease(i int) *fakePolicyLease {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return f.leases[i]
+}
+
+func policyLeaseServer(f *fakePolicyLeaser) *Server {
+ return &Server{cfg: Config{Bucket: "deploy", PolicyLease: f.leaser}}
+}
+
+func wantPolicyRefusal(t *testing.T, ref *doorRefusal, msg string) {
+ t.Helper()
+ if ref == nil {
+ t.Fatalf("no refusal, want 503 %q", msg)
+ }
+ if ref.Status != http.StatusServiceUnavailable || ref.Msg != msg {
+ t.Fatalf("refusal %d %q, want 503 %q", ref.Status, ref.Msg, msg)
+ }
+}
+
+func TestPolicyWriterIsSoleWithoutABucket(t *testing.T) {
+ t.Parallel()
+ f := &fakePolicyLeaser{}
+ s := &Server{cfg: Config{PolicyLease: f.leaser}}
+ h, ref := s.policyWriter(context.Background())
+ if ref != nil {
+ t.Fatalf("refused: %+v", ref)
+ }
+ if h.n != 1 || !h.sole || h.lease != nil {
+ t.Fatalf("holding %+v, want {1 true nil}", h)
+ }
+ if f.callCount() != 0 {
+ t.Fatalf("the leaser was called %d times on a directory deployment", f.callCount())
+ }
+ if !s.policyHeld() {
+ t.Fatal("policyHeld false on a sole console")
+ }
+}
+
+func TestPolicyWriterRefusesWithoutALeaser(t *testing.T) {
+ t.Parallel()
+ s := &Server{cfg: Config{Bucket: "deploy"}}
+ _, ref := s.policyWriter(context.Background())
+ wantPolicyRefusal(t, ref, policyLeaseMissingMsg)
+ if s.policyHeld() {
+ t.Fatal("policyHeld true without a leaser")
+ }
+}
+
+func TestPolicyWriterAcquiresLazilyAndKeepsTheHolding(t *testing.T) {
+ t.Parallel()
+ f := &fakePolicyLeaser{}
+ s := policyLeaseServer(f)
+ if f.callCount() != 0 || s.policyHeld() {
+ t.Fatalf("a new server: %d calls, held %v", f.callCount(), s.policyHeld())
+ }
+
+ h, ref := s.policyWriter(context.Background())
+ if ref != nil {
+ t.Fatalf("refused: %+v", ref)
+ }
+ want := policyLeaseCall{"deploy", "control/app-policy-lease", lease.LeaseTTL}
+ if f.callCount() != 1 || f.calls[0] != want {
+ t.Fatalf("calls %+v, want [%+v]", f.calls, want)
+ }
+ if h.n != 1 || h.sole || h.lease != PolicyLease(f.lease(0)) {
+ t.Fatalf("holding %+v", h)
+ }
+ if !s.policyHeld() {
+ t.Fatal("policyHeld false while held")
+ }
+
+ h, ref = s.policyWriter(context.Background())
+ if ref != nil || h.n != 1 || f.callCount() != 1 {
+ t.Fatalf("second call: holding %+v, refusal %+v, %d calls", h, ref, f.callCount())
+ }
+
+ f.lease(0).set(errors.New("renewal lost"), time.Now().Add(10*time.Minute), nil)
+ if s.policyHeld() {
+ t.Fatal("policyHeld true once Err is set")
+ }
+ h, ref = s.policyWriter(context.Background())
+ if ref != nil {
+ t.Fatalf("refused after a lost renewal: %+v", ref)
+ }
+ if got := f.lease(0).released(); got != 1 {
+ t.Fatalf("the lost lease was released %d times, want 1", got)
+ }
+ if f.callCount() != 2 || h.n != 2 || h.lease != PolicyLease(f.lease(1)) {
+ t.Fatalf("after loss: %d calls, holding %+v", f.callCount(), h)
+ }
+ if !s.policyHeld() {
+ t.Fatal("policyHeld false on the fresh holding")
+ }
+}
+
+func TestPolicyWriterNamesTheHolder(t *testing.T) {
+ t.Parallel()
+ t1 := time.Date(2026, 10, 25, 14, 30, 0, 0, time.FixedZone("x", 3600))
+ f := &fakePolicyLeaser{err: fmt.Errorf("acquire lease: %w", &lease.HeldError{Owner: "console-b:4242", ExpiresAt: t1})}
+ s := policyLeaseServer(f)
+ _, ref := s.policyWriter(context.Background())
+ if ref == nil || ref.Status != http.StatusServiceUnavailable {
+ t.Fatalf("refusal %+v, want 503", ref)
+ }
+ if !strings.Contains(ref.Msg, "another console (console-b:4242)") || !strings.Contains(ref.Msg, t1.UTC().Format(time.RFC3339)) {
+ t.Fatalf("refusal %q does not name the holder and its expiry", ref.Msg)
+ }
+ if ref.Msg != fmt.Sprintf(policyLeaseHeldMsg, "console-b:4242", "2026-10-25T13:30:00Z") {
+ t.Fatalf("refusal %q", ref.Msg)
+ }
+ if s.policyHeld() {
+ t.Fatal("policyHeld true after a refused acquire")
+ }
+
+ f.mu.Lock()
+ f.err = errors.New("dial tcp: no route to host")
+ f.mu.Unlock()
+ _, ref = s.policyWriter(context.Background())
+ wantPolicyRefusal(t, ref, policyLeaseUnavailableMsg)
+}
+
+func TestReleasePolicyWriterEndsTheHolding(t *testing.T) {
+ t.Parallel()
+ f := &fakePolicyLeaser{}
+ s := policyLeaseServer(f)
+ s.ReleasePolicyWriter(context.Background()) // nothing held: a no-op
+ if _, ref := s.policyWriter(context.Background()); ref != nil {
+ t.Fatalf("refused: %+v", ref)
+ }
+ s.ReleasePolicyWriter(context.Background())
+ if got := f.lease(0).released(); got != 1 {
+ t.Fatalf("released %d times, want 1", got)
+ }
+ if s.policyHeld() {
+ t.Fatal("policyHeld true after ReleasePolicyWriter")
+ }
+ s.ReleasePolicyWriter(context.Background())
+ if got := f.lease(0).released(); got != 1 {
+ t.Fatalf("a second ReleasePolicyWriter released again: %d", got)
+ }
+}
+
+// Astra plan R3-1. Fails if any check is dropped.
+func TestPolicyAttemptBindsToTheSameLiveHolding(t *testing.T) {
+ t.Parallel()
+ bg := context.Background()
+
+ // Sole: the caller's context, untouched.
+ sole := &Server{}
+ sh, _ := sole.policyWriter(bg)
+ actx, cancel, ref := sole.policyAttempt(bg, sh)
+ if ref != nil || actx != bg {
+ t.Fatalf("sole: ctx %v, refusal %+v", actx, ref)
+ }
+ if _, ok := actx.Deadline(); ok {
+ t.Fatal("sole: a deadline was set")
+ }
+ cancel()
+
+ f := &fakePolicyLeaser{}
+ s := policyLeaseServer(f)
+ attempt := func(h policyHolding) (context.Context, *doorRefusal) {
+ t.Helper()
+ actx, cancel, ref := s.policyAttempt(bg, h)
+ t.Cleanup(cancel)
+ return actx, ref
+ }
+ wantWindow := func(actx context.Context, before time.Time) {
+ t.Helper()
+ dl, ok := actx.Deadline()
+ if !ok {
+ t.Fatal("no deadline on the attempt context")
+ }
+ if dl.Before(before.Add(policyAttemptWindow)) || dl.After(time.Now().Add(policyAttemptWindow)) {
+ t.Fatalf("deadline %v is not policyAttemptWindow from now (%v)", dl, before)
+ }
+ }
+
+ h1, ref := s.policyWriter(bg)
+ if ref != nil {
+ t.Fatalf("refused: %+v", ref)
+ }
+ l1 := f.lease(0)
+
+ // Healthy, safe 10 minutes out: the window bounds the attempt.
+ before := time.Now()
+ actx, ref = attempt(h1)
+ if ref != nil {
+ t.Fatalf("healthy holding refused: %+v", ref)
+ }
+ wantWindow(actx, before)
+
+ // Safe 30 minutes out: still the window, never longer.
+ l1.set(nil, time.Now().Add(30*time.Minute), nil)
+ before = time.Now()
+ actx, ref = attempt(h1)
+ if ref != nil {
+ t.Fatalf("refused: %+v", ref)
+ }
+ wantWindow(actx, before)
+
+ // Safe inside the window: refused, and the holding ends.
+ l1.set(nil, time.Now().Add(8*time.Second), nil)
+ _, ref = attempt(h1)
+ wantPolicyRefusal(t, ref, policyLeaseLostMsg)
+ if l1.released() != 1 || s.policyHeld() {
+ t.Fatalf("a refused holding did not end: released %d, held %v", l1.released(), s.policyHeld())
+ }
+
+ // SafeUntil failing: refused.
+ h2, ref := s.policyWriter(bg)
+ if ref != nil || h2.n != 2 {
+ t.Fatalf("holding %+v, refusal %+v", h2, ref)
+ }
+ l2 := f.lease(1)
+ l2.set(nil, time.Now().Add(10*time.Minute), errors.New("lease spent"))
+ _, ref = attempt(h2)
+ wantPolicyRefusal(t, ref, policyLeaseLostMsg)
+
+ // Err set: refused.
+ h3, ref := s.policyWriter(bg)
+ if ref != nil || h3.n != 3 {
+ t.Fatalf("holding %+v, refusal %+v", h3, ref)
+ }
+ l3 := f.lease(2)
+ l3.set(errors.New("renewal lost"), time.Now().Add(10*time.Minute), nil)
+ _, ref = attempt(h3)
+ wantPolicyRefusal(t, ref, policyLeaseLostMsg)
+
+ // A later holding does not revive an earlier one, even one whose lease
+ // looks healthy again.
+ h4, ref := s.policyWriter(bg)
+ if ref != nil || h4.n != 4 {
+ t.Fatalf("holding %+v, refusal %+v", h4, ref)
+ }
+ l3.set(nil, time.Now().Add(10*time.Minute), nil)
+ if _, ref = attempt(h3); ref == nil {
+ t.Fatal("an earlier holding was accepted after a later one began")
+ }
+ wantPolicyRefusal(t, ref, policyLeaseLostMsg)
+ if _, ref = attempt(h1); ref == nil {
+ t.Fatal("holding 1 was accepted while holding 4 is current")
+ }
+ actx, ref = attempt(h4)
+ if ref != nil {
+ t.Fatalf("the live holding was refused: %+v", ref)
+ }
+
+ // The lease's Lost channel closing ends the attempt at once.
+ select {
+ case <-actx.Done():
+ t.Fatal("the attempt context ended before the lease was lost")
+ default:
+ }
+ close(f.lease(3).lost)
+ select {
+ case <-actx.Done():
+ case <-time.After(5 * time.Second):
+ t.Fatal("the attempt context outlived the lease's Lost")
+ }
+ // And a lease already lost refuses the next attempt outright.
+ _, ref = attempt(h4)
+ wantPolicyRefusal(t, ref, policyLeaseLostMsg)
+}
diff --git a/internal/console/server.go b/internal/console/server.go
index 433d08e15..99e089fdc 100644
--- a/internal/console/server.go
+++ b/internal/console/server.go
@@ -520,6 +520,13 @@ type Config struct {
// struct. "" is legal — a local CARLOS_DEPLOYMENT_DIR deployment has no
// bucket name to draw — and simply draws no bucket node.
Bucket string
+ // PolicyLease acquires the app policy writer lease in Bucket
+ // (policylease.go, ruling 10): ProductionPolicyLeaser when Bucket is set.
+ // With Bucket set and this nil, every policy write refuses 503; with
+ // Bucket "" the console is the sole writer and this is never called.
+ PolicyLease PolicyLeaser
+ // PolicyLeaseTTL is the writer lease's TTL; 0 is lease.LeaseTTL.
+ PolicyLeaseTTL time.Duration
// RegionHA is CARLOS_REGION_HA as strconv.ParseBool reads it (region HA
// spec 1 §8 step 2): with it off the console chooses no executor at
// create and relocation has no console half. A deployment fact, on
@@ -971,6 +978,14 @@ type Server struct {
// under doorPublishMu, so the zero Server needs no setup.
doorPublishMu sync.Mutex
doorPublish map[string]*sync.Mutex
+ // The app policy writer lease (policylease.go). policyLeaseMu serializes
+ // acquisition and the holding count policyN; policyCur is the current
+ // holding, nil when none, read without the mutex so policyHeld and
+ // policyAttempt never wait on an acquisition. The zero Server needs no
+ // setup.
+ policyLeaseMu sync.Mutex
+ policyN uint64
+ policyCur atomic.Pointer[policyHolding]
// sendApp is how the alert watcher (alertwatch.go) delivers keymail:
// km.SendApp in production, a capture in tests. A field rather than a
// direct call so an alertPass assertion needs no keymail server.