mirror of
https://github.com/ZmnSCPxj/clboss.git
synced 2026-08-13 12:33:20 +02:00
AskreneUpdates: clboss-status block + clboss-askrene-updates command
Surface the learned-updates store for observation. Add an askrene_updates clboss-status block with row counts, distinct nodes / channel-dirs, and the counts currently within the projection windows (what a getroutes gets now), plus the oldest/newest timestamps and the retention horizon. Add a read-only clboss-askrene-updates command: with no argument it lists the updates being applied now -- node disables grouped per node and channel-update overrides grouped per (scid, dir) with the latest policy -- each with its age, occurrence count and a projected flag. An optional hours argument widens the view to the last N hours of the retained log so aged-out rows appear too (projected=false), for mining which nodes churn and which channels re-price. Read-only; no writes to any layer or table.
This commit is contained in:
Ken Sedgwick
parent
89346c7ce0
commit
1e3fdbb3e4
1 changed files with 254 additions and 1 deletions
|
|
@ -3,13 +3,19 @@
|
|||
#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/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/log.hpp"
|
||||
#include"Ev/Io.hpp"
|
||||
|
|
@ -59,7 +65,20 @@ private:
|
|||
});
|
||||
bus.subscribe<Msg::Manifestation
|
||||
>([this](Msg::Manifestation const&) {
|
||||
return bus.raise(Msg::ManifestOption{
|
||||
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),
|
||||
|
|
@ -139,6 +158,22 @@ private:
|
|||
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
|
||||
|
|
@ -341,6 +376,224 @@ private:
|
|||
});
|
||||
}
|
||||
|
||||
/* 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();
|
||||
|
||||
auto nf = tx.query(R"QRY(
|
||||
SELECT COUNT(*), COUNT(DISTINCT node)
|
||||
, COALESCE(MIN(time), 0), COALESCE(MAX(time), 0)
|
||||
, COUNT(DISTINCT CASE WHEN time >= :cut
|
||||
THEN node END)
|
||||
FROM "AskreneNodeDisableUpdates";
|
||||
)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 COUNT(*), COUNT(DISTINCT scid || '/' || dir)
|
||||
, COALESCE(MIN(time), 0), COALESCE(MAX(time), 0)
|
||||
, COUNT(DISTINCT CASE WHEN time >= :cut
|
||||
THEN scid || '/' || dir END)
|
||||
FROM "AskreneChannelUpdates";
|
||||
)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();
|
||||
}
|
||||
|
||||
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;
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue