Skip to main content

nros_node/executor/
action_core.rs

1//! Type-agnostic action protocol core types.
2//!
3//! [`ActionServerCore`] and [`ActionClientCore`] handle the raw-bytes
4//! action protocol (GoalUuid framing, status publishing, result slab)
5//! without requiring `RosAction` type parameters. The typed
6//! [`ActionServer`](super::handles::ActionServer) and
7//! [`ActionClient`](super::handles::ActionClient) wrap these cores
8//! and add serialization/deserialization at the boundary.
9
10use nros_core::{
11    CdrReader, CdrWriter, GoalInfo, GoalStatusCode, GoalStatusStamped, GoalUuid, Serialize, Time,
12};
13use nros_rmw::{ClientTrait, Publisher, ServiceTrait, Subscription, TransportError};
14
15use super::types::NodeError;
16use crate::session;
17
18/// Scratch buffer for a decoded CancelGoal request. Cancel payloads are
19/// tiny (a `GoalUuid` + a `builtin_interfaces/Time` stamp), so a fixed
20/// 256-byte buffer covers them without a const-generic parameter like the
21/// goal/result/feedback slabs.
22pub(crate) const CANCEL_BUF: usize = 256;
23
24/// DDS type name of an action's `send_goal` / `get_result` service.
25///
26/// ROS 2 names the action's two services by their per-channel service types —
27/// `<Action>_SendGoal` / `<Action>_GetResult` — **not** the bare action type
28/// (`<Action>`). A real `rcl_action` peer matches on those, so advertising the
29/// bare action type leaves our send_goal/get_result services undiscovered and
30/// every goal times out.
31///
32/// The codegen emits each channel's *request* envelope type name in DDS form,
33/// e.g. `example_interfaces::action::dds_::Fibonacci_SendGoal_Request_`. The
34/// service layer (`xrce_dds_request_type` / the Zenoh shim) extends a base type
35/// ending in `_` with `Request_` / `Response_`, so the base we must pass is the
36/// request type name with its trailing `Request_` stripped:
37/// `…Fibonacci_SendGoal_Request_` → `…Fibonacci_SendGoal_`. Falls back to the
38/// bare action type if the request name has an unexpected shape.
39pub(crate) fn action_service_base_type<'a>(
40    request_type_name: &'a str,
41    fallback_action_type: &'a str,
42) -> &'a str {
43    request_type_name
44        .strip_suffix("Request_")
45        .unwrap_or(fallback_action_type)
46}
47
48/// The per-channel DDS type name for a RAW-registered action server.
49///
50/// The typed path derives these from `A::SendGoalRequest::TYPE_NAME` via
51/// `action_service_base_type`. The raw path (`register_action_server_raw*`)
52/// has only the BARE action type, so it must construct them — and until
53/// phase-338 W3 it did not, advertising `…Fibonacci_` on send_goal / get_result
54/// / feedback where ROS 2 expects `…Fibonacci_SendGoal_`,
55/// `…Fibonacci_GetResult_` and `…Fibonacci_FeedbackMessage_`. The type name is
56/// baked into the keyexpr, so a client's query never matched the server's
57/// queryable and every goal timed out with `Transport(Timeout)` — exactly the
58/// failure `action_service_base_type`'s own doc warns about, on the other
59/// registration path.
60///
61/// `action_type` is DDS-form and ends in `_` (`…::dds_::Fibonacci_`); the
62/// result replaces that suffix with `_<Channel>_`. Returns the bare type
63/// unchanged if it has an unexpected shape, matching the typed path's
64/// fallback.
65pub fn action_channel_type<const N: usize>(
66    action_type: &str,
67    channel: &str,
68) -> heapless::String<N> {
69    let mut out: heapless::String<N> = heapless::String::new();
70    let base = action_type.strip_suffix('_').unwrap_or(action_type);
71    if out.push_str(base).is_err()
72        || out.push('_').is_err()
73        || out.push_str(channel).is_err()
74        || out.push('_').is_err()
75    {
76        let mut fallback: heapless::String<N> = heapless::String::new();
77        let _ = fallback.push_str(action_type);
78        return fallback;
79    }
80    out
81}
82
83/// Scratch buffer for serializing a `GoalStatusArray` before publishing it
84/// on the status topic. It is a local in
85/// [`publish_status_array`](crate::executor::ffi::ActionServerCoreFfi::publish_status_array), so it is
86/// stack, not static — which is why it is not simply made large enough for any
87/// `MAX_GOALS` anyone might pick. What it must hold is stated exactly by
88/// [`status_array_bytes`] and checked at COMPILE time per instantiation; see
89/// [`ActionServerCore::STATUS_ARRAY_FITS`].
90pub(crate) const STATUS_ARRAY_BUF: usize = 512;
91
92/// Bytes one `GoalStatusStamped` occupies when another entry follows it.
93///
94/// `GoalInfo` is a fixed `uint8[16]` (16 unaligned bytes) + `int32 sec` +
95/// `uint32 nanosec`, then `int8 status` — 25 bytes — and the NEXT entry's `sec`
96/// forces a 4-byte alignment pad, so the stride is 28. MEASURED against the
97/// real `CdrWriter` by `status_array_entry_layout_is_measured`, not
98/// asserted from this comment: the const and the serializer have to agree or
99/// the compile-time bound below is checking the wrong number.
100pub(crate) const STATUS_ENTRY_STRIDE: usize = 28;
101
102/// The LAST entry carries no trailing alignment pad, so it is 3 bytes shorter
103/// than [`STATUS_ENTRY_STRIDE`].
104pub(crate) const STATUS_ENTRY_TAIL: usize = 25;
105
106/// CDR encapsulation header (4) + the sequence length prefix (4).
107pub(crate) const STATUS_ARRAY_PREAMBLE: usize = 8;
108
109/// Exact serialized size of a `GoalStatusArray` carrying `entries` entries.
110pub(crate) const fn status_array_bytes(entries: usize) -> usize {
111    if entries == 0 {
112        STATUS_ARRAY_PREAMBLE
113    } else {
114        STATUS_ARRAY_PREAMBLE + STATUS_ENTRY_STRIDE * (entries - 1) + STATUS_ENTRY_TAIL
115    }
116}
117
118// ============================================================================
119// Supporting types
120// ============================================================================
121
122/// Goal tracked by the core — GoalUuid, status and acceptance stamp; no typed data.
123#[derive(Clone, Copy)]
124pub struct RawActiveGoal {
125    /// Goal ID.
126    pub goal_id: GoalUuid,
127    /// Current status.
128    pub status: GoalStatusCode,
129    /// Issue 2029 — when the server ACCEPTED the goal, on the ROS clock
130    /// (`Clock::ros_time()`: `/clock` when a source drives it, the port's wall
131    /// clock otherwise — rcl_action stamps with the server's clock the same
132    /// way). A `CancelGoal` request with a stamp cancels the goals accepted at
133    /// or before it, and every `GoalInfo` this server writes carries it. A
134    /// board with no wall clock stamps zero, so a stamped cancel there selects
135    /// every goal.
136    pub stamp: Time,
137}
138
139/// A `get_result` request held until its goal terminates (Phase 237).
140///
141/// `sequence_number` is the service-backend reply-correlation token; the backend
142/// must be able to `send_response(sequence_number, …)` after the handler returned
143/// (Cyclone native; XRCE/Zenoh via the Phase 237 seq-keyed reply tables).
144#[derive(Clone, Copy)]
145pub struct PendingGetResult {
146    /// Goal whose terminal result the requester is waiting for.
147    pub goal_id: GoalUuid,
148    /// Backend reply-correlation token for the deferred `send_response`.
149    pub sequence_number: i64,
150}
151
152/// Completed goal result metadata — indexes into the result slab.
153///
154/// Entries are kept in **completion order**, which is also **increasing
155/// `offset` order**: results are appended at `result_slab_used` and the slab is
156/// compacted in place whenever an entry is reclaimed. Every reclamation path
157/// preserves that ordering (`heapless::Vec::remove` / `retain`, never
158/// `swap_remove`) because `ActionServerCore::compact_result_slab` moves
159/// survivors *down* and would clobber a not-yet-moved entry otherwise.
160#[derive(Clone, Copy)]
161pub struct CompletedResultEntry {
162    /// Unique identifier for the completed goal.
163    pub goal_id: GoalUuid,
164    /// Terminal status of the goal.
165    pub status: GoalStatusCode,
166    /// Byte offset into the result slab.
167    pub offset: usize,
168    /// Length of the serialised result in bytes.
169    pub len: usize,
170    /// `true` once a `get_result` reply carrying this result has been sent —
171    /// either the deferred flush in
172    /// [`ActionServerCore::complete_goal_raw`](crate::executor::ffi::ActionServerCoreFfi::complete_goal_raw) or the immediate reply in
173    /// [`ActionServerCore::try_handle_get_result_raw`](crate::executor::ffi::ActionServerCoreFfi::try_handle_get_result_raw).
174    ///
175    /// Issue 0796: this is the reclamation priority. A delivered result has
176    /// served its purpose (rcl would let its `result_timeout` retire it), so it
177    /// is evicted before any result nobody has fetched yet.
178    pub delivered: bool,
179}
180
181/// Phase 122.3.c.6.d — information about a peeked cancel-goal
182/// request. Returned by
183/// [`ActionServerCore::try_recv_cancel_request`](crate::executor::ffi::ActionServerCoreFfi::try_recv_cancel_request).
184pub struct PendingCancelRequest {
185    /// The goal_id named in the cancel request.
186    pub goal_id: GoalUuid,
187    /// Service sequence number — pass back to
188    /// [`ActionServerCore::send_cancel_reply`](crate::executor::ffi::ActionServerCoreFfi::send_cancel_reply).
189    pub sequence_number: i64,
190    /// Snapshot of the goal's current status at peek time
191    /// (`GoalStatusCode::Unknown` if no matching active goal).
192    pub current_status: GoalStatusCode,
193}
194
195/// Information about a received goal request.
196pub struct RawGoalRequest {
197    /// The parsed goal ID.
198    pub goal_id: GoalUuid,
199    /// Sequence number for the service reply.
200    pub sequence_number: i64,
201    /// Offset into the goal buffer where the CDR payload begins.
202    /// Backends may prepend a sequence-number header (DDS) or place
203    /// the payload at offset 0 (zenoh).
204    pub data_offset: usize,
205    /// Length of valid CDR data starting at `data_offset`.
206    pub data_len: usize,
207}
208
209// ============================================================================
210// GoalUuid CDR helpers
211// ============================================================================
212
213/// Read a GoalUuid from a CDR reader as a fixed `uint8[16]` array.
214///
215/// ROS 2 actions carry the goal id as `unique_identifier_msgs/UUID`, whose
216/// single field is a **fixed-size** `uint8[16]` array — CDR fixed arrays have
217/// **no** length prefix. We must read exactly 16 bytes with no leading count,
218/// matching `unique_identifier_msgs::msg::UUID::deserialize`. (The pre-233.6
219/// framing wrote a `u32(16)` sequence prefix, which self-matched nano-ros peers
220/// but added 4 bytes a real `rcl_action` peer rejects.)
221fn read_goal_id(reader: &mut CdrReader<'_>) -> Result<GoalUuid, NodeError> {
222    let mut goal_id = GoalUuid::default();
223    for byte in &mut goal_id.uuid {
224        *byte = reader
225            .read_u8()
226            .map_err(|_| NodeError::Transport(TransportError::DeserializationError))?;
227    }
228    Ok(goal_id)
229}
230
231/// Write a GoalUuid into a CDR writer as a fixed `uint8[16]` array (no length
232/// prefix) — see [`read_goal_id`] for why the prefix must be absent.
233fn write_goal_id(writer: &mut CdrWriter<'_>, goal_id: &GoalUuid) -> Result<(), NodeError> {
234    for b in &goal_id.uuid {
235        writer.write_u8(*b).map_err(|_| NodeError::Serialization)?;
236    }
237    Ok(())
238}
239
240// ============================================================================
241// ActionServerCore
242// ============================================================================
243
244/// Type-agnostic action server core handling the raw-bytes protocol.
245///
246/// Manages active goal tracking (GoalUuid + status), completed result storage
247/// in a fixed-size slab, and all CDR framing for the action protocol.
248///
249/// The typed [`ActionServer`](super::handles::ActionServer) wraps this
250/// and adds `A::Goal` / `A::Feedback` / `A::Result` (de)serialization.
251pub struct ActionServerCore<
252    const GOAL_BUF: usize = { crate::config::DEFAULT_RX_BUF_SIZE },
253    const RESULT_BUF: usize = { crate::config::DEFAULT_RX_BUF_SIZE },
254    const FEEDBACK_BUF: usize = { crate::config::DEFAULT_RX_BUF_SIZE },
255    const MAX_GOALS: usize = 4,
256> {
257    pub(crate) send_goal_server: session::RmwServiceServer,
258    pub(crate) cancel_goal_server: session::RmwServiceServer,
259    pub(crate) get_result_server: session::RmwServiceServer,
260    pub(crate) feedback_publisher: session::RmwPublisher,
261    pub(crate) status_publisher: session::RmwPublisher,
262    pub(crate) active_goals: heapless::Vec<RawActiveGoal, MAX_GOALS>,
263    pub(crate) completed_results: heapless::Vec<CompletedResultEntry, MAX_GOALS>,
264    /// `get_result` requests that arrived while their goal was still active
265    /// (Phase 237). `rclcpp_action` sends `get_result` immediately after
266    /// acceptance and expects the reply only once the goal terminates, so we
267    /// hold the request's correlation token (`sequence_number`) here and flush
268    /// it in [`Self::complete_goal_raw`]. Deferral relies on the service
269    /// backend honoring `send_response(seq)` after the handler returns — the
270    /// seq-keyed reply contract (Cyclone native; XRCE/Zenoh per Phase 237).
271    pub(crate) pending_get_results: heapless::Vec<PendingGetResult, MAX_GOALS>,
272    /// Slab storage for completed result CDR bytes.
273    ///
274    /// A bump region whose survivors are compacted on reclamation — see
275    /// [`CompletedResultEntry`] and
276    /// [`reserve_result_space`](Self::reserve_result_space).
277    pub(crate) result_slab: [u8; RESULT_BUF],
278    /// Bytes of [`result_slab`](Self::result_slab) currently held by the
279    /// entries in `completed_results`. Compaction keeps the live bytes in
280    /// `[0, result_slab_used)` with no holes, so this is exactly the sum of the
281    /// entries' `len`.
282    pub(crate) result_slab_used: usize,
283    pub(crate) goal_buffer: [u8; GOAL_BUF],
284    pub(crate) feedback_buffer: [u8; FEEDBACK_BUF],
285    pub(crate) cancel_buffer: [u8; CANCEL_BUF],
286}
287
288impl<
289    const GOAL_BUF: usize,
290    const RESULT_BUF: usize,
291    const FEEDBACK_BUF: usize,
292    const MAX_GOALS: usize,
293> ActionServerCore<GOAL_BUF, RESULT_BUF, FEEDBACK_BUF, MAX_GOALS>
294{
295    /// Phase 122.3.c.6.b — construct an `ActionServerCore` from the
296    /// 5 already-built transport channels. Caller (typically the C
297    /// API's `rclc_polling_action_server_init`) owns wiring the
298    /// channels via the session's `create_*` methods.
299    pub(crate) fn from_channels(
300        send_goal_server: session::RmwServiceServer,
301        cancel_goal_server: session::RmwServiceServer,
302        get_result_server: session::RmwServiceServer,
303        feedback_publisher: session::RmwPublisher,
304        status_publisher: session::RmwPublisher,
305    ) -> Self {
306        Self {
307            send_goal_server,
308            cancel_goal_server,
309            get_result_server,
310            feedback_publisher,
311            status_publisher,
312            active_goals: heapless::Vec::new(),
313            completed_results: heapless::Vec::new(),
314            pending_get_results: heapless::Vec::new(),
315            result_slab: [0u8; RESULT_BUF],
316            result_slab_used: 0,
317            goal_buffer: [0u8; GOAL_BUF],
318            feedback_buffer: [0u8; FEEDBACK_BUF],
319            cancel_buffer: [0u8; CANCEL_BUF],
320        }
321    }
322
323    /// Try to receive a goal request from the send_goal service.
324    ///
325    /// Returns the parsed GoalUuid, sequence number, and data length.
326    /// The full CDR data (including GoalUuid) remains in `goal_buffer`.
327    pub(crate) fn try_recv_goal_request(&mut self) -> Result<Option<RawGoalRequest>, NodeError> {
328        // Capture buf base ptr before borrowing through `take_request`
329        // so we can recover the data offset after the borrow ends.
330        // DDS-style backends place a sequence-number prefix before the
331        // CDR payload; reading the buffer from offset 0 unconditionally
332        // would feed the prefix bytes to the deserializer.
333        let buf_start = self.goal_buffer.as_ptr() as usize;
334        // Phase 120: NoData (no pending request) is the steady-state
335        // expected condition — collapse it to `Ok(None)` instead of
336        // surfacing as ServiceRequestFailed. Any other transport
337        // error remains ServiceRequestFailed.
338        let request = match self.send_goal_server.take_request(&mut self.goal_buffer) {
339            Ok(opt) => opt,
340            Err(TransportError::NoData) => return Ok(None),
341            // issue 1088 — pending but no free reply slot: keep the kind.
342            Err(TransportError::WouldBlock) => {
343                return Err(NodeError::Transport(TransportError::WouldBlock));
344            }
345            Err(_) => return Err(NodeError::Transport(TransportError::ServiceRequestFailed)),
346        };
347
348        let request = match request {
349            Some(r) => r,
350            None => return Ok(None),
351        };
352
353        let data_offset = (request.data.as_ptr() as usize).saturating_sub(buf_start);
354        let data_len = request.data.len();
355        let sequence_number = request.sequence_number;
356        #[allow(clippy::drop_non_drop)]
357        drop(request);
358
359        let mut reader =
360            CdrReader::new_with_header(&self.goal_buffer[data_offset..data_offset + data_len])
361                .map_err(|_| NodeError::Transport(TransportError::DeserializationError))?;
362
363        let goal_id = read_goal_id(&mut reader)?;
364
365        Ok(Some(RawGoalRequest {
366            goal_id,
367            sequence_number,
368            data_offset,
369            data_len,
370        }))
371    }
372
373    /// Get a reference to the goal buffer (valid after `try_recv_goal_request`).
374    pub(crate) fn goal_buffer(&self) -> &[u8] {
375        &self.goal_buffer
376    }
377
378    /// Accept a goal: records it, sends the acceptance reply, publishes status.
379    ///
380    /// # Ordering
381    ///
382    /// The goal is recorded in `active_goals` **before** the reply goes out,
383    /// and the recording is rolled back if the reply fails. Issue 0322: this
384    /// used to reply first and then `let _ = self.active_goals.push(...)`, so
385    /// once `MAX_GOALS` (default 4) were active, a 5th `send_goal` was
386    /// answered `accepted=true` and then dropped — no execution, no feedback,
387    /// no result, no terminal status. An rclcpp/rclpy client that saw
388    /// `accepted=true` waited on its result future forever.
389    ///
390    /// A full table is now answered truthfully with `accepted=false` via
391    /// [`Self::reject_goal`], which is a contract the client already handles.
392    pub(crate) fn accept_goal(&mut self, goal_id: GoalUuid, seq: i64) -> Result<(), NodeError> {
393        // Serialize first: a serialization failure here must leave no trace,
394        // and nothing below depends on the table.
395        // Issue 2029 — the acceptance stamp: written into the reply (upstream's
396        // `SendGoal_Response.stamp`) and kept for cancel-before.
397        let stamp = Self::accept_stamp();
398        let mut writer =
399            crate::tx_writer(&mut self.cancel_buffer).map_err(|_| NodeError::BufferTooSmall)?;
400        writer.write_u8(1).map_err(|_| NodeError::Serialization)?;
401        writer
402            .write_i32(stamp.sec)
403            .map_err(|_| NodeError::Serialization)?;
404        writer
405            .write_u32(stamp.nanosec)
406            .map_err(|_| NodeError::Serialization)?;
407        let reply_len = writer.position();
408
409        // Capacity is decided BEFORE anything reaches the wire, so a full
410        // table becomes an honest rejection rather than a lie.
411        if self
412            .active_goals
413            .push(RawActiveGoal {
414                goal_id,
415                status: GoalStatusCode::Accepted,
416                stamp,
417            })
418            .is_err()
419        {
420            return self.reject_goal(seq);
421        }
422
423        if self
424            .send_goal_server
425            .send_response(seq, &self.cancel_buffer[..reply_len])
426            .is_err()
427        {
428            // The client never learned it was accepted, so un-record it —
429            // otherwise the slot leaks and lowers the effective capacity for
430            // every later goal. `pop` removes the entry pushed just above:
431            // `&mut self` means nothing else can have touched the table.
432            self.active_goals.pop();
433            return Err(NodeError::ServiceReplyFailed);
434        }
435
436        // Past this point the acceptance is on the wire and irreversible.
437        //
438        // The status-array publish is therefore NOT propagated: both C and C++
439        // callers collapse `Err` to a generic error code
440        // (`nros-c/src/action/server.rs`, `nros-cpp/src/action.rs`), so
441        // returning one here would report "accept failed" for a goal that IS
442        // accepted and running — inviting the caller to reject or retry it.
443        // The client already holds `accepted=true` and will still receive the
444        // result; a missed status sample is degraded, not broken. Issue 0322
445        // proposed propagating this; see that issue for why it is deliberately
446        // not done.
447        let _status = self.publish_status_array();
448        Ok(())
449    }
450
451    /// Issue 2029 — "now" for a goal being accepted, on the ROS clock (see
452    /// [`RawActiveGoal::stamp`]).
453    fn accept_stamp() -> Time {
454        nros_core::clock::Clock::ros_time().now()
455    }
456
457    /// Reject a goal: sends the rejection reply.
458    pub(crate) fn reject_goal(&mut self, seq: i64) -> Result<(), NodeError> {
459        // Serialize response: accepted=false + stamp
460        let mut writer =
461            crate::tx_writer(&mut self.cancel_buffer).map_err(|_| NodeError::BufferTooSmall)?;
462        writer.write_u8(0).map_err(|_| NodeError::Serialization)?;
463        writer.write_i32(0).map_err(|_| NodeError::Serialization)?;
464        writer.write_u32(0).map_err(|_| NodeError::Serialization)?;
465        let reply_len = writer.position();
466
467        self.send_goal_server
468            .send_response(seq, &self.cancel_buffer[..reply_len])
469            .map_err(|_| NodeError::ServiceReplyFailed)
470    }
471
472    /// Publish feedback with raw CDR bytes.
473    ///
474    /// Writes GoalUuid framing + raw feedback bytes into the feedback buffer
475    /// and publishes.
476    pub(crate) fn publish_feedback_raw(
477        &mut self,
478        goal_id: &GoalUuid,
479        feedback_cdr: &[u8],
480    ) -> Result<(), NodeError> {
481        // CDR header + GoalUuid + feedback must fit in FEEDBACK_BUF. Issue 2014 —
482        // the framing from `action_frame`; this read `4 + 20`, the pre-233.6
483        // GoalUuid with a sequence-length prefix.
484        let needed = crate::action_frame::FEEDBACK_PAYLOAD_OFFSET + feedback_cdr.len();
485        if needed > FEEDBACK_BUF {
486            return Err(NodeError::BufferTooSmall);
487        }
488
489        let mut writer =
490            crate::tx_writer(&mut self.feedback_buffer).map_err(|_| NodeError::BufferTooSmall)?;
491
492        write_goal_id(&mut writer, goal_id)?;
493
494        // Copy raw feedback bytes directly after GoalUuid
495        let pos = writer.position();
496        if pos + feedback_cdr.len() > FEEDBACK_BUF {
497            return Err(NodeError::BufferTooSmall);
498        }
499        self.feedback_buffer[pos..pos + feedback_cdr.len()].copy_from_slice(feedback_cdr);
500        let len = pos + feedback_cdr.len();
501
502        self.feedback_publisher
503            .publish_raw(&self.feedback_buffer[..len])
504            .map_err(|_| NodeError::Transport(TransportError::PublishFailed))
505    }
506
507    /// Set a goal's status and publish the updated GoalStatusArray.
508    pub(crate) fn set_goal_status(&mut self, goal_id: &GoalUuid, status: GoalStatusCode) {
509        for goal in &mut self.active_goals {
510            if goal.goal_id.uuid == goal_id.uuid {
511                goal.status = status;
512                break;
513            }
514        }
515        let _ = self.publish_status_array();
516    }
517
518    /// Compact the result slab: walk the retained entries in offset order,
519    /// move each survivor down to the first free byte, and rewrite its offset.
520    ///
521    /// Issue 0796. `completed_results` is ordered by `offset` (see
522    /// [`CompletedResultEntry`]), so `write <= entry.offset` at every step and
523    /// the `copy_within` can never clobber an entry that has not moved yet.
524    /// Bounded work: at most `MAX_GOALS` (default 4) moves totalling at most
525    /// `RESULT_BUF` bytes, and only on a reclamation.
526    fn compact_result_slab(&mut self) {
527        let mut write = 0usize;
528        for entry in self.completed_results.iter_mut() {
529            debug_assert!(
530                write <= entry.offset,
531                "completed_results must stay in increasing-offset order"
532            );
533            if entry.offset != write {
534                self.result_slab
535                    .copy_within(entry.offset..entry.offset + entry.len, write);
536                entry.offset = write;
537            }
538            write += entry.len;
539        }
540        self.result_slab_used = write;
541    }
542
543    /// Drop the retained result for `goal_id`, if any, and compact.
544    /// Returns `true` when an entry was dropped.
545    fn drop_completed_result(&mut self, goal_id: &GoalUuid) -> bool {
546        match self
547            .completed_results
548            .iter()
549            .position(|e| e.goal_id.uuid == goal_id.uuid)
550        {
551            Some(idx) => {
552                self.completed_results.remove(idx);
553                self.compact_result_slab();
554                true
555            }
556            None => false,
557        }
558    }
559
560    /// Reclaim exactly one completed result. Returns `false` when there is
561    /// nothing left to reclaim.
562    ///
563    /// **Eviction policy** (issue 0796). rcl keeps a terminated goal's result
564    /// until it has been collected *and* a per-goal `result_timeout` elapses,
565    /// then `rcl_action_expire_goals()` reclaims it. We have no clock in the
566    /// core — every RTOS port spells one differently and the core is `no_std`
567    /// with no time source threaded through it — so reclamation is **on demand
568    /// and priority-ordered** instead of timed:
569    ///
570    /// 1. the oldest **delivered** result (a client already has these bytes;
571    ///    this is the case rcl's timeout is really for), else
572    /// 2. the oldest result overall.
573    ///
574    /// Rule 1 means a fetched result never pins storage. Rule 2 means a client
575    /// that asks for nothing cannot wedge the server: its stale result is
576    /// displaced by newer ones rather than blocking every later goal. A goal
577    /// evicted under rule 2 whose client asks *later* is answered
578    /// `GoalStatusCode::Unknown` with the default result by
579    /// [`try_handle_get_result_raw`](Self::try_handle_get_result_raw) — degraded,
580    /// but an answer, where the pre-0796 code left the requester hanging
581    /// forever.
582    fn evict_one_completed_result(&mut self) -> bool {
583        if self.completed_results.is_empty() {
584            return false;
585        }
586        let idx = self
587            .completed_results
588            .iter()
589            .position(|e| e.delivered)
590            .unwrap_or(0);
591        self.completed_results.remove(idx);
592        self.compact_result_slab();
593        true
594    }
595
596    /// Make room for one more entry of `needed` bytes, reclaiming completed
597    /// results as required.
598    ///
599    /// `Err(NodeError::BufferTooSmall)` means the result can never be retained
600    /// because it exceeds `RESULT_BUF` outright — an empty slab would not hold
601    /// it either, so the caller must raise the action server's `RESULT_BUF`.
602    /// That is the ONLY failure: a full slab or a full entry table is
603    /// reclaimed, not refused.
604    fn reserve_result_space(&mut self, needed: usize) -> Result<(), NodeError> {
605        if needed > RESULT_BUF {
606            return Err(NodeError::BufferTooSmall);
607        }
608        loop {
609            if self.completed_results.len() < MAX_GOALS
610                && RESULT_BUF - self.result_slab_used >= needed
611            {
612                return Ok(());
613            }
614            if !self.evict_one_completed_result() {
615                // Unreachable while `needed <= RESULT_BUF` and `MAX_GOALS >= 1`,
616                // but the loop must not spin on a degenerate instantiation.
617                return Err(NodeError::BufferTooSmall);
618            }
619        }
620    }
621
622    /// Mark `goal_id`'s retained result as fetched, making it the first
623    /// candidate for reclamation.
624    fn mark_result_delivered(&mut self, goal_id: &GoalUuid) {
625        for entry in self.completed_results.iter_mut() {
626            if entry.goal_id.uuid == goal_id.uuid {
627                entry.delivered = true;
628                break;
629            }
630        }
631    }
632
633    /// Reclaim every completed result whose `get_result` reply has already been
634    /// sent — nano-ros's analogue of `rcl_action_expire_goals()`. Returns the
635    /// number of entries reclaimed.
636    ///
637    /// Calling this is **optional**: [`complete_goal_raw`](Self::complete_goal_raw)
638    /// reclaims on demand, so a server that never calls it still runs forever.
639    /// It exists for a server that would rather return the memory eagerly (e.g.
640    /// before a long idle period) than at the next completion.
641    pub(crate) fn expire_completed_results(&mut self) -> usize {
642        let before = self.completed_results.len();
643        self.completed_results.retain(|e| !e.delivered);
644        let removed = before - self.completed_results.len();
645        if removed > 0 {
646            self.compact_result_slab();
647        }
648        removed
649    }
650
651    /// Number of completed results currently retained.
652    pub(crate) fn completed_result_count(&self) -> usize {
653        self.completed_results.len()
654    }
655
656    /// Bytes of the result slab currently held by retained results.
657    pub(crate) fn result_slab_used(&self) -> usize {
658        self.result_slab_used
659    }
660
661    /// Is `goal_id` a goal this server still knows about?
662    ///
663    /// rcl's `rcl_action_server_goal_exists` — phase-417 W4.b, ledger row
664    /// `c:action_server_goal_exists`. rcl searches the server's goal-handle
665    /// table, which retains a goal after it terminates until
666    /// `rcl_action_expire_goals` reclaims it, so a terminated-but-unreclaimed
667    /// goal EXISTS there. Ours asks the same question over the two places a
668    /// goal can live here: the active set, and the completed-result slab.
669    ///
670    /// Both halves are load-bearing, and that is why this row could not be
671    /// closed as a documentation fix pointing at the status lookup. A status
672    /// lookup reads the ACTIVE set only, so it answers "not found" — which the
673    /// C surface maps to `RCLC_RET_NOT_FOUND` — for a goal that has completed
674    /// and whose result is still sitting there to be fetched. "Not found" and
675    /// "does not exist" therefore disagreed over exactly the window in which a
676    /// client is expected to call `get_result`.
677    pub(crate) fn goal_exists(&self, goal_id: &GoalUuid) -> bool {
678        self.active_goals
679            .iter()
680            .any(|g| g.goal_id.uuid == goal_id.uuid)
681            || self.has_completed_result(goal_id)
682    }
683
684    /// `true` while `goal_id`'s completed result is still retained (i.e. a
685    /// `get_result` for it would be answered from the slab rather than as
686    /// `Unknown`).
687    pub(crate) fn has_completed_result(&self, goal_id: &GoalUuid) -> bool {
688        self.completed_results
689            .iter()
690            .any(|e| e.goal_id.uuid == goal_id.uuid)
691    }
692
693    /// Complete a goal: remove from active, store raw result CDR in slab,
694    /// publish status.
695    ///
696    /// The goal moves from `active_goals` to `completed_results`, and the
697    /// status array published at the end of this function carries BOTH lists
698    /// (issue 1361), so the terminal status is on the wire. See
699    /// [`publish_status_array`](Self::publish_status_array) for how long it
700    /// stays there.
701    ///
702    /// One case still publishes no terminal entry, and it is the `Err` arm
703    /// below: when `result_cdr` exceeds `RESULT_BUF` the goal enters NEITHER
704    /// list, so the array omits it exactly as it omitted every terminated goal
705    /// before 1361. That is not a regression — it was already the behaviour —
706    /// and the fix for it is the one this function's `Err` already names: raise
707    /// `RESULT_BUF`.
708    ///
709    /// # Errors
710    ///
711    /// `NodeError::BufferTooSmall` when `result_cdr` is larger than
712    /// `RESULT_BUF` and therefore cannot be retained for a later `get_result`.
713    /// Any client already waiting on `~/_action/get_result` is still answered
714    /// (straight from `result_cdr`), and a status array is still published —
715    /// but it carries no entry for this goal (see above), and a *later*
716    /// `get_result` for it gets `GoalStatusCode::Unknown`. Raise the server's
717    /// `RESULT_BUF`.
718    ///
719    /// Issue 0796: this returned `()`, so the pre-fix slab exhaustion — the
720    /// server silently dropping every result once the bump allocator hit
721    /// `RESULT_BUF`, and silently stranding every waiting requester with it —
722    /// was invisible to the caller. A full slab is no longer a failure at all
723    /// (it is reclaimed), and the one remaining failure is reported.
724    pub(crate) fn complete_goal_raw(
725        &mut self,
726        goal_id: &GoalUuid,
727        status: GoalStatusCode,
728        result_cdr: &[u8],
729    ) -> Result<(), NodeError> {
730        // Remove from active goals
731        if let Some(pos) = self
732            .active_goals
733            .iter()
734            .position(|g| g.goal_id.uuid == goal_id.uuid)
735        {
736            self.active_goals.swap_remove(pos);
737        }
738
739        // Re-completing a goal replaces its retained result rather than
740        // stacking a second copy of it in the slab.
741        self.drop_completed_result(goal_id);
742
743        // Store result CDR in the slab, reclaiming older results if needed.
744        let stored = match self.reserve_result_space(result_cdr.len()) {
745            Ok(()) => {
746                let offset = self.result_slab_used;
747                let end = offset + result_cdr.len();
748                self.result_slab[offset..end].copy_from_slice(result_cdr);
749                self.result_slab_used = end;
750                // `reserve_result_space` guaranteed a free entry slot.
751                let pushed = self
752                    .completed_results
753                    .push(CompletedResultEntry {
754                        goal_id: *goal_id,
755                        status,
756                        offset,
757                        len: result_cdr.len(),
758                        delivered: false,
759                    })
760                    .is_ok();
761                debug_assert!(pushed, "reserve_result_space must leave an entry slot");
762                Some((offset, result_cdr.len()))
763            }
764            Err(_) => None,
765        };
766
767        // Phase 237 — flush any get_result requests that arrived while this goal
768        // was still active. Reply to each held requester via its retained
769        // `sequence_number`.
770        //
771        // Issue 0796: this used to be skipped entirely when the result could not
772        // be stored, so an oversized (or, pre-fix, merely unlucky) result left
773        // an `rclcpp_action` client waiting on its result future forever. The
774        // waiter is now answered from `result_cdr` directly when the slab could
775        // not take it — the bytes are right here; only the *retention* failed.
776        let mut delivered_any = false;
777        let mut i = 0;
778        while i < self.pending_get_results.len() {
779            if self.pending_get_results[i].goal_id.uuid == goal_id.uuid {
780                let seq = self.pending_get_results[i].sequence_number;
781                // swap_remove moves the last entry into slot `i`; re-check `i`.
782                let _ = self.pending_get_results.swap_remove(i);
783                let sent = match stored {
784                    Some((offset, len)) => {
785                        self.reply_get_result_from_slab(seq, status, offset, len)
786                    }
787                    None => self.reply_get_result_bytes(seq, status, result_cdr),
788                };
789                delivered_any |= sent.is_ok();
790            } else {
791                i += 1;
792            }
793        }
794        if delivered_any {
795            self.mark_result_delivered(goal_id);
796        }
797
798        let _ = self.publish_status_array();
799
800        if stored.is_some() {
801            Ok(())
802        } else {
803            Err(NodeError::BufferTooSmall)
804        }
805    }
806
807    /// Phase 122.3.c.6.e — register a `Waker` that fires when a new
808    /// send_goal request arrives. Event-driven action servers
809    /// register here in place of polling `try_recv_goal_request` on
810    /// a timer.
811    pub(crate) fn register_goal_waker(&self, waker: &core::task::Waker) -> bool {
812        use nros_rmw::ServiceTrait;
813        self.send_goal_server.register_waker(waker)
814    }
815
816    /// Phase 122.3.c.6.e — register a `Waker` that fires when a
817    /// cancel-goal request arrives.
818    pub(crate) fn register_cancel_waker(&self, waker: &core::task::Waker) -> bool {
819        use nros_rmw::ServiceTrait;
820        self.cancel_goal_server.register_waker(waker)
821    }
822
823    /// Phase 122.3.c.6.e — register a `Waker` that fires when a
824    /// get_result query arrives.
825    pub(crate) fn register_get_result_waker(&self, waker: &core::task::Waker) -> bool {
826        use nros_rmw::ServiceTrait;
827        self.get_result_server.register_waker(waker)
828    }
829
830    /// Phase 122.3.c.6.d — peek a pending cancel-goal request without
831    /// generating a reply. Returns the goal_id named in the request,
832    /// the matching service sequence number (use it with
833    /// [`send_cancel_reply`](Self::send_cancel_reply)), and the
834    /// goal's current status (`GoalStatusCode::Unknown` if no such
835    /// active goal). Returns `Ok(None)` when no cancel request is
836    /// pending.
837    ///
838    /// Used by L1 polling-mode action servers (nros-c / nros-cpp C
839    /// FFI) that want to drive cancel-decision policy without
840    /// passing a Rust closure across the C ABI. See the matching
841    /// [`send_cancel_reply`](Self::send_cancel_reply) for the reply
842    /// side. The high-level closure-based
843    /// [`try_handle_cancel`](Self::try_handle_cancel) keeps working
844    /// and now delegates to this pair.
845    pub(crate) fn try_recv_cancel_request(
846        &mut self,
847    ) -> Result<Option<PendingCancelRequest>, NodeError> {
848        let buf_start = self.cancel_buffer.as_ptr() as usize;
849        let request = match self
850            .cancel_goal_server
851            .take_request(&mut self.cancel_buffer)
852        {
853            Ok(Some(r)) => r,
854            Ok(None) | Err(TransportError::NoData) => return Ok(None),
855            // issue 1088 — pending but no free reply slot: keep the kind.
856            Err(TransportError::WouldBlock) => {
857                return Err(NodeError::Transport(TransportError::WouldBlock));
858            }
859            Err(_) => return Err(NodeError::Transport(TransportError::ServiceRequestFailed)),
860        };
861
862        let data_offset = (request.data.as_ptr() as usize).saturating_sub(buf_start);
863        let data_len = request.data.len();
864        let sequence_number = request.sequence_number;
865        #[allow(clippy::drop_non_drop)]
866        drop(request);
867
868        let mut reader =
869            CdrReader::new_with_header(&self.cancel_buffer[data_offset..data_offset + data_len])
870                .map_err(|_| NodeError::Transport(TransportError::DeserializationError))?;
871
872        let goal_id = read_goal_id(&mut reader)?;
873        let current_status = self.find_goal_status(&goal_id);
874
875        Ok(Some(PendingCancelRequest {
876            goal_id,
877            sequence_number,
878            current_status,
879        }))
880    }
881
882    /// Phase 122.3.c.6.d — send the reply to a previously-peeked
883    /// cancel-goal request. `sequence_number` must match the value
884    /// returned by [`try_recv_cancel_request`](Self::try_recv_cancel_request).
885    ///
886    /// `return_code` is the overall RPC status (`CancelResponseCode::Ok`
887    /// = at least one cancel honoured; other variants = whole-request
888    /// failure). `accepted` lists the goals that transition to
889    /// `Canceling`; this function flips their stored status before
890    /// publishing the status array.
891    ///
892    /// Issue 0796 — the parameter is a [`nros_core::CancelResponseCode`], not
893    /// the per-goal [`nros_core::CancelResponse`] a cancel callback returns.
894    /// The two were one type and their discriminants overlap with opposite
895    /// meanings (`Reject`/`Ok` are both 0), so this signature is what stops a
896    /// per-goal answer being written into the RPC field.
897    pub(crate) fn send_cancel_reply(
898        &mut self,
899        sequence_number: i64,
900        return_code: nros_core::CancelResponseCode,
901        accepted: &[GoalUuid],
902    ) -> Result<(), NodeError> {
903        for id in accepted {
904            self.set_goal_status(id, GoalStatusCode::Canceling);
905        }
906
907        let mut writer =
908            crate::tx_writer(&mut self.goal_buffer).map_err(|_| NodeError::BufferTooSmall)?;
909        writer
910            .write_i8(return_code as i8)
911            .map_err(|_| NodeError::Serialization)?;
912        let count = u32::try_from(accepted.len()).unwrap_or(u32::MAX);
913        writer
914            .write_u32(count)
915            .map_err(|_| NodeError::Serialization)?;
916        for id in accepted {
917            write_goal_id(&mut writer, id)?;
918            // GoalInfo.stamp — the goal's acceptance stamp (issue 2029), zero
919            // for an id this server does not hold.
920            let stamp = self
921                .active_goals
922                .iter()
923                .find(|g| g.goal_id.uuid == id.uuid)
924                .map(|g| g.stamp)
925                .unwrap_or_default();
926            writer
927                .write_i32(stamp.sec)
928                .map_err(|_| NodeError::Serialization)?;
929            writer
930                .write_u32(stamp.nanosec)
931                .map_err(|_| NodeError::Serialization)?;
932        }
933        let reply_len = writer.position();
934
935        self.cancel_goal_server
936            .send_response(sequence_number, &self.goal_buffer[..reply_len])
937            .map_err(|_| NodeError::ServiceReplyFailed)?;
938
939        if !accepted.is_empty() {
940            let _ = self.publish_status_array();
941        }
942
943        Ok(())
944    }
945
946    /// Try to handle a cancel_goal request (type-agnostic) — upstream
947    /// `rcl_action_process_cancel_request` + rclcpp's per-goal callback loop.
948    ///
949    /// Issue 2029 — the request's `GoalInfo` is read as a PAIR, the way
950    /// `action_msgs/srv/CancelGoal` defines it:
951    ///
952    /// | goal id  | stamp    | goals selected                                   |
953    /// |----------|----------|--------------------------------------------------|
954    /// | zero     | zero     | every cancelable goal (cancel-all)               |
955    /// | zero     | non-zero | every cancelable goal accepted at or before it   |
956    /// | non-zero | zero     | that goal                                        |
957    /// | non-zero | non-zero | that goal, plus those accepted at or before it   |
958    ///
959    /// A goal is cancelable while `Accepted` or `Executing`. `cancel_handler`
960    /// is asked ONCE PER SELECTED GOAL; the goals it accepts move to
961    /// `Canceling` and are listed in the reply's `goals_canceling`, each with
962    /// its acceptance stamp. Return codes: a non-zero id naming no active goal
963    /// answers `UnknownGoal` (or `GoalTerminated` when its result is still
964    /// retained) without asking the handler, exactly as `rcl_action` does;
965    /// otherwise an empty `goals_canceling` is `Rejected` and a non-empty one
966    /// `Ok`. The status array is published once if anything moved.
967    ///
968    /// Before 2029 only the goal id was read: the zero id was an unknown goal,
969    /// so cancel-all and cancel-before cancelled nothing — and the handler was
970    /// asked about a goal that did not exist.
971    ///
972    /// Returns the goal id the request NAMED (zero for the two zero-id forms)
973    /// and [`nros_core::CancelResponse::Accept`] if at least one goal moved to
974    /// `Canceling`.
975    pub(crate) fn try_handle_cancel(
976        &mut self,
977        mut cancel_handler: impl FnMut(&GoalUuid, GoalStatusCode) -> nros_core::CancelResponse,
978    ) -> Result<Option<(GoalUuid, nros_core::CancelResponse)>, NodeError> {
979        // (issue 0796) the handler answers about ONE goal; the reply below
980        // carries the RPC-level return code. They are different types with
981        // overlapping discriminants, so the translation is explicit.
982        let buf_start = self.cancel_buffer.as_ptr() as usize;
983        // Phase 120: NoData == steady-state idle; map to Ok(None).
984        let request = match self
985            .cancel_goal_server
986            .take_request(&mut self.cancel_buffer)
987        {
988            Ok(Some(r)) => r,
989            Ok(None) | Err(TransportError::NoData) => return Ok(None),
990            // issue 1088 — pending but no free reply slot: keep the kind.
991            Err(TransportError::WouldBlock) => {
992                return Err(NodeError::Transport(TransportError::WouldBlock));
993            }
994            Err(_) => return Err(NodeError::Transport(TransportError::ServiceRequestFailed)),
995        };
996
997        let data_offset = (request.data.as_ptr() as usize).saturating_sub(buf_start);
998        let data_len = request.data.len();
999        let sequence_number = request.sequence_number;
1000        #[allow(clippy::drop_non_drop)]
1001        drop(request);
1002
1003        let mut reader =
1004            CdrReader::new_with_header(&self.cancel_buffer[data_offset..data_offset + data_len])
1005                .map_err(|_| NodeError::Transport(TransportError::DeserializationError))?;
1006
1007        let goal_id = read_goal_id(&mut reader)?;
1008        // GoalInfo.stamp (builtin_interfaces/Time). A pre-2029 nano-ros client
1009        // sent the id alone; a missing stamp reads as zero, which is what that
1010        // client meant.
1011        let stamp_sec = reader.read_i32().unwrap_or(0);
1012        let stamp_nanosec = reader.read_u32().unwrap_or(0);
1013        let request_stamp = Time::new(stamp_sec, stamp_nanosec);
1014
1015        let (selected, early) = self.select_cancel_goals(&goal_id, request_stamp);
1016
1017        // Ask the handler per selected goal; keep the ones it accepts.
1018        let mut canceling = [false; MAX_GOALS];
1019        let mut any = false;
1020        for (i, goal) in self.active_goals.iter().enumerate() {
1021            if !selected[i] {
1022                continue;
1023            }
1024            if cancel_handler(&goal.goal_id, goal.status) == nros_core::CancelResponse::Accept {
1025                canceling[i] = true;
1026                any = true;
1027            }
1028        }
1029        for (i, goal) in self.active_goals.iter_mut().enumerate() {
1030            if canceling[i] {
1031                goal.status = GoalStatusCode::Canceling;
1032            }
1033        }
1034
1035        let return_code = match early {
1036            Some(code) => code,
1037            None if any => nros_core::CancelResponseCode::Ok,
1038            None => nros_core::CancelResponseCode::Rejected,
1039        };
1040
1041        // Serialize response: return_code (i8) + goals_canceling (sequence of GoalInfo)
1042        let mut writer =
1043            crate::tx_writer(&mut self.goal_buffer).map_err(|_| NodeError::BufferTooSmall)?;
1044        writer
1045            .write_i8(return_code as i8)
1046            .map_err(|_| NodeError::Serialization)?;
1047        let count = canceling.iter().filter(|c| **c).count() as u32;
1048        writer
1049            .write_u32(count)
1050            .map_err(|_| NodeError::Serialization)?;
1051        for (i, goal) in self.active_goals.iter().enumerate() {
1052            if !canceling[i] {
1053                continue;
1054            }
1055            write_goal_id(&mut writer, &goal.goal_id)?;
1056            writer
1057                .write_i32(goal.stamp.sec)
1058                .map_err(|_| NodeError::Serialization)?;
1059            writer
1060                .write_u32(goal.stamp.nanosec)
1061                .map_err(|_| NodeError::Serialization)?;
1062        }
1063        let reply_len = writer.position();
1064
1065        self.cancel_goal_server
1066            .send_response(sequence_number, &self.goal_buffer[..reply_len])
1067            .map_err(|_| NodeError::ServiceReplyFailed)?;
1068
1069        if any {
1070            let _ = self.publish_status_array();
1071        }
1072
1073        let response = if any {
1074            nros_core::CancelResponse::Accept
1075        } else {
1076            nros_core::CancelResponse::Reject
1077        };
1078        Ok(Some((goal_id, response)))
1079    }
1080
1081    /// Issue 2029 — which active goals a `CancelGoal` request selects, and the
1082    /// return code that ends it before any handler runs. See
1083    /// [`try_handle_cancel`](Self::try_handle_cancel) for the table.
1084    fn select_cancel_goals(
1085        &self,
1086        goal_id: &GoalUuid,
1087        request_stamp: Time,
1088    ) -> ([bool; MAX_GOALS], Option<nros_core::CancelResponseCode>) {
1089        let mut selected = [false; MAX_GOALS];
1090        let zero_id = goal_id.uuid == [0u8; 16];
1091        let zero_stamp = request_stamp.sec == 0 && request_stamp.nanosec == 0;
1092        let cancelable = |g: &RawActiveGoal| {
1093            matches!(
1094                g.status,
1095                GoalStatusCode::Accepted | GoalStatusCode::Executing
1096            )
1097        };
1098        let before = |g: &RawActiveGoal| g.stamp.to_nanos() <= request_stamp.to_nanos();
1099
1100        if !zero_id {
1101            match self
1102                .active_goals
1103                .iter()
1104                .position(|g| g.goal_id.uuid == goal_id.uuid)
1105            {
1106                Some(i) => {
1107                    if cancelable(&self.active_goals[i]) {
1108                        selected[i] = true;
1109                    }
1110                }
1111                None => {
1112                    let code = if self
1113                        .completed_results
1114                        .iter()
1115                        .any(|c| c.goal_id.uuid == goal_id.uuid)
1116                    {
1117                        nros_core::CancelResponseCode::GoalTerminated
1118                    } else {
1119                        nros_core::CancelResponseCode::UnknownGoal
1120                    };
1121                    return (selected, Some(code));
1122                }
1123            }
1124            if zero_stamp {
1125                return (selected, None);
1126            }
1127        }
1128        for (i, g) in self.active_goals.iter().enumerate() {
1129            if cancelable(g) && (zero_stamp || before(g)) {
1130                selected[i] = true;
1131            }
1132        }
1133        (selected, None)
1134    }
1135
1136    /// Try to handle a get_result request using raw bytes.
1137    ///
1138    /// For completed goals, sends the stored raw result CDR from the slab.
1139    /// For active/unknown goals, sends the provided `default_result_cdr` bytes.
1140    ///
1141    /// `default_result_cdr` should contain serialized result data (without CDR
1142    /// header or status byte) — typically `A::Result::default()` serialized.
1143    pub(crate) fn try_handle_get_result_raw(
1144        &mut self,
1145        default_result_cdr: &[u8],
1146    ) -> Result<Option<GoalUuid>, NodeError> {
1147        let buf_start = self.goal_buffer.as_ptr() as usize;
1148        // Phase 120: NoData == steady-state idle; map to Ok(None).
1149        let request = match self.get_result_server.take_request(&mut self.goal_buffer) {
1150            Ok(Some(r)) => r,
1151            Ok(None) | Err(TransportError::NoData) => return Ok(None),
1152            // issue 1088 — pending but no free reply slot: keep the kind.
1153            Err(TransportError::WouldBlock) => {
1154                return Err(NodeError::Transport(TransportError::WouldBlock));
1155            }
1156            Err(_) => return Err(NodeError::Transport(TransportError::ServiceRequestFailed)),
1157        };
1158
1159        let data_offset = (request.data.as_ptr() as usize).saturating_sub(buf_start);
1160        let data_len = request.data.len();
1161        let sequence_number = request.sequence_number;
1162        #[allow(clippy::drop_non_drop)]
1163        drop(request);
1164
1165        let mut reader =
1166            CdrReader::new_with_header(&self.goal_buffer[data_offset..data_offset + data_len])
1167                .map_err(|_| NodeError::Transport(TransportError::DeserializationError))?;
1168
1169        let goal_id = read_goal_id(&mut reader)?;
1170
1171        // Look up in completed results
1172        let completed = self
1173            .completed_results
1174            .iter()
1175            .find(|c| c.goal_id.uuid == goal_id.uuid);
1176
1177        if let Some(entry) = completed {
1178            // Completed: send status + stored result CDR from the slab.
1179            let (off, len, status) = (entry.offset, entry.len, entry.status);
1180            self.reply_get_result_from_slab(sequence_number, status, off, len)?;
1181            // Issue 0796 — the client now holds these bytes, so this entry
1182            // becomes the first candidate for reclamation.
1183            self.mark_result_delivered(&goal_id);
1184        } else if self
1185            .active_goals
1186            .iter()
1187            .any(|g| g.goal_id.uuid == goal_id.uuid)
1188        {
1189            // Active goal → DEFER (Phase 237). `rclcpp_action` sends get_result
1190            // right after acceptance and expects the reply only once the goal
1191            // terminates; replying now with a non-terminal status makes the
1192            // client treat an unfinished goal as done. Hold the request's
1193            // correlation token; `complete_goal_raw` flushes it. The backend
1194            // retains the reply token keyed by `sequence_number`.
1195            if self
1196                .pending_get_results
1197                .push(PendingGetResult {
1198                    goal_id,
1199                    sequence_number,
1200                })
1201                .is_err()
1202            {
1203                // Table full — fail loud rather than silently strand the
1204                // requester (caller surfaces it; the request is not re-queued).
1205                return Err(NodeError::BufferTooSmall);
1206            }
1207        } else {
1208            // Unknown goal → reply immediately with UNKNOWN + default result.
1209            // Issue 0796: a goal whose result was reclaimed (evicted by a newer
1210            // completion) lands here too. That is deliberate — an honest
1211            // `Unknown` beats an unanswered query.
1212            self.reply_get_result_bytes(
1213                sequence_number,
1214                GoalStatusCode::Unknown,
1215                default_result_cdr,
1216            )?;
1217        }
1218
1219        Ok(Some(goal_id))
1220    }
1221
1222    /// Write the `get_result` reply prologue — `[CDR header][status i8][pad to
1223    /// 4]` — into `goal_buffer` and return the offset the result bytes go at.
1224    ///
1225    /// The align-to-4 matters: the result CDR starts with a `u32` sequence
1226    /// length the reader will `align(4)` to.
1227    fn write_get_result_header(&mut self, status: GoalStatusCode) -> Result<usize, NodeError> {
1228        let mut writer =
1229            crate::tx_writer(&mut self.goal_buffer).map_err(|_| NodeError::BufferTooSmall)?;
1230        writer
1231            .write_i8(status as i8)
1232            .map_err(|_| NodeError::Serialization)?;
1233        writer.align(4).map_err(|_| NodeError::Serialization)?;
1234        Ok(writer.position())
1235    }
1236
1237    /// Build + send a `get_result` reply — `[status i8][align(4)][result CDR]`
1238    /// — copying the result bytes from the slab. Shared by the immediate
1239    /// completed-goal path and the deferred flush in `complete_goal_raw`
1240    /// (Phase 237). `goal_buffer` and `result_slab` are disjoint fields, so the
1241    /// header build (into `goal_buffer`) and the slab copy don't alias.
1242    fn reply_get_result_from_slab(
1243        &mut self,
1244        sequence_number: i64,
1245        status: GoalStatusCode,
1246        slab_offset: usize,
1247        slab_len: usize,
1248    ) -> Result<(), NodeError> {
1249        let pos = self.write_get_result_header(status)?;
1250        if pos + slab_len > GOAL_BUF {
1251            return Err(NodeError::BufferTooSmall);
1252        }
1253        self.goal_buffer[pos..pos + slab_len]
1254            .copy_from_slice(&self.result_slab[slab_offset..slab_offset + slab_len]);
1255        let reply_len = pos + slab_len;
1256        self.get_result_server
1257            .send_response(sequence_number, &self.goal_buffer[..reply_len])
1258            .map_err(|_| NodeError::ServiceReplyFailed)
1259    }
1260
1261    /// Same reply, from a caller-owned slice instead of the slab.
1262    ///
1263    /// Used for the `Unknown` reply and — issue 0796 — for a waiter whose
1264    /// result was too large to retain: the bytes exist in the caller's buffer
1265    /// even when the slab could not take a copy. `bytes` must NOT alias
1266    /// `self` (the slab path is [`reply_get_result_from_slab`]).
1267    fn reply_get_result_bytes(
1268        &mut self,
1269        sequence_number: i64,
1270        status: GoalStatusCode,
1271        bytes: &[u8],
1272    ) -> Result<(), NodeError> {
1273        let pos = self.write_get_result_header(status)?;
1274        if pos + bytes.len() > GOAL_BUF {
1275            return Err(NodeError::BufferTooSmall);
1276        }
1277        self.goal_buffer[pos..pos + bytes.len()].copy_from_slice(bytes);
1278        let reply_len = pos + bytes.len();
1279        self.get_result_server
1280            .send_response(sequence_number, &self.goal_buffer[..reply_len])
1281            .map_err(|_| NodeError::ServiceReplyFailed)
1282    }
1283
1284    /// Get the number of active goals.
1285    pub(crate) fn active_goal_count(&self) -> usize {
1286        self.active_goals.len()
1287    }
1288
1289    /// Get a reference to all active goals.
1290    pub(crate) fn active_goals(&self) -> &[RawActiveGoal] {
1291        &self.active_goals
1292    }
1293
1294    /// Issue 2029 — the acceptance stamp of an active goal, zero if unknown.
1295    pub(crate) fn goal_stamp(&self, goal_id: &GoalUuid) -> Time {
1296        self.active_goals
1297            .iter()
1298            .find(|g| g.goal_id.uuid == goal_id.uuid)
1299            .map(|g| g.stamp)
1300            .unwrap_or_default()
1301    }
1302
1303    /// Find the status of a goal (active or unknown).
1304    pub(crate) fn find_goal_status(&self, goal_id: &GoalUuid) -> GoalStatusCode {
1305        self.active_goals
1306            .iter()
1307            .find(|g| g.goal_id.uuid == goal_id.uuid)
1308            .map(|g| g.status)
1309            .unwrap_or(GoalStatusCode::Unknown)
1310    }
1311
1312    /// Compile-time bound on the status array (issue 1361).
1313    ///
1314    /// `publish_status_array` now writes up to `2 * MAX_GOALS` entries, and the
1315    /// scratch buffer is a fixed [`STATUS_ARRAY_BUF`]. An instantiation whose
1316    /// worst case does not fit would not fail loudly — `CdrWriter` would answer
1317    /// `BufferTooSmall`, `publish_status_array` would return `Err`, and every
1318    /// caller discards that with `let _ =`, so the terminal status would go
1319    /// missing exactly the way 1361 is about. That silent overflow was reachable
1320    /// BEFORE this change too (at `MAX_GOALS > 18`, active goals alone); nothing
1321    /// guarded it. This is the guard, and it names the numbers.
1322    ///
1323    /// 512 bytes hold 18 entries, so `MAX_GOALS` may be at most 9. The default
1324    /// is 4 and no in-tree instantiation exceeds it.
1325    const STATUS_ARRAY_FITS: () = assert!(
1326        status_array_bytes(2 * MAX_GOALS) <= STATUS_ARRAY_BUF,
1327        "MAX_GOALS is too large for STATUS_ARRAY_BUF: the status array carries \
1328         active goals AND retained completed results, i.e. up to 2 * MAX_GOALS \
1329         entries of 28 bytes. 512 bytes hold 18 entries, so MAX_GOALS <= 9."
1330    );
1331
1332    /// Publish the current `GoalStatusArray` on the status topic.
1333    ///
1334    /// The array carries **active goals and retained completed results**, not
1335    /// active goals alone (issue 1361, option 2). Before that, a terminated goal
1336    /// left `active_goals` in `complete_goal_raw`'s first statement and the
1337    /// publish at the end of the same function was therefore structurally
1338    /// guaranteed to omit the goal it was about: every client saw ACCEPTED and
1339    /// then an empty array, on every backend, and a transient-local late joiner
1340    /// retained that empty array.
1341    ///
1342    /// **The visibility promise this makes, which is NOT upstream's.**
1343    /// `rcl_action` retires a terminated goal on a per-goal `result_timeout`
1344    /// (15 minutes by default). The core has no clock, so a nano-ros action
1345    /// server promises instead: *a terminated goal stays in `/status` until its
1346    /// result is reclaimed* — that is, until the client fetches it and
1347    /// [`expire_completed_results`](Self::expire_completed_results) runs, or
1348    /// until `MAX_GOALS` newer goals displace it through
1349    /// `evict_one_completed_result`. It is a
1350    /// rule a caller can reason about, and it is the closest thing to upstream
1351    /// reachable without a time source; it is deliberately not a duration.
1352    ///
1353    /// Reclamation itself does NOT republish, which is what the late joiner
1354    /// rides on: the last sample the publisher retained still names the terminal
1355    /// status after the result behind it has been returned to the slab.
1356    ///
1357    /// Consequence for a reader: an entry in the array is no longer necessarily
1358    /// an ACTIVE goal, so a count of entries is not a count of active goals —
1359    /// read each entry's `status`. That is what `rcl_action` has always
1360    /// published, so a stock `rclcpp_action` client already does it, and
1361    /// nothing in this tree subscribes to `/status` at all
1362    /// (`ActionClientCore` holds three service clients and a feedback
1363    /// subscriber, no status subscriber). Callers wanting the active count have
1364    /// [`active_goal_count`](Self::active_goal_count), which still reads
1365    /// `active_goals` alone.
1366    pub(crate) fn publish_status_array(&self) -> Result<(), NodeError> {
1367        // Force the bound above to be evaluated for this instantiation.
1368        let () = Self::STATUS_ARRAY_FITS;
1369
1370        let mut buf = [0u8; STATUS_ARRAY_BUF];
1371        let mut writer = crate::tx_writer(&mut buf).map_err(|_| NodeError::BufferTooSmall)?;
1372
1373        let entries = self.active_goals.len() + self.completed_results.len();
1374        writer
1375            .write_u32(entries as u32)
1376            .map_err(|_| NodeError::Serialization)?;
1377
1378        for goal in &self.active_goals {
1379            let info = GoalInfo::new(goal.goal_id, goal.stamp.sec, goal.stamp.nanosec);
1380            let stamped = GoalStatusStamped::new(info, goal.status);
1381            stamped
1382                .serialize(&mut writer)
1383                .map_err(|_| NodeError::Serialization)?;
1384        }
1385
1386        for done in &self.completed_results {
1387            let stamped = GoalStatusStamped::new(GoalInfo::with_id(done.goal_id), done.status);
1388            stamped
1389                .serialize(&mut writer)
1390                .map_err(|_| NodeError::Serialization)?;
1391        }
1392
1393        let len = writer.position();
1394        self.status_publisher
1395            .publish_raw(&buf[..len])
1396            .map_err(|_| NodeError::Transport(TransportError::PublishFailed))
1397    }
1398}
1399
1400// ============================================================================
1401// ActionClientCore
1402// ============================================================================
1403
1404/// Type-agnostic action client core handling the raw-bytes protocol.
1405///
1406/// The typed [`ActionClient`](super::handles::ActionClient) wraps this
1407/// and adds serialization/deserialization at the boundary.
1408pub struct ActionClientCore<
1409    const GOAL_BUF: usize = { crate::config::DEFAULT_RX_BUF_SIZE },
1410    const RESULT_BUF: usize = { crate::config::DEFAULT_RX_BUF_SIZE },
1411    const FEEDBACK_BUF: usize = { crate::config::DEFAULT_RX_BUF_SIZE },
1412> {
1413    pub(crate) send_goal_client: session::RmwServiceClient,
1414    pub(crate) cancel_goal_client: session::RmwServiceClient,
1415    pub(crate) get_result_client: session::RmwServiceClient,
1416    pub(crate) feedback_subscriber: session::RmwSubscriber,
1417    pub(crate) goal_buffer: [u8; GOAL_BUF],
1418    pub(crate) result_buffer: [u8; RESULT_BUF],
1419    pub(crate) feedback_buffer: [u8; FEEDBACK_BUF],
1420    pub(crate) goal_counter: u64,
1421    /// Phase 84.D3: per-sub-client in-flight flags. Each of the three
1422    /// sub-clients (send_goal / cancel / get_result) is an independent
1423    /// request/reply channel and tracks its own "unconsumed reply"
1424    /// state. Cleared by `Promise::take` on success.
1425    pub(crate) in_flight_send_goal: bool,
1426    pub(crate) in_flight_cancel: bool,
1427    pub(crate) in_flight_get_result: bool,
1428    /// Issue 2009 — the requests each sub-client has in flight, keyed by the
1429    /// sequence id the backend assigned and carrying the goal each one is
1430    /// about. A reply is attributed to the goal of the request it NAMES, and a
1431    /// reply naming no pending request (an abandoned one's) is dropped. Before
1432    /// this, every reply was attributed to the most recently SENT goal
1433    /// (`goal_counter`), so two goals in flight swapped acceptances and
1434    /// results, and an abandoned `get_result`'s late reply answered the next.
1435    pub(crate) pending_send_goal: ActionRequests,
1436    pub(crate) pending_cancel: ActionRequests,
1437    pub(crate) pending_get_result: ActionRequests,
1438}
1439
1440/// Issue 2009 — one action sub-client's requests in flight: at most
1441/// [`MAX_CONCURRENT_GOALS`](crate::limits::MAX_CONCURRENT_GOALS) — the number of
1442/// goals a server holds at once — each keyed by its sequence id and carrying
1443/// its goal. A send past that is refused with
1444/// [`NodeError::RequestInFlight`] before it reaches the wire.
1445pub type ActionRequests =
1446    nros_rmw::RequestCorrelator<GoalUuid, { crate::limits::MAX_CONCURRENT_GOALS }>;
1447
1448impl<const GOAL_BUF: usize, const RESULT_BUF: usize, const FEEDBACK_BUF: usize>
1449    ActionClientCore<GOAL_BUF, RESULT_BUF, FEEDBACK_BUF>
1450{
1451    /// Whether the action server's `send_goal` service is CURRENTLY visible.
1452    /// See `ActionClient::action_server_is_ready` for the semantic; the C and
1453    /// C++ `wait_for_action_server` bindings spin on this (phase-428 W13 — the
1454    /// asynchronous discovery pair they used to drive is gone).
1455    ///
1456    /// phase-379 W6 / issue 1008 — `Err` (the backend cannot answer) reports
1457    /// NOT ready. The deleted `is_server_ready` defaulted to `true`, so an
1458    /// image whose backend has no discovery claimed the server was up.
1459    /// Answering `false` makes a caller wait; answering `true` makes it send
1460    /// into the void.
1461    pub(crate) fn is_server_ready(&self) -> bool {
1462        matches!(self.server_readiness(), Ok(true))
1463    }
1464
1465    /// The raw three-answer probe behind [`Self::is_server_ready`] — what a
1466    /// WAIT loop reads, through `ServerVisibility::of`, so that "the backend
1467    /// cannot know" ends the wait instead of being waited out (issue 1686).
1468    pub(crate) fn server_readiness(&self) -> Result<bool, nros_rmw::TransportError> {
1469        use nros_rmw::ClientTrait;
1470        self.send_goal_client.service_is_ready()
1471    }
1472
1473    /// Create a new action client core from the raw transport handles.
1474    pub(crate) fn new(
1475        send_goal_client: session::RmwServiceClient,
1476        cancel_goal_client: session::RmwServiceClient,
1477        get_result_client: session::RmwServiceClient,
1478        feedback_subscriber: session::RmwSubscriber,
1479    ) -> Self {
1480        Self {
1481            send_goal_client,
1482            cancel_goal_client,
1483            get_result_client,
1484            feedback_subscriber,
1485            goal_buffer: [0u8; GOAL_BUF],
1486            result_buffer: [0u8; RESULT_BUF],
1487            feedback_buffer: [0u8; FEEDBACK_BUF],
1488            goal_counter: 0,
1489            in_flight_send_goal: false,
1490            in_flight_cancel: false,
1491            in_flight_get_result: false,
1492            pending_send_goal: ActionRequests::new(),
1493            pending_cancel: ActionRequests::new(),
1494            pending_get_result: ActionRequests::new(),
1495        }
1496    }
1497
1498    /// Send a goal with raw CDR bytes. Returns the generated GoalUuid.
1499    ///
1500    /// The `goal_cdr` bytes are the serialized goal data (without GoalUuid framing).
1501    /// This writes GoalUuid + goal_cdr into the goal buffer and sends the request.
1502    ///
1503    /// After calling, use `send_goal_client` and `result_buffer` to construct
1504    /// a Promise for the acceptance reply.
1505    pub(crate) fn send_goal_raw(&mut self, goal_cdr: &[u8]) -> Result<GoalUuid, NodeError> {
1506        // Issue 2009 — refuse BEFORE the counter moves or the request is sent.
1507        self.pending_send_goal
1508            .can_send()
1509            .map_err(|_| NodeError::RequestInFlight)?;
1510        self.goal_counter += 1;
1511        let mut goal_id = GoalUuid::default();
1512        let counter_bytes = self.goal_counter.to_le_bytes();
1513        goal_id.uuid[..8].copy_from_slice(&counter_bytes);
1514
1515        let mut writer =
1516            crate::tx_writer(&mut self.goal_buffer).map_err(|_| NodeError::BufferTooSmall)?;
1517
1518        write_goal_id(&mut writer, &goal_id)?;
1519
1520        // Copy raw goal CDR bytes after GoalUuid
1521        let pos = writer.position();
1522        if pos + goal_cdr.len() > GOAL_BUF {
1523            return Err(NodeError::BufferTooSmall);
1524        }
1525        self.goal_buffer[pos..pos + goal_cdr.len()].copy_from_slice(goal_cdr);
1526        let req_len = pos + goal_cdr.len();
1527
1528        let seq = self
1529            .send_goal_client
1530            .send_request_raw(&self.goal_buffer[..req_len])
1531            .map_err(|_| NodeError::ServiceRequestFailed)?;
1532        let _ = self.pending_send_goal.insert(seq, goal_id);
1533
1534        Ok(goal_id)
1535    }
1536
1537    /// Try to receive feedback (non-blocking, raw bytes).
1538    ///
1539    /// Returns the GoalUuid and total data length. The full CDR data
1540    /// (including GoalUuid) is in `feedback_buffer`.
1541    pub(crate) fn try_recv_feedback_raw(&mut self) -> Result<Option<(GoalUuid, usize)>, NodeError> {
1542        let data = self
1543            .feedback_subscriber
1544            .take_serialized(&mut self.feedback_buffer)
1545            .map_err(NodeError::Transport)?;
1546
1547        let len = match data {
1548            Some(len) => len,
1549            None => return Ok(None),
1550        };
1551
1552        let mut reader = CdrReader::new_with_header(&self.feedback_buffer[..len])
1553            .map_err(|_| NodeError::Transport(TransportError::DeserializationError))?;
1554
1555        let goal_id = read_goal_id(&mut reader)?;
1556
1557        Ok(Some((goal_id, len)))
1558    }
1559
1560    /// Cancel a goal (non-blocking). Sends the cancel request.
1561    ///
1562    /// After calling, use `cancel_goal_client` and `result_buffer` to construct
1563    /// a Promise for the cancel response.
1564    pub(crate) fn send_cancel_request(&mut self, goal_id: &GoalUuid) -> Result<(), NodeError> {
1565        self.send_cancel_request_stamped(goal_id, 0, 0)
1566            .map(|_seq| ())
1567    }
1568
1569    /// [`Self::send_cancel_request`] with the request's `GoalInfo.stamp` — issue 2027.
1570    ///
1571    /// `action_msgs/srv/CancelGoal` reads the pair (goal id, stamp): a zero id
1572    /// with a zero stamp cancels EVERY goal, a zero id with a stamp cancels
1573    /// the goals accepted at or before it, and a non-zero id cancels that goal
1574    /// (plus, with a stamp, those before it). The reply is correlated under
1575    /// `goal_id` — the zero id for the two broadcast forms. Returns the
1576    /// request's sequence id, which is what tells two pending requests about
1577    /// the SAME goal (or two broadcasts) apart when their replies come back
1578    /// ([`Self::try_recv_cancel_reply_seq`]).
1579    pub(crate) fn send_cancel_request_stamped(
1580        &mut self,
1581        goal_id: &GoalUuid,
1582        stamp_sec: i32,
1583        stamp_nanosec: u32,
1584    ) -> Result<i64, NodeError> {
1585        self.pending_cancel
1586            .can_send()
1587            .map_err(|_| NodeError::RequestInFlight)?;
1588        let mut writer =
1589            crate::tx_writer(&mut self.goal_buffer).map_err(|_| NodeError::BufferTooSmall)?;
1590
1591        write_goal_id(&mut writer, goal_id)?;
1592        writer
1593            .write_i32(stamp_sec)
1594            .map_err(|_| NodeError::Serialization)?;
1595        writer
1596            .write_u32(stamp_nanosec)
1597            .map_err(|_| NodeError::Serialization)?;
1598
1599        let req_len = writer.position();
1600
1601        let seq = self
1602            .cancel_goal_client
1603            .send_request_raw(&self.goal_buffer[..req_len])
1604            .map_err(|_| NodeError::ServiceRequestFailed)?;
1605        let _ = self.pending_cancel.insert(seq, *goal_id);
1606        Ok(seq)
1607    }
1608
1609    /// Send a get_result request.
1610    ///
1611    /// After calling, use `get_result_client` and `result_buffer` to construct
1612    /// a Promise for the result response.
1613    pub(crate) fn send_get_result_request(&mut self, goal_id: &GoalUuid) -> Result<(), NodeError> {
1614        self.pending_get_result
1615            .can_send()
1616            .map_err(|_| NodeError::RequestInFlight)?;
1617        let mut writer =
1618            crate::tx_writer(&mut self.goal_buffer).map_err(|_| NodeError::BufferTooSmall)?;
1619
1620        write_goal_id(&mut writer, goal_id)?;
1621
1622        let req_len = writer.position();
1623
1624        let seq = self
1625            .get_result_client
1626            .send_request_raw(&self.goal_buffer[..req_len])
1627            .map_err(|_| NodeError::ServiceRequestFailed)?;
1628        let _ = self.pending_get_result.insert(seq, *goal_id);
1629        Ok(())
1630    }
1631
1632    /// Phase 122.3.c.6.e — register a `Waker` that fires when the
1633    /// send_goal RPC reply lands.
1634    pub(crate) fn register_goal_response_waker(&self, waker: &core::task::Waker) -> bool {
1635        use nros_rmw::ClientTrait;
1636        self.send_goal_client.register_waker(waker)
1637    }
1638
1639    /// Phase 122.3.c.6.e — register a `Waker` for cancel-RPC replies.
1640    pub(crate) fn register_cancel_response_waker(&self, waker: &core::task::Waker) -> bool {
1641        use nros_rmw::ClientTrait;
1642        self.cancel_goal_client.register_waker(waker)
1643    }
1644
1645    /// Phase 122.3.c.6.e — register a `Waker` for get_result replies.
1646    pub(crate) fn register_result_waker(&self, waker: &core::task::Waker) -> bool {
1647        use nros_rmw::ClientTrait;
1648        self.get_result_client.register_waker(waker)
1649    }
1650
1651    /// Phase 122.3.c.6.e — register a `Waker` for feedback messages.
1652    pub(crate) fn register_feedback_waker(&self, waker: &core::task::Waker) -> bool {
1653        use nros_rmw::Subscription;
1654        self.feedback_subscriber.register_waker(waker)
1655    }
1656
1657    /// Phase 122.3.c.6.c — poll for a cancel reply (non-blocking,
1658    /// raw bytes). Returns `Ok(Some(len))` when a reply landed; the
1659    /// CDR payload is in `result_buffer_ref()[..len]`. Reply layout
1660    /// is action_msgs/srv/CancelGoal_Response wire CDR.
1661    pub(crate) fn try_recv_cancel_reply(&mut self) -> Result<Option<usize>, NodeError> {
1662        Ok(self.try_recv_cancel_reply_for()?.map(|(len, _goal)| len))
1663    }
1664
1665    /// Issue 2009 — [`Self::try_recv_cancel_reply`] with the goal the reply is ABOUT: the
1666    /// goal of the request whose sequence id it names. A reply naming no
1667    /// pending request is dropped (it answers an abandoned request).
1668    pub(crate) fn try_recv_cancel_reply_for(
1669        &mut self,
1670    ) -> Result<Option<(usize, GoalUuid)>, NodeError> {
1671        Ok(self
1672            .try_recv_cancel_reply_seq()?
1673            .map(|(len, _seq, goal)| (len, goal)))
1674    }
1675
1676    /// Issue 2027 — [`Self::try_recv_cancel_reply_for`] with the sequence id
1677    /// of the request the reply answers too ([`nros_rmw::SEQUENCE_ID_UNKNOWN`]
1678    /// from a backend that assigns none — it then holds one cancel in flight).
1679    pub(crate) fn try_recv_cancel_reply_seq(
1680        &mut self,
1681    ) -> Result<Option<(usize, i64, GoalUuid)>, NodeError> {
1682        // Phase 120: NoData == steady-state polling; map to Ok(None).
1683        match nros_rmw::take_correlated(
1684            &mut self.cancel_goal_client,
1685            &mut self.result_buffer,
1686            &mut self.pending_cancel,
1687        ) {
1688            Ok(opt) => Ok(opt),
1689            Err(TransportError::NoData) => Ok(None),
1690            Err(_) => Err(NodeError::Transport(TransportError::DeserializationError)),
1691        }
1692    }
1693
1694    /// Poll for a get_result reply (non-blocking, raw bytes).
1695    ///
1696    /// Returns `Ok(Some(total_len))` if a reply arrived (data in result buffer),
1697    /// `Ok(None)` if no reply yet.
1698    ///
1699    /// After receiving, use [`result_buffer_ref()`](Self::result_buffer_ref)
1700    /// to access the raw CDR data. The layout is: CDR header (4) + status
1701    /// byte (1) + result data.
1702    pub(crate) fn try_recv_get_result_reply(&mut self) -> Result<Option<usize>, NodeError> {
1703        Ok(self
1704            .try_recv_get_result_reply_for()?
1705            .map(|(len, _goal)| len))
1706    }
1707
1708    /// Issue 2009 — [`Self::try_recv_get_result_reply`] with the goal the reply is ABOUT: the
1709    /// goal of the request whose sequence id it names. A reply naming no
1710    /// pending request is dropped (it answers an abandoned request).
1711    pub(crate) fn try_recv_get_result_reply_for(
1712        &mut self,
1713    ) -> Result<Option<(usize, GoalUuid)>, NodeError> {
1714        // Phase 120: NoData == steady-state polling; map to Ok(None).
1715        match nros_rmw::take_correlated(
1716            &mut self.get_result_client,
1717            &mut self.result_buffer,
1718            &mut self.pending_get_result,
1719        ) {
1720            Ok(opt) => Ok(opt.map(|(len, _seq, goal)| (len, goal))),
1721            Err(TransportError::NoData) => Ok(None),
1722            Err(_) => Err(NodeError::Transport(TransportError::DeserializationError)),
1723        }
1724    }
1725
1726    /// Poll for the send_goal acceptance reply (non-blocking, raw bytes).
1727    ///
1728    /// Returns `Ok(Some(total_len))` if a reply arrived (data in result buffer),
1729    /// `Ok(None)` if no reply yet.
1730    ///
1731    /// The reply CDR contains: header (4) + accepted (u8) + stamp (i32 + u32).
1732    pub(crate) fn try_recv_send_goal_reply(&mut self) -> Result<Option<usize>, NodeError> {
1733        Ok(self.try_recv_send_goal_reply_for()?.map(|(len, _goal)| len))
1734    }
1735
1736    /// Issue 2009 — [`Self::try_recv_send_goal_reply`] with the goal the reply is ABOUT: the
1737    /// goal of the request whose sequence id it names. A reply naming no
1738    /// pending request is dropped (it answers an abandoned request).
1739    pub(crate) fn try_recv_send_goal_reply_for(
1740        &mut self,
1741    ) -> Result<Option<(usize, GoalUuid)>, NodeError> {
1742        // Phase 120: NoData == steady-state polling; map to Ok(None).
1743        match nros_rmw::take_correlated(
1744            &mut self.send_goal_client,
1745            &mut self.result_buffer,
1746            &mut self.pending_send_goal,
1747        ) {
1748            Ok(opt) => Ok(opt.map(|(len, _seq, goal)| (len, goal))),
1749            Err(TransportError::NoData) => Ok(None),
1750            Err(_) => Err(NodeError::Transport(TransportError::DeserializationError)),
1751        }
1752    }
1753
1754    /// Issue 2027 — forget the oldest pending `send_goal` request for
1755    /// `goal_id`: its slot (one of
1756    /// [`MAX_CONCURRENT_GOALS`](crate::limits::MAX_CONCURRENT_GOALS)) is free
1757    /// for the next goal, and its reply, if it still comes, is dropped. For a
1758    /// binding whose last owner of that request let go before the reply came
1759    /// (a dropped C++ future). `true` if one was pending.
1760    pub(crate) fn forget_send_goal(&mut self, goal_id: &GoalUuid) -> bool {
1761        self.pending_send_goal.remove_key(goal_id)
1762    }
1763
1764    /// Issue 2027 — [`Self::forget_send_goal`] for a `get_result` request.
1765    pub(crate) fn forget_get_result(&mut self, goal_id: &GoalUuid) -> bool {
1766        self.pending_get_result.remove_key(goal_id)
1767    }
1768
1769    /// Issue 2027 — [`Self::forget_send_goal`] for a `cancel_goal` request
1770    /// (the zero id names a cancel-all / cancel-before request).
1771    pub(crate) fn forget_cancel(&mut self, goal_id: &GoalUuid) -> bool {
1772        self.pending_cancel.remove_key(goal_id)
1773    }
1774
1775    /// Read-only access to the result buffer (after polling a reply).
1776    pub(crate) fn result_buffer_ref(&self) -> &[u8] {
1777        &self.result_buffer
1778    }
1779
1780    /// Read-only access to the feedback buffer (after receiving feedback).
1781    pub(crate) fn feedback_buffer_ref(&self) -> &[u8] {
1782        &self.feedback_buffer
1783    }
1784
1785    /// Get the current goal counter (used to reconstruct the last goal ID).
1786    pub(crate) fn goal_counter(&self) -> u64 {
1787        self.goal_counter
1788    }
1789}