15838907453

Committed 24 Jun 2025 01:26AM UTC coverage: 57.079% (-11.1%) from 68.172%

Build # 15838907453

Build Type

Pull #9982

github

Committed by

web-flow

Commit Message

Merge e42780be2 into 45c15646c

Pull Request Pull Request #9982: lnwire+lnwallet: add LocalNonces field for splice nonce coordination w/ taproot channels

Run Details

103 of 167 new or added lines in 5 files covered. (61.68%)

30191 existing lines in 463 files now uncovered.

96331 of 168768 relevant lines covered (57.08%)

0.6 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

83.69

/discovery/validation_barrier.go

package discovery

import (
        "fmt"
        "sync"
        "sync/atomic"

        "github.com/go-errors/errors"
        "github.com/lightningnetwork/lnd/fn/v2"
        "github.com/lightningnetwork/lnd/lnwire"
        "github.com/lightningnetwork/lnd/routing/route"
)

var (
        // ErrVBarrierShuttingDown signals that the barrier has been requested
        // to shutdown, and that the caller should not treat the wait condition
        // as fulfilled.
        ErrVBarrierShuttingDown = errors.New("ValidationBarrier shutting down")
)

// JobID identifies an active job in the validation barrier. It is large so
// that we don't need to worry about overflows.
type JobID uint64

// jobInfo stores job dependency info for a set of dependent gossip messages.
type jobInfo struct {
        // activeParentJobIDs is the set of active parent job ids.
        activeParentJobIDs fn.Set[JobID]

        // activeDependentJobs is the set of active dependent job ids.
        activeDependentJobs fn.Set[JobID]
}

// ValidationBarrier is a barrier used to enforce a strict validation order
// while concurrently validating other updates for channel edges. It uses a set
// of maps to track validation dependencies. This is needed in practice because
// gossip messages for a given channel may arive in order, but then due to
// scheduling in different goroutines, may be validated in the wrong order.
// With the ValidationBarrier, the dependent update will wait until the parent
// update completes.
type ValidationBarrier struct {
        // validationSemaphore is a channel of structs which is used as a
        // semaphore. Initially we'll fill this with a buffered channel of the
        // size of the number of active requests. Each new job will consume
        // from this channel, then restore the value upon completion.
        validationSemaphore chan struct{}

        // jobInfoMap stores the set of job ids for each channel.
        // NOTE: This MUST be used with the mutex.
        // NOTE: This currently stores string representations of
        // lnwire.ShortChannelID and route.Vertex. Since these are of different
        // lengths, collision cannot occur in their string representations.
        // N.B.: Check that any new string-converted types don't collide with
        // existing string-converted types.
        jobInfoMap map[string]*jobInfo

        // jobDependencies is a mapping from a child's JobID to the set of
        // parent JobID that it depends on.
        // NOTE: This MUST be used with the mutex.
        jobDependencies map[JobID]fn.Set[JobID]

        // childJobChans stores the notification channel that each child job
        // listens on for parent job completions.
        // NOTE: This MUST be used with the mutex.
        childJobChans map[JobID]chan struct{}

        // idCtr is an atomic integer that is used to assign JobIDs.
        idCtr atomic.Uint64

        quit chan struct{}
        sync.Mutex
}

// NewValidationBarrier creates a new instance of a validation barrier given
// the total number of active requests, and a quit channel which will be used
// to know when to kill pending, but unfilled jobs.
func NewValidationBarrier(numActiveReqs int,
        quitChan chan struct{}) *ValidationBarrier {

        v := &ValidationBarrier{
                jobInfoMap:      make(map[string]*jobInfo),
                jobDependencies: make(map[JobID]fn.Set[JobID]),
                childJobChans:   make(map[JobID]chan struct{}),
                quit:            quitChan,
        }

        // We'll first initialize a set of semaphores to limit our concurrency
        // when validating incoming requests in parallel.
        v.validationSemaphore = make(chan struct{}, numActiveReqs)
        for i := 0; i < numActiveReqs; i++ {
                v.validationSemaphore <- struct{}{}
        }

        return v
}

