12312390362

Committed 13 Dec 2024 08:44AM UTC coverage: 57.458% (+8.5%) from 48.92%

Build # 12312390362

Build Type

Pull #9343

github

Committed by

ellemouton

Commit Message

fn: rework the ContextGuard and add tests

In this commit, the ContextGuard struct is re-worked such that the
context that its new main WithCtx method provides is cancelled in sync
with a parent context being cancelled or with it's quit channel being
cancelled. Tests are added to assert the behaviour. In order for the
close of the quit channel to be consistent with the cancelling of the
derived context, the quit channel _must_ be contained internal to the
ContextGuard so that callers are only able to close the channel via the
exposed Quit method which will then take care to first cancel any
derived context that depend on the quit channel before returning.

Pull Request Pull Request #9343: fn: expand the ContextGuard and add tests

Run Details

101853 of 177264 relevant lines covered (57.46%)

24972.93 hits per line

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

76.44

/routing/missioncontrol_state.go

package routing

import (
        "time"

        "github.com/lightningnetwork/lnd/routing/route"
)

// missionControlState is an object that manages the internal mission control
// state. Note that it isn't thread safe and synchronization needs to be
// enforced externally.
type missionControlState struct {
        // lastPairResult tracks the last payment result (on a pair basis) for
        // each transited node. This is a multi-layer map that allows us to look
        // up the failure history of all connected channels (node pairs) for a
        // particular node.
        lastPairResult map[route.Vertex]NodeResults

        // lastSecondChance tracks the last time a second chance was granted for
        // a directed node pair.
        lastSecondChance map[DirectedNodePair]time.Time

        // minFailureRelaxInterval is the minimum time that must have passed
        // since the previously recorded failure before the failure amount may
        // be raised.
        minFailureRelaxInterval time.Duration
}

// newMissionControlState instantiates a new mission control state object.
func newMissionControlState(
        minFailureRelaxInterval time.Duration) *missionControlState {

        return &missionControlState{
                lastPairResult:          make(map[route.Vertex]NodeResults),
                lastSecondChance:        make(map[DirectedNodePair]time.Time),
                minFailureRelaxInterval: minFailureRelaxInterval,
        }
}

// getLastPairResult returns the current state for connections to the given
// node.
func (m *missionControlState) getLastPairResult(node route.Vertex) (NodeResults,
        bool) {

        result, ok := m.lastPairResult[node]
        return result, ok
}

// ResetHistory resets the history of missionControlState returning it to a
// state as if no payment attempts have been made.
func (m *missionControlState) resetHistory() {
        m.lastPairResult = make(map[route.Vertex]NodeResults)
        m.lastSecondChance = make(map[DirectedNodePair]time.Time)
}

