rns_core/resource/

sender.rs

use alloc::borrow::Cow;
use alloc::boxed::Box;
use alloc::vec;
use alloc::vec::Vec;

use rns_crypto::Rng;

use super::advertisement::ResourceAdvertisement;
use super::parts::{build_hashmap, has_collision, map_hash, prepend_metadata};
use super::proof::{compute_expected_proof, compute_resource_hash, validate_proof};
use super::types::*;
use crate::buffer::types::Compressor;
use crate::constants::*;
use crate::hash::get_random_hash;

/// Resource sender state machine.
///
/// Creates an advertisement, handles part requests, sends parts, and validates proofs.
/// Returns `Vec<ResourceAction>` — no I/O, no callbacks.
pub struct ResourceSender {
    /// Current status
    pub status: ResourceStatus,
    /// Resource hash (SHA-256 of unencrypted data + random_hash), 32 bytes
    pub resource_hash: [u8; 32],
    /// Truncated hash (first 16 bytes of resource_hash)
    pub truncated_hash: [u8; 16],
    /// Expected proof (SHA-256 of unencrypted data + resource_hash)
    pub expected_proof: [u8; 32],
    /// Original hash (for multi-segment, first segment's hash)
    pub original_hash: [u8; 32],
    /// Random hash for map hashing (4 bytes)
    pub random_hash: Vec<u8>,
    /// SDU size
    pub sdu: usize,
    /// Encrypted bytes retained once; parts are slices of this buffer.
    parts: Box<[u8]>,
    /// Original partition size, independent of the public SDU field.
    part_sdu: usize,
    /// Part map hashes (4 bytes each)
    pub part_hashes: Vec<[u8; RESOURCE_MAPHASH_LEN]>,
    /// Concatenated hashmap bytes
    hashmap: Vec<u8>,
    /// Number of parts
    total_parts: usize,
    /// Number of unique parts sent
    pub sent_parts: usize,
    /// Tracks which part indices have been sent (for dedup)
    sent_indices: Vec<bool>,
    /// Flags
    pub flags: AdvFlags,
    /// Transfer size (encrypted data size)
    pub transfer_size: usize,
    /// Total uncompressed data size
    pub data_size: usize,
    /// Segment index (1-based)
    pub segment_index: u64,
    /// Total segments
    pub total_segments: u64,
    /// Request ID
    pub request_id: Option<Vec<u8>>,
    /// Retries left
    pub retries_left: usize,
    /// Max retries
    pub max_retries: usize,
    /// Max advertisement retries
    pub max_adv_retries: usize,
    /// RTT estimate (seconds)
    pub rtt: Option<f64>,
    /// Link RTT estimate (from link establishment)
    pub link_rtt: f64,
    /// Timeout factor
    pub timeout_factor: f64,
    /// Last activity timestamp
    pub last_activity: f64,
    /// Advertisement sent timestamp
    pub adv_sent: f64,
    /// Last part sent timestamp
    pub last_part_sent: f64,
    /// Sender grace time
    pub sender_grace_time: f64,
    /// Receiver min consecutive height (for search optimization)
    receiver_min_consecutive_height: usize,
}