// InitJobDependencies will wait for a new job slot to become open, and then
// sets up any dependent signals/trigger for the new job.
func (v *ValidationBarrier) InitJobDependencies(job interface{}) (JobID,
        error) {

        // We'll wait for either a new slot to become open, or for the quit
        // channel to be closed.
        select {
        case <-v.validationSemaphore:
        case <-v.quit:
        }

        v.Lock()
        defer v.Unlock()

        // updateOrCreateJobInfo modifies the set of activeParentJobs for this
        // annID and updates jobInfoMap.
        updateOrCreateJobInfo := func(annID string, annJobID JobID) {
                info, ok := v.jobInfoMap[annID]
                if ok {
                        // If an entry already exists for annID, then a job
                        // related to it is being validated. Add to the set of
                        // parent job ids. This addition will only affect
                        // _later_, _child_ jobs for the annID.
                        info.activeParentJobIDs.Add(annJobID)
                        return
                }

                // No entry exists for annID, meaning that we should create
                // one.
                parentJobSet := fn.NewSet(annJobID)

                info = &jobInfo{
                        activeParentJobIDs:  parentJobSet,
                        activeDependentJobs: fn.NewSet[JobID](),
                }
                v.jobInfoMap[annID] = info
        }

        // populateDependencies populates the job dependency mappings (i.e.
        // which should complete after another) for the (annID, childJobID)
        // tuple.
        populateDependencies := func(annID string, childJobID JobID) {
                // If there is no entry in the jobInfoMap, we don't have to
                // wait on any parent jobs to finish.
                info, ok := v.jobInfoMap[annID]
                if !ok {
                        return
                }

                // We want to see a snapshot of active parent jobs for this
                // annID that are already registered in activeParentJobIDs. The
                // child job identified by childJobID can only run after these
                // parent jobs have run. After grabbing the snapshot, we then
                // want to persist a slice of these jobs.

                // Create the notification chan that parent jobs will send (or
                // close) on when they complete.
                jobChan := make(chan struct{})

                // Add to set of activeDependentJobs for this annID.
                info.activeDependentJobs.Add(childJobID)

                // Store in childJobChans. The parent jobs will fetch this chan
                // to notify on. The child job will later fetch this chan to
                // listen on when WaitForParents is called.
                v.childJobChans[childJobID] = jobChan

                // Copy over the parent job IDs at this moment for this annID.
                // This job must be processed AFTER those parent IDs.
                parentJobs := info.activeParentJobIDs.Copy()

                // Populate the jobDependencies mapping.
                v.jobDependencies[childJobID] = parentJobs
        }

        // Once a slot is open, we'll examine the message of the job, to see if
        // there need to be any dependent barriers set up.
        switch msg := job.(type) {
        case *lnwire.ChannelAnnouncement1:
                id := JobID(v.idCtr.Add(1))

                updateOrCreateJobInfo(msg.ShortChannelID.String(), id)
                updateOrCreateJobInfo(route.Vertex(msg.NodeID1).String(), id)
                updateOrCreateJobInfo(route.Vertex(msg.NodeID2).String(), id)

                return id, nil

        // Populate the dependency mappings for the below child jobs.
        case *lnwire.ChannelUpdate1:
                childJobID := JobID(v.idCtr.Add(1))
                populateDependencies(msg.ShortChannelID.String(), childJobID)

                return childJobID, nil
        case *lnwire.NodeAnnouncement:
                childJobID := JobID(v.idCtr.Add(1))
                populateDependencies(
                        route.Vertex(msg.NodeID).String(), childJobID,
                )

                return childJobID, nil
        case *lnwire.AnnounceSignatures1:
                // TODO(roasbeef): need to wait on chan ann?
                // - We can do the above by calling populateDependencies. For
                //   now, while we evaluate potential side effects, don't do
                //   anything with childJobID and just return it.
                childJobID := JobID(v.idCtr.Add(1))
                return childJobID, nil

        default:
                // An invalid message was passed into InitJobDependencies.
                // Return an error.
                return JobID(0), errors.New("invalid message")
        }
}

// CompleteJob returns a free slot to the set of available job slots. This
// should be called once a job has been fully completed. Otherwise, slots may
// not be returned to the internal scheduling, causing a deadlock when a new
// overflow job is attempted.
func (v *ValidationBarrier) CompleteJob() {
        select {
        case v.validationSemaphore <- struct{}{}:
        case <-v.quit:
        }
}

