pamoja_ffi/

ladder.rs

//! The C ABI for the cost-aware transport ladder.
//!
//! These functions wrap [`pamoja_ladder`] for callers that reach the SDK through
//! the flat C boundary. A ladder is the answer to a node that has more than one
//! way to reach the network and no single one that always works: rungs are tried
//! in the order they were added, cheapest first, and a message no rung accepts
//! goes into a buffer rather than being lost.
//!
//! The Rust ladder is generic over its buffer, which cannot cross a C ABI, so
//! this one is built over the store handle from [`crate::sync`]. That handle
//! already covers both an in-memory and a file-backed queue, so nothing is given
//! up: a caller still chooses whether the buffer survives a restart.

use std::ffi::c_char;
use std::ptr;

use pamoja_ladder::{Delivery, TransportLadder};

use crate::sync::{take_store, PamojaStore, StoreKind};
use crate::transport::{receive_within, take_transport, PamojaMessage, PamojaTransport};
use crate::{read_bytes, read_str, runtime, set_last_error, PamojaStatus};

/// What became of a message handed to a ladder.
#[repr(C)]
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum PamojaDelivery {
    /// A rung took the message and it is on its way.
    Sent = 0,
    /// No rung would take it, so it is in the buffer awaiting a flush.
    Buffered = 1,
}

/// An opaque handle to a ladder and the buffer behind it.
///
/// The Rust builder takes the ladder by value to add a rung, which would move
/// the handle out from under a caller holding a pointer to it. Holding the
/// ladder in an option lets it be taken and put back so the handle address stays
/// good for the life of the ladder.
pub struct PamojaLadder {
    inner: Option<TransportLadder<StoreKind>>,
}

/// Creates a ladder with no rungs, buffering into a store.
///
/// # Arguments
///
/// * `store` - the buffer to hold messages no rung would take, consumed by this
///   call.
///
/// # Returns
///
/// A handle the caller must release with [`pamoja_ladder_free`], or null if
/// `store` is null.
///
/// # Safety
///
/// `store` must be a live handle from [`crate::sync`] that has not been freed or
/// consumed. After this call it must not be used again, whatever the result.
#[no_mangle]
pub unsafe extern "C" fn pamoja_ladder_new(store: *mut PamojaStore) -> *mut PamojaLadder {
    let Some(store) = take_store(store) else {
        return ptr::null_mut();
    };
    Box::into_raw(Box::new(PamojaLadder {
        inner: Some(TransportLadder::new(store)),
    }))
}

/// Adds a rung, which is tried after the rungs already added.
///
/// Add the cheapest, most-preferred link first and the costliest fallback last,
/// because a send takes the first rung that accepts it. A transport that delivers
/// is subscribed and listened on; one that only sends, a host transport without a
/// `recv` callback, is an uplink the ladder never listens on.
///
/// # Arguments
///
/// * `ladder` - the ladder to add to.
/// * `transport` - the transport to add, consumed by this call.
///
/// # Returns
///
/// [`PamojaStatus::Ok`] once the rung is added.
///
/// # Safety
///
/// `ladder` must be a live handle, and `transport` a live transport handle that
/// has not been freed or consumed. After this call the transport must not be
/// used again, whatever the result.
#[no_mangle]
pub unsafe extern "C" fn pamoja_ladder_rung(
    ladder: *mut PamojaLadder,
    transport: *mut PamojaTransport,
) -> PamojaStatus {
    add(ladder, transport, false)
}

