DemandTracker: trigger demand rebalance cycles from observed forwards

New module DemandTracker registers an htlc_accepted deferrer that
never holds an HTLC: for each forward it raises DemandObserved
naming the outgoing channel, then immediately declines.  The
message deliberately carries no amount: unforwardable probe HTLCs
cost an attacker nothing, so sizing from a demanded amount would
be a free lever over our spend.

XRebalancer consumes the message.  A trigger arriving while any
cycle runs is discarded (traffic recurrence re-arms real demand);
otherwise it runs a demand cycle: the same fetch/join pipeline as
a matched cycle, targeting the peer whose channel the forward
exited through.  Fill-pool membership is the entire criterion --
demand controls when we rebalance, never who qualifies or how
much.  The request restores the peer to the fill edge, priced at
target NetPpm plus the minimum offered NetPpm, executed through
the existing executor with cycle tag [demand].

The new in_flight flag serializes matched and demand cycles; both
paths clear it behind a catch-all so an exception cannot wedge
it.  The catch-all around the matched tick also keeps the Poisson
loop alive on RPC errors, which previously terminated it with
only a stderr notice.

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<Msg::SolicitHtlcAcceptedDeferrer
+		     >([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<bool>
+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);
+	});
+}
+
+}}

Boss/Mod/DemandTracker.hpp

@@ -0,0 +1,39 @@
+#ifndef BOSS_MOD_DEMANDTRACKER_HPP
+#define BOSS_MOD_DEMANDTRACKER_HPP
+
+namespace Ev { template<typename a> 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<bool> 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) */

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<Msg::DbResource
 			     >([this](Msg::DbResource const& m) {
@@ -185,6 +194,11 @@ private:
 		});
 
 
+		bus.subscribe<Msg::DemandObserved
+			     >([this](Msg::DemandObserved const& m) {
+			return handle_demand(m);
+		});
+
 		bus.subscribe<Msg::Init
 			     >([this](Msg::Init const& _) {
 			if (started)
@@ -366,6 +380,11 @@ private:
 	}
 
 	Ev::Io<void> 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<std::exception>([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<void> 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<void> 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<std::exception>([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<void> 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<void> 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<std::vector<Chan>>(
-				build_chans(res)));
+				build_chans(res)), demand_scid);
 		});
 	}
 
 	Ev::Io<void>
-	run_cycle_with(std::shared_ptr<std::vector<Chan>> chans) {
+	run_cycle_with( std::shared_ptr<std::vector<Chan>> 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<std::map<Ln::NodeId, NetPpm>>();
 			/* 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<void>
 	plan_and_log( std::shared_ptr<std::vector<Chan>> chans
 		    , std::shared_ptr<std::map<Ln::NodeId, NetPpm>> 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<PoolItem> 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<void>
+	plan_demand( std::vector<PoolItem> const& fill
+		   , std::vector<PoolItem> 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 = &it;
+					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<ScidCap>();
+		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

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<void> all( std::ostream& cout
 	all->install<XRebalancePredictor>(bus);
 	all->install<XRebalancer>(bus, *waiter);
 	all->install<XRebalancePartMonitor>(bus);
+	all->install<DemandTracker>(bus);
 	all->install<MoveFundsCommand>(bus);
 	all->install<EarningsTracker>(bus);
 	all->install<JitRebalancer>(bus);

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) */

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 \