mirror of
https://github.com/ZmnSCPxj/clboss.git
synced 2026-08-17 13:07:24 +02:00
f22a02e98a
A third clboss-rebalance-mode value, xrebalance2, runs the same XRebalancer planner but executes cycles through the external xrebalance plugin's RPC instead of the in-clboss clboss-xmovefunds executor. The plugin does the layer splitting on stock askrene and owns constraint knowledge and failure feedback, so the in-clboss layer machinery (including the predictor) stays idle in this mode. The new XRebalancePartMonitor subscribes to the plugin's xrebalance_part notifications and raises Msg::XRebalanceAttribution for each completed part, so EarningsTracker accounts plugin-moved funds regardless of which client initiated the transfer. Attribution is notification-only: a part that reaches terminal state while clboss is down goes unaccounted. Subscribing to the topic is safe without the plugin loaded (lightningd only warns about unknown notification topics), and a cycle fired with the plugin missing logs one line and retries next cycle.
128 lines
3.4 KiB
C++
128 lines
3.4 KiB
C++
#include"Boss/Mod/XRebalancePartMonitor.hpp"
|
|
#include"Boss/Msg/ManifestNotification.hpp"
|
|
#include"Boss/Msg/Manifestation.hpp"
|
|
#include"Boss/Msg/Notification.hpp"
|
|
#include"Boss/Msg/XRebalanceAttribution.hpp"
|
|
#include"Boss/concurrent.hpp"
|
|
#include"Boss/log.hpp"
|
|
#include"Ev/Io.hpp"
|
|
#include"Ev/map.hpp"
|
|
#include"Jsmn/Object.hpp"
|
|
#include"Ln/Amount.hpp"
|
|
#include"Ln/NodeId.hpp"
|
|
#include"Ln/Scid.hpp"
|
|
#include"S/Bus.hpp"
|
|
#include"Util/stringify.hpp"
|
|
#include<vector>
|
|
|
|
namespace Boss { namespace Mod {
|
|
|
|
void XRebalancePartMonitor::start() {
|
|
bus.subscribe<Msg::Manifestation
|
|
>([this](Msg::Manifestation const& _) {
|
|
/* lightningd only warns about subscriptions to topics no
|
|
* loaded plugin provides, so this is safe without the
|
|
* xrebalance plugin present. */
|
|
return bus.raise(Msg::ManifestNotification{
|
|
"xrebalance_part"
|
|
});
|
|
});
|
|
bus.subscribe<Msg::Notification
|
|
>([this](Msg::Notification const& n) {
|
|
if (n.notification != "xrebalance_part")
|
|
return Ev::lift();
|
|
|
|
auto first_scid = Ln::Scid();
|
|
auto return_scid = Ln::Scid();
|
|
auto amount = Ln::Amount();
|
|
auto fee = Ln::Amount();
|
|
try {
|
|
auto payload = n.params["xrebalance_part"];
|
|
if ( !payload.has("status")
|
|
|| !payload.has("first_hop")
|
|
|| !payload.has("return_hop")
|
|
|| !payload.has("delivered_msat")
|
|
|| !payload.has("fee_msat")
|
|
)
|
|
return Ev::lift();
|
|
/* Only completed parts carry earnings; failed and
|
|
* pending parts are the plugin's business. */
|
|
if (std::string(payload["status"]) != "complete")
|
|
return Ev::lift();
|
|
|
|
first_scid = Ln::Scid(std::string(
|
|
payload["first_hop"]
|
|
));
|
|
return_scid = Ln::Scid(std::string(
|
|
payload["return_hop"]
|
|
));
|
|
amount = Ln::Amount::object(
|
|
payload["delivered_msat"]
|
|
);
|
|
fee = Ln::Amount::object(
|
|
payload["fee_msat"]
|
|
);
|
|
} catch (std::runtime_error const& err) {
|
|
return Boss::log( bus, Error
|
|
, "XRebalancePartMonitor: unexpected "
|
|
"xrebalance_part payload: %s: %s"
|
|
, Util::stringify(n.params).c_str()
|
|
, err.what()
|
|
);
|
|
}
|
|
|
|
auto f = [this](Ln::Scid scid) {
|
|
return peer_from_scid_rr.execute(Msg::RequestPeerFromScid{
|
|
nullptr, scid
|
|
}).then([](Msg::ResponsePeerFromScid r) {
|
|
return Ev::lift(std::move(r.peer));
|
|
});
|
|
};
|
|
auto scids = std::vector<Ln::Scid>{first_scid, return_scid};
|
|
return Ev::map( std::move(f), std::move(scids)
|
|
).then([ this
|
|
, amount
|
|
, fee
|
|
](std::vector<Ln::NodeId> nids) {
|
|
return cont( std::move(nids[0])
|
|
, std::move(nids[1])
|
|
, amount
|
|
, fee
|
|
);
|
|
});
|
|
});
|
|
}
|
|
|
|
Ev::Io<void> XRebalancePartMonitor::cont( Ln::NodeId source
|
|
, Ln::NodeId destination
|
|
, Ln::Amount amount
|
|
, Ln::Amount fee
|
|
) {
|
|
if (!source || !destination)
|
|
/* Funds moved but the channel is gone from
|
|
* listpeerchannels (closed between part completion and
|
|
* this lookup); the earnings go unattributed. */
|
|
return Boss::log( bus, Warn
|
|
, "XRebalancePartMonitor: completed part on "
|
|
"unknown channel, not attributed."
|
|
);
|
|
|
|
auto act = Ev::lift();
|
|
act += Boss::log( bus, Debug
|
|
, "XRebalancePartMonitor: %s -> %s, "
|
|
"moved %s, fee %s."
|
|
, std::string(source).c_str()
|
|
, std::string(destination).c_str()
|
|
, std::string(amount).c_str()
|
|
, std::string(fee).c_str()
|
|
);
|
|
act += Boss::concurrent(bus.raise(Msg::XRebalanceAttribution{
|
|
std::move(source),
|
|
std::move(destination),
|
|
amount,
|
|
fee
|
|
}));
|
|
return act;
|
|
}
|
|
|
|
}}
|