From 4596c75e07e4aedd4d93a432277ac31e7d3d107e Mon Sep 17 00:00:00 2001 From: Ken Sedgwick Date: Tue, 16 Jun 2026 09:48:44 -0700 Subject: [PATCH] XMoveFunds: bound the synchronous part wait; settle stragglers in the background A part whose HTLC is held in flight by some downstream node used to freeze the whole xrebalance executor: do_execute waited on every part's waitsendpay with no timeout, the XRebalancer driver awaits the cycle, and the driver runs one cycle at a time -- so a single stuck HTLC paused all cycles and all learning until it resolved (observed: a 633-sat part held 40+ minutes). There is no protocol way to cancel an in-flight HTLC; it resolves on its own (settle, or fail at its CLTV, up to ~100s of blocks away). Bound the wait instead. do_execute now waits synchronously for parts only up to clboss-xrebalance-part-wait-secs (new dynamic option, default 180) from the start of the wait phase, via CLN's native waitsendpay timeout. A part still in flight at the deadline comes back as error 200; rather than treating that as a routing failure (which would write a bogus capacity constraint and delpay a live part), we DETACH it: a background greenthread (background_wait_part via Boss::concurrent) re-waits it with no timeout and applies the SAME askrene learning and earnings attribution the foreground would have. Because the Claimer keeps each self-payment's claim entry for 24h, the detached part still settles and is accounted -- no accounting gap, nothing discarded -- while the driver loop is freed to start the next cycle. Why bounded-synchronous rather than fully async: the synchronous phase is what lets the next cycle re-plan on mostly-settled balances (it re-fetches listpeerchannels each cycle) and what produces the clboss-xmovefunds summary (delivered / fee / parts_complete) the manual RPC and the cycle log report. Attribution, askrene learning, delpay, and the transient mask-layer removal do not require the wait, so detaching a rare straggler costs only that one part's contribution to the next re-plan -- self-correcting, and reserve-protected against double-spend. The reply gains a parts_detached count; the XRebalancer cycle log notes "(N detached, settling in background)" so detached parts do not read as failures. Track A (FundsMover) is unaffected: it runs each attempt in its own Boss::concurrent Runner, so a stuck attempt blocks only that Runner, never the whole rebalancer, and its attribution already fires in the background when the part settles. --- Boss/Mod/XMoveFunds/Main.cpp | 296 ++++++++++++++++++++++++++++++++++- Boss/Mod/XRebalancer.cpp | 20 ++- 2 files changed, 304 insertions(+), 12 deletions(-) diff --git a/Boss/Mod/XMoveFunds/Main.cpp b/Boss/Mod/XMoveFunds/Main.cpp index f3fb869..c0b37d9 100644 --- a/Boss/Mod/XMoveFunds/Main.cpp +++ b/Boss/Mod/XMoveFunds/Main.cpp @@ -17,6 +17,7 @@ #include"Boss/concurrent.hpp" #include"Boss/log.hpp" #include"Ev/Io.hpp" +#include"Ev/now.hpp" #include"Ev/yield.hpp" #include"Jsmn/Object.hpp" #include"Json/Out.hpp" @@ -343,6 +344,17 @@ private: * via setconfig to keep accumulated capacity knowledge longer * before constraints expire. */ std::uint64_t aging_window_secs; + /* How long (seconds) do_execute waits SYNCHRONOUSLY for parts + * to terminate before it stops blocking the cycle. A part still + * in flight at this deadline is detached (background_wait_part): + * its waitsendpay+attribution continue in the background so it + * still settles and is accounted within the Claimer's 24h claim + * window, while the driver loop is freed to start the next cycle. + * This bounds a single stuck HTLC from freezing the whole + * xrebalance executor (no protocol way to cancel an in-flight + * HTLC; it resolves on its own). Dynamic via setconfig + * `clboss-xrebalance-part-wait-secs`. Default 180s. */ + double part_wait_secs; /* For generating MPP groupids -- a u64 random value shared * across all parts of one xmovefunds invocation. */ std::mt19937_64 rng; @@ -990,7 +1002,8 @@ private: Ev::Io waitsendpay_part(Sha256::Hash const& payment_hash, std::uint64_t partid, - std::uint64_t groupid) { + std::uint64_t groupid, + std::uint64_t timeout_secs = 0) { auto parms = Json::Out(); auto obj = parms.start_object(); obj.field("payment_hash", @@ -999,6 +1012,13 @@ private: obj.field("partid", partid); obj.field("groupid", groupid); } + /* timeout_secs > 0 bounds the synchronous wait: CLN returns + * error 200 ("Timed out") after this many seconds while + * leaving the HTLC in flight, which the caller detects and + * detaches. timeout_secs == 0 waits indefinitely (used by + * the detached background re-wait). */ + if (timeout_secs > 0) + obj.field("timeout", timeout_secs); obj.end_object(); return rpc->command("waitsendpay", std::move(parms)); } @@ -1637,6 +1657,131 @@ private: return rpc->command("getroutes", std::move(parms)); } + /* Run a batch of accumulated feedback actions in sequence. */ + Ev::Io + run_feedback(std::shared_ptr>> fb) { + return Ev::lift().then([fb]() { + auto act = Ev::lift(); + for (auto& a : *fb) + act = std::move(act) + std::move(a); + return act; + }); + } + + /* Push the positive-reinforcement feedback for a settled part into + * `sink`: inform-unconstrained every network middle hop at the + * amount it carried, plus a Success observation per hop. Shared by + * the foreground wait handler (sink = the cycle's deferred batch, + * drained once after the wait loop) and the detached background + * re-wait (sink = a local batch drained inline). */ + void + push_success_feedback( std::size_t i + , std::vector>> + const& per_part_middle + , std::vector>& sink + ) { + for (auto const& hop : per_part_middle[i]) { + sink.push_back( + Boss::Mod::AskreneLayer:: + inform_channel_unconstrained( + *rpc, + Boss::Mod::AskreneLayer:: + xrebalance_layer_name, + std::get<0>(hop), std::get<1>(hop), + std::get<2>(hop))); + auto obs = Msg::XRebalanceObservation{ + std::uint64_t(std::time(nullptr)), + std::get<0>(hop), std::get<1>(hop), + Msg::XRebalanceObservationKind::Success, + std::get<2>(hop), 0, Ln::NodeId()}; + sink.push_back(bus.raise(std::move(obs))); + } + } + + /* Re-wait (no timeout) a part that exceeded the synchronous wait + * budget and detached. Runs in a background greenthread via + * Boss::concurrent. When the part finally settles (within the + * Claimer's 24h claim window) or fails, it applies the SAME + * attribution / askrene learning the foreground path would have -- + * so a slow part is still accounted, with no gap, while the driver + * loop was freed to start the next cycle. Mirrors the foreground + * success/failure handlers, but drains its own feedback inline (the + * cycle's feedback batch was already flushed). */ + Ev::Io + background_wait_part( + std::shared_ptr payment_hash, + std::uint64_t partid, std::size_t i, + std::shared_ptr groupid_actual, + std::shared_ptr>>> per_part_middle, + std::shared_ptr> our_scids, + std::shared_ptr askrene_response) { + return Boss::log( bus, Info + , "XMoveFunds: part %zu (hash %s partid %" + PRIu64 ") still in flight at the %.0fs " + "synchronous wait budget; detaching to " + "settle in the background." + , i + , std::string(*payment_hash).substr(0, 16) + .c_str() + , partid + , part_wait_secs + ).then([ this, payment_hash, partid, i + , groupid_actual, per_part_middle + , our_scids, askrene_response + ]() { + return waitsendpay_part( *payment_hash, partid + , *groupid_actual) + .then([ this, i, per_part_middle, askrene_response + , payment_hash, partid + ](Jsmn::Object r) { + auto fb = std::make_shared< + std::vector>>(); + push_success_feedback(i, *per_part_middle, *fb); + auto path = (*askrene_response) + ["routes"][i]["path"]; + return run_feedback(fb) + + raise_attribution(path, r) + + Boss::log( bus, Info + , "XMoveFunds: detached part " + "%zu (hash %s partid %" PRIu64 + ") settled in the background." + , i + , std::string(*payment_hash) + .substr(0, 16).c_str() + , partid + ); + }).catching([ this, i, our_scids + , askrene_response + , payment_hash, partid + , groupid_actual + ](RpcError const& e) { + auto fb = std::make_shared< + std::vector>>(); + auto path = (*askrene_response) + ["routes"][i]["path"]; + accumulate_failure_feedback( + e, *our_scids, path, *fb); + return run_feedback(fb) + + delpay_part( *payment_hash, partid + , *groupid_actual) + + Boss::log( bus, Info + , "XMoveFunds: detached part " + "%zu (hash %s partid %" PRIu64 + ") failed in the background: " + "%s" + , i + , std::string(*payment_hash) + .substr(0, 16).c_str() + , partid + , failure_summary(e, path) + .c_str() + ); + }); + }); + } + /* Drive the sendpay + waitsendpay sequence for one or more * askrene-returned routes (multi-part for MPP). All parts * share payment_hash + payment_secret + groupid + label; @@ -1941,21 +2086,53 @@ private: * payment-critical path is not slowed by feedback RPCs * and to preserve a single coherent observation set * across MPP parts. */ + /* Synchronous-wait deadline. We block the cycle waiting for + * parts only up to part_wait_secs from when the wait phase + * begins (set at execute time below, after the sends). A part + * still in flight then is detached to settle in the background + * (background_wait_part), so one stuck HTLC cannot freeze the + * executor; num_detached feeds the reply. */ + auto deadline = std::make_shared(0.0); + auto num_detached = std::make_shared(0); + chain = std::move(chain) + Ev::lift().then([deadline, this]() { + *deadline = Ev::now() + part_wait_secs; + return Ev::lift(); + }); for (auto i = std::size_t(0); i < num_parts; ++i) { if (skipped[i]) continue; auto partid = multi ? (i + 1) : 0; chain = std::move(chain) + Ev::lift().then( - [this, payment_hash, partid, i, + [this, deadline, payment_hash, partid, i, groupid_actual, results, err_msgs, per_part_middle, our_scids, askrene_response, - feedback_actions] - () { + feedback_actions, num_detached] + () -> Ev::Io { + auto remaining = + *deadline - Ev::now(); + if (remaining <= 0.0) { + /* Budget already spent by + * earlier parts: detach this + * one without a foreground + * wait. */ + ++*num_detached; + results->push_back( + Jsmn::Object()); + return Boss::concurrent( + background_wait_part( + payment_hash, partid, + i, groupid_actual, + per_part_middle, + our_scids, + askrene_response)); + } return waitsendpay_part( *payment_hash, partid, - *groupid_actual) + *groupid_actual, + std::uint64_t(remaining) + + 1) .then([results, i, per_part_middle, feedback_actions, @@ -2025,8 +2202,39 @@ private: askrene_response, i, feedback_actions, payment_hash, partid, - groupid_actual, this] - (RpcError const& e) { + groupid_actual, + per_part_middle, + num_detached, this] + (RpcError const& e) + -> Ev::Io { + /* Timed out (code 200): the + * part is still in flight, not + * failed. Detach a background + * re-wait so it still settles + * and is attributed within the + * Claimer's 24h claim window; + * write no constraint and do + * NOT delpay (the HTLC is + * alive). Frees the cycle to + * proceed. */ + if (e.error.has("code") + && e.error["code"] + .is_number() + && int(double( + e.error["code"])) + == 200) { + ++*num_detached; + results->push_back( + Jsmn::Object()); + return Boss::concurrent( + background_wait_part( + payment_hash, + partid, i, + groupid_actual, + per_part_middle, + our_scids, + askrene_response)); + } results->push_back( Jsmn::Object()); /* Negative reinforcement: @@ -2090,7 +2298,7 @@ private: return std::move(chain).then( [payment_hash, preimage, label, groupid_actual, num_parts, num_skipped, - results, err_msgs]() { + num_detached, results, err_msgs]() { auto out = Json::Out(); auto obj = out.start_object(); obj.field("payment_hash", @@ -2106,6 +2314,12 @@ private: obj.field("parts", num_parts - num_skipped); if (num_skipped > 0) obj.field("parts_skipped", num_skipped); + /* Parts still in flight at the synchronous-wait + * deadline, detached to settle in the background. + * They are NOT yet in parts_complete; their + * attribution lands later via background_wait_part. */ + if (*num_detached > 0) + obj.field("parts_detached", *num_detached); /* Per-call summary across successful parts: * delivered_msat = sum r.amount_msat for parts @@ -2338,6 +2552,7 @@ public: , claimer(bus_) , layer_ready(false) , aging_window_secs(3600) + , part_wait_secs(180.0) , rng(static_cast( std::chrono::system_clock::now() .time_since_epoch().count())) { @@ -2393,6 +2608,24 @@ public: "Default 3600 (1h); operators on slower " "networks (signet) typically widen this.", /* dynamic = */ true + }) + bus.raise(Msg::ManifestOption{ + "clboss-xrebalance-part-wait-secs", + Msg::OptionType_Int, + Json::Out::direct( + std::uint64_t(part_wait_secs)), + "Seconds clboss-xmovefunds waits synchronously " + "for its parts to settle before returning and " + "letting the next cycle start. A part still in " + "flight at this deadline is detached: its " + "waitsendpay + earnings/askrene attribution " + "continue in the background, so it still " + "settles and is accounted within the claim " + "window -- one stuck HTLC no longer freezes the " + "xrebalance executor. Dynamic: settable at " + "runtime via `lightning-cli setconfig " + "clboss-xrebalance-part-wait-secs `. " + "Default 180 (3 minutes).", + /* dynamic = */ true }); }); bus.subscribe([this](Msg::Option const& o) { + if (o.name != "clboss-xrebalance-part-wait-secs") + return Ev::lift(); + /* Number at startup, string via setconfig -- same + * dual encoding as clboss-xrebalance-age-secs. */ + auto secs = double(0.0); + try { + if (o.value.is_number()) { + secs = double(o.value); + } else if (o.value.is_string()) { + secs = std::stod(std::string(o.value)); + } else { + return Boss::log( bus, Warn + , "XMoveFunds: clboss-" + "xrebalance-part-wait-" + "secs: unsupported " + "value type; keeping " + "%.0f." + , part_wait_secs + ); + } + } catch (std::exception const& e) { + return Boss::log( bus, Warn + , "XMoveFunds: clboss-" + "xrebalance-part-wait-secs: " + "parse error '%s'; keeping " + "%.0f." + , e.what() + , part_wait_secs + ); + } + if (secs <= 0.0) { + return Boss::log( bus, Warn + , "XMoveFunds: clboss-" + "xrebalance-part-wait-secs: " + "must be > 0; keeping %.0f." + , part_wait_secs + ); + } + part_wait_secs = secs; + return Boss::log( bus, Info + , "XMoveFunds: part synchronous-wait " + "budget = %.0f seconds" + , part_wait_secs + ); + }); bus.subscribe([this](Msg::TimerRandomHourly const&) { return wait_for_ready().then([this]() { diff --git a/Boss/Mod/XRebalancer.cpp b/Boss/Mod/XRebalancer.cpp index 6b39fa0..05025e3 100644 --- a/Boss/Mod/XRebalancer.cpp +++ b/Boss/Mod/XRebalancer.cpp @@ -904,6 +904,16 @@ private: << " ppm)"; ppm = os.str(); } + auto detached = std::string(); + { + auto nd = num("parts_detached"); + if (nd > 0.0) { + auto os = std::ostringstream(); + os << " (" << std::llround(nd) + << " detached, settling in background)"; + detached = os.str(); + } + } /* Chokepoint: among the failed parts, surface the one that got * CLOSEST to delivery -- the smallest from_target magnitude -- * because that frontier (how near the best attempt came, and @@ -958,7 +968,7 @@ private: * economics; reason is present only on a partial. */ return Boss::log( bus, Info , "XRebalancer: transfer done: %.0f/%.0f parts, " - "delivered %s msat, fee %s msat%s%s." + "delivered %s msat, fee %s msat%s%s%s." , num("parts_complete"), num("parts") , Util::Str::group_digits( std::int64_t(std::llround( @@ -966,12 +976,14 @@ private: , Util::Str::group_digits( std::int64_t(std::llround( fee))).c_str() - , ppm.c_str(), reason.c_str() ); + , ppm.c_str(), reason.c_str() + , detached.c_str() ); /* Nothing delivered: a clean failure -- show the part count * attempted and the chokepoint, not three zeros. */ return Boss::log( bus, Info - , "XRebalancer: transfer failed: %.0f part(s)%s." - , num("parts"), reason.c_str() ); + , "XRebalancer: transfer failed: %.0f part(s)%s%s." + , num("parts"), reason.c_str() + , detached.c_str() ); } static std::string join_scids(std::vector const& v) {