fix(health): probe predastore raft state instead of echoing the config (#786)

/health reported service_health.predastore == "ok" for every node simply because predastore was named in spinifex.toml, without contacting it — on env19 all three nodes stayed green for the whole time raft had no leader and volume creates were failing. A probe registry now backs /health's predastore entry with a real raft status dial (ok/no_leader/unreachable, cached briefly and bounded by a short timeout), while every other service keeps its unprobed "ok" default unchanged. spx admin storage status had the same root problem from the other side: it dialed HTTPS /health and /status endpoints predastore's meta nodes never served (they speak an opcode-multiplexed RPC), so every meta node always reported unhealthy with blank raft fields. It now dials the same raft status RPC predastore#82 exposes, verified against the cluster CA, with the JSON output shape unchanged.
2026-10-15 09:01:31 +00:00 · 2026-08-13 15:24:54 +10:00
parent f7a2d85691
commit 560b437f3b
12 changed files with 532 additions and 134 deletions
@@ -23,7 +23,7 @@ require (
 	github.com/miekg/dns v1.1.72
 	github.com/mulgadc/bluebottle v1.16.1-0.20260812012137-683e88e465e3
 	github.com/mulgadc/northstar v1.16.1-0.20260812012604-5d7a622ad059
-	github.com/mulgadc/predastore v1.16.1-0.20260812012531-12c90bd2e14a
+	github.com/mulgadc/predastore v1.16.1-0.20260813022738-0b4392b43cae
 	github.com/mulgadc/viperblock v1.16.1-0.20260812124023-7e0df2323c75
 	github.com/nats-io/nats-server/v2 v2.14.4
 	github.com/nats-io/nats.go v1.52.0
@@ -1323,12 +1323,8 @@ github.com/mulgadc/bluebottle v1.16.1-0.20260812012137-683e88e465e3 h1:EJ8JMQB9Y
 github.com/mulgadc/bluebottle v1.16.1-0.20260812012137-683e88e465e3/go.mod h1:bhJitsX6VKkbmQ1dVh3vPUmoSeQsvtn5Jm4D8hUyq2s=
 github.com/mulgadc/northstar v1.16.1-0.20260812012604-5d7a622ad059 h1:CK3Ed0Rg5dGvpSeMFTGpE1nf+390Rbf5hePOcnm1+kM=
 github.com/mulgadc/northstar v1.16.1-0.20260812012604-5d7a622ad059/go.mod h1:vyfhxWlBlAnJfUeSEt+z/Z6KrfReC/uzZM3WTllorYM=
-github.com/mulgadc/predastore v1.16.1-0.20260812012531-12c90bd2e14a h1:XSvL4FucGZaEDPUXCG/mUshfFskGO9tE69SYMJJ28L4=
-github.com/mulgadc/predastore v1.16.1-0.20260812012531-12c90bd2e14a/go.mod h1:S5vcni7R3VYi3nccelMagKbWdX58WtbKf0I1R/7MmNI=
-github.com/mulgadc/viperblock v1.16.1-0.20260812012547-bb1225320f0c h1:5PCU4696Gelc+4X4K+E+EzNVsyoLLWEiF/TcsIaAGi4=
-github.com/mulgadc/viperblock v1.16.1-0.20260812012547-bb1225320f0c/go.mod h1:8NT5u+udY1CsoKz/JAPY2SSMcQ98Omi+42r8BkoA4FY=
-github.com/mulgadc/viperblock v1.16.1-0.20260812033921-fc825c93eff6 h1:QfdcaqnrDDzsO4neJNX7BdL8vzyga/WPn4A/04QZcVw=
-github.com/mulgadc/viperblock v1.16.1-0.20260812033921-fc825c93eff6/go.mod h1:8NT5u+udY1CsoKz/JAPY2SSMcQ98Omi+42r8BkoA4FY=
+github.com/mulgadc/predastore v1.16.1-0.20260813022738-0b4392b43cae h1:itP5arN88Xv/xe4kI/5TWC/fkawnOeIR8MF1AL7aC48=
+github.com/mulgadc/predastore v1.16.1-0.20260813022738-0b4392b43cae/go.mod h1:S5vcni7R3VYi3nccelMagKbWdX58WtbKf0I1R/7MmNI=
 github.com/mulgadc/viperblock v1.16.1-0.20260812124023-7e0df2323c75 h1:eqB0VOuFPldBHuQ7ELYZHSRxq+3MXbMMi9D5bfVUwe8=
 github.com/mulgadc/viperblock v1.16.1-0.20260812124023-7e0df2323c75/go.mod h1:8NT5u+udY1CsoKz/JAPY2SSMcQ98Omi+42r8BkoA4FY=
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
@@ -184,6 +184,9 @@ type Daemon struct {
 	startTime     time.Time
 	configPath    string

+	// predastoreHealth caches the /health predastore probe's verdict; see health.go.
+	predastoreHealth predastoreHealthCache
+
 	// System credentials for ALB agent SigV4 auth (loaded from system-credentials.json)
 	systemAccessKey string
 	systemSecretKey string
@@ -2307,10 +2310,7 @@ func (d *Daemon) ClusterManager() error {
 			return
 		}

-		serviceHealth := make(map[string]string)
-		for _, svc := range d.config.GetServices() {
-			serviceHealth[svc] = "ok"
-		}
+		serviceHealth := d.probeServiceHealth(r.Context())
 		if !d.config.HasService("nats") {
 			if d.natsConn != nil && d.natsConn.IsConnected() {
 				serviceHealth["nats"] = "remote_ok"
@@ -3,7 +3,6 @@ package daemon
 import (
 	"log/slog"
 	"os"
-	"path/filepath"

 	"github.com/mulgadc/spinifex/spinifex/types"
 	"github.com/nats-io/nats.go"
@@ -55,8 +54,7 @@ type predastoreBucket struct {
 // handleStorageConfig responds with parsed predastore config (topology only, no creds).
 // Used by the gateway to build the GetStorageStatus response.
 func (d *Daemon) handleStorageConfig(msg *nats.Msg) {
-	configDir := filepath.Dir(d.configPath)
-	predastorePath := filepath.Join(configDir, "predastore", "predastore.toml")
+	predastorePath := predastoreConfigPath(d.configPath)

 	data, err := os.ReadFile(predastorePath)
 	if err != nil {
@@ -0,0 +1,182 @@
+package daemon
+
+import (
+	"context"
+	"crypto/x509"
+	"fmt"
+	"log/slog"
+	"path/filepath"
+	"sync"
+	"time"
+
+	pds "github.com/mulgadc/predastore"
+	"github.com/mulgadc/spinifex/spinifex/utils"
+)
+
+// Predastore health verdicts, as reported under service_health.predastore in
+// the /health response.
+const (
+	predastoreHealthOK          = "ok"
+	predastoreHealthNoLeader    = "no_leader"
+	predastoreHealthUnreachable = "unreachable"
+)
+
+// predastoreProbeTimeout bounds every dial a /health request makes to this
+// host's own meta node(s), so a wedged or unreachable peer can never stall
+// the handler. A var, not a const, so tests can shrink it to exercise the
+// timeout path without a multi-second sleep.
+var predastoreProbeTimeout = 2 * time.Second
+
+// predastoreHealthCacheTTL bounds how stale a cached predastore verdict may
+// be. /health is a monitoring target polled far more often than raft state
+// changes, and a fresh QUIC dial per poll would put that polling load on the
+// raft plane; a few seconds of staleness is a good trade for that. A var for
+// the same reason as predastoreProbeTimeout.
+var predastoreHealthCacheTTL = 5 * time.Second
+
+// nodeStatusFn is pds.NodeStatus, indirected so tests can stub the raft dial
+// and exercise every verdict — including the timeout path — without a live
+// predastore process.
+var nodeStatusFn = pds.NodeStatus
+
+// probeFunc reports one service's real health, in place of the "ok" every
+// service otherwise reports simply for being named in spinifex.toml.
+type probeFunc func(ctx context.Context, d *Daemon) string
+
+// serviceProbes maps a service name to the probe that reports its real
+// health. A service absent from this map keeps reporting "ok" as soon as it
+// is named in spinifex.toml: service_health.<svc> == "ok" gates several
+// ansible bootstrap playbooks, and changing that default in the same change
+// that adds the first probe would stall them for a reason unrelated to any
+// bug in those services. Probes for the other services are a separate piece
+// of work.
+var serviceProbes = map[string]probeFunc{
+	"predastore": probePredastore,
+}
+
+// probeServiceHealth reports every service named in spinifex.toml: a probed
+// service returns its probe's verdict, and everything else returns "ok" —
+// the same restatement of the config /health has always reported for
+// services with no probe.
+func (d *Daemon) probeServiceHealth(ctx context.Context) map[string]string {
+	services := d.config.GetServices()
+	health := make(map[string]string, len(services))
+	for _, svc := range services {
+		if probe, ok := serviceProbes[svc]; ok {
+			health[svc] = probe(ctx, d)
+			continue
+		}
+		health[svc] = "ok"
+	}
+	return health
+}
+
+// predastoreHealthCache holds the last predastore verdict this node computed,
+// so concurrent or rapid /health polls do not each dial the local meta
+// node(s) themselves.
+type predastoreHealthCache struct {
+	mu     sync.Mutex
+	at     time.Time
+	result string
+}
+
+// probePredastore reports the raft health of the meta node(s) MetaNodesOnHost
+// pins to this host, serving a cached verdict when it is fresh enough.
+func probePredastore(ctx context.Context, d *Daemon) string {
+	d.predastoreHealth.mu.Lock()
+	if !d.predastoreHealth.at.IsZero() && time.Since(d.predastoreHealth.at) < predastoreHealthCacheTTL {
+		result := d.predastoreHealth.result
+		d.predastoreHealth.mu.Unlock()
+		return result
+	}
+	d.predastoreHealth.mu.Unlock()
+
+	result := computePredastoreHealth(ctx, d)
+
+	d.predastoreHealth.mu.Lock()
+	d.predastoreHealth.result = result
+	d.predastoreHealth.at = time.Now()
+	d.predastoreHealth.mu.Unlock()
+
+	return result
+}
+
+// computePredastoreHealth dials every meta node MetaNodesOnHost pins to this
+// host and reports the worst of them: unreachable beats no_leader beats ok,
+// since any one of them failing is this host's predastore service failing.
+// A host that runs no meta node at all — a pure gate/blob host, or a node
+// that runs no predastore — has nothing local to probe and reports ok: that
+// is not a failure of this host's own service, and the raft quorum as a
+// whole is what `spx admin storage status` reports on, across every host
+// that does run a meta node.
+func computePredastoreHealth(ctx context.Context, d *Daemon) string {
+	nodes, cfg, err := localMetaNodes(d)
+	if err != nil {
+		slog.Warn("predastore health probe: could not resolve local meta nodes", "err", err)
+		return predastoreHealthUnreachable
+	}
+	if len(nodes) == 0 {
+		return predastoreHealthOK
+	}
+
+	rootCAs, err := loadPredastoreTrustRoot(d)
+	if err != nil {
+		slog.Warn("predastore health probe: could not load cluster CA", "err", err)
+		return predastoreHealthUnreachable
+	}
+
+	probeCtx, cancel := context.WithTimeout(ctx, predastoreProbeTimeout)
+	defer cancel()
+
+	verdict := predastoreHealthOK
+	for _, node := range nodes {
+		status, err := nodeStatusFn(probeCtx, cfg, node, rootCAs)
+		if err != nil {
+			slog.Debug("predastore health probe: node unreachable", "node", node, "err", err)
+			return predastoreHealthUnreachable
+		}
+		if status.Leader == "" {
+			verdict = predastoreHealthNoLeader
+		}
+	}
+	return verdict
+}
+
+// localMetaNodes returns the meta node ids pinned to this daemon's predastore
+// host, and the config they were parsed from. A node with no predastore host
+// id configured — this node runs no predastore at all — returns an empty
+// slice and no error, which computePredastoreHealth treats as "ok" rather
+// than "unreachable": it is not this host's job to answer for a service it
+// was never asked to run.
+func localMetaNodes(d *Daemon) ([]pds.NodeID, *pds.Config, error) {
+	if d.config.Predastore.HostID <= 0 {
+		return nil, nil, nil
+	}
+
+	cfgPath := predastoreConfigPath(d.configPath)
+	cfg, err := pds.LoadConfig(cfgPath)
+	if err != nil {
+		return nil, nil, fmt.Errorf("load predastore config %s: %w", cfgPath, err)
+	}
+
+	nodes := pds.MetaNodesOnHost(cfg, pds.HostID(d.config.Predastore.HostID))
+	return nodes, cfg, nil
+}
+
+// predastoreConfigPath is where the predastore process reads its own config:
+// a fixed "predastore/predastore.toml" alongside the daemon's own
+// spinifex.toml, matching how the predastore service and handleStorageConfig
+// both resolve it.
+func predastoreConfigPath(daemonConfigPath string) string {
+	return filepath.Join(filepath.Dir(daemonConfigPath), "predastore", "predastore.toml")
+}
+
+// loadPredastoreTrustRoot returns the cluster CA pool predastore's meta
+// replicas are issued from — the same CA (d.config.NATS.CACert, conventionally
+// /etc/spinifex/ca.pem) the daemon already trusts NATS peers with.
+func loadPredastoreTrustRoot(d *Daemon) (*x509.CertPool, error) {
+	if d.config.NATS.CACert == "" {
+		return nil, fmt.Errorf("cluster CA not configured (nats.cacert)")
+	}
+	return utils.LoadCertPool(d.config.NATS.CACert)
+}
@@ -0,0 +1,203 @@
+// Exercises unexported probe internals (serviceProbes, computePredastoreHealth,
+// nodeStatusFn, predastoreHealthCache) that have no exported surface to drive
+// them through.
+//
+//test:in-package
+package daemon
+
+import (
+	"context"
+	"crypto/x509"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	pds "github.com/mulgadc/predastore"
+	"github.com/mulgadc/spinifex/spinifex/config"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// healthTestTOML pins one meta node (id 3) to host 1, and gives host 2 no
+// meta node at all — the two shapes localMetaNodes has to tell apart.
+const healthTestTOML = `
+version = 1
+region = "ap-southeast-2"
+
+[rs]
+data = 1
+parity = 1
+
+[[host]]
+id = 1
+bind_addr = "0.0.0.0"
+addr = "10.0.0.1"
+data_dir = "/var/lib/spinifex/predastore/cluster"
+
+[[host.node]]
+id = 1
+role = "gate"
+port = 8443
+
+[[host.node]]
+id = 2
+role = "blob"
+port = 6660
+
+[[host.node]]
+id = 3
+role = "meta"
+port = 7660
+
+[[host]]
+id = 2
+bind_addr = "0.0.0.0"
+addr = "10.0.0.2"
+data_dir = "/var/lib/spinifex/predastore/cluster"
+
+[[host.node]]
+id = 4
+role = "blob"
+port = 6660
+`
+
+// newHealthTestDaemon writes healthTestTOML and a CA cert under a temp
+// config dir and returns a Daemon whose predastore host id is hostID.
+func newHealthTestDaemon(t *testing.T, hostID int) *Daemon {
+	t.Helper()
+	dir := t.TempDir()
+	require.NoError(t, os.MkdirAll(filepath.Join(dir, "predastore"), 0o750))
+	require.NoError(t, os.WriteFile(filepath.Join(dir, "predastore", "predastore.toml"), []byte(healthTestTOML), 0o600))
+
+	certPEM, _ := generateTestCert(t)
+	caPath := filepath.Join(dir, "ca.pem")
+	require.NoError(t, os.WriteFile(caPath, certPEM, 0o600))
+
+	return &Daemon{
+		configPath: filepath.Join(dir, "spinifex.toml"),
+		config: &config.Config{
+			Predastore: config.PredastoreConfig{HostID: hostID},
+			NATS:       config.NATSConfig{CACert: caPath},
+		},
+	}
+}
+
+// setNodeStatusFn stubs nodeStatusFn for the test's lifetime, restoring the
+// real pds.NodeStatus on cleanup, and returns a pointer to a call counter.
+func setNodeStatusFn(t *testing.T, fn func(context.Context, *pds.Config, pds.NodeID, *x509.CertPool) (pds.Status, error)) *int {
+	t.Helper()
+	calls := 0
+	orig := nodeStatusFn
+	nodeStatusFn = func(ctx context.Context, cfg *pds.Config, node pds.NodeID, rootCAs *x509.CertPool) (pds.Status, error) {
+		calls++
+		return fn(ctx, cfg, node, rootCAs)
+	}
+	t.Cleanup(func() { nodeStatusFn = orig })
+	return &calls
+}
+
+func TestProbeServiceHealth_UnprobedServiceDefaultsToOK(t *testing.T) {
+	d := &Daemon{config: &config.Config{Services: []string{"viperblock"}}}
+	health := d.probeServiceHealth(t.Context())
+	assert.Equal(t, "ok", health["viperblock"])
+}
+
+func TestComputePredastoreHealth_NoMetaNodeOnHost(t *testing.T) {
+	// Host 2 in healthTestTOML runs only a blob node — nothing local to
+	// probe, which is not this host's service failing.
+	d := newHealthTestDaemon(t, 2)
+	calls := setNodeStatusFn(t, func(context.Context, *pds.Config, pds.NodeID, *x509.CertPool) (pds.Status, error) {
+		return pds.Status{}, nil
+	})
+
+	got := computePredastoreHealth(t.Context(), d)
+	assert.Equal(t, predastoreHealthOK, got)
+	assert.Equal(t, 0, *calls, "a host with no meta node must never dial one")
+}
+
+func TestComputePredastoreHealth_NoHostIDConfigured(t *testing.T) {
+	// HostID <= 0: this node runs no predastore at all.
+	d := newHealthTestDaemon(t, 0)
+	got := computePredastoreHealth(t.Context(), d)
+	assert.Equal(t, predastoreHealthOK, got)
+}
+
+func TestComputePredastoreHealth_OK(t *testing.T) {
+	d := newHealthTestDaemon(t, 1)
+	setNodeStatusFn(t, func(context.Context, *pds.Config, pds.NodeID, *x509.CertPool) (pds.Status, error) {
+		return pds.Status{State: "Leader", Leader: "node-3", IsLeader: true}, nil
+	})
+
+	got := computePredastoreHealth(t.Context(), d)
+	assert.Equal(t, predastoreHealthOK, got)
+}
+
+func TestComputePredastoreHealth_NoLeader(t *testing.T) {
+	d := newHealthTestDaemon(t, 1)
+	setNodeStatusFn(t, func(context.Context, *pds.Config, pds.NodeID, *x509.CertPool) (pds.Status, error) {
+		return pds.Status{State: "Follower", Leader: ""}, nil
+	})
+
+	got := computePredastoreHealth(t.Context(), d)
+	assert.Equal(t, predastoreHealthNoLeader, got)
+}
+
+func TestComputePredastoreHealth_UnreachableOnError(t *testing.T) {
+	d := newHealthTestDaemon(t, 1)
+	setNodeStatusFn(t, func(context.Context, *pds.Config, pds.NodeID, *x509.CertPool) (pds.Status, error) {
+		return pds.Status{}, assert.AnError
+	})
+
+	got := computePredastoreHealth(t.Context(), d)
+	assert.Equal(t, predastoreHealthUnreachable, got)
+}
+
+func TestComputePredastoreHealth_UnreachableOnTimeout(t *testing.T) {
+	d := newHealthTestDaemon(t, 1)
+
+	// Shrink the probe timeout so a wedged peer is provably bounded rather
+	// than actually waiting the production default.
+	origTimeout := predastoreProbeTimeout
+	predastoreProbeTimeout = 50 * time.Millisecond
+	t.Cleanup(func() { predastoreProbeTimeout = origTimeout })
+
+	setNodeStatusFn(t, func(ctx context.Context, cfg *pds.Config, node pds.NodeID, rootCAs *x509.CertPool) (pds.Status, error) {
+		<-ctx.Done()
+		return pds.Status{}, ctx.Err()
+	})
+
+	start := time.Now()
+	got := computePredastoreHealth(t.Context(), d)
+	elapsed := time.Since(start)
+
+	assert.Equal(t, predastoreHealthUnreachable, got)
+	assert.Less(t, elapsed, 2*time.Second, "a wedged peer must not stall the handler beyond the probe timeout")
+}
+
+func TestComputePredastoreHealth_NoCAConfigured(t *testing.T) {
+	d := newHealthTestDaemon(t, 1)
+	d.config.NATS.CACert = ""
+
+	got := computePredastoreHealth(t.Context(), d)
+	assert.Equal(t, predastoreHealthUnreachable, got)
+}
+
+func TestProbePredastore_CachesWithinTTL(t *testing.T) {
+	d := newHealthTestDaemon(t, 1)
+
+	origTTL := predastoreHealthCacheTTL
+	predastoreHealthCacheTTL = time.Minute
+	t.Cleanup(func() { predastoreHealthCacheTTL = origTTL })
+
+	calls := setNodeStatusFn(t, func(context.Context, *pds.Config, pds.NodeID, *x509.CertPool) (pds.Status, error) {
+		return pds.Status{State: "Leader", Leader: "node-3"}, nil
+	})
+
+	first := probePredastore(t.Context(), d)
+	second := probePredastore(t.Context(), d)
+
+	assert.Equal(t, predastoreHealthOK, first)
+	assert.Equal(t, predastoreHealthOK, second)
+	assert.Equal(t, 1, *calls, "a second poll within the TTL must not re-dial")
+}
@@ -2,6 +2,7 @@ package gateway

 import (
 	"context"
+	"crypto/x509"
 	"encoding/json"
 	"encoding/xml"
 	"errors"
@@ -87,9 +88,14 @@ type GatewayConfig struct {
 	DisableLogging bool       `json:"disable_logging"`
 	NATSConn       *nats.Conn // Shared NATS connection for service communication
 	Config         string     // Shared AWS Gateway config for S3 auth
-	ExpectedNodes  int        // Number of expected spinifex nodes for multi-node operations
-	Region         string     // Region this gateway is running in
-	InternalSuffix string     // Internal DNS suffix for AWS-parity endpoints (e.g. spinifex.internal)
+	// RootCAs is the cluster CA pool, used to verify predastore meta nodes
+	// dialed directly for GetStorageStatus. Nil disables that verification,
+	// so admin storage status reports every meta node unreachable rather than
+	// dialing with no trust root.
+	RootCAs        *x509.CertPool
+	ExpectedNodes  int    // Number of expected spinifex nodes for multi-node operations
+	Region         string // Region this gateway is running in
+	InternalSuffix string // Internal DNS suffix for AWS-parity endpoints (e.g. spinifex.internal)
 	// RegistryPort is the gateway's advertised port, appended to the ECR
 	// registry host so docker login/tag/push dial the right port. Empty or
 	// "443" renders a port-less host (standard HTTPS parity).
@@ -73,7 +73,7 @@ func (gw *GatewayConfig) Spinifex_Request(w http.ResponseWriter, r *http.Request
 		if gw.NATSConn == nil {
 			return errors.New(awserrors.ErrorServerInternal)
 		}
-		output, err = gateway_spx.GetStorageStatus(gw.NATSConn)
+		output, err = gateway_spx.GetStorageStatus(gw.NATSConn, gw.RootCAs)
 	case "PromoteImage":
 		if gw.NATSConn == nil {
 			return errors.New(awserrors.ErrorServerInternal)
@@ -2,15 +2,14 @@ package spx

 import (
 	"context"
+	"crypto/x509"
 	"encoding/json"
 	"fmt"
 	"log/slog"
-	"net"
-	"net/http"
-	"strconv"
 	"sync"
 	"time"

+	pds "github.com/mulgadc/predastore"
 	"github.com/mulgadc/spinifex/spinifex/types"
 	"github.com/nats-io/nats.go"
 )
@@ -45,13 +44,17 @@ type StorageEncodingOutput struct {
 	ParityShards int    `json:"parity_shards"`
 }

-var storageHTTPClient = &http.Client{
-	Timeout: 1 * time.Second,
-}
+// metaNodeQueryTimeout bounds the whole parallel round of meta node dials, so
+// one unreachable node cannot stall the CLI command waiting on the others.
+const metaNodeQueryTimeout = 2 * time.Second

-// GetStorageStatus fetches predastore topology via NATS, then queries each
-// meta node's /status and /health endpoints in parallel.
-func GetStorageStatus(nc *nats.Conn) (*StorageStatusOutput, error) {
+// GetStorageStatus fetches predastore topology via NATS, then dials each meta
+// node's raft status directly and in parallel. rootCAs is the cluster CA
+// pool: every meta node in the cluster is queried, not just a local one, so
+// verifying each one's identity against the shared CA (rather than a local
+// TLS config only the node owning it could read) is what makes every dial
+// trusted.
+func GetStorageStatus(nc *nats.Conn, rootCAs *x509.CertPool) (*StorageStatusOutput, error) {
 	msg, err := nc.Request("spinifex.storage.config", []byte("{}"), 3*time.Second)
 	if err != nil {
 		return nil, fmt.Errorf("storage config request: %w", err)
@@ -64,7 +67,7 @@ func GetStorageStatus(nc *nats.Conn) (*StorageStatusOutput, error) {

 	metaStatuses := make([]MetaNodeStatus, len(cfg.MetaNodes))
 	var wg sync.WaitGroup
-	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), metaNodeQueryTimeout)
 	defer cancel()

 	for i, meta := range cfg.MetaNodes {
@@ -76,7 +79,7 @@ func GetStorageStatus(nc *nats.Conn) (*StorageStatusOutput, error) {
 		wg.Add(1)
 		go func(idx int, host string, port int) {
 			defer wg.Done()
-			queryMetaNodeStatus(ctx, &metaStatuses[idx], host, port)
+			queryMetaNodeStatus(ctx, &metaStatuses[idx], rootCAs, host, port)
 		}(i, meta.Host, meta.Port)
 	}
 	wg.Wait()
@@ -93,61 +96,65 @@ func GetStorageStatus(nc *nats.Conn) (*StorageStatusOutput, error) {
 	}, nil
 }

-// predastoreStatusResponse matches the predastore s3db.StatusResponse JSON shape.
-type predastoreStatusResponse struct {
-	NodeID     string `json:"node_id"`
-	State      string `json:"state"`
-	Leader     string `json:"leader"`
-	LeaderAddr string `json:"leader_addr"`
-	Term       string `json:"term"`
-	CommitIdx  string `json:"commit_index"`
-	AppliedIdx string `json:"applied_index"`
-	IsLeader   bool   `json:"is_leader"`
-}
+// queryMetaNodeStatus dials one meta node's raft status directly — the
+// opcode-multiplexed RPC predastore's meta nodes actually speak, not the
+// HTTPS endpoints they never served — and fills out with what it reports. A
+// node that does not answer within ctx leaves out untouched (Healthy stays
+// false and every raft field stays blank), matching the CLI's existing
+// contract for an unhealthy node.
+func queryMetaNodeStatus(ctx context.Context, out *MetaNodeStatus, rootCAs *x509.CertPool, host string, port int) {
+	nodeID, ok := metaNodeID(out.ID)
+	if !ok {
+		slog.Debug("queryMetaNodeStatus: invalid node id", "id", out.ID, "host", host, "port", port)
+		return
+	}

-func queryMetaNodeStatus(ctx context.Context, out *MetaNodeStatus, host string, port int) {
-	// Resolve 0.0.0.0 to a routable address for HTTPS.
+	// Resolve 0.0.0.0 to a routable address.
 	queryHost := host
 	if queryHost == "0.0.0.0" {
 		queryHost = "127.0.0.1"
 	}

-	healthURL := "https://" + net.JoinHostPort(queryHost, strconv.Itoa(port)) + "/health"
-	req, err := http.NewRequestWithContext(ctx, http.MethodGet, healthURL, nil)
-	if err != nil {
-		return
-	}
-	resp, err := storageHTTPClient.Do(req)
-	if err != nil {
-		slog.Debug("queryMetaNodeStatus: health check failed", "host", host, "port", port, "err", err)
-		return
-	}
-	resp.Body.Close()
-	out.Healthy = resp.StatusCode == http.StatusOK
-
-	statusURL := "https://" + net.JoinHostPort(queryHost, strconv.Itoa(port)) + "/status"
-	req, err = http.NewRequestWithContext(ctx, http.MethodGet, statusURL, nil)
-	if err != nil {
-		return
-	}
-	resp, err = storageHTTPClient.Do(req)
-	if err != nil {
-		slog.Debug("queryMetaNodeStatus: status check failed", "host", host, "port", port, "err", err)
-		return
-	}
-	defer resp.Body.Close()
-
-	var status predastoreStatusResponse
-	if err := json.NewDecoder(resp.Body).Decode(&status); err != nil {
-		slog.Debug("queryMetaNodeStatus: failed to decode status", "host", host, "port", port, "err", err)
-		return
+	// A single-host, single-node config is all NodeStatus needs to resolve
+	// and dial this one node: it queries that replica only, never redirects,
+	// and asks for no local TLS identity of its own.
+	cfg := &pds.Config{
+		Hosts: []pds.HostConfig{
+			{
+				ID:    pds.HostID(nodeID),
+				Addr:  queryHost,
+				Nodes: []pds.NodeConfig{{ID: nodeID, Role: pds.RoleMeta, Port: port}},
+			},
+		},
 	}

+	status, err := pds.NodeStatus(ctx, cfg, nodeID, rootCAs)
+	if err != nil {
+		slog.Debug("queryMetaNodeStatus: status probe failed", "id", out.ID, "host", host, "port", port, "err", err)
+		return
+	}
+	applyMetaStatus(out, status)
+}
+
+// metaNodeID converts a meta node's id — plain int on the wire, predastore's
+// NodeID (uint64) internally — validating it is never negative first, since
+// predastore ids never are and a blind conversion could wrap.
+func metaNodeID(id int) (pds.NodeID, bool) {
+	if id < 0 {
+		return 0, false
+	}
+	return pds.NodeID(id), true //#nosec G115 -- id validated non-negative above
+}
+
+// applyMetaStatus maps a predastore raft Status into a MetaNodeStatus.
+// Reaching here means the node answered, so Healthy is unconditionally true.
+func applyMetaStatus(out *MetaNodeStatus, status pds.Status) {
+	out.Healthy = true
 	out.State = status.State
 	out.Leader = status.Leader
 	out.LeaderAddr = status.LeaderAddr
 	out.Term = status.Term
-	out.CommitIdx = status.CommitIdx
-	out.AppliedIdx = status.AppliedIdx
+	out.CommitIdx = status.CommitIndex
+	out.AppliedIdx = status.AppliedIndex
 	out.IsLeader = status.IsLeader
 }
@@ -1,14 +1,12 @@
 package spx

 import (
-	"crypto/tls"
+	"context"
 	"encoding/json"
-	"net/http"
-	"net/http/httptest"
-	"net/url"
-	"strconv"
 	"testing"
+	"time"

+	pds "github.com/mulgadc/predastore"
 	"github.com/mulgadc/spinifex/spinifex/types"
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
@@ -23,7 +21,7 @@ func TestGetStorageStatus_Success(t *testing.T) {
 		resp := types.StorageConfigResponse{
 			Encoding: types.StorageEncoding{DataShards: 2, ParityShards: 1},
 			MetaNodes: []types.StorageMetaNode{
-				{ID: 1, Host: "127.0.0.1", Port: 0}, // port set below
+				{ID: 1, Host: "127.0.0.1", Port: 1}, // nothing listens on port 1
 			},
 			BlobNodes: []types.StorageBlobNode{
 				{ID: 1, Host: "0.0.0.0", Port: 9991},
@@ -40,7 +38,7 @@ func TestGetStorageStatus_Success(t *testing.T) {
 	defer sub.Unsubscribe()
 	nc.Flush()

-	out, err := GetStorageStatus(nc)
+	out, err := GetStorageStatus(nc, nil)
 	require.NoError(t, err)

 	assert.Equal(t, "Reed-Solomon", out.Encoding.Type)
@@ -49,75 +47,59 @@ func TestGetStorageStatus_Success(t *testing.T) {
 	assert.Len(t, out.BlobNodes, 2)
 	assert.Len(t, out.Buckets, 1)
 	assert.Equal(t, "predastore", out.Buckets[0].Name)
-	// Meta node health check will fail (no real predastore running) — expected
+	// Meta node status probe will fail (no real predastore running) — expected
 	require.Len(t, out.MetaNodes, 1)
 	assert.Equal(t, 1, out.MetaNodes[0].ID)
-	assert.False(t, out.MetaNodes[0].Healthy) // no predastore server to respond
+	assert.False(t, out.MetaNodes[0].Healthy)
 }

 func TestGetStorageStatus_NoNATSResponse(t *testing.T) {
 	_, nc := startEmbeddedNATS(t)
 	// No subscriber — should timeout
-	_, err := GetStorageStatus(nc)
+	_, err := GetStorageStatus(nc, nil)
 	require.Error(t, err)
 	assert.Contains(t, err.Error(), "storage config request")
 }

-func TestQueryMetaNodeStatus_HealthyNode(t *testing.T) {
-	// Start a mock predastore meta node server
-	mux := http.NewServeMux()
-	mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
-		w.Header().Set("Content-Type", "application/json")
-		w.WriteHeader(http.StatusOK)
-		json.NewEncoder(w).Encode(map[string]any{"status": "healthy"})
+func TestApplyMetaStatus_MapsRaftFields(t *testing.T) {
+	out := &MetaNodeStatus{ID: 1, Host: "10.0.0.1", Port: 6100}
+	applyMetaStatus(out, pds.Status{
+		NodeID:       "1",
+		State:        "Leader",
+		Leader:       "node-1",
+		LeaderAddr:   "",
+		Term:         "42",
+		CommitIndex:  "1000",
+		AppliedIndex: "998",
+		IsLeader:     true,
 	})
-	mux.HandleFunc("/status", func(w http.ResponseWriter, r *http.Request) {
-		w.Header().Set("Content-Type", "application/json")
-		json.NewEncoder(w).Encode(predastoreStatusResponse{
-			NodeID:     "1",
-			State:      "Leader",
-			Leader:     "1",
-			LeaderAddr: "127.0.0.1:7660",
-			Term:       "42",
-			CommitIdx:  "1000",
-			AppliedIdx: "998",
-			IsLeader:   true,
-		})
-	})
-
-	srv := httptest.NewTLSServer(mux)
-	defer srv.Close()
-
-	// Use the test server's TLS client
-	origClient := storageHTTPClient
-	storageHTTPClient = &http.Client{
-		Transport: &http.Transport{
-			TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
-		},
-	}
-	defer func() { storageHTTPClient = origClient }()
-
-	// Parse host:port from the test server URL
-	u, err := url.Parse(srv.URL)
-	require.NoError(t, err)
-	host := u.Hostname()
-	port, err := strconv.Atoi(u.Port())
-	require.NoError(t, err)
-
-	out := &MetaNodeStatus{ID: 1, Host: host, Port: port}
-	queryMetaNodeStatus(t.Context(), out, host, port)

 	assert.True(t, out.Healthy)
 	assert.Equal(t, "Leader", out.State)
+	assert.Equal(t, "node-1", out.Leader)
 	assert.True(t, out.IsLeader)
 	assert.Equal(t, "42", out.Term)
 	assert.Equal(t, "1000", out.CommitIdx)
 	assert.Equal(t, "998", out.AppliedIdx)
 }

-func TestQueryMetaNodeStatus_UnreachableNode(t *testing.T) {
-	out := &MetaNodeStatus{ID: 1, Host: "127.0.0.1", Port: 1} // port 1 won't respond
-	queryMetaNodeStatus(t.Context(), out, "127.0.0.1", 1)
+func TestApplyMetaStatus_NoLeaderStaysHealthy(t *testing.T) {
+	// A replica that answers but observes no leader is still a replica that
+	// answered: Healthy describes reachability, not raft convergence.
+	out := &MetaNodeStatus{ID: 2}
+	applyMetaStatus(out, pds.Status{State: "Follower"})
+
+	assert.True(t, out.Healthy)
+	assert.Equal(t, "Follower", out.State)
+	assert.Empty(t, out.Leader)
+	assert.False(t, out.IsLeader)
+}
+
+func TestQueryMetaNodeStatus_Unreachable(t *testing.T) {
+	out := &MetaNodeStatus{ID: 1, Host: "127.0.0.1", Port: 1} // nothing listens here
+	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+	defer cancel()
+	queryMetaNodeStatus(ctx, out, nil, "127.0.0.1", 1)

 	assert.False(t, out.Healthy)
 	assert.Empty(t, out.State)
@@ -3,6 +3,7 @@ package awsgw
 import (
 	"context"
 	"crypto/tls"
+	"crypto/x509"
 	"errors"
 	"fmt"
 	"log/slog"
@@ -151,6 +152,16 @@ func launchService(config *config.ClusterConfig) error {
 	}
 	defer natsConn.Close()

+	// The same cluster CA verifies predastore meta nodes GetStorageStatus
+	// dials directly, across every host in the cluster.
+	var rootCAs *x509.CertPool
+	if nodeConfig.NATS.CACert != "" {
+		rootCAs, err = utils.LoadCertPool(nodeConfig.NATS.CACert)
+		if err != nil {
+			return fmt.Errorf("load cluster CA: %w", err)
+		}
+	}
+
 	// Append Base dir if config has no leading path
 	if nodeConfig.BaseDir != "" && !strings.HasPrefix(nodeConfig.AWSGW.Config, "/") {
 		nodeConfig.AWSGW.Config = fmt.Sprintf("%s/%s", nodeConfig.BaseDir, nodeConfig.AWSGW.Config)
@@ -388,6 +399,7 @@ func launchService(config *config.ClusterConfig) error {
 		Debug:                   nodeConfig.AWSGW.Debug,
 		DisableLogging:          false,
 		NATSConn:                natsConn,
+		RootCAs:                 rootCAs,
 		Config:                  nodeConfig.AWSGW.Config,
 		ExpectedNodes:           len(config.Nodes),
 		Region:                  nodeConfig.Region,
@@ -25,6 +25,22 @@ var (
 // ErrClusterUnavailable is returned when the NATS connection is not currently connected.
 var ErrClusterUnavailable = errors.New("cluster unavailable: NATS disconnected")

+// LoadCertPool reads a PEM CA certificate from path and returns a pool
+// trusting exactly it. Shared by every caller that verifies a peer against
+// the cluster CA (e.g. /etc/spinifex/ca.pem) rather than the system trust
+// store.
+func LoadCertPool(path string) (*x509.CertPool, error) {
+	caCert, err := os.ReadFile(path)
+	if err != nil {
+		return nil, fmt.Errorf("%w %s: %w", ErrCACertRead, path, err)
+	}
+	pool := x509.NewCertPool()
+	if !pool.AppendCertsFromPEM(caCert) {
+		return nil, fmt.Errorf("%w from %s", ErrCACertParse, path)
+	}
+	return pool, nil
+}
+
 // natsRetryEscalateAttempt is the threshold at which retry logs escalate from Warn to Error (rate-limited to 1/min).
 // ~30 attempts at the 60 s backoff cap ≈ ~30 min disconnected, suggesting config error not transient restart.
 const natsRetryEscalateAttempt = 30
@@ -59,13 +75,9 @@ func ConnectNATS(host, token, caCertPath string, opts ...RetryOption) (*nats.Con
 	}

 	if caCertPath != "" {
-		caCert, err := os.ReadFile(caCertPath)
+		pool, err := LoadCertPool(caCertPath)
 		if err != nil {
-			return nil, fmt.Errorf("%w %s: %w", ErrCACertRead, caCertPath, err)
-		}
-		pool := x509.NewCertPool()
-		if !pool.AppendCertsFromPEM(caCert) {
-			return nil, fmt.Errorf("%w from %s", ErrCACertParse, caCertPath)
+			return nil, err
 		}
 		natsOpts = append(natsOpts, nats.Secure(&tls.Config{
 			RootCAs:          pool,