simple-someip 0.8.0

//! Service Discovery session-state tracking, decoupled from socket ownership.
//!
//! [`SdStateManager`] owns the session-ID counter used by every outgoing
//! SOME/IP-SD message this server emits (`OfferService` announcements,
//! unicast Offer replies, `SubscribeAck`, `SubscribeNack`). It also builds
//! and sends `OfferService` announcements when given a socket.
//!
//! Keeping this state in its own type prepares the server for upcoming
//! transport abstraction: once `TransportSocket` lands, the `&UdpSocket`
//! parameter on [`SdStateManager::send_offer_service`] becomes the single
//! migration point for the announcement path.

use core::net::SocketAddrV4;
use core::sync::atomic::{AtomicU32, Ordering};

use crate::protocol::sd::{
    self, Entry, Flags, OptionsCount, RebootFlag, ServiceEntry, TransportProtocol,
};
use crate::transport::TransportSocket;

use super::{Error, ServerConfig};

/// Tracks the SD session-ID counter and emits `OfferService` announcements.
///
/// Session IDs increment with each SD message and wrap from `0xFFFF` back
/// to `0x0001` (skipping `0`, which is reserved). Per AUTOSAR SOME/IP-SD,
/// the reboot flag on emitted SD messages is
/// [`RebootFlag::RecentlyRebooted`] from startup until the counter wraps
/// once, then [`RebootFlag::Continuous`] permanently — `SdStateManager`
/// tracks that transition and bundles the `(session_id, reboot_flag)` pair
/// atomically through
/// [`Self::next_session_id_with_reboot_flag`] so every server-side SD
/// emission path reads from a single source of truth and concurrent
/// emitters cannot race around the wrap boundary.
#[derive(Debug)]
pub struct SdStateManager {
    /// Packed `(has_wrapped, session_id)` state.
    ///
    /// - bits 0..16: current session id (1..=0xFFFF, never 0).
    /// - bit 16: `has_wrapped` flag — once set, never cleared.
    /// - bits 17..32: reserved, must remain 0.
    ///
    /// Packed into a single `AtomicU32` so a single `fetch_update`
    /// produces a consistent `(session_id, reboot_flag)` pair across
    /// concurrent emitters around the `0xFFFF → 0x0001` wrap boundary.
    /// Two separate atomics could be interleaved by another emitter
    /// between the increment and the wrap-flag latch; with one atomic,
    /// the pair is computed in one CAS step.
    session_state: AtomicU32,
}

const SID_MASK: u32 = 0xFFFF;
const WRAPPED_BIT: u32 = 1 << 16;

impl SdStateManager {
    /// Construct an `SdStateManager` with a fresh session counter
    /// (starts at `1`, reboot flag = `RecentlyRebooted`).
    ///
    /// `const fn` so consumers can declare a `static`-storage instance
    /// without an allocator:
    ///
    /// ```ignore
    /// static SD_STATE: SdStateManager = SdStateManager::new();
    /// // pass `&SD_STATE` (an `&'static SdStateManager`) into the
    /// // appropriate `Server` constructor.
    /// ```
    #[must_use]
    pub const fn new() -> Self {
        Self::with_initial(1)
    }
}

impl Default for SdStateManager {
    /// Equivalent to [`Self::new`]. Provided for clippy-pedantic
    /// completeness; bare-metal callers should prefer the explicit
    /// `SdStateManager::new()` because it is `const` and works in a
    /// `static` initializer.
    fn default() -> Self {
        Self::new()
    }
}

impl SdStateManager {
    /// Construct with a specific starting session counter. `pub`
    /// (rather than `pub(super)`) so external test harnesses — e.g.
    /// `tests/vsomeip_sd_compat.rs`'s wire-format checks — can
    /// pre-seed counter state to validate wrap-around behaviour
    /// without driving a full Server lifecycle. Production callers
    /// should use [`Self::new`].
    #[must_use]
    pub const fn with_initial(initial: u16) -> Self {
        Self {
            // has_wrapped starts false; session_id starts at `initial`.
            session_state: AtomicU32::new(initial as u32),
        }
    }

