update gossip mesh

Author: xzyaoi

src/internal/common/utils.go

@@ -0,0 +1,13 @@
+package common
+
+func DeduplicateStrings(input []string) []string {
+	output := []string{}
+	seen := make(map[string]struct{})
+	for _, s := range input {
+		if _, ok := seen[s]; !ok {
+			seen[s] = struct{}{}
+			output = append(output, s)
+		}
+	}
+	return output
+}

src/internal/protocol/clock.go

@@ -37,9 +37,11 @@ func StartTicker() {
 			// check if peer is still connected
 			p, error := GetPeerFromTable(peer_id.String())
 			if error == nil {
-				p.Connected = true
-				if peer_id != host.ID() && host.Network().Connectedness(peer_id) != network.Connected {
+				if host.Network().Connectedness(peer_id) == network.Connected {
+					p.Connected = true
+				} else if peer_id != host.ID() && host.Network().Connectedness(peer_id) != network.Connected {
 					// try to dial the peer, if cannot dial, then mark it as disconnected
+					common.Logger.Info("Dialing ", peer_id.String(), "...")
 					ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 					defer cancel()
 					addrInfo := libpeer.AddrInfo{ID: peer_id, Addrs: host.Peerstore().Addrs(peer_id)}
@@ -60,6 +62,8 @@ func StartTicker() {
 				value, err := json.Marshal(p)
 				if err == nil {
 					UpdateNodeTableHook(ds.NewKey(peer_id.String()), value)
+				} else {
+					common.Logger.Error("Error while marshalling peer: ", peer_id.String(), err)
 				}
 			}
 		}

src/internal/protocol/crdt.go

@@ -39,8 +39,11 @@ func GetCRDTStore() (*crdt.Datastore, context.CancelFunc) {
 
 		ipfs, err = ipfslite.New(ctx, store, nil, host, &dht, nil)
 		common.ReportError(err, "Error while creating ipfs lite node")
-
-		psub, err := pubsub.NewGossipSub(ctx, host)
+		pubsubParams := pubsub.DefaultGossipSubParams()
+		pubsubParams.D = 128
+		pubsubParams.Dlo = 16
+		pubsubParams.Dhi = 256
+		psub, err := pubsub.NewGossipSub(ctx, host, pubsub.WithGossipSubParams(pubsubParams))
 		common.ReportError(err, "Error while creating pubsub")
 
 		topic, err := psub.Join(pubsubNet)
@@ -57,6 +60,15 @@ func GetCRDTStore() (*crdt.Datastore, context.CancelFunc) {
 					break
 				}
 				host.ConnManager().TagPeer(msg.ReceivedFrom, "keep", 100)
+				// Update LastSeen when we receive a message from a peer
+				p, gerr := GetPeerFromTable(msg.ReceivedFrom.String())
+				if gerr != nil {
+					p = Peer{ID: msg.ReceivedFrom.String()}
+				}
+				p.LastSeen = time.Now().Unix()
+				if b, merr := json.Marshal(p); merr == nil {
+					UpdateNodeTableHook(ds.NewKey(msg.ReceivedFrom.String()), b)
+				}
 			}
 		}()
 

src/internal/protocol/host.go

@@ -3,6 +3,7 @@ package protocol
 import (
 	"context"
 	"crypto/rand"
+	"encoding/json"
 	mrand "math/rand"
 	"ocf/internal/common"
 	"strconv"
@@ -21,6 +22,7 @@ import (
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/libp2p/go-libp2p/core/routing"
 	rcmgr "github.com/libp2p/go-libp2p/p2p/host/resource-manager"
+	connmgr "github.com/libp2p/go-libp2p/p2p/net/connmgr"
 	"github.com/libp2p/go-libp2p/p2p/security/noise"
 	libp2ptls "github.com/libp2p/go-libp2p/p2p/security/tls"
 	"github.com/spf13/viper"
@@ -52,15 +54,17 @@ func GetP2PNode(ds datastore.Batching) (host.Host, dualdht.DHT) {
 }
 
 func newHost(ctx context.Context, seed int64, ds datastore.Batching) (host.Host, error) {
-	// connmgr, err := connmgr.NewConnManager(
-	// 	50,
-	// 	500,
-	// 	connmgr.WithGracePeriod(time.Minute),
-	// )
-	// if err != nil {
-	// 	common.Logger.Error("Error while creating connection manager: %v", err)
-	// }
 	var err error
+	// Connection manager: maintain a larger pool of connections so we can exceed
+	// the pubsub mesh degree and keep more peers around.
+	cm, err := connmgr.NewConnManager(
+		100, // Low watermark
+		800, // High watermark
+		connmgr.WithGracePeriod(5*time.Minute),
+	)
+	if err != nil {
+		common.Logger.Error("Error while creating connection manager: ", err)
+	}
 	var priv crypto.PrivKey
 	// try to load the private key from file
 	if seed == 0 {
@@ -103,9 +107,9 @@ func newHost(ctx context.Context, seed int64, ds datastore.Batching) (host.Host,
 		System: systemLimits,
 		// Keep default peer limits but increase them slightly
 		PeerDefault: rcmgr.ResourceLimits{
-			ConnsInbound:  16, // Allow more connections per peer
-			ConnsOutbound: 16,
-			Conns:         32,
+			ConnsInbound:  512, // Allow more connections per peer
+			ConnsOutbound: 512,
+			Conns:         1024,
 		},
 	}.Build(limits)
 
@@ -119,7 +123,7 @@ func newHost(ctx context.Context, seed int64, ds datastore.Batching) (host.Host,
 		libp2p.DefaultTransports,
 		libp2p.Identity(priv),
 		libp2p.ResourceManager(mgr), // Use our custom resource manager
-		// libp2p.ConnectionManager(connmgr),
+		libp2p.ConnectionManager(cm),
 		libp2p.NATPortMap(),
 		libp2p.ListenAddrStrings(
 			"/ip4/0.0.0.0/tcp/"+viper.GetString("tcpport"),
@@ -149,9 +153,45 @@ func newHost(ctx context.Context, seed int64, ds datastore.Batching) (host.Host,
 			common.Logger.Info("Connected to peer: ", c.RemotePeer(), " Total connections: ", len(n.Conns()))
 			// On (re)connections, re-announce local services
 			go ReannounceLocalServices()
+
+			// Mark peer as connected in node table immediately
+			go func(pid peer.ID) {
+				// Avoid updating self
+				if pid == host.ID() {
+					return
+				}
+				p, err := GetPeerFromTable(pid.String())
+				if err != nil {
+					p = Peer{ID: pid.String()}
+				}
+				p.Connected = true
+				p.LastSeen = time.Now().Unix()
+				if b, e := json.Marshal(p); e == nil {
+					UpdateNodeTableHook(datastore.NewKey(pid.String()), b)
+				} else {
+					common.Logger.Error("Failed to marshal peer on connect: ", e)
+				}
+			}(c.RemotePeer())
 		},
 		DisconnectedF: func(n network.Network, c network.Conn) {
 			common.Logger.Info("Disconnected from peer: ", c.RemotePeer(), " Total connections: ", len(n.Conns()))
+			// Mark peer as disconnected in node table immediately
+			go func(pid peer.ID) {
+				if pid == host.ID() {
+					return
+				}
+				p, err := GetPeerFromTable(pid.String())
+				if err != nil {
+					p = Peer{ID: pid.String()}
+				}
+				p.Connected = false
+				// keep LastSeen as last known good; do not bump here
+				if b, e := json.Marshal(p); e == nil {
+					UpdateNodeTableHook(datastore.NewKey(pid.String()), b)
+				} else {
+					common.Logger.Error("Failed to marshal peer on disconnect: ", e)
+				}
+			}(c.RemotePeer())
 		},
 	})
 
@@ -247,6 +287,11 @@ func ConnectedBootstraps() []string {
 			}
 		}
 	}
+	// add myself as bootstrap
+	myaddr := host.Addrs()[0].String() + "/p2p/" + host.ID().String()
+	bootstraps = append(bootstraps, myaddr)
+	// deduplicate
+	bootstraps = common.DeduplicateStrings(bootstraps)
 	return bootstraps
 }
 

src/internal/protocol/node_table.go

@@ -124,7 +124,6 @@ func UpdateNodeTableHook(key ds.Key, value []byte) {
 	common.ReportError(err, "Error while unmarshalling peer")
 	// Preserve locally computed connectivity status if we already know this peer
 	if existing, ok := table[key.String()]; ok {
-		peer.Connected = existing.Connected
 		// If LastSeen is missing in the update, keep the existing one
 		if peer.LastSeen == 0 {
 			peer.LastSeen = existing.LastSeen

tools/establish_connection.py

@@ -5,15 +5,15 @@
   -H 'Authorization: Bearer YOUR_API_KEY' \
   -H 'Content-Type: application/json' \
   -d '{
-    "model": "Qwen/Qwen3-32B",
+    "model": "swissai/apertus3-70b-15T-sft",
     "messages": [
       { "role": "system", "content": "You are a helpful assistant." },
       { "role": "user", "content": "What is the capital of France?" }
     ],
     "temperature": 0.7
   }'"""
 
-address = "QmNNRmz2etg76yFhc15cqvVPEKjCywSLSRFiN3brzYEV6u"
+address = "QmSNB58JK6TvpWpKqAQMJSmvZbzWLy5Qp9jkT8pNp9cJf5"
 cmd = command.replace("<address>", address)
 print(cmd)
 # Use subprocess to avoid shell interpretation issues