Fix: LDK remove mutex

ldk.go

@@ -10,7 +10,6 @@ import (
 	"sort"
 	"strconv"
 	"strings"
-	"sync"
 	"time"
 
 	"github.com/getAlby/ldk-node-go/ldk_node"
@@ -20,12 +19,12 @@ import (
 )
 
 type LDKService struct {
-	svc                       *Service
-	workdir                   string
-	node                      *ldk_node.LdkNode
-	cancelLdkEventListenerCtx context.CancelFunc
-	subscribeLdkEvents        func() chan ldk_node.Event
-	unsubscribeLdkEvents      func(chan ldk_node.Event)
+	svc                 *Service
+	workdir             string
+	node                *ldk_node.LdkNode
+	ldkEventBroadcaster LDKEventBroadcaster
+	ctx                 context.Context
+	cancel              context.CancelFunc
 }
 
 func NewLDKService(svc *Service, mnemonic, workDir string, network string, esploraServer string, gossipSource string) (result lnclient.LNClient, err error) {
@@ -72,43 +71,14 @@ func NewLDKService(svc *Service, mnemonic, workDir string, network string, esplo
 		return nil, err
 	}
 
-	// TODO: move this event handler code
-	ldkEventListenerCtx, cancelLdkEventListenerCtx := context.WithCancel(context.Background())
-	ldkEventHandlers := []chan ldk_node.Event{}
-	var ldkEventHandlersMutex sync.Mutex
-
-	subscribeLdkEvents := func() chan ldk_node.Event {
-		ldkEventHandler := make(chan ldk_node.Event)
-		svc.Logger.Debugf("Locking event handler mutex")
-		ldkEventHandlersMutex.Lock()
-		svc.Logger.Debugf("Locked event handler mutex")
-		ldkEventHandlers = append(ldkEventHandlers, ldkEventHandler)
-		ldkEventHandlersMutex.Unlock()
-		svc.Logger.Debugf("Unlocked event handler mutex")
-		return ldkEventHandler
-	}
-
-	unsubscribeLdkEvents := func(eventHandler chan ldk_node.Event) {
-		svc.Logger.Debugf("Locking event handler mutex")
-		ldkEventHandlersMutex.Lock()
-		svc.Logger.Debugf("Locked event handler mutex")
-		for i := 0; i < len(ldkEventHandlers); i++ {
-			if eventHandler == ldkEventHandlers[i] {
-				// Replace the element to be removed with the last element of the slice
-				ldkEventHandlers[i] = ldkEventHandlers[len(ldkEventHandlers)-1]
-				// Slice off the last element
-				ldkEventHandlers = ldkEventHandlers[:len(ldkEventHandlers)-1]
-				break
-			}
-		}
-		ldkEventHandlersMutex.Unlock()
-		svc.Logger.Debugf("Unlocked event handler mutex")
-	}
+	ldkEventConsumer := make(chan *ldk_node.Event)
+	ctx, cancel := context.WithCancel(svc.ctx)
 
+	// check for and forward new LDK events to LDKEventBroadcaster (through ldkEventConsumer)
 	go func() {
 		for {
 			select {
-			case <-ldkEventListenerCtx.Done():
+			case <-ctx.Done():
 				return
 			default:
 				// NOTE: do not use WaitNextEvent() as it can block the LDK thread
@@ -117,15 +87,9 @@ func NewLDKService(svc *Service, mnemonic, workDir string, network string, esplo
 					time.Sleep(time.Duration(1) * time.Millisecond)
 					continue
 				}
-				svc.Logger.Debugf("Locking event handler mutex")
-				ldkEventHandlersMutex.Lock()
-				svc.Logger.Debugf("Locked event handler mutex")
-				svc.Logger.Infof("Received LDK event %+v (%d listeners)", *event, len(ldkEventHandlers))
-				for _, eventHandler := range ldkEventHandlers {
-					eventHandler <- *event
-				}
-				ldkEventHandlersMutex.Unlock()
-				svc.Logger.Debugf("Unlocked event handler mutex")
+
+				svc.Logger.Infof("Received LDK event %+v", *event)
+				ldkEventConsumer <- event
 
 				node.EventHandled()
 			}
@@ -136,10 +100,10 @@ func NewLDKService(svc *Service, mnemonic, workDir string, network string, esplo
 		workdir: newpath,
 		node:    node,
 		//listener: &listener,
-		svc:                       svc,
-		cancelLdkEventListenerCtx: cancelLdkEventListenerCtx,
-		subscribeLdkEvents:        subscribeLdkEvents,
-		unsubscribeLdkEvents:      unsubscribeLdkEvents,
+		svc:                 svc,
+		ctx:                 ctx,
+		cancel:              cancel,
+		ldkEventBroadcaster: NewLDKEventBroadcaster(svc.Logger, ctx, ldkEventConsumer),
 	}
 
 	nodeId := node.NodeId()
@@ -155,16 +119,16 @@ func NewLDKService(svc *Service, mnemonic, workDir string, network string, esplo
 
 func (gs *LDKService) Shutdown() error {
 	gs.svc.Logger.Infof("shutting down LDK client")
-	gs.cancelLdkEventListenerCtx()
+	gs.cancel()
 	gs.node.Destroy()
 
 	return nil
 }
 
 func (gs *LDKService) SendPaymentSync(ctx context.Context, payReq string) (preimage string, err error) {
 	paymentStart := time.Now()
-	eventListener := gs.subscribeLdkEvents()
-	defer gs.unsubscribeLdkEvents(eventListener)
+	ldkEventSubscription := gs.ldkEventBroadcaster.Subscribe()
+	defer gs.ldkEventBroadcaster.CancelSubscription(ldkEventSubscription)
 
 	paymentHash, err := gs.node.SendPayment(payReq)
 	if err != nil {
@@ -173,10 +137,10 @@ func (gs *LDKService) SendPaymentSync(ctx context.Context, payReq string) (preim
 	}
 
 	for start := time.Now(); time.Since(start) < time.Second*60; {
-		event := <-eventListener
+		event := <-ldkEventSubscription
 
-		eventPaymentSuccessful, isEventPaymentSuccessfulEvent := event.(ldk_node.EventPaymentSuccessful)
-		eventPaymentFailed, isEventPaymentFailedEvent := event.(ldk_node.EventPaymentFailed)
+		eventPaymentSuccessful, isEventPaymentSuccessfulEvent := (*event).(ldk_node.EventPaymentSuccessful)
+		eventPaymentFailed, isEventPaymentFailedEvent := (*event).(ldk_node.EventPaymentFailed)
 
 		if isEventPaymentSuccessfulEvent && eventPaymentSuccessful.PaymentHash == paymentHash {
 			gs.svc.Logger.Infof("Got payment success event")
@@ -432,8 +396,8 @@ func (gs *LDKService) OpenChannel(ctx context.Context, openChannelRequest *lncli
 		return nil, errors.New("node is not peered yet")
 	}
 
-	eventListener := gs.subscribeLdkEvents()
-	defer gs.unsubscribeLdkEvents(eventListener)
+	ldkEventSubscription := gs.ldkEventBroadcaster.Subscribe()
+	defer gs.ldkEventBroadcaster.CancelSubscription(ldkEventSubscription)
 
 	gs.svc.Logger.Infof("Opening channel with: %v", foundPeer.NodeId)
 	userChannelId, err := gs.node.ConnectOpenChannel(foundPeer.NodeId, foundPeer.Address, uint64(openChannelRequest.Amount), nil, nil, openChannelRequest.Public)
@@ -446,9 +410,9 @@ func (gs *LDKService) OpenChannel(ctx context.Context, openChannelRequest *lncli
 	gs.svc.Logger.Infof("Funded channel: %v", userChannelId)
 
 	for start := time.Now(); time.Since(start) < time.Second*60; {
-		event := <-eventListener
+		event := <-ldkEventSubscription
 
-		channelPendingEvent, isChannelPendingEvent := event.(ldk_node.EventChannelPending)
+		channelPendingEvent, isChannelPendingEvent := (*event).(ldk_node.EventChannelPending)
 
 		if !isChannelPendingEvent {
 			continue

ldk_event_broadcaster.go

@@ -0,0 +1,77 @@
+package main
+
+import (
+	"context"
+	"slices"
+
+	"github.com/getAlby/ldk-node-go/ldk_node"
+	"github.com/sirupsen/logrus"
+)
+
+// based on https://betterprogramming.pub/how-to-broadcast-messages-in-go-using-channels-b68f42bdf32e
+type ldkEventBroadcastServer struct {
+	logger         *logrus.Logger
+	source         <-chan *ldk_node.Event
+	listeners      []chan *ldk_node.Event
+	addListener    chan chan *ldk_node.Event
+	removeListener chan (<-chan *ldk_node.Event)
+}
+
+type LDKEventBroadcaster interface {
+	Subscribe() <-chan *ldk_node.Event
+	CancelSubscription(<-chan *ldk_node.Event)
+}
+
+func NewLDKEventBroadcaster(logger *logrus.Logger, ctx context.Context, source <-chan *ldk_node.Event) LDKEventBroadcaster {
+	service := &ldkEventBroadcastServer{
+		logger:         logger,
+		source:         source,
+		listeners:      make([]chan *ldk_node.Event, 0),
+		addListener:    make(chan chan *ldk_node.Event),
+		removeListener: make(chan (<-chan *ldk_node.Event)),
+	}
+	go service.serve(ctx)
+	return service
+}
+
+func (s *ldkEventBroadcastServer) Subscribe() <-chan *ldk_node.Event {
+	newListener := make(chan *ldk_node.Event)
+	s.addListener <- newListener
+	return newListener
+}
+
+func (s *ldkEventBroadcastServer) CancelSubscription(channel <-chan *ldk_node.Event) {
+	s.removeListener <- channel
+}
+
+func (s *ldkEventBroadcastServer) serve(ctx context.Context) {
+	defer func() {
+		for _, listener := range s.listeners {
+			if listener != nil {
+				close(listener)
+			}
+		}
+	}()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case newListener := <-s.addListener:
+			s.listeners = append(s.listeners, newListener)
+		case listenerToRemove := <-s.removeListener:
+			for i, listener := range s.listeners {
+				if listener == listenerToRemove {
+					s.listeners[i] = s.listeners[len(s.listeners)-1]
+					s.listeners = slices.Delete(s.listeners, len(s.listeners)-1, len(s.listeners))
+					close(listener)
+					break
+				}
+			}
+		case event := <-s.source:
+			for _, listener := range s.listeners {
+				listener <- event
+			}
+		}
+	}
+}