diff --git a/CHANGELOG.md b/CHANGELOG.md index 3572486a..8235b507 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,33 @@ ## [0.8.0] +### Breaking — bare-metal server buffer extraction (PR #125 / PR 3) + +These changes are free before the 0.8.0 release — no published version +exposes the pre-refactor API. + +- **`Server::run_with_buffers` takes two additional send-scratch buffers** — + the signature now requires `recv_send_buf: &mut [u8]` and + `announce_send_buf: &mut [u8]` after the existing `unicast_buf` / `sd_buf` + receive buffers. Callers that used the pre-refactor four-buffer form must + add two more `static [u8; N]` arguments (or equivalent heap slices). The + `_alloc` convenience `Server::run` is unchanged. + +- **New `Server::announce_only_with_buffer`** — bare-metal supplementary + servers that need *only* the SD `OfferService` announcement loop (no recv) + now call this method instead of `announce_only_future`. It accepts a + caller-owned `&mut [u8]` scratch so the future does NOT park a + `[u8; UDP_BUFFER_SIZE]` (≈ 1500 B) in its own state. + `announce_only_future` (alloc-only) now delegates to this method internally + and carries a `#[cfg(feature = "_alloc")]` gate. + +- **`EventPublisher::publish_event_with_buffers` / + `publish_raw_event_with_buffers` take caller scratch** — the two methods + now accept explicit `msg_buf: &mut [u8]` / `protected_buf: &mut [u8]` + slices. The future no longer parks two `[u8; UDP_BUFFER_SIZE]` arrays; + bare-metal callers supply `static` buffers. The `_alloc` wrappers + `publish_event` / `publish_raw_event` are unchanged. + ### Client/Server API symmetry & ergonomics The 0.8.0 ergonomics pass aligning the public Client and Server surfaces, removing the tokio-only `Server::new` cliff, and improving discoverability for bare-metal adopters. Six bundled changes: generic-parameter alignment, tokio-defaulted `Deps` builders, channel-types rustdoc, `ServerConfig` fluent builder, `Server::new` constructor reshape (the one breaking change in this set), and `SubscriptionHandle` GAT promotion. Migration shapes below are written against the previous published version (0.7.0); `cargo build` will surface every remaining call-site. diff --git a/docs/simple_someip/plans/2026-06-17-pr2-125-client-async-state-reduction.md b/docs/simple_someip/plans/2026-06-17-pr2-125-client-async-state-reduction.md index 3dc80182..c7fc5d25 100644 --- a/docs/simple_someip/plans/2026-06-17-pr2-125-client-async-state-reduction.md +++ b/docs/simple_someip/plans/2026-06-17-pr2-125-client-async-state-reduction.md @@ -289,17 +289,19 @@ use crate::static_channels::{BufferLease, BufferPool}; use crate::transport::BufferProvider; use crate::UDP_BUFFER_SIZE; -/// Tokio-path buffer provider: a single leaked `BufferPool` sized at -/// `UNICAST_SOCKETS_CAP + 1` × `UDP_BUFFER_SIZE` (one per possible socket -/// plus discovery). Leaking is fine — a client process holds one for its -/// lifetime; the API hides it entirely from callers. -#[derive(Clone, Copy, Debug)] -pub struct TokioBufferProvider(&'static BufferPool<9, UDP_BUFFER_SIZE>); +/// Tokio-path buffer provider: a single `Arc`-backed `BufferPool` sized at +/// `UNICAST_SOCKETS_CAP + 1 (discovery) + 1 (release-lag slack)` × +/// `UDP_BUFFER_SIZE`. `Arc`-backed, NOT leaked — the pool is freed when the +/// last provider/lease drops; the API hides it entirely from callers. +/// (Rev: the original sketch used `Box::leak`; the merged implementation is +/// `Arc`-backed so dynamically-created clients don't leak a pool each.) +#[derive(Clone, Debug)] +pub struct TokioBufferProvider(alloc::sync::Arc>); impl TokioBufferProvider { #[must_use] pub fn new() -> Self { - Self(Box::leak(Box::new(BufferPool::new()))) + Self(alloc::sync::Arc::new(BufferPool::new())) } } diff --git a/docs/simple_someip/plans/2026-06-17-pr3-125-server-buffer-extraction.md b/docs/simple_someip/plans/2026-06-17-pr3-125-server-buffer-extraction.md new file mode 100644 index 00000000..0c2d9522 --- /dev/null +++ b/docs/simple_someip/plans/2026-06-17-pr3-125-server-buffer-extraction.md @@ -0,0 +1,325 @@ +# PR 3 — #125 Server Buffer Extraction + Final Numbers Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Move the server's 6 future-resident `[u8; UDP_BUFFER_SIZE]` send-path buffers out of the run/publish futures into caller-provided scratch — matching the server's existing caller-provided *receive* buffer model (`run_with_buffers`) — then capture issue #125's final client+server before/after numbers and close it. + +**Architecture:** The server runs a single combined future (`run_combined` = `recv_loop` + `announce_loop` via `select`) plus an app-driven publish path (`EventPublisher`); it does NOT spawn per-socket loops, so there is no buffer *pool* — buffers are caller-provided fixed scratch (as the receive buffers already are). The SD/Subscribe/offer send helpers and `EventPublisher::publish_*` stop stack-allocating `[u8; UDP_BUFFER_SIZE]` and instead take `&mut [u8]` scratch; the run path is fed two scratch buffers (recv-path + announce-path, which can be mid-send concurrently); the tokio path heap-allocates them internally so the public tokio API is unchanged. + +**Tech Stack:** Rust (edition 2024, crate stays stable-buildable), `heapless`, `embassy-sync`, `tokio` (std path only), cargo, `cargo-nextest`. + +**Spec:** `docs/simple_someip/plans/2026-06-09-phase22-125-memory-reduction-design.md` (PR 3 section — note its `recv_loop`/`announce_loop` *receive* extraction is already done via `run_with_buffers`; the real targets are the send paths below). + +**Base:** stacked on PR 2 (`feature/pr2_125_client_async_state`, tip `6b0aaa3`). + +## Global Constraints + +- **Crate stays stable-buildable**; no nightly-only features. **`server,bare_metal` stays alloc-free** where it is today (the new scratch params must not pull `alloc` into the bare-metal path — tokio-only heap allocation goes behind the `_alloc`/`server-tokio` gate). +- **Wire format untouched.** No change to any byte emitted. +- **Public tokio/std API unchanged** for `Server::new`/`run` and `EventPublisher::publish_*` callers: the tokio path allocates the new scratch internally (mirroring how `run_inner` already heap-allocates the 65535-byte receive buffers). +- **Only behavioral change allowed:** an encode or E2E-protect output that exceeds the *provided scratch length* is rejected with `Error::Capacity("udp_buffer")` (today it's checked against the `UDP_BUFFER_SIZE` constant, which is correct only because the buffers are currently full-size). No other behavioral change. +- **Existing suite green on every commit** (514 nextest `client-tokio,server-tokio`; `bare_metal_e2e` 6/6; no-alloc witness; all clippy + doc gates). +- **Every memory change validated against the `bm_server_run_future` witness** (`tests/bare_metal_e2e.rs`); a change that doesn't move the number is dropped. +- **Exact constants (verbatim):** `UDP_BUFFER_SIZE = 1500` (`src/lib.rs:158`); `BM_SERVER_RUN_FUTURE_BUDGET = 9664` (measured 7696; `tests/bare_metal_e2e.rs:606`). + +--- + +## File Structure + +| File | Action | Responsibility | +|---|---|---| +| `src/server/runtime.rs` | Modify | `send_unicast_offer`/`send_subscribe_ack_from_view`/`send_subscribe_nack_from_view` take `buf: &mut [u8]`; `recv_loop`/`announce_loop`/`run_combined` thread the two send-scratch buffers to them. | +| `src/server/sd_state.rs` | Modify | `SdStateManager::send_offer_service` takes `buf: &mut [u8]`; bounds on `buf.len()`. | +| `src/server/event_publisher.rs` | Modify | `publish_event` takes `msg_buf`/`protected_buf: &mut [u8]`; `publish_raw_event` takes `buf: &mut [u8]`; all encode + E2E bounds on `buf.len()`. | +| `src/server/mod.rs` | Modify | Extend `run_with_buffers` with the two send buffers; `run_inner` (tokio) heap-allocates them; tokio `EventPublisher` wrapper allocates publish scratch internally. | +| `tests/bare_metal_e2e.rs` | Modify | Server send-path regression tests (small-scratch → `Capacity`, not panic) + retighten `BM_SERVER_RUN_FUTURE_BUDGET`. | + +--- + +### Task 1: SD send helpers take caller scratch (`runtime.rs`) + +**Files:** Modify `src/server/runtime.rs` (`send_unicast_offer` ~`:29-78`, `send_subscribe_ack_from_view` ~`:81-129`, `send_subscribe_nack_from_view` ~`:132-182`); Test `tests/bare_metal_e2e.rs`. + +**Interfaces:** +- Produces: each helper gains a leading `buf: &mut [u8]` parameter (replacing its internal `let mut buffer = [0u8; UDP_BUFFER_SIZE]`). + +- [ ] **Step 1: Write the failing test** (a too-small scratch rejects, doesn't panic/OOB) + +```rust +// tests/bare_metal_e2e.rs — drives send_subscribe_ack via the public Subscribe path +// with a deliberately tiny send-scratch buffer; the encoded SD-ACK exceeds it. +#[tokio::test] +async fn server_send_with_undersized_scratch_returns_capacity_not_panic() { + // Harness: build the server with a send-scratch buffer of, say, 24 bytes — + // big enough for the 16-byte header but not the SD-ACK payload — drive a + // Subscribe, and assert the run loop surfaces Capacity("udp_buffer") and + // does NOT panic / OOB. (Model on the existing bare_metal_e2e server harness.) + let outcome = run_server_subscribe_with_send_scratch_len(24).await; + assert!(matches!(outcome, Err(Error::Capacity("udp_buffer")))); +} +``` + +- [ ] **Step 2: Run it — expect FAIL** (helpers don't take a buffer yet / no harness) + +Run: `cargo test --features client,server,bare_metal --test bare_metal_e2e server_send_with_undersized_scratch` +Expected: FAIL (signature mismatch / harness missing). + +- [ ] **Step 3: Change the three helpers to take `buf` and bound on `buf.len()`** + +Pattern (apply to all three — `send_unicast_offer`, `send_subscribe_ack_from_view`, `send_subscribe_nack_from_view`): + +```rust +// was: fn send_subscribe_ack_from_view(... sd_socket, ...) { let mut buffer = [0u8; UDP_BUFFER_SIZE]; ... } +async fn send_subscribe_ack_from_view( + buf: &mut [u8], // ← caller scratch (was the local array) + /* existing params... */ +) -> Result<(), Error> { + if buf.len() < 16 { + return Err(Error::Capacity("udp_buffer")); + } + let sd_data_len = sd_payload.encode_to_slice(&mut buf[16..])?; // encode_to_slice already errors if the slice is too small; map/propagate to Capacity if it surfaces a different error + let someip_header = SomeIpHeader::new_sd(sid, sd_data_len); + someip_header.encode_to_slice(&mut buf[..16])?; + let total_len = 16 + sd_data_len; + if total_len > buf.len() { // defensive: should already be caught by encode_to_slice + return Err(Error::Capacity("udp_buffer")); + } + sd_socket.send_to(&buf[..total_len], subscriber_v4).await?; + Ok(()) +} +``` +Note: `encode_to_slice` into `&mut buf[16..]` already fails on a too-small slice — verify which error it returns and ensure the helper surfaces `Error::Capacity("udp_buffer")` for the over-capacity case (add the explicit `buf.len()` guards above so the contract is the typed capacity error, not a generic encode error). + +- [ ] **Step 4: Run the test — expect PASS**, then the full server suite. + +Run: `cargo test --features client,server,bare_metal --test bare_metal_e2e server_send_with_undersized_scratch` → PASS +Run: `cargo nextest run --no-default-features --features client-tokio,server-tokio` → still green. + +- [ ] **Step 5: Commit** + +```bash +git add src/server/runtime.rs tests/bare_metal_e2e.rs +git commit -m "feat(server): SD send helpers take caller scratch, bound on buf.len() (#125)" +``` + +--- + +### Task 2: `SdStateManager::send_offer_service` takes caller scratch (`sd_state.rs`) + +**Files:** Modify `src/server/sd_state.rs:~219-240`. + +**Interfaces:** Consumes nothing new. Produces: `send_offer_service` gains a `buf: &mut [u8]` parameter. + +- [ ] **Step 1: Change the signature and bound on `buf.len()`** (same pattern as Task 1; this is the `announce_loop`'s send path): + +```rust +// was: pub(crate) async fn send_offer_service(&self, config, socket) { let mut buffer = [0u8; UDP_BUFFER_SIZE]; ... } +pub(crate) async fn send_offer_service( + &self, + buf: &mut [u8], // ← caller scratch + config: &ServerConfig, + socket: &impl TransportSocket, +) -> Result<(), Error> { + if buf.len() < 16 { return Err(Error::Capacity("udp_buffer")); } + let sd_data_len = sd_payload.encode_to_slice(&mut buf[16..])?; + let someip_header = SomeIpHeader::new_sd(sid, sd_data_len); + someip_header.encode_to_slice(&mut buf[..16])?; + let total_len = 16 + sd_data_len; + if total_len > buf.len() { return Err(Error::Capacity("udp_buffer")); } + let multicast_addr = SocketAddrV4::new(sd::MULTICAST_IP, sd::MULTICAST_PORT); + socket.send_to(&buf[..total_len], multicast_addr).await?; + Ok(()) +} +``` + +- [ ] **Step 2: Build (callers updated in Task 3) — verify it compiles in isolation by temporarily building after Task 3 wires the caller.** (This task's deliverable is verified together with Task 3, since `send_offer_service`'s only caller is `announce_loop`.) + +- [ ] **Step 3: Commit** (fold into Task 3's commit if cleaner — `send_offer_service` has no standalone caller). + +```bash +git add src/server/sd_state.rs +git commit -m "feat(server): send_offer_service takes caller scratch, bound on buf.len() (#125)" +``` + +--- + +### Task 3: Thread two send-scratch buffers through the run path (`runtime.rs`, `server/mod.rs`) + +**Files:** Modify `src/server/runtime.rs` (`run_combined` ~`:587-642`, `recv_loop` ~`:435-571`, `announce_loop` ~`:397-430`); `src/server/mod.rs` (`run_with_buffers` ~`:1260-1312`, `run_inner` ~`:1403-1453`). + +**Interfaces:** +- Consumes: the `buf`-taking helpers (Tasks 1–2). +- Produces: `Server::run_with_buffers(unicast_buf, sd_buf, recv_send_buf, announce_send_buf: &mut [u8])` — two new send-scratch params. `recv_loop` and `announce_loop` each receive their send-scratch buffer. + +**Why two:** `run_combined` drives `recv_loop` and `announce_loop` concurrently via `select`; both can be suspended at a `send_to().await` at the same time, so a single shared send buffer would alias. `recv_loop` itself handles one inbound message at a time (one send in flight), so it needs exactly one; `announce_loop` needs exactly one. + +- [ ] **Step 1: Extend `run_with_buffers` + `run_combined` + the loops to pass the buffers down** + +```rust +// server/mod.rs — run_with_buffers gains two params +pub fn run_with_buffers<'a>( + &self, + unicast_buf: &'a mut [u8], + sd_buf: &'a mut [u8], + recv_send_buf: &'a mut [u8], // ← new: recv_loop's send scratch + announce_send_buf: &'a mut [u8], // ← new: announce_loop's send scratch +) -> impl core::future::Future> + 'a + use<'a, F, Tm, R, Sub, H, Hsd, Hep> { /* forward all four into run_combined */ } +``` +```rust +// runtime.rs — run_combined forwards recv_send_buf into recv_loop, announce_send_buf into announce_loop; +// recv_loop passes its buffer to send_unicast_offer / send_subscribe_ack_from_view / send_subscribe_nack_from_view; +// announce_loop passes announce_send_buf to sd_state.send_offer_service. +``` + +- [ ] **Step 2: tokio `run_inner` allocates the two send buffers internally** (API unchanged for `Server::run` callers): + +```rust +// server/mod.rs run_inner (tokio) — alongside the existing two recv vecs +let mut unicast_buf = alloc::vec![0u8; 65535]; +let mut sd_buf = alloc::vec![0u8; 65535]; +let mut recv_send_buf = alloc::vec![0u8; crate::UDP_BUFFER_SIZE]; // ← new +let mut announce_send_buf = alloc::vec![0u8; crate::UDP_BUFFER_SIZE]; // ← new +// ...run_with_buffers(&mut unicast_buf, &mut sd_buf, &mut recv_send_buf, &mut announce_send_buf).await +``` + +- [ ] **Step 3: Update bare-metal callers** (`examples/bare_metal_server`, any `run_with_buffers` test caller) to declare and pass the two send buffers. Grep for `run_with_buffers(` and fix each call site. + +- [ ] **Step 4: Verify** — full suite + bare-metal builds: + +Run: `cargo nextest run --no-default-features --features client-tokio,server-tokio` → green +Run: `cargo build --target thumbv7em-none-eabihf --no-default-features --features server,bare_metal` → builds; `cargo build ... --features client,server,bare_metal` → builds +Run: `cargo test --features client,server,bare_metal --test bare_metal_e2e` → 6/6 + Task 1's new test + +- [ ] **Step 5: Commit** + +```bash +git add src/server/runtime.rs src/server/sd_state.rs src/server/mod.rs examples/ tests/ +git commit -m "feat(server): thread recv+announce send-scratch buffers through run_with_buffers (#125)" +``` + +--- + +### Task 4: `EventPublisher` publish paths take caller scratch (`event_publisher.rs`, `server/mod.rs`) + +**Files:** Modify `src/server/event_publisher.rs` (`publish_event` ~`:158-218`, `publish_raw_event` ~`:313-344`); `src/server/mod.rs` (tokio publisher wrapper / publish entry). + +**Interfaces:** +- Produces: `publish_event(&self, /* msg */, msg_buf: &mut [u8], protected_buf: &mut [u8])` and `publish_raw_event(&self, /* hdr+payload */, buf: &mut [u8])`. E2E needs `protected_buf` because `E2ERegistry::protect(key, input: &[u8], hdr, output: &mut [u8])` requires disjoint in/out slices. + +- [ ] **Step 1: Write the failing test** (E2E publish on undersized scratch → `Capacity`, mirroring PR 2's client regression): + +```rust +// tests/bare_metal_e2e.rs — register an E2E key, publish an event whose protected +// frame exceeds the provided msg_buf/protected_buf, assert Capacity not panic/OOB. +#[tokio::test] +async fn e2e_publish_with_undersized_scratch_returns_capacity_not_panic() { + let outcome = publish_e2e_event_with_scratch_len(40).await; // 36 fits, +12 P4 protect = 48 > 40 + assert!(matches!(outcome, Err(Error::Capacity("udp_buffer")))); +} +``` + +- [ ] **Step 2: Run it — expect FAIL.** + +- [ ] **Step 3: Change `publish_event` / `publish_raw_event` to take scratch and bound on `buf.len()`** (the E2E guard is the PR-2 lesson applied here — `> msg_buf.len()` / `> protected_buf.len()`, not `> UDP_BUFFER_SIZE`): + +```rust +// publish_event: was two stack arrays; now caller scratch. +pub async fn publish_event(&self, /* message */, msg_buf: &mut [u8], protected_buf: &mut [u8]) -> Result<(), Error> { + let mut message_length = message.encode_to_slice(msg_buf)?; // errors if msg_buf too small + if self.e2e_registry.contains_key(&key) { + let upper_header: [u8; 8] = msg_buf[8..16].try_into().expect("upper header slice"); + let result = self.e2e_registry.protect(key, &msg_buf[16..message_length], upper_header, protected_buf); + if let Some(Ok(protected_len)) = result { + if 16 + protected_len > msg_buf.len() { // ← buf.len(), not UDP_BUFFER_SIZE + return Err(Error::Capacity("udp_buffer")); + } + msg_buf[16..16 + protected_len].copy_from_slice(&protected_buf[..protected_len]); + message_length = 16 + protected_len; + } /* preserve the existing Some(Err)/None arms */ + } + let datagram = &msg_buf[..message_length]; + for addr in &subscribers { /* existing send_to(datagram).await loop, unchanged */ } + Ok(()) +} +``` + +- [ ] **Step 4: tokio publisher keeps the app API ergonomic** — the `server-tokio` publish wrapper allocates `msg_buf`/`protected_buf` (Vecs) internally per publish so existing `publish_event(message)` callers are unchanged; only the bare-metal publish path surfaces the `&mut [u8]` params. Gate the internal allocation behind `_alloc`/`server-tokio`. + +- [ ] **Step 5: Run the test (PASS) + full suite + no-alloc witness.** + +Run: `cargo test --features client,server,bare_metal --test bare_metal_e2e e2e_publish_with_undersized_scratch` → PASS +Run: `cargo nextest run --no-default-features --features client-tokio,server-tokio` → green +Run: `cargo test --features client,bare_metal --test no_alloc_witness` → still alloc-free (publish scratch alloc is tokio-gated) + +- [ ] **Step 6: Commit** + +```bash +git add src/server/event_publisher.rs src/server/mod.rs tests/bare_metal_e2e.rs +git commit -m "feat(server): EventPublisher publish paths take caller scratch; E2E bound on buf.len() (#125)" +``` + +--- + +### Task 5: Measure + retighten the server run-future witness + +**Files:** Modify `tests/bare_metal_e2e.rs:606`. + +- [ ] **Step 1: Capture before/after** + +Run: `cargo test --features client,server,bare_metal --test bare_metal_e2e -- --nocapture | grep bm_server_run_future` +Record the new `bm_server_run_future` (expected to drop from 7696 as the recv-path + announce-path send buffers leave `run_combined`'s state). + +- [ ] **Step 2: Set the budget to `ceil64(new × 1.25)`** + +```rust +// tests/bare_metal_e2e.rs:606 +const BM_SERVER_RUN_FUTURE_BUDGET: usize = /* ceil64(new bm_server_run_future × 1.25) */; +``` + +- [ ] **Step 3: Verify** the witness passes at the tightened budget. If the number did NOT move, the run path's buffers weren't the dominant term — record that and keep the extraction only if it helps (per the binding constraint). + +- [ ] **Step 4: Commit** + +```bash +git add tests/bare_metal_e2e.rs +git commit -m "test(server): retighten server run-future budget after send-buffer extraction (#125)" +``` + +--- + +### Task 6: Final #125 numbers + close-out + +**Files:** none (verification + PR description); optionally `CHANGELOG.md`. + +- [ ] **Step 1: Full gate matrix** + +```bash +cargo nextest run --no-default-features --features client-tokio,server-tokio +cargo test --features client,server,bare_metal --test bare_metal_e2e +cargo test --features client,bare_metal --test no_alloc_witness +cargo clippy --workspace --all-features -- -D warnings -D clippy::pedantic +cargo clippy --no-default-features -- -D warnings -D clippy::pedantic +cargo build --target thumbv7em-none-eabihf --no-default-features --features server,bare_metal +cargo build --target thumbv7em-none-eabihf --no-default-features --features client,server,bare_metal +# nm: server,bare_metal alloc status is documented (server uses alloc today); client,bare_metal stays alloc-free +RUSTDOCFLAGS="-D warnings" cargo doc --no-deps --all-features +cargo test --doc --all-features +``` + +- [ ] **Step 2: Capture authoritative thumb numbers** (if `tools/capture_type_sizes.sh` exists) and assemble issue #125's acceptance table: client (PR 2: socket-loop 2224→776) + server (PR 3: `bm_server_run_future` before→after) TaskStorage sizes + any pool/`nm` symbol deltas. + +- [ ] **Step 3: Record the before/after in the PR description**; note in CHANGELOG that the server run/publish buffers are now caller-sized (breaking the `run_with_buffers` and `publish_*` bare-metal signatures — free pre-0.8.0). + +- [ ] **Step 4: Finish the branch** — announce and use superpowers:finishing-a-development-branch; push `feature/pr3_125_server_buffers` and open the draft PR **based on `feature/pr2_125_client_async_state`** (keep it stacked; never merge-down). + +--- + +## Self-Review + +**Spec coverage:** all 6 future-resident send buffers → Tasks 1 (3 SD helpers), 2 (offer), 4 (publish ×2 incl. E2E `protected`). Threading → Task 3 (run) + Task 4 (publish). E2E `buf.len()` bound → Task 4. Witness retighten → Task 5. Final #125 tables → Task 6. The receive buffers are already caller-provided (no task). ✓ + +**Placeholder scan:** the witness budget (Task 5) and the test harness shims (`run_server_subscribe_with_send_scratch_len`, `publish_e2e_event_with_scratch_len`) are resolved at implementation against the real `bare_metal_e2e` harness; the `encode_to_slice` error-mapping in Tasks 1–2 must be matched to the real return type. Flagged inline. + +**Type consistency:** `buf: &mut [u8]` is the uniform scratch param across Tasks 1/2/4; `run_with_buffers` gains exactly `recv_send_buf`/`announce_send_buf` (Task 3) consumed by the Task 1/2 helpers; `publish_event` gains `msg_buf`/`protected_buf`, `publish_raw_event` gains `buf` (Task 4). `Error::Capacity("udp_buffer")` reused verbatim. + +**Design note for the reviewer:** unlike the client (a `BufferProvider` *pool* for dynamically-spawned per-socket loops), the server uses fixed caller-provided scratch because it runs one combined future + an app publish path — there is no dynamic socket spawning, so claim/release is unnecessary. This is the deliberate, architecture-driven divergence from the client. diff --git a/examples/bare_metal_client/src/main.rs b/examples/bare_metal_client/src/main.rs index 37e32cdc..de65d5a9 100644 --- a/examples/bare_metal_client/src/main.rs +++ b/examples/bare_metal_client/src/main.rs @@ -299,11 +299,14 @@ async fn main() { timer: MockTimer, e2e_registry: e2e, interface: iface, - // Caller-declared static buffer pool (#125): one slot per - // possible socket. On real firmware this is a `static`; here it - // is a function-local `static` for the example. + // Caller-declared static buffer pool (#125): UNICAST_SOCKETS_CAP + // (8) + 1 discovery + 1 release-lag slack = 10 slots. An evicted + // socket's lease frees asynchronously, so size one above the max + // live socket count to avoid a transient Capacity("udp_buffer") + // on evict-then-rebind. On real firmware this is a `static`; here + // it is a function-local `static` for the example. buffer_provider: { - static POOL: BufferPool<9, UDP_BUFFER_SIZE> = BufferPool::new(); + static POOL: BufferPool<10, UDP_BUFFER_SIZE> = BufferPool::new(); StaticBufferProvider(&POOL) }, }, diff --git a/examples/embassy_net_client/src/main.rs b/examples/embassy_net_client/src/main.rs index b79e3e79..b4fa5eda 100644 --- a/examples/embassy_net_client/src/main.rs +++ b/examples/embassy_net_client/src/main.rs @@ -399,6 +399,8 @@ async fn main() { tokio::task::spawn_local(server.run_with_buffers( Box::leak(vec![0u8; 65535].into_boxed_slice()), Box::leak(vec![0u8; 65535].into_boxed_slice()), + Box::leak(vec![0u8; simple_someip::UDP_BUFFER_SIZE].into_boxed_slice()), + Box::leak(vec![0u8; simple_someip::UDP_BUFFER_SIZE].into_boxed_slice()), )); println!( "[server] run loop spawned, emitting OfferService(0x{SERVICE_ID:04X}) every 1s" diff --git a/simple-someip-embassy-net/tests/loopback.rs b/simple-someip-embassy-net/tests/loopback.rs index 0f7e571f..2886bdd2 100644 --- a/simple-someip-embassy-net/tests/loopback.rs +++ b/simple-someip-embassy-net/tests/loopback.rs @@ -633,6 +633,8 @@ async fn client_receives_server_sd_announcement() { tokio::task::spawn_local(server.run_with_buffers( Box::leak(Box::new([0u8; 65535])), Box::leak(Box::new([0u8; 65535])), + Box::leak(Box::new([0u8; simple_someip::UDP_BUFFER_SIZE])), + Box::leak(Box::new([0u8; simple_someip::UDP_BUFFER_SIZE])), )); // ── Client on stack B ──────────────────────────────── diff --git a/src/client/mod.rs b/src/client/mod.rs index f3a2b5e3..0283b4d9 100644 --- a/src/client/mod.rs +++ b/src/client/mod.rs @@ -341,10 +341,11 @@ impl /// Build a `ClientDeps` with the tokio defaults. /// /// `buffer_provider` is a single `TokioBufferProvider::new()` - /// constructed here exactly once. `TokioBufferProvider::new()` does - /// a `Box::leak`, so it MUST be one-per-client and never called on a - /// per-bind / hot path — this constructor is the canonical single - /// call site for the tokio path. + /// constructed here exactly once. It is `Arc`-backed (the pool is freed + /// when the last provider/lease drops — not leaked); keeping it + /// one-per-client shares that pool and avoids a fresh heap allocation on + /// every bind, so it should not be reconstructed on a per-bind / hot + /// path — this constructor is the canonical single call site. #[must_use] pub fn tokio(interface: Ipv4Addr) -> Self { Self { @@ -691,8 +692,10 @@ where interface: Arc::new(RwLock::new(interface)), spawner, // One `TokioBufferProvider::new()` per client construction. - // It `Box::leak`s internally, so it must not be moved to a - // per-bind path; this single call covers every `bind_*`. + // It is `Arc`-backed (freed when the last provider/lease + // drops); keeping it one-per-client shares the pool and + // avoids a per-bind heap allocation. This single call + // covers every `bind_*`. buffer_provider: crate::tokio_transport::TokioBufferProvider::new(), }, multicast_loopback, diff --git a/src/client/socket_manager.rs b/src/client/socket_manager.rs index fec0028c..a0b52193 100644 --- a/src/client/socket_manager.rs +++ b/src/client/socket_manager.rs @@ -879,9 +879,9 @@ mod tests { } /// Claim a single socket-loop buffer for a direct `bind_with_transport` - /// call in these unit tests. Each call leaks one small `BufferPool` - /// (acceptable in a test); production paths claim from one shared - /// provider per client. + /// call in these unit tests. Each call builds a fresh `Arc`-backed + /// `TokioBufferProvider` (the pool is freed when the lease drops — no + /// leak); production paths claim from one shared provider per client. fn test_buf() -> crate::buffer_pool::BufferLease { use crate::tokio_transport::TokioBufferProvider; use crate::transport::BufferProvider; diff --git a/src/server/event_publisher.rs b/src/server/event_publisher.rs index 82f6903f..6057f017 100644 --- a/src/server/event_publisher.rs +++ b/src/server/event_publisher.rs @@ -2,7 +2,6 @@ use super::Error; use super::subscription_manager::{SUBSCRIBERS_PER_GROUP, SubscriptionHandle}; -use crate::UDP_BUFFER_SIZE; use crate::e2e::E2EKey; use crate::protocol::{Header, Message}; use crate::traits::{PayloadWireFormat, WireFormat}; @@ -84,29 +83,48 @@ where } } - /// Publish an event to all subscribers of an event group + /// Publish an event to all subscribers of an event group using caller-provided scratch. + /// + /// The `msg_buf` and `protected_buf` slices are the two scratch areas + /// needed for the send path: + /// + /// - `msg_buf` — receives the serialized SOME/IP frame (including any + /// post-E2E copy-back). Must be large enough to hold the full + /// protected datagram: at minimum `16 + E2E_header_overhead + payload_len`. + /// On the bare-metal path, callers typically supply a `static [u8; N]`. + /// + /// - `protected_buf` — temporary scratch for the E2E protect output + /// (`E2ERegistry::protect` requires disjoint in/out slices). Must be + /// at least as large as the protected payload. Ignored when no E2E key + /// is registered for the message. /// /// # Arguments /// * `service_id` - Service ID /// * `instance_id` - Instance ID /// * `event_group_id` - Event group ID /// * `message` - The SOME/IP message to send (must be a notification/event) + /// * `msg_buf` - Caller-supplied scratch for the outgoing datagram + /// * `protected_buf` - Caller-supplied scratch for E2E protect output /// /// # Errors /// - /// Returns an error if the message fails to serialize. + /// Returns an error if the message fails to serialize, or + /// [`Error::Capacity`]`("udp_buffer")` if either scratch buffer is too small + /// for the encoded or E2E-protected frame. /// /// # Panics /// /// May panic if the underlying [`E2ERegistryHandle`](crate::transport::E2ERegistryHandle) /// implementation panics (e.g., `Arc>` on mutex poison). #[allow(clippy::too_many_lines)] - pub async fn publish_event( + pub async fn publish_event_with_buffers( &self, service_id: u16, instance_id: u16, event_group_id: u16, message: &Message

