@@ -35,6 +35,12 @@ const (
3535
3636 // peerLimit defines limit of number of peers returned during active peer discovery.
3737 peerLimit = 60
38+
39+ // seedReconnectBackoff is the initial backoff when reconnecting to a disconnected seed peer.
40+ seedReconnectBackoff = 1 * time .Second
41+
42+ // seedReconnectMaxBackoff is the maximum backoff for seed peer reconnection attempts.
43+ seedReconnectMaxBackoff = 30 * time .Second
3844)
3945
4046// Client is a P2P client, implemented with libp2p.
@@ -56,6 +62,11 @@ type Client struct {
5662 ps * pubsub.PubSub
5763 started bool
5864
65+ ctx context.Context
66+ cancel context.CancelFunc
67+
68+ seedPeers []peer.AddrInfo
69+
5970 metrics * Metrics
6071}
6172
@@ -140,6 +151,7 @@ func (c *Client) Start(ctx context.Context) error {
140151
141152func (c * Client ) startWithHost (ctx context.Context , h host.Host ) error {
142153 c .host = h
154+ c .ctx , c .cancel = context .WithCancel (ctx )
143155 for _ , a := range c .host .Addrs () {
144156 c .logger .Info ().Str ("address" , fmt .Sprintf ("%s/p2p/%s" , a , c .host .ID ())).Msg ("listening on address" )
145157 }
@@ -170,11 +182,17 @@ func (c *Client) startWithHost(ctx context.Context, h host.Host) error {
170182 }
171183
172184 c .started = true
185+
186+ c .host .Network ().Notify (c .newDisconnectNotifee ())
187+
173188 return nil
174189}
175190
176191// Close gently stops Client.
177192func (c * Client ) Close () error {
193+ if c .cancel != nil {
194+ c .cancel ()
195+ }
178196 var err error
179197 if c .dht != nil {
180198 err = errors .Join (err , c .dht .Close ())
@@ -245,6 +263,77 @@ func (c *Client) Peers() []PeerConnection {
245263 return res
246264}
247265
266+ type disconnectNotifee struct {
267+ c * Client
268+ }
269+
270+ func (n disconnectNotifee ) Connected (_ network.Network , conn network.Conn ) {
271+ p := conn .RemotePeer ()
272+ for _ , sp := range n .c .seedPeers {
273+ if sp .ID == p {
274+ n .c .logger .Info ().Str ("peer" , p .String ()).Msg ("connected to seed peer" )
275+ return
276+ }
277+ }
278+ }
279+ func (n disconnectNotifee ) OpenedStream (_ network.Network , _ network.Stream ) {}
280+ func (n disconnectNotifee ) ClosedStream (_ network.Network , _ network.Stream ) {}
281+ func (n disconnectNotifee ) Listen (_ network.Network , _ multiaddr.Multiaddr ) {}
282+ func (n disconnectNotifee ) ListenClose (_ network.Network , _ multiaddr.Multiaddr ) {}
283+
284+ func (n disconnectNotifee ) Disconnected (_ network.Network , conn network.Conn ) {
285+ p := conn .RemotePeer ()
286+ for _ , sp := range n .c .seedPeers {
287+ if sp .ID == p {
288+ n .c .logger .Warn ().Str ("peer" , p .String ()).Msg ("disconnected from seed peer, scheduling reconnect" )
289+ go n .c .reconnectSeedPeer (sp )
290+ return
291+ }
292+ }
293+ }
294+
295+ func (c * Client ) reconnectSeedPeer (sp peer.AddrInfo ) {
296+ backoff := seedReconnectBackoff
297+ for {
298+ if c .ctx .Err () != nil {
299+ return
300+ }
301+ if c .isConnected (sp .ID ) {
302+ return
303+ }
304+
305+ err := c .host .Connect (c .ctx , sp )
306+ if err == nil {
307+ c .logger .Info ().Str ("peer" , sp .ID .String ()).Msg ("reconnected to seed peer" )
308+ return
309+ }
310+ if c .ctx .Err () != nil {
311+ return
312+ }
313+
314+ c .logger .Debug ().Str ("peer" , sp .ID .String ()).Dur ("backoff" , backoff ).Err (err ).Msg ("failed to reconnect to seed peer, retrying" )
315+ select {
316+ case <- c .ctx .Done ():
317+ return
318+ case <- time .After (backoff ):
319+ }
320+
321+ backoff *= 2
322+ if backoff > seedReconnectMaxBackoff {
323+ backoff = seedReconnectMaxBackoff
324+ }
325+ }
326+ }
327+
328+ func (c * Client ) newDisconnectNotifee () disconnectNotifee {
329+ return disconnectNotifee {c : c }
330+ }
331+
332+ // isConnected returns true if there is an active connection to the given peer.
333+ func (c * Client ) isConnected (id peer.ID ) bool {
334+ return c .host .Network ().Connectedness (id ) == network .Connected
335+ }
336+
248337func (c * Client ) listen () (host.Host , error ) {
249338 maddr , err := multiaddr .NewMultiaddr (c .conf .ListenAddress )
250339 if err != nil {
@@ -256,6 +345,7 @@ func (c *Client) listen() (host.Host, error) {
256345
257346func (c * Client ) setupDHT (ctx context.Context ) error {
258347 peers := c .parseAddrInfoList (c .conf .Peers )
348+ c .seedPeers = peers
259349 if len (peers ) == 0 {
260350 c .logger .Info ().Msg ("no peers - only listening for connections" )
261351 }
0 commit comments