1646 lines
59 KiB
Rust
1646 lines
59 KiB
Rust
//! Board-agnostic configuration client over Kaptan's generic file service.
|
|
|
|
use std::time::{Duration, Instant};
|
|
|
|
use kaptan_config::{ConfigValue, FieldId, KcfgResultDetail, ValueKind};
|
|
use kaptan_file_mavlink::{
|
|
Content, ControlRequest, ControlResponse, DATA_BYTES, DataRequest, DataResponse, FileOperation,
|
|
FileRequest, FileResponse, FileStatus, Principal, TRANSFER_REPLAY_LEASE_SECONDS, Target,
|
|
crc32c,
|
|
wire::{decode_response, encode_request},
|
|
};
|
|
#[cfg(test)]
|
|
use kaptan_mavlink::receiver::IntegrityMavlinkReceiver;
|
|
use kaptan_mavlink::{MavlinkFrame, encoder::MavlinkEncoder, generated::kaptan::MavMessage};
|
|
|
|
const CONFIG_PATH: &[u8] = b"/config/user.kcfg";
|
|
const CHUNK_BYTES: usize = DATA_BYTES;
|
|
const MAX_CONFIG_FILE_BYTES: usize = 1024 * 1024;
|
|
const REQUEST_TIMEOUT: Duration = Duration::from_millis(500);
|
|
// Closing a write may erase a 128-KiB STM32F7 sector. The controller cannot
|
|
// service serial traffic during that erase, so this operation gets its own
|
|
// bounded timeout.
|
|
const CLOSE_WRITE_TIMEOUT: Duration = Duration::from_secs(5);
|
|
const REQUEST_ATTEMPTS: u8 = 3;
|
|
// The extra second keeps the retry beyond the controller's lease despite
|
|
// independent host/controller clocks and service-loop scheduling.
|
|
const OPEN_CONFLICT_LEASE_WAIT: Duration =
|
|
Duration::from_secs(TRANSFER_REPLAY_LEASE_SECONDS as u64 + 1);
|
|
|
|
#[cfg(test)]
|
|
fn parse_test_message(frame: &MavlinkFrame) -> Option<MavMessage> {
|
|
let mut receiver = IntegrityMavlinkReceiver::<1>::new();
|
|
for byte in frame.raw_bytes() {
|
|
receiver.receive_byte(*byte);
|
|
}
|
|
receiver.take_received()?.decode()
|
|
}
|
|
|
|
pub(crate) fn scalar_json(value: ConfigValue) -> serde_json::Value {
|
|
match value {
|
|
ConfigValue::Bool(value) => value.into(),
|
|
ConfigValue::U8(value) => value.into(),
|
|
ConfigValue::U16(value) => value.into(),
|
|
ConfigValue::I32(value) => value.into(),
|
|
ConfigValue::U32(value) => value.into(),
|
|
// JSON numbers cannot preserve the full u64 domain in a browser.
|
|
ConfigValue::U64(value) => value.to_string().into(),
|
|
ConfigValue::F32(value) => serde_json::Value::from(value),
|
|
ConfigValue::F64(value) => serde_json::Value::from(value),
|
|
}
|
|
}
|
|
|
|
pub(crate) const fn scalar_less_than(left: ConfigValue, right: ConfigValue) -> bool {
|
|
match (left, right) {
|
|
(ConfigValue::I32(left), ConfigValue::I32(right)) => left < right,
|
|
(ConfigValue::U8(left), ConfigValue::U8(right)) => left < right,
|
|
(ConfigValue::U16(left), ConfigValue::U16(right)) => left < right,
|
|
(ConfigValue::U32(left), ConfigValue::U32(right)) => left < right,
|
|
(ConfigValue::U64(left), ConfigValue::U64(right)) => left < right,
|
|
(ConfigValue::F32(left), ConfigValue::F32(right)) => left < right,
|
|
(ConfigValue::F64(left), ConfigValue::F64(right)) => left < right,
|
|
(ConfigValue::Bool(left), ConfigValue::Bool(right)) => !left && right,
|
|
_ => false,
|
|
}
|
|
}
|
|
|
|
#[derive(Clone)]
|
|
pub(crate) struct RemoteEnumOption {
|
|
pub(crate) value: u8,
|
|
pub(crate) label: String,
|
|
}
|
|
|
|
#[derive(Clone)]
|
|
pub(crate) struct RemoteField {
|
|
pub(crate) id: FieldId,
|
|
pub(crate) key: String,
|
|
pub(crate) unit: String,
|
|
pub(crate) kind: ValueKind,
|
|
pub(crate) minimum: Option<ConfigValue>,
|
|
pub(crate) maximum: Option<ConfigValue>,
|
|
pub(crate) factory: ConfigValue,
|
|
pub(crate) value: ConfigValue,
|
|
pub(crate) options: Vec<RemoteEnumOption>,
|
|
}
|
|
|
|
#[derive(Clone)]
|
|
pub(crate) struct RemoteConfig {
|
|
pub(crate) schema_id: [u8; 16],
|
|
pub(crate) restart_required: bool,
|
|
pub(crate) fields: Vec<RemoteField>,
|
|
}
|
|
|
|
pub(crate) struct Edit {
|
|
pub(crate) field_id: FieldId,
|
|
pub(crate) value: ConfigValue,
|
|
}
|
|
|
|
pub(crate) enum ClientEvent {
|
|
Configuration(RemoteConfig),
|
|
Failure(String),
|
|
RediscoveryRequired(String),
|
|
DurableStateUnknown(String),
|
|
}
|
|
|
|
#[expect(
|
|
clippy::large_enum_variant,
|
|
reason = "the bounded host result keeps one encoded frame allocation-free"
|
|
)]
|
|
pub(crate) enum Outcome {
|
|
Idle,
|
|
Event(ClientEvent),
|
|
Send(MavlinkFrame),
|
|
}
|
|
|
|
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
|
enum PendingKind {
|
|
OpenRead { transfer_id: u16 },
|
|
Read { position: u32 },
|
|
CloseRead { position: u32 },
|
|
OpenWrite,
|
|
Write { position: u32, next: u32 },
|
|
CloseWrite { position: u32 },
|
|
Abort { transfer_id: u16 },
|
|
}
|
|
|
|
struct Pending {
|
|
kind: PendingKind,
|
|
raw: MavlinkFrame,
|
|
sent_at: Instant,
|
|
attempts_left: u8,
|
|
}
|
|
|
|
struct ReadTransfer {
|
|
id: u16,
|
|
size: u32,
|
|
checksum: u32,
|
|
bytes: Vec<u8>,
|
|
}
|
|
|
|
struct WriteTransfer {
|
|
id: u16,
|
|
bytes: Vec<u8>,
|
|
position: u32,
|
|
checksum: u32,
|
|
}
|
|
|
|
enum Transfer {
|
|
Read(ReadTransfer),
|
|
Write(WriteTransfer),
|
|
}
|
|
|
|
struct DownloadedConfig {
|
|
document: Vec<u8>,
|
|
config: RemoteConfig,
|
|
}
|
|
|
|
#[derive(Clone, Copy)]
|
|
enum OpenConflictRecovery {
|
|
Disabled,
|
|
FreshId,
|
|
LeaseDelay,
|
|
Waiting(Instant),
|
|
}
|
|
|
|
/// Downloads and imports exactly one self-describing configuration artifact.
|
|
///
|
|
/// The client contains no board IDs, field names, schema IDs, or field catalog.
|
|
/// The sole known resource is the stable generic virtual path.
|
|
pub(crate) struct ConfigClient {
|
|
local: Principal,
|
|
encoder: MavlinkEncoder,
|
|
next_transfer_id: u16,
|
|
target: Option<Target>,
|
|
downloaded: Option<DownloadedConfig>,
|
|
transfer: Option<Transfer>,
|
|
pending: Option<Pending>,
|
|
abort_failure: Option<String>,
|
|
open_conflict_recovery: OpenConflictRecovery,
|
|
}
|
|
|
|
impl ConfigClient {
|
|
pub(crate) const fn new(local: Principal) -> Self {
|
|
Self {
|
|
local,
|
|
encoder: MavlinkEncoder::new(local.system(), local.component()),
|
|
next_transfer_id: 1,
|
|
target: None,
|
|
downloaded: None,
|
|
transfer: None,
|
|
pending: None,
|
|
abort_failure: None,
|
|
open_conflict_recovery: OpenConflictRecovery::Disabled,
|
|
}
|
|
}
|
|
|
|
pub(crate) fn reset(&mut self) {
|
|
self.target = None;
|
|
self.downloaded = None;
|
|
self.transfer = None;
|
|
self.pending = None;
|
|
self.abort_failure = None;
|
|
self.open_conflict_recovery = OpenConflictRecovery::Disabled;
|
|
}
|
|
|
|
pub(crate) fn discover(&mut self, target: Target) -> Result<MavlinkFrame, String> {
|
|
self.reset();
|
|
self.target = Some(target);
|
|
self.open_conflict_recovery = OpenConflictRecovery::FreshId;
|
|
self.request_open_read()
|
|
}
|
|
|
|
fn request_open_read(&mut self) -> Result<MavlinkFrame, String> {
|
|
let target = self.target()?;
|
|
let transfer_id = self.transfer_id();
|
|
Ok(self.request_control(
|
|
PendingKind::OpenRead { transfer_id },
|
|
ControlRequest {
|
|
operation: FileOperation::OpenRead,
|
|
target,
|
|
position: 0,
|
|
file_size: 0,
|
|
checksum_crc32c: 0,
|
|
transfer_id,
|
|
content: Content::try_new(CONFIG_PATH).expect("static config path is bounded"),
|
|
},
|
|
))
|
|
}
|
|
|
|
pub(crate) fn save(&mut self, edits: Vec<Edit>) -> Result<MavlinkFrame, String> {
|
|
if edits.is_empty() {
|
|
return Err("configuration has no changed fields".into());
|
|
}
|
|
if self.pending.is_some() || self.transfer.is_some() {
|
|
return Err("configuration file transfer is busy".into());
|
|
}
|
|
let downloaded = self
|
|
.downloaded
|
|
.as_ref()
|
|
.ok_or("configuration has not been downloaded")?;
|
|
validate_edits(&downloaded.config, &edits)?;
|
|
let bytes = encode_edited_document(&downloaded.document, &edits)?;
|
|
let size = u32::try_from(bytes.len())
|
|
.map_err(|_| "configuration artifact exceeds the file protocol".to_owned())?;
|
|
let checksum = crc32c(&bytes);
|
|
let transfer_id = self.transfer_id();
|
|
let target = self.target()?;
|
|
self.transfer = Some(Transfer::Write(WriteTransfer {
|
|
id: transfer_id,
|
|
bytes,
|
|
position: 0,
|
|
checksum,
|
|
}));
|
|
Ok(self.request_control(
|
|
PendingKind::OpenWrite,
|
|
ControlRequest {
|
|
operation: FileOperation::OpenWrite,
|
|
target,
|
|
position: 0,
|
|
file_size: size,
|
|
checksum_crc32c: checksum,
|
|
transfer_id,
|
|
content: Content::try_new(CONFIG_PATH).expect("static config path is bounded"),
|
|
},
|
|
))
|
|
}
|
|
|
|
pub(crate) fn observe(&mut self, source: Target, message: &MavMessage) -> Outcome {
|
|
let Some(target) = self.target else {
|
|
return Outcome::Idle;
|
|
};
|
|
if source != target {
|
|
return Outcome::Idle;
|
|
}
|
|
match decode_response(message) {
|
|
Ok(FileResponse::Control(control)) => self.observe_control(control),
|
|
Ok(FileResponse::Data(data)) => self.observe_data(data),
|
|
Err(_) => Outcome::Idle,
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
fn observe_frame(&mut self, frame: &MavlinkFrame) -> Outcome {
|
|
let Some(message) = parse_test_message(frame) else {
|
|
return Outcome::Idle;
|
|
};
|
|
let Some(source) = Target::try_new(frame.system_id(), frame.component_id()) else {
|
|
return Outcome::Idle;
|
|
};
|
|
self.observe(source, &message)
|
|
}
|
|
|
|
pub(crate) fn retry_due(&mut self) -> Outcome {
|
|
if let OpenConflictRecovery::Waiting(retry_at) = self.open_conflict_recovery {
|
|
if Instant::now() < retry_at {
|
|
return Outcome::Idle;
|
|
}
|
|
self.open_conflict_recovery = OpenConflictRecovery::Disabled;
|
|
let request = self.request_open_read();
|
|
return self.follow_up(request);
|
|
}
|
|
let Some(pending) = self.pending.as_mut() else {
|
|
return Outcome::Idle;
|
|
};
|
|
if pending.sent_at.elapsed() < timeout(pending.kind) {
|
|
return Outcome::Idle;
|
|
}
|
|
if pending.attempts_left != 0 {
|
|
pending.attempts_left -= 1;
|
|
pending.sent_at = Instant::now();
|
|
return Outcome::Send(pending.raw);
|
|
}
|
|
let kind = pending.kind;
|
|
self.pending = None;
|
|
let detail = format!(
|
|
"configuration file {} timed out after bounded retries",
|
|
operation_name(kind)
|
|
);
|
|
if matches!(kind, PendingKind::CloseWrite { .. }) {
|
|
return self.durable_state_unknown(format!(
|
|
"{detail}; the durable state is unknown, so power-cycle the controller and reconnect before another save"
|
|
));
|
|
}
|
|
if matches!(kind, PendingKind::Abort { .. }) {
|
|
let original = self.abort_failure.take().unwrap_or(detail);
|
|
return self.rediscovery_required(format!(
|
|
"{original}; the file transfer could not be released"
|
|
));
|
|
}
|
|
self.abort_after(detail)
|
|
}
|
|
|
|
fn observe_control(&mut self, data: ControlResponse) -> Outcome {
|
|
if data.target != self.local_target() {
|
|
return Outcome::Idle;
|
|
}
|
|
let Some(kind) = self.pending.as_ref().map(|pending| pending.kind) else {
|
|
return Outcome::Idle;
|
|
};
|
|
if !control_matches(kind, &data) {
|
|
// A delayed duplicate may arrive after its retry was already
|
|
// accepted. It must not disturb the current stop-and-wait request.
|
|
return Outcome::Idle;
|
|
}
|
|
if !self.response_identity_matches(kind, data.transfer_id) {
|
|
return Outcome::Idle;
|
|
}
|
|
self.pending = None;
|
|
if matches!(kind, PendingKind::Abort { .. }) {
|
|
let original = self
|
|
.abort_failure
|
|
.take()
|
|
.unwrap_or_else(|| "configuration file transfer aborted".into());
|
|
self.transfer = None;
|
|
return if data.status == FileStatus::Ok {
|
|
self.fail(original)
|
|
} else {
|
|
self.rediscovery_required(format!(
|
|
"{original}; controller rejected the transfer abort: {:?}",
|
|
data.status
|
|
))
|
|
};
|
|
}
|
|
if matches!(kind, PendingKind::OpenRead { .. })
|
|
&& data.status == FileStatus::RequestConflict
|
|
{
|
|
self.transfer = None;
|
|
match self.open_conflict_recovery {
|
|
OpenConflictRecovery::FreshId => {
|
|
self.open_conflict_recovery = OpenConflictRecovery::LeaseDelay;
|
|
let request = self.request_open_read();
|
|
return self.follow_up(request);
|
|
}
|
|
OpenConflictRecovery::LeaseDelay => {
|
|
self.open_conflict_recovery =
|
|
OpenConflictRecovery::Waiting(Instant::now() + OPEN_CONFLICT_LEASE_WAIT);
|
|
return Outcome::Idle;
|
|
}
|
|
OpenConflictRecovery::Disabled | OpenConflictRecovery::Waiting(_) => {}
|
|
}
|
|
}
|
|
if data.status != FileStatus::Ok {
|
|
let detail = rejected_detail(&data);
|
|
return match kind {
|
|
PendingKind::OpenRead { .. } => self.rediscovery_required(detail),
|
|
PendingKind::OpenWrite => self.fail(detail),
|
|
PendingKind::CloseWrite { .. } if data.status == FileStatus::StorageFailure => self
|
|
.durable_state_unknown(format!(
|
|
"{detail}; power-cycle the controller and reconnect before another save"
|
|
)),
|
|
PendingKind::Read { .. }
|
|
| PendingKind::Write { .. }
|
|
| PendingKind::CloseRead { .. }
|
|
| PendingKind::CloseWrite { .. }
|
|
if data.status == FileStatus::InvalidTransfer =>
|
|
{
|
|
self.rediscovery_required(detail)
|
|
}
|
|
PendingKind::Read { .. }
|
|
| PendingKind::Write { .. }
|
|
| PendingKind::CloseRead { .. }
|
|
| PendingKind::CloseWrite { .. }
|
|
if matches!(
|
|
data.status,
|
|
FileStatus::InvalidPosition
|
|
| FileStatus::InvalidState
|
|
| FileStatus::RequestConflict
|
|
) =>
|
|
{
|
|
self.abort_after(detail)
|
|
}
|
|
PendingKind::Read { .. }
|
|
| PendingKind::Write { .. }
|
|
| PendingKind::CloseRead { .. }
|
|
| PendingKind::CloseWrite { .. } => self.fail(detail),
|
|
PendingKind::Abort { .. } => unreachable!("abort responses returned above"),
|
|
};
|
|
}
|
|
match kind {
|
|
PendingKind::OpenRead { transfer_id } => {
|
|
self.open_conflict_recovery = OpenConflictRecovery::Disabled;
|
|
self.opened_read(data, transfer_id)
|
|
}
|
|
PendingKind::CloseRead { .. } => self.closed_read(data),
|
|
PendingKind::OpenWrite => self.opened_write(data),
|
|
PendingKind::Write { next, .. } => self.accepted_write(data, next),
|
|
PendingKind::CloseWrite { .. } => self.closed_write(data),
|
|
PendingKind::Abort { .. } => unreachable!("abort responses returned above"),
|
|
PendingKind::Read { .. } => Outcome::Idle,
|
|
}
|
|
}
|
|
|
|
fn observe_data(&mut self, data: DataResponse) -> Outcome {
|
|
if data.target != self.local_target() {
|
|
return Outcome::Idle;
|
|
}
|
|
let Some(Pending {
|
|
kind: PendingKind::Read { position },
|
|
..
|
|
}) = self.pending.as_ref()
|
|
else {
|
|
return Outcome::Idle;
|
|
};
|
|
let position = *position;
|
|
let Some(Transfer::Read(read)) = self.transfer.as_mut() else {
|
|
return self.fail("configuration read response has no open transfer");
|
|
};
|
|
if data.transfer_id != read.id || data.position != position {
|
|
return Outcome::Idle;
|
|
}
|
|
let remaining = read.size.saturating_sub(position);
|
|
let count = usize::try_from(remaining.min(CHUNK_BYTES as u32))
|
|
.expect("bounded file chunk length fits usize");
|
|
if count == 0 || read.bytes.len() != position as usize {
|
|
return self.abort_after("controller returned an invalid file position".into());
|
|
}
|
|
if data.data[count..].iter().any(|byte| *byte != 0) {
|
|
return self.abort_after("controller returned a noncanonical file chunk".into());
|
|
}
|
|
read.bytes.extend_from_slice(&data.data[..count]);
|
|
self.pending = None;
|
|
let next = position + count as u32;
|
|
if next < read.size {
|
|
let request = self.request_read(next);
|
|
self.follow_up(request)
|
|
} else if crc32c(&read.bytes) != read.checksum {
|
|
self.abort_after("configuration file checksum did not match".into())
|
|
} else {
|
|
let transfer_id = read.id;
|
|
let document = std::mem::take(&mut read.bytes);
|
|
let config = match decode_document(&document) {
|
|
Ok(config) => config,
|
|
Err(error) => return self.abort_after(error),
|
|
};
|
|
self.downloaded = Some(DownloadedConfig { document, config });
|
|
let request =
|
|
self.request_close(PendingKind::CloseRead { position: next }, transfer_id, next);
|
|
self.follow_up(request)
|
|
}
|
|
}
|
|
|
|
fn opened_read(&mut self, data: ControlResponse, requested_id: u16) -> Outcome {
|
|
let Ok(size) = usize::try_from(data.file_size) else {
|
|
return self.fail("controller returned an unsupported configuration file size");
|
|
};
|
|
if data.status != FileStatus::Ok
|
|
|| data.transfer_id != requested_id
|
|
|| data.position != 0
|
|
|| size == 0
|
|
|| size > MAX_CONFIG_FILE_BYTES
|
|
{
|
|
return self.abort_transfer_after(
|
|
requested_id,
|
|
"controller returned an invalid configuration file size".into(),
|
|
);
|
|
}
|
|
self.transfer = Some(Transfer::Read(ReadTransfer {
|
|
id: data.transfer_id,
|
|
size: data.file_size,
|
|
checksum: data.checksum_crc32c,
|
|
bytes: Vec::with_capacity(size),
|
|
}));
|
|
let request = self.request_read(0);
|
|
self.follow_up(request)
|
|
}
|
|
|
|
fn closed_read(&mut self, data: ControlResponse) -> Outcome {
|
|
let expected_position = match self.transfer.as_ref() {
|
|
Some(Transfer::Read(read)) => read.size,
|
|
_ => return self.fail("configuration export close has no transfer"),
|
|
};
|
|
if data.status != FileStatus::Ok || data.position != expected_position {
|
|
return self.fail("controller did not close the configuration export");
|
|
}
|
|
self.transfer = None;
|
|
let Some(config) = self
|
|
.downloaded
|
|
.as_ref()
|
|
.map(|downloaded| downloaded.config.clone())
|
|
else {
|
|
return self.fail("configuration artifact was not decoded");
|
|
};
|
|
Outcome::Event(ClientEvent::Configuration(config))
|
|
}
|
|
|
|
fn opened_write(&mut self, data: ControlResponse) -> Outcome {
|
|
let Some(Transfer::Write(write)) = self.transfer.as_mut() else {
|
|
return self.fail("configuration write response has no pending artifact");
|
|
};
|
|
if data.status != FileStatus::Ok
|
|
|| data.transfer_id == 0
|
|
|| data.transfer_id != write.id
|
|
|| data.position != 0
|
|
|| usize::try_from(data.file_size).ok() != Some(write.bytes.len())
|
|
|| data.checksum_crc32c != write.checksum
|
|
{
|
|
return self.abort_after("controller returned an invalid write transfer".into());
|
|
}
|
|
let request = self.request_write();
|
|
self.follow_up(request)
|
|
}
|
|
|
|
fn accepted_write(&mut self, data: ControlResponse, next: u32) -> Outcome {
|
|
let Some(Transfer::Write(write)) = self.transfer.as_mut() else {
|
|
return self.fail("configuration write acknowledgement has no transfer");
|
|
};
|
|
if data.status != FileStatus::Ok
|
|
|| data.transfer_id != write.id
|
|
|| data.position != next
|
|
|| usize::try_from(data.file_size).ok() != Some(write.bytes.len())
|
|
|| data.checksum_crc32c != write.checksum
|
|
{
|
|
return self.abort_after("controller returned an invalid write acknowledgement".into());
|
|
}
|
|
write.position = next;
|
|
if usize::try_from(next).ok() == Some(write.bytes.len()) {
|
|
let transfer_id = write.id;
|
|
let request = self.request_close(
|
|
PendingKind::CloseWrite { position: next },
|
|
transfer_id,
|
|
next,
|
|
);
|
|
self.follow_up(request)
|
|
} else {
|
|
let request = self.request_write();
|
|
self.follow_up(request)
|
|
}
|
|
}
|
|
|
|
fn closed_write(&mut self, data: ControlResponse) -> Outcome {
|
|
let (expected_position, expected_checksum) = match self.transfer.as_ref() {
|
|
Some(Transfer::Write(write)) => (write.bytes.len() as u32, write.checksum),
|
|
_ => return self.rediscovery_required("completed save has no local artifact"),
|
|
};
|
|
let Some((result, _)) = KcfgResultDetail::decode(data.content.as_bytes()) else {
|
|
return self.rediscovery_required(
|
|
"controller returned a malformed configuration import result",
|
|
);
|
|
};
|
|
if data.status != FileStatus::Ok
|
|
|| data.position != expected_position
|
|
|| data.file_size != expected_position
|
|
|| data.checksum_crc32c != expected_checksum
|
|
|| result.primary().is_some()
|
|
|| result.related_field().is_some()
|
|
|| result.rule().is_some()
|
|
{
|
|
return self.rediscovery_required(
|
|
"controller returned an invalid configuration import result",
|
|
);
|
|
}
|
|
let restart_required = result.restart_required();
|
|
let Some(Transfer::Write(write)) = self.transfer.take() else {
|
|
return self.rediscovery_required("completed save has no local artifact");
|
|
};
|
|
let document = match encode_document_state(&write.bytes, restart_required) {
|
|
Ok(document) => document,
|
|
Err(error) => return self.rediscovery_required(error),
|
|
};
|
|
let config = match decode_document(&document) {
|
|
Ok(config) => config,
|
|
Err(error) => return self.rediscovery_required(error),
|
|
};
|
|
self.downloaded = Some(DownloadedConfig {
|
|
document,
|
|
config: config.clone(),
|
|
});
|
|
Outcome::Event(ClientEvent::Configuration(config))
|
|
}
|
|
|
|
fn request_read(&mut self, position: u32) -> Result<MavlinkFrame, String> {
|
|
let id = match self.transfer.as_ref() {
|
|
Some(Transfer::Read(read)) => read.id,
|
|
_ => return Err("configuration read transfer is not open".into()),
|
|
};
|
|
let target = self.target()?;
|
|
Ok(self.request_control(
|
|
PendingKind::Read { position },
|
|
ControlRequest {
|
|
operation: FileOperation::Read,
|
|
target,
|
|
position,
|
|
file_size: 0,
|
|
checksum_crc32c: 0,
|
|
transfer_id: id,
|
|
content: Content::EMPTY,
|
|
},
|
|
))
|
|
}
|
|
|
|
fn request_write(&mut self) -> Result<MavlinkFrame, String> {
|
|
let (id, position, data, next) = match self.transfer.as_ref() {
|
|
Some(Transfer::Write(write)) => {
|
|
let position = usize::try_from(write.position)
|
|
.map_err(|_| "configuration write position is invalid")?;
|
|
let end = (position + CHUNK_BYTES).min(write.bytes.len());
|
|
if position >= end {
|
|
return Err("configuration write has no remaining bytes".into());
|
|
}
|
|
(
|
|
write.id,
|
|
write.position,
|
|
&write.bytes[position..end],
|
|
end as u32,
|
|
)
|
|
}
|
|
_ => return Err("configuration write transfer is not open".into()),
|
|
};
|
|
let target = self.target()?;
|
|
let mut bytes = [0_u8; CHUNK_BYTES];
|
|
bytes[..data.len()].copy_from_slice(data);
|
|
let message = encode_request(FileRequest::Data(DataRequest {
|
|
target,
|
|
position,
|
|
transfer_id: id,
|
|
data: bytes,
|
|
}));
|
|
Ok(self.send(PendingKind::Write { position, next }, message))
|
|
}
|
|
|
|
fn request_close(
|
|
&mut self,
|
|
kind: PendingKind,
|
|
transfer_id: u16,
|
|
position: u32,
|
|
) -> Result<MavlinkFrame, String> {
|
|
let target = self.target()?;
|
|
Ok(self.request_control(
|
|
kind,
|
|
ControlRequest {
|
|
operation: FileOperation::Close,
|
|
target,
|
|
position,
|
|
file_size: 0,
|
|
checksum_crc32c: 0,
|
|
transfer_id,
|
|
content: Content::EMPTY,
|
|
},
|
|
))
|
|
}
|
|
|
|
fn request_control(&mut self, kind: PendingKind, request: ControlRequest) -> MavlinkFrame {
|
|
let message = encode_request(FileRequest::Control(request));
|
|
self.send(kind, message)
|
|
}
|
|
|
|
fn abort_after(&mut self, detail: String) -> Outcome {
|
|
let transfer_id = match self.transfer.as_ref() {
|
|
Some(Transfer::Read(read)) => read.id,
|
|
Some(Transfer::Write(write)) => write.id,
|
|
None => 0,
|
|
};
|
|
self.abort_transfer_after(transfer_id, detail)
|
|
}
|
|
|
|
fn abort_transfer_after(&mut self, transfer_id: u16, detail: String) -> Outcome {
|
|
if transfer_id == 0 {
|
|
return self.fail(detail);
|
|
}
|
|
self.abort_failure = Some(detail);
|
|
let target = match self.target() {
|
|
Ok(target) => target,
|
|
Err(error) => return self.rediscovery_required(error),
|
|
};
|
|
let request = self.request_control(
|
|
PendingKind::Abort { transfer_id },
|
|
ControlRequest {
|
|
operation: FileOperation::Abort,
|
|
target,
|
|
position: 0,
|
|
file_size: 0,
|
|
checksum_crc32c: 0,
|
|
transfer_id,
|
|
content: Content::EMPTY,
|
|
},
|
|
);
|
|
Outcome::Send(request)
|
|
}
|
|
|
|
pub(crate) fn target(&self) -> Result<Target, String> {
|
|
self.target
|
|
.ok_or_else(|| "configuration target is unknown".into())
|
|
}
|
|
|
|
fn response_identity_matches(&self, kind: PendingKind, transfer_id: u16) -> bool {
|
|
let expected = match kind {
|
|
PendingKind::OpenRead {
|
|
transfer_id: expected,
|
|
} => Some(expected),
|
|
PendingKind::OpenWrite | PendingKind::Write { .. } | PendingKind::CloseWrite { .. } => {
|
|
match self.transfer.as_ref() {
|
|
Some(Transfer::Write(write)) => Some(write.id),
|
|
_ => None,
|
|
}
|
|
}
|
|
PendingKind::Read { .. } | PendingKind::CloseRead { .. } => {
|
|
match self.transfer.as_ref() {
|
|
Some(Transfer::Read(read)) => Some(read.id),
|
|
_ => None,
|
|
}
|
|
}
|
|
PendingKind::Abort {
|
|
transfer_id: expected,
|
|
} => Some(expected),
|
|
};
|
|
expected == Some(transfer_id)
|
|
}
|
|
|
|
fn transfer_id(&mut self) -> u16 {
|
|
let id = self.next_transfer_id;
|
|
self.next_transfer_id = self.next_transfer_id.wrapping_add(1).max(1);
|
|
id
|
|
}
|
|
|
|
fn send(&mut self, kind: PendingKind, message: MavMessage) -> MavlinkFrame {
|
|
let raw = match message {
|
|
MavMessage::FILE_CONTROL(data) => self.encoder.encode(&data),
|
|
MavMessage::FILE_DATA(data) => self.encoder.encode(&data),
|
|
_ => unreachable!("file request encoder returned another message"),
|
|
};
|
|
self.pending = Some(Pending {
|
|
kind,
|
|
raw,
|
|
sent_at: Instant::now(),
|
|
attempts_left: REQUEST_ATTEMPTS - 1,
|
|
});
|
|
raw
|
|
}
|
|
|
|
fn fail(&mut self, detail: impl Into<String>) -> Outcome {
|
|
self.pending = None;
|
|
self.transfer = None;
|
|
self.abort_failure = None;
|
|
self.open_conflict_recovery = OpenConflictRecovery::Disabled;
|
|
Outcome::Event(ClientEvent::Failure(detail.into()))
|
|
}
|
|
|
|
fn durable_state_unknown(&mut self, detail: impl Into<String>) -> Outcome {
|
|
let detail = detail.into();
|
|
self.downloaded = None;
|
|
self.transfer = None;
|
|
self.pending = None;
|
|
self.abort_failure = None;
|
|
self.open_conflict_recovery = OpenConflictRecovery::Disabled;
|
|
Outcome::Event(ClientEvent::DurableStateUnknown(detail))
|
|
}
|
|
|
|
const fn local_target(&self) -> Target {
|
|
Target::new(self.local.system(), self.local.component())
|
|
}
|
|
|
|
fn follow_up(&mut self, request: Result<MavlinkFrame, String>) -> Outcome {
|
|
match request {
|
|
Ok(request) => Outcome::Send(request),
|
|
Err(error) => self.fail(error),
|
|
}
|
|
}
|
|
|
|
fn rediscovery_required(&mut self, detail: impl Into<String>) -> Outcome {
|
|
let detail = detail.into();
|
|
self.reset();
|
|
Outcome::Event(ClientEvent::RediscoveryRequired(detail))
|
|
}
|
|
}
|
|
|
|
fn validate_edits(config: &RemoteConfig, edits: &[Edit]) -> Result<(), String> {
|
|
for edit in edits {
|
|
let Some(field) = config.fields.iter().find(|field| field.id == edit.field_id) else {
|
|
return Err(format!("unknown field id {}", edit.field_id.get()));
|
|
};
|
|
if edit.value.kind() != field.kind {
|
|
return Err(format!(
|
|
"field {} has the wrong value type",
|
|
edit.field_id.get()
|
|
));
|
|
}
|
|
if !field.options.is_empty()
|
|
&& !matches!(
|
|
edit.value,
|
|
ConfigValue::U8(value)
|
|
if field.options.iter().any(|option| option.value == value)
|
|
)
|
|
{
|
|
return Err(format!(
|
|
"field {} has an unnamed enum value",
|
|
edit.field_id.get()
|
|
));
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
fn timeout(kind: PendingKind) -> Duration {
|
|
if matches!(kind, PendingKind::CloseWrite { .. }) {
|
|
CLOSE_WRITE_TIMEOUT
|
|
} else {
|
|
REQUEST_TIMEOUT
|
|
}
|
|
}
|
|
|
|
fn operation_name(kind: PendingKind) -> &'static str {
|
|
match kind {
|
|
PendingKind::OpenRead { .. } => "open-read",
|
|
PendingKind::Read { .. } => "read",
|
|
PendingKind::CloseRead { .. } => "close-read",
|
|
PendingKind::OpenWrite => "open-write",
|
|
PendingKind::Write { .. } => "write",
|
|
PendingKind::CloseWrite { .. } => "close-write",
|
|
PendingKind::Abort { .. } => "abort",
|
|
}
|
|
}
|
|
|
|
fn control_matches(kind: PendingKind, data: &ControlResponse) -> bool {
|
|
let successful = data.status == FileStatus::Ok;
|
|
match kind {
|
|
PendingKind::OpenRead { .. } => {
|
|
data.operation == FileOperation::OpenRead && data.position == 0
|
|
}
|
|
PendingKind::OpenWrite => data.operation == FileOperation::OpenWrite && data.position == 0,
|
|
PendingKind::Read { position } => {
|
|
data.operation == FileOperation::Read && !successful && data.position == position
|
|
}
|
|
PendingKind::Write { position, next } => {
|
|
data.operation == FileOperation::WriteAck
|
|
&& data.position == if successful { next } else { position }
|
|
}
|
|
PendingKind::CloseRead { position } | PendingKind::CloseWrite { position } => {
|
|
data.operation == FileOperation::Close && data.position == position
|
|
}
|
|
PendingKind::Abort { .. } => data.operation == FileOperation::Abort && data.position == 0,
|
|
}
|
|
}
|
|
|
|
fn rejected_detail(data: &ControlResponse) -> String {
|
|
if let Some((detail, text)) = KcfgResultDetail::decode(data.content.as_bytes()) {
|
|
let mut identities = Vec::new();
|
|
if let Some(field) = detail.primary() {
|
|
identities.push(format!("field {}", field.get()));
|
|
}
|
|
if let Some(field) = detail.related_field() {
|
|
identities.push(format!("related field {}", field.get()));
|
|
}
|
|
if let Some(rule) = detail.rule() {
|
|
identities.push(format!("rule {} ({rule:?})", rule.code()));
|
|
}
|
|
let suffix = if identities.is_empty() {
|
|
String::new()
|
|
} else {
|
|
format!(" [{}]", identities.join(", "))
|
|
};
|
|
return if text.is_empty() {
|
|
format!("configuration file rejected: {:?}{suffix}", data.status)
|
|
} else {
|
|
format!(
|
|
"configuration file rejected: {:?}: {text}{suffix}",
|
|
data.status
|
|
)
|
|
};
|
|
}
|
|
let context = std::str::from_utf8(data.content.as_bytes())
|
|
.ok()
|
|
.filter(|text| !text.is_empty());
|
|
match context {
|
|
Some(context) => format!("configuration file rejected: {:?}: {context}", data.status),
|
|
None => format!("configuration file rejected: {:?}", data.status),
|
|
}
|
|
}
|
|
|
|
// KCFG owns the schema and value representation. These two adapters are the
|
|
// only place where the host configurator depends on that artifact codec.
|
|
fn decode_document(bytes: &[u8]) -> Result<RemoteConfig, String> {
|
|
kaptan_config::decode_kcfg(bytes)
|
|
.map_err(|error| format!("invalid KCFG artifact: {error:?}"))
|
|
.and_then(remote_config)
|
|
}
|
|
|
|
fn remote_config(artifact: kaptan_config::KcfgArtifact<'_>) -> Result<RemoteConfig, String> {
|
|
let mut fields = Vec::with_capacity(usize::from(artifact.field_count()));
|
|
for field in artifact.fields() {
|
|
let mut options = Vec::with_capacity(field.enum_options().len());
|
|
for option in field.enum_options() {
|
|
options.push(RemoteEnumOption {
|
|
value: option.value(),
|
|
label: option.label().into(),
|
|
});
|
|
}
|
|
let field = RemoteField {
|
|
id: field.id(),
|
|
key: field.key().into(),
|
|
unit: field.unit().into(),
|
|
kind: field.kind(),
|
|
minimum: field.minimum(),
|
|
maximum: field.maximum(),
|
|
factory: field.factory_value(),
|
|
value: field.value(),
|
|
options,
|
|
};
|
|
if fields
|
|
.iter()
|
|
.any(|previous: &RemoteField| previous.key == field.key)
|
|
{
|
|
return Err("KCFG artifact contains a duplicate display key".into());
|
|
}
|
|
validate_remote_field(&field)?;
|
|
fields.push(field);
|
|
}
|
|
Ok(RemoteConfig {
|
|
schema_id: artifact.schema_id(),
|
|
restart_required: artifact.restart_required(),
|
|
fields,
|
|
})
|
|
}
|
|
|
|
fn validate_remote_field(field: &RemoteField) -> Result<(), String> {
|
|
let values = [
|
|
Some(field.value),
|
|
Some(field.factory),
|
|
field.minimum,
|
|
field.maximum,
|
|
];
|
|
if values.into_iter().flatten().any(non_finite) {
|
|
return Err(format!(
|
|
"KCFG field {} contains a non-finite value",
|
|
field.id.get()
|
|
));
|
|
}
|
|
if field.minimum.is_some_and(|minimum| {
|
|
field
|
|
.maximum
|
|
.is_some_and(|maximum| scalar_less_than(maximum, minimum))
|
|
}) || field
|
|
.minimum
|
|
.is_some_and(|minimum| scalar_less_than(field.value, minimum))
|
|
|| field
|
|
.maximum
|
|
.is_some_and(|maximum| scalar_less_than(maximum, field.value))
|
|
|| field
|
|
.minimum
|
|
.is_some_and(|minimum| scalar_less_than(field.factory, minimum))
|
|
|| field
|
|
.maximum
|
|
.is_some_and(|maximum| scalar_less_than(maximum, field.factory))
|
|
{
|
|
return Err(format!(
|
|
"KCFG field {} has inconsistent bounds",
|
|
field.id.get()
|
|
));
|
|
}
|
|
if !field.options.is_empty()
|
|
&& ![field.value, field.factory].into_iter().all(|value| {
|
|
matches!(
|
|
value,
|
|
ConfigValue::U8(value)
|
|
if field.options.iter().any(|option| option.value == value)
|
|
)
|
|
})
|
|
{
|
|
return Err(format!(
|
|
"KCFG field {} contains an unnamed enum value",
|
|
field.id.get()
|
|
));
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
const fn non_finite(value: ConfigValue) -> bool {
|
|
match value {
|
|
ConfigValue::F32(value) => !value.is_finite(),
|
|
ConfigValue::F64(value) => !value.is_finite(),
|
|
ConfigValue::Bool(_)
|
|
| ConfigValue::U8(_)
|
|
| ConfigValue::U16(_)
|
|
| ConfigValue::I32(_)
|
|
| ConfigValue::U32(_)
|
|
| ConfigValue::U64(_) => false,
|
|
}
|
|
}
|
|
|
|
fn encode_edited_document(document: &[u8], edits: &[Edit]) -> Result<Vec<u8>, String> {
|
|
let artifact = kaptan_config::decode_kcfg(document)
|
|
.map_err(|error| format!("cannot edit KCFG artifact: {error:?}"))?;
|
|
let mut output = vec![0_u8; document.len()];
|
|
let encoded = kaptan_config::encode_edited_kcfg(&artifact, true, &mut output, |field| {
|
|
edits
|
|
.iter()
|
|
.find(|edit| edit.field_id == field.id())
|
|
.map_or_else(|| field.value(), |edit| edit.value)
|
|
})
|
|
.map_err(|error| format!("cannot encode KCFG artifact: {error:?}"))?;
|
|
let length = encoded.len();
|
|
output.truncate(length);
|
|
Ok(output)
|
|
}
|
|
|
|
fn encode_document_state(document: &[u8], restart_required: bool) -> Result<Vec<u8>, String> {
|
|
let artifact = kaptan_config::decode_kcfg(document)
|
|
.map_err(|error| format!("cannot update KCFG artifact state: {error:?}"))?;
|
|
let mut output = vec![0_u8; document.len()];
|
|
let encoded =
|
|
kaptan_config::encode_edited_kcfg(&artifact, restart_required, &mut output, |field| {
|
|
field.value()
|
|
})
|
|
.map_err(|error| format!("cannot encode KCFG artifact state: {error:?}"))?;
|
|
let length = encoded.len();
|
|
output.truncate(length);
|
|
Ok(output)
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use kaptan_file_mavlink::{
|
|
ControlResponse, DataResponse, FileRequest, FileResponse, FileStatus,
|
|
wire::{decode_request, encode_response},
|
|
};
|
|
|
|
const CONTROLLER: Target = Target::new(42, 7);
|
|
const CLIENT: Principal = Principal::new(255, 190);
|
|
const FIELD_ID: FieldId = FieldId::new(401);
|
|
|
|
fn artifact(current: u8) -> Vec<u8> {
|
|
let key = b"telemetry.heartbeat_hz";
|
|
let unit = b"hz";
|
|
let mut bytes = Vec::new();
|
|
bytes.extend_from_slice(b"KCFG");
|
|
bytes.push(1); // KCFG version 1.
|
|
bytes.push(0);
|
|
bytes.extend_from_slice(b"UNKNOWN-SCHEMA01");
|
|
bytes.extend_from_slice(&1_u16.to_le_bytes());
|
|
bytes.extend_from_slice(&FIELD_ID.get().to_le_bytes());
|
|
bytes.push(ValueKind::U8 as u8);
|
|
bytes.push(3);
|
|
bytes.push(key.len() as u8);
|
|
bytes.push(unit.len() as u8);
|
|
bytes.push(0);
|
|
bytes.extend_from_slice(&[current, 2, 1, 100]);
|
|
bytes.extend_from_slice(key);
|
|
bytes.extend_from_slice(unit);
|
|
let checksum = crc32c(&bytes);
|
|
bytes.extend_from_slice(&checksum.to_be_bytes());
|
|
bytes
|
|
}
|
|
|
|
fn response(response: FileResponse) -> MavlinkFrame {
|
|
let mut encoder = MavlinkEncoder::new(CONTROLLER.system(), CONTROLLER.component());
|
|
match encode_response(response) {
|
|
MavMessage::FILE_CONTROL(data) => encoder.encode(&data),
|
|
MavMessage::FILE_DATA(data) => encoder.encode(&data),
|
|
_ => unreachable!("file response encoder returned another message"),
|
|
}
|
|
}
|
|
|
|
fn request(raw: &MavlinkFrame) -> FileRequest {
|
|
let message = parse_test_message(raw).unwrap();
|
|
decode_request(&message).unwrap()
|
|
}
|
|
|
|
fn sent(outcome: Outcome) -> MavlinkFrame {
|
|
let Outcome::Send(frame) = outcome else {
|
|
panic!("expected a follow-up frame");
|
|
};
|
|
frame
|
|
}
|
|
|
|
fn event(outcome: Outcome) -> ClientEvent {
|
|
let Outcome::Event(event) = outcome else {
|
|
panic!("expected a client event");
|
|
};
|
|
event
|
|
}
|
|
|
|
fn control_response(
|
|
operation: FileOperation,
|
|
transfer_id: u16,
|
|
position: u32,
|
|
file_size: u32,
|
|
checksum: u32,
|
|
status: FileStatus,
|
|
) -> MavlinkFrame {
|
|
control_response_with_content(
|
|
operation,
|
|
transfer_id,
|
|
position,
|
|
file_size,
|
|
checksum,
|
|
status,
|
|
Content::EMPTY,
|
|
)
|
|
}
|
|
|
|
fn control_response_with_content(
|
|
operation: FileOperation,
|
|
transfer_id: u16,
|
|
position: u32,
|
|
file_size: u32,
|
|
checksum: u32,
|
|
status: FileStatus,
|
|
content: Content,
|
|
) -> MavlinkFrame {
|
|
response(FileResponse::Control(ControlResponse {
|
|
operation,
|
|
target: Target::new(CLIENT.system(), CLIENT.component()),
|
|
position,
|
|
file_size,
|
|
checksum_crc32c: checksum,
|
|
transfer_id,
|
|
status,
|
|
content,
|
|
}))
|
|
}
|
|
|
|
fn discover(client: &mut ConfigClient, document: &[u8]) -> RemoteConfig {
|
|
let open = client.discover(CONTROLLER).unwrap();
|
|
let FileRequest::Control(open) = request(&open) else {
|
|
panic!("open-read must be a control request");
|
|
};
|
|
assert_eq!(open.operation, FileOperation::OpenRead);
|
|
assert_eq!(open.content.as_bytes(), CONFIG_PATH);
|
|
assert_ne!(open.transfer_id, 0);
|
|
|
|
let size = document.len() as u32;
|
|
let checksum = crc32c(document);
|
|
let read = sent(client.observe_frame(&control_response(
|
|
FileOperation::OpenRead,
|
|
open.transfer_id,
|
|
0,
|
|
size,
|
|
checksum,
|
|
FileStatus::Ok,
|
|
)));
|
|
let FileRequest::Control(read) = request(&read) else {
|
|
panic!("read must be a control request");
|
|
};
|
|
assert_eq!(read.operation, FileOperation::Read);
|
|
assert_eq!(read.position, 0);
|
|
|
|
let mut data = [0_u8; CHUNK_BYTES];
|
|
data[..document.len()].copy_from_slice(document);
|
|
let close = sent(
|
|
client.observe_frame(&response(FileResponse::Data(DataResponse {
|
|
target: Target::new(CLIENT.system(), CLIENT.component()),
|
|
position: 0,
|
|
transfer_id: open.transfer_id,
|
|
data,
|
|
}))),
|
|
);
|
|
let FileRequest::Control(close) = request(&close) else {
|
|
panic!("close-read must be a control request");
|
|
};
|
|
assert_eq!(close.operation, FileOperation::Close);
|
|
assert_eq!(close.position, size);
|
|
|
|
let event = event(client.observe_frame(&control_response(
|
|
FileOperation::Close,
|
|
open.transfer_id,
|
|
size,
|
|
size,
|
|
checksum,
|
|
FileStatus::Ok,
|
|
)));
|
|
let ClientEvent::Configuration(config) = event else {
|
|
panic!("closed export must publish configuration");
|
|
};
|
|
config
|
|
}
|
|
|
|
#[test]
|
|
fn transfer_crc_is_crc32c() {
|
|
assert_eq!(crc32c(b"123456789"), 0xe306_9283);
|
|
}
|
|
|
|
#[test]
|
|
fn configurator_knows_only_the_canonical_config_path() {
|
|
assert_eq!(CONFIG_PATH, b"/config/user.kcfg");
|
|
}
|
|
|
|
#[test]
|
|
fn file_data_uses_the_full_declared_chunk() {
|
|
assert_eq!(CHUNK_BYTES, 246);
|
|
}
|
|
|
|
#[test]
|
|
fn exact_local_endpoints_isolate_clients_with_the_same_transfer_id() {
|
|
let left_endpoint = Principal::new(250, 190);
|
|
let right_endpoint = Principal::new(251, 190);
|
|
let mut left = ConfigClient::new(left_endpoint);
|
|
let mut right = ConfigClient::new(right_endpoint);
|
|
let left_open = left.discover(CONTROLLER).unwrap();
|
|
let right_open = right.discover(CONTROLLER).unwrap();
|
|
assert_eq!(left_open.system_id(), left_endpoint.system());
|
|
assert_eq!(right_open.system_id(), right_endpoint.system());
|
|
let FileRequest::Control(left_request) = request(&left_open) else {
|
|
panic!("left discovery must open a file");
|
|
};
|
|
let FileRequest::Control(right_request) = request(&right_open) else {
|
|
panic!("right discovery must open a file");
|
|
};
|
|
assert_eq!(left_request.transfer_id, right_request.transfer_id);
|
|
|
|
let document = artifact(20);
|
|
let reply = response(FileResponse::Control(ControlResponse {
|
|
operation: FileOperation::OpenRead,
|
|
target: Target::new(left_endpoint.system(), left_endpoint.component()),
|
|
position: 0,
|
|
file_size: document.len() as u32,
|
|
checksum_crc32c: crc32c(&document),
|
|
transfer_id: left_request.transfer_id,
|
|
status: FileStatus::Ok,
|
|
content: Content::EMPTY,
|
|
}));
|
|
assert!(matches!(left.observe_frame(&reply), Outcome::Send(_)));
|
|
assert!(matches!(right.observe_frame(&reply), Outcome::Idle));
|
|
assert!(matches!(
|
|
right.pending.as_ref().map(|pending| pending.kind),
|
|
Some(PendingKind::OpenRead { .. })
|
|
));
|
|
}
|
|
|
|
#[test]
|
|
fn discovery_rejection_resets_for_a_later_heartbeat() {
|
|
let mut client = ConfigClient::new(CLIENT);
|
|
let open = client.discover(CONTROLLER).unwrap();
|
|
let FileRequest::Control(open) = request(&open) else {
|
|
panic!("discovery must open a file");
|
|
};
|
|
let event = event(client.observe_frame(&control_response(
|
|
FileOperation::OpenRead,
|
|
open.transfer_id,
|
|
0,
|
|
0,
|
|
0,
|
|
FileStatus::Busy,
|
|
)));
|
|
assert!(matches!(event, ClientEvent::RediscoveryRequired(_)));
|
|
assert_eq!(client.target, None);
|
|
assert!(client.pending.is_none());
|
|
}
|
|
|
|
#[test]
|
|
fn stateful_rejection_and_noncanonical_data_emit_abort() {
|
|
let document = artifact(20);
|
|
for noncanonical_data in [false, true] {
|
|
let mut client = ConfigClient::new(CLIENT);
|
|
let open = client.discover(CONTROLLER).unwrap();
|
|
let FileRequest::Control(open) = request(&open) else {
|
|
panic!("discovery must open a file");
|
|
};
|
|
let read = sent(client.observe_frame(&control_response(
|
|
FileOperation::OpenRead,
|
|
open.transfer_id,
|
|
0,
|
|
document.len() as u32,
|
|
crc32c(&document),
|
|
FileStatus::Ok,
|
|
)));
|
|
assert!(matches!(request(&read), FileRequest::Control(_)));
|
|
|
|
let abort = sent(if noncanonical_data {
|
|
let mut data = [0_u8; CHUNK_BYTES];
|
|
data[..document.len()].copy_from_slice(&document);
|
|
data[document.len()] = 1;
|
|
client.observe_frame(&response(FileResponse::Data(DataResponse {
|
|
target: Target::new(CLIENT.system(), CLIENT.component()),
|
|
position: 0,
|
|
transfer_id: open.transfer_id,
|
|
data,
|
|
})))
|
|
} else {
|
|
client.observe_frame(&control_response(
|
|
FileOperation::Read,
|
|
open.transfer_id,
|
|
0,
|
|
document.len() as u32,
|
|
crc32c(&document),
|
|
FileStatus::InvalidPosition,
|
|
))
|
|
});
|
|
let FileRequest::Control(abort) = request(&abort) else {
|
|
panic!("rejected active transfer must abort");
|
|
};
|
|
assert_eq!(abort.operation, FileOperation::Abort);
|
|
assert_eq!(abort.transfer_id, open.transfer_id);
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn delayed_write_ack_does_not_consume_the_next_pending_chunk() {
|
|
let response = |position, status| ControlResponse {
|
|
operation: FileOperation::WriteAck,
|
|
target: Target::new(CLIENT.system(), CLIENT.component()),
|
|
position,
|
|
file_size: 600,
|
|
checksum_crc32c: 0,
|
|
transfer_id: 9,
|
|
status,
|
|
content: Content::EMPTY,
|
|
};
|
|
let pending = PendingKind::Write {
|
|
position: CHUNK_BYTES as u32,
|
|
next: 2 * CHUNK_BYTES as u32,
|
|
};
|
|
|
|
assert!(!control_matches(
|
|
pending,
|
|
&response(CHUNK_BYTES as u32, FileStatus::Ok)
|
|
));
|
|
assert!(control_matches(
|
|
pending,
|
|
&response(2 * CHUNK_BYTES as u32, FileStatus::Ok)
|
|
));
|
|
assert!(control_matches(
|
|
pending,
|
|
&response(CHUNK_BYTES as u32, FileStatus::InvalidPosition)
|
|
));
|
|
}
|
|
|
|
#[test]
|
|
fn one_file_export_discovers_an_unknown_schema_without_a_catalog() {
|
|
let mut client = ConfigClient::new(CLIENT);
|
|
let config = discover(&mut client, &artifact(20));
|
|
|
|
assert_eq!(config.schema_id, *b"UNKNOWN-SCHEMA01");
|
|
assert!(!config.restart_required);
|
|
assert_eq!(config.fields.len(), 1);
|
|
assert_eq!(config.fields[0].id, FIELD_ID);
|
|
assert_eq!(config.fields[0].key, "telemetry.heartbeat_hz");
|
|
assert_eq!(config.fields[0].value, ConfigValue::U8(20));
|
|
assert_eq!(config.fields[0].factory, ConfigValue::U8(2));
|
|
assert_eq!(config.fields[0].minimum, Some(ConfigValue::U8(1)));
|
|
assert_eq!(config.fields[0].maximum, Some(ConfigValue::U8(100)));
|
|
}
|
|
|
|
#[test]
|
|
fn reconnect_retries_once_then_waits_quietly_past_the_replay_lease() {
|
|
let mut client = ConfigClient::new(CLIENT);
|
|
let first = client.discover(CONTROLLER).unwrap();
|
|
let FileRequest::Control(first) = request(&first) else {
|
|
panic!("discovery must open the configuration file");
|
|
};
|
|
|
|
let retry = sent(client.observe_frame(&control_response(
|
|
FileOperation::OpenRead,
|
|
first.transfer_id,
|
|
0,
|
|
0,
|
|
0,
|
|
FileStatus::RequestConflict,
|
|
)));
|
|
let FileRequest::Control(retry) = request(&retry) else {
|
|
panic!("conflict recovery must reopen the configuration file");
|
|
};
|
|
assert_eq!(retry.operation, FileOperation::OpenRead);
|
|
assert_ne!(retry.transfer_id, first.transfer_id);
|
|
|
|
assert!(matches!(
|
|
client.observe_frame(&control_response(
|
|
FileOperation::OpenRead,
|
|
retry.transfer_id,
|
|
0,
|
|
0,
|
|
0,
|
|
FileStatus::RequestConflict,
|
|
)),
|
|
Outcome::Idle
|
|
));
|
|
|
|
let OpenConflictRecovery::Waiting(retry_at) = client.open_conflict_recovery else {
|
|
panic!("the second conflict must wait for the controller lease");
|
|
};
|
|
assert!(
|
|
retry_at.duration_since(Instant::now())
|
|
> Duration::from_secs(u64::from(TRANSFER_REPLAY_LEASE_SECONDS))
|
|
);
|
|
let next_transfer_id = client.next_transfer_id;
|
|
for _ in 0..3 {
|
|
assert!(matches!(client.retry_due(), Outcome::Idle));
|
|
assert_eq!(client.next_transfer_id, next_transfer_id);
|
|
}
|
|
|
|
client.open_conflict_recovery =
|
|
OpenConflictRecovery::Waiting(Instant::now() - REQUEST_TIMEOUT);
|
|
let delayed = sent(client.retry_due());
|
|
let FileRequest::Control(delayed) = request(&delayed) else {
|
|
panic!("lease expiry must reopen the configuration file");
|
|
};
|
|
assert_eq!(delayed.operation, FileOperation::OpenRead);
|
|
assert_ne!(delayed.transfer_id, first.transfer_id);
|
|
assert_ne!(delayed.transfer_id, retry.transfer_id);
|
|
|
|
let document = artifact(20);
|
|
let read = sent(client.observe_frame(&control_response(
|
|
FileOperation::OpenRead,
|
|
delayed.transfer_id,
|
|
0,
|
|
document.len() as u32,
|
|
crc32c(&document),
|
|
FileStatus::Ok,
|
|
)));
|
|
let FileRequest::Control(read) = request(&read) else {
|
|
panic!("the reopened file must advance to its first read");
|
|
};
|
|
assert_eq!(read.operation, FileOperation::Read);
|
|
assert_eq!(read.transfer_id, delayed.transfer_id);
|
|
}
|
|
|
|
#[test]
|
|
fn save_uploads_one_complete_edited_artifact_and_closes_it_once() {
|
|
let mut client = ConfigClient::new(CLIENT);
|
|
let original = artifact(20);
|
|
discover(&mut client, &original);
|
|
|
|
let open = client
|
|
.save(vec![Edit {
|
|
field_id: FIELD_ID,
|
|
value: ConfigValue::U8(30),
|
|
}])
|
|
.unwrap();
|
|
let FileRequest::Control(open) = request(&open) else {
|
|
panic!("open-write must be a control request");
|
|
};
|
|
assert_eq!(open.operation, FileOperation::OpenWrite);
|
|
assert_eq!(open.content.as_bytes(), CONFIG_PATH);
|
|
assert_eq!(open.file_size as usize, original.len());
|
|
|
|
let data = sent(client.observe_frame(&control_response(
|
|
FileOperation::OpenWrite,
|
|
open.transfer_id,
|
|
0,
|
|
open.file_size,
|
|
open.checksum_crc32c,
|
|
FileStatus::Ok,
|
|
)));
|
|
let FileRequest::Data(data) = request(&data) else {
|
|
panic!("save body must be file data");
|
|
};
|
|
let written = &data.data[..open.file_size as usize];
|
|
assert_eq!(crc32c(written), open.checksum_crc32c);
|
|
let edited = kaptan_config::decode_kcfg(written).unwrap();
|
|
assert!(edited.restart_required());
|
|
assert_eq!(edited.fields().next().unwrap().value(), ConfigValue::U8(30));
|
|
|
|
let close = sent(client.observe_frame(&control_response(
|
|
FileOperation::WriteAck,
|
|
open.transfer_id,
|
|
open.file_size,
|
|
open.file_size,
|
|
open.checksum_crc32c,
|
|
FileStatus::Ok,
|
|
)));
|
|
let FileRequest::Control(close) = request(&close) else {
|
|
panic!("complete upload must close");
|
|
};
|
|
assert_eq!(close.operation, FileOperation::Close);
|
|
assert_eq!(close.position, open.file_size);
|
|
|
|
let event = event(client.observe_frame(&control_response_with_content(
|
|
FileOperation::Close,
|
|
open.transfer_id,
|
|
open.file_size,
|
|
open.file_size,
|
|
open.checksum_crc32c,
|
|
FileStatus::Ok,
|
|
Content::try_new(&KcfgResultDetail::success(true).encode()).unwrap(),
|
|
)));
|
|
let ClientEvent::Configuration(config) = event else {
|
|
panic!("completed import must publish configuration");
|
|
};
|
|
assert!(config.restart_required);
|
|
assert_eq!(config.fields[0].value, ConfigValue::U8(30));
|
|
}
|
|
|
|
#[test]
|
|
fn save_can_restore_the_boot_active_values_without_forcing_restart_state() {
|
|
let document = encode_edited_document(
|
|
&artifact(20),
|
|
&[Edit {
|
|
field_id: FIELD_ID,
|
|
value: ConfigValue::U8(2),
|
|
}],
|
|
)
|
|
.unwrap();
|
|
let mut client = ConfigClient::new(CLIENT);
|
|
client.transfer = Some(Transfer::Write(WriteTransfer {
|
|
id: 9,
|
|
position: document.len() as u32,
|
|
bytes: document.clone(),
|
|
checksum: crc32c(&document),
|
|
}));
|
|
|
|
let event = event(client.closed_write(ControlResponse {
|
|
operation: FileOperation::Close,
|
|
target: Target::new(CLIENT.system(), CLIENT.component()),
|
|
position: document.len() as u32,
|
|
file_size: document.len() as u32,
|
|
checksum_crc32c: crc32c(&document),
|
|
transfer_id: 9,
|
|
status: FileStatus::Ok,
|
|
content: Content::try_new(&KcfgResultDetail::success(false).encode()).unwrap(),
|
|
}));
|
|
|
|
let ClientEvent::Configuration(config) = event else {
|
|
panic!("restored values must publish configuration");
|
|
};
|
|
assert!(!config.restart_required);
|
|
assert!(
|
|
!kaptan_config::decode_kcfg(&client.downloaded.as_ref().unwrap().document)
|
|
.unwrap()
|
|
.restart_required()
|
|
);
|
|
}
|
|
|
|
fn client_awaiting_close() -> (ConfigClient, Vec<u8>) {
|
|
let document = encode_edited_document(
|
|
&artifact(20),
|
|
&[Edit {
|
|
field_id: FIELD_ID,
|
|
value: ConfigValue::U8(30),
|
|
}],
|
|
)
|
|
.unwrap();
|
|
let mut client = ConfigClient::new(CLIENT);
|
|
client.target = Some(CONTROLLER);
|
|
client.transfer = Some(Transfer::Write(WriteTransfer {
|
|
id: 9,
|
|
position: document.len() as u32,
|
|
checksum: crc32c(&document),
|
|
bytes: document.clone(),
|
|
}));
|
|
client
|
|
.request_close(
|
|
PendingKind::CloseWrite {
|
|
position: document.len() as u32,
|
|
},
|
|
9,
|
|
document.len() as u32,
|
|
)
|
|
.unwrap();
|
|
(client, document)
|
|
}
|
|
|
|
#[test]
|
|
fn malformed_success_and_storage_uncertainty_never_publish_configuration() {
|
|
let (mut malformed, document) = client_awaiting_close();
|
|
let malformed_event = event(malformed.observe_frame(&control_response(
|
|
FileOperation::Close,
|
|
9,
|
|
document.len() as u32,
|
|
document.len() as u32,
|
|
crc32c(&document),
|
|
FileStatus::Ok,
|
|
)));
|
|
assert!(matches!(
|
|
malformed_event,
|
|
ClientEvent::RediscoveryRequired(_)
|
|
));
|
|
|
|
let (mut uncertain, document) = client_awaiting_close();
|
|
let uncertain_event = event(uncertain.observe_frame(&control_response(
|
|
FileOperation::Close,
|
|
9,
|
|
document.len() as u32,
|
|
document.len() as u32,
|
|
crc32c(&document),
|
|
FileStatus::StorageFailure,
|
|
)));
|
|
assert!(matches!(
|
|
uncertain_event,
|
|
ClientEvent::DurableStateUnknown(_)
|
|
));
|
|
}
|
|
|
|
#[test]
|
|
fn exhausted_close_retries_leave_durable_state_unknown() {
|
|
let (mut client, _) = client_awaiting_close();
|
|
let pending = client.pending.as_mut().unwrap();
|
|
pending.attempts_left = 0;
|
|
pending.sent_at = Instant::now() - CLOSE_WRITE_TIMEOUT;
|
|
let event = event(client.retry_due());
|
|
assert!(matches!(event, ClientEvent::DurableStateUnknown(_)));
|
|
assert!(client.pending.is_none());
|
|
assert!(client.transfer.is_none());
|
|
}
|
|
|
|
#[test]
|
|
fn typed_kcfg_rejection_retains_field_and_rule_identity() {
|
|
let violation = kaptan_config::ConfigViolation::related(
|
|
kaptan_config::FieldId::new(401),
|
|
kaptan_config::FieldId::new(402),
|
|
kaptan_config::ValidationRule::Inconsistent,
|
|
);
|
|
let prefix = KcfgResultDetail::violation(violation).encode();
|
|
let detail = rejected_detail(&ControlResponse {
|
|
operation: FileOperation::Close,
|
|
target: Target::new(CLIENT.system(), CLIENT.component()),
|
|
position: 0,
|
|
file_size: 0,
|
|
checksum_crc32c: 0,
|
|
transfer_id: 9,
|
|
status: FileStatus::ContentRejected,
|
|
content: Content::try_new(&prefix).unwrap(),
|
|
});
|
|
assert!(detail.contains("field 401"));
|
|
assert!(detail.contains("related field 402"));
|
|
assert!(detail.contains("rule 5 (Inconsistent)"));
|
|
}
|
|
}
|