use crate::atom::{Atom, AtomTable};
use crate::distribution::control::{
self, ControlDelivery, ControlError, ControlMessage, ControlRegistry, PG_UPDATE, PgDelivery,
REG_SEND, SEND,
};
use crate::etf::decode::decode_term;
use crate::etf::encode::EncodeError;
use crate::native::ProcessContext;
use crate::process::{ExitReason, Process, RemotePid};
use crate::term::Term;
use crate::term::boxed::Tuple;
use crate::term::pid_ref::PidRef;
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
#[repr(i64)]
pub enum ControlOp {
Link = 1,
Send = 2,
Exit = 3,
Unlink = 4,
RegSend = 6,
Exit2 = 8,
MonitorP = 19,
DemonitorP = 20,
MonitorPExit = 21,
SpawnRequest = 29,
SpawnReply = 31,
}
impl ControlOp {
#[must_use]
pub const fn from_opcode(opcode: i64) -> Option<Self> {
match opcode {
1 => Some(Self::Link),
2 => Some(Self::Send),
3 => Some(Self::Exit),
4 => Some(Self::Unlink),
6 => Some(Self::RegSend),
8 => Some(Self::Exit2),
19 => Some(Self::MonitorP),
20 => Some(Self::DemonitorP),
21 => Some(Self::MonitorPExit),
29 => Some(Self::SpawnRequest),
31 => Some(Self::SpawnReply),
_ => None,
}
}
#[must_use]
pub const fn opcode(self) -> i64 {
self as i64
}
}
pub const BEAMR_MONITOR: i64 = 102;
const LINK: i64 = ControlOp::Link.opcode();
const EXIT: i64 = ControlOp::Exit.opcode();
const UNLINK: i64 = ControlOp::Unlink.opcode();
const EXIT2: i64 = ControlOp::Exit2.opcode();
pub fn encode_link_frame(
local_node: Atom,
from_pid: u64,
to: RemotePid,
atom_table: &AtomTable,
) -> Result<Vec<u8>, EncodeError> {
encode_control_frame(ControlOp::Link, local_node, from_pid, to, None, atom_table)
}
pub fn encode_unlink_frame(
local_node: Atom,
from_pid: u64,
to: RemotePid,
atom_table: &AtomTable,
) -> Result<Vec<u8>, EncodeError> {
encode_control_frame(
ControlOp::Unlink,
local_node,
from_pid,
to,
None,
atom_table,
)
}
pub fn encode_exit_frame(
op: ControlOp,
local_node: Atom,
from_pid: u64,
to: RemotePid,
reason: ExitReason,
atom_table: &AtomTable,
) -> Result<Vec<u8>, EncodeError> {
if !matches!(op, ControlOp::Exit | ControlOp::Exit2) {
return Err(EncodeError::UnsupportedTerm);
}
encode_control_frame(op, local_node, from_pid, to, Some(reason), atom_table)
}
fn encode_control_frame(
op: ControlOp,
local_node: Atom,
from_pid: u64,
to: RemotePid,
reason: Option<ExitReason>,
atom_table: &AtomTable,
) -> Result<Vec<u8>, EncodeError> {
let mut process = Process::new(0, 64);
let mut context = ProcessContext::new();
context.attach_process(&mut process, 0);
let from = context
.alloc_external_pid(local_node, from_pid, 0)
.map_err(|_| EncodeError::UnsupportedTerm)?;
let to = context
.alloc_external_pid(to.node, to.pid_number, to.serial)
.map_err(|_| EncodeError::UnsupportedTerm)?;
let control = match reason {
None => context.alloc_tuple(&[Term::small_int(op.opcode()), from, to]),
Some(reason) => {
context.alloc_tuple(&[Term::small_int(op.opcode()), from, to, reason.as_term()])
}
}
.map_err(|_| EncodeError::UnsupportedTerm)?;
control::encode_frame(control, Term::NIL, atom_table)
}
pub(crate) fn exit_reason_from_wire(atom: Atom) -> ExitReason {
match atom {
Atom::NORMAL => ExitReason::Normal,
Atom::KILL => ExitReason::Kill,
Atom::KILLED => ExitReason::Killed,
Atom::NOCONNECTION => ExitReason::NoConnection,
Atom::NOPROC => ExitReason::NoProc,
_ => ExitReason::Error,
}
}
pub fn decode_control_addressed(
control_etf: &[u8],
atom_table: &AtomTable,
local_node: Option<Atom>,
) -> Result<ControlMessage, ControlError> {
let mut process = Process::new(0, 64);
let mut context = ProcessContext::new();
context.attach_process(&mut process, 0);
let term = decode_term(control_etf, &mut context, atom_table)?;
let tuple = Tuple::new(term).ok_or(ControlError::InvalidControl)?;
match tuple.get(0).and_then(Term::as_small_int) {
Some(SEND) if tuple.arity() == 3 => {
let to = PidRef::new(tuple.get(2).ok_or(ControlError::InvalidControl)?)
.ok_or(ControlError::InvalidControl)?;
if let (Some(local), Some(node)) = (local_node, to.node())
&& node != local
&& !is_node_less_marker(atom_table, node)
{
return Err(ControlError::MisAddressed);
}
Ok(ControlMessage::Send {
to_pid: to.pid_number(),
})
}
Some(REG_SEND) if tuple.arity() == 4 => {
let to_name = tuple
.get(3)
.and_then(Term::as_atom)
.ok_or(ControlError::InvalidControl)?;
Ok(ControlMessage::RegSend { to_name })
}
Some(PG_UPDATE) if tuple.arity() == 5 => control::decode_pg_update(&tuple),
Some(LINK) if tuple.arity() == 3 => {
let (from, to_pid) = link_endpoints(&tuple, atom_table, local_node)?;
Ok(ControlMessage::Link { from, to_pid })
}
Some(UNLINK) if tuple.arity() == 3 => {
let (from, to_pid) = link_endpoints(&tuple, atom_table, local_node)?;
Ok(ControlMessage::Unlink { from, to_pid })
}
Some(EXIT) if tuple.arity() == 4 => {
let (from, to_pid) = link_endpoints(&tuple, atom_table, local_node)?;
let reason = wire_reason(&tuple)?;
Ok(ControlMessage::Exit {
from,
to_pid,
reason,
})
}
Some(EXIT2) if tuple.arity() == 4 => {
let (from, to_pid) = link_endpoints(&tuple, atom_table, local_node)?;
let reason = wire_reason(&tuple)?;
Ok(ControlMessage::Exit2 {
from,
to_pid,
reason,
})
}
_ => Err(ControlError::InvalidControl),
}
}
fn link_endpoints(
tuple: &Tuple,
atom_table: &AtomTable,
local_node: Option<Atom>,
) -> Result<(RemotePid, u64), ControlError> {
let from = PidRef::new(tuple.get(1).ok_or(ControlError::InvalidControl)?)
.ok_or(ControlError::InvalidControl)?
.remote_pid()
.ok_or(ControlError::InvalidControl)?;
if is_node_less_marker(atom_table, from.node) {
return Err(ControlError::InvalidControl);
}
let to = PidRef::new(tuple.get(2).ok_or(ControlError::InvalidControl)?)
.ok_or(ControlError::InvalidControl)?;
let to_node = to.node().ok_or(ControlError::InvalidControl)?;
if is_node_less_marker(atom_table, to_node) {
return Err(ControlError::InvalidControl);
}
if let Some(local) = local_node
&& to_node != local
{
return Err(ControlError::MisAddressed);
}
Ok((from, to.pid_number()))
}
fn is_node_less_marker(atom_table: &AtomTable, node: Atom) -> bool {
atom_table.resolve(node) == Some(crate::etf::encode::NONODE_NOHOST)
}
fn wire_reason(tuple: &Tuple) -> Result<ExitReason, ControlError> {
let atom = tuple
.get(3)
.and_then(Term::as_atom)
.ok_or(ControlError::InvalidControl)?;
Ok(exit_reason_from_wire(atom))
}
pub trait LinkControlDelivery: Send + Sync {
fn apply_link(&self, from: RemotePid, to_pid: u64);
fn apply_unlink(&self, from: RemotePid, to_pid: u64);
fn apply_link_exit(&self, from: RemotePid, to_pid: u64, reason: ExitReason);
fn apply_exit2(&self, from: RemotePid, to_pid: u64, reason: ExitReason);
}
pub struct ControlSinks<'a> {
pub delivery: &'a dyn ControlDelivery,
pub registry: Option<&'a dyn ControlRegistry>,
pub pg: Option<&'a dyn PgDelivery>,
pub links: Option<&'a dyn LinkControlDelivery>,
pub origin_node: Option<Atom>,
pub local_node: Option<Atom>,
}
pub fn dispatch_frame(
control_etf: &[u8],
payload_etf: &[u8],
atom_table: &AtomTable,
sinks: &ControlSinks<'_>,
) -> Result<bool, ControlError> {
match decode_control_addressed(control_etf, atom_table, sinks.local_node)? {
ControlMessage::Send { to_pid } => Ok(sinks.delivery.deliver_payload(to_pid, payload_etf)),
ControlMessage::RegSend { to_name } => {
let Some(registry) = sinks.registry else {
return Ok(false);
};
let Some(pid) = registry.whereis(to_name) else {
return Ok(false);
};
Ok(sinks.delivery.deliver_payload(pid, payload_etf))
}
ControlMessage::PgJoin {
scope,
group,
node,
pid_number,
serial,
} => {
let Some(pg) = sinks.pg else {
return Ok(false);
};
pg.apply_pg_join(scope, group, node, pid_number, serial);
Ok(true)
}
ControlMessage::PgLeave {
scope,
group,
node,
pid_number,
serial,
} => {
let Some(pg) = sinks.pg else {
return Ok(false);
};
pg.apply_pg_leave(scope, group, node, pid_number, serial);
Ok(true)
}
ControlMessage::Link { from, to_pid } => {
let Some(links) = link_sink(sinks, from) else {
return Ok(false);
};
links.apply_link(from, to_pid);
Ok(true)
}
ControlMessage::Unlink { from, to_pid } => {
let Some(links) = link_sink(sinks, from) else {
return Ok(false);
};
links.apply_unlink(from, to_pid);
Ok(true)
}
ControlMessage::Exit {
from,
to_pid,
reason,
} => {
let Some(links) = link_sink(sinks, from) else {
return Ok(false);
};
links.apply_link_exit(from, to_pid, reason);
Ok(true)
}
ControlMessage::Exit2 {
from,
to_pid,
reason,
} => {
let Some(links) = link_sink(sinks, from) else {
return Ok(false);
};
links.apply_exit2(from, to_pid, reason);
Ok(true)
}
}
}
fn link_sink<'a>(sinks: &ControlSinks<'a>, from: RemotePid) -> Option<&'a dyn LinkControlDelivery> {
if let Some(origin) = sinks.origin_node
&& from.node != origin
{
return None;
}
sinks.links
}
#[cfg(test)]
mod tests {
use std::sync::Mutex;
use super::*;
use crate::distribution::control::{SPAWN_REPLY, encode_send_frame, split_frame};
use crate::etf::encode::encode_term;
use crate::term::boxed::write_external_pid;
fn remote(node: Atom, pid_number: u64, serial: u64) -> RemotePid {
RemotePid {
node,
pid_number,
serial,
}
}
fn hostile_frame(
atom_table: &AtomTable,
build: impl FnOnce(&mut ProcessContext) -> Term,
) -> Vec<u8> {
let mut process = Process::new(0, 256);
let mut context = ProcessContext::new();
context.attach_process(&mut process, 0);
let control = build(&mut context);
control::encode_frame(control, Term::NIL, atom_table).expect("frame encodes")
}
fn ext_pid(context: &mut ProcessContext, node: Atom, pid_number: u64, serial: u64) -> Term {
context
.alloc_external_pid(node, pid_number, serial)
.expect("external pid allocates")
}
#[test]
fn beamr_private_opcodes_are_outside_otp_control_range() {
const _: () = assert!(PG_UPDATE > SPAWN_REPLY && BEAMR_MONITOR > SPAWN_REPLY);
assert!(ControlOp::from_opcode(BEAMR_MONITOR).is_none());
assert!(ControlOp::from_opcode(PG_UPDATE).is_none());
}
#[test]
fn control_op_mapping_is_exact() {
assert_eq!(ControlOp::from_opcode(1), Some(ControlOp::Link));
assert_eq!(ControlOp::from_opcode(2), Some(ControlOp::Send));
assert_eq!(ControlOp::from_opcode(3), Some(ControlOp::Exit));
assert_eq!(ControlOp::from_opcode(4), Some(ControlOp::Unlink));
assert_eq!(ControlOp::from_opcode(6), Some(ControlOp::RegSend));
assert_eq!(ControlOp::from_opcode(8), Some(ControlOp::Exit2));
assert_eq!(ControlOp::from_opcode(19), Some(ControlOp::MonitorP));
assert_eq!(ControlOp::from_opcode(20), Some(ControlOp::DemonitorP));
assert_eq!(ControlOp::from_opcode(21), Some(ControlOp::MonitorPExit));
assert_eq!(ControlOp::from_opcode(29), Some(ControlOp::SpawnRequest));
assert_eq!(ControlOp::from_opcode(31), Some(ControlOp::SpawnReply));
assert_eq!(ControlOp::from_opcode(255), None);
}
#[test]
fn control_op_round_trips_all_tabled_opcodes() {
for op in [
ControlOp::Link,
ControlOp::Send,
ControlOp::Exit,
ControlOp::Unlink,
ControlOp::RegSend,
ControlOp::Exit2,
ControlOp::MonitorP,
ControlOp::DemonitorP,
ControlOp::MonitorPExit,
ControlOp::SpawnRequest,
ControlOp::SpawnReply,
] {
assert_eq!(ControlOp::from_opcode(op.opcode()), Some(op));
}
assert_eq!(ControlOp::from_opcode(255), None);
}
#[test]
fn link_frame_encodes_the_pinned_exact_bytes() {
let atom_table = AtomTable::with_common_atoms();
let node_a = atom_table.intern("a@host");
let node_b = atom_table.intern("b@host");
let frame = encode_link_frame(node_a, 42, remote(node_b, 7, 3), &atom_table)
.expect("link frame encodes");
#[rustfmt::skip]
let control: &[u8] = &[
0x83, 0x68, 0x03, 0x61, 0x01, 0x58, 0x77, 0x06, b'a', b'@', b'h', b'o', b's', b't',
0x00, 0x00, 0x00, 0x2A, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x58, 0x77, 0x06, b'b', b'@', b'h', b'o', b's', b't',
0x00, 0x00, 0x00, 0x07, 0x00, 0x00, 0x00, 0x03, 0x00, 0x00, 0x00, 0x00, ];
let payload: &[u8] = &[0x83, 0x6A]; let mut expected = Vec::new();
expected.extend_from_slice(&(control.len() as u32).to_be_bytes());
expected.extend_from_slice(&(payload.len() as u32).to_be_bytes());
expected.extend_from_slice(control);
expected.extend_from_slice(payload);
assert_eq!(control.len(), 47);
assert_eq!(frame, expected);
}
#[test]
fn link_frame_round_trips_through_split_and_addressed_decode() {
let atom_table = AtomTable::with_common_atoms();
let node_a = atom_table.intern("a@host");
let node_b = atom_table.intern("b@host");
let frame = encode_link_frame(node_a, 42, remote(node_b, 7, 3), &atom_table)
.expect("link frame encodes");
let (control, payload) = split_frame(&frame).expect("frame splits");
assert_eq!(
payload,
encode_term(Term::NIL, &atom_table)
.expect("nil encodes")
.as_slice()
);
let decoded =
decode_control_addressed(control, &atom_table, Some(node_b)).expect("control decodes");
assert_eq!(
decoded,
ControlMessage::Link {
from: remote(node_a, 42, 0),
to_pid: 7,
}
);
}
#[test]
fn unlink_frame_round_trips_through_split_and_addressed_decode() {
let atom_table = AtomTable::with_common_atoms();
let node_a = atom_table.intern("a@host");
let node_b = atom_table.intern("b@host");
let frame = encode_unlink_frame(node_a, 5, remote(node_b, 9, 1), &atom_table)
.expect("unlink frame encodes");
let (control, _payload) = split_frame(&frame).expect("frame splits");
let decoded =
decode_control_addressed(control, &atom_table, Some(node_b)).expect("control decodes");
assert_eq!(
decoded,
ControlMessage::Unlink {
from: remote(node_a, 5, 0),
to_pid: 9,
}
);
}
#[test]
fn exit_frame_round_trips_carrying_the_terminal_reason() {
let atom_table = AtomTable::with_common_atoms();
let node_a = atom_table.intern("a@host");
let node_b = atom_table.intern("b@host");
let frame = encode_exit_frame(
ControlOp::Exit,
node_a,
8,
remote(node_b, 11, 0),
ExitReason::Killed,
&atom_table,
)
.expect("exit frame encodes");
let (control, _payload) = split_frame(&frame).expect("frame splits");
let decoded =
decode_control_addressed(control, &atom_table, Some(node_b)).expect("control decodes");
assert_eq!(
decoded,
ControlMessage::Exit {
from: remote(node_a, 8, 0),
to_pid: 11,
reason: ExitReason::Killed,
}
);
}
#[test]
fn exit2_frame_round_trips_carrying_a_raw_kill() {
let atom_table = AtomTable::with_common_atoms();
let node_a = atom_table.intern("a@host");
let node_b = atom_table.intern("b@host");
let frame = encode_exit_frame(
ControlOp::Exit2,
node_a,
3,
remote(node_b, 4, 2),
ExitReason::Kill,
&atom_table,
)
.expect("exit2 frame encodes");
let (control, _payload) = split_frame(&frame).expect("frame splits");
let decoded =
decode_control_addressed(control, &atom_table, Some(node_b)).expect("control decodes");
assert_eq!(
decoded,
ControlMessage::Exit2 {
from: remote(node_a, 3, 0),
to_pid: 4,
reason: ExitReason::Kill,
}
);
}
#[test]
fn exit_encoder_rejects_non_exit_ops() {
let atom_table = AtomTable::with_common_atoms();
let node_a = atom_table.intern("a@host");
let node_b = atom_table.intern("b@host");
for op in [ControlOp::Link, ControlOp::Unlink, ControlOp::Send] {
assert_eq!(
encode_exit_frame(
op,
node_a,
1,
remote(node_b, 2, 0),
ExitReason::Error,
&atom_table,
),
Err(EncodeError::UnsupportedTerm)
);
}
}
#[test]
fn exit_reason_from_wire_maps_known_atoms_and_coerces_unknown_to_error() {
let atom_table = AtomTable::with_common_atoms();
assert_eq!(exit_reason_from_wire(Atom::NORMAL), ExitReason::Normal);
assert_eq!(exit_reason_from_wire(Atom::KILL), ExitReason::Kill);
assert_eq!(exit_reason_from_wire(Atom::KILLED), ExitReason::Killed);
assert_eq!(exit_reason_from_wire(Atom::ERROR), ExitReason::Error);
assert_eq!(
exit_reason_from_wire(Atom::NOCONNECTION),
ExitReason::NoConnection
);
assert_eq!(exit_reason_from_wire(Atom::NOPROC), ExitReason::NoProc);
let weird = atom_table.intern("weird_reason");
assert_eq!(exit_reason_from_wire(weird), ExitReason::Error);
}
#[test]
fn exit_decode_coerces_an_unknown_reason_atom_to_error() {
let atom_table = AtomTable::with_common_atoms();
let node_a = atom_table.intern("a@host");
let node_b = atom_table.intern("b@host");
let weird = atom_table.intern("weird_reason");
let frame = hostile_frame(&atom_table, |context| {
let from = ext_pid(context, node_a, 1, 0);
let to = ext_pid(context, node_b, 2, 0);
context
.alloc_tuple(&[Term::small_int(EXIT), from, to, Term::atom(weird)])
.expect("tuple allocates")
});
let (control, _payload) = split_frame(&frame).expect("frame splits");
assert_eq!(
decode_control_addressed(control, &atom_table, Some(node_b)),
Ok(ControlMessage::Exit {
from: remote(node_a, 1, 0),
to_pid: 2,
reason: ExitReason::Error,
})
);
}
#[test]
fn unknown_opcode_is_invalid_control() {
let atom_table = AtomTable::with_common_atoms();
let node_b = atom_table.intern("b@host");
let frame = hostile_frame(&atom_table, |context| {
let from = ext_pid(context, node_b, 1, 0);
let to = ext_pid(context, node_b, 2, 0);
context
.alloc_tuple(&[Term::small_int(77), from, to])
.expect("tuple allocates")
});
let (control, _payload) = split_frame(&frame).expect("frame splits");
assert_eq!(
decode_control_addressed(control, &atom_table, Some(node_b)),
Err(ControlError::InvalidControl)
);
}
#[test]
fn wrong_arities_are_invalid_control() {
let atom_table = AtomTable::with_common_atoms();
let node_a = atom_table.intern("a@host");
let node_b = atom_table.intern("b@host");
let link_arity_4 = hostile_frame(&atom_table, |context| {
let from = ext_pid(context, node_a, 1, 0);
let to = ext_pid(context, node_b, 2, 0);
context
.alloc_tuple(&[Term::small_int(LINK), from, to, Term::atom(Atom::NORMAL)])
.expect("tuple allocates")
});
let exit_arity_3 = hostile_frame(&atom_table, |context| {
let from = ext_pid(context, node_a, 1, 0);
let to = ext_pid(context, node_b, 2, 0);
context
.alloc_tuple(&[Term::small_int(EXIT), from, to])
.expect("tuple allocates")
});
for frame in [link_arity_4, exit_arity_3] {
let (control, _payload) = split_frame(&frame).expect("frame splits");
assert_eq!(
decode_control_addressed(control, &atom_table, Some(node_b)),
Err(ControlError::InvalidControl)
);
}
}
#[test]
fn node_less_pids_on_link_controls_are_invalid_control() {
let atom_table = AtomTable::with_common_atoms();
let node_a = atom_table.intern("a@host");
let node_b = atom_table.intern("b@host");
let local_from = hostile_frame(&atom_table, |context| {
let to = ext_pid(context, node_b, 2, 0);
context
.alloc_tuple(&[Term::small_int(LINK), Term::pid(1), to])
.expect("tuple allocates")
});
let local_to = hostile_frame(&atom_table, |context| {
let from = ext_pid(context, node_a, 1, 0);
context
.alloc_tuple(&[Term::small_int(UNLINK), from, Term::pid(2)])
.expect("tuple allocates")
});
for frame in [local_from, local_to] {
let (control, _payload) = split_frame(&frame).expect("frame splits");
assert_eq!(
decode_control_addressed(control, &atom_table, Some(node_b)),
Err(ControlError::InvalidControl),
);
assert_eq!(
decode_control_addressed(control, &atom_table, None),
Err(ControlError::InvalidControl),
);
}
}
#[test]
fn in_memory_node_less_pids_are_rejected_by_the_endpoint_extractor() {
let atom_table = AtomTable::with_common_atoms();
let node_b = atom_table.intern("b@host");
let mut process = Process::new(0, 64);
let mut context = ProcessContext::new();
context.attach_process(&mut process, 0);
let ext = ext_pid(&mut context, node_b, 2, 0);
let from_local = context
.alloc_tuple(&[Term::small_int(LINK), Term::pid(1), ext])
.expect("tuple allocates");
let to_local = context
.alloc_tuple(&[Term::small_int(LINK), ext, Term::pid(1)])
.expect("tuple allocates");
for term in [from_local, to_local] {
let tuple = Tuple::new(term).expect("tuple view");
assert_eq!(
link_endpoints(&tuple, &atom_table, Some(node_b)),
Err(ControlError::InvalidControl)
);
}
}
#[test]
fn non_atom_exit_reason_is_shape_corruption() {
let atom_table = AtomTable::with_common_atoms();
let node_a = atom_table.intern("a@host");
let node_b = atom_table.intern("b@host");
for op in [EXIT, EXIT2] {
let frame = hostile_frame(&atom_table, |context| {
let from = ext_pid(context, node_a, 1, 0);
let to = ext_pid(context, node_b, 2, 0);
let reason = context
.alloc_tuple(&[Term::atom(Atom::ERROR), Term::small_int(1)])
.expect("reason tuple allocates");
context
.alloc_tuple(&[Term::small_int(op), from, to, reason])
.expect("tuple allocates")
});
let (control, _payload) = split_frame(&frame).expect("frame splits");
assert_eq!(
decode_control_addressed(control, &atom_table, Some(node_b)),
Err(ControlError::InvalidControl)
);
}
}
#[test]
fn link_control_addressed_to_a_third_node_is_misaddressed() {
let atom_table = AtomTable::with_common_atoms();
let node_a = atom_table.intern("a@host");
let node_b = atom_table.intern("b@host");
let node_c = atom_table.intern("c@host");
let link = encode_link_frame(node_a, 1, remote(node_c, 2, 0), &atom_table)
.expect("link frame encodes");
let exit = encode_exit_frame(
ControlOp::Exit,
node_a,
1,
remote(node_c, 2, 0),
ExitReason::Error,
&atom_table,
)
.expect("exit frame encodes");
for frame in [link, exit] {
let (control, _payload) = split_frame(&frame).expect("frame splits");
assert_eq!(
decode_control_addressed(control, &atom_table, Some(node_b)),
Err(ControlError::MisAddressed)
);
}
}
#[test]
fn send_to_an_external_pid_naming_another_node_is_misaddressed() {
let atom_table = AtomTable::with_common_atoms();
let node_b = atom_table.intern("b@host");
let node_c = atom_table.intern("c@host");
let mut heap = [0_u64; 4];
let to = write_external_pid(&mut heap, node_c, 7, 0).expect("external pid fits");
let frame = encode_send_frame(Term::atom(Atom::OK), to, Term::atom(Atom::OK), &atom_table)
.expect("send frame encodes");
let (control, _payload) = split_frame(&frame).expect("frame splits");
assert_eq!(
decode_control_addressed(control, &atom_table, Some(node_b)),
Err(ControlError::MisAddressed)
);
assert_eq!(
decode_control_addressed(control, &atom_table, None),
Ok(ControlMessage::Send { to_pid: 7 })
);
}
#[test]
fn send_to_a_node_less_pid_still_delivers_for_legacy_tolerance() {
let atom_table = AtomTable::with_common_atoms();
let node_b = atom_table.intern("b@host");
let frame = encode_send_frame(
Term::atom(Atom::OK),
Term::pid(7),
Term::atom(Atom::OK),
&atom_table,
)
.expect("send frame encodes");
let (control, _payload) = split_frame(&frame).expect("frame splits");
assert_eq!(
decode_control_addressed(control, &atom_table, Some(node_b)),
Ok(ControlMessage::Send { to_pid: 7 })
);
}
#[test]
fn worst_case_link_control_fits_the_64_word_decode_heap() {
const TUPLE_WORDS: usize = 5;
const EXTERNAL_PID_WORDS: usize = 4;
const _: () = assert!(TUPLE_WORDS + 2 * EXTERNAL_PID_WORDS <= 64);
let atom_table = AtomTable::with_common_atoms();
let node_a = atom_table.intern("a-rather-long-node-name@some-long-host.example.com");
let node_b = atom_table.intern("b-rather-long-node-name@some-long-host.example.com");
let frame = encode_exit_frame(
ControlOp::Exit,
node_a,
u64::from(u32::MAX),
remote(node_b, u64::from(u32::MAX), u64::from(u32::MAX)),
ExitReason::NoConnection,
&atom_table,
)
.expect("exit frame encodes");
let (control, _payload) = split_frame(&frame).expect("frame splits");
assert!(decode_control_addressed(control, &atom_table, Some(node_b)).is_ok());
}
struct NoopDelivery;
impl ControlDelivery for NoopDelivery {
fn deliver_payload(&self, _target_pid: u64, _payload_etf: &[u8]) -> bool {
true
}
}
type LinkRecord = (ControlOp, RemotePid, u64, Option<ExitReason>);
#[derive(Default)]
struct RecordingLinks {
events: Mutex<Vec<LinkRecord>>,
}
impl RecordingLinks {
fn events(&self) -> Vec<LinkRecord> {
self.events
.lock()
.unwrap_or_else(|error| error.into_inner())
.clone()
}
fn record(&self, record: LinkRecord) {
self.events
.lock()
.unwrap_or_else(|error| error.into_inner())
.push(record);
}
}
impl LinkControlDelivery for RecordingLinks {
fn apply_link(&self, from: RemotePid, to_pid: u64) {
self.record((ControlOp::Link, from, to_pid, None));
}
fn apply_unlink(&self, from: RemotePid, to_pid: u64) {
self.record((ControlOp::Unlink, from, to_pid, None));
}
fn apply_link_exit(&self, from: RemotePid, to_pid: u64, reason: ExitReason) {
self.record((ControlOp::Exit, from, to_pid, Some(reason)));
}
fn apply_exit2(&self, from: RemotePid, to_pid: u64, reason: ExitReason) {
self.record((ControlOp::Exit2, from, to_pid, Some(reason)));
}
}
#[test]
fn dispatch_routes_all_four_link_controls_to_the_links_sink() {
let atom_table = AtomTable::with_common_atoms();
let node_a = atom_table.intern("a@host");
let node_b = atom_table.intern("b@host");
let delivery = NoopDelivery;
let links = RecordingLinks::default();
let sinks = ControlSinks {
delivery: &delivery,
registry: None,
pg: None,
links: Some(&links),
origin_node: Some(node_a),
local_node: Some(node_b),
};
let frames = [
encode_link_frame(node_a, 1, remote(node_b, 2, 0), &atom_table)
.expect("link frame encodes"),
encode_unlink_frame(node_a, 1, remote(node_b, 2, 0), &atom_table)
.expect("unlink frame encodes"),
encode_exit_frame(
ControlOp::Exit,
node_a,
1,
remote(node_b, 2, 0),
ExitReason::Error,
&atom_table,
)
.expect("exit frame encodes"),
encode_exit_frame(
ControlOp::Exit2,
node_a,
1,
remote(node_b, 2, 0),
ExitReason::Kill,
&atom_table,
)
.expect("exit2 frame encodes"),
];
for frame in &frames {
let (control, payload) = split_frame(frame).expect("frame splits");
assert_eq!(
dispatch_frame(control, payload, &atom_table, &sinks),
Ok(true)
);
}
let from = remote(node_a, 1, 0);
assert_eq!(
links.events(),
vec![
(ControlOp::Link, from, 2, None),
(ControlOp::Unlink, from, 2, None),
(ControlOp::Exit, from, 2, Some(ExitReason::Error)),
(ControlOp::Exit2, from, 2, Some(ExitReason::Kill)),
]
);
}
#[test]
fn dispatch_drops_link_controls_whose_from_forges_another_origin() {
let atom_table = AtomTable::with_common_atoms();
let node_a = atom_table.intern("a@host");
let node_b = atom_table.intern("b@host");
let node_c = atom_table.intern("c@host");
let delivery = NoopDelivery;
let links = RecordingLinks::default();
let sinks = ControlSinks {
delivery: &delivery,
registry: None,
pg: None,
links: Some(&links),
origin_node: Some(node_c),
local_node: Some(node_b),
};
let frame = encode_link_frame(node_a, 1, remote(node_b, 2, 0), &atom_table)
.expect("link frame encodes");
let (control, payload) = split_frame(&frame).expect("frame splits");
assert_eq!(
dispatch_frame(control, payload, &atom_table, &sinks),
Ok(false)
);
assert!(links.events().is_empty(), "forged frame must not apply");
}
#[test]
fn dispatch_without_a_links_sink_drops_link_controls() {
let atom_table = AtomTable::with_common_atoms();
let node_a = atom_table.intern("a@host");
let node_b = atom_table.intern("b@host");
let delivery = NoopDelivery;
let sinks = ControlSinks {
delivery: &delivery,
registry: None,
pg: None,
links: None,
origin_node: None,
local_node: None,
};
let frame = encode_exit_frame(
ControlOp::Exit,
node_a,
1,
remote(node_b, 2, 0),
ExitReason::Error,
&atom_table,
)
.expect("exit frame encodes");
let (control, payload) = split_frame(&frame).expect("frame splits");
assert_eq!(
dispatch_frame(control, payload, &atom_table, &sinks),
Ok(false)
);
}
#[test]
fn legacy_handle_frame_now_drops_link_controls_instead_of_erroring() {
let atom_table = AtomTable::with_common_atoms();
let node_a = atom_table.intern("a@host");
let node_b = atom_table.intern("b@host");
let delivery = NoopDelivery;
let frame = encode_link_frame(node_a, 1, remote(node_b, 2, 0), &atom_table)
.expect("link frame encodes");
let (control, payload) = split_frame(&frame).expect("frame splits");
assert_eq!(
control::handle_frame(control, payload, &atom_table, &delivery, None, None),
Ok(false)
);
}
}