clboss/Boss/Mod/XMoveFunds/Main.cpp

#include"Boss/Mod/AskreneLayer.hpp"
#include"Boss/Mod/Rpc.hpp"
#include"Boss/Mod/XMoveFunds/Main.hpp"
#include"Boss/Msg/CommandFail.hpp"
#include"Boss/Msg/CommandRequest.hpp"
#include"Boss/Msg/CommandResponse.hpp"
#include"Boss/Msg/Init.hpp"
#include"Boss/Msg/ManifestCommand.hpp"
#include"Boss/Msg/Manifestation.hpp"
#include"Boss/concurrent.hpp"
#include"Boss/log.hpp"
#include"Ev/Io.hpp"
#include"Ev/yield.hpp"
#include"Jsmn/Object.hpp"
#include"Json/Out.hpp"
#include"Ln/Amount.hpp"
#include"Ln/CommandId.hpp"
#include"Ln/NodeId.hpp"
#include"Ln/Scid.hpp"
#include"S/Bus.hpp"
#include"Util/make_unique.hpp"
#include"Util/stringify.hpp"
#include"Uuid.hpp"
#include<algorithm>
#include<cinttypes>
#include<memory>
#include<set>
#include<sstream>
#include<vector>

namespace {

/* JSON-RPC error code we use for malformed parameters.  Matches
 * the JSONRPC2 invalid-params constant used elsewhere in clboss
 * (e.g. Dowser, MoveFundsCommand). */
constexpr int RPC_INVALID_PARAMS = -32602;

/* Decode either a single scid string or an array of scid strings
 * from a JSON value into a vector.  Throws on type/format error
 * with a message suitable for surfacing in the RPC reply. */
std::vector<Ln::Scid>
parse_scid_list(Jsmn::Object const& j, char const* fieldname) {
	std::vector<Ln::Scid> out;
	auto push_one = [&out, fieldname](Jsmn::Object const& s) {
		if (!s.is_string())
			throw std::runtime_error(
				std::string(fieldname)
				+ " must be a scid string or array of "
				  "scid strings");
		out.emplace_back(std::string(s));
	};
	if (j.is_string()) {
		push_one(j);
	} else if (j.is_array()) {
		for (auto i = std::size_t(0); i < j.size(); ++i)
			push_one(j[i]);
	} else {
		throw std::runtime_error(
			std::string(fieldname)
			+ " must be a scid string or array of scid "
			  "strings");
	}
	if (out.empty())
		throw std::runtime_error(
			std::string(fieldname)
			+ " must be non-empty");
	return out;
}

std::string
join_scids(std::vector<Ln::Scid> const& v) {
	auto os = std::ostringstream();
	auto first = true;
	for (auto const& s : v) {
		if (!first) os << ",";
		os << std::string(s);
		first = false;
	}
	return os.str();
}

/* Parse a JSON value as a u32, accepting either a JSON number or a
 * numeric string.  lightning-cli encodes unquoted CLI values as JSON
 * numbers (so `maxparts=10` arrives as the number 10), while object-
 * form RPC calls sometimes pass them as strings.  Mirrors the
 * permissive shape of Ln::Amount::object. */
std::uint32_t
parse_u32(Jsmn::Object const& o, char const* fieldname) {
	if (o.is_number()) {
		return std::uint32_t(double(o));
	}
	if (o.is_string()) {
		try {
			return std::uint32_t(
				std::stoul(std::string(o)));
		} catch (std::exception const&) {
			throw std::runtime_error(
				std::string(fieldname)
				+ " must be an integer");
		}
	}
	throw std::runtime_error(
		std::string(fieldname) + " must be an integer");
}

}

namespace Boss { namespace Mod { namespace XMoveFunds {

class Main::Impl {
private:
	S::Bus& bus;
	Boss::Mod::Rpc* rpc;
	Ln::NodeId self_id;
	/* True once create_xrebalance_layer() has resolved (either by
	 * successfully creating/finding the persistent layer, or by
	 * logging a non-fatal RpcError on older CLN).  Gated on by
	 * wait_for_ready() so that command handling never tries to
	 * use the layer before askrene is told about it. */
	bool layer_ready;

