From 2b224f8eb945a50d49d8ff234b9db0c1c917e3c1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Onur=20=C3=96zkan?= Date: Sun, 9 Aug 2026 22:06:42 +0300 Subject: [PATCH] feat(config): add MAVLink transaction service --- Cargo.toml | 1 + libs/kaptan-config-mavlink/Cargo.toml | 20 + libs/kaptan-config-mavlink/src/command.rs | 271 ++++ libs/kaptan-config-mavlink/src/lib.rs | 20 + libs/kaptan-config-mavlink/src/protocol.rs | 194 +++ libs/kaptan-config-mavlink/src/service.rs | 14 + .../src/service/import.rs | 248 ++++ .../src/service/query.rs | 208 ++++ .../src/service/replay.rs | 272 ++++ .../src/service/session.rs | 1094 +++++++++++++++++ libs/kaptan-config-mavlink/src/wire.rs | 501 ++++++++ libs/kaptan-mavlink/build.rs | 3 +- libs/kaptan-mavlink/definitions/kaptan.xml | 1 + .../definitions/kaptan/config.xml | 270 ++++ 14 files changed, 3116 insertions(+), 1 deletion(-) create mode 100644 libs/kaptan-config-mavlink/Cargo.toml create mode 100644 libs/kaptan-config-mavlink/src/command.rs create mode 100644 libs/kaptan-config-mavlink/src/lib.rs create mode 100644 libs/kaptan-config-mavlink/src/protocol.rs create mode 100644 libs/kaptan-config-mavlink/src/service.rs create mode 100644 libs/kaptan-config-mavlink/src/service/import.rs create mode 100644 libs/kaptan-config-mavlink/src/service/query.rs create mode 100644 libs/kaptan-config-mavlink/src/service/replay.rs create mode 100644 libs/kaptan-config-mavlink/src/service/session.rs create mode 100644 libs/kaptan-config-mavlink/src/wire.rs create mode 100644 libs/kaptan-mavlink/definitions/kaptan/config.xml diff --git a/Cargo.toml b/Cargo.toml index 31b7166..717d14c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,6 +7,7 @@ members = [ "libs/kaptan-control", "libs/kaptan-config", "libs/kaptan-config-derive", + "libs/kaptan-config-mavlink", "libs/kaptan-data-channel", "libs/kaptan-dma", "libs/kaptan-drivers", diff --git a/libs/kaptan-config-mavlink/Cargo.toml b/libs/kaptan-config-mavlink/Cargo.toml new file mode 100644 index 0000000..e5c215e --- /dev/null +++ b/libs/kaptan-config-mavlink/Cargo.toml @@ -0,0 +1,20 @@ +[package] +name = "kaptan-config-mavlink" +edition.workspace = true +license.workspace = true +repository.workspace = true +rust-version.workspace = true +version.workspace = true + +[lints] +workspace = true + +[features] +# Explicitly opt-in host/SITL test authority. Production cannot construct an +# authorization token from a CRC-valid raw MAVLink frame. +host-authority = [] + +[dependencies] +kaptan-config = { path = "../kaptan-config" } +kaptan-log = { path = "../kaptan-log" } +kaptan-mavlink = { path = "../kaptan-mavlink", default-features = false, features = ["dialect-kaptan"] } diff --git a/libs/kaptan-config-mavlink/src/command.rs b/libs/kaptan-config-mavlink/src/command.rs new file mode 100644 index 0000000..2874c54 --- /dev/null +++ b/libs/kaptan-config-mavlink/src/command.rs @@ -0,0 +1,271 @@ +//! Authorized, transport-neutral configuration commands. + +use kaptan_config::{ConfigValue, FieldId}; + +/// Exact vehicle endpoint selected by a configuration request. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct Target { + /// MAVLink system identity. + system: u8, + /// MAVLink component identity. + component: u8, +} + +/// Selects the configuration snapshot read by a value request. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum ValueSource { + /// Immutable boot-active snapshot. + Active, + /// Last durably committed candidate. + Committed, + /// The caller's open RAM transaction. + Transaction, +} + +impl Target { + /// Creates an exact system/component target. + pub const fn new(system: u8, component: u8) -> Self { + assert!( + system != 0 && component != 0, + "configuration target must be exact" + ); + Self { system, component } + } + + /// Parses an exact non-broadcast target from an untrusted transport. + pub const fn try_new(system: u8, component: u8) -> Option { + if system == 0 || component == 0 { + None + } else { + Some(Self { system, component }) + } + } +} + +/// One command accepted by the Kaptan configuration microservice. +#[derive(Clone, Copy, Debug, PartialEq)] +pub enum ConfigCommand { + /// Reads protocol/schema metadata before any descriptor or upload request. + InfoRequest { + /// Client correlation identity. + request_id: u32, + /// Exact addressed endpoint. + target: Target, + }, + /// Enumerates a bounded page of schema descriptors. + DescriptorRequest { + /// Client correlation identity. + request_id: u32, + /// First stable descriptor index. + first: u16, + /// Requested number of descriptors, clamped to the protocol bound. + count: u16, + /// Exact addressed endpoint. + target: Target, + }, + /// Reads one typed scalar from an immutable or transaction snapshot. + ValueRequest { + /// Client correlation identity. + request_id: u32, + /// Lease-bound session for transaction reads, otherwise zero. + session_id: u32, + /// Stable field identity. + field: FieldId, + /// Selected snapshot. + source: ValueSource, + /// Exact addressed endpoint. + target: Target, + }, + /// Starts one bounded CBOR snapshot import into the caller's transaction. + ImportBegin { + /// Client correlation identity. + request_id: u32, + /// Lease-bound transaction identity. + session_id: u32, + /// Optimistic transaction revision. + expected_revision: u32, + /// Complete CBOR byte length. + length: u16, + /// Exact schema identifier advertised by `KAPTAN_CONFIG_INFO`. + schema_id: [u8; 16], + /// CRC-32C over exactly `length` CBOR bytes. + crc32c: u32, + /// Exact addressed endpoint. + target: Target, + }, + /// Supplies one fixed-bounded CBOR import chunk. + ImportChunk { + /// Client correlation identity. + request_id: u32, + /// Lease-bound transaction identity. + session_id: u32, + /// Server-issued ID returned by the matching import-begin result. + transfer_id: u32, + /// Byte offset into the complete import. + offset: u16, + /// Used bytes of `data`. + length: u8, + /// Fixed MAVLink payload data. + data: [u8; 192], + /// Exact addressed endpoint. + target: Target, + }, + /// Checks the completed import and replaces only the RAM transaction. + ImportFinish { + /// Client correlation identity. + request_id: u32, + /// Lease-bound transaction identity. + session_id: u32, + /// Server-issued ID returned by the matching import-begin result. + transfer_id: u32, + /// Optimistic transaction revision. + expected_revision: u32, + /// Exact addressed endpoint. + target: Target, + }, + /// Starts an edit transaction from the expected committed generation. + Begin { + /// Client correlation identity. + request_id: u32, + /// Optimistic concurrency generation. + expected_generation: u64, + /// Exact addressed endpoint. + target: Target, + }, + /// Sets one typed scalar in the open transaction. + Set { + /// Client correlation identity. + request_id: u32, + /// Lease-bound transaction identity. + session_id: u32, + /// Optimistic transaction revision. + expected_revision: u32, + /// Stable schema-local field identity. + field: FieldId, + /// Typed value. + value: ConfigValue, + /// Exact addressed endpoint. + target: Target, + }, + /// Validates the complete candidate without persistence. + Validate { + /// Client correlation identity. + request_id: u32, + /// Lease-bound transaction identity. + session_id: u32, + /// Optimistic transaction revision. + expected_revision: u32, + /// Exact addressed endpoint. + target: Target, + }, + /// Validates and atomically persists the complete candidate. + Commit { + /// Client correlation identity. + request_id: u32, + /// Lease-bound transaction identity. + session_id: u32, + /// Optimistic committed generation. + expected_generation: u64, + /// Optimistic transaction revision. + expected_revision: u32, + /// Exact addressed endpoint. + target: Target, + }, + /// Discards the RAM transaction. + Abort { + /// Client correlation identity. + request_id: u32, + /// Lease-bound transaction identity. + session_id: u32, + /// Exact addressed endpoint. + target: Target, + }, +} + +impl ConfigCommand { + /// Returns the client request identity used for idempotency. + pub const fn request_id(self) -> u32 { + match self { + Self::InfoRequest { request_id, .. } + | Self::DescriptorRequest { request_id, .. } + | Self::ValueRequest { request_id, .. } + | Self::ImportBegin { request_id, .. } + | Self::ImportChunk { request_id, .. } + | Self::ImportFinish { request_id, .. } + | Self::Begin { request_id, .. } + | Self::Set { request_id, .. } + | Self::Validate { request_id, .. } + | Self::Commit { request_id, .. } + | Self::Abort { request_id, .. } => request_id, + } + } + + /// Returns the endpoint selected by this command. + pub const fn target(self) -> Target { + match self { + Self::InfoRequest { target, .. } + | Self::DescriptorRequest { target, .. } + | Self::ValueRequest { target, .. } + | Self::ImportBegin { target, .. } + | Self::ImportChunk { target, .. } + | Self::ImportFinish { target, .. } + | Self::Begin { target, .. } + | Self::Set { target, .. } + | Self::Validate { target, .. } + | Self::Commit { target, .. } + | Self::Abort { target, .. } => target, + } + } +} + +/// A capability token produced only by the future signing/replay authority. +/// +/// The token carries a stable principal identity so the service can bind a +/// transaction lease to one authenticated source. It is intentionally opaque: +/// an unverified raw MAVLink frame cannot create one. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct VerifiedAuthority { + principal: u64, + _private: (), +} + +impl VerifiedAuthority { + /// Returns the authenticated principal identity for ownership checks. + pub const fn principal(self) -> u64 { + self.principal + } + + #[cfg(any(test, feature = "host-authority"))] + /// Creates an explicit host/SITL authority token. + /// + /// This exists only for host integration and protocol tests. Flight code + /// must receive the equivalent token from the reviewed signing/replay path. + pub const fn host(principal: u64) -> Self { + Self { + principal, + _private: (), + } + } +} + +/// A command whose authority was verified before it reached this adapter. +#[derive(Clone, Copy, Debug, PartialEq)] +pub struct Authorized { + authority: VerifiedAuthority, + command: T, +} + +impl Authorized { + #[cfg(any(test, feature = "host-authority"))] + /// Wraps a host/SITL command in an explicit test-only authority token. + pub const fn host(principal: u64, command: T) -> Self { + Self { + authority: VerifiedAuthority::host(principal), + command, + } + } + + pub(crate) fn into_parts(self) -> (VerifiedAuthority, T) { + (self.authority, self.command) + } +} diff --git a/libs/kaptan-config-mavlink/src/lib.rs b/libs/kaptan-config-mavlink/src/lib.rs new file mode 100644 index 0000000..c0e5b06 --- /dev/null +++ b/libs/kaptan-config-mavlink/src/lib.rs @@ -0,0 +1,20 @@ +//! Kaptan custom-MAVLink configuration microservice adapter. +//! +//! This crate translates generated `kaptan.xml` records to the transport- +//! neutral configuration lifecycle. It does not parse raw frames, authenticate +//! peers, deserialize CBOR, access storage directly, or mutate active flight +//! consumers. + +#![no_std] + +mod command; +mod protocol; +pub mod service; +pub mod wire; + +pub use command::{Authorized, ConfigCommand, Target, ValueSource, VerifiedAuthority}; +pub use protocol::{ConfigReply, Operation, ProtocolError, ResultKind}; +pub use service::{ + CAPABILITY_CBOR_IMPORT, ConfigInfo, ConfigProtocol, DescriptorPage, NoSnapshotImporter, + ServiceResponse, SnapshotImporter, ValueResponse, +}; diff --git a/libs/kaptan-config-mavlink/src/protocol.rs b/libs/kaptan-config-mavlink/src/protocol.rs new file mode 100644 index 0000000..77fe003 --- /dev/null +++ b/libs/kaptan-config-mavlink/src/protocol.rs @@ -0,0 +1,194 @@ +//! Stable typed outcomes for the Kaptan configuration microservice. + +use kaptan_config::{ConfigState, ConfigViolation, FieldId}; +use kaptan_log::{EventHub, KaptanError, LogFlags, PublishOutcome}; +/// Operation reported in an authoritative result record. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum Operation { + /// Query/read operation. + Query, + /// Begin transaction. + Begin, + /// Scalar write. + Set, + /// Whole-candidate validation. + Validate, + /// Durable commit. + Commit, + /// Transaction abort. + Abort, + /// CBOR snapshot import into the isolated transaction. + Import, +} + +/// Stable result classification. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum ResultKind { + /// The operation was accepted. + Accepted, + /// A full snapshot committed but requires restart before activation. + CommittedRestartRequired, + /// The operation was rejected without an active-state transition. + Rejected, +} + +/// Compact protocol error code. Detail belongs in [`ConfigViolation`] when a +/// configuration invariant was responsible. +#[derive(Clone, Copy, Debug, Eq, PartialEq, KaptanError)] +pub enum ProtocolError { + /// The request target did not match this component. + BadTarget, + /// Authorization was absent or insufficient. + AuthorizationDenied, + /// The optimistic durable generation did not match. + GenerationConflict, + /// The single transaction slot is occupied. + SessionBusy, + /// The session is unknown, expired, or belongs to a different principal. + InvalidSession, + /// Reusing one request ID with different content is forbidden. + RequestConflict, + /// The request was not based on the current transaction revision. + StaleRevision, + /// A command is invalid in the current lifecycle state. + BadState, + /// The field is not present in this schema. + UnknownField(FieldId), + /// The typed setter does not match the field descriptor. + TypeMismatch(FieldId), + /// A local or complete validation issue. + Violation(ConfigViolation), + /// Persistence failed after a complete validation pass. + StorageFailure, + /// A CBOR snapshot could not be decoded into this schema. + CborMalformed, + /// A CBOR snapshot exceeded the fixed import budget. + CborTooLarge, + /// The complete CBOR snapshot did not match the supplied CRC-32C. + BlobCrc, + /// The uploaded CBOR blob was prepared for a different schema. + SchemaMismatch, + /// This deployment does not expose bounded CBOR transaction import. + ImportUnavailable, + /// The adapter reached an impossible internal state. + Internal, +} + +/// Authoritative completion for one request. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct ConfigReply { + /// Client request identity. + pub request_id: u32, + /// The current session, or zero when none exists. + pub session_id: u32, + /// Server-assigned import transfer identity, or zero when not importing. + pub transfer_id: u32, + /// The transaction revision, or zero when no transaction exists. + pub revision: u32, + /// Latest committed generation, or zero for the factory profile. + pub generation: u64, + /// Completed operation. + pub operation: Operation, + /// High-level result. + pub result: ResultKind, + /// Exact typed failure, if rejected. + pub error: Option, + /// Current lifecycle state after handling the request. + pub state: ConfigState, +} + +impl ConfigReply { + /// Stable primary field for the wire response, or the reserved sentinel. + pub const fn primary_field(self) -> u16 { + match self.error { + Some(ProtocolError::Violation(violation)) => match violation.primary() { + Some(field) => field.get(), + None => u16::MAX, + }, + Some(ProtocolError::UnknownField(field)) | Some(ProtocolError::TypeMismatch(field)) => { + field.get() + } + _ => u16::MAX, + } + } + + /// Stable related field for the wire response, or the reserved sentinel. + pub const fn related_field(self) -> u16 { + match self.error { + Some(ProtocolError::Violation(violation)) => match violation.related_field() { + Some(field) => field.get(), + None => u16::MAX, + }, + _ => u16::MAX, + } + } + + /// Stable validation-rule code, or zero when no validation rule applies. + pub const fn validation_rule(self) -> u16 { + match self.error { + Some(ProtocolError::Violation(violation)) => violation.rule().code(), + _ => 0, + } + } + + /// Enqueues a critical diagnostic companion for a rejected structured + /// result. The `KAPTAN_CONFIG_RESULT` remains the correlated, + /// field/rule-precise authority; this sends only the derive-generated + /// high-level `ProtocolError` display identity through `KAPTAN_ERROR`. + /// + /// The helper deliberately carries no request/session/generation fields + /// into the error channel. A composition owner may call it after it has + /// emitted or retained the authoritative result. + pub fn publish_rejection( + self, + hub: &mut EventHub, + ) -> Option { + if self.result != ResultKind::Rejected { + return None; + } + self.error + .as_ref() + .map(|error| hub.publish_error(LogFlags::NONE, error)) + } +} + +#[cfg(test)] +mod tests { + use kaptan_log::{EventHub, LogFilter, PublishOutcome, RecordChannel}; + + use super::{ConfigReply, Operation, ProtocolError, ResultKind}; + use kaptan_config::ConfigState; + + #[test] + fn rejected_result_enqueues_only_its_high_level_error_identity() { + let reply = ConfigReply { + request_id: 42, + session_id: 7, + transfer_id: 0, + revision: 3, + generation: 9, + operation: Operation::Set, + result: ResultKind::Rejected, + error: Some(ProtocolError::BadState), + state: ConfigState::Editing, + }; + let mut hub = EventHub::<1, 1>::new(LogFilter::default()); + assert_eq!( + reply.publish_rejection(&mut hub), + Some(PublishOutcome::Queued) + ); + let record = hub.peek().expect("rejection is retained"); + assert_eq!(record.channel(), RecordChannel::Error); + assert_eq!( + record.message().as_str(), + "kaptan_config_mavlink::protocol::ProtocolError::BadState" + ); + + let accepted = ConfigReply { + result: ResultKind::Accepted, + error: None, + ..reply + }; + assert_eq!(accepted.publish_rejection(&mut hub), None); + } +} diff --git a/libs/kaptan-config-mavlink/src/service.rs b/libs/kaptan-config-mavlink/src/service.rs new file mode 100644 index 0000000..8793966 --- /dev/null +++ b/libs/kaptan-config-mavlink/src/service.rs @@ -0,0 +1,14 @@ +//! Transport-neutral façade for the custom configuration service. +//! +//! Public callers see one small protocol API here. Its bounded query, +//! session, replay, and CBOR-import mechanics stay in private sibling modules +//! so no one concern becomes a second service owner. + +mod import; +mod query; +mod replay; +mod session; + +pub use import::{CAPABILITY_CBOR_IMPORT, NoSnapshotImporter, SnapshotImporter}; +pub use query::{ConfigInfo, DescriptorPage, ServiceResponse, ValueResponse}; +pub use session::ConfigProtocol; diff --git a/libs/kaptan-config-mavlink/src/service/import.rs b/libs/kaptan-config-mavlink/src/service/import.rs new file mode 100644 index 0000000..ed49c6a --- /dev/null +++ b/libs/kaptan-config-mavlink/src/service/import.rs @@ -0,0 +1,248 @@ +//! Bounded CBOR snapshot transfer state. + +use kaptan_config::{CommitError, ConfigManager, ConfigModel, ProfileStore}; + +use super::session::{ConfigProtocol, Session}; +use crate::{ConfigCommand, ConfigReply, Operation, ProtocolError, ResultKind}; + +/// Maximum complete CBOR snapshot accepted by the protocol. +pub(super) const IMPORT_MAXIMUM_BYTES: usize = 4096; +/// Payload bytes carried by each aligned import chunk. +pub(super) const IMPORT_CHUNK_BYTES: usize = 192; +pub(super) const IMPORT_CHUNK_COUNT: usize = IMPORT_MAXIMUM_BYTES.div_ceil(IMPORT_CHUNK_BYTES); + +/// `KAPTAN_CONFIG_INFO.capability_flags`: bounded CBOR transaction import. +pub const CAPABILITY_CBOR_IMPORT: u8 = 1; + +/// Decodes a complete bounded CBOR snapshot into this service's candidate. +/// +/// Vehicle-specific implementations live above this transport adapter so the +/// generic MAVLink layer never owns a profile schema or board capabilities. +pub trait SnapshotImporter { + /// Whether this concrete deployment exposes the bounded import path. + const IMPORT_SUPPORTED: bool; + /// Current format carried inside the deployment's raw CBOR profile blob. + const PROFILE_FORMAT: u16; + /// Converts one complete CBOR payload to a candidate. The service performs + /// the subsequent complete-context validation before replacing the RAM + /// transaction. + fn decode(&self, bytes: &[u8]) -> Result; +} + +/// Default importer used when a deployment deliberately does not expose CBOR +/// import yet. It keeps the service type usable for descriptor-only tests. +#[derive(Clone, Copy, Debug, Default)] +pub struct NoSnapshotImporter; + +impl SnapshotImporter for NoSnapshotImporter { + const IMPORT_SUPPORTED: bool = false; + const PROFILE_FORMAT: u16 = 0; + + fn decode(&self, _: &[u8]) -> Result { + Err(ProtocolError::CborMalformed) + } +} + +#[derive(Clone, Copy)] +pub(super) struct ImportBuffer { + owner: u64, + session_id: u32, + transfer_id: u32, + revision: u32, + length: u16, + crc32c: u32, + received: [bool; IMPORT_CHUNK_COUNT], + data: [u8; IMPORT_MAXIMUM_BYTES], +} + +impl ConfigProtocol { + pub(super) fn begin_import( + &mut self, + owner: u64, + command: ConfigCommand, + session: Session, + ) -> ConfigReply + where + C: ConfigModel, + I: SnapshotImporter, + { + let ConfigCommand::ImportBegin { + request_id, + length, + schema_id, + crc32c, + .. + } = command + else { + return self.rejected(command, ProtocolError::Internal); + }; + if !I::IMPORT_SUPPORTED { + return self.rejected(command, ProtocolError::ImportUnavailable); + } + if schema_id != C::SCHEMA_ID { + return self.rejected(command, ProtocolError::SchemaMismatch); + } + if length == 0 || usize::from(length) > IMPORT_MAXIMUM_BYTES { + return self.rejected(command, ProtocolError::CborTooLarge); + } + let Some(transfer_id) = self.allocate_transfer_id() else { + return self.rejected(command, ProtocolError::Internal); + }; + self.import = Some(ImportBuffer { + owner, + session_id: session.id, + transfer_id, + revision: session.revision, + length, + crc32c, + received: [false; IMPORT_CHUNK_COUNT], + data: [0; IMPORT_MAXIMUM_BYTES], + }); + let mut reply = self.reply( + request_id, + session.id, + session.revision, + Operation::Import, + ResultKind::Accepted, + None, + ); + reply.transfer_id = transfer_id; + reply + } + + pub(super) fn import_chunk(&mut self, command: ConfigCommand, session: Session) -> ConfigReply { + let ConfigCommand::ImportChunk { + request_id, + transfer_id, + offset, + length, + data, + .. + } = command + else { + return self.rejected(command, ProtocolError::Internal); + }; + let offset = usize::from(offset); + let length = usize::from(length); + let Some(import) = self.import.as_mut() else { + return self.rejected(command, ProtocolError::BadState); + }; + if import.owner != session.owner + || import.session_id != session.id + || import.transfer_id != transfer_id + { + return self.rejected(command, ProtocolError::InvalidSession); + } + if import.revision != session.revision { + return self.rejected(command, ProtocolError::StaleRevision); + } + let total = usize::from(import.length); + let Some(end) = offset.checked_add(length) else { + return self.rejected(command, ProtocolError::CborMalformed); + }; + if length == 0 + || offset % IMPORT_CHUNK_BYTES != 0 + || end > total + || (length != IMPORT_CHUNK_BYTES && end != total) + { + return self.rejected(command, ProtocolError::CborMalformed); + } + let chunk = offset / IMPORT_CHUNK_BYTES; + if import.received[chunk] { + if import.data[offset..end] != data[..length] { + return self.rejected(command, ProtocolError::RequestConflict); + } + } else { + import.data[offset..end].copy_from_slice(&data[..length]); + import.received[chunk] = true; + } + self.reply( + request_id, + session.id, + session.revision, + Operation::Import, + ResultKind::Accepted, + None, + ) + } + + pub(super) fn finish_import( + &mut self, + manager: &mut ConfigManager, + command: ConfigCommand, + session: Session, + request_id: u32, + transfer_id: u32, + ) -> ConfigReply + where + C: ConfigModel, + S: ProfileStore, + I: SnapshotImporter, + { + let Some(import) = self.import else { + return self.rejected(command, ProtocolError::BadState); + }; + if import.owner != session.owner + || import.session_id != session.id + || import.transfer_id != transfer_id + { + return self.rejected(command, ProtocolError::InvalidSession); + } + if import.revision != session.revision { + return self.rejected(command, ProtocolError::StaleRevision); + } + let total = usize::from(import.length); + let chunks = total.div_ceil(IMPORT_CHUNK_BYTES); + if import.received[..chunks].iter().any(|received| !received) { + return self.rejected(command, ProtocolError::CborMalformed); + } + if crc32c(&import.data[..total]) != import.crc32c { + return self.rejected(command, ProtocolError::BlobCrc); + } + let candidate = match self.importer.decode(&import.data[..total]) { + Ok(candidate) => candidate, + Err(error) => return self.rejected(command, error), + }; + let Some(next_revision) = session.revision.checked_add(1) else { + return self.rejected(command, ProtocolError::Internal); + }; + match manager.replace_transaction(candidate) { + Ok(()) => { + let Some(session) = self.session.as_mut() else { + return self.rejected(command, ProtocolError::Internal); + }; + session.revision = next_revision; + let (session_id, revision) = (session.id, session.revision); + self.import = None; + self.reply( + request_id, + session_id, + revision, + Operation::Import, + ResultKind::Accepted, + None, + ) + } + Err(CommitError::NoTransaction) => self.rejected(command, ProtocolError::BadState), + Err(CommitError::InvalidCandidate(violation)) => { + self.rejected(command, ProtocolError::Violation(violation)) + } + Err(CommitError::Store(_)) => self.rejected(command, ProtocolError::Internal), + } + } +} + +pub(super) fn crc32c(bytes: &[u8]) -> u32 { + let mut crc = !0_u32; + for byte in bytes { + crc ^= u32::from(*byte); + for _ in 0..8 { + crc = if crc & 1 == 0 { + crc >> 1 + } else { + (crc >> 1) ^ 0x82f6_3b78 + }; + } + } + !crc +} diff --git a/libs/kaptan-config-mavlink/src/service/query.rs b/libs/kaptan-config-mavlink/src/service/query.rs new file mode 100644 index 0000000..c56030b --- /dev/null +++ b/libs/kaptan-config-mavlink/src/service/query.rs @@ -0,0 +1,208 @@ +//! Read-only protocol discovery and value-query responses. + +use kaptan_config::{ + ConfigManager, ConfigModel, ConfigState, ConfigValue, FieldDescriptor, FieldId, Generation, + ProfileStore, +}; + +use super::{import::SnapshotImporter, session::ConfigProtocol}; +use crate::{ProtocolError, ValueSource}; + +/// Bounded descriptor page returned by a schema discovery request. +#[derive(Clone, Copy, Debug, PartialEq)] +pub struct DescriptorPage { + /// Request being answered. + pub request_id: u32, + /// First requested descriptor index. + pub first: u16, + /// Number of descriptors present in `descriptors`. + pub len: u8, + /// Fixed-capacity response page; `None` slots are unused. + pub descriptors: [Option<&'static FieldDescriptor>; 4], +} + +/// One typed value read from an explicitly selected configuration snapshot. +#[derive(Clone, Copy, Debug, PartialEq)] +pub struct ValueResponse { + /// Request being answered. + pub request_id: u32, + /// Transaction session for transaction values, otherwise zero. + pub session_id: u32, + /// Current transaction revision, otherwise zero. + pub revision: u32, + /// Requested stable field identity. + pub field: FieldId, + /// Declared scalar value. + pub value: ConfigValue, +} + +/// Correlated protocol/schema discovery response. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct ConfigInfo { + /// Request being answered. + pub request_id: u32, + /// Boot-active generation, or `None` for the factory profile. + pub active_generation: Option, + /// Last committed generation, or `None` for the factory profile. + pub committed_generation: Option, + /// Exact schema identifier expected by profile import. + pub schema_id: [u8; 16], + /// Version carried inside the one raw CBOR profile blob. + pub profile_format: u16, + /// Number of editable fields in the registry. + pub field_count: u16, + /// Current configuration lifecycle state. + pub state: ConfigState, + /// Deployment capabilities such as bounded CBOR import. + pub capability_flags: u8, +} + +/// Exact response produced by the custom configuration service. +#[derive(Clone, Copy, Debug, PartialEq)] +pub enum ServiceResponse { + /// Correlated protocol/schema discovery response. + Info(ConfigInfo), + /// Completion for a mutating operation or rejected query. + Result(crate::ConfigReply), + /// Bounded self-describing schema page. + Descriptors(DescriptorPage), + /// One typed scalar value. + Value(ValueResponse), +} + +impl ConfigProtocol { + /// Returns static descriptor metadata by stable schema index. + pub fn descriptor( + &self, + manager: &ConfigManager, + index: u16, + ) -> Option<&'static FieldDescriptor> + where + C: ConfigModel, + S: ProfileStore, + { + manager.descriptor(index) + } + + /// Reads one typed value without opening or changing a transaction. + pub fn value( + &self, + manager: &ConfigManager, + source: ValueSource, + field: FieldId, + ) -> Option + where + C: ConfigModel, + S: ProfileStore, + { + match source { + ValueSource::Active => manager.active_value(field), + ValueSource::Committed => manager.committed_value(field), + ValueSource::Transaction => manager.transaction_value(field), + } + } + + pub(super) fn descriptors( + &self, + manager: &ConfigManager, + request_id: u32, + first: u16, + requested: u16, + ) -> DescriptorPage + where + C: ConfigModel, + S: ProfileStore, + { + let mut descriptors = [None; 4]; + let count = requested.min(descriptors.len() as u16); + let mut len = 0; + for offset in 0..count { + let Some(index) = first.checked_add(offset) else { + break; + }; + let Some(descriptor) = manager.descriptor(index) else { + break; + }; + descriptors[usize::from(len)] = Some(descriptor); + len += 1; + } + DescriptorPage { + request_id, + first, + len, + descriptors, + } + } + + pub(super) fn info(&self, manager: &ConfigManager, request_id: u32) -> ConfigInfo + where + C: ConfigModel, + S: ProfileStore, + I: SnapshotImporter, + { + ConfigInfo { + request_id, + active_generation: manager.active_generation(), + committed_generation: manager.committed_generation(), + schema_id: C::SCHEMA_ID, + profile_format: I::PROFILE_FORMAT, + field_count: manager.field_count(), + state: self.state, + capability_flags: self.capability_flags_for::(), + } + } + + pub(super) fn value_response( + &self, + manager: &ConfigManager, + owner: u64, + request_id: u32, + session_id: u32, + field: FieldId, + source: ValueSource, + ) -> ServiceResponse + where + C: ConfigModel, + S: ProfileStore, + { + let (value, reply_session, revision) = match source { + ValueSource::Active => (manager.active_value(field), 0, 0), + ValueSource::Committed => (manager.committed_value(field), 0, 0), + ValueSource::Transaction => { + let Some(session) = self.session else { + return ServiceResponse::Result(self.rejected_for( + request_id, + session_id, + ProtocolError::InvalidSession, + )); + }; + if session.id != session_id || session.owner != owner { + return ServiceResponse::Result(self.rejected_for( + request_id, + session_id, + ProtocolError::InvalidSession, + )); + } + ( + manager.transaction_value(field), + session.id, + session.revision, + ) + } + }; + let Some(value) = value else { + return ServiceResponse::Result(self.rejected_for( + request_id, + session_id, + ProtocolError::UnknownField(field), + )); + }; + ServiceResponse::Value(ValueResponse { + request_id, + session_id: reply_session, + revision, + field, + value, + }) + } +} diff --git a/libs/kaptan-config-mavlink/src/service/replay.rs b/libs/kaptan-config-mavlink/src/service/replay.rs new file mode 100644 index 0000000..08643c7 --- /dev/null +++ b/libs/kaptan-config-mavlink/src/service/replay.rs @@ -0,0 +1,272 @@ +//! Bounded request-replay cache for idempotent client retries. + +use kaptan_config::ConfigValue; + +use super::session::ConfigProtocol; +use crate::{ConfigCommand, ConfigReply, ProtocolError}; + +pub(super) const REQUEST_CACHE_CAPACITY: usize = 8; + +#[derive(Clone, Copy)] +pub(super) struct RememberedResult { + pub(super) owner: u64, + pub(super) command: ConfigCommand, + pub(super) reply: ConfigReply, +} + +impl ConfigProtocol { + pub(super) fn cached(&self, owner: u64, command: ConfigCommand) -> Option { + for cached in self.remembered.iter().flatten() { + if cached.owner != owner || cached.command.request_id() != command.request_id() { + continue; + } + if same_command(cached.command, command) { + return Some(cached.reply); + } + return Some(self.rejected(command, ProtocolError::RequestConflict)); + } + None + } + + pub(super) fn remember(&mut self, result: RememberedResult) { + self.remembered[self.next_remembered] = Some(result); + self.next_remembered = (self.next_remembered + 1) % REQUEST_CACHE_CAPACITY; + } +} + +fn same_command(left: ConfigCommand, right: ConfigCommand) -> bool { + match (left, right) { + ( + ConfigCommand::InfoRequest { + request_id: left_request, + target: left_target, + }, + ConfigCommand::InfoRequest { + request_id: right_request, + target: right_target, + }, + ) => left_request == right_request && left_target == right_target, + ( + ConfigCommand::DescriptorRequest { + request_id: left_request, + first: left_first, + count: left_count, + target: left_target, + }, + ConfigCommand::DescriptorRequest { + request_id: right_request, + first: right_first, + count: right_count, + target: right_target, + }, + ) => { + left_request == right_request + && left_first == right_first + && left_count == right_count + && left_target == right_target + } + ( + ConfigCommand::ValueRequest { + request_id: left_request, + session_id: left_session, + field: left_field, + source: left_source, + target: left_target, + }, + ConfigCommand::ValueRequest { + request_id: right_request, + session_id: right_session, + field: right_field, + source: right_source, + target: right_target, + }, + ) => { + left_request == right_request + && left_session == right_session + && left_field == right_field + && left_source == right_source + && left_target == right_target + } + ( + ConfigCommand::Begin { + request_id: left_request, + expected_generation: left_generation, + target: left_target, + }, + ConfigCommand::Begin { + request_id: right_request, + expected_generation: right_generation, + target: right_target, + }, + ) => { + left_request == right_request + && left_generation == right_generation + && left_target == right_target + } + ( + ConfigCommand::Set { + request_id: left_request, + session_id: left_session, + expected_revision: left_revision, + field: left_field, + value: left_value, + target: left_target, + }, + ConfigCommand::Set { + request_id: right_request, + session_id: right_session, + expected_revision: right_revision, + field: right_field, + value: right_value, + target: right_target, + }, + ) => { + left_request == right_request + && left_session == right_session + && left_revision == right_revision + && left_field == right_field + && same_value(left_value, right_value) + && left_target == right_target + } + ( + ConfigCommand::Validate { + request_id: left_request, + session_id: left_session, + expected_revision: left_revision, + target: left_target, + }, + ConfigCommand::Validate { + request_id: right_request, + session_id: right_session, + expected_revision: right_revision, + target: right_target, + }, + ) => { + left_request == right_request + && left_session == right_session + && left_revision == right_revision + && left_target == right_target + } + ( + ConfigCommand::Commit { + request_id: left_request, + session_id: left_session, + expected_generation: left_generation, + expected_revision: left_revision, + target: left_target, + }, + ConfigCommand::Commit { + request_id: right_request, + session_id: right_session, + expected_generation: right_generation, + expected_revision: right_revision, + target: right_target, + }, + ) => { + left_request == right_request + && left_session == right_session + && left_generation == right_generation + && left_revision == right_revision + && left_target == right_target + } + ( + ConfigCommand::Abort { + request_id: left_request, + session_id: left_session, + target: left_target, + }, + ConfigCommand::Abort { + request_id: right_request, + session_id: right_session, + target: right_target, + }, + ) => { + left_request == right_request + && left_session == right_session + && left_target == right_target + } + ( + ConfigCommand::ImportBegin { + request_id: left_request, + session_id: left_session, + expected_revision: left_revision, + length: left_length, + schema_id: left_schema, + crc32c: left_crc, + target: left_target, + }, + ConfigCommand::ImportBegin { + request_id: right_request, + session_id: right_session, + expected_revision: right_revision, + length: right_length, + schema_id: right_schema, + crc32c: right_crc, + target: right_target, + }, + ) => { + left_request == right_request + && left_session == right_session + && left_revision == right_revision + && left_length == right_length + && left_schema == right_schema + && left_crc == right_crc + && left_target == right_target + } + ( + ConfigCommand::ImportChunk { + request_id: left_request, + session_id: left_session, + transfer_id: left_transfer, + offset: left_offset, + length: left_length, + data: left_data, + target: left_target, + }, + ConfigCommand::ImportChunk { + request_id: right_request, + session_id: right_session, + transfer_id: right_transfer, + offset: right_offset, + length: right_length, + data: right_data, + target: right_target, + }, + ) => { + left_request == right_request + && left_session == right_session + && left_transfer == right_transfer + && left_offset == right_offset + && left_length == right_length + && left_data == right_data + && left_target == right_target + } + ( + ConfigCommand::ImportFinish { + request_id: left_request, + session_id: left_session, + transfer_id: left_transfer, + expected_revision: left_revision, + target: left_target, + }, + ConfigCommand::ImportFinish { + request_id: right_request, + session_id: right_session, + transfer_id: right_transfer, + expected_revision: right_revision, + target: right_target, + }, + ) => { + left_request == right_request + && left_session == right_session + && left_transfer == right_transfer + && left_revision == right_revision + && left_target == right_target + } + _ => false, + } +} + +fn same_value(left: ConfigValue, right: ConfigValue) -> bool { + left.kind() == right.kind() && left.raw_bits() == right.raw_bits() +} diff --git a/libs/kaptan-config-mavlink/src/service/session.rs b/libs/kaptan-config-mavlink/src/service/session.rs new file mode 100644 index 0000000..33e63bf --- /dev/null +++ b/libs/kaptan-config-mavlink/src/service/session.rs @@ -0,0 +1,1094 @@ +//! Lease-bound command service over the transport-neutral configuration core. + +use core::num::NonZeroU32; + +use kaptan_config::{ + BeginError, CommitError, ConfigManager, ConfigModel, ConfigState, FieldWriteError, Generation, + ProfileStore, SetError, +}; + +use super::{ + import::{CAPABILITY_CBOR_IMPORT, ImportBuffer, NoSnapshotImporter, SnapshotImporter}, + query::ServiceResponse, + replay::{REQUEST_CACHE_CAPACITY, RememberedResult}, +}; +#[cfg(test)] +use super::{ + import::{IMPORT_CHUNK_BYTES, crc32c}, + query::{ConfigInfo, DescriptorPage, ValueResponse}, +}; +use crate::{ + command::{Authorized, ConfigCommand, Target}, + protocol::{ConfigReply, Operation, ProtocolError, ResultKind}, +}; + +#[derive(Clone, Copy)] +pub(super) struct Session { + pub(super) id: u32, + pub(super) owner: u64, + pub(super) revision: u32, +} + +/// Owns only the one GCS transaction lease and bounded protocol state. +/// +/// A future signing/replay dispatcher creates [`Authorized`] commands. This +/// protocol never accepts raw frames and can be expired by the composition +/// owner on lease or link loss. The same composition owner retains the one +/// [`ConfigManager`] and one runtime log hub. +pub struct ConfigProtocol { + pub(super) session: Option, + pub(super) next_session_id: Option, + pub(super) next_transfer_id: Option, + pub(super) target: Target, + pub(super) remembered: [Option; REQUEST_CACHE_CAPACITY], + pub(super) next_remembered: usize, + pub(super) importer: I, + pub(super) import: Option, + pub(super) generation: u64, + pub(super) state: ConfigState, +} + +impl ConfigProtocol { + /// Creates a custom-protocol state machine without CBOR import support. + pub const fn new(target: Target) -> Self { + Self::with_importer(target, NoSnapshotImporter) + } +} + +impl ConfigProtocol { + /// Creates a custom-protocol state machine with a vehicle-specific CBOR + /// importer. Profile/store ownership stays outside this adapter. + pub const fn with_importer(target: Target, importer: I) -> Self { + Self { + session: None, + next_session_id: Some(NonZeroU32::MIN), + next_transfer_id: Some(NonZeroU32::MIN), + target, + remembered: [None; REQUEST_CACHE_CAPACITY], + next_remembered: 0, + importer, + import: None, + generation: 0, + state: ConfigState::Active, + } + } +} + +impl ConfigProtocol { + /// Feature bits for `KAPTAN_CONFIG_INFO` for a concrete candidate type. + pub const fn capability_flags_for(&self) -> u8 + where + I: SnapshotImporter, + { + if I::IMPORT_SUPPORTED { + CAPABILITY_CBOR_IMPORT + } else { + 0 + } + } + + /// Dispatches one authorized configuration command. The authoritative + /// `KAPTAN_CONFIG_RESULT` response remains the complete configuration + /// audit; generic human logging is intentionally independent. + pub fn dispatch( + &mut self, + manager: &mut ConfigManager, + authorized: Authorized, + ) -> ServiceResponse + where + C: ConfigModel, + S: ProfileStore, + I: SnapshotImporter, + { + self.refresh(manager); + let (authority, command) = authorized.into_parts(); + self.dispatch_inner(manager, authority.principal(), command) + } + + /// Expires the bounded session and drops only its RAM transaction. + pub fn expire(&mut self, manager: &mut ConfigManager) -> bool + where + C: ConfigModel, + S: ProfileStore, + { + let expired = self.session.take().is_some(); + if expired { + manager.abort(); + self.import = None; + self.remembered = [None; REQUEST_CACHE_CAPACITY]; + self.next_remembered = 0; + } + expired + } + + fn dispatch_inner( + &mut self, + manager: &mut ConfigManager, + owner: u64, + command: ConfigCommand, + ) -> ServiceResponse + where + C: ConfigModel, + S: ProfileStore, + I: SnapshotImporter, + { + self.refresh(manager); + if command.target() != self.target { + return ServiceResponse::Result(self.rejected(command, ProtocolError::BadTarget)); + } + if let Some(cached) = self.cached(owner, command) { + return ServiceResponse::Result(cached); + } + let response = match command { + ConfigCommand::InfoRequest { request_id, .. } => { + ServiceResponse::Info(self.info(manager, request_id)) + } + ConfigCommand::DescriptorRequest { + request_id, + first, + count, + .. + } => ServiceResponse::Descriptors(self.descriptors(manager, request_id, first, count)), + ConfigCommand::ValueRequest { + request_id, + session_id, + field, + source, + .. + } => self.value_response(manager, owner, request_id, session_id, field, source), + ConfigCommand::ImportBegin { .. } + | ConfigCommand::ImportChunk { .. } + | ConfigCommand::ImportFinish { .. } => { + ServiceResponse::Result(self.in_session(manager, owner, command)) + } + ConfigCommand::Begin { .. } => { + ServiceResponse::Result(self.begin(manager, owner, command)) + } + ConfigCommand::Set { .. } + | ConfigCommand::Validate { .. } + | ConfigCommand::Commit { .. } + | ConfigCommand::Abort { .. } => { + ServiceResponse::Result(self.in_session(manager, owner, command)) + } + }; + self.refresh(manager); + match response { + ServiceResponse::Result(reply) => { + self.remember(RememberedResult { + owner, + command, + reply, + }); + ServiceResponse::Result(reply) + } + response => response, + } + } + + fn begin( + &mut self, + manager: &mut ConfigManager, + owner: u64, + command: ConfigCommand, + ) -> ConfigReply + where + C: ConfigModel, + S: ProfileStore, + { + let ConfigCommand::Begin { + request_id, + expected_generation, + .. + } = command + else { + return self.rejected(command, ProtocolError::Internal); + }; + if self.session.is_some() { + return self.rejected(command, ProtocolError::SessionBusy); + } + if expected_generation != self.generation() { + return self.rejected(command, ProtocolError::GenerationConflict); + } + let Some(id) = self.allocate_session_id() else { + return self.rejected(command, ProtocolError::Internal); + }; + match manager.begin() { + Ok(()) => { + self.session = Some(Session { + id, + owner, + revision: 0, + }); + self.reply( + request_id, + id, + 0, + Operation::Begin, + ResultKind::Accepted, + None, + ) + } + Err(BeginError::TransactionActive) => { + self.rejected(command, ProtocolError::SessionBusy) + } + } + } + + fn in_session( + &mut self, + manager: &mut ConfigManager, + owner: u64, + command: ConfigCommand, + ) -> ConfigReply + where + C: ConfigModel, + S: ProfileStore, + I: SnapshotImporter, + { + let Some(session) = self.session else { + return self.rejected(command, ProtocolError::InvalidSession); + }; + let session_id = command_session_id(command); + if session.id != session_id || session.owner != owner { + return self.rejected(command, ProtocolError::InvalidSession); + } + if let Some(expected_revision) = command_expected_revision(command) + && expected_revision != session.revision + { + return self.rejected(command, ProtocolError::StaleRevision); + } + match command { + ConfigCommand::ImportBegin { .. } => self.begin_import::(owner, command, session), + ConfigCommand::ImportChunk { .. } => self.import_chunk(command, session), + ConfigCommand::ImportFinish { + request_id, + transfer_id, + .. + } => self.finish_import(manager, command, session, request_id, transfer_id), + ConfigCommand::Set { + request_id, + field, + value, + .. + } => { + let Some(next_revision) = session.revision.checked_add(1) else { + return self.rejected(command, ProtocolError::Internal); + }; + match manager.set(field, value) { + Ok(_) => { + let Some(session) = self.session.as_mut() else { + return self.rejected(command, ProtocolError::Internal); + }; + session.revision = next_revision; + let (session_id, revision) = (session.id, session.revision); + self.reply( + request_id, + session_id, + revision, + Operation::Set, + ResultKind::Accepted, + None, + ) + } + Err(SetError::NoTransaction) => self.rejected(command, ProtocolError::BadState), + Err(SetError::Field(error)) => self.rejected(command, field_error(error)), + } + } + ConfigCommand::Validate { request_id, .. } => match manager.validate() { + Ok(()) => self.reply( + request_id, + session.id, + session.revision, + Operation::Validate, + ResultKind::Accepted, + None, + ), + Err(CommitError::NoTransaction) => self.rejected(command, ProtocolError::BadState), + Err(CommitError::InvalidCandidate(violation)) => { + self.rejected(command, ProtocolError::Violation(violation)) + } + Err(CommitError::Store(_)) => self.rejected(command, ProtocolError::Internal), + }, + ConfigCommand::Commit { + request_id, + expected_generation, + .. + } => { + if expected_generation != self.generation() { + return self.rejected(command, ProtocolError::GenerationConflict); + } + match manager.commit() { + Ok(_) => { + self.session = None; + self.import = None; + self.refresh(manager); + self.reply( + request_id, + 0, + session.revision, + Operation::Commit, + ResultKind::CommittedRestartRequired, + None, + ) + } + Err(CommitError::NoTransaction) => { + self.rejected(command, ProtocolError::BadState) + } + Err(CommitError::InvalidCandidate(violation)) => { + self.rejected(command, ProtocolError::Violation(violation)) + } + Err(CommitError::Store(_)) => { + self.rejected(command, ProtocolError::StorageFailure) + } + } + } + ConfigCommand::Abort { request_id, .. } => { + manager.abort(); + self.session = None; + self.import = None; + self.refresh(manager); + self.reply( + request_id, + 0, + session.revision, + Operation::Abort, + ResultKind::Accepted, + None, + ) + } + ConfigCommand::InfoRequest { .. } + | ConfigCommand::DescriptorRequest { .. } + | ConfigCommand::ValueRequest { .. } + | ConfigCommand::Begin { .. } => self.rejected(command, ProtocolError::Internal), + } + } + + fn allocate_session_id(&mut self) -> Option { + let id = self.next_session_id?; + self.next_session_id = id.get().checked_add(1).and_then(NonZeroU32::new); + Some(id.get()) + } + + pub(super) fn allocate_transfer_id(&mut self) -> Option { + let id = self.next_transfer_id?; + self.next_transfer_id = id.get().checked_add(1).and_then(NonZeroU32::new); + Some(id.get()) + } + + fn refresh(&mut self, manager: &ConfigManager) + where + C: ConfigModel, + S: ProfileStore, + { + self.generation = manager.generation().map_or(0, Generation::get); + self.state = manager.state(); + } + + fn generation(&self) -> u64 { + self.generation + } + + pub(super) fn reply( + &self, + request_id: u32, + session_id: u32, + revision: u32, + operation: Operation, + result: ResultKind, + error: Option, + ) -> ConfigReply { + ConfigReply { + request_id, + session_id, + transfer_id: 0, + revision, + generation: self.generation, + operation, + result, + error, + state: self.state, + } + } + + pub(super) fn rejected(&self, command: ConfigCommand, error: ProtocolError) -> ConfigReply { + self.reply( + command.request_id(), + command_session_id(command), + self.session.map_or(0, |session| session.revision), + operation(command), + ResultKind::Rejected, + Some(error), + ) + } + + pub(super) fn rejected_for( + &self, + request_id: u32, + session_id: u32, + error: ProtocolError, + ) -> ConfigReply { + self.reply( + request_id, + session_id, + self.session.map_or(0, |session| session.revision), + Operation::Query, + ResultKind::Rejected, + Some(error), + ) + } +} + +fn command_session_id(command: ConfigCommand) -> u32 { + match command { + ConfigCommand::InfoRequest { .. } + | ConfigCommand::DescriptorRequest { .. } + | ConfigCommand::Begin { .. } => 0, + ConfigCommand::ValueRequest { session_id, .. } => session_id, + ConfigCommand::Set { session_id, .. } + | ConfigCommand::ImportBegin { session_id, .. } + | ConfigCommand::ImportChunk { session_id, .. } + | ConfigCommand::ImportFinish { session_id, .. } + | ConfigCommand::Validate { session_id, .. } + | ConfigCommand::Commit { session_id, .. } + | ConfigCommand::Abort { session_id, .. } => session_id, + } +} + +fn command_expected_revision(command: ConfigCommand) -> Option { + match command { + ConfigCommand::Set { + expected_revision, .. + } + | ConfigCommand::ImportBegin { + expected_revision, .. + } + | ConfigCommand::ImportFinish { + expected_revision, .. + } + | ConfigCommand::Validate { + expected_revision, .. + } + | ConfigCommand::Commit { + expected_revision, .. + } => Some(expected_revision), + ConfigCommand::InfoRequest { .. } + | ConfigCommand::DescriptorRequest { .. } + | ConfigCommand::ValueRequest { .. } + | ConfigCommand::ImportChunk { .. } + | ConfigCommand::Begin { .. } + | ConfigCommand::Abort { .. } => None, + } +} + +const fn operation(command: ConfigCommand) -> Operation { + match command { + ConfigCommand::InfoRequest { .. } + | ConfigCommand::DescriptorRequest { .. } + | ConfigCommand::ValueRequest { .. } => Operation::Query, + ConfigCommand::ImportBegin { .. } + | ConfigCommand::ImportChunk { .. } + | ConfigCommand::ImportFinish { .. } => Operation::Import, + ConfigCommand::Begin { .. } => Operation::Begin, + ConfigCommand::Set { .. } => Operation::Set, + ConfigCommand::Validate { .. } => Operation::Validate, + ConfigCommand::Commit { .. } => Operation::Commit, + ConfigCommand::Abort { .. } => Operation::Abort, + } +} + +fn field_error(error: FieldWriteError) -> ProtocolError { + match error { + FieldWriteError::UnknownField(field) => ProtocolError::UnknownField(field), + FieldWriteError::WrongType { field, .. } => ProtocolError::TypeMismatch(field), + FieldWriteError::Violation(violation) => ProtocolError::Violation(violation), + } +} + +#[cfg(test)] +mod tests { + use kaptan_config::{ + ConfigManager, ConfigModel, ConfigValue, ConfigViolation, FieldDescriptor, FieldId, + FieldWriteError, Generation, ProfileStore, ValidationRule, + }; + + use super::*; + use crate::{Authorized, ConfigCommand, ProtocolError, ResultKind, Target, ValueSource}; + + const VALUE: FieldId = FieldId::new(1); + const DESCRIPTOR: FieldDescriptor = FieldDescriptor { + id: VALUE, + display_key: "test.value", + kind: kaptan_config::ValueKind::U32, + minimum: Some(ConfigValue::U32(0)), + maximum: Some(ConfigValue::U32(10)), + default: ConfigValue::U32(2), + unit: "count", + apply_policy: kaptan_config::ApplyPolicy::RestartRequired, + }; + + #[derive(Clone)] + struct Candidate(u32); + + impl ConfigModel for Candidate { + type Active = u32; + type Context = (); + const SCHEMA_ID: [u8; 16] = *b"CONFIGSERVICE-01"; + + fn descriptor(index: u16) -> Option<&'static FieldDescriptor> { + (index == 0).then_some(&DESCRIPTOR) + } + fn field_count() -> u16 { + 1 + } + fn field_value(&self, id: FieldId) -> Option { + (id == VALUE).then_some(ConfigValue::U32(self.0)) + } + fn active_field_value(active: &Self::Active, id: FieldId) -> Option { + (id == VALUE).then_some(ConfigValue::U32(*active)) + } + fn set_field(&mut self, id: FieldId, value: ConfigValue) -> Result<(), FieldWriteError> { + if id != VALUE { + return Err(FieldWriteError::UnknownField(id)); + } + let ConfigValue::U32(value) = value else { + return Err(FieldWriteError::WrongType { + field: id, + expected: kaptan_config::ValueKind::U32, + actual: value.kind(), + }); + }; + if value > 10 { + return Err(FieldWriteError::Violation(ConfigViolation::field( + id, + ValidationRule::AboveMaximum, + ))); + } + self.0 = value; + Ok(()) + } + fn validate(&self, _: &Self::Context) -> Result { + Ok(self.0) + } + } + + struct Store; + impl ProfileStore for Store { + type Error = (); + fn load(&mut self) -> Result, Self::Error> { + Ok(None) + } + fn persist(&mut self, _: &Candidate) -> Result { + Ok(Generation::new(1)) + } + } + + struct Importer; + + impl SnapshotImporter for Importer { + const IMPORT_SUPPORTED: bool = true; + const PROFILE_FORMAT: u16 = 1; + + fn decode(&self, bytes: &[u8]) -> Result { + if bytes != [7] + && !(bytes.len() == IMPORT_CHUNK_BYTES + 1 + && bytes[0] == 7 + && bytes[IMPORT_CHUNK_BYTES] == 9) + { + return Err(ProtocolError::CborMalformed); + } + Ok(Candidate(7)) + } + } + + /// Test-only composition root. Production code must retain these two + /// objects separately; this wrapper keeps protocol examples compact while + /// exercising the same externally-owned-manager dispatch API. + struct TestService { + manager: ConfigManager, + protocol: ConfigProtocol, + } + + impl TestService { + fn new(manager: ConfigManager, target: Target) -> Self { + Self { + manager, + protocol: ConfigProtocol::new(target), + } + } + } + + impl TestService + where + I: SnapshotImporter, + { + fn with_importer( + manager: ConfigManager, + target: Target, + importer: I, + ) -> Self { + Self { + manager, + protocol: ConfigProtocol::with_importer(target, importer), + } + } + + fn dispatch(&mut self, command: Authorized) -> ServiceResponse { + self.protocol.dispatch(&mut self.manager, command) + } + + fn manager(&self) -> &ConfigManager { + &self.manager + } + + fn capability_flags(&self) -> u8 { + self.protocol.capability_flags_for::() + } + } + + fn result(response: ServiceResponse) -> ConfigReply { + match response { + ServiceResponse::Result(reply) => reply, + ServiceResponse::Info(_) + | ServiceResponse::Descriptors(_) + | ServiceResponse::Value(_) => { + panic!("test expected a result response") + } + } + } + + #[test] + fn retries_are_idempotent_and_wrong_owner_cannot_mutate() { + let manager = ConfigManager::new(Candidate(2), (), Store).unwrap(); + let mut service = TestService::new(manager, Target::new(1, 1)); + let begin = ConfigCommand::Begin { + request_id: 1, + expected_generation: 0, + target: Target::new(1, 1), + }; + let started = result(service.dispatch(Authorized::host(7, begin))); + assert_eq!(started.result, ResultKind::Accepted); + let set = ConfigCommand::Set { + request_id: 2, + session_id: started.session_id, + expected_revision: 0, + field: VALUE, + value: ConfigValue::U32(3), + target: Target::new(1, 1), + }; + let accepted = result(service.dispatch(Authorized::host(7, set))); + assert_eq!(accepted.result, ResultKind::Accepted); + assert_eq!(result(service.dispatch(Authorized::host(7, set))), accepted); + let foreign = result(service.dispatch(Authorized::host( + 8, + ConfigCommand::Set { + request_id: 3, + session_id: started.session_id, + expected_revision: 1, + field: VALUE, + value: ConfigValue::U32(3), + target: Target::new(1, 1), + }, + ))); + assert_eq!(foreign.error, Some(ProtocolError::InvalidSession)); + } + + #[test] + fn exact_validation_issue_reaches_result() { + let manager = ConfigManager::new(Candidate(2), (), Store).unwrap(); + let mut service = TestService::new(manager, Target::new(1, 1)); + let begin = result(service.dispatch(Authorized::host( + 7, + ConfigCommand::Begin { + request_id: 1, + expected_generation: 0, + target: Target::new(1, 1), + }, + ))); + let rejected = result(service.dispatch(Authorized::host( + 7, + ConfigCommand::Set { + request_id: 2, + session_id: begin.session_id, + expected_revision: 0, + field: VALUE, + value: ConfigValue::U32(11), + target: Target::new(1, 1), + }, + ))); + assert!( + matches!(rejected.error, Some(ProtocolError::Violation(issue)) if issue.rule() == ValidationRule::AboveMaximum) + ); + } + + #[test] + fn discovery_and_value_reads_are_service_responses() { + let manager = ConfigManager::new(Candidate(2), (), Store).unwrap(); + let mut service = TestService::new(manager, Target::new(1, 1)); + let descriptors = service.dispatch(Authorized::host( + 7, + ConfigCommand::DescriptorRequest { + request_id: 1, + first: 0, + count: 4, + target: Target::new(1, 1), + }, + )); + assert!(matches!( + descriptors, + ServiceResponse::Descriptors(DescriptorPage { + len: 1, + descriptors: [Some(&DESCRIPTOR), None, None, None], + .. + }) + )); + let value = service.dispatch(Authorized::host( + 7, + ConfigCommand::ValueRequest { + request_id: 2, + session_id: 0, + field: VALUE, + source: ValueSource::Active, + target: Target::new(1, 1), + }, + )); + assert!(matches!( + value, + ServiceResponse::Value(ValueResponse { + value: ConfigValue::U32(2), + .. + }) + )); + + let info = service.dispatch(Authorized::host( + 7, + ConfigCommand::InfoRequest { + request_id: 3, + target: Target::new(1, 1), + }, + )); + assert!(matches!( + info, + ServiceResponse::Info(ConfigInfo { + request_id: 3, + field_count: 1, + profile_format: 0, + capability_flags: 0, + .. + }) + )); + } + + #[test] + fn complete_cbor_import_replaces_only_the_open_transaction() { + let manager = ConfigManager::new(Candidate(2), (), Store).unwrap(); + let mut service = TestService::with_importer(manager, Target::new(1, 1), Importer); + let begin = result(service.dispatch(Authorized::host( + 7, + ConfigCommand::Begin { + request_id: 1, + expected_generation: 0, + target: Target::new(1, 1), + }, + ))); + let bytes = [7]; + let mut chunk = [0; IMPORT_CHUNK_BYTES]; + chunk[0] = 7; + let import_begin = result(service.dispatch(Authorized::host( + 7, + ConfigCommand::ImportBegin { + request_id: 2, + session_id: begin.session_id, + expected_revision: 0, + length: 1, + schema_id: Candidate::SCHEMA_ID, + crc32c: crc32c(&bytes), + target: Target::new(1, 1), + }, + ))); + assert_eq!(import_begin.result, ResultKind::Accepted); + assert_ne!(import_begin.transfer_id, 0); + assert_eq!( + result(service.dispatch(Authorized::host( + 7, + ConfigCommand::ImportChunk { + request_id: 3, + session_id: begin.session_id, + transfer_id: import_begin.transfer_id, + offset: 0, + length: 1, + data: chunk, + target: Target::new(1, 1), + }, + ))) + .result, + ResultKind::Accepted + ); + let complete = result(service.dispatch(Authorized::host( + 7, + ConfigCommand::ImportFinish { + request_id: 4, + session_id: begin.session_id, + transfer_id: import_begin.transfer_id, + expected_revision: 0, + target: Target::new(1, 1), + }, + ))); + assert_eq!(complete.result, ResultKind::Accepted); + assert_eq!(complete.revision, 1); + assert_eq!( + service.manager().active_value(VALUE), + Some(ConfigValue::U32(2)) + ); + assert_eq!( + service.manager().transaction_value(VALUE), + Some(ConfigValue::U32(7)) + ); + } + + #[test] + fn stale_import_transfer_cannot_feed_a_new_import() { + let manager = ConfigManager::new(Candidate(2), (), Store).unwrap(); + let mut service = TestService::with_importer(manager, Target::new(1, 1), Importer); + let session = result(service.dispatch(Authorized::host( + 7, + ConfigCommand::Begin { + request_id: 1, + expected_generation: 0, + target: Target::new(1, 1), + }, + ))); + let first = result(service.dispatch(Authorized::host( + 7, + ConfigCommand::ImportBegin { + request_id: 2, + session_id: session.session_id, + expected_revision: 0, + length: 1, + schema_id: Candidate::SCHEMA_ID, + crc32c: crc32c(&[7]), + target: Target::new(1, 1), + }, + ))); + let second = result(service.dispatch(Authorized::host( + 7, + ConfigCommand::ImportBegin { + request_id: 3, + session_id: session.session_id, + expected_revision: 0, + length: 1, + schema_id: Candidate::SCHEMA_ID, + crc32c: crc32c(&[7]), + target: Target::new(1, 1), + }, + ))); + assert_ne!(first.transfer_id, second.transfer_id); + let mut chunk = [0; IMPORT_CHUNK_BYTES]; + chunk[0] = 7; + let stale = result(service.dispatch(Authorized::host( + 7, + ConfigCommand::ImportChunk { + request_id: 4, + session_id: session.session_id, + transfer_id: first.transfer_id, + offset: 0, + length: 1, + data: chunk, + target: Target::new(1, 1), + }, + ))); + assert_eq!(stale.error, Some(ProtocolError::InvalidSession)); + } + + #[test] + fn import_accepts_aligned_out_of_order_idempotent_chunks_and_rejects_conflicts() { + let manager = ConfigManager::new(Candidate(2), (), Store).unwrap(); + let mut service = TestService::with_importer(manager, Target::new(1, 1), Importer); + let session = result(service.dispatch(Authorized::host( + 7, + ConfigCommand::Begin { + request_id: 1, + expected_generation: 0, + target: Target::new(1, 1), + }, + ))); + let mut bytes = [0_u8; IMPORT_CHUNK_BYTES + 1]; + bytes[0] = 7; + bytes[IMPORT_CHUNK_BYTES] = 9; + let imported = result(service.dispatch(Authorized::host( + 7, + ConfigCommand::ImportBegin { + request_id: 2, + session_id: session.session_id, + expected_revision: 0, + length: bytes.len() as u16, + schema_id: Candidate::SCHEMA_ID, + crc32c: crc32c(&bytes), + target: Target::new(1, 1), + }, + ))); + let mut tail = [0_u8; IMPORT_CHUNK_BYTES]; + tail[0] = 9; + let tail_command = ConfigCommand::ImportChunk { + request_id: 3, + session_id: session.session_id, + transfer_id: imported.transfer_id, + offset: IMPORT_CHUNK_BYTES as u16, + length: 1, + data: tail, + target: Target::new(1, 1), + }; + assert_eq!( + result(service.dispatch(Authorized::host(7, tail_command))).result, + ResultKind::Accepted + ); + + let mut head = [0_u8; IMPORT_CHUNK_BYTES]; + head[0] = 7; + let head_command = ConfigCommand::ImportChunk { + request_id: 4, + session_id: session.session_id, + transfer_id: imported.transfer_id, + offset: 0, + length: IMPORT_CHUNK_BYTES as u8, + data: head, + target: Target::new(1, 1), + }; + assert_eq!( + result(service.dispatch(Authorized::host(7, head_command))).result, + ResultKind::Accepted + ); + let retry = ConfigCommand::ImportChunk { + request_id: 5, + session_id: session.session_id, + transfer_id: imported.transfer_id, + offset: 0, + length: IMPORT_CHUNK_BYTES as u8, + data: head, + target: Target::new(1, 1), + }; + assert_eq!( + result(service.dispatch(Authorized::host(7, retry))).result, + ResultKind::Accepted + ); + let mut conflicting = head; + conflicting[0] = 8; + let conflict = result(service.dispatch(Authorized::host( + 7, + ConfigCommand::ImportChunk { + request_id: 6, + data: conflicting, + session_id: session.session_id, + transfer_id: imported.transfer_id, + offset: 0, + length: IMPORT_CHUNK_BYTES as u8, + target: Target::new(1, 1), + }, + ))); + assert_eq!(conflict.error, Some(ProtocolError::RequestConflict)); + let complete = result(service.dispatch(Authorized::host( + 7, + ConfigCommand::ImportFinish { + request_id: 7, + session_id: session.session_id, + transfer_id: imported.transfer_id, + expected_revision: 0, + target: Target::new(1, 1), + }, + ))); + assert_eq!(complete.result, ResultKind::Accepted); + assert_eq!( + service.manager().transaction_value(VALUE), + Some(ConfigValue::U32(7)) + ); + } + + #[test] + fn configuration_result_remains_the_structured_failure_authority() { + let manager = ConfigManager::new(Candidate(2), (), Store).unwrap(); + let mut service = TestService::new(manager, Target::new(1, 1)); + let begin = result(service.dispatch(Authorized::host( + 7, + ConfigCommand::Begin { + request_id: 1, + expected_generation: 0, + target: Target::new(1, 1), + }, + ))); + let rejected = result(service.dispatch(Authorized::host( + 7, + ConfigCommand::Set { + request_id: 2, + session_id: begin.session_id, + expected_revision: 0, + field: FieldId::new(2), + value: ConfigValue::U32(3), + target: Target::new(1, 1), + }, + ))); + assert_eq!( + rejected.error, + Some(ProtocolError::UnknownField(FieldId::new(2))) + ); + assert_eq!(rejected.request_id, 2); + assert_eq!(rejected.primary_field(), FieldId::new(2).get()); + } + + #[test] + fn import_capability_truthfully_reflects_the_installed_importer() { + let default_manager = ConfigManager::new(Candidate(2), (), Store).unwrap(); + let mut default_service = TestService::new(default_manager, Target::new(1, 1)); + assert_eq!(default_service.capability_flags(), 0); + let begin = result(default_service.dispatch(Authorized::host( + 7, + ConfigCommand::Begin { + request_id: 1, + expected_generation: 0, + target: Target::new(1, 1), + }, + ))); + let unavailable = result(default_service.dispatch(Authorized::host( + 7, + ConfigCommand::ImportBegin { + request_id: 2, + session_id: begin.session_id, + expected_revision: 0, + length: 1, + schema_id: [0; 16], + crc32c: 0, + target: Target::new(1, 1), + }, + ))); + assert_eq!(unavailable.error, Some(ProtocolError::ImportUnavailable)); + let importing_manager = ConfigManager::new(Candidate(2), (), Store).unwrap(); + let importing_service = + TestService::with_importer(importing_manager, Target::new(1, 1), Importer); + assert_eq!(importing_service.capability_flags(), CAPABILITY_CBOR_IMPORT); + } + + #[test] + fn import_requires_the_discovered_schema_id_before_receiving_bytes() { + let manager = ConfigManager::new(Candidate(2), (), Store).unwrap(); + let mut service = TestService::with_importer(manager, Target::new(1, 1), Importer); + let begin = result(service.dispatch(Authorized::host( + 7, + ConfigCommand::Begin { + request_id: 1, + expected_generation: 0, + target: Target::new(1, 1), + }, + ))); + let rejected = result(service.dispatch(Authorized::host( + 7, + ConfigCommand::ImportBegin { + request_id: 2, + session_id: begin.session_id, + expected_revision: 0, + length: 1, + schema_id: [0; 16], + crc32c: crc32c(&[7]), + target: Target::new(1, 1), + }, + ))); + assert_eq!(rejected.error, Some(ProtocolError::SchemaMismatch)); + } +} diff --git a/libs/kaptan-config-mavlink/src/wire.rs b/libs/kaptan-config-mavlink/src/wire.rs new file mode 100644 index 0000000..614bc7a --- /dev/null +++ b/libs/kaptan-config-mavlink/src/wire.rs @@ -0,0 +1,501 @@ +//! Generated Kaptan-dialect message translation. + +use kaptan_config::{ + ApplyPolicy, ConfigState, ConfigValue, FieldDescriptor, FieldId, Generation, ValidationRule, + ValueKind, +}; +use kaptan_log::KaptanError; +use kaptan_mavlink::message::Message; +use kaptan_mavlink::message::kaptan::{ + KAPTAN_CONFIG_DESCRIPTOR_DATA, KAPTAN_CONFIG_INFO_DATA, KAPTAN_CONFIG_RESULT_DATA, + KAPTAN_CONFIG_VALUE_DATA, KaptanConfigError, KaptanConfigOperation, KaptanConfigPolicy, + KaptanConfigResult, KaptanConfigState, KaptanConfigValueKind, KaptanConfigValueSource, + MavMessage, +}; + +use crate::{ + ConfigCommand, ConfigReply, Operation, ProtocolError, ResultKind, Target, ValueSource, + service::{ConfigInfo, ValueResponse}, +}; + +/// Rejects a generated message that is not a Kaptan configuration command. +#[derive(Clone, Copy, Debug, Eq, PartialEq, KaptanError)] +pub enum DecodeError { + /// The message belongs to a response, telemetry, or another service. + UnsupportedMessage, + /// A wire field identity used a reserved sentinel. + InvalidFieldId, + /// The message used an unknown value-source discriminator. + InvalidValueSource, + /// A boolean setter used a representation other than zero or one. + InvalidBoolean, + /// A configuration request used a broadcast/zero target endpoint. + InvalidTarget, +} + +/// Converts one generated custom request into a transport-neutral command. +/// +/// This only decodes a validated MAVLink payload. An authenticated ingress must +/// still wrap the result in [`crate::Authorized`] before dispatch. +pub fn decode_command(message: &MavMessage) -> Result { + if matches!(message.target_system_id(), Some(0)) + || matches!(message.target_component_id(), Some(0)) + { + return Err(DecodeError::InvalidTarget); + } + match message { + MavMessage::KAPTAN_CONFIG_INFO_REQUEST(data) => Ok(ConfigCommand::InfoRequest { + request_id: data.request_id, + target: Target::new(data.target_system, data.target_component), + }), + MavMessage::KAPTAN_CONFIG_DESCRIPTOR_REQUEST(data) => { + Ok(ConfigCommand::DescriptorRequest { + request_id: data.request_id, + first: data.first, + count: data.count, + target: Target::new(data.target_system, data.target_component), + }) + } + MavMessage::KAPTAN_CONFIG_VALUE_REQUEST(data) => Ok(ConfigCommand::ValueRequest { + request_id: data.request_id, + session_id: data.session_id, + field: field_id(data.field_id)?, + source: value_source(data.source)?, + target: Target::new(data.target_system, data.target_component), + }), + MavMessage::KAPTAN_CONFIG_IMPORT_BEGIN(data) => Ok(ConfigCommand::ImportBegin { + request_id: data.request_id, + session_id: data.session_id, + expected_revision: data.expected_revision, + length: data.length, + schema_id: data.schema_id, + crc32c: data.crc32c, + target: Target::new(data.target_system, data.target_component), + }), + MavMessage::KAPTAN_CONFIG_IMPORT_CHUNK(data) => Ok(ConfigCommand::ImportChunk { + request_id: data.request_id, + session_id: data.session_id, + transfer_id: data.transfer_id, + offset: data.offset, + length: data.length, + data: data.data, + target: Target::new(data.target_system, data.target_component), + }), + MavMessage::KAPTAN_CONFIG_IMPORT_FINISH(data) => Ok(ConfigCommand::ImportFinish { + request_id: data.request_id, + session_id: data.session_id, + transfer_id: data.transfer_id, + expected_revision: data.expected_revision, + target: Target::new(data.target_system, data.target_component), + }), + MavMessage::KAPTAN_CONFIG_BEGIN(data) => Ok(ConfigCommand::Begin { + request_id: data.request_id, + expected_generation: data.expected_generation, + target: Target::new(data.target_system, data.target_component), + }), + MavMessage::KAPTAN_CONFIG_SET_F32(data) => Ok(ConfigCommand::Set { + request_id: data.request_id, + session_id: data.session_id, + expected_revision: data.expected_revision, + field: field_id(data.field_id)?, + value: ConfigValue::F32(data.value), + target: Target::new(data.target_system, data.target_component), + }), + MavMessage::KAPTAN_CONFIG_SET_BOOL(data) => Ok(ConfigCommand::Set { + request_id: data.request_id, + session_id: data.session_id, + expected_revision: data.expected_revision, + field: field_id(data.field_id)?, + value: ConfigValue::Bool(boolean(data.value)?), + target: Target::new(data.target_system, data.target_component), + }), + MavMessage::KAPTAN_CONFIG_SET_I32(data) => Ok(ConfigCommand::Set { + request_id: data.request_id, + session_id: data.session_id, + expected_revision: data.expected_revision, + field: field_id(data.field_id)?, + value: ConfigValue::I32(data.value), + target: Target::new(data.target_system, data.target_component), + }), + MavMessage::KAPTAN_CONFIG_SET_U32(data) => Ok(ConfigCommand::Set { + request_id: data.request_id, + session_id: data.session_id, + expected_revision: data.expected_revision, + field: field_id(data.field_id)?, + value: ConfigValue::U32(data.value), + target: Target::new(data.target_system, data.target_component), + }), + MavMessage::KAPTAN_CONFIG_SET_U64(data) => Ok(ConfigCommand::Set { + request_id: data.request_id, + session_id: data.session_id, + expected_revision: data.expected_revision, + field: field_id(data.field_id)?, + value: ConfigValue::U64(data.value), + target: Target::new(data.target_system, data.target_component), + }), + MavMessage::KAPTAN_CONFIG_SET_F64(data) => Ok(ConfigCommand::Set { + request_id: data.request_id, + session_id: data.session_id, + expected_revision: data.expected_revision, + field: field_id(data.field_id)?, + value: ConfigValue::F64(data.value), + target: Target::new(data.target_system, data.target_component), + }), + MavMessage::KAPTAN_CONFIG_VALIDATE(data) => Ok(ConfigCommand::Validate { + request_id: data.request_id, + session_id: data.session_id, + expected_revision: data.expected_revision, + target: Target::new(data.target_system, data.target_component), + }), + MavMessage::KAPTAN_CONFIG_COMMIT(data) => Ok(ConfigCommand::Commit { + request_id: data.request_id, + session_id: data.session_id, + expected_generation: data.expected_generation, + expected_revision: data.expected_revision, + target: Target::new(data.target_system, data.target_component), + }), + MavMessage::KAPTAN_CONFIG_ABORT(data) => Ok(ConfigCommand::Abort { + request_id: data.request_id, + session_id: data.session_id, + target: Target::new(data.target_system, data.target_component), + }), + _ => Err(DecodeError::UnsupportedMessage), + } +} + +const fn field_id(value: u16) -> Result { + match FieldId::try_new(value) { + Some(value) => Ok(value), + None => Err(DecodeError::InvalidFieldId), + } +} + +const fn value_source(value: KaptanConfigValueSource) -> Result { + match value { + KaptanConfigValueSource::KAPTAN_CONFIG_VALUE_SOURCE_ACTIVE => Ok(ValueSource::Active), + KaptanConfigValueSource::KAPTAN_CONFIG_VALUE_SOURCE_COMMITTED => Ok(ValueSource::Committed), + KaptanConfigValueSource::KAPTAN_CONFIG_VALUE_SOURCE_TRANSACTION => { + Ok(ValueSource::Transaction) + } + } +} + +const fn boolean(value: u8) -> Result { + match value { + 0 => Ok(false), + 1 => Ok(true), + _ => Err(DecodeError::InvalidBoolean), + } +} + +/// Encodes one authoritative service completion. +pub fn encode_result(reply: ConfigReply) -> KAPTAN_CONFIG_RESULT_DATA { + KAPTAN_CONFIG_RESULT_DATA { + generation: reply.generation, + request_id: reply.request_id, + session_id: reply.session_id, + transfer_id: reply.transfer_id, + revision: reply.revision, + primary_field: reply.primary_field(), + related_field: reply.related_field(), + validation_rule: reply.validation_rule(), + operation: operation(reply.operation), + result: result(reply.result), + error: error(reply.error), + config_state: state(reply.state), + } +} + +/// Encodes configuration protocol/schema status for a Kaptan control station. +pub fn encode_info(info: ConfigInfo) -> KAPTAN_CONFIG_INFO_DATA { + KAPTAN_CONFIG_INFO_DATA { + active_generation: info.active_generation.map_or(0, Generation::get), + committed_generation: info.committed_generation.map_or(0, Generation::get), + request_id: info.request_id, + schema_id: info.schema_id, + profile_format: info.profile_format, + field_count: info.field_count, + protocol_major: 1, + protocol_minor: 0, + config_state: state(info.state), + capability_flags: info.capability_flags, + } +} + +/// Encodes self-describing metadata for one schema field. +pub fn encode_descriptor( + request_id: u32, + descriptor_index: u16, + descriptor: FieldDescriptor, +) -> KAPTAN_CONFIG_DESCRIPTOR_DATA { + KAPTAN_CONFIG_DESCRIPTOR_DATA { + minimum_bits: descriptor.minimum.map_or(0, ConfigValue::raw_bits), + maximum_bits: descriptor.maximum.map_or(0, ConfigValue::raw_bits), + default_bits: descriptor.default.raw_bits(), + request_id, + descriptor_index, + field_id: descriptor.id.get(), + display_key: descriptor.display_key.into(), + unit: descriptor.unit.into(), + value_kind: value_kind(descriptor.kind), + apply_policy: policy(descriptor.apply_policy), + flags: 0, + } +} + +/// Encodes a typed scalar read response for the Kaptan control station. +pub fn encode_value(value: ValueResponse) -> KAPTAN_CONFIG_VALUE_DATA { + KAPTAN_CONFIG_VALUE_DATA { + value_bits: value.value.raw_bits(), + request_id: value.request_id, + session_id: value.session_id, + revision: value.revision, + field_id: value.field.get(), + value_kind: value_kind(value.value.kind()), + } +} + +const fn value_kind(kind: ValueKind) -> KaptanConfigValueKind { + match kind { + ValueKind::Bool => KaptanConfigValueKind::KAPTAN_CONFIG_VALUE_KIND_BOOL, + ValueKind::I32 => KaptanConfigValueKind::KAPTAN_CONFIG_VALUE_KIND_I32, + ValueKind::U32 => KaptanConfigValueKind::KAPTAN_CONFIG_VALUE_KIND_U32, + ValueKind::U64 => KaptanConfigValueKind::KAPTAN_CONFIG_VALUE_KIND_U64, + ValueKind::F32 => KaptanConfigValueKind::KAPTAN_CONFIG_VALUE_KIND_F32, + ValueKind::F64 => KaptanConfigValueKind::KAPTAN_CONFIG_VALUE_KIND_F64, + } +} + +const fn policy(policy: ApplyPolicy) -> KaptanConfigPolicy { + match policy { + ApplyPolicy::Live => KaptanConfigPolicy::KAPTAN_CONFIG_POLICY_LIVE, + ApplyPolicy::DisarmedReset => KaptanConfigPolicy::KAPTAN_CONFIG_POLICY_DISARMED_RESET, + ApplyPolicy::RestartRequired => KaptanConfigPolicy::KAPTAN_CONFIG_POLICY_RESTART_REQUIRED, + } +} + +const fn operation(operation: Operation) -> KaptanConfigOperation { + match operation { + Operation::Query => KaptanConfigOperation::KAPTAN_CONFIG_OPERATION_QUERY, + Operation::Begin => KaptanConfigOperation::KAPTAN_CONFIG_OPERATION_BEGIN, + Operation::Set => KaptanConfigOperation::KAPTAN_CONFIG_OPERATION_SET, + Operation::Validate => KaptanConfigOperation::KAPTAN_CONFIG_OPERATION_VALIDATE, + Operation::Commit => KaptanConfigOperation::KAPTAN_CONFIG_OPERATION_COMMIT, + Operation::Abort => KaptanConfigOperation::KAPTAN_CONFIG_OPERATION_ABORT, + Operation::Import => KaptanConfigOperation::KAPTAN_CONFIG_OPERATION_IMPORT, + } +} + +const fn result(result: ResultKind) -> KaptanConfigResult { + match result { + ResultKind::Accepted => KaptanConfigResult::KAPTAN_CONFIG_RESULT_ACCEPTED, + ResultKind::CommittedRestartRequired => { + KaptanConfigResult::KAPTAN_CONFIG_RESULT_COMMITTED_RESTART_REQUIRED + } + ResultKind::Rejected => KaptanConfigResult::KAPTAN_CONFIG_RESULT_REJECTED, + } +} + +const fn state(state: ConfigState) -> KaptanConfigState { + match state { + ConfigState::Active => KaptanConfigState::KAPTAN_CONFIG_STATE_ACTIVE, + ConfigState::Editing => KaptanConfigState::KAPTAN_CONFIG_STATE_EDITING, + ConfigState::RestartRequired => KaptanConfigState::KAPTAN_CONFIG_STATE_RESTART_REQUIRED, + } +} + +const fn error(error: Option) -> KaptanConfigError { + match error { + None => KaptanConfigError::KAPTAN_CONFIG_ERROR_NONE, + Some(ProtocolError::BadTarget) => KaptanConfigError::KAPTAN_CONFIG_ERROR_BAD_TARGET, + Some(ProtocolError::AuthorizationDenied) => { + KaptanConfigError::KAPTAN_CONFIG_ERROR_AUTHORIZATION_DENIED + } + Some(ProtocolError::GenerationConflict) => { + KaptanConfigError::KAPTAN_CONFIG_ERROR_GENERATION_CONFLICT + } + Some(ProtocolError::SessionBusy) => KaptanConfigError::KAPTAN_CONFIG_ERROR_SESSION_BUSY, + Some(ProtocolError::InvalidSession) => { + KaptanConfigError::KAPTAN_CONFIG_ERROR_INVALID_SESSION + } + Some(ProtocolError::RequestConflict) => { + KaptanConfigError::KAPTAN_CONFIG_ERROR_REQUEST_CONFLICT + } + Some(ProtocolError::StaleRevision) => KaptanConfigError::KAPTAN_CONFIG_ERROR_STALE_REVISION, + Some(ProtocolError::BadState) => KaptanConfigError::KAPTAN_CONFIG_ERROR_BAD_STATE, + Some(ProtocolError::UnknownField(_)) => { + KaptanConfigError::KAPTAN_CONFIG_ERROR_UNKNOWN_FIELD + } + Some(ProtocolError::TypeMismatch(_)) => { + KaptanConfigError::KAPTAN_CONFIG_ERROR_TYPE_MISMATCH + } + Some(ProtocolError::Violation(violation)) => match violation.rule() { + ValidationRule::NonFinite => KaptanConfigError::KAPTAN_CONFIG_ERROR_NON_FINITE, + ValidationRule::BelowMinimum | ValidationRule::Zero => { + KaptanConfigError::KAPTAN_CONFIG_ERROR_BELOW_MINIMUM + } + ValidationRule::AboveMaximum => KaptanConfigError::KAPTAN_CONFIG_ERROR_ABOVE_MAXIMUM, + ValidationRule::Inconsistent => KaptanConfigError::KAPTAN_CONFIG_ERROR_CROSS_FIELD, + ValidationRule::UnsupportedCapability => { + KaptanConfigError::KAPTAN_CONFIG_ERROR_UNSUPPORTED_CAPABILITY + } + ValidationRule::UnsupportedFormat => { + KaptanConfigError::KAPTAN_CONFIG_ERROR_UNSUPPORTED_FORMAT + } + }, + Some(ProtocolError::StorageFailure) => { + KaptanConfigError::KAPTAN_CONFIG_ERROR_STORAGE_FAILURE + } + Some(ProtocolError::CborMalformed) => KaptanConfigError::KAPTAN_CONFIG_ERROR_CBOR_MALFORMED, + Some(ProtocolError::CborTooLarge) => KaptanConfigError::KAPTAN_CONFIG_ERROR_CBOR_TOO_LARGE, + Some(ProtocolError::BlobCrc) => KaptanConfigError::KAPTAN_CONFIG_ERROR_BLOB_CRC, + Some(ProtocolError::SchemaMismatch) => { + KaptanConfigError::KAPTAN_CONFIG_ERROR_SCHEMA_MISMATCH + } + Some(ProtocolError::ImportUnavailable) => { + KaptanConfigError::KAPTAN_CONFIG_ERROR_IMPORT_UNAVAILABLE + } + Some(ProtocolError::Internal) => KaptanConfigError::KAPTAN_CONFIG_ERROR_INTERNAL, + } +} + +#[cfg(test)] +mod tests { + use kaptan_config::{ConfigState, ConfigViolation, ValidationRule}; + use kaptan_mavlink::message::kaptan::{ + KAPTAN_CONFIG_INFO_REQUEST_DATA, KAPTAN_CONFIG_SET_F32_DATA, MavMessage, + }; + + use super::*; + use crate::CAPABILITY_CBOR_IMPORT; + + #[test] + fn generated_set_message_preserves_typed_value_and_identifiers() { + let message = MavMessage::KAPTAN_CONFIG_SET_F32(KAPTAN_CONFIG_SET_F32_DATA { + request_id: 7, + session_id: 8, + expected_revision: 9, + field_id: 10, + value: 1.5, + target_system: 1, + target_component: 2, + }); + assert_eq!( + decode_command(&message), + Ok(ConfigCommand::Set { + request_id: 7, + session_id: 8, + expected_revision: 9, + field: FieldId::new(10), + value: ConfigValue::F32(1.5), + target: Target::new(1, 2), + }) + ); + } + + #[test] + fn generated_info_request_is_correlated_and_exactly_targeted() { + let message = MavMessage::KAPTAN_CONFIG_INFO_REQUEST(KAPTAN_CONFIG_INFO_REQUEST_DATA { + request_id: 31, + target_system: 1, + target_component: 2, + }); + assert_eq!( + decode_command(&message), + Ok(ConfigCommand::InfoRequest { + request_id: 31, + target: Target::new(1, 2), + }) + ); + } + + #[test] + fn generated_config_requests_reject_broadcast_targets() { + let zero_system = MavMessage::KAPTAN_CONFIG_SET_F32(KAPTAN_CONFIG_SET_F32_DATA { + request_id: 7, + session_id: 8, + expected_revision: 9, + field_id: 10, + value: 1.5, + target_system: 0, + target_component: 2, + }); + let zero_component = MavMessage::KAPTAN_CONFIG_SET_F32(KAPTAN_CONFIG_SET_F32_DATA { + request_id: 7, + session_id: 8, + expected_revision: 9, + field_id: 10, + value: 1.5, + target_system: 1, + target_component: 0, + }); + assert_eq!( + decode_command(&zero_system), + Err(DecodeError::InvalidTarget) + ); + assert_eq!( + decode_command(&zero_component), + Err(DecodeError::InvalidTarget) + ); + } + + #[test] + fn result_encodes_exact_validation_identity() { + let reply = ConfigReply { + request_id: 1, + session_id: 2, + transfer_id: 0, + revision: 3, + generation: 4, + operation: Operation::Set, + result: ResultKind::Rejected, + error: Some(ProtocolError::Violation(ConfigViolation::field( + FieldId::new(11), + ValidationRule::NonFinite, + ))), + state: ConfigState::Editing, + }; + let encoded = encode_result(reply); + assert_eq!(encoded.primary_field, 11); + assert_eq!(encoded.validation_rule, ValidationRule::NonFinite.code()); + assert_eq!( + encoded.error, + KaptanConfigError::KAPTAN_CONFIG_ERROR_NON_FINITE + ); + } + + #[test] + fn unavailable_import_has_its_own_structured_error_code() { + let reply = ConfigReply { + request_id: 1, + session_id: 0, + transfer_id: 0, + revision: 0, + generation: 0, + operation: Operation::Import, + result: ResultKind::Rejected, + error: Some(ProtocolError::ImportUnavailable), + state: ConfigState::Active, + }; + assert_eq!( + encode_result(reply).error, + KaptanConfigError::KAPTAN_CONFIG_ERROR_IMPORT_UNAVAILABLE + ); + } + + #[test] + fn info_preserves_service_capabilities() { + let info = encode_info(ConfigInfo { + request_id: 7, + active_generation: None, + committed_generation: None, + schema_id: [0; 16], + profile_format: 1, + field_count: 23, + state: ConfigState::Active, + capability_flags: CAPABILITY_CBOR_IMPORT, + }); + assert_eq!(info.request_id, 7); + assert_eq!(info.capability_flags, CAPABILITY_CBOR_IMPORT); + } +} diff --git a/libs/kaptan-mavlink/build.rs b/libs/kaptan-mavlink/build.rs index b57bc29..d6c314c 100644 --- a/libs/kaptan-mavlink/build.rs +++ b/libs/kaptan-mavlink/build.rs @@ -3,8 +3,9 @@ use std::path::PathBuf; fn main() { let manifest_directory = PathBuf::from(std::env::var("CARGO_MANIFEST_DIR").unwrap()); let dialect = manifest_directory.join("definitions/kaptan.xml"); + let config = manifest_directory.join("definitions/kaptan/config.xml"); let log = manifest_directory.join("definitions/kaptan/log.xml"); - for definition in [&dialect, &log] { + for definition in [&dialect, &config, &log] { println!("cargo::rerun-if-changed={}", definition.display()); } println!("cargo::rustc-check-cfg=cfg(kaptan_has_messages)"); diff --git a/libs/kaptan-mavlink/definitions/kaptan.xml b/libs/kaptan-mavlink/definitions/kaptan.xml index ad3bb02..71e12bf 100644 --- a/libs/kaptan-mavlink/definitions/kaptan.xml +++ b/libs/kaptan-mavlink/definitions/kaptan.xml @@ -2,5 +2,6 @@ + kaptan/config.xml kaptan/log.xml diff --git a/libs/kaptan-mavlink/definitions/kaptan/config.xml b/libs/kaptan-mavlink/definitions/kaptan/config.xml new file mode 100644 index 0000000..71bb8e8 --- /dev/null +++ b/libs/kaptan-mavlink/definitions/kaptan/config.xml @@ -0,0 +1,270 @@ + + + 3 + 0 + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + Correlated configuration protocol and active snapshot metadata. + + + + + + + + + + + + + Requests correlated configuration protocol and schema metadata. + + + + + + Requests a bounded page of field descriptors. + + + + + + + + One self-describing editable scalar field. + + + + + + + + + + + + + + Reads one active, committed, or transaction value. + + + + + + + + + One typed configuration scalar as canonical raw bits. + + + + + + + + + Opens the one lease-bound edit transaction. + + + + + + + Writes one f32 field in the open transaction. + + + + + + + + + + Writes one i32 field in the open transaction. + + + + + + + + + + Writes one u32 field in the open transaction. + + + + + + + + + + Writes one u64 field in the open transaction. + + + + + + + + + + Validates the complete open transaction. + + + + + + + + Validates then atomically persists the open transaction. + + + + + + + + + Discards the open RAM transaction. + + + + + + + Authoritative typed completion for a configuration request. + + + + + + + + + + + + + + + Writes one boolean field in the open transaction. + + + + + + + + + + Writes one f64 field in the open transaction. + + + + + + + + + + Starts a fixed-bounded CBOR import inside an open transaction. + + + + + + + + + + + One aligned, idempotent 192-byte-or-final CBOR import chunk. Chunks may arrive out of order. + + + + + + + + + + + Validates a complete received CBOR import into the transaction. + + + + + + + + +