/// @file dpf/net/async_round_sink.hpp
/// @brief Event-driven `RoundSink` over an `async_stream_array`.
/// @details Both ends first exchange a hello on lane 0 carrying the plan shape
///          (rounds, slot widths, lanes, instances, framing) and an epoch. A
///          mismatch fails both ends with a message that names the field.
///          Unframed lanes carry one round each (round == lane). Framed lanes
///          carry `{round, nbytes}` headers; payloads are read straight into the
///          round's inbox, and a partial prefix of instances is ready as soon
///          as it lands. Every outbound round stays in memory, so after a
///          transport error the sink can take a replacement link from
///          `sink_options::reconnect`, exchange what each side received, and
///          resend only the missing bytes. The schedule above never rewinds.
#ifndef LIBDPF_INCLUDE_DPF_NET_ASYNC_ROUND_SINK_HPP__
#define LIBDPF_INCLUDE_DPF_NET_ASYNC_ROUND_SINK_HPP__

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

#include "dpf/net/asio_ns.hpp"

#include "dpf/net/async_stream_array.hpp"
#include "dpf/net/policy.hpp"
#include "dpf/net/round_lane.hpp"
#include "dpf/net/round_sink.hpp"

namespace dpf
{
namespace net
{

/// @brief Construction options for `async_round_sink`.
struct sink_options
{
    framing_mode framing = framing_mode::automatic;
    /// Exchange the plan-shape hello. Both ends must agree.
    bool hello = true;
    /// Bound on waiting for a full write window to drain.
    std::chrono::milliseconds drain_timeout{30000};
    /// Bound on the hello exchange after a reconnect.
    std::chrono::milliseconds handshake_timeout{30000};
    /// On a transport error, return a replacement link (the peer must replace
    /// its end too) or nullptr to fail. Runs on the drive thread.
    std::function<async_stream_array *(const std::error_code &)> reconnect;
    unsigned max_reconnects = 3;
};

/// @brief Per-sink counters.
struct sink_stats
{
    std::uint64_t bytes_sent = 0;
    std::uint64_t bytes_received = 0;
    std::uint64_t flushes = 0;
    std::uint64_t window_waits = 0;
    std::uint64_t window_wait_ns = 0;
    std::uint64_t resumes = 0;
    std::uint64_t resent_bytes = 0;
    bool framed = false;
    std::size_t lanes = 0;
};

/// @brief `RoundSink` whose I/O completes through an `io_context`, never spins.
class async_round_sink : public RoundSink
{
public:
    async_round_sink(async_stream_array & streams,
        std::vector<std::size_t> slot_bytes_per_round, std::size_t count = 1,
        sink_options opt = {})
      : core_(std::make_shared<core>(streams, streams,
            std::move(slot_bytes_per_round), count, std::move(opt)))
    {
        core_->start();
    }

    /// @brief Send on `out`, receive on `in` (a ring edge: to the previous
    ///        party, from the next). Both must run on the same `io_context`.
    ///        Reconnect resume needs a single link.
    async_round_sink(async_stream_array & out, async_stream_array & in,
        std::vector<std::size_t> slot_bytes_per_round, std::size_t count = 1,
        sink_options opt = {})
      : core_(std::make_shared<core>(out, in, std::move(slot_bytes_per_round),
            count, std::move(opt)))
    {
        core_->start();
    }

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

    ~async_round_sink() override
    {
        std::lock_guard<std::mutex> lock(core_->mu);
        core_->dead = true;
    }

    std::size_t count() const noexcept override { return core_->count; }
    std::size_t rounds() const noexcept override { return core_->slots.size(); }
    bool framed() const noexcept { return core_->map.framed; }
    std::size_t lanes() const noexcept { return core_->map.n_lanes; }

    std::size_t slot_bytes(std::uint16_t round) const override
    {
        if (round >= core_->slots.size())
            throw std::out_of_range("async_round_sink round");
        return core_->slots[round];
    }

    asio::io_context & context() noexcept { return *core_->io; }
    async_stream_array & link() noexcept { return *core_->streams; }
    async_stream_array & in_link() noexcept { return *core_->in; }

