fix(manager): surface liveness subscribe failures instead of reporting success

livenessMonitor.Subscribe armed the epoll subscription inside a
syscall.RawConn.Control callback that assigned failures to the function's
named error return -- which the enclosing `err = rawConn.Control(...)`
statement then overwrote with Control's own nil result. A failed
unix.SetNonblock or unix.EpollCtl was therefore logged, "Subscribe daemon
... liveness event" was still printed, and Subscribe returned nil.

The caller then believes the daemon is monitored when it was never added
to the epoll interest list: when that nydusd later dies no death event
fires, so restart/failover recovery never starts and the daemon's
containers are left with dead FUSE mounts -- with nothing in the logs
explaining why. The dialed unix connection also leaked on this path.

Introduce a controlFD helper that returns the first error from either
Control itself or the callback, make Subscribe all-or-nothing (close the
connection and register nothing on failure), and use the same helper in
unsubscribe, where an EPOLL_CTL_DEL failure previously left a stale fd
entry in the interest set that could collide with a future subscription
reusing the same fd number.

Signed-off-by: Iaroslav Geraskin <iaroslav@reflection.ai>

Author: iaroslav-reflection

pkg/manager/monitor.go

@@ -115,39 +115,60 @@ func (m *livenessMonitor) Subscribe(id string, path string, notifier chan<- deat
 	}
 
 	if rawConn, err = uc.SyscallConn(); err != nil {
+		if closeErr := uc.Close(); closeErr != nil {
+			log.L.WithError(closeErr).Warnf("close liveness connection for daemon %s", id)
+		}
 		return
 	}
 
-	err = rawConn.Control(func(fd FD) {
-		err = unix.SetNonblock(int(fd), true)
-		if err != nil {
-			log.L.Errorf("Failed to set file. daemon id %s path %s. %v", id, path, err)
-			return
+	if err = controlFD(rawConn, func(fd FD) error {
+		if err := unix.SetNonblock(int(fd), true); err != nil {
+			return errors.Wrapf(err, "set connection non-blocking, daemon id %s path %s", id, path)
 		}
 
 		event := unix.EpollEvent{
 			Fd:     int32(fd),
 			Events: unix.EPOLLHUP | unix.EPOLLERR | unix.EPOLLET,
 		}
 
-		err = unix.EpollCtl(m.epollFd, unix.EPOLL_CTL_ADD, int(fd), &event)
-		if err != nil {
-			log.L.Errorf("Failed to control epoll. daemon id %s path %s. %v", id, path, err)
-			return
+		if err := unix.EpollCtl(m.epollFd, unix.EPOLL_CTL_ADD, int(fd), &event); err != nil {
+			return errors.Wrapf(err, "add connection to epoll interest list, daemon id %s path %s", id, path)
 		}
 		target := &target{uc: uc, id: id, path: path}
 
 		// Only add subscribed target when everything is OK.
 		m.set[fd] = target
 		m.subscribers[id] = target
 		target.notifier = notifier
-	})
+		return nil
+	}); err != nil {
+		// Leave nothing behind on failure: reporting success for a daemon that
+		// was never added to the epoll interest list means its death would
+		// never be noticed and recovery would never start.
+		if closeErr := uc.Close(); closeErr != nil {
+			log.L.WithError(closeErr).Warnf("close liveness connection for daemon %s", id)
+		}
+		return err
+	}
 
 	log.L.Infof("Subscribe daemon %s liveness event, path=%s.", id, path)
 
 	return
 }
 
+// controlFD runs fn on the raw connection's file descriptor and returns the
+// first error from either the Control call itself or from fn.
+// syscall.RawConn.Control only reports its own failure; an error assigned
+// inside the callback to a variable that Control's result is later assigned to
+// would be silently overwritten.
+func controlFD(rc syscall.RawConn, fn func(fd FD) error) error {
+	var fnErr error
+	if err := rc.Control(func(fd uintptr) { fnErr = fn(fd) }); err != nil {
+		return err
+	}
+	return fnErr
+}
+
 func (m *livenessMonitor) Unsubscribe(id string) (err error) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
@@ -170,14 +191,18 @@ func (m *livenessMonitor) unsubscribe(id string) (err error) {
 	}
 
 	// No longer wait for event, delete it from interest list.
-	if err = rawConn.Control(func(fd uintptr) {
+	if err = controlFD(rawConn, func(fd FD) error {
+		// The set entry must go even if EPOLL_CTL_DEL fails: the connection is
+		// closed below, which removes the fd from the interest list anyway, and
+		// a stale entry would collide with a future subscription reusing the
+		// same fd number.
+		delete(m.set, fd)
 		if err := unix.EpollCtl(m.epollFd, unix.EPOLL_CTL_DEL, int(fd), &unix.EpollEvent{}); err != nil {
-			log.L.Errorf("Fail to delete event fd %d for supervisor %s", int(fd), id)
-			return
+			return errors.Wrapf(err, "delete event fd %d for supervisor %s", int(fd), id)
 		}
-		delete(m.set, fd)
+		return nil
 	}); err != nil {
-		return errors.Wrapf(err, "remove target FD in the interested list, id=%s", id)
+		log.L.WithError(err).Warnf("remove target FD from the interest list, id=%s", id)
 	}
 
 	if err = target.uc.Close(); err != nil {

pkg/manager/monitor_test.go

@@ -15,6 +15,7 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
+	"golang.org/x/sys/unix"
 )
 
 func startUnixServer(ctx context.Context, sock string) {
@@ -44,6 +45,36 @@ func startUnixServer(ctx context.Context, sock string) {
 
 }
 
+func TestLivenessMonitorSubscribeReportsArmFailure(t *testing.T) {
+	sock, err := os.CreateTemp("", "liveness_monitor_sock")
+	assert.Nil(t, err)
+	sock.Close()
+	t.Cleanup(func() { os.Remove(sock.Name()) })
+
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	go startUnixServer(ctx, sock.Name())
+
+	monitor, err := newMonitor()
+	assert.Nil(t, err)
+	assert.NotNil(t, monitor)
+	time.Sleep(time.Millisecond * 200)
+
+	// Sabotage the epoll fd so that arming the subscription (EPOLL_CTL_ADD)
+	// fails. Subscribe must report the failure instead of pretending the
+	// daemon is monitored: an unmonitored daemon would never produce a death
+	// event, so its recovery would never start.
+	assert.Nil(t, unix.Close(monitor.epollFd))
+
+	notifier := make(chan deathEvent, 1)
+	err = monitor.Subscribe("daemon_1", sock.Name(), notifier)
+	assert.NotNil(t, err)
+
+	// All-or-nothing: no half-registered state may remain.
+	assert.Equal(t, 0, len(monitor.subscribers))
+	assert.Equal(t, 0, len(monitor.set))
+}
+
 func TestLivenessMonitor(t *testing.T) {
 	sockPattern := "liveness_monitor_sock"