From 0195a611d050e294049b6e2b700312d2d535de18 Mon Sep 17 00:00:00 2001 From: Ivan Malison Date: Sun, 19 Jul 2026 01:19:37 -0700 Subject: [PATCH 1/3] split: add a bounded application-message side channel MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add an opaque, bounded application payload that firmware can exchange between the split central and peripheral alongside the normal split traffic, without ever taking priority over key events. * `split_app` module: `SplitAppData` (a small, `MaxSize`, postcard length-prefixed payload) plus four statics — `SPLIT_APP_TX` (central -> peripheral), `SPLIT_APP_PERIPH_TX` (peripheral -> central), the symmetric `SPLIT_APP_RX` inbox, and the `SPLIT_APP_LINK` watch that reports split-link state to the application. * Producers use `try_send` only (bounded, drop-on-full) so the split read/write loops never block or get starved by application traffic; key events are always polled first. * The split driver and peripheral drain the application queues as the lowest-priority arm of their outgoing selects and forward received `SplitMessage::Application` payloads into the inbox. * `SPLIT_APP_LINK` is state-based (a `Watch`), so a late-subscribing application still observes the current link state; the `false -> true` edge is a resync trigger. Link-down edges are emitted from a drop guard so they survive async cancellation of the split session. * On the peripheral the link is raised on the FIRST inbound message from the central rather than on bare connection: over BLE, notifications to a central that has not yet subscribed are silently dropped, so the connection alone is not proof the application channel is usable. Developed for a split keyboard port. --- rmk/src/lib.rs | 3 + rmk/src/split/driver.rs | 34 ++++- rmk/src/split/mod.rs | 6 + rmk/src/split/peripheral.rs | 266 ++++++++++++++++++++++-------------- rmk/src/split_app.rs | 124 +++++++++++++++++ 5 files changed, 328 insertions(+), 105 deletions(-) create mode 100644 rmk/src/split_app.rs diff --git a/rmk/src/lib.rs b/rmk/src/lib.rs index 1be3ed35a..a69f69d8c 100644 --- a/rmk/src/lib.rs +++ b/rmk/src/lib.rs @@ -105,6 +105,9 @@ pub mod matrix; pub mod processor; #[cfg(feature = "split")] pub mod split; +// Bounded application-message hook for the split protocol. +#[cfg(feature = "split")] +pub mod split_app; pub mod state; #[cfg(feature = "storage")] pub mod storage; diff --git a/rmk/src/split/driver.rs b/rmk/src/split/driver.rs index 857574716..7e94e26a8 100644 --- a/rmk/src/split/driver.rs +++ b/rmk/src/split/driver.rs @@ -186,6 +186,24 @@ impl SplitMessage::Wpm(e.0), with_feature("display"): e = modifier_sub.next_event().fuse() => SplitMessage::Modifier(e.modifier.into_bits()), with_feature("display"): e = sleep_sub.next_event().fuse() => SplitMessage::SleepState(e.0), + // Application messages, deliberately the + // last (lowest-priority) outgoing arm; the read arm of the + // outer select still beats all outgoing traffic. + m = crate::split_app::SPLIT_APP_TX.receive().fuse() => SplitMessage::Application(m), } }; @@ -256,7 +278,7 @@ impl { if self.send(&msg).await.is_err() { - return; + return; // guard sends the link-down edge } } } @@ -285,6 +307,14 @@ impl publish_event(e), + // Forward peripheral → central application + // payloads into the (symmetric) inbox; drop-on-full so a slow + // consumer can never stall the split read loop. + SplitMessage::Application(data) => { + if crate::split_app::SPLIT_APP_RX.try_send(data).is_err() { + warn!("split app message dropped (inbox full)"); + } + } #[cfg(feature = "_ble")] SplitMessage::BatteryStatus(state) => set_peripheral_battery(self.id, state.0), #[cfg(feature = "dfu_split")] diff --git a/rmk/src/split/mod.rs b/rmk/src/split/mod.rs index de5ad13b6..2bd46c0c6 100644 --- a/rmk/src/split/mod.rs +++ b/rmk/src/split/mod.rs @@ -92,6 +92,12 @@ pub(crate) enum SplitMessage { /// Peripheral → Central: confirm mark_updated succeeded, about to reset. #[cfg(feature = "dfu_split")] FirmwareUpdateConfirm, + + /// opaque bounded application payload, central → + /// peripheral (see `crate::split_app`). Kept as the LAST variant so + /// the postcard discriminants of all existing messages stay stable across + /// halves flashed at different revisions. + Application(crate::split_app::SplitAppData), } // ----------------------------------------------------------------------- diff --git a/rmk/src/split/peripheral.rs b/rmk/src/split/peripheral.rs index f1a84104e..182b36954 100644 --- a/rmk/src/split/peripheral.rs +++ b/rmk/src/split/peripheral.rs @@ -78,6 +78,40 @@ impl SplitPeripheral { /// The peripheral uses the general matrix, does scanning and sends key events through `SplitWriter`. /// It also receives split messages from the central through `SplitReader`. pub(crate) async fn run(&mut self) { + // Expose the split-link state to the application. + // `run` executes exactly while a central session is up (for BLE it is + // invoked per connection and returns on disconnect). + // + // As on the central side (split/driver.rs), the link-down edge MUST + // come from a drop guard: this future can be cancelled from outside + // when the session dies, so no in-line `send(false)` is guaranteed to + // run. Without the false edge, reconnects look like true->true and + // link-up-triggered behavior (e.g. the version announcement) never + // re-arms. + // + // The link-UP edge is deliberately NOT sent here at session start: + // for BLE the connection is up, but the central has not yet + // subscribed to the peripheral's notify characteristic (CCCD write), + // and trouble-host silently drops notifications to an unsubscribed + // peer — anything the application sent in that window would vanish. + // Instead, link-up is declared on the FIRST message received from + // the central: the central's `PeripheralManager` sends its + // `ConnectionStatus` snapshot immediately after subscribing, and ATT + // bearer ordering guarantees the CCCD write was processed before + // that message, so peripheral → central traffic is deliverable from + // this point on. (Serial split has no subscription step and reaches + // the same first message; the edge is simply "the session is + // bidirectionally live".) + struct LinkDownGuard; + impl Drop for LinkDownGuard { + fn drop(&mut self) { + crate::split_app::SPLIT_APP_LINK.sender().send(false); + } + } + let _link_guard = LinkDownGuard; + let app_link = crate::split_app::SPLIT_APP_LINK.sender(); + let mut link_up_sent = false; + // Proactively announce our firmware hash so the central can detect // us even when it booted first and already gave up waiting for a query response. #[cfg(feature = "dfu_split")] @@ -108,127 +142,150 @@ impl SplitPeripheral { }, e = pointing_sub.next_message_pure().fuse() => SplitMessage::Pointing(e), with_feature("_ble"): e = battery_sub.next_event().fuse() => SplitMessage::BatteryStatus(e), + // Peripheral → central application + // messages, deliberately the last (lowest-priority) + // outgoing arm behind key events. + m = crate::split_app::SPLIT_APP_PERIPH_TX.receive().fuse() => SplitMessage::Application(m), } }; match select(self.split_driver.read(), read_message_to_send).await { Either::First(m) => match m { // Process split messages from the central - Ok(split_message) => match split_message { - SplitMessage::ConnectionStatus(status) => { - trace!("Received central connection status: {:?}", status); - update_status(|c| *c = status); - } - #[cfg(all(feature = "_ble", feature = "storage"))] - SplitMessage::ClearPeer => { - // Clear the peer address - FLASH_CHANNEL - .send(crate::storage::FlashOperationMessage::PeerAddress(PeerAddress::new( - 0, false, [0; 6], - ))) - .await; - } - SplitMessage::KeyboardIndicator(indicator) => { - // Publish KeyboardIndicator event - publish_event(LedIndicatorEvent::new( - rmk_types::led_indicator::LedIndicator::from_bits(indicator), - )); - } - SplitMessage::Layer(layer) => { - // Publish Layer event - publish_event(LayerChangeEvent::new(layer)); - } - #[cfg(feature = "display")] - SplitMessage::Wpm(wpm) => publish_event(WpmUpdateEvent::new(wpm)), - #[cfg(feature = "display")] - SplitMessage::Modifier(bits) => { - publish_event(ModifierEvent { - modifier: rmk_types::modifier::ModifierCombination::from_bits(bits), - }); - } - #[cfg(feature = "display")] - SplitMessage::SleepState(sleeping) => { - publish_event(SleepStateEvent::new(sleeping)); + Ok(split_message) => { + // First traffic from the central ⇒ the + // session is bidirectionally live (see the LinkDownGuard + // comment above for why link-up is not sent at start). + if !link_up_sent { + app_link.send(true); + link_up_sent = true; } - // --- dfu_split: firmware update handlers --- - #[cfg(feature = "dfu_split")] - SplitMessage::FirmwareHashQuery => { - let hash = crate::dfu::read_embedded_firmware_hash(); - info!("dfu_split: hash query, responding with {:#x}", hash); - self.split_driver - .write(&SplitMessage::FirmwareHashResponse(hash)) - .await - .ok(); - } - #[cfg(feature = "dfu_split")] - SplitMessage::FirmwareChunk { offset, len, data } => { - if self.dfu_handler.is_none() { - self.dfu_handler = crate::dfu::SplitDfuHandler::new(); + match split_message { + SplitMessage::ConnectionStatus(status) => { + trace!("Received central connection status: {:?}", status); + update_status(|c| *c = status); + } + #[cfg(all(feature = "_ble", feature = "storage"))] + SplitMessage::ClearPeer => { + // Clear the peer address + FLASH_CHANNEL + .send(crate::storage::FlashOperationMessage::PeerAddress(PeerAddress::new( + 0, false, [0; 6], + ))) + .await; + } + SplitMessage::KeyboardIndicator(indicator) => { + // Publish KeyboardIndicator event + publish_event(LedIndicatorEvent::new( + rmk_types::led_indicator::LedIndicator::from_bits(indicator), + )); + } + SplitMessage::Layer(layer) => { + // Publish Layer event + publish_event(LayerChangeEvent::new(layer)); + } + #[cfg(feature = "display")] + SplitMessage::Wpm(wpm) => publish_event(WpmUpdateEvent::new(wpm)), + #[cfg(feature = "display")] + SplitMessage::Modifier(bits) => { + publish_event(ModifierEvent { + modifier: rmk_types::modifier::ModifierCombination::from_bits(bits), + }); + } + #[cfg(feature = "display")] + SplitMessage::SleepState(sleeping) => { + publish_event(SleepStateEvent::new(sleeping)); + } + // --- dfu_split: firmware update handlers --- + #[cfg(feature = "dfu_split")] + SplitMessage::FirmwareHashQuery => { + let hash = crate::dfu::read_embedded_firmware_hash(); + info!("dfu_split: hash query, responding with {:#x}", hash); + self.split_driver + .write(&SplitMessage::FirmwareHashResponse(hash)) + .await + .ok(); + } + #[cfg(feature = "dfu_split")] + SplitMessage::FirmwareChunk { offset, len, data } => { if self.dfu_handler.is_none() { - error!("dfu_split: FlashManager not initialized, skipping chunk"); - continue; + self.dfu_handler = crate::dfu::SplitDfuHandler::new(); + if self.dfu_handler.is_none() { + error!("dfu_split: FlashManager not initialized, skipping chunk"); + continue; + } } - } - let handler = self.dfu_handler.as_mut().unwrap(); - let actual_len = len as usize; - let chunk_data = &data.0[..actual_len]; - match handler.write_chunk(offset as u32, chunk_data) { - Ok(()) => { - debug!("dfu_split: wrote {} bytes at offset {}", actual_len, offset); - let ack = SplitMessage::FirmwareChunkAck { - offset, - crc: crate::crc32::crc32(chunk_data), - }; - self.split_driver.write(&ack).await.ok(); + let handler = self.dfu_handler.as_mut().unwrap(); + let actual_len = len as usize; + let chunk_data = &data.0[..actual_len]; + match handler.write_chunk(offset as u32, chunk_data) { + Ok(()) => { + debug!("dfu_split: wrote {} bytes at offset {}", actual_len, offset); + let ack = SplitMessage::FirmwareChunkAck { + offset, + crc: crate::crc32::crc32(chunk_data), + }; + self.split_driver.write(&ack).await.ok(); + } + Err(()) => error!("dfu_split: write error at offset {}", offset), } - Err(()) => error!("dfu_split: write error at offset {}", offset), } - } - #[cfg(feature = "dfu_split")] - SplitMessage::FirmwareUpdateComplete => { - if let Some(ref mut handler) = self.dfu_handler { - let dfu_crc = handler.compute_dfu_crc(); - info!("dfu_split: DFU partition CRC: {:#010x}", dfu_crc); - let crc_msg = SplitMessage::FirmwareCrcReport(dfu_crc); - self.split_driver.write(&crc_msg).await.ok(); - info!("dfu_split: CRC report sent"); + #[cfg(feature = "dfu_split")] + SplitMessage::FirmwareUpdateComplete => { + if let Some(ref mut handler) = self.dfu_handler { + let dfu_crc = handler.compute_dfu_crc(); + info!("dfu_split: DFU partition CRC: {:#010x}", dfu_crc); + let crc_msg = SplitMessage::FirmwareCrcReport(dfu_crc); + self.split_driver.write(&crc_msg).await.ok(); + info!("dfu_split: CRC report sent"); - let deadline = embassy_time::Instant::now() + embassy_time::Duration::from_secs(5); - let ok = loop { - match select(self.split_driver.read(), embassy_time::Timer::at(deadline)).await { - Either::First(Ok(SplitMessage::FirmwareCrcOk)) => { - info!("dfu_split: central confirmed CRC, resetting"); - break true; - } - Either::First(Ok(SplitMessage::FirmwareCrcFail)) => { - warn!("dfu_split: central rejected CRC, stopping update"); - break false; - } - Either::First(Ok(_)) => {} - Either::First(Err(e)) => { - error!("read error: {:?}", e); - break false; - } - Either::Second(_) => { - error!("timeout"); - break false; + let deadline = embassy_time::Instant::now() + embassy_time::Duration::from_secs(5); + let ok = loop { + match select(self.split_driver.read(), embassy_time::Timer::at(deadline)).await + { + Either::First(Ok(SplitMessage::FirmwareCrcOk)) => { + info!("dfu_split: central confirmed CRC, resetting"); + break true; + } + Either::First(Ok(SplitMessage::FirmwareCrcFail)) => { + warn!("dfu_split: central rejected CRC, stopping update"); + break false; + } + Either::First(Ok(_)) => {} + Either::First(Err(e)) => { + error!("read error: {:?}", e); + break false; + } + Either::Second(_) => { + error!("timeout"); + break false; + } } - } - }; + }; - if ok { - self.split_driver.write(&SplitMessage::FirmwareUpdateConfirm).await.ok(); - embassy_time::Timer::after_millis(50).await; - handler.mark_updated_and_reset().ok(); + if ok { + self.split_driver.write(&SplitMessage::FirmwareUpdateConfirm).await.ok(); + embassy_time::Timer::after_millis(50).await; + handler.mark_updated_and_reset().ok(); + } else { + self.dfu_handler = None; + } } else { - self.dfu_handler = None; + error!("dfu_split: no active DFU session"); + } + } + // Forward application payloads; + // drop-on-full so a slow consumer can never stall the + // split read loop (the application resyncs on + // reconnect and must tolerate loss). + SplitMessage::Application(data) => { + if crate::split_app::SPLIT_APP_RX.try_send(data).is_err() { + warn!("split app message dropped (inbox full)"); } - } else { - error!("dfu_split: no active DFU session"); } + _ => (), } - _ => (), - }, + } Err(e) => { error!("Split message read error: {:?}", e); if let crate::split::driver::SplitDriverError::Disconnected = e { @@ -242,5 +299,8 @@ impl SplitPeripheral { } } } + + // The loop only exits on disconnect. + app_link.send(false); } } diff --git a/rmk/src/split_app.rs b/rmk/src/split_app.rs new file mode 100644 index 000000000..7dc37f3dd --- /dev/null +++ b/rmk/src/split_app.rs @@ -0,0 +1,124 @@ +//! Bounded application-message hook for the split protocol. +//! +//! A small, opaque, bounded payload that an application can send between the +//! split central and its peripheral alongside — but never in front of — the +//! normal split traffic. RMK itself attaches no meaning to the bytes; a +//! firmware built on RMK can use it for application-level side-band state +//! (for example, propagating lighting or UI state across the split link). +//! +//! Flow: +//! +//! - Central: the application queues [`SplitAppData`] into [`SPLIT_APP_TX`] +//! (bounded, `try_send` only — the queue must never be awaited full). +//! `PeripheralManager` drains it as one more (lowest-priority) arm of its +//! outgoing-message select and wraps each payload in +//! `SplitMessage::Application`. Peripheral→central key events always win: +//! they arrive on the read arm, which is polled first. +//! - Peripheral: `SplitPeripheral` forwards received `Application` messages +//! into [`SPLIT_APP_RX`] with `try_send` (drop-on-full keeps the split read +//! loop responsive; the application is expected to tolerate loss, e.g. by +//! resyncing on reconnect). +//! - Peripheral → central: the peripheral application queues +//! [`SplitAppData`] into [`SPLIT_APP_PERIPH_TX`] (bounded, `try_send` +//! only); `SplitPeripheral` drains it as one more outgoing arm of its +//! select, behind key events. The central's `PeripheralManager` forwards +//! received `Application` messages into [`SPLIT_APP_RX`] the same way the +//! peripheral does — the inbox is symmetric ("this side's received +//! application messages"), only the senders differ. +//! - Both sides: [`SPLIT_APP_LINK`] carries the split-link state (central: +//! "peripheral link up"; peripheral: "central link up"), set by the split +//! driver. The central raises it at session start; the peripheral raises +//! it on the FIRST message received from the central — for BLE the bare +//! connection is not enough, since notifications to a central that has +//! not yet subscribed are silently dropped (see `split/peripheral.rs`). +//! Both lower it at session end. Applications use the `false → true` edge +//! to trigger an idempotent resync. +//! +//! Note: the queues assume a single split peripheral. Extending this hook to +//! multiple peripherals would key them by peripheral id. + +use embassy_sync::channel::Channel; +use embassy_sync::watch::Watch; +use postcard::experimental::max_size::MaxSize; +use serde::{Deserialize, Deserializer, Serialize, Serializer}; + +use crate::RawMutex; + +/// Maximum payload of one application split message. Deliberately small: +/// every split BLE transfer is `SPLIT_MESSAGE_MAX_SIZE` bytes on the wire, +/// so this bound also taxes key-event messages. Hard cap: the trouble +/// `gatt_service` macro initializes its characteristic arrays via +/// `Default`, which arrays only implement up to 32 elements — so +/// `SPLIT_MESSAGE_MAX_SIZE` (this + 1-byte length prefix + 1-byte enum +/// discriminant + 4 bytes margin) must stay ≤ 32. +pub const SPLIT_APP_MSG_MAX: usize = 26; + +/// One opaque application payload. Only `data[..len]` is meaningful. +#[derive(Debug, Clone, Copy)] +#[cfg_attr(feature = "defmt", derive(defmt::Format))] +pub struct SplitAppData { + pub len: u8, + pub data: [u8; SPLIT_APP_MSG_MAX], +} + +impl SplitAppData { + /// Wrap `payload`; `None` if it exceeds [`SPLIT_APP_MSG_MAX`]. + pub fn new(payload: &[u8]) -> Option { + if payload.len() > SPLIT_APP_MSG_MAX { + return None; + } + let mut data = [0u8; SPLIT_APP_MSG_MAX]; + data[..payload.len()].copy_from_slice(payload); + Some(Self { + len: payload.len() as u8, + data, + }) + } + + pub fn payload(&self) -> &[u8] { + &self.data[..(self.len as usize).min(SPLIT_APP_MSG_MAX)] + } +} + +// Postcard stores the payload as `&[u8]` (varint length prefix + bytes), so +// only the used bytes travel inside the (fixed-size) split transfer buffer. +impl Serialize for SplitAppData { + fn serialize(&self, serializer: S) -> Result { + self.payload().serialize(serializer) + } +} + +impl<'de> Deserialize<'de> for SplitAppData { + fn deserialize>(deserializer: D) -> Result { + use serde::de::Error; + let buf: &[u8] = Deserialize::deserialize(deserializer)?; + SplitAppData::new(buf).ok_or_else(|| D::Error::custom("split app message too long")) + } +} + +impl MaxSize for SplitAppData { + // 1-byte varint length prefix + payload. + const POSTCARD_MAX_SIZE: usize = SPLIT_APP_MSG_MAX + 1; +} + +/// Central → peripheral queue, drained by `PeripheralManager` while the link +/// is up. Producers MUST use `try_send` (bounded, never block); capacity is +/// sized so one full application resync burst fits with headroom. +pub static SPLIT_APP_TX: Channel = Channel::new(); + +/// This side's inbox of received application messages (peripheral: from the +/// central's `SPLIT_APP_TX`; central: from the peripheral's +/// `SPLIT_APP_PERIPH_TX`). Filled with `try_send` (drop-on-full) by the +/// split read loops. +pub static SPLIT_APP_RX: Channel = Channel::new(); + +/// Peripheral → central queue, drained by `SplitPeripheral` while the link +/// is up. Producers MUST use `try_send`. Small: the application announces +/// tiny, rare state (e.g. its build identity once per link-up). +pub static SPLIT_APP_PERIPH_TX: Channel = Channel::new(); + +/// Split-link state for the application: on the central, "peripheral link +/// up"; on the peripheral, "central link up". Written by the split driver at +/// session start/end; state-based (a late receiver still observes the latest +/// value), so edges cannot be lost the way pub/sub events can. +pub static SPLIT_APP_LINK: Watch = Watch::new(); From 3bdbd253b1f40ab6b8e12d496d2e08bed234401d Mon Sep 17 00:00:00 2001 From: Ivan Malison Date: Wed, 5 Aug 2026 01:31:50 -0700 Subject: [PATCH 2/3] fix(split): satisfy collapsible match lint --- rmk/src/split/peripheral.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/rmk/src/split/peripheral.rs b/rmk/src/split/peripheral.rs index 182b36954..a02f581c6 100644 --- a/rmk/src/split/peripheral.rs +++ b/rmk/src/split/peripheral.rs @@ -279,7 +279,8 @@ impl SplitPeripheral { // split read loop (the application resyncs on // reconnect and must tolerate loss). SplitMessage::Application(data) => { - if crate::split_app::SPLIT_APP_RX.try_send(data).is_err() { + let queued = crate::split_app::SPLIT_APP_RX.try_send(data); + if queued.is_err() { warn!("split app message dropped (inbox full)"); } } From c4b4cc0ba1049874402d1a0ca1f4796e81b713fd Mon Sep 17 00:00:00 2001 From: Ivan Malison Date: Wed, 5 Aug 2026 02:16:51 -0700 Subject: [PATCH 3/3] refactor(split): dedupe link guard and flatten peripheral handling Move the duplicated inline LinkDownGuard structs into split_app as a single LinkGuard that also owns the idempotent link-up edge, extract the peripheral's central-message match into handle_central_message so it is no longer nested five levels deep inside run(), share the drop-on-full inbox delivery between both sides, and trim comments down to the non-obvious facts. Co-Authored-By: Claude Fable 5 --- rmk/src/lib.rs | 1 - rmk/src/split/driver.rs | 39 ++--- rmk/src/split/mod.rs | 7 +- rmk/src/split/peripheral.rs | 303 ++++++++++++++++-------------------- rmk/src/split_app.rs | 117 +++++++------- 5 files changed, 206 insertions(+), 261 deletions(-) diff --git a/rmk/src/lib.rs b/rmk/src/lib.rs index a69f69d8c..2c48d76ff 100644 --- a/rmk/src/lib.rs +++ b/rmk/src/lib.rs @@ -105,7 +105,6 @@ pub mod matrix; pub mod processor; #[cfg(feature = "split")] pub mod split; -// Bounded application-message hook for the split protocol. #[cfg(feature = "split")] pub mod split_app; pub mod state; diff --git a/rmk/src/split/driver.rs b/rmk/src/split/driver.rs index 7e94e26a8..10ef5e8f7 100644 --- a/rmk/src/split/driver.rs +++ b/rmk/src/split/driver.rs @@ -186,23 +186,11 @@ impl SplitMessage::Wpm(e.0), with_feature("display"): e = modifier_sub.next_event().fuse() => SplitMessage::Modifier(e.modifier.into_bits()), with_feature("display"): e = sleep_sub.next_event().fuse() => SplitMessage::SleepState(e.0), - // Application messages, deliberately the - // last (lowest-priority) outgoing arm; the read arm of the - // outer select still beats all outgoing traffic. + // Deliberately the last (lowest-priority) outgoing arm. m = crate::split_app::SPLIT_APP_TX.receive().fuse() => SplitMessage::Application(m), } }; @@ -278,7 +264,7 @@ impl { if self.send(&msg).await.is_err() { - return; // guard sends the link-down edge + return; } } } @@ -307,14 +293,7 @@ impl publish_event(e), - // Forward peripheral → central application - // payloads into the (symmetric) inbox; drop-on-full so a slow - // consumer can never stall the split read loop. - SplitMessage::Application(data) => { - if crate::split_app::SPLIT_APP_RX.try_send(data).is_err() { - warn!("split app message dropped (inbox full)"); - } - } + SplitMessage::Application(data) => crate::split_app::deliver_received(data), #[cfg(feature = "_ble")] SplitMessage::BatteryStatus(state) => set_peripheral_battery(self.id, state.0), #[cfg(feature = "dfu_split")] diff --git a/rmk/src/split/mod.rs b/rmk/src/split/mod.rs index 2bd46c0c6..d057676b2 100644 --- a/rmk/src/split/mod.rs +++ b/rmk/src/split/mod.rs @@ -93,10 +93,9 @@ pub(crate) enum SplitMessage { #[cfg(feature = "dfu_split")] FirmwareUpdateConfirm, - /// opaque bounded application payload, central → - /// peripheral (see `crate::split_app`). Kept as the LAST variant so - /// the postcard discriminants of all existing messages stay stable across - /// halves flashed at different revisions. + /// Opaque application payload, either direction (see `crate::split_app`). + /// Kept as the last variant so the postcard discriminants of existing + /// messages stay stable across halves flashed at different revisions. Application(crate::split_app::SplitAppData), } diff --git a/rmk/src/split/peripheral.rs b/rmk/src/split/peripheral.rs index a02f581c6..0809af523 100644 --- a/rmk/src/split/peripheral.rs +++ b/rmk/src/split/peripheral.rs @@ -78,39 +78,14 @@ impl SplitPeripheral { /// The peripheral uses the general matrix, does scanning and sends key events through `SplitWriter`. /// It also receives split messages from the central through `SplitReader`. pub(crate) async fn run(&mut self) { - // Expose the split-link state to the application. - // `run` executes exactly while a central session is up (for BLE it is - // invoked per connection and returns on disconnect). - // - // As on the central side (split/driver.rs), the link-down edge MUST - // come from a drop guard: this future can be cancelled from outside - // when the session dies, so no in-line `send(false)` is guaranteed to - // run. Without the false edge, reconnects look like true->true and - // link-up-triggered behavior (e.g. the version announcement) never - // re-arms. - // - // The link-UP edge is deliberately NOT sent here at session start: - // for BLE the connection is up, but the central has not yet - // subscribed to the peripheral's notify characteristic (CCCD write), - // and trouble-host silently drops notifications to an unsubscribed - // peer — anything the application sent in that window would vanish. - // Instead, link-up is declared on the FIRST message received from - // the central: the central's `PeripheralManager` sends its - // `ConnectionStatus` snapshot immediately after subscribing, and ATT - // bearer ordering guarantees the CCCD write was processed before - // that message, so peripheral → central traffic is deliverable from - // this point on. (Serial split has no subscription step and reaches - // the same first message; the edge is simply "the session is - // bidirectionally live".) - struct LinkDownGuard; - impl Drop for LinkDownGuard { - fn drop(&mut self) { - crate::split_app::SPLIT_APP_LINK.sender().send(false); - } - } - let _link_guard = LinkDownGuard; - let app_link = crate::split_app::SPLIT_APP_LINK.sender(); - let mut link_up_sent = false; + // `run` executes exactly while a central session is up; the guard's + // `Drop` sends the link-down edge even when this future is cancelled + // from outside. Link-up waits for the first message from the central: + // a bare BLE connection is not enough, since trouble-host silently + // drops notifications until the central subscribes (CCCD write), and + // the central's `ConnectionStatus` snapshot arrives right after it + // subscribes — so first message ⇒ bidirectionally live. + let mut app_link = crate::split_app::LinkGuard::new(); // Proactively announce our firmware hash so the central can detect // us even when it booted first and already gave up waiting for a query response. @@ -142,9 +117,7 @@ impl SplitPeripheral { }, e = pointing_sub.next_message_pure().fuse() => SplitMessage::Pointing(e), with_feature("_ble"): e = battery_sub.next_event().fuse() => SplitMessage::BatteryStatus(e), - // Peripheral → central application - // messages, deliberately the last (lowest-priority) - // outgoing arm behind key events. + // Deliberately the last (lowest-priority) outgoing arm, behind key events. m = crate::split_app::SPLIT_APP_PERIPH_TX.receive().fuse() => SplitMessage::Application(m), } }; @@ -153,139 +126,10 @@ impl SplitPeripheral { Either::First(m) => match m { // Process split messages from the central Ok(split_message) => { - // First traffic from the central ⇒ the - // session is bidirectionally live (see the LinkDownGuard - // comment above for why link-up is not sent at start). - if !link_up_sent { - app_link.send(true); - link_up_sent = true; - } - match split_message { - SplitMessage::ConnectionStatus(status) => { - trace!("Received central connection status: {:?}", status); - update_status(|c| *c = status); - } - #[cfg(all(feature = "_ble", feature = "storage"))] - SplitMessage::ClearPeer => { - // Clear the peer address - FLASH_CHANNEL - .send(crate::storage::FlashOperationMessage::PeerAddress(PeerAddress::new( - 0, false, [0; 6], - ))) - .await; - } - SplitMessage::KeyboardIndicator(indicator) => { - // Publish KeyboardIndicator event - publish_event(LedIndicatorEvent::new( - rmk_types::led_indicator::LedIndicator::from_bits(indicator), - )); - } - SplitMessage::Layer(layer) => { - // Publish Layer event - publish_event(LayerChangeEvent::new(layer)); - } - #[cfg(feature = "display")] - SplitMessage::Wpm(wpm) => publish_event(WpmUpdateEvent::new(wpm)), - #[cfg(feature = "display")] - SplitMessage::Modifier(bits) => { - publish_event(ModifierEvent { - modifier: rmk_types::modifier::ModifierCombination::from_bits(bits), - }); - } - #[cfg(feature = "display")] - SplitMessage::SleepState(sleeping) => { - publish_event(SleepStateEvent::new(sleeping)); - } - // --- dfu_split: firmware update handlers --- - #[cfg(feature = "dfu_split")] - SplitMessage::FirmwareHashQuery => { - let hash = crate::dfu::read_embedded_firmware_hash(); - info!("dfu_split: hash query, responding with {:#x}", hash); - self.split_driver - .write(&SplitMessage::FirmwareHashResponse(hash)) - .await - .ok(); - } - #[cfg(feature = "dfu_split")] - SplitMessage::FirmwareChunk { offset, len, data } => { - if self.dfu_handler.is_none() { - self.dfu_handler = crate::dfu::SplitDfuHandler::new(); - if self.dfu_handler.is_none() { - error!("dfu_split: FlashManager not initialized, skipping chunk"); - continue; - } - } - let handler = self.dfu_handler.as_mut().unwrap(); - let actual_len = len as usize; - let chunk_data = &data.0[..actual_len]; - match handler.write_chunk(offset as u32, chunk_data) { - Ok(()) => { - debug!("dfu_split: wrote {} bytes at offset {}", actual_len, offset); - let ack = SplitMessage::FirmwareChunkAck { - offset, - crc: crate::crc32::crc32(chunk_data), - }; - self.split_driver.write(&ack).await.ok(); - } - Err(()) => error!("dfu_split: write error at offset {}", offset), - } - } - #[cfg(feature = "dfu_split")] - SplitMessage::FirmwareUpdateComplete => { - if let Some(ref mut handler) = self.dfu_handler { - let dfu_crc = handler.compute_dfu_crc(); - info!("dfu_split: DFU partition CRC: {:#010x}", dfu_crc); - let crc_msg = SplitMessage::FirmwareCrcReport(dfu_crc); - self.split_driver.write(&crc_msg).await.ok(); - info!("dfu_split: CRC report sent"); - - let deadline = embassy_time::Instant::now() + embassy_time::Duration::from_secs(5); - let ok = loop { - match select(self.split_driver.read(), embassy_time::Timer::at(deadline)).await - { - Either::First(Ok(SplitMessage::FirmwareCrcOk)) => { - info!("dfu_split: central confirmed CRC, resetting"); - break true; - } - Either::First(Ok(SplitMessage::FirmwareCrcFail)) => { - warn!("dfu_split: central rejected CRC, stopping update"); - break false; - } - Either::First(Ok(_)) => {} - Either::First(Err(e)) => { - error!("read error: {:?}", e); - break false; - } - Either::Second(_) => { - error!("timeout"); - break false; - } - } - }; - - if ok { - self.split_driver.write(&SplitMessage::FirmwareUpdateConfirm).await.ok(); - embassy_time::Timer::after_millis(50).await; - handler.mark_updated_and_reset().ok(); - } else { - self.dfu_handler = None; - } - } else { - error!("dfu_split: no active DFU session"); - } - } - // Forward application payloads; - // drop-on-full so a slow consumer can never stall the - // split read loop (the application resyncs on - // reconnect and must tolerate loss). - SplitMessage::Application(data) => { - let queued = crate::split_app::SPLIT_APP_RX.try_send(data); - if queued.is_err() { - warn!("split app message dropped (inbox full)"); - } - } - _ => (), - } + // First traffic from the central ⇒ the session is + // bidirectionally live. + app_link.mark_up(); + self.handle_central_message(split_message).await; } Err(e) => { error!("Split message read error: {:?}", e); @@ -300,8 +144,125 @@ impl SplitPeripheral { } } } + } - // The loop only exits on disconnect. - app_link.send(false); + /// Process a single message from the central. + async fn handle_central_message(&mut self, split_message: SplitMessage) { + match split_message { + SplitMessage::ConnectionStatus(status) => { + trace!("Received central connection status: {:?}", status); + update_status(|c| *c = status); + } + #[cfg(all(feature = "_ble", feature = "storage"))] + SplitMessage::ClearPeer => { + // Clear the peer address + FLASH_CHANNEL + .send(crate::storage::FlashOperationMessage::PeerAddress(PeerAddress::new( + 0, false, [0; 6], + ))) + .await; + } + SplitMessage::KeyboardIndicator(indicator) => { + // Publish KeyboardIndicator event + publish_event(LedIndicatorEvent::new( + rmk_types::led_indicator::LedIndicator::from_bits(indicator), + )); + } + SplitMessage::Layer(layer) => { + // Publish Layer event + publish_event(LayerChangeEvent::new(layer)); + } + #[cfg(feature = "display")] + SplitMessage::Wpm(wpm) => publish_event(WpmUpdateEvent::new(wpm)), + #[cfg(feature = "display")] + SplitMessage::Modifier(bits) => { + publish_event(ModifierEvent { + modifier: rmk_types::modifier::ModifierCombination::from_bits(bits), + }); + } + #[cfg(feature = "display")] + SplitMessage::SleepState(sleeping) => { + publish_event(SleepStateEvent::new(sleeping)); + } + // --- dfu_split: firmware update handlers --- + #[cfg(feature = "dfu_split")] + SplitMessage::FirmwareHashQuery => { + let hash = crate::dfu::read_embedded_firmware_hash(); + info!("dfu_split: hash query, responding with {:#x}", hash); + self.split_driver + .write(&SplitMessage::FirmwareHashResponse(hash)) + .await + .ok(); + } + #[cfg(feature = "dfu_split")] + SplitMessage::FirmwareChunk { offset, len, data } => { + if self.dfu_handler.is_none() { + self.dfu_handler = crate::dfu::SplitDfuHandler::new(); + if self.dfu_handler.is_none() { + error!("dfu_split: FlashManager not initialized, skipping chunk"); + return; + } + } + let handler = self.dfu_handler.as_mut().unwrap(); + let actual_len = len as usize; + let chunk_data = &data.0[..actual_len]; + match handler.write_chunk(offset as u32, chunk_data) { + Ok(()) => { + debug!("dfu_split: wrote {} bytes at offset {}", actual_len, offset); + let ack = SplitMessage::FirmwareChunkAck { + offset, + crc: crate::crc32::crc32(chunk_data), + }; + self.split_driver.write(&ack).await.ok(); + } + Err(()) => error!("dfu_split: write error at offset {}", offset), + } + } + #[cfg(feature = "dfu_split")] + SplitMessage::FirmwareUpdateComplete => { + if let Some(ref mut handler) = self.dfu_handler { + let dfu_crc = handler.compute_dfu_crc(); + info!("dfu_split: DFU partition CRC: {:#010x}", dfu_crc); + let crc_msg = SplitMessage::FirmwareCrcReport(dfu_crc); + self.split_driver.write(&crc_msg).await.ok(); + info!("dfu_split: CRC report sent"); + + let deadline = embassy_time::Instant::now() + embassy_time::Duration::from_secs(5); + let ok = loop { + match select(self.split_driver.read(), embassy_time::Timer::at(deadline)).await { + Either::First(Ok(SplitMessage::FirmwareCrcOk)) => { + info!("dfu_split: central confirmed CRC, resetting"); + break true; + } + Either::First(Ok(SplitMessage::FirmwareCrcFail)) => { + warn!("dfu_split: central rejected CRC, stopping update"); + break false; + } + Either::First(Ok(_)) => {} + Either::First(Err(e)) => { + error!("read error: {:?}", e); + break false; + } + Either::Second(_) => { + error!("timeout"); + break false; + } + } + }; + + if ok { + self.split_driver.write(&SplitMessage::FirmwareUpdateConfirm).await.ok(); + embassy_time::Timer::after_millis(50).await; + handler.mark_updated_and_reset().ok(); + } else { + self.dfu_handler = None; + } + } else { + error!("dfu_split: no active DFU session"); + } + } + SplitMessage::Application(data) => crate::split_app::deliver_received(data), + _ => (), + } } } diff --git a/rmk/src/split_app.rs b/rmk/src/split_app.rs index 7dc37f3dd..b69271083 100644 --- a/rmk/src/split_app.rs +++ b/rmk/src/split_app.rs @@ -1,41 +1,18 @@ //! Bounded application-message hook for the split protocol. //! -//! A small, opaque, bounded payload that an application can send between the -//! split central and its peripheral alongside — but never in front of — the -//! normal split traffic. RMK itself attaches no meaning to the bytes; a -//! firmware built on RMK can use it for application-level side-band state -//! (for example, propagating lighting or UI state across the split link). +//! An application built on RMK can exchange small opaque payloads +//! ([`SplitAppData`]) between the split halves alongside — but never in front +//! of — normal split traffic. RMK attaches no meaning to the bytes. //! -//! Flow: +//! Each side queues outgoing payloads with `try_send` ([`SPLIT_APP_TX`] on +//! the central, [`SPLIT_APP_PERIPH_TX`] on the peripheral). The split loops +//! drain them as their lowest-priority outgoing arm and deliver received +//! payloads into [`SPLIT_APP_RX`], dropping on full so application traffic +//! can never delay key events. [`SPLIT_APP_LINK`] carries the link state; its +//! `false → true` edge is the application's cue to resync, which also covers +//! messages lost to a full queue. //! -//! - Central: the application queues [`SplitAppData`] into [`SPLIT_APP_TX`] -//! (bounded, `try_send` only — the queue must never be awaited full). -//! `PeripheralManager` drains it as one more (lowest-priority) arm of its -//! outgoing-message select and wraps each payload in -//! `SplitMessage::Application`. Peripheral→central key events always win: -//! they arrive on the read arm, which is polled first. -//! - Peripheral: `SplitPeripheral` forwards received `Application` messages -//! into [`SPLIT_APP_RX`] with `try_send` (drop-on-full keeps the split read -//! loop responsive; the application is expected to tolerate loss, e.g. by -//! resyncing on reconnect). -//! - Peripheral → central: the peripheral application queues -//! [`SplitAppData`] into [`SPLIT_APP_PERIPH_TX`] (bounded, `try_send` -//! only); `SplitPeripheral` drains it as one more outgoing arm of its -//! select, behind key events. The central's `PeripheralManager` forwards -//! received `Application` messages into [`SPLIT_APP_RX`] the same way the -//! peripheral does — the inbox is symmetric ("this side's received -//! application messages"), only the senders differ. -//! - Both sides: [`SPLIT_APP_LINK`] carries the split-link state (central: -//! "peripheral link up"; peripheral: "central link up"), set by the split -//! driver. The central raises it at session start; the peripheral raises -//! it on the FIRST message received from the central — for BLE the bare -//! connection is not enough, since notifications to a central that has -//! not yet subscribed are silently dropped (see `split/peripheral.rs`). -//! Both lower it at session end. Applications use the `false → true` edge -//! to trigger an idempotent resync. -//! -//! Note: the queues assume a single split peripheral. Extending this hook to -//! multiple peripherals would key them by peripheral id. +//! The queues assume a single split peripheral. use embassy_sync::channel::Channel; use embassy_sync::watch::Watch; @@ -44,13 +21,11 @@ use serde::{Deserialize, Deserializer, Serialize, Serializer}; use crate::RawMutex; -/// Maximum payload of one application split message. Deliberately small: -/// every split BLE transfer is `SPLIT_MESSAGE_MAX_SIZE` bytes on the wire, -/// so this bound also taxes key-event messages. Hard cap: the trouble -/// `gatt_service` macro initializes its characteristic arrays via -/// `Default`, which arrays only implement up to 32 elements — so -/// `SPLIT_MESSAGE_MAX_SIZE` (this + 1-byte length prefix + 1-byte enum -/// discriminant + 4 bytes margin) must stay ≤ 32. +/// Maximum payload of one application split message. Every split transfer is +/// `SPLIT_MESSAGE_MAX_SIZE` bytes on the wire, and trouble's `gatt_service` +/// macro caps that at 32 (it initializes characteristic arrays via `Default`, +/// which arrays only implement up to 32 elements), so this plus postcard +/// overhead must keep `SPLIT_MESSAGE_MAX_SIZE` ≤ 32. pub const SPLIT_APP_MSG_MAX: usize = 26; /// One opaque application payload. Only `data[..len]` is meaningful. @@ -101,24 +76,56 @@ impl MaxSize for SplitAppData { const POSTCARD_MAX_SIZE: usize = SPLIT_APP_MSG_MAX + 1; } -/// Central → peripheral queue, drained by `PeripheralManager` while the link -/// is up. Producers MUST use `try_send` (bounded, never block); capacity is -/// sized so one full application resync burst fits with headroom. +/// Central → peripheral queue. Producers must use `try_send` (never await a +/// full queue); capacity fits one full application resync burst. pub static SPLIT_APP_TX: Channel = Channel::new(); -/// This side's inbox of received application messages (peripheral: from the -/// central's `SPLIT_APP_TX`; central: from the peripheral's -/// `SPLIT_APP_PERIPH_TX`). Filled with `try_send` (drop-on-full) by the -/// split read loops. +/// This side's inbox of received application messages. Filled with `try_send` +/// (drop-on-full) by the split read loops. pub static SPLIT_APP_RX: Channel = Channel::new(); -/// Peripheral → central queue, drained by `SplitPeripheral` while the link -/// is up. Producers MUST use `try_send`. Small: the application announces -/// tiny, rare state (e.g. its build identity once per link-up). +/// Peripheral → central queue. Producers must use `try_send`. Small: meant +/// for tiny, rare state such as a build identity announced once per link-up. pub static SPLIT_APP_PERIPH_TX: Channel = Channel::new(); -/// Split-link state for the application: on the central, "peripheral link -/// up"; on the peripheral, "central link up". Written by the split driver at -/// session start/end; state-based (a late receiver still observes the latest -/// value), so edges cannot be lost the way pub/sub events can. +/// Deliver a received payload into [`SPLIT_APP_RX`]. Drop-on-full: a slow +/// application consumer must never stall the split read loops, and the +/// application resyncs on reconnect anyway. +pub(crate) fn deliver_received(data: SplitAppData) { + if SPLIT_APP_RX.try_send(data).is_err() { + warn!("split app message dropped (inbox full)"); + } +} + +/// Split-link state for the application, written by the split loops via +/// [`LinkGuard`]. State-based, so a late receiver still observes the current +/// value. pub static SPLIT_APP_LINK: Watch = Watch::new(); + +/// Owns one split session's [`SPLIT_APP_LINK`] state: `mark_up` sends the +/// link-up edge once, and `Drop` sends the link-down edge. The down edge must +/// come from `Drop` because the session futures holding a guard are cancelled +/// from outside on connection loss. +pub(crate) struct LinkGuard { + up: bool, +} + +impl LinkGuard { + pub(crate) fn new() -> Self { + Self { up: false } + } + + /// Send the link-up edge; idempotent. + pub(crate) fn mark_up(&mut self) { + if !self.up { + SPLIT_APP_LINK.sender().send(true); + self.up = true; + } + } +} + +impl Drop for LinkGuard { + fn drop(&mut self) { + SPLIT_APP_LINK.sender().send(false); + } +}