Add mempool support to /get_address_txs and /feed (#251)

Co-authored-by: William Swanson <swansontec@gmail.com>

src/CMakeLists.txt

@@ -71,7 +71,7 @@ set(monero-lws-common_headers config.h error.h fwd.h)
 add_library(monero-lws-common ${monero-lws-common_sources} ${monero-lws-common_headers})
 target_link_libraries(monero-lws-common monero::libraries)
 
-add_library(monero-lws-daemon-common rest_server.cpp scanner.cpp)
+add_library(monero-lws-daemon-common rest_server.cpp scanner.cpp mempool.cpp)
 target_include_directories(monero-lws-daemon-common PUBLIC ${ZMQ_INCLUDE_PATH})
 target_link_libraries(monero-lws-daemon-common
   PUBLIC

src/db/data.cpp

@@ -541,6 +541,22 @@ namespace db
     );
   }
 
+  std::size_t get_hash(account_address const& value) noexcept
+  {
+    /* use view_public - it has good distribution and cannot be selected
+      by user (and has been verified as such at login). */
+    std::size_t out;
+    static_assert(sizeof(out) <= sizeof(value.view_public));
+    static_assert(std::is_trivially_copyable<account_address>::value);
+    std::memcpy(std::addressof(out), std::addressof(value.view_public), sizeof(out));
+    return out;
+  }
+  bool operator==(account_address const& left, account_address const& right) noexcept
+  {
+    static_assert(std::is_trivially_copyable<account_address>::value);
+    return std::memcmp(std::addressof(left), std::addressof(right), sizeof(left)) == 0;
+  }
+
   bool operator<(const webhook_dupsort& left, const webhook_dupsort& right) noexcept
   {
     return left.payment_id == right.payment_id ?

src/db/data.h

@@ -433,6 +433,13 @@ namespace db
   };
   void write_bytes(wire::writer&, const webhook_new_account&);
 
+  std::size_t get_hash(account_address const& value) noexcept;
+  bool operator==(account_address const& left, account_address const& right) noexcept;
+  inline bool operator!=(account_address const& left, account_address const& right) noexcept
+  {
+    return !(left == right);
+  }
+
   inline constexpr bool operator==(address_index const& left, address_index const& right) noexcept
   {
     return left.maj_i == right.maj_i && left.min_i == right.min_i;
@@ -508,3 +515,13 @@ namespace wire
     : std::true_type
   {};
 }
+
+namespace std
+{
+  template<>
+  struct hash<lws::db::account_address>
+  {
+    std::size_t operator()(lws::db::account_address const& value) const noexcept
+    { return lws::db::get_hash(value); }
+  };
+}

src/fwd.h

@@ -30,6 +30,7 @@
 namespace lws
 {
   class account;
+  class mempool;
   class rest_server;
   class scanner;
 }

src/mempool.cpp

@@ -0,0 +1,215 @@
+// Copyright (c) 2024, 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 "mempool.h"
+
+#include <algorithm>
+#include <boost/numeric/conversion/cast.hpp>
+#include <boost/thread/lock_guard.hpp>
+#include "common/error.h"
+#include "cryptonote_basic/cryptonote_format_utils.h"
+#include "error.h"
+#include "misc_log_ex.h"
+#include "rpc/daemon_messages.h" // external/monero/src
+#include "rpc/daemon_zmq.h"
+#include "rpc/json.h"
+#include "util/ownership_test.h"
+
+namespace lws
+{
+  const size_t max_cache_size = 1000;
+
+  void mempool::add_txs(epee::span<cryptonote::transaction> txs)
+  {
+    const auto now = std::chrono::system_clock::now();
+    const boost::lock_guard<boost::mutex> lock{mutex_};
+    snapshot_.reset();
+
+    for (auto& tx: txs)
+    {
+      auto found = state_.try_emplace(get_transaction_hash(tx), nullptr);
+      if (!found.first->second)
+        found.first->second = std::make_shared<pool_entry>(std::move(tx), now);
+    }
+  }
+
+  void mempool::reset_txs(pool_table&& txs)
+  {
+    const boost::lock_guard<boost::mutex> lock{mutex_};
+    snapshot_.reset();
+
+    MDEBUG("Mempool reset from size " << state_.size() << " to " << txs.size());
+    state_ = std::move(txs);
+  }
+
+  std::unordered_set<crypto::hash> mempool::address_cache::get(const db::account_address& address) const
+  {
+    auto it = map_.find(address);
+    if (it == map_.end())
+      return {};
+    return {it->second.txids.begin(), it->second.txids.end()};
+  }
+
+  void mempool::address_cache::set(const db::account_address& address, std::vector<crypto::hash>&& txids)
+  {
+    // find or create entry for this address
+    auto status = map_.emplace(address, entry{});
+    entry& e = status.first->second;
+    e.txids = std::move(txids);
+    if (status.second)
+    {
+      // add new address to order list
+      order_.push_front(address);
+      e.order = order_.begin();
+
+      // evict old address if needed
+      if (map_.size() >= max_cache_size)
+      {
+        auto& evict_address = order_.back();
+        map_.erase(evict_address);
+        order_.pop_back();
+      }
+    }
+    else
+      // move existing address to front of order list
+      order_.splice(order_.begin(), order_, e.order);
+  }
+
+  std::vector<found_pool_tx> mempool::scan_account(lws::account& user) const
+  {
+    std::unordered_set<crypto::hash> skip;
+    std::shared_ptr<const pool_snapshot> snapshot;
+    {
+      const boost::lock_guard<boost::mutex> lock{mutex_};
+      skip = cache_.get(user.db_address());
+
+      if (!snapshot_)
+      {
+         // otherwise make a fresh snapshot
+        pool_snapshot temp;
+        temp.resize(state_.size());
+        std::copy(state_.begin(), state_.end(), temp.begin());
+        snapshot_ = std::make_shared<const pool_snapshot>(std::move(temp));
+      }
+      snapshot = snapshot_;
+    }
+
+    std::vector<found_pool_tx> found{};
+    ownership_test scan_transaction{
+      [&found](account&, db::spend&& spend)
+      {
+        if (found.empty() || found.back().hash != spend.link.tx_hash)
+          found.push_back({spend.link.tx_hash});
+        found.back().spends.push_back(std::move(spend));
+      },
+      [&found](account&, db::output&& output)
+      {
+        if (found.empty() || found.back().hash != output.link.tx_hash)
+          found.push_back({output.link.tx_hash});
+        found.back().outputs.push_back(std::move(output));
+      }
+    };
+
+    // uint64::max is for txpool
+    static const std::vector<std::uint64_t> fake_outs(
+      256, std::numeric_limits<std::uint64_t>::max()
+    );
+
+    std::vector<crypto::hash> skipped;
+    for (const auto& pair: *snapshot) {
+      if (!skip.count(pair.first))
+        scan_transaction(
+          epee::span<lws::account>(&user, 1),
+          db::block_id::txpool,
+          boost::numeric_cast<std::uint64_t>(std::chrono::system_clock::to_time_t(pair.second->timestamp)),
+          std::addressof(pair.first),
+          pair.second->tx,
+          fake_outs
+        );
+      if (found.empty() || found.back().hash != pair.first)
+        skipped.push_back(pair.first);
+    }
+
+    {
+      const boost::lock_guard<boost::mutex> lock{mutex_};
+      cache_.set(user.db_address(), std::move(skipped));
+    }
+
+    return found;
+  }
+
+  expect<void> pool_update_loop(std::shared_ptr<mempool> pool, rpc::client& client)
+  {
+    LWS_VERIFY(pool);
+    return client.event_loop(
+      [pool](std::string&& json) -> expect<void>
+      {
+        auto msg = rpc::parse_json_response<rpc::get_transaction_pool>(std::move(json));
+        if (!msg)
+        {
+          MERROR("Pool failed to parse block response" << msg.error());
+          return msg.error();
+        }
+
+        mempool::pool_table txs{};
+        txs.reserve(msg->transactions.size());
+        const auto now = std::chrono::system_clock::now();
+        for (auto& tx: msg->transactions)
+        {
+          auto shared_tx = std::make_shared<mempool::pool_entry>(std::move(tx.tx), now);
+          txs.emplace(tx.tx_hash, std::move(shared_tx));
+        }
+        pool->reset_txs(std::move(txs));
+        return {};
+      },
+
+      [&client](rpc::minimal_chain_pub&&) -> expect<void>
+      {
+        cryptonote::rpc::GetTransactionPool::Request req{};
+
+        expect<void> status = success();
+        for (unsigned i = 0; i < 2; ++i)
+        {
+          status = client.send(
+            rpc::client::make_message("get_transaction_pool", req), std::chrono::seconds(0)
+          );
+          if (status || status != net::zmq::make_error_code(EFSM))
+            return status;
+          MONERO_CHECK(client.daemon_reconnect());
+        }
+        return status;
+      },
+
+      [pool](rpc::full_txpool_pub&& msg) -> expect<void>
+      {
+        pool->add_txs(epee::to_mut_span(msg.txes));
+        return {};
+      }
+    );
+  }
+
+} // namespace lws

src/mempool.h

@@ -0,0 +1,116 @@
+// Copyright (c) 2024, 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/thread/mutex.hpp>
+#include <chrono>
+#include <memory>
+#include <unordered_map>
+#include <vector>
+
+#include "common/expect.h"                     // external/monero/src
+#include "cryptonote_basic/cryptonote_basic.h" // external/monero/src
+#include "db/data.h"
+#include "db/fwd.h"
+#include "rpc/client.h"
+
+namespace lws
+{
+  struct found_pool_tx {
+    crypto::hash hash;
+    std::vector<db::spend> spends;
+    std::vector<db::output> outputs;
+  };
+
+  /*!
+    Thread-safe mempool cache, designed to perform well with a single writer
+    and multiple concurrent readers.
+
+    The primary mutable state is guarded with a mutex, while each reader
+    receives an immutable snapshot of the latest state. These snapshots
+    use structural sharing to avoid copying the heavyweight transactions.
+
+    The two update methods are `add_tx` and `filter_txs`.
+  */
+  class mempool
+  {
+  public:
+    struct pool_entry
+    {
+      pool_entry(cryptonote::transaction&& tx, const std::chrono::system_clock::time_point timestamp)
+        : tx(std::move(tx)), timestamp(timestamp)
+      {}
+
+      cryptonote::transaction tx;
+      std::chrono::system_clock::time_point timestamp;
+    };
+
+    using pool_snapshot = std::vector<std::pair<crypto::hash, std::shared_ptr<const pool_entry>>>;
+    using pool_table = std::unordered_map<crypto::hash, std::shared_ptr<pool_entry>>;
+
+    //! Adds transactions to the pool, if they are not already present.
+    // Thread-safe.
+    void add_txs(epee::span<cryptonote::transaction>);
+
+    //! Replaces all transactions to the pool. Thread-safe.
+    void reset_txs(pool_table&&);
+
+    //! Scans the pool for transactions belonging to an address. Thread-safe.
+    std::vector<found_pool_tx> scan_account(lws::account& user) const;
+
+  private:
+    //! Remembers which txid's that do *not* belong to an address,
+    // since there is no need for the scanner to revisit these.
+    class address_cache
+    {
+    public:
+      std::unordered_set<crypto::hash> get(const db::account_address& address) const;
+      void set(const db::account_address& address, std::vector<crypto::hash>&& txids);
+
+    private:
+      struct entry
+      {
+        std::vector<crypto::hash> txids;
+        std::list<db::account_address>::iterator order;
+      };
+
+      std::unordered_map<db::account_address, entry> map_;
+      std::list<db::account_address> order_;
+    };
+
+    pool_table state_;
+    mutable address_cache cache_;
+    mutable std::shared_ptr<const pool_snapshot> snapshot_;
+    mutable boost::mutex mutex_;
+  };
+
+  //! Run this on a thread to keep the pool in sync.
+  // Loops until an error or shutdown signal occurs.
+  expect<void> pool_update_loop(std::shared_ptr<mempool> pool, rpc::client& client);
+
+} // namespace lws

src/rest_server.cpp

@@ -65,12 +65,14 @@
 #include "common/expect.h"         // monero/src
 #include "config.h"
 #include "crypto/crypto.h"         // monero/src
+#include "cryptonote_basic/cryptonote_format_utils.h" // monero/srcqq
 #include "cryptonote_config.h"     // monero/src
 #include "db/data.h"
 #include "db/storage.h"
 #include "db/string.h"
 #include "error.h"
 #include "lmdb/util.h"             // monero/src
+#include "mempool.h"
 #include "net/http/client.h"
 #include "net/http/slice_body.h"
 #include "net/net_parse_helpers.h" // monero/contrib/epee/include
@@ -86,7 +88,9 @@
 #include "rpc/login.h"
 #include "rpc/rates.h"
 #include "rpc/webhook.h"
+#include "string_tools.h"          // monero/contrib/epee/include
 #include "util/gamma_picker.h"
+#include "util/ownership_test.h"
 #include "util/random_outputs.h"
 #include "util/source_location.h"
 #include "wire/adapted/crypto.h"
@@ -118,15 +122,17 @@ namespace lws
     boost::asio::io_context io;
     const db::storage disk;
     const rpc::client client;
+    const std::shared_ptr<lws::mempool> mempool;
     const runtime_options options;
     std::vector<net::zmq::async_client> clients;
     net::http::client webhook_client;
     boost::mutex sync;
 
-    rest_server_data(db::storage disk, rpc::client client, runtime_options options)
+    rest_server_data(db::storage disk, rpc::client client, std::shared_ptr<lws::mempool> mempool, runtime_options options)
       : io(),
         disk(std::move(disk)),
         client(std::move(client)),
+        mempool(std::move(mempool)),
         options(std::move(options)),
         webhook_client(options.webhook_verify),
         clients(),
@@ -551,7 +557,7 @@ namespace lws
           return user.error();
 
         data.passed_login = true;
-        return response::load(user->second, user->first, false);
+        return response::load(user->second, user->first, data.global->mempool.get(), false);
       }
     };
 
@@ -1433,6 +1439,10 @@ namespace lws
       {
         using transaction_rpc = cryptonote::rpc::SendRawTxHex;
 
+        cryptonote::blobdata tx_blob;
+        if (!epee::string_tools::parse_hexstr_to_binbuff(req.tx, tx_blob))
+          return {lws::error::bad_client_tx};
+
         struct frame
         {
           rest_server_data* parent;
@@ -1440,7 +1450,7 @@ namespace lws
           net::zmq::async_client client;
           boost::asio::steady_timer timer;
           boost::asio::io_context::strand strand;
-          std::deque<std::pair<epee::byte_slice, std::function<async_complete>>> resumers;
+          std::deque<std::tuple<epee::byte_slice, std::function<async_complete>, cryptonote::transaction>> resumers;
           std::size_t outstanding;
 
           frame(rest_server_data& parent, net::zmq::async_client client)
@@ -1480,7 +1490,11 @@ namespace lws
             return {error::load};
 
           active->outstanding += msg.size(); 
-          active->resumers.emplace_back(std::move(msg), std::move(resume));
+          auto& elem = active->resumers.emplace_back();
+          std::get<0>(elem) = std::move(msg);
+          std::get<1>(elem) = std::move(resume);
+          if (!cryptonote::parse_and_validate_tx_from_blob(tx_blob, std::get<2>(elem)))
+            return {lws::error::bad_client_tx};
           return success();
         }
 
@@ -1497,7 +1511,7 @@ namespace lws
             assert(self_ != nullptr);
             assert(self_->strand.running_in_this_thread());
 
-            std::deque<std::pair<epee::byte_slice, std::function<async_complete>>> resumers;
+            std::deque<std::tuple<epee::byte_slice, std::function<async_complete>, cryptonote::transaction>> resumers;
             {
               const boost::lock_guard<boost::mutex> lock{cache.sync};
               if (error)
@@ -1510,6 +1524,9 @@ namespace lws
               }
               else
               {
+                if (self_->parent && self_->parent->mempool)
+                  self_->parent->mempool->add_txs({std::addressof(std::get<2>(self_->resumers.front())), 1});
+
                 MDEBUG("Completed ZMQ request in /submit_raw_tx");
                 resumers.push_back(std::move(self_->resumers.front()));
                 self_->resumers.pop_front();
@@ -1517,7 +1534,7 @@ namespace lws
             }
 
             for (const auto& r : resumers)
-              r.second(value);
+              std::get<1>(r)(value);
           }
 
           bool set_timeout(std::chrono::steady_clock::duration timeout, const bool expecting) const
