chanfitness: refactor to store channels by peer

When dealing with online events, we actually need to track our events
by peer, not by channel. All we need to track channels is to have a
set of online events for a peer which at least contain those events.
This change refactors chanfitness to track by peer.

chanfitness/chaneventstore.go

@@ -32,6 +32,10 @@ var (
 	// ErrChannelNotFound is returned when a query is made for a channel
 	// that the event store does not have knowledge of.
 	ErrChannelNotFound = errors.New("channel not found in event store")
+
+	// ErrPeerNotFound is returned when a query is made for a channel
+	// that has a peer that the event store is not currently tracking.
+	ErrPeerNotFound = errors.New("peer not found in event store")
 )
 
 // ChannelEventStore maintains a set of event logs for the node's channels to
@@ -39,12 +43,8 @@ var (
 type ChannelEventStore struct {
 	cfg *Config
 
-	// channels maps channel points to event logs.
-	channels map[wire.OutPoint]*chanEventLog
-
-	// peers tracks the current online status of peers based on online
-	// and offline events.
-	peers map[route.Vertex]bool
+	// peers tracks all of our currently monitored peers and their channels.
+	peers map[route.Vertex]peerMonitor
 
 	// chanInfoRequests serves requests for information about our channel.
 	chanInfoRequests chan channelInfoRequest
@@ -77,6 +77,7 @@ type Config struct {
 }
 
 type channelInfoRequest struct {
+	peer         route.Vertex
 	channelPoint wire.OutPoint
 	responseChan chan channelInfoResponse
 }
@@ -92,8 +93,7 @@ type channelInfoResponse struct {
 func NewChannelEventStore(config *Config) *ChannelEventStore {
 	store := &ChannelEventStore{
 		cfg:              config,
-		channels:         make(map[wire.OutPoint]*chanEventLog),
-		peers:            make(map[route.Vertex]bool),
+		peers:            make(map[route.Vertex]peerMonitor),
 		chanInfoRequests: make(chan channelInfoRequest),
 		quit:             make(chan struct{}),
 	}
@@ -173,60 +173,56 @@ func (c *ChannelEventStore) Stop() {
 	c.wg.Wait()
 }
 
-// addChannel adds a new channel to the ChannelEventStore's map of channels with
-// an initial peer online state (if the peer is online). If the channel is
-// already present in the map, the function returns early. This function should
-// be called to add existing channels on startup and when open channel events
-// are observed.
+// addChannel checks whether we are already tracking a channel's peer, creates a
+// new peer log to track it if we are not yet monitoring it, and adds the
+// channel.
 func (c *ChannelEventStore) addChannel(channelPoint wire.OutPoint,
 	peer route.Vertex) {
 
-	// Check for the unexpected case where the channel is already in the
-	// store.
-	_, ok := c.channels[channelPoint]
-	if ok {
-		log.Errorf("Channel %v duplicated in channel store",
-			channelPoint)
-		return
+	peerMonitor, ok := c.peers[peer]
+	if !ok {
+		peerMonitor = newPeerLog(c.cfg.Clock)
+		c.peers[peer] = peerMonitor
 	}
 
-	// Create an event log for the channel.
-	eventLog := newEventLog(channelPoint, peer, c.cfg.Clock)
-
-	// If the peer is already online, add a peer online event to record
-	// the starting state of the peer.
-	if c.peers[peer] {
-		eventLog.add(peerOnlineEvent)
+	if err := peerMonitor.addChannel(channelPoint); err != nil {
+		log.Errorf("could not add channel: %v", err)
 	}
-
-	c.channels[channelPoint] = eventLog
 }
 
 // closeChannel records a closed time for a channel, and returns early is the
-// channel is not known to the event store.
-func (c *ChannelEventStore) closeChannel(channelPoint wire.OutPoint) {
-	// Check for the unexpected case where the channel is unknown to the
-	// store.
-	eventLog, ok := c.channels[channelPoint]
+// channel is not known to the event store. We log warnings (rather than errors)
+// when we cannot find a peer/channel because channels that we restore from a
+// static channel backup do not have their open notified, so the event store
+// never learns about them, but they are closed using the regular flow so we
+// will try to remove them on close. At present, we cannot easily distinguish
+// between these closes and others.
+func (c *ChannelEventStore) closeChannel(channelPoint wire.OutPoint,
+	peer route.Vertex) {
+
+	peerMonitor, ok := c.peers[peer]
 	if !ok {
-		log.Errorf("Close channel %v unknown to store", channelPoint)
+		log.Warnf("peer not known to store: %v", peer)
 		return
 	}
 
-	eventLog.close()
+	if err := peerMonitor.removeChannel(channelPoint); err != nil {
+		log.Warnf("could not remove channel: %v", err)
+	}
 }
 
-// peerEvent adds a peer online or offline event to all channels we currently
-// have open with a peer.
-func (c *ChannelEventStore) peerEvent(peer route.Vertex, event eventType) {
-	// Track current online status of peers in the channelEventStore.
-	c.peers[peer] = event == peerOnlineEvent
-
-	for _, eventLog := range c.channels {
-		if eventLog.peer == peer {
-			eventLog.add(event)
-		}
+// peerEvent creates a peer monitor for a peer if we do not currently have
+// one, and adds an online event to it.
+func (c *ChannelEventStore) peerEvent(peer route.Vertex, online bool) {
+	// If we are not currently tracking events for this peer, add a peer
+	// log for it.
+	peerMonitor, ok := c.peers[peer]
+	if !ok {
+		peerMonitor = newPeerLog(c.cfg.Clock)
+		c.peers[peer] = peerMonitor
 	}
+
+	peerMonitor.onlineEvent(online)
 }
 
 // subscriptions abstracts away from subscription clients to allow for mocking.
@@ -268,7 +264,19 @@ func (c *ChannelEventStore) consume(subscriptions *subscriptions) {
 			// A channel has been closed, we must remove the channel
 			// from the store and record a channel closed event.
 			case channelnotifier.ClosedChannelEvent:
-				c.closeChannel(event.CloseSummary.ChanPoint)
+				compressed := event.CloseSummary.RemotePub.SerializeCompressed()
+				peerKey, err := route.NewVertexFromBytes(
+					compressed,
+				)
+				if err != nil {
+					log.Errorf("Could not get vertex "+
+						"from: %v", compressed)
+					continue
+				}
+
+				c.closeChannel(
+					event.CloseSummary.ChanPoint, peerKey,
+				)
 			}
 
 		// Process peer online and offline events.
@@ -278,13 +286,13 @@ func (c *ChannelEventStore) consume(subscriptions *subscriptions) {
 			// and should record an online event for any channels
 			// with that peer.
 			case peernotifier.PeerOnlineEvent:
-				c.peerEvent(event.PubKey, peerOnlineEvent)
+				c.peerEvent(event.PubKey, true)
 
 			// We have lost a connection with our peer, and should
 			// record an offline event for any channels with that
 			// peer.
 			case peernotifier.PeerOfflineEvent:
-				c.peerEvent(event.PubKey, peerOfflineEvent)
+				c.peerEvent(event.PubKey, false)
 			}
 
 		// Serve all requests for channel lifetime.
@@ -314,10 +322,11 @@ type ChannelInfo struct {
 }
 
 // GetChanInfo gets all the information we have on a channel in the event store.
-func (c *ChannelEventStore) GetChanInfo(channelPoint wire.OutPoint) (
-	*ChannelInfo, error) {
+func (c *ChannelEventStore) GetChanInfo(channelPoint wire.OutPoint,
+	peer route.Vertex) (*ChannelInfo, error) {
 
 	request := channelInfoRequest{
+		peer:         peer,
 		channelPoint: channelPoint,
 		responseChan: make(chan channelInfoResponse),
 	}
@@ -347,26 +356,18 @@ func (c *ChannelEventStore) GetChanInfo(channelPoint wire.OutPoint) (
 func (c *ChannelEventStore) getChanInfo(req channelInfoRequest) (*ChannelInfo,
 	error) {
 
-	// Look for the channel in our current set.
-	channel, ok := c.channels[req.channelPoint]
+	peerMonitor, ok := c.peers[req.peer]
 	if !ok {
-		return nil, ErrChannelNotFound
+		return nil, ErrPeerNotFound
 	}
 
-	// If our channel is not closed, we want to calculate uptime until the
-	// present.
-	endTime := channel.closedAt
-	if endTime.IsZero() {
-		endTime = c.cfg.Clock.Now()
-	}
-
-	uptime, err := channel.uptime(channel.openedAt, endTime)
+	lifetime, uptime, err := peerMonitor.channelUptime(req.channelPoint)
 	if err != nil {
 		return nil, err
 	}
 
 	return &ChannelInfo{
-		Lifetime: endTime.Sub(channel.openedAt),
+		Lifetime: lifetime,
 		Uptime:   uptime,
 	}, nil
 }