store.py

# -*- coding: utf-8 -*-
"""The submission ledger, kept in a private Hugging Face dataset.

A Space's container filesystem is ephemeral. Every rebuild, restart or sleep wipes it,
and with it every submission and every score. That is survivable for a demo and not
survivable for a contest with a prize attached, so the ledger lives outside the container
in a dataset repo that the entrant's work outlives the service.

Layout - one file per record, never appended to:

    submissions/<id>.json   written by the web service when an entry is accepted
    results/<id>.json       written by the GPU worker when it finishes scoring
    leaderboard.json        rolled up by the worker so the page reads one file

One file per record is what makes concurrent writers safe. Two entrants submitting at the
same moment touch different paths, so neither commit can clobber the other - which an
append to a shared JSONL absolutely would.

The rollup exists because a page load must not fan out into one request per entry. The
worker already holds every score at the moment it writes one, so it is the natural place
to rebuild the table.
"""
import base64
import json
import os
import time
import urllib.error
import urllib.request

REPO = os.environ.get("ODC_DATASET", "FINAL-Bench/odc-submissions")
TOKEN = os.environ.get("HF_TOKEN", "")
API = "https://huggingface.co/api/datasets/%s" % REPO
RESOLVE = "https://huggingface.co/datasets/%s/resolve/main" % REPO


def _hdr(extra=None):
    h = {"User-Agent": "VIDRAFT-ODC/1.0"}
    if TOKEN:
        h["Authorization"] = "Bearer " + TOKEN
    if extra:
        h.update(extra)
    return h


def _get_json(url, timeout=30):
    req = urllib.request.Request(url, headers=_hdr())
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return json.loads(r.read().decode())


def read(path, default=None):
    """Fetch one record. Missing files are a normal state, not an error."""
    try:
        return _get_json("%s/%s" % (RESOLVE, path))
    except urllib.error.HTTPError as e:
        if e.code in (404, 401, 403):
            return default
        raise
    except Exception:
        return default


def write(path, obj, summary=None):
    """Commit one record. NDJSON is the format this endpoint takes - a plain JSON body
    is accepted and then quietly does nothing, which is a long way to debug."""
    blob = base64.b64encode(json.dumps(obj, ensure_ascii=False).encode()).decode()
    lines = [
        json.dumps({"key": "header", "value": {"summary": summary or ("write " + path)}}),
        json.dumps({"key": "file", "value": {"path": path, "content": blob,
                                             "encoding": "base64"}}),
    ]
    body = ("\n".join(lines) + "\n").encode()
    req = urllib.request.Request(API + "/commit/main", data=body,
                                 headers=_hdr({"Content-Type": "application/x-ndjson"}))
    with urllib.request.urlopen(req, timeout=60) as r:
        return json.loads(r.read().decode())


MAX_PAGES = int(os.environ.get("ODC_MAX_PAGES", "200"))


def _tree_pages(prefix):
    """Every page of a tree listing, following the Link cursor.

    The Hub caps a page at 1,000 entries. Reading only the first page was silent while
    the dataset was small and started dropping records the moment it was not."""
    url = "%s/tree/main/%s" % (API, prefix)
    seen = 0
    for _ in range(MAX_PAGES):
        req = urllib.request.Request(url, headers=_hdr())
        with urllib.request.urlopen(req, timeout=60) as r:
            page = json.loads(r.read().decode())
            link = r.headers.get("Link") or ""
        yield page
        seen += len(page)
        nxt = ""
        for part in link.split(","):
            if 'rel="next"' in part and "<" in part:
                nxt = part[part.index("<") + 1:part.index(">")]
        if not nxt:
            return
        url = nxt
    raise RuntimeError("tree listing exceeded %d pages at %s (%d entries)"
                       % (MAX_PAGES, prefix, seen))


def listdir(prefix):
    """Record ids under a prefix. Absent directory means nothing has been written yet."""
    out = []
    try:
        for page in _tree_pages(prefix):
            for e in page:
                q = e.get("path", "")
                if q.endswith(".json") and not os.path.basename(q).startswith("_"):
                    out.append(os.path.basename(q)[:-5])
    except urllib.error.HTTPError as e:
        if e.code == 404:
            return []
        raise
    except Exception:
        # a partial listing is worse than none: the caller would treat the missing ids as
        # unscored and the rollup would drop them, which is exactly the failure this fixes
        raise
    return out


class Cached:
    """A small TTL cache so a page refresh does not become a round trip to the Hub.

    Staleness is bounded and harmless here: the worst case is a leaderboard a few seconds
    behind, and the page polls anyway.
    """

    def __init__(self, ttl=20):
        self.ttl = ttl
        self._v = {}

    def get(self, key, produce):
        now = time.time()
        hit = self._v.get(key)
        if hit and now - hit[0] < self.ttl:
            return hit[1]
        val = produce()
        self._v[key] = (now, val)
        return val

    def drop(self, key=None):
        if key is None:
            self._v.clear()
        else:
            self._v.pop(key, None)