use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU8, Ordering};
use std::sync::Arc;
use std::time::Instant;
use bytes::Bytes;
use parking_lot::Mutex;
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use super::channel::{ChannelHash, ChannelId, ChannelName, ChannelPublisher, PublishConfig};
use super::mesh_rpc_metrics::{CallMetricsGuard, CallOutcome, ServiceMetricsAtomic};
use crate::adapter::net::cortex::{
build_trace_headers, encode_request_grant, encode_rpc_route, encode_stream_grant,
parse_request_window_initial, peek_request_service, EventMeta, RpcAsyncResponseEmitter,
RpcCancellationToken, RpcClientFold, RpcClientStreamingHandler, RpcContext, RpcDuplexFold,
RpcDuplexHandler, RpcHandler, RpcHandlerError, RpcInboundDispatcher, RpcInboundEvent,
RpcRequestChunkPayload, RpcRequestGrantEmitter, RpcRequestPayload, RpcResponseEmitter,
RpcResponsePayload, RpcServerFold, RpcServerStreamingFold, RpcStatus, RpcStreamingHandler,
RpcStreamingRequestFold, StreamItem, TraceContext, DISPATCH_RPC_CANCEL, DISPATCH_RPC_REQUEST,
DISPATCH_RPC_REQUEST_CHUNK, DISPATCH_RPC_REQUEST_GRANT, DISPATCH_RPC_STREAM_GRANT,
EVENT_META_SIZE, FLAG_RPC_CLIENT_STREAMING_REQUEST, FLAG_RPC_PROPAGATE_TRACE,
FLAG_RPC_REQUEST_END, FLAG_RPC_STREAMING_RESPONSE, HEADER_NRPC_REQUEST_WINDOW_INITIAL,
HEADER_NRPC_STREAM_WINDOW_INITIAL, RPC_FRAME_BODY_OFFSET, RPC_ROUTE_V1_SIZE,
};
use crate::error::AdapterError;
use super::behavior::org::{OrgId, OrgMembershipCert};
use super::behavior::org_admission::OrgAdmission;
use super::behavior::org_call::{OrgCallProof, MAX_ORG_PROOF_TTL_SECS, ORG_ADMISSION_HEADER};
use super::behavior::org_grant::{CapabilityAuthorityId, OrgCapabilityGrant, OrgDispatcherGrant};
use super::mesh::{MeshNode, PeerPublishOutcome, ReplySubscription};
use super::org_admission_gate::{
org_request_digest, CapabilityVisibility, OrgProviderPolicy, RegisteredRpcService,
};
use super::subnet::SubnetExportBinding;
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub enum RoutingPolicy {
#[default]
RoundRobin,
Random,
Sticky {
key: u64,
},
LowestLatency,
}
#[derive(Debug, Clone)]
pub struct CallOptions {
pub deadline: Option<Instant>,
pub routing_policy: RoutingPolicy,
pub filter_unhealthy: bool,
pub trace_context: Option<TraceContext>,
pub max_in_flight_per_target: u32,
pub stream_window_initial: Option<u32>,
pub request_window_initial: Option<u32>,
pub request_headers: Vec<(String, Vec<u8>)>,
pub cancel_token: Option<u64>,
pub org_proof_intent: Option<OrgProofIntent>,
}
impl Default for CallOptions {
fn default() -> Self {
Self {
deadline: None,
routing_policy: RoutingPolicy::default(),
filter_unhealthy: true,
trace_context: None,
max_in_flight_per_target: 64,
stream_window_initial: None,
request_window_initial: None,
request_headers: Vec::new(),
cancel_token: None,
org_proof_intent: None,
}
}
}
#[derive(Clone)]
pub struct OrgProofIntent {
pub caller: Arc<crate::adapter::net::identity::EntityKeypair>,
pub membership: OrgMembershipCert,
pub dispatcher: OrgDispatcherGrant,
pub capability_grant: Option<OrgCapabilityGrant>,
pub acting_org: OrgId,
pub provider_owner_org: OrgId,
pub provider: crate::adapter::net::identity::EntityId,
pub capability: CapabilityAuthorityId,
pub proof_ttl_secs: u64,
}
impl std::fmt::Debug for OrgProofIntent {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("OrgProofIntent")
.field("acting_org", &self.acting_org)
.field("provider_owner_org", &self.provider_owner_org)
.field("provider", &self.provider)
.field("capability", &self.capability)
.field("proof_ttl_secs", &self.proof_ttl_secs)
.field("credentials", &"<redacted>")
.finish()
}
}
#[derive(Debug, Clone)]
pub struct RpcReply {
pub body: Bytes,
pub headers: Vec<(String, Vec<u8>)>,
pub latency_ns: u64,
}
#[derive(Debug, thiserror::Error)]
pub enum RpcError {
#[error("no route to target {target:#x}: {reason}")]
NoRoute {
target: u64,
reason: String,
},
#[error("timeout after {elapsed_ms}ms")]
Timeout {
elapsed_ms: u64,
},
#[error("server returned status {status:#06x}: {message}")]
ServerError {
status: u16,
message: String,
headers: Vec<(String, Vec<u8>)>,
},
#[error("transport: {0}")]
Transport(#[from] AdapterError),
#[error("codec ({direction:?}): {message}")]
Codec {
direction: CodecDirection,
message: String,
},
#[error("capability denied: target {target:#x} does not authorize nrpc:{capability}")]
CapabilityDenied {
target: u64,
capability: String,
},
#[error("call cancelled by caller")]
Cancelled,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CodecDirection {
Encode,
Decode,
}
pub struct ServeHandle {
channel_hash: ChannelHash,
registration_id: u64,
service: String,
_bridge: JoinHandle<()>,
_response_drain: Option<JoinHandle<()>>,
mesh: Arc<MeshNode>,
#[cfg(test)]
origin_node_cache: RpcOriginNodeCache,
}
impl Drop for ServeHandle {
fn drop(&mut self) {
self.mesh
.unregister_rpc_inbound(self.channel_hash, self.registration_id);
self.mesh
.rpc_local_services_arc()
.remove_if(&self.service, self.registration_id);
}
}
fn is_cross_service_request(frame: &[u8], expected: &str) -> bool {
let Some(meta) = (if frame.len() >= EVENT_META_SIZE {
EventMeta::from_bytes(&frame[..EVENT_META_SIZE])
} else {
None
}) else {
return false;
};
if meta.dispatch != DISPATCH_RPC_REQUEST {
return false;
}
match peek_request_service(frame) {
Some(svc) => svc != expected,
None => false,
}
}
fn cache_authenticated_response_destination(
mesh: &MeshNode,
cache: &RpcOriginNodeCache,
inbound: &RpcInboundEvent,
) {
let meta = if inbound.payload.len() >= EVENT_META_SIZE {
EventMeta::from_bytes(&inbound.payload[..EVENT_META_SIZE])
} else {
None
};
let authenticated_peer_origin = mesh
.peer_entity_id(inbound.from_node)
.map(|e| e.origin_hash());
if response_route_is_trustworthy(
inbound.from_node,
meta.as_ref().map(|m| m.dispatch),
inbound.origin_hash,
authenticated_peer_origin,
) {
if let Some(m) = meta {
cache.insert(
(inbound.from_node, inbound.origin_hash, m.seq_or_ts),
inbound.from_node,
);
}
}
}
fn response_route_is_trustworthy(
from_node: u64,
dispatch: Option<u8>,
claimed_origin: u64,
authenticated_peer_origin: Option<u64>,
) -> bool {
from_node != 0
&& dispatch == Some(DISPATCH_RPC_REQUEST)
&& authenticated_peer_origin == Some(claimed_origin)
}
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
enum RequestGrantRoute {
TrustedDirect(u64),
Loopback,
RelayedOrUntrusted,
}
fn classify_request_grant_route(
from_node: u64,
claimed_origin: u64,
authenticated_peer_origin: Option<u64>,
) -> RequestGrantRoute {
if from_node == 0 {
RequestGrantRoute::Loopback
} else if authenticated_peer_origin == Some(claimed_origin) {
RequestGrantRoute::TrustedDirect(from_node)
} else {
RequestGrantRoute::RelayedOrUntrusted
}
}
fn streaming_response_is_terminal(resp: &RpcResponsePayload) -> bool {
resp.status != RpcStatus::Ok
|| resp.headers.iter().any(|(name, value)| {
name.eq_ignore_ascii_case(crate::adapter::net::cortex::HEADER_NRPC_STREAMING)
&& value.as_slice() == crate::adapter::net::cortex::HEADER_NRPC_STREAMING_END
})
}
enum BridgePreflight {
Proceed(RpcInboundEvent),
Drop,
Deny {
claimed_origin: u64,
call_id: u64,
from_node: u64,
},
}
fn bridge_origin_check(
inbound: &RpcInboundEvent,
expected_service: &str,
tag: &str,
metrics: &ServiceMetricsAtomic,
) -> Option<EventMeta> {
if is_cross_service_request(&inbound.payload, expected_service) {
return None;
}
let meta = (if inbound.payload.len() >= EVENT_META_SIZE {
EventMeta::from_bytes(&inbound.payload[..EVENT_META_SIZE])
} else {
None
})?;
let from_node = inbound.from_node;
if from_node != 0 && inbound.origin_hash != meta.origin_hash {
metrics
.packet_origin_mismatch_dropped_total
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
tracing::warn!(
service = expected_service,
tag = tag,
from_node = format!("{from_node:#x}"),
packet_origin = format!("{:#x}", inbound.origin_hash),
payload_origin = format!("{:#x}", meta.origin_hash),
call_id = meta.seq_or_ts,
"nrpc: dropping frame whose packet origin != payload (EventMeta) origin \
before admission — a direct peer must not run the fold under a forged \
payload origin",
);
return None;
}
Some(meta)
}
fn bridge_preflight(
mesh: &MeshNode,
cache: &RpcOriginNodeCache,
inbound: &RpcInboundEvent,
expected_service: &str,
tag: &str,
metrics: &ServiceMetricsAtomic,
) -> BridgePreflight {
let Some(meta) = bridge_origin_check(inbound, expected_service, tag, metrics) else {
return BridgePreflight::Drop;
};
let from_node = inbound.from_node;
if from_node != 0
&& !crate::adapter::net::behavior::fold::capability_bridge::may_admit(
mesh.capability_fold(),
mesh.node_id(),
tag,
from_node,
)
{
return BridgePreflight::Deny {
claimed_origin: inbound.origin_hash,
call_id: meta.seq_or_ts,
from_node,
};
}
cache_authenticated_response_destination(mesh, cache, inbound);
let frame = strip_public_admission_header(inbound).unwrap_or_else(|| inbound.clone());
BridgePreflight::Proceed(frame)
}
async fn emit_capability_denial(
mesh: &MeshNode,
service: &str,
claimed_origin: u64,
call_id: u64,
from_node: u64,
) {
let resp = crate::adapter::net::cortex::RpcResponsePayload {
status: RpcStatus::CapabilityDenied,
headers: vec![],
body: Bytes::from(format!(
"callee-side capability-auth gate denied nrpc:{service}"
)),
};
let meta = EventMeta::new(
crate::adapter::net::cortex::DISPATCH_RPC_RESPONSE,
0,
mesh.identity_origin_hash(),
call_id,
0,
);
let mut buf = Vec::with_capacity(EVENT_META_SIZE + 64);
buf.extend_from_slice(&meta.to_bytes());
resp.encode_into(&mut buf);
let reply_origin = mesh
.peer_entity_id(from_node)
.map(|e| e.origin_hash())
.unwrap_or(claimed_origin);
let reply_channel_name = format!("{service}.replies.{reply_origin:016x}");
let Ok(reply_channel) = ChannelName::new(&reply_channel_name) else {
return;
};
let reply_channel_id = ChannelId::new(reply_channel.clone());
let reply_channel_hash = reply_channel_id.hash();
let reply_stream_id = MeshNode::publish_stream_id(&reply_channel_id);
let _ = publish_response_to_caller(
mesh,
reply_origin,
Some(from_node),
&reply_channel,
reply_channel_hash,
reply_stream_id,
Bytes::from(buf),
ResponseRouteFallback::DirectOnly,
)
.await;
}
fn emit_admission_denial(
mesh: &MeshNode,
resp_tx: &mpsc::Sender<RpcResponseJob>,
service: &str,
claimed_origin: u64,
call_id: u64,
from_node: u64,
coarse: crate::adapter::net::behavior::org_admission::CoarseAdmissionReason,
) {
let resp = crate::adapter::net::cortex::RpcResponsePayload {
status: RpcStatus::AdmissionDenied,
headers: vec![],
body: Bytes::copy_from_slice(&[coarse.to_wire()]),
};
let meta = EventMeta::new(
crate::adapter::net::cortex::DISPATCH_RPC_RESPONSE,
0,
mesh.identity_origin_hash(),
call_id,
0,
);
let mut buf = Vec::with_capacity(EVENT_META_SIZE + 16);
buf.extend_from_slice(&meta.to_bytes());
resp.encode_into(&mut buf);
let reply_origin = mesh
.peer_entity_id(from_node)
.map(|e| e.origin_hash())
.unwrap_or(claimed_origin);
let Ok(reply_channel) = ChannelName::new(&format!("{service}.replies.{reply_origin:016x}"))
else {
return;
};
let reply_channel_id = ChannelId::new(reply_channel.clone());
let reply_channel_hash = reply_channel_id.hash();
let reply_stream_id = MeshNode::publish_stream_id(&reply_channel_id);
if resp_tx
.try_send(RpcResponseJob {
caller_origin: reply_origin,
call_id,
target_hint: Some(from_node),
reply_channel,
reply_channel_hash,
reply_stream_id,
payload: Bytes::from(buf),
})
.is_err()
{
tracing::debug!(
from_node = format!("{:#x}", from_node),
call_id,
"rpc admission: response drainer at capacity; dropping denial"
);
}
}
fn strip_public_admission_header(inbound: &RpcInboundEvent) -> Option<RpcInboundEvent> {
if inbound.payload.len() < EVENT_META_SIZE {
return None;
}
match EventMeta::from_bytes(&inbound.payload[..EVENT_META_SIZE]) {
Some(meta) if meta.dispatch == DISPATCH_RPC_REQUEST => {}
_ => return None,
}
let needle = ORG_ADMISSION_HEADER.as_bytes();
if inbound.payload.len() < RPC_FRAME_BODY_OFFSET
|| !inbound.payload.windows(needle.len()).any(|w| w == needle)
{
return None;
}
let mut req = RpcRequestPayload::decode(inbound.payload.slice(RPC_FRAME_BODY_OFFSET..)).ok()?;
if !req.headers.iter().any(|(n, _)| n == ORG_ADMISSION_HEADER) {
return None;
}
req.headers.retain(|(n, _)| n != ORG_ADMISSION_HEADER);
let mut buf = inbound.payload[..RPC_FRAME_BODY_OFFSET].to_vec();
req.encode_into(&mut buf);
Some(RpcInboundEvent {
channel_hash: inbound.channel_hash,
origin_hash: inbound.origin_hash,
from_node: inbound.from_node,
payload: Bytes::from(buf),
})
}
#[allow(clippy::too_many_arguments)]
async fn admit_and_dispatch_protected(
mesh: &Arc<MeshNode>,
cache: &RpcOriginNodeCache,
inbound: &RpcInboundEvent,
service: &str,
tag: &str,
metrics: &ServiceMetricsAtomic,
reg: &crate::adapter::net::org_admission_gate::RegisteredRpcService,
replay: &crate::adapter::net::behavior::org_admission_replay::AdmissionReplayGuard,
fold: &Arc<Mutex<RpcServerFold>>,
resp_tx: &mpsc::Sender<RpcResponseJob>,
) {
use crate::adapter::net::behavior::org_admission::{AdmissionContext, CoarseAdmissionReason};
use crate::adapter::net::org_admission_gate as gate;
let Some(meta) = bridge_origin_check(inbound, service, tag, metrics) else {
return;
};
let from_node = inbound.from_node;
let claimed_origin = meta.origin_hash;
let call_id = meta.seq_or_ts;
if meta.dispatch != DISPATCH_RPC_REQUEST {
if let Err(e) = fold.lock().apply_inbound(inbound) {
tracing::warn!(error = %e, "rpc serve_rpc_protected: fold apply error");
}
return;
}
let caller = match mesh.resolve_direct_caller(from_node, claimed_origin) {
Ok(caller) => caller,
Err(_) => {
emit_admission_denial(
mesh,
resp_tx,
service,
claimed_origin,
call_id,
from_node,
CoarseAdmissionReason::Denied,
);
return;
}
};
if !crate::adapter::net::behavior::fold::capability_bridge::has_local_capability(
mesh.capability_fold(),
mesh.node_id(),
tag,
) {
emit_admission_denial(
mesh,
resp_tx,
service,
claimed_origin,
call_id,
from_node,
CoarseAdmissionReason::Denied,
);
return;
}
if inbound.payload.len() < RPC_FRAME_BODY_OFFSET {
return;
}
let Ok(payload) = RpcRequestPayload::decode(inbound.payload.slice(RPC_FRAME_BODY_OFFSET..))
else {
emit_admission_denial(
mesh,
resp_tx,
service,
claimed_origin,
call_id,
from_node,
CoarseAdmissionReason::Denied,
);
return;
};
let Ok(request_digest) = gate::org_request_digest(&payload) else {
emit_admission_denial(
mesh,
resp_tx,
service,
claimed_origin,
call_id,
from_node,
CoarseAdmissionReason::Denied,
);
return;
};
let admission_headers: Vec<&[u8]> = payload
.headers
.iter()
.filter(|(n, _)| n == crate::adapter::net::behavior::org_call::ORG_ADMISSION_HEADER)
.map(|(_, v)| v.as_slice())
.collect();
let is_unary =
payload.flags & (FLAG_RPC_CLIENT_STREAMING_REQUEST | FLAG_RPC_STREAMING_RESPONSE) == 0;
let clock = crate::adapter::net::behavior::admission_clock::ClockSample::now();
let subnet_export_facts = match reg.subnet_export() {
Some(binding) => match gate::verify_subnet_export(mesh, binding, &clock) {
Ok(facts) => Some(facts),
Err(denied) => {
emit_admission_denial(
mesh,
resp_tx,
service,
claimed_origin,
call_id,
from_node,
denied.coarse(),
);
return;
}
},
None => None,
};
let facts = match gate::verify_provider_authority(mesh, &clock) {
Ok(f) => f,
Err(d) => {
emit_admission_denial(
mesh,
resp_tx,
service,
claimed_origin,
call_id,
from_node,
d.coarse(),
);
return;
}
};
let invoked_capability =
crate::adapter::net::behavior::org_grant::CapabilityAuthorityId::for_tag(tag);
let ctx = AdmissionContext {
mode: reg.admission(),
authenticated_caller: &caller,
provider: &facts.provider,
provider_owner_org: facts.provider_owner_org,
invoked_capability,
call_id,
request_digest,
is_unary,
floors: facts.floors.as_ref(),
skew_secs: facts.skew_secs,
};
let captured_stamp = facts.stamp;
#[cfg(test)]
if reg.red_witness_admission_disabled() {
cache_authenticated_response_destination(mesh, cache, inbound);
let admitted = crate::adapter::net::behavior::org_admission::Admitted {
caller: caller.clone(),
acting_org: facts.provider_owner_org,
provider_org: facts.provider_owner_org,
provider: facts.provider.clone(),
capability: invoked_capability,
};
if let Err(e) = fold.lock().apply_inbound_admitted(inbound, admitted) {
tracing::warn!(error = %e, "rpc serve_rpc_protected: fold apply error");
}
return;
}
if !mesh
.admission_rate_limiter()
.may_attempt(from_node, clock.monotonic)
{
tracing::warn!(
service = service,
from_node = format!("{:#x}", from_node),
"nrpc: org admission throttled — peer exhausted its failed-admission budget",
);
metrics
.capability_denied_total
.fetch_add(1, Ordering::Relaxed);
emit_admission_denial(
mesh,
resp_tx,
service,
claimed_origin,
call_id,
from_node,
CoarseAdmissionReason::Unavailable,
);
return;
}
let outcome = crate::adapter::net::behavior::org_admission::verify_org_admission(
&ctx,
&admission_headers,
replay,
clock,
|| {
captured_stamp.is_current(&gate::capture_admission_stamp(mesh))
&& subnet_export_facts
.as_ref()
.is_none_or(|facts| facts.is_current(mesh))
},
|proof| (reg.provider_policy())(proof),
);
match outcome {
Ok(admitted) => {
cache_authenticated_response_destination(mesh, cache, inbound);
if let Err(e) = fold.lock().apply_inbound_admitted(inbound, admitted) {
tracing::warn!(error = %e, "rpc serve_rpc_protected: fold apply error");
}
}
Err(denied) => {
if !matches!(
denied,
crate::adapter::net::behavior::org_admission::AdmissionDenied::AuthorityChanged
) {
mesh.admission_rate_limiter()
.on_failure(from_node, clock.monotonic);
}
tracing::warn!(service = service, reason = ?denied, "nrpc: org admission denied");
emit_admission_denial(
mesh,
resp_tx,
service,
claimed_origin,
call_id,
from_node,
denied.coarse(),
);
}
}
}
fn reject_relayed_flow_controlled_request(
mesh: &MeshNode,
metrics: &ServiceMetricsAtomic,
inbound: &RpcInboundEvent,
service: &str,
tag: &str,
) -> bool {
let Some(meta) = (if inbound.payload.len() >= EVENT_META_SIZE {
EventMeta::from_bytes(&inbound.payload[..EVENT_META_SIZE])
} else {
None
}) else {
return false;
};
if meta.dispatch != DISPATCH_RPC_REQUEST {
return false;
}
if inbound.payload.len() < RPC_FRAME_BODY_OFFSET {
return false;
}
let flow_controlled = matches!(
RpcRequestPayload::decode(inbound.payload.slice(RPC_FRAME_BODY_OFFSET..)),
Ok(request) if parse_request_window_initial(&request.headers).is_some()
);
if !flow_controlled {
return false;
}
let authenticated_peer_origin = mesh
.peer_entity_id(inbound.from_node)
.map(|e| e.origin_hash());
if RequestGrantRoute::RelayedOrUntrusted
== classify_request_grant_route(
inbound.from_node,
inbound.origin_hash,
authenticated_peer_origin,
)
{
metrics
.relayed_flow_controlled_rejected_total
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
tracing::warn!(
service = service,
tag = tag,
from_node = format!("{:#x}", inbound.from_node),
claimed_origin = format!("{:#x}", inbound.origin_hash),
call_id = meta.seq_or_ts,
"nrpc: rejecting relayed/untrusted flow-controlled upload before fold — \
secure upload grants require a directly authenticated caller session; \
relayed flow-controlled nRPC is unsupported until end-to-end recipient \
correlation exists",
);
return true;
}
false
}
struct RpcResponseJob {
caller_origin: u64,
call_id: u64,
target_hint: Option<u64>,
reply_channel: ChannelName,
reply_channel_hash: ChannelHash,
reply_stream_id: u64,
payload: Bytes,
}
#[derive(Clone)]
struct CachedReplyChannel {
name: ChannelName,
hash: ChannelHash,
stream_id: u64,
}
pub struct RpcStream {
mesh: Arc<MeshNode>,
target_node_id: u64,
request_channel: ChannelName,
request_channel_hash: ChannelHash,
request_stream_id: u64,
self_origin: u64,
call_id: u64,
inner: tokio::sync::mpsc::UnboundedReceiver<StreamItem>,
done: bool,
stream_window: Option<u32>,
grant_pending: u32,
observer: StreamingObserverState,
_cancel_keep_alive: StreamCancelKeepAlive,
}
impl RpcStream {
pub fn call_id(&self) -> u64 {
self.call_id
}
pub fn flow_controlled(&self) -> bool {
self.stream_window.is_some()
}
pub fn grant(&self, amount: u32) {
if !self.flow_controlled() || amount == 0 {
return;
}
spawn_grant_publish(
Arc::clone(&self.mesh),
self.target_node_id,
self.request_channel_hash,
self.request_stream_id,
self.self_origin,
self.call_id,
amount,
);
}
}
fn accumulate_auto_grant(pending: &mut u32, window: u32) -> Option<u32> {
*pending = pending.saturating_add(1);
let threshold = (window / 2).max(1);
if *pending >= threshold {
let amount = *pending;
*pending = 0;
Some(amount)
} else {
None
}
}
fn spawn_grant_publish(
mesh: Arc<MeshNode>,
target: u64,
request_channel_hash: ChannelHash,
request_stream_id: u64,
self_origin: u64,
call_id: u64,
amount: u32,
) {
tokio::spawn(async move {
let meta = EventMeta::new(DISPATCH_RPC_STREAM_GRANT, 0, self_origin, call_id, 0);
let mut buf = Vec::with_capacity(EVENT_META_SIZE + RPC_ROUTE_V1_SIZE + 4);
buf.extend_from_slice(&meta.to_bytes());
encode_rpc_route(&mut buf, request_channel_hash);
buf.extend_from_slice(&encode_stream_grant(amount));
let payload = Bytes::from(buf);
let _ = mesh
.publish_to_peer(
target,
request_channel_hash,
request_stream_id,
true,
std::slice::from_ref(&payload),
)
.await;
});
}
impl futures::Stream for RpcStream {
type Item = Result<Bytes, RpcError>;
fn poll_next(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<Option<Self::Item>> {
if self.done {
return std::task::Poll::Ready(None);
}
match self.inner.poll_recv(cx) {
std::task::Poll::Ready(Some(StreamItem::Chunk(body))) => {
if let Some(window) = self.stream_window {
let mut pending = self.grant_pending;
if let Some(amount) = accumulate_auto_grant(&mut pending, window) {
spawn_grant_publish(
Arc::clone(&self.mesh),
self.target_node_id,
self.request_channel_hash,
self.request_stream_id,
self.self_origin,
self.call_id,
amount,
);
}
self.grant_pending = pending;
}
self.observer.add_response_bytes(body.len() as u32);
std::task::Poll::Ready(Some(Ok(body)))
}
std::task::Poll::Ready(Some(StreamItem::End)) => {
self.done = true;
self.observer.latch_ok();
std::task::Poll::Ready(None)
}
std::task::Poll::Ready(Some(StreamItem::Error(resp))) => {
self.done = true;
let status = resp.status.to_wire();
let message = String::from_utf8(resp.body.to_vec()).unwrap_or_else(|e| {
format!("<{} bytes of non-utf8 body>", e.into_bytes().len())
});
self.observer
.latch_error(format!("server returned status {status:#06x}: {message}"));
std::task::Poll::Ready(Some(Err(RpcError::ServerError {
status,
message,
headers: resp.headers,
})))
}
std::task::Poll::Ready(None) => {
self.done = true;
std::task::Poll::Ready(None)
}
std::task::Poll::Pending => std::task::Poll::Pending,
}
}
}
impl Drop for RpcStream {
fn drop(&mut self) {
self.mesh.rpc_client_pending_arc().cancel(self.call_id);
spawn_cancel_publish(
Arc::clone(&self.mesh),
self.target_node_id,
self.request_channel.clone(),
self.self_origin,
self.call_id,
);
self.observer.fire();
}
}
async fn publish_request_chunk(
mesh: &Arc<MeshNode>,
target: u64,
request_channel_hash: ChannelHash,
request_stream_id: u64,
self_origin: u64,
chunk: &RpcRequestChunkPayload,
) -> Result<(), RpcError> {
let meta = EventMeta::new(DISPATCH_RPC_REQUEST_CHUNK, 0, self_origin, chunk.call_id, 0);
let mut buf = Vec::with_capacity(EVENT_META_SIZE + RPC_ROUTE_V1_SIZE + chunk.encoded_len());
buf.extend_from_slice(&meta.to_bytes());
encode_rpc_route(&mut buf, request_channel_hash);
chunk.encode_into(&mut buf);
let payload = Bytes::from(buf);
mesh.publish_to_peer(
target,
request_channel_hash,
request_stream_id,
true,
std::slice::from_ref(&payload),
)
.await
.map_err(RpcError::Transport)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ClientStreamState {
JustOpened,
Sending,
Finishing,
Done,
}
pub struct ClientStreamCallRaw {
mesh: Arc<MeshNode>,
target_node_id: u64,
request_channel: ChannelName,
request_channel_hash: ChannelHash,
request_stream_id: u64,
self_origin: u64,
call_id: u64,
service: String,
initial_headers: Vec<(String, Vec<u8>)>,
initial_flags: u16,
deadline_ns: u64,
credit_sem: Option<Arc<tokio::sync::Semaphore>>,
grant_pump: Option<JoinHandle<()>>,
terminal_rx: Option<tokio::sync::oneshot::Receiver<RpcResponsePayload>>,
state: ClientStreamState,
started: Instant,
observer: StreamingObserverState,
_cancel_keep_alive: StreamCancelKeepAlive,
}
impl ClientStreamCallRaw {
pub fn call_id(&self) -> u64 {
self.call_id
}
pub fn flow_controlled(&self) -> bool {
self.credit_sem.is_some()
}
pub async fn send(&mut self, body: Bytes) -> Result<(), RpcError> {
match self.state {
ClientStreamState::Finishing | ClientStreamState::Done => {
return Err(RpcError::Codec {
direction: CodecDirection::Encode,
message: "send() called after finish()".to_string(),
});
}
_ => {}
}
if let Some(sem) = self.credit_sem.as_ref() {
let permit = sem.clone().acquire_owned().await.map_err(|_| {
RpcError::Transport(AdapterError::Connection("credit semaphore closed".into()))
})?;
permit.forget();
}
self.observer.add_request_bytes(body.len() as u32);
match self.state {
ClientStreamState::JustOpened => {
let req = RpcRequestPayload {
service: self.service.clone(),
deadline_ns: self.deadline_ns,
flags: self.initial_flags,
headers: std::mem::take(&mut self.initial_headers),
body: body.clone(),
};
self.publish_initial_request(&req).await?;
self.state = ClientStreamState::Sending;
}
ClientStreamState::Sending => {
let chunk = RpcRequestChunkPayload {
call_id: self.call_id,
flags: 0,
headers: vec![],
body: body.clone(),
};
publish_request_chunk(
&self.mesh,
self.target_node_id,
self.request_channel_hash,
self.request_stream_id,
self.self_origin,
&chunk,
)
.await?;
}
ClientStreamState::Finishing | ClientStreamState::Done => unreachable!(),
}
Ok(())
}
pub async fn finish(mut self) -> Result<RpcReply, RpcError> {
match self.state {
ClientStreamState::JustOpened => {
let req = RpcRequestPayload {
service: self.service.clone(),
deadline_ns: self.deadline_ns,
flags: self.initial_flags | FLAG_RPC_REQUEST_END,
headers: std::mem::take(&mut self.initial_headers),
body: Bytes::new(),
};
self.publish_initial_request(&req).await?;
}
ClientStreamState::Sending => {
let chunk = RpcRequestChunkPayload {
call_id: self.call_id,
flags: FLAG_RPC_REQUEST_END,
headers: vec![],
body: Bytes::new(),
};
publish_request_chunk(
&self.mesh,
self.target_node_id,
self.request_channel_hash,
self.request_stream_id,
self.self_origin,
&chunk,
)
.await?;
}
ClientStreamState::Finishing | ClientStreamState::Done => {
return Err(RpcError::Codec {
direction: CodecDirection::Encode,
message: "finish() called twice".to_string(),
});
}
}
self.state = ClientStreamState::Finishing;
let terminal_rx = self.terminal_rx.take().ok_or_else(|| {
RpcError::Transport(AdapterError::Connection(
"terminal receiver already consumed".into(),
))
})?;
let resp = if self.deadline_ns > 0 {
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos() as u64)
.unwrap_or(0);
let remaining = self.deadline_ns.saturating_sub(now);
match tokio::time::timeout(std::time::Duration::from_nanos(remaining), terminal_rx)
.await
{
Ok(Ok(r)) => r,
Ok(Err(_)) => {
let msg = "terminal sender dropped before response arrived";
self.observer.latch_error(msg);
return Err(RpcError::Transport(AdapterError::Connection(msg.into())));
}
Err(_elapsed) => {
let elapsed_ms = self.started.elapsed().as_millis() as u64;
self.observer.latch_timeout();
return Err(RpcError::Timeout { elapsed_ms });
}
}
} else {
match terminal_rx.await {
Ok(r) => r,
Err(_) => {
let msg = "terminal sender dropped before response arrived";
self.observer.latch_error(msg);
return Err(RpcError::Transport(AdapterError::Connection(msg.into())));
}
}
};
self.state = ClientStreamState::Done;
self.observer.add_response_bytes(resp.body.len() as u32);
if !resp.status.is_ok() {
let message = String::from_utf8(resp.body.to_vec())
.unwrap_or_else(|e| format!("<{} bytes of non-utf8 body>", e.into_bytes().len()));
self.observer.latch_error(format!(
"server returned status {:#06x}: {message}",
resp.status.to_wire()
));
return Err(RpcError::ServerError {
status: resp.status.to_wire(),
message,
headers: resp.headers,
});
}
self.observer.latch_ok();
let latency_ns = self.started.elapsed().as_nanos() as u64;
Ok(RpcReply {
body: resp.body,
headers: resp.headers,
latency_ns,
})
}
async fn publish_initial_request(&self, req: &RpcRequestPayload) -> Result<(), RpcError> {
let meta = EventMeta::new(DISPATCH_RPC_REQUEST, 0, self.self_origin, self.call_id, 0);
let mut buf = Vec::with_capacity(EVENT_META_SIZE + RPC_ROUTE_V1_SIZE + req.encoded_len());
buf.extend_from_slice(&meta.to_bytes());
encode_rpc_route(&mut buf, self.request_channel_hash);
req.encode_into(&mut buf);
let payload = Bytes::from(buf);
self.mesh
.publish_to_peer(
self.target_node_id,
self.request_channel_hash,
self.request_stream_id,
true,
std::slice::from_ref(&payload),
)
.await
.map_err(RpcError::Transport)
}
}
impl Drop for ClientStreamCallRaw {
fn drop(&mut self) {
if let Some(task) = self.grant_pump.take() {
task.abort();
}
self.observer.fire();
if matches!(self.state, ClientStreamState::Done) {
return;
}
self.mesh.rpc_client_pending_arc().cancel(self.call_id);
if !matches!(self.state, ClientStreamState::JustOpened) {
spawn_cancel_publish(
Arc::clone(&self.mesh),
self.target_node_id,
self.request_channel.clone(),
self.self_origin,
self.call_id,
);
}
}
}
struct DuplexInner {
mesh: Arc<MeshNode>,
target_node_id: u64,
request_channel: ChannelName,
request_channel_hash: ChannelHash,
request_stream_id: u64,
self_origin: u64,
call_id: u64,
initial_sent: std::sync::atomic::AtomicBool,
clean_close: std::sync::atomic::AtomicBool,
observer: StreamingObserverState,
_cancel_keep_alive: Option<StreamCancelKeepAlive>,
}
impl Drop for DuplexInner {
fn drop(&mut self) {
self.mesh.rpc_client_pending_arc().cancel(self.call_id);
self.observer.fire();
if self.clean_close.load(Ordering::SeqCst) {
return;
}
if !self.initial_sent.load(Ordering::SeqCst) {
return;
}
spawn_cancel_publish(
Arc::clone(&self.mesh),
self.target_node_id,
self.request_channel.clone(),
self.self_origin,
self.call_id,
);
}
}
pub struct DuplexSink {
inner: Arc<DuplexInner>,
service: String,
initial_headers: Vec<(String, Vec<u8>)>,
initial_flags: u16,
deadline_ns: u64,
credit_sem: Option<Arc<tokio::sync::Semaphore>>,
grant_pump: Option<JoinHandle<()>>,
state: ClientStreamState,
}
impl DuplexSink {
pub async fn send(&mut self, body: Bytes) -> Result<(), RpcError> {
match self.state {
ClientStreamState::Finishing | ClientStreamState::Done => {
return Err(RpcError::Codec {
direction: CodecDirection::Encode,
message: "send() called after finish_sending()".to_string(),
});
}
_ => {}
}
if let Some(sem) = self.credit_sem.as_ref() {
let permit = sem.clone().acquire_owned().await.map_err(|_| {
RpcError::Transport(AdapterError::Connection("credit semaphore closed".into()))
})?;
permit.forget();
}
self.inner.observer.add_request_bytes(body.len() as u32);
match self.state {
ClientStreamState::JustOpened => {
let req = RpcRequestPayload {
service: self.service.clone(),
deadline_ns: self.deadline_ns,
flags: self.initial_flags,
headers: std::mem::take(&mut self.initial_headers),
body: body.clone(),
};
self.publish_initial_request(&req).await?;
self.inner.initial_sent.store(true, Ordering::SeqCst);
self.state = ClientStreamState::Sending;
}
ClientStreamState::Sending => {
let chunk = RpcRequestChunkPayload {
call_id: self.inner.call_id,
flags: 0,
headers: vec![],
body: body.clone(),
};
publish_request_chunk(
&self.inner.mesh,
self.inner.target_node_id,
self.inner.request_channel_hash,
self.inner.request_stream_id,
self.inner.self_origin,
&chunk,
)
.await?;
}
ClientStreamState::Finishing | ClientStreamState::Done => unreachable!(),
}
Ok(())
}
pub async fn finish_sending(mut self) -> Result<(), RpcError> {
match self.state {
ClientStreamState::JustOpened => {
let req = RpcRequestPayload {
service: self.service.clone(),
deadline_ns: self.deadline_ns,
flags: self.initial_flags | FLAG_RPC_REQUEST_END,
headers: std::mem::take(&mut self.initial_headers),
body: Bytes::new(),
};
self.publish_initial_request(&req).await?;
self.inner.initial_sent.store(true, Ordering::SeqCst);
}
ClientStreamState::Sending => {
let chunk = RpcRequestChunkPayload {
call_id: self.inner.call_id,
flags: FLAG_RPC_REQUEST_END,
headers: vec![],
body: Bytes::new(),
};
publish_request_chunk(
&self.inner.mesh,
self.inner.target_node_id,
self.inner.request_channel_hash,
self.inner.request_stream_id,
self.inner.self_origin,
&chunk,
)
.await?;
}
ClientStreamState::Finishing | ClientStreamState::Done => {
return Err(RpcError::Codec {
direction: CodecDirection::Encode,
message: "finish_sending() called twice".to_string(),
});
}
}
self.state = ClientStreamState::Finishing;
Ok(())
}
pub fn call_id(&self) -> u64 {
self.inner.call_id
}
pub fn flow_controlled(&self) -> bool {
self.credit_sem.is_some()
}
async fn publish_initial_request(&self, req: &RpcRequestPayload) -> Result<(), RpcError> {
let meta = EventMeta::new(
DISPATCH_RPC_REQUEST,
0,
self.inner.self_origin,
self.inner.call_id,
0,
);
let mut buf = Vec::with_capacity(EVENT_META_SIZE + RPC_ROUTE_V1_SIZE + req.encoded_len());
buf.extend_from_slice(&meta.to_bytes());
encode_rpc_route(&mut buf, self.inner.request_channel_hash);
req.encode_into(&mut buf);
let payload = Bytes::from(buf);
self.inner
.mesh
.publish_to_peer(
self.inner.target_node_id,
self.inner.request_channel_hash,
self.inner.request_stream_id,
true,
std::slice::from_ref(&payload),
)
.await
.map_err(RpcError::Transport)
}
}
impl Drop for DuplexSink {
fn drop(&mut self) {
if let Some(task) = self.grant_pump.take() {
task.abort();
}
}
}
pub struct DuplexStream {
inner: Arc<DuplexInner>,
chunks_rx: tokio::sync::mpsc::UnboundedReceiver<StreamItem>,
done: bool,
}
impl DuplexStream {
pub fn call_id(&self) -> u64 {
self.inner.call_id
}
}
impl futures::Stream for DuplexStream {
type Item = Result<Bytes, RpcError>;
fn poll_next(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<Option<Self::Item>> {
if self.done {
return std::task::Poll::Ready(None);
}
match self.chunks_rx.poll_recv(cx) {
std::task::Poll::Ready(Some(StreamItem::Chunk(body))) => {
self.inner.observer.add_response_bytes(body.len() as u32);
std::task::Poll::Ready(Some(Ok(body)))
}
std::task::Poll::Ready(Some(StreamItem::End)) => {
self.done = true;
self.inner.clean_close.store(true, Ordering::SeqCst);
self.inner.observer.latch_ok();
std::task::Poll::Ready(None)
}
std::task::Poll::Ready(Some(StreamItem::Error(resp))) => {
self.done = true;
self.inner.clean_close.store(true, Ordering::SeqCst);
let status = resp.status.to_wire();
let message = String::from_utf8(resp.body.to_vec()).unwrap_or_else(|e| {
format!("<{} bytes of non-utf8 body>", e.into_bytes().len())
});
self.inner
.observer
.latch_error(format!("server returned status {status:#06x}: {message}"));
std::task::Poll::Ready(Some(Err(RpcError::ServerError {
status,
message,
headers: resp.headers,
})))
}
std::task::Poll::Ready(None) => {
self.done = true;
std::task::Poll::Ready(None)
}
std::task::Poll::Pending => std::task::Poll::Pending,
}
}
}
pub struct DuplexCallRaw {
sink: DuplexSink,
stream: DuplexStream,
}
impl DuplexCallRaw {
pub fn call_id(&self) -> u64 {
self.sink.call_id()
}
pub fn flow_controlled(&self) -> bool {
self.sink.flow_controlled()
}
pub async fn send(&mut self, body: Bytes) -> Result<(), RpcError> {
self.sink.send(body).await
}
pub async fn finish_sending(&mut self) -> Result<(), RpcError> {
let placeholder = DuplexSink {
inner: Arc::clone(&self.sink.inner),
service: String::new(),
initial_headers: Vec::new(),
initial_flags: 0,
deadline_ns: 0,
credit_sem: None,
grant_pump: None,
state: ClientStreamState::Done,
};
let sink = std::mem::replace(&mut self.sink, placeholder);
sink.finish_sending().await
}
pub async fn next(&mut self) -> Option<Result<Bytes, RpcError>> {
use futures::StreamExt;
self.stream.next().await
}
pub fn into_split(self) -> (DuplexSink, DuplexStream) {
(self.sink, self.stream)
}
}
impl futures::Stream for DuplexCallRaw {
type Item = Result<Bytes, RpcError>;
fn poll_next(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<Option<Self::Item>> {
std::pin::Pin::new(&mut self.stream).poll_next(cx)
}
}
struct UnaryCallGuard {
pending: Arc<crate::adapter::net::cortex::RpcClientPending>,
mesh: Arc<MeshNode>,
target_node_id: u64,
request_channel: ChannelName,
self_origin: u64,
call_id: u64,
completed: bool,
}
impl Drop for UnaryCallGuard {
fn drop(&mut self) {
self.pending.cancel(self.call_id);
if !self.completed {
spawn_cancel_publish(
Arc::clone(&self.mesh),
self.target_node_id,
self.request_channel.clone(),
self.self_origin,
self.call_id,
);
}
}
}
pub(crate) struct StreamingObserverState {
mesh: Arc<MeshNode>,
target_node_id: u64,
service: String,
started: Instant,
request_bytes: AtomicU32,
response_bytes: AtomicU32,
observer_status: AtomicU8,
observer_msg: parking_lot::Mutex<Option<String>>,
fired: AtomicBool,
}
impl StreamingObserverState {
pub(crate) fn new(
mesh: Arc<MeshNode>,
target_node_id: u64,
service: impl Into<String>,
request_bytes: u32,
) -> Self {
Self {
mesh,
target_node_id,
service: service.into(),
started: Instant::now(),
request_bytes: AtomicU32::new(request_bytes),
response_bytes: AtomicU32::new(0),
observer_status: AtomicU8::new(0),
observer_msg: parking_lot::Mutex::new(None),
fired: AtomicBool::new(false),
}
}
pub(crate) fn add_request_bytes(&self, n: u32) {
self.request_bytes.fetch_add(n, Ordering::Relaxed);
}
pub(crate) fn add_response_bytes(&self, n: u32) {
self.response_bytes.fetch_add(n, Ordering::Relaxed);
}
pub(crate) fn latch_ok(&self) {
self.observer_status.store(1, Ordering::Relaxed);
}
pub(crate) fn latch_error(&self, msg: impl Into<String>) {
*self.observer_msg.lock() = Some(msg.into());
self.observer_status.store(2, Ordering::Relaxed);
}
pub(crate) fn latch_timeout(&self) {
self.observer_status.store(3, Ordering::Relaxed);
}
pub(crate) fn fire(&self) {
if self.fired.swap(true, Ordering::SeqCst) {
return;
}
let status_code = self.observer_status.load(Ordering::Relaxed);
let status = match status_code {
1 => crate::adapter::net::cortex::rpc_observer::RpcCallStatus::Ok,
2 => {
let msg = self.observer_msg.lock().clone().unwrap_or_default();
crate::adapter::net::cortex::rpc_observer::RpcCallStatus::Error(msg)
}
3 => crate::adapter::net::cortex::rpc_observer::RpcCallStatus::Timeout,
_ => crate::adapter::net::cortex::rpc_observer::RpcCallStatus::Canceled,
};
self.mesh.fire_rpc_observer_outbound(
self.target_node_id,
&self.service,
self.started.elapsed().as_millis() as u32,
status,
self.request_bytes.load(Ordering::Relaxed),
self.response_bytes.load(Ordering::Relaxed),
);
}
}
const REQUEST_GRANT_PER_CALL_CAP: usize = 1_000_000;
fn add_request_grant_credits(sem: &tokio::sync::Semaphore, credits: u32) {
if credits == 0 {
return;
}
let current = sem.available_permits();
let remaining = REQUEST_GRANT_PER_CALL_CAP.saturating_sub(current);
let safe = (credits as usize).min(usize::MAX >> 4).min(remaining);
if safe > 0 {
sem.add_permits(safe);
}
}
fn request_grant_route(from_node: u64) -> (Option<u64>, ResponseRouteFallback) {
if from_node != 0 {
(Some(from_node), ResponseRouteFallback::DirectOnly)
} else {
(None, ResponseRouteFallback::RosterOnStaleDirect)
}
}
fn build_request_grant_emitter(
mesh: Arc<MeshNode>,
service: String,
server_origin: u64,
diag_tag: &'static str,
) -> RpcRequestGrantEmitter {
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<(u64, u64, u64, u32)>();
tokio::spawn(async move {
while let Some(first) = rx.recv().await {
let mut summed: std::collections::HashMap<(u64, u64, u64), u32> =
std::collections::HashMap::new();
let (from_node, caller, call_id, credits) = first;
summed.insert((from_node, caller, call_id), credits);
while let Ok((from_node, caller, call_id, credits)) = rx.try_recv() {
let entry = summed.entry((from_node, caller, call_id)).or_insert(0);
*entry = entry.saturating_add(credits);
}
for ((from_node, caller, call_id), credits) in summed {
let reply_channel_name = format!("{service}.replies.{caller:016x}");
let reply_channel = match ChannelName::new(&reply_channel_name) {
Ok(c) => c,
Err(e) => {
tracing::warn!(
error = %e,
channel = %reply_channel_name,
tag = diag_tag,
"rpc grant drainer: invalid reply channel name");
continue;
}
};
let reply_channel_id = ChannelId::new(reply_channel.clone());
let reply_channel_hash = reply_channel_id.hash();
let reply_stream_id = MeshNode::publish_stream_id(&reply_channel_id);
let meta = EventMeta::new(DISPATCH_RPC_REQUEST_GRANT, 0, server_origin, call_id, 0);
let mut buf = Vec::with_capacity(EVENT_META_SIZE + 12);
buf.extend_from_slice(&meta.to_bytes());
buf.extend_from_slice(&encode_request_grant(call_id, credits));
let (target_hint, fallback) = request_grant_route(from_node);
if let Err(e) = publish_response_to_caller(
&mesh,
caller,
target_hint,
&reply_channel,
reply_channel_hash,
reply_stream_id,
Bytes::from(buf),
fallback,
)
.await
{
tracing::warn!(
error = %e,
caller_origin = format!("{:#x}", caller),
call_id,
tag = diag_tag,
"rpc grant drainer: REQUEST_GRANT publish failed");
}
}
}
});
Arc::new(move |from_node, caller_origin, call_id, credits| {
let _ = tx.send((from_node, caller_origin, call_id, credits));
})
}
type RpcOriginNodeCache = Arc<BoundedLru<(u64, u64, u64), u64>>;
const RPC_CALLER_CACHE_CAP: usize = 4096;
const RPC_CALLER_CACHE_CAP_NZ: std::num::NonZeroUsize =
match std::num::NonZeroUsize::new(RPC_CALLER_CACHE_CAP) {
Some(n) => n,
None => panic!("RPC_CALLER_CACHE_CAP must be non-zero"),
};
struct BoundedLru<K, V>(Mutex<lru::LruCache<K, V>>);
impl<K: std::hash::Hash + Eq, V: Clone> BoundedLru<K, V> {
fn new() -> Self {
Self(Mutex::new(lru::LruCache::new(RPC_CALLER_CACHE_CAP_NZ)))
}
fn get(&self, key: K) -> Option<V> {
self.0.lock().get(&key).cloned()
}
fn insert(&self, key: K, value: V) {
self.0.lock().put(key, value);
}
fn remove(&self, key: K) {
self.0.lock().pop(&key);
}
}
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
enum ResponseRouteFallback {
RosterOnStaleDirect,
DirectOnly,
}
#[allow(clippy::too_many_arguments)]
async fn publish_response_to_caller(
mesh: &MeshNode,
caller_origin: u64,
target_hint: Option<u64>,
reply_channel: &ChannelName,
reply_channel_hash: ChannelHash,
reply_stream_id: u64,
payload: Bytes,
fallback: ResponseRouteFallback,
) -> Result<(), AdapterError> {
let payload = crate::adapter::net::cortex::insert_rpc_route(payload, reply_channel_hash);
let resolved = match fallback {
ResponseRouteFallback::DirectOnly => target_hint,
ResponseRouteFallback::RosterOnStaleDirect => {
target_hint.or_else(|| mesh.get_node_by_origin_hash(caller_origin))
}
};
if let Some(target_node_id) = resolved {
match mesh
.try_publish_to_peer(
target_node_id,
reply_channel_hash,
reply_stream_id,
true,
std::slice::from_ref(&payload),
)
.await
{
PeerPublishOutcome::Sent => return Ok(()),
PeerPublishOutcome::SendFailed(e) => return Err(e),
PeerPublishOutcome::NoSession => {
if fallback == ResponseRouteFallback::DirectOnly {
tracing::debug!(
caller_origin = format!("{caller_origin:#x}"),
target_node = format!("{target_node_id:#x}"),
"rpc direct-only frame: peer session gone at send time; dropping",
);
return Ok(());
}
tracing::debug!(
caller_origin = format!("{caller_origin:#x}"),
target_node = format!("{target_node_id:#x}"),
"rpc response: resolved route has no peer session; roster fallback",
);
}
}
} else if fallback == ResponseRouteFallback::DirectOnly {
return Ok(());
}
let publisher = ChannelPublisher::new(reply_channel.clone(), PublishConfig::default());
mesh.publish(&publisher, payload).await.map(|_| ())
}
static CANCEL_PUBLISH_RT: std::sync::OnceLock<tokio::runtime::Handle> = std::sync::OnceLock::new();
pub(crate) fn remember_cancel_publish_runtime() {
if let Ok(handle) = tokio::runtime::Handle::try_current() {
let _ = CANCEL_PUBLISH_RT.set(handle);
}
}
fn spawn_cancel_publish(
mesh: Arc<MeshNode>,
target: u64,
request_channel: ChannelName,
self_origin: u64,
call_id: u64,
) {
let handle = tokio::runtime::Handle::try_current()
.ok()
.or_else(|| CANCEL_PUBLISH_RT.get().cloned());
let Some(handle) = handle else {
tracing::debug!(
call_id,
target_node = format!("{target:#x}"),
"rpc CANCEL: no runtime to publish on (dropped during teardown); \
relying on the peer's keep-alive expiry",
);
return;
};
handle.spawn(async move {
let meta = EventMeta::new(DISPATCH_RPC_CANCEL, 0, self_origin, call_id, 0);
let request_channel_id = ChannelId::new(request_channel);
let request_channel_hash = request_channel_id.hash();
let stream_id = MeshNode::publish_stream_id(&request_channel_id);
let mut buf = meta.to_bytes().to_vec();
encode_rpc_route(&mut buf, request_channel_hash);
let payload = Bytes::from(buf);
let _ = mesh
.publish_to_peer(
target,
request_channel_hash,
stream_id,
true,
std::slice::from_ref(&payload),
)
.await;
});
}
type StreamCancelKeepAlive = tokio::sync::oneshot::Sender<()>;
fn spawn_stream_cancel_watcher(
cancel_notify: Arc<tokio::sync::Notify>,
cancel_token: u64,
cancel_registry: Arc<crate::adapter::net::cancel_registry::CancelRegistry>,
pending: Arc<crate::adapter::net::cortex::RpcClientPending>,
call_id: u64,
) -> StreamCancelKeepAlive {
let (done_tx, done_rx) = tokio::sync::oneshot::channel();
if cancel_token == 0 {
return done_tx;
}
tokio::spawn(async move {
tokio::select! {
biased;
_ = cancel_notify.notified() => {
pending.cancel(call_id);
cancel_registry.release(cancel_token);
}
_ = done_rx => {
cancel_registry.release(cancel_token);
}
}
});
done_tx
}
fn arm_stream_cancel(
mesh: &Arc<MeshNode>,
opts: &CallOptions,
pending: &Arc<crate::adapter::net::cortex::RpcClientPending>,
call_id: u64,
) -> StreamCancelKeepAlive {
let cancel_token = opts.cancel_token.unwrap_or(0);
let cancel_notify = mesh.cancel_registry().register_notify(cancel_token);
spawn_stream_cancel_watcher(
cancel_notify,
cancel_token,
Arc::clone(mesh.cancel_registry()),
Arc::clone(pending),
call_id,
)
}
fn fire_unary_cancel_outcome(
mesh: &Arc<MeshNode>,
metrics_guard: &mut crate::adapter::net::mesh_rpc_metrics::CallMetricsGuard,
cancel_token: u64,
target_node_id: u64,
service: &str,
started_total: Instant,
request_bytes_len: u32,
) -> RpcError {
mesh.cancel_registry().release(cancel_token);
metrics_guard.record(crate::adapter::net::mesh_rpc_metrics::CallOutcome::Transport);
mesh.fire_rpc_observer_outbound(
target_node_id,
service,
started_total.elapsed().as_millis() as u32,
crate::adapter::net::cortex::rpc_observer::RpcCallStatus::Canceled,
request_bytes_len,
0,
);
RpcError::Cancelled
}
impl MeshNode {
pub fn serve_rpc<H: RpcHandler>(
self: &Arc<Self>,
service: &str,
handler: Arc<H>,
) -> Result<ServeHandle, ServeError> {
self.serve_rpc_unary_impl(service, handler, UnaryAdmission::Public)
}
pub fn serve_rpc_protected<H: RpcHandler>(
self: &Arc<Self>,
service: &str,
handler: Arc<H>,
admission: OrgAdmission,
provider_policy: OrgProviderPolicy,
) -> Result<ServeHandle, ServeError> {
if matches!(admission, OrgAdmission::PublicAuthenticated) {
return Err(ServeError::InvalidProtectedRegistration(
"admission mode must be org-protected (OwnerDelegated / CrossOrgGranted), \
not PublicAuthenticated"
.to_string(),
));
}
if self.node_authority().is_none() {
return Err(ServeError::ProtectedAuthorityRequired(service.to_string()));
}
self.serve_rpc_unary_impl(
service,
handler,
UnaryAdmission::Protected {
admission,
provider_policy,
},
)
}
pub fn serve_rpc_subnet_exported<H: RpcHandler>(
self: &Arc<Self>,
service: &str,
handler: Arc<H>,
admission: OrgAdmission,
export: SubnetExportBinding,
provider_policy: OrgProviderPolicy,
) -> Result<ServeHandle, ServeError> {
if matches!(admission, OrgAdmission::PublicAuthenticated) {
return Err(ServeError::InvalidProtectedRegistration(
"admission mode must be org-protected (OwnerDelegated / CrossOrgGranted), \
not PublicAuthenticated"
.to_string(),
));
}
if self.node_authority().is_none() {
return Err(ServeError::ProtectedAuthorityRequired(service.to_string()));
}
let clock = crate::adapter::net::behavior::admission_clock::ClockSample::now();
if let Err(denied) =
crate::adapter::net::org_admission_gate::verify_subnet_export(self, &export, &clock)
{
return Err(ServeError::SubnetExportUnauthorized(format!(
"{service}: {denied:?}"
)));
}
self.serve_rpc_unary_impl(
service,
handler,
UnaryAdmission::SubnetExported {
admission,
export,
provider_policy,
},
)
}
#[cfg(test)]
pub(crate) fn serve_rpc_protected_red_witness_disabled<H: RpcHandler>(
self: &Arc<Self>,
service: &str,
handler: Arc<H>,
admission: OrgAdmission,
provider_policy: OrgProviderPolicy,
) -> Result<ServeHandle, ServeError> {
if matches!(admission, OrgAdmission::PublicAuthenticated) {
return Err(ServeError::InvalidProtectedRegistration(
"admission mode must be org-protected".to_string(),
));
}
if self.node_authority().is_none() {
return Err(ServeError::ProtectedAuthorityRequired(service.to_string()));
}
self.serve_rpc_unary_impl(
service,
handler,
UnaryAdmission::ProtectedRedWitnessDisabled {
admission,
provider_policy,
},
)
}
pub fn serve_rpc_owner_scoped<H: RpcHandler>(
self: &Arc<Self>,
service: &str,
handler: Arc<H>,
provider_policy: OrgProviderPolicy,
) -> Result<ServeHandle, ServeError> {
if self.node_authority().is_none() {
return Err(ServeError::ProtectedAuthorityRequired(service.to_string()));
}
self.serve_rpc_unary_impl(
service,
handler,
UnaryAdmission::OwnerScoped { provider_policy },
)
}
pub fn serve_rpc_granted<H: RpcHandler>(
self: &Arc<Self>,
service: &str,
handler: Arc<H>,
provider_policy: OrgProviderPolicy,
) -> Result<ServeHandle, ServeError> {
if self.node_authority().is_none() {
return Err(ServeError::ProtectedAuthorityRequired(service.to_string()));
}
self.serve_rpc_unary_impl(
service,
handler,
UnaryAdmission::Granted { provider_policy },
)
}
fn serve_rpc_unary_impl<H: RpcHandler>(
self: &Arc<Self>,
service: &str,
handler: Arc<H>,
mode: UnaryAdmission,
) -> Result<ServeHandle, ServeError> {
let request_channel = ChannelName::new(&format!("{service}.requests"))
.map_err(|e| ServeError::InvalidServiceName(e.to_string()))?;
let channel_hash = request_channel.hash();
if let Some(registry) = self.channel_configs() {
registry.install_rpc_service_defaults(service)?;
}
let response_fallback = mode.response_route_fallback();
let (tx, mut rx) = mpsc::channel::<RpcInboundEvent>(1024);
let origin_node_cache: RpcOriginNodeCache = Arc::new(BoundedLru::new());
let service_for_emit = service.to_string();
let server_origin = self.identity_origin_hash();
let origin_node_cache_for_emit = Arc::clone(&origin_node_cache);
let reply_channel_cache: Arc<BoundedLru<u64, CachedReplyChannel>> =
Arc::new(BoundedLru::new());
let (resp_tx, mut resp_rx) = mpsc::channel::<RpcResponseJob>(1024);
let resp_tx_for_denials = resp_tx.clone();
let emit: RpcResponseEmitter = Arc::new(move |from_node, caller_origin, call_id, resp| {
let target_hint = origin_node_cache_for_emit.get((from_node, caller_origin, call_id));
let cached = match reply_channel_cache.get(caller_origin) {
Some(c) => c,
None => {
let name = format!("{service_for_emit}.replies.{caller_origin:016x}");
match ChannelName::new(&name) {
Ok(channel_name) => {
let channel_id = ChannelId::new(channel_name.clone());
let triple = CachedReplyChannel {
hash: channel_id.hash(),
stream_id: MeshNode::publish_stream_id(&channel_id),
name: channel_name,
};
reply_channel_cache.insert(caller_origin, triple.clone());
triple
}
Err(e) => {
tracing::warn!(error = %e, channel = %name,
"rpc serve_rpc: invalid reply channel name");
return;
}
}
}
};
let meta = EventMeta::new(
crate::adapter::net::cortex::DISPATCH_RPC_RESPONSE,
0,
server_origin,
call_id,
0,
);
let mut buf = Vec::with_capacity(EVENT_META_SIZE + 64);
buf.extend_from_slice(&meta.to_bytes());
resp.encode_into(&mut buf);
if resp_tx
.try_send(RpcResponseJob {
caller_origin,
call_id,
target_hint,
reply_channel: cached.name,
reply_channel_hash: cached.hash,
reply_stream_id: cached.stream_id,
payload: Bytes::from(buf),
})
.is_err()
{
tracing::debug!(
caller_origin = format!("{:#x}", caller_origin),
call_id,
"rpc serve_rpc: response drainer at capacity; dropping response"
);
}
origin_node_cache_for_emit.remove((from_node, caller_origin, call_id));
});
let metrics_handle = self.rpc_metrics_arc().for_service(service);
let metrics_for_bridge = Arc::clone(&metrics_handle);
let fold = Arc::new(Mutex::new(
RpcServerFold::new(handler as Arc<dyn RpcHandler>, emit).with_metrics(metrics_handle),
));
let dispatcher: RpcInboundDispatcher = Arc::new(move |ev| {
let _ = tx.try_send(ev);
});
let Some(registration_id) = self.register_rpc_inbound(channel_hash, dispatcher) else {
return Err(ServeError::AlreadyServing(service.to_string()));
};
let visibility = mode.visibility();
self.rpc_local_services_arc()
.insert(service.to_string(), registration_id, visibility);
self.index_self_with_local_services();
let reg = Arc::new(match mode {
UnaryAdmission::Public => {
RegisteredRpcService::public(registration_id, Arc::from(service))
}
UnaryAdmission::Protected {
admission,
provider_policy,
} => match RegisteredRpcService::protected(
registration_id,
Arc::from(service),
admission,
provider_policy,
) {
Ok(reg) => reg,
Err(e) => {
self.unregister_rpc_inbound(channel_hash, registration_id);
self.rpc_local_services_arc()
.remove_if(service, registration_id);
return Err(ServeError::InvalidProtectedRegistration(e.to_string()));
}
},
UnaryAdmission::OwnerScoped { provider_policy } => RegisteredRpcService::owner_scoped(
registration_id,
Arc::from(service),
provider_policy,
),
UnaryAdmission::Granted { provider_policy } => {
RegisteredRpcService::granted(registration_id, Arc::from(service), provider_policy)
}
UnaryAdmission::SubnetExported {
admission,
export,
provider_policy,
} => match RegisteredRpcService::subnet_exported(
registration_id,
Arc::from(service),
admission,
export,
provider_policy,
) {
Ok(reg) => reg,
Err(e) => {
self.unregister_rpc_inbound(channel_hash, registration_id);
self.rpc_local_services_arc()
.remove_if(service, registration_id);
return Err(ServeError::InvalidProtectedRegistration(e.to_string()));
}
},
#[cfg(test)]
UnaryAdmission::ProtectedRedWitnessDisabled {
admission,
provider_policy,
} => match RegisteredRpcService::protected(
registration_id,
Arc::from(service),
admission,
provider_policy,
) {
Ok(reg) => reg.with_red_witness_disabled(),
Err(e) => {
self.unregister_rpc_inbound(channel_hash, registration_id);
self.rpc_local_services_arc()
.remove_if(service, registration_id);
return Err(ServeError::InvalidProtectedRegistration(e.to_string()));
}
},
});
let admission_replay = self.rpc_admission_replay_arc();
let mesh_for_bridge = Arc::clone(self);
let service_for_bridge = service.to_string();
let origin_node_cache_for_bridge = Arc::clone(&origin_node_cache);
let reg_for_bridge = Arc::clone(®);
let replay_for_bridge = Arc::clone(&admission_replay);
let bridge = tokio::spawn(async move {
let tag = format!("nrpc:{}", service_for_bridge);
while let Some(inbound) = rx.recv().await {
match reg_for_bridge.admission() {
OrgAdmission::PublicAuthenticated => {
match bridge_preflight(
&mesh_for_bridge,
&origin_node_cache_for_bridge,
&inbound,
&service_for_bridge,
&tag,
&metrics_for_bridge,
) {
BridgePreflight::Proceed(frame) => {
if let Err(e) = fold.lock().apply_inbound(&frame) {
tracing::warn!(error = %e, "rpc serve_rpc: fold apply error");
}
}
BridgePreflight::Drop => continue,
BridgePreflight::Deny {
claimed_origin,
call_id,
from_node,
} => {
metrics_for_bridge
.capability_denied_total
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
emit_capability_denial(
&mesh_for_bridge,
&service_for_bridge,
claimed_origin,
call_id,
from_node,
)
.await;
continue;
}
}
}
OrgAdmission::OwnerDelegated | OrgAdmission::CrossOrgGranted => {
admit_and_dispatch_protected(
&mesh_for_bridge,
&origin_node_cache_for_bridge,
&inbound,
&service_for_bridge,
&tag,
&metrics_for_bridge,
®_for_bridge,
&replay_for_bridge,
&fold,
&resp_tx_for_denials,
)
.await;
}
}
}
});
let response_drain_mesh = Arc::clone(self);
let response_drain = tokio::spawn(async move {
while let Some(job) = resp_rx.recv().await {
if let Err(e) = publish_response_to_caller(
&response_drain_mesh,
job.caller_origin,
job.target_hint,
&job.reply_channel,
job.reply_channel_hash,
job.reply_stream_id,
job.payload,
response_fallback,
)
.await
{
tracing::warn!(
error = %e,
caller_origin = format!("{:#x}", job.caller_origin),
call_id = job.call_id,
"rpc serve_rpc: response publish failed"
);
}
}
});
let mesh_for_announce = Arc::clone(self);
let service_for_log = service.to_string();
tokio::spawn(async move {
if let Err(e) = mesh_for_announce.reannounce_current_capabilities().await {
tracing::warn!(
error = %e,
service = %service_for_log,
"serve_rpc: auto re-announce failed",
);
}
});
Ok(ServeHandle {
channel_hash,
registration_id,
service: service.to_string(),
_bridge: bridge,
_response_drain: Some(response_drain),
mesh: Arc::clone(self),
#[cfg(test)]
origin_node_cache: origin_node_cache.clone(),
})
}
pub fn serve_rpc_streaming<H: RpcStreamingHandler>(
self: &Arc<Self>,
service: &str,
handler: Arc<H>,
) -> Result<ServeHandle, ServeError> {
let request_channel = ChannelName::new(&format!("{service}.requests"))
.map_err(|e| ServeError::InvalidServiceName(e.to_string()))?;
let channel_hash = request_channel.hash();
if let Some(registry) = self.channel_configs() {
registry.install_rpc_service_defaults(service)?;
}
let (tx, mut rx) = tokio::sync::mpsc::channel::<RpcInboundEvent>(1024);
let origin_node_cache: RpcOriginNodeCache = Arc::new(BoundedLru::new());
let mesh_for_emit = Arc::clone(self);
let service_for_emit = service.to_string();
let server_origin = self.identity_origin_hash();
let origin_node_cache_for_emit = Arc::clone(&origin_node_cache);
let emit: RpcAsyncResponseEmitter =
Arc::new(move |from_node, caller_origin, call_id, resp| {
let mesh = Arc::clone(&mesh_for_emit);
let service = service_for_emit.clone();
let target_hint =
origin_node_cache_for_emit.get((from_node, caller_origin, call_id));
if streaming_response_is_terminal(&resp) {
origin_node_cache_for_emit.remove((from_node, caller_origin, call_id));
}
Box::pin(async move {
let reply_channel_name = format!("{service}.replies.{caller_origin:016x}");
let reply_channel = match ChannelName::new(&reply_channel_name) {
Ok(c) => c,
Err(e) => {
tracing::warn!(error = %e, channel = %reply_channel_name,
"rpc serve_rpc_streaming: invalid reply channel name");
return;
}
};
let meta = EventMeta::new(
crate::adapter::net::cortex::DISPATCH_RPC_RESPONSE,
0,
server_origin,
call_id,
0,
);
let mut buf = Vec::with_capacity(EVENT_META_SIZE + 64);
buf.extend_from_slice(&meta.to_bytes());
resp.encode_into(&mut buf);
let reply_channel_id = ChannelId::new(reply_channel.clone());
let reply_channel_hash = reply_channel_id.hash();
let reply_stream_id = MeshNode::publish_stream_id(&reply_channel_id);
if let Err(e) = publish_response_to_caller(
&mesh,
caller_origin,
target_hint,
&reply_channel,
reply_channel_hash,
reply_stream_id,
Bytes::from(buf),
ResponseRouteFallback::RosterOnStaleDirect,
)
.await
{
tracing::warn!(error = %e,
caller_origin = format!("{:#x}", caller_origin),
call_id,
"rpc serve_rpc_streaming: chunk publish failed");
}
})
});
let metrics_handle = self.rpc_metrics_arc().for_service(service);
let metrics_for_bridge = Arc::clone(&metrics_handle);
let fold = Arc::new(Mutex::new(
RpcServerStreamingFold::new(handler as Arc<dyn RpcStreamingHandler>, emit)
.with_metrics(metrics_handle),
));
let dispatcher: RpcInboundDispatcher = Arc::new(move |ev| {
let _ = tx.try_send(ev);
});
let Some(registration_id) = self.register_rpc_inbound(channel_hash, dispatcher) else {
return Err(ServeError::AlreadyServing(service.to_string()));
};
self.rpc_local_services_arc().insert(
service.to_string(),
registration_id,
CapabilityVisibility::Public,
);
self.index_self_with_local_services();
let origin_node_cache_for_bridge = Arc::clone(&origin_node_cache);
let mesh_for_bridge = Arc::clone(self);
let service_for_bridge = service.to_string();
let bridge = tokio::spawn(async move {
let tag = format!("nrpc:{}", service_for_bridge);
while let Some(inbound) = rx.recv().await {
match bridge_preflight(
&mesh_for_bridge,
&origin_node_cache_for_bridge,
&inbound,
&service_for_bridge,
&tag,
&metrics_for_bridge,
) {
BridgePreflight::Proceed(frame) => {
if let Err(e) = fold.lock().apply_inbound(&frame) {
tracing::warn!(error = %e, "rpc serve_rpc_streaming: fold apply error");
}
}
BridgePreflight::Drop => continue,
BridgePreflight::Deny {
claimed_origin,
call_id,
from_node,
} => {
metrics_for_bridge
.capability_denied_total
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
emit_capability_denial(
&mesh_for_bridge,
&service_for_bridge,
claimed_origin,
call_id,
from_node,
)
.await;
continue;
}
}
}
});
Ok(ServeHandle {
channel_hash,
registration_id,
service: service.to_string(),
_bridge: bridge,
_response_drain: None,
mesh: Arc::clone(self),
#[cfg(test)]
origin_node_cache: origin_node_cache.clone(),
})
}
pub fn serve_rpc_client_stream<H: RpcClientStreamingHandler>(
self: &Arc<Self>,
service: &str,
handler: Arc<H>,
) -> Result<ServeHandle, ServeError> {
let request_channel = ChannelName::new(&format!("{service}.requests"))
.map_err(|e| ServeError::InvalidServiceName(e.to_string()))?;
let channel_hash = request_channel.hash();
if let Some(registry) = self.channel_configs() {
registry.install_rpc_service_defaults(service)?;
}
let (tx, mut rx) = tokio::sync::mpsc::channel::<RpcInboundEvent>(1024);
let origin_node_cache: RpcOriginNodeCache = Arc::new(BoundedLru::new());
let mesh_for_emit = Arc::clone(self);
let service_for_emit = service.to_string();
let server_origin = self.identity_origin_hash();
let emit_resp_mesh = Arc::clone(&mesh_for_emit);
let emit_resp_service = service_for_emit.clone();
let origin_node_cache_for_emit = Arc::clone(&origin_node_cache);
let emit_resp: RpcResponseEmitter =
Arc::new(move |from_node, caller_origin, call_id, resp| {
let mesh = Arc::clone(&emit_resp_mesh);
let service = emit_resp_service.clone();
let target_hint =
origin_node_cache_for_emit.get((from_node, caller_origin, call_id));
tokio::spawn(async move {
let reply_channel_name = format!("{service}.replies.{caller_origin:016x}");
let reply_channel = match ChannelName::new(&reply_channel_name) {
Ok(c) => c,
Err(e) => {
tracing::warn!(error = %e, channel = %reply_channel_name,
"rpc serve_rpc_client_stream: invalid reply channel name");
return;
}
};
let meta = EventMeta::new(
crate::adapter::net::cortex::DISPATCH_RPC_RESPONSE,
0,
server_origin,
call_id,
0,
);
let mut buf = Vec::with_capacity(EVENT_META_SIZE + 64);
buf.extend_from_slice(&meta.to_bytes());
resp.encode_into(&mut buf);
let reply_channel_id = ChannelId::new(reply_channel.clone());
let reply_channel_hash = reply_channel_id.hash();
let reply_stream_id = MeshNode::publish_stream_id(&reply_channel_id);
if let Err(e) = publish_response_to_caller(
&mesh,
caller_origin,
target_hint,
&reply_channel,
reply_channel_hash,
reply_stream_id,
Bytes::from(buf),
ResponseRouteFallback::RosterOnStaleDirect,
)
.await
{
tracing::warn!(error = %e,
caller_origin = format!("{:#x}", caller_origin),
call_id,
"rpc serve_rpc_client_stream: terminal RESPONSE publish failed");
}
});
origin_node_cache_for_emit.remove((from_node, caller_origin, call_id));
});
let emit_grant = build_request_grant_emitter(
Arc::clone(&mesh_for_emit),
service_for_emit.clone(),
server_origin,
"serve_rpc_client_stream",
);
let metrics_handle = self.rpc_metrics_arc().for_service(service);
let metrics_for_bridge = Arc::clone(&metrics_handle);
let fold = Arc::new(Mutex::new(
RpcStreamingRequestFold::new(handler as Arc<dyn RpcClientStreamingHandler>, emit_resp)
.with_grant_emitter(emit_grant)
.with_metrics(metrics_handle),
));
let dispatcher: RpcInboundDispatcher = Arc::new(move |ev| {
let _ = tx.try_send(ev);
});
let Some(registration_id) = self.register_rpc_inbound(channel_hash, dispatcher) else {
return Err(ServeError::AlreadyServing(service.to_string()));
};
self.rpc_local_services_arc().insert(
service.to_string(),
registration_id,
CapabilityVisibility::Public,
);
self.index_self_with_local_services();
let origin_node_cache_for_bridge = Arc::clone(&origin_node_cache);
let service_for_bridge = service.to_string();
let mesh_for_bridge = Arc::clone(self);
let bridge = tokio::spawn(async move {
let tag = format!("nrpc:{}", service_for_bridge);
while let Some(inbound) = rx.recv().await {
match bridge_preflight(
&mesh_for_bridge,
&origin_node_cache_for_bridge,
&inbound,
&service_for_bridge,
&tag,
&metrics_for_bridge,
) {
BridgePreflight::Proceed(frame) => {
if reject_relayed_flow_controlled_request(
&mesh_for_bridge,
&metrics_for_bridge,
&inbound,
&service_for_bridge,
&tag,
) {
continue;
}
if let Err(e) = fold.lock().apply_inbound(&frame) {
tracing::warn!(error = %e,
"rpc serve_rpc_client_stream: fold apply error");
}
}
BridgePreflight::Drop => continue,
BridgePreflight::Deny {
claimed_origin,
call_id,
from_node,
} => {
metrics_for_bridge
.capability_denied_total
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
emit_capability_denial(
&mesh_for_bridge,
&service_for_bridge,
claimed_origin,
call_id,
from_node,
)
.await;
continue;
}
}
}
});
Ok(ServeHandle {
channel_hash,
registration_id,
service: service.to_string(),
_bridge: bridge,
_response_drain: None,
mesh: Arc::clone(self),
#[cfg(test)]
origin_node_cache: origin_node_cache.clone(),
})
}
pub async fn call_client_stream(
self: &Arc<Self>,
target_node_id: u64,
service: &str,
opts: CallOptions,
) -> Result<ClientStreamCallRaw, RpcError> {
if opts.org_proof_intent.is_some() {
return Err(RpcError::Codec {
direction: CodecDirection::Encode,
message: "org admission (org_proof_intent) is unary-only; use `call` for a \
protected service"
.to_string(),
});
}
if matches!(opts.request_window_initial, Some(0)) {
return Err(RpcError::Codec {
direction: CodecDirection::Encode,
message: "request_window_initial must be None or >= 1; Some(0) deadlocks send"
.to_string(),
});
}
let route = self.rpc_route_or_no_route(target_node_id, service)?;
let self_origin = self.identity_origin_hash();
self.ensure_reply_subscription(
target_node_id,
service,
route.reply_channel.clone(),
route.reply_hash,
)
.await?;
let call_id = mint_random_call_id();
let pending = self.rpc_client_pending();
let (terminal_rx, mut grant_rx) =
pending.register_client_streaming(call_id, target_node_id);
let mut initial_flags = FLAG_RPC_CLIENT_STREAMING_REQUEST;
let mut initial_headers: Vec<(String, Vec<u8>)> = Vec::new();
if let Some(tc) = opts.trace_context.as_ref() {
initial_flags |= FLAG_RPC_PROPAGATE_TRACE;
initial_headers.extend(build_trace_headers(tc));
}
if let Some(window) = opts.request_window_initial {
initial_headers.push((
HEADER_NRPC_REQUEST_WINDOW_INITIAL.to_string(),
window.to_string().into_bytes(),
));
}
initial_headers.extend(opts.request_headers.iter().cloned());
let credit_sem = opts
.request_window_initial
.map(|n| Arc::new(tokio::sync::Semaphore::new(n as usize)));
let grant_pump = credit_sem.as_ref().map(|sem| {
let sem = Arc::clone(sem);
tokio::spawn(async move {
while let Some(credits) = grant_rx.recv().await {
add_request_grant_credits(&sem, credits);
}
})
});
let deadline_ns = opts.deadline.map(instant_to_unix_nanos).unwrap_or(0);
let observer = StreamingObserverState::new(Arc::clone(self), target_node_id, service, 0);
let cancel_keep_alive = arm_stream_cancel(self, &opts, &pending, call_id);
remember_cancel_publish_runtime();
Ok(ClientStreamCallRaw {
mesh: Arc::clone(self),
target_node_id,
request_channel: route.request_channel.clone(),
request_channel_hash: route.request_channel_hash,
request_stream_id: route.request_stream_id,
self_origin,
call_id,
service: service.to_string(),
initial_headers,
initial_flags,
deadline_ns,
credit_sem,
grant_pump,
terminal_rx: Some(terminal_rx),
state: ClientStreamState::JustOpened,
started: Instant::now(),
observer,
_cancel_keep_alive: cancel_keep_alive,
})
}
pub fn serve_rpc_duplex<H: RpcDuplexHandler>(
self: &Arc<Self>,
service: &str,
handler: Arc<H>,
) -> Result<ServeHandle, ServeError> {
let request_channel = ChannelName::new(&format!("{service}.requests"))
.map_err(|e| ServeError::InvalidServiceName(e.to_string()))?;
let channel_hash = request_channel.hash();
if let Some(registry) = self.channel_configs() {
registry.install_rpc_service_defaults(service)?;
}
let (tx, mut rx) = tokio::sync::mpsc::channel::<RpcInboundEvent>(1024);
let origin_node_cache: RpcOriginNodeCache = Arc::new(BoundedLru::new());
let mesh_for_emit = Arc::clone(self);
let service_for_emit = service.to_string();
let server_origin = self.identity_origin_hash();
let emit_resp_mesh = Arc::clone(&mesh_for_emit);
let emit_resp_service = service_for_emit.clone();
let origin_node_cache_for_emit = Arc::clone(&origin_node_cache);
let emit_resp: RpcAsyncResponseEmitter =
Arc::new(move |from_node, caller_origin, call_id, resp| {
let mesh = Arc::clone(&emit_resp_mesh);
let service = emit_resp_service.clone();
let target_hint =
origin_node_cache_for_emit.get((from_node, caller_origin, call_id));
if streaming_response_is_terminal(&resp) {
origin_node_cache_for_emit.remove((from_node, caller_origin, call_id));
}
Box::pin(async move {
let reply_channel_name = format!("{service}.replies.{caller_origin:016x}");
let reply_channel = match ChannelName::new(&reply_channel_name) {
Ok(c) => c,
Err(e) => {
tracing::warn!(error = %e, channel = %reply_channel_name,
"rpc serve_rpc_duplex: invalid reply channel name");
return;
}
};
let meta = EventMeta::new(
crate::adapter::net::cortex::DISPATCH_RPC_RESPONSE,
0,
server_origin,
call_id,
0,
);
let mut buf = Vec::with_capacity(EVENT_META_SIZE + 64);
buf.extend_from_slice(&meta.to_bytes());
resp.encode_into(&mut buf);
let reply_channel_id = ChannelId::new(reply_channel.clone());
let reply_channel_hash = reply_channel_id.hash();
let reply_stream_id = MeshNode::publish_stream_id(&reply_channel_id);
if let Err(e) = publish_response_to_caller(
&mesh,
caller_origin,
target_hint,
&reply_channel,
reply_channel_hash,
reply_stream_id,
Bytes::from(buf),
ResponseRouteFallback::RosterOnStaleDirect,
)
.await
{
tracing::warn!(error = %e,
caller_origin = format!("{:#x}", caller_origin),
call_id,
"rpc serve_rpc_duplex: chunk publish failed");
}
})
});
let emit_grant = build_request_grant_emitter(
Arc::clone(&mesh_for_emit),
service_for_emit.clone(),
server_origin,
"serve_rpc_duplex",
);
let metrics_handle = self.rpc_metrics_arc().for_service(service);
let metrics_for_bridge = Arc::clone(&metrics_handle);
let fold = Arc::new(Mutex::new(
RpcDuplexFold::new(handler as Arc<dyn RpcDuplexHandler>, emit_resp)
.with_grant_emitter(emit_grant)
.with_metrics(metrics_handle),
));
let dispatcher: RpcInboundDispatcher = Arc::new(move |ev| {
let _ = tx.try_send(ev);
});
let Some(registration_id) = self.register_rpc_inbound(channel_hash, dispatcher) else {
return Err(ServeError::AlreadyServing(service.to_string()));
};
self.rpc_local_services_arc().insert(
service.to_string(),
registration_id,
CapabilityVisibility::Public,
);
self.index_self_with_local_services();
let origin_node_cache_for_bridge = Arc::clone(&origin_node_cache);
let service_for_bridge = service.to_string();
let mesh_for_bridge = Arc::clone(self);
let bridge = tokio::spawn(async move {
let tag = format!("nrpc:{}", service_for_bridge);
while let Some(inbound) = rx.recv().await {
match bridge_preflight(
&mesh_for_bridge,
&origin_node_cache_for_bridge,
&inbound,
&service_for_bridge,
&tag,
&metrics_for_bridge,
) {
BridgePreflight::Proceed(frame) => {
if reject_relayed_flow_controlled_request(
&mesh_for_bridge,
&metrics_for_bridge,
&inbound,
&service_for_bridge,
&tag,
) {
continue;
}
if let Err(e) = fold.lock().apply_inbound(&frame) {
tracing::warn!(error = %e,
"rpc serve_rpc_duplex: fold apply error");
}
}
BridgePreflight::Drop => continue,
BridgePreflight::Deny {
claimed_origin,
call_id,
from_node,
} => {
metrics_for_bridge
.capability_denied_total
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
emit_capability_denial(
&mesh_for_bridge,
&service_for_bridge,
claimed_origin,
call_id,
from_node,
)
.await;
continue;
}
}
}
});
Ok(ServeHandle {
channel_hash,
registration_id,
service: service.to_string(),
_bridge: bridge,
_response_drain: None,
mesh: Arc::clone(self),
#[cfg(test)]
origin_node_cache: origin_node_cache.clone(),
})
}
pub async fn call_duplex(
self: &Arc<Self>,
target_node_id: u64,
service: &str,
opts: CallOptions,
) -> Result<DuplexCallRaw, RpcError> {
if opts.org_proof_intent.is_some() {
return Err(RpcError::Codec {
direction: CodecDirection::Encode,
message: "org admission (org_proof_intent) is unary-only; use `call` for a \
protected service"
.to_string(),
});
}
if matches!(opts.request_window_initial, Some(0)) {
return Err(RpcError::Codec {
direction: CodecDirection::Encode,
message: "request_window_initial must be None or >= 1; Some(0) deadlocks send"
.to_string(),
});
}
if matches!(opts.stream_window_initial, Some(0)) {
return Err(RpcError::Codec {
direction: CodecDirection::Encode,
message:
"stream_window_initial must be None or >= 1; Some(0) deadlocks the response pump"
.to_string(),
});
}
let route = self.rpc_route_or_no_route(target_node_id, service)?;
let self_origin = self.identity_origin_hash();
self.ensure_reply_subscription(
target_node_id,
service,
route.reply_channel.clone(),
route.reply_hash,
)
.await?;
let call_id = mint_random_call_id();
let pending = self.rpc_client_pending();
let (chunks_rx, mut grant_rx) = pending.register_duplex(call_id, target_node_id);
let mut initial_flags = FLAG_RPC_CLIENT_STREAMING_REQUEST | FLAG_RPC_STREAMING_RESPONSE;
let mut initial_headers: Vec<(String, Vec<u8>)> = Vec::new();
if let Some(tc) = opts.trace_context.as_ref() {
initial_flags |= FLAG_RPC_PROPAGATE_TRACE;
initial_headers.extend(build_trace_headers(tc));
}
if let Some(window) = opts.request_window_initial {
initial_headers.push((
HEADER_NRPC_REQUEST_WINDOW_INITIAL.to_string(),
window.to_string().into_bytes(),
));
}
if let Some(window) = opts.stream_window_initial {
initial_headers.push((
HEADER_NRPC_STREAM_WINDOW_INITIAL.to_string(),
window.to_string().into_bytes(),
));
}
initial_headers.extend(opts.request_headers.iter().cloned());
let credit_sem = opts
.request_window_initial
.map(|n| Arc::new(tokio::sync::Semaphore::new(n as usize)));
let grant_pump = credit_sem.as_ref().map(|sem| {
let sem = Arc::clone(sem);
tokio::spawn(async move {
while let Some(credits) = grant_rx.recv().await {
add_request_grant_credits(&sem, credits);
}
})
});
let deadline_ns = opts.deadline.map(instant_to_unix_nanos).unwrap_or(0);
let observer = StreamingObserverState::new(Arc::clone(self), target_node_id, service, 0);
let cancel_keep_alive = arm_stream_cancel(self, &opts, &pending, call_id);
remember_cancel_publish_runtime();
let inner = Arc::new(DuplexInner {
mesh: Arc::clone(self),
target_node_id,
request_channel: route.request_channel.clone(),
request_channel_hash: route.request_channel_hash,
request_stream_id: route.request_stream_id,
self_origin,
call_id,
initial_sent: std::sync::atomic::AtomicBool::new(false),
clean_close: std::sync::atomic::AtomicBool::new(false),
observer,
_cancel_keep_alive: Some(cancel_keep_alive),
});
let sink = DuplexSink {
inner: Arc::clone(&inner),
service: service.to_string(),
initial_headers,
initial_flags,
deadline_ns,
credit_sem,
grant_pump,
state: ClientStreamState::JustOpened,
};
let stream = DuplexStream {
inner,
chunks_rx,
done: false,
};
Ok(DuplexCallRaw { sink, stream })
}
pub async fn call_streaming(
self: &Arc<Self>,
target_node_id: u64,
service: &str,
payload: Bytes,
opts: CallOptions,
) -> Result<RpcStream, RpcError> {
if opts.org_proof_intent.is_some() {
return Err(RpcError::Codec {
direction: CodecDirection::Encode,
message: "org admission (org_proof_intent) is unary-only; use `call` for a \
protected service"
.to_string(),
});
}
if matches!(opts.stream_window_initial, Some(0)) {
return Err(RpcError::Codec {
direction: CodecDirection::Encode,
message: "stream_window_initial must be None or >= 1; Some(0) deadlocks the response pump"
.to_string(),
});
}
let route = self.rpc_route_or_no_route(target_node_id, service)?;
let self_origin = self.identity_origin_hash();
self.ensure_reply_subscription(
target_node_id,
service,
route.reply_channel.clone(),
route.reply_hash,
)
.await?;
let call_id = mint_random_call_id();
let pending = self.rpc_client_pending();
let rx = pending.register_streaming(call_id, target_node_id);
let mut flags = FLAG_RPC_STREAMING_RESPONSE;
let mut headers = Vec::new();
if let Some(tc) = opts.trace_context.as_ref() {
flags |= FLAG_RPC_PROPAGATE_TRACE;
headers.extend(build_trace_headers(tc));
}
if let Some(window) = opts.stream_window_initial {
headers.push((
HEADER_NRPC_STREAM_WINDOW_INITIAL.to_string(),
window.to_string().into_bytes(),
));
}
headers.extend(opts.request_headers.iter().cloned());
let req = RpcRequestPayload {
service: service.to_string(),
deadline_ns: opts.deadline.map(instant_to_unix_nanos).unwrap_or(0),
flags,
headers,
body: payload.clone(),
};
let meta = EventMeta::new(DISPATCH_RPC_REQUEST, 0, self_origin, call_id, 0);
let mut buf = Vec::with_capacity(EVENT_META_SIZE + RPC_ROUTE_V1_SIZE + req.body.len() + 32);
buf.extend_from_slice(&meta.to_bytes());
encode_rpc_route(&mut buf, route.request_channel_hash);
req.encode_into(&mut buf);
let payload_bytes = Bytes::from(buf);
if let Err(e) = self
.publish_to_peer(
target_node_id,
route.request_channel_hash,
route.request_stream_id,
true,
std::slice::from_ref(&payload_bytes),
)
.await
{
pending.cancel(call_id);
return Err(RpcError::Transport(e));
}
let request_bytes_len = payload_bytes.len() as u32;
let cancel_keep_alive = arm_stream_cancel(self, &opts, &pending, call_id);
remember_cancel_publish_runtime();
Ok(RpcStream {
mesh: Arc::clone(self),
target_node_id,
request_channel: route.request_channel.clone(),
request_channel_hash: route.request_channel_hash,
request_stream_id: route.request_stream_id,
self_origin,
call_id,
inner: rx,
done: false,
stream_window: opts.stream_window_initial,
grant_pending: 0,
_cancel_keep_alive: cancel_keep_alive,
observer: StreamingObserverState::new(
Arc::clone(self),
target_node_id,
service,
request_bytes_len,
),
})
}
pub fn find_service_nodes(&self, service: &str) -> Vec<u64> {
use crate::adapter::net::behavior::capability::CapabilityFilter;
use crate::adapter::net::behavior::fold::capability_bridge;
let tag = format!("nrpc:{service}");
let filter = CapabilityFilter::default().require_tag(tag);
capability_bridge::find_nodes_matching(self.capability_fold(), &filter)
}
pub fn public_owned_service_providers(&self, service: &str) -> Vec<PublicOwnedProvider> {
use crate::adapter::net::behavior::fold::capability_bridge;
let tag = format!("nrpc:{service}");
let mut out: Vec<PublicOwnedProvider> =
capability_bridge::public_owned_providers(self.capability_fold(), &tag)
.into_iter()
.filter_map(|publisher| {
let pinned = self.peer_entity_id(publisher.node_id)?;
if pinned != publisher.member {
tracing::debug!(
node_id = format!("{:#x}", publisher.node_id),
"dropping exported candidate: the live session pin names a \
different entity than the verified ownership projection"
);
return None;
}
Some(PublicOwnedProvider {
provider: pinned,
owner_org: publisher.owner_org,
})
})
.collect();
out.sort_unstable_by(|a, b| a.provider.as_bytes().cmp(b.provider.as_bytes()));
out
}
pub async fn call_service(
self: &Arc<Self>,
service: &str,
payload: Bytes,
opts: CallOptions,
) -> Result<RpcReply, RpcError> {
let mut candidates = self.find_service_nodes(service);
if candidates.is_empty() {
return Err(RpcError::NoRoute {
target: 0,
reason: format!(
"no nodes advertise `nrpc:{service}` (have any servers \
for this service called serve_rpc + announce_capabilities?)"
),
});
}
if opts.filter_unhealthy {
let proximity = self.proximity_graph();
candidates.retain(|node_id| match self.entity_id_for_node(*node_id) {
Some(entity_id) => match proximity.get_node(&entity_id) {
Some(node) => node.is_available(),
None => true, },
None => true, });
if candidates.is_empty() {
return Err(RpcError::NoRoute {
target: 0,
reason: format!(
"every node advertising `nrpc:{service}` is marked \
unhealthy by the local proximity graph",
),
});
}
}
candidates.sort_unstable();
if let Some(intent) = opts.org_proof_intent.as_ref() {
candidates
.retain(|node_id| self.peer_entity_id(*node_id).as_ref() == Some(&intent.provider));
if candidates.is_empty() {
return Err(RpcError::Codec {
direction: CodecDirection::Encode,
message: format!(
"org admission: no advertising candidate for `nrpc:{service}` matches \
the proof's bound provider"
),
});
}
} else {
let tag = format!("nrpc:{service}");
use crate::adapter::net::behavior::fold::capability_bridge;
let self_id = self.node_id();
let any_candidate = candidates[0];
let fold = self.capability_fold();
let verdicts = capability_bridge::may_execute_batch(fold, &candidates, &tag, self_id);
let mut iter = verdicts.into_iter();
candidates.retain(|_| iter.next().unwrap_or(false));
if candidates.is_empty() {
return Err(RpcError::CapabilityDenied {
target: any_candidate,
capability: service.to_string(),
});
}
}
let target = self.select_target(&candidates, &opts.routing_policy);
self.call(target, service, payload, opts).await
}
pub async fn call_service_streaming(
self: &Arc<Self>,
service: &str,
payload: Bytes,
opts: CallOptions,
) -> Result<RpcStream, RpcError> {
if opts.org_proof_intent.is_some() {
return Err(RpcError::Codec {
direction: CodecDirection::Encode,
message: "org admission (org_proof_intent) is unary-only; use `call_service` \
for a protected service"
.to_string(),
});
}
let mut candidates = self.find_service_nodes(service);
if candidates.is_empty() {
return Err(RpcError::NoRoute {
target: 0,
reason: format!(
"no nodes advertise `nrpc:{service}` (have any servers \
for this service called serve_rpc + announce_capabilities?)"
),
});
}
if opts.filter_unhealthy {
let proximity = self.proximity_graph();
candidates.retain(|node_id| match self.entity_id_for_node(*node_id) {
Some(entity_id) => match proximity.get_node(&entity_id) {
Some(node) => node.is_available(),
None => true,
},
None => true,
});
if candidates.is_empty() {
return Err(RpcError::NoRoute {
target: 0,
reason: format!(
"every node advertising `nrpc:{service}` is marked \
unhealthy by the local proximity graph",
),
});
}
}
candidates.sort_unstable();
let tag = format!("nrpc:{service}");
use crate::adapter::net::behavior::fold::capability_bridge;
let self_id = self.node_id();
let any_candidate = candidates[0];
let fold = self.capability_fold();
let verdicts = capability_bridge::may_execute_batch(fold, &candidates, &tag, self_id);
let mut iter = verdicts.into_iter();
candidates.retain(|_| iter.next().unwrap_or(false));
if candidates.is_empty() {
return Err(RpcError::CapabilityDenied {
target: any_candidate,
capability: service.to_string(),
});
}
let target = self.select_target(&candidates, &opts.routing_policy);
self.call_streaming(target, service, payload, opts).await
}
fn select_target(&self, candidates: &[u64], policy: &RoutingPolicy) -> u64 {
match policy {
RoutingPolicy::RoundRobin => {
let n = self
.rpc_round_robin_cursor_arc()
.fetch_add(1, Ordering::Relaxed);
candidates[(n as usize) % candidates.len()]
}
RoutingPolicy::Random => {
let n = self
.rpc_round_robin_cursor_arc()
.fetch_add(1, Ordering::Relaxed);
let mixed = xxhash_rust::xxh3::xxh3_64(&n.to_le_bytes());
candidates[(mixed as usize) % candidates.len()]
}
RoutingPolicy::Sticky { key } => {
let h = xxhash_rust::xxh3::xxh3_64(&key.to_le_bytes());
candidates[(h as usize) % candidates.len()]
}
RoutingPolicy::LowestLatency => {
let proximity = self.proximity_graph();
let mut best_node = candidates[0];
let mut best_latency = u64::MAX;
for &node_id in candidates {
let lat = self
.entity_id_for_node(node_id)
.and_then(|eid| proximity.get_node(&eid))
.map(|n| n.latency_us)
.unwrap_or(u64::MAX);
if lat < best_latency {
best_latency = lat;
best_node = node_id;
}
}
best_node
}
}
}
pub async fn call(
self: &Arc<Self>,
target_node_id: u64,
service: &str,
payload: Bytes,
mut opts: CallOptions,
) -> Result<RpcReply, RpcError> {
let started_total = Instant::now();
let request_bytes_len = payload.len() as u32;
let route = self.rpc_route_or_no_route(target_node_id, service)?;
let self_origin = self.identity_origin_hash();
let call_id = mint_random_call_id();
let (flags, mut headers) = match opts.trace_context.as_ref() {
Some(tc) => (FLAG_RPC_PROPAGATE_TRACE, build_trace_headers(tc)),
None => (0u16, Vec::new()),
};
headers.append(&mut opts.request_headers);
let mut req = RpcRequestPayload {
service: service.to_string(),
deadline_ns: opts.deadline.map(instant_to_unix_nanos).unwrap_or(0),
flags,
headers,
body: payload.clone(),
};
if let Some(intent) = opts.org_proof_intent.as_ref() {
match self.peer_entity_id(target_node_id) {
Some(pinned) if pinned == intent.provider => {}
Some(_) => {
return Err(RpcError::Codec {
direction: CodecDirection::Encode,
message: format!(
"org admission: proof provider does not match the pinned entity \
of target {target_node_id:#x}"
),
});
}
None => {
return Err(RpcError::Codec {
direction: CodecDirection::Encode,
message: format!(
"org admission: target {target_node_id:#x} has no pinned entity \
to bind the proof to"
),
});
}
}
if req
.headers
.iter()
.any(|(name, _)| name == ORG_ADMISSION_HEADER)
{
return Err(RpcError::Codec {
direction: CodecDirection::Encode,
message: "org admission: request already carries a net-org-admission header"
.to_string(),
});
}
let header = sign_admission_proof(intent, call_id, &req)?;
req.headers.push(header);
req.validate_wire_bounds().map_err(|e| RpcError::Codec {
direction: CodecDirection::Encode,
message: format!("org admission: finalized request exceeds wire bounds: {e}"),
})?;
}
let metrics_registry = self.rpc_metrics_arc();
let mut metrics_guard = CallMetricsGuard::new(metrics_registry.for_service(service));
if let Err(e) = self
.ensure_reply_subscription(
target_node_id,
service,
route.reply_channel.clone(),
route.reply_hash,
)
.await
{
metrics_guard.record(CallOutcome::NoRoute);
self.fire_rpc_observer_outbound(
target_node_id,
service,
started_total.elapsed().as_millis() as u32,
crate::adapter::net::cortex::rpc_observer::RpcCallStatus::Error(e.to_string()),
request_bytes_len,
0,
);
return Err(e);
}
let pending = self.rpc_client_pending();
let rx = pending.register(call_id, target_node_id);
let meta = EventMeta::new(DISPATCH_RPC_REQUEST, 0, self_origin, call_id, 0);
let mut buf = Vec::with_capacity(EVENT_META_SIZE + RPC_ROUTE_V1_SIZE + req.body.len() + 32);
buf.extend_from_slice(&meta.to_bytes());
encode_rpc_route(&mut buf, route.request_channel_hash);
req.encode_into(&mut buf);
let started = Instant::now();
let payload_bytes = Bytes::from(buf);
if let Err(e) = self
.publish_to_peer(
target_node_id,
route.request_channel_hash,
route.request_stream_id,
true,
std::slice::from_ref(&payload_bytes),
)
.await
{
pending.cancel(call_id);
let err = if classify_publish_no_session(&e) {
metrics_guard.record(CallOutcome::NoRoute);
RpcError::NoRoute {
target: target_node_id,
reason: e.to_string(),
}
} else {
metrics_guard.record(CallOutcome::Transport);
RpcError::Transport(e)
};
self.fire_rpc_observer_outbound(
target_node_id,
service,
started_total.elapsed().as_millis() as u32,
crate::adapter::net::cortex::rpc_observer::RpcCallStatus::Error(err.to_string()),
request_bytes_len,
0,
);
return Err(err);
}
remember_cancel_publish_runtime();
let mut guard = UnaryCallGuard {
pending: Arc::clone(&pending),
mesh: Arc::clone(self),
target_node_id,
request_channel: route.request_channel.clone(),
self_origin,
call_id,
completed: false,
};
let cancel_token = opts.cancel_token.unwrap_or(0);
let cancel_notify = self.cancel_registry().register_notify(cancel_token);
let outcome: Result<Result<RpcResponsePayload, _>, tokio::time::error::Elapsed> =
match opts.deadline {
None => {
tokio::select! {
biased;
_ = cancel_notify.notified() => {
return Err(fire_unary_cancel_outcome(
self,
&mut metrics_guard,
cancel_token,
target_node_id,
service,
started_total,
request_bytes_len,
));
}
r = rx => Ok(r),
}
}
Some(deadline) => {
let timeout_at = deadline.saturating_duration_since(Instant::now());
tokio::select! {
biased;
_ = cancel_notify.notified() => {
return Err(fire_unary_cancel_outcome(
self,
&mut metrics_guard,
cancel_token,
target_node_id,
service,
started_total,
request_bytes_len,
));
}
r = tokio::time::timeout(timeout_at, rx) => r,
}
}
};
self.cancel_registry().release(cancel_token);
let resp = match outcome {
Ok(Ok(resp)) => {
guard.completed = true;
resp
}
Ok(Err(_recv_err)) => {
guard.completed = true;
metrics_guard.record(CallOutcome::Transport);
let err = RpcError::Transport(AdapterError::Connection(
"rpc client pending sender dropped (no response will arrive)".into(),
));
self.fire_rpc_observer_outbound(
target_node_id,
service,
started_total.elapsed().as_millis() as u32,
crate::adapter::net::cortex::rpc_observer::RpcCallStatus::Error(
err.to_string(),
),
request_bytes_len,
0,
);
return Err(err);
}
Err(_elapsed) => {
metrics_guard.record(CallOutcome::Timeout);
self.fire_rpc_observer_outbound(
target_node_id,
service,
started_total.elapsed().as_millis() as u32,
crate::adapter::net::cortex::rpc_observer::RpcCallStatus::Timeout,
request_bytes_len,
0,
);
return Err(RpcError::Timeout {
elapsed_ms: started.elapsed().as_millis() as u64,
});
}
};
if resp.status.is_ok() {
metrics_guard.record(CallOutcome::Ok);
let response_bytes_len = resp.body.len() as u32;
self.fire_rpc_observer_outbound(
target_node_id,
service,
started_total.elapsed().as_millis() as u32,
crate::adapter::net::cortex::rpc_observer::RpcCallStatus::Ok,
request_bytes_len,
response_bytes_len,
);
Ok(RpcReply {
body: resp.body,
headers: resp.headers,
latency_ns: started.elapsed().as_nanos() as u64,
})
} else {
metrics_guard.record(CallOutcome::ServerError);
let status = resp.status.to_wire();
let response_bytes_len = resp.body.len() as u32;
let message = String::from_utf8(resp.body.to_vec())
.unwrap_or_else(|e| format!("<{} bytes of non-utf8 body>", e.into_bytes().len()));
self.fire_rpc_observer_outbound(
target_node_id,
service,
started_total.elapsed().as_millis() as u32,
crate::adapter::net::cortex::rpc_observer::RpcCallStatus::Error(message.clone()),
request_bytes_len,
response_bytes_len,
);
if matches!(resp.status, RpcStatus::CapabilityDenied) {
return Err(RpcError::CapabilityDenied {
target: target_node_id,
capability: service.to_string(),
});
}
Err(RpcError::ServerError {
status,
message,
headers: resp.headers,
})
}
}
async fn ensure_reply_subscription(
self: &Arc<Self>,
target_node_id: u64,
service: &str,
reply_channel: ChannelName,
reply_hash: ChannelHash,
) -> Result<(), RpcError> {
let registry = self.rpc_reply_subscriptions_arc();
let service_hash = xxhash_rust::xxh3::xxh3_64(service.as_bytes());
if reply_subscription_covers(®istry, target_node_id, service_hash, service) {
return Ok(());
}
let mut gen_before: Option<u64> = None;
if registry.len() >= MAX_REPLY_SUBSCRIPTIONS {
return Err(RpcError::NoRoute {
target: target_node_id,
reason: format!(
"reply-subscription registry at cap ({} entries); refusing new \
(target={target_node_id:#x}, service={service:?}). Caller should \
reuse an existing target+service pair or shrink the active set.",
MAX_REPLY_SUBSCRIPTIONS,
),
});
}
let mut last_err = None;
for attempt in 0..REPLY_SUBSCRIBE_ATTEMPTS {
let gen_this_attempt = self.peer_failure_generation(target_node_id);
match self
.subscribe_channel_reporting_reason(target_node_id, reply_channel.clone())
.await
{
Ok(()) => {
gen_before = Some(gen_this_attempt);
last_err = None;
break;
}
Err(failure) => {
let retryable = failure.warrants_reannounce();
last_err = Some(failure.into_adapter_error());
if !retryable || attempt + 1 == REPLY_SUBSCRIBE_ATTEMPTS {
break;
}
if self.claim_corrective_announce(target_node_id) {
if let Err(e) = self.reannounce_for_authorization().await {
tracing::debug!(
target = format!("{target_node_id:#x}"),
error = %e,
"corrective capability announce failed to send; \
releasing the claim so a later call can retry"
);
self.release_corrective_announce(target_node_id);
}
}
tokio::time::sleep(REPLY_SUBSCRIBE_BACKOFF * (attempt as u32 + 1)).await;
}
}
}
if let Some(e) = last_err {
return Err(RpcError::NoRoute {
target: target_node_id,
reason: format!(
"reply-channel subscribe rejected by {target_node_id:#x} ({e}). \
The reply channel is bound to this node's announced identity; \
if this persists, the target has not pinned our EntityId \
(no signature-verified direct capability announcement has \
been received from us)."
),
});
}
if !self.rpc_inbound_dispatcher_registered(reply_hash) {
let pending = self.rpc_client_pending();
let fold = Arc::new(Mutex::new(RpcClientFold::new(pending)));
let dispatcher: RpcInboundDispatcher = Arc::new(move |ev| {
fold.lock().apply_inbound(&ev);
});
let _ = self.register_rpc_inbound(reply_hash, dispatcher);
}
let _ = reply_hash; let Some(gen_before) = gen_before else {
return Err(RpcError::NoRoute {
target: target_node_id,
reason: format!(
"reply-channel subscribe to {target_node_id:#x} completed with \
no recorded outcome; refusing to cache an unverified \
subscription. Retry — the next call re-subscribes."
),
});
};
registry.insert(
(target_node_id, service_hash),
ReplySubscription {
service: Arc::from(service),
written_at_generation: gen_before,
},
);
if self.peer_failure_generation(target_node_id) != gen_before {
registry.remove_if(&(target_node_id, service_hash), |_, entry| {
entry.written_at_generation == gen_before
});
return Err(RpcError::NoRoute {
target: target_node_id,
reason: format!(
"session with {target_node_id:#x} failed while subscribing to the \
reply channel; the target dropped our roster entry, so the \
subscribe did not survive. Retry — the next call re-subscribes."
),
});
}
Ok(())
}
}
fn reply_subscription_covers(
registry: &dashmap::DashMap<(u64, u64), ReplySubscription>,
target_node_id: u64,
service_hash: u64,
service: &str,
) -> bool {
registry
.get(&(target_node_id, service_hash))
.is_some_and(|entry| entry.value().service.as_ref() == service)
}
pub const MAX_REPLY_SUBSCRIPTIONS: usize = 1024;
const REPLY_SUBSCRIBE_ATTEMPTS: usize = 3;
const REPLY_SUBSCRIBE_BACKOFF: std::time::Duration = std::time::Duration::from_millis(8);
fn sign_admission_proof(
intent: &OrgProofIntent,
call_id: u64,
req: &RpcRequestPayload,
) -> Result<(String, Vec<u8>), RpcError> {
let expected = CapabilityAuthorityId::for_tag(&format!("nrpc:{}", req.service));
if intent.capability != expected {
return Err(RpcError::Codec {
direction: CodecDirection::Encode,
message: format!(
"org admission: intent capability does not match the invoked service `{}`",
req.service
),
});
}
if intent.proof_ttl_secs == 0 || intent.proof_ttl_secs > MAX_ORG_PROOF_TTL_SECS {
return Err(RpcError::Codec {
direction: CodecDirection::Encode,
message: format!(
"org admission: proof TTL {}s out of range (1..={MAX_ORG_PROOF_TTL_SECS})",
intent.proof_ttl_secs
),
});
}
let digest = org_request_digest(req).map_err(|e| RpcError::Codec {
direction: CodecDirection::Encode,
message: format!("org admission: request digest failed: {e}"),
})?;
let ttl_secs = intent.proof_ttl_secs;
let now_ns = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos() as u64)
.unwrap_or(0);
let expiry = now_ns.saturating_add(ttl_secs.saturating_mul(1_000_000_000));
let proof = OrgCallProof::sign_for_call(
&intent.caller,
intent.membership.clone(),
intent.dispatcher.clone(),
intent.capability_grant.clone(),
intent.acting_org,
intent.provider_owner_org,
intent.provider.clone(),
call_id,
intent.capability,
expiry,
digest,
);
let bytes = proof.encode().map_err(|e| RpcError::Codec {
direction: CodecDirection::Encode,
message: format!("org admission: proof encode failed: {e}"),
})?;
Ok((ORG_ADMISSION_HEADER.to_string(), bytes))
}
fn mint_random_call_id() -> u64 {
thread_local! {
static CALL_ID_ENTROPY_POOL: std::cell::RefCell<([u8; CALL_ID_ENTROPY_POOL_BYTES], usize)> = const {
std::cell::RefCell::new(([0u8; CALL_ID_ENTROPY_POOL_BYTES], CALL_ID_ENTROPY_POOL_BYTES))
};
}
CALL_ID_ENTROPY_POOL.with(|cell| {
let mut pool = cell.borrow_mut();
let (buf, cursor) = &mut *pool;
if *cursor >= CALL_ID_ENTROPY_POOL_BYTES {
if getrandom::fill(buf).is_err() {
static CALL_ID_FALLBACK: std::sync::atomic::AtomicU64 =
std::sync::atomic::AtomicU64::new(1);
let id = CALL_ID_FALLBACK.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
return if id == 0 { 1 } else { id };
}
*cursor = 0;
}
let mut id = [0u8; 8];
id.copy_from_slice(&buf[*cursor..*cursor + 8]);
*cursor += 8;
match u64::from_le_bytes(id) {
0 => 1,
id => id,
}
})
}
const CALL_ID_ENTROPY_POOL_BYTES: usize = 64 * 8;
impl MeshNode {
fn rpc_client_pending(&self) -> Arc<crate::adapter::net::cortex::RpcClientPending> {
self.rpc_client_pending_arc()
}
fn identity_origin_hash(&self) -> u64 {
self.public_key_origin_hash()
}
fn rpc_route_or_no_route(
&self,
target_node_id: u64,
service: &str,
) -> Result<Arc<super::mesh::RpcRoute>, RpcError> {
self.rpc_route_for_service(service)
.map_err(|reason| RpcError::NoRoute {
target: target_node_id,
reason,
})
}
}
#[derive(Debug, thiserror::Error)]
pub enum ServeError {
#[error("invalid service name: {0}")]
InvalidServiceName(String),
#[error("already serving service `{0}` on this node")]
AlreadyServing(String),
#[error(
"tool `{0}` announces pricing_terms but this serve path cannot enforce payment — \
serve paid tools via Mesh::serve_tool_paid, or publish via \
ServerPublisher::publish_tools with payment_admission"
)]
UnenforceablePricing(String),
#[error(
"tool `{0}` is served through the payment gate but announces no pricing_terms — \
attach terms to the descriptor, or serve it free via Mesh::serve_tool"
)]
MissingPricingTerms(String),
#[error(
"protected service `{0}` requires an installed node authority; adopt one before serving"
)]
ProtectedAuthorityRequired(String),
#[error("invalid protected registration: {0}")]
InvalidProtectedRegistration(String),
#[error("subnet export not authorized: {0}")]
SubnetExportUnauthorized(String),
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PublicOwnedProvider {
pub provider: crate::adapter::net::identity::EntityId,
pub owner_org: crate::adapter::net::behavior::org::OrgId,
}
enum UnaryAdmission {
Public,
Protected {
admission: OrgAdmission,
provider_policy: OrgProviderPolicy,
},
OwnerScoped { provider_policy: OrgProviderPolicy },
Granted { provider_policy: OrgProviderPolicy },
SubnetExported {
admission: OrgAdmission,
export: SubnetExportBinding,
provider_policy: OrgProviderPolicy,
},
#[cfg(test)]
ProtectedRedWitnessDisabled {
admission: OrgAdmission,
provider_policy: OrgProviderPolicy,
},
}
impl UnaryAdmission {
fn visibility(&self) -> CapabilityVisibility {
match self {
Self::Public | Self::Protected { .. } | Self::SubnetExported { .. } => {
CapabilityVisibility::Public
}
Self::OwnerScoped { .. } => CapabilityVisibility::OwnerScoped,
Self::Granted { .. } => CapabilityVisibility::GrantedAudience,
#[cfg(test)]
Self::ProtectedRedWitnessDisabled { .. } => CapabilityVisibility::Public,
}
}
fn response_route_fallback(&self) -> ResponseRouteFallback {
match self {
Self::Public => ResponseRouteFallback::RosterOnStaleDirect,
Self::Protected { .. }
| Self::OwnerScoped { .. }
| Self::Granted { .. }
| Self::SubnetExported { .. } => ResponseRouteFallback::DirectOnly,
#[cfg(test)]
Self::ProtectedRedWitnessDisabled { .. } => ResponseRouteFallback::DirectOnly,
}
}
}
#[derive(Debug, thiserror::Error)]
pub enum TypedCallError {
#[error("transport: {0}")]
Transport(#[from] RpcError),
#[error("codec: {0}")]
Codec(String),
}
impl From<postcard::Error> for TypedCallError {
fn from(e: postcard::Error) -> Self {
Self::Codec(e.to_string())
}
}
pub async fn typed_call<Req, Resp>(
mesh: &std::sync::Arc<crate::adapter::net::MeshNode>,
target_node_id: u64,
service: &str,
request: &Req,
deadline: std::time::Duration,
) -> Result<Resp, TypedCallError>
where
Req: serde::Serialize,
Resp: serde::de::DeserializeOwned,
{
let body = postcard::to_allocvec(request)?;
let opts = CallOptions {
deadline: Some(std::time::Instant::now() + deadline),
..Default::default()
};
let reply = mesh
.call(target_node_id, service, Bytes::from(body), opts)
.await?;
Ok(postcard::from_bytes(&reply.body)?)
}
fn classify_publish_no_session(err: &AdapterError) -> bool {
match err {
AdapterError::Connection(msg) => {
msg.contains("no session for subscriber") || msg.contains("no session to publisher")
}
_ => false,
}
}
fn instant_to_unix_nanos(instant: Instant) -> u64 {
let now_instant = Instant::now();
let now_wall = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos() as u64)
.unwrap_or(0);
if instant >= now_instant {
let delta = instant.duration_since(now_instant);
now_wall.saturating_add(delta.as_nanos() as u64)
} else {
let delta = now_instant.duration_since(instant);
now_wall.saturating_sub(delta.as_nanos() as u64)
}
}
#[allow(dead_code)]
fn _ensure_send_sync() {
fn assert_send_sync<T: Send + Sync>() {}
assert_send_sync::<ServeHandle>();
assert_send_sync::<RpcCancellationToken>();
assert_send_sync::<RpcContext>();
assert_send_sync::<RpcHandlerError>();
assert_send_sync::<RpcStatus>();
assert_send_sync::<RpcReply>();
assert_send_sync::<CallOptions>();
}
#[cfg(test)]
mod reply_subscribe_retry_tests {
#[test]
fn the_announce_latch_does_not_gate_the_retry() {
let src = include_str!("mesh_rpc.rs");
let start = src
.find(" async fn ensure_reply_subscription(")
.expect("ensure_reply_subscription must exist");
let end = src
.find("\n#[cfg(test)]")
.expect("this file has test modules");
assert!(
start < end,
"ensure_reply_subscription moved below the test modules; this scan \
would read its own source"
);
let body = &src[start..end];
let retryable = body
.find("let retryable = ")
.expect("the retry decision must still be a named binding");
let stmt_end = body[retryable..]
.find(';')
.expect("the retryable binding must terminate");
let binding = &body[retryable..retryable + stmt_end];
assert!(
!binding.contains("claim_corrective_announce"),
"regression: the once-per-target announce latch is back inside the \
retry condition ({binding:?}). Attempt 1 would then read a spent \
latch as a permanent denial and abandon the announcement attempt \
0 had just sent. The latch bounds ANNOUNCES, not attempts."
);
assert!(
binding.contains("warrants_reannounce"),
"the retry condition must remain the `Unauthorized` test — \
retrying anything else cannot change the answer"
);
assert!(
body.contains("if self.claim_corrective_announce(target_node_id) {"),
"regression: the corrective announce must stay behind the \
once-per-target latch, or one persistently-denying target turns \
every RPC into extra rate-limit-bypassing capability broadcasts"
);
}
}
#[cfg(test)]
mod origin_cache_tests {
use super::*;
#[test]
fn response_route_trust_requires_authenticated_direct_origin() {
let victim_origin = 0x1111_2222_3333_4444u64;
let peer_node = 0xABCDu64;
assert!(response_route_is_trustworthy(
peer_node,
Some(DISPATCH_RPC_REQUEST),
victim_origin,
Some(victim_origin),
));
assert!(!response_route_is_trustworthy(
peer_node,
Some(DISPATCH_RPC_REQUEST),
victim_origin,
Some(0x9999_9999_9999_9999),
));
for dispatch in [
DISPATCH_RPC_CANCEL,
DISPATCH_RPC_REQUEST_CHUNK,
DISPATCH_RPC_REQUEST_GRANT,
] {
assert!(!response_route_is_trustworthy(
peer_node,
Some(dispatch),
victim_origin,
Some(victim_origin),
));
}
assert!(!response_route_is_trustworthy(
peer_node,
Some(DISPATCH_RPC_REQUEST),
victim_origin,
None,
));
assert!(!response_route_is_trustworthy(
0,
Some(DISPATCH_RPC_REQUEST),
victim_origin,
Some(victim_origin),
));
assert!(!response_route_is_trustworthy(
peer_node,
None,
victim_origin,
Some(victim_origin),
));
}
#[test]
fn origin_keyed_lru_bounds_under_crafted_origin_flood() {
let cache: BoundedLru<(u64, u64), u64> = BoundedLru::new();
let flood = (RPC_CALLER_CACHE_CAP as u64) * 4;
for origin in 0..flood {
cache.insert((origin, origin), origin);
}
assert_eq!(
cache.0.lock().len(),
RPC_CALLER_CACHE_CAP,
"cache must stay at its capacity bound under a crafted-origin flood"
);
assert_eq!(cache.get((flood - 1, flood - 1)), Some(flood - 1));
assert_eq!(cache.get((0, 0)), None);
}
#[test]
fn mint_random_call_id_produces_distinct_values_across_thousands_of_calls() {
let mut seen = std::collections::HashSet::with_capacity(100_000);
for _ in 0..100_000 {
let id = super::mint_random_call_id();
assert_ne!(id, 0, "fallback-zero path triggered unexpectedly");
assert!(seen.insert(id), "duplicate call_id minted: {:#x}", id);
}
}
#[test]
fn mint_random_call_id_set_bit_density_is_balanced() {
let n = 10_000u64;
let mut total_set: u64 = 0;
for _ in 0..n {
total_set += super::mint_random_call_id().count_ones() as u64;
}
let bits_total = n * 64;
let fraction = total_set as f64 / bits_total as f64;
assert!(
(fraction - 0.5).abs() < 0.02,
"set-bit density {} is too far from 0.5 — pool may be mismanaged",
fraction
);
}
#[test]
fn reply_subscriptions_keyed_by_target_and_service_hash() {
use dashmap::DashMap;
let registry: DashMap<(u64, u64), ReplySubscription> = DashMap::new();
let h_a = xxhash_rust::xxh3::xxh3_64(b"svc-a");
let h_b = xxhash_rust::xxh3::xxh3_64(b"svc-b");
registry.insert(
(0xAA, h_a),
ReplySubscription {
service: Arc::from("svc-a"),
written_at_generation: 0,
},
);
registry.insert(
(0xAA, h_b),
ReplySubscription {
service: Arc::from("svc-b"),
written_at_generation: 0,
},
);
assert!(super::reply_subscription_covers(
®istry, 0xAA, h_a, "svc-a"
));
assert!(super::reply_subscription_covers(
®istry, 0xAA, h_b, "svc-b"
));
assert!(!super::reply_subscription_covers(
®istry, 0xBB, h_a, "svc-a"
));
registry.insert(
(0xBB, h_a),
ReplySubscription {
service: Arc::from("svc-a"),
written_at_generation: 0,
},
);
assert!(super::reply_subscription_covers(
®istry, 0xBB, h_a, "svc-a"
));
registry.insert(
(0xAA, h_a),
ReplySubscription {
service: Arc::from("svc-a"),
written_at_generation: 0,
},
);
assert!(super::reply_subscription_covers(
®istry, 0xAA, h_a, "svc-a"
));
assert_eq!(registry.len(), 3);
assert!(
!super::reply_subscription_covers(®istry, 0xAA, h_a, "svc-evil"),
"hash collision must not satisfy the membership check for a \
different service name"
);
registry.insert(
(0xAA, h_a),
ReplySubscription {
service: Arc::from("svc-evil"),
written_at_generation: 0,
},
);
assert!(super::reply_subscription_covers(
®istry, 0xAA, h_a, "svc-evil"
));
assert!(!super::reply_subscription_covers(
®istry, 0xAA, h_a, "svc-a"
));
}
#[test]
fn the_fence_rollback_spares_a_newer_calls_entry() {
use dashmap::DashMap;
let registry: DashMap<(u64, u64), ReplySubscription> = DashMap::new();
let key = (0xAAu64, xxhash_rust::xxh3::xxh3_64(b"svc"));
registry.insert(
key,
ReplySubscription {
service: Arc::from("svc"),
written_at_generation: 0,
},
);
registry.insert(
key,
ReplySubscription {
service: Arc::from("svc"),
written_at_generation: 1,
},
);
registry.remove_if(&key, |_, entry| entry.written_at_generation == 0);
assert!(
super::reply_subscription_covers(®istry, key.0, key.1, "svc"),
"the late rollback deleted a newer call's entry; that caller is \
genuinely subscribed and will now re-subscribe for nothing"
);
registry.remove_if(&key, |_, entry| entry.written_at_generation == 1);
assert!(
!super::reply_subscription_covers(®istry, key.0, key.1, "svc"),
"a call must still be able to roll back its own entry — leaving it \
is the stale-cache bug the fence exists to prevent"
);
}
#[test]
fn auto_grant_coalescing_never_starves_the_server_pump() {
for window in 1u32..=64 {
let chunks = 1_000u32;
let mut server_credits = window as u64;
let mut pending = 0u32;
let mut sent = 0u32;
let mut delivered = 0u32;
let mut grants = 0u32;
while delivered < chunks {
while server_credits > 0 && sent < chunks {
server_credits -= 1;
sent += 1;
}
assert!(
sent > delivered,
"window {window}: server starved while the consumer is actively \
polling (credits {server_credits}, pending {pending}, \
sent {sent}, delivered {delivered})"
);
delivered += 1;
if let Some(amount) = super::accumulate_auto_grant(&mut pending, window) {
grants += 1;
server_credits += amount as u64;
}
}
let threshold = (window / 2).max(1);
assert!(
grants <= chunks / threshold + 1,
"window {window}: {grants} grant packets exceeds the \
coalesced cadence bound of {}",
chunks / threshold + 1
);
if window >= 4 {
assert!(
grants < chunks,
"window {window}: coalescing must emit fewer grants than chunks"
);
}
}
}
#[test]
fn origin_keyed_lru_get_promotes_to_mru() {
let cache: BoundedLru<(u64, u64), u64> = BoundedLru::new();
for origin in 0..(RPC_CALLER_CACHE_CAP as u64) {
cache.insert((origin, 0), origin);
}
assert_eq!(cache.get((0, 0)), Some(0));
cache.insert((u64::MAX, 0), 1);
assert_eq!(
cache.get((0, 0)),
Some(0),
"touched entry must survive eviction"
);
assert_eq!(
cache.get((1, 0)),
None,
"the now-LRU entry (1) must be evicted"
);
}
#[test]
fn response_route_cache_is_session_scoped_and_retires_per_call() {
let cache: BoundedLru<(u64, u64, u64), u64> = BoundedLru::new();
const ORIGIN: u64 = 0xAA;
const CALL: u64 = 7; const NODE_A: u64 = 0x10;
const NODE_B: u64 = 0x20;
cache.insert((NODE_A, ORIGIN, CALL), NODE_A);
cache.insert((NODE_B, ORIGIN, CALL), NODE_B);
assert_eq!(cache.get((NODE_A, ORIGIN, CALL)), Some(NODE_A));
assert_eq!(cache.get((NODE_B, ORIGIN, CALL)), Some(NODE_B));
cache.remove((NODE_A, ORIGIN, CALL));
assert_eq!(cache.get((NODE_A, ORIGIN, CALL)), None);
assert_eq!(cache.get((NODE_B, ORIGIN, CALL)), Some(NODE_B));
}
#[test]
fn streaming_terminal_detection_recognizes_end_and_errors() {
use crate::adapter::net::cortex::{
HEADER_NRPC_STREAMING, HEADER_NRPC_STREAMING_CONTINUE, HEADER_NRPC_STREAMING_END,
};
let continue_chunk = RpcResponsePayload {
status: RpcStatus::Ok,
headers: vec![(
HEADER_NRPC_STREAMING.to_string(),
HEADER_NRPC_STREAMING_CONTINUE.to_vec(),
)],
body: Bytes::from_static(b"chunk"),
};
assert!(
!streaming_response_is_terminal(&continue_chunk),
"a continue chunk must NOT be treated as terminal",
);
let end_chunk = RpcResponsePayload {
status: RpcStatus::Ok,
headers: vec![(
HEADER_NRPC_STREAMING.to_string(),
HEADER_NRPC_STREAMING_END.to_vec(),
)],
body: Bytes::new(),
};
assert!(
streaming_response_is_terminal(&end_chunk),
"the nrpc-streaming end marker is terminal",
);
let error_frame = RpcResponsePayload {
status: RpcStatus::Internal,
headers: vec![],
body: Bytes::from_static(b"boom"),
};
assert!(
streaming_response_is_terminal(&error_frame),
"a non-Ok status is terminal",
);
}
}
#[cfg(test)]
mod roster_fallback_tests {
use super::*;
use crate::adapter::net::{EntityKeypair, MeshNodeConfig};
use std::net::SocketAddr;
use std::time::Duration;
#[test]
fn request_grant_route_is_direct_only_for_authenticated_sessions() {
assert_eq!(
request_grant_route(0xABCD),
(Some(0xABCD), ResponseRouteFallback::DirectOnly),
"a real authenticated session must get a DirectOnly grant to its own node",
);
assert_eq!(
request_grant_route(0),
(None, ResponseRouteFallback::RosterOnStaleDirect),
"the loopback/relayed sentinel keeps the roster path",
);
}
#[test]
fn classify_request_grant_route_matches_authenticated_origin_equality() {
const ORIGIN: u64 = 0x1111_2222_3333_4444;
const NODE_A: u64 = 0xABCD;
const NODE_B: u64 = 0xBEEF;
assert_eq!(
classify_request_grant_route(NODE_A, ORIGIN, Some(ORIGIN)),
RequestGrantRoute::TrustedDirect(NODE_A),
);
assert_eq!(
classify_request_grant_route(NODE_A, ORIGIN, Some(0x9999_9999_9999_9999)),
RequestGrantRoute::RelayedOrUntrusted,
);
assert_eq!(
classify_request_grant_route(NODE_A, ORIGIN, None),
RequestGrantRoute::RelayedOrUntrusted,
);
assert_eq!(
classify_request_grant_route(0, ORIGIN, Some(ORIGIN)),
RequestGrantRoute::Loopback,
);
assert_eq!(
classify_request_grant_route(NODE_A, ORIGIN, Some(ORIGIN)),
RequestGrantRoute::TrustedDirect(NODE_A),
);
assert_eq!(
classify_request_grant_route(NODE_B, ORIGIN, Some(ORIGIN)),
RequestGrantRoute::TrustedDirect(NODE_B),
);
}
fn rpc_request_frame(
payload_origin: u64,
call_id: u64,
dispatch: u8,
service: &str,
flags: u16,
window: Option<&[u8]>,
) -> Bytes {
let headers = match window {
Some(w) => vec![(HEADER_NRPC_REQUEST_WINDOW_INITIAL.to_string(), w.to_vec())],
None => vec![],
};
let payload = RpcRequestPayload {
service: service.to_string(),
deadline_ns: 0,
flags,
headers,
body: Bytes::new(),
};
let mut buf = EventMeta::new(dispatch, 0, payload_origin, call_id, 0)
.to_bytes()
.to_vec();
encode_rpc_route(&mut buf, 0);
buf.extend_from_slice(&payload.encode());
Bytes::from(buf)
}
#[tokio::test]
async fn reject_relayed_flow_controlled_request_rejects_only_relayed_flow_controlled_uploads() {
use std::sync::atomic::Ordering;
let server = build_server().await;
let metrics = server.rpc_metrics_arc().for_service("svc.upload");
const DIRECT_NODE: u64 = 0x51;
const RELAY_NODE: u64 = 0x52;
let direct_entity = EntityKeypair::generate().entity_id().clone();
let relay_entity = EntityKeypair::generate().entity_id().clone();
let direct_origin = direct_entity.origin_hash();
let relay_origin = relay_entity.origin_hash();
server.test_pin_peer_entity(DIRECT_NODE, direct_entity);
server.test_pin_peer_entity(RELAY_NODE, relay_entity);
let victim_origin = relay_origin ^ 0xFFFF_FFFF;
let chan = ChannelId::new(ChannelName::new("svc.upload.requests").unwrap()).hash();
let frame = |from_node: u64,
claimed_origin: u64,
call_id: u64,
dispatch: u8,
window: Option<&[u8]>| {
RpcInboundEvent {
channel_hash: chan,
origin_hash: claimed_origin,
from_node,
payload: rpc_request_frame(
claimed_origin,
call_id,
dispatch,
"svc.upload",
FLAG_RPC_CLIENT_STREAMING_REQUEST,
window,
),
}
};
let reject = |ev: &RpcInboundEvent| {
reject_relayed_flow_controlled_request(
&server,
&metrics,
ev,
"svc.upload",
"nrpc:svc.upload",
)
};
let rejected = || {
metrics
.relayed_flow_controlled_rejected_total
.load(Ordering::Relaxed)
};
assert!(
reject(&frame(
RELAY_NODE,
victim_origin,
1,
DISPATCH_RPC_REQUEST,
Some(b"32")
)),
"a relayed flow-controlled caller must be rejected before the fold",
);
assert_eq!(
rejected(),
1,
"the relayed flow-controlled REQUEST metered once"
);
assert!(
!reject(&frame(
RELAY_NODE,
victim_origin,
2,
DISPATCH_RPC_REQUEST,
None
)),
"a relayed caller without a window header is unbounded-upload and admitted",
);
assert!(
!reject(&frame(
RELAY_NODE,
victim_origin,
3,
DISPATCH_RPC_REQUEST,
Some(b"not-a-number")
)),
"a malformed window header parses as absent, matching the fold",
);
assert!(
!reject(&frame(
DIRECT_NODE,
direct_origin,
4,
DISPATCH_RPC_REQUEST,
Some(b"32")
)),
"a directly authenticated flow-controlled caller is admitted",
);
assert!(
!reject(&frame(0, 0xDEAD_BEEF, 5, DISPATCH_RPC_REQUEST, Some(b"32"))),
"the loopback sentinel is admitted",
);
assert!(
!reject(&frame(
RELAY_NODE,
victim_origin,
1,
DISPATCH_RPC_REQUEST_CHUNK,
Some(b"32")
)),
"only the initial REQUEST is classified; control frames pass through",
);
assert_eq!(
rejected(),
1,
"only the one relayed flow-controlled REQUEST was metered"
);
}
struct RanClientStream(std::sync::Arc<std::sync::atomic::AtomicUsize>);
#[async_trait::async_trait]
impl crate::adapter::net::cortex::RpcClientStreamingHandler for RanClientStream {
async fn call(
&self,
_ctx: crate::adapter::net::cortex::RpcStreamingContext,
mut requests: crate::adapter::net::cortex::RequestStream,
) -> Result<RpcResponsePayload, RpcHandlerError> {
use futures::StreamExt;
self.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
while requests.next().await.is_some() {}
Ok(RpcResponsePayload {
status: RpcStatus::Ok,
headers: vec![],
body: Bytes::new(),
})
}
}
struct RanDuplex(std::sync::Arc<std::sync::atomic::AtomicUsize>);
#[async_trait::async_trait]
impl crate::adapter::net::cortex::RpcDuplexHandler for RanDuplex {
async fn call(
&self,
_ctx: crate::adapter::net::cortex::RpcStreamingContext,
mut requests: crate::adapter::net::cortex::RequestStream,
_responses: crate::adapter::net::cortex::RpcResponseSink,
) -> Result<(), RpcHandlerError> {
use futures::StreamExt;
self.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
while requests.next().await.is_some() {}
Ok(())
}
}
async fn wait_until_at_least(get: impl Fn() -> u64, want: u64) -> bool {
for _ in 0..200 {
if get() >= want {
return true;
}
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
get() >= want
}
#[tokio::test]
async fn client_stream_bridge_rejects_before_fold_end_to_end() {
use std::sync::atomic::{AtomicUsize, Ordering};
let server = build_server().await;
let ran = std::sync::Arc::new(AtomicUsize::new(0));
let serve = server
.serve_rpc_client_stream("cs", std::sync::Arc::new(RanClientStream(ran.clone())))
.expect("serve client-stream");
let channel_hash = serve.channel_hash;
let metrics = server.rpc_metrics_arc().for_service("cs");
const DIRECT_NODE: u64 = 0x61;
const RELAY_NODE: u64 = 0x62;
let direct_entity = EntityKeypair::generate().entity_id().clone();
let relay_entity = EntityKeypair::generate().entity_id().clone();
let direct_origin = direct_entity.origin_hash();
let relay_origin = relay_entity.origin_hash();
server.test_pin_peer_entity(DIRECT_NODE, direct_entity);
server.test_pin_peer_entity(RELAY_NODE, relay_entity);
let victim = relay_origin ^ 0xFFFF_FFFF;
let event = |from_node: u64,
packet_origin: u64,
payload_origin: u64,
dispatch: u8,
call_id: u64,
window: Option<&[u8]>| {
RpcInboundEvent {
channel_hash,
origin_hash: packet_origin,
from_node,
payload: rpc_request_frame(
payload_origin,
call_id,
dispatch,
"cs",
FLAG_RPC_CLIENT_STREAMING_REQUEST,
window,
),
}
};
assert!(server.deliver_rpc_inbound_for_test(
channel_hash,
event(
DIRECT_NODE,
direct_origin,
victim,
DISPATCH_RPC_REQUEST,
1,
Some(b"32")
)
));
assert!(
wait_until_at_least(
|| metrics
.packet_origin_mismatch_dropped_total
.load(Ordering::Relaxed),
1
)
.await,
"an origin-mismatch frame is dropped and metered",
);
assert_eq!(
serve.origin_node_cache.get((DIRECT_NODE, direct_origin, 1)),
None,
"an origin-mismatch frame must not mutate the response-route cache",
);
assert!(server.deliver_rpc_inbound_for_test(
channel_hash,
event(
RELAY_NODE,
victim,
victim,
DISPATCH_RPC_REQUEST,
2,
Some(b"32")
)
));
assert!(
wait_until_at_least(
|| metrics
.relayed_flow_controlled_rejected_total
.load(Ordering::Relaxed),
1
)
.await,
"a relayed flow-controlled frame is rejected and metered",
);
assert_eq!(
serve.origin_node_cache.get((RELAY_NODE, victim, 2)),
None,
"a relayed frame must not mutate the response-route cache",
);
assert!(server.deliver_rpc_inbound_for_test(
channel_hash,
event(
RELAY_NODE,
victim,
victim,
DISPATCH_RPC_REQUEST_CHUNK,
2,
None
)
));
assert!(server.deliver_rpc_inbound_for_test(
channel_hash,
event(RELAY_NODE, victim, victim, DISPATCH_RPC_REQUEST, 3, None)
));
assert!(
wait_until_at_least(|| ran.load(Ordering::SeqCst) as u64, 1).await,
"a relayed non-flow-controlled REQUEST is admitted and the handler runs",
);
assert_eq!(
metrics
.relayed_flow_controlled_rejected_total
.load(Ordering::Relaxed),
1,
"per-call counter: the continuation CHUNK must not re-count the rejection",
);
assert!(server.deliver_rpc_inbound_for_test(
channel_hash,
event(
DIRECT_NODE,
direct_origin,
direct_origin,
DISPATCH_RPC_REQUEST,
4,
Some(b"32")
)
));
assert!(
wait_until_at_least(|| ran.load(Ordering::SeqCst) as u64, 2).await,
"a direct flow-controlled REQUEST is admitted and the handler runs",
);
assert_eq!(
serve.origin_node_cache.get((DIRECT_NODE, direct_origin, 4)),
Some(DIRECT_NODE),
"an accepted authenticated direct frame DOES cache its response route",
);
assert_eq!(
ran.load(Ordering::SeqCst),
2,
"only the two admitted frames ever ran the handler",
);
assert_eq!(
metrics
.packet_origin_mismatch_dropped_total
.load(Ordering::Relaxed),
1,
);
assert_eq!(
metrics
.relayed_flow_controlled_rejected_total
.load(Ordering::Relaxed),
1,
);
}
#[tokio::test]
async fn duplex_bridge_rejects_before_fold_end_to_end() {
use std::sync::atomic::{AtomicUsize, Ordering};
let server = build_server().await;
let ran = std::sync::Arc::new(AtomicUsize::new(0));
let serve = server
.serve_rpc_duplex("dx", std::sync::Arc::new(RanDuplex(ran.clone())))
.expect("serve duplex");
let channel_hash = serve.channel_hash;
let metrics = server.rpc_metrics_arc().for_service("dx");
const DIRECT_NODE: u64 = 0x71;
const RELAY_NODE: u64 = 0x72;
let direct_entity = EntityKeypair::generate().entity_id().clone();
let relay_entity = EntityKeypair::generate().entity_id().clone();
let direct_origin = direct_entity.origin_hash();
let relay_origin = relay_entity.origin_hash();
server.test_pin_peer_entity(DIRECT_NODE, direct_entity);
server.test_pin_peer_entity(RELAY_NODE, relay_entity);
let victim = relay_origin ^ 0xFFFF_FFFF;
let dx_flags = FLAG_RPC_CLIENT_STREAMING_REQUEST | FLAG_RPC_STREAMING_RESPONSE;
let event = |from_node: u64,
packet_origin: u64,
payload_origin: u64,
dispatch: u8,
call_id: u64,
window: Option<&[u8]>| {
RpcInboundEvent {
channel_hash,
origin_hash: packet_origin,
from_node,
payload: rpc_request_frame(
payload_origin,
call_id,
dispatch,
"dx",
dx_flags,
window,
),
}
};
assert!(server.deliver_rpc_inbound_for_test(
channel_hash,
event(
DIRECT_NODE,
direct_origin,
victim,
DISPATCH_RPC_REQUEST,
1,
Some(b"32")
)
));
assert!(
wait_until_at_least(
|| metrics
.packet_origin_mismatch_dropped_total
.load(Ordering::Relaxed),
1
)
.await,
"an origin-mismatch frame is dropped and metered on the duplex bridge",
);
assert_eq!(
serve.origin_node_cache.get((DIRECT_NODE, direct_origin, 1)),
None,
"an origin-mismatch frame must not mutate the response-route cache",
);
assert!(server.deliver_rpc_inbound_for_test(
channel_hash,
event(
RELAY_NODE,
victim,
victim,
DISPATCH_RPC_REQUEST,
2,
Some(b"32")
)
));
assert!(
wait_until_at_least(
|| metrics
.relayed_flow_controlled_rejected_total
.load(Ordering::Relaxed),
1
)
.await,
"a relayed flow-controlled frame is rejected and metered on the duplex bridge",
);
assert_eq!(
serve.origin_node_cache.get((RELAY_NODE, victim, 2)),
None,
"a relayed frame must not mutate the response-route cache",
);
assert!(server.deliver_rpc_inbound_for_test(
channel_hash,
event(
RELAY_NODE,
victim,
victim,
DISPATCH_RPC_REQUEST_CHUNK,
2,
None
)
));
assert!(server.deliver_rpc_inbound_for_test(
channel_hash,
event(
DIRECT_NODE,
direct_origin,
direct_origin,
DISPATCH_RPC_REQUEST,
3,
Some(b"32")
)
));
assert!(
wait_until_at_least(|| ran.load(Ordering::SeqCst) as u64, 1).await,
"a direct flow-controlled duplex REQUEST is admitted and the handler runs",
);
assert_eq!(
serve.origin_node_cache.get((DIRECT_NODE, direct_origin, 3)),
Some(DIRECT_NODE),
"an accepted authenticated direct frame DOES cache its response route",
);
assert_eq!(
ran.load(Ordering::SeqCst),
1,
"only the one admitted frame ran the duplex handler",
);
assert_eq!(
metrics
.packet_origin_mismatch_dropped_total
.load(Ordering::Relaxed),
1,
);
assert_eq!(
metrics
.relayed_flow_controlled_rejected_total
.load(Ordering::Relaxed),
1,
);
}
async fn build_server() -> Arc<MeshNode> {
let addr: SocketAddr = "127.0.0.1:0".parse().unwrap();
let cfg = MeshNodeConfig::new(addr, [0x42u8; 32])
.with_heartbeat_interval(Duration::from_millis(200))
.with_session_timeout(Duration::from_secs(10))
.with_handshake(3, Duration::from_secs(2));
Arc::new(
MeshNode::new(EntityKeypair::generate(), cfg)
.await
.expect("MeshNode::new"),
)
}
async fn protected_provider(
tag: &str,
) -> (
Arc<MeshNode>,
crate::adapter::net::identity::EntityId,
crate::adapter::net::behavior::org::OrgKeypair,
std::path::PathBuf,
) {
use crate::adapter::net::behavior::org::{OrgKeypair, OrgMembershipCert};
use crate::adapter::net::behavior::org_authority::NodeAuthority;
let server = build_server().await;
let node_entity = server.entity_id().clone();
let org_b = OrgKeypair::from_bytes([0x42u8; 32]);
let node_cert =
OrgMembershipCert::try_issue(&org_b, node_entity.clone(), 1, 3600).expect("node cert");
let dir = std::env::temp_dir().join(format!("net-oa2-{tag}-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
let authority =
NodeAuthority::adopt(&dir, node_cert, &node_entity, 0, None).expect("adopt authority");
server
.install_node_authority(std::sync::Arc::new(authority))
.expect("install authority");
(server, node_entity, org_b, dir)
}
fn owner_delegated_intent(
caller_kp: EntityKeypair,
org_b: &crate::adapter::net::behavior::org::OrgKeypair,
provider: crate::adapter::net::identity::EntityId,
) -> OrgProofIntent {
use crate::adapter::net::behavior::org::OrgMembershipCert;
use crate::adapter::net::behavior::org_grant::{
CapabilityAuthorityId, DispatcherScope, OrgDispatcherGrant,
};
let caller_entity = caller_kp.entity_id().clone();
let cap = CapabilityAuthorityId::for_tag("nrpc:svc");
let membership =
OrgMembershipCert::try_issue(org_b, caller_entity.clone(), 1, 3600).expect("cert");
let dispatcher =
OrgDispatcherGrant::try_issue(org_b, caller_entity, DispatcherScope::Exact(cap), 3600)
.expect("dispatcher");
OrgProofIntent {
caller: std::sync::Arc::new(caller_kp),
membership,
dispatcher,
capability_grant: None,
acting_org: org_b.org_id(),
provider_owner_org: org_b.org_id(),
provider,
capability: cap,
proof_ttl_secs: 30,
}
}
#[tokio::test]
async fn admission_replay_guard_is_node_owned_across_reregistration() {
use crate::adapter::net::behavior::org_admission::OrgAdmission;
use std::sync::atomic::{AtomicUsize, Ordering};
struct Counter(std::sync::Arc<AtomicUsize>);
#[async_trait::async_trait]
impl RpcHandler for Counter {
async fn call(&self, _ctx: RpcContext) -> Result<RpcResponsePayload, RpcHandlerError> {
self.0.fetch_add(1, Ordering::SeqCst);
Ok(RpcResponsePayload {
status: RpcStatus::Ok,
headers: vec![],
body: Bytes::new(),
})
}
}
let (server, node_entity, org_b, _dir) = protected_provider("b2").await;
const CALLER_NODE: u64 = 0x9b;
let caller_kp = EntityKeypair::generate();
let caller_origin = caller_kp.entity_id().origin_hash();
server.test_pin_peer_entity(CALLER_NODE, caller_kp.entity_id().clone());
let intent = owner_delegated_intent(caller_kp, &org_b, node_entity);
let base = RpcRequestPayload {
service: "svc".to_string(),
deadline_ns: 0,
flags: 0,
headers: vec![],
body: Bytes::from_static(b"ping"),
};
let header1 = sign_admission_proof(&intent, 1, &base).expect("sign");
let make = |header: &(String, Vec<u8>), call_id: u64, channel_hash: ChannelHash| {
let mut req = base.clone();
req.headers.push(header.clone());
let mut f = EventMeta::new(DISPATCH_RPC_REQUEST, 0, caller_origin, call_id, 0)
.to_bytes()
.to_vec();
encode_rpc_route(&mut f, 0);
f.extend_from_slice(&req.encode());
RpcInboundEvent {
channel_hash,
origin_hash: caller_origin,
from_node: CALLER_NODE,
payload: Bytes::from(f),
}
};
let admits1 = std::sync::Arc::new(AtomicUsize::new(0));
let serve1 = server
.serve_rpc_protected(
"svc",
std::sync::Arc::new(Counter(admits1.clone())),
OrgAdmission::OwnerDelegated,
std::sync::Arc::new(|_| true),
)
.expect("serve #1");
let channel_hash = serve1.channel_hash;
assert!(server.deliver_rpc_inbound_for_test(channel_hash, make(&header1, 1, channel_hash)));
assert!(
wait_until_at_least(|| admits1.load(Ordering::SeqCst) as u64, 1).await,
"the first call admits",
);
drop(serve1);
let admits2 = std::sync::Arc::new(AtomicUsize::new(0));
let serve2 = server
.serve_rpc_protected(
"svc",
std::sync::Arc::new(Counter(admits2.clone())),
OrgAdmission::OwnerDelegated,
std::sync::Arc::new(|_| true),
)
.expect("serve #2");
let ch2 = serve2.channel_hash;
assert!(server.deliver_rpc_inbound_for_test(ch2, make(&header1, 1, ch2))); let header2 = sign_admission_proof(&intent, 2, &base).expect("sign fresh");
assert!(server.deliver_rpc_inbound_for_test(ch2, make(&header2, 2, ch2))); assert!(
wait_until_at_least(|| admits2.load(Ordering::SeqCst) as u64, 1).await,
"the fresh call admits on the second registration",
);
assert_eq!(
admits2.load(Ordering::SeqCst),
1,
"the replayed proof was DENIED across re-registration (node-owned guard)",
);
}
#[tokio::test]
async fn admission_replay_guard_collides_across_services() {
use crate::adapter::net::behavior::org::OrgMembershipCert;
use crate::adapter::net::behavior::org_admission::OrgAdmission;
use crate::adapter::net::behavior::org_grant::{
CapabilityAuthorityId, DispatcherScope, OrgDispatcherGrant,
};
use std::sync::atomic::{AtomicUsize, Ordering};
struct Counter(std::sync::Arc<AtomicUsize>);
#[async_trait::async_trait]
impl RpcHandler for Counter {
async fn call(&self, _ctx: RpcContext) -> Result<RpcResponsePayload, RpcHandlerError> {
self.0.fetch_add(1, Ordering::SeqCst);
Ok(RpcResponsePayload {
status: RpcStatus::Ok,
headers: vec![],
body: Bytes::new(),
})
}
}
let (server, node_entity, org_b, _dir) = protected_provider("b2-xsvc").await;
const CALLER_NODE: u64 = 0x9c;
let caller_kp = std::sync::Arc::new(EntityKeypair::generate());
let caller_origin = caller_kp.entity_id().origin_hash();
server.test_pin_peer_entity(CALLER_NODE, caller_kp.entity_id().clone());
let intent_for = |service: &str| -> OrgProofIntent {
let caller_entity = caller_kp.entity_id().clone();
let cap = CapabilityAuthorityId::for_tag(&format!("nrpc:{service}"));
let membership =
OrgMembershipCert::try_issue(&org_b, caller_entity.clone(), 1, 3600).expect("cert");
let dispatcher = OrgDispatcherGrant::try_issue(
&org_b,
caller_entity,
DispatcherScope::Exact(cap),
3600,
)
.expect("dispatcher");
OrgProofIntent {
caller: caller_kp.clone(),
membership,
dispatcher,
capability_grant: None,
acting_org: org_b.org_id(),
provider_owner_org: org_b.org_id(),
provider: node_entity.clone(),
capability: cap,
proof_ttl_secs: 30,
}
};
let req_for = |service: &str| RpcRequestPayload {
service: service.to_string(),
deadline_ns: 0,
flags: 0,
headers: vec![],
body: Bytes::from_static(b"ping"),
};
let make = |base: &RpcRequestPayload,
header: &(String, Vec<u8>),
call_id: u64,
channel_hash: ChannelHash| {
let mut req = base.clone();
req.headers.push(header.clone());
let mut f = EventMeta::new(DISPATCH_RPC_REQUEST, 0, caller_origin, call_id, 0)
.to_bytes()
.to_vec();
encode_rpc_route(&mut f, 0);
f.extend_from_slice(&req.encode());
RpcInboundEvent {
channel_hash,
origin_hash: caller_origin,
from_node: CALLER_NODE,
payload: Bytes::from(f),
}
};
let admits_a = std::sync::Arc::new(AtomicUsize::new(0));
let admits_b = std::sync::Arc::new(AtomicUsize::new(0));
let policy_calls_b = std::sync::Arc::new(AtomicUsize::new(0));
let serve_a = server
.serve_rpc_protected(
"a",
std::sync::Arc::new(Counter(admits_a.clone())),
OrgAdmission::OwnerDelegated,
std::sync::Arc::new(|_| true),
)
.expect("serve a");
let serve_b = server
.serve_rpc_protected(
"b",
std::sync::Arc::new(Counter(admits_b.clone())),
OrgAdmission::OwnerDelegated,
{
let policy_calls_b = policy_calls_b.clone();
std::sync::Arc::new(move |_: &_| {
policy_calls_b.fetch_add(1, Ordering::SeqCst);
true
})
},
)
.expect("serve b");
let ch_a = serve_a.channel_hash;
let ch_b = serve_b.channel_hash;
let intent_a = intent_for("a");
let req_a = req_for("a");
let header_a7 = sign_admission_proof(&intent_a, 7, &req_a).expect("sign a7");
assert!(server.deliver_rpc_inbound_for_test(ch_a, make(&req_a, &header_a7, 7, ch_a)));
assert!(
wait_until_at_least(|| admits_a.load(Ordering::SeqCst) as u64, 1).await,
"service A admits (caller, call_id=7)",
);
let intent_b = intent_for("b");
let req_b = req_for("b");
let header_b7 = sign_admission_proof(&intent_b, 7, &req_b).expect("sign b7");
let header_b8 = sign_admission_proof(&intent_b, 8, &req_b).expect("sign b8");
assert!(server.deliver_rpc_inbound_for_test(ch_b, make(&req_b, &header_b7, 7, ch_b)));
assert!(server.deliver_rpc_inbound_for_test(ch_b, make(&req_b, &header_b8, 8, ch_b)));
assert!(
wait_until_at_least(|| admits_b.load(Ordering::SeqCst) as u64, 1).await,
"service B admits the fresh call_id 8",
);
assert_eq!(
policy_calls_b.load(Ordering::SeqCst),
1,
"only the fresh call reached service B's policy — the reused call_id 7 was denied at \
the replay insert (node-wide (caller, call_id) collision)",
);
assert_eq!(
admits_b.load(Ordering::SeqCst),
1,
"service B's handler ran exactly once (only the fresh call)",
);
drop(serve_a);
drop(serve_b);
}
#[test]
fn public_admission_strip_removes_only_the_proof_header() {
let frame = |dispatch: u8, headers: Vec<(String, Vec<u8>)>| -> RpcInboundEvent {
let req = RpcRequestPayload {
service: "svc".to_string(),
deadline_ns: 0,
flags: 0,
headers,
body: Bytes::from_static(b"ping"),
};
let mut f = EventMeta::new(dispatch, 0, 0, 1, 0).to_bytes().to_vec();
encode_rpc_route(&mut f, 0);
f.extend_from_slice(&req.encode());
RpcInboundEvent {
channel_hash: 0,
origin_hash: 0,
from_node: 1,
payload: Bytes::from(f),
}
};
assert!(
strip_public_admission_header(&frame(
DISPATCH_RPC_REQUEST,
vec![("keep".to_string(), b"v".to_vec())]
))
.is_none(),
"a request with no proof header is a no-op",
);
assert!(
strip_public_admission_header(&frame(
DISPATCH_RPC_CANCEL,
vec![(ORG_ADMISSION_HEADER.to_string(), b"proofbytes".to_vec())]
))
.is_none(),
"a non-REQUEST frame is never rewritten",
);
let with_proof = frame(
DISPATCH_RPC_REQUEST,
vec![
("keep".to_string(), b"v".to_vec()),
(ORG_ADMISSION_HEADER.to_string(), b"proofbytes".to_vec()),
],
);
let stripped = strip_public_admission_header(&with_proof)
.expect("rewrites when the proof header is present");
let req = RpcRequestPayload::decode(stripped.payload.slice(RPC_FRAME_BODY_OFFSET..))
.expect("decode rewritten request");
assert!(
!req.headers.iter().any(|(n, _)| n == ORG_ADMISSION_HEADER),
"the proof header was removed",
);
assert!(
req.headers.iter().any(|(n, _)| n == "keep"),
"an unrelated header was retained",
);
}
#[tokio::test]
async fn bridge_preflight_strips_a_stray_proof_header_for_every_bridge() {
let server = build_server().await;
let cache: RpcOriginNodeCache = Arc::new(BoundedLru::new());
let metrics = server.rpc_metrics_arc().for_service("svc");
let req = RpcRequestPayload {
service: "svc".to_string(),
deadline_ns: 0,
flags: 0,
headers: vec![
("keep".to_string(), b"v".to_vec()),
(ORG_ADMISSION_HEADER.to_string(), b"proofbytes".to_vec()),
],
body: Bytes::from_static(b"ping"),
};
let mut payload = EventMeta::new(DISPATCH_RPC_REQUEST, 0, 0, 1, 0)
.to_bytes()
.to_vec();
encode_rpc_route(&mut payload, 0);
payload.extend_from_slice(&req.encode());
let inbound = RpcInboundEvent {
channel_hash: 0,
origin_hash: 0,
from_node: 0,
payload: Bytes::from(payload),
};
match bridge_preflight(&server, &cache, &inbound, "svc", "nrpc:svc", &metrics) {
BridgePreflight::Proceed(frame) => {
let decoded =
RpcRequestPayload::decode(frame.payload.slice(RPC_FRAME_BODY_OFFSET..))
.expect("decode the frame the bridge would fold");
assert!(
!decoded
.headers
.iter()
.any(|(n, _)| n == ORG_ADMISSION_HEADER),
"the frame handed to EVERY bridge must carry no proof header",
);
assert!(
decoded.headers.iter().any(|(n, _)| n == "keep"),
"an unrelated header must survive",
);
assert_eq!(decoded.body.as_ref(), b"ping", "the body is untouched");
}
_ => panic!("expected Proceed for a loopback frame on the captured service"),
}
let clean = RpcRequestPayload {
service: "svc".to_string(),
deadline_ns: 0,
flags: 0,
headers: vec![("keep".to_string(), b"v".to_vec())],
body: Bytes::from_static(b"ping"),
};
let mut payload = EventMeta::new(DISPATCH_RPC_REQUEST, 0, 0, 2, 0)
.to_bytes()
.to_vec();
encode_rpc_route(&mut payload, 0);
payload.extend_from_slice(&clean.encode());
let inbound = RpcInboundEvent {
channel_hash: 0,
origin_hash: 0,
from_node: 0,
payload: Bytes::from(payload),
};
match bridge_preflight(&server, &cache, &inbound, "svc", "nrpc:svc", &metrics) {
BridgePreflight::Proceed(frame) => assert_eq!(
frame.payload, inbound.payload,
"a frame with no proof header is passed through byte-for-byte",
),
_ => panic!("expected Proceed"),
}
}
#[tokio::test]
async fn protected_call_finalization_is_atomic_bounded_and_leak_free() {
use crate::adapter::net::behavior::org::OrgKeypair;
use crate::adapter::net::cortex::MAX_RPC_HEADERS;
const TARGET: u64 = 0xDEAD_BEEF;
let server = build_server().await;
let org_b = OrgKeypair::from_bytes([0x42u8; 32]);
let provider = crate::adapter::net::identity::EntityId::from_bytes([0x99u8; 32]);
server.test_pin_peer_entity(TARGET, provider.clone());
let pending = server.rpc_client_pending();
assert_eq!(pending.pending_count(), 0);
let dup = CallOptions {
org_proof_intent: Some(owner_delegated_intent(
EntityKeypair::generate(),
&org_b,
provider.clone(),
)),
request_headers: vec![("net-org-admission".to_string(), b"forged".to_vec())],
..Default::default()
};
assert!(
matches!(
server
.call(TARGET, "svc", Bytes::from_static(b"ping"), dup)
.await,
Err(RpcError::Codec { .. })
),
"a caller-supplied admission header must fail locally",
);
assert_eq!(
pending.pending_count(),
0,
"the dup-header failure must not leak a pending entry",
);
let over = CallOptions {
org_proof_intent: Some(owner_delegated_intent(
EntityKeypair::generate(),
&org_b,
provider,
)),
request_headers: (0..MAX_RPC_HEADERS)
.map(|i| (format!("h{i}"), b"v".to_vec()))
.collect(),
..Default::default()
};
assert!(
matches!(
server
.call(TARGET, "svc", Bytes::from_static(b"ping"), over)
.await,
Err(RpcError::Codec { .. })
),
"an over-cap finalized request must fail locally",
);
assert_eq!(
pending.pending_count(),
0,
"the over-cap failure must not leak a pending entry",
);
}
#[tokio::test]
async fn org_proof_intent_rejected_on_streaming_and_capability_mismatch() {
use crate::adapter::net::behavior::org::OrgKeypair;
const TARGET: u64 = 0xDEAD_BEEF;
let server = build_server().await;
let org_b = OrgKeypair::from_bytes([0x42u8; 32]);
let provider = crate::adapter::net::identity::EntityId::from_bytes([0x99u8; 32]);
server.test_pin_peer_entity(TARGET, provider.clone());
let intent_opts = || CallOptions {
org_proof_intent: Some(owner_delegated_intent(
EntityKeypair::generate(),
&org_b,
provider.clone(),
)),
..Default::default()
};
assert!(matches!(
server
.call_streaming(TARGET, "svc", Bytes::new(), intent_opts())
.await,
Err(RpcError::Codec { .. })
));
assert!(matches!(
server
.call_client_stream(TARGET, "svc", intent_opts())
.await,
Err(RpcError::Codec { .. })
));
assert!(matches!(
server.call_duplex(TARGET, "svc", intent_opts()).await,
Err(RpcError::Codec { .. })
));
assert!(matches!(
server
.call_service_streaming("svc", Bytes::new(), intent_opts())
.await,
Err(RpcError::Codec { .. })
));
assert!(matches!(
server
.call(TARGET, "other", Bytes::from_static(b"x"), intent_opts())
.await,
Err(RpcError::Codec { .. })
));
}
#[test]
fn org_proof_ttl_out_of_range_fails_locally() {
use crate::adapter::net::behavior::org::OrgKeypair;
let org_b = OrgKeypair::from_bytes([0x42u8; 32]);
let provider = crate::adapter::net::identity::EntityId::from_bytes([0x99u8; 32]);
let base = RpcRequestPayload {
service: "svc".to_string(),
deadline_ns: 0,
flags: 0,
headers: vec![],
body: Bytes::from_static(b"ping"),
};
let intent_with = |ttl: u64| {
let mut i = owner_delegated_intent(EntityKeypair::generate(), &org_b, provider.clone());
i.proof_ttl_secs = ttl;
i
};
assert!(matches!(
sign_admission_proof(&intent_with(0), 1, &base),
Err(RpcError::Codec { .. })
));
assert!(sign_admission_proof(&intent_with(MAX_ORG_PROOF_TTL_SECS), 2, &base).is_ok());
assert!(matches!(
sign_admission_proof(&intent_with(MAX_ORG_PROOF_TTL_SECS + 1), 3, &base),
Err(RpcError::Codec { .. })
));
}
#[tokio::test]
async fn protected_call_refuses_provider_target_mismatch() {
use crate::adapter::net::behavior::org::OrgKeypair;
const TARGET: u64 = 0xDEAD_BEEF;
let server = build_server().await;
let org_b = OrgKeypair::from_bytes([0x42u8; 32]);
let provider = crate::adapter::net::identity::EntityId::from_bytes([0x99u8; 32]);
let other = crate::adapter::net::identity::EntityId::from_bytes([0x11u8; 32]);
let pending = server.rpc_client_pending();
let opts = || CallOptions {
org_proof_intent: Some(owner_delegated_intent(
EntityKeypair::generate(),
&org_b,
provider.clone(),
)),
..Default::default()
};
assert!(matches!(
server
.call(TARGET, "svc", Bytes::from_static(b"x"), opts())
.await,
Err(RpcError::Codec { .. })
));
assert_eq!(
pending.pending_count(),
0,
"no pending leak on unpinned target"
);
server.test_pin_peer_entity(TARGET, other);
assert!(matches!(
server
.call(TARGET, "svc", Bytes::from_static(b"x"), opts())
.await,
Err(RpcError::Codec { .. })
));
assert_eq!(
pending.pending_count(),
0,
"no pending leak on provider mismatch"
);
}
#[tokio::test]
async fn serve_rpc_protected_refuses_bad_mode_and_missing_authority() {
use crate::adapter::net::behavior::org_admission::OrgAdmission;
struct H;
#[async_trait::async_trait]
impl RpcHandler for H {
async fn call(&self, _ctx: RpcContext) -> Result<RpcResponsePayload, RpcHandlerError> {
Ok(RpcResponsePayload {
status: RpcStatus::Ok,
headers: vec![],
body: Bytes::new(),
})
}
}
let server = build_server().await; let policy: OrgProviderPolicy = std::sync::Arc::new(|_| true);
assert!(
matches!(
server.serve_rpc_protected(
"p",
std::sync::Arc::new(H),
OrgAdmission::PublicAuthenticated,
policy.clone(),
),
Err(ServeError::InvalidProtectedRegistration(_))
),
"PublicAuthenticated is not a protected mode",
);
assert!(
matches!(
server.serve_rpc_protected(
"p",
std::sync::Arc::new(H),
OrgAdmission::CrossOrgGranted,
policy,
),
Err(ServeError::ProtectedAuthorityRequired(_))
),
"protected registration requires an installed authority",
);
assert!(
server.serve_rpc("p", std::sync::Arc::new(H)).is_ok(),
"no dangling protected registration blocks the slot",
);
}
#[test]
fn protected_registrations_never_roster_fan_responses() {
use crate::adapter::net::behavior::org_admission::OrgAdmission;
let policy: OrgProviderPolicy = std::sync::Arc::new(|_| true);
assert_eq!(
UnaryAdmission::Public.response_route_fallback(),
ResponseRouteFallback::RosterOnStaleDirect,
"public services keep the AV-5 roster fallback",
);
for (label, mode) in [
(
"Protected/OwnerDelegated",
UnaryAdmission::Protected {
admission: OrgAdmission::OwnerDelegated,
provider_policy: policy.clone(),
},
),
(
"Protected/CrossOrgGranted",
UnaryAdmission::Protected {
admission: OrgAdmission::CrossOrgGranted,
provider_policy: policy.clone(),
},
),
(
"OwnerScoped",
UnaryAdmission::OwnerScoped {
provider_policy: policy.clone(),
},
),
(
"Granted",
UnaryAdmission::Granted {
provider_policy: policy.clone(),
},
),
(
"ProtectedRedWitnessDisabled",
UnaryAdmission::ProtectedRedWitnessDisabled {
admission: OrgAdmission::OwnerDelegated,
provider_policy: policy.clone(),
},
),
] {
assert_eq!(
mode.response_route_fallback(),
ResponseRouteFallback::DirectOnly,
"{label} must never roster-fan an org-confidential response",
);
}
}
#[tokio::test]
async fn protected_owner_delegated_call_admits_end_to_end() {
use crate::adapter::net::behavior::org::{
current_timestamp, OrgKeypair, OrgMembershipCert,
};
use crate::adapter::net::behavior::org_admission::{Admitted, OrgAdmission};
use crate::adapter::net::behavior::org_authority::NodeAuthority;
use crate::adapter::net::behavior::org_call::{OrgCallProof, ORG_ADMISSION_HEADER};
use crate::adapter::net::behavior::org_grant::{
CapabilityAuthorityId, DispatcherScope, OrgDispatcherGrant,
};
use crate::adapter::net::org_admission_gate::org_request_digest;
type Seen = std::sync::Arc<Mutex<Option<Admitted>>>;
struct SpyHandler(Seen);
#[async_trait::async_trait]
impl RpcHandler for SpyHandler {
async fn call(&self, ctx: RpcContext) -> Result<RpcResponsePayload, RpcHandlerError> {
*self.0.lock() = Some(ctx.org_admission.clone().expect("admitted call"));
Ok(RpcResponsePayload {
status: RpcStatus::Ok,
headers: vec![],
body: Bytes::new(),
})
}
}
let server = build_server().await;
let node_entity = server.entity_id().clone();
let org_b = OrgKeypair::from_bytes([0x42u8; 32]);
let node_cert =
OrgMembershipCert::try_issue(&org_b, node_entity.clone(), 1, 3600).expect("node cert");
let dir = std::env::temp_dir().join(format!("net-oa2-wire-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
let authority =
NodeAuthority::adopt(&dir, node_cert, &node_entity, 0, None).expect("adopt authority");
server
.install_node_authority(std::sync::Arc::new(authority))
.expect("install authority");
let seen: Seen = std::sync::Arc::new(Mutex::new(None));
let serve = server
.serve_rpc_protected(
"svc",
std::sync::Arc::new(SpyHandler(seen.clone())),
OrgAdmission::OwnerDelegated,
std::sync::Arc::new(|_| true),
)
.expect("serve protected");
let channel_hash = serve.channel_hash;
const CALLER_NODE: u64 = 0x91;
let caller_kp = EntityKeypair::generate();
let caller_entity = caller_kp.entity_id().clone();
let caller_origin = caller_entity.origin_hash();
server.test_pin_peer_entity(CALLER_NODE, caller_entity.clone());
let cap = CapabilityAuthorityId::for_tag("nrpc:svc");
let call_id = 7u64;
let base = RpcRequestPayload {
service: "svc".to_string(),
deadline_ns: 0,
flags: 0,
headers: vec![],
body: Bytes::from_static(b"ping"),
};
let digest = org_request_digest(&base).expect("digest");
let membership = OrgMembershipCert::try_issue(&org_b, caller_entity.clone(), 1, 3600)
.expect("caller cert");
let dispatcher = OrgDispatcherGrant::try_issue(
&org_b,
caller_entity.clone(),
DispatcherScope::Exact(cap),
3600,
)
.expect("dispatcher grant");
let expiry = (current_timestamp() + 20) * 1_000_000_000;
let proof = OrgCallProof::sign_for_call(
&caller_kp,
membership,
dispatcher,
None,
org_b.org_id(),
org_b.org_id(),
node_entity.clone(),
call_id,
cap,
expiry,
digest,
);
let proof_bytes = proof.encode().expect("encode proof");
let mut payload = base.clone();
payload
.headers
.push((ORG_ADMISSION_HEADER.to_string(), proof_bytes));
let mut frame = EventMeta::new(DISPATCH_RPC_REQUEST, 0, caller_origin, call_id, 0)
.to_bytes()
.to_vec();
encode_rpc_route(&mut frame, 0);
frame.extend_from_slice(&payload.encode());
let event = RpcInboundEvent {
channel_hash,
origin_hash: caller_origin,
from_node: CALLER_NODE,
payload: Bytes::from(frame),
};
assert!(server.deliver_rpc_inbound_for_test(channel_hash, event));
assert!(
wait_until_at_least(|| seen.lock().is_some() as u64, 1).await,
"the handler must run for an admitted protected call",
);
let admitted = seen.lock().clone().expect("admitted");
assert_eq!(admitted.caller, caller_entity, "caller S");
assert_eq!(
admitted.acting_org,
org_b.org_id(),
"acting org A == owner B"
);
assert_eq!(admitted.provider_org, org_b.org_id(), "provider org B");
assert_eq!(admitted.provider, node_entity, "exact provider P");
assert_eq!(admitted.capability, cap, "capability C");
}
#[tokio::test]
async fn protected_replayed_and_tampered_proof_frames_are_denied() {
use crate::adapter::net::behavior::org::{
current_timestamp, OrgKeypair, OrgMembershipCert,
};
use crate::adapter::net::behavior::org_admission::OrgAdmission;
use crate::adapter::net::behavior::org_authority::NodeAuthority;
use crate::adapter::net::behavior::org_call::{OrgCallProof, ORG_ADMISSION_HEADER};
use crate::adapter::net::behavior::org_grant::{
CapabilityAuthorityId, DispatcherScope, OrgDispatcherGrant,
};
use crate::adapter::net::org_admission_gate::org_request_digest;
let admits = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
struct CountingHandler(std::sync::Arc<std::sync::atomic::AtomicUsize>);
#[async_trait::async_trait]
impl RpcHandler for CountingHandler {
async fn call(&self, _ctx: RpcContext) -> Result<RpcResponsePayload, RpcHandlerError> {
self.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
Ok(RpcResponsePayload {
status: RpcStatus::Ok,
headers: vec![],
body: Bytes::from_static(b"pong"),
})
}
}
let server = build_server().await;
let node_entity = server.entity_id().clone();
let org_b = OrgKeypair::from_bytes([0x42u8; 32]);
let node_cert =
OrgMembershipCert::try_issue(&org_b, node_entity.clone(), 1, 3600).expect("node cert");
let dir = std::env::temp_dir().join(format!("net-oa2-replay-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
let authority =
NodeAuthority::adopt(&dir, node_cert, &node_entity, 0, None).expect("adopt authority");
server
.install_node_authority(std::sync::Arc::new(authority))
.expect("install authority");
let serve = server
.serve_rpc_protected(
"svc",
std::sync::Arc::new(CountingHandler(admits.clone())),
OrgAdmission::OwnerDelegated,
std::sync::Arc::new(|_| true),
)
.expect("serve protected");
let channel_hash = serve.channel_hash;
const CALLER_NODE: u64 = 0x91;
let caller_kp = EntityKeypair::generate();
let caller_entity = caller_kp.entity_id().clone();
let caller_origin = caller_entity.origin_hash();
server.test_pin_peer_entity(CALLER_NODE, caller_entity.clone());
let cap = CapabilityAuthorityId::for_tag("nrpc:svc");
let call_id = 7u64;
let base = RpcRequestPayload {
service: "svc".to_string(),
deadline_ns: 0,
flags: 0,
headers: vec![],
body: Bytes::from_static(b"ping"),
};
let digest = org_request_digest(&base).expect("digest");
let membership = OrgMembershipCert::try_issue(&org_b, caller_entity.clone(), 1, 3600)
.expect("caller cert");
let dispatcher = OrgDispatcherGrant::try_issue(
&org_b,
caller_entity.clone(),
DispatcherScope::Exact(cap),
3600,
)
.expect("dispatcher grant");
let expiry = (current_timestamp() + 20) * 1_000_000_000;
let proof = OrgCallProof::sign_for_call(
&caller_kp,
membership,
dispatcher,
None,
org_b.org_id(),
org_b.org_id(),
node_entity.clone(),
call_id,
cap,
expiry,
digest,
);
let proof_bytes = proof.encode().expect("encode proof");
let mut payload = base.clone();
payload
.headers
.push((ORG_ADMISSION_HEADER.to_string(), proof_bytes.clone()));
let mut frame = EventMeta::new(DISPATCH_RPC_REQUEST, 0, caller_origin, call_id, 0)
.to_bytes()
.to_vec();
encode_rpc_route(&mut frame, 0);
frame.extend_from_slice(&payload.encode());
let frame = Bytes::from(frame);
let event = |payload: Bytes| RpcInboundEvent {
channel_hash,
origin_hash: caller_origin,
from_node: CALLER_NODE,
payload,
};
assert!(server.deliver_rpc_inbound_for_test(channel_hash, event(frame.clone())));
assert!(
wait_until_at_least(
|| admits.load(std::sync::atomic::Ordering::SeqCst) as u64,
1
)
.await,
"the first delivery must be admitted",
);
assert!(server.deliver_rpc_inbound_for_test(channel_hash, event(frame.clone())));
tokio::time::sleep(Duration::from_millis(200)).await;
assert_eq!(
admits.load(std::sync::atomic::Ordering::SeqCst),
1,
"a byte-identical replay must NOT reach the handler a second time",
);
let mut tampered_proof = proof_bytes.clone();
let last = tampered_proof.len() - 1;
tampered_proof[last] ^= 0xFF;
assert_ne!(tampered_proof, proof_bytes, "the tamper must change bytes");
let mut tampered_payload = base.clone();
tampered_payload
.headers
.push((ORG_ADMISSION_HEADER.to_string(), tampered_proof));
let tampered_call_id = call_id + 1;
let mut tampered_frame =
EventMeta::new(DISPATCH_RPC_REQUEST, 0, caller_origin, tampered_call_id, 0)
.to_bytes()
.to_vec();
encode_rpc_route(&mut tampered_frame, 0);
tampered_frame.extend_from_slice(&tampered_payload.encode());
assert!(
server.deliver_rpc_inbound_for_test(channel_hash, event(Bytes::from(tampered_frame)))
);
tokio::time::sleep(Duration::from_millis(200)).await;
assert_eq!(
admits.load(std::sync::atomic::Ordering::SeqCst),
1,
"a wire-tampered proof must never reach the handler",
);
}
#[tokio::test]
async fn protected_call_with_tampered_binding_is_denied_end_to_end() {
use crate::adapter::net::behavior::org::{
current_timestamp, OrgKeypair, OrgMembershipCert,
};
use crate::adapter::net::behavior::org_admission::OrgAdmission;
use crate::adapter::net::behavior::org_authority::NodeAuthority;
use crate::adapter::net::behavior::org_call::{OrgCallProof, ORG_ADMISSION_HEADER};
use crate::adapter::net::behavior::org_grant::{
CapabilityAuthorityId, DispatcherScope, OrgDispatcherGrant,
};
use crate::adapter::net::org_admission_gate::org_request_digest;
use std::sync::atomic::{AtomicUsize, Ordering};
struct Counter(std::sync::Arc<AtomicUsize>);
#[async_trait::async_trait]
impl RpcHandler for Counter {
async fn call(&self, ctx: RpcContext) -> Result<RpcResponsePayload, RpcHandlerError> {
assert!(
ctx.org_admission.is_some(),
"only admitted calls may reach the handler",
);
self.0.fetch_add(1, Ordering::SeqCst);
Ok(RpcResponsePayload {
status: RpcStatus::Ok,
headers: vec![],
body: Bytes::new(),
})
}
}
let server = build_server().await;
let node_entity = server.entity_id().clone();
let org_b = OrgKeypair::from_bytes([0x42u8; 32]);
let node_cert =
OrgMembershipCert::try_issue(&org_b, node_entity.clone(), 1, 3600).expect("node cert");
let dir = std::env::temp_dir().join(format!("net-oa2-wire-deny-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
let authority =
NodeAuthority::adopt(&dir, node_cert, &node_entity, 0, None).expect("adopt authority");
server
.install_node_authority(std::sync::Arc::new(authority))
.expect("install authority");
let admits = std::sync::Arc::new(AtomicUsize::new(0));
let serve = server
.serve_rpc_protected(
"svc",
std::sync::Arc::new(Counter(admits.clone())),
OrgAdmission::OwnerDelegated,
std::sync::Arc::new(|_| true),
)
.expect("serve protected");
let channel_hash = serve.channel_hash;
const CALLER_NODE: u64 = 0x93;
let caller_kp = EntityKeypair::generate();
let caller_entity = caller_kp.entity_id().clone();
let caller_origin = caller_entity.origin_hash();
server.test_pin_peer_entity(CALLER_NODE, caller_entity.clone());
let cap = CapabilityAuthorityId::for_tag("nrpc:svc");
let base = RpcRequestPayload {
service: "svc".to_string(),
deadline_ns: 0,
flags: 0,
headers: vec![],
body: Bytes::from_static(b"ping"),
};
let digest = org_request_digest(&base).expect("digest");
let membership = OrgMembershipCert::try_issue(&org_b, caller_entity.clone(), 1, 3600)
.expect("caller cert");
let dispatcher = OrgDispatcherGrant::try_issue(
&org_b,
caller_entity.clone(),
DispatcherScope::Exact(cap),
3600,
)
.expect("dispatcher grant");
let expiry = (current_timestamp() + 20) * 1_000_000_000;
let make_frame = |signed_call_id: u64, frame_call_id: u64| {
let proof = OrgCallProof::sign_for_call(
&caller_kp,
membership.clone(),
dispatcher.clone(),
None,
org_b.org_id(),
org_b.org_id(),
node_entity.clone(),
signed_call_id,
cap,
expiry,
digest,
);
let mut payload = base.clone();
payload.headers.push((
ORG_ADMISSION_HEADER.to_string(),
proof.encode().expect("encode"),
));
let mut frame =
EventMeta::new(DISPATCH_RPC_REQUEST, 0, caller_origin, frame_call_id, 0)
.to_bytes()
.to_vec();
encode_rpc_route(&mut frame, 0);
frame.extend_from_slice(&payload.encode());
RpcInboundEvent {
channel_hash,
origin_hash: caller_origin,
from_node: CALLER_NODE,
payload: Bytes::from(frame),
}
};
assert!(server.deliver_rpc_inbound_for_test(channel_hash, make_frame(7, 8)));
assert!(server.deliver_rpc_inbound_for_test(channel_hash, make_frame(9, 9)));
assert!(
wait_until_at_least(|| admits.load(Ordering::SeqCst) as u64, 1).await,
"the valid call must be admitted",
);
assert_eq!(
admits.load(Ordering::SeqCst),
1,
"the tampered-binding call was denied (handler never ran for it)",
);
}
#[tokio::test]
async fn caller_proof_intent_produces_an_admissible_proof() {
use crate::adapter::net::behavior::org::{OrgKeypair, OrgMembershipCert};
use crate::adapter::net::behavior::org_admission::{Admitted, OrgAdmission};
use crate::adapter::net::behavior::org_authority::NodeAuthority;
use crate::adapter::net::behavior::org_grant::{
CapabilityAuthorityId, DispatcherScope, OrgDispatcherGrant,
};
type Seen = std::sync::Arc<Mutex<Option<Admitted>>>;
struct SpyHandler(Seen);
#[async_trait::async_trait]
impl RpcHandler for SpyHandler {
async fn call(&self, ctx: RpcContext) -> Result<RpcResponsePayload, RpcHandlerError> {
*self.0.lock() = Some(ctx.org_admission.clone().expect("admitted call"));
Ok(RpcResponsePayload {
status: RpcStatus::Ok,
headers: vec![],
body: Bytes::new(),
})
}
}
let server = build_server().await;
let node_entity = server.entity_id().clone();
let org_b = OrgKeypair::from_bytes([0x42u8; 32]);
let node_cert =
OrgMembershipCert::try_issue(&org_b, node_entity.clone(), 1, 3600).expect("node cert");
let dir = std::env::temp_dir().join(format!("net-oa2-caller-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
let authority =
NodeAuthority::adopt(&dir, node_cert, &node_entity, 0, None).expect("adopt authority");
server
.install_node_authority(std::sync::Arc::new(authority))
.expect("install authority");
let seen: Seen = std::sync::Arc::new(Mutex::new(None));
let serve = server
.serve_rpc_protected(
"svc",
std::sync::Arc::new(SpyHandler(seen.clone())),
OrgAdmission::OwnerDelegated,
std::sync::Arc::new(|_| true),
)
.expect("serve protected");
let channel_hash = serve.channel_hash;
const CALLER_NODE: u64 = 0x95;
let caller_kp = EntityKeypair::generate();
let caller_entity = caller_kp.entity_id().clone();
let caller_origin = caller_entity.origin_hash();
server.test_pin_peer_entity(CALLER_NODE, caller_entity.clone());
let cap = CapabilityAuthorityId::for_tag("nrpc:svc");
let membership = OrgMembershipCert::try_issue(&org_b, caller_entity.clone(), 1, 3600)
.expect("caller cert");
let dispatcher = OrgDispatcherGrant::try_issue(
&org_b,
caller_entity.clone(),
DispatcherScope::Exact(cap),
3600,
)
.expect("dispatcher grant");
let intent = OrgProofIntent {
caller: std::sync::Arc::new(caller_kp),
membership,
dispatcher,
capability_grant: None,
acting_org: org_b.org_id(),
provider_owner_org: org_b.org_id(),
provider: node_entity.clone(),
capability: cap,
proof_ttl_secs: 30,
};
let call_id = 11u64;
let mut req = RpcRequestPayload {
service: "svc".to_string(),
deadline_ns: 0,
flags: 0,
headers: vec![],
body: Bytes::from_static(b"ping"),
};
req.headers
.push(sign_admission_proof(&intent, call_id, &req).expect("sign proof"));
let mut frame = EventMeta::new(DISPATCH_RPC_REQUEST, 0, caller_origin, call_id, 0)
.to_bytes()
.to_vec();
encode_rpc_route(&mut frame, 0);
frame.extend_from_slice(&req.encode());
assert!(server.deliver_rpc_inbound_for_test(
channel_hash,
RpcInboundEvent {
channel_hash,
origin_hash: caller_origin,
from_node: CALLER_NODE,
payload: Bytes::from(frame),
}
));
assert!(
wait_until_at_least(|| seen.lock().is_some() as u64, 1).await,
"the caller-built proof must be admitted by the live gate",
);
let admitted = seen.lock().clone().expect("admitted");
assert_eq!(admitted.caller, caller_entity);
assert_eq!(admitted.provider, node_entity);
assert_eq!(admitted.capability, cap);
}
#[tokio::test]
async fn protected_cross_org_call_admits_end_to_end() {
use crate::adapter::net::behavior::org::{OrgKeypair, OrgMembershipCert};
use crate::adapter::net::behavior::org_admission::{Admitted, OrgAdmission};
use crate::adapter::net::behavior::org_authority::NodeAuthority;
use crate::adapter::net::behavior::org_grant::{
CapabilityAuthorityId, DispatcherScope, GrantRights, GrantTargetScope,
OrgCapabilityGrant, OrgDispatcherGrant,
};
type Seen = std::sync::Arc<Mutex<Option<Admitted>>>;
struct SpyHandler(Seen);
#[async_trait::async_trait]
impl RpcHandler for SpyHandler {
async fn call(&self, ctx: RpcContext) -> Result<RpcResponsePayload, RpcHandlerError> {
*self.0.lock() = Some(ctx.org_admission.clone().expect("admitted call"));
Ok(RpcResponsePayload {
status: RpcStatus::Ok,
headers: vec![],
body: Bytes::new(),
})
}
}
let server = build_server().await;
let node_entity = server.entity_id().clone();
let org_b = OrgKeypair::from_bytes([0x42u8; 32]); let org_a = OrgKeypair::from_bytes([0x77u8; 32]); let node_cert =
OrgMembershipCert::try_issue(&org_b, node_entity.clone(), 1, 3600).expect("node cert");
let dir = std::env::temp_dir().join(format!("net-oa2-xorg-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
let authority =
NodeAuthority::adopt(&dir, node_cert, &node_entity, 0, None).expect("adopt authority");
server
.install_node_authority(std::sync::Arc::new(authority))
.expect("install authority");
let seen: Seen = std::sync::Arc::new(Mutex::new(None));
let serve = server
.serve_rpc_protected(
"svc",
std::sync::Arc::new(SpyHandler(seen.clone())),
OrgAdmission::CrossOrgGranted,
std::sync::Arc::new(|_| true),
)
.expect("serve protected");
let channel_hash = serve.channel_hash;
const CALLER_NODE: u64 = 0x97;
let caller_kp = EntityKeypair::generate();
let caller_entity = caller_kp.entity_id().clone();
let caller_origin = caller_entity.origin_hash();
server.test_pin_peer_entity(CALLER_NODE, caller_entity.clone());
let cap = CapabilityAuthorityId::for_tag("nrpc:svc");
let membership = OrgMembershipCert::try_issue(&org_a, caller_entity.clone(), 1, 3600)
.expect("caller cert");
let dispatcher = OrgDispatcherGrant::try_issue(
&org_a,
caller_entity.clone(),
DispatcherScope::Exact(cap),
3600,
)
.expect("dispatcher grant");
let (grant, _) = OrgCapabilityGrant::try_issue(
&org_b,
org_a.org_id(),
cap,
GrantRights::INVOKE,
GrantTargetScope::ExactNode(node_entity.clone()),
3600,
)
.expect("capability grant");
let intent = OrgProofIntent {
caller: std::sync::Arc::new(caller_kp),
membership,
dispatcher,
capability_grant: Some(grant),
acting_org: org_a.org_id(),
provider_owner_org: org_b.org_id(),
provider: node_entity.clone(),
capability: cap,
proof_ttl_secs: 30,
};
let call_id = 13u64;
let mut req = RpcRequestPayload {
service: "svc".to_string(),
deadline_ns: 0,
flags: 0,
headers: vec![],
body: Bytes::from_static(b"ping"),
};
req.headers
.push(sign_admission_proof(&intent, call_id, &req).expect("sign proof"));
let mut frame = EventMeta::new(DISPATCH_RPC_REQUEST, 0, caller_origin, call_id, 0)
.to_bytes()
.to_vec();
encode_rpc_route(&mut frame, 0);
frame.extend_from_slice(&req.encode());
assert!(server.deliver_rpc_inbound_for_test(
channel_hash,
RpcInboundEvent {
channel_hash,
origin_hash: caller_origin,
from_node: CALLER_NODE,
payload: Bytes::from(frame),
}
));
assert!(
wait_until_at_least(|| seen.lock().is_some() as u64, 1).await,
"a valid cross-org call must be admitted",
);
let admitted = seen.lock().clone().expect("admitted");
assert_eq!(admitted.acting_org, org_a.org_id(), "acting org A");
assert_eq!(admitted.provider_org, org_b.org_id(), "provider org B");
assert_ne!(
admitted.acting_org, admitted.provider_org,
"cross-org: A and B are distinct",
);
}
#[tokio::test]
async fn cross_org_admission_denial_matrix() {
use crate::adapter::net::behavior::org::{
current_timestamp, OrgKeypair, OrgMembershipCert,
};
use crate::adapter::net::behavior::org_admission::OrgAdmission;
use crate::adapter::net::behavior::org_authority::NodeAuthority;
use crate::adapter::net::behavior::org_call::{OrgCallProof, ORG_ADMISSION_HEADER};
use crate::adapter::net::behavior::org_grant::{
CapabilityAuthorityId, DispatcherScope, GrantRights, GrantTargetScope,
OrgCapabilityGrant, OrgDispatcherGrant,
};
use crate::adapter::net::org_admission_gate::org_request_digest;
use std::sync::atomic::{AtomicUsize, Ordering};
struct Counter(std::sync::Arc<AtomicUsize>);
#[async_trait::async_trait]
impl RpcHandler for Counter {
async fn call(&self, ctx: RpcContext) -> Result<RpcResponsePayload, RpcHandlerError> {
assert!(
ctx.org_admission.is_some(),
"only admitted calls may reach the handler",
);
self.0.fetch_add(1, Ordering::SeqCst);
Ok(RpcResponsePayload {
status: RpcStatus::Ok,
headers: vec![],
body: Bytes::new(),
})
}
}
let server = build_server().await;
let node_entity = server.entity_id().clone();
let org_b = OrgKeypair::from_bytes([0x42u8; 32]); let org_a = OrgKeypair::from_bytes([0x7au8; 32]); let org_c = OrgKeypair::from_bytes([0x33u8; 32]); let node_cert =
OrgMembershipCert::try_issue(&org_b, node_entity.clone(), 1, 3600).expect("node cert");
let dir = std::env::temp_dir().join(format!("net-oa4-xorg-deny-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
let authority =
NodeAuthority::adopt(&dir, node_cert, &node_entity, 0, None).expect("adopt authority");
server
.install_node_authority(std::sync::Arc::new(authority))
.expect("install authority");
let admits = std::sync::Arc::new(AtomicUsize::new(0));
let serve = server
.serve_rpc_protected(
"svc",
std::sync::Arc::new(Counter(admits.clone())),
OrgAdmission::CrossOrgGranted,
std::sync::Arc::new(|_| true),
)
.expect("serve protected");
let channel_hash = serve.channel_hash;
const CALLER_NODE: u64 = 0xA1;
let caller_kp = EntityKeypair::from_bytes([0x1bu8; 32]);
let caller_entity = caller_kp.entity_id().clone();
let caller_origin = caller_entity.origin_hash();
server.test_pin_peer_entity(CALLER_NODE, caller_entity.clone());
let cap = CapabilityAuthorityId::for_tag("nrpc:svc");
let base = RpcRequestPayload {
service: "svc".to_string(),
deadline_ns: 0,
flags: 0,
headers: vec![],
body: Bytes::from_static(b"ping"),
};
let digest = org_request_digest(&base).expect("digest");
let future = (current_timestamp() + 20) * 1_000_000_000;
let membership_a = OrgMembershipCert::try_issue(&org_a, caller_entity.clone(), 1, 3600)
.expect("membership");
let dispatcher_a = OrgDispatcherGrant::try_issue(
&org_a,
caller_entity.clone(),
DispatcherScope::Exact(cap),
3600,
)
.expect("dispatcher");
let good_grant = OrgCapabilityGrant::try_issue(
&org_b,
org_a.org_id(),
cap,
GrantRights::INVOKE,
GrantTargetScope::ExactNode(node_entity.clone()),
3600,
)
.expect("grant")
.0;
let make_event = |headers: Vec<Vec<u8>>, call_id: u64, body: &'static [u8]| {
let mut payload = base.clone();
payload.body = Bytes::from_static(body);
payload.headers = headers
.into_iter()
.map(|h| (ORG_ADMISSION_HEADER.to_string(), h))
.collect();
let mut frame = EventMeta::new(DISPATCH_RPC_REQUEST, 0, caller_origin, call_id, 0)
.to_bytes()
.to_vec();
encode_rpc_route(&mut frame, 0);
frame.extend_from_slice(&payload.encode());
RpcInboundEvent {
channel_hash,
origin_hash: caller_origin,
from_node: CALLER_NODE,
payload: Bytes::from(frame),
}
};
let sign = |kp: &EntityKeypair,
membership: OrgMembershipCert,
dispatcher: OrgDispatcherGrant,
grant: Option<OrgCapabilityGrant>,
call_id: u64,
expiry: u64,
dgst: [u8; 32]|
-> Vec<u8> {
OrgCallProof::sign_for_call(
kp,
membership,
dispatcher,
grant,
org_a.org_id(),
org_b.org_id(),
node_entity.clone(),
call_id,
cap,
expiry,
dgst,
)
.encode()
.expect("encode proof")
};
let good = |call_id: u64| {
sign(
&caller_kp,
membership_a.clone(),
dispatcher_a.clone(),
Some(good_grant.clone()),
call_id,
future,
digest,
)
};
let wrong_grantee = OrgCapabilityGrant::try_issue(
&org_b,
org_c.org_id(),
cap,
GrantRights::INVOKE,
GrantTargetScope::ExactNode(node_entity.clone()),
3600,
)
.expect("grant")
.0;
server.deliver_rpc_inbound_for_test(
channel_hash,
make_event(
vec![sign(
&caller_kp,
membership_a.clone(),
dispatcher_a.clone(),
Some(wrong_grantee),
101,
future,
digest,
)],
101,
b"ping",
),
);
let foreign_issuer = OrgCapabilityGrant::try_issue(
&org_c,
org_a.org_id(),
cap,
GrantRights::INVOKE,
GrantTargetScope::ExactNode(node_entity.clone()),
3600,
)
.expect("grant")
.0;
server.deliver_rpc_inbound_for_test(
channel_hash,
make_event(
vec![sign(
&caller_kp,
membership_a.clone(),
dispatcher_a.clone(),
Some(foreign_issuer),
102,
future,
digest,
)],
102,
b"ping",
),
);
let discover_only = OrgCapabilityGrant::try_issue(
&org_b,
org_a.org_id(),
cap,
GrantRights::DISCOVER,
GrantTargetScope::ExactNode(node_entity.clone()),
3600,
)
.expect("grant")
.0;
server.deliver_rpc_inbound_for_test(
channel_hash,
make_event(
vec![sign(
&caller_kp,
membership_a.clone(),
dispatcher_a.clone(),
Some(discover_only),
103,
future,
digest,
)],
103,
b"ping",
),
);
server.deliver_rpc_inbound_for_test(
channel_hash,
make_event(
vec![sign(
&caller_kp,
membership_a.clone(),
dispatcher_a.clone(),
None,
104,
future,
digest,
)],
104,
b"ping",
),
);
let other_entity = EntityKeypair::from_bytes([0xeeu8; 32]).entity_id().clone();
let wrong_target = OrgCapabilityGrant::try_issue(
&org_b,
org_a.org_id(),
cap,
GrantRights::INVOKE,
GrantTargetScope::ExactNode(other_entity),
3600,
)
.expect("grant")
.0;
server.deliver_rpc_inbound_for_test(
channel_hash,
make_event(
vec![sign(
&caller_kp,
membership_a.clone(),
dispatcher_a.clone(),
Some(wrong_target),
105,
future,
digest,
)],
105,
b"ping",
),
);
let wrong_cap = OrgCapabilityGrant::try_issue(
&org_b,
org_a.org_id(),
CapabilityAuthorityId::for_tag("nrpc:other"),
GrantRights::INVOKE,
GrantTargetScope::ExactNode(node_entity.clone()),
3600,
)
.expect("grant")
.0;
server.deliver_rpc_inbound_for_test(
channel_hash,
make_event(
vec![sign(
&caller_kp,
membership_a.clone(),
dispatcher_a.clone(),
Some(wrong_cap),
106,
future,
digest,
)],
106,
b"ping",
),
);
server
.deliver_rpc_inbound_for_test(channel_hash, make_event(vec![good(107)], 107, b"pong"));
let past = current_timestamp().saturating_sub(100) * 1_000_000_000;
server.deliver_rpc_inbound_for_test(
channel_hash,
make_event(
vec![sign(
&caller_kp,
membership_a.clone(),
dispatcher_a.clone(),
Some(good_grant.clone()),
108,
past,
digest,
)],
108,
b"ping",
),
);
let x_kp = EntityKeypair::from_bytes([0x2cu8; 32]);
let x_entity = x_kp.entity_id().clone();
let membership_x =
OrgMembershipCert::try_issue(&org_a, x_entity.clone(), 1, 3600).expect("x membership");
let dispatcher_x =
OrgDispatcherGrant::try_issue(&org_a, x_entity, DispatcherScope::Exact(cap), 3600)
.expect("x dispatcher");
server.deliver_rpc_inbound_for_test(
channel_hash,
make_event(
vec![sign(
&x_kp,
membership_x,
dispatcher_x,
Some(good_grant.clone()),
109,
future,
digest,
)],
109,
b"ping",
),
);
server.deliver_rpc_inbound_for_test(channel_hash, make_event(vec![], 110, b"ping"));
server.deliver_rpc_inbound_for_test(
channel_hash,
make_event(vec![good(111), good(111)], 111, b"ping"),
);
let v1 = good(200);
server
.deliver_rpc_inbound_for_test(channel_hash, make_event(vec![v1.clone()], 200, b"ping"));
server.deliver_rpc_inbound_for_test(channel_hash, make_event(vec![v1], 200, b"ping"));
server
.deliver_rpc_inbound_for_test(channel_hash, make_event(vec![good(201)], 201, b"ping"));
assert!(
wait_until_at_least(|| admits.load(Ordering::SeqCst) as u64, 2).await,
"the two genuine cross-org calls must be admitted",
);
assert_eq!(
admits.load(Ordering::SeqCst),
2,
"exactly the two genuine calls admitted — every adversarial frame was denied \
(handler dark) and the replay of the first genuine call was refused",
);
}
#[tokio::test]
async fn owner_delegated_admission_denial_matrix() {
use crate::adapter::net::behavior::org::{
current_timestamp, OrgKeypair, OrgMembershipCert,
};
use crate::adapter::net::behavior::org_admission::OrgAdmission;
use crate::adapter::net::behavior::org_authority::NodeAuthority;
use crate::adapter::net::behavior::org_call::{OrgCallProof, ORG_ADMISSION_HEADER};
use crate::adapter::net::behavior::org_grant::{
CapabilityAuthorityId, DispatcherScope, GrantRights, GrantTargetScope,
OrgCapabilityGrant, OrgDispatcherGrant,
};
use crate::adapter::net::org_admission_gate::org_request_digest;
use std::sync::atomic::{AtomicUsize, Ordering};
struct Counter(std::sync::Arc<AtomicUsize>);
#[async_trait::async_trait]
impl RpcHandler for Counter {
async fn call(&self, ctx: RpcContext) -> Result<RpcResponsePayload, RpcHandlerError> {
assert!(
ctx.org_admission.is_some(),
"only admitted calls may reach the handler",
);
self.0.fetch_add(1, Ordering::SeqCst);
Ok(RpcResponsePayload {
status: RpcStatus::Ok,
headers: vec![],
body: Bytes::new(),
})
}
}
let server = build_server().await;
let node_entity = server.entity_id().clone();
let org_b = OrgKeypair::from_bytes([0x42u8; 32]); let org_a = OrgKeypair::from_bytes([0x7au8; 32]); let node_cert =
OrgMembershipCert::try_issue(&org_b, node_entity.clone(), 1, 3600).expect("node cert");
let dir = std::env::temp_dir().join(format!("net-oa4-owner-deny-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
let authority =
NodeAuthority::adopt(&dir, node_cert, &node_entity, 0, None).expect("adopt authority");
server
.install_node_authority(std::sync::Arc::new(authority))
.expect("install authority");
let admits = std::sync::Arc::new(AtomicUsize::new(0));
let serve = server
.serve_rpc_protected(
"svc",
std::sync::Arc::new(Counter(admits.clone())),
OrgAdmission::OwnerDelegated,
std::sync::Arc::new(|_| true),
)
.expect("serve protected");
let channel_hash = serve.channel_hash;
const CALLER_NODE: u64 = 0xB1;
let caller_kp = EntityKeypair::from_bytes([0x3bu8; 32]);
let caller_entity = caller_kp.entity_id().clone();
let caller_origin = caller_entity.origin_hash();
server.test_pin_peer_entity(CALLER_NODE, caller_entity.clone());
let cap = CapabilityAuthorityId::for_tag("nrpc:svc");
let base = RpcRequestPayload {
service: "svc".to_string(),
deadline_ns: 0,
flags: 0,
headers: vec![],
body: Bytes::from_static(b"ping"),
};
let digest = org_request_digest(&base).expect("digest");
let future = (current_timestamp() + 20) * 1_000_000_000;
let membership_b = OrgMembershipCert::try_issue(&org_b, caller_entity.clone(), 1, 3600)
.expect("membership");
let dispatcher_b = OrgDispatcherGrant::try_issue(
&org_b,
caller_entity.clone(),
DispatcherScope::Exact(cap),
3600,
)
.expect("dispatcher");
let make_event = |headers: Vec<Vec<u8>>, call_id: u64, body: &'static [u8]| {
let mut payload = base.clone();
payload.body = Bytes::from_static(body);
payload.headers = headers
.into_iter()
.map(|h| (ORG_ADMISSION_HEADER.to_string(), h))
.collect();
let mut frame = EventMeta::new(DISPATCH_RPC_REQUEST, 0, caller_origin, call_id, 0)
.to_bytes()
.to_vec();
encode_rpc_route(&mut frame, 0);
frame.extend_from_slice(&payload.encode());
RpcInboundEvent {
channel_hash,
origin_hash: caller_origin,
from_node: CALLER_NODE,
payload: Bytes::from(frame),
}
};
#[allow(clippy::too_many_arguments)]
let sign = |kp: &EntityKeypair,
membership: OrgMembershipCert,
dispatcher: OrgDispatcherGrant,
grant: Option<OrgCapabilityGrant>,
acting_org: crate::adapter::net::behavior::org::OrgId,
callee: crate::adapter::net::identity::EntityId,
capability: CapabilityAuthorityId,
call_id: u64,
expiry: u64,
dgst: [u8; 32]|
-> Vec<u8> {
OrgCallProof::sign_for_call(
kp,
membership,
dispatcher,
grant,
acting_org,
org_b.org_id(),
callee,
call_id,
capability,
expiry,
dgst,
)
.encode()
.expect("encode proof")
};
let good = |call_id: u64| {
sign(
&caller_kp,
membership_b.clone(),
dispatcher_b.clone(),
None,
org_b.org_id(),
node_entity.clone(),
cap,
call_id,
future,
digest,
)
};
let dispatcher_other = OrgDispatcherGrant::try_issue(
&org_b,
caller_entity.clone(),
DispatcherScope::Exact(CapabilityAuthorityId::for_tag("nrpc:other")),
3600,
)
.expect("dispatcher");
server.deliver_rpc_inbound_for_test(
channel_hash,
make_event(
vec![sign(
&caller_kp,
membership_b.clone(),
dispatcher_other,
None,
org_b.org_id(),
node_entity.clone(),
cap,
401,
future,
digest,
)],
401,
b"ping",
),
);
let x_kp = EntityKeypair::from_bytes([0x3cu8; 32]);
let x_entity = x_kp.entity_id().clone();
let membership_x =
OrgMembershipCert::try_issue(&org_b, x_entity.clone(), 1, 3600).expect("x membership");
let dispatcher_x =
OrgDispatcherGrant::try_issue(&org_b, x_entity, DispatcherScope::Exact(cap), 3600)
.expect("x dispatcher");
server.deliver_rpc_inbound_for_test(
channel_hash,
make_event(
vec![sign(
&x_kp,
membership_x,
dispatcher_x,
None,
org_b.org_id(),
node_entity.clone(),
cap,
402,
future,
digest,
)],
402,
b"ping",
),
);
let other_entity = EntityKeypair::from_bytes([0xefu8; 32]).entity_id().clone();
server.deliver_rpc_inbound_for_test(
channel_hash,
make_event(
vec![sign(
&caller_kp,
membership_b.clone(),
dispatcher_b.clone(),
None,
org_b.org_id(),
other_entity,
cap,
403,
future,
digest,
)],
403,
b"ping",
),
);
server.deliver_rpc_inbound_for_test(
channel_hash,
make_event(
vec![sign(
&caller_kp,
membership_b.clone(),
dispatcher_b.clone(),
None,
org_b.org_id(),
node_entity.clone(),
CapabilityAuthorityId::for_tag("nrpc:other"),
404,
future,
digest,
)],
404,
b"ping",
),
);
server
.deliver_rpc_inbound_for_test(channel_hash, make_event(vec![good(405)], 405, b"pong"));
let past = current_timestamp().saturating_sub(100) * 1_000_000_000;
server.deliver_rpc_inbound_for_test(
channel_hash,
make_event(
vec![sign(
&caller_kp,
membership_b.clone(),
dispatcher_b.clone(),
None,
org_b.org_id(),
node_entity.clone(),
cap,
406,
past,
digest,
)],
406,
b"ping",
),
);
let stray_grant = OrgCapabilityGrant::try_issue(
&org_b,
org_b.org_id(),
cap,
GrantRights::INVOKE,
GrantTargetScope::ExactNode(node_entity.clone()),
3600,
)
.expect("grant")
.0;
server.deliver_rpc_inbound_for_test(
channel_hash,
make_event(
vec![sign(
&caller_kp,
membership_b.clone(),
dispatcher_b.clone(),
Some(stray_grant),
org_b.org_id(),
node_entity.clone(),
cap,
407,
future,
digest,
)],
407,
b"ping",
),
);
let membership_a = OrgMembershipCert::try_issue(&org_a, caller_entity.clone(), 1, 3600)
.expect("a membership");
let dispatcher_a = OrgDispatcherGrant::try_issue(
&org_a,
caller_entity.clone(),
DispatcherScope::Exact(cap),
3600,
)
.expect("a dispatcher");
server.deliver_rpc_inbound_for_test(
channel_hash,
make_event(
vec![sign(
&caller_kp,
membership_a,
dispatcher_a,
None,
org_a.org_id(),
node_entity.clone(),
cap,
408,
future,
digest,
)],
408,
b"ping",
),
);
server.deliver_rpc_inbound_for_test(channel_hash, make_event(vec![], 409, b"ping"));
server.deliver_rpc_inbound_for_test(
channel_hash,
make_event(vec![good(410), good(410)], 410, b"ping"),
);
let v1 = good(500);
server
.deliver_rpc_inbound_for_test(channel_hash, make_event(vec![v1.clone()], 500, b"ping"));
server.deliver_rpc_inbound_for_test(channel_hash, make_event(vec![v1], 500, b"ping"));
server
.deliver_rpc_inbound_for_test(channel_hash, make_event(vec![good(501)], 501, b"ping"));
assert!(
wait_until_at_least(|| admits.load(Ordering::SeqCst) as u64, 2).await,
"the two genuine owner-delegated calls must be admitted",
);
assert_eq!(
admits.load(Ordering::SeqCst),
2,
"exactly the two genuine calls admitted — every adversarial frame was denied \
(handler dark) and the replay of the first genuine call was refused",
);
}
#[tokio::test]
async fn seam_red_org_admission_is_load_bearing() {
use crate::adapter::net::behavior::fold::capability::{
CapabilityFold, CapabilityMembership,
};
use crate::adapter::net::behavior::fold::capability_bridge::{
has_local_capability, may_execute,
};
use crate::adapter::net::behavior::fold::{
EnvelopeMeta, FoldKind, NodeState, SignedAnnouncement,
};
use crate::adapter::net::behavior::org::{
current_timestamp, OrgKeypair, OrgMembershipCert,
};
use crate::adapter::net::behavior::org_admission::OrgAdmission;
use crate::adapter::net::behavior::org_authority::NodeAuthority;
use crate::adapter::net::behavior::org_call::{OrgCallProof, ORG_ADMISSION_HEADER};
use crate::adapter::net::behavior::org_grant::{
CapabilityAuthorityId, DispatcherScope, GrantRights, GrantTargetScope,
OrgCapabilityGrant, OrgDispatcherGrant,
};
use crate::adapter::net::org_admission_gate::org_request_digest;
use std::sync::atomic::{AtomicUsize, Ordering};
struct Counter(std::sync::Arc<AtomicUsize>);
#[async_trait::async_trait]
impl RpcHandler for Counter {
async fn call(&self, _ctx: RpcContext) -> Result<RpcResponsePayload, RpcHandlerError> {
self.0.fetch_add(1, Ordering::SeqCst);
Ok(RpcResponsePayload {
status: RpcStatus::Ok,
headers: vec![],
body: Bytes::new(),
})
}
}
let server = build_server().await;
let node_entity = server.entity_id().clone();
let org_b = OrgKeypair::from_bytes([0x42u8; 32]); let org_a = OrgKeypair::from_bytes([0x7au8; 32]); let node_cert =
OrgMembershipCert::try_issue(&org_b, node_entity.clone(), 1, 3600).expect("node cert");
let dir = std::env::temp_dir().join(format!("net-oa4-seamred-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
let authority =
NodeAuthority::adopt(&dir, node_cert, &node_entity, 0, None).expect("adopt authority");
server
.install_node_authority(std::sync::Arc::new(authority))
.expect("install authority");
const CALLER_NODE: u64 = 0xC1;
let caller_kp = EntityKeypair::from_bytes([0x4bu8; 32]);
let caller_entity = caller_kp.entity_id().clone();
let caller_origin = caller_entity.origin_hash();
server.test_pin_peer_entity(CALLER_NODE, caller_entity.clone());
let cap = CapabilityAuthorityId::for_tag("nrpc:svc");
let base = RpcRequestPayload {
service: "svc".to_string(),
deadline_ns: 0,
flags: 0,
headers: vec![],
body: Bytes::from_static(b"ping"),
};
let digest = org_request_digest(&base).expect("digest");
let future = (current_timestamp() + 20) * 1_000_000_000;
let membership = OrgMembershipCert::try_issue(&org_a, caller_entity.clone(), 1, 3600)
.expect("membership");
let dispatcher = OrgDispatcherGrant::try_issue(
&org_a,
caller_entity.clone(),
DispatcherScope::Exact(cap),
3600,
)
.expect("dispatcher");
let grant = OrgCapabilityGrant::try_issue(
&org_b,
org_a.org_id(),
cap,
GrantRights::INVOKE,
GrantTargetScope::ExactNode(node_entity.clone()),
3600,
)
.expect("grant")
.0;
let valid_proof = |call_id: u64| {
OrgCallProof::sign_for_call(
&caller_kp,
membership.clone(),
dispatcher.clone(),
Some(grant.clone()),
org_a.org_id(),
org_b.org_id(),
node_entity.clone(),
call_id,
cap,
future,
digest,
)
.encode()
.expect("encode proof")
};
let make_event = |ch: u64, headers: Vec<Vec<u8>>, call_id: u64| {
let mut payload = base.clone();
payload.headers = headers
.into_iter()
.map(|h| (ORG_ADMISSION_HEADER.to_string(), h))
.collect();
let mut frame = EventMeta::new(DISPATCH_RPC_REQUEST, 0, caller_origin, call_id, 0)
.to_bytes()
.to_vec();
encode_rpc_route(&mut frame, 0);
frame.extend_from_slice(&payload.encode());
RpcInboundEvent {
channel_hash: ch,
origin_hash: caller_origin,
from_node: CALLER_NODE,
payload: Bytes::from(frame),
}
};
let _serve_d = server
.serve_rpc(
"d",
std::sync::Arc::new(Counter(std::sync::Arc::new(AtomicUsize::new(0)))),
)
.expect("serve public D");
let admits = std::sync::Arc::new(AtomicUsize::new(0));
let serve_enforced = server
.serve_rpc_protected(
"svc",
std::sync::Arc::new(Counter(admits.clone())),
OrgAdmission::CrossOrgGranted,
std::sync::Arc::new(|_| true),
)
.expect("serve enforced");
let ch = serve_enforced.channel_hash;
assert!(
may_execute(
server.capability_fold(),
server.node_id(),
"nrpc:svc",
CALLER_NODE
),
"before the unrelated restricted D entry, C's own class-0 entry permits S \
(empty allow-lists)",
);
let d_kp = EntityKeypair::generate();
let d_entry = SignedAnnouncement::sign(
&d_kp,
CapabilityFold::KIND_ID,
0xD00D,
server.node_id(),
1,
EnvelopeMeta::default(),
CapabilityMembership {
class_hash: 0xD00D,
tags: vec!["nrpc:d".to_string()],
hardware: None,
state: NodeState::Idle,
region: None,
price_quote: None,
reflex_addr: None,
allowed_nodes: vec![0xDEAD],
allowed_subnets: vec![],
allowed_groups: vec![],
metadata: std::collections::BTreeMap::new(),
owner: None,
},
)
.expect("sign restrictive D");
server
.capability_fold()
.apply(d_entry)
.expect("apply restrictive D entry");
assert!(
has_local_capability(server.capability_fold(), server.node_id(), "nrpc:svc"),
"the provider holds C locally",
);
assert!(
!may_execute(
server.capability_fold(),
server.node_id(),
"nrpc:svc",
CALLER_NODE
),
"may_execute(P, C, S) is now false ONLY via the unrelated D entry's \
target-wide aggregation — C's own entry is still unrestricted",
);
assert!(
!may_execute(
server.capability_fold(),
server.node_id(),
"nrpc:d",
CALLER_NODE
),
"the unrelated D is itself governed by may_execute (S excluded)",
);
server.deliver_rpc_inbound_for_test(ch, make_event(ch, vec![valid_proof(700)], 700));
server.deliver_rpc_inbound_for_test(ch, make_event(ch, vec![], 701));
server.deliver_rpc_inbound_for_test(ch, make_event(ch, vec![valid_proof(702)], 702));
assert!(
wait_until_at_least(|| admits.load(Ordering::SeqCst) as u64, 2).await,
"both valid proofs are admitted through the protected path",
);
assert_eq!(
admits.load(Ordering::SeqCst),
2,
"the two valid proofs admitted (legacy gate cannot block C); the no-proof \
request was denied under enforced admission",
);
drop(serve_enforced);
let serve_disabled = server
.serve_rpc_protected_red_witness_disabled(
"svc",
std::sync::Arc::new(Counter(admits.clone())),
OrgAdmission::CrossOrgGranted,
std::sync::Arc::new(|_| true),
)
.expect("serve red-witness-disabled");
let ch2 = serve_disabled.channel_hash;
assert!(
has_local_capability(server.capability_fold(), server.node_id(), "nrpc:svc"),
"C is present locally after the re-registration",
);
assert!(
!may_execute(
server.capability_fold(),
server.node_id(),
"nrpc:svc",
CALLER_NODE
),
"may_execute(P, C, S) stays false via the surviving unrelated D entry",
);
server.deliver_rpc_inbound_for_test(ch2, make_event(ch2, vec![], 703));
assert!(
wait_until_at_least(|| admits.load(Ordering::SeqCst) as u64, 3).await,
"RED: with org admission disabled, the unauthorized protected call ran the handler",
);
assert_eq!(
admits.load(Ordering::SeqCst),
3,
"disabling ONLY verify_org_admission let the unauthorized protected call execute — \
OrgAdmission is load-bearing, independent of the legacy may_execute verdict",
);
drop(serve_disabled);
}
#[tokio::test]
async fn protected_gate_binds_authenticated_identity_to_claimed_origin() {
use crate::adapter::net::behavior::org::{OrgKeypair, OrgMembershipCert};
use crate::adapter::net::behavior::org_admission::OrgAdmission;
use crate::adapter::net::behavior::org_authority::NodeAuthority;
use crate::adapter::net::behavior::org_grant::{
CapabilityAuthorityId, DispatcherScope, OrgDispatcherGrant,
};
use std::sync::atomic::{AtomicUsize, Ordering};
struct Counter(std::sync::Arc<AtomicUsize>);
#[async_trait::async_trait]
impl RpcHandler for Counter {
async fn call(&self, _ctx: RpcContext) -> Result<RpcResponsePayload, RpcHandlerError> {
self.0.fetch_add(1, Ordering::SeqCst);
Ok(RpcResponsePayload {
status: RpcStatus::Ok,
headers: vec![],
body: Bytes::new(),
})
}
}
let server = build_server().await;
let node_entity = server.entity_id().clone();
let org_b = OrgKeypair::from_bytes([0x42u8; 32]);
let node_cert =
OrgMembershipCert::try_issue(&org_b, node_entity.clone(), 1, 3600).expect("node cert");
let dir = std::env::temp_dir().join(format!("net-oa2-b1-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
let authority =
NodeAuthority::adopt(&dir, node_cert, &node_entity, 0, None).expect("adopt authority");
server
.install_node_authority(std::sync::Arc::new(authority))
.expect("install authority");
let admits = std::sync::Arc::new(AtomicUsize::new(0));
let serve = server
.serve_rpc_protected(
"svc",
std::sync::Arc::new(Counter(admits.clone())),
OrgAdmission::OwnerDelegated,
std::sync::Arc::new(|_| true),
)
.expect("serve protected");
let channel_hash = serve.channel_hash;
const CALLER_NODE: u64 = 0x99;
let caller_kp = EntityKeypair::generate();
let caller_entity = caller_kp.entity_id().clone();
let caller_origin = caller_entity.origin_hash();
let victim_origin = caller_origin ^ 0xFFFF_FFFF; server.test_pin_peer_entity(CALLER_NODE, caller_entity.clone());
let cap = CapabilityAuthorityId::for_tag("nrpc:svc");
let membership = OrgMembershipCert::try_issue(&org_b, caller_entity.clone(), 1, 3600)
.expect("caller cert");
let dispatcher = OrgDispatcherGrant::try_issue(
&org_b,
caller_entity.clone(),
DispatcherScope::Exact(cap),
3600,
)
.expect("dispatcher grant");
let intent = OrgProofIntent {
caller: std::sync::Arc::new(caller_kp),
membership,
dispatcher,
capability_grant: None,
acting_org: org_b.org_id(),
provider_owner_org: org_b.org_id(),
provider: node_entity.clone(),
capability: cap,
proof_ttl_secs: 30,
};
let frame = |call_id: u64, frame_origin: u64| {
let mut req = RpcRequestPayload {
service: "svc".to_string(),
deadline_ns: 0,
flags: 0,
headers: vec![],
body: Bytes::from_static(b"ping"),
};
req.headers
.push(sign_admission_proof(&intent, call_id, &req).expect("sign"));
let mut f = EventMeta::new(DISPATCH_RPC_REQUEST, 0, frame_origin, call_id, 0)
.to_bytes()
.to_vec();
encode_rpc_route(&mut f, 0);
f.extend_from_slice(&req.encode());
RpcInboundEvent {
channel_hash,
origin_hash: frame_origin,
from_node: CALLER_NODE,
payload: Bytes::from(f),
}
};
assert!(server.deliver_rpc_inbound_for_test(channel_hash, frame(1, victim_origin)));
assert!(server.deliver_rpc_inbound_for_test(channel_hash, frame(2, caller_origin)));
assert!(
wait_until_at_least(|| admits.load(Ordering::SeqCst) as u64, 1).await,
"the origin-matched call is admitted",
);
assert_eq!(
admits.load(Ordering::SeqCst),
1,
"the origin-mismatched call was denied — the handler stayed dark",
);
}
async fn build_server_with_token_gated_reply(reply: &ChannelName) -> Arc<MeshNode> {
use crate::adapter::net::channel::{ChannelConfig, ChannelConfigRegistry};
let addr: SocketAddr = "127.0.0.1:0".parse().unwrap();
let cfg = MeshNodeConfig::new(addr, [0x42u8; 32])
.with_heartbeat_interval(Duration::from_millis(200))
.with_session_timeout(Duration::from_secs(10))
.with_handshake(3, Duration::from_secs(2));
let mut node = MeshNode::new(EntityKeypair::generate(), cfg)
.await
.expect("MeshNode::new");
let registry = ChannelConfigRegistry::new();
let foreign_root = EntityKeypair::generate().entity_id().clone();
registry.insert(
ChannelConfig::new(ChannelId::new(reply.clone())).with_token_roots(vec![foreign_root]),
);
node.set_channel_configs(Arc::new(registry));
Arc::new(node)
}
#[tokio::test]
async fn try_publish_to_peer_reports_pre_send_no_session() {
let server = build_server().await;
let reply = ChannelName::new("svc.replies.0000000000000002").unwrap();
let cid = ChannelId::new(reply);
let reply_hash = cid.hash();
let reply_sid = MeshNode::publish_stream_id(&cid);
const NEVER_CONNECTED: u64 = 0xDEAD_BEEF_0000_0002;
assert!(!server.has_peer_session(NEVER_CONNECTED));
let outcome = server
.try_publish_to_peer(
NEVER_CONNECTED,
reply_hash,
reply_sid,
true,
std::slice::from_ref(&Bytes::from_static(b"resp")),
)
.await;
assert!(
matches!(outcome, PeerPublishOutcome::NoSession),
"a never-connected peer must report the pre-send NoSession outcome",
);
}
#[tokio::test]
async fn direct_only_frame_drops_while_normal_response_rosters_when_peer_gone() {
let reply = ChannelName::new("svc.replies.0000000000000003").unwrap();
let server = build_server_with_token_gated_reply(&reply).await;
let cid = ChannelId::new(reply.clone());
let reply_hash = cid.hash();
let reply_sid = MeshNode::publish_stream_id(&cid);
const GONE_NODE: u64 = 0xDEAD_BEEF_0000_0003;
assert!(!server.has_peer_session(GONE_NODE));
let direct = publish_response_to_caller(
&server,
0x3,
Some(GONE_NODE),
&reply,
reply_hash,
reply_sid,
Bytes::from_static(b"deny"),
ResponseRouteFallback::DirectOnly,
)
.await;
assert!(
direct.is_ok(),
"a direct-only frame to a gone peer must DROP (never roster-fallback): {direct:?}",
);
let roster = publish_response_to_caller(
&server,
0x3,
Some(GONE_NODE),
&reply,
reply_hash,
reply_sid,
Bytes::from_static(b"resp"),
ResponseRouteFallback::RosterOnStaleDirect,
)
.await;
assert!(
roster.is_err(),
"a normal response to a gone peer must REACH the (ACL-denied) roster, \
proving the divergence is the routing policy not an empty roster: {roster:?}",
);
}
#[tokio::test]
async fn stale_route_hint_falls_back_to_the_roster() {
let server = build_server().await;
let reply = ChannelName::new("svc.replies.0000000000000001").unwrap();
let cid = ChannelId::new(reply.clone());
let reply_hash = cid.hash();
let reply_sid = MeshNode::publish_stream_id(&cid);
const STALE_NODE: u64 = 0xDEAD_BEEF_0000_0001;
assert!(!server.has_peer_session(STALE_NODE));
let result = publish_response_to_caller(
&server,
0x1,
Some(STALE_NODE),
&reply,
reply_hash,
reply_sid,
Bytes::from_static(b"resp"),
ResponseRouteFallback::RosterOnStaleDirect,
)
.await;
assert!(
result.is_ok(),
"a stale route hint must fall back to the roster, not error out: {result:?}",
);
}
}