st: review nits N1-N7

N1 Expected operator*/operator-> are not UB on an error: operator* is
   noexcept + std::get so it terminates; operator-> returns nullptr. Correct
   the docs to say so.
N2 Drop redundant unit prose ("in milliseconds"/"in seconds") from
   std::chrono fields/params in Policies, Consumer (AckPolicy) and the
   sendTimeout setter; the type already states the unit. (ProducerConfig's
   int64 sendTimeoutMs keeps its "milliseconds" note - it is not a chrono type.)
N3 decodeBigEndian: replace the dead `i < data.size()` guard (all codecs
   length-check first) with an assert of that precondition.
N4 ProtobufNativeSchema: guard the size_t->int narrowing in encode/decode,
   rejecting messages larger than INT_MAX instead of passing a wrapped size.
N5 OutgoingMessage: one-line note for the usesView<->payloadView invariant.
N7 Wrap the SerDeFor concept in clang-format off/on so clang-format-11 stops
   mangling the `{ expr } -> Concept;` compound requirements.
N6 Normalize config-struct field docs to the dominant /** */-before style
   (OutgoingMessage, CheckpointConsumerConfig, Stream/QueueConsumerConfig);
   enum-value ///< trailing docs are left as-is.

Verified: clang-format-11 clean; examples compile (clang); N3 runtime test
and N4 (protobuf stub) pass on clang and gcc.

Signed-off-by: Matteo Merli <mmerli@apache.org>

Author: merlimat

include/pulsar/st/CheckpointConsumer.h

@@ -45,15 +45,18 @@ namespace pulsar::st {
  * not construct this directly.
  */
 struct CheckpointConsumerConfig {
-    std::string topic;  ///< Scalable topic to read. REQUIRED; no default.
-    Checkpoint startPosition =
-        Checkpoint::latest();  ///< Position to start from. Default `Checkpoint::latest()`.
-    std::optional<std::string>
-        consumerGroup;  ///< Consumer group to join. Unset (default) => ungrouped, reads every segment.
-    std::optional<std::string>
-        consumerName;       ///< Human-readable consumer name. Unset (default) => auto-generated.
-    Properties properties;  ///< Free-form key/value metadata attached to the consumer. Default empty.
-    SchemaInfo schema;      ///< Schema descriptor; filled in from `Schema<T>` by the builder.
+    /** Scalable topic to read. REQUIRED; no default. */
+    std::string topic;
+    /** Position to start from. Default `Checkpoint::latest()`. */
+    Checkpoint startPosition = Checkpoint::latest();
+    /** Consumer group to join. Unset (default) => ungrouped, reads every segment. */
+    std::optional<std::string> consumerGroup;
+    /** Human-readable consumer name. Unset (default) => auto-generated. */
+    std::optional<std::string> consumerName;
+    /** Free-form key/value metadata attached to the consumer. Default empty. */
+    Properties properties;
+    /** Schema descriptor; filled in from `Schema<T>` by the builder. */
+    SchemaInfo schema;
 };
 
 template <typename T>

include/pulsar/st/Consumer.h

@@ -52,10 +52,10 @@ enum class SubscriptionInitialPosition
  * optional and fall back to the client default when unset.
  */
 struct AckPolicy {
-    /** Time window over which acknowledgments are batched before being sent, in milliseconds; 0 acks
+    /** Time window over which acknowledgments are batched before being sent; 0 acks
      * immediately. Unset uses the client default. */
     std::optional<std::chrono::milliseconds> groupTime = std::nullopt;
-    /** Delay before a negatively-acknowledged message is redelivered, in milliseconds. QueueConsumer only.
+    /** Delay before a negatively-acknowledged message is redelivered. QueueConsumer only.
      * Unset uses the client default. */
     std::optional<std::chrono::milliseconds> negativeAckRedeliveryDelay = std::nullopt;
 };

include/pulsar/st/Expected.h

@@ -139,8 +139,10 @@ class [[nodiscard]] Expected {
     /**
      * Unchecked access to the contained value.
      *
-     * Unlike `value()`, this never throws and performs no check. Behaviour is
-     * undefined if this holds an error; verify with `operator bool` first.
+     * Performs no check and is `noexcept`. It does not throw on an error; instead,
+     * because it reads the wrong `std::variant` alternative through a `noexcept`
+     * boundary, accessing the value when this holds an error terminates the program.
+     * Verify with `operator bool` first.
      *
      * @return a reference to the contained value (lvalue or rvalue per ref-qualifier).
      */
@@ -153,10 +155,10 @@ class [[nodiscard]] Expected {
     /**
      * Unchecked member access to the contained value.
      *
-     * Behaviour is undefined if this holds an error; verify with `operator bool`
-     * first.
+     * Returns `nullptr` if this holds an error (so `e->member` would then dereference
+     * a null pointer); verify with `operator bool` first.
      *
-     * @return a pointer to the contained value.
+     * @return a pointer to the contained value, or `nullptr` if this holds an error.
      */
     const T* operator->() const noexcept { return std::get_if<0>(&storage_); }
     /** @copydoc operator->() const */

include/pulsar/st/Policies.h

@@ -90,21 +90,21 @@ struct MemorySize {
 struct ConnectionPolicy {
     /** Number of physical connections opened to each broker. Unset uses the client default. */
     std::optional<int> connectionsPerBroker = std::nullopt;
-    /** Maximum time to wait for a TCP/TLS connection to be established, in milliseconds. Unset uses the
+    /** Maximum time to wait for a TCP/TLS connection to be established. Unset uses the
      * client default. */
     std::optional<std::chrono::milliseconds> connectionTimeout = std::nullopt;
-    /** Maximum time to wait for a broker request (e.g. produce/consume control ops) to complete, in
-     * milliseconds. Unset uses the client default. */
+    /** Maximum time to wait for a broker request (e.g. produce/consume control ops) to complete. Unset
+     * uses the client default. */
     std::optional<std::chrono::milliseconds> operationTimeout = std::nullopt;
-    /** Interval between keep-alive pings sent on an idle connection, in seconds. Unset uses the client
+    /** Interval between keep-alive pings sent on an idle connection. Unset uses the client
      * default. */
     std::optional<std::chrono::seconds> keepAliveInterval = std::nullopt;
     /** Maximum number of concurrent topic-lookup requests in flight. Unset uses the client default. */
     std::optional<int> maxLookupRequests = std::nullopt;
     /** Maximum number of lookup redirects to follow before failing a lookup. Unset uses the client default.
      */
     std::optional<int> maxLookupRedirects = std::nullopt;
-    /** Time an idle pooled connection may stay open before being closed, in milliseconds. Unset uses the
+    /** Time an idle pooled connection may stay open before being closed. Unset uses the
      * client default. */
     std::optional<std::chrono::milliseconds> maxConnectionIdleTime = std::nullopt;
     /** Advertised listener name for broker discovery (multi-listener deployments). Unset uses the
@@ -120,9 +120,9 @@ struct ConnectionPolicy {
  * applies its built-in default for that bound.
  */
 struct BackoffPolicy {
-    /** Delay before the first reconnection attempt, in milliseconds. Unset uses the client default. */
+    /** Delay before the first reconnection attempt. Unset uses the client default. */
     std::optional<std::chrono::milliseconds> initialBackoff = std::nullopt;
-    /** Upper bound on the backoff delay as it grows across retries, in milliseconds. Unset uses the client
+    /** Upper bound on the backoff delay as it grows across retries. Unset uses the client
      * default. */
     std::optional<std::chrono::milliseconds> maxBackoff = std::nullopt;
 };
@@ -161,7 +161,7 @@ struct TlsPolicy {
  * optional and the client supplies a built-in default when it is unset.
  */
 struct TransactionPolicy {
-    /** Default lifetime of a transaction before it is automatically aborted, in milliseconds. Unset uses the
+    /** Default lifetime of a transaction before it is automatically aborted. Unset uses the
      * client default. */
     std::optional<std::chrono::milliseconds> timeout = std::nullopt;
 };

include/pulsar/st/Producer.h

@@ -101,6 +101,10 @@ struct ProducerConfig {
  * This is the encoded, schema-agnostic form of a message: the typed value has
  * already been serialized to `payload` bytes. The builder fills these fields from
  * its fluent setters and hands the result to the producer core for publishing.
+ *
+ * Invariant (maintained by `MessageBuilder`): when `usesView` is true the payload is
+ * read from `payloadView` (which must stay valid until the send completes); otherwise
+ * it is read from `payload`.
  */
 struct OutgoingMessage {
     /** Encoded message payload — the value serialized to bytes through `Schema<T>`.
@@ -115,12 +119,16 @@ struct OutgoingMessage {
     std::optional<std::string> key;
     /** Per-message user metadata. Empty by default. */
     Properties properties;
-    std::optional<Timestamp> eventTime;  ///< Application event time; unset (nullopt) by default.
-    std::optional<int64_t> sequenceId;   ///< Explicit sequence id; unset = auto-assign.
-    std::optional<Timestamp> deliverAt;  ///< Scheduled delivery time; unset = deliver immediately.
+    /** Application event time; unset (nullopt) by default. */
+    std::optional<Timestamp> eventTime;
+    /** Explicit sequence id; unset = auto-assign. */
+    std::optional<int64_t> sequenceId;
+    /** Scheduled delivery time; unset = deliver immediately. */
+    std::optional<Timestamp> deliverAt;
     /** Target clusters for geo-replication; empty applies the topic's default. */
     std::vector<std::string> replicationClusters;
-    std::optional<Transaction> transaction;  ///< Enlisting transaction; unset = non-transactional.
+    /** Enlisting transaction; unset = non-transactional. */
+    std::optional<Transaction> transaction;
 };
 
 template <typename T>
@@ -451,8 +459,7 @@ class ProducerBuilder {
      * Set the per-message send timeout: how long a send may stay unacknowledged
      * before failing.
      *
-     * @param d the timeout, in milliseconds. Optional; when unset the SDK default
-     *          applies.
+     * @param d the timeout. Optional; when unset the SDK default applies.
      * @return `*this`, for chaining.
      */
     ProducerBuilder& sendTimeout(std::chrono::milliseconds d) {

include/pulsar/st/ProtobufNativeSchema.h

@@ -24,6 +24,7 @@
 #include <pulsar/st/Schema.h>
 
 #include <cstddef>
+#include <limits>
 #include <span>
 #include <string>
 #include <type_traits>
@@ -41,12 +42,17 @@ struct ProtobufNativeSerDe {
                   "protobufNativeSchema<T> requires T to be a generated protobuf Message");
     SchemaInfo info() const { return pulsar::createProtobufNativeSchema(T::descriptor()); }
     Expected<void> encode(const T& value, std::vector<std::byte>& out) const {
-        out.resize(value.ByteSizeLong());
-        if (!value.SerializeToArray(out.data(), static_cast<int>(out.size())))
+        const std::size_t size = value.ByteSizeLong();
+        if (size > static_cast<std::size_t>(std::numeric_limits<int>::max()))
+            return unexpected(pulsar::ResultInvalidMessage, "protobuf message too large to serialize");
+        out.resize(size);
+        if (!value.SerializeToArray(out.data(), static_cast<int>(size)))
             return unexpected(pulsar::ResultInvalidMessage, "failed to serialize protobuf message");
         return {};
     }
     Expected<T> decode(std::span<const std::byte> data) const {
+        if (data.size() > static_cast<std::size_t>(std::numeric_limits<int>::max()))
+            return unexpected(pulsar::ResultInvalidMessage, "protobuf message too large to parse");
         T message;
         if (message.ParseFromArray(data.data(), static_cast<int>(data.size()))) return message;
         return unexpected(pulsar::ResultInvalidMessage, "failed to parse protobuf message");

include/pulsar/st/QueueConsumer.h

@@ -55,38 +55,38 @@ namespace pulsar::st {
  * connecting. Fields not selected by `useNamespace` are ignored.
  */
 struct QueueConsumerConfig {
-    /// Selects namespace mode over single-topic mode. When `false` (the default),
-    /// `topic` is used; when `true`, `namespaceName` (and `propertyFilters`) apply.
+    /** Selects namespace mode over single-topic mode. When `false` (the default),
+     *  `topic` is used; when `true`, `namespaceName` (and `propertyFilters`) apply. */
     bool useNamespace = false;
-    /// Fully-qualified topic name. Used only when `useNamespace == false`. Mutually
-    /// exclusive with `namespaceName`.
+    /** Fully-qualified topic name. Used only when `useNamespace == false`. Mutually
+     *  exclusive with `namespaceName`. */
     std::string topic;  // when !useNamespace
-    /// Namespace name (`tenant/namespace`). Used only when `useNamespace == true`.
-    /// Subscribes to all scalable topics in the namespace with live membership.
+    /** Namespace name (`tenant/namespace`). Used only when `useNamespace == true`.
+     *  Subscribes to all scalable topics in the namespace with live membership. */
     std::string namespaceName;  // when useNamespace
-    /// Namespace mode only: AND filters matched against topic properties to select
-    /// which topics in the namespace are included. Empty means no filtering (all
-    /// topics). Ignored in single-topic mode.
+    /** Namespace mode only: AND filters matched against topic properties to select
+     *  which topics in the namespace are included. Empty means no filtering (all
+     *  topics). Ignored in single-topic mode. */
     Properties propertyFilters;  // namespace mode: AND filters over topic properties
-    /// REQUIRED. Subscription name shared by all consumers of this subscription.
+    /** REQUIRED. Subscription name shared by all consumers of this subscription. */
     std::string subscriptionName;  // REQUIRED
-    /// Where the subscription starts when it is first created. Default
-    /// `SubscriptionInitialPosition::Latest` (skip the backlog). Has no effect once
-    /// the subscription already exists.
+    /** Where the subscription starts when it is first created. Default
+     *  `SubscriptionInitialPosition::Latest` (skip the backlog). Has no effect once
+     *  the subscription already exists. */
     SubscriptionInitialPosition initialPosition = SubscriptionInitialPosition::Latest;
-    /// Optional consumer name (useful for diagnostics and metrics). Default unset, in
-    /// which case the broker assigns one.
+    /** Optional consumer name (useful for diagnostics and metrics). Default unset, in
+     *  which case the broker assigns one. */
     std::optional<std::string> consumerName;
-    /// Acknowledgment tuning (e.g. the ack-grouping/batching window and negative-ack
-    /// redelivery delay). Default-constructed `AckPolicy` when unset.
+    /** Acknowledgment tuning (e.g. the ack-grouping/batching window and negative-ack
+     *  redelivery delay). Default-constructed `AckPolicy` when unset. */
     AckPolicy ackPolicy;
-    /// Optional dead-letter policy: route messages to a dead-letter topic after
-    /// repeated redelivery. Default unset (no dead-lettering).
+    /** Optional dead-letter policy: route messages to a dead-letter topic after
+     *  repeated redelivery. Default unset (no dead-lettering). */
     std::optional<DeadLetterPolicy> deadLetterPolicy;
-    /// Arbitrary client-side consumer properties (reported in topic stats). Default empty.
+    /** Arbitrary client-side consumer properties (reported in topic stats). Default empty. */
     Properties properties;
-    /// Schema descriptor for the value type `T`. Populated automatically by the builder
-    /// from the `Schema<T>` it was constructed with.
+    /** Schema descriptor for the value type `T`. Populated automatically by the builder
+     *  from the `Schema<T>` it was constructed with. */
     SchemaInfo schema;
 };
 

include/pulsar/st/Schema.h

@@ -24,6 +24,7 @@
 #include <pulsar/st/Error.h>
 #include <pulsar/st/Expected.h>
 
+#include <cassert>
 #include <concepts>
 #include <cstddef>
 #include <cstdint>
@@ -81,16 +82,15 @@ using BytesView = std::span<const std::byte>;
  * @tparam S the candidate SerDe type.
  * @tparam T the value type the SerDe handles.
  */
+// clang-format off
 template <typename S, typename T>
 concept SerDeFor = requires(const S& serde, const T& value, std::span<const std::byte> data,
                             std::vector<std::byte>& out) {
-    { serde.info() }
-    ->std::convertible_to<SchemaInfo>;
-    { serde.encode(value, out) }
-    ->std::convertible_to<Expected<void>>;
-    { serde.decode(data) }
-    ->std::convertible_to<Expected<T>>;
+    { serde.info() } -> std::convertible_to<SchemaInfo>;
+    { serde.encode(value, out) } -> std::convertible_to<Expected<void>>;
+    { serde.decode(data) } -> std::convertible_to<Expected<T>>;
 };
+// clang-format on
 
 /**
  * `Schema<T>` is the typed seam of the API: `Producer<T>`, `Consumer<T>` and
@@ -223,8 +223,9 @@ inline void encodeBigEndian(U value, std::vector<std::byte>& out) {
 template <typename U>
 inline U decodeBigEndian(std::span<const std::byte> data) {
     static_assert(std::is_integral_v<U>, "integral only");
+    assert(data.size() >= sizeof(U) && "callers validate the payload length before decoding");
     std::make_unsigned_t<U> u = 0;
-    for (std::size_t i = 0; i < sizeof(U) && i < data.size(); ++i) {
+    for (std::size_t i = 0; i < sizeof(U); ++i) {
         u = (u << 8) | std::to_integer<unsigned char>(data[i]);
     }
     return static_cast<U>(u);