diff --git a/tests/test_dev_miner.py b/tests/test_dev_miner.py index fcc25df..da2337e 100644 --- a/tests/test_dev_miner.py +++ b/tests/test_dev_miner.py @@ -28,6 +28,7 @@ from txstratum.dev.block_miner import BlockMiner, solve_block from txstratum.dev.manager import DevMiningManager from txstratum.dev.tx_miner import solve_tx +from txstratum.jobs import JobStatus, TxJob # Same TX1_DATA from test_api.py — a serialized transaction that we use as a # realistic input for PoW solving. The weight is ~32, so without @@ -55,6 +56,89 @@ def update_timestamp(tx_bytes: bytes, *, delta: int = 0) -> bytes: return bytes(tx) +class FakeResponse: + """Stand-in for an aiohttp response, usable as an async context manager.""" + + def __init__(self, status, payload): + self.status = status + self._payload = payload + + async def json(self): + return self._payload + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc_info): + return False + + +class FakeSession: + """Stand-in for HathorClient._session, serving GET /transaction?id=... + + The dev-miner reads parent timestamps straight off the fullnode, so the + tests need a session that answers that one endpoint. Requests are recorded + so tests can assert on caching. + """ + + def __init__(self, timestamps=None, *, status=200): + # Maps hex tx id -> timestamp. A hash absent from the map is answered + # the way the fullnode answers an unknown tx: success=False. + self.timestamps = timestamps or {} + self.status = status + self.requested_ids = [] + + def get(self, url, params=None, **kwargs): + # **kwargs mirrors aiohttp's real signature, which takes `timeout` and + # friends — without it the caller's TypeError is swallowed by the + # helper's own degradation path and every lookup silently returns None. + tx_id = (params or {}).get("id") + self.requested_ids.append(tx_id) + timestamp = self.timestamps.get(tx_id) + if timestamp is None: + return FakeResponse(self.status, {"success": False}) + return FakeResponse( + self.status, {"success": True, "tx": {"timestamp": timestamp}} + ) + + +class FixedPayloadSession: + """Serves one caller-supplied payload with HTTP 200, well-formed or not.""" + + def __init__(self, payload): + self.payload = payload + self.requested_ids = [] + + def get(self, url, params=None, **kwargs): + self.requested_ids.append((params or {}).get("id")) + return FakeResponse(200, self.payload) + + +def make_backend(parents=None, *, parent_timestamps=None, session_status=200): + """Mock HathorClient with the session and URL builder the lookup needs. + + Predecessors are unknown unless `parent_timestamps` names them. + """ + parents = parents if parents is not None else [b"\x00" * 32, b"\x01" * 32] + backend = MagicMock() + backend.get_tx_parents = AsyncMock(return_value=parents) + backend.push_tx_or_block = AsyncMock(return_value=True) + backend._session = FakeSession(parent_timestamps, status=session_status) + backend._get_url = lambda path: f"http://fullnode/v1a/{path}" + return backend + + +def make_backend_with_session(session, parents=None): + """Build a mocked HathorClient around an arbitrary session stand-in.""" + parents = parents if parents is not None else [b"\x00" * 32, b"\x01" * 32] + backend = MagicMock() + backend.get_tx_parents = AsyncMock(return_value=parents) + backend.push_tx_or_block = AsyncMock(return_value=True) + backend._session = session + backend._get_url = lambda path: f"http://fullnode/v1a/{path}" + return backend + + # --------------------------------------------------------------------------- # Transaction PoW tests # --------------------------------------------------------------------------- @@ -106,11 +190,7 @@ class TestDevMiningManager(AioHTTPTestCase): __test__ = True async def get_application(self): - self.backend = MagicMock() - self.backend.get_tx_parents = AsyncMock( - return_value=[b"\x00" * 32, b"\x01" * 32] - ) - self.backend.push_tx_or_block = AsyncMock(return_value=True) + self.backend = make_backend() self.manager = DevMiningManager(backend=self.backend) await self.manager.start() self.healthcheck = MagicMock() @@ -260,6 +340,312 @@ async def test_duplicate_job_submission(self): ) +class TestParentTimestampClamp(unittest.IsolatedAsyncioTestCase): + """The stamped timestamp must come out strictly above every parent's. + + The fullnode enforces `tx.timestamp > parent.timestamp` at 1-second + granularity. Wall-clock time alone does not satisfy that when a parent was + stamped in the same second — the rejection reported in + HathorNetwork/tx-mining-service#172. + """ + + __test__ = True + + # A fixed "now" so the expected timestamps are exact rather than windowed. + NOW = 1785951640 + + def setUp(self): + txstratum.time.set_time_function(lambda: float(self.NOW)) + + def tearDown(self): + txstratum.time.set_time_function(None) + + async def test_uses_now_when_parents_are_older(self): + """The common case: nothing to clamp, so "now" stands unchanged.""" + parent = b"\x0a" * 32 + backend = make_backend(parent_timestamps={parent.hex(): self.NOW - 10}) + manager = DevMiningManager(backend=backend) + self.assertEqual(self.NOW, await manager._get_timestamp_for([parent])) + + async def test_clamps_past_a_same_second_parent(self): + """The reported bug: parent stamped this second, so we must exceed it.""" + parent = b"\x0a" * 32 + backend = make_backend(parent_timestamps={parent.hex(): self.NOW}) + manager = DevMiningManager(backend=backend) + self.assertEqual(self.NOW + 1, await manager._get_timestamp_for([parent])) + + async def test_clamps_past_the_newest_parent(self): + """With two parents, only the newest constrains the result.""" + older, newer = b"\x0a" * 32, b"\x0b" * 32 + backend = make_backend( + parent_timestamps={older.hex(): self.NOW - 5, newer.hex(): self.NOW} + ) + manager = DevMiningManager(backend=backend) + self.assertEqual(self.NOW + 1, await manager._get_timestamp_for([older, newer])) + + async def test_clamps_past_a_parent_already_ahead_of_now(self): + """A same-second chain, which is what defeats a blind `now + 1`. + + The parent was itself clamped to NOW + 1 a moment ago. Stamping this tx + `now + 1` would tie with it and be rejected; reading the parent's real + timestamp is what makes the chain resolve. + """ + parent = b"\x0a" * 32 + backend = make_backend(parent_timestamps={parent.hex(): self.NOW + 1}) + manager = DevMiningManager(backend=backend) + self.assertEqual(self.NOW + 2, await manager._get_timestamp_for([parent])) + + async def test_falls_back_to_now_when_parent_is_unknown(self): + """An unreadable parent degrades to the old behavior, never an error.""" + backend = make_backend(parent_timestamps={}) + manager = DevMiningManager(backend=backend) + self.assertEqual(self.NOW, await manager._get_timestamp_for([b"\x0a" * 32])) + + async def test_falls_back_to_now_on_http_error(self): + parent = b"\x0a" * 32 + backend = make_backend( + parent_timestamps={parent.hex(): self.NOW}, session_status=500 + ) + manager = DevMiningManager(backend=backend) + self.assertEqual(self.NOW, await manager._get_timestamp_for([parent])) + + async def test_no_parents_skips_the_lookup_entirely(self): + backend = make_backend() + manager = DevMiningManager(backend=backend) + self.assertEqual(self.NOW, await manager._get_timestamp_for([])) + self.assertEqual([], backend._session.requested_ids) + + async def test_parent_timestamps_are_cached(self): + """A confirmed tx's timestamp is immutable, so one read per hash.""" + parent = b"\x0a" * 32 + backend = make_backend(parent_timestamps={parent.hex(): self.NOW}) + manager = DevMiningManager(backend=backend) + await manager._get_timestamp_for([parent]) + await manager._get_timestamp_for([parent]) + self.assertEqual([parent.hex()], backend._session.requested_ids) + + async def test_clamps_past_a_spent_tx_that_is_not_a_parent(self): + """Inputs are checked too, by a different verifier than parents. + + `transaction_verifier` rejects `tx.timestamp <= spent_tx.timestamp` for + every input, independently of the parent rule. A spent tx that + something else already confirmed is not a tip, so it can never be + chosen as a parent — clamping only above parents would miss it. + """ + tx = tx_or_block_from_bytes(TX1_DATA) + tx.weight = 1.0 + spent = tx.inputs[0].tx_id + self.assertNotIn(spent, tx.parents) + + # Only the spent tx has a known timestamp; the parents are unreadable, + # so a parents-only clamp would fall back to NOW and be rejected. + backend = make_backend_with_session( + FakeSession({spent.hex(): self.NOW}), parents=list(tx.parents) + ) + manager = DevMiningManager(backend=backend) + job = TxJob(bytes(tx)) + + await manager._mine_job(job) + + self.assertEqual(JobStatus.DONE, job.status) + self.assertEqual(self.NOW + 1, job.get_tx().timestamp) + self.assertIn(spent.hex(), backend._session.requested_ids) + + async def test_duplicate_predecessors_are_fetched_once(self): + """A tx often spends several outputs of the same parent.""" + shared = b"\x0a" * 32 + backend = make_backend(parent_timestamps={shared.hex(): self.NOW}) + manager = DevMiningManager(backend=backend) + + self.assertEqual( + self.NOW + 1, await manager._get_timestamp_for([shared, shared, shared]) + ) + self.assertEqual([shared.hex()], backend._session.requested_ids) + + async def test_cache_evicts_the_least_recently_used_entry(self): + """Eviction order, not just the bound. + + A plain FIFO would rotate out the hot tips that nearly every job + re-reads, costing a fullnode round-trip per predecessor per job. + """ + keys = [bytes([i]) * 32 for i in (1, 2, 3)] + backend = make_backend( + parent_timestamps={k.hex(): self.NOW - 100 for k in keys} + ) + manager = DevMiningManager(backend=backend) + manager.PARENT_TIMESTAMP_CACHE_SIZE = 2 + + await manager._get_timestamp_for([keys[0]]) + await manager._get_timestamp_for([keys[1]]) + await manager._get_timestamp_for([keys[0]]) # keys[0] becomes the MRU + await manager._get_timestamp_for([keys[2]]) # evicts the LRU, keys[1] + + self.assertEqual([keys[0], keys[2]], list(manager._parent_timestamps)) + + async def test_cache_does_not_grow_without_bound(self): + parents = [bytes([i]) * 32 for i in range(1, 11)] + backend = make_backend( + parent_timestamps={p.hex(): self.NOW - 100 for p in parents} + ) + manager = DevMiningManager(backend=backend) + manager.PARENT_TIMESTAMP_CACHE_SIZE = 4 + for parent in parents: + await manager._get_timestamp_for([parent]) + self.assertLessEqual(len(manager._parent_timestamps), 4) + + async def test_malformed_payloads_degrade_instead_of_raising(self): + """A structurally invalid payload must behave like an unreadable parent. + + This helper runs inside the mining task, so an exception escaping it + propagates through asyncio.gather into _mine_job and strands the job in + MINING with no cleanup scheduled — see the companion test below. + """ + malformed = [ + ("json array", []), + ("null", None), + ("string", "not json"), + ("tx is an array", {"success": True, "tx": []}), + ("tx is null", {"success": True, "tx": None}), + ("tx is a string", {"success": True, "tx": "nope"}), + ("timestamp missing", {"success": True, "tx": {}}), + ("timestamp is a string", {"success": True, "tx": {"timestamp": "1"}}), + ] + for label, payload in malformed: + with self.subTest(payload=label): + backend = make_backend_with_session(FixedPayloadSession(payload)) + manager = DevMiningManager(backend=backend) + self.assertEqual( + self.NOW, await manager._get_timestamp_for([b"\x0a" * 32]) + ) + + async def test_malformed_payload_does_not_strand_the_job(self): + """The job still reaches DONE rather than being left in MINING.""" + tx = tx_or_block_from_bytes(TX1_DATA) + tx.weight = 1.0 + backend = make_backend_with_session( + FixedPayloadSession({"success": True, "tx": []}), parents=list(tx.parents) + ) + manager = DevMiningManager(backend=backend) + job = TxJob(bytes(tx)) + + await manager._mine_job(job) + + self.assertEqual(JobStatus.DONE, job.status) + self.assertEqual(self.NOW, job.get_tx().timestamp) + + async def test_missing_private_client_attribute_degrades(self): + """A hathorlib rename of `_session` degrades to wall-clock time. + + The helper reaches into HathorClient's private attributes, so an + upstream rename must fall back rather than raise. + """ + backend = MagicMock() + backend.get_tx_parents = AsyncMock(return_value=[]) + del backend._session + manager = DevMiningManager(backend=backend) + self.assertEqual(self.NOW, await manager._get_timestamp_for([b"\x0a" * 32])) + + async def test_mined_tx_carries_the_clamped_timestamp(self): + """The clamp reaches the solved tx, not just the helper in isolation.""" + tx = tx_or_block_from_bytes(TX1_DATA) + tx.weight = 1.0 + backend = make_backend( + parent_timestamps={p.hex(): self.NOW for p in tx.parents} + ) + manager = DevMiningManager(backend=backend) + job = TxJob(bytes(tx)) + + await manager._mine_job(job) + + self.assertEqual(JobStatus.DONE, job.status) + self.assertEqual(self.NOW + 1, job.get_tx().timestamp) + self.assertTrue(job.get_tx().verify_pow()) + + async def test_client_timestamp_is_honoured_when_it_leads(self): + """The client picked the inputs, so its timestamp is information. + + It is the one party that already knows what its inputs spend, and + api.py has bounded the value to MAX_TIMESTAMP_DELTA of now before we + see it. A client that clamps above its own inputs therefore needs no + per-input lookup here at all. + """ + tx = tx_or_block_from_bytes(TX1_DATA) + tx.weight = 1.0 + tx.timestamp = self.NOW + 50 + backend = make_backend( + parent_timestamps={p.hex(): self.NOW - 10 for p in tx.parents} + ) + manager = DevMiningManager(backend=backend) + job = TxJob(bytes(tx)) + + await manager._mine_job(job) + + self.assertEqual(self.NOW + 50, job.get_tx().timestamp) + + async def test_client_timestamp_never_lowers_the_clamp(self): + """Honouring the client is a floor, never a ceiling. + + A stale or careless client value must not defeat the parent clamp — + otherwise trusting it would reintroduce the rejection this fixes. + """ + tx = tx_or_block_from_bytes(TX1_DATA) + tx.weight = 1.0 + tx.timestamp = self.NOW - 100 + backend = make_backend( + parent_timestamps={p.hex(): self.NOW for p in tx.parents} + ) + manager = DevMiningManager(backend=backend) + job = TxJob(bytes(tx)) + + await manager._mine_job(job) + + self.assertEqual(self.NOW + 1, job.get_tx().timestamp) + + async def test_our_own_output_is_remembered(self): + """A tx mined here is the likeliest predecessor of the next job. + + Reading it back from the fullnode would be a round-trip for a timestamp + this service chose itself moments ago — and a chained flock is, by + construction, mostly this service's own output. + """ + tx = tx_or_block_from_bytes(TX1_DATA) + tx.weight = 1.0 + backend = make_backend( + parent_timestamps={p.hex(): self.NOW for p in tx.parents} + ) + manager = DevMiningManager(backend=backend) + job = TxJob(bytes(tx)) + + await manager._mine_job(job) + mined = job.get_tx() + backend._session.requested_ids.clear() + + self.assertEqual( + mined.timestamp + 1, await manager._get_timestamp_for([mined.hash]) + ) + self.assertEqual([], backend._session.requested_ids) + + async def test_remembered_timestamp_is_the_one_actually_stamped(self): + """What gets recorded must be the final value, not a pre-clamp draft. + + The client's floor is applied after the predecessor lookups, so + recording the wrong one would under-clamp every tx chained onto this. + """ + tx = tx_or_block_from_bytes(TX1_DATA) + tx.weight = 1.0 + tx.timestamp = self.NOW + 50 + backend = make_backend( + parent_timestamps={p.hex(): self.NOW for p in tx.parents} + ) + manager = DevMiningManager(backend=backend) + job = TxJob(bytes(tx)) + + await manager._mine_job(job) + mined = job.get_tx() + + self.assertEqual(self.NOW + 50, manager._parent_timestamps[mined.hash]) + + class TestSolveBlock(AioHTTPTestCase): """Test the block PoW solver (solve_block).""" diff --git a/txstratum/dev/manager.py b/txstratum/dev/manager.py index b076bc6..a0f0a33 100644 --- a/txstratum/dev/manager.py +++ b/txstratum/dev/manager.py @@ -25,8 +25,10 @@ """ import asyncio +from collections import OrderedDict from typing import TYPE_CHECKING, Any, Dict, List, Optional +import aiohttp from structlog import get_logger import txstratum.time @@ -55,6 +57,13 @@ class DevMiningManager: TX_CLEAN_UP_INTERVAL = 300.0 # seconds + # Upper bound for the timestamp cache. A tx's timestamp is covered by its + # hash, so a hash->timestamp entry can never go stale whatever happens to + # the tx afterwards; the bound only keeps the dict from growing over a long + # run. Note these hashes are mostly DAG tips — the least confirmed vertices + # in the graph — so confirmation state is deliberately not part of this. + PARENT_TIMESTAMP_CACHE_SIZE = 1024 + def __init__(self, backend: "HathorClient"): """Initialize the dev mining manager.""" self.log = logger.new() @@ -63,6 +72,7 @@ def __init__(self, backend: "HathorClient"): self.tx_jobs: Dict[bytes, TxJob] = {} self._tasks: Dict[bytes, asyncio.Task[None]] = {} self.refuse_new_jobs = False + self._parent_timestamps: "OrderedDict[bytes, int]" = OrderedDict() # Statistics — same fields as TxMiningManager for API compatibility. self.txs_solved: int = 0 @@ -182,6 +192,122 @@ async def _mine_with_parents(self, job: TxJob) -> None: job.set_parents(parents) await self._mine_job(job) + async def _fetch_tx_timestamp(self, tx_hash: bytes) -> Optional[int]: + """Read one transaction's timestamp from the fullnode, or None. + + hathorlib's HathorClient has no "get transaction by hash", and this fix + is deliberately confined to this repo, so the client's session and URL + builder are used directly. Both are private, which is exactly why this + is the only place that touches them: if hathorlib later grows a proper + accessor (or the parents endpoint starts returning timestamps, as + HathorNetwork/tx-mining-service#172 sketches), this body is the only + thing that has to change. + + Returns None on any failure — a timestamp we cannot read is one we + cannot clamp against, and degrading to plain wall-clock time is far + better than failing the job. That promise has to hold absolutely: this + runs inside the mining task, so an exception escaping here strands the + job in MINING with no cleanup scheduled. Hence both the private-attribute + access and the payload parsing below are guarded. + """ + cached = self._parent_timestamps.get(tx_hash) + if cached is not None: + self._parent_timestamps.move_to_end(tx_hash) + return cached + + try: + # Inside the try on purpose: these attributes are private, so a + # rename in a future hathorlib degrades to wall-clock time with a + # warning instead of stranding every job that needs a parent. + session = self.backend._session + if session is None: + return None + url = self.backend._get_url("transaction") + # An explicit timeout: the session carries aiohttp's 5-minute + # default, and this manager never arms a job timeout, so a stalled + # fullnode would otherwise hold the job in MINING for that long. + # A lookup slower than this is worthless anyway — the clock it is + # being compared against has already moved on. + async with session.get( + url, + params={"id": tx_hash.hex()}, + timeout=aiohttp.ClientTimeout(total=2), + ) as resp: + if resp.status != 200: + self.log.warn( + "Parent timestamp lookup failed", + parent=tx_hash.hex(), + status=resp.status, + ) + return None + data = await resp.json() + except Exception as e: + self.log.warn( + "Could not read parent timestamp", + parent=tx_hash.hex(), + error=f"{type(e).__name__}: {e}", + ) + return None + + # The payload is not assumed well-formed. A JSON array, a null, or a + # non-object "tx" must degrade like any other unreadable predecessor + # rather than raise, so every access below is type-checked first. + if not isinstance(data, dict): + self.log.warn("Malformed transaction payload", parent=tx_hash.hex()) + return None + if not data.get("success"): + # Expected: the fullnode has not indexed this tx yet. Not logged — + # the clamp exists for txs broadcast moments ago, so this is the + # ordinary case rather than a fault. + return None + tx_data = data.get("tx") + timestamp = tx_data.get("timestamp") if isinstance(tx_data, dict) else None + if not isinstance(timestamp, int): + self.log.warn("Malformed transaction payload", parent=tx_hash.hex()) + return None + + self._remember_timestamp(tx_hash, timestamp) + return timestamp + + def _remember_timestamp(self, tx_hash: bytes, timestamp: int) -> None: + """Record one hash's timestamp, evicting the oldest past the bound.""" + self._parent_timestamps[tx_hash] = timestamp + if len(self._parent_timestamps) > self.PARENT_TIMESTAMP_CACHE_SIZE: + self._parent_timestamps.popitem(last=False) + + async def _get_timestamp_for(self, predecessors: List[bytes]) -> int: + """Return a timestamp strictly greater than every predecessor's. + + The fullnode enforces this twice, in two separate verifiers, and both + must hold: a tx must beat each of its parents (`vertex_verifier`) and + each tx it spends (`transaction_verifier`). Neither set contains the + other — parents are drawn from the current tips, while a spent tx may + have been confirmed long ago and so can never be selected as a parent. + Pass both. + + Wall-clock time alone does not satisfy either rule at 1-second + granularity: two txs in the same second, where one precedes the other, + produce a rejection the client cannot prevent or repair, since both the + timestamp and the parents are assigned here, under the nonce. + + The lead over wall-clock grows only along a chain of same-second + predecessors, at most one second per link, against hathor-core's + 300s MAX_FUTURE_TIMESTAMP_ALLOWED. Sustained chaining above one tx per + second could in principle reach that ceiling; measured workloads stay + at a lead of one second. See HathorNetwork/tx-mining-service#172 for + the full analysis and why deferring is not worth its livelock risk. + """ + # dict.fromkeys dedups while preserving order: several inputs commonly + # spend the same tx, and a parent is often an input too. + timestamps: List[Optional[int]] = list( + await asyncio.gather( + *(self._fetch_tx_timestamp(h) for h in dict.fromkeys(predecessors)) + ) + ) + # Read the clock after the lookups, not before — they are network I/O. + now = int(txstratum.time.time()) + return max([now, *(ts + 1 for ts in timestamps if ts is not None)]) + async def _mine_job(self, job: TxJob) -> None: """Solve PoW for a transaction job. @@ -195,9 +321,23 @@ async def _mine_job(self, job: TxJob) -> None: start = txstratum.time.time() tx = job.get_tx() - # Update timestamp to current time — the fullnode validates that tx - # timestamps are within an acceptable delta of the current time. - tx.timestamp = int(txstratum.time.time()) + # Stamp above everything the fullnode compares this tx against: its DAG + # parents and every tx it spends. Both are checked, in separate + # verifiers, and a spent tx is frequently not among the parents. + # + # The client's own timestamp is a floor here, not something to discard. + # It chose the inputs, so it is the one party that already knows what + # they spend — and api.py has already bounded it to within + # MAX_TIMESTAMP_DELTA of now, so honouring it cannot push this tx past + # what the fullnode accepts. A client that clamps above its own inputs + # therefore needs no per-input lookup here at all, which is the only + # version of this that survives a tx with 200 inputs. + tx.timestamp = max( + tx.timestamp, + await self._get_timestamp_for( + [*tx.parents, *(txin.tx_id for txin in tx.inputs)] + ), + ) loop = asyncio.get_event_loop() solved = await loop.run_in_executor(None, solve_tx, tx) @@ -207,6 +347,14 @@ async def _mine_job(self, job: TxJob) -> None: self._schedule_cleanup(job) return + # Everything mined here is a candidate predecessor for the next job: a + # chained flock arriving at one service is, by construction, mostly this + # service's own recent output. Recording it turns those lookups into + # memory reads at exactly the concurrency that makes the clamp + # load-bearing — and the hash covers the timestamp, so the entry is + # valid whatever happens to the tx afterwards. + self._remember_timestamp(tx.hash, tx.timestamp) + elapsed_ms = (txstratum.time.time() - start) * 1000 nonce = tx.get_struct_nonce()