AskreneLayer: coalesce redundant inform-channel writes per channel-direction

askrene's get_constraints folds every constraint for a (layer, scid-dir) down
to a single tightest [min, max] at query time, so the per-HTLC stream of
inform-channel writes only bloats the layer. Hot rebalance corridors accreted
hundreds-to-thousands of dominated min_msat copies -- one survey found a single
channel-direction holding 5026 constraints -- every one of which askrene must
re-fold on each getroutes through that dir.

Coalesce at the inform_channel chokepoint: per (layer, scid-dir, inform-kind)
keep the tightest bound emitted in the current time bucket and write through
only on a new bucket (a keep-alive against the layer aging) or a tightening.
Dropping a dominated write is lossless -- it is exactly the entry
get_constraints discards when it folds. On the surveyed hot set this is a ~42x
depth reduction.

The bucket length is a fixed fraction (1/12) of the layer aging window
(clboss-classic-layer-age-secs), so it always stays well under the aging window
-- the once-per-bucket keep-alive refreshes a constraint before it can age out
-- and the aging window is then always exactly 12 buckets whatever its value,
making the prune and the depth floor scale-invariant. 30 minutes at the default
6h aging. FundsMover feeds the live value through set_aging_window_secs from its
clboss-classic-layer-age-secs handler, so a setconfig retunes the bucket
immediately.

A small amortized prune drops cache entries idle past the aging window so the
cache does not grow with the set of channel-directions seen over the process
lifetime. inform_coalesce_emit is factored out as a pure decision; the test
covers the dominance and oscillation cases plus a behavioural drop test.

Boss/Mod/AskreneLayer.cpp

@@ -1,10 +1,13 @@
 #include"Boss/Mod/AskreneLayer.hpp"
 #include"Boss/Mod/Rpc.hpp"
 #include"Ev/Io.hpp"
+#include"Ev/now.hpp"
 #include"Jsmn/Object.hpp"
 #include"Json/Out.hpp"
 #include"Util/stringify.hpp"
 #include<assert.h>
