diff --git a/Boss/Mod/JitRebalancer.cpp b/Boss/Mod/JitRebalancer.cpp index d204afe..47654fd 100644 --- a/Boss/Mod/JitRebalancer.cpp +++ b/Boss/Mod/JitRebalancer.cpp @@ -31,6 +31,7 @@ #include"Util/make_unique.hpp" #include"Util/stringify.hpp" #include +#include namespace { @@ -110,6 +111,14 @@ private: ModG::RebalanceUnmanagerProxy unmanager; std::uint32_t max_rebalance_fee_ppm; + /* Nodes with a JIT rebalance currently in flight. + * The budget check reads expenditures that are only + * persisted once a rebalance completes, so concurrent + * rebalances to the same destination would each + * authorize against the same stale budget. + */ + std::set in_flight; + void start() { max_rebalance_fee_ppm = default_max_rebalance_fee_ppm; @@ -186,6 +195,18 @@ private: return Ev::lift(false); }); } + if (in_flight.count(node) != 0) { + return Boss::log( bus, Debug + , "JitRebalancer: HTLC %s to " + "%s: rebalance already in " + "flight, will ignore." + , stringify_cid(id).c_str() + , Util::stringify(node).c_str() + ).then([]() { + return Ev::lift(false); + }); + } + in_flight.insert(node); return Boss::concurrent( check_and_move(node, amount, id) ).then([]() { return Ev::lift(true); @@ -228,6 +249,9 @@ private: , unmanager, max_rebalance_fee_ppm ); return r.execute(); + }).then([this, node]() { + in_flight.erase(node); + return Ev::lift(); }); } diff --git a/tests/boss/test_jitrebalancer.cpp b/tests/boss/test_jitrebalancer.cpp index afa0e8d..c7af59f 100644 --- a/tests/boss/test_jitrebalancer.cpp +++ b/tests/boss/test_jitrebalancer.cpp @@ -14,8 +14,6 @@ #include"Boss/Msg/ResponseRpcCommand.hpp" #include"Boss/Msg/SolicitHtlcAcceptedDeferrer.hpp" #include"Ev/Io.hpp" -#include"Ev/concurrent.hpp" -#include"Ev/foreach.hpp" #include"Ev/map.hpp" #include"Ev/now.hpp" #include"Ev/start.hpp" @@ -311,6 +309,11 @@ int main() { void* requester = nullptr; auto source = Ln::NodeId(); auto destination = Ln::NodeId(); + /* Parallel-call check: the calls, and which one was + * let in. + */ + auto ids = std::vector{3, 4, 5}; + auto deferred_id = std::uint64_t(0); bus.subscribe< RequestMoveFunds >([&](RequestMoveFunds const& m) { ++num_move_funds; @@ -355,27 +358,51 @@ int main() { }).then([&]() { assert(num_move_funds == 0); - /* Check parallel calls. */ - auto ids = std::vector{3, 4, 5}; - auto act = Ev::lift(); - /* Perform parallel calls. */ - act += Ev::concurrent(Ev::map([&](std::uint64_t id) { - return deferrer(htlc("1000x1x0", Ln::Amount::msat(1), id)); - }, ids).then([&](std::vector flags) { - /* Every forward should get in. */ - for (auto flag : flags) - assert(flag); - return Ev::lift(); - })); - act += Ev::yield(); - act += Ev::foreach([&](std::uint64_t id) { - return release_monitor.wait_release(id); + /* Check parallel calls to the same underfunded + * node: exactly one is let in and requests the + * rebalance; the rest are skipped because a + * run for the node is already in flight. + */ + return Ev::map([&](std::uint64_t id) { + return deferrer(htlc("1000x1x1", Ln::Amount::msat(90000), id)); }, ids); - return act; + }).then([&](std::vector flags) { + auto num_in = std::size_t(0); + for (auto i = std::size_t(0); i < flags.size(); ++i) { + if (flags[i]) { + ++num_in; + deferred_id = ids[i]; + } + } + assert(num_in == 1); + /* Wait for the let-in run to reach its + * move-funds request. + */ + return multiyield(); + }).then([&]() { + /* Only the let-in run requests a rebalance. */ + assert(num_move_funds == 1); + /* The 02 would not have fit. */ + assert(source == Ln::NodeId("020000000000000000000000000000000000000000000000000000000000000000")); + assert(destination == Ln::NodeId("020000000000000000000000000000000000000000000000000000000000000001")); + /* Let the in-flight run finish. */ + return bus.raise(ResponseMoveFunds{ + requester, + Ln::Amount::sat(0), + Ln::Amount::sat(0) + }); + }).then([&]() { + return release_monitor.wait_release(deferred_id); + }).then([&]() { + /* Let the finished run clean up. */ + return multiyield(); }).then([&]() { - assert(num_move_funds == 0); - /* Check for a forward that does not fit. */ + /* The guard clears once the run completes: + * a new forward that does not fit gets in + * again. + */ + num_move_funds = 0; return deferrer(htlc("1000x1x1", Ln::Amount::msat(90000), 6)); }).then([&](bool flag) { assert(flag == true);