XRebalancer: honor the "balance" unmanage tag

The classic rebalancers consult RebalanceUnmanager, and FundsMover
rejects any RequestMoveFunds naming an unmanaged node, but the
XRebalancer executes through clboss-xmovefunds or the external
xrebalance plugin, bypassing both checks, so peers unmanaged for
"balance" were still filled and drained.

Fetch the unmanaged set at planning time and exclude those peers
from the fill and drain pools.  Sources and destinations are drawn
only from the pools, so the one filter covers matched and demand
cycles in both executor modes.  Channeled peers the set withholds
are named in one Debug line per cycle, so an operator can tell an
unmanaged exclusion from a peer that merely failed candidacy.

Adds a test that drives a demand cycle over the bus with the
best-paying drain peer unmanaged and checks the resulting request
names it on neither side and that the exclusion line names it.

Boss/Mod/XRebalancer.cpp

@@ -2,6 +2,7 @@
 #include"Boss/Mod/Waiter.hpp"
 #include"Boss/Mod/Rpc.hpp"
 #include"Boss/ModG/RebalanceModeProxy.hpp"
+#include"Boss/ModG/RebalanceUnmanagerProxy.hpp"
 #include"Boss/ModG/RpcProxy.hpp"
 #include"Boss/Msg/DbResource.hpp"
 #include"Boss/Msg/DemandObserved.hpp"
@@ -31,6 +32,7 @@
 #include<iomanip>
 #include<map>
 #include<random>
+#include<set>
 #include<sstream>
 #include<string>
 #include<vector>
@@ -76,6 +78,7 @@ private:
 	Waiter& waiter;
 	ModG::RebalanceModeProxy mode_proxy;
 	ModG::RpcProxy rpc;
+	ModG::RebalanceUnmanagerProxy unmanager;
 	Sqlite3::Db db;
 
 	double per_hour;
@@ -712,11 +715,52 @@ private:
 		return ladder;
 	}
 
+	/* Fetch the "balance"-unmanaged set (clboss-unmanage), then
+	 * plan; unmanaged peers are excluded from both pools, which
+	 * covers matched and demand cycles in either executor mode.
+	 * Channeled peers the set withholds are named in one Debug
+	 * line per cycle, so an operator can see the exclusion doing
+	 * its work.  */
 	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
 		    ) {
+		return unmanager.get_unmanaged().then([ this, chans, net
+						      , demand_scid
+						      ](std::set<Ln::NodeId> const* unmanaged) {
+			auto excluded = std::ostringstream();
+			auto seen = std::set<Ln::NodeId>();
+			for (auto const& c : *chans) {
+				if (unmanaged->count(c.node) == 0)
+					continue;
+				if (!seen.insert(c.node).second)
+					continue;
+				if (seen.size() > 1)
+					excluded << " ";
+				excluded << std::string(c.node);
+			}
+			auto act = Ev::lift();
+			if (!seen.empty())
+				act = Boss::log( bus, Debug
+					, "XRebalancer: unmanaged (balance), "
+					  "excluded: %s"
+					, excluded.str().c_str() );
+			return std::move(act).then([ this, chans, net
+						   , demand_scid, unmanaged
+						   ]() {
+				return plan_and_log( chans, net, demand_scid
+						   , *unmanaged );
+			});
+		});
+	}
+
+	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
+		    , std::set<Ln::NodeId> const& unmanaged
+		    ) {
 		/* Aggregate channels into peers; deficits aim at the band
 		 * edges on the aggregate Loc%.  A peer with one full and
 		 * one empty channel nets out balanced and is left alone --
@@ -754,6 +798,8 @@ private:
 		for (auto const& p : peers) {
 			if (!p.online)
 				continue;
+			if (unmanaged.count(p.node) != 0)
+				continue;
 			auto it = net->find(p.node);
 			if (it == net->end())
 				continue;
@@ -1488,7 +1534,8 @@ public:
 
 	explicit
 	Impl(S::Bus& bus_, Waiter& waiter_)
-		: bus(bus_), waiter(waiter_), mode_proxy(bus_), rpc(bus_) {
+		: bus(bus_), waiter(waiter_), mode_proxy(bus_), rpc(bus_)
+		, unmanager(bus_) {
 		start();
 	}
 };

Makefile.am

@@ -670,6 +670,7 @@ TESTS = \
 	tests/boss/test_xrebalancepartmonitor \
 	tests/boss/test_xrebalancepredict \
 	tests/boss/test_xrebalancepredictor \
+	tests/boss/test_xrebalancer \
 	tests/boss/test_peerjudge_algo \
 	tests/boss/test_peerjudge_datagatherer \
 	tests/boss/test_peerstatistician \

tests/boss/test_xrebalancer.cpp

@@ -0,0 +1,305 @@
+#undef NDEBUG
+
+#include"Boss/Mod/RebalanceUnmanager.hpp"
+#include"Boss/Mod/Rpc.hpp"
+#include"Boss/Mod/Waiter.hpp"
+#include"Boss/Mod/XRebalancer.hpp"
+#include"Boss/Msg/DbResource.hpp"
+#include"Boss/Msg/DemandObserved.hpp"
+#include"Boss/Msg/Init.hpp"
+#include"Boss/Msg/JsonCout.hpp"
+#include"Boss/Msg/Option.hpp"
+#include"Boss/Msg/RequestRebalanceMode.hpp"
+#include"Boss/Msg/RequestRpcCommand.hpp"
+#include"Boss/Msg/ResponseRebalanceMode.hpp"
+#include"Boss/Msg/ResponseRpcCommand.hpp"
+#include"Boss/RebalanceMode.hpp"
+#include"Boss/Shutdown.hpp"
+#include"Ev/Io.hpp"
+#include"Ev/now.hpp"
+#include"Ev/start.hpp"
+#include"Ev/yield.hpp"
+#include"Jsmn/Object.hpp"
+#include"Json/Out.hpp"
+#include"Ln/NodeId.hpp"
+#include"Ln/Scid.hpp"
+#include"Net/Connector.hpp"
+#include"Net/Fd.hpp"
+#include"Net/SocketFd.hpp"
+#include"S/Bus.hpp"
+#include"Secp256k1/PubKey.hpp"
+#include"Secp256k1/Signature.hpp"
+#include"Secp256k1/SignerIF.hpp"
+#include"Sha256/Hash.hpp"
+#include"Sqlite3.hpp"
+#include<assert.h>
+#include<ctime>
+#include<iostream>
+#include<sstream>
+#include<string>
+#include<sys/socket.h>
+
+/* Exercises the XRebalancer demand-cycle pipeline end to end on the
+ * bus, checking that a peer unmanaged for "balance" is excluded from
+ * both the fill and drain pools: peer C below is deliberately the
+ * best-paying drain candidate, so without the exclusion it would top
+ * the source list of every cycle.  */
+
+namespace {
+
+/* Peer A: 5% local -> fill candidate; the demand target.  */
+auto const node_a = "020000000000000000000000000000000000000000000000000000000000000000";
+/* Peer B: 95% local -> drain candidate.  */
+auto const node_b = "020000000000000000000000000000000000000000000000000000000000000001";
+/* Peer C: 95% local, best in_net -> drain candidate, but unmanaged.  */
+auto const node_c = "020000000000000000000000000000000000000000000000000000000000000002";
+
+auto const scid_a = "103x1x0";
+auto const scid_b = "103x1x1";
+auto const scid_c = "103x2x0";
+
+auto const listpeerchannels_result = R"JSON(
+{
+  "channels": [
+    {
+      "state": "CHANNELD_NORMAL",
+      "to_us_msat": "50000000msat",
+      "total_msat": "1000000000msat",
+      "short_channel_id": "103x1x0",
+      "peer_id": "020000000000000000000000000000000000000000000000000000000000000000",
+      "peer_connected": true
+    },
+    {
+      "state": "CHANNELD_NORMAL",
+      "to_us_msat": "950000000msat",
+      "total_msat": "1000000000msat",
+      "short_channel_id": "103x1x1",
+      "peer_id": "020000000000000000000000000000000000000000000000000000000000000001",
+      "peer_connected": true
+    },
+    {
+      "state": "CHANNELD_NORMAL",
+      "to_us_msat": "950000000msat",
+      "total_msat": "1000000000msat",
+      "short_channel_id": "103x2x0",
+      "peer_id": "020000000000000000000000000000000000000000000000000000000000000002",
+      "peer_connected": true
+    }
+  ]
+}
+)JSON";
+
+class DummyConnector : public Net::Connector {
+public:
+	Net::SocketFd
+	connect(std::string const& host, int port) override {
+		(void) host;
+		(void) port;
+		return Net::SocketFd();
+	}
+};
+
+class DummySigner : public Secp256k1::SignerIF {
+public:
+	Secp256k1::PubKey
+	get_pubkey_tweak(Secp256k1::PrivKey const& tweak) override {
+		(void) tweak;
+		return Secp256k1::PubKey();
+	}
+
+	Secp256k1::Signature
+	get_signature_tweak( Secp256k1::PrivKey const& tweak
+			   , Sha256::Hash const& m
+			   ) override {
+		(void) tweak;
+		(void) m;
+		return Secp256k1::Signature();
+	}
+
+	Sha256::Hash
+	get_privkey_salted_hash(std::uint8_t salt[32]) override {
+		if (!salt)
+			return Sha256::Hash();
+		auto hash = Sha256::Hash();
+		hash.from_buffer(salt);
+		return hash;
+	}
+};
+
+bool has_scid(Jsmn::Object const& arr, char const* scid) {
+	for (auto i = std::size_t(0); i < arr.size(); ++i)
+		if (std::string(arr[i]) == scid)
+			return true;
+	return false;
+}
+
+Ev::Io<void> wait_flag(bool& flag, double start) {
+	return Ev::yield().then([&flag, start]() {
+		if (flag)
+			return Ev::lift();
+		assert(Ev::now() - start < 10.0); /* Time out.  */
+		return wait_flag(flag, start);
+	});
+}
+
+}
+
+int main() {
+	auto bus = S::Bus();
+	Boss::Mod::Waiter waiter(bus);
+
+	/* Mode stub: always xrebalance (the in-clboss executor, so the
+	 * cycle surfaces as a clboss-xmovefunds RPC we can capture).  */
+	bus.subscribe<Boss::Msg::RequestRebalanceMode
+		     >([&](Boss::Msg::RequestRebalanceMode const& m) {
+		return bus.raise(Boss::Msg::ResponseRebalanceMode{
+			m.requester, Boss::RebalanceMode::xrebalance
+		});
+	});
+
+	/* Collect log output (Boss::log emits JsonCout).  */
+	auto log_lines = std::string();
+	bus.subscribe<Boss::Msg::JsonCout
+		     >([&](Boss::Msg::JsonCout const& m) {
+		log_lines += m.obj.output();
+		log_lines += "\n";
+		return Ev::lift();
+	});
+
+	/* RPC stub.  */
+	auto xmf_called = false;
+	auto xmf_params = std::string();
+	bus.subscribe<Boss::Msg::RequestRpcCommand
+		     >([&](Boss::Msg::RequestRpcCommand const& m) {
+		auto respond = [&](char const* res) {
+			return bus.raise(Boss::Msg::ResponseRpcCommand{
+				m.requester, true,
+				Jsmn::Object::parse_json(res), ""
+			});
+		};
+		if (m.command == "listpeerchannels")
+			return respond(listpeerchannels_result);
+		if (m.command == "clboss-xmovefunds") {
+			xmf_params = m.params.output();
+			xmf_called = true;
+			return respond(R"JSON({"execution": {}})JSON");
+		}
+		std::cerr << "UNMOCKED COMMAND " << m.command << std::endl;
+		assert(0);
+		return Ev::lift();
+	});
+
+	Boss::Mod::RebalanceUnmanager unmanager(bus, {node_c});
+
+	/* Module under test.  */
+	auto mut = Boss::Mod::XRebalancer(bus, waiter);
+
+	/* Init carries references the XRebalancer never touches; the
+	 * Rpc rides on an idle socketpair.  */
+	auto connector = DummyConnector();
+	auto signer = DummySigner();
+	auto db = Sqlite3::Db(":memory:");
+	int sockets[2];
+	auto sockres = socketpair(AF_UNIX, SOCK_STREAM, 0, sockets);
+	assert(sockres >= 0);
+	auto server_socket = Net::Fd(sockets[0]);
+	auto client_socket = Net::Fd(sockets[1]);
+	auto rpc = Boss::Mod::Rpc(bus, std::move(client_socket));
+
+	auto code = Ev::lift().then([&]() {
+		return db.transact();
+	}).then([&](Sqlite3::Tx tx) {
+		/* The columns the XRebalancer NetPpm query reads.  All
+		 * three peers get a positive net on the side that makes
+		 * them pool candidates; C pays the most on the drain
+		 * side.  */
+		tx.query_execute(R"QRY(
+		CREATE TABLE "EarningsTracker"
+		     ( node TEXT NOT NULL
+		     , time_bucket REAL NOT NULL
+		     , in_earnings INTEGER NOT NULL
+		     , in_forwarded INTEGER NOT NULL
+		     , in_expenditures INTEGER NOT NULL
+		     , out_earnings INTEGER NOT NULL
+		     , out_forwarded INTEGER NOT NULL
+		     , out_expenditures INTEGER NOT NULL
+		     );
+		)QRY");
+		auto now = double(std::time(nullptr));
+		auto insert = [&]( char const* node
+				 , std::int64_t in_e, std::int64_t in_f
+				 , std::int64_t out_e, std::int64_t out_f
+				 ) {
+			tx.query(R"QRY(
+			INSERT INTO "EarningsTracker"
+			VALUES( :node, :time_bucket
+			      , :in_e, :in_f, 0
+			      , :out_e, :out_f, 0
+			      );
+			)QRY")
+				.bind(":node", node)
+				.bind(":time_bucket", now)
+				.bind(":in_e", in_e)
+				.bind(":in_f", in_f)
+				.bind(":out_e", out_e)
+				.bind(":out_f", out_f)
+				.execute();
+		};
+		insert(node_a, 0, 0, 1000000, 1000000000); /* out 1000ppm */
+		insert(node_b, 500000, 1000000000, 0, 0);  /* in 500ppm */
+		insert(node_c, 2000000, 1000000000, 0, 0); /* in 2000ppm */
+		tx.commit();
+
+		/* Pause the Poisson loop so only the demand trigger
+		 * below can start a cycle.  */
+		return bus.raise(Boss::Msg::Option{
+			"clboss-xrebalance-per-hour",
+			Jsmn::Object::parse_json(R"JSON("0")JSON"),
+			nullptr
+		});
+	}).then([&]() {
+		return bus.raise(Boss::Msg::DbResource{db});
+	}).then([&]() {
+		return bus.raise(Boss::Msg::Init{
+			Boss::Msg::Network_Regtest,
+			rpc,
+			Ln::NodeId(node_a),
+			db,
+			connector,
+			signer,
+			std::string(),
+			false
+		});
+	}).then([&]() {
+		/* A forward just spent A's outgoing liquidity.  */
+		return bus.raise(Boss::Msg::DemandObserved{
+			Ln::Scid(std::string(scid_a))
+		});
+	}).then([&]() {
+		return wait_flag(xmf_called, Ev::now());
+	}).then([&]() {
+		auto req = Jsmn::Object::parse_json(xmf_params.c_str());
+		auto srcs = req["source_scid"];
+		auto dsts = req["dest_scid"];
+
+		/* The demand target fills A.  */
+		assert(has_scid(dsts, scid_a));
+		/* B is the only remaining drain source.  */
+		assert(has_scid(srcs, scid_b));
+		/* The unmanaged C appears on neither side, even though
+		 * it out-pays B.  */
+		assert(!has_scid(srcs, scid_c));
+		assert(!has_scid(dsts, scid_c));
+
+		/* The exclusion is named in the cycle's log.  */
+		assert(log_lines.find(std::string(
+			"unmanaged (balance), excluded: ") + node_c)
+			!= std::string::npos);
+
+		return bus.raise(Boss::Shutdown{});
+	}).then([&]() {
+		return Ev::lift(0);
+	});
+
+	return Ev::start(code);
+}