Merge pull request #1988 from valentinewallace/subscribe-chans-rpc

rpc: Add SubscribeChannels RPC.

Author: Roasbeef

channelnotifier/channelnotifier.go

@@ -0,0 +1,137 @@
+package channelnotifier
+
+import (
+	"sync/atomic"
+
+	"github.com/btcsuite/btcd/wire"
+	"github.com/lightningnetwork/lnd/channeldb"
+	"github.com/lightningnetwork/lnd/subscribe"
+)
+
+// ChannelNotifier is a subsystem which all active, inactive, and closed channel
+// events pipe through. It takes subscriptions for its events, and whenever
+// it receives a new event it notifies its subscribers over the proper channel.
+type ChannelNotifier struct {
+	started uint32
+	stopped uint32
+
+	ntfnServer *subscribe.Server
+
+	chanDB *channeldb.DB
+}
+
+// OpenChannelEvent represents a new event where a channel goes from pending
+// open to open.
+type OpenChannelEvent struct {
+	// Channel is the channel that has become open.
+	Channel *channeldb.OpenChannel
+}
+
+// ActiveChannelEvent represents a new event where a channel becomes active.
+type ActiveChannelEvent struct {
+	// ChannelPoint is the channelpoint for the newly active channel.
+	ChannelPoint *wire.OutPoint
+}
+
+// InactiveChannelEvent represents a new event where a channel becomes inactive.
+type InactiveChannelEvent struct {
+	// ChannelPoint is the channelpoint for the newly inactive channel.
+	ChannelPoint *wire.OutPoint
+}
+
+// ClosedChannelEvent represents a new event where a channel becomes closed.
+type ClosedChannelEvent struct {
+	// CloseSummary is the summary of the channel close that has occurred.
+	CloseSummary *channeldb.ChannelCloseSummary
+}
+
+// New creates a new channel notifier. The ChannelNotifier gets channel
+// events from peers and from the chain arbitrator, and dispatches them to
+// its clients.
+func New(chanDB *channeldb.DB) *ChannelNotifier {
+	return &ChannelNotifier{
+		ntfnServer: subscribe.NewServer(),
+		chanDB:     chanDB,
+	}
+}
+
+// Start starts the ChannelNotifier and all goroutines it needs to carry out its task.
+func (c *ChannelNotifier) Start() error {
+	if !atomic.CompareAndSwapUint32(&c.started, 0, 1) {
+		return nil
+	}
+
+	log.Tracef("ChannelNotifier %v starting", c)
+
+	if err := c.ntfnServer.Start(); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// Stop signals the notifier for a graceful shutdown.
+func (c *ChannelNotifier) Stop() {
+	if !atomic.CompareAndSwapUint32(&c.stopped, 0, 1) {
+		return
+	}
+
+	c.ntfnServer.Stop()
+}
+
+// SubscribeChannelEvents returns a subscribe.Client that will receive updates
+// any time the Server is made aware of a new event.
+func (c *ChannelNotifier) SubscribeChannelEvents() (*subscribe.Client, error) {
+	return c.ntfnServer.Subscribe()
+}
+
+// NotifyOpenChannelEvent notifies the channelEventNotifier goroutine that a
+// channel has gone from pending open to open.
+func (c *ChannelNotifier) NotifyOpenChannelEvent(chanPoint wire.OutPoint) {
+
+	// Fetch the relevant channel from the database.
+	channel, err := c.chanDB.FetchChannel(chanPoint)
+	if err != nil {
+		log.Warnf("Unable to fetch open channel from the db: %v", err)
+	}
+
+	// Send the open event to all channel event subscribers.
+	event := OpenChannelEvent{Channel: channel}
+	if err := c.ntfnServer.SendUpdate(event); err != nil {
+		log.Warnf("Unable to send open channel update: %v", err)
+	}
+}
+
+// NotifyClosedChannelEvent notifies the channelEventNotifier goroutine that a
+// channel has closed.
+func (c *ChannelNotifier) NotifyClosedChannelEvent(chanPoint wire.OutPoint) {
+	// Fetch the relevant closed channel from the database.
+	closeSummary, err := c.chanDB.FetchClosedChannel(&chanPoint)
+	if err != nil {
+		log.Warnf("Unable to fetch closed channel summary from the db: %v", err)
+	}
+
+	// Send the closed event to all channel event subscribers.
+	event := ClosedChannelEvent{CloseSummary: closeSummary}
+	if err := c.ntfnServer.SendUpdate(event); err != nil {
+		log.Warnf("Unable to send closed channel update: %v", err)
+	}
+}
+
+// NotifyActiveChannelEvent notifies the channelEventNotifier goroutine that a
+// channel is active.
+func (c *ChannelNotifier) NotifyActiveChannelEvent(chanPoint wire.OutPoint) {
+	event := ActiveChannelEvent{ChannelPoint: &chanPoint}
+	if err := c.ntfnServer.SendUpdate(event); err != nil {
+		log.Warnf("Unable to send active channel update: %v", err)
+	}
+}
+
+// NotifyInactiveChannelEvent notifies the channelEventNotifier goroutine that a
+// channel is inactive.
+func (c *ChannelNotifier) NotifyInactiveChannelEvent(chanPoint wire.OutPoint) {
+	event := InactiveChannelEvent{ChannelPoint: &chanPoint}
+	if err := c.ntfnServer.SendUpdate(event); err != nil {
+		log.Warnf("Unable to send inactive channel update: %v", err)
+	}
+}

channelnotifier/log.go

@@ -0,0 +1,45 @@
+package channelnotifier
+
+import (
+	"github.com/btcsuite/btclog"
+	"github.com/lightningnetwork/lnd/build"
+)
+
+// log is a logger that is initialized with no output filters.  This means the
+// package will not perform any logging by default until the caller requests
+// it.
+var log btclog.Logger
+
+// The default amount of logging is none.
+func init() {
+	UseLogger(build.NewSubLogger("CHNF", nil))
+}
+
+// DisableLog disables all library log output.  Logging output is disabled by
+// default until UseLogger is called.
+func DisableLog() {
+	UseLogger(btclog.Disabled)
+}
+
+// UseLogger uses a specified Logger to output package logging info.  This
+// should be used in preference to SetLogWriter if the caller is also using
+// btclog.
+func UseLogger(logger btclog.Logger) {
+	log = logger
+}
+
+// logClosure is used to provide a closure over expensive logging operations so
+// don't have to be performed when the logging level doesn't warrant it.
+type logClosure func() string
+
+// String invokes the underlying function and returns the result.
+func (c logClosure) String() string {
+	return c()
+}
+
+// newLogClosure returns a new closure over a function that returns a string
+// which itself provides a Stringer interface so that it can be used with the
+// logging system.
+func newLogClosure(c func() string) logClosure {
+	return logClosure(c)
+}

contractcourt/chain_arbitrator.go

@@ -141,6 +141,10 @@ type ChainArbitratorConfig struct {
 	// the given payment hash. ErrInvoiceNotFound is returned if an invoice
 	// is not found.
 	SettleInvoice func(lntypes.Hash, lnwire.MilliSatoshi) error
+
+	// NotifyClosedChannel is a function closure that the ChainArbitrator
+	// will use to notify the ChannelNotifier about a newly closed channel.
+	NotifyClosedChannel func(wire.OutPoint)
 }
 
 // ChainArbitrator is a sub-system that oversees the on-chain resolution of all
@@ -245,10 +249,16 @@ func newActiveChannelArbitrator(channel *channeldb.OpenChannel,
 			return chanMachine.ForceClose()
 		},
 		MarkCommitmentBroadcasted: channel.MarkCommitmentBroadcasted,
-		MarkChannelClosed:         channel.CloseChannel,
-		IsPendingClose:            false,
-		ChainArbitratorConfig:     c.cfg,
-		ChainEvents:               chanEvents,
+		MarkChannelClosed: func(summary *channeldb.ChannelCloseSummary) error {
+			if err := channel.CloseChannel(summary); err != nil {
+				return err
+			}
+			c.cfg.NotifyClosedChannel(summary.ChanPoint)
+			return nil
+		},
+		IsPendingClose:        false,
+		ChainArbitratorConfig: c.cfg,
+		ChainEvents:           chanEvents,
 	}
 
 	// The final component needed is an arbitrator log that the arbitrator
