A friend request stays answerable for 30 days, across restarts

handshake.go

@@ -98,6 +98,17 @@ const (
 	// operators ample time to decide without letting the queue grow
 	// unboundedly under sustained inbound handshake pressure.
 	pendingHandshakeTTL = 30 * 24 * time.Hour
+	// outgoingRequestTTL is how long a request this node sent stays
+	// answerable: the peer keeps it pending for pendingHandshakeTTL, and an
+	// acceptance is honoured only while the matching outgoing entry exists.
+	// It used to be 2 x handshakeMaxAge (10 minutes), so a person who
+	// accepted later than that, as people do, was trusted on their side and
+	// silently dropped as "unsolicited" on the requester's, which then showed
+	// the request as waiting forever.
+	outgoingRequestTTL = pendingHandshakeTTL
+	// maxOutgoingRequests bounds the outgoing set; past it the oldest
+	// request is forgotten first.
+	maxOutgoingRequests = 4096
 )
 
 // replayEntry is one recorded handshake message hash: when it was first
@@ -301,7 +312,7 @@ func (hm *Manager) goRPCLocked(fn func()) {
 func (hm *Manager) markTrustedLocked(nodeID uint32, rec *TrustRecord) {
 	hm.trusted[nodeID] = rec
 	hm.deletePendingLocked(nodeID)
-	delete(hm.outgoing, nodeID)
+	hm.dropOutgoingLocked(nodeID)
 	hm.trustWaitersMu.Lock()
 	waiters := hm.trustWaiters[nodeID]
 	delete(hm.trustWaiters, nodeID)
@@ -379,6 +390,14 @@ type trustSnapshot struct {
 	Trusted []trustSnapshotEntry   `json:"trusted"`
 	Pending []pendingSnapshotEntry `json:"pending,omitempty"`
 	Revoked []revokedSnapshotEntry `json:"revoked,omitempty"`
+	// Outgoing are the requests this node sent and is still waiting on, so
+	// an acceptance that arrives after a restart is still honoured.
+	Outgoing []outgoingSnapshotEntry `json:"outgoing,omitempty"`
+}
+
+type outgoingSnapshotEntry struct {
+	NodeID uint32 `json:"node_id"`
+	SentAt string `json:"sent_at"`
 }
 
 type trustSnapshotEntry struct {
@@ -463,6 +482,11 @@ func (hm *Manager) saveTrustLocked() {
 			Source:        p.Source,
 		})
 	}
+	for nodeID, sentAt := range hm.outgoing {
+		if time.Since(sentAt) < outgoingRequestTTL {
+			snap.Outgoing = append(snap.Outgoing, outgoingSnapshotEntry{NodeID: nodeID, SentAt: sentAt.Format(time.RFC3339)})
+		}
+	}
 	for nodeID, until := range hm.revoked {
 		// Only persist entries whose cooldown hasn't expired yet.
 		if time.Now().Before(until) {
@@ -585,7 +609,18 @@ func (hm *Manager) loadTrust() {
 			hm.revoked[e.NodeID] = until
 		}
 	}
-	slog.Info("loaded trust state", "peers", len(hm.trusted), "pending", len(hm.pending), "revoked", len(hm.revoked))
+	for _, e := range snap.Outgoing {
+		sent, err := time.Parse(time.RFC3339, e.SentAt)
+		if err != nil {
+			slog.Warn("outgoing entry has malformed SentAt — skipping",
+				"node_id", e.NodeID, "sent_at", e.SentAt, "err", err)
+			continue
+		}
+		if _, trusted := hm.trusted[e.NodeID]; !trusted && time.Since(sent) < outgoingRequestTTL {
+			hm.outgoing[e.NodeID] = sent
+		}
+	}
+	slog.Info("loaded trust state", "peers", len(hm.trusted), "pending", len(hm.pending), "revoked", len(hm.revoked), "outgoing", len(hm.outgoing))
 }
 
 // Start binds port 444 and begins handling handshake connections.
@@ -917,8 +952,8 @@ func (hm *Manager) reapReplayLocked(now time.Time) {
 //
 // outgoing: peers we sent a handshake request to but never heard back from.
 // A peer that never responds (wrong address, dropped, ignore-listed) would
-// otherwise hold a map entry forever. We evict after 2× handshakeMaxAge —
-// long enough for any in-flight response to arrive.
+// otherwise hold a map entry forever. We evict after outgoingRequestTTL,
+// as long as the peer keeps the request pending for a person to answer.
 //
 // revoked: peers in post-revocation cooldown. The lazy-delete in handleRequest
 // / handleRelayRequest only fires on the next *contact from that peer*, which
@@ -927,10 +962,9 @@ func (hm *Manager) reapOutgoingAndRevoked() {
 	hm.mu.Lock()
 	defer hm.mu.Unlock()
 	now := time.Now()
-	outgoingTTL := 2 * handshakeMaxAge
 	for nodeID, sentAt := range hm.outgoing {
-		if now.Sub(sentAt) > outgoingTTL {
-			delete(hm.outgoing, nodeID)
+		if now.Sub(sentAt) > outgoingRequestTTL {
+			hm.dropOutgoingLocked(nodeID)
 		}
 	}
 	for nodeID, until := range hm.revoked {
@@ -940,6 +974,33 @@ func (hm *Manager) reapOutgoingAndRevoked() {
 	}
 }
 
+// dropOutgoingLocked forgets a request this node sent, and marks the state
+// for saving, since outgoing requests are persisted. Caller MUST hold
+// hm.mu (write-locked).
+func (hm *Manager) dropOutgoingLocked(nodeID uint32) {
+	if _, ok := hm.outgoing[nodeID]; ok {
+		delete(hm.outgoing, nodeID)
+		hm.markDirty()
+	}
+}
+
+// addOutgoingLocked records a request this node sent, forgetting the oldest
+// when the set is full. Caller MUST hold hm.mu (write-locked).
+func (hm *Manager) addOutgoingLocked(nodeID uint32, at time.Time) {
+	if _, ok := hm.outgoing[nodeID]; !ok && len(hm.outgoing) >= maxOutgoingRequests {
+		var oldest uint32
+		var oldestAt time.Time
+		for id, t := range hm.outgoing {
+			if oldestAt.IsZero() || t.Before(oldestAt) {
+				oldest, oldestAt = id, t
+			}
+		}
+		delete(hm.outgoing, oldest)
+	}
+	hm.outgoing[nodeID] = at
+	hm.markDirty()
+}
+
 func (hm *Manager) deletePendingLocked(nodeID uint32) {
 	p, ok := hm.pending[nodeID]
 	if !ok {
@@ -1027,7 +1088,7 @@ func (hm *Manager) handleRequest(stream coreapi.Stream, msg *HandshakeMsg, regis
 	// cannot claim any NodeID with their own keypair and slip into trust.
 	if _, ok := hm.outgoing[peerNodeID]; ok && registryBound {
 		// Mutual! Auto-approve (registry confirmed pubkey binding)
-		delete(hm.outgoing, peerNodeID)
+		hm.dropOutgoingLocked(peerNodeID)
 		hm.markTrustedLocked(peerNodeID, &TrustRecord{
 			NodeID:     peerNodeID,
 			PublicKey:  msg.PublicKey,
@@ -1179,7 +1240,7 @@ func (hm *Manager) handleAccept(msg *HandshakeMsg) {
 	if until, ok := hm.revoked[peerNodeID]; ok {
 		if time.Now().Before(until) {
 			slog.Info("ignoring handshake acceptance from recently-revoked peer", "peer_node_id", peerNodeID)
-			delete(hm.outgoing, peerNodeID)
+			hm.dropOutgoingLocked(peerNodeID)
 			return
 		}
 		delete(hm.revoked, peerNodeID)
@@ -1199,7 +1260,7 @@ func (hm *Manager) handleAccept(msg *HandshakeMsg) {
 	if existing, ok := hm.trusted[peerNodeID]; ok {
 		bound := existing.PublicKey
 		if !hm.reconcileTrustBindingLocked(peerNodeID, msg.PublicKey) {
-			delete(hm.outgoing, peerNodeID)
+			hm.dropOutgoingLocked(peerNodeID)
 			return
 		}
 		// Preserve an already-bound key when the accept omits one.
@@ -1208,7 +1269,7 @@ func (hm *Manager) handleAccept(msg *HandshakeMsg) {
 		}
 	}
 
-	delete(hm.outgoing, peerNodeID)
+	hm.dropOutgoingLocked(peerNodeID)
 	hm.markTrustedLocked(peerNodeID, &TrustRecord{
 		NodeID:     peerNodeID,
 		PublicKey:  msg.PublicKey,
@@ -1229,7 +1290,7 @@ func (hm *Manager) handleRejectMsg(msg *HandshakeMsg) {
 	slog.Warn("handshake rejected by peer", "peer_node_id", msg.NodeID, "reason", msg.Reason)
 
 	hm.mu.Lock()
-	delete(hm.outgoing, msg.NodeID)
+	hm.dropOutgoingLocked(msg.NodeID)
 	hm.mu.Unlock()
 }
 
@@ -1257,7 +1318,7 @@ func (hm *Manager) SendRequest(peerNodeID uint32, justification string) error {
 	// themselves locked out of a peer that was auto-pruned by policy
 	// (trust-decay, etc.) with no clean recovery path.
 	delete(hm.revoked, peerNodeID)
-	hm.outgoing[peerNodeID] = time.Now()
+	hm.addOutgoingLocked(peerNodeID, time.Now())
 	hm.mu.Unlock()
 
 	pubKeyStr := ""
@@ -1324,7 +1385,7 @@ func (hm *Manager) processRelayedRequest(fromNodeID uint32, justification string
 
 	// Check if we have an outgoing request to this peer (mutual handshake)
 	if _, ok := hm.outgoing[fromNodeID]; ok {
-		delete(hm.outgoing, fromNodeID)
+		hm.dropOutgoingLocked(fromNodeID)
 		hm.markTrustedLocked(fromNodeID, &TrustRecord{
 			NodeID:     fromNodeID,
 			ApprovedAt: time.Now(),
@@ -1468,7 +1529,7 @@ func (hm *Manager) processRelayedApproval(fromNodeID uint32) {
 	if until, ok := hm.revoked[fromNodeID]; ok {
 		if time.Now().Before(until) {
 			slog.Info("ignoring relayed approval for recently-revoked peer", "peer_node_id", fromNodeID)
-			delete(hm.outgoing, fromNodeID)
+			hm.dropOutgoingLocked(fromNodeID)
 			return
 		}
 		delete(hm.revoked, fromNodeID)
@@ -1479,7 +1540,7 @@ func (hm *Manager) processRelayedApproval(fromNodeID uint32) {
 		return
 	}
 
-	delete(hm.outgoing, fromNodeID)
+	hm.dropOutgoingLocked(fromNodeID)
 	hm.markTrustedLocked(fromNodeID, &TrustRecord{
 		NodeID:     fromNodeID,
 		ApprovedAt: time.Now(),
@@ -1555,7 +1616,7 @@ func (hm *Manager) ProcessRelayedRejection(fromNodeID uint32) {
 // processRelayedRejection handles a handshake rejection received via registry relay.
 func (hm *Manager) processRelayedRejection(fromNodeID uint32) {
 	hm.mu.Lock()
-	delete(hm.outgoing, fromNodeID)
+	hm.dropOutgoingLocked(fromNodeID)
 	hm.mu.Unlock()
 	slog.Info("handshake rejected via relay", "peer_node_id", fromNodeID)
 }
@@ -1654,7 +1715,7 @@ func (hm *Manager) RevokeTrust(peerNodeID uint32) error {
 	_, wasPending := hm.pending[peerNodeID]
 	delete(hm.trusted, peerNodeID)
 	hm.deletePendingLocked(peerNodeID)
-	delete(hm.outgoing, peerNodeID)
+	hm.dropOutgoingLocked(peerNodeID)
 	// Block stale relayed approvals still sitting in the registry inbox from
 	// re-establishing trust right after a local revoke. 5-minute cooldown covers
 	// the normal poll cycle many times over.
@@ -1719,7 +1780,7 @@ func (hm *Manager) handleRevokeMsg(msg *HandshakeMsg) {
 	_, wasPending := hm.pending[peerNodeID]
 	delete(hm.trusted, peerNodeID)
 	hm.deletePendingLocked(peerNodeID)
-	delete(hm.outgoing, peerNodeID)
+	hm.dropOutgoingLocked(peerNodeID)
 	if wasTrusted || wasPending {
 		hm.markDirty()
 	}

zz_outgoing_lifetime_test.go

@@ -0,0 +1,126 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package handshake
+
+import (
+	"path/filepath"
+	"testing"
+	"time"
+)
+
+// A person accepts a friend request when they see it, often hours later.
+// The acceptance must still complete trust on the requester's side; it used
+// to be dropped as unsolicited once the outgoing entry passed 10 minutes.
+func TestOutgoing_AcceptanceHoursLaterIsHonoured(t *testing.T) {
+	t.Parallel()
+	hm := newTestHM(t, "")
+	t.Cleanup(hm.Stop)
+	hm.mu.Lock()
+	hm.addOutgoingLocked(77, time.Now().Add(-3*time.Hour))
+	hm.mu.Unlock()
+	hm.reapOutgoingAndRevoked() // the reaper must not drop it
+
+	hm.handleAccept(&HandshakeMsg{Type: HandshakeAccept, NodeID: 77, PublicKey: "k", Timestamp: time.Now().Unix()})
+
+	hm.mu.RLock()
+	defer hm.mu.RUnlock()
+	if _, ok := hm.trusted[77]; !ok {
+		t.Fatal("an acceptance three hours after the request was dropped")
+	}
+}
+
+// The same, relayed through the registry, which is how most private nodes
+// learn of an acceptance.
+func TestOutgoing_RelayedApprovalHoursLaterIsHonoured(t *testing.T) {
+	t.Parallel()
+	hm := newTestHM(t, "")
+	t.Cleanup(hm.Stop)
+	hm.mu.Lock()
+	hm.addOutgoingLocked(78, time.Now().Add(-5*time.Hour))
+	hm.mu.Unlock()
+	hm.reapOutgoingAndRevoked()
+
+	hm.ProcessRelayedApproval(78)
+
+	hm.mu.RLock()
+	defer hm.mu.RUnlock()
+	if _, ok := hm.trusted[78]; !ok {
+		t.Fatal("a relayed approval five hours after the request was dropped")
+	}
+}
+
+// A request survives a restart: it is saved with the trust state, and an
+// acceptance that arrives after the restart completes trust.
+func TestOutgoing_SurvivesRestart(t *testing.T) {
+	t.Parallel()
+	path := filepath.Join(t.TempDir(), "trust.json")
+	before := newTestHM(t, path)
+	before.mu.Lock()
+	before.addOutgoingLocked(90, time.Now().Add(-time.Hour))
+	before.addOutgoingLocked(91, time.Now().Add(-outgoingRequestTTL-time.Hour)) // too old to keep
+	before.mu.Unlock()
+	before.saveTrust()
+
+	after := newTestHM(t, path)
+	t.Cleanup(after.Stop)
+	after.loadTrust()
+	after.mu.RLock()
+	_, kept := after.outgoing[90]
+	_, stale := after.outgoing[91]
+	after.mu.RUnlock()
+	if !kept || stale {
+		t.Fatalf("after a restart: request 90 kept=%v (want true), expired 91 kept=%v (want false)", kept, stale)
+	}
+
+	after.ProcessRelayedApproval(90)
+	after.mu.RLock()
+	defer after.mu.RUnlock()
+	if _, ok := after.trusted[90]; !ok {
+		t.Fatal("an acceptance after a restart was dropped")
+	}
+}
+
+// Cancelling a request (revoking) forgets it for good, also across a restart,
+// so a late acceptance of a cancelled request is not honoured.
+func TestOutgoing_CancelledRequestIsForgottenAcrossRestart(t *testing.T) {
+	t.Parallel()
+	path := filepath.Join(t.TempDir(), "trust.json")
+	before := newTestHM(t, path)
+	before.mu.Lock()
+	before.addOutgoingLocked(95, time.Now())
+	before.mu.Unlock()
+	before.saveTrust()
+	before.RevokeTrust(95)
+	before.saveTrust()
+
+	after := newTestHM(t, path)
+	t.Cleanup(after.Stop)
+	after.loadTrust()
+	after.mu.RLock()
+	defer after.mu.RUnlock()
+	if _, ok := after.outgoing[95]; ok {
+		t.Fatal("a cancelled request came back after a restart")
+	}
+}
+
+// The outgoing set is bounded: past maxOutgoingRequests the oldest goes.
+func TestOutgoing_Bounded(t *testing.T) {
+	t.Parallel()
+	hm := newTestHM(t, "")
+	t.Cleanup(hm.Stop)
+	start := time.Now().Add(-time.Hour)
+	hm.mu.Lock()
+	defer hm.mu.Unlock()
+	for i := 0; i <= maxOutgoingRequests; i++ {
+		hm.addOutgoingLocked(uint32(1000+i), start.Add(time.Duration(i)*time.Millisecond))
+	}
+	if len(hm.outgoing) != maxOutgoingRequests {
+		t.Fatalf("outgoing holds %d, want %d", len(hm.outgoing), maxOutgoingRequests)
+	}
+	if _, ok := hm.outgoing[1000]; ok {
+		t.Fatal("the oldest request should have been forgotten first")
+	}
+	if _, ok := hm.outgoing[uint32(1000+maxOutgoingRequests)]; !ok {
+		t.Fatal("the newest request is missing")
+	}
+}

zz_relayed_more_test.go

@@ -173,10 +173,12 @@ func TestReapOutgoingAndRevoked_PrunesExpired(t *testing.T) {
 	t.Parallel()
 	hm := newTestHM(t, "")
 	hm.mu.Lock()
-	// Old outgoing — should be reaped.
-	hm.outgoing[10] = time.Now().Add(-2 * time.Hour)
+	// Past outgoingRequestTTL — should be reaped.
+	hm.outgoing[10] = time.Now().Add(-outgoingRequestTTL - time.Hour)
 	// Fresh outgoing — should remain.
 	hm.outgoing[20] = time.Now()
+	// Hours old: a person may still accept it — should remain.
+	hm.outgoing[30] = time.Now().Add(-2 * time.Hour)
 	hm.mu.Unlock()
 
 	hm.reapOutgoingAndRevoked()
@@ -189,4 +191,7 @@ func TestReapOutgoingAndRevoked_PrunesExpired(t *testing.T) {
 	if _, ok := hm.outgoing[20]; !ok {
 		t.Error("fresh outgoing should remain")
 	}
+	if _, ok := hm.outgoing[30]; !ok {
+		t.Error("a request sent two hours ago should still be answerable")
+	}
 }