Bind sshd, dns and stats listeners in the reload so a bind failure fails it

dns_server.go

@@ -38,8 +38,8 @@ type dnsServer struct {
 
 	serverMu sync.Mutex
 	server   *dns.Server
-	// started is closed once `server` has finished binding (or after
-	// ListenAndServe returns on a bind failure). Stop waits on it before
+	// started is closed once `server` is serving (or after serve returns
+	// without ever starting). Stop waits on it before
 	// calling Shutdown to avoid the miekg/dns "server not started" race
 	// where a Shutdown that arrives before bind completes is silently
 	// ignored, leaving the listener running forever.
@@ -84,7 +84,7 @@ func newDnsServerFromConfig(ctx context.Context, l *slog.Logger, pki *PKI, hostM
 }
 
 // reload applies the latest config and reconciles the running state with it:
-//   - enabled toggled on  -> spawn a runner
+//   - enabled toggled on  -> bind, then serve in the background
 //   - enabled toggled off -> stop the runner
 //   - listen address changed (while running) -> restart on the new address
 //   - everything else     -> no-op
@@ -121,13 +121,20 @@ func (d *dnsServer) reload(c *config.C, initial bool) error {
 		return nil
 	}
 
-	if !running {
-		// Was disabled (or never started); bring it up now.
-		go d.Start()
-	} else if !sameAddr {
-		// Stop clears the slot before shutting down, otherwise the Start below can find the dying server and refuse
+	if running && !sameAddr {
+		// Stop clears the slot before shutting down, otherwise the listen below can find the dying server and refuse
 		d.Stop()
-		go d.Start()
+		running = false
+	}
+	if !running {
+		// Bind here rather than in the background, so a reload that can't listen says so
+		server, started, err := d.listen()
+		if err != nil {
+			return err
+		}
+		if server != nil {
+			go d.serve(server, started)
+		}
 	}
 
 	// Refresh the self entry every enabled reload so cert renewals that change our name or VPN addresses are picked up.
@@ -156,30 +163,50 @@ func (d *dnsServer) shutdownServer(srv *dns.Server, started chan struct{}, reaso
 // Must be invoked after the tun device is active so that lighthouse.dns.host
 // may bind to a nebula IP.
 func (d *dnsServer) Start() {
-	if !d.enabled.Load() {
+	server, started, err := d.listen()
+	if err != nil {
+		d.l.Warn("Failed to run the DNS responder", "error", err)
 		return
 	}
+	if server != nil {
+		d.serve(server, started)
+	}
+}
+
+// listen binds the responder and claims the server slot for serve. It returns a nil server, and no error, when DNS is
+// disabled or a server is already running.
+func (d *dnsServer) listen() (*dns.Server, chan struct{}, error) {
+	if !d.enabled.Load() {
+		return nil, nil, nil
+	}
 
 	started := make(chan struct{})
 	d.serverMu.Lock()
+	defer d.serverMu.Unlock()
 	// Re-check enabled under the lock, a disable that raced our check above snapshots the slot under it too.
-	// Two reloads in quick succession can both spawn a Start, the loser would orphan the live listener past Stop
+	// A reload and Control.Start can both get here, the loser would orphan the live listener past Stop
 	if d.ctx.Err() != nil || d.server != nil || !d.enabled.Load() {
-		d.serverMu.Unlock()
-		return
+		return nil, nil, nil
+	}
+	pc, err := net.ListenPacket("udp", d.addr)
+	if err != nil {
+		return nil, nil, err
 	}
-	addr := d.addr
 	server := &dns.Server{
-		Addr:              addr,
-		Net:               "udp",
+		Addr:              d.addr,
+		PacketConn:        pc,
 		Handler:           d.mux,
 		NotifyStartedFunc: func() { close(started) },
 	}
 	d.server = server
 	d.started = started
-	d.serverMu.Unlock()
+	d.l.Info("Starting DNS responder", "dnsListener", pc.LocalAddr())
+	return server, started, nil
+}
 
-	// Per-invocation ctx watcher. Exits when Start does, so we don't leak a
+// serve answers queries on a server from listen until Stop or the context ends it.
+func (d *dnsServer) serve(server *dns.Server, started chan struct{}) {
+	// Per-invocation ctx watcher. Exits when serve does, so we don't leak a
 	// watcher per reload-driven restart.
 	done := make(chan struct{})
 	go func() {
@@ -190,16 +217,16 @@ func (d *dnsServer) Start() {
 		}
 	}()
 
-	d.l.Info("Starting DNS responder", "dnsListener", addr)
-	err := server.ListenAndServe()
+	err := server.ActivateAndServe()
 	close(done)
 
-	// If the listener never bound (bind error) NotifyStartedFunc never fires,
-	// so close started here to release any Stop caller waiting on it.
+	// If serving failed before it started NotifyStartedFunc never fires, so close started here to release any Stop
+	// caller waiting on it. ActivateAndServe doesn't close the socket on that path either
 	select {
 	case <-started:
 	default:
 		close(started)
+		_ = server.PacketConn.Close()
 	}
 
 	// Release our slot, unless a reload already replaced us, so a dead listener can't block a future Start

dns_server_test.go

@@ -629,3 +629,22 @@ func TestDnsServer_Start_refusesWhenDisabledUnderLock(t *testing.T) {
 	require.NoError(t, err, "an orphaned listener is still holding the port")
 	_ = pc.Close()
 }
+
+// A reload that can't bind says so, rather than finding out in the background
+func TestDnsServer_reload_busyPortFails(t *testing.T) {
+	busy, err := net.ListenPacket("udp", "127.0.0.1:0")
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = busy.Close() })
+	_, port, err := net.SplitHostPort(busy.LocalAddr().String())
+	require.NoError(t, err)
+
+	ds, c := newTestDnsServer(t)
+	setDnsConfig(c, "127.0.0.1", port, false, true)
+	require.NoError(t, ds.reload(c, true))
+
+	setDnsConfig(c, "127.0.0.1", port, true, true)
+	require.Error(t, ds.reload(c, false))
+	ds.serverMu.Lock()
+	assert.Nil(t, ds.server, "a failed bind kept the slot")
+	ds.serverMu.Unlock()
+}

ssh.go

@@ -30,15 +30,20 @@ func wireSSHReload(l *slog.Logger, ssh *sshd.SSHServer, c *config.C) {
 				}
 				return
 			}
-			sshRun, err := configSSH(l, ssh, c, r)
-			if err != nil {
+			if _, err := configSSH(l, ssh, c, r); err != nil {
 				l.Error("Failed to reconfigure the sshd", "error", err)
 				ssh.Stop()
 				r.Failed()
+				return
 			}
-			if sshRun != nil {
-				go sshRun()
+			// Bind here rather than in the background, so a reload that can't listen says so
+			listener, err := ssh.Listen(c.GetString("sshd.listen", ""))
+			if err != nil {
+				l.Error("Failed to run the SSH server", "error", err)
+				r.Failed()
+				return
 			}
+			go ssh.Serve(listener)
 		} else {
 			ssh.Stop()
 		}

sshd/server.go

@@ -184,19 +184,34 @@ func (s *SSHServer) AddAuthorizedKey(user, pubKey string) error {
 // from the constructor-supplied ctx so a Stop+Run sequence (used by config reload) starts clean
 // rather than carrying a permanently-cancelled context across runs.
 func (s *SSHServer) Run(addr string) error {
-	if s.ctx.Err() != nil {
-		return s.ctx.Err()
-	}
-
-	listener, err := net.Listen("tcp", addr)
+	listener, err := s.Listen(addr)
 	if err != nil {
 		return err
 	}
+	s.Serve(listener)
+	return nil
+}
+
+// Listen binds addr for Serve, so a caller can tell the bind failed before serving in the background. Stop can close
+// the listener from here on, Serve then returns as soon as it starts.
+func (s *SSHServer) Listen(addr string) (net.Listener, error) {
+	if s.ctx.Err() != nil {
+		return nil, s.ctx.Err()
+	}
+	listener, err := net.Listen("tcp", addr)
+	if err != nil {
+		return nil, err
+	}
 	// s.listener is the public handle Stop uses to interrupt the active run; listener (the local) is what
 	// this run owns. They start equal but a fast reload may overwrite s.listener with the next run's
 	// listener before this run's watcher fires, so each run must close its own listener via the local
 	// reference.
 	s.listener.Store(&listener)
+	return listener, nil
+}
+
+// Serve accepts connections on a listener from Listen until Stop, or the context passed to NewSSHServer, ends it.
+func (s *SSHServer) Serve(listener net.Listener) {
 	s.runs.Add(1)
 	bound := listener.Addr()
 	s.addr.Store(&bound)
@@ -224,8 +239,6 @@ func (s *SSHServer) Run(addr string) error {
 	s.run(runCtx, listener)
 
 	s.l.Info("SSH server stopped listening")
-	// We don't return an error because run logs for us
-	return nil
 }
 
 func (s *SSHServer) run(ctx context.Context, listener net.Listener) {

sshd_reload_test.go

@@ -279,3 +279,15 @@ func TestSSHExecExitStatusReflectsTheCommand(t *testing.T) {
 	assert.Equal(t, 1, exit.ExitStatus())
 	assert.True(t, strings.HasPrefix(string(out), "did not understand: nope\nAvailable commands:\n"), string(out))
 }
+
+// A reload that can't bind the new address says so, rather than finding out in the background
+func TestSSHDReload_BusyPortFails(t *testing.T) {
+	h := newSSHDReloadHarness(t)
+	busy, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = busy.Close() })
+
+	res := h.c.ReloadConfigString(strings.Replace(h.config(), "127.0.0.1:0", busy.Addr().String(), 1))
+	require.NoError(t, res.Err)
+	assert.True(t, res.Failed)
+}

stats.go

@@ -140,8 +140,13 @@ func (s *statsServer) reload(c *config.C, initial bool) error {
 
 	// Even with nothing running, a runtime whose listener failed can still be finishing its last pass
 	s.Stop()
-	if enabled && !s.configTest {
-		go s.Start()
+	// Bind here rather than in the background, so a reload that can't listen says so
+	run, err := s.start()
+	if err != nil {
+		return err
+	}
+	if run != nil {
+		go run()
 	}
 	return nil
 }
@@ -151,36 +156,64 @@ func (s *statsServer) reload(c *config.C, initial bool) error {
 // the HTTP listener. For graphite it blocks on the capture loop's context.
 // Safe to call when stats are disabled or already running (both no-op).
 func (s *statsServer) Start() {
-	if !s.enabled.Load() || s.configTest {
+	run, err := s.start()
+	if err != nil {
+		s.l.Error("Failed to start the prometheus stats listener", "error", err)
 		return
 	}
+	if run != nil {
+		run()
+	}
+}
+
+// start claims the runtime and binds the prometheus listener. The returned run serves it and blocks until Stop or ctx
+// ends it. A nil run and no error means stats are off or already running.
+func (s *statsServer) start() (func(), error) {
+	if !s.enabled.Load() || s.configTest {
+		return nil, nil
+	}
 
 	s.runMu.Lock()
+	defer s.runMu.Unlock()
 	if s.ctx.Err() != nil || s.run != nil || s.runCfg == nil {
-		s.runMu.Unlock()
-		return
+		return nil, nil
 	}
 	cfg := *s.runCfg
 	runCtx, cancel := context.WithCancel(s.ctx)
 	captureFns, listener := s.buildRuntime(runCtx, cfg)
+	var ln net.Listener
+	if listener != nil {
+		var err error
+		ln, err = net.Listen("tcp", listener.Addr)
+		if err != nil {
+			cancel()
+			// Drop the config so a SIGHUP with the same one tries again once the problem is fixed
+			s.runCfg = nil
+			return nil, err
+		}
+	}
 	rt := &statsRuntime{cancel: cancel, listener: listener}
 	s.run = rt
 	done := make(chan struct{})
 	s.loopDone = done
-	s.runMu.Unlock()
 
+	return func() { s.serve(runCtx, rt, captureFns, cfg.interval, ln, done) }, nil
+}
+
+// serve runs the capture loop, and the prometheus listener on ln, until Stop or ctx ends them.
+func (s *statsServer) serve(runCtx context.Context, rt *statsRuntime, captureFns []func(), interval time.Duration, ln net.Listener, done chan struct{}) {
 	// Prime before serving, so a scrape that lands before the first tick sees real values instead of zeros (issue #907)
 	for _, fn := range captureFns {
 		fn()
 	}
-	go captureStatsLoop(runCtx, cfg.interval, captureFns, done)
+	go captureStatsLoop(runCtx, interval, captureFns, done)
 
 	cleanExit := true
-	if listener == nil {
+	if rt.listener == nil {
 		// Graphite: no HTTP listener to serve; block until teardown.
 		<-runCtx.Done()
 	} else {
-		cleanExit = s.serveListener(listener)
+		cleanExit = s.serveListener(rt.listener, ln)
 	}
 
 	// Clear our runtime only if nothing has replaced it. Stop races through
@@ -189,10 +222,9 @@ func (s *statsServer) Start() {
 	if s.run == rt {
 		rt.cancel()
 		s.run = nil
-		// A listener that exited with an error (e.g., bind conflict) leaves
-		// runCfg cached as if it were applied. Drop it so a SIGHUP with the
-		// same config re-triggers Start once the user fixes the underlying
-		// problem.
+		// A listener that exited with an error leaves runCfg cached as if it
+		// were applied. Drop it so a SIGHUP with the same config re-triggers
+		// Start once the user fixes the underlying problem.
 		if !cleanExit {
 			s.runCfg = nil
 		}
@@ -200,12 +232,12 @@ func (s *statsServer) Start() {
 	s.runMu.Unlock()
 }
 
-// serveListener runs ListenAndServe and ensures ctx cancellation unblocks it.
+// serveListener serves listener on ln and ensures ctx cancellation unblocks it.
 // Returns true if the listener exited cleanly (Stop, ctx cancellation, or any
 // other http.ErrServerClosed path), false on an unexpected error.
-func (s *statsServer) serveListener(listener *http.Server) bool {
+func (s *statsServer) serveListener(listener *http.Server, ln net.Listener) bool {
 	// Per-invocation watcher: ctx cancellation triggers a listener shutdown
-	// which in turn unblocks ListenAndServe. Closing `done` on exit keeps
+	// which in turn unblocks Serve. Closing `done` on exit keeps
 	// the watcher from outliving this call.
 	done := make(chan struct{})
 	go func() {
@@ -221,8 +253,8 @@ func (s *statsServer) serveListener(listener *http.Server) bool {
 	}()
 	defer close(done)
 
-	s.l.Info("Starting prometheus stats listener", "addr", listener.Addr)
-	err := listener.ListenAndServe()
+	s.l.Info("Starting prometheus stats listener", "addr", ln.Addr())
+	err := listener.Serve(ln)
 	if err == nil || errors.Is(err, http.ErrServerClosed) {
 		return true
 	}

stats_test.go

@@ -673,3 +673,28 @@ func TestStatsServer_Start_duringStopWaitsForThePass(t *testing.T) {
 	waitFor(t, func() bool { return len(entered) == 1 })
 	assert.Equal(t, int64(1), most.Load(), "a Start captured alongside the pass Stop was waiting out")
 }
+
+// A reload that can't bind says so rather than finding out in the background, and the same config tries again once
+// the port is free
+func TestStatsServer_reload_bindFailureFails(t *testing.T) {
+	blocker, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	port := strconv.Itoa(blocker.Addr().(*net.TCPAddr).Port)
+
+	s, c := newTestStatsServer(t)
+	stopAtCleanup(t, &s)
+	require.NoError(t, s.reload(c, true))
+
+	setStatsConfig(c, map[string]any{
+		"type":     "prometheus",
+		"interval": "1s",
+		"listen":   "127.0.0.1:" + port,
+		"path":     "/metrics",
+	})
+	require.Error(t, s.reload(c, false))
+	assert.Nil(t, currentRuntime(s))
+
+	require.NoError(t, blocker.Close())
+	require.NoError(t, s.reload(c, false))
+	waitForListening(t, "127.0.0.1:"+port)
+}