Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,12 @@ After a short receive buffer the frame is retained locally even though the queue
slot has been released. Explicit session invalidation prevents later endpoint
operations; detected queue corruption also latches invalidation. A malformed
complete message is retained and reported separately. Neither path resets a
queue, detects an unobservable peer reset or reclaims peer memory.
queue, detects an unobservable peer reset or automatically reclaims peer
memory. Once the platform proves that both peers and all users are quiescent,
the endpoint or its session can explicitly reclaim both backends through
`reclaim_after_peer_quiescent`; its safety contract also requires reconciliation
of caches and acceptance that a locally staged receive will be discarded.
Invalidation alone does not establish quiescence.

Run independent host tests with `cargo test --locked --target
x86_64-unknown-linux-gnu --features defmt`; the RISC-V check uses the project's
Expand Down
6 changes: 6 additions & 0 deletions src/session.rs
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,12 @@ impl<E> Session<E> {
pub fn is_valid(&self) -> bool {
self.valid
}

// Only for a transport-specific, quiescent reclaim path. Returning an
// endpoint to public callers could bypass this session's invalidation.
pub(crate) fn into_reclaim_endpoint(self) -> E {
self.endpoint
}
}
impl<E: Endpoint> Endpoint for Session<E> {
fn try_send(&mut self, message: &[u8]) -> Result<(), SendError> {
Expand Down
45 changes: 45 additions & 0 deletions src/transport/spsc_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -455,3 +455,48 @@ fn polling_endpoint_checks_authority_backpressure_and_preserves_short_receive()
Err(ReceiveError::Invalidated)
);
}

#[test]
fn invalidated_session_can_reclaim_both_quiescent_queues_with_staged_message() {
use crate::endpoint::{Endpoint, ReceiveError};
use crate::protocol::FirmwareProtocol;
use crate::session::Session;
use crate::transport::ws53::{FrameAuthority, PollingEndpoint};

struct RejectAll;
// SAFETY: the policy never authorizes publication to the synthetic peer.
unsafe impl FrameAuthority for RejectAll {
fn permits(&self, _: Message<'_>) -> bool {
false
}
}
let tx_memory = prepared(8, true);
let rx_memory = prepared(8, false);
let tx_drops = tx_memory.drops.clone();
let rx_drops = rx_memory.drops.clone();
let peer_rx = rx_memory.bytes.clone();
let tx = unsafe { Transmit::from_prepared(tx_memory, Channel::Channel1) }.unwrap();
let rx = unsafe { Receive::from_prepared(rx_memory, Channel::Channel1) }.unwrap();
let endpoint =
unsafe { PollingEndpoint::from_prepared(tx, rx, RejectAll, FirmwareProtocol::Sdk163d9150) };
let mut session = Session::new(endpoint);
assert!(sdk_push(&peer_rx, 8, &[1, 0, 2, 0, 0xaa, 0xbb]));
assert_eq!(
session.try_receive(&mut [0; 1]),
Err(ReceiveError::BufferTooSmall { required: 6 })
);
session.invalidate();
assert_eq!(
session.try_receive(&mut [0; 8]),
Err(ReceiveError::Invalidated)
);
// SAFETY: no peer thread exists, no shared payload references remain, and
// this test explicitly accepts discarding the endpoint's staged message.
let (tx_memory, rx_memory, _) = unsafe { session.reclaim_after_peer_quiescent() };
assert_eq!(tx_drops.load(AtomicOrdering::SeqCst), 0);
assert_eq!(rx_drops.load(AtomicOrdering::SeqCst), 0);
drop(tx_memory);
drop(rx_memory);
assert_eq!(tx_drops.load(AtomicOrdering::SeqCst), 1);
assert_eq!(rx_drops.load(AtomicOrdering::SeqCst), 1);
}
35 changes: 35 additions & 0 deletions src/transport/ws53.rs
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ use crate::{
endpoint::{Endpoint, ReceiveError, SendError},
mailbox::MailboxError,
protocol::FirmwareProtocol,
session::Session,
spsc::{Error, Receive, SharedMailbox, Transmit},
};

Expand Down Expand Up @@ -64,6 +65,40 @@ impl<T: SharedMailbox, R: SharedMailbox, A: FrameAuthority> PollingEndpoint<T, R
pending_len: None,
}
}

/// Recover both retained mailbox backends and the authority policy.
/// Any staged received message is discarded; this does not acknowledge it
/// to the peer or restore queue metadata.
///
/// # Safety
/// Both peers and all other users must be quiescent; no notification,
/// outstanding payload reference, cache operation or memory access may
/// remain. Reconcile ownership and cache state before reusing either
/// mapping. The caller must explicitly accept loss of any staged message.
/// This does not make stale queues safe to attach again.
pub unsafe fn reclaim_after_peer_quiescent(self) -> (T, R, A) {
// SAFETY: the caller supplies the quiescence and ownership proof for
// both queues, including any references retained by the peer.
let tx = unsafe { self.tx.reclaim_after_peer_quiescent() };
// SAFETY: same proof applies independently to the receive queue.
let rx = unsafe { self.rx.reclaim_after_peer_quiescent() };
(tx, rx, self.authority)
}
}

impl<T: SharedMailbox, R: SharedMailbox, A: FrameAuthority> Session<PollingEndpoint<T, R, A>> {
/// Consume this session and reclaim both queues without reopening it.
/// This is also available after explicit or corruption-driven invalidation.
///
/// # Safety
/// The full quiescence, cache, reference and staged-message conditions of
/// [`PollingEndpoint::reclaim_after_peer_quiescent`] must hold. Invalidating
/// a session does not itself stop the peer or establish these conditions.
pub unsafe fn reclaim_after_peer_quiescent(self) -> (T, R, A) {
let endpoint = self.into_reclaim_endpoint();
// SAFETY: caller supplies the endpoint's complete reclaim contract.
unsafe { endpoint.reclaim_after_peer_quiescent() }
}
}
impl<T: SharedMailbox, R: SharedMailbox, A: FrameAuthority> Endpoint for PollingEndpoint<T, R, A> {
fn try_send(&mut self, bytes: &[u8]) -> Result<(), SendError> {
Expand Down
Loading