diff --git a/.github/workflows/go.yaml b/.github/workflows/go.yaml index 20eb364..52f3c88 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,13 +15,26 @@ 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: + # 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: persist-credentials: false @@ -22,6 +42,7 @@ jobs: 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 .)" @@ -36,3 +57,68 @@ jobs: 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. + 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}." diff --git a/Readme.md b/Readme.md index 2d8927e..7413e42 100644 --- a/Readme.md +++ b/Readme.md @@ -80,33 +80,41 @@ 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 | — | — | — | +| `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 @@ -115,7 +123,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 +134,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 +146,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 +187,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 +285,12 @@ 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 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 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.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]]] 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..179e143 100644 --- a/codegen/xbot_codegen/xbot_codegen.py +++ b/codegen/xbot_codegen/xbot_codegen.py @@ -87,11 +87,63 @@ 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") + +# 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. + + 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} + 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']}: 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: # 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 +151,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..879f22d --- /dev/null +++ b/include/xbot/datatypes/DescriptionRequestPayload.hpp @@ -0,0 +1,31 @@ +// +// 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, 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 + +#endif // DESCRIPTION_REQUEST_PAYLOAD_HPP diff --git a/include/xbot/datatypes/XbotHeader.hpp b/include/xbot/datatypes/XbotHeader.hpp index 3f3d02e..a0101a9 100644 --- a/include/xbot/datatypes/XbotHeader.hpp +++ b/include/xbot/datatypes/XbotHeader.hpp @@ -39,7 +39,20 @@ 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 = 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 = 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 }; #pragma pack(push, 1) diff --git a/libxbot-service/include/xbot-service/Service.hpp b/libxbot-service/include/xbot-service/Service.hpp index bf67d55..10e3c27 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,30 @@ class Service : public ServiceIo { protected: size_t rpc_max_response_size_{0}; + /** + * @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: + /** @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); 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..f0236c8 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,85 @@ 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)) { + 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; + } + + /* 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); + 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 = 0; + header_.arg2 = static_cast(total); + 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, size_t payload_len) { (void)header; 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..8a05440 --- /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(`(?m)^\s*(\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..815d465 100644 --- a/xbot_service_interface_go/xbot/spec_test.go +++ b/xbot_service_interface_go/xbot/spec_test.go @@ -109,23 +109,16 @@ 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") } - 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 { diff --git a/xbot_service_interface_py/README.md b/xbot_service_interface_py/README.md index a4ab2c4..de1f5f0 100644 --- a/xbot_service_interface_py/README.md +++ b/xbot_service_interface_py/README.md @@ -89,7 +89,19 @@ 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 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 +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/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" 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..b56eea0 --- /dev/null +++ b/xbot_service_interface_py/tests/test_description_fetch.py @@ -0,0 +1,220 @@ +"""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 ( + DESCRIPTION_PART_OFFSET_FORMAT, DESCRIPTION_REQUEST_FORMAT, 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, offset = struct.unpack(DESCRIPTION_REQUEST_FORMAT, data[HEADER_SIZE:]) + if sid != SERVICE_ID: + continue + self.requests.append((offset, hdr['arg2'])) + chunk = self.blob[offset:offset + self.window] + payload = struct.pack(DESCRIPTION_PART_OFFSET_FORMAT, offset) + chunk + answer = pack_header(MessageType.DESCRIPTION_PART, SERVICE_ID, 0, len(self.blob), + 0, 0, len(payload)) + payload + # target_ip is the host-order integer of the address (ClaimPayload convention). + self.sock.sendto(answer, (socket.inet_ntoa(struct.pack('!I', ip)), port)) + + 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 + 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()) + 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) == 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(): + 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 == [] + + +# --------------------------------------------------------------------------------- +# 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, 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. + DESCRIPTION_PART = 0x83 class RpcStatus(IntEnum): @@ -112,6 +120,25 @@ 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, 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, offset) + + 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..ea2682f 100644 --- a/xbot_service_interface_py/xbot_service_interface/discovery.py +++ b/xbot_service_interface_py/xbot_service_interface/discovery.py @@ -1,19 +1,76 @@ +import fcntl import socket 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, + MAX_PACKET_SIZE, MessageType, unpack_header, pack_header, + DESCRIPTION_PART_OFFSET_FORMAT, DESCRIPTION_PART_OFFSET_SIZE, + 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 +79,26 @@ class ServiceDiscovery: def on_service_found(service_id, ip, port, schema: ServiceSchema): ... """ - def __init__(self, bind_ip: str = '0.0.0.0'): + #: 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 + #: (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 @@ -96,6 +171,125 @@ 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, 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 + # reaches the service from (the request payload carries ip:port, see + # DescriptionRequestPayload). + target_ip = _route_source_ip(ip) + + data = bytearray() + total = None + sock = None + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP) + sock.bind(('', 0)) + 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] + + # 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) + 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 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, 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['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 + DESCRIPTION_PART_OFFSET_SIZE:] + break + if total == 0: + return {} + except OSError as e: + log.warning(f"Service {service_id}: description fetch failed: {e}") + return None + finally: + if sock is not None: + 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 + + log.debug(f"Service {service_id}: fetched omitted schema blocks {list(blocks)}") + return blocks + # ------------------------------------------------------------------ # Internal # ------------------------------------------------------------------ @@ -137,7 +331,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 +342,20 @@ 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_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, desc.get('type', ''), ip, port, + desc.get('version', 0)) + if details is None: + log.debug(f"Service {service_id}: description stays incomplete ({omitted})") + else: + del desc['omitted'] + desc.update(details) + elif 'omitted' in desc: + del desc['omitted'] try: schema = ServiceSchema.from_dict(desc) except Exception as e: @@ -158,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}")