14614595224

Committed 23 Apr 2025 09:21AM UTC coverage: 68.901% (+10.3%) from 58.562%

Build # 14614595224

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #9739 from ellemouton/rpcInterceptorMD

lnrpc+rpcperms: add ctx metadata pairs to RPCMiddlewareRequest

Run Details

17 of 17 new or added lines in 1 file covered. (100.0%)

300 existing lines in 24 files now uncovered.

133638 of 193957 relevant lines covered (68.9%)

22022.08 hits per line

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

84.4

/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 {	35✔
79		35✔
80	v := &ValidationBarrier{	35✔
81	jobInfoMap: make(map[string]*jobInfo),	35✔
82	jobDependencies: make(map[JobID]fn.Set[JobID]),	35✔
83	childJobChans: make(map[JobID]chan struct{}),	35✔
84	quit: quitChan,	35✔
85	}	35✔
86		35✔
87	// We'll first initialize a set of semaphores to limit our concurrency	35✔
88	// when validating incoming requests in parallel.	35✔
89	v.validationSemaphore = make(chan struct{}, numActiveReqs)	35✔
90	for i := 0; i < numActiveReqs; i++ {	30,069✔
91	v.validationSemaphore <- struct{}{}	30,034✔
92	}	30,034✔
93
94	return v	35✔
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) {	355✔
101		355✔
102	// We'll wait for either a new slot to become open, or for the quit	355✔
103	// channel to be closed.	355✔
104	select {	355✔
105	case <-v.validationSemaphore:	355✔
106	case <-v.quit:	×
107	}
108
109	v.Lock()	355✔
110	defer v.Unlock()	355✔
111		355✔
112	// updateOrCreateJobInfo modifies the set of activeParentJobs for this	355✔
113	// annID and updates jobInfoMap.	355✔
114	updateOrCreateJobInfo := func(annID string, annJobID JobID) {	1,083✔
115	info, ok := v.jobInfoMap[annID]	728✔
116	if ok {	737✔
117	// If an entry already exists for annID, then a job	9✔
118	// related to it is being validated. Add to the set of	9✔
119	// parent job ids. This addition will only affect	9✔
120	// _later_, _child_ jobs for the annID.	9✔
121	info.activeParentJobIDs.Add(annJobID)	9✔
122	return	9✔
123	}	9✔
124
125	// No entry exists for annID, meaning that we should create
126	// one.
127	parentJobSet := fn.NewSet(annJobID)	721✔
128		721✔
129	info = &jobInfo{	721✔
130	activeParentJobIDs: parentJobSet,	721✔
131	activeDependentJobs: fn.NewSet[JobID](),	721✔
132	}	721✔
133	v.jobInfoMap[annID] = info	721✔
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) {	468✔
140	// If there is no entry in the jobInfoMap, we don't have to	113✔
141	// wait on any parent jobs to finish.	113✔
142	info, ok := v.jobInfoMap[annID]	113✔
143	if !ok {	216✔
144	return	103✔
145	}	103✔
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{})	12✔
156		12✔
157	// Add to set of activeDependentJobs for this annID.	12✔
158	info.activeDependentJobs.Add(childJobID)	12✔
159		12✔
160	// Store in childJobChans. The parent jobs will fetch this chan	12✔
161	// to notify on. The child job will later fetch this chan to	12✔
162	// listen on when WaitForParents is called.	12✔
163	v.childJobChans[childJobID] = jobChan	12✔
164		12✔
165	// Copy over the parent job IDs at this moment for this annID.	12✔
166	// This job must be processed AFTER those parent IDs.	12✔
167	parentJobs := info.activeParentJobIDs.Copy()	12✔
168		12✔
169	// Populate the jobDependencies mapping.	12✔
170	v.jobDependencies[childJobID] = parentJobs	12✔
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) {	355✔
176	case *lnwire.ChannelAnnouncement1:	244✔
177	id := JobID(v.idCtr.Add(1))	244✔
178		244✔
179	updateOrCreateJobInfo(msg.ShortChannelID.String(), id)	244✔
180	updateOrCreateJobInfo(route.Vertex(msg.NodeID1).String(), id)	244✔
181	updateOrCreateJobInfo(route.Vertex(msg.NodeID2).String(), id)	244✔
182		244✔
183	return id, nil	244✔
184
185	// Populate the dependency mappings for the below child jobs.
186	case *lnwire.ChannelUpdate1:	88✔
187	childJobID := JobID(v.idCtr.Add(1))	88✔
188	populateDependencies(msg.ShortChannelID.String(), childJobID)	88✔
189		88✔
190	return childJobID, nil	88✔
191	case *lnwire.NodeAnnouncement:	27✔
192	childJobID := JobID(v.idCtr.Add(1))	27✔
193	populateDependencies(	27✔
194	route.Vertex(msg.NodeID).String(), childJobID,	27✔
195	)	27✔
196		27✔
197	return childJobID, nil	27✔
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() {	330✔
218	select {	330✔
219	case v.validationSemaphore <- struct{}{}:	330✔
UNCOV 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 {	328✔
231		328✔
232	var (	328✔
233	ok bool	328✔
234	jobDesc string	328✔
235		328✔
236	parentJobIDs fn.Set[JobID]	328✔
237	annID string	328✔
238	jobChan chan struct{}	328✔
239	)	328✔
240		328✔
241	// Acquire a lock to read ValidationBarrier.	328✔
242	v.Lock()	328✔
243		328✔
244	switch msg := job.(type) {	328✔
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:	72✔
248	annID = msg.ShortChannelID.String()	72✔
249		72✔
250	parentJobIDs, ok = v.jobDependencies[childJobID]	72✔
251	if !ok {	135✔
252	// If ok is false, it means that this child job never	63✔
253	// had any parent jobs to wait on.	63✔
254	v.Unlock()	63✔
255	return nil	63✔
256	}	63✔
257
258	jobDesc = fmt.Sprintf("job=lnwire.ChannelUpdate, scid=%v",	11✔
259	msg.ShortChannelID.ToUint64())	11✔
260
261	case *lnwire.NodeAnnouncement:	27✔
262	annID = route.Vertex(msg.NodeID).String()	27✔
263		27✔
264	parentJobIDs, ok = v.jobDependencies[childJobID]	27✔
265	if !ok {	53✔
266	// If ok is false, it means that this child job never	26✔
267	// had any parent jobs to wait on.	26✔
268	v.Unlock()	26✔
269	return nil	26✔
270	}	26✔
271
272	jobDesc = fmt.Sprintf("job=lnwire.NodeAnnouncement, pub=%s",	3✔
273	route.Vertex(msg.NodeID))	3✔
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:	233✔
283	v.Unlock()	233✔
284	return nil	233✔
285	}
286
287	// Release the lock once the above read is finished.
288	v.Unlock()	12✔
289		12✔
290	log.Debugf("Waiting for dependent on %s", jobDesc)	12✔
291		12✔
292	v.Lock()	12✔
293	jobChan, ok = v.childJobChans[childJobID]	12✔
294	if !ok {	12✔
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()	12✔
302		12✔
303	for {	24✔
304	select {	12✔
305	case <-v.quit:	4✔
306	return ErrVBarrierShuttingDown	4✔
307
308	case <-jobChan:	8✔
309	// Every time this is sent on or if it's closed, a	8✔
310	// parent job has finished. The parent jobs have to	8✔
311	// also potentially close the channel because if all	8✔
312	// the parent jobs finish and call SignalDependents	8✔
313	// before the goroutine running WaitForParents has a	8✔
314	// chance to grab the notification chan from	8✔
315	// childJobChans, then the running goroutine will wait	8✔
316	// here for a notification forever. By having the last	8✔
317	// parent job close the notificiation chan, we avoid	8✔
318	// this issue.	8✔
319		8✔
320	// Check and see if we have any parent jobs left. If we	8✔
321	// don't, we can finish up.	8✔
322	v.Lock()	8✔
323	info, found := v.jobInfoMap[annID]	8✔
324	if !found {	16✔
325	v.Unlock()	8✔
326		8✔
327	// No parent job info found, proceed with	8✔
328	// validation.	8✔
329	return nil	8✔
330	}	8✔
331
332	x := parentJobIDs.Intersect(info.activeParentJobIDs)	2✔
333	v.Unlock()	2✔
334	if x.IsEmpty() {	4✔
335	// The parent jobs have all completed. We can	2✔
336	// proceed with validation.	2✔
337	return nil	2✔
338	}	2✔
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 {	325✔
349	v.Lock()	325✔
350	defer v.Unlock()	325✔
351		325✔
352	// removeJob either removes a child job or a parent job. If it is	325✔
353	// removing a child job, then it removes the child's JobID from the set	325✔
354	// of dependent jobs for the announcement ID. If this is removing a	325✔
355	// parent job, then it removes the parentJobID from the set of active	325✔
356	// parent jobs and notifies the child jobs that it has finished	325✔
357	// validating.	325✔
358	removeJob := func(annID string, id JobID, child bool) error {	1,126✔
359	if child {	888✔
360	// If we're removing a child job, check jobInfoMap and	87✔
361	// remove this job from activeDependentJobs.	87✔
362	info, ok := v.jobInfoMap[annID]	87✔
363	if ok {	89✔
364	info.activeDependentJobs.Remove(id)	2✔
365	}	2✔
366
367	// Remove the notification chan from childJobChans.
368	delete(v.childJobChans, id)	87✔
369		87✔
370	// Remove this job's dependency mapping.	87✔
371	delete(v.jobDependencies, id)	87✔
372		87✔
373	return nil	87✔
374	}
375
376	// Otherwise, we are removing a parent job.
377	jobInfo, found := v.jobInfoMap[annID]	716✔
378	if !found {	716✔
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)	716✔
386		716✔
387	lastJob := jobInfo.activeParentJobIDs.IsEmpty()	716✔
388		716✔
389	// Notify all dependent jobs that a parent job has completed.	716✔
390	for child := range jobInfo.activeDependentJobs {	727✔
391	notifyChan, ok := v.childJobChans[child]	11✔
392	if !ok {	11✔
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 {	11✔
401	case notifyChan <- struct{}{}:	8✔
402	default:	5✔
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 {	19✔
415	close(notifyChan)	8✔
416	}	8✔
417	}
418
419	// Remove from jobInfoMap if last job.
420	if lastJob {	1,425✔
421	delete(v.jobInfoMap, annID)	709✔
422	}	709✔
423
424	return nil	716✔
425	}
426
427	switch msg := job.(type) {	325✔
428	case *lnwire.ChannelAnnouncement1:	240✔
429	// Signal to the child jobs that parent validation has	240✔
430	// finished. We have to call removeJob for each annID	240✔
431	// that this ChannelAnnouncement can be associated with.	240✔
432	err := removeJob(msg.ShortChannelID.String(), id, false)	240✔
433	if err != nil {	240✔
434	return err	×
435	}	×
436
437	err = removeJob(route.Vertex(msg.NodeID1).String(), id, false)	240✔
438	if err != nil {	240✔
439	return err	×
440	}	×
441
442	err = removeJob(route.Vertex(msg.NodeID2).String(), id, false)	240✔
443	if err != nil {	240✔
444	return err	×
445	}	×
446
447	return nil	240✔
448
449	case *lnwire.NodeAnnouncement:	26✔
450	// Remove child job info.	26✔
451	return removeJob(route.Vertex(msg.NodeID).String(), id, true)	26✔
452
453	case *lnwire.ChannelUpdate1:	63✔
454	// Remove child job info.	63✔
455	return removeJob(msg.ShortChannelID.String(), id, true)	63✔
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 / 14614595224

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