Real Bug over here

This commit is contained in:
lda
2026-04-12 00:22:45 +07:00 Unverified
parent f17111a239
commit b8b8d951f1
5 changed files with 166 additions and 8 deletions
+1
View File
@@ -1,3 +1,4 @@
We work towards clean, modular, maintainable design, clear documentation... im not the best at this thing, my choices can be bad, so you help me.
DRY: spawn subagents.
Context: break those files out.
+9 -1
View File
@@ -6,10 +6,13 @@ USE_PROCD=1
NAME=wakey
BIN=/root/.bin/wakey-agent
CONFIG=/etc/wakey-agent/config.toml
start_service() {
procd_open_instance
procd_set_param command "$BIN" serve
procd_set_param command "$BIN" serve --config "$CONFIG"
procd_set_param file "$CONFIG"
procd_set_param env RUST_LOG=wakey_agent=debug,wakey=debug
procd_set_param respawn 5 1 0
procd_set_param stdout 1
procd_set_param stderr 1
@@ -19,3 +22,8 @@ start_service() {
stop_service() {
: # procd manages the process; nothing to do here
}
reload_service() {
# Prefer in-process reload; fallback to full restart if signal path is unavailable.
procd_send_signal "$NAME" HUP 2>/dev/null || restart
}
+2 -2
View File
@@ -41,7 +41,7 @@ async fn dispatch_leases(req: LeasesRequest) -> Result<CommandResult> {
})
.await?;
debug!(rows = leases.len(), include_state = req.include_state, "dispatched leases command");
Ok(CommandResult::Leases(leases))
Ok(CommandResult::Leases { rows: leases })
}
async fn dispatch_devs(req: DevsRequest) -> Result<CommandResult> {
@@ -54,7 +54,7 @@ async fn dispatch_devs(req: DevsRequest) -> Result<CommandResult> {
devs.retain(|dev| dev.operstate == "up");
}
debug!(rows = devs.len(), up_only = req.up_only, "dispatched devs command");
Ok(CommandResult::Devs(devs))
Ok(CommandResult::Devs { rows: devs })
}
async fn dispatch_inventory(req: InventoryRequest) -> Result<CommandResult> {
+14 -2
View File
@@ -175,8 +175,8 @@ pub enum AgentCommand {
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum CommandResult {
Status(Status<NeighborEntry>),
Leases(Vec<DhcpLeaseWithState>),
Devs(Vec<InterfaceSummary>),
Leases { rows: Vec<DhcpLeaseWithState> },
Devs { rows: Vec<InterfaceSummary> },
Inventory(DeviceInventory),
Wake(WakeResult),
}
@@ -236,4 +236,16 @@ mod tests {
let err = RequestId::try_from(" ".to_string()).expect_err("must fail");
assert!(err.contains("must not be empty"));
}
#[test]
fn client_result_with_devs_serializes() {
let msg = ClientMessage::Result {
request_id: RequestId::try_from("req-devs-1".to_string()).expect("request id"),
result: CommandResult::Devs { rows: vec![] },
};
let json = serde_json::to_string(&msg).expect("serialize");
assert!(json.contains("\"type\":\"result\""));
assert!(json.contains("\"kind\":\"devs\""));
}
}
+139 -2
View File
@@ -128,7 +128,7 @@ where
match dispatch_command(command).await {
Ok(result) => {
info!(request_id = %request_id, command = %kind, "command execution completed");
send_json(sink, &ClientMessage::Result { request_id, result }).await?;
send_command_result(sink, request_id, kind, result).await?;
}
Err(err) => {
error!(request_id = %request_id, command = %kind, error = %err, "command dispatch failed");
@@ -151,6 +151,49 @@ where
Ok(())
}
async fn send_command_result<S>(
sink: &mut S,
request_id: crate::protocol::RequestId,
kind: &str,
result: crate::protocol::CommandResult,
) -> Result<()>
where
S: SinkExt<Message> + Unpin,
<S as futures_util::Sink<Message>>::Error: std::error::Error + Send + Sync + 'static,
{
if let Err(err) = send_json(
sink,
&ClientMessage::Result {
request_id: request_id.clone(),
result,
},
)
.await
{
error!(
request_id = %request_id,
command = %kind,
error = %err,
"failed to send command result; sending explicit error frame"
);
send_json(
sink,
&ClientMessage::Error {
request_id,
error: ErrorPayload {
code: "command_result_serialize_failed".into(),
message: err.to_string(),
retryable: Some(true),
},
},
)
.await?;
}
Ok(())
}
async fn send_json<S>(sink: &mut S, message: &ClientMessage) -> Result<()>
where
S: SinkExt<Message> + Unpin,
@@ -205,7 +248,56 @@ pub fn websocket_url(server_url: &str) -> Result<url::Url> {
#[cfg(test)]
mod tests {
use super::{next_backoff_ms, websocket_url};
use std::collections::VecDeque;
use std::io;
use std::pin::Pin;
use std::task::{Context, Poll};
use futures_util::Sink;
use serde_json::Value;
use tokio_tungstenite::tungstenite::Message;
use super::{next_backoff_ms, send_command_result, websocket_url};
use crate::protocol::{CommandResult, RequestId};
struct RecordingSink {
fail_sends_remaining: usize,
sent: VecDeque<Message>,
}
impl RecordingSink {
fn new(fail_sends_remaining: usize) -> Self {
Self {
fail_sends_remaining,
sent: VecDeque::new(),
}
}
}
impl Sink<Message> for RecordingSink {
type Error = io::Error;
fn poll_ready(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn start_send(mut self: Pin<&mut Self>, item: Message) -> Result<(), Self::Error> {
if self.fail_sends_remaining > 0 {
self.fail_sends_remaining -= 1;
return Err(io::Error::other("injected send failure"));
}
self.sent.push_back(item);
Ok(())
}
fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn poll_close(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
}
#[test]
fn websocket_url_uses_expected_path() {
@@ -224,4 +316,49 @@ mod tests {
fn backoff_never_shrinks_when_max_is_lower() {
assert_eq!(next_backoff_ms(8_000, 1_000), 8_000);
}
#[tokio::test]
async fn send_command_result_emits_result_on_success() {
let mut sink = RecordingSink::new(0);
let request_id = RequestId::try_from("req-success".to_string()).expect("request id");
send_command_result(
&mut sink,
request_id,
"wake",
CommandResult::Wake(wakey_core::WakeResult { result: vec![] }),
)
.await
.expect("send should succeed");
assert_eq!(sink.sent.len(), 1);
let Some(Message::Text(payload)) = sink.sent.pop_front() else {
panic!("expected text websocket message");
};
let v: Value = serde_json::from_str(payload.as_ref()).expect("json");
assert_eq!(v["type"], "result");
}
#[tokio::test]
async fn send_command_result_falls_back_to_error_frame() {
let mut sink = RecordingSink::new(1);
let request_id = RequestId::try_from("req-fallback".to_string()).expect("request id");
send_command_result(
&mut sink,
request_id,
"devs",
CommandResult::Devs { rows: vec![] },
)
.await
.expect("fallback error frame should be sent");
assert_eq!(sink.sent.len(), 1);
let Some(Message::Text(payload)) = sink.sent.pop_front() else {
panic!("expected text websocket message");
};
let v: Value = serde_json::from_str(payload.as_ref()).expect("json");
assert_eq!(v["type"], "error");
assert_eq!(v["error"]["code"], "command_result_serialize_failed");
}
}