    /// Advance the counter and return the next SOME/IP-SD session ID
    /// (`client_id = 0`, session ID in the low 16 bits) together with the
    /// reboot flag that *belongs to this same emission*. Skips 0 on wrap,
    /// and latches the `has_wrapped` bit the first time the counter
    /// crosses the `0xFFFF → 0x0001` boundary so the reboot flag flips
    /// to [`RebootFlag::Continuous`] permanently.
    ///
    /// `(session_id, reboot_flag)` is computed atomically inside one
    /// `fetch_update` so concurrent emitters always agree on the pair.
    /// A previous implementation used two separate atomics and could
    /// race around the wrap boundary, advertising
    /// `(0xFFFF, Continuous)` or `(0x0001, RecentlyRebooted)` — both
    /// violations of AUTOSAR SOME/IP-SD's stated semantics that the
    /// wrap message itself carries `Continuous`.
    ///
    /// # Panics
    ///
    /// Cannot panic in practice: the inner `fetch_update` closure
    /// always returns `Some(_)` (the wrap step is unconditional), so
    /// the `.unwrap()` is statically infallible. Documented for
    /// clippy's `missing_panics_doc` and as a tripwire if the closure
    /// is ever changed to be conditional.
    pub fn next_session_id_with_reboot_flag(&self) -> (u32, RebootFlag) {
        let prev_state = self
            .session_state
            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |state| {
                let prev_sid = (state & SID_MASK) as u16;
                let prev_wrapped = (state & WRAPPED_BIT) != 0;
                let next_sid = match prev_sid.wrapping_add(1) {
                    0 => 1u16,
                    n => n,
                };
                // Latch wrap on the 0xFFFF → 0x0001 transition.
                let next_wrapped = prev_wrapped || prev_sid == u16::MAX;
                let next_state =
                    (u32::from(next_sid)) | (if next_wrapped { WRAPPED_BIT } else { 0 });
                Some(next_state)
            })
            .unwrap();
        // Re-derive the new state from the prev we observed; this is
        // the *same* computation the closure performed and produces
        // exactly the new state we just stored.
        let prev_sid = (prev_state & SID_MASK) as u16;
        let prev_wrapped = (prev_state & WRAPPED_BIT) != 0;
        let next_sid = match prev_sid.wrapping_add(1) {
            0 => 1u16,
            n => n,
        };
        let next_wrapped = prev_wrapped || prev_sid == u16::MAX;
        let session_id = u32::from(next_sid);
        let reboot_flag = if next_wrapped {
            RebootFlag::Continuous
        } else {
            RebootFlag::RecentlyRebooted
        };
        (session_id, reboot_flag)
    }

    /// Convenience: advance the counter and return only the session id.
    /// Use [`Self::next_session_id_with_reboot_flag`] when the same
    /// emission also needs the reboot flag — calling these two methods
    /// separately races around the wrap boundary. Only used by unit
    /// tests; production paths take the atomic pair.
    #[cfg(test)]
    pub(super) fn next_session_id(&self) -> u32 {
        self.next_session_id_with_reboot_flag().0
    }

    /// Current SD reboot flag for this server.
    ///
    /// Returns [`RebootFlag::RecentlyRebooted`] until the session counter
    /// has wrapped past `0xFFFF` at least once, then
    /// [`RebootFlag::Continuous`] permanently. Production emission paths
    /// must use [`Self::next_session_id_with_reboot_flag`] instead to
    /// avoid a TOCTOU race around the wrap boundary; this accessor is
    /// `#[cfg(test)]`-only so future code cannot accidentally reach for
    /// the racy pair.
    #[cfg(test)]
    pub(super) fn reboot_flag(&self) -> RebootFlag {
        if (self.session_state.load(Ordering::Acquire) & WRAPPED_BIT) != 0 {
            RebootFlag::Continuous
        } else {
            RebootFlag::RecentlyRebooted
        }
    }

    /// Send a multicast `OfferService` announcement for the given config.
    ///
    /// `buf` is a caller-provided scratch buffer used for encoding the
    /// outgoing frame. Returns [`Error::Capacity`]`("udp_buffer")` if the
    /// encoded frame does not fit in `buf`.
    pub(super) async fn send_offer_service<T: TransportSocket>(
        &self,
        buf: &mut [u8],
        config: &ServerConfig,
        socket: &T,
    ) -> Result<(), Error> {
        use crate::protocol::Header as SomeIpHeader;
        use crate::traits::WireFormat;

        let entry = Entry::OfferService(ServiceEntry {
            index_first_options_run: 0,
            index_second_options_run: 0,
            options_count: OptionsCount::new(1, 0),
            service_id: config.service_id,
            instance_id: config.instance_id,
            major_version: config.major_version,
            ttl: config.ttl,
            minor_version: config.minor_version,
        });

        let option = sd::Options::IpV4Endpoint {
            ip: config.interface,
            port: config.local_port,
            protocol: TransportProtocol::Udp,
        };

        let entries = [entry];
        let options = [option];
        // Atomic (sid, reboot_flag) pair so that concurrent emissions
        // around the wrap boundary cannot disagree about whether this
        // very message advertises `RecentlyRebooted` or `Continuous`.
        // See `next_session_id_with_reboot_flag` docs for the race.
        let (sid, reboot_flag) = self.next_session_id_with_reboot_flag();
        let sd_payload = sd::Header::new(Flags::new_sd(reboot_flag), &entries, &options);

        // Caller-provided send scratch — keeps the per-tick path
        // alloc-free without parking a `[u8; UDP_BUFFER_SIZE]` in the
        // announce future. 16-byte SOME/IP header + the SD payload.
        if buf.len() < 16 {
            return Err(Error::Capacity("udp_buffer"));
        }
        let sd_data_len = sd_payload
            .encode_to_slice(&mut buf[16..])
            .map_err(|_| Error::Capacity("udp_buffer"))?;
        let total_len = 16 + sd_data_len;
        // The `< 16` guard plus `encode_to_slice`'s own over-capacity
        // error already cover the fit; this stays as a debug-only
        // sanity check rather than a live branch.
        debug_assert!(total_len <= buf.len());
        let someip_header = SomeIpHeader::new_sd(sid, sd_data_len);
        someip_header
            .encode_to_slice(&mut buf[..16])
            .map_err(|_| Error::Capacity("udp_buffer"))?;

        let multicast_addr = SocketAddrV4::new(sd::MULTICAST_IP, sd::MULTICAST_PORT);

        crate::log::trace!(
            "Sending OfferService: service=0x{:04X}, instance={}, port={}, size={} bytes",
            config.service_id,
            config.instance_id,
            config.local_port,
            total_len
        );
        crate::log::trace!("OfferService data: {:02X?}", &buf[..total_len.min(64)]);

        socket.send_to(&buf[..total_len], multicast_addr).await?;
        crate::log::trace!("Sent to {}", multicast_addr);

        Ok(())
    }
}

// `SdStateHandle` / `WrappableSdStateHandle` were collapsed into the
// unified `crate::transport::SharedHandle<SdStateManager>` /
// `WrappableSharedHandle<SdStateManager>` traits. The blanket impls
// there cover both `&'static SdStateManager` and
// `Arc<SdStateManager>`; no dedicated trait survives here.

