registry: tell the beacon which node a relayed handshake was parked for

CHANGELOG.md

@@ -9,6 +9,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Relayed trust handshakes no longer wait for the recipient's next poll. When
+  `request_handshake` or `respond_handshake` parks a message for a node, the
+  registry tells its co-located beacon (`SetHandshakeNotifier`, wired in
+  `cmd/rendezvous` when the beacon has `NotifyNode`), which sends that node a
+  two-byte notify; an updated daemon then polls at once. The notify carries
+  nothing about the handshake or the other node. Older daemons ignore it and
+  poll once a minute as before.
 - `resolve_hostname_ok` carries `last_seen_unix`, the same field `lookup_ok`
   reports, so a caller can tell a hostname held by a node that has stopped
   heartbeating from a live one. Additive; hostname claimability and

cmd/rendezvous/main.go

@@ -127,6 +127,12 @@ func main() {
 	r.SetBuildInfo(buildInfo)
 	r.SetStaleNodeThreshold(*staleThreshold)
 	r.SetBeaconStats(b)
+	// Relayed trust handshakes: have the beacon prompt the recipient to
+	// poll. Asserted rather than called so this builds against a beacon
+	// release that predates NotifyNode (the hook is then simply not wired).
+	if n, ok := interface{}(b).(interface{ NotifyNode(uint32) error }); ok {
+		r.SetHandshakeNotifier(func(nodeID uint32) { _ = n.NotifyNode(nodeID) })
+	}
 	b.SetAuthoritativeKeyLookup(func(nodeID uint32) (ed25519.PublicKey, bool) {
 		raw, ok := r.LookupPublicKey(nodeID)
 		if !ok {

dispatch.go

@@ -123,13 +123,21 @@ var handlers = map[string]Handler{
 		return s.trust.HandleCheckTrust(msg)
 	},
 	"request_handshake": func(s *Server, msg map[string]interface{}, _ string) (map[string]interface{}, error) {
-		return s.trust.HandleRequestHandshake(msg)
+		resp, err := s.trust.HandleRequestHandshake(msg)
+		if err == nil {
+			s.notifyHandshake(jsonUint32(msg, "to_node_id"))
+		}
+		return resp, err
 	},
 	"poll_handshakes": func(s *Server, msg map[string]interface{}, _ string) (map[string]interface{}, error) {
 		return s.trust.HandlePollHandshakes(msg)
 	},
 	"respond_handshake": func(s *Server, msg map[string]interface{}, _ string) (map[string]interface{}, error) {
-		return s.trust.HandleRespondHandshake(msg)
+		resp, err := s.trust.HandleRespondHandshake(msg)
+		if err == nil {
+			s.notifyHandshake(jsonUint32(msg, "peer_id"))
+		}
+		return resp, err
 	},
 	"heartbeat": func(s *Server, msg map[string]interface{}, _ string) (map[string]interface{}, error) {
 		return s.directory.HandleHeartbeat(msg)

server.go

@@ -186,6 +186,11 @@ type Server struct {
 	// without coupling the registry directly to pkg/beacon.
 	beaconStats BeaconStatsProvider
 
+	// handshakeNotifier, when set by the host (cmd/rendezvous), is told
+	// which node a relayed handshake request or answer was just parked
+	// for, so the co-located beacon can prompt that node to poll now.
+	handshakeNotifier atomic.Pointer[func(nodeID uint32)]
+
 	// Delta log for incremental replication
 	deltaLog *deltaLog
 

server_stats.go

@@ -118,6 +118,21 @@ type BeaconStatsProvider = dashpkg.BeaconStatsProvider
 // DashboardStats is an alias for dashpkg.DashboardStats (moved in R5.2).
 type DashboardStats = dashpkg.DashboardStats
 
+// SetHandshakeNotifier installs the hook called with the node ID a relayed
+// handshake request or answer was just stored for. The host wires it to the
+// beacon, which prompts that node to poll instead of leaving the handshake
+// parked until the node's next once-a-minute poll. The hook must not block;
+// it is told only the recipient's node ID.
+func (s *Server) SetHandshakeNotifier(fn func(nodeID uint32)) {
+	s.handshakeNotifier.Store(&fn)
+}
+
+func (s *Server) notifyHandshake(nodeID uint32) {
+	if fn := s.handshakeNotifier.Load(); fn != nil && *fn != nil && nodeID != 0 {
+		(*fn)(nodeID)
+	}
+}
+
 // SetBeaconStats wires a BeaconStatsProvider into the registry so
 // /api/stats can return relay-forward counts. Also installs the metrics-side
 // adapter so pilot_beacon_relay_forwarded_total / _dropped_total /

zz_handshake_notifier_test.go

@@ -0,0 +1,88 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package server
+
+import (
+	"encoding/base64"
+	"fmt"
+	"testing"
+
+	"github.com/pilot-protocol/common/crypto"
+)
+
+// TestHandshakeNotifierNamesTheRecipient: a relayed handshake request tells
+// the notifier the target's node ID, the relayed answer tells it the
+// original requester's, and a refused message tells it nothing.
+func TestHandshakeNotifierNamesTheRecipient(t *testing.T) {
+	t.Parallel()
+	s := newTestServer(t, "")
+
+	var notified []uint32
+	s.SetHandshakeNotifier(func(nodeID uint32) { notified = append(notified, nodeID) })
+
+	register := func() (*crypto.Identity, uint32) {
+		t.Helper()
+		id, err := crypto.GenerateIdentity()
+		if err != nil {
+			t.Fatal(err)
+		}
+		resp, err := s.handleMessage(map[string]interface{}{
+			"type":        "register",
+			"listen_addr": "203.0.113.5:4000",
+			"public_key":  crypto.EncodePublicKey(id.PublicKey),
+		}, "203.0.113.5:50000")
+		if err != nil {
+			t.Fatalf("register: %v", err)
+		}
+		var nodeID uint32
+		switch v := resp["node_id"].(type) {
+		case uint32:
+			nodeID = v
+		case float64:
+			nodeID = uint32(v)
+		default:
+			t.Fatalf("node_id has type %T", resp["node_id"])
+		}
+		return id, nodeID
+	}
+	sign := func(id *crypto.Identity, format string, a, b uint32) string {
+		return base64.StdEncoding.EncodeToString(id.Sign([]byte(fmt.Sprintf(format, a, b))))
+	}
+	idA, a := register()
+	idB, b := register()
+
+	request := map[string]interface{}{
+		"type":          "request_handshake",
+		"from_node_id":  float64(a),
+		"to_node_id":    float64(b),
+		"justification": "hello",
+		"signature":     sign(idA, "handshake:%d:%d", a, b),
+	}
+	if _, err := s.handleMessage(request, "203.0.113.5:50000"); err != nil {
+		t.Fatalf("request_handshake: %v", err)
+	}
+	if len(notified) != 1 || notified[0] != b {
+		t.Fatalf("after the request: notified %v, want [%d] (the target)", notified, b)
+	}
+
+	// The duplicate is refused; nothing new is waiting, so no notify.
+	if _, err := s.handleMessage(request, "203.0.113.5:50000"); err == nil {
+		t.Fatal("duplicate request accepted")
+	}
+	if len(notified) != 1 {
+		t.Fatalf("refused request notified: %v", notified)
+	}
+
+	if _, err := s.handleMessage(map[string]interface{}{
+		"type":      "respond_handshake",
+		"node_id":   float64(b),
+		"peer_id":   float64(a),
+		"accept":    true,
+		"signature": sign(idB, "respond:%d:%d", b, a),
+	}, "203.0.113.5:50000"); err != nil {
+		t.Fatalf("respond_handshake: %v", err)
+	}
+	if len(notified) != 2 || notified[1] != a {
+		t.Fatalf("after the answer: notified %v, want [%d %d] (target, then requester)", notified, b, a)
+	}
+}