From cbc661457be4594cb70157550f52faee30898d88 Mon Sep 17 00:00:00 2001 From: Ken Sedgwick Date: Fri, 19 Jun 2026 09:21:07 -0700 Subject: [PATCH] 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 | 52 +++++++++------------------ Boss/Mod/FundsMover/Runner.cpp | 3 +- Boss/Msg/ResponseMoveFunds.hpp | 7 ++++ tests/boss/test_earningshistory.cpp | 6 ++-- tests/boss/test_earningstracker.cpp | 39 +++++++++++--------- tests/boss/test_initialrebalancer.cpp | 8 ++++- tests/boss/test_jitrebalancer.cpp | 4 ++- tests/boss/test_recentearnings.cpp | 6 ++-- 8 files changed, 66 insertions(+), 59 deletions(-) diff --git a/Boss/Mod/EarningsTracker.cpp b/Boss/Mod/EarningsTracker.cpp index b15e269..8d28bc5 100644 --- a/Boss/Mod/EarningsTracker.cpp +++ b/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 -#include #include #include @@ -88,14 +86,6 @@ private: std::function 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 pendings; - void start() { bus.subscribe([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([this](Msg::RequestMoveFunds const& req) { - return request_move_funds(req); - }); bus.subscribe([this](Msg::ResponseMoveFunds const& rsp) { return response_move_funds(rsp); @@ -406,28 +392,25 @@ private: }); } Ev::Io - 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 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(); }); diff --git a/Boss/Mod/FundsMover/Runner.cpp b/Boss/Mod/FundsMover/Runner.cpp index 6b7e608..041eb54 100644 --- a/Boss/Mod/FundsMover/Runner.cpp +++ b/Boss/Mod/FundsMover/Runner.cpp @@ -288,7 +288,8 @@ Ev::Io Runner::finish() { ); }).then([this]() { return bus.raise(Msg::ResponseMoveFunds{ - requester, transferred, orig_budget - *fee_budget + requester, transferred, orig_budget - *fee_budget, + source, destination }); }); } diff --git a/Boss/Msg/ResponseMoveFunds.hpp b/Boss/Msg/ResponseMoveFunds.hpp index d068517..1e3c7f9 100644 --- a/Boss/Msg/ResponseMoveFunds.hpp +++ b/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; }; }} diff --git a/tests/boss/test_earningshistory.cpp b/tests/boss/test_earningshistory.cpp index 5b8b40f..2ab7252 100644 --- a/tests/boss/test_earningshistory.cpp +++ b/tests/boss/test_earningshistory.cpp @@ -63,9 +63,11 @@ Ev::Io 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]() { diff --git a/tests/boss/test_earningstracker.cpp b/tests/boss/test_earningstracker.cpp index d0f29c0..b561f82 100644 --- a/tests/boss/test_earningstracker.cpp +++ b/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")); diff --git a/tests/boss/test_initialrebalancer.cpp b/tests/boss/test_initialrebalancer.cpp index 88aed63..1d4d011 100644 --- a/tests/boss/test_initialrebalancer.cpp +++ b/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([&]() { diff --git a/tests/boss/test_jitrebalancer.cpp b/tests/boss/test_jitrebalancer.cpp index 4ebd77d..c1ef1fa 100644 --- a/tests/boss/test_jitrebalancer.cpp +++ b/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. */ diff --git a/tests/boss/test_recentearnings.cpp b/tests/boss/test_recentearnings.cpp index e2be9cf..24f5ffd 100644 --- a/tests/boss/test_recentearnings.cpp +++ b/tests/boss/test_recentearnings.cpp @@ -63,9 +63,11 @@ Ev::Io 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]() {