EarningsTracker: attribute rebalances from the move response, not a requester-keyed map

Completed rebalances were correlated back to their (source,
destination) through an in-memory map keyed only by the
RequestMoveFunds requester pointer. When one rebalancer issues several
moves at once they share a single requester (EarningsRebalancer uses
its module pointer; JitRebalancer uses nullptr), so each request
overwrote the previous map entry. As the moves completed, the first
response consumed and erased the shared entry -- booking against
whatever pair happened to be written last -- and every later response,
including the actually-successful one, hit the not-in-our-table path
and was silently dropped.

Effect: rebalance spend and volume were under-counted and sometimes
mis-attributed to the wrong peer. That also starved the
EarningsRebalancer refusing-to-throw-good-money-after-bad guard of
accurate expenditure data. JIT-dominated periods looked fine only
because those moves complete one at a time and rarely overlap; an
EarningsRebalancer-dominated period booked almost nothing.

Fix: ResponseMoveFunds now carries source and destination
directly. FundsMover/Runner populates them and EarningsTracker books
the move straight from the response. The requester-keyed pendings map,
the RequestMoveFunds subscription, and request_move_funds are
removed. This mirrors the existing XRebalanceAttribution path, which
already passes source/destination in its message.

Tests: ResponseMoveFunds constructions updated across the rebalancer
and earnings tests; test_earningstracker now issues two moves under
the same requester and asserts both are booked and correctly
attributed -- a regression guard for the collision. Full suite 85/85.

Boss/Mod/EarningsTracker.cpp

@@ -8,7 +8,6 @@
 #include"Boss/Msg/Manifestation.hpp"
 #include"Boss/Msg/ProvideStatus.hpp"
 #include"Boss/Msg/RequestEarningsInfo.hpp"
-#include"Boss/Msg/RequestMoveFunds.hpp"
 #include"Boss/Msg/ResponseEarningsInfo.hpp"
 #include"Boss/Msg/ResponseMoveFunds.hpp"
 #include"Boss/Msg/SolicitStatus.hpp"
@@ -21,7 +20,6 @@
 #include"Util/make_unique.hpp"
 
 #include<cmath>
-#include<map>
 #include<iostream>
 #include<unordered_set>
 
@@ -88,14 +86,6 @@ private:
 	std::function<double()> get_now;
 	Sqlite3::Db db;
 
