File size: 10,231 Bytes
e1acc15 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 | """PP4 shared-owner integration; CPU fixtures only, real scopes for receiving hops."""
from __future__ import annotations
import json
import shlex
import subprocess
from pathlib import Path
from typing import cast
import pytest
from offload.launch.windows import hold_admission
from offload.launch.windows import pf1_window as pf
from offload.launch.windows import pp4_window as w
from offload.launch.windows import scoped_remote as owner
from offload.launch.windows.tests.test_pp4_window import WORKER
from offload.launch.windows.tests.test_scoped_remote import Sandbox, wait_for
pytest_plugins = ["offload.launch.windows.tests.test_scoped_remote"]
@pytest.mark.parametrize('kind', ['stop', 'run'])
@pytest.mark.parametrize('pause', ['before-registration', 'active'])
def test_real_pp4_owned_second_hop_and_retry(sandbox: Sandbox, kind: str, pause: str, monkeypatch: pytest.MonkeyPatch) -> None:
s = sandbox
tag = s.tag('pp4-' + kind)
home = s.homes['sparkworker-wired']
state = home/'production'
state.write_text('active')
release = home/'release'
if pause == 'before-registration':
(s.root/'sparkworker-wired-transport-gate').touch()
body = f'touch {home}/active; while [ ! -e {release} ]; do sleep .02; done; echo {kind} > {state}'
live = w.Live(s.root, pf.Supervisor(s.root), native=WORKER, remote='experiment', expected=WORKER.commit,
data_dir=s.root, tag=tag, admission={'verifier_sha256': {}})
live.sup.abort_at = s.deadlines[tag]
submitted: list[str] = []
def ssh(label: str, host: str, command: str, stdin: str | None = None, *, timeout: float = 120) -> str:
submitted.append(command)
return ''
monkeypatch.setattr(live, 'ssh', ssh)
live.owned('engine-start', 'sparkhead', tag, 'ssh sparkworker ' + shlex.quote(body), timeout=30)
assert live.owner_tags == [tag] and live.owner_deadlines[tag] == s.deadlines[tag]
process = subprocess.Popen(['bash', '-c', submitted[0]], env=s.env(), stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL, text=True)
s.processes.append(process)
wait_for(s.root/'sparkworker-wired-accepted' if pause == 'before-registration' else home/'active')
s.close('new', tag)
(s.root/'sparkworker-wired-transport-gate').unlink(missing_ok=True)
release.touch()
process.wait(timeout=10)
assert state.read_text() == 'active'
retry = s.run(s.command('new', 'launch', tag, ['bash', '-c', body]), 'sparkworker-wired')
assert retry.returncode == 75 and state.read_text() == 'active'
live.sup.close()
def test_s55_lease_origin_admission_and_no_renewal(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
clock = [1000.]
sup = pf.Supervisor(tmp_path, clock=lambda: clock[0])
live = w.Live(tmp_path, sup, native=WORKER, remote='experiment', expected=WORKER.commit,
data_dir=tmp_path, tag='lease', admission={'verifier_sha256': {}}, clock=lambda: clock[0])
clock[0] += 1080 # long preflight must not consume placeholder lifetime
before = clock[0]
calls: list[str] = []
def ssh(label: str, host: str, command: str, stdin: str | None = None, *, timeout: float = 120) -> str:
calls.append(label)
if label == 'hold-open':
assert 'SCOPED_REMOTE_DEADLINE=' in command and hold_admission.NOT_OWNED in command
clock[0] += pf.DRAIN_S - 10
from offload.launch.windows.tests.test_pf1_window import iso
return iso(clock[0]) + ' DSV41-OFFLOAD (session coordinator): PP4-WINDOW lease: fixture'
monkeypatch.setattr(live, 'ssh', ssh)
monkeypatch.setattr(sup, 'tunnel', lambda argv: None)
live.open_hold()
assert live.open_deadline == before + pf.DRAIN_S + 60
assert live.owner_deadlines['lease-hold'] == live.hold_scope_deadline >= live.restore_by
saved = live.restore_by, live.hold_scope_deadline
with pytest.raises(ValueError, match='cannot be reopened or renewed'):
live.open_hold()
assert saved == (live.restore_by, live.hold_scope_deadline)
sup.close()
@pytest.mark.parametrize('mode', ['fence', 'reap', 'absent'])
@pytest.mark.parametrize('host', w.OWNER_HOSTS)
def test_all_host_control_failures_prevent_pair_and_w8(tmp_path: Path, mode: str, host: str) -> None:
calls: list[str] = []
def plain(label: str, argv: list[str], timeout: float) -> tuple[int, str]:
calls.append(label)
if label == f'owned-{mode}-engine-{host}':
return 1, 'injected missing proof'
return 0, owner.FENCED if label.startswith('owned-fence-') else owner.ABSENT
sup = pf.Supervisor(tmp_path)
live = w.Live(tmp_path, sup, native=WORKER, remote='experiment', expected=WORKER.commit,
data_dir=tmp_path, tag='fixture', admission={'verifier_sha256': {}}, plain=plain)
live.hold_written = True
live.remember('engine', live.restore_by - 90)
with pytest.raises(ValueError, match='HOLD stays open'):
live.restore(production_stopped=True)
assert 'pair-close' not in calls and 'start' not in calls and 'commit-close' not in calls
if mode == 'fence':
assert not any(label.startswith(('owned-reap-', 'owned-absent-')) for label in calls)
def test_pins_only_shared_ownership_and_recovery_is_scoped() -> None:
assert 'offload/launch/windows/scoped_remote.py' in w.PIN_FILES
assert not (pf.ROOT/'offload/launch/windows/pp4_remote.py').exists()
assert {'offload/launch/windows/'+name+'.py' for name in
('post_close_watch', 'hold_admission', 'spark_quiet', 'production_health')} <= set(w.PIN_FILES)
def test_repeated_recovery_proofs_keep_distinct_logs(tmp_path: Path) -> None:
sup = pf.Supervisor(tmp_path)
(tmp_path/'cmd').mkdir(exist_ok=True)
live = w.Live(tmp_path, sup, native=WORKER, remote='experiment', expected=WORKER.commit,
data_dir=tmp_path, tag='repeat', admission={'verifier_sha256': {}})
for text in ('first proof', 'second proof'):
assert live.local('same-phase-absent', ['bash', '-c', 'printf %s '+shlex.quote(text)], 5) == (0, text)
logs = sorted((tmp_path/'cmd').glob('restore-*.log'))
assert [p.name for p in logs] == ['restore-0001-same-phase-absent.log', 'restore-0002-same-phase-absent.log']
assert not (tmp_path/'cmd'/'restore-same-phase-absent.log').exists()
assert [p.read_text() for p in logs] == ['first proof', 'second proof']
sup.close()
@pytest.mark.parametrize('restore,watch_status,exit_code', [
('RESTORED: verified', 'PASS', 0), ('RESTORED: verified', 'FAILED', 2),
('FAILED', 'PASS', 2), ('NOT NEEDED: preflight only', 'PASS', 0),
])
def test_main_watchdog_is_mandatory_after_restore(tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
restore: str, watch_status: str, exit_code: int) -> None:
from types import SimpleNamespace
from offload.launch.windows import post_close_watch as watch
from offload.launch.windows import pp4_prerequisites as prereq
from offload.prefill_maxlen.evidence import Json, record
monkeypatch.setenv('WINDOW_COORDINATOR', 'Deepseek/coordinator')
monkeypatch.setattr(pf, 'CAMPAIGN', tmp_path)
monkeypatch.setattr(subprocess, 'check_output', lambda argv, **kwargs: WORKER.commit if argv[1] == 'rev-parse' else '')
monkeypatch.setattr(prereq, 'staging', lambda *args: {'worker_identity': WORKER.identity, 'markov_identity': WORKER.markov})
calls: list[str] = []
monkeypatch.setattr(pf, 'Supervisor', lambda out: SimpleNamespace(close=lambda: calls.append('close')))
monkeypatch.setattr(w, 'Live', lambda *args, **kwargs: SimpleNamespace())
monkeypatch.setattr(w, 'window', lambda *args, **kwargs: {
'status': 'PAIR_QUALIFIED', 'restore': {'status': restore}, 'decision': {'status': 'CANDIDATE'}})
monkeypatch.setattr(watch, 'ProductionHealth', object)
def recovery(tag: str, out: Path) -> object:
assert calls == ['close']
assert tag == 'PP4-test-watch'
calls.append('recovery')
return object()
def watchdog(health: object, recovery: object, *, log: object) -> watch.Result:
calls.append('watch')
return watch.Result(watch_status, 30, 0, 0)
monkeypatch.setattr(watch, 'ScopedRecovery', recovery)
monkeypatch.setattr(watch, 'watch', watchdog)
import signal
saved = {sig: signal.getsignal(sig) for sig in (signal.SIGTERM, signal.SIGINT, signal.SIGHUP)}
try:
rc = w.main(['--execute', '--pair-only', '--qualification', str(tmp_path/'staging.json'),
'--expected-experiment-commit', WORKER.commit, '--worker-commit', WORKER.commit,
'--worker-tree', WORKER.tree, '--worker-identity', WORKER.identity,
'--worker-markov-identity', WORKER.markov, '--out', str(tmp_path/'PP4-test')])
finally:
for sig, handler in saved.items():
signal.signal(sig, handler)
assert rc == exit_code
assert ('watch' in calls) == restore.startswith('RESTORED')
result = record(cast(Json, json.loads((tmp_path/'PP4-test/summary.json').read_text())))
if restore.startswith('RESTORED'):
assert record(result['post_close_watch'])['status'] == watch_status
assert record(result['decision'])['status'] == ('CANDIDATE' if watch_status == 'PASS' else 'REJECT')
def test_local_control_precheck_never_requires_loopback_ssh(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
sup = pf.Supervisor(tmp_path)
live = w.Live(tmp_path, sup, native=WORKER, remote='experiment', expected=WORKER.commit,
data_dir=tmp_path, tag='fresh', admission={'verifier_sha256': {}})
calls: list[list[str]] = []
def call(label: str, argv: list[str], **kwargs: object) -> str:
calls.append(argv)
return ''
monkeypatch.setattr(live, 'call', call)
live.fresh_controls(['fresh-hold', 'fresh-worker'])
assert [argv[-2] for argv in calls if argv[0] == 'ssh'] == list(owner.NODES)
local = [argv for argv in calls if argv[0] != 'ssh']
assert len(local) == 1 and local[0][:2] == ['bash', '-c']
assert 'scoped-control/fresh-hold' in local[0][-1] and 'scoped-control/fresh-worker' in local[0][-1]
sup.close()
|