GlcoinCore/clboss
1
0
Fork 0
mirror of https://github.com/ZmnSCPxj/clboss.git synced 2026-08-13 12:33:20 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 5
clboss/Boss/Mod/AskreneUpdates.cpp
Ken Sedgwick c7fa9bd662
SetConfigHandler: fail setconfig when the owner rejects the value 
setconfig was acknowledged with unconditional success after
broadcasting Msg::Option, while owning modules reject bad values by
log-and-keep.  lightningd persists a setconfig value (configvar_save)
only on a success response, so the blanket ack recorded values clboss
never applied: listconfigs and config.setconfig diverged from the
running configuration, and a non-numeric value persisted for an
int-typed option fails lightningd's own option parse on the next
start -- lightningd refuses to boot.

Add a rejection back-channel to Msg::Option: SetConfigHandler
allocates a shared reject_reason (null on init-time raises, so
aggregate initialization at existing sites is unaffected), the owning
subscriber reports rejection via the new Msg::Option::reject() helper
(a no-op at init time, where quietly keeping the default is right),
and SetConfigHandler -- whose bus.raise() returns only after all
subscribers ran -- fails the command with invalid-params when a
reason was set.  All dynamic-option owners in this tree report their
rejections: RebalanceModeManager (unrecognized mode), FundsMover's
three numeric handlers, and AskreneUpdates' shared age/retain
handler.
2026-08-04 11:01:56 -07:00

806 lines
25 KiB
C++
Raw Permalink Blame History

