Commit 86b39cf8 authored by jan.koester's avatar jan.koester
Browse files

test

parent 49288cfc

src/client.cpp

+73 −57
Original line number Diff line number Diff line
@@ -141,7 +141,14 @@ void client::ensure_connected(size_t node_index) { 
std::vector<uint8_t> client::send_recv(size_t node_index,

                                         msg_type type,

                                         const std::vector<uint8_t>& payload) {

    for (int attempt = 0; attempt < 2; ++attempt) {

        try {

            ensure_connected(node_index);

        } catch (...) {

            if (attempt == 0) continue;

            return {};

        }

        if (!connections_[node_index].socket) return {};

        auto& conn = *connections_[node_index].socket;



        uint64_t stream = conn.openStream(true);
@@ -149,22 +156,32 @@ std::vector<uint8_t> client::send_recv(size_t node_index, 


        auto sr_deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5);



    // Non-blocking send: wait for writable, then send

        bool sent = false;

        while (int poll_ms = ms_left(sr_deadline)) {

            if (wait_.waitWrite(conn, poll_ms)) {

                conn.sendStreamData(stream, msg, true);

                conn.pumpNetwork(MSG_DONTWAIT);

                sent = true;

                break;

            }

            conn.pumpNetwork(MSG_DONTWAIT);

        }



    // Wait for response using event-driven waitRead (epoll/kqueue/IOCP/poll)

        if (!sent) {

            try { conn.closeStream(stream); } catch (...) {}

            connections_[node_index].socket.reset();

            connections_[node_index].connected = false;

            if (attempt == 0) continue;

            connections_[node_index].dead_until = std::chrono::steady_clock::now() + std::chrono::seconds(10);

            return {};

        }



        // Wait for response

        std::vector<uint8_t> response;

        while (int poll_ms = ms_left(sr_deadline)) {

            if (!wait_.waitRead(conn, poll_ms)) continue;

            conn.pumpNetwork(MSG_DONTWAIT);

            if (conn.hasStreamData(stream)) {

            // Read header

                uint8_t hdr[protocol::HEADER_SIZE];

                size_t r = conn.recvStreamData(stream, hdr, protocol::HEADER_SIZE);

                if (r < protocol::HEADER_SIZE) continue;
@@ -191,14 +208,18 @@ std::vector<uint8_t> client::send_recv(size_t node_index, 
            }

        }

        conn.closeStream(stream);

    // No response after all attempts — mark connection as broken and dead

    // so subsequent calls fail fast instead of retrying.

    if (response.empty()) {

        connections_[node_index].connected = false;



        if (!response.empty())

            return response;



        // Empty response — stale connection, retry once with fresh connection

        connections_[node_index].socket.reset();

        connections_[node_index].connected = false;

        if (attempt > 0) {

            connections_[node_index].dead_until = std::chrono::steady_clock::now() + std::chrono::seconds(10);

        }

    return response;

    }

    return {};

}



bool client::store_block_on_node(size_t node_index, uint64_t group_id,
@@ -768,8 +789,8 @@ void client::store(uint64_t group_id, const uint8_t* data, size_t data_len) { 
    DBG_LOG("[PARITY] store gid=" << group_id << " data_len=" << data_len

              << " stripes=" << total_stripes << "\n");



    // Retry once on failure: stale QUIC connections from a previous operation

    // cause the first stripe to fail; resetting connections fixes it.

    // store_stripe already handles per-node failures and retries internally.

    // Only retry at this level if the stripe itself throws (> m nodes failed).

    for (int attempt = 0; attempt < 2; ++attempt) {

        try {

            uint32_t stripe_idx = 0;
@@ -806,12 +827,7 @@ void client::store(uint64_t group_id, const uint8_t* data, size_t data_len) { 
        } catch (const std::exception& e) {

            if (attempt == 0) {

                DBG_LOG("[PARITY] store attempt 1 failed: " << e.what()

                          << ", resetting connections and retrying\n");

                // Reset all remote connections so ensure_connected() opens fresh ones

                for (size_t i = 0; i < connections_.size(); ++i) {

                    connections_[i].socket.reset();

                    connections_[i].connected = false;

                }

                          << ", retrying (dead nodes already cached)\n");

            } else {

                throw; // second attempt failed, propagate

            }