    sink_stats stats() const
    {
        std::lock_guard<std::mutex> lock(core_->mu);
        sink_stats s = core_->st;
        s.framed = core_->map.framed;
        s.lanes = core_->map.n_lanes;
        return s;
    }

    void submit(std::uint16_t round, std::size_t index,
        const std::uint8_t * bytes, std::size_t n) override
    {
        std::lock_guard<std::mutex> lock(core_->mu);
        core_->window(round).submit(index, bytes, n);
    }

    bool peer_ready(std::uint16_t round, std::size_t index) const override
    {
        {
            // Bytes that already arrived stay readable after a link error.
            std::lock_guard<std::mutex> lock(core_->mu);
            if (core_->ready_locked(round, index))
                return true;
        }
        core_->raise_if_failed();
        std::lock_guard<std::mutex> lock(core_->mu);
        return core_->ready_locked(round, index);
    }

    void read_peer(std::uint16_t round, std::size_t index, std::uint8_t * out,
        std::size_t n) const override
    {
        {
            std::lock_guard<std::mutex> lock(core_->mu);
            if (round < core_->slots.size() && core_->ready_locked(round, index))
            {
                if (n != core_->slots[round])
                    throw std::invalid_argument("async_round_sink read size");
                if (n != 0)
                    std::memcpy(out, core_->inbox[round].data() + index * n, n);
                return;
            }
        }
        core_->raise_if_failed();
        std::lock_guard<std::mutex> lock(core_->mu);
        if (round >= core_->slots.size())
            throw std::out_of_range("async_round_sink read_peer round");
        if (!core_->ready_locked(round, index))
            throw std::logic_error("async_round_sink: peer not ready");
        if (n != core_->slots[round])
            throw std::invalid_argument("async_round_sink read size");
        if (n != 0)
            std::memcpy(out, core_->inbox[round].data() + index * n, n);
    }

    void flush() override
    {
        for (std::uint16_t r = 0; r < core_->slots.size(); ++r)
        {
            bool due = false;
            {
                std::lock_guard<std::mutex> lock(core_->mu);
                const auto & w = core_->win[r];
                due = w.next_unwritten() > w.flushed()
                    || (core_->slots[r] == 0 && !core_->announced[r]);
            }
            if (due)
                flush_round(r);
        }
    }

    void flush_round(std::uint16_t round) override
    {
        core_->raise_if_failed();
        const std::size_t lane = core_->flush_one(round);
        core_->await_window(lane);
    }

    void poll() override
    {
        core_->restart_if_stopped();
        core_->io->poll();
    }

    bool wait_io() override
    {
        core_->raise_if_failed();
        core_->restart_if_stopped();
        const bool ran = core_->io->run_one() > 0;
        core_->raise_if_failed();
        return ran;
    }

    bool wait_io_for(std::chrono::milliseconds budget) override
    {
        core_->raise_if_failed();
        core_->restart_if_stopped();
        const bool ran = budget.count() <= 0 ? core_->io->poll_one() > 0
                                              : core_->io->run_one_for(budget) > 0;
        core_->raise_if_failed();
        return ran;
    }

    bool can_block() const noexcept override { return true; }

    bool can_send_ahead() const noexcept override
    {
        for (std::size_t l = 0; l < core_->map.n_lanes; ++l)
        {
            const std::size_t w = core_->streams->lane_window_bytes(l);
            if (w != 0 && core_->streams->lane_buffered_bytes(l) > w)
                return false;
        }
        return true;
    }

    std::uint64_t progress() const noexcept override
    {
        return core_->progress.load(std::memory_order_relaxed);
    }

    const void * wait_domain() const noexcept override { return core_->io; }

private:
    static constexpr std::uint32_t k_magic = 0x48535044u;  // 'DPSH'
    static constexpr std::uint16_t k_version = 1;
    static constexpr std::size_t k_hello_fixed = 36;
    static constexpr std::uint32_t k_zero_seen = 1;