#include"Boss/Mod/AskreneUpdates.hpp"
#include"Boss/Mod/AskreneLayer.hpp"
#include"Boss/Mod/Rpc.hpp"
#include"Boss/Msg/AskreneChannelUpdate.hpp"
#include"Boss/Msg/AskreneNodeDisableUpdate.hpp"
#include"Boss/Msg/CommandFail.hpp"
#include"Boss/Msg/CommandRequest.hpp"
#include"Boss/Msg/CommandResponse.hpp"
#include"Boss/Msg/DbResource.hpp"
#include"Boss/Msg/Init.hpp"
#include"Boss/Msg/ManifestCommand.hpp"
#include"Boss/Msg/ManifestOption.hpp"
#include"Boss/Msg/Manifestation.hpp"
#include"Boss/Msg/Option.hpp"
#include"Boss/Msg/OptionType.hpp"
#include"Boss/Msg/ProvideStatus.hpp"
#include"Boss/Msg/RequestAskreneUpdates.hpp"
#include"Boss/Msg/ResponseAskreneUpdates.hpp"
#include"Boss/Msg/SolicitStatus.hpp"
#include"Boss/Msg/TimerRandomHourly.hpp"
#include"Boss/concurrent.hpp"
#include"Boss/log.hpp"
#include"Ev/Io.hpp"
#include"Jsmn/Object.hpp"
#include"Json/Out.hpp"
#include"Ln/Amount.hpp"
#include"Ln/NodeId.hpp"
#include"Ln/Scid.hpp"
#include"S/Bus.hpp"
#include"Sqlite3.hpp"
#include"Util/make_unique.hpp"
#include"Uuid.hpp"
#include<cinttypes>
#include<cstdint>
#include<string>
#include<vector>
namespace {
/* How long after its last occurrence a learned update is still projected
* into the per-request layer. Separate knobs -- a down node and a
* re-priced channel may deserve different half-lives. */
auto constexpr default_node_disable_age_secs = std::uint64_t(3600); /* 1h */
auto constexpr default_channel_update_age_secs = std::uint64_t(3600); /* 1h */
/* How long a row survives in the log at all -- long, because the log
* doubles as a mineable history of what got disabled / re-priced. */
auto constexpr default_retain_secs = std::uint64_t(2592000); /* 30d */
}
namespace Boss { namespace Mod {
class AskreneUpdates::Impl {
private:
S::Bus& bus;
std::function<double()> get_now;
Sqlite3::Db db;
Boss::Mod::Rpc* rpc;
std::uint64_t node_disable_age_secs;
std::uint64_t channel_update_age_secs;
std::uint64_t retain_secs;
void start() {
bus.subscribe<Msg::DbResource
>([this](Msg::DbResource const& r) {
db = r.db;
return init();
});
bus.subscribe<Msg::Init
>([this](Msg::Init const& init) {
rpc = &init.rpc;
return Boss::concurrent(sweep_stale_layers());
});
bus.subscribe<Msg::Manifestation
>([this](Msg::Manifestation const&) {
return bus.raise(Msg::ManifestCommand{
"clboss-askrene-updates",
"[hours]",
"Show the learned askrene updates CLBOSS is "
"applying: the node disables and channel_update "
"overrides still within their projection window "
"(what a rebalance getroutes gets right now), "
"each with its age, occurrence count and -- for "
"channels -- the overridden policy. Optional "
"{hours} widens the view to the last {hours} "
"hours of the retained log, so aged-out entries "
"appear too (projected=false). Read-only.",
false
}) + bus.raise(Msg::ManifestOption{
"clboss-node-disable-age-secs",
Msg::OptionType_Int,
Json::Out::direct(default_node_disable_age_secs),
"How long (seconds) after the most recent "
"NODE-level routing failure CLBOSS keeps "
"disabling that node in rebalance route "
"searches. Once this elapses with no fresh "
"failure the node is no longer projected and "
"becomes routable again. Dynamic via "
"`lightning-cli setconfig`. Default 3600 (1h).",
/* dynamic = */ true
}) + bus.raise(Msg::ManifestOption{
"clboss-channel-update-age-secs",
Msg::OptionType_Int,
Json::Out::direct(default_channel_update_age_secs),
"How long (seconds) after the most recent "
"failure-learned channel_update CLBOSS keeps "
"applying that policy override (fees, htlc "
"bounds, enabled flag) in rebalance route "
"searches. Once this elapses with no fresh "
"update the channel reverts to gossip policy. "
"Dynamic via `lightning-cli setconfig`. "
"Default 3600 (1h).",
/* dynamic = */ true
}) + bus.raise(Msg::ManifestOption{
"clboss-update-retain-secs",
Msg::OptionType_Int,
Json::Out::direct(default_retain_secs),
"How long (seconds) learned node-disable and "
"channel-update rows are kept in the CLBOSS "
"database before pruning. Independent of the "
"projection windows above: the log is retained "
"well past when an update stops being applied, "
"so it can be mined (which nodes churn, which "
"channels re-price). Dynamic via `lightning-cli "
"setconfig`. Default 2592000 (30d).",
/* dynamic = */ true
});
});
bus.subscribe<Msg::Option
>([this](Msg::Option const& o) {
if (o.name == "clboss-node-disable-age-secs")
return handle_option( o, node_disable_age_secs
, "clboss-node-disable-age-secs");
if (o.name == "clboss-channel-update-age-secs")
return handle_option( o, channel_update_age_secs
, "clboss-channel-update-age-secs");
if (o.name == "clboss-update-retain-secs")
return handle_option( o, retain_secs
, "clboss-update-retain-secs");
return Ev::lift();
});
bus.subscribe<Msg::AskreneNodeDisableUpdate
>([this](Msg::AskreneNodeDisableUpdate const& m) {
if (!db)
return Ev::lift();
return record_node(m.node);
});
bus.subscribe<Msg::AskreneChannelUpdate
>([this](Msg::AskreneChannelUpdate const& m) {
if (!db)
return Ev::lift();
return record_channel(m);
});
bus.subscribe<Msg::RequestAskreneUpdates
>([this](Msg::RequestAskreneUpdates const& req) {
auto requester = req.requester;
if (!db)
return bus.raise(Msg::ResponseAskreneUpdates{
requester, {}, {}
});
return provide(requester);
});
bus.subscribe<Msg::TimerRandomHourly
>([this](Msg::TimerRandomHourly const&) {
if (!db)
return Ev::lift();
return prune();
});
bus.subscribe<Msg::SolicitStatus
>([this](Msg::SolicitStatus const&) {
if (!db)
return Ev::lift();
return status();
});
bus.subscribe<Msg::CommandRequest
>([this](Msg::CommandRequest const& req) {
if (req.command != "clboss-askrene-updates")
return Ev::lift();
if (!db)
return bus.raise(Msg::CommandResponse{
req.id, Json::Out::empty_object()
});
return report(req);
});
}
/* Parse and apply one *-age-secs / retain-secs option, tolerating
* both the number-at-startup and string-via-setconfig encodings. */
Ev::Io<void> handle_option( Msg::Option const& o
, std::uint64_t& target
, char const* name
) {
/* Signed so a negative value is rejected below rather than
* wrapping to a huge unsigned: std::stoull accepts a leading
* minus and negates modulo 2^64, and the double->uint64
* conversion of a negative is undefined. Matches the
* FundsMover option handlers. A wrapped-huge window would
* silently project every retained row into every rebalance
* layer. */
auto secs = std::int64_t(0);
try {
if (o.value.is_number()) {
secs = std::int64_t(double(o.value));
} else if (o.value.is_string()) {
secs = std::stoll(std::string(o.value));
} else {
o.reject( std::string(name)
+ ": unsupported value type");
return Boss::log( bus, Warn
, "AskreneUpdates: %s: "
"unsupported value type; "
"keeping %" PRIu64 "."
, name, target
);
}
} catch (std::exception const& e) {
o.reject( std::string(name)
+ ": not a valid number");
return Boss::log( bus, Warn
, "AskreneUpdates: %s: parse error "
"'%s'; keeping %" PRIu64 "."
, name, e.what(), target
);
}
if (secs <= 0) {
o.reject(std::string(name) + ": must be > 0");
return Boss::log( bus, Warn
, "AskreneUpdates: %s: must be > 0; "
"keeping %" PRIu64 "."
, name, target
);
}
target = std::uint64_t(secs);
return Boss::log( bus, Info
, "AskreneUpdates: %s = %" PRIu64 " seconds."
, name, target
);
}
Ev::Io<void> init() {
return db.transact().then([](Sqlite3::Tx tx) {
tx.query_execute(R"QRY(
CREATE TABLE IF NOT EXISTS "AskreneNodeDisableUpdates"
( time INTEGER NOT NULL -- unix seconds
, node TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS
idx_askrenenodedisableupdates_node_time
ON "AskreneNodeDisableUpdates" (node, time);
CREATE INDEX IF NOT EXISTS
idx_askrenenodedisableupdates_time
ON "AskreneNodeDisableUpdates" (time);
CREATE TABLE IF NOT EXISTS "AskreneChannelUpdates"
( time INTEGER NOT NULL -- unix seconds
, scid TEXT NOT NULL
, dir INTEGER NOT NULL -- askrene direction 0/1
, enabled INTEGER NOT NULL
, htlc_min_msat INTEGER NOT NULL
, htlc_max_msat INTEGER NOT NULL
, base_fee_msat INTEGER NOT NULL
, prop_fee_ppm INTEGER NOT NULL
, cltv_delta INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS
idx_askrenechannelupdates_scid_dir_time
ON "AskreneChannelUpdates" (scid, dir, time);
CREATE INDEX IF NOT EXISTS
idx_askrenechannelupdates_time
ON "AskreneChannelUpdates" (time);
)QRY");
tx.commit();
return Ev::lift();
});
}
/* Remove stale private per-request layers left over by a previous
* run. Runner (FundsMover) and XMoveFunds normally remove their
* clboss-updates-tmp-<uuid> layers when a request finishes, but an
* exception escaping mid-request skips those continuations -- in
* particular, a plugin stop with moves in flight throws
* Boss::Shutdown into every pending RPC, and cleanup at shutdown
* cannot work anyway since the Rpc module is already rejecting new
* commands by then. The layers are non-persistent, so a lightningd
* restart clears them; a clboss-only restart does not, and repeated
* redeploys accumulate junk in askrene-listlayers. Sweeping at
* init restores the invariant that only in-flight requests hold
* private layers. */
Ev::Io<void> sweep_stale_layers() {
return rpc->command( "askrene-listlayers"
, Json::Out::empty_object()
).then([this](Jsmn::Object res) {
auto stale = std::vector<std::string>();
try {
auto layers = res["layers"];
for (auto l : layers) {
auto name = std::string(l["layer"]);
if (name.rfind("clboss-updates-tmp-", 0) == 0)
stale.push_back(name);
}
} catch (std::exception const&) {
return Boss::log( bus, Warn
, "AskreneUpdates: unexpected "
"askrene-listlayers response; "
"skipping stale-layer sweep."
);
}
if (stale.empty())
return Ev::lift();
auto act = Ev::lift();
for (auto const& name : stale)
act = std::move(act)
+ AskreneUpdates::close_layer(*rpc, name);
return std::move(act)
+ Boss::log( bus, Info
, "AskreneUpdates: removed %zu stale "
"clboss-updates-tmp layer(s) left "
"by a previous run."
, stale.size()
);
}).catching<RpcError>([](RpcError const&) {
/* CLN without askrene: nothing to sweep. */
return Ev::lift();
});
}
Ev::Io<void> record_node(Ln::NodeId node) {
auto now = std::uint64_t(get_now());
auto node_s = std::string(node);
return db.transact().then([now, node_s](Sqlite3::Tx tx) {
tx.query(R"QRY(
INSERT INTO "AskreneNodeDisableUpdates"
VALUES(:time, :node);
)QRY")
.bind(":time", now)
.bind(":node", node_s)
.execute()
;
tx.commit();
return Ev::lift();
});
}
Ev::Io<void> record_channel(Msg::AskreneChannelUpdate cu) {
auto now = std::uint64_t(get_now());
return db.transact().then([now, cu](Sqlite3::Tx tx) {
tx.query(R"QRY(
INSERT INTO "AskreneChannelUpdates"
VALUES( :time, :scid, :dir, :enabled, :hmin, :hmax
, :base, :prop, :cltv);
)QRY")
.bind(":time", now)
.bind(":scid", std::string(cu.scid))
.bind(":dir", cu.direction)
.bind(":enabled", cu.enabled)
.bind(":hmin", cu.htlc_minimum_msat.to_msat())
.bind(":hmax", cu.htlc_maximum_msat.to_msat())
.bind(":base", cu.fee_base_msat.to_msat())
.bind(":prop", cu.fee_proportional_millionths)
.bind(":cltv", cu.cltv_expiry_delta)
.execute()
;
tx.commit();
return Ev::lift();
});
}
Ev::Io<void> provide(void* requester) {
auto now = std::uint64_t(get_now());
auto node_cutoff = (now > node_disable_age_secs)
? now - node_disable_age_secs : std::uint64_t(0);
auto chan_cutoff = (now > channel_update_age_secs)
? now - channel_update_age_secs : std::uint64_t(0);
return db.transact().then([this, requester, node_cutoff, chan_cutoff
](Sqlite3::Tx tx) {
auto resp = Msg::ResponseAskreneUpdates{requester, {}, {}};
/* INDEXED BY the bare (time) index: the planner
* otherwise picks the (node, time) index and
* full-scans it with time>= demoted to a per-row
* filter -- the whole retained table (30d) walked
* per projection instead of just the window (1h).
*/
auto nq = tx.query(R"QRY(
SELECT DISTINCT node
FROM "AskreneNodeDisableUpdates"
INDEXED BY idx_askrenenodedisableupdates_time
WHERE time >= :cutoff;
)QRY");
nq.bind(":cutoff", node_cutoff);
for (auto& r : nq.execute())
resp.node_disables.push_back(
Ln::NodeId(r.get<std::string>(0)));
/* Latest override per (scid, dir) still in window.
* sqlite fills the bare columns from the MAX(time)
* row of each group (preserved under INDEXED BY).
* The bare (time) index is forced for the same
* reason as the node query above: the planner's
* choice, (scid, dir, time), degrades to a
* full-index scan with time>= as a per-row filter.
*/
auto cq = tx.query(R"QRY(
SELECT scid, dir, enabled, htlc_min_msat, htlc_max_msat
, base_fee_msat, prop_fee_ppm, cltv_delta, MAX(time)
FROM "AskreneChannelUpdates"
INDEXED BY idx_askrenechannelupdates_time
WHERE time >= :cutoff
GROUP BY scid, dir;
)QRY");
cq.bind(":cutoff", chan_cutoff);
for (auto& r : cq.execute()) {
auto ndx = 0;
auto scid = r.get<std::string>(ndx++);
auto dir = r.get<std::uint32_t>(ndx++);
auto enabled = r.get<int>(ndx++);
auto hmin = r.get<std::uint64_t>(ndx++);
auto hmax = r.get<std::uint64_t>(ndx++);
auto base = r.get<std::uint64_t>(ndx++);
auto prop = r.get<std::uint64_t>(ndx++);
auto cltv = r.get<std::uint64_t>(ndx++);
resp.channel_updates.push_back(
Msg::AskreneChannelUpdate{
Ln::Scid(scid),
dir,
enabled != 0,
Ln::Amount::msat(hmin),
Ln::Amount::msat(hmax),
Ln::Amount::msat(base),
std::uint32_t(prop),
std::uint16_t(cltv)
});
}
tx.commit();
return bus.raise(std::move(resp));
});
}
Ev::Io<void> prune() {
auto now = std::uint64_t(get_now());
auto cutoff = (now > retain_secs) ? now - retain_secs
: std::uint64_t(0);
return db.transact().then([cutoff](Sqlite3::Tx tx) {
tx.query(R"QRY(
DELETE FROM "AskreneNodeDisableUpdates"
WHERE time < :cutoff;
)QRY")
.bind(":cutoff", cutoff)
.execute()
;
tx.query(R"QRY(
DELETE FROM "AskreneChannelUpdates"
WHERE time < :cutoff;
)QRY")
.bind(":cutoff", cutoff)
.execute()
;
tx.commit();
return Ev::lift();
});
}
/* clboss-status block: counts of what is stored and, within the
* projection windows, what is being applied right now. */
Ev::Io<void> status() {
auto now = std::uint64_t(get_now());
auto ncut = (now > node_disable_age_secs)
? now - node_disable_age_secs : std::uint64_t(0);
auto ccut = (now > channel_update_age_secs)
? now - channel_update_age_secs : std::uint64_t(0);
return db.transact().then([this, now, ncut, ccut
](Sqlite3::Tx tx) {
auto out = Json::Out();
auto obj = out.start_object();
/* Scalar subqueries instead of one aggregate pass:
* MIN/MAX become O(1) index seeks, the projected
* count reads only the window via the (time)
* index, and the distinct counts scan a covering
* index without materializing per-row values. The
* previous whole-table aggregate form cost
* clboss-status a full-table scan (with a string
* concatenation per row on the channel side). */
auto nf = tx.query(R"QRY(
SELECT (SELECT COUNT(*)
FROM "AskreneNodeDisableUpdates")
, (SELECT COUNT(*)
FROM (SELECT DISTINCT node
FROM "AskreneNodeDisableUpdates"))
, COALESCE((SELECT MIN(time)
FROM "AskreneNodeDisableUpdates"), 0)
, COALESCE((SELECT MAX(time)
FROM "AskreneNodeDisableUpdates"), 0)
, (SELECT COUNT(DISTINCT node)
FROM "AskreneNodeDisableUpdates"
INDEXED BY idx_askrenenodedisableupdates_time
WHERE time >= :cut);
)QRY");
nf.bind(":cut", ncut);
for (auto& r : nf.execute()) {
auto nd = obj.start_object("node_disables");
nd
.field("rows", r.get<std::uint64_t>(0))
.field( "distinct_nodes"
, r.get<std::uint64_t>(1))
.field( "projected_nodes"
, r.get<std::uint64_t>(4))
.field("window_secs", node_disable_age_secs)
.field( "oldest_time"
, r.get<std::uint64_t>(2))
.field( "newest_time"
, r.get<std::uint64_t>(3))
;
nd.end_object();
}
auto cf = tx.query(R"QRY(
SELECT (SELECT COUNT(*)
FROM "AskreneChannelUpdates")
, (SELECT COUNT(*)
FROM (SELECT DISTINCT scid, dir
FROM "AskreneChannelUpdates"))
, COALESCE((SELECT MIN(time)
FROM "AskreneChannelUpdates"), 0)
, COALESCE((SELECT MAX(time)
FROM "AskreneChannelUpdates"), 0)
, (SELECT COUNT(*)
FROM (SELECT DISTINCT scid, dir
FROM "AskreneChannelUpdates"
INDEXED BY idx_askrenechannelupdates_time
WHERE time >= :cut));
)QRY");
cf.bind(":cut", ccut);
for (auto& r : cf.execute()) {
auto cu = obj.start_object("channel_updates");
cu
.field("rows", r.get<std::uint64_t>(0))
.field( "distinct_channel_dirs"
, r.get<std::uint64_t>(1))
.field( "projected_channel_dirs"
, r.get<std::uint64_t>(4))
.field( "window_secs"
, channel_update_age_secs)
.field( "oldest_time"
, r.get<std::uint64_t>(2))
.field( "newest_time"
, r.get<std::uint64_t>(3))
;
cu.end_object();
}
obj.field("retain_secs", retain_secs);
obj.field("now", now);
obj.end_object();
tx.commit();
return bus.raise(Msg::ProvideStatus{
"askrene_updates", std::move(out)
});
});
}
/* clboss-askrene-updates command: list the updates being applied
* now (default) or, with {hours}, everything in the last {hours}
* hours of the retained log (with projected=false for aged-out
* rows). Node disables grouped per node; channel updates grouped
* per (scid, dir) with the latest overridden policy. */
Ev::Io<void> report(Msg::CommandRequest const& req) {
auto id = req.id;
auto paramfail = [this, id]() {
return bus.raise(Msg::CommandFail{
id, -32602, "Parameter failure",
Json::Out::empty_object()
});
};
auto hours = double(0.0);
auto hours_j = Jsmn::Object();
auto params = req.params;
if (params.is_object()) {
auto known = std::size_t(0);
if (params.has("hours")) {
hours_j = params["hours"];
++known;
}
if (params.size() != known)
return paramfail();
} else if (params.is_array()) {
if (params.size() > 1)
return paramfail();
for (auto p : params)
hours_j = p;
}
if (!hours_j.is_null()) {
if (!hours_j.is_number())
return paramfail();
hours = double(hours_j);
if (hours <= 0)
return paramfail();
/* Clamp to the retention window: beyond it a
* wider view adds nothing (rows are pruned), and
* an unbounded value would make the
* double->uint64 conversion of hours*3600 below
* undefined. Clamping rather than failing --
* "everything retained" is the obvious intent of
* a huge value. */
if (hours * 3600.0 > double(retain_secs))
hours = double(retain_secs) / 3600.0;
}
auto now = std::uint64_t(get_now());
/* With {hours}, both kinds use that window; otherwise each
* uses its own projection window (the applied-now view). */
auto ncut = std::uint64_t(0);
auto ccut = std::uint64_t(0);
if (hours > 0) {
auto w = std::uint64_t(hours * 3600.0);
ncut = (w < now) ? now - w : std::uint64_t(0);
ccut = ncut;
} else {
ncut = (now > node_disable_age_secs)
? now - node_disable_age_secs : std::uint64_t(0);
ccut = (now > channel_update_age_secs)
? now - channel_update_age_secs : std::uint64_t(0);
}
return db.transact().then([this, id, now, ncut, ccut
](Sqlite3::Tx tx) {
auto out = Json::Out();
auto obj = out.start_object();
auto nf = tx.query(R"QRY(
SELECT node, COUNT(*), MAX(time)
FROM "AskreneNodeDisableUpdates"
WHERE time >= :cut
GROUP BY node
ORDER BY MAX(time) DESC;
)QRY");
nf.bind(":cut", ncut);
auto narr = obj.start_array("node_disables");
for (auto& r : nf.execute()) {
auto node = r.get<std::string>(0);
auto occ = r.get<std::uint64_t>(1);
auto last = r.get<std::uint64_t>(2);
auto age = (now >= last) ? now - last
: std::uint64_t(0);
auto o = narr.start_object();
o
.field("node", node)
.field("occurrences", occ)
.field("last_time", last)
.field("age_secs", age)
.field( "projected"
, age <= node_disable_age_secs)
;
o.end_object();
}
narr.end_array();
auto cf = tx.query(R"QRY(
SELECT scid, dir, enabled, htlc_min_msat, htlc_max_msat
, base_fee_msat, prop_fee_ppm, cltv_delta
, COUNT(*), MAX(time)
FROM "AskreneChannelUpdates"
WHERE time >= :cut
GROUP BY scid, dir
ORDER BY MAX(time) DESC;
)QRY");
cf.bind(":cut", ccut);
auto carr = obj.start_array("channel_updates");
for (auto& r : cf.execute()) {
auto ndx = 0;
auto scid = r.get<std::string>(ndx++);
auto dir = r.get<std::uint32_t>(ndx++);
auto enabled = r.get<int>(ndx++);
auto hmin = r.get<std::uint64_t>(ndx++);
auto hmax = r.get<std::uint64_t>(ndx++);
auto base = r.get<std::uint64_t>(ndx++);
auto prop = r.get<std::uint64_t>(ndx++);
auto cltv = r.get<std::uint64_t>(ndx++);
auto occ = r.get<std::uint64_t>(ndx++);
auto last = r.get<std::uint64_t>(ndx++);
auto age = (now >= last) ? now - last
: std::uint64_t(0);
auto o = carr.start_object();
o
.field("scid", scid)
.field("dir", dir)
.field("enabled", enabled != 0)
.field("htlc_min_msat", hmin)
.field("htlc_max_msat", hmax)
.field("base_fee_msat", base)
.field("prop_fee_ppm", prop)
.field("cltv_delta", cltv)
.field("occurrences", occ)
.field("last_time", last)
.field("age_secs", age)
.field( "projected"
, age <= channel_update_age_secs)
;
o.end_object();
}
carr.end_array();
obj.field("now", now);
obj.end_object();
tx.commit();
return bus.raise(Msg::CommandResponse{
id, std::move(out)
});
});
}
public:
Impl() =delete;
Impl(Impl&&) =delete;
Impl(Impl const&) =delete;
explicit
Impl(S::Bus& bus_, std::function<double()> get_now_)
: bus(bus_)
, get_now(std::move(get_now_))
, rpc(nullptr)
, node_disable_age_secs(default_node_disable_age_secs)
, channel_update_age_secs(default_channel_update_age_secs)
, retain_secs(default_retain_secs) { start(); }
};
/* ~~~~ static projection helpers ~~~~ */
Ev::Io<std::string>
AskreneUpdates::open_layer( Boss::Mod::Rpc& rpc
, Boss::Msg::ResponseAskreneUpdates const& updates
) {
auto layer = std::string("clboss-updates-tmp-")
+ std::string(Uuid::random());
auto parms = Json::Out()
.start_object()
.field("layer", layer)
.field("persistent", false)
.end_object()
;
return rpc.command( "askrene-create-layer"
, std::move(parms)
).then([&rpc, layer, updates](Jsmn::Object) {
auto chain = Ev::lift();
for (auto const& node : updates.node_disables)
chain = std::move(chain)
+ Boss::Mod::AskreneLayer::disable_node(
rpc, layer, node);
for (auto const& cu : updates.channel_updates)
chain = std::move(chain)
+ Boss::Mod::AskreneLayer::update_channel(
rpc, layer,
cu.scid, cu.direction, cu.enabled,
cu.htlc_minimum_msat,
cu.htlc_maximum_msat,
cu.fee_base_msat,
cu.fee_proportional_millionths,
cu.cltv_expiry_delta);
return std::move(chain).then([layer]() {
return Ev::lift(layer);
});
}).catching<RpcError>([](RpcError const&) {
/* create-layer failed (e.g. CLN < v24.11 has no askrene):
* return an empty name so the caller omits it from its
* getroutes layers array rather than naming a layer that
* does not exist -- the exact condition that aborts the
* (important) askrene plugin. Degraded (no update
* projection this call), never a crash. */
return Ev::lift(std::string());
});
}
Ev::Io<void>
AskreneUpdates::close_layer( Boss::Mod::Rpc& rpc
, std::string const& 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>([](RpcError const&) {
/* Best-effort: the layer is non-persistent and private to
* this finished request; a failed remove leaks nothing that
* a restart would not clear. */
return Ev::lift();
});
}
AskreneUpdates::AskreneUpdates(AskreneUpdates&&) =default;
AskreneUpdates::~AskreneUpdates() =default;
AskreneUpdates::AskreneUpdates( S::Bus& bus
, std::function<double()> get_now_
)
: pimpl(Util::make_unique<Impl>(bus, get_now_)) { }
}}
Reference in a new issue View git blame Copy permalink