feat(vpcd): drain held DHCP leases on teardown (#346)

DHCP-source external-pool leases are DHCPRELEASEd only on explicit
instance/EIP terminate. Manager.Stop() preserves leases for
adopt-on-reboot, so env teardown/reset kills vpcd and destroys the
per-AZ lease KV without releasing — stranding every live lease
(instance ENIs, EIP allocations, per-VPC LRP gateways) on the upstream
DHCP server until TTL. Across CI churn with fresh client-ids the
upstream scope fills and new launches get no public IP.

Add Manager.DrainAll + a vpc.dhcp.drain RPC and a standalone
'spx admin cluster drain-dhcp' command that releases every held lease.
Wired into both teardown paths (single-node teardown role and
multi-node cluster-teardown.yml) before services stop. Stop() is
unchanged, so normal restarts still adopt leases on reboot.

cmd/spinifex/cmd/admin.go

@@ -115,6 +115,16 @@ Each phase waits for all nodes to ACK before proceeding to the next.`,
 	Run: runClusterShutdown,
 }
 
+var clusterDrainDHCPCmd = &cobra.Command{
+	Use:   "drain-dhcp",
+	Short: "Release all upstream DHCP leases held by vpcd",
+	Long: `Ask each vpcd to DHCPRELEASE every external-pool DHCP lease it currently
+holds, returning them to the upstream DHCP server. Run this on teardown before
+stopping services: an env reset otherwise strands held leases on the upstream
+server until their TTL expires, eventually exhausting the upstream scope.`,
+	Run: runClusterDrainDHCP,
+}
+
 var adminInitCmd = &cobra.Command{
 	Use:   "init",
 	Short: "Initialize Spinifex platform configuration",
@@ -245,6 +255,9 @@ func init() {
 	clusterShutdownCmd.Flags().Duration("timeout", 120*time.Second, "Maximum time to wait per phase")
 	clusterShutdownCmd.Flags().Bool("dry-run", false, "Print phase plan without executing")
 
+	clusterCmd.AddCommand(clusterDrainDHCPCmd)
+	clusterDrainDHCPCmd.Flags().Duration("timeout", 30*time.Second, "Reply-collection window for vpcd drain responders")
+
 	adminCmd.AddCommand(imagesCmd)
 	imagesCmd.AddCommand(imagesImportCmd)
 	imagesCmd.AddCommand(imagesListCmd)

cmd/spinifex/cmd/cluster.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/mulgadc/spinifex/spinifex/daemon"
+	"github.com/mulgadc/spinifex/spinifex/network/external/dhcp"
 	"github.com/nats-io/nats.go"
 	"github.com/spf13/cobra"
 )
@@ -160,6 +161,79 @@ func runClusterShutdown(cmd *cobra.Command, args []string) {
 	fmt.Printf("Cluster shutdown complete (%s)\n", time.Since(start).Round(time.Millisecond))
 }
 
+// runClusterDrainDHCP asks every vpcd to DHCPRELEASE all external-pool leases
+// it currently holds, returning them to the upstream DHCP server. Intended for
+// the teardown path: an env reset otherwise strands held leases until TTL
+// because vpcd's Stop() preserves them for adopt-on-reboot. Best-effort — if
+// the cluster is already partly down it warns and exits 0 so teardown proceeds.
+func runClusterDrainDHCP(cmd *cobra.Command, args []string) {
+	timeout, _ := cmd.Flags().GetDuration("timeout")
+	if timeout <= 0 {
+		timeout = 30 * time.Second
+	}
+
+	_, nc, err := loadConfigAndConnect()
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Warning: drain-dhcp could not connect (cluster may be down): %v\n", err)
+		return
+	}
+	defer nc.Close()
+
+	total, responders := drainDHCPLeases(nc, timeout)
+	fmt.Printf("DHCP drain complete: released %d lease(s) from %d vpcd responder(s)\n", total, responders)
+}
+
+// drainDHCPLeases publishes a drain request to vpc.dhcp.drain and collects
+// replies until timeout — one responder per AZ (vpcd uses a per-AZ queue
+// group). Returns the summed released count and the number of responders.
+func drainDHCPLeases(nc *nats.Conn, timeout time.Duration) (released, responders int) {
+	reqData, err := json.Marshal(struct{}{})
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Error: %v\n", err)
+		return 0, 0
+	}
+
+	inbox := nats.NewInbox()
+	sub, err := nc.SubscribeSync(inbox)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Warning: drain-dhcp subscribe failed: %v\n", err)
+		return 0, 0
+	}
+	defer sub.Unsubscribe()
+
+	if err := nc.PublishRequest(dhcp.TopicDrain, inbox, reqData); err != nil {
+		fmt.Fprintf(os.Stderr, "Warning: drain-dhcp publish failed: %v\n", err)
+		return 0, 0
+	}
+	nc.Flush()
+
+	deadline := time.Now().Add(timeout)
+	for {
+		remaining := time.Until(deadline)
+		if remaining <= 0 {
+			break
+		}
+		msg, err := sub.NextMsg(remaining)
+		if err != nil {
+			break
+		}
+		var reply struct {
+			Released int    `json:"released"`
+			Error    string `json:"error,omitempty"`
+		}
+		if err := json.Unmarshal(msg.Data, &reply); err != nil {
+			continue
+		}
+		responders++
+		if reply.Error != "" {
+			fmt.Fprintf(os.Stderr, "  vpcd drain error: %s\n", reply.Error)
+			continue
+		}
+		released += reply.Released
+	}
+	return released, responders
+}
+
 // collectShutdownACKs publishes a shutdown request and collects ACKs from nodes.
 func collectShutdownACKs(nc *nats.Conn, topic string, reqData []byte, nodeCount int, timeout time.Duration) ([]daemon.ShutdownACK, error) {
 	inbox := nats.NewInbox()

cmd/spinifex/cmd/cluster_test.go

@@ -0,0 +1,58 @@
+package cmd
+
+import (
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/network/external/dhcp"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// fakeDrainResponder subscribes a stand-in vpcd to vpc.dhcp.drain that
+// replies with the given released count (or error), mirroring
+// Manager.handleDrainMsg without a real lease store.
+func fakeDrainResponder(t *testing.T, nc *nats.Conn, released int, errMsg string) {
+	t.Helper()
+	sub, err := nc.Subscribe(dhcp.TopicDrain, func(msg *nats.Msg) {
+		body, _ := json.Marshal(struct {
+			Released int    `json:"released"`
+			Error    string `json:"error,omitempty"`
+		}{Released: released, Error: errMsg})
+		_ = msg.Respond(body)
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+}
+
+func TestDrainDHCPLeasesSumsReleased(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	fakeDrainResponder(t, nc, 7, "")
+
+	released, responders := drainDHCPLeases(nc, 2*time.Second)
+	assert.Equal(t, 7, released)
+	assert.Equal(t, 1, responders)
+}
+
+func TestDrainDHCPLeasesReportsResponderError(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	fakeDrainResponder(t, nc, 0, "list leases for drain: boom")
+
+	// An erroring responder still counts as a responder but contributes
+	// zero released leases.
+	released, responders := drainDHCPLeases(nc, 2*time.Second)
+	assert.Equal(t, 0, released)
+	assert.Equal(t, 1, responders)
+}
+
+func TestDrainDHCPLeasesNoResponders(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+
+	// No vpcd subscribed: the collection window elapses with zero replies.
+	released, responders := drainDHCPLeases(nc, 200*time.Millisecond)
+	assert.Equal(t, 0, released)
+	assert.Equal(t, 0, responders)
+}

docs/COMMANDS.md

@@ -65,6 +65,7 @@ Operational commands for inspecting cluster state. These fan out NATS requests t
 | Command | Flags | Description |
 |---------|-------|-------------|
 | `spx admin cluster shutdown` | `--force` (shutdown even if nodes don't respond), `--timeout` (max wait per phase, default 120s), `--dry-run` (print phase plan without executing) | Performs coordinated, phased shutdown of entire cluster. Phases execute in order: GATE (stop API/UI) → DRAIN (stop VMs) → STORAGE (stop viperblock) → PERSIST (stop predastore) → INFRA (stop NATS/daemon). Each phase waits for all nodes to ACK before proceeding. Uses JetStream state tracking. |
+| `spx admin cluster drain-dhcp` | `--timeout` (reply-collection window, default 30s) | Asks each vpcd to DHCPRELEASE every external-pool DHCP lease it currently holds, returning them to the upstream DHCP server. Run on teardown before stopping services — an env reset otherwise strands held leases upstream until their TTL expires, eventually exhausting the upstream scope. Best-effort: warns and exits 0 if the cluster is already down. |
 
 ### Certificate Management
 

spinifex/network/external/dhcp/manager.go

@@ -186,6 +186,7 @@ func (m *Manager) Subscribe(nc *nats.Conn) ([]*nats.Subscription, error) {
 	subs := []sub{
 		{TopicAcquire, m.handleAcquireMsg},
 		{TopicRelease, m.handleReleaseMsg},
+		{TopicDrain, m.handleDrainMsg},
 	}
 	var out []*nats.Subscription
 	for _, s := range subs {
@@ -486,6 +487,51 @@ func (m *Manager) acquireLocked(ctx context.Context, req acquireWireRequest) (*E
 	return &entry, nil
 }
 
+// DrainAll DHCPRELEASEs every lease currently held in this vpcd's per-AZ
+// store and deletes the KV records. Best-effort: a per-lease release
+// failure is logged and skipped so one stuck lease can't block the rest.
+// Returns the number of leases successfully released. Used by the teardown
+// path so an env reset returns its leases to the upstream pool rather than
+// stranding them until TTL (Stop() deliberately preserves leases for
+// adopt-on-reboot, which is wrong on a destructive teardown).
+func (m *Manager) DrainAll(ctx context.Context) (int, error) {
+	entries, err := m.store.List()
+	if err != nil {
+		return 0, fmt.Errorf("list leases for drain: %w", err)
+	}
+	released := 0
+	for _, e := range entries {
+		if e.Lease == nil {
+			continue
+		}
+		if relErr := m.handleRelease(ctx, e.Lease.ClientID); relErr != nil {
+			slog.Warn("dhcp manager: drain release failed", "client_id", e.Lease.ClientID, "err", relErr)
+			continue
+		}
+		released++
+	}
+	slog.Info("dhcp manager: drained leases", "released", released, "total", len(entries))
+	return released, nil
+}
+
+func (m *Manager) handleDrainMsg(msg *nats.Msg) {
+	if msg.Reply == "" {
+		slog.Warn("dhcp manager: drain request missing reply subject; dropping")
+		return
+	}
+	n, err := m.DrainAll(context.Background())
+	reply := drainWireReply{Released: n}
+	if err != nil {
+		reply.Error = err.Error()
+	}
+	body, mErr := json.Marshal(reply)
+	if mErr != nil {
+		slog.Warn("dhcp manager: encode drain reply failed", "err", mErr)
+		return
+	}
+	_ = msg.Respond(body)
+}
+
 func (m *Manager) handleRelease(ctx context.Context, clientID string) error {
 	if clientID == "" {
 		return errors.New("client_id required")

spinifex/network/external/dhcp/manager_test.go

@@ -2,6 +2,7 @@ package dhcp_test
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"net"
@@ -124,6 +125,123 @@ func TestNATSClientAcquireReleaseRoundtrip(t *testing.T) {
 	assert.Equal(t, 1, fake.ReleaseCount())
 }
 
+func seedLeases(t *testing.T, store *dhcp.Store, ids []string) {
+	t.Helper()
+	hw, _ := net.ParseMAC("02:00:00:aa:bb:cc")
+	for i, id := range ids {
+		lease := &dhcp.Lease{
+			Bridge:        "br-wan",
+			ClientID:      id,
+			HWAddr:        hw,
+			IP:            net.IPv4(192, 0, 2, byte(10+i)),
+			AcquiredAt:    time.Now(),
+			LeaseDuration: time.Hour,
+		}
+		require.NoError(t, store.Put(dhcp.Entry{Purpose: "eip", PoolName: "default", Lease: lease}))
+	}
+}
+
+func TestManagerDrainAll(t *testing.T) {
+	fake := dhcp.NewFake()
+	mgr, store, _ := newTestManager(t, "az1", fake, time.Now)
+
+	// Seed before Start so the manager adopts them and spawns a renewal
+	// loop per lease — drain must release every lease AND cancel its loop.
+	ids := []string{"eipalloc-1", "eni-2", "i-3", "gw-vpc-4"}
+	seedLeases(t, store, ids)
+	require.NoError(t, mgr.Start(context.Background()))
+	require.Eventually(t, func() bool { return mgr.LoopCount() == len(ids) }, time.Second, 10*time.Millisecond)
+
+	n, err := mgr.DrainAll(context.Background())
+	require.NoError(t, err)
+	assert.Equal(t, len(ids), n)
+	assert.Equal(t, len(ids), fake.ReleaseCount())
+
+	entries, err := store.List()
+	require.NoError(t, err)
+	assert.Empty(t, entries)
+	assert.Equal(t, 0, mgr.LoopCount())
+}
+
+func TestManagerDrainAllBestEffortOnReleaseError(t *testing.T) {
+	fake := dhcp.NewFake()
+	// A stuck upstream release on one lease must not block draining the rest.
+	fake.ReleaseHook = func(l *dhcp.Lease) error {
+		if l.ClientID == "eni-2" {
+			return errors.New("upstream unreachable")
+		}
+		return nil
+	}
+	mgr, store, _ := newTestManager(t, "az1", fake, time.Now)
+
+	ids := []string{"eipalloc-1", "eni-2", "i-3"}
+	seedLeases(t, store, ids)
+
+	n, err := mgr.DrainAll(context.Background())
+	require.NoError(t, err)
+	assert.Equal(t, len(ids), n)
+	assert.Equal(t, len(ids), fake.ReleaseCount())
+
+	entries, err := store.List()
+	require.NoError(t, err)
+	assert.Empty(t, entries)
+}
+
+func TestManagerDrainMsgWithoutReplyIsDropped(t *testing.T) {
+	fake := dhcp.NewFake()
+	mgr, store, nc := newTestManager(t, "az1", fake, time.Now)
+	require.NoError(t, mgr.Start(context.Background()))
+	subs, err := mgr.Subscribe(nc)
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		for _, s := range subs {
+			_ = s.Unsubscribe()
+		}
+	})
+
+	seedLeases(t, store, []string{"eipalloc-x"})
+
+	// A drain publish with no reply subject must be dropped before draining —
+	// the lease stays put and nothing is released.
+	require.NoError(t, nc.Publish(dhcp.TopicDrain, []byte("{}")))
+	require.NoError(t, nc.Flush())
+	time.Sleep(100 * time.Millisecond)
+
+	entries, err := store.List()
+	require.NoError(t, err)
+	assert.Len(t, entries, 1)
+	assert.Equal(t, 0, fake.ReleaseCount())
+}
+
+func TestManagerDrainRPCRoundtrip(t *testing.T) {
+	fake := dhcp.NewFake()
+	mgr, store, nc := newTestManager(t, "az1", fake, time.Now)
+	require.NoError(t, mgr.Start(context.Background()))
+	subs, err := mgr.Subscribe(nc)
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		for _, s := range subs {
+			_ = s.Unsubscribe()
+		}
+	})
+
+	seedLeases(t, store, []string{"eipalloc-A", "eipalloc-B"})
+
+	msg, err := nc.Request(dhcp.TopicDrain, []byte("{}"), 2*time.Second)
+	require.NoError(t, err)
+	var reply struct {
+		Released int    `json:"released"`
+		Error    string `json:"error,omitempty"`
+	}
+	require.NoError(t, json.Unmarshal(msg.Data, &reply))
+	assert.Empty(t, reply.Error)
+	assert.Equal(t, 2, reply.Released)
+
+	entries, err := store.List()
+	require.NoError(t, err)
+	assert.Empty(t, entries)
+}
+
 func TestNATSClientAcquireIdempotent(t *testing.T) {
 	fake := dhcp.NewFake()
 	mgr, _, nc := newTestManager(t, "az1", fake, time.Now)

spinifex/network/external/dhcp/wire.go

@@ -14,6 +14,10 @@ import (
 const (
 	TopicAcquire = "vpc.dhcp.acquire"
 	TopicRelease = "vpc.dhcp.release"
+	// TopicDrain asks the bridge-owning vpcd to DHCPRELEASE every lease it
+	// currently holds. Invoked on cluster teardown so an env reset returns
+	// its leases to the upstream pool instead of stranding them until TTL.
+	TopicDrain = "vpc.dhcp.drain"
 )
 
 // acquireWireRequest is the JSON payload sent by daemon-side
@@ -51,6 +55,14 @@ type releaseWireReply struct {
 	Error string `json:"error,omitempty"`
 }
 
+// drainWireReply reports how many leases the responding vpcd released. The
+// drain request itself carries no body — the manager drains its entire
+// per-AZ lease store, so there are no parameters.
+type drainWireReply struct {
+	Released int    `json:"released"`
+	Error    string `json:"error,omitempty"`
+}
+
 // wireLease is Lease in JSON-friendly form: dotted-quad IPs, MAC string,
 // durations as nanoseconds.
 type wireLease struct {