discovery+graph: track job set dependencies in vb

discovery/gossiper.go

@@ -512,6 +512,9 @@ type AuthenticatedGossiper struct {
 	// AuthenticatedGossiper lock.
 	chanUpdateRateLimiter map[uint64][2]*rate.Limiter
 
+	// vb is used to enforce job dependency ordering of gossip messages.
+	vb *graph.ValidationBarrier
+
 	sync.Mutex
 }
 
@@ -537,6 +540,8 @@ func New(cfg Config, selfKeyDesc *keychain.KeyDescriptor) *AuthenticatedGossiper
 		banman:                newBanman(),
 	}
 
+	gossiper.vb = graph.NewValidationBarrier(1000, gossiper.quit)
+
 	gossiper.syncMgr = newSyncManager(&SyncManagerCfg{
 		ChainHash:               cfg.ChainHash,
 		ChanSeries:              cfg.ChanSeries,
@@ -1398,10 +1403,6 @@ func (d *AuthenticatedGossiper) networkHandler() {
 		log.Errorf("Unable to rebroadcast stale announcements: %v", err)
 	}
 
-	// We'll use this validation to ensure that we process jobs in their
-	// dependency order during parallel validation.
-	validationBarrier := graph.NewValidationBarrier(1000, d.quit)
-
 	for {
 		select {
 		// A new policy update has arrived. We'll commit it to the
@@ -1470,11 +1471,17 @@ func (d *AuthenticatedGossiper) networkHandler() {
 			// We'll set up any dependent, and wait until a free
 			// slot for this job opens up, this allow us to not
 			// have thousands of goroutines active.
-			validationBarrier.InitJobDependencies(announcement.msg)
+			annJobID, err := d.vb.InitJobDependencies(
+				announcement.msg,
+			)
+			if err != nil {
+				announcement.err <- err
+				continue
+			}
 
 			d.wg.Add(1)
 			go d.handleNetworkMessages(
-				announcement, &announcements, validationBarrier,
+				announcement, &announcements, annJobID,
 			)
 
 		// The trickle timer has ticked, which indicates we should
@@ -1525,18 +1532,18 @@ func (d *AuthenticatedGossiper) networkHandler() {
 //
 // NOTE: must be run as a goroutine.
 func (d *AuthenticatedGossiper) handleNetworkMessages(nMsg *networkMsg,
-	deDuped *deDupedAnnouncements, vb *graph.ValidationBarrier) {
+	deDuped *deDupedAnnouncements, jobID graph.JobID) {
 
 	defer d.wg.Done()
-	defer vb.CompleteJob()
+	defer d.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)
+	err := d.vb.WaitForParents(jobID, nMsg.msg)
 	if err != nil {
 		log.Debugf("Validating network message %s got err: %v",
 			nMsg.msg.MsgType(), err)
@@ -1566,7 +1573,16 @@ func (d *AuthenticatedGossiper) handleNetworkMessages(nMsg *networkMsg,
 
 	// If this message had any dependencies, then we can now signal them to
 	// continue.
-	vb.SignalDependants(nMsg.msg, allow)
+	err = d.vb.SignalDependents(nMsg.msg, jobID)
+	if err != nil {
+		// Something is wrong if SignalDependents returns an error.
+		log.Errorf("SignalDependents returned error for msg=%v with "+
+			"JobID=%v", spew.Sdump(nMsg.msg), jobID)
+
+		nMsg.err <- err
+
+		return
+	}
 
 	// If the announcement was accepted, then add the emitted announcements
 	// to our announce batch to be broadcast once the trickle timer ticks

docs/release-notes/release-notes-0.19.0.md

@@ -182,6 +182,9 @@
   added and they will be removed in a future release. The defaults values for 
   these options have also been increased from max 3 log files to 10 and from 
   max 10 MB to 20 MB. 
+
+* Refactored the `ValidationBarrier` to use
+  [set-based dependency tracking](https://github.com/lightningnetwork/lnd/pull/9241).
 
 * [Deprecate `dust-threshold`
 config option](https://github.com/lightningnetwork/lnd/pull/9182) and introduce

graph/builder.go

@@ -675,45 +675,20 @@ func (b *Builder) handleNetworkUpdate(vb *ValidationBarrier,
 	defer b.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()) //nolint
-
-		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 = b.processUpdate(update.msg, update.op...)
+	err := b.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)
+	notError := err == nil || IsError(err, ErrIgnored, ErrOutdated)
 
 	// 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 as a debug message if this is not an error we need to be
+		// concerned about.
+		if notError {
 			log.Debugf("process network updates got: %v", err)
 		} else {
 			log.Errorf("process network updates got: %v", err)
@@ -789,11 +764,6 @@ func (b *Builder) networkHandler() {
 		// result we'll modify the channel graph accordingly depending
 		// on the exact type of the message.
 		case update := <-b.networkUpdates:
-			// We'll set up any dependants, and wait until a free
-			// slot for this job opens up, this allows us to not
-			// have thousands of goroutines active.
-			validationBarrier.InitJobDependencies(update.msg)
-
 			b.wg.Add(1)
 			go b.handleNetworkUpdate(validationBarrier, update)