ck, daemon: say when a module is running something other than what its config asks for (#108)

crates/ck-bus/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ck-bus"
-version = "0.1.11"
+version = "0.1.13"
 edition = "2021"
 publish = false
 description = "Supervised owner of the CortexKit NATS message plane."
@@ -28,15 +28,15 @@ data-encoding = "=2.11.1"
 nkeys = "=0.4.5"
 serde_json = "1"
 sha2 = "0.10"
-subc-client-rs = { path = "../subc-client-rs", version = "0.18" }
+subc-client-rs = { path = "../subc-client-rs", version = "0.19" }
 subc-protocol = { path = "../subc-protocol", version = "0.25.0" }
 tokio = { version = "1", features = ["macros", "process", "rt-multi-thread", "sync", "time"] }
 
 [dev-dependencies]
 futures-util = "0.3"
 tempfile = "3"
 tokio = { version = "1", features = ["io-util", "net"] }
-subc-control = { path = "../subc-control", version = "0.19" }
+subc-control = { path = "../subc-control", version = "0.20" }
 subc-daemon = { path = "../subc-daemon", version = "0.21.0", features = ["test-support"] }
 subc-jsonc = { path = "../subc-jsonc", version = "0.1.0" }
 subc-transport = { path = "../subc-transport", version = "0.7.0" }

crates/subc-client-rs/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "subc-client-rs"
-version = "0.18.5"
+version = "0.19.0"
 edition = "2021"
 publish = true
 description = "Shared serve + consume client for Rust subc modules."
@@ -11,7 +11,7 @@ repository = "https://github.com/cortexkit/subconscious"
 async-trait = "0.1"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
-subc-control = { path = "../subc-control", version = "0.19" }
+subc-control = { path = "../subc-control", version = "0.20" }
 subc-protocol = { path = "../subc-protocol", version = "0.25.0" }
 subc-transport = { path = "../subc-transport", version = "0.7" }
 tokio = { version = "1", features = ["io-util", "macros", "net", "rt", "sync", "time"] }

crates/subc-control/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "subc-control"
-version = "0.19.0"
+version = "0.20.0"
 edition = "2021"
 publish = true
 description = "Client-facing subc control-plane wire shapes."

crates/subc-control/src/lib.rs

@@ -609,6 +609,39 @@ pub struct SupervisorObservedProcess {
     pub running_image: RunningImageAgreement,
 }
 
+/// Independent comparisons of configured path and running image at list time.
+/// An absent verdict means the daemon predates this field, not agreement.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+pub struct PendingReloadVerdict {
+    pub path: ReloadPathAgreement,
+    pub image: RunningImageAgreement,
+}
+
+/// Whether the running process was spawned from the currently configured program.
+#[derive(Debug, Clone, PartialEq)]
+pub enum ReloadPathAgreement {
+    Match,
+    Mismatch {
+        configured: PathBuf,
+        spawned_from: PathBuf,
+    },
+    Unavailable {
+        reason: ReloadPathUnavailableReason,
+    },
+    Unknown {
+        tag: String,
+        body: OrderedJsonObject,
+    },
+}
+
+open_string_enum! {
+    /// Why configured and spawned paths cannot be compared.
+    ReloadPathUnavailableReason {
+        NotRunning => "not_running",
+        SpawnedPathUnavailable => "spawned_path_unavailable",
+    }
+}
+
 /// Daemon provenance paired with its runtime process observation.
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 pub struct SupervisorDaemonProvenance {
@@ -779,6 +812,19 @@ enum RunningImageAgreementWire {
     },
 }
 
+#[derive(Debug, Serialize, Deserialize)]
+#[serde(tag = "status", rename_all = "snake_case")]
+enum ReloadPathAgreementWire {
+    Match,
+    Mismatch {
+        configured: PathBuf,
+        spawned_from: PathBuf,
+    },
+    Unavailable {
+        reason: ReloadPathUnavailableReason,
+    },
+}
+
 #[derive(Debug, Serialize, Deserialize)]
 #[serde(tag = "method", rename_all = "snake_case")]
 enum RunningImageEvidenceWire {
@@ -1185,6 +1231,61 @@ impl Serialize for RunningImageAgreement {
     }
 }
 
