refactor!(blankhost): make blankhost more amenable to a lifecycle

Author: MarcoPolo

p2p/host/blank/blank.go

@@ -14,30 +14,33 @@ import (
 	"github.com/libp2p/go-libp2p/core/peerstore"
 	"github.com/libp2p/go-libp2p/core/protocol"
 	"github.com/libp2p/go-libp2p/core/record"
-	"github.com/libp2p/go-libp2p/p2p/host/eventbus"
-
 	logging "github.com/libp2p/go-libp2p/gologshim"
+	"github.com/libp2p/go-libp2p/p2p/host/eventbus"
 
 	ma "github.com/multiformats/go-multiaddr"
 	mstream "github.com/multiformats/go-multistream"
 )
 
 var log = logging.Logger("blankhost")
 
-// BlankHost is the thinnest implementation of the host.Host interface
+// BlankHost is a thin implementation of the host.Host interface
 type BlankHost struct {
-	n        network.Network
-	mux      *mstream.MultistreamMuxer[protocol.ID]
-	cmgr     connmgr.ConnManager
-	eventbus event.Bus
-	emitters struct {
+	N       network.Network
+	M       *mstream.MultistreamMuxer[protocol.ID]
+	E       event.Bus
+	ConnMgr connmgr.ConnManager
+	// SkipInitSignedRecord is a flag to skip the initialization of a signed record for the host
+	SkipInitSignedRecord bool
+	emitters             struct {
 		evtLocalProtocolsUpdated event.Emitter
 	}
+	onStop []func() error
 }
 
 type config struct {
-	cmgr     connmgr.ConnManager
-	eventBus event.Bus
+	cmgr                 connmgr.ConnManager
+	eventBus             event.Bus
+	skipInitSignedRecord bool
 }
 
 type Option = func(cfg *config)
@@ -54,6 +57,12 @@ func WithEventBus(eventBus event.Bus) Option {
 	}
 }
 