    struct core : std::enable_shared_from_this<core>
    {
        core(async_stream_array & s, async_stream_array & rx,
            std::vector<std::size_t> slots_, std::size_t count_, sink_options opt_)
          : streams(&s),
            in(&rx),
            io(&s.context()),
            count(count_),
            slots(std::move(slots_)),
            map(std::min(s.size(), rx.size()), slots.size(), opt_.framing),
            opt(std::move(opt_))
        {
            if (count == 0)
                throw std::invalid_argument("async_round_sink: count 0");
            if (s.size() == 0 || rx.size() == 0)
                throw std::invalid_argument("async_round_sink: empty streams");
            if (&s.context() != &rx.context())
                throw std::invalid_argument(
                    "async_round_sink: out and in links need one io_context");
            if (&s != &rx && opt.reconnect)
                throw std::invalid_argument(
                    "async_round_sink: reconnect needs a single link");
            const std::size_t nr = slots.size();
            win.reserve(nr);
            inbox.resize(nr);
            in_filled.assign(nr, 0);
            zero_seen.assign(nr, false);
            announced.assign(nr, false);
            legacy_started.assign(nr, false);
            legacy_wanted.assign(nr, false);
            for (std::size_t k = 0; k < nr; ++k)
            {
                win.emplace_back(count, slots[k]);
                inbox[k].assign(count * slots[k], 0);
            }
            fingerprint = slots_fingerprint(slots);
        }

        // --- lifecycle -------------------------------------------------
        void start()
        {
            std::lock_guard<std::mutex> lock(mu);
            if (opt.hello)
                begin_hello_locked();
            else
                hello_done_locked();
        }

        void restart_if_stopped()
        {
            if (io->stopped())
                io->restart();
        }

        // --- hello ------------------------------------------------------
        std::shared_ptr<std::vector<std::uint8_t>> build_hello_locked() const
        {
            const std::size_t r = slots.size();
            auto buf = acquire_buffer(k_hello_fixed + 4 * r);
            auto * p = buf->data();
            std::memset(p, 0, buf->size());
            detail::put_u32(p + 0, k_magic);
            p[4] = static_cast<std::uint8_t>(k_version & 0xffu);
            p[5] = static_cast<std::uint8_t>(k_version >> 8);
            p[6] = map.framed ? 1 : 0;
            detail::put_u32(p + 8, static_cast<std::uint32_t>(r));
            detail::put_u32(p + 12, static_cast<std::uint32_t>(map.n_lanes));
            detail::put_u32(p + 16, static_cast<std::uint32_t>(count));
            detail::put_u32(p + 20, gen);
            for (int i = 0; i < 8; ++i)
                p[24 + i] = static_cast<std::uint8_t>((fingerprint >> (8 * i)) & 0xffu);
            for (std::size_t k = 0; k < r; ++k)
            {
                const std::uint32_t got = slots[k] == 0
                    ? (zero_seen[k] ? k_zero_seen : 0)
                    : static_cast<std::uint32_t>(in_filled[k]);
                detail::put_u32(p + k_hello_fixed + 4 * k, got);
            }
            return buf;
        }

        void begin_hello_locked()
        {
            hello_ok = false;
            auto self = shared_from_this();
            const std::uint32_t g = gen;
            streams->async_write_owned(0, build_hello_locked(),
                [self, g](const std::error_code & ec) { self->on_write(g, ec); });
            hello_in = std::make_shared<std::vector<std::uint8_t>>(
                k_hello_fixed + 4 * slots.size());
            auto buf = hello_in;
            in->async_read(0, buf->data(), buf->size(),
                [self, g, buf](const std::error_code & ec) {
                    self->on_hello(g, buf, ec);
                });
        }

