client/daemon: route liveness / bfd / run error

Author: snormore

client/doublezerod/internal/liveness/manager.go

@@ -47,6 +47,7 @@ type RouteKey struct {
 type ManagerConfig struct {
 	Logger    *slog.Logger
 	Netlinker RouteReaderWriter
+	UDP       *UDPService
 
 	BindIP string // local bind address for the UDP socket (IPv4)
 	Port   int    // UDP port to listen/transmit on
@@ -124,6 +125,7 @@ type Manager struct {
 	ctx    context.Context
 	cancel context.CancelFunc
 	wg     sync.WaitGroup
+	errCh  chan error
 
 	log *slog.Logger
 	cfg *ManagerConfig
@@ -150,9 +152,13 @@ func NewManager(ctx context.Context, cfg *ManagerConfig) (*Manager, error) {
 		return nil, fmt.Errorf("error validating manager config: %v", err)
 	}
 
-	udp, err := ListenUDP(cfg.BindIP, cfg.Port)
-	if err != nil {
-		return nil, fmt.Errorf("error creating UDP connection: %v", err)
+	udp := cfg.UDP
+	if udp == nil {
+		var err error
+		udp, err = ListenUDP(cfg.BindIP, cfg.Port)
+		if err != nil {
+			return nil, fmt.Errorf("error creating UDP connection: %w", err)
+		}
 	}
 
 	log := cfg.Logger
@@ -162,6 +168,7 @@ func NewManager(ctx context.Context, cfg *ManagerConfig) (*Manager, error) {
 	m := &Manager{
 		ctx:    ctx,
 		cancel: cancel,
+		errCh:  make(chan error, 10),
 
 		log: log,
 		cfg: cfg,
@@ -184,22 +191,32 @@ func NewManager(ctx context.Context, cfg *ManagerConfig) (*Manager, error) {
 		defer m.wg.Done()
 		err := m.recv.Run(m.ctx)
 		if err != nil {
-			// TODO(snormore): What should we do when this returns an error? Reconnect/retry or
-			// propagate up and exit the daemon?
 			m.log.Error("liveness: error running receiver", "error", err)
+			cancel()
+			m.errCh <- err
 		}
 	}()
 
 	// Scheduler goroutine: handles periodic TX and detect expirations.
 	m.wg.Add(1)
 	go func() {
 		defer m.wg.Done()
-		m.sched.Run(m.ctx)
+		err := m.sched.Run(m.ctx)
+		if err != nil {
+			m.log.Error("liveness: error running scheduler", "error", err)
+			cancel()
+			m.errCh <- err
+		}
 	}()
 
 	return m, nil
 }
 
+// Err returns a channel that will receive any errors from the manager.
+func (m *Manager) Err() chan error {
+	return m.errCh
+}
+
 // RegisterRoute declares interest in monitoring reachability for route r via iface.
 // It optionally installs the route immediately in PassiveMode, then creates or
 // reuses a liveness Session and schedules immediate TX to begin handshake.

client/doublezerod/internal/liveness/manager_test.go

@@ -452,6 +452,71 @@ func TestClient_LivenessManager_PeerKey_IPv4Canonicalization(t *testing.T) {
 	require.True(t, ok, "peer key should use IPv4 string forms")
 }
 
+func TestClient_Liveness_Manager_ReceiverFailure_PropagatesOnErr(t *testing.T) {
+	t.Parallel()
+	m, err := newTestManager(t, nil)
+	require.NoError(t, err)
+	defer func() { _ = m.Close() }()
+
+	errCh := m.Err()
+
+	// Close the UDP socket directly to force Receiver.Run to error out.
+	var udp *UDPService
+	m.mu.Lock()
+	udp = m.udp
+	m.mu.Unlock()
+	require.NotNil(t, udp)
+	_ = udp.Close()
+
+	// Expect an error to surface on Err().
+	select {
+	case e := <-errCh:
+		require.Error(t, e)
+	default:
+		select {
+		case e := <-errCh:
+			require.Error(t, e)
+		case <-time.After(2 * time.Second):
+			t.Fatalf("timeout waiting for error from manager.Err after UDP close")
+		}
+	}
+
+	// Close should complete cleanly after the receiver failure.
+	require.NoError(t, m.Close())
+}
+
+func TestClient_Liveness_Manager_Close_NoErrOnErrCh(t *testing.T) {
+	t.Parallel()
+	m, err := newTestManager(t, nil)
+	require.NoError(t, err)
+
+	// No spurious errors before close.
+	func() {
+		timer := time.NewTimer(200 * time.Millisecond)
+		defer timer.Stop()
+		select {
+		case <-timer.C:
+			return
+		case <-m.Err():
+			t.Fatalf("unexpected error before Close")
+		}
+	}()
+
+	require.NoError(t, m.Close())
+
+	// No spurious errors after close either.
+	func() {
+		timer := time.NewTimer(200 * time.Millisecond)
+		defer timer.Stop()
+		select {
+		case <-timer.C:
+			return
+		case <-m.Err():
+			t.Fatalf("unexpected error after Close")
+		}
+	}()
+}
+
 func newTestManager(t *testing.T, mutate func(*ManagerConfig)) (*Manager, error) {
 	cfg := &ManagerConfig{
 		Logger:     newTestLogger(t),

client/doublezerod/internal/liveness/scheduler.go

@@ -136,7 +136,7 @@ func NewScheduler(log *slog.Logger, udp *UDPService, onSessionDown func(s *Sessi
 // It continuously pops and processes due events, sleeping until the next one if necessary.
 // Each TX event sends a control packet and re-schedules the next TX;
 // each Detect event checks for timeout and invokes onSessionDown if expired.
-func (s *Scheduler) Run(ctx context.Context) {
+func (s *Scheduler) Run(ctx context.Context) error {
 	s.log.Debug("liveness.scheduler: tx loop started")
 
 	t := time.NewTimer(time.Hour)
@@ -146,7 +146,7 @@ func (s *Scheduler) Run(ctx context.Context) {
 		select {
 		case <-ctx.Done():
 			s.log.Debug("liveness.scheduler: stopped by context done", "reason", ctx.Err())
-			return
+			return nil
 		default:
 		}
 
@@ -167,7 +167,7 @@ func (s *Scheduler) Run(ctx context.Context) {
 			select {
 			case <-ctx.Done():
 				s.log.Debug("liveness.scheduler: stopped by context done", "reason", ctx.Err())
-				return
+				return nil
 			case <-t.C:
 				continue
 			}

client/doublezerod/internal/liveness/scheduler_test.go

@@ -183,7 +183,9 @@ func TestClient_Liveness_Scheduler_Run_SendsAndReschedules(t *testing.T) {
 	s := NewScheduler(log, w, func(*Session) {})
 	ctx, cancel := context.WithCancel(t.Context())
 	defer cancel()
-	go s.Run(ctx)
+	go func() {
+		require.NoError(t, s.Run(ctx))
+	}()
 
 	sess := &Session{
 		state:         StateInit,

client/doublezerod/internal/runtime/run.go

@@ -133,5 +133,7 @@ func Run(ctx context.Context, sockFile string, routeConfigPath string, enableLat
 		return nil
 	case err := <-errCh:
 		return err
+	case err := <-lm.Err():
+		return err
 	}
 }

client/doublezerod/internal/runtime/run_test.go

@@ -33,6 +33,7 @@ import (
 	"github.com/malbeclabs/doublezero/client/doublezerod/internal/liveness"
 	"github.com/malbeclabs/doublezero/client/doublezerod/internal/pim"
 	"github.com/malbeclabs/doublezero/client/doublezerod/internal/runtime"
+	"github.com/stretchr/testify/require"
 	"golang.org/x/net/ipv4"
 	"golang.org/x/sys/unix"
 
@@ -1649,12 +1650,11 @@ func TestServiceNoCoExistence(t *testing.T) {
 	}()
 
 	errChan := make(chan error, 1)
-	ctx, _ := context.WithCancel(context.Background())
 
 	sockFile := filepath.Join(rootPath, "doublezerod.sock")
 	go func() {
 		programId := ""
-		err := runtime.Run(ctx, sockFile, "", false, false, programId, "", 30, 30, newTestLivenessManagerConfig())
+		err := runtime.Run(t.Context(), sockFile, "", false, false, programId, "", 30, 30, newTestLivenessManagerConfig())
 		errChan <- err
 	}()
 
@@ -2052,6 +2052,120 @@ func TestServiceCoexistence(t *testing.T) {
 	})
 }
 
+func TestRuntime_Run_ReturnsOnContextCancel(t *testing.T) {
+	errChan := make(chan error, 1)
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+
+	rootPath, err := os.MkdirTemp("", "doublezerod")
+	require.NoError(t, err)
+	defer os.RemoveAll(rootPath)
+	t.Setenv("XDG_STATE_HOME", rootPath)
+
+	path := filepath.Join(rootPath, "doublezerod")
+	if err := os.Mkdir(path, 0766); err != nil {
+		t.Fatalf("error creating state dir: %v", err)
+	}
+
+	sockFile := filepath.Join(rootPath, "doublezerod.sock")
+	go func() {
+		programId := ""
+		err := runtime.Run(ctx, sockFile, "", false, false, programId, "", 30, 30, newTestLivenessManagerConfig())
+		errChan <- err
+	}()
+
+	// Give the runtime a moment to start, then cancel the context to force exit.
+	select {
+	case err := <-errChan:
+		require.NoError(t, err)
+	case <-time.After(300 * time.Millisecond):
+	}
+
+	cancel()
+	select {
+	case err := <-errChan:
+		require.NoError(t, err)
+	case <-time.After(5 * time.Second):
+		t.Fatalf("timed out waiting for runtime to exit after context cancel")
+	}
+}
+
+func TestRuntime_Run_PropagatesLivenessStartupError(t *testing.T) {
+	errChan := make(chan error, 1)
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+
+	rootPath, err := os.MkdirTemp("", "doublezerod")
+	require.NoError(t, err)
+	defer os.RemoveAll(rootPath)
+	t.Setenv("XDG_STATE_HOME", rootPath)
+
+	// Invalid liveness config (port < 0) -> NewManager.Validate() error.
+	bad := *newTestLivenessManagerConfig()
+	bad.Port = -1
+
+	sockFile := filepath.Join(rootPath, "doublezerod.sock")
+	go func() {
+		programId := ""
+		err := runtime.Run(ctx, sockFile, "", false, false, programId, "", 30, 30, &bad)
+		errChan <- err
+	}()
+
+	select {
+	case err := <-errChan:
+		require.Error(t, err)
+		require.Contains(t, err.Error(), "port must be greater than or equal to 0")
+	case <-time.After(5 * time.Second):
+		t.Fatalf("expected startup error from runtime.Run with bad liveness config")
+	}
+}
+
+func TestRuntime_Run_PropagatesLivenessError_FromUDPClosure(t *testing.T) {
+	errCh := make(chan error, 1)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	// Minimal state dir + socket path
+	rootPath, err := os.MkdirTemp("", "doublezerod")
+	if err != nil {
+		t.Fatalf("mktemp: %v", err)
+	}
+	defer os.RemoveAll(rootPath)
+	t.Setenv("XDG_STATE_HOME", rootPath)
+	sockFile := filepath.Join(rootPath, "doublezerod.sock")
+
+	// Create a real UDPService we can close to induce a receiver error.
+	udp, err := liveness.ListenUDP("127.0.0.1", 0)
+	if err != nil {
+		t.Fatalf("ListenUDP: %v", err)
+	}
+
+	// Build a liveness config that uses our injected UDP service.
+	cfg := newTestLivenessManagerConfig()
+	cfg.UDP = udp
+	cfg.PassiveMode = true
+
+	// Start the runtime.
+	go func() {
+		programID := ""
+		errCh <- runtime.Run(ctx, sockFile, "", false, false, programID, "", 30, 30, cfg)
+	}()
+
+	// Give the liveness receiver a moment to start, then close the UDP socket.
+	time.Sleep(200 * time.Millisecond)
+	_ = udp.Close()
+
+	// The receiver should error, Manager should send on lm.Err(), and Run should return that error.
+	select {
+	case err := <-errCh:
+		if err == nil {
+			t.Fatalf("expected non-nil error propagated from liveness manager, got nil")
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatalf("timeout waiting for runtime to return error from liveness manager")
+	}
+}
+
 func setupTest(t *testing.T) (func(), error) {
 	abortIfLinksAreUp(t)
 	rootPath, err := os.MkdirTemp("", "doublezerod")