From 1299ea56eebe980241e4133cbb0891cd8b45735e Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Fri, 2 Oct 2026 01:51:20 +0300 Subject: [PATCH 1/3] accept: lock-free whitelist + throttled read-deadline refresh IsWhitelisted took the same mutex as the token buckets and re-parsed the CIDR list on every call; the live mutex profile showed it as the #2 contention site (~3.5k s of wait per 5 min). The whitelist is written rarely (operator) and read on every connection, so publish it copy-on-write behind an atomic.Pointer and drop the lock. The JSON and binary connection loops also re-armed a 5-minute read deadline on every message, one runtime timer modification per request (~43k/s under load). Re-arm at most once a minute; the deadline keeps at least 4 minutes of headroom, so idle-connection semantics are unchanged. --- accept/accept.go | 65 +++++++++++++++++++++++++++++++++--------------- 1 file changed, 45 insertions(+), 20 deletions(-) diff --git a/accept/accept.go b/accept/accept.go index f6703f0..d81954d 100644 --- a/accept/accept.go +++ b/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 { From f1910a9a0e00af9386b7afef527f973fd18732b3 Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Fri, 2 Oct 2026 01:51:20 +0300 Subject: [PATCH 2/3] cmd/rendezvous: handle slog records asynchronously The default slog handler formats and writes under its own mutex on the caller goroutine; the live mutex profile showed it as a top contention site (~1.1k s of wait per 5 min). Wrap the configured handler in a buffered async handler so hot request paths only enqueue, with formatting and writing on a dedicated goroutine. Overflow drops are counted instead of blocking callers. The dynamic log-level endpoint reinstalls the wrapper so it survives level changes. --- cmd/rendezvous/asynclog.go | 75 ++++++++++++++++++++++++++++++++++++++ cmd/rendezvous/main.go | 5 +-- 2 files changed, 77 insertions(+), 3 deletions(-) create mode 100644 cmd/rendezvous/asynclog.go diff --git a/cmd/rendezvous/asynclog.go b/cmd/rendezvous/asynclog.go new file mode 100644 index 0000000..fecb4ad --- /dev/null +++ b/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()))) +} diff --git a/cmd/rendezvous/main.go b/cmd/rendezvous/main.go index cebdf96..9b77c28 100644 --- a/cmd/rendezvous/main.go +++ b/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 From 703a13fab7c9c3f208c9758509ee2ed71dd5133a Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Fri, 2 Oct 2026 01:51:20 +0300 Subject: [PATCH 3/3] directory: drop the lock before signalling save in HandleSetHostname The handler held the global write lock (and the node shard) across st.cb.Save(). Save is a non-blocking trigger and reads state under its own locks when it runs, so moving it after the unlock shortens the critical section without changing ordering. --- directory/directory.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/directory/directory.go b/directory/directory.go index 2cb8288..797b8e5 100644 --- a/directory/directory.go +++ b/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)