AdithyaSK HF Staff commited on
Commit
858027d
·
verified ·
1 Parent(s): a55d825

Deploy HF RL Explorer

Browse files
Files changed (2) hide show
  1. DESIGN.md +5 -0
  2. app/space_checks.py +65 -17
DESIGN.md CHANGED
@@ -258,6 +258,11 @@ Spaces. Their sitemaps expand ranges one shard at a time, without downloading ta
258
  or creating millions of URLs in memory. These counts describe published entries,
259
  not individually tested episodes or Google-indexed pages.
260
 
 
 
 
 
 
261
  Task HTML includes public task text or structured input facts. Public row datasets
262
  and OpenEnv Task API records can be read anonymously on demand, with four concurrent
263
  reads and bounded caches. These reads do not build an index, wake a Space, run an
 
258
  or creating millions of URLs in memory. These counts describe published entries,
259
  not individually tested episodes or Google-indexed pages.
260
 
261
+ Space-check inventory refreshes outside request threads and never holds the reader
262
+ lock during bucket I/O. Writes update the cached record immediately; concurrent
263
+ refreshes preserve newer evidence. A cold process temporarily has no verified
264
+ Spaces, and cached evidence still expires according to its original check time.
265
+
266
  Task HTML includes public task text or structured input facts. Public row datasets
267
  and OpenEnv Task API records can be read anonymously on demand, with four concurrent
268
  reads and bounded caches. These reads do not build an index, wake a Space, run an
app/space_checks.py CHANGED
@@ -27,6 +27,8 @@ PASS = "API checked"
27
  UNKNOWN = "Not checked"
28
  _cache = (None, 0.0, {})
29
  _lock = threading.Lock()
 
 
30
  _stop = threading.Event()
31
  _worker = None
32
  log = logging.getLogger("rlx")
@@ -39,7 +41,8 @@ def _path(spec):
39
  def save(rec):
40
  global _cache
41
  path = _path(rec["id"])
42
- with _lock:
 
43
  try:
44
  old = json.loads(path.read_bytes())
45
  if old.get("checked_at", 0) > rec["checked_at"]:
@@ -47,29 +50,74 @@ def save(rec):
47
  except (OSError, ValueError):
48
  pass
49
  catalog.atomic_write(path, json.dumps(rec, separators=(",", ":")).encode())
50
- _cache = (None, 0.0, {})
 
 
 
 
51
 
52
 
53
- def inventory():
54
  """Public evidence only; invalid or oversized files are ignored, never trusted."""
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
55
  global _cache
56
  directory = config.STORAGE_DIR / "openenv-checks"
57
  with _lock:
58
- if _cache[0] == directory and time.time() - _cache[1] < 15:
 
 
59
  return _cache[2]
60
- records = {}
61
- for path in sorted(directory.glob("*.json"))[:20000]:
62
- try:
63
- if path.stat().st_size > 32 * 1024:
64
- continue
65
- r = json.loads(path.read_bytes())
66
- spec = catalog.check_spec(r["id"])
67
- if r.get("schema") == 1 and isinstance(r.get("checked_at"), (float, int)):
68
- records[spec] = r
69
- except (OSError, ValueError, KeyError, TypeError):
70
- continue
71
- _cache = (directory, time.time(), records)
72
  return records
 
 
 
73
 
74
 
75
  def status(rec, now=None):
@@ -310,7 +358,7 @@ def scan():
310
 
311
  def _scan():
312
  from . import snapshot
313
- old = inventory()
314
  with snapshot.use() as (_, conn):
315
  candidates = list(conn.execute("SELECT id, stage FROM envs WHERE kind = 'space' ORDER BY trending DESC, likes DESC"))
316
  due = pending(candidates, old)
 
27
  UNKNOWN = "Not checked"
28
  _cache = (None, 0.0, {})
29
  _lock = threading.Lock()
30
+ _write_lock = threading.Lock()
31
+ _refreshing = {}
32
  _stop = threading.Event()
33
  _worker = None
34
  log = logging.getLogger("rlx")
 
41
  def save(rec):
42
  global _cache
43
  path = _path(rec["id"])
44
+ # Bucket I/O must not hold the catalog's reader lock.
45
+ with _write_lock:
46
  try:
47
  old = json.loads(path.read_bytes())
48
  if old.get("checked_at", 0) > rec["checked_at"]:
 
50
  except (OSError, ValueError):
51
  pass
52
  catalog.atomic_write(path, json.dumps(rec, separators=(",", ":")).encode())
53
+ with _lock:
54
+ directory = path.parent
55
+ records = dict(_cache[2]) if _cache[0] == directory else {}
56
+ records[rec["id"]] = rec
57
+ _cache = (directory, _cache[1] if _cache[0] == directory else 0.0, records)
58
 
59
 
60
+ def _read_inventory(directory):
61
  """Public evidence only; invalid or oversized files are ignored, never trusted."""
62
+ records = {}
63
+ for path in sorted(directory.glob("*.json"))[:20000]:
64
+ try:
65
+ if path.stat().st_size > 32 * 1024:
66
+ continue
67
+ r = json.loads(path.read_bytes())
68
+ spec = catalog.check_spec(r["id"])
69
+ if r.get("schema") == 1 and isinstance(r.get("checked_at"), (float, int)):
70
+ records[spec] = r
71
+ except (OSError, ValueError, KeyError, TypeError):
72
+ continue
73
+ return records
74
+
75
+
76
+ def _refresh_inventory(directory, done):
77
+ global _cache
78
+ try:
79
+ records = _read_inventory(directory)
80
+ with _lock:
81
+ if _cache[0] == directory:
82
+ # A check may finish during the read. Its newer evidence,
83
+ # including a failure superseding a pass, must win.
84
+ for spec, rec in _cache[2].items():
85
+ if rec.get("checked_at", 0) > records.get(spec, {}).get("checked_at", 0):
86
+ records[spec] = rec
87
+ _cache = (directory, time.time(), records)
88
+ except Exception:
89
+ log.exception("Could not read stored OpenEnv checks")
90
+ finally:
91
+ with _lock:
92
+ _refreshing.pop(directory, None)
93
+ done.set()
94
+
95
+
96
+ def inventory(*, wait=False):
97
+ """Serve cached evidence immediately; refresh bucket files off the request path.
98
+
99
+ Cold readers see no verified Spaces until the first read completes. Expiry
100
+ still uses checked_at. Only the background checker waits for storage so it
101
+ does not re-probe everything on boot.
102
+ """
103
  global _cache
104
  directory = config.STORAGE_DIR / "openenv-checks"
105
  with _lock:
106
+ if _cache[0] != directory:
107
+ _cache = (directory, 0.0, {})
108
+ if time.time() - _cache[1] < 60:
109
  return _cache[2]
110
+ done = _refreshing.get(directory)
111
+ if done is None:
112
+ done = _refreshing[directory] = threading.Event()
113
+ threading.Thread(target=_refresh_inventory, args=(directory, done), daemon=True,
114
+ name="read-openenv-checks").start()
115
+ records = _cache[2]
116
+ if not wait:
 
 
 
 
 
117
  return records
118
+ done.wait()
119
+ with _lock:
120
+ return _cache[2]
121
 
122
 
123
  def status(rec, now=None):
 
358
 
359
  def _scan():
360
  from . import snapshot
361
+ old = inventory(wait=True)
362
  with snapshot.use() as (_, conn):
363
  candidates = list(conn.execute("SELECT id, stage FROM envs WHERE kind = 'space' ORDER BY trending DESC, likes DESC"))
364
  due = pending(candidates, old)