/// Adds a rung that only sends, tried after the rungs already added.
///
/// The ladder sends over it in its turn but never subscribes it or listens on it,
/// whatever the transport could do. This is the shape of a satellite messenger, a
/// LoRa uplink, or any link a node reports over but takes no commands from.
///
/// # Arguments
///
/// * `ladder` - the ladder to add to.
/// * `transport` - the transport to add, consumed by this call.
///
/// # Returns
///
/// [`PamojaStatus::Ok`] once the rung is added.
///
/// # Safety
///
/// `ladder` must be a live handle, and `transport` a live transport handle that
/// has not been freed or consumed. After this call the transport must not be
/// used again, whatever the result.
#[no_mangle]
pub unsafe extern "C" fn pamoja_ladder_uplink(
    ladder: *mut PamojaLadder,
    transport: *mut PamojaTransport,
) -> PamojaStatus {
    add(ladder, transport, true)
}

/// Adds a transport to a ladder, as an uplink when asked or when it cannot
/// deliver, and as a rung that is listened on otherwise.
///
/// # Safety
///
/// As for [`pamoja_ladder_rung`].
unsafe fn add(
    ladder: *mut PamojaLadder,
    transport: *mut PamojaTransport,
    uplink: bool,
) -> PamojaStatus {
    let Some(handle) = ladder_handle(ladder) else {
        // The transport was promised to this call, so it is released rather than
        // left for a caller who has already been told not to touch it again.
        drop(take_transport(transport));
        return PamojaStatus::InvalidArgument;
    };
    let Some(transport) = take_transport(transport) else {
        return PamojaStatus::InvalidArgument;
    };
    let Some(inner) = handle.inner.take() else {
        set_last_error("this ladder is no longer usable".to_owned());
        return PamojaStatus::InvalidArgument;
    };
    handle.inner = Some(if uplink || !transport.listens() {
        inner.uplink(transport)
    } else {
        inner.rung(transport)
    });
    PamojaStatus::Ok
}

/// Connects every rung, so a send can be tried against each in turn.
///
/// A rung that will not connect is left in the ladder: it may come back, and a
/// send simply falls through it until it does.
///
/// # Arguments
///
/// * `ladder` - the ladder.
///
/// # Returns
///
/// [`PamojaStatus::Ok`] once the rungs have been tried.
///
/// # Safety
///
/// `ladder` must be a live handle from [`pamoja_ladder_new`].
#[no_mangle]
pub unsafe extern "C" fn pamoja_ladder_connect(ladder: *mut PamojaLadder) -> PamojaStatus {
    let Some(inner) = ladder_inner(ladder) else {
        return PamojaStatus::InvalidArgument;
    };
    match runtime().block_on(inner.connect()) {
        Ok(()) => PamojaStatus::Ok,
        Err(error) => fail(error),
    }
}

/// Sends a payload, falling through the rungs and buffering if none take it.
///
/// # Arguments
///
/// * `ladder` - the ladder.
/// * `topic` - the destination topic, as null-terminated UTF-8.
/// * `payload` - the bytes to send.
/// * `payload_len` - the length of `payload`.
/// * `out_delivery` - receives whether the message went out or was buffered.
///
/// # Returns
///
/// [`PamojaStatus::Ok`] on success, with `out_delivery` saying which happened.
/// Buffering is a success, not a failure: it is what the ladder exists to do.
///
/// # Safety
///
/// `ladder` must be a live handle, `topic` a valid null-terminated UTF-8 string,
/// `payload` must point to at least `payload_len` readable bytes or be null when
/// that length is 0, and `out_delivery` must be writable or null.
#[no_mangle]
pub unsafe extern "C" fn pamoja_ladder_send(
    ladder: *mut PamojaLadder,
    topic: *const c_char,
    payload: *const u8,
    payload_len: usize,
    out_delivery: *mut PamojaDelivery,
) -> PamojaStatus {
    let Some(topic) = read_str(topic, "topic") else {
        return PamojaStatus::InvalidArgument;
    };
    let payload = match read_bytes(payload, payload_len) {
        Ok(payload) => payload,
        Err(status) => return status,
    };
    let Some(inner) = ladder_inner(ladder) else {
        return PamojaStatus::InvalidArgument;
    };
    match runtime().block_on(inner.send(topic, &payload)) {
        Ok(delivery) => {
            if !out_delivery.is_null() {
                *out_delivery = match delivery {
                    Delivery::Sent => PamojaDelivery::Sent,
                    Delivery::Buffered => PamojaDelivery::Buffered,
                };
            }
            PamojaStatus::Ok
        }
        Err(error) => fail(error),
    }
}