#[cfg(all(test, feature = "server-tokio"))]
mod tests {
    use super::{Error, SdStateManager, ServerConfig};
    use crate::protocol::sd::{self, EntryType, Flags, RebootFlag, TransportProtocol};
    use crate::protocol::{MessageType, MessageView, ReturnCode};
    use crate::tokio_transport::TokioSocket;
    use crate::transport::{SocketOptions, TransportFactory};
    use std::net::{IpAddr, Ipv4Addr, SocketAddr, SocketAddrV4};
    use std::time::Duration;
    use tokio::net::UdpSocket;

    /// Test-only `service_id` for `send_offer_service` tests. Distinct from
    /// the 0x5B / 0x5C values used elsewhere in this crate so that parallel
    /// tests joined to the same SD multicast group do not produce false
    /// matches. If you add a new test that emits a multicast `OfferService`,
    /// give it its own dedicated `service_id` too.
    const TEST_SERVICE_ID: u16 = 0xFE01;
    const TEST_INSTANCE_ID: u16 = 0x42;
    /// Port value placed in the emitted `IpV4Endpoint` option so the
    /// round-trip assertion has something non-zero to check. The test does
    /// not bind this port — it only appears in the announcement payload.
    const TEST_ADVERTISED_PORT: u16 = 40210;

    #[test]
    fn next_session_id_wraps_past_ffff_skipping_zero() {
        let sd = SdStateManager::with_initial(0xFFFE);

        // 0xFFFE -> 0xFFFF
        assert_eq!(sd.next_session_id(), 0xFFFF);

        // 0xFFFF -> wraps to 0x0001 (0 is skipped)
        assert_eq!(sd.next_session_id(), 0x0001);
    }

    #[test]
    fn next_session_id_starts_at_two_from_default_new() {
        let sd = SdStateManager::new();
        // new() seeds at 1; first next_session_id increments to 2
        assert_eq!(sd.next_session_id(), 2);
    }

    /// Concurrent emitters around the wrap boundary must never produce
    /// a `(session_id, reboot_flag)` pair where one is pre-wrap and the
    /// other is post-wrap. Regression for the two-atomic TOCTOU race.
    ///
    /// We seed near the wrap and have many threads call
    /// `next_session_id_with_reboot_flag` concurrently. Every
    /// `(0xFFFF, _)` must carry `RecentlyRebooted`, every `(0x0001, _)`
    /// (the wrap message) and beyond must carry `Continuous`.
    #[test]
    fn next_session_id_with_reboot_flag_never_mismatches_around_wrap() {
        use std::sync::Arc;
        for _trial in 0..20 {
            let sd = Arc::new(SdStateManager::with_initial(0xFFF0));
            let mut handles = std::vec::Vec::new();
            for _ in 0..32 {
                let s = Arc::clone(&sd);
                handles.push(std::thread::spawn(move || {
                    let (sid, flag) = s.next_session_id_with_reboot_flag();
                    (sid, flag)
                }));
            }
            for h in handles {
                let (sid, flag) = h.join().unwrap();
                // sid is u32 in 1..=0xFFFF (never 0).
                assert!((1..=0xFFFF).contains(&sid), "sid out of range: {sid:#x}");
                if sid == 0xFFFF {
                    // The 0xFFFF emission is the LAST pre-wrap.
                    assert_eq!(
                        flag,
                        RebootFlag::RecentlyRebooted,
                        "sid=0xFFFF must carry RecentlyRebooted"
                    );
                } else if sid <= 0xFFEF {
                    // We seeded at 0xFFF0, so any sid in 1..=0xFFEF
                    // means the counter wrapped past 0xFFFF. Must be
                    // Continuous.
                    assert_eq!(
                        flag,
                        RebootFlag::Continuous,
                        "post-wrap sid={sid:#x} must carry Continuous"
                    );
                }
                // sids in 0xFFF0..=0xFFFE are the pre-wrap window —
                // both flags are valid depending on whether this trial
                // wrapped before/after the emission. Don't assert.
            }
        }
    }

    // ── Reboot-flag tracking ────────────────────────────────────────────
    //
    // AUTOSAR SOME/IP-SD: the reboot bit on emitted SD messages must be
    // set until the session counter wraps past `0xFFFF` for the first
    // time, then cleared permanently. These tests drive `SdStateManager`
    // directly (no socket) and verify the state machine that every
    // server-side SD emission path (`send_offer_service`, plus unicast
    // offer / `SubscribeAck` / `SubscribeNack` in `server::Server`) now
    // reads from via [`SdStateManager::reboot_flag`].

    #[test]
    fn reboot_flag_is_recently_rebooted_on_fresh_manager() {
        // Default constructor: counter hasn't wrapped, flag must indicate
        // a recent reboot so peers can re-synchronize SD state.
        let sd = SdStateManager::new();
        assert_eq!(sd.reboot_flag(), RebootFlag::RecentlyRebooted);
    }

    #[test]
    fn reboot_flag_stays_recently_rebooted_below_wrap() {
        // Advancing the counter short of a wrap must not flip the flag —
        // it's specifically the 0xFFFF → 0x0001 transition that matters,
        // not "has next_session_id been called more than once".
        let sd = SdStateManager::with_initial(0x1233);
        for _ in 0..10 {
            sd.next_session_id();
        }
        assert_eq!(sd.reboot_flag(), RebootFlag::RecentlyRebooted);
    }