        void on_hello(std::uint32_t g,
            const std::shared_ptr<std::vector<std::uint8_t>> & buf,
            const std::error_code & ec)
        {
            std::lock_guard<std::mutex> lock(mu);
            if (dead || g != gen)
                return;
            if (ec)
            {
                fail_locked(ec, "hello read", false);
                return;
            }
            const auto * p = buf->data();
            if (detail::get_u32(p) != k_magic)
            {
                fail_locked(std::make_error_code(std::errc::protocol_error),
                    "peer did not send a sink hello (is it an async_round_sink "
                    "or stream_array_sink with hello enabled?)", true);
                return;
            }
            const std::uint16_t ver = static_cast<std::uint16_t>(p[4] | (p[5] << 8));
            const bool peer_framed = (p[6] & 1) != 0;
            const std::uint32_t pr = detail::get_u32(p + 8);
            const std::uint32_t pl = detail::get_u32(p + 12);
            const std::uint32_t pc = detail::get_u32(p + 16);
            const std::uint32_t pg = detail::get_u32(p + 20);
            std::uint64_t pf = 0;
            for (int i = 0; i < 8; ++i)
                pf |= static_cast<std::uint64_t>(p[24 + i]) << (8 * i);
            std::string why;
            if (ver != k_version)
                why += " version " + std::to_string(ver) + " vs "
                    + std::to_string(k_version) + ";";
            if (pr != slots.size())
                why += " rounds " + std::to_string(pr) + " vs "
                    + std::to_string(slots.size()) + ";";
            else if (pf != fingerprint)
                why += " slot widths differ;";
            if (pl != map.n_lanes)
                why += " lanes " + std::to_string(pl) + " vs "
                    + std::to_string(map.n_lanes) + ";";
            if (pc != count)
                why += " instances " + std::to_string(pc) + " vs "
                    + std::to_string(count) + ";";
            if (peer_framed != map.framed)
                why += std::string(" framing ") + (peer_framed ? "on" : "off")
                    + " vs " + (map.framed ? "on" : "off") + ";";
            if (pg != gen)
                why += " epoch " + std::to_string(pg) + " vs "
                    + std::to_string(gen) + ";";
            if (!why.empty())
            {
                fail_locked(std::make_error_code(std::errc::protocol_error),
                    "peer sink disagrees (peer vs this side):" + why, true);
                return;
            }
            std::vector<std::uint32_t> peer_got(slots.size());
            for (std::size_t k = 0; k < slots.size(); ++k)
                peer_got[k] = detail::get_u32(p + k_hello_fixed + 4 * k);
            if (gen != 0 && !resend_locked(peer_got))
                return;
            hello_done_locked();
        }

        void hello_done_locked()
        {
            hello_ok = true;
            if (map.framed)
            {
                hdr_bufs.clear();
                for (std::size_t lane = 0; lane < map.n_lanes; ++lane)
                {
                    hdr_bufs.push_back(
                        std::make_shared<std::array<std::uint8_t, round_lane_hdr::size>>());
                    read_lane_hdr_locked(lane);
                }
            }
            else
            {
                for (std::size_t r = 0; r < slots.size(); ++r)
                    if (legacy_wanted[r])
                        issue_legacy_locked(static_cast<std::uint16_t>(r));
            }
        }

        /// @brief After a reconnect: send what the peer reported missing.
        bool resend_locked(const std::vector<std::uint32_t> & peer_got)
        {
            auto self = shared_from_this();
            const std::uint32_t g = gen;
            for (std::size_t k = 0; k < slots.size(); ++k)
            {
                const std::size_t sb = slots[k];
                const auto round = static_cast<std::uint16_t>(k);
                const std::size_t lane = map.lane(round);
                if (sb == 0)
                {
                    if (map.framed && announced[k] && peer_got[k] != k_zero_seen)
                    {
                        auto frame = acquire_buffer(round_lane_hdr::size);
                        round_lane_hdr{round, 0}.pack(frame->data());
                        streams->async_write_owned(lane, std::move(frame),
                            [self, g](const std::error_code & ec) {
                                self->on_write(g, ec);
                            });
                    }
                    continue;
                }
                const std::size_t sent = win[k].flushed() * sb;
                const std::size_t got = peer_got[k];
                if (got > sent || got % sb != 0)
                {
                    fail_locked(std::make_error_code(std::errc::protocol_error),
                        "resume: peer reports " + std::to_string(got)
                            + " bytes of round " + std::to_string(k)
                            + ", this side sent " + std::to_string(sent),
                        true);
                    return false;
                }
                if (got == sent)
                    continue;
                const std::size_t nbytes = sent - got;
                const std::uint8_t * src = win[k].out_at(got / sb);
                st.resent_bytes += nbytes;
                if (map.framed)
                {
                    auto frame = acquire_buffer(round_lane_hdr::size + nbytes);
                    round_lane_hdr{round, static_cast<std::uint32_t>(nbytes)}.pack(
                        frame->data());
                    std::memcpy(frame->data() + round_lane_hdr::size, src, nbytes);
                    streams->async_write_owned(lane, std::move(frame),
                        [self, g](const std::error_code & ec) {
                            self->on_write(g, ec);
                        });
                }
                else
                {
                    streams->async_write(lane, src, nbytes,
                        [self, g](const std::error_code & ec) {
                            self->on_write(g, ec);
                        });
                }
            }
            return true;
        }