, + msg_buf: &mut [u8], + protected_buf: &mut [u8], ) -> Result { // Snapshot subscriber addresses into a stack-allocated buffer so // we can release the subscription read lock before doing async @@ -141,71 +159,75 @@ where // when it runs out of buffer. Matches the raw-event path below // and the client socket_manager path. let required_size = message.required_size(); - if required_size > UDP_BUFFER_SIZE { + if required_size > msg_buf.len() { crate::log::error!( - "Message size ({} bytes) exceeds UDP_BUFFER_SIZE ({}); dropping publish", + "Message size ({} bytes) exceeds msg_buf.len() ({}); dropping publish", required_size, - UDP_BUFFER_SIZE + msg_buf.len() ); return Err(Error::Capacity("udp_buffer")); } - // Serialize the message into a fixed-size buffer of - // `UDP_BUFFER_SIZE` bytes. (In this `async fn` the buffer lives - // in the future state, not literally on the stack; "MTU-sized" - // is a misleading description since the cap is a UDP payload - // limit, not an Ethernet MTU — see `UDP_BUFFER_SIZE` docs.) - let mut buffer = [0u8; UDP_BUFFER_SIZE]; - let mut message_length = message.encode_to_slice(&mut buffer)?; - - // Apply E2E protect if configured. The `protected` stack buffer is - // disjoint from `buffer`, so we can read the unprotected payload - // directly out of `buffer[16..]` without a separate copy. + // Serialize the message into the caller-provided buffer. + // (PR-3 #125 change: no longer uses an in-future `[u8; UDP_BUFFER_SIZE]`; + // the caller decides the buffer size and lifetime.) + let mut message_length = message.encode_to_slice(msg_buf)?; + + // Apply E2E protect if configured. `protected_buf` is disjoint from + // `msg_buf`, so we can read the unprotected payload directly out of + // `msg_buf[16..]` without a separate copy. The guard is keyed off + // `msg_buf.len()` (not `UDP_BUFFER_SIZE`) — the PR-2 lesson applied + // to the server publish path. { let key = E2EKey::from_message_id(message.header().message_id()); if self.e2e_registry.contains_key(&key) { - let upper_header: [u8; 8] = buffer[8..16].try_into().expect("upper header slice"); - let mut protected = [0u8; UDP_BUFFER_SIZE]; + let upper_header: [u8; 8] = msg_buf[8..16].try_into().expect("upper header slice"); let result = self.e2e_registry.protect( key, - &buffer[16..message_length], + &msg_buf[16..message_length], upper_header, - &mut protected, + protected_buf, ); match result { Some(Ok(protected_len)) => { - if 16 + protected_len > UDP_BUFFER_SIZE { + if 16 + protected_len > msg_buf.len() { crate::log::error!( "E2E-protected datagram ({} bytes, header + protected payload) \ - exceeds UDP_BUFFER_SIZE ({}); dropping publish", + exceeds msg_buf.len() ({}); dropping publish", 16 + protected_len, - UDP_BUFFER_SIZE + msg_buf.len() ); return Err(Error::Capacity("udp_buffer")); } #[allow(clippy::cast_possible_truncation)] let new_length: u32 = 8 + protected_len as u32; - buffer[4..8].copy_from_slice(&new_length.to_be_bytes()); - buffer[16..16 + protected_len].copy_from_slice(&protected[..protected_len]); + msg_buf[4..8].copy_from_slice(&new_length.to_be_bytes()); + msg_buf[16..16 + protected_len] + .copy_from_slice(&protected_buf[..protected_len]); message_length = 16 + protected_len; } - Some(Err(e)) => { - // Surface protect failures as `Err(Error::E2e(_))` - // rather than logging-and-falling-through, which - // would silently send the UNPROTECTED payload - // claiming an E2E-protected channel and break the - // receiver's CRC/counter checks. Counter - // exhaustion, key-lookup races, and similar - // backend errors all funnel here. - crate::log::error!("E2E protect error: {:?}; dropping publish", e); - return Err(Error::E2e(e)); + Some(Err(e @ crate::e2e::Error::BufferTooSmall { .. })) => { + // `protect` returned `BufferTooSmall`, meaning the + // caller-supplied `protected_buf` was too short. + // Map to `Capacity("udp_buffer")` for symmetry with + // the pre-encode and post-protect `msg_buf` guards + // above — the PR-3 contract is "undersized scratch → + // `Error::Capacity`". If `crate::e2e::Error` gains + // new variants in the future, they should be mapped + // here explicitly (new variants would require a + // new arm or the exhaustiveness check will catch it). + crate::log::error!( + "E2E protect error (buffer too small): {:?}; dropping publish", + e + ); + return Err(Error::Capacity("udp_buffer")); } None => unreachable!("contains_key was true"), } } } - let datagram = &buffer[..message_length]; + let datagram = &msg_buf[..message_length]; // Send to all snapshotted subscribers. Track the last // transport error so we can surface "every send failed" as @@ -249,15 +271,70 @@ where Ok(sent_count) } - /// Publish raw event data (already serialized with E2E protection) + /// Publish an event to all subscribers of an event group. + /// + /// Convenience wrapper over [`Self::publish_event_with_buffers`] that + /// internally allocates the two scratch `Vec`s required for the send + /// path. Available only when an allocator is present (`_alloc` feature). + /// Bare-metal callers without an allocator must supply their own + /// scratch via [`Self::publish_event_with_buffers`] directly. + /// + /// Existing `server-tokio` callers — which call `publish_event(...)` via + /// an `Arc>` handle — are unchanged by the PR-3 #125 + /// scratch-extraction refactor: the allocation is invisible at the call + /// site and the signature is identical to the pre-refactor version. + /// + /// # Arguments + /// * `service_id` - Service ID + /// * `instance_id` - Instance ID + /// * `event_group_id` - Event group ID + /// * `message` - The SOME/IP message to send + /// + /// # Errors + /// + /// Returns an error if serialization fails or the serialized frame + /// exceeds the internally-allocated scratch buffer (which is sized + /// to `crate::UDP_BUFFER_SIZE`). Callers that need to control the + /// buffer length must use [`Self::publish_event_with_buffers`] + /// directly. + #[cfg(feature = "_alloc")] + pub async fn publish_event( + &self, + service_id: u16, + instance_id: u16, + event_group_id: u16, + message: &Message

, + ) -> Result { + let mut msg_buf = alloc::vec![0u8; crate::UDP_BUFFER_SIZE]; + let mut protected_buf = alloc::vec![0u8; crate::UDP_BUFFER_SIZE]; + self.publish_event_with_buffers( + service_id, + instance_id, + event_group_id, + message, + &mut msg_buf, + &mut protected_buf, + ) + .await + } + + /// Publish raw event data using a caller-provided scratch buffer. /// - /// This is useful when you've already applied E2E protection to the payload + /// The `buf` slice receives the serialized SOME/IP header + payload + /// datagram before being sent to each subscriber. The caller must + /// supply a buffer large enough to hold `16 + payload.len()` bytes; + /// [`Error::Capacity`]`("udp_buffer")` is returned if the buffer is + /// too small, without writing any bytes. On the bare-metal path, + /// callers typically supply a `static [u8; N]`. + /// + /// This is useful when you've already applied E2E protection to the payload. /// /// # Errors /// - /// Returns an error if the SOME/IP header fails to serialize. + /// Returns an error if the SOME/IP header fails to serialize, or + /// [`Error::Capacity`]`("udp_buffer")` if `buf` is too small for the frame. #[allow(clippy::too_many_arguments)] - pub async fn publish_raw_event( + pub async fn publish_raw_event_with_buffers( &self, service_id: u16, instance_id: u16, @@ -267,9 +344,10 @@ where protocol_version: u8, interface_version: u8, payload: &[u8], + buf: &mut [u8], ) -> Result { // Snapshot subscriber addresses into a stack buffer (see - // publish_event for rationale). + // publish_event_with_buffers for rationale). let mut subscribers: HeaplessVec = HeaplessVec::new(); let _total = self .subscriptions @@ -284,16 +362,27 @@ where // Pre-build size check. Fail fast with `Error::Capacity` BEFORE // calling `Header::new_event`, which `assert!`s on payloads - // larger than `u32::MAX as usize - 8`. The earlier - // `checked_add(header_len, payload.len())` guard below was dead - // for that reason; keeping it for defence-in-depth on platforms - // where `Header::SIZE + payload` could overflow `usize`. The - // `16` here is the SOME/IP header size in bytes. - if payload.len() > UDP_BUFFER_SIZE.saturating_sub(16) { + // larger than `u32::MAX as usize - 8`. The `16` here is the + // SOME/IP header size in bytes. Guard is keyed off `buf.len()` + // (not `UDP_BUFFER_SIZE`) — the PR-2 lesson applied here too. + // + // Guard `buf.len() < 16` explicitly: with an empty payload and a + // sub-header buffer, the `payload.len() > buf.len() - 16` check + // below (saturating to 0) would not fire, and `encode_to_slice` + // would then surface a protocol I/O error instead of the typed + // `Capacity`. (#133 review.) + if buf.len() < 16 { crate::log::error!( - "raw event payload ({} bytes) + 16-byte header exceeds UDP_BUFFER_SIZE ({}); dropping publish", + "raw event buffer ({} bytes) too small for the 16-byte SOME/IP header; dropping publish", + buf.len() + ); + return Err(Error::Capacity("udp_buffer")); + } + if payload.len() > buf.len().saturating_sub(16) { + crate::log::error!( + "raw event payload ({} bytes) + 16-byte header exceeds buf.len() ({}); dropping publish", payload.len(), - UDP_BUFFER_SIZE + buf.len() ); return Err(Error::Capacity("udp_buffer")); } @@ -308,10 +397,9 @@ where payload.len(), ); - // Serialize header + payload into a fixed-size buffer of - // `UDP_BUFFER_SIZE` bytes. See note in `publish_event` above. - let mut buffer = [0u8; UDP_BUFFER_SIZE]; - let header_len = header.encode_to_slice(&mut buffer)?; + // Serialize header + payload into the caller-provided buffer. + // (PR-3 #125 change: no longer uses an in-future `[u8; UDP_BUFFER_SIZE]`.) + let header_len = header.encode_to_slice(buf)?; let Some(total_len) = header_len.checked_add(payload.len()) else { crate::log::error!( "raw event length computation overflowed usize (header_len={}, payload.len()={}); dropping publish", @@ -324,20 +412,20 @@ where // oversize payloads, but a future caller adding optional // post-encode tail bytes (e.g. another protect profile) would // need this branch. Cheap to keep. - if total_len > UDP_BUFFER_SIZE { + if total_len > buf.len() { crate::log::error!( - "raw event ({} bytes) exceeds UDP_BUFFER_SIZE ({}); dropping publish", + "raw event ({} bytes) exceeds buf.len() ({}); dropping publish", total_len, - UDP_BUFFER_SIZE + buf.len() ); return Err(Error::Capacity("udp_buffer")); } - buffer[header_len..total_len].copy_from_slice(payload); - let datagram = &buffer[..total_len]; + buf[header_len..total_len].copy_from_slice(payload); + let datagram = &buf[..total_len]; // Send to all snapshotted subscribers; surface total-failure // as `Err(Transport(_))` rather than `Ok(0)` (see - // `publish_event`). + // `publish_event_with_buffers`). let mut sent_count = 0usize; let mut last_err: Option = None; for addr in &subscribers { @@ -360,6 +448,53 @@ where Ok(sent_count) } + /// Publish raw event data (already serialized with E2E protection). + /// + /// Convenience wrapper over [`Self::publish_raw_event_with_buffers`] that + /// internally allocates the scratch `Vec` required for the send path. + /// Available only when an allocator is present (`_alloc` feature). + /// Bare-metal callers without an allocator must supply their own scratch + /// via [`Self::publish_raw_event_with_buffers`] directly. + /// + /// Existing `server-tokio` callers are unchanged by the PR-3 #125 + /// scratch-extraction refactor — the allocation is invisible at the call + /// site and the signature is identical to the pre-refactor version. + /// + /// # Errors + /// + /// Returns an error if the SOME/IP header fails to serialize or the + /// frame exceeds the internally-allocated scratch buffer (which is + /// sized to `crate::UDP_BUFFER_SIZE`). Callers that need to control + /// the buffer length must use [`Self::publish_raw_event_with_buffers`] + /// directly. + #[cfg(feature = "_alloc")] + #[allow(clippy::too_many_arguments)] + pub async fn publish_raw_event( + &self, + service_id: u16, + instance_id: u16, + event_group_id: u16, + event_id: u16, + request_id: u32, + protocol_version: u8, + interface_version: u8, + payload: &[u8], + ) -> Result { + let mut buf = alloc::vec![0u8; crate::UDP_BUFFER_SIZE]; + self.publish_raw_event_with_buffers( + service_id, + instance_id, + event_group_id, + event_id, + request_id, + protocol_version, + interface_version, + payload, + &mut buf, + ) + .await + } + /// Check if there are any active subscribers for a specific event group /// /// # Arguments @@ -479,6 +614,7 @@ where #[cfg(all(test, feature = "server-tokio"))] mod tests { use super::*; + use crate::UDP_BUFFER_SIZE; use crate::e2e::E2ERegistry; use crate::protocol::sd::test_support::{TestPayload, empty_sd_header}; use crate::server::SubscriptionManager; diff --git a/src/server/mod.rs b/src/server/mod.rs index 6884f9bf..c13fd873 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -1201,7 +1201,7 @@ where /// Register an E2E profile for the given key. /// - /// Once registered, outgoing events published via [`EventPublisher::publish_event`] + /// Once registered, outgoing events published via `EventPublisher::publish_event` /// will have E2E protection applied automatically. /// /// # Errors @@ -1230,7 +1230,7 @@ where /// 1-Hz `OfferService` announcement loop. The two are combined /// into a single future so callers cannot forget to spawn the /// announcement side; passing - /// [`ServerConfig::with_announce`]`(false)` suppresses the + /// [`ServerConfig::with_announce`] with `false` suppresses the /// announcement arm for dispatcher topologies where a co-located /// `Client` drives SD on the server's behalf. /// @@ -1261,6 +1261,8 @@ where &self, unicast_buf: &'a mut [u8], sd_buf: &'a mut [u8], + recv_send_buf: &'a mut [u8], + announce_send_buf: &'a mut [u8], ) -> impl core::future::Future> + 'a + use<'a, F, Tm, R, Sub, H, Hsd, Hep> where Tm: 'a, @@ -1305,12 +1307,75 @@ where is_passive, unicast_buf, sd_buf, + recv_send_buf, + announce_send_buf, non_sd_observer, ) .await } } + /// Run *only* the SD `OfferService` announcement loop with a + /// caller-provided scratch buffer. Use this on bare-metal + /// supplementary Servers that share a `sd_socket` / + /// `unicast_socket` handle (via [`Self::new_with_handles`]) with a + /// primary Server already running [`Self::run_with_buffers`]: the + /// primary owns the inbound recv loops, supplementary Servers add + /// their own `OfferService` to the same SD multicast group without + /// competing for inbound datagrams. + /// + /// The caller provides the send scratch `announce_send_buf` so the + /// future does NOT park a `[u8; UDP_BUFFER_SIZE]` (≈ 1500 B) in + /// its own state. Bare-metal callers typically supply a + /// `static [u8; N]`: + /// + /// ```ignore + /// static mut ANNOUNCE_BUF: [u8; simple_someip::UDP_BUFFER_SIZE] = + /// [0u8; simple_someip::UDP_BUFFER_SIZE]; + /// // SAFETY: only one future accesses this buffer concurrently. + /// let fut = server.announce_only_with_buffer(unsafe { &mut ANNOUNCE_BUF }); + /// executor.spawn(fut); + /// ``` + /// + /// std / alloc callers can use `Self::announce_only_future` + /// instead, which heap-allocates the buffer internally. + /// + /// Design note: this partially reintroduces the split-future shape + /// phase 21 removed — deliberately. An announce-only future never + /// touches the receive path, so the invariant that motivated the + /// phase-21 combined run-future (no two futures racing the same + /// sockets and SD session counter) is preserved: the `Self::run` + /// path is still guarded by the first-poll `started` latch, and + /// supplementary announce loops only ever *send* on the shared SD + /// socket. + /// + /// The returned future loops forever (1 s tick between + /// announcements); spawn it on your executor. + pub fn announce_only_with_buffer<'a>( + &self, + announce_send_buf: &'a mut [u8], + ) -> impl core::future::Future + 'a + use<'a, F, Tm, R, Sub, H, Hsd, Hep> + where + Tm: 'a, + Hsd: 'a, + H: 'a, + { + let config = self.config.clone(); + let sd_socket = self.sd_socket.clone(); + let sd_state = self.sd_state.clone(); + let timer = self.timer.clone(); + async move { + runtime::announce_loop( + &config, + sd_socket.get(), + sd_state.get(), + &timer, + announce_send_buf, + ) + .await; + } + } + /// Run *only* the SD `OfferService` announcement loop, without /// driving the receive path. Use this on supplementary Servers /// that share a `sd_socket` / `unicast_socket` handle (via @@ -1320,6 +1385,13 @@ where /// `OfferService` to the same SD multicast group without /// competing for inbound datagrams. /// + /// This is the `_alloc` convenience wrapper — it heap-allocates + /// the send scratch internally. Bare-metal callers that cannot + /// park a `[u8; UDP_BUFFER_SIZE]` (≈ 1500 B) on the heap should + /// use [`Self::announce_only_with_buffer`] instead, which accepts + /// a caller-provided buffer so the heap allocation is avoided + /// entirely. + /// /// Design note: this partially reintroduces the split-future shape /// phase 21 removed — deliberately. An announce-only future never /// touches the receive path, so the invariant that motivated the @@ -1331,6 +1403,7 @@ where /// /// The returned future loops forever (1 s tick between /// announcements); spawn it on your executor. + #[cfg(feature = "_alloc")] pub fn announce_only_future<'a>( &self, ) -> impl core::future::Future + 'a + use<'a, F, Tm, R, Sub, H, Hsd, Hep> @@ -1344,7 +1417,19 @@ where let sd_state = self.sd_state.clone(); let timer = self.timer.clone(); async move { - runtime::announce_loop(&config, sd_socket.get(), sd_state.get(), &timer).await; + // Heap-allocate the send scratch here so the caller does + // not need to manage the buffer lifetime. Bare-metal callers + // that cannot use the allocator should call + // `announce_only_with_buffer` with a static scratch buffer. + let mut announce_send_buf = alloc::vec![0u8; crate::UDP_BUFFER_SIZE]; + runtime::announce_loop( + &config, + sd_socket.get(), + sd_state.get(), + &timer, + &mut announce_send_buf, + ) + .await; } } @@ -1436,6 +1521,13 @@ where let mut unicast_buf = alloc::vec![0u8; 65535]; let mut sd_buf = alloc::vec![0u8; 65535]; + // Two DISTINCT send-scratch buffers — `recv_loop` and + // `announce_loop` run concurrently and can each be parked at a + // `send_to().await`, so a shared buffer would mutably alias. + // Heap-backed here (this is the `_alloc` path); bare-metal + // callers pass their own via `run_with_buffers`. + let mut recv_send_buf = alloc::vec![0u8; crate::UDP_BUFFER_SIZE]; + let mut announce_send_buf = alloc::vec![0u8; crate::UDP_BUFFER_SIZE]; runtime::run_combined::( config, unicast_socket, @@ -1446,6 +1538,8 @@ where is_passive, &mut unicast_buf, &mut sd_buf, + &mut recv_send_buf, + &mut announce_send_buf, non_sd_observer, ) .await @@ -1738,7 +1832,16 @@ mod tests { let server = TestServer::new_passive_with_handles(handles, config).expect("passive ctor"); let mut unicast_buf = vec![0u8; 1500]; let mut sd_buf = vec![0u8; 1500]; - let result = server.run_with_buffers(&mut unicast_buf, &mut sd_buf).await; + let mut recv_send_buf = vec![0u8; 1500]; + let mut announce_send_buf = vec![0u8; 1500]; + let result = server + .run_with_buffers( + &mut unicast_buf, + &mut sd_buf, + &mut recv_send_buf, + &mut announce_send_buf, + ) + .await; match result { Err(Error::InvalidUsage(tag)) => assert_eq!(tag, "passive_server_run"), other => { @@ -1903,6 +2006,7 @@ mod tests { &server.subscriptions, &sd_view, sender, + &mut [0u8; crate::UDP_BUFFER_SIZE], ) .await; assert!( @@ -2073,6 +2177,7 @@ mod tests { &server.subscriptions, &sd_view, addr, + &mut [0u8; crate::UDP_BUFFER_SIZE], ) .await .unwrap(); @@ -2136,6 +2241,7 @@ mod tests { &server.subscriptions, &sd_view, addr, + &mut [0u8; crate::UDP_BUFFER_SIZE], ) .await .unwrap(); @@ -2196,6 +2302,7 @@ mod tests { &server.subscriptions, &sd_view, addr, + &mut [0u8; crate::UDP_BUFFER_SIZE], ) .await .unwrap(); @@ -2254,6 +2361,7 @@ mod tests { &server.subscriptions, &sd_view, addr, + &mut [0u8; crate::UDP_BUFFER_SIZE], ) .await .unwrap(); @@ -2315,6 +2423,7 @@ mod tests { &server.subscriptions, &sd_view, addr, + &mut [0u8; crate::UDP_BUFFER_SIZE], ) .await .unwrap(); @@ -2373,6 +2482,7 @@ mod tests { &server.subscriptions, &sd_view, addr, + &mut [0u8; crate::UDP_BUFFER_SIZE], ) .await .unwrap(); @@ -2424,6 +2534,7 @@ mod tests { &server.subscriptions, &sd_view, addr, + &mut [0u8; crate::UDP_BUFFER_SIZE], ) .await .unwrap(); @@ -2457,7 +2568,10 @@ mod tests { let recv_addr = receiver.local_addr().unwrap(); let (server, _) = create_test_server(0x5B, 1).await; + // PR3/#125 Task 1: the SD send helpers now take a caller-provided + // scratch buffer (was a future-resident `[u8; UDP_BUFFER_SIZE]`). runtime::send_unicast_offer( + &mut [0u8; crate::UDP_BUFFER_SIZE], &server.config, server.sd_socket.get(), server.sd_state.get(), @@ -2652,6 +2766,7 @@ mod tests { &server.subscriptions, &sd_view, "127.0.0.1:12345".parse().unwrap(), + &mut [0u8; crate::UDP_BUFFER_SIZE], ) .await; assert!(result.is_ok()); @@ -2692,6 +2807,7 @@ mod tests { &server.subscriptions, &sd_view, addr, + &mut [0u8; crate::UDP_BUFFER_SIZE], ) .await .unwrap(); @@ -3007,6 +3123,7 @@ mod tests { &server.subscriptions, &sd_view, sender, + &mut [0u8; crate::UDP_BUFFER_SIZE], ) .await .unwrap(); @@ -3176,7 +3293,16 @@ mod tests { // Same gate on `run_with_buffers`. let mut unicast_buf = vec![0u8; 1500]; let mut sd_buf = vec![0u8; 1500]; - let third = server.run_with_buffers(&mut unicast_buf, &mut sd_buf).await; + let mut recv_send_buf = vec![0u8; 1500]; + let mut announce_send_buf = vec![0u8; 1500]; + let third = server + .run_with_buffers( + &mut unicast_buf, + &mut sd_buf, + &mut recv_send_buf, + &mut announce_send_buf, + ) + .await; match third { Err(Error::InvalidUsage(tag)) => { assert_eq!(tag, "server_already_running"); diff --git a/src/server/runtime.rs b/src/server/runtime.rs index 9f07235b..7fd1d68a 100644 --- a/src/server/runtime.rs +++ b/src/server/runtime.rs @@ -26,7 +26,12 @@ use super::{Error, ServerConfig}; /// Send a unicast `OfferService` to a specific address (typically in /// response to a `FindService`). +/// +/// `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_unicast_offer( + buf: &mut [u8], config: &ServerConfig, sd_socket: &T, sd_state: &SdStateManager, @@ -60,14 +65,25 @@ where let (sid, reboot_flag) = sd_state.next_session_id_with_reboot_flag(); let sd_payload = sd::Header::new(Flags::new_sd(reboot_flag), &entries, &options); - let mut buffer = [0u8; crate::UDP_BUFFER_SIZE]; - let sd_data_len = sd_payload.encode_to_slice(&mut buffer[16..])?; - let someip_header = SomeIpHeader::new_sd(sid, sd_data_len); - someip_header.encode_to_slice(&mut buffer[..16])?; + // Guard: SOME/IP header needs 16 bytes; SD payload needs the rest. + 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 (dead) 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 target_v4 = socket_addr_v4(target)?; - sd_socket.send_to(&buffer[..total_len], target_v4).await?; + sd_socket.send_to(&buf[..total_len], target_v4).await?; crate::log::debug!( "Sent unicast OfferService to {} for service 0x{:04X}", target, @@ -78,7 +94,12 @@ where } /// Send `SubscribeAck` derived from a peer's `Subscribe` entry view. +/// +/// `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_subscribe_ack_from_view( + buf: &mut [u8], config: &ServerConfig, sd_socket: &T, sd_state: &SdStateManager, @@ -107,15 +128,26 @@ where let (sid, reboot_flag) = sd_state.next_session_id_with_reboot_flag(); let sd_payload = sd::Header::new(Flags::new_sd(reboot_flag), &entries, &[]); - let mut buffer = [0u8; crate::UDP_BUFFER_SIZE]; - let sd_data_len = sd_payload.encode_to_slice(&mut buffer[16..])?; - let someip_header = SomeIpHeader::new_sd(sid, sd_data_len); - someip_header.encode_to_slice(&mut buffer[..16])?; + // Guard: SOME/IP header needs 16 bytes; SD payload needs the rest. + 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 (dead) 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 subscriber_v4 = socket_addr_v4(subscriber)?; sd_socket - .send_to(&buffer[..total_len], subscriber_v4) + .send_to(&buf[..total_len], subscriber_v4) .await?; crate::log::debug!( @@ -129,7 +161,12 @@ where } /// Send `SubscribeNack` (`SubscribeAckEventGroup` with `ttl = 0`). +/// +/// `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_subscribe_nack_from_view( + buf: &mut [u8], _config: &ServerConfig, sd_socket: &T, sd_state: &SdStateManager, @@ -159,15 +196,26 @@ where let (sid, reboot_flag) = sd_state.next_session_id_with_reboot_flag(); let sd_payload = sd::Header::new(Flags::new_sd(reboot_flag), &entries, &[]); - let mut buffer = [0u8; crate::UDP_BUFFER_SIZE]; - let sd_data_len = sd_payload.encode_to_slice(&mut buffer[16..])?; - let someip_header = SomeIpHeader::new_sd(sid, sd_data_len); - someip_header.encode_to_slice(&mut buffer[..16])?; + // Guard: SOME/IP header needs 16 bytes; SD payload needs the rest. + 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 (dead) 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 subscriber_v4 = socket_addr_v4(subscriber)?; sd_socket - .send_to(&buffer[..total_len], subscriber_v4) + .send_to(&buf[..total_len], subscriber_v4) .await?; crate::log::warn!( @@ -190,6 +238,7 @@ pub(super) async fn handle_sd_message( subscriptions: &Sub, sd_view: &sd::SdHeaderView<'_>, sender: core::net::SocketAddr, + send_buf: &mut [u8], ) -> Result<(), Error> where T: TransportSocket, @@ -197,6 +246,12 @@ where { crate::log::trace!("Handling SD message from {}", sender); + // `send_buf` is the caller-owned send scratch threaded down from + // `recv_loop` (which holds exactly one — only one inbound SD message + // is handled at a time, so only one helper send is ever in flight). + // It replaces the former future-resident `[u8; UDP_BUFFER_SIZE]`, + // keeping that buffer out of the run future's frame. + for entry_view in sd_view.entries() { let entry_type = entry_view.entry_type()?; match entry_type { @@ -216,6 +271,7 @@ where entry_view.service_id() ); send_subscribe_nack_from_view( + send_buf, config, sd_socket, sd_state, @@ -231,6 +287,7 @@ where entry_view.instance_id() ); send_subscribe_nack_from_view( + send_buf, config, sd_socket, sd_state, @@ -246,6 +303,7 @@ where entry_view.major_version() ); if let Err(e) = send_subscribe_nack_from_view( + send_buf, config, sd_socket, sd_state, @@ -264,6 +322,7 @@ where entry_view.service_id() ); if let Err(e) = send_subscribe_nack_from_view( + send_buf, config, sd_socket, sd_state, @@ -299,6 +358,7 @@ where match subscribe_result { Ok(()) => { if let Err(e) = send_subscribe_ack_from_view( + send_buf, config, sd_socket, sd_state, @@ -334,6 +394,7 @@ where }; crate::log::debug!("Subscription rejected: {reason}"); if let Err(e) = send_subscribe_nack_from_view( + send_buf, config, sd_socket, sd_state, @@ -350,6 +411,7 @@ where } else { crate::log::warn!("No endpoint found in Subscribe message options"); if let Err(e) = send_subscribe_nack_from_view( + send_buf, config, sd_socket, sd_state, @@ -373,7 +435,15 @@ where find_service_id, config.service_id ); - if let Err(e) = send_unicast_offer(config, sd_socket, sd_state, sender).await { + if let Err(e) = send_unicast_offer( + send_buf, + config, + sd_socket, + sd_state, + sender, + ) + .await + { crate::log::warn!("Unicast OfferService send failed: {e}"); } } else { @@ -399,13 +469,17 @@ pub(super) async fn announce_loop( sd_socket: &T, sd_state: &SdStateManager, timer: &Tm, + announce_send_buf: &mut [u8], ) where T: TransportSocket, Tm: Timer, { let mut announcement_count = 0u32; loop { - match sd_state.send_offer_service(config, sd_socket).await { + match sd_state + .send_offer_service(announce_send_buf, config, sd_socket) + .await + { Ok(()) => { announcement_count += 1; if announcement_count == 1 { @@ -440,6 +514,7 @@ async fn recv_loop( subscriptions: &Sub, unicast_buf: &mut [u8], sd_buf: &mut [u8], + send_buf: &mut [u8], non_sd_observer: Option<(super::NonSdRequestCallback, usize)>, ) -> Result<(), Error> where @@ -528,6 +603,7 @@ where subscriptions, &sd_view, addr, + send_buf, ) .await?; } @@ -594,6 +670,8 @@ pub(super) async fn run_combined( is_passive: bool, unicast_buf: &mut [u8], sd_buf: &mut [u8], + recv_send_buf: &mut [u8], + announce_send_buf: &mut [u8], non_sd_observer: Option<(super::NonSdRequestCallback, usize)>, ) -> Result<(), Error> where @@ -626,11 +704,16 @@ where &subscriptions, unicast_buf, sd_buf, + recv_send_buf, non_sd_observer, ); if config.announce { - let announce_fut = announce_loop(&config, sd, sd_state_ref, &timer); + // Two DISTINCT send buffers: `recv_loop` and `announce_loop` run + // concurrently under the `select` below, and both can be suspended + // at a `send_to().await` simultaneously. Sharing one buffer would + // mutably alias it across the two live futures — UB / corruption. + let announce_fut = announce_loop(&config, sd, sd_state_ref, &timer, announce_send_buf); pin_mut!(recv_fut, announce_fut); match futures_util::future::select(recv_fut, announce_fut).await { Either::Left((recv_result, _)) => recv_result, @@ -712,3 +795,289 @@ pub(super) fn extract_subscriber_endpoint( } } } + +// ── Unit tests for SD send helpers ─────────────────────────────────────────── +// +// These tests live in `runtime.rs` (rather than `tests/bare_metal_e2e.rs`) +// because the three helpers are `pub(super)` and are not reachable from +// integration-test crates. Task 3 will expose a public surface (via +// `run_with_buffers` threading the real scratch) that integration tests can +// exercise end-to-end; until then, the helper-level contract is verified here. +#[cfg(test)] +mod tests { + use super::*; + + use core::net::{Ipv4Addr, SocketAddrV4}; + + use crate::transport::{ReceivedDatagram, TransportError, TransportSocket}; + + // ── Minimal no-op mock socket ───────────────────────────────────────── + + struct NullSocket; + + struct NullSend; + impl core::future::Future for NullSend { + type Output = Result<(), TransportError>; + fn poll( + self: core::pin::Pin<&mut Self>, + _cx: &mut core::task::Context<'_>, + ) -> core::task::Poll { + core::task::Poll::Ready(Ok(())) + } + } + + struct NullRecv; + impl core::future::Future for NullRecv { + type Output = Result; + fn poll( + self: core::pin::Pin<&mut Self>, + _cx: &mut core::task::Context<'_>, + ) -> core::task::Poll { + core::task::Poll::Pending + } + } + + impl TransportSocket for NullSocket { + type SendFuture<'a> = NullSend; + type RecvFuture<'a> = NullRecv; + + fn send_to<'a>(&'a self, _buf: &'a [u8], _target: SocketAddrV4) -> Self::SendFuture<'a> { + NullSend + } + fn recv_from<'a>(&'a self, _buf: &'a mut [u8]) -> Self::RecvFuture<'a> { + NullRecv + } + fn local_addr(&self) -> Result { + Ok(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 0)) + } + fn join_multicast_v4( + &self, + _group: Ipv4Addr, + _iface: Ipv4Addr, + ) -> Result<(), TransportError> { + Ok(()) + } + fn leave_multicast_v4( + &self, + _group: Ipv4Addr, + _iface: Ipv4Addr, + ) -> Result<(), TransportError> { + Ok(()) + } + } + + fn make_config() -> ServerConfig { + ServerConfig::new(0x1234, 1) + .with_interface(Ipv4Addr::LOCALHOST) + .with_local_port(30500) + } + + fn make_sd_state() -> SdStateManager { + SdStateManager::new() + } + + fn subscriber_addr() -> core::net::SocketAddr { + core::net::SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(127, 0, 0, 1), 40000)) + } + + // ── Task 1 RED/GREEN: undersized buf rejects with Capacity, not panic ─ + + /// `send_unicast_offer` with a 24-byte buf (fits 16-byte header but + /// not the SD payload) must return `Err(Capacity("udp_buffer"))`. + #[tokio::test] + async fn send_unicast_offer_undersized_buf_returns_capacity() { + let config = make_config(); + let sd_state = make_sd_state(); + let socket = NullSocket; + let target = subscriber_addr(); + + let result = + send_unicast_offer(&mut [0u8; 24], &config, &socket, &sd_state, target).await; + + assert!( + matches!(result, Err(Error::Capacity("udp_buffer"))), + "expected Capacity(\"udp_buffer\"), got {result:?}" + ); + } + + /// `send_unicast_offer` with a buf shorter than the 16-byte SOME/IP + /// header must return `Err(Capacity("udp_buffer"))`. + #[tokio::test] + async fn send_unicast_offer_buf_shorter_than_header_returns_capacity() { + let config = make_config(); + let sd_state = make_sd_state(); + let socket = NullSocket; + let target = subscriber_addr(); + + let result = + send_unicast_offer(&mut [0u8; 8], &config, &socket, &sd_state, target).await; + + assert!( + matches!(result, Err(Error::Capacity("udp_buffer"))), + "expected Capacity(\"udp_buffer\"), got {result:?}" + ); + } + + /// `send_unicast_offer` with a full-size buf must succeed (no regression). + #[tokio::test] + async fn send_unicast_offer_full_size_buf_succeeds() { + let config = make_config(); + let sd_state = make_sd_state(); + let socket = NullSocket; + let target = subscriber_addr(); + + let result = send_unicast_offer( + &mut [0u8; crate::UDP_BUFFER_SIZE], + &config, + &socket, + &sd_state, + target, + ) + .await; + + assert!(result.is_ok(), "full-size buf must succeed, got {result:?}"); + } + + // ── send_subscribe_ack_from_view: build a minimal EntryView via wire bytes + + /// Encode a minimal Subscribe SD payload and return `(wire_bytes, sd_len)` so + /// callers can parse an `SdHeaderView` and extract an `EntryView`. + fn subscribe_wire_bytes() -> ([u8; 512], usize) { + use crate::traits::WireFormat; + + let entry = sd::Entry::SubscribeEventGroup(sd::EventGroupEntry { + index_first_options_run: 0, + index_second_options_run: 0, + options_count: sd::OptionsCount::new(1, 0), + service_id: 0x1234, + instance_id: 1, + major_version: 1, + ttl: 3, + counter: 0, + event_group_id: 0x0001, + }); + let option = sd::Options::IpV4Endpoint { + ip: Ipv4Addr::LOCALHOST, + port: 40000, + protocol: sd::TransportProtocol::Udp, + }; + let entries = [entry]; + let options = [option]; + let sd_payload = + sd::Header::new(sd::Flags::new_sd(sd::RebootFlag::RecentlyRebooted), &entries, &options); + + let mut wire = [0u8; 512]; + let sd_len = sd_payload.encode_to_slice(&mut wire).expect("encode"); + (wire, sd_len) + } + + /// `send_subscribe_ack_from_view` with a 24-byte buf must return + /// `Err(Capacity("udp_buffer"))` without panicking. + #[tokio::test] + async fn send_subscribe_ack_undersized_buf_returns_capacity_not_panic() { + let config = make_config(); + let sd_state = make_sd_state(); + let socket = NullSocket; + let subscriber = subscriber_addr(); + + let (wire, sd_len) = subscribe_wire_bytes(); + let sd_view = sd::SdHeaderView::parse(&wire[..sd_len]).expect("parse"); + let entry_view = sd_view.entries().next().expect("one entry"); + + let result = send_subscribe_ack_from_view( + &mut [0u8; 24], + &config, + &socket, + &sd_state, + &entry_view, + subscriber, + ) + .await; + + assert!( + matches!(result, Err(Error::Capacity("udp_buffer"))), + "expected Capacity(\"udp_buffer\"), got {result:?}" + ); + } + + /// `send_subscribe_ack_from_view` with a full-size buf must succeed. + #[tokio::test] + async fn send_subscribe_ack_full_size_buf_succeeds() { + let config = make_config(); + let sd_state = make_sd_state(); + let socket = NullSocket; + let subscriber = subscriber_addr(); + + let (wire, sd_len) = subscribe_wire_bytes(); + let sd_view = sd::SdHeaderView::parse(&wire[..sd_len]).expect("parse"); + let entry_view = sd_view.entries().next().expect("one entry"); + + let result = send_subscribe_ack_from_view( + &mut [0u8; crate::UDP_BUFFER_SIZE], + &config, + &socket, + &sd_state, + &entry_view, + subscriber, + ) + .await; + + assert!(result.is_ok(), "full-size buf must succeed, got {result:?}"); + } + + /// `send_subscribe_nack_from_view` with a 24-byte buf must return + /// `Err(Capacity("udp_buffer"))` without panicking. + #[tokio::test] + async fn send_subscribe_nack_undersized_buf_returns_capacity_not_panic() { + let config = make_config(); + let sd_state = make_sd_state(); + let socket = NullSocket; + let subscriber = subscriber_addr(); + + let (wire, sd_len) = subscribe_wire_bytes(); + let sd_view = sd::SdHeaderView::parse(&wire[..sd_len]).expect("parse"); + let entry_view = sd_view.entries().next().expect("one entry"); + + let result = send_subscribe_nack_from_view( + &mut [0u8; 24], + &config, + &socket, + &sd_state, + &entry_view, + subscriber, + "test_reason", + ) + .await; + + assert!( + matches!(result, Err(Error::Capacity("udp_buffer"))), + "expected Capacity(\"udp_buffer\"), got {result:?}" + ); + } + + /// `send_subscribe_nack_from_view` with a full-size buf must succeed. + #[tokio::test] + async fn send_subscribe_nack_full_size_buf_succeeds() { + let config = make_config(); + let sd_state = make_sd_state(); + let socket = NullSocket; + let subscriber = subscriber_addr(); + + let (wire, sd_len) = subscribe_wire_bytes(); + let sd_view = sd::SdHeaderView::parse(&wire[..sd_len]).expect("parse"); + let entry_view = sd_view.entries().next().expect("one entry"); + + let result = send_subscribe_nack_from_view( + &mut [0u8; crate::UDP_BUFFER_SIZE], + &config, + &socket, + &sd_state, + &entry_view, + subscriber, + "test_reason", + ) + .await; + + assert!(result.is_ok(), "full-size buf must succeed, got {result:?}"); + } +} diff --git a/src/server/sd_state.rs b/src/server/sd_state.rs index 3b321f6d..fd7c5217 100644 --- a/src/server/sd_state.rs +++ b/src/server/sd_state.rs @@ -182,8 +182,13 @@ impl SdStateManager { } /// 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( &self, + buf: &mut [u8], config: &ServerConfig, socket: &T, ) -> Result<(), Error> { @@ -216,14 +221,24 @@ impl SdStateManager { let (sid, reboot_flag) = self.next_session_id_with_reboot_flag(); let sd_payload = sd::Header::new(Flags::new_sd(reboot_flag), &entries, &options); - // Stack-allocated send buffer — alloc-free per-tick path. - // 16-byte SOME/IP header + the SD payload, capped at the UDP - // datagram limit. - let mut buffer = [0u8; crate::UDP_BUFFER_SIZE]; - let sd_data_len = sd_payload.encode_to_slice(&mut buffer[16..])?; - let someip_header = SomeIpHeader::new_sd(sid, sd_data_len); - someip_header.encode_to_slice(&mut buffer[..16])?; + // 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); @@ -234,9 +249,9 @@ impl SdStateManager { config.local_port, total_len ); - crate::log::trace!("OfferService data: {:02X?}", &buffer[..total_len.min(64)]); + crate::log::trace!("OfferService data: {:02X?}", &buf[..total_len.min(64)]); - socket.send_to(&buffer[..total_len], multicast_addr).await?; + socket.send_to(&buf[..total_len], multicast_addr).await?; crate::log::trace!("Sent to {}", multicast_addr); Ok(()) @@ -582,7 +597,7 @@ mod tests { let sock = CapturingSocket::new(); sd_state - .send_offer_service(&config, &sock) + .send_offer_service(&mut [0u8; crate::UDP_BUFFER_SIZE], &config, &sock) .await .expect("send_offer_service should succeed against the mock"); @@ -608,8 +623,8 @@ mod tests { let sd_state = SdStateManager::with_initial(0x1233); let sock = CapturingSocket::new(); - sd_state.send_offer_service(&config, &sock).await.unwrap(); - sd_state.send_offer_service(&config, &sock).await.unwrap(); + 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); @@ -629,8 +644,8 @@ mod tests { // (Continuous). let sd_state = SdStateManager::with_initial(0xFFFE); let sock = CapturingSocket::new(); - sd_state.send_offer_service(&config, &sock).await.unwrap(); - sd_state.send_offer_service(&config, &sock).await.unwrap(); + 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); @@ -662,7 +677,7 @@ mod tests { config.ttl = 0; let sd_state = SdStateManager::with_initial(0x1233); let sock = CapturingSocket::new(); - sd_state.send_offer_service(&config, &sock).await.unwrap(); + 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(); @@ -677,7 +692,7 @@ mod tests { .with_local_port(TEST_ADVERTISED_PORT); let sd_state = SdStateManager::with_initial(0x1233); let sock = FailingSocket; - let result = sd_state.send_offer_service(&config, &sock).await; + 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 @@ -893,7 +908,7 @@ mod tests { // Seed with a recognisable value so on-wire session_id is exact. let sd_state = SdStateManager::with_initial(0x1233); sd_state - .send_offer_service(&config, &tx) + .send_offer_service(&mut [0u8; crate::UDP_BUFFER_SIZE], &config, &tx) .await .expect("send_offer_service should succeed on a configured socket"); @@ -918,8 +933,8 @@ mod tests { let (rx, tx) = mcast_rx_tx().await; let sd_state = SdStateManager::with_initial(0x1233); - sd_state.send_offer_service(&config, &tx).await.unwrap(); - sd_state.send_offer_service(&config, &tx).await.unwrap(); + 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; @@ -941,8 +956,8 @@ mod tests { let (rx, tx) = mcast_rx_tx().await; let sd_state = SdStateManager::with_initial(0xFFFE); - sd_state.send_offer_service(&config, &tx).await.unwrap(); - sd_state.send_offer_service(&config, &tx).await.unwrap(); + 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; @@ -983,7 +998,7 @@ mod tests { let (rx, tx) = mcast_rx_tx().await; let sd_state = SdStateManager::with_initial(0x1233); - sd_state.send_offer_service(&config, &tx).await.unwrap(); + 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); diff --git a/tests/bare_metal_e2e.rs b/tests/bare_metal_e2e.rs index 5ceeff3a..673d3e93 100644 --- a/tests/bare_metal_e2e.rs +++ b/tests/bare_metal_e2e.rs @@ -30,19 +30,22 @@ use simple_someip::PayloadWireFormat; use simple_someip::client::Error as ClientError; use simple_someip::client::{ClientUpdate, ControlMessage, ReceivedMessage, SendMessage}; use simple_someip::define_static_channels; -use simple_someip::e2e::E2ERegistry; +use simple_someip::e2e::{E2EProfile, E2ERegistry, Profile4Config}; use simple_someip::protocol::sd::RebootFlag; use simple_someip::protocol::{ Header, Message, MessageId, MessageType, MessageTypeField, ReturnCode, }; -use simple_someip::server::{ServerConfig, SubscribeError, Subscriber, SubscriptionHandle}; +use simple_someip::server::{ + Error as ServerError, EventPublisher, ServerConfig, SubscribeError, Subscriber, + SubscriptionHandle, +}; use simple_someip::WireFormat; use simple_someip::static_channels::BufferPool; use simple_someip::transport::{ - ReceivedDatagram, SocketOptions, Spawner, StaticBufferProvider, Timer, TransportError, - TransportFactory, TransportSocket, + E2ERegistryHandle, ReceivedDatagram, SocketOptions, Spawner, StaticBufferProvider, Timer, + TransportError, TransportFactory, TransportSocket, }; -use simple_someip::{Client, ClientDeps, RawPayload, Server, ServerDeps, UDP_BUFFER_SIZE}; +use simple_someip::{Client, ClientDeps, E2EKey, RawPayload, Server, ServerDeps, UDP_BUFFER_SIZE}; // ── Static-pool channel factory ─────────────────────────────────────── // @@ -603,7 +606,7 @@ async fn client_send_request_server_runloop_stable() { /// `tools/capture_type_sizes.sh` (thumbv7em). const BM_CLIENT_RUN_FUTURE_BUDGET: usize = 34048; // = ceil64(27224 × 1.25) const BM_CLIENT_SOCKET_LOOP_BUDGET: usize = 1024; // = ceil64(776 × 1.25); receive buffer moved to BufferProvider pool (Tasks 3+4) -const BM_SERVER_RUN_FUTURE_BUDGET: usize = 9664; // = ceil64(7696 × 1.25) +const BM_SERVER_RUN_FUTURE_BUDGET: usize = 4416; // = ceil64(3528 × 1.25); send buffers moved to caller scratch (PR3 T2+T3) #[tokio::test] async fn future_size_witness_bare_metal_channels() { @@ -989,7 +992,7 @@ async fn binding_sockets_claims_one_buffer_each_until_pool_exhausted() { /// - Buffer slot: 40 bytes. /// /// Pre-guard (unprotected) : 36 ≤ 40 → passes. -/// Post-protect guard (before fix): 48 > UDP_BUFFER_SIZE (1400) → false → no guard fires. +/// Post-protect guard (before fix): 48 > UDP_BUFFER_SIZE (1500) → false → no guard fires. /// `copy_from_slice` into buf[16..48] on a 40-byte buf → out-of-bounds panic (RED). /// /// Post-protect guard (after fix): 48 > 40 → true → `Capacity` error returned (GREEN). @@ -1076,6 +1079,26 @@ async fn e2e_protect_expanding_payload_beyond_leased_buffer_returns_capacity_err run_handle.abort(); } +// ── Task 1 (PR 3, #125): SD send-helper buf.len() rejection ────────────────── +// +// The three SD send helpers (`send_unicast_offer`, `send_subscribe_ack_from_view`, +// `send_subscribe_nack_from_view`) are `pub(super)` and therefore not reachable +// from this integration-test crate. The helper-level RED/GREEN tests live in +// `src/server/runtime.rs` under `mod tests`: +// +// - `send_unicast_offer_undersized_buf_returns_capacity` +// - `send_unicast_offer_buf_shorter_than_header_returns_capacity` +// - `send_unicast_offer_full_size_buf_succeeds` +// - `send_subscribe_ack_undersized_buf_returns_capacity_not_panic` +// - `send_subscribe_ack_full_size_buf_succeeds` +// - `send_subscribe_nack_undersized_buf_returns_capacity_not_panic` +// - `send_subscribe_nack_full_size_buf_succeeds` +// +// Task 3 will thread the real caller-owned scratch buffer through `recv_loop` +// → `handle_sd_message` → the helpers, at which point an end-to-end +// integration test here can drive a Subscribe through the server harness with +// a tiny SD send-scratch and assert `Error::Capacity("udp_buffer")`. + /// An empty `VecSdHeader` for building a minimal valid SD message. fn empty_vec_sd_header() -> simple_someip::VecSdHeader { use simple_someip::protocol::sd::{Flags, RebootFlag}; @@ -1085,3 +1108,276 @@ fn empty_vec_sd_header() -> simple_someip::VecSdHeader { options: vec![], } } + +// ── Task 4 (PR 3, #125): server EventPublisher publish paths take caller scratch ─ + +/// Task 4 regression (server-side PR-2 lesson): E2E-protected publish whose +/// expanded payload exceeds the caller-provided `msg_buf` / `protected_buf` +/// must return `Err(ServerError::Capacity("udp_buffer"))`, NOT panic from +/// an out-of-bounds copy. +/// +/// # Why the window is deterministic +/// +/// Profile 4 protect prepends a 12-byte E2E header. With a 20-byte payload: +/// - Unprotected SOME/IP frame: 16 (header) + 20 (payload) = 36 bytes. +/// - Post-protect SOME/IP frame: 16 (header) + (12 + 20) (P4 output) = 48 bytes. +/// - Both scratch buffers: 40 bytes each. +/// +/// Pre-guard (unprotected) : 36 ≤ 40 → passes. +/// Post-protect guard (before fix): 48 > UDP_BUFFER_SIZE (1500) → false → no guard fires. +/// `copy_from_slice` into msg_buf[16..48] on a 40-byte buf → out-of-bounds panic (RED). +/// +/// Post-protect guard (after fix): 48 > 40 → true → `Capacity` returned (GREEN). +#[tokio::test] +async fn e2e_publish_with_undersized_scratch_returns_capacity_not_panic() { + // Construct a bare-metal EventPublisher using mock infrastructure from + // this test file (MockSocket / MockSubscriptions / Arc>). + + // ── Build a MockSocket that discards sends (we never reach the send step) ── + let network = SharedNetwork::new(); + let server_factory = MockFactory { + tx_pipe: Arc::clone(&network.server_to_client), + rx_pipe: Arc::clone(&network.client_to_server), + next_port: Arc::new(Mutex::new(50)), + }; + let socket = server_factory + .bind( + SocketAddrV4::new(Ipv4Addr::LOCALHOST, 0), + &SocketOptions::new(), + ) + .await + .expect("bind mock socket"); + let socket = Arc::new(socket); + + // ── Subscription: one subscriber so we don't short-circuit on "no subs" ── + let subs = MockSubscriptions::default(); + let subscriber_addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, 39999); + subs.subscribe(0xABCD, 1, 0x01, subscriber_addr) + .await + .expect("subscribe"); + + // ── E2E: register Profile 4 for service 0xABCD, method 0x0001 ── + let registry: Arc> = Arc::new(Mutex::new(E2ERegistry::new())); + let e2e_key = E2EKey::new(0xABCD, 0x0001); + registry + .register(e2e_key, E2EProfile::Profile4(Profile4Config::new(0, 15))) + .expect("register E2E key"); + + let publisher: EventPublisher< + Arc>, + MockSubscriptions, + Arc, + MockSocket, + > = EventPublisher::new(subs, socket, registry); + + // ── Build the SOME/IP message with 20-byte payload ── + // Unprotected: 16 + 20 = 36 B ≤ 40 B (fits msg_buf). + // Post-P4-protect: 16 + 12 + 20 = 48 B > 40 B (exceeds msg_buf) → must Capacity. + let service_id: u16 = 0xABCD; + let method_id: u16 = 0x0001; + let payload_bytes = [0x55u8; 20]; + let msg_id = MessageId::new_from_service_and_method(service_id, method_id); + let payload = RawPayload::from_payload_bytes(msg_id, &payload_bytes).expect("create payload"); + let message = Message::::new( + Header::new_event( + service_id, + method_id, + 0x0001_0001, + 1, + 1, + payload_bytes.len(), + ), + payload, + ); + + // ── 40-byte scratch buffers: fits unprotected (36 B), too small for P4 (48 B) ── + let mut msg_buf = [0u8; 40]; + let mut protected_buf = [0u8; 40]; + + let result = publisher + .publish_event_with_buffers( + service_id, + 1, + 0x01, + &message, + &mut msg_buf, + &mut protected_buf, + ) + .await; + + // Must return typed Capacity error — NOT panic from out-of-bounds copy. + assert!( + matches!(result, Err(ServerError::Capacity("udp_buffer"))), + "expected Err(Capacity(\"udp_buffer\")), got {result:?}" + ); +} + +/// Task 4 regression (raw event path): `publish_raw_event_with_buffers` with a +/// buffer too small to hold `16 + payload` must return +/// `Err(ServerError::Capacity("udp_buffer"))`, NOT panic. +/// +/// # Why the window is deterministic +/// +/// SOME/IP header is 16 bytes. With a 10-byte payload: +/// - Frame: 16 + 10 = 26 bytes. +/// - Buffer: 20 bytes. +/// +/// `16 + 10 = 26 > 20` → `Error::Capacity` (RED before guard, GREEN after). +#[tokio::test] +async fn publish_raw_event_with_undersized_buf_returns_capacity_not_panic() { + let network = SharedNetwork::new(); + let server_factory = MockFactory { + tx_pipe: Arc::clone(&network.server_to_client), + rx_pipe: Arc::clone(&network.client_to_server), + next_port: Arc::new(Mutex::new(70)), + }; + let socket = server_factory + .bind( + SocketAddrV4::new(Ipv4Addr::LOCALHOST, 0), + &SocketOptions::new(), + ) + .await + .expect("bind mock socket"); + let socket = Arc::new(socket); + + let subs = MockSubscriptions::default(); + let subscriber_addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, 39998); + subs.subscribe(0xABCD, 1, 0x01, subscriber_addr) + .await + .expect("subscribe"); + + let registry: Arc> = Arc::new(Mutex::new(E2ERegistry::new())); + + let publisher: EventPublisher< + Arc>, + MockSubscriptions, + Arc, + MockSocket, + > = EventPublisher::new(subs, socket, registry); + + // 20-byte buf, 10-byte payload: 16 + 10 = 26 > 20 → must Capacity. + let mut buf = [0u8; 20]; + let payload = [0xAA_u8; 10]; + + let result = publisher + .publish_raw_event_with_buffers( + 0xABCD, + 1, + 0x01, + 0x8001, + 0x0001_0001, + 1, + 1, + &payload, + &mut buf, + ) + .await; + + assert!( + matches!(result, Err(ServerError::Capacity("udp_buffer"))), + "expected Err(Capacity(\"udp_buffer\")), got {result:?}" + ); + + // #133 review: empty payload + sub-header buffer must ALSO return + // Capacity (not a protocol I/O error). The `payload.len() > + // buf.len() - 16` guard saturates to `0 > 0` = false here, so this + // path relies on the explicit `buf.len() < 16` guard. + let mut tiny = [0u8; 10]; + let empty: [u8; 0] = []; + let result = publisher + .publish_raw_event_with_buffers( + 0xABCD, 1, 0x01, 0x8001, 0x0001_0001, 1, 1, &empty, &mut tiny, + ) + .await; + assert!( + matches!(result, Err(ServerError::Capacity("udp_buffer"))), + "sub-16 buffer + empty payload must Capacity, got {result:?}" + ); +} + +/// Task 4 future-size witness: measure the size of a `publish_event_with_buffers` +/// future constructed with bare-metal channel / mock infrastructure + caller +/// buffers. This is the app's future (separate from `run_combined`), so it does +/// NOT appear in the `bm_server_run_future` witness. +/// +/// The budget is `ceil64(measured × 1.25)`. Update this constant when the +/// implementation changes (and verify on thumbv7em with `tools/capture_type_sizes.sh`). +/// +/// # Budget rationale +/// +/// With caller-provided scratch (PR-3 #125), the future no longer holds +/// two `[u8; UDP_BUFFER_SIZE]` arrays — those live in the app's stack frame +/// instead. The future retains only the subscriber snapshot +/// (`HeaplessVec`) and the E2E + socket +/// handle clones, which are pointer-sized. +/// Budget: ceil64(320 B × 1.25) = 448 B (x86-64 host measurement). +/// Before PR-3 #125 scratch-extraction, the future held two `[u8; 1500]` arrays +/// in-future: ~3320 B. After: caller holds the arrays; future is ~320 B (host). +const BM_SERVER_PUBLISH_FUTURE_BUDGET: usize = 448; // = ceil64(320 × 1.25) + +#[tokio::test] +async fn future_size_witness_bm_server_publish_future() { + // ── Build minimal bare-metal-flavored infrastructure ── + let network = SharedNetwork::new(); + let server_factory = MockFactory { + tx_pipe: Arc::clone(&network.server_to_client), + rx_pipe: Arc::clone(&network.client_to_server), + next_port: Arc::new(Mutex::new(60)), + }; + let socket = server_factory + .bind( + SocketAddrV4::new(Ipv4Addr::LOCALHOST, 0), + &SocketOptions::new(), + ) + .await + .expect("bind mock socket"); + let socket = Arc::new(socket); + + let subs = MockSubscriptions::default(); + let registry: Arc> = Arc::new(Mutex::new(E2ERegistry::new())); + + let publisher: EventPublisher< + Arc>, + MockSubscriptions, + Arc, + MockSocket, + > = EventPublisher::new(subs, socket, registry); + + // ── Build the message payload ── + let service_id: u16 = 0x1234; + let method_id: u16 = 0x0001; + let payload_bytes = [0u8; 20]; + let msg_id = MessageId::new_from_service_and_method(service_id, method_id); + let payload = RawPayload::from_payload_bytes(msg_id, &payload_bytes).expect("create payload"); + let message = Message::::new( + Header::new_event(service_id, method_id, 0x0001_0001, 1, 1, payload_bytes.len()), + payload, + ); + + // ── Caller-provided scratch buffers (simulate app-side static arrays) ── + let mut msg_buf = [0u8; UDP_BUFFER_SIZE]; + let mut protected_buf = [0u8; UDP_BUFFER_SIZE]; + + // Construct the future WITHOUT awaiting it so we can measure its size. + let publish_future = publisher.publish_event_with_buffers( + service_id, + 1, + 0x01, + &message, + &mut msg_buf, + &mut protected_buf, + ); + + let future_size = core::mem::size_of_val(&publish_future); + // Drop the future (do not drive it — no real subscribers in this witness). + drop(publish_future); + + println!("FUTURE_SIZE bm_server_publish_future {future_size}"); + + assert!( + future_size <= BM_SERVER_PUBLISH_FUTURE_BUDGET, + "publish future grew: {future_size} B > budget {BM_SERVER_PUBLISH_FUTURE_BUDGET} B — \ + update BM_SERVER_PUBLISH_FUTURE_BUDGET in tests/bare_metal_e2e.rs after verifying \ + the new size is acceptable" + ); +}