	struct Params {
		std::vector<Ln::Scid> source_scids;
		std::vector<Ln::Scid> dest_scids;
		Ln::Amount amount;
		Ln::Amount maxfee;
		std::uint32_t maxparts;
		bool execute;
	};

	/* Parse the JSON params object.  Throws on bad input. */
	Params parse_params(Jsmn::Object const& params) {
		auto p = Params();
		p.maxparts = 10;
		p.execute = false;

		if (!params.is_object())
			throw std::runtime_error(
				"params must be an object "
				"(named-parameter form required)");

		if (!params.has("source_scid"))
			throw std::runtime_error("source_scid required");
		p.source_scids =
		    parse_scid_list(params["source_scid"], "source_scid");

		if (!params.has("dest_scid"))
			throw std::runtime_error("dest_scid required");
		p.dest_scids =
		    parse_scid_list(params["dest_scid"], "dest_scid");

		if (!params.has("amount_msat"))
			throw std::runtime_error("amount_msat required");
		try {
			p.amount = Ln::Amount::object(params["amount_msat"]);
		} catch (std::exception const&) {
			throw std::runtime_error(
				"amount_msat must be an integer number of "
				"msat (as a JSON number or string)");
		}
		auto amount_msat = std::uint64_t(p.amount.to_msat());
		if (amount_msat == 0)
			throw std::runtime_error("amount_msat must be > 0");

		/* Default maxfee: 5000 ppm of amount, capped at sensible
		 * minimum so we never pass 0 to askrene for tiny
		 * amounts.  Caller can override via maxfee_msat. */
		auto default_maxfee_msat =
		    std::max(std::uint64_t(1000), amount_msat * 5000 / 1000000);
		p.maxfee = Ln::Amount::msat(default_maxfee_msat);
		if (params.has("maxfee_msat")) {
			try {
				p.maxfee = Ln::Amount::object(
				    params["maxfee_msat"]);
			} catch (std::exception const&) {
				throw std::runtime_error(
					"maxfee_msat must be an integer "
					"number of msat");
			}
		}

		if (params.has("maxparts")) {
			p.maxparts = parse_u32(params["maxparts"],
					       "maxparts");
			if (p.maxparts == 0)
				throw std::runtime_error(
					"maxparts must be > 0");
		}

		if (params.has("execute")) {
			auto e = params["execute"];
			if (e.is_boolean()) {
				p.execute = bool(e);
			} else if (e.is_string()) {
				auto s = std::string(e);
				if (s == "true") p.execute = true;
				else if (s == "false") p.execute = false;
				else throw std::runtime_error(
					"execute must be a boolean");
			} else {
				throw std::runtime_error(
					"execute must be a boolean");
			}
		}

		return p;
	}

	/* Ensure the persistent xrebalance askrene layer exists.
	 * Called once at startup, fire-and-forget.  Idempotent: when
	 * persistent is true, askrene-create-layer succeeds even if
	 * the layer already exists.  Failures (e.g. CLN < v24.11
	 * where the RPC does not exist, or stock CLN that lacks the
	 * circular-routing patch) are logged but non-fatal --
	 * subsequent xmovefunds calls will surface the underlying
	 * crash if the caller invokes them. */
	Ev::Io<void> create_xrebalance_layer() {
		return Ev::lift().then([this]() {
			auto parms = Json::Out()
				.start_object()
					.field("layer",
					       Boss::Mod::AskreneLayer::
					           xrebalance_layer_name)
					.field("persistent", true)
				.end_object()
				;
			return rpc->command( "askrene-create-layer"
					   , std::move(parms)
					   );
		}).then([this](Jsmn::Object _) {
			layer_ready = true;
			return Boss::log( bus, Debug
					, "XMoveFunds: persistent "
					  "askrene layer '%s' ready"
					, Boss::Mod::AskreneLayer::
					      xrebalance_layer_name
					      .c_str()
					);
		}).catching<RpcError>([this](RpcError const& e) {
			/* Mark ready even on failure: degraded mode
			 * must still allow plan calls to proceed
			 * (their getroutes call will surface a clearer
			 * error than us deadlocking on
			 * wait_for_ready). */
			layer_ready = true;
			return Boss::log( bus, Error
					, "XMoveFunds: askrene-create-"
					  "layer (%s) failed: %s; will "
					  "proceed in degraded mode "
					  "(no persistent learning "
					  "layer)."
					, Boss::Mod::AskreneLayer::
					      xrebalance_layer_name
					      .c_str()
					, Util::stringify(e.error).c_str()
					);
		});
	}