// WaitForParents will block until all parent job dependencies have went
// through the validation pipeline. This allows us a graceful way to run jobs
// in goroutines and still have strict ordering guarantees. If this job doesn't
// have any parent job dependencies, then this function will return
// immediately.
func (v *ValidationBarrier) WaitForParents(childJobID JobID,
        job interface{}) error {

        var (
                ok      bool
                jobDesc string

                parentJobIDs fn.Set[JobID]
                annID        string
                jobChan      chan struct{}
        )

        // Acquire a lock to read ValidationBarrier.
        v.Lock()

        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 *lnwire.ChannelUpdate1:
                annID = msg.ShortChannelID.String()

                parentJobIDs, ok = v.jobDependencies[childJobID]
                if !ok {
                        // If ok is false, it means that this child job never
                        // had any parent jobs to wait on.
                        v.Unlock()
                        return nil
                }

                jobDesc = fmt.Sprintf("job=lnwire.ChannelUpdate, scid=%v",
                        msg.ShortChannelID.ToUint64())

        case *lnwire.NodeAnnouncement:
                annID = route.Vertex(msg.NodeID).String()

                parentJobIDs, ok = v.jobDependencies[childJobID]
                if !ok {
                        // If ok is false, it means that this child job never
                        // had any parent jobs to wait on.
                        v.Unlock()
                        return nil
                }

                jobDesc = fmt.Sprintf("job=lnwire.NodeAnnouncement, pub=%s",
                        route.Vertex(msg.NodeID))

        // Other types of jobs can be executed immediately, so we'll just
        // return directly.
        case *lnwire.AnnounceSignatures1:
                // TODO(roasbeef): need to wait on chan ann?
                v.Unlock()
                return nil

        case *lnwire.ChannelAnnouncement1:
                v.Unlock()
                return nil
        }

        // Release the lock once the above read is finished.
        v.Unlock()

        log.Debugf("Waiting for dependent on %s", jobDesc)

        v.Lock()
        jobChan, ok = v.childJobChans[childJobID]
        if !ok {
                v.Unlock()

                // The entry may not exist because this job does not depend on
                // any parent jobs.
                return nil
        }
        v.Unlock()

        for {
                select {
                case <-v.quit:
                        return ErrVBarrierShuttingDown

                case <-jobChan:
                        // Every time this is sent on or if it's closed, a
                        // parent job has finished. The parent jobs have to
                        // also potentially close the channel because if all
                        // the parent jobs finish and call SignalDependents
                        // before the goroutine running WaitForParents has a
                        // chance to grab the notification chan from
                        // childJobChans, then the running goroutine will wait
                        // here for a notification forever. By having the last
                        // parent job close the notificiation chan, we avoid
                        // this issue.

                        // Check and see if we have any parent jobs left. If we
                        // don't, we can finish up.
                        v.Lock()
                        info, found := v.jobInfoMap[annID]
                        if !found {
                                v.Unlock()

                                // No parent job info found, proceed with
                                // validation.
                                return nil
                        }

                        x := parentJobIDs.Intersect(info.activeParentJobIDs)
                        v.Unlock()
                        if x.IsEmpty() {
                                // The parent jobs have all completed. We can
                                // proceed with validation.
                                return nil
                        }

                        // If we've reached this point, we are still waiting on
                        // a parent job to complete.
                }
        }
}

