let one refused datagram fail without stalling the batch

README.md

@@ -43,7 +43,8 @@ for (const auto& d : in->datagrams() | std::views::filter(dgram::is_intact)) {
         (void)out->stage(piece, d.from());
     }
 }
-(void)out->flush(*sock);   // discarding the result keeps a failed send staged
+(void)out->flush(*sock);   // a refused datagram is dropped; the rest still go
+out->discard();            // whatever a stalled flush kept still points into `in`
 ```
 
 `then`, `map`, `recover` and `tap` compose a `result` left to right, in the order

docs/batches.md

@@ -105,14 +105,16 @@ bool stage(std::span<const std::byte> payload, const endpoint& to);
 bool stage(std::span<const std::byte> payload, const endpoint& to, const control<Features>&);
 bool stage_copy(std::span<const std::byte> payload, const endpoint& to,
                 const control<Features>& = {});                          // SlotBytes > 0
-result<std::size_t> flush(const socket& sock, int flags = 0);
+flushed flush(const socket& sock, int flags = 0);
+void discard();
 bool full() const;
 std::size_t staged() const;
 ```
 
 `stage` references the caller's bytes rather than copying, so bouncing a datagram
 straight out of a receive slot costs nothing. **The referenced bytes must stay
-valid until `flush` returns.** It yields `false` when the batch is full.
+valid until a flush sends them or `discard()` drops them.** It yields `false`
+when the batch is full.
 
 `stage_copy` copies into the batch's own slot, for bytes that will not outlive
 the flush. It requires `SlotBytes > 0` and yields `false` if the payload exceeds
@@ -122,24 +124,45 @@ The `control` overload attaches ancillary data to that one datagram; see
 [metadata](metadata.md). `msg_controllen` is set from what was actually built,
 zero included, so nothing carries over between datagrams staged in the same slot.
 
-`flush` sends everything staged and clears the batch, returning how many were
-taken. A short send is not an error and the remainder is dropped: datagram
-delivery is unreliable by definition, so requeueing buys nothing a protocol layer
-cannot do better.
-
-**A failed send leaves the batch staged.** That case is not a short send: nothing
-went out at all, so clearing would hand the caller an empty batch and no way to
-tell what it lost. The argument for dropping a short send's tail is that the
-protocol layer can do better, which is only true while the protocol still has the
-bytes. `EAGAIN` on a non-blocking socket is the ordinary way to get here; flush
-again once the socket is writable, or call `discard()` to drop it.
+`flush` sends everything staged and reports what happened:
+
+| `flushed` member | Description |
+|------------------|-------------|
+| `sent` | Datagrams the kernel took. |
+| `rejected` | Entries dropped because the kernel refused that datagram. |
+| `last_rejection` | Why the last of those was refused. |
+| `stalled` | Why the flush stopped early. Set means entries are still staged. |
+| `drained()` | `stalled` is empty: every entry was sent or rejected. |
+
+`sendmmsg` stops at the first datagram it cannot send, so `flush` decides who the
+error belongs to.
+
+- **The datagram's own** (`EMSGSIZE`, `EHOSTUNREACH`, `EPERM` from a firewall,
+  `EACCES` on broadcast, ...): that entry is dropped and the rest are still sent.
+  One bad destination cannot hold up everyone else in the batch, and a
+  per-destination error that repeats cannot wedge it.
+- **The socket's** (`EAGAIN`, `ENOBUFS`, `ENOMEM`, `EINTR`, `EBADF`, `ENOTSOCK`,
+  `EOPNOTSUPP`): the next entry would hit it too, so `flush` stops. Everything
+  not yet sent stays staged, in order, including the tail of a send that got
+  partway. Flush again once the socket is writable, or `discard()` it.
+
+**Entries a stalled flush keeps still reference the caller's bytes.** If those
+are receive slots, the next `receive` overwrites them, and a later flush would
+send new datagrams' bytes to old datagrams' peers. Flush again before receiving,
+or discard:
 
 ```cpp
-if (const auto sent{tx->flush(*sock)}; !sent && sent.error() == dgram::would_block) {
-    tx->discard();  // or leave it staged and flush again when writable
+const auto sent{tx->flush(*sock)};
+if (sent.stalled == dgram::would_block) {
+    wait_until_writable(*sock);  // then flush again, before the next receive
+} else if (sent.stalled) {
+    tx->discard();
 }
 ```
 
+Stalled entries keep their slots until they go out, so `full()` can be true with
+fewer than `Capacity` staged.
+
 ## Retaining bytes past the batch
 
 There is no retention mechanism. A protocol that reassembles or retransmits, as

docs/testing.md

