diff --git a/discovery/gossiper.go b/discovery/gossiper.go index bd2d3b605..8bb5a5214 100644 --- a/discovery/gossiper.go +++ b/discovery/gossiper.go @@ -658,6 +658,7 @@ func (d *AuthenticatedGossiper) Stop() error { func (d *AuthenticatedGossiper) stop() { log.Info("Authenticated Gossiper is stopping") + defer log.Info("Authenticated Gossiper stopped") d.blockEpochs.Cancel() @@ -1187,12 +1188,6 @@ func (d *AuthenticatedGossiper) networkHandler() { announcement.msg.MsgType(), announcement.isRemote) - // We should only broadcast this message forward if it - // originated from us or it wasn't received as part of - // our initial historical sync. - shouldBroadcast := !announcement.isRemote || - d.syncMgr.IsGraphSynced() - switch announcement.msg.(type) { // Channel announcement signatures are amongst the only // messages that we'll process serially. @@ -1231,70 +1226,9 @@ func (d *AuthenticatedGossiper) networkHandler() { validationBarrier.InitJobDependencies(announcement.msg) d.wg.Add(1) - go func() { - defer d.wg.Done() - defer validationBarrier.CompleteJob() - - // If this message has an existing dependency, - // then we'll wait until that has been fully - // validated before we proceed. - err := validationBarrier.WaitForDependants( - announcement.msg, - ) - if err != nil { - if !routing.IsError( - err, - routing.ErrVBarrierShuttingDown, - routing.ErrParentValidationFailed, - ) { - - log.Warnf("unexpected error "+ - "during validation "+ - "barrier shutdown: %v", - err) - } - announcement.err <- err - return - } - - // Process the network announcement to - // determine if this is either a new - // announcement from our PoV or an edges to a - // prior vertex/edge we previously proceeded. - emittedAnnouncements, allowDependents := d.processNetworkAnnouncement( - announcement, - ) - - log.Tracef("Processed network message %s, "+ - "returned len(announcements)=%v, "+ - "allowDependents=%v", - announcement.msg.MsgType(), - len(emittedAnnouncements), - allowDependents) - - // If this message had any dependencies, then - // we can now signal them to continue. - validationBarrier.SignalDependants( - announcement.msg, allowDependents, - ) - - // If the announcement was accepted, then add - // the emitted announcements to our announce - // batch to be broadcast once the trickle timer - // ticks gain. - if emittedAnnouncements != nil && shouldBroadcast { - // TODO(roasbeef): exclude peer that - // sent. - announcements.AddMsgs( - emittedAnnouncements..., - ) - } else if emittedAnnouncements != nil { - log.Trace("Skipping broadcast of " + - "announcements received " + - "during initial graph sync") - } - - }() + go d.handleNetworkMessages( + announcement, &announcements, validationBarrier, + ) // The trickle timer has ticked, which indicates we should // flush to the network the pending batch of new announcements @@ -1359,6 +1293,67 @@ func (d *AuthenticatedGossiper) networkHandler() { } } +// handleNetworkMessages is responsible for waiting for dependencies for a +// given network message and processing the message. Once processed, it will +// signal its dependants and add the new announcements to the announce batch. +// +// NOTE: must be run as a goroutine. +func (d *AuthenticatedGossiper) handleNetworkMessages(nMsg *networkMsg, + deDuped *deDupedAnnouncements, vb *routing.ValidationBarrier) { + + defer d.wg.Done() + defer vb.CompleteJob() + + // We should only broadcast this message forward if it originated from + // us or it wasn't received as part of our initial historical sync. + shouldBroadcast := !nMsg.isRemote || d.syncMgr.IsGraphSynced() + + // If this message has an existing dependency, then we'll wait until + // that has been fully validated before we proceed. + err := vb.WaitForDependants(nMsg.msg) + if err != nil { + log.Debugf("Validating network message %s got err: %v", + nMsg.msg.MsgType(), err) + + if !routing.IsError( + err, + routing.ErrVBarrierShuttingDown, + routing.ErrParentValidationFailed, + ) { + + log.Warnf("unexpected error during validation "+ + "barrier shutdown: %v", err) + } + nMsg.err <- err + + return + } + + // Process the network announcement to determine if this is either a + // new announcement from our PoV or an edges to a prior vertex/edge we + // previously proceeded. + newAnns, allow := d.processNetworkAnnouncement(nMsg) + + log.Tracef("Processed network message %s, returned "+ + "len(announcements)=%v, allowDependents=%v", + nMsg.msg.MsgType(), len(newAnns), allow) + + // If this message had any dependencies, then we can now signal them to + // continue. + vb.SignalDependants(nMsg.msg, allow) + + // If the announcement was accepted, then add the emitted announcements + // to our announce batch to be broadcast once the trickle timer ticks + // gain. + if newAnns != nil && shouldBroadcast { + // TODO(roasbeef): exclude peer that sent. + deDuped.AddMsgs(newAnns...) + } else if newAnns != nil { + log.Trace("Skipping broadcast of announcements received " + + "during initial graph sync") + } +} + // TODO(roasbeef): d/c peers that send updates not on our chain // InitSyncState is called by outside sub-systems when a connection is @@ -1824,10 +1819,6 @@ func (d *AuthenticatedGossiper) isPremature(chanID lnwire.ShortChannelID, func (d *AuthenticatedGossiper) processNetworkAnnouncement( nMsg *networkMsg) ([]networkMsg, bool) { - log.Debugf("Processing network message: peer=%v, source=%x, msg=%s, "+ - "is_remote=%v", nMsg.peer, nMsg.source.SerializeCompressed(), - nMsg.msg.MsgType(), nMsg.isRemote) - // If this is a remote update, we set the scheduler option to lazily // add it to the graph. var schedulerOp []batch.SchedulerOption @@ -1947,7 +1938,7 @@ func (d *AuthenticatedGossiper) isMsgStale(msg lnwire.Message) bool { } if err != nil { log.Debugf("Unable to retrieve channel=%v from graph: "+ - "%v", err) + "%v", chanInfo.ChannelID, err) return false } @@ -2145,6 +2136,9 @@ func (d *AuthenticatedGossiper) handleNodeAnnouncement(nMsg *networkMsg, timestamp := time.Unix(int64(nodeAnn.Timestamp), 0) + log.Debugf("Processing NodeAnnouncement: peer=%v, timestamp=%v, "+ + "node=%x", nMsg.peer, timestamp, nodeAnn.NodeID) + // We'll quickly ask the router if it already has a newer update for // this node so we can skip validating signatures if not required. if d.cfg.Router.IsStaleNode(nodeAnn.NodeID, timestamp) { @@ -2199,6 +2193,10 @@ func (d *AuthenticatedGossiper) handleNodeAnnouncement(nMsg *networkMsg, nMsg.err <- nil // TODO(roasbeef): get rid of the above + + log.Debugf("Processed NodeAnnouncement: peer=%v, timestamp=%v, "+ + "node=%x", nMsg.peer, timestamp, nodeAnn.NodeID) + return announcements, true } @@ -2207,6 +2205,9 @@ func (d *AuthenticatedGossiper) handleChanAnnouncement(nMsg *networkMsg, ann *lnwire.ChannelAnnouncement, ops []batch.SchedulerOption) ([]networkMsg, bool) { + log.Debugf("Processing ChannelAnnouncement: peer=%v, short_chan_id=%v", + nMsg.peer, ann.ShortChannelID.ToUint64()) + // We'll ignore any channel announcements that target any chain other // than the set of chains we know of. if !bytes.Equal(ann.ChainHash[:], d.cfg.ChainHash[:]) { @@ -2327,6 +2328,9 @@ func (d *AuthenticatedGossiper) handleChanAnnouncement(nMsg *networkMsg, } } + log.Debugf("Adding edge for short_chan_id: %v", + ann.ShortChannelID.ToUint64()) + // We will add the edge to the channel router. If the nodes present in // this channel are not present in the database, a partial node will be // added to represent each node while we wait for a node announcement. @@ -2338,6 +2342,9 @@ func (d *AuthenticatedGossiper) handleChanAnnouncement(nMsg *networkMsg, d.channelMtx.Lock(ann.ShortChannelID.ToUint64()) err := d.cfg.Router.AddEdge(edge, ops...) if err != nil { + log.Debugf("Router rejected edge for short_chan_id(%v): %v", + ann.ShortChannelID.ToUint64(), err) + defer d.channelMtx.Unlock(ann.ShortChannelID.ToUint64()) // If the edge was rejected due to already being known, then it @@ -2359,19 +2366,20 @@ func (d *AuthenticatedGossiper) handleChanAnnouncement(nMsg *networkMsg, return nil, false } + log.Debugf("Extracted %v announcements from rejected "+ + "msgs", len(anns)) + // If while processing this rejected edge, we realized // there's a set of announcements we could extract, // then we'll return those directly. - if len(anns) != 0 { - nMsg.err <- nil - return anns, true - } + // + // NOTE: since this is an ErrIgnored, we can return + // true here to signal "allow" to its dependants. + nMsg.err <- nil - // Otherwise, this is just a regular rejected edge. - log.Debugf("Router rejected channel edge: %v", err) + return anns, true } else { - log.Debugf("Router rejected channel edge: %v", err) - + // Otherwise, this is just a regular rejected edge. key := newRejectCacheKey( ann.ShortChannelID.ToUint64(), sourceToPub(nMsg.source), @@ -2386,6 +2394,9 @@ func (d *AuthenticatedGossiper) handleChanAnnouncement(nMsg *networkMsg, // If err is nil, release the lock immediately. d.channelMtx.Unlock(ann.ShortChannelID.ToUint64()) + log.Debugf("Finish adding edge for short_chan_id: %v", + ann.ShortChannelID.ToUint64()) + // If we earlier received any ChannelUpdates for this channel, we can // now process them, as the channel is added to the graph. shortChanID := ann.ShortChannelID.ToUint64() @@ -2456,6 +2467,10 @@ func (d *AuthenticatedGossiper) handleChanAnnouncement(nMsg *networkMsg, } nMsg.err <- nil + + log.Debugf("Processed ChannelAnnouncement: peer=%v, short_chan_id=%v", + nMsg.peer, ann.ShortChannelID.ToUint64()) + return announcements, true } @@ -2464,6 +2479,9 @@ func (d *AuthenticatedGossiper) handleChanUpdate(nMsg *networkMsg, upd *lnwire.ChannelUpdate, ops []batch.SchedulerOption) ([]networkMsg, bool) { + log.Debugf("Processing ChannelUpdate: peer=%v, short_chan_id=%v, ", + nMsg.peer, upd.ShortChannelID.ToUint64()) + // We'll ignore any channel updates that target any chain other than // the set of chains we know of. if !bytes.Equal(upd.ChainHash[:], d.cfg.ChainHash[:]) { @@ -2523,10 +2541,10 @@ func (d *AuthenticatedGossiper) handleChanUpdate(nMsg *networkMsg, graphScid, timestamp, upd.ChannelFlags, ) { - log.Debugf("Ignored stale edge policy: peer=%v, source=%x, "+ - "msg=%s, is_remote=%v", nMsg.peer, - nMsg.source.SerializeCompressed(), nMsg.msg.MsgType(), - nMsg.isRemote, + log.Debugf("Ignored stale edge policy for short_chan_id(%v): "+ + "peer=%v, source=%x, msg=%s, is_remote=%v", shortChanID, + nMsg.peer, nMsg.source.SerializeCompressed(), + nMsg.msg.MsgType(), nMsg.isRemote, ) nMsg.err <- nil @@ -2649,6 +2667,10 @@ func (d *AuthenticatedGossiper) handleChanUpdate(nMsg *networkMsg, edgeToUpdate = e2 } + log.Debugf("Validating ChannelUpdate: channel=%v, from node=%x, has "+ + "edge=%v", chanInfo.ChannelID, pubKey.SerializeCompressed(), + edgeToUpdate != nil) + // Validate the channel announcement with the expected public key and // channel capacity. In the case of an invalid channel update, we'll // return an error to the caller and exit early. @@ -2743,7 +2765,8 @@ func (d *AuthenticatedGossiper) handleChanUpdate(nMsg *networkMsg, routing.ErrVBarrierShuttingDown, ) { - log.Debug(err) + log.Debugf("Update edge for short_chan_id(%v) got: %v", + shortChanID, err) } else { // Since we know the stored SCID in the graph, we'll // cache that SCID. @@ -2753,7 +2776,8 @@ func (d *AuthenticatedGossiper) handleChanUpdate(nMsg *networkMsg, ) _, _ = d.recentRejects.Put(key, &cachedReject{}) - log.Error(err) + log.Errorf("Update edge for short_chan_id(%v) got: %v", + shortChanID, err) } nMsg.err <- err @@ -2801,8 +2825,7 @@ func (d *AuthenticatedGossiper) handleChanUpdate(nMsg *networkMsg, ) log.Debugf("The message %v has no AuthProof, sending the "+ - "update to remote peer %x", upd.MsgType(), - remotePubKey) + "update to remote peer %x", upd.MsgType(), remotePubKey) // Now we'll attempt to send the channel update message // reliably to the remote peer in the background, so that we @@ -2832,6 +2855,10 @@ func (d *AuthenticatedGossiper) handleChanUpdate(nMsg *networkMsg, } nMsg.err <- nil + + log.Debugf("Processed ChannelUpdate: peer=%v, short_chan_id=%v, "+ + "timestamp=%v", nMsg.peer, upd.ShortChannelID.ToUint64(), + timestamp) return announcements, true } @@ -2848,7 +2875,7 @@ func (d *AuthenticatedGossiper) handleAnnSig(nMsg *networkMsg, prefix = "remote" } - log.Infof("Received new %v channel announcement for %v", prefix, + log.Infof("Received new %v announcement signature for %v", prefix, ann.ShortChannelID) // By the specification, channel announcement proofs should be sent diff --git a/discovery/reliable_sender.go b/discovery/reliable_sender.go index b22a15ae1..b4d32e73f 100644 --- a/discovery/reliable_sender.go +++ b/discovery/reliable_sender.go @@ -84,6 +84,9 @@ func (s *reliableSender) Start() error { // Stop halts the reliable sender from sending messages to peers. func (s *reliableSender) Stop() { s.stop.Do(func() { + log.Debugf("reliableSender is stopping") + defer log.Debugf("reliableSender stopped") + close(s.quit) s.wg.Wait() }) diff --git a/discovery/sync_manager.go b/discovery/sync_manager.go index d834be6b2..8123c37eb 100644 --- a/discovery/sync_manager.go +++ b/discovery/sync_manager.go @@ -183,6 +183,9 @@ func (m *SyncManager) Start() { // Stop stops the SyncManager from performing its duties. func (m *SyncManager) Stop() { m.stop.Do(func() { + log.Debugf("SyncManager is stopping") + defer log.Debugf("SyncManager stopped") + close(m.quit) m.wg.Wait() diff --git a/discovery/syncer.go b/discovery/syncer.go index 11424a61e..f5c1bace5 100644 --- a/discovery/syncer.go +++ b/discovery/syncer.go @@ -450,6 +450,9 @@ func (g *GossipSyncer) Start() { // exited. func (g *GossipSyncer) Stop() { g.stopped.Do(func() { + log.Debugf("Stopping GossipSyncer(%x)", g.cfg.peerPub[:]) + defer log.Debugf("GossipSyncer(%x) stopped", g.cfg.peerPub[:]) + close(g.quit) g.wg.Wait() }) diff --git a/docs/release-notes/release-notes-0.16.0.md b/docs/release-notes/release-notes-0.16.0.md index 642fdf24b..97cf2cf06 100644 --- a/docs/release-notes/release-notes-0.16.0.md +++ b/docs/release-notes/release-notes-0.16.0.md @@ -215,6 +215,9 @@ certain large transactions](https://github.com/lightningnetwork/lnd/pull/7100). `getblockhash`, and `getbestblock`. These commands provide access to chain block data. +* [Fixed a bug](https://github.com/lightningnetwork/lnd/pull/7186) that might + lead to channel updates being missed, causing channel graph being incomplete. + ## Code Health * [test: use `T.TempDir` to create temporary test diff --git a/peer/brontide.go b/peer/brontide.go index 018f81245..7ac65aaf8 100644 --- a/peer/brontide.go +++ b/peer/brontide.go @@ -1401,6 +1401,8 @@ func newChanMsgStream(p *Brontide, cid lnwire.ChannelID) *msgStream { // channel announcements. func newDiscMsgStream(p *Brontide) *msgStream { apply := func(msg lnwire.Message) { + // TODO(yy): `ProcessRemoteAnnouncement` returns an error chan + // and we need to process it. p.cfg.AuthGossiper.ProcessRemoteAnnouncement(msg, p) } diff --git a/routing/router.go b/routing/router.go index 01d4b5022..0588d5122 100644 --- a/routing/router.go +++ b/routing/router.go @@ -991,6 +991,78 @@ func (r *ChannelRouter) pruneZombieChans() error { return nil } +// handleNetworkUpdate is responsible for processing the update message and +// notifies topology changes, if any. +// +// NOTE: must be run inside goroutine. +func (r *ChannelRouter) handleNetworkUpdate(vb *ValidationBarrier, + update *routingMsg) { + + defer r.wg.Done() + defer vb.CompleteJob() + + // If this message has an existing dependency, then we'll wait until + // that has been fully validated before we proceed. + err := vb.WaitForDependants(update.msg) + if err != nil { + switch { + case IsError(err, ErrVBarrierShuttingDown): + update.err <- err + + case IsError(err, ErrParentValidationFailed): + update.err <- newErrf(ErrIgnored, err.Error()) + + default: + log.Warnf("unexpected error during validation "+ + "barrier shutdown: %v", err) + update.err <- err + } + + return + } + + // Process the routing update to determine if this is either a new + // update from our PoV or an update to a prior vertex/edge we + // previously accepted. + err = r.processUpdate(update.msg, update.op...) + update.err <- err + + // If this message had any dependencies, then we can now signal them to + // continue. + allowDependents := err == nil || IsError(err, ErrIgnored, ErrOutdated) + vb.SignalDependants(update.msg, allowDependents) + + // If the error is not nil here, there's no need to send topology + // change. + if err != nil { + // We now decide to log an error or not. If allowDependents is + // false, it means there is an error and the error is neither + // ErrIgnored or ErrOutdated. In this case, we'll log an error. + // Otherwise, we'll add debug log only. + if allowDependents { + log.Debugf("process network updates got: %v", err) + } else { + log.Errorf("process network updates got: %v", err) + } + + return + } + + // Otherwise, we'll send off a new notification for the newly accepted + // update, if any. + topChange := &TopologyChange{} + err = addToTopologyChange(r.cfg.Graph, topChange, update.msg) + if err != nil { + log.Errorf("unable to update topology change notification: %v", + err) + return + } + + if !topChange.isEmpty() { + r.notifyTopologyChange(topChange) + } +} + // networkHandler is the primary goroutine for the ChannelRouter. The roles of // this goroutine include answering queries related to the state of the // network, pruning the graph on new block notification, applying network @@ -1050,74 +1122,7 @@ func (r *ChannelRouter) networkHandler() { validationBarrier.InitJobDependencies(update.msg) r.wg.Add(1) - go func() { - defer r.wg.Done() - defer validationBarrier.CompleteJob() - - // If this message has an existing dependency, - // then we'll wait until that has been fully - // validated before we proceed. - err := validationBarrier.WaitForDependants( - update.msg, - ) - if err != nil { - switch { - case IsError( - err, ErrVBarrierShuttingDown, - ): - update.err <- err - - case IsError( - err, ErrParentValidationFailed, - ): - update.err <- newErrf( - ErrIgnored, err.Error(), - ) - - default: - log.Warnf("unexpected error "+ - "during validation "+ - "barrier shutdown: %v", - err) - update.err <- err - } - return - } - - // Process the routing update to determine if - // this is either a new update from our PoV or - // an update to a prior vertex/edge we - // previously accepted. - err = r.processUpdate(update.msg, update.op...) - update.err <- err - - // If this message had any dependencies, then - // we can now signal them to continue. - allowDependents := err == nil || - IsError(err, ErrIgnored, ErrOutdated) - validationBarrier.SignalDependants( - update.msg, allowDependents, - ) - if err != nil { - return - } - - // Send off a new notification for the newly - // accepted update. - topChange := &TopologyChange{} - err = addToTopologyChange( - r.cfg.Graph, topChange, update.msg, - ) - if err != nil { - log.Errorf("unable to update topology "+ - "change notification: %v", err) - return - } - - if !topChange.isEmpty() { - r.notifyTopologyChange(topChange) - } - }() + go r.handleNetworkUpdate(validationBarrier, update) // TODO(roasbeef): remove all unconnected vertexes // after N blocks pass with no corresponding @@ -1439,6 +1444,9 @@ func (r *ChannelRouter) processUpdate(msg interface{}, r.stats.incNumNodeUpdates() case *channeldb.ChannelEdgeInfo: + log.Debugf("Received ChannelEdgeInfo for channel %v", + msg.ChannelID) + // Prior to processing the announcement we first check if we // already know of this channel, if so, then we can exit early. _, _, exists, isZombie, err := r.cfg.Graph.HasChannelEdge( @@ -1584,7 +1592,7 @@ func (r *ChannelRouter) processUpdate(msg interface{}, return errors.Errorf("unable to add edge: %v", err) } - log.Tracef("New channel discovered! Link "+ + log.Debugf("New channel discovered! Link "+ "connects %x and %x with ChannelPoint(%v): "+ "chan_id=%v, capacity=%v", msg.NodeKey1Bytes, msg.NodeKey2Bytes, @@ -1610,6 +1618,9 @@ func (r *ChannelRouter) processUpdate(msg interface{}, } case *channeldb.ChannelEdgePolicy: + log.Debugf("Received ChannelEdgePolicy for channel %v", + msg.ChannelID) + // We make sure to hold the mutex for this channel ID, // such that no other goroutine is concurrently doing // database accesses for the same channel ID. @@ -1685,7 +1696,7 @@ func (r *ChannelRouter) processUpdate(msg interface{}, return err } - log.Tracef("New channel update applied: %v", + log.Debugf("New channel update applied: %v", newLogClosure(func() string { return spew.Sdump(msg) })) r.stats.incNumChannelUpdates() @@ -2608,10 +2619,18 @@ func (r *ChannelRouter) AddProof(chanID lnwire.ShortChannelID, // target node with a more recent timestamp. // // NOTE: This method is part of the ChannelGraphSource interface. -func (r *ChannelRouter) IsStaleNode(node route.Vertex, timestamp time.Time) bool { +func (r *ChannelRouter) IsStaleNode(node route.Vertex, + timestamp time.Time) bool { + // If our attempt to assert that the node announcement is fresh fails, // then we know that this is actually a stale announcement. - return r.assertNodeAnnFreshness(node, timestamp) != nil + err := r.assertNodeAnnFreshness(node, timestamp) + if err != nil { + log.Debugf("Checking stale node %x got %v", node, err) + return true + } + + return false } // IsPublicNode determines whether the given vertex is seen as a public node in @@ -2641,6 +2660,7 @@ func (r *ChannelRouter) IsStaleEdgePolicy(chanID lnwire.ShortChannelID, edge1Timestamp, edge2Timestamp, exists, isZombie, err := r.cfg.Graph.HasChannelEdge(chanID.ToUint64()) if err != nil { + log.Debugf("Check stale edge policy got error: %v", err) return false } @@ -2788,6 +2808,7 @@ func (r *ChannelRouter) BuildRoute(amt *lnwire.MilliSatoshi, // Exit if there are no channels. unifiedPolicy, ok := u.policies[fromNode] if !ok { + log.Errorf("Cannot find policy for node %v", fromNode) return nil, ErrNoChannel{ fromNode: fromNode, position: i, @@ -2806,6 +2827,8 @@ func (r *ChannelRouter) BuildRoute(amt *lnwire.MilliSatoshi, // to forward. policy := unifiedPolicy.getPolicy(runningAmt, bandwidthHints) if policy == nil { + log.Errorf("Cannot find policy with amt=%v for node %v", + runningAmt, fromNode) return nil, ErrNoChannel{ fromNode: fromNode, position: i, diff --git a/routing/validation_barrier.go b/routing/validation_barrier.go index 80866acd5..ed1ac71f0 100644 --- a/routing/validation_barrier.go +++ b/routing/validation_barrier.go @@ -1,6 +1,7 @@ package routing import ( + "fmt" "sync" "github.com/lightningnetwork/lnd/channeldb" @@ -177,56 +178,77 @@ func (v *ValidationBarrier) WaitForDependants(job interface{}) error { var ( signals *validationSignals ok bool + jobDesc string ) + // Acquire a lock to read ValidationBarrier. v.Lock() - switch msg := job.(type) { + switch msg := job.(type) { // Any ChannelUpdate or NodeAnnouncement jobs will need to wait on the // completion of any active ChannelAnnouncement jobs related to them. case *channeldb.ChannelEdgePolicy: shortID := lnwire.NewShortChanIDFromInt(msg.ChannelID) signals, ok = v.chanEdgeDependencies[shortID] + + jobDesc = fmt.Sprintf("job=lnwire.ChannelEdgePolicy, scid=%v", + msg.ChannelID) + case *channeldb.LightningNode: vertex := route.Vertex(msg.PubKeyBytes) signals, ok = v.nodeAnnDependencies[vertex] + + jobDesc = fmt.Sprintf("job=channeldb.LightningNode, pub=%x", + vertex) + case *lnwire.ChannelUpdate: signals, ok = v.chanEdgeDependencies[msg.ShortChannelID] + + jobDesc = fmt.Sprintf("job=lnwire.ChannelUpdate, scid=%v", + msg.ShortChannelID.ToUint64()) + case *lnwire.NodeAnnouncement: vertex := route.Vertex(msg.NodeID) signals, ok = v.nodeAnnDependencies[vertex] + jobDesc = fmt.Sprintf("job=lnwire.NodeAnnouncement, pub=%x", + vertex) // Other types of jobs can be executed immediately, so we'll just // return directly. case *lnwire.AnnounceSignatures: // TODO(roasbeef): need to wait on chan ann? - v.Unlock() - return nil case *channeldb.ChannelEdgeInfo: - v.Unlock() - return nil case *lnwire.ChannelAnnouncement: - v.Unlock() + } + + // Release the lock once the above read is finished. + v.Unlock() + + // If it's not ok, it means either the job is not a dependent type, or + // it doesn't have a dependency signal. Either way, we can return + // early. + if !ok { return nil } - v.Unlock() + + log.Debugf("Waiting for dependent on %s", jobDesc) // If we do have an active job, then we'll wait until either the signal // is closed, or the set of jobs exits. - if ok { - select { - case <-v.quit: - return newErrf(ErrVBarrierShuttingDown, - "validation barrier shutting down") - case <-signals.deny: - return newErrf(ErrParentValidationFailed, - "parent validation failed") - case <-signals.allow: - return nil - } - } + select { + case <-v.quit: + return newErrf(ErrVBarrierShuttingDown, + "validation barrier shutting down") - return nil + case <-signals.deny: + log.Debugf("Signal deny for %s", jobDesc) + return newErrf(ErrParentValidationFailed, + "parent validation failed") + + case <-signals.allow: + log.Tracef("Signal allow for %s", jobDesc) + return nil + } } // SignalDependants will allow/deny any jobs that are dependent on this job that