/// @file dpf/net/async_stream_array.hpp
/// @brief Truly asynchronous, overlapped indexed byte streams.
/// @details Every backend completes through an `asio::io_context` and never
///          spins on a `peer_ready` flag or blocks the calling thread. The
///          calling thread of `async_read` only registers a waiter and returns;
///          the completion handler fires later on an `io_context` thread.
///            * `async_dual_memory_stream_array` — paired in-process ends.
///            * `async_mux_stream_array`         — N lanes on ONE TCP socket.
///            * `async_parallel_stream_array`    — one socket PER lane.
///          Writes are split into `wire_policy::chunk_bytes` frames. The mux
///          sends chunks round-robin across lanes, so a large write on one lane
///          does not queue small writes on another lane behind it. When a reader
///          is already waiting, socket payloads land directly in its buffer.
#ifndef LIBDPF_INCLUDE_DPF_NET_ASYNC_STREAM_ARRAY_HPP__
#define LIBDPF_INCLUDE_DPF_NET_ASYNC_STREAM_ARRAY_HPP__

#include <algorithm>
#include <atomic>
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <cstring>
#include <deque>
#include <functional>
#include <memory>
#include <mutex>
#include <stdexcept>
#include <string>
#include <system_error>
#include <utility>
#include <vector>

#include "dpf/net/asio_ns.hpp"

#include "hedley/hedley.h"

#include "dpf/net/buffer_pool.hpp"
#include "dpf/net/connect.hpp"
#include "dpf/net/policy.hpp"
#include "dpf/net/socket_tune.hpp"
#include "dpf/net/stream_array.hpp"  // stream_frame_hdr

namespace dpf
{
namespace net
{

/// @brief Completion handler for a single async I/O operation.
using async_handler = std::function<void(const std::error_code &)>;

/// @brief Counters for one stream array end.
/// @details `bytes_*` include framing; `payload_*` are caller bytes. Times are
///          `steady_clock` nanoseconds since epoch (0 = never).
struct stream_stats
{
    std::uint64_t bytes_out = 0;
    std::uint64_t bytes_in = 0;
    std::uint64_t payload_out = 0;
    std::uint64_t payload_in = 0;
    std::uint64_t frames_out = 0;
    std::uint64_t frames_in = 0;
    std::uint64_t write_calls = 0;
    std::size_t buffered = 0;
    std::size_t unread = 0;
    std::int64_t last_write_ns = 0;
    std::int64_t last_read_ns = 0;
    /// TCP payload bytes the kernel reports sent (acknowledged) and received
    /// on this link's sockets, so TLS records and handshakes included; 0 for
    /// in-process, unix, and SCTP links, and once the link is closed.
    std::uint64_t socket_bytes_out = 0;
    std::uint64_t socket_bytes_in = 0;
    std::error_code error;
    bool closed = false;
};

namespace detail
{

inline std::int64_t steady_ns() noexcept
{
    return static_cast<std::int64_t>(
        std::chrono::duration_cast<std::chrono::nanoseconds>(
            std::chrono::steady_clock::now().time_since_epoch())
            .count());
}

/// @brief Completion shared by the chunks of one write.
struct write_op
{
    async_handler h;
    std::size_t left = 0;
    std::error_code ec;
};

struct io_counters
{
    std::atomic<std::uint64_t> bytes_out{0};
    std::atomic<std::uint64_t> bytes_in{0};
    std::atomic<std::uint64_t> payload_out{0};
    std::atomic<std::uint64_t> payload_in{0};
    std::atomic<std::uint64_t> frames_out{0};
    std::atomic<std::uint64_t> frames_in{0};
    std::atomic<std::uint64_t> write_calls{0};
    std::atomic<std::int64_t> last_write_ns{0};
    std::atomic<std::int64_t> last_read_ns{0};

    void wrote(std::uint64_t wire, std::uint64_t payload, std::uint64_t frames)
    {
        bytes_out.fetch_add(wire, std::memory_order_relaxed);
        payload_out.fetch_add(payload, std::memory_order_relaxed);
        frames_out.fetch_add(frames, std::memory_order_relaxed);
        write_calls.fetch_add(1, std::memory_order_relaxed);
        last_write_ns.store(steady_ns(), std::memory_order_relaxed);
    }

    void read(std::uint64_t wire, std::uint64_t payload, std::uint64_t frames)
    {
        bytes_in.fetch_add(wire, std::memory_order_relaxed);
        payload_in.fetch_add(payload, std::memory_order_relaxed);
        frames_in.fetch_add(frames, std::memory_order_relaxed);
        last_read_ns.store(steady_ns(), std::memory_order_relaxed);
    }

    void fill(stream_stats & s) const
    {
        s.bytes_out = bytes_out.load(std::memory_order_relaxed);
        s.bytes_in = bytes_in.load(std::memory_order_relaxed);
        s.payload_out = payload_out.load(std::memory_order_relaxed);
        s.payload_in = payload_in.load(std::memory_order_relaxed);
        s.frames_out = frames_out.load(std::memory_order_relaxed);
        s.frames_in = frames_in.load(std::memory_order_relaxed);
        s.write_calls = write_calls.load(std::memory_order_relaxed);
        s.last_write_ns = last_write_ns.load(std::memory_order_relaxed);
        s.last_read_ns = last_read_ns.load(std::memory_order_relaxed);
    }
};

/// @brief `(offset, length)` pieces of a `len`-byte write.
inline std::vector<std::pair<std::size_t, std::size_t>> split_chunks(
    std::size_t len, std::size_t chunk)
{
    std::vector<std::pair<std::size_t, std::size_t>> out;
    if (len == 0)
        return out;
    const std::size_t step = chunk == 0 ? len : chunk;
    for (std::size_t off = 0; off < len; off += step)
        out.emplace_back(off, std::min(step, len - off));
    return out;
}

/// @brief One reader waiting for `n` bytes on a stream.
struct stream_waiter
{
    bool active = false;
    bool reading = false;  ///< a socket read is filling `dst` directly
    void * dst = nullptr;
    std::size_t n = 0;
    std::size_t filled = 0;
    async_handler h;
    std::error_code abort;
};

/// @brief Byte inbox with a consumed-prefix cursor.
struct stream_inbox
{
    std::vector<std::uint8_t> bytes;
    std::size_t pos = 0;

    std::size_t avail() const noexcept { return bytes.size() - pos; }

    std::size_t take(void * dst, std::size_t n, std::size_t compact_at)
    {
        const std::size_t k = std::min(n, avail());
        if (k != 0)
            std::memcpy(dst, bytes.data() + pos, k);
        pos += k;
        if (pos == bytes.size())
        {
            bytes.clear();
            pos = 0;
        }
        else if (compact_at != 0 && pos > compact_at)
        {
            bytes.erase(bytes.begin(),
                bytes.begin() + static_cast<std::ptrdiff_t>(pos));
            pos = 0;
        }
        return k;
    }
};

inline void post_handler(asio::io_context & io, async_handler h,
    std::error_code ec)
{
    if (h)
        asio::post(io, [h = std::move(h), ec]() { h(ec); });
}

/// @brief One contiguous copy of a gather list of `bytes` total.
inline std::shared_ptr<std::vector<std::uint8_t>> flatten(
    const std::vector<asio::const_buffer> & bufs, std::size_t bytes)
{
    auto flat = acquire_buffer(bytes);
    std::size_t at = 0;
    for (const auto & b : bufs)
    {
        if (b.size() != 0)
            std::memcpy(flat->data() + at, b.data(), b.size());
        at += b.size();
    }
    return flat;
}

/// @brief Add the kernel's TCP byte counters for `fd` (no-op off Linux or
///        for non-TCP sockets).
inline void add_socket_bytes(int fd, stream_stats & s)
{
#if defined(__linux__) && defined(TCP_INFO)
    tcp_info ti{};
    socklen_t len = sizeof(ti);
    if (fd >= 0 && ::getsockopt(fd, IPPROTO_TCP, TCP_INFO, &ti, &len) == 0
        && len >= offsetof(tcp_info, tcpi_bytes_received) + sizeof(ti.tcpi_bytes_received))
    {
        s.socket_bytes_out += ti.tcpi_bytes_acked;
        s.socket_bytes_in += ti.tcpi_bytes_received;
    }
#else
    (void)fd;
    (void)s;
#endif
}

/// @brief A TLS peer that closes without `close_notify` reads as a truncated
///        stream. Frames already say what is owed, so that is an ordinary end
///        of stream.
inline std::error_code end_of_stream(const std::error_code & ec)
{
    if (ec.value() == 1 && std::strcmp(ec.category().name(), "asio.ssl.stream") == 0)
        return asio::error::eof;
    return ec;
}

}  // namespace detail

/// @brief Virtual base for N duplex streams with event-driven completion.
/// @details `async_write` / `async_read` are non-blocking initiators. The
///          handler is invoked exactly once, on a thread running `context()`
///          (never inline on the caller's thread). A short read or write is
///          reported through the handler's `error_code`. After `close()`, no
///          backend copies into a caller buffer from its inbox again; pending
///          reads complete with `operation_aborted`.
class async_stream_array
{
public:
    virtual ~async_stream_array() = default;

