Commit 5e75a1d0 authored by Ramon Nou's avatar Ramon Nou
Browse files

complete RPC disconnect containment coverage

parent 02a62065

WIP.md

+17 −13
Original line number Diff line number Diff line
@@ -37,8 +37,11 @@ documentation, or experimental work. 


### 1. Prevent daemon termination from one failed client RPC



**Status: Partial.** Core response containment and basic rate limiting are

implemented; peer/request-ID metadata and full disconnect coverage remain.

**Status: Done for supported RPC response paths.** Daemon and proxy response

submission is contained and rate-limited. Transport diagnostics include the RPC

handler name, peer endpoint, and cause; the installed Thallium API does not

expose a request ID. Disconnect regressions cover data, metadata, and

malleability RPC clients.



**Problem:** A client disappearing during an RPC can cause an exception while

responding and terminate the daemon. One failed client or resize must not take
@@ -46,19 +49,20 @@ down unrelated clients or the whole filesystem. 


**Work:**



- Establish one exception boundary around decoding, handler execution, and

  response submission.

- Convert transport failures into structured RPC errors containing peer,

  request type, request ID, and cause.

- Audit all handlers for `MARGO_ASSERT`, `req.respond()`, and equivalent calls

  that can throw outside cleanup logic.

- Define which failures are recoverable, retryable, or daemon-fatal.

- Rate-limit repeated transport diagnostics.

- Keep all response submissions behind `safe_respond()`; direct

  `req.respond()` is forbidden in daemon and proxy handlers.

- Treat disconnected peers and response-serialization failures as recoverable:

  log the failed response and continue serving. The request ID is recorded as

  `unavailable` because this Thallium request API does not provide one.

- Treat handler exceptions as request failures (`errno` responses) and process

  initialization, storage, or corruption failures as daemon-fatal according to

  their existing startup/backend error paths.

- Maintain rate-limited transport diagnostics and coverage for data, metadata,

  and malleability clients.



**Acceptance:** Disconnect clients during data, metadata, and malleability RPCs;

the daemon stays alive, later operations work, and the client receives a

deterministic error or retry result. Extend

`tests/integration/malleability/test_client_disconnect_during_rpc.py`.

the daemon stays alive and later operations work. The disconnected client cannot

receive a response; surviving clients receive normal deterministic results.



### 2. Make hostfile updates transactional and observable

include/common/rpc/handler_util.hpp

+27 −6
Original line number Diff line number Diff line
@@ -3,9 +3,24 @@ 
#include <spdlog/spdlog.h>

#include <atomic>

#include <cstdint>

#include <string>



