From 13feac55b6fd801aaaa74478e582537dcb73c7a9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=B6rg=20Ebeling?= Date: Mon, 2 Nov 2026 20:36:01 +0100 Subject: [PATCH 1/7] feat(protocol): serve schema blocks the advertisement cannot carry The advertisement is one UDP packet (max_packet_size 1472), so a growing schema no longer fits: SoundService needs 1562 bytes of CBOR for its 32 SoundId values. The generator now fills the advertisement as far as it goes and lists the rest: "desc": { "type": ..., "version": ..., "inputs": [...], "functions": [...], "registers": [...], "omitted": ["enums"] } Two message types serve the remainder on request: DESCRIPTION_REQUEST 0x82 arg1 = offset, arg2 = max payload, payload = DescriptionRequestPayload (sid, ip, port) DESCRIPTION_PART 0x83 arg1 = offset, arg2 = total, payload = window Advertised and omitted bytes partition the schema, so nothing is stored twice and nothing is lost. Hosts with generated classes never ask; the Python interface fetches the blocks after discovery, merges them and caches per (sid, version) - a repeating advertisement never fetches again. A failed fetch keeps the advertised part and only logs a warning. SoundService: 894 B advertised + 684 B omitted (was 1605 B in one packet), i.e. 474 B headroom; every other service still advertises its complete schema. --- Readme.md | 83 ++++----- codegen/README.md | 14 ++ codegen/templates/ServiceTemplate.hpp | 13 ++ codegen/xbot_codegen/xbot_codegen.py | 46 ++++- .../datatypes/DescriptionRequestPayload.hpp | 26 +++ include/xbot/datatypes/XbotHeader.hpp | 13 +- .../include/xbot-service/Service.hpp | 19 +- libxbot-service/src/Service.cpp | 56 ++++++ xbot_service_interface_py/README.md | 12 +- .../tests/test_description_fetch.py | 138 +++++++++++++++ .../xbot_service_interface/datatypes.py | 19 ++ .../xbot_service_interface/discovery.py | 165 +++++++++++++++++- 12 files changed, 555 insertions(+), 49 deletions(-) create mode 100644 include/xbot/datatypes/DescriptionRequestPayload.hpp create mode 100644 xbot_service_interface_py/tests/test_description_fetch.py diff --git a/Readme.md b/Readme.md index 2d8927e..17fd4c8 100644 --- a/Readme.md +++ b/Readme.md @@ -80,33 +80,33 @@ message-specific payload. ### XbotHeader (24 bytes, packed) -| Offset | Size | Field | Notes | -|--------|------|------------------|---------------------------------------------------------| -| 0 | 1 | protocol_version | Currently `1` | -| 1 | 1 | message_type | See message type table below | +| Offset | Size | Field | Notes | +| ------ | ---- | ---------------- | --------------------------------------------------------- | +| 0 | 1 | protocol_version | Currently `1` | +| 1 | 1 | message_type | See message type table below | | 2 | 1 | flags | Bit 0: reboot flag (set on boot, cleared on seq rollover) | -| 3 | 1 | reserved1 | | -| 4 | 2 | service_id | Identifies the service | -| 6 | 1 | arg1 | Message-specific (see table below) | -| 7 | 1 | reserved2 | | -| 8 | 2 | arg2 | Message-specific (see table below) | -| 10 | 2 | sequence_no | Increments per message; rollover clears reboot flag | -| 12 | 8 | timestamp | Unix timestamp in **microseconds** | -| 20 | 4 | payload_size | Byte length of payload following the header | +| 3 | 1 | reserved1 | | +| 4 | 2 | service_id | Identifies the service | +| 6 | 1 | arg1 | Message-specific (see table below) | +| 7 | 1 | reserved2 | | +| 8 | 2 | arg2 | Message-specific (see table below) | +| 10 | 2 | sequence_no | Increments per message; rollover clears reboot flag | +| 12 | 8 | timestamp | Unix timestamp in **microseconds** | +| 20 | 4 | payload_size | Byte length of payload following the header | ### Message Types -| Value | Name | arg1 | arg2 | Payload | -|--------|-------------------------|-----------------------------|-------------|----------------------------------------| -| `0x00` | UNKNOWN | — | — | — | -| `0x01` | DATA | — | `target_id` | Raw value bytes | -| `0x02` | CONFIGURATION_REQUEST | — | — | Empty | -| `0x03` | CLAIM | `0`=request, `1`=ack | — | `ClaimPayload` (request) or empty (ack) | -| `0x04` | HEARTBEAT | — | — | Empty | -| `0x05` | TRANSACTION | `0`=data, `1`=configuration | — | Sequence of `DataDescriptor`+payload | -| `0x7F` | LOG | log level (1–7) | — | UTF-8 string (max 255 bytes) | -| `0x80` | SERVICE_ADVERTISEMENT | — | — | CBOR-encoded service JSON | -| `0x81` | SERVICE_QUERY | — | — | — | +| Value | Name | arg1 | arg2 | Payload | +| ------ | --------------------- | --------------------------- | ----------- | --------------------------------------- | +| `0x00` | UNKNOWN | — | — | — | +| `0x01` | DATA | — | `target_id` | Raw value bytes | +| `0x02` | CONFIGURATION_REQUEST | — | — | Empty | +| `0x03` | CLAIM | `0`=request, `1`=ack | — | `ClaimPayload` (request) or empty (ack) | +| `0x04` | HEARTBEAT | — | — | Empty | +| `0x05` | TRANSACTION | `0`=data, `1`=configuration | — | Sequence of `DataDescriptor`+payload | +| `0x7F` | LOG | log level (1–7) | — | UTF-8 string (max 255 bytes) | +| `0x80` | SERVICE_ADVERTISEMENT | — | — | CBOR-encoded service JSON | +| `0x81` | SERVICE_QUERY | — | — | — | ### TRANSACTION payload format @@ -115,7 +115,7 @@ A TRANSACTION payload is a sequence of chunks, each preceded by a `DataDescripto **DataDescriptor (8 bytes, packed):** | Offset | Size | Field | -|--------|------|--------------| +| ------ | ---- | ------------ | | 0 | 2 | target_id | | 2 | 2 | reserved | | 4 | 4 | payload_size | @@ -126,11 +126,11 @@ Chunks are read sequentially until `payload_size` bytes of the transaction are c Sent by the interface as the payload of a `CLAIM` message: -| Offset | Size | Field | Notes | -|--------|------|-------------------|--------------------------------| -| 0 | 4 | target_ip | Interface IP (network byte order) | -| 4 | 2 | target_port | Interface UDP port | -| 6 | 4 | heartbeat_micros | Requested heartbeat interval | +| Offset | Size | Field | Notes | +| ------ | ---- | ---------------- | --------------------------------- | +| 0 | 4 | target_ip | Interface IP (network byte order) | +| 4 | 2 | target_port | Interface UDP port | +| 6 | 4 | heartbeat_micros | Requested heartbeat interval | --- @@ -138,10 +138,10 @@ Sent by the interface as the payload of a `CLAIM` message: Service discovery uses **UDP multicast**. -| Parameter | Value | -|-------------------|-----------------| -| Multicast address | `233.255.255.0` | -| Port | `4242` | +| Parameter | Value | +| ----------------- | ------------------------------ | +| Multicast address | `233.255.255.0` | +| Port | `4242` | | Message type | `SERVICE_ADVERTISEMENT` (0x80) | **Advertisement payload:** CBOR-encoded JSON with the following structure: @@ -179,18 +179,18 @@ Service discovery uses **UDP multicast**. Remote logging uses **UDP multicast**. -| Parameter | Value | -|-------------------|-------------------| -| Multicast address | `233.255.255.1` | -| Port | `4242` | -| Message type | `LOG` (0x7F) | +| Parameter | Value | +| ----------------- | --------------- | +| Multicast address | `233.255.255.1` | +| Port | `4242` | +| Message type | `LOG` (0x7F) | **Payload:** UTF-8 string, max 255 bytes. When a `service_id` context is available the message is prefixed as `[ID=X] message`. **Log levels (arg1):** | arg1 | Level | -|------|----------| +| ---- | -------- | | 1 | TRACE | | 2 | DEBUG | | 3 | INFO | @@ -277,8 +277,9 @@ Two ways to use a service: - **Schema provided:** pass a `service.json` file (or dict) to `ServiceInterface`. The schema is validated against the service's advertisement on connect, and an `IncompatibleServiceError` aborts the connection on a mismatch. -- **Schema-free:** omit the schema entirely. The full service description is embedded in the advertisement itself, so - inputs, outputs, registers, and RPC functions are all available immediately after discovery. +- **Schema-free:** omit the schema entirely. The advertisement carries the interface (inputs, outputs, registers, RPC functions) and as much of the rest of the schema as fits into one packet. + A service whose schema is bigger, lists what did not fit under `omitted`; the client fetches those blocks with `DESCRIPTION_REQUEST`/`DESCRIPTION_PART` once after discovery (one round trip, unicast) and caches them per `(service id, version)`. + Hosts with generated classes never fetch: they know the types. ```python from xbot_service_interface import XbotServiceIo, ServiceInterface diff --git a/codegen/README.md b/codegen/README.md index 5de502b..5f566c1 100644 --- a/codegen/README.md +++ b/codegen/README.md @@ -50,3 +50,17 @@ from the `service.json` stays visible, so the schema remains the reference a rea (`// PlayMP3 calls the RPC function 3 ("PlayMp3" in the service.json).`), - trailing comments on the enum values (`SoundIDBootPing SoundID = 0 // BOOT_PING`), - nothing on the `Fn*` function-id constants: they are only used together with the RPC method whose doc comment names the schema function. + +## The advertised service description + +The advertisement is a single UDP packet (`xbot::config::max_packet_size`, 1472 bytes for one +unfragmented Ethernet frame), so the code generator advertises as much of the schema as fits and +lists the rest under `omitted`: + +```json +"desc": { "type": "SoundService", "version": 1, "inputs": [...], "functions": [...], + "registers": [...], "omitted": ["enums"] } +``` + +Blocks are added in schema order while the budget is not exceeded; whatever does not fit is +available through `DESCRIPTION_REQUEST` -> `DESCRIPTION_PART` (see `libxbot-service`), served from a second static array. The two arrays partition the schema, so nothing is stored twice and nothing is lost: a host that needs the block (a client without generated classes) fetches it, a host with generated code never asks. A service whose schema fits is advertised completely and ignores description requests. diff --git a/codegen/templates/ServiceTemplate.hpp b/codegen/templates/ServiceTemplate.hpp index 861b40e..bf910ba 100644 --- a/codegen/templates/ServiceTemplate.hpp +++ b/codegen/templates/ServiceTemplate.hpp @@ -76,6 +76,19 @@ class ServiceTemplateBase : public xbot::service::Service { /*[[[cog cog.out("static constexpr unsigned char SERVICE_DESCRIPTION_CBOR[] = "); cog.outl(xbot_codegen.binary2c_array(service["service_cbor"])); + cog.outl("") + cog.outl(" // Blocks that did not fit into the advertisement; served on request") + cog.outl(" // (DESCRIPTION_REQUEST -> DESCRIPTION_PART), so a host without generated") + cog.outl(" // classes can still fetch the complete schema.") + cog.out(" static constexpr unsigned char SERVICE_OMITTED_CBOR[] = "); + cog.outl(xbot_codegen.binary2c_array(service["omitted_cbor"])); + cog.outl(" static constexpr size_t SERVICE_OMITTED_CBOR_SIZE = %d;" % + (len(service["omitted_cbor"]) if service["omitted_names"] else 0)) + cog.outl("") + cog.outl(" const unsigned char* GetOmittedDescription(size_t& size) override {") + cog.outl(" size = SERVICE_OMITTED_CBOR_SIZE;") + cog.outl(" return SERVICE_OMITTED_CBOR;") + cog.outl(" }") ]]]*/ static constexpr unsigned char SERVICE_DESCRIPTION_CBOR[] = { 0xA7, 0x64, 0x74, 0x79, 0x70, 0x65, 0x6F, 0x53, diff --git a/codegen/xbot_codegen/xbot_codegen.py b/codegen/xbot_codegen/xbot_codegen.py index b561b8c..53746b5 100644 --- a/codegen/xbot_codegen/xbot_codegen.py +++ b/codegen/xbot_codegen/xbot_codegen.py @@ -87,11 +87,52 @@ def common_attrs(json, valid_types, callback_name, send_method_name): } +# The advertisement is one UDP packet. What does not fit is not advertised but served +# on request (Service::HandleDescriptionRequest), so a host without generated classes can +# still fetch the complete schema; hosts with generated classes never need it. +# +# Budget: max_packet_size(1472) - sizeof(XbotHeader)(24) - the bytes the base class adds +# around the description (80, see the static_assert in templates/ServiceTemplate.cpp). +ADVERTISED_DESCRIPTION_BUDGET = 1472 - 24 - 80 + +# Blocks the advertisement may drop to fit, widest/most optional first. `type` and +# `version` always stay: they identify the service. +DROPPABLE_BLOCKS = ("enums", "functions", "registers", "outputs", "inputs") + + +def splitDescription(json_service: dict): + """Split the schema into what the advertisement carries and what it leaves out. + + Returns (advertised_cbor, omitted_cbor, omitted_names). Blocks are added in their + schema order while they fit, so the packet is filled as far as possible; the names of + everything that did not fit are listed under "omitted". + """ + droppable = [key for key in DROPPABLE_BLOCKS if key in json_service] + for dropped in range(len(droppable) + 1): + omitted_names = droppable[:dropped] + advertised = {"type": json_service["type"], "version": json_service["version"]} + for key, value in json_service.items(): + if key in ("type", "version") or key in omitted_names: + continue + advertised[key] = value + if omitted_names: + advertised["omitted"] = omitted_names + advertised_cbor = cbor2.dumps(advertised) + if len(advertised_cbor) <= ADVERTISED_DESCRIPTION_BUDGET: + omitted = {key: json_service[key] for key in omitted_names} + return advertised_cbor, cbor2.dumps(omitted), omitted_names + raise Exception( + f"{json_service['type']}: type/version alone do not fit into the advertisement " + f"budget of {ADVERTISED_DESCRIPTION_BUDGET} bytes") + + def loadService(path: str) -> dict: # Fetch the service definition with open(path) as f: json_service = json.load(f) + advertised_cbor, omitted_cbor, omitted_names = splitDescription(json_service) + # Build the dict for code generation. service = { "type": json_service["type"], @@ -99,7 +140,10 @@ def loadService(path: str) -> dict: "class_name": toCamelCase(json_service["type"]) + "Base", "interface_class_name": toCamelCase(json_service["type"]) + "InterfaceBase", "service_json": json.dumps(json_service, indent=2), - "service_cbor": cbor2.dumps(json_service) + # Advertised as far as it fits; the rest is served on request. + "service_cbor": advertised_cbor, + "omitted_cbor": omitted_cbor, + "omitted_names": omitted_names } # Consistency checks diff --git a/include/xbot/datatypes/DescriptionRequestPayload.hpp b/include/xbot/datatypes/DescriptionRequestPayload.hpp new file mode 100644 index 0000000..d83cc97 --- /dev/null +++ b/include/xbot/datatypes/DescriptionRequestPayload.hpp @@ -0,0 +1,26 @@ +// +// Payload of a DESCRIPTION_REQUEST. +// + +#ifndef DESCRIPTION_REQUEST_PAYLOAD_HPP +#define DESCRIPTION_REQUEST_PAYLOAD_HPP + +#include + +namespace xbot::datatypes { + +/** + * @brief Where the service should send the DESCRIPTION_PART to. + * + * The portable packet API only hands out the payload, not the sender's address, so the + * host brings its own endpoint along - the same idea as ClaimPayload. + */ +struct DescriptionRequestPayload { + uint16_t service_id{}; + uint32_t target_ip{}; + uint16_t target_port{}; +} __attribute__((packed)); + +} // namespace xbot::datatypes + +#endif // DESCRIPTION_REQUEST_PAYLOAD_HPP diff --git a/include/xbot/datatypes/XbotHeader.hpp b/include/xbot/datatypes/XbotHeader.hpp index 3f3d02e..ae87c2a 100644 --- a/include/xbot/datatypes/XbotHeader.hpp +++ b/include/xbot/datatypes/XbotHeader.hpp @@ -39,7 +39,18 @@ enum class MessageType : uint8_t { // Use for complex, infrequent, non-realtime critical messages (e.g. service // discovery). SERVICE_ADVERTISEMENT = 0x80, - SERVICE_QUERY = 0x81 + SERVICE_QUERY = 0x81, + // DESCRIPTION_REQUEST asks a service for the part of its schema that the advertisement + // could not carry: the advertisement is one packet (config::max_packet_size), so a + // growing schema does not fit into it forever. It is sent unicast by a host that needs + // the type definitions - a host with generated classes never asks. + // arg1 = byte offset into the omitted description, arg2 = maximum answer payload. + // Payload = DescriptionRequestPayload (where to send the answer). + DESCRIPTION_REQUEST = 0x82, + // DESCRIPTION_PART answers a DESCRIPTION_REQUEST with one window of the omitted + // description. arg1 = byte offset, arg2 = total size of the omitted description. + // Payload = the window bytes; the host repeats until offset + payload >= total. + DESCRIPTION_PART = 0x83 }; #pragma pack(push, 1) diff --git a/libxbot-service/include/xbot-service/Service.hpp b/libxbot-service/include/xbot-service/Service.hpp index bf67d55..8921998 100644 --- a/libxbot-service/include/xbot-service/Service.hpp +++ b/libxbot-service/include/xbot-service/Service.hpp @@ -111,7 +111,8 @@ class Service : public ServiceIo { /** * Called whenever a transaction was finished */ - virtual void OnTransactionEnd() {} + virtual void OnTransactionEnd() { + } const bool &IsRunning() { return is_running_; @@ -168,13 +169,27 @@ class Service : public ServiceIo { protected: size_t rpc_max_response_size_{0}; - private: + /** + * @brief Blocks of the schema that the advertisement could not carry. + * + * The advertisement is one packet (config::max_packet_size), so a service advertises + * as much of its schema as fits and serves the rest on request + * (DESCRIPTION_REQUEST -> DESCRIPTION_PART). Generated service bases override this with + * their static CBOR; size = 0 (the default) means everything is advertised and makes + * the service ignore description requests. + */ + virtual const uint8_t *GetOmittedDescription(size_t &size) { + size = 0; + return nullptr; + } + private: void HandleClaimMessage(datatypes::XbotHeader *header, const void *payload, size_t payload_len); void HandleDataMessage(datatypes::XbotHeader *header, const void *payload, size_t payload_len); void HandleDataTransaction(datatypes::XbotHeader *header, const void *payload, size_t payload_len); void HandleConfigurationTransaction(datatypes::XbotHeader *header, const void *payload, size_t payload_len); void HandleRpcCall(datatypes::XbotHeader *header, const void *payload, size_t payload_len); + void HandleDescriptionRequest(datatypes::XbotHeader *header, const void *payload, size_t payload_len); bool SetRegistersFromConfigurationMessage(const void *payload, size_t payload_len); // Builds and transmits an RPC_RESPONSE packet without touching rpc_in_progress_. diff --git a/libxbot-service/src/Service.cpp b/libxbot-service/src/Service.cpp index f8a8b47..0580552 100644 --- a/libxbot-service/src/Service.cpp +++ b/libxbot-service/src/Service.cpp @@ -9,6 +9,7 @@ #include #include #include +#include xbot::service::Service::Service(uint16_t service_id, void *processing_thread_stack, size_t processing_thread_stack_size) : ServiceIo(service_id), @@ -280,6 +281,9 @@ void xbot::service::Service::runProcessing() { HandleConfigurationTransaction(header, payload_buffer, header->payload_size); } break; + case datatypes::MessageType::DESCRIPTION_REQUEST: + HandleDescriptionRequest(header, payload_buffer, header->payload_size); + break; case datatypes::MessageType::RPC_CALL: if (is_running_) { HandleRpcCall(header, payload_buffer, header->payload_size); @@ -310,6 +314,58 @@ static uint32_t CalculateHeartbeatInterval(uint32_t heartbeat_micros) { return heartbeat_micros / 2; } +void xbot::service::Service::HandleDescriptionRequest(xbot::datatypes::XbotHeader *header, const void *payload, + size_t payload_len) { + if (payload_len != sizeof(datatypes::DescriptionRequestPayload)) { + ULOG_ARG_WARNING(&service_id_, "description request with invalid payload size"); + return; + } + const auto *request = static_cast(payload); + if (request->service_id != service_id_) { + return; + } + + /* A service whose whole schema fits into the advertisement has nothing to serve. */ + size_t total = 0; + const uint8_t *const omitted = GetOmittedDescription(total); + if (omitted == nullptr || total == 0U) { + ULOG_ARG_DEBUG(&service_id_, "description request, but nothing is omitted"); + return; + } + + const size_t offset = header->arg1; + if (offset >= total) { + ULOG_ARG_WARNING(&service_id_, "description request beyond the omitted description (%u >= %u)", + static_cast(offset), static_cast(total)); + return; + } + + size_t length = total - offset; + if (length > header->arg2) { + length = header->arg2; + } + constexpr size_t kMaxPayload = config::max_packet_size - sizeof(datatypes::XbotHeader); + if (length > kMaxPayload) { + length = kMaxPayload; + } + if (length == 0U) { + return; + } + + packet::PacketPtr ptr = packet::allocatePacket(); + { + Lock lk(&state_mutex_); + fillHeader(); + header_.message_type = datatypes::MessageType::DESCRIPTION_PART; + header_.arg1 = static_cast(offset); + header_.arg2 = static_cast(total); + header_.payload_size = static_cast(length); + packet::packetAppendData(ptr, &header_, sizeof(header_)); + } + packet::packetAppendData(ptr, omitted + offset, length); + Io::transmitPacket(ptr, request->target_ip, request->target_port); +} + void xbot::service::Service::HandleClaimMessage(xbot::datatypes::XbotHeader *header, const void *payload, size_t payload_len) { (void)header; diff --git a/xbot_service_interface_py/README.md b/xbot_service_interface_py/README.md index a4ab2c4..c1289ca 100644 --- a/xbot_service_interface_py/README.md +++ b/xbot_service_interface_py/README.md @@ -89,7 +89,17 @@ def connected(): print([i['name'] for i in schema.inputs]) ``` -The full service description is embedded in the CBOR advertisement packet. All inputs, outputs, registers and RPC functions are available immediately after connection. +The advertisement carries the interface - inputs, outputs, registers and RPC functions - plus as much +of the rest of the schema as fits into one UDP packet. A service whose schema is bigger (a long +`SoundId` enum, for example) lists what did not fit under `omitted`; the interface fetches those +blocks right after discovery (`DESCRIPTION_REQUEST` -> `DESCRIPTION_PART`, unicast, one round trip per +window), merges them and caches the result per `(service id, version)`, so a repeating +advertisement never fetches again. + +A schema passed in Mode 1 is validated once the description is complete, i.e. after those blocks +arrived. If the fetch fails - an older service that advertises everything, or a lost packet - the +interface keeps working with the blocks the advertisement carried and logs a warning; enum-typed +values then cannot be encoded by name. --- diff --git a/xbot_service_interface_py/tests/test_description_fetch.py b/xbot_service_interface_py/tests/test_description_fetch.py new file mode 100644 index 0000000..9b356f0 --- /dev/null +++ b/xbot_service_interface_py/tests/test_description_fetch.py @@ -0,0 +1,138 @@ +"""The advertisement is one packet, so a service advertises only as much of its schema as +fits and lists the rest under "omitted". A client that needs those blocks (one without +generated classes) fetches them with DESCRIPTION_REQUEST / DESCRIPTION_PART. +""" + +import socket +import struct +import threading + +import cbor2 + +from xbot_service_interface.datatypes import ( + HEADER_SIZE, MAX_PACKET_SIZE, MessageType, pack_header, unpack_header, +) +from xbot_service_interface.discovery import ServiceDiscovery + +SERVICE_ID = 13 +ENUMS = {'enums': [{'id': 'SoundId', 'base_type': 'uint8_t', + 'values': {'BOOT_PING': 0, 'MOWING_START': 14, 'RECOVERED': 31}}]} + + +class FakeService: + """Answers DESCRIPTION_REQUEST with `window` bytes per packet.""" + + def __init__(self, payload: dict, window: int = 40): + self.blob = cbor2.dumps(payload) + self.window = window + self.requests: list[tuple[int, int]] = [] + self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + self.sock.bind(('127.0.0.1', 0)) + self.sock.settimeout(0.2) + self.port = self.sock.getsockname()[1] + self._thread = threading.Thread(target=self._serve, daemon=True) + self._thread.start() + + def _serve(self) -> None: + while True: + try: + data, _ = self.sock.recvfrom(MAX_PACKET_SIZE) + except OSError: + return + hdr = unpack_header(data) + if hdr['message_type'] != MessageType.DESCRIPTION_REQUEST: + continue + sid, ip, port = struct.unpack(' bytes: + desc = {'type': 'SoundService', 'version': 1, + 'inputs': [{'id': 0, 'name': 'Volume', 'type': 'uint8_t'}], + 'functions': []} + desc['omitted'] = ['enums'] if omitted is None else omitted + adv = {'sid': SERVICE_ID, 'endpoint': {'ip': '127.0.0.1', 'port': self.port}, 'desc': desc} + body = cbor2.dumps(adv) + return pack_header(MessageType.SERVICE_ADVERTISEMENT, SERVICE_ID, 0, 0, 0, 0, len(body)) + body + + def close(self) -> None: + self.sock.close() + + +class Collector: + def __init__(self): + self.found = [] + + def on_service_found(self, service_id, ip, port, schema): + self.found.append(schema) + + +def _discovery(service: FakeService) -> tuple[ServiceDiscovery, Collector]: + discovery = ServiceDiscovery(bind_ip='127.0.0.1') + collector = Collector() + discovery.register_listener(collector) + return discovery, collector + + +def test_omitted_blocks_are_fetched_in_windows(): + service = FakeService(ENUMS, window=40) + discovery, collector = _discovery(service) + try: + discovery._handle_packet(service.advertisement()) + finally: + service.close() + + assert len(collector.found) == 1 + schema = collector.found[0] + assert schema.type == 'SoundService' + assert 'SoundId' in schema.enums_dict, "the fetched enums must end up in the schema" + assert [offset for offset, _ in service.requests] == [0, 40, 80] + assert all(want == MAX_PACKET_SIZE - HEADER_SIZE for _, want in service.requests) + + +def test_fetch_is_cached_per_version(): + service = FakeService(ENUMS) + discovery, collector = _discovery(service) + try: + discovery._handle_packet(service.advertisement()) + requests_after_first = len(service.requests) + discovery._handle_packet(service.advertisement()) + finally: + service.close() + + assert len(collector.found) == 1, "the second advertisement changes nothing" + assert len(service.requests) == requests_after_first, "a repeating advertisement must not fetch again" + + +def test_failed_fetch_keeps_the_advertisement_only(): + service = FakeService(ENUMS) + request_count = 0 + discovery, collector = _discovery(service) + discovery.FETCH_TIMEOUT_S = 0.05 + service.window = 0 # every window is empty -> the client never makes progress + try: + discovery._handle_packet(service.advertisement()) + finally: + service.close() + request_count = len(service.requests) + + assert len(collector.found) == 1 + assert collector.found[0].enums_dict == {}, "a failed fetch must not abort discovery" + assert request_count >= 1 + + +def test_nothing_is_fetched_when_the_service_advertises_everything(): + service = FakeService(ENUMS) + discovery, collector = _discovery(service) + try: + discovery._handle_packet(service.advertisement(omitted=[])) + finally: + service.close() + + assert len(collector.found) == 1 + assert service.requests == [] diff --git a/xbot_service_interface_py/xbot_service_interface/datatypes.py b/xbot_service_interface_py/xbot_service_interface/datatypes.py index fdff1bc..9f271ce 100644 --- a/xbot_service_interface_py/xbot_service_interface/datatypes.py +++ b/xbot_service_interface_py/xbot_service_interface/datatypes.py @@ -49,6 +49,12 @@ class MessageType(IntEnum): LOG = 0x7F SERVICE_ADVERTISEMENT = 0x80 SERVICE_QUERY = 0x81 + # Ask a service for the schema blocks its advertisement could not carry. + # arg1 = byte offset, arg2 = maximum answer payload, payload = request struct. + DESCRIPTION_REQUEST = 0x82 + # Answer to DESCRIPTION_REQUEST: arg1 = byte offset, arg2 = total size, + # payload = the window bytes. Repeat until offset + payload >= total. + DESCRIPTION_PART = 0x83 class RpcStatus(IntEnum): @@ -112,6 +118,19 @@ def pack_descriptor(target_id: int, payload_size: int) -> bytes: return struct.pack(DESCRIPTOR_FORMAT, target_id, 0, payload_size) +DESCRIPTION_REQUEST_FORMAT = ' bytes: + """Where the service should send the DESCRIPTION_PART to. + + The portable packet API only hands out the payload, not the sender's address, so + the host brings its own endpoint along - the same idea as the claim payload. + """ + return struct.pack(DESCRIPTION_REQUEST_FORMAT, service_id, target_ip_host, target_port) + + def pack_claim_payload(target_ip_host: int, target_port: int, heartbeat_micros: int) -> bytes: return struct.pack(CLAIM_FORMAT, target_ip_host, target_port, heartbeat_micros) diff --git a/xbot_service_interface_py/xbot_service_interface/discovery.py b/xbot_service_interface_py/xbot_service_interface/discovery.py index 7201a19..ff9f789 100644 --- a/xbot_service_interface_py/xbot_service_interface/discovery.py +++ b/xbot_service_interface_py/xbot_service_interface/discovery.py @@ -1,3 +1,4 @@ +import fcntl import socket import struct import threading @@ -7,13 +8,67 @@ from .datatypes import ( HEADER_SIZE, MULTICAST_PORT, SD_MULTICAST_ADDR, - MAX_PACKET_SIZE, MessageType, unpack_header, + MAX_PACKET_SIZE, MessageType, unpack_header, pack_header, + pack_description_request, ) from .schema import ServiceSchema log = logging.getLogger(__name__) +_SKIP_IFACE_PREFIXES = ('lo', 'docker', 'veth', 'virbr', 'br-', 'wg', 'tun', 'tap') + + +def _route_source_ip(destination: str) -> str: + """Local address the kernel would send a packet to @p destination from. + + A UDP connect() only performs the route lookup (nothing is transmitted), so this + picks the interface that can actually reach the service. + + Falls back to _primary_ip() when the host has no route at all. + """ + try: + probe = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + probe.connect((destination, 9)) # 'discard' port, only the route matters + address = probe.getsockname()[0] + if address and address != '0.0.0.0': + return address + finally: + probe.close() + except OSError: + pass + return _primary_ip() + + +def _primary_ip() -> str: + """Fallback for _route_source_ip(): first plausible non-loopback interface address. + + Skips loopback and virtual interfaces (docker, veth, bridges, VPNs), mirroring + io._get_primary_ip() to avoid a circular import. + + The discovery socket is bound to the multicast group, so its own address is useless + as a reply target. + """ + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + for _idx, name in socket.if_nameindex(): + if any(name.startswith(prefix) for prefix in _SKIP_IFACE_PREFIXES): + continue + try: + ifreq = struct.pack('16sH14s', name.encode(), socket.AF_INET, b'\x00' * 14) + result = fcntl.ioctl(sock, 0x8915, ifreq) # SIOCGIFADDR + return socket.inet_ntoa(result[20:24]) + except OSError: + continue + finally: + sock.close() + except Exception: + pass + return '127.0.0.1' + + class ServiceDiscovery: """Listens for SERVICE_ADVERTISEMENT packets on the multicast group 233.255.255.0:4242 and notifies registered listeners. @@ -22,8 +77,21 @@ class ServiceDiscovery: def on_service_found(service_id, ip, port, schema: ServiceSchema): ... """ - def __init__(self, bind_ip: str = '0.0.0.0'): + #: How long to wait for the DESCRIPTION_PART answers of one window. + FETCH_TIMEOUT_S = 1 + + #: Upper bound on the windows one fetch may take (guards a misbehaving service). + FETCH_MAX_WINDOWS = 16 + + def __init__(self, bind_ip: str = '0.0.0.0', fetch_omitted: bool = True): self._bind_ip = bind_ip + #: A service advertises only as much of its schema as fits into one packet and + #: lists the rest under "omitted". Set False to keep the schema incomplete - + #: clients that have generated classes do not need the blocks at all. + self._fetch_omitted = fetch_omitted + #: service_id -> {"version": int, "blocks": dict} so a repeating advertisement + #: does not fetch again. + self._omitted_cache: dict[int, dict] = {} self._sock: socket.socket | None = None self._thread: threading.Thread | None = None self._running = False @@ -96,6 +164,84 @@ def stop(self) -> None: self._sock = None log.info("ServiceDiscovery stopped") + # ------------------------------------------------------------------ + # Omitted schema blocks (see DESCRIPTION_REQUEST in libxbot-service) + # ------------------------------------------------------------------ + + def _omitted_blocks(self, service_id: int, ip: str, port: int, version: int) -> dict | None: + """Fetch the schema blocks the advertisement left out; None when it did not work.""" + cached = self._omitted_cache.get(service_id) + if cached is not None and cached['version'] == version: + return cached['blocks'] + + target_ip = self._bind_ip + if target_ip == '0.0.0.0': + # The service has to send the answer somewhere: hand it the address this host + # reaches the service from (the request payload carries ip:port, see + # DescriptionRequestPayload). + target_ip = _route_source_ip(ip) + + data = bytearray() + total = None + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP) + sock.bind(('', 0)) + sock.settimeout(self.FETCH_TIMEOUT_S) + my_port = sock.getsockname()[1] + request = pack_description_request(service_id, struct.unpack('= total: + break + offset = len(data) + header = pack_header(MessageType.DESCRIPTION_REQUEST, service_id, offset, + MAX_PACKET_SIZE - HEADER_SIZE, 0, 0, len(request)) + sock.sendto(header + request, (ip, port)) + + # One window per request: wait for the answer that starts at len(data). + # Everything else (a stale window, a foreign packet) is skipped. + while True: + try: + answer, _ = sock.recvfrom(MAX_PACKET_SIZE) + except socket.timeout: + return None + if len(answer) < HEADER_SIZE: + continue + hdr = unpack_header(answer) + if hdr['message_type'] != MessageType.DESCRIPTION_PART: + continue + if len(answer) != HEADER_SIZE + hdr['payload_size']: + continue + if hdr['arg1'] != len(data): + continue + total = hdr['arg2'] + data += answer[HEADER_SIZE:] + break + if total == 0: + return {} + except OSError as e: + log.warning(f"Service {service_id}: description fetch failed: {e}") + return None + finally: + try: + sock.close() + except OSError: + pass + + if total is None or len(data) != total: + return None + try: + blocks = cbor2.loads(bytes(data)) + except Exception as e: + log.warning(f"Service {service_id}: omitted schema is not valid CBOR: {e}") + return None + if not isinstance(blocks, dict): + return None + + self._omitted_cache[service_id] = {'version': version, 'blocks': blocks} + log.debug(f"Service {service_id}: fetched omitted schema blocks {list(blocks)}") + return blocks + # ------------------------------------------------------------------ # Internal # ------------------------------------------------------------------ @@ -137,7 +283,7 @@ def _handle_packet(self, data: bytes) -> None: service_id = msg.get('sid') endpoint = msg.get('endpoint', {}) - desc = msg.get('desc', {}) + desc = dict(msg.get('desc', {})) if service_id is None: return @@ -148,6 +294,19 @@ def _handle_packet(self, data: bytes) -> None: log.warning(f"Service {service_id} advertised with invalid endpoint, ignoring") return + # The advertisement is one packet: whatever did not fit is fetched on demand. + omitted = desc.get('omitted') or [] + if omitted and self._fetch_omitted: + details = self._omitted_blocks(service_id, ip, port, desc.get('version', 0)) + if details is None: + log.warning( + f"Service {service_id}: could not fetch the omitted schema blocks {omitted};" + " using the advertisement only") + else: + del desc['omitted'] + desc.update(details) + elif 'omitted' in desc: + del desc['omitted'] try: schema = ServiceSchema.from_dict(desc) except Exception as e: From e470df61b0bdfa3d7402a29023436b583faa4132 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=B6rg=20Ebeling?= Date: Mon, 2 Nov 2026 20:54:34 +0100 Subject: [PATCH 2/7] ci: auto-regenerate Go message types from C++ header Add a workflow step that runs `go generate` and commits the regenerated Go sources back to the branch when they drift from `XbotHeader.hpp`, so contributors can add message types without a local Go toolchain. Only branches that may be pushed to are committed to (non-default, non-fork); otherwise the run fails with instructions. The regenerate/push steps use scoped write permissions and a separate push step to keep the token out of the `go generate` environment, with a fallback to GITHUB_TOKEN and a warning when BOT_TOKEN is unset. Concurrency is grouped per branch to serialize pushes. --- .github/workflows/go.yaml | 72 ++++++++++ .../cmd/xbot-codegen/main.go | 10 ++ .../cmd/xbot-codegen/messagetypes.go | 125 ++++++++++++++++++ xbot_service_interface_go/xbot/generate.go | 7 + xbot_service_interface_go/xbot/header.go | 53 -------- .../xbot/message_type_generated.go | 55 ++++++++ xbot_service_interface_go/xbot/spec_test.go | 17 +-- 7 files changed, 274 insertions(+), 65 deletions(-) create mode 100644 xbot_service_interface_go/cmd/xbot-codegen/messagetypes.go create mode 100644 xbot_service_interface_go/xbot/generate.go create mode 100644 xbot_service_interface_go/xbot/message_type_generated.go diff --git a/.github/workflows/go.yaml b/.github/workflows/go.yaml index 20eb364..522e090 100644 --- a/.github/workflows/go.yaml +++ b/.github/workflows/go.yaml @@ -1,5 +1,12 @@ name: go +# The Go module is part of a polyglot repository: `xbot/message_type_generated.go` mirrors +# `enum class MessageType` in include/xbot/datatypes/XbotHeader.hpp. Regenerating and +# committing that here means a developer can add a message type (or a service definition) +# without installing Go - the same pattern as in the definitions repository. +# +# Fix and verification share one run on purpose: a push made with GITHUB_TOKEN does not +# trigger further workflow runs. on: pull_request: push: @@ -8,20 +15,85 @@ on: permissions: contents: read +# A push to a branch with an open pull request fires both events, and github.ref differs +# between them, so they have to share one group per branch to serialize the commit back. +concurrency: + group: go-${{ github.event.pull_request.head.repo.full_name || github.repository }}-${{ github.head_ref || github.ref_name }} + cancel-in-progress: false + jobs: test: runs-on: ubuntu-latest + # The regenerate step commits back; every other step stays read-only. + permissions: + contents: write defaults: run: working-directory: xbot_service_interface_go steps: - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4 with: + # A pull_request checkout is the synthetic merge commit (detached), and a fork + # PR branch only exists in the fork. + repository: ${{ github.event.pull_request.head.repo.full_name || github.repository }} + ref: ${{ github.head_ref || github.ref_name }} persist-credentials: false - uses: actions/setup-go@40f1582b2485089dde7abd97c1529aa768e1baff # v5 with: go-version-file: xbot_service_interface_go/go.mod cache-dependency-path: xbot_service_interface_go/go.sum + + - name: Regenerate the Go sources + id: regenerate + env: + BRANCH: ${{ github.head_ref || github.ref_name }} + # Only a branch we may push to: not the default branch, and not a fork. + CAN_COMMIT: >- + ${{ (github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name == github.repository) + || (github.event_name == 'push' && github.ref_name != github.event.repository.default_branch) }} + run: | + go generate ./... + cd .. + # Only the Go module is generated; the C++ header is the author's edit. + if git diff --quiet -- xbot_service_interface_go && + [ -z "$(git ls-files --others --exclude-standard -- xbot_service_interface_go)" ]; then + echo "::notice::The Go sources already match the C++ header." + exit 0 + fi + if [ "$CAN_COMMIT" != "true" ]; then + git --no-pager diff --stat -- xbot_service_interface_go + echo "::error::The Go sources do not match the C++ header and this run must not commit them (default branch or fork). Run 'cd xbot_service_interface_go && go generate ./...' and commit the result." + exit 1 + fi + git config user.name 'github-actions[bot]' + git config user.email '41898282+github-actions[bot]@users.noreply.github.com' + git add -A -- xbot_service_interface_go + git commit -m 'chore(go): regenerate the message types from XbotHeader.hpp' + echo "committed=true" >> "$GITHUB_OUTPUT" + + - name: Push the regenerated sources + # Separate step so the token never sits in the environment of the step that runs + # `go generate` on a possibly forked branch. + if: steps.regenerate.outputs.committed == 'true' + env: + # GITHUB_TOKEN pushes do not trigger workflow runs, so the regenerated commit + # would carry no check runs; a PAT (contents: write) in BOT_TOKEN does. + BOT_TOKEN: ${{ secrets.BOT_TOKEN }} + GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} + BRANCH: ${{ github.head_ref || github.ref_name }} + run: | + if [ -n "${BOT_TOKEN}" ]; then + token="${BOT_TOKEN}" + else + token="${GH_TOKEN}" + echo "::warning::BOT_TOKEN is not set, pushing with GITHUB_TOKEN - the regenerated commit will not trigger a workflow run, so required status checks stay pending." + fi + if ! git push "https://x-access-token:${token}@github.com/${GITHUB_REPOSITORY}.git" "HEAD:${BRANCH}"; then + echo "::error::Could not push the regenerated sources to ${BRANCH}." + exit 1 + fi + echo "::notice::Committed the regenerated Go sources to ${BRANCH}." + - name: Format run: | unformatted="$(gofmt -l .)" diff --git a/xbot_service_interface_go/cmd/xbot-codegen/main.go b/xbot_service_interface_go/cmd/xbot-codegen/main.go index 1264807..8fb3be5 100644 --- a/xbot_service_interface_go/cmd/xbot-codegen/main.go +++ b/xbot_service_interface_go/cmd/xbot-codegen/main.go @@ -21,11 +21,21 @@ import ( func main() { service := flag.String("service", "", "path to the service.json (required)") + messageTypes := flag.Bool("message-types", false, "generate the Go message types from enum class MessageType in the C++ header") + header := flag.String("header", "../../include/xbot/datatypes/XbotHeader.hpp", "C++ header with enum class MessageType (-message-types)") out := flag.String("out", "", "output file (default: stdout)") serviceID := flag.Uint("service-id", 0, "service id from service_ids.h (0: Dial takes the id as an argument)") pkg := flag.String("package", "", "package name (default: derived from the service type)") flag.Parse() + if *messageTypes { + if err := generateMessageTypes(*header, *out); err != nil { + fmt.Fprintf(os.Stderr, "xbot-codegen: %v\n", err) + os.Exit(1) + } + return + } + if *service == "" { fmt.Fprintln(os.Stderr, "xbot-codegen: -service is required") flag.Usage() diff --git a/xbot_service_interface_go/cmd/xbot-codegen/messagetypes.go b/xbot_service_interface_go/cmd/xbot-codegen/messagetypes.go new file mode 100644 index 0000000..11dffe6 --- /dev/null +++ b/xbot_service_interface_go/cmd/xbot-codegen/messagetypes.go @@ -0,0 +1,125 @@ +package main + +// Generates xbot/message_type_generated.go from the C++ header, so the Go message types +// cannot drift away from include/xbot/datatypes/XbotHeader.hpp. The header is the single +// source of truth for the wire protocol; this generator is what CI runs, which is why a +// developer can add a message type without installing Go. + +import ( + "fmt" + "go/format" + "os" + "regexp" + "sort" + "strings" +) + +var ( + enumBlock = regexp.MustCompile(`(?s)enum class MessageType[^{]*\{(.*?)\}`) + enumEntry = regexp.MustCompile(`(\w+)\s*=\s*(0x[0-9A-Fa-f]+)`) +) + +// initialisms stay upper case in Go names (MsgRPCCall, not MsgRpcCall). +var initialisms = map[string]bool{ + "API": true, "CBOR": true, "ID": true, "IO": true, "JSON": true, + "RPC": true, "TCP": true, "UDP": true, "URL": true, +} + +func goName(symbol string) string { + var b strings.Builder + b.WriteString("Msg") + for _, part := range strings.Split(symbol, "_") { + if part == "" { + continue + } + if initialisms[part] { + b.WriteString(part) + continue + } + b.WriteString(strings.ToUpper(part[:1])) + b.WriteString(strings.ToLower(part[1:])) + } + return b.String() +} + +func generateMessageTypes(headerPath, outPath string) error { + src, err := os.ReadFile(headerPath) + if err != nil { + return err + } + block := enumBlock.FindSubmatch(src) + if block == nil { + return fmt.Errorf("%s: enum class MessageType not found - the header format changed", headerPath) + } + entries := enumEntry.FindAllSubmatch(block[1], -1) + if len(entries) == 0 { + return fmt.Errorf("%s: no enumerators found in enum class MessageType", headerPath) + } + + type entry struct { + symbol string + goName string + value string + } + var all []entry + for _, m := range entries { + symbol := string(m[1]) + all = append(all, entry{symbol: symbol, goName: goName(symbol), value: string(m[2])}) + } + + // Align the const block like gofmt would. + width := 0 + for _, e := range all { + if len(e.goName) > width { + width = len(e.goName) + } + } + + var b strings.Builder + b.WriteString("// Code generated by 'go generate' (xbot-codegen -message-types) from\n") + b.WriteString("// include/xbot/datatypes/XbotHeader.hpp. DO NOT EDIT.\n\n") + b.WriteString("package xbot\n\n") + b.WriteString("import \"fmt\"\n\n") + b.WriteString("// MessageType is the 8 bit message type of an XbotHeader\n") + b.WriteString("// (see include/xbot/datatypes/XbotHeader.hpp).\n") + b.WriteString("//\n") + b.WriteString("// The first bit of the value tells how the payload is encoded: 0 = raw binary\n") + b.WriteString("// (DATA, RPC, CLAIM, ...), 1 = CBOR (service discovery).\n") + b.WriteString("type MessageType uint8\n\n") + b.WriteString("// Message types, generated from the C++ header.\nconst (\n") + for _, e := range all { + fmt.Fprintf(&b, "\t%-*s MessageType = %s\n", width, e.goName, e.value) + } + b.WriteString(")\n\n") + b.WriteString("// messageTypeNames is the wire symbol of every message type.\n") + b.WriteString("var messageTypeNames = map[MessageType]string{\n") + for _, e := range all { + fmt.Fprintf(&b, "\t%-*s %q,\n", width, e.goName+":", e.symbol) + } + b.WriteString("}\n\n") + b.WriteString("// String returns the symbol of the message type, e.g. \"RPC_CALL\".\n") + b.WriteString("func (m MessageType) String() string {\n") + b.WriteString("\tif name, ok := messageTypeNames[m]; ok {\n\t\treturn name\n\t}\n") + b.WriteString("\treturn fmt.Sprintf(\"MessageType(0x%02x)\", uint8(m))\n}\n") + + // go/format instead of hand alignment: the result is gofmt-clean by construction, + // which is what the CI format check expects right after a regeneration. + formatted, err := format.Source([]byte(b.String())) + if err != nil { + return fmt.Errorf("formatting the generated code: %w", err) + } + if outPath == "" { + _, err = os.Stdout.Write(formatted) + return err + } + if err = os.WriteFile(outPath, formatted, 0o644); err != nil { + return err + } + names := make([]string, 0, len(all)) + for _, e := range all { + names = append(names, e.symbol) + } + sort.Strings(names) + fmt.Printf("wrote %s (%d message types)\n", outPath, len(names)) + return nil +} diff --git a/xbot_service_interface_go/xbot/generate.go b/xbot_service_interface_go/xbot/generate.go new file mode 100644 index 0000000..09f7b20 --- /dev/null +++ b/xbot_service_interface_go/xbot/generate.go @@ -0,0 +1,7 @@ +package xbot + +// The message types mirror `enum class MessageType` in the C++ header, so they are +// generated instead of maintained twice. CI runs this and commits the result, which is +// why adding a message type needs no Go toolchain on the developer's machine. +// +//go:generate go run ../cmd/xbot-codegen -message-types -header ../../include/xbot/datatypes/XbotHeader.hpp -out message_type_generated.go diff --git a/xbot_service_interface_go/xbot/header.go b/xbot_service_interface_go/xbot/header.go index c5762dc..65938b0 100644 --- a/xbot_service_interface_go/xbot/header.go +++ b/xbot_service_interface_go/xbot/header.go @@ -7,59 +7,6 @@ import ( "time" ) -// MessageType is the 8 bit message type of an XbotHeader -// (see include/xbot/datatypes/XbotHeader.hpp). -// -// The first bit of the value tells how the payload is encoded: 0 = raw binary -// (DATA, RPC, CLAIM, ...), 1 = CBOR (service discovery). -type MessageType uint8 - -// Message types. Not all of them are used by this client yet — the list -// documents the wire protocol. -const ( - MsgUnknown MessageType = 0x00 - MsgData MessageType = 0x01 - MsgConfigurationRequest MessageType = 0x02 - MsgClaim MessageType = 0x03 - MsgHeartbeat MessageType = 0x04 - MsgTransaction MessageType = 0x05 - MsgRPCCall MessageType = 0x06 - MsgRPCResponse MessageType = 0x07 - MsgLog MessageType = 0x7F - MsgServiceAdvertisement MessageType = 0x80 - MsgServiceQuery MessageType = 0x81 -) - -// String returns the symbol of the message type, e.g. "RPC_CALL". -func (m MessageType) String() string { - switch m { - case MsgData: - return "DATA" - case MsgConfigurationRequest: - return "CONFIGURATION_REQUEST" - case MsgClaim: - return "CLAIM" - case MsgHeartbeat: - return "HEARTBEAT" - case MsgTransaction: - return "TRANSACTION" - case MsgRPCCall: - return "RPC_CALL" - case MsgRPCResponse: - return "RPC_RESPONSE" - case MsgLog: - return "LOG" - case MsgServiceAdvertisement: - return "SERVICE_ADVERTISEMENT" - case MsgServiceQuery: - return "SERVICE_QUERY" - case MsgUnknown: - return "UNKNOWN" - default: - return fmt.Sprintf("MessageType(0x%02x)", uint8(m)) - } -} - // RPC status codes (arg1 of an RPC_RESPONSE). const ( RPCStatusSuccess uint8 = 0 diff --git a/xbot_service_interface_go/xbot/message_type_generated.go b/xbot_service_interface_go/xbot/message_type_generated.go new file mode 100644 index 0000000..2da9d24 --- /dev/null +++ b/xbot_service_interface_go/xbot/message_type_generated.go @@ -0,0 +1,55 @@ +// Code generated by 'go generate' (xbot-codegen -message-types) from +// include/xbot/datatypes/XbotHeader.hpp. DO NOT EDIT. + +package xbot + +import "fmt" + +// MessageType is the 8 bit message type of an XbotHeader +// (see include/xbot/datatypes/XbotHeader.hpp). +// +// The first bit of the value tells how the payload is encoded: 0 = raw binary +// (DATA, RPC, CLAIM, ...), 1 = CBOR (service discovery). +type MessageType uint8 + +// Message types, generated from the C++ header. +const ( + MsgUnknown MessageType = 0x00 + MsgData MessageType = 0x01 + MsgConfigurationRequest MessageType = 0x02 + MsgClaim MessageType = 0x03 + MsgHeartbeat MessageType = 0x04 + MsgTransaction MessageType = 0x05 + MsgRPCCall MessageType = 0x06 + MsgRPCResponse MessageType = 0x07 + MsgLog MessageType = 0x7F + MsgServiceAdvertisement MessageType = 0x80 + MsgServiceQuery MessageType = 0x81 + MsgDescriptionRequest MessageType = 0x82 + MsgDescriptionPart MessageType = 0x83 +) + +// messageTypeNames is the wire symbol of every message type. +var messageTypeNames = map[MessageType]string{ + MsgUnknown: "UNKNOWN", + MsgData: "DATA", + MsgConfigurationRequest: "CONFIGURATION_REQUEST", + MsgClaim: "CLAIM", + MsgHeartbeat: "HEARTBEAT", + MsgTransaction: "TRANSACTION", + MsgRPCCall: "RPC_CALL", + MsgRPCResponse: "RPC_RESPONSE", + MsgLog: "LOG", + MsgServiceAdvertisement: "SERVICE_ADVERTISEMENT", + MsgServiceQuery: "SERVICE_QUERY", + MsgDescriptionRequest: "DESCRIPTION_REQUEST", + MsgDescriptionPart: "DESCRIPTION_PART", +} + +// String returns the symbol of the message type, e.g. "RPC_CALL". +func (m MessageType) String() string { + if name, ok := messageTypeNames[m]; ok { + return name + } + return fmt.Sprintf("MessageType(0x%02x)", uint8(m)) +} diff --git a/xbot_service_interface_go/xbot/spec_test.go b/xbot_service_interface_go/xbot/spec_test.go index 21e990c..7de841f 100644 --- a/xbot_service_interface_go/xbot/spec_test.go +++ b/xbot_service_interface_go/xbot/spec_test.go @@ -114,18 +114,11 @@ func TestMessageTypesMatchHeaderHpp(t *testing.T) { t.Fatal("no enumerators parsed from XbotHeader.hpp") } - ours := map[string]MessageType{ - "UNKNOWN": MsgUnknown, - "DATA": MsgData, - "CONFIGURATION_REQUEST": MsgConfigurationRequest, - "CLAIM": MsgClaim, - "HEARTBEAT": MsgHeartbeat, - "TRANSACTION": MsgTransaction, - "RPC_CALL": MsgRPCCall, - "RPC_RESPONSE": MsgRPCResponse, - "LOG": MsgLog, - "SERVICE_ADVERTISEMENT": MsgServiceAdvertisement, - "SERVICE_QUERY": MsgServiceQuery, + // The Go side is generated from the header (see generate.go); build the name -> value + // map from it so this test only fails when the generated file is stale. + ours := make(map[string]MessageType, len(messageTypeNames)) + for value, name := range messageTypeNames { + ours[name] = value } seen := make(map[string]bool, len(entries)) for _, e := range entries { From 7d11dfa87f3e3b4bf08c28e4928866267a8eff3c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=B6rg=20Ebeling?= Date: Mon, 2 Nov 2026 21:44:58 +0100 Subject: [PATCH 3/7] fix(discovery): use claim IP convention for description fetch and merge schemas Description fetch requests packed target_ip as network-order bytes, but services interpret it as a host-order integer (converting via htonl), silently sending responses to byte-swapped addresses. Fixed test fake to unpack with '!I' and added regression tests verifying the request matches the claim payload layout. Also extended tests to cover both usage modes receiving the merged schema after a fetch: mode 2 builds its schema from discovery, and mode 1 validates the provided schema against the completed one while still rejecting version mismatches. --- .../tests/test_description_fetch.py | 76 ++++++++++++++++++- .../xbot_service_interface/discovery.py | 5 +- 2 files changed, 79 insertions(+), 2 deletions(-) diff --git a/xbot_service_interface_py/tests/test_description_fetch.py b/xbot_service_interface_py/tests/test_description_fetch.py index 9b356f0..f8c1928 100644 --- a/xbot_service_interface_py/tests/test_description_fetch.py +++ b/xbot_service_interface_py/tests/test_description_fetch.py @@ -49,7 +49,8 @@ def _serve(self) -> None: chunk = self.blob[hdr['arg1']:hdr['arg1'] + self.window] answer = pack_header(MessageType.DESCRIPTION_PART, SERVICE_ID, hdr['arg1'], len(self.blob), 0, 0, len(chunk)) + chunk - self.sock.sendto(answer, (socket.inet_ntoa(struct.pack(' bytes: desc = {'type': 'SoundService', 'version': 1, @@ -136,3 +137,76 @@ def test_nothing_is_fetched_when_the_service_advertises_everything(): assert len(collector.found) == 1 assert service.requests == [] + + +# --------------------------------------------------------------------------------- +# Both usage modes must work with a description that had to be completed by a fetch: +# the interface receives the *merged* schema from the discovery, never a partial one. +# --------------------------------------------------------------------------------- + +def _advertised_desc(version: int = 1) -> dict: + return {'type': 'SoundService', 'version': version, + 'inputs': [{'id': 0, 'name': 'Volume', 'type': 'uint8_t'}], + 'functions': []} + + +def _merged_schema(): + from xbot_service_interface.schema import ServiceSchema + desc = _advertised_desc() + desc.pop('omitted', None) + desc.update(ENUMS) + return ServiceSchema.from_dict(desc) + + +def test_mode_2_uses_the_completed_schema(): + from xbot_service_interface.interface import ServiceInterface + from xbot_service_interface.serialization import pack_value + + interface = ServiceInterface(service_id=SERVICE_ID) # no schema= argument + interface._on_service_discovered('127.0.0.1', 4711, _merged_schema()) + + enums = interface._active_schema.enums_dict + assert 'SoundId' in enums + # An enum-typed argument can be given by name, which is the whole point of the enums. + assert pack_value('SoundId', 'MOWING_START', enums) == b'\x0e' + + +def test_mode_1_validates_against_the_completed_schema(): + from xbot_service_interface.interface import ServiceInterface + + provided = _advertised_desc() + provided['functions'] = [] + interface = ServiceInterface(service_id=SERVICE_ID, schema=provided) + # Must not raise: the advertised type/version still match, the enums were fetched. + interface._on_service_discovered('127.0.0.1', 4711, _merged_schema()) + assert interface._active_schema is interface._schema + + +def test_mode_1_still_rejects_an_incompatible_service(): + import pytest + + from xbot_service_interface.exceptions import IncompatibleServiceError + from xbot_service_interface.interface import ServiceInterface + from xbot_service_interface.schema import ServiceSchema + + provided = _advertised_desc(version=2) + interface = ServiceInterface(service_id=SERVICE_ID, schema=provided) + with pytest.raises(IncompatibleServiceError): + interface._on_service_discovered('127.0.0.1', 4711, _merged_schema()) + + +def test_description_request_uses_the_claim_ip_convention(): + """The wire carries the host-order integer of the address, not the network bytes. + + The service turns target_ip into a destination with htonl(), exactly like the claim + response does. Packing it the other way round sends every answer to a byte-swapped + address - which is silent: the request is served, the answer just goes nowhere. + """ + from xbot_service_interface.datatypes import pack_claim_payload, pack_description_request + + ip_host = struct.unpack('!I', socket.inet_aton('172.16.78.1'))[0] + request = pack_description_request(13, ip_host, 4711) + claim = pack_claim_payload(ip_host, 4711, 1_000_000) + + assert request[2:6] == claim[0:4], "target_ip must be packed like the claim payload" + assert request[6:8] == claim[4:6], "target_port must be packed like the claim payload" diff --git a/xbot_service_interface_py/xbot_service_interface/discovery.py b/xbot_service_interface_py/xbot_service_interface/discovery.py index ff9f789..9a0f936 100644 --- a/xbot_service_interface_py/xbot_service_interface/discovery.py +++ b/xbot_service_interface_py/xbot_service_interface/discovery.py @@ -188,7 +188,10 @@ def _omitted_blocks(self, service_id: int, ip: str, port: int, version: int) -> sock.bind(('', 0)) sock.settimeout(self.FETCH_TIMEOUT_S) my_port = sock.getsockname()[1] - request = pack_description_request(service_id, struct.unpack('= total: From 5153bf4cb78c568bf2dff339f14135917fa3d8f8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=B6rg=20Ebeling?= Date: Tue, 3 Nov 2026 17:42:19 +0100 Subject: [PATCH 4/7] fix(codegen): define SERVICE_OMITTED_CBOR out of line GetOmittedDescription() returns the member as a pointer (odr-use), so it needs a definition. The firmware compiles as C++17 where static constexpr members are implicitly inline, but catkin/mower_simulation uses gnu++14 - the ROS build then fails to link with "undefined reference to ...SERVICE_OMITTED_CBOR". --- codegen/templates/ServiceTemplate.cpp | 2 ++ 1 file changed, 2 insertions(+) diff --git a/codegen/templates/ServiceTemplate.cpp b/codegen/templates/ServiceTemplate.cpp index ff04aeb..e17fec4 100644 --- a/codegen/templates/ServiceTemplate.cpp +++ b/codegen/templates/ServiceTemplate.cpp @@ -21,8 +21,10 @@ cog.outl(f'#include "{service["class_name"]}.hpp"') /*[[[cog cog.outl(f"constexpr unsigned char {service['class_name']}::SERVICE_DESCRIPTION_CBOR[];") +cog.outl(f"constexpr unsigned char {service['class_name']}::SERVICE_OMITTED_CBOR[];") ]]]*/ constexpr unsigned char ServiceTemplateBase::SERVICE_DESCRIPTION_CBOR[]; +constexpr unsigned char ServiceTemplateBase::SERVICE_OMITTED_CBOR[]; //[[[end]]] From 0db7e07ab8167d352be720572a1a54981d3e3bc5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=B6rg=20Ebeling?= Date: Wed, 4 Nov 2026 19:43:16 +0100 Subject: [PATCH 5/7] docs: list DESCRIPTION_REQUEST/PART in the message type table The table ended at SERVICE_QUERY, so the two new message types were only described in prose further down. Adds the rows (with their arg1/arg2 meaning) and a short note on the request/part window handshake. --- Readme.md | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/Readme.md b/Readme.md index 17fd4c8..c57fd5c 100644 --- a/Readme.md +++ b/Readme.md @@ -107,6 +107,14 @@ message-specific payload. | `0x7F` | LOG | log level (1–7) | — | UTF-8 string (max 255 bytes) | | `0x80` | SERVICE_ADVERTISEMENT | — | — | CBOR-encoded service JSON | | `0x81` | SERVICE_QUERY | — | — | — | +| `0x82` | DESCRIPTION_REQUEST | offset into the omitted desc | `max_size` | `DescriptionRequestPayload` (ip:port) | +| `0x83` | DESCRIPTION_PART | offset into the omitted desc | `total_size`| Window of the omitted description | + +`DESCRIPTION_REQUEST` is sent unicast by a host that needs the part of the schema the +advertisement could not carry (see *Service description* below); the service answers with one +`DESCRIPTION_PART` window and the host repeats until `offset + payload_size >= total`. A +service whose whole schema fits into one advertisement reports size 0 and ignores the +request. ### TRANSACTION payload format From 953c89dcf0188fbecbf1f05d20bc975458d94e1e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=B6rg=20Ebeling?= Date: Wed, 4 Nov 2026 19:44:02 +0100 Subject: [PATCH 6/7] chore(py): 0.2.0 for the description fetch The interface now fetches and merges the schema blocks a service omits from its advertisement (DESCRIPTION_REQUEST -> DESCRIPTION_PART), so the package needs its own release: 0.1.3 -> 0.2.0. --- xbot_service_interface_py/pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/xbot_service_interface_py/pyproject.toml b/xbot_service_interface_py/pyproject.toml index bc08ea7..cf527bf 100644 --- a/xbot_service_interface_py/pyproject.toml +++ b/xbot_service_interface_py/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "xbot-service-interface-py" -version = "0.1.3" +version = "0.2.0" description = "Python service interface for the xBot Framework — connect to xBot services over UDP without code generation" readme = "README.md" license = "MIT" From 5cad9941f83db5a859cc54719c59b51ac00eab34 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=B6rg=20Ebeling?= Date: Wed, 4 Nov 2026 20:16:16 +0100 Subject: [PATCH 7/7] Fix CodeRabbit findings --- .github/workflows/go.yaml | 52 +++++---- Readme.md | 7 +- codegen/xbot_codegen/xbot_codegen.py | 17 ++- .../datatypes/DescriptionRequestPayload.hpp | 7 +- include/xbot/datatypes/XbotHeader.hpp | 10 +- .../include/xbot-service/Service.hpp | 3 + libxbot-service/src/Service.cpp | 37 +++++- .../cmd/xbot-codegen/messagetypes.go | 2 +- xbot_service_interface_go/xbot/spec_test.go | 2 +- xbot_service_interface_py/README.md | 14 ++- .../tests/test_description_fetch.py | 30 +++-- .../xbot_service_interface/datatypes.py | 22 ++-- .../xbot_service_interface/discovery.py | 106 +++++++++++++----- 13 files changed, 222 insertions(+), 87 deletions(-) diff --git a/.github/workflows/go.yaml b/.github/workflows/go.yaml index 522e090..52f3c88 100644 --- a/.github/workflows/go.yaml +++ b/.github/workflows/go.yaml @@ -31,20 +31,49 @@ jobs: run: working-directory: xbot_service_interface_go steps: + # Format, vet, test and build check what merging the pull request would produce, so they + # run on the default checkout (for a pull_request event that is the synthetic merge + # commit). Only the regenerate step below needs the head branch, because that is where + # its commit goes. - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4 with: - # A pull_request checkout is the synthetic merge commit (detached), and a fork - # PR branch only exists in the fork. - repository: ${{ github.event.pull_request.head.repo.full_name || github.repository }} - ref: ${{ github.head_ref || github.ref_name }} persist-credentials: false - uses: actions/setup-go@40f1582b2485089dde7abd97c1529aa768e1baff # v5 with: go-version-file: xbot_service_interface_go/go.mod cache-dependency-path: xbot_service_interface_go/go.sum + - name: Format + run: | + unformatted="$(gofmt -l .)" + if [ -n "$unformatted" ]; then + echo "gofmt needed for:" + echo "$unformatted" + exit 1 + fi + - name: Vet + run: go vet ./... + - name: Test + run: go test ./... -race -count=1 + - name: Build examples + run: go build ./... + + # The merge commit is detached and only exists in this repository, so the fix has to be + # worked on in the head branch's own worktree. !cancelled() on purpose: a red Format or + # Test above is exactly what a stale generated file looks like, and the commit below is + # the fix - the bot push then triggers a run that sees the corrected branch. + - name: Checkout the branch to regenerate on + if: ${{ !cancelled() }} + uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4 + with: + # A fork PR branch only exists in the fork. + repository: ${{ github.event.pull_request.head.repo.full_name || github.repository }} + ref: ${{ github.head_ref || github.ref_name }} + persist-credentials: false + - name: Regenerate the Go sources id: regenerate + if: ${{ !cancelled() }} env: BRANCH: ${{ github.head_ref || github.ref_name }} # Only a branch we may push to: not the default branch, and not a fork. @@ -93,18 +122,3 @@ jobs: exit 1 fi echo "::notice::Committed the regenerated Go sources to ${BRANCH}." - - - name: Format - run: | - unformatted="$(gofmt -l .)" - if [ -n "$unformatted" ]; then - echo "gofmt needed for:" - echo "$unformatted" - exit 1 - fi - - name: Vet - run: go vet ./... - - name: Test - run: go test ./... -race -count=1 - - name: Build examples - run: go build ./... diff --git a/Readme.md b/Readme.md index c57fd5c..7413e42 100644 --- a/Readme.md +++ b/Readme.md @@ -285,8 +285,11 @@ Two ways to use a service: - **Schema provided:** pass a `service.json` file (or dict) to `ServiceInterface`. The schema is validated against the service's advertisement on connect, and an `IncompatibleServiceError` aborts the connection on a mismatch. -- **Schema-free:** omit the schema entirely. The advertisement carries the interface (inputs, outputs, registers, RPC functions) and as much of the rest of the schema as fits into one packet. - A service whose schema is bigger, lists what did not fit under `omitted`; the client fetches those blocks with `DESCRIPTION_REQUEST`/`DESCRIPTION_PART` once after discovery (one round trip, unicast) and caches them per `(service id, version)`. +- **Schema-free:** omit the schema entirely. The advertisement carries as much of the schema as fits + into one UDP packet; a service whose schema is bigger lists the blocks it had to leave out under + `omitted` - any of `enums`, `functions`, `registers`, `outputs`, `inputs`, whatever did not fit. + The client fetches those blocks with `DESCRIPTION_REQUEST`/`DESCRIPTION_PART` right after discovery + (unicast, one round trip per window) and caches them per `(service id, version)`. Hosts with generated classes never fetch: they know the types. ```python diff --git a/codegen/xbot_codegen/xbot_codegen.py b/codegen/xbot_codegen/xbot_codegen.py index 53746b5..179e143 100644 --- a/codegen/xbot_codegen/xbot_codegen.py +++ b/codegen/xbot_codegen/xbot_codegen.py @@ -99,6 +99,11 @@ def common_attrs(json, valid_types, callback_name, send_method_name): # `version` always stay: they identify the service. DROPPABLE_BLOCKS = ("enums", "functions", "registers", "outputs", "inputs") +# The omitted description is handed out in DESCRIPTION_PART windows whose total size field +# is 16 bits, so more than this cannot be addressed and the split has to fail loudly here +# instead of silently truncating on the wire. +MAX_OMITTED_CBOR_SIZE = 0xFFFF + def splitDescription(json_service: dict): """Split the schema into what the advertisement carries and what it leaves out. @@ -120,10 +125,16 @@ def splitDescription(json_service: dict): advertised_cbor = cbor2.dumps(advertised) if len(advertised_cbor) <= ADVERTISED_DESCRIPTION_BUDGET: omitted = {key: json_service[key] for key in omitted_names} - return advertised_cbor, cbor2.dumps(omitted), omitted_names + omitted_cbor = cbor2.dumps(omitted) + if len(omitted_cbor) > MAX_OMITTED_CBOR_SIZE: + raise Exception( + f"{json_service['type']}: the omitted description is {len(omitted_cbor)} bytes, " + f"more than the 16-bit DESCRIPTION_PART total limit of {MAX_OMITTED_CBOR_SIZE}") + return advertised_cbor, omitted_cbor, omitted_names + kept = [key for key in json_service if key not in DROPPABLE_BLOCKS] raise Exception( - f"{json_service['type']}: type/version alone do not fit into the advertisement " - f"budget of {ADVERTISED_DESCRIPTION_BUDGET} bytes") + f"{json_service['type']}: the blocks that cannot be omitted ({', '.join(kept)}) do not " + f"fit into the advertisement budget of {ADVERTISED_DESCRIPTION_BUDGET} bytes") def loadService(path: str) -> dict: diff --git a/include/xbot/datatypes/DescriptionRequestPayload.hpp b/include/xbot/datatypes/DescriptionRequestPayload.hpp index d83cc97..879f22d 100644 --- a/include/xbot/datatypes/DescriptionRequestPayload.hpp +++ b/include/xbot/datatypes/DescriptionRequestPayload.hpp @@ -10,15 +10,20 @@ namespace xbot::datatypes { /** - * @brief Where the service should send the DESCRIPTION_PART to. + * @brief Where the service should send the DESCRIPTION_PART to, and where in the omitted + * description this window starts. * * The portable packet API only hands out the payload, not the sender's address, so the * host brings its own endpoint along - the same idea as ClaimPayload. + * + * The offset lives here instead of in the header: `arg1` is one byte, and the omitted + * description may be up to 64 KiB (the header's 16-bit total). */ struct DescriptionRequestPayload { uint16_t service_id{}; uint32_t target_ip{}; uint16_t target_port{}; + uint16_t offset{}; } __attribute__((packed)); } // namespace xbot::datatypes diff --git a/include/xbot/datatypes/XbotHeader.hpp b/include/xbot/datatypes/XbotHeader.hpp index ae87c2a..a0101a9 100644 --- a/include/xbot/datatypes/XbotHeader.hpp +++ b/include/xbot/datatypes/XbotHeader.hpp @@ -44,12 +44,14 @@ enum class MessageType : uint8_t { // could not carry: the advertisement is one packet (config::max_packet_size), so a // growing schema does not fit into it forever. It is sent unicast by a host that needs // the type definitions - a host with generated classes never asks. - // arg1 = byte offset into the omitted description, arg2 = maximum answer payload. - // Payload = DescriptionRequestPayload (where to send the answer). + // arg1 = 0 (reserved), arg2 = maximum answer payload. + // Payload = DescriptionRequestPayload (destination + byte offset). The offset is not in + // the header on purpose: arg1 is one byte and the omitted description may be up to 64 KiB. DESCRIPTION_REQUEST = 0x82, // DESCRIPTION_PART answers a DESCRIPTION_REQUEST with one window of the omitted - // description. arg1 = byte offset, arg2 = total size of the omitted description. - // Payload = the window bytes; the host repeats until offset + payload >= total. + // description. arg1 = 0 (reserved), arg2 = total size of the omitted description. + // Payload = uint16 byte offset of this window, then the window bytes; the host repeats + // until offset + window >= total. DESCRIPTION_PART = 0x83 }; diff --git a/libxbot-service/include/xbot-service/Service.hpp b/libxbot-service/include/xbot-service/Service.hpp index 8921998..10e3c27 100644 --- a/libxbot-service/include/xbot-service/Service.hpp +++ b/libxbot-service/include/xbot-service/Service.hpp @@ -184,6 +184,9 @@ class Service : public ServiceIo { } private: + /** @brief getTimeMicros() of the last DESCRIPTION_PART, for the reply rate limit. */ + uint32_t last_description_part_micros_{0}; + void HandleClaimMessage(datatypes::XbotHeader *header, const void *payload, size_t payload_len); void HandleDataMessage(datatypes::XbotHeader *header, const void *payload, size_t payload_len); void HandleDataTransaction(datatypes::XbotHeader *header, const void *payload, size_t payload_len); diff --git a/libxbot-service/src/Service.cpp b/libxbot-service/src/Service.cpp index 0580552..f0236c8 100644 --- a/libxbot-service/src/Service.cpp +++ b/libxbot-service/src/Service.cpp @@ -314,6 +314,14 @@ static uint32_t CalculateHeartbeatInterval(uint32_t heartbeat_micros) { return heartbeat_micros / 2; } +namespace { +/* One DESCRIPTION_PART per request, but a request is cheap to forge and the reply + destination comes from its payload (the sender is not exposed by the portable packet + API), so without a limit the service works as a reflector. 20 ms keeps a 64 KiB + description - about 45 windows - fetchable in well under a second. */ +constexpr uint32_t kDescriptionPartIntervalMicros = 20000U; +} // namespace + void xbot::service::Service::HandleDescriptionRequest(xbot::datatypes::XbotHeader *header, const void *payload, size_t payload_len) { if (payload_len != sizeof(datatypes::DescriptionRequestPayload)) { @@ -333,37 +341,56 @@ void xbot::service::Service::HandleDescriptionRequest(xbot::datatypes::XbotHeade return; } - const size_t offset = header->arg1; + /* The offset comes from the payload, not from arg1: that is one byte, and the omitted + description may be up to 64 KiB (offset 1448 and beyond are normal). */ + const size_t offset = request->offset; if (offset >= total) { ULOG_ARG_WARNING(&service_id_, "description request beyond the omitted description (%u >= %u)", static_cast(offset), static_cast(total)); return; } + /* A reply is a reflection primitive: the destination comes from the request payload (the + portable packet API does not expose the sender, so it cannot be verified - the same + trade-off ClaimPayload makes). Keep the amplification bounded with a minimum interval + between windows; a host fetching a 64 KiB description needs ~45 windows, which stays + below a second. */ + const uint32_t now = system::getTimeMicros(); + if (last_description_part_micros_ != 0U && + static_cast(now - last_description_part_micros_) < kDescriptionPartIntervalMicros) { + ULOG_ARG_DEBUG(&service_id_, "description request dropped (reply rate limit)"); + return; + } + size_t length = total - offset; if (length > header->arg2) { length = header->arg2; } + /* The payload carries the window offset in front of the bytes. */ constexpr size_t kMaxPayload = config::max_packet_size - sizeof(datatypes::XbotHeader); - if (length > kMaxPayload) { - length = kMaxPayload; + constexpr size_t kWindowOffsetSize = sizeof(uint16_t); + if (length > kMaxPayload - kWindowOffsetSize) { + length = kMaxPayload - kWindowOffsetSize; } if (length == 0U) { return; } + const uint16_t window_offset = static_cast(offset); packet::PacketPtr ptr = packet::allocatePacket(); { Lock lk(&state_mutex_); fillHeader(); header_.message_type = datatypes::MessageType::DESCRIPTION_PART; - header_.arg1 = static_cast(offset); + header_.arg1 = 0; header_.arg2 = static_cast(total); - header_.payload_size = static_cast(length); + header_.payload_size = static_cast(kWindowOffsetSize + length); packet::packetAppendData(ptr, &header_, sizeof(header_)); } + packet::packetAppendData(ptr, &window_offset, sizeof(window_offset)); packet::packetAppendData(ptr, omitted + offset, length); Io::transmitPacket(ptr, request->target_ip, request->target_port); + last_description_part_micros_ = now; } void xbot::service::Service::HandleClaimMessage(xbot::datatypes::XbotHeader *header, const void *payload, diff --git a/xbot_service_interface_go/cmd/xbot-codegen/messagetypes.go b/xbot_service_interface_go/cmd/xbot-codegen/messagetypes.go index 11dffe6..8a05440 100644 --- a/xbot_service_interface_go/cmd/xbot-codegen/messagetypes.go +++ b/xbot_service_interface_go/cmd/xbot-codegen/messagetypes.go @@ -16,7 +16,7 @@ import ( var ( enumBlock = regexp.MustCompile(`(?s)enum class MessageType[^{]*\{(.*?)\}`) - enumEntry = regexp.MustCompile(`(\w+)\s*=\s*(0x[0-9A-Fa-f]+)`) + enumEntry = regexp.MustCompile(`(?m)^\s*(\w+)\s*=\s*(0x[0-9A-Fa-f]+)`) ) // initialisms stay upper case in Go names (MsgRPCCall, not MsgRpcCall). diff --git a/xbot_service_interface_go/xbot/spec_test.go b/xbot_service_interface_go/xbot/spec_test.go index 7de841f..815d465 100644 --- a/xbot_service_interface_go/xbot/spec_test.go +++ b/xbot_service_interface_go/xbot/spec_test.go @@ -109,7 +109,7 @@ func TestMessageTypesMatchHeaderHpp(t *testing.T) { if enumBody == nil { t.Fatal("enum class MessageType not found - the header format changed") } - entries := regexp.MustCompile(`(\w+)\s*=\s*(0x[0-9A-Fa-f]+)`).FindAllStringSubmatch(enumBody[1], -1) + entries := regexp.MustCompile(`(?m)^\s*(\w+)\s*=\s*(0x[0-9A-Fa-f]+)`).FindAllStringSubmatch(enumBody[1], -1) if len(entries) == 0 { t.Fatal("no enumerators parsed from XbotHeader.hpp") } diff --git a/xbot_service_interface_py/README.md b/xbot_service_interface_py/README.md index c1289ca..de1f5f0 100644 --- a/xbot_service_interface_py/README.md +++ b/xbot_service_interface_py/README.md @@ -89,12 +89,14 @@ def connected(): print([i['name'] for i in schema.inputs]) ``` -The advertisement carries the interface - inputs, outputs, registers and RPC functions - plus as much -of the rest of the schema as fits into one UDP packet. A service whose schema is bigger (a long -`SoundId` enum, for example) lists what did not fit under `omitted`; the interface fetches those -blocks right after discovery (`DESCRIPTION_REQUEST` -> `DESCRIPTION_PART`, unicast, one round trip per -window), merges them and caches the result per `(service id, version)`, so a repeating -advertisement never fetches again. +The advertisement carries as much of the schema as fits into one UDP packet; if the schema is bigger +- a long `SoundId` enum, for example - the blocks that had to be left out are listed under `omitted`. +Any block can be omitted that way, `inputs`, `outputs`, `registers` and `functions` included; only +`type` and `version` are always advertised. The interface fetches the omitted blocks right after +discovery (`DESCRIPTION_REQUEST` -> `DESCRIPTION_PART`, unicast, one round trip per window), merges +them into the schema and caches the result per `(type, service id, version)`, so a repeating +advertisement never fetches again. Blocks the advertisement did carry are never fetched either - the +fetch only fills in the gaps. A schema passed in Mode 1 is validated once the description is complete, i.e. after those blocks arrived. If the fetch fails - an older service that advertises everything, or a lost packet - the diff --git a/xbot_service_interface_py/tests/test_description_fetch.py b/xbot_service_interface_py/tests/test_description_fetch.py index f8c1928..b56eea0 100644 --- a/xbot_service_interface_py/tests/test_description_fetch.py +++ b/xbot_service_interface_py/tests/test_description_fetch.py @@ -10,7 +10,8 @@ import cbor2 from xbot_service_interface.datatypes import ( - HEADER_SIZE, MAX_PACKET_SIZE, MessageType, pack_header, unpack_header, + DESCRIPTION_PART_OFFSET_FORMAT, DESCRIPTION_REQUEST_FORMAT, HEADER_SIZE, MAX_PACKET_SIZE, + MessageType, pack_header, unpack_header, ) from xbot_service_interface.discovery import ServiceDiscovery @@ -42,18 +43,19 @@ def _serve(self) -> None: hdr = unpack_header(data) if hdr['message_type'] != MessageType.DESCRIPTION_REQUEST: continue - sid, ip, port = struct.unpack(' bytes: - desc = {'type': 'SoundService', 'version': 1, + def advertisement(self, omitted: list[str] | None = None, version: int = 1) -> bytes: + desc = {'type': 'SoundService', 'version': version, 'inputs': [{'id': 0, 'name': 'Volume', 'type': 'uint8_t'}], 'functions': []} desc['omitted'] = ['enums'] if omitted is None else omitted @@ -103,11 +105,15 @@ def test_fetch_is_cached_per_version(): discovery._handle_packet(service.advertisement()) requests_after_first = len(service.requests) discovery._handle_packet(service.advertisement()) + requests_after_repeat = len(service.requests) + discovery._handle_packet(service.advertisement(version=2)) + requests_after_version = len(service.requests) finally: service.close() - assert len(collector.found) == 1, "the second advertisement changes nothing" - assert len(service.requests) == requests_after_first, "a repeating advertisement must not fetch again" + assert len(collector.found) == 2, "a changed version is a new schema" + assert requests_after_repeat == requests_after_first, "a repeating advertisement must not fetch again" + assert requests_after_version > requests_after_repeat, "a new version must be fetched" def test_failed_fetch_keeps_the_advertisement_only(): @@ -205,8 +211,10 @@ def test_description_request_uses_the_claim_ip_convention(): from xbot_service_interface.datatypes import pack_claim_payload, pack_description_request ip_host = struct.unpack('!I', socket.inet_aton('172.16.78.1'))[0] - request = pack_description_request(13, ip_host, 4711) + request = pack_description_request(13, ip_host, 4711, 1448) claim = pack_claim_payload(ip_host, 4711, 1_000_000) assert request[2:6] == claim[0:4], "target_ip must be packed like the claim payload" assert request[6:8] == claim[4:6], "target_port must be packed like the claim payload" + # The offset is in the payload because the header's arg1 is one byte; 1448 is beyond it. + assert struct.unpack_from('= total. + # Answer to DESCRIPTION_REQUEST: arg1 = 0 (reserved), arg2 = total size, payload = a + # uint16 window offset followed by the window bytes. + # Repeat until offset + window >= total. DESCRIPTION_PART = 0x83 @@ -118,17 +120,23 @@ def pack_descriptor(target_id: int, payload_size: int) -> bytes: return struct.pack(DESCRIPTOR_FORMAT, target_id, 0, payload_size) -DESCRIPTION_REQUEST_FORMAT = ' bytes: - """Where the service should send the DESCRIPTION_PART to. + target_port: int, offset: int) -> bytes: + """Where the service should send the DESCRIPTION_PART to, and where the window starts. The portable packet API only hands out the payload, not the sender's address, so the host brings its own endpoint along - the same idea as the claim payload. + + The offset is part of the payload on purpose: the header's ``arg1`` is one byte and + the omitted description may be up to 64 KiB, so a window beyond offset 255 could not + be requested otherwise. """ - return struct.pack(DESCRIPTION_REQUEST_FORMAT, service_id, target_ip_host, target_port) + return struct.pack(DESCRIPTION_REQUEST_FORMAT, service_id, target_ip_host, target_port, offset) def pack_claim_payload(target_ip_host: int, target_port: int, diff --git a/xbot_service_interface_py/xbot_service_interface/discovery.py b/xbot_service_interface_py/xbot_service_interface/discovery.py index 9a0f936..ea2682f 100644 --- a/xbot_service_interface_py/xbot_service_interface/discovery.py +++ b/xbot_service_interface_py/xbot_service_interface/discovery.py @@ -3,12 +3,14 @@ import struct import threading import logging +import time import cbor2 from .datatypes import ( HEADER_SIZE, MULTICAST_PORT, SD_MULTICAST_ADDR, MAX_PACKET_SIZE, MessageType, unpack_header, pack_header, + DESCRIPTION_PART_OFFSET_FORMAT, DESCRIPTION_PART_OFFSET_SIZE, pack_description_request, ) from .schema import ServiceSchema @@ -77,21 +79,26 @@ class ServiceDiscovery: def on_service_found(service_id, ip, port, schema: ServiceSchema): ... """ - #: How long to wait for the DESCRIPTION_PART answers of one window. + #: Deadline for one complete fetch (all windows together, not per window). FETCH_TIMEOUT_S = 1 #: Upper bound on the windows one fetch may take (guards a misbehaving service). FETCH_MAX_WINDOWS = 16 + #: After a failed fetch, do not try again for this long. The fetch runs on the discovery + #: thread, so an unreachable service must not block it once per advertisement. + FETCH_RETRY_S = 30 + def __init__(self, bind_ip: str = '0.0.0.0', fetch_omitted: bool = True): self._bind_ip = bind_ip #: A service advertises only as much of its schema as fits into one packet and #: lists the rest under "omitted". Set False to keep the schema incomplete - #: clients that have generated classes do not need the blocks at all. self._fetch_omitted = fetch_omitted - #: service_id -> {"version": int, "blocks": dict} so a repeating advertisement - #: does not fetch again. - self._omitted_cache: dict[int, dict] = {} + #: (type, service_id, version) -> {"blocks": dict | None, "retry_at": float}. Keyed + #: on the type as well: a different service reusing the id and version must not + #: inherit blocks. A None entry is a cached failure with a retry deadline. + self._omitted_cache: dict[tuple[str, int, int], dict] = {} self._sock: socket.socket | None = None self._thread: threading.Thread | None = None self._running = False @@ -168,12 +175,34 @@ def stop(self) -> None: # Omitted schema blocks (see DESCRIPTION_REQUEST in libxbot-service) # ------------------------------------------------------------------ - def _omitted_blocks(self, service_id: int, ip: str, port: int, version: int) -> dict | None: - """Fetch the schema blocks the advertisement left out; None when it did not work.""" - cached = self._omitted_cache.get(service_id) - if cached is not None and cached['version'] == version: - return cached['blocks'] + def _omitted_blocks(self, service_id: int, service_type: str, ip: str, port: int, + version: int) -> dict | None: + """The blocks the advertisement left out; None when they are not available. + + Cached per (type, id, version), and a *failed* fetch is cached too: the fetch runs + on the discovery thread, so an unreachable service must not block it once per + advertisement. + """ + key = (service_type, service_id, version) + cached = self._omitted_cache.get(key) + if cached is not None: + if cached['blocks'] is not None: + return cached['blocks'] + if time.monotonic() < cached['retry_at']: + return None + + blocks = self._fetch_blocks(service_id, ip, port) + if blocks is None: + self._omitted_cache[key] = {'blocks': None, + 'retry_at': time.monotonic() + self.FETCH_RETRY_S} + log.warning(f"Service {service_id}: could not fetch the omitted schema blocks; " + f"using the advertisement only (retry in {self.FETCH_RETRY_S:.0f} s)") + else: + self._omitted_cache[key] = {'blocks': blocks} + return blocks + def _fetch_blocks(self, service_id: int, ip: str, port: int) -> dict | None: + """Request the omitted description window by window; None when that did not work.""" target_ip = self._bind_ip if target_ip == '0.0.0.0': # The service has to send the answer somewhere: hand it the address this host @@ -183,42 +212,58 @@ def _omitted_blocks(self, service_id: int, ip: str, port: int, version: int) -> data = bytearray() total = None + sock = None try: sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP) sock.bind(('', 0)) - sock.settimeout(self.FETCH_TIMEOUT_S) my_port = sock.getsockname()[1] # Same convention as the claim payload: the wire carries the host-order # integer of the address (io._send_claim uses struct.unpack('!I', ...)). target_ip_host = struct.unpack('!I', socket.inet_aton(target_ip))[0] - request = pack_description_request(service_id, target_ip_host, my_port) + # One deadline for the whole fetch: a stream of foreign packets must not be + # able to stretch the wait window by window. + deadline = time.monotonic() + self.FETCH_TIMEOUT_S for _ in range(self.FETCH_MAX_WINDOWS): if total is not None and len(data) >= total: break offset = len(data) - header = pack_header(MessageType.DESCRIPTION_REQUEST, service_id, offset, + request = pack_description_request(service_id, target_ip_host, my_port, offset) + header = pack_header(MessageType.DESCRIPTION_REQUEST, service_id, 0, MAX_PACKET_SIZE - HEADER_SIZE, 0, 0, len(request)) sock.sendto(header + request, (ip, port)) # One window per request: wait for the answer that starts at len(data). - # Everything else (a stale window, a foreign packet) is skipped. + # Everything else - a stale window, a packet from another host or another + # service - is skipped, and skipping does not extend the deadline. while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + return None + sock.settimeout(remaining) try: - answer, _ = sock.recvfrom(MAX_PACKET_SIZE) + answer, addr = sock.recvfrom(MAX_PACKET_SIZE) except socket.timeout: return None + if (addr[0], addr[1]) != (ip, port): + continue if len(answer) < HEADER_SIZE: continue hdr = unpack_header(answer) if hdr['message_type'] != MessageType.DESCRIPTION_PART: continue + if hdr['service_id'] != service_id: + continue if len(answer) != HEADER_SIZE + hdr['payload_size']: continue - if hdr['arg1'] != len(data): + if hdr['payload_size'] <= DESCRIPTION_PART_OFFSET_SIZE: + continue + part_offset = struct.unpack_from(DESCRIPTION_PART_OFFSET_FORMAT, answer, + HEADER_SIZE)[0] + if part_offset != len(data): continue total = hdr['arg2'] - data += answer[HEADER_SIZE:] + data += answer[HEADER_SIZE + DESCRIPTION_PART_OFFSET_SIZE:] break if total == 0: return {} @@ -226,10 +271,11 @@ def _omitted_blocks(self, service_id: int, ip: str, port: int, version: int) -> log.warning(f"Service {service_id}: description fetch failed: {e}") return None finally: - try: - sock.close() - except OSError: - pass + if sock is not None: + try: + sock.close() + except OSError: + pass if total is None or len(data) != total: return None @@ -241,7 +287,6 @@ def _omitted_blocks(self, service_id: int, ip: str, port: int, version: int) -> if not isinstance(blocks, dict): return None - self._omitted_cache[service_id] = {'version': version, 'blocks': blocks} log.debug(f"Service {service_id}: fetched omitted schema blocks {list(blocks)}") return blocks @@ -298,13 +343,14 @@ def _handle_packet(self, data: bytes) -> None: return # The advertisement is one packet: whatever did not fit is fetched on demand. + # _omitted_blocks() caches failures and warns once, so this stays cheap and quiet + # for a service that keeps advertising the same incomplete schema. omitted = desc.get('omitted') or [] if omitted and self._fetch_omitted: - details = self._omitted_blocks(service_id, ip, port, desc.get('version', 0)) + details = self._omitted_blocks(service_id, desc.get('type', ''), ip, port, + desc.get('version', 0)) if details is None: - log.warning( - f"Service {service_id}: could not fetch the omitted schema blocks {omitted};" - " using the advertisement only") + log.debug(f"Service {service_id}: description stays incomplete ({omitted})") else: del desc['omitted'] desc.update(details) @@ -320,9 +366,15 @@ def _handle_packet(self, data: bytes) -> None: with self._lock: existing = self._services.get(service_id) if existing: - if existing['ip'] == ip and existing['port'] == port: + if (existing['ip'] == ip and existing['port'] == port + and existing['schema'].raw == schema.raw): return # Nothing changed - log.info(f"Service {service_id} endpoint changed to {ip}:{port}") + if existing['ip'] != ip or existing['port'] != port: + log.info(f"Service {service_id} endpoint changed to {ip}:{port}") + else: + # Same endpoint, different schema: a fetch that completed late, or a + # service that changed without bumping its version. + log.info(f"Service {service_id} description changed") else: log.info(f"New service: {schema.type!r} id={service_id} at {ip}:{port}")