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 { 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 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)