        // --- errors and resume ------------------------------------------
        void fail_locked(const std::error_code & ec, const std::string & what,
            bool is_fatal)
        {
            if (fail)
                return;
            fail = ec;
            fail_what = "async_round_sink: " + what;
            fatal = is_fatal;
        }

        void on_write(std::uint32_t g, const std::error_code & ec)
        {
            if (!ec)
                return;
            std::lock_guard<std::mutex> lock(mu);
            if (dead || g != gen)
                return;
            fail_locked(ec, "write", false);
        }

        void raise_if_failed()
        {
            std::error_code ec;
            std::string what;
            {
                std::lock_guard<std::mutex> lock(mu);
                if (!fail)
                    return;
                if (fatal || !opt.reconnect || reconnects >= opt.max_reconnects)
                    throw std::system_error(fail, fail_what);
                ec = fail;
                what = fail_what;
            }
            async_stream_array * next = nullptr;
            try
            {
                next = opt.reconnect(ec);
            }
            catch (const std::exception & e)
            {
                throw std::system_error(ec, what + "; reconnect failed: " + e.what());
            }
            if (next == nullptr)
                throw std::system_error(ec, what + "; no replacement link");
            resume(*next);
        }

        void resume(async_stream_array & next)
        {
            {
                std::lock_guard<std::mutex> lock(mu);
                if (next.size() < map.n_lanes)
                    throw std::invalid_argument(
                        "async_round_sink resume: replacement link has fewer lanes");
                ++gen;
                ++reconnects;
                ++st.resumes;
                fail.clear();
                fail_what.clear();
                fatal = false;
                streams = &next;
                in = &next;
                io = &next.context();
                for (std::size_t r = 0; r < slots.size(); ++r)
                {
                    if (legacy_started[r] && in_filled[r] < count * slots[r])
                    {
                        legacy_started[r] = false;
                        legacy_wanted[r] = true;
                    }
                }
                begin_hello_locked();
            }
            const auto deadline = std::chrono::steady_clock::now()
                + opt.handshake_timeout;
            for (;;)
            {
                {
                    std::lock_guard<std::mutex> lock(mu);
                    if (hello_ok)
                        return;
                    if (fail)
                        throw std::system_error(fail, fail_what + " (during resume)");
                }
                const auto now = std::chrono::steady_clock::now();
                if (now >= deadline)
                    throw std::system_error(std::make_error_code(std::errc::timed_out),
                        "async_round_sink: resume hello timed out");
                restart_if_stopped();
                io->run_one_for(std::min<std::chrono::milliseconds>(
                    std::chrono::duration_cast<std::chrono::milliseconds>(
                        deadline - now),
                    std::chrono::milliseconds(10)));
            }
        }

        // --- reads ------------------------------------------------------
        /// @brief Every round's peer bytes (and zero-width announcements) are in.
        bool complete_locked() const
        {
            for (std::size_t r = 0; r < slots.size(); ++r)
            {
                const std::size_t sb = slots[r];
                if (sb == 0 ? (map.framed && !zero_seen[r])
                            : in_filled[r] < count * sb)
                    return false;
            }
            return true;
        }

        bool ready_locked(std::uint16_t round, std::size_t index)
        {
            if (round >= slots.size())
                return false;
            const std::size_t sb = slots[round];
            if (map.framed)
            {
                if (sb == 0)
                    return zero_seen[round];
                return in_filled[round] >= (index + 1) * sb;
            }
            if (sb == 0)
                return true;
            if (!legacy_started[round])
            {
                if (hello_ok)
                    issue_legacy_locked(round);
                else
                    legacy_wanted[round] = true;
            }
            return in_filled[round] == count * sb;
        }

