diff --git a/.env.example b/.env.example index c220582a..8736e850 100644 --- a/.env.example +++ b/.env.example @@ -23,3 +23,8 @@ BIGTABLE_PROJECT= BIGTABLE_INSTANCE= BIGTABLE_TABLE_PREDICTION_LOW=miner_predictions_low BIGTABLE_TABLE_PREDICTION_HIGH=miner_predictions_high + +# Optional salt seeding density-tapering keeper selection: it decides +# which request per bucket is kept. Keep it stable. Unset falls back to +# md5(id). Any non-empty random string (e.g. `openssl rand -hex 32`). +THINNING_SALT= diff --git a/synth/validator/miner_data_handler.py b/synth/validator/miner_data_handler.py index 441ce93b..89933210 100644 --- a/synth/validator/miner_data_handler.py +++ b/synth/validator/miner_data_handler.py @@ -3,6 +3,7 @@ import typing import logging import math +import os import bittensor as bt @@ -85,6 +86,17 @@ def __init__( # When set, prediction payloads are shipped to Bigtable and the # Postgres `prediction` column holds a sentinel + `bigtable_key`. self.bigtable_storage = bigtable_storage + # Optional salt for density-tapering keeper selection. The kept + # validator_request per bucket is picked by md5(id || salt), so the + # keeper is spread across the bucket rather than always its earliest + # member. Unset falls back to md5(id); warn once so that's not + # silent. + self.thinning_salt = os.getenv("THINNING_SALT", "") + if not self.thinning_salt: + bt.logging.warning( + "THINNING_SALT is not set; density-tapering keeper " + "selection falls back to md5(id)." + ) def get_miner_uids(self, connection: Connection): ranked_miners = select( @@ -788,10 +800,16 @@ def density_tapering_predictions( Selects validator_requests with this cycle's `time_length` whose `start_time` is older than `prompt_config.thin_after_minutes`, buckets them by (asset, floor(epoch(start_time)/thin_bucket_seconds)), - keeps the smallest-id row per bucket, and soft-deletes every - miner_prediction under the non-keeper requests by setting - `deleted_at` and replacing `prediction` with a tombstone (same - pattern as `cleanup_old_history`). + keeps one row per bucket, and soft-deletes every miner_prediction + under the non-keeper requests by setting `deleted_at` and replacing + `prediction` with a tombstone (same pattern as `cleanup_old_history`). + + The keeper is the row with the smallest `md5(id || thinning_salt)`, + not the smallest id, so the kept request is spread across the bucket + rather than always being its earliest member. The hash is stable per + row (idempotent across runs): the global-min-hash row, once eligible, + is always rn=1 and never thinned, so every bucket keeps exactly one + row and converges to that keeper well before it enters scoring range. The most recent request per asset is additionally preserved (until it is older than `time_length`, i.e. enters scoring range) so @@ -822,7 +840,7 @@ def density_tapering_predictions( SELECT vr_id, ROW_NUMBER() OVER ( PARTITION BY asset, bucket - ORDER BY vr_id ASC + ORDER BY md5(vr_id::text || :salt), vr_id ASC ) AS rn FROM old ), @@ -862,6 +880,7 @@ def density_tapering_predictions( "thin_cutoff": thin_cutoff, "scoring_cutoff": scoring_cutoff, "now": now, + "salt": self.thinning_salt, }, ) except Exception as e: diff --git a/tests/conftest.py b/tests/conftest.py index cfff9ecf..7b30b78f 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -31,3 +31,11 @@ def db_engine(setup): engine = create_engine(os.environ["DB_URL_TEST"]) yield engine engine.dispose() + + +@pytest.fixture(scope="session", autouse=True) +def thinning_salt_env(): + # Provide a stable THINNING_SALT so the salted-hash keeper selection is + # reproducible across the suite. + os.environ["THINNING_SALT"] = "test-thinning-salt" + yield diff --git a/tests/test_miner_data_handler.py b/tests/test_miner_data_handler.py index b8563821..7d8bafdb 100644 --- a/tests/test_miner_data_handler.py +++ b/tests/test_miner_data_handler.py @@ -1,4 +1,6 @@ from datetime import datetime, timedelta +import hashlib +import os import pytest from sqlalchemy import Engine, select, delete @@ -665,6 +667,24 @@ def _fetch_prediction_state(connection, mp_id: int): ).fetchone() +def _split_alive_thinned(connection, mp_ids: list[int]): + """Partition the given prediction ids into (alive, thinned) after a + tapering run. Keeper selection is a salted hash of the request id, so + tests assert the survivor *count* per bucket, not which id survives. + Every thinned row must carry the tombstone.""" + alive, thinned = [], [] + for mp_id in mp_ids: + row = _fetch_prediction_state(connection, mp_id) + assert row is not None + if row.deleted_at is None: + assert isinstance(row.prediction, list) + alive.append(mp_id) + else: + assert row.prediction == {"deleted": True, "reason": "thinned"} + thinned.append(mp_id) + return alive, thinned + + def test_prune_leaves_recent_requests_untouched(db_engine: Engine): """Validator_requests newer than `thin_after_minutes` are not pruned — even when many in the same bucket. Short-term density survives for the @@ -701,32 +721,20 @@ def test_prune_leaves_recent_requests_untouched(db_engine: Engine): def test_prune_collapses_old_requests_in_same_bucket(db_engine: Engine): - """Three old validator_requests for the same asset land in the same - hour bucket → the smallest-id request is the keeper; every prediction - on the other two requests gets the `thinned` tombstone, including - predictions for additional miners on those requests.""" + """Several old validator_requests for the same asset land in the same + hour bucket → exactly one request is the keeper (salted-hash pick) and + every prediction on the other requests gets the `thinned` tombstone.""" bucket_anchor = datetime(2026, 1, 1, 12, 0, 0) + ids: list[int] = [] with db_engine.connect() as connection: with connection.begin(): miner_a = _insert_miner(connection, miner_uid=310) miner_b = _insert_miner(connection, miner_uid=311) - # Keeper request — smallest id, both miners. - kept_a = _insert_prediction( - connection, - miner_id=miner_a, - asset="SOL", - time_length=LOW_TEST_CONFIG.time_length, - start_time=bucket_anchor + timedelta(minutes=5), - created_at=bucket_anchor + timedelta(minutes=5), - payload=[[{"price": 1.0}]], - ) - # Extra request 1 in the same bucket — both miners. - extras: list[int] = [] - for minute in (25, 50): + for minute in (5, 25, 50): start = bucket_anchor + timedelta(minutes=minute) for miner_id in (miner_a, miner_b): - extras.append( + ids.append( _insert_prediction( connection, miner_id=miner_id, @@ -741,21 +749,16 @@ def test_prune_collapses_old_requests_in_same_bucket(db_engine: Engine): MinerDataHandler(db_engine).density_tapering_predictions(LOW_TEST_CONFIG) with db_engine.connect() as connection: - keeper = _fetch_prediction_state(connection, kept_a) - assert keeper is not None - assert keeper.deleted_at is None - assert keeper.prediction == [[{"price": 1.0}]] - - for mp_id in extras: - row = _fetch_prediction_state(connection, mp_id) - assert row is not None - assert row.deleted_at is not None - assert row.prediction == {"deleted": True, "reason": "thinned"} + alive, thinned = _split_alive_thinned(connection, ids) + # Each _insert_prediction creates its own validator_request, so the + # single surviving request contributes exactly one alive prediction. + assert len(alive) == 1 + assert len(thinned) == len(ids) - 1 def test_prune_keeps_one_request_per_asset_per_bucket(db_engine: Engine): """Different assets in the same bucket are independent — each asset - keeps its smallest-id request. Different buckets are independent — + keeps exactly one request. Different buckets are independent — each bucket keeps its own request.""" bucket_anchor = datetime(2026, 1, 1, 12, 0, 0) with db_engine.connect() as connection: @@ -804,11 +807,10 @@ def test_prune_keeps_one_request_per_asset_per_bucket(db_engine: Engine): MinerDataHandler(db_engine).density_tapering_predictions(LOW_TEST_CONFIG) with db_engine.connect() as connection: - assert _fetch_prediction_state(connection, btc_keep).deleted_at is None - assert ( - _fetch_prediction_state(connection, btc_prune).deleted_at - is not None - ) + # BTC bucket 1 has two requests → exactly one survives. + btc_alive, _ = _split_alive_thinned(connection, [btc_keep, btc_prune]) + assert len(btc_alive) == 1 + # Single-request buckets are always kept. assert _fetch_prediction_state(connection, eth_keep).deleted_at is None assert ( _fetch_prediction_state(connection, btc_next_bucket).deleted_at @@ -862,11 +864,9 @@ def test_prune_scopes_by_time_length(db_engine: Engine): MinerDataHandler(db_engine).density_tapering_predictions(LOW_TEST_CONFIG) with db_engine.connect() as connection: - assert _fetch_prediction_state(connection, low_kept).deleted_at is None - assert ( - _fetch_prediction_state(connection, low_extra).deleted_at - is not None - ) + # LOW bucket has two requests → exactly one survives. + low_alive, _ = _split_alive_thinned(connection, [low_kept, low_extra]) + assert len(low_alive) == 1 # HIGH untouched. assert _fetch_prediction_state(connection, high_a).deleted_at is None assert _fetch_prediction_state(connection, high_b).deleted_at is None @@ -910,13 +910,139 @@ def test_prune_high_bucket_is_ten_minutes(db_engine: Engine): MinerDataHandler(db_engine).density_tapering_predictions(HIGH_TEST_CONFIG) with db_engine.connect() as connection: - assert _fetch_prediction_state(connection, kept_a).deleted_at is None - assert ( - _fetch_prediction_state(connection, extra_a).deleted_at is not None - ) + # Two requests in the same 10-min bucket → exactly one survives. + alive, _ = _split_alive_thinned(connection, [kept_a, extra_a]) + assert len(alive) == 1 + # Adjacent bucket is independent → kept. assert _fetch_prediction_state(connection, kept_b).deleted_at is None +def test_thinning_without_salt_falls_back_to_md5_id( + db_engine: Engine, monkeypatch +): + """THINNING_SALT is optional: unset, the handler still constructs and + tapering collapses each bucket to exactly one keeper via md5(id).""" + monkeypatch.delenv("THINNING_SALT", raising=False) + bucket_anchor = datetime(2026, 1, 1, 7, 0, 0) + ids: list[int] = [] + with db_engine.connect() as connection: + with connection.begin(): + miner_id = _insert_miner(connection, miner_uid=370) + for minute in (5, 25, 45): + start = bucket_anchor + timedelta(minutes=minute) + ids.append( + _insert_prediction( + connection, + miner_id=miner_id, + asset="SOL", + time_length=LOW_TEST_CONFIG.time_length, + start_time=start, + created_at=start, + payload=[[{"price": float(minute)}]], + ) + ) + + MinerDataHandler(db_engine).density_tapering_predictions(LOW_TEST_CONFIG) + + with db_engine.connect() as connection: + vr_by_mp = { + mp_id: connection.execute( + select(MinerPrediction.validator_requests_id).where( + MinerPrediction.id == mp_id + ) + ).scalar_one() + for mp_id in ids + } + expected_keeper_mp = min( + ids, + key=lambda mp: hashlib.md5(str(vr_by_mp[mp]).encode()).hexdigest(), + ) + alive, thinned = _split_alive_thinned(connection, ids) + assert alive == [expected_keeper_mp] + assert len(thinned) == len(ids) - 1 + + +def test_tapering_keeper_is_stable_across_runs(db_engine: Engine): + """Keeper selection is deterministic per salt, so re-running tapering + keeps the same survivor (idempotent) instead of thinning the bucket + down to zero over successive runs.""" + bucket_anchor = datetime(2026, 1, 1, 9, 0, 0) + ids: list[int] = [] + with db_engine.connect() as connection: + with connection.begin(): + miner_id = _insert_miner(connection, miner_uid=350) + for minute in (5, 20, 35, 50): + start = bucket_anchor + timedelta(minutes=minute) + ids.append( + _insert_prediction( + connection, + miner_id=miner_id, + asset="XRP", + time_length=LOW_TEST_CONFIG.time_length, + start_time=start, + created_at=start, + payload=[[{"price": float(minute)}]], + ) + ) + + handler = MinerDataHandler(db_engine) + handler.density_tapering_predictions(LOW_TEST_CONFIG) + with db_engine.connect() as connection: + alive_first, _ = _split_alive_thinned(connection, ids) + assert len(alive_first) == 1 + + handler.density_tapering_predictions(LOW_TEST_CONFIG) + with db_engine.connect() as connection: + alive_second, _ = _split_alive_thinned(connection, ids) + assert alive_second == alive_first + + +def test_tapering_keeper_matches_salted_hash(db_engine: Engine): + """The surviving request is the one with the smallest md5(id || salt), + proving the keeper is driven by the secret salt and not by id order.""" + salt = os.environ["THINNING_SALT"] + bucket_anchor = datetime(2026, 1, 1, 8, 0, 0) + mp_ids: list[int] = [] + with db_engine.connect() as connection: + with connection.begin(): + miner_id = _insert_miner(connection, miner_uid=360) + for minute in (2, 12, 22, 32, 42, 52): + start = bucket_anchor + timedelta(minutes=minute) + mp_ids.append( + _insert_prediction( + connection, + miner_id=miner_id, + asset="XAU", + time_length=LOW_TEST_CONFIG.time_length, + start_time=start, + created_at=start, + payload=[[{"price": float(minute)}]], + ) + ) + + with db_engine.connect() as connection: + vr_by_mp = { + mp_id: connection.execute( + select(MinerPrediction.validator_requests_id).where( + MinerPrediction.id == mp_id + ) + ).scalar_one() + for mp_id in mp_ids + } + expected_keeper_mp = min( + mp_ids, + key=lambda mp: hashlib.md5( + f"{vr_by_mp[mp]}{salt}".encode() + ).hexdigest(), + ) + + MinerDataHandler(db_engine).density_tapering_predictions(LOW_TEST_CONFIG) + + with db_engine.connect() as connection: + alive, _ = _split_alive_thinned(connection, mp_ids) + assert alive == [expected_keeper_mp] + + def test_scoring_path_skips_thinned_requests(db_engine: Engine): """Regression: density_tapering_predictions tombstones every prediction on redundant validator_requests ~30 min after start_time, but those @@ -961,15 +1087,22 @@ def test_scoring_path_skips_thinned_requests(db_engine: Engine): handler = MinerDataHandler(db_engine) handler.density_tapering_predictions(LOW_TEST_CONFIG) + # The keeper is a salted-hash pick, so discover which of the two + # same-bucket requests survived and assert the scorer sees only it. with db_engine.connect() as connection: - keeper_vr_id = connection.execute( + alive, thinned = _split_alive_thinned( + connection, [keeper_mp, thinned_mp] + ) + assert len(alive) == 1 + assert len(thinned) == 1 + survivor_vr_id = connection.execute( select(MinerPrediction.validator_requests_id).where( - MinerPrediction.id == keeper_mp + MinerPrediction.id == alive[0] ) ).scalar_one() thinned_vr_id = connection.execute( select(MinerPrediction.validator_requests_id).where( - MinerPrediction.id == thinned_mp + MinerPrediction.id == thinned[0] ) ).scalar_one() @@ -980,7 +1113,7 @@ def test_scoring_path_skips_thinned_requests(db_engine: Engine): asset_list=["ETH"], ) returned_ids = {int(r.id) for r in (requests or [])} - assert keeper_vr_id in returned_ids + assert survivor_vr_id in returned_ids assert thinned_vr_id not in returned_ids # Defense in depth: even if a future caller bypasses the filter @@ -1211,15 +1344,14 @@ def test_prune_preserves_latest_request_per_asset_during_gap( was being tombstoned as a non-keeper): same asset, same hourly bucket, both older than `thin_after_minutes`: - - "Keeper" : start_time = now - 90 min (smaller id, rn = 1). - - "Latest" : start_time = now - 50 min (larger id, rn = 2 in - bucket; without the new protection it would be - soft-deleted as redundant). - - Both must stay alive after `density_tapering_predictions`: - the keeper because rn = 1 (scoring still uses it), the latest - because the new `latest_per_asset` clause shields the freshest - request per asset from the rn > 1 deletion. + - "Older" : start_time = now - 90 min. + - "Latest" : start_time = now - 50 min (freshest per asset). + + The keeper is a randomized (salted-hash) pick, but the freshest + request per asset is always alive after `density_tapering_predictions`: + as the keeper itself, or — when it is a non-keeper — via the + `latest_per_asset` clause that shields the freshest request per asset + from the rn > 1 deletion. """ now = datetime.now() keeper_start = now - timedelta(minutes=90) @@ -1228,7 +1360,8 @@ def test_prune_preserves_latest_request_per_asset_during_gap( with db_engine.connect() as connection: with connection.begin(): miner = _insert_miner(connection, miner_uid=320) - keeper_id = _insert_prediction( + # Older, non-latest request that contends for the bucket keeper. + _insert_prediction( connection, miner_id=miner, asset="ETH", @@ -1250,14 +1383,8 @@ def test_prune_preserves_latest_request_per_asset_during_gap( MinerDataHandler(db_engine).density_tapering_predictions(LOW_TEST_CONFIG) with db_engine.connect() as connection: - keeper_row = _fetch_prediction_state(connection, keeper_id) latest_row = _fetch_prediction_state(connection, latest_id) - assert keeper_row is not None - assert ( - keeper_row.deleted_at is None - ), "Bucket keeper (rn = 1) must stay alive — scoring still uses it." - assert latest_row is not None assert latest_row.deleted_at is None, ( "Latest request per asset must stay alive even as a " @@ -1281,12 +1408,16 @@ def test_prune_drops_protection_once_latest_ages_past_time_length( crosses its 24h forecast window while still being the newest): same asset, same hourly bucket, BOTH older than 24h (`time_length`): - - "Keeper" : start_time = anchor + 5 min (rn = 1, kept). - - "Stale latest" : start_time = anchor + 40 min (rn = 2; - is the latest_per_asset row, but its + - "Older" : start_time = anchor + 5 min. + - "Stale latest" : start_time = anchor + 40 min (freshest, but its `start_time` is older than `now - time_length`, - so the CTE filters it out → no protection - → soft-deleted by the normal rn > 1 rule). + so latest_per_asset filters it out → no + protection). + + latest_per_asset is empty, so the bucket collapses to a single + (randomized) keeper: exactly one of the two survives and the other is + thinned. If the stale latest were still protected, both would stay + alive. """ now = datetime.now() # Pin to an hour boundary 26h ago so both requests fall in the same @@ -1322,20 +1453,11 @@ def test_prune_drops_protection_once_latest_ages_past_time_length( MinerDataHandler(db_engine).density_tapering_predictions(LOW_TEST_CONFIG) with db_engine.connect() as connection: - keeper_row = _fetch_prediction_state(connection, keeper_id) - stale_row = _fetch_prediction_state(connection, stale_latest_id) - - assert keeper_row is not None - assert keeper_row.deleted_at is None - assert keeper_row.prediction == [[{"price": 1.0}]] - - assert stale_row is not None - assert stale_row.deleted_at is not None, ( - "A latest request older than `time_length` must NOT be " - "protected — otherwise it would stay alive into its scoring " - "window and create a duplicate scorable row in its bucket." + # Protection is off (latest_per_asset empty), so exactly one row — + # the randomized keeper — survives; both surviving would mean the + # stale latest was wrongly protected into its scoring window. + alive, thinned = _split_alive_thinned( + connection, [keeper_id, stale_latest_id] ) - assert stale_row.prediction == { - "deleted": True, - "reason": "thinned", - } + assert len(alive) == 1 + assert len(thinned) == 1