// setLastPairResult stores a result for a node pair.
func (m *missionControlState) setLastPairResult(fromNode, toNode route.Vertex,
        timestamp time.Time, result *pairResult, force bool) {

        nodePairs, ok := m.lastPairResult[fromNode]
        if !ok {
                nodePairs = make(NodeResults)
                m.lastPairResult[fromNode] = nodePairs
        }

        current := nodePairs[toNode]

        // Apply the new result to the existing data for this pair. If there is
        // no existing data, apply it to the default values for TimedPairResult.
        if result.success {
                successAmt := result.amt
                current.SuccessTime = timestamp

                // Only update the success amount if this amount is higher. This
                // prevents the success range from shrinking when there is no
                // reason to do so. For example: small amount probes shouldn't
                // affect a previous success for a much larger amount.
                if force || successAmt > current.SuccessAmt {
                        current.SuccessAmt = successAmt
                }

                // If the success amount goes into the failure range, move the
                // failure range up. Future attempts up to the success amount
                // are likely to succeed. We don't want to clear the failure
                // completely, because we haven't learnt much for amounts above
                // the current success amount.
                if force || (!current.FailTime.IsZero() &&
                        successAmt >= current.FailAmt) {

                        current.FailAmt = successAmt + 1
                }
        } else {
                // For failures we always want to update both the amount and the
                // time. Those need to relate to the same result, because the
                // time is used to gradually diminish the penalty for that
                // specific result. Updating the timestamp but not the amount
                // could cause a failure for a lower amount (a more severe
                // condition) to be revived as if it just happened.
                failAmt := result.amt

                // Drop result if it would increase the failure amount too soon
                // after a previous failure. This can happen if htlc results
                // come in out of order. This check makes it easier for payment
                // processes to converge to a final state.
                failInterval := timestamp.Sub(current.FailTime)
                if !force && failAmt > current.FailAmt &&
                        failInterval < m.minFailureRelaxInterval {

                        log.Debugf("Ignoring higher amount failure within min "+
                                "failure relaxation interval: prev_fail_amt=%v, "+
                                "fail_amt=%v, interval=%v",
                                current.FailAmt, failAmt, failInterval)

                        return
                }

                current.FailTime = timestamp
                current.FailAmt = failAmt

                switch {
                // The failure amount is set to zero when the failure is
                // amount-independent, meaning that the attempt would have
                // failed regardless of the amount. This should also reset the
                // success amount to zero.
                case failAmt == 0:
                        current.SuccessAmt = 0

                // If the failure range goes into the success range, move the
                // success range down.
                case failAmt <= current.SuccessAmt:
                        current.SuccessAmt = failAmt - 1
                }
        }

        log.Debugf("Setting %v->%v range to [%v-%v]",
                fromNode, toNode, current.SuccessAmt, current.FailAmt)

        nodePairs[toNode] = current
}

// setAllFail stores a fail result for all known connections to and from the
// given node.
func (m *missionControlState) setAllFail(node route.Vertex,
        timestamp time.Time) {

        for fromNode, nodePairs := range m.lastPairResult {
                for toNode := range nodePairs {
                        if fromNode == node || toNode == node {
                                nodePairs[toNode] = TimedPairResult{
                                        FailTime: timestamp,
                                }
                        }
                }
        }
}

// requestSecondChance checks whether the node fromNode can have a second chance
// at providing a channel update for its channel with toNode.
func (m *missionControlState) requestSecondChance(timestamp time.Time,
        fromNode, toNode route.Vertex) bool {

        // Look up previous second chance time.
        pair := DirectedNodePair{
                From: fromNode,
                To:   toNode,
        }
        lastSecondChance, ok := m.lastSecondChance[pair]

        // If the channel hasn't already be given a second chance or its last
        // second chance was long ago, we give it another chance.
        if !ok || timestamp.Sub(lastSecondChance) > minSecondChanceInterval {
                m.lastSecondChance[pair] = timestamp

                log.Debugf("Second chance granted for %v->%v", fromNode, toNode)

                return true
        }

        // Otherwise penalize the channel, because we don't allow channel
        // updates that are that frequent. This is to prevent nodes from keeping
        // us busy by continuously sending new channel updates.
        log.Debugf("Second chance denied for %v->%v, remaining interval: %v",
                fromNode, toNode, timestamp.Sub(lastSecondChance))

        return false
}

// GetHistorySnapshot takes a snapshot from the current mission control state
// and actual probability estimates.
func (m *missionControlState) getSnapshot() *MissionControlSnapshot {
        log.Debugf("Requesting history snapshot from mission control: "+
                "pair_result_count=%v", len(m.lastPairResult))

        pairs := make([]MissionControlPairSnapshot, 0, len(m.lastPairResult))

        for fromNode, fromPairs := range m.lastPairResult {
                for toNode, result := range fromPairs {
                        pair := NewDirectedNodePair(fromNode, toNode)

                        pairSnapshot := MissionControlPairSnapshot{
                                Pair:            pair,
                                TimedPairResult: result,
                        }

                        pairs = append(pairs, pairSnapshot)
                }
        }

        snapshot := MissionControlSnapshot{
                Pairs: pairs,
        }

        return &snapshot
}