@@ -18,7 +18,7 @@ checkout.
 |-------|--------|
 | `doc_examples.cpp` | Every snippet printed in `README.md` and `docs/`, compiled and run. A snippet in a document is untested code that looks authoritative; this is what stops one rotting. |
 | `compile_checks.cpp` | Pure `static_assert`. Feature concepts, control-buffer sizing, batch geometry, which resources may be carved from, which metadata is readable, and SipHash against its published vectors, split at every word boundary. Compiling is the test; it is never run. |
-| `loopback_tests.cpp` | Batch reuse, truncation reporting (including a caller-supplied `MSG_TRUNC`), nonblocking receive, staging limits, a failed flush leaving the batch staged, the error combinators. |
+| `loopback_tests.cpp` | Batch reuse, truncation reporting (including a caller-supplied `MSG_TRUNC`), nonblocking receive, staging limits, the error combinators. Flushing: a refused datagram dropped while the rest still go, socket-level errors stalling with everything kept staged in order, and `send_all` against scripted `sendmmsg` results, since loopback never returns `EAGAIN` on send. |
 | `metadata_tests.cpp` | Destination address and ECN over loopback, both families, including a feature that was never enabled and an undersized control buffer. DSCP and ECN together in both families, and a per-datagram marking replacing the socket's DSCP. Dual-stack sockets: ECN from and to IPv4 peers, and a reply from the arrival address. |
 | `malformed_control_tests.cpp` | Control buffers the kernel would never write. Every case here was found by the fuzzer first. |
 | `offload_tests.cpp` | GRO and GSO over loopback, and the segmentation walk on its own: short tail, exact multiple, stride of zero, stride past the end, empty payload. |

examples/echo.cpp

@@ -103,8 +103,14 @@ int main(const int argc, const char* const* argv) {
             }
         }
 
-        if (const auto sent{tx->flush(*sock)}; !sent) {
-            std::println(stderr, "flush: {}", dgram::describe(sent.error()));
+        const auto sent{tx->flush(*sock)};
+        if (sent.last_rejection) {
+            std::println(stderr, "flush: {} refused, last: {}", sent.rejected, dgram::describe(*sent.last_rejection));
+        }
+        if (sent.stalled) {
+            // What is still staged points into the receive slots the next receive overwrites.
+            std::println(stderr, "flush stalled: {}", dgram::describe(*sent.stalled));
+            tx->discard();
         }
     }
 }

src/dgram/batch.cppm

@@ -327,13 +327,69 @@ private:
  * Transmit
  * ============================================================================ */
 
+/** @brief What one `flush` did. */
+export struct flushed {
+    std::size_t sent{};                   ///< datagrams the kernel took
+    std::size_t rejected{};               ///< entries dropped because the kernel refused that datagram
+    std::optional<errc> last_rejection{}; ///< why the last of those was refused
+    std::optional<errc> stalled{};        ///< why it stopped early; every unsent entry is still staged
+
+    /** @brief Nothing is left staged: every entry was sent or rejected. */
+    [[nodiscard]] constexpr bool drained() const noexcept { return !stalled.has_value(); }
+};
+
+namespace detail {
+
+/**
+ * @brief Errors that belong to the socket or the moment, not to one datagram.
+ *
+ * Retrying the next entry would hit them again, so they stop a flush. Anything
+ * else is the datagram's own fault and drops only that entry.
+ */
+inline constexpr std::array socket_errors{would_block, interrupted, out_of_memory, errc{ENOBUFS}, errc{EBADF}, errc{ENOTSOCK}, errc{EOPNOTSUPP}};
+
+/**
+ * @brief Send `pending` front to back, through `send` standing in for one `sendmmsg`.
+ *
+ * `sendmmsg` stops at the first datagram it cannot send and, if it sent any
+ * before it, reports only the count. The next call starts at that datagram and
+ * gets its error. Exactly `sent + rejected` entries from the front are consumed.
+ */
+export template <typename Send>
+    requires std::is_invocable_r_v<result<int>, Send&, std::span<::mmsghdr>>
+[[nodiscard]] flushed send_all(std::span<::mmsghdr> pending, Send&& send) noexcept {
+    flushed out{};
+    while (!pending.empty()) {
+        const auto n{std::invoke(send, pending)};
+        if (n.has_value() && *n > 0) {
+            const auto taken{std::min(static_cast<std::size_t>(*n), pending.size())};
+            out.sent += taken;
+            pending = pending.subspan(taken);
+            continue;
+        }
+        // A zero count, which a non-empty vector never gets, stalls rather than spins.
+        const auto e{n.error_or(would_block)};
+        if (std::ranges::contains(socket_errors, e)) {
+            out.stalled = e;
+            return out;
+        }
+        ++out.rejected;
+        out.last_rejection = e;
+        pending = pending.subspan(1);
+    }
+    return out;
+}
+
+} // namespace detail
+
 /**
  * @brief Fixed-size `sendmmsg` argument block over caller-provided memory.
  *
  * `stage` references the caller's bytes instead of copying them, so echoing
  * straight out of a `receive_batch` slot costs nothing. The referenced bytes
- * must stay put until `flush` returns. `SlotBytes` of zero carves no payload
- * memory at all and leaves only the referencing path available.
+ * must stay put until a flush sends them or `discard()` drops them, which a
+ * stalled flush postpones. `SlotBytes` of zero carves no payload memory at all
+ * and leaves only the referencing path available.
  */
 export template <std::size_t Capacity, std::size_t SlotBytes = 0, feature_set Features = no_features> class transmit_batch {
     static_assert(Capacity > 0 && Capacity <= IOV_MAX, "capacity must fit IOV_MAX");
@@ -368,9 +424,16 @@ public:
         return b;
     }
 