    #[test]
    fn reboot_flag_flips_to_continuous_exactly_on_wrap() {
        // Step the counter across the wrap boundary and assert the flag
        // transitions on the precise call that crosses 0xFFFF → 0x0001.
        let sd = SdStateManager::with_initial(0xFFFE);
        assert_eq!(sd.reboot_flag(), RebootFlag::RecentlyRebooted);

        // 0xFFFE -> 0xFFFF: prev=0xFFFE, no wrap.
        assert_eq!(sd.next_session_id(), 0xFFFF);
        assert_eq!(
            sd.reboot_flag(),
            RebootFlag::RecentlyRebooted,
            "counter reached 0xFFFF but has not yet wrapped — flag must still be RecentlyRebooted",
        );

        // 0xFFFF -> 0x0001 (skip 0): prev=0xFFFF, wrap latches.
        assert_eq!(sd.next_session_id(), 0x0001);
        assert_eq!(
            sd.reboot_flag(),
            RebootFlag::Continuous,
            "wrap just occurred — flag must now be Continuous",
        );
    }

    #[test]
    fn reboot_flag_is_monotonic_after_wrap() {
        // Once the flag latches to Continuous it never goes back, even
        // after the counter wraps a second time or is advanced
        // indefinitely. Guard against a regression that would re-derive
        // the flag from the current counter value (which would wrongly
        // flip back to RecentlyRebooted at 0x0001).
        let sd = SdStateManager::with_initial(0xFFFE);
        sd.next_session_id(); // -> 0xFFFF
        sd.next_session_id(); // wrap -> 0x0001
        assert_eq!(sd.reboot_flag(), RebootFlag::Continuous);

        // Many further advances, including crossing 0xFFFF again.
        for _ in 0..(u32::from(u16::MAX) + 5) {
            sd.next_session_id();
        }
        assert_eq!(sd.reboot_flag(), RebootFlag::Continuous);
    }

    // ── Mock-socket tests (default `cargo test` runs these) ────────────
    //
    // These exercise `send_offer_service`'s encoding + framing path
    // through a mock `TransportSocket` that captures the bytes the
    // loop would have emitted. They run in default `cargo test` (no
    // MULTICAST flag required) and provide the primary coverage for
    // the encoding lines.
    //
    // The `#[ignore]`d multicast tests further down test the SAME
    // properties (session-id advancement, wrap, TTL=0 round-trip)
    // through a real kernel-loopback multicast socket. Those tests
    // remain because they additionally exercise the
    // `socket.send_to(multicast_addr, ...)` kernel path, which the
    // mock can't observe. Don't delete them in favour of the mocks
    // — the kernel-multicast verification is the only signal we
    // have for "the wire form actually leaves the OS correctly."

    use crate::transport::{IoErrorKind, ReceivedDatagram, TransportError, TransportSocket};
    use std::sync::Mutex;
    use std::vec::Vec;

    /// `TransportSocket` impl that captures every `send_to` call
    /// instead of touching a real network. `recv_from` and the
    /// socket-level queries are stubbed because `send_offer_service`
    /// never touches them.
    struct CapturingSocket {
        sent: Mutex<Vec<(SocketAddrV4, Vec<u8>)>>,
    }

    impl CapturingSocket {
        fn new() -> Self {
            Self {
                sent: Mutex::new(Vec::new()),
            }
        }

        fn drain_sent(&self) -> Vec<(SocketAddrV4, Vec<u8>)> {
            std::mem::take(&mut *self.sent.lock().unwrap())
        }
    }

    impl TransportSocket for CapturingSocket {
        type SendFuture<'a> = core::future::Ready<Result<(), TransportError>>;
        type RecvFuture<'a> = core::future::Pending<Result<ReceivedDatagram, TransportError>>;

        fn send_to<'a>(&'a self, buf: &'a [u8], target: SocketAddrV4) -> Self::SendFuture<'a> {
            self.sent.lock().unwrap().push((target, buf.to_vec()));
            core::future::ready(Ok(()))
        }

        fn recv_from<'a>(&'a self, _buf: &'a mut [u8]) -> Self::RecvFuture<'a> {
            core::future::pending()
        }

        fn local_addr(&self) -> Result<SocketAddrV4, TransportError> {
            Err(TransportError::Io(IoErrorKind::Other))
        }

        fn join_multicast_v4(
            &self,
            _group: Ipv4Addr,
            _iface: Ipv4Addr,
        ) -> Result<(), TransportError> {
            Ok(())
        }

        fn leave_multicast_v4(
            &self,
            _group: Ipv4Addr,
            _iface: Ipv4Addr,
        ) -> Result<(), TransportError> {
            Ok(())
        }
    }

    /// `TransportSocket` that fails on `send_to` so we can test
    /// `send_offer_service`'s error propagation path.
    struct FailingSocket;

    impl TransportSocket for FailingSocket {
        type SendFuture<'a> = core::future::Ready<Result<(), TransportError>>;
        type RecvFuture<'a> = core::future::Pending<Result<ReceivedDatagram, TransportError>>;

        fn send_to<'a>(&'a self, _buf: &'a [u8], _target: SocketAddrV4) -> Self::SendFuture<'a> {
            core::future::ready(Err(TransportError::Io(IoErrorKind::NetworkUnreachable)))
        }

        fn recv_from<'a>(&'a self, _buf: &'a mut [u8]) -> Self::RecvFuture<'a> {
            core::future::pending()
        }

        fn local_addr(&self) -> Result<SocketAddrV4, TransportError> {
            Err(TransportError::Io(IoErrorKind::Other))
        }

        fn join_multicast_v4(
            &self,
            _group: Ipv4Addr,
            _iface: Ipv4Addr,
        ) -> Result<(), TransportError> {
            Ok(())
        }