    virtual std::size_t size() const noexcept = 0;

    virtual asio::io_context & context() noexcept = 0;

    /// @brief Write `n` bytes on stream `i`. Backends copy `src` before return.
    virtual void async_write(std::size_t i, const void * src, std::size_t n,
        async_handler h) = 0;

    /// @brief Read exactly `n` bytes on stream `i` into `dst`. `dst` must stay
    ///        valid until `h` fires. One outstanding read per stream.
    virtual void async_read(std::size_t i, void * dst, std::size_t n,
        async_handler h) = 0;

    /// @brief Write an owned buffer without copying the payload again.
    virtual void async_write_owned(std::size_t i,
        std::shared_ptr<std::vector<std::uint8_t>> buf, async_handler h)
    {
        const std::size_t n = buf ? buf->size() : 0;
        const void * p = (buf && n) ? buf->data() : nullptr;
        async_write(i, p, n, [buf = std::move(buf), h = std::move(h)](
                                 const std::error_code & ec) {
            if (h)
                h(ec);
        });
    }

    /// @brief Bytes accepted by `async_write*` and not yet on the wire (or not
    ///        yet read, for in-process backends).
    virtual std::size_t buffered_bytes() const noexcept { return 0; }

    /// @brief High-water mark for `buffered_bytes` (0 = unlimited).
    virtual std::size_t window_bytes() const noexcept { return 0; }

    virtual void set_window_bytes(std::size_t) {}

    /// @brief Backpressure that applies to lane `i` (whole connection on mux
    ///        and SCTP, that lane's socket on parallel).
    virtual std::size_t lane_buffered_bytes(std::size_t) const noexcept
    {
        return buffered_bytes();
    }

    virtual std::size_t lane_window_bytes(std::size_t) const noexcept
    {
        return window_bytes();
    }

    virtual stream_stats stats() const { return {}; }

    /// @brief Abort all I/O: pending reads and queued writes fail with
    ///        `operation_aborted`; the peer sees EOF. Destroying a backend is
    ///        graceful instead: accepted writes still reach the peer.
    virtual void close() noexcept {}
};

// ---------------------------------------------------------------------------
// Lane-range view
// ---------------------------------------------------------------------------

/// @brief Lanes `[first, first + count)` of another array, re-indexed from 0.
/// @details Lets one connection carry several logical edges (peer, ring,
///          dealer) on disjoint lanes. Does not own or close the base.
class async_stream_view final : public async_stream_array
{
public:
    async_stream_view(async_stream_array & base, std::size_t first,
        std::size_t count)
      : base_(&base), first_(first), count_(count)
    {
        if (count == 0 || first + count > base.size())
            throw std::invalid_argument("async_stream_view: lane range");
    }

    std::size_t size() const noexcept override { return count_; }
    asio::io_context & context() noexcept override { return base_->context(); }

    void async_write(std::size_t i, const void * src, std::size_t n,
        async_handler h) override
    {
        base_->async_write(map(i), src, n, std::move(h));
    }

    void async_read(std::size_t i, void * dst, std::size_t n,
        async_handler h) override
    {
        base_->async_read(map(i), dst, n, std::move(h));
    }

    void async_write_owned(std::size_t i,
        std::shared_ptr<std::vector<std::uint8_t>> buf, async_handler h) override
    {
        base_->async_write_owned(map(i), std::move(buf), std::move(h));
    }

    std::size_t buffered_bytes() const noexcept override
    {
        return base_->buffered_bytes();
    }
    std::size_t window_bytes() const noexcept override
    {
        return base_->window_bytes();
    }
    void set_window_bytes(std::size_t b) override { base_->set_window_bytes(b); }
    std::size_t lane_buffered_bytes(std::size_t i) const noexcept override
    {
        return base_->lane_buffered_bytes(first_ + i);
    }
    std::size_t lane_window_bytes(std::size_t i) const noexcept override
    {
        return base_->lane_window_bytes(first_ + i);
    }
    stream_stats stats() const override { return base_->stats(); }

private:
    std::size_t map(std::size_t i) const
    {
        if (i >= count_)
            throw std::out_of_range("async_stream_view index");
        return first_ + i;
    }

    async_stream_array * base_ = nullptr;
    std::size_t first_ = 0;
    std::size_t count_ = 0;
};

// ---------------------------------------------------------------------------
// In-process backend (paired ends; each end may own its own io_context)
// ---------------------------------------------------------------------------

/// @brief Shared state for a pair of in-process async ends.
/// @details Direction `d` carries bytes written by side `d` to side `1 - d`.
///          A write that would push a direction's unread bytes past that
///          side's window waits (FIFO) until the reader consumes. Completions
///          post to the owning side's `io_context`. Closing one side fails its
///          pending I/O and gives the other side EOF once its data runs out.
class async_dual_memory_hub
{
public:
    async_dual_memory_hub(asio::io_context & io_a, asio::io_context & io_b,
        std::size_t streams, std::size_t window = default_wire_window())
      : n_(streams)
    {
        if (streams == 0)
            throw std::invalid_argument("async memory hub: empty");
        io_[0] = &io_a;
        io_[1] = &io_b;
        for (int d = 0; d < 2; ++d)
        {
            in_[d].resize(streams);
            wait_[d].resize(streams);
            window_[d] = window;
        }
    }

    async_dual_memory_hub(asio::io_context & io, std::size_t streams)
      : async_dual_memory_hub(io, io, streams)
    {
    }

    std::size_t size() const noexcept { return n_; }
    asio::io_context & context(int side) noexcept { return *io_[side]; }

    std::size_t window(int writer_side) const
    {
        std::lock_guard<std::mutex> lock(mu_);
        return window_[writer_side];
    }

    void set_window(int writer_side, std::size_t bytes)
    {
        std::lock_guard<std::mutex> lock(mu_);
        window_[writer_side] = bytes;
        release_locked(writer_side);
    }

    /// @brief Bytes `writer_side` wrote that its peer has not read yet.
    std::size_t unread_side(int writer_side) const
    {
        std::lock_guard<std::mutex> lock(mu_);
        return unread_[writer_side] + pending_bytes_[writer_side];
    }

    stream_stats stats(int side) const
    {
        stream_stats s;
        std::lock_guard<std::mutex> lock(mu_);
        s.bytes_out = s.payload_out = out_bytes_[side];
        s.bytes_in = s.payload_in = in_bytes_[side];
        s.frames_out = s.write_calls = out_frames_[side];
        s.frames_in = in_frames_[side];
        s.buffered = unread_[side] + pending_bytes_[side];
        s.unread = unread_[1 - side];
        s.last_write_ns = last_write_[side];
        s.last_read_ns = last_read_[side];
        s.closed = closed_[side];
        return s;
    }