	/* Gate command handling on startup completion: rpc must have
	 * arrived via Msg::Init, and create_xrebalance_layer() must
	 * have completed (either successfully or via the logged-
	 * RpcError graceful-degradation path). */
	Ev::Io<void> wait_for_ready() {
		return Ev::lift().then([this]() {
			if (!rpc || !layer_ready)
				return Ev::yield() + wait_for_ready();
			return Ev::lift();
		});
	}

	/* Fetch our channels via listpeerchannels.  Returns the
	 * "channels" array. */
	Ev::Io<Jsmn::Object> list_my_channels() {
		auto parms = Json::Out()
			.start_object()
			.end_object()
			;
		return rpc->command( "listpeerchannels"
				   , std::move(parms)
				   ).then([](Jsmn::Object res) {
			return Ev::lift(res["channels"]);
		});
	}

	/* Create a transient (persistent=false) askrene layer.  Used
	 * for the per-request mask state. */
	Ev::Io<void> create_transient_layer(std::string layer) {
		auto parms = Json::Out()
			.start_object()
				.field("layer", layer)
				.field("persistent", false)
			.end_object()
			;
		return rpc->command( "askrene-create-layer"
				   , std::move(parms)
				   ).then([](Jsmn::Object _) {
			return Ev::lift();
		});
	}

	/* Remove a transient askrene layer.  Best-effort: any RpcError
	 * is logged but swallowed because we may be on a cleanup path
	 * after some other failure and the caller has already given
	 * up. */
	Ev::Io<void> remove_layer(std::string layer) {
		auto parms = Json::Out()
			.start_object()
				.field("layer", layer)
			.end_object()
			;
		return rpc->command( "askrene-remove-layer"
				   , std::move(parms)
				   ).then([](Jsmn::Object _) {
			return Ev::lift();
		}).catching<RpcError>([this, layer](RpcError const& e) {
			return Boss::log( bus, Debug
					, "XMoveFunds: askrene-remove-"
					  "layer (%s) failed: %s "
					  "(non-fatal, ignored)"
					, layer.c_str()
					, Util::stringify(e.error).c_str()
					);
		});
	}

	/* Compute the direction (0 or 1) corresponding to "us
	 * sending into this channel" per BOLT 7 canonical ordering:
	 * direction 0 is the lower-id node as sender, direction 1 is
	 * the higher-id node as sender. */
	std::uint32_t us_to_peer_dir(Ln::NodeId const& peer) const {
		return self_id < peer ? 0 : 1;
	}
	std::uint32_t peer_to_us_dir(Ln::NodeId const& peer) const {
		return self_id < peer ? 1 : 0;
	}

