pw_rpc2/cpp/client_call_test.cc

// 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/client_call.h"

#include <cstddef>
#include <cstring>
#include <optional>
#include <type_traits>
#include <utility>

#include "pw_allocator/testing.h"
#include "pw_assert/check.h"
#include "pw_async2/dispatcher_for_test.h"
#include "pw_bytes/endian.h"
#include "pw_rpc2/client.h"
#include "pw_rpc2/internal/call.h"
#include "pw_rpc2/internal/handshake.h"
#include "pw_rpc2/internal/packet.h"
#include "pw_rpc2/internal/packet_testing.h"
#include "pw_rpc2/internal/test_utils.h"
#include "pw_rpc2/server.h"
#include "pw_transport/transport.h"
#include "pw_unit_test/framework.h"

namespace pw::rpc2 {
namespace {

namespace flags = ::pw::rpc2::internal::flags;

static_assert(!std::is_copy_constructible_v<RawReadFuture>);
static_assert(!std::is_copy_assignable_v<RawReadFuture>);
static_assert(std::is_move_constructible_v<RawReadFuture>);
static_assert(std::is_move_assignable_v<RawReadFuture>);

static_assert(!std::is_copy_constructible_v<ReadFuture<pw::ConstBuf>>);
static_assert(!std::is_copy_assignable_v<ReadFuture<pw::ConstBuf>>);
static_assert(std::is_move_constructible_v<ReadFuture<pw::ConstBuf>>);
static_assert(std::is_move_assignable_v<ReadFuture<pw::ConstBuf>>);

static_assert(
    !std::is_copy_constructible_v<UnaryCallFuture<pw::ConstBuf, pw::ConstBuf>>);
static_assert(
    !std::is_copy_assignable_v<UnaryCallFuture<pw::ConstBuf, pw::ConstBuf>>);
static_assert(
    std::is_move_constructible_v<UnaryCallFuture<pw::ConstBuf, pw::ConstBuf>>);
static_assert(
    std::is_move_assignable_v<UnaryCallFuture<pw::ConstBuf, pw::ConstBuf>>);

static_assert(!std::is_copy_constructible_v<
              ServerStreamFuture<pw::ConstBuf, pw::ConstBuf>>);
static_assert(
    !std::is_copy_assignable_v<ServerStreamFuture<pw::ConstBuf, pw::ConstBuf>>);
static_assert(std::is_move_constructible_v<
              ServerStreamFuture<pw::ConstBuf, pw::ConstBuf>>);
static_assert(
    std::is_move_assignable_v<ServerStreamFuture<pw::ConstBuf, pw::ConstBuf>>);

static_assert(!std::is_copy_constructible_v<
              ClientStreamFuture<pw::ConstBuf, pw::ConstBuf>>);
static_assert(
    !std::is_copy_assignable_v<ClientStreamFuture<pw::ConstBuf, pw::ConstBuf>>);
static_assert(std::is_move_constructible_v<
              ClientStreamFuture<pw::ConstBuf, pw::ConstBuf>>);
static_assert(
    std::is_move_assignable_v<ClientStreamFuture<pw::ConstBuf, pw::ConstBuf>>);

static_assert(!std::is_copy_constructible_v<
              BidiStreamFuture<pw::ConstBuf, pw::ConstBuf>>);
static_assert(
    !std::is_copy_assignable_v<BidiStreamFuture<pw::ConstBuf, pw::ConstBuf>>);
static_assert(
    std::is_move_constructible_v<BidiStreamFuture<pw::ConstBuf, pw::ConstBuf>>);
static_assert(
    std::is_move_assignable_v<BidiStreamFuture<pw::ConstBuf, pw::ConstBuf>>);

class ClientConnectTask : public async2::Task {
 public:
  explicit ClientConnectTask(ClientFuture fut) : fut_(std::move(fut)) {}

  async2::Poll<> DoPend(async2::Context& cx) override {
    auto poll = fut_.Pend(cx);
    if (poll.IsPending()) {
      return async2::Pending();
    }
    client_ = std::move(*poll);
    return async2::Ready();
  }

  std::optional<Result<Client>>& client() { return client_; }
  const std::optional<Result<Client>>& client() const { return client_; }

  ~ClientConnectTask() override {
    Deregister();
    // A `Client` must be closed before it is destroyed. If nothing took
    // this one, close it here.
    if (client_.has_value() && client_->ok()) {
      auto closed = (*client_)->Close();
    }
  }

 private:
  ClientFuture fut_;
  std::optional<Result<Client>> client_;
};

template <typename Fut = UnaryCallFuture<pw::ConstBuf, pw::ConstBuf>>
class UnaryTestTask : public async2::Task {
 public:
  explicit UnaryTestTask(Fut fut) : fut_(std::move(fut)) {}

  async2::Poll<> DoPend(async2::Context& cx) override {
    auto poll = fut_.Pend(cx);
    if (poll.IsPending()) {
      return async2::Pending();
    }
    result_ = std::move(*poll);
    return async2::Ready();
  }

  const std::optional<Result<pw::ConstBuf>>& result() const { return result_; }

  // Exposed so tests can assert the future's state after it resolves.
  const Fut& future() const { return fut_; }

 private:
  Fut fut_;
  std::optional<Result<pw::ConstBuf>> result_;
};

template <typename Fut>
UnaryTestTask(Fut) -> UnaryTestTask<Fut>;

template <typename Fut = RawReadFuture>
class ReadTestTask : public async2::Task {
 public:
  explicit ReadTestTask(Fut fut) : fut_(std::move(fut)) {}

  async2::Poll<> DoPend(async2::Context& cx) override {
    auto poll = fut_.Pend(cx);
    if (poll.IsPending()) {
      return async2::Pending();
    }
    result_ = std::move(*poll);
    return async2::Ready();
  }

  const std::optional<Result<pw::ConstBuf>>& result() const { return result_; }

 private:
  Fut fut_;
  std::optional<Result<pw::ConstBuf>> result_;
};

template <typename Fut>
ReadTestTask(Fut) -> ReadTestTask<Fut>;

template <typename Fut>
class WriteTestTask : public async2::Task {
 public:
  explicit WriteTestTask(Fut fut) : fut_(std::move(fut)) {}

  async2::Poll<> DoPend(async2::Context& cx) override {
    auto poll = fut_.Pend(cx);
    if (poll.IsPending()) {
      return async2::Pending();
    }
    result_ = *poll;
    return async2::Ready();
  }

  const std::optional<Status>& result() const { return result_; }

 private:
  Fut fut_;
  std::optional<Status> result_;
};

template <typename Fut>
WriteTestTask(Fut) -> WriteTestTask<Fut>;

template <typename Fut = PacketReservationFuture>
class ReserveTestTask : public async2::Task {
 public:
  explicit ReserveTestTask(Fut fut) : fut_(std::move(fut)) {}

