diff --git a/p2p_sync.go b/p2p_sync.go index 9660bfb..0fccf03 100644 --- a/p2p_sync.go +++ b/p2p_sync.go @@ -4,19 +4,26 @@ import ( "bufio" "context" "encoding/json" + "errors" "fmt" "io" "log" + "strings" "sync" "time" + pubsub "github.com/libp2p/go-libp2p-pubsub" "github.com/libp2p/go-libp2p/core/host" "github.com/libp2p/go-libp2p/core/network" "github.com/libp2p/go-libp2p/core/peer" + "github.com/libp2p/go-libp2p/p2p/net/swarm" ) const ( - ProtocolID = "/clset/sync/1.0.0" + ProtocolID = "/clset/sync/1.0.0" + PeerSummaryTopic = "/clset/summary/1.0.0" + discoverConnectBackoffDrift = 100 * time.Millisecond + discoverConnectRetryCount = 3 ) // SyncMessage represents the protocol messages for synchronization @@ -37,12 +44,95 @@ type Peer struct { peersMu sync.RWMutex syncInProgress map[peer.ID]bool syncMu sync.RWMutex + + ps *pubsub.PubSub + topic *pubsub.Topic + sub *pubsub.Subscription + summaryCache map[string]cachedSummary + cacheMu sync.Mutex + + summaryCh chan struct{} // channel for summary message notifications + lastSummary map[string]uint64 // last advertised tracked state + lastPub time.Time // last publish time + + // Latency tracking for peer scoring + peerLatency map[peer.ID]time.Duration + latencyMu sync.RWMutex + + config PeerConfig // configurable parameters +} + +type SummaryMessage struct { + PeerID string `json:"peer_id"` + Tracked map[string]uint64 `json:"tracked"` + Metadata map[string]string `json:"metadata"` // NEW: Peer metadata + TTL uint64 `json:"ttl"` // NEW: BestBefore timestamp +} + +type cachedSummary struct { + Msg SummaryMessage + Timestamp time.Time +} + +// PeerConfig holds configurable parameters for the Peer +type PeerConfig struct { + PeriodicSyncInterval time.Duration // Interval for periodic sync with all peers (default: 5 minutes) + ScheduledSyncInterval time.Duration // Interval for gossip-triggered sync scheduling (default: 5 seconds) + SummaryCacheTTL time.Duration // TTL for cached peer summaries (default: 10 minutes) + MinSummaryInterval time.Duration // Minimum interval between identical summary broadcasts (default: 1 minute) + EnablePeerScoring bool // Enable latency-based peer scoring for GossipSub (default: true) +} + +// DefaultPeerConfig returns the default configuration for a Peer +func DefaultPeerConfig() PeerConfig { + return PeerConfig{ + PeriodicSyncInterval: 5 * time.Minute, + ScheduledSyncInterval: 5 * time.Second, + SummaryCacheTTL: 10 * time.Minute, + MinSummaryInterval: 1 * time.Minute, + EnablePeerScoring: true, + } +} + +// PeerOption is a functional option for configuring a Peer +type PeerOption func(*PeerConfig) + +// WithPeriodicSyncInterval sets the interval for periodic sync with all peers +func WithPeriodicSyncInterval(d time.Duration) PeerOption { + return func(c *PeerConfig) { c.PeriodicSyncInterval = d } +} + +// WithScheduledSyncInterval sets the interval for gossip-triggered sync scheduling +func WithScheduledSyncInterval(d time.Duration) PeerOption { + return func(c *PeerConfig) { c.ScheduledSyncInterval = d } +} + +// WithSummaryCacheTTL sets the TTL for cached peer summaries +func WithSummaryCacheTTL(d time.Duration) PeerOption { + return func(c *PeerConfig) { c.SummaryCacheTTL = d } +} + +// WithMinSummaryInterval sets the minimum interval between identical summary broadcasts +func WithMinSummaryInterval(d time.Duration) PeerOption { + return func(c *PeerConfig) { c.MinSummaryInterval = d } +} + +// WithPeerScoring enables or disables latency-based peer scoring +func WithPeerScoring(enabled bool) PeerOption { + return func(c *PeerConfig) { c.EnablePeerScoring = enabled } } // NewP2PSync creates a P2P-enabled CRDT instance -func NewPeer(crdt *CRDT, ctx context.Context, host host.Host) (*Peer, error) { +func NewPeer(crdt *CRDT, ctx context.Context, host host.Host, opts ...PeerOption) (*Peer, error) { syncCtx, cancel := context.WithCancel(ctx) + // Apply configuration options + config := DefaultPeerConfig() + for _, opt := range opts { + opt(&config) + } + + // Create the peer struct first so we can reference it in the score function p2p := &Peer{ crdt: crdt, Host: host, @@ -50,8 +140,68 @@ func NewPeer(crdt *CRDT, ctx context.Context, host host.Host) (*Peer, error) { cancel: cancel, syncPeers: make(map[peer.ID]time.Time), syncInProgress: make(map[peer.ID]bool), + summaryCache: make(map[string]cachedSummary), + summaryCh: make(chan struct{}, 1), + lastSummary: make(map[string]uint64), + peerLatency: make(map[peer.ID]time.Duration), + config: config, + } + + // Initialize pubsub with optional peer scoring + var ps *pubsub.PubSub + var err error + + if config.EnablePeerScoring { + // Configure peer scoring to prefer low-latency peers + scoreParams := &pubsub.PeerScoreParams{ + AppSpecificScore: func(peerID peer.ID) float64 { + return p2p.peerScoreFunc(peerID) + }, + AppSpecificWeight: 1.0, + DecayInterval: time.Second, + DecayToZero: 0.01, + } + + thresholds := &pubsub.PeerScoreThresholds{ + GossipThreshold: -100, + PublishThreshold: -200, + GraylistThreshold: -500, + } + + ps, err = pubsub.NewGossipSub( + syncCtx, + host, + pubsub.WithPeerScore(scoreParams, thresholds), + ) + } else { + ps, err = pubsub.NewGossipSub(syncCtx, host) + } + + if err != nil { + cancel() + return nil, fmt.Errorf("failed to create pubsub: %w", err) } + topic, err := ps.Join(PeerSummaryTopic) + if err != nil { + cancel() + return nil, fmt.Errorf("failed to join pubsub topic: %w", err) + } + + sub, err := topic.Subscribe() + if err != nil { + cancel() + return nil, fmt.Errorf("failed to subscribe to pubsub topic: %w", err) + } + + p2p.ps = ps + p2p.topic = topic + p2p.sub = sub + + crdt.AddInsertHook(func(k string, v []byte, m CRDTKeyMeta) { p2p.notifySummary() }) + crdt.AddUpdateHook(func(k string, oldV []byte, oldM CRDTKeyMeta, newV []byte, newM CRDTKeyMeta) { p2p.notifySummary() }) + crdt.AddDeleteHook(func(k string, oldV []byte, oldM CRDTKeyMeta) { p2p.notifySummary() }) + // Set stream handler for incoming sync requests host.SetStreamHandler(ProtocolID, p2p.handleSyncStream) @@ -61,6 +211,14 @@ func NewPeer(crdt *CRDT, ctx context.Context, host host.Host) (*Peer, error) { // Start periodic sync with connected peers go p2p.periodicSync() + // Start gossip routines + go p2p.readSummaries() + go p2p.broadcastSummaries() + go p2p.scheduledSync() + + // Start TTL cleanup for expired peers + go p2p.cleanupExpiredPeers() + log.Printf("P2P CRDT node started") log.Printf("Peer ID: %s", host.ID()) log.Printf("Listening on: %v", host.Addrs()) @@ -79,6 +237,59 @@ func (p *Peer) GetCRDT() *CRDT { return p.crdt } +// recordLatency records the latency for a peer using exponential moving average +func (p *Peer) recordLatency(peerID peer.ID, duration time.Duration) { + p.latencyMu.Lock() + defer p.latencyMu.Unlock() + + // Exponential moving average with weight 0.75 on existing value + // This smooths out spikes while still adapting to changes + if existing, ok := p.peerLatency[peerID]; ok { + p.peerLatency[peerID] = (existing*3 + duration) / 4 + } else { + p.peerLatency[peerID] = duration + } +} + +// peerScoreFunc calculates a score for a peer based on its latency +// Lower latency = higher score. This function is called by GossipSub +// to preferentially select low-latency peers for the mesh network. +func (p *Peer) peerScoreFunc(peerID peer.ID) float64 { + p.latencyMu.RLock() + latency, ok := p.peerLatency[peerID] + p.latencyMu.RUnlock() + + if !ok { + return 0.0 // Unknown peer, neutral score + } + + // Lower latency = higher score + // Score = 1000 / latency_in_ms + // Examples: + // 1ms latency = score 1000 + // 10ms latency = score 100 + // 100ms latency = score 10 + // 1000ms latency = score 1 + latencyMs := float64(latency.Milliseconds()) + if latencyMs < 1.0 { + latencyMs = 1.0 // Avoid division by zero, minimum 1ms + } + + return 1000.0 / latencyMs +} + +// GetPeerLatencies returns a copy of all peer latencies for monitoring +func (p *Peer) GetPeerLatencies() map[peer.ID]time.Duration { + p.latencyMu.RLock() + defer p.latencyMu.RUnlock() + + latencies := make(map[peer.ID]time.Duration, len(p.peerLatency)) + for pid, lat := range p.peerLatency { + latencies[pid] = lat + } + return latencies +} + // handleSyncStream handles incoming sync streams from peers func (p *Peer) handleSyncStream(s network.Stream) { defer s.Close() @@ -165,6 +376,9 @@ func (p *Peer) handleSyncResponse(resp SyncMessage) { // syncWithPeer initiates a sync with a specific peer func (p *Peer) syncWithPeer(peerID peer.ID) error { + // Start measuring latency + startTime := time.Now() + // Check if sync is already in progress p.syncMu.Lock() if p.syncInProgress[peerID] { @@ -178,6 +392,10 @@ func (p *Peer) syncWithPeer(peerID peer.ID) error { p.syncMu.Lock() delete(p.syncInProgress, peerID) p.syncMu.Unlock() + + // Record the latency for this sync operation + duration := time.Since(startTime) + p.recordLatency(peerID, duration) }() // Check if we're still connected to the peer @@ -259,7 +477,7 @@ func (p *Peer) syncWithPeer(peerID peer.ID) error { // periodicSync runs periodic synchronization with all connected peers func (p *Peer) periodicSync() { - ticker := time.NewTicker(5 * time.Second) + ticker := time.NewTicker(p.config.PeriodicSyncInterval) defer ticker.Stop() for { @@ -325,14 +543,11 @@ func (p *Peer) HandlePeerFound(pi peer.AddrInfo) { return // Already connected to this peer } - log.Printf("Discovered new peer: %s", pi.ID) - - // Create a context with timeout for connection - connectCtx, cancel := context.WithTimeout(p.ctx, 15*time.Second) - defer cancel() + log.Printf("Discovered new peer: %s, %+v", pi.ID, pi) // Connect to the peer - if err := p.Host.Connect(connectCtx, pi); err != nil { + err := p.discoveryConnect(pi, discoverConnectRetryCount) + if err != nil { log.Printf("Failed to connect to discovered peer %s: %v", pi.ID, err) return } @@ -353,6 +568,48 @@ func (p *Peer) HandlePeerFound(pi peer.AddrInfo) { }() } +const unexpectedHandshake = "received unexpected handshake message of type *tls.clientHelloMsg when waiting for *tls.serverHelloMsg" + +func (p *Peer) discoveryConnect(pi peer.AddrInfo, maxRetries int) error { + // To avoid simultaneous connection attempts, add a small drift based on peer IDs + backoffDrift := discoverConnectBackoffDrift + if p.Host.ID().String() > pi.ID.String() { + backoffDrift = 0 + } + + // Initial wait before first attempt. It's more effective to wait here than after the first failure, + // as default libp2p backoff is quite long. + if backoffDrift > 0 { + time.Sleep(backoffDrift) + } + + for attempt := 0; attempt < maxRetries; attempt++ { + connectCtx, cancel := context.WithTimeout(p.ctx, 15*time.Second) + defer cancel() + + backoffTime := swarm.BackoffBase + swarm.BackoffCoef*time.Duration(attempt*attempt) + backoffDrift + + err := p.Host.Connect(connectCtx, pi) + switch { + case errors.Is(err, swarm.ErrDialBackoff): + // Backoff requested by libp2p, wait and retry. + log.Printf("Backoff connecting to peer %s, retrying in %s", pi.ID, backoffTime) + time.Sleep(backoffTime) + continue + case err != nil && strings.Contains(err.Error(), unexpectedHandshake): + // Simultaneous connection attempt detected, wait and retry. + log.Printf("Simultaneous connection attempt with peer %s, retrying in %s", pi.ID, backoffTime) + time.Sleep(backoffTime) + continue + case err != nil: + // Other connection error, exit function. + return fmt.Errorf("failed to connect to peer %s: %w", pi.ID, err) + } + } + + return nil +} + // ManualConnect allows manually connecting to a peer by multiaddr func (p *Peer) ManualConnect(addr string) error { maddr, err := peer.AddrInfoFromString(addr) @@ -419,3 +676,298 @@ func (p *Peer) SyncNow() { }(peerID) } } + +// broadcastSummaries periodically publishes CRDT summaries via pubsub. +// It sends updates when the local CRDT state changes, avoiding redundant or too frequent broadcasts. +func (p *Peer) broadcastSummaries() { + for { + select { + case <-p.ctx.Done(): + return + case <-p.summaryCh: + tracked := p.crdt.GetTrackedPeers() + + // Get local metadata + p.crdt.metadataMu.RLock() + metadata := make(map[string]string) + for k, v := range p.crdt.localMetadata { + metadata[k] = v + } + p.crdt.metadataMu.RUnlock() + + // Calculate TTL (BestBefore timestamp) + ttl := uint64(time.Now().Add(p.crdt.peerTTL).Unix()) + + // Avoid re-sending identical state too often + if equalTracked(tracked, p.lastSummary) && time.Since(p.lastPub) < p.config.MinSummaryInterval { + continue + } + p.lastSummary = tracked + p.lastPub = time.Now() + + msg := SummaryMessage{ + PeerID: p.Host.ID().String(), + Tracked: tracked, + Metadata: metadata, + TTL: ttl, + } + data, err := json.Marshal(msg) + if err != nil { + log.Printf("Failed to marshal summary: %v", err) + continue + } + if err := p.topic.Publish(p.ctx, data); err != nil { + log.Printf("Failed to publish summary: %v", err) + } + } + } +} + +// equalTracked compares two Tracked Peers maps and returns true if they are identical. +// It is used to avoid re-broadcasting identical CRDT summaries. +func equalTracked(a, b map[string]uint64) bool { + if len(a) != len(b) { + return false + } + for k, v := range a { + if b[k] != v { + return false + } + } + return true +} + +// notifySummary signals that the CRDT state has changed, prompting a new summary broadcast. +// It ensures that only one signal is queued to prevent redundant triggers. +func (p *Peer) notifySummary() { + select { + case p.summaryCh <- struct{}{}: + default: + // drop if already signalled + } +} + +// readSummaries listens for summary messages from other peers over pubsub. +// Received summaries are cached for later evaluation by the sync scheduler (the scheduledSync function). +func (p *Peer) readSummaries() { + for { + m, err := p.sub.Next(p.ctx) + if err != nil { + // If context is cancelled, exit gracefully + if p.ctx.Err() != nil { + log.Printf("Summary reader shutting down: %v", p.ctx.Err()) + return + } + // For other errors, log and continue + log.Printf("Error reading summary: %v", err) + continue + } + if m.ReceivedFrom == p.Host.ID() { + continue + } + + var msg SummaryMessage + if err := json.Unmarshal(m.Data, &msg); err != nil { + continue + } + + // Update peer metadata + p.crdt.metadataMu.Lock() + changed := false + + existing := p.crdt.peerMetadata[msg.PeerID] + if existing == nil { + p.crdt.peerMetadata[msg.PeerID] = &PeerMetadata{ + BestBefore: msg.TTL, + Metadata: msg.Metadata, + } + changed = true + } else { + // Update if changed + if existing.BestBefore != msg.TTL || + !mapsEqual(existing.Metadata, msg.Metadata) { + existing.BestBefore = msg.TTL + existing.Metadata = msg.Metadata + changed = true + } + } + p.crdt.metadataMu.Unlock() + + // Trigger membership hook if metadata changed + if changed { + p.crdt.runMembershipHooks() + } + + // Cache summary for sync scheduling + p.cacheMu.Lock() + p.summaryCache[msg.PeerID] = cachedSummary{ + Msg: msg, + Timestamp: time.Now(), + } + p.cacheMu.Unlock() + } +} + +// mapsEqual compares two string maps for equality +func mapsEqual(a, b map[string]string) bool { + if len(a) != len(b) { + return false + } + for k, v := range a { + if b[k] != v { + return false + } + } + return true +} + +// scheduledSync periodically evaluates cached peer summaries and chooses a peer to synchronize with. +// It selects the peer with the largest sequence number gap in tracked states. +func (p *Peer) scheduledSync() { + ticker := time.NewTicker(p.config.ScheduledSyncInterval) + defer ticker.Stop() + + for { + select { + case <-p.ctx.Done(): + return + case <-ticker.C: + p.cacheMu.Lock() + cacheCopy := make(map[string]cachedSummary) + for k, v := range p.summaryCache { + cacheCopy[k] = v + } + p.cacheMu.Unlock() + + local := p.crdt.GetTrackedPeers() + bestPeer := "" + maxGap := uint64(0) + + for pid, summary := range cacheCopy { + gap := calcGap(local, summary.Msg.Tracked) + if gap > maxGap { + maxGap = gap + bestPeer = pid + } + } + + if bestPeer != "" { + if pid, err := peer.Decode(bestPeer); err == nil { + // Check if sync is already in progress before spawning goroutine + p.syncMu.RLock() + inProgress := p.syncInProgress[pid] + p.syncMu.RUnlock() + if !inProgress { + go p.syncWithPeer(pid) + } + } + } + + p.cleanupSummaryCache() + } + } +} + +// calcGap computes the total sequence number gap between two Tracked Peers maps. +// It is used to quantify how ahead a remote peer is compared to the local peer. +func calcGap(local, remote map[string]uint64) uint64 { + var gap uint64 + for pid, seq := range remote { + if local[pid] < seq { + gap += seq - local[pid] + } + } + return gap +} + +// cleanupSummaryCache removes stale summary entries from the peer summary cache, +// i.e., entries that have not been updated for longer than the configured TTL. +func (p *Peer) cleanupSummaryCache() { + p.cacheMu.Lock() + defer p.cacheMu.Unlock() + + now := time.Now() + for pid, entry := range p.summaryCache { + if now.Sub(entry.Timestamp) > p.config.SummaryCacheTTL { + delete(p.summaryCache, pid) + } + } +} + +// LogMeshStats logs mesh composition statistics, showing distribution of +// local vs remote peers based on latency. This is useful for verifying that +// peer scoring is working correctly and preferring low-latency peers. +func (p *Peer) LogMeshStats() { + p.latencyMu.RLock() + defer p.latencyMu.RUnlock() + + if len(p.peerLatency) == 0 { + log.Printf("Mesh stats: No peer latency data available") + return + } + + var localPeers, remotePeers int + var avgLocalLatency, avgRemoteLatency time.Duration + const localThreshold = 10 * time.Millisecond + + for _, latency := range p.peerLatency { + if latency < localThreshold { + localPeers++ + avgLocalLatency += latency + } else { + remotePeers++ + avgRemoteLatency += latency + } + } + + // Calculate averages + if localPeers > 0 { + avgLocalLatency = avgLocalLatency / time.Duration(localPeers) + } + if remotePeers > 0 { + avgRemoteLatency = avgRemoteLatency / time.Duration(remotePeers) + } + + log.Printf("Mesh composition: %d local peers (avg %v), %d remote peers (avg %v)", + localPeers, avgLocalLatency, remotePeers, avgRemoteLatency) + + // Log individual peer latencies for detailed analysis + if localPeers+remotePeers <= 10 { + for peerID, latency := range p.peerLatency { + score := p.peerScoreFunc(peerID) + log.Printf(" Peer %s: latency=%v, score=%.2f", peerID.ShortString(), latency, score) + } + } +} + +// cleanupExpiredPeers periodically removes peers whose TTL has expired. +// This runs in a background goroutine and triggers the membership hook when peers are removed. +func (p *Peer) cleanupExpiredPeers() { + ticker := time.NewTicker(1 * time.Minute) + defer ticker.Stop() + + for { + select { + case <-p.ctx.Done(): + return + case <-ticker.C: + p.crdt.metadataMu.Lock() + now := uint64(time.Now().Unix()) + removed := false + + for peerID, meta := range p.crdt.peerMetadata { + if meta.BestBefore < now { + delete(p.crdt.peerMetadata, peerID) + removed = true + log.Printf("Removed expired peer: %s (TTL expired)", peerID) + } + } + p.crdt.metadataMu.Unlock() + + // Trigger membership hook if any peers were removed + if removed { + p.crdt.runMembershipHooks() + } + } + } +} diff --git a/p2p_sync_test.go b/p2p_sync_test.go index 4813852..eeedb9b 100644 --- a/p2p_sync_test.go +++ b/p2p_sync_test.go @@ -148,3 +148,127 @@ func TestPeerTracking(t *testing.T) { assert.Contains(t, crdt2.GetTrackedPeers(), "p1") assert.True(t, crdt2.GetTrackedPeers()["p1"] > 0) } + +func TestLatencyTracking(t *testing.T) { + crdt1 := createTestCRDT(t, "peer1") + crdt2 := createTestCRDT(t, "peer2") + + p2p1 := createPeer(t, crdt1, 15001) + defer p2p1.Close() + + p2p2 := createPeer(t, crdt2, 15002) + defer p2p2.Close() + + time.Sleep(1 * time.Second) + + // Connect peers + connectAddr := p2p1.Host.Addrs()[0].String() + "/p2p/" + p2p1.Host.ID().String() + require.NoError(t, p2p2.ManualConnect(connectAddr)) + + // Set some data and sync + require.NoError(t, crdt1.Set("key1", []byte("value1"))) + p2p2.SyncNow() + time.Sleep(2 * time.Second) + + // Check that latency was recorded + latencies := p2p2.GetPeerLatencies() + assert.Contains(t, latencies, p2p1.Host.ID()) + assert.Greater(t, latencies[p2p1.Host.ID()], time.Duration(0)) + + t.Logf("Recorded latency for peer %s: %v", p2p1.Host.ID().ShortString(), latencies[p2p1.Host.ID()]) +} + +func TestPeerScoring(t *testing.T) { + crdt1 := createTestCRDT(t, "peer1") + crdt2 := createTestCRDT(t, "peer2") + + // Enable peer scoring explicitly + p2p1 := createPeer(t, crdt1, 16001) + defer p2p1.Close() + + p2p2 := createPeer(t, crdt2, 16002) + defer p2p2.Close() + + time.Sleep(1 * time.Second) + + // Connect and sync + connectAddr := p2p1.Host.Addrs()[0].String() + "/p2p/" + p2p1.Host.ID().String() + require.NoError(t, p2p2.ManualConnect(connectAddr)) + + // Perform multiple syncs to build latency data + for i := 0; i < 3; i++ { + require.NoError(t, crdt1.Set(fmt.Sprintf("key%d", i), []byte(fmt.Sprintf("value%d", i)))) + p2p2.SyncNow() + time.Sleep(1 * time.Second) + } + + // Verify latency tracking and scoring + latencies := p2p2.GetPeerLatencies() + require.Contains(t, latencies, p2p1.Host.ID()) + + latency := latencies[p2p1.Host.ID()] + t.Logf("Average latency: %v", latency) + + // Log mesh stats + p2p2.LogMeshStats() +} + +func TestPeerScoringDisabled(t *testing.T) { + crdt1 := createTestCRDT(t, "peer1") + crdt2 := createTestCRDT(t, "peer2") + + privateKey1, _, err := crypto.GenerateKeyPairWithReader(crypto.RSA, 2048, rand.Reader) + require.NoError(t, err) + + privateKey2, _, err := crypto.GenerateKeyPairWithReader(crypto.RSA, 2048, rand.Reader) + require.NoError(t, err) + + h1, err := libp2p.New( + libp2p.Identity(privateKey1), + libp2p.ListenAddrStrings("/ip4/0.0.0.0/tcp/17001"), + libp2p.DisableRelay(), + libp2p.Ping(false), + libp2p.DefaultSecurity, + libp2p.DefaultTransports, + ) + require.NoError(t, err) + + h2, err := libp2p.New( + libp2p.Identity(privateKey2), + libp2p.ListenAddrStrings("/ip4/0.0.0.0/tcp/17002"), + libp2p.DisableRelay(), + libp2p.Ping(false), + libp2p.DefaultSecurity, + libp2p.DefaultTransports, + ) + require.NoError(t, err) + + // Create peers with scoring disabled + p2p1, err := clset.NewPeer(crdt1, t.Context(), h1, clset.WithPeerScoring(false)) + require.NoError(t, err) + defer p2p1.Close() + + p2p2, err := clset.NewPeer(crdt2, t.Context(), h2, clset.WithPeerScoring(false)) + require.NoError(t, err) + defer p2p2.Close() + + time.Sleep(1 * time.Second) + + // Connect and sync + connectAddr := p2p1.Host.Addrs()[0].String() + "/p2p/" + p2p1.Host.ID().String() + require.NoError(t, p2p2.ManualConnect(connectAddr)) + + require.NoError(t, crdt1.Set("key1", []byte("value1"))) + p2p2.SyncNow() + time.Sleep(2 * time.Second) + + // Verify sync still works without scoring + val, exists, err := crdt2.Get("key1") + require.NoError(t, err) + assert.True(t, exists) + assert.Equal(t, []byte("value1"), val) + + // Latency should still be tracked + latencies := p2p2.GetPeerLatencies() + assert.Contains(t, latencies, p2p1.Host.ID()) +}