-	/* Information of a pending MoveFunds.  */
-	struct Pending {
-		Ln::NodeId source;
-		Ln::NodeId destination;
-	};
-	/* Maps a requester to the source-destination of the movefunds.  */
-	std::map<void*, Pending> pendings;
-
 	void start() {
 		bus.subscribe<Msg::DbResource
 			     >([this](Msg::DbResource const& r) {
@@ -106,10 +96,6 @@ private:
 			     >([this](Msg::ForwardFee const& f) {
 				     return forward_fee(f.in_id, f.out_id, f.fee, f.amount);
 		});
-		bus.subscribe<Msg::RequestMoveFunds
-			     >([this](Msg::RequestMoveFunds const& req) {
-			return request_move_funds(req);
-		});
 		bus.subscribe<Msg::ResponseMoveFunds
 			     >([this](Msg::ResponseMoveFunds const& rsp) {
 			return response_move_funds(rsp);
@@ -406,28 +392,25 @@ private:
 		});
 	}
 	Ev::Io<void>
-	request_move_funds(Boss::Msg::RequestMoveFunds const& req) {
-		auto& pending = pendings[req.requester];
-		pending.source = req.source;
-		pending.destination = req.destination;
-		return Ev::lift();
-	}
-	Ev::Io<void>
 	response_move_funds(Boss::Msg::ResponseMoveFunds const& rsp) {
-		auto requester = rsp.requester;
+		/* The response carries its own source/destination, so we
+		 * attribute the move directly rather than correlating back
+		 * to the request by `requester`.  A `requester`-keyed map
+		 * collides when one rebalancer issues several moves at once
+		 * (they share the requester pointer): the first response to
+		 * arrive consumed the shared entry and every later
+		 * response -- including the actually-successful one -- was
+		 * silently dropped.  */
+		auto source = rsp.source;
+		auto destination = rsp.destination;
 		auto fee = rsp.fee_spent;
 		auto amount = rsp.amount_moved;
-		return db.transact().then([this, requester, fee, amount
+		return db.transact().then([ this, source, destination
+					  , fee, amount
 					  ](Sqlite3::Tx tx) {
-			auto it = pendings.find(requester);
-			if (it == pendings.end())
-				/* Not in our table, huh, weird.  */
-				return Ev::lift();
-			auto& pending = it->second;
-
 			auto bucket = bucket_time(get_now());
-			ensure(tx, pending.source, bucket);
-			ensure(tx, pending.destination, bucket);
+			ensure(tx, source, bucket);
+			ensure(tx, destination, bucket);
 
 			/* Source gets in-expenditures since it gets more
 			 * incoming capacity (for more earnings for the
@@ -440,7 +423,7 @@ private:
 			   AND time_bucket = :bucket
 			     ;
 			)QRY")
-				.bind(":node", std::string(pending.source))
+				.bind(":node", std::string(source))
 				.bind(":bucket", bucket)
 				.bind(":fee", fee.to_msat())
 				.bind(":amount", amount.to_msat())
@@ -458,7 +441,7 @@ private:
 			     ;
 			)QRY")
 				.bind( ":node"
-				     , std::string(pending.destination)
+				     , std::string(destination)
 				     )
 				.bind(":bucket", bucket)
 				.bind(":fee", fee.to_msat())
@@ -466,9 +449,6 @@ private:
 				.execute()
 				;
 
-			/* Erase it.  */
-			pendings.erase(it);
-
 			tx.commit();
 			return Ev::lift();
 		});

Boss/Mod/FundsMover/Runner.cpp

@@ -288,7 +288,8 @@ Ev::Io<void> Runner::finish() {
 				);
 	}).then([this]() {
 		return bus.raise(Msg::ResponseMoveFunds{
-			requester, transferred, orig_budget - *fee_budget
+			requester, transferred, orig_budget - *fee_budget,
+			source, destination
 		});
 	});
 }

Boss/Msg/ResponseMoveFunds.hpp

@@ -2,6 +2,7 @@
 #define BOSS_MSG_RESPONSEMOVEFUNDS_HPP
 
 #include"Ln/Amount.hpp"
+#include"Ln/NodeId.hpp"
 
 namespace Boss { namespace Msg {
 
@@ -17,6 +18,12 @@ struct ResponseMoveFunds {
 	Ln::Amount amount_moved;
 	/* Actual amount spent on fees.  */
 	Ln::Amount fee_spent;
+
+	/* The peers this move was between.  Carried in the response so
+	 * consumers (e.g. earnings accounting) can attribute the move
+	 * without correlating back to the request by `requester`.  */
+	Ln::NodeId source;
+	Ln::NodeId destination;
 };
 
 }}

tests/boss/test_earningshistory.cpp

@@ -63,9 +63,11 @@ Ev::Io<void> raiseMoveFundsLoop(S::Bus& bus, int count) {
 		.then([&bus]() {
 			return bus.raise(
 				Boss::Msg::ResponseMoveFunds{
-					nullptr,        	// requester (see RequestMoveFunds)
+					nullptr,        	// requester (unused for attribution)
 					Ln::Amount::sat(1000),  // amount_moved
-					Ln::Amount::sat(1)      // fee_spent
+					Ln::Amount::sat(1),     // fee_spent
+					C,              	// source
+					A               	// destination
 				});
 		})
 		.then([&bus, count]() {

tests/boss/test_earningstracker.cpp

@@ -5,7 +5,6 @@
 #include"Boss/Msg/ForwardFee.hpp"
 #include"Boss/Msg/ProvideStatus.hpp"
 #include"Boss/Msg/RequestEarningsInfo.hpp"
-#include"Boss/Msg/RequestMoveFunds.hpp"
 #include"Boss/Msg/ResponseEarningsInfo.hpp"
 #include"Boss/Msg/ResponseMoveFunds.hpp"
 #include"Boss/Msg/SolicitStatus.hpp"
@@ -167,22 +166,30 @@ int main() {
                         )JSON"));
 		return Ev::lift();
 	}).then([&]() {
+		/* Two rebalance moves reported under the SAME requester.
+		 * The response carries its own source/destination, so both
+		 * are booked and attributed correctly.  The tracker used to
+		 * correlate moves back to requests by `requester`, so every
+		 * move after the first of a same-requester batch was
+		 * silently dropped -- regression guard for that bug.  */
 		mock_now = 4000.0;
 		return bus.raise(
-			Boss::Msg::RequestMoveFunds{
-				NULL,			// requester (match ResponseMoveFunds)
+			Boss::Msg::ResponseMoveFunds{
+				NULL,			// requester (shared)
+				Ln::Amount::sat(1000),	// amount_moved
+				Ln::Amount::sat(2),	// fee_spent
 				C,			// source
-				A,			// destination
-				Ln::Amount::sat(1000),	// amount
-				Ln::Amount::sat(3)	// fee_budget
+				A			// destination
 			});
 	}).then([&]() {
 		mock_now = 5000.0;
 		return bus.raise(
 			Boss::Msg::ResponseMoveFunds{
-				NULL,			// requester (match RequestMoveFunds)
+				NULL,			// requester (same as above)
 				Ln::Amount::sat(1000),	// amount_moved
-				Ln::Amount::sat(2)	// fee_spent
+				Ln::Amount::sat(2),	// fee_spent
+				C,			// source
+				B			// destination
 			});
 	}).then([&]() {
 		return bus.raise(Boss::Msg::SolicitStatus{});
@@ -207,31 +214,31 @@ int main() {
 		    "in_earnings": 0,
 		    "in_expenditures": 0,
 		    "out_earnings": 2000,
-		    "out_expenditures": 0,
+		    "out_expenditures": 2000,
 		    "in_forwarded": 0,
 		    "in_rebalanced": 0,
 		    "out_forwarded": 3000000,
-		    "out_rebalanced": 0
+		    "out_rebalanced": 1000000
 		  },
 		  "020000000000000000000000000000000000000000000000000000000000000003": {
 		    "in_earnings": 0,
-		    "in_expenditures": 2000,
+		    "in_expenditures": 4000,
 		    "out_earnings": 0,
 		    "out_expenditures": 0,
 		    "in_forwarded": 0,
-		    "in_rebalanced": 1000000,
+		    "in_rebalanced": 2000000,
 		    "out_forwarded": 0,
 		    "out_rebalanced": 0
 		  },
 		  "total": {
 		    "in_earnings": 2000,
-		    "in_expenditures": 2000,
+		    "in_expenditures": 4000,
 		    "out_earnings": 2000,
-		    "out_expenditures": 2000,
+		    "out_expenditures": 4000,
 		    "in_forwarded": 3000000,
-		    "in_rebalanced": 1000000,
+		    "in_rebalanced": 2000000,
 		    "out_forwarded": 3000000,
-		    "out_rebalanced": 1000000
+		    "out_rebalanced": 2000000
 		  }
 		}
                         )JSON"));

tests/boss/test_initialrebalancer.cpp

@@ -142,10 +142,14 @@ int main() {
 
 	auto num_move_funds = std::size_t(0);
 	void* last_requester = nullptr;
+	auto last_source = Ln::NodeId();
+	auto last_destination = Ln::NodeId();
 	bus.subscribe< RequestMoveFunds
 		     >([&](RequestMoveFunds const& m) {
 		++num_move_funds;
 		last_requester = m.requester;
+		last_source = m.source;
+		last_destination = m.destination;
 		return Ev::lift();
 	});
 
@@ -174,7 +178,9 @@ int main() {
 		return bus.raise(ResponseMoveFunds{
 			last_requester,
 			Ln::Amount::sat(0),
-			Ln::Amount::sat(0)
+			Ln::Amount::sat(0),
+			last_source,
+			last_destination
 		});
 	}).then([&]() {
 

tests/boss/test_jitrebalancer.cpp

@@ -395,7 +395,9 @@ int main() {
 		return bus.raise(ResponseMoveFunds{
 			requester,
 			Ln::Amount::sat(0),
-			Ln::Amount::sat(0)
+			Ln::Amount::sat(0),
+			source,
+			destination
 		});
 	}).then([&]() {
 		/* We should then release.  */

tests/boss/test_recentearnings.cpp

@@ -63,9 +63,11 @@ Ev::Io<void> raiseMoveFundsLoop(S::Bus& bus, int count) {
 		.then([&bus]() {
 			return bus.raise(
 				Boss::Msg::ResponseMoveFunds{
-					nullptr,        	// requester (see RequestMoveFunds)
+					nullptr,        	// requester (unused for attribution)
 					Ln::Amount::sat(1000),  // amount_moved
-					Ln::Amount::sat(1)      // fee_spent
+					Ln::Amount::sat(1),     // fee_spent
+					C,              	// source
+					B               	// destination
 				});
 		})
 		.then([&bus, count]() {