From 325ed08ed2470634fef6350d5587b93b1ca2e1e5 Mon Sep 17 00:00:00 2001 From: kronecker <2026692687@qq.com> Date: Tue, 13 Oct 2026 05:06:24 +0800 Subject: [PATCH] feat(release): add fail-closed admission and bounded service draining --- backend/src/kf_oj/api/app.py | 2 + backend/src/kf_oj/api/availability.py | 50 ++ backend/src/kf_oj/contracts/availability.py | 18 + backend/src/kf_oj/contracts/export.py | 2 + backend/src/kf_oj/core/settings.py | 2 + backend/src/kf_oj/infra/availability.py | 52 ++ .../src/kf_oj/infra/messaging/publisher.py | 40 ++ backend/src/kf_oj/workers/outbox_status.py | 27 + backend/src/kf_oj/workers/problem_import.py | 12 +- backend/tests/test_availability.py | 88 +++ backend/tests/test_delivery.py | 99 ++++ .../delivery-availability.v1.schema.json | 39 ++ judge/crates/sandbox/src/factory.rs | 3 +- judge/crates/sandbox/src/factory_signals.rs | 55 ++ judge/crates/sandbox/src/lib.rs | 2 + scripts/release/README.md | 27 + scripts/release/availability.py | 66 +++ scripts/release/factory_service.py | 2 +- scripts/release/lifecycle.py | 536 ++++++++++++++++++ scripts/release/node_health.py | 58 +- scripts/release/operator_units.py | 48 +- scripts/release/outbox_health.py | 40 ++ scripts/release/service_bootstrap.py | 5 + .../delivery-availability.valid.json | 6 + 24 files changed, 1261 insertions(+), 18 deletions(-) create mode 100644 backend/src/kf_oj/api/availability.py create mode 100644 backend/src/kf_oj/contracts/availability.py create mode 100644 backend/src/kf_oj/infra/availability.py create mode 100644 backend/src/kf_oj/workers/outbox_status.py create mode 100644 backend/tests/test_availability.py create mode 100644 contracts/schemas/delivery-availability.v1.schema.json create mode 100644 judge/crates/sandbox/src/factory_signals.rs create mode 100644 scripts/release/availability.py create mode 100644 scripts/release/lifecycle.py create mode 100644 scripts/release/outbox_health.py create mode 100644 tests/fixtures/contracts/delivery-availability.valid.json diff --git a/backend/src/kf_oj/api/app.py b/backend/src/kf_oj/api/app.py index 07ef48f..05d2625 100644 --- a/backend/src/kf_oj/api/app.py +++ b/backend/src/kf_oj/api/app.py @@ -6,6 +6,7 @@ from fastapi.exceptions import RequestValidationError from starlette.responses import JSONResponse from kf_oj import __version__ +from kf_oj.api.availability import AvailabilityMiddleware from kf_oj.api.body_limits import BodyLimitMiddleware from kf_oj.api.v1.admin_imports import router as import_router from kf_oj.api.v1.admin_problems import router as maintenance_router @@ -132,6 +133,7 @@ def create_app(settings: Settings | None = None) -> FastAPI: ) app.state.settings = settings app.add_middleware(BodyLimitMiddleware) + app.add_middleware(AvailabilityMiddleware, settings=settings) @app.middleware("http") async def private_responses(request, call_next): diff --git a/backend/src/kf_oj/api/availability.py b/backend/src/kf_oj/api/availability.py new file mode 100644 index 0000000..6a06a5f --- /dev/null +++ b/backend/src/kf_oj/api/availability.py @@ -0,0 +1,50 @@ +"""Admission barrier while already accepted work retains its read/cancel paths.""" + +import re + +from starlette.responses import JSONResponse + +from kf_oj.infra.availability import current + + +def needs_open(method: str, path: str) -> bool: + if method in {"GET", "HEAD", "OPTIONS"}: + return False + if method == "POST" and path == "/api/v1/auth/logout": + return False + if method == "POST" and re.fullmatch(r"/api/v1/submissions/[0-9a-f-]{36}/cancel", path): + return False + if method == "PUT" and re.fullmatch( + r"/api/v1/admin/problem-imports/[0-9a-f-]{36}/archive", path + ): + # The domain route still requires a previously authorized immutable + # receipt. This cannot create a new upload or bypass its lease/epoch. + return False + return path.startswith("/api/v1/") + + +class AvailabilityMiddleware: + def __init__(self, app, *, settings): + self.app, self.settings = app, settings + + async def __call__(self, scope, receive, send): + if ( + scope["type"] == "http" + and self.settings.environment == "production" + and needs_open(scope["method"], scope.get("path", "")) + ): + try: + state = current( + self.settings.delivery_release_id, self.settings.delivery_configuration_id + ) + allowed = state.state == "open" + except (OSError, ValueError, RecursionError): + allowed = False + if not allowed: + await JSONResponse( + {"detail": "platform mutations unavailable"}, + status_code=503, + headers={"Cache-Control": "no-store", "Retry-After": "10"}, + )(scope, receive, send) + return + await self.app(scope, receive, send) diff --git a/backend/src/kf_oj/contracts/availability.py b/backend/src/kf_oj/contracts/availability.py new file mode 100644 index 0000000..220c555 --- /dev/null +++ b/backend/src/kf_oj/contracts/availability.py @@ -0,0 +1,18 @@ +"""Root-to-API availability binding; independent from immutable judging protocols.""" + +from typing import Annotated, Literal + +from pydantic import BaseModel, BeforeValidator, ConfigDict, Field + +from kf_oj.contracts.judging import integer_version + +AVAILABILITY_PATH = "/etc/kf-admission/status.json" + + +class AvailabilityRecord(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True, strict=True) + + schema_version: Annotated[Literal[1], BeforeValidator(integer_version)] + release_id: Annotated[str, Field(pattern=r"^sd-[0-9a-f]{12}$")] + configuration_id: Annotated[str, Field(pattern=r"^cfg-[0-9a-f]{24}$")] + state: Literal["open", "draining", "quiescent"] diff --git a/backend/src/kf_oj/contracts/export.py b/backend/src/kf_oj/contracts/export.py index b4c0531..da08f3b 100644 --- a/backend/src/kf_oj/contracts/export.py +++ b/backend/src/kf_oj/contracts/export.py @@ -10,6 +10,7 @@ from kf_oj.contracts.admin_problems import ( ProblemImportCapabilities, ProblemImportDetail, ) +from kf_oj.contracts.availability import AvailabilityRecord from kf_oj.contracts.judge import JudgeTaskNotification from kf_oj.contracts.judging import WIRE_MODELS from kf_oj.contracts.judging_v2 import WIRE_MODELS as WIRE_MODELS_V2 @@ -55,6 +56,7 @@ def main() -> None: ) outputs[root / "contracts/schemas/submission-performance.v1.schema.json"] = performance for name, model in { + "delivery-availability": AvailabilityRecord, "problem-publish": PublishCandidate, "problem-availability": ChangeProblemAvailability, "problem-maintenance": MaintenanceReceipt, diff --git a/backend/src/kf_oj/core/settings.py b/backend/src/kf_oj/core/settings.py index 495536b..504085d 100644 --- a/backend/src/kf_oj/core/settings.py +++ b/backend/src/kf_oj/core/settings.py @@ -30,6 +30,8 @@ class Settings(BaseSettings): judge_admission: NodeAdmission | None = None coordinator_url: str | None = None coordinator_token: SecretStr | None = None + delivery_release_id: str | None = Field(default=None, pattern=r"^sd-[0-9a-f]{12}$") + delivery_configuration_id: str | None = Field(default=None, pattern=r"^cfg-[0-9a-f]{24}$") @model_validator(mode="after") def secure_origins(self): diff --git a/backend/src/kf_oj/infra/availability.py b/backend/src/kf_oj/infra/availability.py new file mode 100644 index 0000000..3bd909d --- /dev/null +++ b/backend/src/kf_oj/infra/availability.py @@ -0,0 +1,52 @@ +"""Ordinary read-only view of the fixed protected Root admission record.""" + +import json +import os +import stat +from pathlib import Path + +from kf_oj.contracts.availability import AVAILABILITY_PATH, AvailabilityRecord + +GATE = Path(AVAILABILITY_PATH) + + +def protected_record() -> AvailabilityRecord: + for path in GATE.parents: + info = path.lstat() + if not stat.S_ISDIR(info.st_mode) or info.st_uid != 0 or info.st_mode & 0o022: + raise ValueError("protected availability directory required") + fd = os.open(GATE, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC) + try: + info = os.fstat(fd) + if ( + not stat.S_ISREG(info.st_mode) + or info.st_nlink != 1 + or info.st_uid != 0 + or info.st_gid != 0 + or stat.S_IMODE(info.st_mode) != 0o644 + or info.st_size > 4096 + ): + raise ValueError("protected availability file required") + with os.fdopen(fd, "rb", closefd=False) as source: + contents = source.read(4097) + if len(contents) > 4096: + raise ValueError("availability size bound") + finally: + os.close(fd) + + def unique(pairs): + value = {} + for key, item in pairs: + if key in value: + raise ValueError("duplicate availability field") + value[key] = item + return value + + return AvailabilityRecord.model_validate(json.loads(contents, object_pairs_hook=unique)) + + +def current(release: str | None, batch: str | None) -> AvailabilityRecord: + value = protected_record() + if value.release_id != release or value.configuration_id != batch: + raise ValueError("availability is not this running API binding") + return value diff --git a/backend/src/kf_oj/infra/messaging/publisher.py b/backend/src/kf_oj/infra/messaging/publisher.py index 07d8a18..0cac114 100644 --- a/backend/src/kf_oj/infra/messaging/publisher.py +++ b/backend/src/kf_oj/infra/messaging/publisher.py @@ -25,6 +25,46 @@ class OutboxRepository: def __init__(self, sessions): self.sessions = sessions + async def status(self): + async with self.sessions() as session, session.begin(): + await session.execute(text("SET TRANSACTION READ ONLY")) + await session.execute(text("SET LOCAL statement_timeout='1500'")) + await session.execute(text("SET LOCAL lock_timeout='500'")) + safe = await session.scalar( + text(""" + SELECT rolcanlogin AND NOT(rolsuper OR rolcreatedb OR rolcreaterole + OR rolreplication OR rolbypassrls) + AND pg_has_role(current_user,'kf_oj_platform','USAGE') + AND NOT pg_has_role(current_user,'kf_oj_coordinator','MEMBER') + AND NOT pg_has_role(current_user,'kf_oj_identity_authority_owner','MEMBER') + AND NOT has_column_privilege(current_user,'evaluations','state','UPDATE') + FROM pg_roles WHERE rolname=session_user + """) + ) + if safe is not True: + raise ValueError("dedicated outbox projection role required") + row = ( + ( + await session.execute( + text(""" + SELECT + (SELECT count(*) FROM (SELECT 1 FROM outbox + WHERE delivered_at IS NULL LIMIT 129) p) pending, + (SELECT count(*) FROM (SELECT 1 FROM outbox WHERE delivered_at IS NULL + AND locked_until>clock_timestamp() LIMIT 129) l) locked + """) + ) + ) + .mappings() + .one() + ) + return { + "schema_version": 1, + "pending_capped": min(row["pending"], 128), + "locked_capped": min(row["locked"], 128), + "truncated": row["pending"] > 128 or row["locked"] > 128, + } + async def claim(self): owner = uuid4() async with self.sessions.begin() as s: diff --git a/backend/src/kf_oj/workers/outbox_status.py b/backend/src/kf_oj/workers/outbox_status.py new file mode 100644 index 0000000..7dacb71 --- /dev/null +++ b/backend/src/kf_oj/workers/outbox_status.py @@ -0,0 +1,27 @@ +"""Read only publisher-owned counts; never claim, publish or acknowledge a task.""" + +import asyncio +import json + +from kf_oj.core.settings import Settings +from kf_oj.infra.db.session import build_engine, build_session_factory +from kf_oj.infra.messaging.publisher import OutboxRepository + + +async def run() -> dict: + engine = build_engine(Settings()) + try: + return await OutboxRepository(build_session_factory(engine)).status() + finally: + await engine.dispose() + + +def main() -> None: + try: + print(json.dumps(asyncio.run(asyncio.wait_for(run(), timeout=6)), sort_keys=True)) + except Exception: + raise SystemExit("outbox status unavailable") from None + + +if __name__ == "__main__": + main() diff --git a/backend/src/kf_oj/workers/problem_import.py b/backend/src/kf_oj/workers/problem_import.py index 17d8983..f31c7fe 100644 --- a/backend/src/kf_oj/workers/problem_import.py +++ b/backend/src/kf_oj/workers/problem_import.py @@ -1,6 +1,7 @@ """Ordinary import service composition. Provision its dedicated DB role and cgroup separately.""" import asyncio +import signal import sys from sqlalchemy.exc import DBAPIError @@ -16,6 +17,10 @@ from kf_oj.modules.problems.import_worker import ImportWorker, verify_worker_lim async def run(): + stop = asyncio.Event() + loop = asyncio.get_running_loop() + for event in (signal.SIGTERM, signal.SIGINT): + loop.add_signal_handler(event, stop.set) settings = Settings() if ( settings.import_staging_root is None @@ -42,9 +47,12 @@ async def run(): settings.judge_admission, settings.import_worker_cgroup, ) - while True: + while not stop.is_set(): if not await worker.once(): - await asyncio.sleep(1) + try: + await asyncio.wait_for(stop.wait(), timeout=1) + except TimeoutError: + pass finally: staging.close() store.close() diff --git a/backend/tests/test_availability.py b/backend/tests/test_availability.py new file mode 100644 index 0000000..a003f44 --- /dev/null +++ b/backend/tests/test_availability.py @@ -0,0 +1,88 @@ +"""Contract/admission-route tests; Root file/atomicity and drain acceptance are VM-only.""" + +import asyncio +import json +from pathlib import Path +from types import SimpleNamespace + +import pytest +from jsonschema import Draft202012Validator + +from kf_oj.api.availability import AvailabilityMiddleware, needs_open +from kf_oj.contracts.availability import AvailabilityRecord + + +@pytest.mark.parametrize("state", ["open", "draining", "quiescent"]) +def test_availability_is_exactly_bound_to_a_known_version_and_state(state): + path = ( + Path(__file__).resolve().parents[2] + / "tests/fixtures/contracts/delivery-availability.valid.json" + ) + value = json.loads(path.read_text()) + value["state"] = state + assert AvailabilityRecord.model_validate(value).state == state + schema = path.parents[3] / "contracts/schemas/delivery-availability.v1.schema.json" + Draft202012Validator(json.loads(schema.read_text())).validate(value) + for key, replacement in ( + ("schema_version", True), + ("release_id", "../release"), + ("state", "always_allow"), + ): + with pytest.raises(ValueError): + AvailabilityRecord.model_validate({**value, key: replacement}) + + +@pytest.mark.parametrize( + "method,path,blocked", + [ + ("POST", "/api/v1/submissions", True), + ("POST", "/api/v1/admin/problem-imports", True), + ("POST", "/api/v1/admin/problems/publish", True), + ("POST", "/api/v1/auth/login", True), + ("GET", "/api/v1/internal/evaluations/artifacts", False), + ("POST", "/api/v1/auth/logout", False), + ("POST", "/api/v1/submissions/11111111-1111-4111-8111-111111111111/cancel", False), + ( + "PUT", + "/api/v1/admin/problem-imports/11111111-1111-4111-8111-111111111111/archive", + False, + ), + ], +) +def test_drain_blocks_new_mutations_without_cutting_accepted_reads_and_cancel( + method, path, blocked +): + assert needs_open(method, path) is blocked + + +def test_missing_admission_record_rejects_before_body_or_domain_execution(monkeypatch): + # Transport policy only: this stub does not verify Root ownership or Linux file safety. + calls = [] + messages = [] + + def missing(*_): + raise FileNotFoundError("record missing") + + monkeypatch.setattr("kf_oj.api.availability.current", missing) + + async def receive(): + calls.append("body") + return {"type": "http.request", "body": b""} + + async def domain(*_): + calls.append("domain") + + async def send(message): + messages.append(message) + + settings = SimpleNamespace( + environment="production", delivery_release_id=None, delivery_configuration_id=None + ) + asyncio.run( + AvailabilityMiddleware(domain, settings=settings)( + {"type": "http", "method": "POST", "path": "/api/v1/submissions"}, receive, send + ) + ) + assert calls == [] + assert messages[0]["status"] == 503 + assert (b"cache-control", b"no-store") in messages[0]["headers"] diff --git a/backend/tests/test_delivery.py b/backend/tests/test_delivery.py index 79045ea..d843b1d 100644 --- a/backend/tests/test_delivery.py +++ b/backend/tests/test_delivery.py @@ -25,6 +25,7 @@ dependency_install = importlib.import_module("scripts.release.dependency_install dependency_initialize = importlib.import_module("scripts.release.dependency_initialize") node_health = importlib.import_module("scripts.release.node_health") broker_health = importlib.import_module("scripts.release.broker_health") +lifecycle = importlib.import_module("scripts.release.lifecycle") def test_delivery_authorities_and_layout(): @@ -843,3 +844,101 @@ def test_counter_projection_rejects_truncated_empty_and_source_admission_mismatc value["counters"]["cleanup_unconfirmed"]["truncated"] = True with pytest.raises(policy.DeliveryError, match="bound"): node_health.validate_execution(value, value["admission"]) + + +def empty_drain_projection(): + return { + "execution": { + "database_clock_unix_ms": 1791830000000, + **{ + name: {"count_capped": 0, "truncated": False} + for name in ("queued", "in_flight", "cleanup_unconfirmed") + }, + }, + "outbox": { + "schema_version": 1, + "pending_capped": 0, + "locked_capped": 0, + "truncated": False, + }, + "imports": { + "schema_version": 1, + "worker_ready": True, + "reserved_bytes": 0, + "uploaded_capped": 0, + "validating_capped": 0, + "truncated": False, + }, + "broker": { + "schema_version": 1, + "metadata_verified": True, + "messages_ready": 0, + "messages_unacknowledged": 0, + "message_bytes": 0, + "consumers": 1, + }, + "factory": { + "schema_version": 1, + "configuration_sha256": "b" * 64, + "supervisor_authorized": True, + "execution_gate": "required_per_execution", + "isolation_verified": False, + "max_concurrent_runs": 1, + "active_helpers": 0, + "cleanup_complete": True, + "remaining_requests": 4096, + "remaining_lifetime_seconds": 3600, + }, + } + + +@pytest.mark.parametrize( + "component,key,value", + [ + ("outbox", "pending_capped", 1), + ("imports", "uploaded_capped", 1), + ("imports", "validating_capped", 1), + ("imports", "truncated", True), + ("broker", "messages_unacknowledged", 1), + ("factory", "cleanup_complete", False), + ("factory", "active_helpers", 1), + ], +) +def test_drain_cannot_release_with_pending_owner_work_or_unconfirmed_cleanup(component, key, value): + snapshot = empty_drain_projection() + assert lifecycle.projection_empty(snapshot) + snapshot[component][key] = value + if component == "factory" and key == "active_helpers": + snapshot["factory"]["cleanup_complete"] = False + assert not lifecycle.projection_empty(snapshot) + + +@pytest.mark.parametrize("counter", ["queued", "in_flight", "cleanup_unconfirmed"]) +def test_drain_rejects_unknown_or_truncated_execution_counters(counter): + snapshot = empty_drain_projection() + snapshot["execution"][counter]["count_capped"] = False + with pytest.raises(policy.DeliveryError): + lifecycle.projection_empty(snapshot) + snapshot["execution"][counter] = {"count_capped": 0, "truncated": True} + assert not lifecycle.projection_empty(snapshot) + + +def test_maintenance_receipt_cannot_silently_adopt_unknown_history_or_release_proof(): + value = { + "schema_version": 1, + "source_sha": "b" * 40, + "configuration_id": "cfg-" + "c" * 24, + "action": "drain", + "phase": "quiescent", + "success": False, + "events": [], + "root_cleanup_confirmed": True, + "boot_id": "11111111-1111-4111-8111-111111111111", + "drain_projection": empty_drain_projection(), + } + lifecycle.validate_record(value) + with pytest.raises(policy.DeliveryError): + lifecycle.validate_record({**value, "clear_lease": True}) + value["drain_projection"]["factory"]["cleanup_complete"] = False + with pytest.raises(policy.DeliveryError): + lifecycle.validate_record(value) diff --git a/contracts/schemas/delivery-availability.v1.schema.json b/contracts/schemas/delivery-availability.v1.schema.json new file mode 100644 index 0000000..9820889 --- /dev/null +++ b/contracts/schemas/delivery-availability.v1.schema.json @@ -0,0 +1,39 @@ +{ + "$id": "urn:kf-oj:delivery-availability:v1", + "$schema": "https://json-schema.org/draft/2020-12/schema", + "additionalProperties": false, + "properties": { + "configuration_id": { + "pattern": "^cfg-[0-9a-f]{24}$", + "title": "Configuration Id", + "type": "string" + }, + "release_id": { + "pattern": "^sd-[0-9a-f]{12}$", + "title": "Release Id", + "type": "string" + }, + "schema_version": { + "const": 1, + "title": "Schema Version", + "type": "integer" + }, + "state": { + "enum": [ + "open", + "draining", + "quiescent" + ], + "title": "State", + "type": "string" + } + }, + "required": [ + "schema_version", + "release_id", + "configuration_id", + "state" + ], + "title": "AvailabilityRecord", + "type": "object" +} diff --git a/judge/crates/sandbox/src/factory.rs b/judge/crates/sandbox/src/factory.rs index c5bf9f2..3344041 100644 --- a/judge/crates/sandbox/src/factory.rs +++ b/judge/crates/sandbox/src/factory.rs @@ -387,6 +387,7 @@ pub fn serve_with_limit( )?) ); execution::authorize_registry(&config)?; + let stop_signals = crate::factory_signals::StopSignals::install()?; let policy = config.config(); if policy.aggregate.supervisor_processes < 4 + 8 * policy.max_concurrent_runs { return Err(denied()); @@ -472,7 +473,7 @@ pub fn serve_with_limit( let mut health_checks = 0_usize; let mut last_health = None; let result = (|| { - while Instant::now() < deadline { + while Instant::now() < deadline && !stop_signals.requested() { let mut i = 0; while i < active.len() { if let Some(status) = active[i].child.try_wait()? { diff --git a/judge/crates/sandbox/src/factory_signals.rs b/judge/crates/sandbox/src/factory_signals.rs new file mode 100644 index 0000000..82f9c17 --- /dev/null +++ b/judge/crates/sandbox/src/factory_signals.rs @@ -0,0 +1,55 @@ +//! Single-threaded factory termination notice; handlers perform one lock-free store. +use std::io; +use std::sync::atomic::{AtomicBool, Ordering}; + +static STOP_REQUESTED: AtomicBool = AtomicBool::new(false); +extern "C" fn notice(_: libc::c_int) { + STOP_REQUESTED.store(true, Ordering::Relaxed); +} +pub(crate) struct StopSignals { + previous: Vec<(libc::c_int, libc::sigaction)>, +} +impl StopSignals { + #[allow(unsafe_code)] // Fixed signal installation belongs to the Sandbox boundary. + pub(crate) fn install() -> io::Result { + // Factory authorization has already proved a single-threaded process. + // Caught handlers reset across exec, so fresh helper executables retain + // their own signal/guardian semantics; no signal mask is inherited. + STOP_REQUESTED.store(false, Ordering::Relaxed); + let mut guard = Self { + previous: Vec::new(), + }; + for signal in [libc::SIGTERM, libc::SIGINT] { + // SAFETY: initialized action storage, an async-signal-safe handler, + // valid signal numbers, and process-global use in the fixed factory. + unsafe { + let mut action: libc::sigaction = std::mem::zeroed(); + action.sa_sigaction = notice as *const () as usize; + action.sa_flags = libc::SA_RESTART; + if libc::sigemptyset(&mut action.sa_mask) != 0 { + return Err(io::Error::last_os_error()); + } + let mut previous: libc::sigaction = std::mem::zeroed(); + if libc::sigaction(signal, &action, &mut previous) != 0 { + return Err(io::Error::last_os_error()); + } + guard.previous.push((signal, previous)); + } + } + Ok(guard) + } + pub(crate) fn requested(&self) -> bool { + STOP_REQUESTED.load(Ordering::Relaxed) + } +} +impl Drop for StopSignals { + #[allow(unsafe_code)] // Restore only this invocation's saved signal dispositions. + fn drop(&mut self) { + for (signal, previous) in self.previous.iter().rev() { + // SAFETY: restore only dispositions saved by this fixed invocation. + unsafe { + libc::sigaction(*signal, previous, std::ptr::null_mut()); + } + } + } +} diff --git a/judge/crates/sandbox/src/lib.rs b/judge/crates/sandbox/src/lib.rs index 88a237e..ec3db3f 100644 --- a/judge/crates/sandbox/src/lib.rs +++ b/judge/crates/sandbox/src/lib.rs @@ -45,3 +45,5 @@ pub mod factory_cycles; pub mod factory_health; #[cfg(target_os = "linux")] pub mod factory_preparation; +#[cfg(target_os = "linux")] +mod factory_signals; diff --git a/scripts/release/README.md b/scripts/release/README.md index ebf7257..575230a 100644 --- a/scripts/release/README.md +++ b/scripts/release/README.md @@ -277,3 +277,30 @@ creates no evaluation/grade and starts no program; ordinary periodic observation does not publish messages. Probe failure is reported separately. Underlying interfaces: [RabbitMQ CLI](https://www.rabbitmq.com/docs/man/rabbitmqctl.8), [PostgreSQL privilege functions](https://www.postgresql.org/docs/18/functions-info.html). + +`python3.13 -m scripts.release.lifecycle --action start|resume|drain|stop|restart` +is the explicit fixed-unit maintenance entry. Root publishes a protected, versioned +availability record bound to the selected release/configuration. The installed API +checks that record before accepting mutations; an absent, unknown, closed or wrong +binding returns 503. Existing reads, logout, cancellation and receipt-bound archive +uploads retain their ordinary identity/lease checks. Health alone cannot open this +barrier, and service startup does not create schemas, accounts or credentials. + +Drain first closes new mutations, keeps the API available to already accepted +work, and polls the Coordinator, outbox, import, broker and factory owners within +a finite deadline. Only their untruncated empty projections permit ordinary +service shutdown. Final owner counts, actual stopped process groups, delegated +tasks/identity/capture resources and absent factory socket must all agree before +quiescence is published. Timeout preserves unfinished rows, leases, queues and +replay journals; no forced grade or cleanup proof is written. `drain` leaves DB/MQ +running, `stop` additionally stops them, and `restart` drains before a fresh start. + +Fresh start requires stopped known application units. `resume` explicitly checks +known units that are already running after an interrupted operation; it does not +adopt foreign units or repair unknown resources. Both keep admission closed until +full service preconditions pass and opening is read back. Failed final readiness +closes the barrier again. A bounded Root maintenance receipt records manager +operations and cleanup results. Root factory termination uses a signal notice and +the existing bounded guardian/recovery path; ordinary import termination finishes +its current authorized operation before claiming another. These interfaces require +actual independent-VM lifecycle acceptance and are not installation/safety proof. diff --git a/scripts/release/availability.py b/scripts/release/availability.py new file mode 100644 index 0000000..0390ce5 --- /dev/null +++ b/scripts/release/availability.py @@ -0,0 +1,66 @@ +"""Only lifecycle checks may open this fixed Root-owned API admission binding.""" + +import os +import secrets +import stat +from pathlib import Path + +from kf_oj.contracts.availability import AVAILABILITY_PATH, AvailabilityRecord + +from . import install, preflight +from .policy import DeliveryError, canonical, read_json + +GATE = Path(AVAILABILITY_PATH) + + +def read() -> AvailabilityRecord | None: + try: + GATE.lstat() + except FileNotFoundError: + return None + preflight.protected(GATE) + info = GATE.stat() + if info.st_gid != 0 or stat.S_IMODE(info.st_mode) != 0o644: + raise DeliveryError("unknown availability record; refused adoption") + return AvailabilityRecord.model_validate(read_json(GATE, 4096)) + + +def write(selected: dict, state: str) -> dict: + if os.getuid() != 0 or os.geteuid() != 0 or os.getgid() != 0: + raise DeliveryError("root availability operator required") + value = AvailabilityRecord( + schema_version=1, + release_id=selected["release_id"], + configuration_id=selected["configuration_id"], + state=state, + ) + if install.selection(check_configuration=False) != selected: + raise DeliveryError("availability selection changed") + install.directory(GATE.parent, 0o711) + before = read() + if before is not None and ( + before.release_id != selected["release_id"] + or before.configuration_id != selected["configuration_id"] + ): + # A new stopped installation can replace an old quiescent binding only. + install.require_stopped() + if before.state != "quiescent" or state != "draining": + raise DeliveryError("availability binding replacement requires quiescence") + temporary = GATE.parent / (".availability-" + secrets.token_hex(8)) + created = None + try: + install.exclusive_file(temporary, canonical(value.model_dump(mode="json")), 0o644) + info = temporary.stat() + created = (info.st_dev, info.st_ino) + if read() != before or install.selection(check_configuration=False) != selected: + raise DeliveryError("availability changed during publication") + os.replace(temporary, GATE) + install.sync_directory(GATE.parent) + finally: + if created is not None and temporary.exists(): + info = temporary.lstat() + if (info.st_dev, info.st_ino) == created: + temporary.unlink() + if read() != value: + raise DeliveryError("availability publication readback differs") + return value.model_dump(mode="json") diff --git a/scripts/release/factory_service.py b/scripts/release/factory_service.py index 1985996..fd208b6 100644 --- a/scripts/release/factory_service.py +++ b/scripts/release/factory_service.py @@ -145,7 +145,7 @@ def render(receipt: dict) -> bytes: "LimitSTACK=16M", "TimeoutStartSec=1800", "RuntimeMaxSec=3700", - "TimeoutStopSec=30", + "TimeoutStopSec=40", "KillMode=mixed", "Restart=no", "StandardOutput=journal", diff --git a/scripts/release/lifecycle.py b/scripts/release/lifecycle.py new file mode 100644 index 0000000..7c72f33 --- /dev/null +++ b/scripts/release/lifecycle.py @@ -0,0 +1,536 @@ +"""Fixed Root maintenance orchestration; owners drain work, never forged terminal state.""" + +import argparse +import fcntl +import json +import os +import re +import secrets +import stat +import subprocess +import time +from pathlib import Path +from uuid import UUID + +from . import ( + availability, + configuration, + dependency_install, + factory_service, + install, + node_health, + operator_units, + preflight, + service_install, + services, +) +from .policy import DeliveryError, canonical, digest, fields, read_json, sha256 + +STATE = Path("/var/lib/kf-onlinejudge/operations") +LAST = STATE / "maintenance.last.json" +APPS = ("api", "outbox", "worker", "problem-import", "coordinator", "sandbox-factory") +DEPENDENCIES = ("postgres", "rabbitmq") +START_ORDER = ( + "sandbox-factory", + "coordinator", + "api", + "outbox", + "problem-import", + "worker", +) +DRAIN_SECONDS = 360 + + +def validate_record(value: dict) -> dict: + required = { + "schema_version", + "source_sha", + "configuration_id", + "action", + "phase", + "success", + "events", + "root_cleanup_confirmed", + "boot_id", + } + if ( + not isinstance(value, dict) + or not required <= set(value) + or not set(value) - required <= {"drain_projection", "failure_code"} + ): + raise DeliveryError("unknown maintenance receipt fields") + if ( + type(value["schema_version"]) is not int + or value["schema_version"] != 1 + or not isinstance(value["source_sha"], str) + or not re.fullmatch(r"[0-9a-f]{40}", value["source_sha"]) + or not isinstance(value["configuration_id"], str) + or not re.fullmatch(r"cfg-[0-9a-f]{24}", value["configuration_id"]) + or not isinstance(value["action"], str) + or value["action"] not in {"start", "resume", "drain", "stop", "restart"} + or not isinstance(value["phase"], str) + or value["phase"] + not in { + "prepared", + "dispatching_start", + "dispatching_stop", + "start_confirmed", + "stop_confirmed", + "admission_closed", + "root_stop_proven", + "quiescent", + "accepting", + "complete", + "failed", + } + or type(value["success"]) is not bool + or type(value["root_cleanup_confirmed"]) is not bool + or not isinstance(value["events"], list) + or len(value["events"]) > 32 + or "failure_code" in value + and value["failure_code"] != "maintenance_unconfirmed" + ): + raise DeliveryError("unknown maintenance receipt binding") + if not isinstance(value["boot_id"], str): + raise DeliveryError("maintenance boot identity differs") + identifier = UUID(value["boot_id"]) + if identifier.int == 0 or str(identifier) != value["boot_id"]: + raise DeliveryError("maintenance boot identity differs") + for event in value["events"]: + fields(event, {"unit", "operation", "confirmed"}, "maintenance event") + if ( + not isinstance(event["unit"], str) + or event["unit"] not in {*APPS, *DEPENDENCIES} + or not isinstance(event["operation"], str) + or event["operation"] not in {"start", "stop"} + or type(event["confirmed"]) is not bool + ): + raise DeliveryError("unknown maintenance event authority") + if "drain_projection" in value and not projection_empty(value["drain_projection"]): + raise DeliveryError("maintenance quiescence proof differs") + return value + + +def projection_empty(snapshot: dict) -> bool: + fields( + snapshot, + {"execution", "outbox", "imports", "broker", "factory"}, + "drain projection", + ) + execution = fields( + snapshot["execution"], + {"queued", "in_flight", "cleanup_unconfirmed", "database_clock_unix_ms"}, + "execution drain projection", + ) + if ( + type(execution["database_clock_unix_ms"]) is not int + or execution["database_clock_unix_ms"] <= 0 + ): + raise DeliveryError("execution drain clock unavailable") + for name in ("queued", "in_flight", "cleanup_unconfirmed"): + item = fields(execution[name], {"count_capped", "truncated"}, "execution drain counter") + if ( + type(item["count_capped"]) is not int + or not 0 <= item["count_capped"] <= 128 + or type(item["truncated"]) is not bool + ): + raise DeliveryError("execution drain counter unavailable") + if execution[name] != {"count_capped": 0, "truncated": False}: + return False + node_health.validate_outbox(snapshot["outbox"]) + outbox = snapshot["outbox"] + imports = snapshot["imports"] + fields( + imports, + { + "schema_version", + "worker_ready", + "reserved_bytes", + "uploaded_capped", + "validating_capped", + "truncated", + }, + "import drain projection", + ) + if ( + type(imports["schema_version"]) is not int + or imports["schema_version"] != 1 + or type(imports["worker_ready"]) is not bool + or type(imports["truncated"]) is not bool + or any( + type(imports[k]) is not int or not 0 <= imports[k] <= maximum + for k, maximum in ( + ("reserved_bytes", 268435456), + ("uploaded_capped", 128), + ("validating_capped", 128), + ) + ) + ): + raise DeliveryError("import drain counters unavailable") + broker, root = snapshot["broker"], snapshot["factory"] + if not isinstance(root, dict): + raise DeliveryError("factory drain proof unavailable") + node_health.validate_factory(root, sha256(root.get("configuration_sha256"))) + fields( + broker, + { + "schema_version", + "metadata_verified", + "messages_ready", + "messages_unacknowledged", + "message_bytes", + "consumers", + }, + "broker drain projection", + ) + if ( + type(broker["schema_version"]) is not int + or broker["schema_version"] != 1 + or broker["metadata_verified"] is not True + or any( + type(broker[k]) is not int or not 0 <= broker[k] <= maximum + for k, maximum in ( + ("messages_ready", 64), + ("messages_unacknowledged", 1), + ("message_bytes", 32768 + 4096), + ("consumers", 1), + ) + ) + ): + raise DeliveryError("broker drain projection unavailable") + if ( + type(root["active_helpers"]) is not int + or not 0 <= root["active_helpers"] <= 1 + or type(root["cleanup_complete"]) is not bool + ): + raise DeliveryError("factory drain proof unavailable") + return bool( + not outbox["truncated"] + and outbox["pending_capped"] == outbox["locked_capped"] == 0 + and imports["truncated"] is False + and imports["uploaded_capped"] == imports["validating_capped"] == 0 + and broker["messages_ready"] == broker["messages_unacknowledged"] == 0 + and root["active_helpers"] == 0 + and root["cleanup_complete"] is True + ) + + +class Maintenance: + def __init__(self, selected: dict, paths: dict, payload: Path, action: str): + self.selected, self.paths, self.payload = selected, paths, payload + self.coordinator = read_json(paths["coordinator.json"], 32768) + self.record = { + "schema_version": 1, + "source_sha": read_json(payload.parent / "manifest.json")["source_sha"], + "configuration_id": selected["configuration_id"], + "action": action, + "phase": "prepared", + "success": False, + "events": [], + "root_cleanup_confirmed": False, + "boot_id": Path("/proc/sys/kernel/random/boot_id").read_text().strip(), + } + + def save(self, phase: str) -> None: + self.record["phase"] = phase + if len(self.record["events"]) > 32: + raise DeliveryError("maintenance event bound") + contents = canonical(validate_record(self.record)) + if len(contents) > 32768: + raise DeliveryError("maintenance receipt bound") + if LAST.exists() or LAST.is_symlink(): + preflight.protected(LAST) + configuration.private(LAST) + validate_record(read_json(LAST, 32768)) + temporary = STATE / (".maintenance-" + secrets.token_hex(8)) + created = None + try: + install.exclusive_file(temporary, contents, 0o600) + info = temporary.stat() + created = (info.st_dev, info.st_ino) + os.replace(temporary, LAST) + install.sync_directory(STATE) + finally: + if created is not None and temporary.exists(): + info = temporary.lstat() + if (info.st_dev, info.st_ino) == created: + temporary.unlink() + + def unit(self, name: str) -> dict: + if name not in {*APPS, *DEPENDENCIES}: + raise DeliveryError("unknown lifecycle target") + path = service_install.SYSTEM_UNITS / ("kf-" + name + ".service") + if name in services.PROFILES: + directory = service_install.known_link(path) + base = service_install.UNITS + elif name == "sandbox-factory": + directory = factory_service.known_link(path) + base = factory_service.UNITS + else: + directory = dependency_install.known_link(path) + base = dependency_install.ROOT + if directory != base / ( + self.selected["release_id"] + "." + self.selected["configuration_id"] + ): + raise DeliveryError("lifecycle target is not the selected snapshot") + values = dict( + line.split("=", 1) + for line in service_install.systemctl( + "show", + path.name, + "--property=LoadState,ActiveState,SubState,MainPID,FragmentPath", + ).splitlines() + if "=" in line + ) + if values.get("LoadState") != "loaded" or values.get("FragmentPath") not in { + str(path), + str(directory / path.name), + }: + raise DeliveryError("lifecycle manager target differs") + return values + + @staticmethod + def empty_cgroup(name: str) -> None: + directory = Path("/sys/fs/cgroup/kfonlinejudge.slice/kf-" + name + ".service") + try: + directory.lstat() + except FileNotFoundError: + return + if directory.is_symlink() or not directory.is_dir(): + raise DeliveryError("unknown residual service cgroup") + pending = [directory] + seen = 0 + while pending: + group = pending.pop() + seen += 1 + if seen > 64: + raise DeliveryError("service cgroup tree exceeds bound") + processes = group / "cgroup.procs" + with processes.open() as source: + members = source.read(4097) + if len(members) > 4096 or members.strip(): + raise DeliveryError("service processes remain; no capacity release") + for child in group.iterdir(): + if child.is_symlink(): + raise DeliveryError("unknown cgroup entry") + if child.is_dir(): + pending.append(child) + + def change(self, name: str, operation: str) -> None: + if name == "sandbox-factory" and operation == "start": + self.record["root_cleanup_confirmed"] = False + values = self.unit(name) + if operation == "start" and values["ActiveState"] == "active": + node_health.inspect_unit(name, self.selected) + return + if operation == "stop" and values["ActiveState"] == "inactive" and values["MainPID"] == "0": + self.empty_cgroup(name) + return + if operation not in {"start", "stop"}: + raise DeliveryError("unexpected lifecycle operation") + self.record["events"].append({"unit": name, "operation": operation, "confirmed": False}) + self.save("dispatching_" + operation) + service_install.systemctl("--no-block", operation, "kf-" + name + ".service") + deadline = time.monotonic() + (130 if operation == "start" else 50) + while True: + if install.selection(check_configuration=False) != self.selected: + raise DeliveryError("selected binding changed during maintenance") + values = self.unit(name) + if ( + operation == "start" + and values["ActiveState"] == "active" + and values["SubState"] == "running" + ): + node_health.inspect_unit(name, self.selected) + break + if ( + operation == "stop" + and values["ActiveState"] == "inactive" + and values["MainPID"] == "0" + ): + self.empty_cgroup(name) + break + if time.monotonic() >= deadline or values["ActiveState"] == "failed": + raise DeliveryError("bounded lifecycle operation unconfirmed") + time.sleep(0.5) + self.record["events"][-1]["confirmed"] = True + self.save(operation + "_confirmed") + + def snapshot(self) -> dict: + owner = node_health.http( + "http://127.0.0.1:8100/internal/operations/status", + token=self.coordinator["platform_token"], + ) + counters = node_health.validate_execution(owner, self.coordinator["admission"]) + result = { + "execution": counters, + "outbox": operator_units.run("outbox-health", self.payload), + "imports": operator_units.run("import-health", self.payload), + "broker": operator_units.run("broker-health", self.payload), + "factory": operator_units.run("factory-health", self.payload), + } + node_health.validate_factory(result["factory"], digest(self.paths["runtime.json"])) + return result + + def root_stopped(self) -> None: + values = self.unit("sandbox-factory") + if values["ActiveState"] != "inactive" or values["MainPID"] != "0": + raise DeliveryError("Root factory stop unconfirmed") + self.empty_cgroup("sandbox-factory") + for name in ("tasks", "identities", "captures"): + path = Path("/var/lib/kf-judge") / name + try: + path.lstat() + except FileNotFoundError: + continue + preflight.protected(path, directory=True) + if next(path.iterdir(), None) is not None: + raise DeliveryError("task/identity/capture resources remain") + delegated = Path("/sys/fs/cgroup/kfonlinejudge.slice/kf-judge-tasks") + if delegated.exists(): + preflight.protected(delegated, directory=True) + if (delegated / "cgroup.procs").read_text().strip() or any( + p.is_dir() or p.is_symlink() for p in delegated.iterdir() + ): + raise DeliveryError("delegated task resources remain") + socket = Path("/var/lib/kf-judge/factory/control.sock") + try: + socket.lstat() + except FileNotFoundError: + pass + else: + raise DeliveryError("factory socket remains; explicit recovery required") + self.record["root_cleanup_confirmed"] = True + self.save("root_stop_proven") + + def drain(self) -> None: + availability.write(self.selected, "draining") + self.save("admission_closed") + # API remains available for accepted artifact/source reads, cancellation + # and receipt-bound uploads. No new submission/import/write is admitted. + deadline = time.monotonic() + DRAIN_SECONDS + while True: + snapshot = self.snapshot() + if projection_empty(snapshot): + self.record["drain_projection"] = snapshot + break + if time.monotonic() >= deadline: + raise DeliveryError("drain deadline expired; owners retain unfinished work") + time.sleep(2) + # Every process/child and final Root resource proof remains required. + # No row, notification, identity lease or replay journal is removed here. + for name in ("api", "outbox", "problem-import", "worker"): + self.change(name, "stop") + final = self.snapshot() + if not projection_empty(final): + raise DeliveryError("work changed during final stop; quiescence unconfirmed") + self.change("coordinator", "stop") + self.change("sandbox-factory", "stop") + self.root_stopped() + availability.write(self.selected, "quiescent") + self.save("quiescent") + + def start(self, *, resume: bool) -> None: + if not resume: + for name in APPS: + value = self.unit(name) + if value["ActiveState"] != "inactive" or value["MainPID"] != "0": + raise DeliveryError("fresh start requires stopped application units") + self.empty_cgroup(name) + self.root_stopped() + availability.write(self.selected, "draining") + for name in DEPENDENCIES: + self.change(name, "start") + for name in START_ORDER: + self.change(name, "start") + ready = operator_units.run("node-health", self.payload) + if ready.get("service_preconditions_met") is not True: + raise DeliveryError("service preconditions unavailable; admission remains closed") + availability.write(self.selected, "open") + try: + ready = operator_units.run("node-health", self.payload) + if ready.get("ready_for_admission") is not True: + raise DeliveryError("admission readiness readback unconfirmed") + except (OSError, ValueError, subprocess.SubprocessError): + availability.write(self.selected, "draining") + raise + self.save("accepting") + + +def run(action: str) -> dict: + operator_units.guarded("node-maintenance") + selected, paths, payload = node_health.binding() + for name in ("lifecycle.py", "availability.py"): + shipped = payload / "delivery/scripts/release" / name + preflight.protected(shipped) + if digest(shipped) != digest(Path(__file__).parent / name): + raise DeliveryError("maintenance tool differs from selected delivery") + preflight.protected(STATE, directory=True) + if stat.S_IMODE(STATE.stat().st_mode) != 0o700: + raise DeliveryError("private operator state required") + lockpath = STATE / "maintenance.lock" + if not lockpath.exists(): + install.exclusive_file(lockpath, b"", 0o600) + preflight.protected(lockpath) + configuration.private(lockpath) + fd = os.open(lockpath, os.O_RDWR | os.O_NOFOLLOW | os.O_CLOEXEC) + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + maintenance = Maintenance(selected, paths, payload, action) + try: + if action in {"drain", "stop", "restart"}: + maintenance.drain() + if action == "stop": + for name in ("rabbitmq", "postgres"): + maintenance.change(name, "stop") + if action in {"start", "resume", "restart"}: + maintenance.start(resume=action == "resume") + maintenance.record["success"] = True + maintenance.save("complete") + except (OSError, ValueError, subprocess.SubprocessError) as error: + maintenance.record["failure_code"] = "maintenance_unconfirmed" + # Publication errors are not hidden by a successful-looking receipt. + maintenance.save("failed") + raise DeliveryError( + "maintenance failed; admission not reopened automatically" + ) from error + return maintenance.record + finally: + os.close(fd) + + +def main() -> None: + parser = argparse.ArgumentParser(description="Fixed bounded single-node lifecycle") + parser.add_argument( + "--action", + choices=("start", "resume", "drain", "stop", "restart"), + required=True, + ) + parser.add_argument("--inside", action="store_true") + args = parser.parse_args() + try: + if args.inside: + result = run(args.action) + else: + if os.getuid() != 0 or os.geteuid() != 0: + raise DeliveryError("root operator required") + selected = install.selection(check_configuration=False) + install.directory(availability.GATE.parent, 0o711) + install.directory(STATE, 0o700) + result = operator_units.run( + "node-maintenance", + install.RELEASES / selected["release_id"] / "payload", + action=args.action, + ) + print(json.dumps(result, sort_keys=True)) + except (OSError, ValueError, subprocess.SubprocessError) as error: + parser.exit( + 1, + "Lifecycle refused: " + + (str(error) if isinstance(error, DeliveryError) else type(error).__name__) + + "\n", + ) + + +if __name__ == "__main__": + main() diff --git a/scripts/release/node_health.py b/scripts/release/node_health.py index b48176f..d65d413 100644 --- a/scripts/release/node_health.py +++ b/scripts/release/node_health.py @@ -11,6 +11,7 @@ from pathlib import Path from uuid import UUID from . import ( + availability, configuration, database, dependencies, @@ -58,11 +59,13 @@ def binding() -> tuple[dict, dict, Path]: selected = install.selection(check_configuration=False) payload = install.RELEASES / selected["release_id"] / "payload" for name in ( + "availability.py", "node_health.py", "broker_health.py", "factory_health.py", "import_health.py", "operator_units.py", + "outbox_health.py", ): shipped = payload / "delivery/scripts/release" / name preflight.protected(shipped) @@ -345,7 +348,13 @@ def validate_factory(value: dict, expected: str) -> None: def validate_execution(value: dict, admission: dict) -> dict: fields( value, - {"schema_version", "admission", "configured_workers", "lease_seconds", "counters"}, + { + "schema_version", + "admission", + "configured_workers", + "lease_seconds", + "counters", + }, "execution snapshot", ) if ( @@ -412,7 +421,11 @@ async def notification_probe(url: str, identifier: str) -> None: # A synthetically absent evaluation notification cannot create a grade or # execute a program: Coordinator must find and lease an existing row. body = canonical( - {"schema_version": 1, "kind": "judge.evaluation", "evaluation_id": identifier} + { + "schema_version": 1, + "kind": "judge.evaluation", + "evaluation_id": identifier, + } ).strip() reply = await channel.default_exchange.publish( aio_pika.Message( @@ -438,6 +451,7 @@ def observe(*, probe: bool) -> dict: "source_sha": read_json(payload.parent / "manifest.json")["source_sha"], "configuration_id": selected["configuration_id"], "ready_for_admission": False, + "service_preconditions_met": False, "isolation_verified": False, "components": {}, "errors": [], @@ -476,6 +490,12 @@ def observe(*, probe: bool) -> dict: components["broker"] = operator_units.run("broker-health", payload) except (OSError, ValueError, subprocess.SubprocessError): errors.append("broker_metadata_unavailable") + try: + state = operator_units.run("outbox-health", payload) + validate_outbox(state) + components["outbox_pipeline"] = state + except (OSError, ValueError, subprocess.SubprocessError): + errors.append("outbox_pipeline_unavailable") try: state = operator_units.run("import-health", payload) fields( @@ -534,7 +554,8 @@ def observe(*, probe: bool) -> dict: raise DeliveryError("notification probe authority unavailable") asyncio.run( asyncio.wait_for( - notification_probe(endpoints["amqp"]["outbox"], str(identifier)), timeout=6 + notification_probe(endpoints["amqp"]["outbox"], str(identifier)), + timeout=6, ) ) components["notification_delivery_confirmed"] = True @@ -545,12 +566,41 @@ def observe(*, probe: bool) -> dict: if not errors: root = components["factory_status"] counters = components["execution"] - result["ready_for_admission"] = admission_ready( + result["service_preconditions_met"] = admission_ready( root, counters, components["broker"], coordinator["admission"] ) + record = availability.read() + result["ready_for_admission"] = bool( + result["service_preconditions_met"] + and record is not None + and record.release_id == selected["release_id"] + and record.configuration_id == selected["configuration_id"] + and record.state == "open" + ) return result +def validate_outbox(state: dict) -> None: + fields( + state, + {"schema_version", "pending_capped", "locked_capped", "truncated"}, + "outbox status", + ) + if ( + type(state["schema_version"]) is not int + or state["schema_version"] != 1 + or type(state["truncated"]) is not bool + ): + raise DeliveryError("outbox projection binding differs") + for name in ("pending_capped", "locked_capped"): + if type(state[name]) is not int or not 0 <= state[name] <= 128: + raise DeliveryError("outbox projection bound differs") + if state["locked_capped"] > state["pending_capped"] or ( + state["truncated"] and state["pending_capped"] != 128 + ): + raise DeliveryError("outbox projection counters inconsistent") + + def main() -> None: parser = argparse.ArgumentParser(description="Bounded installed service readiness") parser.add_argument("--inside", action="store_true") diff --git a/scripts/release/operator_units.py b/scripts/release/operator_units.py index ebe1ddd..12eb782 100644 --- a/scripts/release/operator_units.py +++ b/scripts/release/operator_units.py @@ -15,6 +15,8 @@ PROFILES = { "broker-health": ("rabbitmq", 1204, 256, 64, 50, 45), "factory-health": ("kf-worker", 1202, 32, 4, 10, 10), "import-health": ("kf-oj", 1200, 128, 16, 25, 10), + "outbox-health": ("kf-oj", 1200, 128, 16, 25, 10), + "node-maintenance": ("root", 0, 256, 32, 50, 900), } @@ -73,9 +75,20 @@ def decode(contents: bytes, maximum: int = 32768): ) -def run(name: str, payload: Path, *, notification_probe: bool = False) -> dict: +def run( + name: str, + payload: Path, + *, + notification_probe: bool = False, + action: str | None = None, +) -> dict: if os.getuid() != 0 or os.geteuid() != 0 or name not in PROFILES: raise DeliveryError("fixed root operator helper required") + if name == "node-maintenance": + if action not in {"start", "resume", "drain", "stop", "restart"} or notification_probe: + raise DeliveryError("explicit fixed maintenance action required") + elif action is not None: + raise DeliveryError("unexpected operator action override") user, _, memory, pids, cpu, seconds = PROFILES[name] unit = "kf-" + name + ".service" current = dict( @@ -140,17 +153,26 @@ def run(name: str, payload: Path, *, notification_probe: bool = False) -> dict: } for key, value in properties.items(): command.append("--property=" + key + "=" + value) - if name == "import-health": + if name in {"import-health", "outbox-health"}: path = configuration.installed_path( - selected["release_id"], selected["configuration_id"], "import.env" + selected["release_id"], + selected["configuration_id"], + "import.env" if name == "import-health" else "outbox.env", ) command += [ "--property=EnvironmentFile=" + str(path), "--property=PrivatePIDs=yes", - "--property=PrivateNetwork=yes", - "--property=RestrictAddressFamilies=AF_UNIX", ] - if name == "import-health": + if name == "import-health": + command += [ + "--property=PrivateNetwork=yes", + "--property=RestrictAddressFamilies=AF_UNIX", + ] + if name == "node-maintenance": + command += [ + "--property=ReadWritePaths=/etc/kf-admission /var/lib/kf-onlinejudge/operations" + ] + if name in {"import-health", "outbox-health"}: if notification_probe: raise DeliveryError("unexpected import probe override") command += [ @@ -162,7 +184,7 @@ def run(name: str, payload: Path, *, notification_probe: bool = False) -> dict: "-s", "-P", "-m", - "scripts.release.import_health", + "scripts.release." + name.replace("-", "_"), "--inside", ] else: @@ -179,16 +201,20 @@ def run(name: str, payload: Path, *, notification_probe: bool = False) -> dict: "scripts.release.factory_health", "--inside", ] - elif name != "import-health": + elif name not in {"import-health", "outbox-health"}: command += [ *python_environment(payload).split(), str(payload / "python/bin/python3.13"), "-s", "-P", "-m", - "scripts.release." + name.replace("-", "_"), + "scripts.release.lifecycle" + if name == "node-maintenance" + else "scripts.release." + name.replace("-", "_"), "--inside", ] + if name == "node-maintenance": + command += ["--action", action] if notification_probe and name == "node-health": command.append("--probe-notification") elif notification_probe: @@ -211,7 +237,9 @@ def run(name: str, payload: Path, *, notification_probe: bool = False) -> dict: actual = dict( line.split("=", 1) for line in service_install.systemctl( - "show", unit, "--property=LoadState,ActiveState,MainPID,Transient,Description" + "show", + unit, + "--property=LoadState,ActiveState,MainPID,Transient,Description", ).splitlines() if "=" in line ) diff --git a/scripts/release/outbox_health.py b/scripts/release/outbox_health.py new file mode 100644 index 0000000..1db86a2 --- /dev/null +++ b/scripts/release/outbox_health.py @@ -0,0 +1,40 @@ +"""Ordinary publisher-only read projection with a separate private PID view.""" + +import argparse +import os +from pathlib import Path + +from . import install, operator_units, service_bootstrap +from .policy import DeliveryError + + +def main() -> None: + parser = argparse.ArgumentParser(description="Private ordinary outbox observation") + parser.add_argument("--inside", action="store_true", required=True) + parser.parse_args() + try: + operator_units.guarded("outbox-health") + status = dict( + line.split(":", 1) + for line in Path("/proc/self/status").read_text().splitlines() + if ":" in line + ) + if status.get("NSpid", "").split()[-1:] != ["1"]: + raise DeliveryError("private outbox observation view required") + payload = Path(__file__).resolve().parents[3] + if payload.parent.parent != install.RELEASES or payload.name != "payload": + raise DeliveryError("installed outbox observer required") + command, env = service_bootstrap.python_command("outbox", payload.parent.name) + command[-1] = "kf_oj.workers.outbox_status" + os.execve(command[0], command, env) + except (OSError, ValueError, KeyError) as error: + parser.exit( + 1, + "Outbox health refused: " + + (str(error) if isinstance(error, DeliveryError) else type(error).__name__) + + "\n", + ) + + +if __name__ == "__main__": + main() diff --git a/scripts/release/service_bootstrap.py b/scripts/release/service_bootstrap.py index cdce2fc..3cee846 100644 --- a/scripts/release/service_bootstrap.py +++ b/scripts/release/service_bootstrap.py @@ -273,6 +273,11 @@ def main() -> None: if args.credential_directory is not None or args.sha256 is not None: raise DeliveryError("unexpected ordinary credential override") command, environment = python_command(args.role, args.release_id) + if args.role == "api": + # Added only after fixed argument and environment-file checks; + # the batch ID cannot be embedded in its own hashed env file. + environment["KF_OJ_DELIVERY_RELEASE_ID"] = args.release_id + environment["KF_OJ_DELIVERY_CONFIGURATION_ID"] = args.configuration_id else: if args.credential_directory is None or args.sha256 is None: raise DeliveryError("fixed ordinary credential required") diff --git a/tests/fixtures/contracts/delivery-availability.valid.json b/tests/fixtures/contracts/delivery-availability.valid.json new file mode 100644 index 0000000..6da318c --- /dev/null +++ b/tests/fixtures/contracts/delivery-availability.valid.json @@ -0,0 +1,6 @@ +{ + "schema_version": 1, + "release_id": "sd-123456789abc", + "configuration_id": "cfg-bbbbbbbbbbbbbbbbbbbbbbbb", + "state": "draining" +}