        void issue_legacy_locked(std::uint16_t round)
        {
            if (legacy_started[round])
                return;
            legacy_started[round] = true;
            legacy_wanted[round] = false;
            const std::size_t bytes = count * slots[round];
            if (bytes == 0)
            {
                in_filled[round] = 0;
                return;
            }
            auto self = shared_from_this();
            const std::uint32_t g = gen;
            this->in->async_read(round, inbox[round].data(), bytes,
                [self, g, round, bytes](const std::error_code & ec) {
                    std::lock_guard<std::mutex> lock(self->mu);
                    if (self->dead || g != self->gen)
                        return;
                    if (ec)
                    {
                        self->fail_locked(ec, "read round "
                            + std::to_string(round), false);
                        return;
                    }
                    self->in_filled[round] = bytes;
                    self->st.bytes_received += bytes;
                    self->progress.fetch_add(bytes, std::memory_order_relaxed);
                });
        }

        void read_lane_hdr_locked(std::size_t lane)
        {
            auto self = shared_from_this();
            const std::uint32_t g = gen;
            auto buf = hdr_bufs[lane];
            in->async_read(lane, buf->data(), buf->size(),
                [self, g, lane, buf](const std::error_code & ec) {
                    self->on_lane_hdr(g, lane, buf, ec);
                });
        }

        void on_lane_hdr(std::uint32_t g, std::size_t lane,
            const std::shared_ptr<std::array<std::uint8_t, round_lane_hdr::size>> & buf,
            const std::error_code & ec)
        {
            std::lock_guard<std::mutex> lock(mu);
            if (dead || g != gen)
                return;
            if (ec)
            {
                // A peer that closes after its last frame finished cleanly.
                if (!complete_locked())
                    fail_locked(ec, "read lane " + std::to_string(lane), false);
                return;
            }
            const auto hdr = round_lane_hdr::unpack(buf->data());
            if (hdr.round >= slots.size() || map.lane(hdr.round) != lane)
            {
                fail_locked(std::make_error_code(std::errc::protocol_error),
                    "frame for round " + std::to_string(hdr.round) + " on lane "
                        + std::to_string(lane),
                    true);
                return;
            }
            const std::size_t sb = slots[hdr.round];
            const std::size_t cap = count * sb;
            if (hdr.nbytes == 0)
            {
                if (sb == 0 && !zero_seen[hdr.round])
                {
                    zero_seen[hdr.round] = true;
                    progress.fetch_add(1, std::memory_order_relaxed);
                }
                read_lane_hdr_locked(lane);
                return;
            }
            if (sb == 0 || hdr.nbytes % sb != 0
                || in_filled[hdr.round] + hdr.nbytes > cap)
            {
                fail_locked(std::make_error_code(std::errc::message_size),
                    "round " + std::to_string(hdr.round) + " frame of "
                        + std::to_string(hdr.nbytes) + " bytes does not fit",
                    true);
                return;
            }
            const std::uint16_t round = hdr.round;
            const std::size_t nbytes = hdr.nbytes;
            auto self = shared_from_this();
            in->async_read(lane, inbox[round].data() + in_filled[round], nbytes,
                [self, g, lane, round, nbytes](const std::error_code & xec) {
                    std::lock_guard<std::mutex> lk(self->mu);
                    if (self->dead || g != self->gen)
                        return;
                    if (xec)
                    {
                        self->fail_locked(xec, "read round "
                            + std::to_string(round), false);
                        return;
                    }
                    self->in_filled[round] += nbytes;
                    self->st.bytes_received += nbytes;
                    self->progress.fetch_add(nbytes, std::memory_order_relaxed);
                    self->read_lane_hdr_locked(lane);
                });
        }

        // --- writes -----------------------------------------------------
        round_window & window(std::uint16_t round)
        {
            if (round >= win.size())
                throw std::out_of_range("async_round_sink round");
            return win[round];
        }