        fn leave_multicast_v4(
            &self,
            _group: Ipv4Addr,
            _iface: Ipv4Addr,
        ) -> Result<(), TransportError> {
            Ok(())
        }
    }

    /// Assert that the captured datagram is byte-for-byte the
    /// `OfferService` we expected — SOME/IP envelope, SD entry, and
    /// the IPv4 endpoint option.
    fn assert_captured_offer_matches(
        bytes: &[u8],
        target: SocketAddrV4,
        config: &ServerConfig,
        expected_session_id: u32,
        expected_reboot: RebootFlag,
    ) {
        // Goes to the SD multicast group on port 30490.
        assert_eq!(
            target,
            SocketAddrV4::new(sd::MULTICAST_IP, sd::MULTICAST_PORT),
        );
        let view = MessageView::parse(bytes).expect("parses as SOME/IP");
        // SD envelope. message_id = (0xFFFF, 0x8100), notification, ok.
        assert_eq!(view.header().message_id().service_id(), 0xFFFF);
        assert_eq!(view.header().message_id().method_id(), 0x8100);
        assert_eq!(view.header().request_id(), expected_session_id);
        assert_eq!(
            view.header().message_type().message_type(),
            MessageType::Notification,
        );
        assert_eq!(view.header().return_code(), ReturnCode::Ok);
        // SD body.
        let sd_view = view.sd_header().expect("sd header parses");
        assert_eq!(sd_view.flags().reboot(), expected_reboot);
        assert!(sd_view.flags().unicast(), "unicast flag must be set");
        // Exactly one OfferService entry, exactly one IPv4 endpoint.
        assert_eq!(sd_view.entries().count(), 1);
        assert_eq!(sd_view.options().count(), 1);
        let entry = sd_view.entries().next().unwrap();
        assert!(matches!(entry.entry_type(), Ok(EntryType::OfferService),));
        assert_eq!(entry.service_id(), config.service_id);
        assert_eq!(entry.instance_id(), config.instance_id);
        assert_eq!(entry.major_version(), config.major_version);
        assert_eq!(entry.ttl(), config.ttl);
        assert_eq!(entry.minor_version(), config.minor_version);
        let opts_count = entry.options_count();
        assert_eq!(opts_count.first_options_count, 1);
        assert_eq!(opts_count.second_options_count, 0);
        let option = sd_view.options().next().unwrap();
        let (ip, protocol, port) = option.as_ipv4().expect("ipv4 endpoint option");
        assert_eq!(ip, config.interface);
        assert_eq!(port, config.local_port);
        assert_eq!(protocol, TransportProtocol::Udp);
    }

    #[tokio::test]
    async fn send_offer_service_through_mock_emits_full_someip_sd_envelope() {
        let config = ServerConfig::new(TEST_SERVICE_ID, TEST_INSTANCE_ID)
            .with_interface(Ipv4Addr::LOCALHOST)
            .with_local_port(TEST_ADVERTISED_PORT);
        let sd_state = SdStateManager::with_initial(0x1233);
        let sock = CapturingSocket::new();

        sd_state
            .send_offer_service(&mut [0u8; crate::UDP_BUFFER_SIZE], &config, &sock)
            .await
            .expect("send_offer_service should succeed against the mock");

        let captured = sock.drain_sent();
        assert_eq!(captured.len(), 1, "expected exactly one send_to call");
        let (target, bytes) = &captured[0];
        // Initial counter 0x1233 → first emission carries 0x0000_1234,
        // and the manager has not wrapped, so reboot=RecentlyRebooted.
        assert_captured_offer_matches(
            bytes,
            *target,
            &config,
            0x0000_1234,
            RebootFlag::RecentlyRebooted,
        );
    }

    #[tokio::test]
    async fn send_offer_service_through_mock_advances_session_id_across_calls() {
        let config = ServerConfig::new(TEST_SERVICE_ID, TEST_INSTANCE_ID)
            .with_interface(Ipv4Addr::LOCALHOST)
            .with_local_port(TEST_ADVERTISED_PORT);
        let sd_state = SdStateManager::with_initial(0x1233);
        let sock = CapturingSocket::new();

        sd_state
            .send_offer_service(&mut [0u8; crate::UDP_BUFFER_SIZE], &config, &sock)
            .await
            .unwrap();
        sd_state
            .send_offer_service(&mut [0u8; crate::UDP_BUFFER_SIZE], &config, &sock)
            .await
            .unwrap();

        let sent = sock.drain_sent();
        assert_eq!(sent.len(), 2);
        let v0 = MessageView::parse(&sent[0].1).unwrap();
        let v1 = MessageView::parse(&sent[1].1).unwrap();
        assert_eq!(v0.header().request_id(), 0x0000_1234);
        assert_eq!(v1.header().request_id(), 0x0000_1235);
    }

