A deadline for the coroutine receive, contract-wide (#130)

CHANGELOG.md

@@ -45,6 +45,22 @@ policy in [docs/versioning.md](docs/versioning.md).
 
 ### Added
 
+- **A deadline for the coroutine receive** (#130).
+  `co_await stream.Receive(timeout)` on `AsyncEventStream` (and
+  `ReceiveMessage(socket, timeout)` for the raw awaitable) resolves with
+  `Error::Timeout` (code `TimeoutError`) when nothing arrives in time — the
+  completion-driven twin of the blocking `Receive(timeout)`, with the same
+  rule: a timeout is not terminal, the receive slot is released, and a
+  message the wire delivers after the deadline waits for the next await. A
+  `Detached` serve loop can now bound every await instead of parking its
+  frame forever on a peer that never sends. One layer down it is
+  `WebSocket::ReceiveAsync(timeout, callback)` — a refusing default (the
+  async family is opt-in), implemented by both transports (a timer on the
+  connection's executor for Beast; a watchdog for the executor-less
+  in-memory pair) and passed through by `JsonRpcStreamSocket`, with the
+  timeout/delivery race settled exactly once by a park generation. The
+  shared WebSocket contract suite grows four deadline cases every
+  implementation runs, in-repo and out.
 - **Servers validate intEnum membership** (#109). String enums already
   failed request validation outside the modeled value set; intEnum members
   were accepted silently. They now produce the same suite-exact

docs/production-guide.md

@@ -461,6 +461,15 @@ code holding a raw session; a hand-rolled socket (a test fake, an adapter
 over another WebSocket library) owes its callers a wait that actually ends
 and an `Error::Timeout` when it does.
 
+The completion-driven half has the same deadline (#130):
+`co_await stream.Receive(std::chrono::seconds(2))` on an `AsyncEventStream`
+resolves with the same four outcomes and the same rule — a timeout releases
+the receive slot and leaves the session usable, and an event the wire
+delivers after the deadline waits for the next await. One layer down it is
+`WebSocket::ReceiveAsync(timeout, callback)`, a refusing default rather than
+a pure virtual (the async family is opt-in), with the shared contract suite
+holding every `SupportsAsync()` implementation to it.
+
 Not every `ClientConfig` knob reaches a streaming dial — the upgrade GET is
 not a unary request:
 

docs/server-guide.md

@@ -140,6 +140,12 @@ while (true) {
 }
 ```
 
+The completion-driven seam has the same watchdog (#130): a `Detached` loop bounds its
+await with `co_await stream.Receive(std::chrono::seconds(1))` and gets the same
+`TimeoutError` on a session that keeps serving — the loop above works verbatim as a
+coroutine. `ReceiveMessage(socket, timeout)` bounds the raw pre-stream await the same
+way (the jsonRpc2 opening-envelope read).
+
 Whatever method the operation models, upgrades are always GET — a WebSocket upgrade is a
 GET on the wire, and the generated routes register accordingly: on the modeled URI for
 the binding protocols, on the shared `/` endpoint for jsonRpc2 (ADR-0023), whose opening

examples/bazel-consumer/async_acceptance_test.cc

@@ -101,6 +101,50 @@ Detached Serve(smithy::server::SessionRegistry<Message>& registry, std::string i
   stream.Close();
 }
 
+// The #130 watchdog, out of tree and over a real socket: a hand-written
+// Detached loop — the deadline's stated audience — bounds its await
+// against a peer that says nothing unsolicited, treats the timeout as a
+// tick rather than an end, and the same coroutine still serves the
+// message that arrives after one.
+TEST(AsyncAcceptanceTest, AnAwaitedReceiveDeadlineTicksWithoutEndingTheSession) {
+  BeastServerTransport::Options options;
+  options.on_websocket = [](const HttpRequest&, WebSocket& socket) {
+    while (true) {
+      auto message = socket.Receive();
+      if (!message.ok() || !message->has_value()) return;
+      if (!socket.Send(Event("echo", "echo:" + (*message)->payload.ToString())).ok()) return;
+    }
+  };
+  BeastServerTransport server(options);
+  ASSERT_TRUE(server.Start(NotFound).ok());
+  auto dialed = BeastWebSocketClient::Dial({.host = "127.0.0.1", .port = server.port()});
+  ASSERT_TRUE(dialed.ok()) << dialed.error().message();
+
+  std::promise<std::string> first_timeout_code;
+  std::promise<std::string> echoed;
+  [](std::shared_ptr<WebSocket> socket, std::promise<std::string>* timeout_code,
+     std::promise<std::string>* echoed) -> Detached {
+    AsyncEventStream<Message, Message> stream(std::move(socket), Identity, Identity);
+    // Quiet wire: the bounded await resolves with the timeout, terminally
+    // for nothing — the session and the loop both continue.
+    auto nothing = co_await stream.Receive(std::chrono::milliseconds(75));
+    timeout_code->set_value(nothing.ok() ? "ok" : nothing.error().code());
+    (void)co_await stream.Send(Event("chat", "after the deadline"));
+    auto echo = co_await stream.Receive(std::chrono::seconds(10));
+    echoed->set_value(echo.ok() && echo->has_value() ? (**echo).payload.ToString() : "<no echo>");
+    stream.Close();
+  }(*dialed, &first_timeout_code, &echoed);
+
+  auto code = first_timeout_code.get_future();
+  ASSERT_EQ(code.wait_for(std::chrono::seconds(5)), std::future_status::ready);
+  EXPECT_EQ(code.get(), "TimeoutError");
+
+  auto reply = echoed.get_future();
+  ASSERT_EQ(reply.wait_for(std::chrono::seconds(5)), std::future_status::ready);
+  EXPECT_EQ(reply.get(), "echo:after the deadline");
+  server.Stop();
+}
+
 TEST(AsyncAcceptanceTest, ThreeSessionsShareOneHandlerThreadAndAFanOutRegistry) {
   // Declared before the transport on purpose: sessions reference the
   // registry from their coroutines, so it must outlive them.

examples/bazel-consumer/model/todo.smithy

@@ -22,6 +22,10 @@ operation AddTask {
         @required
         @length(min: 1, max: 256)
         title: String
+
+        priority: Priority
+
+        effortHours: Float
     }
 
     output := {
@@ -55,6 +59,14 @@ operation GetTask {
     errors: [NoSuchTask]
 }
 
+/// int32 on the wire: out-of-range values fail the parse and unknown
+/// in-range values fail server validation (#109) — the members above give
+/// those checks out-of-tree, real-socket coverage.
+intEnum Priority {
+    LOW = 1
+    HIGH = 2
+}
+
 @error("client")
 @httpError(404)
 structure NoSuchTask {

examples/bazel-consumer/todo_beast_acceptance_test.cc

@@ -107,6 +107,50 @@ TEST_F(TodoBeastAcceptanceTest, RoundTripsAndModeledErrorsWork) {
   ASSERT_NE(missing.error().detail<NoSuchTask>(), nullptr);
 }
 
+// The #109 numeric bounds, out of tree and over the real socket: hostile
+// wire values for the model's intEnum and float members are rejected by
+// the generated server before the handler runs, and the generated client
+// serializes valid ones — the same generator output consumers ship.
+TEST_F(TodoBeastAcceptanceTest, NumericBoundsAreEnforcedOverTheRealSocket) {
+  smithy::http::BeastHttpClient raw({.host = "127.0.0.1", .port = transport_->port()});
+  const auto post = [&raw](const std::string& body) {
+    smithy::http::HttpRequest request;
+    request.method = "POST";
+    request.target = "/tasks";
+    request.headers.Set("content-type", "application/json");
+    request.body = body;
+    return raw.Send(request);
+  };
+
+  // 2^32+2 would alias onto a valid Priority under a truncating cast; the
+  // parse rejects it instead.
+  auto aliased = post(R"({"title":"t","priority":4294967298})");
+  ASSERT_TRUE(aliased.ok()) << aliased.error().message();
+  EXPECT_EQ(aliased->status, 400);
+  EXPECT_EQ(aliased->headers.Get("x-error-type").value_or("<missing>"), "SerializationException");
+
+  // In range but outside the modeled set: membership validation, with the
+  // suite-exact message shape.
+  auto unknown = post(R"({"title":"t","priority":3})");
+  ASSERT_TRUE(unknown.ok()) << unknown.error().message();
+  EXPECT_EQ(unknown->status, 400);
+  EXPECT_EQ(unknown->headers.Get("x-error-type").value_or("<missing>"), "ValidationException");
+  EXPECT_NE(unknown->body.find("Member must satisfy enum value set: [1, 2]"), std::string::npos)
+      << unknown->body;
+
+  // A finite double beyond float range was UB to cast; it fails the parse.
+  auto overflow = post(R"({"title":"t","effortHours":1e300})");
+  ASSERT_TRUE(overflow.ok()) << overflow.error().message();
+  EXPECT_EQ(overflow->status, 400);
+  EXPECT_EQ(overflow->headers.Get("x-error-type").value_or("<missing>"), "SerializationException");
+
+  // Valid values ride the generated client end to end.
+  auto added = client_->AddTask(AddTaskInput{
+      .title = "well bounded", .priority = acme::todo::Priority::kHigh, .effortHours = 1.5F});
+  ASSERT_TRUE(added.ok()) << added.error().message();
+  EXPECT_EQ(added->title, "well bounded");
+}
+
 TEST_F(TodoBeastAcceptanceTest, LifecycleStopsAndRestartsAcrossTransportGenerations) {
   // The rolling-restart pattern from the production guide, composed entirely
   // from the consumer-visible API: serve, Stop() (bounded — the drain knob

examples/bazel-consumer/websocket_contract_consumer_test.cc

@@ -12,13 +12,16 @@
 
 #include <gtest/gtest.h>
 
+#include <algorithm>
 #include <chrono>
 #include <condition_variable>
 #include <cstddef>
+#include <cstdint>
 #include <memory>
 #include <mutex>
 #include <optional>
 #include <string>
+#include <thread>
 #include <utility>
 
 #include "smithy/eventstream/frame.h"
@@ -37,7 +40,7 @@ using smithy::http::WebSocket;
 // wire but that the terminal transition is expressed with TerminalWaiters,
 // so this implementation inherits the ordering rule without its author
 // having to rediscover why the rule exists.
-class ConsumerSocket final : public WebSocket {
+class ConsumerSocket final : public WebSocket, public std::enable_shared_from_this<ConsumerSocket> {
  public:
   // Small on purpose: the contract suite wedges the wire by sending, and a
   // shallow queue gets there in a few messages.
@@ -74,17 +77,54 @@ class ConsumerSocket final : public WebSocket {
   // lock is released — the seam's documented shapes, nothing more.
   void ReceiveAsync(ReceiveCallback callback) override {
     Outcome<std::optional<Message>> immediate = std::optional<Message>();  // the clean end
+    // The same deliver-slot shape as the timed overload below: parking
+    // hands the completion to the session and empties this, so one park
+    // style covers both receives.
+    ReceiveCallback deliver = std::move(callback);
     {
       const std::lock_guard<std::mutex> lock(mutex_);
       if (!closed_ && !pending_receive_) {
-        pending_receive_ = std::move(callback);
-        return;  // EndSession completes it
+        pending_receive_ = std::exchange(deliver, nullptr);
+        ++receive_park_generation_;  // a stale deadline must not fire this park
+      } else if (!closed_) {
+        immediate = smithy::Error::Validation("consumer socket: a receive is already outstanding");
       }
-      if (!closed_) {
+    }
+    if (!deliver) return;  // parked: EndSession completes it
+    deliver(std::move(immediate));
+  }
+
+  // The deadline overload (#130): the same park, bounded by a watchdog
+  // thread racing the terminal transition — the executor-less shape. The
+  // park generation settles the race exactly once under the lock, so a
+  // stale deadline can never time out a later receive; the watchdog holds
+  // the socket alive (shared_from_this) for at most its own deadline, and
+  // it is spawned outside the lock so a park never stalls the session
+  // behind pthread_create.
+  void ReceiveAsync(std::chrono::milliseconds timeout, ReceiveCallback callback) override {
+    Outcome<std::optional<Message>> immediate = std::optional<Message>();  // the clean end
+    // The completion this call still owes; parking hands it to the session
+    // and leaves this empty, so the slot itself says which happened.
+    ReceiveCallback deliver = std::move(callback);
+    std::uint64_t parked_generation = 0;  // 0 = not parked (the counter starts at 1)
+    {
+      const std::lock_guard<std::mutex> lock(mutex_);
+      if (!closed_ && !pending_receive_ && timeout > std::chrono::milliseconds::zero()) {
+        pending_receive_ = std::exchange(deliver, nullptr);
+        parked_generation = ++receive_park_generation_;
+      } else if (!closed_ && pending_receive_) {
         immediate = smithy::Error::Validation("consumer socket: a receive is already outstanding");
+      } else if (!closed_) {
+        // The non-positive poll — this peer never sends, so nothing is
+        // ever already in hand.
+        immediate = smithy::Error::Timeout("consumer socket: no message within the deadline");
       }
     }
-    callback(std::move(immediate));
+    if (!deliver) {
+      ArmDeadline(parked_generation, timeout);  // EndSession or the deadline completes it
+      return;
+    }
+    deliver(std::move(immediate));
   }
 
   void SendAsync(const Message& message, SendCallback callback) override {
@@ -126,10 +166,54 @@ class ConsumerSocket final : public WebSocket {
   }
 
  private:
+  // Arms the watchdog for the park `generation`. It sleeps on the shared
+  // condition variable, so a completed park (EndSession, or a fresh park's
+  // bumped generation) releases it early; at the deadline, a park still
+  // bearing its generation is timed out — the slot released exactly as a
+  // completion releases it. A spawn that fails must not leave the park
+  // unbounded: it is taken back (unless the session ended it first) and
+  // refused, keeping the callback exactly-once with nothing thrown at the
+  // caller. The saturation guards milliseconds::max()-style "practically
+  // forever" deadlines from overflowing into the past.
+  void ArmDeadline(std::uint64_t generation, std::chrono::milliseconds timeout) {
+    constexpr std::chrono::milliseconds kMaxWait = std::chrono::hours(24 * 365);
+    const auto deadline = std::chrono::steady_clock::now() + std::min(timeout, kMaxWait);
+    const auto watchdog = [self = shared_from_this(), generation, deadline] {
+      ReceiveCallback expired;
+      {
+        std::unique_lock<std::mutex> lock(self->mutex_);
+        self->changed_.wait_until(lock, deadline, [&] {
+          return !self->pending_receive_ || self->receive_park_generation_ != generation;
+        });
+        if (!self->pending_receive_ || self->receive_park_generation_ != generation) {
+          return;  // the park this deadline bounded already completed
+        }
+        expired = std::exchange(self->pending_receive_, nullptr);
+        self->changed_.notify_all();
+      }
+      expired(smithy::Error::Timeout("consumer socket: no message within the deadline"));
+    };
+    ReceiveCallback refused;
+    try {
+      std::thread(watchdog).detach();
+      return;
+    } catch (...) {
+      const std::lock_guard<std::mutex> lock(mutex_);
+      if (pending_receive_ && receive_park_generation_ == generation) {
+        refused = std::exchange(pending_receive_, nullptr);
+        changed_.notify_all();
+      }
+    }
+    if (refused) {
+      refused(smithy::Error::Transport("consumer socket: cannot arm the receive deadline"));
+    }
+  }
+
   std::mutex mutex_;
   std::condition_variable changed_;
   ReceiveCallback pending_receive_;
   SendCallback pending_send_;
+  std::uint64_t receive_park_generation_ = 0;
   std::size_t queued_ = 0;
   bool closed_ = false;
 };