        std::size_t flush_one(std::uint16_t round)
        {
            std::lock_guard<std::mutex> lock(mu);
            auto & w = window(round);
            std::size_t begin = 0;
            std::size_t nslots = 0;
            const std::uint8_t * pend = w.pending_out(begin, nslots);
            const std::size_t sb = slots[round];
            const std::size_t nbytes = nslots * sb;
            const std::size_t lane = map.lane(round);
            auto self = shared_from_this();
            const std::uint32_t g = gen;
            auto done = [self, g](const std::error_code & ec) {
                self->on_write(g, ec);
            };
            if (map.framed)
            {
                if (nslots != 0 || (sb == 0 && !announced[round]))
                {
                    auto frame = acquire_buffer(round_lane_hdr::size + nbytes);
                    round_lane_hdr{round, static_cast<std::uint32_t>(nbytes)}.pack(
                        frame->data());
                    if (nbytes != 0)
                        std::memcpy(frame->data() + round_lane_hdr::size, pend,
                            nbytes);
                    streams->async_write_owned(lane, std::move(frame), done);
                    if (sb == 0)
                        announced[round] = true;
                    ++st.flushes;
                    st.bytes_sent += nbytes;
                }
                if (nslots != 0)
                    w.mark_flushed(nslots);
            }
            else
            {
                if (nslots != 0 && sb != 0)
                {
                    streams->async_write(lane, pend, nbytes, done);
                    ++st.flushes;
                    st.bytes_sent += nbytes;
                }
                if (nslots != 0)
                    w.mark_flushed(nslots);
                if (!legacy_started[round])
                {
                    if (hello_ok)
                        issue_legacy_locked(round);
                    else
                        legacy_wanted[round] = true;
                }
            }
            return lane;
        }

        /// @brief Stall the producer while lane `lane`'s window is full.
        void await_window(std::size_t lane)
        {
            const std::size_t w = streams->lane_window_bytes(lane);
            if (w == 0 || streams->lane_buffered_bytes(lane) <= w)
                return;
            const auto t0 = std::chrono::steady_clock::now();
            auto since = t0;
            std::size_t last = streams->lane_buffered_bytes(lane);
            for (;;)
            {
                raise_if_failed();
                const std::size_t now_buf = streams->lane_buffered_bytes(lane);
                if (now_buf <= streams->lane_window_bytes(lane))
                    break;
                const auto now = std::chrono::steady_clock::now();
                if (now_buf < last)
                {
                    last = now_buf;
                    since = now;
                }
                else if (now - since > opt.drain_timeout)
                {
                    throw std::runtime_error("async_round_sink: lane "
                        + std::to_string(lane) + " did not drain for "
                        + std::to_string(opt.drain_timeout.count()) + " ms ("
                        + std::to_string(now_buf) + " bytes buffered, window "
                        + std::to_string(w) + ")");
                }
                restart_if_stopped();
                io->run_one_for(std::chrono::milliseconds(10));
            }
            const auto dt = std::chrono::steady_clock::now() - t0;
            std::lock_guard<std::mutex> lock(mu);
            ++st.window_waits;
            st.window_wait_ns += static_cast<std::uint64_t>(
                std::chrono::duration_cast<std::chrono::nanoseconds>(dt).count());
        }

        async_stream_array * streams = nullptr;
        async_stream_array * in = nullptr;
        asio::io_context * io = nullptr;
        std::size_t count = 1;
        std::vector<std::size_t> slots;
        round_lane_map map;
        sink_options opt;
        std::uint64_t fingerprint = 0;

        std::mutex mu;
        std::vector<round_window> win;
        std::vector<std::vector<std::uint8_t>> inbox;
        std::vector<std::size_t> in_filled;
        std::vector<bool> zero_seen;
        std::vector<bool> announced;
        std::vector<bool> legacy_started;
        std::vector<bool> legacy_wanted;
        std::vector<std::shared_ptr<std::array<std::uint8_t, round_lane_hdr::size>>>
            hdr_bufs;
        std::shared_ptr<std::vector<std::uint8_t>> hello_in;
        std::uint32_t gen = 0;
        bool hello_ok = false;
        std::error_code fail;
        std::string fail_what;
        bool fatal = false;
        unsigned reconnects = 0;
        bool dead = false;
        std::atomic<std::uint64_t> progress{0};
        sink_stats st;
    };

    std::shared_ptr<core> core_;
};

}  // namespace net
}  // namespace dpf

#endif  // LIBDPF_INCLUDE_DPF_NET_ASYNC_ROUND_SINK_HPP__