diff --git a/internal/anchor/client.go b/internal/anchor/client.go index a55b7a6..20b309f 100644 --- a/internal/anchor/client.go +++ b/internal/anchor/client.go @@ -100,6 +100,46 @@ func runOnce(ctx context.Context, anchorURL, netHash string, id *crypto.Identity s := &sender{id: id, sendCh: sendCh} + // Drain gossip-discovered peers and initiate offers. Stopped when runOnce + // returns (via drainDone) so stale sessions from a dead connection are never reused. + drainDone := make(chan struct{}) + defer close(drainDone) + go func() { + for { + select { + case pid, ok := <-m.PendingConnect: + if !ok { + return + } + mu.RLock() + _, already := sessions[pid] + mu.RUnlock() + if already { + continue + } + // Use the same lexicographic tiebreak as anchor join to avoid + // both sides trying to offer simultaneously. + if strings.Compare(string(id.PeerID()), string(pid)) <= 0 { + continue + } + go func(pid proto.PeerID) { + sess, err := startOffer(ctx, pid, id, m, s) + if err != nil { + log.Printf("anchor: gossip offer to %s: %v", pid.Short(), err) + return + } + mu.Lock() + sessions[pid] = sess + mu.Unlock() + }(pid) + case <-drainDone: + return + case <-ctx.Done(): + return + } + } + }() + for { var msg proto.AnchorMessage if err := wsjson.Read(ctx, conn, &msg); err != nil { diff --git a/internal/mesh/mesh.go b/internal/mesh/mesh.go index 137b288..385574c 100644 --- a/internal/mesh/mesh.go +++ b/internal/mesh/mesh.go @@ -38,6 +38,10 @@ type Mesh struct { outbound map[string]*outboundTransfer // xid → pending outbound inbound map[string]*inboundTransfer // xid → pending inbound + // PendingConnect receives peer IDs discovered via gossip that we should + // attempt to connect to. Drained by the anchor client's runOnce loop. + PendingConnect chan proto.PeerID + // subscribers receive a copy of every event (fan-out to IPC clients) subMu sync.Mutex subs []chan proto.IpcMessage @@ -47,11 +51,12 @@ type Mesh struct { // Pass a non-nil store to enable message and peer persistence. func New(id *crypto.Identity, st *store.Store) *Mesh { return &Mesh{ - Identity: id, - Store: st, - peers: make(map[proto.PeerID]*PeerConn), - outbound: make(map[string]*outboundTransfer), - inbound: make(map[string]*inboundTransfer), + Identity: id, + Store: st, + peers: make(map[proto.PeerID]*PeerConn), + outbound: make(map[string]*outboundTransfer), + inbound: make(map[string]*inboundTransfer), + PendingConnect: make(chan proto.PeerID, 32), } } diff --git a/internal/mesh/peer.go b/internal/mesh/peer.go index 5e6ca36..e2c1cf4 100644 --- a/internal/mesh/peer.go +++ b/internal/mesh/peer.go @@ -150,6 +150,8 @@ func handleDCMessage(data []byte, from proto.PeerID, localID *crypto.Identity, m PeerID: peerIDPtr(from), Nick: hello.Nick, }) + // Tell the new peer about everyone we can currently see. + go m.sendGossipTo(from) return } @@ -224,9 +226,28 @@ func dispatchPeerMessage(msg proto.PeerMessage, from proto.PeerID, m *Mesh) { log.Printf("mesh: file-done from %s xid=%s", from.Short(), msg.Xid) case proto.MsgPeerGossip: - if msg.Gossip != nil { - log.Printf("mesh: gossip from %s: %d peer hints", from.Short(), len(msg.Gossip.Peers)) + if msg.Gossip == nil { + return } + // Build set of peers we already know about (connected + self). + connected := m.ConnectedPeers() + known := make(map[proto.PeerID]bool, len(connected)+1) + known[m.Identity.PeerID()] = true + for _, p := range connected { + known[p.ID] = true + } + newPeers := 0 + for _, entry := range msg.Gossip.Peers { + if known[entry.Peer.ID] { + continue + } + select { + case m.PendingConnect <- entry.Peer.ID: + newPeers++ + default: + } + } + log.Printf("mesh: gossip from %s: %d hints, %d new", from.Short(), len(msg.Gossip.Peers), newPeers) case proto.MsgPing: log.Printf("mesh: ping from %s", from.Short()) case proto.MsgPong: @@ -272,4 +293,29 @@ func fingerprintFromSDP(sdp string) []byte { return nil } +// sendGossipTo sends the current live peer list to peerID so they can discover +// and connect to peers they haven't met yet. Called after hello is verified. +func (m *Mesh) sendGossipTo(peerID proto.PeerID) { + peers := m.ConnectedPeers() + entries := make([]proto.GossipEntry, 0, len(peers)) + for _, p := range peers { + if p.ID == peerID { + continue // don't include the recipient in their own gossip + } + entries = append(entries, proto.GossipEntry{Peer: p}) + } + if len(entries) == 0 { + return + } + wire, err := json.Marshal(proto.PeerMessage{ + Type: proto.MsgPeerGossip, + Gossip: &proto.PeerGossip{Peers: entries}, + }) + if err != nil { + return + } + m.SendTo(peerID, wire) + log.Printf("mesh: sent gossip to %s: %d peer(s)", peerID.Short(), len(entries)) +} + func peerIDPtr(p proto.PeerID) *proto.PeerID { return &p }