cmd/frostfs-node/morph.go

@@ -181,14 +181,6 @@ func waitNotaryDeposit(ctx context.Context, c *cfg, tx util.Uint256) error {
 }
 
 func listenMorphNotifications(ctx context.Context, c *cfg) {
-	// listenerPoolCap is a capacity of a
-	// worker pool inside the listener. It
-	// is used to prevent blocking in neo-go:
-	// the client cannot make RPC requests if
-	// the notification channel is not being
-	// read by another goroutine.
-	const listenerPoolCap = 10
-
 	var (
 		err  error
 		subs subscriber.Subscriber
@@ -208,9 +200,8 @@ func listenMorphNotifications(ctx context.Context, c *cfg) {
 	fatalOnErr(err)
 
 	lis, err := event.NewListener(event.ListenerParams{
-		Logger:             c.log,
+		Logger:     c.log,
-		Subscriber:         subs,
+		Subscriber: subs,
-		WorkerPoolCapacity: listenerPoolCap,
 	})
 	fatalOnErr(err)
 

pkg/innerring/innerring.go

@@ -407,14 +407,6 @@ func New(ctx context.Context, log *logger.Logger, cfg *viper.Viper, errChan chan
 }
 
 func createListener(ctx context.Context, cli *client.Client, p *chainParams) (event.Listener, error) {
-	// listenerPoolCap is a capacity of a
-	// worker pool inside the listener. It
-	// is used to prevent blocking in neo-go:
-	// the client cannot make RPC requests if
-	// the notification channel is not being
-	// read by another goroutine.
-	const listenerPoolCap = 10
-
 	var (
 		sub subscriber.Subscriber
 		err error
@@ -430,9 +422,8 @@ func createListener(ctx context.Context, cli *client.Client, p *chainParams) (ev
 	}
 
 	listener, err := event.NewListener(event.ListenerParams{
-		Logger:             &logger.Logger{Logger: p.log.With(zap.String("chain", p.name))},
+		Logger:     &logger.Logger{Logger: p.log.With(zap.String("chain", p.name))},
-		Subscriber:         sub,
+		Subscriber: sub,
-		WorkerPoolCapacity: listenerPoolCap,
 	})
 	if err != nil {
 		return nil, err

pkg/morph/event/listener.go

@@ -609,11 +609,6 @@ func (l *listener) RegisterBlockHandler(handler BlockHandler) {
 
 // NewListener create the notification event listener instance and returns Listener interface.
 func NewListener(p ListenerParams) (Listener, error) {
-	// defaultPoolCap is a default worker
-	// pool capacity if it was not specified
-	// via params
-	const defaultPoolCap = 10
-
 	switch {
 	case p.Logger == nil:
 		return nil, fmt.Errorf("%s: %w", newListenerFailMsg, errNilLogger)
@@ -621,12 +616,9 @@ func NewListener(p ListenerParams) (Listener, error) {
 		return nil, fmt.Errorf("%s: %w", newListenerFailMsg, errNilSubscriber)
 	}
 
-	poolCap := p.WorkerPoolCapacity
+	// The pool here must be blocking, otherwise notifications could be dropped.
-	if poolCap == 0 {
+	// The default capacity is 0, which means "infinite".
-		poolCap = defaultPoolCap
+	pool, err := ants.NewPool(p.WorkerPoolCapacity)
-	}
-
-	pool, err := ants.NewPool(poolCap, ants.WithNonblocking(true))
 	if err != nil {
 		return nil, fmt.Errorf("could not init worker pool: %w", err)
 	}