@@ -1568,7 +1585,7 @@ namespace lws
                     self_->parent->store_async_client(std::move(self_->client));
                     return;
                   }
-                  next = std::move(self_->resumers.front().first);
+                  next = std::move(std::get<0>(self_->resumers.front()));
                   self_->outstanding -= std::min(self_->outstanding, next.size());
                 }
 
@@ -1613,7 +1630,11 @@ namespace lws
         cache.status = active;
 
         active->outstanding += msg.size();
-        active->resumers.emplace_back(std::move(msg), std::move(resume));
+        auto& elem = active->resumers.emplace_back();
+        std::get<0>(elem) = std::move(msg);
+        std::get<1>(elem) = std::move(resume);
+        if (!cryptonote::parse_and_validate_tx_from_blob(tx_blob, std::get<2>(elem)))
+          return {lws::error::bad_client_tx};
         lock.unlock();
 
         MDEBUG("Starting new ZMQ request in /submit_raw_tx");
@@ -2072,7 +2093,7 @@ namespace lws
         if (!feed_disabled && boost::beast::websocket::is_upgrade(self_->parser_->get()))
         {
           MDEBUG("Attempting websocket '" << feed->name << "' on " << self_.get());
-          if (!rpc::feed::start(std::move(self_->sock()), from->io, from->client, self_->parser_->get(), from->disk, feed_timeout))
+          if (!rpc::feed::start(std::move(self_->sock()), from->io, from->client, from->mempool, self_->parser_->get(), from->disk, feed_timeout))
             return self_->bad_request(boost::beast::http::status::bad_request, std::forward<F>(resume));
           return; // else a websocket has I/O queued on `io`.
         }
@@ -2240,8 +2261,8 @@ namespace lws
     }
   }
 
-  rest_server::rest_server(epee::span<const std::string> addresses, std::vector<std::string> admin, db::storage disk, rpc::client client, configuration config)
-    : global_(std::make_unique<rest_server_data>(std::move(disk), std::move(client), runtime_options{config})),
+  rest_server::rest_server(epee::span<const std::string> addresses, std::vector<std::string> admin, db::storage disk, rpc::client client, std::shared_ptr<lws::mempool> mempool, configuration config)
+    : global_(std::make_unique<rest_server_data>(std::move(disk), std::move(client), std::move(mempool), runtime_options{config})),
       ports_(),
       workers_()
   {

src/rest_server.h

@@ -41,6 +41,7 @@
 
 namespace lws
 {
+  class mempool;
   struct rest_server_data;
   class rest_server
   {
@@ -70,7 +71,13 @@ namespace lws
       bool auto_accept_import;
     };
 
-    explicit rest_server(epee::span<const std::string> addresses, std::vector<std::string> admin, db::storage disk, rpc::client client, configuration config);
+    explicit rest_server(
+      epee::span<const std::string> addresses,
+      std::vector<std::string> admin,
+      db::storage disk,
+      rpc::client client,
+      std::shared_ptr<lws::mempool> mempool,
+      configuration config);
 
     rest_server(rest_server&&) = delete;
     rest_server(rest_server const&) = delete;

src/rpc/client.cpp

@@ -27,6 +27,7 @@
 
 #include "client.h"
 
+#include <array>
 #include <boost/algorithm/string/predicate.hpp>
 #include <boost/thread/mutex.hpp>
 #include <boost/utility/string_ref.hpp>
@@ -433,6 +434,15 @@ namespace rpc
     return ctx && (ctx->external_pub || ctx->rmq.is_available());
   }
 
+  expect<void> client::daemon_reconnect() noexcept
+  {
+    auto new_daemon = make_daemon(ctx);
+    if (!new_daemon)
+      return new_daemon.error();
+    daemon = std::move(*new_daemon);
+    return success();
+  }
+
   expect<void> client::watch_scan_signals() noexcept
   {
     MONERO_PRECOND(ctx != nullptr);
@@ -502,6 +512,111 @@ namespace rpc
     return {lws::error::bad_daemon_response};
   }
 
+  expect<void> client::event_loop(rpc_handler on_rpc, block_pub_handler on_block,  txpool_pub_handler on_txpool)
+  {
+    MONERO_PRECOND(ctx != nullptr);
+    assert(signal_sub != nullptr);
+
+    std::array<zmq_pollitem_t, 3> poll_items{};
+    std::size_t count = 0;
+
+    poll_items[count++] = {signal_sub.get(), 0, short(ZMQ_POLLIN | ZMQ_POLLERR), 0};
+    if (daemon != nullptr)
+      poll_items[count++] = {daemon.get(), 0, short(ZMQ_POLLIN | ZMQ_POLLERR), 0};
+    if (daemon_sub != nullptr)
+      poll_items[count++] = {daemon_sub.get(), 0, short(ZMQ_POLLIN | ZMQ_POLLERR), 0};
+
+    while (true)
+    {
+      const int ready = zmq_poll(poll_items.data(), count, -1);
+      if (ready < 0)
+      {
+        const int err = zmq_errno();
+        if (err == EINTR)
+          continue;
+        MERROR("Failed polling sockets: " << err);
+        return net::zmq::make_error_code(err);
+      }
+
+      // check for abort messages:
+      if (poll_items[0].revents & (ZMQ_POLLIN | ZMQ_POLLERR))
+      {
+        char buf[1];
+        MONERO_ZMQ_CHECK(zmq_recv(signal_sub.get(), buf, 1, 0));
+        switch (buf[0])
+        {
+        case 'P':
+          return {lws::error::signal_abort_process};
+        case 'S':
+          return {lws::error::signal_abort_scan};
+        default:
+          return {lws::error::signal_unknown};
+        }
+      }
+
+      // check for rpc responses:
+      if (daemon && (poll_items[1].revents & (ZMQ_POLLIN | ZMQ_POLLERR)))
+      {
+        auto json = net::zmq::receive(daemon.get(), ZMQ_DONTWAIT);
+        if (!json)
+        {
+          if (json != net::zmq::make_error_code(EAGAIN))
+          {
+            MERROR("Failed reading RPC response: " << json.error());
+            return json.error();
+          }
+        } else if (on_rpc)
+          MONERO_CHECK(on_rpc(std::move(*json)));
+      }
+
+      // check for PUB notificatons:
+      std::size_t sub_index = daemon ? 2 : 1;
+      if (daemon_sub && (poll_items[sub_index].revents & (ZMQ_POLLIN | ZMQ_POLLERR)))
+      {
+        while (true)
+        {
+          auto json = net::zmq::receive(daemon_sub.get(), ZMQ_DONTWAIT);
+          if (!json)
+          {
+            if (json == net::zmq::make_error_code(EAGAIN))
+              break;
+            MERROR("Failed reading pub message: " << json.error());
+            return json.error();
+          }
+
+          if (boost::string_ref{*json}.starts_with(minimal_chain_topic))
+          {
+            json->erase(0, sizeof(minimal_chain_topic));
+            auto parsed = rpc::minimal_chain_pub::from_json(std::move(*json));
+            if (!parsed)
+            {
+              MERROR("Failed parsing chain pub: " << parsed.error().message());
+              return parsed.error();
+            }
+            if (on_block)
+              MONERO_CHECK(on_block(std::move(*parsed)));
+          }
+          else if (boost::string_ref{*json}.starts_with(full_txpool_topic))
+          {
+            json->erase(0, sizeof(full_txpool_topic));
+            auto parsed = rpc::full_txpool_pub::from_json(std::move(*json));
+            if (!parsed)
+            {
+              MERROR("Failed parsing txpool pub: " << parsed.error().message());
+              return parsed.error();
+            }
+            if (on_txpool)
+              MONERO_CHECK(on_txpool(std::move(*parsed)));
+          }
+          else
+            return {lws::error::bad_daemon_response};
+        }
+      }
+    }
+
+    return {};
+  }
+
   expect<account_sub> client::make_account_sub(boost::asio::io_context& io) const
   {
     MONERO_PRECOND(ctx != nullptr);

src/rpc/client.h

@@ -29,6 +29,7 @@
 #include <boost/asio/io_context.hpp>
 #include <boost/optional/optional.hpp>
 #include <chrono>
+#include <functional>
 #include <memory>
 #include <string>
 #include <utility>
@@ -158,12 +159,19 @@ namespace rpc
     //! \return True if an external pub/sub was setup
     bool has_publish() const noexcept;
 
+    //! Reconnect ZMQ connection to daemon
+    expect<void> daemon_reconnect() noexcept;
+
     //! `wait`, `send`, and `receive` will watch for `raise_abort_scan()`.
     expect<void> watch_scan_signals() noexcept;
 
     //! Wait for new block announce or internal timeout.
     expect<std::vector<std::pair<topic, std::string>>> wait_for_block();
 
+    using rpc_handler = std::function<expect<void>(std::string&&)>;
+    using block_pub_handler = std::function<expect<void>(minimal_chain_pub&&)>;
+    using txpool_pub_handler = std::function<expect<void>(full_txpool_pub&&)>;
+
     //! \return A JSON message for RPC request `M`.
     template<typename M>
     static epee::byte_slice make_message(char const* const name, const M& message)
@@ -171,6 +179,13 @@ namespace rpc
       return cryptonote::rpc::FullMessage::getRequest(name, message, 0);
     }
 
+    //! Loops until an error or shutdown happens, emitting events.
+    expect<void> event_loop(
+      rpc_handler on_rpc,
+      block_pub_handler on_block,
+      txpool_pub_handler on_txpool
+    );
+
     //! \return An async ZMQ_SUB socket that is bound to the account update publisher
     expect<account_sub> make_account_sub(boost::asio::io_context& io) const;
 

src/rpc/feed.h

@@ -34,6 +34,7 @@
 #include <boost/beast/http/string_body.hpp>
 #include <chrono>
 #include <cstdint>
+#include <memory>
 #include <system_error>
 
 #include "db/fwd.h"
@@ -72,8 +73,8 @@ namespace lws { namespace rpc { namespace feed
     return std::error_code{int(value), error_category()};
   }
 
-  bool start(boost::asio::ip::tcp::socket&& sock, boost::asio::io_context& io, const lws::rpc::client& client, const request& req, const lws::db::storage& disk, std::chrono::seconds timeout);
-  bool start(boost::asio::ssl::stream<boost::asio::ip::tcp::socket>&& sock, boost::asio::io_context& io, const lws::rpc::client& client, const request& req, const lws::db::storage& disk, std::chrono::seconds idle_timeout);
+  bool start(boost::asio::ip::tcp::socket&& sock, boost::asio::io_context& io, const lws::rpc::client& client, std::shared_ptr<const mempool> pool, const request& req, const lws::db::storage& disk, std::chrono::seconds timeout);
+  bool start(boost::asio::ssl::stream<boost::asio::ip::tcp::socket>&& sock, boost::asio::io_context& io, const lws::rpc::client& client, std::shared_ptr<mempool> pool, const request& req, const lws::db::storage& disk, std::chrono::seconds idle_timeout);
 }}} // lws // rpc // feed
 
 namespace std

src/rpc/feed.inl