// importSnapshot takes an existing snapshot and merges it with our current
// state if the result provided are fresher than our current results. It returns
// the number of pairs that were used.
func (m *missionControlState) importSnapshot(snapshot *MissionControlSnapshot,
        force bool) int {

        var imported int

        for _, pair := range snapshot.Pairs {
                fromNode := pair.Pair.From
                toNode := pair.Pair.To

                results, found := m.getLastPairResult(fromNode)
                if !found {
                        results = make(map[route.Vertex]TimedPairResult)
                }

                lastResult := results[toNode]

                failResult := failPairResult(pair.FailAmt)
                imported += m.importResult(
                        lastResult.FailTime, pair.FailTime, failResult,
                        fromNode, toNode, force,
                )

                successResult := successPairResult(pair.SuccessAmt)
                imported += m.importResult(
                        lastResult.SuccessTime, pair.SuccessTime, successResult,
                        fromNode, toNode, force,
                )
        }

        return imported
}

func (m *missionControlState) importResult(currentTs, importedTs time.Time,
        importedResult pairResult, fromNode, toNode route.Vertex,
        force bool) int {

        if !force && currentTs.After(importedTs) {
                log.Debugf("Not setting pair result for %v->%v (%v) "+
                        "success=%v, timestamp %v older than last result %v",
                        fromNode, toNode, importedResult.amt,
                        importedResult.success, importedTs, currentTs)

                return 0
        }

        m.setLastPairResult(
                fromNode, toNode, importedTs, &importedResult, force,
        )

        return 1
}

