From d0f062747ddca654e1791c124056f5620495bf70 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Sat, 25 Jul 2026 17:38:09 -0700 Subject: [PATCH] routing: add atomic MPP with held liquidity and mid-payment traffic In this commit, we give the simulator hold-and-release shard semantics behind a per-scenario atomic_mpp flag, the exp-010b arena change. A successful shard no longer settles instantly: each hop reserves its outgoing liquidity in a hold ledger owned by the graph, sibling shards and background traffic see availability net of holds, and the payment's holds settle together on completion or release together on failure -- so failed MPP payments become atomic and no longer move the network. The hop walk is shared between the settle and hold paths via a commit-mode parameter, keeping the forwarding checks and the amount arithmetic identical between modes. The world also keeps turning while an atomic payment probes: each attempt runs the background traffic engine pro-rated by the attempt duration (with fractional carry across attempts), so the exogenous process drifts at one rate throughout instead of freezing whenever a payment is in flight. Sequential probe-learn-resize now pays for its information in time and reservation, which is what it costs on mainnet. With the flag off, behavior is byte-identical over full corpora, verified end to end against a HEAD-built binary. --- routing/sim_atomic_test.go | 555 +++++++++++++++++++++++++++++++++++++ routing/sim_graph.go | 191 +++++++++++-- routing/sim_run.go | 124 ++++++++- routing/sim_traffic.go | 9 +- 4 files changed, 849 insertions(+), 30 deletions(-) create mode 100644 routing/sim_atomic_test.go diff --git a/routing/sim_atomic_test.go b/routing/sim_atomic_test.go new file mode 100644 index 000000000..bcf9c2ac3 --- /dev/null +++ b/routing/sim_atomic_test.go @@ -0,0 +1,555 @@ +package routing + +import ( + "fmt" + "testing" + + "github.com/btcsuite/btcd/btcutil/v2" + "github.com/lightningnetwork/lnd/lnwire" + "github.com/lightningnetwork/lnd/routing/route" + "github.com/stretchr/testify/require" +) + +// atomicFinalCltv is the final cltv delta the hand-built test routes use. +const atomicFinalCltv = 40 + +// atomicChanCapSat is the capacity of every channel in the atomic mpp +// fixture, large enough that the shard amounts the tests send are a small +// fraction of it. +const atomicChanCapSat btcutil.Amount = 1_000_000 + +// atomicTestPolicy is the forwarding policy both ends of every fixture +// channel announce. The fees are deliberately non-zero so that the settlement +// tests compare something more interesting than a bare amount. +var atomicTestPolicy = SimPolicy{ + BaseFeeMsat: 1_000, + FeeRatePPM: 100, + TimeLockDelta: 40, +} + +// atomicTestGraph builds the fixture the atomic mpp tests route over: a +// source with two disjoint two-hop paths to a single target. +// +// S ── c1 ── A ── c2 ── T +// └── c3 ── B ── c4 ──┘ +// +// Every channel starts at an even split; the tests move the balances they +// care about with atomicSetBalance. +func atomicTestGraph(t *testing.T) (*SimGraph, [4]route.Vertex) { + t.Helper() + + graph := NewSimGraph() + + var nodes [4]route.Vertex + for i := range nodes { + nodes[i] = SimNodePubKey(uint32(i + 1)) + + _, err := graph.AddNode(nodes[i], fmt.Sprintf("n%d", i+1)) + require.NoError(t, err) + } + + source, nodeA, nodeB, target := nodes[0], nodes[1], nodes[2], nodes[3] + + links := []struct { + id uint64 + a, b route.Vertex + }{ + {1, source, nodeA}, + {2, nodeA, target}, + {3, source, nodeB}, + {4, nodeB, target}, + } + for _, link := range links { + require.NoError(t, graph.AddChannel( + link.id, link.a, link.b, atomicChanCapSat, + atomicTestPolicy, atomicTestPolicy, + )) + } + + return graph, nodes +} + +// atomicSetBalance pins the outbound balance of one end of a channel, giving +// the remainder of the capacity to the other end. +func atomicSetBalance(t *testing.T, g *SimGraph, chanID uint64, + owner route.Vertex, balance lnwire.MilliSatoshi) { + + t.Helper() + + channel, ok := g.channels[chanID] + require.True(t, ok, "unknown channel %d", chanID) + + capacity := lnwire.NewMSatFromSatoshis(channel.Capacity) + require.LessOrEqual(t, balance, capacity, "balance above capacity") + + end := channel.end(owner) + require.NotNil(t, end, "node is not a party to channel %d", chanID) + + end.balance = balance + channel.otherEnd(owner).balance = capacity - balance +} + +// atomicBalance returns the current outbound balance of one end of a channel. +func atomicBalance(t *testing.T, g *SimGraph, chanID uint64, + owner route.Vertex) lnwire.MilliSatoshi { + + t.Helper() + + channel, ok := g.channels[chanID] + require.True(t, ok, "unknown channel %d", chanID) + + end := channel.end(owner) + require.NotNil(t, end, "node is not a party to channel %d", chanID) + + return end.balance +} + +// atomicTestRoute builds a well-formed route from the source over the given +// channels, delivering amt to the far end of the last one. Amounts and +// expiries accumulate backward the way a real sender's path finding does, so +// the route clears the forwarding checks of every node it crosses. +func atomicTestRoute(t *testing.T, g *SimGraph, source route.Vertex, + chanIDs []uint64, amt lnwire.MilliSatoshi) *route.Route { + + t.Helper() + + require.NotEmpty(t, chanIDs, "route needs at least one channel") + + // Walk forward to learn the node sequence the channels describe. + nodes := []route.Vertex{source} + for _, id := range chanIDs { + channel, ok := g.channels[id] + require.True(t, ok, "unknown channel %d", id) + + next := channel.otherEnd(nodes[len(nodes)-1]) + require.NotNil(t, next, "channel %d does not extend the "+ + "route", id) + + nodes = append(nodes, next.owner) + } + + // Walk backward to accumulate the amount and expiry each channel has + // to carry, adding the fee and delta of the node that forwards onto + // the following channel. + last := len(chanIDs) - 1 + amts := make([]lnwire.MilliSatoshi, len(chanIDs)) + expiries := make([]uint32, len(chanIDs)) + amts[last] = amt + expiries[last] = atomicFinalCltv + + for k := last - 1; k >= 0; k-- { + policy := &g.channels[chanIDs[k+1]].end(nodes[k+1]).policy + + amts[k] = amts[k+1] + policy.fee(amts[k+1]) + expiries[k] = expiries[k+1] + uint32(policy.TimeLockDelta) + } + + // AmtToForward is the amount the hop's node sends ONWARD, which is + // what the next channel carries; the final hop forwards nothing, so it + // carries the delivered amount itself. + hops := make([]*route.Hop, len(chanIDs)) + for j := range chanIDs { + amtToForward, outgoingTimeLock := amt, uint32(atomicFinalCltv) + if j < last { + amtToForward = amts[j+1] + outgoingTimeLock = expiries[j+1] + } + + hops[j] = &route.Hop{ + PubKeyBytes: nodes[j+1], + ChannelID: chanIDs[j], + AmtToForward: amtToForward, + OutgoingTimeLock: outgoingTimeLock, + } + } + + return &route.Route{ + TotalAmount: amts[0], + TotalTimeLock: expiries[0], + SourcePubKey: source, + Hops: hops, + } +} + +// scriptedRouter is a SimRouter that hands back a fixed list of routes in +// order and records what came of each. It stands in for a candidate algorithm +// wherever a test needs the shard sequence to be exactly what it asked for. +type scriptedRouter struct { + routes []*route.Route + + // next is the index of the route the following RequestRoute returns. + next int + + // results holds the resolution of every attempt, in order. + results []SimHtlcResult + + // onReport, when set, runs after each attempt is recorded, the hook a + // test uses to observe the network mid-payment. + onReport func() +} + +// RequestRoute returns the next scripted route, failing the payment once the +// script runs out. +// +// NOTE: Part of the SimRouter interface. +func (s *scriptedRouter) RequestRoute(_ lnwire.MilliSatoshi, + _ uint32) (*route.Route, error) { + + if s.next >= len(s.routes) { + return nil, fmt.Errorf("scripted router out of routes") + } + + rt := s.routes[s.next] + s.next++ + + return rt, nil +} + +// ReportAttempt records the outcome and runs the observation hook. +// +// NOTE: Part of the SimRouter interface. +func (s *scriptedRouter) ReportAttempt(_ uint64, _ *route.Route, + result SimHtlcResult) error { + + s.results = append(s.results, result) + if s.onReport != nil { + s.onReport() + } + + return nil +} + +// atomicRunner builds a runner over the given graph that always uses the +// supplied scripted router. +func atomicRunner(t *testing.T, g *SimGraph, source route.Vertex, + router *scriptedRouter) *SimRunner { + + t.Helper() + + runner, err := NewSimRunner(g, DefaultSimParams(), source, t.TempDir()) + require.NoError(t, err) + t.Cleanup(runner.Close) + + runner.SetRouterFactory(func(_ SimNetworkView, _ route.Vertex, + _ map[uint64]lnwire.MilliSatoshi, + _ *SimPaymentSpec) (SimRouter, error) { + + return router, nil + }) + + return runner +} + +// requireNoHolds asserts that the graph is not holding any liquidity, the +// invariant that must be true whenever a payment has finished resolving. +func requireNoHolds(t *testing.T, g *SimGraph) { + t.Helper() + + require.Empty(t, g.holds, "holds outlived the payment") + + for id, channel := range g.channels { + for i := range channel.ends { + require.Zero( + t, channel.ends[i].held, + "channel %d end %d still holds liquidity", + id, i, + ) + } + } +} + +// TestSimAtomicMppRollsBackFailedPayment is the atomicity test: a payment that +// delivers a shard and then runs out of routes must leave every hidden balance +// exactly where it found it, and must pay nothing for the shard it rolled +// back. +func TestSimAtomicMppRollsBackFailedPayment(t *testing.T) { + t.Parallel() + + const shard = lnwire.MilliSatoshi(100_000_000) + + graph, nodes := atomicTestGraph(t) + source, nodeB, target := nodes[0], nodes[2], nodes[3] + + // The path through B is a dead end: B has next to nothing on its side + // of the channel into the target, so the second shard cannot cross it. + atomicSetBalance(t, graph, 4, nodeB, 1_000) + + routes := []*route.Route{ + atomicTestRoute(t, graph, source, []uint64{1, 2}, shard), + atomicTestRoute(t, graph, source, []uint64{3, 4}, shard), + } + router := &scriptedRouter{routes: routes} + runner := atomicRunner(t, graph, source, router) + + before := balanceSnapshot(graph) + + result, err := runner.RunScenario(&SimScenario{ + Target: target.String(), + AmtMsat: uint64(2 * shard), + MaxParts: 2, + AtomicMpp: true, + }) + require.NoError(t, err) + + // The first shard arrived, the second failed on B's depleted channel, + // and the router then ran out of routes. + require.False(t, result.Success) + require.Len(t, result.Attempts, 2) + require.True(t, result.Attempts[0].Success) + require.False(t, result.Attempts[1].Success) + + // Nothing settled, so nothing moved and nothing was paid for. + require.Equal(t, before, balanceSnapshot(graph), "failed atomic mpp "+ + "left balances moved") + require.Zero(t, result.FeeMsat) + require.EqualValues(t, shard, result.HeldReleasedMsat) + requireNoHolds(t, graph) +} + +// TestSimAtomicMppFlagOff is the regression guard on the historical +// behavior: with the flag off the same payment settles its first shard the +// instant it arrives, so a failure leaves that shard's balances moved and its +// fee paid. +func TestSimAtomicMppFlagOff(t *testing.T) { + t.Parallel() + + const shard = lnwire.MilliSatoshi(100_000_000) + + graph, nodes := atomicTestGraph(t) + source, nodeA, nodeB, target := nodes[0], nodes[1], nodes[2], nodes[3] + + atomicSetBalance(t, graph, 4, nodeB, 1_000) + + first := atomicTestRoute(t, graph, source, []uint64{1, 2}, shard) + routes := []*route.Route{ + first, + atomicTestRoute(t, graph, source, []uint64{3, 4}, shard), + } + router := &scriptedRouter{routes: routes} + runner := atomicRunner(t, graph, source, router) + + var ( + sourceOnC1 = atomicBalance(t, graph, 1, source) + nodeAOnC1 = atomicBalance(t, graph, 1, nodeA) + nodeAOnC2 = atomicBalance(t, graph, 2, nodeA) + targetOnC2 = atomicBalance(t, graph, 2, target) + ) + beforeB := balanceSnapshot(graph) + + result, err := runner.RunScenario(&SimScenario{ + Target: target.String(), + AmtMsat: uint64(2 * shard), + MaxParts: 2, + }) + require.NoError(t, err) + + require.False(t, result.Success) + require.Len(t, result.Attempts, 2) + + // The settled shard moved its route total onto the first channel and + // the shard amount onto the second, leaving A the fee in between. + require.Equal( + t, sourceOnC1-first.TotalAmount, + atomicBalance(t, graph, 1, source), + ) + require.Equal( + t, nodeAOnC1+first.TotalAmount, + atomicBalance(t, graph, 1, nodeA), + ) + require.Equal(t, nodeAOnC2-shard, atomicBalance(t, graph, 2, nodeA)) + require.Equal(t, targetOnC2+shard, atomicBalance(t, graph, 2, target)) + + // The failed shard unwound completely, as it always has. + after := balanceSnapshot(graph) + require.Equal(t, beforeB[3], after[3], "failed shard moved channel 3") + require.Equal(t, beforeB[4], after[4], "failed shard moved channel 4") + + // The fee of the settled shard is charged even though the payment + // failed, and the held-liquidity accounting stays out of the way. + require.EqualValues(t, first.TotalFees(), result.FeeMsat) + require.Zero(t, result.HeldReleasedMsat) + requireNoHolds(t, graph) +} + +// TestSimAtomicMppShardContention asserts that held shards genuinely reserve +// liquidity: two shards over one channel that only covers one of them fail the +// same way a plain liquidity shortfall does today. +func TestSimAtomicMppShardContention(t *testing.T) { + t.Parallel() + + const shard = lnwire.MilliSatoshi(100_000_000) + + // run sends two identical shards down the same path, over a channel + // whose middle hop can fund one and a half of them. + run := func(atomicMpp bool) ([]SimHtlcResult, []SimAttemptTrace) { + graph, nodes := atomicTestGraph(t) + source, nodeA, target := nodes[0], nodes[1], nodes[3] + + atomicSetBalance(t, graph, 2, nodeA, shard+shard/2) + + rt := atomicTestRoute(t, graph, source, []uint64{1, 2}, shard) + router := &scriptedRouter{routes: []*route.Route{rt, rt}} + runner := atomicRunner(t, graph, source, router) + + result, err := runner.RunScenario(&SimScenario{ + Target: target.String(), + AmtMsat: uint64(2 * shard), + MaxParts: 2, + AtomicMpp: atomicMpp, + }) + require.NoError(t, err) + require.False(t, result.Success) + requireNoHolds(t, graph) + + return router.results, result.Attempts + } + + held, heldTraces := run(true) + settled, settledTraces := run(false) + + require.Len(t, held, 2) + require.Len(t, settled, 2) + + // The first shard clears in both modes; the second runs into the same + // shortfall at the same node whether the first one is held or settled. + require.Nil(t, held[0].Failure) + require.Nil(t, settled[0].Failure) + + require.IsType( + t, &lnwire.FailTemporaryChannelFailure{}, held[1].Failure, + "held shard did not reserve the channel", + ) + require.Equal(t, settled[1].FailureSource, held[1].FailureSource) + require.IsType(t, settled[1].Failure, held[1].Failure) + require.Equal(t, settledTraces, heldTraces) +} + +// TestSimAtomicMppSettlesLikeNonAtomic asserts that a payment that does +// complete moves exactly the balances and charges exactly the fees the +// eagerly settling simulator would have. +func TestSimAtomicMppSettlesLikeNonAtomic(t *testing.T) { + t.Parallel() + + const shard = lnwire.MilliSatoshi(100_000_000) + + run := func(atomicMpp bool) (*SimScenarioResult, + map[uint64][2]lnwire.MilliSatoshi) { + + graph, nodes := atomicTestGraph(t) + source, target := nodes[0], nodes[3] + + routes := []*route.Route{ + atomicTestRoute( + t, graph, source, []uint64{1, 2}, shard, + ), + atomicTestRoute( + t, graph, source, []uint64{3, 4}, shard, + ), + } + router := &scriptedRouter{routes: routes} + runner := atomicRunner(t, graph, source, router) + + result, err := runner.RunScenario(&SimScenario{ + Target: target.String(), + AmtMsat: uint64(2 * shard), + MaxParts: 2, + AtomicMpp: atomicMpp, + }) + require.NoError(t, err) + require.True(t, result.Success, "two shard payment failed") + requireNoHolds(t, graph) + + return result, balanceSnapshot(graph) + } + + atomicResult, atomicBalances := run(true) + plainResult, plainBalances := run(false) + + require.Equal(t, plainBalances, atomicBalances, "atomic settlement "+ + "moved different balances") + + // The echoed scenario is the one thing that legitimately differs, so + // normalize the flag away and require everything else to match. + atomicResult.Scenario.AtomicMpp = false + require.Equal(t, plainResult, atomicResult, "atomic settlement "+ + "produced a different result") + require.Positive(t, atomicResult.FeeMsat, "fee assertion is vacuous") +} + +// TestSimAtomicMppDrift asserts that the world keeps turning during an atomic +// payment: background traffic moves hidden liquidity between attempts, it does +// so identically for a given seed, and with the flag off the network stays +// frozen for the duration of the payment as it always has. +func TestSimAtomicMppDrift(t *testing.T) { + t.Parallel() + + const shard = lnwire.MilliSatoshi(1_000_000) + + // run scripts three shards down a path whose second hop is switched + // off, so every attempt fails and the only thing that can move a + // balance is the background traffic. + run := func(atomicMpp bool, + trafficSeed int64) []map[uint64][2]lnwire.MilliSatoshi { + + graph, nodes := atomicTestGraph(t) + source, nodeB, target := nodes[0], nodes[2], nodes[3] + + graph.channels[4].end(nodeB).policy.Disabled = true + + rt := atomicTestRoute(t, graph, source, []uint64{3, 4}, shard) + router := &scriptedRouter{routes: []*route.Route{rt, rt, rt}} + + var snapshots []map[uint64][2]lnwire.MilliSatoshi + router.onReport = func() { + snapshots = append(snapshots, balanceSnapshot(graph)) + } + + runner := atomicRunner(t, graph, source, router) + runner.SetVirtualClock(&SimClockParams{ + StartUnix: 1_800_000_000, + PaymentGapSec: 600, + AttemptSec: 60, + }) + require.NoError(t, runner.SetBackgroundTraffic( + &SimTrafficParams{ + PaymentsPerGap: 100, + MinAmtMsat: 100_000, + MaxAmtMsat: 10_000_000, + Seed: trafficSeed, + }, + )) + + result, err := runner.RunScenario(&SimScenario{ + Target: target.String(), + AmtMsat: uint64(4 * shard), + MaxParts: 4, + AtomicMpp: atomicMpp, + }) + require.NoError(t, err) + require.False(t, result.Success) + require.Len(t, result.Attempts, 3) + requireNoHolds(t, graph) + + return snapshots + } + + // One attempt is a tenth of a gap, so ten of the hundred background + // payments of a gap land in each attempt's window. + drifting := run(true, 21) + require.Len(t, drifting, 3) + require.NotEqual( + t, drifting[0], drifting[2], + "liquidity did not drift during the payment", + ) + + // The exogenous process is still a function of its seed alone. + require.Equal(t, drifting, run(true, 21), "same seed diverged") + require.NotEqual(t, drifting, run(true, 22), "different seeds agreed") + + // With the flag off the world freezes for the duration of a payment, + // exactly as it did before. + frozen := run(false, 21) + require.Len(t, frozen, 3) + require.Equal(t, frozen[0], frozen[2], "liquidity drifted with atomic "+ + "mpp off") +} diff --git a/routing/sim_graph.go b/routing/sim_graph.go index b5cc0389b..b555740e0 100644 --- a/routing/sim_graph.go +++ b/routing/sim_graph.go @@ -54,6 +54,17 @@ type simChannelEnd struct { owner route.Vertex balance lnwire.MilliSatoshi policy SimPolicy + + // held is the part of the balance that in-flight htlcs have reserved + // but not settled yet. It is always at most the balance, and it is + // zero unless a payment is running with hold semantics. + held lnwire.MilliSatoshi +} + +// available returns the liquidity this end can put behind a new htlc: its +// balance less whatever the htlcs already in flight over it hold. +func (e *simChannelEnd) available() lnwire.MilliSatoshi { + return e.balance - e.held } // SimChannel is a single channel in the simulated network. The two ends are @@ -115,6 +126,15 @@ type SimNode struct { type SimGraph struct { nodes map[route.Vertex]*SimNode channels map[uint64]*SimChannel + + // holds tracks the liquidity reservations of the htlcs that have + // traversed the network but are not settled yet, keyed by hold id. + // It is empty unless a payment is running with hold semantics. + holds map[uint64][]balanceMove + + // nextHoldID hands out hold ids; zero is never used so that it can + // stand for "no hold". + nextHoldID uint64 } // NewSimGraph instantiates an empty simulated network. @@ -122,6 +142,7 @@ func NewSimGraph() *SimGraph { return &SimGraph{ nodes: make(map[route.Vertex]*SimNode), channels: make(map[uint64]*SimChannel), + holds: make(map[uint64][]balanceMove), } } @@ -234,7 +255,7 @@ func (g *SimGraph) LocalBalances( } for _, channel := range node.channels { - balances[channel.ID] = channel.end(pubKey).balance + balances[channel.ID] = channel.end(pubKey).available() } return balances @@ -318,26 +339,146 @@ type SimHtlcResult struct { Failure lnwire.FailureMessage } -// balanceMove records a single applied balance mutation so that it can be -// unwound when a downstream hop fails. +// balanceMove records a single hop's liquidity commitment so that it can be +// unwound when a downstream hop fails, and so that a held htlc can later be +// settled or released as a unit. type balanceMove struct { from *simChannelEnd to *simChannelEnd amt lnwire.MilliSatoshi } +// apply moves the amount across the channel: the settlement of one hop. +func (m *balanceMove) apply() { + m.from.balance -= m.amt + m.to.balance += m.amt +} + +// unapply undoes apply. +func (m *balanceMove) unapply() { + m.from.balance += m.amt + m.to.balance -= m.amt +} + +// reserve holds the amount on the sending end, taking it out of the +// liquidity available to every other htlc without moving it yet. +func (m *balanceMove) reserve() { + m.from.held += m.amt +} + +// unreserve gives back the reservation made by reserve. +func (m *balanceMove) unreserve() { + m.from.held -= m.amt +} + +// simCommitMode selects what a route walk does with the liquidity it +// traverses. +type simCommitMode uint8 + +const ( + // simCommitSettle moves the balance of every hop as the htlc passes + // it, the instantly settling htlc the simulator has always used. + simCommitSettle simCommitMode = iota + + // simCommitHold only reserves the outgoing liquidity of every hop, + // leaving the balances untouched until the resulting hold is settled + // or released. This is the htlc a receiver sits on while it waits for + // the rest of an mpp set to arrive. + simCommitHold +) + // SendHtlc sends an htlc along the given route through the simulated // network and synchronously returns its resolution. Forwarding applies the // same policy checks a real node would: disabled channels, min/max htlc // limits, fee sufficiency, cltv deltas and (hidden) liquidity. func (g *SimGraph) SendHtlc(rt *route.Route) (SimHtlcResult, error) { + result, _, err := g.walkHtlc(rt, simCommitSettle) + + return result, err +} + +// HoldHtlc sends an htlc along the given route but stops short of settling +// it: each hop reserves its outgoing liquidity instead of moving it, so +// sibling shards and background traffic see the reduced availability while +// the htlc is in flight. A settled resolution returns a non-zero hold id that +// must eventually be passed to either SettleHold or ReleaseHold. A failure +// leaves nothing reserved and returns a zero hold id. +func (g *SimGraph) HoldHtlc(rt *route.Route) (SimHtlcResult, uint64, error) { + result, moves, err := g.walkHtlc(rt, simCommitHold) + if err != nil || result.Failure != nil { + return result, 0, err + } + + g.nextHoldID++ + id := g.nextHoldID + g.holds[id] = moves + + return result, id, nil +} + +// SettleHold turns a hold into real balance movement: every reservation +// becomes the transfer the settling htlc would have made all along, which +// pays each forwarding node the difference between what it received and what +// it sent on. Settling an unknown hold is a no-op. +func (g *SimGraph) SettleHold(id uint64) { + moves, ok := g.holds[id] + if !ok { + return + } + delete(g.holds, id) + + for i := range moves { + moves[i].unreserve() + moves[i].apply() + } +} + +// ReleaseHold cancels a hold: the reserved liquidity becomes available again +// and no balance moves at all, so an htlc that is never settled leaves the +// network exactly as it found it. Releasing an unknown hold is a no-op. +func (g *SimGraph) ReleaseHold(id uint64) { + moves, ok := g.holds[id] + if !ok { + return + } + delete(g.holds, id) + + for i := range moves { + moves[i].unreserve() + } +} + +// walkHtlc walks an htlc along the given route, applying the same policy and +// liquidity checks a real forwarding node would. The commit mode decides what +// happens to the liquidity of each hop it clears: simCommitSettle moves it, +// simCommitHold merely reserves it and hands the reservations back so that +// the caller can settle or release them later. Either way a failure part of +// the way down the route unwinds everything committed so far, so the graph is +// never left holding a half-forwarded htlc. +func (g *SimGraph) walkHtlc(rt *route.Route, + mode simCommitMode) (SimHtlcResult, []balanceMove, error) { + var moves []balanceMove - // revert unwinds all balance mutations applied so far. + // commit applies one hop's liquidity commitment in the current mode. + commit := func(m *balanceMove) { + if mode == simCommitHold { + m.reserve() + return + } + + m.apply() + } + + // revert unwinds all liquidity commitments applied so far. revert := func() { - for _, m := range moves { - m.from.balance += m.amt - m.to.balance -= m.amt + for i := range moves { + if mode == simCommitHold { + moves[i].unreserve() + continue + } + + moves[i].unapply() } } @@ -351,8 +492,8 @@ func (g *SimGraph) SendHtlc(rt *route.Route) (SimHtlcResult, error) { channel, ok := g.channels[routeHop.ChannelID] if !ok { revert() - return SimHtlcResult{}, fmt.Errorf("unknown channel "+ - "%v in route", routeHop.ChannelID) + return SimHtlcResult{}, nil, fmt.Errorf("unknown "+ + "channel %v in route", routeHop.ChannelID) } sendingEnd := channel.end(prevNode) @@ -361,8 +502,8 @@ func (g *SimGraph) SendHtlc(rt *route.Route) (SimHtlcResult, error) { receivingEnd.owner != routeHop.PubKeyBytes { revert() - return SimHtlcResult{}, fmt.Errorf("channel %v does "+ - "not connect %v to %v", routeHop.ChannelID, + return SimHtlcResult{}, nil, fmt.Errorf("channel %v "+ + "does not connect %v to %v", routeHop.ChannelID, prevNode, routeHop.PubKeyBytes) } @@ -389,31 +530,34 @@ func (g *SimGraph) SendHtlc(rt *route.Route) (SimHtlcResult, error) { return SimHtlcResult{ FailureSource: prevNode, Failure: failure, - }, nil + }, nil, nil } } // Liquidity check: the sending end must have the outgoing - // amount available. This is the hidden state that path - // finding is trying to predict. - if sendingEnd.balance < amtOut { + // amount available. Liquidity that an in-flight htlc already + // holds does not count, so sibling shards and background + // payments contend for the same balance. This is the hidden + // state that path finding is trying to predict. + if sendingEnd.available() < amtOut { revert() return SimHtlcResult{ FailureSource: prevNode, Failure: lnwire.NewTemporaryChannelFailure( nil, ), - }, nil + }, nil, nil } - // Move the balance and record the move for potential unwind. - sendingEnd.balance -= amtOut - receivingEnd.balance += amtOut - moves = append(moves, balanceMove{ + // Commit the hop's liquidity and record it so that it can be + // unwound, settled or released later. + move := balanceMove{ from: sendingEnd, to: receivingEnd, amt: amtOut, - }) + } + commit(&move) + moves = append(moves, move) // Advance to the next hop. amtIn = amtOut @@ -421,8 +565,9 @@ func (g *SimGraph) SendHtlc(rt *route.Route) (SimHtlcResult, error) { prevNode = routeHop.PubKeyBytes } - // All hops succeeded, the htlc is settled at the final node. - return SimHtlcResult{}, nil + // All hops succeeded: the htlc has reached the final node, either + // settled outright or held there pending its siblings. + return SimHtlcResult{}, moves, nil } // checkPolicy applies the forwarding policy checks of a node to an htlc that diff --git a/routing/sim_run.go b/routing/sim_run.go index fd7c4606c..4644cf9b2 100644 --- a/routing/sim_run.go +++ b/routing/sim_run.go @@ -138,6 +138,20 @@ type SimScenario struct { // MaxParts caps the number of MPP shards. 1 disables splitting. MaxParts uint32 `json:"max_parts"` + + // AtomicMpp switches the payment onto hold-and-release shard + // semantics: a shard that reaches the destination reserves the + // liquidity of every hop it crossed instead of settling it, and the + // whole set only moves balances once the full amount has arrived. A + // payment that never completes releases everything it held, so a + // failed mpp is atomic and costs no fees. It also makes sequential + // probing expensive rather than free: the shards a router leaves in + // flight while it probes contend with its own siblings and with + // background traffic, and time keeps passing between attempts. + // + // With the flag off the simulator keeps its historical behavior, in + // which every shard settles the instant it arrives. + AtomicMpp bool `json:"atomic_mpp,omitempty"` } // SimHopTrace records one hop of an attempted route. @@ -167,6 +181,12 @@ type SimScenarioResult struct { // FeeMsat is the total fee paid over all settled htlcs. FeeMsat uint64 `json:"fee_msat"` + // HeldReleasedMsat is how much of the payment had already reached the + // destination and was rolled back when the payment failed. It is only + // ever non-zero under atomic mpp, where it measures the liquidity a + // router tied up on the way to failing. + HeldReleasedMsat uint64 `json:"held_released_msat,omitempty"` + // Error records a terminal payment error, e.g. no path found. Error string `json:"error,omitempty"` } @@ -220,6 +240,12 @@ type SimRunner struct { // traffic is the background traffic engine, nil when disabled. traffic *simTraffic + // 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 + // instead of rounding it away. + trafficCarry float64 + cleanup func() } @@ -361,8 +387,12 @@ func (r *SimRunner) advanceGap() { } } -// advanceAttempt moves virtual time forward by one attempt's duration. -func (r *SimRunner) advanceAttempt() { +// 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) { if r.virtualClk == nil || r.clockParams.AttemptSec <= 0 { return } @@ -371,6 +401,31 @@ func (r *SimRunner) advanceAttempt() { time.Duration(r.clockParams.AttemptSec * float64(time.Second)), )) + + if !atomicMpp || r.traffic == nil { + return + } + + r.traffic.runN(r.attemptTrafficPayments()) +} + +// attemptTrafficPayments returns how many background payments belong to the +// virtual time one attempt consumes. The per-gap volume is pro-rated by the +// attempt duration so that the exogenous process runs at one rate throughout, +// and the fractional remainder carries into the next attempt rather than +// rounding away. +func (r *SimRunner) attemptTrafficPayments() int { + if r.clockParams.PaymentGapSec <= 0 { + return 0 + } + + r.trafficCarry += float64(r.traffic.params.PaymentsPerGap) * + r.clockParams.AttemptSec / r.clockParams.PaymentGapSec + + n := int(r.trafficCarry) + r.trafficCarry -= float64(n) + + return n } // Close releases the runner's resources. @@ -454,8 +509,33 @@ func (r *SimRunner) RunScenario(s *SimScenario) (*SimScenarioResult, error) { 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) @@ -474,9 +554,20 @@ func (r *SimRunner) RunScenario(s *SimScenario) (*SimScenarioResult, error) { // Each attempt consumes virtual time: htlcs take real seconds // to resolve on a live network. - r.advanceAttempt() + r.advanceAttempt(s.AtomicMpp) - htlcResult, err := r.graph.SendHtlc(rt) + // 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 @@ -486,7 +577,9 @@ func (r *SimRunner) RunScenario(s *SimScenario) (*SimScenarioResult, error) { result.Attempts, traceAttempt(rt, htlcResult), ) - // Let the router learn from the outcome. + // 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. err = router.ReportAttempt(attemptID, rt, htlcResult) if err != nil { return nil, err @@ -497,7 +590,16 @@ func (r *SimRunner) RunScenario(s *SimScenario) (*SimScenarioResult, error) { } inFlightHtlcs++ - result.FeeMsat += uint64(rt.TotalFees()) + + // 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. @@ -509,6 +611,16 @@ func (r *SimRunner) RunScenario(s *SimScenario) (*SimScenarioResult, error) { 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 } diff --git a/routing/sim_traffic.go b/routing/sim_traffic.go index 5189b20d5..d25b56b61 100644 --- a/routing/sim_traffic.go +++ b/routing/sim_traffic.go @@ -97,7 +97,14 @@ func newSimTraffic(graph *SimGraph, params *SimTrafficParams) (*simTraffic, // run executes one gap's worth of background payments. func (t *simTraffic) run() { - for i := 0; i < t.params.PaymentsPerGap; i++ { + t.runN(t.params.PaymentsPerGap) +} + +// runN executes the given number of background payments, the slice of the +// exogenous process that belongs to some stretch of virtual time shorter than +// a full gap. +func (t *simTraffic) runN(n int) { + for i := 0; i < n; i++ { t.sendOne() } }