diff --git a/allways/validator/event_watcher.py b/allways/validator/event_watcher.py index 379d762b..c94507e3 100644 --- a/allways/validator/event_watcher.py +++ b/allways/validator/event_watcher.py @@ -991,6 +991,14 @@ def _emit_reservation_pin_ends(self, block_num: int, miner: str) -> None: rate=0.0, ) + def expire_stale_reservation_pins(self) -> int: + """Close crown pins for reservations that lapsed without a swap (no + contract event fires on natural expiry), then purge them. End emitted at + ``reserved_until + 1`` so crown stops at the last live block. Returns rows purged.""" + for pin in self.state_store.get_expired_reservation_pins(): + self._emit_reservation_pin_ends(pin.reserved_until + 1, pin.miner_hotkey) + return self.state_store.purge_expired_reservation_pins() + def record_active_transition(self, block_num: int, hotkey: str, active: bool) -> None: """Apply an on-chain active-flag transition to both the current-state snapshot and the historical event log. A no-op if the flag already diff --git a/allways/validator/forward.py b/allways/validator/forward.py index de2d0c2d..49d15e4a 100644 --- a/allways/validator/forward.py +++ b/allways/validator/forward.py @@ -80,7 +80,10 @@ async def forward(self: Validator) -> None: f'forward: purged {purged} expired pending_confirms (reservations elapsed without tx confirmation)' ) - purged_pins = self.state_store.purge_expired_reservation_pins() + # Emits crown pin-end events for expired reservations *before* purging, so a + # reservation that lapses without a swap can't keep earning crown at its + # pinned rate. Runs before scoring in this same forward pass. + purged_pins = self.event_watcher.expire_stale_reservation_pins() if purged_pins: bt.logging.info(f'forward: purged {purged_pins} expired reservation_pins (reservations elapsed without a swap)') diff --git a/allways/validator/state_store.py b/allways/validator/state_store.py index 970b081a..09dcfab7 100644 --- a/allways/validator/state_store.py +++ b/allways/validator/state_store.py @@ -267,6 +267,22 @@ def update_reservation_pin_reserved_until(self, miner_hotkey: str, reserved_unti (reserved_until, miner_hotkey), ) + def get_expired_reservation_pins(self) -> List[ReservationPin]: + """Pins whose reservation has lapsed as of the current block. + + Read before ``purge_expired_reservation_pins`` so the caller can emit a + scoring pin-end event per expired pin — otherwise the crown overlay's + 'start' outlives the on-chain reservation and keeps earning crown at the + pinned rate after expiry. + """ + if self.current_block_fn is None: + return [] + rows = self._fetchall( + 'SELECT * FROM reservation_pins WHERE reserved_until < ?', + (self.current_block_fn(),), + ) + return [self.row_to_reservation_pin(row) for row in rows] + def purge_expired_reservation_pins(self) -> int: """Drop pins whose reservation has already expired.""" if self.current_block_fn is None: diff --git a/tests/test_event_watcher.py b/tests/test_event_watcher.py index 9b8f50d5..0c1b1ba8 100644 --- a/tests/test_event_watcher.py +++ b/tests/test_event_watcher.py @@ -567,6 +567,66 @@ def test_swap_timed_out_clears_pin(self, tmp_path: Path): assert w.state_store.get_reservation_pin('hk_a') is None w.state_store.close() + def test_expired_reservation_emits_pin_end_and_purges(self, tmp_path: Path): + """A reservation that lapses without a swap emits no contract event, so + nothing closes the crown pin. ``expire_stale_reservation_pins`` must + emit an 'end' (at reserved_until + 1) for every open direction and purge + the synchronous pin row — otherwise the miner keeps earning crown at the + pinned rate with no live reservation.""" + w = make_watcher(tmp_path, netuid=2, subtensor=MagicMock()) + w.state_store.current_block_fn = lambda: 1001 # past reserved_until=1000 + with patch( + 'allways.validator.event_watcher.read_miner_commitment', + return_value=make_pinned_commitment(), + ): + w.apply_event(900, 'MinerReserved', {'miner': 'hk_a', 'reserved_until': 1000}) + + purged = w.expire_stale_reservation_pins() + + assert purged == 1 + assert w.state_store.get_reservation_pin('hk_a') is None + ends = [ev for ev in w.reservation_pin_events if ev.hotkey == 'hk_a' and ev.kind == 'end'] + # One 'end' per pinned direction (primary btc→tao + counter tao→btc), + # at reserved_until + 1 so crown is credited through the last live block. + assert {(ev.from_chain, ev.to_chain) for ev in ends} == {('btc', 'tao'), ('tao', 'btc')} + assert all(ev.block_num == 1001 for ev in ends) + w.state_store.close() + + def test_live_reservation_not_closed_by_expiry_sweep(self, tmp_path: Path): + """A reservation whose TTL is still in the future must be left alone — + no pin-end, no purge.""" + w = make_watcher(tmp_path, netuid=2, subtensor=MagicMock()) + w.state_store.current_block_fn = lambda: 950 # before reserved_until=1000 + with patch( + 'allways.validator.event_watcher.read_miner_commitment', + return_value=make_pinned_commitment(), + ): + w.apply_event(900, 'MinerReserved', {'miner': 'hk_a', 'reserved_until': 1000}) + + purged = w.expire_stale_reservation_pins() + + assert purged == 0 + assert w.state_store.get_reservation_pin('hk_a') is not None + assert not [ev for ev in w.reservation_pin_events if ev.hotkey == 'hk_a' and ev.kind == 'end'] + w.state_store.close() + + def test_expiry_sweep_is_idempotent(self, tmp_path: Path): + """Re-running after the row is purged emits no further 'end' events.""" + w = make_watcher(tmp_path, netuid=2, subtensor=MagicMock()) + w.state_store.current_block_fn = lambda: 1001 + with patch( + 'allways.validator.event_watcher.read_miner_commitment', + return_value=make_pinned_commitment(), + ): + w.apply_event(900, 'MinerReserved', {'miner': 'hk_a', 'reserved_until': 1000}) + + assert w.expire_stale_reservation_pins() == 1 + ends_after_first = [ev for ev in w.reservation_pin_events if ev.kind == 'end'] + assert w.expire_stale_reservation_pins() == 0 + ends_after_second = [ev for ev in w.reservation_pin_events if ev.kind == 'end'] + assert ends_after_second == ends_after_first + w.state_store.close() + def test_reservation_extension_finalized_bumps_pin_ttl(self, tmp_path: Path): w = make_watcher(tmp_path, netuid=2, subtensor=MagicMock()) with patch( diff --git a/tests/test_scoring_v1.py b/tests/test_scoring_v1.py index 952e84c7..ecff2305 100644 --- a/tests/test_scoring_v1.py +++ b/tests/test_scoring_v1.py @@ -26,7 +26,7 @@ snapshot_current_crown_holders, success_rate, ) -from allways.validator.state_store import ValidatorStateStore +from allways.validator.state_store import ReservationPin, ValidatorStateStore POOL_TAO_BTC = 0.5 POOL_BTC_TAO = 0.5 @@ -934,6 +934,68 @@ def test_pin_end_lets_live_rate_take_over(self, tmp_path: Path): assert crown == {'hk_a': 750.0, 'hk_b': 250.0} store.close() + def test_expired_reservation_pin_stops_earning_crown(self, tmp_path: Path): + """End-to-end: a reservation that lapses without a swap must stop + earning crown at the pinned rate. ``expire_stale_reservation_pins`` + emits the missing pin-end at reserved_until + 1, so once the live rate + reverts to junk after expiry the miner can no longer crown-squat on the + stale pinned rate.""" + store = ValidatorStateStore(db_path=tmp_path / 'state.db') + watcher = make_watcher(store, active={'hk_a', 'hk_b'}) + conn = store.require_connection() + for hk in ('hk_a', 'hk_b'): + conn.execute( + 'INSERT INTO rate_events (hotkey, from_chain, to_chain, rate, block) VALUES (?, ?, ?, ?, ?)', + (hk, 'btc', 'tao', 200.0, 0), + ) + # A bumps its live rate to junk at block 700 — only the stale pin would + # keep it winning crown at 200 past expiry. + conn.execute( + 'INSERT INTO rate_events (hotkey, from_chain, to_chain, rate, block) VALUES (?, ?, ?, ?, ?)', + ('hk_a', 'btc', 'tao', 25000.0, 700), + ) + conn.commit() + + # A is reserved at block 300, TTL through 600, and never swaps — the + # synchronous pin row + the scoring 'start' both land, but no terminal + # event ever closes the pin. + store.upsert_reservation_pin( + ReservationPin( + miner_hotkey='hk_a', + reserve_block=300, + from_chain='btc', + to_chain='tao', + rate_str='200', + counter_rate_str='0', + miner_from_address='bc1-a', + miner_to_address='5a', + reserved_until=600, + created_at=1.0, + ) + ) + watcher._record_reservation_pin_event( + block_num=300, hotkey='hk_a', from_chain='btc', to_chain='tao', kind='start', rate=200.0 + ) + + # The forward loop's expiry sweep fires once the block passes the TTL. + store.current_block_fn = lambda: 601 + watcher.expire_stale_reservation_pins() + + crown = replay_crown_time_window( + store=store, + event_watcher=watcher, + from_chain='btc', + to_chain='tao', + window_start=100, + window_end=1100, + rewardable_hotkeys={'hk_a', 'hk_b'}, + ) + # (100, 601]: A pinned at 200, tie with B → 250.5 each. + # (601, 700]: pin closed, A live at 200, tie with B → 49.5 each. + # (700, 1100]: A live at junk 25000 — wins the rate sort, earns 400. + assert crown == {'hk_a': 700.0, 'hk_b': 300.0} + store.close() + def test_pin_only_applies_to_pinned_direction(self, tmp_path: Path): """A reservation pin is direction-specific — pinning btc→tao must not affect crown ranking in the tao→btc direction.""" diff --git a/tests/test_state_store.py b/tests/test_state_store.py index 20384a33..af3ae49d 100644 --- a/tests/test_state_store.py +++ b/tests/test_state_store.py @@ -113,6 +113,26 @@ def test_purge_without_current_block_fn_is_noop(self, tmp_path: Path): assert store.get_reservation_pin('miner-1') == PIN_SAMPLE1 store.close() + def test_get_expired_returns_only_expired_rows(self, tmp_path: Path): + store = ValidatorStateStore( + db_path=tmp_path / 'state.db', + current_block_fn=lambda: 1001, + ) + store.upsert_reservation_pin(PIN_SAMPLE1) # reserved_until=1000 → expired at 1001 + store.upsert_reservation_pin(PIN_SAMPLE2) # reserved_until=1005 → still live + + expired = store.get_expired_reservation_pins() + assert expired == [PIN_SAMPLE1] + # Read-only: the row is left for the caller to emit a pin-end before purging. + assert store.get_reservation_pin('miner-1') == PIN_SAMPLE1 + store.close() + + def test_get_expired_without_current_block_fn_is_empty(self, tmp_path: Path): + store = ValidatorStateStore(db_path=tmp_path / 'state.db') + store.upsert_reservation_pin(PIN_SAMPLE1) + assert store.get_expired_reservation_pins() == [] + store.close() + def test_update_reserved_until_keeps_row_a_purge_would_drop(self, tmp_path: Path): """Regression: after the contract extends a reservation, refreshing the pin's reserved_until must keep it alive past its original TTL — exactly