feat(firmware): validate configuration and migrate sensor-driven flight APIs

Validate configuration through its owning schemas and keep board
declarations separate from shared F722 resource ownership. Bound USB
transitions, isolate transport failures and retain restart-only activation.

Migrate navigation and flight consumers with the sensor and schema
contracts. Require explicit installation data and synchronous output
authority; board motor outputs remain inhibited.

Validation: assembled-prefix host tests, strict Clippy, platform/USB tests
and both ARM release builds. Final-source coverage and scoped Miri/Kani
evidence are retained separately from hardware qualification.
This commit is contained in:
Onur Özkan committed 2026-11-07 18:12:11 +03:00
1 parent fb1b8af2d1
commit f9a61a086f
211 files changed
+45657 -12740

No files matched your search

+194 -69
View File
@@ -4,33 +4,105 @@ use std::{
io::{Read, Write},
net::TcpStream,
sync::Arc,
time::{Duration, Instant},
};
use serde::{Deserialize, Serialize};
use crate::{
serial_worker::{Command, SerialController},
state::SharedState,
serial_worker::{Action, SerialController},
state::{SharedState, ViewContext},
};
const INDEX: &str = include_str!("../web/index.html");
const APP: &str = include_str!("../web/app.js");
const VIEW: &str = include_str!("../web/view.js");
const CSS: &str = include_str!("../web/styles.css");
const CSP: &str = "default-src 'self'; connect-src 'self'; img-src 'self'; style-src 'self'; script-src 'self'; base-uri 'none'; form-action 'self'; frame-ancestors 'none'";
const HTTP_PHASE_TIMEOUT: Duration = Duration::from_secs(2);
const MAX_REQUEST_BYTES: usize = 64 * 1024;
const MAX_BODY_BYTES: usize = 60 * 1024;
struct HttpConnection {
stream: TcpStream,
read_deadline: Instant,
write_deadline: Option<Instant>,
timeout: Duration,
}
impl HttpConnection {
fn new(stream: TcpStream, timeout: Duration) -> Self {
Self {
stream,
read_deadline: Instant::now() + timeout,
write_deadline: None,
timeout,
}
}
}
fn remaining(deadline: Instant) -> std::io::Result<Duration> {
deadline
.checked_duration_since(Instant::now())
.filter(|duration| !duration.is_zero())
.ok_or_else(|| std::io::Error::new(std::io::ErrorKind::TimedOut, "HTTP deadline expired"))
}
impl Read for HttpConnection {
fn read(&mut self, bytes: &mut [u8]) -> std::io::Result<usize> {
self.stream
.set_read_timeout(Some(remaining(self.read_deadline)?))?;
self.stream.read(bytes)
}
}
impl Write for HttpConnection {
fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
let deadline = *self
.write_deadline
.get_or_insert_with(|| Instant::now() + self.timeout);
self.stream.set_write_timeout(Some(remaining(deadline)?))?;
self.stream.write(bytes)
}
fn flush(&mut self) -> std::io::Result<()> {
self.stream.flush()
}
}
#[derive(Deserialize)]
struct Connect {
context: ViewContext,
path: String,
}
#[derive(Deserialize)]
struct EditDraft {
context: ViewContext,
field_id: u16,
value: serde_json::Value,
}
#[derive(Deserialize)]
struct Mutation {
context: ViewContext,
}
pub fn serve_connection(
mut stream: TcpStream,
stream: TcpStream,
state: &Arc<SharedState>,
serial: &SerialController,
origin: &str,
) -> std::io::Result<()> {
serve_http(
HttpConnection::new(stream, HTTP_PHASE_TIMEOUT),
state,
serial,
origin,
)
}
fn serve_http(
mut stream: HttpConnection,
state: &Arc<SharedState>,
serial: &SerialController,
origin: &str,
@@ -39,30 +111,42 @@ pub fn serve_connection(
let header_end = loop {
let mut chunk = [0_u8; 4096];
let length = stream.read(&mut chunk)?;
if length == 0 || request.len() + length > 64 * 1024 {
if length == 0 || request.len() + length > MAX_REQUEST_BYTES {
return reply(&mut stream, 400, "text/plain", "bad request");
}
// Only the previous chunk's last three bytes can start a split separator.
let search_start = request.len().saturating_sub(3);
request.extend_from_slice(&chunk[..length]);
if let Some(separator) = request[search_start..]
.windows(4)
.position(|window| window == b"\r\n\r\n")
{
break search_start + separator + 4;
}
};
let Some(content_length) = content_length(&request[..header_end - 4]) else {
return reply(&mut stream, 400, "text/plain", "bad request");
};
if content_length > MAX_BODY_BYTES || content_length > MAX_REQUEST_BYTES - header_end {
return reply(&mut stream, 400, "text/plain", "bad request");
}
let body_end = header_end + content_length;
while request.len() < body_end {
let mut chunk = [0_u8; 4096];
let remaining = (body_end - request.len()).min(chunk.len());
let length = stream.read(&mut chunk[..remaining])?;
if length == 0 {
return reply(&mut stream, 400, "text/plain", "bad request");
}
request.extend_from_slice(&chunk[..length]);
if let Some(separator) = request.windows(4).position(|window| window == b"\r\n\r\n") {
let end = separator + 4;
let head = String::from_utf8_lossy(&request[..separator]);
let content_length = header(&head, "Content-Length")
.and_then(|value| value.parse::<usize>().ok())
.unwrap_or(0);
if content_length > 60 * 1024 {
return reply(&mut stream, 400, "text/plain", "bad request");
}
if request.len() >= end + content_length {
break end;
}
}
};
let head = String::from_utf8_lossy(&request[..header_end - 4]);
let body = &request[header_end..];
}
// This connection serves one request and closes. Ignore any pipelined bytes.
let head = std::str::from_utf8(&request[..header_end - 4]).expect("validated HTTP headers");
let body = &request[header_end..body_end];
let mut parts = head.lines().next().unwrap_or_default().split_whitespace();
let method = parts.next().unwrap_or_default();
let path = parts.next().unwrap_or_default();
if method == "POST" && !same_origin(&head, origin) {
if method == "POST" && !same_origin(head, origin) {
return reply(&mut stream, 403, "text/plain", "forbidden");
}
match (method, path) {
@@ -74,32 +158,54 @@ pub fn serve_connection(
APP,
),
("GET", "/styles.css") => reply(&mut stream, 200, "text/css; charset=utf-8", CSS),
("GET", "/view.js") => reply(
&mut stream,
200,
"application/javascript; charset=utf-8",
VIEW,
),
("GET", "/api/view") => json(&mut stream, 200, &state.view()),
("GET", "/api/state") => json(&mut stream, 200, &state.snapshot()),
("GET", "/api/config") => json(&mut stream, 200, &state.configuration()),
("GET", "/api/packets") => json(&mut stream, 200, &state.packets()),
("GET", "/api/logs") => json(&mut stream, 200, &state.logs()),
("POST", "/api/connect") => match serde_json::from_slice::<Connect>(body) {
Ok(value) if !value.path.trim().is_empty() => match state.begin_connect() {
Ok(()) => {
let result = serial.connect(value.path.trim().into());
if let Err(error) = &result {
state.worker_failed(error.clone());
}
endpoint(&mut stream, result)
}
Err(error) => json(&mut stream, 409, &serde_json::json!({"error":error})),
},
Ok(value) if !value.path.trim().is_empty() => endpoint(
&mut stream,
state,
serial.connect(value.path.trim().into(), value.context),
),
_ => json(
&mut stream,
400,
&serde_json::json!({"error":"A serial device path is required."}),
),
},
("POST", "/api/disconnect") => endpoint(&mut stream, serial.send(Command::Disconnect)),
("POST", "/api/disconnect" | "/api/config/save" | "/api/config/reset") => {
match serde_json::from_slice::<Mutation>(body) {
Ok(mutation) => {
let action = match path {
"/api/disconnect" => Action::Disconnect,
"/api/config/save" => Action::Save,
_ => Action::ResetSaved,
};
endpoint(&mut stream, state, serial.submit(mutation.context, action))
}
Err(_) => json(
&mut stream,
400,
&serde_json::json!({"error":"Expected the displayed view context."}),
),
}
}
("POST", "/api/config/edit") => match serde_json::from_slice::<EditDraft>(body) {
Ok(edit) => match state.edit(edit.field_id, &edit.value) {
Ok(config) => json(&mut stream, 200, &config),
Err(error) => json(&mut stream, 409, &serde_json::json!({"error":error})),
Ok(edit) => match state.edit_context(edit.context, edit.field_id, &edit.value) {
Ok(view) => json(&mut stream, 200, &view),
Err(error) => json(
&mut stream,
409,
&serde_json::json!({"error":error, "view":state.view()}),
),
},
Err(_) => json(
&mut stream,
@@ -107,20 +213,46 @@ pub fn serve_connection(
&serde_json::json!({"error":"Expected a stable field_id and typed value."}),
),
},
("POST", "/api/config/save") => match state.begin_save() {
Ok(save) => {
let result = serial.send(Command::Save(save));
if let Err(error) = &result {
state.worker_failed(error.clone());
}
endpoint(&mut stream, result)
}
Err(error) => json(&mut stream, 409, &serde_json::json!({"error":error})),
},
_ => reply(&mut stream, 404, "text/plain", "not found"),
}
}
fn content_length(head: &[u8]) -> Option<usize> {
let head = std::str::from_utf8(head).ok()?;
let mut lines = head.split("\r\n");
if lines.next()?.bytes().any(|byte| byte.is_ascii_control()) {
return None;
}
let mut content_length = None;
for line in lines {
let (key, value) = line.split_once(':')?;
if key.is_empty()
|| !key
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || b"!#$%&'*+-.^_`|~".contains(&byte))
|| value
.bytes()
.any(|byte| byte.is_ascii_control() && byte != b'\t')
{
return None;
}
if key.eq_ignore_ascii_case("Transfer-Encoding") {
return None;
}
if key.eq_ignore_ascii_case("Content-Length") {
let value = value.trim_matches([' ', '\t']);
if content_length.is_some()
|| value.is_empty()
|| !value.bytes().all(|byte| byte.is_ascii_digit())
{
return None;
}
content_length = Some(value.parse::<usize>().ok()?);
}
}
Some(content_length.unwrap_or(0))
}
fn same_origin(head: &str, origin: &str) -> bool {
let Some(expected_host) = origin.strip_prefix("http://") else {
return false;
@@ -135,14 +267,22 @@ fn header<'a>(head: &'a str, name: &str) -> Option<&'a str> {
})
}
fn endpoint(stream: &mut TcpStream, result: Result<(), String>) -> std::io::Result<()> {
fn endpoint(
stream: &mut HttpConnection,
state: &SharedState,
result: Result<(), String>,
) -> std::io::Result<()> {
match result {
Ok(()) => json(stream, 202, &serde_json::json!({"ok":true})),
Err(error) => json(stream, 409, &serde_json::json!({"error":error})),
Ok(()) => json(stream, 202, &state.view()),
Err(error) => json(
stream,
409,
&serde_json::json!({"error":error, "view":state.view()}),
),
}
}
fn json<T: Serialize>(stream: &mut TcpStream, status: u16, value: &T) -> std::io::Result<()> {
fn json<T: Serialize>(stream: &mut HttpConnection, status: u16, value: &T) -> std::io::Result<()> {
match serde_json::to_string(value) {
Ok(body) => reply(stream, status, "application/json; charset=utf-8", &body),
Err(_) => reply(stream, 500, "text/plain", "serialization error"),
@@ -150,7 +290,7 @@ fn json<T: Serialize>(stream: &mut TcpStream, status: u16, value: &T) -> std::io
}
fn reply(
stream: &mut TcpStream,
stream: &mut HttpConnection,
status: u16,
content_type: &str,
body: &str,
@@ -171,6 +311,10 @@ fn reply(
)
}
#[cfg(test)]
#[path = "api_hosted_tests.rs"]
mod hosted_tests;
#[cfg(test)]
mod tests {
use super::*;
@@ -213,23 +357,4 @@ mod tests {
assert!(APP.contains("truncated.textContent = 'TRUNCATED'"));
assert!(INDEX.contains("id=\"log-entries\""));
}
#[test]
fn browser_clears_recovered_errors_and_retries_failed_configuration_fetches() {
assert!(APP.contains("showError(state.error ?? '')"));
assert!(APP.contains("q('#configuration').hidden = !state.has_config"));
let fetch = APP
.find("const config = await api('/api/config')")
.expect("configuration refresh fetches the latest document");
let render = APP[fetch..]
.find("renderConfig(config)")
.map(|offset| fetch + offset)
.expect("a fetched configuration is rendered");
let revision = APP[render..]
.find("configRevision = state.config_revision")
.map(|offset| render + offset)
.expect("a successful render advances the observed revision");
assert!(fetch < render && render < revision);
}
}
@@ -0,0 +1,416 @@
use super::*;
use crate::{config_client::ClientEvent, serial_worker::SerialBaud, state::tests::unknown_board};
use kaptan_file_mavlink::Principal;
use std::{net::TcpListener, os::fd::AsRawFd, sync::mpsc, thread};
fn controller(state: &Arc<SharedState>) -> Arc<SerialController> {
Arc::new(
SerialController::new(
Arc::clone(state),
SerialBaud::DEFAULT,
Principal::new(255, 190),
)
.unwrap(),
)
}
fn request(
state: &Arc<SharedState>,
serial: &Arc<SerialController>,
path: &str,
body: Option<serde_json::Value>,
) -> (u16, serde_json::Value) {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let address = listener.local_addr().unwrap();
let state = Arc::clone(state);
let serial = Arc::clone(serial);
let server = thread::spawn(move || {
serve_connection(
listener.accept().unwrap().0,
&state,
&serial,
&format!("http://{address}"),
)
.unwrap();
});
let mut client = TcpStream::connect(address).unwrap();
client
.set_read_timeout(Some(Duration::from_secs(2)))
.unwrap();
let method = if body.is_some() { "POST" } else { "GET" };
let body = body.map(|value| value.to_string()).unwrap_or_default();
write!(client, "{method} {path} HTTP/1.1\r\nHost: {address}\r\nOrigin: http://{address}\r\nContent-Length: {}\r\n\r\n{body}", body.len()).unwrap();
let mut response = String::new();
client.read_to_string(&mut response).unwrap();
server.join().unwrap();
let (head, body) = response.split_once("\r\n\r\n").unwrap();
let status = head.split_whitespace().nth(1).unwrap().parse().unwrap();
(status, serde_json::from_str(body).unwrap())
}
#[test]
fn stale_tab_cannot_edit_save_reset_or_disconnect_a_new_controller_over_http() {
let state = Arc::new(SharedState::default());
let serial = controller(&state);
state.connected("controller-A".into());
state.configuration_event(ClientEvent::Configuration(unknown_board()));
let (status, a) = request(&state, &serial, "/api/view", None);
assert_eq!(status, 200);
let context = a["state"]["context"].clone();
assert!(context["connection"].is_string());
state.disconnected(None);
state.connected("controller-B".into());
state.configuration_event(ClientEvent::Configuration(unknown_board()));
state.edit(401, &serde_json::json!(50)).unwrap();
let before = state.snapshot();
for path in [
"/api/config/edit",
"/api/config/save",
"/api/config/reset",
"/api/disconnect",
] {
let (status, response) = request(
&state,
&serial,
path,
Some(serde_json::json!({
"context": context, "field_id": 401, "value": 60,
})),
);
assert_eq!(status, 409, "{path}: {response}");
assert!(response["error"].as_str().unwrap().contains("stale view"));
assert_eq!(state.snapshot().config_revision, before.config_revision);
assert_eq!(state.snapshot().path.as_deref(), Some("controller-B"));
assert_eq!(
state.configuration().fields[0].draft,
Some(serde_json::json!(50))
);
}
state.disconnected(None);
let (status, _) = request(
&state,
&serial,
"/api/connect",
Some(serde_json::json!({
"context": context, "path": "/not-opened",
})),
);
assert_eq!(status, 409);
assert!(!state.snapshot().can_disconnect);
assert_eq!(
request(
&state,
&serial,
"/api/config/reset",
Some(serde_json::json!({}))
)
.0,
400
);
}
#[test]
fn idle_http_connection_expires_and_the_sequential_server_serves_the_next_client() {
let state = Arc::new(SharedState::default());
let serial = controller(&state);
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let address = listener.local_addr().unwrap();
let (accepted, ready) = mpsc::channel();
let server = thread::spawn(move || {
let stream = listener.accept().unwrap().0;
accepted.send(()).unwrap();
let started = Instant::now();
let error = serve_http(
HttpConnection::new(stream, Duration::from_millis(40)),
&state,
&serial,
&format!("http://{address}"),
)
.unwrap_err();
assert!(matches!(
error.kind(),
std::io::ErrorKind::TimedOut | std::io::ErrorKind::WouldBlock
));
assert!(started.elapsed() < Duration::from_secs(1));
serve_connection(
listener.accept().unwrap().0,
&state,
&serial,
&format!("http://{address}"),
)
.unwrap();
});
let _idle = TcpStream::connect(address).unwrap();
ready.recv_timeout(Duration::from_secs(1)).unwrap();
let mut next = TcpStream::connect(address).unwrap();
next.set_read_timeout(Some(Duration::from_secs(1))).unwrap();
next.write_all(b"GET /api/view HTTP/1.1\r\n\r\n").unwrap();
let mut response = String::new();
next.read_to_string(&mut response).unwrap();
assert!(response.starts_with("HTTP/1.1 200"));
server.join().unwrap();
}
#[test]
fn http_deadlines_are_total_even_after_partial_progress() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let mut client = TcpStream::connect(listener.local_addr().unwrap()).unwrap();
let server = listener.accept().unwrap().0;
let mut connection = HttpConnection::new(server, Duration::from_millis(40));
client.write_all(b"xy").unwrap();
let mut byte = [0];
connection.read_exact(&mut byte).unwrap();
connection.read_deadline = Instant::now();
assert_eq!(
connection.read(&mut byte).unwrap_err().kind(),
std::io::ErrorKind::TimedOut
);
let size: libc::c_int = 4096;
// SAFETY: the live TCP fd and pointer/length describe one integer option.
let configured = unsafe {
libc::setsockopt(
connection.stream.as_raw_fd(),
libc::SOL_SOCKET,
libc::SO_SNDBUF,
(&size as *const libc::c_int).cast(),
std::mem::size_of_val(&size) as libc::socklen_t,
)
};
assert_eq!(configured, 0);
let started = Instant::now();
let error = connection.write_all(&vec![0; 4 * 1024 * 1024]).unwrap_err();
assert!(matches!(
error.kind(),
std::io::ErrorKind::TimedOut | std::io::ErrorKind::WouldBlock
));
assert!(started.elapsed() < Duration::from_secs(1));
connection.write_deadline = Some(Instant::now());
assert_eq!(
connection.write(b"z").unwrap_err().kind(),
std::io::ErrorKind::TimedOut
);
}
fn raw_request(
state: &Arc<SharedState>,
serial: &Arc<SerialController>,
fragments: impl FnOnce(String) -> Vec<Vec<u8>>,
) -> (u16, String) {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let address = listener.local_addr().unwrap();
let state = Arc::clone(state);
let serial = Arc::clone(serial);
let server = thread::spawn(move || {
serve_connection(
listener.accept().unwrap().0,
&state,
&serial,
&format!("http://{address}"),
)
.unwrap();
});
let mut client = TcpStream::connect(address).unwrap();
client.set_nodelay(true).unwrap();
client
.set_read_timeout(Some(Duration::from_secs(3)))
.unwrap();
for (index, fragment) in fragments(address.to_string()).iter().enumerate() {
if index != 0 {
thread::sleep(Duration::from_millis(2));
}
client.write_all(fragment).unwrap();
}
if let Err(error) = client.shutdown(std::net::Shutdown::Write) {
assert!(matches!(
error.kind(),
std::io::ErrorKind::NotConnected | std::io::ErrorKind::ConnectionReset
));
}
let mut response = String::new();
if let Err(error) = client.read_to_string(&mut response) {
// Closing a one-request connection with unread pipeline bytes can reset
// TCP. A complete response is still required below.
assert_eq!(error.kind(), std::io::ErrorKind::ConnectionReset);
}
server.join().unwrap();
let (head, body) = response.split_once("\r\n\r\n").unwrap();
assert_eq!(
header(head, "Content-Length")
.unwrap()
.parse::<usize>()
.unwrap(),
body.len()
);
assert_eq!(header(head, "Connection"), Some("close"));
let status = head.split_whitespace().nth(1).unwrap().parse().unwrap();
(status, body.into())
}
fn active_edit() -> (Arc<SharedState>, Arc<SerialController>, String) {
let state = Arc::new(SharedState::default());
let serial = controller(&state);
state.connected("parser-test-controller".into());
state.configuration_event(ClientEvent::Configuration(unknown_board()));
let body = serde_json::json!({
"context": state.snapshot().context, "field_id": 401, "value": 50,
})
.to_string();
(state, serial, body)
}
#[test]
fn malformed_framing_cannot_apply_a_current_context_edit() {
let (state, serial, body) = active_edit();
let before = serde_json::to_value(state.view()).unwrap();
for framing in [
"Content-Length:".into(),
"Content-Length: invalid".into(),
format!("Content-Length: +{}", body.len()),
"Content-Length: -1".into(),
format!("Content-Length: {}", usize::MAX),
format!("Content-Length: {}0", usize::MAX),
format!("Content-Length: {}, {}", body.len(), body.len()),
format!(
"Content-Length: {}\r\ncOnTeNt-LeNgTh: {}",
body.len(),
body.len()
),
format!("Content-Length: {}\r\nContent-Length: 0", body.len()),
format!("Content-Length : {}", body.len()),
"Transfer-Encoding: chunked".into(),
format!(
"Content-Length: {}\r\nTransfer-Encoding: identity",
body.len()
),
format!("Content-Length: {}", body.len() + 1),
] {
let (status, response) = raw_request(&state, &serial, |address| {
vec![format!(
"POST /api/config/edit HTTP/1.1\r\nHost: {address}\r\nOrigin: http://{address}\r\n{framing}\r\n\r\n{body}"
).into_bytes()]
});
assert_eq!(status, 400, "{framing}: {response}");
assert_eq!(serde_json::to_value(state.view()).unwrap(), before);
}
// The same context and body are actionable when framing is valid.
assert_eq!(
request(
&state,
&serial,
"/api/config/edit",
Some(serde_json::from_str(&body).unwrap())
)
.0,
200
);
assert_eq!(
state.configuration().fields[0].draft,
Some(serde_json::json!(50))
);
}
#[test]
fn fragmented_headers_separator_and_body_apply_one_current_edit() {
let (state, serial, body) = active_edit();
let (status, response) = raw_request(&state, &serial, |address| {
let mut fragments = vec![
format!("POST /api/config/edit HTTP/1.1\r\nHost: {address}\r\nOrigin: http://{address}\r\nContent-Len").into_bytes(),
format!("gth: {}", body.len()).into_bytes(),
b"\r".to_vec(), b"\n".to_vec(), b"\r".to_vec(), b"\n".to_vec(),
];
fragments.extend(body.as_bytes().chunks(13).map(<[u8]>::to_vec));
fragments
});
assert_eq!(status, 200, "{response}");
assert_eq!(
state.configuration().fields[0].draft,
Some(serde_json::json!(50))
);
}
#[test]
fn malformed_request_line_and_wrong_origin_cannot_edit_an_active_context() {
let (state, serial, body) = active_edit();
let before = serde_json::to_value(state.view()).unwrap();
for (line, same_origin, expected) in [
(
"POST /api/config/edit HTTP/1.1\nContent-Length: 0",
true,
400,
),
("POST /api/config/edit HTTP/1.1", false, 403),
] {
let (status, _) = raw_request(&state, &serial, |address| {
let origin = if same_origin {
format!("http://{address}")
} else {
"http://example.invalid".into()
};
vec![format!(
"{line}\r\nHost: {address}\r\nOrigin: {origin}\r\nContent-Length: {}\r\n\r\n{body}", body.len()
).into_bytes()]
});
assert_eq!(status, expected);
assert_eq!(serde_json::to_value(state.view()).unwrap(), before);
}
}
#[test]
fn exact_body_length_excludes_trailing_bytes_and_pipeline_is_not_dispatched() {
let (state, serial, body) = active_edit();
let before = serde_json::to_value(state.view()).unwrap();
let (status, _) = raw_request(&state, &serial, |address| {
vec![format!(
"GET /api/view HTTP/1.1\r\n\r\nPOST /api/config/edit HTTP/1.1\r\nHost: {address}\r\nOrigin: http://{address}\r\nContent-Length: {}\r\n\r\n{body}", body.len()
).into_bytes()]
});
assert_eq!(status, 200);
assert_eq!(serde_json::to_value(state.view()).unwrap(), before);
let (status, response) = raw_request(&state, &serial, |address| {
vec![format!(
"POST /api/config/edit HTTP/1.1\r\nHost: {address}\r\nOrigin: http://{address}\r\nContent-Length: {}\r\n\r\n{body}GET /api/view HTTP/1.1\r\n\r\n", body.len()
).into_bytes()]
});
assert_eq!(status, 200, "{response}");
assert_eq!(
state.configuration().fields[0].draft,
Some(serde_json::json!(50))
);
}
#[test]
fn request_and_body_caps_reject_before_mutation_and_accept_exact_body_cap() {
let (state, serial, body) = active_edit();
let before = serde_json::to_value(state.view()).unwrap();
for (extra_header, length) in [
(String::new(), MAX_BODY_BYTES + 1),
(
format!("X-Padding: {}\r\n", "x".repeat(4096)),
MAX_BODY_BYTES,
),
(
format!("X-Padding: {}\r\n", "x".repeat(MAX_REQUEST_BYTES)),
body.len(),
),
] {
let (status, _) = raw_request(&state, &serial, |address| {
vec![format!(
"POST /api/config/edit HTTP/1.1\r\nHost: {address}\r\nOrigin: http://{address}\r\n{extra_header}Content-Length: {length}\r\n\r\n{body}"
).into_bytes()]
});
assert_eq!(status, 400);
assert_eq!(serde_json::to_value(state.view()).unwrap(), before);
}
let (status, response) = raw_request(&state, &serial, |address| {
let padding = " ".repeat(MAX_BODY_BYTES - body.len());
vec![format!(
"POST /api/config/edit HTTP/1.1\r\nHost: {address}\r\nOrigin: http://{address}\r\nContent-Length: {MAX_BODY_BYTES}\r\n\r\n{body}{padding}"
).into_bytes()]
});
assert_eq!(status, 200, "{response}");
assert_eq!(
state.configuration().fields[0].draft,
Some(serde_json::json!(50))
);
}
File diff suppressed because it is too large. Load diff
@@ -0,0 +1,648 @@
//! Host client interoperability with the real file service and configuration lifecycle.
use std::{cell::RefCell, rc::Rc};
use kaptan_config::{
Config, ConfigEnum, ConfigManager, ConfigModel, ConfigStore, ConfigViolation, FieldDescriptor,
FieldWriteError, KcfgExporter, KcfgImporter, ReplaceError,
};
use kaptan_file_mavlink::{
FileFailure, FileMetadata, FilePath, FileProvider, FileService, IngressId, TransferPeer,
wire::{decode_request, encode_response},
};
use super::*;
const CONTROLLER: Target = Target::new(42, 7);
const CLIENT: Principal = Principal::new(255, 190);
const PRIMARY_ROLE: FieldId = FieldId::new(11);
const PRIMARY_BAUD: FieldId = FieldId::new(12);
#[derive(Clone, Copy, Debug, PartialEq, ConfigEnum)]
#[repr(u8)]
enum Role {
Disabled = 0,
Mavlink = 1,
Sensor = 2,
}
// Generated metadata makes this a multi-chunk artifact without padding or a
// dependency on a particular board or its port catalog.
#[derive(Clone, Debug, PartialEq, Config)]
struct Candidate {
#[config(enum, id = 11, key = "ports.primary.role", unit = "")]
primary_role: Role,
#[config(
id = 12,
key = "ports.primary.baud",
unit = "baud",
range(min = 9600, max = 921600)
)]
primary_baud: u32,
#[config(enum, id = 21, key = "ports.secondary.role", unit = "")]
secondary_role: Role,
#[config(
id = 22,
key = "ports.secondary.baud",
unit = "baud",
range(min = 9600, max = 921600)
)]
secondary_baud: u32,
#[config(enum, id = 31, key = "ports.auxiliary.role", unit = "")]
auxiliary_role: Role,
#[config(
id = 32,
key = "ports.auxiliary.baud",
unit = "baud",
range(min = 9600, max = 921600)
)]
auxiliary_baud: u32,
#[config(id = 41, key = "scalar.free", unit = "")]
free: f32,
#[config(id = 42, key = "scalar.nonzero", unit = "", nonzero)]
nonzero: f64,
#[config(id = 43, key = "scalar.finite", unit = "", finite)]
finite: f64,
}
impl Candidate {
fn factory() -> Self {
Self {
primary_role: Role::Disabled,
primary_baud: 115_200,
secondary_role: Role::Disabled,
secondary_baud: 115_200,
auxiliary_role: Role::Disabled,
auxiliary_baud: 115_200,
free: 0.0,
nonzero: 1.0,
finite: 2.0,
}
}
fn boot() -> Self {
Self {
primary_role: Role::Mavlink,
secondary_role: Role::Sensor,
auxiliary_baud: 57_600,
..Self::factory()
}
}
}
impl ConfigModel for Candidate {
type Active = Self;
type Context = ();
const SCHEMA_ID: [u8; 16] = <Self as Config>::SCHEMA_ID;
fn descriptor(index: u16) -> Option<&'static FieldDescriptor> {
<Self as Config>::descriptor(usize::from(index))
}
fn field_count() -> u16 {
Self::FIELD_COUNT.try_into().unwrap()
}
fn field_value(&self, id: FieldId) -> Option<ConfigValue> {
<Self as Config>::field_value(self, id)
}
fn active_field_value(active: &Self, id: FieldId) -> Option<ConfigValue> {
<Self as Config>::field_value(active, id)
}
fn set_field(&mut self, id: FieldId, value: ConfigValue) -> Result<(), FieldWriteError> {
<Self as Config>::set_field(self, id, value)
}
fn validate(&self, _: &()) -> Result<Self, ConfigViolation> {
<Self as Config>::validate(self)?;
Ok(self.clone())
}
}
struct Disk {
saved: Option<Candidate>,
persist_calls: usize,
fail_next: bool,
}
struct MemoryStore(Rc<RefCell<Disk>>);
impl ConfigStore<Candidate> for MemoryStore {
type Error = ();
fn load(&mut self, _: &Candidate) -> Result<Option<Candidate>, ()> {
Ok(self.0.borrow().saved.clone())
}
fn persist(&mut self, candidate: &Candidate) -> Result<(), ()> {
let mut disk = self.0.borrow_mut();
disk.persist_calls += 1;
disk.saved = Some(candidate.clone());
// A store error can occur after durability. The manager must invalidate
// its committed view instead of assuming rollback.
if std::mem::take(&mut disk.fail_next) {
Err(())
} else {
Ok(())
}
}
}
enum OpenFile {
Read(Vec<u8>),
Write {
importer: KcfgImporter<Candidate>,
reset: bool,
},
}
struct ConfigFiles {
manager: ConfigManager<Candidate, MemoryStore>,
open: Option<OpenFile>,
}
fn failure(status: FileStatus) -> FileFailure {
FileFailure::new(status, Content::EMPTY).unwrap()
}
impl FileProvider for ConfigFiles {
fn open_read(&mut self, path: &FilePath) -> Result<FileMetadata, FileFailure> {
let (current, restart_required) = match path.as_bytes() {
CONFIG_PATH => (
self.manager
.committed()
.ok_or_else(|| failure(FileStatus::StorageFailure))?,
self.manager.restart_required(),
),
ACTIVE_PATH => (self.manager.active(), false),
_ => return Err(failure(FileStatus::NotFound)),
};
let exporter =
KcfgExporter::new(current, self.manager.factory(), restart_required).unwrap();
let mut bytes = vec![0; exporter.length()];
assert_eq!(exporter.read(0, &mut bytes).unwrap(), bytes.len());
let metadata = FileMetadata {
size: bytes.len().try_into().unwrap(),
checksum_crc32c: exporter.checksum_crc32c(),
};
self.open = Some(OpenFile::Read(bytes));
Ok(metadata)
}
fn open_write(&mut self, path: &FilePath, _: FileMetadata) -> Result<(), FileFailure> {
let reset = match path.as_bytes() {
CONFIG_PATH => false,
RESET_PATH => true,
_ => return Err(failure(FileStatus::NotFound)),
};
if !reset && self.manager.committed().is_none() {
return Err(failure(FileStatus::StorageFailure));
}
self.open = Some(OpenFile::Write {
importer: KcfgImporter::new(self.manager.factory()).unwrap(),
reset,
});
Ok(())
}
fn read(&mut self, position: u32, output: &mut [u8]) -> Result<(), FileFailure> {
let Some(OpenFile::Read(bytes)) = &self.open else {
return Err(failure(FileStatus::InvalidState));
};
let start = usize::try_from(position).unwrap();
output.copy_from_slice(&bytes[start..start + output.len()]);
Ok(())
}
fn write(&mut self, _: u32, data: &[u8]) -> Result<(), FileFailure> {
let Some(OpenFile::Write { importer, .. }) = &mut self.open else {
return Err(failure(FileStatus::InvalidState));
};
importer
.push(self.manager.factory(), data)
.map_err(|_| failure(FileStatus::ContentRejected))
}
fn close(&mut self) -> Result<Content, FileFailure> {
let open = self
.open
.take()
.ok_or_else(|| failure(FileStatus::InvalidState))?;
let OpenFile::Write { importer, reset } = open else {
return Ok(Content::EMPTY);
};
let update = importer
.finish()
.map_err(|_| failure(FileStatus::ContentRejected))?;
let result = if reset {
if !self.manager.is_factory(&update.candidate) {
return Err(failure(FileStatus::ContentRejected));
}
self.manager.reset_to_factory()
} else {
self.manager
.replace_if_current(update.candidate, &update.expected_base)
};
let outcome = result.map_err(|error| {
failure(match error {
ReplaceError::Conflict => FileStatus::RequestConflict,
ReplaceError::Store(()) | ReplaceError::DurableStateUnknown => {
FileStatus::StorageFailure
}
ReplaceError::InvalidCandidate(_) => FileStatus::ContentRejected,
})
})?;
Ok(
Content::try_new(&KcfgResultDetail::success(outcome.restart_required).encode())
.unwrap(),
)
}
fn abort(&mut self) {
self.open = None;
}
}
struct Link {
service: FileService,
provider: ConfigFiles,
disk: Rc<RefCell<Disk>>,
encoder: MavlinkEncoder,
exchanges: Vec<(FileRequest, FileResponse)>,
}
impl Link {
fn new() -> Self {
let disk = Rc::new(RefCell::new(Disk {
saved: Some(Candidate::boot()),
persist_calls: 0,
fail_next: false,
}));
let manager =
ConfigManager::recover(Candidate::factory(), (), MemoryStore(Rc::clone(&disk)))
.unwrap();
Self {
service: FileService::new(CONTROLLER),
provider: ConfigFiles {
manager,
open: None,
},
disk,
encoder: MavlinkEncoder::new(CONTROLLER.system(), CONTROLLER.component()),
exchanges: Vec::new(),
}
}
fn exchange(&mut self, frame: MavlinkFrame) -> MavlinkFrame {
let request = decode_request(&parse_test_message(&frame).unwrap()).unwrap();
let peer = TransferPeer {
principal: Principal::new(frame.system_id(), frame.component_id()),
ingress: IngressId(0),
};
let result = self
.service
.dispatch(&mut self.provider, peer, request)
.expect("client must address the service endpoint");
self.exchanges.push((request, result.response));
match encode_response(result.response) {
MavMessage::FILE_CONTROL(data) => self.encoder.encode(&data),
MavMessage::FILE_DATA(data) => self.encoder.encode(&data),
_ => unreachable!(),
}
}
fn run(
&mut self,
client: &mut ConfigClient,
mut request: MavlinkFrame,
drop_close_ack: bool,
) -> ClientEvent {
let mut dropped = false;
// One write and its two readbacks each need at most chunks + 2
// exchanges. The extra exchange permits one dropped-close retry.
let chunks = kaptan_config::kcfg_length::<Candidate>()
.unwrap()
.div_ceil(DATA_BYTES);
for _ in 0..3 * (chunks + 2) {
let mut reply = self.exchange(request);
if drop_close_ack
&& !dropped
&& matches!(
client.pending.as_ref().map(|pending| pending.kind),
Some(PendingKind::CloseWrite { .. })
)
{
dropped = true;
client.pending.as_mut().unwrap().sent_at = Instant::now() - CLOSE_WRITE_TIMEOUT;
let Outcome::Send(retry) = client.retry_due() else {
panic!("lost close acknowledgement must retry")
};
assert_eq!(retry.raw_bytes(), request.raw_bytes());
reply = self.exchange(retry);
}
match client.observe_frame(&reply) {
Outcome::Send(next) => request = next,
Outcome::Event(event) => {
assert_eq!(dropped, drop_close_ack);
return event;
}
Outcome::Idle => panic!("a real service reply must advance the client"),
}
}
panic!("configuration exchange exceeded its artifact-derived bound")
}
fn discover(&mut self, client: &mut ConfigClient) -> RemoteConfig {
let request = client.discover(CONTROLLER).unwrap();
configuration(self.run(client, request, false))
}
fn expire_lease(&mut self) {
// Switching editors requires expiry of the real service's replay lease.
assert!(!self.service.expire(&mut self.provider));
assert!(self.provider.open.is_none());
}
}
fn configuration(event: ClientEvent) -> RemoteConfig {
match event {
ClientEvent::Configuration(config) => config,
ClientEvent::Failure(detail)
| ClientEvent::RediscoveryRequired(detail)
| ClientEvent::DurableStateUnknown(detail) => panic!("configuration failed: {detail}"),
}
}
fn baud_edit(value: u32) -> Vec<Edit> {
vec![Edit {
field_id: PRIMARY_BAUD,
value: ConfigValue::U32(value),
}]
}
fn field(config: &RemoteConfig, id: FieldId) -> &RemoteField {
config.field(id).unwrap()
}
#[test]
fn scalar_bits_and_rules_interoperate_through_browser_drafts_and_the_real_file_service() {
let mut link = Link::new();
let mut client = ConfigClient::new(CLIENT);
let state = crate::state::SharedState::default();
state.connected("test".into());
let initial = link.discover(&mut client);
assert!(field(&initial, FieldId::new(42)).nonzero);
assert!(field(&initial, FieldId::new(43)).finite);
state.configuration_event(ClientEvent::Configuration(initial));
state.edit(41, &serde_json::json!("-0")).unwrap();
state
.edit(42, &serde_json::json!("bits:0xfff0000000000042"))
.unwrap();
assert!(state.edit(43, &serde_json::json!("inf")).is_err());
state.edit(43, &serde_json::json!("2")).unwrap();
let request = client.save(state.begin_save().unwrap()).unwrap();
let saved = configuration(link.run(&mut client, request, true));
let nan = ConfigValue::F64(f64::from_bits(0xfff0_0000_0000_0042));
assert_eq!(
field(&saved, FieldId::new(41)).value,
ConfigValue::F32(-0.0)
);
assert_eq!(field(&saved, FieldId::new(42)).value, nan);
assert_eq!(link.disk.borrow().persist_calls, 1);
assert!(saved.restart_required);
state.configuration_event(ClientEvent::Configuration(saved));
state
.edit(42, &serde_json::json!("bits:0xfff0000000000042"))
.unwrap();
assert!(!state.snapshot().can_save);
let request = client
.save(vec![Edit {
field_id: FieldId::new(42),
value: nan,
}])
.unwrap();
configuration(link.run(&mut client, request, false));
assert_eq!(
link.disk.borrow().persist_calls,
1,
"bit-identical NaN is a no-op"
);
let request = client.save(baud_edit(230_400)).unwrap();
configuration(link.run(&mut client, request, false));
assert_eq!(
link.disk.borrow().persist_calls,
2,
"NaN expected base matches exact bits"
);
}
#[test]
fn multichunk_commit_replays_lost_close_once_and_refreshes_the_next_save_base() {
let mut link = Link::new();
let mut client = ConfigClient::new(CLIENT);
let initial = link.discover(&mut client);
assert!(!initial.restart_required);
assert!(initial.committed_error.is_none());
assert_eq!(initial.fields.len(), Candidate::FIELD_COUNT);
assert_eq!(field(&initial, PRIMARY_ROLE).options[1].label, "Mavlink");
assert_eq!(
field(&initial, PRIMARY_ROLE).active,
Some(ConfigValue::U8(1))
);
let paths: Vec<_> = link
.exchanges
.iter()
.filter_map(|(request, _)| match request {
FileRequest::Control(control) if control.operation == FileOperation::OpenRead => {
Some(control.content.as_bytes())
}
_ => None,
})
.collect();
assert_eq!(
paths,
[b"/config/user.kcfg".as_slice(), b"/config/active.kcfg"]
);
let bytes = client.downloaded.as_ref().unwrap().document.len();
assert!(
bytes > DATA_BYTES,
"the complete candidate must span chunks"
);
let second_read_chunks = link.exchanges.iter().filter(|(_, response)| {
matches!(response, FileResponse::Data(data) if data.position == DATA_BYTES as u32)
}).count();
assert_eq!(
second_read_chunks, 2,
"both snapshots must cross the chunk boundary"
);
let start = link.exchanges.len();
let mut edits = baud_edit(230_400);
edits.push(Edit {
field_id: PRIMARY_ROLE,
value: ConfigValue::U8(0),
});
let request = client.save(edits).unwrap();
let saved = configuration(link.run(&mut client, request, true));
assert_eq!(link.disk.borrow().persist_calls, 1);
let mut expected = Candidate::boot();
expected.primary_role = Role::Disabled;
expected.primary_baud = 230_400;
assert_eq!(link.provider.manager.committed(), Some(&expected));
assert_eq!(link.provider.manager.active(), &Candidate::boot());
assert!(saved.restart_required);
assert_eq!(field(&saved, PRIMARY_ROLE).value, ConfigValue::U8(0));
assert_eq!(field(&saved, PRIMARY_ROLE).active, Some(ConfigValue::U8(1)));
assert_eq!(field(&saved, PRIMARY_BAUD).value, ConfigValue::U32(230_400));
assert_eq!(
field(&saved, PRIMARY_BAUD).active,
Some(ConfigValue::U32(115_200))
);
assert!(link.exchanges[start..].iter().any(|(request, _)| {
matches!(request, FileRequest::Data(data) if data.position == DATA_BYTES as u32)
}));
let replay = link.exchanges[start..].windows(2).find(|pair| {
pair[0].0 == pair[1].0
&& matches!(pair[0].0, FileRequest::Control(control) if control.operation == FileOperation::Close)
}).expect("close must traverse the real service twice");
assert_eq!(replay[0].1, replay[1].1);
let baseline = super::decode_artifact(&client.downloaded.as_ref().unwrap().document).unwrap();
assert_eq!(
baseline
.fields()
.find(|field| field.id() == PRIMARY_BAUD)
.unwrap()
.expected_base_value(),
ConfigValue::U32(230_400)
);
let request = client.save(baud_edit(460_800)).unwrap();
let second = configuration(link.run(&mut client, request, false));
assert_eq!(link.disk.borrow().persist_calls, 2);
expected.primary_baud = 460_800;
assert_eq!(link.disk.borrow().saved.as_ref(), Some(&expected));
assert_eq!(link.provider.manager.committed(), Some(&expected));
assert_eq!(link.provider.manager.active(), &Candidate::boot());
assert_eq!(
field(&second, PRIMARY_BAUD).value,
ConfigValue::U32(460_800)
);
}
#[test]
fn another_editors_commit_rejects_stale_close_without_persisting_again() {
let mut link = Link::new();
let mut first = ConfigClient::new(CLIENT);
let mut stale = ConfigClient::new(Principal::new(254, 190));
link.discover(&mut first);
link.expire_lease();
link.discover(&mut stale);
link.expire_lease();
let request = first.save(baud_edit(230_400)).unwrap();
configuration(link.run(&mut first, request, false));
assert_eq!(link.disk.borrow().persist_calls, 1);
link.expire_lease();
let request = stale.save(baud_edit(460_800)).unwrap();
let ClientEvent::RediscoveryRequired(detail) = link.run(&mut stale, request, false) else {
panic!("obsolete expected base must fail at close")
};
assert!(detail.contains("RequestConflict"));
assert!(detail.contains("stale draft discarded"));
assert!(matches!(link.exchanges.last().unwrap().1,
FileResponse::Control(control) if control.operation == FileOperation::Close
&& control.status == FileStatus::RequestConflict));
assert_eq!(link.disk.borrow().persist_calls, 1);
assert_eq!(
link.provider.manager.committed().unwrap().primary_baud,
230_400
);
assert_eq!(link.provider.manager.active(), &Candidate::boot());
assert!(link.provider.open.is_none());
assert!(stale.save(baud_edit(57_600)).is_err());
let refreshed = link.discover(&mut stale);
assert_eq!(
field(&refreshed, PRIMARY_BAUD).value,
ConfigValue::U32(230_400)
);
}
#[test]
fn storage_failure_exports_active_and_factory_reset_recovers_unknown_committed() {
let mut link = Link::new();
let mut client = ConfigClient::new(CLIENT);
link.discover(&mut client);
link.disk.borrow_mut().fail_next = true;
let request = client.save(baud_edit(230_400)).unwrap();
assert!(matches!(
link.run(&mut client, request, false),
ClientEvent::DurableStateUnknown(_)
));
assert_eq!(link.disk.borrow().persist_calls, 1);
assert!(link.provider.manager.committed().is_none());
assert_eq!(
link.disk.borrow().saved.as_ref().unwrap().primary_baud,
230_400
);
let start = link.exchanges.len();
let unknown = link.discover(&mut client);
assert!(
unknown
.committed_error
.as_ref()
.unwrap()
.contains("StorageFailure")
);
assert!(matches!(link.exchanges[start].1,
FileResponse::Control(control) if control.operation == FileOperation::OpenRead
&& control.status == FileStatus::StorageFailure));
assert_eq!(
field(&unknown, PRIMARY_ROLE).active,
Some(ConfigValue::U8(1))
);
assert_eq!(
field(&unknown, PRIMARY_BAUD).active,
Some(ConfigValue::U32(115_200))
);
assert!(client.downloaded.is_none());
assert!(client.save(baud_edit(460_800)).is_err());
let request = client.reset_saved().unwrap();
let start = link.exchanges.len();
let recovered = configuration(link.run(&mut client, request, true));
assert!(matches!(&link.exchanges[start].0,
FileRequest::Control(control) if control.operation == FileOperation::OpenWrite
&& control.content.as_bytes() == b"/config/reset.kcfg"));
assert_eq!(
link.disk.borrow().persist_calls,
2,
"reset retry must not persist twice"
);
assert_eq!(
link.disk.borrow().saved.as_ref(),
Some(&Candidate::factory())
);
assert_eq!(
link.provider.manager.committed(),
Some(&Candidate::factory())
);
assert_eq!(link.provider.manager.active(), &Candidate::boot());
assert!(recovered.committed_error.is_none());
assert!(recovered.restart_required);
assert!(
recovered
.fields
.iter()
.all(|field| field.value == field.factory)
);
assert_eq!(
field(&recovered, PRIMARY_ROLE).active,
Some(ConfigValue::U8(1))
);
let request = client.save(baud_edit(57_600)).unwrap();
configuration(link.run(&mut client, request, false));
assert_eq!(link.disk.borrow().persist_calls, 3);
}
+1 -1
View File
@@ -9,4 +9,4 @@ mod state;
pub use api::serve_connection;
pub use serial_worker::{SerialBaud, SerialController};
pub use state::SharedState;
pub use state::{SharedState, ViewContext};
+3 -1
View File
@@ -19,7 +19,9 @@ fn main() -> std::io::Result<()> {
)
.map_err(std::io::Error::other)?;
if let Some(path) = arguments.serial_path {
serial.connect(path).map_err(std::io::Error::other)?;
serial
.connect(path, state.snapshot().context)
.map_err(std::io::Error::other)?;
}
let listener = TcpListener::bind(LISTEN_ADDRESS)?;
let address = listener.local_addr()?;
@@ -0,0 +1,199 @@
use super::*;
use crate::{
config_client::{ClientEvent, tests::pending_close},
state::tests::unknown_board,
};
use std::{io, os::fd::FromRawFd, sync::mpsc};
fn pty() -> (File, File, String) {
let (mut master, mut slave) = (-1, -1);
// SAFETY: valid out-pointers receive two owned fds. Optional settings are null.
let opened = unsafe {
libc::openpty(
&mut master,
&mut slave,
std::ptr::null_mut(),
std::ptr::null(),
std::ptr::null(),
)
};
assert_eq!(opened, 0);
// SAFETY: openpty returned fresh fds whose ownership is transferred once.
let master = unsafe { File::from_raw_fd(master) };
// SAFETY: the second fresh fd is distinct and is transferred once.
let slave = unsafe { File::from_raw_fd(slave) };
// SAFETY: this live fd accepts the integer flags passed to fcntl.
let configured = unsafe { libc::fcntl(master.as_raw_fd(), libc::F_SETFL, libc::O_NONBLOCK) };
assert_eq!(configured, 0);
let path = format!("/proc/self/fd/{}", slave.as_raw_fd());
(master, slave, path)
}
fn fill_transmit_queue(file: &mut File) {
for _ in 0..4096 {
match file.write(&[0; 4096]) {
Ok(count) => assert_ne!(count, 0),
Err(error) if error.kind() == io::ErrorKind::WouldBlock => return,
Err(error) => panic!("cannot fill PTY: {error}"),
}
}
panic!("PTY transmit queue did not reach its bound")
}
fn stall_frame(session: &mut Session, frame: MavlinkFrame) {
// PTY line-discipline work can free a little space after the first EAGAIN.
// Keep filling the unread peer until an actual frame write is backpressured.
for _ in 0..64 {
fill_transmit_queue(session.link.as_mut().unwrap());
write_raw(session, frame).unwrap();
if !flush_pending(session).unwrap() {
return;
}
}
panic!("PTY did not backpressure a configuration frame")
}
#[test]
fn queued_save_and_reset_recheck_connection_and_configuration_before_effects() {
for reconnect in [false, true] {
for action in [Action::Save, Action::ResetSaved] {
let state = SharedState::default();
let connection = state.connected("A".into());
state.configuration_event(ClientEvent::Configuration(unknown_board()));
state.edit(401, &serde_json::json!(30)).unwrap();
let intent = state
.begin_intent(state.snapshot().context, action.kind())
.unwrap();
let (_master, _slave, path) = pty();
let mut session = Session::new(Principal::new(255, 190));
session.link = Some(open_serial(&path, SerialBaud::DEFAULT).unwrap());
session.connection = Some(connection);
if reconnect {
state.disconnected(None);
session.connection = Some(state.connected("B".into()));
}
state.configuration_event(ClientEvent::Configuration(unknown_board()));
let before = state.snapshot();
handle_command(Command { intent, action }, &mut session, &state);
assert!(session.outbound.is_none());
assert!(session.client.target().is_err());
assert_eq!(state.snapshot().config_revision, before.config_revision);
assert_eq!(state.snapshot().context, before.context);
assert_eq!(state.snapshot().path, before.path);
}
}
}
#[test]
fn partial_writes_keep_exact_offsets_and_total_deadlines() {
struct ShortWriter {
bytes: Vec<u8>,
stalled: bool,
}
impl Write for ShortWriter {
fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
if self.stalled {
return Err(io::ErrorKind::WouldBlock.into());
}
let count = bytes.len().min(3);
self.bytes.extend_from_slice(&bytes[..count]);
Ok(count)
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
let (_, frame) = pending_close();
let mut pending = PendingWrite {
frame,
position: 0,
deadline: Instant::now() + WRITE_TIMEOUT,
};
let mut output = ShortWriter {
bytes: Vec::new(),
stalled: false,
};
assert!(!pending.advance(&mut output).unwrap());
output.stalled = true;
assert!(!pending.advance(&mut output).unwrap());
assert_eq!(pending.position, 3);
output.stalled = false;
for _ in 0..frame.raw_bytes().len() {
if pending.advance(&mut output).unwrap() {
break;
}
}
assert_eq!(output.bytes, frame.raw_bytes());
pending.position = 0;
pending.deadline = Instant::now();
assert!(
pending
.advance(&mut output)
.unwrap_err()
.contains("deadline")
);
assert_eq!(output.bytes, frame.raw_bytes());
}
#[test]
fn nonblocking_pty_backpressure_can_be_cancelled_and_shutdown_preserves_close_uncertainty() {
let (_master, _slave, path) = pty();
let mut file = open_serial(&path, SerialBaud::DEFAULT).unwrap();
// SAFETY: fcntl reads flags for this still-owned valid fd.
let flags = unsafe { libc::fcntl(file.as_raw_fd(), libc::F_GETFL) };
assert_ne!(flags & libc::O_NONBLOCK, 0);
fill_transmit_queue(&mut file);
let state = Arc::new(SharedState::default());
let connection = state.connected(path);
state.configuration_event(ClientEvent::Configuration(unknown_board()));
let (client, frame) = pending_close();
let mut session = Session::new(Principal::new(255, 190));
session.client = client;
session.connection = Some(connection);
session.link = Some(file);
let started = Instant::now();
stall_frame(&mut session, frame);
assert!(started.elapsed() < Duration::from_millis(100));
let stopping = Arc::new(AtomicBool::new(false));
let (commands, receiver) = sync_channel(COMMAND_CAPACITY);
let worker_state = Arc::clone(&state);
let worker_stop = Arc::clone(&stopping);
let worker = thread::spawn(move || worker(receiver, worker_state, session, worker_stop));
let controller = SerialController {
commands,
worker: Mutex::new(Some(worker)),
baud: SerialBaud::DEFAULT,
state: Arc::clone(&state),
stopping,
};
let (done, completed) = mpsc::channel();
let shutdown = thread::spawn(move || {
drop(controller);
done.send(()).unwrap();
});
completed
.recv_timeout(Duration::from_millis(500))
.expect("shutdown must cancel a stalled write");
shutdown.join().unwrap();
assert!(state.snapshot().path.is_none());
assert!(state.snapshot().durable_state_unknown);
assert!(state.snapshot().error.unwrap().contains("unknown"));
}
#[test]
fn stalled_pty_write_hits_its_deadline_without_blocking_the_worker() {
let (_master, _slave, path) = pty();
let mut file = open_serial(&path, SerialBaud::DEFAULT).unwrap();
fill_transmit_queue(&mut file);
let mut session = Session::new(Principal::new(255, 190));
session.link = Some(file);
let (_, frame) = pending_close();
stall_frame(&mut session, frame);
session.outbound.as_mut().unwrap().deadline = Instant::now();
assert!(
flush_pending(&mut session)
.unwrap_err()
.contains("deadline")
);
}
+225 -65
View File
@@ -7,17 +7,18 @@ use std::{
os::unix::fs::OpenOptionsExt,
sync::{
Arc, Mutex, MutexGuard,
atomic::{AtomicBool, Ordering},
mpsc::{Receiver, SyncSender, TryRecvError, TrySendError, sync_channel},
},
thread::{self, JoinHandle},
time::Duration,
time::{Duration, Instant},
};
use crate::{
config_client::{ConfigClient, Edit, Outcome},
config_client::{ConfigClient, Outcome},
log_monitor::{LogObservation, event_level_from_wire, wire_level_label},
packet_monitor::PacketObservation,
state::SharedState,
state::{IntentKind, IntentToken, Revision, SharedState, ViewContext},
};
use kaptan_file_mavlink::{Principal, Target};
use kaptan_log::Level;
@@ -30,6 +31,7 @@ use mavlink_core::Message;
const COMMAND_CAPACITY: usize = 8;
const IDLE_POLL: Duration = Duration::from_millis(20);
const WRITE_TIMEOUT: Duration = Duration::from_secs(1);
/// A line rate that the host termios adapter can represent exactly.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
@@ -64,17 +66,35 @@ impl SerialBaud {
}
}
pub(crate) enum Command {
pub(crate) enum Action {
Connect { path: String, baud: SerialBaud },
Disconnect,
Save(Vec<Edit>),
Shutdown,
Save,
ResetSaved,
}
impl Action {
fn kind(&self) -> IntentKind {
match self {
Self::Connect { .. } => IntentKind::Connect,
Self::Disconnect => IntentKind::Disconnect,
Self::Save => IntentKind::Save,
Self::ResetSaved => IntentKind::Reset,
}
}
}
struct Command {
intent: IntentToken,
action: Action,
}
pub struct SerialController {
commands: SyncSender<Command>,
worker: Mutex<Option<JoinHandle<()>>>,
baud: SerialBaud,
state: Arc<SharedState>,
stopping: Arc<AtomicBool>,
}
impl SerialController {
@@ -84,36 +104,48 @@ impl SerialController {
local: Principal,
) -> Result<Self, String> {
let (commands, receiver) = sync_channel(COMMAND_CAPACITY);
let stopping = Arc::new(AtomicBool::new(false));
let worker_state = Arc::clone(&state);
let worker_stopping = Arc::clone(&stopping);
let worker = thread::Builder::new()
.name("kaptan-config-serial".into())
.spawn(move || worker(receiver, state, local))
.spawn(move || worker(receiver, worker_state, Session::new(local), worker_stopping))
.map_err(|error| error.to_string())?;
Ok(Self {
commands,
worker: Mutex::new(Some(worker)),
baud,
state,
stopping,
})
}
/// Opens one explicit point-to-point serial device at the selected CLI rate.
pub fn connect(&self, path: String) -> Result<(), String> {
self.send(Command::Connect {
path,
baud: self.baud,
pub fn connect(&self, path: String, context: ViewContext) -> Result<(), String> {
self.submit(
context,
Action::Connect {
path,
baud: self.baud,
},
)
}
pub(crate) fn submit(&self, context: ViewContext, action: Action) -> Result<(), String> {
self.state.queue_intent(context, action.kind(), |intent| {
self.commands
.try_send(Command { intent, action })
.map_err(|error| match error {
TrySendError::Full(_) => "configuration transport is busy".into(),
TrySendError::Disconnected(_) => {
"configuration transport is unavailable".into()
}
})
})
}
pub(crate) fn send(&self, command: Command) -> Result<(), String> {
self.commands
.try_send(command)
.map_err(|error| match error {
TrySendError::Full(_) => "configuration transport is busy".into(),
TrySendError::Disconnected(_) => "configuration transport is unavailable".into(),
})
}
fn shutdown(&self) {
let _ = self.commands.send(Command::Shutdown);
self.stopping.store(true, Ordering::Release);
if let Some(worker) = lock(&self.worker).take() {
let _ = worker.join();
}
@@ -130,6 +162,37 @@ struct Session {
link: Option<File>,
receiver: IntegrityMavlinkReceiver<1>,
client: ConfigClient,
connection: Option<Revision>,
outbound: Option<PendingWrite>,
}
struct PendingWrite {
frame: MavlinkFrame,
position: usize,
deadline: Instant,
}
impl PendingWrite {
fn advance(&mut self, writer: &mut impl Write) -> Result<bool, String> {
if Instant::now() >= self.deadline {
return Err("serial write deadline expired".into());
}
let bytes = self.frame.raw_bytes();
match writer.write(&bytes[self.position..]) {
Ok(0) => return Err("serial write returned zero bytes".into()),
Ok(count) => self.position += count,
Err(error)
if matches!(
error.kind(),
std::io::ErrorKind::WouldBlock | std::io::ErrorKind::Interrupted
) =>
{
return Ok(false);
}
Err(error) => return Err(error.to_string()),
}
Ok(self.position == bytes.len())
}
}
impl Session {
@@ -138,29 +201,35 @@ impl Session {
link: None,
receiver: IntegrityMavlinkReceiver::default(),
client: ConfigClient::new(local),
connection: None,
outbound: None,
}
}
fn close(&mut self) {
fn close(&mut self, state: &SharedState) {
if let Some(event) = self.client.interruption() {
state.configuration_event(event);
}
self.link = None;
self.connection = None;
self.outbound = None;
self.receiver.reset();
self.client.reset();
}
}
fn worker(commands: Receiver<Command>, state: Arc<SharedState>, local: Principal) {
let mut session = Session::new(local);
loop {
loop {
fn worker(
commands: Receiver<Command>,
state: Arc<SharedState>,
mut session: Session,
stopping: Arc<AtomicBool>,
) {
'worker: while !stopping.load(Ordering::Acquire) {
for _ in 0..COMMAND_CAPACITY {
match commands.try_recv() {
Ok(Command::Shutdown) => {
session.close();
state.disconnected(None);
return;
}
Ok(command) => handle_command(command, &mut session, &state),
Err(TryRecvError::Empty) => break,
Err(TryRecvError::Disconnected) => return,
Err(TryRecvError::Disconnected) => break 'worker,
}
}
@@ -168,48 +237,116 @@ fn worker(commands: Receiver<Command>, state: Arc<SharedState>, local: Principal
thread::sleep(IDLE_POLL);
continue;
}
if stopping.load(Ordering::Acquire) {
break;
}
if state.link_cancelled() {
session.close(&state);
state.disconnected(None);
continue;
}
let link = session.link.as_ref().unwrap();
let mut descriptor = libc::pollfd {
fd: link.as_raw_fd(),
events: if session.outbound.is_some() {
libc::POLLOUT
} else {
libc::POLLIN
},
revents: 0,
};
// SAFETY: poll receives one initialized descriptor that stays open.
let ready = unsafe { libc::poll(&mut descriptor, 1, IDLE_POLL.as_millis() as i32) };
if stopping.load(Ordering::Acquire) {
break;
}
if state.link_cancelled() {
continue;
}
if ready < 0 {
let error = std::io::Error::last_os_error();
if error.kind() != std::io::ErrorKind::Interrupted {
session.close(&state);
state.disconnected(Some(format!("serial poll failed: {error}")));
}
continue;
}
if descriptor.revents & (libc::POLLERR | libc::POLLHUP | libc::POLLNVAL) != 0 {
session.close(&state);
state.disconnected(Some("serial link closed".into()));
continue;
}
if let Err(error) = service_link(&mut session, &state) {
session.close();
session.close(&state);
state.disconnected(Some(format!("serial link failed: {error}")));
}
}
session.close(&state);
state.disconnected(None);
}
fn handle_command(command: Command, session: &mut Session, state: &SharedState) {
match command {
Command::Connect { path, baud } => {
session.close();
if matches!(command.action, Action::Save | Action::ResetSaved)
&& session.connection != Some(command.intent.context.connection)
{
state.reject_intent(
command.intent,
"queued action belongs to a different serial connection".into(),
);
return;
}
let edits = match state.claim_intent(command.intent) {
Ok(edits) => edits,
Err(error) => {
state.reject_intent(command.intent, error);
return;
}
};
match command.action {
Action::Connect { path, baud } => {
session.close(state);
match open_serial(&path, baud) {
Ok(link) => {
let Some(connection) = state.connection_opened(command.intent, path) else {
drop(link);
state.disconnected(None);
return;
};
session.link = Some(link);
state.connected(path);
session.connection = Some(connection);
}
Err(error) => {
state.disconnected(Some(format!("cannot open serial device: {error}")))
}
}
}
Command::Disconnect => {
session.close();
Action::Disconnect => {
session.close(state);
state.disconnected(None);
}
Command::Save(edits) => {
let request = session.client.save(edits);
action @ (Action::Save | Action::ResetSaved) => {
let request = match action {
Action::Save => session.client.save(edits),
Action::ResetSaved => session.client.reset_saved(),
_ => unreachable!(),
};
match request {
Ok(request) => {
if let Err(error) = write_raw(session, request.raw_bytes()) {
session.close();
if let Err(error) = write_raw(session, request) {
session.close(state);
state.disconnected(Some(format!("serial link failed: {error}")));
}
}
Err(error) => state.worker_failed(error),
}
}
Command::Shutdown => unreachable!("shutdown handled by worker"),
}
}
fn service_link(session: &mut Session, state: &SharedState) -> Result<(), String> {
if !flush_pending(session)? {
return Ok(());
}
let mut bytes = [0_u8; 512];
let read = match session
.link
@@ -221,7 +358,9 @@ fn service_link(session: &mut Session, state: &SharedState) -> Result<(), String
Err(error)
if matches!(
error.kind(),
std::io::ErrorKind::TimedOut | std::io::ErrorKind::WouldBlock
std::io::ErrorKind::TimedOut
| std::io::ErrorKind::WouldBlock
| std::io::ErrorKind::Interrupted
) =>
{
0
@@ -229,14 +368,11 @@ fn service_link(session: &mut Session, state: &SharedState) -> Result<(), String
Err(error) => return Err(error.to_string()),
};
let mut frames = Vec::new();
for byte in &bytes[..read] {
session.receiver.receive_byte(*byte);
if let Some(frame) = session.receiver.take_received() {
frames.push(frame);
}
}
for frame in frames {
let Some(frame) = session.receiver.take_received() else {
continue;
};
let Some(message) = frame.decode() else {
continue;
};
@@ -249,7 +385,7 @@ fn service_link(session: &mut Session, state: &SharedState) -> Result<(), String
{
state.discovering();
let request = session.client.discover(target)?;
write_raw(session, request.raw_bytes())?;
write_raw(session, request)?;
}
let Some(source) = Target::try_new(raw.system_id(), raw.component_id()) else {
continue;
@@ -258,8 +394,10 @@ fn service_link(session: &mut Session, state: &SharedState) -> Result<(), String
handle_client_outcome(session, state, outcome)?;
}
let outcome = session.client.retry_due();
handle_client_outcome(session, state, outcome)?;
if session.outbound.is_none() {
let outcome = session.client.retry_due();
handle_client_outcome(session, state, outcome)?;
}
Ok(())
}
@@ -419,17 +557,35 @@ fn handle_client_outcome(
state.configuration_event(event);
Ok(())
}
Outcome::Send(request) => write_raw(session, request.raw_bytes()),
Outcome::Send(request) => write_raw(session, request),
}
}
fn write_raw(session: &mut Session, bytes: &[u8]) -> Result<(), String> {
session
.link
.as_mut()
.ok_or("serial link is closed")?
.write_all(bytes)
.map_err(|error| error.to_string())
fn write_raw(session: &mut Session, frame: MavlinkFrame) -> Result<(), String> {
if session.outbound.is_some() {
return Err("serial frame is still being transmitted".into());
}
session.outbound = Some(PendingWrite {
frame,
position: 0,
deadline: Instant::now() + WRITE_TIMEOUT,
});
Ok(())
}
/// One nonblocking write per worker iteration. Commands and cancellation are
/// serviced between partial writes, including a stalled kernel transmit queue.
fn flush_pending(session: &mut Session) -> Result<bool, String> {
let Some(write) = &mut session.outbound else {
return Ok(true);
};
if write.advance(session.link.as_mut().ok_or("serial link is closed")?)? {
session.outbound = None;
session.client.transmitted();
Ok(true)
} else {
Ok(false)
}
}
fn decoded_heartbeat_target(frame: &MavlinkFrame, message: &MavMessage) -> Option<Target> {
@@ -452,7 +608,7 @@ fn open_serial(path: &str, baud: SerialBaud) -> Result<File, String> {
let file = OpenOptions::new()
.read(true)
.write(true)
.custom_flags(libc::O_NOCTTY)
.custom_flags(libc::O_NOCTTY | libc::O_NONBLOCK | libc::O_CLOEXEC)
.open(path)
.map_err(|error| error.to_string())?;
let descriptor = file.as_raw_fd();
@@ -492,6 +648,10 @@ fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
}
}
#[cfg(test)]
#[path = "serial_io_tests.rs"]
mod io_tests;
#[cfg(test)]
mod tests {
use super::*;
@@ -746,7 +906,7 @@ mod tests {
&IMU_DATA {
time_us: 123,
acceleration_milligravity: [1, 2, 3],
angular_rate_milliradians_per_second: [-4, 5, -6],
angular_rate_milliradians_per_second: [-34_871, 34_870, -8_731],
temperature_centidegrees_celsius: 2_500,
},
));
@@ -755,7 +915,7 @@ mod tests {
assert_eq!(observation.values["time_us"], 123);
assert_eq!(
observation.values["angular_rate_milliradians_per_second"],
serde_json::json!([-4, 5, -6])
serde_json::json!([-34_871, 34_870, -8_731])
);
assert!(observation.values.as_str().is_none());
}
File diff suppressed because it is too large. Load diff
+84 -44
View File
@@ -1,14 +1,27 @@
const q = selector => document.querySelector(selector);
const pendingActions = new Set();
let latestState = null;
let latestConfig = null;
let configRevision = -1;
let editorStructure = '';
const expandedPackets = new Set();
const packetRows = new Map();
let logSession = null;
let latestLogId = -1;
const views = new ViewSession({
read: () => api('/api/view'),
send: post,
render: view => { renderState(view.state); renderConfig(view.config); },
error: showError,
permissions: renderPermissions,
edited: (before, after) => {
for (const control of q('#draft-fields').elements) {
if (control.dataset.dirty === 'true'
&& sameContext(control.intentContext ?? control.viewContext, before)) {
control.intentContext = after;
}
}
},
});
function setText(element, value) {
const text = value ?? '';
@@ -30,7 +43,9 @@ async function api(path, options = {}) {
const value = await responseValue(response);
if (!response.ok) {
const detail = typeof value === 'object' && value !== null ? value.error : value;
throw Error(detail || `Request failed with HTTP ${response.status}`);
const error = Error(detail || `Request failed with HTTP ${response.status}`);
error.view = value?.view;
throw error;
}
return value;
}
@@ -47,41 +62,30 @@ function showError(message) {
setText(q('#connect-error'), message);
}
async function runAction(id, action) {
pendingActions.add(id);
renderPermissions();
try {
const value = await action();
showError('');
return value;
} catch (error) {
showError(error.message);
return null;
} finally {
pendingActions.delete(id);
renderPermissions();
}
}
q('#connect-form').addEventListener('submit', event => {
event.preventDefault();
runAction('connect', () => post('/api/connect', {path: q('#path').value}));
views.mutate('/api/connect', {path: q('#path').value}, q('#connect-serial').viewContext);
});
q('#disconnect').addEventListener('click', () => {
runAction('disconnect', () => post('/api/disconnect'));
views.mutate('/api/disconnect', {}, q('#disconnect').viewContext);
});
q('#draft-fields').addEventListener('submit', event => event.preventDefault());
q('#save').addEventListener('click', () => {
runAction('save', () => post('/api/config/save'));
views.mutate('/api/config/save', {}, q('#save').viewContext);
});
q('#reset-saved').addEventListener('click', () => {
views.mutate('/api/config/reset', {}, q('#reset-saved').viewContext);
});
function fieldStructure(field) {
return [
field.id, field.key, field.label, field.group, field.unit, field.kind,
field.minimum, field.maximum, field.factory, field.help, field.options,
field.finite, field.nonzero,
];
}
@@ -104,24 +108,37 @@ function fieldControl(field) {
if (field.kind === 'bool') {
control.type = 'checkbox';
} else {
control.type = 'number';
// Text retains signed zero and IEEE bit strings for nonfinite values.
control.type = field.kind === 'decimal' ? 'text' : 'number';
control.step = field.kind === 'decimal' ? 'any' : '1';
control.inputMode = field.kind === 'decimal' ? 'decimal' : 'numeric';
if (field.minimum !== null) control.min = String(field.minimum);
if (field.maximum !== null) control.max = String(field.maximum);
}
}
control.addEventListener('input', () => {
if (control.dataset.dirty !== 'true') control.intentContext = control.viewContext;
control.dataset.dirty = 'true';
renderPermissions();
});
control.addEventListener('change', async () => {
let value = control.value;
if (field.kind === 'bool') {
value = field.options.length === 0 ? control.checked : control.value === 'true';
}
const action = `edit:${field.id}`;
const config = await runAction(action, () => post('/api/config/edit', {
control.dataset.dirty = 'true';
const displayed = control.type === 'checkbox' ? control.checked : control.value;
await views.mutate('/api/config/edit', {
field_id: field.id,
value,
}));
if (config !== null) renderConfig(config);
}, control.intentContext ?? control.viewContext);
if (displayed === (control.type === 'checkbox' ? control.checked : control.value)
&& !views.editing(field.id)) {
delete control.dataset.dirty;
delete control.intentContext;
}
if (latestConfig) syncEditorValues(latestConfig.fields);
renderPermissions();
});
return control;
}
@@ -148,11 +165,19 @@ function buildEditor(fields) {
key.textContent = `${field.key} [${field.id}]`;
const control = fieldControl(field);
label.append(title, key, control);
const values = document.createElement('small');
values.id = `values-${field.id}`;
const error = document.createElement('small');
error.id = `error-${field.id}`;
error.className = 'error';
error.setAttribute('role', 'alert');
label.append(values, error);
const bounds = field.minimum !== null || field.maximum !== null
? `Inclusive bounds: ${field.minimum ?? 'unbounded'} to ${field.maximum ?? 'unbounded'}.`
: '';
const helpText = [field.help, `Factory: ${field.factory}.`, bounds].filter(Boolean).join(' ');
const rules = [field.finite ? 'Must be finite.' : '', field.nonzero ? 'Must be nonzero.' : ''].filter(Boolean).join(' ');
const helpText = [field.help, `Factory: ${field.factory}.`, bounds, rules].filter(Boolean).join(' ');
if (helpText) {
const help = document.createElement('small');
help.id = `help-${field.id}`;
@@ -171,9 +196,18 @@ function syncEditorValues(fields) {
for (const field of fields) {
const control = q(`[data-field-id="${field.id}"]`);
if (!control) continue;
const display = value => {
if (value === null) return 'unknown';
return field.options.find(option => String(option.value) === String(value))?.label ?? String(value);
};
setText(q(`#values-${field.id}`), `Boot-active: ${display(field.active)} · Saved: ${display(field.saved)} · Draft: ${field.draft === null ? 'unchanged' : display(field.draft)}`);
setText(q(`#error-${field.id}`), field.error ?? '');
control.setAttribute('aria-invalid', String(field.error !== null));
if (views.editing(field.id) || control.dataset.dirty === 'true') continue;
control.viewContext = latestState.context;
if (field.kind === 'bool' && field.options.length === 0) {
if (document.activeElement !== control) control.checked = field.value;
} else if (document.activeElement !== control && control.value !== String(field.value)) {
control.checked = field.value;
} else if (control.value !== String(field.value)) {
control.value = String(field.value);
}
}
@@ -191,25 +225,32 @@ function renderConfig(config) {
q('#restart-guidance').hidden = config.state !== 'restart required';
renderValidation(config.validation);
const structure = JSON.stringify(config.fields.map(fieldStructure));
const structure = JSON.stringify([
latestState.context.instance, latestState.context.connection, latestState.context.configuration,
config.fields.map(fieldStructure),
]);
if (structure !== editorStructure) {
editorStructure = structure;
buildEditor(config.fields);
}
syncEditorValues(config.fields);
q('#save').viewContext = latestState.context;
q('#reset-saved').viewContext = latestState.context;
renderPermissions();
}
function renderPermissions() {
if (latestState === null) return;
const busy = pendingActions.size !== 0;
const busy = views.busy;
const dirty = [...q('#draft-fields').elements].some(control => control.dataset.dirty === 'true');
q('#path').disabled = latestState.can_disconnect || busy;
q('#connect-serial').disabled = latestState.can_disconnect || busy;
q('#disconnect').hidden = !latestState.can_disconnect;
q('#disconnect').disabled = busy;
q('#save').disabled = !latestState.can_save || busy;
q('#save').disabled = !latestState.can_save || busy || dirty;
q('#reset-saved').disabled = !latestState.can_reset || busy;
for (const control of q('#draft-fields').elements) {
control.disabled = !latestState.can_edit || busy;
control.disabled = !latestState.can_edit || (busy && !views.onlyEdits);
}
}
@@ -406,24 +447,23 @@ function renderState(state) {
q('#configuration').hidden = !state.has_config;
q('#packet-monitor').hidden = !state.can_disconnect;
q('#log-monitor').hidden = !state.can_disconnect;
q('#connect-serial').viewContext = state.context;
q('#disconnect').viewContext = state.context;
renderPermissions();
}
async function refresh() {
try {
const [state, packets, logs] = await Promise.all([
api('/api/state'),
const context = views.view?.state.context;
const [, packets, logs] = await Promise.all([
views.refresh(),
api('/api/packets'),
api('/api/logs'),
]);
renderState(state);
renderPackets(packets);
renderLogs(logs);
if (state.config_revision !== configRevision) {
const config = await api('/api/config');
renderConfig(config);
configRevision = state.config_revision;
q('#configuration').hidden = latestConfig.schema_id === null;
if (context?.connection === views.view?.state.context.connection
&& context?.instance === views.view?.state.context.instance) {
renderPackets(packets);
renderLogs(logs);
}
} catch (error) {
showError(error.message);
@@ -35,11 +35,15 @@
<aside id="restart-guidance" class="important" role="status" hidden>
Saved configuration becomes active after the controller restarts.
The current serial link keeps its boot-active settings until restart.
</aside>
<p class="hint">Inputs edit the saved draft. Boot-active values stay in use until restart.
USB remains available for configuration and recovery.</p>
<ul id="validation"></ul>
<form id="draft-fields" class="draft-fields"></form>
<div class="actions" aria-label="Configuration actions">
<button id="save" type="button">Save</button>
<button id="reset-saved" type="button">Reset saved configuration</button>
</div>
</section>
@@ -69,6 +73,7 @@
</div>
</section>
</main>
<script src="/view.js"></script>
<script src="/app.js"></script>
</body>
</html>
+102
View File
@@ -0,0 +1,102 @@
// Local view ordering and intent binding. Contexts are not credentials.
function sameContext(left, right) {
return ['instance', 'connection', 'configuration', 'draft'].every(key => left?.[key] === right?.[key]);
}
class ViewSession {
constructor({read, send, render, error, permissions, edited = () => {}}) {
Object.assign(this, {read, send, render, error, permissions, edited});
this.view = null;
this.queue = [];
this.ticket = 0;
this.acceptedTicket = 0;
this.retired = new Set();
this.message = '';
}
get busy() { return this.queue.length !== 0; }
get onlyEdits() { return this.queue.every(entry => entry.path === '/api/config/edit'); }
editing(id) {
return this.queue.some(entry => entry.body.field_id === id
&& entry.context.instance === this.view?.state.context.instance
&& entry.context.connection === this.view?.state.context.connection
&& entry.context.configuration === this.view?.state.context.configuration);
}
accept(view, ticket) {
const current = this.view?.state;
const next = view.state;
if (current) {
if (current.context.instance === next.context.instance) {
if (BigInt(next.config_revision) < BigInt(current.config_revision)) return false;
} else {
if (ticket < this.acceptedTicket || this.retired.has(next.context.instance)) return false;
this.retired.add(current.context.instance);
}
}
this.acceptedTicket = Math.max(ticket, this.acceptedTicket);
this.view = view;
this.render(view);
if (this.message) this.error(this.message);
return true;
}
async refresh() {
const ticket = ++this.ticket;
const view = await this.read();
return this.accept(view, ticket);
}
mutate(path, body = {}, context = this.view?.state.context) {
if (!this.view) return Promise.resolve(null);
return new Promise(resolve => {
this.queue.push({path, body, context, resolve});
this.permissions();
if (this.queue.length === 1) this.drain();
});
}
async drain() {
while (this.queue.length) {
const entry = this.queue[0];
const ticket = ++this.ticket;
let result = null;
try {
if (!sameContext(entry.context, this.view.state.context)) {
const error = Error('View changed; queued action discarded. Review the current configuration.');
error.staleIntent = true;
throw error;
}
const view = await this.send(entry.path, {...entry.body, context: entry.context});
if (this.accept(view, ticket)) {
this.message = '';
result = view;
// Only our own accepted edits can advance already queued edits.
// A poll, another editor, a reset, or a new connection cannot rebase intent.
if (entry.path === '/api/config/edit'
&& view.state.context.instance === entry.context.instance
&& view.state.context.connection === entry.context.connection
&& view.state.context.configuration === entry.context.configuration) {
for (const queued of this.queue.slice(1)) {
if (sameContext(queued.context, entry.context)) queued.context = view.state.context;
}
this.edited(entry.context, view.state.context);
}
}
} catch (error) {
const relevant = error.view
? this.accept(error.view, ticket)
: error.staleIntent || sameContext(entry.context, this.view.state.context);
if (relevant) this.message = error.message;
} finally {
this.queue.shift();
this.render(this.view);
if (this.message) this.error(this.message);
this.permissions();
entry.resolve(result);
}
}
}
}
if (typeof module !== 'undefined') module.exports = {ViewSession, sameContext};
@@ -0,0 +1,160 @@
const {test} = require('node:test');
const assert = require('node:assert/strict');
const {ViewSession, sameContext} = require('./view.js');
function view(revision, connection = '1', draft = '0', value = 20, instance = 'host') {
return {
state: {config_revision: String(revision), context: {instance, connection, configuration: '1', draft}},
config: {fields: [{id: 401, value}]},
};
}
function deferred() {
let resolve, reject;
const promise = new Promise((yes, no) => { resolve = yes; reject = no; });
return {promise, resolve, reject};
}
function harness(send = async () => { throw Error('unexpected mutation'); }, edited = () => {}) {
const reads = [], rendered = [], errors = [];
const session = new ViewSession({
read: () => { const next = deferred(); reads.push(next); return next.promise; },
send, edited, render: snapshot => rendered.push(snapshot), error: message => errors.push(message), permissions() {},
});
return {session, reads, rendered, errors};
}
test('late A view cannot replace displayed B or supply a reset context', async () => {
const sent = [];
const h = harness(async (path, body) => { sent.push({path, body}); return view(4, '2'); });
h.session.accept(view(1), 0);
const old = h.session.refresh();
const fresh = h.session.refresh();
h.reads[1].resolve(view(3, '2'));
await fresh;
h.reads[0].resolve(view(2));
assert.equal(await old, false);
assert.equal(h.rendered.at(-1).state.context.connection, '2');
await h.session.mutate('/api/config/reset');
assert.equal(sent[0].body.context.connection, '2');
});
test('an A action already queued is discarded rather than rebound to B', async () => {
const first = deferred(), sent = [];
const h = harness((path, body) => { sent.push({path, body}); return first.promise; });
h.session.accept(view(1), 0);
const edit = h.session.mutate('/api/config/edit', {field_id: 401, value: 30});
const reset = h.session.mutate('/api/config/reset');
const fresh = h.session.refresh();
h.reads[0].resolve(view(5, '2'));
await fresh;
first.resolve(view(2, '1', '1', 30));
await Promise.all([edit, reset]);
assert.equal(sent.length, 1);
assert.equal(h.rendered.at(-1).state.context.connection, '2');
assert.match(h.errors.at(-1), /queued action discarded/);
});
test('an old input change event retains the context of its displayed value', async () => {
const h = harness();
const a = view(1);
h.session.accept(a, 0);
h.session.accept(view(3, '2'), 1);
assert.equal(await h.session.mutate('/api/config/edit', {field_id: 401, value: 30}, a.state.context), null);
assert.match(h.errors.at(-1), /queued action discarded/);
assert.equal(h.rendered.at(-1).state.context.connection, '2');
});
test('rapid same-input edits serialize and only our accepted draft advances queued context', async () => {
const replies = [deferred(), deferred(), deferred()], sent = [];
let displayedContext = view('9007199254740993').state.context;
const h = harness((path, body) => { sent.push({path, body}); return replies[sent.length - 1].promise; },
(before, after) => { if (sameContext(displayedContext, before)) displayedContext = after; });
h.session.accept(view('9007199254740993'), 0);
const first = h.session.mutate('/api/config/edit', {field_id: 401, value: 30}, displayedContext);
const second = h.session.mutate('/api/config/edit', {field_id: 401, value: 40}, displayedContext);
assert.equal(sent.length, 1);
assert.equal(h.session.editing(401), true);
replies[0].resolve(view('9007199254740994', '1', '1', 30));
await first;
assert.equal(sent.length, 2);
assert.equal(sent[1].body.context.draft, '1');
assert.equal(sent[1].body.value, 40);
assert.equal(h.session.editing(401), true);
const third = h.session.mutate('/api/config/edit', {field_id: 401, value: 50}, displayedContext);
replies[1].resolve(view('9007199254740995', '1', '2', 40));
await second;
assert.equal(sent[2].body.context.draft, '2');
replies[2].resolve(view('9007199254740996', '1', '3', 50));
await third;
assert.equal(h.session.editing(401), false);
assert.equal(h.rendered.at(-1).config.fields[0].value, 50);
assert.equal(h.session.accept(view('9007199254740994', '1', '1', 30), 100), false);
});
test('another editors draft cannot silently rebase queued local edits', async () => {
const reply = deferred(), sent = [];
const h = harness((path, body) => { sent.push(body); return reply.promise; });
h.session.accept(view(1), 0);
const first = h.session.mutate('/api/config/edit', {field_id: 401, value: 30});
const next = h.session.mutate('/api/config/edit', {field_id: 401, value: 40});
const poll = h.session.refresh();
h.reads[0].resolve(view(4, '1', '3', 50));
await poll;
reply.resolve(view(2, '1', '1', 30));
await Promise.all([first, next]);
assert.equal(sent.length, 1);
assert.equal(h.rendered.at(-1).config.fields[0].value, 50);
});
test('failed refresh can retry and prior server-instance responses stay retired', async () => {
const h = harness();
const first = h.session.refresh();
h.reads[0].reject(Error('HTTP failure'));
await assert.rejects(first, /HTTP failure/);
const retry = h.session.refresh();
h.reads[1].resolve(view(100));
assert.equal(await retry, true);
const prior = h.session.refresh();
const current = h.session.refresh();
h.reads[3].resolve(view(1, '1', '0', 60, 'new-host'));
await current;
h.reads[2].resolve(view(101));
assert.equal(await prior, false);
assert.equal(h.rendered.at(-1).config.fields[0].value, 60);
});
test('decimal controls retain signed zero and nonfinite bit strings when submitting edits', async () => {
const vm = require('node:vm');
const fs = require('node:fs');
const sent = [];
function element() {
return {
dataset: {}, value: '', events: {},
addEventListener(name, callback) { this.events[name] = callback; },
};
}
const pending = () => new Promise(() => {});
const context = vm.createContext({
document: {querySelector: element, createElement: element},
fetch: pending,
ViewSession: class {
refresh() { return pending(); }
editing() { return false; }
async mutate(path, body) { sent.push({path, body}); }
},
});
vm.runInContext(fs.readFileSync(`${__dirname}/app.js`, 'utf8'), context);
const field = {id: 401, kind: 'decimal', options: [], minimum: null, maximum: null};
context.field = field;
const control = vm.runInContext('fieldControl(field)', context);
assert.equal(control.type, 'text');
for (const value of ['-0', 'bits:0xfff0000000000042', 'bits:0x7f800000']) {
control.value = value;
await control.events.change();
const serialized = JSON.parse(JSON.stringify(sent.at(-1)));
assert.equal(serialized.body.value, value);
assert.equal(serialized.path, '/api/config/edit');
}
const original = vm.runInContext('JSON.stringify(fieldStructure(field))', context);
field.finite = true;
assert.notEqual(vm.runInContext('JSON.stringify(fieldStructure(field))', context), original);
});