-    /** @brief Whether another datagram fits before a flush is required. */
-    [[nodiscard]] constexpr bool full() const noexcept { return staged_ == Capacity; }
-    [[nodiscard]] constexpr std::size_t staged() const noexcept { return staged_; }
+    /**
+     * @brief Whether another datagram fits before a flush is required.
+     *
+     * Entries a stalled flush left behind keep their slots, so this can be
+     * true with fewer than `Capacity` staged.
+     */
+    [[nodiscard]] constexpr bool full() const noexcept { return end_ == Capacity; }
+
+    /** @brief Entries waiting to be sent. */
+    [[nodiscard]] constexpr std::size_t staged() const noexcept { return end_ - head_; }
 
     /**
      * @brief Queue `payload` for `to`, referencing the bytes in place.
@@ -396,7 +459,7 @@ public:
         if (full()) [[unlikely]] {
             return false;
         }
-        const auto i{staged_++};
+        const auto i{end_++};
         addrs_[i] = to;
         // sendmmsg only reads through iov_base, but the field is not const-qualified.
         iovs_[i].iov_base = const_cast<std::byte*>(payload.data());
@@ -417,38 +480,38 @@ public:
         if (full() || payload.size() > SlotBytes) [[unlikely]] {
             return false;
         }
-        const auto dest{payload_.subspan(staged_ * SlotBytes, payload.size())};
+        const auto dest{payload_.subspan(end_ * SlotBytes, payload.size())};
         std::ranges::copy(payload, dest.begin());
         return stage(dest, to, ancillary);
     }
 
     /**
-     * @brief Send everything staged, then clear the batch.
-     *
-     * A short send is not an error: `sendmmsg` reports how many it took and the
-     * remainder is dropped. Datagram delivery is unreliable by definition, so
-     * requeueing the tail buys nothing a protocol layer cannot do better.
+     * @brief Send everything staged, dropping only what the kernel refuses.
      *
-     * **A failed send is different and the batch is left staged.** Nothing went
-     * out, so clearing would hand the caller an empty batch and no way to know
-     * what it lost, and "a protocol layer can do better" is only true while the
-     * protocol still has the bytes. `EAGAIN` on a non-blocking socket is the
-     * common case: flush again once it is writable, or `discard()` to drop it.
+     * An error that belongs to one datagram (`EMSGSIZE`, `EHOSTUNREACH`,
+     * `EPERM`, ...) drops that entry, counts it in `rejected`, and the rest are
+     * still sent, so one bad destination cannot hold up the batch. An error that
+     * belongs to the socket (`EAGAIN`, `ENOBUFS`, `EINTR`, `EBADF`, ...) stops
+     * the flush and sets `stalled`. Everything not yet sent stays staged and
+     * still references the caller's bytes: flush again once the socket is
+     * writable, or `discard()` it before those bytes change.
      */
-    [[nodiscard]] result<std::size_t> flush(const socket& sock, const int flags = 0) noexcept {
-        if (staged_ == 0) {
-            return std::size_t{0};
-        }
-        const int n{::sendmmsg(sock.native(), msgs_.data(), static_cast<unsigned>(staged_), flags)};
-        if (n < 0) [[unlikely]] {
-            return fail<std::size_t>();
+    [[nodiscard]] flushed flush(const socket& sock, const int flags = 0) noexcept {
+        const auto out{detail::send_all(msgs_.subspan(head_, end_ - head_), [&sock, flags](const std::span<::mmsghdr> pending) noexcept {
+            return invoke_syscall(::sendmmsg, sock.native(), pending.data(), static_cast<unsigned>(pending.size()), flags);
+        })};
+        head_ += out.sent + out.rejected;
+        if (head_ == end_) {
+            discard();
         }
-        staged_ = 0;
-        return static_cast<std::size_t>(n);
+        return out;
     }
 
-    /** @brief Drop everything staged without sending it, after a failed `flush`. */
-    void discard() noexcept { staged_ = 0; }
+    /** @brief Drop everything staged without sending it. */
+    void discard() noexcept {
+        head_ = 0;
+        end_ = 0;
+    }
 
 private:
     transmit_batch() noexcept = default;
@@ -481,7 +544,8 @@ private:
     std::span<endpoint> addrs_{};
     std::span<std::byte> control_{};
     std::span<std::byte> payload_{};
-    std::size_t staged_{};
+    std::size_t head_{}; ///< first entry not yet sent; a stalled flush leaves it past zero
+    std::size_t end_{};  ///< one past the last staged entry, where `stage` writes next
 };
 
 } // namespace dgram

