diff --git a/routing/sim_concurrency.go b/routing/sim_concurrency.go index 0b4cbddf7..62a7a07bc 100644 --- a/routing/sim_concurrency.go +++ b/routing/sim_concurrency.go @@ -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() +} diff --git a/routing/sim_concurrency_test.go b/routing/sim_concurrency_test.go index d04c6a135..9a60d29ab 100644 --- a/routing/sim_concurrency_test.go +++ b/routing/sim_concurrency_test.go @@ -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], + ) +} diff --git a/routing/sim_router.go b/routing/sim_router.go index 4c0de41a1..822285e50 100644 --- a/routing/sim_router.go +++ b/routing/sim_router.go @@ -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 diff --git a/routing/sim_run.go b/routing/sim_run.go index 8680d6ff9..2db0ea4fa 100644 --- a/routing/sim_run.go +++ b/routing/sim_run.go @@ -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.