feat: node cordon/drain (#36)

docs/deploy.md

@@ -195,6 +195,12 @@ WantedBy=multi-user.target
 Stopping sandboxd leaves VMs alive; the next start reconciles them. Claimed
 sandboxes are reaped when their TTL expires (default 5m, capped at 24h).
 
+To empty a node for maintenance, cordon it first:
+[`POST /v1/drain`](sandboxd-api.md#post-v1drain) (root) stops new claims and
+drains the warm pools; poll `GET /v1/info` until `claimed` reaches zero (or
+let the leases expire), then stop sandboxd. `DELETE /v1/drain` uncordons.
+The drain is not persisted — a restarted node serves again.
+
 ## Verifying a node
 
 ```bash

docs/sandboxd-api.md

@@ -165,6 +165,21 @@ Answers the fresh `GET /v1/info` payload. 400 bad key, negative warm/idle,
 `warm_max` below `warm`, or duplicate pool; 401 bad api token; 409 egress
 pool on a node without an egress attachment.
 
+## POST /v1/drain
+
+Auth: root only. Cordons the node for maintenance: claim/fork/branch answer
+429 `node draining` (on a cluster the warm-peer redirect is tried first, and
+gossip stops naming this node within a tick as its warm counts hit zero),
+unclaimed warm VMs are destroyed, and live claims keep serving until release
+or TTL. Pool ownership is untouched — no pools.json write, no config change.
+Answers the fresh `GET /v1/info` payload; poll `claimed` to zero to know the
+node is empty. Deliberately not persisted: a restarted node serves again.
+
+## DELETE /v1/drain
+
+Auth: root only. Lifts the drain and kicks an immediate refill. Answers the
+fresh info payload.
+
 ## POST /v1/sandboxes/{id}/preview
 
 Auth: like fork — node `api_token` in the header, the sandbox's own token
@@ -270,6 +285,8 @@ peers:
 `hibernated` counts claims whose VM is currently hibernated, `archived` those
 checkpointed to the store with the local VM dropped (see
 [archive tiers](deploy.md#configuration)); both are included in `claimed`.
+A node cordoned via [`POST /v1/drain`](#post-v1drain) additionally reports
+`"draining": true`.
 
 `golden` reports whether the pool's snapshot exists (refill can clone);
 `warm` at `target` with `golden: true` means warm claims are served in

sandboxd/pool/claim.go

@@ -183,8 +183,12 @@ func (m *Manager) overQuota(extra int, tenant string) error {
 }
 
 // quotaErr answers ErrQuota when extra more claims for tenant would cross
-// the node-wide or per-tenant cap; callers hold m.mu.
+// the node-wide or per-tenant cap, or when the node is draining; callers
+// hold m.mu.
 func (m *Manager) quotaErr(extra int, tenant string) error {
+	if m.draining {
+		return fmt.Errorf("%w: node draining", ErrQuota)
+	}
 	if m.maxClaims > 0 && len(m.claimed)+extra > m.maxClaims {
 		return fmt.Errorf("%w: %d live claims, cap %d", ErrQuota, len(m.claimed), m.maxClaims)
 	}

sandboxd/pool/drain.go

@@ -0,0 +1,31 @@
+package pool
+
+import "context"
+
+// Drain cordons the node for maintenance: claim/fork/branch answer 429 like
+// a node at max_claims (a warm peer wins the redirect first on a cluster),
+// unclaimed warm VMs are destroyed, and live claims run to their leases.
+// Deliberately not persisted — a restarted node serves again.
+func (m *Manager) Drain(ctx context.Context) {
+	var trim []string
+	m.mu.Lock()
+	m.draining = true
+	for _, p := range m.pools {
+		for _, sb := range p.warm {
+			trim = append(trim, sb.VMName)
+		}
+		p.warm = p.warm[:0]
+	}
+	m.mu.Unlock()
+	m.runBounded(context.WithoutCancel(ctx), len(trim), func(ctx context.Context, i int) {
+		m.destroy(ctx, trim[i])
+	}).Wait()
+}
+
+// Uncordon lifts a drain and kicks an immediate refill.
+func (m *Manager) Uncordon(ctx context.Context) {
+	m.mu.Lock()
+	m.draining = false
+	m.mu.Unlock()
+	m.refillOnce(context.WithoutCancel(ctx))
+}

sandboxd/pool/drain_test.go

@@ -0,0 +1,55 @@
+package pool
+
+import (
+	"errors"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/cocoonstack/sandbox/sandboxd/config"
+)
+
+func TestDrainRefusesClaimsTrimsWarmAndUncordonRefills(t *testing.T) {
+	eng := newFakeEngine()
+	m := newTestManager(t, eng)
+	goldenDir := filepath.Join(m.goldensDir(), testKey.Hash())
+	if err := os.MkdirAll(goldenDir, 0o750); err != nil {
+		t.Fatalf("setup golden: %v", err)
+	}
+	if err := m.SetPools(t.Context(), []config.PoolSpec{{PoolKey: testKey, Warm: 2}}); err != nil {
+		t.Fatalf("SetPools: %v", err)
+	}
+	waitFor(t, func() bool {
+		infos, _ := m.Info()
+		return len(infos) == 1 && infos[0].Warm == 2
+	})
+
+	m.Drain(t.Context())
+
+	if _, err := m.ClaimWarm(t.Context(), testKey, 0, ""); !errors.Is(err, ErrQuota) {
+		t.Fatalf("ClaimWarm during drain: %v, want ErrQuota", err)
+	}
+	if _, err := m.ClaimProvision(t.Context(), testKey, 0, ""); !errors.Is(err, ErrQuota) {
+		t.Fatalf("ClaimProvision during drain: %v, want ErrQuota", err)
+	}
+	infos, g := m.Info()
+	if !g.Draining || infos[0].Warm != 0 {
+		t.Fatalf("draining=%v warm=%d, want true/0", g.Draining, infos[0].Warm)
+	}
+	if removed := eng.removedNames(); len(removed) != 2 {
+		t.Fatalf("removed=%v, want both warm VMs destroyed", removed)
+	}
+	m.refillOnce(t.Context())
+	if infos, _ = m.Info(); infos[0].Warm != 0 || infos[0].Refilling != 0 {
+		t.Fatalf("refill ran during drain: %+v", infos)
+	}
+
+	m.Uncordon(t.Context())
+	waitFor(t, func() bool {
+		infos, g := m.Info()
+		return !g.Draining && len(infos) == 1 && infos[0].Warm == 2
+	})
+	if _, err := claimAny(t.Context(), m, testKey, 0); err != nil {
+		t.Fatalf("claim after uncordon: %v", err)
+	}
+}

sandboxd/pool/pool.go

@@ -110,11 +110,12 @@ type PoolInfo struct {
 	Golden    bool          `json:"golden"`
 }
 
-// Gauges are the manager's point-in-time claim counts.
+// Gauges are the manager's point-in-time claim counts and drain state.
 type Gauges struct {
 	Claimed    int
 	Hibernated int
 	Archived   int
+	Draining   bool
 }
 
 type pool struct {
@@ -191,6 +192,7 @@ type Manager struct {
 	// stays O(1). usage is the always-on billing event stream, audit the
 	// config-gated request tap.
 	maxClaims    int
+	draining     bool // cordoned for maintenance; guarded by m.mu, deliberately not persisted
 	tenantMax    map[string]int
 	tenantLive   map[string]int
 	tenantEgress map[string]*egress.Policy // per-tenant allow-list; nil = no tenant policy
@@ -446,7 +448,7 @@ func (m *Manager) Info() ([]PoolInfo, Gauges) {
 		})
 	}
 	slices.SortFunc(pools, func(a, b PoolInfo) int { return strings.Compare(a.Key.Hash(), b.Key.Hash()) })
-	g := Gauges{Claimed: len(m.claimed)}
+	g := Gauges{Claimed: len(m.claimed), Draining: m.draining}
 	for _, sb := range m.claimed {
 		if sb.HibernateSnap != "" {
 			g.Hibernated++

sandboxd/pool/refill.go

@@ -19,6 +19,9 @@ import (
 func (m *Manager) refillOnce(ctx context.Context) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
+	if m.draining {
+		return
+	}
 	now := time.Now()
 	for _, p := range m.pools {
 		target := p.effectiveTarget(now)
@@ -55,12 +58,12 @@ func (m *Manager) refillOne(ctx context.Context, p *pool, golden string) {
 	m.mu.Lock()
 	p.refilling--
 	target := p.effectiveTarget(time.Now())
-	if err == nil && len(p.warm) < target {
+	if err == nil && !m.draining && len(p.warm) < target {
 		p.warm = append(p.warm, sb)
 		p.noteLead(time.Since(start))
 		keep = true
 	}
-	if err == nil && p.goldenDir != "" && len(p.warm)+p.refilling < target {
+	if err == nil && !m.draining && p.goldenDir != "" && len(p.warm)+p.refilling < target {
 		p.refilling++
 		releaseSlot = false
 		go m.refillOne(ctx, p, p.goldenDir)

sandboxd/server/metrics.go

@@ -31,6 +31,12 @@ func (s *Server) handleMetrics(w http.ResponseWriter, _ *http.Request) {
 	_, _ = fmt.Fprintf(w, "sandboxd_hibernated %d\n", g.Hibernated)
 	metric("archived", "gauge", "claims archived to the checkpoint store")
 	_, _ = fmt.Fprintf(w, "sandboxd_archived %d\n", g.Archived)
+	draining := 0
+	if g.Draining {
+		draining = 1
+	}
+	metric("draining", "gauge", "1 while the node is cordoned for maintenance")
+	_, _ = fmt.Fprintf(w, "sandboxd_draining %d\n", draining)
 
 	if tenants := s.mgr.TenantClaims(); len(tenants) > 0 {
 		metric("tenant_claims", "gauge", "live claims per configured tenant")

sandboxd/server/server.go

@@ -76,6 +76,8 @@ type Manager interface {
 	AgentSocket(id, token string) (string, error)
 	WakeAgentSocket(ctx context.Context, id, token string) (string, error)
 	SetPools(ctx context.Context, pools []config.PoolSpec) error
+	Drain(ctx context.Context)
+	Uncordon(ctx context.Context)
 	Info() ([]pool.PoolInfo, pool.Gauges)
 }
 
@@ -100,6 +102,7 @@ type InfoResponse struct {
 	Claimed    int             `json:"claimed"`
 	Hibernated int             `json:"hibernated"`
 	Archived   int             `json:"archived"`
+	Draining   bool            `json:"draining,omitempty"`
 	Peers      []string        `json:"peers,omitempty"`
 }
 
@@ -165,6 +168,8 @@ func (s *Server) Handler() http.Handler {
 	mux.HandleFunc("DELETE /v1/checkpoints/{id}", s.requireToken(s.handleDeleteCheckpoint))
 	mux.HandleFunc("DELETE /v1/templates", s.requireToken(s.handleDeleteTemplate))
 	mux.HandleFunc("PUT /v1/pools", s.requireRoot(s.handlePutPools))
+	mux.HandleFunc("POST /v1/drain", s.requireRoot(s.handleDrain))
+	mux.HandleFunc("DELETE /v1/drain", s.requireRoot(s.handleUncordon))
 	mux.HandleFunc("GET /v1/sandboxes/{id}/agent", s.handleAgent)
 	mux.HandleFunc("GET /v1/sandboxes/{id}/owner", s.handleOwner)
 	mux.HandleFunc("GET /v1/info", s.requireRoot(s.handleInfo))
@@ -411,6 +416,16 @@ func (s *Server) handlePutPools(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+func (s *Server) handleDrain(w http.ResponseWriter, r *http.Request) {
+	s.mgr.Drain(r.Context())
+	s.handleInfo(w, r)
+}
+
+func (s *Server) handleUncordon(w http.ResponseWriter, r *http.Request) {
+	s.mgr.Uncordon(r.Context())
+	s.handleInfo(w, r)
+}
+
 // handlePeers lists the cluster's node addresses for the SDK's redirect
 // follow and Lookup scatter — cluster topology, not operator state, so any
 // valid token (root or tenant) may read it.
@@ -424,7 +439,7 @@ func (s *Server) handlePeers(w http.ResponseWriter, _ *http.Request) {
 
 func (s *Server) handleInfo(w http.ResponseWriter, _ *http.Request) {
 	pools, g := s.mgr.Info()
-	resp := InfoResponse{Pools: pools, Claimed: g.Claimed, Hibernated: g.Hibernated, Archived: g.Archived}
+	resp := InfoResponse{Pools: pools, Claimed: g.Claimed, Hibernated: g.Hibernated, Archived: g.Archived, Draining: g.Draining}
 	if s.placer != nil {
 		resp.Peers = s.placer.PeerAddrs()
 	}

sandboxd/server/server_test.go

@@ -239,6 +239,36 @@ func TestPutPoolsRejectsUnknownFields(t *testing.T) {
 	}
 }
 
+func TestDrainEndpoints(t *testing.T) {
+	mgr := &fakeManager{}
+	ts := newTestServer(t, "sekret", mgr, nil)
+	for _, tt := range []struct {
+		method   string
+		draining bool
+	}{
+		{http.MethodPost, true},
+		{http.MethodDelete, false},
+	} {
+		req, err := http.NewRequestWithContext(t.Context(), tt.method, ts.URL+"/v1/drain", nil)
+		if err != nil {
+			t.Fatalf("request: %v", err)
+		}
+		req.Header.Set("Authorization", "Bearer sekret")
+		resp, err := http.DefaultClient.Do(req)
+		if err != nil {
+			t.Fatalf("do: %v", err)
+		}
+		var info InfoResponse
+		if err := json.NewDecoder(resp.Body).Decode(&info); err != nil {
+			t.Fatalf("decode: %v", err)
+		}
+		resp.Body.Close()
+		if resp.StatusCode != http.StatusOK || info.Draining != tt.draining || mgr.draining != tt.draining {
+			t.Fatalf("%s: status=%d draining=%v mgr=%v, want 200/%v", tt.method, resp.StatusCode, info.Draining, mgr.draining, tt.draining)
+		}
+	}
+}
+
 func TestPutPoolsUpdatesTargets(t *testing.T) {
 	var got []config.PoolSpec
 	mgr := &fakeManager{
@@ -836,6 +866,7 @@ type fakeManager struct {
 
 	gotTenant    string
 	tenantClaims map[string]int
+	draining     bool
 }
 
 func (f *fakeManager) ClaimWarm(_ context.Context, _ types.PoolKey, _ time.Duration, tenant string) (*types.Sandbox, error) {
@@ -967,9 +998,13 @@ func (f *fakeManager) SetPools(_ context.Context, pools []config.PoolSpec) error
 }
 
 func (f *fakeManager) Info() ([]pool.PoolInfo, pool.Gauges) {
-	return f.infoPools, pool.Gauges{}
+	return f.infoPools, pool.Gauges{Draining: f.draining}
 }
 
+func (f *fakeManager) Drain(context.Context) { f.draining = true }
+
+func (f *fakeManager) Uncordon(context.Context) { f.draining = false }
+
 type fakeDialer struct {
 	dial func(ctx context.Context, sock string) (net.Conn, error)
 }

sdk/go/drain.go

@@ -0,0 +1,26 @@
+package sandbox
+
+import (
+	"context"
+	"net/http"
+)
+
+// Drain cordons the entry node for maintenance (root token): new claims,
+// forks, and branches are refused while live claims run to their leases.
+// Answers the fresh node info — poll Info until Claimed reaches zero.
+func (c *Client) Drain(ctx context.Context) (*NodeInfo, error) {
+	info, err := doJSON[NodeInfo](ctx, c, http.MethodPost, c.addr, "/v1/drain", nil, c.apiToken, "drain")
+	if err != nil {
+		return nil, err
+	}
+	return &info, nil
+}
+
+// Uncordon lifts a drain on the entry node (root token).
+func (c *Client) Uncordon(ctx context.Context) (*NodeInfo, error) {
+	info, err := doJSON[NodeInfo](ctx, c, http.MethodDelete, c.addr, "/v1/drain", nil, c.apiToken, "uncordon")
+	if err != nil {
+		return nil, err
+	}
+	return &info, nil
+}

sdk/go/drain_test.go

@@ -0,0 +1,42 @@
+package sandbox
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+)
+
+func TestDrainAndUncordon(t *testing.T) {
+	var methods []string
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/v1/drain" {
+			t.Errorf("path = %s, want /v1/drain", r.URL.Path)
+		}
+		if auth := r.Header.Get("Authorization"); auth != "Bearer root" {
+			t.Errorf("auth = %q, want Bearer root", auth)
+		}
+		methods = append(methods, r.Method)
+		_ = json.NewEncoder(w).Encode(NodeInfo{Claimed: 3, Draining: r.Method == http.MethodPost})
+	}))
+	defer ts.Close()
+
+	c := testClient(t, ts, WithAPIToken("root"))
+	info, err := c.Drain(t.Context())
+	if err != nil {
+		t.Fatalf("Drain: %v", err)
+	}
+	if !info.Draining || info.Claimed != 3 {
+		t.Errorf("info = %+v, want Draining true, Claimed 3", info)
+	}
+	info, err = c.Uncordon(t.Context())
+	if err != nil {
+		t.Fatalf("Uncordon: %v", err)
+	}
+	if info.Draining {
+		t.Errorf("info = %+v, want Draining false", info)
+	}
+	if len(methods) != 2 || methods[0] != http.MethodPost || methods[1] != http.MethodDelete {
+		t.Errorf("methods = %v, want [POST DELETE]", methods)
+	}
+}