JitRebalancer: skip rebalance if one is already in flight for the destination

The fee budget check reads out_expenditures, which is only
persisted once a rebalance completes (up to 120 s per run).  Each
incoming HTLC spawned an independent rebalance run with no guard,
so concurrent HTLCs to the same underfunded channel each passed
the check against the same stale value, multiplying the intended
25%-of-earnings aggregate cap by the number of concurrent triggers.

Skip HTLC-triggered rebalances for a node that already has one in
flight, mirroring the working guard in EarningsRebalancer.  Skipped
HTLCs are released immediately and proceed without JIT rebalancing;
a retry after the in-flight run completes sees both the refilled
channel and the updated budget.

Update the parallel-calls unit test to the new semantics: exactly
one of three concurrent calls is let in, only it requests a
rebalance, and the guard clears once the run completes.

Reported by an external security researcher via private disclosure.

Fixes #323

Boss/Mod/JitRebalancer.cpp

@@ -31,6 +31,7 @@
 #include"Util/make_unique.hpp"
 #include"Util/stringify.hpp"
 #include<map>
+#include<set>
 
 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<Ln::NodeId> 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();
 		});
 	}
 

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<std::uint64_t>{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<std::uint64_t>{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<bool> 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<bool> 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);