    void do_write(int ws, std::size_t i, const void * src, std::size_t n,
        async_handler h)
    {
        check(i);
        std::lock_guard<std::mutex> lock(mu_);
        if (closed_[ws])
        {
            detail::post_handler(*io_[ws], std::move(h),
                asio::error::operation_aborted);
            return;
        }
        if (closed_[1 - ws])
        {
            detail::post_handler(*io_[ws], std::move(h), asio::error::broken_pipe);
            return;
        }
        if (n == 0)
        {
            detail::post_handler(*io_[ws], std::move(h), {});
            return;
        }
        const auto * p = static_cast<const std::uint8_t *>(src);
        if (!pending_[ws].empty() || !fits_locked(ws, n))
        {
            pending_[ws].push_back(
                pending_write{i, std::vector<std::uint8_t>(p, p + n), std::move(h)});
            pending_bytes_[ws] += n;
            return;
        }
        deliver_locked(ws, i, p, n);
        detail::post_handler(*io_[ws], std::move(h), {});
        release_locked(ws);
    }

    void do_read(int rs, std::size_t i, void * dst, std::size_t n,
        async_handler h)
    {
        check(i);
        const int d = 1 - rs;
        std::lock_guard<std::mutex> lock(mu_);
        if (closed_[rs])
        {
            detail::post_handler(*io_[rs], std::move(h),
                asio::error::operation_aborted);
            return;
        }
        detail::stream_waiter & w = wait_[d][i];
        if (w.active)
            throw std::logic_error("async memory stream: overlapping read on "
                "stream " + std::to_string(i));
        const std::size_t k = in_[d][i].take(dst, n, compact_);
        consumed_locked(d, rs, k);
        if (k == n)
        {
            detail::post_handler(*io_[rs], std::move(h), {});
            release_locked(d);
            return;
        }
        if (closed_[d] && pending_[d].empty())
        {
            detail::post_handler(*io_[rs], std::move(h), asio::error::eof);
            return;
        }
        w.active = true;
        w.dst = dst;
        w.n = n;
        w.filled = k;
        w.h = std::move(h);
        release_locked(d);
    }

    void close_side(int s)
    {
        std::lock_guard<std::mutex> lock(mu_);
        if (closed_[s])
            return;
        closed_[s] = true;
        // Reads by `s` fail.
        for (auto & w : wait_[1 - s])
        {
            if (!w.active)
                continue;
            w.active = false;
            detail::post_handler(*io_[s], std::move(w.h),
                asio::error::operation_aborted);
        }
        // Accepted writes by `s` still reach the peer (window no longer applies).
        while (!pending_[s].empty())
        {
            pending_write pw = std::move(pending_[s].front());
            pending_[s].pop_front();
            pending_bytes_[s] -= pw.bytes.size();
            deliver_locked(s, pw.i, pw.bytes.data(), pw.bytes.size());
            detail::post_handler(*io_[s], std::move(pw.h), {});
        }
        // Peer reads that the delivered bytes cannot satisfy end with EOF.
        for (auto & w : wait_[s])
        {
            if (!w.active)
                continue;
            w.active = false;
            detail::post_handler(*io_[1 - s], std::move(w.h), asio::error::eof);
        }
    }

private:
    struct pending_write
    {
        std::size_t i = 0;
        std::vector<std::uint8_t> bytes;
        async_handler h;
    };

    void check(std::size_t i) const
    {
        if (i >= n_)
            throw std::out_of_range("async memory stream index "
                + std::to_string(i));
    }

    bool fits_locked(int ws, std::size_t n) const
    {
        const std::size_t w = window_[ws];
        return w == 0 || unread_[ws] == 0 || unread_[ws] + n <= w;
    }

    void deliver_locked(int d, std::size_t i, const std::uint8_t * p,
        std::size_t n)
    {
        auto & box = in_[d][i];
        box.bytes.insert(box.bytes.end(), p, p + n);
        unread_[d] += n;
        out_bytes_[d] += n;
        ++out_frames_[d];
        last_write_[d] = detail::steady_ns();
        satisfy_locked(d, i);
    }

    void consumed_locked(int d, int rs, std::size_t k)
    {
        if (k == 0)
            return;
        unread_[d] -= std::min(unread_[d], k);
        in_bytes_[rs] += k;
        ++in_frames_[rs];
        last_read_[rs] = detail::steady_ns();
    }

    void satisfy_locked(int d, std::size_t i)
    {
        detail::stream_waiter & w = wait_[d][i];
        if (!w.active)
            return;
        const int rs = 1 - d;
        const std::size_t k = in_[d][i].take(
            static_cast<std::uint8_t *>(w.dst) + w.filled, w.n - w.filled,
            compact_);
        consumed_locked(d, rs, k);
        w.filled += k;
        if (w.filled == w.n)
        {
            w.active = false;
            detail::post_handler(*io_[rs], std::move(w.h), {});
        }
    }

    void release_locked(int d)
    {
        while (!pending_[d].empty() && !closed_[1 - d]
            && fits_locked(d, pending_[d].front().bytes.size()))
        {
            pending_write pw = std::move(pending_[d].front());
            pending_[d].pop_front();
            pending_bytes_[d] -= pw.bytes.size();
            deliver_locked(d, pw.i, pw.bytes.data(), pw.bytes.size());
            detail::post_handler(*io_[d], std::move(pw.h), {});
        }
    }

    asio::io_context * io_[2] = {nullptr, nullptr};
    std::size_t n_ = 0;
    std::size_t compact_ = wire_policy{}.compact_bytes;
    mutable std::mutex mu_;
    std::vector<detail::stream_inbox> in_[2];
    std::vector<detail::stream_waiter> wait_[2];
    std::deque<pending_write> pending_[2];
    std::size_t pending_bytes_[2] = {0, 0};
    std::size_t unread_[2] = {0, 0};
    std::size_t window_[2] = {0, 0};
    bool closed_[2] = {false, false};
    std::uint64_t out_bytes_[2] = {0, 0};
    std::uint64_t in_bytes_[2] = {0, 0};
    std::uint64_t out_frames_[2] = {0, 0};
    std::uint64_t in_frames_[2] = {0, 0};
    std::int64_t last_write_[2] = {0, 0};
    std::int64_t last_read_[2] = {0, 0};
};

/// @brief One end of an in-process async pair. Destroying it closes its side.
class async_dual_memory_stream_array final : public async_stream_array
{
public:
    async_dual_memory_stream_array(std::shared_ptr<async_dual_memory_hub> hub,
        bool side_a)
      : hub_(std::move(hub)), side_(side_a ? 0 : 1)
    {
        if (!hub_)
            throw std::invalid_argument("async memory stream needs a hub");
    }

    async_dual_memory_stream_array(async_dual_memory_stream_array && o) noexcept
      : hub_(std::move(o.hub_)), side_(o.side_)
    {
    }

    async_dual_memory_stream_array & operator=(
        async_dual_memory_stream_array && o) noexcept
    {
        if (this != &o)
        {
            if (hub_)
                hub_->close_side(side_);
            hub_ = std::move(o.hub_);
            side_ = o.side_;
        }
        return *this;
    }

    async_dual_memory_stream_array(const async_dual_memory_stream_array &) = delete;
    async_dual_memory_stream_array & operator=(
        const async_dual_memory_stream_array &) = delete;

    ~async_dual_memory_stream_array() override
    {
        if (hub_)
            hub_->close_side(side_);
    }

    std::size_t size() const noexcept override { return hub_->size(); }
    asio::io_context & context() noexcept override
    {
        return hub_->context(side_);
    }

    void async_write(std::size_t i, const void * src, std::size_t n,
        async_handler h) override
    {
        hub_->do_write(side_, i, src, n, std::move(h));
    }

    void async_read(std::size_t i, void * dst, std::size_t n,
        async_handler h) override
    {
        hub_->do_read(side_, i, dst, n, std::move(h));
    }

    std::size_t buffered_bytes() const noexcept override
    {
        return hub_->unread_side(side_);
    }

    std::size_t window_bytes() const noexcept override
    {
        return hub_->window(side_);
    }

    void set_window_bytes(std::size_t bytes) override
    {
        hub_->set_window(side_, bytes);
    }

    stream_stats stats() const override { return hub_->stats(side_); }

