From f5fc74770a30562bf040021fc5c9c0663ecd010d Mon Sep 17 00:00:00 2001 From: Brian Searls Date: Sun, 4 Oct 2026 03:23:41 +0000 Subject: [PATCH 1/5] Spark remote legs carry a liveness deadline; a leg that never began is typed dark-or-down undecided Co-Authored-By: Claude Opus 5.5 (1M context) --- dag/extdeps/ssh/client_options.dag | 21 +++- dag/gunbc/fleet/fleet_known_hosts_anchor.dag | 53 +++++++++-- ...lan_leg_hangs_and_reads_as_a_down_host.dag | 22 +++++ ..._leg_reads_as_one_that_ran_and_refused.dag | 2 +- dag/gunbc/spark/fabric_blob_peer.dag | 10 +- dag/gunbc/spark/glm_canary_converge.dag | 8 +- dag/gunbc/spark/host_occupancy_admission.dag | 2 +- .../spark/model_snapshot_materialize.dag | 95 ++++++++++++++++--- dag/gunbc/spark/serving_load_probe.dag | 2 +- dag/gunbc/spark/serving_load_runner.dag | 2 +- dag/gunbc/spark/serving_rank_observe.dag | 2 +- .../spark/serving_relaunch_transaction.dag | 4 +- dag/gunbc/spark/spark_file_blob.dag | 10 +- .../spark/v41_checkpoint_materialize.dag | 16 ++-- .../spark/v41_engram_differential_run.dag | 12 +-- dag/gunbc/spark/v41_group_a_launch.dag | 2 +- dag/gunbc/spark/v41_row_store_encode_run.dag | 10 +- .../spark/v41_row_store_readback_run.dag | 2 +- dag/gunbc/spark/v41_runtime_image_probe.dag | 6 +- dag/gunbc/spark/v41_serving_load.dag | 2 +- dag/gunbc/spark/vllm_runtime_image_build.dag | 6 +- .../spark/serving_load_probe_witness_test.dag | 4 +- ...spark_remote_leg_deadline_witness_test.dag | 95 +++++++++++++++++++ 23 files changed, 311 insertions(+), 77 deletions(-) create mode 100644 dag/gunbc/recurring_failure_mode/a_dark_management_lan_leg_hangs_and_reads_as_a_down_host.dag create mode 100644 dag/test/claim/spark/spark_remote_leg_deadline_witness_test.dag diff --git a/dag/extdeps/ssh/client_options.dag b/dag/extdeps/ssh/client_options.dag index fd56eea0163..2d578d9d691 100644 --- a/dag/extdeps/ssh/client_options.dag +++ b/dag/extdeps/ssh/client_options.dag @@ -1,6 +1,6 @@ module extdeps.ssh.client_options -import std.types { NonEmptyStr, String, Bool, List } +import std.types { NonEmptyStr, String, Bool, Int, List } import std.measure { Second, second, second_count } import std.decl_ref { DeclarationRef, WholeDeclaration } import extdeps.external_authority { ExternalAuthority, ExternalModelScope, ExternalSubjectRef } @@ -99,7 +99,16 @@ type UpdateHostKeysValue | UpdateHostKeysAsk // One variant per keyword, each carrying the value domain THAT keyword admits, so a nonsensical -// pairing has no constructor. ConnectTimeout takes an Int because ssh_config(5) specifies seconds; +// pairing has no constructor. ServerAliveInterval and ServerAliveCountMax are the LIVENESS half of a +// deadline, distinct from ConnectTimeout, which bounds only the TCP connect: ssh_config(5) -- +// ServerAliveInterval "Sets a timeout interval in seconds after which if no data has been received +// from the server, ssh(1) will send a message through the encrypted channel to request a response", +// default 0 (never); ServerAliveCountMax "Sets the number of server alive messages ... which may be +// sent without ssh(1) receiving any messages back from the server. If this threshold is reached while +// server alive messages are being sent, ssh will disconnect from the server, terminating the session", +// default 3. With the interval at its default of 0 an established session whose path goes dark waits +// on TCP alone, which is how fleet-converge run 37156689444 sat ~1h inside one leg. +// ConnectTimeout takes an Int because ssh_config(5) specifies seconds; // BatchMode takes a Bool because its domain is yes|no and Bool is that domain already grounded // (section 2: no net new concepts by re-invention). type SshClientOption @@ -112,6 +121,8 @@ type SshClientOption | BatchMode { enabled: Bool } | ExitOnForwardFailure { enabled: Bool } | ConnectTimeout { seconds: Second } + | ServerAliveInterval { seconds: Second } + | ServerAliveCountMax { count: Int } | ConfigFile { path: NonEmptyStr } | IdentityFile { path: NonEmptyStr } | IdentitiesOnly { enabled: Bool } @@ -134,6 +145,8 @@ fn ssh_client_option_keyword(option: SshClientOption) -> String { BatchMode { enabled: _ } => "BatchMode" ExitOnForwardFailure { enabled: _ } => "ExitOnForwardFailure" ConnectTimeout { seconds: _ } => "ConnectTimeout" + ServerAliveInterval { seconds: _ } => "ServerAliveInterval" + ServerAliveCountMax { count: _ } => "ServerAliveCountMax" ConfigFile { path: _ } => "ConfigFile" IdentityFile { path: _ } => "IdentityFile" IdentitiesOnly { enabled: _ } => "IdentitiesOnly" @@ -182,6 +195,8 @@ fn ssh_client_option_value(option: SshClientOption) -> String { BatchMode { enabled: e } => if e { "yes" } else { "no" } ExitOnForwardFailure { enabled: e } => if e { "yes" } else { "no" } ConnectTimeout { seconds: s } => to_string(second_count(s: s)) + ServerAliveInterval { seconds: s } => to_string(second_count(s: s)) + ServerAliveCountMax { count: n } => to_string(n) ConfigFile { path: p } => p as String IdentityFile { path: p } => p as String IdentitiesOnly { enabled: e } => if e { "yes" } else { "no" } @@ -338,6 +353,8 @@ fn serialize_ssh_config_file_flag(option: SshClientOption) -> List { BatchMode { enabled: _ } => serialize_ssh_client_option(option: option) ExitOnForwardFailure { enabled: _ } => serialize_ssh_client_option(option: option) ConnectTimeout { seconds: _ } => serialize_ssh_client_option(option: option) + ServerAliveInterval { seconds: _ } => serialize_ssh_client_option(option: option) + ServerAliveCountMax { count: _ } => serialize_ssh_client_option(option: option) IdentityFile { path: _ } => serialize_ssh_client_option(option: option) IdentitiesOnly { enabled: _ } => serialize_ssh_client_option(option: option) IdentityAgent { path: _ } => serialize_ssh_client_option(option: option) diff --git a/dag/gunbc/fleet/fleet_known_hosts_anchor.dag b/dag/gunbc/fleet/fleet_known_hosts_anchor.dag index 0cef2730e4b..01eb265502b 100644 --- a/dag/gunbc/fleet/fleet_known_hosts_anchor.dag +++ b/dag/gunbc/fleet/fleet_known_hosts_anchor.dag @@ -1,7 +1,7 @@ module gunbc.fleet_known_hosts_anchor import std.types { String, Bool, List, NonEmptyStr, PathSegment, path_segment_is_safe, brand } -import std.measure { second } +import std.measure { Second, second } import std.algebra { trim } import std.content_hash { ContentHash, @@ -28,6 +28,8 @@ import extdeps.ssh.client_options { User, BatchMode, ConnectTimeout, + ServerAliveInterval, + ServerAliveCountMax, } import gunbc.fleet_host_key_enrollment { KnownHostKeyRow, fleet_host_key_enrollments, known_hosts_line, host_key_provenance_label } import gunbc.spark.fabric_reach { fabric_rail_known_host_rows } @@ -667,12 +669,43 @@ fn fleet_ssh_password_client_options( trust: List, principal: NonEmptyStr, ) -> List { - concat(trust, [ + concat(trust, concat([ PreferredAuthentications { methods: "password,keyboard-interactive" as NonEmptyStr }, User { name: principal }, BatchMode { enabled: false }, - ConnectTimeout { seconds: second(count: 10) }, - ]) + ], fleet_ssh_leg_deadline_options(deadline: fleet_ssh_leg_deadline))) +} + +// EVERY FLEET LEG CARRIES A DEADLINE, AND IT IS ONE FACT. A leg has two ways to stall and ssh bounds +// them with different keywords: the TCP connect (ConnectTimeout) and an established session whose +// path then goes dark (ServerAliveInterval x ServerAliveCountMax, extdeps.ssh.client_options). The +// shapers below carried the first, inline, and never the second -- so on 2026-10-03 when the Spark +// management LAN (wlP9s9, every Spark's only route to the executor) stopped forwarding mid-session, +// fleet-converge run 37156689444 sat ~1h inside one ssh leg instead of refusing. +// +// UNCONSTRUCTIBLE WITHOUT IT, at the grain this module owns: neither shaper takes an options +// parameter, so no caller can shape a fleet leg that omits these words, and both derive them from +// this ONE row rather than each spelling its own number. A dark path now ends the leg with ssh's own +// 255 within interval x count_max (60 s) of the last reply; the sshd answers keepalives itself, so +// a remote command that is merely silent is NOT cut off -- only a path that returns nothing is. +type FleetSshLegDeadline { + connect: Second + alive_interval: Second + alive_count_max: Int +} + +data fleet_ssh_leg_deadline: FleetSshLegDeadline = FleetSshLegDeadline { + connect: second(count: 10), + alive_interval: second(count: 15), + alive_count_max: 4, +} + +fn fleet_ssh_leg_deadline_options(deadline: FleetSshLegDeadline) -> List { + [ + ConnectTimeout { seconds: deadline.connect }, + ServerAliveInterval { seconds: deadline.alive_interval }, + ServerAliveCountMax { count: deadline.alive_count_max }, + ] } // The password session needs only the trust anchor, so a caller holding no fleet key (the first @@ -714,11 +747,13 @@ fn shape_fleet_ssh_exec( fleet_openssh_trust_options(policy: context.trust), concat( fleet_ssh_credential_options(binding: context.credential), - [ - User { name: target.principal }, - BatchMode { enabled: true }, - ConnectTimeout { seconds: second(count: 10) }, - ], + concat( + [ + User { name: target.principal }, + BatchMode { enabled: true }, + ], + fleet_ssh_leg_deadline_options(deadline: fleet_ssh_leg_deadline), + ), ), ) concat( diff --git a/dag/gunbc/recurring_failure_mode/a_dark_management_lan_leg_hangs_and_reads_as_a_down_host.dag b/dag/gunbc/recurring_failure_mode/a_dark_management_lan_leg_hangs_and_reads_as_a_down_host.dag new file mode 100644 index 00000000000..1234ed7db51 --- /dev/null +++ b/dag/gunbc/recurring_failure_mode/a_dark_management_lan_leg_hangs_and_reads_as_a_down_host.dag @@ -0,0 +1,22 @@ +module gunbc.recurring_failure_mode.a_dark_management_lan_leg_hangs_and_reads_as_a_down_host + +import std.types { NonEmptyStr } +import gunbc.recurring_failure_mode { RecurringFailureMode } + +data a_dark_management_lan_leg_hangs_and_reads_as_a_down_host: RecurringFailureMode = RecurringFailureMode { + identity: "a_dark_management_lan_leg_hangs_and_reads_as_a_down_host" as NonEmptyStr, + + receipts: [ + "**a dark management-LAN leg hangs, then reads as a down host** (a fleet ssh leg bounded only at connect waits indefinitely when an established session's path goes dark, and when it does end, the run reports the host as down although the evidence in hand cannot tell a dark LAN leg from a host that is off).", + + "INVALID STATE: gunbc.fleet_known_hosts_anchor shape_fleet_ssh_exec and fleet_ssh_password_client_options carried ConnectTimeout=10 and no ServerAliveInterval (ssh_config(5) default 0, never), so an established session had no liveness bound; and gunbc.spark.model_snapshot_materialize spark_remote_run folded ssh's own 255 into RemoteRan, so a leg that never began read as a command that exited 255.", + + "HARM: 2026-10-03, Spark Group A bring-up. Every Spark reaches the executor only over wlP9s9 (MediaTek mt7925e). Router band steering roamed the hosts; a failed 4-way handshake left NetworkManager DISCONNECTED ('no secrets: No agents were available') on srv8 and Group B spark-3336, and srv7 stayed associated but forwarded nothing 05:33-06:53 EDT (read on the hosts by valiant-crab-775). Fleet-converge run 37156689444 sat ~1h inside one ssh leg. No host rebooted or wedged; the runs called them down.", + + "DISTINGUISHING FACTS: liveness is a different bound from connect (ssh_config(5) ServerAliveInterval x ServerAliveCountMax); sshd answers keepalives itself, so a silent remote command is not cut off, only a path that returns nothing. Whether the leg BEGAN is read in band (spark_remote_leg_began_word on stdout), not from stderr. 'Host down' needs an observation this boundary does not hold, so the type has no HostDown arm: LegNeverBegan is dark-or-down UNDECIDED and names the readback (uptime, wlP9s9 journal) that closes it.", + + "RUNG: found at 1 (a human read the hang and the hosts). Now 2: gunbc.fleet_known_hosts_anchor fleet_ssh_leg_deadline is the one row both fleet shapers derive their deadline words from and neither shaper takes an options parameter, and dag/test/claim/spark/spark_remote_leg_deadline_witness_test.dag pins the argv, the route through the password session, a zeroed-keepalive red control and the never-began reading. CEILING 3: the deadline as a required field of the operation every fleet ssh transport declares, so no ssh leg in the corpus (including extdeps.bmc.openbmc_password_ssh_transport and extdeps.ssh.password_session CopyFile, which still carry ConnectTimeout only) is expressible without it. NEXT-RUNG TRIGGER: those transport literals moved onto extdeps.ssh.client_options with the deadline derived, SUFFICIENT FOR no ssh argv in the corpus lacking a liveness bound.", + ], + + evidence: [], +} diff --git a/dag/gunbc/recurring_failure_mode/an_unreached_effect_leg_reads_as_one_that_ran_and_refused.dag b/dag/gunbc/recurring_failure_mode/an_unreached_effect_leg_reads_as_one_that_ran_and_refused.dag index ab46646ca69..3d9220fe979 100644 --- a/dag/gunbc/recurring_failure_mode/an_unreached_effect_leg_reads_as_one_that_ran_and_refused.dag +++ b/dag/gunbc/recurring_failure_mode/an_unreached_effect_leg_reads_as_one_that_ran_and_refused.dag @@ -17,7 +17,7 @@ data an_unreached_effect_leg_reads_as_one_that_ran_and_refused: RecurringFailure "RESIDUE, STATED: a connection lost after the shell started but before its began-line crossed also reads Unreached. gunbc.spark.native_serving_apply therefore retries only idempotent stages: preflight and the front door read, preserve/commit/rollback are no-ops where this transaction's desired state already holds, and apply's unconditional restart is the module's declared epoch semantics.", - "RUNG: found at 1 (mitigatable -- a human could read the log and delete the files). Now 2 for the native serving arm: witnesses in dag/test/claim/spark/native_serving_resumable_apply_witness_test.dag pin the reading, the bounded retry (spark_native_unreached_leg_attempt_budget) and the leftover decision (spark_native_leftover_decision) with discriminating reds. CEILING 3: an outcome type every remote-effect leg must produce, so no consumer can match a leg result without facing the Unreached arm. NEXT-RUNG TRIGGER: the other SshSessionExecResult consumers that fold 255 into a refusal (gunbc.spark.model_snapshot_materialize spark_remote_run, gunbc.host_command_binding host_command_outcome_of's Unconfirmed arm) consuming one shared in-band leg reading, SUFFICIENT FOR no remote-effect consumer in the corpus to construct a ran-and-refused value from a leg that never began.", + "RUNG: found at 1 (mitigatable -- a human could read the log and delete the files). Now 2 for the native serving arm: witnesses in dag/test/claim/spark/native_serving_resumable_apply_witness_test.dag pin the reading, the bounded retry (spark_native_unreached_leg_attempt_budget) and the leftover decision (spark_native_leftover_decision) with discriminating reds. CEILING 3: an outcome type every remote-effect leg must produce, so no consumer can match a leg result without facing the Unreached arm. NEXT-RUNG TRIGGER: the other SshSessionExecResult consumers that fold 255 into a refusal (gunbc.spark.model_snapshot_materialize spark_remote_run, gunbc.host_command_binding host_command_outcome_of's Unconfirmed arm) consuming one shared in-band leg reading, SUFFICIENT FOR no remote-effect consumer in the corpus to construct a ran-and-refused value from a leg that never began. PROGRESS (2026-10-04): gunbc.spark.model_snapshot_materialize spark_remote_run now reads the in-band began-line (spark_remote_leg_began_word) and returns RemoteUnreachable with reach LegNeverBegan for ssh's 255 with no began-line (see a_dark_management_lan_leg_hangs_and_reads_as_a_down_host); gunbc.host_command_binding host_command_outcome_of remains.", ], evidence: [], diff --git a/dag/gunbc/spark/fabric_blob_peer.dag b/dag/gunbc/spark/fabric_blob_peer.dag index b1590138a5a..f8573cb84f4 100644 --- a/dag/gunbc/spark/fabric_blob_peer.dag +++ b/dag/gunbc/spark/fabric_blob_peer.dag @@ -94,7 +94,7 @@ fn fabric_blob_serve(ctx: SparkMaterializeContext) -> FabricBlobServe { Absent => fabric_blob_serve_refused(ctx: ctx, step: "serve-rail-address", cause: join(["the holder ", ctx.host as String, " has no fabric rail address, so it cannot serve over the rail"], "")) Present { value: rail } => match spark_remote_run(ctx: ctx, argv: ["systemctl", "is-active", fabric_blob_serve_unit as String]) { - RemoteUnreachable { cause: c } => fabric_blob_serve_refused(ctx: ctx, step: "serve-state", cause: c) + RemoteUnreachable { cause: c, reach: _ } => fabric_blob_serve_refused(ctx: ctx, step: "serve-state", cause: c) RemoteRan { exit_code: _, stdout: state, stderr: _ } => if trim(s: state) == "active" { FabricBlobServing { serving: FabricBlobHolder { holder: ctx.host, rail_address: rail as NonEmptyStr } } @@ -109,11 +109,11 @@ fn fabric_blob_serve(ctx: SparkMaterializeContext) -> FabricBlobServe { // the script, so its newlines and `=` reach the file exactly as rsyncd_conf_text rendered them. fn fabric_blob_serve_start(ctx: SparkMaterializeContext, rail: NonEmptyStr) -> FabricBlobServe { match spark_remote_run(ctx: ctx, argv: ["mkdir", "-p", fabric_blob_root as String, fabric_blob_staging_directory as String]) { - RemoteUnreachable { cause: c } => fabric_blob_serve_refused(ctx: ctx, step: "serve-root", cause: c) + RemoteUnreachable { cause: c, reach: _ } => fabric_blob_serve_refused(ctx: ctx, step: "serve-root", cause: c) RemoteRan { exit_code: e, stdout: _, stderr: err } => if e != 0 { fabric_blob_serve_refused(ctx: ctx, step: "serve-root", cause: join(["mkdir exited ", to_string(e), ": ", err], "")) } else { match spark_remote_run(ctx: ctx, argv: ["sh", "-c", "printf '%s' \"$1\" > \"$2\"", "sh", rsyncd_conf_text(module: fabric_blob_rsync_module), fabric_blob_rsyncd_conf_path as String]) { - RemoteUnreachable { cause: c } => fabric_blob_serve_refused(ctx: ctx, step: "serve-config", cause: c) + RemoteUnreachable { cause: c, reach: _ } => fabric_blob_serve_refused(ctx: ctx, step: "serve-config", cause: c) RemoteRan { exit_code: e2, stdout: _, stderr: err2 } => if e2 != 0 { fabric_blob_serve_refused(ctx: ctx, step: "serve-config", cause: join(["writing the daemon configuration exited ", to_string(e2), ": ", err2], "")) } else { let cleared = spark_remote_run(ctx: ctx, argv: ["systemctl", "reset-failed", fabric_blob_serve_unit as String]) @@ -121,7 +121,7 @@ fn fabric_blob_serve_start(ctx: SparkMaterializeContext, rail: NonEmptyStr) -> F ["systemd-run", join(["--unit=", fabric_blob_serve_unit as String], ""), "--description=gunbc fabric blob serve (read-only rsync daemon on the rail)", "--"], rsync_daemon_foreground_argv(address: rail, port: rsync_daemon_default_port, config_path: fabric_blob_rsyncd_conf_path), )) { - RemoteUnreachable { cause: c } => fabric_blob_serve_refused(ctx: ctx, step: "serve-start", cause: c) + RemoteUnreachable { cause: c, reach: _ } => fabric_blob_serve_refused(ctx: ctx, step: "serve-start", cause: c) RemoteRan { exit_code: e3, stdout: _, stderr: err3 } => if e3 != 0 { fabric_blob_serve_refused(ctx: ctx, step: "serve-start", cause: join(["systemd-run exited ", to_string(e3), ": ", err3], "")) } else { FabricBlobServing { serving: FabricBlobHolder { holder: ctx.host, rail_address: rail } } @@ -136,7 +136,7 @@ fn fabric_blob_serve_start(ctx: SparkMaterializeContext, rail: NonEmptyStr) -> F // STOPPING IS BEST-EFFORT AND SAYS SO IN ITS TYPE: a daemon that is already gone is the ordinary case. fn fabric_blob_serve_stop(ctx: SparkMaterializeContext) -> Bool { match spark_remote_run(ctx: ctx, argv: ["systemctl", "stop", fabric_blob_serve_unit as String]) { - RemoteUnreachable { cause: _ } => false + RemoteUnreachable { cause: _, reach: _ } => false RemoteRan { exit_code: e, stdout: _, stderr: _ } => e == 0 } } diff --git a/dag/gunbc/spark/glm_canary_converge.dag b/dag/gunbc/spark/glm_canary_converge.dag index bd95df8f7dd..8b01ef000c1 100644 --- a/dag/gunbc/spark/glm_canary_converge.dag +++ b/dag/gunbc/spark/glm_canary_converge.dag @@ -582,7 +582,7 @@ fn glm_canary_observe_serve(ctx: SparkMaterializeContext, arm: GlmCanaryArm) -> Absent => ServeUnreadable { cause: "this arm renders no launch line, so there is nothing to compare the running container against" } Present { value: argv } => match spark_remote_run(ctx: ctx, argv: glm_canary_inspect_argv(arm: arm)) { - RemoteUnreachable { cause: c } => ServeUnreadable { cause: c } + RemoteUnreachable { cause: c, reach: _ } => ServeUnreadable { cause: c } RemoteRan { exit_code: e, stdout: out, stderr: err } => if e != 0 { ServeNotRunning { detail: join(["docker inspect exited ", to_string(e), " for container ", arm.container_name as String, ": ", err], "") } @@ -691,7 +691,7 @@ fn glm_canary_cutover(ctx: SparkMaterializeContext, arm: GlmCanaryArm, agreement fn glm_canary_replace_container(ctx: SparkMaterializeContext, arm: GlmCanaryArm, run: List) -> CanaryCutover { match spark_remote_run(ctx: ctx, argv: glm_canary_inspect_argv(arm: arm)) { - RemoteUnreachable { cause: c } => CutoverRefusedBeforeActing { cause: join(["the rollback line could not be read, so nothing was touched: ", c], "") } + RemoteUnreachable { cause: c, reach: _ } => CutoverRefusedBeforeActing { cause: join(["the rollback line could not be read, so nothing was touched: ", c], "") } RemoteRan { exit_code: e, stdout: rollback, stderr: err } => if e != 0 { CutoverRefusedBeforeActing { cause: join(["the rollback line could not be read, so nothing was touched: ", err], "") } @@ -704,7 +704,7 @@ fn glm_canary_replace_container(ctx: SparkMaterializeContext, arm: GlmCanaryArm, fn glm_canary_stop_and_start(ctx: SparkMaterializeContext, arm: GlmCanaryArm, run: List, rollback: String) -> CanaryCutover { let removed = spark_remote_run(ctx: ctx, argv: ["docker", "rm", "-f", arm.container_name as String]) match spark_remote_run(ctx: ctx, argv: run) { - RemoteUnreachable { cause: c } => CutoverStartedButNotServing { rollback: rollback, detail: join(["the replacement could not be started: ", c], "") } + RemoteUnreachable { cause: c, reach: _ } => CutoverStartedButNotServing { rollback: rollback, detail: join(["the replacement could not be started: ", c], "") } RemoteRan { exit_code: e, stdout: _, stderr: err } => if e != 0 { CutoverStartedButNotServing { rollback: rollback, detail: join(["docker run exited ", to_string(e), ": ", err], "") } @@ -728,7 +728,7 @@ fn glm_canary_health_argv(arm: GlmCanaryArm) -> List { fn glm_canary_await_serving(ctx: SparkMaterializeContext, arm: GlmCanaryArm, rollback: String, remaining: Nat) -> CanaryCutover { match spark_remote_run(ctx: ctx, argv: glm_canary_health_argv(arm: arm)) { - RemoteUnreachable { cause: c } => + RemoteUnreachable { cause: c, reach: _ } => if remaining == 0 { CutoverStartedButNotServing { rollback: rollback, detail: join(["the health probe never reached the host: ", c], "") } } else { diff --git a/dag/gunbc/spark/host_occupancy_admission.dag b/dag/gunbc/spark/host_occupancy_admission.dag index def3229e14e..dbcda9d75a1 100644 --- a/dag/gunbc/spark/host_occupancy_admission.dag +++ b/dag/gunbc/spark/host_occupancy_admission.dag @@ -38,7 +38,7 @@ fn spark_serving_units() -> List { fn spark_ctx_argv_run(ctx: SparkMaterializeContext) -> fn(List) -> ArgvRun { fn(argv) { match spark_remote_run(ctx: ctx, argv: argv) { - RemoteUnreachable { cause: c } => ArgvLegDidNotRun { cause: c } + RemoteUnreachable { cause: c, reach: _ } => ArgvLegDidNotRun { cause: c } RemoteRan { exit_code: code, stdout: out, stderr: err } => ArgvRan { exit_code: code, stdout: out, stderr: err } } } diff --git a/dag/gunbc/spark/model_snapshot_materialize.dag b/dag/gunbc/spark/model_snapshot_materialize.dag index 25a3640fc75..76d2cb1f399 100644 --- a/dag/gunbc/spark/model_snapshot_materialize.dag +++ b/dag/gunbc/spark/model_snapshot_materialize.dag @@ -13,6 +13,7 @@ import gunbc.fleet_bootstrap_principal_session { FleetPrincipalLegRefused, } import extdeps.exec.command { shell_quote } +import extdeps.ssh.session { ssh_client_error_exit_status } import gunbc.fleet_known_hosts_anchor { FleetOpenSshTrustPolicy, SshTarget, portable_remote_words, } @@ -60,7 +61,71 @@ type SparkMaterializeContext { type SparkRemoteRun = RemoteRan { exit_code: Int, stdout: String, stderr: String } - | RemoteUnreachable { cause: String } + | RemoteUnreachable { cause: String, reach: SparkLegUnreachability } + +// WHY A LEG DID NOT RUN, TYPED, AND THE ARM THAT IS DELIBERATELY ABSENT. Every Spark reaches the +// executor only over its management Wi-Fi (wlP9s9). On 2026-10-03 that link went dark on srv8 and +// Group B spark-3336 (NetworkManager left it DISCONNECTED after a failed 4-way handshake) and srv7 +// stayed associated but forwarded nothing 05:33-06:53 EDT; none of the hosts rebooted or wedged, yet +// the runs reported them as down. From ONE ssh leg the executor cannot tell a dark LAN leg from a host +// that is off: both are a connect that times out or a session that stops answering. So there is no +// HostDown arm here -- this boundary holds no observation that grounds it, and minting one is the +// over-assertion DESIGN §4d forbids. What it CAN say, and now says by name: +// +// LegNotAttempted -- nothing was sent: no endpoint, an inexpressible command, a refused password +// prerequisite. A fact about this run, not about the host or the LAN. +// LegNeverBegan -- ssh exited with its own 255 and the remote shell never reported beginning +// (spark_remote_leg_began_word). The LAN leg is dark OR the host is down, +// UNDECIDED; the discriminator that closes it is a readback once the host is +// reachable again (uptime / the NetworkManager journal for wlP9s9), and the +// 2026-10-03 incident closed it as LAN-dark every time. +// +// A leg that BEGAN and then lost its session is not here: it ran to an unknown point, so it stays +// RemoteRan with ssh's 255 -- retrying it as if nothing ran is the opposite, non-idempotent decision +// (gunbc.recurring_failure_mode an_unreached_effect_leg_reads_as_one_that_ran_and_refused). +type SparkLegUnreachability + = LegNotAttempted { cause: String } + | LegNeverBegan { endpoint: String, stderr: String } + +fn spark_leg_unreachability_text(u: SparkLegUnreachability) -> String { + match u { + LegNotAttempted { cause: c } => c + LegNeverBegan { endpoint: e, stderr: err } => + join([ + "the ssh leg to ", e, " never began (ssh exit 255, no remote began-line): management LAN leg dark ", + "or host down, UNDECIDED from this leg -- read the host's uptime and wlP9s9 journal once reachable; stderr=", + err, + ], "") + } +} + +fn spark_remote_unreachable(u: SparkLegUnreachability) -> SparkRemoteRun { + RemoteUnreachable { cause: spark_leg_unreachability_text(u: u), reach: u } +} + +// THE IN-BAND BEGAN-LINE, the discriminator gunbc.command_runner ssh_exit_255_conflation_dissolution +// names instead of stderr-sniffing: the remote shell prints it before the command, on stdout, and it +// is stripped before any consumer reads stdout, so consumers see exactly the command's output. +data spark_remote_leg_began_word: String = "gunbc-spark-remote-leg-began" + +fn spark_remote_leg_reported_began(stdout: String) -> Bool { + length(filter(split(s: stdout, delimiter: "\n"), l => trim(s: l) == spark_remote_leg_began_word)) > 0 +} + +fn spark_remote_leg_command_stdout(stdout: String) -> String { + trim(s: join(filter(split(s: stdout, delimiter: "\n"), l => trim(s: l) != spark_remote_leg_began_word), "\n")) +} + +// THE READING OF ONE LANDED SESSION RESULT, pure so a witness can supply the result rather than a +// host. 255 with no began-line is the only case read as unreached; 255 WITH a began-line is a +// command that ran (or a session lost after it began) and is reported as ran. +fn spark_remote_run_of_session(endpoint: String, exit_code: Int, stdout: String, stderr: String) -> SparkRemoteRun { + if exit_code == ssh_client_error_exit_status && !spark_remote_leg_reported_began(stdout: stdout) { + spark_remote_unreachable(u: LegNeverBegan { endpoint: endpoint, stderr: trim(s: stderr) }) + } else { + RemoteRan { exit_code: exit_code, stdout: spark_remote_leg_command_stdout(stdout: stdout), stderr: trim(s: stderr) } + } +} // EVERY LEG IS PRIVILEGED, BECAUSE THE DAEMON IT MUST REACH IS. Acquisition drives docker on the // target, and docker's socket is root-owned on these hosts -- an unprivileged run answers @@ -78,10 +143,10 @@ type SparkRemoteRun // so this module and the grant installer can no longer disagree about how briansrls is reached. fn spark_remote_run(ctx: SparkMaterializeContext, argv: List) -> SparkRemoteRun { match portable_remote_words(raws: ["sudo", "-S", "sh", "-s"]) { - Absent => RemoteUnreachable { cause: "the remote invocation is not expressible as portable remote words" } + Absent => spark_remote_unreachable(u: LegNotAttempted { cause: "the remote invocation is not expressible as portable remote words" }) Present { value: words } => match spark_reach_endpoint(host: ctx.host, path: ctx.path) { - Absent => RemoteUnreachable { cause: join(["no ", spark_reach_path_wire(p: ctx.path), " endpoint for ", ctx.host as String], "") } + Absent => spark_remote_unreachable(u: LegNotAttempted { cause: join(["no ", spark_reach_path_wire(p: ctx.path), " endpoint for ", ctx.host as String], "") }) Present { value: endpoint } => match exec_as_fleet_principal_trusting_with_stdin( trust: ctx.trust, @@ -90,12 +155,12 @@ fn spark_remote_run(ctx: SparkMaterializeContext, argv: List) -> SparkRe principal: fleet_bootstrap_principal_login(), }, words: words, - stdin_payload: join([ctx.admin_credential as String, "\n", shell_command_line(argv: argv), "\n"], ""), + stdin_payload: join([ctx.admin_credential as String, "\n", "echo ", spark_remote_leg_began_word, "\n", shell_command_line(argv: argv), "\n"], ""), credential: BootstrapCredentialFileHeld { path: ctx.admin_credential_file }, ) { - FleetPrincipalLegRefused { cause: why } => RemoteUnreachable { cause: why as String } + FleetPrincipalLegRefused { cause: why } => spark_remote_unreachable(u: LegNotAttempted { cause: why as String }) FleetPrincipalLegRan { session: _, result: r } => - RemoteRan { exit_code: r.exit_code, stdout: trim(s: r.stdout), stderr: trim(s: r.stderr) } + spark_remote_run_of_session(endpoint: endpoint as String, exit_code: r.exit_code, stdout: r.stdout, stderr: r.stderr) } } } @@ -312,7 +377,7 @@ fn read_instrument_failed_claim(exit_code: Int) -> InstrumentReading { // engine launch on a partial checkpoint. fn spark_snapshot_observe(ctx: SparkMaterializeContext, verify: List, subject: AcquisitionSubject) -> AcquisitionObservation { match spark_remote_run(ctx: ctx, argv: ["test", "-d", subject.install_path as String]) { - RemoteUnreachable { cause: c } => AcquisitionProbeFailed { reason: c as NonEmptyStr } + RemoteUnreachable { cause: c, reach: _ } => AcquisitionProbeFailed { reason: c as NonEmptyStr } RemoteRan { exit_code: present, stdout: _, stderr: _ } => if present != 0 { AcquisitionAbsent @@ -350,7 +415,7 @@ fn spark_snapshot_observe(ctx: SparkMaterializeContext, verify: List, su // distinction, at which point this arm may narrow to the digest claim it currently declines to make. fn spark_snapshot_verify_observation(ctx: SparkMaterializeContext, verify: List) -> AcquisitionObservation { match spark_remote_run(ctx: ctx, argv: executor_wrap(executor: ctx.executor, argv: verify)) { - RemoteUnreachable { cause: c } => AcquisitionProbeFailed { reason: c as NonEmptyStr } + RemoteUnreachable { cause: c, reach: _ } => AcquisitionProbeFailed { reason: c as NonEmptyStr } RemoteRan { exit_code: code, stdout: out, stderr: err } => match read_instrument(exit_code: code, stdout: out) { InstrumentDidNotRun => @@ -366,7 +431,7 @@ fn spark_snapshot_verify_observation(ctx: SparkMaterializeContext, verify: List< fn spark_snapshot_ensure_parent(ctx: SparkMaterializeContext, subject: AcquisitionSubject) -> Bool { match spark_remote_run(ctx: ctx, argv: ["mkdir", "-p", subject.parent_directory as String]) { - RemoteUnreachable { cause: _ } => false + RemoteUnreachable { cause: _, reach: _ } => false RemoteRan { exit_code: e, stdout: _, stderr: _ } => e == 0 } } @@ -466,7 +531,7 @@ type SparkTransferState fn spark_transfer_state(ctx: SparkMaterializeContext, subject: AcquisitionSubject) -> SparkTransferState { match spark_remote_run(ctx: ctx, argv: ["systemctl", "show", spark_transfer_unit_for(subject: subject) as String, "--property=ActiveState", "--value"]) { - RemoteUnreachable { cause: c } => TransferStateUnreadable { cause: c } + RemoteUnreachable { cause: c, reach: _ } => TransferStateUnreadable { cause: c } RemoteRan { exit_code: _, stdout: state, stderr: _ } => if state == "active" || state == "activating" || state == "deactivating" { TransferRunning @@ -474,7 +539,7 @@ fn spark_transfer_state(ctx: SparkMaterializeContext, subject: AcquisitionSubjec TransferAbsent } else { match spark_remote_run(ctx: ctx, argv: ["systemctl", "show", spark_transfer_unit_for(subject: subject) as String, "--property=ExecMainStatus", "--value"]) { - RemoteUnreachable { cause: c } => TransferStateUnreadable { cause: c } + RemoteUnreachable { cause: c, reach: _ } => TransferStateUnreadable { cause: c } RemoteRan { exit_code: _, stdout: code, stderr: _ } => if trim(s: code) == "0" { TransferExitedClean } else { TransferExitedFailed { status: trim(s: code) } } } @@ -494,7 +559,7 @@ fn spark_transfer_state(ctx: SparkMaterializeContext, subject: AcquisitionSubjec // place a reader will believe. fn spark_transfer_clear(ctx: SparkMaterializeContext, subject: AcquisitionSubject) -> Bool { match spark_remote_run(ctx: ctx, argv: ["systemctl", "reset-failed", spark_transfer_unit_for(subject: subject) as String]) { - RemoteUnreachable { cause: _ } => false + RemoteUnreachable { cause: _, reach: _ } => false RemoteRan { exit_code: _, stdout: _, stderr: _ } => true } } @@ -544,7 +609,7 @@ fn spark_transfer_start_argv(ctx: SparkMaterializeContext, subject: AcquisitionS ], executor_wrap(executor: ctx.executor, argv: argv), )) { - RemoteUnreachable { cause: c } => TransferStartUnreachable { cause: c as NonEmptyStr } + RemoteUnreachable { cause: c, reach: _ } => TransferStartUnreachable { cause: c as NonEmptyStr } RemoteRan { exit_code: e, stdout: _, stderr: err } => if e == 0 { TransferStarted @@ -664,7 +729,7 @@ fn spark_snapshot_after_transfer(ctx: SparkMaterializeContext, verify: List, subject: AcquisitionSubject, origin: AcquisitionSource) -> AcquisitionAttempt { match spark_remote_run(ctx: ctx, argv: executor_wrap(executor: ctx.executor, argv: verify)) { - RemoteUnreachable { cause: c } => + RemoteUnreachable { cause: c, reach: _ } => AttemptTerminal { outcome: AcquisitionRefused { cause: VerifyMechanismFailed { identity: subject.identity, source: origin.identity, reason: c as NonEmptyStr, } } } @@ -773,7 +838,7 @@ fn spark_materialize_seed( // host_effect_recover_over and its release observes with the privilege its effect used. fn spark_remote_argv_run(ctx: SparkMaterializeContext, argv: List) -> ArgvRun { match spark_remote_run(ctx: ctx, argv: argv) { - RemoteUnreachable { cause: why } => ArgvLegDidNotRun { cause: why as String } + RemoteUnreachable { cause: why, reach: _ } => ArgvLegDidNotRun { cause: why as String } RemoteRan { exit_code: c, stdout: o, stderr: e } => ArgvRan { exit_code: c, stdout: o, stderr: e } } } diff --git a/dag/gunbc/spark/serving_load_probe.dag b/dag/gunbc/spark/serving_load_probe.dag index 9110ab13c1a..375f4fd9fb1 100644 --- a/dag/gunbc/spark/serving_load_probe.dag +++ b/dag/gunbc/spark/serving_load_probe.dag @@ -746,7 +746,7 @@ type BenchServeExecution fn bench_serve_from_remote(protocol: VllmBenchServeRequest, remote: SparkRemoteRun) -> BenchServeExecution { match remote { - RemoteUnreachable { cause: c } => + RemoteUnreachable { cause: c, reach: _ } => BenchServeUnreachable { protocol: protocol, cause: c as NonEmptyStr } RemoteRan { exit_code: code, stdout: out, stderr: err } => BenchServeRan { protocol: protocol, exit_code: code, stdout: out, stderr: err } diff --git a/dag/gunbc/spark/serving_load_runner.dag b/dag/gunbc/spark/serving_load_runner.dag index e2a2df112b7..76e85970ef3 100644 --- a/dag/gunbc/spark/serving_load_runner.dag +++ b/dag/gunbc/spark/serving_load_runner.dag @@ -116,7 +116,7 @@ fn bench_exec_argv(subject: ServingLoadSubject, req: VllmBenchServeRequest) -> L fn scrape_metrics(ctx: SparkMaterializeContext, subject: ServingLoadSubject, step: String) -> MetricsScrape { match spark_remote_run(ctx: ctx, argv: curl_metrics_argv(subject: subject)) { - RemoteUnreachable { cause: c } => ScrapeUnreachable { step: step, cause: c } + RemoteUnreachable { cause: c, reach: _ } => ScrapeUnreachable { step: step, cause: c } RemoteRan { exit_code: code, stdout: out, stderr: err } => if code != 0 { ScrapeExitedNonZero { step: step, exit_code: code, stdout: out, stderr: err } diff --git a/dag/gunbc/spark/serving_rank_observe.dag b/dag/gunbc/spark/serving_rank_observe.dag index fe2d9abacce..b53d6bc82de 100644 --- a/dag/gunbc/spark/serving_rank_observe.dag +++ b/dag/gunbc/spark/serving_rank_observe.dag @@ -164,7 +164,7 @@ fn rank_leg( ) -> RankLegOutcome { match spark_remote_run(ctx: ctx, argv: argv_words(command: command)) { - RemoteUnreachable { cause: why } => + RemoteUnreachable { cause: why, reach: _ } => RankLegFailed { refusal: rank_observation_refusal(subject: subject, step: step, cause: why as NonEmptyStr), stderr: "" } RemoteRan { exit_code: code, stdout: out, stderr: err } => if code != 0 { diff --git a/dag/gunbc/spark/serving_relaunch_transaction.dag b/dag/gunbc/spark/serving_relaunch_transaction.dag index d57bc535a48..c3557b59674 100644 --- a/dag/gunbc/spark/serving_relaunch_transaction.dag +++ b/dag/gunbc/spark/serving_relaunch_transaction.dag @@ -941,7 +941,7 @@ fn relaunch_leg( ) -> NonEmptyStr? { match spark_remote_run(ctx: ctx, argv: argv_words(command: command)) { - RemoteUnreachable { cause: why } => + RemoteUnreachable { cause: why, reach: _ } => Present { value: join([relaunch_step_wire(s: step) as String, ": ", why], "") as NonEmptyStr } RemoteRan { exit_code: code, stdout: _, stderr: err } => if code != 0 { @@ -1565,7 +1565,7 @@ fn await_after_interval(c: ArmCapture, remaining: Nat) -> RankTransportObservati ) } SleepArgv { words: words } => match spark_remote_run(ctx: c.rank.ctx, argv: words) { - RemoteUnreachable { cause: why } => RankTransportRefused { refusal: rank_observation_refusal( + RemoteUnreachable { cause: why, reach: _ } => RankTransportRefused { refusal: rank_observation_refusal( subject: c.rank.subject, step: relaunch_step_wire(s: StepAwaitAnnouncement), cause: join(["the wait between announcement polls could not be taken: ", why], "") as NonEmptyStr, diff --git a/dag/gunbc/spark/spark_file_blob.dag b/dag/gunbc/spark/spark_file_blob.dag index c902cd925f5..fc103887326 100644 --- a/dag/gunbc/spark/spark_file_blob.dag +++ b/dag/gunbc/spark/spark_file_blob.dag @@ -64,7 +64,7 @@ type SparkFileDigestReading fn spark_read_file_digest(ctx: SparkMaterializeContext, path: NonEmptyStr) -> SparkFileDigestReading { match spark_remote_run(ctx: ctx, argv: sha256sum_argv(path: path as String)) { - RemoteUnreachable { cause: c } => SparkFileDigestUnread { cause: join(["the digest leg did not run: ", c], "") as NonEmptyStr } + RemoteUnreachable { cause: c, reach: _ } => SparkFileDigestUnread { cause: join(["the digest leg did not run: ", c], "") as NonEmptyStr } RemoteRan { exit_code: code, stdout: out, stderr: err } => if code != 0 { SparkFileDigestUnread { cause: join(["sha256sum exited ", to_string(code), ": ", trim(s: err)], "") as NonEmptyStr } } else { match sha256sum_line_digest(line: out) { @@ -96,7 +96,7 @@ fn spark_file_blob_observe(ctx: SparkMaterializeContext, subject: AcquisitionSub Absent => AcquisitionProbeFailed { reason: join(["the subject ", subject.identity as String, " is not keyed by a sha256, so a single file cannot be checked against it"], "") as NonEmptyStr } Present { value: _ } => match spark_remote_run(ctx: ctx, argv: ["test", "-e", subject.install_path as String]) { - RemoteUnreachable { cause: c } => AcquisitionProbeFailed { reason: c as NonEmptyStr } + RemoteUnreachable { cause: c, reach: _ } => AcquisitionProbeFailed { reason: c as NonEmptyStr } RemoteRan { exit_code: present, stdout: _, stderr: err } => if present == 1 { AcquisitionAbsent } else if present != 0 { AcquisitionProbeFailed { reason: join(["the presence test exited ", to_string(present), ": ", trim(s: err)], "") as NonEmptyStr } @@ -113,7 +113,7 @@ fn spark_file_blob_observe(ctx: SparkMaterializeContext, subject: AcquisitionSub fn spark_file_blob_ensure_parent(ctx: SparkMaterializeContext, policy: SparkFileBlobPolicy, subject: AcquisitionSubject) -> Bool { match spark_remote_run(ctx: ctx, argv: ["mkdir", "-p", subject.parent_directory as String, policy.staging_directory as String]) { - RemoteUnreachable { cause: _ } => false + RemoteUnreachable { cause: _, reach: _ } => false RemoteRan { exit_code: e, stdout: _, stderr: _ } => e == 0 } } @@ -163,7 +163,7 @@ fn spark_file_blob_adopt_or_start(ctx: SparkMaterializeContext, policy: SparkFil Absent => AttemptRejected { detail: join(["the subject ", subject.identity as String, " names no staging path, and this handler publishes only staged bytes"], "") as NonEmptyStr } Present { value: staged } => match spark_remote_run(ctx: ctx, argv: ["test", "-f", staged as String]) { - RemoteUnreachable { cause: _ } => AttemptUnreachable + RemoteUnreachable { cause: _, reach: _ } => AttemptUnreachable RemoteRan { exit_code: e, stdout: _, stderr: _ } => if e != 0 { spark_file_blob_start(ctx: ctx, policy: policy, subject: subject, origin: origin) } else { match spark_read_file_digest(ctx: ctx, path: staged) { @@ -240,7 +240,7 @@ fn spark_file_blob_publish(ctx: SparkMaterializeContext, subject: AcquisitionSub } else { AttemptTerminal { outcome: AcquisitionRefused { cause: PublishFailed { identity: subject.identity, source: origin.identity, install_path: subject.install_path } } } } - RemoteUnreachable { cause: _ } => + RemoteUnreachable { cause: _, reach: _ } => AttemptTerminal { outcome: AcquisitionRefused { cause: PublishFailed { identity: subject.identity, source: origin.identity, install_path: subject.install_path } } } } } diff --git a/dag/gunbc/spark/v41_checkpoint_materialize.dag b/dag/gunbc/spark/v41_checkpoint_materialize.dag index 022b508be77..88234cac09d 100644 --- a/dag/gunbc/spark/v41_checkpoint_materialize.dag +++ b/dag/gunbc/spark/v41_checkpoint_materialize.dag @@ -214,7 +214,7 @@ data v41_checkpoint_blob_policy: SparkFileBlobPolicy = SparkFileBlobPolicy { fn v41_attempt(ctx: SparkMaterializeContext, subject: AcquisitionSubject, origin: AcquisitionSource) -> AcquisitionAttempt { if (origin.identity as String) == (v41_hub_cache_source_identity as String) { match spark_remote_run(ctx: ctx, argv: ["test", "-f", v41_hub_cache_copy_path(path: subject.identity) as String]) { - RemoteUnreachable { cause: _ } => AttemptUnreachable + RemoteUnreachable { cause: _, reach: _ } => AttemptUnreachable RemoteRan { exit_code: e, stdout: _, stderr: _ } => if e != 0 { AttemptUnreachable } else { spark_file_blob_attempt(ctx: ctx, policy: v41_checkpoint_blob_policy, subject: subject, origin: origin) } } @@ -286,18 +286,18 @@ fn v41_required_bytes(files: List, listing: List) -> V41DiskReading { let dir = v41_checkpoint_directory as String match spark_remote_run(ctx: ctx, argv: ["test", "-d", dir]) { - RemoteUnreachable { cause: c } => V41DiskUnread { cause: join(["the directory probe did not run: ", c], "") as NonEmptyStr } + RemoteUnreachable { cause: c, reach: _ } => V41DiskUnread { cause: join(["the directory probe did not run: ", c], "") as NonEmptyStr } RemoteRan { exit_code: present, stdout: _, stderr: _ } => if present != 0 && present != 1 { V41DiskUnread { cause: join(["the directory probe exited ", to_string(present)], "") as NonEmptyStr } } else { match v41_list_directory(ctx: ctx, present: present == 0) { Absent => V41DiskUnread { cause: "the directory listing did not answer" as NonEmptyStr } Present { value: listing } => match spark_remote_run(ctx: ctx, argv: ["mkdir", "-p", dir]) { - RemoteUnreachable { cause: c } => V41DiskUnread { cause: join(["the directory could not be created for the free-space read: ", c], "") as NonEmptyStr } + RemoteUnreachable { cause: c, reach: _ } => V41DiskUnread { cause: join(["the directory could not be created for the free-space read: ", c], "") as NonEmptyStr } RemoteRan { exit_code: m, stdout: _, stderr: merr } => if m != 0 { V41DiskUnread { cause: join(["mkdir exited ", to_string(m), ": ", trim(s: merr)], "") as NonEmptyStr } } else { match spark_remote_run(ctx: ctx, argv: host_disk_free_space_argv(path: dir)) { - RemoteUnreachable { cause: c } => V41DiskUnread { cause: join(["the free-space read did not run: ", c], "") as NonEmptyStr } + RemoteUnreachable { cause: c, reach: _ } => V41DiskUnread { cause: join(["the free-space read did not run: ", c], "") as NonEmptyStr } RemoteRan { exit_code: e2, stdout: avail, stderr: err2 } => match host_disk_observation_of_df(exit_code: e2, stdout: avail, stderr: err2, observed_on: "spark_v41_checkpoint_materialize" as NonEmptyStr) { DiskObservationRefused { detail: d } => V41DiskUnread { cause: d } @@ -319,7 +319,7 @@ fn v41_read_disk(ctx: SparkMaterializeContext, files: List String? { if !present { Present { value: "" } } else { match spark_remote_run(ctx: ctx, argv: find_regular_file_sizes_argv(root: v41_checkpoint_directory as String)) { - RemoteUnreachable { cause: _ } => none + RemoteUnreachable { cause: _, reach: _ } => none RemoteRan { exit_code: e, stdout: out, stderr: _ } => if e != 0 { none } else { Present { value: out } } } } @@ -349,7 +349,7 @@ type V41SpanReadOutcome fn v41_read_span(ctx: SparkMaterializeContext, sp: V41PublishedSpan) -> V41SpanReadOutcome { match spark_remote_run(ctx: ctx, argv: v41_span_read_argv(sp: sp)) { - RemoteUnreachable { cause: c } => V41SpanUnread { span: sp, cause: c as NonEmptyStr } + RemoteUnreachable { cause: c, reach: _ } => V41SpanUnread { span: sp, cause: c as NonEmptyStr } RemoteRan { exit_code: code, stdout: out, stderr: err } => if code != 0 { V41SpanUnread { span: sp, cause: join(["the span read exited ", to_string(code), ": ", trim(s: err)], "") as NonEmptyStr } } else { match sha256sum_line_digest(line: out) { @@ -474,14 +474,14 @@ fn v41_index_selection_of_verified(ctx: SparkMaterializeContext, path: NonEmptyS SparkFileDigestUnread { cause: c } => V41IndexSelectionUnread { cause: c } SparkFileDigestRead { hex: h } => match spark_remote_run(ctx: ctx, argv: ["wc", "--bytes", path as String]) { - RemoteUnreachable { cause: c } => V41IndexSelectionUnread { cause: join(["the size read did not run: ", c], "") as NonEmptyStr } + RemoteUnreachable { cause: c, reach: _ } => V41IndexSelectionUnread { cause: join(["the size read did not run: ", c], "") as NonEmptyStr } RemoteRan { exit_code: e, stdout: out, stderr: _ } => if e != 0 { V41IndexSelectionUnread { cause: join(["wc exited ", to_string(e)], "") as NonEmptyStr } } else { match v41_wc_bytes(out: out) { Absent => V41IndexSelectionUnread { cause: join(["wc answered no byte count: ", out], "") as NonEmptyStr } Present { value: b } => match spark_remote_run(ctx: ctx, argv: ["cat", "--", path as String]) { - RemoteUnreachable { cause: c } => V41IndexSelectionUnread { cause: join(["the index read did not run: ", c], "") as NonEmptyStr } + RemoteUnreachable { cause: c, reach: _ } => V41IndexSelectionUnread { cause: join(["the index read did not run: ", c], "") as NonEmptyStr } RemoteRan { exit_code: e2, stdout: text, stderr: _ } => if e2 != 0 { V41IndexSelectionUnread { cause: join(["cat exited ", to_string(e2)], "") as NonEmptyStr } } else { V41IndexSelectionRead { sha256_hex: h, bytes: b, parse: v41_select_index_document(text: text, selector: v41_index_engram_selector) } diff --git a/dag/gunbc/spark/v41_engram_differential_run.dag b/dag/gunbc/spark/v41_engram_differential_run.dag index f70636ab28d..94a477bda30 100644 --- a/dag/gunbc/spark/v41_engram_differential_run.dag +++ b/dag/gunbc/spark/v41_engram_differential_run.dag @@ -235,11 +235,11 @@ fn v41_judge_occupancy(meminfo: String, compute_csv: String) -> V41HostOccupancy fn v41_read_occupancy(ctx: SparkMaterializeContext) -> V41HostOccupancy { match spark_remote_run(ctx: ctx, argv: ["cat", "/proc/meminfo"]) { - RemoteUnreachable { cause: why } => V41HostOccupied { cause: join(["/proc/meminfo could not be read: ", why as String], "") as NonEmptyStr } + RemoteUnreachable { cause: why, reach: _ } => V41HostOccupied { cause: join(["/proc/meminfo could not be read: ", why as String], "") as NonEmptyStr } RemoteRan { exit_code: c, stdout: mem, stderr: e } => if c != 0 { V41HostOccupied { cause: join(["/proc/meminfo exited ", to_string(c), ": ", trim(s: e)], "") as NonEmptyStr } } else { match spark_remote_run(ctx: ctx, argv: ["nvidia-smi", "--query-compute-apps=pid,process_name,used_memory", "--format=csv"]) { - RemoteUnreachable { cause: why } => V41HostOccupied { cause: join(["nvidia-smi could not run: ", why as String], "") as NonEmptyStr } + RemoteUnreachable { cause: why, reach: _ } => V41HostOccupied { cause: join(["nvidia-smi could not run: ", why as String], "") as NonEmptyStr } RemoteRan { exit_code: c2, stdout: apps, stderr: e2 } => if c2 != 0 { V41HostOccupied { cause: join(["nvidia-smi exited ", to_string(c2), ": ", trim(s: e2)], "") as NonEmptyStr } } else { v41_judge_occupancy(meminfo: mem, compute_csv: apps) } } @@ -250,12 +250,12 @@ fn v41_read_occupancy(ctx: SparkMaterializeContext) -> V41HostOccupancy { // ── ASKING IT ─────────────────────────────────────────────────────────────────────────────────── fn v41_differential_remove_containers(ctx: SparkMaterializeContext) -> String? { match spark_remote_run(ctx: ctx, argv: argv_words(command: docker_ps_running_command(filter: DockerPsByLabel { label: v41_differential_lane_label }))) { - RemoteUnreachable { cause: why } => Present { value: join(["docker ps did not run: ", why as String], "") } + RemoteUnreachable { cause: why, reach: _ } => Present { value: join(["docker ps did not run: ", why as String], "") } RemoteRan { exit_code: c, stdout: out, stderr: e } => if c != 0 { Present { value: join(["docker ps exited ", to_string(c), ": ", trim(s: e)], "") } } else { let failures = flat_map(filter(split(s: out, delimiter: "\n"), w => trim(s: w) != ""), id => match spark_remote_run(ctx: ctx, argv: argv_words(command: docker_remove_force_command(name: trim(s: id) as NonEmptyStr))) { - RemoteUnreachable { cause: why } => [join(["docker rm -f ", trim(s: id), " did not run: ", why as String], "")] + RemoteUnreachable { cause: why, reach: _ } => [join(["docker rm -f ", trim(s: id), " did not run: ", why as String], "")] RemoteRan { exit_code: rc, stdout: _, stderr: re } => if rc == 0 { [] as List } else { [join(["docker rm -f ", trim(s: id), " exited ", to_string(rc), ": ", trim(s: re)], "")] } }) if length(failures) != 0 { Present { value: join(failures, "; ") } } else { none } @@ -265,7 +265,7 @@ fn v41_differential_remove_containers(ctx: SparkMaterializeContext) -> String? { fn v41_differential_on(ctx: SparkMaterializeContext, claim: NonEmptyStr, image: NonEmptyStr, manifest: String, owed: List) -> V41DifferentialOutcome { match spark_remote_run(ctx: ctx, argv: argv_words(command: docker_image_inspect_command(image: image))) { - RemoteUnreachable { cause: why } => V41DifferentialRefused { step: "image-config-id" as NonEmptyStr, cause: why } + RemoteUnreachable { cause: why, reach: _ } => V41DifferentialRefused { step: "image-config-id" as NonEmptyStr, cause: why } RemoteRan { exit_code: code, stdout: inspect_stdout, stderr: err } => if code != 0 { if docker_image_inspect_names_no_such_image(stderr: err) { @@ -278,7 +278,7 @@ fn v41_differential_on(ctx: SparkMaterializeContext, claim: NonEmptyStr, image: DockerImageInspectIdUnreadable { cause: why } => V41DifferentialRefused { step: "image-config-id" as NonEmptyStr, cause: why } DockerImageInspectId { id: config_id } => { let outcome = match spark_remote_run(ctx: ctx, argv: v41_differential_argv(image: config_id, claim: claim, manifest: manifest)) { - RemoteUnreachable { cause: why } => V41DifferentialRefused { step: "differential-run" as NonEmptyStr, cause: why } + RemoteUnreachable { cause: why, reach: _ } => V41DifferentialRefused { step: "differential-run" as NonEmptyStr, cause: why } RemoteRan { exit_code: c, stdout: out, stderr: e } => if c == 0 || c == 1 { v41_admit_differential(image_config: config_id, owed: owed, stdout: out) diff --git a/dag/gunbc/spark/v41_group_a_launch.dag b/dag/gunbc/spark/v41_group_a_launch.dag index 842b229e006..fd319c09ec9 100644 --- a/dag/gunbc/spark/v41_group_a_launch.dag +++ b/dag/gunbc/spark/v41_group_a_launch.dag @@ -881,7 +881,7 @@ fn v41_sm120_host_gate(ctx: SparkMaterializeContext, l: ArmNodeLaunch) -> V41Sm1 Absent => V41Sm120HostGate { passed: false, receipt: join(["sm120 self-test ", host, ": REFUSED the rank's engine argv is empty, so it names no model path to run against\n"], "") } Present { value: model_path } => match spark_remote_run(ctx: vllm_image_target_ctx(source_ctx: ctx, host: l.node.host), argv: v41_sm120_selftest_argv(l: l, model_path: model_path)) { - RemoteUnreachable { cause: c } => V41Sm120HostGate { passed: false, receipt: join(["sm120 self-test ", host, ": DID NOT RUN ", c, "\n"], "") } + RemoteUnreachable { cause: c, reach: _ } => V41Sm120HostGate { passed: false, receipt: join(["sm120 self-test ", host, ": DID NOT RUN ", c, "\n"], "") } RemoteRan { exit_code: code, stdout: out, stderr: err } => { let passed = code == 0 && trim(s: out).ends_with(suffix: "SELFTEST PASS") V41Sm120HostGate { diff --git a/dag/gunbc/spark/v41_row_store_encode_run.dag b/dag/gunbc/spark/v41_row_store_encode_run.dag index b757ca7bd37..8f16a25109f 100644 --- a/dag/gunbc/spark/v41_row_store_encode_run.dag +++ b/dag/gunbc/spark/v41_row_store_encode_run.dag @@ -107,7 +107,7 @@ type V41DigestRead fn v41_store_digest(ctx: SparkMaterializeContext, out: NonEmptyStr) -> V41DigestRead { match spark_remote_run(ctx: ctx, argv: sha256sum_argv(path: out as String)) { - RemoteUnreachable { cause: c } => V41DigestReadFailed { cause: join(["the sha256sum leg did not run: ", c], "") as NonEmptyStr } + RemoteUnreachable { cause: c, reach: _ } => V41DigestReadFailed { cause: join(["the sha256sum leg did not run: ", c], "") as NonEmptyStr } RemoteRan { exit_code: code, stdout: s, stderr: err } => if code != 0 { V41DigestReadFailed { cause: join(["sha256sum exited ", to_string(code), ": ", trim(s: err)], "") as NonEmptyStr } } else { match sha256sum_line_digest(line: s) { @@ -137,7 +137,7 @@ fn v41_store_presence_of(exit_code: Int, stderr: String) -> V41StorePresence { fn v41_store_present(ctx: SparkMaterializeContext, out: NonEmptyStr) -> V41StorePresence { match spark_remote_run(ctx: ctx, argv: ["test", "-f", out as String]) { - RemoteUnreachable { cause: c } => V41StorePresenceUnread { cause: join(["the presence check did not run: ", c], "") as NonEmptyStr } + RemoteUnreachable { cause: c, reach: _ } => V41StorePresenceUnread { cause: join(["the presence check did not run: ", c], "") as NonEmptyStr } RemoteRan { exit_code: code, stdout: _, stderr: err } => v41_store_presence_of(exit_code: code, stderr: err) } } @@ -155,7 +155,7 @@ type V41SizeRead fn v41_store_size(ctx: SparkMaterializeContext, out: NonEmptyStr) -> V41SizeRead { match spark_remote_run(ctx: ctx, argv: ["stat", "-c", "%s", "--", out as String]) { - RemoteUnreachable { cause: c } => V41SizeReadFailed { cause: join(["the stat leg did not run: ", c], "") as NonEmptyStr } + RemoteUnreachable { cause: c, reach: _ } => V41SizeReadFailed { cause: join(["the stat leg did not run: ", c], "") as NonEmptyStr } RemoteRan { exit_code: code, stdout: o, stderr: err } => if code != 0 { V41SizeReadFailed { cause: join(["stat exited ", to_string(code), ": ", trim(s: err)], "") as NonEmptyStr } } else { match parse_int(s: trim(s: o)) { @@ -191,7 +191,7 @@ fn v41_run_store(ctx: SparkMaterializeContext, a: V41StoreEncodeArgs) -> V41Stor V41StoreFound => v41_digested(ctx: ctx, a: a, encoded: false, read: v41_store_digest(ctx: ctx, out: a.out)) V41StoreAbsent => match spark_remote_run(ctx: ctx, argv: v41_store_encode_argv(shard_directory: v41_checkpoint_directory, a: a)) { - RemoteUnreachable { cause: c } => V41StoreEncodeFailed { store: a, cause: join(["the encode leg did not run: ", c], "") as NonEmptyStr } + RemoteUnreachable { cause: c, reach: _ } => V41StoreEncodeFailed { store: a, cause: join(["the encode leg did not run: ", c], "") as NonEmptyStr } RemoteRan { exit_code: code, stdout: _, stderr: err } => if code != 0 { V41StoreEncodeFailed { store: a, cause: join(["the host program exited ", to_string(code), ": ", trim(s: err)], "") as NonEmptyStr } @@ -238,7 +238,7 @@ fn v41_encode_all(ctx: SparkMaterializeContext) -> V41EncodeRun { V41StoreEncodeUnplannable { defects: d } => V41EncodeRefusedBeforeRun { cause: join(["the encode plan refused: ", join(map(d, x => x as String), "; ")], "") as NonEmptyStr } V41StoreEncodePlanned { stores: st } => match spark_remote_run(ctx: ctx, argv: ["mkdir", "-p", v41_row_store_output_directory() as String]) { - RemoteUnreachable { cause: c } => V41EncodeRefusedBeforeRun { cause: join(["the output directory could not be made: ", c], "") as NonEmptyStr } + RemoteUnreachable { cause: c, reach: _ } => V41EncodeRefusedBeforeRun { cause: join(["the output directory could not be made: ", c], "") as NonEmptyStr } RemoteRan { exit_code: code, stdout: _, stderr: err } => if code != 0 { V41EncodeRefusedBeforeRun { cause: join(["mkdir of the output directory exited ", to_string(code), ": ", trim(s: err)], "") as NonEmptyStr } } else { V41EncodeRan { planned: st, stores: map(st, a => v41_run_store(ctx: ctx, a: a)) } diff --git a/dag/gunbc/spark/v41_row_store_readback_run.dag b/dag/gunbc/spark/v41_row_store_readback_run.dag index e2d269b3b98..becc2ef540f 100644 --- a/dag/gunbc/spark/v41_row_store_readback_run.dag +++ b/dag/gunbc/spark/v41_row_store_readback_run.dag @@ -110,7 +110,7 @@ fn v41_od_hex(stdout: String) -> String { fn v41_read_sample(ctx: SparkMaterializeContext, r: V41HostRegion) -> V41SampleRead { match spark_remote_run(ctx: ctx, argv: argv_words(command: od_hex_span_command(path: r.path, offset: r.offset, length: r.length))) { - RemoteUnreachable { cause: c } => V41SampleReadFailed { region: r, cause: join(["the od leg did not run: ", c], "") as NonEmptyStr } + RemoteUnreachable { cause: c, reach: _ } => V41SampleReadFailed { region: r, cause: join(["the od leg did not run: ", c], "") as NonEmptyStr } RemoteRan { exit_code: code, stdout: o, stderr: err } => if code != 0 { V41SampleReadFailed { region: r, cause: join(["od exited ", to_string(code), ": ", trim(s: err)], "") as NonEmptyStr } } else { V41SampleReadOk { region: r, hex: v41_od_hex(stdout: o) } diff --git a/dag/gunbc/spark/v41_runtime_image_probe.dag b/dag/gunbc/spark/v41_runtime_image_probe.dag index 23c023286d0..dfd859c52d0 100644 --- a/dag/gunbc/spark/v41_runtime_image_probe.dag +++ b/dag/gunbc/spark/v41_runtime_image_probe.dag @@ -743,7 +743,7 @@ fn v41_probe_config_id_argv(image_reference: NonEmptyStr) -> List { // already keeps them apart; this preserves that rather than folding both into "the probe failed". fn v41_privileged_outcome(step: NonEmptyStr, run: SparkRemoteRun) -> V41ProbeOutcome? { match run { - RemoteUnreachable { cause: why } => Present { value: V41ProbeRefused { step: step, cause: why } } + RemoteUnreachable { cause: why, reach: _ } => Present { value: V41ProbeRefused { step: step, cause: why } } RemoteRan { exit_code: code, stdout: _, stderr: err } => if code != 0 { Present { value: V41ProbeRefused { step: step, cause: join(["exit=", to_string(code), " stderr=", trim(s: err)], "") } } @@ -755,7 +755,7 @@ fn v41_privileged_outcome(step: NonEmptyStr, run: SparkRemoteRun) -> V41ProbeOut fn v41_privileged_stdout(run: SparkRemoteRun) -> String { match run { - RemoteUnreachable { cause: _ } => "" + RemoteUnreachable { cause: _, reach: _ } => "" RemoteRan { exit_code: _, stdout: out, stderr: _ } => out } } @@ -775,7 +775,7 @@ fn v41_probe_image_in_place(ctx: SparkMaterializeContext, reference: NonEmptyStr { let inspect_run = spark_remote_run(ctx: ctx, argv: v41_probe_config_id_argv(image_reference: reference)) match inspect_run { - RemoteUnreachable { cause: why } => V41ProbeRefused { step: "image-config-id" as NonEmptyStr, cause: why } + RemoteUnreachable { cause: why, reach: _ } => V41ProbeRefused { step: "image-config-id" as NonEmptyStr, cause: why } RemoteRan { exit_code: code, stdout: inspect_stdout, stderr: err } => if code != 0 { if docker_image_inspect_names_no_such_image(stderr: err) { diff --git a/dag/gunbc/spark/v41_serving_load.dag b/dag/gunbc/spark/v41_serving_load.dag index 0f60d01b1db..57e9f552b72 100644 --- a/dag/gunbc/spark/v41_serving_load.dag +++ b/dag/gunbc/spark/v41_serving_load.dag @@ -218,7 +218,7 @@ type V41RemoteText fn v41_remote_text(ctx: SparkMaterializeContext, what: String, argv: List) -> V41RemoteText { match spark_remote_run(ctx: ctx, argv: argv) { - RemoteUnreachable { cause: c } => V41RemoteTextRefused { cause: join([what, " unreachable: ", c], "") } + RemoteUnreachable { cause: c, reach: _ } => V41RemoteTextRefused { cause: join([what, " unreachable: ", c], "") } RemoteRan { exit_code: code, stdout: out, stderr: err } => if code != 0 { V41RemoteTextRefused { cause: join([what, " exited ", to_string(code), ": ", err], "") } diff --git a/dag/gunbc/spark/vllm_runtime_image_build.dag b/dag/gunbc/spark/vllm_runtime_image_build.dag index fc03906dcb5..8339d5cdecd 100644 --- a/dag/gunbc/spark/vllm_runtime_image_build.dag +++ b/dag/gunbc/spark/vllm_runtime_image_build.dag @@ -629,7 +629,7 @@ type ImageLegOutcome fn vllm_docker_leg(ctx: SparkMaterializeContext, step: NonEmptyStr, command: ArgvCommand) -> ImageLegOutcome { match spark_remote_run(ctx: ctx, argv: argv_words(command: command)) { - RemoteUnreachable { cause: why } => ImageLegFailed { step: step, cause: why } + RemoteUnreachable { cause: why, reach: _ } => ImageLegFailed { step: step, cause: why } RemoteRan { exit_code: code, stdout: out, stderr: err } => if code != 0 { ImageLegFailed { step: step, cause: join(["exit=", to_string(code), " stderr=", trim(s: err)], "") } @@ -1310,7 +1310,7 @@ fn vllm_image_held_of_inspect(step: NonEmptyStr, stdout: String) -> VllmImageHel fn vllm_image_read_held(ctx: SparkMaterializeContext, step: NonEmptyStr, tag: NonEmptyStr) -> VllmImageHeld { match spark_remote_run(ctx: ctx, argv: argv_words(command: docker_image_inspect_command(image: tag))) { - RemoteUnreachable { cause: why } => VllmImageHeldUnread { step: step, cause: why } + RemoteUnreachable { cause: why, reach: _ } => VllmImageHeldUnread { step: step, cause: why } RemoteRan { exit_code: code, stdout: out, stderr: err } => if code == 0 { vllm_image_held_of_inspect(step: step, stdout: out) @@ -1344,7 +1344,7 @@ fn vllm_image_disk_refusal(step: NonEmptyStr, place: String, required: ByteSize, fn vllm_image_remote_disk_refusal(ctx: SparkMaterializeContext, step: NonEmptyStr, path: NonEmptyStr, required: ByteSize) -> String? { let place = join([ctx.host as String, ":", path as String], "") match spark_remote_run(ctx: ctx, argv: host_disk_free_space_argv(path: path as String)) { - RemoteUnreachable { cause: why } => Present { value: join([step as String, ": the free-space read at ", place, " did not run: ", why], "") } + RemoteUnreachable { cause: why, reach: _ } => Present { value: join([step as String, ": the free-space read at ", place, " did not run: ", why], "") } RemoteRan { exit_code: code, stdout: out, stderr: err } => vllm_image_disk_refusal(step: step, place: place, required: required, reading: host_disk_observation_of_df(exit_code: code, stdout: out, stderr: err, observed_on: ctx.host as NonEmptyStr)) diff --git a/dag/test/claim/spark/serving_load_probe_witness_test.dag b/dag/test/claim/spark/serving_load_probe_witness_test.dag index 2c229d88a8d..9c70328ca65 100644 --- a/dag/test/claim/spark/serving_load_probe_witness_test.dag +++ b/dag/test/claim/spark/serving_load_probe_witness_test.dag @@ -14,7 +14,7 @@ import gunbc.spark.serving_rank_observe { container_incarnation_id, ObservedRankIncarnation, } import product.fabric.demand { ObservationReceiptRef } -import gunbc.spark.model_snapshot_materialize { RemoteRan, RemoteUnreachable } +import gunbc.spark.model_snapshot_materialize { RemoteRan, RemoteUnreachable, LegNotAttempted } import gunbc.spark.vllm_endpoint_process_launch { vllm_endpoint_process_launch, VllmEndpointProcessLaunchObservation, LaunchObserved, } @@ -569,7 +569,7 @@ test fn a_nonzero_bench_exit_does_not_parse_stdout() -> Bool { test fn an_unreachable_bench_does_not_parse_stdout() -> Bool { client_output_is_unread(e: bench_serve_from_remote( protocol: glm_native_smoke_protocol, - remote: RemoteUnreachable { cause: "ssh refused" }, + remote: RemoteUnreachable { cause: "ssh refused", reach: LegNotAttempted { cause: "ssh refused" } }, )) } diff --git a/dag/test/claim/spark/spark_remote_leg_deadline_witness_test.dag b/dag/test/claim/spark/spark_remote_leg_deadline_witness_test.dag new file mode 100644 index 00000000000..e846b861bdb --- /dev/null +++ b/dag/test/claim/spark/spark_remote_leg_deadline_witness_test.dag @@ -0,0 +1,95 @@ +module test.claim.spark.spark_remote_leg_deadline_witness_test + +import std.types { String, Bool, List, NonEmptyStr } +import std.measure { second, second_count } +import extdeps.ssh.client_options { serialize_ssh_client_options } +import gunbc.fleet_known_hosts_anchor { + FleetSshLegDeadline, + fleet_ssh_leg_deadline, + fleet_ssh_leg_deadline_options, + fleet_ssh_password_client_options, +} +import gunbc.spark.model_snapshot_materialize { + SparkRemoteRun, RemoteRan, RemoteUnreachable, + LegNotAttempted, LegNeverBegan, + spark_remote_run_of_session, spark_remote_leg_began_word, +} + +// THE DEADLINE THE SPARK LEGS CARRY, asserted as exact argv. Fleet-converge run 37156689444 sat ~1h +// inside one ssh leg on 2026-10-03 because the shapers bounded the connect and nothing after it. + +test fn w_fleet_deadline_serializes_connect_and_liveness_words() -> Bool { + serialize_ssh_client_options(options: fleet_ssh_leg_deadline_options(deadline: fleet_ssh_leg_deadline)) == [ + "-o", "ConnectTimeout=10", + "-o", "ServerAliveInterval=15", + "-o", "ServerAliveCountMax=4", + ] +} + +// THE ROUTE, not only the row: the password session spark_remote_run reaches through +// gunbc.fleet_bootstrap_principal_session derives its client options here, so deleting the deadline +// from the shaper reds this claim even while the row above still serializes. +test fn w_password_session_options_carry_the_liveness_bound() -> Bool { + let line = join(serialize_ssh_client_options(options: fleet_ssh_password_client_options(trust: [], principal: "briansrls" as NonEmptyStr)), " ") + contains(line, "-o ConnectTimeout=10 -o ServerAliveInterval=15 -o ServerAliveCountMax=4") +} + +// ssh_config(5): ServerAliveInterval 0 is "never send", so a zeroed interval or count is the +// deadline-less leg in disguise. The live row must bound a dark session in finite time; the red +// control below asserts the zeroed shape is recognized as unbounded. +fn deadline_bounds_a_dark_session(d: FleetSshLegDeadline) -> Bool { + second_count(s: d.connect) > 0 && second_count(s: d.alive_interval) > 0 && d.alive_count_max > 0 +} + +test fn w_live_deadline_bounds_a_dark_session() -> Bool { + deadline_bounds_a_dark_session(d: fleet_ssh_leg_deadline) +} + +test fn red_control_zeroed_keepalive_is_not_a_bound() -> Bool { + !deadline_bounds_a_dark_session(d: FleetSshLegDeadline { connect: second(count: 10), alive_interval: second(count: 0), alive_count_max: 4 }) +} + +// THE CLASSIFIER, over supplied session results (DESIGN §3: the claim is about the reading, not the +// transport). 255 with no began-line is the only unreached reading. + +fn is_never_began(r: SparkRemoteRun) -> Bool { + match r { + RemoteUnreachable { cause: _, reach: LegNeverBegan { endpoint: _, stderr: _ } } => true + RemoteUnreachable { cause: _, reach: LegNotAttempted { cause: _ } } => false + RemoteRan { exit_code: _, stdout: _, stderr: _ } => false + } +} + +test fn w_dark_connect_reads_as_never_began_not_as_a_command_exit() -> Bool { + is_never_began(r: spark_remote_run_of_session( + endpoint: "192.168.1.232", + exit_code: 255, + stdout: "", + stderr: "ssh: connect to host 192.168.1.232 port 22: Connection timed out", + )) +} + +// The cause text names the undecided pair, never "host down" alone (DESIGN §4d). +test fn w_never_began_text_says_dark_or_down_undecided() -> Bool { + match spark_remote_run_of_session(endpoint: "192.168.1.232", exit_code: 255, stdout: "", stderr: "Timeout, server 192.168.1.232 not responding.") { + RemoteUnreachable { cause: c, reach: _ } => contains(c, "LAN leg dark or host down, UNDECIDED") + RemoteRan { exit_code: _, stdout: _, stderr: _ } => false + } +} + +// A session lost AFTER the began-line ran to an unknown point: it stays a ran result with ssh's 255, +// never an unreached one a caller could retry as if nothing happened. +test fn w_session_lost_after_begin_stays_ran() -> Bool { + match spark_remote_run_of_session(endpoint: "h", exit_code: 255, stdout: join([spark_remote_leg_began_word, "partial"], "\n"), stderr: "Timeout, server h not responding.") { + RemoteRan { exit_code: c, stdout: o, stderr: _ } => c == 255 && o == "partial" + RemoteUnreachable { cause: _, reach: _ } => false + } +} + +// The began-line is stripped, so every consumer reads exactly the command's stdout. +test fn w_began_line_never_reaches_a_consumer() -> Bool { + match spark_remote_run_of_session(endpoint: "h", exit_code: 0, stdout: join([spark_remote_leg_began_word, "active"], "\n"), stderr: "") { + RemoteRan { exit_code: c, stdout: o, stderr: _ } => c == 0 && o == "active" + RemoteUnreachable { cause: _, reach: _ } => false + } +} From d442c7dea25e9a39173f5cd40cf74e7adde955b7 Mon Sep 17 00:00:00 2001 From: Brian Searls Date: Sun, 4 Oct 2026 03:45:37 +0000 Subject: [PATCH 2/5] Name the fabric-rail probe from a same-group peer as the first settling readback for LegNeverBegan Co-Authored-By: Claude Opus 5.5 (1M context) --- ...gement_lan_leg_hangs_and_reads_as_a_down_host.dag | 2 +- dag/gunbc/spark/model_snapshot_materialize.dag | 12 ++++++++---- 2 files changed, 9 insertions(+), 5 deletions(-) diff --git a/dag/gunbc/recurring_failure_mode/a_dark_management_lan_leg_hangs_and_reads_as_a_down_host.dag b/dag/gunbc/recurring_failure_mode/a_dark_management_lan_leg_hangs_and_reads_as_a_down_host.dag index 1234ed7db51..ab4eb08eaab 100644 --- a/dag/gunbc/recurring_failure_mode/a_dark_management_lan_leg_hangs_and_reads_as_a_down_host.dag +++ b/dag/gunbc/recurring_failure_mode/a_dark_management_lan_leg_hangs_and_reads_as_a_down_host.dag @@ -13,7 +13,7 @@ data a_dark_management_lan_leg_hangs_and_reads_as_a_down_host: RecurringFailureM "HARM: 2026-10-03, Spark Group A bring-up. Every Spark reaches the executor only over wlP9s9 (MediaTek mt7925e). Router band steering roamed the hosts; a failed 4-way handshake left NetworkManager DISCONNECTED ('no secrets: No agents were available') on srv8 and Group B spark-3336, and srv7 stayed associated but forwarded nothing 05:33-06:53 EDT (read on the hosts by valiant-crab-775). Fleet-converge run 37156689444 sat ~1h inside one ssh leg. No host rebooted or wedged; the runs called them down.", - "DISTINGUISHING FACTS: liveness is a different bound from connect (ssh_config(5) ServerAliveInterval x ServerAliveCountMax); sshd answers keepalives itself, so a silent remote command is not cut off, only a path that returns nothing. Whether the leg BEGAN is read in band (spark_remote_leg_began_word on stdout), not from stderr. 'Host down' needs an observation this boundary does not hold, so the type has no HostDown arm: LegNeverBegan is dark-or-down UNDECIDED and names the readback (uptime, wlP9s9 journal) that closes it.", + "DISTINGUISHING FACTS: liveness is a different bound from connect (ssh_config(5) ServerAliveInterval x ServerAliveCountMax); sshd answers keepalives itself, so a silent remote command is not cut off, only a path that returns nothing. Whether the leg BEGAN is read in band (spark_remote_leg_began_word on stdout), not from stderr. 'Host down' needs an observation this boundary does not hold, so the type has no HostDown arm: LegNeverBegan is dark-or-down UNDECIDED and names the readbacks that close it: first a probe of the host's fabric-rail address (192.168.110.x) from a same-group peer (rail answers -> LAN dark, host up), then uptime and the wlP9s9 journal. A second trigger, separate from the deadline one: an automatic rail-jump classifier with its own consumer, SUFFICIENT FOR LegNeverBegan to be split into LAN-dark-host-up vs rail-also-dark without a human.", "RUNG: found at 1 (a human read the hang and the hosts). Now 2: gunbc.fleet_known_hosts_anchor fleet_ssh_leg_deadline is the one row both fleet shapers derive their deadline words from and neither shaper takes an options parameter, and dag/test/claim/spark/spark_remote_leg_deadline_witness_test.dag pins the argv, the route through the password session, a zeroed-keepalive red control and the never-began reading. CEILING 3: the deadline as a required field of the operation every fleet ssh transport declares, so no ssh leg in the corpus (including extdeps.bmc.openbmc_password_ssh_transport and extdeps.ssh.password_session CopyFile, which still carry ConnectTimeout only) is expressible without it. NEXT-RUNG TRIGGER: those transport literals moved onto extdeps.ssh.client_options with the deadline derived, SUFFICIENT FOR no ssh argv in the corpus lacking a liveness bound.", ], diff --git a/dag/gunbc/spark/model_snapshot_materialize.dag b/dag/gunbc/spark/model_snapshot_materialize.dag index 76d2cb1f399..2a602a27f01 100644 --- a/dag/gunbc/spark/model_snapshot_materialize.dag +++ b/dag/gunbc/spark/model_snapshot_materialize.dag @@ -76,9 +76,13 @@ type SparkRemoteRun // prerequisite. A fact about this run, not about the host or the LAN. // LegNeverBegan -- ssh exited with its own 255 and the remote shell never reported beginning // (spark_remote_leg_began_word). The LAN leg is dark OR the host is down, -// UNDECIDED; the discriminator that closes it is a readback once the host is -// reachable again (uptime / the NetworkManager journal for wlP9s9), and the -// 2026-10-03 incident closed it as LAN-dark every time. +// UNDECIDED. The FIRST settling readback is a probe of the host's fabric-rail +// address (enp1s0f1np1, 192.168.110.x) from a same-group peer: rail answers -> +// LAN dark, host up; rail dark too -> host down or both links down, still +// undecided. Uptime and the wlP9s9 NetworkManager journal come after, since they +// need a session. This is how the 2026-10-03 incident was closed by hand, as +// LAN-dark every time. NEXT RUNG: an automatic rail-jump classifier with its own +// consumer turns LegNeverBegan into LanLegDark / RailAlsoDark; not built here. // // A leg that BEGAN and then lost its session is not here: it ran to an unknown point, so it stays // RemoteRan with ssh's 255 -- retrying it as if nothing ran is the opposite, non-idempotent decision @@ -93,7 +97,7 @@ fn spark_leg_unreachability_text(u: SparkLegUnreachability) -> String { LegNeverBegan { endpoint: e, stderr: err } => join([ "the ssh leg to ", e, " never began (ssh exit 255, no remote began-line): management LAN leg dark ", - "or host down, UNDECIDED from this leg -- read the host's uptime and wlP9s9 journal once reachable; stderr=", + "or host down, UNDECIDED from this leg -- settle it first by probing the host's fabric-rail address from a same-group peer, then uptime and the wlP9s9 journal; stderr=", err, ], "") } From 75d04210fe66763ec40d9938c77afbcfb5049ffe Mon Sep 17 00:00:00 2001 From: Brian Searls Date: Sun, 4 Oct 2026 04:49:40 +0000 Subject: [PATCH 3/5] Begin-unobserved, not never-began: a 255 with no began-line may have run and projects to ArgvLegMayHaveRun (review of #13204) Co-Authored-By: Claude Opus 5.5 (1M context) --- dag/gunbc/compute/host_occupancy.dag | 6 +- dag/gunbc/fleet/fleet_wireless_link.dag | 4 +- ...lan_leg_hangs_and_reads_as_a_down_host.dag | 4 +- ..._leg_reads_as_one_that_ran_and_refused.dag | 2 +- dag/gunbc/spark/host_effect_quiescence.dag | 8 ++ dag/gunbc/spark/host_occupancy_admission.dag | 10 ++- .../spark/model_snapshot_materialize.dag | 76 +++++++++++-------- ...spark_remote_leg_deadline_witness_test.dag | 49 +++++++++--- 8 files changed, 109 insertions(+), 50 deletions(-) diff --git a/dag/gunbc/compute/host_occupancy.dag b/dag/gunbc/compute/host_occupancy.dag index ff78a6a8a5d..239b738a233 100644 --- a/dag/gunbc/compute/host_occupancy.dag +++ b/dag/gunbc/compute/host_occupancy.dag @@ -19,7 +19,7 @@ import extdeps.systemd { SystemdUnitActiveState, Active, Inactive, Activating, Deactivating, Reloading, Failed, parse_systemd_unit_active_state, systemd_unit_active_state_wire_label, } -import gunbc.spark.host_effect_quiescence { ArgvRun, ArgvRan, ArgvLegDidNotRun } +import gunbc.spark.host_effect_quiescence { ArgvRun, ArgvRan, ArgvLegDidNotRun, ArgvLegMayHaveRun } // ── WHO IS ALREADY USING THIS HOST: A READING, TAKEN ON THE HOST, BEFORE AN EFFECT IS ADMITTED ─ // @@ -85,6 +85,7 @@ type HostOperationLeg type HostOperationRan = HostOperationExited { exit_code: Int, stdout: String, stderr: String } | HostOperationNotRun { cause: String } + | HostOperationResultUnreceived { cause: String } fn host_operation_ran(run: fn(List) -> ArgvRun, operation: HostOperation) -> HostOperationRan { match host_operation_materialize_argv(operation: operation) { @@ -92,6 +93,7 @@ fn host_operation_ran(run: fn(List) -> ArgvRun, operation: HostOperation ArgvMaterialized { argv: argv } => match run(argv) { ArgvLegDidNotRun { cause: c } => HostOperationNotRun { cause: join([host_operation_label(operation: operation), " did not run: ", c], "") } + ArgvLegMayHaveRun { cause: c } => HostOperationResultUnreceived { cause: join([host_operation_label(operation: operation), " may have run, no result received: ", c], "") } ArgvRan { exit_code: code, stdout: out, stderr: err } => HostOperationExited { exit_code: code, stdout: out, stderr: err } } } @@ -104,6 +106,7 @@ fn host_operation_exit_unread(operation: HostOperation, code: Int, err: String) fn host_operation_leg(run: fn(List) -> ArgvRun, operation: HostOperation, accepted_exits: List) -> HostOperationLeg { match host_operation_ran(run: run, operation: operation) { HostOperationNotRun { cause: c } => HostOperationLegUnread { cause: c } + HostOperationResultUnreceived { cause: c } => HostOperationLegUnread { cause: c } HostOperationExited { exit_code: code, stdout: out, stderr: err } => if any(accepted_exits, e => e == code) { HostOperationLegRead { stdout: out } } else { HostOperationLegUnread { cause: host_operation_exit_unread(operation: operation, code: code, err: err) } } @@ -125,6 +128,7 @@ fn read_unit_activity(run: fn(List) -> ArgvRun, unit: NonEmptyStr) -> Un let operation = SystemctlIsActive { unit: unit } match host_operation_ran(run: run, operation: operation) { HostOperationNotRun { cause: c } => UnitUnread { cause: c } + HostOperationResultUnreceived { cause: c } => UnitUnread { cause: c } HostOperationExited { exit_code: code, stdout: out, stderr: err } => match systemctl_is_active_exit(code: code) { Absent => UnitUnread { cause: host_operation_exit_unread(operation: operation, code: code, err: err) } diff --git a/dag/gunbc/fleet/fleet_wireless_link.dag b/dag/gunbc/fleet/fleet_wireless_link.dag index 89bec0d4913..a700d098503 100644 --- a/dag/gunbc/fleet/fleet_wireless_link.dag +++ b/dag/gunbc/fleet/fleet_wireless_link.dag @@ -20,7 +20,7 @@ import extdeps.networkmanager.nmcli { nmcli_connection_property_get_argv, nmcli_wireless_powersave_modify_argv, } import extdeps.systemd.journalctl { journalctl_kernel_current_boot_argv } -import gunbc.spark.host_effect_quiescence { ArgvRun, ArgvRan, ArgvLegDidNotRun } +import gunbc.spark.host_effect_quiescence { ArgvRun, ArgvRan, ArgvLegDidNotRun, ArgvLegMayHaveRun } import gunbc.fleet_intent_network { operator_host_srv5, operator_host_srv6, operator_host_srv7, operator_host_srv8, operator_host_srv9, operator_host_srv10, operator_host_srv11, @@ -105,6 +105,7 @@ type WirelessLinkObservation { fn argv_run_failure(run: ArgvRun) -> String? { match run { ArgvLegDidNotRun { cause: c } => Present { value: join(["leg did not run: ", c], "") } + ArgvLegMayHaveRun { cause: c } => Present { value: join(["leg may have run, no result received (read back before retrying): ", c], "") } ArgvRan { exit_code: code, stdout: _, stderr: err } => if code != 0 { Present { value: join(["exit=", to_string(code), " stderr=", trim(s: err)], "") } } else { none } } @@ -113,6 +114,7 @@ fn argv_run_failure(run: ArgvRun) -> String? { fn argv_run_stdout(run: ArgvRun) -> String { match run { ArgvLegDidNotRun { cause: _ } => "" + ArgvLegMayHaveRun { cause: _ } => "" ArgvRan { exit_code: _, stdout: o, stderr: _ } => o } } diff --git a/dag/gunbc/recurring_failure_mode/a_dark_management_lan_leg_hangs_and_reads_as_a_down_host.dag b/dag/gunbc/recurring_failure_mode/a_dark_management_lan_leg_hangs_and_reads_as_a_down_host.dag index ab4eb08eaab..127f9f38dfe 100644 --- a/dag/gunbc/recurring_failure_mode/a_dark_management_lan_leg_hangs_and_reads_as_a_down_host.dag +++ b/dag/gunbc/recurring_failure_mode/a_dark_management_lan_leg_hangs_and_reads_as_a_down_host.dag @@ -9,11 +9,11 @@ data a_dark_management_lan_leg_hangs_and_reads_as_a_down_host: RecurringFailureM receipts: [ "**a dark management-LAN leg hangs, then reads as a down host** (a fleet ssh leg bounded only at connect waits indefinitely when an established session's path goes dark, and when it does end, the run reports the host as down although the evidence in hand cannot tell a dark LAN leg from a host that is off).", - "INVALID STATE: gunbc.fleet_known_hosts_anchor shape_fleet_ssh_exec and fleet_ssh_password_client_options carried ConnectTimeout=10 and no ServerAliveInterval (ssh_config(5) default 0, never), so an established session had no liveness bound; and gunbc.spark.model_snapshot_materialize spark_remote_run folded ssh's own 255 into RemoteRan, so a leg that never began read as a command that exited 255.", + "INVALID STATE: gunbc.fleet_known_hosts_anchor shape_fleet_ssh_exec and fleet_ssh_password_client_options carried ConnectTimeout=10 and no ServerAliveInterval (ssh_config(5) default 0, never), so an established session had no liveness bound; and gunbc.spark.model_snapshot_materialize spark_remote_run folded ssh's own 255 into RemoteRan, so a leg whose session died read as a command that exited 255.", "HARM: 2026-10-03, Spark Group A bring-up. Every Spark reaches the executor only over wlP9s9 (MediaTek mt7925e). Router band steering roamed the hosts; a failed 4-way handshake left NetworkManager DISCONNECTED ('no secrets: No agents were available') on srv8 and Group B spark-3336, and srv7 stayed associated but forwarded nothing 05:33-06:53 EDT (read on the hosts by valiant-crab-775). Fleet-converge run 37156689444 sat ~1h inside one ssh leg. No host rebooted or wedged; the runs called them down.", - "DISTINGUISHING FACTS: liveness is a different bound from connect (ssh_config(5) ServerAliveInterval x ServerAliveCountMax); sshd answers keepalives itself, so a silent remote command is not cut off, only a path that returns nothing. Whether the leg BEGAN is read in band (spark_remote_leg_began_word on stdout), not from stderr. 'Host down' needs an observation this boundary does not hold, so the type has no HostDown arm: LegNeverBegan is dark-or-down UNDECIDED and names the readbacks that close it: first a probe of the host's fabric-rail address (192.168.110.x) from a same-group peer (rail answers -> LAN dark, host up), then uptime and the wlP9s9 journal. A second trigger, separate from the deadline one: an automatic rail-jump classifier with its own consumer, SUFFICIENT FOR LegNeverBegan to be split into LAN-dark-host-up vs rail-also-dark without a human.", + "DISTINGUISHING FACTS: liveness is a different bound from connect (ssh_config(5) ServerAliveInterval x ServerAliveCountMax); sshd answers keepalives itself, so a silent remote command is not cut off, only a path that returns nothing. Whether the leg's began-line ARRIVED is read in band (spark_remote_leg_began_word on stdout), not from stderr -- and its absence does not establish that nothing ran, because the marker and the command share one unacknowledged stream: so the arm is LegBeginUnobserved, projected to ArgvLegMayHaveRun, never to ArgvLegDidNotRun (review of gunbc#13204). 'Host down' needs an observation this boundary does not hold, so the type has no HostDown arm: LegBeginUnobserved is dark-or-down UNDECIDED and names the readbacks that close it: first a probe of the host's fabric-rail address (192.168.110.x) from a same-group peer (rail answers -> LAN dark, host up), then uptime and the wlP9s9 journal. Two further triggers, separate from the deadline one: a two-phase leg protocol (the client receives an acknowledgement before it sends the effect command), SUFFICIENT FOR a NeverBegan arm a consumer may retry without a readback; and an automatic rail-jump classifier with its own consumer, SUFFICIENT FOR LegBeginUnobserved to be split into LAN-dark-host-up vs rail-also-dark without a human.", "RUNG: found at 1 (a human read the hang and the hosts). Now 2: gunbc.fleet_known_hosts_anchor fleet_ssh_leg_deadline is the one row both fleet shapers derive their deadline words from and neither shaper takes an options parameter, and dag/test/claim/spark/spark_remote_leg_deadline_witness_test.dag pins the argv, the route through the password session, a zeroed-keepalive red control and the never-began reading. CEILING 3: the deadline as a required field of the operation every fleet ssh transport declares, so no ssh leg in the corpus (including extdeps.bmc.openbmc_password_ssh_transport and extdeps.ssh.password_session CopyFile, which still carry ConnectTimeout only) is expressible without it. NEXT-RUNG TRIGGER: those transport literals moved onto extdeps.ssh.client_options with the deadline derived, SUFFICIENT FOR no ssh argv in the corpus lacking a liveness bound.", ], diff --git a/dag/gunbc/recurring_failure_mode/an_unreached_effect_leg_reads_as_one_that_ran_and_refused.dag b/dag/gunbc/recurring_failure_mode/an_unreached_effect_leg_reads_as_one_that_ran_and_refused.dag index 3d9220fe979..9821591be7a 100644 --- a/dag/gunbc/recurring_failure_mode/an_unreached_effect_leg_reads_as_one_that_ran_and_refused.dag +++ b/dag/gunbc/recurring_failure_mode/an_unreached_effect_leg_reads_as_one_that_ran_and_refused.dag @@ -17,7 +17,7 @@ data an_unreached_effect_leg_reads_as_one_that_ran_and_refused: RecurringFailure "RESIDUE, STATED: a connection lost after the shell started but before its began-line crossed also reads Unreached. gunbc.spark.native_serving_apply therefore retries only idempotent stages: preflight and the front door read, preserve/commit/rollback are no-ops where this transaction's desired state already holds, and apply's unconditional restart is the module's declared epoch semantics.", - "RUNG: found at 1 (mitigatable -- a human could read the log and delete the files). Now 2 for the native serving arm: witnesses in dag/test/claim/spark/native_serving_resumable_apply_witness_test.dag pin the reading, the bounded retry (spark_native_unreached_leg_attempt_budget) and the leftover decision (spark_native_leftover_decision) with discriminating reds. CEILING 3: an outcome type every remote-effect leg must produce, so no consumer can match a leg result without facing the Unreached arm. NEXT-RUNG TRIGGER: the other SshSessionExecResult consumers that fold 255 into a refusal (gunbc.spark.model_snapshot_materialize spark_remote_run, gunbc.host_command_binding host_command_outcome_of's Unconfirmed arm) consuming one shared in-band leg reading, SUFFICIENT FOR no remote-effect consumer in the corpus to construct a ran-and-refused value from a leg that never began. PROGRESS (2026-10-04): gunbc.spark.model_snapshot_materialize spark_remote_run now reads the in-band began-line (spark_remote_leg_began_word) and returns RemoteUnreachable with reach LegNeverBegan for ssh's 255 with no began-line (see a_dark_management_lan_leg_hangs_and_reads_as_a_down_host); gunbc.host_command_binding host_command_outcome_of remains.", + "RUNG: found at 1 (mitigatable -- a human could read the log and delete the files). Now 2 for the native serving arm: witnesses in dag/test/claim/spark/native_serving_resumable_apply_witness_test.dag pin the reading, the bounded retry (spark_native_unreached_leg_attempt_budget) and the leftover decision (spark_native_leftover_decision) with discriminating reds. CEILING 3: an outcome type every remote-effect leg must produce, so no consumer can match a leg result without facing the Unreached arm. NEXT-RUNG TRIGGER: the other SshSessionExecResult consumers that fold 255 into a refusal (gunbc.spark.model_snapshot_materialize spark_remote_run, gunbc.host_command_binding host_command_outcome_of's Unconfirmed arm) consuming one shared in-band leg reading, SUFFICIENT FOR no remote-effect consumer in the corpus to construct a ran-and-refused value from a leg that never began. PROGRESS (2026-10-04): gunbc.spark.model_snapshot_materialize spark_remote_run now reads the in-band began-line (spark_remote_leg_began_word) and returns RemoteUnreachable with reach LegBeginUnobserved for ssh's 255 with no began-line received, projected to ArgvLegMayHaveRun, never ArgvLegDidNotRun: the marker and the command share one unacknowledged stream, so a missing marker does not establish that nothing ran (see a_dark_management_lan_leg_hangs_and_reads_as_a_down_host). The same residue applies to this row's own Unreached reading. gunbc.host_command_binding host_command_outcome_of remains.", ], evidence: [], diff --git a/dag/gunbc/spark/host_effect_quiescence.dag b/dag/gunbc/spark/host_effect_quiescence.dag index a7aa170980e..a2c2171d46f 100644 --- a/dag/gunbc/spark/host_effect_quiescence.dag +++ b/dag/gunbc/spark/host_effect_quiescence.dag @@ -145,13 +145,20 @@ type HostQuiescence // THE LEG IS SUPPLIED: production hands the fleet-ssh typed-argv leg to the host, a witness hands a // local one. The observer decides from exit codes and stdout only. +// THREE ARMS, BECAUSE "DID NOT RUN" IS A CLAIM. ArgvLegDidNotRun is admissible only when nothing was +// sent. A leg whose transport failed after the command may have reached the host -- the session went +// dark before any acknowledgement arrived -- is ArgvLegMayHaveRun: a reader reads it as UNREAD, and an +// effect consumer must read the host back before retrying a non-idempotent effect, never treat it as +// a leg that did not run. type ArgvRun = ArgvRan { exit_code: Int, stdout: String, stderr: String } | ArgvLegDidNotRun { cause: String } + | ArgvLegMayHaveRun { cause: String } fn observe_host_effect_quiescence(run: fn(List) -> ArgvRun, claim: EventId, host: HostIdentity, spec: HostEffectResidueSpec, observer: NonEmptyStr, at: EpochSecs) -> HostQuiescence { match run(pgrep_match_full_command_line_argv(pattern: spec.process_pattern)) { ArgvLegDidNotRun { cause: c } => HostQuiescenceUnread { host: host, cause: join(["the process scan did not run: ", c], "") } + ArgvLegMayHaveRun { cause: c } => HostQuiescenceUnread { host: host, cause: join(["the process scan's result was not received: ", c], "") } ArgvRan { exit_code: code, stdout: out, stderr: err } => match pgrep_scan_outcome(exit_code: code, stdout: out, stderr: err) { PgrepScanFailed { exit_code: c2, stderr: e2 } => HostQuiescenceUnread { host: host, cause: join(["the process scan failed (exit ", to_string(c2), "): ", e2], "") } @@ -162,6 +169,7 @@ fn observe_host_effect_quiescence(run: fn(List) -> ArgvRun, claim: Event Present { value: f } => match run(argv_words(command: docker_ps_running_command(filter: f))) { ArgvLegDidNotRun { cause: c } => HostQuiescenceUnread { host: host, cause: join(["docker ps did not run: ", c], "") } + ArgvLegMayHaveRun { cause: c } => HostQuiescenceUnread { host: host, cause: join(["docker ps's result was not received: ", c], "") } ArgvRan { exit_code: code2, stdout: out2, stderr: err2 } => if code2 != 0 { HostQuiescenceUnread { host: host, cause: join(["docker ps exited ", to_string(code2), ": ", trim(s: err2)], "") } diff --git a/dag/gunbc/spark/host_occupancy_admission.dag b/dag/gunbc/spark/host_occupancy_admission.dag index dbcda9d75a1..a53326982b3 100644 --- a/dag/gunbc/spark/host_occupancy_admission.dag +++ b/dag/gunbc/spark/host_occupancy_admission.dag @@ -13,7 +13,7 @@ import gunbc.compute.host_occupancy { read_host_occupancy, admit_host_occupancy, host_occupancy_refusal, host_occupant_wire, } import gunbc.spark.host_effect_quiescence { ArgvRun, ArgvRan, ArgvLegDidNotRun } -import gunbc.spark.model_snapshot_materialize { SparkMaterializeContext, RemoteRan, RemoteUnreachable, spark_remote_run } +import gunbc.spark.model_snapshot_materialize { SparkMaterializeContext, RemoteRan, RemoteUnreachable, spark_remote_run, spark_argv_run_of_unreachable } import gunbc.spark.pair_serving_realization { spark_pair_head_unit_name, spark_pair_worker_unit_name } import gunbc.spark.pair_serving_apply { spark_pair_retired_head_unit, spark_pair_retired_worker_unit } @@ -33,12 +33,14 @@ fn spark_serving_units() -> List { [spark_pair_head_unit_name, spark_pair_worker_unit_name, spark_pair_retired_head_unit, spark_pair_retired_worker_unit] } -// THE PRIVILEGED LEG THE V4.1 EFFECTS ALREADY HOLD, as an ArgvRun. A leg that did not reach the host -// is a leg that did not run -- the reader turns that into Unread, never into vacant. +// THE PRIVILEGED LEG THE V4.1 EFFECTS ALREADY HOLD, as an ArgvRun, through the one projection +// (gunbc.spark.model_snapshot_materialize spark_argv_run_of_unreachable): a leg that sent nothing did +// not run, a leg whose result never arrived may have run -- the reader turns both into Unread, never +// into vacant. fn spark_ctx_argv_run(ctx: SparkMaterializeContext) -> fn(List) -> ArgvRun { fn(argv) { match spark_remote_run(ctx: ctx, argv: argv) { - RemoteUnreachable { cause: c, reach: _ } => ArgvLegDidNotRun { cause: c } + RemoteUnreachable { cause: c, reach: r } => spark_argv_run_of_unreachable(cause: c, reach: r) RemoteRan { exit_code: code, stdout: out, stderr: err } => ArgvRan { exit_code: code, stdout: out, stderr: err } } } diff --git a/dag/gunbc/spark/model_snapshot_materialize.dag b/dag/gunbc/spark/model_snapshot_materialize.dag index 2a602a27f01..35080050647 100644 --- a/dag/gunbc/spark/model_snapshot_materialize.dag +++ b/dag/gunbc/spark/model_snapshot_materialize.dag @@ -5,7 +5,7 @@ import std.nat { Nat } import std.content_hash { content_hash_atom } import product.placement_supply { HostIdentity } import std.algebra { trim } -import gunbc.spark.host_effect_quiescence { ArgvRun, ArgvRan, ArgvLegDidNotRun } +import gunbc.spark.host_effect_quiescence { ArgvRun, ArgvRan, ArgvLegDidNotRun, ArgvLegMayHaveRun } import gunbc.fleet_bootstrap_principal_session { exec_as_fleet_principal_trusting_with_stdin, BootstrapCredentialFileHeld, @@ -63,41 +63,47 @@ type SparkRemoteRun = RemoteRan { exit_code: Int, stdout: String, stderr: String } | RemoteUnreachable { cause: String, reach: SparkLegUnreachability } -// WHY A LEG DID NOT RUN, TYPED, AND THE ARM THAT IS DELIBERATELY ABSENT. Every Spark reaches the -// executor only over its management Wi-Fi (wlP9s9). On 2026-10-03 that link went dark on srv8 and +// WHY A LEG'S RESULT DID NOT ARRIVE, TYPED, AND THE TWO CLAIMS IT DOES NOT MAKE. Every Spark reaches +// the executor only over its management Wi-Fi (wlP9s9). On 2026-10-03 that link went dark on srv8 and // Group B spark-3336 (NetworkManager left it DISCONNECTED after a failed 4-way handshake) and srv7 // stayed associated but forwarded nothing 05:33-06:53 EDT; none of the hosts rebooted or wedged, yet -// the runs reported them as down. From ONE ssh leg the executor cannot tell a dark LAN leg from a host -// that is off: both are a connect that times out or a session that stops answering. So there is no -// HostDown arm here -- this boundary holds no observation that grounds it, and minting one is the -// over-assertion DESIGN §4d forbids. What it CAN say, and now says by name: -// -// LegNotAttempted -- nothing was sent: no endpoint, an inexpressible command, a refused password -// prerequisite. A fact about this run, not about the host or the LAN. -// LegNeverBegan -- ssh exited with its own 255 and the remote shell never reported beginning -// (spark_remote_leg_began_word). The LAN leg is dark OR the host is down, -// UNDECIDED. The FIRST settling readback is a probe of the host's fabric-rail -// address (enp1s0f1np1, 192.168.110.x) from a same-group peer: rail answers -> -// LAN dark, host up; rail dark too -> host down or both links down, still -// undecided. Uptime and the wlP9s9 NetworkManager journal come after, since they -// need a session. This is how the 2026-10-03 incident was closed by hand, as -// LAN-dark every time. NEXT RUNG: an automatic rail-jump classifier with its own -// consumer turns LegNeverBegan into LanLegDark / RailAlsoDark; not built here. -// -// A leg that BEGAN and then lost its session is not here: it ran to an unknown point, so it stays -// RemoteRan with ssh's 255 -- retrying it as if nothing ran is the opposite, non-idempotent decision -// (gunbc.recurring_failure_mode an_unreached_effect_leg_reads_as_one_that_ran_and_refused). +// the runs reported them as down. +// +// LegNotAttempted -- nothing was sent: no endpoint, an inexpressible command, a refused password +// prerequisite. The only arm that may be read as "did not run". +// LegBeginUnobserved -- ssh exited with its own 255 and the remote shell's began-line +// (spark_remote_leg_began_word) never ARRIVED. That is not "never began": the +// marker and the command travel in one unacknowledged stream, so the path can +// go dark after the shell wrote the marker, or after the command started, +// before the marker reached the client. The command MAY HAVE RUN. +// +// FIRST CLAIM NOT MADE: "host down". From one leg the executor cannot tell a dark LAN leg from a host +// that is off (DESIGN §4d), so there is no HostDown arm. The FIRST settling readback is a probe of the +// host's fabric-rail address (enp1s0f1np1, 192.168.110.x) from a same-group peer: rail answers -> LAN +// dark, host up; rail dark too -> host down or both links down, still undecided. Uptime and the wlP9s9 +// NetworkManager journal come after, since they need a session. This is how the 2026-10-03 incident +// was closed by hand, as LAN-dark every time. +// +// SECOND CLAIM NOT MADE: "did not run". spark_remote_argv_run projects LegBeginUnobserved to +// ArgvLegMayHaveRun, never ArgvLegDidNotRun, so an effect consumer reads the host back before retrying +// a non-idempotent effect. +// +// NEXT RUNGS, not built here: (1) a two-phase protocol -- the client receives an acknowledgement before +// it sends the effect command -- after which a NeverBegan arm is supportable; (2) an automatic +// rail-jump classifier with its own consumer, splitting LegBeginUnobserved into LAN-dark-host-up vs +// rail-also-dark. type SparkLegUnreachability = LegNotAttempted { cause: String } - | LegNeverBegan { endpoint: String, stderr: String } + | LegBeginUnobserved { endpoint: String, stderr: String } fn spark_leg_unreachability_text(u: SparkLegUnreachability) -> String { match u { LegNotAttempted { cause: c } => c - LegNeverBegan { endpoint: e, stderr: err } => + LegBeginUnobserved { endpoint: e, stderr: err } => join([ - "the ssh leg to ", e, " never began (ssh exit 255, no remote began-line): management LAN leg dark ", - "or host down, UNDECIDED from this leg -- settle it first by probing the host's fabric-rail address from a same-group peer, then uptime and the wlP9s9 journal; stderr=", + "the ssh leg to ", e, " lost its session before the remote began-line arrived (ssh exit 255): the command MAY HAVE RUN; ", + "management LAN leg dark or host down, UNDECIDED from this leg -- settle it first by probing the host's fabric-rail address ", + "from a same-group peer, then uptime and the wlP9s9 journal; stderr=", err, ], "") } @@ -121,11 +127,11 @@ fn spark_remote_leg_command_stdout(stdout: String) -> String { } // THE READING OF ONE LANDED SESSION RESULT, pure so a witness can supply the result rather than a -// host. 255 with no began-line is the only case read as unreached; 255 WITH a began-line is a -// command that ran (or a session lost after it began) and is reported as ran. +// host. 255 with no began-line received is LegBeginUnobserved (may have run); 255 WITH a began-line +// is a command that ran (or a session lost after it began) and is reported as ran. fn spark_remote_run_of_session(endpoint: String, exit_code: Int, stdout: String, stderr: String) -> SparkRemoteRun { if exit_code == ssh_client_error_exit_status && !spark_remote_leg_reported_began(stdout: stdout) { - spark_remote_unreachable(u: LegNeverBegan { endpoint: endpoint, stderr: trim(s: stderr) }) + spark_remote_unreachable(u: LegBeginUnobserved { endpoint: endpoint, stderr: trim(s: stderr) }) } else { RemoteRan { exit_code: exit_code, stdout: spark_remote_leg_command_stdout(stdout: stdout), stderr: trim(s: stderr) } } @@ -840,9 +846,17 @@ fn spark_materialize_seed( // declares that container as host-effect residue, and the fleet agent cannot read the docker socket, // so the lane hands this leg to gunbc.spark.host_commitment release_host_effect_live_over / // host_effect_recover_over and its release observes with the privilege its effect used. +// THE PROJECTION, and the arm it must not take: only a leg that sent nothing is ArgvLegDidNotRun. +fn spark_argv_run_of_unreachable(cause: String, reach: SparkLegUnreachability) -> ArgvRun { + match reach { + LegNotAttempted { cause: _ } => ArgvLegDidNotRun { cause: cause } + LegBeginUnobserved { endpoint: _, stderr: _ } => ArgvLegMayHaveRun { cause: cause } + } +} + fn spark_remote_argv_run(ctx: SparkMaterializeContext, argv: List) -> ArgvRun { match spark_remote_run(ctx: ctx, argv: argv) { - RemoteUnreachable { cause: why, reach: _ } => ArgvLegDidNotRun { cause: why as String } + RemoteUnreachable { cause: why, reach: r } => spark_argv_run_of_unreachable(cause: why, reach: r) RemoteRan { exit_code: c, stdout: o, stderr: e } => ArgvRan { exit_code: c, stdout: o, stderr: e } } } diff --git a/dag/test/claim/spark/spark_remote_leg_deadline_witness_test.dag b/dag/test/claim/spark/spark_remote_leg_deadline_witness_test.dag index e846b861bdb..aeecd985c84 100644 --- a/dag/test/claim/spark/spark_remote_leg_deadline_witness_test.dag +++ b/dag/test/claim/spark/spark_remote_leg_deadline_witness_test.dag @@ -3,6 +3,7 @@ module test.claim.spark.spark_remote_leg_deadline_witness_test import std.types { String, Bool, List, NonEmptyStr } import std.measure { second, second_count } import extdeps.ssh.client_options { serialize_ssh_client_options } +import gunbc.spark.host_effect_quiescence { ArgvRun, ArgvRan, ArgvLegDidNotRun, ArgvLegMayHaveRun } import gunbc.fleet_known_hosts_anchor { FleetSshLegDeadline, fleet_ssh_leg_deadline, @@ -11,8 +12,8 @@ import gunbc.fleet_known_hosts_anchor { } import gunbc.spark.model_snapshot_materialize { SparkRemoteRun, RemoteRan, RemoteUnreachable, - LegNotAttempted, LegNeverBegan, - spark_remote_run_of_session, spark_remote_leg_began_word, + LegNotAttempted, LegBeginUnobserved, + spark_remote_run_of_session, spark_remote_leg_began_word, spark_argv_run_of_unreachable, } // THE DEADLINE THE SPARK LEGS CARRY, asserted as exact argv. Fleet-converge run 37156689444 sat ~1h @@ -50,18 +51,18 @@ test fn red_control_zeroed_keepalive_is_not_a_bound() -> Bool { } // THE CLASSIFIER, over supplied session results (DESIGN §3: the claim is about the reading, not the -// transport). 255 with no began-line is the only unreached reading. +// transport). 255 with no began-line received is the only begin-unobserved reading. -fn is_never_began(r: SparkRemoteRun) -> Bool { +fn is_begin_unobserved(r: SparkRemoteRun) -> Bool { match r { - RemoteUnreachable { cause: _, reach: LegNeverBegan { endpoint: _, stderr: _ } } => true + RemoteUnreachable { cause: _, reach: LegBeginUnobserved { endpoint: _, stderr: _ } } => true RemoteUnreachable { cause: _, reach: LegNotAttempted { cause: _ } } => false RemoteRan { exit_code: _, stdout: _, stderr: _ } => false } } -test fn w_dark_connect_reads_as_never_began_not_as_a_command_exit() -> Bool { - is_never_began(r: spark_remote_run_of_session( +test fn w_dark_connect_reads_as_begin_unobserved_not_as_a_command_exit() -> Bool { + is_begin_unobserved(r: spark_remote_run_of_session( endpoint: "192.168.1.232", exit_code: 255, stdout: "", @@ -69,10 +70,11 @@ test fn w_dark_connect_reads_as_never_began_not_as_a_command_exit() -> Bool { )) } -// The cause text names the undecided pair, never "host down" alone (DESIGN §4d). -test fn w_never_began_text_says_dark_or_down_undecided() -> Bool { +// The cause text names the undecided pair and that the command may have run -- never "host down" +// alone and never "did not run" (DESIGN §4d). +test fn w_begin_unobserved_text_says_may_have_run_and_dark_or_down_undecided() -> Bool { match spark_remote_run_of_session(endpoint: "192.168.1.232", exit_code: 255, stdout: "", stderr: "Timeout, server 192.168.1.232 not responding.") { - RemoteUnreachable { cause: c, reach: _ } => contains(c, "LAN leg dark or host down, UNDECIDED") + RemoteUnreachable { cause: c, reach: _ } => contains(c, "the command MAY HAVE RUN") && contains(c, "LAN leg dark or host down, UNDECIDED") && contains(c, "fabric-rail address") RemoteRan { exit_code: _, stdout: _, stderr: _ } => false } } @@ -93,3 +95,30 @@ test fn w_began_line_never_reaches_a_consumer() -> Bool { RemoteUnreachable { cause: _, reach: _ } => false } } + +// THE PROJECTION THE REVIEW OF gunbc#13204 REQUIRED. A begin-unobserved leg may have run, so it +// projects to ArgvLegMayHaveRun; only a leg that sent nothing is ArgvLegDidNotRun. Reverting the +// projection to "every unreachable did not run" reds the first claim. +fn is_may_have_run(r: ArgvRun) -> Bool { + match r { + ArgvLegMayHaveRun { cause: _ } => true + ArgvLegDidNotRun { cause: _ } => false + ArgvRan { exit_code: _, stdout: _, stderr: _ } => false + } +} + +fn is_did_not_run(r: ArgvRun) -> Bool { + match r { + ArgvLegDidNotRun { cause: _ } => true + ArgvLegMayHaveRun { cause: _ } => false + ArgvRan { exit_code: _, stdout: _, stderr: _ } => false + } +} + +test fn w_begin_unobserved_projects_to_may_have_run_never_did_not_run() -> Bool { + is_may_have_run(r: spark_argv_run_of_unreachable(cause: "c", reach: LegBeginUnobserved { endpoint: "h", stderr: "Timeout, server h not responding." })) +} + +test fn w_not_attempted_projects_to_did_not_run() -> Bool { + is_did_not_run(r: spark_argv_run_of_unreachable(cause: "no endpoint", reach: LegNotAttempted { cause: "no endpoint" })) +} From c1fecd270b72c3c089d0f39c9a88e56c95b77b2f Mon Sep 17 00:00:00 2001 From: Brian Searls Date: Sun, 4 Oct 2026 06:23:14 +0000 Subject: [PATCH 4/5] One began-line protocol: lift it to gunbc.spark.remote_leg_began, consumed by pair_serving_apply and spark_remote_run (review 75397) Co-Authored-By: Claude Opus 5.5 (1M context) --- ...lan_leg_hangs_and_reads_as_a_down_host.dag | 2 +- ..._leg_reads_as_one_that_ran_and_refused.dag | 2 +- .../spark/model_snapshot_materialize.dag | 22 +++------- dag/gunbc/spark/pair_serving_apply.dag | 20 +++------- dag/gunbc/spark/remote_leg_began.dag | 40 +++++++++++++++++++ ...e_serving_resumable_apply_witness_test.dag | 12 +++--- ...spark_remote_leg_deadline_witness_test.dag | 7 ++-- 7 files changed, 64 insertions(+), 41 deletions(-) create mode 100644 dag/gunbc/spark/remote_leg_began.dag diff --git a/dag/gunbc/recurring_failure_mode/a_dark_management_lan_leg_hangs_and_reads_as_a_down_host.dag b/dag/gunbc/recurring_failure_mode/a_dark_management_lan_leg_hangs_and_reads_as_a_down_host.dag index 127f9f38dfe..3306865725f 100644 --- a/dag/gunbc/recurring_failure_mode/a_dark_management_lan_leg_hangs_and_reads_as_a_down_host.dag +++ b/dag/gunbc/recurring_failure_mode/a_dark_management_lan_leg_hangs_and_reads_as_a_down_host.dag @@ -13,7 +13,7 @@ data a_dark_management_lan_leg_hangs_and_reads_as_a_down_host: RecurringFailureM "HARM: 2026-10-03, Spark Group A bring-up. Every Spark reaches the executor only over wlP9s9 (MediaTek mt7925e). Router band steering roamed the hosts; a failed 4-way handshake left NetworkManager DISCONNECTED ('no secrets: No agents were available') on srv8 and Group B spark-3336, and srv7 stayed associated but forwarded nothing 05:33-06:53 EDT (read on the hosts by valiant-crab-775). Fleet-converge run 37156689444 sat ~1h inside one ssh leg. No host rebooted or wedged; the runs called them down.", - "DISTINGUISHING FACTS: liveness is a different bound from connect (ssh_config(5) ServerAliveInterval x ServerAliveCountMax); sshd answers keepalives itself, so a silent remote command is not cut off, only a path that returns nothing. Whether the leg's began-line ARRIVED is read in band (spark_remote_leg_began_word on stdout), not from stderr -- and its absence does not establish that nothing ran, because the marker and the command share one unacknowledged stream: so the arm is LegBeginUnobserved, projected to ArgvLegMayHaveRun, never to ArgvLegDidNotRun (review of gunbc#13204). 'Host down' needs an observation this boundary does not hold, so the type has no HostDown arm: LegBeginUnobserved is dark-or-down UNDECIDED and names the readbacks that close it: first a probe of the host's fabric-rail address (192.168.110.x) from a same-group peer (rail answers -> LAN dark, host up), then uptime and the wlP9s9 journal. Two further triggers, separate from the deadline one: a two-phase leg protocol (the client receives an acknowledgement before it sends the effect command), SUFFICIENT FOR a NeverBegan arm a consumer may retry without a readback; and an automatic rail-jump classifier with its own consumer, SUFFICIENT FOR LegBeginUnobserved to be split into LAN-dark-host-up vs rail-also-dark without a human.", + "DISTINGUISHING FACTS: liveness is a different bound from connect (ssh_config(5) ServerAliveInterval x ServerAliveCountMax); sshd answers keepalives itself, so a silent remote command is not cut off, only a path that returns nothing. Whether the leg's began-line ARRIVED is read in band (gunbc.spark.remote_leg_began remote_leg_began_word on stdout, the one protocol gunbc.spark.pair_serving_apply also reads), not from stderr -- and its absence does not establish that nothing ran, because the marker and the command share one unacknowledged stream: so the arm is LegBeginUnobserved, projected to ArgvLegMayHaveRun, never to ArgvLegDidNotRun (review of gunbc#13204). 'Host down' needs an observation this boundary does not hold, so the type has no HostDown arm: LegBeginUnobserved is dark-or-down UNDECIDED and names the readbacks that close it: first a probe of the host's fabric-rail address (192.168.110.x) from a same-group peer (rail answers -> LAN dark, host up), then uptime and the wlP9s9 journal. Two further triggers, separate from the deadline one: a two-phase leg protocol (the client receives an acknowledgement before it sends the effect command), SUFFICIENT FOR a NeverBegan arm a consumer may retry without a readback; and an automatic rail-jump classifier with its own consumer, SUFFICIENT FOR LegBeginUnobserved to be split into LAN-dark-host-up vs rail-also-dark without a human.", "RUNG: found at 1 (a human read the hang and the hosts). Now 2: gunbc.fleet_known_hosts_anchor fleet_ssh_leg_deadline is the one row both fleet shapers derive their deadline words from and neither shaper takes an options parameter, and dag/test/claim/spark/spark_remote_leg_deadline_witness_test.dag pins the argv, the route through the password session, a zeroed-keepalive red control and the never-began reading. CEILING 3: the deadline as a required field of the operation every fleet ssh transport declares, so no ssh leg in the corpus (including extdeps.bmc.openbmc_password_ssh_transport and extdeps.ssh.password_session CopyFile, which still carry ConnectTimeout only) is expressible without it. NEXT-RUNG TRIGGER: those transport literals moved onto extdeps.ssh.client_options with the deadline derived, SUFFICIENT FOR no ssh argv in the corpus lacking a liveness bound.", ], diff --git a/dag/gunbc/recurring_failure_mode/an_unreached_effect_leg_reads_as_one_that_ran_and_refused.dag b/dag/gunbc/recurring_failure_mode/an_unreached_effect_leg_reads_as_one_that_ran_and_refused.dag index 9821591be7a..3a11145d1f1 100644 --- a/dag/gunbc/recurring_failure_mode/an_unreached_effect_leg_reads_as_one_that_ran_and_refused.dag +++ b/dag/gunbc/recurring_failure_mode/an_unreached_effect_leg_reads_as_one_that_ran_and_refused.dag @@ -17,7 +17,7 @@ data an_unreached_effect_leg_reads_as_one_that_ran_and_refused: RecurringFailure "RESIDUE, STATED: a connection lost after the shell started but before its began-line crossed also reads Unreached. gunbc.spark.native_serving_apply therefore retries only idempotent stages: preflight and the front door read, preserve/commit/rollback are no-ops where this transaction's desired state already holds, and apply's unconditional restart is the module's declared epoch semantics.", - "RUNG: found at 1 (mitigatable -- a human could read the log and delete the files). Now 2 for the native serving arm: witnesses in dag/test/claim/spark/native_serving_resumable_apply_witness_test.dag pin the reading, the bounded retry (spark_native_unreached_leg_attempt_budget) and the leftover decision (spark_native_leftover_decision) with discriminating reds. CEILING 3: an outcome type every remote-effect leg must produce, so no consumer can match a leg result without facing the Unreached arm. NEXT-RUNG TRIGGER: the other SshSessionExecResult consumers that fold 255 into a refusal (gunbc.spark.model_snapshot_materialize spark_remote_run, gunbc.host_command_binding host_command_outcome_of's Unconfirmed arm) consuming one shared in-band leg reading, SUFFICIENT FOR no remote-effect consumer in the corpus to construct a ran-and-refused value from a leg that never began. PROGRESS (2026-10-04): gunbc.spark.model_snapshot_materialize spark_remote_run now reads the in-band began-line (spark_remote_leg_began_word) and returns RemoteUnreachable with reach LegBeginUnobserved for ssh's 255 with no began-line received, projected to ArgvLegMayHaveRun, never ArgvLegDidNotRun: the marker and the command share one unacknowledged stream, so a missing marker does not establish that nothing ran (see a_dark_management_lan_leg_hangs_and_reads_as_a_down_host). The same residue applies to this row's own Unreached reading. gunbc.host_command_binding host_command_outcome_of remains.", + "RUNG: found at 1 (mitigatable -- a human could read the log and delete the files). Now 2 for the native serving arm: witnesses in dag/test/claim/spark/native_serving_resumable_apply_witness_test.dag pin the reading, the bounded retry (spark_native_unreached_leg_attempt_budget) and the leftover decision (spark_native_leftover_decision) with discriminating reds. CEILING 3: an outcome type every remote-effect leg must produce, so no consumer can match a leg result without facing the Unreached arm. NEXT-RUNG TRIGGER: the other SshSessionExecResult consumers that fold 255 into a refusal (gunbc.spark.model_snapshot_materialize spark_remote_run, gunbc.host_command_binding host_command_outcome_of's Unconfirmed arm) consuming one shared in-band leg reading, SUFFICIENT FOR no remote-effect consumer in the corpus to construct a ran-and-refused value from a leg that never began. PROGRESS (2026-10-04): gunbc.spark.model_snapshot_materialize spark_remote_run now consumes the ONE shared in-band reading, gunbc.spark.remote_leg_began (lifted out of gunbc.spark.pair_serving_apply, which consumes it too; review of gunbc#13204), and returns RemoteUnreachable with reach LegBeginUnobserved for ssh's 255 with no began-line received, projected to ArgvLegMayHaveRun, never ArgvLegDidNotRun: the marker and the command share one unacknowledged stream, so a missing marker does not establish that nothing ran (see a_dark_management_lan_leg_hangs_and_reads_as_a_down_host). The same residue applies to this row's own Unreached reading. gunbc.host_command_binding host_command_outcome_of remains.", ], evidence: [], diff --git a/dag/gunbc/spark/model_snapshot_materialize.dag b/dag/gunbc/spark/model_snapshot_materialize.dag index 35080050647..f35a62877a6 100644 --- a/dag/gunbc/spark/model_snapshot_materialize.dag +++ b/dag/gunbc/spark/model_snapshot_materialize.dag @@ -14,6 +14,7 @@ import gunbc.fleet_bootstrap_principal_session { } import extdeps.exec.command { shell_quote } import extdeps.ssh.session { ssh_client_error_exit_status } +import gunbc.spark.remote_leg_began { remote_leg_began_payload_line, remote_leg_reported_began, remote_leg_stdout_without_began } import gunbc.fleet_known_hosts_anchor { FleetOpenSshTrustPolicy, SshTarget, portable_remote_words, } @@ -72,7 +73,7 @@ type SparkRemoteRun // LegNotAttempted -- nothing was sent: no endpoint, an inexpressible command, a refused password // prerequisite. The only arm that may be read as "did not run". // LegBeginUnobserved -- ssh exited with its own 255 and the remote shell's began-line -// (spark_remote_leg_began_word) never ARRIVED. That is not "never began": the +// (gunbc.spark.remote_leg_began remote_leg_began_word) never ARRIVED. That is not "never began": the // marker and the command travel in one unacknowledged stream, so the path can // go dark after the shell wrote the marker, or after the command started, // before the marker reached the client. The command MAY HAVE RUN. @@ -113,27 +114,14 @@ fn spark_remote_unreachable(u: SparkLegUnreachability) -> SparkRemoteRun { RemoteUnreachable { cause: spark_leg_unreachability_text(u: u), reach: u } } -// THE IN-BAND BEGAN-LINE, the discriminator gunbc.command_runner ssh_exit_255_conflation_dissolution -// names instead of stderr-sniffing: the remote shell prints it before the command, on stdout, and it -// is stripped before any consumer reads stdout, so consumers see exactly the command's output. -data spark_remote_leg_began_word: String = "gunbc-spark-remote-leg-began" - -fn spark_remote_leg_reported_began(stdout: String) -> Bool { - length(filter(split(s: stdout, delimiter: "\n"), l => trim(s: l) == spark_remote_leg_began_word)) > 0 -} - -fn spark_remote_leg_command_stdout(stdout: String) -> String { - trim(s: join(filter(split(s: stdout, delimiter: "\n"), l => trim(s: l) != spark_remote_leg_began_word), "\n")) -} - // THE READING OF ONE LANDED SESSION RESULT, pure so a witness can supply the result rather than a // host. 255 with no began-line received is LegBeginUnobserved (may have run); 255 WITH a began-line // is a command that ran (or a session lost after it began) and is reported as ran. fn spark_remote_run_of_session(endpoint: String, exit_code: Int, stdout: String, stderr: String) -> SparkRemoteRun { - if exit_code == ssh_client_error_exit_status && !spark_remote_leg_reported_began(stdout: stdout) { + if exit_code == ssh_client_error_exit_status && !remote_leg_reported_began(stdout: stdout) { spark_remote_unreachable(u: LegBeginUnobserved { endpoint: endpoint, stderr: trim(s: stderr) }) } else { - RemoteRan { exit_code: exit_code, stdout: spark_remote_leg_command_stdout(stdout: stdout), stderr: trim(s: stderr) } + RemoteRan { exit_code: exit_code, stdout: remote_leg_stdout_without_began(stdout: stdout), stderr: trim(s: stderr) } } } @@ -165,7 +153,7 @@ fn spark_remote_run(ctx: SparkMaterializeContext, argv: List) -> SparkRe principal: fleet_bootstrap_principal_login(), }, words: words, - stdin_payload: join([ctx.admin_credential as String, "\n", "echo ", spark_remote_leg_began_word, "\n", shell_command_line(argv: argv), "\n"], ""), + stdin_payload: join([ctx.admin_credential as String, "\n", remote_leg_began_payload_line(), shell_command_line(argv: argv), "\n"], ""), credential: BootstrapCredentialFileHeld { path: ctx.admin_credential_file }, ) { FleetPrincipalLegRefused { cause: why } => spark_remote_unreachable(u: LegNotAttempted { cause: why as String }) diff --git a/dag/gunbc/spark/pair_serving_apply.dag b/dag/gunbc/spark/pair_serving_apply.dag index 256fa3a9eb7..9783255df29 100644 --- a/dag/gunbc/spark/pair_serving_apply.dag +++ b/dag/gunbc/spark/pair_serving_apply.dag @@ -9,6 +9,7 @@ import product.placement_supply { HostIdentity } import extdeps.filesystem.filesystem_io { Filesystem } import extdeps.exec.command { ArgvCommand, shell_quote } import extdeps.ssh.session { SshSessionExecResult, ssh_client_error_exit_status } +import gunbc.spark.remote_leg_began { remote_leg_began_word, remote_leg_began_payload_line, remote_leg_lines, remote_leg_reported_began } import gunbc.fleet_bootstrap_principal_session { exec_as_fleet_principal_trusting_with_stdin, BootstrapCredentialFileHeld, @@ -256,7 +257,7 @@ fn spark_pair_apply_plan_refusals(plans: List) -> List the script RAN; its own exit decides Applied or Failed. @@ -267,7 +268,6 @@ fn spark_pair_apply_plan_refusals(plans: List) -> List Bool { // still reports it; the outer exit then mirrors it so ssh's own status keeps its meaning too. fn spark_pair_leg_payload(script: String) -> String { join([ - "echo ", spark_pair_leg_began_word, "\n", + remote_leg_began_payload_line(), "(\n", script, "\n)\n", "gunbc_leg_rc=$?\n", "echo ", spark_pair_leg_exit_prefix, "$gunbc_leg_rc\n", @@ -312,25 +312,17 @@ fn spark_pair_leg_payload(script: String) -> String { ], "") } -fn spark_pair_leg_lines(stdout: String) -> List { - map(split(s: stdout, delimiter: "\n"), l => trim(s: l)) -} - fn spark_pair_leg_reported_exit(stdout: String) -> Int? { - match first(filter(spark_pair_leg_lines(stdout: stdout), l => starts_with(s: l, prefix: spark_pair_leg_exit_prefix))) { + match first(filter(remote_leg_lines(stdout: stdout), l => starts_with(s: l, prefix: spark_pair_leg_exit_prefix))) { Absent => none Present { value: l } => parse_int(s: substring(s: l, start: length(spark_pair_leg_exit_prefix), end: length(l))) } } -fn spark_pair_leg_reported_began(stdout: String) -> Bool { - length(filter(spark_pair_leg_lines(stdout: stdout), l => l == spark_pair_leg_began_word)) > 0 -} - // THE SCRIPT'S OWN OUTPUT, WITHOUT THE WRAPPER'S TWO LINES, so readers of `detail` see what they saw // before the wrapper existed. fn spark_pair_leg_script_stdout(stdout: String) -> String { - trim(s: join(filter(spark_pair_leg_lines(stdout: stdout), l => l != spark_pair_leg_began_word && !starts_with(s: l, prefix: spark_pair_leg_exit_prefix)), "\n")) + trim(s: join(filter(remote_leg_lines(stdout: stdout), l => l != remote_leg_began_word && !starts_with(s: l, prefix: spark_pair_leg_exit_prefix)), "\n")) } // THE READING, PURE OVER ONE SESSION RESULT so a witness supplies the result at this boundary. @@ -347,7 +339,7 @@ fn spark_pair_leg_outcome_of(host: HostIdentity, wire: String, r: SshSessionExec SparkPairHostApplyFailed { host: host, cause: join([wire, " script-exit=", to_string(code), " stdout=", spark_pair_leg_script_stdout(stdout: r.stdout), " stderr=", trim(s: r.stderr)], "") } } Absent => - if spark_pair_leg_reported_began(stdout: r.stdout) { + if remote_leg_reported_began(stdout: r.stdout) { SparkPairHostApplyFailed { host: host, cause: join([wire, " the script BEGAN and its exit never arrived (ssh exit=", to_string(r.exit_code), "), so it ran to an unknown point and is not retried: stdout=", spark_pair_leg_script_stdout(stdout: r.stdout), " stderr=", trim(s: r.stderr)], "") } } else if r.exit_code == ssh_client_error_exit_status { SparkPairHostUnreached { host: host, cause: join([wire, " ssh exit=", to_string(r.exit_code), " stderr=", trim(s: r.stderr)], "") } diff --git a/dag/gunbc/spark/remote_leg_began.dag b/dag/gunbc/spark/remote_leg_began.dag new file mode 100644 index 00000000000..d9837a23d9a --- /dev/null +++ b/dag/gunbc/spark/remote_leg_began.dag @@ -0,0 +1,40 @@ +module gunbc.spark.remote_leg_began + +import std.types { String, Bool, List } +import std.algebra { trim } + +// THE IN-BAND BEGAN-LINE, ONE PROTOCOL FOR EVERY PRIVILEGED SPARK LEG. extdeps.ssh.session +// ssh_client_error_exit_status says why ssh's 255 alone does not establish that nothing ran, and +// gunbc.command_runner ssh_exit_255_conflation_dissolution rejects stderr-sniffing and names the richer +// source: the remote side reporting on stdout. So a privileged leg's payload prints this line before +// its command (remote_leg_began_payload_line), and a reader asks whether it ARRIVED. +// +// WHAT ITS ABSENCE DOES NOT ESTABLISH. The line and the command travel in one unacknowledged stream: a +// path can go dark after the remote shell wrote the line, or after the command started, before the line +// reaches the client. So "no began-line and ssh exited 255" means the start was UNOBSERVED, not that +// nothing ran. The consumers state what they do with that residue: gunbc.spark.pair_serving_apply +// retries only idempotent legs; gunbc.spark.model_snapshot_materialize reads it as LegBeginUnobserved, +// projected to ArgvLegMayHaveRun. A two-phase protocol (acknowledge before the command is sent) is the +// next rung, after which a never-began reading is supportable. +// +// It was two protocols with two marker strings (gunbc.spark.pair_serving_apply and +// gunbc.spark.model_snapshot_materialize); this is the one home both consume. +data remote_leg_began_word: String = "gunbc-privileged-leg-began" + +fn remote_leg_began_payload_line() -> String { + join(["echo ", remote_leg_began_word, "\n"], "") +} + +fn remote_leg_lines(stdout: String) -> List { + map(split(s: stdout, delimiter: "\n"), l => trim(s: l)) +} + +fn remote_leg_reported_began(stdout: String) -> Bool { + length(filter(remote_leg_lines(stdout: stdout), l => l == remote_leg_began_word)) > 0 +} + +// The command's own stdout, without the began-line, so a consumer reads exactly what the command +// printed. +fn remote_leg_stdout_without_began(stdout: String) -> String { + trim(s: join(filter(remote_leg_lines(stdout: stdout), l => l != remote_leg_began_word), "\n")) +} diff --git a/dag/test/claim/spark/native_serving_resumable_apply_witness_test.dag b/dag/test/claim/spark/native_serving_resumable_apply_witness_test.dag index 10b89d9106b..cbe2bf67afb 100644 --- a/dag/test/claim/spark/native_serving_resumable_apply_witness_test.dag +++ b/dag/test/claim/spark/native_serving_resumable_apply_witness_test.dag @@ -1,12 +1,14 @@ module test.claim.spark.native_serving_resumable_apply_witness_test +import gunbc.spark.remote_leg_began { remote_leg_began_word } + import std.types { Bool, String, NonEmptyStr, Int } import gunbc.fleet_intent_network { operator_host_srv9 } import extdeps.ssh.session { SshSessionExecResult } import gunbc.spark.pair_serving_apply { SparkPairHostApplyOutcome, SparkPairHostApplied, SparkPairHostApplyFailed, SparkPairHostUnreached, spark_pair_host_applied, spark_pair_host_unreached, spark_pair_leg_outcome_of, spark_pair_leg_payload, - spark_pair_leg_began_word, spark_pair_leg_exit_prefix, + spark_pair_leg_exit_prefix, } import gunbc.spark.pair_serving_launch_receipt { serving_standing } import gunbc.spark.native_serving_apply { @@ -56,7 +58,7 @@ test fn w_a_connect_timeout_with_no_began_line_reads_unreached() -> Bool { // Failed and never retry-eligible. test fn w_a_255_after_the_script_began_is_failed_not_unreached() -> Bool { let o = spark_pair_leg_outcome_of(host: operator_host_srv9, wire: "management-lan", r: SshSessionExecResult { - exit_code: 255, success: false, stdout: join([spark_pair_leg_began_word, "\nrestarted\n"], ""), + exit_code: 255, success: false, stdout: join([remote_leg_began_word, "\nrestarted\n"], ""), stderr: "Connection to 192.168.1.232 closed by remote host.", }) !spark_pair_host_unreached(o: o) && !spark_pair_host_applied(o: o) @@ -65,7 +67,7 @@ test fn w_a_255_after_the_script_began_is_failed_not_unreached() -> Bool { test fn w_a_script_that_ran_and_refused_is_failed_and_its_own_exit_is_read() -> Bool { let o = spark_pair_leg_outcome_of(host: operator_host_srv9, wire: "management-lan", r: SshSessionExecResult { exit_code: 1, success: false, - stdout: join([spark_pair_leg_began_word, "\nPREFLIGHT-REFUSED-image-absent\n", spark_pair_leg_exit_prefix, "1\n"], ""), + stdout: join([remote_leg_began_word, "\nPREFLIGHT-REFUSED-image-absent\n", spark_pair_leg_exit_prefix, "1\n"], ""), stderr: "", }) match o { @@ -78,7 +80,7 @@ test fn w_a_script_that_ran_and_refused_is_failed_and_its_own_exit_is_read() -> test fn w_an_applied_leg_detail_omits_the_wrapper_lines() -> Bool { match spark_pair_leg_outcome_of(host: operator_host_srv9, wire: "management-lan", r: SshSessionExecResult { exit_code: 0, success: true, - stdout: join([spark_pair_leg_began_word, "\npreflight-ok\n", spark_pair_leg_exit_prefix, "0\n"], ""), + stdout: join([remote_leg_began_word, "\npreflight-ok\n", spark_pair_leg_exit_prefix, "0\n"], ""), stderr: "", }) { SparkPairHostApplied { host: _, detail: d, standing: _ } => @@ -93,7 +95,7 @@ test fn w_an_applied_leg_detail_omits_the_wrapper_lines() -> Bool { // the exit line. test fn w_the_privileged_payload_reports_began_and_exit_around_the_script() -> Bool { let p = spark_pair_leg_payload(script: "set -e\necho preflight-ok\n") - string_contains(s: p, pattern: join(["echo ", spark_pair_leg_began_word, "\n(\nset -e\necho preflight-ok\n"], "")) + string_contains(s: p, pattern: join(["echo ", remote_leg_began_word, "\n(\nset -e\necho preflight-ok\n"], "")) && string_contains(s: p, pattern: join(["echo ", spark_pair_leg_exit_prefix, "$gunbc_leg_rc"], "")) && string_contains(s: p, pattern: "exit $gunbc_leg_rc") } diff --git a/dag/test/claim/spark/spark_remote_leg_deadline_witness_test.dag b/dag/test/claim/spark/spark_remote_leg_deadline_witness_test.dag index aeecd985c84..441219e02f8 100644 --- a/dag/test/claim/spark/spark_remote_leg_deadline_witness_test.dag +++ b/dag/test/claim/spark/spark_remote_leg_deadline_witness_test.dag @@ -3,6 +3,7 @@ module test.claim.spark.spark_remote_leg_deadline_witness_test import std.types { String, Bool, List, NonEmptyStr } import std.measure { second, second_count } import extdeps.ssh.client_options { serialize_ssh_client_options } +import gunbc.spark.remote_leg_began { remote_leg_began_word } import gunbc.spark.host_effect_quiescence { ArgvRun, ArgvRan, ArgvLegDidNotRun, ArgvLegMayHaveRun } import gunbc.fleet_known_hosts_anchor { FleetSshLegDeadline, @@ -13,7 +14,7 @@ import gunbc.fleet_known_hosts_anchor { import gunbc.spark.model_snapshot_materialize { SparkRemoteRun, RemoteRan, RemoteUnreachable, LegNotAttempted, LegBeginUnobserved, - spark_remote_run_of_session, spark_remote_leg_began_word, spark_argv_run_of_unreachable, + spark_remote_run_of_session, spark_argv_run_of_unreachable, } // THE DEADLINE THE SPARK LEGS CARRY, asserted as exact argv. Fleet-converge run 37156689444 sat ~1h @@ -82,7 +83,7 @@ test fn w_begin_unobserved_text_says_may_have_run_and_dark_or_down_undecided() - // A session lost AFTER the began-line ran to an unknown point: it stays a ran result with ssh's 255, // never an unreached one a caller could retry as if nothing happened. test fn w_session_lost_after_begin_stays_ran() -> Bool { - match spark_remote_run_of_session(endpoint: "h", exit_code: 255, stdout: join([spark_remote_leg_began_word, "partial"], "\n"), stderr: "Timeout, server h not responding.") { + match spark_remote_run_of_session(endpoint: "h", exit_code: 255, stdout: join([remote_leg_began_word, "partial"], "\n"), stderr: "Timeout, server h not responding.") { RemoteRan { exit_code: c, stdout: o, stderr: _ } => c == 255 && o == "partial" RemoteUnreachable { cause: _, reach: _ } => false } @@ -90,7 +91,7 @@ test fn w_session_lost_after_begin_stays_ran() -> Bool { // The began-line is stripped, so every consumer reads exactly the command's stdout. test fn w_began_line_never_reaches_a_consumer() -> Bool { - match spark_remote_run_of_session(endpoint: "h", exit_code: 0, stdout: join([spark_remote_leg_began_word, "active"], "\n"), stderr: "") { + match spark_remote_run_of_session(endpoint: "h", exit_code: 0, stdout: join([remote_leg_began_word, "active"], "\n"), stderr: "") { RemoteRan { exit_code: c, stdout: o, stderr: _ } => c == 0 && o == "active" RemoteUnreachable { cause: _, reach: _ } => false } From 3b361bafcca787a249e2cf76640fbfcb01bc2507 Mon Sep 17 00:00:00 2001 From: Brian Searls Date: Sun, 4 Oct 2026 06:50:24 +0000 Subject: [PATCH 5/5] remote_leg_began: import filter and length (floor unimported-bare-provider gate) Co-Authored-By: Claude Opus 5.5 (1M context) --- dag/gunbc/spark/remote_leg_began.dag | 1 + 1 file changed, 1 insertion(+) diff --git a/dag/gunbc/spark/remote_leg_began.dag b/dag/gunbc/spark/remote_leg_began.dag index d9837a23d9a..f885ccbaca8 100644 --- a/dag/gunbc/spark/remote_leg_began.dag +++ b/dag/gunbc/spark/remote_leg_began.dag @@ -2,6 +2,7 @@ module gunbc.spark.remote_leg_began import std.types { String, Bool, List } import std.algebra { trim } +import v2.std.algebra { filter, length } // THE IN-BAND BEGAN-LINE, ONE PROTOCOL FOR EVERY PRIVILEGED SPARK LEG. extdeps.ssh.session // ssh_client_error_exit_status says why ssh's 255 alone does not establish that nothing ran, and