	/* For each of our channels, decide whether each direction
	 * should be masked off in the transient layer, and return a
	 * chained Ev::Io<void> that writes all the masks
	 * sequentially. */
	Ev::Io<void>
	write_masks(std::string layer,
		    Jsmn::Object channels,
		    Params const& p) {
		auto source_set = std::set<std::string>();
		for (auto const& s : p.source_scids)
			source_set.insert(std::string(s));
		auto dest_set = std::set<std::string>();
		for (auto const& s : p.dest_scids)
			dest_set.insert(std::string(s));

		auto chain = Ev::lift();
		auto count = std::size_t(0);

		for (auto i = std::size_t(0); i < channels.size(); ++i) {
			auto ch = channels[i];
			if (!ch.has("state")
			    || std::string(ch["state"])
			        != "CHANNELD_NORMAL")
				continue;
			if (!ch.has("short_channel_id")
			    || !ch.has("peer_id"))
				continue;
			auto scid_str =
			    std::string(ch["short_channel_id"]);
			auto scid = Ln::Scid(scid_str);
			auto peer = Ln::NodeId(
			    std::string(ch["peer_id"]));

			auto disable_dir =
			    [this, layer]
			    (Ln::Scid s, std::uint32_t dir) {
				return Boss::Mod::AskreneLayer::
				    update_channel(
					*rpc, layer, s, dir,
					/* enabled = */ false,
					Ln::Amount::msat(0),
					Ln::Amount::msat(0),
					Ln::Amount::msat(0),
					/* fee_prop = */ 0,
					/* cltv = */ 0);
			};

			if (!source_set.count(scid_str)) {
				chain = std::move(chain)
				      + disable_dir(scid,
						    us_to_peer_dir(peer));
				++count;
			}
			if (!dest_set.count(scid_str)) {
				chain = std::move(chain)
				      + disable_dir(scid,
						    peer_to_us_dir(peer));
				++count;
			}
		}

		return std::move(chain)
		     + Boss::log( bus, Debug
				, "XMoveFunds: wrote %zu mask "
				  "entries to transient layer %s"
				, count
				, layer.c_str()
				);
	}

	/* Build and issue the askrene-getroutes call with
	 * source = destination = self_id (the patched askrene
	 * interprets this as circular self-rebalance routing).
	 * Includes auto.localchans, the persistent xrebalance
	 * layer, and the per-request transient layer. */
	Ev::Io<Jsmn::Object>
	call_getroutes(std::string transient, Params const& p) {
		auto parms = Json::Out();
		auto obj = parms.start_object();
		obj.field("source", std::string(self_id));
		obj.field("destination", std::string(self_id));
		obj.field("amount_msat",
			  std::uint64_t(p.amount.to_msat()));
		auto la = obj.start_array("layers");
		la.entry(std::string("auto.localchans"));
		la.entry(Boss::Mod::AskreneLayer::
			     xrebalance_layer_name);
		la.entry(transient);
		la.end_array();
		obj.field("maxfee_msat",
			  std::uint64_t(p.maxfee.to_msat()));
		obj.field("final_cltv", std::uint32_t(14));
		obj.field("maxparts", p.maxparts);
		obj.end_object();
		return rpc->command("getroutes", std::move(parms));
	}

	/* Per-request flow.  Builds + uses a uuid-suffixed transient
	 * layer, calls getroutes, and returns the askrene response
	 * embedded in our plan reply.  The transient layer is
	 * removed before returning (success or failure). */
	Ev::Io<void>
	do_plan(std::shared_ptr<Params> p, Ln::CommandId id) {
		auto transient =
		    Boss::Mod::AskreneLayer::xrebalance_layer_name
		    + "-tmp-"
		    + std::string(Uuid::random());
		auto routes = std::make_shared<Jsmn::Object>();
		auto err_msg = std::make_shared<std::string>();
		auto err_code = std::make_shared<int>(0);

		return create_transient_layer(transient
		).then([this, p, transient]() {
			return list_my_channels();
		}).then([this, p, transient]
			(Jsmn::Object channels) {
			return write_masks(transient, channels, *p);
		}).then([this, p, transient]() {
			return call_getroutes(transient, *p);
		}).then([routes](Jsmn::Object r) {
			*routes = r;
			return Ev::lift();
		}).catching<RpcError>(
		    [err_msg, err_code](RpcError const& e) {
			*err_code = -32603;
			*err_msg = Util::stringify(e.error);
			return Ev::lift();
		}).then([this, transient]() {
			return remove_layer(transient);
		}).then([this, p, id, routes, err_msg, err_code]() {
			if (*err_code != 0) {
				return bus.raise(Msg::CommandFail{
					id, *err_code,
					"getroutes failed: " + *err_msg,
					Json::Out::empty_object()
				});
			}

			auto plan = Json::Out();
			auto obj = plan.start_object();
			obj.field("status",
				  std::string(p->execute
					      ? "ready"
					      : "planned"));
			{
				auto arr =
				    obj.start_array("source_scids");
				for (auto const& s : p->source_scids)
					arr.entry(std::string(s));
				arr.end_array();
			}
			{
				auto arr =
				    obj.start_array("dest_scids");
				for (auto const& s : p->dest_scids)
					arr.entry(std::string(s));
				arr.end_array();
			}
			obj.field("amount_msat",
				  std::uint64_t(
				      p->amount.to_msat()));
			obj.field("maxfee_msat",
				  std::uint64_t(
				      p->maxfee.to_msat()));
			obj.field("maxparts", p->maxparts);
			obj.field("execute", p->execute);
			/* Echo the askrene response in full so the
			 * caller (and the spike harness) can inspect
			 * the planned routes, per-hop amounts, and
			 * probabilities. */
			obj.field("askrene", *routes);
			if (!p->execute) {
				obj.field("note",
					  std::string(
					      "plan-only stage: "
					      "sendpay not yet "
					      "implemented (Task "
					      "#87)"));
			}
			obj.end_object();
			return bus.raise(Msg::CommandResponse{
				id, std::move(plan)
			});
		});
	}