    void close() noexcept override
    {
        if (hub_)
            hub_->close_side(side_);
    }

private:
    std::shared_ptr<async_dual_memory_hub> hub_;
    int side_ = 0;
};

using async_memory_hub = async_dual_memory_hub;
using async_memory_stream_array = async_dual_memory_stream_array;

/// @brief Two ends sharing one `io_context`.
HEDLEY_WARN_UNUSED_RESULT
inline std::pair<async_memory_stream_array, async_memory_stream_array>
make_async_memory_stream_pair(asio::io_context & io, std::size_t streams)
{
    auto hub = std::make_shared<async_dual_memory_hub>(io, streams);
    return {async_memory_stream_array(hub, true),
        async_memory_stream_array(hub, false)};
}

/// @brief Two ends, each completing on its own `io_context`.
HEDLEY_WARN_UNUSED_RESULT
inline std::pair<async_dual_memory_stream_array, async_dual_memory_stream_array>
make_async_dual_memory_stream_pair(asio::io_context & io_a,
    asio::io_context & io_b, std::size_t streams,
    std::size_t window = default_wire_window())
{
    auto hub = std::make_shared<async_dual_memory_hub>(io_a, io_b, streams,
        window);
    return {async_dual_memory_stream_array(hub, true),
        async_dual_memory_stream_array(hub, false)};
}

// ---------------------------------------------------------------------------
// Mux backend (N lanes on ONE TCP socket)
// ---------------------------------------------------------------------------

/// @brief N lanes multiplexed on one TCP socket, fully overlapped.
/// @details Wire frame: `u32 total` (little-endian) `|| stream_frame_hdr`
///          (`u16 index`, `u32 length`) `|| payload`, with
///          `total = sizeof(stream_frame_hdr) + length`. The synchronous
///          `mux_stream_array` writes the same frames, so the two interoperate.
///          All socket operations run on one strand. The window covers the
///          whole connection: every lane shares one TCP byte stream.
template <typename Stream>
class basic_async_mux_stream_array final : public async_stream_array
{
public:
    basic_async_mux_stream_array(asio::io_context & io, Stream sock,
        unsigned self_id, unsigned peer_id, std::size_t nstreams,
        const wire_policy & pol = {})
    {
        if (nstreams == 0)
            throw std::invalid_argument("async_mux_stream_array empty");
        if (nstreams > 0xffffu)
            throw std::invalid_argument("async_mux_stream_array: lane ids are u16");
        if (self_id == peer_id)
            throw std::invalid_argument("async_mux_stream_array roles");
        pol.validate();
        tune_stream(sock, pol.socket);
        impl_ = std::make_shared<impl>(io, std::move(sock), nstreams, pol);
        impl_->start();
    }

    basic_async_mux_stream_array(const basic_async_mux_stream_array &) = delete;
    basic_async_mux_stream_array & operator=(const basic_async_mux_stream_array &)
        = delete;

    ~basic_async_mux_stream_array() override
    {
        if (impl_)
            impl_->close_graceful();
    }

    std::size_t size() const noexcept override { return impl_->n; }
    asio::io_context & context() noexcept override { return impl_->io; }

    void async_write(std::size_t i, const void * src, std::size_t n,
        async_handler h) override
    {
        impl_->write(i, n == 0 ? nullptr : copy_buffer(src, n), std::move(h));
    }

    void async_write_owned(std::size_t i,
        std::shared_ptr<std::vector<std::uint8_t>> buf, async_handler h) override
    {
        impl_->write(i, std::move(buf), std::move(h));
    }

    void async_read(std::size_t i, void * dst, std::size_t n,
        async_handler h) override
    {
        impl_->read(i, dst, n, std::move(h));
    }

    std::size_t buffered_bytes() const noexcept override
    {
        return impl_->buffered.load(std::memory_order_relaxed);
    }

    std::size_t window_bytes() const noexcept override
    {
        return impl_->window.load(std::memory_order_relaxed);
    }

    void set_window_bytes(std::size_t bytes) override
    {
        impl_->window.store(bytes, std::memory_order_relaxed);
    }

    stream_stats stats() const override { return impl_->stats(); }

    void close() noexcept override
    {
        impl_->close(asio::error::operation_aborted);
    }

private:
    struct impl : std::enable_shared_from_this<impl>
    {
        static constexpr std::size_t hdr_wire = 4 + sizeof(stream_frame_hdr);

        struct chunk
        {
            std::uint32_t total_le = 0;
            stream_frame_hdr hdr{};
            std::shared_ptr<std::vector<std::uint8_t>> payload;
            std::size_t off = 0;
            std::size_t len = 0;
            std::shared_ptr<detail::write_op> op;
        };

        impl(asio::io_context & io_, Stream sock_,
            std::size_t nstreams, const wire_policy & pol_)
          : io(io_),
            sock(std::move(sock_)),
            strand(asio::make_strand(io_)),
            n(nstreams),
            pol(pol_),
            inbox(nstreams),
            wait(nstreams),
            outq(nstreams),
            window(pol_.window_bytes)
        {
            fd = base_socket(sock).native_handle();
        }

        void check(std::size_t i) const
        {
            if (i >= n)
                throw std::out_of_range("async_mux_stream_array index "
                    + std::to_string(i));
        }

        void start()
        {
            auto self = this->shared_from_this();
            asio::post(strand, [self] { self->read_hdr(); });
        }

        // --- reads ---------------------------------------------------------
        void read(std::size_t i, void * dst, std::size_t nbytes, async_handler h)
        {
            check(i);
            std::lock_guard<std::mutex> lock(mu);
            if (fail)
            {
                detail::post_handler(io, std::move(h), fail);
                return;
            }
            detail::stream_waiter & w = wait[i];
            if (w.active)
                throw std::logic_error(
                    "async_mux_stream_array: overlapping read on stream "
                    + std::to_string(i));
            const std::size_t k = inbox[i].take(dst, nbytes, pol.compact_bytes);
            if (k == nbytes)
            {
                detail::post_handler(io, std::move(h), {});
                return;
            }
            w.active = true;
            w.reading = false;
            w.dst = dst;
            w.n = nbytes;
            w.filled = k;
            w.h = std::move(h);
            w.abort.clear();
        }

        void satisfy_locked(std::size_t i)
        {
            detail::stream_waiter & w = wait[i];
            if (!w.active || w.reading)
                return;
            w.filled += inbox[i].take(static_cast<std::uint8_t *>(w.dst) + w.filled,
                w.n - w.filled, pol.compact_bytes);
            if (w.filled == w.n)
            {
                w.active = false;
                detail::post_handler(io, std::move(w.h), {});
            }
        }

        void read_hdr()
        {
            auto self = this->shared_from_this();
            asio::async_read(sock, asio::buffer(hdr_buf, hdr_wire),
                asio::bind_executor(strand,
                    [self](const std::error_code & ec, std::size_t) {
                        self->on_hdr(ec);
                    }));
        }

        void on_hdr(const std::error_code & ec)
        {
            if (ec)
            {
                deliver_error(detail::end_of_stream(ec));
                return;
            }
            std::uint32_t total = 0;
            std::memcpy(&total, hdr_buf, 4);
            stream_frame_hdr hdr{};
            std::memcpy(&hdr, hdr_buf + 4, sizeof(hdr));
            if (hdr.index >= n || total != sizeof(stream_frame_hdr) + hdr.length)
            {
                deliver_error(std::make_error_code(std::errc::protocol_error));
                return;
            }
            if (hdr.length > pol.max_frame)
            {
                deliver_error(std::make_error_code(std::errc::message_size));
                return;
            }
            cur_index = hdr.index;
            cur_len = hdr.length;
            if (cur_len == 0)
            {
                counters.read(hdr_wire, 0, 1);
                read_hdr();
                return;
            }
            void * dst = nullptr;
            cur_direct = false;
            {
                std::lock_guard<std::mutex> lock(mu);
                if (closed)
                    return;
                detail::stream_waiter & w = wait[cur_index];
                if (w.active && !w.reading && inbox[cur_index].avail() == 0
                    && w.filled + cur_len <= w.n)
                {
                    w.reading = true;
                    cur_direct = true;
                    dst = static_cast<std::uint8_t *>(w.dst) + w.filled;
                }
            }
            if (!cur_direct)
            {
                rbuf.resize(cur_len);
                dst = rbuf.data();
            }
            auto self = this->shared_from_this();
            asio::async_read(sock, asio::buffer(dst, cur_len),
                asio::bind_executor(strand,
                    [self](const std::error_code & xec, std::size_t) {
                        self->on_body(xec);
                    }));
        }