tests/concurrency_tests.cpp

@@ -71,7 +71,7 @@ TEST(Concurrency, WorkersShareNothing) {
                 if (!tx->stage_copy(bytes_of(text), target, marked)) {
                     continue;
                 }
-                ASSERT_TRUE(tx->flush(*peer).has_value());
+                ASSERT_TRUE(tx->flush(*peer).drained());
 
                 // Nonblocking, so poll briefly rather than parking a thread.
                 for (int attempt{}; attempt < 2000; ++attempt) {

tests/demux_tests.cpp

@@ -382,7 +382,7 @@ TEST(Route, DeliversToTheMatchingFlowAndReportsTheRest) {
 
     const auto say = [&](dgram::socket& s, std::string_view text) {
         ASSERT_TRUE(tx->stage_copy({reinterpret_cast<const std::byte*>(text.data()), text.size()}, net.target));
-        ASSERT_TRUE(tx->flush(s).has_value());
+        ASSERT_TRUE(tx->flush(s).drained());
     };
     say(*known, "from-known");
     say(*stranger, "from-stranger");
@@ -443,7 +443,7 @@ TEST(Route, CoalescedSlotRoutesPerDatagramNotPerSlot) {
     dgram::control<tx_set> ancillary{};
     ancillary.set<dgram::segment>(mtu);
     ASSERT_TRUE(tx->stage(payload, net.target, ancillary));
-    ASSERT_TRUE(tx->flush(*sender).has_value());
+    ASSERT_TRUE(tx->flush(*sender).drained());
 
     ASSERT_EQ(*rx->receive(net.receiver), 1u) << "GRO should coalesce into one slot";
     {
@@ -475,7 +475,7 @@ TEST(Route, TruncatedDatagramsAreCountedNotDelivered) {
 
     constexpr std::string_view text{"far longer than eight bytes"};
     ASSERT_TRUE(tx->stage_copy({reinterpret_cast<const std::byte*>(text.data()), text.size()}, net.target));
-    ASSERT_TRUE(tx->flush(*sender).has_value());
+    ASSERT_TRUE(tx->flush(*sender).drained());
 
     ASSERT_EQ(*rx->receive(net.receiver), 1u);
     const auto counts{dgram::route(*rx, dgram::by_peer{}, *table, 0ns)};
@@ -507,7 +507,7 @@ struct routing_fixture {
     template <typename Table, typename Unmatched, typename... Context>
     dgram::routed exchange(rx_batch& rx, tx_batch& tx, dgram::socket& s, std::string_view text, Table& table, Unmatched&& unmatched, Context&&... context) {
         EXPECT_TRUE(tx.stage_copy({reinterpret_cast<const std::byte*>(text.data()), text.size()}, net.target));
-        EXPECT_TRUE(tx.flush(s).has_value());
+        EXPECT_TRUE(tx.flush(s).drained());
 
         for (int attempt{}; attempt < 100; ++attempt) {
             if (const auto got{rx.receive(net.receiver)}; got && *got > 0) {
@@ -674,7 +674,7 @@ TEST(Route, ReportsHowManySlotsTheDatagramsCameIn) {
     dgram::control<tx_set> ancillary{};
     (void)ancillary.set<dgram::segment>(stride);
     ASSERT_TRUE(tx->stage_copy(payload, net.target, ancillary));
-    ASSERT_TRUE(tx->flush(*sender).has_value());
+    ASSERT_TRUE(tx->flush(*sender).drained());
 
     dgram::routed counts{};
     for (int attempt{}; attempt < 100 && counts.handled() < 4; ++attempt) {

tests/doc_examples.cpp

@@ -96,7 +96,10 @@ TEST(DocExamples, AtAGlance) {
             }
         }
     }
-    ASSERT_TRUE(out->flush(*sock).has_value());
+    const auto sent{out->flush(*sock)};
+    out->discard();
+    EXPECT_TRUE(sent.drained());
+    EXPECT_EQ(sent.sent, 1u);
     EXPECT_EQ(echoed, 1u);
 }