Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .env.example
Original file line number Diff line number Diff line change
Expand Up @@ -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=
29 changes: 24 additions & 5 deletions synth/validator/miner_data_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import typing
import logging
import math
import os


import bittensor as bt
Expand Down Expand Up @@ -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)."
)
Comment on lines +89 to +99

def get_miner_uids(self, connection: Connection):
ranked_miners = select(
Expand Down Expand Up @@ -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.
Comment on lines +807 to +812

The most recent request per asset is additionally preserved (until
it is older than `time_length`, i.e. enters scoring range) so
Expand Down Expand Up @@ -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
),
Expand Down Expand Up @@ -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:
Expand Down
8 changes: 8 additions & 0 deletions tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Comment on lines +36 to +41
Loading
Loading