diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index d6c0f8547..eaf7aefed 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -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 diff --git a/spinifex/daemon/daemon_test.go b/spinifex/daemon/daemon_test.go index 34724fe45..ccef2d1cf 100644 --- a/spinifex/daemon/daemon_test.go +++ b/spinifex/daemon/daemon_test.go @@ -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() diff --git a/spinifex/services/awsgw/awsgw.go b/spinifex/services/awsgw/awsgw.go index e46de75b9..add0370b3 100644 --- a/spinifex/services/awsgw/awsgw.go +++ b/spinifex/services/awsgw/awsgw.go @@ -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. diff --git a/spinifex/utils/nats.go b/spinifex/utils/nats.go index e4b7b84e2..3d7feceb3 100644 --- a/spinifex/utils/nats.go +++ b/spinifex/utils/nats.go @@ -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" diff --git a/spinifex/utils/nats_test.go b/spinifex/utils/nats_test.go index b382ee6ac..71cd4649f 100644 --- a/spinifex/utils/nats_test.go +++ b/spinifex/utils/nats_test.go @@ -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") +}