diff --git a/lntemp/node/watcher.go b/lntemp/node/watcher.go new file mode 100644 index 000000000..84d90a6f9 --- /dev/null +++ b/lntemp/node/watcher.go @@ -0,0 +1,663 @@ +package node + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "sync" + "time" + + "github.com/btcsuite/btcd/wire" + "github.com/lightningnetwork/lnd/lnrpc" + "github.com/lightningnetwork/lnd/lntemp/rpc" + "github.com/lightningnetwork/lnd/lntest" + "github.com/lightningnetwork/lnd/lntest/wait" +) + +type chanWatchType uint8 + +const ( + // watchOpenChannel specifies that this is a request to watch an open + // channel event. + watchOpenChannel chanWatchType = iota + + // watchCloseChannel specifies that this is a request to watch a close + // channel event. + watchCloseChannel + + // watchPolicyUpdate specifies that this is a request to watch a policy + // update event. + watchPolicyUpdate + + // TODO(yy): remove once temp tests is finished. + DefaultTimeout = lntest.DefaultTimeout +) + +// closeChanWatchRequest is a request to the lightningNetworkWatcher to be +// notified once it's detected within the test Lightning Network, that a +// channel has either been added or closed. +type chanWatchRequest struct { + chanPoint wire.OutPoint + + chanWatchType chanWatchType + + eventChan chan struct{} + + advertisingNode string + policy *lnrpc.RoutingPolicy + includeUnannounced bool +} + +// nodeWatcher is a topology watcher for a HarnessNode. It keeps track of all +// the topology updates seen in a given node, including NodeUpdate, +// ChannelEdgeUpdate, and ClosedChannelUpdate. +type nodeWatcher struct { + // rpc is the RPC clients used for the current node. + rpc *rpc.HarnessRPC + + // state is the node's current state. + state *State + + // chanWatchRequests receives a request for watching a particular event + // for a given channel. + chanWatchRequests chan *chanWatchRequest + + // For each outpoint, we'll track an integer which denotes the number + // of edges seen for that channel within the network. When this number + // reaches 2, then it means that both edge advertisements has + // propagated through the network. + // openChanWatchers map[wire.OutPoint][]chan struct{} + openChanWatchers *sync.Map + + // closeChanWatchers map[wire.OutPoint][]chan struct{} + closeChanWatchers *sync.Map + + wg sync.WaitGroup +} + +func newNodeWatcher(rpc *rpc.HarnessRPC, state *State) *nodeWatcher { + return &nodeWatcher{ + rpc: rpc, + state: state, + chanWatchRequests: make(chan *chanWatchRequest, 100), + openChanWatchers: &sync.Map{}, + closeChanWatchers: &sync.Map{}, + } +} + +// GetNumChannelUpdates reads the num of channel updates inside a lock and +// returns the value. +func (nw *nodeWatcher) GetNumChannelUpdates(op wire.OutPoint) int { + result, ok := nw.state.numChanUpdates.Load(op) + if ok { + return result.(int) + } + return 0 +} + +// GetPolicyUpdates returns the node's policyUpdates state. +func (nw *nodeWatcher) GetPolicyUpdates(op wire.OutPoint) PolicyUpdate { + result, ok := nw.state.policyUpdates.Load(op) + if ok { + return result.(PolicyUpdate) + } + return nil +} + +// GetNodeUpdates reads the node updates inside a lock and returns the value. +func (nw *nodeWatcher) GetNodeUpdates(pubkey string) []*lnrpc.NodeUpdate { + result, ok := nw.state.nodeUpdates.Load(pubkey) + if ok { + return result.([]*lnrpc.NodeUpdate) + } + return nil +} + +// WaitForNumChannelUpdates will block until a given number of updates has been +// seen in the node's network topology. +func (nw *nodeWatcher) WaitForNumChannelUpdates(op wire.OutPoint, + expected int) error { + + checkNumUpdates := func() error { + num := nw.GetNumChannelUpdates(op) + if num >= expected { + return nil + } + + return fmt.Errorf("timeout waiting for num channel updates, "+ + "want %d, got %d", expected, num) + } + + return wait.NoError(checkNumUpdates, DefaultTimeout) +} + +// WaitForNumNodeUpdates will block until a given number of node updates has +// been seen in the node's network topology. +func (nw *nodeWatcher) WaitForNumNodeUpdates(pubkey string, + expected int) error { + + checkNumUpdates := func() error { + num := len(nw.GetNodeUpdates(pubkey)) + if num >= expected { + return nil + } + + return fmt.Errorf("timeout waiting for num node updates, "+ + "want %d, got %d", expected, num) + } + + return wait.NoError(checkNumUpdates, DefaultTimeout) +} + +// WaitForChannelOpen will block until a channel with the target outpoint is +// seen as being fully advertised within the network. A channel is considered +// "fully advertised" once both of its directional edges has been advertised in +// the node's network topology. +func (nw *nodeWatcher) WaitForChannelOpen(chanPoint *lnrpc.ChannelPoint) error { + op := nw.rpc.MakeOutpoint(chanPoint) + eventChan := make(chan struct{}) + nw.chanWatchRequests <- &chanWatchRequest{ + chanPoint: op, + eventChan: eventChan, + chanWatchType: watchOpenChannel, + } + + timer := time.After(DefaultTimeout) + select { + case <-eventChan: + return nil + + case <-timer: + updates, err := syncMapToJSON(nw.state.openChans) + if err != nil { + return err + } + + return fmt.Errorf("channel:%s not heard before timeout: "+ + "node has heard: %s", op, updates) + } +} + +// WaitForChannelClose will block until a channel with the target outpoint is +// seen as closed within the node's network topology. A channel is considered +// closed once a transaction spending the funding outpoint is seen within a +// confirmed block. +func (nw *nodeWatcher) WaitForChannelClose( + chanPoint *lnrpc.ChannelPoint) (*lnrpc.ClosedChannelUpdate, error) { + + op := nw.rpc.MakeOutpoint(chanPoint) + eventChan := make(chan struct{}) + nw.chanWatchRequests <- &chanWatchRequest{ + chanPoint: op, + eventChan: eventChan, + chanWatchType: watchCloseChannel, + } + + timer := time.After(DefaultTimeout) + select { + case <-eventChan: + closedChan, ok := nw.state.closedChans.Load(op) + if !ok { + return nil, fmt.Errorf("channel:%s expected to find "+ + "a closed channel in node's state:%s", op, + nw.state) + } + return closedChan.(*lnrpc.ClosedChannelUpdate), nil + + case <-timer: + return nil, fmt.Errorf("channel:%s not closed before timeout: "+ + "%s", op, nw.state) + } +} + +// syncMapToJSON is a helper function that creates json bytes from the sync.Map +// used in the node. Expect the sync.Map to have map[string]interface. +func syncMapToJSON(state *sync.Map) ([]byte, error) { + m := map[string]interface{}{} + state.Range(func(k, v interface{}) bool { + op := k.(wire.OutPoint) + m[op.String()] = v + return true + }) + policies, err := json.MarshalIndent(m, "", "\t") + if err != nil { + return nil, fmt.Errorf("encode polices err: %v", err) + } + + return policies, nil +} + +// topologyWatcher is a goroutine which is able to dispatch notifications once +// it has been observed that a target channel has been closed or opened within +// the network. In order to dispatch these notifications, the +// GraphTopologySubscription client exposed as part of the gRPC interface is +// used. +// +// NOTE: must be run as a goroutine. +func (nw *nodeWatcher) topologyWatcher(ctxb context.Context, + started chan error) { + + graphUpdates := make(chan *lnrpc.GraphTopologyUpdate) + + client, err := nw.newTopologyClient(ctxb) + started <- err + + // Exit if there's an error. + if err != nil { + return + } + + // Start a goroutine to receive graph updates. + nw.wg.Add(1) + go func() { + defer nw.wg.Done() + + // With the client being created, we now start receiving the + // updates. + err = nw.receiveTopologyClientStream(ctxb, client, graphUpdates) + if err != nil { + started <- fmt.Errorf("receiveTopologyClientStream "+ + "got err: %v", err) + } + }() + + for { + select { + // A new graph update has just been received, so we'll examine + // the current set of registered clients to see if we can + // dispatch any requests. + case graphUpdate := <-graphUpdates: + nw.handleChannelEdgeUpdates(graphUpdate.ChannelUpdates) + nw.handleClosedChannelUpdate(graphUpdate.ClosedChans) + nw.handleNodeUpdates(graphUpdate.NodeUpdates) + + // A new watch request, has just arrived. We'll either be able + // to dispatch immediately, or need to add the client for + // processing later. + case watchRequest := <-nw.chanWatchRequests: + switch watchRequest.chanWatchType { + case watchOpenChannel: + // TODO(roasbeef): add update type also, checks + // for multiple of 2 + nw.handleOpenChannelWatchRequest(watchRequest) + + case watchCloseChannel: + nw.handleCloseChannelWatchRequest(watchRequest) + + case watchPolicyUpdate: + nw.handlePolicyUpdateWatchRequest(watchRequest) + } + + case <-ctxb.Done(): + return + } + } +} + +func (nw *nodeWatcher) handleNodeUpdates(updates []*lnrpc.NodeUpdate) { + for _, nodeUpdate := range updates { + nw.updateNodeStateNodeUpdates(nodeUpdate) + } +} + +// handleChannelEdgeUpdates takes a series of channel edge updates, extracts +// the outpoints, and saves them to harness node's internal state. +func (nw *nodeWatcher) handleChannelEdgeUpdates( + updates []*lnrpc.ChannelEdgeUpdate) { + + // For each new channel, we'll increment the number of edges seen by + // one. + for _, newChan := range updates { + op := nw.rpc.MakeOutpoint(newChan.ChanPoint) + + // Update the num of channel updates. + nw.updateNodeStateNumChanUpdates(op) + + // Update the open channels. + nw.updateNodeStateOpenChannel(op, newChan) + + // Check whether there's a routing policy update. If so, save + // it to the node state. + if newChan.RoutingPolicy != nil { + nw.updateNodeStatePolicy(op, newChan) + } + } +} + +// updateNodeStateNumChanUpdates updates the internal state of the node +// regarding the num of channel update seen. +func (nw *nodeWatcher) updateNodeStateNumChanUpdates(op wire.OutPoint) { + var oldNum int + result, ok := nw.state.numChanUpdates.Load(op) + if ok { + oldNum = result.(int) + } + nw.state.numChanUpdates.Store(op, oldNum+1) +} + +// updateNodeStateNodeUpdates updates the internal state of the node regarding +// the node updates seen. +func (nw *nodeWatcher) updateNodeStateNodeUpdates(update *lnrpc.NodeUpdate) { + var oldUpdates []*lnrpc.NodeUpdate + + result, ok := nw.state.nodeUpdates.Load(update.IdentityKey) + if ok { + oldUpdates = result.([]*lnrpc.NodeUpdate) + } + nw.state.nodeUpdates.Store( + update.IdentityKey, append(oldUpdates, update), + ) +} + +// updateNodeStateOpenChannel updates the internal state of the node regarding +// the open channels. +func (nw *nodeWatcher) updateNodeStateOpenChannel(op wire.OutPoint, + newChan *lnrpc.ChannelEdgeUpdate) { + + // Load the old updates the node has heard so far. + updates := make([]*OpenChannelUpdate, 0) + result, ok := nw.state.openChans.Load(op) + if ok { + updates = result.([]*OpenChannelUpdate) + } + + // Create a new update based on this newChan. + newUpdate := &OpenChannelUpdate{ + AdvertisingNode: newChan.AdvertisingNode, + ConnectingNode: newChan.ConnectingNode, + Timestamp: time.Now(), + } + + // Update the node's state. + updates = append(updates, newUpdate) + nw.state.openChans.Store(op, updates) + + // For this new channel, if the number of edges seen is less + // than two, then the channel hasn't been fully announced yet. + if len(updates) < 2 { + return + } + + // Otherwise, we'll notify all the registered watchers and + // remove the dispatched watchers. + watcherResult, loaded := nw.openChanWatchers.LoadAndDelete(op) + if !loaded { + return + } + events := watcherResult.([]chan struct{}) + for _, eventChan := range events { + close(eventChan) + } +} + +// updateNodeStatePolicy updates the internal state of the node regarding the +// policy updates. +func (nw *nodeWatcher) updateNodeStatePolicy(op wire.OutPoint, + newChan *lnrpc.ChannelEdgeUpdate) { + + // Init an empty policy map and overwrite it if the channel point can + // be found in the node's policyUpdates. + policies := make(PolicyUpdate) + result, ok := nw.state.policyUpdates.Load(op) + if ok { + policies = result.(PolicyUpdate) + } + + node := newChan.AdvertisingNode + + // Append the policy to the slice and update the node's state. + newPolicy := PolicyUpdateInfo{ + newChan.RoutingPolicy, newChan.ConnectingNode, time.Now(), + } + policies[node] = append(policies[node], &newPolicy) + nw.state.policyUpdates.Store(op, policies) +} + +// handleOpenChannelWatchRequest processes a watch open channel request by +// checking the number of the edges seen for a given channel point. If the +// number is no less than 2 then the channel is considered open. Otherwise, we +// will attempt to find it in its channel graph. If neither can be found, the +// request is added to a watch request list than will be handled by +// handleChannelEdgeUpdates. +func (nw *nodeWatcher) handleOpenChannelWatchRequest(req *chanWatchRequest) { + targetChan := req.chanPoint + + // If this is an open request, then it can be dispatched if the number + // of edges seen for the channel is at least two. + result, ok := nw.state.openChans.Load(targetChan) + if ok && len(result.([]*OpenChannelUpdate)) >= 2 { + close(req.eventChan) + return + } + + // Otherwise, we'll add this to the list of open channel watchers for + // this out point. + oldWatchers := make([]chan struct{}, 0) + watchers, ok := nw.openChanWatchers.Load(targetChan) + if ok { + oldWatchers = watchers.([]chan struct{}) + } + nw.openChanWatchers.Store( + targetChan, append(oldWatchers, req.eventChan), + ) +} + +// handleClosedChannelUpdate takes a series of closed channel updates, extracts +// the outpoints, saves them to harness node's internal state, and notifies all +// registered clients. +func (nw *nodeWatcher) handleClosedChannelUpdate( + updates []*lnrpc.ClosedChannelUpdate) { + + // For each channel closed, we'll mark that we've detected a channel + // closure while lnd was pruning the channel graph. + for _, closedChan := range updates { + op := nw.rpc.MakeOutpoint(closedChan.ChanPoint) + + nw.state.closedChans.Store(op, closedChan) + + // As the channel has been closed, we'll notify all register + // watchers. + result, loaded := nw.closeChanWatchers.LoadAndDelete(op) + if !loaded { + continue + } + + watchers := result.([]chan struct{}) + for _, eventChan := range watchers { + close(eventChan) + } + } +} + +// handleCloseChannelWatchRequest processes a watch close channel request by +// checking whether the given channel point can be found in the node's internal +// state. If not, the request is added to a watch request list than will be +// handled by handleCloseChannelWatchRequest. +func (nw *nodeWatcher) handleCloseChannelWatchRequest(req *chanWatchRequest) { + targetChan := req.chanPoint + + // If this is a close request, then it can be immediately dispatched if + // we've already seen a channel closure for this channel. + if _, ok := nw.state.closedChans.Load(targetChan); ok { + close(req.eventChan) + return + } + + // Otherwise, we'll add this to the list of close channel watchers for + // this out point. + oldWatchers := make([]chan struct{}, 0) + result, ok := nw.closeChanWatchers.Load(targetChan) + if ok { + oldWatchers = result.([]chan struct{}) + } + + nw.closeChanWatchers.Store( + targetChan, append(oldWatchers, req.eventChan), + ) +} + +// handlePolicyUpdateWatchRequest checks that if the expected policy can be +// found either in the node's interval state or describe graph response. If +// found, it will signal the request by closing the event channel. Otherwise it +// does nothing but returns nil. +func (nw *nodeWatcher) handlePolicyUpdateWatchRequest(req *chanWatchRequest) { + op := req.chanPoint + + var policies []*PolicyUpdateInfo + + // Get a list of known policies for this chanPoint+advertisingNode + // combination. Start searching in the node state first. + result, ok := nw.state.policyUpdates.Load(op) + if ok { + policyMap := result.(PolicyUpdate) + policies, ok = policyMap[req.advertisingNode] + if !ok { + return + } + } else { + // If it cannot be found in the node state, try searching it + // from the node's DescribeGraph. + policyMap := nw.getChannelPolicies(req.includeUnannounced) + result, ok := policyMap[op.String()][req.advertisingNode] + if !ok { + return + } + for _, policy := range result { + // Use empty from node to mark it being loaded from + // DescribeGraph. + policies = append( + policies, &PolicyUpdateInfo{ + policy, "", time.Now(), + }, + ) + } + } + + // Check if the latest policy is matched. + policy := policies[len(policies)-1] + if checkChannelPolicy(policy.RoutingPolicy, req.policy) == nil { + close(req.eventChan) + return + } +} + +type topologyClient lnrpc.Lightning_SubscribeChannelGraphClient + +// newTopologyClient creates a topology client. +func (nw *nodeWatcher) newTopologyClient( + ctx context.Context) (topologyClient, error) { + + req := &lnrpc.GraphTopologySubscription{} + client, err := nw.rpc.LN.SubscribeChannelGraph(ctx, req) + if err != nil { + return nil, fmt.Errorf("%s: unable to create topology client: "+ + "%v (%s)", nw.rpc.Name, err, time.Now().String()) + } + + return client, nil +} + +// receiveTopologyClientStream takes a topologyClient and receives graph +// updates. +// +// NOTE: must be run as a goroutine. +func (nw *nodeWatcher) receiveTopologyClientStream(ctxb context.Context, + client topologyClient, + receiver chan *lnrpc.GraphTopologyUpdate) error { + + for { + update, err := client.Recv() + + switch { + case err == nil: + // Good case. We will send the update to the receiver. + + case strings.Contains(err.Error(), "EOF"): + // End of subscription stream. Do nothing and quit. + return nil + + case strings.Contains(err.Error(), context.Canceled.Error()): + // End of subscription stream. Do nothing and quit. + return nil + + default: + // An expected error is returned, return and leave it + // to be handled by the caller. + return fmt.Errorf("graph subscription err: %w", err) + } + + // Send the update or quit. + select { + case receiver <- update: + case <-ctxb.Done(): + return nil + } + } +} + +// getChannelPolicies queries the channel graph and formats the policies into +// the format defined in type policyUpdateMap. +func (nw *nodeWatcher) getChannelPolicies(include bool) policyUpdateMap { + req := &lnrpc.ChannelGraphRequest{IncludeUnannounced: include} + graph := nw.rpc.DescribeGraph(req) + + policyUpdates := policyUpdateMap{} + + for _, e := range graph.Edges { + policies := policyUpdates[e.ChanPoint] + + // If the map[op] is nil, we need to initialize the map first. + if policies == nil { + policies = make(map[string][]*lnrpc.RoutingPolicy) + } + + if e.Node1Policy != nil { + policies[e.Node1Pub] = append( + policies[e.Node1Pub], e.Node1Policy, + ) + } + + if e.Node2Policy != nil { + policies[e.Node2Pub] = append( + policies[e.Node2Pub], e.Node2Policy, + ) + } + + policyUpdates[e.ChanPoint] = policies + } + + return policyUpdates +} + +// checkChannelPolicy checks that the policy matches the expected one. +func checkChannelPolicy(policy, expectedPolicy *lnrpc.RoutingPolicy) error { + if policy.FeeBaseMsat != expectedPolicy.FeeBaseMsat { + return fmt.Errorf("expected base fee %v, got %v", + expectedPolicy.FeeBaseMsat, policy.FeeBaseMsat) + } + if policy.FeeRateMilliMsat != expectedPolicy.FeeRateMilliMsat { + return fmt.Errorf("expected fee rate %v, got %v", + expectedPolicy.FeeRateMilliMsat, + policy.FeeRateMilliMsat) + } + if policy.TimeLockDelta != expectedPolicy.TimeLockDelta { + return fmt.Errorf("expected time lock delta %v, got %v", + expectedPolicy.TimeLockDelta, + policy.TimeLockDelta) + } + if policy.MinHtlc != expectedPolicy.MinHtlc { + return fmt.Errorf("expected min htlc %v, got %v", + expectedPolicy.MinHtlc, policy.MinHtlc) + } + if policy.MaxHtlcMsat != expectedPolicy.MaxHtlcMsat { + return fmt.Errorf("expected max htlc %v, got %v", + expectedPolicy.MaxHtlcMsat, policy.MaxHtlcMsat) + } + if policy.Disabled != expectedPolicy.Disabled { + return errors.New("edge should be disabled but isn't") + } + + return nil +}