+impl Serialize for ReloadPathAgreement {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        match self {
+            Self::Match => ReloadPathAgreementWire::Match.serialize(serializer),
+            Self::Mismatch {
+                configured,
+                spawned_from,
+            } => ReloadPathAgreementWire::Mismatch {
+                configured: configured.clone(),
+                spawned_from: spawned_from.clone(),
+            }
+            .serialize(serializer),
+            Self::Unavailable { reason } => ReloadPathAgreementWire::Unavailable {
+                reason: reason.clone(),
+            }
+            .serialize(serializer),
+            Self::Unknown { body, .. } => body.serialize(serializer),
+        }
+    }
+}
+
+impl<'de> Deserialize<'de> for ReloadPathAgreement {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let (tag, body) = read_tagged(deserializer, "status")?;
+        match tag.as_str() {
+            "match" => Ok(Self::Match),
+            "mismatch" => {
+                match serde_json::from_value(body.into_value()).map_err(D::Error::custom)? {
+                    ReloadPathAgreementWire::Mismatch {
+                        configured,
+                        spawned_from,
+                    } => Ok(Self::Mismatch {
+                        configured,
+                        spawned_from,
+                    }),
+                    _ => unreachable!(),
+                }
+            }
+            "unavailable" => match serde_json::from_value(body.into_value())
+                .map_err(D::Error::custom)?
+            {
+                ReloadPathAgreementWire::Unavailable { reason } => Ok(Self::Unavailable { reason }),
+                _ => unreachable!(),
+            },
+            _ => Ok(Self::Unknown { tag, body }),
+        }
+    }
+}
+
 impl<'de> Deserialize<'de> for RunningImageAgreement {
     fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
     where
@@ -1708,7 +1809,7 @@ pub enum ModuleProtocol {
     None,
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 pub struct SupervisorEntry {
     pub module_id: String,
     pub state: String,
@@ -1729,6 +1830,10 @@ pub struct SupervisorEntry {
     #[serde(default)]
     pub protocol: ModuleProtocol,
     pub health: SupervisorHealthStatus,
+    /// Computed from the stored launch spec and observed process at list time;
+    /// None means an older daemon did not report this comparison.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub pending_reload: Option<PendingReloadVerdict>,
     /// When the daemon last collected this module's health, as unix
     /// milliseconds. Absent means NEVER PROBED (a module inside its first probe
     /// window, whose `health` is therefore `Unknown` rather than good), not

crates/subc-control/tests/golden/supervisor_entry_reload_verdict_states.json

@@ -0,0 +1,102 @@
+[
+  {
+    "enabled": true,
+    "health": "degraded",
+    "last_exit_ms": 1700000000123,
+    "last_probe_ms": 1700000000000,
+    "live": true,
+    "max_restarts": 3,
+    "module_id": "aft-tools",
+    "pending_reload": {
+      "image": {
+        "evidence": {
+          "digest": "old",
+          "method": "linux_proc_sha256"
+        },
+        "status": "match"
+      },
+      "path": {
+        "configured": "/bin/new",
+        "spawned_from": "/bin/old",
+        "status": "mismatch"
+      }
+    },
+    "protocol": "subc",
+    "restart_count": 2,
+    "state": "running"
+  },
+  {
+    "enabled": true,
+    "health": "degraded",
+    "last_exit_ms": 1700000000123,
+    "last_probe_ms": 1700000000000,
+    "live": true,
+    "max_restarts": 3,
+    "module_id": "aft-tools",
+    "pending_reload": {
+      "image": {
+        "disk": {
+          "digest": "new",
+          "method": "linux_proc_sha256"
+        },
+        "running": {
+          "digest": "old",
+          "method": "linux_proc_sha256"
+        },
+        "status": "mismatch"
+      },
+      "path": {
+        "status": "match"
+      }
+    },
+    "protocol": "subc",
+    "restart_count": 2,
+    "state": "running"
+  },
+  {
+    "enabled": true,
+    "health": "degraded",
+    "last_exit_ms": 1700000000123,
+    "last_probe_ms": 1700000000000,
+    "live": true,
+    "max_restarts": 3,
+    "module_id": "aft-tools",
+    "pending_reload": {
+      "image": {
+        "reason": "not_running",
+        "status": "unavailable"
+      },
+      "path": {
+        "reason": "not_running",
+        "status": "unavailable"
+      }
+    },
+    "protocol": "subc",
+    "restart_count": 2,
+    "state": "running"
+  },
+  {
+    "enabled": true,
+    "health": "degraded",
+    "last_exit_ms": 1700000000123,
+    "last_probe_ms": 1700000000000,
+    "live": true,
+    "max_restarts": 3,
+    "module_id": "aft-tools",
+    "pending_reload": {
+      "image": {
+        "evidence": {
+          "digest": "same",
+          "method": "linux_proc_sha256"
+        },
+        "status": "match"
+      },
+      "path": {
+        "status": "match"
+      }
+    },
+    "protocol": "subc",
+    "restart_count": 2,
+    "state": "running"
+  }
+]