#include"Boss/Mod/HtlcAcceptor.hpp" #include"Boss/Mod/Waiter.hpp" #include"Boss/Msg/CommandRequest.hpp" #include"Boss/Msg/CommandResponse.hpp" #include"Boss/Msg/ManifestHook.hpp" #include"Boss/Msg/Manifestation.hpp" #include"Boss/Msg/ProvideHtlcAcceptedDeferrer.hpp" #include"Boss/Msg/ReleaseHtlcAccepted.hpp" #include"Boss/Msg/SolicitHtlcAcceptedDeferrer.hpp" #include"Boss/concurrent.hpp" #include"Boss/log.hpp" #include"Ev/Io.hpp" #include"Ev/map.hpp" #include"Jsmn/Object.hpp" #include"Json/Out.hpp" #include"S/Bus.hpp" #include"Util/Str.hpp" #include"Util/make_unique.hpp" #include"Util/stringify.hpp" #include #include #include #include namespace { /* Number of seconds to defer response of htlc accepted. */ auto const defer_time = double(6.0); std::string stringify_cid(Ln::CommandId const& id) { auto rv = std::string(); id.cmatch([&](std::uint64_t nid) { rv = Util::stringify(nid); }, [&](std::string const& sid) { rv = sid; }); return rv; } } namespace Boss { namespace Mod { class HtlcAcceptor::Impl { private: S::Bus& bus; Boss::Mod::Waiter& waiter; /* Deferrer functions. */ std::vector( Ln::HtlcAccepted::Request const& )>> deferrers; bool soliciting; bool solicited; /* `htlc_accepted` hooks that we are informing to * deferrers. */ std::set deferring; /* `htlc_accepted` hooks that we know are deferred. */ std::set deferred; void start() { bus.subscribe([this](Msg::CommandRequest const& req) { if (req.command != "htlc_accepted") return Ev::lift(); return parse_payload(req.id, req.params ).then([this ](std::shared_ptr ptr) { if (!ptr) return Ev::lift(); return htlc_accepted(std::move(ptr)); }); }); bus.subscribe([this ](Msg::ProvideHtlcAcceptedDeferrer const& m) { if (!soliciting) return Ev::lift(); deferrers.push_back(m.deferrer); return Ev::lift(); }); bus.subscribe([this] (Msg::ReleaseHtlcAccepted const& r) { return release_htlc_accepted(r.response); }); bus.subscribe([this](Msg::Manifestation const& _) { return bus.raise(Msg::ManifestHook{"htlc_accepted"}); }); } Ev::Io solicit() { return Ev::yield().then([this]() { if (solicited) return Ev::lift(); if (soliciting) /* Busy-wait. */ return solicit(); soliciting = true; return bus.raise(Msg::SolicitHtlcAcceptedDeferrer{ } ).then([this]() { soliciting = false; solicited = true; return Ev::lift(); }); }); } Ev::Io> parse_payload( Ln::CommandId id , Jsmn::Object const& payload ) { auto rv = std::make_shared(); rv->id = id; try { auto onion = payload["onion"]; auto htlc = payload["htlc"]; rv->incoming_payload = Util::Str::hexread(std::string( onion["payload"] )); rv->incoming_amount = Ln::Amount::object( htlc["amount_msat"] ); rv->incoming_cltv = std::uint32_t(double( htlc["cltv_expiry"] )); if (onion.has("type")) rv->type_tlv = "tlv" == std::string( onion["type"] ); else rv->type_tlv = false; if (onion.has("short_channel_id")) { rv->next_channel = Ln::Scid(std::string( onion["short_channel_id"] )); rv->next_amount = Ln::Amount::object( onion["forward_msat"] ); rv->next_cltv = std::uint32_t(double( onion["outgoing_cltv_value"] )); } else { rv->next_channel = Ln::Scid(nullptr); rv->next_amount = rv->incoming_amount; rv->next_cltv = rv->incoming_cltv; } rv->next_onion = Util::Str::hexread(std::string( onion["next_onion"] )); rv->payment_hash = Sha256::Hash(std::string( htlc["payment_hash"] )); if (onion.has("payment_secret")) rv->payment_secret = Ln::Preimage(std::string( onion["payment_secret"] )); } catch (Jsmn::TypeError const& e) { return Boss::log( bus, Error , "HtlcAcceptor: Unexpected payload " "of htlc_accepted: %s" , Util::stringify(payload).c_str() ).then([e]() { throw e; return Ev::lift(std::shared_ptr< Ln::HtlcAccepted::Request >()); }); } return Ev::lift(std::move(rv)); } Ev::Io htlc_accepted(std::shared_ptr req) { auto id = req->id; return solicit().then([this, req]() { return Boss::log( bus, Debug , "HtlcAcceptor: " "HTLC %s (%s) arrived." , stringify_cid(req->id).c_str() , std::string(req->payment_hash) .c_str() ); }).then([this, req]() { deferring.insert(req->id); auto f = [ req ](std::function(Ln::HtlcAccepted::Request const&)> deferrer) { return deferrer(*req); }; return Ev::map(std::move(f), deferrers); }).then([this, id](std::vector results) { auto it = deferring.find(id); /* There is a race condition where one of the * deferrers takes too long to decide, then one * of the *other* deferrers has already given * the result. * So if it is no longer in the deferring set, * quit. */ if (it == deferring.end()) return Ev::lift(); deferring.erase(it); deferred.insert(id); /* If anyone wanted to defer, wait then raise. */ if (std::any_of( results.begin(), results.end() , [](bool x) { return x; } )) { auto act = waiter.wait(defer_time) + finish(id) ; return Boss::concurrent(act); } return finish(id); }); } Ev::Io finish(Ln::CommandId id) { return Ev::lift().then([this, id]() { auto it = deferred.find(id); if (it == deferred.end()) /* Somebody got to it first, * silently finish. */ return Ev::lift(); return bus.raise(Msg::ReleaseHtlcAccepted{ Ln::HtlcAccepted::Response::cont(id) }); }); } Ev::Io release_htlc_accepted(Ln::HtlcAccepted::Response const& resp) { /* First get the id and check if it is in either * deferred or deferring. */ auto id = resp.id(); auto found = false; auto it1 = deferred.find(id); if (it1 != deferred.end()) { deferred.erase(it1); found = true; } else { auto it2 = deferring.find(id); if (it2 != deferring.end()) { deferring.erase(it2); found = true; } } /* Silently fail. */ if (!found) return Ev::lift(); /* Generate hook result. */ auto os = std::ostringstream(); auto result = Json::Out(); if (resp.is_cont()) { os << "Continue."; result = Json::Out() .start_object() .field("result", "continue") .end_object() ; } else if (resp.is_fail()) { auto const& msg = resp.fail_message(); auto msg_s = Util::Str::hexdump(&msg[0], msg.size()); os << "Fail: " << msg_s; result = Json::Out() .start_object() .field("result", "fail") .field("failure_message", msg_s) .end_object() ; } else { auto key = std::string(resp.resolve_preimage()); os << "Resolve: " << key; result = Json::Out() .start_object() .field("result", "resolve") .field("payment_key", key) .end_object() ; } return Boss::log( bus, Debug , "HtlcAcceptor: HTLC %s: %s" , stringify_cid(id).c_str() , os.str().c_str() ) + bus.raise(Msg::CommandResponse{id, std::move(result)}) ; } public: Impl() =delete; Impl(Impl&&) =delete; Impl(Impl const&) =delete; explicit Impl( S::Bus& bus_ , Boss::Mod::Waiter& waiter_ ) : bus(bus_), waiter(waiter_) , soliciting(false) , solicited(false) { start(); } }; HtlcAcceptor::HtlcAcceptor(HtlcAcceptor&&) =default; HtlcAcceptor::~HtlcAcceptor() =default; HtlcAcceptor::HtlcAcceptor( S::Bus& bus , Boss::Mod::Waiter& waiter ) : pimpl(Util::make_unique(bus, waiter)) { } }}