diff --git a/docs/ko/recovery-qualification.md b/docs/ko/recovery-qualification.md index 292a200..95d4a2a 100644 --- a/docs/ko/recovery-qualification.md +++ b/docs/ko/recovery-qualification.md @@ -59,7 +59,7 @@ env -i HOME="$HOME" TMPDIR="$TMPDIR" PATH="/bin:/usr/bin:/bin:/usr/ ## 기능 스위트 -시나리오마다 작업 디렉터리, 작업 데이터베이스, Runner를 따로 쓰고, `required`이면 fixture authority도 따로 씁니다. 시나리오가 남긴 실행 중인 것은 그 뒤에 끝냅니다. 시나리오는 확인 항목이 모두 통과해야 통과합니다. 시나리오가 따로 말하지 않으면 gateway의 프로세스 timeout은 10분이고 종료 의도는 `detach`입니다. I/O 칸의 "둘 다"는 pipe로 한 번, PTY로 한 번 실행한다는 뜻이고, 참여 칸은 그 시나리오를 실행하는 참여를 나열합니다. +시나리오마다 작업 디렉터리, 작업 데이터베이스, Runner를 따로 쓰고, `required`이면 fixture authority도 따로 씁니다. 시나리오가 남긴 실행 중인 것은 그 뒤에 끝냅니다. 시나리오는 확인 항목이 모두 통과해야 통과합니다. 시나리오가 따로 말하지 않으면 gateway의 프로세스 timeout은 10분이고 종료 의도는 `detach`입니다. I/O 칸의 "둘 다"는 pipe로 한 번, PTY로 한 번 실행한다는 뜻이고, 참여 칸은 그 시나리오를 실행하는 참여를 나열합니다. 재시작 직후 gateway는 작업이 시작된 적 있는 workspace마다 그 workspace를 대조할 때까지 새 작업을 `RECOVERY_PENDING`으로 거부합니다. 프로세스를 위해 workspace를 점유한 gateway는 Runner가 그 프로세스를 회수하면 끝났다고 보고하지만, workspace는 프로세스의 출력이 닫혀 Runner의 종료 이벤트가 뒤따를 때에야 풀며, 그때까지 그 workspace의 새 작업을 `WORKSPACE_BUSY`로 거부합니다. 시나리오가 새 작업을 시작해야 할 때는 이 거부 가운데 하나를 받는 동안 최대 30초까지 다시 보내고, 통과 조건이 이 거부 가운데 하나를 기대하는 시나리오는 그 거부를 확인합니다. @@ -147,7 +147,7 @@ env -i HOME="$HOME" TMPDIR="$TMPDIR" PATH="/bin:/usr/bin:/bin:/usr/ ### 구간, 주기, 표본 -한 번의 반복은 10분 유휴 구간과 그 뒤 30분 이상의 부하 구간이고, 절차는 세 번 반복합니다. 구간마다 새 Runner를 시작하며, 이 Runner는 `codespace` 소비자의 유일한 인스턴스로 등록합니다(결정 D12). 첫 gateway는 구간 내내 실행되는 프로세스를 시작합니다. 각각 25 mCPU, 16 MiB, 2 tasks를 요청하는 관리 대상 workspace의 `cat`(pipe)과 `sleep`(PTY), 그리고 관리하지 않는 workspace에서 0.2초마다 번호 붙은 줄을 쓰는 PTY와, 하네스가 신호를 줄 때마다 출력 ring이 담는 것보다 많은 번호 붙은 줄 40,000개를 한꺼번에 쓰는 pipe입니다. 요청을 받는 gateway가 무엇이든 그것을 통해 `control` 스위트의 일정대로 제어 경로를 표본으로 잽니다([구간과 표본](cs-rg-qualification.md#control-intervals)). status는 1초마다, 새 관리 대상 종료는 20초마다, I/O echo는 5초마다, 관리 대상 거부는 30초마다, 그리고 host, service, footprint(Runner와 gateway의 것)입니다. 두 출력 흐름은 2초마다 읽습니다. 요청을 받는 gateway가 없을 때 돌아온 일정은 표본이 아닌 `restarting`으로 기록하며, 기대하는 표본 수는 gateway가 요청을 받은 시간을 따릅니다. +한 번의 반복은 10분 유휴 구간과 그 뒤 30분 이상의 부하 구간이고, 절차는 세 번 반복합니다. 구간마다 새 Runner를 시작하며, 이 Runner는 `codespace` 소비자의 유일한 인스턴스로 등록합니다(결정 D12). 첫 gateway는 구간 내내 실행되는 프로세스를 시작합니다. 각각 25 mCPU, 16 MiB, 2 tasks를 요청하는 관리 대상 workspace의 `cat`(pipe)과 `sleep`(PTY), 그리고 관리하지 않는 workspace에서 0.2초마다 번호 붙은 줄을 쓰는 PTY와, 하네스가 신호를 줄 때마다 출력 ring이 담는 것보다 많은 번호 붙은 줄 40,000개를 한꺼번에 쓰는 pipe입니다. 요청을 받는 gateway가 무엇이든 그것을 통해 `control` 스위트의 일정대로 제어 경로를 표본으로 잽니다([구간과 표본](cs-rg-qualification.md#control-intervals)). status는 1초마다, 새 관리 대상 종료는 20초마다, I/O echo는 5초마다, 관리 대상 거부는 30초마다, 그리고 host, service, footprint(Runner와 gateway의 것)입니다. gateway는 모든 workspace를 대조하기 전에 요청을 받기 시작하므로, 요청을 받는 gateway가 종료 대상이나 거부 표본의 실행을 `RECOVERY_PENDING`으로 거부하면 최대 10초까지 다시 보내며, 그때까지도 거부된 종료 대상은 오류입니다. 두 출력 흐름은 2초마다 읽습니다. 요청을 받는 gateway가 없을 때 돌아온 일정은 표본이 아닌 `restarting`으로 기록하며, 기대하는 표본 수는 gateway가 요청을 받은 시간을 따릅니다. 구간은 gateway 주기의 연속입니다. 첫 주기는 20초이고, 그 뒤의 gateway는 모두 프로세스 timeout 20초로 실행을 시작하며, 그 주기의 길이는 그 주기의 timeout 대상이 정합니다. 각 주기에서: diff --git a/docs/recovery-qualification.md b/docs/recovery-qualification.md index 193a7b8..c13760c 100644 --- a/docs/recovery-qualification.md +++ b/docs/recovery-qualification.md @@ -59,7 +59,7 @@ env -i HOME="$HOME" TMPDIR="$TMPDIR" PATH="/bin:/usr/bin:/bin:/usr/ ## Functional suites -Each scenario gets its own work directory, operations database and Runner, and with `required` its own fixture authority; whatever a scenario leaves running is ended afterwards. A scenario passes only when each of its checks passed. Unless a scenario says otherwise, a gateway's process timeout is 10 minutes and its exit intent is `detach`. "Both" in the I/O column means that the scenario runs once with a pipe and once with a PTY, and the participation column lists the participations the scenario runs with. +Each scenario gets its own work directory, operations database and Runner, and with `required` its own fixture authority; whatever a scenario leaves running is ended afterwards. A scenario passes only when each of its checks passed. Unless a scenario says otherwise, a gateway's process timeout is 10 minutes and its exit intent is `detach`. "Both" in the I/O column means that the scenario runs once with a pipe and once with a PTY, and the participation column lists the participations the scenario runs with. Right after a restart a gateway refuses new work with `RECOVERY_PENDING` in each workspace where work started, until it has reconciled it. A gateway that holds a workspace for a process reports the process exited once the Runner has reaped it, but frees the workspace only when the Runner's exit event follows, once the process's output has closed, and refuses new work there with `WORKSPACE_BUSY` until then. Where a scenario needs new work started, it sends it again while it meets one of these refusals, for up to 30 seconds; where its row expects one of these refusals, it checks it. @@ -147,7 +147,7 @@ The measurement restarts the gateway again and again while governed pipe and PTY ### Intervals, cycles and samples -A repetition is a 10-minute idle interval followed by a load interval of at least 30 minutes; the protocol has three repetitions. Each interval starts a fresh Runner, which registers as the `codespace` consumer's one instance (decision D12), and its first gateway starts the processes that run the whole interval: a `cat` (pipe) and a `sleep` (PTY) in governed workspaces, which request 25 mCPU, 16 MiB and 2 tasks each, and in ungoverned ones a PTY that writes a numbered line every 0.2 seconds and a pipe that writes 40,000 numbered lines at once, more than the output ring holds, whenever the harness cues it. Through whichever gateway serves, the control path is sampled on the `control` suite's schedules ([intervals and samples](cs-rg-qualification.md#control-intervals)): status every second, a fresh governed termination every 20 seconds, the I/O echo every 5 seconds, a governed refusal every 30 seconds, and the host, service and footprint (the Runner's and the gateway's). Both output streams are read every 2 seconds. A slot that falls while no gateway serves is recorded as `restarting`, not as a sample, and the samples expected follow the time a gateway served. +A repetition is a 10-minute idle interval followed by a load interval of at least 30 minutes; the protocol has three repetitions. Each interval starts a fresh Runner, which registers as the `codespace` consumer's one instance (decision D12), and its first gateway starts the processes that run the whole interval: a `cat` (pipe) and a `sleep` (PTY) in governed workspaces, which request 25 mCPU, 16 MiB and 2 tasks each, and in ungoverned ones a PTY that writes a numbered line every 0.2 seconds and a pipe that writes 40,000 numbered lines at once, more than the output ring holds, whenever the harness cues it. Through whichever gateway serves, the control path is sampled on the `control` suite's schedules ([intervals and samples](cs-rg-qualification.md#control-intervals)): status every second, a fresh governed termination every 20 seconds, the I/O echo every 5 seconds, a governed refusal every 30 seconds, and the host, service and footprint (the Runner's and the gateway's). Since a gateway serves before it has reconciled every workspace, a termination's target or a refusal's execution that the serving gateway refuses with `RECOVERY_PENDING` is sent again, for up to 10 seconds; a target still refused then is an error. Both output streams are read every 2 seconds. A slot that falls while no gateway serves is recorded as `restarting`, not as a sample, and the samples expected follow the time a gateway served. The interval is a sequence of gateway cycles. The first lasts 20 seconds; every later gateway starts its executions with a 20-second process timeout, and its cycle is timed by that cycle's timeout target. In each cycle: diff --git a/docs/translations.json b/docs/translations.json index 76f540e..116fd71 100644 --- a/docs/translations.json +++ b/docs/translations.json @@ -249,8 +249,8 @@ "회귀", "후보-조합" ], - "source_sha256": "e49e67542a40b866f58e973546541fd3abf5790416db9726fab9e884c586f43f", - "translation_sha256": "01ccea2d97e827ea9d210909ff903b68c5e2bd5d73704a130293e3180ec7632f" + "source_sha256": "90775a4bde26e863cf717e0b0af77f9a492ca76cfc9a721e47b52dc97efcd6c9", + "translation_sha256": "537841d793983da1d7ac7bd3327355cae57e7b87e64d9deb9be3449a3ee8d1e9" }, { "id": "overview", diff --git a/scripts/qualify-devguard.py b/scripts/qualify-devguard.py index 467e735..0f052e1 100644 --- a/scripts/qualify-devguard.py +++ b/scripts/qualify-devguard.py @@ -2376,19 +2376,28 @@ def stopped_checks(bench, runner, label, terminated=None): return checks -def once_reconciled(call, limit=30.0, period=0.05): - """`call()` again while it is refused with RECOVERY_PENDING, for up to `limit` seconds: its - last result, the refusals before it and how long that took (ms).""" +def once_reconciled(call, limit=30.0, period=0.05, codes=('RECOVERY_PENDING',)): + """`call()` again while it is refused with RECOVERY_PENDING (or another of `codes`), for up + to `limit` seconds: its last result, the refusals before it and how long that took (ms).""" began = time.monotonic() pending = 0 while True: result = call() - if result.get('code') != 'RECOVERY_PENDING' or time.monotonic() - began >= limit: + if result.get('code') not in codes or time.monotonic() - began >= limit: return result, pending, round((time.monotonic() - began) * 1000, 3) pending += 1 time.sleep(period) +def once_free(call, limit=30.0): + """`call()`, new work in a workspace, again while the workspace is not free yet: refused with + RECOVERY_PENDING while the gateway reconciles it, or with WORKSPACE_BUSY while the gateway + still holds it for a process it already reports exited. The Runner reports a process exited + once it has reaped it, and the gateway frees its workspace when the Runner's exit event + follows, once the process's output has closed. As once_reconciled.""" + return once_reconciled(call, limit=limit, codes=('RECOVERY_PENDING', 'WORKSPACE_BUSY')) + + def resume_result(gateway, approval, timeout=host.CALL_TIMEOUT_S): """operation_resume of a granted approval as an outcome: started, with its process, refused with CodeSpace's code, or an error of the transport.""" @@ -2837,7 +2846,9 @@ def recovery_writer_lock(bench): 'waiting for it to end' in log and 'the only writer of the operations database' in log and ended['returncode'] == 0, ended=ended), check('it claimed the Runner under epoch 2', bool(claim and claim['epoch'] == 2), claim=claim)] - later, _, _ = ran(g3, g3.exec('a', ['/bin/echo', 'next']), 'next\n', 'the next gateway') + # The first gateway worked in `a`: the next one starts work there once it has reconciled it. + following, _, _ = once_reconciled(lambda: g3.exec('a', ['/bin/echo', 'next'])) + later, _, _ = ran(g3, following, 'next\n', 'the next gateway') return checks + later, {'second': {key: second[key] for key in ('returncode', 'elapsed_s')}} @@ -2878,7 +2889,8 @@ def recovery_schema(bench): claim = claimed_by(runner, g2.pid) checks.append(check('the gateway of the database the Runner knows claimed it under epoch 2', bool(claim and claim['epoch'] == 2), claim=claim)) - later, _, _ = ran(g2, g2.exec('a', ['/bin/echo', 'next']), 'next\n', 'it') + following, _, _ = once_reconciled(lambda: g2.exec('a', ['/bin/echo', 'next'])) + later, _, _ = ran(g2, following, 'next\n', 'it') return checks + later, detail @@ -3245,8 +3257,8 @@ def recovery_reconcile_running(bench, tty): """A gateway killed while its process runs: the next gateway refuses new work in that workspace with RECOVERY_PENDING until it has reconciled it, then holds the workspace for the process again (WORKSPACE_BUSY), while a workspace where no work was left takes new work at - once; terminating the inherited process frees its workspace. No refused command ran - (P1R-C04).""" + once; terminating the inherited process frees its workspace (once_free: as soon as the + Runner's exit event follows its exited status). No refused command ran (P1R-C04).""" runner = bench.runner() checks = registered_checks(bench, runner) g1 = bench.gateway('g1') @@ -3262,7 +3274,7 @@ def recovery_reconcile_running(bench, tty): status = status_of(g2, held['process_id']) g2.terminate(held['process_id']) ended = wait_state(g2, held['process_id'], exited) - fresh, _, _ = once_reconciled(lambda: g2.exec('a', tagged(bench.roots['a'], 'fresh', ['/bin/echo', 'fresh']), tty=tty)) + fresh, _, _ = once_free(lambda: g2.exec('a', tagged(bench.roots['a'], 'fresh', ['/bin/echo', 'fresh']), tty=tty)) fresh_checks, _, _ = ran(g2, fresh, 'fresh\n', 'new work once it was free') checks += [check('a workspace where no work was left took new work at once', elsewhere['outcome'] == 'started', result=elsewhere), @@ -3312,7 +3324,9 @@ def recovery_reconcile_approval(bench): killed is resumed by the next gateway once, and resuming it again returns the same process; one resumed while the gateway was killed, at a range of moments, is resumed by the next gateway with the process the Runner created for it, or started then when the Runner proved - it created none: each command ran exactly once.""" + it created none: each command ran exactly once. Each round's execution is held through the + gateway of the round before once it has reconciled `confirm`: when the Runner had already + started the resumed process, that gateway answers both resumes from its record at once.""" runner = bench.runner() checks = registered_checks(bench, runner) root = bench.roots['confirm'] @@ -3320,7 +3334,7 @@ def recovery_reconcile_approval(bench): rounds = [] for index, delay in enumerate((None,) + KILL_DELAYS_S): tag = f'approval-{index}' - held = gateway.exec('confirm', tagged(root, tag, ['/bin/sh', '-c', 'exit 0'])) + held, _, _ = once_reconciled(lambda: gateway.exec('confirm', tagged(root, tag, ['/bin/sh', '-c', 'exit 0']))) approval = held.get('approval_id') row = {'tag': tag, 'held': held.get('code'), 'approval_id': approval, 'delay_s': delay} rounds.append(row) @@ -3478,17 +3492,20 @@ def recovery_kill_during_exec(bench, tty): def recovery_kill_during_terminate(bench, tty): """Gateways killed while terminate_process is in flight, at a range of moments: the next gateway reports the process terminated, or still running and then terminates it; each process - ends once, terminated, and ran once; its workspace takes new work afterwards (P1R-C04).""" + ends once, terminated, and ran once; its workspace takes new work afterwards, once free + (once_free: the next gateway reconciles it, and frees it when the Runner's exit event follows + the exited status) (P1R-C04).""" runner = bench.runner() checks = registered_checks(bench, runner) gateway = bench.gateway('g1') rounds = [] for index, delay in enumerate(KILL_DELAYS_S): tag = f'held-{index}' - started, _, _ = once_reconciled(lambda: gateway.exec('a', tagged(bench.roots['a'], tag, ['/bin/sleep', '600']), - tty=tty)) + started, refusals, waited = once_free(lambda: gateway.exec('a', tagged(bench.roots['a'], tag, + ['/bin/sleep', '600']), tty=tty)) if started['outcome'] != 'started': - rounds.append({'tag': tag, 'delay_s': delay, 'started': started}) + rounds.append({'tag': tag, 'delay_s': delay, 'started': started, 'start_refusals': refusals, + 'start_waited_ms': waited}) break pid = tag_pid(bench.roots['a'], tag) flight = {} @@ -3511,13 +3528,14 @@ def terminate(process_id=started['process_id'], target=gateway, result=flight): status = wait_state(gateway, started['process_id'], exited) rounds.append({'tag': tag, 'delay_s': delay, 'in_flight': flight, 'terminated_again': again is not None, 'state': status.get('state'), 'termination': status.get('termination'), - 'pid_gone': gone(pid, limit=10) if pid else None}) + 'pid_gone': gone(pid, limit=10) if pid else None, 'start_refusals': refusals, + 'start_waited_ms': waited}) tags = markers(bench.roots['a']) checks += [check('each process ended once, terminated, whichever gateway terminated it', len(rounds) == len(KILL_DELAYS_S) and all(row.get('termination') == 'terminated' and row.get('pid_gone') for row in rounds), rounds=rounds), check('each process ran once', all(tags.count(row['tag']) == 1 for row in rounds), markers=tags)] - fresh, _, _ = once_reconciled(lambda: gateway.exec('a', ['/bin/echo', 'fresh'], tty=tty)) + fresh, _, _ = once_free(lambda: gateway.exec('a', ['/bin/echo', 'fresh'], tty=tty)) fresh_checks, _, _ = ran(gateway, fresh, 'fresh\n', 'new work in its workspace') return checks + fresh_checks, {'rounds': rounds} @@ -3575,7 +3593,8 @@ def recovery_retention(bench): g1 = bench.gateway('g1') ids = [] for index in range(RETENTION_COUNT): - started = g1.exec('off', ['/bin/sh', '-c', f'echo {index}; exit {index % 7}']) + # Each starts once the gateway has freed `off` from the one before (once_free). + started, _, _ = once_free(lambda: g1.exec('off', ['/bin/sh', '-c', f'echo {index}; exit {index % 7}'])) if started['outcome'] != 'started': return [started_checks(f'process {index}', started)], {} ids.append(started['process_id']) @@ -3833,6 +3852,9 @@ def run(args, live, evidence): # Refusals a tagged command meets while capacity, a reconciliation or the workspace's last # process stands in its way: it is tried again under a new tag. RETRIED = ('RESOURCE_UNAVAILABLE', 'RECOVERY_PENDING', 'WORKSPACE_BUSY') +# A sample that starts new work (a termination's target, a refusal) right after a restart, while +# the serving gateway still reconciles its workspace, is sent again for up to this long. +RECONCILE_WAIT_S = 10.0 # The workspace and terminal of each restart's timeout target, in turn. TIMEOUT_KINDS = (('timeout-a', False), ('timeout-off', False), ('timeout-a', True), ('timeout-off', True)) RECOVERY_SAMPLES = analysis.SAMPLE_FILES + ('read_beat', 'read_burst') @@ -3948,10 +3970,11 @@ class Execs: def __init__(self, bench, writer, path): self.bench, self.writer, self.path = bench, writer, path - def run(self, gateway, workspace, tag, command, tty=False, retry=0.0, approval=False): - """exec_command of a tagged command. Refused for capacity, for a reconciliation or for its - workspace's last process, it is sent again under a new tag for up to `retry` seconds; the - result names the tag that was sent last.""" + def run(self, gateway, workspace, tag, command, tty=False, retry=0.0, approval=False, retried=RETRIED): + """exec_command of a tagged command. Refused with one of `retried` (for capacity, for a + reconciliation or for its workspace's last process), it is sent again under a new tag for up + to `retry` seconds; the result names the tag that was sent last and how many were refused + before it.""" deadline = time.monotonic() + retry for attempt in itertools.count(): name = tag if attempt == 0 else f'{tag}-{attempt}' @@ -3959,26 +3982,32 @@ def run(self, gateway, workspace, tag, command, tty=False, retry=0.0, approval=F self.writer.write(self.path, {'tag': name, 'workspace': workspace, 'outcome': result['outcome'], 'code': result.get('code'), 'process_id': result.get('process_id'), 'approval': approval, 'gateway': gateway.pid, 'unix': time.time()}) - if result.get('code') not in RETRIED or time.monotonic() >= deadline: - return dict(result, tag=name) + if result.get('code') not in retried or time.monotonic() >= deadline: + return dict(result, tag=name, refused_before=attempt) time.sleep(0.2) class GatedTerminations: """Each slot starts a fresh target in `term` through the gateway that serves, alternating pipe and PTY, terminates it and times it as Terminations does: a governed target's end by its - scope's process group, an ungoverned one's by its own PID, its release by the watcher.""" + scope's process group, an ungoverned one's by its own PID, its release by the watcher. Right + after a restart the gateway refuses new work in `term` with RECOVERY_PENDING until it has + reconciled it: the target is sent again meanwhile, for up to RECONCILE_WAIT_S, and one still + refused then is an error.""" def __init__(self, bench, execs, watcher): self.bench, self.execs, self.watcher = bench, execs, watcher def __call__(self, gateway, slot): tty = bool(slot % 2) - started = self.execs.run(gateway, 'term', f'term-{slot}', ['/bin/sleep', '600'], tty=tty) + started = self.execs.run(gateway, 'term', f'term-{slot}', ['/bin/sleep', '600'], tty=tty, + retry=RECONCILE_WAIT_S, retried=('RECOVERY_PENDING',)) + pending = started['refused_before'] if started['outcome'] != 'started': code = started.get('code') return {'outcome': 'not_started' if code == 'RESOURCE_UNAVAILABLE' else 'error', 'code': code, - 'attempt_id': started.get('attempt_id'), 'latency_ms': started.get('latency_ms'), 'tty': tty} + 'attempt_id': started.get('attempt_id'), 'latency_ms': started.get('latency_ms'), 'tty': tty, + 'pending_refusals': pending} process_id = started['process_id'] entry = (gateway.execution(process_id) or {}) if self.bench.governed else {} root, attempt = entry.get('scope_root_pid'), entry.get('attempt_id') @@ -3992,15 +4021,19 @@ def __call__(self, gateway, slot): ended = host.wait_group_gone(root, limit=30) is not None if root else bool(pid) and gone(pid, limit=30) return {'outcome': 'acknowledged', 'tty': tty, 'ack_ms': ack, 'termination': (status or {}).get('termination'), 'exited_ms': exited, 'scope_exit_ms': round((time.monotonic() - asked) * 1000, 3) if ended else None, - 'attempt_id': attempt, 'scope_root_pid': root, 'pid': pid} + 'attempt_id': attempt, 'scope_root_pid': root, 'pid': pid, 'pending_refusals': pending} def gated_admission(execs): """A governed execution in `hog`, which the authority must refuse for capacity; a command it - refused must never run (its tag never in the marker file).""" + refused must never run (its tag never in the marker file). As a termination's target, it is + sent again while the serving gateway reconciles `hog` (RECOVERY_PENDING), so that the refusal + timed is the authority's.""" def sample(gateway, slot): - result = execs.run(gateway, 'hog', f'hog-{slot}', ['/usr/bin/true']) - row = {'latency_ms': result.get('latency_ms'), 'code': result.get('code'), 'attempt_id': result.get('attempt_id')} + result = execs.run(gateway, 'hog', f'hog-{slot}', ['/usr/bin/true'], retry=RECONCILE_WAIT_S, + retried=('RECOVERY_PENDING',)) + row = {'latency_ms': result.get('latency_ms'), 'code': result.get('code'), 'attempt_id': result.get('attempt_id'), + 'pending_refusals': result['refused_before']} if result['outcome'] == 'started': row.update(outcome='admitted', attempt_id=attempt_of(gateway, result['process_id'])) gateway.wait_exited(result['process_id'], limit=30) diff --git a/scripts/tests/test_qualify_devguard.py b/scripts/tests/test_qualify_devguard.py index 32b8e15..d18906c 100644 --- a/scripts/tests/test_qualify_devguard.py +++ b/scripts/tests/test_qualify_devguard.py @@ -909,6 +909,24 @@ def test_verify_refuses_an_unsealed_directory(self): self.assertEqual(qualify.verify(self.dir / 'open')['verdict'], 'failed') +class ScriptedGateway: + """A gateway whose exec_command answers in turn from `answers`, the last one again and again; + the processes it starts end at once when terminated.""" + + def __init__(self, answers, pid=4141): + self.answers, self.pid, self.sent = list(answers), pid, [] + + def exec(self, workspace, command, tty=False): + self.sent.append({'workspace': workspace, 'tag': command[3], 'tty': tty}) + return dict(self.answers.pop(0) if len(self.answers) > 1 else self.answers[0]) + + def terminate(self, process_id, timeout=None): + return 12.5 + + def wait_exited(self, process_id, limit=30, period=0.02): + return {'state': 'exited', 'termination': 'terminated'} + + class RecoveryTests(unittest.TestCase): """P1R-C06: the recovery suites' analysis, the Runner's log and wire protocol, the serving gate, the command line and `verify` of a recovery run.""" @@ -1119,6 +1137,65 @@ def slow(gateway, slot): self.assertEqual(serving.close(), 0) self.assertEqual(serving.served_s, served) + def test_new_work_is_sent_again_until_its_workspace_is_free(self): + def answers(*codes): + results = iter([{'outcome': 'refused', 'code': code} for code in codes] + + [{'outcome': 'started', 'process_id': 'proc-1'}]) + return lambda: next(results) + + def busy(): + return {'outcome': 'refused', 'code': 'WORKSPACE_BUSY'} + # once_reconciled waits for the reconciliation only: a workspace still held is its answer. + result, pending, _ = qualify.once_reconciled(answers('RECOVERY_PENDING', 'WORKSPACE_BUSY'), period=0) + self.assertEqual((result['code'], pending), ('WORKSPACE_BUSY', 1)) + # once_free also waits while the gateway still holds the workspace for a process it + # reported exited, until the Runner's exit event frees it. + result, pending, _ = qualify.once_free(answers('RECOVERY_PENDING', 'WORKSPACE_BUSY', 'WORKSPACE_BUSY')) + self.assertEqual((result['outcome'], pending), ('started', 3)) + # Any other refusal is the answer at once, and one that lasts is the answer at the limit. + result, pending, _ = qualify.once_free(answers('RESOURCE_UNAVAILABLE')) + self.assertEqual((result['code'], pending), ('RESOURCE_UNAVAILABLE', 0)) + result, pending, waited = qualify.once_free(busy, limit=0.12) + self.assertEqual(result['code'], 'WORKSPACE_BUSY') + self.assertGreaterEqual(waited, 120) + self.assertGreater(pending, 0) + + def test_samples_wait_for_the_reconciliation_of_their_workspace(self): + rows = [] + bench = SimpleNamespace(roots={'term': self.dir, 'hog': self.dir}, governed=False) + execs = qualify.Execs(bench, SimpleNamespace(write=lambda path, row: rows.append(row)), + self.dir / 'execs.jsonl') + terminations = qualify.GatedTerminations(bench, execs, None) + pending = {'outcome': 'refused', 'code': 'RECOVERY_PENDING'} + with mock.patch.object(qualify, 'tag_pid', return_value=4242), \ + mock.patch.object(qualify, 'gone', return_value=True): + # A target the serving gateway refuses while it reconciles `term` is sent again, under + # a new tag, and timed once it started. + gateway = ScriptedGateway([pending, {'outcome': 'started', 'process_id': 'proc-1'}]) + sample = terminations(gateway, 1) + self.assertEqual((sample['outcome'], sample['termination'], sample['ack_ms'], sample['pending_refusals']), + ('acknowledged', 'terminated', 12.5, 1)) + self.assertEqual([(row['tag'], row['code']) for row in rows], + [('term-1', 'RECOVERY_PENDING'), ('term-1-1', None)]) + self.assertTrue(all(sent['tty'] for sent in gateway.sent)) + # A target refused for capacity is not a sample, and is not sent again. + gateway = ScriptedGateway([{'outcome': 'refused', 'code': 'RESOURCE_UNAVAILABLE'}, pending]) + sample = terminations(gateway, 2) + self.assertEqual((sample['outcome'], sample['pending_refusals'], len(gateway.sent)), ('not_started', 0, 1)) + # One still refused when the wait ends is an error, a missing sample. + gateway = ScriptedGateway([pending]) + with mock.patch.object(qualify, 'RECONCILE_WAIT_S', 0.3): + sample = terminations(gateway, 3) + self.assertEqual((sample['outcome'], sample['code']), ('error', 'RECOVERY_PENDING')) + self.assertEqual(sample['pending_refusals'], len(gateway.sent) - 1) + self.assertGreater(sample['pending_refusals'], 0) + # The refusal timed is the authority's, not the reconciliation's. + gateway = ScriptedGateway([pending, {'outcome': 'refused', 'code': 'RESOURCE_UNAVAILABLE', 'latency_ms': 41.0}]) + row = qualify.gated_admission(execs)(gateway, 4) + self.assertEqual((row['outcome'], row['code'], row['latency_ms'], row['pending_refusals']), + ('refused', 'RESOURCE_UNAVAILABLE', 41.0, 1)) + self.assertEqual([sent['tag'] for sent in gateway.sent], ['hog-4', 'hog-4-1']) + def test_recovery_arguments(self): common = ['--out', 'x', '--mcp-bin', 'm', '--runtime-bin', 'r'] self.assertEqual(qualify.arguments(['recovery-fence', *common, '--participation', 'off']).participation,