use core::fmt;
use serde::{Deserialize, Deserializer, Serialize};
use crate::config::AppConfig;
use crate::status::ProcStatus;
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Hello {
pub client_version: String,
pub protocol: u32,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct HelloAck {
pub daemon_version: String,
pub protocol: u32,
pub pid: u32,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", content = "value", rename_all = "snake_case")]
pub enum SelectorSpec {
All,
Id(u32),
Name(String),
Regex(String),
Fold(String),
Instance {
name: String,
slot: u32,
},
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct Smit(String);
impl Smit {
pub const MAX_CHARS: usize = 48;
#[must_use]
pub fn as_str(&self) -> &str {
&self.0
}
}
impl fmt::Display for Smit {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(&self.0)
}
}
impl core::str::FromStr for Smit {
type Err = SmitError;
fn from_str(text: &str) -> Result<Self, Self::Err> {
if text.trim().is_empty() {
return Err(SmitError::Empty);
}
let chars = text.chars().count();
if chars > Self::MAX_CHARS {
return Err(SmitError::TooLong { chars });
}
if text.chars().any(char::is_control) {
return Err(SmitError::Unprintable);
}
Ok(Self(text.to_string()))
}
}
impl<'de> Deserialize<'de> for Smit {
fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
let text = String::deserialize(deserializer)?;
text.parse().map_err(serde::de::Error::custom)
}
}
#[non_exhaustive]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SmitError {
TooLong {
chars: usize,
},
Unprintable,
Empty,
}
impl fmt::Display for SmitError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::TooLong { chars } => write!(
f,
"a smit is at most {} characters; this one is {chars}",
Smit::MAX_CHARS
),
Self::Unprintable => {
f.write_str("a smit may not contain a control character, an escape included")
}
Self::Empty => f.write_str("a smit may not be empty"),
}
}
}
impl core::error::Error for SmitError {}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
#[non_exhaustive]
pub enum Request {
Ping,
ListFlock,
Describe {
selector: SelectorSpec,
},
Start {
apps: Vec<AppConfig>,
},
ConfigDrift {
apps: Vec<AppConfig>,
},
Stop {
selector: SelectorSpec,
},
Restart {
selector: SelectorSpec,
},
Reload {
selector: SelectorSpec,
},
Delete {
selector: SelectorSpec,
},
Scale {
name: String,
count: u32,
},
SetSmit {
sheep: String,
smit: Option<Smit>,
},
Reopen {
selector: SelectorSpec,
},
Flush {
selector: SelectorSpec,
},
Trigger {
selector: SelectorSpec,
action: String,
params: Option<String>,
},
Signal {
selector: SelectorSpec,
signal: String,
},
SendLine {
selector: SelectorSpec,
line: String,
},
SaveRoll,
Muster,
DogConfig {
name: String,
},
EnableDog {
name: String,
source: DogSource,
},
DisableDog {
name: String,
},
KillDaemon,
Subscribe {
topics: Vec<String>,
},
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
#[non_exhaustive]
pub enum DogSource {
BuiltIn,
Adopted {
path: String,
},
}
#[non_exhaustive]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Lamb {
pub pid: u32,
pub name: String,
}
impl Lamb {
#[must_use]
pub fn new(pid: u32, name: impl Into<String>) -> Self {
Self {
pid,
name: name.into(),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct ExitInfo {
pub code: Option<i32>,
pub signal: Option<i32>,
}
#[non_exhaustive]
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ProcessInfo {
pub id: u32,
pub name: String,
pub status: ProcStatus,
pub pid: Option<u32>,
pub restarts: u32,
pub uptime_ms: u64,
pub fold: Option<String>,
pub out_file: Option<String>,
pub err_file: Option<String>,
pub cpu_percent: Option<f32>,
pub memory_bytes: Option<u64>,
pub dog: Option<DogSource>,
pub lambs: Option<Vec<Lamb>>,
pub last_exit: Option<ExitInfo>,
pub smit: Option<String>,
pub instance: Option<u32>,
}
pub fn sort_flock(listing: &mut [ProcessInfo]) {
listing.sort_unstable_by(|a, b| {
(a.name.as_str(), a.instance, a.id).cmp(&(b.name.as_str(), b.instance, b.id))
});
}
impl ProcessInfo {
pub fn builder(id: u32, name: impl Into<String>, status: ProcStatus) -> ProcessInfoBuilder {
ProcessInfoBuilder {
info: Self {
id,
name: name.into(),
status,
pid: None,
restarts: 0,
uptime_ms: 0,
fold: None,
out_file: None,
err_file: None,
cpu_percent: None,
memory_bytes: None,
dog: None,
lambs: None,
last_exit: None,
smit: None,
instance: None,
},
}
}
}
#[derive(Debug, Clone)]
#[must_use = "a builder that is never `build`-ed produces no ProcessInfo"]
pub struct ProcessInfoBuilder {
info: ProcessInfo,
}
impl ProcessInfoBuilder {
pub fn pid(mut self, pid: Option<u32>) -> Self {
self.info.pid = pid;
self
}
pub fn restarts(mut self, restarts: u32) -> Self {
self.info.restarts = restarts;
self
}
pub fn uptime_ms(mut self, uptime_ms: u64) -> Self {
self.info.uptime_ms = uptime_ms;
self
}
pub fn fold(mut self, fold: Option<String>) -> Self {
self.info.fold = fold;
self
}
pub fn out_file(mut self, out_file: Option<String>) -> Self {
self.info.out_file = out_file;
self
}
pub fn err_file(mut self, err_file: Option<String>) -> Self {
self.info.err_file = err_file;
self
}
pub fn cpu_percent(mut self, cpu_percent: Option<f32>) -> Self {
self.info.cpu_percent = cpu_percent;
self
}
pub fn memory_bytes(mut self, memory_bytes: Option<u64>) -> Self {
self.info.memory_bytes = memory_bytes;
self
}
pub fn dog(mut self, dog: Option<DogSource>) -> Self {
self.info.dog = dog;
self
}
pub fn lambs(mut self, lambs: Option<Vec<Lamb>>) -> Self {
self.info.lambs = lambs;
self
}
pub fn last_exit(mut self, last_exit: Option<ExitInfo>) -> Self {
self.info.last_exit = last_exit;
self
}
pub fn smit(mut self, smit: Option<String>) -> Self {
self.info.smit = smit;
self
}
pub fn instance(mut self, instance: Option<u32>) -> Self {
self.info.instance = instance;
self
}
#[must_use]
pub fn build(self) -> ProcessInfo {
self.info
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
#[non_exhaustive]
pub enum ActionOutcome {
Replied {
body: String,
},
NoChannel,
Skipped,
TimedOut,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ActionReply {
pub id: u32,
pub name: String,
pub outcome: ActionOutcome,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
#[non_exhaustive]
pub enum SignalOutcome {
Delivered,
NotRunning,
Failed {
reason: String,
},
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SignalReply {
pub id: u32,
pub name: String,
pub outcome: SignalOutcome,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
#[non_exhaustive]
pub enum LineOutcome {
Sent,
NoStdin,
NotWritten {
reason: String,
},
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct LineReply {
pub id: u32,
pub name: String,
pub outcome: LineOutcome,
}
#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(transparent)]
pub struct DogSectionToml(String);
impl DogSectionToml {
#[must_use]
pub fn as_str(&self) -> &str {
&self.0
}
}
impl From<String> for DogSectionToml {
fn from(toml: String) -> Self {
Self(toml)
}
}
impl core::ops::Deref for DogSectionToml {
type Target = str;
fn deref(&self) -> &str {
&self.0
}
}
impl fmt::Debug for DogSectionToml {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "DogSectionToml(<{} bytes>)", self.0.len())
}
}
#[non_exhaustive]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SheepDrift {
pub name: String,
pub fields: Vec<String>,
}
impl SheepDrift {
#[must_use]
pub fn new(name: impl Into<String>, fields: Vec<String>) -> Self {
Self {
name: name.into(),
fields,
}
}
}
#[allow(clippy::large_enum_variant)]
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", content = "data", rename_all = "snake_case")]
#[non_exhaustive]
pub enum Response {
Pong,
Flock(Vec<ProcessInfo>),
Described(Vec<ProcessInfo>),
Started(Vec<ProcessInfo>),
Drifted(Vec<SheepDrift>),
Stopped(Vec<ProcessInfo>),
Restarted(Vec<ProcessInfo>),
Reloading(Vec<ProcessInfo>),
Scaled(Vec<ProcessInfo>),
SmitPainted(Vec<ProcessInfo>),
Deleted(Vec<u32>),
Reopened(Vec<ProcessInfo>),
Flushed(Vec<ProcessInfo>),
Triggered(Vec<ActionReply>),
Signalled(Vec<SignalReply>),
SentLine(Vec<LineReply>),
RollSaved {
path: String,
apps: u32,
},
Mustered(Vec<ProcessInfo>),
DogSection {
toml: DogSectionToml,
},
DogStarted(ProcessInfo),
Subscribed,
ShuttingDown,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Envelope {
pub id: u64,
pub deadline_ms: Option<u64>,
pub body: Request,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Reply {
pub id: u64,
pub result: Result<Response, RpcError>,
}
pub type HelloReply = Result<HelloAck, RpcError>;
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RpcError {
pub code: RpcErrorCode,
pub message: String,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum RpcErrorCode {
NotFound,
InvalidConfig,
SpawnFailed,
ProtocolMismatch,
Internal,
DeadlineExceeded,
}
impl RpcErrorCode {
pub const ALL: [Self; 6] = [
Self::NotFound,
Self::InvalidConfig,
Self::SpawnFailed,
Self::ProtocolMismatch,
Self::Internal,
Self::DeadlineExceeded,
];
#[allow(dead_code)]
const fn assert_all_lists_every_variant(code: Self) -> Self {
match code {
Self::NotFound => Self::ALL[0],
Self::InvalidConfig => Self::ALL[1],
Self::SpawnFailed => Self::ALL[2],
Self::ProtocolMismatch => Self::ALL[3],
Self::Internal => Self::ALL[4],
Self::DeadlineExceeded => Self::ALL[5],
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::config::AppConfig;
use crate::protocol::PROTOCOL_VERSION;
use crate::status::ProcStatus;
fn sample_info() -> ProcessInfo {
ProcessInfo {
id: 3,
name: "web".to_string(),
status: ProcStatus::Online,
pid: Some(4242),
restarts: 1,
uptime_ms: 60_000,
fold: Some("backend".to_string()),
out_file: Some("/home/ada/.shep/logs/web-0-out.log".to_string()),
err_file: Some("/home/ada/.shep/logs/web-0-err.log".to_string()),
cpu_percent: Some(12.5),
memory_bytes: Some(48 * 1024 * 1024),
dog: None,
lambs: None,
last_exit: Some(ExitInfo {
code: Some(1),
signal: None,
}),
smit: None,
instance: None,
}
}
#[test]
fn a_builder_with_nothing_set_is_a_sheep_that_has_not_run() {
let info = ProcessInfo::builder(3, "web", ProcStatus::Stopped).build();
assert_eq!(info.id, 3);
assert_eq!(info.name, "web");
assert_eq!(info.status, ProcStatus::Stopped);
assert_eq!(info.pid, None);
assert_eq!(info.restarts, 0);
assert_eq!(info.uptime_ms, 0);
assert_eq!(info.fold, None);
assert_eq!(info.out_file, None);
assert_eq!(info.err_file, None);
assert_eq!(info.cpu_percent, None);
assert_eq!(info.memory_bytes, None);
assert_eq!(info.dog, None);
assert_eq!(info.lambs, None);
assert_eq!(info.last_exit, None);
}
#[test]
fn every_setter_writes_its_own_field_and_no_other() {
let built = ProcessInfo::builder(3, "web", ProcStatus::Online)
.pid(Some(4242))
.restarts(1)
.uptime_ms(60_000)
.fold(Some("backend".to_string()))
.out_file(Some("/home/ada/.shep/logs/web-0-out.log".to_string()))
.err_file(Some("/home/ada/.shep/logs/web-0-err.log".to_string()))
.cpu_percent(Some(12.5))
.memory_bytes(Some(48 * 1024 * 1024))
.dog(None)
.last_exit(Some(ExitInfo {
code: Some(1),
signal: None,
}))
.build();
assert_eq!(built, sample_info());
assert_eq!(
ProcessInfo::builder(1, "metrics", ProcStatus::Online)
.dog(Some(DogSource::BuiltIn))
.build()
.dog,
Some(DogSource::BuiltIn),
"an empty `dog` setter body is invisible to the comparison above"
);
assert_eq!(
ProcessInfo::builder(1, "web", ProcStatus::Online)
.lambs(Some(vec![Lamb::new(4243, "node")]))
.build()
.lambs,
Some(vec![Lamb::new(4243, "node")]),
"an empty `lambs` setter body is invisible to the comparison above"
);
assert_eq!(
ProcessInfo::builder(1, "web", ProcStatus::Online)
.smit(Some("\u{25b2} main@a1b2c3".to_string()))
.build()
.smit
.as_deref(),
Some("\u{25b2} main@a1b2c3"),
"an empty `smit` setter body is invisible to the comparison above"
);
}
#[test]
fn lambs_distinguishes_not_walked_from_walked_and_empty() {
let not_walked = ProcessInfo::builder(1, "web", ProcStatus::Online).build();
assert_eq!(not_walked.lambs, None);
let walked_empty = ProcessInfo::builder(1, "web", ProcStatus::Online)
.lambs(Some(Vec::new()))
.build();
assert_eq!(walked_empty.lambs, Some(Vec::new()));
}
#[test]
fn a_process_info_without_a_lambs_key_still_deserializes() {
let fixture = r#"{
"id": 3, "name": "web", "status": "online", "pid": 4242,
"restarts": 0, "uptime_ms": 100, "fold": null,
"out_file": null, "err_file": null,
"cpu_percent": null, "memory_bytes": null, "dog": null
}"#;
let info: ProcessInfo = serde_json::from_str(fixture).unwrap();
assert_eq!(info.lambs, None);
}
#[test]
fn a_lamb_is_a_pid_and_an_executable_name() {
let lamb = Lamb::new(4243, "node");
let json = serde_json::to_string(&lamb).unwrap();
assert_eq!(json, r#"{"pid":4243,"name":"node"}"#);
assert_eq!(serde_json::from_str::<Lamb>(&json).unwrap(), lamb);
}
#[test]
fn a_dog_source_serializes_snake_case_under_its_kind() {
assert_eq!(
serde_json::to_string(&DogSource::BuiltIn).unwrap(),
r#"{"kind":"built_in"}"#
);
let adopted = DogSource::Adopted {
path: "/usr/local/bin/shep-otel".to_string(),
};
let wire = r#"{"kind":"adopted","path":"/usr/local/bin/shep-otel"}"#;
assert_eq!(serde_json::to_string(&adopted).unwrap(), wire);
assert_eq!(serde_json::from_str::<DogSource>(wire).unwrap(), adopted);
}
#[test]
fn v1_process_info_without_a_dog_marker_still_deserializes() {
let fixture = r#"{"id":3,"name":"web","status":"online","pid":4242,"restarts":1,"uptime_ms":60000,"fold":"backend","out_file":"/l/o.log","err_file":"/l/e.log","cpu_percent":12.5,"memory_bytes":50331648}"#;
let info: ProcessInfo = serde_json::from_str(fixture).unwrap();
assert_eq!(info.dog, None);
}
#[test]
fn a_process_info_without_a_last_exit_key_still_deserializes() {
let fixture = r#"{"id":3,"name":"web","status":"online","pid":4242,"restarts":1,"uptime_ms":60000,"fold":"backend","out_file":"/l/o.log","err_file":"/l/e.log","cpu_percent":12.5,"memory_bytes":50331648,"dog":null,"lambs":null}"#;
let info: ProcessInfo = serde_json::from_str(fixture).unwrap();
assert_eq!(info.last_exit, None);
}
#[test]
fn a_signal_request_and_its_reply_round_trip() {
let request = Request::Signal {
selector: SelectorSpec::Name("web".to_string()),
signal: "SIGHUP".to_string(),
};
let json = serde_json::to_string(&request).unwrap();
assert_eq!(serde_json::from_str::<Request>(&json).unwrap(), request);
let reply = Response::Signalled(vec![
SignalReply {
id: 1,
name: "web".to_string(),
outcome: SignalOutcome::Delivered,
},
SignalReply {
id: 2,
name: "web".to_string(),
outcome: SignalOutcome::NotRunning,
},
SignalReply {
id: 3,
name: "api".to_string(),
outcome: SignalOutcome::Failed {
reason: "no such process".to_string(),
},
},
]);
let json = serde_json::to_string(&reply).unwrap();
assert_eq!(serde_json::from_str::<Response>(&json).unwrap(), reply);
assert!(json.contains(r#""kind":"delivered""#), "{json}");
assert!(json.contains(r#""kind":"not_running""#), "{json}");
assert!(json.contains(r#""kind":"failed""#), "{json}");
}
#[test]
fn a_scale_request_names_one_app_and_a_count() {
let request = Request::Scale {
name: "web".to_string(),
count: 4,
};
let json = serde_json::to_string(&request).unwrap();
assert_eq!(serde_json::from_str::<Request>(&json).unwrap(), request);
assert!(json.contains(r#""kind":"scale""#), "{json}");
assert!(json.contains(r#""name":"web""#), "{json}");
assert!(!json.contains("selector"), "{json}");
}
#[test]
fn a_scaled_reply_carries_its_own_tag() {
let json = serde_json::to_string(&Response::Scaled(vec![])).unwrap();
assert_eq!(json, r#"{"kind":"scaled","data":[]}"#);
}
#[test]
fn a_send_line_request_and_its_reply_round_trip() {
let request = Request::SendLine {
selector: SelectorSpec::Name("repl".to_string()),
line: "reload-config".to_string(),
};
let json = serde_json::to_string(&request).unwrap();
assert_eq!(serde_json::from_str::<Request>(&json).unwrap(), request);
let reply = Response::SentLine(vec![
LineReply {
id: 1,
name: "repl".to_string(),
outcome: LineOutcome::Sent,
},
LineReply {
id: 2,
name: "web".to_string(),
outcome: LineOutcome::NoStdin,
},
LineReply {
id: 3,
name: "stuck".to_string(),
outcome: LineOutcome::NotWritten {
reason: "the app did not read its stdin within 2s".to_string(),
},
},
]);
let json = serde_json::to_string(&reply).unwrap();
assert_eq!(serde_json::from_str::<Response>(&json).unwrap(), reply);
assert!(json.contains(r#""kind":"sent""#), "{json}");
assert!(json.contains(r#""kind":"no_stdin""#), "{json}");
assert!(json.contains("did not read its stdin"), "{json}");
}
#[test]
fn a_line_carrying_a_newline_is_still_one_field_on_the_wire() {
let request = Request::SendLine {
selector: SelectorSpec::All,
line: "a\nb".to_string(),
};
let json = serde_json::to_string(&request).unwrap();
assert!(json.contains(r#""line":"a\nb""#), "{json}");
assert_eq!(serde_json::from_str::<Request>(&json).unwrap(), request);
}
#[test]
fn request_wire_snapshots() {
let requests = vec![
Envelope {
id: 1,
deadline_ms: Some(5000),
body: Request::Ping,
},
Envelope {
id: 2,
deadline_ms: None,
body: Request::ListFlock,
},
Envelope {
id: 3,
deadline_ms: None,
body: Request::Stop {
selector: SelectorSpec::Name("web".to_string()),
},
},
Envelope {
id: 4,
deadline_ms: None,
body: Request::Start {
apps: vec![AppConfig::minimal("web", "./srv")],
},
},
Envelope {
id: 5,
deadline_ms: None,
body: Request::Reopen {
selector: SelectorSpec::All,
},
},
Envelope {
id: 6,
deadline_ms: None,
body: Request::Flush {
selector: SelectorSpec::All,
},
},
Envelope {
id: 7,
deadline_ms: None,
body: Request::Reload {
selector: SelectorSpec::Name("web".to_string()),
},
},
Envelope {
id: 8,
deadline_ms: None,
body: Request::Trigger {
selector: SelectorSpec::Name("web".to_string()),
action: "set-log-level".to_string(),
params: Some("debug".to_string()),
},
},
Envelope {
id: 9,
deadline_ms: None,
body: Request::SaveRoll,
},
Envelope {
id: 10,
deadline_ms: None,
body: Request::Muster,
},
Envelope {
id: 11,
deadline_ms: None,
body: Request::DogConfig {
name: "bark".to_string(),
},
},
Envelope {
id: 12,
deadline_ms: None,
body: Request::EnableDog {
name: "metrics".to_string(),
source: DogSource::BuiltIn,
},
},
Envelope {
id: 13,
deadline_ms: None,
body: Request::DisableDog {
name: "metrics".to_string(),
},
},
Envelope {
id: 14,
deadline_ms: None,
body: Request::Describe {
selector: SelectorSpec::Id(7),
},
},
Envelope {
id: 15,
deadline_ms: None,
body: Request::Describe {
selector: SelectorSpec::Regex("^web-".to_string()),
},
},
Envelope {
id: 16,
deadline_ms: None,
body: Request::Describe {
selector: SelectorSpec::Fold("api".to_string()),
},
},
Envelope {
id: 17,
deadline_ms: None,
body: Request::Signal {
selector: SelectorSpec::Name("web".to_string()),
signal: "SIGHUP".to_string(),
},
},
Envelope {
id: 18,
deadline_ms: None,
body: Request::Scale {
name: "web".to_string(),
count: 4,
},
},
Envelope {
id: 19,
deadline_ms: None,
body: Request::SendLine {
selector: SelectorSpec::All,
line: "reload-config".to_string(),
},
},
Envelope {
id: 20,
deadline_ms: None,
body: Request::SetSmit {
sheep: "web".to_string(),
smit: Some(
"\u{25b2} main@a1b2c3"
.parse()
.expect("the reference smit is valid"),
),
},
},
Envelope {
id: 21,
deadline_ms: None,
body: Request::SetSmit {
sheep: "web".to_string(),
smit: None,
},
},
Envelope {
id: 22,
deadline_ms: None,
body: Request::ConfigDrift { apps: Vec::new() },
},
Envelope {
id: 23,
deadline_ms: None,
body: Request::Restart {
selector: SelectorSpec::Instance {
name: "web".to_string(),
slot: 2,
},
},
},
];
insta::assert_json_snapshot!("request_wire_v2", requests);
}
#[test]
fn reply_wire_snapshots() {
let replies = vec![
Reply {
id: 1,
result: Ok(Response::Pong),
},
Reply {
id: 2,
result: Ok(Response::Flock(vec![sample_info()])),
},
Reply {
id: 3,
result: Err(RpcError {
code: RpcErrorCode::NotFound,
message: "no sheep matches `web`".to_string(),
}),
},
Reply {
id: 4,
result: Ok(Response::Triggered(vec![ActionReply {
id: 3,
name: "web".to_string(),
outcome: ActionOutcome::Replied {
body: "ok".to_string(),
},
}])),
},
Reply {
id: 5,
result: Ok(Response::RollSaved {
path: "/home/ada/.shep/flock.json".to_string(),
apps: 2,
}),
},
Reply {
id: 6,
result: Ok(Response::Flock(vec![ProcessInfo {
id: 7,
name: "otel".to_string(),
dog: Some(DogSource::Adopted {
path: "/usr/local/bin/shep-otel".to_string(),
}),
..sample_info()
}])),
},
Reply {
id: 7,
result: Ok(Response::DogSection {
toml: "port = 9615\n".to_string().into(),
}),
},
Reply {
id: 8,
result: Ok(Response::DogStarted(ProcessInfo {
id: 4,
name: "metrics".to_string(),
dog: Some(DogSource::BuiltIn),
..sample_info()
})),
},
Reply {
id: 9,
result: Ok(Response::Described(vec![])),
},
Reply {
id: 10,
result: Ok(Response::Started(vec![])),
},
Reply {
id: 11,
result: Ok(Response::Stopped(vec![])),
},
Reply {
id: 12,
result: Ok(Response::Restarted(vec![])),
},
Reply {
id: 13,
result: Ok(Response::Reloading(vec![])),
},
Reply {
id: 14,
result: Ok(Response::Deleted(vec![7, 8])),
},
Reply {
id: 15,
result: Ok(Response::Reopened(vec![])),
},
Reply {
id: 16,
result: Ok(Response::Flushed(vec![])),
},
Reply {
id: 17,
result: Ok(Response::Mustered(vec![])),
},
Reply {
id: 18,
result: Ok(Response::Subscribed),
},
Reply {
id: 19,
result: Ok(Response::ShuttingDown),
},
Reply {
id: 20,
result: Ok(Response::Signalled(vec![
SignalReply {
id: 1,
name: "web".to_string(),
outcome: SignalOutcome::Delivered,
},
SignalReply {
id: 2,
name: "web".to_string(),
outcome: SignalOutcome::NotRunning,
},
SignalReply {
id: 3,
name: "api".to_string(),
outcome: SignalOutcome::Failed {
reason: "no such process".to_string(),
},
},
])),
},
Reply {
id: 21,
result: Ok(Response::Scaled(vec![sample_info()])),
},
Reply {
id: 22,
result: Ok(Response::SentLine(vec![
LineReply {
id: 1,
name: "repl".to_string(),
outcome: LineOutcome::Sent,
},
LineReply {
id: 2,
name: "web".to_string(),
outcome: LineOutcome::NoStdin,
},
LineReply {
id: 3,
name: "stuck".to_string(),
outcome: LineOutcome::NotWritten {
reason: "the app did not read its stdin within 2s".to_string(),
},
},
])),
},
Reply {
id: 23,
result: Ok(Response::Described(vec![
ProcessInfo::builder(3, "web", ProcStatus::Online)
.pid(Some(4242))
.lambs(Some(vec![Lamb::new(4243, "node"), Lamb::new(4244, "sh")]))
.build(),
])),
},
Reply {
id: 24,
result: Ok(Response::Flock(vec![
ProcessInfo::builder(5, "worker", ProcStatus::Stopped)
.restarts(1)
.last_exit(Some(ExitInfo {
code: None,
signal: Some(15),
}))
.build(),
])),
},
Reply {
id: 25,
result: Ok(Response::SmitPainted(vec![
ProcessInfo::builder(3, "web", ProcStatus::Online)
.pid(Some(4242))
.smit(Some("\u{25b2} main@a1b2c3".to_string()))
.build(),
])),
},
Reply {
id: 26,
result: Ok(Response::Drifted(vec![
SheepDrift::new("web", vec!["cwd".to_string()]),
SheepDrift::new(
"api",
vec!["args".to_string(), "env".to_string(), "script".to_string()],
),
])),
},
Reply {
id: 27,
result: Ok(Response::Flock(vec![
ProcessInfo::builder(9, "web", ProcStatus::Online)
.pid(Some(5150))
.instance(Some(2))
.build(),
])),
},
];
insta::assert_json_snapshot!("reply_wire_v2", replies);
}
#[test]
fn a_process_info_without_a_smit_key_still_deserializes() {
let fixture = r#"{"id":1,"name":"web","status":"online","pid":42,"restarts":0,"uptime_ms":10,"fold":null,"out_file":null,"err_file":null,"cpu_percent":null,"memory_bytes":null,"dog":null,"lambs":null,"last_exit":null}"#;
let info: ProcessInfo = serde_json::from_str(fixture).unwrap();
assert_eq!(info.smit, None);
}
#[test]
fn a_smit_is_validated_when_it_is_deserialized_not_only_when_parsed() {
for bad in [
r#""\u001b[2Jgone""#.to_string(), r#""a\nb""#.to_string(), r#""""#.to_string(), r#"" ""#.to_string(), format!(r#""{}""#, "x".repeat(Smit::MAX_CHARS + 1)), ] {
assert!(
serde_json::from_str::<Smit>(&bad).is_err(),
"a daemon must refuse this on the wire: {bad}"
);
}
assert!(serde_json::from_str::<Smit>(r#""\u25b2 main@a1b2c3""#).is_ok());
}
#[test]
fn a_smit_travels_as_a_bare_string() {
let smit: Smit = "\u{25b2} main@a1b2c3".parse().expect("valid");
let json = serde_json::to_string(&smit).unwrap();
assert_eq!(json, "\"\u{25b2} main@a1b2c3\"");
assert_eq!(serde_json::from_str::<Smit>(&json).unwrap(), smit);
}
#[test]
fn a_smit_is_capped_in_characters_not_bytes() {
let cjk = "\u{7f8a}".repeat(Smit::MAX_CHARS);
assert_eq!(cjk.len(), Smit::MAX_CHARS * 3);
assert!(cjk.parse::<Smit>().is_ok(), "{cjk}");
assert_eq!(
"x".repeat(Smit::MAX_CHARS + 1).parse::<Smit>(),
Err(SmitError::TooLong {
chars: Smit::MAX_CHARS + 1
})
);
}
#[test]
fn a_smit_is_stored_exactly_as_it_arrived() {
let padded: Smit = " main@a1b2c3 ".parse().expect("valid");
assert_eq!(padded.as_str(), " main@a1b2c3 ");
assert_eq!(padded.to_string(), " main@a1b2c3 ");
}
#[test]
fn v1_fixture_still_deserializes() {
let fixture = r#"{"id":7,"deadline_ms":null,"body":{"kind":"stop","selector":{"kind":"name","value":"web"}}}"#;
let env: Envelope = serde_json::from_str(fixture).unwrap();
assert_eq!(env.id, 7);
assert!(matches!(
env.body,
Request::Stop { selector: SelectorSpec::Name(ref n) } if n == "web"
));
}
#[test]
fn hello_handshake_shape() {
let hello = Hello {
client_version: "0.1.0".to_string(),
protocol: PROTOCOL_VERSION,
};
let json = serde_json::to_string(&hello).unwrap();
assert_eq!(json, r#"{"client_version":"0.1.0","protocol":2}"#);
}
#[test]
fn hello_reply_carries_typed_skew_error() {
let refusal: HelloReply = Err(RpcError {
code: RpcErrorCode::ProtocolMismatch,
message: "daemon speaks protocol 1, client sent 2".to_string(),
});
let json = serde_json::to_string(&refusal).unwrap();
assert_eq!(
json,
r#"{"Err":{"code":"protocol_mismatch","message":"daemon speaks protocol 1, client sent 2"}}"#
);
let back: HelloReply = serde_json::from_str(&json).unwrap();
assert_eq!(back, refusal);
}
#[test]
fn v1_reply_fixture_still_deserializes() {
let ok = r#"{"id":1,"result":{"Ok":{"kind":"pong"}}}"#;
let reply: Reply = serde_json::from_str(ok).unwrap();
assert!(matches!(reply.result, Ok(Response::Pong)));
let err = r#"{"id":2,"result":{"Err":{"code":"not_found","message":"no sheep"}}}"#;
let reply: Reply = serde_json::from_str(err).unwrap();
assert_eq!(reply.result.unwrap_err().code, RpcErrorCode::NotFound);
}
#[test]
fn v1_hello_ack_fixture_still_deserializes() {
let fixture = r#"{"Ok":{"daemon_version":"0.1.0","protocol":1,"pid":4242}}"#;
let ack: HelloReply = serde_json::from_str(fixture).unwrap();
assert_eq!(ack.unwrap().pid, 4242);
}
#[test]
fn v1_process_info_without_stats_still_deserializes() {
let fixture = r#"{"id":3,"name":"web","status":"online","pid":4242,"restarts":1,"uptime_ms":60000,"fold":"backend","out_file":"/l/o.log","err_file":"/l/e.log"}"#;
let info: ProcessInfo = serde_json::from_str(fixture).unwrap();
assert_eq!(info.cpu_percent, None);
assert_eq!(info.memory_bytes, None);
}
#[test]
fn v1_process_info_without_log_paths_still_deserializes() {
let fixture = r#"{"id":3,"name":"web","status":"online","pid":4242,"restarts":1,"uptime_ms":60000,"fold":"backend"}"#;
let info: ProcessInfo = serde_json::from_str(fixture).unwrap();
assert_eq!(info.id, 3);
assert_eq!(info.out_file, None);
assert_eq!(info.err_file, None);
}
#[test]
fn an_old_client_still_decodes_a_new_process_info() {
#[derive(Deserialize)]
struct V1ProcessInfo {
id: u32,
fold: Option<String>,
}
let current = serde_json::to_string(&sample_info()).unwrap();
let old: V1ProcessInfo = serde_json::from_str(¤t).unwrap();
assert_eq!(old.id, 3);
assert_eq!(old.fold.as_deref(), Some("backend"));
}
#[test]
fn deadline_exceeded_code_serializes_snake_case() {
assert_eq!(
serde_json::to_string(&RpcErrorCode::DeadlineExceeded).unwrap(),
"\"deadline_exceeded\""
);
assert_eq!(
serde_json::from_str::<RpcErrorCode>("\"deadline_exceeded\"").unwrap(),
RpcErrorCode::DeadlineExceeded
);
}
#[test]
fn action_outcome_kinds_serialize_snake_case_and_round_trip() {
let cases = [
(
ActionOutcome::Replied {
body: "pong".to_string(),
},
r#"{"kind":"replied","body":"pong"}"#,
),
(ActionOutcome::NoChannel, r#"{"kind":"no_channel"}"#),
(ActionOutcome::Skipped, r#"{"kind":"skipped"}"#),
(ActionOutcome::TimedOut, r#"{"kind":"timed_out"}"#),
];
for (outcome, wire) in cases {
assert_eq!(
serde_json::to_string(&outcome).unwrap(),
wire,
"{outcome:?}"
);
assert_eq!(
serde_json::from_str::<ActionOutcome>(wire).unwrap(),
outcome
);
}
}
#[test]
fn save_roll_serializes_snake_case_with_its_payload_under_data() {
assert_eq!(
serde_json::to_string(&Request::SaveRoll).unwrap(),
r#"{"kind":"save_roll"}"#
);
let reply = Response::RollSaved {
path: "/tmp/flock.json".to_string(),
apps: 3,
};
let wire = r#"{"kind":"roll_saved","data":{"path":"/tmp/flock.json","apps":3}}"#;
assert_eq!(serde_json::to_string(&reply).unwrap(), wire);
assert_eq!(serde_json::from_str::<Response>(wire).unwrap(), reply);
}
#[test]
fn muster_serializes_snake_case_with_its_listing_under_data() {
assert_eq!(
serde_json::to_string(&Request::Muster).unwrap(),
r#"{"kind":"muster"}"#
);
let reply = Response::Mustered(Vec::new());
let wire = r#"{"kind":"mustered","data":[]}"#;
assert_eq!(serde_json::to_string(&reply).unwrap(), wire);
assert_eq!(serde_json::from_str::<Response>(wire).unwrap(), reply);
}
#[test]
fn the_dog_verbs_serialize_snake_case_with_their_payloads_under_data() {
assert_eq!(
serde_json::to_string(&Request::DogConfig {
name: "bark".to_string()
})
.unwrap(),
r#"{"kind":"dog_config","name":"bark"}"#
);
assert_eq!(
serde_json::to_string(&Request::DisableDog {
name: "bark".to_string()
})
.unwrap(),
r#"{"kind":"disable_dog","name":"bark"}"#
);
let section = Response::DogSection {
toml: "port = 9615\n".to_string().into(),
};
let wire = r#"{"kind":"dog_section","data":{"toml":"port = 9615\n"}}"#;
assert_eq!(serde_json::to_string(§ion).unwrap(), wire);
assert_eq!(serde_json::from_str::<Response>(wire).unwrap(), section);
}
#[test]
fn dog_section_toml_debug_does_not_leak() {
let toml: DogSectionToml =
"webhook_url = \"https://discord.com/api/webhooks/1/super-secret-token\"\n"
.to_string()
.into();
assert_eq!(format!("{toml:?}"), "DogSectionToml(<70 bytes>)");
let response = Response::DogSection { toml };
assert_eq!(
format!("{response:?}"),
"DogSection { toml: DogSectionToml(<70 bytes>) }"
);
}
#[test]
fn a_listing_sorts_by_name_then_by_id() {
let mut listing = vec![
ProcessInfo::builder(1, "web", ProcStatus::Online).build(),
ProcessInfo::builder(2, "api", ProcStatus::Online).build(),
ProcessInfo::builder(0, "web", ProcStatus::Online).build(),
];
sort_flock(&mut listing);
let seen: Vec<(&str, u32)> = listing
.iter()
.map(|info| (info.name.as_str(), info.id))
.collect();
assert_eq!(
seen,
vec![("api", 2), ("web", 0), ("web", 1)],
"name first, then id inside a name"
);
}
#[test]
fn an_instance_slot_survives_a_round_trip_and_defaults_to_absent() {
let with = ProcessInfo::builder(1, "web", ProcStatus::Online)
.instance(Some(2))
.build();
assert_eq!(with.instance, Some(2));
let without = ProcessInfo::builder(1, "web", ProcStatus::Online).build();
assert_eq!(
without.instance, None,
"a row nobody set a slot on says so, rather than claiming slot 0"
);
}
#[test]
fn a_reply_from_a_daemon_without_the_field_deserializes_as_absent() {
let json = r#"{"id":1,"name":"web","status":"online","pid":null,
"restarts":0,"uptime_ms":0,"fold":null,"out_file":null,
"err_file":null,"cpu_percent":null,"memory_bytes":null,"dog":null,
"lambs":null,"last_exit":null,"smit":null}"#;
let info: ProcessInfo = serde_json::from_str(json).expect("older reply still parses");
assert_eq!(info.instance, None);
}
#[test]
fn sort_flock_orders_by_slot_before_id() {
let mut listing = vec![
ProcessInfo::builder(9, "web", ProcStatus::Online)
.instance(Some(0))
.build(),
ProcessInfo::builder(2, "web", ProcStatus::Online)
.instance(Some(1))
.build(),
];
sort_flock(&mut listing);
assert_eq!(
listing.iter().map(|i| i.id).collect::<Vec<_>>(),
vec![9, 2],
"slot 0 leads even though its id is higher"
);
}
#[test]
fn sort_flock_falls_back_to_id_when_no_row_carries_a_slot() {
let mut listing = vec![
ProcessInfo::builder(5, "web", ProcStatus::Online).build(),
ProcessInfo::builder(3, "web", ProcStatus::Online).build(),
];
sort_flock(&mut listing);
assert_eq!(
listing.iter().map(|i| i.id).collect::<Vec<_>>(),
vec![3, 5],
"an older daemon's listing sorts exactly as it does today"
);
}
}