    #[tokio::test]
    async fn send_offer_service_through_mock_reboot_flag_flips_on_wrap() {
        let config = ServerConfig::new(TEST_SERVICE_ID, TEST_INSTANCE_ID)
            .with_interface(Ipv4Addr::LOCALHOST)
            .with_local_port(TEST_ADVERTISED_PORT);
        // Seed so the FIRST send takes 0xFFFE → 0xFFFF (still
        // RecentlyRebooted) and the SECOND sees the wrap to 0x0001
        // (Continuous).
        let sd_state = SdStateManager::with_initial(0xFFFE);
        let sock = CapturingSocket::new();
        sd_state
            .send_offer_service(&mut [0u8; crate::UDP_BUFFER_SIZE], &config, &sock)
            .await
            .unwrap();
        sd_state
            .send_offer_service(&mut [0u8; crate::UDP_BUFFER_SIZE], &config, &sock)
            .await
            .unwrap();

        let sent = sock.drain_sent();
        assert_eq!(sent.len(), 2);
        let v0 = MessageView::parse(&sent[0].1).unwrap();
        let v1 = MessageView::parse(&sent[1].1).unwrap();
        assert_eq!(v0.header().request_id(), 0x0000_FFFF);
        assert_eq!(
            v1.header().request_id(),
            0x0000_0001,
            "must skip reserved 0 on wrap",
        );
        assert_eq!(
            v0.sd_header().unwrap().flags().reboot(),
            RebootFlag::RecentlyRebooted,
            "first emit is pre-wrap",
        );
        assert_eq!(
            v1.sd_header().unwrap().flags().reboot(),
            RebootFlag::Continuous,
            "second emit is post-wrap",
        );
    }

    #[tokio::test]
    async fn send_offer_service_through_mock_preserves_zero_ttl() {
        let mut config = ServerConfig::new(TEST_SERVICE_ID, TEST_INSTANCE_ID)
            .with_interface(Ipv4Addr::LOCALHOST)
            .with_local_port(TEST_ADVERTISED_PORT);
        config.ttl = 0;
        let sd_state = SdStateManager::with_initial(0x1233);
        let sock = CapturingSocket::new();
        sd_state
            .send_offer_service(&mut [0u8; crate::UDP_BUFFER_SIZE], &config, &sock)
            .await
            .unwrap();

        let sent = sock.drain_sent();
        let view = MessageView::parse(&sent[0].1).unwrap();
        let entry = view.sd_header().unwrap().entries().next().unwrap();
        assert_eq!(entry.ttl(), 0, "TTL=0 must round-trip end-to-end");
    }

    #[tokio::test]
    async fn send_offer_service_through_mock_propagates_socket_errors() {
        let config = ServerConfig::new(TEST_SERVICE_ID, TEST_INSTANCE_ID)
            .with_interface(Ipv4Addr::LOCALHOST)
            .with_local_port(TEST_ADVERTISED_PORT);
        let sd_state = SdStateManager::with_initial(0x1233);
        let sock = FailingSocket;
        let result = sd_state
            .send_offer_service(&mut [0u8; crate::UDP_BUFFER_SIZE], &config, &sock)
            .await;
        // Narrow assertion: the error must specifically be the
        // `Io(NetworkUnreachable)` propagated from `FailingSocket::send_to`.
        // `Err(_)` would also pass on unrelated regressions (encoding
        // failures, internal panics) and falsely attribute them to
        // socket-error propagation.
        match result {
            Err(Error::Transport(TransportError::Io(IoErrorKind::NetworkUnreachable))) => {}
            other => panic!(
                "expected Err(Transport(Io(NetworkUnreachable))) propagated from \
                 FailingSocket::send_to; got {other:?}",
            ),
        }
    }

    // ── Multicast-loopback harness ──────────────────────────────────────
    //
    // All tests below drive `send_offer_service` against a real UDP socket
    // and read the emitted packet off a second socket joined to the SD
    // multicast group. These are `#[ignore]`d on environments whose
    // loopback interface does not carry the `MULTICAST` flag (check with
    // `ip link show lo`); on such hosts the kernel drops multicast on
    // `lo` before loopback reflection, so the receiver times out. Runs
    // in any environment where loopback multicast is available.

    /// Bind a receiver socket on the SD multicast port, ready to
    /// `join_multicast_v4`.
    fn build_mcast_receiver(interface: Ipv4Addr) -> std::io::Result<UdpSocket> {
        let raw = socket2::Socket::new(
            socket2::Domain::IPV4,
            socket2::Type::DGRAM,
            Some(socket2::Protocol::UDP),
        )?;
        raw.set_reuse_address(true)?;
        #[cfg(unix)]
        raw.set_reuse_port(true)?;
        raw.set_multicast_loop_v4(true)?;
        raw.bind(&SocketAddr::new(IpAddr::V4(interface), sd::MULTICAST_PORT).into())?;
        raw.set_nonblocking(true)?;
        UdpSocket::from_std(raw.into())
    }

    /// Bind a sender [`TokioSocket`] on an ephemeral port with
    /// `multicast_if` pinned to the loopback interface so emitted
    /// packets loop back to any receiver joined to the same group on
    /// that interface. Uses the [`TransportFactory`] surface so the
    /// resulting socket implements [`crate::transport::TransportSocket`]
    /// — which is what the now-generic
    /// [`SdStateManager::send_offer_service`] requires.
    async fn build_mcast_sender(
        interface: Ipv4Addr,
    ) -> Result<TokioSocket, crate::transport::TransportError> {
        let mut opts = SocketOptions::new();
        opts.reuse_address = true;
        opts.reuse_port = true;
        opts.multicast_if_v4 = Some(interface);
        opts.multicast_loop_v4 = Some(true);
        crate::tokio_transport::TokioTransport
            .bind(SocketAddrV4::new(interface, 0), &opts)
            .await
    }

    /// Fields extracted from a received SOME/IP-SD `OfferService` packet.
    /// Keeping these together makes per-test assertions a straight list of
    /// `assert_eq!`s against expected values.
    struct ReceivedOffer {
        request_id: u32,
        someip_service_id: u16,
        someip_method_id: u16,
        message_type: MessageType,
        return_code: ReturnCode,
        protocol_version: u8,
        interface_version: u8,
        flags: Flags,
        entry_service_id: u16,
        entry_instance_id: u16,
        entry_major_version: u8,
        entry_minor_version: u32,
        entry_ttl: u32,
        endpoint_ip: Ipv4Addr,
        endpoint_port: u16,
        endpoint_protocol: TransportProtocol,
    }

