just add check to DisconnectedF

op-node/p2p/node.go

@@ -11,7 +11,6 @@ import (
 	"github.com/hashicorp/go-multierror"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/libp2p/go-libp2p/core/connmgr"
-	"github.com/libp2p/go-libp2p/core/event"
 	"github.com/libp2p/go-libp2p/core/host"
 	p2pmetrics "github.com/libp2p/go-libp2p/core/metrics"
 	"github.com/libp2p/go-libp2p/core/network"
@@ -108,20 +107,17 @@ func (n *NodeP2P) init(resourcesCtx context.Context, rollupCfg *rollup.Config, l
 		// Activate the P2P req-resp sync if enabled by feature-flag.
 		if setup.ReqRespSyncEnabled() {
 			n.syncCl = NewSyncClient(log, rollupCfg, n.host.NewStream, gossipIn.OnUnsafeL2Payload, metrics, n.appScorer)
-			subscribe, err := n.host.EventBus().Subscribe(&event.EvtPeerConnectednessChanged{})
-			if err != nil {
-				return fmt.Errorf("failed to subscribe peer connectedness changed event: %w", err)
-			}
-			go func() {
-				for evt := range subscribe.Out() {
-					evto := evt.(event.EvtPeerConnectednessChanged)
-					if evto.Connectedness == network.Connected {
-						n.syncCl.AddPeer(evto.Peer)
-					} else if evto.Connectedness == network.NotConnected {
-						n.syncCl.RemovePeer(evto.Peer)
-					}
-				}
-			}()
+			n.host.Network().Notify(&network.NotifyBundle{
+				ConnectedF: func(nw network.Network, conn network.Conn) {
+					n.syncCl.AddPeer(conn.RemotePeer())
+				},
+				DisconnectedF: func(nw network.Network, conn network.Conn) {
+					// only when no connection is available, we can remove the peer
+					if nw.Connectedness(conn.RemotePeer()) == network.NotConnected {
+						n.syncCl.RemovePeer(conn.RemotePeer())
+					}
+				},
+			})
 			n.syncCl.Start()
 			// the host may already be connected to peers, add them all to the sync client
 			for _, peerID := range n.host.Network().Peers() {

op-node/p2p/sync_test.go

@@ -7,7 +7,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/libp2p/go-libp2p/core/event"
 	"github.com/libp2p/go-libp2p/core/host"
 	"github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/peer"
@@ -290,7 +289,7 @@ func TestMultiPeerSync(t *testing.T) {
 	}
 }
 
-func TestUseEvtPeerConnectednessChangedEvent(t *testing.T) {
+func TestNetworkNotifyAddPeerAndRemovePeer(t *testing.T) {
 	t.Parallel()
 	log := testlog.Logger(t, log.LvlDebug)
 
@@ -308,20 +307,17 @@ func TestUseEvtPeerConnectednessChangedEvent(t *testing.T) {
 	syncCl := NewSyncClient(log, cfg, hostA.NewStream, func(ctx context.Context, from peer.ID, payload *eth.ExecutionPayload) error {
 		return nil
 	}, metrics.NoopMetrics, &NoopApplicationScorer{})
-	subscribe, err := hostA.EventBus().Subscribe(&event.EvtPeerConnectednessChanged{})
-	require.NoError(t, err, "subscribe peerConnectednessChanged fail")
-	go func() {
-		for evt := range subscribe.Out() {
-			evto := evt.(event.EvtPeerConnectednessChanged)
-			if evto.Connectedness == network.Connected {
-				log.Info("event: connect peer", "peer", evto.Peer)
-				syncCl.AddPeer(evto.Peer)
-			} else if evto.Connectedness == network.NotConnected {
-				log.Info("event: disconnect peer", "peer", evto.Peer)
-				syncCl.RemovePeer(evto.Peer)
-			}
+	hostA.Network().Notify(&network.NotifyBundle{
+		ConnectedF: func(nw network.Network, conn network.Conn) {
+			syncCl.AddPeer(conn.RemotePeer())
+		},
+		DisconnectedF: func(nw network.Network, conn network.Conn) {
+			// only when no connection is available, we can remove the peer
+			if nw.Connectedness(conn.RemotePeer()) == network.NotConnected {
+				syncCl.RemovePeer(conn.RemotePeer())
 			}
-	}()
+		},
+	})
 	syncCl.Start()
 
 	err = hostA.Connect(context.Background(), peer.AddrInfo{ID: hostB.ID(), Addrs: hostB.Addrs()})