diff --git a/.fusa-reqs.json b/.fusa-reqs.json index 727034c..f7d8602 100644 --- a/.fusa-reqs.json +++ b/.fusa-reqs.json @@ -1451,7 +1451,7 @@ { "id": "REQ-LOAN-001", "title": "loan() returns zeroed buffer", - "text": "loan::Controller::loan(n) shall return a zeroed byte vector of exactly n bytes.", + "text": "loan::BufferPool::loan(n) shall return a zeroed byte vector of exactly n bytes.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" @@ -1459,47 +1459,47 @@ { "id": "REQ-LOAN-002", "title": "loan() returns ErrClosed when closed", - "text": "loan() shall return ErrClosed if the controller has been closed.", + "text": "loan() shall return ErrClosed once the pool has been closed.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" }, { "id": "REQ-LOAN-003", - "title": "loan() returns error for negative size", - "text": "loan() shall return an error if size < 0.", + "title": "loan() returns an error for negative size", + "text": "loan() shall return std::errc::invalid_argument if size < 0.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" }, { "id": "REQ-LOAN-004", - "title": "send_loaned() delegates to inner", - "text": "send_loaned() shall delegate command delivery to the inner controller.", + "title": "Released buffers become available for reuse", + "text": "A Loan's buffer, once released (destruction or explicit ret()), shall become available for a subsequent loan() call to draw from the pool rather than freshly allocating.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" }, { "id": "REQ-LOAN-005", - "title": "Loan RAII releases buffer to pool", - "text": "When a Loan is destroyed without being sent, its destructor shall return the buffer to the pool.", + "title": "Released buffers are re-zeroed before reuse", + "text": "A buffer drawn from the pool via loan() shall be re-zeroed, even if a prior borrower left non-zero data in it, so no stale data leaks across reuse.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" }, { "id": "REQ-LOAN-006", - "title": "zone() delegates to inner", - "text": "zone() shall return the inner controller's zone.", + "title": "close() is idempotent and safe with outstanding Loans alive", + "text": "BufferPool::close() shall be safe to call more than once, including while Loans obtained before the call are still alive.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" }, { "id": "REQ-TLS-001", - "title": "Mutual TLS authentication", - "text": "TLS transport shall require both client and server to present valid X.509 certificates.", + "title": "Mutual TLS authentication for the UDP/IP transport", + "text": "The secure channel option for rcp/udp.hpp's transport shall require both client and server to present valid X.509 certificates once a real (D)TLS backend is enabled.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" @@ -1530,16 +1530,16 @@ }, { "id": "REQ-TLS-005", - "title": "Command frames encrypted", - "text": "All Command and Response frames shall be encrypted in transit.", + "title": "SecureClient never sends a request in the clear without a real backend", + "text": "Without RCP_TLS_OPENSSL, SecureClient::request shall return function_not_supported for every call rather than transmitting an unencrypted AVTPDU-framed request.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" }, { "id": "REQ-TLS-006", - "title": "Status frames encrypted", - "text": "All Status subscription frames shall be encrypted in transit.", + "title": "SecureServer never reports itself usable without a real backend", + "text": "Without RCP_TLS_OPENSSL, SecureServer::ok() shall report false, so an embedding cannot mistake the stub for a securely bound server.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" @@ -1562,16 +1562,16 @@ }, { "id": "REQ-TLS-009", - "title": "close() terminates TLS session", - "text": "close() shall perform a clean TLS shutdown before closing the underlying socket.", + "title": "close() terminates the secure session cleanly", + "text": "SecureClient::close and SecureServer::close shall perform a clean shutdown and be safe to call more than once.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" }, { "id": "REQ-TLS-010", - "title": "Error propagation", - "text": "Any TLS handshake or I/O error shall be returned as a std::error_code to the caller.", + "title": "stream_id and Config are carried unchanged", + "text": "SecureClient and SecureServer shall expose the stream_id and Config they were constructed with, unchanged, for the embedding application to introspect.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" @@ -1579,127 +1579,127 @@ { "id": "REQ-SHMEM-001", "title": "In-process zero-copy delivery", - "text": "shmem::Controller::send() shall deliver commands to the ZoneServer without serialising to a byte buffer.", + "text": "shmem::Channel::request() shall deliver a request to the registered Handler directly, without serialising to a byte buffer.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" }, { "id": "REQ-SHMEM-002", - "title": "Custom handler invoked", - "text": "If set_handler() has been called, the handler shall be invoked for each command; otherwise a default OK response is returned.", + "title": "Custom handler invoked, else default Acknowledge", + "text": "If set_handler() has been called, the handler shall be invoked for each request() call; otherwise Channel::request() shall answer ResponseKind::Acknowledge.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" }, { "id": "REQ-SHMEM-003", - "title": "ErrZoneMismatch on wrong zone", - "text": "send() shall return ErrZoneMismatch if cmd.zone differs from the server's zone.", + "title": "ErrClosed after close()", + "text": "Channel::request() shall return ErrClosed once the Channel has been closed.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" }, { "id": "REQ-SHMEM-004", - "title": "ErrClosed after close()", - "text": "send() shall return ErrClosed after the Controller or ZoneServer has been closed.", + "title": "Registry::lookup finds a channel registered via add_channel", + "text": "shmem::Registry::lookup() shall return the Channel for a stream_key registered via add_channel().", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" }, { "id": "REQ-SHMEM-005", - "title": "publish() delivers to all subscribers", - "text": "ZoneServer::publish() shall push the Status to every active StatusChannel.", + "title": "Registry::close() closes all registered channels", + "text": "Registry::close() shall close every registered Channel and cause subsequent lookup() calls to return ErrClosed.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" }, { "id": "REQ-SHMEM-006", - "title": "Payload copied before handler", - "text": "The controller shall copy the Command payload before passing it to the ZoneServer handler.", + "title": "Channel::request is thread-safe for concurrent callers", + "text": "Concurrent calls to Channel::request() from different threads shall not result in data races or lost/duplicated handler invocations.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" }, { "id": "REQ-SHMEM-007", - "title": "Registry lookup finds registered server", - "text": "shmem::Registry::lookup() shall return the controller for a zone registered via add_server().", + "title": "Client id is forwarded to the handler unchanged", + "text": "The caller-supplied client id passed to Channel::request() shall be forwarded to the registered Handler unchanged.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" }, { "id": "REQ-SHMEM-008", - "title": "Registry::close() releases all controllers", - "text": "Registry::close() shall close all registered controllers and cause subsequent lookup() to return ErrClosed.", + "title": "close() is idempotent", + "text": "Channel::close() and Registry::close() shall be safe to call more than once without crashing or blocking.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" }, { "id": "REQ-TSN-001", - "title": "SO_PRIORITY set per command priority", - "text": "tsn::Controller::send() shall set the SO_PRIORITY socket option proportional to the command's rcp::Priority before sending.", + "title": "apply_priority sets SO_PRIORITY per RequestCategory", + "text": "tsn::apply_priority(fd, cfg, category) shall set the SO_PRIORITY socket option proportional to the request's rcp::request::RequestCategory before the next send on that socket, on platforms where RCP_TSN_SO_PRIORITY is defined; on other platforms it shall succeed as a no-op.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" }, { "id": "REQ-TSN-002", - "title": "PCPMap maps Normal/High/Critical to PCP values", - "text": "The PCPMap struct shall provide separate PCP values for Priority::Normal, Priority::High, and Priority::Critical.", + "title": "PCPMap maps every RequestCategory to a PCP value", + "text": "The PCPMap struct shall provide a separate PCP value for each of the seven rcp::request::RequestCategory values.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" }, { "id": "REQ-TSN-003", - "title": "Critical mapped to highest PCP", - "text": "Priority::Critical shall be mapped to PCP 7 by default.", + "title": "Cancellation mapped to the highest PCP", + "text": "RequestCategory::Cancellation (execution-priority rank 0) shall be mapped to PCP 7 by default.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" }, { "id": "REQ-TSN-004", - "title": "High mapped to mid-range PCP", - "text": "Priority::High shall be mapped to PCP 5 by default.", + "title": "Triggered mapped to a mid/high-range PCP", + "text": "RequestCategory::Triggered shall be mapped to PCP 6 by default.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" }, { "id": "REQ-TSN-005", - "title": "Normal mapped to low PCP", - "text": "Priority::Normal shall be mapped to PCP 2 by default.", + "title": "Standard mapped to the lowest PCP", + "text": "RequestCategory::Standard (the mandatory baseline kind, execution-priority rank 6) shall be mapped to PCP 1 by default, and the default PCPMap shall preserve the full seven-category execution-priority ordering.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" }, { "id": "REQ-TSN-006", - "title": "subscribe() delegates to inner", - "text": "subscribe() shall delegate directly to the inner controller without modifying socket priority.", + "title": "apply_priority rejects an invalid socket fd", + "text": "apply_priority shall return std::errc::invalid_argument for a negative fd, on every platform.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" }, { "id": "REQ-MDNS-001", - "title": "Discoverer interface for zone discovery", - "text": "mdns::Discoverer shall define the abstract interface for mDNS-based zone controller discovery.", + "title": "Discoverer interface for udp::Server host:port discovery", + "text": "mdns::Discoverer shall define the abstract interface for mDNS-based discovery of udp::Server instances reachable at a host:port, for the UDP/IP transport variant only.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" }, { "id": "REQ-MDNS-002", - "title": "start() fires Added event for each discovered zone", - "text": "Discoverer::start() shall invoke the callback with EventType::Added for each discovered ZoneInfo.", + "title": "start() fires Added event for each discovered server", + "text": "Discoverer::start() shall invoke the callback with EventType::Added for each discovered ServerInfo.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" @@ -1714,16 +1714,16 @@ }, { "id": "REQ-MDNS-004", - "title": "StaticDiscoverer emits configured zones", - "text": "StaticDiscoverer::start() shall emit all configured ZoneInfo entries synchronously before returning.", + "title": "StaticDiscoverer emits configured servers", + "text": "StaticDiscoverer::start() shall emit all configured ServerInfo entries synchronously before returning.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" }, { "id": "REQ-MDNS-005", - "title": "ZoneInfo carries host, port, and zone", - "text": "ZoneInfo shall carry the Zone, host string, port number, and service instance name.", + "title": "ServerInfo carries stream_key, host, and port", + "text": "ServerInfo shall carry the opaque stream_key (typically avtp::StreamId::to_u64()), host string, port number, and service instance name.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" @@ -1731,15 +1731,15 @@ { "id": "REQ-MDNS-006", "title": "instance_name follows _rcp._udp.local convention", - "text": "make_instance_name() shall produce names of the form '.._rcp._udp.local'.", + "text": "make_instance_name() shall produce names of the form '.._rcp._udp.local', distinct for every distinct stream_id.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" }, { "id": "REQ-MDNS-007", - "title": "Announcer interface for zone registration", - "text": "mdns::Announcer shall define the abstract interface for announcing zone controllers via the local mDNS responder.", + "title": "Announcer interface for udp::Server registration", + "text": "mdns::Announcer shall define the abstract interface for announcing a udp::Server via the local mDNS responder.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" @@ -1747,7 +1747,7 @@ { "id": "REQ-MDNS-008", "title": "withdraw() removes mDNS record", - "text": "Announcer::withdraw(z) shall remove the mDNS record for the given zone.", + "text": "Announcer::withdraw(stream_key) shall remove the mDNS record for the given stream_key.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" @@ -2106,16 +2106,16 @@ }, { "id": "REQ-OBS-001", - "title": "ObservingController records a span per send", - "text": "ObservingController::send() shall record one Span for every call to the underlying controller.", + "title": "ObservingClient records a span per request", + "text": "ObservingClient::request() shall record one Span for every call to the wrapped RequestFn.", "standard": "iso26262", "level": "ASIL-A", "asil": "ASIL-A" }, { "id": "REQ-OBS-002", - "title": "Span captures zone and command type", - "text": "The Span recorded by ObservingController shall include the Command zone and CommandType.", + "title": "Span captures byte_bus_id and acf_msg_type", + "text": "The Span recorded by ObservingClient shall include the request's byte_bus_id and acf_msg_type.", "standard": "iso26262", "level": "ASIL-A", "asil": "ASIL-A" @@ -2123,7 +2123,7 @@ { "id": "REQ-OBS-003", "title": "Span duration is non-negative", - "text": "The duration() of every recorded Span shall be \u2265 0 microseconds.", + "text": "The duration() of every recorded Span shall be >= 0 microseconds.", "standard": "iso26262", "level": "ASIL-A", "asil": "ASIL-A" @@ -2131,7 +2131,7 @@ { "id": "REQ-OBS-004", "title": "Span captures error code", - "text": "The Span result field shall reflect the error_code returned by the inner controller.", + "text": "The Span result field shall reflect the error_code returned by the wrapped RequestFn.", "standard": "iso26262", "level": "ASIL-A", "asil": "ASIL-A" @@ -2154,16 +2154,16 @@ }, { "id": "REQ-OBS-007", - "title": "ObservingController increments rcp.commands.total counter", - "text": "ObservingController::send() shall call record_counter('rcp.commands.total') once per send.", + "title": "ObservingClient increments rcp.requests.total counter", + "text": "ObservingClient::request() shall call record_counter('rcp.requests.total') once per call.", "standard": "iso26262", "level": "ASIL-A", "asil": "ASIL-A" }, { "id": "REQ-OBS-008", - "title": "ObservingController increments error counter on failure", - "text": "When send() returns a non-zero error_code, ObservingController shall also call record_counter('rcp.commands.errors').", + "title": "ObservingClient increments error counter on failure", + "text": "When request() returns a non-zero error_code, ObservingClient shall also call record_counter('rcp.requests.errors').", "standard": "iso26262", "level": "ASIL-A", "asil": "ASIL-A" @@ -2234,24 +2234,24 @@ }, { "id": "REQ-REC-001", - "title": "RecordingController captures command and response", - "text": "RecordingController::send() shall append one Entry to the Record containing the Command, Response, and error_code for every send call.", + "title": "RecordingClient captures request and response", + "text": "RecordingClient::request() shall append one Entry to the Record containing the request AcfMessageInfo/payload, the response AcfMessageInfo/payload, and the resulting error_code, for every call.", "standard": "iso26262", "level": "ASIL-A", "asil": "ASIL-A" }, { "id": "REQ-REC-002", - "title": "Record::size() reflects number of sent commands", - "text": "Record::size() shall equal the total number of completed send() calls through the RecordingController.", + "title": "Record::size() reflects number of completed request() calls", + "text": "Record::size() shall equal the total number of completed request() calls through the RecordingClient.", "standard": "iso26262", "level": "ASIL-A", "asil": "ASIL-A" }, { "id": "REQ-REC-003", - "title": "Record::write_binary creates a non-empty file", - "text": "Record::write_binary() shall create a file at the specified path, write all captured entries, and return no error.", + "title": "Record::write_binary/read_binary round-trip entries losslessly", + "text": "Record::write_binary() shall serialise every captured Entry to a file, and Record::read_binary() shall recover the timestamp, request/response AcfMessageInfo fields, and request/response payloads unchanged.", "standard": "iso26262", "level": "ASIL-A", "asil": "ASIL-A" @@ -2259,7 +2259,7 @@ { "id": "REQ-REC-004", "title": "Playback::run_all replays all entries", - "text": "Playback::run_all() shall send one command for every entry in the Record to the target controller.", + "text": "Playback::run_all() shall replay one request for every entry in the Record against the target RequestFn.", "standard": "iso26262", "level": "ASIL-A", "asil": "ASIL-A" @@ -2275,15 +2275,15 @@ { "id": "REQ-REC-006", "title": "Entry timestamp_ns is monotonically increasing", - "text": "The timestamp_ns fields of consecutive entries in a Record shall be non-decreasing.", + "text": "The timestamp_ns fields of consecutive entries appended to a Record shall be non-decreasing.", "standard": "iso26262", "level": "ASIL-A", "asil": "ASIL-A" }, { "id": "REQ-REC-007", - "title": "RecordingController forwards send result unchanged", - "text": "RecordingController::send() shall return the same error_code as the inner controller.", + "title": "RecordingClient forwards the inner error_code unchanged", + "text": "RecordingClient::request() shall return the same error_code as the wrapped RequestFn.", "standard": "iso26262", "level": "ASIL-B", "asil": "ASIL-B" diff --git a/CMakeLists.txt b/CMakeLists.txt index 6be7b75..1b917da 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -1,7 +1,7 @@ cmake_minimum_required(VERSION 3.21) project(cpp-rcp - VERSION 2.13.0 + VERSION 2.14.0 DESCRIPTION "Remote Control Protocol for automotive zonal architecture (C++)" HOMEPAGE_URL "https://github.com/SoundMatt/cpp-RCP" LANGUAGES CXX diff --git a/README.md b/README.md index 3ac015e..a6cd1a8 100644 --- a/README.md +++ b/README.md @@ -64,19 +64,19 @@ reference. | `` | Transparent zone proxy for cascaded zonal topologies | | `` | Atomic multi-zone command broadcast with typed zone group sets | | `` | Runtime schema registry and dynamic payload encoding | -| `` | `LoaningController` — zero-copy payload loaning via a pre-allocated pool | -| `` | Binary record and replay of RCP traffic | -| `` | OpenTelemetry-style observability: spans, gauges, counters | -| `` | mDNS/DNS-SD zone controller discovery (RFC 6762/6763) | +| `` | `loan::BufferPool` — zero-copy payload loaning via a pre-allocated pool, for AVTPDU/ACF request-building (ROADMAP.md v2.14.0) | +| `` | Binary record and replay of RC-Client-level request/response traffic (ROADMAP.md v2.14.0) | +| `` | OpenTelemetry-style observability: spans and counters around a client-side send-equivalent call (ROADMAP.md v2.14.0) | +| `` | mDNS/DNS-SD host:port discovery (RFC 6762/6763), scoped to the UDP/IP transport variant (ROADMAP.md v2.14.0) | ### Transports | Header | Description | |---|---| | `` | Native IEEE 1722-over-UDP/IP transport (Annex J): `Server`/`Client` carry AVTPDU (NTSCF/TSCF) + ACF_ABB/ACF_GBB frames over UDP sockets (ROADMAP.md v2.13.0) | -| `` | Mutual TLS transport for zone controller communication | -| `` | Zero-copy intra-host command delivery via shared in-process memory | -| `` | IEEE 802.1Qbv-aware UDP transport adapter for hard real-time Ethernet | +| `` | Secure-channel option for the UDP/IP transport variant (DTLS/application-layer, `SecureClient`/`SecureServer`); the specification's own preferred link security is MACsec (802.1AE) at layer 2, which this package does not address (ROADMAP.md v2.14.0) | +| `` | Zero-copy in-process request `Channel`/`Registry`, keyed by opaque stream_key (ROADMAP.md v2.14.0) | +| `` | IEEE 802.1p PCP-priority hint (`apply_priority`) keyed off `rcp::request::RequestCategory`'s execution-priority ordering; prefer genuine IEEE 1722 stream reservation where available (ROADMAP.md v2.14.0) | | `` | TC18 wire codec, framing half — IEEE 1722 AVTPDU (NTSCF/TSCF) header framing (ROADMAP.md v2.0.0) | | `` | TC18 wire codec, message half — ACF_ABB/ACF_GBB message format (ROADMAP.md v2.0.0) | | `` | Timing-realistic RC Server simulator for SiL/HIL testing — wraps `` with latency/jitter and Fault/Recover controls, watchdog wired via `` (ROADMAP.md v2.12.0) | diff --git a/ROADMAP.md b/ROADMAP.md index e031a56..768147e 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -1089,6 +1089,82 @@ not claimed, same as the equivalent disclaimers in `rcp/avtp.hpp`, - Depends on: Wire format core (v2.0.0); UDP/IP transport (v2.13.0) for the `mdns`/`tls` items specifically +**Done (v2.14.0):** All seven files are ADAPTed in place per the Satellite +Package Disposition table's entries — each concept survives, rebound onto +the new stream/`byte_bus_id` addressing model instead of the removed +`Zone`/`Command`/`Controller`/`Registry` types. `rcp/rcp.hpp`'s own +"nothing new should build on this" notice held throughout: none of the +seven include it for anything beyond `rcp::Context`/the generic +`ErrClosed`/`ErrNotFound`/`ErrAlreadyExists`/`ErrTimeout` error constants +(also not part of that pre-replacement model) and, for `loan.hpp`, the +already-generic `rcp::Loan` RAII buffer holder. + +`mdns.hpp`'s `Discoverer`/`StaticDiscoverer`/`Announcer` interfaces are +unchanged in shape; `ZoneInfo` becomes `ServerInfo`, keyed by an opaque +`stream_key` (typically `avtp::StreamId::to_u64()`) instead of `Zone`. +`tls.hpp`'s `Config` (cert/key/ca/verify_peer) survives unchanged; +`Controller`/`ZoneServer`/`Registry` are replaced by `SecureClient`/ +`SecureServer` wrapping `rcp/udp.hpp`'s `Server`/`Client` (v2.13.0), same +OpenSSL-gated-real-backend/`function_not_supported`-stub split as before. +The header now also states plainly, per this milestone's own instruction, +that the specification's preferred link-security mechanism is MACsec +(802.1AE) at layer 2 — this package addresses the UDP/IP path only and +does not implement MACsec itself. + +`tsn.hpp`'s `PCPMap` now maps all seven `rcp::request::RequestCategory` +values (v2.5.0) instead of the removed `Priority` enum's three, preserving +the same cancellation > triggered > timed > compound > compound-wait > +chained > standard execution-priority ordering `rcp/request.hpp`'s +`priority_rank` already defines (extraction §3.14). The `Controller` +wrapper is dropped for a standalone `apply_priority(fd, cfg, category)` +free function — there is no unified client-side `send()` chokepoint left +to wrap (that unification, if any, does not land until the CLI/capi/adapt +rebuilds at v2.16.0, per `rcp/authz.hpp`'s equivalent v2.11.0 note) — and +its header comment now notes that genuine IEEE 1722 stream reservation +(802.1Qat SRP), where available, is preferable to this socket-level hint. + +`shmem.hpp`'s `ZoneServer`/`Controller` pair becomes `Channel`/`Registry`, +keyed by `stream_key` the same way `rcp/watchdog.hpp`'s `Manager` already +is; `Channel::request()` still dispatches to a caller-supplied handler +in-process with no wire encode/decode step, the same zero-copy value +proposition as before, now shaped to match `udp::Server::Handler`/ +`rcp::mock::Server::dispatch`'s signature. + +`loan.hpp`'s `loan::Controller` (which wrapped the now-gone +`rcp::LoaningController`) becomes a standalone `BufferPool`, the same +"primitive, not a wrapped chokepoint" choice `tsn.hpp` makes at this +milestone — a caller draws a `rcp::Loan` from it immediately before +building an AVTPDU/ACF-framed request. Buffers are re-zeroed on reuse +(closing a small aliasing gap the pre-replacement pool didn't guarantee). + +`record.hpp` and `observe.hpp` both now wrap a `RequestFn` — a +`std::function` shaped like `udp::Client::request`'s core signature (the +"new client-side send-equivalent call" this milestone calls for) — rather +than the removed `rcp::Controller`. `record::Entry` carries +`acf::AcfMessageInfo`/payload pairs for both the request and response +instead of `Command`/`Response`/`Status`; `Record::write_binary` now has a +matching `read_binary`, encoding each message via `rcp::acf::encode_acf_abb`/ +`encode_acf_gbb` (the same codec the wire path itself uses) so the on-disk +format round-trips losslessly rather than only being asserted non-empty. +`observe::Span`/`Metric` swap `Zone`/`CommandType` for `byte_bus_id`/ +`acf_msg_type` and an opaque `stream_key`; the OTel-style span/counter +approach itself (`MetricsSink`/`NoopSink`/`InMemorySink`) is unchanged. + +Grepping the tree for consumers of each file's pre-v2.14.0 API beyond its +own (now-rewritten) test found none, so no legacy-shim split file was +needed for any of the seven, same as `rcp/udp.hpp`'s v2.13.0 rebuild. +New/rewritten coverage lives in `tests/test_mdns.cpp` (`REQ-MDNS-001..008`), +`tests/test_tls.cpp` (`REQ-TLS-001..010`), `tests/test_tsn.cpp` +(`REQ-TSN-001..006`), `tests/test_shmem.cpp` (`REQ-SHMEM-001..008`), +`tests/test_loan.cpp` (`REQ-LOAN-001..006`), `tests/test_record.cpp` +(`REQ-REC-001..008`), and `tests/test_observe.cpp` (`REQ-OBS-001..008`) — +all entirely rewritten under the same file/prefix identity as the +discarded pre-replacement coverage, the same convention `tests/test_udp.cpp` +followed at v2.13.0 — traced in `.fusa-reqs.json`. Full bit-for-bit +conformance against other TC18/Annex-J implementations is not claimed, +same as the equivalent disclaimers in `rcp/avtp.hpp`, `rcp/acf.hpp`, +`rcp/discovery.hpp`, and `rcp/udp.hpp`. + ### 59. Application-Layer Protocol Bridge Rebind (v2.15.0) - Rebind `mqttbr.hpp`, `ddsbr.hpp`, `someipbr.hpp`, `restbridge.hpp`, diff --git a/include/rcp/loan.hpp b/include/rcp/loan.hpp index dbbee8a..a08a229 100644 --- a/include/rcp/loan.hpp +++ b/include/rcp/loan.hpp @@ -5,12 +5,31 @@ // fusa:req REQ-LOAN-005 // fusa:req REQ-LOAN-006 -// LoaningController wrapper — extends any rcp::Controller with zero-copy -// payload loaning via a pool of pre-allocated byte vectors. +// BufferPool — zero-copy payload loaning for AVTPDU-framed request/response +// construction, via a pool of pre-allocated byte vectors. // -// loan() obtains a zeroed buffer from the pool; send_loaned() delivers it to -// the inner controller without copying. The Loan's destructor returns the -// buffer to the pool automatically if not sent. +// ROADMAP.md milestone 58, "Auxiliary Transport & Cross-Cutting Rebind +// (v2.14.0)": this header is ADAPTed, per the Satellite Package +// Disposition table's entry for `loan.hpp` — the pre-replacement +// loan::Controller wrapped rcp::LoaningController, a chokepoint that no +// longer exists (there is no unified client-side send() interface yet; +// that unification, if any, does not land until the CLI/capi/adapt +// rebuilds at v2.16.0 — see rcp/authz.hpp's equivalent v2.11.0 note). This +// rebind drops the wrapper and keeps the pool as a standalone primitive a +// caller draws a buffer from immediately before building an +// AVTPDU/ACF-framed request (rcp/avtp.hpp, rcp/acf.hpp) — e.g. the +// req_payload/out_resp_payload arguments to rcp/udp.hpp's Client::request +// — same "primitives driven by the embedding application" convention every +// Phase 14/15 header has used since v2.9.0. This likely matters more now +// than it did against the old 16-byte header: AVTPDU + ACF_ABB/ACF_GBB +// framing carries more bytes worth avoiding a fresh heap allocation for on +// a hot request path. +// +// loan() obtains a zeroed buffer from the pool; the returned rcp::Loan (a +// generic RAII buffer holder — not itself part of the pre-replacement +// Zone/Command/Controller/Registry model rcp/rcp.hpp's own header comment +// warns against building on) returns its buffer to the pool automatically +// once it goes out of scope, or immediately via its own ret() method. #pragma once #include "rcp.hpp" @@ -23,65 +42,60 @@ namespace rcp { namespace loan { -class Controller final : public rcp::LoaningController { +class BufferPool { public: - explicit Controller(std::shared_ptr inner) - : inner_(std::move(inner)) {} - - ~Controller() override { auto ec = close(); (void)ec; } - - Zone zone() const noexcept override { return inner_->zone(); } - - std::error_code send(const rcp::Context& ctx, - const Command& cmd, - Response& out) override { - return inner_->send(ctx, cmd, out); - } - - std::error_code subscribe(const rcp::Context& ctx, - std::shared_ptr& out) override { - return inner_->subscribe(ctx, out); - } - - std::error_code close() override { - closed_.store(true, std::memory_order_release); - return inner_->close(); - } + ~BufferPool() { close(); } - // loan returns a zeroed buffer of exactly size bytes from the pool. - std::error_code loan(int size, std::unique_ptr& out) override { + // loan returns a zeroed buffer of exactly size bytes, drawn from the + // pool if a same-or-larger buffer is available for reuse, or freshly + // allocated otherwise. + std::error_code loan(int size, std::unique_ptr& out) { if (closed_.load(std::memory_order_acquire)) return ErrClosed; - if (size < 0) return make_error_code(Errc::busy); + if (size < 0) return std::make_error_code(std::errc::invalid_argument); + + std::vector buf; + { + std::lock_guard lk(pool_mu_); + if (!pool_.empty()) { + buf = std::move(*pool_.back()); + pool_.pop_back(); + } + } + buf.assign(static_cast(size), 0); // re-zero: no stale data leaks across reuse - std::vector buf(static_cast(size), 0); - // Release function returns the buffer to the pool. auto* raw = new std::vector(std::move(buf)); out = std::make_unique( *raw, [this, raw]() mutable { std::lock_guard lk(pool_mu_); - raw->clear(); pool_.push_back(std::unique_ptr>(raw)); }); return {}; } - std::error_code send_loaned(const rcp::Context& ctx, - Command& cmd, - Response& out) override { - if (closed_.load(std::memory_order_acquire)) return ErrClosed; - return inner_->send(ctx, cmd, out); + // close is idempotent — safe to call more than once, including while + // Loans obtained before the call are still alive (their eventual + // release simply grows a pool nobody will draw from again). + void close() { closed_.store(true, std::memory_order_release); } + + bool ok() const noexcept { return !closed_.load(std::memory_order_acquire); } + + // pooled_count reports how many released buffers are currently held + // for reuse — introspection for tests, not part of the loan/release + // contract itself. + size_t pooled_count() const { + std::lock_guard lk(pool_mu_); + return pool_.size(); } private: - std::shared_ptr inner_; std::atomic closed_{false}; - std::mutex pool_mu_; + mutable std::mutex pool_mu_; std::vector>> pool_; }; -inline std::shared_ptr new_controller(std::shared_ptr inner) { - return std::make_shared(std::move(inner)); +inline std::unique_ptr new_buffer_pool() { + return std::make_unique(); } } // namespace loan diff --git a/include/rcp/mdns.hpp b/include/rcp/mdns.hpp index e5c2f65..a146193 100644 --- a/include/rcp/mdns.hpp +++ b/include/rcp/mdns.hpp @@ -7,31 +7,54 @@ // fusa:req REQ-MDNS-007 // fusa:req REQ-MDNS-008 -// mDNS/DNS-SD zone controller discovery (RFC 6762 + RFC 6763). +// mDNS/DNS-SD host:port discovery (RFC 6762 + RFC 6763) for the UDP/IP +// transport variant specifically. // -// Provides the abstract Discoverer interface and a StaticDiscoverer for testing. -// A full mDNS responder (Avahi-compatible) requires platform APIs not available -// in a header-only library. Use the avahi-client or systemd-resolved backend on -// Linux, or dns_sd on macOS, and wrap it with this Discoverer interface. +// ROADMAP.md milestone 58, "Auxiliary Transport & Cross-Cutting Rebind +// (v2.14.0)": this header is ADAPTed, per the Satellite Package Disposition +// table's entry for `mdns.hpp` — its scope is narrowed rather than carried +// forward unchanged. The OPEN Alliance TC18 Remote Control Protocol +// Specification v0.5.1_RC already defines its own wire-level discovery over +// an Ethernet-addressed stream (v2.2.0, rcp/discovery.hpp) that needs no +// name service at all: a native-Ethernet-framed RC Server is found by +// broadcasting a discovery request on the link, not by resolving a service +// name. mDNS/DNS-SD only earns its keep here when host:port discovery is +// separately needed for the IEEE1722-over-UDP/IP path (v2.13.0, +// rcp/udp.hpp), where "the link" is an IP network with no broadcast-domain +// guarantee and a server's UDP socket has no MAC-address-derived identity a +// client could otherwise rendezvous on. +// +// This is still an abstract interface, not a working mDNS responder — full +// mDNS requires platform APIs (avahi-client/systemd-resolved on Linux, +// dns_sd on macOS) not available in a header-only library. StaticDiscoverer +// below exists for testing and static configuration. #pragma once -#include "rcp.hpp" +#include "rcp.hpp" // for rcp::Context — see this header's own scope note above +#include "avtp.hpp" #include #include +#include +#include #include #include namespace rcp { namespace mdns { -// ── ZoneInfo ───────────────────────────────────────────────────────────────── +// ── ServerInfo ──────────────────────────────────────────────────────────────── +// What a discovered (or announced) udp::Server is reachable at. `stream_id` +// replaces the old ZoneInfo::zone — it is the same opaque per-sender +// identity rcp/avtp.hpp's StreamId models, carried here as its uint64_t +// form (avtp::StreamId::to_u64()) for the same "opaque index" convention +// rcp/watchdog.hpp's Manager and rcp/regmap.hpp's Ep0 already use. -struct ZoneInfo { - Zone zone; +struct ServerInfo { + uint64_t stream_key = 0; std::string host; uint16_t port = 0; - std::string instance_name; // e.g. "front-left.myhost._rcp._udp.local" + std::string instance_name; // e.g. "0002xxxxxxxx1234.myhost._rcp._udp.local" }; // ── DiscoveryEvent ──────────────────────────────────────────────────────────── @@ -39,13 +62,13 @@ struct ZoneInfo { enum class EventType : uint8_t { Added = 0, Removed = 1 }; struct DiscoveryEvent { - EventType event; - ZoneInfo info; + EventType event; + ServerInfo info; }; // ── Discoverer ──────────────────────────────────────────────────────────────── -// Discoverer is the abstract interface for mDNS-based zone discovery. +// Discoverer is the abstract interface for mDNS-based udp::Server discovery. // Implementations call the registered callback for each add/remove event. class Discoverer { public: @@ -62,18 +85,19 @@ class Discoverer { // ── StaticDiscoverer ───────────────────────────────────────────────────────── -// StaticDiscoverer emits a fixed set of ZoneInfo entries immediately on start(). -// Suitable for testing and static configuration. +// StaticDiscoverer emits a fixed set of ServerInfo entries immediately on +// start(). Suitable for testing and static configuration where the set of +// UDP/IP servers is known in advance and does not need real mDNS resolution. class StaticDiscoverer final : public Discoverer { public: - explicit StaticDiscoverer(std::vector zones) - : zones_(std::move(zones)) {} + explicit StaticDiscoverer(std::vector servers) + : servers_(std::move(servers)) {} std::error_code start(Callback cb) override { stopped_.store(false); - for (auto& z : zones_) { + for (auto& s : servers_) { if (stopped_.load()) break; - cb({EventType::Added, z}); + cb({EventType::Added, s}); } return {}; } @@ -81,31 +105,40 @@ class StaticDiscoverer final : public Discoverer { void stop() override { stopped_.store(true); } private: - std::vector zones_; - std::atomic stopped_{false}; + std::vector servers_; + std::atomic stopped_{false}; }; // ── Announcer (passive interface) ───────────────────────────────────────────── -// Announcer registers a ZoneInfo record in the local mDNS responder. +// Announcer registers a ServerInfo record in the local mDNS responder so +// that a udp::Server bound to `info.host`:`info.port` becomes discoverable. // Platform implementations call the underlying mDNS API (Avahi, dns_sd, etc.) class Announcer { public: virtual ~Announcer() = default; - virtual std::error_code announce(const ZoneInfo& info) = 0; - virtual void withdraw(Zone z) = 0; + virtual std::error_code announce(const ServerInfo& info) = 0; + virtual void withdraw(uint64_t stream_key) = 0; }; // ── Helpers ─────────────────────────────────────────────────────────────────── -// make_instance_name builds the standard _rcp._udp.local service instance name. -inline std::string make_instance_name(Zone z, const std::string& hostname) { - return rcp::to_string(z) + "." + hostname + "._rcp._udp.local"; +// make_instance_name builds a standard _rcp._udp.local service instance +// name from a stream_id and hostname, using stream_id's hex-encoded uint64_t +// form as the unique-per-sender label (the same identity avtp::StreamId +// already models — no zone/instance name of its own to draw from now). +inline std::string make_instance_name(avtp::StreamId stream_id, const std::string& hostname) { + std::ostringstream oss; + oss << std::hex << std::uppercase; + oss.width(16); + oss.fill('0'); + oss << stream_id.to_u64(); + return oss.str() + "." + hostname + "._rcp._udp.local"; } -inline std::unique_ptr new_static_discoverer(std::vector zones) { - return std::make_unique(std::move(zones)); +inline std::unique_ptr new_static_discoverer(std::vector servers) { + return std::make_unique(std::move(servers)); } } // namespace mdns diff --git a/include/rcp/observe.hpp b/include/rcp/observe.hpp index 9b4a6de..f1fefa3 100644 --- a/include/rcp/observe.hpp +++ b/include/rcp/observe.hpp @@ -7,20 +7,29 @@ // fusa:req REQ-OBS-007 // fusa:req REQ-OBS-008 -// OpenTelemetry-style observability: spans, gauges, and counters (v0.25.0). +// OpenTelemetry-style observability: spans and counters around RC-Client +// request/response traffic. // -// ObservingController wraps any rcp::Controller and records a latency span -// around every send(). Metrics are exported via a configurable MetricsSink. -// The default sink is a no-op; users supply a real sink for production use. -// -// Span and Metric types are intentionally plain so they can be adapted to any -// observability backend (OTel gRPC, Prometheus, etc.) without pulling in -// heavy dependencies. +// ROADMAP.md milestone 58, "Auxiliary Transport & Cross-Cutting Rebind +// (v2.14.0)": this header is ADAPTed, per the Satellite Package +// Disposition table's entry for `observe.hpp` — the OTel-style span/ +// counter approach itself is unaffected by the protocol replacement, only +// what it wraps changes. ObservingClient below wraps a RequestFn — the +// same client-side send-equivalent callable rcp/record.hpp's +// RecordingClient wraps at this same milestone; see that file's header +// comment for why it mirrors rcp/udp.hpp's Client::request's core shape +// rather than depending on rcp/udp.hpp directly — and records one Span per +// call, the same latency-around-send() concept the pre-replacement +// ObservingController used. #pragma once -#include "rcp.hpp" +#include "acf.hpp" +#include "avtp.hpp" +#include "rcp.hpp" // for rcp::Context only — see this header's own scope note above #include +#include +#include #include #include #include @@ -29,12 +38,20 @@ namespace rcp { namespace observe { +// RequestFn — see rcp/record.hpp's header comment for why this shape is +// shared between the two v2.14.0 wrapper headers. +using RequestFn = std::function&, + acf::AcfMessageInfo&, + std::vector&)>; + // ── Span ────────────────────────────────────────────────────────────────────── struct Span { std::string name; - Zone zone; - CommandType cmd_type; + avtp::ByteBusId byte_bus_id = 0; + uint8_t acf_msg_type = acf::kAcfMsgTypeAbb; std::chrono::steady_clock::time_point start_time; std::chrono::steady_clock::time_point end_time; std::error_code result; @@ -46,11 +63,14 @@ struct Span { }; // ── Metric ──────────────────────────────────────────────────────────────────── +// `stream_key` replaces the old Metric::zone — the same opaque uint64_t +// per-sender identity (typically avtp::StreamId::to_u64()) rcp/watchdog.hpp's +// Manager and rcp/shmem.hpp's Registry already key on. struct Metric { std::string name; double value; - Zone zone; + uint64_t stream_key; }; // ── MetricsSink ─────────────────────────────────────────────────────────────── @@ -60,14 +80,14 @@ class MetricsSink { virtual ~MetricsSink() = default; virtual void record_span(const Span&) = 0; virtual void record_gauge(const Metric&) = 0; - virtual void record_counter(const std::string& name, Zone zone, double delta) = 0; + virtual void record_counter(const std::string& name, uint64_t stream_key, double delta) = 0; }; class NoopSink final : public MetricsSink { public: void record_span(const Span&) override {} void record_gauge(const Metric&) override {} - void record_counter(const std::string&, Zone, double) override {} + void record_counter(const std::string&, uint64_t, double) override {} }; // ── InMemorySink ────────────────────────────────────────────────────────────── @@ -80,7 +100,7 @@ class InMemorySink final : public MetricsSink { spans_.push_back(s); } void record_gauge(const Metric&) override {} - void record_counter(const std::string&, Zone, double) override {} + void record_counter(const std::string&, uint64_t, double) override {} std::vector spans() const { std::lock_guard lk(mu_); @@ -96,53 +116,52 @@ class InMemorySink final : public MetricsSink { std::vector spans_; }; -// ── ObservingController ─────────────────────────────────────────────────────── +// ── ObservingClient ─────────────────────────────────────────────────────────── -class ObservingController final : public rcp::Controller { +class ObservingClient { public: - ObservingController(std::shared_ptr inner, - std::shared_ptr sink) + // `stream_key` identifies which stream this client observes (typically + // avtp::StreamId::to_u64() for the underlying transport, e.g. + // rcp/udp.hpp's Client) — RequestFn's own signature carries no stream + // identity of its own (see this file's header comment), so it is + // supplied once here rather than re-derived per call. + ObservingClient(RequestFn inner, uint64_t stream_key, std::shared_ptr sink) : inner_(std::move(inner)) + , stream_key_(stream_key) , sink_(std::move(sink)) {} - Zone zone() const noexcept override { return inner_->zone(); } - - std::error_code send(const rcp::Context& ctx, - const Command& cmd, - Response& out) override { + std::error_code request(const rcp::Context& ctx, + const acf::AcfMessageInfo& req, + const std::vector& req_payload, + acf::AcfMessageInfo& out_resp, + std::vector& out_resp_payload) { Span span; - span.name = "rcp.send"; - span.zone = cmd.zone; - span.cmd_type = cmd.type; - span.start_time = std::chrono::steady_clock::now(); + span.name = "rcp.request"; + span.byte_bus_id = req.byte_bus_id; + span.acf_msg_type = req.acf_msg_type; + span.start_time = std::chrono::steady_clock::now(); - auto ec = inner_->send(ctx, cmd, out); + auto ec = inner_(ctx, req, req_payload, out_resp, out_resp_payload); span.end_time = std::chrono::steady_clock::now(); span.result = ec; sink_->record_span(span); - sink_->record_counter("rcp.commands.total", cmd.zone, 1.0); - if (ec) sink_->record_counter("rcp.commands.errors", cmd.zone, 1.0); + sink_->record_counter("rcp.requests.total", stream_key_, 1.0); + if (ec) sink_->record_counter("rcp.requests.errors", stream_key_, 1.0); return ec; } - std::error_code subscribe(const rcp::Context& ctx, - std::shared_ptr& out) override { - return inner_->subscribe(ctx, out); - } - - std::error_code close() override { return inner_->close(); } - private: - std::shared_ptr inner_; - std::shared_ptr sink_; + RequestFn inner_; + uint64_t stream_key_; + std::shared_ptr sink_; }; -inline std::shared_ptr new_controller( - std::shared_ptr inner, - std::shared_ptr sink = std::make_shared()) { - return std::make_shared(std::move(inner), std::move(sink)); +inline std::shared_ptr new_observing_client( + RequestFn inner, uint64_t stream_key, + std::shared_ptr sink = std::make_shared()) { + return std::make_shared(std::move(inner), stream_key, std::move(sink)); } } // namespace observe diff --git a/include/rcp/record.hpp b/include/rcp/record.hpp index 1237b99..0f841ab 100644 --- a/include/rcp/record.hpp +++ b/include/rcp/record.hpp @@ -7,29 +7,41 @@ // fusa:req REQ-REC-007 // fusa:req REQ-REC-008 -// Binary record and replay of RCP traffic (v0.27.0). +// Binary record and replay of RC-Client-level request/response traffic. // -// RecordingController wraps any Controller and writes a timestamped binary log -// of every Command/Response pair to a Record. Playback::send() reads the log -// and drives a mock controller with the recorded responses. +// ROADMAP.md milestone 58, "Auxiliary Transport & Cross-Cutting Rebind +// (v2.14.0)": this header is ADAPTed, per the Satellite Package +// Disposition table's entry for `record.hpp` — its on-disk entry format is +// reworked to capture RC-Client-level request/response pairs (each +// rcp::acf::AcfMessageInfo header plus its payload, per rcp/acf.hpp, +// v2.0.0) instead of the pre-replacement Command/Response/Status/Priority +// tuples. `RequestFn` below is shaped identically to rcp/udp.hpp's +// Client::request's core signature (minus the TSCF-specific trailing +// knobs, which are a framing detail this transport-agnostic layer does not +// need to see) so a rcp::udp::Client instance, or any other client-side +// send-equivalent call, can be wrapped without adapting its call shape — +// the same generic-callable pattern rcp/observe.hpp's ObservingClient uses +// alongside this file at the same milestone. // -// Log format per entry (little-endian): -// [8 bytes] wall-clock timestamp in nanoseconds since epoch -// [2 bytes] CommandType -// [1 byte] Zone -// [1 byte] Priority -// [4 bytes] payload length -// [N bytes] payload -// [1 byte] ResponseStatus -// [4 bytes] response payload length -// [N bytes] response payload +// RecordingClient wraps a RequestFn and writes a timestamped binary log of +// every request/response pair to a Record. Playback::run_all() reads a +// Record back and drives a target RequestFn (e.g. a mock, per +// rcp/mock.hpp) with the recorded requests, honoring the recorded +// inter-entry timing. +// +// Full bit-for-bit conformance of the on-disk format against any other +// implementation is not claimed — it is this implementation's own logging +// encoding, same as the equivalent disclaimers in rcp/avtp.hpp, rcp/acf.hpp, +// and rcp/discovery.hpp. #pragma once -#include "rcp.hpp" +#include "acf.hpp" +#include "rcp.hpp" // for rcp::Context only — see this header's own scope note above #include #include #include +#include #include #include #include @@ -38,13 +50,24 @@ namespace rcp { namespace record { +// RequestFn is the "client-side send-equivalent call" this header wraps — +// see this file's header comment for why it mirrors udp::Client::request's +// core shape rather than depending on rcp/udp.hpp directly. +using RequestFn = std::function&, + acf::AcfMessageInfo&, + std::vector&)>; + // ── Entry ───────────────────────────────────────────────────────────────────── struct Entry { - int64_t timestamp_ns; // nanoseconds since epoch - Command cmd; - Response resp; - std::error_code error; + int64_t timestamp_ns = 0; // nanoseconds since epoch + acf::AcfMessageInfo request; + std::vector request_payload; + acf::AcfMessageInfo response; + std::vector response_payload; + std::error_code error; }; // ── Record ──────────────────────────────────────────────────────────────────── @@ -56,108 +79,175 @@ class Record { entries_.push_back(std::move(e)); } - const std::vector& entries() const noexcept { return entries_; } - size_t size() const noexcept { return entries_.size(); } + std::vector entries() const { + std::lock_guard lk(mu_); + return entries_; + } + size_t size() const { + std::lock_guard lk(mu_); + return entries_.size(); + } - // write_binary serialises the record to a file. + // write_binary serialises the record to a file. Each entry's request + // and response are encoded via rcp::acf::encode_acf_abb/encode_acf_gbb + // (selected by AcfMessageInfo::acf_msg_type) — the same codec the wire + // path itself uses — length-prefixed so read_binary can recover them + // without re-deriving ACF's own length field semantics. std::error_code write_binary(const std::string& path) const { - std::lock_guard lk(mu_); + std::vector snapshot; + { + std::lock_guard lk(mu_); + snapshot = entries_; + } std::ofstream f(path, std::ios::binary | std::ios::trunc); if (!f) return std::make_error_code(std::errc::io_error); - for (auto& e : entries_) { - auto write8 = [&](int64_t v) { - f.write(reinterpret_cast(&v), 8); - }; - auto write2 = [&](uint16_t v) { - f.write(reinterpret_cast(&v), 2); - }; - auto write1 = [&](uint8_t v) { - f.write(reinterpret_cast(&v), 1); - }; - auto write4 = [&](uint32_t v) { - f.write(reinterpret_cast(&v), 4); - }; - auto writeVec = [&](const std::vector& d) { - write4(static_cast(d.size())); - if (!d.empty()) - f.write(reinterpret_cast(d.data()), - static_cast(d.size())); - }; - - write8(e.timestamp_ns); - write2(static_cast(e.cmd.type)); - write1(static_cast(e.cmd.zone)); - write1(static_cast(e.cmd.priority)); - writeVec(e.cmd.payload); - write1(static_cast(e.resp.status)); - writeVec(e.resp.payload); + for (auto& e : snapshot) write_entry(f, e); + return f.good() ? std::error_code{} : std::make_error_code(std::errc::io_error); + } + + // read_binary is write_binary's inverse: it replaces this Record's + // entries with the ones decoded from `path`. + std::error_code read_binary(const std::string& path) { + std::ifstream f(path, std::ios::binary); + if (!f) return std::make_error_code(std::errc::io_error); + + std::vector loaded; + while (f.peek() != std::ifstream::traits_type::eof()) { + Entry e; + if (auto ec = read_entry(f, e)) return ec; + loaded.push_back(std::move(e)); } + std::lock_guard lk(mu_); + entries_ = std::move(loaded); return {}; } private: + static void write8(std::ofstream& f, int64_t v) { f.write(reinterpret_cast(&v), 8); } + static void write1(std::ofstream& f, uint8_t v) { f.write(reinterpret_cast(&v), 1); } + static void write4(std::ofstream& f, uint32_t v) { f.write(reinterpret_cast(&v), 4); } + static void write_msg(std::ofstream& f, const acf::AcfMessageInfo& info, + const std::vector& payload) { + // RequestFn (see this file's header comment) does not carry ACF_GBB's + // separate message_timestamp parameter, only AcfMessageInfo/payload — + // a GBB entry's recorded message_timestamp slot is therefore always + // zero on replay. Harmless for this module's own record/replay round + // trip (mtv/opcode-repurposing semantics, rcp/request.hpp, still + // survive intact); a caller needing exact GBB timestamp fidelity + // should record at a layer that sees it directly instead. + std::vector encoded = (info.acf_msg_type == acf::kAcfMsgTypeGbb) + ? acf::encode_acf_gbb(info, /*message_timestamp=*/0, payload) + : acf::encode_acf_abb(info, payload); + write4(f, static_cast(encoded.size())); + if (!encoded.empty()) + f.write(reinterpret_cast(encoded.data()), + static_cast(encoded.size())); + } + + static void write_entry(std::ofstream& f, const Entry& e) { + write8(f, e.timestamp_ns); + write_msg(f, e.request, e.request_payload); + write_msg(f, e.response, e.response_payload); + write4(f, static_cast(e.error.value())); + write1(f, e.error ? 1 : 0); + } + + static std::error_code read8(std::ifstream& f, int64_t& v) { + f.read(reinterpret_cast(&v), 8); + return f.good() ? std::error_code{} : std::make_error_code(std::errc::io_error); + } + static std::error_code read1(std::ifstream& f, uint8_t& v) { + f.read(reinterpret_cast(&v), 1); + return f.good() ? std::error_code{} : std::make_error_code(std::errc::io_error); + } + static std::error_code read4(std::ifstream& f, uint32_t& v) { + f.read(reinterpret_cast(&v), 4); + return f.good() ? std::error_code{} : std::make_error_code(std::errc::io_error); + } + static std::error_code read_msg(std::ifstream& f, acf::AcfMessageInfo& info, + std::vector& payload) { + uint32_t len = 0; + if (auto ec = read4(f, len)) return ec; + std::vector encoded(len); + if (len > 0) { + f.read(reinterpret_cast(encoded.data()), static_cast(len)); + if (!f.good()) return std::make_error_code(std::errc::io_error); + } + if (!encoded.empty() && encoded[0] == acf::kAcfMsgTypeGbb) { + uint64_t ts = 0; + return acf::decode_acf_gbb(encoded.data(), encoded.size(), info, ts, payload); + } + return acf::decode_acf_abb(encoded.data(), encoded.size(), info, payload); + } + + static std::error_code read_entry(std::ifstream& f, Entry& e) { + if (auto ec = read8(f, e.timestamp_ns)) return ec; + if (auto ec = read_msg(f, e.request, e.request_payload)) return ec; + if (auto ec = read_msg(f, e.response, e.response_payload)) return ec; + uint32_t err_value = 0; + uint8_t has_error = 0; + if (auto ec = read4(f, err_value)) return ec; + if (auto ec = read1(f, has_error)) return ec; + e.error = has_error + ? std::error_code(static_cast(err_value), std::generic_category()) + : std::error_code{}; + return {}; + } + mutable std::mutex mu_; std::vector entries_; }; -// ── RecordingController ─────────────────────────────────────────────────────── +// ── RecordingClient ─────────────────────────────────────────────────────────── -class RecordingController final : public rcp::Controller { +class RecordingClient { public: - RecordingController(std::shared_ptr inner, - std::shared_ptr rec) + RecordingClient(RequestFn inner, std::shared_ptr rec) : inner_(std::move(inner)), rec_(std::move(rec)) {} - Zone zone() const noexcept override { return inner_->zone(); } - - std::error_code send(const rcp::Context& ctx, - const Command& cmd, - Response& out) override { - auto ec = inner_->send(ctx, cmd, out); + std::error_code request(const rcp::Context& ctx, + const acf::AcfMessageInfo& req, + const std::vector& req_payload, + acf::AcfMessageInfo& out_resp, + std::vector& out_resp_payload) { + auto ec = inner_(ctx, req, req_payload, out_resp, out_resp_payload); Entry e; e.timestamp_ns = std::chrono::duration_cast( std::chrono::system_clock::now().time_since_epoch()).count(); - e.cmd = cmd; - e.resp = out; - e.error = ec; + e.request = req; + e.request_payload = req_payload; + e.response = out_resp; + e.response_payload = out_resp_payload; + e.error = ec; rec_->append(std::move(e)); return ec; } - std::error_code subscribe(const rcp::Context& ctx, - std::shared_ptr& out) override { - return inner_->subscribe(ctx, out); - } - - std::error_code close() override { return inner_->close(); } - private: - std::shared_ptr inner_; - std::shared_ptr rec_; + RequestFn inner_; + std::shared_ptr rec_; }; // ── PlaybackConfig ──────────────────────────────────────────────────────────── struct PlaybackConfig { - double speed_factor = 1.0; // 2.0 = 2× faster, 0.0 = no delays + double speed_factor = 1.0; // 2.0 = 2x faster, 0.0 = no delays }; // ── Playback ────────────────────────────────────────────────────────────────── -// Playback replays a Record against a target controller using the recorded +// Playback replays a Record against a target RequestFn using the recorded // inter-entry timing (scaled by speed_factor). class Playback { public: - Playback(std::shared_ptr target, - const Record& rec, - PlaybackConfig cfg = {}) + Playback(RequestFn target, const Record& rec, PlaybackConfig cfg = {}) : target_(std::move(target)), rec_(rec), cfg_(cfg) {} // run_all replays every entry synchronously, pausing between entries to // respect the original timing (adjusted by speed_factor). std::error_code run_all(const Context& ctx) { - const auto& entries = rec_.entries(); + auto entries = rec_.entries(); if (entries.empty()) return {}; auto prev_ts = entries.front().timestamp_ns; @@ -173,23 +263,23 @@ class Playback { } prev_ts = e.timestamp_ns; - Response out; - auto ec = target_->send(ctx, e.cmd, out); + acf::AcfMessageInfo out_resp; + std::vector out_resp_payload; + auto ec = target_(ctx, e.request, e.request_payload, out_resp, out_resp_payload); (void)ec; } return {}; } private: - std::shared_ptr target_; - const Record& rec_; - PlaybackConfig cfg_; + RequestFn target_; + const Record& rec_; + PlaybackConfig cfg_; }; -inline std::shared_ptr new_controller( - std::shared_ptr inner, - std::shared_ptr rec) { - return std::make_shared(std::move(inner), std::move(rec)); +inline std::shared_ptr new_recording_client(RequestFn inner, + std::shared_ptr rec) { + return std::make_shared(std::move(inner), std::move(rec)); } } // namespace record diff --git a/include/rcp/shmem.hpp b/include/rcp/shmem.hpp index 97165d3..4ecedbc 100644 --- a/include/rcp/shmem.hpp +++ b/include/rcp/shmem.hpp @@ -7,214 +7,164 @@ // fusa:req REQ-SHMEM-007 // fusa:req REQ-SHMEM-008 -// Zero-copy intra-host command delivery via shared in-process memory. +// Zero-copy intra-host request delivery via shared in-process memory. // -// shmem::Controller and shmem::ZoneServer communicate through buffered -// in-process channels (queue + condition variable), avoiding serialisation -// overhead when both sides run in the same process. This is the default -// "shared memory" implementation for unit testing and single-process deployments. +// ROADMAP.md milestone 58, "Auxiliary Transport & Cross-Cutting Rebind +// (v2.14.0)": this header is ADAPTed, per the Satellite Package +// Disposition table's entry for `shmem.hpp` — the "avoid serialisation +// overhead for a co-located RC Client/RC Server" value proposition is +// unaffected by the protocol replacement, so the concept survives; only +// the request/response shapes it carries change. shmem::Channel and +// shmem::Registry below play the same role rcp/udp.hpp's Server/Client and +// Registry-shaped lookup play for a real socket transport, just for two +// endpoints that happen to live in the same process: no bytes are ever +// serialised to (or decoded from) an AVTPDU/ACF wire encoding, since both +// sides already share the same acf::AcfMessageInfo/std::vector +// objects in memory. // -// For true OS shared memory across processes (shm_open/mmap), define -// RCP_SHMEM_POSIX. That path is Linux-only and is planned for v0.8.1+. +// This header has no clock, thread, or socket of its own — Channel:: +// request() dispatches synchronously on the calling thread, same +// "primitives driven by the embedding application" convention every Phase +// 14/15 header has used since v2.9.0. #pragma once -#include "rcp.hpp" +#include "acf.hpp" +#include "avtp.hpp" +#include "rcp.hpp" // for rcp::ErrClosed/ErrNotFound/ErrAlreadyExists only — see this header's own scope note above -#include #include +#include #include #include #include #include #include -#include #include namespace rcp { namespace shmem { -// ── ZoneServer ──────────────────────────────────────────────────────────────── +// ── Channel ─────────────────────────────────────────────────────────────────── -// ZoneServer is the zone-controller side of the in-process transport. -// Create one per zone, then pair it with a Controller. -class ZoneServer { +// Channel is the in-process, zero-copy analog of rcp/udp.hpp's Server: +// a caller-supplied Handler answers each request directly, in the same +// process, with no wire encode/decode step in between. Handler is shaped +// identically to udp::Server::Handler so the same handler (e.g. +// rcp::mock::Server::dispatch) can be wired to either transport. +class Channel { public: - using Handler = std::function; + using Handler = std::function& req_payload, + acf::AcfMessageInfo& out_resp, + std::vector& out_resp_payload)>; - explicit ZoneServer(Zone zone) : zone_(zone) {} + explicit Channel(uint64_t stream_key) : stream_key_(stream_key) {} - Zone zone() const noexcept { return zone_; } + uint64_t stream_key() const noexcept { return stream_key_; } void set_handler(Handler h) { std::lock_guard lk(mu_); handler_ = std::move(h); } - void set_healthy(bool v) { healthy_.store(v, std::memory_order_release); } + // request delivers `req`/`req_payload` straight to the registered + // Handler and returns its response, with no serialisation step between + // caller and handler. If no Handler is set, the default response is + // ResponseKind::Acknowledge — the same "answer something, don't hang" + // default rcp/udp.hpp's Server uses when unhandled. + std::error_code request(size_t client, const acf::AcfMessageInfo& req, + const std::vector& req_payload, + acf::AcfMessageInfo& out_resp, + std::vector& out_resp_payload) { + if (closed_.load(std::memory_order_acquire)) return ErrClosed; - // publish sends a Status to all active subscribers. - void publish(const std::vector& payload) { - uint32_t seq = ++seq_; - Status st{zone_, seq, healthy_.load(), payload}; std::lock_guard lk(mu_); - for (auto& s : subs_) s->push(st); - } - - // dispatch_one processes one pending command (called by Controller). - // Returns false if the server is closed. - bool dispatch_one(const Command& cmd, Response& out) { - if (closed_.load()) return false; - std::lock_guard lk(mu_); - if (handler_) out = handler_(cmd); - else out = Response{cmd.id, zone_, ResponseStatus::OK, {}}; - return true; - } - - // add_sub registers a subscriber channel. - void add_sub(std::shared_ptr ch) { - std::lock_guard lk(mu_); - subs_.push_back(std::move(ch)); - } - - // remove_sub deregisters a subscriber channel. - void remove_sub(const std::shared_ptr& ch) { - std::lock_guard lk(mu_); - auto it = std::find(subs_.begin(), subs_.end(), ch); - if (it != subs_.end()) subs_.erase(it); - } - - void close() { - closed_.store(true, std::memory_order_release); - std::lock_guard lk(mu_); - for (auto& s : subs_) s->close(); - subs_.clear(); - } - - bool ok() const noexcept { return !closed_.load(); } - -private: - Zone zone_; - Handler handler_; - std::atomic closed_{false}; - std::atomic healthy_{true}; - std::atomic seq_{0}; - std::mutex mu_; - std::vector> subs_; -}; - -// ── Controller ──────────────────────────────────────────────────────────────── - -class Controller final : public rcp::Controller { -public: - explicit Controller(std::shared_ptr server) - : server_(std::move(server)) {} - - Zone zone() const noexcept override { return server_->zone(); } - - std::error_code send(const rcp::Context& ctx, - const Command& cmd, - Response& out) override { - if (closed_.load()) return ErrClosed; - if (ctx.done()) return ErrTimeout; - if (cmd.zone != server_->zone()) return ErrZoneMismatch; - - // Copy payload (REQ-SHMEM-006: no aliasing). - Command safe = cmd; - if (!cmd.payload.empty()) safe.payload = cmd.payload; - - if (!server_->dispatch_one(safe, out)) return ErrClosed; + if (handler_) { + return handler_(client, req, req_payload, out_resp, out_resp_payload); + } + out_resp = acf::make_response(req, acf::ResponseKind::Acknowledge); + out_resp_payload.clear(); return {}; } - std::error_code subscribe(const rcp::Context& ctx, - std::shared_ptr& out) override { - if (closed_.load()) return ErrClosed; - auto ch = std::make_shared(16); - server_->add_sub(ch); - std::thread([this, weak_ch = std::weak_ptr(ch), ctx]() mutable { - while (!ctx.done() && !closed_.load()) { - auto c = weak_ch.lock(); - if (!c || c->is_closed()) break; - std::this_thread::sleep_for(std::chrono::milliseconds(1)); - } - auto c = weak_ch.lock(); - if (!c) return; - server_->remove_sub(c); - c->close(); - }).detach(); - out = std::move(ch); - return {}; - } + // close is idempotent — safe to call more than once. + void close() { closed_.store(true, std::memory_order_release); } - std::error_code close() override { - closed_.store(true, std::memory_order_release); - return {}; - } + bool ok() const noexcept { return !closed_.load(std::memory_order_acquire); } private: - std::shared_ptr server_; - std::atomic closed_{false}; + uint64_t stream_key_; + std::atomic closed_{false}; + std::mutex mu_; + Handler handler_; }; // ── Registry ────────────────────────────────────────────────────────────────── -class Registry final : public rcp::Registry { +// Registry looks up a Channel by its stream_key, the same opaque +// uint64_t-keyed lookup role rcp/watchdog.hpp's Manager and +// rcp/regmap.hpp's Ep0 already establish for this codebase (typically an +// avtp::StreamId::to_u64()). +class Registry { public: - std::error_code add_server(std::shared_ptr server) { - return register_ctrl(std::make_shared(std::move(server))); - } - - std::error_code register_ctrl(std::shared_ptr ctrl) override { + std::error_code add_channel(std::shared_ptr ch) { std::unique_lock lk(mu_); if (closed_) return ErrClosed; - if (ctrls_.count(ctrl->zone())) return ErrAlreadyExists; - ctrls_[ctrl->zone()] = std::move(ctrl); + if (channels_.count(ch->stream_key())) return ErrAlreadyExists; + channels_[ch->stream_key()] = std::move(ch); return {}; } - std::error_code deregister(Zone z) override { + std::error_code deregister(uint64_t stream_key) { std::unique_lock lk(mu_); - auto it = ctrls_.find(z); - if (it == ctrls_.end()) return ErrNotFound; - auto ctrl = it->second; - ctrls_.erase(it); + auto it = channels_.find(stream_key); + if (it == channels_.end()) return ErrNotFound; + auto ch = it->second; + channels_.erase(it); lk.unlock(); - return ctrl->close(); + ch->close(); + return {}; } - std::error_code lookup(Zone z, std::shared_ptr& out) override { + std::error_code lookup(uint64_t stream_key, std::shared_ptr& out) { std::shared_lock lk(mu_); if (closed_) return ErrClosed; - auto it = ctrls_.find(z); - if (it == ctrls_.end()) return ErrNotFound; + auto it = channels_.find(stream_key); + if (it == channels_.end()) return ErrNotFound; out = it->second; return {}; } - std::vector> controllers() override { + std::vector> channels() { std::shared_lock lk(mu_); - std::vector> out; - out.reserve(ctrls_.size()); - for (auto& kv : ctrls_) out.push_back(kv.second); + std::vector> out; + out.reserve(channels_.size()); + for (auto& kv : channels_) out.push_back(kv.second); return out; } - std::error_code close() override { + // close is idempotent — safe to call more than once. + std::error_code close() { std::unique_lock lk(mu_); if (closed_) return {}; closed_ = true; - auto local = std::move(ctrls_); + auto local = std::move(channels_); lk.unlock(); - for (auto& kv : local) { auto ec = kv.second->close(); (void)ec; } + for (auto& kv : local) kv.second->close(); return {}; } private: - mutable std::shared_mutex mu_; - std::map> ctrls_; - bool closed_ = false; + mutable std::shared_mutex mu_; + std::map> channels_; + bool closed_ = false; }; +inline std::shared_ptr new_channel(uint64_t stream_key) { + return std::make_shared(stream_key); +} + inline std::unique_ptr new_registry() { return std::make_unique(); } diff --git a/include/rcp/tls.hpp b/include/rcp/tls.hpp index 75a917c..cf93a8d 100644 --- a/include/rcp/tls.hpp +++ b/include/rcp/tls.hpp @@ -9,16 +9,41 @@ // fusa:req REQ-TLS-009 // fusa:req REQ-TLS-010 -// Mutual TLS transport for zone controller communication (SG-006, IEC 62443 SL-2). +// Secure-channel option for the UDP/IP transport variant (SG-006, IEC 62443 +// SL-2): DTLS or an application-layer encrypted channel wrapping +// rcp/udp.hpp's Server/Client. // -// Full implementation requires OpenSSL or an equivalent TLS library. -// Define RCP_TLS_OPENSSL and link -lssl -lcrypto to enable the real backend. -// Without that flag, this header provides the interface types and a stub that -// returns std::errc::function_not_supported, so the rest of the codebase can -// be built and tested without TLS on CI hosts that lack OpenSSL. +// ROADMAP.md milestone 58, "Auxiliary Transport & Cross-Cutting Rebind +// (v2.14.0)": this header is ADAPTed, per the Satellite Package Disposition +// table's entry for `tls.hpp`, to build on rcp/udp.hpp's udp::Server/ +// udp::Client (v2.13.0) instead of the pre-replacement Zone-addressed +// ZoneServer/Controller pair. SecureClient/SecureServer below carry the +// same certificate/CA configuration surface as the pre-replacement Config, +// and preserve the same "interface + stub" split: full support requires +// OpenSSL or an equivalent (D)TLS library. Define RCP_TLS_OPENSSL and link +// -lssl -lcrypto to enable a real backend; without that flag every transport +// operation below returns std::errc::function_not_supported rather than +// ever falling back to sending a request in the clear, so the rest of the +// codebase (and CI hosts without OpenSSL) can build and test against this +// header's interface regardless. +// +// IMPORTANT — link-security posture: the specification's own preferred +// mechanism for securing the link this protocol rides on is MACsec (IEEE +// 802.1AE) operating at layer 2, beneath native AVTP-over-Ethernet framing +// (extraction §3.12, §1.3). This package does not implement, and does not +// otherwise address, MACsec at all — it only ever secures the UDP/IP +// transport variant (rcp/udp.hpp's Annex J encapsulation) at the +// application layer. An embedding that has native Ethernet framing +// available end to end should prefer a MACsec-capable NIC/switch fabric +// over deploying this header at all; this header exists for the UDP/IP +// path specifically, where MACsec's layer-2 guarantee does not apply once +// the AVTPDU is riding inside an ordinary UDP/IP datagram. #pragma once -#include "rcp.hpp" +#include "acf.hpp" +#include "avtp.hpp" +#include "rcp.hpp" // for rcp::Context/ErrClosed/ErrTimeout only — see this header's own scope note above +#include "udp.hpp" #include #include @@ -28,6 +53,9 @@ namespace rcp { namespace tls { // ── Config ──────────────────────────────────────────────────────────────────── +// Unchanged in shape from the pre-replacement design — the certificate/CA +// configuration surface this package exposes is orthogonal to which wire +// protocol rides on top of the secured channel. struct Config { std::string cert_file; // PEM certificate for this endpoint @@ -36,67 +64,92 @@ struct Config { bool verify_peer = true; // enforce mutual authentication }; -// ── Controller (interface) ──────────────────────────────────────────────────── - -// Controller connects to a ZoneServer over mTLS. -// Use new_controller() to obtain an implementation. -class Controller final : public rcp::Controller { +namespace detail { +inline std::error_code unsupported() noexcept { + return std::make_error_code(std::errc::function_not_supported); +} +} // namespace detail + +// ── SecureClient ────────────────────────────────────────────────────────────── +// SecureClient wraps a udp::Client, adding a DTLS (or application-layer) +// encrypted channel per `cfg` before any AVTPDU-framed ACF request leaves +// this process. Its request() signature mirrors udp::Client::request's +// core shape so a caller can swap one for the other without reshaping its +// own request-building code. +class SecureClient { public: - Controller(Zone z, const char*, uint16_t, Config) - : zone_(z) {} - - Zone zone() const noexcept override { return zone_; } - - std::error_code send(const rcp::Context&, const Command&, Response&) override { - return std::make_error_code(std::errc::function_not_supported); + SecureClient(avtp::StreamId stream_id, const char* server_host, uint16_t server_port, + Config cfg) + : stream_id_(stream_id), cfg_(std::move(cfg)) { + (void)server_host; + (void)server_port; } - std::error_code subscribe(const rcp::Context&, - std::shared_ptr&) override { - return std::make_error_code(std::errc::function_not_supported); + std::error_code request(const rcp::Context& ctx, + const acf::AcfMessageInfo& req, + const std::vector& req_payload, + acf::AcfMessageInfo& out_resp, + std::vector& out_resp_payload) { + (void)ctx; (void)req; (void)req_payload; (void)out_resp; (void)out_resp_payload; + return detail::unsupported(); } - std::error_code close() override { return {}; } + std::error_code close() { return {}; } + + bool ok() const noexcept { return false; } + + avtp::StreamId stream_id() const noexcept { return stream_id_; } + const Config& config() const noexcept { return cfg_; } private: - Zone zone_; + avtp::StreamId stream_id_; + Config cfg_; }; -// ── ZoneServer (interface stub) ─────────────────────────────────────────────── - -class ZoneServer { +// ── SecureServer ────────────────────────────────────────────────────────────── +// SecureServer wraps a udp::Server the same way SecureClient wraps +// udp::Client. Handler is shaped identically to udp::Server::Handler so the +// same handler (e.g. rcp::mock::Server::dispatch) can be wired to either. +class SecureServer { public: - ZoneServer(Zone, const char*, uint16_t, Config) {} - void set_handler(std::function) {} - void set_healthy(bool) {} - void publish(const std::vector&) {} + using Handler = udp::Server::Handler; + + SecureServer(avtp::StreamId stream_id, const char* addr, uint16_t port, Config cfg) + : stream_id_(stream_id), cfg_(std::move(cfg)) { + (void)addr; + (void)port; + } + + void set_handler(Handler h) { handler_ = std::move(h); } + + std::string addr_string() const { return {}; } + uint16_t port() const { return 0; } + void close() {} - bool ok() const noexcept { return false; } -}; -// ── Registry ────────────────────────────────────────────────────────────────── + bool ok() const noexcept { return false; } -class Registry final : public rcp::Registry { -public: - std::error_code dial(Zone, const char*, uint16_t, Config) { - return std::make_error_code(std::errc::function_not_supported); - } + avtp::StreamId stream_id() const noexcept { return stream_id_; } + const Config& config() const noexcept { return cfg_; } - std::error_code register_ctrl(std::shared_ptr) override { - return std::make_error_code(std::errc::function_not_supported); - } - std::error_code deregister(Zone) override { - return std::make_error_code(std::errc::function_not_supported); - } - std::error_code lookup(Zone, std::shared_ptr&) override { - return std::make_error_code(std::errc::function_not_supported); - } - std::vector> controllers() override { return {}; } - std::error_code close() override { return {}; } +private: + avtp::StreamId stream_id_; + Config cfg_; + Handler handler_; }; -inline std::unique_ptr new_registry() { - return std::make_unique(); +inline std::unique_ptr new_secure_client(avtp::StreamId stream_id, + const char* server_host, + uint16_t server_port, + Config cfg) { + return std::make_unique(stream_id, server_host, server_port, std::move(cfg)); +} + +inline std::unique_ptr new_secure_server(avtp::StreamId stream_id, + const char* addr, + uint16_t port, + Config cfg) { + return std::make_unique(stream_id, addr, port, std::move(cfg)); } } // namespace tls diff --git a/include/rcp/tsn.hpp b/include/rcp/tsn.hpp index 70bd347..633fd79 100644 --- a/include/rcp/tsn.hpp +++ b/include/rcp/tsn.hpp @@ -5,28 +5,46 @@ // fusa:req REQ-TSN-005 // fusa:req REQ-TSN-006 -// IEEE 802.1Qbv-aware UDP transport adapter for hard real-time Ethernet. +// IEEE 802.1Qbv-aware priority hint for the UDP/IP transport, keyed off the +// specification's own execution-priority ordering across request kinds. // -// Maps rcp::Priority values to IEEE 802.1p PCP (Priority Code Point) classes -// and applies SO_PRIORITY on the UDP socket so the Linux traffic shaper routes -// frames into the correct egress queue. +// ROADMAP.md milestone 58, "Auxiliary Transport & Cross-Cutting Rebind +// (v2.14.0)": this header is ADAPTed, per the Satellite Package +// Disposition table's entry for `tsn.hpp` — this file's own pre-v2.14.0 +// header comment already flagged this exact rebind as pending. The old +// 3-level rcp::Priority enum (rcp/prioqueue.hpp, DEPRECATEd outright per +// the disposition table, no TC18 analog) is gone; PCPMap below instead maps +// rcp::request::RequestCategory (v2.5.0) — the seven-way cancellation > +// triggered > timed > compound > compound-wait > chained > standard +// ordering extraction §3.14 already defines for simultaneously-due +// requests — onto IEEE 802.1p PCP classes. apply_priority() is a free +// function taking a raw socket fd, same as the pre-replacement Controller +// wrapper's only real dependency (its own header comment already noted it +// "depends only on the socket fd passed in, not on any concrete transport +// header"); there is no unified client-side send() chokepoint yet to wrap +// (that unification, if any, does not land until the CLI/capi/adapt +// rebuilds at v2.16.0 — see rcp/authz.hpp's equivalent v2.11.0 note), so +// this is a primitive an embedding calls directly before udp::Client:: +// request, the same "primitives driven by the embedding application" +// pattern every Phase 14/15 header has used since v2.9.0. // -// Full 802.1Qbv gate scheduling requires a TSN-capable NIC and kernel ≥ 4.15. -// On standard hardware this provides best-effort priority mapping only. -// -// Wraps any rcp::Controller (e.g. the pre-replacement rcp/legacy_mock.hpp's -// Controller); it depends only on the socket fd passed in, not on any -// concrete transport header. See ROADMAP.md's Satellite Package Disposition -// table's entry for this file — a v2.14.0 rebind to the post-replacement -// request model, once rcp/udp.hpp itself no longer implements this -// pre-replacement Controller interface (v2.13.0) — for why this class still -// targets the old model today. +// Full 802.1Qbv gate scheduling requires a TSN-capable NIC and kernel +// >= 4.15; on standard hardware this provides best-effort priority mapping +// only. Where genuine IEEE 1722 stream reservation (802.1Qat SRP) is +// available on the platform/NIC, prefer it over this socket-level +// SO_PRIORITY hint — SRP carries an admission-controlled bandwidth +// reservation end to end, while SO_PRIORITY only ever influences local +// egress queueing on this host; this header does not implement SRP itself, +// same "primitive, not the full mechanism" scope as this file's other +// pre-v2.14.0 disclaimers. #pragma once -#include "rcp.hpp" +#include "rcp.hpp" // for std::error_code convention only — see this header's own scope note above +#include "request.hpp" +#include #include -#include +#include #if defined(__linux__) # include @@ -37,22 +55,32 @@ namespace rcp { namespace tsn { // ── PCPMap ──────────────────────────────────────────────────────────────────── - -// PCPMap maps each rcp::Priority to an IEEE 802.1p PCP value (0–7). -// PCP 7 is the highest priority (used for Priority::Critical). +// Maps each rcp::request::RequestCategory to an IEEE 802.1p PCP value +// (0-7). Category rank 0 (Cancellation) is the highest execution priority +// (extraction §3.14) and gets the highest default PCP; rank 6 (Standard, +// the mandatory baseline kind) gets the lowest. struct PCPMap { - uint8_t normal = 2; // PCP for Priority::Normal - uint8_t high = 5; // PCP for Priority::High - uint8_t critical = 7; // PCP for Priority::Critical + uint8_t cancellation = 7; // RequestCategory::Cancellation (rank 0, highest) + uint8_t triggered = 6; // RequestCategory::Triggered (rank 1) + uint8_t timed = 5; // RequestCategory::Timed (rank 2) + uint8_t compound = 4; // RequestCategory::Compound (rank 3) + uint8_t compound_wait = 3; // RequestCategory::CompoundWait (rank 4) + uint8_t chained = 2; // RequestCategory::Chained (rank 5) + uint8_t standard = 1; // RequestCategory::Standard (rank 6, lowest) }; inline PCPMap default_pcp_map() { return PCPMap{}; } -inline uint8_t pcp_for(const PCPMap& m, Priority p) noexcept { - switch (p) { - case Priority::High: return m.high; - case Priority::Critical: return m.critical; - default: return m.normal; +inline uint8_t pcp_for(const PCPMap& m, request::RequestCategory c) noexcept { + switch (c) { + case request::RequestCategory::Cancellation: return m.cancellation; + case request::RequestCategory::Triggered: return m.triggered; + case request::RequestCategory::Timed: return m.timed; + case request::RequestCategory::Compound: return m.compound; + case request::RequestCategory::CompoundWait: return m.compound_wait; + case request::RequestCategory::Chained: return m.chained; + case request::RequestCategory::Standard: + default: return m.standard; } } @@ -60,54 +88,36 @@ inline uint8_t pcp_for(const PCPMap& m, Priority p) noexcept { struct TSNConfig { PCPMap pcp_map; - int vlan_id = 0; // 0 = untagged - int cycle_ns = 0; // 802.1Qbv gate cycle in nanoseconds (0 = disabled) + int vlan_id = 0; // 0 = untagged + int cycle_ns = 0; // 802.1Qbv gate cycle in nanoseconds (0 = disabled) }; inline TSNConfig default_tsn_config() { return TSNConfig{}; } -// ── Controller wrapper ──────────────────────────────────────────────────────── - -// Controller wraps any rcp::Controller and applies SO_PRIORITY to the kernel -// socket before each send() so the egress qdisc places the frame in the correct -// 802.1p traffic class. -class Controller final : public rcp::Controller { -public: - Controller(std::shared_ptr inner, - int socket_fd, - TSNConfig cfg = {}) - : inner_(std::move(inner)), fd_(socket_fd), cfg_(cfg) {} - - Zone zone() const noexcept override { return inner_->zone(); } - - std::error_code send(const rcp::Context& ctx, - const Command& cmd, - Response& out) override { +// ── apply_priority ──────────────────────────────────────────────────────────── +// apply_priority sets SO_PRIORITY on `fd` to the PCP value `cfg.pcp_map` +// assigns `category`, so the egress qdisc places the next datagram sent on +// this socket into the correct 802.1p traffic class. A caller invokes this +// once per outbound request, immediately before udp::Client::request, with +// the RequestCategory that request's decoded (or about-to-be-encoded) +// request_type opcode maps to via rcp::request::category_of. +// +// Outside RCP_TSN_SO_PRIORITY (non-Linux platforms, where this hint has no +// portable socket-option equivalent) this is a no-op success rather than a +// failure — the caller's send path is unaffected either way, same as the +// pre-replacement Controller wrapper's own `#if defined(...)` gating. +inline std::error_code apply_priority(int fd, const TSNConfig& cfg, + request::RequestCategory category) noexcept { + if (fd < 0) return std::make_error_code(std::errc::invalid_argument); #if defined(RCP_TSN_SO_PRIORITY) - int pcp = static_cast(pcp_for(cfg_.pcp_map, cmd.priority)); - if (fd_ >= 0) - ::setsockopt(fd_, SOL_SOCKET, SO_PRIORITY, &pcp, sizeof(pcp)); + int pcp = static_cast(pcp_for(cfg.pcp_map, category)); + if (::setsockopt(fd, SOL_SOCKET, SO_PRIORITY, &pcp, sizeof(pcp)) < 0) + return std::error_code(errno, std::generic_category()); +#else + (void)cfg; + (void)category; #endif - return inner_->send(ctx, cmd, out); - } - - std::error_code subscribe(const rcp::Context& ctx, - std::shared_ptr& out) override { - return inner_->subscribe(ctx, out); - } - - std::error_code close() override { return inner_->close(); } - -private: - std::shared_ptr inner_; - [[maybe_unused]] int fd_; - TSNConfig cfg_; -}; - -inline std::shared_ptr new_controller(std::shared_ptr inner, - int socket_fd, - TSNConfig cfg = {}) { - return std::make_shared(std::move(inner), socket_fd, cfg); + return {}; } } // namespace tsn diff --git a/include/rcp/version.hpp b/include/rcp/version.hpp index 15ee9ee..25b4251 100644 --- a/include/rcp/version.hpp +++ b/include/rcp/version.hpp @@ -6,6 +6,6 @@ namespace rcp { // Semantic version of the cpp-RCP implementation (matches the latest git tag). -constexpr std::string_view kVersion = "2.13.0"; +constexpr std::string_view kVersion = "2.14.0"; } // namespace rcp diff --git a/tests/test_loan.cpp b/tests/test_loan.cpp index 7f4ad6e..437820f 100644 --- a/tests/test_loan.cpp +++ b/tests/test_loan.cpp @@ -5,90 +5,94 @@ // fusa:test REQ-LOAN-005 // fusa:test REQ-LOAN-006 +// Tests for rcp/loan.hpp — BufferPool, the zero-copy payload-loaning +// primitive (ROADMAP.md milestone 58, "Auxiliary Transport & Cross-Cutting +// Rebind", v2.14.0). + #include -#include #include using namespace rcp; -static std::shared_ptr make_mock(Zone z = Zone::FrontLeft) { - return std::make_shared(z); -} - // ── loan() ──────────────────────────────────────────────────────────────────── -TEST_CASE("loan::Controller::loan returns zeroed buffer of requested size", "[loan][REQ-LOAN-001]") { - auto inner = make_mock(); - loan::Controller lc(inner); +TEST_CASE("loan::BufferPool::loan returns a zeroed buffer of exactly size bytes", + "[loan][REQ-LOAN-001]") { + loan::BufferPool pool; std::unique_ptr loan_out; - REQUIRE_FALSE(lc.loan(16, loan_out)); + REQUIRE_FALSE(pool.loan(16, loan_out)); REQUIRE(loan_out != nullptr); REQUIRE(loan_out->payload.size() == 16); for (auto b : loan_out->payload) REQUIRE(b == 0); } -TEST_CASE("loan::Controller::loan returns ErrClosed when closed", "[loan][REQ-LOAN-002]") { - auto inner = make_mock(); - loan::Controller lc(inner); - { auto ec = lc.close(); (void)ec; } +TEST_CASE("loan::BufferPool::loan returns ErrClosed once the pool is closed", + "[loan][REQ-LOAN-002]") { + loan::BufferPool pool; + pool.close(); std::unique_ptr loan_out; - auto ec = lc.loan(8, loan_out); - REQUIRE(ec == ErrClosed); + REQUIRE(pool.loan(8, loan_out) == ErrClosed); } -TEST_CASE("loan::Controller::loan returns error for negative size", "[loan][REQ-LOAN-003]") { - auto inner = make_mock(); - loan::Controller lc(inner); +TEST_CASE("loan::BufferPool::loan returns an error for negative size", + "[loan][REQ-LOAN-003]") { + loan::BufferPool pool; std::unique_ptr loan_out; - auto ec = lc.loan(-1, loan_out); - REQUIRE(ec); // any error is acceptable for invalid size + auto ec = pool.loan(-1, loan_out); + REQUIRE(ec == std::make_error_code(std::errc::invalid_argument)); } -// ── send_loaned() ───────────────────────────────────────────────────────────── +// ── Pool reuse ──────────────────────────────────────────────────────────────── -TEST_CASE("loan::Controller::send_loaned delivers command and returns OK", "[loan][REQ-LOAN-004]") { - auto inner = make_mock(); - loan::Controller lc(inner); +TEST_CASE("A released Loan's buffer becomes available for a subsequent loan() to reuse", + "[loan][REQ-LOAN-004]") { + loan::BufferPool pool; + REQUIRE(pool.pooled_count() == 0); - std::unique_ptr loan_out; - REQUIRE_FALSE(lc.loan(4, loan_out)); - loan_out->payload = {0x01, 0x02, 0x03, 0x04}; - - Command cmd; - cmd.zone = Zone::FrontLeft; - cmd.payload = loan_out->payload; - loan_out->ret(); // release back to pool (simulate send_loaned transfer) - - Response resp; - auto ec = lc.send_loaned(Context::background(), cmd, resp); - REQUIRE_FALSE(ec); - REQUIRE(resp.status == ResponseStatus::OK); + { + std::unique_ptr loan_out; + REQUIRE_FALSE(pool.loan(64, loan_out)); + } // destructor releases the buffer back to the pool + + REQUIRE(pool.pooled_count() == 1); + + std::unique_ptr loan_out2; + REQUIRE_FALSE(pool.loan(64, loan_out2)); // drawn from the pool, not freshly allocated + REQUIRE(pool.pooled_count() == 0); } -// ── Pool return ─────────────────────────────────────────────────────────────── +// ── Re-zeroing on reuse ─────────────────────────────────────────────────────── -TEST_CASE("Loan RAII releases buffer to pool on destruction", "[loan][REQ-LOAN-005]") { - auto inner = make_mock(); - loan::Controller lc(inner); +TEST_CASE("A released buffer is re-zeroed before being handed out again", + "[loan][REQ-LOAN-005]") { + loan::BufferPool pool; { std::unique_ptr loan_out; - REQUIRE_FALSE(lc.loan(8, loan_out)); - } // loan_out goes out of scope here — should return to pool without crash + REQUIRE_FALSE(pool.loan(4, loan_out)); + loan_out->payload = {0xDE, 0xAD, 0xBE, 0xEF}; // stamp with non-zero data + } // released back to the pool with that data still present - // Loan again to verify pool is reusable. std::unique_ptr loan_out2; - REQUIRE_FALSE(lc.loan(8, loan_out2)); - REQUIRE(loan_out2 != nullptr); + REQUIRE_FALSE(pool.loan(4, loan_out2)); + for (auto b : loan_out2->payload) REQUIRE(b == 0); // no stale data leaked across reuse } -// ── Zone passthrough ────────────────────────────────────────────────────────── +// ── Idempotent close ────────────────────────────────────────────────────────── + +TEST_CASE("loan::BufferPool::close is idempotent and safe with outstanding Loans still alive", + "[loan][REQ-LOAN-006]") { + loan::BufferPool pool; + std::unique_ptr loan_out; + REQUIRE_FALSE(pool.loan(8, loan_out)); + + pool.close(); + pool.close(); // second call must not crash -TEST_CASE("loan::Controller::zone returns inner zone", "[loan][REQ-LOAN-006]") { - auto inner = make_mock(Zone::RearRight); - loan::Controller lc(inner); - REQUIRE(lc.zone() == Zone::RearRight); + // The still-alive Loan's eventual release (destructor below) must not + // crash even though the pool has already been closed. + loan_out.reset(); } diff --git a/tests/test_mdns.cpp b/tests/test_mdns.cpp index eb48340..328fd51 100644 --- a/tests/test_mdns.cpp +++ b/tests/test_mdns.cpp @@ -7,121 +7,134 @@ // fusa:test REQ-MDNS-007 // fusa:test REQ-MDNS-008 -// mDNS / DNS-SD zone-discovery tests (RFC 6762 + RFC 6763). -// -// Exercises the StaticDiscoverer (the header-only Discoverer used for testing -// and static config) and the Announcer interface via a recording test double. +// Tests for rcp/mdns.hpp — host:port discovery for the UDP/IP transport +// variant (ROADMAP.md milestone 58, "Auxiliary Transport & Cross-Cutting +// Rebind", v2.14.0). + #include #include "rcp/mdns.hpp" -#include -#include #include using namespace rcp; +using namespace rcp::mdns; namespace { - -std::vector sample_zones() { - return { - {Zone::FrontLeft, "fl.local", 5000, "front-left.fl.local._rcp._udp.local"}, - {Zone::FrontRight, "fr.local", 5001, "front-right.fr.local._rcp._udp.local"}, - {Zone::Central, "c.local", 5002, "central.c.local._rcp._udp.local"}, - }; +avtp::StreamId make_stream_id(uint8_t mac_seed, uint16_t suffix) { + avtp::StreamId id; + for (auto& b : id.mac) b = mac_seed++; + id.suffix = suffix; + return id; } - -// TestAnnouncer records announce()/withdraw() calls for assertions. -class TestAnnouncer final : public mdns::Announcer { -public: - std::error_code announce(const mdns::ZoneInfo& info) override { - announced[info.zone] = info; - return {}; - } - void withdraw(Zone z) override { announced.erase(z); } - std::map announced; -}; - } // namespace -TEST_CASE("mdns: StaticDiscoverer is a Discoverer that emits on start", - "[mdns][REQ-MDNS-001]") { - auto disc = mdns::new_static_discoverer(sample_zones()); - mdns::Discoverer& as_iface = *disc; // satisfies the abstract interface +// ── Discoverer interface / StaticDiscoverer ────────────────────────────────── - int count = 0; - auto ec = as_iface.start([&](const mdns::DiscoveryEvent&) { ++count; }); - REQUIRE_FALSE(ec); - REQUIRE(count == 3); +TEST_CASE("Discoverer is an abstract interface implemented by StaticDiscoverer", + "[mdns][REQ-MDNS-001]") { + std::unique_ptr d = new_static_discoverer({}); + REQUIRE(d != nullptr); } -TEST_CASE("mdns: start fires an Added event for each discovered zone", +TEST_CASE("StaticDiscoverer::start emits an Added event for each configured server", "[mdns][REQ-MDNS-002][REQ-MDNS-004]") { - auto disc = mdns::new_static_discoverer(sample_zones()); + ServerInfo a{make_stream_id(0x02, 1).to_u64(), "10.0.0.1", 8000, "srv-a"}; + ServerInfo b{make_stream_id(0x02, 2).to_u64(), "10.0.0.2", 8001, "srv-b"}; - std::vector added; - auto ec = disc->start([&](const mdns::DiscoveryEvent& ev) { - REQUIRE(ev.event == mdns::EventType::Added); - added.push_back(ev.info.zone); - }); - REQUIRE_FALSE(ec); + auto d = new_static_discoverer({a, b}); - // Exactly the configured zones, in order (REQ-MDNS-004). - REQUIRE(added == std::vector{Zone::FrontLeft, Zone::FrontRight, Zone::Central}); -} + std::vector events; + REQUIRE_FALSE(d->start([&](const DiscoveryEvent& ev) { events.push_back(ev); })); -TEST_CASE("mdns: stop terminates discovery", "[mdns][REQ-MDNS-003]") { - auto disc = mdns::new_static_discoverer(sample_zones()); - - int count = 0; - // Calling stop() from within the callback halts emission after the first event. - auto ec = disc->start([&](const mdns::DiscoveryEvent&) { - ++count; - disc->stop(); - }); - REQUIRE_FALSE(ec); - REQUIRE(count == 1); // discovery stopped despite three configured zones + REQUIRE(events.size() == 2); + REQUIRE(events[0].event == EventType::Added); + REQUIRE(events[0].info.stream_key == a.stream_key); + REQUIRE(events[0].info.host == "10.0.0.1"); + REQUIRE(events[0].info.port == 8000); + REQUIRE(events[1].info.stream_key == b.stream_key); } -TEST_CASE("mdns: ZoneInfo carries host, port and zone", "[mdns][REQ-MDNS-005]") { - auto disc = mdns::new_static_discoverer(sample_zones()); +TEST_CASE("StaticDiscoverer::stop halts emission of further events", + "[mdns][REQ-MDNS-003]") { + std::vector servers; + for (int i = 0; i < 5; ++i) { + servers.push_back({make_stream_id(0x02, static_cast(i)).to_u64(), + "10.0.0.1", 8000, "srv"}); + } + auto d = new_static_discoverer(servers); - mdns::ZoneInfo first{}; - bool got = false; - auto ec = disc->start([&](const mdns::DiscoveryEvent& ev) { - if (!got) { first = ev.info; got = true; } + int seen = 0; + d->start([&](const DiscoveryEvent&) { + ++seen; + if (seen == 2) d->stop(); }); - REQUIRE_FALSE(ec); - REQUIRE(got); - REQUIRE(first.zone == Zone::FrontLeft); - REQUIRE(first.host == "fl.local"); - REQUIRE(first.port == 5000); + REQUIRE(seen == 2); // stop() during the callback halts the remaining emissions } -TEST_CASE("mdns: instance_name follows the _rcp._udp.local convention", +// ── ServerInfo shape ────────────────────────────────────────────────────────── + +TEST_CASE("ServerInfo carries stream_key, host, port, and instance_name", + "[mdns][REQ-MDNS-005]") { + auto sid = make_stream_id(0x02, 0x99); + ServerInfo info; + info.stream_key = sid.to_u64(); + info.host = "192.168.1.10"; + info.port = 12345; + info.instance_name = make_instance_name(sid, "192.168.1.10"); + + REQUIRE(info.stream_key == sid.to_u64()); + REQUIRE(info.host == "192.168.1.10"); + REQUIRE(info.port == 12345); + REQUIRE_FALSE(info.instance_name.empty()); +} + +// ── make_instance_name convention ──────────────────────────────────────────── + +TEST_CASE("make_instance_name follows .._rcp._udp.local", "[mdns][REQ-MDNS-006]") { - auto name = mdns::make_instance_name(Zone::FrontLeft, "myhost"); - REQUIRE(name == rcp::to_string(Zone::FrontLeft) + ".myhost._rcp._udp.local"); + auto sid = make_stream_id(0x02, 0x1234); + auto name = make_instance_name(sid, "myhost"); + REQUIRE(name.find("._rcp._udp.local") != std::string::npos); + REQUIRE(name.find(".myhost.") != std::string::npos); + + // Two distinct stream_ids must never collide on the same instance name. + auto other = make_instance_name(make_stream_id(0x03, 0x1234), "myhost"); + REQUIRE(name != other); } -TEST_CASE("mdns: Announcer registers a zone record", "[mdns][REQ-MDNS-007]") { - TestAnnouncer ann; - mdns::Announcer& iface = ann; // abstract registration interface +// ── Announcer interface ─────────────────────────────────────────────────────── - mdns::ZoneInfo info{Zone::RearLeft, "rl.local", 6000, "rear-left.rl.local._rcp._udp.local"}; - REQUIRE_FALSE(iface.announce(info)); - REQUIRE(ann.announced.count(Zone::RearLeft) == 1); - REQUIRE(ann.announced[Zone::RearLeft].port == 6000); -} +namespace { +class RecordingAnnouncer final : public Announcer { +public: + std::error_code announce(const ServerInfo& info) override { + announced.push_back(info); + return {}; + } + void withdraw(uint64_t stream_key) override { withdrawn.push_back(stream_key); } -TEST_CASE("mdns: withdraw removes the mDNS record", "[mdns][REQ-MDNS-008]") { - TestAnnouncer ann; - mdns::ZoneInfo info{Zone::RearRight, "rr.local", 6001, "rr._rcp._udp.local"}; - REQUIRE_FALSE(ann.announce(info)); - REQUIRE(ann.announced.count(Zone::RearRight) == 1); + std::vector announced; + std::vector withdrawn; +}; +} // namespace + +TEST_CASE("Announcer interface is implementable and announce() registers a ServerInfo", + "[mdns][REQ-MDNS-007]") { + RecordingAnnouncer a; + ServerInfo info{make_stream_id(0x02, 1).to_u64(), "10.0.0.5", 9000, "srv"}; + REQUIRE_FALSE(a.announce(info)); + REQUIRE(a.announced.size() == 1); + REQUIRE(a.announced[0].stream_key == info.stream_key); +} - ann.withdraw(Zone::RearRight); - REQUIRE(ann.announced.count(Zone::RearRight) == 0); +TEST_CASE("Announcer::withdraw removes the record for the given stream_key", + "[mdns][REQ-MDNS-008]") { + RecordingAnnouncer a; + auto key = make_stream_id(0x02, 1).to_u64(); + a.withdraw(key); + REQUIRE(a.withdrawn.size() == 1); + REQUIRE(a.withdrawn[0] == key); } diff --git a/tests/test_observe.cpp b/tests/test_observe.cpp index f8394d7..ebd21e5 100644 --- a/tests/test_observe.cpp +++ b/tests/test_observe.cpp @@ -6,9 +6,13 @@ // fusa:test REQ-OBS-006 // fusa:test REQ-OBS-007 // fusa:test REQ-OBS-008 + +// Tests for rcp/observe.hpp — OpenTelemetry-style spans/counters around +// RC-Client request/response traffic (ROADMAP.md milestone 58, "Auxiliary +// Transport & Cross-Cutting Rebind", v2.14.0). + #include -#include "rcp/legacy_mock.hpp" #include "rcp/observe.hpp" #include @@ -18,18 +22,20 @@ #include using namespace rcp; +using namespace rcp::observe; namespace { -// CountingSink records counter deltas by metric name and the last span seen, -// so tests can assert on rcp.commands.total / rcp.commands.errors and span.result. -class CountingSink final : public observe::MetricsSink { +// CountingSink records counter deltas by metric name and the last span +// seen, so tests can assert on rcp.requests.total / rcp.requests.errors +// and span.result. +class CountingSink final : public MetricsSink { public: - void record_span(const observe::Span& s) override { + void record_span(const Span& s) override { std::lock_guard lk(mu_); last_span_ = s; } - void record_gauge(const observe::Metric&) override {} - void record_counter(const std::string& name, Zone, double delta) override { + void record_gauge(const Metric&) override {} + void record_counter(const std::string& name, uint64_t, double delta) override { std::lock_guard lk(mu_); counters_[name] += delta; } @@ -38,115 +44,98 @@ class CountingSink final : public observe::MetricsSink { auto it = counters_.find(name); return it == counters_.end() ? 0.0 : it->second; } - observe::Span last_span() const { + Span last_span() const { std::lock_guard lk(mu_); return last_span_; } private: mutable std::mutex mu_; std::map counters_; - observe::Span last_span_{}; + Span last_span_{}; }; + +acf::AcfMessageInfo standard_request(avtp::ByteBusId bus_id, uint8_t transaction_num) { + return acf::make_standard_request(bus_id, transaction_num, /*write=*/false, /*read_size=*/2); +} + +RequestFn ok_request(std::error_code result = {}) { + return [result](const Context&, const acf::AcfMessageInfo& req, const std::vector&, + acf::AcfMessageInfo& out_resp, std::vector&) { + out_resp = acf::make_response(req, acf::ResponseKind::Acknowledge); + return result; + }; +} } // namespace -TEST_CASE("observe: span recorded on successful send", "[observe]") { - auto sink = std::make_shared(); - auto inner = std::make_shared(Zone::FrontLeft); - auto ctrl = observe::new_controller(inner, sink); +TEST_CASE("observe: span recorded on successful request", "[observe][REQ-OBS-001]") { + auto sink = std::make_shared(); + auto oc = new_observing_client(ok_request(), /*stream_key=*/1, sink); - Command cmd; - cmd.zone = Zone::FrontLeft; - cmd.type = CommandType::Get; - Response resp; - REQUIRE_FALSE(ctrl->send(Context{}, cmd, resp)); + auto req = standard_request(3, 1); + acf::AcfMessageInfo resp; + std::vector resp_payload; + REQUIRE_FALSE(oc->request(Context{}, req, {}, resp, resp_payload)); REQUIRE(sink->span_count() == 1); - auto spans = sink->spans(); - REQUIRE(spans[0].zone == Zone::FrontLeft); - REQUIRE(spans[0].cmd_type == CommandType::Get); - REQUIRE(!spans[0].result); } -TEST_CASE("observe: multiple sends accumulate spans", "[observe]") { - auto sink = std::make_shared(); - auto inner = std::make_shared(Zone::FrontLeft); - auto ctrl = observe::new_controller(inner, sink); - - for (int i = 0; i < 5; ++i) { - Command cmd; - cmd.zone = Zone::FrontLeft; - cmd.type = CommandType::Set; - Response resp; - auto ec = ctrl->send(Context{}, cmd, resp); - (void)ec; - } +TEST_CASE("observe: span captures byte_bus_id and acf_msg_type", "[observe][REQ-OBS-002]") { + auto sink = std::make_shared(); + auto oc = new_observing_client(ok_request(), 1, sink); + + auto req = standard_request(/*bus_id=*/7, 1); + acf::AcfMessageInfo resp; + std::vector resp_payload; + REQUIRE_FALSE(oc->request(Context{}, req, {}, resp, resp_payload)); - REQUIRE(sink->span_count() == 5); + auto spans = sink->spans(); + REQUIRE(spans[0].byte_bus_id == 7); + REQUIRE(spans[0].acf_msg_type == acf::kAcfMsgTypeAbb); } -TEST_CASE("observe: span duration is non-negative", "[observe]") { - auto sink = std::make_shared(); - auto inner = std::make_shared(Zone::Central); - auto ctrl = observe::new_controller(inner, sink); +TEST_CASE("observe: span duration is non-negative", "[observe][REQ-OBS-003]") { + auto sink = std::make_shared(); + auto oc = new_observing_client(ok_request(), 1, sink); - Command cmd; - cmd.zone = Zone::Central; - cmd.type = CommandType::Watchdog; - Response resp; - auto ec = ctrl->send(Context{}, cmd, resp); + auto req = standard_request(1, 1); + acf::AcfMessageInfo resp; + std::vector resp_payload; + auto ec = oc->request(Context{}, req, {}, resp, resp_payload); (void)ec; REQUIRE(sink->span_count() == 1); REQUIRE(sink->spans()[0].duration().count() >= 0); } -TEST_CASE("observe: noop sink does not crash", "[observe][REQ-OBS-005]") { - auto inner = std::make_shared(Zone::FrontLeft); - auto ctrl = observe::new_controller(inner); // default noop sink - - Command cmd; - cmd.zone = Zone::FrontLeft; - cmd.type = CommandType::Get; - Response resp; - REQUIRE_FALSE(ctrl->send(Context{}, cmd, resp)); -} +TEST_CASE("observe: span captures error code and error counter increments on failure", + "[observe][REQ-OBS-004][REQ-OBS-008]") { + auto sink = std::make_shared(); + auto oc = new_observing_client(ok_request(ErrClosed), 1, sink); -TEST_CASE("observe: commands.total counter increments per send", "[observe][REQ-OBS-007]") { - auto sink = std::make_shared(); - auto inner = std::make_shared(Zone::FrontLeft); - auto ctrl = observe::new_controller(inner, sink); + auto req = standard_request(1, 1); + acf::AcfMessageInfo resp; + std::vector resp_payload; + auto ec = oc->request(Context{}, req, {}, resp, resp_payload); + REQUIRE(ec == ErrClosed); - for (int i = 0; i < 3; ++i) { - Command cmd; cmd.zone = Zone::FrontLeft; cmd.type = CommandType::Set; - Response resp; - REQUIRE_FALSE(ctrl->send(Context{}, cmd, resp)); - } - REQUIRE(sink->counter("rcp.commands.total") == 3.0); - REQUIRE(sink->counter("rcp.commands.errors") == 0.0); + REQUIRE(sink->last_span().result == ErrClosed); // span records the error (REQ-OBS-004) + REQUIRE(sink->counter("rcp.requests.total") == 1.0); + REQUIRE(sink->counter("rcp.requests.errors") == 1.0); // error counter (REQ-OBS-008) } -TEST_CASE("observe: span captures error code and error counter increments on failure", - "[observe][REQ-OBS-004][REQ-OBS-008]") { - auto sink = std::make_shared(); - auto inner = std::make_shared(Zone::FrontLeft); - REQUIRE_FALSE(inner->close()); // a closed controller fails every send with ErrClosed - auto ctrl = observe::new_controller(inner, sink); - - Command cmd; cmd.zone = Zone::FrontLeft; cmd.type = CommandType::Get; - Response resp; - auto ec = ctrl->send(Context{}, cmd, resp); - REQUIRE(ec == ErrClosed); +TEST_CASE("observe: NoopSink does not crash on any metric", "[observe][REQ-OBS-005]") { + auto oc = new_observing_client(ok_request(), 1); // default noop sink - REQUIRE(sink->last_span().result == ErrClosed); // span records the error (REQ-OBS-004) - REQUIRE(sink->counter("rcp.commands.total") == 1.0); - REQUIRE(sink->counter("rcp.commands.errors") == 1.0); // error counter (REQ-OBS-008) + auto req = standard_request(1, 1); + acf::AcfMessageInfo resp; + std::vector resp_payload; + REQUIRE_FALSE(oc->request(Context{}, req, {}, resp, resp_payload)); } TEST_CASE("observe: InMemorySink is thread-safe under concurrent spans", "[observe][REQ-OBS-006]") { - auto sink = std::make_shared(); - auto inner = std::make_shared(Zone::Central); - auto ctrl = observe::new_controller(inner, sink); + auto sink = std::make_shared(); + auto oc = new_observing_client(ok_request(), 1, sink); constexpr int kThreads = 8; constexpr int kPerThread = 500; @@ -154,9 +143,10 @@ TEST_CASE("observe: InMemorySink is thread-safe under concurrent spans", for (int t = 0; t < kThreads; ++t) { ts.emplace_back([&] { for (int i = 0; i < kPerThread; ++i) { - Command cmd; cmd.zone = Zone::Central; cmd.type = CommandType::Get; - Response resp; - auto ec = ctrl->send(Context{}, cmd, resp); + auto req = standard_request(1, 1); + acf::AcfMessageInfo resp; + std::vector resp_payload; + auto ec = oc->request(Context{}, req, {}, resp, resp_payload); (void)ec; } }); @@ -164,3 +154,17 @@ TEST_CASE("observe: InMemorySink is thread-safe under concurrent spans", for (auto& th : ts) th.join(); REQUIRE(sink->span_count() == kThreads * kPerThread); } + +TEST_CASE("observe: rcp.requests.total counter increments per request", "[observe][REQ-OBS-007]") { + auto sink = std::make_shared(); + auto oc = new_observing_client(ok_request(), 1, sink); + + for (int i = 0; i < 3; ++i) { + auto req = standard_request(1, static_cast(i)); + acf::AcfMessageInfo resp; + std::vector resp_payload; + REQUIRE_FALSE(oc->request(Context{}, req, {}, resp, resp_payload)); + } + REQUIRE(sink->counter("rcp.requests.total") == 3.0); + REQUIRE(sink->counter("rcp.requests.errors") == 0.0); +} diff --git a/tests/test_record.cpp b/tests/test_record.cpp index 93899bc..1a67afe 100644 --- a/tests/test_record.cpp +++ b/tests/test_record.cpp @@ -6,108 +6,145 @@ // fusa:test REQ-REC-006 // fusa:test REQ-REC-007 // fusa:test REQ-REC-008 + +// Tests for rcp/record.hpp — binary record/replay of RC-Client-level +// request/response traffic (ROADMAP.md milestone 58, "Auxiliary Transport +// & Cross-Cutting Rebind", v2.14.0). + #include -#include "rcp/legacy_mock.hpp" #include "rcp/record.hpp" #include +#include #include #include using namespace rcp; +using namespace rcp::record; + +namespace { +acf::AcfMessageInfo standard_request(avtp::ByteBusId bus_id, uint8_t transaction_num) { + return acf::make_standard_request(bus_id, transaction_num, /*write=*/false, /*read_size=*/2); +} -TEST_CASE("record: recording controller captures entries", "[record]") { - auto rec = std::make_shared(); - auto inner = std::make_shared(Zone::FrontLeft); - auto ctrl = record::new_controller(inner, rec); +// echo_request is a stand-in "send-equivalent call" (RequestFn) that always +// answers a ReadResponse echoing a fixed payload. +RequestFn echo_request(std::vector payload = {0xAB}, std::error_code result = {}) { + return [payload, result](const Context&, const acf::AcfMessageInfo& req, + const std::vector&, + acf::AcfMessageInfo& out_resp, std::vector& out_payload) { + out_resp = acf::make_response(req, acf::ResponseKind::ReadResponse); + out_payload = payload; + return result; + }; +} +} // namespace + +TEST_CASE("record: RecordingClient captures request and response for every call", + "[record][REQ-REC-001]") { + auto rec = std::make_shared(); + auto rc = new_recording_client(echo_request({0xAB}), rec); - Command cmd; - cmd.zone = Zone::FrontLeft; - cmd.type = CommandType::Get; - Response resp; - REQUIRE_FALSE(ctrl->send(Context{}, cmd, resp)); + auto req = standard_request(1, 5); + acf::AcfMessageInfo resp; + std::vector resp_payload; + REQUIRE_FALSE(rc->request(Context{}, req, {}, resp, resp_payload)); REQUIRE(rec->size() == 1); - REQUIRE(rec->entries()[0].cmd.type == CommandType::Get); - REQUIRE(!rec->entries()[0].error); + auto entries = rec->entries(); + REQUIRE(entries[0].request.byte_bus_id == 1); + REQUIRE(entries[0].request.transaction_num == 5); + REQUIRE(entries[0].response_payload == std::vector{0xAB}); + REQUIRE_FALSE(entries[0].error); } -TEST_CASE("record: multiple sends produce sequential entries", "[record]") { - auto rec = std::make_shared(); - auto inner = std::make_shared(Zone::FrontLeft); - auto ctrl = record::new_controller(inner, rec); +TEST_CASE("record: Record::size reflects the number of completed request() calls", + "[record][REQ-REC-002]") { + auto rec = std::make_shared(); + auto rc = new_recording_client(echo_request(), rec); for (int i = 0; i < 3; ++i) { - Command cmd; - cmd.zone = Zone::FrontLeft; - cmd.type = CommandType::Set; - Response resp; - auto ec = ctrl->send(Context{}, cmd, resp); + auto req = standard_request(1, static_cast(i)); + acf::AcfMessageInfo resp; + std::vector resp_payload; + auto ec = rc->request(Context{}, req, {}, resp, resp_payload); (void)ec; } REQUIRE(rec->size() == 3); } -TEST_CASE("record: write_binary creates file", "[record]") { - auto rec = std::make_shared(); - auto inner = std::make_shared(Zone::FrontLeft); - auto ctrl = record::new_controller(inner, rec); +TEST_CASE("record: write_binary/read_binary round-trips entries losslessly", + "[record][REQ-REC-003]") { + auto rec = std::make_shared(); + auto rc = new_recording_client(echo_request({0x11, 0x22, 0x33}), rec); - Command cmd; - cmd.zone = Zone::FrontLeft; - cmd.type = CommandType::Get; - Response resp; - auto ec = ctrl->send(Context{}, cmd, resp); + auto req = standard_request(3, 9); + acf::AcfMessageInfo resp; + std::vector resp_payload; + auto ec = rc->request(Context{}, req, std::vector{0x01, 0x02}, resp, resp_payload); (void)ec; auto path = (std::filesystem::temp_directory_path() / "rcp_test_record.bin").string(); REQUIRE_FALSE(rec->write_binary(path)); - // Verify file is non-empty std::ifstream f(path, std::ios::binary | std::ios::ate); REQUIRE(f.is_open()); REQUIRE(f.tellg() > 0); + f.close(); + + Record roundtrip; + REQUIRE_FALSE(roundtrip.read_binary(path)); + REQUIRE(roundtrip.size() == rec->size()); + + auto original = rec->entries(); + auto reread = roundtrip.entries(); + REQUIRE(reread[0].timestamp_ns == original[0].timestamp_ns); + REQUIRE(reread[0].request.byte_bus_id == 3); + REQUIRE(reread[0].request.transaction_num == 9); + REQUIRE(reread[0].request_payload == std::vector{0x01, 0x02}); + REQUIRE(reread[0].response_payload == std::vector{0x11, 0x22, 0x33}); + + std::filesystem::remove(path); } TEST_CASE("record: entry timestamps are monotonically non-decreasing", "[record][REQ-REC-006]") { - auto rec = std::make_shared(); - auto inner = std::make_shared(Zone::FrontLeft); - auto ctrl = record::new_controller(inner, rec); + auto rec = std::make_shared(); + auto rc = new_recording_client(echo_request(), rec); for (int i = 0; i < 20; ++i) { - Command cmd; cmd.zone = Zone::FrontLeft; cmd.type = CommandType::Set; - Response resp; - auto ec = ctrl->send(Context{}, cmd, resp); + auto req = standard_request(1, static_cast(i)); + acf::AcfMessageInfo resp; + std::vector resp_payload; + auto ec = rc->request(Context{}, req, {}, resp, resp_payload); (void)ec; } - const auto& es = rec->entries(); - REQUIRE(es.size() == 20); - for (size_t i = 1; i < es.size(); ++i) { - REQUIRE(es[i].timestamp_ns >= es[i - 1].timestamp_ns); + auto entries = rec->entries(); + REQUIRE(entries.size() == 20); + for (size_t i = 1; i < entries.size(); ++i) { + REQUIRE(entries[i].timestamp_ns >= entries[i - 1].timestamp_ns); } } -TEST_CASE("record: forwards inner send result unchanged", "[record][REQ-REC-007]") { - auto rec = std::make_shared(); - auto inner = std::make_shared(Zone::FrontLeft); - REQUIRE_FALSE(inner->close()); // closed inner returns ErrClosed - auto ctrl = record::new_controller(inner, rec); +TEST_CASE("record: RecordingClient::request forwards the inner error_code unchanged", + "[record][REQ-REC-007]") { + auto rec = std::make_shared(); + auto rc = new_recording_client(echo_request({}, ErrClosed), rec); - Command cmd; cmd.zone = Zone::FrontLeft; cmd.type = CommandType::Get; - Response resp; - auto ec = ctrl->send(Context{}, cmd, resp); + auto req = standard_request(1, 1); + acf::AcfMessageInfo resp; + std::vector resp_payload; + auto ec = rc->request(Context{}, req, {}, resp, resp_payload); REQUIRE(ec == ErrClosed); // result passed through verbatim REQUIRE(rec->size() == 1); REQUIRE(rec->entries()[0].error == ErrClosed); // and captured in the log } -TEST_CASE("record: Record tolerates concurrent appends", "[record][REQ-REC-008]") { - auto rec = std::make_shared(); - auto inner = std::make_shared(Zone::Central); - auto ctrl = record::new_controller(inner, rec); +TEST_CASE("record: Record::append tolerates concurrent callers", "[record][REQ-REC-008]") { + auto rec = std::make_shared(); + auto rc = new_recording_client(echo_request(), rec); constexpr int kThreads = 8; constexpr int kPerThread = 500; @@ -115,9 +152,10 @@ TEST_CASE("record: Record tolerates concurrent appends", "[record][REQ-REC-008]" for (int t = 0; t < kThreads; ++t) { ts.emplace_back([&] { for (int i = 0; i < kPerThread; ++i) { - Command cmd; cmd.zone = Zone::Central; cmd.type = CommandType::Get; - Response resp; - auto ec = ctrl->send(Context{}, cmd, resp); + auto req = standard_request(1, 1); + acf::AcfMessageInfo resp; + std::vector resp_payload; + auto ec = rc->request(Context{}, req, {}, resp, resp_payload); (void)ec; } }); @@ -126,21 +164,30 @@ TEST_CASE("record: Record tolerates concurrent appends", "[record][REQ-REC-008]" REQUIRE(rec->size() == kThreads * kPerThread); } -TEST_CASE("record: playback replays entries against target", "[record][REQ-REC-004][REQ-REC-005]") { - auto rec = std::make_shared(); - auto inner = std::make_shared(Zone::FrontLeft); - auto ctrl = record::new_controller(inner, rec); +TEST_CASE("record: Playback::run_all replays every entry against the target", + "[record][REQ-REC-004][REQ-REC-005]") { + auto rec = std::make_shared(); + auto rc = new_recording_client(echo_request(), rec); - Command cmd; - cmd.zone = Zone::FrontLeft; - cmd.type = CommandType::Get; - Response resp; - auto ec = ctrl->send(Context{}, cmd, resp); - (void)ec; + for (int i = 0; i < 3; ++i) { + auto req = standard_request(1, static_cast(i)); + acf::AcfMessageInfo resp; + std::vector resp_payload; + auto ec = rc->request(Context{}, req, {}, resp, resp_payload); + (void)ec; + } + REQUIRE(rec->size() == 3); - REQUIRE(rec->size() == 1); + int replayed = 0; + RequestFn target = [&](const Context&, const acf::AcfMessageInfo& req, + const std::vector&, acf::AcfMessageInfo& out_resp, + std::vector&) { + ++replayed; + out_resp = acf::make_response(req, acf::ResponseKind::Acknowledge); + return std::error_code{}; + }; - auto target = std::make_shared(Zone::FrontLeft); - record::Playback pb(target, *rec, {0.0}); // no delays + Playback pb(target, *rec, PlaybackConfig{/*speed_factor=*/0.0}); // no delays REQUIRE_FALSE(pb.run_all(Context{})); + REQUIRE(replayed == 3); } diff --git a/tests/test_shmem.cpp b/tests/test_shmem.cpp index 9602130..0825f17 100644 --- a/tests/test_shmem.cpp +++ b/tests/test_shmem.cpp @@ -7,130 +7,164 @@ // fusa:test REQ-SHMEM-007 // fusa:test REQ-SHMEM-008 +// Tests for rcp/shmem.hpp — the zero-copy in-process request channel +// (ROADMAP.md milestone 58, "Auxiliary Transport & Cross-Cutting Rebind", +// v2.14.0). + #include #include -using namespace rcp; - -// ── ZoneServer + Controller ─────────────────────────────────────────────────── +#include +#include +#include -TEST_CASE("shmem Controller::send returns OK via default handler", "[shmem][REQ-SHMEM-001]") { - auto server = std::make_shared(Zone::FrontLeft); - shmem::Controller ctrl(server); +using namespace rcp; +using namespace rcp::shmem; - Command cmd; - cmd.zone = Zone::FrontLeft; - Response resp; - REQUIRE_FALSE(ctrl.send(Context::background(), cmd, resp)); - REQUIRE(resp.status == ResponseStatus::OK); +namespace { +acf::AcfMessageInfo standard_request(avtp::ByteBusId bus_id, uint8_t transaction_num) { + return acf::make_standard_request(bus_id, transaction_num, /*write=*/false, /*read_size=*/4); } - -TEST_CASE("shmem Controller dispatches to custom handler", "[shmem][REQ-SHMEM-002]") { - auto server = std::make_shared(Zone::FrontLeft); - server->set_handler([](const Command& c) { - Response r; - r.command_id = c.id; - r.zone = c.zone; - r.status = ResponseStatus::OK; - r.payload = {0xAB}; - return r; +} // namespace + +// ── Zero-copy delivery ──────────────────────────────────────────────────────── + +TEST_CASE("shmem Channel::request delivers to the handler directly, without a wire encode/decode", + "[shmem][REQ-SHMEM-001]") { + auto ch = new_channel(/*stream_key=*/42); + ch->set_handler([](size_t, const acf::AcfMessageInfo& req, + const std::vector& payload, + acf::AcfMessageInfo& out_resp, std::vector& out_payload) { + out_resp = acf::make_response(req, acf::ResponseKind::ReadResponse); + out_payload = payload; // echo unchanged -- proves no lossy round trip occurred + return std::error_code{}; }); - shmem::Controller ctrl(server); - Command cmd; - cmd.id = 7; - cmd.zone = Zone::FrontLeft; - Response resp; - REQUIRE_FALSE(ctrl.send(Context::background(), cmd, resp)); - REQUIRE(resp.payload == std::vector{0xAB}); + auto req = standard_request(1, 5); + std::vector req_payload{0xDE, 0xAD, 0xBE, 0xEF}; + acf::AcfMessageInfo resp; + std::vector resp_payload; + + REQUIRE_FALSE(ch->request(0, req, req_payload, resp, resp_payload)); + REQUIRE(resp_payload == req_payload); + REQUIRE(resp.byte_bus_id == req.byte_bus_id); + REQUIRE(resp.transaction_num == req.transaction_num); } -TEST_CASE("shmem Controller::send returns ErrZoneMismatch for wrong zone", "[shmem][REQ-SHMEM-003]") { - auto server = std::make_shared(Zone::FrontLeft); - shmem::Controller ctrl(server); +TEST_CASE("shmem Channel::request answers Acknowledge by default when no handler is set", + "[shmem][REQ-SHMEM-002]") { + auto ch = new_channel(1); + auto req = standard_request(1, 1); + acf::AcfMessageInfo resp; + std::vector resp_payload; - Command cmd; - cmd.zone = Zone::RearLeft; - Response resp; - REQUIRE(ctrl.send(Context::background(), cmd, resp) == ErrZoneMismatch); + REQUIRE_FALSE(ch->request(0, req, {}, resp, resp_payload)); + REQUIRE(acf::response_kind_of(resp) == acf::ResponseKind::Acknowledge); } -TEST_CASE("shmem Controller::send returns ErrClosed after close", "[shmem][REQ-SHMEM-004]") { - auto server = std::make_shared(Zone::FrontLeft); - shmem::Controller ctrl(server); - { auto ec = ctrl.close(); (void)ec; } +// ── Lifecycle ───────────────────────────────────────────────────────────────── + +TEST_CASE("shmem Channel::request returns ErrClosed after close()", "[shmem][REQ-SHMEM-003]") { + auto ch = new_channel(1); + ch->close(); - Command cmd; - cmd.zone = Zone::FrontLeft; - Response resp; - REQUIRE(ctrl.send(Context::background(), cmd, resp) == ErrClosed); + auto req = standard_request(1, 1); + acf::AcfMessageInfo resp; + std::vector resp_payload; + REQUIRE(ch->request(0, req, {}, resp, resp_payload) == ErrClosed); } -// ── Publish / subscribe ─────────────────────────────────────────────────────── +// ── Registry ────────────────────────────────────────────────────────────────── -TEST_CASE("shmem ZoneServer::publish delivers status to subscribers", "[shmem][REQ-SHMEM-005]") { - auto server = std::make_shared(Zone::FrontLeft); - server->set_healthy(true); - shmem::Controller ctrl(server); +TEST_CASE("shmem Registry::lookup finds a channel registered via add_channel", + "[shmem][REQ-SHMEM-004]") { + auto reg = new_registry(); + auto ch = new_channel(7); + REQUIRE_FALSE(reg->add_channel(ch)); - std::shared_ptr ch; - REQUIRE_FALSE(ctrl.subscribe(Context::background(), ch)); - REQUIRE(ch != nullptr); + std::shared_ptr out; + REQUIRE_FALSE(reg->lookup(7, out)); + REQUIRE(out == ch); +} - server->publish({0x01, 0x02}); - std::this_thread::sleep_for(std::chrono::milliseconds(10)); +TEST_CASE("shmem Registry::close closes all registered channels", "[shmem][REQ-SHMEM-005]") { + auto reg = new_registry(); + auto ch = new_channel(7); + REQUIRE_FALSE(reg->add_channel(ch)); + REQUIRE_FALSE(reg->close()); - auto st = ch->try_recv(); - REQUIRE(st.has_value()); - REQUIRE(st->zone == Zone::FrontLeft); - REQUIRE(st->healthy == true); - REQUIRE(st->payload == std::vector{0x01, 0x02}); + std::shared_ptr out; + REQUIRE(reg->lookup(7, out) == ErrClosed); + REQUIRE_FALSE(ch->ok()); // the channel itself was closed too, not just delisted - ch->close(); + auto req = standard_request(1, 1); + acf::AcfMessageInfo resp; + std::vector resp_payload; + REQUIRE(ch->request(0, req, {}, resp, resp_payload) == ErrClosed); } -// ── Payload copy isolation ──────────────────────────────────────────────────── - -TEST_CASE("shmem controller copies payload (no aliasing)", "[shmem][REQ-SHMEM-006]") { - auto server = std::make_shared(Zone::FrontLeft); - std::vector received_payload; - server->set_handler([&](const Command& c) { - received_payload = c.payload; - return Response{c.id, c.zone, ResponseStatus::OK, {}}; +// ── Concurrency ─────────────────────────────────────────────────────────────── + +TEST_CASE("shmem Channel::request is thread-safe for concurrent callers", + "[shmem][REQ-SHMEM-006]") { + auto ch = new_channel(1); + std::atomic call_count{0}; + ch->set_handler([&](size_t, const acf::AcfMessageInfo& req, const std::vector&, + acf::AcfMessageInfo& out_resp, std::vector&) { + call_count.fetch_add(1); + out_resp = acf::make_response(req, acf::ResponseKind::Acknowledge); + return std::error_code{}; }); - shmem::Controller ctrl(server); - std::vector original{1, 2, 3}; - Command cmd; - cmd.zone = Zone::FrontLeft; - cmd.payload = original; + constexpr int kThreads = 8; + constexpr int kPerThread = 50; + std::vector threads; + for (int t = 0; t < kThreads; ++t) { + threads.emplace_back([&, t] { + for (int i = 0; i < kPerThread; ++i) { + auto req = standard_request(1, static_cast(t)); + acf::AcfMessageInfo resp; + std::vector resp_payload; + auto ec = ch->request(static_cast(t), req, {}, resp, resp_payload); + (void)ec; + } + }); + } + for (auto& th : threads) th.join(); + + REQUIRE(call_count.load() == kThreads * kPerThread); +} - Response resp; - { auto ec = ctrl.send(Context::background(), cmd, resp); (void)ec; } +// ── Per-call client id pass-through ────────────────────────────────────────── + +TEST_CASE("shmem Channel::request forwards the caller-supplied client id to the handler unchanged", + "[shmem][REQ-SHMEM-007]") { + auto ch = new_channel(1); + size_t seen = 999; + ch->set_handler([&](size_t client, const acf::AcfMessageInfo& req, + const std::vector&, acf::AcfMessageInfo& out_resp, + std::vector&) { + seen = client; + out_resp = acf::make_response(req, acf::ResponseKind::Acknowledge); + return std::error_code{}; + }); - // Mutate original after send — should not affect what the handler saw. - original[0] = 0xFF; - REQUIRE(received_payload[0] == 1); + auto req = standard_request(1, 1); + acf::AcfMessageInfo resp; + std::vector resp_payload; + REQUIRE_FALSE(ch->request(/*client=*/42, req, {}, resp, resp_payload)); + REQUIRE(seen == 42); } -// ── Registry ────────────────────────────────────────────────────────────────── - -TEST_CASE("shmem Registry::lookup finds registered server", "[shmem][REQ-SHMEM-007]") { - auto reg = shmem::new_registry(); - auto server = std::make_shared(Zone::FrontLeft); - REQUIRE_FALSE(reg->add_server(server)); +// ── Idempotent close ────────────────────────────────────────────────────────── - std::shared_ptr ctrl_out; - REQUIRE_FALSE(reg->lookup(Zone::FrontLeft, ctrl_out)); - REQUIRE(ctrl_out != nullptr); -} +TEST_CASE("shmem Channel::close and Registry::close are idempotent", "[shmem][REQ-SHMEM-008]") { + auto ch = new_channel(1); + ch->close(); + ch->close(); // second call must not crash -TEST_CASE("shmem Registry::close releases all controllers", "[shmem][REQ-SHMEM-008]") { - auto reg = shmem::new_registry(); - auto server = std::make_shared(Zone::FrontLeft); - { auto ec = reg->add_server(server); (void)ec; } + auto reg = new_registry(); + REQUIRE_FALSE(reg->add_channel(new_channel(2))); REQUIRE_FALSE(reg->close()); - - std::shared_ptr ctrl_out; - REQUIRE(reg->lookup(Zone::FrontLeft, ctrl_out) == ErrClosed); + REQUIRE_FALSE(reg->close()); // second call must not crash } diff --git a/tests/test_tls.cpp b/tests/test_tls.cpp index 0b7e4a9..b6dec5d 100644 --- a/tests/test_tls.cpp +++ b/tests/test_tls.cpp @@ -9,16 +9,19 @@ // fusa:test REQ-TLS-009 // fusa:test REQ-TLS-010 -// Mutual-TLS transport tests (SG-006, IEC 62443 SL-2). +// Secure-channel option tests for rcp/tls.hpp — the UDP/IP transport +// variant's DTLS/application-layer channel (ROADMAP.md milestone 58, +// "Auxiliary Transport & Cross-Cutting Rebind", v2.14.0, SG-006, IEC 62443 +// SL-2). // -// Without RCP_TLS_OPENSSL the TLS module is a compile-time interface stub: it -// carries the secure-transport configuration (certs, CA, verify_peer) and never -// performs an insecure send — every transport call returns -// function_not_supported rather than transmitting plaintext. These tests pin -// that configuration surface and the secure-by-default refusal. The cipher / -// protocol-version enforcement of REQ-TLS-007/008 is exercised end-to-end by -// the OpenSSL backend build; here we assert the stub never falls back to an -// insecure path on hosts that lack OpenSSL. +// Without RCP_TLS_OPENSSL the module is a compile-time interface stub: it +// carries the secure-transport configuration (certs, CA, verify_peer) and +// never performs an insecure send — every transport call returns +// function_not_supported rather than transmitting plaintext. These tests +// pin that configuration surface and the secure-by-default refusal. The +// cipher/protocol-version enforcement of REQ-TLS-007/008 is exercised +// end-to-end by the OpenSSL backend build; here we assert the stub never +// falls back to an insecure path on hosts that lack OpenSSL. #include #include "rcp/tls.hpp" @@ -31,11 +34,18 @@ namespace { const std::error_code kUnsupported = std::make_error_code(std::errc::function_not_supported); +avtp::StreamId make_stream_id(uint8_t mac_seed, uint16_t suffix) { + avtp::StreamId id; + for (auto& b : id.mac) b = mac_seed++; + id.suffix = suffix; + return id; +} + tls::Config mtls_config() { tls::Config c; - c.cert_file = "/etc/rcp/client.pem"; - c.key_file = "/etc/rcp/client.key"; - c.ca_file = "/etc/rcp/ca.pem"; + c.cert_file = "/etc/rcp/client.pem"; + c.key_file = "/etc/rcp/client.key"; + c.ca_file = "/etc/rcp/ca.pem"; c.verify_peer = true; return c; } @@ -61,45 +71,56 @@ TEST_CASE("tls: PEM cert/key/ca files are carried in Config", "[tls][REQ-TLS-004 REQUIRE(c.ca_file == "/etc/rcp/ca.pem"); } -TEST_CASE("tls: command frames never sent in the clear (stub refuses send)", +TEST_CASE("tls: SecureClient::request never sends a request frame in the clear (stub refuses)", "[tls][REQ-TLS-005]") { - tls::Controller ctrl(Zone::FrontLeft, "127.0.0.1", 8443, mtls_config()); - Command cmd; cmd.zone = Zone::FrontLeft; cmd.type = CommandType::Set; - Response resp; - // No OpenSSL backend → refuse rather than transmit an unencrypted frame. - REQUIRE(ctrl.send(Context{}, cmd, resp) == kUnsupported); + tls::SecureClient client(make_stream_id(0x02, 1), "127.0.0.1", 8443, mtls_config()); + auto req = acf::make_standard_request(/*bus_id=*/1, /*transaction_num=*/1, + /*write=*/false, /*read_size=*/4); + acf::AcfMessageInfo resp; + std::vector resp_payload; + REQUIRE(client.request(Context{}, req, {}, resp, resp_payload) == kUnsupported); } -TEST_CASE("tls: status frames never sent in the clear (stub refuses subscribe)", +TEST_CASE("tls: SecureServer never answers in the clear (stub reports not ok())", "[tls][REQ-TLS-006]") { - tls::Controller ctrl(Zone::FrontLeft, "127.0.0.1", 8443, mtls_config()); - std::shared_ptr ch; - REQUIRE(ctrl.subscribe(Context{}, ch) == kUnsupported); + tls::SecureServer server(make_stream_id(0x03, 1), "127.0.0.1", 8443, mtls_config()); + // No OpenSSL backend -> the server can never actually bind/serve + // securely, so it must not report itself usable. + REQUIRE_FALSE(server.ok()); } TEST_CASE("tls: no insecure fallback on hosts without OpenSSL (TLS1.2+ posture)", "[tls][REQ-TLS-007][REQ-TLS-008]") { - // The secure backend (RCP_TLS_OPENSSL) enforces TLS >=1.2 and disables weak - // ciphers. The stub's safety contract is that it must NOT silently downgrade - // to plaintext: dialing/sending returns a hard error instead. - auto reg = tls::new_registry(); - REQUIRE(reg->dial(Zone::Central, "127.0.0.1", 8443, mtls_config()) == kUnsupported); - tls::Controller ctrl(Zone::Central, "127.0.0.1", 8443, mtls_config()); - Command cmd; cmd.zone = Zone::Central; Response resp; - REQUIRE(ctrl.send(Context{}, cmd, resp) == kUnsupported); + // The secure backend (RCP_TLS_OPENSSL) enforces TLS >=1.2 and disables + // weak ciphers. The stub's safety contract is that it must NOT silently + // downgrade to plaintext: every transport call returns a hard error + // instead. + tls::SecureClient client(make_stream_id(0x02, 2), "127.0.0.1", 8443, mtls_config()); + REQUIRE_FALSE(client.ok()); + + auto req = acf::make_standard_request(1, 1, false, 4); + acf::AcfMessageInfo resp; + std::vector resp_payload; + REQUIRE(client.request(Context{}, req, {}, resp, resp_payload) == kUnsupported); } TEST_CASE("tls: close terminates the session cleanly", "[tls][REQ-TLS-009]") { - tls::Controller ctrl(Zone::RearLeft, "127.0.0.1", 8443, mtls_config()); - REQUIRE_FALSE(ctrl.close()); - auto reg = tls::new_registry(); - REQUIRE_FALSE(reg->close()); + tls::SecureClient client(make_stream_id(0x02, 3), "127.0.0.1", 8443, mtls_config()); + REQUIRE_FALSE(client.close()); + + tls::SecureServer server(make_stream_id(0x03, 3), "127.0.0.1", 8443, mtls_config()); + server.close(); // must not hang or crash } -TEST_CASE("tls: transport errors propagate to the caller", "[tls][REQ-TLS-010]") { - auto reg = tls::new_registry(); - std::shared_ptr out; - REQUIRE(reg->lookup(Zone::FrontLeft, out) == kUnsupported); - REQUIRE(reg->register_ctrl(nullptr) == kUnsupported); - REQUIRE(reg->deregister(Zone::FrontLeft) == kUnsupported); +TEST_CASE("tls: SecureClient/SecureServer carry the configured stream_id and Config unchanged", + "[tls][REQ-TLS-010]") { + auto sid = make_stream_id(0x02, 4); + auto cfg = mtls_config(); + tls::SecureClient client(sid, "127.0.0.1", 8443, cfg); + REQUIRE(client.stream_id() == sid); + REQUIRE(client.config().cert_file == cfg.cert_file); + + tls::SecureServer server(sid, "127.0.0.1", 8443, cfg); + REQUIRE(server.stream_id() == sid); + REQUIRE(server.config().ca_file == cfg.ca_file); } diff --git a/tests/test_tsn.cpp b/tests/test_tsn.cpp index 52df862..dabb022 100644 --- a/tests/test_tsn.cpp +++ b/tests/test_tsn.cpp @@ -5,76 +5,85 @@ // fusa:test REQ-TSN-005 // fusa:test REQ-TSN-006 -// TSN / IEEE 802.1p priority-mapping tests. +// TSN / IEEE 802.1p priority-mapping tests (ROADMAP.md milestone 58, +// "Auxiliary Transport & Cross-Cutting Rebind", v2.14.0). // -// The TSN wrapper maps rcp::Priority to an 802.1p PCP value and applies -// SO_PRIORITY to the egress socket before delegating to the inner controller. -// The setsockopt() call is platform-gated (Linux); these tests pin the -// priority→PCP mapping and the pass-through behaviour, which are portable. +// apply_priority() maps rcp::request::RequestCategory to an 802.1p PCP +// value and applies SO_PRIORITY to the given socket fd. The setsockopt() +// call itself is platform-gated (Linux); these tests pin the +// category -> PCP mapping and the portable fd-validation behavior. #include -#include "rcp/legacy_mock.hpp" #include "rcp/tsn.hpp" using namespace rcp; +using rcp::request::RequestCategory; -TEST_CASE("tsn: PCPMap maps Normal/High/Critical to PCP values", "[tsn][REQ-TSN-002]") { +TEST_CASE("tsn: PCPMap maps every RequestCategory to a distinct PCP value", + "[tsn][REQ-TSN-002]") { auto m = tsn::default_pcp_map(); - REQUIRE(tsn::pcp_for(m, Priority::Normal) == m.normal); - REQUIRE(tsn::pcp_for(m, Priority::High) == m.high); - REQUIRE(tsn::pcp_for(m, Priority::Critical) == m.critical); + REQUIRE(tsn::pcp_for(m, RequestCategory::Cancellation) == m.cancellation); + REQUIRE(tsn::pcp_for(m, RequestCategory::Triggered) == m.triggered); + REQUIRE(tsn::pcp_for(m, RequestCategory::Timed) == m.timed); + REQUIRE(tsn::pcp_for(m, RequestCategory::Compound) == m.compound); + REQUIRE(tsn::pcp_for(m, RequestCategory::CompoundWait) == m.compound_wait); + REQUIRE(tsn::pcp_for(m, RequestCategory::Chained) == m.chained); + REQUIRE(tsn::pcp_for(m, RequestCategory::Standard) == m.standard); } -TEST_CASE("tsn: Critical maps to the highest PCP", "[tsn][REQ-TSN-003]") { +TEST_CASE("tsn: Cancellation (highest execution priority) maps to the highest PCP", + "[tsn][REQ-TSN-003]") { auto m = tsn::default_pcp_map(); - REQUIRE(tsn::pcp_for(m, Priority::Critical) == 7); - REQUIRE(tsn::pcp_for(m, Priority::Critical) > tsn::pcp_for(m, Priority::High)); + REQUIRE(tsn::pcp_for(m, RequestCategory::Cancellation) == 7); + REQUIRE(tsn::pcp_for(m, RequestCategory::Cancellation) > + tsn::pcp_for(m, RequestCategory::Triggered)); } -TEST_CASE("tsn: High maps to a mid-range PCP", "[tsn][REQ-TSN-004]") { +TEST_CASE("tsn: Triggered maps to a mid/high-range PCP", "[tsn][REQ-TSN-004]") { auto m = tsn::default_pcp_map(); - auto high = tsn::pcp_for(m, Priority::High); - REQUIRE(high == 5); - REQUIRE(high > tsn::pcp_for(m, Priority::Normal)); - REQUIRE(high < tsn::pcp_for(m, Priority::Critical)); + auto triggered = tsn::pcp_for(m, RequestCategory::Triggered); + REQUIRE(triggered == 6); + REQUIRE(triggered < tsn::pcp_for(m, RequestCategory::Cancellation)); + REQUIRE(triggered > tsn::pcp_for(m, RequestCategory::Standard)); } -TEST_CASE("tsn: Normal maps to a low PCP", "[tsn][REQ-TSN-005]") { +TEST_CASE("tsn: Standard (the mandatory baseline kind) maps to the lowest PCP", + "[tsn][REQ-TSN-005]") { auto m = tsn::default_pcp_map(); - REQUIRE(tsn::pcp_for(m, Priority::Normal) == 2); - REQUIRE(tsn::pcp_for(m, Priority::Normal) < tsn::pcp_for(m, Priority::High)); + REQUIRE(tsn::pcp_for(m, RequestCategory::Standard) == 1); + // Every priority-rank ordering the specification defines across the + // seven categories (extraction §3.14) is preserved in the PCP mapping: + // higher execution priority (lower rank) never yields a lower PCP. + REQUIRE(tsn::pcp_for(m, RequestCategory::Standard) < + tsn::pcp_for(m, RequestCategory::Chained)); + REQUIRE(tsn::pcp_for(m, RequestCategory::Chained) < + tsn::pcp_for(m, RequestCategory::CompoundWait)); + REQUIRE(tsn::pcp_for(m, RequestCategory::CompoundWait) < + tsn::pcp_for(m, RequestCategory::Compound)); + REQUIRE(tsn::pcp_for(m, RequestCategory::Compound) < + tsn::pcp_for(m, RequestCategory::Timed)); + REQUIRE(tsn::pcp_for(m, RequestCategory::Timed) < + tsn::pcp_for(m, RequestCategory::Triggered)); } -TEST_CASE("tsn: send applies priority class then delegates to inner", - "[tsn][REQ-TSN-001]") { - // Capture the command type/priority the inner controller actually receives, - // proving the wrapper forwards after applying the (platform-gated) PCP. - Priority seen = Priority::Normal; - auto inner = std::make_shared( - Zone::FrontLeft, [&](const Command& c) { - seen = c.priority; - return Response{c.id, Zone::FrontLeft, ResponseStatus::OK, {}}; - }); - // fd = -1 → SO_PRIORITY is skipped, send still delegates. - auto ctrl = tsn::new_controller(inner, -1, tsn::default_tsn_config()); - - Command cmd; - cmd.zone = Zone::FrontLeft; - cmd.type = CommandType::Set; - cmd.priority = Priority::Critical; - Response resp; - REQUIRE_FALSE(ctrl->send(Context{}, cmd, resp)); - REQUIRE(resp.status == ResponseStatus::OK); - REQUIRE(seen == Priority::Critical); // forwarded unchanged to the inner controller +TEST_CASE("tsn: apply_priority succeeds (as a portable no-op where SO_PRIORITY " + "is unavailable) for a valid fd", "[tsn][REQ-TSN-001]") { + // fd = 0 is a valid descriptor value (stdin) so this exercises the real + // code path on every platform without needing an actual UDP socket; + // on non-Linux builds RCP_TSN_SO_PRIORITY is undefined and this is + // unconditionally a no-op success. + auto ec = tsn::apply_priority(0, tsn::default_tsn_config(), RequestCategory::Cancellation); +#if defined(RCP_TSN_SO_PRIORITY) + // setsockopt on fd 0 (not a socket) fails on Linux -- apply_priority + // must propagate that failure rather than silently swallowing it. + REQUIRE(ec); +#else + REQUIRE_FALSE(ec); +#endif } -TEST_CASE("tsn: subscribe delegates to inner", "[tsn][REQ-TSN-006]") { - auto inner = std::make_shared(Zone::Central); - auto ctrl = tsn::new_controller(inner, -1, tsn::default_tsn_config()); - - std::shared_ptr ch; - REQUIRE_FALSE(ctrl->subscribe(Context{}, ch)); - REQUIRE(ch != nullptr); - REQUIRE(ctrl->zone() == Zone::Central); - REQUIRE_FALSE(ctrl->close()); +TEST_CASE("tsn: apply_priority rejects a negative fd on every platform", + "[tsn][REQ-TSN-006]") { + auto ec = tsn::apply_priority(-1, tsn::default_tsn_config(), RequestCategory::Standard); + REQUIRE(ec == std::make_error_code(std::errc::invalid_argument)); }