impl ResourceSender {
    /// Create a new ResourceSender from unencrypted data.
    ///
    /// - `data`: raw application data (no metadata prefix)
    /// - `metadata`: optional pre-serialized metadata bytes
    /// - `sdu`: SDU size (usually RESOURCE_SDU = 464)
    /// - `encrypt_fn`: closure to encrypt the full data blob
    /// - `compressor`: Compressor trait for optional compression
    /// - `rng`: random number generator
    /// - `now`: current timestamp
    /// - `auto_compress`: whether to attempt compression
    /// - `is_response`: whether this is a response to a request
    /// - `request_id`: optional request ID
    /// - `segment_index`: 1-based segment number (1 for single-segment)
    /// - `total_segments`: total number of segments
    /// - `original_hash`: original hash from first segment (None for first segment)
    /// - `link_rtt`: current link RTT estimate
    /// - `traffic_timeout_factor`: link traffic timeout factor
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        data: &[u8],
        metadata: Option<&[u8]>,
        sdu: usize,
        encrypt_fn: &dyn Fn(&[u8]) -> Vec<u8>,
        compressor: &dyn Compressor,
        rng: &mut dyn Rng,
        now: f64,
        auto_compress: bool,
        is_response: bool,
        request_id: Option<Vec<u8>>,
        segment_index: u64,
        total_segments: u64,
        original_hash: Option<[u8; 32]>,
        link_rtt: f64,
        traffic_timeout_factor: f64,
    ) -> Result<Self, ResourceError> {
        // Build unencrypted data (metadata prefix + data)
        let uncompressed_data = match metadata {
            Some(meta) => Cow::Owned(prepend_metadata(data, meta)),
            None => Cow::Borrowed(data),
        };
        let has_metadata = metadata.is_some();

        let data_size = uncompressed_data.len();

        // Try compression
        let (working_data, compressed) =
            if auto_compress && uncompressed_data.len() <= RESOURCE_AUTO_COMPRESS_MAX_SIZE {
                match compressor.compress(&uncompressed_data) {
                    Some(compressed_data) if compressed_data.len() < uncompressed_data.len() => {
                        (Cow::Owned(compressed_data), true)
                    }
                    _ => (Cow::Borrowed(uncompressed_data.as_ref()), false),
                }
            } else {
                (Cow::Borrowed(uncompressed_data.as_ref()), false)
            };

        // Prepend random hash (4 bytes)
        let random_prefix: [u8; RESOURCE_RANDOM_HASH_SIZE] = {
            let rh = get_random_hash(rng);
            let mut buf = [0u8; RESOURCE_RANDOM_HASH_SIZE];
            buf.copy_from_slice(&rh[..RESOURCE_RANDOM_HASH_SIZE]);
            buf
        };
        let mut data_with_random =
            Vec::with_capacity(RESOURCE_RANDOM_HASH_SIZE + working_data.len());
        data_with_random.extend_from_slice(&random_prefix);
        data_with_random.extend_from_slice(&working_data);

        // Encrypt
        let encrypted_data = encrypt_fn(&data_with_random);
        let transfer_size = encrypted_data.len();

        // Keep trying until no collision in hashmap (max 100 attempts)
        let mut resource_hash;
        let mut truncated_resource_hash;
        let mut expected_proof;
        let mut final_random_hash;
        let mut part_hashes;
        let mut collision_retries = 0;
        const MAX_COLLISION_RETRIES: usize = 100;

        loop {
            final_random_hash = {
                let rh = get_random_hash(rng);
                rh[..RESOURCE_RANDOM_HASH_SIZE].to_vec()
            };

            resource_hash = compute_resource_hash(&uncompressed_data, &final_random_hash);
            truncated_resource_hash = {
                let mut t = [0u8; 16];
                t.copy_from_slice(&resource_hash[..16]);
                t
            };
            expected_proof = compute_expected_proof(&uncompressed_data, &resource_hash);

            part_hashes = if sdu == 0 {
                Vec::new()
            } else {
                encrypted_data
                    .chunks(sdu)
                    .map(|part| map_hash(part, &final_random_hash))
                    .collect()
            };

            if !has_collision(&part_hashes) {
                break;
            }
            // Collision detected, retry with new random hash
            collision_retries += 1;
            if collision_retries >= MAX_COLLISION_RETRIES {
                return Err(ResourceError::CollisionDetected);
            }
        }

        let hashmap = build_hashmap(&part_hashes);
        let total_parts = part_hashes.len();

        let orig_hash = original_hash.unwrap_or(resource_hash);

        let flags = AdvFlags {
            encrypted: true,
            compressed,
            split: total_segments > 1,
            is_request: request_id.is_some() && !is_response,
            is_response: request_id.is_some() && is_response,
            has_metadata,
        };

        Ok(ResourceSender {
            status: ResourceStatus::Queued,
            resource_hash,
            truncated_hash: truncated_resource_hash,
            expected_proof,
            original_hash: orig_hash,
            random_hash: final_random_hash,
            sdu,
            // Discard spare encryption capacity rather than retaining it per transfer.
            parts: if sdu == 0 {
                Box::default()
            } else {
                encrypted_data.into_boxed_slice()
            },
            part_sdu: sdu.max(1),
            part_hashes,
            hashmap,
            total_parts,
            sent_parts: 0,
            sent_indices: vec![false; total_parts],
            flags,
            transfer_size,
            data_size,
            segment_index,
            total_segments,
            request_id,
            retries_left: RESOURCE_MAX_RETRIES,
            max_retries: RESOURCE_MAX_RETRIES,
            max_adv_retries: RESOURCE_MAX_ADV_RETRIES,
            rtt: None,
            link_rtt,
            timeout_factor: traffic_timeout_factor,
            last_activity: now,
            adv_sent: now,
            last_part_sent: now,
            sender_grace_time: RESOURCE_SENDER_GRACE_TIME,
            receiver_min_consecutive_height: 0,
        })
    }

    /// Generate the advertisement for the given hashmap segment.
    pub fn get_advertisement(&self, segment: usize) -> Vec<u8> {
        let adv = ResourceAdvertisement {
            transfer_size: self.transfer_size as u64,
            data_size: self.data_size as u64,
            num_parts: self.total_parts as u64,
            resource_hash: self.resource_hash.to_vec(),
            random_hash: self.random_hash.clone(),
            original_hash: self.original_hash.to_vec(),
            hashmap: self.hashmap.clone(),
            flags: self.flags,
            segment_index: self.segment_index,
            total_segments: self.total_segments,
            request_id: self.request_id.clone(),
        };
        adv.pack(segment)
    }

    /// Advertise the resource. Returns SendAdvertisement action.
    pub fn advertise(&mut self, now: f64) -> Vec<ResourceAction> {
        self.status = ResourceStatus::Advertised;
        self.last_activity = now;
        self.adv_sent = now;
        self.retries_left = self.max_adv_retries;
        let adv_data = self.get_advertisement(0);
        vec![ResourceAction::SendAdvertisement(adv_data)]
    }

    /// Handle a request for parts (RESOURCE_REQ context).
    ///
    /// request_data format:
    /// [exhausted_flag: u8][last_map_hash: 4 bytes if exhausted][resource_hash: 32 bytes][requested_hashes: N*4 bytes]
    pub fn handle_request(&mut self, request_data: &[u8], now: f64) -> Vec<ResourceAction> {
        if self.status == ResourceStatus::Failed {
            return vec![];
        }

        let wants_more_hashmap = request_data.first() == Some(&RESOURCE_HASHMAP_IS_EXHAUSTED);
        let pad = if wants_more_hashmap {
            1 + RESOURCE_MAPHASH_LEN
        } else {
            1
        };

        if request_data.len() < pad + 32 || request_data[pad..pad + 32] != self.resource_hash {
            return vec![];
        }

        // Measure RTT from advertisement
        if self.rtt.is_none() {
            self.rtt = Some(now - self.adv_sent);
        }

        if self.status != ResourceStatus::Transferring {
            self.status = ResourceStatus::Transferring;
        }

        self.retries_left = self.max_retries;
        self.last_activity = now;

        let requested_hashes_data = &request_data[pad + 32..];
        let mut actions = Vec::new();

        // Parse requested map hashes
        let num_requested = requested_hashes_data.len() / RESOURCE_MAPHASH_LEN;
        let mut map_hashes_requested = Vec::with_capacity(num_requested);
        for i in 0..num_requested {
            let start = i * RESOURCE_MAPHASH_LEN;
            let end = start + RESOURCE_MAPHASH_LEN;
            if end <= requested_hashes_data.len() {
                let mut h = [0u8; RESOURCE_MAPHASH_LEN];
                h.copy_from_slice(&requested_hashes_data[start..end]);
                map_hashes_requested.push(h);
            }
        }

        // Search for requested parts within guard window
        let search_start = self.receiver_min_consecutive_height;
        let search_end = core::cmp::min(
            search_start + RESOURCE_COLLISION_GUARD_SIZE,
            self.total_parts,
        );

        for part_idx in search_start..search_end {
            if map_hashes_requested.contains(&self.part_hashes[part_idx]) {
                actions.push(ResourceAction::SendPart(self.part(part_idx).to_vec()));
                if !self.sent_indices[part_idx] {
                    self.sent_indices[part_idx] = true;
                    self.sent_parts += 1;
                }
                self.last_part_sent = now;
            }
        }

        // Handle hashmap exhaustion
        if wants_more_hashmap {
            match self.build_hmu(request_data, now) {
                Ok(hmu) => actions.push(ResourceAction::SendHmu(hmu)),
                Err(_) => {
                    actions.extend(self.cancel());
                    return actions;
                }
            }
        }

        // Check if all parts sent
        if self.sent_parts >= self.total_parts {
            self.status = ResourceStatus::AwaitingProof;
            self.retries_left = 3; // hardcoded in Python
        }

        actions
    }

    /// Build hashmap update data.
    fn build_hmu(&mut self, request_data: &[u8], now: f64) -> Result<Vec<u8>, ResourceError> {
        if request_data.len() < 1 + RESOURCE_MAPHASH_LEN {
            return Err(ResourceError::InvalidState);
        }

        let last_map_hash_bytes = &request_data[1..1 + RESOURCE_MAPHASH_LEN];
        let mut last_map_hash = [0u8; RESOURCE_MAPHASH_LEN];
        last_map_hash.copy_from_slice(last_map_hash_bytes);

        // Find the part index of the last map hash
        let search_start = self.receiver_min_consecutive_height;
        let search_end = core::cmp::min(
            search_start + RESOURCE_COLLISION_GUARD_SIZE,
            self.total_parts,
        );

        let mut part_index = search_start;
        for idx in search_start..search_end {
            part_index = idx + 1;
            if self.part_hashes[idx] == last_map_hash {
                break;
            }
        }

        // Update receiver min consecutive height
        self.receiver_min_consecutive_height = if part_index > RESOURCE_WINDOW_MAX {
            part_index - 1 - RESOURCE_WINDOW_MAX
        } else {
            0
        };

        // Verify alignment
        if !part_index.is_multiple_of(RESOURCE_HASHMAP_MAX_LEN) {
            return Err(ResourceError::InvalidState);
        }

        let segment = part_index / RESOURCE_HASHMAP_MAX_LEN;
        let hashmap_start = segment * RESOURCE_HASHMAP_MAX_LEN;
        let hashmap_end =
            core::cmp::min((segment + 1) * RESOURCE_HASHMAP_MAX_LEN, self.total_parts);

        let mut hashmap_segment = Vec::new();
        for i in hashmap_start..hashmap_end {
            hashmap_segment.extend_from_slice(
                &self.hashmap[i * RESOURCE_MAPHASH_LEN..(i + 1) * RESOURCE_MAPHASH_LEN],
            );
        }

        if hashmap_segment.is_empty() {
            return Err(ResourceError::InvalidState);
        }

        // Build HMU: resource_hash + msgpack([segment, hashmap])
        let hmu_payload = crate::msgpack::pack(&crate::msgpack::Value::Array(vec![
            crate::msgpack::Value::UInt(segment as u64),
            crate::msgpack::Value::Bin(hashmap_segment),
        ]));

        let mut hmu = Vec::with_capacity(32 + hmu_payload.len());
        hmu.extend_from_slice(&self.resource_hash);
        hmu.extend_from_slice(&hmu_payload);

        self.last_activity = now;
        Ok(hmu)
    }

    /// Handle proof from receiver.
    pub fn handle_proof(&mut self, proof_data: &[u8], _now: f64) -> Vec<ResourceAction> {
        if self.status == ResourceStatus::Failed {
            return vec![];
        }

        match validate_proof(proof_data, &self.resource_hash, &self.expected_proof) {
            Ok(true) => {
                self.status = ResourceStatus::Complete;
                vec![ResourceAction::Completed]
            }
            Ok(false) => {
                self.status = ResourceStatus::Failed;
                vec![ResourceAction::Failed(ResourceError::InvalidProof)]
            }
            Err(e) => {
                self.status = ResourceStatus::Failed;
                vec![ResourceAction::Failed(e)]
            }
        }
    }

    /// Handle rejection from receiver.
    pub fn handle_reject(&mut self) -> Vec<ResourceAction> {
        self.status = ResourceStatus::Rejected;
        vec![ResourceAction::Failed(ResourceError::Rejected)]
    }

    /// Cancel the transfer.
    pub fn cancel(&mut self) -> Vec<ResourceAction> {
        if self.status < ResourceStatus::Complete {
            self.status = ResourceStatus::Failed;
            vec![ResourceAction::SendCancelInitiator(
                self.resource_hash.to_vec(),
            )]
        } else {
            vec![]
        }
    }

    /// Periodic tick. Checks for timeouts.
    pub fn tick(&mut self, now: f64) -> Vec<ResourceAction> {
        if self.status >= ResourceStatus::Complete {
            return vec![];
        }

        match self.status {
            ResourceStatus::Advertised => {
                let timeout = self.adv_sent
                    + self.rtt.unwrap_or(self.link_rtt * self.timeout_factor)
                    + RESOURCE_PROCESSING_GRACE;
                if now > timeout {
                    if self.retries_left == 0 {
                        self.status = ResourceStatus::Failed;
                        return vec![ResourceAction::Failed(ResourceError::Timeout)];
                    }
                    self.retries_left -= 1;
                    self.last_activity = now;
                    self.adv_sent = now;
                    let adv_data = self.get_advertisement(0);
                    return vec![ResourceAction::SendAdvertisement(adv_data)];
                }
            }
            ResourceStatus::Transferring => {
                let rtt = self.rtt.unwrap_or(1.0);
                let max_extra_wait: f64 = (0..self.max_retries)
                    .map(|r| (r as f64 + 1.0) * RESOURCE_PER_RETRY_DELAY)
                    .sum();
                let max_wait = rtt * self.timeout_factor * self.max_retries as f64
                    + self.sender_grace_time
                    + max_extra_wait;
                if now > self.last_activity + max_wait {
                    self.status = ResourceStatus::Failed;
                    return vec![ResourceAction::Failed(ResourceError::Timeout)];
                }
            }
            ResourceStatus::AwaitingProof => {
                let rtt = self.rtt.unwrap_or(1.0);
                let timeout = self.last_part_sent
                    + rtt * RESOURCE_PROOF_TIMEOUT_FACTOR
                    + self.sender_grace_time;
                if now > timeout {
                    if self.retries_left == 0 {
                        self.status = ResourceStatus::Failed;
                        return vec![ResourceAction::Failed(ResourceError::Timeout)];
                    }
                    self.retries_left -= 1;
                    self.last_part_sent = now;
                    // In Python, this queries network cache. We just signal retry.
                    return vec![];
                }
            }
            _ => {}
        }

        vec![]
    }

    fn part(&self, index: usize) -> &[u8] {
        let start = index * self.part_sdu;
        &self.parts[start..start + self.part_sdu.min(self.parts.len() - start)]
    }

    /// Get the total number of parts.
    pub fn total_parts(&self) -> usize {
        self.total_parts
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::buffer::types::NoopCompressor;

    fn identity_encrypt(data: &[u8]) -> Vec<u8> {
        data.to_vec()
    }

    fn make_sender(data: &[u8]) -> ResourceSender {
        let mut rng = rns_crypto::FixedRng::new(&[0x42; 64]);
        ResourceSender::new(
            data,
            None,
            RESOURCE_SDU,
            &identity_encrypt,
            &NoopCompressor,
            &mut rng,
            1000.0,
            false,
            false,
            None,
            1,
            1,
            None,
            0.5,
            6.0,
        )
        .unwrap()
    }

    fn varying_data(len: usize) -> Vec<u8> {
        let mut state = 0x1234_5678u32;
        (0..len)
            .map(|_| {
                state = state.wrapping_mul(1_664_525).wrapping_add(1_013_904_223);
                (state >> 16) as u8
            })
            .collect()
    }

    #[test]
    fn preparation_wire_transcript_matches_owned_buffer_baseline() {
        // Exercise absent/empty metadata and compression skipped, unavailable,
        // equal-sized, larger, or accepted. The synthetic codec isolates buffer
        // selection; full codec round trips are covered separately.
        struct Codec(u8);
        impl Compressor for Codec {
            fn compress(&self, data: &[u8]) -> Option<Vec<u8>> {
                match self.0 {
                    0 | 1 => None,
                    2 => Some(data.to_vec()),
                    3 => Some([data, &[0]].concat()),
                    _ => Some(data[..data.len() / 2].to_vec()),
                }
            }
            fn decompress_bounded(
                &self,
                _: &[u8],
                _: usize,
            ) -> Result<Vec<u8>, crate::buffer::types::DecompressError> {
                unreachable!("sender-only transcript")
            }
        }
        let mut transcript = Vec::new();
        for len in [0, 1, 1000] {
            let data = varying_data(len);
            for metadata in [None, Some(&b""[..]), Some(&b"metadata"[..])] {
                for mode in 0..5 {
                    for response in [false, true] {
                        let mut rng = rns_crypto::FixedRng::new(&[0x42; 64]);
                        let mut sender = ResourceSender::new(
                            &data,
                            metadata,
                            RESOURCE_SDU,
                            &identity_encrypt,
                            &Codec(mode),
                            &mut rng,
                            1000.0,
                            mode != 0,
                            response,
                            Some(vec![0xAB; 16]),
                            2,
                            3,
                            Some([0xCD; 32]),
                            0.5,
                            6.0,
                        )
                        .unwrap();
                        transcript.extend_from_slice(&sender.resource_hash);
                        transcript.extend_from_slice(&sender.expected_proof);
                        transcript.extend_from_slice(&(sender.data_size as u64).to_be_bytes());
                        for action in sender.advertise(1000.0) {
                            if let ResourceAction::SendAdvertisement(raw) = action {
                                transcript.extend_from_slice(&(raw.len() as u64).to_be_bytes());
                                transcript.extend_from_slice(&raw);
                            } else {
                                panic!("expected advertisement");
                            }
                        }
                        for part in sender.parts.chunks(sender.part_sdu) {
                            transcript.extend_from_slice(&(part.len() as u64).to_be_bytes());
                            transcript.extend_from_slice(part);
                        }
                    }
                }
            }
        }
        // Recorded from the original preparation code before borrowing buffers.
        assert_eq!(
            crate::hash::full_hash(&transcript),
            [
                67, 44, 221, 45, 186, 73, 169, 206, 206, 134, 142, 225, 182, 205, 22, 233, 153,
                131, 88, 228, 84, 36, 149, 65, 72, 213, 69, 130, 12, 198, 159, 159
            ]
        );
    }

    #[test]
    fn test_new_sender_status() {
        let sender = make_sender(b"test data");
        assert_eq!(sender.status, ResourceStatus::Queued);
    }

    #[test]
    fn test_new_sender_parts() {
        let data = vec![0xAA; 1000];
        let sender = make_sender(&data);
        // 4 (random) + 1000 data = 1004 encrypted bytes
        // 1004 / 464 = 3 parts (464, 464, 76)
        assert_eq!(sender.total_parts(), 3);
    }

    #[test]
    fn test_advertise() {
        let mut sender = make_sender(b"test data");
        let actions = sender.advertise(1000.0);
        assert_eq!(sender.status, ResourceStatus::Advertised);
        assert_eq!(actions.len(), 1);
        match &actions[0] {
            ResourceAction::SendAdvertisement(data) => {
                assert!(!data.is_empty());
            }
            _ => panic!("Expected SendAdvertisement"),
        }
    }

    #[test]
    fn test_handle_request_basic() {
        let mut sender = make_sender(b"short");
        sender.advertise(1000.0);

        // Build a request: [not_exhausted][resource_hash][first part hash]
        let mut request = Vec::new();
        request.push(RESOURCE_HASHMAP_IS_NOT_EXHAUSTED);
        request.extend_from_slice(&sender.resource_hash);
        request.extend_from_slice(&sender.part_hashes[0]);

        let actions = sender.handle_request(&request, 1001.0);
        assert!(!actions.is_empty());
        // Should have sent a part
        let has_part = actions
            .iter()
            .any(|a| matches!(a, ResourceAction::SendPart(_)));
        assert!(has_part);
    }

    #[test]
    fn requested_parts_and_retries_preserve_original_partition() {
        for len in [
            0,
            1,
            RESOURCE_SDU - 4,
            RESOURCE_SDU - 3,
            RESOURCE_SDU * 3 + 17,
        ] {
            let data = varying_data(len);
            let mut sender = make_sender(&data);
            // FixedRng and identity encryption leave the original bytes visible.
            let mut encrypted = vec![0; RESOURCE_RANDOM_HASH_SIZE];
            // The random prefix is a hash of RNG output, not raw RNG bytes.
            encrypted[..RESOURCE_RANDOM_HASH_SIZE]
                .copy_from_slice(&crate::hash::full_hash(&[0x42; 16])[..RESOURCE_RANDOM_HASH_SIZE]);
            encrypted.extend_from_slice(&data);
            let expected: Vec<Vec<u8>> =
                encrypted.chunks(RESOURCE_SDU).map(<[u8]>::to_vec).collect();
            sender.advertise(1000.0);
            let mut request = vec![RESOURCE_HASHMAP_IS_NOT_EXHAUSTED];
            request.extend_from_slice(&sender.resource_hash);
            for hash in &sender.part_hashes {
                request.extend_from_slice(hash);
            }
            // Public SDU changes must not repartition a transfer already prepared.
            sender.sdu = 1;
            for now in [1001.0, 1002.0] {
                let parts: Vec<Vec<u8>> = sender
                    .handle_request(&request, now)
                    .into_iter()
                    .filter_map(|action| match action {
                        ResourceAction::SendPart(part) => Some(part),
                        _ => None,
                    })
                    .collect();
                assert_eq!(parts, expected);
                assert_eq!(sender.sent_parts, expected.len());
            }
        }
    }

    #[test]
    fn test_request_for_another_resource_is_ignored_without_mutating_sender() {
        let mut sender = make_sender(b"short");
        sender.advertise(1000.0);
        let status = sender.status;
        let last_activity = sender.last_activity;
        let retries_left = sender.retries_left;
        let mut request = Vec::new();
        request.push(RESOURCE_HASHMAP_IS_NOT_EXHAUSTED);
        request.extend_from_slice(&[0xEE; 32]);
        request.extend_from_slice(&sender.part_hashes[0]);

        let actions = sender.handle_request(&request, 2000.0);

        assert!(actions.is_empty());
        assert_eq!(sender.status, status);
        assert_eq!(sender.last_activity, last_activity);
        assert_eq!(sender.retries_left, retries_left);
        assert_eq!(sender.sent_parts, 0);
    }

    fn exhausted_request(
        sender: &ResourceSender,
        last_map_hash: [u8; RESOURCE_MAPHASH_LEN],
    ) -> Vec<u8> {
        let mut request = Vec::new();
        request.push(RESOURCE_HASHMAP_IS_EXHAUSTED);
        request.extend_from_slice(&last_map_hash);
        request.extend_from_slice(&sender.resource_hash);
        request
    }

    #[test]
    fn test_empty_hmu_generation_cancels_sender() {
        let data =
            varying_data(RESOURCE_HASHMAP_MAX_LEN * RESOURCE_SDU - RESOURCE_RANDOM_HASH_SIZE);
        let mut sender = make_sender(&data);
        assert_eq!(sender.total_parts(), RESOURCE_HASHMAP_MAX_LEN);
        sender.advertise(1000.0);
        let request = exhausted_request(&sender, sender.part_hashes[RESOURCE_HASHMAP_MAX_LEN - 1]);

        let actions = sender.handle_request(&request, 1001.0);

        assert_eq!(sender.status, ResourceStatus::Failed);
        assert!(actions
            .iter()
            .any(|action| matches!(action, ResourceAction::SendCancelInitiator(_))));
        assert!(!actions
            .iter()
            .any(|action| matches!(action, ResourceAction::SendHmu(_))));
    }

    #[test]
    fn test_misaligned_hmu_request_cancels_sender() {
        let data = varying_data((RESOURCE_HASHMAP_MAX_LEN + 2) * RESOURCE_SDU);
        let mut sender = make_sender(&data);
        sender.advertise(1000.0);
        let request = exhausted_request(&sender, sender.part_hashes[5]);

        let actions = sender.handle_request(&request, 1001.0);

        assert_eq!(sender.status, ResourceStatus::Failed);
        assert!(actions
            .iter()
            .any(|action| matches!(action, ResourceAction::SendCancelInitiator(_))));
        assert!(!actions
            .iter()
            .any(|action| matches!(action, ResourceAction::SendHmu(_))));
    }

    #[test]
    fn test_valid_nonempty_hmu_is_emitted() {
        let data = varying_data((RESOURCE_HASHMAP_MAX_LEN + 2) * RESOURCE_SDU);
        let mut sender = make_sender(&data);
        sender.advertise(1000.0);
        let request = exhausted_request(&sender, sender.part_hashes[RESOURCE_HASHMAP_MAX_LEN - 1]);

        let actions = sender.handle_request(&request, 1001.0);

        assert_ne!(sender.status, ResourceStatus::Failed);
        let hmu = actions
            .iter()
            .find_map(|action| match action {
                ResourceAction::SendHmu(data) => Some(data),
                _ => None,
            })
            .expect("valid exhausted request should produce an HMU");
        let (value, consumed) = crate::msgpack::unpack(&hmu[32..]).unwrap();
        assert_eq!(consumed, hmu.len() - 32);
        let array = value.as_array().unwrap();
        assert_eq!(array[0].as_uint(), Some(1));
        assert_eq!(
            array[1].as_bin().map(<[u8]>::len),
            Some((sender.total_parts() - RESOURCE_HASHMAP_MAX_LEN) * RESOURCE_MAPHASH_LEN)
        );
    }

    #[test]
    fn test_all_parts_sent_awaiting_proof() {
        let mut sender = make_sender(b"hi");
        sender.advertise(1000.0);

        // Request all parts
        let mut request = Vec::new();
        request.push(RESOURCE_HASHMAP_IS_NOT_EXHAUSTED);
        request.extend_from_slice(&sender.resource_hash);
        for h in &sender.part_hashes.clone() {
            request.extend_from_slice(h);
        }

        let _actions = sender.handle_request(&request, 1001.0);
        assert_eq!(sender.status, ResourceStatus::AwaitingProof);
        assert_eq!(sender.retries_left, 3);
    }

    #[test]
    fn test_valid_proof() {
        let mut sender = make_sender(b"data");
        sender.advertise(1000.0);

        let proof_data =
            super::super::proof::build_proof_data(&sender.resource_hash, &sender.expected_proof);
        let actions = sender.handle_proof(&proof_data, 1002.0);
        assert_eq!(sender.status, ResourceStatus::Complete);
        assert!(actions
            .iter()
            .any(|a| matches!(a, ResourceAction::Completed)));
    }

    #[test]
    fn test_invalid_proof() {
        let mut sender = make_sender(b"data");
        sender.advertise(1000.0);

        let wrong_proof = [0xFF; 32];
        let proof_data = super::super::proof::build_proof_data(&sender.resource_hash, &wrong_proof);
        let _actions = sender.handle_proof(&proof_data, 1002.0);
        assert_eq!(sender.status, ResourceStatus::Failed);
    }

    #[test]
    fn test_handle_reject() {
        let mut sender = make_sender(b"data");
        sender.advertise(1000.0);
        let _actions = sender.handle_reject();
        assert_eq!(sender.status, ResourceStatus::Rejected);
    }

    #[test]
    fn test_cancel() {
        let mut sender = make_sender(b"data");
        sender.advertise(1000.0);
        let actions = sender.cancel();
        assert_eq!(sender.status, ResourceStatus::Failed);
        assert!(actions
            .iter()
            .any(|a| matches!(a, ResourceAction::SendCancelInitiator(_))));
    }

    #[test]
    fn test_cancel_already_complete() {
        let mut sender = make_sender(b"data");
        sender.status = ResourceStatus::Complete;
        let actions = sender.cancel();
        assert!(actions.is_empty());
    }

    #[test]
    fn test_tick_advertised_timeout() {
        let mut sender = make_sender(b"data");
        sender.advertise(1000.0);
        sender.retries_left = 0;

        // Way past timeout
        let _actions = sender.tick(2000.0);
        assert_eq!(sender.status, ResourceStatus::Failed);
    }

    #[test]
    fn test_tick_advertised_retry() {
        let mut sender = make_sender(b"data");
        sender.advertise(1000.0);
        assert!(sender.retries_left > 0);

        let actions = sender.tick(2000.0);
        // Should retry advertisement
        assert!(actions
            .iter()
            .any(|a| matches!(a, ResourceAction::SendAdvertisement(_))));
    }

    #[test]
    fn test_resource_hash_is_32_bytes() {
        let sender = make_sender(b"data");
        assert_eq!(sender.resource_hash.len(), 32);
        assert_eq!(sender.expected_proof.len(), 32);
    }

    #[test]
    fn multipart_resource_identity_uses_complete_original_data() {
        let data = varying_data(RESOURCE_SDU * 3 + 17);
        let sender = make_sender(&data);

        assert!(sender.total_parts() > 1);
        assert_eq!(
            sender.resource_hash,
            compute_resource_hash(&data, &sender.random_hash)
        );
        assert_eq!(
            sender.expected_proof,
            compute_expected_proof(&data, &sender.resource_hash)
        );
        assert_ne!(
            sender.resource_hash,
            compute_resource_hash(sender.part(0), &sender.random_hash)
        );
    }

    #[test]
    fn test_sender_with_metadata() {
        let mut rng = rns_crypto::FixedRng::new(&[0x55; 64]);
        let sender = ResourceSender::new(
            b"data",
            Some(b"metadata"),
            RESOURCE_SDU,
            &identity_encrypt,
            &NoopCompressor,
            &mut rng,
            1000.0,
            false,
            false,
            None,
            1,
            1,
            None,
            0.5,
            6.0,
        )
        .unwrap();
        assert!(sender.flags.has_metadata);
    }

    #[test]
    fn test_multi_segment_sender() {
        let orig_hash = [0xBB; 32];
        let mut rng = rns_crypto::FixedRng::new(&[0x66; 64]);
        let sender = ResourceSender::new(
            b"segment 2 data",
            None,
            RESOURCE_SDU,
            &identity_encrypt,
            &NoopCompressor,
            &mut rng,
            1000.0,
            false,
            false,
            None,
            2,
            5,
            Some(orig_hash),
            0.5,
            6.0,
        )
        .unwrap();
        assert_eq!(sender.segment_index, 2);
        assert_eq!(sender.total_segments, 5);
        assert_eq!(sender.original_hash, orig_hash);
        assert!(sender.flags.split);
    }
}