funding+server.go: modify notifications to pass throuhh server

This modifies the various channelnotifier notification functions
to instead hit the server and then call the notification routine.
This allows us to accurately modify the server's maps.

funding/manager.go

@@ -3,6 +3,7 @@ package funding
 import (
 	"bytes"
 	"encoding/binary"
+	"encoding/hex"
 	"fmt"
 	"io"
 	"sync"
@@ -510,7 +511,7 @@ type Config struct {
 
 	// NotifyOpenChannelEvent informs the ChannelNotifier when channels
 	// transition from pending open to open.
-	NotifyOpenChannelEvent func(wire.OutPoint)
+	NotifyOpenChannelEvent func(wire.OutPoint, string) error
 
 	// OpenChannelPredicate is a predicate on the lnwire.OpenChannel message
 	// and on the requesting node's public key that returns a bool which
@@ -520,7 +521,11 @@ type Config struct {
 	// NotifyPendingOpenChannelEvent informs the ChannelNotifier when
 	// channels enter a pending state.
 	NotifyPendingOpenChannelEvent func(wire.OutPoint,
-		*channeldb.OpenChannel)
+		*channeldb.OpenChannel, string) error
+
+	// NotifyFundingTimeout informs the ChannelNotifier when a pending-open
+	// channel times out because the funding transaction hasn't confirmed.
+	NotifyFundingTimeout func(wire.OutPoint, string) error
 
 	// EnableUpfrontShutdown specifies whether the upfront shutdown script
 	// is enabled.
@@ -1312,7 +1317,14 @@ func (f *Manager) advancePendingChannelState(channel *channeldb.OpenChannel,
 
 		// Inform the ChannelNotifier that the channel has transitioned
 		// from pending open to open.
-		f.cfg.NotifyOpenChannelEvent(channel.FundingOutpoint)
+		remoteHex := remotePubHex(channel.IdentityPub)
+		if err := f.cfg.NotifyOpenChannelEvent(
+			channel.FundingOutpoint, remoteHex,
+		); err != nil {
+			log.Errorf("Unable to notify open channel event for "+
+				"ChannelPoint(%v): %v",
+				channel.FundingOutpoint, err)
+		}
 
 		// Find and close the discoverySignal for this channel such
 		// that ChannelReady messages will be processed.
@@ -2653,7 +2665,13 @@ func (f *Manager) fundeeProcessFundingCreated(peer lnpeer.Peer,
 
 	// Inform the ChannelNotifier that the channel has entered
 	// pending open state.
-	f.cfg.NotifyPendingOpenChannelEvent(fundingOut, completeChan)
+	remoteHex := remotePubHex(completeChan.IdentityPub)
+	if err := f.cfg.NotifyPendingOpenChannelEvent(
+		fundingOut, completeChan, remoteHex,
+	); err != nil {
+		log.Errorf("Unable to send pending-open channel event for "+
+			"ChannelPoint(%v) %v", fundingOut, err)
+	}
 
 	// At this point we have sent our last funding message to the
 	// initiating peer before the funding transaction will be broadcast.
@@ -2873,7 +2891,15 @@ func (f *Manager) funderProcessFundingSigned(peer lnpeer.Peer,
 	case resCtx.updates <- upd:
 		// Inform the ChannelNotifier that the channel has entered
 		// pending open state.
-		f.cfg.NotifyPendingOpenChannelEvent(*fundingPoint, completeChan)
+		remoteHex := remotePubHex(completeChan.IdentityPub)
+		if err := f.cfg.NotifyPendingOpenChannelEvent(
+			*fundingPoint, completeChan, remoteHex,
+		); err != nil {
+			log.Errorf("Unable to send pending-open channel "+
+				"event for ChannelPoint(%v) %v", fundingPoint,
+				err)
+		}
+
 	case <-f.quit:
 		return
 	}
@@ -2929,6 +2955,14 @@ func (f *Manager) fundingTimeout(c *channeldb.OpenChannel,
 			c.FundingOutpoint, err)
 	}
 
+	// Notify other subsystems about the funding timeout.
+	remoteHex := remotePubHex(c.IdentityPub)
+	err := f.cfg.NotifyFundingTimeout(c.FundingOutpoint, remoteHex)
+	if err != nil {
+		log.Errorf("failed to notify of funding timeout for "+
+			"ChanPoint(%v): %v", c.FundingOutpoint, err)
+	}
+
 	timeoutErr := fmt.Errorf("timeout waiting for funding tx (%v) to "+
 		"confirm", c.FundingOutpoint)
 
@@ -3297,7 +3331,14 @@ func (f *Manager) handleFundingConfirmation(
 
 	// Inform the ChannelNotifier that the channel has transitioned from
 	// pending open to open.
-	f.cfg.NotifyOpenChannelEvent(completeChan.FundingOutpoint)
+	remoteHex := remotePubHex(completeChan.IdentityPub)
+	if err := f.cfg.NotifyOpenChannelEvent(
+		completeChan.FundingOutpoint, remoteHex,
+	); err != nil {
+		log.Errorf("Unable to notify open channel event for "+
+			"ChannelPoint(%v): %v", completeChan.FundingOutpoint,
+			err)
+	}
 
 	// Close the discoverySignal channel, indicating to a separate
 	// goroutine that the channel now is marked as open in the database
@@ -5374,3 +5415,9 @@ func (f *Manager) waitForPeerOnline(peerPubkey *btcec.PublicKey) (lnpeer.Peer,
 	}
 	return peer, nil
 }
+
+// remotePubHex takes a *btcec.PublicKey and converts it to a hex-encoded
+// string.
+func remotePubHex(remotePub *btcec.PublicKey) string {
+	return hex.EncodeToString(remotePub.SerializeCompressed())
+}

funding/manager_test.go

@@ -231,17 +231,29 @@ type mockChanEvent struct {
 	pendingOpenEvent chan channelnotifier.PendingOpenChannelEvent
 }
 
-func (m *mockChanEvent) NotifyOpenChannelEvent(outpoint wire.OutPoint) {
+func (m *mockChanEvent) NotifyOpenChannelEvent(outpoint wire.OutPoint,
+	remoteHex string) error {
+
 	m.openEvent <- outpoint
+
+	return nil
 }
 
 func (m *mockChanEvent) NotifyPendingOpenChannelEvent(outpoint wire.OutPoint,
-	pendingChannel *channeldb.OpenChannel) {
+	pendingChannel *channeldb.OpenChannel, remoteHex string) error {
 
 	m.pendingOpenEvent <- channelnotifier.PendingOpenChannelEvent{
 		ChannelPoint:   &outpoint,
 		PendingChannel: pendingChannel,
 	}
+
+	return nil
+}
+
+func (m *mockChanEvent) NotifyFundingTimeout(outpoint wire.OutPoint,
+	remoteHex string) error {
+
+	return nil
 }
 
 // mockZeroConfAcceptor always accepts the channel open request for zero-conf
@@ -550,6 +562,7 @@ func createTestFundingManager(t *testing.T, privKey *btcec.PrivateKey,
 		NotifyOpenChannelEvent:        evt.NotifyOpenChannelEvent,
 		OpenChannelPredicate:          chainedAcceptor,
 		NotifyPendingOpenChannelEvent: evt.NotifyPendingOpenChannelEvent,
+		NotifyFundingTimeout:          evt.NotifyFundingTimeout,
 		DeleteAliasEdge: func(scid lnwire.ShortChannelID) (
 			*models.ChannelEdgePolicy, error) {
 

server.go

@@ -1707,9 +1707,10 @@ func newServer(cfg *Config, listenAddrs []net.Addr,
 		MaxPendingChannels:            cfg.MaxPendingChannels,
 		RejectPush:                    cfg.RejectPush,
 		MaxLocalCSVDelay:              chainCfg.MaxLocalDelay,
-		NotifyOpenChannelEvent:        s.channelNotifier.NotifyOpenChannelEvent,
+		NotifyOpenChannelEvent:        s.notifyOpenChannelPeerEvent,
 		OpenChannelPredicate:          chanPredicate,
-		NotifyPendingOpenChannelEvent: s.channelNotifier.NotifyPendingOpenChannelEvent,
+		NotifyPendingOpenChannelEvent: s.notifyPendingOpenChannelPeerEvent,
+		NotifyFundingTimeout:          s.notifyFundingTimeoutPeerEvent,
 		EnableUpfrontShutdown:         cfg.EnableUpfrontShutdown,
 		MaxAnchorsCommitFeeRate: chainfee.SatPerKVByte(
 			s.cfg.MaxCommitFeeRateAnchors * 1000).FeePerKWeight(),