+func SkipInitSignedRecord() Option {
+	return func(cfg *config) {
+		cfg.skipInitSignedRecord = true
+	}
+}
+
 func NewBlankHost(n network.Network, options ...Option) *BlankHost {
 	cfg := config{
 		cmgr: &connmgr.NullConnMgr{},
@@ -63,36 +72,72 @@ func NewBlankHost(n network.Network, options ...Option) *BlankHost {
 	}
 
 	bh := &BlankHost{
-		n:        n,
-		cmgr:     cfg.cmgr,
-		mux:      mstream.NewMultistreamMuxer[protocol.ID](),
-		eventbus: cfg.eventBus,
+		N:       n,
+		ConnMgr: cfg.cmgr,
+		M:       mstream.NewMultistreamMuxer[protocol.ID](),
+		E:       cfg.eventBus,
+
+		SkipInitSignedRecord: cfg.skipInitSignedRecord,
 	}
-	if bh.eventbus == nil {
-		bh.eventbus = eventbus.NewBus(eventbus.WithMetricsTracer(eventbus.NewMetricsTracer()))
+
+	if err := bh.Start(); err != nil {
+		log.Error("error creating blank host", "err", err)
+		return nil
+	}
+
+	return bh
+}
+
+func (bh *BlankHost) Start() error {
+	if bh.E == nil {
+		bh.E = eventbus.NewBus(eventbus.WithMetricsTracer(eventbus.NewMetricsTracer()))
 	}
 
 	// subscribe the connection manager to network notifications (has no effect with NullConnMgr)
-	n.Notify(bh.cmgr.Notifee())
+	notifee := bh.ConnMgr.Notifee()
+	bh.N.Notify(notifee)
+	bh.onStop = append(bh.onStop, func() error {
+		bh.N.StopNotify(notifee)
+		return nil
+	})
 
 	var err error
-	if bh.emitters.evtLocalProtocolsUpdated, err = bh.eventbus.Emitter(&event.EvtLocalProtocolsUpdated{}); err != nil {
-		return nil
+	if bh.emitters.evtLocalProtocolsUpdated, err = bh.E.Emitter(&event.EvtLocalProtocolsUpdated{}); err != nil {
+		return err
 	}
+	bh.onStop = append(bh.onStop, func() error {
+		bh.emitters.evtLocalProtocolsUpdated.Close()
+		return nil
+	})
 
-	n.SetStreamHandler(bh.newStreamHandler)
+	bh.N.SetStreamHandler(bh.newStreamHandler)
+	bh.onStop = append(bh.onStop, func() error {
+		bh.N.SetStreamHandler(func(s network.Stream) { s.Reset() })
+		return nil
+	})
 
 	// persist a signed peer record for self to the peerstore.
-	if err := bh.initSignedRecord(); err != nil {
-		log.Error("error creating blank host", "err", err)
-		return nil
+	if !bh.SkipInitSignedRecord {
+		if err := bh.initSignedRecord(); err != nil {
+			log.Error("error creating blank host", "err", err)
+			return err
+		}
 	}
 
-	return bh
+	return nil
+}
+
+func (bh *BlankHost) Stop() error {
+	var err error
+	for _, f := range bh.onStop {
+		err = errors.Join(err, f())
+	}
+	bh.onStop = nil
+	return err
 }
 
 func (bh *BlankHost) initSignedRecord() error {
-	cab, ok := peerstore.GetCertifiedAddrBook(bh.n.Peerstore())
+	cab, ok := peerstore.GetCertifiedAddrBook(bh.N.Peerstore())
 	if !ok {
 		log.Error("peerstore does not support signed records")
 		return errors.New("peerstore does not support signed records")
@@ -114,7 +159,7 @@ func (bh *BlankHost) initSignedRecord() error {
 var _ host.Host = (*BlankHost)(nil)
 
 func (bh *BlankHost) Addrs() []ma.Multiaddr {
-	addrs, err := bh.n.InterfaceListenAddresses()
+	addrs, err := bh.N.InterfaceListenAddresses()
 	if err != nil {
 		log.Debug("error retrieving network interface addrs", "err", err)
 		return nil
@@ -124,14 +169,18 @@ func (bh *BlankHost) Addrs() []ma.Multiaddr {
 }
 
 func (bh *BlankHost) Close() error {
-	return bh.n.Close()
+	var err error
+	if bh.onStop != nil {
+		err = bh.Stop()
+	}
+	return errors.Join(err, bh.N.Close())
 }
 
 func (bh *BlankHost) Connect(ctx context.Context, ai peer.AddrInfo) error {
 	// absorb addresses into peerstore
 	bh.Peerstore().AddAddrs(ai.ID, ai.Addrs, peerstore.TempAddrTTL)
 
-	cs := bh.n.ConnsToPeer(ai.ID)
+	cs := bh.N.ConnsToPeer(ai.ID)
 	if len(cs) > 0 {
 		return nil
 	}
@@ -144,15 +193,15 @@ func (bh *BlankHost) Connect(ctx context.Context, ai peer.AddrInfo) error {
 }
 
 func (bh *BlankHost) Peerstore() peerstore.Peerstore {
-	return bh.n.Peerstore()
+	return bh.N.Peerstore()
 }
 
 func (bh *BlankHost) ID() peer.ID {
-	return bh.n.LocalPeer()
+	return bh.N.LocalPeer()
 }
 
 func (bh *BlankHost) NewStream(ctx context.Context, p peer.ID, protos ...protocol.ID) (network.Stream, error) {
-	s, err := bh.n.NewStream(ctx, p)
+	s, err := bh.N.NewStream(ctx, p)
 	if err != nil {
 		return nil, fmt.Errorf("failed to open stream: %w", err)
 	}
@@ -204,7 +253,7 @@ func (bh *BlankHost) SetStreamHandlerMatch(pid protocol.ID, m func(protocol.ID)
 func (bh *BlankHost) newStreamHandler(s network.Stream) {
 	protoID, handle, err := bh.Mux().Negotiate(s)
 	if err != nil {
-		log.Info("protocol negotiation failed", "err", err)
+		log.Error("protocol negotiation failed", "err", err)
 		s.Reset()
 		return
 	}
@@ -216,18 +265,18 @@ func (bh *BlankHost) newStreamHandler(s network.Stream) {
 
 // TODO: i'm not sure this really needs to be here
 func (bh *BlankHost) Mux() protocol.Switch {
-	return bh.mux
+	return bh.M
 }
 
 // TODO: also not sure this fits... Might be better ways around this (leaky abstractions)
 func (bh *BlankHost) Network() network.Network {
-	return bh.n
+	return bh.N
 }
 
 func (bh *BlankHost) ConnManager() connmgr.ConnManager {
-	return bh.cmgr
+	return bh.ConnMgr
 }
 
 func (bh *BlankHost) EventBus() event.Bus {
-	return bh.eventbus
+	return bh.E
 }