        void on_body(const std::error_code & ec)
        {
            if (cur_direct)
            {
                std::lock_guard<std::mutex> lock(mu);
                detail::stream_waiter & w = wait[cur_index];
                w.reading = false;
                if (w.active && (ec || w.abort))
                {
                    w.active = false;
                    detail::post_handler(io, std::move(w.h), ec ? ec : w.abort);
                }
                else if (w.active && !ec)
                {
                    w.filled += cur_len;
                    if (w.filled == w.n)
                    {
                        w.active = false;
                        detail::post_handler(io, std::move(w.h), {});
                    }
                }
            }
            if (ec)
            {
                deliver_error(detail::end_of_stream(ec));
                return;
            }
            counters.read(hdr_wire + cur_len, cur_len, 1);
            if (!cur_direct)
            {
                std::lock_guard<std::mutex> lock(mu);
                if (closed)
                    return;
                auto & box = inbox[cur_index].bytes;
                box.insert(box.end(), rbuf.begin(), rbuf.end());
                satisfy_locked(cur_index);
            }
            {
                std::lock_guard<std::mutex> lock(mu);
                if (closed)
                    return;
            }
            if (pol.socket.quickack)
                rearm_quickack(base_socket(sock));
            read_hdr();
        }

        void fail_waiters_locked(const std::error_code & ec)
        {
            for (auto & w : wait)
            {
                if (!w.active)
                    continue;
                if (w.reading)
                {
                    // Completes once the in-flight socket read ends: `dst`
                    // must stay owned by the handler until then.
                    w.abort = ec;
                    continue;
                }
                w.active = false;
                detail::post_handler(io, std::move(w.h), ec);
            }
        }

        void deliver_error(const std::error_code & ec)
        {
            {
                std::lock_guard<std::mutex> lock(mu);
                if (!fail)
                    fail = ec;
                fail_waiters_locked(fail);
            }
            auto self = this->shared_from_this();
            asio::post(strand, [self, ec] {
                std::error_code e;
                base_socket(self->sock).shutdown(asio::socket_base::shutdown_both, e);
                self->fail_queued(ec);
            });
        }

        void close(const std::error_code & ec)
        {
            {
                std::lock_guard<std::mutex> lock(mu);
                if (!fail)
                    fail = ec;
                if (closed && aborted)
                    return;
                closed = true;
                aborted = true;
                fail_waiters_locked(ec);
            }
            auto self = this->shared_from_this();
            asio::post(strand, [self, ec] {
                std::error_code e;
                base_socket(self->sock).shutdown(asio::socket_base::shutdown_both, e);
                base_socket(self->sock).close(e);
                self->fail_queued(ec);
            });
        }

        /// @brief Stop reads now; close the socket once queued writes drain.
        void close_graceful()
        {
            {
                std::lock_guard<std::mutex> lock(mu);
                if (closed)
                    return;
                closed = true;
                fail_waiters_locked(asio::error::operation_aborted);
            }
            auto self = this->shared_from_this();
            asio::post(strand, [self] {
                self->draining = true;
                if (!self->writing)
                    self->finish_close();
            });
        }

        void finish_close()
        {
            std::error_code e;
            base_socket(sock).shutdown(asio::socket_base::shutdown_both, e);
            base_socket(sock).close(e);
        }

        stream_stats stats() const
        {
            stream_stats s;
            counters.fill(s);
            s.buffered = buffered.load(std::memory_order_relaxed);
            std::lock_guard<std::mutex> lock(mu);
            for (const auto & b : inbox)
                s.unread += b.avail();
            s.error = fail;
            s.closed = closed;
            if (!closed)
                detail::add_socket_bytes(fd, s);
            return s;
        }

        // --- writes (strand) ----------------------------------------------
        void write(std::size_t i, std::shared_ptr<std::vector<std::uint8_t>> payload,
            async_handler h)
        {
            check(i);
            const std::size_t len = payload ? payload->size() : 0;
            {
                std::lock_guard<std::mutex> lock(mu);
                if (fail || closed)
                {
                    detail::post_handler(io, std::move(h),
                        fail ? fail : asio::error::operation_aborted);
                    return;
                }
            }
            if (len == 0)
            {
                detail::post_handler(io, std::move(h), {});
                return;
            }
            if (pol.chunk_bytes == 0 && len > 0xffffffffu - sizeof(stream_frame_hdr))
                throw std::invalid_argument("async_mux_stream_array frame too large");
            const auto pieces = detail::split_chunks(len, pol.chunk_bytes);
            auto op = std::make_shared<detail::write_op>();
            op->h = std::move(h);
            op->left = pieces.size();
            std::vector<chunk> cs;
            cs.reserve(pieces.size());
            std::size_t wire = 0;
            for (const auto & pc : pieces)
            {
                chunk c;
                c.hdr.index = static_cast<std::uint16_t>(i);
                c.hdr.length = static_cast<std::uint32_t>(pc.second);
                c.total_le = static_cast<std::uint32_t>(
                    sizeof(stream_frame_hdr) + pc.second);
                c.payload = payload;
                c.off = pc.first;
                c.len = pc.second;
                c.op = op;
                wire += hdr_wire + pc.second;
                cs.push_back(std::move(c));
            }
            buffered.fetch_add(wire, std::memory_order_relaxed);
            auto self = this->shared_from_this();
            asio::post(strand, [self, i, cs = std::move(cs)]() mutable {
                self->enqueue(i, std::move(cs));
            });
        }

        void enqueue(std::size_t i, std::vector<chunk> cs)
        {
            std::error_code ec;
            {
                std::lock_guard<std::mutex> lock(mu);
                ec = fail;
            }
            if (ec)
            {
                for (auto & c : cs)
                    finish_chunk(c, ec);
                return;
            }
            for (auto & c : cs)
                outq[i].push_back(std::move(c));
            if (!writing)
                start_write();
        }

        void finish_chunk(chunk & c, const std::error_code & ec)
        {
            buffered.fetch_sub(std::min(buffered.load(std::memory_order_relaxed),
                                   hdr_wire + c.len),
                std::memory_order_relaxed);
            auto & op = *c.op;
            if (ec && !op.ec)
                op.ec = ec;
            if (--op.left == 0)
                detail::post_handler(io, std::move(op.h), op.ec);
        }

        void start_write()
        {
            inflight.clear();
            std::size_t bytes = 0;
            bool stop = false;
            while (!stop && inflight.size() < pol.coalesce_frames)
            {
                bool found = false;
                for (std::size_t k = 0; k < n; ++k)
                {
                    const std::size_t lane = (rr + k) % n;
                    if (outq[lane].empty())
                        continue;
                    const std::size_t wire = hdr_wire + outq[lane].front().len;
                    if (!inflight.empty() && bytes + wire > pol.coalesce_bytes)
                    {
                        stop = true;
                        break;
                    }
                    inflight.push_back(std::move(outq[lane].front()));
                    outq[lane].pop_front();
                    bytes += wire;
                    rr = (lane + 1) % n;
                    found = true;
                    break;
                }
                if (!found)
                    break;
            }
            if (inflight.empty())
            {
                writing = false;
                if (draining)
                    finish_close();
                return;
            }
            writing = true;
            inflight_wire = bytes;
            std::vector<asio::const_buffer> bufs;
            bufs.reserve(inflight.size() * 3);
            for (auto & c : inflight)
            {
                bufs.push_back(asio::buffer(&c.total_le, 4));
                bufs.push_back(asio::buffer(&c.hdr, sizeof(c.hdr)));
                bufs.push_back(asio::buffer(c.payload->data() + c.off, c.len));
            }
            auto self = this->shared_from_this();
            if constexpr (detail::is_layered_stream<Stream>::value)
            {
                // TLS seals each buffer as its own record; send the batch as one.
                auto flat = detail::flatten(bufs, bytes);
                asio::async_write(sock, asio::buffer(*flat),
                    asio::bind_executor(strand,
                        [self, flat](const std::error_code & ec, std::size_t) {
                            self->on_written(ec);
                        }));
                return;
            }
            asio::async_write(sock, bufs,
                asio::bind_executor(strand,
                    [self](const std::error_code & ec, std::size_t) {
                        self->on_written(ec);
                    }));
        }

