Merge pull request #4911 from mn-ram/fix/ga-sock-reconnect

hostagent: stop destroying ga.sock on guest-agent reconnect

Author: jandubois

pkg/hostagent/ga_sock_forward_test.go

@@ -0,0 +1,141 @@
+// SPDX-FileCopyrightText: Copyright The Lima Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package hostagent
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/lima-vm/sshocker/pkg/ssh"
+	"gotest.tools/v3/assert"
+)
+
+// Regression test for https://github.com/lima-vm/lima/issues/2227.
+//
+// Two invariants must hold for the guest-agent socket forward:
+//  1. forwardGuestAgentSock must issue verbCancel before verbForward on every
+//     call, so the SSH ControlMaster releases the prior registration and the
+//     new bind succeeds cleanly. Otherwise forwardSSH unlinks the local
+//     ga.sock and the socket disappears from disk until limactl stop/start.
+//  2. gaSockForwardMu must serialize the forward path against the cancel
+//     path (cleanup) and against concurrent reconnect ticks. Without
+//     serialization, os.RemoveAll/bind races leave ga.sock missing.
+
+type forwardCall struct {
+	verb  string
+	local string
+}
+
+// stubForwardSSH replaces the package-level forwardSSH with a stub that
+// records calls and optionally runs hook on every call. It returns an
+// accessor for the recorded calls and a restore function.
+func stubForwardSSH(t *testing.T, hook func(verb string)) (calls func() []forwardCall, restore func()) {
+	t.Helper()
+	var mu sync.Mutex
+	var recorded []forwardCall
+	orig := forwardSSH
+	forwardSSH = func(_ context.Context, _ *ssh.SSHConfig, _ string, _ int, local, _, verb string, _ bool) error {
+		if hook != nil {
+			hook(verb)
+		}
+		mu.Lock()
+		recorded = append(recorded, forwardCall{verb: verb, local: local})
+		mu.Unlock()
+		return nil
+	}
+	return func() []forwardCall {
+			mu.Lock()
+			defer mu.Unlock()
+			out := make([]forwardCall, len(recorded))
+			copy(out, recorded)
+			return out
+		}, func() {
+			forwardSSH = orig
+		}
+}
+
+func TestForwardGuestAgentSock_CancelsBeforeForward(t *testing.T) {
+	calls, restore := stubForwardSSH(t, nil)
+	defer restore()
+
+	a := &HostAgent{}
+	a.forwardGuestAgentSock(t.Context(), "/tmp/ga.sock", "/run/lima-guestagent.sock")
+
+	got := calls()
+	assert.Equal(t, len(got), 2, "expected exactly one cancel and one forward")
+	assert.Equal(t, got[0].verb, verbCancel, "cancel must come before forward")
+	assert.Equal(t, got[1].verb, verbForward, "forward must follow cancel")
+}
+
+func TestCancelGuestAgentSockForward_IssuesCancel(t *testing.T) {
+	calls, restore := stubForwardSSH(t, nil)
+	defer restore()
+
+	a := &HostAgent{}
+	err := a.cancelGuestAgentSockForward(t.Context(), "/tmp/ga.sock", "/run/lima-guestagent.sock")
+	assert.NilError(t, err)
+
+	got := calls()
+	assert.Equal(t, len(got), 1)
+	assert.Equal(t, got[0].verb, verbCancel)
+}
+
+// TestGuestAgentSockForward_SerializedUnderContention runs many concurrent
+// forward and cancel callers and asserts that at most one forwardSSH call is
+// in flight at any moment (max observed concurrency == 1). If
+// gaSockForwardMu were missing or scoped incorrectly, two goroutines could
+// enter forwardSSH simultaneously and race on the local ga.sock listener.
+func TestGuestAgentSockForward_SerializedUnderContention(t *testing.T) {
+	const (
+		forwarders = 8
+		cancelers  = 8
+		iterations = 50
+	)
+
+	var (
+		mu        sync.Mutex
+		active    int
+		maxActive int
+	)
+	hook := func(_ string) {
+		mu.Lock()
+		active++
+		if active > maxActive {
+			maxActive = active
+		}
+		mu.Unlock()
+		// Widen the window so any missing serialization can manifest.
+		time.Sleep(50 * time.Microsecond)
+		mu.Lock()
+		active--
+		mu.Unlock()
+	}
+
+	_, restore := stubForwardSSH(t, hook)
+	defer restore()
+
+	a := &HostAgent{}
+	var wg sync.WaitGroup
+	for range forwarders {
+		wg.Go(func() {
+			for range iterations {
+				a.forwardGuestAgentSock(t.Context(), "/tmp/ga.sock", "/run/lima-guestagent.sock")
+			}
+		})
+	}
+	for range cancelers {
+		wg.Go(func() {
+			for range iterations {
+				_ = a.cancelGuestAgentSockForward(t.Context(), "/tmp/ga.sock", "/run/lima-guestagent.sock")
+			}
+		})
+	}
+	wg.Wait()
+
+	mu.Lock()
+	defer mu.Unlock()
+	assert.Equal(t, maxActive, 1, "forwardSSH calls overlapped — gaSockForwardMu is not serializing the ga.sock forward path")
+}

pkg/hostagent/hostagent.go