  async2::Poll<> DoPend(async2::Context& cx) override {
    auto poll = fut_.Pend(cx);
    if (poll.IsPending()) {
      return async2::Pending();
    }
    result_ = std::move(*poll);
    return async2::Ready();
  }

  std::optional<Result<PacketReservation>>& result() { return result_; }
  const std::optional<Result<PacketReservation>>& result() const {
    return result_;
  }

 private:
  Fut fut_;
  std::optional<Result<PacketReservation>> result_;
};

template <typename Fut>
ReserveTestTask(Fut) -> ReserveTestTask<Fut>;

template <typename Fut>
class StreamCallTestTask : public async2::Task {
 public:
  explicit StreamCallTestTask(Fut fut) : fut_(std::move(fut)) {}

  async2::Poll<> DoPend(async2::Context& cx) override {
    auto poll = fut_.Pend(cx);
    if (poll.IsPending()) {
      return async2::Pending();
    }
    result_ = std::move(*poll);
    return async2::Ready();
  }

  using ResType = typename Fut::value_type;
  const std::optional<ResType>& result() const { return result_; }
  std::optional<ResType>& result() { return result_; }

 private:
  Fut fut_;
  std::optional<ResType> result_;
};

template <typename Fut>
StreamCallTestTask(Fut) -> StreamCallTestTask<Fut>;

class TestServiceClient : public ServiceClient {
 public:
  TestServiceClient(const Client& client, uint32_t service_id)
      : ServiceClient(client, service_id) {}

