use core::task::Poll;
use super::{
core::{CursorEndpoint, RecvRuntimeDesc},
lane_port,
};
use crate::{
binding::EndpointSlot,
control::cap::mint::{EpochTable, MintConfigMarker},
endpoint::{RecvError, RecvResult},
global::{
ControlDesc,
const_dsl::ScopeKind,
typestate::{PassiveArmNavigation, StateIndex, state_index_to_usize},
},
observe::ids,
policy_runtime::PolicySlot,
runtime::{config::Clock, consts::LabelUniverse},
transport::{
Transport,
trace::TapFrameMeta,
wire::{CodecError, FrameFlags, Payload},
},
};
pub(crate) enum RecvPayloadSource<'a> {
Empty,
Direct(Payload<'a>),
BindingWithUnconsumedTransport {
payload: Payload<'a>,
frame: lane_port::ReceivedFrame<'a>,
},
}
impl RecvPayloadSource<'_> {
#[inline]
fn discard_terminal(self) {
if let Self::BindingWithUnconsumedTransport { frame, .. } = self {
frame.discard_uncommitted();
}
}
}
enum RecvCommitEffect<'a> {
None,
RequeueTransport(lane_port::ReceivedFrame<'a>),
}
impl RecvCommitEffect<'_> {
#[inline]
fn discard_uncommitted(self) {
if let Self::RequeueTransport(frame) = self {
frame.discard_uncommitted();
}
}
}
struct RecvCommitPlan<'a> {
desc: RecvDescriptor,
payload: Payload<'a>,
next_index: StateIndex,
commit_effect: RecvCommitEffect<'a>,
}
pub(crate) struct RecvState {
prepared: Option<PreparedRecv>,
pending_recv: lane_port::PendingRecv,
}
impl RecvState {
#[inline]
pub(crate) const fn new() -> Self {
Self {
prepared: None,
pending_recv: lane_port::PendingRecv::new(),
}
}
#[inline]
pub(crate) fn prepared(&self) -> Option<PreparedRecv> {
self.prepared
}
#[inline]
pub(crate) fn set_prepared(&mut self, prepared: PreparedRecv) {
self.prepared = Some(prepared);
}
#[inline]
pub(crate) fn clear_prepared(&mut self) {
self.prepared = None;
}
}
#[derive(Clone, Copy)]
pub(crate) struct PreparedRecv {
pub(crate) descriptor: RecvDescriptor,
pub(crate) runtime: RecvRuntimeDesc,
}
#[derive(Clone, Copy)]
pub(crate) struct RecvDescriptor {
pub(crate) meta: crate::global::typestate::RecvMeta,
pub(crate) sid_raw: u32,
pub(crate) lane_idx: usize,
pub(crate) lane_wire: u8,
}
impl<'r, const ROLE: u8, T, U, C, E, const MAX_RV: usize, Mint, B>
CursorEndpoint<'r, ROLE, T, U, C, E, MAX_RV, Mint, B>
where
T: Transport + 'r,
U: LabelUniverse,
C: Clock,
E: EpochTable,
Mint: MintConfigMarker,
B: EndpointSlot + 'r,
{
fn prepare_recv_descriptor(
&mut self,
target_label: u8,
expects_control: bool,
accepts_empty_payload: bool,
) -> RecvResult<PreparedRecv> {
self.try_select_lane_for_label(target_label);
let mut iter_count = 0usize;
let descriptor_bound = self
.cursor
.local_steps_len()
.saturating_add(self.cursor.route_scope_count())
.saturating_add(1);
loop {
iter_count += 1;
debug_assert!(
iter_count <= descriptor_bound,
"recv() descriptor navigation exceeded structural bound iter={} bound={}",
iter_count,
descriptor_bound
);
if iter_count > descriptor_bound {
return Err(RecvError::PhaseInvariant);
}
if let Some(region) = self.cursor.scope_region()
&& region.kind == ScopeKind::Route
&& self.cursor.index() == region.start
{
let scope_id = region.scope_id;
let lane_wire = self.lane_for_label_or_offer(scope_id, target_label);
let existing_arm = self.route_arm_for(lane_wire, scope_id);
if let Some(arm) = existing_arm {
let recv_idx = self.cursor.route_scope_arm_recv_index(scope_id, arm);
if let Some(idx) = recv_idx {
self.set_cursor_index(idx);
continue;
}
if let Some(nav) = self.cursor.follow_passive_observer_arm(arm) {
let PassiveArmNavigation::WithinArm { entry } = nav;
self.set_cursor_index(state_index_to_usize(entry));
continue;
}
if self.cursor.advance_scope_if_kind_in_place(ScopeKind::Route) {
continue;
}
} else {
return Err(RecvError::PhaseInvariant);
}
}
if self.cursor.is_recv() {
break;
}
if let Some(region) = self.cursor.scope_region()
&& region.kind == ScopeKind::Route
&& self.can_advance_route_scope(region.scope_id, target_label)
&& self.cursor.advance_scope_if_kind_in_place(ScopeKind::Route)
{
continue;
}
return Err(RecvError::PhaseInvariant);
}
let meta = self
.cursor
.try_recv_meta()
.ok_or(RecvError::PhaseInvariant)?;
if meta.label != target_label {
return Err(RecvError::LabelMismatch {
expected: meta.label,
actual: target_label,
});
}
let lane_idx = meta.lane as usize;
let lane_wire = self.port_for_lane(lane_idx).lane().as_wire();
let descriptor = RecvDescriptor {
meta,
sid_raw: self.sid.raw(),
lane_idx,
lane_wire,
};
let runtime = RecvRuntimeDesc::new(
target_label,
crate::transport::FrameLabel::new(meta.frame_label),
expects_control,
accepts_empty_payload,
);
if meta.is_control != runtime.expects_control() {
return Err(RecvError::PhaseInvariant);
}
Ok(PreparedRecv {
descriptor,
runtime,
})
}
fn poll_recv_payload_source(
&mut self,
desc: RecvDescriptor,
accepts_empty_payload: bool,
pending_recv: &mut lane_port::PendingRecv,
cx: &mut core::task::Context<'_>,
) -> Poll<RecvResult<RecvPayloadSource<'r>>> {
if let Some(payload) = {
let scratch_ptr = {
let port = self.port_for_lane(desc.lane_idx);
lane_port::scratch_ptr(port)
};
self.try_recv_from_binding(desc.meta.lane, desc.meta.frame_label, scratch_ptr)
}? {
return Poll::Ready(Ok(RecvPayloadSource::Direct(payload)));
}
loop {
let frame = {
let port = self.port_for_lane(desc.lane_idx);
match lane_port::poll_recv_frame(pending_recv, port, cx) {
Poll::Pending => return Poll::Pending,
Poll::Ready(Ok(frame)) => frame,
Poll::Ready(Err(err)) => return Poll::Ready(Err(RecvError::Transport(err))),
}
};
let transport_payload_is_semantic = !frame.is_empty() || accepts_empty_payload;
let late_binding = {
let scratch_ptr = {
let port = self.port_for_lane(desc.lane_idx);
lane_port::scratch_ptr(port)
};
self.try_recv_from_binding(desc.meta.lane, desc.meta.frame_label, scratch_ptr)
};
let late_binding = match late_binding {
Ok(payload) => payload,
Err(err) => {
frame.discard_uncommitted();
return Poll::Ready(Err(err));
}
};
if let Some(payload) = late_binding {
if transport_payload_is_semantic {
return Poll::Ready(Ok(RecvPayloadSource::BindingWithUnconsumedTransport {
payload,
frame,
}));
} else {
frame.discard_uncommitted();
return Poll::Ready(Ok(RecvPayloadSource::Direct(payload)));
}
}
if frame.is_empty() {
if accepts_empty_payload {
let _ = frame.into_payload();
return Poll::Ready(Ok(RecvPayloadSource::Empty));
}
if !self.binding_inbox.is_enabled() {
frame.discard_uncommitted();
return Poll::Ready(Ok(RecvPayloadSource::Empty));
}
frame.discard_uncommitted();
continue;
}
return Poll::Ready(Ok(RecvPayloadSource::Direct(frame.into_payload())));
}
}
fn build_recv_commit_plan(
&self,
desc: RecvDescriptor,
payload_source: RecvPayloadSource<'r>,
erased: RecvRuntimeDesc,
control: Option<ControlDesc>,
validate: for<'a> fn(Payload<'a>) -> Result<(), CodecError>,
) -> RecvResult<RecvCommitPlan<'r>> {
let meta = desc.meta;
if erased.frame_label() != crate::transport::FrameLabel::new(meta.frame_label) {
payload_source.discard_terminal();
return Err(RecvError::PhaseInvariant);
}
if meta.is_control != erased.expects_control() {
payload_source.discard_terminal();
return Err(RecvError::PhaseInvariant);
}
let (payload, commit_effect) = match payload_source {
RecvPayloadSource::Empty if erased.accepts_empty_payload() => {
(Payload::new(&[]), RecvCommitEffect::None)
}
RecvPayloadSource::Empty => return Err(RecvError::PhaseInvariant),
RecvPayloadSource::Direct(payload) => (payload, RecvCommitEffect::None),
RecvPayloadSource::BindingWithUnconsumedTransport { payload, frame } => {
(payload, RecvCommitEffect::RequeueTransport(frame))
}
};
if let Err(err) = validate(payload) {
commit_effect.discard_uncommitted();
return Err(RecvError::Codec(err));
}
if let Err(err) = self.validate_inbound_explicit_wire_control(desc, control, payload) {
commit_effect.discard_uncommitted();
return Err(err);
}
let next_index = match self.cursor.try_next_index_past_jumps() {
Ok(index) => index,
Err(_) => {
commit_effect.discard_uncommitted();
return Err(RecvError::PhaseInvariant);
}
};
Ok(RecvCommitPlan {
desc,
payload,
next_index,
commit_effect,
})
}
fn publish_recv_commit_plan(&mut self, plan: RecvCommitPlan<'r>) -> RecvResult<Payload<'r>> {
let RecvCommitPlan {
desc,
payload,
next_index,
commit_effect,
} = plan;
let meta = desc.meta;
if let RecvCommitEffect::RequeueTransport(frame) = commit_effect {
let port = self.port_for_lane(frame.lane_idx());
lane_port::requeue_recv_frame(port, frame).map_err(RecvError::Transport)?;
}
self.emit_endpoint_policy_audit(
PolicySlot::EndpointRx,
ids::ENDPOINT_RECV,
desc.sid_raw,
Self::endpoint_policy_args(
crate::control::types::Lane::new(meta.lane as u32),
meta.label,
FrameFlags::empty(),
),
crate::control::types::Lane::new(meta.lane as u32),
);
let logical_meta = TapFrameMeta::new(
desc.sid_raw,
desc.lane_wire,
ROLE,
meta.label,
FrameFlags::empty(),
);
let scope_trace = self.scope_trace(meta.scope);
let event_id = if meta.is_control {
ids::ENDPOINT_CONTROL
} else {
ids::ENDPOINT_RECV
};
self.emit_endpoint_event(event_id, logical_meta, scope_trace, meta.lane);
self.set_cursor_index(next_index.as_usize());
self.advance_lane_cursor(desc.lane_idx, meta.eff_index);
self.maybe_skip_remaining_route_arm(meta.scope, meta.lane, meta.route_arm, meta.eff_index);
self.publish_scope_settlement(meta.scope, meta.route_arm, Some(meta.eff_index), meta.lane);
self.maybe_advance_phase();
Ok(payload)
}
fn finish_recv_payload(
&mut self,
desc: RecvDescriptor,
payload_source: RecvPayloadSource<'r>,
erased: RecvRuntimeDesc,
control: Option<ControlDesc>,
validate: for<'a> fn(Payload<'a>) -> Result<(), CodecError>,
) -> RecvResult<Payload<'r>> {
let plan = self.build_recv_commit_plan(desc, payload_source, erased, control, validate)?;
let payload = self.publish_recv_commit_plan(plan)?;
Ok(payload)
}
}
impl<'r, const ROLE: u8, T, U, C, E, const MAX_RV: usize, Mint, B>
super::core::RecvKernelEndpoint<'r> for CursorEndpoint<'r, ROLE, T, U, C, E, MAX_RV, Mint, B>
where
T: Transport + 'r,
U: LabelUniverse,
C: Clock,
E: EpochTable,
Mint: MintConfigMarker,
B: EndpointSlot + 'r,
{
#[inline]
fn prepare_recv_kernel_descriptor(
&mut self,
label: u8,
expects_control: bool,
accepts_empty_payload: bool,
) -> RecvResult<PreparedRecv> {
self.prepare_recv_descriptor(label, expects_control, accepts_empty_payload)
}
#[inline]
fn poll_recv_kernel_payload_source(
&mut self,
desc: RecvDescriptor,
accepts_empty_payload: bool,
state: &mut RecvState,
cx: &mut core::task::Context<'_>,
) -> Poll<RecvResult<RecvPayloadSource<'r>>> {
let pending_recv = &mut state.pending_recv;
self.poll_recv_payload_source(desc, accepts_empty_payload, pending_recv, cx)
}
#[inline]
fn finish_recv_kernel_payload(
&mut self,
desc: RecvDescriptor,
payload_source: RecvPayloadSource<'r>,
erased: RecvRuntimeDesc,
control: Option<ControlDesc>,
validate: for<'a> fn(Payload<'a>) -> Result<(), CodecError>,
) -> RecvResult<Payload<'r>> {
self.finish_recv_payload(desc, payload_source, erased, control, validate)
}
}