        void on_written(const std::error_code & ec)
        {
            if (!ec)
            {
                std::size_t payload = 0;
                for (const auto & c : inflight)
                    payload += c.len;
                counters.wrote(inflight_wire, payload, inflight.size());
            }
            for (auto & c : inflight)
                finish_chunk(c, ec);
            inflight.clear();
            if (ec)
            {
                writing = false;
                deliver_error(detail::end_of_stream(ec));
                if (draining)
                    finish_close();
                return;
            }
            if (pol.socket.quickack)
                rearm_quickack(base_socket(sock));
            start_write();
        }

        void fail_queued(const std::error_code & ec)
        {
            for (auto & q : outq)
            {
                for (auto & c : q)
                    finish_chunk(c, ec);
                q.clear();
            }
        }

        asio::io_context & io;
        Stream sock;
        int fd = -1;
        asio::strand<asio::io_context::executor_type> strand;
        std::size_t n = 0;
        wire_policy pol;

        mutable std::mutex mu;
        std::vector<detail::stream_inbox> inbox;
        std::vector<detail::stream_waiter> wait;
        std::error_code fail;
        bool closed = false;
        bool aborted = false;

        // strand-only
        std::vector<std::deque<chunk>> outq;
        std::size_t rr = 0;
        bool writing = false;
        bool draining = false;
        std::vector<chunk> inflight;
        std::size_t inflight_wire = 0;
        std::uint8_t hdr_buf[hdr_wire]{};
        std::vector<std::uint8_t> rbuf;
        std::size_t cur_index = 0;
        std::size_t cur_len = 0;
        bool cur_direct = false;

        std::atomic<std::size_t> buffered{0};
        std::atomic<std::size_t> window;
        detail::io_counters counters;
    };

    std::shared_ptr<impl> impl_;
};

/// @brief Mux over a plain TCP socket.
using async_mux_stream_array = basic_async_mux_stream_array<asio::ip::tcp::socket>;

// ---------------------------------------------------------------------------
// Parallel backend (one socket PER lane)
// ---------------------------------------------------------------------------

/// @brief One dedicated socket per lane; each lane fully independent.
/// @details Wire frame per socket: `u32 length || payload`. Each socket has its
///          own strand, write queue, window, and read loop. Queued frames on a
///          socket are written in one gathered syscall.
template <typename Socket>
class basic_async_parallel_stream_array final : public async_stream_array
{
public:
    basic_async_parallel_stream_array(asio::io_context & io,
        std::vector<Socket> socks, const wire_policy & pol = {})
    {
        if (socks.empty())
            throw std::invalid_argument("async_parallel_stream_array empty");
        pol.validate();
        impl_ = std::make_shared<impl>(io, std::move(socks), pol);
        impl_->start();
    }

    basic_async_parallel_stream_array(const basic_async_parallel_stream_array &)
        = delete;
    basic_async_parallel_stream_array & operator=(
        const basic_async_parallel_stream_array &) = delete;

    ~basic_async_parallel_stream_array() override
    {
        if (impl_)
            impl_->close_graceful();
    }

    std::size_t size() const noexcept override { return impl_->conns.size(); }
    asio::io_context & context() noexcept override { return impl_->io; }

    void async_write(std::size_t i, const void * src, std::size_t n,
        async_handler h) override
    {
        impl_->write(i, n == 0 ? nullptr : copy_buffer(src, n), std::move(h));
    }

    void async_write_owned(std::size_t i,
        std::shared_ptr<std::vector<std::uint8_t>> buf, async_handler h) override
    {
        impl_->write(i, std::move(buf), std::move(h));
    }

    void async_read(std::size_t i, void * dst, std::size_t n,
        async_handler h) override
    {
        impl_->read(i, dst, n, std::move(h));
    }

    std::size_t buffered_bytes() const noexcept override
    {
        std::size_t total = 0;
        for (const auto & c : impl_->conns)
            total += c->buffered.load(std::memory_order_relaxed);
        return total;
    }

    /// @brief Sum of per-socket windows (matches `buffered_bytes`).
    std::size_t window_bytes() const noexcept override
    {
        std::size_t total = 0;
        for (const auto & c : impl_->conns)
        {
            const std::size_t w = c->window.load(std::memory_order_relaxed);
            if (w == 0)
                return 0;
            total += w;
        }
        return total;
    }

    /// @brief Set every socket's window to `bytes`.
    void set_window_bytes(std::size_t bytes) override
    {
        for (auto & c : impl_->conns)
            c->window.store(bytes, std::memory_order_relaxed);
    }

    std::size_t lane_buffered_bytes(std::size_t i) const noexcept override
    {
        return i < impl_->conns.size()
            ? impl_->conns[i]->buffered.load(std::memory_order_relaxed)
            : 0;
    }

    std::size_t lane_window_bytes(std::size_t i) const noexcept override
    {
        return i < impl_->conns.size()
            ? impl_->conns[i]->window.load(std::memory_order_relaxed)
            : 0;
    }

    stream_stats stats() const override { return impl_->stats(); }

    void close() noexcept override
    {
        impl_->close(asio::error::operation_aborted);
    }

private:
    struct chunk
    {
        std::uint32_t len_le = 0;
        std::shared_ptr<std::vector<std::uint8_t>> payload;
        std::size_t off = 0;
        std::size_t len = 0;
        std::shared_ptr<detail::write_op> op;
    };

    struct conn
    {
        conn(Socket s, asio::io_context & io, const wire_policy & pol)
          : sock(std::move(s)),
            strand(asio::make_strand(io)),
            window(pol.window_bytes)
        {
            tune_stream(sock, pol.socket);
            fd = base_socket(sock).native_handle();
        }

        Socket sock;
        int fd = -1;
        asio::strand<asio::io_context::executor_type> strand;
        detail::stream_inbox inbox;          // guarded by impl::mu
        detail::stream_waiter wait;          // guarded by impl::mu
        std::deque<chunk> outq;              // strand
        std::vector<chunk> inflight;         // strand
        std::size_t inflight_wire = 0;       // strand
        bool writing = false;                // strand
        bool draining = false;               // strand
        std::uint32_t rlen_le = 0;           // strand
        std::vector<std::uint8_t> rbuf;      // strand
        std::size_t cur_len = 0;             // strand
        bool cur_direct = false;             // strand
        std::atomic<std::size_t> buffered{0};
        std::atomic<std::size_t> window;
    };

    struct impl : std::enable_shared_from_this<impl>
    {
        impl(asio::io_context & io_, std::vector<Socket> socks,
            const wire_policy & pol_)
          : io(io_), pol(pol_)
        {
            conns.reserve(socks.size());
            for (auto & s : socks)
                conns.push_back(std::make_unique<conn>(std::move(s), io_, pol_));
        }

        void check(std::size_t i) const
        {
            if (i >= conns.size())
                throw std::out_of_range("async_parallel_stream_array index "
                    + std::to_string(i));
        }

        void start()
        {
            for (std::size_t i = 0; i < conns.size(); ++i)
            {
                auto self = this->shared_from_this();
                asio::post(conns[i]->strand, [self, i] { self->read_len(i); });
            }
        }

        // --- reads ---------------------------------------------------------
        void read(std::size_t i, void * dst, std::size_t nbytes, async_handler h)
        {
            check(i);
            conn & c = *conns[i];
            std::lock_guard<std::mutex> lock(mu);
            if (fail)
            {
                detail::post_handler(io, std::move(h), fail);
                return;
            }
            if (c.wait.active)
                throw std::logic_error(
                    "async_parallel_stream_array: overlapping read on stream "
                    + std::to_string(i));
            const std::size_t k = c.inbox.take(dst, nbytes, pol.compact_bytes);
            if (k == nbytes)
            {
                detail::post_handler(io, std::move(h), {});
                return;
            }
            c.wait.active = true;
            c.wait.reading = false;
            c.wait.dst = dst;
            c.wait.n = nbytes;
            c.wait.filled = k;
            c.wait.h = std::move(h);
            c.wait.abort.clear();
        }

        void satisfy_locked(conn & c)
        {
            auto & w = c.wait;
            if (!w.active || w.reading)
                return;
            w.filled += c.inbox.take(static_cast<std::uint8_t *>(w.dst) + w.filled,
                w.n - w.filled, pol.compact_bytes);
            if (w.filled == w.n)
            {
                w.active = false;
                detail::post_handler(io, std::move(w.h), {});
            }
        }

        void read_len(std::size_t i)
        {
            auto self = this->shared_from_this();
            conn & c = *conns[i];
            asio::async_read(c.sock, asio::buffer(&c.rlen_le, 4),
                asio::bind_executor(c.strand,
                    [self, i](const std::error_code & ec, std::size_t) {
                        self->on_len(i, ec);
                    }));
        }

        void on_len(std::size_t i, const std::error_code & ec)
        {
            conn & c = *conns[i];
            if (ec)
            {
                deliver_error(detail::end_of_stream(ec));
                return;
            }
            std::uint32_t len = 0;
            std::memcpy(&len, &c.rlen_le, 4);
            if (len > pol.max_frame)
            {
                deliver_error(std::make_error_code(std::errc::message_size));
                return;
            }
            c.cur_len = len;
            if (len == 0)
            {
                counters.read(4, 0, 1);
                read_len(i);
                return;
            }
            void * dst = nullptr;
            c.cur_direct = false;
            {
                std::lock_guard<std::mutex> lock(mu);
                if (closed)
                    return;
                auto & w = c.wait;
                if (w.active && !w.reading && c.inbox.avail() == 0
                    && w.filled + len <= w.n)
                {
                    w.reading = true;
                    c.cur_direct = true;
                    dst = static_cast<std::uint8_t *>(w.dst) + w.filled;
                }
            }
            if (!c.cur_direct)
            {
                c.rbuf.resize(len);
                dst = c.rbuf.data();
            }
            auto self = this->shared_from_this();
            asio::async_read(c.sock, asio::buffer(dst, len),
                asio::bind_executor(c.strand,
                    [self, i](const std::error_code & xec, std::size_t) {
                        self->on_body(i, xec);
                    }));
        }

        void on_body(std::size_t i, const std::error_code & ec)
        {
            conn & c = *conns[i];
            if (c.cur_direct)
            {
                std::lock_guard<std::mutex> lock(mu);
                auto & w = c.wait;
                w.reading = false;
                if (w.active && (ec || w.abort))
                {
                    w.active = false;
                    detail::post_handler(io, std::move(w.h), ec ? ec : w.abort);
                }
                else if (w.active && !ec)
                {
                    w.filled += c.cur_len;
                    if (w.filled == w.n)
                    {
                        w.active = false;
                        detail::post_handler(io, std::move(w.h), {});
                    }
                }
            }
            if (ec)
            {
                deliver_error(detail::end_of_stream(ec));
                return;
            }
            counters.read(4 + c.cur_len, c.cur_len, 1);
            {
                std::lock_guard<std::mutex> lock(mu);
                if (closed)
                    return;
                if (!c.cur_direct)
                {
                    c.inbox.bytes.insert(c.inbox.bytes.end(), c.rbuf.begin(),
                        c.rbuf.end());
                    satisfy_locked(c);
                }
            }
            rearm_stream(c.sock, pol.socket);
            read_len(i);
        }

        void fail_waiters_locked(const std::error_code & ec)
        {
            for (auto & cp : conns)
            {
                auto & w = cp->wait;
                if (!w.active)
                    continue;
                if (w.reading)
                {
                    w.abort = ec;
                    continue;
                }
                w.active = false;
                detail::post_handler(io, std::move(w.h), ec);
            }
        }

        void deliver_error(const std::error_code & ec)
        {
            {
                std::lock_guard<std::mutex> lock(mu);
                if (!fail)
                    fail = ec;
                fail_waiters_locked(fail);
            }
            shutdown_all(ec, false);
        }

        void close(const std::error_code & ec)
        {
            {
                std::lock_guard<std::mutex> lock(mu);
                if (!fail)
                    fail = ec;
                if (closed && aborted)
                    return;
                closed = true;
                aborted = true;
                fail_waiters_locked(ec);
            }
            shutdown_all(ec, true);
        }

        void close_graceful()
        {
            {
                std::lock_guard<std::mutex> lock(mu);
                if (closed)
                    return;
                closed = true;
                fail_waiters_locked(asio::error::operation_aborted);
            }
            for (std::size_t i = 0; i < conns.size(); ++i)
            {
                auto self = this->shared_from_this();
                asio::post(conns[i]->strand, [self, i] {
                    conn & c = *self->conns[i];
                    c.draining = true;
                    if (!c.writing)
                        finish_close(c);
                });
            }
        }

        static void finish_close(conn & c)
        {
            std::error_code e;
            base_socket(c.sock).shutdown(asio::socket_base::shutdown_both, e);
            base_socket(c.sock).close(e);
        }

        void shutdown_all(const std::error_code & ec, bool close_sockets)
        {
            for (std::size_t i = 0; i < conns.size(); ++i)
            {
                auto self = this->shared_from_this();
                asio::post(conns[i]->strand, [self, i, ec, close_sockets] {
                    conn & c = *self->conns[i];
                    std::error_code e;
                    base_socket(c.sock).shutdown(asio::socket_base::shutdown_both, e);
                    if (close_sockets)
                        base_socket(c.sock).close(e);
                    for (auto & ch : c.outq)
                        self->finish_chunk(c, ch, ec);
                    c.outq.clear();
                });
            }
        }

        stream_stats stats() const
        {
            stream_stats s;
            counters.fill(s);
            for (const auto & c : conns)
                s.buffered += c->buffered.load(std::memory_order_relaxed);
            std::lock_guard<std::mutex> lock(mu);
            for (const auto & c : conns)
            {
                s.unread += c->inbox.avail();
                if (!closed)
                    detail::add_socket_bytes(c->fd, s);
            }
            s.error = fail;
            s.closed = closed;
            return s;
        }

        // --- writes --------------------------------------------------------
        void write(std::size_t i, std::shared_ptr<std::vector<std::uint8_t>> payload,
            async_handler h)
        {
            check(i);
            const std::size_t len = payload ? payload->size() : 0;
            {
                std::lock_guard<std::mutex> lock(mu);
                if (fail || closed)
                {
                    detail::post_handler(io, std::move(h),
                        fail ? fail : asio::error::operation_aborted);
                    return;
                }
            }
            if (len == 0)
            {
                detail::post_handler(io, std::move(h), {});
                return;
            }
            if (pol.chunk_bytes == 0 && len > 0xffffffffu)
                throw std::invalid_argument(
                    "async_parallel_stream_array frame too large");
            const auto pieces = detail::split_chunks(len, pol.chunk_bytes);
            auto op = std::make_shared<detail::write_op>();
            op->h = std::move(h);
            op->left = pieces.size();
            std::vector<chunk> cs;
            cs.reserve(pieces.size());
            std::size_t wire = 0;
            for (const auto & pc : pieces)
            {
                chunk c;
                c.len_le = static_cast<std::uint32_t>(pc.second);
                c.payload = payload;
                c.off = pc.first;
                c.len = pc.second;
                c.op = op;
                wire += 4 + pc.second;
                cs.push_back(std::move(c));
            }
            conn & cn = *conns[i];
            cn.buffered.fetch_add(wire, std::memory_order_relaxed);
            auto self = this->shared_from_this();
            asio::post(cn.strand, [self, i, cs = std::move(cs)]() mutable {
                self->enqueue(i, std::move(cs));
            });
        }

