Implemented websocket "/feed" (i.e "push" updates) on HTTP (REST) API (#237)

This commit is contained in:
Lee *!* Clagett
2026-05-22 23:45:34 -04:00
committed by Lee *!* Clagett
parent a3abb637af
commit d2d03b5c6f
40 changed files with 3321 additions and 307 deletions
+258
View File
@@ -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
}
```
+147
View File
@@ -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"]
}
}
}
+21
View File
@@ -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"]
}
+10
View File
@@ -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"]
}
+36
View File
@@ -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"
}
}
}
+33
View File
@@ -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"
]
}
+20
View File
@@ -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"]
}
+12
View File
@@ -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"]
}
+10
View File
@@ -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:
+1
View File
@@ -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
+9 -3
View File
@@ -195,7 +195,7 @@ namespace
{
std::shared_ptr<lws::rpc::scanner::client> client_;
bool operator()(boost::asio::io_context&, lws::rpc::client&, net::http::client&, epee::span<const crypto::hash> chain, epee::span<const lws::account> users, epee::span<const lws::db::pow_sync> pow)
bool operator()(boost::asio::io_context&, lws::rpc::client&, net::http::client&, epee::span<const crypto::hash> chain, epee::span<lws::account> users, epee::span<const lws::db::pow_sync> 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<crypto::hash> 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<lws::account> 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};
+4
View File
@@ -1,5 +1,6 @@
#pragma once
#include <chrono>
#include <cstdint>
#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;
+25 -3
View File
@@ -51,6 +51,17 @@ namespace lws
return std::memcmp(std::addressof(lhs), std::addressof(rhs), sizeof(lhs)) < 0;
}
};
template<typename F, typename T>
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
+33 -2
View File
@@ -26,6 +26,7 @@
// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#pragma once
#include <array>
#include <boost/optional/optional.hpp>
#include <cstdint>
#include <memory>
@@ -40,7 +41,25 @@
#include "wire/msgpack/fwd.h"
namespace lws
{
{
//! Sent internallly, same wire format as `account::update`
struct mempool_receive
{
std::array<db::output, 1> outputs;
db::block_id old_height;
db::block_id new_height;
std::array<db::spend, 0> 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<db::spend> spends;
std::vector<db::output> 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;
+2
View File
@@ -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;
+12
View File
@@ -26,6 +26,8 @@
// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#include "error.h"
#include <cstring>
#include <stdexcept>
#include <string>
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{};
+8
View File
@@ -29,6 +29,13 @@
#include <system_error>
#include <type_traits>
#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
+59 -217
View File
@@ -47,6 +47,7 @@
#include <boost/beast/http/verb.hpp>
#include <boost/beast/http/write.hpp>
#include <boost/beast/version.hpp>
#include <boost/beast/websocket/rfc6455.hpp>
#include <boost/optional/optional.hpp>
#include <boost/range/counting_range.hpp>
#include <boost/thread/lock_guard.hpp>
@@ -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<db::output::spend_meta_>::const_iterator
find_metadata(std::vector<db::output::spend_meta_> 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<std::pair<db::account, db::storage_reader>> 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<response> handle(const request& req, connection_data& data, std::function<async_complete>&&)
{
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<MONERO_FIELD(db::output, unlock_time)>(), last->id, output.get_value<MONERO_FIELD(db::output, link)>().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<response> handle(const request& req, connection_data& data, std::function<async_complete>&&)
{
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<db::block_info> 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<db::output::spend_meta_> 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<MONERO_FIELD(db::output, link)>();
if (!spend.is_end())
next_spend = spend.get_value<MONERO_FIELD(db::spend, link)>();
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<MONERO_FIELD(db::output, spend_meta.amount)>();
resp.transactions.back().info.spend_meta.amount += amount;
}
const db::output::spend_meta_ meta = output.get_value<MONERO_FIELD(db::output, spend_meta)>();
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<MONERO_FIELD(db::output, link)>();
}
else if (output.is_end() || (next_spend < next_output))
{
const db::output_id source_id = spend.get_value<MONERO_FIELD(db::spend, source)>();
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<MONERO_FIELD(db::spend, link)>();
}
}
return resp;
return response::load(user->second, user->first, false);
}
};
@@ -1066,7 +911,7 @@ namespace lws
static expect<response> handle(request const& req, connection_data& data, std::function<async_complete>&&)
{
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<response> handle(request req, connection_data& data, std::function<async_complete>&&)
{
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<daemon_status>, 1024, true},
{"/feed", nullptr, 0, false},
{"/get_address_info", call<get_address_info>, 2 * 1024, false},
{"/get_address_txs", call<get_address_txs>, 2 * 1024, false},
{"/get_random_outs", call<get_random_outs>, 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<bool is_request, typename Fields>
explicit writer(boost::beast::http::header<is_request, Fields> const&, value_type const& body)
: body_(body.clone())
{}
void init(boost::beast::error_code& ec)
{
ec = {};
}
boost::optional<std::pair<const_buffers_type, bool>> 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<F>(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<F>(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<F>(resume));
}
if (handler->run == nullptr)
return self_->bad_request(boost::beast::http::status::not_implemented, std::forward<F>(resume));
@@ -2399,7 +2241,7 @@ namespace lws
}
rest_server::rest_server(epee::span<const std::string> addresses, std::vector<std::string> admin, db::storage disk, rpc::client client, configuration config)
: global_(std::make_unique<rest_server_data>(std::move(disk), std::move(client), runtime_options{config.max_subaddresses, config.webhook_verify, config.disable_admin_auth, config.auto_accept_creation, config.auto_accept_import})),
: global_(std::make_unique<rest_server_data>(std::move(disk), std::move(client), runtime_options{config})),
ports_(),
workers_()
{
+1
View File
@@ -60,6 +60,7 @@ namespace lws
{
epee::net_utils::ssl_authentication_t auth;
std::vector<std::string> access_controls;
std::chrono::seconds feed_timeout;
std::size_t threads;
std::uint32_t max_subaddresses;
epee::net_utils::ssl_verification_t webhook_verify;
+3 -3
View File
@@ -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})
+161 -4
View File
@@ -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 <amqp.h>
@@ -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<std::uint32_t> 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<std::uint32_t> 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<net::zmq::socket> client::make_daemon(const std::shared_ptr<detail::context>& ctx) noexcept
{
assert(ctx != nullptr);
@@ -481,6 +502,19 @@ namespace rpc
return {lws::error::bad_daemon_response};
}
expect<account_sub> 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<net::zmq::async_client> client::make_async_client(boost::asio::io_context& io) const
{
MONERO_PRECOND(ctx != nullptr);
@@ -558,6 +592,114 @@ namespace rpc
return rc;
}
namespace
{
template<typename T>
expect<void> 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<typename T>
expect<epee::byte_slice> 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<void> 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<boost::mutex> lock{ctx->sync_internal_pub};
net::zmq::send(std::move(*msg), ctx->internal_pub.get(), ZMQ_DONTWAIT);
}
return success();
}
expect<void> client::push_updates(const epee::span<account> 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<std::size_t> offsets;
offsets.reserve(accts.size());
epee::byte_stream sink;
std::size_t last = 0;
expect<void> 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<void> 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<boost::mutex> 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<void> push_warning_(const std::shared_ptr<detail::context>& ctx, const std::error_code error, const db::block_id height)
{
MONERO_PRECOND(ctx != nullptr);
if (ctx->internal_pub)
{
const boost::lock_guard<boost::mutex> lock{ctx->sync_internal_pub};
expect<epee::byte_slice> 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<void> client::push_warning(const std::error_code error, const db::block_id height) const
{
return push_warning_(ctx, error, height);
}
expect<rates> 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<detail::context>(
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<void> context::push_warning(const std::error_code error) const
{
return push_warning_(ctx, error, db::block_id::txpool);
}
expect<void> context::raise_abort_scan() noexcept
{
MONERO_PRECOND(ctx != nullptr);
+45 -2
View File
@@ -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<void> parse_response(cryptonote::rpc::Message& parser, std::string msg, source_location loc = {});
class account_sub
{
std::shared_ptr<const detail::context> ctx;
net::zmq::async_client sub;
public:
account_sub(std::shared_ptr<const detail::context> 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<std::uint32_t> 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<account_sub> make_account_sub(boost::asio::io_context& io) const;
//! \return `async_client` to daemon. Thread safe.
expect<net::zmq::async_client> make_async_client(boost::asio::io_context& io) const;
@@ -163,6 +199,10 @@ namespace rpc
return publish(epee::byte_slice{std::move(bytes)});
}
expect<void> push_update(const account& acct, const mempool_receive& src) const;
expect<void> push_updates(epee::span<account> accts, db::block_id new_height) const;
expect<void> push_warning(std::error_code error, db::block_id height) const;
//! \return Next available RPC message response from server
expect<std::string> get_message(std::chrono::seconds timeout);
@@ -179,7 +219,7 @@ namespace rpc
\return Recent exchange rates. */
expect<rates> 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<void> 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
+85
View File
@@ -0,0 +1,85 @@
// Copyright (c) 2026, The Monero Project
// All rights reserved.
//
// Redistribution and use in source and binary forms, with or without modification, are
// permitted provided that the following conditions are met:
//
// 1. Redistributions of source code must retain the above copyright notice, this list of
// conditions and the following disclaimer.
//
// 2. Redistributions in binary form must reproduce the above copyright notice, this list
// of conditions and the following disclaimer in the documentation and/or other
// materials provided with the distribution.
//
// 3. Neither the name of the copyright holder nor the names of its contributors may be
// used to endorse or promote products derived from this software without specific
// prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY
// EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
// MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
// THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
// STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#pragma once
#include <boost/asio/io_context.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/ssl/stream.hpp>
#include <boost/beast/http/message.hpp>
#include <boost/beast/http/string_body.hpp>
#include <chrono>
#include <cstdint>
#include <system_error>
#include "db/fwd.h"
#include "rpc/fwd.h"
#include "wire/fwd.h"
namespace lws { namespace rpc { namespace feed
{
using request =
boost::beast::http::request<boost::beast::http::string_body>;
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<boost::asio::ip::tcp::socket>&& sock, boost::asio::io_context& io, const lws::rpc::client& client, const request& req, const lws::db::storage& disk, std::chrono::seconds idle_timeout);
}}} // lws // rpc // feed
namespace std
{
template<>
struct is_error_code_enum<::lws::rpc::feed::error>
: true_type
{};
}
+600
View File
@@ -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 <array>
#include <boost/asio/bind_executor.hpp>
#include <boost/asio/dispatch.hpp>
#include <boost/asio/steady_timer.hpp>
#include <boost/asio/strand.hpp>
#include <boost/beast/core/error.hpp>
#include <boost/beast/core/flat_buffer.hpp>
#include <boost/beast/core/string_type.hpp>
#include <boost/beast/http/message.hpp>
#include <boost/beast/http/status.hpp>
#include <boost/beast/http/write.hpp>
#include <boost/beast/version.hpp>
#include <boost/beast/websocket/stream.hpp>
#include <chrono>
#include <type_traits>
#include <utility>
#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<lws::rpc::feed::event>
: 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<int>(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<connection> self_;
std::size_t iteration_;
explicit zmq_loop(std::shared_ptr<connection> 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<epee::byte_slice> prep_error(std::error_code error, protocol proto);
expect<epee::byte_slice> prep_login(const db::storage& disk, account_sub* sub, connection_sync* sync, boost::beast::flat_buffer& buffer, protocol proto);
expect<epee::byte_slice> prep_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<epee::byte_slice> 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<connection> self) = 0;
protected:
template<typename T>
void do_async_close(T& sock, std::shared_ptr<connection> 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<connection>)
{
do_cleanup(sock);
}
void do_async_shutdown(boost::asio::ssl::stream<boost::asio::ip::tcp::socket>& sock, std::shared_ptr<connection> 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<boost::asio::ip::tcp::socket>& sock)
{
do_cleanup(sock.next_layer());
}
template<typename T>
void do_stop_reads(T& sock, std::shared_ptr<connection> 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<typename T>
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<typename T>
static void do_async_write_queue(T& sock, std::shared_ptr<connection> 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<connection> 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<connection> self, expect<epee::byte_slice>&& 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<connection> 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<connection> 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<typename F>
auto bind(F&& f) -> decltype(boost::asio::bind_executor(strand_, std::forward<F>(f)))
{
return boost::asio::bind_executor(strand_, std::forward<F>(f));
}
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<connection> 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<connection> self)
{
struct force_close
{
std::shared_ptr<connection> 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<connection> self) = 0;
virtual void cleanup() = 0;
virtual void stop_reads(std::shared_ptr<connection> self) = 0;
virtual void async_read(boost::asio::steady_timer::duration timeout, zmq_loop callback) = 0;
virtual void async_write_queue(std::shared_ptr<connection> self, epee::byte_slice&& msg) = 0;
};
template<typename T>
class connection_ final : public connection
{
boost::beast::websocket::stream<T, false> sock_;
virtual void async_shutdown(std::shared_ptr<connection> 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<T, false>& sock() { return sock_; }
virtual void async_close(std::shared_ptr<connection> 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<connection> 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<connection> 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<boost::asio::ip::tcp::socket>& sock) { return sock.next_layer(); }
template<typename T>
inline void do_start(std::shared_ptr<connection_<T>> self, const request& req, const std::chrono::seconds timeout)
{
struct on_control
{
std::weak_ptr<connection> 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_<T>& 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<typename T>
inline bool do_start(T&& sock, boost::asio::io_context& io, const lws::rpc::client& client, const request& req, const lws::db::storage& disk, const std::chrono::seconds timeout)
{
const protocol proto = get_protocol(req.base()[boost::beast::http::field::sec_websocket_protocol]);
switch (proto)
{
case protocol::v0_msgpack:
case protocol::v0_json:
do_start(std::make_shared<connection_<T>>(std::forward<T>(sock), io, client, disk, proto), req, timeout);
return true;
default:
case protocol::invalid:
break;
}
return false;
}
}}} // lws // rpc // feed
+44
View File
@@ -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 <boost/asio/ssl.hpp>
#include <boost/beast/websocket/ssl.hpp>
#include "feed.inl"
namespace lws { namespace rpc { namespace feed
{
bool start(boost::asio::ssl::stream<boost::asio::ip::tcp::socket>&& sock, boost::asio::io_context& io, const lws::rpc::client& client, const request& req, const lws::db::storage& disk, const std::chrono::seconds timeout)
{
// 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
}
}}}
+374
View File
@@ -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 <cstring>
#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<boost::beast::string_view, protocol> 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<const char*>(buf.data()), buf.size()};
}
epee::byte_slice get_slice(const boost::beast::net::const_buffer buf)
{
return epee::byte_slice{{reinterpret_cast<const std::uint8_t*>(buf.data()), buf.size()}};
}
template<typename T>
expect<epee::byte_slice> 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<std::pair<boost::beast::string_view, protocol>, 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<const char*>(buf.data());
void const* const match = std::memchr(begin, u8':', buf.size());
if (match)
return {begin, std::size_t(reinterpret_cast<const char*>(match) - begin + 1)};
return {};
}
expect<epee::byte_slice> prep_error(const std::error_code error, const protocol proto)
{
return prep(feed_error{error}, proto);
}
expect<epee::byte_slice> prep_login(const db::storage& disk, account_sub* const sub, connection_sync* const sync, boost::beast::flat_buffer& buffer, const protocol proto)
{
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<std::uint32_t> 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<epee::byte_slice> 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
+9
View File
@@ -27,11 +27,20 @@
#pragma once
#include <cstdint>
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;
+415 -3
View File
@@ -35,11 +35,13 @@
#include <type_traits>
#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 std::uint8_t> const* payment_id = nullptr;
epee::span<const std::uint8_t> 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<typename T>
struct sync_wrapper
{
const T& data;
};
struct sync_transform_
{
template<typename T>
sync_wrapper<T> 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<db::output> 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<rpc::transaction_spend> 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<rpc::get_transaction> self)
{
epee::span<const std::uint8_t> const* payment_id = nullptr;
epee::span<const std::uint8_t> 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<typename F, typename T>
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<const get_address_txs_response::transaction&> self)
static void write_bytes(wire::json_writer& dest, boost::range::index_value<const get_transaction&> self)
{
epee::span<const std::uint8_t> const* payment_id = nullptr;
epee::span<const std::uint8_t> payment_id_bytes;
@@ -311,6 +525,204 @@ namespace lws
);
}
} // rpc
std::vector<db::output::spend_meta_>::const_iterator
rpc::get_address_txs_response::find_metadata(std::vector<db::output::spend_meta_> 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<typename T>
struct vec_lmdb_
{
using iterator = typename std::vector<T>::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<typename U, typename G = U, std::size_t uoffset = 0>
G get_value() const noexcept
{
static_assert(std::is_same<U, T>(), "bad MONERO_FIELD usage?");
static_assert(std::is_trivially_copyable<U>(), "value type must be memcpy safe");
static_assert(std::is_trivially_copyable<G>(), "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<const std::uint8_t*>(std::addressof(*pos)) + uoffset, sizeof(value));
return value;
}
const value_type& operator*() const noexcept { return *pos; }
};
template<typename T>
vec_lmdb_<T> vec_lmdb(const std::vector<T>& src) { return {src.begin(), src.end()}; }
template<typename T, typename U>
std::pair<std::vector<rpc::get_transaction>, std::uint64_t> merge_into_txes(T output, U spend, const std::size_t reserve, const bool all_outputs, const bool skip_spend_meta)
{
// merge input and output info into a single set of txes.
std::uint64_t total = 0;
std::vector<rpc::get_transaction> out{};
std::vector<db::output::spend_meta_> 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<MONERO_FIELD(db::output, link)>();
if (!spend.is_end())
next_spend = spend.template get_value<MONERO_FIELD(db::spend, link)>();
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<MONERO_FIELD(db::output, spend_meta.amount)>();
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<MONERO_FIELD(db::output, spend_meta)>();
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<MONERO_FIELD(db::output, link)>();
}
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<MONERO_FIELD(db::spend, source)>();
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<MONERO_FIELD(db::spend, link)>();
}
}
return {std::move(out), total};
}
struct tx_link_sorter_
{
template<typename T>
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_transaction> rpc::get_address_txs_response::load(std::vector<db::output> outputs, std::vector<db::spend> 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> 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<db::block_info> 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));
+118 -39
View File
@@ -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<db::subaddress_dict> new_subaddrs;
std::vector<db::subaddress_dict> 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<db::output> receives;
std::vector<transaction_spend> spends;
std::uint64_t spent;
};
struct get_address_txs_response
{
get_address_txs_response() = delete;
static std::vector<db::output::spend_meta_>::const_iterator
find_metadata(std::vector<db::output::spend_meta_> const& metas, db::output_id id);
static std::vector<get_transaction> load(std::vector<db::output> outputs, std::vector<db::spend> spends);
static expect<get_address_txs_response> load(db::storage_reader& reader, const db::account& acct, const bool all_outputs);
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<get_transaction> 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<get_transaction> 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<db::subaddress_dict> new_subaddrs;
std::vector<db::subaddress_dict> 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<transaction_spend> 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<transaction> 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<std::uint32_t> n_min;
boost::optional<bool> 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;
+80
View File
@@ -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<std::pair<db::account, db::storage_reader>> 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
+45
View File
@@ -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 <utility>
#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<std::pair<db::account, db::storage_reader>> open_account(const account_credentials& creds, const db::storage& disk);
}} // lws // rpc
+2 -2
View File
@@ -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();
+43 -17
View File
@@ -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<void> pushed = client_.push_update(user, mempool_receive{out});
if (!pushed)
MERROR("Failed to send mempool update to " << user.address() << ": " << pushed.error().message());
return true;
}
};
@@ -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<lws::account> 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<std::uint64_t> 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<crypto::hash> prefix_hash;
boost::optional<cryptonote::tx_extra_nonce> extra_nonce;
@@ -356,6 +363,16 @@ namespace lws
cryptonote::tx_extra_additional_pub_keys additional_tx_pub_keys;
std::vector<crypto::key_derivation> additional_derivations;
const auto get_tx_hash = [&] () -> const crypto::hash&
{
if (!tx_hash)
{
tx_hash_lazy = get_transaction_hash(tx);
tx_hash = std::addressof(tx_hash_lazy);
}
return *tx_hash;
};
{
std::vector<cryptonote::tx_extra_field> extra;
cryptonote::parse_tx_extra(tx.extra, extra);
@@ -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<std::uint64_t> 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<lws::account> 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<db::storage>{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<thread_data> 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<crypto::hash> new_hashes{};
std::vector<db::pow_sync> 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<const crypto::hash> chain, const epee::span<const lws::account> users, const epee::span<const db::pow_sync> pow)
bool user_data::store(boost::asio::io_context& io, db::storage& disk, rpc::client& client, net::http::client& webhook, const epee::span<const crypto::hash> chain, const epee::span<lws::account> users, const epee::span<const db::pow_sync> 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<const crypto::hash> chain, const epee::span<const lws::account> users, const epee::span<const db::pow_sync> pow)
bool user_data::operator()(boost::asio::io_context& io, rpc::client& client, net::http::client& webhook, const epee::span<const crypto::hash> chain, const epee::span<lws::account> users, const epee::span<const db::pow_sync> 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);
+3 -3
View File
@@ -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<const crypto::hash> chain, epee::span<const lws::account> users, epee::span<const db::pow_sync> pow);
static bool store(boost::asio::io_context& io, db::storage& disk, rpc::client& zclient, net::http::client& webhook ,epee::span<const crypto::hash> chain, epee::span<lws::account> users, epee::span<const db::pow_sync> 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<const crypto::hash> chain, epee::span<const lws::account> users, epee::span<const db::pow_sync> pow);
bool operator()(boost::asio::io_context& io, rpc::client& zclient, net::http::client& webhook, epee::span<const crypto::hash> chain, epee::span<lws::account> users, epee::span<const db::pow_sync> 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<bool(boost::asio::io_context&, rpc::client&, net::http::client&, epee::span<const crypto::hash>, epee::span<const lws::account>, epee::span<const db::pow_sync>)>;
using store_func = std::function<bool(boost::asio::io_context&, rpc::client&, net::http::client&, epee::span<const crypto::hash>, epee::span<lws::account>, epee::span<const db::pow_sync>)>;
/*! Run _just_ the inner scanner loop while `self.is_running() == true`.
*
+13 -1
View File
@@ -92,6 +92,7 @@ namespace
const command_line::arg_descriptor<double> split_sync_threads;
const command_line::arg_descriptor<std::uint64_t> split_sync_depth;
const command_line::arg_descriptor<bool> balance_new_addresses;
const command_line::arg_descriptor<std::chrono::seconds::rep> 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};
+389 -6
View File
@@ -27,7 +27,11 @@
#include "framework.test.h"
#include <boost/asio/ip/tcp.hpp>
#include <boost/beast/websocket/error.hpp>
#include <memory>
#include <optional>
#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<StringBuffer> 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<std::string, unsigned> 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<std::string> 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<pull::connection> 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<crypto::hash>()
};
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<crypto::hash>();
const crypto::public_key pub = crypto::rand<crypto::public_key>();
const crypto::public_key pub2 = crypto::rand<crypto::public_key>();
const rct::key ringct = crypto::rand<rct::key>();
const auto extra = lws::db::extra(lws::db::extra::ringct_output);
const auto payment_id_ = crypto::rand<lws::db::output::payment_id_>();
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<crypto::key_image>();
{
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<crypto::hash, 2> chain{{
last_block.hash, crypto::rand<crypto::hash>()
}};
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);
}
}
}
+1 -1
View File
@@ -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
+140
View File
@@ -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 <boost/asio/bind_executor.hpp>
#include <boost/asio/coroutine.hpp>
#include <boost/asio/strand.hpp>
#include <boost/beast/core/error.hpp>
#include <boost/beast/core/flat_buffer.hpp>
#include <boost/beast/core/string_type.hpp>
#include <boost/beast/http/message.hpp>
#include <boost/beast/http/status.hpp>
#include <boost/beast/http/write.hpp>
#include <boost/beast/version.hpp>
#include <boost/beast/websocket/stream.hpp>
#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<boost::asio::ip::tcp::socket> 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<connection> self_;
std::function<response> 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<const char*>(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<connection> 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<connection>(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<connection> self, std::string init_message, std::function<response> resp)
{
LWS_VERIFY(self);
self->out_ = std::move(init_message);
connector{wrapper{self, std::move(resp)}}();
}
void async_next_message(std::shared_ptr<connection> self, std::function<response> resp)
{
LWS_VERIFY(self);
self->stream_.async_read(self->buffer_, wrapper{self, std::move(resp)});
}
void async_close(std::shared_ptr<connection> self, std::function<response> resp)
{
LWS_VERIFY(self);
self->stream_.async_close(boost::beast::websocket::close_reason(), wrapper{self, std::move(resp)});
}
}}} // lws_Test // rpc // pull
+49
View File
@@ -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 <boost/asio/ip/tcp.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/system/error_code.hpp>
#include <functional>
#include <memory>
#include <string>
#include <string_view>
namespace lws_test { namespace rpc { namespace pull
{
using response = void(boost::system::error_code, std::string);
struct connection;
std::shared_ptr<connection> make(boost::asio::io_context& io, const boost::asio::ip::tcp::endpoint& remote, const std::string& version);
void async_handshake(std::shared_ptr<connection> self, std::string init_message, std::function<response> resp);
void async_next_message(std::shared_ptr<connection> self, std::function<response> resp);
void async_close(std::shared_ptr<connection> self, std::function<response> resp);
}}} // lws_test // rpc // pull
+1 -1
View File
@@ -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{};