    /// Wait for a multicast `OfferService` matching `expected_service_id`,
    /// returning its decoded fields. Other packets on the group (from
    /// concurrent tests) are ignored; a single outer timeout bounds the
    /// whole filter loop.
    async fn recv_our_offer(
        rx: &UdpSocket,
        expected_service_id: u16,
        within: Duration,
    ) -> ReceivedOffer {
        let recv_loop = async {
            let mut buf = [0u8; 2048];
            loop {
                let (len, _from) = rx
                    .recv_from(&mut buf)
                    .await
                    .expect("recv_from should succeed");
                let Ok(view) = MessageView::parse(&buf[..len]) else {
                    continue;
                };
                if view.header().message_id().service_id() != 0xFFFF {
                    continue;
                }
                let Ok(sd_view) = view.sd_header() else {
                    continue;
                };
                let Some(entry) = sd_view.entries().next() else {
                    continue;
                };
                if !matches!(entry.entry_type(), Ok(EntryType::OfferService)) {
                    continue;
                }
                if entry.service_id() != expected_service_id {
                    continue;
                }
                let first_option = sd_view
                    .options()
                    .next()
                    .expect("OfferService should carry an endpoint option");
                let (endpoint_ip, endpoint_protocol, endpoint_port) = first_option
                    .as_ipv4()
                    .expect("endpoint option should decode as IPv4");
                return ReceivedOffer {
                    request_id: view.header().request_id(),
                    someip_service_id: view.header().message_id().service_id(),
                    someip_method_id: view.header().message_id().method_id(),
                    message_type: view.header().message_type().message_type(),
                    return_code: view.header().return_code(),
                    protocol_version: view.header().protocol_version(),
                    interface_version: view.header().interface_version(),
                    flags: sd_view.flags(),
                    entry_service_id: entry.service_id(),
                    entry_instance_id: entry.instance_id(),
                    entry_major_version: entry.major_version(),
                    entry_minor_version: entry.minor_version(),
                    entry_ttl: entry.ttl(),
                    endpoint_ip,
                    endpoint_port,
                    endpoint_protocol,
                };
            }
        };
        tokio::time::timeout(within, recv_loop)
            .await
            .expect("timed out waiting for our OfferService")
    }

    /// Assert every field of the SOME/IP + SD envelope that
    /// `send_offer_service` is responsible for — not just the entry body.
    /// A future regression that garbles the endpoint option, flips a flag,
    /// or changes the SOME/IP message type should fail here.
    ///
    /// `expected_reboot` lets pre-wrap callers assert `RecentlyRebooted`
    /// and post-wrap callers assert `Continuous`; the flag is tracked by
    /// `SdStateManager::has_wrapped` and read via `reboot_flag()` at each
    /// send.
    fn assert_offer_matches(
        offer: &ReceivedOffer,
        config: &ServerConfig,
        expected_request_id: u32,
        expected_reboot: RebootFlag,
    ) {
        // SOME/IP envelope
        assert_eq!(offer.someip_service_id, 0xFFFF, "SD uses service_id 0xFFFF");
        assert_eq!(offer.someip_method_id, 0x8100, "SD uses method_id 0x8100");
        assert_eq!(offer.message_type, MessageType::Notification);
        assert_eq!(offer.return_code, ReturnCode::Ok);
        assert_eq!(offer.protocol_version, 0x01);
        assert_eq!(offer.interface_version, 0x01);
        assert_eq!(
            offer.request_id, expected_request_id,
            "request_id is session_id in low 16 bits, client_id zero in high 16",
        );
        // SD flags — reboot comes from SdStateManager::reboot_flag (latches
        // to Continuous after the session counter wraps past 0xFFFF);
        // unicast is always true for SD.
        assert_eq!(offer.flags.reboot(), expected_reboot);
        assert!(offer.flags.unicast());
        // OfferService entry
        assert_eq!(offer.entry_service_id, config.service_id);
        assert_eq!(offer.entry_instance_id, config.instance_id);
        assert_eq!(offer.entry_major_version, config.major_version);
        assert_eq!(offer.entry_minor_version, config.minor_version);
        assert_eq!(offer.entry_ttl, config.ttl);
        // Endpoint option
        assert_eq!(offer.endpoint_ip, config.interface);
        assert_eq!(offer.endpoint_port, config.local_port);
        assert_eq!(offer.endpoint_protocol, TransportProtocol::Udp);
    }

    /// Standard loopback receiver/sender pair used by the send-path tests.
    async fn mcast_rx_tx() -> (UdpSocket, TokioSocket) {
        let interface = Ipv4Addr::LOCALHOST;
        let rx = build_mcast_receiver(interface).expect("bind receiver");
        rx.join_multicast_v4(sd::MULTICAST_IP, interface)
            .expect("join SD multicast group");
        let tx = build_mcast_sender(interface).await.expect("bind sender");
        (rx, tx)
    }