// SignalDependents signals to any child jobs that this parent job has
// finished.
func (v *ValidationBarrier) SignalDependents(job interface{}, id JobID) error {
        v.Lock()
        defer v.Unlock()

        // removeJob either removes a child job or a parent job. If it is
        // removing a child job, then it removes the child's JobID from the set
        // of dependent jobs for the announcement ID. If this is removing a
        // parent job, then it removes the parentJobID from the set of active
        // parent jobs and notifies the child jobs that it has finished
        // validating.
        removeJob := func(annID string, id JobID, child bool) error {
                if child {
                        // If we're removing a child job, check jobInfoMap and
                        // remove this job from activeDependentJobs.
                        info, ok := v.jobInfoMap[annID]
                        if ok {
                                info.activeDependentJobs.Remove(id)
                        }

                        // Remove the notification chan from childJobChans.
                        delete(v.childJobChans, id)

                        // Remove this job's dependency mapping.
                        delete(v.jobDependencies, id)

                        return nil
                }

                // Otherwise, we are removing a parent job.
                jobInfo, found := v.jobInfoMap[annID]
                if !found {
                        // NOTE: Some sort of consistency guarantee has been
                        // broken.
                        return fmt.Errorf("no job info found for "+
                                "identifier(%v)", id)
                }

                jobInfo.activeParentJobIDs.Remove(id)

                lastJob := jobInfo.activeParentJobIDs.IsEmpty()

                // Notify all dependent jobs that a parent job has completed.
                for child := range jobInfo.activeDependentJobs {
                        notifyChan, ok := v.childJobChans[child]
                        if !ok {
                                // NOTE: Some sort of consistency guarantee has
                                // been broken.
                                return fmt.Errorf("no job info found for "+
                                        "identifier(%v)", id)
                        }

                        // We don't want to block when sending out the signal.
                        select {
                        case notifyChan <- struct{}{}:
                        default:
                        }

                        // If this is the last parent job for this annID, also
                        // close the channel. This is needed because it's
                        // possible that the parent job cleans up the job
                        // mappings before the goroutine handling the child job
                        // has a chance to call WaitForParents and catch the
                        // signal sent above. We are allowed to close because
                        // no other parent job will be able to send along the
                        // channel (or close) as we're removing the entry from
                        // the jobInfoMap below.
                        if lastJob {
                                close(notifyChan)
                        }
                }

                // Remove from jobInfoMap if last job.
                if lastJob {
                        delete(v.jobInfoMap, annID)
                }

                return nil
        }

        switch msg := job.(type) {
        case *lnwire.ChannelAnnouncement1:
                // Signal to the child jobs that parent validation has
                // finished. We have to call removeJob for each annID
                // that this ChannelAnnouncement can be associated with.
                err := removeJob(msg.ShortChannelID.String(), id, false)
                if err != nil {
                        return err
                }

                err = removeJob(route.Vertex(msg.NodeID1).String(), id, false)
                if err != nil {
                        return err
                }

                err = removeJob(route.Vertex(msg.NodeID2).String(), id, false)
                if err != nil {
                        return err
                }

                return nil

        case *lnwire.NodeAnnouncement:
                // Remove child job info.
                return removeJob(route.Vertex(msg.NodeID).String(), id, true)

        case *lnwire.ChannelUpdate1:
                // Remove child job info.
                return removeJob(msg.ShortChannelID.String(), id, true)

        case *lnwire.AnnounceSignatures1:
                // No dependency mappings are stored for AnnounceSignatures1,
                // so do nothing.
                return nil
        }

        return errors.New("invalid message - no job dependencies")
}