@@ -78,6 +78,12 @@ type HostAgent struct {
 	clientMu sync.RWMutex
 	client   *guestagentclient.GuestAgentClient
 
+	// gaSockForwardMu serializes (re-)establishment of the SSH forward for
+	// the guest-agent unix socket. The reconnect loop in watchGuestAgentEvents
+	// and the inotify goroutine both touch the same local socket path; without
+	// this lock they can race on os.RemoveAll/bind and leave ga.sock missing.
+	gaSockForwardMu sync.Mutex
+
 	guestAgentAliveCh     chan struct{} // closed on establishing the connection
 	guestAgentAliveChOnce sync.Once
 
@@ -719,7 +725,7 @@ func (a *HostAgent) watchGuestAgentEvents(ctx context.Context) {
 			}
 		}
 		if a.driver.ForwardGuestAgent(ctx) {
-			if err := forwardSSH(ctx, a.sshConfig, sshAddress, sshPort, localUnix, remoteUnix, verbCancel, false); err != nil {
+			if err := a.cancelGuestAgentSockForward(ctx, localUnix, remoteUnix); err != nil {
 				errs = append(errs, err)
 			}
 		}
@@ -733,8 +739,7 @@ func (a *HostAgent) watchGuestAgentEvents(ctx context.Context) {
 		client := a.getClient()
 		if client == nil || !isGuestAgentSocketAccessible(ctx, client) {
 			if a.driver.ForwardGuestAgent(ctx) {
-				sshAddress, sshPort := a.sshAddressPort()
-				_ = forwardSSH(ctx, a.sshConfig, sshAddress, sshPort, localUnix, remoteUnix, verbForward, false)
+				a.forwardGuestAgentSock(ctx, localUnix, remoteUnix)
 			}
 		}
 		// Re-spawn startInotify when its gRPC stream dies (typically because
@@ -769,8 +774,7 @@ func (a *HostAgent) watchGuestAgentEvents(ctx context.Context) {
 		client := a.getClient()
 		if client == nil || !isGuestAgentSocketAccessible(ctx, client) {
 			if a.driver.ForwardGuestAgent(ctx) {
-				sshAddress, sshPort := a.sshAddressPort()
-				_ = forwardSSH(ctx, a.sshConfig, sshAddress, sshPort, localUnix, remoteUnix, verbForward, false)
+				a.forwardGuestAgentSock(ctx, localUnix, remoteUnix)
 			}
 		}
 		client, err := a.getOrCreateClient(ctx)
@@ -949,7 +953,52 @@ func executeSSH(ctx context.Context, sshConfig *ssh.SSHConfig, sshAddress string
 	return nil
 }
 
-func forwardSSH(ctx context.Context, sshConfig *ssh.SSHConfig, sshAddress string, sshPort int, local, remote, verb string, reverse bool) error {
+// forwardGuestAgentSock establishes (or re-establishes) the SSH local forward
+// of the guest-agent unix socket. It is used both for the initial setup and
+// to bring the forward back up after the guest agent has been restarted, the
+// VM has been rebooted, or the gRPC stream has otherwise become unhealthy.
+//
+// The previous behavior was to call forwardSSH(verbForward) directly on every
+// reconnect tick. forwardSSH unlinks the local socket file as its first step
+// (so a fresh listener can bind), and the SSH ControlMaster still has the
+// previous forward registered for the same listen path. The duplicate
+// registration causes ssh -O forward to exit non-zero and forwardSSH to unlink
+// the socket a second time on its failure branch — leaving ga.sock permanently
+// missing on disk and breaking host↔guest gRPC, dynamic port forwarding, and
+// inotify mount invalidation until limactl stop && limactl start. See #2227.
+//
+// The fix is twofold:
+//  1. Best-effort verbCancel before verbForward, so the ControlMaster releases
+//     the prior registration and the new bind succeeds cleanly.
+//  2. Serialize via gaSockForwardMu, so the reconnect loop in
+//     watchGuestAgentEvents, the inotify setup goroutine, and the cleanup
+//     path cannot race on os.RemoveAll/bind of the same path.
+func (a *HostAgent) forwardGuestAgentSock(ctx context.Context, localUnix, remoteUnix string) {
+	a.gaSockForwardMu.Lock()
+	defer a.gaSockForwardMu.Unlock()
+	sshAddress, sshPort := a.sshAddressPort()
+	// Best-effort teardown of any prior forward registered with the
+	// ControlMaster. Errors are expected (e.g. on the very first call when
+	// no forward exists yet) and intentionally ignored. Use ctx so shutdown
+	// can unblock this call if the ControlMaster is unresponsive.
+	_ = forwardSSH(ctx, a.sshConfig, sshAddress, sshPort, localUnix, remoteUnix, verbCancel, false)
+	if err := forwardSSH(ctx, a.sshConfig, sshAddress, sshPort, localUnix, remoteUnix, verbForward, false); err != nil {
+		logrus.WithError(err).Warn("failed to (re-)establish forward for the guest agent socket; will retry")
+	}
+}
+
+// cancelGuestAgentSockForward tears down the SSH forward for the guest-agent
+// unix socket. Serialized via gaSockForwardMu so it cannot race with the
+// reconnect path in forwardGuestAgentSock.
+func (a *HostAgent) cancelGuestAgentSockForward(ctx context.Context, localUnix, remoteUnix string) error {
+	a.gaSockForwardMu.Lock()
+	defer a.gaSockForwardMu.Unlock()
+	sshAddress, sshPort := a.sshAddressPort()
+	return forwardSSH(ctx, a.sshConfig, sshAddress, sshPort, localUnix, remoteUnix, verbCancel, false)
+}
+
+// forwardSSH is a var (not a func) so tests can stub it without touching real ssh.
+var forwardSSH = func(ctx context.Context, sshConfig *ssh.SSHConfig, sshAddress string, sshPort int, local, remote, verb string, reverse bool) error {
 	args := sshConfig.Args()
 	args = append(args,
 		"-T",