namespace gkfs::utils {



namespace detail {



template <typename RequestType>

std::string

request_peer(const RequestType& req) noexcept {

    try {

        return static_cast<std::string>(req.get_endpoint());

    } catch(...) {

        return "unavailable";

    }

}



} // namespace detail



/**

 * @internal

 * Safe wrapper around thallium::request::respond() that contains
@@ -26,16 +41,18 @@ safe_respond(RequestType& req, const ResponseType& resp, 
        const auto failure = transport_failures.fetch_add(1) + 1;

        if(failure == 1 || failure % 100 == 0) {

            try {

                const auto peer = detail::request_peer(req);

                auto logger = spdlog::get("main");

                if(!logger) {

                    logger = spdlog::default_logger();

                }

                if(logger) {

                    logger->warn(

                            "rpc_response_failed=1 rpc_context={} "

                            "rpc_response_failed=1 rpc_name={} peer='{}' "

                            "request_id=unavailable "

                            "failure_kind=client_disconnect failure_count={} "

                            "cause='{}'",

                            context, failure, e.what());

                            context, peer, failure, e.what());

                }

            } catch(...) {

            }
@@ -43,28 +60,32 @@ safe_respond(RequestType& req, const ResponseType& resp, 
        return false;

    } catch(const std::exception& e) {

        try {

            const auto peer = detail::request_peer(req);

            auto logger = spdlog::get("main");

            if(!logger) {

                logger = spdlog::default_logger();

            }

            if(logger) {

                logger->error("rpc_response_failed=1 rpc_context={} "

                logger->error("rpc_response_failed=1 rpc_name={} peer='{}' "

                              "request_id=unavailable "

                              "failure_kind=unexpected cause='{}'",

                              context, e.what());

                              context, peer, e.what());

            }

        } catch(...) {

        }

        return false;

    } catch(...) {

        try {

            const auto peer = detail::request_peer(req);

            auto logger = spdlog::get("main");

            if(!logger) {

                logger = spdlog::default_logger();

            }

            if(logger) {

                logger->error("rpc_response_failed=1 rpc_context={} "

                logger->error("rpc_response_failed=1 rpc_name={} peer='{}' "

                              "request_id=unavailable "

                              "failure_kind=unknown",

                              context);

                              context, peer);

            }

        } catch(...) {

        }

src/proxy/rpc/srv_data.cpp

+7 −7
Original line number Diff line number Diff line
@@ -46,7 +46,7 @@ proxy_rpc_srv_write(const tl::request& req, 
            PROXY_DATA->log()->error("{}() Bulk size '{}' != write_size '{}'",

                                     __func__, bulk_size, in.write_size);

            out.err = EINVAL;

            gkfs::utils::safe_respond(req, out);

            gkfs::utils::safe_respond(req, out, __func__);

            return;

        }
@@ -84,7 +84,7 @@ proxy_rpc_srv_write(const tl::request& req, 
        out.err = EBUSY;

    }



    gkfs::utils::safe_respond(req, out);

    gkfs::utils::safe_respond(req, out, __func__);

}



void
@@ -100,7 +100,7 @@ proxy_rpc_srv_read(const tl::request& req, 
            PROXY_DATA->log()->error("{}() Bulk size '{}' != read_size '{}'",

                                     __func__, bulk_size, in.read_size);

            out.err = EINVAL;

            gkfs::utils::safe_respond(req, out);

            gkfs::utils::safe_respond(req, out, __func__);

            return;

        }
@@ -118,7 +118,7 @@ proxy_rpc_srv_read(const tl::request& req, 
                    "{}() Failure when forwarding to daemon with err '{}'",

                    __func__, daemon_out.first);

            out.err = daemon_out.first;

            gkfs::utils::safe_respond(req, out);

            gkfs::utils::safe_respond(req, out, __func__);

            return;

        }
@@ -148,7 +148,7 @@ proxy_rpc_srv_read(const tl::request& req, 
        out.err = EBUSY;

    }



    gkfs::utils::safe_respond(req, out);

    gkfs::utils::safe_respond(req, out, __func__);

}



void
@@ -173,7 +173,7 @@ proxy_rpc_srv_truncate(const tl::request& req, 
    }



    PROXY_DATA->log()->debug("{}() Sending output err '{}'", __func__, out.err);

    gkfs::utils::safe_respond(req, out);

    gkfs::utils::safe_respond(req, out, __func__);

}



void
@@ -197,5 +197,5 @@ proxy_rpc_srv_chunk_stat(const tl::request& req, 
        out.err = EBUSY;

    }

    PROXY_DATA->log()->debug("{}() Sending output err '{}'", __func__, out.err);

    gkfs::utils::safe_respond(req, out);

    gkfs::utils::safe_respond(req, out, __func__);

}
 
 No newline at end of file

src/proxy/rpc/srv_metadata.cpp

+9 −9
Original line number Diff line number Diff line
@@ -49,7 +49,7 @@ proxy_rpc_srv_create(const tl::request& req, gkfs::rpc::rpc_mk_node_in_t& in) { 
    }



    PROXY_DATA->log()->debug("{}() Sending output err '{}'", __func__, out.err);

    gkfs::utils::safe_respond(req, out);

    gkfs::utils::safe_respond(req, out, __func__);

}



void
@@ -73,7 +73,7 @@ proxy_rpc_srv_stat(const tl::request& req, gkfs::rpc::rpc_path_only_in_t& in) { 
    // out.inline_data is implicit?



    PROXY_DATA->log()->debug("{}() Sending output err '{}'", __func__, out.err);

    gkfs::utils::safe_respond(req, out);

    gkfs::utils::safe_respond(req, out, __func__);

}



void
@@ -94,7 +94,7 @@ proxy_rpc_srv_remove(const tl::request& req, gkfs::rpc::rpc_rm_node_in_t& in) { 
    }



    PROXY_DATA->log()->debug("{}() Sending output err '{}'", __func__, out.err);

    gkfs::utils::safe_respond(req, out);

    gkfs::utils::safe_respond(req, out, __func__);

}



void
@@ -117,7 +117,7 @@ proxy_rpc_srv_decr_size(const tl::request& req, gkfs::rpc::rpc_trunc_in_t& in) { 
    }



    PROXY_DATA->log()->debug("{}() Sending output err '{}'", __func__, out.err);

    gkfs::utils::safe_respond(req, out);

    gkfs::utils::safe_respond(req, out, __func__);

}



void
@@ -139,7 +139,7 @@ proxy_rpc_srv_get_metadentry_size(const tl::request& req, 


    PROXY_DATA->log()->debug("{}() Sending output err '{}' ret_size '{}'",

                             __func__, out.err, out.ret_size);

    gkfs::utils::safe_respond(req, out);

    gkfs::utils::safe_respond(req, out, __func__);

}



void
@@ -167,7 +167,7 @@ proxy_rpc_srv_update_metadentry_size( 


    PROXY_DATA->log()->debug("{}() Sending output err '{}' ret_offset '{}'",

                             __func__, out.err, out.ret_offset);

    gkfs::utils::safe_respond(req, out);

    gkfs::utils::safe_respond(req, out, __func__);

}



void
@@ -193,7 +193,7 @@ proxy_rpc_srv_get_dirents_extended( 
                    __func__, strerror(daemon_err));

            out.err = daemon_err;

            out.dirents_size = 0;

            gkfs::utils::safe_respond(req, out);

            gkfs::utils::safe_respond(req, out, __func__);

            return;

        }
@@ -209,7 +209,7 @@ proxy_rpc_srv_get_dirents_extended( 


            out.err = ENOBUFS;

            out.dirents_size = payload_size;

            gkfs::utils::safe_respond(req, out);

            gkfs::utils::safe_respond(req, out, __func__);

            return;

        }
@@ -243,5 +243,5 @@ proxy_rpc_srv_get_dirents_extended( 


    PROXY_DATA->log()->debug("{}() Sending output err '{}' dirents_size '{}'",

                             __func__, out.err, out.dirents_size);

    gkfs::utils::safe_respond(req, out);

    gkfs::utils::safe_respond(req, out, __func__);

}

tests/integration/malleability/test_client_disconnect_during_rpc.py

+53 −15
Original line number Diff line number Diff line
@@ -14,27 +14,16 @@ import subprocess 
import time





def test_daemon_survives_client_abort_during_write(gkfwd_daemon_factory,

                                                   gkfs_shell):

    """A killed intercepted writer must not terminate the forwarding daemon."""

    daemon = gkfwd_daemon_factory.create()

    aborted_path = daemon.mountdir / "aborted_client_write"

    health_path = daemon.mountdir / "post_abort_health"



def _kill_client_and_check_health(daemon, gkfs_shell, command, health_name):

    client = subprocess.Popen(

            [

                    "bash",

                    "-c",

                    f"exec dd if=/dev/zero of='{aborted_path}' "

                    "bs=4M count=1024 status=none conv=fsync",

            ],

            ["bash", "-c", command],

            env=gkfs_shell._env,

            stdout=subprocess.PIPE,

            stderr=subprocess.PIPE,

    )

    try:

        # Give the intercepted client time to enter a data RPC, but do not wait

        # for the 4 GiB write to finish.

        # Give the client time to enter its RPC, but do not wait for it to

        # complete.

        time.sleep(0.25)

        if client.poll() is None:

            client.send_signal(signal.SIGKILL)
@@ -43,6 +32,7 @@ def test_daemon_survives_client_abort_during_write(gkfwd_daemon_factory, 


        assert daemon._proc.poll() is None, "daemon exited after client abort"



        health_path = daemon.mountdir / health_name

        health = gkfs_shell.script(

                f"printf healthy > '{health_path}' && test -s '{health_path}'",

                timeout=30,
@@ -53,4 +43,52 @@ def test_daemon_survives_client_abort_during_write(gkfwd_daemon_factory, 
        if client.poll() is None:

            client.kill()

            client.communicate()





def test_daemon_survives_client_abort_during_write(gkfwd_daemon_factory,

                                                   gkfs_shell):

    """A killed intercepted writer must not terminate the forwarding daemon."""

    daemon = gkfwd_daemon_factory.create()

    aborted_path = daemon.mountdir / "aborted_client_write"

    try:

        _kill_client_and_check_health(

                daemon,

                gkfs_shell,

                f"exec dd if=/dev/zero of='{aborted_path}' "

                "bs=4M count=1024 status=none conv=fsync",

                "post_write_abort_health",

        )

    finally:

        daemon.shutdown()







def test_daemon_survives_client_abort_during_metadata_rpc(gkfwd_daemon_factory,

                                                          gkfs_shell):

    """Repeated metadata RPCs from a killed client must not stop the daemon."""

    daemon = gkfwd_daemon_factory.create()

    metadata_path = daemon.mountdir / "aborted_client_metadata"

    try:

        _kill_client_and_check_health(

                daemon,

                gkfs_shell,

                f"while :; do touch '{metadata_path}'; rm -f '{metadata_path}'; done",

                "post_metadata_abort_health",

        )

    finally:

        daemon.shutdown()





def test_daemon_survives_client_abort_during_malleability_rpc(

        gkfwd_daemon_factory, gkfs_shell):

    """A killed mutate-status client must not terminate the daemon."""

    daemon = gkfwd_daemon_factory.create()

    try:

        _kill_client_and_check_health(

                daemon,

                gkfs_shell,

                "while :; do gkfs_malleability mutate status; done",

                "post_malleability_abort_health",

        )

    finally:

        daemon.shutdown()
 
 No newline at end of file