1	package discovery
2
3	import (
4	"fmt"
5	"sync"
6	"sync/atomic"
7
8	"github.com/go-errors/errors"
9	"github.com/lightningnetwork/lnd/fn/v2"
10	"github.com/lightningnetwork/lnd/lnwire"
11	"github.com/lightningnetwork/lnd/routing/route"
12	)
13
14	var (
15	// ErrVBarrierShuttingDown signals that the barrier has been requested
16	// to shutdown, and that the caller should not treat the wait condition
17	// as fulfilled.
18	ErrVBarrierShuttingDown = errors.New("ValidationBarrier shutting down")
19	)
20
21	// JobID identifies an active job in the validation barrier. It is large so
22	// that we don't need to worry about overflows.
23	type JobID uint64
24
25	// jobInfo stores job dependency info for a set of dependent gossip messages.
26	type jobInfo struct {
27	// activeParentJobIDs is the set of active parent job ids.
28	activeParentJobIDs fn.Set[JobID]
29
30	// activeDependentJobs is the set of active dependent job ids.
31	activeDependentJobs fn.Set[JobID]
32	}
33
34	// ValidationBarrier is a barrier used to enforce a strict validation order
35	// while concurrently validating other updates for channel edges. It uses a set
36	// of maps to track validation dependencies. This is needed in practice because
37	// gossip messages for a given channel may arive in order, but then due to
38	// scheduling in different goroutines, may be validated in the wrong order.
39	// With the ValidationBarrier, the dependent update will wait until the parent
40	// update completes.
41	type ValidationBarrier struct {
42	// validationSemaphore is a channel of structs which is used as a
43	// semaphore. Initially we'll fill this with a buffered channel of the
44	// size of the number of active requests. Each new job will consume
45	// from this channel, then restore the value upon completion.
46	validationSemaphore chan struct{}
47
48	// jobInfoMap stores the set of job ids for each channel.
49	// NOTE: This MUST be used with the mutex.
50	// NOTE: This currently stores string representations of
51	// lnwire.ShortChannelID and route.Vertex. Since these are of different
52	// lengths, collision cannot occur in their string representations.
53	// N.B.: Check that any new string-converted types don't collide with
54	// existing string-converted types.
55	jobInfoMap map[string]*jobInfo
56
57	// jobDependencies is a mapping from a child's JobID to the set of
58	// parent JobID that it depends on.
59	// NOTE: This MUST be used with the mutex.
60	jobDependencies map[JobID]fn.Set[JobID]
61
62	// childJobChans stores the notification channel that each child job
63	// listens on for parent job completions.
64	// NOTE: This MUST be used with the mutex.
65	childJobChans map[JobID]chan struct{}
66
67	// idCtr is an atomic integer that is used to assign JobIDs.
68	idCtr atomic.Uint64
69
70	quit chan struct{}
71	sync.Mutex
72	}
73
74	// NewValidationBarrier creates a new instance of a validation barrier given
75	// the total number of active requests, and a quit channel which will be used
76	// to know when to kill pending, but unfilled jobs.
77	func NewValidationBarrier(numActiveReqs int,
78	quitChan chan struct{}) *ValidationBarrier {	1✔
79		1✔
80	v := &ValidationBarrier{	1✔
81	jobInfoMap: make(map[string]*jobInfo),	1✔
82	jobDependencies: make(map[JobID]fn.Set[JobID]),	1✔
83	childJobChans: make(map[JobID]chan struct{}),	1✔
84	quit: quitChan,	1✔
85	}	1✔
86		1✔
87	// We'll first initialize a set of semaphores to limit our concurrency	1✔
88	// when validating incoming requests in parallel.	1✔
89	v.validationSemaphore = make(chan struct{}, numActiveReqs)	1✔
90	for i := 0; i < numActiveReqs; i++ {	2✔
91	v.validationSemaphore <- struct{}{}	1✔
92	}	1✔
93
94	return v	1✔
95	}
96
97	// InitJobDependencies will wait for a new job slot to become open, and then
98	// sets up any dependent signals/trigger for the new job.
99	func (v *ValidationBarrier) InitJobDependencies(job interface{}) (JobID,
100	error) {	1✔
101		1✔
102	// We'll wait for either a new slot to become open, or for the quit	1✔
103	// channel to be closed.	1✔
104	select {	1✔
105	case <-v.validationSemaphore:	1✔
106	case <-v.quit:	×
107	}
108
109	v.Lock()	1✔
110	defer v.Unlock()	1✔
111		1✔
112	// updateOrCreateJobInfo modifies the set of activeParentJobs for this	1✔
113	// annID and updates jobInfoMap.	1✔
114	updateOrCreateJobInfo := func(annID string, annJobID JobID) {	2✔
115	info, ok := v.jobInfoMap[annID]	1✔
116	if ok {	2✔
117	// If an entry already exists for annID, then a job	1✔
118	// related to it is being validated. Add to the set of	1✔
119	// parent job ids. This addition will only affect	1✔
120	// _later_, _child_ jobs for the annID.	1✔
121	info.activeParentJobIDs.Add(annJobID)	1✔
122	return	1✔
123	}	1✔
124
125	// No entry exists for annID, meaning that we should create
126	// one.
127	parentJobSet := fn.NewSet(annJobID)	1✔
128		1✔
129	info = &jobInfo{	1✔
130	activeParentJobIDs: parentJobSet,	1✔
131	activeDependentJobs: fn.NewSet[JobID](),	1✔
132	}	1✔
133	v.jobInfoMap[annID] = info	1✔
134	}
135
136	// populateDependencies populates the job dependency mappings (i.e.
137	// which should complete after another) for the (annID, childJobID)
138	// tuple.
139	populateDependencies := func(annID string, childJobID JobID) {	2✔
140	// If there is no entry in the jobInfoMap, we don't have to	1✔
141	// wait on any parent jobs to finish.	1✔
142	info, ok := v.jobInfoMap[annID]	1✔
143	if !ok {	2✔
144	return	1✔
145	}	1✔
146
147	// We want to see a snapshot of active parent jobs for this
148	// annID that are already registered in activeParentJobIDs. The
149	// child job identified by childJobID can only run after these
150	// parent jobs have run. After grabbing the snapshot, we then
151	// want to persist a slice of these jobs.
152
153	// Create the notification chan that parent jobs will send (or
154	// close) on when they complete.
155	jobChan := make(chan struct{})	1✔
156		1✔
157	// Add to set of activeDependentJobs for this annID.	1✔
158	info.activeDependentJobs.Add(childJobID)	1✔
159		1✔
160	// Store in childJobChans. The parent jobs will fetch this chan	1✔
161	// to notify on. The child job will later fetch this chan to	1✔
162	// listen on when WaitForParents is called.	1✔
163	v.childJobChans[childJobID] = jobChan	1✔
164		1✔
165	// Copy over the parent job IDs at this moment for this annID.	1✔
166	// This job must be processed AFTER those parent IDs.	1✔
167	parentJobs := info.activeParentJobIDs.Copy()	1✔
168		1✔
169	// Populate the jobDependencies mapping.	1✔
170	v.jobDependencies[childJobID] = parentJobs	1✔
171	}
172
173	// Once a slot is open, we'll examine the message of the job, to see if
174	// there need to be any dependent barriers set up.
175	switch msg := job.(type) {	1✔
176	case *lnwire.ChannelAnnouncement1:	1✔
177	id := JobID(v.idCtr.Add(1))	1✔
178		1✔
179	updateOrCreateJobInfo(msg.ShortChannelID.String(), id)	1✔
180	updateOrCreateJobInfo(route.Vertex(msg.NodeID1).String(), id)	1✔
181	updateOrCreateJobInfo(route.Vertex(msg.NodeID2).String(), id)	1✔
182		1✔
183	return id, nil	1✔
184
185	// Populate the dependency mappings for the below child jobs.
186	case *lnwire.ChannelUpdate1:	1✔
187	childJobID := JobID(v.idCtr.Add(1))	1✔
188	populateDependencies(msg.ShortChannelID.String(), childJobID)	1✔
189		1✔
190	return childJobID, nil	1✔
191	case *lnwire.NodeAnnouncement:	1✔
192	childJobID := JobID(v.idCtr.Add(1))	1✔
193	populateDependencies(	1✔
194	route.Vertex(msg.NodeID).String(), childJobID,	1✔
195	)	1✔
196		1✔
197	return childJobID, nil	1✔
198	case *lnwire.AnnounceSignatures1:	×
199	// TODO(roasbeef): need to wait on chan ann?	×
200	// - We can do the above by calling populateDependencies. For	×
201	// now, while we evaluate potential side effects, don't do	×
202	// anything with childJobID and just return it.	×
203	childJobID := JobID(v.idCtr.Add(1))	×
204	return childJobID, nil	×
205
206	default:	×
207	// An invalid message was passed into InitJobDependencies.	×
208	// Return an error.	×
209	return JobID(0), errors.New("invalid message")	×
210	}
211	}
212
213	// CompleteJob returns a free slot to the set of available job slots. This
214	// should be called once a job has been fully completed. Otherwise, slots may
215	// not be returned to the internal scheduling, causing a deadlock when a new
216	// overflow job is attempted.
217	func (v *ValidationBarrier) CompleteJob() {	1✔
218	select {	1✔
219	case v.validationSemaphore <- struct{}{}:	1✔
220	case <-v.quit:	×
221	}
222	}
223
224	// WaitForParents will block until all parent job dependencies have went
225	// through the validation pipeline. This allows us a graceful way to run jobs
226	// in goroutines and still have strict ordering guarantees. If this job doesn't
227	// have any parent job dependencies, then this function will return
228	// immediately.
229	func (v *ValidationBarrier) WaitForParents(childJobID JobID,
230	job interface{}) error {	1✔
231		1✔
232	var (	1✔
233	ok bool	1✔
234	jobDesc string	1✔
235		1✔
236	parentJobIDs fn.Set[JobID]	1✔
237	annID string	1✔
238	jobChan chan struct{}	1✔
239	)	1✔
240		1✔
241	// Acquire a lock to read ValidationBarrier.	1✔
242	v.Lock()	1✔
243		1✔
244	switch msg := job.(type) {	1✔
245	// Any ChannelUpdate or NodeAnnouncement jobs will need to wait on the
246	// completion of any active ChannelAnnouncement jobs related to them.
247	case *lnwire.ChannelUpdate1:	1✔
248	annID = msg.ShortChannelID.String()	1✔
249		1✔
250	parentJobIDs, ok = v.jobDependencies[childJobID]	1✔
251	if !ok {	2✔
252	// If ok is false, it means that this child job never	1✔
253	// had any parent jobs to wait on.	1✔
254	v.Unlock()	1✔
255	return nil	1✔
256	}	1✔
257
258	jobDesc = fmt.Sprintf("job=lnwire.ChannelUpdate, scid=%v",	1✔
259	msg.ShortChannelID.ToUint64())	1✔
260
261	case *lnwire.NodeAnnouncement:	1✔
262	annID = route.Vertex(msg.NodeID).String()	1✔
263		1✔
264	parentJobIDs, ok = v.jobDependencies[childJobID]	1✔
265	if !ok {	2✔
266	// If ok is false, it means that this child job never	1✔
267	// had any parent jobs to wait on.	1✔
268	v.Unlock()	1✔
269	return nil	1✔
270	}	1✔
271
272	jobDesc = fmt.Sprintf("job=lnwire.NodeAnnouncement, pub=%s",	1✔
273	route.Vertex(msg.NodeID))	1✔
274
275	// Other types of jobs can be executed immediately, so we'll just
276	// return directly.
277	case *lnwire.AnnounceSignatures1:	×
278	// TODO(roasbeef): need to wait on chan ann?	×
279	v.Unlock()	×
280	return nil	×
281
282	case *lnwire.ChannelAnnouncement1:	1✔
283	v.Unlock()	1✔
284	return nil	1✔
285	}
286
287	// Release the lock once the above read is finished.
288	v.Unlock()	1✔
289		1✔
290	log.Debugf("Waiting for dependent on %s", jobDesc)	1✔
291		1✔
292	v.Lock()	1✔
293	jobChan, ok = v.childJobChans[childJobID]	1✔
294	if !ok {	1✔
295	v.Unlock()	×
296		×
297	// The entry may not exist because this job does not depend on	×
298	// any parent jobs.	×
299	return nil	×
300	}	×
301	v.Unlock()	1✔
302		1✔
303	for {	2✔
304	select {	1✔
UNCOV 305	case <-v.quit:	×
UNCOV 306	return ErrVBarrierShuttingDown	×
307
308	case <-jobChan:	1✔
309	// Every time this is sent on or if it's closed, a	1✔
310	// parent job has finished. The parent jobs have to	1✔
311	// also potentially close the channel because if all	1✔
312	// the parent jobs finish and call SignalDependents	1✔
313	// before the goroutine running WaitForParents has a	1✔
314	// chance to grab the notification chan from	1✔
315	// childJobChans, then the running goroutine will wait	1✔
316	// here for a notification forever. By having the last	1✔
317	// parent job close the notificiation chan, we avoid	1✔
318	// this issue.	1✔
319		1✔
320	// Check and see if we have any parent jobs left. If we	1✔
321	// don't, we can finish up.	1✔
322	v.Lock()	1✔
323	info, found := v.jobInfoMap[annID]	1✔
324	if !found {	2✔
325	v.Unlock()	1✔
326		1✔
327	// No parent job info found, proceed with	1✔
328	// validation.	1✔
329	return nil	1✔
330	}	1✔
331
332	x := parentJobIDs.Intersect(info.activeParentJobIDs)	1✔
333	v.Unlock()	1✔
334	if x.IsEmpty() {	2✔
335	// The parent jobs have all completed. We can	1✔
336	// proceed with validation.	1✔
337	return nil	1✔
338	}	1✔
339
340	// If we've reached this point, we are still waiting on
341	// a parent job to complete.
342	}
343	}
344	}
345
346	// SignalDependents signals to any child jobs that this parent job has
347	// finished.
348	func (v *ValidationBarrier) SignalDependents(job interface{}, id JobID) error {	1✔
349	v.Lock()	1✔
350	defer v.Unlock()	1✔
351		1✔
352	// removeJob either removes a child job or a parent job. If it is	1✔
353	// removing a child job, then it removes the child's JobID from the set	1✔
354	// of dependent jobs for the announcement ID. If this is removing a	1✔
355	// parent job, then it removes the parentJobID from the set of active	1✔
356	// parent jobs and notifies the child jobs that it has finished	1✔
357	// validating.	1✔
358	removeJob := func(annID string, id JobID, child bool) error {	2✔
359	if child {	2✔
360	// If we're removing a child job, check jobInfoMap and	1✔
361	// remove this job from activeDependentJobs.	1✔
362	info, ok := v.jobInfoMap[annID]	1✔
363	if ok {	2✔
364	info.activeDependentJobs.Remove(id)	1✔
365	}	1✔
366
367	// Remove the notification chan from childJobChans.
368	delete(v.childJobChans, id)	1✔
369		1✔
370	// Remove this job's dependency mapping.	1✔
371	delete(v.jobDependencies, id)	1✔
372		1✔
373	return nil	1✔
374	}
375
376	// Otherwise, we are removing a parent job.
377	jobInfo, found := v.jobInfoMap[annID]	1✔
378	if !found {	1✔
379	// NOTE: Some sort of consistency guarantee has been	×
380	// broken.	×
381	return fmt.Errorf("no job info found for "+	×
382	"identifier(%v)", id)	×
383	}	×
384
385	jobInfo.activeParentJobIDs.Remove(id)	1✔
386		1✔
387	lastJob := jobInfo.activeParentJobIDs.IsEmpty()	1✔
388		1✔
389	// Notify all dependent jobs that a parent job has completed.	1✔
390	for child := range jobInfo.activeDependentJobs {	2✔
391	notifyChan, ok := v.childJobChans[child]	1✔
392	if !ok {	1✔
393	// NOTE: Some sort of consistency guarantee has	×
394	// been broken.	×
395	return fmt.Errorf("no job info found for "+	×
396	"identifier(%v)", id)	×
397	}	×
398
399	// We don't want to block when sending out the signal.
400	select {	1✔
401	case notifyChan <- struct{}{}:	1✔
402	default:	1✔
403	}
404
405	// If this is the last parent job for this annID, also
406	// close the channel. This is needed because it's
407	// possible that the parent job cleans up the job
408	// mappings before the goroutine handling the child job
409	// has a chance to call WaitForParents and catch the
410	// signal sent above. We are allowed to close because
411	// no other parent job will be able to send along the
412	// channel (or close) as we're removing the entry from
413	// the jobInfoMap below.
414	if lastJob {	2✔
415	close(notifyChan)	1✔
416	}	1✔
417	}
418
419	// Remove from jobInfoMap if last job.
420	if lastJob {	2✔
421	delete(v.jobInfoMap, annID)	1✔
422	}	1✔
423
424	return nil	1✔
425	}
426
427	switch msg := job.(type) {	1✔
428	case *lnwire.ChannelAnnouncement1:	1✔
429	// Signal to the child jobs that parent validation has	1✔
430	// finished. We have to call removeJob for each annID	1✔
431	// that this ChannelAnnouncement can be associated with.	1✔
432	err := removeJob(msg.ShortChannelID.String(), id, false)	1✔
433	if err != nil {	1✔
434	return err	×
435	}	×
436
437	err = removeJob(route.Vertex(msg.NodeID1).String(), id, false)	1✔
438	if err != nil {	1✔
439	return err	×
440	}	×
441
442	err = removeJob(route.Vertex(msg.NodeID2).String(), id, false)	1✔
443	if err != nil {	1✔
444	return err	×
445	}	×
446
447	return nil	1✔
448
449	case *lnwire.NodeAnnouncement:	1✔
450	// Remove child job info.	1✔
451	return removeJob(route.Vertex(msg.NodeID).String(), id, true)	1✔
452
453	case *lnwire.ChannelUpdate1:	1✔
454	// Remove child job info.	1✔
455	return removeJob(msg.ShortChannelID.String(), id, true)	1✔
456
457	case *lnwire.AnnounceSignatures1:	×
458	// No dependency mappings are stored for AnnounceSignatures1,	×
459	// so do nothing.	×
460	return nil	×
461	}
462
463	return errors.New("invalid message - no job dependencies")	×
464	}

lightningnetwork / lnd / 15838907453

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous