Robert Wojciechowski
6b94242110
* feat: add Runpod serverless image collector pools * fix: prevent worker settings clipping and remove preview copy * fix: initialize worker settings before mounting the form * fix: validate complete serverless requests against 10 MB limit * fix: simplify serverless cards and keep status dots circular * fix: match serverless endpoint box to master card * fix: preserve serverless capacity and recovery across job failures Retain durable reservations until provider termination or explicit recovery, protect unresolved pools and credentials, and add scoped reconciliation. Consolidate lifecycle and secret policies and split prompt/media stages with regression coverage. * fix(serverless): disable Clear saved API key while the worker is enabled The checkbox read "Clear saved API key (disable pool first)", but the panel never defines "pool" and saving with it ticked on an enabled worker was rejected by the backend. - Disable the checkbox while the worker is enabled, with a tooltip saying to turn the worker off first. Also disable it when no key is saved. - Use "worker" instead of "pool" in user-facing serverless errors, the serverless guide and the README entry. * feat(serverless): replace Concurrent jobs with a single Images per run setting - Remove the separate Concurrent jobs field; rename Jobs per run to Images per run (1-16) with a tooltip explaining that Runpod's max workers decides how many GPUs run them. - Backend uses a fixed per-worker limit of 16 outstanding jobs, so all images of a run are submitted at once. Unresolved jobs still count against it. The limit is constant, so it never has to change while jobs are unresolved. - Legacy 'concurrency' in saved configs and older browser tabs is accepted and ignored/dropped; it no longer triggers the busy guard. - Revert the enable toggle and show the reason when the backend rejects it. - Trim a pasted api.runpod.ai/v2/<id> URL to the endpoint ID; plainer endpoint-ID error otherwise. - Confirm before a type change deletes a saved Runpod API key. - Update docs and tests.
58 lines
2.4 KiB
Python
58 lines
2.4 KiB
Python
"""Private, atomic job-ID journal. Contains identifiers/status only, never payloads."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import tempfile
|
|
import threading
|
|
|
|
from .serverless_state import TERMINAL
|
|
FIELDS = {"record_id", "execution_id", "worker_id", "participant_id", "endpoint_id", "job_id", "status", "created_at", "updated_at", "error", "prompt_id"}
|
|
|
|
|
|
class JobLedger:
|
|
def __init__(self, path):
|
|
self.path = Path(path)
|
|
self._lock = threading.RLock()
|
|
|
|
def _read(self):
|
|
if self.path.is_symlink():
|
|
raise RuntimeError("Unsafe serverless job journal path")
|
|
if not self.path.exists():
|
|
return {}
|
|
try:
|
|
data = json.loads(self.path.read_text(encoding="utf-8"))
|
|
if not isinstance(data, dict) or any(not isinstance(v, dict) or set(v) - FIELDS for v in data.values()):
|
|
raise ValueError()
|
|
return data
|
|
except (ValueError, OSError):
|
|
raise RuntimeError("Serverless job journal is unreadable; preserve it for recovery before submitting more work") from None
|
|
|
|
def all(self):
|
|
with self._lock:
|
|
return list(self._read().values())
|
|
|
|
def update(self, record_id, **fields):
|
|
if set(fields) - FIELDS:
|
|
raise ValueError("Unexpected serverless journal field")
|
|
with self._lock:
|
|
data = self._read()
|
|
data.setdefault(record_id, {"record_id": record_id}).update(fields)
|
|
# Keep every unresolved job, plus a bounded terminal history.
|
|
completed = sorted((k for k, v in data.items() if v.get("status") in TERMINAL), key=lambda k: data[k].get("updated_at", 0))
|
|
for key in completed[:-200]:
|
|
data.pop(key)
|
|
if self.path.parent.is_symlink():
|
|
raise RuntimeError("Unsafe serverless journal directory")
|
|
self.path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
|
|
fd, temporary = tempfile.mkstemp(prefix=".jobs-", dir=self.path.parent)
|
|
try:
|
|
with os.fdopen(fd, "w", encoding="utf-8") as stream:
|
|
json.dump(data, stream, allow_nan=False)
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
os.replace(temporary, self.path)
|
|
finally:
|
|
if os.path.exists(temporary):
|
|
os.unlink(temporary)
|
|
return dict(data[record_id])
|