Add ZMQ-PUB notification for scan events

docs/zmq.md

@@ -15,6 +15,10 @@ option. Users are still required to "subscribe" to topics:
     creation that has recently triggered (identical output as webhook).
   * `msgpack-full-new_account_hook`: A msgpack object of a single new account
     creation that has recently triggered (identical output as webhook).
+  * `json-minimal-scanned`: A JSON object of a list of user primary addresses,
+    with their new height and block hash.
+  * `msgpack-minimal-scanned:` A msgpack object of a list of user primary
+    addresses with their new height and block hash.
 
 
 ### `json-full-payment_hook`/`msgpack-full-payment_hook`
@@ -95,3 +99,25 @@ PUB/SUB subscription model. The subscriber requests data from a certain "topic"
 where matching is done by string prefixes.
 
 > `index` is a counter used to detect dropped messages.
+
+
+### `json-minimal-scanned`/`msgpack-minimal-scanned`
+These topics receive PUB messages when a thread has finished scanning 1+
+accounts. The last block height and hash is sent.
+
+Example of the "raw" output from ZMQ-SUB side:
+```json
+json-minimal-scanned:{
+  "index": 13,
+  "event": {
+    "height": 2438536,
+    "id": "9197e1c6f3de28a98dfc579325903e5416ef1ba2681043c54b5fff0d39645a7f",
+    "addresses": [
+      "9xkhhJSa7ZhS5sAcTix6ozL14RwdgxbV7JZVFW4rCghN7GidutaykfxDHfgW45UPiCTXncuvZ91GNSGgxs3b2Cin9TU8nP3"
+    ]
+  }
+}
+```
+
+> `index` is a counter used to detect dropped messages
+

src/db/fwd.h

@@ -31,6 +31,8 @@
 
 namespace lws
 {
+  class account;
+
 namespace db
 {
   enum account_flags : std::uint8_t;

src/db/storage.h

@@ -252,7 +252,7 @@ namespace db
       \param chain List of block hashes that `accts` were scanned against.
       \param accts Updated to `height + chain.size()` scan height.
 
-      \return True iff LMDB successfully committed the update.
+      \return Number of updated accounts, and a list of webhooks that triggered.
     */
     expect<std::pair<std::size_t, std::vector<webhook_tx_confirmation>>>
       update(block_id height, epee::span<const crypto::hash> chain, epee::span<const lws::account> accts);

src/rpc/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-rpc_sources admin.cpp client.cpp daemon_pub.cpp daemon_zmq.cpp light_wallet.cpp rates.cpp)
-set(monero-lws-rpc_headers admin.h client.h daemon_pub.h daemon_zmq.h fwd.h json.h light_wallet.h rates.h)
+set(monero-lws-rpc_sources admin.cpp client.cpp daemon_pub.cpp daemon_zmq.cpp light_wallet.cpp lws_pub.cpp rates.cpp)
+set(monero-lws-rpc_headers admin.h client.h daemon_pub.h daemon_zmq.h fwd.h json.h light_wallet.h lws_pub.h rates.h)
 
 add_library(monero-lws-rpc ${monero-lws-rpc_sources} ${monero-lws-rpc_headers})
 target_include_directories(monero-lws-rpc PRIVATE ${RMQ_INCLUDE_DIR})

src/rpc/fwd.h

@@ -29,5 +29,9 @@
 
 namespace lws
 {
+  namespace rpc
+  {
+    class client;
+  }
   struct rates;
 }

src/rpc/webhook.h

@@ -1,3 +1,31 @@
+// Copyright (c) 2023-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 <boost/utility/string_ref.hpp>

src/scanner.cpp

@@ -55,6 +55,7 @@
 #include "rpc/daemon_messages.h"   // monero/src
 #include "rpc/daemon_zmq.h"
 #include "rpc/json.h"
+#include "rpc/lws_pub.h"
 #include "rpc/message_data_structs.h" // monero/src
 #include "rpc/webhook.h"
 #include "util/source_location.h"
@@ -692,7 +693,7 @@ namespace lws
 
           blockchain.push_back(cryptonote::get_block_hash(fetched->blocks.front().block));
 
-          auto blocks = epee::to_span(fetched->blocks);
+          auto blocks = epee::to_mut_span(fetched->blocks);
           auto indices = epee::to_span(fetched->output_indices);
 
           if (fetched->start_height != 1)
@@ -769,6 +770,7 @@ namespace lws
 
           MINFO("Processed " << blocks.size() << " block(s) against " << users.size() << " account(s)");
           send_payment_hook(client, epee::to_span(updated->second), opts.webhook_verify);
+ 
           if (updated->first != users.size())
           {
             MWARNING("Only updated " << updated->first << " account(s) out of " << users.size() << ", resetting");
@@ -777,6 +779,10 @@ namespace lws
 
           for (account& user : users)
             user.updated(db::block_id(fetched->start_height));
+
+          // Publish when all scan threads have past this block
+          if (!blockchain.empty() && client.has_publish())
+            rpc::publish_scanned(client, blockchain.back(), epee::to_span(users));
         }
       }
       catch (std::exception const& e)

src/wire/write.h

@@ -193,7 +193,13 @@ namespace wire_write
 
   template<typename T>
   inline std::size_t array_size_(std::true_type, const T& source)
-  { return boost::size(source); }
+  {
+    static_assert(
+      !std::is_same<typename std::iterator_traits<typename T::const_iterator>::iterator_category, std::input_iterator_tag>{},
+      "Input iterators must use json (or similar) derived classes directly"
+    );
+    return boost::size(source);
+  }
 
   template<typename T>
   inline constexpr std::size_t array_size_(std::false_type, const T&) noexcept