+#include<cstdint>
+#include<map>
 
 namespace Boss { namespace Mod { namespace AskreneLayer {
 
@@ -12,6 +15,42 @@ std::string const clboss_layer_name = "clboss";
 
 namespace {
 
+/* Coalescing state, keyed by "layer|scid-dir|inform-kind" -> the
+ * tightest bound emitted in the current time bucket (see InformObs in
+ * the header).  Safe as a file-static: the whole plugin runs on a
+ * single Ev event-loop thread, so there is no concurrent access. */
+std::map<std::string, InformObs> inform_cache;
+
+/* The coalescing bucket length, derived as a fixed fraction
+ * (1 / coalesce_window_divisor) of the layer aging window
+ * (clboss-classic-layer-age-secs).  Making it a fraction of the aging
+ * window means the keep-alive re-emit (once per bucket) always refreshes
+ * a constraint well before it can age out, and the aging window is
+ * always exactly coalesce_window_divisor buckets long whatever its
+ * value -- so the prune and the depth floor are scale-invariant.
+ * FundsMover feeds the live aging value via set_aging_window_secs();
+ * this default matches aging/12 at the default 6h aging (30-min bucket). */
+std::uint64_t constexpr coalesce_window_divisor = 12;
+double coalesce_window_secs = 21600.0 / double(coalesce_window_divisor);
+
+/* Drop coalescing entries idle past the aging window, so the cache does
+ * not grow with the set of channel-dirs seen over the process lifetime.
+ * Amortised -- swept once per 4096 emits, not per call. */
+void prune_inform_cache(std::uint64_t now_bucket) {
+	static std::uint64_t emits = 0;
+	if ((++emits & 0xFFF) != 0)
+		return;
+	/* The aging window is always coalesce_window_divisor buckets, so a
+	 * few more than that covers any still-active dir at any aging value. */
+	auto constexpr keep_buckets = std::uint64_t(coalesce_window_divisor + 4);
+	for (auto it = inform_cache.begin(); it != inform_cache.end(); ) {
+		if (it->second.bucket + keep_buckets < now_bucket)
+			it = inform_cache.erase(it);
+		else
+			++it;
+	}
+}
+
 /* Common machinery for the two inform_channel variants.  askrene
  * accepts inform=succeeded / constrained / unconstrained as the
  * only difference between them; everything else (scid_dir,
@@ -38,6 +77,27 @@ inform_channel( Boss::Mod::Rpc& rpc
 	if (direction > 1)
 		return Ev::lift();
 	auto sdir = std::string(scid) + "/" + Util::stringify(direction);
+
+	/* Coalesce redundant writes (see InformObs in the header): keep the
+	 * tightest bound per (layer, scid-dir, kind) within one aging-derived
+	 * bucket and emit only on a new bucket (keep-alive against the layer
+	 * aging) or a tightening.  Dropping a dominated write is lossless --
+	 * get_constraints folds the dir down to one tightest [min,max], so the
+	 * dropped entry would not have changed any route. */
+	auto const bucket = std::uint64_t(Ev::now() / coalesce_window_secs);
+	auto const is_lower_bound = (std::string(inform) != "constrained");
+	auto const key = layer + "|" + sdir + "|" + inform;
+	auto const cache_it = inform_cache.find(key);
+	auto const* prior = (cache_it == inform_cache.end())
+			  ? nullptr : &cache_it->second;
+	if (!inform_coalesce_emit( prior, bucket
+				 , std::uint64_t(amount.to_msat())
+				 , is_lower_bound
+				 ))
+		return Ev::lift();
+	inform_cache[key] = InformObs{ bucket, std::uint64_t(amount.to_msat()) };
+	prune_inform_cache(bucket);
+
 	auto parms = Json::Out()
 		.start_object()
 			.field("layer", layer)
@@ -63,6 +123,34 @@ inform_channel( Boss::Mod::Rpc& rpc
 
 }
 
+/* The coalescing decision (see InformObs in the header).  Defined out
+ * here rather than in the anonymous namespace so the unit test can call
+ * it directly; inform_channel reaches it via the header declaration. */
+bool
+inform_coalesce_emit( InformObs const* prior
+		    , std::uint64_t bucket
+		    , std::uint64_t amount_msat
+		    , bool is_lower_bound
+		    ) {
+	if (!prior)
+		return true;                 /* first observation for this key */
+	if (prior->bucket != bucket)
+		return true;                 /* new bucket: keep-alive emit */
+	/* Same bucket: emit only if this observation tightens the bound. */
+	return is_lower_bound ? amount_msat > prior->tightest_msat
+			      : amount_msat < prior->tightest_msat;
+}
+
+/* Set the coalescing bucket length from the current layer aging window
+ * (clboss-classic-layer-age-secs), as aging / coalesce_window_divisor.
+ * Called by FundsMover when that option loads or changes, so the
+ * coalescing window tracks the aging window live.  Floored at 1s so a
+ * pathological aging value can never produce a zero-length bucket. */
+void set_aging_window_secs(std::uint64_t aging_secs) {
+	auto const w = double(aging_secs) / double(coalesce_window_divisor);
+	coalesce_window_secs = (w >= 1.0) ? w : 1.0;
+}
+
 Ev::Io<void>
 inform_channel_constrained( Boss::Mod::Rpc& rpc
 			  , std::string const& layer

Boss/Mod/AskreneLayer.hpp

@@ -129,6 +129,52 @@ Ev::Io<void> update_channel( Boss::Mod::Rpc& rpc
 			   , std::uint16_t cltv_expiry_delta
 			   );
 
+/* --- inform-channel write coalescing -------------------------------
+ *
+ * askrene's get_constraints folds every constraint for a given
+ * (layer, scid-dir) down to a single tightest [min, max] at query
+ * time, so a stream of repeated inform-channel writes for the same dir
+ * only bloats the layer: hot rebalance corridors accreted hundreds-to-
+ * thousands of dominated min_msat copies, every one of which askrene
+ * must re-fold on each getroutes through that dir.  inform_channel
+ * coalesces them -- per (layer, scid-dir, inform-kind) it keeps the
+ * tightest bound seen in the current time bucket and emits a write only
+ * when a new bucket opens (a keep-alive against the layer aging) or the
+ * bound tightens.  Dropping a dominated write is lossless: it is exactly
+ * what get_constraints discards.
+ *
+ * InformObs is the per-key state; inform_coalesce_emit is the pure
+ * decision, exposed here so it can be unit-tested directly.
+ */
+struct InformObs {
+	/* Time bucket (Ev::now() / window) the tightest bound was last
+	 * emitted in. */
+	std::uint64_t bucket;
+	/* That tightest bound, in msat: the highest min for a lower-bound
+	 * (unconstrained) kind, the lowest max for an upper-bound
+	 * (constrained) kind. */
+	std::uint64_t tightest_msat;
+};
+
+/* Decide whether a new observation should be written through to
+ * askrene.  prior is the last emitted state for this
+ * (layer, scid-dir, kind), or nullptr if none.  is_lower_bound is true
+ * for the unconstrained/min kind, false for the constrained/max kind.
+ * Emit on no prior, on a new bucket, or on a tightening within the
+ * same bucket. */
+bool inform_coalesce_emit( InformObs const* prior
+			 , std::uint64_t bucket
+			 , std::uint64_t amount_msat
+			 , bool is_lower_bound
+			 );
+
+/* Set the inform-channel coalescing bucket length from the layer aging
+ * window: the bucket is a fixed fraction of aging_secs, so the
+ * once-per-bucket keep-alive always refreshes a constraint before it
+ * ages out.  Wired from FundsMover's clboss-classic-layer-age-secs
+ * handler so the window tracks the aging window live. */
+void set_aging_window_secs(std::uint64_t aging_secs);
+
 }}}
 
 #endif /* !defined(BOSS_MOD_ASKRENELAYER_HPP_) */

Boss/Mod/FundsMover/Main.cpp

@@ -178,6 +178,12 @@ private:
 						);
 			}
 			aging_window_secs = std::uint64_t(secs);
