Commit fc84df5a authored by Dominik Charousset's avatar Dominik Charousset

Report errors while connecting two CAF nodes

Connecting two CAF nodes can fail for several reasons:
- unexpected TCP disconnects
- an application ID mismatch
- a CAF version mismatch

Currently, CAF reports all of the above errors as
`disconnect_during_handshake`. Masking the true error makes it
impossible for users to troubleshoot CAF applications or to respond to
errors programmatically. For example, an application may try to
reconnect on actual TCP disconnects but there's no point in
disconnecting to an incompatible node.
parent b05a4b7b
......@@ -124,6 +124,17 @@ enum class sec : uint8_t {
socket_operation_failed = 45,
/// A resource is temporarily unavailable or would block.
unavailable_or_would_block,
/// Connection refused because of incompatible CAF versions.
incompatible_versions,
/// Connection refused because of incompatible application IDs.
incompatible_application_ids,
/// The middleman received a malformed BASP message from another node.
malformed_basp_message,
/// The middleman closed a connection because it failed to serialize or
/// deserialize a payload.
serializing_basp_payload_failed,
/// The middleman closed a connection to itself or an already connected node.
redundant_connection,
/// Resolving a path on a remote node failed.
remote_lookup_failed,
};
......
......@@ -18,6 +18,7 @@
#pragma once
#include "caf/sec.hpp"
namespace caf {
namespace io {
......@@ -25,7 +26,8 @@ namespace basp {
/// @addtogroup BASP
/// Denotes the state of a connection between to BASP nodes.
/// Denotes the state of a connection between two BASP nodes. Overlaps with
/// `sec` (these states get converted to an error by the BASP instance).
enum connection_state {
/// Indicates that a connection is established and this node is
/// waiting for the next BASP header.
......@@ -34,14 +36,67 @@ enum connection_state {
/// and is waiting for the data.
await_payload,
/// Indicates that this connection no longer exists.
close_connection
close_connection,
/// See `sec::incompatible_versions`.
incompatible_versions,
/// See `sec::incompatible_application_ids`.
incompatible_application_ids,
/// See `sec::malformed_basp_message`.
malformed_basp_message,
/// See `sec::serializing_basp_payload_failed`.
serializing_basp_payload_failed,
/// See `sec::redundant_connection`.
redundant_connection,
/// See `sec::no_route_to_receiving_node`.
no_route_to_receiving_node
};
/// Returns whether the connection state requries a shutdown of the socket
/// connection.
/// @relates connection_state
inline bool requires_shutdown(connection_state x) {
// Any enum value other than await_header (0) and await_payload (1) signal the
// BASP broker to shutdown the connection.
return static_cast<int>(x) > 1;
}
/// Converts the connection state to a system error code if it holds one of the
/// overlapping values. Otherwise returns `sec::none`.
/// @relates connection_state
inline sec to_sec(connection_state x) {
switch (x) {
default:
return sec::none;
case incompatible_versions:
return sec::incompatible_versions;
case incompatible_application_ids:
return sec::incompatible_application_ids;
case malformed_basp_message:
return sec::malformed_basp_message;
case serializing_basp_payload_failed:
return sec::serializing_basp_payload_failed;
case redundant_connection:
return sec::redundant_connection;
case no_route_to_receiving_node:
return sec::no_route_to_receiving_node;
}
}
/// @relates connection_state
inline std::string to_string(connection_state x) {
return x == await_header ? "await_header"
: (x == await_payload ? "await_payload"
: "close_connection");
// TODO: auto-generate this to_string implementation
static constexpr const char* strings[] = {
"await_header",
"await_payload",
"close_connection",
"incompatible_versions",
"incompatible_application_ids",
"malformed_basp_message",
"serializing_basp_payload_failed",
"redundant_connection",
"no_route_to_receiving_node",
};
return strings[static_cast<size_t>(x)];
}
/// @}
......
......@@ -225,8 +225,8 @@ public:
return system().config();
}
bool handle(execution_unit* ctx, connection_handle hdl, header& hdr,
std::vector<char>* payload);
connection_state handle(execution_unit* ctx, connection_handle hdl,
header& hdr, std::vector<char>* payload);
private:
void forward(execution_unit* ctx, const node_id& dest_node, const header& hdr,
......
......@@ -111,7 +111,7 @@ public:
void set_context(connection_handle hdl);
/// Cleans up any state for `hdl`.
void connection_cleanup(connection_handle hdl);
void connection_cleanup(connection_handle hdl, sec code);
/// Sends a basp::down_message message to a remote node.
void send_basp_down_message(const node_id& nid, actor_id aid, error err);
......
......@@ -54,10 +54,10 @@ connection_state instance::handle(execution_unit* ctx, new_data_msg& dm,
header& hdr, bool is_payload) {
CAF_LOG_TRACE(CAF_ARG(dm) << CAF_ARG(is_payload));
// function object providing cleanup code on errors
auto err = [&]() -> connection_state {
auto err = [&](connection_state code) {
if (auto nid = tbl_.erase_direct(dm.handle))
callee_.purge_state(nid);
return close_connection;
return code;
};
std::vector<char>* payload = nullptr;
if (is_payload) {
......@@ -65,14 +65,14 @@ connection_state instance::handle(execution_unit* ctx, new_data_msg& dm,
if (payload->size() != hdr.payload_len) {
CAF_LOG_WARNING("received invalid payload, expected"
<< hdr.payload_len << "bytes, got" << payload->size());
return err();
return err(malformed_basp_message);
}
} else {
binary_deserializer bd{ctx, dm.buf};
auto e = bd(hdr);
if (e || !valid(hdr)) {
CAF_LOG_WARNING("received invalid header:" << CAF_ARG(hdr));
return err();
return err(malformed_basp_message);
}
if (hdr.payload_len > 0) {
CAF_LOG_DEBUG("await payload before processing further");
......@@ -80,9 +80,7 @@ connection_state instance::handle(execution_unit* ctx, new_data_msg& dm,
}
}
CAF_LOG_DEBUG(CAF_ARG(hdr));
if (!handle(ctx, dm.handle, hdr, payload))
return err();
return await_header;
return handle(ctx, dm.handle, hdr, payload);
}
void instance::handle_heartbeat(execution_unit* ctx) {
......@@ -286,18 +284,18 @@ void instance::write_heartbeat(execution_unit* ctx, buffer_type& buf) {
write(ctx, buf, hdr);
}
bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
std::vector<char>* payload) {
connection_state instance::handle(execution_unit* ctx, connection_handle hdl,
header& hdr, std::vector<char>* payload) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(hdr));
// Check payload validity.
if (payload == nullptr) {
if (hdr.payload_len != 0) {
CAF_LOG_WARNING("invalid payload");
return false;
CAF_LOG_WARNING("missing payload");
return malformed_basp_message;
}
} else if (hdr.payload_len != payload->size()) {
CAF_LOG_WARNING("invalid payload");
return false;
CAF_LOG_WARNING("actual payload size differs from advertised size");
return malformed_basp_message;
}
// Dispatch by message type.
switch (hdr.operation) {
......@@ -311,7 +309,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
if (auto err = bd(source_node, app_ids, aid, sigs)) {
CAF_LOG_WARNING("unable to deserialize payload of server handshake:"
<< ctx->system().render(err));
return false;
return serializing_basp_payload_failed;
}
// Check the application ID.
auto whitelist = get_or(config(), "middleman.app-identifiers",
......@@ -321,20 +319,20 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
if (i == app_ids.end()) {
CAF_LOG_WARNING("refuse to connect to server due to app ID mismatch:"
<< CAF_ARG(app_ids) << CAF_ARG(whitelist));
return false;
return incompatible_application_ids;
}
// Close connection to ourselves immediately after sending client HS.
if (source_node == this_node_) {
CAF_LOG_DEBUG("close connection to self immediately");
callee_.finalize_handshake(source_node, aid, sigs);
return false;
return redundant_connection;
}
// Close this connection if we already have a direct connection.
if (tbl_.lookup_direct(source_node)) {
CAF_LOG_DEBUG(
"close redundant direct connection:" << CAF_ARG(source_node));
callee_.finalize_handshake(source_node, aid, sigs);
return false;
return redundant_connection;
}
// Add direct route to this node and remove any indirect entry.
CAF_LOG_DEBUG("new direct connection:" << CAF_ARG(source_node));
......@@ -344,7 +342,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
auto path = tbl_.lookup(source_node);
if (!path) {
CAF_LOG_ERROR("no route to host after server handshake");
return false;
return no_route_to_receiving_node;
}
write_client_handshake(ctx, callee_.get_buffer(path->hdl));
callee_.learned_new_node_directly(source_node, was_indirect);
......@@ -359,7 +357,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
if (auto err = bd(source_node)) {
CAF_LOG_WARNING("unable to deserialize payload of client handshake:"
<< ctx->system().render(err));
return false;
return serializing_basp_payload_failed;
}
// Drop repeated handshakes.
if (tbl_.lookup_direct(source_node)) {
......@@ -383,11 +381,11 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
CAF_LOG_WARNING(
"unable to deserialize source and destination for routed message:"
<< ctx->system().render(err));
return false;
return serializing_basp_payload_failed;
}
if (dest_node != this_node_) {
forward(ctx, dest_node, hdr, *payload);
return true;
return await_header;
}
auto last_hop = tbl_.lookup_direct(hdl);
if (source_node != none && source_node != this_node_
......@@ -441,7 +439,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
if (auto err = bd(source_node, dest_node)) {
CAF_LOG_WARNING("unable to deserialize payload of monitor message:"
<< ctx->system().render(err));
return false;
return serializing_basp_payload_failed;
}
if (dest_node == this_node_)
callee_.proxy_announced(source_node, hdr.dest_actor);
......@@ -458,7 +456,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
if (auto err = bd(source_node, dest_node, fail_state)) {
CAF_LOG_WARNING("unable to deserialize payload of down message:"
<< ctx->system().render(err));
return false;
return serializing_basp_payload_failed;
}
if (dest_node == this_node_) {
// Delay this message to make sure we don't skip in-flight messages.
......@@ -481,10 +479,10 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
}
default: {
CAF_LOG_ERROR("invalid operation");
return false;
return malformed_basp_message;
}
}
return true;
return await_header;
}
void instance::forward(execution_unit* ctx, const node_id& dest_node,
......
......@@ -128,8 +128,8 @@ behavior basp_broker::make_behavior() {
auto& ctx = *this_context;
auto next = instance.handle(context(), msg, ctx.hdr,
ctx.cstate == basp::await_payload);
if (next == basp::close_connection) {
connection_cleanup(msg.handle);
if (requires_shutdown(next)) {
connection_cleanup(msg.handle, to_sec(next));
close(msg.handle);
return;
}
......@@ -226,7 +226,9 @@ behavior basp_broker::make_behavior() {
delete_atom::value, msg.handle));
},
// received from the message handler above for connection_closed_msg
[=](delete_atom, connection_handle hdl) { connection_cleanup(hdl); },
[=](delete_atom, connection_handle hdl) {
connection_cleanup(hdl, sec::none);
},
// received from underlying broker implementation
[=](const acceptor_closed_msg& msg) {
CAF_LOG_TRACE("");
......@@ -564,8 +566,8 @@ void basp_broker::set_context(connection_handle hdl) {
t_last_hop = &i->second.id;
}
void basp_broker::connection_cleanup(connection_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(hdl));
void basp_broker::connection_cleanup(connection_handle hdl, sec code) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(code));
// Remove handle from the routing table and clean up any node-specific state
// we might still have.
if (auto nid = instance.tbl().erase_direct(hdl))
......@@ -577,8 +579,9 @@ void basp_broker::connection_cleanup(connection_handle hdl) {
auto& ref = i->second;
CAF_ASSERT(i->first == ref.hdl);
if (ref.callback) {
CAF_LOG_DEBUG("connection closed during handshake");
ref.callback->deliver(sec::disconnect_during_handshake);
CAF_LOG_DEBUG("connection closed during handshake:" << CAF_ARG(code));
auto x = code != sec::none ? code : sec::disconnect_during_handshake;
ref.callback->deliver(x);
}
ctx.erase(i);
}
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment