: peer_addr( endpoint ),

strand( net_plugin_impl::get()->thread_pool->get_executor() ),

socket( new tcp_socket( net_plugin_impl::get()->thread_pool->get_executor() ) ),

log_p2p_address( endpoint ),

connection_id( ++net_plugin_impl::get()->current_connection_id ),

response_expected_timer( net_plugin_impl::get()->thread_pool->get_executor() ),

last_handshake_recv(),

last_handshake_sent(),

handshake_backoff_floor(std::chrono::milliseconds(net_plugin_impl::get_handshake_backoff_floor_ms())),

handshake_backoff_cap(std::chrono::milliseconds(net_plugin_impl::get_handshake_backoff_cap_ms()))

{

fc_ilog( net_plugin_impl::get_logger(), "created connection {c} to {n}", ("c", connection_id)("n", endpoint) );

fc_ilog(net_plugin_impl::get_logger(), "handshake backoff control: floor={f}ms, cap={c}ms",

("f", net_plugin_impl::get_handshake_backoff_floor_ms())

("c", net_plugin_impl::get_handshake_backoff_cap_ms()));

: peer_addr(),

strand( net_plugin_impl::get()->thread_pool->get_executor() ),

socket( new tcp_socket( net_plugin_impl::get()->thread_pool->get_executor() ) ),

connection_id( ++net_plugin_impl::get()->current_connection_id ),

response_expected_timer( net_plugin_impl::get()->thread_pool->get_executor() ),

last_handshake_recv(),

last_handshake_sent(),

handshake_backoff_floor(std::chrono::milliseconds(net_plugin_impl::get_handshake_backoff_floor_ms())),

handshake_backoff_cap(std::chrono::milliseconds(net_plugin_impl::get_handshake_backoff_cap_ms()))

{

fc_dlog( net_plugin_impl::get_logger(), "new connection object created" );

fc_ilog(net_plugin_impl::get_logger(), "handshake backoff control: floor={f}ms, cap={c}ms",

("f", net_plugin_impl::get_handshake_backoff_floor_ms())

("c", net_plugin_impl::get_handshake_backoff_cap_ms()));

boost::system::error_code ec;

boost::system::error_code ec2;

auto rep = socket->remote_endpoint(ec);

auto lep = socket->local_endpoint(ec2);

log_remote_endpoint_ip = ec ? unknown : rep.address().to_string();

log_remote_endpoint_port = ec ? unknown : std::to_string(rep.port());

local_endpoint_ip = ec2 ? unknown : lep.address().to_string();

local_endpoint_port = ec2 ? unknown : std::to_string(lep.port());

std::lock_guard<std::mutex> g_conn( conn_mtx );

remote_endpoint_ip = log_remote_endpoint_ip;

ci.log_p2p_address = log_p2p_address;

ci.connection_id = connection_id;

ci.conn_node_id = conn_node_id;

ci.short_conn_node_id = short_conn_node_id;

ci.log_remote_endpoint_ip = log_remote_endpoint_ip;

ci.log_remote_endpoint_port = log_remote_endpoint_port;

ci.local_endpoint_ip = local_endpoint_ip;

ci.local_endpoint_port = local_endpoint_port;

// host:port:[<trx>|<blk>]

string::size_type colon = peer_add.find(':');

string::size_type colon2 = peer_add.find(':', colon + 1);

string::size_type end = colon2 == string::npos

? string::npos : peer_add.find_first_of( " :+=.,<>!$%^&(*)|-#@\t", colon2 + 1 ); // future proof by including most symbols without using regex

string host = peer_add.substr( 0, colon );

string port = peer_add.substr( colon + 1, colon2 == string::npos ? string::npos : colon2 - (colon + 1));

string type = colon2 == string::npos ? "" : end == string::npos ?

peer_add.substr( colon2 + 1 ) : peer_add.substr( colon2 + 1, end - (colon2 + 1) );

if( type.empty() ) {

fc_dlog( net_plugin_impl::get_logger(), "Setting connection {c} type for: {peer} to both transactions and blocks", ("c", connection_id)("peer", peer_add) );

connection_type = both;

} else if( type == "trx" ) {

fc_dlog( net_plugin_impl::get_logger(), "Setting connection {c} type for: {peer} to transactions only", ("c", connection_id)("peer", peer_add) );

connection_type = transactions_only;

} else if( type == "blk" ) {

fc_dlog( net_plugin_impl::get_logger(), "Setting connection {c} type for: {peer} to blocks only", ("c", connection_id)("peer", peer_add) );

connection_type = blocks_only;

} else {

fc_wlog( net_plugin_impl::get_logger(), "Unknown connection {c} type: {t}, for {peer}", ("c", connection_id)("t", type)("peer", peer_add) );

}

connection_status stat;

stat.peer = peer_addr;

stat.connecting = connecting;

stat.syncing = syncing;

std::lock_guard<std::mutex> g( conn_mtx );

stat.last_handshake = last_handshake_recv;

return stat;

verify_strand_in_this_thread( strand, __func__, __LINE__ );

update_endpoints();

update_logger_connection_info();

boost::asio::ip::tcp::no_delay nodelay( true );

boost::system::error_code ec;

socket->set_option( nodelay, ec );

if( ec ) {

peer_elog( this, "connection failed (set_option): {e1}", ( "e1", ec.message() ) );

close();

return false;

} else {

peer_dlog( this, "connected" );

socket_open = true;

start_read_message();

return true;

}

strand.post( [self = shared_from_this(), reconnect, shutdown]() {

connection::_close( self, reconnect, shutdown );

});

self->socket_open = false;

boost::system::error_code ec;

if( self->socket->is_open() ) {

self->socket->shutdown( tcp_socket::shutdown_both, ec );

self->socket->close( ec );

}

self->socket.reset( new tcp_socket( net_plugin_impl::get()->thread_pool->get_executor() ) );

self->flush_queues();

self->connecting = false;

self->syncing = false;

self->block_status_monitor_.reset();

++self->consecutive_immediate_connection_close;

bool has_last_req = false;

{

std::lock_guard<std::mutex> g_conn( self->conn_mtx );

has_last_req = self->last_req.has_value();

self->last_handshake_recv = handshake_message();

self->last_handshake_sent = handshake_message();

self->last_close = fc::time_point::now();

self->conn_node_id = fc::sha256();

}

if( has_last_req && !shutdown ) {

net_plugin_impl::get()->dispatcher->retry_fetch( self->shared_from_this() );

}

self->peer_requested.reset();

self->sent_handshake_count = 0;

if( !shutdown) {

try {

auto lock = net_plugin_impl::get()->sm_impl().locked_sml_mutex();

net_plugin_impl::get()->sync_sm->process_event( net_plugin_impl::sync_man_sm_impl::close_connection{self} );

} FC_LOG_AND_RETHROW();

}

peer_ilog( self, "closing" );

self->cancel_wait();

if( reconnect && !shutdown ) {

net_plugin_impl::get()->start_conn_timer( std::chrono::milliseconds( 100 ), wptr() );

}

uint32_t head_num = 0;

std::tie( std::ignore, std::ignore, head_num,

std::ignore, std::ignore, std::ignore ) = net_plugin_impl::get()->get_chain_info();

peer_dlog(this, "head_num = {h}",("h",head_num));

if(head_num == 0) {

notice_message note;

note.known_blocks.mode = normal;

note.known_blocks.pending = 0;

enqueue(note);

return;

}

std::unique_lock<std::mutex> g_conn( conn_mtx );

if( last_handshake_recv.generation >= 1 ) {

peer_dlog( this, "maybe truncating branch at = {h}:{id}",

("h", block_header::num_from_id(last_handshake_recv.head_id))("id", last_handshake_recv.head_id) );

}

block_id_type lib_id = last_handshake_recv.last_irreversible_block_id;

g_conn.unlock();

const auto lib_num = block_header::num_from_id(lib_id);

if( lib_num == 0 ) return; // if last_irreversible_block_id is null (we have not received handshake or reset)

app().post( priority::medium, [chain_plug = net_plugin_impl::get()->chain_plug, c = shared_from_this(),

lib_num, head_num, msg_head_id]() {

auto msg_head_num = block_header::num_from_id(msg_head_id);

bool on_fork = msg_head_num == 0;

bool unknown_block = false;

if( !on_fork ) {

try {

const controller& cc = chain_plug->chain();

block_id_type my_id = cc.get_block_id_for_num( msg_head_num );

on_fork = my_id != msg_head_id;

} catch( const unknown_block_exception& ) {

unknown_block = true;

} catch( ... ) {

on_fork = true;

}

}

if( unknown_block ) {

c->strand.post( [msg_head_num, c]() {

peer_ilog( c, "Peer asked for unknown block {mn}, sending: benign_other go away", ("mn", msg_head_num) );

c->no_retry = benign_other;

c->enqueue( go_away_message( benign_other ) );

} );

} else {

if( on_fork ) msg_head_num = 0;

// if peer on fork, start at their last lib, otherwise we can start at msg_head+1

c->strand.post( [c, msg_head_num, lib_num, head_num]() {

c->blk_send_branch_impl( msg_head_num, lib_num, head_num );

} );

}

} );

if( !peer_requested ) {

auto last = msg_head_num != 0 ? msg_head_num : lib_num;

peer_requested = peer_sync_state( last+1, head_num, last );

} else {

auto last = msg_head_num != 0 ? msg_head_num : std::min( peer_requested->last, lib_num );

uint32_t end = std::max( peer_requested->end_block, head_num );

peer_requested = peer_sync_state( last+1, end, last );

}

if( peer_requested->start_block <= peer_requested->end_block ) {

peer_ilog( this, "enqueue {s} - {e}", ("s", peer_requested->start_block)("e", peer_requested->end_block) );

enqueue_sync_block();

} else {

peer_ilog( this, "nothing to enqueue" );

peer_requested.reset();

}

wptr weak = shared_from_this();

app().post( priority::medium, [blkid, weak{std::move(weak)}]() {

ptr c = weak.lock();

if( !c ) return;

try {

controller& cc = net_plugin_impl::get()->chain_plug->chain();

signed_block_ptr b = cc.fetch_block_by_id( blkid );

if( b ) {

fc_dlog( net_plugin_impl::get_logger(), "fetch_block_by_id num {n}, connection {cid}",

("n", b->block_num())("cid", c->connection_id) );

net_plugin_impl::get()->dispatcher->add_peer_block( blkid, c->connection_id );

c->strand.post( [c, b{std::move(b)}]() {

c->enqueue_block( b );

} );

} else {

fc_ilog( net_plugin_impl::get_logger(), "fetch block by id returned null, id {id}, connection {cid}",

("id", blkid)("cid", c->connection_id) );

}

} catch( const assert_exception& ex ) {

fc_elog( net_plugin_impl::get_logger(), "caught assert on fetch_block_by_id, {ex}, id {id}, connection {cid}",

("ex", ex.to_string())("id", blkid)("cid", c->connection_id) );

} catch( ... ) {

fc_elog( net_plugin_impl::get_logger(), "caught other exception fetching block id {id}, connection {cid}",

("id", blkid)("cid", c->connection_id) );

}

});

strand.post( [c = shared_from_this()]() {

if (net_plugin_impl::get()->in_shutdown) {

peer_dlog(c, "net plugin is in shutdown, will not enqueue the handshake, return");

return;

}

std::unique_lock<std::mutex> g_conn( c->conn_mtx );

// backoff should take place before populate_handshake(), during which the clock time is written in the message

c->backoff_handshake();

if( c->populate_handshake( c->last_handshake_sent ) ) {

static_assert( std::is_same_v<decltype( c->sent_handshake_count ), int16_t>, "INT16_MAX based on int16_t" );

if( c->sent_handshake_count == INT16_MAX ) c->sent_handshake_count = 1; // do not wrap

c->last_handshake_sent.generation = ++c->sent_handshake_count;

auto last_handshake_sent = c->last_handshake_sent;

g_conn.unlock();

peer_dlog( c, "Sending handshake generation {g}, lib {lib}, head {head}, id {id}",

("g", last_handshake_sent.generation)

("lib", last_handshake_sent.last_irreversible_block_num)

("head", last_handshake_sent.head_num)("id", last_handshake_sent.head_id.str().substr(8,16)) );

c->enqueue( last_handshake_sent );

}

});

if( protocol_version >= heartbeat_interval && latest_msg_time > 0 ) {

if( current_time > latest_msg_time + hb_timeout ) {

no_retry = benign_other;

if( !peer_address().empty() ) {

peer_wlog(this, "heartbeat timed out for peer address");

close(true);

} else {

peer_wlog( this, "heartbeat timed out" );

close(false);

}

return;

} else {

const tstamp timeout = std::max(hb_timeout/2, 2*std::chrono::milliseconds(config::block_interval_ms).count());

if ( current_time > latest_blk_time + timeout ) {

send_handshake();

return;

}

}

}

send_time();

time_message xpkt;

xpkt.org = rec;

xpkt.rec = dst;

xpkt.xmt = get_time();

org = xpkt.xmt;

enqueue(xpkt);

time_message xpkt;

xpkt.org = msg.xmt;

xpkt.rec = msg.dst;

xpkt.xmt = get_time();

enqueue(xpkt);

std::function<void(boost::system::error_code, std::size_t)> callback,

bool to_sync_queue) {

if( !buffer_queue.add_write_queue( buff, callback, to_sync_queue )) {

peer_wlog( this, "write_queue full {s} bytes, giving up on connection", ("s", buffer_queue.write_queue_size()) );

close();

return;

}

do_queue_write();

if( !buffer_queue.ready_to_send() )

return;

if (net_plugin_impl::get()->in_shutdown) {

peer_dlog(this, "net plugin is in shutdown, will not do queue write, return");

return;

}

ptr c(shared_from_this());

std::vector<boost::asio::const_buffer> bufs;

buffer_queue.fill_out_buffer( bufs );

strand.post( [c{std::move(c)}, bufs{std::move(bufs)}]() {

boost::asio::async_write( *c->socket, bufs,

boost::asio::bind_executor( c->strand, [c, socket=c->socket]( boost::system::error_code ec, std::size_t w ) {

try {

c->buffer_queue.clear_out_queue();

// May have closed connection and cleared buffer_queue

if( !c->socket_is_open() || socket != c->socket ) {

peer_dlog( c, "async write socket {r} before callback", ("r", c->socket_is_open() ? "changed" : "closed") );

c->close();

return;

}

if( ec ) {

if( ec.value() != boost::asio::error::eof ) {

peer_elog( c, "Error sending to peer: {i}", ( "i", ec.message() ) );

} else {

peer_wlog( c, "connection closure detected on write" );

}

c->close();

return;

}

c->buffer_queue.out_callback( ec, w );

c->enqueue_sync_block();

c->do_queue_write();

} catch ( const std::bad_alloc& ) {

throw;

} catch ( const boost::interprocess::bad_alloc& ) {

throw;

} catch( const fc::exception& ex ) {

peer_elog( c, "fc::exception in do_queue_write: {s}", ("s", ex.to_string()) );

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

peer_elog( c, "std::exception in do_queue_write: {s}", ("s", ex.what()) );

} catch( ... ) {

peer_elog( c, "Unknown exception in do_queue_write" );

}

}));

});

peer_dlog( this, "cancel sync reason = {m}, write queue size {o} bytes",

("m", reason_str( reason ))("o", buffer_queue.write_queue_size()) );

cancel_wait();

flush_queues();

switch (reason) {

case validation :

case fatal_other : {

no_retry = reason;

enqueue( go_away_message( reason ));

break;

}

default:

peer_ilog(this, "sending empty request but not calling sync wait");

enqueue( sync_request_message{0,0} );

}

if( !peer_requested ) {

return false;

} else {

peer_dlog( this, "enqueue sync block {num}", ("num", peer_requested->last + 1) );

}

uint32_t num = ++peer_requested->last;

if(num == peer_requested->end_block) {

peer_requested.reset();

peer_ilog( this, "completing enqueue_sync_block {num}", ("num", num) );

}

wptr weak = shared_from_this();

app().post( priority::medium, [num, weak{std::move(weak)}]() {

ptr c = weak.lock();

if( !c ) return;

controller& cc = net_plugin_impl::get()->chain_plug->chain();

signed_block_ptr sb;

try {

sb = cc.fetch_block_by_number( num );

} FC_LOG_AND_DROP();

if( sb ) {

c->strand.post( [c, sb{std::move(sb)}]() {

c->enqueue_block( sb, true );

});

} else {

c->strand.post( [c, num]() {

peer_ilog( c, "enqueue sync, unable to fetch block {num}", ("num", num) );

c->send_handshake();

});

}

});

return true;

verify_strand_in_this_thread( strand, __func__, __LINE__ );

go_away_reason close_after_send = no_reason;

if (std::holds_alternative<go_away_message>(m)) {

close_after_send = std::get<go_away_message>(m).reason;

}

buffer_factory buff_factory;

auto send_buffer = buff_factory.get_send_buffer( m );

enqueue_buffer( send_buffer, close_after_send );

peer_dlog( this, "enqueue block {num}", ("num", b->block_num()) );

verify_strand_in_this_thread( strand, __func__, __LINE__ );

block_buffer_factory buff_factory;

auto sb = buff_factory.get_send_buffer( b, protocol_version.load() );

if( !sb ) {

peer_wlog( this, "Sending go away for incomplete block #{n} {id}...",

("n", b->block_num())("id", b->calculate_id().str().substr(8,16)) );

// unable to convert to v0 signed block and client doesn't support proto_pruned_types, so tell it to go away

no_retry = go_away_reason::fatal_other;

enqueue( go_away_message( fatal_other ) );

return;

}

latest_blk_time = get_time();

enqueue_buffer( sb, no_reason, to_sync_queue);

go_away_reason close_after_send,

bool to_sync_queue)

{

ptr self = shared_from_this();

queue_write(send_buffer,

[conn{std::move(self)}, close_after_send](boost::system::error_code ec, std::size_t ) {

if (ec) return;

if (close_after_send != no_reason) {

fc_ilog( net_plugin_impl::get_logger(), "sent a go away message: {r}, closing connection {cid}",

("r", reason_str(close_after_send))("cid", conn->connection_id) );

conn->close();

return;

}

},

to_sync_queue);

ptr c(shared_from_this());

std::lock_guard<std::mutex> g( response_expected_timer_mtx );

response_expected_timer.expires_from_now( net_plugin_impl::get()->resp_expected_period );

response_expected_timer.async_wait(

boost::asio::bind_executor( c->strand, [c]( boost::system::error_code ec ) {

c->sync_timeout( ec );

} ) );

ptr c( shared_from_this() );

std::lock_guard<std::mutex> g( response_expected_timer_mtx );

response_expected_timer.expires_from_now( net_plugin_impl::get()->resp_expected_period );

response_expected_timer.async_wait(

boost::asio::bind_executor( c->strand, [c]( boost::system::error_code ec ) {

c->fetch_timeout(ec);

} ) );

if( !ec ) {

{

if (!net_plugin_impl::get()->sync_man().is_sync_source(*this))

return;

}

cancel_sync(benign_other);

{

auto lock = net_plugin_impl::get()->sm_impl().locked_sml_mutex();

try {

net_plugin_impl::get()->sync_sm->process_event(

net_plugin_impl::sync_man_sm_impl::reassign_fetch{shared_from_this()}

);

} FC_LOG_AND_RETHROW();

}

} else if( ec != boost::asio::error::operation_aborted ) { // don't log on operation_aborted, called on destroy

peer_elog( this, "setting timer for sync request got error {ec}", ("ec", ec.message()) );

}

if( !ec ) {

net_plugin_impl::get()->dispatcher->retry_fetch( shared_from_this() );

} else if( ec != boost::asio::error::operation_aborted ) { // don't log on operation_aborted, called on destroy

peer_elog( this, "setting timer for fetch request got error {ec}", ("ec", ec.message() ) );

}

sync_request_message srm = {start,end};

enqueue( net_message(srm) );

sync_wait();

switch ( no_retry ) {

case no_reason:

case wrong_version:

case benign_other:

break;

default:

fc_dlog( net_plugin_impl::get_logger(), "Skipping connect due to go_away reason {r}",("r", reason_str( no_retry )));

return false;

}

string::size_type colon = peer_address().find(':');

if (colon == std::string::npos || colon == 0) {

fc_elog( net_plugin_impl::get_logger(), "Invalid peer address. must be \"host:port[:<blk>|<trx>]\": {p}", ("p", peer_address()) );

return false;

}

ptr c = shared_from_this();

if( consecutive_immediate_connection_close > def_max_consecutive_immediate_connection_close || no_retry == benign_other ) {

auto connector_period_us = std::chrono::duration_cast<std::chrono::microseconds>( net_plugin_impl::get()->connector_period );

std::lock_guard<std::mutex> g( c->conn_mtx );

if( last_close == fc::time_point() || last_close > fc::time_point::now() - fc::microseconds( connector_period_us.count() ) ) {

return true; // true so doesn't remove from valid connections

}

}

strand.post([c]() {

string::size_type colon = c->peer_address().find(':');

string::size_type colon2 = c->peer_address().find(':', colon + 1);

string host = c->peer_address().substr( 0, colon );

string port = c->peer_address().substr( colon + 1, colon2 == string::npos ? string::npos : colon2 - (colon + 1));

c->set_connection_type( c->peer_address() );

auto resolver = std::make_shared<tcp_resolver>( net_plugin_impl::get()->thread_pool->get_executor() );

wptr weak_conn = c;

// Note: need to add support for IPv6 too

resolver->async_resolve( tcp::v4(), host, port, boost::asio::bind_executor( c->strand,

[resolver, weak_conn, host, port]( const boost::system::error_code& err, tcp::resolver::results_type endpoints ) {

auto c = weak_conn.lock();

if( !c ) return;

if( !err ) {

c->connect( resolver, endpoints );

} else {

fc_elog( net_plugin_impl::get_logger(), "Unable to resolve {host}:{port} {error}",

("host", host)("port", port)( "error", err.message() ) );

c->connecting = false;

++c->consecutive_immediate_connection_close;

}

} ) );

} );

return true;

switch ( no_retry ) {

case no_reason:

case wrong_version:

case benign_other:

break;

default:

return;

}

connecting = true;

pending_message_buffer.reset();

buffer_queue.clear_out_queue();

boost::asio::async_connect( *socket, endpoints,

boost::asio::bind_executor( strand,

[resolver, c = shared_from_this(), socket=socket]( const boost::system::error_code& err, const tcp::endpoint& endpoint ) {

if( !err && socket->is_open() && socket == c->socket ) {

if( c->start_session() ) {

c->send_handshake();

}

} else {

fc_elog( net_plugin_impl::get_logger(), "connection failed to {host}:{port} {error}",

("host", endpoint.address().to_string())("port", endpoint.port())( "error", err.message()));

c->close( false );

}

} ) );

try {

std::size_t minimum_read =

std::atomic_exchange<decltype(outstanding_read_bytes.load())>( &outstanding_read_bytes, 0 );

minimum_read = minimum_read != 0 ? minimum_read : message_header_size;

if (net_plugin_impl::get()->use_socket_read_watermark) {

const size_t max_socket_read_watermark = 4096;

std::size_t socket_read_watermark = std::min<std::size_t>(minimum_read, max_socket_read_watermark);

boost::asio::socket_base::receive_low_watermark read_watermark_opt(socket_read_watermark);

boost::system::error_code ec;

socket->set_option( read_watermark_opt, ec );

if( ec ) {

peer_elog( this, "unable to set read watermark: {e1}", ("e1", ec.message()) );

}

}

auto completion_handler = [minimum_read](boost::system::error_code ec, std::size_t bytes_transferred) -> std::size_t {

if (ec || bytes_transferred >= minimum_read ) {

return 0;

} else {

return minimum_read - bytes_transferred;

}

};

uint32_t write_queue_size = buffer_queue.write_queue_size();

if( write_queue_size > def_max_write_queue_size ) {

peer_elog( this, "write queue full {s} bytes, giving up on connection, closing", ("s", write_queue_size) );

close( false );

return;

}

boost::asio::async_read( *socket,

pending_message_buffer.get_buffer_sequence_for_boost_async_read(), completion_handler,

boost::asio::bind_executor( strand,

[conn = shared_from_this(), socket=socket]( boost::system::error_code ec, std::size_t bytes_transferred ) {

// may have closed connection and cleared pending_message_buffer

if( !conn->socket_is_open() || socket != conn->socket ) return;

bool close_connection = false;

try {

if( !ec ) {

if (bytes_transferred > conn->pending_message_buffer.bytes_to_write()) {

peer_elog( conn, "async_read_some callback: bytes_transferred = {bt}, buffer.bytes_to_write = {btw}",

("bt",bytes_transferred)("btw",conn->pending_message_buffer.bytes_to_write()) );

}

EOS_ASSERT(bytes_transferred <= conn->pending_message_buffer.bytes_to_write(), plugin_exception, "");

conn->pending_message_buffer.advance_write_ptr(bytes_transferred);

while (conn->pending_message_buffer.bytes_to_read() > 0) {

uint32_t bytes_in_buffer = conn->pending_message_buffer.bytes_to_read();

if (bytes_in_buffer < message_header_size) {

conn->outstanding_read_bytes = message_header_size - bytes_in_buffer;

break;

} else {

uint32_t message_length;

auto index = conn->pending_message_buffer.read_index();

conn->pending_message_buffer.peek(&message_length, sizeof(message_length), index);

if(message_length > def_send_buffer_size*2 || message_length == 0) {

peer_elog( conn, "incoming message length unexpected ({i})", ("i", message_length) );

close_connection = true;

break;

}

auto total_message_bytes = message_length + message_header_size;

if (bytes_in_buffer >= total_message_bytes) {

conn->pending_message_buffer.advance_read_ptr(message_header_size);

conn->consecutive_immediate_connection_close = 0;

if (!conn->process_next_message(message_length)) {

return;

}

} else {

auto outstanding_message_bytes = total_message_bytes - bytes_in_buffer;

auto available_buffer_bytes = conn->pending_message_buffer.bytes_to_write();

if (outstanding_message_bytes > available_buffer_bytes) {

conn->pending_message_buffer.add_space( outstanding_message_bytes - available_buffer_bytes );

}

conn->outstanding_read_bytes = outstanding_message_bytes;

break;

}

}

}

if( !close_connection ) conn->start_read_message();

} else {

if (ec.value() != boost::asio::error::eof) {

peer_elog( conn, "Error reading message: {m}", ( "m", ec.message() ) );

} else {

peer_ilog( conn, "Peer closed connection" );

}

close_connection = true;

}

}

catch ( const std::bad_alloc& )

{

throw;

}

catch ( const boost::interprocess::bad_alloc& )

{

throw;

}

catch(const fc::exception &ex)

{

peer_elog( conn, "Exception in handling read data {s}", ("s",ex.to_string()) );

close_connection = true;

}

catch(const std::exception &ex) {

peer_elog( conn, "Exception in handling read data: {s}", ("s",ex.what()) );

close_connection = true;

}

catch (...) {

peer_elog( conn, "Undefined exception handling read data" );

close_connection = true;

}

if( close_connection ) {

peer_elog( conn, "Closing connection" );

conn->close();

}

}));

} catch (...) {

peer_elog( this, "Undefined exception in start_read_message, closing connection" );

close();

}

try {

latest_msg_time = get_time();

// if next message is a block we already have, exit early

auto peek_ds = pending_message_buffer.create_peek_datastream();

unsigned_int which{};

fc::raw::unpack( peek_ds, which );

if( which == signed_block_which || which == signed_block_v0_which ) {

latest_blk_time = get_time();

return process_next_block_message( message_length );

} else if( which == trx_message_v1_which || which == packed_transaction_v0_which ) {

return process_next_trx_message( message_length );

} else {

auto ds = pending_message_buffer.create_datastream();

net_message msg;

fc::raw::unpack( ds, msg );

msg_handler m( shared_from_this() );

std::visit( m, msg );

}

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

peer_elog( this, "Exception in handling message: {s}", ("s", e.to_detail_string()) );

close();

return false;

}

return true;

auto peek_ds = pending_message_buffer.create_peek_datastream();

unsigned_int which{};

fc::raw::unpack( peek_ds, which ); // throw away

block_header bh;

fc::raw::unpack( peek_ds, bh );

const block_id_type blk_id = bh.calculate_id();

const uint32_t blk_num = bh.block_num();

if( net_plugin_impl::get()->dispatcher->have_block( blk_id ) ) {

peer_dlog( this, "canceling wait, already received block {num}, id {id}...",

("num", blk_num)("id", blk_id.str().substr(8,16)) );

if( app().is_quiting() ) {

close( false, true );

} else {

sync_recv_block( blk_id, blk_num, false );

}

cancel_wait();

pending_message_buffer.advance_read_ptr( message_length );

return true;

}

peer_dlog( this, "received block {num}, id {id}..., latency: {latency}",

("num", bh.block_num())("id", blk_id.str().substr(8,16))

("latency", (fc::time_point::now() - bh.timestamp).count()/1000) );

if( !net_plugin_impl::get()->syncing_with_peer() ) { // guard against peer thinking it needs to send us old blocks

uint32_t lib = 0;

std::tie( lib, std::ignore, std::ignore, std::ignore, std::ignore, std::ignore ) = net_plugin_impl::get()->get_chain_info();

if( blk_num < lib ) {

std::unique_lock<std::mutex> g( conn_mtx );

const auto last_sent_lib = last_handshake_sent.last_irreversible_block_num;

g.unlock();

if( blk_num < last_sent_lib ) {

peer_ilog( this, "received block {n} less than sent lib {lib}", ("n", blk_num)("lib", last_sent_lib) );

close();

} else {

peer_ilog( this, "received block {n} less than lib {lib}", ("n", blk_num)("lib", lib) );

net_plugin_impl::get()->sync_man().reset_last_requested_num();

enqueue( sync_request_message{0, 0} );

send_handshake();

cancel_wait();

}

pending_message_buffer.advance_read_ptr( message_length );

return true;

}

}

auto ds = pending_message_buffer.create_datastream();

fc::raw::unpack( ds, which );

shared_ptr<signed_block> ptr;

if( which == signed_block_which ) {

ptr = std::make_shared<signed_block>();

fc::raw::unpack( ds, *ptr );

} else {

signed_block_v0 sb_v0;

fc::raw::unpack( ds, sb_v0 );

ptr = std::make_shared<signed_block>( std::move( sb_v0 ), true );

}

auto is_webauthn_sig = []( const fc::crypto::signature& s ) {

return static_cast<size_t>(s.which()) == fc::get_index<fc::crypto::signature::storage_type, fc::crypto::webauthn::signature>();

};

bool has_webauthn_sig = is_webauthn_sig( ptr->producer_signature );

constexpr auto additional_sigs_eid = additional_block_signatures_extension::extension_id();

auto exts = ptr->validate_and_extract_extensions();

if( exts.count( additional_sigs_eid ) ) {

const auto &additional_sigs = std::get<additional_block_signatures_extension>(exts.lower_bound( additional_sigs_eid )->second).signatures;

has_webauthn_sig |= std::any_of( additional_sigs.begin(), additional_sigs.end(), is_webauthn_sig );

}

if( has_webauthn_sig ) {

peer_dlog( this, "WebAuthn signed block received, closing connection" );

close();

return false;

}

handle_message( blk_id, std::move( ptr ) );

return true;

if( !net_plugin_impl::get()->p2p_accept_transactions ) {

peer_dlog( this, "p2p-accept-transaction=false - dropping txn" );

pending_message_buffer.advance_read_ptr( message_length );