mirror of
https://github.com/lightningnetwork/lnd.git
synced 2026-08-18 13:07:58 +02:00
routing: run a payment batch on a virtual-time scheduler
In this commit, we turn the attempt loop into a state machine and put a deterministic event loop in front of it, so that the sender can have several of its own payments in the air at once. The loop is NOT goroutines, and the reasons are concrete. SimGraph has no locking on balances or on the holds map. The traffic rng is a single stream whose draw order IS the exogenous process. The attribution degrader consumes a fixed number of draws per attempt precisely so that two routers face the same sequence. Real parallelism would destroy all three, and with them the reproducibility every sealed tier depends on. So each live payment carries a next-event timestamp, and the scheduler picks the earliest, breaking ties by scenario index, advances the clock to it, runs the background traffic owed for the interval, and executes exactly one step: one RequestRoute, or one dispatch and one ReportAttempt. A payment splits between those two because the window between choosing a route and hearing what happened to it is the only window two of the sender's payments can share. One payment is a batch of one on the same loop, so the sequential behavior the concurrent path has to reproduce is the behavior it actually reproduces rather than a second implementation of it. The gap between payments is measured from the moment the slot freed, which is what makes max_in_flight one identical to the loop it replaces: 256 paired whole-output runs across the sealed hard, ood and corpus-mix tiers and regenerated default, hard, drift, split and atomic corpora, both arms, zero diffs, plus 132 mainnet runs compared as aggregate sets because of stage B's finding 4. Two things follow from running the traffic per interval rather than per payment, and both are deliberate. A concurrent batch clears in less virtual time and therefore churns the network for less of it. And the whole window fills at once, since every slot is free when the batch starts, so the first max_in_flight payments plan against the same balances. SimBalanceRefresher is the optional half of the contract: a router that wants to be told its own outbound liquidity moved under it can be, and router_accepts_balance_refresh says who asked. No router in this program does, which is exp-016's lesson applied before the fact rather than after it. self_contention_failures is the number the stage exists to produce, and it is causal rather than coincidental. The runner reads the failing edge's true balance and reservation at the moment of the failure and asks whether the edge would have carried the amount with the siblings' share removed. Without atomic mpp nothing reserves anything, so the counter is structurally zero and that tier is the free control.
This commit is contained in:
Olaoluwa Osuntokun
parent
3af6ed3161
commit
08dc442b68
4 changed files with 1509 additions and 259 deletions
|
|
@ -2,6 +2,10 @@ package routing
|
|||
|
||||
import (
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/lightningnetwork/lnd/lnwire"
|
||||
"github.com/lightningnetwork/lnd/routing/route"
|
||||
)
|
||||
|
||||
// simArrivalWindow keeps at most max_in_flight payments live and starts the
|
||||
|
|
@ -90,3 +94,788 @@ func (p *SimConcurrencyParams) maxInFlight() int {
|
|||
|
||||
return p.MaxInFlight
|
||||
}
|
||||
|
||||
// simPaymentState is where a live payment sits inside its own attempt loop.
|
||||
// The sequential loop this replaces did a whole iteration at a time; the
|
||||
// scheduler has to be able to stop between the moment a route is chosen and
|
||||
// the moment its htlc resolves, because that is the only window in which two
|
||||
// of the sender's payments can be in the air together.
|
||||
type simPaymentState uint8
|
||||
|
||||
const (
|
||||
// simPaymentRequest is a payment about to ask its router for the next
|
||||
// route to try. Nothing of this payment's is in the air.
|
||||
simPaymentRequest simPaymentState = iota
|
||||
|
||||
// simPaymentResolve is a payment whose route has been chosen and whose
|
||||
// htlc is on the wire. The step that runs here dispatches it and
|
||||
// reports the outcome.
|
||||
simPaymentResolve
|
||||
|
||||
// simPaymentDone is a payment that has resolved and released whatever
|
||||
// it still held.
|
||||
simPaymentDone
|
||||
)
|
||||
|
||||
// simLivePayment is one of the sender's payments, mid-flight. Everything here
|
||||
// was a local variable of the sequential attempt loop; the scheduler needs it
|
||||
// to survive between steps.
|
||||
type simLivePayment struct {
|
||||
// index is the payment's position in the scenario list, and the
|
||||
// scheduler's tie break. Two payments due at the same virtual instant
|
||||
// run in the order the file lists them, always.
|
||||
index int
|
||||
|
||||
scenario SimScenario
|
||||
result *SimScenarioResult
|
||||
spec *SimPaymentSpec
|
||||
router SimRouter
|
||||
|
||||
// refresher is the optional half of the contract, non-nil only when
|
||||
// this payment's router asked to be told that its own outbound
|
||||
// liquidity moved under it.
|
||||
refresher SimBalanceRefresher
|
||||
|
||||
state simPaymentState
|
||||
nextAt time.Time
|
||||
|
||||
nextAttemptID uint64
|
||||
amtRemaining lnwire.MilliSatoshi
|
||||
inFlightHtlcs uint32
|
||||
|
||||
// holdIDs are the shards that have reached the destination but are not
|
||||
// settled yet, only ever populated under atomic mpp.
|
||||
holdIDs []uint64
|
||||
heldMsat uint64
|
||||
heldFees uint64
|
||||
|
||||
// held is what this payment's own in-flight shards reserve, per
|
||||
// directed edge. It is the runner's copy of the graph's holds map,
|
||||
// split by payment, which the graph itself does not track.
|
||||
held map[simHoldEdge]lnwire.MilliSatoshi
|
||||
|
||||
// pending is the route the request step chose and the resolve step
|
||||
// will dispatch, with the attempt id it was given.
|
||||
pending *route.Route
|
||||
pendingID uint64
|
||||
}
|
||||
|
||||
// SimConcurrencyStats is what a concurrent batch reports about its own
|
||||
// scheduling. None of it enters the objective.
|
||||
//
|
||||
// MaxConcurrent and MeanConcurrent are the MANIPULATION CHECK, and they are
|
||||
// the reason this stage ships counters at all: if a file's payments never
|
||||
// actually overlap then the tier tests nothing, and the scores would look
|
||||
// perfectly reasonable while measuring the sequential batch under a new name.
|
||||
// exp-012 shipped a staleness knob without one and spent a whole experiment
|
||||
// finding out there was no regime there.
|
||||
//
|
||||
// SelfContentionFailures is the number the stage exists to produce. Read it
|
||||
// against MeanConcurrent: a tier that does not overlap cannot contend, and a
|
||||
// zero here means one of those two things.
|
||||
type SimConcurrencyStats struct {
|
||||
// MaxConcurrent is the largest number of the sender's own payments
|
||||
// live at one instant.
|
||||
MaxConcurrent int
|
||||
|
||||
// MeanConcurrent is the time-weighted mean number of payments live,
|
||||
// over the virtual time in which ANY payment was live. The busy-time
|
||||
// denominator is deliberate: over the whole makespan the gaps between
|
||||
// payments would drag the mean below one even in a batch that overlaps
|
||||
// heavily, and the question this answers is whether payments overlap
|
||||
// while they run. It reads exactly 1.0 for a sequential batch.
|
||||
MeanConcurrent float64
|
||||
|
||||
// SelfContentionFailures counts the attempts that failed for lack of
|
||||
// liquidity on a directed edge where ANOTHER of the sender's own
|
||||
// payments was holding some, and would have cleared had that other
|
||||
// payment not been holding it. It is causal rather than coincidental:
|
||||
// the runner reads the edge's true balance and reservation at the
|
||||
// moment of the failure and asks whether the siblings' share of the
|
||||
// reservation is what made the difference.
|
||||
//
|
||||
// It is structurally ZERO without atomic mpp. A shard that settles the
|
||||
// instant it arrives reserves nothing, so a batch of non-atomic
|
||||
// payments can overlap perfectly and never contend through a hold.
|
||||
// That is the stage's free control rather than a defect, and a
|
||||
// concurrency tier is expected to set atomic_mpp.
|
||||
//
|
||||
// It is also computed from the TRUE result rather than the one the
|
||||
// router is told. A degraded attribution section damages what came
|
||||
// back over the wire, and this is the runner's own book keeping.
|
||||
SelfContentionFailures int
|
||||
|
||||
// MakespanSec is the virtual time the batch took to clear, from the
|
||||
// scheduler starting to the last payment resolving. It does NOT enter
|
||||
// the objective and the program's rule says why: it is a new axis
|
||||
// trading against success in an unmeasured way, and this stage changes
|
||||
// one thing.
|
||||
MakespanSec float64
|
||||
|
||||
// RouterAcceptsBalanceRefresh reports whether the routing strategy
|
||||
// under test implements the optional refresh half of the contract, so
|
||||
// that "refresh did not help" is distinguishable from "refresh was
|
||||
// never delivered". exp-016 had to hand-write importer variants of two
|
||||
// champions after the fact because nothing in the contract had ever
|
||||
// asked for the capability, and shipping the flag with the capability
|
||||
// is what keeps this stage from repeating it.
|
||||
RouterAcceptsBalanceRefresh bool
|
||||
}
|
||||
|
||||
// simScheduler is the deterministic virtual-time event loop that replaces the
|
||||
// sequential attempt loop.
|
||||
//
|
||||
// It is NOT goroutines, and the reasons are concrete rather than stylistic.
|
||||
// SimGraph has no locking on balances or on the holds map. The traffic rng is
|
||||
// a single stream whose draw order IS the exogenous process. The attribution
|
||||
// degrader consumes a fixed number of draws per attempt precisely so that two
|
||||
// routers face the same sequence. Real parallelism would destroy all three,
|
||||
// and with them the reproducibility every sealed tier depends on.
|
||||
//
|
||||
// So the loop is: pick the payment whose next event is earliest, breaking ties
|
||||
// by scenario index, advance the clock to it, run the background traffic owed
|
||||
// for the interval, and execute exactly one step of that payment.
|
||||
type simScheduler struct {
|
||||
r *SimRunner
|
||||
source route.Vertex
|
||||
scenarios []SimScenario
|
||||
|
||||
// interArrival is how long a freed slot waits before the next payment
|
||||
// takes it, and gapStep by default.
|
||||
interArrival time.Duration
|
||||
|
||||
// attemptStep is how much virtual time one htlc attempt consumes.
|
||||
attemptStep time.Duration
|
||||
|
||||
// inFlight is the window size, one for the sequential batch.
|
||||
inFlight int
|
||||
|
||||
results []*SimScenarioResult
|
||||
live []*simLivePayment
|
||||
|
||||
// next is the index of the next scenario waiting for a slot.
|
||||
next int
|
||||
|
||||
// slotFree holds the instants at which the free slots became free, in
|
||||
// ascending order. A payment is admitted interArrival after the
|
||||
// EARLIEST of them, which is what makes max_in_flight=1 reproduce the
|
||||
// sequential batch: there the single slot frees when the previous
|
||||
// payment resolved, and the gap is measured from exactly there.
|
||||
slotFree []time.Time
|
||||
|
||||
start time.Time
|
||||
lastAccount time.Time
|
||||
lastFinish time.Time
|
||||
|
||||
liveIntegral time.Duration
|
||||
busy time.Duration
|
||||
|
||||
stats SimConcurrencyStats
|
||||
}
|
||||
|
||||
// newSimScheduler builds the loop for one batch.
|
||||
func newSimScheduler(r *SimRunner, source route.Vertex,
|
||||
scenarios []SimScenario,
|
||||
params *SimConcurrencyParams) (*simScheduler, error) {
|
||||
|
||||
if err := params.validate(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
inFlight := params.maxInFlight()
|
||||
interArrival := r.simGapStep()
|
||||
if params != nil && params.InterArrivalSec > 0 {
|
||||
interArrival = time.Duration(
|
||||
params.InterArrivalSec * float64(time.Second),
|
||||
)
|
||||
}
|
||||
|
||||
// Concurrency without a clock is not concurrency. Every event ties at
|
||||
// the zero instant, so the loop degenerates to running the payments in
|
||||
// index order with nothing ever overlapping, and the scheduling
|
||||
// counters would report a window that never opened. A tier asking for
|
||||
// one gets told rather than measured.
|
||||
if inFlight > 1 && r.simAttemptStep() == 0 {
|
||||
return nil, fmt.Errorf("concurrency: max_in_flight %d needs a "+
|
||||
"clock section with a positive attempt_sec; with no "+
|
||||
"virtual time payments cannot overlap", inFlight)
|
||||
}
|
||||
|
||||
return &simScheduler{
|
||||
r: r,
|
||||
source: source,
|
||||
scenarios: scenarios,
|
||||
interArrival: interArrival,
|
||||
attemptStep: r.simAttemptStep(),
|
||||
inFlight: inFlight,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// run executes the batch and returns the results in scenario order. On a fatal
|
||||
// error it names the scenario that produced it, so that the caller can report
|
||||
// it the way the sequential batch always has.
|
||||
func (s *simScheduler) run() ([]*SimScenarioResult, int, error) {
|
||||
if s.r.graph.Node(s.source) == nil {
|
||||
return nil, 0, fmt.Errorf("source node %v not in graph",
|
||||
s.source)
|
||||
}
|
||||
|
||||
s.start = s.r.simSchedTime()
|
||||
s.lastAccount = s.start
|
||||
s.lastFinish = s.start
|
||||
s.results = make([]*SimScenarioResult, len(s.scenarios))
|
||||
|
||||
s.slotFree = make([]time.Time, s.inFlight)
|
||||
for i := range s.slotFree {
|
||||
s.slotFree[i] = s.start
|
||||
}
|
||||
|
||||
for {
|
||||
admitAt, hasAdmit := s.nextAdmission()
|
||||
payAt, pay := s.nextEvent()
|
||||
|
||||
switch {
|
||||
case !hasAdmit && pay == nil:
|
||||
s.finalize()
|
||||
|
||||
return s.results, 0, nil
|
||||
|
||||
// An admission and a payment event due at the same instant fill
|
||||
// the window first, which is the only reading of "keep
|
||||
// max_in_flight payments live" that does not leave a slot idle
|
||||
// while a payment that could have started waits.
|
||||
case hasAdmit && (pay == nil || !payAt.Before(admitAt)):
|
||||
s.advanceTo(admitAt)
|
||||
|
||||
idx := s.next
|
||||
if err := s.admit(); err != nil {
|
||||
s.abandon()
|
||||
|
||||
return nil, idx, err
|
||||
}
|
||||
|
||||
default:
|
||||
if err := s.stepAt(payAt, pay); err != nil {
|
||||
s.abandon()
|
||||
|
||||
return nil, pay.index, err
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// nextAdmission returns when the next waiting payment starts, if one is
|
||||
// waiting and a slot is free for it.
|
||||
func (s *simScheduler) nextAdmission() (time.Time, bool) {
|
||||
if s.next >= len(s.scenarios) || len(s.slotFree) == 0 {
|
||||
return time.Time{}, false
|
||||
}
|
||||
|
||||
return s.slotFree[0].Add(s.interArrival), true
|
||||
}
|
||||
|
||||
// nextEvent returns the live payment whose next step is due earliest, ties
|
||||
// broken by scenario index.
|
||||
func (s *simScheduler) nextEvent() (time.Time, *simLivePayment) {
|
||||
var best *simLivePayment
|
||||
for _, p := range s.live {
|
||||
switch {
|
||||
case best == nil:
|
||||
case p.nextAt.Before(best.nextAt):
|
||||
case p.nextAt.Equal(best.nextAt) && p.index < best.index:
|
||||
default:
|
||||
continue
|
||||
}
|
||||
|
||||
best = p
|
||||
}
|
||||
|
||||
if best == nil {
|
||||
return time.Time{}, nil
|
||||
}
|
||||
|
||||
return best.nextAt, best
|
||||
}
|
||||
|
||||
// advanceTo moves the clock to an instant and charges the interval to the
|
||||
// concurrency accounting.
|
||||
//
|
||||
// Whether the background traffic runs for the interval is the sequential
|
||||
// loop's own rule, generalized: the gap between payments always churned, and
|
||||
// the time inside a payment churned only under atomic mpp. With one payment
|
||||
// live those two cases are exactly "no payment live" and "the live payment is
|
||||
// atomic", and that is what this asks.
|
||||
func (s *simScheduler) advanceTo(target time.Time) {
|
||||
s.r.simAdvanceTo(target, s.trafficRuns())
|
||||
s.accountTo(s.r.simSchedTime())
|
||||
}
|
||||
|
||||
// trafficRuns reports whether the exogenous process should run over the
|
||||
// interval about to elapse.
|
||||
func (s *simScheduler) trafficRuns() bool {
|
||||
if len(s.live) == 0 {
|
||||
return true
|
||||
}
|
||||
|
||||
for _, p := range s.live {
|
||||
if p.scenario.AtomicMpp {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
// accountTo charges the stretch since the last accounting point to the
|
||||
// concurrency integral, at the live count that held over it. Every caller that
|
||||
// is about to change the live set calls it first, so the integral is exact
|
||||
// rather than sampled.
|
||||
func (s *simScheduler) accountTo(now time.Time) {
|
||||
d := now.Sub(s.lastAccount)
|
||||
if d <= 0 {
|
||||
return
|
||||
}
|
||||
|
||||
s.liveIntegral += time.Duration(len(s.live)) * d
|
||||
if len(s.live) > 0 {
|
||||
s.busy += d
|
||||
}
|
||||
s.lastAccount = now
|
||||
}
|
||||
|
||||
// admit starts the next waiting payment. Everything it does before the first
|
||||
// route request is what the sequential loop did in the same order: the gap's
|
||||
// churn, then the target, then the budget, then the router, then any served
|
||||
// observations still waiting for one.
|
||||
func (s *simScheduler) admit() error {
|
||||
// The payment gap's churn happens whether or not any virtual time
|
||||
// passes. That is what the sequential loop did unconditionally, so a
|
||||
// scenario file with background traffic and no clock still churns once
|
||||
// per payment; the prorating path has no duration to work with there,
|
||||
// so this is the case that keeps it.
|
||||
if s.r.virtualClk == nil || s.r.clockParams.PaymentGapSec <= 0 {
|
||||
if s.r.traffic != nil {
|
||||
s.r.traffic.run()
|
||||
}
|
||||
}
|
||||
|
||||
now := s.r.simSchedTime()
|
||||
s.accountTo(now)
|
||||
|
||||
idx := s.next
|
||||
s.next++
|
||||
s.slotFree = s.slotFree[1:]
|
||||
|
||||
scenario := s.scenarios[idx]
|
||||
result := &SimScenarioResult{Scenario: scenario}
|
||||
s.results[idx] = result
|
||||
|
||||
target, err := s.r.graph.ResolveNode(scenario.Target)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
maxParts := scenario.MaxParts
|
||||
if maxParts == 0 {
|
||||
maxParts = 16
|
||||
}
|
||||
|
||||
amount := lnwire.MilliSatoshi(scenario.AmtMsat)
|
||||
spec := &SimPaymentSpec{
|
||||
Target: target,
|
||||
Amount: amount,
|
||||
MaxParts: maxParts,
|
||||
|
||||
// The budget is quoted as a share of the payment's own amount,
|
||||
// so that one number describes a corpus whose amounts run over
|
||||
// four orders of magnitude. With no limit set this is
|
||||
// lnwire.MaxMilliSatoshi, which is the value the lnd arm has
|
||||
// been constructed with for the whole program.
|
||||
FeeLimitMsat: simFeeBudgetMsat(amount, scenario.FeeLimitPPM),
|
||||
}
|
||||
|
||||
if spec.FeeLimitMsat != lnwire.MaxMilliSatoshi {
|
||||
s.r.feeLimitStats.Payments++
|
||||
}
|
||||
|
||||
// Build the routing strategy under test for this payment, handing it
|
||||
// the public graph view and the sender's exact local balances. The view
|
||||
// wrapper hides the concrete graph so that a candidate router cannot
|
||||
// reach the hidden balances.
|
||||
//
|
||||
// LocalBalances is read HERE, which is what makes concurrency bite
|
||||
// without any new plumbing: it returns each end's available liquidity,
|
||||
// net of what the sender's other in-flight shards already hold, so a
|
||||
// payment starting while a sibling holds sees the reduced balance.
|
||||
router, err := s.r.routerFactory(
|
||||
&simGossipView{g: s.r.graph, now: s.r.clk.Now}, s.source,
|
||||
s.r.graph.LocalBalances(s.source), spec,
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Hand over any served knowledge before the router plans anything, so
|
||||
// that imported beliefs are available to the very first route request
|
||||
// rather than arriving after the payment has committed.
|
||||
if err := s.r.deliverPendingImport(router); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
p := &simLivePayment{
|
||||
index: idx,
|
||||
scenario: scenario,
|
||||
result: result,
|
||||
spec: spec,
|
||||
router: router,
|
||||
state: simPaymentRequest,
|
||||
nextAt: now,
|
||||
amtRemaining: amount,
|
||||
held: make(map[simHoldEdge]lnwire.MilliSatoshi),
|
||||
}
|
||||
|
||||
refresher, accepts := router.(SimBalanceRefresher)
|
||||
if accepts {
|
||||
p.refresher = refresher
|
||||
}
|
||||
s.r.noteBalanceRefreshCapability(accepts)
|
||||
|
||||
s.live = append(s.live, p)
|
||||
if len(s.live) > s.stats.MaxConcurrent {
|
||||
s.stats.MaxConcurrent = len(s.live)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// stepAt advances the clock to a payment's next event and runs exactly one
|
||||
// step of it: one RequestRoute, or one dispatch and one ReportAttempt.
|
||||
func (s *simScheduler) stepAt(at time.Time, p *simLivePayment) error {
|
||||
s.advanceTo(at)
|
||||
|
||||
switch p.state {
|
||||
case simPaymentRequest:
|
||||
return s.stepRequest(p)
|
||||
|
||||
case simPaymentResolve:
|
||||
return s.stepResolve(p)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// stepRequest asks the router for the next route to try, prices it against the
|
||||
// payment's budget, and puts the htlc in the air.
|
||||
func (s *simScheduler) stepRequest(p *simLivePayment) error {
|
||||
// The attempt cap is checked here because here is the top of the
|
||||
// sequential loop: a degenerate router cannot spin forever.
|
||||
if len(p.result.Attempts) >= simMaxAttempts {
|
||||
s.finish(p)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Tell the router that its own outbound liquidity moved under it, if it
|
||||
// asked to be told. Every router in this program keeps the map it was
|
||||
// handed at construction and nothing updates it, which is fine when one
|
||||
// payment runs at a time and wrong the moment two do.
|
||||
if p.refresher != nil {
|
||||
p.refresher.RefreshLocalBalances(
|
||||
s.r.graph.LocalBalances(s.source),
|
||||
)
|
||||
}
|
||||
|
||||
rt, err := p.router.RequestRoute(p.amtRemaining, p.inFlightHtlcs)
|
||||
if err != nil {
|
||||
p.result.Error = err.Error()
|
||||
p.result.GaveUp = true
|
||||
s.finish(p)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
attemptID := p.nextAttemptID
|
||||
p.nextAttemptID++
|
||||
|
||||
// The fee budget is enforced HERE, at the point the runner would
|
||||
// dispatch, and not inside any router. That is the exp-019
|
||||
// construction: a constraint that lives at the shared delivery point is
|
||||
// the same constraint for the lnd stack and for an evolved candidate.
|
||||
//
|
||||
// What the budget has left is what it started with less the fees this
|
||||
// payment has already committed, which is the fees of the shards that
|
||||
// settled plus the fees riding on the ones still held.
|
||||
committed := lnwire.MilliSatoshi(p.result.FeeMsat + p.heldFees)
|
||||
remaining := simRemainingBudget(p.spec.FeeLimitMsat, committed)
|
||||
if rt.TotalFees() > remaining {
|
||||
// The refusal is a fact about this sender, not about the
|
||||
// network: no forwarding node saw the htlc, so nothing is
|
||||
// recorded in the observation stream, no virtual time passes,
|
||||
// and the result is handed to the router undegraded.
|
||||
//
|
||||
// It does cost an attempt. A router that keeps offering routes
|
||||
// it cannot afford spends its attempt budget on them, which is
|
||||
// the whole point of putting the pressure in the environment.
|
||||
refusal := SimHtlcResult{
|
||||
FailureSource: rt.SourcePubKey,
|
||||
Failure: SimFeeLimitFailure{},
|
||||
}
|
||||
|
||||
p.result.Attempts = append(
|
||||
p.result.Attempts, traceAttempt(rt, refusal),
|
||||
)
|
||||
s.r.feeLimitStats.Failures++
|
||||
|
||||
return p.router.ReportAttempt(attemptID, rt, refusal)
|
||||
}
|
||||
|
||||
// The htlc is now in the air and resolves one attempt's worth of
|
||||
// virtual time from now, which is the window another of the sender's
|
||||
// payments can run inside.
|
||||
p.pending = rt
|
||||
p.pendingID = attemptID
|
||||
p.state = simPaymentResolve
|
||||
p.nextAt = s.r.simSchedTime().Add(s.attemptStep)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// stepResolve sends the pending htlc through the simulated network and reports
|
||||
// what came back.
|
||||
func (s *simScheduler) stepResolve(p *simLivePayment) error {
|
||||
rt := p.pending
|
||||
p.pending = nil
|
||||
p.state = simPaymentRequest
|
||||
|
||||
// A malformed route (unknown channel, disconnected hops) is a router
|
||||
// bug: it terminates this payment with an error rather than killing the
|
||||
// whole batch, so one bad edge case doesn't zero out an otherwise
|
||||
// functional candidate.
|
||||
//
|
||||
// An atomic shard is held at the destination rather than settled there,
|
||||
// reserving the liquidity of every hop it crossed until the payment as
|
||||
// a whole resolves.
|
||||
var (
|
||||
htlcResult SimHtlcResult
|
||||
holdID uint64
|
||||
err error
|
||||
)
|
||||
if p.scenario.AtomicMpp {
|
||||
htlcResult, holdID, err = s.r.graph.HoldHtlc(rt)
|
||||
} else {
|
||||
htlcResult, err = s.r.graph.SendHtlc(rt)
|
||||
}
|
||||
if err != nil {
|
||||
p.result.Error = fmt.Sprintf("malformed route: %v", err)
|
||||
s.finish(p)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
p.result.Attempts = append(
|
||||
p.result.Attempts, traceAttempt(rt, htlcResult),
|
||||
)
|
||||
|
||||
s.noteSelfContention(p, rt, htlcResult)
|
||||
|
||||
// Record what this attempt revealed about the edges it crossed, which
|
||||
// is the raw material a weight-serving node would have to offer.
|
||||
s.r.observations = append(s.r.observations, observationsFromAttempt(
|
||||
rt, htlcResult, s.r.clk.Now(),
|
||||
)...)
|
||||
|
||||
// Let the router learn from the outcome. Everything above this line
|
||||
// records what actually happened; what the router is TOLD may be less
|
||||
// than that, since an attribution section ages and damages the result
|
||||
// on its way over.
|
||||
err = p.router.ReportAttempt(
|
||||
p.pendingID, rt, s.r.deliverAttempt(rt, htlcResult),
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// deliverAttempt can age the network by whole attempt-sized slices, so
|
||||
// the next request is due at whatever the clock says now.
|
||||
p.nextAt = s.r.simSchedTime()
|
||||
|
||||
if htlcResult.Failure != nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
p.inFlightHtlcs++
|
||||
|
||||
// A settling shard pays its fee right away; a held one only pays when
|
||||
// the whole set settles.
|
||||
if p.scenario.AtomicMpp {
|
||||
p.holdIDs = append(p.holdIDs, holdID)
|
||||
p.heldMsat += uint64(rt.ReceiverAmt())
|
||||
p.heldFees += uint64(rt.TotalFees())
|
||||
|
||||
for _, res := range s.r.graph.holdReservations(holdID) {
|
||||
p.held[res.edge] += res.amt
|
||||
}
|
||||
} else {
|
||||
p.result.FeeMsat += uint64(rt.TotalFees())
|
||||
}
|
||||
|
||||
// Guard against a buggy router delivering more than asked: unsigned
|
||||
// underflow here would loop until the attempt cap.
|
||||
recv := rt.ReceiverAmt()
|
||||
if recv > p.amtRemaining {
|
||||
p.result.Error = "router over-delivered payment amount"
|
||||
s.finish(p)
|
||||
|
||||
return nil
|
||||
}
|
||||
p.amtRemaining -= recv
|
||||
|
||||
if p.amtRemaining == 0 {
|
||||
// The full amount has arrived, so the held set becomes real
|
||||
// balance movement all at once and the fees it carried finally
|
||||
// come due. Without atomic mpp there is nothing held and this is
|
||||
// a no-op.
|
||||
for _, id := range p.holdIDs {
|
||||
s.r.graph.SettleHold(id)
|
||||
}
|
||||
p.holdIDs = nil
|
||||
p.result.FeeMsat += p.heldFees
|
||||
|
||||
p.result.Success = true
|
||||
s.finish(p)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// noteSelfContention attributes an attempt that failed for want of liquidity
|
||||
// to the sender's OWN other payments, when they are what made the difference.
|
||||
//
|
||||
// The test is causal rather than coincidental. The runner reads the failing
|
||||
// edge's true balance and its total reservation at this instant, and asks
|
||||
// whether the edge would have carried the amount with the siblings' share of
|
||||
// that reservation removed. An edge that was short anyway is not contention,
|
||||
// and an edge that had room to spare did not fail for liquidity at all.
|
||||
func (s *simScheduler) noteSelfContention(p *simLivePayment, rt *route.Route,
|
||||
res SimHtlcResult) {
|
||||
|
||||
if res.Failure == nil {
|
||||
return
|
||||
}
|
||||
|
||||
// A liquidity shortfall is the temporary channel failure the forwarding
|
||||
// check returns. The announced-limit refusals of stage A share the code
|
||||
// and are filtered out below by the balance test: an edge with room to
|
||||
// spare cannot have failed on liquidity.
|
||||
if _, ok := res.Failure.(*lnwire.FailTemporaryChannelFailure); !ok {
|
||||
return
|
||||
}
|
||||
|
||||
idx := getNodeIndexSim(rt, res.FailureSource)
|
||||
if idx == nil || *idx >= len(rt.Hops) {
|
||||
return
|
||||
}
|
||||
|
||||
edge := simHoldEdge{
|
||||
ChanID: rt.Hops[*idx].ChannelID,
|
||||
From: res.FailureSource,
|
||||
}
|
||||
|
||||
// The amount the failing hop was asked to send is the route total at
|
||||
// the first channel and the previous hop's amt-to-forward after that,
|
||||
// matching walkHtlc's own accounting.
|
||||
amtOut := rt.TotalAmount
|
||||
if *idx > 0 {
|
||||
amtOut = rt.Hops[*idx-1].AmtToForward
|
||||
}
|
||||
|
||||
var siblings lnwire.MilliSatoshi
|
||||
for _, q := range s.live {
|
||||
if q == p {
|
||||
continue
|
||||
}
|
||||
|
||||
siblings += q.held[edge]
|
||||
}
|
||||
if siblings == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
balance, held, ok := s.r.graph.endLiquidity(edge.ChanID, edge.From)
|
||||
if !ok || held > balance {
|
||||
return
|
||||
}
|
||||
|
||||
available := balance - held
|
||||
if available >= amtOut {
|
||||
return
|
||||
}
|
||||
if available+siblings < amtOut {
|
||||
return
|
||||
}
|
||||
|
||||
s.stats.SelfContentionFailures++
|
||||
}
|
||||
|
||||
// finish resolves a payment: it releases whatever the payment still held,
|
||||
// takes it out of the live set and frees its slot.
|
||||
func (s *simScheduler) finish(p *simLivePayment) {
|
||||
now := s.r.simSchedTime()
|
||||
s.accountTo(now)
|
||||
|
||||
// Under atomic mpp a payment that never completes settles nothing:
|
||||
// every shard still held gives its reserved liquidity back, so a failed
|
||||
// mpp leaves the hidden balances exactly as it found them and charges
|
||||
// no fees. The success path settles the set and clears holdIDs first,
|
||||
// so this only ever fires on a failure path, whichever one it is.
|
||||
if len(p.holdIDs) > 0 {
|
||||
for _, id := range p.holdIDs {
|
||||
s.r.graph.ReleaseHold(id)
|
||||
}
|
||||
p.result.HeldReleasedMsat = p.heldMsat
|
||||
}
|
||||
p.holdIDs = nil
|
||||
p.held = nil
|
||||
p.state = simPaymentDone
|
||||
|
||||
for i, q := range s.live {
|
||||
if q != p {
|
||||
continue
|
||||
}
|
||||
|
||||
s.live = append(s.live[:i], s.live[i+1:]...)
|
||||
|
||||
break
|
||||
}
|
||||
|
||||
// Virtual time never runs backwards, so the freed slot belongs at the
|
||||
// end of the ascending list.
|
||||
s.slotFree = append(s.slotFree, now)
|
||||
s.lastFinish = now
|
||||
}
|
||||
|
||||
// abandon releases everything the still-live payments hold, which is what the
|
||||
// sequential loop's deferred release did on the error paths that killed a
|
||||
// batch. Without it a fatal error would leave the graph reserving liquidity
|
||||
// for htlcs that will never resolve.
|
||||
func (s *simScheduler) abandon() {
|
||||
for _, p := range s.live {
|
||||
for _, id := range p.holdIDs {
|
||||
s.r.graph.ReleaseHold(id)
|
||||
}
|
||||
p.holdIDs = nil
|
||||
}
|
||||
s.live = nil
|
||||
}
|
||||
|
||||
// finalize closes the accounting and computes the reported ratios.
|
||||
func (s *simScheduler) finalize() {
|
||||
s.accountTo(s.r.simSchedTime())
|
||||
|
||||
if s.busy > 0 {
|
||||
s.stats.MeanConcurrent = s.liveIntegral.Seconds() /
|
||||
s.busy.Seconds()
|
||||
}
|
||||
s.stats.MakespanSec = s.lastFinish.Sub(s.start).Seconds()
|
||||
s.stats.RouterAcceptsBalanceRefresh = s.r.RouterAcceptsBalanceRefresh()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2,7 +2,10 @@ package routing
|
|||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/lightningnetwork/lnd/lnwire"
|
||||
"github.com/lightningnetwork/lnd/routing/route"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
|
|
@ -97,3 +100,560 @@ func TestSimConcurrencyMaxInFlightDefault(t *testing.T) {
|
|||
present := &SimConcurrencyParams{MaxInFlight: 4}
|
||||
require.Equal(t, 4, present.maxInFlight())
|
||||
}
|
||||
|
||||
// concurrencyRunner builds a runner over the given graph whose router factory
|
||||
// is called once per payment, handing each payment its own router. That is the
|
||||
// contract SimRouterFactory has always had and the one this stage keeps: a
|
||||
// shared instance would need RequestRoute and ReportAttempt to carry a payment
|
||||
// identifier, which would break every router ever evolved.
|
||||
func concurrencyRunner(t *testing.T, g *SimGraph, source route.Vertex,
|
||||
factory SimRouterFactory) *SimRunner {
|
||||
|
||||
t.Helper()
|
||||
|
||||
runner, err := NewSimRunner(g, DefaultSimParams(), source, t.TempDir())
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(runner.Close)
|
||||
|
||||
runner.SetRouterFactory(factory)
|
||||
|
||||
return runner
|
||||
}
|
||||
|
||||
// scriptedFactory hands each payment a fresh scripted router built from the
|
||||
// per-payment route lists, in scenario order.
|
||||
func scriptedFactory(t *testing.T,
|
||||
scripts [][]*route.Route) (SimRouterFactory, *[]*scriptedRouter) {
|
||||
|
||||
t.Helper()
|
||||
|
||||
built := make([]*scriptedRouter, 0, len(scripts))
|
||||
|
||||
return func(_ SimNetworkView, _ route.Vertex,
|
||||
_ map[uint64]lnwire.MilliSatoshi,
|
||||
_ *SimPaymentSpec) (SimRouter, error) {
|
||||
|
||||
require.Less(t, len(built), len(scripts),
|
||||
"factory called more times than there are scripts")
|
||||
|
||||
router := &scriptedRouter{routes: scripts[len(built)]}
|
||||
built = append(built, router)
|
||||
|
||||
return router, nil
|
||||
}, &built
|
||||
}
|
||||
|
||||
// TestSimSchedulerSequentialTimeline pins the timeline the sequential batch
|
||||
// has always produced, now that it comes out of the event loop: a payment
|
||||
// starts one payment gap after the previous one resolved, each of its attempts
|
||||
// takes one attempt step, and nothing ever overlaps.
|
||||
func TestSimSchedulerSequentialTimeline(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const shard = lnwire.MilliSatoshi(10_000_000)
|
||||
|
||||
graph, nodes := atomicTestGraph(t)
|
||||
source, target := nodes[0], nodes[3]
|
||||
|
||||
rt := atomicTestRoute(t, graph, source, []uint64{1, 2}, shard)
|
||||
|
||||
// Two payments, the first taking two attempts and the second one. The
|
||||
// first route is deliberately repeated so the payment needs a second
|
||||
// attempt to deliver its full amount.
|
||||
factory, _ := scriptedFactory(t, [][]*route.Route{{rt, rt}, {rt}})
|
||||
|
||||
var starts []time.Time
|
||||
runner := concurrencyRunner(t, graph, source,
|
||||
func(view SimNetworkView, src route.Vertex,
|
||||
balances map[uint64]lnwire.MilliSatoshi,
|
||||
spec *SimPaymentSpec) (SimRouter, error) {
|
||||
|
||||
starts = append(starts, view.Now())
|
||||
|
||||
return factory(view, src, balances, spec)
|
||||
},
|
||||
)
|
||||
|
||||
const start = int64(1_800_000_000)
|
||||
runner.SetVirtualClock(&SimClockParams{
|
||||
StartUnix: start,
|
||||
PaymentGapSec: 600,
|
||||
AttemptSec: 30,
|
||||
})
|
||||
|
||||
results, err := runner.RunBatch([]SimScenario{
|
||||
{Target: target.String(), AmtMsat: uint64(2 * shard),
|
||||
MaxParts: 2},
|
||||
{Target: target.String(), AmtMsat: uint64(shard),
|
||||
MaxParts: 1},
|
||||
}, nil)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, results, 2)
|
||||
require.True(t, results[0].Success)
|
||||
require.True(t, results[1].Success)
|
||||
|
||||
epoch := time.Unix(start, 0)
|
||||
require.Len(t, starts, 2)
|
||||
|
||||
// The first payment starts one gap after the epoch.
|
||||
require.Equal(t, 600*time.Second, starts[0].Sub(epoch))
|
||||
|
||||
// Its two attempts take one step each, and the second payment starts a
|
||||
// gap after the first resolved.
|
||||
require.Equal(
|
||||
t, (2*30+600)*time.Second, starts[1].Sub(starts[0]),
|
||||
)
|
||||
|
||||
// A sequential batch never overlaps, and the mean is taken over the
|
||||
// virtual time in which anything was live, so it reads exactly one.
|
||||
stats := runner.ConcurrencyStats()
|
||||
require.Equal(t, 1, stats.MaxConcurrent)
|
||||
require.InDelta(t, 1.0, stats.MeanConcurrent, 1e-9)
|
||||
require.Zero(t, stats.SelfContentionFailures)
|
||||
|
||||
// The makespan runs from the scheduler starting to the last payment
|
||||
// resolving: two gaps plus three attempts.
|
||||
require.InDelta(t, 2*600+3*30, stats.MakespanSec, 1e-9)
|
||||
}
|
||||
|
||||
// TestSimSchedulerTrafficIsPerInterval asserts that the background traffic a
|
||||
// batch runs is a function of the virtual time that elapsed, not of the number
|
||||
// of payments or attempts that elapsed it.
|
||||
//
|
||||
// The carry is what makes that true: the per-gap volume is pro-rated by
|
||||
// duration and the fractional remainder is kept rather than rounded away, so
|
||||
// however finely the scheduler slices a window, the window's total is the
|
||||
// same. Slicing is exactly what a concurrent batch does differently, which is
|
||||
// why this is the invariant the stage has to hold.
|
||||
func TestSimSchedulerTrafficCarryIsSliceInvariant(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
// owed replays the prorating path over a window divided into the given
|
||||
// number of equal slices, and reports the total it dispatched.
|
||||
owed := func(slices int) int {
|
||||
runner := &SimRunner{
|
||||
traffic: &simTraffic{
|
||||
params: SimTrafficParams{PaymentsPerGap: 8},
|
||||
},
|
||||
clockParams: SimClockParams{PaymentGapSec: 600},
|
||||
}
|
||||
|
||||
var total int
|
||||
for i := 0; i < slices; i++ {
|
||||
total += runner.trafficPaymentsFor(600 / float64(slices))
|
||||
}
|
||||
|
||||
return total
|
||||
}
|
||||
|
||||
// One gap's worth, however the gap is cut up. The guarantee is within
|
||||
// one payment rather than exact: a remainder that should land on an
|
||||
// integer boundary can land a hair under it in floating point, which
|
||||
// pushes one payment into the next slice and no further.
|
||||
for _, slices := range []int{1, 2, 3, 7, 20, 600} {
|
||||
require.InDelta(t, 8, owed(slices), 1,
|
||||
"a window cut into %d slices moved a different "+
|
||||
"amount of liquidity", slices)
|
||||
}
|
||||
|
||||
// A whole gap in one piece is exact, which is what makes the sequential
|
||||
// batch's churn per payment exactly payments_per_gap and what lets the
|
||||
// scheduler's admission reproduce it byte for byte.
|
||||
require.Equal(t, 8, owed(1))
|
||||
}
|
||||
|
||||
// TestSimSchedulerTrafficTracksVirtualTime asserts that a concurrent batch
|
||||
// clears in less virtual time than the sequential one and therefore elapses
|
||||
// less of the exogenous process.
|
||||
//
|
||||
// That is the honest consequence of running the traffic per interval rather
|
||||
// than per payment: concurrency compresses the clock, and the network churns
|
||||
// for as long as the clock says and no longer. A concurrency tier is therefore
|
||||
// a slightly quieter world than its sequential control, and a sweep that
|
||||
// compares the two is reading both effects at once.
|
||||
func TestSimSchedulerTrafficTracksVirtualTime(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
run := func(inFlight int) (int, float64) {
|
||||
graph := trafficTestGraph(t, 3)
|
||||
nodes := sortedNodes(graph)
|
||||
|
||||
runner, err := NewSimRunner(
|
||||
graph, DefaultSimParams(), nodes[0], t.TempDir(),
|
||||
)
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(runner.Close)
|
||||
|
||||
runner.SetVirtualClock(&SimClockParams{
|
||||
StartUnix: 1_800_000_000,
|
||||
PaymentGapSec: 600,
|
||||
AttemptSec: 30,
|
||||
})
|
||||
require.NoError(t, runner.SetBackgroundTraffic(
|
||||
&SimTrafficParams{
|
||||
PaymentsPerGap: 20,
|
||||
MinAmtMsat: 1_000,
|
||||
MaxAmtMsat: 1_000_000,
|
||||
Seed: 11,
|
||||
},
|
||||
))
|
||||
|
||||
scenarios := make([]SimScenario, 0, 6)
|
||||
for i := 1; i <= 6; i++ {
|
||||
scenarios = append(scenarios, SimScenario{
|
||||
Target: nodes[i*3].String(),
|
||||
AmtMsat: 50_000_000,
|
||||
MaxParts: 4,
|
||||
AtomicMpp: true,
|
||||
})
|
||||
}
|
||||
|
||||
var params *SimConcurrencyParams
|
||||
if inFlight > 1 {
|
||||
params = &SimConcurrencyParams{MaxInFlight: inFlight}
|
||||
}
|
||||
|
||||
_, err = runner.RunBatch(scenarios, params)
|
||||
require.NoError(t, err)
|
||||
|
||||
sent, _ := runner.TrafficStats()
|
||||
|
||||
return sent, runner.ConcurrencyStats().MakespanSec
|
||||
}
|
||||
|
||||
sequential, seqSpan := run(1)
|
||||
concurrent, conSpan := run(3)
|
||||
|
||||
require.Less(t, conSpan, seqSpan, "the window did not clear faster")
|
||||
require.Less(t, concurrent, sequential,
|
||||
"a shorter batch churned the network for just as long")
|
||||
}
|
||||
|
||||
// TestSimSchedulerDeterminism asserts that the same batch run twice against
|
||||
// the same seeds produces the same traces, which is the property every sealed
|
||||
// tier depends on and the one real parallelism would have destroyed.
|
||||
func TestSimSchedulerDeterminism(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
run := func() []*SimScenarioResult {
|
||||
graph := trafficTestGraph(t, 3)
|
||||
nodes := sortedNodes(graph)
|
||||
|
||||
runner, err := NewSimRunner(
|
||||
graph, DefaultSimParams(), nodes[0], t.TempDir(),
|
||||
)
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(runner.Close)
|
||||
|
||||
runner.SetVirtualClock(&SimClockParams{
|
||||
StartUnix: 1_800_000_000,
|
||||
PaymentGapSec: 600,
|
||||
AttemptSec: 30,
|
||||
})
|
||||
require.NoError(t, runner.SetBackgroundTraffic(
|
||||
&SimTrafficParams{
|
||||
PaymentsPerGap: 20,
|
||||
MinAmtMsat: 1_000,
|
||||
MaxAmtMsat: 1_000_000,
|
||||
Seed: 11,
|
||||
},
|
||||
))
|
||||
|
||||
scenarios := make([]SimScenario, 0, 6)
|
||||
for i := 1; i <= 6; i++ {
|
||||
scenarios = append(scenarios, SimScenario{
|
||||
Target: nodes[i*3].String(),
|
||||
AmtMsat: 50_000_000,
|
||||
MaxParts: 4,
|
||||
AtomicMpp: true,
|
||||
})
|
||||
}
|
||||
|
||||
results, err := runner.RunBatch(scenarios,
|
||||
&SimConcurrencyParams{MaxInFlight: 3})
|
||||
require.NoError(t, err)
|
||||
|
||||
return results
|
||||
}
|
||||
|
||||
require.Equal(t, run(), run())
|
||||
}
|
||||
|
||||
// TestSimSchedulerOverlapLadder asserts that raising the window actually makes
|
||||
// the sender's payments overlap. This is the manipulation check: a concurrency
|
||||
// tier whose payments never run at the same time is testing nothing, and the
|
||||
// score would say nothing about that.
|
||||
func TestSimSchedulerOverlapLadder(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
run := func(inFlight int) SimConcurrencyStats {
|
||||
graph := trafficTestGraph(t, 3)
|
||||
nodes := sortedNodes(graph)
|
||||
|
||||
runner, err := NewSimRunner(
|
||||
graph, DefaultSimParams(), nodes[0], t.TempDir(),
|
||||
)
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(runner.Close)
|
||||
|
||||
runner.SetVirtualClock(&SimClockParams{
|
||||
StartUnix: 1_800_000_000,
|
||||
PaymentGapSec: 600,
|
||||
AttemptSec: 30,
|
||||
})
|
||||
|
||||
scenarios := make([]SimScenario, 0, 8)
|
||||
for i := 1; i <= 8; i++ {
|
||||
scenarios = append(scenarios, SimScenario{
|
||||
Target: nodes[i*2].String(),
|
||||
AmtMsat: 200_000_000,
|
||||
MaxParts: 4,
|
||||
AtomicMpp: true,
|
||||
})
|
||||
}
|
||||
|
||||
var params *SimConcurrencyParams
|
||||
if inFlight > 1 {
|
||||
params = &SimConcurrencyParams{MaxInFlight: inFlight}
|
||||
}
|
||||
|
||||
_, err = runner.RunBatch(scenarios, params)
|
||||
require.NoError(t, err)
|
||||
|
||||
return runner.ConcurrencyStats()
|
||||
}
|
||||
|
||||
one := run(1)
|
||||
require.Equal(t, 1, one.MaxConcurrent)
|
||||
require.InDelta(t, 1.0, one.MeanConcurrent, 1e-9)
|
||||
|
||||
for _, inFlight := range []int{2, 4} {
|
||||
stats := run(inFlight)
|
||||
require.Equal(t, inFlight, stats.MaxConcurrent,
|
||||
"window %d never filled", inFlight)
|
||||
require.Greater(t, stats.MeanConcurrent, 1.0,
|
||||
"window %d never overlapped", inFlight)
|
||||
require.Less(t, stats.MakespanSec, one.MakespanSec,
|
||||
"window %d did not clear faster", inFlight)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSimSchedulerNeedsAClock asserts that a concurrency section on a file
|
||||
// with no virtual time is refused rather than measured. With no clock every
|
||||
// event ties at the same instant, the loop degenerates to running the payments
|
||||
// in index order, and the scheduling counters would report a window that never
|
||||
// opened.
|
||||
func TestSimSchedulerNeedsAClock(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
graph, nodes := atomicTestGraph(t)
|
||||
source, target := nodes[0], nodes[3]
|
||||
|
||||
factory, _ := scriptedFactory(t, [][]*route.Route{})
|
||||
runner := concurrencyRunner(t, graph, source, factory)
|
||||
|
||||
_, err := runner.RunBatch([]SimScenario{
|
||||
{Target: target.String(), AmtMsat: 1_000_000},
|
||||
}, &SimConcurrencyParams{MaxInFlight: 2})
|
||||
require.ErrorContains(t, err, "needs a clock section")
|
||||
|
||||
// The sequential batch is unaffected: it is what every scenario file
|
||||
// with no clock has always run.
|
||||
_, err = runner.RunBatch([]SimScenario{}, nil)
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
// refreshRouter is a scripted router that also takes balance refreshes, and
|
||||
// records every map it was handed.
|
||||
type refreshRouter struct {
|
||||
scriptedRouter
|
||||
|
||||
refreshed []map[uint64]lnwire.MilliSatoshi
|
||||
}
|
||||
|
||||
// RefreshLocalBalances records the refreshed view of the sender's own
|
||||
// liquidity.
|
||||
//
|
||||
// NOTE: Part of the SimBalanceRefresher interface.
|
||||
func (r *refreshRouter) RefreshLocalBalances(
|
||||
balances map[uint64]lnwire.MilliSatoshi) {
|
||||
|
||||
r.refreshed = append(r.refreshed, balances)
|
||||
}
|
||||
|
||||
// TestSimSchedulerSelfContention asserts that an attempt that fails because
|
||||
// ANOTHER of the sender's own payments is holding the liquidity is counted,
|
||||
// and that the same attempt against the same shortfall with no sibling holding
|
||||
// anything is not.
|
||||
//
|
||||
// This is the number the whole stage exists to produce. Under atomic mpp a
|
||||
// shard that reaches the destination reserves every hop it crossed, so a
|
||||
// second payment finds the sender's own first channel short of liquidity that
|
||||
// nothing in its gossip view can explain.
|
||||
func TestSimSchedulerSelfContention(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const shard = lnwire.MilliSatoshi(100_000_000)
|
||||
|
||||
// run sends two payments over the sender's channel 1, which funds one
|
||||
// and a half shards. The first payment is an mpp that puts one shard
|
||||
// over channel 1 and the rest over channel 3, so it is still holding
|
||||
// channel 1 when the second payment tries to use it.
|
||||
run := func(inFlight int, atomic bool) SimConcurrencyStats {
|
||||
graph, nodes := atomicTestGraph(t)
|
||||
source, target := nodes[0], nodes[3]
|
||||
|
||||
atomicSetBalance(t, graph, 1, source, shard+shard/2)
|
||||
|
||||
viaA := atomicTestRoute(t, graph, source, []uint64{1, 2}, shard)
|
||||
viaB := atomicTestRoute(t, graph, source, []uint64{3, 4}, shard)
|
||||
factory, _ := scriptedFactory(t, [][]*route.Route{
|
||||
{viaA, viaB}, {viaA},
|
||||
})
|
||||
|
||||
runner := concurrencyRunner(t, graph, source, factory)
|
||||
runner.SetVirtualClock(&SimClockParams{
|
||||
StartUnix: 1_800_000_000,
|
||||
PaymentGapSec: 600,
|
||||
AttemptSec: 30,
|
||||
})
|
||||
|
||||
var params *SimConcurrencyParams
|
||||
if inFlight > 1 {
|
||||
params = &SimConcurrencyParams{MaxInFlight: inFlight}
|
||||
}
|
||||
|
||||
_, err := runner.RunBatch([]SimScenario{
|
||||
{Target: target.String(), AmtMsat: uint64(2 * shard),
|
||||
MaxParts: 2, AtomicMpp: atomic},
|
||||
{Target: target.String(), AmtMsat: uint64(shard),
|
||||
MaxParts: 1, AtomicMpp: atomic},
|
||||
}, params)
|
||||
require.NoError(t, err)
|
||||
requireNoHolds(t, graph)
|
||||
|
||||
return runner.ConcurrencyStats()
|
||||
}
|
||||
|
||||
// Sequential: the first payment has settled and moved the liquidity
|
||||
// before the second one is even built, so the second one's shortfall is
|
||||
// a fact about the network rather than about its sibling.
|
||||
sequential := run(1, true)
|
||||
require.Equal(t, 1, sequential.MaxConcurrent)
|
||||
require.Zero(t, sequential.SelfContentionFailures)
|
||||
|
||||
// Concurrent and atomic: the first payment's shard is still held on
|
||||
// channel 1, and it is the reason the second one fails there.
|
||||
contended := run(2, true)
|
||||
require.Equal(t, 2, contended.MaxConcurrent)
|
||||
require.Equal(t, 1, contended.SelfContentionFailures)
|
||||
|
||||
// Concurrent without holds is the free control. A shard that settles
|
||||
// the instant it arrives reserves nothing, so the counter is
|
||||
// structurally zero however heavily the payments overlap.
|
||||
noHolds := run(2, false)
|
||||
require.Equal(t, 2, noHolds.MaxConcurrent)
|
||||
require.Zero(t, noHolds.SelfContentionFailures)
|
||||
}
|
||||
|
||||
// TestRouterAcceptsBalanceRefreshFalseForPlainRouter asserts that a router
|
||||
// without the optional refresh half is reported as such, so a sweep can tell
|
||||
// an ineffective refresh from an undelivered one.
|
||||
func TestRouterAcceptsBalanceRefreshFalseForPlainRouter(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
graph, nodes := atomicTestGraph(t)
|
||||
source, target := nodes[0], nodes[3]
|
||||
|
||||
rt := atomicTestRoute(
|
||||
t, graph, source, []uint64{1, 2}, lnwire.MilliSatoshi(10_000),
|
||||
)
|
||||
factory, _ := scriptedFactory(t, [][]*route.Route{{rt}})
|
||||
|
||||
runner := concurrencyRunner(t, graph, source, factory)
|
||||
require.False(t, runner.RouterAcceptsBalanceRefresh())
|
||||
|
||||
_, err := runner.RunBatch([]SimScenario{
|
||||
{Target: target.String(), AmtMsat: 10_000, MaxParts: 1},
|
||||
}, nil)
|
||||
require.NoError(t, err)
|
||||
|
||||
require.False(t, runner.RouterAcceptsBalanceRefresh())
|
||||
require.False(
|
||||
t, runner.ConcurrencyStats().RouterAcceptsBalanceRefresh,
|
||||
)
|
||||
}
|
||||
|
||||
// TestSimSchedulerBalanceRefreshIsDelivered asserts that the optional half is
|
||||
// not dead plumbing: a router that implements it is told, before every route
|
||||
// request, what its own outbound liquidity is now, and the number it is told is
|
||||
// net of what is currently held.
|
||||
//
|
||||
// The single-payment case is the exact one, and it is the same staleness the
|
||||
// concurrent case has: what a router was handed at construction stops being
|
||||
// true the moment any shard reserves liquidity, whether that shard is its own
|
||||
// or a sibling payment's.
|
||||
func TestSimSchedulerBalanceRefreshIsDelivered(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const shard = lnwire.MilliSatoshi(100_000_000)
|
||||
|
||||
graph, nodes := atomicTestGraph(t)
|
||||
source, target := nodes[0], nodes[3]
|
||||
|
||||
atomicSetBalance(t, graph, 1, source, 3*shard)
|
||||
|
||||
viaA := atomicTestRoute(t, graph, source, []uint64{1, 2}, shard)
|
||||
viaB := atomicTestRoute(t, graph, source, []uint64{3, 4}, shard)
|
||||
|
||||
built := make([]*refreshRouter, 0, 1)
|
||||
runner := concurrencyRunner(t, graph, source,
|
||||
func(_ SimNetworkView, _ route.Vertex,
|
||||
_ map[uint64]lnwire.MilliSatoshi,
|
||||
_ *SimPaymentSpec) (SimRouter, error) {
|
||||
|
||||
router := &refreshRouter{
|
||||
scriptedRouter: scriptedRouter{
|
||||
routes: []*route.Route{viaA, viaB},
|
||||
},
|
||||
}
|
||||
built = append(built, router)
|
||||
|
||||
return router, nil
|
||||
},
|
||||
)
|
||||
runner.SetVirtualClock(&SimClockParams{
|
||||
StartUnix: 1_800_000_000,
|
||||
PaymentGapSec: 600,
|
||||
AttemptSec: 30,
|
||||
})
|
||||
|
||||
require.False(t, runner.RouterAcceptsBalanceRefresh(),
|
||||
"the capability is latched at the first router, not before")
|
||||
|
||||
results, err := runner.RunBatch([]SimScenario{{
|
||||
Target: target.String(),
|
||||
AmtMsat: uint64(2 * shard),
|
||||
MaxParts: 2,
|
||||
AtomicMpp: true,
|
||||
}}, nil)
|
||||
require.NoError(t, err)
|
||||
require.True(t, results[0].Success)
|
||||
|
||||
require.True(t, runner.RouterAcceptsBalanceRefresh())
|
||||
require.True(
|
||||
t, runner.ConcurrencyStats().RouterAcceptsBalanceRefresh,
|
||||
)
|
||||
|
||||
require.Len(t, built, 1)
|
||||
require.Len(t, built[0].refreshed, 2,
|
||||
"the router was not told once per route request")
|
||||
|
||||
// The first request saw the whole balance of channel 1; the second was
|
||||
// made while the first shard was held on it, so what it was told is
|
||||
// short by exactly what that shard reserved.
|
||||
require.Equal(t, 3*shard, built[0].refreshed[0][1])
|
||||
require.Equal(
|
||||
t, 3*shard-viaA.TotalAmount, built[0].refreshed[1][1],
|
||||
)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -44,6 +44,33 @@ type SimRouter interface {
|
|||
result SimHtlcResult) error
|
||||
}
|
||||
|
||||
// SimBalanceRefresher is the optional half of the SimRouter contract that a
|
||||
// router implements if it wants to be told that its own outbound liquidity
|
||||
// changed under it.
|
||||
//
|
||||
// A router is built once per payment and handed the sender's local balances at
|
||||
// that moment. While one payment runs at a time that snapshot stays true for
|
||||
// the whole payment. It stops being true the moment the sender runs several of
|
||||
// its own payments at once: a sibling's shard takes some of the same outbound
|
||||
// liquidity, and the map this router is planning against still shows it.
|
||||
//
|
||||
// A router that does not implement this keeps the snapshot it was built with,
|
||||
// which is what every router in this program does today. That is a legitimate
|
||||
// design, not a bug: the balances a sender is handed at plan time are what a
|
||||
// real node's path finding runs against too. The interface exists so that a
|
||||
// router which wants the update can have it, and so that "the refresh did not
|
||||
// help" is distinguishable from "the refresh was never delivered" — exp-016
|
||||
// had to hand-write importer variants of two champions after the fact because
|
||||
// nothing in the contract had ever asked for the capability.
|
||||
type SimBalanceRefresher interface {
|
||||
// RefreshLocalBalances delivers the sender's current outbound
|
||||
// liquidity per channel id, net of what its own in-flight htlcs hold,
|
||||
// before each route request. Implementations must treat it as a
|
||||
// replacement for the map they were built with rather than as
|
||||
// additional evidence: it is the same measurement, taken later.
|
||||
RefreshLocalBalances(balances map[uint64]lnwire.MilliSatoshi)
|
||||
}
|
||||
|
||||
// SimPaymentSpec describes one payment for a router to complete.
|
||||
//
|
||||
// Everything here is information a real sender has about its own payment
|
||||
|
|
|
|||
|
|
@ -310,6 +310,17 @@ type SimRunner struct {
|
|||
// path see the same degraded result from the same draw.
|
||||
attribution *simAttribution
|
||||
|
||||
// concurrency is what the last scored batch reported about its own
|
||||
// scheduling, zero until one has run.
|
||||
concurrency SimConcurrencyStats
|
||||
|
||||
// balanceRefresh records whether the routing strategy under test
|
||||
// implements the optional refresh half of the contract, set the first
|
||||
// time a router is built. It is a property of the strategy rather than
|
||||
// of any one payment, and it is reported so that a null cannot be
|
||||
// silent.
|
||||
balanceRefresh bool
|
||||
|
||||
// trafficCarry is the fractional background payment left over from
|
||||
// pro-rating the per-gap volume across attempts. Carrying it keeps the
|
||||
// traffic rate inside a payment equal to the rate between payments
|
||||
|
|
@ -594,41 +605,69 @@ func (r *SimRunner) AdvanceIdle(seconds float64) {
|
|||
}
|
||||
}
|
||||
|
||||
// advanceGap moves virtual time forward by the payment gap and lets the
|
||||
// background traffic use that window.
|
||||
func (r *SimRunner) advanceGap() {
|
||||
if r.virtualClk != nil && r.clockParams.PaymentGapSec > 0 {
|
||||
r.virtualClk.SetTime(r.virtualClk.Now().Add(
|
||||
time.Duration(r.clockParams.PaymentGapSec *
|
||||
float64(time.Second)),
|
||||
))
|
||||
// simSchedTime is the scheduler's ordering key: the virtual clock's current
|
||||
// reading, or the zero time on a scenario that configures none, where nothing
|
||||
// the simulator does moves a clock at all and every event ties.
|
||||
func (r *SimRunner) simSchedTime() time.Time {
|
||||
if r.virtualClk == nil {
|
||||
return time.Time{}
|
||||
}
|
||||
|
||||
if r.traffic != nil {
|
||||
r.traffic.run()
|
||||
}
|
||||
return r.virtualClk.Now()
|
||||
}
|
||||
|
||||
// advanceAttempt moves virtual time forward by one attempt's duration. Under
|
||||
// atomic mpp the background traffic engine also runs for that slice of time,
|
||||
// so hidden liquidity keeps drifting while a payment's shards are in flight
|
||||
// rather than freezing until the payment resolves. That is what makes a
|
||||
// serial probe-learn-resize strategy pay for the time it takes.
|
||||
func (r *SimRunner) advanceAttempt(atomicMpp bool) {
|
||||
// simAdvanceTo moves virtual time forward to the given instant and runs the
|
||||
// background traffic that belongs to the interval, through the same prorating
|
||||
// path every other advance uses.
|
||||
//
|
||||
// Advancing to an instant already past does nothing, which is what the
|
||||
// scheduler needs: a step can move the clock under it, since an attribution
|
||||
// delay ages the network in the middle of an attempt, and the events that were
|
||||
// due inside that window are then simply late rather than run backwards.
|
||||
//
|
||||
// runTraffic is the caller's answer to whether the exogenous process should
|
||||
// run for this stretch at all. It exists because the sequential loop this
|
||||
// replaces answered it two different ways: the gap between payments always
|
||||
// churned, and the time inside a payment only churned under atomic mpp, where
|
||||
// shards left in flight are supposed to pay for the time they take.
|
||||
func (r *SimRunner) simAdvanceTo(target time.Time, runTraffic bool) {
|
||||
if r.virtualClk == nil {
|
||||
return
|
||||
}
|
||||
|
||||
now := r.virtualClk.Now()
|
||||
if !target.After(now) {
|
||||
return
|
||||
}
|
||||
|
||||
r.virtualClk.SetTime(target)
|
||||
|
||||
if r.traffic == nil || !runTraffic {
|
||||
return
|
||||
}
|
||||
|
||||
r.traffic.runN(r.trafficPaymentsFor(target.Sub(now).Seconds()))
|
||||
}
|
||||
|
||||
// simAttemptStep is how much virtual time one htlc attempt consumes, and zero
|
||||
// on a scenario with no virtual clock, where an attempt has always taken no
|
||||
// time at all.
|
||||
func (r *SimRunner) simAttemptStep() time.Duration {
|
||||
if r.virtualClk == nil || r.clockParams.AttemptSec <= 0 {
|
||||
return
|
||||
return 0
|
||||
}
|
||||
|
||||
r.virtualClk.SetTime(r.virtualClk.Now().Add(
|
||||
time.Duration(r.clockParams.AttemptSec *
|
||||
float64(time.Second)),
|
||||
))
|
||||
return time.Duration(r.clockParams.AttemptSec * float64(time.Second))
|
||||
}
|
||||
|
||||
if !atomicMpp || r.traffic == nil {
|
||||
return
|
||||
// simGapStep is how much virtual time passes between one payment finishing and
|
||||
// the next starting, and zero on a scenario with no virtual clock.
|
||||
func (r *SimRunner) simGapStep() time.Duration {
|
||||
if r.virtualClk == nil || r.clockParams.PaymentGapSec <= 0 {
|
||||
return 0
|
||||
}
|
||||
|
||||
r.traffic.runN(r.trafficPaymentsFor(r.clockParams.AttemptSec))
|
||||
return time.Duration(r.clockParams.PaymentGapSec * float64(time.Second))
|
||||
}
|
||||
|
||||
// trafficPaymentsFor returns how many background payments belong to the given
|
||||
|
|
@ -708,258 +747,93 @@ func (r *SimRunner) RunScenario(s *SimScenario) (*SimScenarioResult, error) {
|
|||
// gathered: the payment probes real channels and everything it learns lands in
|
||||
// the one shared mission control, which stays anchored to the runner's source
|
||||
// throughout. Whether that knowledge is worth anything to the runner's source
|
||||
// afterwards is exactly the question — pair history is entangled with the
|
||||
// afterwards is exactly the question. Pair history is entangled with the
|
||||
// vantage that observed it, while a belief about a directed channel's
|
||||
// liquidity is a fact about the channel.
|
||||
//
|
||||
// One payment is a batch of one, run on the same scheduler a concurrent batch
|
||||
// runs on, which is what makes the sequential behavior the concurrent path is
|
||||
// required to reproduce the behavior it actually reproduces rather than a
|
||||
// second implementation of it.
|
||||
func (r *SimRunner) RunScenarioFrom(source route.Vertex,
|
||||
s *SimScenario) (*SimScenarioResult, error) {
|
||||
|
||||
if r.graph.Node(source) == nil {
|
||||
return nil, fmt.Errorf("source node %v not in graph", source)
|
||||
}
|
||||
|
||||
result := &SimScenarioResult{Scenario: *s}
|
||||
|
||||
// Let virtual time pass and background traffic move liquidity before
|
||||
// this payment starts, the way a live network keeps churning between
|
||||
// a node's own sends.
|
||||
r.advanceGap()
|
||||
|
||||
target, err := r.graph.ResolveNode(s.Target)
|
||||
results, _, err := r.runBatch(source, []SimScenario{*s}, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
maxParts := s.MaxParts
|
||||
if maxParts == 0 {
|
||||
maxParts = 16
|
||||
}
|
||||
return results[0], nil
|
||||
}
|
||||
|
||||
amount := lnwire.MilliSatoshi(s.AmtMsat)
|
||||
spec := &SimPaymentSpec{
|
||||
Target: target,
|
||||
Amount: amount,
|
||||
MaxParts: maxParts,
|
||||
// RunBatch executes a whole batch of payments from the runner's source, with
|
||||
// as many of them live at once as the concurrency section allows. A nil
|
||||
// section is the sequential batch: one payment at a time, each starting a
|
||||
// payment gap after the last one resolved, which is what every scenario file
|
||||
// written before stage D asks for by omission.
|
||||
//
|
||||
// The concurrency statistics of the batch are recorded on the runner and read
|
||||
// back with ConcurrencyStats.
|
||||
func (r *SimRunner) RunBatch(scenarios []SimScenario,
|
||||
params *SimConcurrencyParams) ([]*SimScenarioResult, error) {
|
||||
|
||||
// The budget is quoted as a share of the payment's own amount,
|
||||
// so that one number describes a corpus whose amounts run over
|
||||
// four orders of magnitude. With no limit set this is
|
||||
// lnwire.MaxMilliSatoshi, which is the value the lnd arm has
|
||||
// been constructed with for the whole program.
|
||||
FeeLimitMsat: simFeeBudgetMsat(amount, s.FeeLimitPPM),
|
||||
}
|
||||
|
||||
if spec.FeeLimitMsat != lnwire.MaxMilliSatoshi {
|
||||
r.feeLimitStats.Payments++
|
||||
}
|
||||
|
||||
// Build the routing strategy under test for this payment, handing it
|
||||
// the public graph view and the sender's exact local balances. The
|
||||
// view wrapper hides the concrete graph so that a candidate router
|
||||
// cannot reach the hidden balances.
|
||||
router, err := r.routerFactory(
|
||||
&simGossipView{g: r.graph, now: r.clk.Now}, source,
|
||||
r.graph.LocalBalances(source), spec,
|
||||
)
|
||||
results, idx, err := r.runBatch(r.source, scenarios, params)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return nil, fmt.Errorf("scenario %d failed: %v", idx, err)
|
||||
}
|
||||
|
||||
// Hand over any served knowledge before the router plans anything,
|
||||
// so that imported beliefs are available to the very first route
|
||||
// request rather than arriving after the payment has committed.
|
||||
if err := r.deliverPendingImport(router); err != nil {
|
||||
return nil, err
|
||||
return results, nil
|
||||
}
|
||||
|
||||
// runBatch drives the scheduler and reports which scenario a fatal error came
|
||||
// from, so that the caller can name it the way the sequential batch always
|
||||
// has.
|
||||
func (r *SimRunner) runBatch(source route.Vertex, scenarios []SimScenario,
|
||||
params *SimConcurrencyParams) ([]*SimScenarioResult, int, error) {
|
||||
|
||||
scheduler, err := newSimScheduler(r, source, scenarios, params)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
|
||||
var (
|
||||
nextAttemptID uint64
|
||||
amtRemaining = spec.Amount
|
||||
inFlightHtlcs uint32
|
||||
|
||||
// holdIDs are the shards that have reached the destination but
|
||||
// are not settled yet, only ever populated under atomic mpp.
|
||||
// Their amount and fees ride along until the whole set either
|
||||
// settles or is released.
|
||||
holdIDs []uint64
|
||||
heldMsat uint64
|
||||
heldFees uint64
|
||||
)
|
||||
|
||||
// Under atomic mpp a payment that never completes settles nothing:
|
||||
// every shard still held when the loop exits gives its reserved
|
||||
// liquidity back, so a failed mpp leaves the hidden balances exactly
|
||||
// as it found them and charges no fees. The success path settles the
|
||||
// set and clears holdIDs before returning, so this only ever fires on
|
||||
// a failure path, whichever one it is.
|
||||
defer func() {
|
||||
if len(holdIDs) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
for _, id := range holdIDs {
|
||||
r.graph.ReleaseHold(id)
|
||||
}
|
||||
result.HeldReleasedMsat = heldMsat
|
||||
}()
|
||||
|
||||
for len(result.Attempts) < simMaxAttempts {
|
||||
// Ask the router for the next route to attempt.
|
||||
rt, err := router.RequestRoute(amtRemaining, inFlightHtlcs)
|
||||
if err != nil {
|
||||
result.Error = err.Error()
|
||||
result.GaveUp = true
|
||||
break
|
||||
}
|
||||
|
||||
attemptID := nextAttemptID
|
||||
nextAttemptID++
|
||||
|
||||
// The fee budget is enforced HERE, at the point the runner
|
||||
// would dispatch, and not inside any router. That is the
|
||||
// exp-019 construction: a constraint that lives at the shared
|
||||
// delivery point is the same constraint for the lnd stack and
|
||||
// for an evolved candidate, and neither of them can be given a
|
||||
// gentler version of it by accident.
|
||||
//
|
||||
// What the budget has left is what it started with less the
|
||||
// fees this payment has already committed, which is the fees
|
||||
// of the shards that settled plus the fees riding on the ones
|
||||
// still held. lnd's own lifecycle subtracts exactly that
|
||||
// (calcFeeBudget over FeesPaid), and the lnd arm is handed the
|
||||
// same remainder, so this backstop should never fire for it.
|
||||
committed := lnwire.MilliSatoshi(result.FeeMsat + heldFees)
|
||||
remaining := simRemainingBudget(spec.FeeLimitMsat, committed)
|
||||
if rt.TotalFees() > remaining {
|
||||
// The refusal is a fact about this sender, not about
|
||||
// the network: no forwarding node saw the htlc, so
|
||||
// nothing is recorded in the observation stream, no
|
||||
// virtual time passes, and the result is handed to the
|
||||
// router undegraded. An attribution section damages
|
||||
// what came back over the wire, and nothing came back
|
||||
// over the wire; running this through the degrader
|
||||
// would also consume draws and shift the sequence
|
||||
// every paired exp-019 run depends on.
|
||||
//
|
||||
// It does cost an attempt. A router that keeps
|
||||
// offering routes it cannot afford spends its attempt
|
||||
// budget on them, which is the whole point of putting
|
||||
// the pressure in the environment.
|
||||
refusal := SimHtlcResult{
|
||||
FailureSource: rt.SourcePubKey,
|
||||
Failure: SimFeeLimitFailure{},
|
||||
}
|
||||
|
||||
result.Attempts = append(
|
||||
result.Attempts, traceAttempt(rt, refusal),
|
||||
)
|
||||
r.feeLimitStats.Failures++
|
||||
|
||||
err = router.ReportAttempt(attemptID, rt, refusal)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
// Send the htlc through the simulated network. A malformed
|
||||
// route (unknown channel, disconnected hops) is a router bug:
|
||||
// it terminates this payment with an error rather than
|
||||
// killing the whole batch, so one bad edge case doesn't zero
|
||||
// out an otherwise functional candidate.
|
||||
|
||||
// Each attempt consumes virtual time: htlcs take real seconds
|
||||
// to resolve on a live network.
|
||||
r.advanceAttempt(s.AtomicMpp)
|
||||
|
||||
// An atomic shard is held at the destination rather than
|
||||
// settled there, reserving the liquidity of every hop it
|
||||
// crossed until the payment as a whole resolves.
|
||||
var (
|
||||
htlcResult SimHtlcResult
|
||||
holdID uint64
|
||||
)
|
||||
if s.AtomicMpp {
|
||||
htlcResult, holdID, err = r.graph.HoldHtlc(rt)
|
||||
} else {
|
||||
htlcResult, err = r.graph.SendHtlc(rt)
|
||||
}
|
||||
if err != nil {
|
||||
result.Error = fmt.Sprintf("malformed route: %v", err)
|
||||
break
|
||||
}
|
||||
|
||||
result.Attempts = append(
|
||||
result.Attempts, traceAttempt(rt, htlcResult),
|
||||
)
|
||||
|
||||
// Record what this attempt revealed about the edges it
|
||||
// crossed, which is the raw material a weight-serving node
|
||||
// would have to offer.
|
||||
r.observations = append(r.observations, observationsFromAttempt(
|
||||
rt, htlcResult, r.clk.Now(),
|
||||
)...)
|
||||
|
||||
// Let the router learn from the outcome. The feedback is the
|
||||
// same either way: what atomic mpp changes is the price of a
|
||||
// serial probe, not the information it returns.
|
||||
//
|
||||
// Everything above this line records what actually happened.
|
||||
// What the router is TOLD may be less than that: with an
|
||||
// attribution section configured the result is aged and its
|
||||
// attribution damaged first, so the trace and the served
|
||||
// observations keep the truth while the router works from the
|
||||
// same imperfect channel a mainnet sender has.
|
||||
err = router.ReportAttempt(
|
||||
attemptID, rt, r.deliverAttempt(rt, htlcResult),
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if htlcResult.Failure != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
inFlightHtlcs++
|
||||
|
||||
// A settling shard pays its fee right away; a held one only
|
||||
// pays when the whole set settles.
|
||||
if s.AtomicMpp {
|
||||
holdIDs = append(holdIDs, holdID)
|
||||
heldMsat += uint64(rt.ReceiverAmt())
|
||||
heldFees += uint64(rt.TotalFees())
|
||||
} else {
|
||||
result.FeeMsat += uint64(rt.TotalFees())
|
||||
}
|
||||
|
||||
// Guard against a buggy router delivering more than asked:
|
||||
// unsigned underflow here would loop until the attempt cap.
|
||||
recv := rt.ReceiverAmt()
|
||||
if recv > amtRemaining {
|
||||
result.Error = "router over-delivered payment amount"
|
||||
break
|
||||
}
|
||||
amtRemaining -= recv
|
||||
|
||||
if amtRemaining == 0 {
|
||||
// The full amount has arrived, so the held set becomes
|
||||
// real balance movement all at once and the fees it
|
||||
// carried finally come due. Without atomic mpp there
|
||||
// is nothing held and this is a no-op.
|
||||
for _, id := range holdIDs {
|
||||
r.graph.SettleHold(id)
|
||||
}
|
||||
holdIDs = nil
|
||||
result.FeeMsat += heldFees
|
||||
|
||||
result.Success = true
|
||||
break
|
||||
}
|
||||
results, idx, err := scheduler.run()
|
||||
if err != nil {
|
||||
return nil, idx, err
|
||||
}
|
||||
|
||||
return result, nil
|
||||
r.concurrency = scheduler.stats
|
||||
|
||||
return results, 0, nil
|
||||
}
|
||||
|
||||
// ConcurrencyStats reports what the last batch's scheduling actually did: how
|
||||
// many of the sender's payments overlapped, how often one of them took
|
||||
// liquidity another one wanted, and how long the batch took in virtual time.
|
||||
// None of it enters the objective.
|
||||
func (r *SimRunner) ConcurrencyStats() SimConcurrencyStats {
|
||||
return r.concurrency
|
||||
}
|
||||
|
||||
// noteBalanceRefreshCapability records that the routing strategy under test
|
||||
// does or does not take balance refreshes. It is latched on the first router
|
||||
// built, since the strategy is fixed for the life of the runner.
|
||||
func (r *SimRunner) noteBalanceRefreshCapability(accepts bool) {
|
||||
if accepts {
|
||||
r.balanceRefresh = true
|
||||
}
|
||||
}
|
||||
|
||||
// RouterAcceptsBalanceRefresh reports whether the routing strategy under test
|
||||
// implements the optional refresh half of the contract, so that a sweep can
|
||||
// tell an ineffective refresh from an undelivered one.
|
||||
//
|
||||
// Unlike served observations, there is no second path into a router here: the
|
||||
// lnd stack holds the bandwidth hints it was built with and takes no refresh
|
||||
// either, so this reads false for every arm shipped with this stage. That is
|
||||
// the finding rather than a gap: nothing in the contract had ever asked any
|
||||
// of them to be told, and now that something does, the flag says who answered.
|
||||
func (r *SimRunner) RouterAcceptsBalanceRefresh() bool {
|
||||
return r.balanceRefresh
|
||||
}
|
||||
|
||||
// traceAttempt converts a route and its resolution into a trace record.
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue