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

rpc: Add SubscribeChannels RPC.

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())

lnd_test.go

@@ -5694,15 +5694,137 @@ func testInvoiceSubscriptions(net *lntest.NetworkHarness, t *harnessTest) {
 	closeChannelAndAssert(ctxt, t, net, net.Alice, chanPoint, false)
 }
 
-// testBasicChannelCreation test multiple channel opening and closing.
-func testBasicChannelCreation(net *lntest.NetworkHarness, t *harnessTest) {
-	ctxb := context.Background()
+// channelSubscription houses the proxied update and error chans for a node's
+// channel subscriptions.
+type channelSubscription struct {
+	updateChan chan *lnrpc.ChannelEventUpdate
+	errChan    chan error
+	quit       chan struct{}
+}
 
+// subscribeChannelNotifications subscribes to channel updates and launches a
+// goroutine that forwards these to the returned channel.
+func subscribeChannelNotifications(ctxb context.Context, t *harnessTest,
+	node *lntest.HarnessNode) channelSubscription {
+
+	// We'll first start by establishing a notification client which will
+	// send us notifications upon channels becoming active, inactive or
+	// closed.
+	req := &lnrpc.ChannelEventSubscription{}
+	ctx, cancelFunc := context.WithCancel(ctxb)
+
+	chanUpdateClient, err := node.SubscribeChannelEvents(ctx, req)
+	if err != nil {
+		t.Fatalf("unable to create channel update client: %v", err)
+	}
+
+	// We'll launch a goroutine that will be responsible for proxying all
+	// notifications recv'd from the client into the channel below.
+	errChan := make(chan error, 1)
+	quit := make(chan struct{})
+	chanUpdates := make(chan *lnrpc.ChannelEventUpdate, 20)
+	go func() {
+		defer cancelFunc()
+		for {
+			select {
+			case <-quit:
+				return
+			default:
+				chanUpdate, err := chanUpdateClient.Recv()
+				select {
+				case <-quit:
+					return
+				default:
+				}
+
+				if err == io.EOF {
+					return
+				} else if err != nil {
+					select {
+					case errChan <- err:
+					case <-quit:
+					}
+					return
+				}
+
+				select {
+				case chanUpdates <- chanUpdate:
+				case <-quit:
+					return
+				}
+			}
+		}
+	}()
+
+	return channelSubscription{
+		updateChan: chanUpdates,
+		errChan:    errChan,
+		quit:       quit,
+	}
+}
+
+// verifyCloseUpdate is used to verify that a closed channel update is of the
+// expected type.
+func verifyCloseUpdate(chanUpdate *lnrpc.ChannelEventUpdate,
+	force bool, forceType lnrpc.ChannelCloseSummary_ClosureType) error {
+
+	// We should receive one inactive and one closed notification
+	// for each channel.
+	switch update := chanUpdate.Channel.(type) {
+	case *lnrpc.ChannelEventUpdate_InactiveChannel:
+		if chanUpdate.Type != lnrpc.ChannelEventUpdate_INACTIVE_CHANNEL {
+			return fmt.Errorf("update type mismatch: expected %v, got %v",
+				lnrpc.ChannelEventUpdate_INACTIVE_CHANNEL,
+				chanUpdate.Type)
+		}
+	case *lnrpc.ChannelEventUpdate_ClosedChannel:
+		if chanUpdate.Type !=
+			lnrpc.ChannelEventUpdate_CLOSED_CHANNEL {
+			return fmt.Errorf("update type mismatch: expected %v, got %v",
+				lnrpc.ChannelEventUpdate_CLOSED_CHANNEL,
+				chanUpdate.Type)
+		}
+
+		switch force {
+		case true:
+			if update.ClosedChannel.CloseType != forceType {
+				return fmt.Errorf("channel closure type mismatch: "+
+					"expected %v, got %v",
+					forceType,
+					update.ClosedChannel.CloseType)
+			}
+		case false:
+			if update.ClosedChannel.CloseType !=
+				lnrpc.ChannelCloseSummary_COOPERATIVE_CLOSE {
+				return fmt.Errorf("channel closure type "+
+					"mismatch: expected %v, got %v",
+					lnrpc.ChannelCloseSummary_COOPERATIVE_CLOSE,
+					update.ClosedChannel.CloseType)
+			}
+		}
+	default:
+		return fmt.Errorf("channel update channel of wrong type, "+
+			"expected closed channel, got %T",
+			update)
+	}
+
+	return nil
+}
+
+// testBasicChannelCreationAndUpdates tests multiple channel opening and closing,
+// and ensures that if a node is subscribed to channel updates they will be
+// received correctly for both cooperative and force closed channels.
+func testBasicChannelCreationAndUpdates(net *lntest.NetworkHarness, t *harnessTest) {
+	ctxb := context.Background()
 	const (
 		numChannels = 2
 		amount      = maxBtcFundingAmount
 	)
 
+	// Let Bob subscribe to channel notifications.
+	bobChanSub := subscribeChannelNotifications(ctxb, t, net.Bob)
+	defer close(bobChanSub.quit)
+
 	// Open the channel between Alice and Bob, asserting that the
 	// channel has been properly open on-chain.
 	chanPoints := make([]*lnrpc.ChannelPoint, numChannels)
@@ -5716,11 +5838,89 @@ func testBasicChannelCreation(net *lntest.NetworkHarness, t *harnessTest) {
 		)
 	}
 
-	// Close the channel between Alice and Bob, asserting that the
-	// channel has been properly closed on-chain.
-	for _, chanPoint := range chanPoints {
-		ctxt, _ := context.WithTimeout(ctxb, channelCloseTimeout)
-		closeChannelAndAssert(ctxt, t, net, net.Alice, chanPoint, false)
+	// Since each of the channels just became open, Bob should we receive an
+	// open and an active notification for each channel.
+	var numChannelUpds int
+	for numChannelUpds < 2*numChannels {
+		select {
+		case update := <-bobChanSub.updateChan:
+			switch update.Type {
+			case lnrpc.ChannelEventUpdate_ACTIVE_CHANNEL:
+			case lnrpc.ChannelEventUpdate_OPEN_CHANNEL:
+			default:
+				t.Fatalf("update type mismatch: expected open or active "+
+					"channel notification, got: %v", update.Type)
+			}
+			numChannelUpds++
+		case <-time.After(time.Second * 10):
+			t.Fatalf("timeout waiting for channel notifications, "+
+				"only received %d/%d chanupds", numChannelUpds,
+				numChannels)
+		}
+	}
+
+	// Subscribe Alice to channel updates so we can test that both remote
+	// and local force close notifications are received correctly.
+	aliceChanSub := subscribeChannelNotifications(ctxb, t, net.Alice)
+	defer close(aliceChanSub.quit)
+
+	// Close the channel between Alice and Bob, asserting that the channel
+	// has been properly closed on-chain.
+	for i, chanPoint := range chanPoints {
+		ctx, _ := context.WithTimeout(context.Background(), defaultTimeout)
+
+		// Force close half of the channels.
+		force := i%2 == 0
+		closeChannelAndAssert(ctx, t, net, net.Alice, chanPoint, force)
+		if force {
+			cleanupForceClose(t, net, net.Alice, chanPoint)
+		}
+	}
+
+	// verifyCloseUpdatesReceived is used to verify that Alice and Bob
+	// receive the correct channel updates in order.
+	verifyCloseUpdatesReceived := func(sub channelSubscription,
+		forceType lnrpc.ChannelCloseSummary_ClosureType) error {
+
+		// Ensure one inactive and one closed notification is received for each
+		// closed channel.
+		numChannelUpds := 0
+		for numChannelUpds < 2*numChannels {
+			// Every other channel should be force closed.
+			force := (numChannelUpds/2)%2 == 0
+
+			select {
+			case chanUpdate := <-sub.updateChan:
+				err := verifyCloseUpdate(chanUpdate, force, forceType)
+				if err != nil {
+					return err
+				}
+
+				numChannelUpds++
+			case err := <-sub.errChan:
+				return err
+			case <-time.After(time.Second * 10):
+				return fmt.Errorf("timeout waiting for channel "+
+					"notifications, only received %d/%d "+
+					"chanupds", numChannelUpds, 2*numChannels)
+			}
+		}
+
+		return nil
+	}
+
+	// Verify Bob receives all closed channel notifications. He should
+	// receive a remote force close notification for force closed channels.
+	if err := verifyCloseUpdatesReceived(bobChanSub,
+		lnrpc.ChannelCloseSummary_REMOTE_FORCE_CLOSE); err != nil {
+		t.Fatalf("errored verifying close updates: %v", err)
+	}
+
+	// Verify Alice receives all closed channel notifications. She should
+	// receive a remote force close notification for force closed channels.
+	if err := verifyCloseUpdatesReceived(aliceChanSub,
+		lnrpc.ChannelCloseSummary_LOCAL_FORCE_CLOSE); err != nil {
+		t.Fatalf("errored verifying close updates: %v", err)
 	}
 }
 
@@ -12943,8 +13143,8 @@ var testsCases = []*testCase{
 		test: testMultiHopOverPrivateChannels,
 	},
 	{
-		name: "multiple channel creation",
-		test: testBasicChannelCreation,
+		name: "multiple channel creation and update subscription",
+		test: testBasicChannelCreationAndUpdates,
 	},
 	{
 		name: "invoice update subscription",

lnrpc/autopilotrpc/autopilot.pb.go

@@ -33,7 +33,7 @@ func (m *StatusRequest) Reset()         { *m = StatusRequest{} }
 func (m *StatusRequest) String() string { return proto.CompactTextString(m) }
 func (*StatusRequest) ProtoMessage()    {}
 func (*StatusRequest) Descriptor() ([]byte, []int) {
-	return fileDescriptor_autopilot_52f30cf4d0055211, []int{0}
+	return fileDescriptor_autopilot_7db7978f022d4696, []int{0}
 }
 func (m *StatusRequest) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_StatusRequest.Unmarshal(m, b)
@@ -65,7 +65,7 @@ func (m *StatusResponse) Reset()         { *m = StatusResponse{} }
 func (m *StatusResponse) String() string { return proto.CompactTextString(m) }
 func (*StatusResponse) ProtoMessage()    {}
 func (*StatusResponse) Descriptor() ([]byte, []int) {
-	return fileDescriptor_autopilot_52f30cf4d0055211, []int{1}
+	return fileDescriptor_autopilot_7db7978f022d4696, []int{1}
 }
 func (m *StatusResponse) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_StatusResponse.Unmarshal(m, b)
@@ -104,7 +104,7 @@ func (m *ModifyStatusRequest) Reset()         { *m = ModifyStatusRequest{} }
 func (m *ModifyStatusRequest) String() string { return proto.CompactTextString(m) }
 func (*ModifyStatusRequest) ProtoMessage()    {}
 func (*ModifyStatusRequest) Descriptor() ([]byte, []int) {
-	return fileDescriptor_autopilot_52f30cf4d0055211, []int{2}
+	return fileDescriptor_autopilot_7db7978f022d4696, []int{2}
 }
 func (m *ModifyStatusRequest) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ModifyStatusRequest.Unmarshal(m, b)
@@ -141,7 +141,7 @@ func (m *ModifyStatusResponse) Reset()         { *m = ModifyStatusResponse{} }
 func (m *ModifyStatusResponse) String() string { return proto.CompactTextString(m) }
 func (*ModifyStatusResponse) ProtoMessage()    {}
 func (*ModifyStatusResponse) Descriptor() ([]byte, []int) {
-	return fileDescriptor_autopilot_52f30cf4d0055211, []int{3}
+	return fileDescriptor_autopilot_7db7978f022d4696, []int{3}
 }
 func (m *ModifyStatusResponse) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ModifyStatusResponse.Unmarshal(m, b)
@@ -172,7 +172,7 @@ func (m *QueryScoresRequest) Reset()         { *m = QueryScoresRequest{} }
 func (m *QueryScoresRequest) String() string { return proto.CompactTextString(m) }
 func (*QueryScoresRequest) ProtoMessage()    {}
 func (*QueryScoresRequest) Descriptor() ([]byte, []int) {
-	return fileDescriptor_autopilot_52f30cf4d0055211, []int{4}
+	return fileDescriptor_autopilot_7db7978f022d4696, []int{4}
 }
 func (m *QueryScoresRequest) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_QueryScoresRequest.Unmarshal(m, b)
@@ -210,7 +210,7 @@ func (m *QueryScoresResponse) Reset()         { *m = QueryScoresResponse{} }
 func (m *QueryScoresResponse) String() string { return proto.CompactTextString(m) }
 func (*QueryScoresResponse) ProtoMessage()    {}
 func (*QueryScoresResponse) Descriptor() ([]byte, []int) {
-	return fileDescriptor_autopilot_52f30cf4d0055211, []int{5}
+	return fileDescriptor_autopilot_7db7978f022d4696, []int{5}
 }
 func (m *QueryScoresResponse) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_QueryScoresResponse.Unmarshal(m, b)
@@ -249,7 +249,7 @@ func (m *QueryScoresResponse_HeuristicResult) Reset()         { *m = QueryScores
 func (m *QueryScoresResponse_HeuristicResult) String() string { return proto.CompactTextString(m) }
 func (*QueryScoresResponse_HeuristicResult) ProtoMessage()    {}
 func (*QueryScoresResponse_HeuristicResult) Descriptor() ([]byte, []int) {
-	return fileDescriptor_autopilot_52f30cf4d0055211, []int{5, 0}
+	return fileDescriptor_autopilot_7db7978f022d4696, []int{5, 0}
 }
 func (m *QueryScoresResponse_HeuristicResult) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_QueryScoresResponse_HeuristicResult.Unmarshal(m, b)
@@ -451,10 +451,10 @@ var _Autopilot_serviceDesc = grpc.ServiceDesc{
 }
 
 func init() {
-	proto.RegisterFile("autopilotrpc/autopilot.proto", fileDescriptor_autopilot_52f30cf4d0055211)
+	proto.RegisterFile("autopilotrpc/autopilot.proto", fileDescriptor_autopilot_7db7978f022d4696)
 }
 
-var fileDescriptor_autopilot_52f30cf4d0055211 = []byte{
+var fileDescriptor_autopilot_7db7978f022d4696 = []byte{
 	// 391 bytes of a gzipped FileDescriptorProto
 	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x94, 0x53, 0x4d, 0xaf, 0xd2, 0x40,
 	0x14, 0xcd, 0x94, 0x58, 0xec, 0x05, 0xc5, 0x0c, 0x84, 0x34, 0x95, 0x45, 0xe9, 0xaa, 0x1b, 0xdb,

lnrpc/rpc.proto

@@ -359,6 +359,14 @@ service Lightning {
         };
     }
 
+    /** lncli: `subscribechannelevents`
+    SubscribeChannelEvents creates a uni-directional stream from the server to
+    the client in which any updates relevant to the state of the channels are
+    sent over. Events include new active channels, inactive channels, and closed
+    channels.
+    */
+    rpc SubscribeChannelEvents (ChannelEventSubscription) returns (stream ChannelEventUpdate);
+
     /** lncli: `closedchannels`
     ClosedChannels returns a description of all the closed channels that 
     this node was a participant in.
@@ -1401,6 +1409,27 @@ message PendingChannelsResponse {
     repeated WaitingCloseChannel waiting_close_channels = 5 [ json_name = "waiting_close_channels" ];
 }
 
+message ChannelEventSubscription {
+}
+
+message ChannelEventUpdate {
+    oneof channel {
+        Channel open_channel = 1 [ json_name = "open_channel" ];
+        ChannelCloseSummary closed_channel = 2 [ json_name = "closed_channel" ];
+        ChannelPoint active_channel = 3 [ json_name = "active_channel" ];
+        ChannelPoint inactive_channel = 4 [ json_name = "inactive_channel" ];
+    }
+
+    enum UpdateType {
+         OPEN_CHANNEL = 0;
+         CLOSED_CHANNEL = 1;
+         ACTIVE_CHANNEL = 2;
+         INACTIVE_CHANNEL = 3;
+    }
+
+    UpdateType type = 5 [ json_name = "type" ];
+}
+
 message WalletBalanceRequest {
 }
 message WalletBalanceResponse {

lnrpc/rpc.swagger.json

@@ -1127,6 +1127,16 @@
       ],
       "default": "COOPERATIVE_CLOSE"
     },
+    "ChannelEventUpdateUpdateType": {
+      "type": "string",
+      "enum": [
+        "OPEN_CHANNEL",
+        "CLOSED_CHANNEL",
+        "ACTIVE_CHANNEL",
+        "INACTIVE_CHANNEL"
+      ],
+      "default": "OPEN_CHANNEL"
+    },
     "InvoiceInvoiceState": {
       "type": "string",
       "enum": [
@@ -1548,6 +1558,26 @@
         }
       }
     },
+    "lnrpcChannelEventUpdate": {
+      "type": "object",
+      "properties": {
+        "open_channel": {
+          "$ref": "#/definitions/lnrpcChannel"
+        },
+        "closed_channel": {
+          "$ref": "#/definitions/lnrpcChannelCloseSummary"
+        },
+        "active_channel": {
+          "$ref": "#/definitions/lnrpcChannelPoint"
+        },
+        "inactive_channel": {
+          "$ref": "#/definitions/lnrpcChannelPoint"
+        },
+        "type": {
+          "$ref": "#/definitions/ChannelEventUpdateUpdateType"
+        }
+      }
+    },
     "lnrpcChannelFeeReport": {
       "type": "object",
       "properties": {

log.go

@@ -15,6 +15,7 @@ import (
 	"github.com/lightningnetwork/lnd/build"
 	"github.com/lightningnetwork/lnd/chainntnfs"
 	"github.com/lightningnetwork/lnd/channeldb"
+	"github.com/lightningnetwork/lnd/channelnotifier"
 	"github.com/lightningnetwork/lnd/contractcourt"
 	"github.com/lightningnetwork/lnd/discovery"
 	"github.com/lightningnetwork/lnd/htlcswitch"
@@ -80,6 +81,7 @@ var (
 	wtwrLog = build.NewSubLogger("WTWR", backendLog.Logger)
 	ntfrLog = build.NewSubLogger("NTFR", backendLog.Logger)
 	irpcLog = build.NewSubLogger("IRPC", backendLog.Logger)
+	chnfLog = build.NewSubLogger("CHNF", backendLog.Logger)
 )
 
 // Initialize package-global logger variables.
@@ -105,6 +107,7 @@ func init() {
 	watchtower.UseLogger(wtwrLog)
 	chainrpc.UseLogger(ntfrLog)
 	invoicesrpc.UseLogger(irpcLog)
+	channelnotifier.UseLogger(chnfLog)
 }
 
 // subsystemLoggers maps each subsystem identifier to its associated logger.
@@ -136,6 +139,7 @@ var subsystemLoggers = map[string]btclog.Logger{
 	"WTWR": wtwrLog,
 	"NTFR": ntfnLog,
 	"IRPC": irpcLog,
+	"CHNF": chnfLog,
 }
 
 // initLogRotator initializes the logging rotator to write logs to logFile and

rpcserver.go

@@ -30,6 +30,7 @@ import (
 	"github.com/lightningnetwork/lnd/autopilot"
 	"github.com/lightningnetwork/lnd/build"
 	"github.com/lightningnetwork/lnd/channeldb"
+	"github.com/lightningnetwork/lnd/channelnotifier"
 	"github.com/lightningnetwork/lnd/htlcswitch"
 	"github.com/lightningnetwork/lnd/input"
 	"github.com/lightningnetwork/lnd/invoices"
@@ -245,6 +246,10 @@ var (
 			Entity: "offchain",
 			Action: "read",
 		}},
+		"/lnrpc.Lightning/SubscribeChannelEvents": {{
+			Entity: "offchain",
+			Action: "read",
+		}},
 		"/lnrpc.Lightning/ClosedChannels": {{
 			Entity: "offchain",
 			Action: "read",
@@ -2254,56 +2259,34 @@ func (r *rpcServer) ClosedChannels(ctx context.Context,
 			continue
 		}
 
-		nodePub := dbChannel.RemotePub
-		nodeID := hex.EncodeToString(nodePub.SerializeCompressed())
-
-		var closeType lnrpc.ChannelCloseSummary_ClosureType
 		switch dbChannel.CloseType {
 		case channeldb.CooperativeClose:
 			if filterResults && !in.Cooperative {
 				continue
 			}
-			closeType = lnrpc.ChannelCloseSummary_COOPERATIVE_CLOSE
 		case channeldb.LocalForceClose:
 			if filterResults && !in.LocalForce {
 				continue
 			}
-			closeType = lnrpc.ChannelCloseSummary_LOCAL_FORCE_CLOSE
 		case channeldb.RemoteForceClose:
 			if filterResults && !in.RemoteForce {
 				continue
 			}
-			closeType = lnrpc.ChannelCloseSummary_REMOTE_FORCE_CLOSE
 		case channeldb.BreachClose:
 			if filterResults && !in.Breach {
 				continue
 			}
-			closeType = lnrpc.ChannelCloseSummary_BREACH_CLOSE
 		case channeldb.FundingCanceled:
 			if filterResults && !in.FundingCanceled {
 				continue
 			}
-			closeType = lnrpc.ChannelCloseSummary_FUNDING_CANCELED
 		case channeldb.Abandoned:
 			if filterResults && !in.Abandoned {
 				continue
 			}
-			closeType = lnrpc.ChannelCloseSummary_ABANDONED
-		}
-
-		channel := &lnrpc.ChannelCloseSummary{
-			Capacity:          int64(dbChannel.Capacity),
-			RemotePubkey:      nodeID,
-			CloseHeight:       dbChannel.CloseHeight,
-			CloseType:         closeType,
-			ChannelPoint:      dbChannel.ChanPoint.String(),
-			ChanId:            dbChannel.ShortChanID.ToUint64(),
-			SettledBalance:    int64(dbChannel.SettledBalance),
-			TimeLockedBalance: int64(dbChannel.TimeLockedBalance),
-			ChainHash:         dbChannel.ChainHash.String(),
-			ClosingTxHash:     dbChannel.ClosingTXID.String(),
 		}
 
+		channel := createRPCClosedChannel(dbChannel)
 		resp.Channels = append(resp.Channels, channel)
 	}
 
@@ -2339,14 +2322,8 @@ func (r *rpcServer) ListChannels(ctx context.Context,
 
 	for _, dbChannel := range dbChannels {
 		nodePub := dbChannel.IdentityPub
-		nodeID := hex.EncodeToString(nodePub.SerializeCompressed())
 		chanPoint := dbChannel.FundingOutpoint
 
-		// With the channel point known, retrieve the network channel
-		// ID from the database.
-		var chanID uint64
-		chanID, _ = graph.ChannelID(&chanPoint)
-
 		var peerOnline bool
 		if _, err := r.server.FindPeer(nodePub); err == nil {
 			peerOnline = true
@@ -2364,7 +2341,7 @@ func (r *rpcServer) ListChannels(ctx context.Context,
 		// Next, we'll determine whether we should add this channel to
 		// our list depending on the type of channels requested to us.
 		isActive := peerOnline && linkActive
-		isPublic := dbChannel.ChannelFlags&lnwire.FFAnnounceChannel != 0
+		channel := createRPCOpenChannel(r, graph, dbChannel, isActive)
 
 		// We'll only skip returning this channel if we were requested
 		// for a specific kind and this channel doesn't satisfy it.
@@ -2373,78 +2350,213 @@ func (r *rpcServer) ListChannels(ctx context.Context,
 			continue
 		case in.InactiveOnly && isActive:
 			continue
-		case in.PublicOnly && !isPublic:
+		case in.PublicOnly && channel.Private:
 			continue
-		case in.PrivateOnly && isPublic:
+		case in.PrivateOnly && !channel.Private:
 			continue
 		}
 
-		// As this is required for display purposes, we'll calculate
-		// the weight of the commitment transaction. We also add on the
-		// estimated weight of the witness to calculate the weight of
-		// the transaction if it were to be immediately unilaterally
-		// broadcast.
-		localCommit := dbChannel.LocalCommitment
-		utx := btcutil.NewTx(localCommit.CommitTx)
-		commitBaseWeight := blockchain.GetTransactionWeight(utx)
-		commitWeight := commitBaseWeight + input.WitnessCommitmentTxWeight
-
-		localBalance := localCommit.LocalBalance
-		remoteBalance := localCommit.RemoteBalance
-
-		// As an artifact of our usage of mSAT internally, either party
-		// may end up in a state where they're holding a fractional
-		// amount of satoshis which can't be expressed within the
-		// actual commitment output. Since we round down when going
-		// from mSAT -> SAT, we may at any point be adding an
-		// additional SAT to miners fees. As a result, we display a
-		// commitment fee that accounts for this externally.
-		var sumOutputs btcutil.Amount
-		for _, txOut := range localCommit.CommitTx.TxOut {
-			sumOutputs += btcutil.Amount(txOut.Value)
-		}
-		externalCommitFee := dbChannel.Capacity - sumOutputs
-
-		channel := &lnrpc.Channel{
-			Active:                isActive,
-			Private:               !isPublic,
-			RemotePubkey:          nodeID,
-			ChannelPoint:          chanPoint.String(),
-			ChanId:                chanID,
-			Capacity:              int64(dbChannel.Capacity),
-			LocalBalance:          int64(localBalance.ToSatoshis()),
-			RemoteBalance:         int64(remoteBalance.ToSatoshis()),
-			CommitFee:             int64(externalCommitFee),
-			CommitWeight:          commitWeight,
-			FeePerKw:              int64(localCommit.FeePerKw),
-			TotalSatoshisSent:     int64(dbChannel.TotalMSatSent.ToSatoshis()),
-			TotalSatoshisReceived: int64(dbChannel.TotalMSatReceived.ToSatoshis()),
-			NumUpdates:            localCommit.CommitHeight,
-			PendingHtlcs:          make([]*lnrpc.HTLC, len(localCommit.Htlcs)),
-			CsvDelay:              uint32(dbChannel.LocalChanCfg.CsvDelay),
-			Initiator:             dbChannel.IsInitiator,
-		}
-
-		for i, htlc := range localCommit.Htlcs {
-			var rHash [32]byte
-			copy(rHash[:], htlc.RHash[:])
-			channel.PendingHtlcs[i] = &lnrpc.HTLC{
-				Incoming:         htlc.Incoming,
-				Amount:           int64(htlc.Amt.ToSatoshis()),
-				HashLock:         rHash[:],
-				ExpirationHeight: htlc.RefundTimeout,
-			}
-
-			// Add the Pending Htlc Amount to UnsettledBalance field.
-			channel.UnsettledBalance += channel.PendingHtlcs[i].Amount
-		}
-
 		resp.Channels = append(resp.Channels, channel)
 	}
 
 	return resp, nil
 }
 
+// createRPCOpenChannel creates an *lnrpc.Channel from the *channeldb.Channel.
+func createRPCOpenChannel(r *rpcServer, graph *channeldb.ChannelGraph,
+	dbChannel *channeldb.OpenChannel, isActive bool) *lnrpc.Channel {
+
+	nodePub := dbChannel.IdentityPub
+	nodeID := hex.EncodeToString(nodePub.SerializeCompressed())
+	chanPoint := dbChannel.FundingOutpoint
+
+	// With the channel point known, retrieve the network channel
+	// ID from the database.
+	var chanID uint64
+	chanID, _ = graph.ChannelID(&chanPoint)
+
+	// Next, we'll determine whether the channel is public or not.
+	isPublic := dbChannel.ChannelFlags&lnwire.FFAnnounceChannel != 0
+
+	// As this is required for display purposes, we'll calculate
+	// the weight of the commitment transaction. We also add on the
+	// estimated weight of the witness to calculate the weight of
+	// the transaction if it were to be immediately unilaterally
+	// broadcast.
+	localCommit := dbChannel.LocalCommitment
+	utx := btcutil.NewTx(localCommit.CommitTx)
+	commitBaseWeight := blockchain.GetTransactionWeight(utx)
+	commitWeight := commitBaseWeight + input.WitnessCommitmentTxWeight
+
+	localBalance := localCommit.LocalBalance
+	remoteBalance := localCommit.RemoteBalance
+
+	// As an artifact of our usage of mSAT internally, either party
+	// may end up in a state where they're holding a fractional
+	// amount of satoshis which can't be expressed within the
+	// actual commitment output. Since we round down when going
+	// from mSAT -> SAT, we may at any point be adding an
+	// additional SAT to miners fees. As a result, we display a
+	// commitment fee that accounts for this externally.
+	var sumOutputs btcutil.Amount
+	for _, txOut := range localCommit.CommitTx.TxOut {
+		sumOutputs += btcutil.Amount(txOut.Value)
+	}
+	externalCommitFee := dbChannel.Capacity - sumOutputs
+
+	channel := &lnrpc.Channel{
+		Active:                isActive,
+		Private:               !isPublic,
+		RemotePubkey:          nodeID,
+		ChannelPoint:          chanPoint.String(),
+		ChanId:                chanID,
+		Capacity:              int64(dbChannel.Capacity),
+		LocalBalance:          int64(localBalance.ToSatoshis()),
+		RemoteBalance:         int64(remoteBalance.ToSatoshis()),
+		CommitFee:             int64(externalCommitFee),
+		CommitWeight:          commitWeight,
+		FeePerKw:              int64(localCommit.FeePerKw),
+		TotalSatoshisSent:     int64(dbChannel.TotalMSatSent.ToSatoshis()),
+		TotalSatoshisReceived: int64(dbChannel.TotalMSatReceived.ToSatoshis()),
+		NumUpdates:            localCommit.CommitHeight,
+		PendingHtlcs:          make([]*lnrpc.HTLC, len(localCommit.Htlcs)),
+		CsvDelay:              uint32(dbChannel.LocalChanCfg.CsvDelay),
+		Initiator:             dbChannel.IsInitiator,
+	}
+
+	for i, htlc := range localCommit.Htlcs {
+		var rHash [32]byte
+		copy(rHash[:], htlc.RHash[:])
+		channel.PendingHtlcs[i] = &lnrpc.HTLC{
+			Incoming:         htlc.Incoming,
+			Amount:           int64(htlc.Amt.ToSatoshis()),
+			HashLock:         rHash[:],
+			ExpirationHeight: htlc.RefundTimeout,
+		}
+
+		// Add the Pending Htlc Amount to UnsettledBalance field.
+		channel.UnsettledBalance += channel.PendingHtlcs[i].Amount
+	}
+
+	return channel
+}
+
+// createRPCClosedChannel creates an *lnrpc.ClosedChannelSummary from a
+// *channeldb.ChannelCloseSummary.
+func createRPCClosedChannel(
+	dbChannel *channeldb.ChannelCloseSummary) *lnrpc.ChannelCloseSummary {
+
+	nodePub := dbChannel.RemotePub
+	nodeID := hex.EncodeToString(nodePub.SerializeCompressed())
+
+	var closeType lnrpc.ChannelCloseSummary_ClosureType
+	switch dbChannel.CloseType {
+	case channeldb.CooperativeClose:
+		closeType = lnrpc.ChannelCloseSummary_COOPERATIVE_CLOSE
+	case channeldb.LocalForceClose:
+		closeType = lnrpc.ChannelCloseSummary_LOCAL_FORCE_CLOSE
+	case channeldb.RemoteForceClose:
+		closeType = lnrpc.ChannelCloseSummary_REMOTE_FORCE_CLOSE
+	case channeldb.BreachClose:
+		closeType = lnrpc.ChannelCloseSummary_BREACH_CLOSE
+	case channeldb.FundingCanceled:
+		closeType = lnrpc.ChannelCloseSummary_FUNDING_CANCELED
+	case channeldb.Abandoned:
+		closeType = lnrpc.ChannelCloseSummary_ABANDONED
+	}
+
+	return &lnrpc.ChannelCloseSummary{
+		Capacity:          int64(dbChannel.Capacity),
+		RemotePubkey:      nodeID,
+		CloseHeight:       dbChannel.CloseHeight,
+		CloseType:         closeType,
+		ChannelPoint:      dbChannel.ChanPoint.String(),
+		ChanId:            dbChannel.ShortChanID.ToUint64(),
+		SettledBalance:    int64(dbChannel.SettledBalance),
+		TimeLockedBalance: int64(dbChannel.TimeLockedBalance),
+		ChainHash:         dbChannel.ChainHash.String(),
+		ClosingTxHash:     dbChannel.ClosingTXID.String(),
+	}
+}
+
+// SubscribeChannelEvents returns a uni-directional stream (server -> client)
+// for notifying the client of newly active, inactive or closed channels.
+func (r *rpcServer) SubscribeChannelEvents(req *lnrpc.ChannelEventSubscription,
+	updateStream lnrpc.Lightning_SubscribeChannelEventsServer) error {
+
+	channelEventSub, err := r.server.channelNotifier.SubscribeChannelEvents()
+	if err != nil {
+		return err
+	}
+
+	// Ensure that the resources for the client is cleaned up once either
+	// the server, or client exits.
+	defer channelEventSub.Cancel()
+
+	graph := r.server.chanDB.ChannelGraph()
+
+	for {
+		select {
+		// A new update has been sent by the channel router, we'll
+		// marshal it into the form expected by the gRPC client, then
+		// send it off to the client(s).
+		case e := <-channelEventSub.Updates():
+			var update *lnrpc.ChannelEventUpdate
+			switch event := e.(type) {
+			case channelnotifier.OpenChannelEvent:
+				channel := createRPCOpenChannel(r, graph,
+					event.Channel, true)
+				update = &lnrpc.ChannelEventUpdate{
+					Type: lnrpc.ChannelEventUpdate_OPEN_CHANNEL,
+					Channel: &lnrpc.ChannelEventUpdate_OpenChannel{
+						OpenChannel: channel,
+					},
+				}
+			case channelnotifier.ClosedChannelEvent:
+				closedChannel := createRPCClosedChannel(event.CloseSummary)
+				update = &lnrpc.ChannelEventUpdate{
+					Type: lnrpc.ChannelEventUpdate_CLOSED_CHANNEL,
+					Channel: &lnrpc.ChannelEventUpdate_ClosedChannel{
+						ClosedChannel: closedChannel,
+					},
+				}
+			case channelnotifier.ActiveChannelEvent:
+				update = &lnrpc.ChannelEventUpdate{
+					Type: lnrpc.ChannelEventUpdate_ACTIVE_CHANNEL,
+					Channel: &lnrpc.ChannelEventUpdate_ActiveChannel{
+						ActiveChannel: &lnrpc.ChannelPoint{
+							FundingTxid: &lnrpc.ChannelPoint_FundingTxidBytes{
+								FundingTxidBytes: event.ChannelPoint.Hash[:],
+							},
+							OutputIndex: event.ChannelPoint.Index,
+						},
+					},
+				}
+			case channelnotifier.InactiveChannelEvent:
+				update = &lnrpc.ChannelEventUpdate{
+					Type: lnrpc.ChannelEventUpdate_INACTIVE_CHANNEL,
+					Channel: &lnrpc.ChannelEventUpdate_InactiveChannel{
+						InactiveChannel: &lnrpc.ChannelPoint{
+							FundingTxid: &lnrpc.ChannelPoint_FundingTxidBytes{
+								FundingTxidBytes: event.ChannelPoint.Hash[:],
+							},
+							OutputIndex: event.ChannelPoint.Index,
+						},
+					},
+				}
+			default:
+				return fmt.Errorf("unexpected channel event update: %v", event)
+			}
+
+			if err := updateStream.Send(update); err != nil {
+				return err
+			}
+		case <-r.quit:
+			return nil
+		}
+	}
+}
+
 // savePayment saves a successfully completed payment to the database for
 // historical record keeping.
 func (r *rpcServer) savePayment(route *routing.Route,

server.go

@@ -28,6 +28,7 @@ import (
 	"github.com/lightningnetwork/lnd/autopilot"
 	"github.com/lightningnetwork/lnd/brontide"
 	"github.com/lightningnetwork/lnd/channeldb"
+	"github.com/lightningnetwork/lnd/channelnotifier"
 	"github.com/lightningnetwork/lnd/contractcourt"
 	"github.com/lightningnetwork/lnd/discovery"
 	"github.com/lightningnetwork/lnd/htlcswitch"
@@ -145,6 +146,8 @@ type server struct {
 
 	invoices *invoices.InvoiceRegistry
 
+	channelNotifier *channelnotifier.ChannelNotifier
+
 	witnessBeacon contractcourt.WitnessBeacon
 
 	breachArbiter *breachArbiter
@@ -274,6 +277,8 @@ func newServer(listenAddrs []net.Addr, chanDB *channeldb.DB, cc *chainControl,
 
 		invoices: invoices.NewRegistry(chanDB, activeNetParams.Params),
 
+		channelNotifier: channelnotifier.New(chanDB),
+
 		identityPriv: privKey,
 		nodeSigner:   netann.NewNodeSigner(privKey),
 
@@ -357,6 +362,8 @@ func newServer(listenAddrs []net.Addr, chanDB *channeldb.DB, cc *chainControl,
 			htlcswitch.DefaultFwdEventInterval),
 		LogEventTicker: ticker.New(
 			htlcswitch.DefaultLogInterval),
+		NotifyActiveChannel:   s.channelNotifier.NotifyActiveChannelEvent,
+		NotifyInactiveChannel: s.channelNotifier.NotifyInactiveChannelEvent,
 	}, uint32(currentHeight))
 	if err != nil {
 		return nil, err
@@ -735,8 +742,9 @@ func newServer(listenAddrs []net.Addr, chanDB *channeldb.DB, cc *chainControl,
 		DisableChannel: func(op wire.OutPoint) error {
 			return s.announceChanStatus(op, true)
 		},
-		Sweeper:       s.sweeper,
-		SettleInvoice: s.invoices.SettleInvoice,
+		Sweeper:             s.sweeper,
+		SettleInvoice:       s.invoices.SettleInvoice,
+		NotifyClosedChannel: s.channelNotifier.NotifyClosedChannelEvent,
 	}, chanDB)
 
 	s.breachArbiter = newBreachArbiter(&BreachConfig{
@@ -922,9 +930,10 @@ func newServer(listenAddrs []net.Addr, chanDB *channeldb.DB, cc *chainControl,
 			// channel bandwidth.
 			return uint16(input.MaxHTLCNumber / 2)
 		},
-		ZombieSweeperInterval: 1 * time.Minute,
-		ReservationTimeout:    10 * time.Minute,
-		MinChanSize:           btcutil.Amount(cfg.MinChanSize),
+		ZombieSweeperInterval:  1 * time.Minute,
+		ReservationTimeout:     10 * time.Minute,
+		MinChanSize:            btcutil.Amount(cfg.MinChanSize),
+		NotifyOpenChannelEvent: s.channelNotifier.NotifyOpenChannelEvent,
 	})
 	if err != nil {
 		return nil, err
@@ -986,6 +995,9 @@ func (s *server) Start() error {
 	if err := s.cc.chainNotifier.Start(); err != nil {
 		return err
 	}
+	if err := s.channelNotifier.Start(); err != nil {
+		return err
+	}
 	if err := s.sphinx.Start(); err != nil {
 		return err
 	}
@@ -1083,6 +1095,7 @@ func (s *server) Stop() error {
 	s.authGossiper.Stop()
 	s.chainArb.Stop()
 	s.sweeper.Stop()
+	s.channelNotifier.Stop()
 	s.cc.wallet.Shutdown()
 	s.cc.chainView.Stop()
 	s.connMgr.Stop()

subscribe/subscribe.go

@@ -0,0 +1,216 @@
+package subscribe
+
+import (
+	"errors"
+	"sync"
+	"sync/atomic"
+
+	"github.com/lightningnetwork/lnd/queue"
+)
+
+// ErrServerShuttingDown is an error returned in case the server is in the
+// process of shutting down.
+var ErrServerShuttingDown = errors.New("subscription server shutting down")
+
+// Client is used to get notified about updates the caller has subscribed to,
+type Client struct {
+	// Cancel should be called in case the client no longer wants to
+	// subscribe for updates from the server.
+	Cancel func()
+
+	updates *queue.ConcurrentQueue
+	quit    chan struct{}
+}
+
+// Updates returns a read-only channel where the updates the client has
+// subscribed to will be delivered.
+func (c *Client) Updates() <-chan interface{} {
+	return c.updates.ChanOut()
+}
+
+// Quit is a channel that will be closed in case the server decides to no
+// longer deliver updates to this client.
+func (c *Client) Quit() <-chan struct{} {
+	return c.quit
+}
+
+// Server is a struct that manages a set of subscriptions and their
+// corresponding clients. Any update will be delivered to all active clients.
+type Server struct {
+	clientCounter uint64 // To be used atomically.
+
+	started uint32 // To be used atomically.
+	stopped uint32 // To be used atomically.
+
+	clients       map[uint64]*Client
+	clientUpdates chan *clientUpdate
+
+	updates chan interface{}
+
+	quit chan struct{}
+	wg   sync.WaitGroup
+}
+
+// clientUpdate is an internal message sent to the subscriptionHandler to
+// either register a new client for subscription or cancel an existing
+// subscription.
+type clientUpdate struct {
+	// cancel indicates if the update to the client is cancelling an
+	// existing client's subscription. If not then this update will be to
+	// subscribe a new client.
+	cancel bool
+
+	// clientID is the unique identifier for this client. Any further
+	// updates (deleting or adding) to this notification client will be
+	// dispatched according to the target clientID.
+	clientID uint64
+
+	// client is the new client that will receive updates. Will be nil in
+	// case this is a cancallation update.
+	client *Client
+}
+
+// NewServer returns a new Server.
+func NewServer() *Server {
+	return &Server{
+		clients:       make(map[uint64]*Client),
+		clientUpdates: make(chan *clientUpdate),
+		updates:       make(chan interface{}),
+		quit:          make(chan struct{}),
+	}
+}
+
+// Start starts the Server, making it ready to accept subscriptions and
+// updates.
+func (s *Server) Start() error {
+	if !atomic.CompareAndSwapUint32(&s.started, 0, 1) {
+		return nil
+	}
+
+	s.wg.Add(1)
+	go s.subscriptionHandler()
+
+	return nil
+}
+
+// Stop stops the server.
+func (s *Server) Stop() error {
+	if !atomic.CompareAndSwapUint32(&s.stopped, 0, 1) {
+		return nil
+	}
+
+	close(s.quit)
+	s.wg.Wait()
+
+	return nil
+}
+
+// Subscribe returns a Client that will receive updates any time the Server is
+// made aware of a new event.
+func (s *Server) Subscribe() (*Client, error) {
+	// We'll first atomically obtain the next ID for this client from the
+	// incrementing client ID counter.
+	clientID := atomic.AddUint64(&s.clientCounter, 1)
+
+	// Create the client that will be returned. The Cancel method is
+	// populated to send the cancellation intent to the
+	// subscriptionHandler.
+	client := &Client{
+		updates: queue.NewConcurrentQueue(20),
+		quit:    make(chan struct{}),
+		Cancel: func() {
+			select {
+			case s.clientUpdates <- &clientUpdate{
+				cancel:   true,
+				clientID: clientID,
+			}:
+			case <-s.quit:
+				return
+			}
+		},
+	}
+
+	select {
+	case s.clientUpdates <- &clientUpdate{
+		cancel:   false,
+		clientID: clientID,
+		client:   client,
+	}:
+	case <-s.quit:
+		return nil, ErrServerShuttingDown
+	}
+
+	return client, nil
+}
+
+// SendUpdate is called to send the passed update to all currently active
+// subscription clients.
+func (s *Server) SendUpdate(update interface{}) error {
+
+	select {
+	case s.updates <- update:
+		return nil
+	case <-s.quit:
+		return ErrServerShuttingDown
+	}
+}
+
+// subscriptionHandler is the main handler for the Server. It will handle
+// incoming updates and subscriptions, and forward the incoming updates to the
+// registered clients.
+//
+// NOTE: MUST be run as a goroutine.
+func (s *Server) subscriptionHandler() {
+	defer s.wg.Done()
+
+	for {
+		select {
+
+		// If a client update is received, the either a new
+		// subscription becomes active, or we cancel and existing one.
+		case update := <-s.clientUpdates:
+			clientID := update.clientID
+
+			// In case this is a cancellation, stop the client's
+			// underlying queue, and remove the client from the set
+			// of active subscription clients.
+			if update.cancel {
+				client, ok := s.clients[update.clientID]
+				if ok {
+					client.updates.Stop()
+					close(client.quit)
+					delete(s.clients, clientID)
+				}
+
+				continue
+			}
+
+			// If this was not a cancellation, start the underlying
+			// queue and add the client to our set of subscription
+			// clients. It will be notified about any new updates
+			// the server receives.
+			update.client.updates.Start()
+			s.clients[update.clientID] = update.client
+
+		// A new update was received, forward it to all active clients.
+		case upd := <-s.updates:
+			for _, client := range s.clients {
+				select {
+				case client.updates.ChanIn() <- upd:
+				case <-client.quit:
+				case <-s.quit:
+					return
+				}
+			}
+
+		// In case the server is shutting down, stop the clients and
+		// close the quit channels to notify them.
+		case <-s.quit:
+			for _, client := range s.clients {
+				client.updates.Stop()
+				close(client.quit)
+			}
+			return
+		}
+	}
+}

subscribe/subscribe_test.go

@@ -0,0 +1,110 @@
+package subscribe_test
+
+import (
+	"testing"
+	"time"
+
+	"github.com/lightningnetwork/lnd/subscribe"
+)
+
+// TestSubscribe tests that the subscription clients receive the updates sent
+// to them after they subscribe, and that cancelled clients don't get more
+// updates.
+func TestSubscribe(t *testing.T) {
+	t.Parallel()
+
+	server := subscribe.NewServer()
+	if err := server.Start(); err != nil {
+		t.Fatalf("unable to start server")
+	}
+
+	const numClients = 300
+	const numUpdates = 1000
+
+	var clients [numClients]*subscribe.Client
+
+	// Start by registering two thirds the clients.
+	for i := 0; i < numClients*2/3; i++ {
+		c, err := server.Subscribe()
+		if err != nil {
+			t.Fatalf("unable to subscribe: %v", err)
+		}
+
+		clients[i] = c
+	}
+
+	// Send half the updates.
+	for i := 0; i < numUpdates/2; i++ {
+		if err := server.SendUpdate(i); err != nil {
+			t.Fatalf("unable to send update")
+		}
+	}
+
+	// Register the rest of the clients.
+	for i := numClients * 2 / 3; i < numClients; i++ {
+		c, err := server.Subscribe()
+		if err != nil {
+			t.Fatalf("unable to subscribe: %v", err)
+		}
+
+		clients[i] = c
+	}
+
+	// Cancel one third of the clients.
+	for i := 0; i < numClients/3; i++ {
+		clients[i].Cancel()
+	}
+
+	// Send the rest of the updates.
+	for i := numUpdates / 2; i < numUpdates; i++ {
+		if err := server.SendUpdate(i); err != nil {
+			t.Fatalf("unable to send update")
+		}
+	}
+
+	// Now ensure the clients got the updates we expect.
+	for i, c := range clients {
+
+		var from, to int
+		switch {
+
+		// We expect the first third of the clients to quit, since they
+		// were cancelled.
+		case i < numClients/3:
+			select {
+			case <-c.Quit():
+				continue
+			case <-time.After(1 * time.Second):
+				t.Fatalf("cancelled client %v did not quit", i)
+			}
+
+		// The next third should receive all updates.
+		case i < numClients*2/3:
+			from = 0
+			to = numUpdates
+
+		// And finally the last third should receive the last half of
+		// the updates.
+		default:
+			from = numUpdates / 2
+			to = numUpdates
+		}
+
+		for cnt := from; cnt < to; cnt++ {
+			select {
+			case upd := <-c.Updates():
+				j := upd.(int)
+				if j != cnt {
+					t.Fatalf("expected %v, got %v, for "+
+						"client %v", cnt, j, i)
+				}
+
+			case <-time.After(1 * time.Second):
+				t.Fatalf("did not receive expected update %v "+
+					"for client %v", cnt, i)
+			}
+		}
+
+	}
+
+}