diff --git a/p2p/connection_pool.go b/p2p/connection_pool.go index 3d7eaa00..d24fd685 100644 --- a/p2p/connection_pool.go +++ b/p2p/connection_pool.go @@ -232,6 +232,11 @@ func ping_loop() { return } c.update(&response.Common) // update common information + latency := atomic.LoadInt64(&c.Latency) + topoH := atomic.LoadInt64(&c.TopoHeight) + if latency > 0 { + Peer_UpdateLatency(Address(c), latency, topoH) + } }() } return true @@ -666,10 +671,16 @@ func trigger_sync() { clist = append(clist, value) } - // sort the list random - // do random shuffling, can we get away with len/2 random shuffling - globals.Global_Random.Shuffle(len(clist), func(i, j int) { - clist[i], clist[j] = clist[j], clist[i] + // sort by height descending (furthest ahead first), then latency ascending (fastest among equal) + sort.SliceStable(clist, func(i, j int) bool { + hi := atomic.LoadInt64(&clist[i].Height) + hj := atomic.LoadInt64(&clist[j].Height) + if hi != hj { + return hi > hj + } + li := atomic.LoadInt64(&clist[i].Latency) + lj := atomic.LoadInt64(&clist[j].Latency) + return li < lj }) for _, connection := range clist { diff --git a/p2p/controller.go b/p2p/controller.go index 58ff34b2..9544109f 100644 --- a/p2p/controller.go +++ b/p2p/controller.go @@ -427,17 +427,60 @@ func maintain_seed_node_connection() { return case <-delay.C: } - endpoint := "" - if globals.IsMainnet() { // choose mainnet seed node - r, _ := rand.Int(rand.Reader, big.NewInt(10240)) - endpoint = config.Mainnet_seed_nodes[r.Int64()%int64(len(config.Mainnet_seed_nodes))] - } else { // choose testnet peer node - r, _ := rand.Int(rand.Reader, big.NewInt(10240)) - endpoint = config.Testnet_seed_nodes[r.Int64()%int64(len(config.Testnet_seed_nodes))] - } - if endpoint != "" { - connect_with_endpoint(endpoint, sync_node) - //connect_with_endpoint(endpoint, true) // seed nodes always have sync mode + + var seeds []string + if globals.IsMainnet() { + seeds = config.Mainnet_seed_nodes + } else { + seeds = config.Testnet_seed_nodes + } + + // split seeds into known (have recent latency data) and unknown + now := uint64(time.Now().UTC().Unix()) + type scoredSeed struct { + addr string + lat int64 + } + var known []scoredSeed + var unknown []string + + peer_mutex.Lock() + for _, s := range seeds { + if p, ok := peer_map[ParseIPNoError(s)]; ok && + p.LastMeasured > 0 && + p.LastLatency > 0 && + (now-p.LastMeasured) < 24*3600 { + + known = append(known, scoredSeed{s, p.LastLatency}) + } else { + unknown = append(unknown, s) + } + } + peer_mutex.Unlock() + + // known seeds: fastest first + sort.Slice(known, func(i, j int) bool { + return known[i].lat < known[j].lat + }) + + // unknown seeds: shuffle for variety + globals.Global_Random.Shuffle(len(unknown), func(i, j int) { + unknown[i], unknown[j] = unknown[j], unknown[i] + }) + + // build ordered list: known (fast first) + unknown (random) + var ordered []string + for _, s := range known { + ordered = append(ordered, s.addr) + } + ordered = append(ordered, unknown...) + + // connect to first non-connected seed + for _, endpoint := range ordered { + if !IsAddressConnected(ParseIPNoError(endpoint)) { + connect_with_endpoint(endpoint, sync_node) + break + } } } } diff --git a/p2p/peer_pool.go b/p2p/peer_pool.go index 1e2813bf..b2b40737 100644 --- a/p2p/peer_pool.go +++ b/p2p/peer_pool.go @@ -58,6 +58,10 @@ type Peer struct { ConnectAfter uint64 `json:"connectafter"` // we should connect when the following timestamp passes BlacklistBefore uint64 `json:"blacklistbefore"` // peer blacklisted till epoch , priority nodes are never blacklisted, 0 if not blacklist GoodCount uint64 `json:"goodcount"` // how many times peer has been shared with us + SuccessCount uint64 `json:"successcount"` // outbound connection successes (for scoring) + LastLatency int64 `json:"lastlatency"` // nanoseconds, from rtt_micro + LastTopoHeight int64 `json:"lasttopoheight"` // peer's topo height at measurement + LastMeasured uint64 `json:"lastmeasured"` // epoch seconds when latency captured Version int `json:"version"` // version 1 is original C daemon peer, version 2 is golang p2p version Whitelist bool `json:"whitelist"` sync.Mutex @@ -202,12 +206,42 @@ func Peer_SetSuccess(address string) { } peer_mutex.Lock() defer peer_mutex.Unlock() - p.FailCount = 0 // fail count is zero again - p.ConnectAfter = 0 - p.Whitelist = true - p.LastConnected = uint64(time.Now().UTC().Unix()) // set time when last connected + p.FailCount = 0 // fail count is zero again + p.ConnectAfter = 0 + p.Whitelist = true + p.LastConnected = uint64(time.Now().UTC().Unix()) // set time when last connected + p.SuccessCount++ - // logger.Infof("Setting peer as white listed") + // logger.Infof("Setting peer as white listed") +} + +// captures live latency/topoheight from the ping path, not from handshake +// call after c.update(&response.Common) in ping_loop when Latency > 0 +func Peer_UpdateLatency(address string, latencyNs int64, topoHeight int64) { + p := GetPeerInList(ParseIPNoError(address)) + if p == nil { + return + } + peer_mutex.Lock() + defer peer_mutex.Unlock() + p.LastLatency = latencyNs + p.LastTopoHeight = topoHeight + p.LastMeasured = uint64(time.Now().UTC().Unix()) +} + +// computes a score for a peer: success/fail history + latency bonus +// latency bonus decays to zero after 24 hours +func peerScore(p *Peer, now uint64) float64 { + score := float64(p.SuccessCount*10) - float64(p.FailCount*50) + + if p.LastMeasured > 0 && p.LastLatency > 0 { + age := now - p.LastMeasured + if age < 24*3600 { + latencyMs := float64(p.LastLatency) / 1e6 // ns → ms + score += 10000.0 / (latencyMs + 1.0) // +1 avoids div-by-zero + } + } + return score } /* @@ -240,24 +274,51 @@ func Peer_Delete(p *Peer) { delete(peer_map, ParseIPNoError(p.Address)) } +func formatAge(lastMeasured uint64) string { + if lastMeasured == 0 { + return "-" + } + age := time.Now().UTC().Unix() - int64(lastMeasured) + if age < 0 { + return "now" + } + switch { + case age < 60: + return fmt.Sprintf("%ds ago", age) + case age < 3600: + return fmt.Sprintf("%dm ago", age/60) + case age < 86400: + return fmt.Sprintf("%dh ago", age/3600) + default: + return fmt.Sprintf("%dd ago", age/86400) + } +} + +func printLatency(p *Peer) string { + if p.LastLatency <= 0 { + return "-" + } + ms := float64(p.LastLatency) / 1e6 + return fmt.Sprintf("%.1fms", ms) +} + // prints all the connection info to screen func PeerList_Print() { peer_mutex.Lock() defer peer_mutex.Unlock() fmt.Printf("Peer List\n") - fmt.Printf("%-22s %-6s %-4s %-5s %-7s %9s %3s\n", "Remote Addr", "Active", "Good", "Fail", " State", "Height", "DIR") + fmt.Printf("%-22s %-6s %4s %5s %4s %8s %8s\n", "Remote Addr", "Active", "Good", "Fail", "Succ", "Lat(ms)", "Age") var list []*Peer greycount := 0 for _, v := range peer_map { - if v.Whitelist { // only display white listed peer + if v.Whitelist { list = append(list, v) } else { greycount++ } } - // sort the list sort.Slice(list, func(i, j int) bool { return list[i].Address < list[j].Address }) for i := range list { @@ -265,7 +326,12 @@ func PeerList_Print() { if IsAddressConnected(ParseIPNoError(list[i].Address)) { connected = "ACTIVE" } - fmt.Printf("%-22s %-6s %4d %5d \n", list[i].Address, connected, list[i].GoodCount, list[i].FailCount) + fmt.Printf("%-22s %-6s %4d %5d %4d %8s %8s\n", + list[i].Address, connected, + list[i].GoodCount, list[i].FailCount, + list[i].SuccessCount, + printLatency(list[i]), + formatAge(list[i].LastMeasured)) } fmt.Printf("\nWhitelist size %d\n", len(peer_map)-greycount) @@ -289,24 +355,53 @@ func find_peer_to_connect(version int) *Peer { peer_mutex.Lock() defer peer_mutex.Unlock() - // first search the whitelisted ones + now := uint64(time.Now().UTC().Unix()) + + // Pass 1: weighted random among eligible whitelist peers (reservoir sampling) + var best *Peer + var totalWeight float64 for _, v := range peer_map { - if uint64(time.Now().Unix()) > v.BlacklistBefore && // if ip is blacklisted skip it - uint64(time.Now().Unix()) > v.ConnectAfter && - !IsAddressConnected(ParseIPNoError(v.Address)) && v.Whitelist && !IsAddressInBanList(ParseIPNoError(v.Address)) { - v.ConnectAfter = uint64(time.Now().UTC().Unix()) + 10 // minimum 10 secs gap - return v + if now > v.BlacklistBefore && + now > v.ConnectAfter && + !IsAddressConnected(ParseIPNoError(v.Address)) && + v.Whitelist && + !IsAddressInBanList(ParseIPNoError(v.Address)) { + + w := peerScore(v, now) + if w < 1 { + w = 1 // minimum weight 1 so all eligible peers have a chance + } + totalWeight += w + if globals.Global_Random.Float64()*totalWeight < w { + best = v + } } } - // if we donot have any white listed, choose from the greylist + if best != nil { + best.ConnectAfter = now + 10 // minimum 10 secs gap + return best + } + + // Pass 2: uniform random among eligible greylist peers (no latency data) + var greyBest *Peer + var greyCount float64 for _, v := range peer_map { - if uint64(time.Now().Unix()) > v.BlacklistBefore && // if ip is blacklisted skip it - uint64(time.Now().Unix()) > v.ConnectAfter && - !IsAddressConnected(ParseIPNoError(v.Address)) && !v.Whitelist && !IsAddressInBanList(ParseIPNoError(v.Address)) { - v.ConnectAfter = uint64(time.Now().UTC().Unix()) + 10 // minimum 10 secs gap - return v + if now > v.BlacklistBefore && + now > v.ConnectAfter && + !IsAddressConnected(ParseIPNoError(v.Address)) && + !v.Whitelist && + !IsAddressInBanList(ParseIPNoError(v.Address)) { + + greyCount++ + if globals.Global_Random.Float64()*greyCount < 1 { + greyBest = v + } } } + if greyBest != nil { + greyBest.ConnectAfter = now + 10 // minimum 10 secs gap + return greyBest + } return nil // if no peer found, return nil } diff --git a/p2p/peer_pool_test.go b/p2p/peer_pool_test.go new file mode 100644 index 00000000..caa6d2f1 --- /dev/null +++ b/p2p/peer_pool_test.go @@ -0,0 +1,203 @@ +package p2p + +import ( + "math" + "testing" +) + +func TestPeerScore(t *testing.T) { + now := uint64(1_000_000_000) // fixed epoch for deterministic tests + + tests := []struct { + name string + peer *Peer + now uint64 + expected float64 + tolerance float64 + }{ + { + name: "zero peer — all fields zero", + peer: &Peer{}, + now: now, + expected: 0.0, + tolerance: 0.01, + }, + { + name: "5 successes, no latency data", + peer: &Peer{SuccessCount: 5}, + now: now, + expected: 50.0, + tolerance: 0.01, + }, + { + name: "3 failures only", + peer: &Peer{FailCount: 3}, + now: now, + expected: -150.0, + tolerance: 0.01, + }, + { + name: "5 successes, 10ms latency", + peer: &Peer{ + SuccessCount: 5, + LastLatency: 10_000_000, // 10ms in ns + LastMeasured: now - 60, // 1 min ago + }, + now: now, + expected: 50.0 + 10000.0/11.0, // ≈ 959.09 + tolerance: 0.01, + }, + { + name: "5 successes, 100ms latency", + peer: &Peer{ + SuccessCount: 5, + LastLatency: 100_000_000, // 100ms in ns + LastMeasured: now - 60, + }, + now: now, + expected: 50.0 + 10000.0/101.0, // ≈ 149.01 + tolerance: 0.01, + }, + { + name: "5 successes, 200ms latency", + peer: &Peer{ + SuccessCount: 5, + LastLatency: 200_000_000, // 200ms in ns + LastMeasured: now - 60, + }, + now: now, + expected: 50.0 + 10000.0/201.0, // ≈ 99.75 + tolerance: 0.01, + }, + { + name: "5 successes, TTL expired (25h old)", + peer: &Peer{ + SuccessCount: 5, + LastLatency: 10_000_000, + LastMeasured: now - 25*3600, // 25 hours ago + }, + now: now, + expected: 50.0, // no latency bonus (age >= 24h) + tolerance: 0.01, + }, + { + name: "5 successes, TTL boundary (exactly 24h)", + peer: &Peer{ + SuccessCount: 5, + LastLatency: 10_000_000, + LastMeasured: now - 24*3600, // exactly 24 hours + }, + now: now, + expected: 50.0, // no bonus: age < 86400 is false + tolerance: 0.01, + }, + { + name: "5 successes, TTL just barely valid (24h - 1s)", + peer: &Peer{ + SuccessCount: 5, + LastLatency: 10_000_000, + LastMeasured: now - (24*3600 - 1), // 23h59m59s ago + }, + now: now, + expected: 50.0 + 10000.0/11.0, // ≈ 959.09 + tolerance: 0.01, + }, + { + name: "5 successes, 1ms latency (very fast)", + peer: &Peer{ + SuccessCount: 5, + LastLatency: 1_000_000, // 1ms in ns + LastMeasured: now - 60, + }, + now: now, + expected: 50.0 + 5000.0, // 10000/(1+1) = 5000, total = 5050.0 + tolerance: 0.01, + }, + { + name: "1 success, 5 failures, 10ms latency (negative base)", + peer: &Peer{ + SuccessCount: 1, + FailCount: 5, + LastLatency: 10_000_000, + LastMeasured: now - 60, + }, + now: now, + expected: -240.0 + 10000.0/11.0, // -240 + 909.09 = 669.09 + tolerance: 0.01, + }, + { + name: "zero latency guard (LastLatency=0, LastMeasured>0)", + peer: &Peer{ + SuccessCount: 5, + LastLatency: 0, + LastMeasured: now - 60, + }, + now: now, + expected: 50.0, // no bonus: LastLatency == 0 skips block + tolerance: 0.01, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := peerScore(tt.peer, tt.now) + diff := math.Abs(got - tt.expected) + if diff > tt.tolerance { + t.Errorf("peerScore() = %.10f, want ≈ %.2f (diff = %.10f)", got, tt.expected, diff) + } + }) + } +} + +func TestPeerScoreRelativeOrdering(t *testing.T) { + now := uint64(1_000_000_000) + + fastPeer := &Peer{ + SuccessCount: 5, + LastLatency: 10_000_000, // 10ms + LastMeasured: now - 60, + } + slowPeer := &Peer{ + SuccessCount: 5, + LastLatency: 100_000_000, // 100ms + LastMeasured: now - 60, + } + verySlowPeer := &Peer{ + SuccessCount: 5, + LastLatency: 200_000_000, // 200ms + LastMeasured: now - 60, + } + noLatencyPeer := &Peer{ + SuccessCount: 5, + } + greyPeer := &Peer{} // greylist: no success, no latency + + scores := []struct { + name string + score float64 + }{ + {"fastPeer (10ms)", peerScore(fastPeer, now)}, + {"slowPeer (100ms)", peerScore(slowPeer, now)}, + {"verySlowPeer (200ms)", peerScore(verySlowPeer, now)}, + {"noLatencyPeer", peerScore(noLatencyPeer, now)}, + {"greyPeer", peerScore(greyPeer, now)}, + } + + for i := 0; i < len(scores)-1; i++ { + if scores[i].score < scores[i+1].score { + t.Errorf( + "ordering violation: %s (%.2f) < %s (%.2f) — expected descending", + scores[i].name, scores[i].score, + scores[i+1].name, scores[i+1].score, + ) + } + } + + // also verify absolute values are sensible + if fastPeerScore := peerScore(fastPeer, now); fastPeerScore < 900 { + t.Errorf("fast peer (10ms) should score > 900, got %.2f", fastPeerScore) + } + if greyPeerScore := peerScore(greyPeer, now); greyPeerScore != 0.0 { + t.Errorf("grey peer should score 0.0, got %.2f", greyPeerScore) + } +} diff --git a/p2p/rpc_handshake.go b/p2p/rpc_handshake.go index 72fb6e5b..1ebdb1dd 100644 --- a/p2p/rpc_handshake.go +++ b/p2p/rpc_handshake.go @@ -121,6 +121,7 @@ func (connection *Connection) dispatch_test_handshake() { p.LastConnected = uint64(time.Now().UTC().Unix()) Peer_Add(&p) + Peer_SetSuccess(connection.Addr.String()) } // parse delivered peer list as grey list @@ -169,9 +170,6 @@ func (c *Connection) Handshake(request Handshake_Struct, response *Handshake_Str } } } - if !c.Incoming { - Peer_SetSuccess(c.Addr.String()) - } return nil }