diff --git a/docs/api/feed.md b/docs/api/feed.md new file mode 100644 index 0000000..8867243 --- /dev/null +++ b/docs/api/feed.md @@ -0,0 +1,258 @@ +# '/feed' API + +The `/feed` websocket API allows for faster synchronization on initalization (msgpack +with less redundant data compared to [wallet api](/api/wallet/)), and "push" +updates instead of polling. New block arrivals, new received/spent transactions, +and reorgs are all pushed to the client in real-time. This should provide an +enterprise-like experience where the (usually mobile) client is seamlessly +synchronized with `monerod` (but only if `monero-lws-daemon` is using ZMQ +PUB/SUB from `monerod`). + +## Basics +The client initiates a `/feed` websocket connection, then the LWS server sends +the entire tx history, and then finally sends a continous stream of updates from +`monerod` (ZMQ PUB mempool+blocks messages) to the client. + +## Flow +### Handshake +The client must attempt a websocket "upgrade" connection to `/feed`. The +initial handshake must have a HTTP/1.1 field, `Sec-WebSocket-Protocol`, which +is set to msgpack or json: + +``` +Sec-WebSocket-Protocol: lws.feed.v0.msgpack +Sec-WebSocket-Protocol: lws.feed.v0.json +``` + +This parameter signals the protocol version and payload type. All message +payloads must match the type set in the header. Msgpack is recommended as both +serialization and de-serialization is faster, and the messages are more +compact (especially since binary crypto blobs are not encoded as hex). + +### Synchronization +After the websocket handshake, the client must send a [login](#login) message, +and the server must respond with a [tx-sync](#tx-sync) message OR +[error](#error) message. An `error` message signals the websocket is now in the +"terminated" state, a `tx-sync` message indicates that server is now ready to +send real-time updates. In either case, the client shall not send any further +data - the server will terminate the connection if this is not followed. + +### Updates +After the `tx-sync`, the server will then push [blocks](#blocks), +[mempool](#mempool), [warning](#warning), or [error](#error) messages until the +stream is terminated. An `error` message indicates the stream is immediately +terminated, whereas the other messages indicate a valid stream where clients +should expected further messages. + +## Message Format +Every message has a prefix before the msgpack/json payload that indicates the +type. Current valid prefixes are `login:`, `tx-sync:`, `blocks:`, `mempool:`, +`warning:`, and `error:`. The `login:` prefix is from the client, and the +remainder are sent by the server. The root type of every payload is an object. +As an example, a client login message (in the json sub-protocol), will look +like: + +```json +login:{ + "account":{ + "address": "47nPhxp2cJeKN2NjamupNUNA13XgzcYPzQBCzzsKcj717s8M2UpFVmmdwSuYwgyy8kPDwU7hpEqTTDvfe5LAb9Aj6nwmEzf", + "view_key": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205" + } +} +``` +This format (with prefixes) allows for single-pass DOMless parsing. + +### login +Messages of this type are sent (only) by the client. The message contains +proof the client has authority to receives updates for a given primary address, +along with optional restrictions on data sent/pushed by the server. + +[schema](schemas/feed_login.json) + +#### Example +```json +login:{ + "account":{ + "address": "47nPhxp2cJeKN2NjamupNUNA13XgzcYPzQBCzzsKcj717s8M2UpFVmmdwSuYwgyy8kPDwU7hpEqTTDvfe5LAb9Aj6nwmEzf", + "view_key": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205" + }, + "tx_sync": false, + "receives_only": true +} +``` + +### tx-sync +Messages of this type are sent only by the server, and only in response to a +successful `login:` command from the client. The message contains everything +needed for the client to have a full tx synchronization with the backend +(unless the client requested no transactions). + +[schema](schemas/feed_tx_sync.json) + +#### Example +```json +tx_sync:{ + "scanned_block_height": 10, + "blockchain_height": 10, + "lookahead_fail": 5, + "lookahead": {"maj_i": 50, "min_i": 200 }, + "transactions": [ + { + "hash": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205", + "prefix_hash": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205", + "timestamp": 0, + "fee": 0, + "unlock_time": 0, + "mixin": 0, + "payment_id": "ee171270296bbc26", + "coinbase": true, + "mempool": true, + "receives": [ + { + "id": {"legacy": {"amount": 0, "index": 0}}, + "amount": 0, + "public_key": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205", + "index": 0, + "tx_pub_key": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205", + "rct": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205", + "recipient": {"maj_i": 1, "min_i": 1} + } + ], + "spends": [ + { + "id": {"unified": 0}, + "key_image": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205" + } + ] + } + ] +} +``` + +### blocks +Messages of this type are sent only by the server when a new block has been +scanned on the account. If another block was added to the Monero chain, this +message type is not sent until a new block has been scanned against the +account. This will also give real-time updates when account is scanning +historical blocks (although the server can choose to group/aggregate messages +to reduce total number of messages). + +[schema](schemas/feed_blocks.json) + +#### Example +```json +blocks:{ + "scan_start": 150, + "scan_end": 151, + "blockchain_height": 151, + "lookahead_fail": 0, + "lookahead": {"maj_i": 50, "min_i": 200 }, + "transactions": [ + { + "hash": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205", + "prefix_hash": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205", + "timestamp": 0, + "fee": 0, + "unlock_time": 0, + "mixin": 0, + "payment_id": "ee171270296bbc26", + "coinbase": true, + "receives": [ + { + "id": {"legacy": {"amount": 0, "index": 0}}, + "amount": 0, + "public_key": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205", + "index": 0, + "tx_pub_key": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205", + "rct": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205", + "recipient": {"maj_i": 1, "min_i": 1} + } + ], + "spends": [ + { + "id": {"unified": 0}, + "key_image": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205" + } + ] + } + ] +} +``` + +### mempool +If the account scanner has caught-up with the most recent block, the server may +send _received outputs_ that are still in the mempool. These outputs will be +re-sent in the `blocks:` message in a transaction; the ID numbers for outputs +are not assigned until the transaction is included in a block. Spends are +assumed to be under the control of the client, and are not sent in mempool +messages. + +[schema](schemas/feed_mempool.json) + +#### Full Example +```json +mempool:{ + "hash": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205", + "prefix_hash": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205", + "timestamp": 0, + "fee": 0, + "unlock_time": 0, + "mixin": 0, + "payment_id": "ee171270296bbc26", + "amount": 0, + "public_key": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205", + "tx_pub_key": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205", + "index": 0, + "recipient": {"maj_i": 1, "min_i": 1}, + "rct": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205" +} +``` + +### warning +The primary purpose of this message type is to alert the client that the server +has detected a block re-organization. The rollback point is transmitted in the +message, and the client is expected to rollback transactions that are part of +the re-organization. Other events, such as lost connection to `monerod` are also +sent as warnings. Most other events will be [error](#error). + +[schema](schemas/feed_warning.json) + +#### Example +```json +warning:{ + "msg": "Descriptive warning message", + "code": 4, + "counter": 10, + "height": 1000 +} +``` + +#### Codes +```c++ + unspecified_error = 0, + account_not_found = 1, + bad_address = 2, + bad_view_key = 3, + blockchain_reorg = 4, + daemon_unresponsive = 5, + parse_error = 6, + protocol_error = 7, + queue_error = 8, + schema_error = 9 +``` + +### error +Servers will send an error of this type when the websocket must be closed; no +further messages are to be sent after sending this type. The codes sent in this +message type are identical to the ones [specified aboved](#codes) for warning +messages. + +[schema](schemas/feed_error.json) + +#### Example +```json +error:{ + "msg": "Descriptive error message", + "code": 8 +} +``` diff --git a/docs/api/schemas/feed_base.json b/docs/api/schemas/feed_base.json new file mode 100644 index 0000000..87f07bf --- /dev/null +++ b/docs/api/schemas/feed_base.json @@ -0,0 +1,147 @@ +{ + "$schema": "https://json-schema.org/draft-07/schema", + "$id": "https://docs.monerolws.com/api/schemas/feed_base.json", + "definitions": { + "address_meta": { + "type": "object", + "properties": { + "maj_i": {"$ref": "#/definitions/uint32"}, + "min_i": {"$ref": "#/definitions/uint32"} + }, + "required": ["maj_i", "min_i"] + }, + "binary8": { + "type": "string", + "minLength": 16, + "maxLength": 16, + "pattern": "^[0-9A-Fa-f]{16}$", + "example": "ee171270296bbc26" + }, + "binary32": { + "type": "string", + "minLength": 64, + "maxLength": 64, + "pattern": "^[0-9A-Fa-f]{64}$", + "example": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205" + }, + "composite_id": { + "anyOf": [ + {"$ref": "#/definitions/legacy_id"}, + {"$ref": "#/definitions/unified_id"} + ] + }, + "int64": { + "type": "integer", + "minimum": -9223372036854775808, + "maximum": 9223372036854775807 + }, + "legacy_id": { + "type": "object", + "properties": { + "legacy": { + "type": "object", + "properties": { + "amount": {"$ref": "#/definitions/uint64"}, + "index": {"$ref": "#/definitions/uint64"} + }, + "required": ["amount", "index"] + } + }, + "required": ["legacy"] + }, + "receive" : { + "type": "object", + "properties": { + "amount": {"$ref": "#/definitions/uint64"}, + "public_key": {"$ref": "#/definitions/binary32"}, + "index": {"$ref": "#/definitions/uint32"}, + "tx_pub_key": {"$ref": "#/definitions/binary32"}, + "id": {"$ref": "#/definitions/composite_id"}, + "rct": {"$ref": "#/definitions/binary32"}, + "recipient": { + "$ref": "#/definitions/address_meta", + "default": "{0, 0}" + } + }, + "required": [ + "amount", + "public_key", + "index", + "tx_pub_key" + ] + }, + "spend": { + "type": "object", + "properties": { + "id": {"$ref": "#/definitions/composite_id"}, + "key_image": {"$ref": "#/definitions/binary32"} + }, + "required": ["key_image"] + }, + "transaction": { + "type": "object", + "properties": { + "hash": {"$ref": "#/definitions/binary32"}, + "prefix_hash": {"$ref": "#/definitions/binary32"}, + "timestamp": {"$ref": "#/definitions/int64"}, + "fee": {"$ref": "#/definitions/uint64"}, + "unlock_time": {"$ref": "#/definitions/uint64"}, + "mixin": {"$ref": "#/definitions/uint32"}, + "height": {"$ref": "#/definitions/uint64"}, + "payment_id": { + "anyOf": [ + {"$ref": "#/definitions/binary8"}, + {"$ref": "#/definitions/binary32"} + ] + }, + "coinbase": { + "type": "boolean", + "default": false + }, + "mempool": { + "type": "boolean", + "default": false + }, + "spends": { + "type": "array", + "minItems": 1, + "items": {"$ref" : "#/definitions/spend"} + }, + "receives": { + "type": "array", + "minItems": 1, + "items": {"$ref" : "#/definitions/receive"} + } + }, + "required": [ + "hash", + "prefix_hash", + "timestamp", + "fee", + "unlock_time" + ] + }, + "uint16": { + "type": "integer", + "minimum": 0, + "maximum": 65535 + }, + "uint32": { + "type": "integer", + "minimum": 0, + "maximum": 4294967295 + }, + "uint64": { + "type": "integer", + "minimum": 0, + "maximum": 18446744073709551615 + }, + "unified_id": { + "type": "object", + "properties": { + "unified": {"$ref" : "#/definitions/uint64"} + }, + "required": ["unified"] + } + } +} diff --git a/docs/api/schemas/feed_blocks.json b/docs/api/schemas/feed_blocks.json new file mode 100644 index 0000000..fecbf0d --- /dev/null +++ b/docs/api/schemas/feed_blocks.json @@ -0,0 +1,21 @@ +{ + "$schema": "https://json-schema.org/draft-07/schema", + "$id": "https://docs.monerolws.com/api/schemas/feed_blocks.json", + "type": "object", + "properties": { + "scan_start": {"$ref": "feed_base.json#/definitions/uint64"}, + "scan_end": {"$ref": "feed_base.json#/definitions/uint64"}, + "blockchain_height": {"$ref": "feed_base.json#/definitions/uint64"}, + "lookahead_fail": {"$ref": "feed_base.json#/definitions/uint64"}, + "lookahead": { + "$ref": "feed_base.json#/definitions/address_meta", + "default": "{0, 0}" + }, + "transactions": { + "type": "array", + "minItems": 1, + "items": {"$ref": "feed_base.json#/definitions/transaction"} + } + }, + "required": ["scan_start", "scan_end", "blockchain_height"] +} diff --git a/docs/api/schemas/feed_error.json b/docs/api/schemas/feed_error.json new file mode 100644 index 0000000..4b95338 --- /dev/null +++ b/docs/api/schemas/feed_error.json @@ -0,0 +1,10 @@ +{ + "$schema": "https://json-schema.org/draft-07/schema", + "$id": "https://docs.monerolws.com/api/schemas/feed_error.json", + "type": "object", + "properties": { + "msg": {"type": "string"}, + "code": {"$ref": "feed_base.json#/definitions/uint16"} + }, + "required": ["msg", "code"] +} diff --git a/docs/api/schemas/feed_login.json b/docs/api/schemas/feed_login.json new file mode 100644 index 0000000..02dd4d2 --- /dev/null +++ b/docs/api/schemas/feed_login.json @@ -0,0 +1,36 @@ +{ + "$schema": "https://json-schema.org/draft-07/schema", + "$id": "https://docs.monerolws.com/api/schemas/feed_login.json", + "type": "object", + "properties": { + "account": { + "type": "object", + "properties": { + "address": {"$ref": "#/definitions/base58-address"}, + "view_key": {"$ref": "feed_base.json#/definitions/binary32"} + }, + "required": ["address", "view_key"] + }, + "tx_sync": { + "description": "Whether to include txes in 'tx_sync:' response", + "type": "boolean", + "default": true + }, + "receives_only": { + "description": "Send only received outputs messages after 'tx_sync:' message", + "type": "boolean", + "default": false + } + }, + "required": ["account"], + "definitions": { + "base58-address": { + "type": "string", + "minLength": 95, + "maxLength": 95, + "pattern": "^[0-9A-Za-z]{95}$", + "example": "47nPhxp2cJeKN2NjamupNUNA13XgzcYPzQBCzzsKcj717s8M2UpFVmmdwSuYwgyy8kPDwU7hpEqTTDvfe5LAb9Aj6nwmEzf" + } + } +} + diff --git a/docs/api/schemas/feed_mempool.json b/docs/api/schemas/feed_mempool.json new file mode 100644 index 0000000..dbce94e --- /dev/null +++ b/docs/api/schemas/feed_mempool.json @@ -0,0 +1,33 @@ +{ + "$schema": "https://json-schema.org/draft-07/schema", + "$id": "https://docs.monerolws.com/api/schemas/feed_mempool.json", + "type": "object", + "properties": { + "hash": {"$ref": "feed_base.json#/definitions/binary32"}, + "prefix_hash": {"$ref": "feed_base.json#/definitions/binary32"}, + "fee": {"$ref": "feed_base.json#/definitions/uint64"}, + "unlock_time": {"$ref": "feed_base.json#/definitions/uint64"}, + "amount": {"$ref": "feed_base.json#/definitions/uint64"}, + "public_key": {"$ref": "feed_base.json#/definitions/binary32"}, + "tx_pub_key": {"$ref": "feed_base.json#/definitions/binary32"}, + "index": {"$ref": "feed_base.json#/definitions/uint32"}, + "payment_id": { + "anyOf": [ + {"$ref": "feed_base.json#/definitions/binary8"}, + {"$ref": "feed_base.json#/definitions/binary32"} + ] + }, + "mixin": {"$ref": "feed_base.json#/definitions/uint32"}, + "recipient": {"$ref": "feed_base.json#/definitions/address_meta"}, + "rct": {"$ref": "feed_base.json#/definitions/binary32"} + }, + "required": [ + "hash", + "prefix_hash", + "fee", + "unlock_time", + "public_key", + "tx_pub_key", + "index" + ] +} diff --git a/docs/api/schemas/feed_tx_sync.json b/docs/api/schemas/feed_tx_sync.json new file mode 100644 index 0000000..ad8c0d7 --- /dev/null +++ b/docs/api/schemas/feed_tx_sync.json @@ -0,0 +1,20 @@ +{ + "$schema": "https://json-schema.org/draft-07/schema", + "$id": "https://docs.monerolws.com/api/schemas/feed_tx_sync.json", + "type": "object", + "properties": { + "scanned_block_height": {"$ref": "feed_base.json#/definitions/uint64"}, + "blockchain_height": {"$ref": "feed_base.json#/definitions/uint64"}, + "lookahead_fail": {"$ref": "feed_base.json#/definitions/uint64"}, + "lookahead": { + "$ref": "feed_base.json#/definitions/address_meta", + "default": "{0, 0}" + }, + "transactions": { + "type": "array", + "minItems": 1, + "items": {"$ref": "feed_base.json#/definitions/transaction"} + } + }, + "required": ["scanned_block_height", "blockchain_height"] +} diff --git a/docs/api/schemas/feed_warning.json b/docs/api/schemas/feed_warning.json new file mode 100644 index 0000000..2bbd94a --- /dev/null +++ b/docs/api/schemas/feed_warning.json @@ -0,0 +1,12 @@ +{ + "$schema": "https://json-schema.org/draft-07/schema", + "$id": "https://docs.monerolws.com/api/schemas/feed_warning.json", + "type": "object", + "properties": { + "msg": {"type": "string"}, + "code": {"$ref": "feed_base.json#/definitions/uint16"}, + "counter": {"$ref": "feed_base.json#/definitions/uint32"}, + "height": {"$ref": "feed_base.json#/definitions/uint64"} + }, + "required": ["msg", "code", "counter", "height"] +} diff --git a/docs/api/wallet.yaml b/docs/api/wallet.yaml index 564fba5..dfdf4b5 100644 --- a/docs/api/wallet.yaml +++ b/docs/api/wallet.yaml @@ -18,6 +18,16 @@ tags: - name: subaddress - name: transaction paths: + /feed: + get: + tags: [transaction] + summary: Check/upgrade to websocket /feed + operationId: feed + responses: + '101': + description: Switching to websocket /feed protocol + '501': + description: /feed is enabled on websocket upgrade only /get_address_info: post: tags: diff --git a/mkdocs.yml b/mkdocs.yml index 65bad70..0ccb1f1 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -19,6 +19,7 @@ nav: - Wallet: api/wallet.md - Admin: api/admin.md - Webhooks/ZeroMQ: api/zmq.md + - Feed: api/feed.md - RabbitMQ: api/rmq.md plugins: - swagger-ui-tag diff --git a/src/client_main.cpp b/src/client_main.cpp index e473670..ca1a961 100644 --- a/src/client_main.cpp +++ b/src/client_main.cpp @@ -195,7 +195,7 @@ namespace { std::shared_ptr client_; - bool operator()(boost::asio::io_context&, lws::rpc::client&, net::http::client&, epee::span chain, epee::span users, epee::span pow) + bool operator()(boost::asio::io_context&, lws::rpc::client&, net::http::client&, epee::span chain, epee::span users, epee::span pow) { if (!client_) return false; @@ -203,15 +203,21 @@ namespace return true; if (!pow.empty() || chain.empty()) return false; + if (!std::is_sorted(users.begin(), users.end(), lws::by_height{})) + return false; std::vector chain_copy{}; chain_copy.reserve(chain.size()); std::copy(chain.begin(), chain.end(), std::back_inserter(chain_copy)); + const lws::db::block_id height = lws::db::block_id(to_uint(users[0].scan_height()) + chain.size() - 1); std::vector users_copy{}; users_copy.reserve(users.size()); - for (const auto& user : users) + for (auto& user : users) + { users_copy.push_back(user.clone()); + user.updated(height); + } MINFO("Processed " << chain.size() << " block(s) against " << users.size() << " account(s)"); lws::rpc::scanner::client::send_update(client_, std::move(users_copy), std::move(chain_copy)); @@ -271,7 +277,7 @@ namespace void run(program prog) { MINFO("Using monerod ZMQ RPC at " << prog.monerod_rpc); - auto ctx = lws::rpc::context::make(std::move(prog.monerod_rpc), std::move(prog.monerod_sub), {}, {}, std::chrono::minutes{0}, false); + auto ctx = lws::rpc::context::make(std::move(prog.monerod_rpc), std::move(prog.monerod_sub), {}, {}, std::chrono::minutes{0}, false, false); lws::scanner_sync self{epee::net_utils::ssl_verification_t::system_ca}; diff --git a/src/config.h b/src/config.h index f294df6..88ccead 100644 --- a/src/config.h +++ b/src/config.h @@ -1,5 +1,6 @@ #pragma once +#include #include #include "cryptonote_config.h" @@ -9,6 +10,9 @@ namespace config { extern cryptonote::network_type network; + //! Default inactivity timeout for `/feed` websocket connections + constexpr const std::chrono::minutes feed_timeout{4}; + //! Max queued clients when waiting for cacheable daemon response constexpr const std::size_t daemon_wait_queue_max = 25000; diff --git a/src/db/account.cpp b/src/db/account.cpp index a6d504d..6d09530 100644 --- a/src/db/account.cpp +++ b/src/db/account.cpp @@ -51,6 +51,17 @@ namespace lws return std::memcmp(std::addressof(lhs), std::addressof(rhs), sizeof(lhs)) < 0; } }; + + template + static void map_update(F& format, T& self) + { + wire::object(format, + wire::field<0>("spends", wire::trusted_array(std::ref(self.spends))), + wire::field<1>("outputs", wire::trusted_array(std::ref(self.outputs))), + WIRE_FIELD_ID(3, old_height), + WIRE_FIELD_ID(4, new_height) + ); + } } struct account::internal @@ -126,6 +137,15 @@ namespace lws ); } + void account::update::read_bytes(::wire::msgpack_reader& source) + { map_update(source, *this); } + + void account::update::write_bytes(::wire::msgpack_writer& dest) const + { map_update(dest, *this); } + + void mempool_receive::write_bytes(::wire::msgpack_writer& dest) const + { map_update(dest, *this); } + account::account() noexcept : immutable_(nullptr), spendable_(), pubs_(), spends_(), outputs_(), height_(db::block_id(0)) {} @@ -163,13 +183,15 @@ namespace lws return result; } - void account::updated(db::block_id new_height) noexcept + account::update account::updated(db::block_id new_height) noexcept { + update out{std::move(spends_), std::move(outputs_), height_, new_height}; + height_ = std::max(height_, new_height); spends_.clear(); - spends_.shrink_to_fit(); outputs_.clear(); - outputs_.shrink_to_fit(); + + return out; } db::account_id account::id() const noexcept diff --git a/src/db/account.h b/src/db/account.h index c78b2d9..c3a64a3 100644 --- a/src/db/account.h +++ b/src/db/account.h @@ -26,6 +26,7 @@ // THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. #pragma once +#include #include #include #include @@ -40,7 +41,25 @@ #include "wire/msgpack/fwd.h" namespace lws -{ +{ + //! Sent internallly, same wire format as `account::update` + struct mempool_receive + { + std::array outputs; + db::block_id old_height; + db::block_id new_height; + std::array spends; + + explicit mempool_receive(const db::output& receive) noexcept + : outputs{{receive}}, + old_height(db::block_id::txpool), + new_height(db::block_id::txpool), + spends() + {} + + void write_bytes(::wire::msgpack_writer& dest) const; + }; + //! Tracks a subset of DB account info for scanning/updating. class account { @@ -61,6 +80,18 @@ namespace lws public: + //! Sent internally via ZMQ from scanner thread to websocket "push" thread + struct update + { + std::vector spends; + std::vector outputs; + db::block_id old_height; + db::block_id new_height; + + void read_bytes(::wire::msgpack_reader& source); + void write_bytes(::wire::msgpack_writer& dest) const; + }; + //! Construct an "invalid" account (for de-serialization) account() noexcept; @@ -89,7 +120,7 @@ namespace lws account clone() const; //! \post `height() == max(new_height, height())`, `outputs().empty()`, and `spends.empty()`. - void updated(db::block_id new_height) noexcept; + update updated(db::block_id new_height) noexcept; //! \return Unique ID from the account database, possibly `db::account_id::kInvalid`. db::account_id id() const noexcept; diff --git a/src/db/fwd.h b/src/db/fwd.h index 583b240..09045b1 100644 --- a/src/db/fwd.h +++ b/src/db/fwd.h @@ -32,6 +32,7 @@ namespace lws { class account; + class mempool_receive; namespace db { @@ -56,6 +57,7 @@ namespace db struct request_info; struct spend; class storage; + class storage_reader; struct subaddress_map; struct transaction_link; struct view_key; diff --git a/src/error.cpp b/src/error.cpp index 3d7b0d6..6ec3725 100644 --- a/src/error.cpp +++ b/src/error.cpp @@ -26,6 +26,8 @@ // THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. #include "error.h" +#include +#include #include namespace lws @@ -144,6 +146,16 @@ namespace lws } }; + void throw_invalid_argument(const int line, const char* file) + { + if (!file) + file = ""; + char const* const end = std::strrchr(file, '/'); + if (end) + file = end + 1; + throw std::invalid_argument{"Failed pre-condition at " + std::string{file} + ":" + std::to_string(line)}; + } + std::error_category const& error_category() noexcept { static const category instance{}; diff --git a/src/error.h b/src/error.h index 2980e87..4ce3dec 100644 --- a/src/error.h +++ b/src/error.h @@ -29,6 +29,13 @@ #include #include +#define LWS_VERIFY(x) \ + do \ + { \ + if (!(x)) \ + ::lws::throw_invalid_argument(__LINE__, __FILE__); \ + } while (0) + namespace lws { enum class error : int @@ -72,6 +79,7 @@ namespace lws tx_relay_failed //!< Daemon failed to relayed tx from REST client }; + [[noreturn]] void throw_invalid_argument(int line, const char* file); std::error_category const& error_category() noexcept; inline std::error_code make_error_code(lws::error value) noexcept diff --git a/src/rest_server.cpp b/src/rest_server.cpp index 8578e91..21b40bc 100644 --- a/src/rest_server.cpp +++ b/src/rest_server.cpp @@ -47,6 +47,7 @@ #include #include #include +#include #include #include #include @@ -80,7 +81,9 @@ #include "rpc/admin.h" #include "rpc/client.h" #include "rpc/daemon_messages.h" // monero/src +#include "rpc/feed.h" #include "rpc/light_wallet.h" +#include "rpc/login.h" #include "rpc/rates.h" #include "rpc/webhook.h" #include "util/gamma_picker.h" @@ -93,11 +96,21 @@ namespace lws { struct runtime_options { + const std::chrono::seconds feed_timeout; const std::uint32_t max_subaddresses; const epee::net_utils::ssl_verification_t webhook_verify; const bool disable_admin_auth; const bool auto_accept_creation; const bool auto_accept_import; + + explicit runtime_options(const rest_server::configuration& config) + : feed_timeout(config.feed_timeout), + max_subaddresses(config.max_subaddresses), + webhook_verify(config.webhook_verify), + disable_admin_auth(config.disable_admin_auth), + auto_accept_creation(config.auto_accept_creation), + auto_accept_import(config.auto_accept_import) + {} }; struct rest_server_data @@ -207,65 +220,7 @@ namespace lws return to_uint(tx_height) + CRYPTONOTE_DEFAULT_TX_SPENDABLE_AGE > to_uint(last); } - std::vector::const_iterator - find_metadata(std::vector const& metas, db::output_id id) - { - struct by_output_id - { - bool operator()(db::output::spend_meta_ const& left, db::output_id right) const noexcept - { - return left.id < right; - } - bool operator()(db::output_id left, db::output::spend_meta_ const& right) const noexcept - { - return left < right.id; - } - }; - return std::lower_bound(metas.begin(), metas.end(), id, by_output_id{}); - } - - bool is_hidden(db::account_status status) noexcept - { - switch (status) - { - case db::account_status::active: - case db::account_status::inactive: - return false; - default: - case db::account_status::hidden: - break; - } - return true; - } - - bool key_check(const rpc::account_credentials& creds) - { - crypto::public_key verify{}; - if (!crypto::secret_key_to_public_key(creds.key, verify)) - return false; - if (verify != creds.address.view_public) - return false; - return true; - } - - //! \return Account info from the DB, iff key matches address AND address is NOT hidden. - expect> open_account(const rpc::account_credentials& creds, db::storage disk) - { - if (!key_check(creds)) - return {lws::error::bad_view_key}; - - auto reader = disk.start_read(); - if (!reader) - return reader.error(); - - const auto user = reader->get_account(creds.address); - if (!user) - return user.error(); - if (is_hidden(user->first)) - return {lws::error::account_not_found}; - return {std::make_pair(user->second, std::move(*reader))}; - } - + bool check_lookahead(connection_data& data, const db::address_index lookahead) { const auto minor = to_uint(lookahead.min_i); @@ -514,7 +469,7 @@ namespace lws static expect handle(const request& req, connection_data& data, std::function&&) { - auto user = open_account(req, data.global->disk.clone()); + auto user = lws::rpc::open_account(req, data.global->disk); if (!user) return user.error(); @@ -553,7 +508,7 @@ namespace lws if (metas.empty() || metas.back().id < meta.id) metas.push_back(meta); else - metas.insert(find_metadata(metas, meta.id), meta); + metas.insert(rpc::get_address_txs_response::find_metadata(metas, meta.id), meta); resp.total_received = rpc::safe_uint64(std::uint64_t(resp.total_received) + meta.amount); if (is_locked(output.get_value(), last->id, output.get_value().height)) @@ -563,7 +518,7 @@ namespace lws resp.spent_outputs.reserve(spends->count()); for (auto const& spend : spends->make_range()) { - const auto meta = find_metadata(metas, spend.source); + const auto meta = rpc::get_address_txs_response::find_metadata(metas, spend.source); if (meta == metas.end() || meta->id != spend.source) { throw std::logic_error{ @@ -591,122 +546,12 @@ namespace lws static expect handle(const request& req, connection_data& data, std::function&&) { - auto user = open_account(req, data.global->disk.clone()); + auto user = lws::rpc::open_account(req, data.global->disk); if (!user) return user.error(); data.passed_login = true; - auto outputs = user->second.get_outputs(user->first.id); - if (!outputs) - return outputs.error(); - - auto spends = user->second.get_spends(user->first.id); - if (!spends) - return spends.error(); - - const expect last = user->second.get_last_block(); - if (!last) - return last.error(); - - response resp{}; - resp.scanned_height = std::uint64_t(user->first.scan_height); - resp.scanned_block_height = resp.scanned_height; - resp.start_height = std::uint64_t(user->first.start_height); - resp.blockchain_height = std::uint64_t(last->id); - resp.transaction_height = resp.blockchain_height; - resp.lookahead_fail = to_uint(user->first.lookahead_fail); - resp.lookahead = user->first.lookahead; - - // merge input and output info into a single set of txes. - - auto output = outputs->make_iterator(); - auto spend = spends->make_iterator(); - - std::vector metas{}; - - resp.transactions.reserve(outputs->count()); - metas.reserve(resp.transactions.capacity()); - - db::transaction_link next_output{}; - db::transaction_link next_spend{}; - - if (!output.is_end()) - next_output = output.get_value(); - if (!spend.is_end()) - next_spend = spend.get_value(); - - while (!output.is_end() || !spend.is_end()) - { - if (!resp.transactions.empty()) - { - db::transaction_link const& last = resp.transactions.back().info.link; - - if ((!output.is_end() && next_output < last) || (!spend.is_end() && next_spend < last)) - { - throw std::logic_error{"DB has unexpected sort order"}; - } - } - - if (spend.is_end() || (!output.is_end() && next_output <= next_spend)) - { - std::uint64_t amount = 0; - if (resp.transactions.empty() || resp.transactions.back().info.link.tx_hash != next_output.tx_hash) - { - resp.transactions.push_back({*output}); - amount = resp.transactions.back().info.spend_meta.amount; - } - else - { - amount = output.get_value(); - resp.transactions.back().info.spend_meta.amount += amount; - } - - const db::output::spend_meta_ meta = output.get_value(); - if (metas.empty() || metas.back().id < meta.id) - metas.push_back(meta); - else - metas.insert(find_metadata(metas, meta.id), meta); - - resp.total_received = rpc::safe_uint64(std::uint64_t(resp.total_received) + amount); - - ++output; - if (!output.is_end()) - next_output = output.get_value(); - } - else if (output.is_end() || (next_spend < next_output)) - { - const db::output_id source_id = spend.get_value(); - const auto meta = find_metadata(metas, source_id); - if (meta == metas.end() || meta->id != source_id) - { - throw std::logic_error{ - "Serious database error, no receive for spend" - }; - } - - if (resp.transactions.empty() || resp.transactions.back().info.link.tx_hash != next_spend.tx_hash) - { - resp.transactions.push_back({}); - resp.transactions.back().spends.push_back({*meta, *spend}); - resp.transactions.back().info.link.height = resp.transactions.back().spends.back().possible_spend.link.height; - resp.transactions.back().info.link.tx_hash = resp.transactions.back().spends.back().possible_spend.link.tx_hash; - resp.transactions.back().info.spend_meta.mixin_count = - resp.transactions.back().spends.back().possible_spend.mixin_count; - resp.transactions.back().info.timestamp = resp.transactions.back().spends.back().possible_spend.timestamp; - resp.transactions.back().info.unlock_time = resp.transactions.back().spends.back().possible_spend.unlock_time; - } - else - resp.transactions.back().spends.push_back({*meta, *spend}); - - resp.transactions.back().spent += meta->amount; - - ++spend; - if (!spend.is_end()) - next_spend = spend.get_value(); - } - } - - return resp; + return response::load(user->second, user->first, false); } }; @@ -1066,7 +911,7 @@ namespace lws static expect handle(request const& req, connection_data& data, std::function&&) { - auto user = open_account(req, data.global->disk.clone()); + auto user = lws::rpc::open_account(req, data.global->disk); if (!user) return user.error(); @@ -1090,7 +935,7 @@ namespace lws if (!rpc) return rpc.error(); - auto user = open_account(req.creds, std::move(disk)); + auto user = lws::rpc::open_account(req.creds, std::move(disk)); if (!user) return user.error(); @@ -1370,7 +1215,7 @@ namespace lws bool fulfilled = false; db::address_index lookahead{}; { - auto user = open_account(req.creds, data.global->disk.clone()); + auto user = lws::rpc::open_account(req.creds, data.global->disk); if (!user) return user.error(); @@ -1453,7 +1298,7 @@ namespace lws static expect handle(request req, connection_data& data, std::function&&) { - if (!key_check(req.creds)) + if (!lws::rpc::key_check(req.creds)) return {lws::error::bad_view_key}; auto disk = data.global->disk.clone(); @@ -1467,7 +1312,7 @@ namespace lws if (account) { - if (is_hidden(account->first)) + if (lws::rpc::is_hidden(account->first)) return {lws::error::account_not_found}; // Do not count a request for account creation as login @@ -1526,7 +1371,7 @@ namespace lws db::account_id id = db::account_id::invalid; { - auto user = open_account(req.creds, data.global->disk.clone()); + auto user = lws::rpc::open_account(req.creds, data.global->disk); if (!user) return user.error(); id = user->first.id; @@ -1789,7 +1634,7 @@ namespace lws db::account_id id = db::account_id::invalid; { - auto user = open_account(req.creds, data.global->disk.clone()); + auto user = lws::rpc::open_account(req.creds, data.global->disk); if (!user) return user.error(); id = user->first.id; @@ -1922,10 +1767,11 @@ namespace lws current_max = std::max(current_max, start->max_size); return current_max; } - + constexpr const endpoint endpoints[] = { {"/daemon_status", call, 1024, true}, + {"/feed", nullptr, 0, false}, {"/get_address_info", call, 2 * 1024, false}, {"/get_address_txs", call, 2 * 1024, false}, {"/get_random_outs", call, 2 * 1024, true}, @@ -1963,6 +1809,18 @@ namespace lws constexpr const unsigned max_endpoint_size = std::max(max_standard_endpoint_size, max_admin_endpoint_size); + constexpr const endpoint* get_endpoint(const std::string_view name) noexcept + { + const endpoint* current = std::begin(endpoints); + endpoint const* const end = std::end(endpoints); + for (; current != end; ++current) + { + if (current->name == name) + break; + } + return current; + } + struct by_name_ { bool operator()(endpoint const& left, endpoint const& right) const noexcept @@ -1985,39 +1843,6 @@ namespace lws } }; constexpr const by_name_ by_name{}; - - struct slice_body - { - using value_type = epee::byte_slice; - - static std::size_t size(const value_type& source) noexcept - { - return source.size(); - } - - struct writer - { - epee::byte_slice body_; - - using const_buffers_type = boost::asio::const_buffer; - - template - explicit writer(boost::beast::http::header const&, value_type const& body) - : body_(body.clone()) - {} - - void init(boost::beast::error_code& ec) - { - ec = {}; - } - - boost::optional> get(boost::beast::error_code& ec) - { - ec = {}; - return {{const_buffers_type{body_.data(), body_.size()}, false}}; - } - }; - }; } // anonymous struct rest_server::internal @@ -2232,12 +2057,29 @@ namespace lws const auto target = self_->parser_.value().get().target(); MDEBUG("Received HTTP request from " << self_.get() << " to target " << target); - + // Checked access for `parser_` on first use endpoint const* const handler = self_->parent_->get_endpoint(target); if (!handler) return self_->bad_request(boost::beast::http::status::not_found, std::forward(resume)); + static constexpr endpoint const* const feed = get_endpoint("/feed"); + if (handler == feed) + { + rest_server_data* const from = self_->data_.global; + const auto feed_timeout = from->options.feed_timeout; + const bool feed_disabled = feed_timeout <= std::chrono::seconds{0}; + 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)) + return self_->bad_request(boost::beast::http::status::bad_request, std::forward(resume)); + return; // else a websocket has I/O queued on `io`. + } + else if (feed_disabled) + return self_->bad_request(boost::beast::http::status::not_found, std::forward(resume)); + } + if (handler->run == nullptr) return self_->bad_request(boost::beast::http::status::not_implemented, std::forward(resume)); @@ -2399,7 +2241,7 @@ namespace lws } rest_server::rest_server(epee::span addresses, std::vector admin, db::storage disk, rpc::client client, configuration config) - : global_(std::make_unique(std::move(disk), std::move(client), runtime_options{config.max_subaddresses, config.webhook_verify, config.disable_admin_auth, config.auto_accept_creation, config.auto_accept_import})), + : global_(std::make_unique(std::move(disk), std::move(client), runtime_options{config})), ports_(), workers_() { diff --git a/src/rest_server.h b/src/rest_server.h index 6b7c6b6..3de0e01 100644 --- a/src/rest_server.h +++ b/src/rest_server.h @@ -60,6 +60,7 @@ namespace lws { epee::net_utils::ssl_authentication_t auth; std::vector access_controls; + std::chrono::seconds feed_timeout; std::size_t threads; std::uint32_t max_subaddresses; epee::net_utils::ssl_verification_t webhook_verify; diff --git a/src/rpc/CMakeLists.txt b/src/rpc/CMakeLists.txt index e869fb7..e9b30f3 100644 --- a/src/rpc/CMakeLists.txt +++ b/src/rpc/CMakeLists.txt @@ -28,9 +28,9 @@ add_subdirectory(scanner) -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) +set(monero-lws-rpc_sources admin.cpp client.cpp daemon_pub.cpp daemon_zmq.cpp feed_ssl.cpp feed_tcp.cpp light_wallet.cpp login.cpp lws_pub.cpp rates.cpp) +set(monero-lws-rpc_headers admin.h client.h daemon_pub.h daemon_zmq.h feed.h fwd.h json.h light_wallet.h login.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}) -target_link_libraries(monero-lws-rpc monero::libraries monero-lws-util monero-lws-wire-json monero-lws-wire-wrapper ${RMQ_LIBRARY}) +target_link_libraries(monero-lws-rpc monero::libraries monero-lws-db monero-lws-util monero-lws-wire-json monero-lws-wire-wrapper ${RMQ_LIBRARY}) diff --git a/src/rpc/client.cpp b/src/rpc/client.cpp index b830e72..e0d0eef 100644 --- a/src/rpc/client.cpp +++ b/src/rpc/client.cpp @@ -35,13 +35,16 @@ #include "common/error.h" // monero/contrib/epee/include #include "db/account.h" +#include "db/data.h" #include "error.h" #include "misc_log_ex.h" // monero/contrib/epee/include #include "net/http_client.h" // monero/contrib/epee/include #include "net/zmq.h" // monero/src #include "net/zmq_async.h" +#include "rpc/light_wallet.h" #include "scanner.h" #include "serialization/json_object.h" // monero/src +#include "wire/msgpack/write.h" #include "wire/json/write.h" #if MLWS_RMQ_ENABLED #include @@ -66,7 +69,7 @@ namespace rpc namespace { constexpr const char signal_endpoint[] = "inproc://signal"; - constexpr const char account_endpoint[] = "inproc://account"; // append integer every new `account_push` + constexpr const char account_endpoint[] = "inproc://account"; constexpr const char abort_scan_signal[] = "SCAN"; constexpr const char abort_process_signal[] = "PROCESS"; constexpr const char minimal_chain_topic[] = "json-minimal-chain_main"; @@ -268,10 +271,11 @@ namespace rpc { struct context { - explicit context(zcontext comm, net::zmq::socket signal_pub, net::zmq::socket external_pub, rcontext rmq_src, std::string daemon_addr, std::string sub_addr, std::chrono::minutes interval, bool untrusted_daemon) + explicit context(zcontext comm, net::zmq::socket signal_pub, net::zmq::socket external_pub, net::zmq::socket internal_pub, rcontext rmq_src, std::string daemon_addr, std::string sub_addr, std::chrono::minutes interval, bool untrusted_daemon) : comm(std::move(comm)) , signal_pub(std::move(signal_pub)) , external_pub(std::move(external_pub)) + , internal_pub(std::move(internal_pub)) , rmq(std::move(rmq_src)) , daemon_addr(std::move(daemon_addr)) , sub_addr(std::move(sub_addr)) @@ -279,8 +283,10 @@ namespace rpc , cache_time() , cache_interval(interval) , cached{} + , warning_count(0) , sync_pub() , sync_rates() + , sync_internal_pub() , untrusted_daemon(untrusted_daemon) , rates_running() { @@ -292,6 +298,7 @@ namespace rpc zcontext comm; net::zmq::socket signal_pub; net::zmq::socket external_pub; + const net::zmq::socket internal_pub; rcontext rmq; const std::string daemon_addr; const std::string sub_addr; @@ -299,8 +306,10 @@ namespace rpc std::chrono::steady_clock::time_point cache_time; const std::chrono::minutes cache_interval; rates cached; + std::atomic warning_count; boost::mutex sync_pub; boost::mutex sync_rates; + boost::mutex sync_internal_pub; const bool untrusted_daemon; std::atomic_flag rates_running; }; @@ -328,6 +337,18 @@ namespace rpc return {lws::error::bad_daemon_response}; } + expect account_sub::watch(const std::string_view address) const + { + MONERO_PRECOND(ctx); + MONERO_PRECOND(sub.zsock); + + const std::string_view warning = feed_warning::prefix(); + const std::uint32_t count = ctx->warning_count; + MONERO_ZMQ_CHECK(zmq_setsockopt(sub.zsock.get(), ZMQ_SUBSCRIBE, warning.data(), warning.size())); + MONERO_ZMQ_CHECK(zmq_setsockopt(sub.zsock.get(), ZMQ_SUBSCRIBE, address.data(), address.size())); + return count; + } + expect client::make_daemon(const std::shared_ptr& ctx) noexcept { assert(ctx != nullptr); @@ -481,6 +502,19 @@ namespace rpc return {lws::error::bad_daemon_response}; } + expect client::make_account_sub(boost::asio::io_context& io) const + { + MONERO_PRECOND(ctx != nullptr); + net::zmq::socket sub{zmq_socket(ctx->comm.get(), ZMQ_SUB)}; + if (!sub) + return {net::zmq::get_error_code()}; + MONERO_ZMQ_CHECK(zmq_connect(sub.get(), account_endpoint)); + auto client = net::zmq::async_client::make(io, std::move(sub)); + if (!client) + return client.error(); + return account_sub{ctx, std::move(*client)}; + } + expect client::make_async_client(boost::asio::io_context& io) const { MONERO_PRECOND(ctx != nullptr); @@ -558,6 +592,114 @@ namespace rpc return rc; } + namespace + { + template + expect prep_message(epee::byte_stream& sink, const std::string_view prefix, const T& message) + { + sink.write({prefix.data(), prefix.size()}); + if (prefix.empty() || prefix.back() != u8':') + sink.push_back(':'); + + wire::msgpack_slice_writer dest{std::move(sink), true /* integer keys */}; + try + { + wire_write::bytes(dest, message); + } + catch (const wire::exception& e) + { + return {e.code()}; + } + + sink = dest.take_sink(); + return success(); + } + + template + expect prep_message(const std::string_view prefix, const T& message) + { + epee::byte_stream sink; + MONERO_CHECK(prep_message(sink, prefix, message)); + return epee::byte_slice{std::move(sink)}; + } + } + + expect client::push_update(const account& acct, const mempool_receive& src) const + { + if (ctx->internal_pub) + { + auto msg = prep_message(acct.address(), src); + if (!msg) + return msg.error(); + const boost::lock_guard lock{ctx->sync_internal_pub}; + net::zmq::send(std::move(*msg), ctx->internal_pub.get(), ZMQ_DONTWAIT); + } + return success(); + } + + expect client::push_updates(const epee::span accts, const db::block_id new_height) const + { + MONERO_PRECOND(ctx != nullptr); + if (!ctx->internal_pub) + { + for (account& acct : accts) + acct.updated(new_height); + return success(); + } + + std::vector offsets; + offsets.reserve(accts.size()); + epee::byte_stream sink; + + std::size_t last = 0; + expect rc = success(); + for (account& acct : accts) + { + if (new_height < acct.scan_height()) + continue; + + // make sure `updated` is called on every account even with errors + expect msg = prep_message(sink, acct.address(), acct.updated(new_height)); + if (!msg) + rc = msg; + + offsets.push_back(sink.size() - last); + last = sink.size(); + } + + if (!rc) + return rc; + + epee::byte_slice msgs{std::move(sink)}; + const boost::lock_guard lock{ctx->sync_internal_pub}; + for (std::size_t offset : offsets) + net::zmq::send(msgs.take_slice(offset), ctx->internal_pub.get(), ZMQ_DONTWAIT); + return success(); + } + + namespace + { + expect push_warning_(const std::shared_ptr& ctx, const std::error_code error, const db::block_id height) + { + MONERO_PRECOND(ctx != nullptr); + if (ctx->internal_pub) + { + const boost::lock_guard lock{ctx->sync_internal_pub}; + expect msg = + prep_message(rpc::feed_warning::prefix(), rpc::feed_warning{error, ctx->warning_count++, height}); + if (!msg) + return msg.error(); + return net::zmq::send(std::move(*msg), ctx->internal_pub.get(), ZMQ_DONTWAIT); + } + return success(); + } + } + + expect client::push_warning(const std::error_code error, const db::block_id height) const + { + return push_warning_(ctx, error, height); + } + expect client::get_rates() const { MONERO_PRECOND(ctx != nullptr); @@ -571,7 +713,7 @@ namespace rpc return ctx->cached; } - context context::make(std::string daemon_addr, std::string sub_addr, std::string pub_addr, rmq_details rmq_info, std::chrono::minutes rates_interval, const bool untrusted_daemon) + context context::make(std::string daemon_addr, std::string sub_addr, std::string pub_addr, rmq_details rmq_info, std::chrono::minutes rates_interval, const bool untrusted_daemon, const bool push_updates) { zcontext comm{zmq_init(1)}; if (comm == nullptr) @@ -597,10 +739,20 @@ namespace rpc if (!rmq_info.address.empty() || !rmq_info.exchange.empty() || !rmq_info.routing.empty() || !rmq_info.credentials.empty()) MONERO_THROW(error::configuration, "RabbitMQ support not enabled"); #endif + + net::zmq::socket internal_pub; + if (push_updates) + { + internal_pub.reset(zmq_socket(comm.get(), ZMQ_PUB)); + if (internal_pub == nullptr) + MONERO_THROW(net::zmq::get_error_code(), "zmq_socket"); + if (zmq_bind(internal_pub.get(), account_endpoint) < 0) + MONERO_THROW(net::zmq::get_error_code(), "zmq_bind"); + } return context{ std::make_shared( - std::move(comm), std::move(pub), std::move(external_pub), rcontext{std::move(rmq_info)}, std::move(daemon_addr), std::move(sub_addr), rates_interval, untrusted_daemon + std::move(comm), std::move(pub), std::move(external_pub), std::move(internal_pub), rcontext{std::move(rmq_info)}, std::move(daemon_addr), std::move(sub_addr), rates_interval, untrusted_daemon ) }; } @@ -632,6 +784,11 @@ namespace rpc return ctx->cache_interval; } + expect context::push_warning(const std::error_code error) const + { + return push_warning_(ctx, error, db::block_id::txpool); + } + expect context::raise_abort_scan() noexcept { MONERO_PRECOND(ctx != nullptr); diff --git a/src/rpc/client.h b/src/rpc/client.h index 719f1b8..667c520 100644 --- a/src/rpc/client.h +++ b/src/rpc/client.h @@ -38,6 +38,8 @@ #include "db/fwd.h" #include "common/expect.h" // monero/src #include "net/zmq.h" // monero/src +#include "net/zmq_async.h" +#include "rpc/fwd.h" #include "rpc/message.h" // monero/src #include "rpc/daemon_pub.h" #include "rpc/rates.h" @@ -63,6 +65,37 @@ namespace rpc expect parse_response(cryptonote::rpc::Message& parser, std::string msg, source_location loc = {}); + class account_sub + { + std::shared_ptr ctx; + net::zmq::async_client sub; + + public: + account_sub(std::shared_ptr ctx, net::zmq::async_client sub) + : ctx(std::move(ctx)), sub(std::move(sub)) + {} + + account_sub(account_sub&&) = default; + account_sub(const account_sub&) = delete; + ~account_sub() = default; + account_sub& operator=(account_sub&&) = default; + account_sub& operator=(const account_sub&) = delete; + + explicit operator bool() const noexcept { return !sub.close; } + net::zmq::async_client& get() noexcept { return sub; } + + //! Subscribe to events from `address` and "warning:" + expect watch(const std::string_view address) const; + + //! Initiate complete shutdown on this sub + void shutdown(boost::system::error_code& ec) + { + sub.close = true; + if (sub.asock) + sub.asock->cancel(ec); + } + }; + //! Abstraction for ZMQ RPC client. All `const` and `static` methods are thread-safe. class client { @@ -138,6 +171,9 @@ namespace rpc return cryptonote::rpc::FullMessage::getRequest(name, message, 0); } + //! \return An async ZMQ_SUB socket that is bound to the account update publisher + expect make_account_sub(boost::asio::io_context& io) const; + //! \return `async_client` to daemon. Thread safe. expect make_async_client(boost::asio::io_context& io) const; @@ -163,6 +199,10 @@ namespace rpc return publish(epee::byte_slice{std::move(bytes)}); } + expect push_update(const account& acct, const mempool_receive& src) const; + expect push_updates(epee::span accts, db::block_id new_height) const; + expect push_warning(std::error_code error, db::block_id height) const; + //! \return Next available RPC message response from server expect get_message(std::chrono::seconds timeout); @@ -179,7 +219,7 @@ namespace rpc \return Recent exchange rates. */ expect get_rates() const; }; - + //! Owns ZMQ context, and ZMQ PUB socket for signalling child `client`s. class context { @@ -205,7 +245,7 @@ namespace rpc \param True if additional size constraints should be placed on daemon messages */ - static context make(std::string daemon_addr, std::string sub_addr, std::string pub_addr, rmq_details rmq_info, std::chrono::minutes rates_interval, const bool untrusted_daemon); + static context make(std::string daemon_addr, std::string sub_addr, std::string pub_addr, rmq_details rmq_info, std::chrono::minutes rates_interval, const bool untrusted_daemon, bool push_updates); context(context&&) = default; context(context const&) = delete; @@ -233,6 +273,9 @@ namespace rpc return client::make(ctx); } + //! Push warning to all websocket push clients. Thread safe + expect push_warning(std::error_code error) const; + /*! All block `client::send`, `client::receive`, and `client::wait` calls originating from `this` object AND whose `watch_scan_signal` method was diff --git a/src/rpc/feed.h b/src/rpc/feed.h new file mode 100644 index 0000000..6bc7d65 --- /dev/null +++ b/src/rpc/feed.h @@ -0,0 +1,85 @@ +// Copyright (c) 2026, The Monero Project +// All rights reserved. +// +// Redistribution and use in source and binary forms, with or without modification, are +// permitted provided that the following conditions are met: +// +// 1. Redistributions of source code must retain the above copyright notice, this list of +// conditions and the following disclaimer. +// +// 2. Redistributions in binary form must reproduce the above copyright notice, this list +// of conditions and the following disclaimer in the documentation and/or other +// materials provided with the distribution. +// +// 3. Neither the name of the copyright holder nor the names of its contributors may be +// used to endorse or promote products derived from this software without specific +// prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY +// EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF +// MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL +// THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, +// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, +// STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF +// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include + +#include "db/fwd.h" +#include "rpc/fwd.h" +#include "wire/fwd.h" + +namespace lws { namespace rpc { namespace feed +{ + using request = + boost::beast::http::request; + + enum class error { none = 0, missed_messages, protocol, queue_max }; + + // Do not change order or value as these are exported to clients (append) + enum class status : std::uint16_t + { + unspecified_error = 0, + account_not_found, + bad_address, + bad_view_key, + blockchain_reorg, + daemon_unresponsive, + parse_error, + protocol_error, + queue_error, + schema_error + }; + WIRE_AS_INTEGER(status); + + status map(std::error_code error) noexcept; + + const char* get_string(feed::error value) noexcept; + std::error_category const& error_category() noexcept; + inline std::error_code make_error_code(feed::error value) noexcept + { + 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&& sock, boost::asio::io_context& io, const lws::rpc::client& client, const request& req, const lws::db::storage& disk, std::chrono::seconds idle_timeout); +}}} // lws // rpc // feed + +namespace std +{ + template<> + struct is_error_code_enum<::lws::rpc::feed::error> + : true_type + {}; +} diff --git a/src/rpc/feed.inl b/src/rpc/feed.inl new file mode 100644 index 0000000..9673092 --- /dev/null +++ b/src/rpc/feed.inl @@ -0,0 +1,600 @@ +// 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. + +/* This file was split into two compilation units `feed_tcp.cpp` and + `feed_ssl.cpp`. This helps reduce the compilation time since each file can be + computed in parallel. The memory usage is also reduced when -j1, which helps + with compilation on smaller machines. + + The functions must be in anonymous namespace or marked inline (implicitly or + explicitly) or linker will have issues. Inline is preferred because the + linker will merge the duplicate copies. */ + +#include "feed.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "db/storage.h" +#include "error.h" +#include "misc_log_ex.h" +#include "net/http/slice_body.h" +#include "net/zmq_async.h" +#include "rpc/client.h" + +namespace lws { namespace rpc { namespace feed +{ + constexpr const std::size_t max_read_size = 2 * 1024; // login call is small + constexpr const std::size_t queue_max = 100; // # messages in queue + + //! Suppress scanner messages in this interval when account is catching up + constexpr const std::chrono::seconds scanner_busy{10}; + + //! Send all block updates if account is within this top blockchain height + constexpr const unsigned on_tip = 10; + + constexpr const std::chrono::seconds close_timeout{8}; // for websocket close + constexpr const std::chrono::seconds handshake_timeout{5}; + constexpr const std::chrono::seconds login_timeout{8}; + constexpr const std::chrono::seconds shutdown_timeout{8}; // for ssl shutdown + constexpr const std::chrono::seconds tx_sync_timeout{20}; // for initial tx sync + + enum class event { none = 0, invalid, shutdown, timeout}; + enum class protocol : unsigned { invalid = 0, v0_json, v0_msgpack }; + + inline constexpr const char* get_string(const protocol proto) noexcept + { + switch (proto) + { + case protocol::v0_json: + return "lws.feed.v0.json"; + case protocol::v0_msgpack: + return "lws.feed.v0.msgpack"; + default: + break; + } + + return "unsupported"; + } + + // defined in `feed_tcp.cpp. Selects preferred protocol + protocol get_protocol(const boost::beast::string_view list) noexcept; + + inline bool is_binary(const protocol proto) noexcept + { + switch (proto) + { + case protocol::v0_msgpack: + return true; + default: + break; + } + return false; + } +}}} // lws // rpc // feed + +namespace boost +{ + namespace system + { + template <> + struct is_error_code_enum + : std::true_type + {}; + } // namespace system +} // namespace boost + +namespace lws { namespace rpc { namespace feed +{ + + // defined in `feed_tcp.cpp + const boost::system::error_category& get_event_category(); + inline boost::system::error_code make_error_code(event e) + { + return {static_cast(e), get_event_category()}; + } + + class connection; + + struct connection_sync + { + db::block_id height; + db::block_id last_reported; + std::chrono::steady_clock::time_point last_update; + db::account_id id; + std::uint32_t warnings; + bool on_tip; + bool receives_only; + + connection_sync() noexcept + : height(db::block_id::txpool), + last_reported(db::block_id::txpool), + last_update(), + id(db::account_id::invalid), + warnings(-1), + on_tip(false), + receives_only(false) + {} + }; + + //! Handles initial login, tx sync, and then waits on zmq messages, pushing them to a queue + struct zmq_loop : boost::asio::coroutine + { + std::shared_ptr self_; + std::size_t iteration_; + + explicit zmq_loop(std::shared_ptr self) + : boost::asio::coroutine(), self_(std::move(self)), iteration_(0) + {} + + void operator()(boost::system::error_code error = {}, std::size_t = {}); + }; + + // defined in `feed_tcp.cpp` + std::string_view get_prefix(const boost::beast::net::const_buffer buf); + expect prep_error(std::error_code error, protocol proto); + expect prep_login(const db::storage& disk, account_sub* sub, connection_sync* sync, boost::beast::flat_buffer& buffer, protocol proto); + expect prep_update(const db::storage& disk, std::string&& source, connection_sync* sync, protocol proto); + + class connection + { + lws::db::storage disk_; + std::string sub_buffer_; + boost::beast::flat_buffer ws_buffer_; + std::deque write_queue_; + account_sub sub_; + boost::asio::io_context::strand strand_; + boost::asio::steady_timer socket_timer_; + std::size_t pull_iter_; + connection_sync sync_; + const protocol proto_; + + virtual void async_shutdown(std::shared_ptr self) = 0; + + protected: + + template + void do_async_close(T& sock, std::shared_ptr self) + { + LWS_VERIFY(self); + MDEBUG("Starting websocket close on " << this); + boost::system::error_code ec{}; + sub_.shutdown(ec); + start_timeout(close_timeout, self); + sock.async_close(boost::beast::websocket::close_reason(), bind(std::bind(&connection::async_shutdown, self, self))); + } + + void do_async_shutdown(boost::asio::ip::tcp::socket& sock, std::shared_ptr) + { + do_cleanup(sock); + } + + void do_async_shutdown(boost::asio::ssl::stream& sock, std::shared_ptr self) + { + LWS_VERIFY(self); + MDEBUG("Starting TLS/SSL shutdown on " << this); + boost::system::error_code ec{}; + sub_.shutdown(ec); + start_timeout(shutdown_timeout, self); + sock.async_shutdown(bind(std::bind(&connection::cleanup, self))); + } + + void do_cleanup(boost::asio::ip::tcp::socket& sock) + { + MDEBUG("Websocket cleanup on " << this); + boost::system::error_code ec{}; + sub_.shutdown(ec); + socket_timer_.cancel(); + sock.shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec); + sock.close(ec); + } + + void do_cleanup(boost::asio::ssl::stream& sock) + { + do_cleanup(sock.next_layer()); + } + + template + void do_stop_reads(T& sock, std::shared_ptr self) + { + LWS_VERIFY(self); + ws_buffer_.shrink_to_fit(); + sock.async_read(ws_buffer_, bind(std::bind(&connection::close_check, self, self, event::invalid))); + } + + template + void do_async_read(T& sock, boost::asio::steady_timer::duration timeout, zmq_loop callback) + { + if (is_closing()) + return callback(event::shutdown); + + LWS_VERIFY(callback.self_); + start_timeout(timeout, callback); + sock.async_read(ws_buffer_, bind(std::move(callback))); + } + + template + static void do_async_write_queue(T& sock, std::shared_ptr self, epee::byte_slice&& first) + { + // do not check `is_closing()`. Instead, try to write out all buffered data + LWS_VERIFY(self && self->write_queue_.empty()); + + struct push_loop + { + T* sock_; + std::shared_ptr self_; + + void async_write() + { + LWS_VERIFY(sock_ && self_ && !self_->write_queue_.empty()); + connection& self = *self_; + + const epee::byte_slice& next = self.write_queue_.front(); + MDEBUG("Sending websocket " << get_prefix({next.data(), next.size()}) << " message on " << self_.get()); + sock_->async_write( + boost::asio::buffer(next.data(), next.size()), self.bind(std::move(*this)) + ); + } + + void operator()(const boost::system::error_code error, std::size_t) + { + LWS_VERIFY(self_ && !self_->write_queue_.empty()); + connection& self = *self_; + + if (error) + { + MDEBUG("Write error on websocket (" << self_.get() << "): " << error.message()); + return self.async_close(std::move(self_)); + } + + self.write_queue_.pop_front(); + if (!self.write_queue_.empty()) + async_write(); + else if (self.is_closing()) + return self.async_close(std::move(self_)); + } + }; + + self->write_queue_.push_back(std::move(first)); + push_loop{std::addressof(sock), std::move(self)}.async_write(); + } + + bool do_write(std::shared_ptr self, expect&& message) + { + const bool rc = message.has_value(); + if (!rc) + { + MDEBUG("Error in websocket (" << this << "): " << message.error().message()); + message = MONERO_UNWRAP(prep_error(message.error(), proto_)); + } + else if (message->empty()) + return true; // nop update (scanner is _very_ active) + + if (write_queue_.empty()) + async_write_queue(std::move(self), std::move(*message)); + else + write_queue_.push_back(std::move(*message)); + return rc; + } + + static bool sync_iter(std::size_t& ours, const std::size_t theirs) noexcept + { + if (ours != theirs) + return false; + ++ours; + return true; + } + + public: + connection(boost::asio::io_context& io, const lws::rpc::client& client, const lws::db::storage& disk, const protocol proto) + : disk_(disk.clone()), + sub_buffer_(), + ws_buffer_(), + write_queue_(), + sub_(MONERO_UNWRAP(client.make_account_sub(io))), + strand_(io), + socket_timer_(io), + pull_iter_(0), + sync_{}, + proto_(proto) + { + ws_buffer_.max_size(max_read_size); + } + + virtual ~connection() noexcept { MDEBUG("Destroying websocket on " << this); } + + protocol proto() const noexcept { return proto_; } + bool is_closing() const noexcept { return !bool(sub_); } + bool sync_pull(const std::size_t iter) noexcept { return sync_iter(pull_iter_, iter); } + + bool login(std::shared_ptr self) + { + const bool rc = do_write(std::move(self), prep_login(disk_, &sub_, &sync_, ws_buffer_, proto_)); + ws_buffer_.consume(ws_buffer_.cdata().size()); + return rc; + } + + // Push message from ZMQ to websocket (maybe) + bool push(std::shared_ptr self) + { + bool rc = false; + if (queue_max < write_queue_.size()) + { + MDEBUG("Websocket max queue reached on " << this); + write_queue_.push_back(MONERO_UNWRAP(prep_error({error::queue_max}, proto_))); + } + else + { + MDEBUG("Websocket queued event from scanner on " << this); + rc = do_write(std::move(self), prep_update(disk_, std::move(sub_buffer_), &sync_, proto_)); + } + return rc; + } + + template + auto bind(F&& f) -> decltype(boost::asio::bind_executor(strand_, std::forward(f))) + { + return boost::asio::bind_executor(strand_, std::forward(f)); + } + + void client_close(const boost::system::error_code error = {}) + { + if (error) + MDEBUG("Preparing websocket close (" << this << "): " << error.message()); + else + MDEBUG("Preparing websocket close on " << this); + boost::system::error_code ec{}; + socket_timer_.cancel(); + sub_.shutdown(ec); + // writes may need to be flushed, so don't cancel socket ops + } + + void close_check(std::shared_ptr self, const boost::system::error_code error = {}) + { + if (write_queue_.empty()) + async_close(std::move(self)); + else + client_close(error); + } + + void start_timeout(boost::asio::steady_timer::duration timeout, std::shared_ptr self) + { + struct force_close + { + std::shared_ptr self_; + + void operator()(const boost::system::error_code error) const + { + if (error != boost::asio::error::operation_aborted) + { + LWS_VERIFY(self_); + MDEBUG("Websocket timeout on " << self_.get()); + self_->cleanup(); + } + } + }; + + LWS_VERIFY(self); + socket_timer_.expires_after(timeout); + socket_timer_.async_wait(bind(force_close{std::move(self)})); + } + + void start_timeout(boost::asio::steady_timer::duration timeout, zmq_loop handler) + { + struct notify + { + zmq_loop f_; + + void operator()(const boost::system::error_code error) + { + if (!error) + f_(event::timeout); + else if (error != boost::asio::error::operation_aborted) + f_(error); + // else another timer was queued via expires_after + } + }; + + LWS_VERIFY(handler.self_); + socket_timer_.expires_after(timeout); + socket_timer_.async_wait(bind(notify{std::move(handler)})); + } + + void async_zmq_wait(zmq_loop callback) + { + if (is_closing()) + return callback(event::shutdown); + + // Skip timeout + LWS_VERIFY(callback.self_); + net::zmq::async_read(sub_.get(), sub_buffer_, bind(std::move(callback))); + } + + //! Start socket close routine. `loop::` must be exited after this. + virtual void async_close(std::shared_ptr self) = 0; + virtual void cleanup() = 0; + virtual void stop_reads(std::shared_ptr self) = 0; + virtual void async_read(boost::asio::steady_timer::duration timeout, zmq_loop callback) = 0; + virtual void async_write_queue(std::shared_ptr self, epee::byte_slice&& msg) = 0; + }; + + template + class connection_ final : public connection + { + boost::beast::websocket::stream sock_; + + virtual void async_shutdown(std::shared_ptr self) override final + { 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)) + { + if (is_binary(proto)) + sock_.binary(true); + } + + boost::beast::websocket::stream& sock() { return sock_; } + + virtual void async_close(std::shared_ptr self) override final + { do_async_close(sock(), std::move(self)); } + + virtual void cleanup() override final { do_cleanup(sock().next_layer()); } + + virtual void stop_reads(std::shared_ptr self) override final + { do_stop_reads(sock(), std::move(self)); } + + virtual void async_read(boost::asio::steady_timer::duration timeout, zmq_loop callback) override final + { do_async_read(sock(), timeout, std::move(callback)); } + + virtual void async_write_queue(std::shared_ptr self, epee::byte_slice&& msg) override final + { do_async_write_queue(sock(), std::move(self), std::move(msg)); } + }; + + + inline void zmq_loop::operator()(const boost::system::error_code error, const std::size_t bytes) + { + LWS_VERIFY(self_); + connection& self = *self_; + if (!self.sync_pull(iteration_)) // sync all copies of loop + return; + ++iteration_; + + BOOST_ASIO_CORO_REENTER(*this) + { + if (!error) // handshake + { + BOOST_ASIO_CORO_YIELD self.async_read(login_timeout, std::move(*this)); + if (!error) + { + self.stop_reads(self_); // client can only send control frames now + if (!self.login(self_)) // always queues a write + return self.client_close(); + } + } + + if (error) + { + MDEBUG("Websocket failure (" << self_.get() << "): " << error.message()); + return self.async_close(std::move(self_)); // login never queued write + } + + while (true) + { + BOOST_ASIO_CORO_YIELD self.async_zmq_wait(std::move(*this)); + if (error || !self.push(self_)) + return self.close_check(self_, error); + } + } + } + + inline const boost::asio::ip::tcp::socket& get_tcp_sock(const boost::asio::ip::tcp::socket& sock) noexcept { return sock; } + inline const boost::asio::ip::tcp::socket& get_tcp_sock(const boost::asio::ssl::stream& sock) { return sock.next_layer(); } + + template + inline void do_start(std::shared_ptr> self, const request& req, const std::chrono::seconds timeout) + { + struct on_control + { + std::weak_ptr self_; + void operator()(boost::beast::websocket::frame_type type, boost::beast::string_view) + { + switch (type) + { + default: + case boost::beast::websocket::frame_type::close: + { + const auto self = self_.lock(); + if (!self) + return; + MDEBUG("Websocket client requested close on " << self.get()); + self->close_check(self); + break; + } + case boost::beast::websocket::frame_type::ping: + case boost::beast::websocket::frame_type::pong: + MDEBUG("Received control frame (ping/pong) from " << self_.lock().get()); + break; + } + } + }; + + LWS_VERIFY(self); + connection_& obj = *self; + + boost::system::error_code ec{}; + MDEBUG("Starting websocket handshake on " << get_tcp_sock(self->sock().next_layer()).remote_endpoint(ec) << " / " << self.get()); + obj.sock().control_callback(on_control{self}); + { + namespace http = boost::beast::http; + using stream = boost::beast::websocket::stream_base; + obj.sock().set_option(stream::timeout{handshake_timeout, timeout, true /* pings */}); + + const protocol proto = obj.proto(); + obj.sock().set_option( + stream::decorator( + [proto] (http::response_header<>& hdr) + { hdr.set(http::field::sec_websocket_protocol, get_string(proto)); } + ) + ); + } + + obj.sock().async_accept(req, obj.bind(zmq_loop{std::move(self)})); + } + + template + 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) + { + 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>(std::forward(sock), io, client, disk, proto), req, timeout); + return true; + default: + case protocol::invalid: + break; + } + return false; + } +}}} // lws // rpc // feed + diff --git a/src/rpc/feed_ssl.cpp b/src/rpc/feed_ssl.cpp new file mode 100644 index 0000000..cb44251 --- /dev/null +++ b/src/rpc/feed_ssl.cpp @@ -0,0 +1,44 @@ +// 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 +#include +#include "feed.inl" + +namespace lws { namespace rpc { namespace feed +{ + bool start(boost::asio::ssl::stream&& sock, boost::asio::io_context& io, const lws::rpc::client& client, 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); +#else + return false; +#endif + } +}}} + diff --git a/src/rpc/feed_tcp.cpp b/src/rpc/feed_tcp.cpp new file mode 100644 index 0000000..06b6550 --- /dev/null +++ b/src/rpc/feed_tcp.cpp @@ -0,0 +1,374 @@ +// 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 "feed.inl" + +#include + +#include "db/account.h" +#include "db/string.h" +#include "error.h" +#include "rpc/light_wallet.h" +#include "rpc/login.h" +#include "wire/error.h" +#include "wire/json.h" +#include "wire/msgpack.h" + +namespace lws { namespace rpc { namespace feed +{ + status map(const std::error_code error) noexcept + { + if (error == lws::error::blockchain_reorg) + return status::blockchain_reorg; + if (error == lws::error::daemon_timeout) + return status::daemon_unresponsive; + if (error == error::queue_max) + return status::queue_error; + if (error == error::missed_messages) + return status::queue_error; + if (error == lws::error::bad_address) + return status::bad_address; + if (error == lws::error::bad_view_key) + return status::bad_view_key; + if (error == lws::error::account_not_found) + return status::account_not_found; + if (error == error::protocol) + return status::protocol_error; + + const std::error_category& category = error.category(); + if (category == wire::error::schema_category()) + return status::schema_error; + if (category == wire::error::msgpack_category()) + return status::parse_error; + if (category == wire::error::rapidjson_category()) + return status::parse_error; + + return status::unspecified_error; + } + + const char* get_string(error value) noexcept + { + switch(value) + { + case error::missed_messages: + return "Internal client thread missed internal ZeroMQ messages"; + case error::protocol: + return "Client did not follow protocool"; + case error::queue_max: + return "Max queue for outgoing push messages was reached"; + default: + break; + } + return "unknown lws::rpc::feed::error"; + } + + namespace + { + class event_category : public boost::system::error_category + { + public: + // Return a short descriptive name for the category + virtual const char *name() const noexcept override final { return "lws::rpc::push::event"; } + // Return what each enum means in text + virtual std::string message(int c) const override final + { + switch(event(c)) + { + case event::none: + return "No error"; + case event::invalid: + return "Client did not follow protocol"; + case event::timeout: + return "Timeout on async operation"; + default: + break; + } + return "unknown"; + } + virtual boost::system::error_condition default_error_condition(int c) const noexcept override final + { + switch(event(c)) + { + case event::timeout: + return make_error_condition(boost::system::errc::timed_out); + default: + break; + } + return boost::system::error_condition(c, *this); + } + }; + + struct category final : std::error_category + { + virtual const char* name() const noexcept override final + { + return "lws::rpc::push:error_category()"; + } + + virtual std::string message(int value) const override final + { + return get_string(lws::rpc::feed::error(value)); + } + + virtual std::error_condition default_error_condition(int value) const noexcept override final + { + switch (lws::rpc::feed::error(value)) + { + case error::protocol: + return std::errc::protocol_error; + case error::queue_max: + return std::errc::no_buffer_space; + default: + break; // map to unmatchable category + } + return std::error_condition{value, *this}; + } + }; + } // anonymous + + const boost::system::error_category& get_event_category() + { + static event_category c; + return c; + } + + + std::error_category const& error_category() noexcept + { + static const category instance{}; + return instance; + } + + namespace + { + constexpr std::pair get_map(const protocol proto) noexcept + { + return {get_string(proto), proto}; + } + + std::string get_string(const boost::beast::net::const_buffer buf) + { + return {reinterpret_cast(buf.data()), buf.size()}; + } + + epee::byte_slice get_slice(const boost::beast::net::const_buffer buf) + { + return epee::byte_slice{{reinterpret_cast(buf.data()), buf.size()}}; + } + + template + expect prep(const T& src, const protocol proto) + { + epee::byte_stream sink{}; + + const std::string_view prefix = T::prefix(); + sink.write({prefix.data(), prefix.size()}); + + const std::error_code error = proto == protocol::v0_msgpack ? + wire::msgpack::to_bytes(sink, src) : wire::json::to_bytes(sink, src); + if (error) + return error; + return epee::byte_slice{std::move(sink)}; + } + } // anonymous + + protocol get_protocol(const boost::beast::string_view list) noexcept + { + // in order of preference + static constexpr const std::array, 2> supported + {{ + get_map(protocol::v0_msgpack), + get_map(protocol::v0_json) + }}; + const boost::beast::http::token_list tokens{list}; + for (const auto token : tokens) + { + for (const auto support : supported) + { + if (support.first == token) + return support.second; + } + } + return protocol::invalid; + } + + + std::string_view get_prefix(const boost::beast::net::const_buffer buf) + { + char const* const begin = reinterpret_cast(buf.data()); + void const* const match = std::memchr(begin, u8':', buf.size()); + if (match) + return {begin, std::size_t(reinterpret_cast(match) - begin + 1)}; + return {}; + } + + expect prep_error(const std::error_code error, const protocol proto) + { + return prep(feed_error{error}, proto); + } + + expect prep_login(const db::storage& disk, 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()); + if (prefix != feed_login::prefix()) + return {error::protocol}; + + buffer.consume(prefix.size()); + feed_login login{}; + std::error_code error = proto == protocol::v0_msgpack ? + wire::msgpack::from_bytes(get_slice(buffer.cdata()), login) : + wire::json::from_bytes(get_string(buffer.cdata()), login); + if (error) + return error; + + /* Must sub to feed before starting db read or else lost txes could + occur. If the client gets txes twice, it should merge via hash. */ + + const expect watcher = sub->watch(db::address_string(login.account.address)); + if (!watcher) + return watcher.error(); + + auto account = open_account(login.account, disk); + if (!account) + return account.error(); + + sync->receives_only = login.receives_only; + sync->id = account->first.id; + sync->warnings = *watcher; + if (!login.tx_sync) + { + const auto block = account->second.get_last_block(); + if (!block) + return block.error(); + + account->second.finish_read(); + const db::account& acct = account->first; + 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); + if (!txs) + return txs.error(); + + account->second.finish_read(); + return prep(feed_tx_sync{std::move(*txs)}, proto); + } + + //! Convert ZMQ_PUSH message from the scanner to websocket `update` message + expect prep_update(const db::storage& disk, std::string&& source, connection_sync* const sync, const protocol proto) + { + LWS_VERIFY(sync); + + epee::byte_slice slice{std::move(source)}; + const std::string_view prefix = get_prefix({slice.data(), slice.size()}); + const bool warning = prefix == feed_warning::prefix(); + slice.remove_prefix(prefix.size()); + if (warning) + { + if (sync->receives_only) + return epee::byte_slice{}; + + feed_warning the_warning{}; + const std::error_code error = wire::msgpack::from_bytes(std::move(slice), the_warning); + if (!error) + return error; + + if (sync->warnings != the_warning.counter) + return prep_error(error::missed_messages, proto); + + ++sync->warnings; + if (the_warning.code == status::blockchain_reorg && sync->height != db::block_id::txpool) + sync->height = std::min(sync->height, the_warning.height); + + return prep(the_warning, proto); + } + + account::update from_scanner{}; + const std::error_code error = wire::msgpack::from_bytes(std::move(slice), from_scanner); + if (error) + return error; + + if (from_scanner.new_height != db::block_id::txpool) + { + if (sync->height != db::block_id::txpool && sync->height != from_scanner.old_height) + return prep_error(error::missed_messages, proto); + sync->height = from_scanner.new_height; + + if (sync->receives_only && from_scanner.outputs.empty()) + return epee::byte_slice{}; + + // suppress _some_ block messages when account is catching up + if (!sync->on_tip && from_scanner.outputs.empty() && from_scanner.spends.empty() && std::chrono::steady_clock::now() - sync->last_update <= scanner_busy) + return epee::byte_slice{}; + + feed_blocks blocks{ + std::min(sync->last_reported, from_scanner.old_height), + from_scanner.new_height, + from_scanner.new_height + }; + + /* A read is unfortunate here, but it also catches accounts that have + been moved to "hidden". The stream will immediately abort on error. */ + + auto reader = disk.start_read(); + if (!reader) + return reader.error(); + const auto last_block = reader->get_last_block(); + if (!last_block) + return last_block.error(); + auto account = reader->get_account(db::account_status::active, sync->id); + if (account == lws::error::account_not_found) + account = reader->get_account(db::account_status::inactive, sync->id); // rare as scanner should not push updates + if (!account) + return account.error(); + + blocks.blockchain_height = last_block->id; + blocks.lookahead_fail = account->lookahead_fail; + blocks.lookahead = account->lookahead; + reader->finish_read(); + + blocks.transactions = + get_address_txs_response::load(std::move(from_scanner.outputs), std::move(from_scanner.spends)); + + sync->on_tip = (to_uint(last_block->id) - to_uint(std::min(last_block->id, from_scanner.new_height))) <= on_tip; + sync->last_update = std::chrono::steady_clock::now(); + sync->last_reported = from_scanner.new_height; + return prep(blocks, proto); + } + + // mempool + LWS_VERIFY(from_scanner.outputs.size() == 1); + 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) + { + return do_start(std::move(sock), io, client, req, disk, timeout); + } +}}} // lws // rpc // feed + diff --git a/src/rpc/fwd.h b/src/rpc/fwd.h index a60fc84..56f2cff 100644 --- a/src/rpc/fwd.h +++ b/src/rpc/fwd.h @@ -27,11 +27,20 @@ #pragma once +#include + namespace lws { namespace rpc { + struct account_credentials; class client; + + namespace feed + { + enum class error; + enum class status : std::uint16_t; + } } struct rates; class scan_manager; diff --git a/src/rpc/light_wallet.cpp b/src/rpc/light_wallet.cpp index 179235a..706b199 100644 --- a/src/rpc/light_wallet.cpp +++ b/src/rpc/light_wallet.cpp @@ -35,11 +35,13 @@ #include #include "config.h" +#include "db/storage.h" #include "db/string.h" #include "error.h" #include "lws_version.h" #include "time_helper.h" // monero/contrib/epee/include #include "ringct/rctOps.h" // monero/src +#include "rpc/feed.h" #include "span.h" // monero/contrib/epee/include #include "util/random_outputs.h" #include "version.h" // monero/src @@ -47,6 +49,7 @@ #include "wire/adapted/crypto.h" #include "wire/error.h" #include "wire/json.h" +#include "wire/msgpack/read.h" #include "wire/traits.h" #include "wire/vector.h" #include "wire/wrapper/array.h" @@ -196,7 +199,7 @@ namespace lws source.end_array(); } - void rpc::read_bytes(wire::json_reader& source, account_credentials& self) + void rpc::read_bytes(wire::reader& source, account_credentials& self) { std::string address; wire::object(source, @@ -237,6 +240,217 @@ namespace lws ); } + rpc::feed_error::feed_error(std::error_code error) + : msg(error.message()), + code(feed::map(error)) + {} + + void rpc::write_bytes(wire::writer& dest, const feed_error& self) + { + wire::object(dest, WIRE_FIELD(msg), WIRE_FIELD(code)); + } + + void rpc::read_bytes(wire::reader& src, feed_login& self) + { + wire::object(src, + WIRE_FIELD(account), + WIRE_FIELD_DEFAULTED(tx_sync, true), + WIRE_FIELD_DEFAULTED(receives_only, false) + ); + } + + void rpc::write_bytes(wire::writer& dest, const feed_mempool& self) + { + epee::span const* payment_id = nullptr; + epee::span payment_id_bytes; + + const auto extra = db::unpack(self.received.extra); + if (extra.second) + { + payment_id = std::addressof(payment_id_bytes); + + if (extra.second == sizeof(self.received.payment_id.short_)) + payment_id_bytes = epee::as_byte_span(self.received.payment_id.short_); + else + payment_id_bytes = epee::as_byte_span(self.received.payment_id.long_); + } + + rct::key const* optional_rct = nullptr; + if (extra.first & lws::db::ringct_output) + optional_rct = std::addressof(self.received.ringct_mask); + + wire::object(dest, + wire::field("hash", std::cref(self.received.link.tx_hash)), + wire::field("prefix_hash", std::cref(self.received.tx_prefix_hash)), + wire::field("fee", self.received.fee), + wire::field("unlock_time", self.received.unlock_time), + wire::optional_field("payment_id", payment_id), + wire::field("mixin", self.received.spend_meta.mixin_count), + wire::field("amount", self.received.spend_meta.amount), + wire::field("public_key", std::cref(self.received.pub)), + wire::field("tx_pub_key", std::cref(self.received.spend_meta.tx_public)), + wire::field("index", self.received.spend_meta.index), + wire::optional_field("recipient", wire::defaulted(std::cref(self.received.recipient), db::address_index{})), + wire::optional_field("rct", optional_rct) + ); + } + + namespace + { + template + struct sync_wrapper + { + const T& data; + }; + + struct sync_transform_ + { + template + sync_wrapper operator()(const T& value) const noexcept + { return {value}; } + }; + constexpr const sync_transform_ sync_transform{}; + + struct legacy_temp { db::output_id id; }; + struct composite_temp { db::output_id id; }; + + bool operator!=(const composite_temp& left, const composite_temp& right) noexcept + { return left.id != right.id; } + + void write_bytes(wire::writer& dest, const legacy_temp& self) + { + wire::object(dest, + wire::field("amount", self.id.high), + wire::field("index", self.id.low) + ); + } + + void write_bytes(wire::writer& dest, const composite_temp& self) + { + wire::object(dest, wire::field("legacy", legacy_temp{self.id})); + } + + void write_bytes(wire::writer& dest, sync_wrapper self) + { + rct::key rct{}; + rct::key const* optional_rct = nullptr; + const auto flags = unpack(self.data.extra).first; + if (flags & lws::db::ringct_output) + { + if (!(flags & lws::db::coinbase_output)) + rct = self.data.ringct_mask; + else + rct = rct::identity(); + + optional_rct = std::addressof(rct); + } + + wire::object(dest, + wire::field("amount", self.data.spend_meta.amount), + wire::field("public_key", std::cref(self.data.pub)), + wire::field("index", self.data.spend_meta.index), + wire::optional_field("id", wire::defaulted(composite_temp{self.data.spend_meta.id}, composite_temp{db::output_id::txpool()})), + wire::field("tx_pub_key", std::cref(self.data.spend_meta.tx_public)), + wire::optional_field("rct", optional_rct), + wire::optional_field("recipient", wire::defaulted(std::cref(self.data.recipient), db::address_index{})) + ); + } + + void write_bytes(wire::writer& dest, sync_wrapper self) + { + wire::object(dest, + wire::field("id", composite_temp{self.data.possible_spend.source}), + wire::field("key_image", std::cref(self.data.possible_spend.image)) + ); + } + + void write_bytes(wire::writer& dest, sync_wrapper self) + { + epee::span const* payment_id = nullptr; + epee::span payment_id_bytes; + + const auto extra = db::unpack(self.data.info.extra); + if (extra.second) + { + payment_id = std::addressof(payment_id_bytes); + + if (extra.second == sizeof(self.data.info.payment_id.short_)) + payment_id_bytes = epee::as_byte_span(self.data.info.payment_id.short_); + else + payment_id_bytes = epee::as_byte_span(self.data.info.payment_id.long_); + } + + const bool is_coinbase = (extra.first & db::coinbase_output); + const bool txpool = self.data.info.link.height == db::block_id::txpool; + wire::object(dest, + wire::field("hash", std::cref(self.data.info.link.tx_hash)), + wire::field("prefix_hash", std::cref(self.data.info.tx_prefix_hash)), + wire::field("timestamp", self.data.info.timestamp), + wire::field("fee", self.data.info.fee), + wire::field("unlock_time", self.data.info.unlock_time), + wire::optional_field("height", wire::defaulted(self.data.info.link.height, db::block_id::txpool)), + wire::optional_field("payment_id", payment_id), + wire::optional_field("coinbase", wire::defaulted(is_coinbase, false)), + wire::optional_field("mempool", wire::defaulted(txpool, false)), + wire::field("mixin", self.data.info.spend_meta.mixin_count), + wire::optional_field("spends", wire::array(boost::adaptors::transform(self.data.spends, sync_transform))), + wire::optional_field("receives", wire::array(boost::adaptors::transform(self.data.receives, sync_transform))) + ); + } + } + + void rpc::write_bytes(wire::writer& dest, const feed_tx_sync& self) + { + wire::object(dest, + wire::field("scanned_block_height", self.info.scanned_block_height), + wire::field("start_height", self.info.start_height), + wire::field("blockchain_height", self.info.blockchain_height), + wire::optional_field("lookahead_fail", wire::defaulted(self.info.lookahead_fail, unsigned(0))), + wire::optional_field("lookahead", wire::defaulted(self.info.lookahead, db::address_index{})), + wire::optional_field("transactions", wire::array(boost::adaptors::transform(self.info.transactions, sync_transform))) + ); + } + + void rpc::write_bytes(wire::writer& dest, const feed_blocks& self) + { + wire::object(dest, + WIRE_FIELD(scan_start), + WIRE_FIELD(scan_end), + WIRE_FIELD(blockchain_height), + wire::optional_field("lookahead_fail", wire::defaulted(self.lookahead_fail, db::block_id(0))), + wire::optional_field("lookahead", wire::defaulted(std::cref(self.lookahead), db::address_index{})), + wire::optional_field("transactions", wire::array(boost::adaptors::transform(self.transactions, sync_transform))) + ); + } + + rpc::feed_warning::feed_warning(std::error_code error, const std::uint32_t counter, const db::block_id height) + : msg(error.message()), + code(feed::map(error)), + counter(counter), + height(height) + {} + + namespace + { + template + void map_feed_warning(F& format, T& self) + { + wire::object(format, + WIRE_FIELD_ID(0, msg), + WIRE_FIELD_ID(1, code), + WIRE_FIELD_ID(2, counter), + WIRE_FIELD_ID(3, height) + ); + } + } + + void rpc::read_bytes(wire::msgpack_reader& src, feed_warning& dest) + { map_feed_warning(src, dest); } + + void rpc::write_bytes(wire::writer& dest, const feed_warning& src) + { map_feed_warning(dest, src); } + + void rpc::write_bytes(wire::json_writer& dest, const new_subaddrs_response& self) { wire::object(dest, WIRE_FIELD(new_subaddrs), WIRE_FIELD(all_subaddrs)); @@ -275,7 +489,7 @@ namespace lws namespace rpc { - static void write_bytes(wire::json_writer& dest, boost::range::index_value self) + static void write_bytes(wire::json_writer& dest, boost::range::index_value self) { epee::span const* payment_id = nullptr; epee::span payment_id_bytes; @@ -311,6 +525,204 @@ namespace lws ); } } // rpc + + std::vector::const_iterator + rpc::get_address_txs_response::find_metadata(std::vector const& metas, const db::output_id id) + { + struct by_output_id + { + bool operator()(db::output::spend_meta_ const& left, db::output_id right) const noexcept + { + return left.id < right; + } + bool operator()(db::output_id left, db::output::spend_meta_ const& right) const noexcept + { + return left < right.id; + } + }; + return std::lower_bound(metas.begin(), metas.end(), id, by_output_id{}); + } + + namespace + { + template + struct vec_lmdb_ + { + using iterator = typename std::vector::const_iterator; + using value_type = T; + + iterator pos; + const iterator end; + + bool is_end() const { return pos == end; } + vec_lmdb_& operator++() + { + ++pos; + return *this; + } + + template + G get_value() const noexcept + { + static_assert(std::is_same(), "bad MONERO_FIELD usage?"); + static_assert(std::is_trivially_copyable(), "value type must be memcpy safe"); + static_assert(std::is_trivially_copyable(), "field type must be memcpy safe"); + static_assert(sizeof(G) + uoffset <= sizeof(U), "bad field and/or offset"); + assert(sizeof(G) + uoffset <= end - pos); + assert(!is_end()); + + G value; + std::memcpy(std::addressof(value), reinterpret_cast(std::addressof(*pos)) + uoffset, sizeof(value)); + return value; + } + + const value_type& operator*() const noexcept { return *pos; } + }; + + template + vec_lmdb_ vec_lmdb(const std::vector& src) { return {src.begin(), src.end()}; } + + template + std::pair, std::uint64_t> merge_into_txes(T output, U spend, 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. + + std::uint64_t total = 0; + std::vector out{}; + std::vector metas{}; + + out.reserve(reserve); + metas.reserve(reserve); + + db::transaction_link next_output{}; + db::transaction_link next_spend{}; + + if (!output.is_end()) + next_output = output.template get_value(); + if (!spend.is_end()) + next_spend = spend.template get_value(); + + while (!output.is_end() || !spend.is_end()) + { + if (!out.empty()) + { + db::transaction_link const& last = out.back().info.link; + LWS_VERIFY((output.is_end() || last <= next_output) && (spend.is_end() || last <= next_spend)); + } + + if (spend.is_end() || (!output.is_end() && next_output <= next_spend)) + { + LWS_VERIFY(!output.is_end()); + + std::uint64_t amount = 0; + if (out.empty() || out.back().info.link.tx_hash != next_output.tx_hash) + { + out.push_back({*output}); + amount = out.back().info.spend_meta.amount; + } + else + { + amount = output.template get_value(); + out.back().info.spend_meta.amount += amount; + } + + if (all_outputs) + out.back().receives.push_back(*output); + + const db::output::spend_meta_ meta = output.template get_value(); + if (metas.empty() || metas.back().id < meta.id) + metas.push_back(meta); + else + metas.insert(rpc::get_address_txs_response::find_metadata(metas, meta.id), meta); + + total += amount; + + ++output; + if (!output.is_end()) + next_output = output.template get_value(); + } + else if (output.is_end() || (next_spend < next_output)) + { + using meta_type = db::output::spend_meta_; + LWS_VERIFY(!spend.is_end()); + + const db::output_id source_id = spend.template get_value(); + const auto meta = rpc::get_address_txs_response::find_metadata(metas, source_id); + LWS_VERIFY(skip_spend_meta || (meta != metas.end() && meta->id == source_id)); + + if (out.empty() || out.back().info.link.tx_hash != next_spend.tx_hash) + { + out.push_back({}); + out.back().spends.push_back({skip_spend_meta ? meta_type{} : *meta, *spend}); + out.back().info.link.height = out.back().spends.back().possible_spend.link.height; + out.back().info.link.tx_hash = out.back().spends.back().possible_spend.link.tx_hash; + out.back().info.spend_meta.mixin_count = + 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; + } + else + out.back().spends.push_back({skip_spend_meta ? meta_type{} : *meta, *spend}); + + if (!skip_spend_meta) + out.back().spent += meta->amount; + + ++spend; + if (!spend.is_end()) + next_spend = spend.template get_value(); + } + } + + return {std::move(out), total}; + } + + struct tx_link_sorter_ + { + template + bool operator()(const T& left, const T& right) const noexcept + { + return left.link < right.link; + } + }; + constexpr const tx_link_sorter_ by_tx_link{}; + } + + std::vector rpc::get_address_txs_response::load(std::vector outputs, std::vector spends) + { + 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; + } + + expect rpc::get_address_txs_response::load(db::storage_reader& reader, const db::account& acct, const bool all_outputs) + { + auto outputs = reader.get_outputs(acct.id); + if (!outputs) + return outputs.error(); + + auto spends = reader.get_spends(acct.id); + if (!spends) + return spends.error(); + + const expect last = reader.get_last_block(); + if (!last) + return last.error(); + + get_address_txs_response resp{}; + resp.scanned_height = std::uint64_t(acct.scan_height); + resp.scanned_block_height = resp.scanned_height; + resp.start_height = std::uint64_t(acct.start_height); + resp.blockchain_height = std::uint64_t(last->id); + resp.transaction_height = resp.blockchain_height; + 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); + resp.total_received = safe_uint64(out.second); + resp.transactions = std::move(out.first); + return resp; + } + void rpc::write_bytes(wire::json_writer& dest, const get_address_txs_response& self) { wire::object(dest, @@ -457,7 +869,7 @@ namespace lws ); convert_address(address, self.creds.address); } - + void rpc::read_bytes(wire::json_reader& source, submit_raw_tx_request& self) { wire::object(source, WIRE_FIELD(tx)); diff --git a/src/rpc/light_wallet.h b/src/rpc/light_wallet.h index 7fe89ac..952224d 100644 --- a/src/rpc/light_wallet.h +++ b/src/rpc/light_wallet.h @@ -38,9 +38,12 @@ #include "cryptonote_basic/difficulty.h" // monero/src #include "crypto/crypto.h" // monero/src #include "db/data.h" +#include "db/fwd.h" +#include "rpc/fwd.h" #include "rpc/rates.h" #include "util/fwd.h" #include "wire/json/fwd.h" +#include "wire/msgpack/fwd.h" namespace lws { @@ -64,7 +67,7 @@ namespace rpc lws::db::account_address address; crypto::secret_key key; }; - void read_bytes(wire::json_reader&, account_credentials&); + void read_bytes(wire::reader&, account_credentials&); enum class daemon_state : std::uint8_t @@ -107,15 +110,6 @@ namespace rpc void write_bytes(wire::json_writer&, const daemon_status_response&); - struct new_subaddrs_response - { - new_subaddrs_response() = delete; - std::vector new_subaddrs; - std::vector all_subaddrs; - }; - void write_bytes(wire::json_writer&, const new_subaddrs_response&); - - struct transaction_spend { transaction_spend() = delete; @@ -124,6 +118,114 @@ namespace rpc }; void write_bytes(wire::json_writer&, const transaction_spend&); + struct get_transaction + { + get_transaction() = delete; + db::output info; + std::vector receives; + std::vector spends; + std::uint64_t spent; + }; + + struct get_address_txs_response + { + get_address_txs_response() = delete; + + static std::vector::const_iterator + find_metadata(std::vector const& metas, db::output_id id); + + static std::vector load(std::vector outputs, std::vector spends); + static expect load(db::storage_reader& reader, const db::account& acct, const bool all_outputs); + + safe_uint64 total_received; + std::uint64_t scanned_height; + std::uint64_t scanned_block_height; + std::uint64_t start_height; + std::uint64_t transaction_height; + std::uint64_t blockchain_height; + std::uint64_t lookahead_fail; + std::vector transactions; + db::address_index lookahead; + }; + void write_bytes(wire::json_writer&, const get_address_txs_response&); + + + struct feed_blocks + { + static constexpr const char* prefix() noexcept { return "blocks:"; } + feed_blocks() = delete; + db::block_id scan_start; + db::block_id scan_end; + db::block_id blockchain_height; + db::block_id lookahead_fail; + db::address_index lookahead; + std::vector transactions; + }; + void write_bytes(wire::writer&, const feed_blocks&); + + struct feed_error + { + static constexpr const char* prefix() noexcept { return "error:"; } + feed_error() = delete; + feed_error(std::error_code); + + std::string msg; + feed::status code; + }; + void write_bytes(wire::writer&, const feed_error&); + + struct feed_login + { + static constexpr const char* prefix() noexcept { return "login:"; } + feed_login() = delete; + account_credentials account; + bool tx_sync; + bool receives_only; + }; + void read_bytes(wire::reader&, feed_login&); + + struct feed_mempool + { + static constexpr const char* prefix() noexcept { return "mempool:"; } + feed_mempool() = delete; + db::output received; + }; + void write_bytes(wire::writer&, const feed_mempool&); + + struct feed_tx_sync + { + static constexpr const char* prefix() noexcept { return "tx_sync:"; } + feed_tx_sync() = delete; + get_address_txs_response info; + }; + void write_bytes(wire::writer&, const feed_tx_sync&); + + struct feed_warning + { + static constexpr const char* prefix() noexcept { return "warning:"; } + + feed_warning() + : msg(), code(feed::status(0)), counter(0), height(db::block_id::txpool) + {} + + feed_warning(std::error_code error, const std::uint32_t counter, const db::block_id height); + + std::string msg; + feed::status code; + std::uint32_t counter; + db::block_id height; + }; + void read_bytes(wire::msgpack_reader&, feed_warning&); + void write_bytes(wire::writer&, const feed_warning&); + + + struct new_subaddrs_response + { + new_subaddrs_response() = delete; + std::vector new_subaddrs; + std::vector all_subaddrs; + }; + void write_bytes(wire::json_writer&, const new_subaddrs_response&); struct get_address_info_response { @@ -138,7 +240,8 @@ namespace rpc blockchain_height(0), lookahead_fail(0), spent_outputs(), - rates(common_error::kInvalidArgument) + rates(common_error::kInvalidArgument), + lookahead{} {} safe_uint64 locked_funds; @@ -155,31 +258,7 @@ namespace rpc db::address_index lookahead; }; void write_bytes(wire::json_writer&, const get_address_info_response&); - - - struct get_address_txs_response - { - get_address_txs_response() = delete; - struct transaction - { - transaction() = delete; - db::output info; - std::vector spends; - std::uint64_t spent; - }; - - safe_uint64 total_received; - std::uint64_t scanned_height; - std::uint64_t scanned_block_height; - std::uint64_t start_height; - std::uint64_t transaction_height; - std::uint64_t blockchain_height; - std::uint64_t lookahead_fail; - std::vector transactions; - db::address_index lookahead; - }; - void write_bytes(wire::json_writer&, const get_address_txs_response&); - + struct get_random_outs_request { @@ -298,7 +377,7 @@ namespace rpc db::address_index lookahead; }; void write_bytes(wire::json_writer&, login_response); - + struct provision_subaddrs_request { @@ -310,9 +389,9 @@ namespace rpc boost::optional n_min; boost::optional get_all; }; - void read_bytes(wire::json_reader&, provision_subaddrs_request&); + void read_bytes(wire::json_reader&, provision_subaddrs_request&); + - struct submit_raw_tx_request { submit_raw_tx_request() = delete; diff --git a/src/rpc/login.cpp b/src/rpc/login.cpp new file mode 100644 index 0000000..fd7376a --- /dev/null +++ b/src/rpc/login.cpp @@ -0,0 +1,80 @@ +// 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 "login.h" + +#include "crypto/crypto.h" // monero/src +#include "db/data.h" +#include "db/storage.h" +#include "error.h" +#include "rpc/light_wallet.h" + +namespace lws { namespace rpc +{ + bool is_hidden(db::account_status status) noexcept + { + switch (status) + { + case db::account_status::active: + case db::account_status::inactive: + return false; + default: + case db::account_status::hidden: + break; + } + return true; + } + + bool key_check(const rpc::account_credentials& creds) + { + crypto::public_key verify{}; + if (!crypto::secret_key_to_public_key(creds.key, verify)) + return false; + if (verify != creds.address.view_public) + return false; + return true; + } + + //! \return Account info from the DB, iff key matches address AND address is NOT hidden. + expect> open_account(const account_credentials& creds, const db::storage& disk) + { + if (!key_check(creds)) + return {lws::error::bad_view_key}; + + auto reader = disk.start_read(); + if (!reader) + return reader.error(); + + const auto user = reader->get_account(creds.address); + if (!user) + return user.error(); + if (is_hidden(user->first)) + return {lws::error::account_not_found}; + return {std::make_pair(user->second, std::move(*reader))}; + } + +}} // lws // rpc diff --git a/src/rpc/login.h b/src/rpc/login.h new file mode 100644 index 0000000..ac2e8ac --- /dev/null +++ b/src/rpc/login.h @@ -0,0 +1,45 @@ +// 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 + +#include "common/expect.h" // monero/src +#include "db/fwd.h" +#include "rpc/fwd.h" + +namespace lws { namespace rpc +{ + bool is_hidden(db::account_status status) noexcept; + + //! Verify that view pub matches view secret. + bool key_check(const rpc::account_credentials& creds); + + //! \return Account info from the DB, iff key matches address AND address is NOT hidden. + expect> open_account(const account_credentials& creds, const db::storage& disk); +}} // lws // rpc diff --git a/src/rpc/scanner/server.cpp b/src/rpc/scanner/server.cpp index 22ecc9a..1091255 100644 --- a/src/rpc/scanner/server.cpp +++ b/src/rpc/scanner/server.cpp @@ -623,9 +623,9 @@ namespace lws { namespace rpc { namespace scanner std::sort(users.begin(), users.end(), by_height{}); boost::asio::dispatch( self->strand_, - [self, users = std::move(users), blocks = std::move(blocks)] () + [self, users = std::move(users), blocks = std::move(blocks)] () mutable { - if (!lws::user_data::store(self->strand_.context(), self->disk_, self->zclient_, self->webhook_, epee::to_span(blocks), epee::to_span(users), nullptr)) + if (!lws::user_data::store(self->strand_.context(), self->disk_, self->zclient_, self->webhook_, epee::to_span(blocks), epee::to_mut_span(users), nullptr)) { self->do_stop(); self->strand_.context().stop(); diff --git a/src/scanner.cpp b/src/scanner.cpp index 00996ba..1bd70c5 100644 --- a/src/scanner.cpp +++ b/src/scanner.cpp @@ -284,6 +284,9 @@ namespace lws events.pop_back(); //cannot compute tx_hash } send_payment_hook(http_.io_, client_, http_.webhooks_, epee::to_span(events)); + const expect pushed = client_.push_update(user, mempool_receive{out}); + if (!pushed) + MERROR("Failed to send mempool update to " << user.address() << ": " << pushed.error().message()); return true; } }; @@ -314,7 +317,7 @@ namespace lws } }; - void update_lookahead(const account& user, subaddress_reader& reader, const db::address_index& match, const db::block_id height) + 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 @@ -322,6 +325,9 @@ namespace lws 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) { @@ -339,7 +345,7 @@ namespace lws epee::span users, const db::block_id height, const std::uint64_t timestamp, - crypto::hash const& tx_hash, + crypto::hash const* tx_hash, cryptonote::transaction const& tx, std::vector const& out_ids, subaddress_reader& reader, @@ -349,6 +355,7 @@ namespace lws if (2 < tx.version) throw std::runtime_error{"Unsupported tx version"}; + crypto::hash tx_hash_lazy; cryptonote::tx_extra_pub_key key; boost::optional prefix_hash; boost::optional extra_nonce; @@ -356,6 +363,16 @@ namespace lws cryptonote::tx_extra_additional_pub_keys additional_tx_pub_keys; std::vector 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 extra; cryptonote::parse_tx_extra(tx.extra, extra); @@ -426,7 +443,7 @@ namespace lws spend_action( user, db::spend{ - db::transaction_link{height, tx_hash}, + db::transaction_link{height, get_tx_hash()}, in_data->k_image, db::output_id{in_data->amount, goffset}, timestamp, @@ -533,7 +550,7 @@ namespace lws ); if (!decrypted) { - MWARNING(user.address() << " failed to decrypt amount for tx " << tx_hash << ", skipping output"); + MWARNING(user.address() << " failed to decrypt amount for tx " << get_tx_hash() << ", skipping output"); continue; // to next output } amount = decrypted->first; @@ -555,7 +572,7 @@ namespace lws reader.reader, user, db::output{ - db::transaction_link{height, tx_hash}, + 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, @@ -591,7 +608,7 @@ namespace lws std::vector const& out_ids, subaddress_reader& reader) { - scan_transaction_base(users, height, timestamp, tx_hash, tx, out_ids, reader, add_spend{}, add_output{}); + 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 users, db::storage const& disk, scanner_sync& self, rpc::client& client, const scanner_options& opts) @@ -614,7 +631,7 @@ namespace lws subaddress_reader reader{std::optional{disk.clone()}, opts.max_subaddresses}; send_webhook sender{disk, client, self}; for (const auto& tx : parsed->txes) - scan_transaction_base(users, db::block_id::txpool, time, crypto::hash{}, tx, fake_outs, reader, null_spend{}, sender); + scan_transaction_base(users, db::block_id::txpool, time, nullptr, tx, fake_outs, reader, null_spend{}, sender); } void do_scan_loop(scanner_sync& self, std::shared_ptr data, const size_t thread_n) noexcept @@ -975,7 +992,7 @@ namespace lws MINFO("Thread " << thread_n << " processed " << blockchain.size() << " blocks(s) @ height " << fetched->start_height << " against " << users.size() << " account(s)"); - if (!store(self.io_, client, self.webhooks_, epee::to_span(blockchain), epee::to_span(users), epee::to_span(new_pow))) + if (!store(self.io_, client, self.webhooks_, epee::to_span(blockchain), epee::to_mut_span(users), epee::to_span(new_pow))) return false; // TODO @@ -984,9 +1001,6 @@ namespace lws MINFO("On chain with hash " << blockchain.back() << " and difficulty " << diff << " at height " << fetched->start_height); } - for (account& user : users) - user.updated(db::block_id(fetched->start_height)); - // Update queue with current minimum scan height (oldest account determines thread progress) queue.update_min_height(users.front().scan_height()); } @@ -1368,6 +1382,7 @@ namespace lws req.start_height = 0; req.known_hashes = MONERO_UNWRAP(MONERO_UNWRAP(disk.start_read()).get_chain_sync()); + bool sent_warning = false; for (;;) { if (req.known_hashes.empty()) @@ -1384,6 +1399,9 @@ namespace lws return {std::move(client)}; MONERO_CHECK(disk.sync_chain(db::block_id(resp->start_height), epee::to_span(resp->hashes), regtest)); + if (!sent_warning) + MONERO_UNWRAP(client.push_warning(error::blockchain_reorg, db::block_id(resp->start_height))); + sent_warning = true; req.known_hashes.erase(req.known_hashes.begin(), --(req.known_hashes.end())); for (std::size_t num = 0; num < 10; ++num) @@ -1408,6 +1426,7 @@ namespace lws req.block_ids = MONERO_UNWRAP(MONERO_UNWRAP(disk.start_read()).get_pow_sync()); req.prune = true; + bool sent_warning = false; std::vector new_hashes{}; std::vector new_pow{}; for (;;) @@ -1439,7 +1458,7 @@ namespace lws return {error::bad_daemon_response}; return {std::move(client)}; } - + // genesis block must be present as last entry req.block_ids.erase(req.block_ids.begin(), --(req.block_ids.end())); @@ -1516,20 +1535,25 @@ namespace lws MONERO_CHECK(disk.sync_pow(db::block_id(resp->start_height), epee::to_span(new_hashes), epee::to_span(new_pow))); MINFO("Verified up to block " << (resp->start_height + new_hashes.size() - 1) << " with hash " << hash << " and difficulty " << diff); + if (!sent_warning) + MONERO_UNWRAP(client.push_warning(error::blockchain_reorg, db::block_id(resp->start_height))); + sent_warning = true; + } // for until sync return {std::move(client)}; } } // anonymous - bool user_data::store(boost::asio::io_context& io, db::storage& disk, rpc::client& client, net::http::client& webhook, const epee::span chain, const epee::span users, const epee::span pow) + bool user_data::store(boost::asio::io_context& io, db::storage& disk, rpc::client& client, net::http::client& webhook, const epee::span chain, const epee::span users, const epee::span pow) { if (users.empty()) return true; - if (!std::is_sorted(users.begin(), users.end(), by_height{})) - throw std::logic_error{"users must be sorted!"}; + LWS_VERIFY(!chain.empty()); + LWS_VERIFY(std::is_sorted(users.begin(), users.end(), by_height{})); - auto updated = disk.update(users[0].scan_height(), chain, users, pow); + const db::block_id base_height = users[0].scan_height(); + auto updated = disk.update(base_height, chain, epee::to_span(users), pow); if (!updated) { if (updated == lws::error::blockchain_reorg) @@ -1542,6 +1566,7 @@ namespace lws send_payment_hook(io, client, webhook, epee::to_span(updated->confirm_pubs)); send_spend_hook(io, client, webhook, epee::to_span(updated->spend_pubs)); + client.push_updates(users, db::block_id(to_uint(base_height) + chain.size() - 1)); if (updated->accounts_updated != users.size()) { MWARNING("Only updated " << updated->accounts_updated << " account(s) out of " << users.size() << ", resetting"); @@ -1556,7 +1581,7 @@ namespace lws return true; } - bool user_data::operator()(boost::asio::io_context& io, rpc::client& client, net::http::client& webhook, const epee::span chain, const epee::span users, const epee::span pow) + bool user_data::operator()(boost::asio::io_context& io, rpc::client& client, net::http::client& webhook, const epee::span chain, const epee::span users, const epee::span pow) { return store(io, disk_, client, webhook, chain, users, pow); } @@ -1673,6 +1698,7 @@ namespace lws MONERO_THROW(synced.error(), "Unable to sync blockchain"); MWARNING("Failed to connect to daemon at " << ctx.daemon_address()); + MONERO_UNWRAP(ctx.push_warning(synced.error())); } else client = std::move(*synced); diff --git a/src/scanner.h b/src/scanner.h index aecc02c..f5156c7 100644 --- a/src/scanner.h +++ b/src/scanner.h @@ -77,10 +77,10 @@ namespace lws /*! Store updated accounts locally (`disk`), and send ZMQ/RMQ/webhook events. `users` must be sorted by height (lowest first). */ - static bool store(boost::asio::io_context& io, db::storage& disk, rpc::client& zclient, net::http::client& webhook ,epee::span chain, epee::span users, epee::span pow); + static bool store(boost::asio::io_context& io, db::storage& disk, rpc::client& zclient, net::http::client& webhook ,epee::span chain, epee::span users, epee::span pow); //! `users` must be sorted by height (lowest first) - bool operator()(boost::asio::io_context& io, rpc::client& zclient, net::http::client& webhook, epee::span chain, epee::span users, epee::span pow); + bool operator()(boost::asio::io_context& io, rpc::client& zclient, net::http::client& webhook, epee::span chain, epee::span users, epee::span pow); }; struct scanner_sync @@ -117,7 +117,7 @@ namespace lws ~scanner(); //! Callback for storing user account (typically local lmdb, but perhaps remote rpc) - using store_func = std::function, epee::span, epee::span)>; + using store_func = std::function, epee::span, epee::span)>; /*! Run _just_ the inner scanner loop while `self.is_running() == true`. * diff --git a/src/server_main.cpp b/src/server_main.cpp index dbcdeec..8a2c0c9 100644 --- a/src/server_main.cpp +++ b/src/server_main.cpp @@ -92,6 +92,7 @@ namespace const command_line::arg_descriptor split_sync_threads; const command_line::arg_descriptor split_sync_depth; const command_line::arg_descriptor balance_new_addresses; + const command_line::arg_descriptor feed_timeout; static std::string get_default_zmq() { @@ -147,6 +148,7 @@ namespace , split_sync_threads{"split-sync-threads", "Percentage of threads to use for fully synced accounts (0-1, requires --block-depth-threading, 0 to disable)", 0.0} , split_sync_depth{"split-sync-depth", "Maximum block depth for an address to be considered synced (defaults to 10)", 10} , balance_new_addresses{"balance-new-addresses", "Assign new addresses to thread with highest blockheight (or fewest addresses if tied)", false} + , feed_timeout{"feed-timeout", "Seconds for timing out inactive websocket '/feed' clients. val <= 0 disables '/feed'", std::chrono::seconds{lws::config::feed_timeout}.count()} {} void prepare(boost::program_options::options_description& description) const @@ -190,6 +192,7 @@ namespace command_line::add_arg(description, split_sync_threads); command_line::add_arg(description, split_sync_depth); command_line::add_arg(description, balance_new_addresses); + command_line::add_arg(description, feed_timeout); } }; @@ -292,6 +295,7 @@ namespace lws::rest_server::configuration{ {command_line::get_arg(args, opts.rest_ssl_key), command_line::get_arg(args, opts.rest_ssl_cert)}, command_line::get_arg(args, opts.access_controls), + std::chrono::seconds{command_line::get_arg(args, opts.feed_timeout)}, command_line::get_arg(args, opts.rest_threads), command_line::get_arg(args, opts.max_subaddresses), webhook_verify, @@ -356,7 +360,15 @@ namespace boost::filesystem::create_directories(prog.db_path); auto disk = lws::db::storage::open(prog.db_path.c_str(), prog.create_queue_max); - auto ctx = lws::rpc::context::make(std::move(prog.daemon_rpc), std::move(prog.daemon_sub), std::move(prog.zmq_pub), std::move(prog.rmq), prog.rates_interval, prog.untrusted_daemon); + auto ctx = lws::rpc::context::make( + std::move(prog.daemon_rpc), + std::move(prog.daemon_sub), + std::move(prog.zmq_pub), + std::move(prog.rmq), + prog.rates_interval, + prog.untrusted_daemon, + std::chrono::seconds{0} < prog.rest_config.feed_timeout + ); //! SIGINT handle registered by `scanner` constructor lws::scanner scanner{disk.clone(), prog.rest_config.webhook_verify}; diff --git a/tests/unit/rest.test.cpp b/tests/unit/rest.test.cpp index 2243c24..3518a97 100644 --- a/tests/unit/rest.test.cpp +++ b/tests/unit/rest.test.cpp @@ -27,7 +27,11 @@ #include "framework.test.h" +#include +#include +#include #include + #include "db/data.h" #include "db/print.test.h" #include "db/storage.test.h" @@ -35,8 +39,26 @@ #include "error.h" #include "hex.h" // monero/epee/contrib/include #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" + +namespace rapidjson +{ + std::ostream& operator<<(std::ostream& out, const Document& src) + { + StringBuffer buffer; + PrettyWriter writer{buffer}; + src.Accept(writer); + + return out << buffer.GetString(); + } +} namespace { @@ -59,14 +81,20 @@ namespace rct::key amount; }; - std::string invoke(enet::http::http_simple_client& client, const boost::string_ref uri, const boost::string_ref body) + std::pair invoke_base(enet::http::http_simple_client& client, const boost::string_ref uri, const boost::string_ref body) { const enet::http::http_response_info* info = nullptr; if (!client.invoke(uri, "POST", body, std::chrono::milliseconds{500}, std::addressof(info), {})) throw std::runtime_error{"HTTP invoke failed"}; - if (info->m_response_code != 200) - throw std::runtime_error{"HTTP invoke not 200, instead " + std::to_string(info->m_response_code)}; - return std::string{info->m_body}; + return {std::string{info->m_body}, info->m_response_code}; + } + + std::string invoke(enet::http::http_simple_client& client, const boost::string_ref uri, const boost::string_ref body) + { + auto result = invoke_base(client, uri, body); + if (result.second != 200) + throw std::runtime_error{"HTTP invoke not 200, instead " + std::to_string(result.second)}; + return result.first; } epee::byte_slice get_fee_response() @@ -105,6 +133,29 @@ namespace return out; } + + void verify_json(lest::env& lest_env, std::string name, std::string actual_json, std::string expected_json) + { + const lest::ctx ctx{lest_env, std::move(name)}; + + rapidjson::Document expected; + expected.Parse(expected_json.c_str()); + EXPECT(!expected.HasParseError()); + + rapidjson::Document actual; + actual.Parse(actual_json.c_str()); + EXPECT(!actual.HasParseError()); + + EXPECT(expected == actual); + } + + std::string get_prefix(const std::string& raw) + { + char const* const sep = std::strchr(raw.c_str(), u8':'); + if (!sep) + return {}; + return {raw.data(), std::size_t(sep - raw.data() + 1)}; + } } LWS_CASE("rest_server") @@ -122,11 +173,11 @@ LWS_CASE("rest_server") lws::db::test::cleanup_db on_scope_exit{}; lws::db::storage db = lws::db::test::get_fresh_db(); auto context = - lws::rpc::context::make(lws_test::rpc_rendevous, {}, {}, {}, std::chrono::minutes{0}, false); + lws::rpc::context::make(lws_test::rpc_rendevous, {}, {}, {}, std::chrono::minutes{0}, false, true); const auto rpc = MONERO_UNWRAP(context.connect()); { const lws::rest_server::configuration config{ - {}, {}, 1, 20, {}, false, true, true + {}, {}, std::chrono::seconds{10}, 1, 20, {}, false, true, true, false }; std::vector addresses{rest_server}; server.emplace( @@ -145,6 +196,13 @@ LWS_CASE("rest_server") return MONERO_UNWRAP(MONERO_UNWRAP(db.start_read()).get_account(account_address)).second; }; + const auto get_full_account = [&db, &get_account] () -> lws::account + { + const lws::db::account acct = get_account(); + return MONERO_UNWRAP(MONERO_UNWRAP(db.start_read()).get_full_account(acct)); + }; + + enet::http::http_simple_client client{}; client.set_server("127.0.0.1", "10000", boost::none); EXPECT(client.connect(std::chrono::milliseconds{500})); @@ -745,6 +803,331 @@ LWS_CASE("rest_server") "{\"key\":0,\"value\":[[1,20]]}]}" ); } + + SECTION("feed check") + { + auto response = invoke_base(client, "/feed", ""); + EXPECT(response.second == 501); + EXPECT(response.first == ""); + } + + SECTION("feed subscription") + { + const std::string scan_height = std::to_string(std::uint64_t(account.scan_height)); + namespace pull = lws_test::rpc::pull; + + const auto local = boost::asio::ip::make_address("127.0.0.1"); + boost::asio::io_context io; + boost::asio::steady_timer timeout{io}; + + timeout.expires_after(std::chrono::seconds{5}); + timeout.async_wait([&io] (auto) { io.stop(); }); + + // + // invalid version + // + std::shared_ptr conn = + pull::make(io, boost::asio::ip::tcp::endpoint(local, 10000), {}); + + bool ran = false; + boost::system::error_code error{}; + std::string response; + message = R"(login:{"account":{"address":")" + address + R"(","view_key":")" + viewkey + R"("}})"; + const auto handler = [&] (auto ec, auto msg) { ran = true; error = ec; response = msg; }; + + pull::async_handshake(conn, message, handler); + + while (!ran && !io.stopped()) + { + io.restart(); + io.run_one(); + } + + EXPECT(!io.stopped()); + EXPECT(error == boost::beast::websocket::error::upgrade_declined); + + // + // empty tx sync + // + conn = pull::make(io, boost::asio::ip::tcp::endpoint(local, 10000), "lws.feed.v0.json"); + pull::async_handshake(conn, message, handler); + + ran = false; + error = {}; + response = {}; + while (!ran && !io.stopped()) + { + io.restart(); + io.run_one(); + } + + std::string expected = + R"({"scanned_block_height":)" + scan_height + R"(, + "start_height":)" + scan_height + R"(, + "blockchain_height":)" + scan_height + "}"; + + 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", response, expected); + + const lws::db::transaction_link link{ + lws::db::block_id::txpool, crypto::rand() + }; + const crypto::public_key tx_public = []() { + crypto::secret_key secret; + crypto::public_key out; + crypto::generate_keys(out, secret); + return out; + }(); + const crypto::hash tx_prefix = crypto::rand(); + const crypto::public_key pub = crypto::rand(); + const crypto::public_key pub2 = crypto::rand(); + const rct::key ringct = crypto::rand(); + const auto extra = lws::db::extra(lws::db::extra::ringct_output); + const auto payment_id_ = crypto::rand(); + + rpc.push_update( + get_full_account(), + lws::mempool_receive{ + lws::db::output{ + link, + lws::db::output::spend_meta_{ + lws::db::output_id::txpool(), + std::uint64_t(40000), + std::uint32_t(16), + std::uint32_t(2), + tx_public + }, + std::uint64_t(7000), + std::uint64_t(4670), + tx_prefix, + pub, + ringct, + {0, 0, 0, 0, 0, 0, 0}, + lws::db::pack(extra, sizeof(crypto::hash)), + payment_id_, + std::uint64_t(100), + lws::db::address_index{lws::db::major_index(2), lws::db::minor_index(66)} + } + } + ); + pull::async_next_message(conn, handler); + + ran = false; + error = {}; + response = {}; + while (!ran && !io.stopped()) + { + io.restart(); + io.run_one(); + } + + expected = + R"({ + "hash":")" + epee::to_hex::string(epee::as_byte_span(link.tx_hash)) + R"(", + "prefix_hash":")" + epee::to_hex::string(epee::as_byte_span(tx_prefix)) + R"(", + "fee":100, + "unlock_time":4670, + "payment_id":")" + epee::to_hex::string(epee::as_byte_span(payment_id_.long_)) + R"(", + "mixin":16, + "amount":40000, + "public_key":")" + epee::to_hex::string(epee::as_byte_span(pub)) + R"(", + "index":2, + "tx_pub_key":")" + epee::to_hex::string(epee::as_byte_span(tx_public)) + R"(", + "rct":")" + epee::to_hex::string(epee::as_byte_span(ringct)) + R"(", + "recipient":{"maj_i":2,"min_i":66} + })"; + + EXPECT(!io.stopped()); + EXPECT(error == boost::system::error_code{}); + EXPECT(get_prefix(response) == "mempool:"); + response.erase(0, std::strlen("mempool:")); + verify_json(lest_env, "mempool", response, expected); + + const lws::db::block_id new_height = lws::db::block_id(to_uint(last_block.id) + 1); + const crypto::key_image image = crypto::rand(); + { + lws::account account = get_full_account(); + account.add_out( + lws::db::output{ + {new_height, link.tx_hash}, + lws::db::output::spend_meta_{ + lws::db::output_id{0, 100}, + std::uint64_t(40000), + std::uint32_t(16), + std::uint32_t(2), + tx_public + }, + std::uint64_t(7000), + std::uint64_t(4670), + tx_prefix, + pub, + ringct, + {0, 0, 0, 0, 0, 0, 0}, + lws::db::pack(extra, sizeof(crypto::hash)), + payment_id_, + std::uint64_t(100), + lws::db::address_index{lws::db::major_index(2), lws::db::minor_index(66)} + } + ); + account.add_out( + lws::db::output{ + {new_height, link.tx_hash}, + lws::db::output::spend_meta_{ + lws::db::output_id{0, 99}, + std::uint64_t(30000), + std::uint32_t(16), + std::uint32_t(1), + tx_public + }, + std::uint64_t(7000), + std::uint64_t(4670), + tx_prefix, + pub2, + ringct, + {0, 0, 0, 0, 0, 0, 0}, + lws::db::pack(extra, sizeof(crypto::hash)), + payment_id_, + std::uint64_t(100), + lws::db::address_index{lws::db::major_index(0), lws::db::minor_index(0)} + } + ); + + account.add_spend( + lws::db::spend{ + {new_height, link.tx_hash}, + image, + lws::db::output_id{0, 90}, + std::uint64_t(7000), + std::uint64_t(0), + std::uint16_t(16), + {0, 0, 0}, + std::uint8_t(0), + {}, + {} + } + ); + + const std::array chain{{ + last_block.hash, crypto::rand() + }}; + MONERO_UNWRAP(db.update(last_block.id, epee::to_span(chain), {std::addressof(account), 1}, nullptr)); + rpc.push_updates({std::addressof(account), 1}, new_height); + } + + pull::async_next_message(conn, handler); + + ran = false; + error = {}; + response = {}; + while (!ran && !io.stopped()) + { + io.restart(); + io.run_one(); + } + + std::string new_height_str = std::to_string(to_uint(new_height)); + expected = + R"({ + "scan_start":)" + scan_height + R"(, + "scan_end":)" + new_height_str + R"(, + "blockchain_height":)" + new_height_str + R"(, + "transactions":[{ + "hash":")" + epee::to_hex::string(epee::as_byte_span(link.tx_hash)) + R"(", + "prefix_hash":")" + epee::to_hex::string(epee::as_byte_span(tx_prefix)) + R"(", + "timestamp":7000, + "fee":100, + "unlock_time":4670, + "height":)" + new_height_str + R"(, + "payment_id":")" + epee::to_hex::string(epee::as_byte_span(payment_id_.long_)) + R"(", + "mixin":16, + "receives":[ + { + "amount":40000, + "public_key":")" + epee::to_hex::string(epee::as_byte_span(pub)) + R"(", + "index":2, + "id": {"legacy": {"amount": 0, "index": 100}}, + "tx_pub_key":")" + epee::to_hex::string(epee::as_byte_span(tx_public)) + R"(", + "rct":")" + epee::to_hex::string(epee::as_byte_span(ringct)) + R"(", + "recipient":{"maj_i":2,"min_i":66} + }, + { + "amount":30000, + "public_key":")" + epee::to_hex::string(epee::as_byte_span(pub2)) + R"(", + "index":1, + "id": {"legacy": {"amount": 0, "index": 99}}, + "tx_pub_key":")" + epee::to_hex::string(epee::as_byte_span(tx_public)) + R"(", + "rct":")" + epee::to_hex::string(epee::as_byte_span(ringct)) + R"(" + } + ], + "spends":[ + { + "id": {"legacy": {"amount": 0, "index": 90}}, + "key_image":")" + epee::to_hex::string(epee::as_byte_span(image)) + R"(" + } + ] + }] + })"; + + EXPECT(!io.stopped()); + EXPECT(error == boost::system::error_code{}); + EXPECT(get_prefix(response) == "blocks:"); + response.erase(0, std::strlen("blocks:")); + verify_json(lest_env, "blocks", 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 == ""); + + // invalid view key + message = R"(login:{"account":{"address":")" + address + R"(","view_key":")" + epee::to_hex::string(epee::as_byte_span(crypto::ec_scalar{})) + R"("}})"; + conn = pull::make(io, boost::asio::ip::tcp::endpoint(local, 10000), "lws.feed.v0.json"); + pull::async_handshake(conn, message, handler); + + ran = false; + error = {}; + response = {}; + while (!ran && !io.stopped()) + { + io.restart(); + io.run_one(); + } + + expected = R"({"msg":"Address/viewkey mismatch", "code":3})"; + + EXPECT(!io.stopped()); + EXPECT(error == boost::system::error_code{}); + EXPECT(get_prefix(response) == "error:"); + response.erase(0, std::strlen("error:")); + verify_json(lest_env, "view key error", response, expected); + + pull::async_next_message(conn, handler); + + ran = false; + error = {}; + response = {}; + while (!ran && !io.stopped()) + { + io.restart(); + io.run_one(); + } + + EXPECT(!io.stopped()); + EXPECT(error == boost::beast::websocket::error::closed); + } } } diff --git a/tests/unit/rpc/CMakeLists.txt b/tests/unit/rpc/CMakeLists.txt index f8c2e40..68c9492 100644 --- a/tests/unit/rpc/CMakeLists.txt +++ b/tests/unit/rpc/CMakeLists.txt @@ -26,7 +26,7 @@ # 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-rpc OBJECT admin.test.cpp) +add_library(monero-lws-unit-rpc OBJECT admin.test.cpp pull.test.cpp pull.test.h) target_link_libraries( monero-lws-unit-rpc monero-lws-unit-db diff --git a/tests/unit/rpc/pull.test.cpp b/tests/unit/rpc/pull.test.cpp new file mode 100644 index 0000000..794abb2 --- /dev/null +++ b/tests/unit/rpc/pull.test.cpp @@ -0,0 +1,140 @@ +// 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 "pull.test.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "error.h" +#include "misc_log_ex.h" +namespace lws_test { namespace rpc { namespace pull +{ + struct connection + { + boost::asio::io_context::strand strand_; + boost::beast::flat_buffer buffer_; + boost::beast::websocket::stream stream_; + const boost::asio::ip::tcp::endpoint remote_; + std::string out_; + + explicit connection(boost::asio::io_context& io, const boost::asio::ip::tcp::endpoint& remote) + : strand_(io), buffer_(), stream_(io), remote_(remote), out_() + {} + }; + + namespace + { + struct wrapper + { + std::shared_ptr self_; + std::function f_; + + void operator()(const boost::system::error_code error, std::size_t = {}) const + { + LWS_VERIFY(self_ && f_); + if (error) + f_(error, {}); + + const boost::beast::net::const_buffer buf = self_->buffer_.cdata(); + std::string msg{reinterpret_cast(buf.data()), buf.size()}; + self_->buffer_.consume(buf.size()); + f_(error, std::move(msg)); + } + }; + + class connector : public boost::asio::coroutine + { + wrapper self_; + + public: + explicit connector(wrapper&& self) + : boost::asio::coroutine(), self_(std::move(self)) + {} + + connector(const connector&) = default; + + void operator()(const boost::system::error_code error = {}, std::size_t = {}) + { + LWS_VERIFY(self_.self_); + connection& self = *self_.self_; + if (error) + return self_(error); + + BOOST_ASIO_CORO_REENTER(*this) + { + BOOST_ASIO_CORO_YIELD self.stream_.next_layer().async_connect(self.remote_, *this); + BOOST_ASIO_CORO_YIELD self.stream_.async_handshake("localhost", "/feed", *this); + BOOST_ASIO_CORO_YIELD self.stream_.async_write(boost::asio::const_buffer(self.out_.data(), self.out_.size()), *this); + BOOST_ASIO_CORO_YIELD self.stream_.async_read(self.buffer_, *this); + self_(error); + } + } + }; + } // anonymous + + std::shared_ptr make(boost::asio::io_context& io, const boost::asio::ip::tcp::endpoint& remote, const std::string& version) + { + namespace ws = boost::beast::websocket; + namespace http = boost::beast::http; + const auto ptr = std::make_shared(io, remote); + ptr->stream_.set_option( + ws::stream_base::decorator( + [version] (http::request_header<>& hdr) { hdr.set(http::field::sec_websocket_protocol, version); } + ) + ); + return ptr; + } + + void async_handshake(std::shared_ptr self, std::string init_message, std::function resp) + { + LWS_VERIFY(self); + self->out_ = std::move(init_message); + connector{wrapper{self, std::move(resp)}}(); + } + + void async_next_message(std::shared_ptr self, std::function resp) + { + LWS_VERIFY(self); + self->stream_.async_read(self->buffer_, wrapper{self, std::move(resp)}); + } + + void async_close(std::shared_ptr self, std::function resp) + { + LWS_VERIFY(self); + self->stream_.async_close(boost::beast::websocket::close_reason(), wrapper{self, std::move(resp)}); + } +}}} // lws_Test // rpc // pull diff --git a/tests/unit/rpc/pull.test.h b/tests/unit/rpc/pull.test.h new file mode 100644 index 0000000..81b8bbf --- /dev/null +++ b/tests/unit/rpc/pull.test.h @@ -0,0 +1,49 @@ +// 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 +#include +#include +#include +#include +#include +#include + +namespace lws_test { namespace rpc { namespace pull +{ + using response = void(boost::system::error_code, std::string); + struct connection; + + std::shared_ptr make(boost::asio::io_context& io, const boost::asio::ip::tcp::endpoint& remote, const std::string& version); + + void async_handshake(std::shared_ptr self, std::string init_message, std::function resp); + void async_next_message(std::shared_ptr self, std::function resp); + + void async_close(std::shared_ptr self, std::function resp); +}}} // lws_test // rpc // pull diff --git a/tests/unit/scanner.test.cpp b/tests/unit/scanner.test.cpp index 7341615..9cf42bc 100644 --- a/tests/unit/scanner.test.cpp +++ b/tests/unit/scanner.test.cpp @@ -332,7 +332,7 @@ LWS_CASE("lws::scanner::sync and lws::scanner::run") SETUP("lws::rpc::context, ZMQ_REP Server, and lws::db::storage") { auto rpc = - lws::rpc::context::make(lws_test::rpc_rendevous, {}, {}, {}, std::chrono::minutes{0}, false); + lws::rpc::context::make(lws_test::rpc_rendevous, {}, {}, {}, std::chrono::minutes{0}, false, true); lws::db::test::cleanup_db on_scope_exit{};