/feed (websocket) implementation, with fallback to poll interface (#8)

src/backend.cpp

@@ -32,6 +32,8 @@
 #include <boost/asio/coroutine.hpp>
 #include <boost/numeric/conversion/cast.hpp>
 #include <boost/thread/lock_guard.hpp>
+#include <system_error>
+
 #include "crypto/crypto.h"     // monero/src
 #include "crypto/crypto-ops.h" // monero/src
 #include "cryptonote_basic/cryptonote_basic_impl.h"   // monero/src
@@ -39,6 +41,7 @@
 #include "error.h"
 #include "hex.h"               // monero/src
 #include "lwsf_config.h"
+#include "net/websocket.h"
 #include "ringct/rctOps.h"
 #include "wire.h"
 #include "wire/adapted/crypto.h"
@@ -387,6 +390,31 @@ namespace lwsf { namespace internal { namespace backend
       return self.subaccounts[sub.maj_i].last < sub.min_i; // last is inclusive
     }
 
+    boost::optional<crypto::public_key> is_real_spend(const account& self, const rpc::transaction_spend& source)
+    {
+      crypto::key_derivation derivation{};
+      if (!crypto::generate_key_derivation(source.tx_pub_key, self.view.sec, derivation))
+        return boost::none;
+
+      crypto::public_key spend_pub{};
+      const crypto::secret_key spend_sec = get_spend_secret(self, source.sender);
+      if (!crypto::secret_key_to_public_key(spend_sec, spend_pub))
+        return boost::none;
+
+      crypto::public_key output_pub{};
+      if (!crypto::derive_public_key(derivation, source.out_index, spend_pub, output_pub))
+        return boost::none;
+
+      crypto::secret_key output_secret{};
+      crypto::derive_secret_key(derivation, source.out_index, spend_sec, output_secret);
+
+      crypto::key_image image{};
+      crypto::generate_key_image(output_pub, output_secret, image);
+      if (source.key_image == image)
+        return output_pub;
+      return boost::none;
+    }
+
     boost::optional<std::vector<rpc::address_meta>> update_tx(const account& self, transaction& out, const rpc::transaction& source)
     {
       /* Let `receives` and `spends` re-populate in `merge_output` and
@@ -412,29 +440,12 @@ namespace lwsf { namespace internal { namespace backend
       std::uint64_t total_spent = 0; 
       for (const auto& spend : source.spent_outputs)
       {
-        crypto::key_derivation derivation{};
+        const auto output_pub = is_real_spend(self, spend);
-        if (!crypto::generate_key_derivation(spend.tx_pub_key, self.view.sec, derivation))
+        if (output_pub)
-          continue;
-
-        crypto::public_key spend_pub{};
-        const crypto::secret_key spend_sec = get_spend_secret(self, spend.sender);
-        if (!crypto::secret_key_to_public_key(spend_sec, spend_pub))
-          continue;
-
-        crypto::public_key output_pub{};
-        if (!crypto::derive_public_key(derivation, spend.out_index, spend_pub, output_pub))
-          continue;
-
-        crypto::secret_key output_secret{};
-        crypto::derive_secret_key(derivation, spend.out_index, spend_sec, output_secret);
-
-        crypto::key_image image{};
-        crypto::generate_key_image(output_pub, output_secret, image);
-        if (image == spend.key_image)
         {
           /* Frontend will typically know about spend before backend. So only
             merge and never erase spends. */
-          const rpc::address_meta sub = update_spend(out.spends.try_emplace(image).first->second, spend, output_pub);
+          const rpc::address_meta sub = update_spend(out.spends.try_emplace(spend.key_image).first->second, spend, *output_pub);
           if (need_expansion(self, sub))
             meta.push_back(sub);
           total_spent += std::uint64_t(spend.amount);
@@ -517,12 +528,9 @@ namespace lwsf { namespace internal { namespace backend
         {
           const bool rescanning =
             source.scanned_block_height < tx.second->height.value_or(0);
-          if (rescanning || !tx.second->description.empty() || !tx.second->transfers.empty())
+          if (rescanning || tx.second->has_metadata())
           {
-            const auto iter = updated_txes.emplace_hint(
+            const auto iter = updated_txes.emplace(tx.first, nullptr).first;
-              updated_txes.end(), tx.first, nullptr
-            );
-
             if (!iter->second)
             {
               iter->second = std::make_shared<transaction>(*tx.second);
@@ -610,15 +618,7 @@ namespace lwsf { namespace internal { namespace backend
       if (self.primary.restore_height <= self.primary.requested_start && !source.lookahead_fail) 
         self.import_error = {};
 
-      self.fee_mask = unspents.fee_mask;
+      self.update_fees(unspents);
-      if (unspents.fees.empty())
-      {
-        self.per_byte_fee.resize(1);
-        self.per_byte_fee[0] = unspents.per_byte_fee;
-      }
-      else
-        self.per_byte_fee = std::move(unspents.fees);
-
       return out;
     }
 
@@ -709,6 +709,447 @@ namespace lwsf { namespace internal { namespace backend
       runner();
     }
 
+    void notify_listener(Monero::WalletListener& listener, const std::uint64_t new_height, const std::uint64_t orig_height, const merge_results& merged)
+    {
+      listener.refreshed();
+      if (!merged.new_transactions.empty() || new_height - orig_height)
+        listener.updated();
+
+      for (std::uint64_t i = orig_height; i < new_height; ++i)
+        listener.newBlock(i);
+
+      for (const auto& tx : merged.new_transactions)
+      {
+        const auto txid = epee::string_tools::pod_to_hex(tx->id);
+        if (tx->direction == Monero::TransactionInfo::Direction_In)
+        {
+          if (tx->height)
+            listener.moneyReceived(txid, tx->amount);
+          else
+            listener.unconfirmedMoneyReceived(txid, tx->amount);
+        }
+        else
+          listener.moneySpent(txid, tx->amount);
+      }
+    }
+
+    class feed_loop
+    {
+      struct frame
+      {
+        std::weak_ptr<wallet> self;
+        std::weak_ptr<websocket::stream> ws;
+        merge_results* merged;
+        std::function<void(std::error_code)> on_sync;
+
+        explicit frame(std::shared_ptr<wallet> in, merge_results* merged, std::function<void(std::error_code)>&& on_sync)
+          : self(in), ws(in->feed), merged(merged), on_sync(std::move(on_sync))
+        {}
+      };
+
+      std::shared_ptr<frame> frame_;
+
+      /* Convert the simplified '/feed:tx_sync:' message to a split
+        '/get_address_txs' and '/get_unspent_outs' format. This way both legacy
+        and new syncing mechanisms use the same code. This slows down the new
+        sync, but keeps the legacy sync fast, which occurs more frequently. */
+      static expect<std::pair<rpc::get_address_txs, rpc::get_unspent_outs_response>> convert_to_legacy(rpc::feed_tx_sync src)
+      {
+        auto out = std::make_pair(rpc::get_address_txs{}, rpc::get_unspent_outs_response{});
+
+        out.first.lookahead_fail = src.lookahead_fail;
+        out.first.scanned_block_height = src.scanned_block_height;
+        out.first.start_height = src.start_height;
+        out.first.blockchain_height = src.blockchain_height;
+        out.first.lookahead = src.lookahead;
+
+        std::sort(
+          src.transactions.begin(), src.transactions.end(),
+          [] (const auto& lhs, const auto& rhs) 
+          {
+            static_assert(std::is_integral<decltype(lhs.height)>(), "optional types sort incorrectly");
+            return lhs.height < rhs.height;
+          }
+        );
+
+        std::map<rpc::feed_tx::output_id, rpc::feed_tx::receive> receives;
+        for (const auto& orig : src.transactions)
+        {
+          for (const auto& receive : orig.receives)
+          {
+            receives.emplace(receive.id, receive);
+            out.second.outputs.emplace_back(receive, orig);
+          }
+
+          auto& tx = out.first.transactions.emplace_back(orig);
+          for (const auto& orig_spend : orig.spends)
+          {
+            const auto receive = receives.find(orig_spend.id);
+            if (receive == receives.end())
+              return {error::feed_missing_output};
+            tx.spent_outputs.emplace_back(orig_spend, receive->second);
+          }
+        }
+
+        return out;
+      }
+
+      void on_error(wallet& self, boost::unique_lock<boost::mutex>& lock, const std::error_code error)
+      {
+        std::shared_ptr<websocket::stream> feed;
+        std::function<void(std::error_code)> on_sync;
+
+        LWSF_VERIFY(lock.owns_lock());
+        const std::function<void(std::error_code)> on_refresh{self.on_feed_refresh};
+        feed.swap(self.feed);
+        on_sync.swap(frame_->on_sync);
+
+        {
+          const boost::lock_guard<boost::mutex> lock_listener{self.sync_listener};
+          lock.unlock();
+          if (self.listener)
+            self.listener->refreshed();
+        }
+
+        MERROR("Error on websocket '/feed': " << error.message());
+        if (feed)
+          websocket::async_close(std::move(feed));
+
+        if (on_sync)
+          on_sync(self.refresh_error = error);
+        else
+          backend::wallet::refresh(frame_->self.lock(), true, [] (auto) {});
+
+        if (on_refresh)
+          on_refresh(error);
+      }
+
+      static std::string_view get_prefix(const epee::byte_slice& msg) noexcept
+      {
+        void const* const sep = std::memchr(msg.data(), u8':', msg.size());
+        if (!sep)
+          return {};
+        return {
+          reinterpret_cast<const char*>(msg.data()),
+          std::size_t(reinterpret_cast<const std::uint8_t*>(sep) - msg.data() + 1)
+        };
+      }
+
+      template<typename T>
+      static expect<T> unpack(epee::byte_slice&& msg)
+      {
+        const std::string_view prefix = get_prefix(msg);
+        if (prefix != T::prefix())
+          return {error::bad_feed_prefix};
+        msg.remove_prefix(prefix.size());
+
+        T out{};
+        const std::error_code error = wire::msgpack::from_bytes(std::move(msg), out);
+        if (error)
+          return error;
+        return out;
+      }
+
+      bool process(wallet& self, expect<rpc::feed_blocks>&& msg, boost::unique_lock<boost::mutex>& lock, merge_results& merged)
+      {
+        if (!msg)
+        {
+          on_error(self, lock, msg.error());
+          return false;
+        }
+
+        LWSF_VERIFY(lock.owns_lock());
+        std::map<rpc::feed_tx::output_id, std::shared_ptr<transaction>> receives;
+        std::unordered_map<crypto::hash, std::shared_ptr<transaction>> updated;
+        std::unordered_multimap<crypto::key_image, std::shared_ptr<transaction>> images;
+        updated.reserve(self.primary.txes.size() + msg->transactions.size());
+        for (const auto& tx : self.primary.txes)
+        {
+          auto elem = updated.try_emplace(tx.first, std::make_shared<transaction>(*tx.second));
+          for (const auto& receive : elem.first->second->receives)
+            receives.emplace(receive.second.get_output_id(), elem.first->second);
+          for (const auto& spend : elem.first->second->spends)
+            images.emplace(spend.first, elem.first->second);
+        }
+
+        for (const auto& new_tx : msg->transactions)
+        {
+          auto tx = updated.try_emplace(new_tx.hash, nullptr).first;
+          const bool is_new = !tx->second;
+          tx->second = std::make_shared<transaction>(new_tx, tx->second.get());
+          if (is_new)
+            merged.new_transactions.push_back(tx->second);
+
+          std::uint64_t receiving = 0;
+          for (const auto& receive : new_tx.receives)
+          {
+            receiving += receive.amount;
+            receives.emplace(receive.id, tx->second);
+            if (need_expansion(self.primary, receive.recipient))
+              merged.merge_subaddr(receive.recipient);
+          }
+
+          std::uint64_t spending = 0;
+          for (const auto& spend : new_tx.spends)
+          {
+            const auto receive = receives.find(spend.id);
+            if (receive == receives.end())
+            {
+              on_error(self, lock, error::feed_missing_output);
+              return false;
+            }
+
+            std::pair<crypto::public_key, transfer_in> const* output = nullptr;
+            for (const auto& existing : receive->second->receives)
+            {
+              if (existing.second.get_output_id() == spend.id)
+              {
+                output = std::addressof(existing);
+                break;
+              }
+            }
+
+            if (output == nullptr)
+            {
+              on_error(self, lock, error::feed_missing_output);
+              return false;
+            }
+
+            if (is_real_spend(self.primary, {spend, output->second.as_receive(output->first)}))
+            {
+              spending += output->second.amount;
+              tx->second->spends.try_emplace(spend.key_image, output->second, output->first);
+              auto failed = images.equal_range(spend.key_image);
+              for ( ; failed.first != failed.second; ++failed.first)
+                failed.first->second->failed = true;
+            }
+          }
+
+          if (receiving || spending)
+          {
+            if (new_tx.fee <= spending)
+              spending -= new_tx.fee;
+
+            if (tx->second->amount <= spending)
+            {
+              tx->second->direction = Monero::TransactionInfo::Direction_Out;
+              tx->second->amount = spending - tx->second->amount;
+            }
+            else
+            {
+              tx->second->direction = Monero::TransactionInfo::Direction_In;
+              tx->second->amount = tx->second->amount - spending;
+            }
+          }
+          else
+            updated.erase(new_tx.hash); // used solely as decoy in this tx
+        }
+
+        self.primary.txes.swap(updated);
+        merged.lookahead_fail = msg->lookahead_fail;
+        self.server_lookahead = {msg->lookahead.maj_i, msg->lookahead.min_i};
+        self.blockchain_height = std::max(self.blockchain_height, msg->blockchain_height);
+        self.primary.scan_height = std::max(self.primary.scan_height, msg->scan_end);
+        return true;
+      }
+
+      void process(wallet& self, expect<rpc::feed_error>&& msg, boost::unique_lock<boost::mutex>& lock)
+      {
+        if (!msg)
+          return on_error(self, lock, msg.error());
+        MERROR("Received websocket '/feed' error from server: " << msg->msg);
+        on_error(self, lock, error::feed);
+      }
+
+      bool process(wallet& self, expect<rpc::feed_mempool>&& msg, boost::unique_lock<boost::mutex>& lock, merge_results& merged)
+      {
+        if (!msg)
+        {
+          on_error(self, lock, msg.error());
+          return false;
+        }
+
+        LWSF_VERIFY(lock.owns_lock());
+        std::unordered_map<crypto::hash, std::shared_ptr<transaction>> updated;
+        for (const auto& tx : self.primary.txes)
+          updated.emplace(tx.first, std::make_shared<transaction>(*tx.second));
+
+        auto tx = updated.emplace(msg->hash, nullptr).first;
+        if (!tx->second)
+        {
+          tx->second = std::make_shared<transaction>(*msg);
+          merged.new_transactions.push_back(tx->second);
+          if (need_expansion(self.primary, msg->recipient))
+            merged.merge_subaddr(msg->recipient);
+
+          auto receive = tx->second->receives.try_emplace(msg->public_key);
+          if (receive.second)
+          {
+            tx->second->amount += msg->amount;
+            const auto sub = update_output(receive.first->second, *msg, self.primary.view.sec);
+            if (need_expansion(self.primary, sub))
+              merged.merge_subaddr(sub);
+          }
+        } // else we saw it because we spent it
+
+        self.primary.txes.swap(updated);
+        return true;
+      }
+
+      bool process(wallet& self, expect<rpc::feed_warning>&& msg, boost::unique_lock<boost::mutex>& lock)
+      {
+        if (!msg)
+        {
+          on_error(self, lock, msg.error());
+          return false;
+        }
+
+        MDEBUG("Received websocket '/feed' warning from server: " << msg->msg);
+        if (msg->code != rpc::feed_status::blockchain_reorg)
+          return false; // skip notifying wallet
+
+        LWSF_VERIFY(lock.owns_lock());
+        std::unordered_map<crypto::hash, std::shared_ptr<transaction>> updated;
+        for (const auto& tx : self.primary.txes)
+        {
+          LWSF_VERIFY(tx.second);
+          const bool by_height =
+            tx.second->height.value_or(std::numeric_limits<std::uint64_t>::max()) <= msg->height;
+          if (by_height || tx.second->has_metadata())
+          {
+            auto fresh = updated.emplace(tx.first, std::make_shared<transaction>(*tx.second));
+            if (!by_height)
+            {
+              fresh.first->second->timestamp = boost::none;
+              fresh.first->second->height = boost::none;
+              fresh.first->second->failed = false;
+            }
+          }
+        }
+
+        self.primary.txes.swap(updated);
+        self.blockchain_height = std::min(self.blockchain_height, msg->height);
+        self.primary.restore_height = std::min(self.primary.restore_height, msg->height);
+        self.primary.scan_height = std::min(self.primary.scan_height, msg->height);
+        return true;
+      }
+
+    public:
+      explicit feed_loop(std::shared_ptr<wallet> self, merge_results* merged, std::function<void(std::error_code)> on_sync)
+        : frame_(std::make_shared<frame>(std::move(self), merged, std::move(on_sync)))
+      {}
+
+      void operator()(std::error_code error = {}, copyable_slice incoming = {})
+      {
+        LWSF_VERIFY(frame_);
+        const auto self_ptr = frame_->self.lock();
+        LWSF_VERIFY(self_ptr);
+        wallet& self = *self_ptr;
+
+        boost::unique_lock<boost::mutex> lock{self.sync};
+        auto ws = frame_->ws.lock();
+        if (ws.get() != self.feed.get()) // sketchy, relies on make_shared not freeing until weak_ptr closes
+        {
+          LWSF_VERIFY(!frame_->on_sync); // violation in `wallet::refresh`
+          if (ws)
+            websocket::async_close(std::move(ws));
+          return;
+        }
+
+        if (error)
+          return on_error(self, lock, error);
+ 
+        self.last_sync = std::chrono::steady_clock::now();
+        if (frame_->merged)
+        {
+          auto msg = unpack<rpc::feed_tx_sync>(std::move(incoming.data));
+          if (!msg)
+            return on_error(self, lock, msg.error());
+
+          const auto split = convert_to_legacy(std::move(*msg));
+          if (!split)
+            return on_error(self, lock, split.error());
+
+          const std::uint64_t orig_height = self.primary.scan_height;
+          msg = {error::feed}; // release memory
+          *frame_->merged = merge_response(self, split->first, split->second);
+
+          std::function<void(std::error_code)> on_sync;
+          on_sync.swap(frame_->on_sync);
+          const std::function<void(std::error_code)> on_refresh{self.on_feed_refresh};
+
+          const std::uint64_t new_height = std::max(self.primary.scan_height, orig_height);
+          const boost::lock_guard<boost::mutex> listener_lock{self.sync_listener};
+          lock.unlock();
+
+          if (self.listener)
+            notify_listener(*self.listener, new_height, orig_height, *frame_->merged);
+
+          frame_->merged = nullptr;
+          if (on_sync)
+            on_sync({});
+          if (on_refresh)
+            on_refresh({});
+        }
+        else
+        {
+          if (incoming.data.empty())
+            return on_error(self, lock, error::bad_feed_prefix);
+
+          merge_results merged{};
+          const std::uint64_t orig_height = self.primary.scan_height;
+          switch (incoming.data.data()[0])
+          {
+            default:
+              return on_error(self, lock, error::bad_feed_prefix);
+            case 'b':
+              if (process(self, unpack<rpc::feed_blocks>(std::move(incoming.data)), lock, merged))
+                break;
+              return;
+            case 'e':
+              return process(self, unpack<rpc::feed_error>(std::move(incoming.data)), lock);
+            case 'm':
+              if (process(self, unpack<rpc::feed_mempool>(std::move(incoming.data)), lock, merged))
+                break;
+              return;
+            case 'w':
+              if (process(self, unpack<rpc::feed_warning>(std::move(incoming.data)), lock))
+                break;
+              return;
+          }
+
+          const boost::lock_guard<boost::mutex> listener_lock{self.sync_listener};
+          LWSF_VERIFY(lock.owns_lock());
+          if (!self.listener && !self.on_feed_refresh)
+            return;
+
+          const std::function<void(std::error_code)> on_refresh{self.on_feed_refresh};
+          const std::uint64_t new_height = self.primary.scan_height;
+
+          lock.unlock();
+          if (self.listener)
+            notify_listener(*self.listener, new_height, orig_height, merged);
+          if (on_refresh)
+            on_refresh({});
+        }
+      }
+    };
+
+    template<typename T>
+    epee::byte_slice generate_feed_msg(const T& msg)
+    {
+      epee::byte_stream sink;
+
+      const std::string_view prefix = T::prefix();
+      sink.write(prefix.data(), prefix.size());
+
+      const std::error_code error = wire::msgpack::to_bytes(sink, msg);
+      if (error)
+        throw std::system_error{error, std::string{"Failed to create "} + T::prefix()};
+
+      return epee::byte_slice{std::move(sink)};
+    }
+
     //! Bind N arguments into a callback that takes zero arguments to run
     template<typename F, typename... T>
     struct binder 
@@ -733,14 +1174,14 @@ namespace lwsf { namespace internal { namespace backend
       F f_;
 
       template<typename... T>
-      void operator()(T... args)
+      void operator()(T... args) const
       {
         // Use `post` instead of `dispatch` due to locking in handlers
         const auto self = self_.lock();
         LWSF_VERIFY(self);
         boost::asio::post(
           self->strand,
-          binder<F, T...>{self, std::move(f_), std::tuple<T...>(std::move(args)...)}
+          binder<F, T...>{self, f_, std::tuple<T...>(std::move(args)...)}
         );
       }
     };
@@ -772,6 +1213,24 @@ namespace lwsf { namespace internal { namespace backend
       LWSF_VERIFY(self);
       boost::asio::post(self->strand, binder<F>{self, std::move(f)});
     }
+
+    template<typename T, typename F>
+    struct store_
+    {
+      T* out;
+      F f;
+
+      void operator()(const std::error_code error, T val)
+      {
+        LWSF_VERIFY(out);
+        *out = val;
+        f(error);
+      }
+    };
+
+    template<typename T, typename F>
+    store_<T, F> store(T* out, F f)
+    { return {out, std::move(f)}; }
   } // anonymous
 
   WIRE_DEFINE_OBJECT(address_book_entry, map_address_book_entry);
@@ -904,6 +1363,64 @@ namespace lwsf { namespace internal { namespace backend
       failed(false)
   {}
 
+  transaction::transaction(const rpc::feed_tx& src, const transaction* existing)
+    : raw_bytes(),
+      spends(),
+      receives(),
+      transfers(),
+      description(),
+      timestamp(src.timestamp),
+      height(src.get_height()),
+      amount(0),
+      fee(src.fee),
+      unlock_time(src.unlock_time),
+      direction(Monero::TransactionInfo::Direction_In),
+      payment_id(src.payment_id),
+      id(src.hash),
+      prefix(src.prefix_hash),
+      coinbase(src.coinbase),
+      failed(false)
+  {
+    for (const auto& receive : src.receives)
+    {
+      amount += receive.amount;
+      receives.try_emplace(receive.public_key, receive);
+    }
+
+    if (existing)
+    {
+      transfers = existing->transfers;
+      description = existing->description;
+    }
+  }
+
+  transaction::transaction(const rpc::feed_mempool& src)
+   : raw_bytes(),
+      spends(),
+      receives(),
+      transfers(),
+      description(),
+      timestamp(std::chrono::system_clock::now()),
+      height(),
+      amount(src.amount),
+      fee(src.fee),
+      unlock_time(src.unlock_time),
+      direction(Monero::TransactionInfo::Direction_In),
+      payment_id(src.payment_id),
+      id(src.hash),
+      prefix(src.prefix_hash),
+      coinbase(false),
+      failed(false) 
+  {
+    auto elem = receives.try_emplace(src.public_key).first;
+    elem->second.global_index = std::numeric_limits<std::uint64_t>::max();
+    elem->second.amount = src.amount;
+    elem->second.recipient = src.recipient;
+    elem->second.index = src.index;
+    elem->second.rct_mask = src.rct;
+    elem->second.tx_pub = src.tx_pub_key;
+  }
+
   transaction::transaction(const transaction& rhs)
     : raw_bytes(rhs.raw_bytes.clone()), // required because of this
       spends(rhs.spends),
@@ -936,8 +1453,11 @@ namespace lwsf { namespace internal { namespace backend
 
   wallet::wallet(boost::asio::io_context& io)
     : listener(nullptr),
+      feed(nullptr),
       strand(io),
       client{},
+      on_feed_refresh(nullptr),
+      on_feed_fail(nullptr),
       primary{},
       per_byte_fee(),
       login_queue(),
@@ -946,13 +1466,16 @@ namespace lwsf { namespace internal { namespace backend
       lookahead_error(default_subaddr_state),
       import_error(),
       last_sync(),
+      last_fees(),
       blockchain_height(0),
       fee_mask(0),
       server_lookahead{},
       sync(),
       sync_listener(),
       sync_queue(),
-      passed_login(false)
+      passed_login(false),
+      probed_lookahead(false),
+      feed_disabled(false)
   {
     prep_primary_account(primary.subaccounts.emplace_back());
   }
@@ -963,6 +1486,11 @@ namespace lwsf { namespace internal { namespace backend
   {
     client.shutdown();
 
+    const boost::lock_guard<boost::mutex> lock1{sync};
+    feed = nullptr;
+    on_feed_refresh = nullptr;
+    on_feed_fail = nullptr;
+
     const boost::lock_guard<boost::mutex> lock{sync_queue};
     login_queue.clear();
     refresh_queue.clear(); 
@@ -1017,6 +1545,24 @@ namespace lwsf { namespace internal { namespace backend
       primary.lookahead.minor <= server_lookahead.minor; 
   }
 
+  bool wallet::need_fees() const noexcept
+  {
+    return config::refresh_interval_min <= std::chrono::steady_clock::now() - last_fees;
+  }
+
+  void wallet::update_fees(const rpc::get_unspent_outs_response& unspents)
+  {
+    last_fees = std::chrono::steady_clock::now();
+    fee_mask = unspents.fee_mask;
+    if (unspents.fees.empty())
+    {
+      per_byte_fee.resize(1);
+      per_byte_fee[0] = unspents.per_byte_fee;
+    }
+    else
+      per_byte_fee = unspents.fees;
+  }
+
   expect<epee::byte_slice> wallet::to_bytes() const
   {
     epee::byte_stream dest{};
@@ -1045,6 +1591,26 @@ namespace lwsf { namespace internal { namespace backend
     return status;
   }
 
+  void wallet::add_pending_tx(std::shared_ptr<transaction> pending)
+  {
+    const auto txid = epee::string_tools::pod_to_hex(pending->id);
+    const std::uint64_t amount = pending->amount;
+
+    {
+      const boost::lock_guard<boost::mutex> lock{sync};
+      auto entry = primary.txes.try_emplace(pending->id).first;
+      if (!entry->second)
+        entry->second = std::move(pending);
+    }
+    const boost::lock_guard<boost::mutex> lock{sync_listener};
+    if (!listener)
+      return;
+
+    listener->refreshed();
+    listener->updated();
+    listener->moneySpent(txid, amount);
+  }
+
   void wallet::login_is_new(std::shared_ptr<wallet> self, std::function<void(expect<bool>)> f)
   {
     struct frame
@@ -1095,6 +1661,10 @@ namespace lwsf { namespace internal { namespace backend
           self.subaddress_error = {};
           self.import_error = {}; 
           self.lookahead_error = self.lookahead_good() ? std::error_code{} :  error::subaddr_upgrade;
+          if (self.feed)
+            websocket::async_close(std::move(self.feed));
+          self.feed = nullptr;
+          self.feed_disabled = false;
 
           frame_->login = rpc::login_request{
             self.primary.address, self.primary.view.sec, self.primary.lookahead, true, self.primary.generated_locally
@@ -1179,6 +1749,7 @@ namespace lwsf { namespace internal { namespace backend
       rpc::get_subaddrs subaddrs;
       merge_results merged;
       std::uint64_t orig_scan_height;
+      rpc::feed feed;
       const bool mandatory;
 
       explicit frame(std::shared_ptr<wallet> in, const bool mandatory)
@@ -1190,6 +1761,7 @@ namespace lwsf { namespace internal { namespace backend
           subaddrs{},
           merged{},
           orig_scan_height(0),
+          feed{},
           mandatory(mandatory)
       {}
 
@@ -1244,6 +1816,57 @@ namespace lwsf { namespace internal { namespace backend
               return self.run(self.refresh_queue, self.refresh_error = error);
           }
 
+          if (!self.feed_disabled)
+          {
+            if (!self.feed)
+            {
+              BOOST_ASIO_CORO_YIELD rpc::invoke_async(
+                self.client, rpc::empty{}, &frame_->feed, wrap(self_ptr, *this)
+              );
+              if (error == http::error(501))
+              {
+  ws_connect:
+                BOOST_ASIO_CORO_YIELD self.client.ws_async(
+                  rpc::feed::endpoint(), rpc::feed::protocol(), store(&self.feed, wrap(self_ptr, *this))
+                );
+
+                if (!error && self.feed)
+                {
+                  // must hold lock to write `self.feed`, thus the frame
+                  BOOST_ASIO_CORO_YIELD websocket::async_subscribe(
+                    self.feed,
+                    generate_feed_msg(rpc::feed_login{rpc::login{self.primary.address, self.primary.view.sec}}),
+                    wrap(self_ptr, feed_loop{self_ptr, &frame_->merged, wrap(self_ptr, *this)})
+                  );
+                  if (!error)
+                    goto txs_merged;
+                }
+              }
+
+              if (self.feed)
+                websocket::async_close(std::move(self.feed));
+              self.feed = nullptr;
+              self.feed_disabled = true;
+              {
+                std::function<void(std::error_code)> on_feed_fail;
+                std::function<void(std::error_code)> on_feed_refresh;
+                on_feed_fail.swap(self.on_feed_fail);
+                on_feed_refresh.swap(self.on_feed_refresh);
+                if (on_feed_fail)
+                  on_feed_fail(error);
+              }
+            }
+            else // existing websocket stream to '/feed'
+            {
+              // ping should be quick, server doesn't hit disk, etc
+              BOOST_ASIO_CORO_YIELD websocket::async_ping(self.feed, wrap(self_ptr, *this));
+              if (!error)
+                return self.run(self.refresh_queue, self.refresh_error);
+              self.feed = nullptr;
+              goto ws_connect;
+            }
+          }
+
           frame_->orig_scan_height = self.primary.scan_height;
           frame_->login = rpc::login{self.primary.address, self.primary.view.sec};
           BOOST_ASIO_CORO_YIELD rpc::invoke_async(
@@ -1273,7 +1896,7 @@ namespace lwsf { namespace internal { namespace backend
 
           // Remember that this function provides the strong exception guarantee.
           frame_->merged = merge_response(self, frame_->txs_response, frame_->outs_response);
-
+  txs_merged:
           if (!self.import_error && self.primary.requested_start < self.primary.restore_height)
           {
             BOOST_ASIO_CORO_YIELD restore_height(
@@ -1342,29 +1965,10 @@ namespace lwsf { namespace internal { namespace backend
             std::max(frame_->orig_scan_height, self.primary.scan_height);
           lock.unlock();
 
-          self.listener->refreshed();
+          notify_listener(*self.listener, new_scan_height, frame_->orig_scan_height, frame_->merged);
-          const auto& merged = frame_->merged;
-          if (!merged.new_transactions.empty() || new_scan_height - frame_->orig_scan_height)
-            self.listener->updated();
-
-          for (std::uint64_t i = frame_->orig_scan_height; i < new_scan_height; ++i)
-            self.listener->newBlock(i);
-
-          for (const auto& tx : merged.new_transactions)
-          {
-            const auto txid = epee::string_tools::pod_to_hex(tx->id);
-            if (tx->direction == Monero::TransactionInfo::Direction_In)
-            {
-              if (tx->height)
-                self.listener->moneyReceived(txid, tx->amount);
-              else
-                self.listener->unconfirmedMoneyReceived(txid, tx->amount);
-            }
-            else
-              self.listener->moneySpent(txid, tx->amount);
-          }
 
           self.run(self.refresh_queue, rc);
+          frame_->self.reset(); // prevent re-invoking listener
         }
       }
     };
@@ -1374,6 +1978,65 @@ namespace lwsf { namespace internal { namespace backend
       post(self, handler{std::make_shared<frame>(self, mandatory)});
   }
 
+  void wallet::get_fees(std::shared_ptr<wallet> self, std::function<void(std::error_code)> f)
+  {
+    struct frame
+    {
+      std::weak_ptr<wallet> self;
+      rpc::get_unspent_outs_response response;
+      const std::function<void(std::error_code)> f;
+
+      explicit frame(std::shared_ptr<wallet> in, std::function<void(std::error_code)>&& f)
+        : self(std::move(in)),
+          response{},
+          f(std::move(f))
+      {}
+    };
+
+    struct handler : boost::asio::coroutine
+    {
+      std::shared_ptr<frame> frame_;
+
+      explicit handler(std::shared_ptr<frame> in)
+        : boost::asio::coroutine(), frame_(std::move(in))
+      {}
+
+      void operator()(const std::error_code error = {})
+      {
+        LWSF_VERIFY(frame_ && frame_->f);
+        if (error)
+          return frame_->f(error);
+
+        const auto self_ptr = frame_->self.lock();
+        LWSF_VERIFY(self_ptr);
+        wallet& self = *self_ptr;
+        assert(self.strand.running_in_this_thread());
+        const boost::lock_guard<boost::mutex> lock{self.sync};
+        BOOST_ASIO_CORO_REENTER(*this)
+        {
+          if (!self.need_fees())
+            return frame_->f(error);
+
+          if (!self.passed_login)
+            BOOST_ASIO_CORO_YIELD login(self_ptr, wrap(self_ptr, *this));
+
+          BOOST_ASIO_CORO_YIELD rpc::invoke_async(
+            self.client,
+            rpc::get_unspent_outs_request{rpc::login{self.primary.address, self.primary.view.sec}, rpc::uint64_string(0), 0, true},
+            &frame_->response,
+            wrap(self_ptr, *this)
+          );
+
+          self.update_fees(frame_->response);
+          frame_->f(error);
+        }
+      }
+    };
+
+    LWSF_VERIFY(self && f);
+    post(self, handler{std::make_shared<frame>(self, std::move(f))});
+  }
+
   void wallet::register_subaccount(std::shared_ptr<wallet> self, const std::uint32_t maj_i, std::function<void(std::error_code)> f)
   {
     struct frame

src/backend.h

@@ -113,21 +113,8 @@ namespace lwsf { namespace internal { namespace backend
   };
   WIRE_DECLARE_OBJECT(sub_account);
 
-  struct transfer_spend
-  {
-    std::uint64_t amount;
-    rpc::address_meta sender;
-    crypto::public_key tx_pub;
-    crypto::public_key output_pub;
-
-    transfer_spend() noexcept
-      : amount(0), sender{}, tx_pub{}, output_pub{}
-    {}
-  };
-  WIRE_DECLARE_OBJECT(transfer_spend);
-
   struct transfer_in
- {
+  {
     std::uint64_t global_index;
     std::uint64_t amount;
     rpc::address_meta recipient;
@@ -143,9 +130,52 @@ namespace lwsf { namespace internal { namespace backend
         rct_mask(),
         tx_pub{}
     {}
+
+    transfer_in(const rpc::feed_tx::receive& src) noexcept
+      : global_index(src.id.legacy.index),
+        amount(src.amount),
+        recipient(src.recipient),
+        index(src.index),
+        rct_mask(src.rct),
+        tx_pub(src.tx_pub_key)
+    {}
+
+    rpc::feed_tx::output_id get_output_id() const noexcept
+    {
+      return {
+        rpc::feed_tx::legacy_id{
+          bool(rct_mask) ? unsigned(0) : amount,
+          global_index
+        }
+      };
+    }
+
+    rpc::feed_tx::receive as_receive(const crypto::public_key& pub) const noexcept
+    {
+      return {
+        amount, get_output_id(), recipient, index, rct_mask, pub, tx_pub
+      };
+    }
   };
   WIRE_DECLARE_OBJECT(transfer_in);
 
+  struct transfer_spend
+  {
+    std::uint64_t amount;
+    rpc::address_meta sender;
+    crypto::public_key tx_pub;
+    crypto::public_key output_pub;
+
+    transfer_spend() noexcept
+      : amount(0), sender{}, tx_pub{}, output_pub{}
+    {}
+
+    transfer_spend(const transfer_in& src, const crypto::public_key& pub) noexcept
+      : amount(src.amount), sender(src.recipient), tx_pub(src.tx_pub), output_pub(pub)
+    {}
+  };
+  WIRE_DECLARE_OBJECT(transfer_spend);
+
   struct transfer_out
   {
     std::string address;
@@ -165,12 +195,16 @@ namespace lwsf { namespace internal { namespace backend
   struct transaction
   {
     transaction();
+
+    transaction(const rpc::feed_tx& src, const transaction* existing = nullptr);
+    transaction(const rpc::feed_mempool& src);
     
     transaction(transaction&&) = default;
     transaction(const transaction& rhs);
     transaction& operator=(transaction&&) = default;
 
     bool is_unlocked(std::uint64_t chain_height, Monero::NetworkType type) const;
+    bool has_metadata() const noexcept { return !description.empty() || !transfers.empty(); }
 
     /*! flat_map is used here for faster copies/merging. A single allocation
       is needed in the copy (done every refresh interval), instead of an
@@ -240,8 +274,11 @@ namespace lwsf { namespace internal { namespace backend
     // Everything uses `sync` for synchronization unless noted
 
     Monero::WalletListener* listener; //!< Use `sync_listener`
+    std::shared_ptr<websocket::stream> feed;
     boost::asio::io_context::strand strand; //<! useful in preventing contention on `sync` 
     http::client client;
+    std::function<void(std::error_code)> on_feed_refresh;
+    std::function<void(std::error_code)> on_feed_fail;
     account primary;
     std::string client_prefix;
     std::vector<std::uint64_t> per_byte_fee; //!< by priority level
@@ -252,18 +289,21 @@ namespace lwsf { namespace internal { namespace backend
     std::error_code lookahead_error; //!< warnings/errors of `server_lookahead` value
     std::error_code import_error; //!< Error from `import_wallet_request`
     std::chrono::steady_clock::time_point last_sync;
+    std::chrono::steady_clock::time_point last_fees;
     std::uint64_t blockchain_height;
     std::uint64_t fee_mask;
     config::lookahead server_lookahead; //!< Status of lookahead server-side
     mutable boost::mutex sync;
     boost::mutex sync_listener;
     boost::mutex sync_queue;
-    bool passed_login;
+    bool passed_login; //!< Reset to false to re-initialize all other values
     bool probed_lookahead; //!< True iff queries were done to determine lookahead re-sync
+    bool feed_disabled;
 
     wallet(boost::asio::io_context& io);
     ~wallet() noexcept;
 
+
     //! Extra guard against memory cycles
     void shutdown();
 
@@ -292,13 +332,14 @@ namespace lwsf { namespace internal { namespace backend
 
 
     // `sync` mutex is NOT acquired for this group
-
     cryptonote::network_type get_net_type() const;
     crypto::public_key get_spend_public(const rpc::address_meta& index) const;
     cryptonote::account_keys get_primary_keys() const;
     cryptonote::account_public_address get_spend_account(const rpc::address_meta& index) const;
     std::string get_spend_address(const rpc::address_meta& index) const;
     bool lookahead_good() const noexcept;
+    bool need_fees() const noexcept;
+    void update_fees(const rpc::get_unspent_outs_response& unspents);
 
     // End GROUP
 
@@ -310,6 +351,9 @@ namespace lwsf { namespace internal { namespace backend
     //! De-serialize `this` from msgpack. Locks+replaces contents.
     std::error_code from_bytes(epee::byte_slice source);
 
+    //! Do not hold `sync` or `sync_listener`
+    void add_pending_tx(std::shared_ptr<transaction> pending);
+
     // End GROUP
 
     // `sync` is acquired in a `asio::post` function - thread-safe and non-blocking until end
@@ -329,6 +373,9 @@ namespace lwsf { namespace internal { namespace backend
     //! Refreshes txes information. Strong exception guarantee.
     static void refresh(std::shared_ptr<wallet> self, bool mandatory, std::function<void(std::error_code)>);
 
+    //! Calls `get_unspent_outs` to update the fees when using websockets
+    static void get_fees(std::shared_ptr<wallet> self, std::function<void(std::error_code)>);
+
     //! Notify server that new major accounts need to be watched.
     static void register_subaccount(std::shared_ptr<wallet> self, std::uint32_t maj_i, std::function<void(std::error_code)> f);
 

src/error.cpp

@@ -40,12 +40,18 @@ namespace lwsf
       return "No error";
     case error::approval:
       return "Account needs approval on light-wallet-server";
+    case error::bad_feed_prefix:
+      return "Websocket prefix on '/feed' message was invalid";
     case error::configuration:
       return "Configuration issue";
     case error::create:
       return "Account creation not possible";
     case error::decryption:
       return "Unable to decrypt file (likely bad password)";
+    case error::feed:
+      return "Websocket '/feed' had error requiring resync";
+    case error::feed_missing_output:
+      return "Websocket '/feed' sent a spent without sending the receive";
     case error::import_fee:
       return "Import/restore from height needs payment, see address book";
     case error::import_invalid:

src/error.h

@@ -43,9 +43,12 @@ namespace lwsf
   {
     none = 0,           //!< Must be zero for `expect<..>`
     approval,           //!< Account needs approval
+    bad_feed_prefix,    //!< Websocket '/feed' received invalid prefix on msg
     configuration,      //!< Bad Configuration
     create,             //!< Account creation not possible
     decryption,         //!< Failed to decrypt data
+    feed,               //!< '/feed' websocket had error requiring restart
+    feed_missing_output,//!< '/feed' websocket sent spend without receive
     import_fee,         //!< `import_wallet_request` has fee
     import_invalid,     //!< `import_wallet_request` has invalid fee
     import_pending,     //!< `import_wallet_request` not fulfilled/approved

src/lwsf_config.h

@@ -58,6 +58,7 @@ namespace lwsf { namespace config
   constexpr const std::size_t max_txes_in_rpc = 2048;
   constexpr const std::size_t min_outputs = 2;
   constexpr const std::uint32_t mixin_default = 15;
+  constexpr const std::chrono::seconds ping_timeout{10};
   constexpr const std::chrono::seconds refresh_interval{30};
   constexpr const std::chrono::seconds refresh_interval_min{5};
   constexpr const std::chrono::seconds rpc_timeout{30};

src/lwsf_rpc.cpp

@@ -34,6 +34,7 @@
 #include <boost/spirit/include/qi_parse_attr.hpp>
 #include <boost/spirit/include/qi_sequence.hpp>
 #include <ctime>
+#include <limits>
 #include "hex.h"             // monero/contrib/epee/include
 #include "lwsf_config.h"
 #include "ringct/rctOps.h"   // monero/contrib/epee/include
@@ -42,6 +43,7 @@
 #include "wire/adapted/crypto.h"
 #include "wire/field.h"
 #include "wire/json.h"
+#include "wire/msgpack.h"
 #include "wire/traits.h"
 #include "wire/wrapper/array.h"
 #include "wire/wrapper/defaulted.h"
@@ -59,7 +61,7 @@ namespace lwsf { namespace internal { namespace rpc
   void read_bytes(wire::reader& source, empty&)
   { wire::object(source); }
 
-  void write_bytes(wire::json_writer& dest, const login& self)
+  void write_bytes(wire::writer& dest, const login& self)
   {
     wire::object(dest, WIRE_FIELD(address), WIRE_FIELD(view_key));
   }
@@ -92,9 +94,144 @@ namespace lwsf { namespace internal { namespace rpc
       WIRE_FIELD(height),
       WIRE_FIELD(target_height)
     );
-
   }
 
+  namespace
+  {
+    std::variant<empty, crypto::hash8, crypto::hash> make_payment_id(const boost::optional<epee::byte_slice>& payment_id)
+    {
+      if (payment_id && !payment_id->empty())
+      {
+        if (payment_id->size() == sizeof(crypto::hash8))
+        {
+          crypto::hash8 out{};
+          std::memcpy(std::addressof(out), payment_id->data(), sizeof(crypto::hash8));
+          return out;
+        }
+        else if (payment_id->size() == sizeof(crypto::hash))
+        {
+          crypto::hash out{};
+          std::memcpy(std::addressof(out), payment_id->data(), sizeof(crypto::hash));
+          return out;
+        }
+        else
+          WIRE_DLOG_THROW(wire::error::schema::fixed_binary, "Invalid payment_id size");
+      }
+
+      return empty{};
+    }
+  }
+
+  void read_bytes(wire::json_reader& source, const feed&)
+  { wire::object(source); }
+
+  static void read_bytes(wire::msgpack_reader& source, feed_tx::legacy_id& self)
+  { wire::object(source, WIRE_FIELD(amount), WIRE_FIELD(index)); }
+
+  static void read_bytes(wire::msgpack_reader& source, feed_tx::output_id& self)
+  { wire::object(source, WIRE_FIELD(legacy)); }
+
+  static void read_bytes(wire::msgpack_reader& source, feed_tx::receive& self)
+  {
+    static constexpr std::uint64_t max = std::numeric_limits<std::uint64_t>::max();
+    static constexpr auto txpool = feed_tx::output_id{feed_tx::legacy_id{max, max}};
+    wire::object(source,
+      WIRE_FIELD(amount),
+      WIRE_FIELD_DEFAULTED(id, txpool),
+      WIRE_FIELD_DEFAULTED(recipient, address_meta{}),
+      WIRE_FIELD(index),
+      WIRE_OPTIONAL_FIELD(rct),
+      WIRE_FIELD(public_key),
+      WIRE_FIELD(tx_pub_key)
+    );
+  }
+
+  static void read_bytes(wire::msgpack_reader& source, feed_tx::spend& self)
+  { wire::object(source, WIRE_FIELD(id), WIRE_FIELD(key_image)); }
+
+  void read_bytes(wire::msgpack_reader& source, feed_tx& self)
+  {
+    using min_receive = wire::min_element_sizeof<crypto::public_key, crypto::public_key>;
+    using min_spend = wire::min_element_sizeof<crypto::key_image>;
+    using time_point = std::chrono::system_clock::time_point;
+
+    time_point::rep timestamp = 0;
+    boost::optional<epee::byte_slice> payment_id;
+
+    wire::object(source,
+      WIRE_FIELD_ARRAY(receives, min_receive),
+      WIRE_FIELD_ARRAY(spends, min_spend),
+      WIRE_FIELD(fee),
+      WIRE_FIELD(unlock_time),
+      WIRE_FIELD_DEFAULTED(height, feed_tx::txpool()),
+      wire::field("timestamp", std::ref(timestamp)),
+      WIRE_FIELD_DEFAULTED(mixin, std::numeric_limits<std::uint32_t>::max()),
+      wire::optional_field("payment_id", std::ref(payment_id)),
+      WIRE_FIELD(hash),
+      WIRE_FIELD(prefix_hash),
+      WIRE_FIELD_DEFAULTED(coinbase, false),
+      WIRE_FIELD_DEFAULTED(mempool, false)
+    );
+
+    self.payment_id = make_payment_id(payment_id);
+    self.timestamp = time_point{std::chrono::seconds{timestamp}};
+  }
+
+  using min_tx_size = wire::min_element_sizeof<crypto::public_key, crypto::public_key>;
+  void read_bytes(wire::msgpack_reader& source, feed_blocks& self)
+  {
+    wire::object(source,
+      WIRE_FIELD_ARRAY(transactions, min_tx_size),
+      WIRE_FIELD(scan_start),
+      WIRE_FIELD(scan_end),
+      WIRE_FIELD(blockchain_height),
+      WIRE_OPTIONAL_FIELD(lookahead_fail),
+      WIRE_FIELD_DEFAULTED(lookahead, address_meta{})
+    );
+  }
+
+  void read_bytes(wire::msgpack_reader& source, feed_error& self)
+  { wire::object(source, WIRE_FIELD(msg), WIRE_FIELD(code)); }
+
+  void write_bytes(wire::msgpack_writer& dest, const feed_login& self)
+  { wire::object(dest, WIRE_FIELD(account)); }
+
+  void read_bytes(wire::msgpack_reader& source, feed_mempool& self)
+  {
+    boost::optional<epee::byte_slice> payment_id;
+    wire::object(source,
+      WIRE_FIELD(amount),
+      WIRE_FIELD(fee),
+      WIRE_FIELD(unlock_time),
+      WIRE_FIELD_DEFAULTED(recipient, address_meta{}),
+      WIRE_FIELD_DEFAULTED(mixin, std::numeric_limits<std::uint32_t>::max()),
+      wire::optional_field("payment_id", std::ref(payment_id)),
+      WIRE_OPTIONAL_FIELD(rct),
+      WIRE_FIELD(hash),
+      WIRE_FIELD(prefix_hash),
+      WIRE_FIELD(public_key),
+      WIRE_FIELD(tx_pub_key),
+      WIRE_FIELD(index)
+    );
+
+    self.payment_id = make_payment_id(payment_id);
+  }
+
+  void read_bytes(wire::msgpack_reader& source, feed_tx_sync& self)
+  {
+    wire::object(source,
+      WIRE_FIELD_ARRAY(transactions, min_tx_size),
+      WIRE_FIELD(scanned_block_height),
+      WIRE_FIELD(start_height),
+      WIRE_FIELD(blockchain_height),
+      WIRE_OPTIONAL_FIELD(lookahead_fail),
+      WIRE_FIELD_DEFAULTED(lookahead, address_meta{})
+    );
+  }
+
+  void read_bytes(wire::msgpack_reader& source, feed_warning& self)
+  { wire::object(source, WIRE_FIELD(msg), WIRE_FIELD(height), WIRE_FIELD(code)); }
+
   void write_bytes(wire::json_writer& dest, const uint64_string source)
   {
     const auto as_string = dest.to_string(std::uintmax_t(source));
@@ -147,26 +284,9 @@ namespace lwsf { namespace internal { namespace rpc
       WIRE_FIELD(mempool)
     );
 
-    if (payment_id && !payment_id->empty())
+    self.payment_id = make_payment_id(payment_id);
-    {
-      if (payment_id->size() == sizeof(crypto::hash8))
-      {
-        self.payment_id = crypto::hash8{};
-        std::memcpy(std::addressof(std::get<crypto::hash8>(self.payment_id)), payment_id->data(), sizeof(crypto::hash8));
-      }
-      else if (payment_id->size() == sizeof(crypto::hash))
-      {
-        self.payment_id = crypto::hash{};
-        std::memcpy(std::addressof(std::get<crypto::hash>(self.payment_id)), payment_id->data(), sizeof(crypto::hash));
-      }
-      else
-        WIRE_DLOG_THROW(wire::error::schema::fixed_binary, "Invalid payment_id size");
-    }
-    else
-      self.payment_id = empty{};
 
     self.timestamp.reset();
-
     if (timestamp)
     {
       //  %Y-%m-%dT%H:%M:%SZ
@@ -349,7 +469,6 @@ namespace lwsf { namespace internal { namespace rpc
     wire::object(source, WIRE_FIELD_ARRAY(all_subaddrs, max_subaddrs));
   }
 
-
   namespace
   {
     // This field is a mess between implementations, this cleans it up a bit

src/lwsf_rpc.h

@@ -34,6 +34,7 @@
 #include <boost/optional/optional.hpp>
 #include <boost/thread/locks.hpp>
 #include <boost/variant.hpp>
+#include <chrono>
 #include <ctime>
 #include <system_error>
 #include <type_traits>
@@ -49,6 +50,7 @@
 #include "wire/basic_value.h"
 #include "wire/fwd.h"
 #include "wire/json.h"
+#include "wire/msgpack/fwd.h"
 #include "wire/traits.h"
 
 namespace lwsf { namespace internal { namespace rpc
@@ -140,7 +142,7 @@ namespace lwsf { namespace internal { namespace rpc
     std::string address;
     crypto::secret_key view_key;
   };
-  void write_bytes(wire::json_writer&, const login&); 
+  void write_bytes(wire::writer&, const login&);
 
  
   struct login_request
@@ -179,6 +181,197 @@ namespace lwsf { namespace internal { namespace rpc
   void read_bytes(wire::json_reader&, daemon_status&);
 
 
+  enum class feed_status : std::uint16_t
+  {
+    unspecified_error = 0,
+    account_not_found,
+    bad_address,
+    bad_view_key,
+    blockchain_reorg,
+    daemon_unresponsive,
+    parse_error,
+    protocol_error,
+    queue_error,
+    schema_error
+  };
+  WIRE_AS_INTEGER(feed_status);
+
+  struct feed
+  {
+    feed() = delete;
+    static constexpr const char* endpoint() noexcept { return "/feed"; }
+    static constexpr const char* protocol() noexcept { return "lws.feed.v0.msgpack"; }
+  };
+  void read_bytes(wire::json_reader&, const feed&);
+
+  struct feed_tx
+  { 
+    struct legacy_id
+    {
+      legacy_id() = delete;
+      std::uint64_t amount;
+      std::uint64_t index;
+    };
+
+    struct output_id { legacy_id legacy; };
+
+    struct receive
+    {
+      std::uint64_t amount;
+      output_id id;
+      address_meta recipient;
+      std::uint16_t index;
+      boost::optional<rct::key> rct;
+      crypto::public_key public_key;
+      crypto::public_key tx_pub_key;
+
+      receive() noexcept
+        : amount(0), id{}, recipient{}, index(0), rct(), public_key{}, tx_pub_key{}
+      {}
+
+      receive(std::uint64_t amount, const output_id& id, const address_meta& recipient, std::uint16_t index, const boost::optional<rct::key> rct, const crypto::public_key& public_key, const crypto::public_key& tx_pub_key) noexcept
+        : amount(amount), id(id), recipient(recipient), index(index), rct(rct), public_key(public_key), tx_pub_key(tx_pub_key)
+      {}
+    };
+
+    struct spend
+    {
+      output_id id;
+      crypto::key_image key_image;
+
+      spend() noexcept
+        : id{}, key_image{}
+      {}
+    };
+
+    std::vector<receive> receives;
+    std::vector<spend> spends;
+    std::uint64_t fee;
+    std::uint64_t unlock_time;
+    std::uint64_t height;
+    std::chrono::system_clock::time_point timestamp;
+    std::uint32_t mixin;
+    std::variant<empty, crypto::hash8, crypto::hash> payment_id;
+    crypto::hash hash;
+    crypto::hash prefix_hash;
+    bool coinbase;
+    bool mempool;
+
+    //! Hack because `boost/std::optional` sort left adjacent
+    static constexpr std::uint64_t txpool() noexcept { return std::uint64_t(-1); }
+
+    feed_tx()
+      : receives(),
+        spends(),
+        unlock_time(0),
+        height(),
+        timestamp(),
+        mixin(),
+        payment_id(),
+        hash{},
+        prefix_hash{},
+        coinbase(false),
+        mempool(false)
+    {}
+
+    boost::optional<std::uint64_t> get_height() const noexcept
+    { return boost::make_optional(height != txpool(), height); }
+  };
+  void read_bytes(wire::msgpack_reader&, feed_tx&);
+
+  inline constexpr bool operator==(const feed_tx::output_id& lhs, const feed_tx::output_id& rhs) noexcept
+  {
+    return lhs.legacy.index == rhs.legacy.index && lhs.legacy.amount == rhs.legacy.amount;
+  }
+  inline constexpr bool operator!=(const feed_tx::output_id& lhs, const feed_tx::output_id& rhs) noexcept
+  {
+    return lhs.legacy.index != rhs.legacy.index || lhs.legacy.amount != rhs.legacy.amount;
+  }
+  inline constexpr bool operator<(const feed_tx::output_id& lhs, const feed_tx::output_id& rhs) noexcept
+  {
+    return lhs.legacy.amount == rhs.legacy.amount ?
+      lhs.legacy.index < rhs.legacy.index : lhs.legacy.amount < rhs.legacy.amount;
+  }
+
+  struct feed_blocks
+  {
+    feed_blocks() = delete;
+    static constexpr const char* prefix() noexcept { return "blocks:"; }
+
+    std::vector<feed_tx> transactions;
+    std::uint64_t scan_start;
+    std::uint64_t scan_end;
+    std::uint64_t blockchain_height;
+    boost::optional<std::uint64_t> lookahead_fail;
+    address_meta lookahead;
+  };
+  void read_bytes(wire::msgpack_reader&, feed_blocks&);
+
+  struct feed_error
+  {
+    feed_error() = delete;
+    static constexpr const char* prefix() noexcept { return "error:"; }
+
+    std::string msg;
+    feed_status code;
+  };
+  void read_bytes(wire::msgpack_reader&, feed_error&);
+
+  struct feed_login
+  {
+    feed_login() = delete;
+    static constexpr const char* prefix() noexcept { return "login:"; }
+    login account;
+  };
+  void write_bytes(wire::msgpack_writer&, const feed_login&);
+
+  //! One specific output is sent per-message
+  struct feed_mempool
+  {
+    feed_mempool() = delete;
+    static constexpr const char* prefix() noexcept { return "mempool:"; }
+
+    std::uint64_t amount;
+    std::uint64_t fee;
+    std::uint64_t unlock_time;
+    address_meta recipient;
+    std::uint32_t mixin;
+    std::variant<empty, crypto::hash8, crypto::hash> payment_id;
+    boost::optional<rct::key> rct;
+    crypto::hash hash;
+    crypto::hash prefix_hash;
+    crypto::public_key public_key;
+    crypto::public_key tx_pub_key;
+    std::uint16_t index;
+  };
+  void read_bytes(wire::msgpack_reader&, feed_mempool&);
+
+  struct feed_tx_sync
+  {
+    feed_tx_sync() = delete;
+    static constexpr const char* prefix() noexcept { return "tx_sync:"; }
+
+    std::vector<feed_tx> transactions;
+    std::uint64_t scanned_block_height;
+    std::uint64_t start_height;
+    std::uint64_t blockchain_height;
+    boost::optional<std::uint64_t> lookahead_fail;
+    address_meta lookahead;
+  };
+  void read_bytes(wire::msgpack_reader&, feed_tx_sync&);
+
+  struct feed_warning
+  {
+    feed_warning() = delete;
+    static constexpr const char* prefix() noexcept { return "warning:"; }
+
+    std::string msg;
+    std::uint64_t height;
+    feed_status code;
+  };
+  void read_bytes(wire::msgpack_reader&, feed_warning&);
+
+
   enum class uint64_string : std::uint64_t {};
   void write_bytes(wire::json_writer&, uint64_string);
   void read_bytes(wire::json_reader&, uint64_string&); 
@@ -198,6 +391,14 @@ namespace lwsf { namespace internal { namespace rpc
         key_image{},
         tx_pub_key{}
     {}
+
+    transaction_spend(const feed_tx::spend& spend, const feed_tx::receive& receive) noexcept
+      : amount(uint64_string(receive.amount)),
+        sender(receive.recipient),
+        out_index(receive.index),
+        key_image(spend.key_image),
+        tx_pub_key(receive.tx_pub_key)
+    {}
   };
   void read_bytes(wire::json_reader&, transaction_spend&);
 
@@ -226,6 +427,24 @@ namespace lwsf { namespace internal { namespace rpc
         coinbase(false),
         mempool(false)
     {}
+
+    transaction(const feed_tx& src)
+      : spent_outputs(),
+        payment_id(src.payment_id),
+        timestamp(std::chrono::system_clock::to_time_t(src.timestamp)),
+        fee(uint64_string(src.fee)),
+        total_received(uint64_string(0)),
+        unlock_time(src.unlock_time),
+        height(src.get_height()),
+        hash(src.hash),
+        coinbase(src.coinbase),
+        mempool(src.mempool)
+    {
+      std::uint64_t total = 0;
+      for (const auto& received : src.receives)
+        total += received.amount;
+      total_received = uint64_string(total);
+    }
   };
   void read_bytes(wire::json_reader&, transaction&);
   
@@ -333,8 +552,6 @@ namespace lwsf { namespace internal { namespace rpc
 
   struct ringct
   {
-    ringct() = delete;
-
     enum class format : std::uint8_t
     {
       none = 0, //!< Not ringct
@@ -343,6 +560,19 @@ namespace lwsf { namespace internal { namespace rpc
       unencrypted
     };
 
+    ringct() noexcept
+      : mask{}, type(format::none)
+    {}
+
+    ringct(const rct::key& mask, const format type) noexcept
+      : mask(mask), type(type)
+    {}
+
+    ringct(const boost::optional<rct::key>& src) noexcept
+      : mask(src.value_or(rct::key{})),
+        type(bool(src) ? format::unencrypted : format::none)
+    {}
+
     rct::key mask;
     format type;
   };
@@ -370,6 +600,30 @@ namespace lwsf { namespace internal { namespace rpc
         public_key{},
         tx_pub_key{}
     {}
+
+    output(const feed_mempool& src) noexcept
+      : amount(uint64_string(src.amount)),
+        global_index(uint64_string(std::numeric_limits<std::uint64_t>::max())),
+        recipient(src.recipient),
+        index(src.index),
+        rct(src.rct),
+        tx_hash(src.hash),
+        tx_prefix_hash(src.prefix_hash),
+        public_key(src.public_key),
+        tx_pub_key(src.tx_pub_key)
+    {}
+
+    output(const feed_tx::receive& src, const feed_tx& tx) noexcept
+      : amount(uint64_string(src.amount)),
+        global_index(uint64_string(src.id.legacy.index)),
+        recipient(src.recipient),
+        index(src.index),
+        rct(src.rct),
+        tx_hash(tx.hash),
+        tx_prefix_hash(tx.prefix_hash),
+        public_key(src.public_key),
+        tx_pub_key(src.tx_pub_key)
+    {}
   };
   void read_bytes(wire::json_reader&, output&);
 

src/net/CMakeLists.txt

@@ -26,8 +26,8 @@
 # STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
 # THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
-set(lwsf-net_sources context.cpp http.cpp wallet_io.cpp)
+set(lwsf-net_sources context.cpp http.cpp wallet_io.cpp websocket.cpp)
-set(lwsf-net_headers context.h http.h wallet_io.h)
+set(lwsf-net_headers context.h http.h wallet_io.h websocket.h)
 
 add_library(lwsf-net ${lwsf-net_sources} ${lwsf-net_headers})
 target_link_libraries(lwsf-net PRIVATE monero::libraries)

src/net/http.cpp

@@ -58,6 +58,7 @@
 #include "misc_log_ex.h"  // monero/contrib/epee/include
 #include "net/net_utils_base.h"
 #include "net/socks.h"    // monero/src
+#include "net/websocket.h"
 #include "string_tools.h" // monero/contrib/epee/include
 
 namespace lwsf { namespace internal { namespace http
@@ -211,7 +212,7 @@ namespace lwsf { namespace internal { namespace http
       void(std::shared_ptr<client_state>, std::function<void(boost::system::error_code)>);
 
     boost::beast::flat_static_buffer<config::http_parser_buffer_size> buffer;
-    std::function<connect_func> connect;
+    const std::function<connect_func> connect;
     std::deque<message> outgoing;
     const std::string host;
     const std::string prefix;
@@ -435,6 +436,14 @@ namespace lwsf { namespace internal { namespace http
           self.is_connected = true;
           while (!self.outgoing.empty())
           {
+            if (self.outgoing.front().verb == boost::beast::http::verb::subscribe)
+            {
+              std::function<server_response_func> notifier;
+              notifier.swap(self.outgoing.front().notifier);
+              self.outgoing.clear();
+              return notifier({}, {});
+            }
+
             set_timeout(self_, config::rpc_timeout);
 
             MDEBUG("Sending " << self.outgoing.front().body.size() << " bytes in HTTP " << (self.outgoing.front().is_get() ? "GET" : "POST") << " to " << self.outgoing.front());
@@ -490,31 +499,35 @@ namespace lwsf { namespace internal { namespace http
         }
       }
     };
+
+    void queue_async_(std::shared_ptr<client_state> state, std::string&& target, epee::byte_slice&& body, std::function<server_response_func>&& notifier, boost::beast::http::verb verb)
+    {
+      message msg{std::move(body), std::move(target), std::move(notifier), verb};
+      if (!state)
+        return msg.notifier(common_error::kInvalidArgument, epee::byte_slice{});
+
+      MDEBUG("Queueing HTTP " << to_string(verb) << " to " << msg << " using " << state.get());
+      boost::asio::dispatch(
+        state->strand,
+        [state, msg = std::move(msg)] () mutable
+        {
+          const bool empty = state->outgoing.empty();
+          state->outgoing.push_back(std::move(msg));
+          if (empty)
+            boost::asio::post(state->strand, client_loop{state});
+        }
+      );
+    }
   } // anonymous
 
   void client::queue_async(std::string&& target, epee::byte_slice&& body, std::function<server_response_func>&& notifier, boost::beast::http::verb verb) const
   {
-    message msg{std::move(body), std::move(target), std::move(notifier), verb};
     std::shared_ptr<client_state> state;
     {
       const boost::lock_guard<boost::mutex> lock{sync_};
       state = state_;
     }
-
+    queue_async_(std::move(state), std::move(target), std::move(body), std::move(notifier), verb);
-    if (!state)
-      return msg.notifier(common_error::kInvalidArgument, epee::byte_slice{});
-
-    MDEBUG("Queueing HTTP " << (msg.is_get() ? "GET" : "POST") << " to " << msg << " using " << this);
-    boost::asio::dispatch(
-      state->strand,
-      [state, msg = std::move(msg)] () mutable
-      {
-        const bool empty = state->outgoing.empty();
-        state->outgoing.push_back(std::move(msg));
-        if (empty)
-          boost::asio::post(state->strand, client_loop{state});
-      }
-    );
   }
 
   void client::direct::operator()(std::shared_ptr<client_state> self, std::function<callback_func> f) const
@@ -685,6 +698,51 @@ namespace lwsf { namespace internal { namespace http
       state_ = std::make_shared<client_state>(state_->strand.context(), state_->host, state_->prefix, state_->port, state_->ssl, proxy_);
   }
 
+  void client::ws_async(std::string target, std::string protocol, std::function<ws_func> notifier)
+  {
+    std::shared_ptr<client_state> self;
+    {
+      const boost::lock_guard<boost::mutex> lock{sync_};
+      self = state_;
+      state_ = std::make_shared<client_state>(self->strand.context(), self->host, self->prefix, self->port, self->ssl, proxy_);
+    }
+    LWSF_VERIFY(self);
+
+    struct start_ws
+    {
+      std::weak_ptr<client_state> self_;
+      std::string target_;
+      std::string protocol_;
+      std::function<ws_func> notifier_;
+
+      void operator()(const std::error_code error, epee::byte_slice)
+      {
+        const auto self = self_.lock();
+        LWSF_VERIFY(self && notifier_);
+        if (error)
+          return notifier_(error, nullptr);
+
+        std::string host;
+        boost::system::error_code ec{};
+        boost::asio::ip::make_address_v6(self->host, ec);
+        const bool https = self->ssl_status == epee::net_utils::ssl_support_t::e_ssl_support_enabled;
+        if (!ec)
+          host = "[" + host + "]:" + std::to_string(self->port);
+        else if ((https && self->port == 443) || (!https && self->port == 80))
+          host = self->host;
+        else
+          host = self->host + ":" + std::to_string(self->port);
+
+        if (self->ssl_status == epee::net_utils::ssl_support_t::e_ssl_support_enabled)
+          websocket::async_start(std::move(self->sock), self->strand.context(), std::move(host), std::move(target_), std::move(protocol_), std::move(notifier_));
+        else
+          websocket::async_start(std::move(self->sock.next_layer()), self->strand.context(), std::move(host), std::move(target_), std::move(protocol_), std::move(notifier_));
+      }
+    };
+ 
+    queue_async_(self, {}, {}, start_ws{self, std::move(target), std::move(protocol), std::move(notifier)}, boost::beast::http::verb::subscribe);
+  }
+
   void client::post_async(std::string url, epee::byte_slice json_body, std::function<server_response_func> notifier) const
   {
     return queue_async(std::move(url), std::move(json_body), std::move(notifier), boost::beast::http::verb::post);

src/net/http.h

@@ -40,7 +40,9 @@
 #include "byte_slice.h"    // monero/contrib/epee/include
 #include "net/net_ssl.h"   // monero/contrib/epee/include
 
-namespace lwsf { namespace internal { namespace http
+namespace lwsf { namespace internal { 
+namespace websocket { class stream; }
+namespace http
 {
   enum class error : int
   {
@@ -57,6 +59,9 @@ namespace lwsf { namespace internal { namespace http
   using server_response_func = void(std::error_code, epee::byte_slice);
   using callback_func = void(boost::system::error_code);
   using connect_func = void(std::shared_ptr<client_state>, std::function<callback_func>);
+  using ws_func = void(std::error_code, std::shared_ptr<websocket::stream>);
+
+  std::string make_host(const std::string& host);
 
   //! Primarily for webhooks, where the response is (basically) ignored.
   class client
@@ -107,6 +112,17 @@ namespace lwsf { namespace internal { namespace http
       `success()` is returned.
       \return `success()` if `url` is valid. */
     void get_async(std::string target, std::function<server_response_func> notifier) const;
+
+
+    /*! Never blocks. Thread safe. Makes a HTTP "upgrade" websocket request
+      using the existing HTTP connection. If the handshake succeeds, the
+      `websocket::stream` handle is sent to `notifier`, else the error is
+      reported by same function.
+
+      Any existing HTTP connection is irrevocably converted to a websocket,
+      however `post_async` and `get_async` will create a new connection, when
+      needed, using the previously supplied values to `init` and `proxy`. */
+    void ws_async(std::string target, std::string protocol, std::function<ws_func> notifier);
   };
 }}} // lwsf // internal // http
 

src/net/websocket.cpp

@@ -0,0 +1,418 @@
+// Copyright (c) 2026, The Monero Project
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without modification, are
+// permitted provided that the following conditions are met:
+//
+// 1. Redistributions of source code must retain the above copyright notice, this list of
+//    conditions and the following disclaimer.
+//
+// 2. Redistributions in binary form must reproduce the above copyright notice, this list
+//    of conditions and the following disclaimer in the documentation and/or other
+//    materials provided with the distribution.
+//
+// 3. Neither the name of the copyright holder nor the names of its contributors may be
+//    used to endorse or promote products derived from this software without specific
+//    prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY
+// EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
+// MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+// THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
+// STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+#include "websocket.h"
+
+#include <boost/asio/bind_executor.hpp>
+#include <boost/asio/buffer.hpp>
+#include <boost/asio/coroutine.hpp>
+#include <boost/asio/dispatch.hpp>
+#include <boost/asio/steady_timer.hpp>
+#include <boost/asio/strand.hpp>
+#include <boost/beast/core/error.hpp>
+#include <boost/beast/core/multi_buffer.hpp>
+#include <boost/beast/core/string_type.hpp>
+#include <boost/beast/http/message.hpp>
+#include <boost/beast/version.hpp>
+#include <boost/beast/websocket/ssl.hpp>
+#include <boost/beast/websocket/stream.hpp>
+#include <cstdint>
+#include <utility>
+
+#include "byte_slice.h"  // monero/contrib/epee/include
+#include "byte_stream.h" // monero/contrib/epee/include
+#include "error.h"
+#include "lwsf_config.h"
+
+namespace lwsf { namespace internal { namespace websocket
+{
+  using const_flat_buffer = boost::beast::net::const_buffer;
+
+  struct read_loop
+  {
+    std::shared_ptr<stream> self_;
+    std::function<update_func> receive_;
+    void operator()(boost::system::error_code, std::size_t);
+  };
+
+  struct subscription : boost::asio::coroutine
+  {
+    std::shared_ptr<stream> self_;
+    copyable_slice topic_;
+    std::function<update_func> receive_;
+
+    subscription(subscription&&) = default;
+    subscription(const subscription&) = default;
+    explicit subscription(std::shared_ptr<stream> self, epee::byte_slice&& topic, std::function<update_func>&& receive)
+      : self_(std::move(self)), topic_(std::move(topic)), receive_(std::move(receive))
+    {}
+
+    void operator()(boost::system::error_code = {});
+  };
+
+  class stream
+  { 
+    std::vector<std::function<ping_func>> pings_;
+    boost::asio::io_context::strand strand_;
+    boost::asio::steady_timer timer_;
+    boost::beast::multi_buffer buffer_;
+    bool writing_;
+    bool closing_;
+    bool subscribed_;
+
+    struct on_timeout
+    {
+      std::shared_ptr<stream> self_;
+      void operator()(const boost::system::error_code error) const
+      {
+        LWSF_VERIFY(self_);
+        if (!error || error != boost::asio::error::operation_aborted)
+        {
+          self_->notify_pings(boost::asio::error::operation_aborted);
+          self_->async_close(self_);
+        }
+      }
+    };
+
+  protected:
+
+    template<typename F>
+    auto bind(F&& f) -> decltype(boost::asio::bind_executor(strand_, std::forward<F>(f)))
+    { return boost::asio::bind_executor(strand_, std::forward<F>(f)); }
+ 
+    template<typename T, typename F>
+    void do_async_read(T& sock, F f)
+    {
+      sock.async_read(buffer_, bind(std::move(f)));
+    }
+
+    template<typename T, typename F>
+    void do_async_write(T& sock, const const_flat_buffer buf, F f)
+    {
+      if (closing_)
+        return f(boost::beast::websocket::error::closed);
+
+      LWSF_VERIFY(f.self_);
+      if (!writing_)
+      {
+        struct on_write
+        {
+          F f_;
+
+          void operator()(const boost::system::error_code error, std::size_t)
+          {
+            const std::shared_ptr<stream> self = f_.self_;
+            LWSF_VERIFY(self);
+
+            self->timer_.cancel();
+            self->writing_ = false;
+            f_(error);
+            if (error)
+            {
+              self->notify_pings(error);
+              self->async_close(self);
+            }
+            else if (!self->pings_.empty())
+              self->async_ping(on_ping{self});
+          }
+        };
+
+        writing_ = true;
+        timer_.expires_after(config::ping_timeout); // timeout for write not response
+        timer_.async_wait(on_timeout{f.self_});
+        sock.async_write(buf, bind(on_write{std::move(f)}));
+      }
+      else
+        pings_.push_back(std::bind(&stream::async_write, f.self_, std::move(f)));
+    }
+
+    template<typename T, typename F>
+    void do_async_ping(T& sock, F&& f)
+    {
+      if (closing_)
+        return f(boost::beast::websocket::error::closed);
+
+      LWSF_VERIFY(!writing_ && f.self_);
+      writing_ = true;
+      timer_.expires_after(config::ping_timeout);
+      timer_.async_wait(on_timeout{f.self_});
+      sock.async_ping(boost::beast::websocket::ping_data(), bind(std::forward<F>(f)));
+    }
+
+    template<typename T>
+    void do_async_close(T& sock, std::shared_ptr<stream> self)
+    {
+      LWSF_VERIFY(self);
+      if (!writing_)
+      {
+        const bool run = !closing_;
+        closing_ = true;
+        if (run)
+          sock.async_close(boost::beast::websocket::close_reason(), bind(std::bind(&stream::async_shutdown, self, self)));
+      }
+      else
+        pings_.push_back(std::bind(&stream::async_close, self, self));
+    }
+
+    void do_async_shutdown(tcp_sock& sock, std::shared_ptr<stream>)
+    { do_cleanup(sock); }
+
+    void do_async_shutdown(boost::asio::ssl::stream<tcp_sock>& sock, std::shared_ptr<stream> self)
+    {
+      LWSF_VERIFY(self);
+      sock.async_shutdown(bind(std::bind(&stream::cleanup, self)));
+    }
+
+    void do_cleanup(tcp_sock& sock)
+    {
+      boost::system::error_code ec{};
+      sock.shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec);
+      sock.close(ec);
+    }
+
+    void do_cleanup(boost::asio::ssl::stream<tcp_sock>& sock)
+    { do_cleanup(sock.next_layer()); }
+
+  public:
+    explicit stream(boost::asio::io_context& io)
+      : strand_(io), timer_(io), buffer_(), writing_(false), closing_(false), subscribed_(false)
+    {}
+
+    //! \return True if this function has never been invoked previously
+    bool subscribe() noexcept
+    {
+      const bool rc = !subscribed_;
+      subscribed_ = true;
+      return rc;
+    }
+
+    template<typename F>
+    void dispatch(F&& f)
+    {
+      boost::asio::dispatch(strand_, std::forward<F>(f));
+    }
+
+    epee::byte_slice gather_buffer()
+    {
+      epee::byte_stream sink{};
+      sink.reserve(buffer_.size());
+
+      const auto buffers = buffer_.cdata();
+      const auto end = boost::asio::buffer_sequence_end(buffers);
+      for (auto cur = boost::asio::buffer_sequence_begin(buffers); cur != end; ++cur)
+      {
+        const const_flat_buffer this_buf = *cur;
+        sink.write(reinterpret_cast<const std::uint8_t*>(this_buf.data()), this_buf.size());
+      }
+
+      buffer_.consume(buffer_.size());
+      return epee::byte_slice{std::move(sink)};
+    }
+
+    struct on_ping
+    {
+      std::shared_ptr<stream> self_;
+      void operator()(const boost::system::error_code& error)
+      {
+        LWSF_VERIFY(self_);
+        self_->writing_ = false;
+        if (self_->timer_.cancel() || error)
+        {
+          self_->notify_pings(error);
+          if (error)
+            self_->async_close(self_);
+        }
+      }
+    };
+
+    void notify_pings(const boost::system::error_code error = {})
+    {
+      std::vector<std::function<ping_func>> pings;
+      pings.swap(pings_);
+      for (const auto& ping : pings)
+        ping(error);
+    }
+
+    bool add_ping_notifier(std::function<ping_func>&& notifier)
+    {
+      pings_.push_back(std::move(notifier));
+      return !writing_;
+    }
+
+    virtual void async_read(read_loop&&) = 0;
+    virtual void async_write(subscription) = 0;
+    virtual void async_ping(on_ping&&) = 0;
+    virtual void async_close(std::shared_ptr<stream>) = 0;
+    virtual void async_shutdown(std::shared_ptr<stream>) = 0;
+    virtual void cleanup() = 0;
+  };
+
+  template<typename T>
+  class stream_ final : public stream
+  {
+    boost::beast::websocket::stream<T> sock_;
+
+  public:
+    explicit stream_(T&& sock, boost::asio::io_context& io)
+      : stream(io), sock_(std::move(sock))
+    {
+      sock_.binary(true);
+    }
+
+    template<typename F>
+    void async_handshake(const std::string& host, const std::string& target, std::string&& protocol, F f)
+    {
+      namespace http = boost::beast::http;
+      struct on_control
+      {
+        std::weak_ptr<stream> self_;
+        void operator()(boost::beast::websocket::frame_type, boost::beast::string_view)
+        {
+          const auto self = self_.lock();
+          if (self)
+            self->notify_pings();
+        }
+      };
+
+      LWSF_VERIFY(f.self_);
+      sock_.control_callback(on_control{f.self_});
+      sock_.set_option(
+        boost::beast::websocket::stream_base::decorator(
+          [protocol = std::move(protocol)] (http::request_header<>& hdr)
+          { hdr.set(http::field::sec_websocket_protocol, protocol); }
+        )
+      );
+      sock_.async_handshake(host, target, bind(std::move(f)));
+    }
+
+    virtual void async_read(read_loop&& handler) override final
+    { do_async_read(sock_, std::move(handler)); }
+
+    virtual void async_write(subscription handler) override final
+    {
+      std::uint8_t const* const data = handler.topic_.data.data();
+      const std::size_t size = handler.topic_.data.size();
+      do_async_write(sock_, const_flat_buffer(data, size), std::move(handler));
+    }
+
+    virtual void async_ping(on_ping&& handler) override final
+    { do_async_ping(sock_, std::move(handler)); }
+
+    virtual void async_close(std::shared_ptr<stream> self) override final
+    { do_async_close(sock_, std::move(self)); }
+
+    virtual void async_shutdown(std::shared_ptr<stream> self) override final
+    { do_async_shutdown(sock_.next_layer(), std::move(self)); }
+
+    virtual void cleanup() override final
+    { do_cleanup(sock_.next_layer()); }
+  };
+
+  struct after_connect
+  {
+    std::shared_ptr<stream> self_;
+    std::function<ready_func> notifier_;
+
+    void operator()(const boost::system::error_code error)
+    {
+      LWSF_VERIFY(notifier_);
+      if (error)
+        notifier_(error, nullptr);
+      else
+        notifier_(error, self_);
+    }
+  };
+
+  void read_loop::operator()(const boost::system::error_code error, std::size_t)
+  {
+    LWSF_VERIFY(self_ && receive_);
+    stream& self = *self_;
+    receive_(error, self.gather_buffer());
+    if (!error)
+      self.async_read(std::move(*this));
+    else
+      self.async_close(self_);
+  }
+
+  void subscription::operator()(boost::system::error_code error)
+  {
+    LWSF_VERIFY(self_ && receive_);
+    stream& self = *self_;
+    BOOST_ASIO_CORO_REENTER(*this)
+    {
+      if (self.subscribe())
+        BOOST_ASIO_CORO_YIELD self.async_write(std::move(*this));
+      else
+        error = boost::asio::error::already_started;
+
+      if (error)
+      {
+        receive_(error, epee::byte_slice{});
+        self.async_close(std::move(self_));
+      }
+      else
+        self.async_read(read_loop{std::move(self_), std::move(receive_)});
+    }
+  }
+
+  template<typename T>
+  inline void do_start(std::shared_ptr<stream_<T>> self, const std::string& host, const std::string& target, std::string&& protocol, std::function<ready_func>&& notifier)
+  {
+    self->async_handshake(host, target, std::move(protocol), after_connect{self, std::move(notifier)});
+  }
+
+  void async_start(tcp_sock&& sock, boost::asio::io_context& io, std::string&& host, std::string&& target, std::string&& protocol, std::function<ready_func>&& notifier)
+  {
+    do_start(std::make_shared<stream_<tcp_sock>>(std::move(sock), io), host, target, std::move(protocol), std::move(notifier));
+  }
+  
+  void async_start(boost::asio::ssl::stream<tcp_sock>&& sock, boost::asio::io_context& io, std::string&& host, std::string&& target, std::string&& protocol, std::function<ready_func>&& notifier)
+  {
+    do_start(std::make_shared<stream_<boost::asio::ssl::stream<tcp_sock>>>(std::move(sock), io), host, target, std::move(protocol), std::move(notifier));
+  }
+
+  void async_subscribe(std::shared_ptr<stream> self, epee::byte_slice&& topic, std::function<update_func>&& receive)
+  { 
+    LWSF_VERIFY(self);
+    self->dispatch(subscription{self, std::move(topic), std::move(receive)});
+  }
+
+  void async_ping(std::shared_ptr<stream> self, std::function<ping_func>&& notifier)
+  {
+    LWSF_VERIFY(self);
+    self->dispatch([self, notifier = std::move(notifier)] () mutable {
+      if(self->add_ping_notifier(std::move(notifier)))
+        self->async_ping(stream::on_ping{self});
+    });
+  }
+
+  void async_close(std::shared_ptr<stream> self)
+  {
+    LWSF_VERIFY(self);
+    self->dispatch(std::bind(&stream::async_close, self, self));
+  }
+}}} // lwsf // internal // websocket

src/net/websocket.h

@@ -0,0 +1,85 @@
+// Copyright (c) 2026, The Monero Project
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without modification, are
+// permitted provided that the following conditions are met:
+//
+// 1. Redistributions of source code must retain the above copyright notice, this list of
+//    conditions and the following disclaimer.
+//
+// 2. Redistributions in binary form must reproduce the above copyright notice, this list
+//    of conditions and the following disclaimer in the documentation and/or other
+//    materials provided with the distribution.
+//
+// 3. Neither the name of the copyright holder nor the names of its contributors may be
+//    used to endorse or promote products derived from this software without specific
+//    prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY
+// EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
+// MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+// THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
+// STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+#pragma once
+
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/asio/ssl.hpp>
+#include <functional>
+#include <memory>
+#include <string>
+#include <system_error>
+
+#include "byte_slice.h" // monero/contrib/epee/include
+
+namespace lwsf { namespace internal
+{
+
+  struct copyable_slice
+  {
+    copyable_slice() noexcept
+      : data(nullptr)
+    {}
+
+    copyable_slice(epee::byte_slice&& data) noexcept
+      : data(std::move(data))
+    {}
+
+    copyable_slice(copyable_slice&&) noexcept = default;
+    copyable_slice(const copyable_slice& rhs) noexcept
+      : data(rhs.data.clone())
+    {}
+
+    epee::byte_slice data;
+  };
+
+  namespace websocket  
+  { 
+    class stream;
+    using tcp_sock = boost::asio::ip::tcp::socket;
+    using ready_func = void(std::error_code, std::shared_ptr<stream>);
+    using update_func = void(std::error_code, copyable_slice);
+    using ping_func = void(std::error_code);
+
+    void async_start(tcp_sock&& sock, boost::asio::io_context& io, std::string&& host, std::string&& target, std::string&& protocol, std::function<ready_func>&& notifier);
+    void async_start(boost::asio::ssl::stream<tcp_sock>&& sock, boost::asio::io_context& io, std::string&& host, std::string&& target, std::string&& protocol, std::function<ready_func>&& notifier);
+
+    /*! Subscribe to `topic` and call `notifier` on each incoming message. Can
+      only invoke once. Thread-safe.
+      \param topic The raw message to be sent to server
+      \param notifier Called on every received message.
+      \throw std::invalid_argument if invoked multiple times. */
+    void async_subscribe(std::shared_ptr<stream> self, epee::byte_slice&& topic, std::function<update_func>&& notifier);
+
+    //! Ping the server at any time. Multiple calls allowed. Thread safe.
+    void async_ping(std::shared_ptr<stream> self, std::function<ping_func>&& notifier);
+
+    //! Terminate the stream gracefully
+    void async_close(std::shared_ptr<stream> self);
+  } // websocket
+}} // lwsf // internal

src/pending_transaction.cpp

@@ -155,10 +155,8 @@ namespace lwsf { namespace internal
           error_ = error.message();
           return false;
         }
-        const boost::lock_guard<boost::mutex> lock{data_.wallet->sync};
+
-        auto entry = data_.wallet->primary.txes.try_emplace(e->id).first;
+        data_.wallet->add_pending_tx(std::move(e));
-        if (!entry->second)
-          entry->second = std::move(e);
       }
     }
     catch(const std::exception& e)

src/wallet.cpp

@@ -59,6 +59,7 @@
 #include "net/http.h"
 #include "net/net_parse_helpers.h" // monero/contrib/epee/include
 #include "net/parse.h"         // monero/src
+#include "net/websocket.h"
 #include "numeric.h"
 #include "pending_transaction.h"
 #include "subaddress_account.h"
@@ -304,6 +305,12 @@ namespace lwsf { namespace internal
     self->set_error(error, false);
   }
 
+  void wallet::frame::async_clear_error::operator()(const std::error_code error) const
+  {
+    LWSF_VERIFY(self);
+    self->set_error(error, true);
+  }
+
   bool wallet::set_error(const std::error_code error, const bool clear) const
   {
     return status_->set_error(error, clear);
@@ -366,10 +373,16 @@ namespace lwsf { namespace internal
     {
       void operator()(wallet* val) const
       {
-        if (!val)
+        LWSF_VERIFY(val && val->data_.wallet);
-          return;
+        {
-        const boost::lock_guard<boost::mutex> lock{val->status_->sync_};
+          const boost::lock_guard<boost::mutex> lock{val->status_->sync_};
-        val->status_->thread_state_ = state::stop;
+          val->status_->thread_state_ = state::stop;
+        }
+        {
+          const boost::lock_guard<boost::mutex> lock{val->data_.wallet->sync};
+          val->data_.wallet->on_feed_refresh = nullptr;
+          val->data_.wallet->on_feed_fail = nullptr;
+        }
       }
     };
 
@@ -445,12 +458,22 @@ namespace lwsf { namespace internal
     const std::unique_ptr<wallet, set_stop_> set_stop{this};
     try
     {
+      LWSF_VERIFY(data_.wallet);
+
       std::shared_ptr<boost::asio::steady_timer> timer;
       boost::unique_lock<boost::mutex> lock{status_->sync_};
       if (status_->thread_state_ != state::stop)
       {
         timer = std::make_shared<boost::asio::steady_timer>(data_.context->io_);
-        backend::wallet::refresh(data_.wallet, false /* mandatory */, on_refresh{timer, status_, data_.wallet});
+        const boost::lock_guard<boost::mutex> lock_wallet{data_.wallet->sync};
+        if (!data_.wallet->feed_disabled)
+        {
+          data_.wallet->on_feed_fail = on_refresh{timer, status_, data_.wallet};
+          data_.wallet->on_feed_refresh = frame::async_clear_error{status_};
+          backend::wallet::refresh(data_.wallet, false /* mandatory */, frame::async_clear_error{status_}); 
+        }
+        else
+          backend::wallet::refresh(data_.wallet, false /* mandatory */, on_refresh{timer, status_, data_.wallet});
       }
       do
       {
@@ -688,7 +711,12 @@ namespace lwsf { namespace internal
       status_->exception_error_ = "view_key, spend_key, and address_string do not match";
   }
 
-  wallet::~wallet() { stop_refresh_loop(); }
+  wallet::~wallet()
+  {
+    stop_refresh_loop();
+    if (data_.wallet)
+      data_.wallet->shutdown();
+  }
 
   void wallet::add_subaddress(const std::uint32_t accountIndex, std::string label)
   {
@@ -870,7 +898,7 @@ namespace lwsf { namespace internal
         return false;
 
       if (!daemon_username.empty() || !daemon_password.empty())
-        throw std::runtime_error{"HTTP loging not supported"};
+        throw std::runtime_error{"HTTP login not supported"};
 
       // verify cert if `use_ssl == true`, otherwise autodetect if `https`
       // specified.
@@ -909,9 +937,15 @@ namespace lwsf { namespace internal
         }
       }
 
+      // backend permanently assumes `/feed` failure even after changing
+      // http server
       const boost::unique_lock<boost::mutex> lock{data_.wallet->sync};
       data_.wallet->passed_login = false;
       data_.wallet->client.init(data_.context->io_, std::move(url.host), std::move(url.uri), boost::numeric_cast<std::uint16_t>(url.port), std::move(options));
+      std::shared_ptr<websocket::stream> feed;
+      feed.swap(data_.wallet->feed);
+      if (feed)
+        websocket::async_close(std::move(feed)); // on_error
     }
     catch (const std::exception& e)
     {
@@ -947,7 +981,7 @@ namespace lwsf { namespace internal
   bool wallet::connectToDaemon()
   {
     boost::unique_lock<boost::mutex> lock{data_.wallet->sync};
-    if (data_.wallet->passed_login && data_.wallet->client.is_connected())
+    if (data_.wallet->passed_login && (data_.wallet->feed || data_.wallet->client.is_connected()))
       return true;
 
     lock.unlock();
@@ -970,7 +1004,8 @@ namespace lwsf { namespace internal
   Monero::Wallet::ConnectionStatus wallet::connected() const
   {
     const boost::lock_guard<boost::mutex> lock{data_.wallet->sync};
-    return data_.wallet->passed_login && data_.wallet->client.is_connected() ?
+    const bool feed_disabled = data_.wallet->feed_disabled;
+    return data_.wallet->passed_login && ((feed_disabled && data_.wallet->client.is_connected()) || (!feed_disabled && data_.wallet->feed)) ?
       ConnectionStatus_Connected : ConnectionStatus_Disconnected;
   }
 
@@ -1309,12 +1344,22 @@ namespace lwsf { namespace internal
           unspent_map unspent{};
           {
             wait_for<std::error_code>(
-              [this] (auto&& callable) { backend::wallet::refresh(this->data_.wallet, false, std::move(callable)); }
+              [this] (auto&& f) { backend::wallet::refresh(this->data_.wallet, false, std::move(f)); }
             );
             
-            const boost::lock_guard<boost::mutex> lock{data_.wallet->sync};
+            boost::unique_lock<boost::mutex> lock{data_.wallet->sync};
-            fee_mask = data_.wallet->fee_mask;
+            if (data_.wallet->need_fees()) // '/feed' doesn't fetch fees
+            {
+              lock.unlock();
+              const std::error_code fee_fail = wait_for<std::error_code>(
+                [this] (auto&& f) { backend::wallet::get_fees(this->data_.wallet, std::move(f)); }
+              );
+              if (fee_fail)
+                throw std::system_error{fee_fail, "Tx construction"};
+              lock.lock();
+            }
 
+            fee_mask = data_.wallet->fee_mask;
             if (data_.wallet->per_byte_fee.empty() || !fee_mask || !data_.wallet->passed_login)
               throw std::system_error{data_.wallet->refresh_error, "Tx construction"};
 

src/wallet.h

@@ -70,7 +70,11 @@ namespace internal
         void operator()(std::error_code error) const;
       };
 
-      bool thead_start();
+      struct async_clear_error
+      {
+        std::shared_ptr<frame> self;
+        void operator()(std::error_code) const;
+      };
     };
 
     const net::wallet_io data_;

tests/unit/net/CMakeLists.txt

@@ -26,6 +26,6 @@
 # STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
 # THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
-add_library(lwsf-unit-net OBJECT http.test.cpp)
+add_library(lwsf-unit-net OBJECT http.test.cpp push.test.cpp push.test.h)
 #target_include_directories(lwsf-unit-net PUBLIC ${CMAKE_CURRENT_SOURCE_DIR} "${CMAKE_SOURCE_DIR}/src")
 target_link_libraries(lwsf-unit-net PRIVATE lwsf-unit-framework lwsf-net monero::libraries)

tests/unit/net/push.test.cpp

@@ -0,0 +1,149 @@
+// Copyright (c) 2026, The Monero Project
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without modification, are
+// permitted provided that the following conditions are met:
+//
+// 1. Redistributions of source code must retain the above copyright notice, this list of
+//    conditions and the following disclaimer.
+//
+// 2. Redistributions in binary form must reproduce the above copyright notice, this list
+//    of conditions and the following disclaimer in the documentation and/or other
+//    materials provided with the distribution.
+//
+// 3. Neither the name of the copyright holder nor the names of its contributors may be
+//    used to endorse or promote products derived from this software without specific
+//    prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY
+// EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
+// MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+// THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
+// STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+#include "push.test.h"
+
+#include <boost/asio/any_io_executor.hpp>
+#include <boost/asio/bind_executor.hpp>
+#include <boost/asio/executor.hpp>
+#include <boost/asio/strand.hpp>
+#include <boost/beast/websocket/stream.hpp>
+#include <deque>
+#include <functional>
+#include <string>
+#include <utility>
+
+#include "error.h"
+
+namespace lwsf_test { namespace net { namespace push
+{
+  struct connection
+  {
+    boost::beast::websocket::stream<socket> ws_;
+    boost::asio::io_context::strand strand_;
+    std::deque<std::pair<std::string, std ::function<pub>>> msgs_;
+    boost::beast::flat_buffer buffer_;
+
+    connection(socket&& sock)
+      : ws_(std::move(sock)), strand_(ws_.next_layer().get_executor().context()), msgs_(), buffer_()
+    {
+      ws_.binary(true);
+    }
+
+    template<typename F>
+    auto bind(F&& f) -> decltype(boost::asio::bind_executor(strand_, std::forward<F>(f)))
+    {
+      return boost::asio::bind_executor(strand_, std::forward<F>(f));
+    }
+  };
+
+  namespace
+  {
+    struct on_read
+    {
+      std::shared_ptr<connection> self_;
+      std::function<sub> read_;
+      void operator()(const boost::system::error_code error, std::size_t)
+      {
+        LWSF_VERIFY(self_ && read_);
+        if (!error)
+        {
+          const auto buf = self_->buffer_.data();
+          std::string out{reinterpret_cast<const char*>(buf.data()), buf.size()};
+          self_->buffer_.consume(buf.size());
+          read_(std::move(out));
+        }
+        else
+          read_(std::error_code{error});
+      }
+    };
+
+    struct on_write
+    {
+      std::shared_ptr<connection> self_;
+      void operator()(const boost::system::error_code error, std::size_t = {})
+      {
+        LWSF_VERIFY(self_ && !self_->msgs_.empty());
+
+        std::pair<std::string, std::function<pub>> msg;
+        msg.swap(self_->msgs_.front());
+        self_->msgs_.pop_front();
+
+        if (!self_->msgs_.empty())
+          start(std::move(self_));
+
+        if (msg.second)
+          msg.second(error);
+      }
+
+      static void start(std::shared_ptr<connection> self)
+      {
+        LWSF_VERIFY(self && !self->msgs_.empty());
+        const std::string& msg = self->msgs_.front().first;
+        self->ws_.async_write(boost::asio::buffer(msg.data(), msg.size()), self->bind(on_write{self}));
+      }
+    };
+  }
+
+  void async_handshake(socket&& sock, const request& req, std::function<ready> done)
+  {
+    auto self = std::make_shared<connection>(std::move(sock));
+    self->ws_.async_accept(req, self->bind([self, done = std::move(done)] (auto error) {
+      if (error)
+        done(std::error_code{error});
+      else
+        done(self);
+    }));
+  }
+
+  void async_read(std::shared_ptr<connection> self, std::function<sub> read)
+  {
+    LWSF_VERIFY(self);
+    boost::asio::dispatch(self->strand_, [self, read = std::move(read)] () mutable {
+      self->ws_.async_read(self->buffer_, self->bind(on_read{self, std::move(read)}));
+    });
+  }
+
+  void async_pub(std::shared_ptr<connection> self, std::string msg, std::function<pub> sent)
+  {
+    LWSF_VERIFY(self);
+    boost::asio::dispatch(self->strand_, [self, msg = std::move(msg), sent = std::move(sent)] () mutable {
+      const bool empty = self->msgs_.empty();
+      self->msgs_.push_back({std::move(msg), std::move(sent)});
+      if (empty)
+        on_write::start(std::move(self));
+    });
+  }
+
+  void async_close(std::shared_ptr<connection> self)
+  {
+    LWSF_VERIFY(self);
+    boost::asio::dispatch(self->strand_, [self] {
+      self->ws_.async_close(boost::beast::websocket::close_reason(), [] (auto) {});
+    });
+  }
+}}} // lwsf_test // net // push

tests/unit/net/push.test.h

@@ -0,0 +1,60 @@
+// Copyright (c) 2026, The Monero Project
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without modification, are
+// permitted provided that the following conditions are met:
+//
+// 1. Redistributions of source code must retain the above copyright notice, this list of
+//    conditions and the following disclaimer.
+//
+// 2. Redistributions in binary form must reproduce the above copyright notice, this list
+//    of conditions and the following disclaimer in the documentation and/or other
+//    materials provided with the distribution.
+//
+// 3. Neither the name of the copyright holder nor the names of its contributors may be
+//    used to endorse or promote products derived from this software without specific
+//    prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY
+// EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
+// MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+// THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
+// STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+#pragma once
+
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/beast/http/message.hpp>
+#include <boost/beast/http/string_body.hpp>
+#include <boost/system/error_code.hpp>
+#include <memory>
+#include <string>
+#include <system_error>
+
+#include "common/expect.h"
+
+namespace lwsf_test { namespace net { namespace push
+{
+  using socket =
+    boost::asio::basic_stream_socket<
+      boost::asio::ip::tcp,
+      boost::asio::io_context::executor_type
+    >;
+  using request = boost::beast::http::request<boost::beast::http::string_body>;
+
+  struct connection;
+  using ready = void(expect<std::shared_ptr<connection>>);
+  using sub = void(expect<std::string>);
+  using pub = void(std::error_code);
+
+  void async_handshake(socket&& sock, const request& req, std::function<ready> done);
+  void async_read(std::shared_ptr<connection> self, std::function<sub> read);
+  void async_pub(std::shared_ptr<connection> self, std::string msg, std::function<pub> sent);
+  void async_close(std::shared_ptr<connection> self);
+    
+}}} // lwsf_test // net // push