        void enqueue(std::size_t i, std::vector<chunk> cs)
        {
            conn & c = *conns[i];
            std::error_code ec;
            {
                std::lock_guard<std::mutex> lock(mu);
                ec = fail;
            }
            if (ec)
            {
                for (auto & ch : cs)
                    finish_chunk(c, ch, ec);
                return;
            }
            for (auto & ch : cs)
                c.outq.push_back(std::move(ch));
            if (!c.writing)
                start_write(i);
        }

        void finish_chunk(conn & c, chunk & ch, const std::error_code & ec)
        {
            const std::size_t wire = 4 + ch.len;
            c.buffered.fetch_sub(
                std::min(c.buffered.load(std::memory_order_relaxed), wire),
                std::memory_order_relaxed);
            auto & op = *ch.op;
            if (ec && !op.ec)
                op.ec = ec;
            if (--op.left == 0)
                detail::post_handler(io, std::move(op.h), op.ec);
        }

        void start_write(std::size_t i)
        {
            conn & c = *conns[i];
            c.inflight.clear();
            std::size_t bytes = 0;
            while (!c.outq.empty() && c.inflight.size() < pol.coalesce_frames)
            {
                const std::size_t wire = 4 + c.outq.front().len;
                if (!c.inflight.empty() && bytes + wire > pol.coalesce_bytes)
                    break;
                c.inflight.push_back(std::move(c.outq.front()));
                c.outq.pop_front();
                bytes += wire;
            }
            if (c.inflight.empty())
            {
                c.writing = false;
                if (c.draining)
                    finish_close(c);
                return;
            }
            c.writing = true;
            c.inflight_wire = bytes;
            std::vector<asio::const_buffer> bufs;
            bufs.reserve(c.inflight.size() * 2);
            for (auto & ch : c.inflight)
            {
                bufs.push_back(asio::buffer(&ch.len_le, 4));
                bufs.push_back(asio::buffer(ch.payload->data() + ch.off, ch.len));
            }
            auto self = this->shared_from_this();
            if constexpr (detail::is_layered_stream<Socket>::value)
            {
                auto flat = detail::flatten(bufs, bytes);
                asio::async_write(c.sock, asio::buffer(*flat),
                    asio::bind_executor(c.strand,
                        [self, i, flat](const std::error_code & ec, std::size_t) {
                            self->on_written(i, ec);
                        }));
                return;
            }
            asio::async_write(c.sock, bufs,
                asio::bind_executor(c.strand,
                    [self, i](const std::error_code & ec, std::size_t) {
                        self->on_written(i, ec);
                    }));
        }

        void on_written(std::size_t i, const std::error_code & ec)
        {
            conn & c = *conns[i];
            if (!ec)
            {
                std::size_t payload = 0;
                for (const auto & ch : c.inflight)
                    payload += ch.len;
                counters.wrote(c.inflight_wire, payload, c.inflight.size());
            }
            for (auto & ch : c.inflight)
                finish_chunk(c, ch, ec);
            c.inflight.clear();
            if (ec)
            {
                c.writing = false;
                for (auto & ch : c.outq)
                    finish_chunk(c, ch, ec);
                c.outq.clear();
                deliver_error(detail::end_of_stream(ec));
                if (c.draining)
                    finish_close(c);
                return;
            }
            rearm_stream(c.sock, pol.socket);
            start_write(i);
        }

        asio::io_context & io;
        wire_policy pol;
        mutable std::mutex mu;
        std::vector<std::unique_ptr<conn>> conns;
        std::error_code fail;
        bool closed = false;
        bool aborted = false;
        detail::io_counters counters;
    };

    std::shared_ptr<impl> impl_;
};

/// @brief Parallel array over TCP sockets (one connection per lane).
using async_parallel_stream_array =
    basic_async_parallel_stream_array<asio::ip::tcp::socket>;

/// @brief Parallel array over unix-domain stream sockets.
using async_local_parallel_stream_array =
    basic_async_parallel_stream_array<asio::local::stream_protocol::socket>;

#if DPF_HAS_OPENSSL
#include "dpf/net/tls.hpp"
/// @brief N lanes on one TLS connection.
using async_tls_mux_stream_array = basic_async_mux_stream_array<tls_stream>;
/// @brief One TLS connection per lane.
using async_tls_parallel_stream_array = basic_async_parallel_stream_array<tls_stream>;
#endif

// ---------------------------------------------------------------------------
// Connection helpers (setup only; protocol I/O stays async)
// ---------------------------------------------------------------------------

/// @brief Accept `nstreams` lane sockets on ONE port.
/// @details Binds `base_port` (0 = ephemeral), publishes the port, and accepts
///          one connection per lane. Each connection announces its lane index
///          as a `u32`, so accept order does not matter.
HEDLEY_WARN_UNUSED_RESULT
inline std::vector<asio::ip::tcp::socket> accept_parallel_tcp(
    asio::io_context & io, std::atomic<unsigned short> & base_port,
    std::size_t nstreams,
    std::chrono::milliseconds budget = deadlines{}.accept)
{
    if (nstreams == 0)
        throw std::invalid_argument("accept_parallel_tcp empty");
    asio::ip::tcp::acceptor acc(io);
    open_listener(acc, base_port.load());
    base_port.store(acc.local_endpoint().port());
    std::vector<asio::ip::tcp::socket> socks;
    std::vector<bool> have(nstreams, false);
    for (std::size_t k = 0; k < nstreams; ++k)
        socks.emplace_back(io);
    for (std::size_t k = 0; k < nstreams; ++k)
    {
        asio::ip::tcp::socket s(io);
        accept_until(acc, s, budget);
        const std::uint32_t lane = exchange_u32(s.native_handle(),
            static_cast<std::uint32_t>(nstreams), budget, "parallel lane handshake");
        if (lane >= nstreams || have[lane])
            throw std::runtime_error("accept_parallel_tcp: bad lane "
                + std::to_string(lane));
        have[lane] = true;
        socks[lane] = std::move(s);
    }
    return socks;
}

/// @brief Connect `nstreams` lane sockets to one port (lane index handshake).
HEDLEY_WARN_UNUSED_RESULT
inline std::vector<asio::ip::tcp::socket> connect_parallel_tcp(
    asio::io_context & io, const std::string & host,
    std::atomic<unsigned short> & base_port, std::size_t nstreams,
    std::chrono::milliseconds budget = deadlines{}.connect)
{
    if (nstreams == 0)
        throw std::invalid_argument("connect_parallel_tcp empty");
    std::vector<asio::ip::tcp::socket> socks;
    socks.reserve(nstreams);
    for (std::size_t k = 0; k < nstreams; ++k)
    {
        asio::ip::tcp::socket s(io);
        connect_until(s, host, base_port.load(), budget);
        const std::uint32_t n = exchange_u32(s.native_handle(),
            static_cast<std::uint32_t>(k), budget, "parallel lane handshake");
        if (n != nstreams)
            throw std::runtime_error("connect_parallel_tcp: peer expects "
                + std::to_string(n) + " lanes, this side "
                + std::to_string(nstreams));
        socks.push_back(std::move(s));
    }
    return socks;
}

/// @brief In-process unix socket pairs: lane `k` connects `ends.first[k]`
///        (on `io_a`) to `ends.second[k]` (on `io_b`).
HEDLEY_WARN_UNUSED_RESULT
inline std::pair<std::vector<asio::local::stream_protocol::socket>,
    std::vector<asio::local::stream_protocol::socket>>
make_local_socket_pairs(asio::io_context & io_a, asio::io_context & io_b,
    std::size_t nstreams)
{
    std::pair<std::vector<asio::local::stream_protocol::socket>,
        std::vector<asio::local::stream_protocol::socket>>
        out;
    out.first.reserve(nstreams);
    out.second.reserve(nstreams);
    for (std::size_t k = 0; k < nstreams; ++k)
    {
        asio::local::stream_protocol::socket a(io_a);
        asio::local::stream_protocol::socket b(io_b);
        asio::local::connect_pair(a, b);
        out.first.push_back(std::move(a));
        out.second.push_back(std::move(b));
    }
    return out;
}

}  // namespace net
}  // namespace dpf

#endif  // LIBDPF_INCLUDE_DPF_NET_ASYNC_STREAM_ARRAY_HPP__