diff --git a/Boss/Mod/DemandTracker.cpp b/Boss/Mod/DemandTracker.cpp new file mode 100644 index 0000000..59a735c --- /dev/null +++ b/Boss/Mod/DemandTracker.cpp @@ -0,0 +1,36 @@ +#include"Boss/Mod/DemandTracker.hpp" +#include"Boss/Msg/DemandObserved.hpp" +#include"Boss/Msg/ProvideHtlcAcceptedDeferrer.hpp" +#include"Boss/Msg/SolicitHtlcAcceptedDeferrer.hpp" +#include"Ev/Io.hpp" +#include"Ln/HtlcAccepted.hpp" +#include"S/Bus.hpp" + +namespace Boss { namespace Mod { + +void DemandTracker::start() { + bus.subscribe([this](Msg::SolicitHtlcAcceptedDeferrer const&) { + auto f = [this](Ln::HtlcAccepted::Request const& req) { + return htlc_accepted(req); + }; + return bus.raise(Msg::ProvideHtlcAcceptedDeferrer{ + std::move(f) + }); + }); +} + +Ev::Io +DemandTracker::htlc_accepted(Ln::HtlcAccepted::Request const& req) { + /* Not a forward (we are the recipient)? */ + if (!req.next_channel) + return Ev::lift(false); + /* This runs in the hook path: raise (subscribers keep their + * synchronous part cheap) and decline the HTLC either way. */ + auto msg = Msg::DemandObserved{req.next_channel}; + return bus.raise(std::move(msg)).then([]() { + return Ev::lift(false); + }); +} + +}} diff --git a/Boss/Mod/DemandTracker.hpp b/Boss/Mod/DemandTracker.hpp new file mode 100644 index 0000000..42c8031 --- /dev/null +++ b/Boss/Mod/DemandTracker.hpp @@ -0,0 +1,39 @@ +#ifndef BOSS_MOD_DEMANDTRACKER_HPP +#define BOSS_MOD_DEMANDTRACKER_HPP + +namespace Ev { template class Io; } +namespace Ln { namespace HtlcAccepted { struct Request; }} +namespace S { class Bus; } + +namespace Boss { namespace Mod { + +/** class Boss::Mod::DemandTracker + * + * @brief Observes forwards through an `htlc_accepted` deferrer + * and raises `Boss::Msg::DemandObserved` naming the outgoing + * channel of each one. + * + * Never holds an HTLC: the deferrer always declines immediately, + * so forwarding latency is unaffected. Consumers decide whether + * an observation warrants action (`Boss::Mod::XRebalancer`'s + * demand cycles). + */ +class DemandTracker { +private: + S::Bus& bus; + + void start(); + Ev::Io htlc_accepted(Ln::HtlcAccepted::Request const& req); + +public: + DemandTracker() =delete; + DemandTracker(DemandTracker&&) =delete; + DemandTracker(DemandTracker const&) =delete; + + explicit + DemandTracker(S::Bus& bus_) : bus(bus_) { start(); } +}; + +}} + +#endif /* !defined(BOSS_MOD_DEMANDTRACKER_HPP) */ diff --git a/Boss/Mod/XRebalancer.cpp b/Boss/Mod/XRebalancer.cpp index eb34227..533e14e 100644 --- a/Boss/Mod/XRebalancer.cpp +++ b/Boss/Mod/XRebalancer.cpp @@ -4,6 +4,7 @@ #include"Boss/ModG/RebalanceModeProxy.hpp" #include"Boss/ModG/RpcProxy.hpp" #include"Boss/Msg/DbResource.hpp" +#include"Boss/Msg/DemandObserved.hpp" #include"Boss/Msg/Init.hpp" #include"Boss/Msg/Manifestation.hpp" #include"Boss/Msg/ManifestOption.hpp" @@ -90,6 +91,13 @@ private: * suffices. */ bool use_plugin; bool started; + /* True while a cycle (matched or demand) runs. The Poisson + * loop and demand triggers exclude each other through it, and + * a demand trigger arriving while it is set is discarded -- + * traffic recurrence re-arms real demand, so there is no + * queue. Both cycle paths clear it behind a catch-all, so an + * exception cannot leave it wedged. */ + bool in_flight; /* One row per CHANNELD_NORMAL channel, built live from * listpeerchannels each cycle (balances and online status must be @@ -123,6 +131,7 @@ private: floor_auto = false; use_plugin = false; started = false; + in_flight = false; bus.subscribe([this](Msg::DbResource const& m) { @@ -185,6 +194,11 @@ private: }); + bus.subscribe([this](Msg::DemandObserved const& m) { + return handle_demand(m); + }); + bus.subscribe([this](Msg::Init const& _) { if (started) @@ -366,6 +380,11 @@ private: } Ev::Io tick() { + if (in_flight) + return Boss::log( bus, Debug + , "XRebalancer: tick skipped, cycle " + "in flight." ); + in_flight = true; return mode_proxy.get_mode().then([this](RebalanceMode m) { if ( m != RebalanceMode::xrebalance && m != RebalanceMode::xrebalance2 ) @@ -377,29 +396,79 @@ private: ); use_plugin = (m == RebalanceMode::xrebalance2); return run_cycle(); + /* The catch-all before the clear is load-bearing twice + * over: an exception on the fail path would skip a bare + * .then clear (wedging in_flight for good), and it would + * kill the awaiting loop greenthread outright. */ + }).catching([this](std::exception const& e) { + return Boss::log( bus, Warn + , "XRebalancer: cycle error: %s" + , e.what() ); + }).then([this]() { + in_flight = false; + return Ev::lift(); + }); + } + + /* Demand triggers (DemandTracker's htlc_accepted deferrer, via + * Msg::DemandObserved) run in the forwarding hook path, so the + * synchronous part stays cheap: the busy test and the spawn. + * Check and set have no await between them, so concurrent + * triggers cannot both pass. */ + Ev::Io handle_demand(Msg::DemandObserved const& m) { + if (!started || in_flight) + return Ev::lift(); + in_flight = true; + return Boss::concurrent( + demand_cycle(std::string(m.out_scid))); + } + + /* A demand-triggered cycle: the same pipeline as a matched one, + * routed to the demand plan by the scid argument. Runs outside + * the loop greenthread, so it carries its own mode gate and + * catch-all. */ + Ev::Io demand_cycle(std::string scid) { + return mode_proxy.get_mode().then([this, scid](RebalanceMode m) { + if ( m != RebalanceMode::xrebalance + && m != RebalanceMode::xrebalance2 ) + return Ev::lift(); + use_plugin = (m == RebalanceMode::xrebalance2); + return run_cycle(scid); + }).catching([this](std::exception const& e) { + return Boss::log( bus, Warn + , "XRebalancer: demand cycle error: %s" + , e.what() ); + }).then([this]() { + in_flight = false; + return Ev::lift(); }); } /* Fetch live balances/online (listpeerchannels), query the windowed - * per-node NetPpm, join, derive the matched-pool cycle, execute. */ - Ev::Io run_cycle() { + * per-node NetPpm, join, derive the cycle, execute. A non-empty + * demand_scid routes planning to the demand style (the channel a + * forward just exited through); empty runs the matched style. */ + Ev::Io run_cycle(std::string demand_scid = "") { return rpc.command( "listpeerchannels" , Json::Out::empty_object() - ).then([this](Jsmn::Object res) { + ).then([this, demand_scid](Jsmn::Object res) { return run_cycle_with(std::make_shared>( - build_chans(res))); + build_chans(res)), demand_scid); }); } Ev::Io - run_cycle_with(std::shared_ptr> chans) { + run_cycle_with( std::shared_ptr> chans + , std::string demand_scid + ) { if (chans->empty()) return Boss::log( bus, Info , "XRebalancer: no channel data, " "skipping cycle." ); auto cutoff = double(std::time(nullptr)) - window_days * 24.0 * 60.0 * 60.0; - return db.transact().then([this, cutoff, chans](Sqlite3::Tx tx) { + return db.transact().then([ this, cutoff, chans, demand_scid + ](Sqlite3::Tx tx) { auto net = std::make_shared>(); /* Per-peer capacity (msat): grant's credit base and * notional volume. */ @@ -469,7 +538,7 @@ private: (*net)[ce.first] = p; } tx.commit(); - return plan_and_log(chans, net); + return plan_and_log(chans, net, demand_scid); }); } @@ -632,6 +701,7 @@ private: Ev::Io plan_and_log( std::shared_ptr> chans , std::shared_ptr> net + , std::string const& demand_scid ) { /* Aggregate channels into peers; deficits aim at the band * edges on the aggregate Loc%. A peer with one full and @@ -707,13 +777,27 @@ private: [](PoolItem const& a, PoolItem const& b){ return a.ppm > b.ppm; }); - if (fill.empty() || drain.empty()) + if (fill.empty() || drain.empty()) { + /* Demand evaluations run per forward, so their + * no-op outcomes log at Debug; the paced matched + * cycle keeps the Info line. */ + if (!demand_scid.empty()) + return Boss::log( bus, Debug + , "XRebalancer: demand on %s: no " + "cycle -- NO_CANDIDATES (fill=%zu " + "drain=%zu)." + , demand_scid.c_str() + , fill.size(), drain.size() ); return Boss::log( bus, Info , "XRebalancer: no cycle -- NO_CANDIDATES " "(fill=%zu drain=%zu; bands fill<=%.1f " "drain>=%.1f, window=%.0fd)." , fill.size(), drain.size() , fill_band, drain_band, window_days ); + } + + if (!demand_scid.empty()) + return plan_demand(fill, drain, demand_scid); /* Cumulative deficit + marginal ppm per side. */ auto cum = [](std::vector const& pool){ @@ -859,6 +943,69 @@ private: }); } + /* Demand cycle: the target is the peer whose channel a forward + * just exited through. Fill-pool membership is the entire + * criterion -- demand controls WHEN we rebalance, never who + * qualifies, how much, or the price. Sized to the peer's + * deficit to the fill edge and priced conservatively: target + * NetPpm plus the minimum NetPpm of the offered pool, so every + * sat moved earns at least the target's side plus at least the + * cheapest offered channel's side. */ + Ev::Io + plan_demand( std::vector const& fill + , std::vector const& drain + , std::string const& scid + ) { + auto target = (PoolItem const*) nullptr; + for (auto const& it : fill) { + for (auto const& c : it.pr->chans) + if (c.scid == scid) { + target = ⁢ + break; + } + if (target) + break; + } + if (!target) + return Boss::log( bus, Debug + , "XRebalancer: demand on %s: peer not a " + "fill candidate, no cycle." + , scid.c_str() ); + /* Pools are sorted NetPpm-descending, so the minimum + * offered NetPpm is the last element. */ + auto min_offered = drain.back().ppm; + auto maxfee = std::uint32_t(std::llround( + target->ppm + min_offered)); + auto requested = target->deficit; + auto dest_caps = target->caps; + auto source_caps = std::vector(); + for (auto const& it : drain) + source_caps.insert( source_caps.end() + , it.caps.begin(), it.caps.end()); + return Boss::log( bus, Info + , "XRebalancer: cycle [demand] trigger=%s target=%s " + "window=%.0fd -> request=%s sat (deficit to fill " + "edge), maxfee=%u ppm (target %.1f + min offered " + "%.1f); sources=%zu dests=%zu; executing." + , scid.c_str() + , join_caps(target->caps).c_str() + , window_days + , Util::Str::group_digits(requested).c_str() + , (unsigned)maxfee + , target->ppm, min_offered + , source_caps.size(), dest_caps.size() + ).then([this, source_caps, dest_caps]() { + return Boss::log( bus, Debug + , "XRebalancer: sources=[%s] dests=[%s]" + , join_caps(source_caps).c_str() + , join_caps(dest_caps).c_str() + ); + }).then([this, source_caps, dest_caps, requested, maxfee]() { + return execute_cycle(source_caps, dest_caps, + requested, maxfee); + }); + } + /* Drive the chosen cycle through the executor the mode selects: * the in-clboss clboss-xmovefunds command (mode xrebalance, * reusing its sendpay/waitsendpay/harvest/attribution), or the diff --git a/Boss/Mod/all.cpp b/Boss/Mod/all.cpp index 63f2912..ce1384f 100644 --- a/Boss/Mod/all.cpp +++ b/Boss/Mod/all.cpp @@ -24,6 +24,7 @@ #include"Boss/Mod/ConnectFinderByHardcode.hpp" #include"Boss/Mod/Connector.hpp" #include"Boss/Mod/CommandReceiver.hpp" +#include"Boss/Mod/DemandTracker.hpp" #include"Boss/Mod/Dowser.hpp" #include"Boss/Mod/EarningsRebalancer.hpp" #include"Boss/Mod/EarningsTracker.hpp" @@ -224,6 +225,7 @@ std::shared_ptr all( std::ostream& cout all->install(bus); all->install(bus, *waiter); all->install(bus); + all->install(bus); all->install(bus); all->install(bus); all->install(bus); diff --git a/Boss/Msg/DemandObserved.hpp b/Boss/Msg/DemandObserved.hpp new file mode 100644 index 0000000..6394ef8 --- /dev/null +++ b/Boss/Msg/DemandObserved.hpp @@ -0,0 +1,26 @@ +#ifndef BOSS_MSG_DEMANDOBSERVED_HPP +#define BOSS_MSG_DEMANDOBSERVED_HPP + +#include"Ln/Scid.hpp" + +namespace Boss { namespace Msg { + +/** struct Boss::Msg::DemandObserved + * + * @brief Raised by `Boss::Mod::DemandTracker` for each forward + * about to exit through one of our channels: someone is spending + * that channel's outgoing liquidity right now. + * + * Pure observation -- the HTLC is never held. The amount is + * deliberately not carried: unforwardable probe HTLCs cost an + * attacker nothing, so any consumer sizing from a demanded + * amount would hand out a free lever over our spend. + */ +struct DemandObserved { + /* The outgoing channel of the forward. */ + Ln::Scid out_scid; +}; + +}} + +#endif /* !defined(BOSS_MSG_DEMANDOBSERVED_HPP) */ diff --git a/Makefile.am b/Makefile.am index 251997c..3528a8d 100644 --- a/Makefile.am +++ b/Makefile.am @@ -145,6 +145,8 @@ libclboss_la_SOURCES = \ Boss/Mod/Connector.hpp \ Boss/Mod/ConstructedListpeers.cpp \ Boss/Mod/ConstructedListpeers.hpp \ + Boss/Mod/DemandTracker.cpp \ + Boss/Mod/DemandTracker.hpp \ Boss/Mod/Dowser.cpp \ Boss/Mod/Dowser.hpp \ Boss/Mod/EarningsRebalancer.cpp \ @@ -320,6 +322,7 @@ libclboss_la_SOURCES = \ Boss/Msg/CommandRequest.hpp \ Boss/Msg/CommandResponse.hpp \ Boss/Msg/DbResource.hpp \ + Boss/Msg/DemandObserved.hpp \ Boss/Msg/MonitorFeeByBalance.hpp \ Boss/Msg/MonitorFeeByTheory.hpp \ Boss/Msg/MonitorFeeSetChannel.hpp \