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) + } +}