Approve: register the trust pair before answering, and retry the answer

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()

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(&registryOverrideRuntime{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)
+	}
+}

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)