1	package routing
2
3	import (
4	"time"
5
6	"github.com/lightningnetwork/lnd/routing/route"
7	)
8
9	// missionControlState is an object that manages the internal mission control
10	// state. Note that it isn't thread safe and synchronization needs to be
11	// enforced externally.
12	type missionControlState struct {
13	// lastPairResult tracks the last payment result (on a pair basis) for
14	// each transited node. This is a multi-layer map that allows us to look
15	// up the failure history of all connected channels (node pairs) for a
16	// particular node.
17	lastPairResult map[route.Vertex]NodeResults
18
19	// lastSecondChance tracks the last time a second chance was granted for
20	// a directed node pair.
21	lastSecondChance map[DirectedNodePair]time.Time
22
23	// minFailureRelaxInterval is the minimum time that must have passed
24	// since the previously recorded failure before the failure amount may
25	// be raised.
26	minFailureRelaxInterval time.Duration
27	}
28
29	// newMissionControlState instantiates a new mission control state object.
30	func newMissionControlState(
31	minFailureRelaxInterval time.Duration) *missionControlState {	34✔
32		34✔
33	return &missionControlState{	34✔
34	lastPairResult: make(map[route.Vertex]NodeResults),	34✔
35	lastSecondChance: make(map[DirectedNodePair]time.Time),	34✔
36	minFailureRelaxInterval: minFailureRelaxInterval,	34✔
37	}	34✔
38	}	34✔
39
40	// getLastPairResult returns the current state for connections to the given
41	// node.
42	func (m *missionControlState) getLastPairResult(node route.Vertex) (NodeResults,
43	bool) {	668✔
44		668✔
45	result, ok := m.lastPairResult[node]	668✔
46	return result, ok	668✔
47	}	668✔
48
49	// ResetHistory resets the history of missionControlState returning it to a
50	// state as if no payment attempts have been made.
51	func (m *missionControlState) resetHistory() {	3✔
52	m.lastPairResult = make(map[route.Vertex]NodeResults)	3✔
53	m.lastSecondChance = make(map[DirectedNodePair]time.Time)	3✔
54	}	3✔
55
56	// setLastPairResult stores a result for a node pair.
57	func (m *missionControlState) setLastPairResult(fromNode, toNode route.Vertex,
58	timestamp time.Time, result *pairResult, force bool) {	152✔
59		152✔
60	nodePairs, ok := m.lastPairResult[fromNode]	152✔
61	if !ok {	224✔
62	nodePairs = make(NodeResults)	72✔
63	m.lastPairResult[fromNode] = nodePairs	72✔
64	}	72✔
65
66	current := nodePairs[toNode]	152✔
67		152✔
68	// Apply the new result to the existing data for this pair. If there is	152✔
69	// no existing data, apply it to the default values for TimedPairResult.	152✔
70	if result.success {	235✔
71	successAmt := result.amt	83✔
72	current.SuccessTime = timestamp	83✔
73		83✔
74	// Only update the success amount if this amount is higher. This	83✔
75	// prevents the success range from shrinking when there is no	83✔
76	// reason to do so. For example: small amount probes shouldn't	83✔
77	// affect a previous success for a much larger amount.	83✔
78	if force \|\| successAmt > current.SuccessAmt {	128✔
79	current.SuccessAmt = successAmt	45✔
80	}	45✔
81
82	// If the success amount goes into the failure range, move the
83	// failure range up. Future attempts up to the success amount
84	// are likely to succeed. We don't want to clear the failure
85	// completely, because we haven't learnt much for amounts above
86	// the current success amount.
87	if force \|\| (!current.FailTime.IsZero() &&	83✔
88	successAmt >= current.FailAmt) {	87✔
89		4✔
90	current.FailAmt = successAmt + 1	4✔
91	}	4✔
92	} else {	69✔
93	// For failures we always want to update both the amount and the	69✔
94	// time. Those need to relate to the same result, because the	69✔
95	// time is used to gradually diminish the penalty for that	69✔
96	// specific result. Updating the timestamp but not the amount	69✔
97	// could cause a failure for a lower amount (a more severe	69✔
98	// condition) to be revived as if it just happened.	69✔
99	failAmt := result.amt	69✔
100		69✔
101	// Drop result if it would increase the failure amount too soon	69✔
102	// after a previous failure. This can happen if htlc results	69✔
103	// come in out of order. This check makes it easier for payment	69✔
104	// processes to converge to a final state.	69✔
105	failInterval := timestamp.Sub(current.FailTime)	69✔
106	if !force && failAmt > current.FailAmt &&	69✔
107	failInterval < m.minFailureRelaxInterval {	70✔
108		1✔
109	log.Debugf("Ignoring higher amount failure within min "+	1✔
110	"failure relaxation interval: prev_fail_amt=%v, "+	1✔
111	"fail_amt=%v, interval=%v",	1✔
112	current.FailAmt, failAmt, failInterval)	1✔
113		1✔
114	return	1✔
115	}	1✔
116
117	current.FailTime = timestamp	68✔
118	current.FailAmt = failAmt	68✔
119		68✔
120	switch {	68✔
121	// The failure amount is set to zero when the failure is
122	// amount-independent, meaning that the attempt would have
123	// failed regardless of the amount. This should also reset the
124	// success amount to zero.
125	case failAmt == 0:	34✔
126	current.SuccessAmt = 0	34✔
127
128	// If the failure range goes into the success range, move the
129	// success range down.
130	case failAmt <= current.SuccessAmt:	5✔
131	current.SuccessAmt = failAmt - 1	5✔
132	}
133	}
134
135	log.Debugf("Setting %v->%v range to [%v-%v]",	151✔
136	fromNode, toNode, current.SuccessAmt, current.FailAmt)	151✔
137		151✔
138	nodePairs[toNode] = current	151✔
139	}
140
141	// setAllFail stores a fail result for all known connections to and from the
142	// given node.
143	func (m *missionControlState) setAllFail(node route.Vertex,
144	timestamp time.Time) {	6✔
145		6✔
146	for fromNode, nodePairs := range m.lastPairResult {	12✔
147	for toNode := range nodePairs {	14✔
148	if fromNode == node \|\| toNode == node {	12✔
149	nodePairs[toNode] = TimedPairResult{	4✔
150	FailTime: timestamp,	4✔
151	}	4✔
152	}	4✔
153	}
154	}
155	}
156
157	// requestSecondChance checks whether the node fromNode can have a second chance
158	// at providing a channel update for its channel with toNode.
159	func (m *missionControlState) requestSecondChance(timestamp time.Time,
160	fromNode, toNode route.Vertex) bool {	9✔
161		9✔
162	// Look up previous second chance time.	9✔
163	pair := DirectedNodePair{	9✔
164	From: fromNode,	9✔
165	To: toNode,	9✔
166	}	9✔
167	lastSecondChance, ok := m.lastSecondChance[pair]	9✔
168		9✔
169	// If the channel hasn't already be given a second chance or its last	9✔
170	// second chance was long ago, we give it another chance.	9✔
171	if !ok \|\| timestamp.Sub(lastSecondChance) > minSecondChanceInterval {	15✔
172	m.lastSecondChance[pair] = timestamp	6✔
173		6✔
174	log.Debugf("Second chance granted for %v->%v", fromNode, toNode)	6✔
175		6✔
176	return true	6✔
177	}	6✔
178
179	// Otherwise penalize the channel, because we don't allow channel
180	// updates that are that frequent. This is to prevent nodes from keeping
181	// us busy by continuously sending new channel updates.
182	log.Debugf("Second chance denied for %v->%v, remaining interval: %v",	3✔
183	fromNode, toNode, timestamp.Sub(lastSecondChance))	3✔
184		3✔
185	return false	3✔
186	}
187
188	// GetHistorySnapshot takes a snapshot from the current mission control state
189	// and actual probability estimates.
190	func (m missionControlState) getSnapshot() MissionControlSnapshot {	1✔
191	log.Debugf("Requesting history snapshot from mission control: "+	1✔
192	"pair_result_count=%v", len(m.lastPairResult))	1✔
193		1✔
194	pairs := make([]MissionControlPairSnapshot, 0, len(m.lastPairResult))	1✔
195		1✔
196	for fromNode, fromPairs := range m.lastPairResult {	4✔
197	for toNode, result := range fromPairs {	7✔
198	pair := NewDirectedNodePair(fromNode, toNode)	4✔
199		4✔
200	pairSnapshot := MissionControlPairSnapshot{	4✔
201	Pair: pair,	4✔
202	TimedPairResult: result,	4✔
203	}	4✔
204		4✔
205	pairs = append(pairs, pairSnapshot)	4✔
206	}	4✔
207	}
208
209	snapshot := MissionControlSnapshot{	1✔
210	Pairs: pairs,	1✔
211	}	1✔
212		1✔
213	return &snapshot	1✔
214	}
215
216	// importSnapshot takes an existing snapshot and merges it with our current
217	// state if the result provided are fresher than our current results. It returns
218	// the number of pairs that were used.
219	func (m missionControlState) importSnapshot(snapshot MissionControlSnapshot,
220	force bool) int {	×
221		×
222	var imported int	×
223		×
224	for _, pair := range snapshot.Pairs {	×
225	fromNode := pair.Pair.From	×
226	toNode := pair.Pair.To	×
227		×
228	results, found := m.getLastPairResult(fromNode)	×
229	if !found {	×
230	results = make(map[route.Vertex]TimedPairResult)	×
231	}	×
232
233	lastResult := results[toNode]	×
234		×
235	failResult := failPairResult(pair.FailAmt)	×
236	imported += m.importResult(	×
237	lastResult.FailTime, pair.FailTime, failResult,	×
238	fromNode, toNode, force,	×
239	)	×
240		×
241	successResult := successPairResult(pair.SuccessAmt)	×
242	imported += m.importResult(	×
243	lastResult.SuccessTime, pair.SuccessTime, successResult,	×
244	fromNode, toNode, force,	×
245	)	×
246	}
247
248	return imported	×
249	}
250
251	func (m *missionControlState) importResult(currentTs, importedTs time.Time,
252	importedResult pairResult, fromNode, toNode route.Vertex,
253	force bool) int {	×
254		×
255	if !force && currentTs.After(importedTs) {	×
256	log.Debugf("Not setting pair result for %v->%v (%v) "+	×
257	"success=%v, timestamp %v older than last result %v",	×
258	fromNode, toNode, importedResult.amt,	×
259	importedResult.success, importedTs, currentTs)	×
260		×
261	return 0	×
262	}	×
263
264	m.setLastPairResult(	×
265	fromNode, toNode, importedTs, &importedResult, force,	×
266	)	×
267		×
268	return 1	×
269	}

lightningnetwork / lnd / 12312390362

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