/// Replays the buffer over the rungs, oldest message first.
///
/// # Arguments
///
/// * `ladder` - the ladder.
/// * `out_sent` - receives how many buffered messages went out, or may be null.
///
/// # Returns
///
/// [`PamojaStatus::Ok`] on success.
///
/// # Safety
///
/// `ladder` must be a live handle and `out_sent` writable or null.
#[no_mangle]
pub unsafe extern "C" fn pamoja_ladder_flush(
    ladder: *mut PamojaLadder,
    out_sent: *mut usize,
) -> PamojaStatus {
    let Some(inner) = ladder_inner(ladder) else {
        return PamojaStatus::InvalidArgument;
    };
    match runtime().block_on(inner.flush()) {
        Ok(sent) => {
            if !out_sent.is_null() {
                *out_sent = sent;
            }
            PamojaStatus::Ok
        }
        Err(error) => fail(error),
    }
}

/// Reports how many messages are waiting in the buffer.
///
/// # Arguments
///
/// * `ladder` - the ladder.
/// * `out_count` - receives the count.
///
/// # Returns
///
/// [`PamojaStatus::Ok`] on success.
///
/// # Safety
///
/// `ladder` must be a live handle and `out_count` must be writable.
#[no_mangle]
pub unsafe extern "C" fn pamoja_ladder_buffered(
    ladder: *mut PamojaLadder,
    out_count: *mut usize,
) -> PamojaStatus {
    let Some(inner) = ladder_inner(ladder) else {
        return PamojaStatus::InvalidArgument;
    };
    if out_count.is_null() {
        set_last_error("out_count must not be null".to_owned());
        return PamojaStatus::InvalidArgument;
    }
    match runtime().block_on(inner.buffered()) {
        Ok(count) => {
            *out_count = count;
            PamojaStatus::Ok
        }
        Err(error) => fail(error),
    }
}

/// Subscribes every rung that listens to a topic.
///
/// The filter is kept for the life of the ladder, so a rung that is down when it
/// is placed receives it when it next connects. Subscribing while no rung is up
/// succeeds, the way a send with no rung up is buffered rather than refused.
///
/// # Arguments
///
/// * `ladder` - the ladder.
/// * `topic` - the topic filter, as null-terminated UTF-8.
///
/// # Returns
///
/// [`PamojaStatus::Ok`] once the filter is live on a rung or held for the next
/// connect, or the status of the refusal if every connected rung refused it.
///
/// # Safety
///
/// `ladder` must be a live handle and `topic` a valid null-terminated UTF-8
/// string.
#[no_mangle]
pub unsafe extern "C" fn pamoja_ladder_subscribe(
    ladder: *mut PamojaLadder,
    topic: *const c_char,
) -> PamojaStatus {
    let Some(topic) = read_str(topic, "topic") else {
        return PamojaStatus::InvalidArgument;
    };
    let Some(inner) = ladder_inner(ladder) else {
        return PamojaStatus::InvalidArgument;
    };
    match runtime().block_on(inner.subscribe(topic)) {
        Ok(()) => PamojaStatus::Ok,
        Err(error) => fail(error),
    }
}

