From af2f42984706406f8cd4364b12318f2b81c651ea Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Thu, 1 Oct 2026 18:40:11 +0300 Subject: [PATCH] registry: tell the beacon which node a relayed handshake was parked for request_handshake and respond_handshake store a message that the recipient only finds on its next poll, once a minute. Add SetHandshakeNotifier: after either succeeds the hook is called with the recipient's node ID. cmd/rendezvous wires it to the co-located beacon's NotifyNode when the beacon in use has it (asserted, so this builds against beacon releases that predate it), which prompts an updated daemon to poll at once. The hook is told nothing but the recipient's node ID; a refused request notifies nobody. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 7 +++ cmd/rendezvous/main.go | 6 +++ dispatch.go | 12 ++++- server.go | 5 ++ server_stats.go | 15 ++++++ zz_handshake_notifier_test.go | 88 +++++++++++++++++++++++++++++++++++ 6 files changed, 131 insertions(+), 2 deletions(-) create mode 100644 zz_handshake_notifier_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index c359bce..2739014 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/cmd/rendezvous/main.go b/cmd/rendezvous/main.go index cebdf96..aa51b7c 100644 --- a/cmd/rendezvous/main.go +++ b/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 { diff --git a/dispatch.go b/dispatch.go index 640fe92..9f387ae 100644 --- a/dispatch.go +++ b/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) diff --git a/server.go b/server.go index 7be9bcc..015d741 100644 --- a/server.go +++ b/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 diff --git a/server_stats.go b/server_stats.go index a6fda8a..c4f191c 100644 --- a/server_stats.go +++ b/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 / diff --git a/zz_handshake_notifier_test.go b/zz_handshake_notifier_test.go new file mode 100644 index 0000000..b6bec79 --- /dev/null +++ b/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) + } +}