GlcoinCore/clboss
1
0
Fork 0
mirror of https://github.com/ZmnSCPxj/clboss.git synced 2026-08-15 12:50:42 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

PeerComplaintsDesk: defer auto-close while the peer is offline
Some checks failed
Code Base Sanity Check / tests (push) Has been cancelled
Details
Code Base Sanity Check / coverage (push) Has been cancelled
Details
Code Base Sanity Check / build-clang (push) Has been cancelled
Details

Browse source 
close was issued with unilateraltimeout=180 on a fixed timer,
without regard to the peer's connection state, while one complaint
source (ComplainerByLowConnectRate) selects peers specifically for
a low connect rate.  Closing while the peer is offline escalates
to a unilateral force-close after 3 minutes, against exactly the
peers least likely to negotiate a mutual close in time.

Check the peer's channels for a live connection (listpeerchannels
peer_connected) before issuing close, and defer while the peer is
offline.  Poll close candidates every 10 minutes rather than once
per solicitation cycle, so a flaky peer's brief online windows are
actually caught.  If the peer stays offline for close_patience
(3 days), close anyway and let the short unilateral timeout
escalate; the first-deferred time is persisted in a new
PeerComplaintsDesk_closepending table so restarts do not reset
the patience window.  The fees_low gate applies only to that
unilateral path: a mutual close even at high feerates is cheaper
than a unilateral at low feerates, so connected peers are closed
immediately regardless of feerate.

Add tests/boss/test_peercomplaintsdesk_main.cpp covering the close
paths: a connected peer closes immediately; an offline peer defers
and the deferral survives a restart; within patience it holds;
expired patience holds at high fees and closes at low fees;
dropping below the complaint threshold sweeps the deferral;
channel destruction clears it; disabled auto-close closes nothing.
The test drives the module over the bus with a mock CLN on a
socketpair.

Reported by an external security researcher via private disclosure.

Fixes #324
This commit is contained in:
Ken Sedgwick 2026-08-13 13:18:04 -07:00
parent 583924184c
commit 986f30d1b4
No known key found for this signature in database
GPG key ID: DBD2AF0849D711A9
7 changed files with 682 additions and 100 deletions
Download patch file Download diff file Expand all files Collapse all files

24
Boss/Mod/JitRebalancer.cpp
View file

@ -31,7 +31,6 @@
#include"Util/make_unique.hpp"
#include"Util/stringify.hpp"
#include<map>
#include<set>
namespace {
@ -111,14 +110,6 @@ private:
ModG::RebalanceUnmanagerProxy unmanager;
std::uint32_t max_rebalance_fee_ppm;
/* Nodes with a JIT rebalance currently in flight.
* The budget check reads expenditures that are only
* persisted once a rebalance completes, so concurrent
* rebalances to the same destination would each
* authorize against the same stale budget.
*/
std::set<Ln::NodeId> in_flight;
void start() {
max_rebalance_fee_ppm = default_max_rebalance_fee_ppm;
@ -195,18 +186,6 @@ private:
return Ev::lift(false);
});
}
if (in_flight.count(node) != 0) {
return Boss::log( bus, Debug
, "JitRebalancer: HTLC %s to "
"%s: rebalance already in "
"flight, will ignore."
, stringify_cid(id).c_str()
, Util::stringify(node).c_str()
).then([]() {
return Ev::lift(false);
});
}
in_flight.insert(node);
return Boss::concurrent( check_and_move(node, amount, id)
).then([]() {
return Ev::lift(true);
@ -249,9 +228,6 @@ private:
, unmanager, max_rebalance_fee_ppm
);
return r.execute();
}).then([this, node]() {
in_flight.erase(node);
return Ev::lift();
});
}

150
Boss/Mod/PeerComplaintsDesk/Main.cpp
View file

