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.
411 lines
17 KiB
Python
411 lines
17 KiB
Python
import importlib.util
|
|
import json
|
|
import os
|
|
import stat
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import types
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
|
|
def _load_serverless_config_module(*, without_fcntl=False):
|
|
module_path = Path(__file__).resolve().parents[1] / "utils" / "serverless_config.py"
|
|
module_name = (
|
|
"dist_serverless_config_no_fcntl_test_module"
|
|
if without_fcntl
|
|
else "dist_serverless_config_test_module"
|
|
)
|
|
package_name = module_name + "_pkg"
|
|
package = types.ModuleType(package_name)
|
|
package.__path__ = [str(module_path.parent)]
|
|
sys.modules[package_name] = package
|
|
module_name = package_name + ".serverless_config"
|
|
sys.modules.pop(module_name, None)
|
|
spec = importlib.util.spec_from_file_location(module_name, module_path)
|
|
assert spec is not None and spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[module_name] = module
|
|
if without_fcntl:
|
|
with mock.patch.dict(sys.modules, {"fcntl": None}):
|
|
spec.loader.exec_module(module)
|
|
else:
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
serverless_config = _load_serverless_config_module()
|
|
|
|
|
|
class ValidateServerlessWorkerTests(unittest.TestCase):
|
|
def test_returns_copy_with_runpod_defaults(self):
|
|
worker = {"id": "worker-1", "endpoint_id": "ep_123-abc", "name": "GPU"}
|
|
|
|
result = serverless_config.validate_serverless_worker(worker)
|
|
|
|
self.assertEqual(
|
|
result,
|
|
{
|
|
"id": "worker-1",
|
|
"endpoint_id": "ep_123-abc",
|
|
"name": "GPU",
|
|
"type": "serverless",
|
|
"provider": "runpod",
|
|
"jobs_per_run": 1,
|
|
"timeout_seconds": 1800,
|
|
},
|
|
)
|
|
self.assertIsNot(result, worker)
|
|
self.assertEqual(worker, {"id": "worker-1", "endpoint_id": "ep_123-abc", "name": "GPU"})
|
|
|
|
def test_accepts_inclusive_numeric_bounds(self):
|
|
for jobs, timeout in ((1, 30), (16, 86400)):
|
|
with self.subTest(jobs=jobs, timeout=timeout):
|
|
result = serverless_config.validate_serverless_worker(
|
|
{
|
|
"id": "worker-1",
|
|
"endpoint_id": "ep-123",
|
|
"jobs_per_run": jobs,
|
|
"timeout_seconds": timeout,
|
|
}
|
|
)
|
|
self.assertEqual(result["jobs_per_run"], jobs)
|
|
self.assertEqual(result["timeout_seconds"], timeout)
|
|
|
|
def test_legacy_concurrency_is_accepted_and_dropped(self):
|
|
# Configs saved before Images per run replaced the separate setting.
|
|
for legacy in (1, 2, 16, 17, True, "3"):
|
|
with self.subTest(legacy=legacy):
|
|
result = serverless_config.validate_serverless_worker(
|
|
{"id": "worker-1", "endpoint_id": "ep-123", "jobs_per_run": 2, "concurrency": legacy}
|
|
)
|
|
self.assertNotIn("concurrency", result)
|
|
self.assertEqual(result["jobs_per_run"], 2)
|
|
|
|
def test_endpoint_url_is_rejected_with_a_plain_message(self):
|
|
with self.assertRaisesRegex(ValueError, "not a URL"):
|
|
serverless_config.validate_serverless_worker(
|
|
{"id": "worker-1", "endpoint_id": "https://api.runpod.ai/v2/abc123/run"}
|
|
)
|
|
|
|
def test_rejects_missing_or_empty_id_and_endpoint(self):
|
|
base: dict[str, object] = {"id": "worker-1", "endpoint_id": "ep-123"}
|
|
for field, value in (("id", None), ("id", " "), ("endpoint_id", ""), ("endpoint_id", "https://api.runpod.io/ep-123")):
|
|
with self.subTest(field=field, value=value):
|
|
worker = dict(base)
|
|
worker[field] = value
|
|
with self.assertRaises(ValueError):
|
|
serverless_config.validate_serverless_worker(worker)
|
|
|
|
worker = dict(base)
|
|
del worker["id"]
|
|
with self.assertRaises(ValueError):
|
|
serverless_config.validate_serverless_worker(worker)
|
|
|
|
def test_endpoint_id_must_be_ascii_opaque_identifier(self):
|
|
for endpoint_id in ("ep/123", "ep 123", "ép-123", "ep:123", "https://example.test"):
|
|
with self.subTest(endpoint_id=endpoint_id):
|
|
with self.assertRaises(ValueError):
|
|
serverless_config.validate_serverless_worker(
|
|
{"id": "worker-1", "endpoint_id": endpoint_id}
|
|
)
|
|
|
|
def test_endpoint_id_matches_runpod_client_length_limit(self):
|
|
valid_id = "e" * 200
|
|
result = serverless_config.validate_serverless_worker(
|
|
{"id": "worker-1", "endpoint_id": valid_id}
|
|
)
|
|
self.assertEqual(result["endpoint_id"], valid_id)
|
|
with self.assertRaises(ValueError):
|
|
serverless_config.validate_serverless_worker(
|
|
{"id": "worker-1", "endpoint_id": "e" * 201}
|
|
)
|
|
|
|
def test_enabled_must_be_boolean_when_supplied(self):
|
|
base = {"id": "worker-1", "endpoint_id": "ep-123"}
|
|
for enabled in (True, False):
|
|
with self.subTest(enabled=enabled):
|
|
self.assertIs(
|
|
serverless_config.validate_serverless_worker(
|
|
{**base, "enabled": enabled}
|
|
)["enabled"],
|
|
enabled,
|
|
)
|
|
for enabled in ("false", 0, 1, None):
|
|
with self.subTest(enabled=enabled):
|
|
with self.assertRaises(ValueError):
|
|
serverless_config.validate_serverless_worker(
|
|
{**base, "enabled": enabled}
|
|
)
|
|
|
|
def test_rejects_unknown_provider_and_non_serverless_type(self):
|
|
base = {"id": "worker-1", "endpoint_id": "ep-123"}
|
|
for extra in ({"provider": "other"}, {"type": "local"}):
|
|
with self.subTest(extra=extra):
|
|
with self.assertRaises(ValueError):
|
|
serverless_config.validate_serverless_worker({**base, **extra})
|
|
|
|
def test_numeric_fields_reject_bool_float_and_out_of_range(self):
|
|
base = {"id": "worker-1", "endpoint_id": "ep-123"}
|
|
invalid_values = {
|
|
"jobs_per_run": (True, 1.0, 0, 17),
|
|
"timeout_seconds": (True, 30.0, 29, 86401),
|
|
}
|
|
for field, values in invalid_values.items():
|
|
for value in values:
|
|
with self.subTest(field=field, value=value):
|
|
with self.assertRaises(ValueError):
|
|
serverless_config.validate_serverless_worker({**base, field: value})
|
|
|
|
def test_requires_worker_object(self):
|
|
for worker in (None, [], "worker"):
|
|
with self.subTest(worker=worker):
|
|
with self.assertRaises(ValueError):
|
|
serverless_config.validate_serverless_worker(worker)
|
|
|
|
|
|
class SecretStoreTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tempdir = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.tempdir.name)
|
|
self.path = self.root / "private" / "runpod-secrets.json"
|
|
|
|
def tearDown(self):
|
|
self.tempdir.cleanup()
|
|
|
|
def test_set_get_has_delete_and_persists_as_json_mapping(self):
|
|
store = serverless_config.SecretStore(self.path)
|
|
store.set("worker-1", "secret-value")
|
|
|
|
self.assertEqual(store.get("worker-1"), "secret-value")
|
|
self.assertTrue(store.has("worker-1"))
|
|
self.assertIsNone(store.get("missing"))
|
|
self.assertFalse(store.has("missing"))
|
|
self.assertEqual(json.loads(self.path.read_text(encoding="utf-8")), {"worker-1": "secret-value"})
|
|
self.assertEqual(serverless_config.SecretStore(self.path).get("worker-1"), "secret-value")
|
|
|
|
store.delete("worker-1")
|
|
self.assertIsNone(store.get("worker-1"))
|
|
self.assertFalse(store.has("worker-1"))
|
|
|
|
def test_rejects_api_keys_with_newlines_or_over_4096_characters(self):
|
|
store = serverless_config.SecretStore(self.path)
|
|
for key in ("key\nwith-line-feed", "key\rwith-carriage-return", "k" * 4097):
|
|
with self.subTest(key_length=len(key)):
|
|
with self.assertRaises(serverless_config.SecretStoreError):
|
|
store.set("worker-1", key)
|
|
store.set("worker-1", "k" * 4096)
|
|
self.assertEqual(store.get("worker-1"), "k" * 4096)
|
|
|
|
def test_new_secret_file_and_directory_have_restrictive_modes(self):
|
|
store = serverless_config.SecretStore(self.path)
|
|
store.set("worker-1", "secret-value")
|
|
|
|
self.assertEqual(stat.S_IMODE(self.path.stat().st_mode), 0o600)
|
|
self.assertEqual(stat.S_IMODE(self.path.parent.stat().st_mode), 0o700)
|
|
|
|
def test_missing_store_reads_as_empty_without_creating_files(self):
|
|
store = serverless_config.SecretStore(self.path)
|
|
|
|
self.assertIsNone(store.get("worker-1"))
|
|
self.assertFalse(store.has("worker-1"))
|
|
self.assertFalse(self.path.parent.exists())
|
|
|
|
def test_corrupt_store_fails_closed_without_overwriting_or_leaking_contents(self):
|
|
self.path.parent.mkdir()
|
|
corrupt = b'{"partial": "NEVER-LEAK-THIS-secret'
|
|
self.path.write_bytes(corrupt)
|
|
store = serverless_config.SecretStore(self.path)
|
|
|
|
for operation in (
|
|
lambda: store.get("worker-1"),
|
|
lambda: store.set("worker-1", "replacement-secret"),
|
|
lambda: store.delete("worker-1"),
|
|
):
|
|
with self.subTest(operation=operation):
|
|
with self.assertRaises(Exception) as raised:
|
|
operation()
|
|
self.assertNotIn("NEVER-LEAK-THIS", str(raised.exception))
|
|
self.assertNotIn("replacement-secret", str(raised.exception))
|
|
self.assertEqual(self.path.read_bytes(), corrupt)
|
|
|
|
def test_rejects_non_mapping_or_non_string_store_values_without_leaking_data(self):
|
|
for corrupt in (b'["opaque-stored-value"]', b'{"worker": 23}', b'{"": "opaque-stored-value"}'):
|
|
with self.subTest(corrupt=corrupt):
|
|
self.path.parent.mkdir(exist_ok=True)
|
|
self.path.write_bytes(corrupt)
|
|
with self.assertRaises(Exception) as raised:
|
|
serverless_config.SecretStore(self.path).get("worker")
|
|
self.assertNotIn("opaque-stored-value", str(raised.exception))
|
|
|
|
def test_rejects_symlink_file_and_preserves_link_target(self):
|
|
target = self.root / "target.json"
|
|
target.write_text('{"worker-1": "target-secret"}', encoding="utf-8")
|
|
self.path.parent.mkdir()
|
|
self.path.symlink_to(target)
|
|
store = serverless_config.SecretStore(self.path)
|
|
|
|
with self.assertRaises(Exception):
|
|
store.get("worker-1")
|
|
with self.assertRaises(Exception):
|
|
store.set("worker-1", "replacement-secret")
|
|
self.assertEqual(target.read_text(encoding="utf-8"), '{"worker-1": "target-secret"}')
|
|
|
|
def test_rejects_symlinked_parent_directory(self):
|
|
real_dir = self.root / "real"
|
|
real_dir.mkdir()
|
|
link_dir = self.root / "linked"
|
|
link_dir.symlink_to(real_dir, target_is_directory=True)
|
|
store = serverless_config.SecretStore(link_dir / "secrets.json")
|
|
|
|
with self.assertRaises(Exception):
|
|
store.set("worker-1", "secret-value")
|
|
self.assertFalse((real_dir / "secrets.json").exists())
|
|
|
|
def test_rejects_parent_traversal_in_store_path(self):
|
|
store = serverless_config.SecretStore(self.root / "nested" / ".." / "secrets.json")
|
|
|
|
with self.assertRaises(Exception):
|
|
store.set("worker-1", "secret-value")
|
|
self.assertFalse((self.root / "secrets.json").exists())
|
|
|
|
def test_threaded_updates_from_independent_instances_are_not_lost(self):
|
|
workers = 12
|
|
barrier = threading.Barrier(workers)
|
|
errors = []
|
|
|
|
def write_one(index):
|
|
try:
|
|
barrier.wait(timeout=5)
|
|
serverless_config.SecretStore(self.path).set(f"worker-{index}", f"key-{index}")
|
|
except Exception as exc: # collected to fail in the test thread
|
|
errors.append(exc)
|
|
|
|
threads = [threading.Thread(target=write_one, args=(i,)) for i in range(workers)]
|
|
for thread in threads:
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join(timeout=10)
|
|
|
|
self.assertFalse(any(thread.is_alive() for thread in threads), "writer thread did not finish")
|
|
self.assertEqual(errors, [])
|
|
saved = json.loads(self.path.read_text(encoding="utf-8"))
|
|
self.assertEqual(saved, {f"worker-{i}": f"key-{i}" for i in range(workers)})
|
|
|
|
|
|
class PublicConfigTests(unittest.TestCase):
|
|
def test_redacts_credentials_preserves_other_fields_and_marks_only_serverless(self):
|
|
with tempfile.TemporaryDirectory() as tempdir:
|
|
store = serverless_config.SecretStore(Path(tempdir) / "private" / "keys.json")
|
|
store.set("serverless-1", "stored-api-key")
|
|
config = {
|
|
"workers": [
|
|
{
|
|
"id": "serverless-1",
|
|
"type": "serverless",
|
|
"provider": "runpod",
|
|
"endpoint_id": "ep-123",
|
|
"api_key": "inline-secret",
|
|
"clear_api_key": "another-secret",
|
|
"credential": "credential-secret",
|
|
"nested": {"runpod_credential": "nested-secret", "safe": "kept"},
|
|
"name": "remote GPU",
|
|
},
|
|
{"id": "serverless-2", "type": "serverless", "name": "not configured"},
|
|
{"id": "local-1", "type": "local", "api_key": "local-secret"},
|
|
],
|
|
"settings": {"mode": "preserved", "credentials_note": "remove-me"},
|
|
}
|
|
original = json.loads(json.dumps(config))
|
|
|
|
result = serverless_config.public_config(config, store)
|
|
|
|
first, second, local = result["workers"]
|
|
self.assertEqual(first["api_key_configured"], True)
|
|
self.assertEqual(second["api_key_configured"], False)
|
|
self.assertNotIn("api_key_configured", local)
|
|
self.assertNotIn("api_key", first)
|
|
self.assertNotIn("clear_api_key", first)
|
|
self.assertNotIn("credential", first)
|
|
self.assertNotIn("runpod_credential", first["nested"])
|
|
self.assertEqual(first["nested"]["safe"], "kept")
|
|
self.assertNotIn("api_key", local)
|
|
self.assertNotIn("credentials_note", result["settings"])
|
|
self.assertEqual(first["name"], "remote GPU")
|
|
self.assertEqual(config, original)
|
|
self.assertIsNot(result, config)
|
|
self.assertIsNot(result["workers"], config["workers"])
|
|
|
|
def test_redacts_normalized_secret_field_aliases(self):
|
|
config = {
|
|
"workers": [{
|
|
"id": "serverless-1",
|
|
"type": "serverless",
|
|
"apiKey": "camel-api-key",
|
|
"APIKEY": "upper-api-key",
|
|
"clearApiKey": "camel-clear-key",
|
|
"CLEAR_API_KEY": "upper-clear-key",
|
|
"runpod_CREDENTIAL": "provider-credential",
|
|
"nested": {"credential_note": "nested-credential"},
|
|
"safe": "preserved",
|
|
}]
|
|
}
|
|
|
|
result = serverless_config.public_config(
|
|
config, serverless_config.SecretStore(self.root_path())
|
|
)
|
|
|
|
worker = result["workers"][0]
|
|
self.assertEqual(worker, {
|
|
"id": "serverless-1",
|
|
"type": "serverless",
|
|
"nested": {},
|
|
"safe": "preserved",
|
|
"api_key_configured": False,
|
|
})
|
|
|
|
def test_handles_non_dict_config_without_mutating_it(self):
|
|
config = ["safe", {"api_key": "secret"}]
|
|
result = serverless_config.public_config(config, serverless_config.SecretStore(self.root_path()))
|
|
self.assertEqual(result, ["safe", {}])
|
|
self.assertEqual(config, ["safe", {"api_key": "secret"}])
|
|
|
|
def test_module_import_and_non_serverless_redaction_when_fcntl_unavailable(self):
|
|
module = _load_serverless_config_module(without_fcntl=True)
|
|
store = module.SecretStore(self.root_path())
|
|
config = {
|
|
"workers": [
|
|
{"id": "local-1", "type": "local", "api_key": "inline-secret"}
|
|
]
|
|
}
|
|
|
|
result = module.public_config(config, store)
|
|
|
|
self.assertEqual(result, {"workers": [{"id": "local-1", "type": "local"}]})
|
|
self.assertEqual(config["workers"][0]["api_key"], "inline-secret")
|
|
|
|
def test_secret_operations_fail_clearly_without_fcntl(self):
|
|
module = _load_serverless_config_module(without_fcntl=True)
|
|
store = module.SecretStore(self.root_path())
|
|
unsupported_message = "Serverless credential storage is currently supported on POSIX masters only"
|
|
|
|
for operation in (
|
|
lambda: store.get("worker-1"),
|
|
lambda: store.set("worker-1", "api-key"),
|
|
lambda: store.delete("worker-1"),
|
|
lambda: store.has("worker-1"),
|
|
):
|
|
with self.subTest(operation=operation):
|
|
with self.assertRaisesRegex(module.SecretStoreError, unsupported_message):
|
|
operation()
|
|
|
|
@staticmethod
|
|
def root_path():
|
|
return Path(tempfile.gettempdir()) / "not-created-serverless-secret-store" / "secrets.json"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|