From 8e8015d3f834d30d7934d8d95d08aef4d799d0a3 Mon Sep 17 00:00:00 2001 From: kronecker <2026692687@qq.com> Date: Wed, 21 Oct 2026 08:03:53 +0800 Subject: [PATCH] perf(delivery): start independent services in bounded groups --- backend/tests/test_delivery.py | 2 +- backend/tests/test_delivery_io.py | 178 ++++++++++++++++++++++++++++++ deploy/release/README.md | 8 ++ scripts/release/lifecycle.py | 57 +++++++++- 4 files changed, 240 insertions(+), 5 deletions(-) diff --git a/backend/tests/test_delivery.py b/backend/tests/test_delivery.py index af35f17..43b4b85 100644 --- a/backend/tests/test_delivery.py +++ b/backend/tests/test_delivery.py @@ -2271,7 +2271,7 @@ def test_maintenance_does_not_leave_admission_open_when_final_confirmation_fails maintenance = object.__new__(lifecycle.Maintenance) maintenance.selected = {} maintenance.payload = Path("/unused") - monkeypatch.setattr(maintenance, "change", lambda *_: None) + monkeypatch.setattr(maintenance, "start_group", lambda *_: None) monkeypatch.setattr(recovery_database, "admission_guard", nullcontext) def publish(_, state): diff --git a/backend/tests/test_delivery_io.py b/backend/tests/test_delivery_io.py index 65438e6..1ee5931 100644 --- a/backend/tests/test_delivery_io.py +++ b/backend/tests/test_delivery_io.py @@ -21,6 +21,7 @@ file_copy = importlib.import_module("scripts.release.file_copy") factory_service = importlib.import_module("scripts.release.factory_service") proxy_install = importlib.import_module("scripts.release.proxy_install") setup_session = importlib.import_module("scripts.release.setup_session") +lifecycle = importlib.import_module("scripts.release.lifecycle") @pytest.fixture @@ -584,3 +585,180 @@ def test_setup_session_failed_alarm_installation_restores_handler(monkeypatch): monkeypatch.setattr(setup_session.signal, "setitimer", real) if original_timer[0]: signal.setitimer(signal.ITIMER_REAL, *original_timer) + + +@pytest.fixture +def ordinary_lifecycle_model(monkeypatch): + # Every manager, clock, identity check and publication is modeled here. + # Actual overlapping starts, limits, Root and cleanup remain VM-only. + maintenance = object.__new__(lifecycle.Maintenance) + maintenance.selected = {"release_id": "sd-" + "b" * 12, "configuration_id": "cfg-" + "c" * 24} + maintenance.payload = Path("/ordinary-unused-payload") + maintenance.record = {"events": [], "root_cleanup_confirmed": True} + states = {name: "inactive" for name in (*lifecycle.DEPENDENCIES, *lifecycle.APPS)} + trace, clock = [], [0.0] + + def unit(name): + trace.append(("read", name)) + return { + "ActiveState": states[name], + "SubState": "running" if states[name] == "active" else "dead", + "MainPID": "123" if states[name] == "active" else "0", + } + + def dispatch(*args): + assert args[:2] == ("--no-block", "start") + name = args[2].removeprefix("kf-").removesuffix(".service") + trace.append(("dispatch", name)) + states[name] = "active" + + monkeypatch.setattr(maintenance, "unit", unit) + monkeypatch.setattr(maintenance, "save", lambda phase: trace.append(("save", phase))) + monkeypatch.setattr(lifecycle.install, "selection", lambda **_: maintenance.selected) + monkeypatch.setattr(lifecycle.service_install, "systemctl", dispatch) + monkeypatch.setattr( + lifecycle.node_health, "inspect_unit", lambda name, _: trace.append(("inspect", name)) + ) + monkeypatch.setattr(lifecycle.time, "monotonic", lambda: clock[0]) + monkeypatch.setattr( + lifecycle.time, "sleep", lambda seconds: clock.__setitem__(0, clock[0] + seconds) + ) + return maintenance, states, trace, clock + + +def test_start_group_dispatches_independent_units_before_confirming_each(ordinary_lifecycle_model): + maintenance, _, trace, _ = ordinary_lifecycle_model + names = lifecycle.START_GROUPS[0] + maintenance.start_group(names) + starts = [(i, name) for i, (kind, name) in enumerate(trace) if kind == "dispatch"] + inspections = [(i, name) for i, (kind, name) in enumerate(trace) if kind == "inspect"] + assert [name for _, name in starts] == list(names) + assert {name for _, name in inspections} == set(names) + assert min(i for i, _ in inspections) > max(i for i, _ in starts) + assert all(event["confirmed"] for event in maintenance.record["events"]) + assert maintenance.record["root_cleanup_confirmed"] is False + + +def test_start_group_validates_all_targets_before_any_dispatch( + ordinary_lifecycle_model, monkeypatch +): + maintenance, _, trace, _ = ordinary_lifecycle_model + original = maintenance.unit + + def unknown(name): + if name == "rabbitmq": + raise policy.DeliveryError("ordinary unknown target model") + return original(name) + + monkeypatch.setattr(maintenance, "unit", unknown) + with pytest.raises(policy.DeliveryError, match="unknown target"): + maintenance.start_group(lifecycle.START_GROUPS[0]) + assert not any(kind == "dispatch" for kind, _ in trace) + assert maintenance.record["events"] == [] + + +def test_later_dispatch_does_not_reset_an_earlier_start_deadline( + ordinary_lifecycle_model, monkeypatch +): + maintenance, _, trace, clock = ordinary_lifecycle_model + original = lifecycle.service_install.systemctl + + def delayed(*args): + original(*args) + if args[-1] == "kf-rabbitmq.service": + clock[0] = 131 + + monkeypatch.setattr(lifecycle.service_install, "systemctl", delayed) + with pytest.raises(policy.DeliveryError, match="bounded lifecycle"): + maintenance.start_group(lifecycle.START_GROUPS[0]) + assert [name for kind, name in trace if kind == "dispatch"] == ["postgres", "rabbitmq"] + assert all(not event["confirmed"] for event in maintenance.record["events"]) + + +def test_start_group_refuses_changed_selection_before_dispatch( + ordinary_lifecycle_model, monkeypatch +): + maintenance, _, trace, _ = ordinary_lifecycle_model + monkeypatch.setattr(lifecycle.install, "selection", lambda **_: {}) + with pytest.raises(policy.DeliveryError, match="selected binding changed"): + maintenance.start_group(lifecycle.START_GROUPS[0]) + assert not any(kind == "dispatch" for kind, _ in trace) + + +def test_start_group_resume_checks_live_profiles_without_duplicate_starts(ordinary_lifecycle_model): + maintenance, states, trace, _ = ordinary_lifecycle_model + group = lifecycle.START_GROUPS[0] + for name in group: + states[name] = "active" + maintenance.start_group(group) + assert {name for kind, name in trace if kind == "inspect"} == set(group) + assert not any(kind == "dispatch" for kind, _ in trace) + assert maintenance.record["events"] == [] + + +def test_group_profile_failure_cannot_confirm_a_started_unit(ordinary_lifecycle_model, monkeypatch): + maintenance, _, trace, _ = ordinary_lifecycle_model + + def refused(name, selected): + raise policy.DeliveryError("ordinary effective profile refusal model") + + monkeypatch.setattr(lifecycle.node_health, "inspect_unit", refused) + with pytest.raises(policy.DeliveryError, match="effective profile refusal"): + maintenance.start_group(lifecycle.START_GROUPS[0]) + assert len([name for kind, name in trace if kind == "dispatch"]) == 3 + assert all(not event["confirmed"] for event in maintenance.record["events"]) + + +def test_startup_dependency_barriers_precede_both_real_readiness_calls( + ordinary_lifecycle_model, monkeypatch +): + maintenance, states, trace, _ = ordinary_lifecycle_model + gate, observations = [], [] + monkeypatch.setattr(lifecycle.availability, "write", lambda _, state: gate.append(state)) + + def observe(*args): + verified = {name for kind, name in trace if kind == "inspect"} + assert verified == set(states) and all(value == "active" for value in states.values()) + observations.append(gate[-1]) + return {"service_preconditions_met": True, "ready_for_admission": gate[-1] == "open"} + + monkeypatch.setattr(lifecycle.operator_units, "run", observe) + maintenance.start_locked(resume=True) + assert gate == observations == ["draining", "open"] + dispatches = [name for kind, name in trace if kind == "dispatch"] + assert dispatches == [*lifecycle.DEPENDENCIES, *lifecycle.START_ORDER] + for dependency in (*lifecycle.DEPENDENCIES, "sandbox-factory"): + assert trace.index(("inspect", dependency)) < trace.index(("dispatch", "coordinator")) + assert trace.index(("inspect", "api")) < trace.index(("dispatch", "proxy")) + + +def test_startup_group_failure_keeps_admission_closed(ordinary_lifecycle_model, monkeypatch): + maintenance, states, trace, _ = ordinary_lifecycle_model + gate = [] + monkeypatch.setattr(lifecycle.availability, "write", lambda _, state: gate.append(state)) + original = lifecycle.service_install.systemctl + + def failing(*args): + original(*args) + states["rabbitmq"] = "failed" + + monkeypatch.setattr(lifecycle.service_install, "systemctl", failing) + monkeypatch.setattr( + lifecycle.operator_units, + "run", + lambda *_: pytest.fail("readiness cannot follow failed stage"), + ) + with pytest.raises(policy.DeliveryError, match="bounded lifecycle"): + maintenance.start_locked(resume=True) + assert gate == ["draining"] + assert not any( + kind == "dispatch" and name in lifecycle.services.PROFILES for kind, name in trace + ) + + +@pytest.mark.parametrize("names", [(), ("api",), ("postgres", "postgres")]) +def test_startup_rejects_unknown_or_duplicate_groups(ordinary_lifecycle_model, names): + maintenance, _, trace, _ = ordinary_lifecycle_model + with pytest.raises(policy.DeliveryError, match="unknown startup dependency group"): + maintenance.start_group(names) + assert trace == [] diff --git a/deploy/release/README.md b/deploy/release/README.md index ff8a906..860907f 100644 --- a/deploy/release/README.md +++ b/deploy/release/README.md @@ -126,6 +126,14 @@ volumes, then follow this order: The tools use offline materials and do not contact a package index. +Startup dispatches PostgreSQL, RabbitMQ and the Root factory together, then the +five ordinary services, then HTTPS after API startup. Each stage waits for every +member's actual manager state and resource/identity checks before continuing. +Every service retains its complete startup verification and its original unit +and observation deadlines, measured from that service's dispatch. Admission stays +closed until both normal readiness checks confirm the complete node. Drain and +stop retain their ordered owner and cleanup checks. + Installer automation can run the same setup CLI entry points in one bounded Python operation. Use the protected delivery interpreter and its normal Python environment, as the trusted Root operator: diff --git a/scripts/release/lifecycle.py b/scripts/release/lifecycle.py index de50a11..7350db6 100644 --- a/scripts/release/lifecycle.py +++ b/scripts/release/lifecycle.py @@ -49,6 +49,13 @@ START_ORDER = ( "worker", "proxy", ) +# Root has no DB/MQ dependency. Ordinary units require the two dependencies; +# the HTTPS unit additionally orders itself after API startup. +START_GROUPS = ( + (*DEPENDENCIES, "sandbox-factory"), + ("coordinator", "api", "outbox", "problem-import", "worker"), + ("proxy",), +) DRAIN_SECONDS = 360 @@ -399,6 +406,50 @@ class Maintenance: self.record["events"][-1]["confirmed"] = True self.save(operation + "_confirmed") + def start_group(self, names: tuple[str, ...]) -> None: + """Dispatch one fixed dependency stage; confirm each real unit separately.""" + if names not in START_GROUPS: + raise DeliveryError("unknown startup dependency group") + if "sandbox-factory" in names: + self.record["root_cleanup_confirmed"] = False + if install.selection(check_configuration=False) != self.selected: + raise DeliveryError("selected binding changed during maintenance") + # Validate every target before dispatching any member of this stage. + inactive = [] + for name in names: + values = self.unit(name) + if values["ActiveState"] == "active": + node_health.inspect_unit(name, self.selected) + else: + inactive.append(name) + pending = {} + for name in inactive: + if any(time.monotonic() >= deadline for _, deadline in pending.values()): + raise DeliveryError("bounded lifecycle operation unconfirmed") + if install.selection(check_configuration=False) != self.selected: + raise DeliveryError("selected binding changed during maintenance") + event = {"unit": name, "operation": "start", "confirmed": False} + self.record["events"].append(event) + self.save("dispatching_start") + service_install.systemctl("--no-block", "start", "kf-" + name + ".service") + seconds = factory_service.START_TIMEOUT_SECONDS if name == "sandbox-factory" else 120 + # A later dispatch or wait cannot reset an earlier unit's deadline. + pending[name] = (event, time.monotonic() + seconds + 10) + while pending: + for name, (event, deadline) in list(pending.items()): + if install.selection(check_configuration=False) != self.selected: + raise DeliveryError("selected binding changed during maintenance") + values = self.unit(name) + if time.monotonic() >= deadline or values["ActiveState"] == "failed": + raise DeliveryError("bounded lifecycle operation unconfirmed") + if values["ActiveState"] == "active" and values["SubState"] == "running": + node_health.inspect_unit(name, self.selected) + event["confirmed"] = True + self.save("start_confirmed") + del pending[name] + if pending: + time.sleep(0.5) + def snapshot(self) -> dict: owner = node_health.http( "http://127.0.0.1:8100/internal/operations/status", @@ -522,10 +573,8 @@ class Maintenance: 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") + for group in START_GROUPS: + self.start_group(group) 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")