/// Waits for the next message from any rung that listens.
///
/// Every connected rung that listens is polled together and the first to deliver
/// wins. A rung whose link ends is left out until the next
/// [`pamoja_ladder_connect`].
///
/// # Arguments
///
/// * `ladder` - the ladder.
/// * `out_message` - receives a message handle to release with
///   [`pamoja_message_free`](crate::transport::pamoja_message_free).
///
/// # Returns
///
/// [`PamojaStatus::Ok`] with a message, or [`PamojaStatus::Closed`] if no
/// connected rung listens: none was added, the ladder is not connected, or every
/// listening link has ended.
///
/// # Safety
///
/// `ladder` must be a live handle and `out_message` must be writable.
#[no_mangle]
pub unsafe extern "C" fn pamoja_ladder_recv(
    ladder: *mut PamojaLadder,
    out_message: *mut *mut PamojaMessage,
) -> PamojaStatus {
    let Some(inner) = ladder_inner(ladder) else {
        return PamojaStatus::InvalidArgument;
    };
    if out_message.is_null() {
        set_last_error("out_message must not be null".to_owned());
        return PamojaStatus::InvalidArgument;
    }
    *out_message = ptr::null_mut();
    match runtime().block_on(inner.recv()) {
        Ok(Some(message)) => {
            *out_message = PamojaMessage::into_raw(message.topic, message.payload);
            PamojaStatus::Ok
        }
        Ok(None) => PamojaStatus::Ok,
        Err(error) => fail(error),
    }
}

/// Waits a limited time for the next message from any rung that listens.
///
/// Running out of time loses nothing: a message that arrives afterwards waits for
/// the next receive.
///
/// # Arguments
///
/// * `ladder` - the ladder.
/// * `timeout_ms` - how long to wait, in milliseconds.
/// * `out_message` - receives a message handle to release with
///   [`pamoja_message_free`](crate::transport::pamoja_message_free), or null when
///   the time ran out.
/// * `out_timed_out` - receives whether the time ran out before a message arrived.
///
/// # Returns
///
/// [`PamojaStatus::Ok`] with a message or with the time run out, or
/// [`PamojaStatus::Closed`] if no connected rung listens.
///
/// # Safety
///
/// `ladder` must be a live handle, and `out_message` and `out_timed_out` must be
/// writable.
#[no_mangle]
pub unsafe extern "C" fn pamoja_ladder_recv_within(
    ladder: *mut PamojaLadder,
    timeout_ms: u64,
    out_message: *mut *mut PamojaMessage,
    out_timed_out: *mut bool,
) -> PamojaStatus {
    let Some(inner) = ladder_inner(ladder) else {
        return PamojaStatus::InvalidArgument;
    };
    receive_within(timeout_ms, out_message, out_timed_out, || inner.recv())
}

/// Releases a ladder handle, and the rungs and buffer it owns.
///
/// Passing null is a no-op.
///
/// # Safety
///
/// `ladder` must be a handle from [`pamoja_ladder_new`] that has not already
/// been freed, or null. After this call it must not be used again.
#[no_mangle]
pub unsafe extern "C" fn pamoja_ladder_free(ladder: *mut PamojaLadder) {
    if !ladder.is_null() {
        drop(Box::from_raw(ladder));
    }
}

/// Borrows a ladder handle, rejecting a null pointer.
///
/// # Safety
///
/// `ladder` must be a live handle from [`pamoja_ladder_new`], or null.
unsafe fn ladder_handle<'a>(ladder: *mut PamojaLadder) -> Option<&'a mut PamojaLadder> {
    if ladder.is_null() {
        set_last_error("ladder must not be null".to_owned());
        return None;
    }
    Some(&mut *ladder)
}

/// Borrows the ladder inside a handle, rejecting a null or spent one.
///
/// # Safety
///
/// `ladder` must be a live handle from [`pamoja_ladder_new`], or null.
unsafe fn ladder_inner<'a>(
    ladder: *mut PamojaLadder,
) -> Option<&'a mut TransportLadder<StoreKind>> {
    let handle = ladder_handle(ladder)?;
    match handle.inner.as_mut() {
        Some(inner) => Some(inner),
        None => {
            set_last_error("this ladder is no longer usable".to_owned());
            None
        }
    }
}