@ -21,12 +21,16 @@
#include"Boss/Msg/TimerTwiceDaily.hpp"
#include"Boss/concurrent.hpp"
#include"Boss/log.hpp"
#include"Ev/now.hpp"
#include"Jsmn/Object.hpp"
#include"Json/Out.hpp"
#include"S/Bus.hpp"
#include"Sqlite3.hpp"
#include"Util/make_unique.hpp"
#include"Util/stringify.hpp"
#include<algorithm>
#include<assert.h>
#include<memory>
namespace {
@ -46,6 +50,12 @@ auto constexpr channel_close_timeout = std::size_t(3 * 60);
/* How strongly do we insist on our own feerate. */
auto const fee_negotiation_step = std::string("1");
/* How long we wait for an offline peer to come back for a
* mutual close before giving up and closing anyway, letting
* the unilateral timeout escalate.
*/
auto constexpr close_patience = double(3 * 24 * 60 * 60);
}
namespace Boss { namespace Mod { namespace PeerComplaintsDesk {
@ -64,7 +74,6 @@ private:
Boss::Mod::Rpc* rpc;
bool soliciting;
bool should_close;
bool fees_low;
std::map<Ln::NodeId, std::string> exempted_nodes;
@ -80,7 +89,6 @@ private:
enabled = default_enabled;
rpc = nullptr;
soliciting = false;
should_close = false;
fees_low = false;
exempted_nodes.clear();
pending_complaints.clear();
@ -140,11 +148,14 @@ private:
+ Boss::concurrent(cleanup())
;
});
/* Poll close candidates every 10 minutes: a flaky
* peer's brief online windows should actually get
* caught for a mutual close.
*/
bus.subscribe<Msg::Timer10Minutes
>([this](Msg::Timer10Minutes const& _) {
if (!should_close || !rpc)
if (!rpc)
return Ev::lift();
should_close = false;
return Boss::concurrent(check_close());
});
@ -243,8 +254,6 @@ private:
/* Finished soliciting. */
soliciting = false;
/* Schedule closures for next 10-minute timer. */
should_close = true;
/* Do not waste memory. */
exempted_nodes.clear();
pending_complaints.clear();
@ -262,26 +271,8 @@ private:
});
}
Ev::Io<void> check_close() {
return Ev::lift().then([this]() {
if (!fees_low) {
/* Do not spam logs if we are not enabled
* anyway. */
if (!enabled)
return Ev::lift();
return Boss::log( bus, Info
, "PeerComplaintsDesk: "
"Fees are not known to be low, "
"will not close high-complaints channels."
);
}
return actual_check_close();
});
}
Ev::Io<void> actual_check_close() {
return db.transact().then([this](Sqlite3::Tx tx) {
auto complaints = Recorder::check_complaints(tx);
tx.commit();
auto const& unmanaged = unmanager.get_unmanaged();
@ -317,6 +308,21 @@ private:
if (c.second >= max_acceptable_complaints)
to_close.push_back(c.first);
}
/* Forget deferred closes for peers no longer
* over the threshold (recovered, unmanaged,
* or channel gone).
*/
for (auto& kv : Recorder::get_close_pendings(tx)) {
if (std::find( to_close.begin()
, to_close.end()
, kv.first
) == to_close.end())
Recorder::clear_close_pending( tx
, kv.first
);
}
tx.commit();
if (!first)
act += Boss::log( bus, Debug
, "PeerComplaintsDesk: Complaints: %s"
@ -357,13 +363,85 @@ private:
auto parms = Json::Out()
.start_object()
.field("id", Util::stringify(p))
.field("unilateraltimeout", channel_close_timeout)
.field("fee_negotiation_step", fee_negotiation_step)
.end_object()
;
return rpc->command("close", parms);
}).then([](Jsmn::Object _) {
return Ev::lift();
return rpc->command("listpeerchannels", std::move(parms));
}).then([this, p](Jsmn::Object res) {
auto connected = false;
if (res.has("channels")) {
auto cs = res["channels"];
for (auto c : cs) {
if (!c.has("peer_connected"))
continue;
auto conn = c["peer_connected"];
if (!conn.is_boolean())
continue;
if (bool(conn)) {
connected = true;
break;
}
}
}
if (connected) {
/* Take the mutual close as soon as the
* peer is reachable, regardless of the
* feerate: a mutual close even at high
* fees is cheaper than a unilateral at
* low fees, and online windows for these
* peers are rare.
*/
return db.transact(
).then([p](Sqlite3::Tx tx) {
Recorder::clear_close_pending(tx, p);
tx.commit();
return Ev::lift();
}).then([this, p]() {
return do_close(p);
});
}
/* Offline: wait for a mutual-close window
* until patience runs out. The deferral is
* persisted, so a restart does not reset
* the patience window.
*/
return db.transact(
).then([p](Sqlite3::Tx tx) {
Recorder::note_close_pending(tx, p, Ev::now());
auto since = Recorder::get_close_pending(tx, p);
tx.commit();
assert(since);
return Ev::lift(*since);
}).then([this, p](double since) {
if (Ev::now() - since < close_patience)
return Boss::log( bus, Debug
, "PeerComplaintsDesk: %s is "
"not connected, deferring "
"close while waiting for a "
"mutual-close window."
, Util::stringify(p).c_str()
);
/* Patience expired: get it over
* with, but only when fees are low —
* the unilateral path is the
* expensive one.
*/
if (!fees_low)
return Boss::log( bus, Debug
, "PeerComplaintsDesk: %s close "
"patience expired, holding "
"unilateral close until fees "
"are low."
, Util::stringify(p).c_str()
);
return Boss::log( bus, Info
, "PeerComplaintsDesk: %s never "
"came back online, closing "
"anyway."
, Util::stringify(p).c_str()
).then([this, p]() {
return do_close(p);
});
});
}).catching<RpcError>([this, p](RpcError e) {
return Boss::log( bus, Error
, "PeerComplaintsDesk: close %s error: %s"
@ -372,8 +450,22 @@ private:
);
});
}
Ev::Io<void> do_close(Ln::NodeId const& p) {
auto parms = Json::Out()
.start_object()
.field("id", Util::stringify(p))
.field("unilateraltimeout", channel_close_timeout)
.field("fee_negotiation_step", fee_negotiation_step)
.end_object()
;
return rpc->command("close", std::move(parms))
.then([](Jsmn::Object _) {
return Ev::lift();
});
}
Ev::Io<void> on_channel_destroy(Ln::NodeId const& p) {
return db.transact().then([p](Sqlite3::Tx tx) {
Recorder::clear_close_pending(tx, p);
Recorder::channel_closed(tx, p);
tx.commit();
return Ev::lift();

76
Boss/Mod/PeerComplaintsDesk/Recorder.cpp
View file

@ -3,6 +3,8 @@
#include"Ln/NodeId.hpp"
#include"Sqlite3.hpp"
#include"Util/date.hpp"
#include"Util/make_unique.hpp"
#include<memory>
#include<sstream>
#include<iostream>
@ -53,6 +55,15 @@ void Recorder::initialize(Sqlite3::Tx& tx) {
"PeerComplaintsDesk_closedcomplaints_time"
ON "PeerComplaintsDesk_closedcomplaints"(closedtime)
;
CREATE TABLE IF NOT EXISTS
"PeerComplaintsDesk_closepending"
( peerdbid INTEGER PRIMARY KEY
, since REAL NOT NULL
, FOREIGN KEY(peerdbid)
REFERENCES "PeerComplaintsDesk_peers"(peerdbid)
ON DELETE CASCADE
);
)QRY");
}
void Recorder::cleanup( Sqlite3::Tx& tx
@ -286,6 +297,71 @@ void Recorder::channel_closed(Sqlite3::Tx& tx, Ln::NodeId const& nid) {
.execute()
;
}
void Recorder::note_close_pending( Sqlite3::Tx& tx
, Ln::NodeId const& peer
, double since
) {
auto peerdbid = get_peerdbid(tx, peer);
tx.query(R"QRY(
INSERT OR IGNORE INTO "PeerComplaintsDesk_closepending"
( peerdbid, since)
VALUES( :peerdbid, :since);
)QRY")
.bind(":peerdbid", peerdbid)
.bind(":since", since)
.execute()
;
}
std::map< Ln::NodeId
, double
> Recorder::get_close_pendings(Sqlite3::Tx& tx) {
auto rv = std::map<Ln::NodeId, double>();
auto fetch = tx.query(R"QRY(
SELECT nodeid, since
FROM "PeerComplaintsDesk_peers" NATURAL JOIN
"PeerComplaintsDesk_closepending"
;
)QRY").execute();
for (auto& r : fetch)
rv[Ln::NodeId(r.get<std::string>(0))] = r.get<double>(1);
return rv;
}
std::unique_ptr<double> Recorder::get_close_pending( Sqlite3::Tx& tx
, Ln::NodeId const& peer
) {
auto rv = std::unique_ptr<double>();
auto fetch = tx.query(R"QRY(
SELECT since
FROM "PeerComplaintsDesk_peers" NATURAL JOIN
"PeerComplaintsDesk_closepending"
WHERE nodeid = :nodeid
;
)QRY")
.bind(":nodeid", std::string(peer))
.execute();
for (auto& r : fetch)
rv = Util::make_unique<double>(r.get<double>(0));
return rv;
}
void Recorder::clear_close_pending( Sqlite3::Tx& tx
, Ln::NodeId const& peer
) {
tx.query(R"QRY(
DELETE FROM "PeerComplaintsDesk_closepending"
WHERE peerdbid = (SELECT peerdbid
FROM "PeerComplaintsDesk_peers"
WHERE nodeid = :nodeid)
;
)QRY")
.bind(":nodeid", std::string(peer))
.execute()
;
}
std::map< Ln::NodeId
, std::vector<std::string>
> Recorder::get_closed_complaints(Sqlite3::Tx& tx) {

33
Boss/Mod/PeerComplaintsDesk/Recorder.hpp
View file

@ -3,6 +3,7 @@
#include<map>
#include<cstddef>
#include<memory>
#include<string>
#include<vector>
@ -81,6 +82,38 @@ namespace Recorder {
* separate area.
*/
void channel_closed(Sqlite3::Tx&, Ln::NodeId const&);
/** Boss::Mod::PeerComplaintsDesk::Recorder::note_close_pending
*
* @brief records the time a close was first deferred
* because the peer was offline. Does not change an
* existing record.
*/
void note_close_pending( Sqlite3::Tx&
, Ln::NodeId const&
, double since
);
/** Boss::Mod::PeerComplaintsDesk::Recorder::get_close_pendings
*
* @brief gathers the first-deferred times of all peers
* with a deferred close.
*/
std::map< Ln::NodeId
, double
> get_close_pendings(Sqlite3::Tx&);
/** Boss::Mod::PeerComplaintsDesk::Recorder::get_close_pending
*
* @brief the first-deferred time of one peer, or nullptr
* if the peer has no deferred close.
*/
std::unique_ptr<double> get_close_pending( Sqlite3::Tx&
, Ln::NodeId const&
);
/** Boss::Mod::PeerComplaintsDesk::Recorder::clear_close_pending
*
* @brief forgets the deferred-close time of a peer.
*/
void clear_close_pending(Sqlite3::Tx&, Ln::NodeId const&);
/** Boss::Mod::PeerComplaintsDesk::Recorder::get_closed_complaints
*
* @brief gathers all remembered non-ignored complaints for

1
Makefile.am
View file

@ -623,6 +623,7 @@ TESTS = \
tests/boss/test_peerjudge_algo \
tests/boss/test_peerjudge_datagatherer \
tests/boss/test_peerstatistician \
tests/boss/test_peercomplaintsdesk_main \
tests/boss/test_peercomplaintsdesk_recorder \
tests/boss/test_reqresp \
tests/boss/test_rpc \

67
tests/boss/test_jitrebalancer.cpp
View file

@ -14,6 +14,8 @@
#include"Boss/Msg/ResponseRpcCommand.hpp"
#include"Boss/Msg/SolicitHtlcAcceptedDeferrer.hpp"
#include"Ev/Io.hpp"
#include"Ev/concurrent.hpp"
#include"Ev/foreach.hpp"
#include"Ev/map.hpp"
#include"Ev/now.hpp"
#include"Ev/start.hpp"
@ -309,11 +311,6 @@ int main() {
void* requester = nullptr;
auto source = Ln::NodeId();
auto destination = Ln::NodeId();
/* Parallel-call check: the calls, and which one was
* let in.
*/
auto ids = std::vector<std::uint64_t>{3, 4, 5};
auto deferred_id = std::uint64_t(0);
bus.subscribe< RequestMoveFunds
>([&](RequestMoveFunds const& m) {
++num_move_funds;
@ -358,51 +355,27 @@ int main() {
}).then([&]() {
assert(num_move_funds == 0);
/* Check parallel calls to the same underfunded
* node: exactly one is let in and requests the
* rebalance; the rest are skipped because a
* run for the node is already in flight.
*/
return Ev::map([&](std::uint64_t id) {
return deferrer(htlc("1000x1x1", Ln::Amount::msat(90000), id));
/* Check parallel calls. */
auto ids = std::vector<std::uint64_t>{3, 4, 5};
auto act = Ev::lift();
/* Perform parallel calls. */
act += Ev::concurrent(Ev::map([&](std::uint64_t id) {
return deferrer(htlc("1000x1x0", Ln::Amount::msat(1), id));
}, ids).then([&](std::vector<bool> flags) {
/* Every forward should get in. */
for (auto flag : flags)
assert(flag);
return Ev::lift();
}));
act += Ev::yield();
act += Ev::foreach([&](std::uint64_t id) {
return release_monitor.wait_release(id);
}, ids);
}).then([&](std::vector<bool> flags) {
auto num_in = std::size_t(0);
for (auto i = std::size_t(0); i < flags.size(); ++i) {
if (flags[i]) {
++num_in;
deferred_id = ids[i];
}
}
assert(num_in == 1);
/* Wait for the let-in run to reach its
* move-funds request.
*/
return multiyield();
}).then([&]() {
/* Only the let-in run requests a rebalance. */
assert(num_move_funds == 1);
/* The 02 would not have fit. */
assert(source == Ln::NodeId("020000000000000000000000000000000000000000000000000000000000000000"));
assert(destination == Ln::NodeId("020000000000000000000000000000000000000000000000000000000000000001"));
/* Let the in-flight run finish. */
return bus.raise(ResponseMoveFunds{
requester,
Ln::Amount::sat(0),
Ln::Amount::sat(0)
});
}).then([&]() {
return release_monitor.wait_release(deferred_id);
}).then([&]() {
/* Let the finished run clean up. */
return multiyield();
return act;
}).then([&]() {
assert(num_move_funds == 0);
/* The guard clears once the run completes:
* a new forward that does not fit gets in
* again.
*/
num_move_funds = 0;
/* Check for a forward that does not fit. */
return deferrer(htlc("1000x1x1", Ln::Amount::msat(90000), 6));
}).then([&](bool flag) {
assert(flag == true);

431
tests/boss/test_peercomplaintsdesk_main.cpp Normal file
View file

@ -0,0 +1,431 @@
#undef NDEBUG
#include"Boss/Mod/PeerComplaintsDesk/Main.hpp"
#include"Boss/Mod/PeerComplaintsDesk/Recorder.hpp"
#include"Boss/Mod/Rpc.hpp"
#include"Boss/Msg/ChannelDestruction.hpp"
#include"Boss/Msg/DbResource.hpp"
#include"Boss/Msg/Init.hpp"
#include"Boss/Msg/OnchainFee.hpp"
#include"Boss/Msg/Option.hpp"
#include"Boss/Msg/Timer10Minutes.hpp"
#include"Boss/Shutdown.hpp"
#include"Ev/Io.hpp"
#include"Ev/concurrent.hpp"
#include"Ev/start.hpp"
#include"Ev/yield.hpp"
#include"Jsmn/Object.hpp"
#include"Json/Out.hpp"
#include"Ln/NodeId.hpp"
#include"Net/Connector.hpp"
#include"Net/Fd.hpp"
#include"Net/SocketFd.hpp"
#include"S/Bus.hpp"
#include"Secp256k1/PrivKey.hpp"
#include"Secp256k1/PubKey.hpp"
#include"Secp256k1/Signature.hpp"
#include"Secp256k1/SignerIF.hpp"
#include"Sha256/Hash.hpp"
#include"Sha256/fun.hpp"
#include"Sqlite3.hpp"
#include"Util/stringify.hpp"
#include<array>
#include<assert.h>
#include<cctype>
#include<cstdint>
#include<errno.h>
#include<fcntl.h>
#include<iostream>
#include<string>
#include<sys/socket.h>
#include<sys/types.h>
#include<unistd.h>
namespace {
/* Mock CLN on the other end of a socketpair: answers
* listpeerchannels with a switchable peer_connected flag and
* records close calls.
*/
class MockCln {
private:
Net::Fd socket;
Ev::Io<void> writeloop(std::string to_write) {
return Ev::yield().then([this, to_write]() {
auto res = write( socket.get()
, to_write.c_str(), to_write.size()
);
if (res < 0 && ( errno == EWOULDBLOCK
|| errno == EAGAIN
))
return writeloop(to_write);
assert(size_t(res) == to_write.size());
return Ev::yield();
});
}
Ev::Io<std::string> slurp() {
return Ev::yield().then([this]() -> Ev::Io<std::string> {
if (done)
return Ev::lift(std::string());
auto buf = std::string();
auto first = true;
for (;;) {
char tmp[256];
auto res = read( socket.get()
, tmp, sizeof(tmp)
);
if (res < 0 && ( errno == EWOULDBLOCK
|| errno == EAGAIN
)) {
if (first)
/* No data yet. */
return slurp();
break;
}
assert(res > 0);
buf.append(tmp, size_t(res));
first = false;
}
return Ev::lift(buf);
});
}
Ev::Io<void> handle(std::string req_s) {
if (req_s.empty())
return Ev::lift();
/* Strip the trailing record separators. */
while ( isspace(req_s.back())
)
req_s.pop_back();
auto req = Jsmn::Object::parse_json(req_s.c_str());
auto id = std::uint64_t(double(req["id"]));
auto method = std::string(req["method"]);
auto result = Json::Out();
auto robj = result.start_object();
if (method == "listpeerchannels") {
auto cs = robj.start_array("channels");
{
auto c = cs.start_object();
c.field("peer_connected", connected_flag);
c.end_object();
}
cs.end_array();
} else if (method == "close") {
++close_calls;
auto params = req["params"];
last_close_id = std::string(params["id"]);
last_close_timeout = std::uint64_t(double(
params["unilateraltimeout"]
));
} else {
std::cerr << "Unexpected command: " << method
<< std::endl;
assert(false);
}
robj.end_object();
auto js = Json::Out()
.start_object()
.field("jsonrpc", std::string("2.0"))
.field("id", double(id))
.field("result", std::move(result))
.end_object()
.output()
;
return writeloop(js).then([this]() {
return serve();
});
}
public:
bool done = false;
bool connected_flag = true;
std::size_t close_calls = 0;
std::string last_close_id;
std::uint64_t last_close_timeout = 0;
explicit
MockCln(Net::Fd socket_) : socket(std::move(socket_)) {
auto flags = fcntl(socket.get(), F_GETFL);
flags |= O_NONBLOCK;
fcntl(socket.get(), F_SETFL, flags);
}
MockCln(MockCln&&) =default;
MockCln(MockCln const&) =delete;
Ev::Io<void> serve() {
return slurp().then([this](std::string req_s) {
return handle(std::move(req_s));
});
}
};
class MockConnector : public Net::Connector {
public:
Net::SocketFd connect(std::string const&, int) override {
return Net::SocketFd(nullptr);
}
};
/* Nothing in the close path uses the signer, so an arbitrary
* fixed key is fine.
*/
class MockSigner : public Secp256k1::SignerIF {
private:
Secp256k1::PrivKey privkey;
public:
MockSigner()
: privkey(Secp256k1::PrivKey(std::string(
"0101010101010101010101010101010101010101010101010101010101010101"
)))
{ }
Secp256k1::PubKey get_pubkey_tweak(Secp256k1::PrivKey const&) override {
return Secp256k1::PubKey(privkey);
}
Secp256k1::Signature get_signature_tweak( Secp256k1::PrivKey const&
, Sha256::Hash const& m
) override {
return Secp256k1::Signature::create(privkey, m);
}
Sha256::Hash get_privkey_salted_hash(std::uint8_t salt[32]) override {
return Sha256::fun(salt, 32);
}
};
}
int main() {
namespace Recorder = Boss::Mod::PeerComplaintsDesk::Recorder;
auto bus = S::Bus();
auto db = Sqlite3::Db(":memory:");
auto sockets = std::array<int, 2>();
auto res = socketpair(AF_UNIX, SOCK_STREAM, 0, sockets.data());
assert(res >= 0);
auto server = MockCln(Net::Fd(sockets[0]));
auto rpc = Boss::Mod::Rpc(bus, Net::Fd(sockets[1]));
auto connector = MockConnector();
auto signer = MockSigner();
auto self_id = Ln::NodeId(std::string(
"02000000000000000000000000000000000000000000000000000000000000FFFF"
));
/* Module under test. */
auto mut = Boss::Mod::PeerComplaintsDesk::Main(bus);
auto peerA = Ln::NodeId(std::string(
"0200000000000000000000000000000000000000000000000000000000000000A1"
));
auto peerB = Ln::NodeId(std::string(
"0200000000000000000000000000000000000000000000000000000000000000B2"
));
auto peerC = Ln::NodeId(std::string(
"0200000000000000000000000000000000000000000000000000000000000000C3"
));
/* Enough non-ignored complaints to cross the close
* threshold.
*/
auto insert_complaints = [&](Ln::NodeId peer) {
return db.transact().then([peer](Sqlite3::Tx tx) {
for (auto i = 0; i < 5; ++i)
Recorder::add_complaint( tx, peer
, "test complaint"
);
tx.commit();
return Ev::lift();
});
};
/* Drive one 10-minute timer cycle and let the whole
* close-check chain run.
*/
auto cycle = [&]() {
return bus.raise(Boss::Msg::Timer10Minutes{}
).then([]() {
return Ev::yield(200);
});
};
/* Number of persisted deferred-close records for the
* peer (0 or 1).
*/
auto pending_count = [&](Ln::NodeId peer) {
return db.transact().then([peer](Sqlite3::Tx tx) {
auto n = std::size_t(0);
auto fetch = tx.query(R"QRY(
SELECT COUNT(*)
FROM "PeerComplaintsDesk_peers" NATURAL JOIN
"PeerComplaintsDesk_closepending"
WHERE nodeid = :nodeid
;
)QRY")
.bind(":nodeid", std::string(peer))
.execute();
for (auto& r : fetch)
n = r.get<std::size_t>(0);
tx.commit();
return Ev::lift(n);
});
};
/* Age the peer's deferred-close record into the past. */
auto age_pending = [&](Ln::NodeId peer, double secs) {
return db.transact().then([peer, secs](Sqlite3::Tx tx) {
tx.query(R"QRY(
UPDATE "PeerComplaintsDesk_closepending"
SET since = since - :secs
WHERE peerdbid = (SELECT peerdbid
FROM "PeerComplaintsDesk_peers"
WHERE nodeid = :nodeid)
;
)QRY")
.bind(":secs", secs)
.bind(":nodeid", std::string(peer))
.execute();
tx.commit();
return Ev::lift();
});
};
auto code = Ev::lift().then([&]() {
return Ev::concurrent(server.serve());
}).then([&]() {
return bus.raise(Boss::Msg::DbResource{db});
}).then([&]() {
return bus.raise(Boss::Msg::Init{
Boss::Msg::Network_Bitcoin, rpc, self_id, db,
connector, signer, "", false
});
}).then([&]() {
/* Enable auto-close. */
return bus.raise(Boss::Msg::Option{
"clboss-auto-close",
Jsmn::Object::parse_json("{\"enabled\": true}")["enabled"]
});
/* A connected peer is closed immediately, regardless of
* the feerate (fees_low starts false).
*/
}).then([&]() {
return insert_complaints(peerA);
}).then([&]() {
server.connected_flag = true;
return cycle();
}).then([&]() {
assert(server.close_calls == 1);
assert(server.last_close_id == std::string(peerA));
assert(server.last_close_timeout == 180);
/* The channel dying removes the peer from the
* candidate set.
*/
return bus.raise(Boss::Msg::ChannelDestruction{peerA});
}).then([&]() {
return Ev::yield(200);
}).then([&]() {
return cycle();
}).then([&]() {
/* Not closed again. */
assert(server.close_calls == 1);
/* An offline peer is deferred, and the deferral is
* persisted so a restart does not reset patience.
*/
return insert_complaints(peerB);
}).then([&]() {
server.connected_flag = false;
return cycle();
}).then([&]() {
assert(server.close_calls == 1);
return pending_count(peerB);
}).then([&](std::size_t n) {
assert(n == 1);
return db.transact();
}).then([&](Sqlite3::Tx tx) {
auto pendings = Recorder::get_close_pendings(tx);
tx.commit();
assert(pendings.count(peerB) == 1);
/* Still within patience: no close. */
return cycle();
}).then([&]() {
assert(server.close_calls == 1);
/* Patience expired, but fees are not low: hold the
* unilateral close.
*/
return age_pending(peerB, 4 * 24 * 60 * 60);
}).then([&]() {
return bus.raise(Boss::Msg::OnchainFee{false, nullptr});
}).then([&]() {
return cycle();
}).then([&]() {
assert(server.close_calls == 1);
/* Patience expired and fees low: close anyway. */
return bus.raise(Boss::Msg::OnchainFee{true, nullptr});
}).then([&]() {
return cycle();
}).then([&]() {
assert(server.close_calls == 2);
assert(server.last_close_id == std::string(peerB));
return bus.raise(Boss::Msg::ChannelDestruction{peerB});
}).then([&]() {
return Ev::yield(200);
/* A peer that drops below the threshold has its
* deferred-close record swept.
*/
}).then([&]() {
server.connected_flag = false;
return insert_complaints(peerC);
}).then([&]() {
return cycle();
}).then([&]() {
return pending_count(peerC);
}).then([&](std::size_t n) {
assert(n == 1);
/* Complaints gone (expired or recovered). */
return db.transact();
}).then([&](Sqlite3::Tx tx) {
tx.query(R"QRY(
DELETE FROM "PeerComplaintsDesk_complaints";
)QRY").execute();
tx.commit();
return Ev::lift();
}).then([&]() {
return cycle();
}).then([&]() {
return pending_count(peerC);
}).then([&](std::size_t n) {
assert(n == 0);
assert(server.close_calls == 2);
/* With auto-close disabled, even a connected peer over
* the threshold is not closed.
*/
return bus.raise(Boss::Msg::Option{
"clboss-auto-close",
Jsmn::Object::parse_json("{\"enabled\": false}")["enabled"]
});
}).then([&]() {
return insert_complaints(peerC);
}).then([&]() {
server.connected_flag = true;
return cycle();
}).then([&]() {
assert(server.close_calls == 2);
/* Stop the Rpc watchers so the event loop can
* drain and Ev::start can return.
*/
server.done = true;
return bus.raise(Boss::Shutdown());
}).then([&]() {
return Ev::lift(0);
});
return Ev::start(code);
}