mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-21 19:56:33 +00:00
fix: sentinel TLS errors + shared NATS retry (#152)
* fix: replace stringly-typed TLS error detection with sentinel errors Export ErrCACertRead and ErrCACertParse from utils/nats.go so callers can detect permanent TLS configuration failures with errors.Is instead of substring matching on error messages. Update daemon.go and awsgw.go retry loops accordingly, and update tests to use assert.ErrorIs. Refs: mulga-siv-1 * refactor: extract shared NATS connect-with-retry into utils daemon.connectNATS() and awsgw.connectNATS() were near-identical retry loops. Extract to utils.ConnectNATSWithRetry() with functional options for test configurability.
This commit is contained in:
@@ -144,9 +144,8 @@ type Daemon struct {
|
||||
// Delay after QMP device_del before blockdev-del (default 1s, 0 in tests)
|
||||
detachDelay time.Duration
|
||||
|
||||
// NATS connect retry parameters (zero values use defaults: 5min max, 500ms initial delay)
|
||||
natsMaxWait time.Duration
|
||||
natsRetryDelay time.Duration
|
||||
// NATS connect retry options (nil uses defaults: 5min max, 500ms initial delay)
|
||||
natsRetryOpts []utils.RetryOption
|
||||
|
||||
// NetworkPlumber handles tap device lifecycle for VPC networking
|
||||
networkPlumber NetworkPlumber
|
||||
@@ -839,40 +838,12 @@ func (d *Daemon) Start() error {
|
||||
// be ready immediately after daemon start (e.g. if start-dev.sh is still
|
||||
// launching services). This retries for up to 5 minutes before giving up.
|
||||
func (d *Daemon) connectNATS() error {
|
||||
maxWait := 5 * time.Minute
|
||||
retryDelay := 500 * time.Millisecond
|
||||
if d.natsMaxWait > 0 {
|
||||
maxWait = d.natsMaxWait
|
||||
}
|
||||
if d.natsRetryDelay > 0 {
|
||||
retryDelay = d.natsRetryDelay
|
||||
}
|
||||
start := time.Now()
|
||||
|
||||
for {
|
||||
nc, err := utils.ConnectNATS(d.config.NATS.Host, d.config.NATS.ACL.Token, d.config.NATS.CACert)
|
||||
if err == nil {
|
||||
d.natsConn = nc
|
||||
if time.Since(start) > time.Second {
|
||||
slog.Info("NATS connection established", "elapsed", time.Since(start))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// TLS configuration errors are permanent — retrying will not help.
|
||||
if strings.Contains(err.Error(), "CA cert") {
|
||||
return fmt.Errorf("NATS TLS configuration error: %w", err)
|
||||
}
|
||||
|
||||
elapsed := time.Since(start)
|
||||
if elapsed >= maxWait {
|
||||
return fmt.Errorf("NATS connect failed after %s: %w", elapsed.Round(time.Second), err)
|
||||
}
|
||||
|
||||
slog.Warn("NATS not ready, retrying...", "error", err, "elapsed", elapsed.Round(time.Second), "retryIn", retryDelay)
|
||||
time.Sleep(retryDelay)
|
||||
retryDelay = min(retryDelay*2, 10*time.Second)
|
||||
nc, err := utils.ConnectNATSWithRetry(d.config.NATS.Host, d.config.NATS.ACL.Token, d.config.NATS.CACert, d.natsRetryOpts...)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
d.natsConn = nc
|
||||
return nil
|
||||
}
|
||||
|
||||
// initJetStream initializes JetStream with retry/backoff and upgrades replicas
|
||||
|
||||
@@ -32,6 +32,7 @@ import (
|
||||
"github.com/mulgadc/spinifex/spinifex/objectstore"
|
||||
"github.com/mulgadc/spinifex/spinifex/qmp"
|
||||
"github.com/mulgadc/spinifex/spinifex/types"
|
||||
"github.com/mulgadc/spinifex/spinifex/utils"
|
||||
"github.com/mulgadc/spinifex/spinifex/vm"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/stretchr/testify/assert"
|
||||
@@ -3744,8 +3745,10 @@ func TestConnectNATS_RetriesOnFailure(t *testing.T) {
|
||||
clusterCfg.Nodes["node-1"] = cfg
|
||||
daemon, err := NewDaemon(clusterCfg)
|
||||
require.NoError(t, err)
|
||||
daemon.natsMaxWait = 500 * time.Millisecond
|
||||
daemon.natsRetryDelay = 50 * time.Millisecond
|
||||
daemon.natsRetryOpts = []utils.RetryOption{
|
||||
utils.WithMaxWait(500 * time.Millisecond),
|
||||
utils.WithRetryDelay(50 * time.Millisecond),
|
||||
}
|
||||
|
||||
start := time.Now()
|
||||
err = daemon.connectNATS()
|
||||
|
||||
@@ -102,7 +102,7 @@ func launchService(config *config.ClusterConfig) error {
|
||||
|
||||
// Connect to NATS for service communication. On concurrent startup the
|
||||
// local NATS server may not be listening yet, so retry with backoff.
|
||||
natsConn, err := connectNATS(nodeConfig.NATS.Host, nodeConfig.NATS.ACL.Token, nodeConfig.NATS.CACert)
|
||||
natsConn, err := utils.ConnectNATSWithRetry(nodeConfig.NATS.Host, nodeConfig.NATS.ACL.Token, nodeConfig.NATS.CACert)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -203,38 +203,6 @@ func launchService(config *config.ClusterConfig) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// connectNATS establishes a connection to NATS with retry/backoff. On concurrent
|
||||
// startup the local NATS server may not be listening yet.
|
||||
func connectNATS(host, token, caCertPath string) (*nats.Conn, error) {
|
||||
const maxWait = 5 * time.Minute
|
||||
retryDelay := 500 * time.Millisecond
|
||||
start := time.Now()
|
||||
|
||||
for {
|
||||
nc, err := utils.ConnectNATS(host, token, caCertPath)
|
||||
if err == nil {
|
||||
if time.Since(start) > time.Second {
|
||||
slog.Info("NATS connection established", "elapsed", time.Since(start).Round(time.Second))
|
||||
}
|
||||
return nc, nil
|
||||
}
|
||||
|
||||
// TLS configuration errors are permanent — retrying will not help.
|
||||
if strings.Contains(err.Error(), "CA cert") {
|
||||
return nil, fmt.Errorf("NATS TLS configuration error: %w", err)
|
||||
}
|
||||
|
||||
elapsed := time.Since(start)
|
||||
if elapsed >= maxWait {
|
||||
return nil, fmt.Errorf("NATS connect failed after %s: %w", elapsed.Round(time.Second), err)
|
||||
}
|
||||
|
||||
slog.Warn("NATS not ready, retrying...", "error", err, "elapsed", elapsed.Round(time.Second), "retryIn", retryDelay)
|
||||
time.Sleep(retryDelay)
|
||||
retryDelay = min(retryDelay*2, 10*time.Second)
|
||||
}
|
||||
}
|
||||
|
||||
// initIAMService initializes the IAM service with retry/backoff. On multi-node
|
||||
// clusters, JetStream requires NATS cluster quorum before KV buckets can be
|
||||
// created. This retries for up to 5 minutes to allow late-joining nodes.
|
||||
|
||||
+68
-2
@@ -13,6 +13,14 @@ import (
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
// Sentinel errors for TLS certificate configuration failures in ConnectNATS.
|
||||
// Callers can use errors.Is to detect permanent TLS errors without relying
|
||||
// on error message text.
|
||||
var (
|
||||
ErrCACertRead = errors.New("failed to read CA cert")
|
||||
ErrCACertParse = errors.New("failed to parse CA cert")
|
||||
)
|
||||
|
||||
// ConnectNATS establishes a connection to a NATS server with standard reconnect
|
||||
// handling and logging. If token is non-empty, token authentication is used.
|
||||
// If caCertPath is non-empty, TLS is enabled using the given CA certificate.
|
||||
@@ -35,11 +43,11 @@ func ConnectNATS(host, token, caCertPath string) (*nats.Conn, error) {
|
||||
if caCertPath != "" {
|
||||
caCert, err := os.ReadFile(caCertPath)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("read CA cert %s: %w", caCertPath, err)
|
||||
return nil, fmt.Errorf("%w %s: %v", ErrCACertRead, caCertPath, err)
|
||||
}
|
||||
pool := x509.NewCertPool()
|
||||
if !pool.AppendCertsFromPEM(caCert) {
|
||||
return nil, fmt.Errorf("failed to parse CA cert from %s", caCertPath)
|
||||
return nil, fmt.Errorf("%w from %s", ErrCACertParse, caCertPath)
|
||||
}
|
||||
opts = append(opts, nats.Secure(&tls.Config{
|
||||
RootCAs: pool,
|
||||
@@ -55,6 +63,64 @@ func ConnectNATS(host, token, caCertPath string) (*nats.Conn, error) {
|
||||
return nc, nil
|
||||
}
|
||||
|
||||
// retryConfig holds parameters for ConnectNATSWithRetry.
|
||||
type retryConfig struct {
|
||||
maxWait time.Duration
|
||||
retryDelay time.Duration
|
||||
}
|
||||
|
||||
// RetryOption configures ConnectNATSWithRetry behavior.
|
||||
type RetryOption func(*retryConfig)
|
||||
|
||||
// WithMaxWait sets the maximum total time to keep retrying before giving up.
|
||||
func WithMaxWait(d time.Duration) RetryOption {
|
||||
return func(c *retryConfig) { c.maxWait = d }
|
||||
}
|
||||
|
||||
// WithRetryDelay sets the initial delay between retries (doubles each attempt, capped at 10s).
|
||||
func WithRetryDelay(d time.Duration) RetryOption {
|
||||
return func(c *retryConfig) { c.retryDelay = d }
|
||||
}
|
||||
|
||||
// ConnectNATSWithRetry calls ConnectNATS in a retry loop with exponential
|
||||
// backoff. It retries for up to 5 minutes (default) before giving up. TLS
|
||||
// configuration errors (ErrCACertRead, ErrCACertParse) are permanent and
|
||||
// cause an immediate return without retrying.
|
||||
func ConnectNATSWithRetry(host, token, caCertPath string, opts ...RetryOption) (*nats.Conn, error) {
|
||||
cfg := retryConfig{
|
||||
maxWait: 5 * time.Minute,
|
||||
retryDelay: 500 * time.Millisecond,
|
||||
}
|
||||
for _, o := range opts {
|
||||
o(&cfg)
|
||||
}
|
||||
|
||||
start := time.Now()
|
||||
for {
|
||||
nc, err := ConnectNATS(host, token, caCertPath)
|
||||
if err == nil {
|
||||
if time.Since(start) > time.Second {
|
||||
slog.Info("NATS connection established", "elapsed", time.Since(start).Round(time.Second))
|
||||
}
|
||||
return nc, nil
|
||||
}
|
||||
|
||||
// TLS configuration errors are permanent — retrying will not help.
|
||||
if errors.Is(err, ErrCACertRead) || errors.Is(err, ErrCACertParse) {
|
||||
return nil, fmt.Errorf("NATS TLS configuration error: %w", err)
|
||||
}
|
||||
|
||||
elapsed := time.Since(start)
|
||||
if elapsed >= cfg.maxWait {
|
||||
return nil, fmt.Errorf("NATS connect failed after %s: %w", elapsed.Round(time.Second), err)
|
||||
}
|
||||
|
||||
slog.Warn("NATS not ready, retrying...", "error", err, "elapsed", elapsed.Round(time.Second), "retryIn", cfg.retryDelay)
|
||||
time.Sleep(cfg.retryDelay)
|
||||
cfg.retryDelay = min(cfg.retryDelay*2, 10*time.Second)
|
||||
}
|
||||
}
|
||||
|
||||
// AccountIDHeader is the NATS message header key used to pass the caller's
|
||||
// AWS account ID from the gateway to daemon handlers.
|
||||
const AccountIDHeader = "X-Account-ID"
|
||||
|
||||
@@ -90,7 +90,7 @@ func TestConnectNATS_BadAddress(t *testing.T) {
|
||||
func TestConnectNATS_MissingCACert(t *testing.T) {
|
||||
_, err := ConnectNATS("nats://127.0.0.1:4222", "", "/nonexistent/ca.pem")
|
||||
require.Error(t, err)
|
||||
assert.Contains(t, err.Error(), "read CA cert")
|
||||
assert.ErrorIs(t, err, ErrCACertRead)
|
||||
assert.Contains(t, err.Error(), "/nonexistent/ca.pem")
|
||||
}
|
||||
|
||||
@@ -101,7 +101,7 @@ func TestConnectNATS_MalformedCACert(t *testing.T) {
|
||||
|
||||
_, err := ConnectNATS("nats://127.0.0.1:4222", "", badCert)
|
||||
require.Error(t, err)
|
||||
assert.Contains(t, err.Error(), "failed to parse CA cert")
|
||||
assert.ErrorIs(t, err, ErrCACertParse)
|
||||
}
|
||||
|
||||
// generateTestCA creates an ephemeral CA cert+key and writes PEM files to dir.
|
||||
@@ -535,3 +535,41 @@ func TestAccountIDFromMsg_NilHeader(t *testing.T) {
|
||||
msg := &nats.Msg{Subject: "test"}
|
||||
assert.Equal(t, "", AccountIDFromMsg(msg))
|
||||
}
|
||||
|
||||
// --- ConnectNATSWithRetry tests ---
|
||||
|
||||
func TestConnectNATSWithRetry_Success(t *testing.T) {
|
||||
ns := startTestNATSServer(t)
|
||||
|
||||
nc, err := ConnectNATSWithRetry(ns.ClientURL(), "", "")
|
||||
require.NoError(t, err)
|
||||
defer nc.Close()
|
||||
assert.True(t, nc.IsConnected())
|
||||
}
|
||||
|
||||
func TestConnectNATSWithRetry_RetriesOnFailure(t *testing.T) {
|
||||
start := time.Now()
|
||||
_, err := ConnectNATSWithRetry("nats://127.0.0.1:14222", "", "",
|
||||
WithMaxWait(500*time.Millisecond),
|
||||
WithRetryDelay(50*time.Millisecond),
|
||||
)
|
||||
elapsed := time.Since(start)
|
||||
|
||||
require.Error(t, err)
|
||||
assert.Contains(t, err.Error(), "NATS connect failed")
|
||||
assert.GreaterOrEqual(t, elapsed, 100*time.Millisecond, "should have retried at least once")
|
||||
assert.Less(t, elapsed, 5*time.Second, "should fail within a few seconds")
|
||||
}
|
||||
|
||||
func TestConnectNATSWithRetry_TLSErrorNoRetry(t *testing.T) {
|
||||
start := time.Now()
|
||||
_, err := ConnectNATSWithRetry("nats://127.0.0.1:4222", "", "/nonexistent/ca.pem",
|
||||
WithMaxWait(5*time.Second),
|
||||
)
|
||||
elapsed := time.Since(start)
|
||||
|
||||
require.Error(t, err)
|
||||
assert.ErrorIs(t, err, ErrCACertRead)
|
||||
assert.Contains(t, err.Error(), "NATS TLS configuration error")
|
||||
assert.Less(t, elapsed, time.Second, "should fail immediately without retrying")
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user