starred/chia-blockchain
1
0
Fork 0
mirror of https://github.com/Chia-Network/chia-blockchain.git synced 2026-08-28 18:14:19 -05:00
Code Issues Packages Projects Releases Wiki Activity
Files
use-structured-RPC-errors
chia-blockchain/chia/full_node/full_node_rpc_api.py
T

1172 lines
53 KiB
Python
Raw Permalink Blame History

from __future__ import annotations
import asyncio
import time
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
from typing import TYPE_CHECKING, Any, ClassVar, cast
from chia_rs import (
MEMPOOL_MODE,
BlockRecord,
CoinRecord,
CoinSpend,
FullBlock,
PlotParam,
SpendBundle,
SpendBundleConditions,
get_spends_for_trusted_block,
get_spends_for_trusted_block_with_conditions,
run_block_generator2,
)
from chia_rs import get_puzzle_and_solution_for_coin2 as get_puzzle_and_solution_for_coin
from chia_rs.sized_bytes import bytes32
from chia_rs.sized_ints import uint32, uint64, uint128
from chia.consensus.blockchain import Blockchain, BlockchainMutexPriority
from chia.consensus.get_block_generator import get_block_generator
from chia.consensus.pos_quality import UI_ACTUAL_SPACE_CONSTANT_FACTOR
from chia.full_node.fee_estimator_interface import FeeEstimatorInterface
from chia.full_node.full_node import FullNode
from chia.full_node.hard_fork_utils import get_flags
from chia.protocols.outbound_message import NodeType
from chia.rpc.rpc_errors import RpcError, RpcErrorCodes
from chia.rpc.rpc_server import Endpoint, EndpointResult
from chia.types.blockchain_format.proof_of_space import calculate_prefix_bits
from chia.types.generator_types import BlockGenerator, NewBlockGenerator
from chia.types.mempool_inclusion_status import MempoolInclusionStatus
from chia.types.unfinished_header_block import UnfinishedHeaderBlock
from chia.util.byte_types import hexstr_to_bytes
from chia.util.log_exceptions import log_exceptions
from chia.util.math import make_monotonically_decreasing
from chia.util.ws_message import WsRpcMessage, create_payload_dict
def coin_record_dict_backwards_compat(coin_record: dict[str, Any]) -> dict[str, bool]:
coin_record["spent"] = coin_record["spent_block_index"] > 0
return coin_record
async def get_nearest_transaction_block(blockchain: Blockchain, block: BlockRecord) -> BlockRecord:
if block.is_transaction_block:
return block
prev_hash = blockchain.height_to_hash(block.prev_transaction_block_height)
# Genesis block is a transaction block, so theoretically `prev_hash` of all blocks
# other than genesis block cannot be `None`.
assert prev_hash
tb = await blockchain.get_block_record_from_db(prev_hash)
assert tb
return tb
async def get_average_block_time(
blockchain: Blockchain,
base_block: BlockRecord,
height_distance: int,
) -> uint32 | None:
newer_block = await get_nearest_transaction_block(blockchain, base_block)
if newer_block.height < 1:
return None
prev_height = uint32(max(1, newer_block.height - height_distance))
prev_hash = blockchain.height_to_hash(prev_height)
assert prev_hash
prev_block = await blockchain.get_block_record_from_db(prev_hash)
assert prev_block
older_block = await get_nearest_transaction_block(blockchain, prev_block)
assert newer_block.timestamp is not None and older_block.timestamp is not None
if newer_block.height == older_block.height: # small chain not long enough to have a block in between
return None
average_block_time = uint32(
(newer_block.timestamp - older_block.timestamp) / (newer_block.height - older_block.height)
)
return average_block_time
class FullNodeRpcApi:
executor: ThreadPoolExecutor
if TYPE_CHECKING:
from chia.rpc.rpc_server import RpcApiProtocol
_protocol_check: ClassVar[RpcApiProtocol] = cast("FullNodeRpcApi", None)
def __init__(self, service: FullNode) -> None:
self.service = service
self.service_name = "chia_full_node"
self.cached_blockchain_state: dict[str, Any] | None = None
self.executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix="node-rpc-")
def get_routes(self) -> dict[str, Endpoint]:
return {
# Blockchain
"/get_blockchain_state": self.get_blockchain_state,
"/get_block": self.get_block,
"/get_blocks": self.get_blocks,
"/get_block_count_metrics": self.get_block_count_metrics,
"/get_block_record_by_height": self.get_block_record_by_height,
"/get_block_record": self.get_block_record,
"/get_block_records": self.get_block_records,
"/get_block_spends": self.get_block_spends,
"/get_block_spends_with_conditions": self.get_block_spends_with_conditions,
"/get_unfinished_block_headers": self.get_unfinished_block_headers,
"/get_network_space": self.get_network_space,
"/get_additions_and_removals": self.get_additions_and_removals,
"/get_aggsig_additional_data": self.get_aggsig_additional_data,
"/get_recent_signage_point_or_eos": self.get_recent_signage_point_or_eos,
# Coins
"/get_coin_records_by_puzzle_hash": self.get_coin_records_by_puzzle_hash,
"/get_coin_records_by_puzzle_hashes": self.get_coin_records_by_puzzle_hashes,
"/get_coin_record_by_name": self.get_coin_record_by_name,
"/get_coin_records_by_names": self.get_coin_records_by_names,
"/get_coin_records_by_parent_ids": self.get_coin_records_by_parent_ids,
"/get_coin_records_by_hint": self.get_coin_records_by_hint,
"/push_tx": self.push_tx,
"/get_puzzle_and_solution": self.get_puzzle_and_solution,
# Mempool
"/get_all_mempool_tx_ids": self.get_all_mempool_tx_ids,
"/get_all_mempool_items": self.get_all_mempool_items,
"/get_mempool_item_by_tx_id": self.get_mempool_item_by_tx_id,
"/get_mempool_items_by_coin_name": self.get_mempool_items_by_coin_name,
"/create_block_generator": self.create_block_generator,
# Fee estimation
"/get_fee_estimate": self.get_fee_estimate,
}
async def _state_changed(self, change: str, change_data: dict[str, Any] | None = None) -> list[WsRpcMessage]:
if change_data is None:
change_data = {}
payloads = []
if change in {"new_peak", "sync_mode"}:
data = await self.get_blockchain_state({})
assert data is not None
payloads.append(
create_payload_dict(
"get_blockchain_state",
data,
self.service_name,
"wallet_ui",
)
)
payloads.append(
create_payload_dict(
"get_blockchain_state",
data,
self.service_name,
"metrics",
)
)
if change in {"block", "signage_point"}:
payloads.append(create_payload_dict(change, change_data, self.service_name, "metrics"))
if change == "unfinished_block":
payloads.append(create_payload_dict(change, change_data, self.service_name, "unfinished_block_info"))
return payloads
async def get_blockchain_state(self, _: dict[str, Any]) -> EndpointResult:
"""
Returns a summary of the node's view of the blockchain.
"""
node_id = self.service.server.node_id.hex()
if self.service.initialized is False:
res = {
"blockchain_state": {
"peak": None,
"genesis_challenge_initialized": self.service.initialized,
"sync": {
"sync_mode": False,
"synced": False,
"sync_tip_height": 0,
"sync_progress_height": 0,
},
"difficulty": 0,
"sub_slot_iters": 0,
"space": 0,
"average_block_time": None,
"mempool_size": 0,
"mempool_cost": 0,
"mempool_min_fees": {
"cost_5000000": 0,
},
"mempool_max_total_cost": 0,
"block_max_cost": 0,
"node_id": node_id,
},
}
return res
peak: BlockRecord | None = self.service.blockchain.get_peak()
if peak is not None and peak.height > 0:
difficulty = uint64(peak.weight - self.service.blockchain.block_record(peak.prev_hash).weight)
sub_slot_iters = peak.sub_slot_iters
else:
difficulty = self.service.constants.DIFFICULTY_STARTING
sub_slot_iters = self.service.constants.SUB_SLOT_ITERS_STARTING
sync_mode: bool = self.service.sync_store.get_sync_mode() or self.service.sync_store.get_long_sync()
sync_tip_height: uint32 | None = uint32(0)
if sync_mode:
target_peak = self.service.sync_store.target_peak
if target_peak is not None:
sync_tip_height = target_peak.height
if peak is not None:
sync_progress_height: uint32 = peak.height
# Don't display we're syncing towards 0, instead show 'Syncing height/height'
# until sync_store retrieves the correct number.
if sync_tip_height == uint32(0):
sync_tip_height = peak.height
else:
sync_progress_height = uint32(0)
else:
sync_progress_height = uint32(0)
average_block_time: uint32 | None = None
if peak is not None and peak.height > 1:
newer_block_hex = peak.header_hash.hex()
# Average over the last day
header_hash = self.service.blockchain.height_to_hash(uint32(max(1, peak.height - 4608)))
assert header_hash is not None
older_block_hex = header_hash.hex()
space = await self.get_network_space(
{"newer_block_header_hash": newer_block_hex, "older_block_header_hash": older_block_hex}
)
average_block_time = await get_average_block_time(self.service.blockchain, peak, 4608)
else:
space = {"space": uint128(0)}
if self.service.mempool_manager is not None:
mempool_size = self.service.mempool_manager.mempool.size()
mempool_cost = self.service.mempool_manager.mempool.total_mempool_cost()
mempool_fees = self.service.mempool_manager.mempool.total_mempool_fees()
mempool_min_fee_5m = self.service.mempool_manager.mempool.get_min_fee_rate(5000000)
mempool_max_total_cost = self.service.mempool_manager.mempool_max_total_cost
else:
mempool_size = 0
mempool_cost = 0
mempool_fees = 0
mempool_min_fee_5m = 0
mempool_max_total_cost = 0
if self.service.server is not None:
is_connected = len(self.service.server.get_connections(NodeType.FULL_NODE)) > 0 or "simulator" in str(
self.service.config.get("selected_network")
)
else:
is_connected = False
synced = await self.service.synced() and is_connected
assert space is not None
response = {
"blockchain_state": {
"peak": peak,
"genesis_challenge_initialized": self.service.initialized,
"sync": {
"sync_mode": sync_mode,
"synced": synced,
"sync_tip_height": sync_tip_height,
"sync_progress_height": sync_progress_height,
},
"difficulty": difficulty,
"sub_slot_iters": sub_slot_iters,
"space": space["space"],
"average_block_time": average_block_time,
"mempool_size": mempool_size,
"mempool_cost": mempool_cost,
"mempool_fees": mempool_fees,
"mempool_min_fees": {
# We may give estimates for varying costs in the future
# This Dict sets us up for that in the future
"cost_5000000": mempool_min_fee_5m,
},
"mempool_max_total_cost": mempool_max_total_cost,
"block_max_cost": self.service.constants.MAX_BLOCK_COST_CLVM,
"node_id": node_id,
},
}
self.cached_blockchain_state = dict(response["blockchain_state"])
return response
async def get_recent_signage_point_or_eos(self, request: dict[str, Any]) -> EndpointResult:
if "sp_hash" not in request:
challenge_hash: bytes32 = bytes32.from_hexstr(request["challenge_hash"])
# This is the case of getting an end of slot
eos_tuple = self.service.full_node_store.recent_eos.get(challenge_hash)
if not eos_tuple:
raise RpcError(
RpcErrorCodes.EOS_NOT_IN_CACHE,
f"Did not find eos {challenge_hash.hex()} in cache",
data={"challenge_hash": challenge_hash.hex()},
structured_message="Did not find eos in cache",
)
eos, time_received = eos_tuple
# If it's still in the full node store, it's not reverted
if self.service.full_node_store.get_sub_slot(eos.challenge_chain.get_hash()):
return {"eos": eos, "time_received": time_received, "reverted": False}
# Otherwise we can backtrack from peak to find it in the blockchain
curr: BlockRecord | None = self.service.blockchain.get_peak()
if curr is None:
raise RpcError.simple(RpcErrorCodes.NO_BLOCKS_IN_CHAIN, "No blocks in the chain")
number_of_slots_searched = 0
while number_of_slots_searched < 10:
if curr.first_in_sub_slot:
assert curr.finished_challenge_slot_hashes is not None
if curr.finished_challenge_slot_hashes[-1] == eos.challenge_chain.get_hash():
# Found this slot in the blockchain
return {"eos": eos, "time_received": time_received, "reverted": False}
number_of_slots_searched += len(curr.finished_challenge_slot_hashes)
curr = self.service.blockchain.try_block_record(curr.prev_hash)
if curr is None:
# Got to the beginning of the blockchain without finding the slot
return {"eos": eos, "time_received": time_received, "reverted": True}
# Backtracked through 10 slots but still did not find it
return {"eos": eos, "time_received": time_received, "reverted": True}
# Now we handle the case of getting a signage point
sp_hash: bytes32 = bytes32.from_hexstr(request["sp_hash"])
sp_tuple = self.service.full_node_store.recent_signage_points.get(sp_hash)
if sp_tuple is None:
raise RpcError(
RpcErrorCodes.SP_NOT_IN_CACHE,
f"Did not find sp {sp_hash.hex()} in cache",
data={"sp_hash": sp_hash.hex()},
structured_message="Did not find sp in cache",
)
sp, time_received = sp_tuple
assert sp.rc_vdf is not None, "Not an EOS, the signage point rewards chain VDF must not be None"
assert sp.cc_vdf is not None, "Not an EOS, the signage point challenge chain VDF must not be None"
# If it's still in the full node store, it's not reverted
if self.service.full_node_store.get_signage_point(sp_hash):
return {"signage_point": sp, "time_received": time_received, "reverted": False}
# Otherwise we can backtrack from peak to find it in the blockchain
rc_challenge: bytes32 = sp.rc_vdf.challenge
next_b: BlockRecord | None = None
curr_b_optional: BlockRecord | None = self.service.blockchain.get_peak()
assert curr_b_optional is not None
curr_b: BlockRecord = curr_b_optional
for _ in range(200):
sp_total_iters = sp.cc_vdf.number_of_iterations + curr_b.ip_sub_slot_total_iters(self.service.constants)
if curr_b.reward_infusion_new_challenge == rc_challenge:
if next_b is None:
return {"signage_point": sp, "time_received": time_received, "reverted": False}
next_b_total_iters = next_b.ip_sub_slot_total_iters(self.service.constants) + next_b.ip_iters(
self.service.constants
)
return {
"signage_point": sp,
"time_received": time_received,
"reverted": sp_total_iters > next_b_total_iters,
}
if curr_b.finished_reward_slot_hashes is not None:
assert curr_b.finished_challenge_slot_hashes is not None
for eos_rc in curr_b.finished_challenge_slot_hashes:
if eos_rc == rc_challenge:
if next_b is None:
return {"signage_point": sp, "time_received": time_received, "reverted": False}
next_b_total_iters = next_b.ip_sub_slot_total_iters(self.service.constants) + next_b.ip_iters(
self.service.constants
)
return {
"signage_point": sp,
"time_received": time_received,
"reverted": sp_total_iters > next_b_total_iters,
}
next_b = curr_b
curr_b_optional = self.service.blockchain.try_block_record(curr_b.prev_hash)
if curr_b_optional is None:
break
curr_b = curr_b_optional
return {"signage_point": sp, "time_received": time_received, "reverted": True}
async def get_block(self, request: dict[str, Any]) -> EndpointResult:
if "header_hash" not in request:
raise RpcError.simple(RpcErrorCodes.NO_HEADER_HASH_IN_REQUEST, "No header_hash in request")
header_hash = bytes32.from_hexstr(request["header_hash"])
block: FullBlock | None = await self.service.block_store.get_full_block(header_hash)
if block is None:
raise RpcError(
RpcErrorCodes.BLOCK_NOT_FOUND,
f"Block {header_hash.hex()} not found",
data={"header_hash": header_hash.hex()},
structured_message="Block not found",
)
return {"block": block}
async def get_blocks(self, request: dict[str, Any]) -> EndpointResult:
if "start" not in request:
raise RpcError.simple(RpcErrorCodes.NO_START_IN_REQUEST, "No start in request")
if "end" not in request:
raise RpcError.simple(RpcErrorCodes.NO_END_IN_REQUEST, "No end in request")
exclude_hh = False
if "exclude_header_hash" in request:
exclude_hh = request["exclude_header_hash"]
exclude_reorged = False
if "exclude_reorged" in request:
exclude_reorged = request["exclude_reorged"]
start = int(request["start"])
end = int(request["end"])
block_range = []
for a in range(start, end):
block_range.append(uint32(a))
blocks: list[FullBlock] = await self.service.block_store.get_full_blocks_at(block_range)
json_blocks = []
for block in blocks:
hh: bytes32 = block.header_hash
if exclude_reorged and self.service.blockchain.height_to_hash(block.height) != hh:
# Don't include forked (reorged) blocks
continue
json = block.to_json_dict()
if not exclude_hh:
json["header_hash"] = hh.hex()
json_blocks.append(json)
return {"blocks": json_blocks}
async def get_block_count_metrics(self, _: dict[str, Any]) -> EndpointResult:
compact_blocks = 0
uncompact_blocks = 0
with log_exceptions(self.service.log, consume=True):
compact_blocks = await self.service.block_store.count_compactified_blocks()
uncompact_blocks = await self.service.block_store.count_uncompactified_blocks()
hint_count = 0
if self.service.hint_store is not None:
with log_exceptions(self.service.log, consume=True):
hint_count = await self.service.hint_store.count_hints()
return {
"metrics": {
"compact_blocks": compact_blocks,
"uncompact_blocks": uncompact_blocks,
"hint_count": hint_count,
}
}
async def get_block_records(self, request: dict[str, Any]) -> EndpointResult:
if "start" not in request:
raise RpcError.simple(RpcErrorCodes.NO_START_IN_REQUEST, "No start in request")
if "end" not in request:
raise RpcError.simple(RpcErrorCodes.NO_END_IN_REQUEST, "No end in request")
start = int(request["start"])
end = int(request["end"])
records = []
peak_height = self.service.blockchain.get_peak_height()
if peak_height is None:
raise RpcError.simple(RpcErrorCodes.PEAK_IS_NONE, "Peak is None")
for a in range(start, end):
if peak_height < uint32(a):
self.service.log.warning("requested block is higher than known peak ")
break
header_hash: bytes32 | None = self.service.blockchain.height_to_hash(uint32(a))
if header_hash is None:
raise RpcError(
RpcErrorCodes.HEIGHT_NOT_IN_BLOCKCHAIN,
f"Height not in blockchain: {a}",
data={"height": a},
structured_message="Height not in blockchain",
)
record: BlockRecord | None = self.service.blockchain.try_block_record(header_hash)
if record is None:
# Fetch from DB
record = await self.service.blockchain.block_store.get_block_record(header_hash)
if record is None:
raise RpcError(
RpcErrorCodes.BLOCK_DOES_NOT_EXIST,
f"Block {header_hash.hex()} does not exist",
data={"header_hash": header_hash.hex()},
structured_message="Block does not exist",
)
records.append(record)
return {"block_records": records}
async def get_block_spends(self, request: dict[str, Any]) -> EndpointResult:
if "header_hash" not in request:
raise RpcError.simple(RpcErrorCodes.NO_HEADER_HASH_IN_REQUEST, "No header_hash in request")
header_hash = bytes32.from_hexstr(request["header_hash"])
full_block: FullBlock | None = await self.service.block_store.get_full_block(header_hash)
if full_block is None:
raise RpcError(
RpcErrorCodes.BLOCK_NOT_FOUND,
f"Block {header_hash.hex()} not found",
data={"header_hash": header_hash.hex()},
structured_message="Block not found",
)
block_generator = await get_block_generator(self.service.blockchain.lookup_block_generators, full_block)
if block_generator is None: # if block is not a transaction block.
return {"block_spends": []}
flags = await get_flags(constants=self.service.constants, blocks=self.service.blockchain, block=full_block)
spends = await asyncio.get_running_loop().run_in_executor(
self.executor,
get_spends_for_trusted_block,
self.service.constants,
block_generator.program,
block_generator.generator_refs,
flags,
)
return spends
async def get_block_spends_with_conditions(self, request: dict[str, Any]) -> EndpointResult:
if "header_hash" not in request:
raise RpcError.simple(RpcErrorCodes.NO_HEADER_HASH_IN_REQUEST, "No header_hash in request")
header_hash = bytes32.from_hexstr(request["header_hash"])
full_block: FullBlock | None = await self.service.block_store.get_full_block(header_hash)
if full_block is None:
raise RpcError(
RpcErrorCodes.BLOCK_NOT_FOUND,
f"Block {header_hash.hex()} not found",
data={"header_hash": header_hash.hex()},
structured_message="Block not found",
)
block_generator = await get_block_generator(self.service.blockchain.lookup_block_generators, full_block)
if block_generator is None: # if block is not a transaction block.
return {"block_spends_with_conditions": []}
flags = await get_flags(constants=self.service.constants, blocks=self.service.blockchain, block=full_block)
spends_with_conditions = await asyncio.get_running_loop().run_in_executor(
self.executor,
get_spends_for_trusted_block_with_conditions,
self.service.constants,
block_generator.program,
block_generator.generator_refs,
flags,
)
return {"block_spends_with_conditions": spends_with_conditions}
async def get_block_record_by_height(self, request: dict[str, Any]) -> EndpointResult:
if "height" not in request:
raise RpcError.simple(RpcErrorCodes.NO_HEIGHT_IN_REQUEST, "No height in request")
height = request["height"]
header_height = uint32(height)
peak_height = self.service.blockchain.get_peak_height()
if peak_height is None or header_height > peak_height:
raise RpcError(
RpcErrorCodes.BLOCK_HEIGHT_NOT_FOUND,
f"Block height {height} not found in chain",
data={"height": height},
structured_message="Block height not found in chain",
)
header_hash: bytes32 | None = self.service.blockchain.height_to_hash(header_height)
if header_hash is None:
raise RpcError(
RpcErrorCodes.BLOCK_HASH_NOT_FOUND,
f"Block hash {height} not found in chain",
data={"height": height},
structured_message="Block hash not found in chain",
)
record: BlockRecord | None = self.service.blockchain.try_block_record(header_hash)
if record is None:
# Fetch from DB
record = await self.service.blockchain.block_store.get_block_record(header_hash)
if record is None:
raise RpcError(
RpcErrorCodes.BLOCK_DOES_NOT_EXIST,
f"Block {header_hash} does not exist",
data={"header_hash": str(header_hash)},
structured_message="Block does not exist",
)
return {"block_record": record}
async def get_block_record(self, request: dict[str, Any]) -> EndpointResult:
if "header_hash" not in request:
raise RpcError.simple(RpcErrorCodes.HEADER_HASH_NOT_IN_REQUEST, "header_hash not in request")
header_hash_str = request["header_hash"]
header_hash = bytes32.from_hexstr(header_hash_str)
record: BlockRecord | None = self.service.blockchain.try_block_record(header_hash)
if record is None:
# Fetch from DB
record = await self.service.blockchain.block_store.get_block_record(header_hash)
if record is None:
raise RpcError(
RpcErrorCodes.BLOCK_DOES_NOT_EXIST,
f"Block {header_hash.hex()} does not exist",
data={"header_hash": header_hash.hex()},
structured_message="Block does not exist",
)
return {"block_record": record}
async def get_unfinished_block_headers(self, _request: dict[str, Any]) -> EndpointResult:
peak: BlockRecord | None = self.service.blockchain.get_peak()
if peak is None:
return {"headers": []}
response_headers: list[UnfinishedHeaderBlock] = []
for block in self.service.full_node_store.get_unfinished_blocks(peak.height):
unfinished_header_block = UnfinishedHeaderBlock(
block.finished_sub_slots,
block.reward_chain_block,
block.challenge_chain_sp_proof,
block.reward_chain_sp_proof,
block.foliage,
block.foliage_transaction_block,
b"",
)
response_headers.append(unfinished_header_block)
return {"headers": response_headers}
async def get_network_space(self, request: dict[str, Any]) -> EndpointResult:
"""
Retrieves an estimate of total space validating the chain
between two block header hashes.
"""
if "newer_block_header_hash" not in request or "older_block_header_hash" not in request:
raise RpcError.simple(
RpcErrorCodes.INVALID_NETWORK_SPACE_REQUEST,
"Invalid request. newer_block_header_hash and older_block_header_hash required",
)
newer_block_hex = request["newer_block_header_hash"]
older_block_hex = request["older_block_header_hash"]
if newer_block_hex == older_block_hex:
raise RpcError.simple(RpcErrorCodes.NEW_AND_OLD_MUST_DIFFER, "New and old must not be the same")
newer_block_bytes = bytes32.from_hexstr(newer_block_hex)
older_block_bytes = bytes32.from_hexstr(older_block_hex)
newer_block = await self.service.block_store.get_block_record(newer_block_bytes)
if newer_block is None:
# It's possible that the peak block has not yet been committed to the DB, so as a fallback, check memory
try:
newer_block = self.service.blockchain.block_record(newer_block_bytes)
except KeyError:
raise RpcError(
RpcErrorCodes.NEWER_BLOCK_NOT_FOUND,
f"Newer block {newer_block_hex} not found",
data={"header_hash": newer_block_hex},
structured_message="Newer block not found",
)
older_block = await self.service.block_store.get_block_record(older_block_bytes)
if older_block is None:
raise RpcError(
RpcErrorCodes.OLDER_BLOCK_NOT_FOUND,
f"Older block {older_block_hex} not found",
data={"header_hash": older_block_hex},
structured_message="Older block not found",
)
delta_weight = newer_block.weight - older_block.weight
# TODO: todo_v2_plots Update this calculation to take v2 plots into
# account
plot_filter_size = calculate_prefix_bits(self.service.constants, newer_block.height, PlotParam.make_v1(32))
delta_iters = newer_block.total_iters - older_block.total_iters
weight_div_iters = delta_weight / delta_iters
additional_difficulty_constant = self.service.constants.DIFFICULTY_CONSTANT_FACTOR
eligible_plots_filter_multiplier = 2**plot_filter_size
network_space_bytes_estimate = (
UI_ACTUAL_SPACE_CONSTANT_FACTOR
* weight_div_iters
* additional_difficulty_constant
* eligible_plots_filter_multiplier
)
return {"space": uint128(network_space_bytes_estimate)}
async def get_coin_records_by_puzzle_hash(self, request: dict[str, Any]) -> EndpointResult:
"""
Retrieves the coins for a given puzzlehash, by default returns unspent coins.
"""
if "puzzle_hash" not in request:
raise RpcError.simple(RpcErrorCodes.PUZZLE_HASH_NOT_IN_REQUEST, "Puzzle hash not in request")
kwargs: dict[str, Any] = {"include_spent_coins": False, "puzzle_hash": hexstr_to_bytes(request["puzzle_hash"])}
if "start_height" in request:
kwargs["start_height"] = uint32(request["start_height"])
if "end_height" in request:
kwargs["end_height"] = uint32(request["end_height"])
if "include_spent_coins" in request:
kwargs["include_spent_coins"] = request["include_spent_coins"]
coin_records = await self.service.blockchain.coin_store.get_coin_records_by_puzzle_hash(**kwargs)
return {"coin_records": [coin_record_dict_backwards_compat(cr.to_json_dict()) for cr in coin_records]}
async def get_coin_records_by_puzzle_hashes(self, request: dict[str, Any]) -> EndpointResult:
"""
Retrieves the coins for a given puzzlehash, by default returns unspent coins.
"""
if "puzzle_hashes" not in request:
raise RpcError.simple(RpcErrorCodes.PUZZLE_HASHES_NOT_IN_REQUEST, "Puzzle hashes not in request")
kwargs: dict[str, Any] = {
"include_spent_coins": False,
"puzzle_hashes": [hexstr_to_bytes(ph) for ph in request["puzzle_hashes"]],
}
if "start_height" in request:
kwargs["start_height"] = uint32(request["start_height"])
if "end_height" in request:
kwargs["end_height"] = uint32(request["end_height"])
if "include_spent_coins" in request:
kwargs["include_spent_coins"] = request["include_spent_coins"]
coin_records = await self.service.blockchain.coin_store.get_coin_records_by_puzzle_hashes(**kwargs)
return {"coin_records": [coin_record_dict_backwards_compat(cr.to_json_dict()) for cr in coin_records]}
async def get_coin_record_by_name(self, request: dict[str, Any]) -> EndpointResult:
"""
Retrieves a coin record by its name.
"""
if "name" not in request:
raise RpcError.simple(RpcErrorCodes.NAME_NOT_IN_REQUEST, "Name not in request")
name = bytes32.from_hexstr(request["name"])
coin_record: CoinRecord | None = await self.service.blockchain.coin_store.get_coin_record(name)
if coin_record is None:
raise RpcError(
RpcErrorCodes.COIN_RECORD_NOT_FOUND,
f"Coin record 0x{name.hex()} not found",
data={"name": name.hex()},
structured_message="Coin record not found",
)
return {"coin_record": coin_record_dict_backwards_compat(coin_record.to_json_dict())}
async def get_coin_records_by_names(self, request: dict[str, Any]) -> EndpointResult:
"""
Retrieves the coins for given coin IDs, by default returns unspent coins.
"""
if "names" not in request:
raise RpcError.simple(RpcErrorCodes.NAMES_NOT_IN_REQUEST, "Names not in request")
kwargs: dict[str, Any] = {
"include_spent_coins": False,
"names": [hexstr_to_bytes(name) for name in request["names"]],
}
if "start_height" in request:
kwargs["start_height"] = uint32(request["start_height"])
if "end_height" in request:
kwargs["end_height"] = uint32(request["end_height"])
if "include_spent_coins" in request:
kwargs["include_spent_coins"] = request["include_spent_coins"]
coin_records = await self.service.blockchain.coin_store.get_coin_records_by_names(**kwargs)
return {"coin_records": [coin_record_dict_backwards_compat(cr.to_json_dict()) for cr in coin_records]}
async def get_coin_records_by_parent_ids(self, request: dict[str, Any]) -> EndpointResult:
"""
Retrieves the coins for given parent coin IDs, by default returns unspent coins.
"""
if "parent_ids" not in request:
raise RpcError.simple(RpcErrorCodes.PARENT_IDS_NOT_IN_REQUEST, "Parent IDs not in request")
kwargs: dict[str, Any] = {
"include_spent_coins": False,
"parent_ids": [hexstr_to_bytes(ph) for ph in request["parent_ids"]],
}
if "start_height" in request:
kwargs["start_height"] = uint32(request["start_height"])
if "end_height" in request:
kwargs["end_height"] = uint32(request["end_height"])
if "include_spent_coins" in request:
kwargs["include_spent_coins"] = request["include_spent_coins"]
coin_records = await self.service.blockchain.coin_store.get_coin_records_by_parent_ids(**kwargs)
return {"coin_records": [coin_record_dict_backwards_compat(cr.to_json_dict()) for cr in coin_records]}
async def get_coin_records_by_hint(self, request: dict[str, Any]) -> EndpointResult:
"""
Retrieves coins by hint, by default returns unspent coins.
"""
if "hint" not in request:
raise RpcError.simple(RpcErrorCodes.HINT_NOT_IN_REQUEST, "Hint not in request")
if self.service.hint_store is None:
return {"coin_records": []}
names: list[bytes32] = await self.service.hint_store.get_coin_ids(bytes32.from_hexstr(request["hint"]))
kwargs: dict[str, Any] = {
"include_spent_coins": False,
"names": names,
}
if "start_height" in request:
kwargs["start_height"] = uint32(request["start_height"])
if "end_height" in request:
kwargs["end_height"] = uint32(request["end_height"])
if "include_spent_coins" in request:
kwargs["include_spent_coins"] = request["include_spent_coins"]
coin_records = await self.service.blockchain.coin_store.get_coin_records_by_names(**kwargs)
return {"coin_records": [coin_record_dict_backwards_compat(cr.to_json_dict()) for cr in coin_records]}
async def push_tx(self, request: dict[str, Any]) -> EndpointResult:
if "spend_bundle" not in request:
raise RpcError.simple(RpcErrorCodes.SPEND_BUNDLE_NOT_IN_REQUEST, "Spend bundle not in request")
spend_bundle: SpendBundle = SpendBundle.from_json_dict(request["spend_bundle"])
spend_name = spend_bundle.name()
if self.service.mempool_manager.get_spendbundle(spend_name) is not None:
status = MempoolInclusionStatus.SUCCESS
error = None
else:
status, error = await self.service.add_transaction(spend_bundle, spend_name)
if status != MempoolInclusionStatus.SUCCESS:
if self.service.mempool_manager.get_spendbundle(spend_name) is not None:
# Already in mempool
status = MempoolInclusionStatus.SUCCESS
error = None
if status == MempoolInclusionStatus.FAILED:
assert error is not None
raise RpcError(
RpcErrorCodes.TRANSACTION_FAILED,
f"Failed to include transaction {spend_name}, error {error.name}",
data={"spend_name": str(spend_name), "error": error.name},
structured_message="Failed to include transaction",
)
return {
"status": status.name,
}
async def get_puzzle_and_solution(self, request: dict[str, Any]) -> EndpointResult:
coin_name: bytes32 = bytes32.from_hexstr(request["coin_id"])
height = request["height"]
coin_record = await self.service.coin_store.get_coin_record(coin_name)
if coin_record is None or not coin_record.spent or coin_record.spent_block_index != height:
raise RpcError(
RpcErrorCodes.INVALID_HEIGHT_FOR_COIN,
f"Invalid height {height}. coin record {coin_record}",
data={"height": height},
structured_message="Invalid height for coin record",
)
header_hash = self.service.blockchain.height_to_hash(height)
assert header_hash is not None
block: FullBlock | None = await self.service.block_store.get_full_block(header_hash)
if block is None or block.transactions_generator is None:
raise RpcError.simple(RpcErrorCodes.INVALID_BLOCK_OR_GENERATOR, "Invalid block or block generator")
block_generator: BlockGenerator | None = await get_block_generator(
self.service.blockchain.lookup_block_generators, block
)
assert block_generator is not None
try:
flags = await get_flags(constants=self.service.constants, blocks=self.service.blockchain, block=block)
puzzle, solution = get_puzzle_and_solution_for_coin(
block_generator.program,
block_generator.generator_refs,
self.service.constants.MAX_BLOCK_COST_CLVM,
coin_record.coin,
flags,
)
return {"coin_solution": CoinSpend(coin_record.coin, puzzle, solution)}
except Exception as e:
raise RpcError(
RpcErrorCodes.PUZZLE_SOLUTION_FAILED,
f"Failed to get puzzle and solution for coin {coin_record.coin}, error: {e}",
data={"coin": str(coin_record.coin), "error": str(e)},
structured_message="Failed to get puzzle and solution for coin",
) from e
async def get_additions_and_removals(self, request: dict[str, Any]) -> EndpointResult:
if "header_hash" not in request:
raise RpcError.simple(RpcErrorCodes.NO_HEADER_HASH_IN_REQUEST, "No header_hash in request")
header_hash = bytes32.from_hexstr(request["header_hash"])
block: FullBlock | None = await self.service.block_store.get_full_block(header_hash)
if block is None:
raise RpcError(
RpcErrorCodes.BLOCK_NOT_FOUND,
f"Block {header_hash.hex()} not found",
data={"header_hash": header_hash.hex()},
structured_message="Block not found",
)
async with self.service.blockchain.priority_mutex.acquire(priority=BlockchainMutexPriority.low):
if self.service.blockchain.height_to_hash(block.height) != header_hash:
raise RpcError(
RpcErrorCodes.BLOCK_IN_FORK,
f"Block at {header_hash.hex()} is no longer in the blockchain (it's in a fork)",
data={"header_hash": header_hash.hex()},
structured_message="Block is no longer in the blockchain (in a fork)",
)
additions: list[CoinRecord] = await self.service.coin_store.get_coins_added_at_height(block.height)
removals: list[CoinRecord] = await self.service.coin_store.get_coins_removed_at_height(block.height)
return {
"additions": [coin_record_dict_backwards_compat(cr.to_json_dict()) for cr in additions],
"removals": [coin_record_dict_backwards_compat(cr.to_json_dict()) for cr in removals],
}
async def get_aggsig_additional_data(self, _: dict[str, Any]) -> EndpointResult:
return {"additional_data": self.service.constants.AGG_SIG_ME_ADDITIONAL_DATA.hex()}
async def get_all_mempool_tx_ids(self, _: dict[str, Any]) -> EndpointResult:
ids = list(self.service.mempool_manager.mempool.all_item_ids())
return {"tx_ids": ids}
async def get_all_mempool_items(self, _: dict[str, Any]) -> EndpointResult:
spends = {}
for item in self.service.mempool_manager.mempool.all_items():
spends[item.name.hex()] = item.to_json_dict()
return {"mempool_items": spends}
async def get_mempool_item_by_tx_id(self, request: dict[str, Any]) -> EndpointResult:
if "tx_id" not in request:
raise RpcError.simple(RpcErrorCodes.NO_TX_ID_IN_REQUEST, "No tx_id in request")
include_pending: bool = request.get("include_pending", False)
tx_id: bytes32 = bytes32.from_hexstr(request["tx_id"])
item = self.service.mempool_manager.get_mempool_item(tx_id, include_pending)
if item is None:
raise RpcError(
RpcErrorCodes.TX_NOT_IN_MEMPOOL,
f"Tx id 0x{tx_id.hex()} not in the mempool",
data={"tx_id": tx_id.hex()},
structured_message="Tx id not in the mempool",
)
return {"mempool_item": item.to_json_dict()}
async def get_mempool_items_by_coin_name(self, request: dict[str, Any]) -> EndpointResult:
if "coin_name" not in request:
raise RpcError.simple(RpcErrorCodes.NO_COIN_NAME_IN_REQUEST, "No coin_name in request")
coin_name: bytes32 = bytes32.from_hexstr(request["coin_name"])
items = self.service.mempool_manager.mempool.get_items_by_coin_id(coin_name)
return {"mempool_items": [item.to_json_dict() for item in items]}
async def create_block_generator(self, _: dict[str, Any]) -> EndpointResult:
gen = NewBlockGenerator()
# Grab best transactions from Mempool for given tip target
async with self.service.blockchain.priority_mutex.acquire(priority=BlockchainMutexPriority.low):
peak: BlockRecord | None = self.service.blockchain.get_peak()
if peak is None:
return {
"generator": gen.program,
"refs": gen.block_refs,
"additions": gen.additions,
"removals": gen.removals,
"sig": gen.signature,
"cost": gen.cost,
}
# Finds the last transaction block before this one
curr_l_tb: BlockRecord = peak
while not curr_l_tb.is_transaction_block:
curr_l_tb = self.service.blockchain.block_record(curr_l_tb.prev_hash)
self.service.log.info("Beginning simulated block construction from mempool")
start_time = time.monotonic()
try:
maybe_gen = self.service.mempool_manager.create_block_generator2(curr_l_tb.header_hash, 2.0)
if maybe_gen is None:
self.service.log.error(f"failed to create block generator, peak: {peak}")
else:
gen = maybe_gen
except Exception:
self.service.log.exception(f"Error creating block generator, peak: {peak}")
self.service.log.info(f"Simulated block constructed in {time.monotonic() - start_time:0.2f} seconds")
if maybe_gen is not None:
# this also validates the signature
err, conds = await asyncio.get_running_loop().run_in_executor(
self.executor,
run_block_generator2,
bytes(gen.program),
gen.generator_refs,
self.service.constants.MAX_BLOCK_COST_CLVM,
MEMPOOL_MODE,
gen.signature,
None,
self.service.constants,
)
if err is not None:
self.service.log.error(f"failed to validate block: {err}")
else:
assert conds is not None
if conds.cost != gen.cost:
self.service.log.error(
f"invalid cost of generated block: {conds.cost} expected {gen.cost}"
f" exe-cost: {conds.execution_cost}"
f" cond-cost: {conds.condition_cost}"
)
# TODO: maybe validate additions and removals too
return {
"generator": gen.program,
"refs": gen.block_refs,
"additions": gen.additions,
"removals": gen.removals,
"sig": gen.signature,
"cost": gen.cost,
}
def _get_spendbundle_type_cost(self, name: str) -> uint64:
"""
This is a stopgap until we modify the wallet RPCs to get exact costs for created SpendBundles
before we send them to the Mempool.
"""
tx_cost_estimates = {
"send_xch_transaction": 9_401_710,
"cat_spend": 36_382_111,
"take_offer": 721_393_265,
"cancel_offer": 212_443_993,
"nft_set_nft_did": 115_540_006,
"nft_transfer_nft": 74_385_541, # burn or transfer
"create_new_pool_wallet": 18_055_407,
"pw_absorb_rewards": 82_668_466,
"create_new_did_wallet": 57_360_396,
}
return uint64(tx_cost_estimates[name])
async def _validate_fee_estimate_cost(self, request: dict[str, Any]) -> uint64:
c = 0
ns = ["spend_bundle", "cost", "spend_type"]
for n in ns:
if n in request:
c += 1
if c != 1:
raise RpcError(
RpcErrorCodes.REQUEST_MUST_CONTAIN_EXACTLY_ONE,
f"Request must contain exactly one of {ns}",
data={"options": ns},
structured_message="Request must contain exactly one of the required options",
)
if "spend_bundle" in request:
spend_bundle = SpendBundle.from_json_dict(request["spend_bundle"])
conds: SpendBundleConditions = await self.service.mempool_manager.pre_validate_spendbundle(spend_bundle)
cost = conds.cost
elif "cost" in request:
cost = request["cost"]
else:
cost = self._get_spendbundle_type_cost(request["spend_type"])
cost *= request.get("spend_count", 1)
try:
return uint64(cost)
except (ValueError, TypeError) as e:
raise RpcError(
RpcErrorCodes.INVALID_COST,
f"Invalid cost value: {cost!r}",
data={"cost": str(cost)},
structured_message="Invalid cost value",
) from e
def _validate_target_times(self, request: dict[str, Any]) -> None:
if "target_times" not in request:
raise RpcError.simple(RpcErrorCodes.TARGET_TIMES_REQUIRED, "Request must contain 'target_times' array")
if any(t < 0 for t in request["target_times"]):
raise RpcError.simple(
RpcErrorCodes.TARGET_TIMES_NON_NEGATIVE, "'target_times' array members must be non-negative"
)
async def get_fee_estimate(self, request: dict[str, Any]) -> dict[str, Any]:
self._validate_target_times(request)
spend_cost = await self._validate_fee_estimate_cost(request)
target_times: list[int] = request["target_times"]
estimator: FeeEstimatorInterface = self.service.mempool_manager.mempool.fee_estimator
target_times.sort()
estimates = [
estimator.estimate_fee_rate(time_offset_seconds=time).mojos_per_clvm_cost * spend_cost
for time in target_times
]
# The Bitcoin Fee Estimator works by observing the most common fee rates that appear
# at set times into the future. This can lead to situations that users do not expect,
# such as estimating a higher fee for a longer transaction time.
estimates = make_monotonically_decreasing(estimates)
estimates = [uint64(e) for e in estimates]
current_fee_rate = estimator.estimate_fee_rate(time_offset_seconds=1)
mempool_size = self.service.mempool_manager.mempool.total_mempool_cost()
mempool_fees = self.service.mempool_manager.mempool.total_mempool_fees()
num_mempool_spends = self.service.mempool_manager.mempool.size()
mempool_max_size = estimator.mempool_max_size()
blockchain_state = await self.get_blockchain_state({})
synced = blockchain_state["blockchain_state"]["sync"]["synced"]
peak = blockchain_state["blockchain_state"]["peak"]
if peak is None:
peak_height = uint32(0)
last_peak_timestamp: uint64 | None = uint64(0)
last_block_cost = 0
fee_rate_last_block = 0.0
last_tx_block_fees = uint64(0)
last_tx_block_height = 0
else:
peak_height = peak.height
last_peak_timestamp = peak.timestamp
peak_with_timestamp = peak_height # Last transaction block height
last_tx_block = self.service.blockchain.height_to_block_record(peak_with_timestamp)
while last_tx_block is None or last_peak_timestamp is None:
peak_with_timestamp -= 1
last_tx_block = self.service.blockchain.height_to_block_record(peak_with_timestamp)
last_peak_timestamp = last_tx_block.timestamp
assert last_tx_block is not None # mypy
assert last_peak_timestamp is not None # mypy
assert last_tx_block.fees is not None # mypy
record = await self.service.blockchain.block_store.get_full_block(last_tx_block.header_hash)
last_block_cost = 0
fee_rate_last_block = 0.0
if record and record.transactions_info and record.transactions_info.cost > 0:
last_block_cost = record.transactions_info.cost
fee_rate_last_block = record.transactions_info.fees / record.transactions_info.cost
last_tx_block_fees = last_tx_block.fees
last_tx_block_height = last_tx_block.height
dt = datetime.now(timezone.utc)
utc_time = dt.replace(tzinfo=timezone.utc)
utc_timestamp = utc_time.timestamp()
return {
"estimates": estimates,
"target_times": target_times,
"current_fee_rate": current_fee_rate.mojos_per_clvm_cost,
"mempool_size": mempool_size,
"mempool_fees": mempool_fees,
"num_spends": num_mempool_spends,
"mempool_max_size": mempool_max_size,
"full_node_synced": synced,
"peak_height": peak_height,
"last_peak_timestamp": last_peak_timestamp,
"node_time_utc": int(utc_timestamp),
"last_block_cost": last_block_cost,
"fees_last_block": last_tx_block_fees,
"fee_rate_last_block": fee_rate_last_block,
"last_tx_block_height": last_tx_block_height,
}
Reference in New Issue View Git Blame Copy Permalink