    #[ignore = "requires MULTICAST on loopback; skipped on hosts whose `lo` \
                lacks the MULTICAST flag. Runs in any environment where \
                loopback multicast is available."]
    #[tokio::test]
    async fn send_offer_service_emits_parseable_offer_to_multicast() {
        let config = ServerConfig::new(TEST_SERVICE_ID, TEST_INSTANCE_ID)
            .with_interface(Ipv4Addr::LOCALHOST)
            .with_local_port(TEST_ADVERTISED_PORT);
        let (rx, tx) = mcast_rx_tx().await;

        // Seed with a recognisable value so on-wire session_id is exact.
        let sd_state = SdStateManager::with_initial(0x1233);
        sd_state
            .send_offer_service(&mut [0u8; crate::UDP_BUFFER_SIZE], &config, &tx)
            .await
            .expect("send_offer_service should succeed on a configured socket");

        let offer = recv_our_offer(&rx, config.service_id, Duration::from_secs(2)).await;
        // next_session_id advances 0x1233 -> 0x1234; client_id is zero.
        // Fresh SdStateManager: counter has not wrapped, reboot flag is
        // RecentlyRebooted.
        assert_offer_matches(&offer, &config, 0x0000_1234, RebootFlag::RecentlyRebooted);
    }

    #[ignore = "requires MULTICAST on loopback; skipped on hosts whose `lo` \
                lacks the MULTICAST flag. Runs in any environment where \
                loopback multicast is available."]
    #[tokio::test]
    async fn send_offer_service_advances_session_id_across_calls() {
        // Back-to-back sends must consume distinct, incrementing session
        // IDs — catches a regression where `send_offer_service` reads the
        // counter without advancing it, or reuses a cached value.
        let config = ServerConfig::new(TEST_SERVICE_ID, TEST_INSTANCE_ID)
            .with_interface(Ipv4Addr::LOCALHOST)
            .with_local_port(TEST_ADVERTISED_PORT);
        let (rx, tx) = mcast_rx_tx().await;

        let sd_state = SdStateManager::with_initial(0x1233);
        sd_state
            .send_offer_service(&mut [0u8; crate::UDP_BUFFER_SIZE], &config, &tx)
            .await
            .unwrap();
        sd_state
            .send_offer_service(&mut [0u8; crate::UDP_BUFFER_SIZE], &config, &tx)
            .await
            .unwrap();

        let first = recv_our_offer(&rx, config.service_id, Duration::from_secs(2)).await;
        let second = recv_our_offer(&rx, config.service_id, Duration::from_secs(2)).await;
        assert_eq!(first.request_id, 0x0000_1234);
        assert_eq!(second.request_id, 0x0000_1235);
    }

    #[ignore = "requires MULTICAST on loopback; skipped on hosts whose `lo` \
                lacks the MULTICAST flag. Runs in any environment where \
                loopback multicast is available."]
    #[tokio::test]
    async fn send_offer_service_wraps_session_id_through_zero_on_send() {
        // Session counter wrap must be visible on the wire: 0xFFFE -> 0xFFFF
        // -> 0x0001 (skipping the reserved 0). Exercises the wrap branch
        // *through* the send path, not only the unit test of next_session_id.
        let config = ServerConfig::new(TEST_SERVICE_ID, TEST_INSTANCE_ID)
            .with_interface(Ipv4Addr::LOCALHOST)
            .with_local_port(TEST_ADVERTISED_PORT);
        let (rx, tx) = mcast_rx_tx().await;

        let sd_state = SdStateManager::with_initial(0xFFFE);
        sd_state
            .send_offer_service(&mut [0u8; crate::UDP_BUFFER_SIZE], &config, &tx)
            .await
            .unwrap();
        sd_state
            .send_offer_service(&mut [0u8; crate::UDP_BUFFER_SIZE], &config, &tx)
            .await
            .unwrap();

        let first = recv_our_offer(&rx, config.service_id, Duration::from_secs(2)).await;
        let second = recv_our_offer(&rx, config.service_id, Duration::from_secs(2)).await;
        assert_eq!(first.request_id, 0x0000_FFFF);
        assert_eq!(
            second.request_id, 0x0000_0001,
            "must skip reserved 0 on wrap"
        );
        // Reboot flag latches: the first emission goes out before the
        // wrap happens (prev=0xFFFE), so it still advertises
        // RecentlyRebooted; the second emission is the one whose
        // next_session_id call crossed 0xFFFF -> 0x0001, so the flag
        // Flips to Continuous permanently from there on.
        assert_eq!(
            first.flags.reboot(),
            RebootFlag::RecentlyRebooted,
            "first emit is pre-wrap and must still advertise RecentlyRebooted",
        );
        assert_eq!(
            second.flags.reboot(),
            RebootFlag::Continuous,
            "post-wrap emit must advertise Continuous",
        );
    }

    #[ignore = "requires MULTICAST on loopback; skipped on hosts whose `lo` \
                lacks the MULTICAST flag. Runs in any environment where \
                loopback multicast is available."]
    #[tokio::test]
    async fn send_offer_service_preserves_zero_ttl() {
        // TTL=0 is a legitimate SOME/IP-SD value meaning "stop offering";
        // `send_offer_service` must preserve it end-to-end rather than,
        // say, defaulting it back to the ServerConfig::new value of 3.
        let mut config = ServerConfig::new(TEST_SERVICE_ID, TEST_INSTANCE_ID)
            .with_interface(Ipv4Addr::LOCALHOST)
            .with_local_port(TEST_ADVERTISED_PORT);
        config.ttl = 0;
        let (rx, tx) = mcast_rx_tx().await;

        let sd_state = SdStateManager::with_initial(0x1233);
        sd_state
            .send_offer_service(&mut [0u8; crate::UDP_BUFFER_SIZE], &config, &tx)
            .await
            .unwrap();

        let offer = recv_our_offer(&rx, config.service_id, Duration::from_secs(2)).await;
        assert_offer_matches(&offer, &config, 0x0000_1234, RebootFlag::RecentlyRebooted);
        // Belt-and-suspenders: assert_offer_matches already checks this,
        // but the purpose of this test is specifically the zero case.
        assert_eq!(offer.entry_ttl, 0);
    }
}