/// Records an error and maps it onto a status.
fn fail(error: pamoja_core::Error) -> PamojaStatus {
    let status = PamojaStatus::from_error(&error);
    set_last_error(error.to_string());
    status
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::loopback::{
        pamoja_loopback_broker_free, pamoja_loopback_broker_is_reachable,
        pamoja_loopback_broker_new, pamoja_loopback_broker_set_reachable,
        pamoja_loopback_transport_connect, pamoja_loopback_transport_free,
        pamoja_loopback_transport_new, pamoja_loopback_transport_recv,
        pamoja_loopback_transport_send, pamoja_loopback_transport_subscribe,
        pamoja_transport_loopback,
    };
    use crate::sync::pamoja_store_memory;
    use crate::transport::{
        pamoja_message_free, pamoja_message_payload, pamoja_message_payload_len,
        pamoja_transport_faulty,
    };

    #[test]
    fn a_message_no_rung_takes_is_buffered_rather_than_lost() {
        unsafe {
            let ladder = pamoja_ladder_new(pamoja_store_memory(0));
            let topic = std::ffi::CString::new("sensors/1").expect("static");

            let mut delivery = PamojaDelivery::Sent;
            assert_eq!(
                pamoja_ladder_send(ladder, topic.as_ptr(), b"21.5".as_ptr(), 4, &mut delivery),
                PamojaStatus::Ok,
                "buffering is a success, not a failure"
            );
            assert_eq!(delivery, PamojaDelivery::Buffered);

            let mut waiting = 0;
            assert_eq!(
                pamoja_ladder_buffered(ladder, &mut waiting),
                PamojaStatus::Ok
            );
            assert_eq!(waiting, 1);

            pamoja_ladder_free(ladder);
        }
    }

    #[test]
    fn a_rung_that_refuses_falls_through_to_the_next() {
        unsafe {
            let broker = pamoja_loopback_broker_new();
            let listener = pamoja_loopback_transport_new(broker);
            pamoja_loopback_transport_connect(listener);
            let topic = std::ffi::CString::new("sensors/1").expect("static");
            pamoja_loopback_transport_subscribe(listener, topic.as_ptr());

            // The first rung fails its next send; the second is the same broker.
            let failing = pamoja_transport_faulty(pamoja_transport_loopback(broker), 1);
            let working = pamoja_transport_loopback(broker);

            let ladder = pamoja_ladder_new(pamoja_store_memory(0));
            assert_eq!(pamoja_ladder_rung(ladder, failing), PamojaStatus::Ok);
            assert_eq!(pamoja_ladder_rung(ladder, working), PamojaStatus::Ok);
            assert_eq!(pamoja_ladder_connect(ladder), PamojaStatus::Ok);

            let mut delivery = PamojaDelivery::Buffered;
            assert_eq!(
                pamoja_ladder_send(ladder, topic.as_ptr(), b"21.5".as_ptr(), 4, &mut delivery),
                PamojaStatus::Ok
            );
            assert_eq!(
                delivery,
                PamojaDelivery::Sent,
                "the second rung carried what the first refused"
            );

            let mut message = ptr::null_mut();
            pamoja_loopback_transport_recv(listener, &mut message);
            assert!(!message.is_null());
            let payload = std::slice::from_raw_parts(
                pamoja_message_payload(message),
                pamoja_message_payload_len(message),
            )
            .to_vec();
            assert_eq!(payload, b"21.5");
            pamoja_message_free(message);

            pamoja_ladder_free(ladder);
            pamoja_loopback_transport_free(listener);
            pamoja_loopback_broker_free(broker);
        }
    }

    #[test]
    fn a_flush_replays_what_was_buffered() {
        unsafe {
            let broker = pamoja_loopback_broker_new();
            let listener = pamoja_loopback_transport_new(broker);
            pamoja_loopback_transport_connect(listener);
            let topic = std::ffi::CString::new("sensors/1").expect("static");
            pamoja_loopback_transport_subscribe(listener, topic.as_ptr());

            let ladder = pamoja_ladder_new(pamoja_store_memory(0));

            // Nothing to send over yet, so it buffers.
            pamoja_ladder_send(ladder, topic.as_ptr(), b"one".as_ptr(), 3, ptr::null_mut());
            pamoja_ladder_send(ladder, topic.as_ptr(), b"two".as_ptr(), 3, ptr::null_mut());

            // The link comes back.
            assert_eq!(
                pamoja_ladder_rung(ladder, pamoja_transport_loopback(broker)),
                PamojaStatus::Ok
            );
            assert_eq!(pamoja_ladder_connect(ladder), PamojaStatus::Ok);

            let mut sent = 0;
            assert_eq!(pamoja_ladder_flush(ladder, &mut sent), PamojaStatus::Ok);
            assert_eq!(sent, 2, "both buffered messages went out");

            let mut waiting = 1;
            pamoja_ladder_buffered(ladder, &mut waiting);
            assert_eq!(waiting, 0, "and the buffer is empty");

            pamoja_ladder_free(ladder);
            pamoja_loopback_transport_free(listener);
            pamoja_loopback_broker_free(broker);
        }
    }

    #[test]
    fn a_command_comes_back_through_the_ladder() {
        unsafe {
            let broker = pamoja_loopback_broker_new();
            let ladder = pamoja_ladder_new(pamoja_store_memory(0));
            assert_eq!(
                pamoja_ladder_rung(ladder, pamoja_transport_loopback(broker)),
                PamojaStatus::Ok
            );
            assert_eq!(pamoja_ladder_connect(ladder), PamojaStatus::Ok);
            let topic = std::ffi::CString::new("commands/1").expect("static");
            assert_eq!(
                pamoja_ladder_subscribe(ladder, topic.as_ptr()),
                PamojaStatus::Ok
            );

            let upstream = pamoja_loopback_transport_new(broker);
            pamoja_loopback_transport_connect(upstream);
            pamoja_loopback_transport_send(upstream, topic.as_ptr(), b"open".as_ptr(), 4);

            let mut message = ptr::null_mut();
            assert_eq!(pamoja_ladder_recv(ladder, &mut message), PamojaStatus::Ok);
            assert!(
                !message.is_null(),
                "the command came back through the ladder"
            );
            let payload = std::slice::from_raw_parts(
                pamoja_message_payload(message),
                pamoja_message_payload_len(message),
            )
            .to_vec();
            assert_eq!(payload, b"open");
            pamoja_message_free(message);

            pamoja_ladder_free(ladder);
            pamoja_loopback_transport_free(upstream);
            pamoja_loopback_broker_free(broker);
        }
    }

    #[test]
    fn a_timed_receive_runs_out_and_leaves_the_next_message_queued() {
        unsafe {
            let broker = pamoja_loopback_broker_new();
            let ladder = pamoja_ladder_new(pamoja_store_memory(0));
            pamoja_ladder_rung(ladder, pamoja_transport_loopback(broker));
            pamoja_ladder_connect(ladder);
            let topic = std::ffi::CString::new("commands/1").expect("static");
            pamoja_ladder_subscribe(ladder, topic.as_ptr());

            let mut message = ptr::null_mut();
            let mut timed_out = false;
            assert_eq!(
                pamoja_ladder_recv_within(ladder, 20, &mut message, &mut timed_out),
                PamojaStatus::Ok
            );
            assert!(timed_out && message.is_null());

            let upstream = pamoja_loopback_transport_new(broker);
            pamoja_loopback_transport_connect(upstream);
            pamoja_loopback_transport_send(upstream, topic.as_ptr(), b"open".as_ptr(), 4);
            assert_eq!(
                pamoja_ladder_recv_within(ladder, 1000, &mut message, &mut timed_out),
                PamojaStatus::Ok
            );
            assert!(!timed_out && !message.is_null());
            let payload = std::slice::from_raw_parts(
                pamoja_message_payload(message),
                pamoja_message_payload_len(message),
            )
            .to_vec();
            assert_eq!(payload, b"open");
            pamoja_message_free(message);

            pamoja_ladder_free(ladder);
            pamoja_loopback_transport_free(upstream);
            pamoja_loopback_broker_free(broker);
        }
    }

    #[test]
    fn an_uplink_carries_what_an_unreachable_rung_refuses_and_is_never_listened_on() {
        unsafe {
            let near = pamoja_loopback_broker_new();
            let far = pamoja_loopback_broker_new();
            let ashore = pamoja_loopback_transport_new(far);
            pamoja_loopback_transport_connect(ashore);
            let reports = std::ffi::CString::new("reports").expect("static");
            let orders = std::ffi::CString::new("orders").expect("static");
            pamoja_loopback_transport_subscribe(ashore, reports.as_ptr());

            let ladder = pamoja_ladder_new(pamoja_store_memory(0));
            assert_eq!(
                pamoja_ladder_rung(ladder, pamoja_transport_loopback(near)),
                PamojaStatus::Ok
            );
            assert_eq!(
                pamoja_ladder_uplink(ladder, pamoja_transport_loopback(far)),
                PamojaStatus::Ok
            );
            pamoja_ladder_connect(ladder);
            assert_eq!(
                pamoja_ladder_subscribe(ladder, orders.as_ptr()),
                PamojaStatus::Ok
            );

            assert_eq!(
                pamoja_loopback_broker_set_reachable(near, false),
                PamojaStatus::Ok
            );
            assert!(!pamoja_loopback_broker_is_reachable(near));
            let mut delivery = PamojaDelivery::Buffered;
            pamoja_ladder_send(ladder, reports.as_ptr(), b"1".as_ptr(), 1, &mut delivery);
            assert_eq!(delivery, PamojaDelivery::Sent, "the uplink carried it");
            let mut message = ptr::null_mut();
            pamoja_loopback_transport_recv(ashore, &mut message);
            assert!(!message.is_null());
            pamoja_message_free(message);

            pamoja_loopback_transport_send(ashore, orders.as_ptr(), b"stop".as_ptr(), 4);
            let mut timed_out = false;
            pamoja_ladder_recv_within(ladder, 20, &mut message, &mut timed_out);
            assert!(timed_out, "an uplink is never listened on");

            pamoja_loopback_broker_set_reachable(far, false);
            pamoja_ladder_send(ladder, reports.as_ptr(), b"2".as_ptr(), 1, &mut delivery);
            assert_eq!(
                delivery,
                PamojaDelivery::Buffered,
                "with every link out of reach"
            );

            pamoja_ladder_free(ladder);
            pamoja_loopback_transport_free(ashore);
            pamoja_loopback_broker_free(near);
            pamoja_loopback_broker_free(far);
        }
    }

    #[test]
    fn a_receive_with_no_listening_rung_reports_closed() {
        unsafe {
            let ladder = pamoja_ladder_new(pamoja_store_memory(0));
            let mut message = ptr::null_mut();
            assert_eq!(
                pamoja_ladder_recv(ladder, &mut message),
                PamojaStatus::Closed
            );
            assert!(message.is_null());
            pamoja_ladder_free(ladder);
        }
    }

    #[test]
    fn a_null_handle_is_refused_rather_than_dereferenced() {
        unsafe {
            assert!(pamoja_ladder_new(ptr::null_mut()).is_null());
            assert_eq!(
                pamoja_ladder_connect(ptr::null_mut()),
                PamojaStatus::InvalidArgument
            );
            assert_eq!(
                pamoja_ladder_buffered(ptr::null_mut(), ptr::null_mut()),
                PamojaStatus::InvalidArgument
            );
            pamoja_ladder_free(ptr::null_mut());
        }
    }
}