@@ -719,13 +729,7 @@ func (c *ChainArbitrator) WatchNewChannel(newChan *channeldb.OpenChannel) error
 // SubscribeChannelEvents returns a new active subscription for the set of
 // possible on-chain events for a particular channel. The struct can be used by
 // callers to be notified whenever an event that changes the state of the
-// channel on-chain occurs. If syncDispatch is true, then the sender of the
-// notification will wait until an error is sent over the ProcessACK before
-// modifying any database state. This allows callers to request a reliable hand
-// off.
-//
-// TODO(roasbeef): can be used later to provide RPC hook for all channel
-// lifetimes
+// channel on-chain occurs.
 func (c *ChainArbitrator) SubscribeChannelEvents(
 	chanPoint wire.OutPoint) (*ChainEventSubscription, error) {
 

fundingmanager.go

@@ -333,6 +333,10 @@ type fundingConfig struct {
 	// flood us with very small channels that would never really be usable
 	// due to fees.
 	MinChanSize btcutil.Amount
+
+	// NotifyOpenChannelEvent informs the ChannelNotifier when channels
+	// transition from pending open to open.
+	NotifyOpenChannelEvent func(wire.OutPoint)
 }
 
 // fundingManager acts as an orchestrator/bridge between the wallet's
@@ -1939,6 +1943,10 @@ func (f *fundingManager) waitForFundingConfirmation(completeChan *channeldb.Open
 		return
 	}
 
+	// Inform the ChannelNotifier that the channel has transitioned from
+	// pending open to open.
+	f.cfg.NotifyOpenChannelEvent(completeChan.FundingOutpoint)
+
 	// TODO(roasbeef): ideally persistent state update for chan above
 	// should be abstracted
 

fundingmanager_test.go

@@ -353,8 +353,9 @@ func createTestFundingManager(t *testing.T, privKey *btcec.PrivateKey,
 			publTxChan <- txn
 			return nil
 		},
-		ZombieSweeperInterval: 1 * time.Hour,
-		ReservationTimeout:    1 * time.Nanosecond,
+		ZombieSweeperInterval:  1 * time.Hour,
+		ReservationTimeout:     1 * time.Nanosecond,
+		NotifyOpenChannelEvent: func(wire.OutPoint) {},
 	})
 	if err != nil {
 		t.Fatalf("failed creating fundingManager: %v", err)

htlcswitch/interfaces.go

@@ -1,6 +1,7 @@
 package htlcswitch
 
 import (
+	"github.com/btcsuite/btcd/wire"
 	"github.com/lightningnetwork/lnd/channeldb"
 	"github.com/lightningnetwork/lnd/lnpeer"
 	"github.com/lightningnetwork/lnd/lntypes"
@@ -58,6 +59,9 @@ type ChannelLink interface {
 	// possible).
 	HandleChannelUpdate(lnwire.Message)
 
+	// ChannelPoint returns the channel outpoint for the channel link.
+	ChannelPoint() *wire.OutPoint
+
 	// ChanID returns the channel ID for the channel link. The channel ID
 	// is a more compact representation of a channel's full outpoint.
 	ChanID() lnwire.ChannelID

htlcswitch/link.go

@@ -9,6 +9,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/btcsuite/btcd/wire"
 	"github.com/davecgh/go-spew/spew"
 	"github.com/go-errors/errors"
 	"github.com/lightningnetwork/lnd/channeldb"
@@ -1743,6 +1744,12 @@ func (l *channelLink) Peer() lnpeer.Peer {
 	return l.cfg.Peer
 }
 
+// ChannelPoint returns the channel outpoint for the channel link.
+// NOTE: Part of the ChannelLink interface.
+func (l *channelLink) ChannelPoint() *wire.OutPoint {
+	return l.channel.ChannelPoint()
+}
+
 // ShortChanID returns the short channel ID for the channel link. The short
 // channel ID encodes the exact location in the main chain that the original
 // funding output can be found.

htlcswitch/mock.go

@@ -162,9 +162,11 @@ func initSwitchWithDB(startingHeight uint32, db *channeldb.DB) (*Switch, error)
 		FetchLastChannelUpdate: func(lnwire.ShortChannelID) (*lnwire.ChannelUpdate, error) {
 			return nil, nil
 		},
-		Notifier:       &mockNotifier{},
-		FwdEventTicker: ticker.MockNew(DefaultFwdEventInterval),
-		LogEventTicker: ticker.MockNew(DefaultLogInterval),
+		Notifier:              &mockNotifier{},
+		FwdEventTicker:        ticker.MockNew(DefaultFwdEventInterval),
+		LogEventTicker:        ticker.MockNew(DefaultLogInterval),
+		NotifyActiveChannel:   func(wire.OutPoint) {},
+		NotifyInactiveChannel: func(wire.OutPoint) {},
 	}
 
 	return New(cfg, startingHeight)
@@ -673,6 +675,7 @@ func (f *mockChannelLink) ChanID() lnwire.ChannelID                     { return
 func (f *mockChannelLink) ShortChanID() lnwire.ShortChannelID           { return f.shortChanID }
 func (f *mockChannelLink) Bandwidth() lnwire.MilliSatoshi               { return 99999999 }
 func (f *mockChannelLink) Peer() lnpeer.Peer                            { return f.peer }
+func (f *mockChannelLink) ChannelPoint() *wire.OutPoint                 { return &wire.OutPoint{} }
 func (f *mockChannelLink) Stop()                                        {}
 func (f *mockChannelLink) EligibleToForward() bool                      { return f.eligible }
 func (f *mockChannelLink) setLiveShortChanID(sid lnwire.ShortChannelID) { f.shortChanID = sid }

htlcswitch/switch.go

@@ -175,6 +175,11 @@ type Config struct {
 	// LogEventTicker is a signal instructing the htlcswitch to log
 	// aggregate stats about it's forwarding during the last interval.
 	LogEventTicker ticker.Ticker
+
+	// NotifyActiveChannel and NotifyInactiveChannel allow the link to tell
+	// the ChannelNotifier when channels become active and inactive.
+	NotifyActiveChannel   func(wire.OutPoint)
+	NotifyInactiveChannel func(wire.OutPoint)
 }
 
 // Switch is the central messaging bus for all incoming/outgoing HTLCs.
@@ -1956,6 +1961,11 @@ func (s *Switch) addLiveLink(link ChannelLink) {
 		s.interfaceIndex[peerPub] = make(map[lnwire.ChannelID]ChannelLink)
 	}
 	s.interfaceIndex[peerPub][link.ChanID()] = link
+
+	// Inform the channel notifier if the link has become active.
+	if link.EligibleToForward() {
+		s.cfg.NotifyActiveChannel(*link.ChannelPoint())
+	}
 }
 
 // GetLink is used to initiate the handling of the get link command. The
@@ -2031,6 +2041,9 @@ func (s *Switch) removeLink(chanID lnwire.ChannelID) ChannelLink {
 		return nil
 	}
 
+	// Inform the Channel Notifier about the link becoming inactive.
+	s.cfg.NotifyInactiveChannel(*link.ChannelPoint())
+
 	// Remove the channel from live link indexes.
 	delete(s.pendingLinkIndex, link.ChanID())
 	delete(s.linkIndex, link.ChanID())