@@ -171,12 +171,13 @@ namespace lws { namespace rpc { namespace feed
   // defined in `feed_tcp.cpp`
   std::string_view get_prefix(const boost::beast::net::const_buffer buf);
   expect<epee::byte_slice> prep_error(std::error_code error, protocol proto);
-  expect<epee::byte_slice> prep_login(const db::storage& disk, account_sub* sub, connection_sync* sync, boost::beast::flat_buffer& buffer, protocol proto);
+  expect<epee::byte_slice> prep_login(const db::storage& disk, const mempool* pool, account_sub* sub, connection_sync* sync, boost::beast::flat_buffer& buffer, protocol proto);
   expect<epee::byte_slice> prep_update(const db::storage& disk, std::string&& source, connection_sync* sync, protocol proto);
 
   class connection
   {
     lws::db::storage disk_;
+    const std::shared_ptr<const mempool> pool_;
     std::string sub_buffer_;
     boost::beast::flat_buffer ws_buffer_;
     std::deque<epee::byte_slice> write_queue_;
@@ -324,8 +325,9 @@ namespace lws { namespace rpc { namespace feed
     }
 
    public: 
-    connection(boost::asio::io_context& io, const lws::rpc::client& client, const lws::db::storage& disk, const protocol proto)
+    connection(boost::asio::io_context& io, const lws::rpc::client& client, std::shared_ptr<const mempool> pool, const lws::db::storage& disk, const protocol proto)
       : disk_(disk.clone()),
+        pool_(std::move(pool)),
         sub_buffer_(),
         ws_buffer_(),
         write_queue_(),
@@ -347,7 +349,7 @@ namespace lws { namespace rpc { namespace feed
 
     bool login(std::shared_ptr<connection> self)
     {
-      const bool rc = do_write(std::move(self), prep_login(disk_, &sub_, &sync_, ws_buffer_, proto_));
+      const bool rc = do_write(std::move(self), prep_login(disk_, pool_.get(), &sub_, &sync_, ws_buffer_, proto_));
       ws_buffer_.consume(ws_buffer_.cdata().size());
       return rc;
     }
@@ -465,8 +467,8 @@ namespace lws { namespace rpc { namespace feed
     { do_async_shutdown(sock().next_layer(), std::move(self)); }
 
   public:
-    explicit connection_(T&& sock, boost::asio::io_context& io, const lws::rpc::client& client, const lws::db::storage& disk, const protocol proto)
-      : connection(io, client, disk, proto), sock_(std::move(sock))
+    explicit connection_(T&& sock, boost::asio::io_context& io, const lws::rpc::client& client, std::shared_ptr<const mempool> pool, const lws::db::storage& disk, const protocol proto)
+      : connection(io, client, std::move(pool), disk, proto), sock_(std::move(sock))
     {
       if (is_binary(proto))
         sock_.binary(true);
@@ -581,14 +583,14 @@ namespace lws { namespace rpc { namespace feed
   }
 
   template<typename T>
-  inline bool do_start(T&& sock, boost::asio::io_context& io, const lws::rpc::client& client, const request& req, const lws::db::storage& disk, const std::chrono::seconds timeout)
+  inline bool do_start(T&& sock, boost::asio::io_context& io, const lws::rpc::client& client, std::shared_ptr<const mempool> pool, const request& req, const lws::db::storage& disk, const std::chrono::seconds timeout)
   {
     const protocol proto = get_protocol(req.base()[boost::beast::http::field::sec_websocket_protocol]);
     switch (proto)
     {
     case protocol::v0_msgpack:
     case protocol::v0_json:
-      do_start(std::make_shared<connection_<T>>(std::forward<T>(sock), io, client, disk, proto), req, timeout);
+      do_start(std::make_shared<connection_<T>>(std::forward<T>(sock), io, client, std::move(pool), disk, proto), req, timeout);
       return true;
     default:
     case protocol::invalid:

src/rpc/feed_ssl.cpp

@@ -31,11 +31,11 @@
 
 namespace lws { namespace rpc { namespace feed
 {
-  bool start(boost::asio::ssl::stream<boost::asio::ip::tcp::socket>&& sock, boost::asio::io_context& io, const lws::rpc::client& client, const request& req, const lws::db::storage& disk, const std::chrono::seconds timeout)
+  bool start(boost::asio::ssl::stream<boost::asio::ip::tcp::socket>&& sock, boost::asio::io_context& io, const lws::rpc::client& client, std::shared_ptr<const mempool> pool, const request& req, const lws::db::storage& disk, const std::chrono::seconds timeout)
   {
     // moving ssl streams only supported in boost 1.74+
 #if BOOST_VERSION >= 107400 
-    return do_start(std::move(sock), io, client, req, disk, timeout);
+    return do_start(std::move(sock), io, client, std::move(pool), req, disk, timeout);
 #else
     return false;
 #endif

src/rpc/feed_tcp.cpp

@@ -231,7 +231,7 @@ namespace lws { namespace rpc { namespace feed
     return prep(feed_error{error}, proto);
   }
 
-  expect<epee::byte_slice> prep_login(const db::storage& disk, account_sub* const sub, connection_sync* const sync, boost::beast::flat_buffer& buffer, const protocol proto)
+  expect<epee::byte_slice> prep_login(const db::storage& disk, mempool const* const pool, account_sub* const sub, connection_sync* const sync, boost::beast::flat_buffer& buffer, const protocol proto)
   {
     LWS_VERIFY(sub && sync);
     const std::string_view prefix = get_prefix(buffer.cdata());
@@ -271,7 +271,7 @@ namespace lws { namespace rpc { namespace feed
       return prep(feed_blocks{acct.start_height, acct.scan_height, block->id, acct.lookahead_fail, acct.lookahead}, proto);
     }
 
-    auto txs = get_address_txs_response::load(account->second, account->first, true);
+    auto txs = get_address_txs_response::load(account->second, account->first, pool, true);
     if (!txs)
       return txs.error();
  
@@ -366,9 +366,9 @@ namespace lws { namespace rpc { namespace feed
     return prep(feed_mempool{from_scanner.outputs.at(0)}, proto);
   } 
 
-  bool start(boost::asio::ip::tcp::socket&& sock, boost::asio::io_context& io, const lws::rpc::client& client, const request& req, const lws::db::storage& disk, const std::chrono::seconds timeout)
+  bool start(boost::asio::ip::tcp::socket&& sock, boost::asio::io_context& io, const lws::rpc::client& client, std::shared_ptr<const mempool> pool, const request& req, const lws::db::storage& disk, const std::chrono::seconds timeout)
   {
-    return do_start(std::move(sock), io, client, req, disk, timeout);
+    return do_start(std::move(sock), io, client, std::move(pool), req, disk, timeout);
   }
 }}} // lws // rpc // feed 
 

src/rpc/fwd.h

@@ -42,6 +42,7 @@ namespace lws
       enum class status : std::uint16_t;
     }
   }
+  class mempool;
   struct rates;
   class scan_manager;  
 }

src/rpc/light_wallet.cpp

@@ -33,12 +33,14 @@
 #include <limits>
 #include <stdexcept>
 #include <type_traits>
+#include <unordered_set>
 
 #include "config.h"
 #include "db/storage.h"
 #include "db/string.h"
 #include "error.h"
 #include "lws_version.h"
+#include "mempool.h"
 #include "time_helper.h"       // monero/contrib/epee/include
 #include "ringct/rctOps.h"     // monero/src
 #include "rpc/feed.h"
@@ -506,19 +508,22 @@ namespace lws
       }
 
       const bool is_coinbase = (extra.first & db::coinbase_output);
+      const auto height = self.value().info.link.height;
+      const bool mempool = height == db::block_id::txpool;
+      const iso_timestamp timestamp{self.value().info.timestamp};
 
       wire::object(dest,
         wire::field("id", std::uint64_t(self.index())),
         wire::field("hash", std::cref(self.value().info.link.tx_hash)),
-        wire::field("timestamp", iso_timestamp(self.value().info.timestamp)),
+        wire::optional_field("timestamp", mempool ? nullptr : &timestamp),
         wire::field("total_received", safe_uint64(self.value().info.spend_meta.amount)),
         wire::field("total_sent", safe_uint64(self.value().spent)),
         wire::field("fee", safe_uint64(self.value().info.fee)),
         wire::field("unlock_time", self.value().info.unlock_time),
-        wire::field("height", self.value().info.link.height),
+        wire::optional_field("height", mempool ? nullptr : &height),
         wire::optional_field("payment_id", payment_id),
         wire::field("coinbase", is_coinbase),
-        wire::field("mempool", false),
+        wire::field("mempool", mempool),
         wire::field("mixin", self.value().info.spend_meta.mixin_count),
         wire::field("recipient", self.value().info.recipient),
         wire::field("spent_outputs", std::cref(self.value().spends))
@@ -583,16 +588,24 @@ namespace lws
     vec_lmdb_<T> vec_lmdb(const std::vector<T>& src) { return {src.begin(), src.end()}; }
 
     template<typename T, typename U>
-    std::pair<std::vector<rpc::get_transaction>, std::uint64_t> merge_into_txes(T output, U spend, const std::size_t reserve, const bool all_outputs, const bool skip_spend_meta)
+    std::pair<std::vector<rpc::get_transaction>, std::uint64_t> merge_into_txes(T output, U spend, db::account const* const user, mempool const* const pool, const std::size_t reserve, const bool all_outputs, const bool skip_spend_meta)
     {
       // merge input and output info into a single set of txes.
 
+      const bool has_pool = (user && pool);
       std::uint64_t total = 0;
       std::vector<rpc::get_transaction> out{};
       std::vector<db::output::spend_meta_> metas{};
+      std::vector<std::pair<db::output_id, db::address_index>> receives{};
+      std::unordered_set<crypto::hash> txes_processed{};
 
       out.reserve(reserve);
       metas.reserve(reserve);
+      if (has_pool)
+      {
+        receives.reserve(reserve);
+        txes_processed.reserve(reserve);
+      }
 
       db::transaction_link next_output{};
       db::transaction_link next_spend{};
@@ -619,6 +632,8 @@ namespace lws
           {
             out.push_back({*output});
             amount = out.back().info.spend_meta.amount;
+            if (has_pool)
+              txes_processed.emplace(next_output.tx_hash);
           }
           else
           {
@@ -626,6 +641,13 @@ namespace lws
             out.back().info.spend_meta.amount += amount;
           }
 
+          if (has_pool)
+          {
+            auto id = output.template get_value<MONERO_FIELD(db::output, spend_meta.id)>();
+            auto subaddr = output.template get_value<MONERO_FIELD(db::output, recipient)>();
+            receives.emplace_back(std::move(id), std::move(subaddr));
+          }
+
           if (all_outputs)
             out.back().receives.push_back(*output);
 
@@ -660,6 +682,8 @@ namespace lws
               out.back().spends.back().possible_spend.mixin_count;
             out.back().info.timestamp = out.back().spends.back().possible_spend.timestamp;
             out.back().info.unlock_time = out.back().spends.back().possible_spend.unlock_time;
+            if (has_pool)
+              txes_processed.emplace(out.back().info.link.tx_hash);
           }
           else
             out.back().spends.push_back({skip_spend_meta ? meta_type{} : *meta, *spend});
@@ -673,6 +697,62 @@ namespace lws
         }
       }
 
+      if (has_pool)
+      {
+        // Add mempool transactions. Order is not important, since
+        // mempool txs cannot depend on each other.
+        lws::account full_user{*user, std::move(receives), {}};
+        const auto pool_txs = pool->scan_account(full_user);
+        for (const auto& row: pool_txs)
+        {
+          // mempool update after block could be delayed
+          if (txes_processed.count(row.hash))
+            continue;
+
+          rpc::get_transaction& tx = out.emplace_back();
+
+          // Ingore spends that use unknown outputs
+          // (perhaps the scanner thread is behind, and hasn't seen them yet).
+          bool from_future = false;
+          for (const auto& spend: row.spends)
+          {
+            const auto meta = rpc::get_address_txs_response::find_metadata(metas, spend.source);
+            if (meta == metas.end() || meta->id != spend.source)
+            {
+              from_future = true;
+              break;
+            }
+            tx.spends.push_back({*meta, spend});
+            tx.spent += meta->amount;
+          }
+
+          uint64_t tx_total = 0;
+          for (const auto& output: row.outputs)
+          {
+            const auto amount = output.spend_meta.amount;
+            total += amount;
+            tx_total += amount;
+            tx.receives.push_back(output);
+          }
+          if (!row.outputs.empty())
+          {
+            tx.info = row.outputs.front();
+            tx.info.spend_meta.amount = tx_total;
+          }
+          else
+          {
+            if (row.spends.empty() || from_future) break;
+            auto spend = row.spends.front();
+            tx.info.link.tx_hash = row.hash;
+            tx.info.link.height = db::block_id::txpool;
+            tx.info.spend_meta.amount = tx_total;
+            tx.info.spend_meta.mixin_count = spend.mixin_count;
+            tx.info.timestamp = spend.timestamp;
+            tx.info.unlock_time = spend.unlock_time;
+          }
+        }
+      }
+      
       return {std::move(out), total};
     }
 
@@ -691,10 +771,10 @@ namespace lws
   {
     std::sort(outputs.begin(), outputs.end(), by_tx_link);
     std::sort(spends.begin(), spends.end(), by_tx_link);
-    return merge_into_txes(vec_lmdb(outputs), vec_lmdb(spends), outputs.size(), true, true).first;
+    return merge_into_txes(vec_lmdb(outputs), vec_lmdb(spends), nullptr, nullptr, outputs.size(), true, true).first;
   }
 
-  expect<rpc::get_address_txs_response> rpc::get_address_txs_response::load(db::storage_reader& reader, const db::account& acct, const bool all_outputs)
+  expect<rpc::get_address_txs_response> rpc::get_address_txs_response::load(db::storage_reader& reader, const db::account& acct, mempool const* const pool, const bool all_outputs)
   {
     auto outputs = reader.get_outputs(acct.id);
     if (!outputs)
@@ -717,7 +797,7 @@ namespace lws
     resp.lookahead_fail = db::to_uint(acct.lookahead_fail);
     resp.lookahead = acct.lookahead;
 
-    auto out = merge_into_txes(outputs->make_iterator(), spends->make_iterator(), outputs->count(), all_outputs, false);
+    auto out = merge_into_txes(outputs->make_iterator(), spends->make_iterator(), std::addressof(acct), pool, outputs->count(), all_outputs, false);
     resp.total_received = safe_uint64(out.second);
     resp.transactions = std::move(out.first);
     return resp;

src/rpc/light_wallet.h

@@ -29,6 +29,7 @@
 
 #include <boost/optional/optional.hpp>
 #include <cstdint>
+
 #include <string>
 #include <utility>
 #include <vector>
@@ -120,7 +121,14 @@ namespace rpc
 
   struct get_transaction
   {
-    get_transaction() = delete;
+    get_transaction()
+      : info{}, receives(), spends(), spent(0)
+    {}
+
+    get_transaction(const db::output& in)
+      : info(in), receives(), spends(), spent(0)
+    {}
+
     db::output info;
     std::vector<db::output> receives;
     std::vector<transaction_spend> spends;
@@ -135,7 +143,7 @@ namespace rpc
       find_metadata(std::vector<db::output::spend_meta_> const& metas, db::output_id id);
 
     static std::vector<get_transaction> load(std::vector<db::output> outputs, std::vector<db::spend> spends);
-    static expect<get_address_txs_response> load(db::storage_reader& reader, const db::account& acct, const bool all_outputs);
+    static expect<get_address_txs_response> load(db::storage_reader& reader, const db::account& acct, const mempool* pool, const bool all_outputs);
 
     safe_uint64 total_received;
     std::uint64_t scanned_height;

src/scanner.cpp

@@ -53,6 +53,7 @@
 #include "db/data.h"
 #include "cryptonote_basic/difficulty.h"              // monero/src
 #include "error.h"
+#include "mempool.h"
 #include "hardforks/hardforks.h"   // monero/src
 #include "misc_log_ex.h"           // monero/contrib/epee/include
 #include "net/net_parse_helpers.h"
@@ -67,6 +68,7 @@
 #include "rpc/scanner/server.h"
 #include "rpc/webhook.h"
 #include "util/blocks.h"
+#include "util/ownership_test.h"
 #include "util/source_location.h"
 #include "util/transactions.h"
 
@@ -194,18 +196,21 @@ namespace lws
  
     struct add_spend
     {
-      void operator()(lws::account& user, const db::spend& spend) const
+      void operator()(lws::account& user, db::spend&& spend) const
       { user.add_spend(spend); }
     };
     struct add_output
     {
-      bool operator()(expect<db::storage_reader>&, lws::account& user, const db::output& out) const
-      { return user.add_out(out); }
+      void operator()(lws::account& user, db::output&& out) const
+      {
+        if (!user.add_out(out))
+          MWARNING("Output not added, duplicate public key encountered");
+      }
     };
 
     struct null_spend
     {
-      void operator()(lws::account&, const db::spend&) const noexcept
+      void operator()(lws::account&, db::spend&&) const noexcept
       {}
     };
     struct send_webhook
@@ -213,404 +218,41 @@ namespace lws
       db::storage const& disk_;
       rpc::client& client_;
       scanner_sync& http_;
-      std::unordered_map<crypto::hash, crypto::hash> txpool_;
 
-      bool operator()(expect<db::storage_reader>& reader, lws::account& user, const db::output& out)
+      void operator()(lws::account& user, db::output&& out)
       {
-        /* Upstream monerod does not send all fields for a transaction, so
-           mempool notifications cannot compute tx_hash correctly (it is not
-           sent separately, a further blunder). Instead, if there are matching
-           outputs with webhooks, fetch mempool to compare tx_prefix_hash and
-           then use corresponding tx_hash. */
         const db::webhook_key key{user.id(), db::webhook_type::tx_confirmation};
         std::vector<db::webhook_value> hooks{};
 
         {
-          db::storage_reader* active_reader = reader ?
-            std::addressof(*reader) : nullptr;
-
-          expect<db::storage_reader> temp_reader{common_error::kInvalidArgument};
-          if (!active_reader)
+          expect<db::storage_reader> reader = disk_.start_read();
+          if (!reader)
           {
-            temp_reader = disk_.start_read();
-            if (!temp_reader)
-            {
-              MERROR("Unable to lookup webhook on tx in pool: " << reader.error().message());
-              return false;
-            }
-            active_reader = std::addressof(*temp_reader);
+            MERROR("Unable to lookup webhook on tx in pool: " << reader.error().message());
+            return;
           }
-          auto found = active_reader->find_webhook(key, out.payment_id.short_);
+          auto found = reader->find_webhook(key, out.payment_id.short_);
           if (!found)
           {
             MERROR("Failed db lookup for webhooks: " << found.error().message());
-            return false;
+            return;
           }
           hooks = std::move(*found);
         }
 
-        if (!hooks.empty() && txpool_.empty())
-        {
-          cryptonote::rpc::GetTransactionPool::Request req{};
-          if (!send(client_, rpc::client::make_message("get_transaction_pool", req)))
-          {
-            MERROR("Unable to compute tx hash for webhook, aborting");
-            return false;
-          }
-          auto resp = client_.get_message(std::chrono::seconds{3});
-          if (!resp)
-          {
-            MERROR("Unable to get txpool: " << resp.error().message());
-            return false;
-          }
-
-          auto txpool = rpc::parse_json_response<rpc::get_transaction_pool>(std::move(*resp));
-          if (!txpool)
-            MONERO_THROW(txpool.error(), "Failed fetching transaction pool");
-          for (auto& tx : txpool->transactions)
-            txpool_.emplace(get_transaction_prefix_hash(tx.tx), tx.tx_hash);
-        }
-
         std::vector<db::webhook_tx_confirmation> events{};
         for (auto& hook : hooks)
         {
           events.push_back(db::webhook_tx_confirmation{key, std::move(hook), out});
           events.back().value.second.confirmations = 0;
-
-          const auto hash = txpool_.find(out.tx_prefix_hash);
-          if (hash != txpool_.end())
-            events.back().tx_info.link.tx_hash = hash->second;
-          else
-            events.pop_back(); //cannot compute tx_hash
         }
         send_payment_hook(http_.io_, client_, http_.webhooks_, epee::to_span(events));
         const expect<void> pushed = client_.push_update(user, mempool_receive{out});
         if (!pushed)
           MERROR("Failed to send mempool update to " << user.address() << ": " << pushed.error().message());
-        return true;
       }
     };
 
-    struct subaddress_reader
-    {
-      expect<db::storage_reader> reader;
-      std::optional<db::storage> disk;
-      db::cursor::subaddress_indexes cur;
-      const std::uint32_t max_subaddresses;
-
-      subaddress_reader(std::optional<db::storage> const& disk_in, const std::uint32_t max_subaddresses)
-        : reader(common_error::kInvalidArgument), disk(), cur(nullptr), max_subaddresses(max_subaddresses)
-      {
-        if (disk_in)
-          disk = disk_in->clone();
-
-        if (max_subaddresses)
-          update_reader();
-      }
-
-      void update_reader()
-      {
-        if (disk)
-          reader = disk->start_read();
-        if (!reader)
-          MERROR("Subadress lookup failure: " << reader.error().message());
-      }
-    };
-
-    void update_lookahead(const account& user, subaddress_reader& reader, const db::address_index& match, db::block_id height)
-    {
-      if (match.is_zero())
-        return; // keep subaddress disabled servers quick
-
-      if (!reader.disk)
-        throw std::runtime_error{"Bad DB handle in scanner"};
-
-      if (height == db::block_id::txpool)
-        height = user.scan_height();
-
-      auto upserted = reader.disk->update_lookahead(user.db_address(), height, match, reader.max_subaddresses);
-      if (upserted)
-      {
-        if (0 < *upserted)
-          reader.update_reader(); // update reader after upsert added new addresses
-        else if (*upserted < 0)
-          upserted = {error::max_subaddresses};
-      }
-
-      if (!upserted)
-        MWARNING("Failed to update lookahead for " << user.address() << ": " << upserted.error());
-    }
-
-    void scan_transaction_base(
-      epee::span<lws::account> users,
-      const db::block_id height,
-      const std::uint64_t timestamp,
-      crypto::hash const* tx_hash,
-      cryptonote::transaction const& tx,
-      std::vector<std::uint64_t> const& out_ids,
-      subaddress_reader& reader,
-      std::function<void(lws::account&, const db::spend&)> spend_action,
-      std::function<bool(expect<db::storage_reader>&, lws::account&, const db::output&)> output_action)
-    {
-      if (2 < tx.version)
-        throw std::runtime_error{"Unsupported tx version"};
-
-      crypto::hash tx_hash_lazy;
-      cryptonote::tx_extra_pub_key key;
-      boost::optional<crypto::hash> prefix_hash;
-      boost::optional<cryptonote::tx_extra_nonce> extra_nonce;
-      std::pair<std::uint8_t, db::output::payment_id_> payment_id;
-      cryptonote::tx_extra_additional_pub_keys additional_tx_pub_keys;
-      std::vector<crypto::key_derivation> additional_derivations;
-
-      const auto get_tx_hash = [&] () -> const crypto::hash&
-      {
-        if (!tx_hash)
-        {
-          tx_hash_lazy = get_transaction_hash(tx);
-          tx_hash = std::addressof(tx_hash_lazy);
-        } 
-        return *tx_hash;
-      };
-
-      {
-        std::vector<cryptonote::tx_extra_field> extra;
-        cryptonote::parse_tx_extra(tx.extra, extra);
-        // allow partial parsing of tx extra (similar to wallet2.cpp)
-
-        if (!cryptonote::find_tx_extra_field_by_type(extra, key))
-          return;
-
-        extra_nonce.emplace();
-        if (cryptonote::find_tx_extra_field_by_type(extra, *extra_nonce))
-        {
-          if (cryptonote::get_payment_id_from_tx_extra_nonce(extra_nonce->nonce, payment_id.second.long_))
-            payment_id.first = sizeof(crypto::hash);
-        }
-        else
-          extra_nonce = boost::none;
-
-        // additional tx pub keys present when there are 3+ outputs in a tx involving subaddresses
-        if (reader.reader)
-          cryptonote::find_tx_extra_field_by_type(extra, additional_tx_pub_keys);
-      } // destruct `extra` vector
-
-      for (account& user : users)
-      {
-        if (height <= user.scan_height())
-          continue; // to next user
-
-        crypto::key_derivation derived;
-        if (!crypto::wallet::generate_key_derivation(key.pub_key, user.view_key(), derived))
-          continue; // to next user
-
-        if (reader.reader && additional_tx_pub_keys.data.size() == tx.vout.size())
-        {
-          additional_derivations.resize(tx.vout.size());
-          std::size_t index = -1;
-          for (auto const& out: tx.vout)
-          {
-            ++index;
-            if (!crypto::wallet::generate_key_derivation(additional_tx_pub_keys.data[index], user.view_key(), additional_derivations[index]))
-            {
-              additional_derivations.clear();
-              break; // vout loop
-            }
-          }
-        }
-
-        db::extra ext{};
-        std::uint32_t mixin = 0;
-        for (auto const& in : tx.vin)
-        {
-          cryptonote::txin_to_key const* const in_data =
-            boost::get<cryptonote::txin_to_key>(std::addressof(in));
-          if (in_data)
-          {
-            mixin = boost::numeric_cast<std::uint32_t>(
-              std::max(std::size_t(1), in_data->key_offsets.size()) - 1
-            );
-
-            std::uint64_t goffset = 0;
-            for (std::uint64_t offset : in_data->key_offsets)
-            {
-              goffset += offset;
-              const boost::optional<db::address_index> subaccount =
-                user.get_spendable(db::output_id{in_data->amount, goffset});
-              if (!subaccount)
-                continue; // to next input
-
-              spend_action(
-                user,
-                db::spend{
-                  db::transaction_link{height, get_tx_hash()},
-                  in_data->k_image,
-                  db::output_id{in_data->amount, goffset},
-                  timestamp,
-                  tx.unlock_time,
-                  mixin,
-                  {0, 0, 0}, // reserved
-                  payment_id.first,
-                  payment_id.second.long_,
-                  *subaccount
-                }
-              );
-            }
-          }
-          else if (boost::get<cryptonote::txin_gen>(std::addressof(in)))
-            ext = db::extra(ext | db::coinbase_output);
-        }
-
-        std::size_t index = -1;
-        for (auto const& out : tx.vout)
-        {
-          ++index;
-
-          crypto::public_key out_pub_key;
-          if (!cryptonote::get_output_public_key(out, out_pub_key))
-            continue; // to next output
-
-          boost::optional<crypto::view_tag> view_tag_opt =
-            cryptonote::get_output_view_tag(out);
-
-          const bool found_tag =
-            (!additional_derivations.empty() && cryptonote::out_can_be_to_acc(view_tag_opt, additional_derivations.at(index), index)) ||
-            cryptonote::out_can_be_to_acc(view_tag_opt, derived, index); 
-
-          if (!found_tag)
-            continue; // to next output
-
-          bool found_pub = false;
-          db::address_index account_index{db::major_index::primary, db::minor_index::primary};
-          crypto::key_derivation active_derived{};
-          crypto::public_key active_pub{};
-
-          // inspect the additional and traditional keys
-          for (std::size_t attempt = 0; attempt < 2; ++attempt)
-          {
-            if (attempt == 0)
-            {
-              active_derived = derived;
-              active_pub = key.pub_key;
-            }
-            else if (!additional_derivations.empty())
-            {
-              active_derived = additional_derivations.at(index);
-              active_pub = additional_tx_pub_keys.data.at(index);
-            }
-            else
-              break; // inspection loop
-
-            crypto::public_key derived_pub;
-            if (!crypto::wallet::derive_subaddress_public_key(out_pub_key, active_derived, index, derived_pub))
-              continue; // to next available active_derived
-
-            if (user.spend_public() != derived_pub)
-            {
-              if (!reader.reader)
-                continue; // to next available active_derived
-
-              const expect<db::address_index> match =
-                reader.reader->find_subaddress(user.id(), derived_pub, reader.cur);
-              if (!match)
-              {
-                if (match != lmdb::error(MDB_NOTFOUND))
-                  MERROR("Failure when doing subaddress search: " << match.error().message());
-                continue; // to next available active_derived
-              }
-
-              update_lookahead(user, reader, *match, height);
-              found_pub = true;
-              account_index = *match;
-              break; // additional_derivations loop
-            }
-            else
-            {
-              found_pub = true;
-              break; // additional_derivations loop
-            }
-          }
-
-          if (!found_pub)
-            continue; // to next output
-
-          if (!prefix_hash)
-          {
-            prefix_hash.emplace();
-            cryptonote::get_transaction_prefix_hash(tx, *prefix_hash);
-          }
-
-          std::uint64_t amount = out.amount;
-          rct::key mask = rct::identity();
-          if (!amount && !(ext & db::coinbase_output) && 1 < tx.version)
-          {
-            const bool bulletproof2 = (rct::RCTTypeBulletproof2 <= tx.rct_signatures.type);
-            const auto decrypted = lws::decode_amount(
-              tx.rct_signatures.outPk.at(index).mask, tx.rct_signatures.ecdhInfo.at(index), active_derived, index, bulletproof2
-            );
-            if (!decrypted)
-            {
-              MWARNING(user.address() << " failed to decrypt amount for tx " << get_tx_hash() << ", skipping output");
-              continue; // to next output
-            }
-            amount = decrypted->first;
-            mask = decrypted->second;
-            ext = db::extra(ext | db::ringct_output);
-          }
-          else if (1 < tx.version)
-            ext = db::extra(ext | db::ringct_output);
-
-          if (extra_nonce)
-          {
-            if (!payment_id.first && cryptonote::get_encrypted_payment_id_from_tx_extra_nonce(extra_nonce->nonce, payment_id.second.short_))
-            {
-              payment_id.first = sizeof(crypto::hash8);
-              lws::decrypt_payment_id(payment_id.second.short_, active_derived);
-            }
-          }
-          const bool added = output_action(
-            reader.reader,
-            user,
-            db::output{
-              db::transaction_link{height, get_tx_hash()},
-              db::output::spend_meta_{
-                db::output_id{tx.version < 2 ? out.amount : 0, out_ids.at(index)},
-                amount,
-                mixin,
-                boost::numeric_cast<std::uint32_t>(index),
-                active_pub
-              },
-              timestamp,
-              tx.unlock_time,
-              *prefix_hash,
-              out_pub_key,
-              mask,
-              {0, 0, 0, 0, 0, 0, 0}, // reserved bytes
-              db::pack(ext, payment_id.first),
-              payment_id.second,
-              cryptonote::get_tx_fee(tx),
-              account_index
-            }
-          );
-
-          if (!added)
-            MWARNING("Output not added, duplicate public key encountered");
-        } // for all tx outs
-      } // for all users
-    }
-
-    void scan_transaction(
-      epee::span<lws::account> users,
-      const db::block_id height,
-      const std::uint64_t timestamp,
-      crypto::hash const& tx_hash,
-      cryptonote::transaction const& tx,
-      std::vector<std::uint64_t> const& out_ids,
-      subaddress_reader& reader)
-    {
-      scan_transaction_base(users, height, timestamp, std::addressof(tx_hash), tx, out_ids, reader, add_spend{}, add_output{});
-    }
-
     void scan_transactions(std::string&& txpool_msg, epee::span<lws::account> users, db::storage const& disk, scanner_sync& self, rpc::client& client, const scanner_options& opts)
     {
       // uint64::max is for txpool
@@ -628,10 +270,11 @@ namespace lws
       const auto time =
         boost::numeric_cast<std::uint64_t>(std::chrono::system_clock::to_time_t(std::chrono::system_clock::now()));
 
-      subaddress_reader reader{std::optional<db::storage>{disk.clone()}, opts.max_subaddresses};
-      send_webhook sender{disk, client, self};
+      ownership_test scan_transaction{null_spend{}, send_webhook{disk, client, self}};
+      if (opts.max_subaddresses > 0)
+        scan_transaction.enable_subaddresses(disk, opts.max_subaddresses);
       for (const auto& tx : parsed->txes)
-        scan_transaction_base(users, db::block_id::txpool, time, nullptr, tx, fake_outs, reader, null_spend{}, sender);
+        scan_transaction(users, db::block_id::txpool, time, nullptr, tx, fake_outs);
     }
 
     void do_scan_loop(scanner_sync& self, std::shared_ptr<thread_data> data, const size_t thread_n) noexcept
@@ -889,7 +532,9 @@ namespace lws
             );
           }
 
-          subaddress_reader reader{disk, opts.max_subaddresses};
+          ownership_test scan_transaction{add_spend{}, add_output{}};
+          if (disk && opts.max_subaddresses)
+            scan_transaction.enable_subaddresses(*disk, opts.max_subaddresses);
           db::block_difficulty::unsigned_int diff{};
           const db::block_id initial_height = db::block_id(fetched->start_height);
           for (auto block_data : boost::combine(blocks, indices))
@@ -906,18 +551,13 @@ namespace lws
             if (indices.empty())
               throw std::runtime_error{"Bad daemon response - missing /coinbase tx indices"};
 
-            crypto::hash miner_tx_hash;
-            if (!cryptonote::get_transaction_hash(block.miner_tx, miner_tx_hash))
-              throw std::runtime_error{"Failed to calculate miner tx hash"};
-
             scan_transaction(
               epee::to_mut_span(users),
               db::block_id(fetched->start_height),
               block.timestamp,
-              miner_tx_hash,
+              nullptr,
               block.miner_tx,
-              *(indices.begin()),
-              reader
+              *(indices.begin())
             );
 
             if (opts.untrusted_daemon)
@@ -966,10 +606,9 @@ namespace lws
                 epee::to_mut_span(users),
                 db::block_id(fetched->start_height),
                 block.timestamp,
-                boost::get<0>(tx_data),
-                boost::get<1>(tx_data),
-                boost::get<2>(tx_data),
-                reader
+                std::addressof(boost::get<0>(tx_data)), // tx_hashes
+                boost::get<1>(tx_data), // txes
+                boost::get<2>(tx_data) // indices
               );
             }
 
@@ -988,10 +627,9 @@ namespace lws
             blockchain.push_back(cryptonote::get_block_hash(block));
           } // for each block
 
-          reader.reader = std::error_code{common_error::kInvalidArgument}; // cleanup reader before next write
-
           MINFO("Thread " << thread_n << " processed " << blockchain.size() << " blocks(s) @ height " << fetched->start_height << " against " << users.size() << " account(s)");
 
+          scan_transaction.disable_subaddresses(); // cleanup reader before next write
           if (!store(self.io_, client, self.webhooks_, epee::to_span(blockchain), epee::to_mut_span(users), epee::to_span(new_pow)))
             return false;
 
@@ -1015,7 +653,17 @@ namespace lws
       Launches `thread_count` threads to run `scan_loop`, and then polls for
       active account changes in background
     */
-    void check_loop(scanner_sync& self, db::storage disk, rpc::context& ctx, const std::size_t thread_count, const std::string& lws_server_addr, std::string lws_server_pass, std::vector<lws::account> users, std::vector<db::account_id> active, const scanner_options& opts)
+    void check_loop(
+      scanner_sync& self,
+      db::storage disk,
+      rpc::context& ctx,
+      std::shared_ptr<lws::mempool> pool,
+      const std::size_t thread_count,
+      const std::string& lws_server_addr,
+      std::string lws_server_pass,
+      std::vector<lws::account> users,
+      std::vector<db::account_id> active,
+      const scanner_options& opts)
     {
       assert(users.size() == active.size());
       assert(thread_count || !lws_server_addr.empty());
@@ -1298,6 +946,35 @@ namespace lws
           }
         }
 
+        if (pool) {
+          auto client = std::make_shared<rpc::client>(MONERO_UNWRAP(ctx.connect()));
+          MONERO_UNWRAP(client->watch_scan_signals());
+          threads.emplace_back(attrs, [pool, client, &self] ()
+          {
+            MINFO("Pool updater thread started");
+            while (self.is_running())
+            {
+              try
+              {
+                auto result = pool_update_loop(pool, *client);
+                if (!result
+                  && result.error() != make_error_code(lws::error::signal_abort_process)
+                  && result.error() != make_error_code(lws::error::signal_abort_scan))
+                  MERROR("Pool update loop failed " << result.error());
+              }
+              catch (std::exception const& e)
+              {
+                MERROR("Pool update threw " << e.what());
+              }
+              catch (...)
+              {
+                MERROR("Pool update threw unknown exception");
+              }
+            }
+            MINFO("Pool updater thread stopped");
+          });
+        }
+
         users.clear();
         users.shrink_to_fit();
 
@@ -1595,7 +1272,13 @@ namespace lws
     return sync_quick(sync_, disk_.clone(), std::move(client), regtest);
   }
 
-  void scanner::run(rpc::context ctx, std::size_t thread_count, const std::string& lws_server_addr, std::string lws_server_pass, const scanner_options& opts)
+  void scanner::run(
+    rpc::context ctx,
+    std::shared_ptr<mempool> pool,
+    std::size_t thread_count,
+    const std::string& lws_server_addr,
+    std::string lws_server_pass,
+    const scanner_options& opts)
   {
     if (has_shutdown())
       MONERO_THROW(common_error::kInvalidArgument, "this has shutdown");
@@ -1683,7 +1366,7 @@ namespace lws
         }
       }
       else
-        check_loop(sync_, disk_.clone(), ctx, thread_count, lws_server_addr, lws_server_pass, std::move(users), std::move(active), opts);
+        check_loop(sync_, disk_.clone(), ctx, pool, thread_count, lws_server_addr, lws_server_pass, std::move(users), std::move(active), opts);
 
       if (has_shutdown())
         return;

src/scanner.h

@@ -35,6 +35,7 @@
 
 #include "db/fwd.h"
 #include "db/storage.h"
+#include "fwd.h"
 #include "net/http/client.h"
 #include "net/net_ssl.h" // monero/contrib/epee/include
 #include "rpc/client.h"
@@ -130,7 +131,13 @@ namespace lws
     expect<rpc::client> sync(rpc::client client, const bool untrusted_daemon = false, const bool regtest = false);
 
     //! Poll daemon until `shutdown()` is called, using `thread_count` threads.
-    void run(rpc::context ctx, std::size_t thread_count, const std::string& server_addr, std::string server_pass, const scanner_options&);
+    void run(
+      rpc::context ctx,
+      std::shared_ptr<mempool> pool,
+      std::size_t thread_count,
+      const std::string& server_addr,
+      std::string server_pass,
+      const scanner_options&);
 
     //! \return True iff `stop()` and `shutdown()` has never been called
     bool is_running() const noexcept { return sync_.is_running(); }

src/server_main.cpp

@@ -45,6 +45,7 @@
 #include "db/storage.h"
 #include "error.h"
 #include "lws_version.h"
+#include "mempool.h"
 #include "options.h"
 #include "rest_server.h"
 #include "scanner.h"
@@ -373,14 +374,23 @@ namespace
     //! SIGINT handle registered by `scanner` constructor
     lws::scanner scanner{disk.clone(), prog.rest_config.webhook_verify};
 
+    std::shared_ptr<lws::mempool> mempool;
     MINFO("Using monerod ZMQ RPC at " << ctx.daemon_address());
     if (!sub_address.empty())
+    {
       MINFO("Using monerod ZMQ sub at " << sub_address);
+      mempool = std::make_shared<lws::mempool>();
+    }
 
     auto client = scanner.sync(ctx.connect().value(), prog.untrusted_daemon).value();
 
     lws::rest_server server{
-      epee::to_span(prog.rest_servers), prog.admin_rest_servers, std::move(disk), std::move(client), std::move(prog.rest_config)
+      epee::to_span(prog.rest_servers),
+      prog.admin_rest_servers,
+      std::move(disk),
+      std::move(client),
+      mempool,
+      std::move(prog.rest_config)
     };
     for (const std::string& address : prog.rest_servers)
       MINFO("Listening for REST clients at " << address);
@@ -390,6 +400,7 @@ namespace
     // blocks until SIGINT
     scanner.run(
       std::move(ctx),
+      mempool,
       prog.scan_threads,
       std::move(prog.lws_server_addr),
       std::move(prog.lws_server_pass),

src/util/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(monero-lws-util_sources blocks.cpp gamma_picker.cpp random_outputs.cpp source_location.cpp transactions.cpp)
-set(monero-lws-util_headers blocks.h fwd.h gamma_picker.h random_outputs.h source_location.h transactions.h)
+set(monero-lws-util_sources blocks.cpp gamma_picker.cpp ownership_test.cpp random_outputs.cpp source_location.cpp transactions.cpp)
+set(monero-lws-util_headers blocks.h fwd.h gamma_picker.h ownership_test.h random_outputs.h source_location.h transactions.h)
 
 add_library(monero-lws-util ${monero-lws-util_sources} ${monero-lws-util_headers})
 target_link_libraries(monero-lws-util monero::libraries monero-lws-db)

src/util/ownership_test.cpp

@@ -0,0 +1,338 @@
+// Copyright (c) 2018-2025, 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 "ownership_test.h"
+
+#include <boost/optional/optional.hpp>
+#include <boost/range/combine.hpp>
+
+#include "common/error.h"
+#include "crypto/wallet/crypto.h"
+#include "cryptonote_basic/cryptonote_format_utils.h"
+#include "cryptonote_basic/tx_extra.h"
+#include "db/account.h"
+#include "db/data.h"
+#include "db/storage.h"
+#include "error.h"
+#include "misc_log_ex.h"
+#include "rpc/daemon_messages.h"
+#include "util/transactions.h"
+
+namespace lws
+{
+  ownership_test::ownership_test(spend_action on_spend, output_action on_output)
+    : on_spend(std::move(on_spend))
+    , on_output(std::move(on_output))
+  {}
+
+  void ownership_test::enable_subaddresses(const db::storage& disk, const std::uint32_t max_subaddresses)
+  {
+    subaddress.emplace(disk, max_subaddresses);
+  }
+
+  void ownership_test::disable_subaddresses()
+  {
+    subaddress.reset();
+  }
+
+  void ownership_test::operator()(
+    epee::span<account> users,
+    db::block_id height,
+    std::uint64_t timestamp,
+    crypto::hash const* tx_hash,
+    const cryptonote::transaction& tx,
+    const std::vector<std::uint64_t>& out_ids)
+  {
+    if (2 < tx.version)
+      throw std::runtime_error{"Unsupported tx version"};
+
+    crypto::hash tx_hash_lazy;
+    cryptonote::tx_extra_pub_key key;
+    boost::optional<crypto::hash> prefix_hash;
+    boost::optional<cryptonote::tx_extra_nonce> extra_nonce;
+    std::pair<std::uint8_t, db::output::payment_id_> payment_id;
+    cryptonote::tx_extra_additional_pub_keys additional_tx_pub_keys;
+    std::vector<crypto::key_derivation> additional_derivations;
+
+    const auto get_tx_hash = [&] () -> const crypto::hash&
+    {
+      if (!tx_hash)
+      {
+        tx_hash_lazy = get_transaction_hash(tx);
+        tx_hash = std::addressof(tx_hash_lazy);
+      } 
+      return *tx_hash;
+    };
+
+    {
+      std::vector<cryptonote::tx_extra_field> extra;
+      cryptonote::parse_tx_extra(tx.extra, extra);
+      if (!cryptonote::find_tx_extra_field_by_type(extra, key))
+        return;
+
+      extra_nonce.emplace();
+      if (cryptonote::find_tx_extra_field_by_type(extra, *extra_nonce))
+      {
+        if (cryptonote::get_payment_id_from_tx_extra_nonce(extra_nonce->nonce, payment_id.second.long_))
+          payment_id.first = sizeof(crypto::hash);
+      }
+      else
+        extra_nonce = boost::none;
+
+      if (subaddress)
+        cryptonote::find_tx_extra_field_by_type(extra, additional_tx_pub_keys);
+    }
+
+    for (account& user : users)
+    {
+      if (height <= user.scan_height())
+        continue;
+
+      crypto::key_derivation derived;
+      if (!crypto::wallet::generate_key_derivation(key.pub_key, user.view_key(), derived))
+        continue;
+
+      if (subaddress && additional_tx_pub_keys.data.size() == tx.vout.size())
+      {
+        additional_derivations.resize(tx.vout.size());
+        for (std::size_t index = 0; index < tx.vout.size(); ++index)
+        {
+          if (!crypto::wallet::generate_key_derivation(additional_tx_pub_keys.data[index], user.view_key(), additional_derivations[index]))
+          {
+            additional_derivations.clear();
+            break;
+          }
+        }
+      }
+
+      db::extra ext{};
+      std::uint32_t mixin = 0;
+      for (auto const& in : tx.vin)
+      {
+        if (const auto in_data = boost::get<cryptonote::txin_to_key>(std::addressof(in)))
+        {
+          mixin = boost::numeric_cast<std::uint32_t>(std::max<std::size_t>(1, in_data->key_offsets.size()) - 1);
+
+          std::uint64_t goffset = 0;
+          for (std::uint64_t offset : in_data->key_offsets)
+          {
+            goffset += offset;
+            const auto address_index = user.get_spendable(db::output_id{in_data->amount, goffset});
+            if (!address_index)
+              continue;
+
+            on_spend(
+              user,
+              db::spend{
+                db::transaction_link{height, get_tx_hash()},
+                in_data->k_image,
+                db::output_id{in_data->amount, goffset},
+                timestamp,
+                tx.unlock_time,
+                mixin,
+                {0, 0, 0},
+                payment_id.first,
+                payment_id.second.long_,
+                *address_index
+              }
+            );
+          }
+        }
+        else if (boost::get<cryptonote::txin_gen>(std::addressof(in)))
+          ext = db::extra(ext | db::coinbase_output);
+      }
+
+      for (std::size_t index = 0; index < tx.vout.size(); ++index)
+      {
+        crypto::public_key out_pub_key;
+        if (!cryptonote::get_output_public_key(tx.vout[index], out_pub_key))
+          continue;
+
+        const auto view_tag = cryptonote::get_output_view_tag(tx.vout[index]);
+        const bool matched =
+          (!additional_derivations.empty() && cryptonote::out_can_be_to_acc(view_tag, additional_derivations.at(index), index)) ||
+          cryptonote::out_can_be_to_acc(view_tag, derived, index);
+
+        if (!matched)
+          continue;
+
+        bool found_pub = false;
+        db::address_index account_index{db::major_index::primary, db::minor_index::primary};
+        crypto::key_derivation active_derivation{};
+        crypto::public_key active_pub{};
+
+        for (std::size_t attempt = 0; attempt < 2; ++attempt)
+        {
+          if (attempt == 0)
+          {
+            active_derivation = derived;
+            active_pub = key.pub_key;
+          }
+          else if (!additional_derivations.empty())
+          {
+            active_derivation = additional_derivations.at(index);
+            active_pub = additional_tx_pub_keys.data.at(index);
+          }
+          else
+            break;
+
+          crypto::public_key derived_pub;
+          if (!crypto::wallet::derive_subaddress_public_key(out_pub_key, active_derivation, index, derived_pub))
+            continue;
+
+          if (user.spend_public() != derived_pub)
+          {
+            if (!subaddress)
+              continue;
+
+            const expect<db::address_index> match = subaddress->find_subaddress(user, derived_pub);
+            if (!match)
+            {
+              if (match != lmdb::error(MDB_NOTFOUND))
+                MERROR("Failure when doing subaddress search: " << match.error().message());
+              continue;
+            }
+
+            auto result = subaddress->update_lookahead(user, *match, height);
+            if (!result)
+              MWARNING("Failed to update lookahead for " << user.address() << ": " << result.error());
+            found_pub = true;
+            account_index = *match;
+            break;
+          }
+          else
+          {
+            found_pub = true;
+            break;
+          }
+        }
+
+        if (!found_pub)
+          continue;
+
+        if (!prefix_hash)
+        {
+          prefix_hash.emplace();
+          cryptonote::get_transaction_prefix_hash(tx, *prefix_hash);
+        }
+
+        std::uint64_t amount = tx.vout[index].amount;
+        rct::key mask = rct::identity();
+        if (!amount && !(ext & db::coinbase_output) && 1 < tx.version)
+        {
+          const bool bulletproof2 = (rct::RCTTypeBulletproof2 <= tx.rct_signatures.type);
+          const auto decrypted = lws::decode_amount(
+            tx.rct_signatures.outPk.at(index).mask,
+            tx.rct_signatures.ecdhInfo.at(index),
+            active_derivation,
+            index,
+            bulletproof2
+          );
+          if (!decrypted)
+          {
+            MWARNING(user.address() << " failed to decrypt amount for tx " << get_tx_hash() << ", skipping output");
+            continue;
+          }
+          amount = decrypted->first;
+          mask = decrypted->second;
+          ext = db::extra(ext | db::ringct_output);
+        }
+        else if (1 < tx.version)
+          ext = db::extra(ext | db::ringct_output);
+
+        if (extra_nonce && !payment_id.first && cryptonote::get_encrypted_payment_id_from_tx_extra_nonce(extra_nonce->nonce, payment_id.second.short_))
+        {
+          payment_id.first = sizeof(crypto::hash8);
+          lws::decrypt_payment_id(payment_id.second.short_, active_derivation);
+        }
+
+        on_output(
+          user,
+          db::output{
+            db::transaction_link{height, get_tx_hash()},
+            db::output::spend_meta_{
+              db::output_id{tx.version < 2 ? tx.vout[index].amount : 0, out_ids.at(index)},
+              amount,
+              mixin,
+              boost::numeric_cast<std::uint32_t>(index),
+              active_pub
+            },
+            timestamp,
+            tx.unlock_time,
+            *prefix_hash,
+            out_pub_key,
+            mask,
+            {0, 0, 0, 0, 0, 0, 0},
+            db::pack(ext, payment_id.first),
+            payment_id.second,
+            cryptonote::get_tx_fee(tx),
+            account_index
+          }
+        );
+      }
+    }
+  }
+
+  subaddress_reader::subaddress_reader(db::storage const& disk_in, const std::uint32_t max_subaddresses)
+    : reader(common_error::kInvalidArgument), disk(disk_in.clone()), cur(nullptr), max_subaddresses(max_subaddresses)
+  {
+    update_reader();
+  }
+
+  void subaddress_reader::update_reader()
+  {
+    reader = disk.start_read();
+    if (!reader)
+      MERROR("Subadress lookup failure: " << reader.error().message());
+  }
+  
+  expect<db::address_index> subaddress_reader::find_subaddress(const account& user, crypto::public_key const& pubkey)
+  {
+    if (!reader)
+      return {lmdb::error(MDB_NOTFOUND)};
+    return reader->find_subaddress(user.id(), pubkey, cur);
+  }
+
+  expect<void> subaddress_reader::update_lookahead(const account& user, const db::address_index& match, db::block_id height)
+  {
+    if (match.is_zero())
+      return {}; // keep subaddress disabled servers quick
+
+    if (height == db::block_id::txpool)
+      height = user.scan_height();
+
+    auto upserted = disk.update_lookahead(user.db_address(), height, match, max_subaddresses);
+    if (!upserted)
+      return upserted.error();
+    if (0 < *upserted)
+      update_reader(); // update reader after upsert added new addresses
+    else if (*upserted < 0)
+      upserted = {error::max_subaddresses};
+    return upserted.error();
+  }
+
+} // namespace lws

src/util/ownership_test.h

@@ -0,0 +1,95 @@
+// Copyright (c) 2018-2025, 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/optional/optional.hpp>
+#include <functional>
+
+#include "cryptonote_basic/cryptonote_basic.h"
+#include "db/account.h"
+#include "db/storage.h"
+#include "span.h"
+
+namespace lws
+{
+  class subaddress_reader
+  {
+  public:
+    subaddress_reader(db::storage const& disk_in, const std::uint32_t max_subaddresses);
+
+    expect<db::address_index> find_subaddress(const account& user, crypto::public_key const& pubkey);
+    expect<void> update_lookahead(const account& user, const db::address_index& match, db::block_id height);
+
+  private:
+    expect<db::storage_reader> reader;
+    db::storage disk;
+    db::cursor::subaddress_indexes cur;
+    const std::uint32_t max_subaddresses;
+
+    void update_reader();
+  };
+
+  class ownership_test {
+  public:
+    using spend_action = std::function<void(account&, db::spend&&)>;
+    using output_action = std::function<void(account&, db::output&&)>;
+
+    ownership_test(spend_action, output_action);
+
+    ownership_test(const ownership_test&) = delete;
+    ownership_test(ownership_test&&) = default;
+
+    ownership_test& operator=(const ownership_test&) = delete;
+    ownership_test& operator=(ownership_test&&) = default;
+
+    /*! Tests the transaction against our accounts,
+      and invokes callbacks for matching inputs or outputs.
+      @param height mined height
+      @param timestamp mined block timestamp
+      @param out_ids maps vout indices to global utxo indexes
+    */
+    void operator()(
+      epee::span<account> users,
+      db::block_id height,
+      std::uint64_t timestamp,
+      crypto::hash const* tx_hash,
+      const cryptonote::transaction& tx,
+      const std::vector<std::uint64_t>& out_ids
+    );
+
+    void enable_subaddresses(const db::storage& disk, const std::uint32_t max_subaddresses);
+    void disable_subaddresses();
+
+  private:
+    spend_action on_spend;
+    output_action on_output;
+
+    boost::optional<subaddress_reader> subaddress;
+  };
+
+} // namespace lws

tests/unit/CMakeLists.txt

@@ -33,9 +33,10 @@ target_link_libraries(monero-lws-unit-framework)
 add_subdirectory(db)
 add_subdirectory(net)
 add_subdirectory(rpc)
+add_subdirectory(util)
 add_subdirectory(wire)
 
-add_executable(monero-lws-unit main.cpp rest.test.cpp scanner.test.cpp)
+add_executable(monero-lws-unit main.cpp mempool.test.cpp rest.test.cpp scanner.test.cpp)
 target_link_libraries(monero-lws-unit
   monero::libraries
   monero-lws-daemon-common
@@ -44,6 +45,7 @@ target_link_libraries(monero-lws-unit
   monero-lws-unit-net
   monero-lws-unit-net-http
   monero-lws-unit-rpc
+  monero-lws-unit-util
   monero-lws-unit-wire
   monero-lws-unit-wire-json
   monero-lws-unit-wire-msgpack

tests/unit/mempool.test.cpp

@@ -0,0 +1,137 @@
+// Copyright (c) 2024, 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 "framework.test.h"
+
+#include <cstring>
+#include "cryptonote_basic/cryptonote_format_utils.h" // monero/src
+#include "cryptonote_core/cryptonote_tx_utils.h"      // monero/src
+#include "db/account.h"
+#include "db/data.h"
+#include "mempool.h"
+#include "util/account.test.h"
+#include "util/transaction.test.h"
+
+namespace
+{
+  
+  lws::db::account make_account(const cryptonote::account_keys& src)
+  {
+    lws::db::account out{};
+    std::memcpy(&out.address.spend_public, &src.m_account_address.m_spend_public_key, sizeof(out.address.spend_public));
+    std::memcpy(&out.address.view_public, &src.m_account_address.m_view_public_key, sizeof(out.address.view_public));
+    std::memcpy(&out.key, &unwrap(unwrap(src.m_view_secret_key)), sizeof(out.key));
+    return out;
+  }
+}
+
+LWS_CASE("mempool")
+{
+  const auto base = lws_test::make_account();
+
+  SETUP("mempool")
+  {
+    lws::mempool pool{};
+    lws::account account{make_account(base), {}, {}};
+
+    SECTION("empty")
+    {
+      EXPECT(pool.scan_account(account).empty());
+    }
+
+    SECTION("add_txs")
+    {
+      const auto other = lws_test::make_account();
+
+      std::vector<cryptonote::tx_destination_entry> destinations;
+
+      destinations.emplace_back();
+      destinations.back().amount = 8000;
+      destinations.back().addr = base.m_account_address;
+      const auto tx1 = lws_test::make_tx(lest_env, base, destinations, 20, true);
+
+      destinations.back().amount = 9000;
+      destinations.back().addr = other.m_account_address;
+      const auto tx2 = lws_test::make_tx(lest_env, other, destinations, 20, true);
+
+      std::array<cryptonote::transaction, 2> txes{{tx1.tx, tx2.tx}};
+      pool.add_txs(epee::to_mut_span(txes));
+
+      auto results = pool.scan_account(account);
+      EXPECT(results.size() == 1);
+      EXPECT(results.at(0).hash == get_transaction_hash(tx1.tx));
+      EXPECT(results.at(0).spends.size() == 0);
+      EXPECT(results.at(0).outputs.size() == 1);
+      EXPECT(results.at(0).outputs.at(0).link.height == lws::db::block_id::txpool);
+      EXPECT(results.at(0).outputs.at(0).link.tx_hash == get_transaction_hash(tx1.tx));
+
+      results = pool.scan_account(account);
+      EXPECT(results.size() == 1);
+      EXPECT(results.at(0).hash == get_transaction_hash(tx1.tx));
+      EXPECT(results.at(0).spends.size() == 0);
+      EXPECT(results.at(0).outputs.size() == 1);
+      EXPECT(results.at(0).outputs.at(0).link.height == lws::db::block_id::txpool);
+      EXPECT(results.at(0).outputs.at(0).link.tx_hash == get_transaction_hash(tx1.tx));
+
+      pool.reset_txs(lws::mempool::pool_table{});
+      results = pool.scan_account(account);
+      EXPECT(results.empty());
+    }
+
+    SECTION("rest_txs")
+    {
+      const auto now = std::chrono::system_clock::now();
+      const auto other = lws_test::make_account();
+
+      std::vector<cryptonote::tx_destination_entry> destinations;
+
+      destinations.emplace_back();
+      destinations.back().amount = 8000;
+      destinations.back().addr = base.m_account_address;
+      const auto tx1 = lws_test::make_tx(lest_env, base, destinations, 20, true);
+
+      destinations.back().amount = 9000;
+      destinations.back().addr = other.m_account_address;
+      const auto tx2 = lws_test::make_tx(lest_env, other, destinations, 20, true);
+
+      pool.reset_txs(
+        lws::mempool::pool_table{
+          {get_transaction_hash(tx1.tx), std::make_shared<lws::mempool::pool_entry>(cryptonote::transaction{tx1.tx}, now)},
+          {get_transaction_hash(tx2.tx), std::make_shared<lws::mempool::pool_entry>(cryptonote::transaction{tx2.tx}, now)}
+        }
+      );
+
+      auto results = pool.scan_account(account);
+      EXPECT(results.size() == 1);
+      EXPECT(results.at(0).hash == get_transaction_hash(tx1.tx));
+      EXPECT(results.at(0).spends.size() == 0);
+      EXPECT(results.at(0).outputs.size() == 1);
+      EXPECT(results.at(0).outputs.at(0).link.height == lws::db::block_id::txpool);
+      EXPECT(results.at(0).outputs.at(0).link.tx_hash == get_transaction_hash(tx1.tx));
+    }
+  }
+}

tests/unit/rest.test.cpp

@@ -31,22 +31,26 @@
 #include <boost/beast/websocket/error.hpp>
 #include <memory>
 #include <optional>
+#include <time.h>
 
+#include "cryptonote_basic/account.h"            // monero/src
+#include "cryptonote_core/cryptonote_tx_utils.h" // monero/src
 #include "db/data.h"
 #include "db/print.test.h"
 #include "db/storage.test.h"
 #include "db/string.h"
 #include "error.h"
 #include "hex.h" // monero/epee/contrib/include
+#include "mempool.h"
 #include "net/http_client.h"
 #include "rapidjson/document.h"     // monero/external/rapidjson/include
 #include "rapidjson/stringbuffer.h" // monero/external/rapidjson/incldue
 #include "rapidjson/prettywriter.h" // monero/external/rapidjson/incldue
 #include "rest_server.h"
-#include "scanner.test.h"
 #include "rpc/pull.test.h"
-
-#include "misc_log_ex.h"
+#include "scanner.test.h"
+#include "util/account.test.h"
+#include "util/transaction.test.h"
 
 namespace rapidjson
 {
@@ -160,10 +164,13 @@ namespace
 
 LWS_CASE("rest_server")
 {
-  lws::db::account_address account_address{};
-  crypto::secret_key view{};
-  crypto::generate_keys(account_address.spend_public, view);
-  crypto::generate_keys(account_address.view_public, view);
+  const auto now = std::chrono::system_clock::now();
+  const auto base = lws_test::make_account();
+  lws::db::account_address account_address{
+    base.m_account_address.m_view_public_key,
+    base.m_account_address.m_spend_public_key
+  };
+  const crypto::secret_key view = base.m_view_secret_key;
   const std::string address = lws::db::address_string(account_address);
   const std::string viewkey = epee::to_hex::string(epee::as_byte_span(unwrap(unwrap(view))));
 
@@ -175,6 +182,7 @@ LWS_CASE("rest_server")
     auto context =
       lws::rpc::context::make(lws_test::rpc_rendevous, {}, {}, {}, std::chrono::minutes{0}, false, true);
     const auto rpc = MONERO_UNWRAP(context.connect());
+    const auto init_server = [&] (std::shared_ptr<lws::mempool> pool) 
     {
       const lws::rest_server::configuration config{
         {}, {}, std::chrono::seconds{10}, 1, 20, {}, false, true, true, false
@@ -185,9 +193,12 @@ LWS_CASE("rest_server")
         std::vector<std::string>{admin_server},
         db.clone(),
         MONERO_UNWRAP(rpc.clone()),
+        std::move(pool),
         config
       );
-    }
+    };
+
+    init_server(nullptr);
 
     const lws::db::block_info last_block =
       MONERO_UNWRAP(MONERO_UNWRAP(db.start_read()).get_last_block());
@@ -592,8 +603,30 @@ LWS_CASE("rest_server")
       );
     }
 
-    SECTION("One Receive, One Spend")
+    SECTION("One Receive, One Spend, and one mempool")
     {
+      const auto pool = std::make_shared<lws::mempool>();
+
+      EXPECT(client.disconnect());
+      init_server(pool);
+      EXPECT(client.connect(std::chrono::milliseconds{500})); 
+
+      lws_test::transaction tx;
+      {
+        std::vector<cryptonote::tx_destination_entry> destinations;
+        destinations.emplace_back();
+        destinations.back().amount = 8000;
+        destinations.back().addr = base.m_account_address;
+        tx = lws_test::make_tx(lest_env, base, destinations, 20, true);
+
+        const auto hash = get_transaction_hash(tx.tx);
+        pool->reset_txs(
+          lws::mempool::pool_table{
+            {hash, std::make_shared<lws::mempool::pool_entry>(cryptonote::transaction{tx.tx}, now)}
+          }
+        );
+      }
+
       const std::string scan_height = std::to_string(std::uint64_t(account.scan_height) + 5);
       const std::string start_height = std::to_string(std::uint64_t(account.start_height));
       message = "{\"address\":\"" + address + "\",\"view_key\":\"" + viewkey + "\"}";
@@ -621,7 +654,7 @@ LWS_CASE("rest_server")
         lws::db::output{
           link,
           lws::db::output::spend_meta_{
-            lws::db::output_id{500, 30},
+            lws::db::output_id{0, 35},
             std::uint64_t(40000),
             std::uint32_t(16),
             std::uint32_t(2),
@@ -643,7 +676,7 @@ LWS_CASE("rest_server")
         lws::db::spend{
           link,
           image,
-          lws::db::output_id{500, 30},
+          lws::db::output_id{0, 35},
           std::uint64_t(66),
           std::uint64_t(1500),
           std::uint32_t(16),
@@ -688,37 +721,61 @@ LWS_CASE("rest_server")
       );
 
       response = invoke(client, "/get_address_txs", message);
-      EXPECT(response ==
-        "{\"total_received\":\"40000\","
-        "\"scanned_height\":" + scan_height + "," +
-        "\"scanned_block_height\":" + scan_height + ","
-        "\"start_height\":" + start_height + ","
-        "\"transaction_height\":" + scan_height + ","
-        "\"blockchain_height\":" + scan_height + ","
-        "\"transactions\":["
-          "{\"id\":0,"
-          "\"hash\":\"" + epee::to_hex::string(epee::as_byte_span(link.tx_hash)) + "\","
-          "\"timestamp\":\"1970-01-01T01:56:40Z\","
-          "\"total_received\":\"40000\","
-          "\"total_sent\":\"40000\","
-          "\"fee\":\"100\","
-          "\"unlock_time\":4670,"
-          "\"height\":4000,"
-          "\"payment_id\":\"" + epee::to_hex::string(epee::as_byte_span(payment_id_.long_)) + "\","
-          "\"coinbase\":true,"
-          "\"mempool\":false,"
-          "\"mixin\":16,"
-          "\"recipient\":{\"maj_i\":2,\"min_i\":66},"
-          "\"spent_outputs\":[{"
-            "\"amount\":\"40000\","
-            "\"key_image\":\"" + epee::to_hex::string(epee::as_byte_span(image)) + "\","
-            "\"tx_pub_key\":\"" + epee::to_hex::string(epee::as_byte_span(tx_public)) + "\","
-            "\"out_index\":2,"
-            "\"mixin\":16,"
-            "\"sender\":{\"maj_i\":4,\"min_i\":55}"
-          "}]}"
-        "]}"
-      );
+      const std::string expected =
+        R"({
+          "total_received":"48000",
+          "scanned_height":)" + scan_height + R"(,
+          "scanned_block_height":)" + scan_height + R"(,
+          "start_height":)" + start_height + R"(,
+          "transaction_height":)" + scan_height + R"(,
+          "blockchain_height":)" + scan_height + R"(,
+          "transactions":[
+            {
+              "id":0,
+              "hash":")" + epee::to_hex::string(epee::as_byte_span(link.tx_hash)) + R"(",
+              "timestamp":"1970-01-01T01:56:40Z",
+              "total_received":"40000",
+              "total_sent":"40000",
+              "fee":"100",
+              "unlock_time":4670,
+              "height":4000,
+              "payment_id":")" + epee::to_hex::string(epee::as_byte_span(payment_id_.long_)) + R"(",
+              "coinbase":true,
+              "mempool":false,
+              "mixin":16,
+              "recipient":{"maj_i":2,"min_i":66},
+              "spent_outputs":[{
+                "amount":"40000",
+                "key_image":")" + epee::to_hex::string(epee::as_byte_span(image)) + R"(",
+                "tx_pub_key":")" + epee::to_hex::string(epee::as_byte_span(tx_public)) + R"(",
+                "out_index":2,
+                "mixin":16,
+                "sender":{"maj_i":4,"min_i":55}
+              }]
+            },{
+              "id":1,
+              "hash":")" + epee::to_hex::string(epee::as_byte_span(get_transaction_hash(tx.tx))) + R"(",
+              "total_received":"8000",
+              "total_sent":"40000",
+              "fee":"12000",
+              "unlock_time":0,
+              "payment_id":"0000000000000000",
+              "coinbase":false,
+              "mempool":true,
+              "mixin":15,
+              "recipient":{"maj_i":0,"min_i":0},
+              "spent_outputs":[{
+                "amount":"40000",
+                "key_image":")" + epee::to_hex::string(epee::as_byte_span(tx.images.at(0))) + R"(",
+                "tx_pub_key":")" + epee::to_hex::string(epee::as_byte_span(tx_public)) + R"(",
+                "out_index":2,
+                "mixin":15,
+                "sender":{"maj_i":2, "min_i": 66}
+              }]
+            }
+          ]
+        })";
+      verify_json(lest_env, "/get_address_txs", response, expected);
 
       std::vector<epee::byte_slice> messages;
       messages.emplace_back(get_fee_response());
@@ -736,8 +793,8 @@ LWS_CASE("rest_server")
           "{\"amount\":\"40000\","
           "\"public_key\":\"" + epee::to_hex::string(epee::as_byte_span(pub)) + "\","
           "\"index\":2,"
-          "\"global_index\":30,"
-          "\"tx_id\":30,"
+          "\"global_index\":35,"
+          "\"tx_id\":35,"
           "\"tx_hash\":\"" + epee::to_hex::string(epee::as_byte_span(link.tx_hash)) + "\","
           "\"tx_prefix_hash\":\"" + epee::to_hex::string(epee::as_byte_span(tx_prefix)) + "\","
           "\"tx_pub_key\":\"" + epee::to_hex::string(epee::as_byte_span(tx_public)) + "\","
@@ -813,6 +870,7 @@ LWS_CASE("rest_server")
 
     SECTION("feed subscription")
     {
+      const auto pool = std::make_shared<lws::mempool>();
       const std::string scan_height = std::to_string(std::uint64_t(account.scan_height));
       namespace pull = lws_test::rpc::pull;
 
@@ -861,7 +919,7 @@ LWS_CASE("rest_server")
         io.run_one();
       }
 
-      std::string expected =
+      std::string expected=
         R"({"scanned_block_height":)" + scan_height + R"(,
           "start_height":)" + scan_height + R"(,
           "blockchain_height":)" + scan_height + "}";
@@ -872,6 +930,83 @@ LWS_CASE("rest_server")
       response.erase(0, std::strlen("tx_sync:"));
       verify_json(lest_env, "tx_sync", response, expected);
 
+      pull::async_close(conn, handler);
+
+      ran = false;
+      error = {};
+      response = {};
+      while (!ran && !io.stopped())
+      {
+        io.restart();
+        io.run_one();
+      }
+
+      EXPECT(!io.stopped());
+      EXPECT(error == boost::system::error_code{});
+      EXPECT(response == "");
+
+      // mempool tx in sync//
+      init_server(pool);
+      conn = pull::make(io, boost::asio::ip::tcp::endpoint(local, 10000), "lws.feed.v0.json");
+      pull::async_handshake(conn, message, handler);
+
+      lws_test::transaction tx;
+      {
+        std::vector<cryptonote::tx_destination_entry> destinations;
+        destinations.emplace_back();
+        destinations.back().amount = 8000;
+        destinations.back().addr = base.m_account_address;
+        tx = lws_test::make_tx(lest_env, base, destinations, 20, true);
+
+        const auto hash = get_transaction_hash(tx.tx);
+        pool->reset_txs(
+          lws::mempool::pool_table{
+            {hash, std::make_shared<lws::mempool::pool_entry>(cryptonote::transaction{tx.tx}, now)}
+          }
+        );
+      }
+ 
+      ran = false;
+      error = {};
+      response = {};
+      while (!ran && !io.stopped())
+      {
+        io.restart();
+        io.run_one();
+      }
+
+      expected =
+        R"({"scanned_block_height":)" + scan_height + R"(,
+          "start_height":)" + scan_height + R"(,
+          "blockchain_height":)" + scan_height + R"(,
+          "transactions":[{
+            "hash":")" + epee::to_hex::string(epee::as_byte_span(get_transaction_hash(tx.tx))) + R"(",
+            "prefix_hash":")" + epee::to_hex::string(epee::as_byte_span(get_transaction_prefix_hash(tx.tx))) + R"(",
+            "timestamp":)" + std::to_string(std::chrono::system_clock::to_time_t(now)) + R"(,
+            "fee":12000,
+            "unlock_time":0,
+            "payment_id": "0000000000000000",
+            "mempool":true,
+            "mixin":15,
+            "receives":[
+              {
+                "amount":8000,
+                "public_key":")" + epee::to_hex::string(epee::as_byte_span(tx.spend_publics.at(0))) + R"(",
+                "index":0,
+                "tx_pub_key":")" + epee::to_hex::string(epee::as_byte_span(tx.pub_keys.at(0))) + R"(",
+                "rct":")" + epee::to_hex::string(epee::as_byte_span(tx.ringct.at(0))) + R"("
+              }
+            ]
+          }]
+        })";
+
+      EXPECT(!io.stopped());
+      EXPECT(error == boost::system::error_code{});
+      EXPECT(get_prefix(response) == "tx_sync:");
+      response.erase(0, std::strlen("tx_sync:"));
+      verify_json(lest_env, "tx_sync non empty", response, expected);
+
+
       const lws::db::transaction_link link{
         lws::db::block_id::txpool, crypto::rand<crypto::hash>()
       };

tests/unit/scanner.test.cpp

@@ -38,11 +38,11 @@
 #include "db/print.test.h"
 #include "db/storage.test.h"
 #include "device/device_default.hpp" // monero/src
-#include "hardforks/hardforks.h"     // monero/src
 #include "net/zmq.h"                 // monero/src
 #include "rpc/client.h"
 #include "rpc/daemon_messages.h"     // monero/src
 #include "scanner.h"
+#include "util/transaction.test.h"
 #include "wire/error.h"
 #include "wire/json/write.h"
 
@@ -88,146 +88,6 @@ namespace
       ~join() { thread.join(); }
   };
    
-  struct transaction
-  {
-    cryptonote::transaction tx;
-    std::vector<crypto::secret_key> additional_keys;
-    std::vector<crypto::public_key> pub_keys;
-    std::vector<crypto::public_key> spend_publics;
-  };
-
-  transaction make_miner_tx(lest::env& lest_env, lws::db::block_id height, const lws::db::account_address& miner_address, bool use_view_tags)
-  {
-    static constexpr std::uint64_t fee = 0;
-
-    transaction tx{};
-    tx.pub_keys.emplace_back();
-    tx.spend_publics.emplace_back();
-
-    crypto::secret_key key;
-    crypto::generate_keys(tx.pub_keys.back(), key);
-    EXPECT(add_tx_pub_key_to_extra(tx.tx, tx.pub_keys.back()));
-
-    cryptonote::txin_gen in;
-    in.height = std::uint64_t(height);
-    tx.tx.vin.push_back(in);
-
-    // This will work, until size of constructed block is less then CRYPTONOTE_BLOCK_GRANTED_FULL_REWARD_ZONE
-    uint64_t block_reward;
-    EXPECT(cryptonote::get_block_reward(0, 0, 1000000, block_reward, num_testnet_hard_forks));
-    block_reward += fee;
-
-    crypto::key_derivation derivation;
-    EXPECT(crypto::generate_key_derivation(miner_address.view_public, key, derivation));
-    EXPECT(crypto::derive_public_key(derivation, 0, miner_address.spend_public, tx.spend_publics.back()));
- 
-    crypto::view_tag view_tag;
-    if (use_view_tags)
-      crypto::derive_view_tag(derivation, 0, view_tag);
-
-    cryptonote::tx_out out;
-    cryptonote::set_tx_out(block_reward, tx.spend_publics.back(), use_view_tags, view_tag, out);
-
-    tx.tx.vout.push_back(out);
-    tx.tx.version = 2;
-    tx.tx.unlock_time = std::uint64_t(height) + CRYPTONOTE_MINED_MONEY_UNLOCK_WINDOW;
-
-    return tx;
-  }
-
-  struct get_spend_public
-  {
-    std::vector<crypto::public_key>& pub_keys;
-
-    template<typename T>
-    void operator()(const T&) const noexcept
-    {}
-   
-    void operator()(const cryptonote::txout_to_key& val) const
-    { pub_keys.push_back(val.key); }
-
-    void operator()(const cryptonote::txout_to_tagged_key& val) const
-    { pub_keys.push_back(val.key); }
-  };
-
-  transaction make_tx(lest::env& lest_env, const cryptonote::account_keys& keys, std::vector<cryptonote::tx_destination_entry> destinations, const std::uint32_t ring_base, const bool use_view_tag)
-  {
-    static constexpr std::uint64_t input_amount = 20000;
-    static constexpr std::uint64_t output_amount = 8000;
-
-    EXPECT(15 < std::numeric_limits<std::uint32_t>::max() - ring_base);
-
-    crypto::secret_key unused_key{};
-    crypto::secret_key og_tx_key{};
-    crypto::public_key og_tx_public{};
-    crypto::generate_keys(og_tx_public, og_tx_key);
-
-    crypto::key_derivation derivation{};
-    crypto::public_key spend_public{};
-    EXPECT(crypto::generate_key_derivation(keys.m_account_address.m_view_public_key, og_tx_key, derivation));
-    EXPECT(crypto::derive_public_key(derivation, 0, keys.m_account_address.m_spend_public_key, spend_public));
-
-    std::uint32_t index = -1;
-    std::unordered_map<crypto::public_key, cryptonote::subaddress_index> subaddresses;
-    for (const auto& destination : destinations)
-    {
-      ++index;
-      subaddresses[destination.addr.m_spend_public_key] = {0, index};
-    }
-
-    if (2 < destinations.size())
-      destinations.erase(destinations.begin() + 1, destinations.end() - 1);
-
-    std::vector<cryptonote::tx_source_entry> sources;
-    sources.emplace_back();
-    sources.back().amount = input_amount;
-    sources.back().rct = true;
-    sources.back().real_output = 15;
-    sources.back().real_output_in_tx_index = 0;
-    sources.back().real_out_tx_key = og_tx_public;
-    for (std::uint32_t i = ring_base; i < 15 + ring_base; ++i)
-    {
-      crypto::public_key next{};
-      crypto::generate_keys(next, unused_key);
-      sources.back().push_output(i, next, 10000);
-    }
-    sources.back().outputs.emplace_back();
-    sources.back().outputs.back().first = 15 + ring_base;
-    sources.back().outputs.back().second.dest = rct::pk2rct(spend_public);
- 
-    transaction out{};
-    EXPECT(
-      cryptonote::construct_tx_and_get_tx_key(
-        keys, subaddresses, sources, destinations, keys.m_account_address, {}, out.tx, /* 0, */ unused_key,
-        out.additional_keys, true, {rct::RangeProofType::RangeProofPaddedBulletproof, 2}, use_view_tag
-      )
-    );
-
-    for (const auto& vout : out.tx.vout)
-      boost::apply_visitor(get_spend_public{out.spend_publics}, vout.target);
-
-    if (out.additional_keys.empty())
-    {
-      std::vector<cryptonote::tx_extra_field> extra;
-      EXPECT(cryptonote::parse_tx_extra(out.tx.extra, extra));
- 
-      cryptonote::tx_extra_pub_key key;
-      EXPECT(cryptonote::find_tx_extra_field_by_type(extra, key));
-
-      out.pub_keys.emplace_back();
-      out.pub_keys.back() = key.pub_key;
-    }
-    else
-    {
-      for (const auto& this_key : out.additional_keys)
-      {
-        out.pub_keys.emplace_back();
-        EXPECT(crypto::secret_key_to_public_key(this_key, out.pub_keys.back()));
-      }
-    }
-    return out;
-  }
-
   void scanner_thread(lws::scanner& scanner, void* ctx, const std::vector<epee::byte_slice>& reply)
   {
     struct stop_
@@ -331,6 +191,7 @@ LWS_CASE("lws::scanner::sync and lws::scanner::run")
 
   SETUP("lws::rpc::context, ZMQ_REP Server, and lws::db::storage")
   {
+    std::shared_ptr<lws::mempool> pool{};
     auto rpc = 
       lws::rpc::context::make(lws_test::rpc_rendevous, {}, {}, {}, std::chrono::minutes{0}, false, true);
 
@@ -448,15 +309,15 @@ LWS_CASE("lws::scanner::sync and lws::scanner::run")
       destinations.back().addr = keys.m_account_address;
 
       std::vector<epee::byte_slice> messages{};
-      transaction tx = make_miner_tx(lest_env, last_block.id, account, false);
+      lws_test::transaction tx = lws_test::make_miner_tx(lest_env, last_block.id, account, false);
       EXPECT(tx.pub_keys.size() == 1);
       EXPECT(tx.spend_publics.size() == 1);
 
-      transaction tx2 = make_tx(lest_env, keys, destinations, 20, true);
+      lws_test::transaction tx2 = lws_test::make_tx(lest_env, keys, destinations, 20, true);
       EXPECT(tx2.pub_keys.size() == 1);
       EXPECT(tx2.spend_publics.size() == 1);
 
-      transaction tx3 = make_tx(lest_env, keys, destinations, 86, false);
+      lws_test::transaction tx3 = lws_test::make_tx(lest_env, keys, destinations, 86, false);
       EXPECT(tx3.pub_keys.size() == 1);
       EXPECT(tx3.spend_publics.size() == 1);
 
@@ -465,7 +326,7 @@ LWS_CASE("lws::scanner::sync and lws::scanner::run")
       destinations.back().addr = keys_subaddr1.m_account_address;
       destinations.back().is_subaddress = true;
 
-      transaction tx4 = make_tx(lest_env, keys, destinations, 50, false);
+      lws_test::transaction tx4 = lws_test::make_tx(lest_env, keys, destinations, 50, false);
       EXPECT(tx4.pub_keys.size() == 1);
       EXPECT(tx4.spend_publics.size() == 2);
 
@@ -474,7 +335,7 @@ LWS_CASE("lws::scanner::sync and lws::scanner::run")
       //destinations.back().addr = keys_subaddr2.m_account_address;
       //destinations.back().is_subaddress = true;
 
-      //transaction tx5 = make_tx(lest_env, keys, destinations, 100, true);
+      //transaction tx5 = lws_test::make_tx(lest_env, keys, destinations, 100, true);
       //EXPECT(tx5.pub_keys.size() == 3);
       //EXPECT(tx5.spend_publics.size() == 3);
 
@@ -542,7 +403,7 @@ LWS_CASE("lws::scanner::sync and lws::scanner::run")
         lws::scanner scanner{db.clone(), epee::net_utils::ssl_verification_t::none};
         boost::thread server_thread(&scanner_thread, std::ref(scanner), rpc.zmq_context(), std::cref(messages));
         const join on_scope_exit{server_thread};
-        scanner.run(std::move(rpc), 1, {}, {}, opts);
+        scanner.run(std::move(rpc), pool, 1, {}, {}, opts);
       }
 
       hashes.push_back(cryptonote::get_block_hash(bmessage.blocks.back().block));
@@ -749,15 +610,15 @@ LWS_CASE("lws::scanner::sync and lws::scanner::run")
       destinations.back().addr = keys.m_account_address;
 
       std::vector<epee::byte_slice> messages{};
-      transaction tx = make_miner_tx(lest_env, last_block.id, account, false);
+      lws_test::transaction tx = lws_test::make_miner_tx(lest_env, last_block.id, account, false);
       EXPECT(tx.pub_keys.size() == 1);
       EXPECT(tx.spend_publics.size() == 1);
 
-      transaction tx2 = make_tx(lest_env, keys, destinations, 20, true);
+      lws_test::transaction tx2 = lws_test::make_tx(lest_env, keys, destinations, 20, true);
       EXPECT(tx2.pub_keys.size() == 1);
       EXPECT(tx2.spend_publics.size() == 1);
 
-      transaction tx3 = make_tx(lest_env, keys, destinations, 86, false);
+      lws_test::transaction tx3 = lws_test::make_tx(lest_env, keys, destinations, 86, false);
       EXPECT(tx3.pub_keys.size() == 1);
       EXPECT(tx3.spend_publics.size() == 1);
 
@@ -766,7 +627,7 @@ LWS_CASE("lws::scanner::sync and lws::scanner::run")
       destinations.back().addr = keys_subaddr1.m_account_address;
       destinations.back().is_subaddress = true;
 
-      transaction tx4 = make_tx(lest_env, keys, destinations, 50, false);
+      lws_test::transaction tx4 = lws_test::make_tx(lest_env, keys, destinations, 50, false);
       EXPECT(tx4.pub_keys.size() == 1);
       EXPECT(tx4.spend_publics.size() == 2);
 
@@ -775,7 +636,7 @@ LWS_CASE("lws::scanner::sync and lws::scanner::run")
       destinations.back().addr = keys_subaddr2.m_account_address;
       destinations.back().is_subaddress = true;
 
-      transaction tx5 = make_tx(lest_env, keys, destinations, 146, true);
+      lws_test::transaction tx5 = lws_test::make_tx(lest_env, keys, destinations, 146, true);
       EXPECT(tx5.pub_keys.size() == 1);
       EXPECT(tx5.spend_publics.size() == 2);
 
@@ -868,7 +729,7 @@ LWS_CASE("lws::scanner::sync and lws::scanner::run")
         lws::scanner scanner{db.clone(), epee::net_utils::ssl_verification_t::none};
         boost::thread server_thread(&scanner_thread, std::ref(scanner), rpc.zmq_context(), std::cref(messages));
         const join on_scope_exit{server_thread};
-        scanner.run(std::move(rpc), 1, {}, {}, opts);
+        scanner.run(std::move(rpc), pool, 1, {}, {}, opts);
       }
 
       hashes.push_back(cryptonote::get_block_hash(bmessage.blocks.back().block));

tests/unit/util/CMakeLists.txt

@@ -0,0 +1,30 @@
+# Copyright (c) 2024, 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.
+
+add_library(monero-lws-unit-util account.test.cpp account.test.h transaction.test.cpp transaction.test.h)
+target_link_libraries(monero-lws-unit-util monero-lws-unit-framework monero-lws-db monero::libraries)

tests/unit/util/account.test.cpp

@@ -0,0 +1,42 @@
+// 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 "account.test.h"
+
+#include "crypto/crypto.h"
+#include "cryptonote_basic/account.h"
+
+namespace lws_test
+{
+  cryptonote::account_keys make_account()
+  {
+    cryptonote::account_keys out{};
+    crypto::generate_keys(out.m_account_address.m_spend_public_key, out.m_spend_secret_key); 
+    crypto::generate_keys(out.m_account_address.m_view_public_key, out.m_view_secret_key);
+    return out;
+  }
+}

tests/unit/util/account.test.h

@@ -0,0 +1,34 @@
+// 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
+
+namespace cryptonote { class account_keys; }
+namespace lws_test
+{
+  cryptonote::account_keys make_account();
+}

tests/unit/util/transaction.test.cpp

@@ -0,0 +1,222 @@
+// 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 "transaction.test.h"
+
+#include "crypto/crypto.h"                       // monero/src
+#include "cryptonote_basic/account.h"            // monero/src
+#include "cryptonote_core/cryptonote_tx_utils.h" // monero/src
+#include "db/data.h"
+#include "hardforks/hardforks.h"                 // monero/src
+#include "util/transactions.h"
+
+namespace lws_test
+{
+  transaction make_miner_tx(lest::env& lest_env, lws::db::block_id height, const lws::db::account_address& miner_address, bool use_view_tags)
+  {
+    static constexpr std::uint64_t fee = 0;
+
+    transaction tx{};
+    tx.pub_keys.emplace_back();
+    tx.spend_publics.emplace_back();
+
+    crypto::secret_key key;
+    crypto::generate_keys(tx.pub_keys.back(), key);
+    EXPECT(add_tx_pub_key_to_extra(tx.tx, tx.pub_keys.back()));
+
+    cryptonote::txin_gen in;
+    in.height = std::uint64_t(height);
+    tx.tx.vin.push_back(in);
+
+    // This will work, until size of constructed block is less then CRYPTONOTE_BLOCK_GRANTED_FULL_REWARD_ZONE
+    uint64_t block_reward;
+    EXPECT(cryptonote::get_block_reward(0, 0, 1000000, block_reward, num_testnet_hard_forks));
+    block_reward += fee;
+
+    crypto::key_derivation derivation;
+    EXPECT(crypto::generate_key_derivation(miner_address.view_public, key, derivation));
+    EXPECT(crypto::derive_public_key(derivation, 0, miner_address.spend_public, tx.spend_publics.back()));
+ 
+    crypto::view_tag view_tag;
+    if (use_view_tags)
+      crypto::derive_view_tag(derivation, 0, view_tag);
+
+    cryptonote::tx_out out;
+    cryptonote::set_tx_out(block_reward, tx.spend_publics.back(), use_view_tags, view_tag, out);
+
+    tx.tx.vout.push_back(out);
+    tx.tx.version = 2;
+    tx.tx.unlock_time = std::uint64_t(height) + CRYPTONOTE_MINED_MONEY_UNLOCK_WINDOW;
+
+    return tx;
+  }
+
+  struct get_key_image
+  {
+    std::vector<crypto::key_image>& images;
+
+    template<typename T>
+    void operator()(const T&) const noexcept
+    {}
+
+    void operator()(const cryptonote::txin_to_key& val) const
+    {
+      images.push_back(val.k_image);
+    }
+  };
+
+  struct get_spend_public
+  {
+    std::vector<crypto::public_key>& pub_keys;
+
+    template<typename T>
+    void operator()(const T&) const noexcept
+    {}
+   
+    void operator()(const cryptonote::txout_to_key& val) const
+    { pub_keys.push_back(val.key); }
+
+    void operator()(const cryptonote::txout_to_tagged_key& val) const
+    { pub_keys.push_back(val.key); }
+  };
+
+  transaction make_tx(lest::env& lest_env, const cryptonote::account_keys& keys, std::vector<cryptonote::tx_destination_entry> destinations, const std::uint32_t ring_base, const bool use_view_tag)
+  {
+    static constexpr std::uint64_t input_amount = 20000;
+    static constexpr std::uint64_t output_amount = 8000;
+
+    EXPECT(15 < std::numeric_limits<std::uint32_t>::max() - ring_base);
+
+    crypto::secret_key unused_key{};
+    crypto::secret_key og_tx_key{};
+    crypto::public_key og_tx_public{};
+    crypto::generate_keys(og_tx_public, og_tx_key);
+
+    crypto::key_derivation derivation{};
+    crypto::public_key spend_public{};
+    EXPECT(crypto::generate_key_derivation(keys.m_account_address.m_view_public_key, og_tx_key, derivation));
+    EXPECT(crypto::derive_public_key(derivation, 0, keys.m_account_address.m_spend_public_key, spend_public));
+
+    std::uint32_t index = -1;
+    std::unordered_map<crypto::public_key, cryptonote::subaddress_index> subaddresses;
+    for (const auto& destination : destinations)
+    {
+      ++index;
+      subaddresses[destination.addr.m_spend_public_key] = {0, index};
+    }
+
+    if (2 < destinations.size())
+      destinations.erase(destinations.begin() + 1, destinations.end() - 1);
+
+    std::vector<cryptonote::tx_source_entry> sources;
+    sources.emplace_back();
+    sources.back().amount = input_amount;
+    sources.back().rct = true;
+    sources.back().real_output = 15;
+    sources.back().real_output_in_tx_index = 0;
+    sources.back().real_out_tx_key = og_tx_public;
+    for (std::uint32_t i = ring_base; i < 15 + ring_base; ++i)
+    {
+      crypto::public_key next{};
+      crypto::generate_keys(next, unused_key);
+      sources.back().push_output(i, next, 10000);
+    }
+    sources.back().outputs.emplace_back();
+    sources.back().outputs.back().first = 15 + ring_base;
+    sources.back().outputs.back().second.dest = rct::pk2rct(spend_public);
+ 
+    transaction out{};
+    EXPECT(
+      cryptonote::construct_tx_and_get_tx_key(
+        keys, subaddresses, sources, destinations, keys.m_account_address, {}, out.tx, /* 0, */ unused_key,
+        out.additional_keys, true, {rct::RangeProofType::RangeProofPaddedBulletproof, 2}, use_view_tag
+      )
+    );
+
+    for (const auto& vin : out.tx.vin)
+      boost::apply_visitor(get_key_image{out.images}, vin);
+
+    for (const auto& vout : out.tx.vout)
+      boost::apply_visitor(get_spend_public{out.spend_publics}, vout.target);
+
+    if (out.additional_keys.empty())
+    {
+      std::vector<cryptonote::tx_extra_field> extra;
+      EXPECT(cryptonote::parse_tx_extra(out.tx.extra, extra));
+ 
+      cryptonote::tx_extra_pub_key key;
+      EXPECT(cryptonote::find_tx_extra_field_by_type(extra, key));
+
+      out.pub_keys.emplace_back();
+      out.pub_keys.back() = key.pub_key;
+
+      crypto::key_derivation derivation{};
+      EXPECT(crypto::generate_key_derivation(key.pub_key, keys.m_view_secret_key, derivation));
+
+      for (index = 0; index < out.tx.vout.size(); ++index)
+      {
+        const auto decrypted = lws::decode_amount(
+          out.tx.rct_signatures.outPk.at(index).mask,
+          out.tx.rct_signatures.ecdhInfo.at(index),
+          derivation,
+          index,
+          true
+        );
+        EXPECT(bool(decrypted));
+        out.ringct.push_back(decrypted->second);
+      }
+    }
+    else
+    {
+      index = -1;
+      for (const auto& this_key : out.additional_keys)
+      {
+        ++index;
+        out.pub_keys.emplace_back();
+        EXPECT(crypto::secret_key_to_public_key(this_key, out.pub_keys.back()));
+
+        crypto::key_derivation derivation{};
+        EXPECT(crypto::generate_key_derivation(out.pub_keys.back(), keys.m_view_secret_key, derivation));
+
+        const auto decrypted = lws::decode_amount(
+          out.tx.rct_signatures.outPk.at(index).mask,
+          out.tx.rct_signatures.ecdhInfo.at(index),
+          derivation,
+          index,
+          true
+        );
+        EXPECT(bool(decrypted));
+        out.ringct.push_back(decrypted->second);
+      }
+    }
+
+    for (const auto& rct : out.tx.rct_signatures.ecdhInfo)
+      out.ringct.push_back(rct.mask);
+
+    return out;
+  }
+} // lws_test

tests/unit/util/transaction.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 <cstdint>
+#include <vector>
+
+#include "cryptonote_basic/cryptonote_basic.h"   // monero/src
+#include "db/fwd.h"
+#include "lest.hpp"
+
+namespace cryptonote
+{
+  class account_keys;
+  struct public_key;
+  class secret_key;
+  class tx_destination_entry;
+}
+namespace rct { struct key; }
+
+namespace lws_test
+{
+  struct transaction
+  {
+    cryptonote::transaction tx;
+    std::vector<crypto::secret_key> additional_keys;
+    std::vector<crypto::public_key> pub_keys;
+    std::vector<crypto::public_key> spend_publics;
+    std::vector<crypto::key_image> images;
+    std::vector<rct::key> ringct;
+  };
+
+  transaction make_miner_tx(lest::env& lest_env, lws::db::block_id height, const lws::db::account_address& miner_address, bool use_view_tags);
+  transaction make_tx(lest::env& lest_env, const cryptonote::account_keys& keys, std::vector<cryptonote::tx_destination_entry> destinations, const std::uint32_t ring_base, const bool use_view_tag);
+}