Cut send-message latency 50-170x: hostname cache + fetchMembers backoff

Symptom: pilotctl send-message took 10-30 seconds per call against a
warm tunnel — actual relay RTT is ~170ms. Per-step tracing showed:

  TRACE parseAddrOrHostname  6879ms  ← shared regConn mutex contention
  TRACE dataexchange.Dial     ~200ms
  TRACE client.Recv           ~200ms

The PolicyRunner's reconcileMembership ticker fires every 5s and calls
ListNodes on the regConn. For a network with thousands of members
(network 9 = ~10k+ peers in this fleet), the response exceeds the
client read window and EOFs mid-stream. The reconnect cycle pounds the
regConn lock continuously, so any other call (resolve_hostname,
lookup, etc.) that needs regConn blocks until the next failure cycle
completes.

Two complementary fixes:

1. Daemon-side hostname cache (pkg/daemon/ipc.go handleResolveHostname):
   60s positive cache keyed on the literal hostname string. Repeated
   send-message calls to the same agent serve from cache without
   touching regConn at all. Negative results are NOT cached — transient
   registry errors don't poison lookups.

2. Exponential backoff on fetchMembers (pkg/daemon/policy_runner.go):
   On EOF/error, skip the next ticks for 5s → 10s → 20s → 40s → 80s,
   capped at 5 minutes. Reset to 5s on first success. Logs only on
   power-of-two failure transitions to avoid spam. Stops the regConn
   lock from being held continuously by a doomed call.

Bonus: pilotctl send-message gains an opt-in PILOTCTL_TRACE_TIME=1 env
flag for latency debugging — prints per-step timings to stderr.

Measured: 6879ms hostname → 0.2ms cached. Cold call 1.6s → warm call
175ms. Total send-message reduced from 10-30s to ~600ms cold / ~170ms
warm.

Author: teovl

cmd/pilotctl/main.go

@@ -2848,10 +2848,23 @@ func cmdSendMessage(args []string) {
 		fatalCode("invalid_argument", "usage: pilotctl send-message <address|hostname> --data <text> [--type text|json|binary]")
 	}
 
+	// Optional latency tracing: set PILOTCTL_TRACE_TIME=1 to dump per-step
+	// timings to stderr. Useful when debugging slow agents — distinguishes
+	// hostname-resolve, dial, and recv contributions.
+	traceTime := os.Getenv("PILOTCTL_TRACE_TIME") != ""
+	t0 := time.Now()
+	tracef := func(label string) {
+		if traceTime {
+			fmt.Fprintf(os.Stderr, "TRACE %-22s %12.3fms\n", label, float64(time.Since(t0).Microseconds())/1000.0)
+		}
+	}
+
 	d := connectDriver()
+	tracef("connectDriver")
 	defer d.Close()
 
 	target, err := parseAddrOrHostname(d, pos[0])
+	tracef("parseAddrOrHostname")
 	if err != nil {
 		fatalCode("not_found", "%v", err)
 	}
