pw_rpc2: Packet framing and encoding/decoding Introduce core packet framing and wire-protocol definitions for pw_rpc2, including InboundPacket, OutboundPacket, and HandshakePacket. This defines the fundamental wire-level header structures and handles encoding and zero-copy decoding/slicing for RPC packets. Change-Id: I1d84d93f6e4ed003b94fd8387cb0e08e3696a382 Reviewed-on: https://pigweed-review.googlesource.com/c/pigweed/pigweed/+/480285
diff --git a/CMakeLists.txt b/CMakeLists.txt index 04dbf64..9562022 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt
@@ -164,6 +164,7 @@ add_subdirectory(pw_ring_buffer EXCLUDE_FROM_ALL) add_subdirectory(pw_router EXCLUDE_FROM_ALL) add_subdirectory(pw_rpc EXCLUDE_FROM_ALL) +add_subdirectory(pw_rpc2 EXCLUDE_FROM_ALL) add_subdirectory(pw_rpc_transport EXCLUDE_FROM_ALL) add_subdirectory(pw_sensor EXCLUDE_FROM_ALL) add_subdirectory(pw_snapshot EXCLUDE_FROM_ALL)
diff --git a/PIGWEED_MODULES b/PIGWEED_MODULES index 0d5eba1..fac02fb 100644 --- a/PIGWEED_MODULES +++ b/PIGWEED_MODULES
@@ -128,6 +128,7 @@ pw_ring_buffer pw_router pw_rpc +pw_rpc2 pw_rpc_transport pw_rust pw_sensor
diff --git a/docs/sphinx/BUILD.bazel b/docs/sphinx/BUILD.bazel index 8ee3aa8..ee585a0 100644 --- a/docs/sphinx/BUILD.bazel +++ b/docs/sphinx/BUILD.bazel
@@ -333,6 +333,7 @@ "//pw_ring_buffer:docs", "//pw_router:docs", "//pw_rpc:docs", + "//pw_rpc2:docs", "//pw_rpc_transport:docs", "//pw_rust:sphinx", "//pw_sensor:docs",
diff --git a/docs/sphinx/module_metadata.json b/docs/sphinx/module_metadata.json index 9a14960..4832f4e 100644 --- a/docs/sphinx/module_metadata.json +++ b/docs/sphinx/module_metadata.json
@@ -745,6 +745,13 @@ "TypeScript" ] }, + "pw_rpc2": { + "tagline": "Next gen, zero-copy async RPC with end-to-end backpressure", + "status": "experimental", + "languages": [ + "C++" + ] + }, "pw_rpc_transport": { "status": "unstable" },
diff --git a/docs/sphinx/modules.rst b/docs/sphinx/modules.rst index 462e0b3..d555436 100644 --- a/docs/sphinx/modules.rst +++ b/docs/sphinx/modules.rst
@@ -102,6 +102,7 @@ pw_ring_buffer/docs pw_router/docs pw_rpc/docs + pw_rpc2/docs pw_rpc_transport/docs pw_rust/docs pw_sensor/docs
diff --git a/pw_build/generated_pigweed_modules_lists.gni b/pw_build/generated_pigweed_modules_lists.gni index 71678c1..0a84f96 100644 --- a/pw_build/generated_pigweed_modules_lists.gni +++ b/pw_build/generated_pigweed_modules_lists.gni
@@ -166,6 +166,7 @@ dir_pw_ring_buffer = get_path_info("../pw_ring_buffer", "abspath") dir_pw_router = get_path_info("../pw_router", "abspath") dir_pw_rpc = get_path_info("../pw_rpc", "abspath") + dir_pw_rpc2 = get_path_info("../pw_rpc2", "abspath") dir_pw_rpc_transport = get_path_info("../pw_rpc_transport", "abspath") dir_pw_rust = get_path_info("../pw_rust", "abspath") dir_pw_sensor = get_path_info("../pw_sensor", "abspath") @@ -371,6 +372,7 @@ dir_pw_ring_buffer, dir_pw_router, dir_pw_rpc, + dir_pw_rpc2, dir_pw_rpc_transport, dir_pw_rust, dir_pw_sensor, @@ -567,6 +569,7 @@ "$dir_pw_ring_buffer:tests", "$dir_pw_router:tests", "$dir_pw_rpc:tests", + "$dir_pw_rpc2:tests", "$dir_pw_rpc_transport:tests", "$dir_pw_rust:tests", "$dir_pw_sensor:tests",
diff --git a/pw_rpc2/BUILD.bazel b/pw_rpc2/BUILD.bazel new file mode 100644 index 0000000..60b240b --- /dev/null +++ b/pw_rpc2/BUILD.bazel
@@ -0,0 +1,60 @@ +# Copyright 2026 The Pigweed Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may not +# use this file except in compliance with the License. You may obtain a copy of +# the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations under +# the License. + +load("@rules_cc//cc:cc_library.bzl", "cc_library") +load("@sphinxdocs//sphinxdocs:sphinx_docs_library.bzl", "sphinx_docs_library") +load("//pw_build:compatibility.bzl", "incompatible_with_mcu") +load("//pw_unit_test:pw_cc_test.bzl", "pw_cc_test") + +package(default_visibility = ["//visibility:public"]) + +licenses(["notice"]) + +cc_library( + name = "packet", + srcs = ["packet.cc"], + hdrs = ["public/pw_rpc2/internal/packet.h"], + strip_include_prefix = "public", + visibility = ["//pw_rpc2:__subpackages__"], + deps = [ + "//pw_assert:assert", + "//pw_assert:check", + "//pw_buf", + "//pw_bytes", + "//pw_preprocessor", + "//pw_result", + "//pw_status", + ], +) + +pw_cc_test( + name = "packet_test", + srcs = ["packet_test.cc"], + deps = [ + ":packet", + "//pw_allocator:testing", + "//pw_assert:check", + "//pw_bytes", + "//pw_unit_test", + ], +) + +sphinx_docs_library( + name = "docs", + srcs = [ + "docs.rst", + ], + prefix = "pw_rpc2/", + target_compatible_with = incompatible_with_mcu(), +)
diff --git a/pw_rpc2/BUILD.gn b/pw_rpc2/BUILD.gn new file mode 100644 index 0000000..42ccdb9 --- /dev/null +++ b/pw_rpc2/BUILD.gn
@@ -0,0 +1,53 @@ +# Copyright 2026 The Pigweed Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may not +# use this file except in compliance with the License. You may obtain a copy of +# the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations under +# the License. + +import("//build_overrides/pigweed.gni") + +import("$dir_pw_build/target_types.gni") +import("$dir_pw_unit_test/test.gni") + +config("public_include_path") { + include_dirs = [ "public" ] + visibility = [ ":*" ] +} + +pw_source_set("packet") { + public_configs = [ ":public_include_path" ] + public = [ "public/pw_rpc2/internal/packet.h" ] + sources = [ "packet.cc" ] + public_deps = [ + "$dir_pw_assert:assert", + "$dir_pw_assert:check", + dir_pw_buf, + dir_pw_bytes, + dir_pw_preprocessor, + dir_pw_result, + dir_pw_status, + ] + visibility = [ "$dir_pw_rpc2/*" ] +} + +pw_test("packet_test") { + deps = [ + ":packet", + "$dir_pw_allocator:testing", + "$dir_pw_assert:check", + dir_pw_bytes, + ] + sources = [ "packet_test.cc" ] +} + +pw_test_group("tests") { + tests = [ ":packet_test" ] +}
diff --git a/pw_rpc2/CMakeLists.txt b/pw_rpc2/CMakeLists.txt new file mode 100644 index 0000000..2ceaae6 --- /dev/null +++ b/pw_rpc2/CMakeLists.txt
@@ -0,0 +1,45 @@ +# Copyright 2026 The Pigweed Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may not +# use this file except in compliance with the License. You may obtain a copy of +# the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations under +# the License. + +include($ENV{PW_ROOT}/pw_build/pigweed.cmake) + +pw_add_library(pw_rpc2._packet STATIC + HEADERS + public/pw_rpc2/internal/packet.h + PUBLIC_INCLUDES + public + PUBLIC_DEPS + pw_assert.assert + pw_assert.check + pw_buf + pw_bytes + pw_preprocessor + pw_result + pw_status + SOURCES + packet.cc +) + +pw_add_test(pw_rpc2.packet_test + SOURCES + packet_test.cc + PRIVATE_DEPS + pw_allocator.testing + pw_assert.check + pw_bytes + pw_rpc2._packet + GROUPS + modules + pw_rpc2 +)
diff --git a/pw_rpc2/docs.rst b/pw_rpc2/docs.rst new file mode 100644 index 0000000..57c3754 --- /dev/null +++ b/pw_rpc2/docs.rst
@@ -0,0 +1,11 @@ +.. _module-pw_rpc2: + +======= +pw_rpc2 +======= +.. pigweed-module:: + :name: pw_rpc2 + +.. attention:: + + ``pw_rpc2`` is under construction and is not yet ready for use.
diff --git a/pw_rpc2/packet.cc b/pw_rpc2/packet.cc new file mode 100644 index 0000000..cc1fe2d --- /dev/null +++ b/pw_rpc2/packet.cc
@@ -0,0 +1,154 @@ +// Copyright 2026 The Pigweed Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); you may not +// use this file except in compliance with the License. You may obtain a copy of +// the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +// License for the specific language governing permissions and limitations under +// the License. + +#include "pw_rpc2/internal/packet.h" + +#include <cstddef> +#include <cstring> + +#include "pw_assert/check.h" +#include "pw_bytes/endian.h" +#include "pw_status/try.h" + +namespace pw::rpc2::internal { +namespace { + +void WriteUint32(ByteSpan buffer, size_t offset, uint32_t value) { + bytes::CopyInOrder<uint32_t>(endian::little, value, buffer.data() + offset); +} + +} // namespace + +Result<HandshakePacket> HandshakePacket::Decode(ConstByteSpan bytes) { + if (bytes.size() != sizeof(HandshakeWireFormat)) { + return Status::DataLoss(); + } + uint32_t magic = bytes::ReadInOrder<uint32_t>( + endian::little, bytes.data() + offsetof(HandshakeWireFormat, magic)); + uint8_t version = + static_cast<uint8_t>(bytes[offsetof(HandshakeWireFormat, version)]); + Type type = static_cast<Type>(bytes[offsetof(HandshakeWireFormat, type)]); + if (magic != kMagic || version != kVersion || + (type != Type::kSyn && type != Type::kSynAck && type != Type::kAck)) { + return Status::DataLoss(); + } + return HandshakePacket(static_cast<Type>(type)); +} + +Status HandshakePacket::Encode(ByteSpan buffer) const { + if (buffer.size() < sizeof(HandshakeWireFormat)) { + return Status::ResourceExhausted(); + } + std::memset(buffer.data(), 0, sizeof(HandshakeWireFormat)); + WriteUint32(buffer, offsetof(HandshakeWireFormat, magic), kMagic); + buffer[offsetof(HandshakeWireFormat, version)] = + static_cast<std::byte>(kVersion); + buffer[offsetof(HandshakeWireFormat, type)] = static_cast<std::byte>(type_); + return OkStatus(); +} + +Result<Buf> HandshakePacket::Encode(Buf buffer) const { + PW_TRY(Encode(ByteSpan(buffer))); + return buffer; +} + +// State PacketType value assumptions used in Decode to validate the type byte. +static_assert(PacketType::kRequest <= PacketType::kMessage); +static_assert(PacketType::kMessage <= PacketType::kStreamEnd); +static_assert(PacketType::kStreamEnd <= PacketType::kError); +static_assert(PacketType::kError <= PacketType::kResponse); + +Result<InboundPacket> InboundPacket::Decode(ConstBuf&& buffer) { + if (buffer.size() < sizeof(PacketHeader)) { + return Status::DataLoss(); // Too short for header + } + + const auto type = + static_cast<PacketType>(buffer[offsetof(PacketHeader, type)]); + if (type < PacketType::kRequest || type > PacketType::kResponse) { + return Status::InvalidArgument(); + } + if (buffer.size() < PacketSizeWithoutPayload(type)) { + return Status::DataLoss(); + } + + // The payload length is not encoded on the wire. The transport frames the + // packet, so the payload is exactly the bytes following the header. + return InboundPacket(std::move(buffer)); +} + +Result<size_t> OutboundPacket::EncodeHeader(ByteSpan buffer, + size_t payload_len) const { + const size_t offset = payload_offset(); + + // Checked this way around so that a large `payload_len` cannot overflow. + if (buffer.size() < offset || buffer.size() - offset < payload_len) { + return Status::ResourceExhausted(); + } + + WriteUint32(buffer, offsetof(PacketHeader, call_id), call_id_); + buffer[offsetof(PacketHeader, type)] = static_cast<std::byte>(type_); + + switch (type_) { + case PacketType::kRequest: { + WriteUint32(buffer, + offsetof(RequestWireFormat, service_id), + fields_.request.service_id); + WriteUint32(buffer, + offsetof(RequestWireFormat, method_id), + fields_.request.method_id); + break; + } + + case PacketType::kMessage: + case PacketType::kResponse: + case PacketType::kStreamEnd: { + // These packet types carry no header fields beyond the common header. + break; + } + + case PacketType::kError: { + WriteUint32( + buffer, offsetof(ErrorWireFormat, status_code), status().code()); + break; + } + } + + return offset + payload_len; +} + +Result<Buf> OutboundPacket::Encode(Buf buffer, size_t payload_len) const { + if (buffer.empty()) { + return Status::FailedPrecondition(); + } + size_t offset = payload_offset(); + if (buffer.size() > offset + payload_len) { + buffer = Truncate(std::move(buffer), offset + payload_len); + } + PW_TRY(EncodeHeader(buffer, payload_len)); + return buffer; +} + +Result<Buf> OutboundPacket::Encode(Buf buffer) const { + size_t payload_len = 0; + if (type_ != PacketType::kStreamEnd && type_ != PacketType::kError) { + const size_t offset = payload_offset(); + if (buffer.size() > offset) { + payload_len = buffer.size() - offset; + } + } + return Encode(std::move(buffer), payload_len); +} + +} // namespace pw::rpc2::internal
diff --git a/pw_rpc2/packet_test.cc b/pw_rpc2/packet_test.cc new file mode 100644 index 0000000..dc6c2a5 --- /dev/null +++ b/pw_rpc2/packet_test.cc
@@ -0,0 +1,427 @@ +// Copyright 2026 The Pigweed Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); you may not +// use this file except in compliance with the License. You may obtain a copy of +// the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +// License for the specific language governing permissions and limitations under +// the License. + +#include "pw_rpc2/internal/packet.h" + +#include <algorithm> +#include <array> +#include <cstddef> +#include <string_view> + +#include "pw_allocator/testing.h" +#include "pw_assert/check.h" +#include "pw_bytes/array.h" +#include "pw_bytes/span.h" +#include "pw_unit_test/framework.h" + +namespace { + +namespace internal = ::pw::rpc2::internal; + +std::string_view AsString(pw::ConstByteSpan bytes) { + return std::string_view(reinterpret_cast<const char*>(bytes.data()), + bytes.size()); +} + +TEST(PacketTest, EncodeDecodeRequest) { + pw::allocator::test::AllocatorForTest<256> allocator; + + constexpr std::string_view kPayload = "hello request"; + size_t total_size = sizeof(internal::RequestWireFormat) + kPayload.size(); + + auto buf = pw::Buf::Allocate(allocator, total_size); + + auto packet = + internal::OutboundPacket::Request(0x12345678, 0xabcdef01, 0x23456789); + pw::ConstByteSpan payload_bytes = pw::as_bytes(pw::span(kPayload)); + std::copy(payload_bytes.begin(), + payload_bytes.end(), + buf.data() + packet.payload_offset()); + + auto encode_result = packet.Encode(std::move(buf)); + ASSERT_EQ(encode_result.status(), pw::OkStatus()); + pw::Buf encoded_buf = std::move(encode_result.value()); + EXPECT_EQ(encoded_buf.size(), total_size); + + auto decode_result = + internal::InboundPacket::Decode(pw::ConstBuf(std::move(encoded_buf))); + ASSERT_EQ(decode_result.status(), pw::OkStatus()); + + internal::InboundPacket decoded = std::move(decode_result.value()); + EXPECT_EQ(decoded.type(), internal::PacketType::kRequest); + EXPECT_EQ(decoded.payload_offset(), sizeof(internal::RequestWireFormat)); + EXPECT_EQ(decoded.call_id(), 0x12345678u); + EXPECT_EQ(decoded.service_id(), 0xabcdef01u); + EXPECT_EQ(decoded.method_id(), 0x23456789u); + EXPECT_EQ(decoded.payload().size(), kPayload.size()); + EXPECT_EQ(AsString(decoded.payload()), kPayload); + + pw::ConstBuf payload_buf = std::move(decoded).TakePayload(); + EXPECT_EQ(payload_buf.size(), kPayload.size()); + EXPECT_EQ(AsString(payload_buf), kPayload); +} + +TEST(PacketTest, EncodeDecodeMessage) { + pw::allocator::test::AllocatorForTest<256> allocator; + + constexpr std::string_view kPayload = "hello msg"; + size_t total_size = sizeof(internal::MessageWireFormat) + kPayload.size(); + + auto buf = pw::Buf::Allocate(allocator, total_size); + + auto packet = internal::OutboundPacket::Message(0x12345678); + pw::ConstByteSpan payload_bytes = pw::as_bytes(pw::span(kPayload)); + std::copy(payload_bytes.begin(), + payload_bytes.end(), + buf.data() + packet.payload_offset()); + + auto encode_result = packet.Encode(std::move(buf)); + ASSERT_EQ(encode_result.status(), pw::OkStatus()); + pw::Buf encoded_buf = std::move(encode_result.value()); + EXPECT_EQ(encoded_buf.size(), total_size); + + auto decode_result = + internal::InboundPacket::Decode(pw::ConstBuf(std::move(encoded_buf))); + ASSERT_EQ(decode_result.status(), pw::OkStatus()); + + internal::InboundPacket decoded = std::move(decode_result.value()); + EXPECT_EQ(decoded.type(), internal::PacketType::kMessage); + EXPECT_EQ(decoded.payload_offset(), sizeof(internal::MessageWireFormat)); + EXPECT_EQ(decoded.call_id(), 0x12345678u); + EXPECT_EQ(decoded.payload().size(), kPayload.size()); + EXPECT_EQ(AsString(decoded.payload()), kPayload); + + pw::ConstBuf payload_buf = std::move(decoded).TakePayload(); + EXPECT_EQ(payload_buf.size(), kPayload.size()); + EXPECT_EQ(AsString(payload_buf), kPayload); +} + +TEST(PacketTest, EncodeDecodeStreamEnd) { + pw::allocator::test::AllocatorForTest<256> allocator; + size_t total_size = sizeof(internal::StreamEndWireFormat); + + auto buf = pw::Buf::Allocate(allocator, total_size); + + auto encode_result = + internal::OutboundPacket::StreamEnd(0x12345678).Encode(std::move(buf)); + ASSERT_EQ(encode_result.status(), pw::OkStatus()); + pw::Buf encoded_buf = std::move(encode_result.value()); + EXPECT_EQ(encoded_buf.size(), total_size); + + auto decode_result = + internal::InboundPacket::Decode(pw::ConstBuf(std::move(encoded_buf))); + ASSERT_EQ(decode_result.status(), pw::OkStatus()); + + internal::InboundPacket decoded = std::move(decode_result.value()); + EXPECT_EQ(decoded.type(), internal::PacketType::kStreamEnd); + EXPECT_EQ(decoded.call_id(), 0x12345678u); +} + +TEST(PacketTest, EncodeDecodeError) { + pw::allocator::test::AllocatorForTest<256> allocator; + size_t total_size = sizeof(internal::ErrorWireFormat); + + auto buf = pw::Buf::Allocate(allocator, total_size); + + auto encode_result = internal::OutboundPacket::Error( + 0x12345678, pw::Status::FailedPrecondition()) + .Encode(std::move(buf)); + ASSERT_EQ(encode_result.status(), pw::OkStatus()); + pw::Buf encoded_buf = std::move(encode_result.value()); + EXPECT_EQ(encoded_buf.size(), total_size); + + auto decode_result = + internal::InboundPacket::Decode(pw::ConstBuf(std::move(encoded_buf))); + ASSERT_EQ(decode_result.status(), pw::OkStatus()); + + internal::InboundPacket decoded = std::move(decode_result.value()); + EXPECT_EQ(decoded.type(), internal::PacketType::kError); + EXPECT_EQ(decoded.call_id(), 0x12345678u); + EXPECT_EQ(decoded.status(), pw::Status::FailedPrecondition()); +} + +TEST(PacketTest, DecodeBufferTooShortForHeader) { + pw::allocator::test::AllocatorForTest<256> allocator; + + auto buf = pw::Buf::Allocate(allocator, sizeof(internal::PacketHeader) - 1); + EXPECT_EQ( + internal::InboundPacket::Decode(pw::ConstBuf(std::move(buf))).status(), + pw::Status::DataLoss()); +} + +TEST(PacketTest, DecodeBufferTooShortForType) { + pw::allocator::test::AllocatorForTest<256> allocator; + + // Long enough for the common header, but not for a request header. + auto buf = pw::Buf::Allocate(allocator, sizeof(internal::PacketHeader)); + ASSERT_EQ(internal::OutboundPacket::Request(1, 2, 3) + .EncodeHeader(pw::ByteSpan(buf)) + .status(), + pw::Status::ResourceExhausted()); + // Write just the type byte by hand, since the header does not fit. + buf[offsetof(internal::PacketHeader, type)] = + static_cast<std::byte>(internal::PacketType::kRequest); + + EXPECT_EQ( + internal::InboundPacket::Decode(pw::ConstBuf(std::move(buf))).status(), + pw::Status::DataLoss()); +} + +TEST(PacketTest, DecodeUnrecognizedType) { + pw::allocator::test::AllocatorForTest<256> allocator; + + for (std::byte invalid_type : + {std::byte{0x00}, std::byte{0x06}, std::byte{0x7f}, std::byte{0xff}}) { + auto buf = pw::Buf::Allocate(allocator, 32); + buf[offsetof(internal::PacketHeader, type)] = invalid_type; + + EXPECT_EQ( + internal::InboundPacket::Decode(pw::ConstBuf(std::move(buf))).status(), + pw::Status::InvalidArgument()); + } +} + +TEST(PacketTest, EncodeDecodeHandshakePacket) { + pw::allocator::test::AllocatorForTest<256> allocator; + size_t total_size = internal::HandshakePacket::kWireSizeBytes; + EXPECT_EQ(total_size, 8u); + + auto buf = pw::Buf::Allocate(allocator, total_size); + + internal::HandshakePacket packet(internal::HandshakePacket::Type::kSyn); + + auto encode_result = packet.Encode(std::move(buf)); + ASSERT_EQ(encode_result.status(), pw::OkStatus()); + pw::Buf encoded_buf = std::move(encode_result.value()); + EXPECT_EQ(encoded_buf.size(), total_size); + + auto decode_result = internal::HandshakePacket::Decode(encoded_buf); + ASSERT_EQ(decode_result.status(), pw::OkStatus()); + + internal::HandshakePacket decoded = decode_result.value(); + EXPECT_EQ(decoded.version(), 1u); + EXPECT_EQ(decoded.type(), internal::HandshakePacket::Type::kSyn); +} + +TEST(PacketTest, DecodeZeroLengthPayload) { + pw::allocator::test::AllocatorForTest<256> allocator; + size_t total_size = sizeof(internal::RequestWireFormat); + + auto buf = pw::Buf::Allocate(allocator, total_size); + + auto encode_result = + internal::OutboundPacket::Request(0x12345678, 0x100, 0x200) + .Encode(std::move(buf)); + ASSERT_EQ(encode_result.status(), pw::OkStatus()); + pw::Buf encoded_buf = std::move(encode_result.value()); + EXPECT_EQ(encoded_buf.size(), total_size); + + auto decode_result = + internal::InboundPacket::Decode(pw::ConstBuf(std::move(encoded_buf))); + ASSERT_EQ(decode_result.status(), pw::OkStatus()); + + internal::InboundPacket decoded = std::move(decode_result.value()); + EXPECT_EQ(decoded.type(), internal::PacketType::kRequest); + EXPECT_EQ(decoded.payload().size(), 0u); + + pw::ConstBuf payload_buf = std::move(decoded).TakePayload(); + EXPECT_TRUE(payload_buf.empty()); +} + +TEST(PacketTest, DecodeCorruptHandshakePacket) { + pw::allocator::test::AllocatorForTest<256> allocator; + auto buf = pw::Buf::Allocate(allocator, 4); + EXPECT_EQ(internal::HandshakePacket::Decode(buf).status(), + pw::Status::DataLoss()); +} + +TEST(PacketTest, EncodeDecodeResponse) { + pw::allocator::test::AllocatorForTest<256> allocator; + + constexpr std::string_view kPayload = "hello response"; + size_t total_size = sizeof(internal::ResponseWireFormat) + kPayload.size(); + + auto buf = pw::Buf::Allocate(allocator, total_size); + + auto packet = internal::OutboundPacket::Response(0x12345678); + pw::ConstByteSpan payload_bytes = pw::as_bytes(pw::span(kPayload)); + std::copy(payload_bytes.begin(), + payload_bytes.end(), + buf.data() + packet.payload_offset()); + + auto encode_result = packet.Encode(std::move(buf)); + ASSERT_EQ(encode_result.status(), pw::OkStatus()); + pw::Buf encoded_buf = std::move(encode_result.value()); + EXPECT_EQ(encoded_buf.size(), total_size); + + auto decode_result = + internal::InboundPacket::Decode(pw::ConstBuf(std::move(encoded_buf))); + ASSERT_EQ(decode_result.status(), pw::OkStatus()); + + internal::InboundPacket decoded = std::move(decode_result.value()); + EXPECT_EQ(decoded.type(), internal::PacketType::kResponse); + EXPECT_EQ(decoded.payload_offset(), sizeof(internal::ResponseWireFormat)); + EXPECT_EQ(decoded.call_id(), 0x12345678u); + EXPECT_EQ(decoded.payload().size(), kPayload.size()); + EXPECT_EQ(AsString(decoded.payload()), kPayload); + + pw::ConstBuf payload_buf = std::move(decoded).TakePayload(); + EXPECT_EQ(payload_buf.size(), kPayload.size()); + EXPECT_EQ(AsString(payload_buf), kPayload); +} + +TEST(PacketTest, EncodeStreamEndTruncatesExtraBuffer) { + pw::allocator::test::AllocatorForTest<256> allocator; + auto buf = pw::Buf::Allocate(allocator, 64); + auto encode_result = + internal::OutboundPacket::StreamEnd(0x12345678).Encode(std::move(buf)); + ASSERT_EQ(encode_result.status(), pw::OkStatus()); + EXPECT_EQ(encode_result->size(), sizeof(internal::StreamEndWireFormat)); +} + +TEST(PacketTest, EncodeErrorTruncatesExtraBuffer) { + pw::allocator::test::AllocatorForTest<256> allocator; + auto buf = pw::Buf::Allocate(allocator, 64); + auto encode_result = internal::OutboundPacket::Error( + 0x12345678, pw::Status::FailedPrecondition()) + .Encode(std::move(buf)); + ASSERT_EQ(encode_result.status(), pw::OkStatus()); + EXPECT_EQ(encode_result->size(), sizeof(internal::ErrorWireFormat)); +} + +TEST(PacketTest, EncodeRejectsBufferSmallerThanHeader) { + pw::allocator::test::AllocatorForTest<256> allocator; + auto buf = + pw::Buf::Allocate(allocator, sizeof(internal::RequestWireFormat) - 1); + EXPECT_EQ(internal::OutboundPacket::Request(1, 2, 3) + .Encode(std::move(buf)) + .status(), + pw::Status::ResourceExhausted()); +} + +TEST(PacketTest, EncodeRejectsEmptyBuffer) { + EXPECT_EQ(internal::OutboundPacket::Message(1).Encode(pw::Buf()).status(), + pw::Status::FailedPrecondition()); +} + +static_assert( + internal::PacketSizeWithoutPayload(internal::PacketType::kRequest) == 13u); +static_assert( + internal::PacketSizeWithoutPayload(internal::PacketType::kMessage) == 5u); +static_assert( + internal::PacketSizeWithoutPayload(internal::PacketType::kResponse) == 5u); +static_assert( + internal::PacketSizeWithoutPayload(internal::PacketType::kStreamEnd) == 5u); +static_assert( + internal::PacketSizeWithoutPayload(internal::PacketType::kError) == 9u); + +static_assert(internal::OutboundPacket::Request(1, 2, 3).payload_offset() == + 13u); +static_assert(internal::OutboundPacket::Message(1).payload_offset() == 5u); +static_assert(internal::OutboundPacket::Response(1).payload_offset() == 5u); +static_assert(internal::OutboundPacket::StreamEnd(1).payload_offset() == 5u); +static_assert(internal::OutboundPacket::Error(1, pw::Status::Internal()) + .payload_offset() == 9u); + +TEST(PacketTest, EncodedMessageHeaderLayout) { + std::array<std::byte, sizeof(internal::MessageWireFormat)> buffer = {}; + + auto result = + internal::OutboundPacket::Message(0x12345678).EncodeHeader(buffer); + ASSERT_EQ(result.status(), pw::OkStatus()); + EXPECT_EQ(*result, buffer.size()); + + // call_id is little endian and comes first, followed by the type byte. + constexpr auto expected = pw::bytes::Array<0x78, 0x56, 0x34, 0x12, 0x02>(); + EXPECT_EQ(buffer, expected); +} + +TEST(PacketTest, EncodedRequestHeaderLayout) { + std::array<std::byte, sizeof(internal::RequestWireFormat)> buffer = {}; + + auto result = + internal::OutboundPacket::Request(0x12345678, 0xabcdef01, 0x23456789) + .EncodeHeader(buffer); + ASSERT_EQ(result.status(), pw::OkStatus()); + EXPECT_EQ(*result, buffer.size()); + + constexpr auto expected = pw::bytes::Array<0x78, + 0x56, + 0x34, + 0x12, + 0x01, + 0x01, + 0xef, + 0xcd, + 0xab, + 0x89, + 0x67, + 0x45, + 0x23>(); + EXPECT_EQ(buffer, expected); +} + +TEST(PacketTest, EncodeHeaderReportsTotalSizeWithoutWritingPayloadLength) { + constexpr auto buffer_init = [](size_t i) { + return static_cast<std::byte>(i + 1); + }; + auto buffer = pw::bytes::Initialized<64>(buffer_init); + + auto result = internal::OutboundPacket::Response(1).EncodeHeader(buffer, 10); + ASSERT_EQ(result.status(), pw::OkStatus()); + EXPECT_EQ(*result, sizeof(internal::ResponseWireFormat) + 10u); + + // Nothing past the header was modified. + for (size_t i = sizeof(internal::ResponseWireFormat); i < buffer.size(); + ++i) { + EXPECT_EQ(buffer[i], buffer_init(i)); + } +} + +TEST(PacketTest, EncodeHeaderRejectsBufferSmallerThanHeaderPlusPayload) { + std::array<std::byte, sizeof(internal::MessageWireFormat) + 4> buffer = {}; + + EXPECT_EQ( + internal::OutboundPacket::Message(1).EncodeHeader(buffer, 5).status(), + pw::Status::ResourceExhausted()); + EXPECT_EQ( + internal::OutboundPacket::Message(1).EncodeHeader(buffer, 4).status(), + pw::OkStatus()); +} + +TEST(PacketTest, DecodedPayloadIsEverythingAfterTheHeader) { + pw::allocator::test::AllocatorForTest<256> allocator; + + // The payload length is not on the wire, so all trailing bytes belong to the + // payload regardless of what was passed to EncodeHeader. + constexpr size_t kTrailingBytes = 7; + auto buf = pw::Buf::Allocate( + allocator, sizeof(internal::MessageWireFormat) + kTrailingBytes); + + auto encode_result = + internal::OutboundPacket::Message(42).EncodeHeader(pw::ByteSpan(buf)); + ASSERT_EQ(encode_result.status(), pw::OkStatus()); + ASSERT_EQ(*encode_result, sizeof(internal::MessageWireFormat)); + + auto decode_result = + internal::InboundPacket::Decode(pw::ConstBuf(std::move(buf))); + ASSERT_EQ(decode_result.status(), pw::OkStatus()); + + internal::InboundPacket decoded = std::move(decode_result.value()); + EXPECT_EQ(decoded.type(), internal::PacketType::kMessage); + EXPECT_EQ(decoded.call_id(), 42u); + EXPECT_EQ(decoded.payload().size(), kTrailingBytes); +} + +} // namespace
diff --git a/pw_rpc2/public/pw_rpc2/internal/packet.h b/pw_rpc2/public/pw_rpc2/internal/packet.h new file mode 100644 index 0000000..84ee831 --- /dev/null +++ b/pw_rpc2/public/pw_rpc2/internal/packet.h
@@ -0,0 +1,330 @@ +// Copyright 2026 The Pigweed Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); you may not +// use this file except in compliance with the License. You may obtain a copy of +// the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +// License for the specific language governing permissions and limitations under +// the License. +#pragma once + +#include <cstddef> +#include <cstdint> +#include <type_traits> +#include <utility> + +#include "pw_assert/assert.h" +#include "pw_buf/buf.h" +#include "pw_bytes/endian.h" +#include "pw_bytes/span.h" +#include "pw_preprocessor/compiler.h" +#include "pw_result/result.h" +#include "pw_status/status.h" + +namespace pw::rpc2::internal { + +/// Identifies the role and wire format of a regular protocol packet in pw_rpc2. +/// +/// These packets are exchanged after the initial handshake is established. +/// Encoded in the packet header to indicate how the packet is framed, +/// decoded, and processed. +enum class PacketType : uint8_t { + /// Initiates an RPC invocation from client to server. + kRequest = 0x01, + + /// Carries streaming payload data for an in-flight RPC. + kMessage = 0x02, + + /// Signals the completion of a client or server stream. + kStreamEnd = 0x03, + + /// Terminates an RPC with an error status. + kError = 0x04, + + /// Completes the RPC and delivers a response payload. + kResponse = 0x05, +}; + +/// Common prefix of every regular protocol packet (following the handshake). +PW_PACKED(struct) PacketHeader { + uint32_t call_id; + uint8_t type; +}; + +/// Initiates an RPC invocation. Sent by the client to request execution of +/// a specific service method, optionally followed by request payload bytes. +PW_PACKED(struct) RequestWireFormat { + PacketHeader header; + uint32_t service_id; + uint32_t method_id; +}; + +/// Carries streaming payload data for an in-flight RPC in either direction. +PW_PACKED(struct) MessageWireFormat { PacketHeader header; }; + +/// Completes a unary RPC or delivers a response payload from server to client. +PW_PACKED(struct) ResponseWireFormat { PacketHeader header; }; + +/// Signals the end of a stream in one direction without an error. +PW_PACKED(struct) StreamEndWireFormat { PacketHeader header; }; + +/// Signals that an RPC has terminated abnormally with an error status code. +PW_PACKED(struct) ErrorWireFormat { + PacketHeader header; + uint32_t status_code; +}; + +/// Connection handshake packet for protocol negotiation and compatibility +/// verification. +/// +/// Handshake packets are sent first before any regular protocol packets. They +/// include a magic value to verify that the remote peer is an RPC endpoint +/// before establishing the session. +PW_PACKED(struct) HandshakeWireFormat { + uint32_t magic; + uint8_t version; + uint8_t type; + uint16_t reserved; +}; + +static_assert(sizeof(PacketHeader) == 5); +static_assert(sizeof(RequestWireFormat) == 13); +static_assert(sizeof(MessageWireFormat) == 5); +static_assert(sizeof(ResponseWireFormat) == 5); +static_assert(sizeof(StreamEndWireFormat) == 5); +static_assert(sizeof(ErrorWireFormat) == 9); +static_assert(sizeof(HandshakeWireFormat) == 8); + +/// Returns the wire format size for `type` excluding any trailing payload +/// bytes. +constexpr size_t PacketSizeWithoutPayload(PacketType type) { + switch (type) { + case PacketType::kRequest: + return sizeof(RequestWireFormat); + case PacketType::kMessage: + return sizeof(MessageWireFormat); + case PacketType::kResponse: + return sizeof(ResponseWireFormat); + case PacketType::kStreamEnd: + return sizeof(StreamEndWireFormat); + case PacketType::kError: + return sizeof(ErrorWireFormat); + } + PW_DASSERT(false); + return 0; +} + +class HandshakePacket { + public: + enum class Type : uint8_t { + kSyn = 1, + kSynAck = 2, + kAck = 3, + }; + + static constexpr size_t kWireSizeBytes = sizeof(HandshakeWireFormat); + + static Result<HandshakePacket> Decode(ConstByteSpan bytes); + + explicit HandshakePacket(Type type) : type_(type) {} + + uint8_t version() const { return kVersion; } + Type type() const { return type_; } + + Status Encode(ByteSpan buffer) const; + Result<Buf> Encode(Buf buffer) const; + + private: + static constexpr uint32_t kMagic = 0x43505250; // 'PRPC' + static constexpr uint8_t kVersion = 1; + + Type type_; +}; + +/// A packet to be written to a connection. +/// +/// This is a small value type that describes a packet header; the payload +/// bytes are written separately by whoever holds the transport reservation. +/// Copies of it are stored per pending write, so it is kept compact. +class OutboundPacket { + public: + static constexpr OutboundPacket Request(uint32_t call_id, + uint32_t service_id, + uint32_t method_id) { + return OutboundPacket( + PacketType::kRequest, call_id, Fields(service_id, method_id)); + } + + static constexpr OutboundPacket Message(uint32_t call_id) { + return OutboundPacket(PacketType::kMessage, call_id, Fields()); + } + + static constexpr OutboundPacket Response(uint32_t call_id) { + return OutboundPacket(PacketType::kResponse, call_id, Fields()); + } + + static constexpr OutboundPacket StreamEnd(uint32_t call_id) { + return OutboundPacket(PacketType::kStreamEnd, call_id, Fields()); + } + + static constexpr OutboundPacket Error(uint32_t call_id, Status status) { + return OutboundPacket(PacketType::kError, call_id, Fields(status)); + } + + constexpr OutboundPacket() + : OutboundPacket(PacketType::kMessage, 0, Fields()) {} + + OutboundPacket(const OutboundPacket&) = default; + OutboundPacket& operator=(const OutboundPacket&) = default; + OutboundPacket(OutboundPacket&&) noexcept = default; + OutboundPacket& operator=(OutboundPacket&&) noexcept = default; + + constexpr PacketType type() const { return type_; } + constexpr uint32_t call_id() const { return call_id_; } + + uint32_t service_id() const { + PW_DASSERT(type_ == PacketType::kRequest); + return fields_.request.service_id; + } + + uint32_t method_id() const { + PW_DASSERT(type_ == PacketType::kRequest); + return fields_.request.method_id; + } + + Status status() const { + PW_DASSERT(type_ == PacketType::kError); + return fields_.status; + } + + constexpr size_t payload_offset() const { + return PacketSizeWithoutPayload(type_); + } + + /// Encodes this packet's header at the start of `buffer`. + /// + /// `payload_len` is not written to the wire; it is only used to check that + /// `buffer` is large enough for the header plus payload, and is included in + /// the returned total packet size. + Result<size_t> EncodeHeader(ByteSpan buffer, size_t payload_len = 0) const; + + Result<Buf> Encode(Buf buffer, size_t payload_len) const; + Result<Buf> Encode(Buf buffer) const; + + private: + struct RequestIds { + uint32_t service_id; + uint32_t method_id; + }; + + // The header fields that are specific to one packet type. Which member is + // live is determined by `type_`, and the public accessors check it before + // reading. Overlaying them keeps an `OutboundPacket` at 16 bytes rather than + // 20, which matters because one is stored in every pending write. + union Fields { + constexpr Fields() : request{0, 0} {} + constexpr Fields(uint32_t service_id, uint32_t method_id) + : request{service_id, method_id} {} + constexpr explicit Fields(Status error_status) : status(error_status) {} + + RequestIds request; + Status status; + }; + + constexpr OutboundPacket(PacketType type, uint32_t call_id, Fields fields) + : type_(type), call_id_(call_id), fields_(fields) {} + + PacketType type_; + uint32_t call_id_; + Fields fields_; +}; + +static_assert(std::is_trivially_copyable_v<OutboundPacket>); + +/// An incoming packet received from a connection. +/// +/// `InboundPacket` wraps an owned buffer (`ConstBuf`) containing an encoded +/// packet, validates the header on decoding, and provides accessors to the +/// header fields and the trailing payload data. +class InboundPacket { + public: + static Result<InboundPacket> Decode(ConstBuf&& buffer); + + constexpr InboundPacket() = default; + constexpr InboundPacket(std::nullptr_t) noexcept : InboundPacket() {} + + InboundPacket(InboundPacket&&) noexcept = default; + InboundPacket& operator=(InboundPacket&&) noexcept = default; + + InboundPacket(const InboundPacket&) = delete; + InboundPacket& operator=(const InboundPacket&) = delete; + + [[nodiscard]] friend constexpr bool operator==(const InboundPacket& lhs, + std::nullptr_t) noexcept { + return lhs.buffer_ == nullptr; + } + [[nodiscard]] friend constexpr bool operator==( + std::nullptr_t, const InboundPacket& rhs) noexcept { + return rhs.buffer_ == nullptr; + } + [[nodiscard]] friend constexpr bool operator!=(const InboundPacket& lhs, + std::nullptr_t) noexcept { + return lhs.buffer_ != nullptr; + } + [[nodiscard]] friend constexpr bool operator!=( + std::nullptr_t, const InboundPacket& rhs) noexcept { + return rhs.buffer_ != nullptr; + } + + PacketType type() const { + return static_cast<PacketType>(buffer_[offsetof(PacketHeader, type)]); + } + + uint32_t call_id() const { + return ReadUint32(offsetof(PacketHeader, call_id)); + } + + uint32_t service_id() const { + PW_DASSERT(type() == PacketType::kRequest); + return ReadUint32(offsetof(RequestWireFormat, service_id)); + } + + uint32_t method_id() const { + PW_DASSERT(type() == PacketType::kRequest); + return ReadUint32(offsetof(RequestWireFormat, method_id)); + } + + Status status() const { + PW_DASSERT(type() == PacketType::kError); + uint32_t status_code = ReadUint32(offsetof(ErrorWireFormat, status_code)); + return Status(static_cast<Status::Code>(status_code)); + } + + size_t payload_offset() const { return PacketSizeWithoutPayload(type()); } + + ConstByteSpan payload() const { + return ConstByteSpan(buffer_).subspan(payload_offset()); + } + + /// Slices and returns the payload into an owned `ConstBuf`. + [[nodiscard]] ConstBuf TakePayload() && { + return Slice(std::move(buffer_), payload_offset()); + } + + private: + explicit InboundPacket(ConstBuf&& buffer) : buffer_(std::move(buffer)) {} + + uint32_t ReadUint32(size_t offset) const { + return bytes::ReadInOrder<uint32_t>(endian::little, + buffer_.data() + offset); + } + + ConstBuf buffer_; +}; + +} // namespace pw::rpc2::internal