feat(validator): salted-hash keeper for density tapering

.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=

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:

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