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
18 changes: 12 additions & 6 deletions database/dashboard/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -206,8 +206,8 @@ async def loss_curve(index: int) -> dict:
meta = _decode_jsonb(row["meta"], {}) or {}
train, val = _aligned_train_val(meta)
if not train and not val and row["round_id"] is not None and row["miner_hotkey"]:
r2_meta = log_helpers.fetch_meta(
state.r2, int(row["round_id"]), row["miner_hotkey"],
r2_meta = await log_helpers.fetch_meta(
state.pool, state.r2, int(row["round_id"]), row["miner_hotkey"],
)
if r2_meta:
train, val = _aligned_train_val(r2_meta)
Expand Down Expand Up @@ -741,11 +741,13 @@ async def logs_meta_json(round_id: int, hotkey: str):
async def logs_stdout_json(round_id: int, hotkey: str, direct: int = 0):
state = get_state()
if direct:
url = log_helpers.presigned_stdout_url(state.r2, round_id, hotkey)
url = await log_helpers.presigned_stdout_url(
state.pool, state.r2, round_id, hotkey,
)
if not url:
raise HTTPException(status_code=503, detail="R2 presign unavailable")
return RedirectResponse(url=url, status_code=302)
payload = log_helpers.fetch_stdout(state.r2, round_id, hotkey)
payload = await log_helpers.fetch_stdout(state.pool, state.r2, round_id, hotkey)
if payload is None:
raise HTTPException(status_code=404, detail="No stdout.log in R2")
return JSONResponse(payload)
Expand All @@ -755,11 +757,15 @@ async def logs_stdout_json(round_id: int, hotkey: str, direct: int = 0):
async def logs_architecture_json(round_id: int, hotkey: str, direct: int = 0):
state = get_state()
if direct:
url = log_helpers.presigned_architecture_url(state.r2, round_id, hotkey)
url = await log_helpers.presigned_architecture_url(
state.pool, state.r2, round_id, hotkey,
)
if not url:
raise HTTPException(status_code=503, detail="R2 presign unavailable")
return RedirectResponse(url=url, status_code=302)
payload = log_helpers.fetch_architecture(state.r2, round_id, hotkey)
payload = await log_helpers.fetch_architecture(
state.pool, state.r2, round_id, hotkey,
)
if payload is None:
raise HTTPException(status_code=404, detail="No architecture.py in R2")
return JSONResponse(payload)
Expand Down
129 changes: 98 additions & 31 deletions database/dashboard/logs.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,12 @@
Wraps ``shared.r2_audit.R2AuditLog`` with a byte cap so the dashboard
never streams pathologically large stdout blobs over HTTP. Callers that
want the raw file use ``?direct=1`` to get a short-lived presigned URL.

Artifacts live at ``round_{id}/submission_{sid}/...`` — the bucket path
hides miner identities from the trainer-host during Phase B. The
dashboard resolves ``(round_id, miner_hotkey) → submission_id`` via the
``round_submissions`` reveal table that the validator populates after
Phase C closes.
"""

from __future__ import annotations
Expand All @@ -15,26 +21,65 @@
logger = logging.getLogger(__name__)


def _stdout_key(round_id: int, hotkey: str) -> str:
return f"round_{round_id}/miner_{hotkey}/stdout.log"
def _stdout_key_sid(round_id: int, submission_id: str) -> str:
return f"round_{round_id}/submission_{submission_id}/stdout.log"


def _meta_key_sid(round_id: int, submission_id: str) -> str:
return f"round_{round_id}/submission_{submission_id}/training_meta.json"


def _arch_key_sid(round_id: int, submission_id: str) -> str:
return f"round_{round_id}/submission_{submission_id}/architecture.py"

def _meta_key(round_id: int, hotkey: str) -> str:
return f"round_{round_id}/miner_{hotkey}/training_meta.json"

async def submission_id_for(
pool, round_id: int, hotkey: str,
) -> Optional[str]:
"""Look up the opaque submission_id for ``(round_id, miner_hotkey)``.

Returns None when the validator hasn't yet published the reveal map
(Phase C still in progress, or this validator never dispatched a
job for that miner this round).
"""
if pool is None or not hotkey:
return None
try:
row = await pool.fetchrow(
"""
SELECT submission_id FROM round_submissions
WHERE round_id = $1 AND miner_hotkey = $2
ORDER BY created_at DESC LIMIT 1
""",
int(round_id), hotkey,
)
except Exception:
logger.exception(
"submission_id lookup failed: round=%s hotkey=%s",
round_id, hotkey[:16],
)
return None
return row["submission_id"] if row else None

def _arch_key(round_id: int, hotkey: str) -> str:
return f"round_{round_id}/miner_{hotkey}/architecture.py"

async def fetch_meta(pool, r2, round_id: int, hotkey: str) -> Optional[dict]:
"""Return ``training_meta.json`` for a (round, miner) or None if missing.

def fetch_meta(r2, round_id: int, hotkey: str) -> Optional[dict]:
"""Return ``training_meta.json`` for a (round, miner) or None if missing."""
Resolves the bucket path through the ``round_submissions`` reveal
table. Returns None when the reveal hasn't been published yet.
"""
if r2 is None:
return None
sid = await submission_id_for(pool, round_id, hotkey)
if not sid:
return None
try:
return r2.download_json(_meta_key(round_id, hotkey))
return r2.download_json(_meta_key_sid(round_id, sid))
except Exception:
logger.exception("fetch_meta failed: round=%s hotkey=%s", round_id, hotkey[:16])
logger.exception(
"fetch_meta failed: round=%s hotkey=%s sid=%s",
round_id, hotkey[:16], sid[:12],
)
return None


Expand Down Expand Up @@ -70,54 +115,64 @@ async def fetch_meta_cached_or_r2(
meta = await cached_meta(pool, round_id, hotkey)
if meta is not None:
return meta
return fetch_meta(r2, round_id, hotkey)
return await fetch_meta(pool, r2, round_id, hotkey)


def fetch_stdout(
async def fetch_stdout(
pool,
r2,
round_id: int,
hotkey: str,
max_bytes: Optional[int] = None,
) -> Optional[dict]:
"""Return ``{text, truncated, size}`` for the stdout file, or None.

``text`` is capped at ``max_bytes`` (defaults to
``Config.DASHBOARD_MAX_LOG_BYTES``). If the file is larger we take the
tail so operators see the most recent lines.
"""
"""Return ``{text, truncated, size}`` for the stdout file, or None."""
if r2 is None:
return None
sid = await submission_id_for(pool, round_id, hotkey)
if not sid:
return None
cap = max_bytes if max_bytes is not None else Config.DASHBOARD_MAX_LOG_BYTES
try:
raw = r2.download_text(_stdout_key(round_id, hotkey))
raw = r2.download_text(_stdout_key_sid(round_id, sid))
except Exception:
logger.exception("fetch_stdout failed: round=%s hotkey=%s", round_id, hotkey[:16])
logger.exception(
"fetch_stdout failed: round=%s hotkey=%s sid=%s",
round_id, hotkey[:16], sid[:12],
)
return None
if raw is None:
return None
data = raw.encode("utf-8", errors="replace")
size = len(data)
if size > cap:
# Keep the tail — most recent lines are the useful ones
tail = data[-cap:].decode("utf-8", errors="replace")
return {"text": tail, "truncated": True, "size": size}
return {"text": raw, "truncated": False, "size": size}


def presigned_stdout_url(r2, round_id: int, hotkey: str, ttl: int = 900) -> str:
async def presigned_stdout_url(
pool, r2, round_id: int, hotkey: str, ttl: int = 900,
) -> str:
"""Return a 15-minute presigned GET URL for the stdout file."""
if r2 is None:
return ""
sid = await submission_id_for(pool, round_id, hotkey)
if not sid:
return ""
try:
return r2.generate_presigned_get_url(_stdout_key(round_id, hotkey), ttl=ttl)
return r2.generate_presigned_get_url(
_stdout_key_sid(round_id, sid), ttl=ttl,
)
except Exception:
logger.exception(
"presigned_stdout_url failed: round=%s hotkey=%s", round_id, hotkey[:16],
"presigned_stdout_url failed: round=%s hotkey=%s sid=%s",
round_id, hotkey[:16], sid[:12],
)
return ""


def fetch_architecture(
async def fetch_architecture(
pool,
r2,
round_id: int,
hotkey: str,
Expand All @@ -126,12 +181,16 @@ def fetch_architecture(
"""Return ``{text, truncated, size}`` for architecture.py, or None."""
if r2 is None:
return None
sid = await submission_id_for(pool, round_id, hotkey)
if not sid:
return None
cap = max_bytes if max_bytes is not None else Config.DASHBOARD_MAX_LOG_BYTES
try:
raw = r2.download_text(_arch_key(round_id, hotkey))
raw = r2.download_text(_arch_key_sid(round_id, sid))
except Exception:
logger.exception(
"fetch_architecture failed: round=%s hotkey=%s", round_id, hotkey[:16],
"fetch_architecture failed: round=%s hotkey=%s sid=%s",
round_id, hotkey[:16], sid[:12],
)
return None
if raw is None:
Expand All @@ -144,21 +203,29 @@ def fetch_architecture(
return {"text": raw, "truncated": False, "size": size}


def presigned_architecture_url(r2, round_id: int, hotkey: str, ttl: int = 900) -> str:
async def presigned_architecture_url(
pool, r2, round_id: int, hotkey: str, ttl: int = 900,
) -> str:
"""Return a 15-minute presigned GET URL for architecture.py."""
if r2 is None:
return ""
sid = await submission_id_for(pool, round_id, hotkey)
if not sid:
return ""
try:
return r2.generate_presigned_get_url(_arch_key(round_id, hotkey), ttl=ttl)
return r2.generate_presigned_get_url(
_arch_key_sid(round_id, sid), ttl=ttl,
)
except Exception:
logger.exception(
"presigned_architecture_url failed: round=%s hotkey=%s",
round_id, hotkey[:16],
"presigned_architecture_url failed: round=%s hotkey=%s sid=%s",
round_id, hotkey[:16], sid[:12],
)
return ""


__all__ = [
"submission_id_for",
"cached_meta",
"fetch_architecture",
"fetch_meta",
Expand Down
20 changes: 14 additions & 6 deletions database/dashboard/views.py
Original file line number Diff line number Diff line change
Expand Up @@ -321,8 +321,10 @@ async def logs_view(request: Request, round_id: int, hotkey: str):
meta = await log_helpers.fetch_meta_cached_or_r2(
state.pool, state.r2, round_id, hotkey,
)
stdout = log_helpers.fetch_stdout(state.r2, round_id, hotkey)
architecture = log_helpers.fetch_architecture(state.r2, round_id, hotkey)
stdout = await log_helpers.fetch_stdout(state.pool, state.r2, round_id, hotkey)
architecture = await log_helpers.fetch_architecture(
state.pool, state.r2, round_id, hotkey,
)
return _html(
"logs.html",
request,
Expand Down Expand Up @@ -354,11 +356,13 @@ async def logs_meta(round_id: int, hotkey: str):
async def logs_stdout(round_id: int, hotkey: str, direct: int = 0):
state = get_state()
if direct:
url = log_helpers.presigned_stdout_url(state.r2, round_id, hotkey)
url = await log_helpers.presigned_stdout_url(
state.pool, state.r2, round_id, hotkey,
)
if not url:
raise HTTPException(status_code=503, detail="R2 presign unavailable")
return RedirectResponse(url=url, status_code=302)
payload = log_helpers.fetch_stdout(state.r2, round_id, hotkey)
payload = await log_helpers.fetch_stdout(state.pool, state.r2, round_id, hotkey)
if payload is None:
raise HTTPException(status_code=404, detail="No stdout.log in R2")
return JSONResponse(payload)
Expand All @@ -368,11 +372,15 @@ async def logs_stdout(round_id: int, hotkey: str, direct: int = 0):
async def logs_architecture(round_id: int, hotkey: str, direct: int = 0):
state = get_state()
if direct:
url = log_helpers.presigned_architecture_url(state.r2, round_id, hotkey)
url = await log_helpers.presigned_architecture_url(
state.pool, state.r2, round_id, hotkey,
)
if not url:
raise HTTPException(status_code=503, detail="R2 presign unavailable")
return RedirectResponse(url=url, status_code=302)
payload = log_helpers.fetch_architecture(state.r2, round_id, hotkey)
payload = await log_helpers.fetch_architecture(
state.pool, state.r2, round_id, hotkey,
)
if payload is None:
raise HTTPException(status_code=404, detail="No architecture.py in R2")
return JSONResponse(payload)
38 changes: 38 additions & 0 deletions database/migrations/004_round_submissions.sql
Original file line number Diff line number Diff line change
@@ -0,0 +1,38 @@
-- Per-round opaque submission_id -> miner_hotkey reveal map.
--
-- Phase B dispatch now hides the architecture owner's hotkey from the
-- trainer-host (which is itself another miner's pod under cross-eval).
-- The validator mints a per-job ``submission_id`` (32 hex chars) at
-- dispatch time and uses it as the bucket path component:
-- round_{id}/submission_{sid}/checkpoint.safetensors
-- round_{id}/submission_{sid}/architecture.py
-- round_{id}/submission_{sid}/training_meta.json
-- round_{id}/submission_{sid}/stdout.log
-- Trainers never see the architecture owner's hotkey on the wire.
--
-- After Phase C closes, the dispatching validator POSTs the reveal map
-- to ``/round_submissions/reveal``. This table stores it so:
-- * the public dashboard can resolve historical artifact paths back
-- to the (miner_hotkey, miner_uid) pair for log/loss views;
-- * miners can independently confirm their own submission_id was the
-- one trained for any past round.
--
-- Idempotent upsert on (round_id, submission_id): repeated submissions
-- from the same dispatching validator are no-ops, and another
-- dispatcher can append entries for jobs it owned without conflict.
--
-- Bootstrap-order safety: matches the pattern in 003.

CREATE TABLE IF NOT EXISTS round_submissions (
round_id BIGINT NOT NULL,
submission_id TEXT NOT NULL,
miner_hotkey TEXT NOT NULL,
miner_uid INTEGER NOT NULL DEFAULT -1,
created_at DOUBLE PRECISION NOT NULL DEFAULT extract(epoch from now()),
PRIMARY KEY (round_id, submission_id)
);

CREATE INDEX IF NOT EXISTS idx_rs_round_hotkey
ON round_submissions(round_id, miner_hotkey);
CREATE INDEX IF NOT EXISTS idx_rs_hotkey
ON round_submissions(miner_hotkey);
Loading
Loading