rendezvous: take the remaining hot-path locks off the request path

accept/accept.go

@@ -168,8 +168,11 @@ type RateLimiter struct {
 	rate       int
 	window     time.Duration
 	maxBuckets int
-	whitelist  []whitelistRule
-	now        func() time.Time
+	// whitelist is read on every connection/message (IsWhitelisted) but
+	// written rarely (operator). Copy-on-write via an atomic pointer keeps
+	// the hot path lock-free instead of contending with the bucket mutex.
+	whitelist atomic.Pointer[[]whitelistRule]
+	now       func() time.Time
 }
 
 // WhitelistEntry pairs a CIDR with its elevated per-window rate. Operator-
@@ -259,7 +262,7 @@ func (rl *RateLimiter) Allow(ip string) bool {
 	if !ok {
 		// Whitelist lookup at bucket-creation time: matched IPs get an
 		// elevated per-bucket rate and bypass the maxBuckets cap.
-		bucketRate, whitelisted := rl.whitelistRateLocked(ip)
+		bucketRate, whitelisted := rl.whitelistRate(ip)
 		if !whitelisted {
 			bucketRate = rl.rate
 			if rl.maxBuckets > 0 && len(rl.buckets) >= rl.maxBuckets {
@@ -310,23 +313,23 @@ func (rl *RateLimiter) Allow(ip string) bool {
 // in handleBinaryConn / handleTextConn for operator-trusted internal
 // infrastructure). Safe for concurrent use.
 func (rl *RateLimiter) IsWhitelisted(ip string) bool {
-	rl.mu.Lock()
-	defer rl.mu.Unlock()
-	_, ok := rl.whitelistRateLocked(ip)
+	_, ok := rl.whitelistRate(ip)
 	return ok
 }
 
-// whitelistRateLocked returns the elevated rate for ip if any whitelist
-// CIDR contains it, else (0, false). Caller must hold rl.mu.
-func (rl *RateLimiter) whitelistRateLocked(ip string) (int, bool) {
-	if len(rl.whitelist) == 0 {
+// whitelistRate returns the elevated rate for ip if any whitelist CIDR
+// contains it, else (0, false). Lock-free: the whitelist is immutable once
+// published (copy-on-write), so no mutex is needed on this hot path.
+func (rl *RateLimiter) whitelistRate(ip string) (int, bool) {
+	wl := rl.whitelist.Load()
+	if wl == nil || len(*wl) == 0 {
 		return 0, false
 	}
 	parsed := net.ParseIP(ip)
 	if parsed == nil {
 		return 0, false
 	}
-	for _, w := range rl.whitelist {
+	for _, w := range *wl {
 		if w.net.Contains(parsed) {
 			return w.rate, true
 		}
@@ -373,18 +376,18 @@ func (rl *RateLimiter) SetWhitelist(entries []WhitelistEntry) error {
 		}
 		parsed = append(parsed, whitelistRule{net: ipnet, rate: e.Rate})
 	}
-	rl.mu.Lock()
-	rl.whitelist = parsed
-	rl.mu.Unlock()
+	rl.whitelist.Store(&parsed)
 	return nil
 }
 
 // WhitelistSize returns the number of installed whitelist rules (for testing
 // and operator-facing introspection).
 func (rl *RateLimiter) WhitelistSize() int {
-	rl.mu.Lock()
-	defer rl.mu.Unlock()
-	return len(rl.whitelist)
+	wl := rl.whitelist.Load()
+	if wl == nil {
+		return 0
+	}
+	return len(*wl)
 }
 
 // Cleanup removes stale buckets. Call periodically.
@@ -476,6 +479,15 @@ func SanitizeListenAddr(remoteAddr, clientAddr string) string {
 // rawResponseKey is a sentinel used by writeMessage: when the response map
 // contains this key with a []byte value, writeMessage writes the bytes verbatim
 // (skipping json.Marshal). Mirrors the same constant in the server package.
+// connectionReadTimeout bounds how long a connection may sit idle between
+// messages. readDeadlineRefresh throttles re-arming that deadline so busy
+// connections don't touch the runtime timer on every message; the effective
+// idle window is connectionReadTimeout..connectionReadTimeout+refresh.
+const (
+	connectionReadTimeout = 5 * time.Minute
+	readDeadlineRefresh   = 1 * time.Minute
+)
+
 const rawResponseKey = "_pilot_raw_body"
 
 // writeMessageDeadline bounds how long a single response write can block.
@@ -803,7 +815,7 @@ func (a *Acceptor) handleConn(conn net.Conn) {
 	// Binary clients send magic 0x50494C54 ("PILT"). JSON clients send a
 	// 4-byte big-endian length prefix which is always < 64 KB, while the magic
 	// value (~1.3 billion) is orders of magnitude larger.
-	conn.SetReadDeadline(time.Now().Add(5 * time.Minute))
+	conn.SetReadDeadline(time.Now().Add(connectionReadTimeout))
 	var peek [4]byte
 	if _, err := io.ReadFull(conn, peek[:]); err != nil {
 		if err != io.EOF {
@@ -859,9 +871,18 @@ func (a *Acceptor) replyRateLimited(conn net.Conn, binary bool, scope string) bo
 func (a *Acceptor) handleJSONConn(conn net.Conn, reader io.Reader) {
 	var connReqCount int64
 	connStart := time.Now()
+	deadlineSet := time.Now()
 
 	for {
-		conn.SetReadDeadline(time.Now().Add(5 * time.Minute))
+		// Refresh the idle deadline at most once per readDeadlineRefresh.
+		// Re-arming a 5-minute deadline on every message cost a runtime
+		// timer modification per request (~43k/s at fleet scale); the
+		// deadline always stays at least connectionReadTimeout-refresh in
+		// the future, so idle-connection semantics are unchanged.
+		if time.Since(deadlineSet) >= readDeadlineRefresh {
+			deadlineSet = time.Now()
+			conn.SetReadDeadline(deadlineSet.Add(connectionReadTimeout))
+		}
 		msg, err := readMessage(reader)
 		if err != nil {
 			if err != io.EOF {
@@ -1003,9 +1024,13 @@ func (a *Acceptor) handleBinaryConn(conn net.Conn) {
 	connStart := time.Now()
 	remoteAddr := conn.RemoteAddr().String()
 	host, _, _ := net.SplitHostPort(remoteAddr)
+	deadlineSet := time.Now()
 
 	for {
-		conn.SetReadDeadline(time.Now().Add(5 * time.Minute))
+		if time.Since(deadlineSet) >= readDeadlineRefresh {
+			deadlineSet = time.Now()
+			conn.SetReadDeadline(deadlineSet.Add(connectionReadTimeout))
+		}
 		msgType, payload, err := wire.ReadFrame(conn)
 		if err != nil {
 			if err != io.EOF {

cmd/rendezvous/asynclog.go

@@ -0,0 +1,75 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package main
+
+import (
+	"context"
+	"log/slog"
+	"sync/atomic"
+
+	"github.com/pilot-protocol/common/logging"
+)
+
+// asyncLogBuffer is the queue depth of the async slog handler. At fleet scale
+// the registry emits several thousand records/sec; the default slog handler
+// formats + writes under a mutex on the caller's goroutine, and the live
+// mutex profile showed that as a top contention site. This handler enqueues
+// the record and formats/writes it on a dedicated goroutine instead, so hot
+// request paths never block on the log mutex.
+const asyncLogBuffer = 16384
+
+type asyncLogSink struct {
+	next    slog.Handler
+	ch      chan slog.Record
+	dropped atomic.Uint64
+}
+
+func (s *asyncLogSink) run() {
+	ctx := context.Background()
+	for r := range s.ch {
+		_ = s.next.Handle(ctx, r)
+	}
+}
+
+type asyncHandler struct {
+	sink *asyncLogSink
+	next slog.Handler
+}
+
+func newAsyncHandler(next slog.Handler) slog.Handler {
+	sink := &asyncLogSink{next: next, ch: make(chan slog.Record, asyncLogBuffer)}
+	go sink.run()
+	return &asyncHandler{sink: sink, next: next}
+}
+
+func (h *asyncHandler) Enabled(ctx context.Context, lvl slog.Level) bool {
+	return h.next.Enabled(ctx, lvl)
+}
+
+func (h *asyncHandler) Handle(_ context.Context, r slog.Record) error {
+	select {
+	case h.sink.ch <- r:
+	default:
+		// Queue full (logging can't keep up with a burst): drop rather than
+		// block the caller. The counter is exposed for operators to see.
+		h.sink.dropped.Add(1)
+	}
+	return nil
+}
+
+func (h *asyncHandler) WithAttrs(attrs []slog.Attr) slog.Handler {
+	return &asyncHandler{sink: h.sink, next: h.next.WithAttrs(attrs)}
+}
+
+func (h *asyncHandler) WithGroup(name string) slog.Handler {
+	return &asyncHandler{sink: h.sink, next: h.next.WithGroup(name)}
+}
+
+// setupLogging installs the configured slog handler and wraps it so log
+// formatting/writing happens off the caller's goroutine. Used both at startup
+// and by the dynamic log-level endpoint so the async wrapper survives
+// level changes.
+func setupLogging(level, format string) {
+	logging.Setup(level, format)
+	slog.SetDefault(slog.New(newAsyncHandler(slog.Default().Handler())))
+}

cmd/rendezvous/main.go

@@ -18,7 +18,6 @@ import (
 
 	"github.com/pilot-protocol/beacon"
 	"github.com/pilot-protocol/common/config"
-	"github.com/pilot-protocol/common/logging"
 	registry "github.com/pilot-protocol/rendezvous"
 	"github.com/pilot-protocol/rendezvous/breakers"
 	dashpkgPub "github.com/pilot-protocol/rendezvous/dashboard"
@@ -97,7 +96,7 @@ func main() {
 		config.ApplyToFlags(cfg)
 	}
 
-	logging.Setup(*logLevel, *logFormat)
+	setupLogging(*logLevel, *logFormat)
 
 	// Enable mutex + block profiling early so contention from startup paths
 	// is captured. Net runtime cost at the chosen rates is ~1% CPU; the
@@ -306,7 +305,7 @@ func main() {
 		func(level string) error {
 			switch strings.ToLower(level) {
 			case "debug", "info", "warn", "warning", "error":
-				logging.Setup(level, *logFormat)
+				setupLogging(level, *logFormat)
 				currentLogLevel = strings.ToLower(level)
 				slog.Info("log level changed", "level", currentLogLevel)
 				return nil

directory/directory.go

@@ -1820,10 +1820,13 @@ func (st *Store) HandleSetHostname(msg map[string]interface{}) (map[string]inter
 	if hostname != "" {
 		st.hostnameIdx[hostname] = nodeID
 	}
-	st.cb.Save()
 	sh.Unlock()
 	st.mu.Unlock()
 
+	// Signal the save outside the write lock — Save is a non-blocking
+	// trigger, so holding the global lock across it just adds contention.
+	st.cb.Save()
+
 	slog.Debug("hostname set", "node_id", nodeID, "hostname", hostname)
 	st.cb.Audit("hostname.changed", "node_id", nodeID, "old_hostname", oldHostname, "new_hostname", hostname)