@@ -2863,6 +2876,7 @@ func cmdSendMessage(args []string) {
 	msgType := flagString(flags, "type", "text")
 
 	client, err := dataexchange.Dial(d, target)
+	tracef("dataexchange.Dial")
 	if err != nil {
 		hint := classifyDaemonError(err)
 		if hint == "" {
@@ -2883,12 +2897,14 @@ func cmdSendMessage(args []string) {
 	default:
 		fatalCode("invalid_argument", "unknown type %q (use text, json, or binary)", msgType)
 	}
+	tracef("client.Send")
 	if err != nil {
 		fatalCode("connection_failed", "send: %v", err)
 	}
 
 	// Read ACK
 	ack, err := client.Recv()
+	tracef("client.Recv")
 	if err != nil {
 		slog.Debug("send-message ACK read failed", "err", err)
 	}
@@ -2902,6 +2918,7 @@ func cmdSendMessage(args []string) {
 		result["ack"] = string(ack.Payload)
 	}
 	outputOK(result)
+	tracef("outputOK")
 }
 
 // ===================== TASK SUBCOMMANDS =====================

pkg/daemon/daemon.go

@@ -198,6 +198,18 @@ type resolveEntry struct {
 	cachedAt time.Time
 }
 
+// hostnameCacheEntry caches a resolve_hostname response so repeated
+// send-message calls to the same peer don't pound the registry.
+type hostnameCacheEntry struct {
+	resp     map[string]interface{}
+	cachedAt time.Time
+}
+
+// hostnameCacheTTL is how long a cached hostname resolution stays
+// valid. Short enough that legitimate hostname rotations propagate
+// within a minute; long enough to absorb rapid-fire CLI invocations.
+const hostnameCacheTTL = 60 * time.Second
+
 type Daemon struct {
 	config          Config
 	addrMu          sync.RWMutex // protects nodeID, addr, publicEndpoint (H6 fix)
@@ -234,6 +246,14 @@ type Daemon struct {
 	resolveCacheMu sync.RWMutex
 	resolveCache   map[uint32]*resolveEntry
 
+	// Hostname cache: hostname string -> cached resolve_hostname response.
+	// Avoids hitting the registry for every send-message to the same peer
+	// when the regConn is contended (policy fetchMembers loops, etc.).
+	// 60s TTL — hostnames do change (rotation, reassignment) but rarely
+	// fast enough for sub-minute staleness to matter.
+	hostnameCacheMu sync.RWMutex
+	hostnameCache   map[string]*hostnameCacheEntry
+
 	// SYN rate limiter (token bucket)
 	synMu       sync.Mutex
 	synTokens   int
@@ -329,6 +349,7 @@ func New(cfg Config) *Daemon {
 		perSrcSYN:     make(map[uint32]*srcSYNBucket),
 		epCache:       make(map[uint32]*endpointEntry),
 		resolveCache:  make(map[uint32]*resolveEntry),
+		hostnameCache: make(map[string]*hostnameCacheEntry),
 		netPolicies:   make(map[uint16][]uint16),
 		managed:       make(map[uint16]*ManagedEngine),
 		policyRunners: make(map[uint16]*PolicyRunner),

pkg/daemon/ipc.go

@@ -820,12 +820,45 @@ func (s *IPCServer) handleResolveHostname(conn *ipcConn, payload []byte) {
 		return
 	}
 
+	// Try the daemon's hostname cache first. The regConn mutex is shared
+	// with the policy_runner's fetchMembers loop; on networks that trip
+	// the list_nodes EOF cycle the lookup can block for seconds. Most
+	// CLI invocations resolve the same hostname repeatedly (a chained
+	// agent script doing dozens of `pilotctl send-message agent X`
+	// calls), so a 60s positive cache turns those calls into IPC-only
+	// round trips.
+	now := time.Now()
+	s.daemon.hostnameCacheMu.RLock()
+	if e, ok := s.daemon.hostnameCache[hostname]; ok && now.Sub(e.cachedAt) < hostnameCacheTTL {
+		cached := e.resp
+		s.daemon.hostnameCacheMu.RUnlock()
+		if data, err := json.Marshal(cached); err == nil {
+			resp := make([]byte, 1+len(data))
+			resp[0] = CmdResolveHostnameOK
+			copy(resp[1:], data)
+			if err := conn.ipcWrite(resp); err != nil {
+				slog.Debug("IPC resolve_hostname (cached) reply failed", "err", err)
+			}
+			return
+		}
+		// fall through on marshal error (shouldn't happen)
+	} else {
+		s.daemon.hostnameCacheMu.RUnlock()
+	}
+
 	result, err := s.daemon.regConn.ResolveHostname(hostname)
 	if err != nil {
 		s.sendError(conn, fmt.Sprintf("resolve_hostname: %v", err))
 		return
 	}
 
+	// Cache the successful response. Only cache positive results so
+	// transient registry hiccups don't poison the cache with a bogus
+	// "not found".
+	s.daemon.hostnameCacheMu.Lock()
+	s.daemon.hostnameCache[hostname] = &hostnameCacheEntry{resp: result, cachedAt: now}
+	s.daemon.hostnameCacheMu.Unlock()
+
 	data, err := json.Marshal(result)
 	if err != nil {
 		s.sendError(conn, fmt.Sprintf("resolve_hostname marshal: %v", err))

pkg/daemon/policy_runner.go

@@ -47,6 +47,16 @@ type PolicyRunner struct {
 	stopCh chan struct{}
 	done   chan struct{}
 	path   string // persistence path (~/.pilot/policy_<netID>.json)
+
+	// fetchMembers backoff. Some networks are too large for the registry's
+	// list_nodes response to fit a single read window — fetchMembers EOFs
+	// every 5s tick and pounds the regConn mutex, which adds 5+ seconds
+	// of latency to ANY other call (resolve_hostname, lookup, etc) that
+	// shares regConn. Track consecutive failures and skip ticks until
+	// the next backoff deadline.
+	fetchFailMu       sync.Mutex
+	fetchFailures     int       // consecutive failure count
+	fetchSkipUntil    time.Time // skip ticks before this time
 }
 
 // policySnapshot is the JSON format persisted to disk.
@@ -1035,11 +1045,47 @@ func (pr *PolicyRunner) fetchMembers() ([]uint32, error) {
 // fetchMembersWithTags returns member IDs and their admin-assigned tags.
 // Also updates the daemon's local member tags cache for the local node.
 func (pr *PolicyRunner) fetchMembersWithTags() []fetchedMember {
+	// Skip if recent failures put us in backoff. Prevents the 5s tick
+	// from re-EOF'ing the regConn and starving co-tenant calls
+	// (resolve_hostname, lookup) of the shared mutex.
+	pr.fetchFailMu.Lock()
+	if !pr.fetchSkipUntil.IsZero() && time.Now().Before(pr.fetchSkipUntil) {
+		pr.fetchFailMu.Unlock()
+		return nil
+	}
+	pr.fetchFailMu.Unlock()
+
 	resp, err := pr.daemon.regConn.ListNodes(pr.netID, pr.daemon.config.AdminToken)
 	if err != nil {
-		slog.Warn("policy: fetchMembers failed", "network_id", pr.netID, "err", err)
+		// Exponential backoff: 5s → 10s → 20s → 40s → 80s, capped at 5min.
+		// Logs only on transitions (1st, 4th, 8th, 16th failure) so the
+		// user sees the issue without log spam.
+		pr.fetchFailMu.Lock()
+		pr.fetchFailures++
+		fails := pr.fetchFailures
+		backoff := time.Duration(1<<min(fails, 6)) * 5 * time.Second
+		if backoff > 5*time.Minute {
+			backoff = 5 * time.Minute
+		}
+		pr.fetchSkipUntil = time.Now().Add(backoff)
+		pr.fetchFailMu.Unlock()
+		// Log on power-of-two transitions to avoid spam.
+		if fails == 1 || fails == 4 || fails == 8 || fails == 16 || fails%32 == 0 {
+			slog.Warn("policy: fetchMembers failed",
+				"network_id", pr.netID, "err", err,
+				"consecutive_failures", fails, "next_attempt_in", backoff.String())
+		}
 		return nil
 	}
+	// Reset failure count on success.
+	pr.fetchFailMu.Lock()
+	if pr.fetchFailures > 0 {
+		slog.Info("policy: fetchMembers recovered", "network_id", pr.netID,
+			"after_failures", pr.fetchFailures)
+		pr.fetchFailures = 0
+		pr.fetchSkipUntil = time.Time{}
+	}
+	pr.fetchFailMu.Unlock()
 
 	nodesRaw, ok := resp["nodes"].([]interface{})
 	if !ok {