From bd21314ab8edd92df03ee343303d0d5318d51d5e Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Wed, 28 Oct 2026 14:55:01 -0700 Subject: [PATCH 1/2] Approve: register the trust pair before answering, and retry the answer Approving a request told the registry about the new trust pair and sent the answer in parallel, once each, discarding every error. A private requester can't be looked up until the registry has the pair, so the direct accept usually failed; the relayed answer needs the registry's in-memory record of the request. When both copies were lost, the approver trusted the requester (one-way, with no key for a relayed request) while the requester still waited, and kept its messages queued: five hosted pairs had been stuck like this for up to three weeks. Now the pair is registered first, a relayed request's key is taken from the registry (backfillPeerKey), and the relayed and direct answers follow, each registry call retried (three tries, unless stopping; "no pending" is final), with a warning when the answer reached neither. Co-Authored-By: Claude Opus 5.5 --- handshake.go | 60 +++++++++++++++++----- zz_approve_answer_order_test.go | 89 +++++++++++++++++++++++++++++++++ 2 files changed, 137 insertions(+), 12 deletions(-) create mode 100644 zz_approve_answer_order_test.go diff --git a/handshake.go b/handshake.go index 6fcc64a..a5ebf67 100644 --- a/handshake.go +++ b/handshake.go @@ -1652,22 +1652,58 @@ func (hm *Manager) ApproveHandshake(peerNodeID uint32) error { "peer_node_id": peerNodeID, "state": "granted", "reason": "approved", }) - // Report trust to registry (creates the trust pair for resolve authorization) - if reg := hm.rt.Registry(); reg != nil { - nodeID := hm.rt.NodeID() - sig := hm.signHandshakeChallenge(fmt.Sprintf("respond:%d:%d", nodeID, peerNodeID)) - hm.goRPC(func() { - reg.ReportTrust(nodeID, peerNodeID) - // Also respond via registry relay in case the request was relayed - reg.RespondHandshake(nodeID, peerNodeID, true, sig) - }) + // Tell the requester, in the order that lets the answer arrive. The + // registry learns the trust pair first: until it has, a private requester + // cannot be looked up, so the direct accept fails. Then the answer goes + // through the relay (for a relayed request) and directly. Each step is + // retried and logged: when every copy of the answer was lost, this side + // trusted the requester while the requester still waited (and kept its + // messages queued) indefinitely. + reg := hm.rt.Registry() + if reg == nil { + hm.sendAccept(peerNodeID) + return nil } - - // Try to send accept directly (may fail if peer is private, that's OK) - hm.sendAccept(peerNodeID) + nodeID := hm.rt.NodeID() + sig := hm.signHandshakeChallenge(fmt.Sprintf("respond:%d:%d", nodeID, peerNodeID)) + relayedRequest := req.PublicKey == "" + hm.goRPC(func() { + if err := hm.retryRPC(func() error { _, err := reg.ReportTrust(nodeID, peerNodeID); return err }); err != nil { + slog.Warn("handshake approved, but the registry did not take the trust pair", "peer_node_id", peerNodeID, "error", err) + } + if relayedRequest { + hm.backfillPeerKey(peerNodeID) // a relayed request carries no key: take the registry's + } + relayErr := hm.retryRPC(func() error { _, err := reg.RespondHandshake(nodeID, peerNodeID, true, sig); return err }) + if err := hm.sendAccept(peerNodeID); err != nil && relayErr != nil { + slog.Warn("handshake approved, but the answer reached neither the relay nor the peer", + "peer_node_id", peerNodeID, "relay_error", relayErr, "direct_error", err) + } + }) return nil } +// retryRPC runs a registry call up to three times, a second apart and then +// two, unless the manager is stopping. A "no pending" answer is final. +func (hm *Manager) retryRPC(call func() error) error { + var err error + for attempt, wait := range []time.Duration{0, time.Second, 2 * time.Second} { + if attempt > 0 { + hm.mu.RLock() + stopping := hm.stopping + hm.mu.RUnlock() + if stopping { + return err + } + time.Sleep(wait) + } + if err = call(); err == nil || strings.Contains(err.Error(), "no pending") { + return err + } + } + return err +} + // RejectHandshake rejects a pending handshake request. func (hm *Manager) RejectHandshake(peerNodeID uint32, reason string) error { hm.mu.Lock() diff --git a/zz_approve_answer_order_test.go b/zz_approve_answer_order_test.go new file mode 100644 index 0000000..35c0002 --- /dev/null +++ b/zz_approve_answer_order_test.go @@ -0,0 +1,89 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package handshake + +import ( + "errors" + "sync" + "testing" + "time" +) + +// orderedRegistry records the order of the calls an approval makes, and +// fails the first RespondHandshake the way a dropped connection does. +type orderedRegistry struct { + *fakeRegistry + mu sync.Mutex + order []string + failOnce bool +} + +func (o *orderedRegistry) note(s string) { + o.mu.Lock() + o.order = append(o.order, s) + o.mu.Unlock() +} +func (o *orderedRegistry) ReportTrust(n, p uint32) (map[string]interface{}, error) { + o.note("report") + return o.fakeRegistry.ReportTrust(n, p) +} +func (o *orderedRegistry) Lookup(n uint32) (map[string]interface{}, error) { + o.note("lookup") + return o.fakeRegistry.Lookup(n) +} +func (o *orderedRegistry) RespondHandshake(n, p uint32, accept bool, sig string) (map[string]interface{}, error) { + o.mu.Lock() + fail := o.failOnce + o.failOnce = false + o.mu.Unlock() + if fail { + o.note("respond-failed") + return nil, errors.New("connection reset by peer") + } + o.note("respond") + return o.fakeRegistry.RespondHandshake(n, p, accept, sig) +} + +// Approving a relayed request (which carries no key): the registry learns the +// trust pair before anything is answered, the requester's key comes from the +// registry, a dropped relay answer is tried again, and the direct accept is +// still attempted. A lost answer once left the two sides disagreeing for weeks. +func TestApproveRegistersThePairBeforeAnsweringAndRetries(t *testing.T) { + reg := &orderedRegistry{fakeRegistry: newFakeRegistry(), failOnce: true} + reg.setLookup(252515, map[string]interface{}{"public_key": "cmVxdWVzdGVyLWtleQ=="}) + frt := newFakeRuntime() + hm := NewManager(®istryOverrideRuntime{fakeRuntime: frt, reg: reg}) + t.Cleanup(hm.Stop) + hm.mu.Lock() + hm.pending[252515] = &PendingHandshake{NodeID: 252515} // relayed: no key + hm.mu.Unlock() + + if err := hm.ApproveHandshake(252515); err != nil { + t.Fatal(err) + } + deadline := time.Now().Add(10 * time.Second) + for { + reg.mu.Lock() + order := append([]string(nil), reg.order...) + reg.mu.Unlock() + frt.mu.RLock() + dialed := len(frt.dialCalls) + frt.mu.RUnlock() + if len(order) >= 4 && dialed > 0 { + if order[0] != "report" || order[1] != "lookup" || order[2] != "respond-failed" || order[3] != "respond" { + t.Fatalf("order: %v", order) + } + break + } + if time.Now().After(deadline) { + t.Fatalf("order so far: %v, direct accepts tried: %d", order, dialed) + } + time.Sleep(20 * time.Millisecond) + } + hm.mu.RLock() + rec := hm.trusted[252515] + hm.mu.RUnlock() + if rec == nil || rec.PublicKey != "cmVxdWVzdGVyLWtleQ==" { + t.Fatalf("trust record: %+v", rec) + } +} From 34f129e1f50ba13ad20fc9f4a97de854c1942595 Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Wed, 28 Oct 2026 15:00:39 -0700 Subject: [PATCH 2/2] Test: stop the first manager before the restart reads its file TestOutgoing_CancelledRequestIsForgottenAcrossRestart never stopped the manager it restarts from, so its background save could write trust.json while the temp dir was being removed ("directory not empty"); it failed 30 of 30 runs under -race on main. Co-Authored-By: Claude Opus 5.5 --- zz_outgoing_lifetime_test.go | 1 + 1 file changed, 1 insertion(+) diff --git a/zz_outgoing_lifetime_test.go b/zz_outgoing_lifetime_test.go index 976e61b..edd1080 100644 --- a/zz_outgoing_lifetime_test.go +++ b/zz_outgoing_lifetime_test.go @@ -92,6 +92,7 @@ func TestOutgoing_CancelledRequestIsForgottenAcrossRestart(t *testing.T) { before.saveTrust() before.RevokeTrust(95) before.saveTrust() + before.Stop() // the restart: its background save is done before the next one reads the file after := newTestHM(t, path) t.Cleanup(after.Stop)