	Ev::Io<void> run_command(Jsmn::Object params, Ln::CommandId id) {
		auto p = std::make_shared<Params>();
		try {
			*p = parse_params(params);
		} catch (std::exception const& ex) {
			return bus.raise(Msg::CommandFail{
				id, RPC_INVALID_PARAMS,
				ex.what(),
				Json::Out::empty_object()
			});
		}

		return Boss::log( bus, Info
				, "XMoveFunds: planning %s -> %s, "
				  "amount=%" PRIu64 " msat, "
				  "maxfee=%" PRIu64 " msat, "
				  "maxparts=%" PRIu32 ", execute=%s"
				, join_scids(p->source_scids).c_str()
				, join_scids(p->dest_scids).c_str()
				, std::uint64_t(p->amount.to_msat())
				, std::uint64_t(p->maxfee.to_msat())
				, p->maxparts
				, p->execute ? "true" : "false"
				)
		     + wait_for_ready()
		     + do_plan(p, id);
	}

public:
	Impl(S::Bus& bus_)
		: bus(bus_), rpc(nullptr), layer_ready(false) {
		bus.subscribe<Msg::Init>([this](Msg::Init const& init) {
			rpc = &init.rpc;
			self_id = init.self_id;
			return Boss::concurrent(create_xrebalance_layer());
		});
		bus.subscribe<Msg::Manifestation
			     >([this](Msg::Manifestation const&) {
			return bus.raise(Msg::ManifestCommand{
				"clboss-xmovefunds",
				"source_scid(s) dest_scid(s) amount_msat "
				"[maxfee_msat] [maxparts] [execute]",
				"Manually move funds in a circular "
				"self-payment via askrene.  Each of "
				"source_scid and dest_scid may be either "
				"a single scid string (e.g. "
				"\"305607x10x0\") or a JSON array of "
				"scid strings (e.g. "
				"[\"305607x10x0\",\"305121x18x2\"]); "
				"the masking layer enables the us->peer "
				"direction of every listed source and "
				"the peer->us direction of every listed "
				"dest, then askrene's MCF distributes "
				"the flow.  Optional maxfee_msat defaults "
				"to 5000 ppm of amount_msat (floor 1000 "
				"msat); maxparts defaults to 10; execute "
				"defaults to false (plan-only).  Lowest-"
				"level primitive used by the xrebalance "
				"algorithm; the caller specifies the "
				"explicit channel set.",
				false
			});
		});
		bus.subscribe<Msg::CommandRequest
			     >([this](Msg::CommandRequest const& m) {
			if (m.command != "clboss-xmovefunds")
				return Ev::lift();
			return run_command(m.params, m.id);
		});
	}
};

Main::Main(Main&&) =default;
Main::~Main() =default;
Main::Main(S::Bus& bus_) : pimpl(Util::make_unique<Impl>(bus_)) { }

}}}