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

@@ -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")
+}

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

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
 		}

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()
 	}
 }