Implemented websocket "/feed" (i.e "push" updates) on HTTP (REST) API (#237)
What changed, and why it matters
This commit adds a new real-time WebSocket '/feed' endpoint to the Monero Light Wallet Server. It lets wallet clients receive live transaction and block updates instead of polling. The change is large (+3,300 lines) and touches networking, authentication, serialization, and database access. It also refactors existing account-opening code so it can be shared between the old REST API and the new feed. There is no indication in the commit that this is a security fix; it appears to be a feature implementation. Because it is brand-new network-facing code, it increases the attack surface, but the diff itself does not contain an obvious, directly exploitable vulnerability.
Treat this as a feature commit rather than a security patch. Reviewers should focus on the new WebSocket feed code for concurrency bugs, authentication bypasses, and resource exhaustion (queue limits, timeouts, memory growth). Pay special attention to the refactored rpc::open_account and key_check helpers to ensure existing REST endpoints retain their original authorization semantics. Fuzzing the msgpack/json parsers and stress-testing the queue_max and timeout paths would be valuable. No immediate emergency action is warranted based solely on this diff.
Security signals we found
New network-facing WebSocket endpoint increases attack surface
Authentication relies on existing address+view_key credentials
Refactored shared account-opening helpers may change authorization behavior for existing REST endpoints
Large feature commit (+3321/-307) with many new code paths and concurrency interactions
No explicit security fix or vulnerability disclosure language in commit message or diff
Evidence from the diff
The commit implements a WebSocket ‘/feed’ push API on top of the existing HTTP REST server. It adds protocol negotiation (lws.feed.v0.json / lws.feed.v0.msgpack), login via address+view_key, an initial tx-sync response, and subsequent pushed blocks/mempool/warning/error messages. Internally it introduces an inproc ZMQ PUB/SUB channel so the scanner thread can push account updates to the WebSocket handler thread. The feed logic is split across src/rpc/feed.h, feed.inl, feed_tcp.cpp, and feed_ssl.cpp. Existing REST handlers were refactored to use shared helpers such as rpc::open_account, rpc::key_check, and get_address_txs_response::load. A new runtime option feed_timeout (default 4 minutes) controls whether the endpoint is enabled. The code uses Boost.Beast for WebSocket I/O and strands for serialization.
Changed components
src/rest_server.cppsrc/rest_server.hsrc/rpc/feed.hsrc/rpc/feed.inlsrc/rpc/feed_tcp.cppsrc/rpc/feed_ssl.cppsrc/rpc/client.cppsrc/rpc/client.hsrc/rpc/light_wallet.cppsrc/rpc/light_wallet.hsrc/db/account.cppsrc/db/account.hsrc/scanner.cppsrc/scanner.hInspect captured patch +3321 / −307
diff --git a/docs/api/feed.md b/docs/api/feed.md
new file mode 100644
index 0000000..8867243
--- /dev/null
+++ b/docs/api/feed.md
@@ -0,0 +1,258 @@
+# '/feed' API
+
+The `/feed` websocket API allows for faster synchronization on initalization (msgpack
+with less redundant data compared to [wallet api](/api/wallet/)), and "push"
+updates instead of polling. New block arrivals, new received/spent transactions,
+and reorgs are all pushed to the client in real-time. This should provide an
+enterprise-like experience where the (usually mobile) client is seamlessly
+synchronized with `monerod` (but only if `monero-lws-daemon` is using ZMQ
+PUB/SUB from `monerod`).
+
+## Basics
+The client initiates a `/feed` websocket connection, then the LWS server sends
+the entire tx history, and then finally sends a continous stream of updates from
+`monerod` (ZMQ PUB mempool+blocks messages) to the client.
+
+## Flow
+### Handshake
+The client must attempt a websocket "upgrade" connection to `/feed`. The
+initial handshake must have a HTTP/1.1 field, `Sec-WebSocket-Protocol`, which
+is set to msgpack or json:
+
+```
+Sec-WebSocket-Protocol: lws.feed.v0.msgpack
+Sec-WebSocket-Protocol: lws.feed.v0.json
+```
+
+This parameter signals the protocol version and payload type. All message
+payloads must match the type set in the header. Msgpack is recommended as both
+serialization and de-serialization is faster, and the messages are more
+compact (especially since binary crypto blobs are not encoded as hex).
+
+### Synchronization
+After the websocket handshake, the client must send a [login](#login) message,
+and the server must respond with a [tx-sync](#tx-sync) message OR
+[error](#error) message. An `error` message signals the websocket is now in the
+"terminated" state, a `tx-sync` message indicates that server is now ready to
+send real-time updates. In either case, the client shall not send any further
+data - the server will terminate the connection if this is not followed.
+
+### Updates
+After the `tx-sync`, the server will then push [blocks](#blocks),
+[mempool](#mempool), [warning](#warning), or [error](#error) messages until the
+stream is terminated. An `error` message indicates the stream is immediately
+terminated, whereas the other messages indicate a valid stream where clients
+should expected further messages.
+
+## Message Format
+Every message has a prefix before the msgpack/json payload that indicates the
+type. Current valid prefixes are `login:`, `tx-sync:`, `blocks:`, `mempool:`,
+`warning:`, and `error:`. The `login:` prefix is from the client, and the
+remainder are sent by the server. The root type of every payload is an object.
+As an example, a client login message (in the json sub-protocol), will look
+like:
+
+```json
+login:{
+ "account":{
+ "address": "47nPhxp2cJeKN2NjamupNUNA13XgzcYPzQBCzzsKcj717s8M2UpFVmmdwSuYwgyy8kPDwU7hpEqTTDvfe5LAb9Aj6nwmEzf",
+ "view_key": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205"
+ }
+}
+```
+This format (with prefixes) allows for single-pass DOMless parsing.
+
+### login
+Messages of this type are sent (only) by the client. The message contains
+proof the client has authority to receives updates for a given primary address,
+along with optional restrictions on data sent/pushed by the server.
+
+[schema](schemas/feed_login.json)
+
+#### Example
+```json
+login:{
+ "account":{
+ "address": "47nPhxp2cJeKN2NjamupNUNA13XgzcYPzQBCzzsKcj717s8M2UpFVmmdwSuYwgyy8kPDwU7hpEqTTDvfe5LAb9Aj6nwmEzf",
+ "view_key": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205"
+ },
+ "tx_sync": false,
+ "receives_only": true
+}
+```
+
+### tx-sync
+Messages of this type are sent only by the server, and only in response to a
+successful `login:` command from the client. The message contains everything
+needed for the client to have a full tx synchronization with the backend
+(unless the client requested no transactions).
+
+[schema](schemas/feed_tx_sync.json)
+
+#### Example
+```json
+tx_sync:{
+ "scanned_block_height": 10,
+ "blockchain_height": 10,
+ "lookahead_fail": 5,
+ "lookahead": {"maj_i": 50, "min_i": 200 },
+ "transactions": [
+ {
+ "hash": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205",
+ "prefix_hash": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205",
+ "timestamp": 0,
+ "fee": 0,
+ "unlock_time": 0,
+ "mixin": 0,
+ "payment_id": "ee171270296bbc26",
+ "coinbase": true,
+ "mempool": true,
+ "receives": [
+ {
+ "id": {"legacy": {"amount": 0, "index": 0}},
+ "amount": 0,
+ "public_key": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205",
+ "index": 0,
+ "tx_pub_key": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205",
+ "rct": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205",
+ "recipient": {"maj_i": 1, "min_i": 1}
+ }
+ ],
+ "spends": [
+ {
+ "id": {"unified": 0},
+ "key_image": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205"
+ }
+ ]
+ }
+ ]
+}
+```
+
+### blocks
+Messages of this type are sent only by the server when a new block has been
+scanned on the account. If another block was added to the Monero chain, this
+message type is not sent until a new block has been scanned against the
+account. This will also give real-time updates when account is scanning
+historical blocks (although the server can choose to group/aggregate messages
+to reduce total number of messages).
+
+[schema](schemas/feed_blocks.json)
+
+#### Example
+```json
+blocks:{
+ "scan_start": 150,
+ "scan_end": 151,
+ "blockchain_height": 151,
+ "lookahead_fail": 0,
+ "lookahead": {"maj_i": 50, "min_i": 200 },
+ "transactions": [
+ {
+ "hash": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205",
+ "prefix_hash": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205",
+ "timestamp": 0,
+ "fee": 0,
+ "unlock_time": 0,
+ "mixin": 0,
+ "payment_id": "ee171270296bbc26",
+ "coinbase": true,
+ "receives": [
+ {
+ "id": {"legacy": {"amount": 0, "index": 0}},
+ "amount": 0,
+ "public_key": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205",
+ "index": 0,
+ "tx_pub_key": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205",
+ "rct": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205",
+ "recipient": {"maj_i": 1, "min_i": 1}
+ }
+ ],
+ "spends": [
+ {
+ "id": {"unified": 0},
+ "key_image": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205"
+ }
+ ]
+ }
+ ]
+}
+```
+
+### mempool
+If the account scanner has caught-up with the most recent block, the server may
+send _received outputs_ that are still in the mempool. These outputs will be
+re-sent in the `blocks:` message in a transaction; the ID numbers for outputs
+are not assigned until the transaction is included in a block. Spends are
+assumed to be under the control of the client, and are not sent in mempool
+messages.
+
+[schema](schemas/feed_mempool.json)
+
+#### Full Example
+```json
+mempool:{
+ "hash": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205",
+ "prefix_hash": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205",
+ "timestamp": 0,
+ "fee": 0,
+ "unlock_time": 0,
+ "mixin": 0,
+ "payment_id": "ee171270296bbc26",
+ "amount": 0,
+ "public_key": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205",
+ "tx_pub_key": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205",
+ "index": 0,
+ "recipient": {"maj_i": 1, "min_i": 1},
+ "rct": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205"
+}
+```
+
+### warning
+The primary purpose of this message type is to alert the client that the server
+has detected a block re-organization. The rollback point is transmitted in the
+message, and the client is expected to rollback transactions that are part of
+the re-organization. Other events, such as lost connection to `monerod` are also
+sent as warnings. Most other events will be [error](#error).
+
+[schema](schemas/feed_warning.json)
+
+#### Example
+```json
+warning:{
+ "msg": "Descriptive warning message",
+ "code": 4,
+ "counter": 10,
+ "height": 1000
+}
+```
+
+#### Codes
+```c++
+ unspecified_error = 0,
+ account_not_found = 1,
+ bad_address = 2,
+ bad_view_key = 3,
+ blockchain_reorg = 4,
+ daemon_unresponsive = 5,
+ parse_error = 6,
+ protocol_error = 7,
+ queue_error = 8,
+ schema_error = 9
+```
+
+### error
+Servers will send an error of this type when the websocket must be closed; no
+further messages are to be sent after sending this type. The codes sent in this
+message type are identical to the ones [specified aboved](#codes) for warning
+messages.
+
+[schema](schemas/feed_error.json)
+
+#### Example
+```json
+error:{
+ "msg": "Descriptive error message",
+ "code": 8
+}
+```
diff --git a/docs/api/schemas/feed_base.json b/docs/api/schemas/feed_base.json
new file mode 100644
index 0000000..87f07bf
--- /dev/null
+++ b/docs/api/schemas/feed_base.json
@@ -0,0 +1,147 @@
+{
+ "$schema": "https://json-schema.org/draft-07/schema",
+ "$id": "https://docs.monerolws.com/api/schemas/feed_base.json",
+ "definitions": {
+ "address_meta": {
+ "type": "object",
+ "properties": {
+ "maj_i": {"$ref": "#/definitions/uint32"},
+ "min_i": {"$ref": "#/definitions/uint32"}
+ },
+ "required": ["maj_i", "min_i"]
+ },
+ "binary8": {
+ "type": "string",
+ "minLength": 16,
+ "maxLength": 16,
+ "pattern": "^[0-9A-Fa-f]{16}$",
+ "example": "ee171270296bbc26"
+ },
+ "binary32": {
+ "type": "string",
+ "minLength": 64,
+ "maxLength": 64,
+ "pattern": "^[0-9A-Fa-f]{64}$",
+ "example": "ee171270296bbc26d5be4455b6313e1a2086a92080e205f77d6f861f8e5fd205"
+ },
+ "composite_id": {
+ "anyOf": [
+ {"$ref": "#/definitions/legacy_id"},
+ {"$ref": "#/definitions/unified_id"}
+ ]
+ },
+ "int64": {
+ "type": "integer",
+ "minimum": -9223372036854775808,
+ "maximum": 9223372036854775807
+ },
+ "legacy_id": {
+ "type": "object",
+ "properties": {
+ "legacy": {
+ "type": "object",
+ "properties": {
+ "amount": {"$ref": "#/definitions/uint64"},
+ "index": {"$ref": "#/definitions/uint64"}
+ },
+ "required": ["amount", "index"]
+ }
+ },
+ "required": ["legacy"]
+ },
+ "receive" : {
+ "type": "object",
+ "properties": {
+ "amount": {"$ref": "#/definitions/uint64"},
+ "public_key": {"$ref": "#/definitions/binary32"},
+ "index": {"$ref": "#/definitions/uint32"},
+ "tx_pub_key": {"$ref": "#/definitions/binary32"},
+ "id": {"$ref": "#/definitions/composite_id"},
+ "rct": {"$ref": "#/definitions/binary32"},
+ "recipient": {
+ "$ref": "#/definitions/address_meta",
+ "default": "{0, 0}"
+ }
+ },
+ "required": [
+ "amount",
+ "public_key",
+ "index",
+ "tx_pub_key"
+ ]
+ },
+ "spend": {
+ "type": "object",
+ "properties": {
+ "id": {"$ref": "#/definitions/composite_id"},
+ "key_image": {"$ref": "#/definitions/binary32"}
+ },
+ "required": ["key_image"]
+ },
+ "transaction": {
+ "type": "object",
+ "properties": {
+ "hash": {"$ref": "#/definitions/binary32"},
+ "prefix_hash": {"$ref": "#/definitions/binary32"},
+ "timestamp": {"$ref": "#/definitions/int64"},
+ "fee": {"$ref": "#/definitions/uint64"},
+ "unlock_time": {"$ref": "#/definitions/uint64"},
+ "mixin": {"$ref": "#/definitions/uint32"},
+ "height": {"$ref": "#/definitions/uint64"},
+ "payment_id": {
+ "anyOf": [
+ {"$ref": "#/definitions/binary8"},
+ {"$ref": "#/definitions/binary32"}
+ ]
+ },
+ "coinbase": {
+ "type": "boolean",
+ "default": false
+ },
+ "mempool": {
+ "type": "boolean",
+ "default": false
+ },
+ "spends": {
+ "type": "array",
+ "minItems": 1,
+ "items": {"$ref" : "#/definitions/spend"}
+ },
+ "receives": {
+ "type": "array",
+ "minItems": 1,
+ "items": {"$ref" : "#/definitions/receive"}
+ }
+ },
+ "required": [
+ "hash",
+ "prefix_hash",
+ "timestamp",
+ "fee",
+ "unlock_time"
+ ]
+ },
+ "uint16": {
+ "type": "integer",
+ "minimum": 0,
+ "maximum": 65535
+ },
+ "uint32": {
+ "type": "integer",
+ "minimum": 0,
+ "maximum": 4294967295
+ },
+ "uint64": {
+ "type": "integer",
+ "minimum": 0,
+ "maximum": 18446744073709551615
+ },
+ "unified_id": {
+ "type": "object",
+ "properties": {
+ "unified": {"$ref" : "#/definitions/uint64"}
+ },
+ "required": ["unified"]
+ }
+ }
+}
diff --git a/docs/api/schemas/feed_blocks.json b/docs/api/schemas/feed_blocks.json
new file mode 100644
index 0000000..fecbf0d
--- /dev/null
+++ b/docs/api/schemas/feed_blocks.json
@@ -0,0 +1,21 @@
+{
+ "$schema": "https://json-schema.org/draft-07/schema",
+ "$id": "https://docs.monerolws.com/api/schemas/feed_blocks.json",
+ "type": "object",
+ "properties": {
+ "scan_start": {"$ref": "feed_base.json#/definitions/uint64"},
+ "scan_end": {"$ref": "feed_base.json#/definitions/uint64"},
+ "blockchain_height": {"$ref": "feed_base.json#/definitions/uint64"},
+ "lookahead_fail": {"$ref": "feed_base.json#/definitions/uint64"},
+ "lookahead": {
+ "$ref": "feed_base.json#/definitions/address_meta",
+ "default": "{0, 0}"
+ },
+ "transactions": {
+ "type": "array",
+ "minItems": 1,
+ "items": {"$ref": "feed_base.json#/definitions/transaction"}
+ }
+ },
+ "required": ["scan_start", "scan_end", "blockchain_height"]
+}
diff --git a/docs/api/schemas/feed_error.json b/docs/api/schemas/feed_error.json
new file mode 100644
index 0000000..4b95338
--- /dev/null
+++ b/docs/api/schemas/feed_error.json
@@ -0,0 +1,10 @@
+{
+ "$schema": "https://json-schema.org/draft-07/schema",
+ "$id": "https://docs.monerolws.com/api/schemas/feed_error.json",
+ "type": "object",
+ "properties": {
+ "msg": {"type": "string"},
+ "code": {"$ref": "feed_base.json#/definitions/uint16"}
+ },
+ "required": ["msg", "code"]
+}
diff --git a/docs/api/schemas/feed_login.json b/docs/api/schemas/feed_login.json
new file mode 100644
index 0000000..02dd4d2
--- /dev/null
+++ b/docs/api/schemas/feed_login.json
@@ -0,0 +1,36 @@
+{
+ "$schema": "https://json-schema.org/draft-07/schema",
+ "$id": "https://docs.monerolws.com/api/schemas/feed_login.json",
+ "type": "object",
+ "properties": {
+ "account": {
+ "type": "object",
+ "properties": {
+ "address": {"$ref": "#/definitions/base58-address"},
+ "view_key": {"$ref": "feed_base.json#/definitions/binary32"}
+ },
+ "required": ["address", "view_key"]
+ },
+ "tx_sync": {
+ "description": "Whether to include txes in 'tx_sync:' response",
+ "type": "boolean",
+ "default": true
+ },
+ "receives_only": {
+ "description": "Send only received outputs messages after 'tx_sync:' message",
+ "type": "boolean",
+ "default": false
+ }
+ },
+ "required": ["account"],
+ "definitions": {
+ "base58-address": {
+ "type": "string",
+ "minLength": 95,
+ "maxLength": 95,
+ "pattern": "^[0-9A-Za-z]{95}$",
+ "example": "47nPhxp2cJeKN2NjamupNUNA13XgzcYPzQBCzzsKcj717s8M2UpFVmmdwSuYwgyy8kPDwU7hpEqTTDvfe5LAb9Aj6nwmEzf"
+ }
+ }
+}
+
diff --git a/docs/api/schemas/feed_mempool.json b/docs/api/schemas/feed_mempool.json
new file mode 100644
index 0000000..dbce94e
--- /dev/null
+++ b/docs/api/schemas/feed_mempool.json
@@ -0,0 +1,33 @@
+{
+ "$schema": "https://json-schema.org/draft-07/schema",
+ "$id": "https://docs.monerolws.com/api/schemas/feed_mempool.json",
+ "type": "object",
+ "properties": {
+ "hash": {"$ref": "feed_base.json#/definitions/binary32"},
+ "prefix_hash": {"$ref": "feed_base.json#/definitions/binary32"},
+ "fee": {"$ref": "feed_base.json#/definitions/uint64"},
+ "unlock_time": {"$ref": "feed_base.json#/definitions/uint64"},
+ "amount": {"$ref": "feed_base.json#/definitions/uint64"},
+ "public_key": {"$ref": "feed_base.json#/definitions/binary32"},
+ "tx_pub_key": {"$ref": "feed_base.json#/definitions/binary32"},
+ "index": {"$ref": "feed_base.json#/definitions/uint32"},
+ "payment_id": {
+ "anyOf": [
+ {"$ref": "feed_base.json#/definitions/binary8"},
+ {"$ref": "feed_base.json#/definitions/binary32"}
+ ]
+ },
+ "mixin": {"$ref": "feed_base.json#/definitions/uint32"},
+ "recipient": {"$ref": "feed_base.json#/definitions/address_meta"},
+ "rct": {"$ref": "feed_base.json#/definitions/binary32"}
+ },
+ "required": [
+ "hash",
+ "prefix_hash",
+ "fee",
+ "unlock_time",
+ "public_key",
+ "tx_pub_key",
+ "index"
+ ]
+}
diff --git a/docs/api/schemas/feed_tx_sync.json b/docs/api/schemas/feed_tx_sync.json
new file mode 100644
index 0000000..ad8c0d7
--- /dev/null
+++ b/docs/api/schemas/feed_tx_sync.json
@@ -0,0 +1,20 @@
+{
+ "$schema": "https://json-schema.org/draft-07/schema",
+ "$id": "https://docs.monerolws.com/api/schemas/feed_tx_sync.json",
+ "type": "object",
+ "properties": {
+ "scanned_block_height": {"$ref": "feed_base.json#/definitions/uint64"},
+ "blockchain_height": {"$ref": "feed_base.json#/definitions/uint64"},
+ "lookahead_fail": {"$ref": "feed_base.json#/definitions/uint64"},
+ "lookahead": {
+ "$ref": "feed_base.json#/definitions/address_meta",
+ "default": "{0, 0}"
+ },
+ "transactions": {
+ "type": "array",
+ "minItems": 1,
+ "items": {"$ref": "feed_base.json#/definitions/transaction"}
+ }
+ },
+ "required": ["scanned_block_height", "blockchain_height"]
+}
diff --git a/docs/api/schemas/feed_warning.json b/docs/api/schemas/feed_warning.json
new file mode 100644
index 0000000..2bbd94a
--- /dev/null
+++ b/docs/api/schemas/feed_warning.json
@@ -0,0 +1,12 @@
+{
+ "$schema": "https://json-schema.org/draft-07/schema",
+ "$id": "https://docs.monerolws.com/api/schemas/feed_warning.json",
+ "type": "object",
+ "properties": {
+ "msg": {"type": "string"},
+ "code": {"$ref": "feed_base.json#/definitions/uint16"},
+ "counter": {"$ref": "feed_base.json#/definitions/uint32"},
+ "height": {"$ref": "feed_base.json#/definitions/uint64"}
+ },
+ "required": ["msg", "code", "counter", "height"]
+}
diff --git a/docs/api/wallet.yaml b/docs/api/wallet.yaml
index 564fba5..dfdf4b5 100644
--- a/docs/api/wallet.yaml
+++ b/docs/api/wallet.yaml
@@ -18,6 +18,16 @@ tags:
- name: subaddress
- name: transaction
paths:
+ /feed:
+ get:
+ tags: [transaction]
+ summary: Check/upgrade to websocket /feed
+ operationId: feed
+ responses:
+ '101':
+ description: Switching to websocket /feed protocol
+ '501':
+ description: /feed is enabled on websocket upgrade only
/get_address_info:
post:
tags:
diff --git a/mkdocs.yml b/mkdocs.yml
index 65bad70..0ccb1f1 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -19,6 +19,7 @@ nav:
- Wallet: api/wallet.md
- Admin: api/admin.md
- Webhooks/ZeroMQ: api/zmq.md
+ - Feed: api/feed.md
- RabbitMQ: api/rmq.md
plugins:
- swagger-ui-tag
diff --git a/src/client_main.cpp b/src/client_main.cpp
index e473670..ca1a961 100644
--- a/src/client_main.cpp
+++ b/src/client_main.cpp
@@ -195,7 +195,7 @@ namespace
{
std::shared_ptr<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};
diff --git a/src/config.h b/src/config.h
index f294df6..88ccead 100644
--- a/src/config.h
+++ b/src/config.h
@@ -1,5 +1,6 @@
#pragma once
+#include <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;
diff --git a/src/db/account.cpp b/src/db/account.cpp
index a6d504d..6d09530 100644
--- a/src/db/account.cpp
+++ b/src/db/account.cpp
@@ -51,6 +51,17 @@ namespace lws
return std::memcmp(std::addressof(lhs), std::addressof(rhs), sizeof(lhs)) < 0;
}
};
+
+ template<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
diff --git a/src/db/account.h b/src/db/account.h
index c78b2d9..c3a64a3 100644
--- a/src/db/account.h
+++ b/src/db/account.h
@@ -26,6 +26,7 @@
// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#pragma once
+#include <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;
diff --git a/src/db/fwd.h b/src/db/fwd.h
index 583b240..09045b1 100644
--- a/src/db/fwd.h
+++ b/src/db/fwd.h
@@ -32,6 +32,7 @@
namespace lws
{
class account;
+ class mempool_receive;
namespace db
{
@@ -56,6 +57,7 @@ namespace db
struct request_info;
struct spend;
class storage;
+ class storage_reader;
struct subaddress_map;
struct transaction_link;
struct view_key;
diff --git a/src/error.cpp b/src/error.cpp
index 3d7b0d6..6ec3725 100644
--- a/src/error.cpp
+++ b/src/error.cpp
@@ -26,6 +26,8 @@
// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#include "error.h"
+#include <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{};
diff --git a/src/error.h b/src/error.h
index 2980e87..4ce3dec 100644
--- a/src/error.h
+++ b/src/error.h
@@ -29,6 +29,13 @@
#include <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
diff --git a/src/rest_server.cpp b/src/rest_server.cpp
index 8578e91..21b40bc 100644
--- a/src/rest_server.cpp
+++ b/src/rest_server.cpp
@@ -47,6 +47,7 @@
#include <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_()
{
diff --git a/src/rest_server.h b/src/rest_server.h
index 6b7c6b6..3de0e01 100644
--- a/src/rest_server.h
+++ b/src/rest_server.h
@@ -60,6 +60,7 @@ namespace lws
{
epee::net_utils::ssl_authentication_t auth;
std::vector<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;
diff --git a/src/rpc/CMakeLists.txt b/src/rpc/CMakeLists.txt
index e869fb7..e9b30f3 100644
--- a/src/rpc/CMakeLists.txt
+++ b/src/rpc/CMakeLists.txt
@@ -28,9 +28,9 @@
add_subdirectory(scanner)
-set(monero-lws-rpc_sources admin.cpp client.cpp daemon_pub.cpp daemon_zmq.cpp light_wallet.cpp lws_pub.cpp rates.cpp)
-set(monero-lws-rpc_headers admin.h client.h daemon_pub.h daemon_zmq.h fwd.h json.h light_wallet.h lws_pub.h rates.h)
+set(monero-lws-rpc_sources admin.cpp client.cpp daemon_pub.cpp daemon_zmq.cpp feed_ssl.cpp feed_tcp.cpp light_wallet.cpp login.cpp lws_pub.cpp rates.cpp)
+set(monero-lws-rpc_headers admin.h client.h daemon_pub.h daemon_zmq.h feed.h fwd.h json.h light_wallet.h login.h lws_pub.h rates.h)
add_library(monero-lws-rpc ${monero-lws-rpc_sources} ${monero-lws-rpc_headers})
target_include_directories(monero-lws-rpc PRIVATE ${RMQ_INCLUDE_DIR})
-target_link_libraries(monero-lws-rpc monero::libraries monero-lws-util monero-lws-wire-json monero-lws-wire-wrapper ${RMQ_LIBRARY})
+target_link_libraries(monero-lws-rpc monero::libraries monero-lws-db monero-lws-util monero-lws-wire-json monero-lws-wire-wrapper ${RMQ_LIBRARY})
diff --git a/src/rpc/client.cpp b/src/rpc/client.cpp
index b830e72..e0d0eef 100644
--- a/src/rpc/client.cpp
+++ b/src/rpc/client.cpp
@@ -35,13 +35,16 @@
#include "common/error.h" // monero/contrib/epee/include
#include "db/account.h"
+#include "db/data.h"
#include "error.h"
#include "misc_log_ex.h" // monero/contrib/epee/include
#include "net/http_client.h" // monero/contrib/epee/include
#include "net/zmq.h" // monero/src
#include "net/zmq_async.h"
+#include "rpc/light_wallet.h"
#include "scanner.h"
#include "serialization/json_object.h" // monero/src
+#include "wire/msgpack/write.h"
#include "wire/json/write.h"
#if MLWS_RMQ_ENABLED
#include <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);
diff --git a/src/rpc/client.h b/src/rpc/client.h
index 719f1b8..667c520 100644
--- a/src/rpc/client.h
+++ b/src/rpc/client.h
@@ -38,6 +38,8 @@
#include "db/fwd.h"
#include "common/expect.h" // monero/src
#include "net/zmq.h" // monero/src
+#include "net/zmq_async.h"
+#include "rpc/fwd.h"
#include "rpc/message.h" // monero/src
#include "rpc/daemon_pub.h"
#include "rpc/rates.h"
@@ -63,6 +65,37 @@ namespace rpc
expect<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
diff --git a/src/rpc/feed.h b/src/rpc/feed.h
new file mode 100644
index 0000000..6bc7d65
--- /dev/null
+++ b/src/rpc/feed.h
@@ -0,0 +1,85 @@
+// Copyright (c) 2026, The Monero Project
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without modification, are
+// permitted provided that the following conditions are met:
+//
+// 1. Redistributions of source code must retain the above copyright notice, this list of
+// conditions and the following disclaimer.
+//
+// 2. Redistributions in binary form must reproduce the above copyright notice, this list
+// of conditions and the following disclaimer in the documentation and/or other
+// materials provided with the distribution.
+//
+// 3. Neither the name of the copyright holder nor the names of its contributors may be
+// used to endorse or promote products derived from this software without specific
+// prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY
+// EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
+// MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+// THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
+// STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+#pragma once
+
+#include <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
+ {};
+}
diff --git a/src/rpc/feed.inl b/src/rpc/feed.inl
new file mode 100644
index 0000000..9673092
--- /dev/null
+++ b/src/rpc/feed.inl
@@ -0,0 +1,600 @@
+// Copyright (c) 2026, The Monero Project
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without modification, are
+// permitted provided that the following conditions are met:
+//
+// 1. Redistributions of source code must retain the above copyright notice, this list of
+// conditions and the following disclaimer.
+//
+// 2. Redistributions in binary form must reproduce the above copyright notice, this list
+// of conditions and the following disclaimer in the documentation and/or other
+// materials provided with the distribution.
+//
+// 3. Neither the name of the copyright holder nor the names of its contributors may be
+// used to endorse or promote products derived from this software without specific
+// prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY
+// EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
+// MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+// THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
+// STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+/* This file was split into two compilation units `feed_tcp.cpp` and
+ `feed_ssl.cpp`. This helps reduce the compilation time since each file can be
+ computed in parallel. The memory usage is also reduced when -j1, which helps
+ with compilation on smaller machines.
+
+ The functions must be in anonymous namespace or marked inline (implicitly or
+ explicitly) or linker will have issues. Inline is preferred because the
+ linker will merge the duplicate copies. */
+
+#include "feed.h"
+
+#include <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
+
diff --git a/src/rpc/feed_ssl.cpp b/src/rpc/feed_ssl.cpp
new file mode 100644
index 0000000..cb44251
--- /dev/null
+++ b/src/rpc/feed_ssl.cpp
@@ -0,0 +1,44 @@
+// Copyright (c) 2026, The Monero Project
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without modification, are
+// permitted provided that the following conditions are met:
+//
+// 1. Redistributions of source code must retain the above copyright notice, this list of
+// conditions and the following disclaimer.
+//
+// 2. Redistributions in binary form must reproduce the above copyright notice, this list
+// of conditions and the following disclaimer in the documentation and/or other
+// materials provided with the distribution.
+//
+// 3. Neither the name of the copyright holder nor the names of its contributors may be
+// used to endorse or promote products derived from this software without specific
+// prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY
+// EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
+// MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+// THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
+// STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+#include <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
+ }
+}}}
+
diff --git a/src/rpc/feed_tcp.cpp b/src/rpc/feed_tcp.cpp
new file mode 100644
index 0000000..06b6550
--- /dev/null
+++ b/src/rpc/feed_tcp.cpp
@@ -0,0 +1,374 @@
+// Copyright (c) 2026, The Monero Project
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without modification, are
+// permitted provided that the following conditions are met:
+//
+// 1. Redistributions of source code must retain the above copyright notice, this list of
+// conditions and the following disclaimer.
+//
+// 2. Redistributions in binary form must reproduce the above copyright notice, this list
+// of conditions and the following disclaimer in the documentation and/or other
+// materials provided with the distribution.
+//
+// 3. Neither the name of the copyright holder nor the names of its contributors may be
+// used to endorse or promote products derived from this software without specific
+// prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY
+// EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
+// MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+// THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
+// STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+#include "feed.inl"
+
+#include <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
+
diff --git a/src/rpc/fwd.h b/src/rpc/fwd.h
index a60fc84..56f2cff 100644
--- a/src/rpc/fwd.h
+++ b/src/rpc/fwd.h
@@ -27,11 +27,20 @@
#pragma once
+#include <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;
diff --git a/src/rpc/light_wallet.cpp b/src/rpc/light_wallet.cpp
index 179235a..706b199 100644
--- a/src/rpc/light_wallet.cpp
+++ b/src/rpc/light_wallet.cpp
@@ -35,11 +35,13 @@
#include <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));
diff --git a/src/rpc/light_wallet.h b/src/rpc/light_wallet.h
index 7fe89ac..952224d 100644
--- a/src/rpc/light_wallet.h
+++ b/src/rpc/light_wallet.h
@@ -38,9 +38,12 @@
#include "cryptonote_basic/difficulty.h" // monero/src
#include "crypto/crypto.h" // monero/src
#include "db/data.h"
+#include "db/fwd.h"
+#include "rpc/fwd.h"
#include "rpc/rates.h"
#include "util/fwd.h"
#include "wire/json/fwd.h"
+#include "wire/msgpack/fwd.h"
namespace lws
{
@@ -64,7 +67,7 @@ namespace rpc
lws::db::account_address address;
crypto::secret_key key;
};
- void read_bytes(wire::json_reader&, account_credentials&);
+ void read_bytes(wire::reader&, account_credentials&);
enum class daemon_state : std::uint8_t
@@ -107,15 +110,6 @@ namespace rpc
void write_bytes(wire::json_writer&, const daemon_status_response&);
- struct new_subaddrs_response
- {
- new_subaddrs_response() = delete;
- std::vector<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;
diff --git a/src/rpc/login.cpp b/src/rpc/login.cpp
new file mode 100644
index 0000000..fd7376a
--- /dev/null
+++ b/src/rpc/login.cpp
@@ -0,0 +1,80 @@
+// Copyright (c) 2026, The Monero Project
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without modification, are
+// permitted provided that the following conditions are met:
+//
+// 1. Redistributions of source code must retain the above copyright notice, this list of
+// conditions and the following disclaimer.
+//
+// 2. Redistributions in binary form must reproduce the above copyright notice, this list
+// of conditions and the following disclaimer in the documentation and/or other
+// materials provided with the distribution.
+//
+// 3. Neither the name of the copyright holder nor the names of its contributors may be
+// used to endorse or promote products derived from this software without specific
+// prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY
+// EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
+// MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+// THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
+// STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+#include "login.h"
+
+#include "crypto/crypto.h" // monero/src
+#include "db/data.h"
+#include "db/storage.h"
+#include "error.h"
+#include "rpc/light_wallet.h"
+
+namespace lws { namespace rpc
+{
+ bool is_hidden(db::account_status status) noexcept
+ {
+ switch (status)
+ {
+ case db::account_status::active:
+ case db::account_status::inactive:
+ return false;
+ default:
+ case db::account_status::hidden:
+ break;
+ }
+ return true;
+ }
+
+ bool key_check(const rpc::account_credentials& creds)
+ {
+ crypto::public_key verify{};
+ if (!crypto::secret_key_to_public_key(creds.key, verify))
+ return false;
+ if (verify != creds.address.view_public)
+ return false;
+ return true;
+ }
+
+ //! \return Account info from the DB, iff key matches address AND address is NOT hidden.
+ expect<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
diff --git a/src/rpc/login.h b/src/rpc/login.h
new file mode 100644
index 0000000..ac2e8ac
--- /dev/null
+++ b/src/rpc/login.h
@@ -0,0 +1,45 @@
+// Copyright (c) 2026, The Monero Project
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without modification, are
+// permitted provided that the following conditions are met:
+//
+// 1. Redistributions of source code must retain the above copyright notice, this list of
+// conditions and the following disclaimer.
+//
+// 2. Redistributions in binary form must reproduce the above copyright notice, this list
+// of conditions and the following disclaimer in the documentation and/or other
+// materials provided with the distribution.
+//
+// 3. Neither the name of the copyright holder nor the names of its contributors may be
+// used to endorse or promote products derived from this software without specific
+// prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY
+// EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
+// MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+// THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
+// STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+#pragma once
+
+#include <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
diff --git a/src/rpc/scanner/server.cpp b/src/rpc/scanner/server.cpp
index 22ecc9a..1091255 100644
--- a/src/rpc/scanner/server.cpp
+++ b/src/rpc/scanner/server.cpp
@@ -623,9 +623,9 @@ namespace lws { namespace rpc { namespace scanner
std::sort(users.begin(), users.end(), by_height{});
boost::asio::dispatch(
self->strand_,
- [self, users = std::move(users), blocks = std::move(blocks)] ()
+ [self, users = std::move(users), blocks = std::move(blocks)] () mutable
{
- if (!lws::user_data::store(self->strand_.context(), self->disk_, self->zclient_, self->webhook_, epee::to_span(blocks), epee::to_span(users), nullptr))
+ if (!lws::user_data::store(self->strand_.context(), self->disk_, self->zclient_, self->webhook_, epee::to_span(blocks), epee::to_mut_span(users), nullptr))
{
self->do_stop();
self->strand_.context().stop();
diff --git a/src/scanner.cpp b/src/scanner.cpp
index 00996ba..1bd70c5 100644
--- a/src/scanner.cpp
+++ b/src/scanner.cpp
@@ -284,6 +284,9 @@ namespace lws
events.pop_back(); //cannot compute tx_hash
}
send_payment_hook(http_.io_, client_, http_.webhooks_, epee::to_span(events));
+ const expect<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);
diff --git a/src/scanner.h b/src/scanner.h
index aecc02c..f5156c7 100644
--- a/src/scanner.h
+++ b/src/scanner.h
@@ -77,10 +77,10 @@ namespace lws
/*! Store updated accounts locally (`disk`), and send ZMQ/RMQ/webhook
events. `users` must be sorted by height (lowest first). */
- static bool store(boost::asio::io_context& io, db::storage& disk, rpc::client& zclient, net::http::client& webhook ,epee::span<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`.
*
diff --git a/src/server_main.cpp b/src/server_main.cpp
index dbcdeec..8a2c0c9 100644
--- a/src/server_main.cpp
+++ b/src/server_main.cpp
@@ -92,6 +92,7 @@ namespace
const command_line::arg_descriptor<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};
diff --git a/tests/unit/rest.test.cpp b/tests/unit/rest.test.cpp
index 2243c24..3518a97 100644
--- a/tests/unit/rest.test.cpp
+++ b/tests/unit/rest.test.cpp
@@ -27,7 +27,11 @@
#include "framework.test.h"
+#include <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);
+ }
}
}
diff --git a/tests/unit/rpc/CMakeLists.txt b/tests/unit/rpc/CMakeLists.txt
index f8c2e40..68c9492 100644
--- a/tests/unit/rpc/CMakeLists.txt
+++ b/tests/unit/rpc/CMakeLists.txt
@@ -26,7 +26,7 @@
# STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
# THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
-add_library(monero-lws-unit-rpc OBJECT admin.test.cpp)
+add_library(monero-lws-unit-rpc OBJECT admin.test.cpp pull.test.cpp pull.test.h)
target_link_libraries(
monero-lws-unit-rpc
monero-lws-unit-db
diff --git a/tests/unit/rpc/pull.test.cpp b/tests/unit/rpc/pull.test.cpp
new file mode 100644
index 0000000..794abb2
--- /dev/null
+++ b/tests/unit/rpc/pull.test.cpp
@@ -0,0 +1,140 @@
+// Copyright (c) 2026 The Monero Project
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without modification, are
+// permitted provided that the following conditions are met:
+//
+// 1. Redistributions of source code must retain the above copyright notice, this list of
+// conditions and the following disclaimer.
+//
+// 2. Redistributions in binary form must reproduce the above copyright notice, this list
+// of conditions and the following disclaimer in the documentation and/or other
+// materials provided with the distribution.
+//
+// 3. Neither the name of the copyright holder nor the names of its contributors may be
+// used to endorse or promote products derived from this software without specific
+// prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY
+// EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
+// MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+// THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
+// STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+#include "pull.test.h"
+
+#include <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
diff --git a/tests/unit/rpc/pull.test.h b/tests/unit/rpc/pull.test.h
new file mode 100644
index 0000000..81b8bbf
--- /dev/null
+++ b/tests/unit/rpc/pull.test.h
@@ -0,0 +1,49 @@
+// Copyright (c) 2026 The Monero Project
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without modification, are
+// permitted provided that the following conditions are met:
+//
+// 1. Redistributions of source code must retain the above copyright notice, this list of
+// conditions and the following disclaimer.
+//
+// 2. Redistributions in binary form must reproduce the above copyright notice, this list
+// of conditions and the following disclaimer in the documentation and/or other
+// materials provided with the distribution.
+//
+// 3. Neither the name of the copyright holder nor the names of its contributors may be
+// used to endorse or promote products derived from this software without specific
+// prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY
+// EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
+// MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+// THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
+// STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+#pragma once
+
+#include <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
diff --git a/tests/unit/scanner.test.cpp b/tests/unit/scanner.test.cpp
index 7341615..9cf42bc 100644
--- a/tests/unit/scanner.test.cpp
+++ b/tests/unit/scanner.test.cpp
@@ -332,7 +332,7 @@ LWS_CASE("lws::scanner::sync and lws::scanner::run")
SETUP("lws::rpc::context, ZMQ_REP Server, and lws::db::storage")
{
auto rpc =
- lws::rpc::context::make(lws_test::rpc_rendevous, {}, {}, {}, std::chrono::minutes{0}, false);
+ lws::rpc::context::make(lws_test::rpc_rendevous, {}, {}, {}, std::chrono::minutes{0}, false, true);
lws::db::test::cleanup_db on_scope_exit{};
Why this scored 39/100
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.