+			/* Keep the AskreneLayer inform-coalescing bucket a
+			 * fixed fraction of the aging window (see
+			 * set_aging_window_secs); tracks setconfig live. */
+			Boss::Mod::AskreneLayer::set_aging_window_secs(
+				aging_window_secs
+			);
 			return Boss::log( bus, Info
 					, "FundsMover: clboss layer aging "
 					  "window = %" PRIu64 " seconds"

tests/boss/test_askrene_layer.cpp

@@ -275,9 +275,102 @@ test_silent_rpc_error( MockRpcServer& server
 	});
 }
 
+/* Test 5 (pure): the write-coalescing decision (inform_coalesce_emit).
+ * No RPC -- exercises the dominance + bucket rule directly. */
+void test_coalesce_decision() {
+	using Boss::Mod::AskreneLayer::InformObs;
+	using Boss::Mod::AskreneLayer::inform_coalesce_emit;
+
+	/* No prior: always emit, either kind. */
+	assert(inform_coalesce_emit(nullptr, 100, 500, true));
+	assert(inform_coalesce_emit(nullptr, 100, 500, false));
+
+	auto const prior = InformObs{ 100, 500 };
+
+	/* New bucket: emit even when not tighter (keep-alive vs aging). */
+	assert(inform_coalesce_emit(&prior, 101, 500, true));
+	assert(inform_coalesce_emit(&prior, 101, 10, true));
+
+	/* Same bucket, lower bound (min): emit iff strictly higher. */
+	assert( inform_coalesce_emit(&prior, 100, 600, true));
+	assert(!inform_coalesce_emit(&prior, 100, 500, true));
+	assert(!inform_coalesce_emit(&prior, 100, 400, true));
+
+	/* Same bucket, upper bound (max): emit iff strictly lower. */
+	assert( inform_coalesce_emit(&prior, 100, 400, false));
+	assert(!inform_coalesce_emit(&prior, 100, 500, false));
+	assert(!inform_coalesce_emit(&prior, 100, 600, false));
+
+	/* Oscillation within a bucket cannot defeat dominance: with 600 the
+	 * running min, nothing at/below it re-emits, whatever the order. */
+	auto const osc = InformObs{ 100, 600 };
+	assert(!inform_coalesce_emit(&osc, 100, 500, true));
+	assert(!inform_coalesce_emit(&osc, 100, 400, true));
+	assert(!inform_coalesce_emit(&osc, 100, 600, true));
+}
+
+/* Test 6 (behavioural): inform_channel coalesces.  A second, dominated
+ * write to the same dir within the same time bucket must NOT hit the
+ * RPC.  Proven via a sentinel inform to a different dir: if the
+ * dominated write leaked through, the second server read would see it
+ * (and its dir assertion would fire) instead of the sentinel. */
+Ev::Io<void>
+test_coalesce_drops_dominated( MockRpcServer& server
+			     , Boss::Mod::Rpc& rpc
+			     ) {
+	auto const layer = std::string("coalesce-layer");
+
+	auto assert_first = [](Jsmn::Object const& req) {
+		auto id = assert_method(req, "askrene-inform-channel");
+		auto params = req["params"];
+		assert(std::string(params["short_channel_id_dir"]) == "400x4x0/0");
+		assert(double(params["amount_msat"]) == 750000.0);
+		return id;
+	};
+	auto assert_sentinel = [](Jsmn::Object const& req) {
+		auto id = assert_method(req, "askrene-inform-channel");
+		auto params = req["params"];
+		/* A leaked dominated 400x4x0/0 write would show up here
+		 * instead of the sentinel -> this assertion fires. */
+		assert(std::string(params["short_channel_id_dir"]) == "500x5x0/0");
+		assert(double(params["amount_msat"]) == 999.0);
+		return id;
+	};
+
+	return Ev::lift().then([&server, assert_first]() {
+		return Ev::concurrent(server.serve_ok(assert_first));
+	}).then([&rpc, layer]() {
+		/* First write: new dir+bucket -> emits, served by assert_first. */
+		return Boss::Mod::AskreneLayer::inform_channel_unconstrained(
+			rpc, layer, Ln::Scid("400x4x0"), std::uint32_t(0),
+			Ln::Amount::msat(750000)
+		);
+	}).then([&server, assert_sentinel]() {
+		/* Spawn the sentinel server BEFORE the dominated write, so a
+		 * leaked write is caught by assert_sentinel rather than
+		 * hanging the test. */
+		return Ev::concurrent(server.serve_ok(assert_sentinel));
+	}).then([&rpc, layer]() {
+		/* Dominated (lower min, same bucket) -> must DROP, no RPC. */
+		return Boss::Mod::AskreneLayer::inform_channel_unconstrained(
+			rpc, layer, Ln::Scid("400x4x0"), std::uint32_t(0),
+			Ln::Amount::msat(500000)
+		);
+	}).then([&rpc, layer]() {
+		/* Sentinel: different dir -> emits, served by assert_sentinel. */
+		return Boss::Mod::AskreneLayer::inform_channel_unconstrained(
+			rpc, layer, Ln::Scid("500x5x0"), std::uint32_t(0),
+			Ln::Amount::msat(999)
+		);
+	});
+}
+
 } // namespace
 
 int main() {
+	/* Pure-logic coalescing test first -- needs no Ev/RPC machinery. */
+	test_coalesce_decision();
+
 	auto bus = S::Bus();
 
 	int sockets[2];
@@ -297,6 +390,8 @@ int main() {
 		return test_disable_node(server, rpc);
 	}).then([&]() {
 		return test_silent_rpc_error(server, rpc);
+	}).then([&]() {
+		return test_coalesce_drops_dominated(server, rpc);
 	}).then([&]() {
 		/* All tests passed; raise Shutdown so concurrent
 		 * server tasks (if any are still alive) and the