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}