  using ServiceClient::CallBidiStream;
  using ServiceClient::CallClientStream;
  using ServiceClient::CallServerStream;
  using ServiceClient::CallUnary;
};

Client SetupClient(pw::Allocator& allocator,
                   async2::DispatcherForTest& dispatcher,
                   test::MockConnection* raw_conn,
                   transport::ReliableDatagramSocket conn) {
  ClientConnectTask connect_task(Client::Create(dispatcher, allocator, conn));
  dispatcher.Post(connect_task);
  dispatcher.RunUntilStalled();

  pw::Buf hs_resp =
      pw::Buf::Allocate(allocator, internal::HandshakePacket::kWireSizeBytes);
  auto hs_pkt =
      internal::HandshakePacket(internal::HandshakePacket::Type::kSynAck);
  auto enc_res = hs_pkt.Encode(std::move(hs_resp));
  PW_CHECK(enc_res.ok());
  raw_conn->SetNextRead(std::move(enc_res.value()));

  dispatcher.RunUntilStalled();
  PW_CHECK(connect_task.client().has_value() && connect_task.client()->ok());
  Client client = std::move(**connect_task.client());
  connect_task.Deregister();
  return client;
}

// Frames a server packet of any type that is neither a start nor an error
// packet.
Result<pw::Buf> FrameServerPacket(pw::Allocator& allocator,
                                  internal::PacketType type,
                                  uint32_t call_id,
                                  pw::ConstByteSpan payload = {}) {
  PW_CHECK(type.is_server());
  return internal::PacketFramer::FramePacket(allocator, type, call_id, payload);
}

// Returns the call ID of the most recent packet the client sent.
uint32_t LastSentCallId(test::MockConnection& raw_conn) {
  auto decoded = internal::InboundPacket::Decode(raw_conn.last_written_buf());
  PW_CHECK_OK(decoded.status());
  return decoded->call_id();
}

// Checks that the most recent packet the client sent reports a method type
// mismatch for `call_id`.
void ExpectSentMethodTypeMismatch(test::MockConnection& raw_conn,
                                  uint32_t call_id) {
  auto decoded = internal::InboundPacket::Decode(raw_conn.last_written_buf());
  ASSERT_TRUE(decoded.ok());
  EXPECT_EQ(decoded->type(),
            internal::PacketType::Make<flags::kErrorTerminal>());
  EXPECT_EQ(decoded->call_id(), call_id);
  EXPECT_EQ(decoded->client_error(),
            internal::ClientError::kMethodTypeMismatch);
}

// Server packets that cannot answer a unary or client-streaming call, which
// takes exactly one packet with a payload that completes the RPC.
constexpr internal::PacketType kNotSingleResponse[] = {
    internal::PacketType::Make<flags::kServer, flags::kHasPayload>(),
    internal::PacketType::Make<flags::kServer, flags::kStreamEnd>(),
    internal::PacketType::
        Make<flags::kServer, flags::kHasPayload, flags::kStreamEnd>(),
    internal::PacketType::Make<flags::kServer, flags::kOkTerminal>(),
};

TEST(ClientCallTest, UnaryCallSendsRequestFirstThenReadsResponse) {
  allocator::test::AllocatorForTest<16384> allocator;
  async2::DispatcherForTest dispatcher;
  auto [conn, raw_conn] = test::MakeMockConnection(allocator);

  test::ScopedClient client(SetupClient(allocator, dispatcher, raw_conn, conn),
                            dispatcher);

  pw::Buf req_buf = pw::Buf::Allocate(allocator, 3);
  req_buf[0] = std::byte(10);
  req_buf[1] = std::byte(20);
  req_buf[2] = std::byte(30);

  TestServiceClient test_service(client, 100u);
  auto unary_call = test_service.CallUnary<pw::ConstBuf, pw::ConstBuf>(
      200u, pw::ConstBuf(std::move(req_buf)));

  UnaryTestTask task(std::move(unary_call));
  dispatcher.Post(task);
  dispatcher.RunUntilStalled();

  EXPECT_GE(raw_conn->commit_count(), 3u);  // 2 handshake + 1 request
  auto decode_req =
      internal::InboundPacket::Decode(raw_conn->last_written_buf());
  ASSERT_TRUE(decode_req.ok());
  EXPECT_EQ(decode_req->type(),
            (internal::PacketType::
                 Make<flags::kStart, flags::kHasPayload, flags::kStreamEnd>()));
  EXPECT_EQ(decode_req->service_id(), 100u);
  EXPECT_EQ(decode_req->method_id(), 200u);

  EXPECT_FALSE(task.result().has_value());

  pw::Buf resp_payload = pw::Buf::Allocate(allocator, 4);
  std::memset(resp_payload.data(), 0xAA, 4);
  auto framed_resp = internal::PacketFramer::FrameResponsePacket(
      allocator, /*call_id=*/1, pw::ConstByteSpan(resp_payload));
  ASSERT_TRUE(framed_resp.ok());
  raw_conn->SetNextRead(std::move(*framed_resp));

  dispatcher.RunUntilStalled();
  ASSERT_TRUE(task.result().has_value() && task.result()->ok());
  EXPECT_EQ(task.result()->value().size(), 4u);

  task.Deregister();
}

// A default-constructed future is empty: it must report itself as neither
// pendable nor complete, so that polling it is detectably a programming error.
TEST(ClientCallTest, DefaultConstructedUnaryCallFutureIsNotPendable) {
  UnaryCallFuture<pw::ConstBuf, pw::ConstBuf> fut;
  EXPECT_FALSE(fut.is_pendable());
  EXPECT_FALSE(fut.is_complete());
}

// Completion is tracked by the future itself rather than inferred from the
// read half, which never starts if the request fails.
TEST(ClientCallTest, UnaryCallFutureReportsCompleteOnceResolved) {
  allocator::test::AllocatorForTest<16384> allocator;
  async2::DispatcherForTest dispatcher;
  auto [conn, raw_conn] = test::MakeMockConnection(allocator);

  test::ScopedClient client(SetupClient(allocator, dispatcher, raw_conn, conn),
                            dispatcher);

  pw::Buf req_buf = pw::Buf::Allocate(allocator, 1);
  req_buf[0] = std::byte(7);

  TestServiceClient test_service(client, 100u);
  auto unary_call = test_service.CallUnary<pw::ConstBuf, pw::ConstBuf>(
      200u, pw::ConstBuf(std::move(req_buf)));

  EXPECT_TRUE(unary_call.is_pendable());
  EXPECT_FALSE(unary_call.is_complete());

  UnaryTestTask task(std::move(unary_call));
  dispatcher.Post(task);
  dispatcher.RunUntilStalled();

  pw::Buf resp_payload = pw::Buf::Allocate(allocator, 2);
  std::memset(resp_payload.data(), 0xBB, 2);
  auto framed_resp = internal::PacketFramer::FrameResponsePacket(
      allocator, /*call_id=*/1, pw::ConstByteSpan(resp_payload));
  ASSERT_TRUE(framed_resp.ok());
  raw_conn->SetNextRead(std::move(*framed_resp));

  dispatcher.RunUntilStalled();
  ASSERT_TRUE(task.result().has_value() && task.result()->ok());

  EXPECT_TRUE(task.future().is_complete());
  EXPECT_FALSE(task.future().is_pendable());

  task.Deregister();
}

TEST(ClientCallTest, ServerStreamingReceivesMessagesAndStreamEnd) {
  allocator::test::AllocatorForTest<16384> allocator;
  async2::DispatcherForTest dispatcher;
  auto [conn, raw_conn] = test::MakeMockConnection(allocator);

  test::ScopedClient client(SetupClient(allocator, dispatcher, raw_conn, conn),
                            dispatcher);
  TestServiceClient test_service(client, 100u);

  auto server_stream_call =
      test_service.CallServerStream<pw::ConstBuf, pw::ConstBuf>(300u,
                                                                pw::ConstBuf());

  StreamCallTestTask call_task(std::move(server_stream_call));
  dispatcher.Post(call_task);
  dispatcher.RunUntilStalled();

  ASSERT_TRUE(call_task.result().has_value() && call_task.result()->ok());
  Reader<pw::ConstBuf> reader = std::move(**call_task.result());
  call_task.Deregister();

  // 1. Read message 1
  ReadTestTask read1(reader.Read());
  dispatcher.Post(read1);
  dispatcher.RunUntilStalled();
  EXPECT_FALSE(read1.result().has_value());

  std::byte chunk1[3] = {std::byte{1}, std::byte{2}, std::byte{3}};
  auto pkt1 =
      internal::PacketFramer::FrameServerMessagePacket(allocator, 1u, chunk1);
  ASSERT_TRUE(pkt1.ok());
  raw_conn->SetNextRead(std::move(*pkt1));
  dispatcher.RunUntilStalled();
  ASSERT_TRUE(read1.result().has_value() && read1.result()->ok());
  EXPECT_EQ(read1.result()->value().size(), 3u);
  read1.Deregister();

  // 2. Read message 2
  ReadTestTask read2(reader.Read());
  dispatcher.Post(read2);
  std::byte chunk2[2] = {std::byte{10}, std::byte{20}};
  auto pkt2 =
      internal::PacketFramer::FrameServerMessagePacket(allocator, 1u, chunk2);
  ASSERT_TRUE(pkt2.ok());
  raw_conn->SetNextRead(std::move(*pkt2));
  dispatcher.RunUntilStalled();
  ASSERT_TRUE(read2.result().has_value() && read2.result()->ok());
  EXPECT_EQ(read2.result()->value().size(), 2u);
  read2.Deregister();

  // 3. Receive StreamEnd
  ReadTestTask read3(reader.Read());
  dispatcher.Post(read3);
  auto end_pkt = internal::PacketFramer::FrameServerFinishPacket(allocator, 1u);
  ASSERT_TRUE(end_pkt.ok());
  raw_conn->SetNextRead(std::move(*end_pkt));
  dispatcher.RunUntilStalled();
  ASSERT_TRUE(read3.result().has_value());
  EXPECT_FALSE(read3.result()->ok());
  EXPECT_EQ(read3.result()->status(), Status::OutOfRange());
  read3.Deregister();
}

TEST(ClientCallTest, ClientStreamingSendsChunksThenReadsResponse) {
  allocator::test::AllocatorForTest<16384> allocator;
  async2::DispatcherForTest dispatcher;
  auto [conn, raw_conn] = test::MakeMockConnection(allocator);

  test::ScopedClient client(SetupClient(allocator, dispatcher, raw_conn, conn),
                            dispatcher);
  TestServiceClient test_service(client, 100u);

  auto client_stream_call =
      test_service.CallClientStream<pw::ConstBuf, pw::ConstBuf>(400u);

  StreamCallTestTask call_task(std::move(client_stream_call));
  dispatcher.Post(call_task);
  dispatcher.RunUntilStalled();

  ASSERT_TRUE(call_task.result().has_value() && call_task.result()->ok());
  auto [writer, response_future] = std::move(**call_task.result());
  call_task.Deregister();

  // Write chunk 1 via ReserveWrite
  ReserveTestTask write1(writer.ReserveWrite(2));
  dispatcher.Post(write1);
  dispatcher.RunUntilStalled();
  ASSERT_TRUE(write1.result().has_value() && write1.result()->ok());
  (*write1.result())->payload()[0] = std::byte{1};
  (*write1.result())->payload()[1] = std::byte{2};
  PW_TEST_EXPECT_OK((*write1.result())->Commit());
  write1.Deregister();

  // Write chunk 2 via ReserveWrite
  ReserveTestTask write2(writer.ReserveWrite(2));
  dispatcher.Post(write2);
  dispatcher.RunUntilStalled();
  ASSERT_TRUE(write2.result().has_value() && write2.result()->ok());
  (*write2.result())->payload()[0] = std::byte{3};
  (*write2.result())->payload()[1] = std::byte{4};
  PW_TEST_EXPECT_OK((*write2.result())->Commit());
  write2.Deregister();

  // Close writer
  WriteTestTask close_task(writer.Finish());
  dispatcher.Post(close_task);
  dispatcher.RunUntilStalled();
  ASSERT_TRUE(close_task.result().has_value() && close_task.result()->ok());
  close_task.Deregister();

  // Read response
  ReadTestTask resp_task(std::move(response_future));
  dispatcher.Post(resp_task);
  dispatcher.RunUntilStalled();
  EXPECT_FALSE(resp_task.result().has_value());

  pw::Buf resp_data = pw::Buf::Allocate(allocator, 4);
  std::memset(resp_data.data(), 0xEE, 4);
  auto framed_resp = internal::PacketFramer::FrameResponsePacket(
      allocator, /*call_id=*/1, pw::ConstByteSpan(resp_data));
  ASSERT_TRUE(framed_resp.ok());
  raw_conn->SetNextRead(std::move(*framed_resp));

  dispatcher.RunUntilStalled();
  ASSERT_TRUE(resp_task.result().has_value() && resp_task.result()->ok());
  EXPECT_EQ(resp_task.result()->value().size(), 4u);
  resp_task.Deregister();
}

TEST(ClientCallTest, BidirectionalStreamingConcurrentReadWrite) {
  allocator::test::AllocatorForTest<16384> allocator;
  async2::DispatcherForTest dispatcher;
  auto [conn, raw_conn] = test::MakeMockConnection(allocator);

  test::ScopedClient client(SetupClient(allocator, dispatcher, raw_conn, conn),
                            dispatcher);
  TestServiceClient test_service(client, 100u);

  auto bidi_call =
      test_service.CallBidiStream<pw::ConstBuf, pw::ConstBuf>(500u);

  StreamCallTestTask call_task(std::move(bidi_call));
  dispatcher.Post(call_task);
  dispatcher.RunUntilStalled();

  ASSERT_TRUE(call_task.result().has_value() && call_task.result()->ok());
  auto [writer, reader] = std::move(**call_task.result());
  call_task.Deregister();

  // Write from client via ReserveWrite
  ReserveTestTask write_task(writer.ReserveWrite(2));
  dispatcher.Post(write_task);
  dispatcher.RunUntilStalled();
  ASSERT_TRUE(write_task.result().has_value() && write_task.result()->ok());
  (*write_task.result())->payload()[0] = std::byte{10};
  (*write_task.result())->payload()[1] = std::byte{20};
  PW_TEST_EXPECT_OK((*write_task.result())->Commit());
  write_task.Deregister();

  // Read server message
  ReadTestTask read_task(reader.Read());
  dispatcher.Post(read_task);
  std::byte srv_msg[3] = {std::byte{1}, std::byte{2}, std::byte{3}};
  auto pkt =
      internal::PacketFramer::FrameServerMessagePacket(allocator, 1u, srv_msg);
  ASSERT_TRUE(pkt.ok());
  raw_conn->SetNextRead(std::move(*pkt));
  dispatcher.RunUntilStalled();
  ASSERT_TRUE(read_task.result().has_value() && read_task.result()->ok());
  EXPECT_EQ(read_task.result()->value().size(), 3u);
  read_task.Deregister();

  // Client half-close
  WriteTestTask close_task(writer.Finish());
  dispatcher.Post(close_task);
  dispatcher.RunUntilStalled();
  ASSERT_TRUE(close_task.result().has_value() && close_task.result()->ok());
  close_task.Deregister();

  // Server sends StreamEnd
  ReadTestTask end_read_task(reader.Read());
  dispatcher.Post(end_read_task);
  auto end_pkt = internal::PacketFramer::FrameServerFinishPacket(allocator, 1u);
  ASSERT_TRUE(end_pkt.ok());
  raw_conn->SetNextRead(std::move(*end_pkt));
  dispatcher.RunUntilStalled();
  ASSERT_TRUE(end_read_task.result().has_value());
  EXPECT_EQ(end_read_task.result()->status(), Status::OutOfRange());
  end_read_task.Deregister();
}

// The server always terminates the RPC when it ends its stream, so a client
// writer that is still open is closed too.
TEST(ClientCallTest, BidirectionalServerStreamEndClosesClientWriter) {
  allocator::test::AllocatorForTest<16384> allocator;
  async2::DispatcherForTest dispatcher;
  auto [conn, raw_conn] = test::MakeMockConnection(allocator);

  test::ScopedClient client(SetupClient(allocator, dispatcher, raw_conn, conn),
                            dispatcher);
  TestServiceClient test_service(client, 100u);

  StreamCallTestTask call_task(
      test_service.CallBidiStream<pw::ConstBuf, pw::ConstBuf>(500u));
  dispatcher.Post(call_task);
  dispatcher.RunUntilStalled();

  ASSERT_TRUE(call_task.result().has_value() && call_task.result()->ok());
  auto [writer, reader] = std::move(**call_task.result());
  call_task.Deregister();

  ReadTestTask end_read_task(reader.Read());
  dispatcher.Post(end_read_task);
  auto end_pkt = internal::PacketFramer::FrameServerFinishPacket(allocator, 1u);
  ASSERT_TRUE(end_pkt.ok());
  raw_conn->SetNextRead(std::move(*end_pkt));
  dispatcher.RunUntilStalled();
  ASSERT_TRUE(end_read_task.result().has_value());
  EXPECT_EQ(end_read_task.result()->status(), Status::OutOfRange());
  end_read_task.Deregister();

  const size_t commits_before = raw_conn->commit_count();

  ReserveTestTask write_task(writer.ReserveWrite(2));
  dispatcher.Post(write_task);
  dispatcher.RunUntilStalled();
  ASSERT_TRUE(write_task.result().has_value());
  EXPECT_FALSE(write_task.result()->ok());
  write_task.Deregister();

  // The client does not end a stream for a call that is already over.
  {
    auto dropped = std::move(writer);
  }
  dispatcher.RunUntilStalled();
  EXPECT_EQ(raw_conn->commit_count(), commits_before);
}

// A server message that also ends the server's stream without terminating the
// RPC (0x0B) delivers the message, but the reader stays open until the terminal
// packet. The C++ server does not send this packet, but it is valid on the
// wire.
TEST(ClientCallTest, ServerFinalMessageWaitsForTerminalPacket) {
  allocator::test::AllocatorForTest<16384> allocator;
  async2::DispatcherForTest dispatcher;
  auto [conn, raw_conn] = test::MakeMockConnection(allocator);

  test::ScopedClient client(SetupClient(allocator, dispatcher, raw_conn, conn),
                            dispatcher);
  TestServiceClient test_service(client, 100u);

  StreamCallTestTask call_task(
      test_service.CallServerStream<pw::ConstBuf, pw::ConstBuf>(
          300u, pw::ConstBuf()));
  dispatcher.Post(call_task);
  dispatcher.RunUntilStalled();

  ASSERT_TRUE(call_task.result().has_value() && call_task.result()->ok());
  Reader<pw::ConstBuf> reader = std::move(**call_task.result());
  call_task.Deregister();

  std::byte chunk[3] = {std::byte{1}, std::byte{2}, std::byte{3}};
  auto pkt = FrameServerPacket(allocator,
                               internal::PacketType::Make<flags::kServer,
                                                          flags::kHasPayload,
                                                          flags::kStreamEnd>(),
                               1u,
                               chunk);
  ASSERT_TRUE(pkt.ok());

  ReadTestTask read_task(reader.Read());
  dispatcher.Post(read_task);
  raw_conn->SetNextRead(std::move(*pkt));
  dispatcher.RunUntilStalled();
  ASSERT_TRUE(read_task.result().has_value() && read_task.result()->ok());
  EXPECT_EQ(read_task.result()->value().size(), sizeof(chunk));
  read_task.Deregister();

  ReadTestTask end_read_task(reader.Read());
  dispatcher.Post(end_read_task);
  dispatcher.RunUntilStalled();
  EXPECT_FALSE(end_read_task.result().has_value());

  auto finish = internal::PacketFramer::FrameServerFinishPacket(allocator, 1u);
  ASSERT_TRUE(finish.ok());
  raw_conn->SetNextRead(std::move(*finish));
  dispatcher.RunUntilStalled();
  ASSERT_TRUE(end_read_task.result().has_value());
  EXPECT_EQ(end_read_task.result()->status(), Status::OutOfRange());
  end_read_task.Deregister();
}

// After a server half-close (0x09), the client drops any further messages and
// reports the status from the terminal packet that follows.
TEST(ClientCallTest, ServerStreamEndThenErrorReportsError) {
  allocator::test::AllocatorForTest<16384> allocator;
  async2::DispatcherForTest dispatcher;
  auto [conn, raw_conn] = test::MakeMockConnection(allocator);

  test::ScopedClient client(SetupClient(allocator, dispatcher, raw_conn, conn),
                            dispatcher);
  TestServiceClient test_service(client, 100u);

  StreamCallTestTask call_task(
      test_service.CallBidiStream<pw::ConstBuf, pw::ConstBuf>(500u));
  dispatcher.Post(call_task);
  dispatcher.RunUntilStalled();

  ASSERT_TRUE(call_task.result().has_value() && call_task.result()->ok());
  auto [writer, reader] = std::move(**call_task.result());
  call_task.Deregister();
  const uint32_t call_id = LastSentCallId(*raw_conn);

  ReadTestTask read_task(reader.Read());
  dispatcher.Post(read_task);

  auto stream_end = FrameServerPacket(
      allocator,
      internal::PacketType::Make<flags::kServer, flags::kStreamEnd>(),
      call_id);
  ASSERT_TRUE(stream_end.ok());
  raw_conn->SetNextRead(std::move(*stream_end));
  dispatcher.RunUntilStalled();
  EXPECT_FALSE(read_task.result().has_value());

  std::byte chunk[1] = {std::byte{1}};
  auto late_message = internal::PacketFramer::FrameServerMessagePacket(
      allocator, call_id, chunk);
  ASSERT_TRUE(late_message.ok());
  raw_conn->SetNextRead(std::move(*late_message));
  dispatcher.RunUntilStalled();
  EXPECT_FALSE(read_task.result().has_value());

  // The client's stream is still open until the RPC ends.
  ReserveTestTask write_task(writer.ReserveWrite(1));
  dispatcher.Post(write_task);
  dispatcher.RunUntilStalled();
  ASSERT_TRUE(write_task.result().has_value());
  PW_TEST_EXPECT_OK(write_task.result()->status());
  write_task.Deregister();

  auto error = internal::PacketFramer::FrameServerErrorPacket(
      allocator, call_id, internal::ServerError::kInternal);
  ASSERT_TRUE(error.ok());
  raw_conn->SetNextRead(std::move(*error));
  dispatcher.RunUntilStalled();
  ASSERT_TRUE(read_task.result().has_value());
  EXPECT_EQ(read_task.result()->status(), Status::Internal());
  read_task.Deregister();
}

// A unary call takes exactly one packet that carries the response and ends the
// RPC. Any other server packet means the server thinks the method streams its
// responses, so the client fails the call rather than taking a streamed message
// as the response. If the packet did not already end the RPC, the client tells
// the server why.
TEST(ClientCallTest, UnaryCallRejectsResponseThatIsNotSingleResponse) {
  allocator::test::AllocatorForTest<16384> allocator;
  async2::DispatcherForTest dispatcher;
  auto [conn, raw_conn] = test::MakeMockConnection(allocator);

  test::ScopedClient client(SetupClient(allocator, dispatcher, raw_conn, conn),
                            dispatcher);
  TestServiceClient test_service(client, 100u);

  constexpr std::byte kPayload[2] = {std::byte{1}, std::byte{2}};
  for (internal::PacketType type : kNotSingleResponse) {
    UnaryTestTask task(test_service.CallUnary<pw::ConstBuf, pw::ConstBuf>(
        200u, pw::ConstBuf()));
    dispatcher.Post(task);
    dispatcher.RunUntilStalled();
    const uint32_t call_id = LastSentCallId(*raw_conn);

    auto pkt = FrameServerPacket(
        allocator,
        type,
        call_id,
        type.has_payload() ? pw::ConstByteSpan(kPayload) : pw::ConstByteSpan());
    ASSERT_TRUE(pkt.ok());
    raw_conn->SetNextRead(std::move(*pkt));
    dispatcher.RunUntilStalled();

    ASSERT_TRUE(task.result().has_value());
    EXPECT_EQ(task.result()->status(), Status::FailedPrecondition());
    if (type ==
        internal::PacketType::Make<flags::kServer, flags::kOkTerminal>()) {
      auto last = internal::InboundPacket::Decode(raw_conn->last_written_buf());
      ASSERT_TRUE(last.ok());
      EXPECT_EQ(last->type(),
                (internal::PacketType::Make<flags::kStart,
                                            flags::kHasPayload,
                                            flags::kStreamEnd>()));
      EXPECT_EQ(last->call_id(), call_id);
    } else {
      ExpectSentMethodTypeMismatch(*raw_conn, call_id);
    }
    task.Deregister();
  }
}

TEST(ClientCallTest, ClientStreamingCallRejectsStreamedResponse) {
  allocator::test::AllocatorForTest<16384> allocator;
  async2::DispatcherForTest dispatcher;
  auto [conn, raw_conn] = test::MakeMockConnection(allocator);

  test::ScopedClient client(SetupClient(allocator, dispatcher, raw_conn, conn),
                            dispatcher);
  TestServiceClient test_service(client, 100u);

  StreamCallTestTask call_task(
      test_service.CallClientStream<pw::ConstBuf, pw::ConstBuf>(400u));
  dispatcher.Post(call_task);
  dispatcher.RunUntilStalled();
  ASSERT_TRUE(call_task.result().has_value() && call_task.result()->ok());
  auto [writer, response_future] = std::move(**call_task.result());
  call_task.Deregister();
  const uint32_t call_id = LastSentCallId(*raw_conn);

  ReadTestTask resp_task(std::move(response_future));
  dispatcher.Post(resp_task);

  std::byte srv_msg[3] = {std::byte{1}, std::byte{2}, std::byte{3}};
  auto pkt = internal::PacketFramer::FrameServerMessagePacket(
      allocator, call_id, srv_msg);
  ASSERT_TRUE(pkt.ok());
  raw_conn->SetNextRead(std::move(*pkt));
  dispatcher.RunUntilStalled();

  ASSERT_TRUE(resp_task.result().has_value());
  EXPECT_EQ(resp_task.result()->status(), Status::FailedPrecondition());
  ExpectSentMethodTypeMismatch(*raw_conn, call_id);
  resp_task.Deregister();

  // The call is over, so the client's stream is closed too.
  ReserveTestTask write_task(writer.ReserveWrite(2));
  dispatcher.Post(write_task);
  dispatcher.RunUntilStalled();
  ASSERT_TRUE(write_task.result().has_value());
  EXPECT_EQ(write_task.result()->status(), Status::FailedPrecondition());
  write_task.Deregister();
}

// A streaming call accepts a single terminal response: it is unambiguously a
// stream of one message, which a streaming server may send.
TEST(ClientCallTest, ServerStreamingCallAcceptsSingleResponse) {
  allocator::test::AllocatorForTest<16384> allocator;
  async2::DispatcherForTest dispatcher;
  auto [conn, raw_conn] = test::MakeMockConnection(allocator);

  test::ScopedClient client(SetupClient(allocator, dispatcher, raw_conn, conn),
                            dispatcher);
  TestServiceClient test_service(client, 100u);

  StreamCallTestTask call_task(
      test_service.CallServerStream<pw::ConstBuf, pw::ConstBuf>(
          300u, pw::ConstBuf()));
  dispatcher.Post(call_task);
  dispatcher.RunUntilStalled();
  ASSERT_TRUE(call_task.result().has_value() && call_task.result()->ok());
  Reader<pw::ConstBuf> reader = std::move(**call_task.result());
  call_task.Deregister();
  const uint32_t call_id = LastSentCallId(*raw_conn);

  ReadTestTask read_task(reader.Read());
  dispatcher.Post(read_task);
  std::byte resp[2] = {std::byte{7}, std::byte{8}};
  auto pkt =
      internal::PacketFramer::FrameResponsePacket(allocator, call_id, resp);
  ASSERT_TRUE(pkt.ok());
  raw_conn->SetNextRead(std::move(*pkt));
  dispatcher.RunUntilStalled();
  ASSERT_TRUE(read_task.result().has_value() && read_task.result()->ok());
  EXPECT_EQ(read_task.result()->value().size(), sizeof(resp));
  read_task.Deregister();

  ReadTestTask end_read_task(reader.Read());
  dispatcher.Post(end_read_task);
  dispatcher.RunUntilStalled();
  ASSERT_TRUE(end_read_task.result().has_value());
  EXPECT_EQ(end_read_task.result()->status(), Status::OutOfRange());
  end_read_task.Deregister();

  // Nothing was reported to the server.
  EXPECT_EQ(LastSentCallId(*raw_conn), call_id);
  auto last = internal::InboundPacket::Decode(raw_conn->last_written_buf());
  ASSERT_TRUE(last.ok());
  EXPECT_EQ(last->type(),
            (internal::PacketType::
                 Make<flags::kStart, flags::kHasPayload, flags::kStreamEnd>()));
}

TEST(ClientCallTest, ServerErrorResolvesClientFutureWithStatus) {
  allocator::test::AllocatorForTest<16384> allocator;
  async2::DispatcherForTest dispatcher;
  auto [conn, raw_conn] = test::MakeMockConnection(allocator);

  test::ScopedClient client(SetupClient(allocator, dispatcher, raw_conn, conn),
                            dispatcher);
  TestServiceClient test_service(client, 100u);

  pw::Buf req_buf = pw::Buf::Allocate(allocator, 2);
  auto unary_call = test_service.CallUnary<pw::ConstBuf, pw::ConstBuf>(
      600u, pw::ConstBuf(std::move(req_buf)));

  UnaryTestTask task(std::move(unary_call));
  dispatcher.Post(task);
  dispatcher.RunUntilStalled();

  // Server emits Error packet with ServerError::kUnknownMethod
  auto err_pkt = internal::PacketFramer::FrameServerErrorPacket(
      allocator,
      /*call_id=*/1,
      internal::ServerError::kUnknownMethod);
  ASSERT_TRUE(err_pkt.ok());
  raw_conn->SetNextRead(std::move(*err_pkt));

  dispatcher.RunUntilStalled();
  ASSERT_TRUE(task.result().has_value());
  EXPECT_FALSE(task.result()->ok());
  EXPECT_EQ(task.result()->status(), Status::NotFound());

  task.Deregister();
}

TEST(ClientCallTest, ClientDropHandleEndsClientStream) {
  allocator::test::AllocatorForTest<16384> allocator;
  async2::DispatcherForTest dispatcher;
  auto [conn, raw_conn] = test::MakeMockConnection(allocator);

  test::ScopedClient client(SetupClient(allocator, dispatcher, raw_conn, conn),
                            dispatcher);
  TestServiceClient test_service(client, 100u);

  size_t initial_commits = raw_conn->commit_count();

  // Keep the response pending: dropping it would cancel the call.
  std::optional<ReadFuture<pw::ConstBuf>> response;
  {
    auto client_stream_call =
        test_service.CallClientStream<pw::ConstBuf, pw::ConstBuf>(700u);

    StreamCallTestTask call_task(std::move(client_stream_call));
    dispatcher.Post(call_task);
    dispatcher.RunUntilStalled();

    ASSERT_TRUE(call_task.result().has_value() && call_task.result()->ok());
    ClientStreamCall<pw::ConstBuf, pw::ConstBuf> stream =
        std::move(**call_task.result());
    call_task.Deregister();
    response.emplace(std::move(stream.response));

    EXPECT_EQ(raw_conn->commit_count(), initial_commits + 1u);  // +1 request
    // stream.writer is destroyed at end of scope without Close()
  }

  // Destruction of an unclosed Writer ends the outbound stream normally.
  EXPECT_EQ(raw_conn->commit_count(), initial_commits + 2u);
  auto decode_end =
      internal::InboundPacket::Decode(raw_conn->last_written_buf());
  ASSERT_TRUE(decode_end.ok());
  EXPECT_EQ(decode_end->type(),
            internal::PacketType::Make<flags::kStreamEnd>());

  // Dropping the response before the server answers cancels the call.
  response.reset();
  dispatcher.RunUntilStalled();
  EXPECT_EQ(raw_conn->commit_count(), initial_commits + 3u);
  auto decode_cancel =
      internal::InboundPacket::Decode(raw_conn->last_written_buf());
  ASSERT_TRUE(decode_cancel.ok());
  EXPECT_EQ(decode_cancel->type(),
            internal::PacketType::Make<flags::kErrorTerminal>());
  EXPECT_EQ(decode_cancel->client_error(), internal::ClientError::kCancelled);
}

// Checks that the most recent packet the client sent cancels `call_id`.
void ExpectSentCancel(test::MockConnection& raw_conn, uint32_t call_id) {
  auto decoded = internal::InboundPacket::Decode(raw_conn.last_written_buf());
  ASSERT_TRUE(decoded.ok());
  EXPECT_EQ(decoded->type(),
            internal::PacketType::Make<flags::kErrorTerminal>());
  EXPECT_EQ(decoded->call_id(), call_id);
  EXPECT_EQ(decoded->client_error(), internal::ClientError::kCancelled);
}

// Cancelling through the writer completes the call locally, so a pending read
// resolves instead of waiting for a server reply that will never come.
TEST(ClientCallTest, BidiWriterCancelResolvesPendingRead) {
  allocator::test::AllocatorForTest<16384> allocator;
  async2::DispatcherForTest dispatcher;
  auto [conn, raw_conn] = test::MakeMockConnection(allocator);

  test::ScopedClient client(SetupClient(allocator, dispatcher, raw_conn, conn),
                            dispatcher);
  TestServiceClient test_service(client, 100u);

  StreamCallTestTask call_task(
      test_service.CallBidiStream<pw::ConstBuf, pw::ConstBuf>(500u));
  dispatcher.Post(call_task);
  dispatcher.RunUntilStalled();
  ASSERT_TRUE(call_task.result().has_value() && call_task.result()->ok());
  auto [writer, reader] = std::move(**call_task.result());
  call_task.Deregister();

  ReadTestTask read_task(reader.Read());
  dispatcher.Post(read_task);
  dispatcher.RunUntilStalled();
  EXPECT_FALSE(read_task.result().has_value());

  const size_t commits_before = raw_conn->commit_count();
  writer.Cancel();
  dispatcher.RunUntilStalled();

  ASSERT_TRUE(read_task.result().has_value());
  EXPECT_EQ(read_task.result()->status(), Status::Cancelled());
  EXPECT_EQ(raw_conn->commit_count(), commits_before + 1u);
  ExpectSentCancel(*raw_conn, 1u);
  read_task.Deregister();

  // The call is over, so dropping the handles sends nothing more.
  {
    auto dropped_writer = std::move(writer);
    auto dropped_reader = std::move(reader);
  }
  dispatcher.RunUntilStalled();
  EXPECT_EQ(raw_conn->commit_count(), commits_before + 1u);
}

TEST(ClientCallTest, ClientStreamWriterCancelResolvesResponse) {
  allocator::test::AllocatorForTest<16384> allocator;
  async2::DispatcherForTest dispatcher;
  auto [conn, raw_conn] = test::MakeMockConnection(allocator);

  test::ScopedClient client(SetupClient(allocator, dispatcher, raw_conn, conn),
                            dispatcher);
  TestServiceClient test_service(client, 100u);

  StreamCallTestTask call_task(
      test_service.CallClientStream<pw::ConstBuf, pw::ConstBuf>(400u));
  dispatcher.Post(call_task);
  dispatcher.RunUntilStalled();
  ASSERT_TRUE(call_task.result().has_value() && call_task.result()->ok());
  auto [writer, response_future] = std::move(**call_task.result());
  call_task.Deregister();

  ReadTestTask resp_task(std::move(response_future));
  dispatcher.Post(resp_task);
  dispatcher.RunUntilStalled();
  EXPECT_FALSE(resp_task.result().has_value());

  writer.Cancel();
  dispatcher.RunUntilStalled();

  ASSERT_TRUE(resp_task.result().has_value());
  EXPECT_EQ(resp_task.result()->status(), Status::Cancelled());
  ExpectSentCancel(*raw_conn, 1u);
  resp_task.Deregister();
}

// A client that drops a server stream's reader before the stream ends has no
// further interest in the call, so the server is told to stop.
TEST(ClientCallTest, DroppingServerStreamReaderCancelsCall) {
  allocator::test::AllocatorForTest<16384> allocator;
  async2::DispatcherForTest dispatcher;
  auto [conn, raw_conn] = test::MakeMockConnection(allocator);

  test::ScopedClient client(SetupClient(allocator, dispatcher, raw_conn, conn),
                            dispatcher);
  TestServiceClient test_service(client, 100u);

  StreamCallTestTask call_task(
      test_service.CallServerStream<pw::ConstBuf, pw::ConstBuf>(
          300u, pw::ConstBuf()));
  dispatcher.Post(call_task);
  dispatcher.RunUntilStalled();
  ASSERT_TRUE(call_task.result().has_value() && call_task.result()->ok());
  std::optional<Reader<pw::ConstBuf>> reader(std::move(**call_task.result()));
  call_task.Deregister();

  const size_t commits_before = raw_conn->commit_count();
  reader.reset();
  dispatcher.RunUntilStalled();

  EXPECT_EQ(raw_conn->commit_count(), commits_before + 1u);
  ExpectSentCancel(*raw_conn, 1u);
}

// Dropping a reader after the stream ended normally sends nothing.
TEST(ClientCallTest, DroppingFinishedServerStreamReaderSendsNothing) {
  allocator::test::AllocatorForTest<16384> allocator;
  async2::DispatcherForTest dispatcher;
  auto [conn, raw_conn] = test::MakeMockConnection(allocator);

  test::ScopedClient client(SetupClient(allocator, dispatcher, raw_conn, conn),
                            dispatcher);
  TestServiceClient test_service(client, 100u);

  StreamCallTestTask call_task(
      test_service.CallServerStream<pw::ConstBuf, pw::ConstBuf>(
          300u, pw::ConstBuf()));
  dispatcher.Post(call_task);
  dispatcher.RunUntilStalled();
  ASSERT_TRUE(call_task.result().has_value() && call_task.result()->ok());
  std::optional<Reader<pw::ConstBuf>> reader(std::move(**call_task.result()));
  call_task.Deregister();

  ReadTestTask end_read(reader->Read());
  dispatcher.Post(end_read);
  auto end_pkt = internal::PacketFramer::FrameServerFinishPacket(allocator, 1u);
  ASSERT_TRUE(end_pkt.ok());
  raw_conn->SetNextRead(std::move(*end_pkt));
  dispatcher.RunUntilStalled();
  ASSERT_TRUE(end_read.result().has_value());
  EXPECT_EQ(end_read.result()->status(), Status::OutOfRange());
  end_read.Deregister();

  const size_t commits_before = raw_conn->commit_count();
  reader.reset();
  dispatcher.RunUntilStalled();
  EXPECT_EQ(raw_conn->commit_count(), commits_before);
}

TEST(ClientCallTest, ReaderCancelResolvesPendingReadAndCancelsCall) {
  allocator::test::AllocatorForTest<16384> allocator;
  async2::DispatcherForTest dispatcher;
  auto [conn, raw_conn] = test::MakeMockConnection(allocator);

  test::ScopedClient client(SetupClient(allocator, dispatcher, raw_conn, conn),
                            dispatcher);
  TestServiceClient test_service(client, 100u);

  StreamCallTestTask call_task(
      test_service.CallServerStream<pw::ConstBuf, pw::ConstBuf>(
          300u, pw::ConstBuf()));
  dispatcher.Post(call_task);
  dispatcher.RunUntilStalled();
  ASSERT_TRUE(call_task.result().has_value() && call_task.result()->ok());
  std::optional<Reader<pw::ConstBuf>> reader(std::move(**call_task.result()));
  call_task.Deregister();

  ReadTestTask read_task(reader->Read());
  dispatcher.Post(read_task);
  dispatcher.RunUntilStalled();
  EXPECT_FALSE(read_task.result().has_value());

  const size_t commits_before = raw_conn->commit_count();
  reader->Cancel();
  dispatcher.RunUntilStalled();

  ASSERT_TRUE(read_task.result().has_value());
  EXPECT_EQ(read_task.result()->status(), Status::Cancelled());
  EXPECT_EQ(raw_conn->commit_count(), commits_before + 1u);
  ExpectSentCancel(*raw_conn, 1u);
  read_task.Deregister();

  // A second cancel, or dropping the reader, sends nothing more.
  reader->Cancel();
  reader.reset();
  dispatcher.RunUntilStalled();
  EXPECT_EQ(raw_conn->commit_count(), commits_before + 1u);
}

// Dropping a unary call after its request was sent cancels the call.
TEST(ClientCallTest, DroppingUnaryCallAfterRequestCancelsCall) {
  allocator::test::AllocatorForTest<16384> allocator;
  async2::DispatcherForTest dispatcher;
  auto [conn, raw_conn] = test::MakeMockConnection(allocator);

  test::ScopedClient client(SetupClient(allocator, dispatcher, raw_conn, conn),
                            dispatcher);
  TestServiceClient test_service(client, 100u);

  size_t commits_after_request = 0;
  {
    UnaryTestTask task(test_service.CallUnary<pw::ConstBuf, pw::ConstBuf>(
        200u, pw::ConstBuf()));
    dispatcher.Post(task);
    dispatcher.RunUntilStalled();
    EXPECT_FALSE(task.result().has_value());
    commits_after_request = raw_conn->commit_count();
    task.Deregister();
  }
  dispatcher.RunUntilStalled();

  EXPECT_EQ(raw_conn->commit_count(), commits_after_request + 1u);
  ExpectSentCancel(*raw_conn, 1u);
}

TEST(ClientCallTest, ClientStreamSendsStartStreamFirst) {
  allocator::test::AllocatorForTest<16384> allocator;
  async2::DispatcherForTest dispatcher;

  auto [conn, raw_conn] = test::MakeMockConnection(allocator);

  ClientConnectTask connect_task(Client::Create(dispatcher, allocator, conn));
  dispatcher.Post(connect_task);
  dispatcher.RunUntilStalled();

  pw::Buf hs_resp =
      pw::Buf::Allocate(allocator, internal::HandshakePacket::kWireSizeBytes);
  auto hs_pkt =
      internal::HandshakePacket(internal::HandshakePacket::Type::kSynAck);
  auto enc_res = hs_pkt.Encode(std::move(hs_resp));
  ASSERT_TRUE(enc_res.ok());
  raw_conn->SetNextRead(std::move(enc_res.value()));

  dispatcher.RunUntilStalled();
  ASSERT_TRUE(connect_task.client().has_value() && connect_task.client()->ok());
  test::ScopedClient client(std::move(**connect_task.client()), dispatcher);
  connect_task.Deregister();

  TestServiceClient test_service(client, 100u);
  auto client_stream_fut =
      test_service.CallClientStream<pw::ConstBuf, pw::ConstBuf>(200u);

  struct Task : public async2::Task {
    ClientStreamFuture<pw::ConstBuf, pw::ConstBuf> fut;
    std::optional<Result<ClientStreamCall<pw::ConstBuf, pw::ConstBuf>>> res;
    Task(ClientStreamFuture<pw::ConstBuf, pw::ConstBuf> f)
        : fut(std::move(f)) {}
    async2::Poll<> DoPend(async2::Context& cx) override {
      auto poll = fut.Pend(cx);
      if (poll.IsPending()) {
        return async2::Pending();
      }
      res = std::move(*poll);
      return async2::Ready();
    }
  } task(std::move(client_stream_fut));

  dispatcher.Post(task);
  dispatcher.RunUntilStalled();

  // Pending RequestFuture transmits the initial payload-less START packet.
  EXPECT_GE(raw_conn->commit_count(), 3u);  // 2 handshake + 1 start
  auto decode_req =
      internal::InboundPacket::Decode(raw_conn->last_written_buf());
  ASSERT_TRUE(decode_req.ok());
  EXPECT_EQ(decode_req->type(), internal::PacketType::Make<flags::kStart>());
  EXPECT_FALSE(decode_req->type().has_payload());
  EXPECT_EQ(decode_req->type().close_mode(), internal::CloseMode::kOpen);
  EXPECT_EQ(decode_req->service_id(), 100u);
  EXPECT_EQ(decode_req->method_id(), 200u);
  EXPECT_EQ(decode_req->payload().size(), 0u);

  ASSERT_TRUE(task.res.has_value() && task.res->ok());
  auto [writer, reader] = std::move(**task.res);

  // Subsequent Write transmits a message packet (not a START packet)
  ReserveTestTask write_task(writer.ReserveWrite(2));
  dispatcher.Post(write_task);
  dispatcher.RunUntilStalled();
  ASSERT_TRUE(write_task.result().has_value() && write_task.result()->ok());
  (*write_task.result())->payload()[0] = std::byte(1);
  (*write_task.result())->payload()[1] = std::byte(2);
  PW_TEST_EXPECT_OK((*write_task.result())->Commit());

  auto decode_msg =
      internal::InboundPacket::Decode(raw_conn->last_written_buf());
  ASSERT_TRUE(decode_msg.ok());
  EXPECT_EQ(decode_msg->type(),
            internal::PacketType::Make<flags::kHasPayload>());
  EXPECT_EQ(decode_msg->payload().size(), 2u);

  task.Deregister();
  write_task.Deregister();
}

}  // namespace
}  // namespace pw::rpc2