CoolBoy/submmit
1
0
Fork 0
Code Issues Pull Requests Actions Projects Releases Wiki Activity

fix: bound unproven profile submissions and support single-worker takeover

Browse Source
This commit is contained in:
CoolBoy
2026-10-26 14:56:55 +08:00
parent 484a10607c
commit e38d5cf823
13 changed files with 376 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
modelhub_submmit_api/state_sync.py
View File

@@ -8,6 +8,9 @@ import os
import shutil
import tempfile
import threading
import time
from contextlib import ExitStack
from itertools import islice
import uuid
import re
from datetime import datetime, timedelta, timezone
@@ -171,6 +174,8 @@ class StateGitSync:
self._pending_push_oid: str | None = None
self._pending_generation: int | None = None
self._pending_manifest: dict[str, Any] | None = None
self._last_success_monotonic = 0.0
self._refresh_workspace_pending = False
@property
def state_dir(self) -> Path:
@@ -236,7 +241,7 @@ class StateGitSync:
parent = Path(tempfile.mkdtemp(prefix="modelhub-agent-state-"))
workspace = parent / "state"
if remote_oid:
porcelain.clone(
cloned = porcelain.clone(
self.remote,
workspace,
branch=self.branch,
@@ -245,10 +250,11 @@ class StateGitSync:
errstream=io.BytesIO(),
**self._auth_kwargs(),
)
cloned.close()
else:
workspace.mkdir(parents=True)
repo = porcelain.init(workspace)
repo.refs.set_symbolic_ref(b"HEAD", f"refs/heads/{self.branch}".encode("utf-8"))
with porcelain.init(workspace) as repo:
repo.refs.set_symbolic_ref(b"HEAD", f"refs/heads/{self.branch}".encode("utf-8"))
self._workspace = workspace
self._expected_remote_oid = remote_oid
@@ -268,7 +274,7 @@ class StateGitSync:
return manifest
def restore(self) -> bool:
with self._mutex:
with self._mutex, ExitStack() as resources:
try:
remote_oid = self._remote_oid()
self._create_workspace(remote_oid)
@@ -277,7 +283,7 @@ class StateGitSync:
manifest = self._validate_manifest(self._workspace)
except Exception as current_error:
manifest = None
repo = Repo(str(self._workspace))
repo = resources.enter_context(Repo(str(self._workspace)))
commits = [entry.commit.id for entry in repo.get_walker(max_entries=self.history_depth)][1:]
for commit in commits:
porcelain.reset(repo, "hard", commit)
@@ -627,7 +633,10 @@ class StateGitSync:
if self._pending_push_oid is None:
return True
assert self._workspace is not None
repo = Repo(str(self._workspace))
with Repo(str(self._workspace)) as repo:
return self._push_pending_repo(repo, phase=phase)
def _push_pending_repo(self, repo: Repo, *, phase: str) -> bool:
remote_oid = self._remote_oid()
if remote_oid == self._pending_push_oid:
pushed = True
@@ -655,11 +664,12 @@ class StateGitSync:
self._pending_manifest = None
self.last_error = None
self.healthy = True
self._last_success_monotonic = time.monotonic()
self.log(f"[state-sync] generation={self.generation} phase={phase} status=ok")
return True
def sync(self, phase: str) -> bool:
with self._mutex:
with self._mutex, ExitStack() as resources:
try:
if self._workspace is None:
raise StateSyncError("state workspace is not initialized")
@@ -667,6 +677,13 @@ class StateGitSync:
# another commit or generation for the same state transition.
if self._pending_push_oid is not None:
return self._push_pending(phase=phase)
# Intent/result WAL writes are never throttled. Idle/full
# cycles need not rewrite multi-megabyte state every 15s.
if (self.healthy and phase in {"cycle", "cycle_no_slots"}
and time.monotonic() - self._last_success_monotonic < 60):
return True
if self._refresh_workspace_pending:
self._refresh_local_workspace()
previous_manifest: dict[str, Any] = {}
try:
previous_manifest = read_json(self._workspace / "manifest.json")
@@ -691,7 +708,7 @@ class StateGitSync:
"checksums": checksums,
}
write_json(self._workspace / "manifest.json", manifest)
repo = Repo(str(self._workspace))
repo = resources.enter_context(Repo(str(self._workspace)))
porcelain.add(repo)
status = porcelain.status(repo)
staged = any(status.staged.get(kind) for kind in ("add", "delete", "modify"))
@@ -704,7 +721,7 @@ class StateGitSync:
author=self._author,
committer=self._author,
)
commit_count = sum(1 for _ in repo.get_walker())
commit_count = sum(1 for _ in islice(repo.get_walker(), self.history_depth + 1))
if commit_count > self.history_depth:
branch_ref = f"refs/heads/{self.branch}".encode("utf-8")
del repo.refs[branch_ref]
@@ -715,6 +732,7 @@ class StateGitSync:
author=self._author,
committer=self._author,
)
self._refresh_workspace_pending = True
local_oid = repo.head().decode("ascii")
self._pending_push_oid = local_oid
self._pending_generation = next_generation
@@ -726,6 +744,32 @@ class StateGitSync:
self.log(f"[state-sync] phase={phase} status=failed reason={self.last_error}")
return False
def _refresh_local_workspace(self) -> None:
"""After a verified orphan push, discard unreachable *local* objects.
Merely resetting the branch bounds history but leaves old blobs on
disk. Rotate only this class's disposable workspace after verification.
"""
old = self._workspace
parent = Path(tempfile.mkdtemp(prefix="modelhub-agent-state-"))
replacement = parent / "state"
try:
with porcelain.clone(self.remote, replacement, branch=self.branch, depth=1,
errstream=io.BytesIO(), **self._auth_kwargs()) as repo:
if repo.head().decode("ascii") != self._expected_remote_oid:
raise StateSyncError("state changed during local workspace rotation")
self._validate_manifest(replacement)
except Exception:
shutil.rmtree(parent)
raise
self._workspace = replacement
self._refresh_workspace_pending = False
if (old is not None and old.name == "state"
and old.parent.name.startswith("modelhub-agent-state-")
and old.parent.parent == Path(tempfile.gettempdir())):
shutil.rmtree(old.parent)
self.log("[state-sync] local_workspace=rotated unreachable_objects=discarded")
def write_readiness(self, *, ready: bool, reason: str | None = None, extra: dict[str, Any] | None = None) -> None:
body = {
"ready": bool(ready),