Line data Source code
1 : /* fd_bundle_client.c steps gRPC related tasks. */
2 :
3 : #define _GNU_SOURCE /* SOL_TCP */
4 : #include "fd_bundle_auth.h"
5 : #include "fd_bundle_tile_private.h"
6 : #include "fd_bundle_tile.h"
7 : #include "../waker/fd_waker.h"
8 : #include "../fd_txn_m.h"
9 : #include "../../waltz/h2/fd_h2_conn.h"
10 : #include "../../waltz/http/fd_url.h" /* fd_url_unescape */
11 : #include "../../ballet/base58/fd_base58.h"
12 : #include "../../ballet/ed25519/fd_x25519.h"
13 : #include "../../ballet/pb/fd_pb_tokenize.h"
14 : #include "../../util/net/fd_ip4.h"
15 :
16 : #include <fcntl.h>
17 : #include <errno.h>
18 : #include <unistd.h> /* close */
19 : #include <poll.h> /* poll */
20 : #include <sys/epoll.h>
21 : #include <sys/socket.h> /* socket */
22 : #include <netinet/in.h>
23 : #include <netinet/ip.h>
24 : #include <netinet/tcp.h>
25 :
26 0 : #define FD_BUNDLE_CLIENT_REQUEST_TIMEOUT ((long)8e9) /* 8 seconds */
27 0 : #define FD_BUNDLE_CONN_TIMEOUT_NS ((long)3e9) /* TCP connect plus TLS and HTTP/2 setup */
28 :
29 :
30 : __attribute__((weak)) long
31 0 : fd_bundle_now( fd_bundle_tile_t const * ctx ) {
32 0 : return fd_clock_tile_now( ctx->clock );
33 0 : }
34 :
35 : void
36 0 : fd_bundle_client_reset( fd_bundle_tile_t * ctx ) {
37 0 : if( FD_UNLIKELY( ctx->tcp_sock >= 0 ) ) {
38 0 : if( FD_UNLIKELY( 0!=close( ctx->tcp_sock ) ) ) {
39 0 : FD_LOG_ERR(( "close(tcp_sock=%i) failed (%i-%s)", ctx->tcp_sock, errno, fd_io_strerror( errno ) ));
40 0 : }
41 0 : ctx->tcp_sock = -1;
42 0 : ctx->tcp_sock_connected = 0;
43 0 : ctx->sock_in_epoll = 0;
44 0 : ctx->epoll_out_armed = 0;
45 0 : }
46 0 : fd_tlsrec_conn_fini( ctx->tls_conn );
47 0 : fd_memzero_explicit( ctx->tls->kex_private_key, 32UL );
48 0 : ctx->defer_reset = 0;
49 :
50 0 : ctx->builder_info_avail = 0;
51 0 : ctx->builder_info_wait = 0;
52 0 : ctx->packet_subscription_live = 0;
53 0 : ctx->packet_subscription_wait = 0;
54 0 : ctx->bundle_subscription_live = 0;
55 0 : ctx->bundle_subscription_wait = 0;
56 :
57 0 : fd_memset( ctx->rtt, 0, sizeof(fd_rtt_estimate_t) );
58 :
59 0 : fd_bundle_tile_backoff( ctx, fd_bundle_now( ctx ) );
60 :
61 0 : fd_bundle_auther_reset( &ctx->auther );
62 0 : fd_grpc_client_reset( ctx->grpc_client );
63 0 : }
64 :
65 : static int
66 : fd_bundle_client_do_connect( fd_bundle_tile_t const * ctx,
67 0 : uint ip4_addr ) {
68 0 : struct sockaddr_in addr = {
69 0 : .sin_family = AF_INET,
70 0 : .sin_addr.s_addr = ip4_addr,
71 0 : .sin_port = fd_ushort_bswap( ctx->server_tcp_port )
72 0 : };
73 0 : int err = connect( ctx->tcp_sock, fd_type_pun_const( &addr ), sizeof(struct sockaddr_in) );
74 : /* FD_LIKELY is used here as EINPROGRESS is expected even to local tcp ports */
75 0 : if( FD_LIKELY( err==-1 ) ) {
76 0 : return errno;
77 0 : }
78 0 : return 0;
79 0 : }
80 :
81 : static int
82 0 : fd_bundle_client_get_connect_result( fd_bundle_tile_t const * ctx ) {
83 0 : int so_err = 0;
84 0 : socklen_t so_err_sz = sizeof(so_err);
85 0 : if( FD_UNLIKELY( getsockopt( ctx->tcp_sock, SOL_SOCKET, SO_ERROR, &so_err, &so_err_sz )==-1 ) ) {
86 0 : return errno;
87 0 : }
88 0 : return so_err;
89 0 : }
90 :
91 : static void
92 0 : fd_bundle_client_create_conn( fd_bundle_tile_t * ctx ) {
93 0 : fd_bundle_client_reset( ctx );
94 :
95 : /* FIXME IPv6 support */
96 0 : fd_addrinfo_t hints = {0};
97 0 : hints.ai_family = AF_INET;
98 0 : fd_addrinfo_t * res = NULL;
99 0 : uchar scratch[ 4096 ];
100 0 : void * pscratch = scratch;
101 0 : int err = fd_getaddrinfo( ctx->server_fqdn, &hints, &res, &pscratch, sizeof(scratch) );
102 0 : if( FD_UNLIKELY( err ) ) {
103 0 : FD_LOG_WARNING(( "fd_getaddrinfo `%s` failed (%d-%s)", ctx->server_fqdn, err, fd_gai_strerror( err ) ));
104 0 : fd_bundle_client_reset( ctx );
105 0 : ctx->metrics.transport_fail_cnt++;
106 0 : return;
107 0 : }
108 0 : uint const ip4_addr = ((struct sockaddr_in *)res->ai_addr)->sin_addr.s_addr;
109 0 : ctx->server_ip4_addr = ip4_addr;
110 :
111 0 : int tcp_sock = socket( AF_INET, SOCK_STREAM|SOCK_CLOEXEC, 0 );
112 0 : if( FD_UNLIKELY( tcp_sock<0 ) ) {
113 0 : FD_LOG_ERR(( "socket(AF_INET,SOCK_STREAM|SOCK_CLOEXEC,0) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
114 0 : }
115 0 : ctx->tcp_sock = tcp_sock;
116 :
117 0 : struct epoll_event ev = { .events = EPOLLIN|EPOLLOUT, .data.fd = tcp_sock };
118 0 : if( FD_UNLIKELY( -1==epoll_ctl( FD_WAKER_INNER_FD( ctx->waker_client_idx ), EPOLL_CTL_ADD, tcp_sock, &ev ) ) ) {
119 0 : FD_LOG_ERR(( "epoll_ctl(ADD,tcp_sock) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
120 0 : }
121 0 : ctx->sock_in_epoll = 1;
122 0 : ctx->epoll_out_armed = 1;
123 :
124 0 : if( FD_UNLIKELY( 0!=setsockopt( tcp_sock, SOL_SOCKET, SO_RCVBUF, &ctx->so_rcvbuf, sizeof(int) ) ) ) {
125 0 : FD_LOG_ERR(( "setsockopt(SOL_SOCKET,SO_RCVBUF,%i) failed (%i-%s)", ctx->so_rcvbuf, errno, fd_io_strerror( errno ) ));
126 0 : }
127 :
128 0 : int tcp_nodelay = 1;
129 0 : if( FD_UNLIKELY( 0!=setsockopt( tcp_sock, SOL_TCP, TCP_NODELAY, &tcp_nodelay, sizeof(int) ) ) ) {
130 0 : FD_LOG_ERR(( "setsockopt failed (%d-%s)", errno, fd_io_strerror( errno ) ));
131 0 : }
132 :
133 0 : if( FD_UNLIKELY( fcntl( tcp_sock, F_SETFL, O_NONBLOCK )==-1 ) ) {
134 0 : FD_LOG_ERR(( "fcntl(tcp_sock,F_SETFL,O_NONBLOCK) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
135 0 : }
136 :
137 0 : char const * scheme = ctx->is_ssl ? "https" : "http";
138 :
139 0 : FD_LOG_INFO(( "Connecting to %s://" FD_IP4_ADDR_FMT ":%hu (%.*s)",
140 0 : scheme,
141 0 : FD_IP4_ADDR_FMT_ARGS( ip4_addr ), ctx->server_tcp_port,
142 0 : (int)ctx->server_sni_len, ctx->server_sni ));
143 :
144 0 : int connect_err = fd_bundle_client_do_connect( ctx, ip4_addr );
145 : /* FD_LIKELY as EINPROGRESS is expected */
146 0 : if( FD_LIKELY( connect_err ) ) {
147 0 : if( FD_UNLIKELY( connect_err!=EINPROGRESS ) ) {
148 0 : FD_LOG_WARNING(( "connect(tcp_sock," FD_IP4_ADDR_FMT ":%u) failed (%i-%s)",
149 0 : FD_IP4_ADDR_FMT_ARGS( ip4_addr ), ctx->server_tcp_port,
150 0 : connect_err, fd_io_strerror( connect_err ) ));
151 0 : fd_bundle_client_reset( ctx );
152 0 : ctx->metrics.transport_fail_cnt++;
153 0 : return;
154 0 : }
155 0 : }
156 :
157 0 : if( ctx->is_ssl ) {
158 0 : fd_tls_t * tls = ctx->tls;
159 0 : ulong sni_len = strlen( ctx->server_sni );
160 0 : fd_memcpy( tls->server_name, ctx->server_sni, sni_len );
161 0 : tls->server_name[ sni_len ] = '\0';
162 0 : tls->server_name_len = (ushort)sni_len;
163 0 : if( FD_UNLIKELY( !fd_rng_secure( tls->kex_private_key, 32UL ) ) ) FD_LOG_CRIT(( "fd_rng_secure failed" ));
164 0 : fd_x25519_public( tls->kex_public_key, tls->kex_private_key );
165 0 : fd_tlsrec_conn_init( ctx->tls_conn, tls, 0 );
166 0 : }
167 :
168 0 : fd_grpc_client_reset( ctx->grpc_client );
169 0 : fd_keepalive_init( ctx->keepalive, ctx->rng, ctx->keepalive_interval, ctx->keepalive_interval, fd_bundle_now( ctx ) );
170 0 : ctx->conn_deadline = fd_bundle_now( ctx ) + FD_BUNDLE_CONN_TIMEOUT_NS;
171 0 : }
172 :
173 : static int
174 : fd_bundle_client_drive_io( fd_bundle_tile_t * ctx,
175 : long now,
176 0 : int * charge_busy ) {
177 0 : if( ctx->is_ssl ) {
178 0 : return fd_grpc_client_rxtx_tls( ctx->grpc_client, ctx->tls_conn, ctx->tcp_sock, now, charge_busy );
179 0 : }
180 :
181 0 : return fd_grpc_client_rxtx_socket( ctx->grpc_client, ctx->tcp_sock, now, charge_busy );
182 0 : }
183 :
184 : static void
185 0 : fd_bundle_client_request_builder_info( fd_bundle_tile_t * ctx ) {
186 0 : if( FD_UNLIKELY( fd_grpc_client_request_is_blocked( ctx->grpc_client ) ) ) return;
187 :
188 0 : static char const path[] = "/block_engine.BlockEngineValidator/GetBlockBuilderFeeInfo";
189 0 : fd_grpc_h2_stream_t * request = fd_grpc_client_request_start(
190 0 : ctx->grpc_client,
191 0 : path, sizeof(path)-1,
192 0 : FD_BUNDLE_CLIENT_REQ_Bundle_GetBlockBuilderFeeInfo,
193 0 : (uchar const *)"", 0UL, /* empty request message */
194 0 : ctx->auther.access_token, ctx->auther.access_token_sz,
195 0 : 0 /* is_streaming */
196 0 : );
197 0 : if( FD_UNLIKELY( !request ) ) return;
198 0 : fd_grpc_client_deadline_set(
199 0 : request,
200 0 : FD_GRPC_DEADLINE_RX_END,
201 0 : fd_log_wallclock() + FD_BUNDLE_CLIENT_REQUEST_TIMEOUT );
202 :
203 0 : ctx->builder_info_wait = 1;
204 0 : }
205 :
206 : static void
207 0 : fd_bundle_client_subscribe_packets( fd_bundle_tile_t * ctx ) {
208 0 : if( FD_UNLIKELY( fd_grpc_client_request_is_blocked( ctx->grpc_client ) ) ) return;
209 :
210 0 : static char const path[] = "/block_engine.BlockEngineValidator/SubscribePackets";
211 0 : fd_grpc_h2_stream_t * request = fd_grpc_client_request_start(
212 0 : ctx->grpc_client,
213 0 : path, sizeof(path)-1,
214 0 : FD_BUNDLE_CLIENT_REQ_Bundle_SubscribePackets,
215 0 : (uchar const *)"", 0UL, /* empty request message */
216 0 : ctx->auther.access_token, ctx->auther.access_token_sz,
217 0 : 0 /* is_streaming */
218 0 : );
219 0 : if( FD_UNLIKELY( !request ) ) return;
220 0 : fd_grpc_client_deadline_set(
221 0 : request,
222 0 : FD_GRPC_DEADLINE_HEADER,
223 0 : fd_log_wallclock() + FD_BUNDLE_CLIENT_REQUEST_TIMEOUT );
224 :
225 0 : ctx->packet_subscription_wait = 1;
226 0 : }
227 :
228 : static void
229 0 : fd_bundle_client_subscribe_bundles( fd_bundle_tile_t * ctx ) {
230 0 : if( FD_UNLIKELY( fd_grpc_client_request_is_blocked( ctx->grpc_client ) ) ) return;
231 :
232 0 : static char const path[] = "/block_engine.BlockEngineValidator/SubscribeBundles";
233 0 : fd_grpc_h2_stream_t * request = fd_grpc_client_request_start(
234 0 : ctx->grpc_client,
235 0 : path, sizeof(path)-1,
236 0 : FD_BUNDLE_CLIENT_REQ_Bundle_SubscribeBundles,
237 0 : (uchar const *)"", 0UL, /* empty request message */
238 0 : ctx->auther.access_token, ctx->auther.access_token_sz,
239 0 : 0 /* is_streaming */
240 0 : );
241 0 : if( FD_UNLIKELY( !request ) ) return;
242 0 : fd_grpc_client_deadline_set(
243 0 : request,
244 0 : FD_GRPC_DEADLINE_HEADER,
245 0 : fd_log_wallclock() + FD_BUNDLE_CLIENT_REQUEST_TIMEOUT );
246 :
247 0 : ctx->bundle_subscription_wait = 1;
248 0 : }
249 :
250 : void
251 0 : fd_bundle_client_send_ping( fd_bundle_tile_t * ctx ) {
252 0 : if( FD_UNLIKELY( !ctx->grpc_client ) ) return; /* no client */
253 0 : fd_h2_conn_t * conn = fd_grpc_client_h2_conn( ctx->grpc_client );
254 0 : if( FD_UNLIKELY( !conn ) ) return; /* no conn */
255 0 : if( FD_UNLIKELY( conn->flags ) ) return; /* conn busy */
256 0 : fd_h2_rbuf_t * rbuf_tx = fd_grpc_client_rbuf_tx( ctx->grpc_client );
257 :
258 0 : if( FD_LIKELY( fd_h2_tx_ping( conn, rbuf_tx ) ) ) {
259 0 : long now = fd_bundle_now( ctx );
260 0 : fd_keepalive_tx( ctx->keepalive, ctx->rng, now );
261 0 : FD_LOG_DEBUG(( "Keepalive TX (deadline=+%gs)", (double)( ctx->keepalive->ts_deadline-now )/1e9 ));
262 0 : }
263 0 : }
264 :
265 : static void
266 : fd_bundle_client_epoll_out( fd_bundle_tile_t * ctx,
267 0 : int want ) {
268 0 : if( FD_UNLIKELY( !ctx->sock_in_epoll ) ) return;
269 0 : if( FD_LIKELY( want==(int)ctx->epoll_out_armed ) ) return;
270 0 : struct epoll_event ev = { .events = EPOLLIN | (want ? EPOLLOUT : 0U), .data.fd = ctx->tcp_sock };
271 0 : if( FD_UNLIKELY( -1==epoll_ctl( FD_WAKER_INNER_FD( ctx->waker_client_idx ), EPOLL_CTL_MOD, ctx->tcp_sock, &ev ) ) )
272 0 : FD_LOG_ERR(( "epoll_ctl(MOD,tcp_sock) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
273 0 : ctx->epoll_out_armed = !!want;
274 0 : }
275 :
276 : long
277 : fd_bundle_client_next_deadline( fd_bundle_tile_t const * ctx,
278 0 : long now ) {
279 : /* Connecting: completion/failure arrives as an EPOLLOUT/EPOLLERR
280 : event, or the setup times out. */
281 0 : if( FD_UNLIKELY( ctx->tcp_sock>=0 && !ctx->tcp_sock_connected ) ) return ctx->conn_deadline;
282 :
283 : /* Disconnected: next action is the reconnect attempt. */
284 0 : if( FD_UNLIKELY( ctx->tcp_sock<0 ) ) return fd_long_max( ctx->backoff_until, now );
285 :
286 : /* Decrypted TLS bytes buffered in the gRPC client do not make the fd
287 : readable: drain now or they wait for the next unrelated event.
288 : Except while TX is blocked: then they are held because the HTTP/2
289 : rings are full behind the parked send, and only the EPOLLOUT armed
290 : for that send can make progress. */
291 0 : if( FD_UNLIKELY( ctx->is_ssl &&
292 0 : fd_grpc_client_tls_rx_pending( ctx->grpc_client ) &&
293 0 : !fd_grpc_client_tls_tx_pending( ctx->grpc_client ) ) ) return now;
294 :
295 0 : long deadline = fd_grpc_client_next_deadline( ctx->grpc_client );
296 0 : if( FD_LIKELY( ctx->keepalive->interval ) )
297 0 : deadline = fd_long_min( deadline, ctx->keepalive->inflight ? ctx->keepalive->ts_deadline
298 0 : : ctx->keepalive->ts_next_tx );
299 0 : if( FD_LIKELY( ctx->builder_info_avail & !ctx->builder_info_wait ) )
300 0 : deadline = fd_long_min( deadline, ctx->builder_info_valid_until );
301 0 : if( FD_UNLIKELY( ctx->backoff_until>now ) )
302 0 : deadline = fd_long_min( deadline, ctx->backoff_until );
303 0 : if( FD_UNLIKELY( !fd_grpc_client_is_connected( ctx->grpc_client ) ) )
304 0 : deadline = fd_long_min( deadline, ctx->conn_deadline );
305 0 : return deadline;
306 0 : }
307 :
308 : int
309 : fd_bundle_client_step_reconnect( fd_bundle_tile_t * ctx,
310 0 : long now ) {
311 : /* Drive auth */
312 0 : if( FD_UNLIKELY( ctx->auther.needs_poll ) ) {
313 0 : fd_bundle_auther_poll( &ctx->auther, ctx->grpc_client, ctx->keyguard_client );
314 0 : return 1;
315 0 : }
316 0 : if( FD_UNLIKELY( ctx->auther.state!=FD_BUNDLE_AUTH_STATE_DONE_WAIT ) ) return 0;
317 :
318 : /* Request block builder info */
319 0 : int const builder_info_expired = ( ctx->builder_info_valid_until - now )<0;
320 0 : if( FD_UNLIKELY( ( ( !ctx->builder_info_avail ) |
321 0 : ( builder_info_expired ) ) &
322 0 : ( !ctx->builder_info_wait ) ) ) {
323 0 : fd_bundle_client_request_builder_info( ctx );
324 0 : return 1;
325 0 : }
326 :
327 : /* Subscribe to packets */
328 0 : if( FD_UNLIKELY( !ctx->packet_subscription_live && !ctx->packet_subscription_wait ) ) {
329 0 : fd_bundle_client_subscribe_packets( ctx );
330 0 : return 1;
331 0 : }
332 :
333 : /* Subscribe to bundles */
334 0 : if( FD_UNLIKELY( !ctx->bundle_subscription_live && !ctx->bundle_subscription_wait ) ) {
335 0 : fd_bundle_client_subscribe_bundles( ctx );
336 0 : return 1;
337 0 : }
338 :
339 : /* Send a PING */
340 0 : if( FD_UNLIKELY( fd_keepalive_should_tx( ctx->keepalive, now ) ) ) {
341 0 : fd_bundle_client_send_ping( ctx );
342 0 : return 1;
343 0 : }
344 :
345 0 : return 0;
346 0 : }
347 :
348 : static void
349 : fd_bundle_client_step1( fd_bundle_tile_t * ctx,
350 0 : int * charge_busy ) {
351 0 : if( FD_UNLIKELY( ctx->tcp_sock>=0 &&
352 0 : !( ctx->tcp_sock_connected && fd_grpc_client_is_connected( ctx->grpc_client ) ) &&
353 0 : fd_bundle_now( ctx )>=ctx->conn_deadline ) ) {
354 0 : FD_LOG_INFO(( "Bundle connect to " FD_IP4_ADDR_FMT ":%hu timed out (%s)", FD_IP4_ADDR_FMT_ARGS( ctx->server_ip4_addr ), ctx->server_tcp_port,
355 0 : ctx->tcp_sock_connected ? "TLS/HTTP/2 setup" : "TCP connect" ));
356 0 : fd_bundle_client_reset( ctx );
357 0 : ctx->metrics.transport_fail_cnt++;
358 0 : *charge_busy = 1;
359 0 : return;
360 0 : }
361 :
362 : /* Wait for TCP socket to connect */
363 0 : if( FD_UNLIKELY( !ctx->tcp_sock_connected ) ) {
364 0 : if( FD_UNLIKELY( ctx->tcp_sock < 0 ) ) goto reconnect;
365 :
366 0 : struct pollfd pfds[1] = {
367 0 : { .fd = ctx->tcp_sock, .events = POLLOUT }
368 0 : };
369 0 : int poll_res = fd_syscall_poll( pfds, 1, 0 );
370 0 : if( FD_UNLIKELY( poll_res<0 ) ) {
371 0 : FD_LOG_ERR(( "fd_syscall_poll(tcp_sock) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
372 0 : }
373 0 : if( poll_res==0 ) return;
374 :
375 0 : int connect_result = 0;
376 0 : if( pfds[0].revents & (POLLERR|POLLHUP) ) {
377 0 : connect_result = fd_bundle_client_get_connect_result( ctx );
378 0 : connect_failed:
379 0 : FD_LOG_INFO(( "Bundle gRPC connect attempt failed (%i-%s)", connect_result, fd_io_strerror( connect_result ) ));
380 0 : fd_bundle_client_reset( ctx );
381 0 : ctx->metrics.transport_fail_cnt++;
382 0 : *charge_busy = 1;
383 0 : return;
384 0 : }
385 0 : if( FD_UNLIKELY( !(pfds[0].revents & POLLOUT) ) ) return;
386 0 : connect_result = fd_bundle_client_get_connect_result( ctx );
387 0 : if( FD_UNLIKELY( connect_result!=0 ) ) {
388 0 : goto connect_failed;
389 0 : }
390 0 : FD_LOG_DEBUG(( "Bundle TCP socket connected" ));
391 0 : ctx->tcp_sock_connected = 1;
392 0 : fd_bundle_client_epoll_out( ctx, 0 );
393 0 : *charge_busy = 1;
394 0 : }
395 :
396 : /* gRPC conn died? */
397 0 : if( FD_UNLIKELY( !ctx->grpc_client ) ) {
398 0 : long sleep_start;
399 0 : reconnect:
400 0 : sleep_start = fd_bundle_now( ctx );
401 0 : if( FD_UNLIKELY( fd_bundle_tile_should_stall( ctx, sleep_start ) ) ) {
402 0 : long wait_dur = ctx->backoff_until - sleep_start;
403 0 : fd_log_sleep( fd_long_min( wait_dur, 1e6 ) );
404 0 : return;
405 0 : }
406 0 : fd_bundle_client_create_conn( ctx );
407 0 : *charge_busy = 1;
408 0 : return;
409 0 : }
410 :
411 : /* Did a HTTP/2 PING time out */
412 0 : long check_ts = ctx->cached_ts = fd_bundle_now( ctx );
413 0 : if( FD_UNLIKELY( fd_keepalive_is_timeout( ctx->keepalive, check_ts ) ) ) {
414 0 : FD_LOG_WARNING(( "Bundle gRPC timed out (HTTP/2 PING went unanswered for %.2f seconds)",
415 0 : (double)( check_ts - ctx->keepalive->ts_last_tx )/1e9 ));
416 0 : ctx->keepalive->inflight = 0;
417 0 : ctx->defer_reset = 1;
418 0 : *charge_busy = 1;
419 0 : return;
420 0 : }
421 :
422 : /* Drive I/O, SSL handshake, and any inflight requests */
423 0 : if( FD_UNLIKELY( -1==fd_bundle_client_drive_io( ctx, check_ts, charge_busy ) || ctx->defer_reset /* new error? */ ) ) {
424 0 : fd_bundle_client_reset( ctx );
425 0 : ctx->metrics.transport_fail_cnt++;
426 0 : *charge_busy = 1;
427 0 : return;
428 0 : }
429 :
430 : /* Are we ready to issue a new request? */
431 0 : if( FD_UNLIKELY( fd_grpc_client_request_is_blocked( ctx->grpc_client ) ) ) return;
432 0 : long io_ts = fd_bundle_now( ctx );
433 0 : if( FD_UNLIKELY( fd_bundle_tile_should_stall( ctx, io_ts ) ) ) return;
434 :
435 0 : *charge_busy |= fd_bundle_client_step_reconnect( ctx, io_ts );
436 0 : }
437 :
438 : static void
439 0 : fd_bundle_client_log_status( fd_bundle_tile_t * ctx ) {
440 0 : int status = fd_bundle_client_status( ctx );
441 :
442 0 : int const connected_now = ( status==FD_BUNDLE_STATE_CONNECTED );
443 0 : int const connected_before = ( ctx->bundle_status_logged==FD_BUNDLE_STATE_CONNECTED );
444 :
445 0 : if( FD_UNLIKELY( connected_now!=connected_before ) ) {
446 0 : long ts = fd_log_wallclock();
447 0 : if( FD_LIKELY( ts-(ctx->last_bundle_status_log_nanos) >= (long)1e6 ) ) {
448 0 : if( connected_now ) FD_LOG_INFO(( "Connected to bundle server" ));
449 0 : else FD_LOG_INFO(( "Disconnected from bundle server" ));
450 0 : ctx->last_bundle_status_log_nanos = ts;
451 0 : ctx->bundle_status_logged = (uchar)status;
452 0 : }
453 0 : }
454 0 : }
455 :
456 : void
457 : fd_bundle_client_step( fd_bundle_tile_t * ctx,
458 0 : int * charge_busy ) {
459 : /* Edge-trigger logging with rate limiting */
460 0 : fd_bundle_client_step1( ctx, charge_busy );
461 0 : fd_bundle_client_log_status( ctx );
462 :
463 0 : if( FD_UNLIKELY( ctx->tcp_sock_connected && fd_grpc_client_tx_pending( ctx->grpc_client ) ) ) {
464 0 : int flush_err = ctx->is_ssl ? fd_grpc_client_tls_flush ( ctx->grpc_client, ctx->tcp_sock )
465 0 : : fd_grpc_client_tx_flush_socket( ctx->grpc_client, ctx->tcp_sock );
466 0 : if( FD_UNLIKELY( -1==flush_err ) ) {
467 0 : fd_bundle_client_reset( ctx );
468 0 : ctx->metrics.transport_fail_cnt++;
469 0 : *charge_busy = 1;
470 0 : return;
471 0 : }
472 0 : }
473 :
474 : /* Arm EPOLLOUT only on write demand: unsent HTTP/2 bytes, or a TLS
475 : handshake blocked on write (a permanently armed idle-writable
476 : socket would keep the epoll set ready forever). */
477 0 : if( FD_LIKELY( ctx->tcp_sock_connected && ctx->grpc_client ) ) {
478 0 : int want = fd_grpc_client_tx_pending( ctx->grpc_client );
479 0 : want |= ctx->is_ssl && fd_grpc_client_tls_tx_pending( ctx->grpc_client );
480 0 : fd_bundle_client_epoll_out( ctx, want );
481 0 : }
482 0 : }
483 :
484 : void
485 : fd_bundle_tile_backoff( fd_bundle_tile_t * ctx,
486 0 : long now ) {
487 0 : uint iter = ctx->backoff_iter;
488 0 : if( now < ctx->backoff_reset ) iter = 0U;
489 0 : iter++;
490 :
491 : /* FIXME proper backoff */
492 0 : long wait_ns = (long)2e9;
493 0 : wait_ns = (long)( fd_rng_ulong( ctx->rng ) & ( (1UL<<fd_ulong_find_msb_w_default( (ulong)wait_ns, 0 ))-1UL ) );
494 :
495 0 : ctx->backoff_until = now + wait_ns;
496 0 : ctx->backoff_reset = now + 2*wait_ns;
497 :
498 0 : ctx->backoff_iter = iter;
499 0 : }
500 :
501 : static void
502 0 : fd_bundle_client_grpc_conn_established( void * app_ctx ) {
503 0 : (void)app_ctx;
504 0 : FD_LOG_INFO(( "Bundle gRPC connection established" ));
505 0 : }
506 :
507 : static void
508 : fd_bundle_client_grpc_conn_dead( void * app_ctx,
509 : uint h2_err,
510 0 : int closed_by ) {
511 0 : fd_bundle_tile_t * ctx = app_ctx;
512 0 : FD_LOG_INFO(( "Bundle gRPC connection closed %s (%u-%s)",
513 0 : closed_by ? "by peer" : "due to error",
514 0 : h2_err, fd_h2_strerror( h2_err ) ));
515 0 : ctx->defer_reset = 1;
516 0 : }
517 :
518 : /* Publishes a transaction to the verify tiles. Assumes the caller
519 : checked that a credit is available. bundle_txn_cnt is zero for a
520 : standalone packet. */
521 :
522 : static void
523 : fd_bundle_tile_publish_txn( fd_bundle_tile_t * ctx,
524 : void const * txn,
525 : ulong txn_sz, /* <=FD_TXN_MTU */
526 : uint source_ipv4,
527 0 : ulong bundle_txn_cnt ) {
528 0 : int is_bundle = bundle_txn_cnt>0UL;
529 0 : fd_txn_m_t * txnm = fd_chunk_to_laddr( ctx->verify_out.mem, ctx->verify_out.chunk );
530 0 : *txnm = (fd_txn_m_t) {
531 0 : .reference_block_height = 0UL,
532 0 : .payload_sz = (ushort)txn_sz,
533 0 : .txn_t_sz = 0U,
534 0 : .source_ipv4 = source_ipv4,
535 0 : .source_tpu = FD_TXN_M_TPU_SOURCE_BUNDLE,
536 0 : .first_seen_nanos = fd_bundle_now( ctx ),
537 0 : .block_engine = {
538 0 : .bundle_id = is_bundle ? ctx->bundle_seq : 0UL,
539 0 : .bundle_txn_cnt = is_bundle ? bundle_txn_cnt : 1UL,
540 0 : .commission = is_bundle ? (uchar)ctx->builder_commission : (uchar)0,
541 0 : },
542 0 : };
543 0 : if( is_bundle ) fd_memcpy( txnm->block_engine.commission_pubkey, ctx->builder_pubkey, 32UL );
544 0 : else fd_memset( txnm->block_engine.commission_pubkey, 0, 32UL );
545 0 : fd_memcpy( fd_txn_m_payload( txnm ), txn, txn_sz );
546 :
547 0 : ulong sz = fd_txn_m_realized_footprint( txnm, 0, 0 );
548 0 : ulong tspub = (ulong)fd_frag_meta_ts_comp( fd_bundle_now( ctx ) );
549 0 : fd_stem_publish( ctx->stem, ctx->verify_out.idx, (ulong)is_bundle, ctx->verify_out.chunk, sz, 0UL, 0UL, tspub );
550 0 : ctx->verify_out.chunk = fd_dcache_compact_next( ctx->verify_out.chunk, sz, ctx->verify_out.chunk0, ctx->verify_out.wmark );
551 0 : ctx->metrics.txn_received_cnt++;
552 0 : }
553 :
554 : static inline ulong
555 0 : fd_bundle_tile_cr_avail( fd_bundle_tile_t const * ctx ) {
556 0 : return ctx->stem->cr_avail[ ctx->verify_out.idx ];
557 0 : }
558 :
559 : /* fd_bundle_packet_t holds the fields of a packet.Packet message used
560 : by the bundle client. */
561 :
562 : struct fd_bundle_packet {
563 : uchar const * data;
564 : ulong data_sz;
565 : int has_meta;
566 : char addr[ 256 ]; /* packet.Meta.addr, NUL terminated */
567 : };
568 :
569 : typedef struct fd_bundle_packet fd_bundle_packet_t;
570 :
571 : /* fd_bundle_client_decode_packet decodes a packet.Packet message.
572 : Returns packet on success and NULL on decode failure. */
573 :
574 : static fd_bundle_packet_t *
575 : fd_bundle_client_decode_packet( fd_pb_inbuf_t in,
576 0 : fd_bundle_packet_t * packet ) {
577 0 : packet->data = NULL;
578 0 : packet->data_sz = 0UL;
579 0 : packet->has_meta = 0;
580 0 : packet->addr[0] = '\0';
581 0 : while( fd_pb_inbuf_sz( &in ) ) {
582 0 : fd_pb_tlv_t tlv[1];
583 0 : if( FD_UNLIKELY( !fd_pb_tlv_read( &in, tlv ) ) ) return NULL;
584 0 : switch( tlv->field_id ) {
585 0 : case 1: /* data */
586 0 : packet->data = fd_pb_tlv_bytes( &in, tlv );
587 0 : if( FD_UNLIKELY( !packet->data ) ) return NULL;
588 0 : packet->data_sz = tlv->len;
589 0 : break;
590 0 : case 2: { /* meta */
591 0 : fd_pb_inbuf_t meta[1];
592 0 : if( FD_UNLIKELY( !fd_pb_tlv_submsg( &in, tlv, meta ) ) ) return NULL;
593 0 : packet->has_meta = 1;
594 0 : while( fd_pb_inbuf_sz( meta ) ) {
595 0 : fd_pb_tlv_t meta_tlv[1];
596 0 : if( FD_UNLIKELY( !fd_pb_tlv_read( meta, meta_tlv ) ) ) return NULL;
597 0 : if( meta_tlv->field_id==2U ) { /* addr */
598 0 : if( FD_UNLIKELY( !fd_pb_tlv_cstr( meta, meta_tlv, packet->addr, sizeof(packet->addr) ) ) ) return NULL;
599 0 : }
600 0 : if( FD_UNLIKELY( !fd_pb_tlv_skip( meta, meta_tlv ) ) ) return NULL;
601 0 : }
602 0 : break;
603 0 : }
604 0 : default:
605 0 : break;
606 0 : }
607 0 : if( FD_UNLIKELY( !fd_pb_tlv_skip( &in, tlv ) ) ) return NULL;
608 0 : }
609 0 : return packet;
610 0 : }
611 :
612 : /* fd_bundle_client_decode_header_ts decodes the timestamp of a
613 : shared.Header message. Returns the timestamp in nanoseconds since
614 : the Unix epoch on success (0 if absent). Returns -1 on decode
615 : failure. */
616 :
617 : static long
618 0 : fd_bundle_client_decode_header_ts( fd_pb_inbuf_t in ) {
619 0 : ulong seconds = 0UL;
620 0 : ulong nanos = 0UL;
621 0 : while( fd_pb_inbuf_sz( &in ) ) {
622 0 : fd_pb_tlv_t tlv[1];
623 0 : if( FD_UNLIKELY( !fd_pb_tlv_read( &in, tlv ) ) ) return -1L;
624 0 : if( tlv->field_id==1U ) { /* ts */
625 0 : fd_pb_inbuf_t ts[1];
626 0 : if( FD_UNLIKELY( !fd_pb_tlv_submsg( &in, tlv, ts ) ) ) return -1L;
627 0 : while( fd_pb_inbuf_sz( ts ) ) {
628 0 : fd_pb_tlv_t ts_tlv[1];
629 0 : if( FD_UNLIKELY( !fd_pb_tlv_read( ts, ts_tlv ) ) ) return -1L;
630 0 : switch( ts_tlv->field_id ) {
631 0 : case 1: /* seconds */
632 0 : if( FD_UNLIKELY( ts_tlv->wire_type!=FD_PB_WIRE_TYPE_VARINT ) ) return -1L;
633 0 : seconds = ts_tlv->varint;
634 0 : break;
635 0 : case 2: /* nanos */
636 0 : if( FD_UNLIKELY( ts_tlv->wire_type!=FD_PB_WIRE_TYPE_VARINT ) ) return -1L;
637 0 : nanos = (ulong)(uint)ts_tlv->varint;
638 0 : break;
639 0 : default:
640 0 : break;
641 0 : }
642 0 : if( FD_UNLIKELY( !fd_pb_tlv_skip( ts, ts_tlv ) ) ) return -1L;
643 0 : }
644 0 : }
645 0 : if( FD_UNLIKELY( !fd_pb_tlv_skip( &in, tlv ) ) ) return -1L;
646 0 : }
647 0 : return (long)( seconds*(ulong)1e9 + nanos );
648 0 : }
649 :
650 : static void
651 : fd_bundle_client_sample_rx_delay( fd_bundle_tile_t * ctx,
652 0 : long ts ) {
653 0 : fd_histf_sample( ctx->metrics.msg_rx_delay, fd_ulong_sat_sub( (ulong)ctx->cached_ts, (ulong)ts ) );
654 0 : }
655 :
656 : static uint
657 : fd_bundle_packet_ip4( fd_bundle_packet_t const * packet,
658 0 : uint dflt ) {
659 0 : uint ip4;
660 0 : if( packet->has_meta && fd_cstr_to_ip4_addr( packet->addr, &ip4 ) ) return ip4;
661 0 : return dflt;
662 0 : }
663 :
664 : /* Handles a bundle.Bundle message. Returns 1 on success and 0 on
665 : decode failure. */
666 :
667 : static int
668 : fd_bundle_client_handle_bundle( fd_bundle_tile_t * ctx,
669 0 : fd_pb_inbuf_t in ) {
670 0 : fd_bundle_packet_t packets[ FD_BUNDLE_CLIENT_MAX_TXN_PER_BUNDLE ];
671 0 : ulong cnt = 0UL;
672 0 : long ts = 0L;
673 :
674 0 : while( fd_pb_inbuf_sz( &in ) ) {
675 0 : fd_pb_tlv_t tlv[1];
676 0 : if( FD_UNLIKELY( !fd_pb_tlv_read( &in, tlv ) ) ) return 0;
677 0 : switch( tlv->field_id ) {
678 0 : case 2: { /* header */
679 0 : fd_pb_inbuf_t header[1];
680 0 : if( FD_UNLIKELY( !fd_pb_tlv_submsg( &in, tlv, header ) ) ) return 0;
681 0 : ts = fd_bundle_client_decode_header_ts( *header );
682 0 : if( FD_UNLIKELY( ts<0L ) ) return 0;
683 0 : break;
684 0 : }
685 0 : case 3: { /* packets */
686 0 : fd_pb_inbuf_t sub[1];
687 0 : fd_bundle_packet_t packet[1];
688 0 : if( FD_UNLIKELY( !fd_pb_tlv_submsg( &in, tlv, sub ) ) ) return 0;
689 0 : if( FD_UNLIKELY( !fd_bundle_client_decode_packet( *sub, packet ) ) ) return 0;
690 0 : if( FD_UNLIKELY( packet->data_sz==0UL || packet->data_sz>FD_TXN_MTU ) ) {
691 0 : FD_LOG_WARNING(( "Bundle server delivered a %lu byte packet, dropping bundle", packet->data_sz ));
692 0 : return 1;
693 0 : }
694 0 : if( FD_UNLIKELY( cnt>=FD_BUNDLE_CLIENT_MAX_TXN_PER_BUNDLE ) ) return 1;
695 0 : packets[ cnt++ ] = *packet;
696 0 : break;
697 0 : }
698 0 : default:
699 0 : break;
700 0 : }
701 0 : if( FD_UNLIKELY( !fd_pb_tlv_skip( &in, tlv ) ) ) return 0;
702 0 : }
703 :
704 0 : if( FD_UNLIKELY( fd_bundle_tile_cr_avail( ctx )<cnt ) ) {
705 0 : ctx->metrics.backpressure_drop_cnt += cnt;
706 0 : return 1;
707 0 : }
708 :
709 0 : ctx->bundle_seq++;
710 0 : ctx->metrics.bundle_received_cnt++;
711 0 : for( ulong i=0UL; i<cnt; i++ ) {
712 0 : uint ip4 = fd_bundle_packet_ip4( &packets[ i ], ctx->server_ip4_addr );
713 0 : fd_bundle_tile_publish_txn( ctx, packets[ i ].data, packets[ i ].data_sz, ip4, cnt );
714 0 : }
715 0 : fd_bundle_client_sample_rx_delay( ctx, ts );
716 0 : return 1;
717 0 : }
718 :
719 : /* Handle a SubscribeBundlesResponse from a SubscribeBundles gRPC call. */
720 :
721 : static void
722 : fd_bundle_client_handle_bundle_batch(
723 : fd_bundle_tile_t * ctx,
724 : fd_pb_inbuf_t in
725 0 : ) {
726 0 : if( FD_UNLIKELY( !ctx->builder_info_avail ) ) {
727 0 : ctx->metrics.missing_builder_info_fail_cnt++; /* unreachable */
728 0 : return;
729 0 : }
730 :
731 0 : while( fd_pb_inbuf_sz( &in ) ) {
732 0 : fd_pb_tlv_t tlv[1];
733 0 : if( FD_UNLIKELY( !fd_pb_tlv_read( &in, tlv ) ) ) goto fail;
734 0 : if( tlv->field_id==1U ) { /* bundles (BundleUuid) */
735 0 : fd_pb_inbuf_t uuid[1];
736 0 : if( FD_UNLIKELY( !fd_pb_tlv_submsg( &in, tlv, uuid ) ) ) goto fail;
737 0 : fd_pb_inbuf_t bundle[1] = {{0}};
738 0 : while( fd_pb_inbuf_sz( uuid ) ) {
739 0 : fd_pb_tlv_t uuid_tlv[1];
740 0 : if( FD_UNLIKELY( !fd_pb_tlv_read( uuid, uuid_tlv ) ) ) goto fail;
741 0 : if( uuid_tlv->field_id==1U ) { /* bundle */
742 0 : if( FD_UNLIKELY( !fd_pb_tlv_submsg( uuid, uuid_tlv, bundle ) ) ) goto fail;
743 0 : }
744 0 : if( FD_UNLIKELY( !fd_pb_tlv_skip( uuid, uuid_tlv ) ) ) goto fail;
745 0 : }
746 0 : if( FD_UNLIKELY( !fd_bundle_client_handle_bundle( ctx, *bundle ) ) ) goto fail;
747 0 : }
748 0 : if( FD_UNLIKELY( !fd_pb_tlv_skip( &in, tlv ) ) ) goto fail;
749 0 : }
750 0 : return;
751 :
752 0 : fail:
753 0 : ctx->metrics.decode_fail_cnt++;
754 0 : FD_LOG_WARNING(( "Protobuf decode of (block_engine.SubscribeBundlesResponse) failed" ));
755 0 : }
756 :
757 : /* Called for each 'Packet' (a regular transaction) of a
758 : SubscribePacketsResponse. */
759 :
760 : static void
761 : fd_bundle_client_handle_packet( fd_bundle_tile_t * ctx,
762 0 : fd_bundle_packet_t const * packet ) {
763 0 : if( FD_UNLIKELY( packet->data_sz==0UL || packet->data_sz>FD_TXN_MTU ) ) {
764 0 : FD_LOG_WARNING(( "Bundle server delivered a %lu byte packet, ignoring", packet->data_sz ));
765 0 : return;
766 0 : }
767 0 : if( FD_UNLIKELY( !fd_bundle_tile_cr_avail( ctx ) ) ) {
768 0 : ctx->metrics.backpressure_drop_cnt++;
769 0 : return;
770 0 : }
771 0 : fd_bundle_tile_publish_txn( ctx, packet->data, packet->data_sz, fd_bundle_packet_ip4( packet, 0U ), 0UL );
772 0 : ctx->metrics.packet_received_cnt++;
773 0 : }
774 :
775 : /* Handle a SubscribePacketsResponse from a SubscribePackets gRPC call. */
776 :
777 : static void
778 : fd_bundle_client_handle_packet_batch(
779 : fd_bundle_tile_t * ctx,
780 : fd_pb_inbuf_t in
781 0 : ) {
782 0 : long ts = 0L;
783 0 : while( fd_pb_inbuf_sz( &in ) ) {
784 0 : fd_pb_tlv_t tlv[1];
785 0 : if( FD_UNLIKELY( !fd_pb_tlv_read( &in, tlv ) ) ) goto fail;
786 0 : switch( tlv->field_id ) {
787 0 : case 1: { /* header */
788 0 : fd_pb_inbuf_t header[1];
789 0 : if( FD_UNLIKELY( !fd_pb_tlv_submsg( &in, tlv, header ) ) ) goto fail;
790 0 : ts = fd_bundle_client_decode_header_ts( *header );
791 0 : if( FD_UNLIKELY( ts<0L ) ) goto fail;
792 0 : break;
793 0 : }
794 0 : case 2: { /* batch (PacketBatch) */
795 0 : fd_pb_inbuf_t batch[1];
796 0 : if( FD_UNLIKELY( !fd_pb_tlv_submsg( &in, tlv, batch ) ) ) goto fail;
797 0 : while( fd_pb_inbuf_sz( batch ) ) {
798 0 : fd_pb_tlv_t batch_tlv[1];
799 0 : if( FD_UNLIKELY( !fd_pb_tlv_read( batch, batch_tlv ) ) ) goto fail;
800 0 : if( batch_tlv->field_id==1U ) { /* packets */
801 0 : fd_pb_inbuf_t sub[1];
802 0 : fd_bundle_packet_t packet[1];
803 0 : if( FD_UNLIKELY( !fd_pb_tlv_submsg( batch, batch_tlv, sub ) ) ) goto fail;
804 0 : if( FD_UNLIKELY( !fd_bundle_client_decode_packet( *sub, packet ) ) ) goto fail;
805 0 : fd_bundle_client_handle_packet( ctx, packet );
806 0 : }
807 0 : if( FD_UNLIKELY( !fd_pb_tlv_skip( batch, batch_tlv ) ) ) goto fail;
808 0 : }
809 0 : break;
810 0 : }
811 0 : default:
812 0 : break;
813 0 : }
814 0 : if( FD_UNLIKELY( !fd_pb_tlv_skip( &in, tlv ) ) ) goto fail;
815 0 : }
816 :
817 0 : fd_bundle_client_sample_rx_delay( ctx, ts );
818 0 : return;
819 :
820 0 : fail:
821 0 : ctx->metrics.decode_fail_cnt++;
822 0 : FD_LOG_WARNING(( "Protobuf decode of (block_engine.SubscribePacketsResponse) failed" ));
823 0 : }
824 :
825 : /* Handle a BlockBuilderFeeInfoResponse from a GetBlockBuilderFeeInfo
826 : gRPC call. */
827 :
828 : static void
829 : fd_bundle_client_handle_builder_fee_info(
830 : fd_bundle_tile_t * ctx,
831 : fd_pb_inbuf_t in
832 0 : ) {
833 0 : char pubkey[ FD_BASE58_ENCODED_32_SZ ] = {0};
834 0 : ulong commission = 0UL;
835 0 : while( fd_pb_inbuf_sz( &in ) ) {
836 0 : fd_pb_tlv_t tlv[1];
837 0 : if( FD_UNLIKELY( !fd_pb_tlv_read( &in, tlv ) ) ) goto decode_fail;
838 0 : switch( tlv->field_id ) {
839 0 : case 1: /* pubkey */
840 0 : if( FD_UNLIKELY( !fd_pb_tlv_cstr( &in, tlv, pubkey, sizeof(pubkey) ) ) ) goto decode_fail;
841 0 : break;
842 0 : case 2: /* commission */
843 0 : if( FD_UNLIKELY( tlv->wire_type!=FD_PB_WIRE_TYPE_VARINT ) ) goto decode_fail;
844 0 : commission = tlv->varint;
845 0 : break;
846 0 : default:
847 0 : break;
848 0 : }
849 0 : if( FD_UNLIKELY( !fd_pb_tlv_skip( &in, tlv ) ) ) goto decode_fail;
850 0 : }
851 0 : if( FD_UNLIKELY( commission > 100 ) ) {
852 0 : ctx->metrics.decode_fail_cnt++;
853 0 : FD_LOG_WARNING(( "BlockBuilderFeeInfoResponse commission out of range (0-100): %lu", commission ));
854 0 : return;
855 0 : }
856 :
857 0 : fd_pubkey_t decoded_builder_pubkey;
858 0 : if( FD_UNLIKELY( !fd_base58_decode_32( pubkey, decoded_builder_pubkey.uc ) ||
859 0 : fd_pubkey_check_zero( &decoded_builder_pubkey ) ) ) {
860 0 : FD_LOG_HEXDUMP_WARNING(( "Invalid pubkey in BlockBuilderFeeInfoResponse", pubkey, strnlen( pubkey, sizeof(pubkey) ) ));
861 0 : return;
862 0 : }
863 :
864 0 : ctx->builder_commission = (uchar)commission; /* Apply update atomically */
865 0 : fd_memcpy( ctx->builder_pubkey, decoded_builder_pubkey.uc, sizeof(ctx->builder_pubkey) );
866 :
867 0 : long validity_duration_ns = (long)( 60e9 * 5. ); /* 5 minutes */
868 0 : ctx->builder_info_avail = 1;
869 0 : ctx->builder_info_valid_until = fd_bundle_now( ctx ) + validity_duration_ns;
870 0 : return;
871 :
872 0 : decode_fail:
873 0 : ctx->metrics.decode_fail_cnt++;
874 0 : FD_LOG_WARNING(( "Protobuf decode of (block_engine.BlockBuilderFeeInfoResponse) failed" ));
875 0 : }
876 :
877 : static void
878 : fd_bundle_client_grpc_tx_complete(
879 : void * app_ctx,
880 : ulong request_ctx
881 0 : ) {
882 0 : (void)app_ctx; (void)request_ctx;
883 0 : }
884 :
885 : void
886 : fd_bundle_client_grpc_rx_start(
887 : void * app_ctx,
888 : ulong request_ctx
889 0 : ) {
890 0 : fd_bundle_tile_t * ctx = app_ctx;
891 0 : switch( request_ctx ) {
892 0 : case FD_BUNDLE_CLIENT_REQ_Bundle_SubscribePackets:
893 0 : ctx->packet_subscription_live = 1;
894 0 : ctx->packet_subscription_wait = 0;
895 0 : break;
896 0 : case FD_BUNDLE_CLIENT_REQ_Bundle_SubscribeBundles:
897 0 : ctx->bundle_subscription_live = 1;
898 0 : ctx->bundle_subscription_wait = 0;
899 0 : break;
900 0 : }
901 0 : }
902 :
903 : void
904 : fd_bundle_client_grpc_rx_msg(
905 : void * app_ctx,
906 : void const * protobuf,
907 : ulong protobuf_sz,
908 : ulong request_ctx
909 0 : ) {
910 0 : fd_bundle_tile_t * ctx = app_ctx;
911 0 : ctx->metrics.proto_received_bytes += protobuf_sz;
912 0 : fd_pb_inbuf_t in[1]; fd_pb_inbuf_init( in, protobuf, protobuf_sz );
913 0 : switch( request_ctx ) {
914 0 : case FD_BUNDLE_CLIENT_REQ_Auth_GenerateAuthChallenge:
915 0 : if( FD_UNLIKELY( !fd_bundle_auther_handle_challenge_resp( &ctx->auther, protobuf, protobuf_sz ) ) ) {
916 0 : ctx->metrics.decode_fail_cnt++;
917 0 : fd_bundle_tile_backoff( ctx, fd_bundle_now( ctx ) );
918 0 : }
919 0 : break;
920 0 : case FD_BUNDLE_CLIENT_REQ_Auth_GenerateAuthTokens:
921 0 : if( FD_UNLIKELY( !fd_bundle_auther_handle_tokens_resp( &ctx->auther, protobuf, protobuf_sz ) ) ) {
922 0 : ctx->metrics.decode_fail_cnt++;
923 0 : fd_bundle_tile_backoff( ctx, fd_bundle_now( ctx ) );
924 0 : }
925 0 : break;
926 0 : case FD_BUNDLE_CLIENT_REQ_Bundle_SubscribeBundles:
927 0 : fd_bundle_client_handle_bundle_batch( ctx, *in );
928 0 : break;
929 0 : case FD_BUNDLE_CLIENT_REQ_Bundle_SubscribePackets:
930 0 : fd_bundle_client_handle_packet_batch( ctx, *in );
931 0 : break;
932 0 : case FD_BUNDLE_CLIENT_REQ_Bundle_GetBlockBuilderFeeInfo:
933 0 : fd_bundle_client_handle_builder_fee_info( ctx, *in );
934 0 : break;
935 0 : default:
936 0 : FD_LOG_ERR(( "Received unexpected gRPC message (request_ctx=%lu)", request_ctx ));
937 0 : }
938 0 : }
939 :
940 : static void
941 : fd_bundle_client_request_failed( fd_bundle_tile_t * ctx,
942 0 : ulong request_ctx ) {
943 0 : fd_bundle_tile_backoff( ctx, fd_bundle_now( ctx ) );
944 0 : switch( request_ctx ) {
945 0 : case FD_BUNDLE_CLIENT_REQ_Auth_GenerateAuthChallenge:
946 0 : case FD_BUNDLE_CLIENT_REQ_Auth_GenerateAuthTokens:
947 0 : fd_bundle_auther_handle_request_fail( &ctx->auther );
948 0 : break;
949 0 : case FD_BUNDLE_CLIENT_REQ_Bundle_GetBlockBuilderFeeInfo:
950 0 : ctx->builder_info_wait = 0;
951 0 : break;
952 0 : case FD_BUNDLE_CLIENT_REQ_Bundle_SubscribePackets:
953 0 : ctx->packet_subscription_live = 0;
954 0 : ctx->packet_subscription_wait = 0;
955 0 : break;
956 0 : case FD_BUNDLE_CLIENT_REQ_Bundle_SubscribeBundles:
957 0 : ctx->bundle_subscription_live = 0;
958 0 : ctx->bundle_subscription_wait = 0;
959 0 : break;
960 0 : }
961 0 : }
962 :
963 : void
964 : fd_bundle_client_grpc_rx_end(
965 : void * app_ctx,
966 : ulong request_ctx,
967 : fd_grpc_resp_hdrs_t * resp
968 0 : ) {
969 0 : fd_bundle_tile_t * ctx = app_ctx;
970 0 : if( FD_UNLIKELY( resp->h2_status!=200 ) ) {
971 0 : FD_LOG_WARNING(( "gRPC request failed (HTTP status %u)", resp->h2_status ));
972 0 : fd_bundle_client_request_failed( ctx, request_ctx );
973 0 : fd_bundle_auther_reset( &ctx->auther );
974 0 : ctx->defer_reset = 1;
975 0 : return;
976 0 : }
977 :
978 0 : resp->grpc_msg_len = (uint)fd_url_unescape( resp->grpc_msg, resp->grpc_msg_len );
979 0 : if( !resp->grpc_msg_len ) {
980 0 : fd_memcpy( resp->grpc_msg, "unknown error", 13 );
981 0 : resp->grpc_msg_len = 13;
982 0 : }
983 :
984 0 : switch( request_ctx ) {
985 0 : case FD_BUNDLE_CLIENT_REQ_Bundle_SubscribePackets:
986 0 : ctx->packet_subscription_live = 0;
987 0 : ctx->packet_subscription_wait = 0;
988 0 : fd_bundle_tile_backoff( ctx, fd_bundle_now( ctx ) );
989 0 : ctx->defer_reset = 1;
990 0 : FD_LOG_INFO(( "SubscribePackets stream failed (gRPC status %u-%s). Reconnecting ...",
991 0 : resp->grpc_status, fd_grpc_status_cstr( resp->grpc_status ) ));
992 0 : return;
993 0 : case FD_BUNDLE_CLIENT_REQ_Bundle_SubscribeBundles:
994 0 : ctx->bundle_subscription_live = 0;
995 0 : ctx->bundle_subscription_wait = 0;
996 0 : fd_bundle_tile_backoff( ctx, fd_bundle_now( ctx ) );
997 0 : ctx->defer_reset = 1;
998 0 : FD_LOG_INFO(( "SubscribeBundles stream failed (gRPC status %u-%s). Reconnecting ...",
999 0 : resp->grpc_status, fd_grpc_status_cstr( resp->grpc_status ) ));
1000 0 : return;
1001 0 : case FD_BUNDLE_CLIENT_REQ_Bundle_GetBlockBuilderFeeInfo:
1002 0 : ctx->builder_info_wait = 0;
1003 0 : break;
1004 0 : default:
1005 0 : break;
1006 0 : }
1007 :
1008 0 : if( FD_UNLIKELY( resp->grpc_status!=FD_GRPC_STATUS_OK ) ) {
1009 0 : FD_LOG_INFO(( "gRPC request failed (gRPC status %u-%s): %.*s",
1010 0 : resp->grpc_status, fd_grpc_status_cstr( resp->grpc_status ),
1011 0 : (int)resp->grpc_msg_len, resp->grpc_msg ));
1012 0 : fd_bundle_client_request_failed( ctx, request_ctx );
1013 0 : if( resp->grpc_status==FD_GRPC_STATUS_UNAUTHENTICATED ||
1014 0 : resp->grpc_status==FD_GRPC_STATUS_PERMISSION_DENIED ) {
1015 0 : fd_bundle_auther_reset( &ctx->auther );
1016 0 : }
1017 0 : return;
1018 0 : }
1019 0 : }
1020 :
1021 : void
1022 : fd_bundle_client_grpc_rx_timeout(
1023 : void * app_ctx,
1024 : ulong request_ctx, /* FD_BUNDLE_CLIENT_REQ_{...} */
1025 : int deadline_kind /* FD_GRPC_DEADLINE_{HEADER|RX_END} */
1026 0 : ) {
1027 0 : (void)deadline_kind;
1028 0 : FD_LOG_WARNING(( "Request timed out: %s", fd_bundle_request_ctx_cstr( request_ctx ) ));
1029 0 : fd_bundle_tile_t * ctx = app_ctx;
1030 0 : ctx->defer_reset = 1;
1031 0 : }
1032 :
1033 : static void
1034 0 : fd_bundle_client_grpc_ping_ack( void * app_ctx ) {
1035 0 : fd_bundle_tile_t * ctx = app_ctx;
1036 0 : long rtt_sample = fd_keepalive_rx( ctx->keepalive, fd_bundle_now( ctx ) );
1037 0 : if( FD_LIKELY( rtt_sample ) ) {
1038 0 : fd_rtt_sample( ctx->rtt, (float)rtt_sample, 0 );
1039 0 : FD_LOG_DEBUG(( "Keepalive ACK" ));
1040 0 : }
1041 0 : ctx->metrics.ping_ack_cnt++;
1042 0 : }
1043 :
1044 : fd_grpc_client_callbacks_t fd_bundle_client_grpc_callbacks = {
1045 : .conn_established = fd_bundle_client_grpc_conn_established,
1046 : .conn_dead = fd_bundle_client_grpc_conn_dead,
1047 : .tx_complete = fd_bundle_client_grpc_tx_complete,
1048 : .rx_start = fd_bundle_client_grpc_rx_start,
1049 : .rx_msg = fd_bundle_client_grpc_rx_msg,
1050 : .rx_end = fd_bundle_client_grpc_rx_end,
1051 : .rx_timeout = fd_bundle_client_grpc_rx_timeout,
1052 : .ping_ack = fd_bundle_client_grpc_ping_ack,
1053 : };
1054 :
1055 : int
1056 0 : fd_bundle_client_status( fd_bundle_tile_t const * ctx ) {
1057 0 : if( FD_UNLIKELY( ( !ctx->tcp_sock_connected ) |
1058 0 : ( !ctx->grpc_client ) ) ) {
1059 0 : return FD_BUNDLE_STATE_DISCONNECTED;
1060 0 : }
1061 :
1062 0 : fd_h2_conn_t * conn = fd_grpc_client_h2_conn( ctx->grpc_client );
1063 0 : if( FD_UNLIKELY( !conn ) ) {
1064 0 : return FD_BUNDLE_STATE_DISCONNECTED; /* no conn */
1065 0 : }
1066 0 : if( FD_UNLIKELY( conn->flags &
1067 0 : ( FD_H2_CONN_FLAGS_DEAD |
1068 0 : FD_H2_CONN_FLAGS_SEND_GOAWAY ) ) ) {
1069 0 : return FD_BUNDLE_STATE_DISCONNECTED;
1070 0 : }
1071 :
1072 0 : if( FD_UNLIKELY( conn->flags &
1073 0 : ( FD_H2_CONN_FLAGS_CLIENT_INITIAL |
1074 0 : FD_H2_CONN_FLAGS_WAIT_SETTINGS_ACK_0 |
1075 0 : FD_H2_CONN_FLAGS_WAIT_SETTINGS_0 |
1076 0 : FD_H2_CONN_FLAGS_SERVER_INITIAL ) ) ) {
1077 0 : return FD_BUNDLE_STATE_CONNECTING; /* connection is not ready */
1078 0 : }
1079 :
1080 0 : if( FD_UNLIKELY( ctx->auther.state != FD_BUNDLE_AUTH_STATE_DONE_WAIT ) ) {
1081 0 : return FD_BUNDLE_STATE_CONNECTING; /* not authenticated */
1082 0 : }
1083 :
1084 0 : if( FD_UNLIKELY( ( !ctx->builder_info_avail ) |
1085 0 : ( !ctx->packet_subscription_live ) |
1086 0 : ( !ctx->bundle_subscription_live ) ) ) {
1087 0 : return FD_BUNDLE_STATE_CONNECTING; /* not fully connected */
1088 0 : }
1089 :
1090 0 : if( FD_UNLIKELY( fd_keepalive_is_timeout( ctx->keepalive, fd_bundle_now( ctx ) ) ) ) {
1091 0 : return FD_BUNDLE_STATE_DISCONNECTED; /* possible timeout */
1092 0 : }
1093 :
1094 0 : if( FD_UNLIKELY( !fd_grpc_client_is_connected( ctx->grpc_client ) ) ) {
1095 0 : return FD_BUNDLE_STATE_CONNECTING;
1096 0 : }
1097 :
1098 : /* As far as we know, the bundle connection is alive and well. */
1099 0 : return FD_BUNDLE_STATE_CONNECTED;
1100 0 : }
1101 :
1102 : #undef DISCONNECTED
1103 : #undef CONNECTING
1104 : #undef CONNECTED
1105 :
1106 : FD_FN_CONST char const *
1107 0 : fd_bundle_request_ctx_cstr( ulong request_ctx ) {
1108 0 : switch( request_ctx ) {
1109 0 : case FD_BUNDLE_CLIENT_REQ_Auth_GenerateAuthChallenge:
1110 0 : return "GenerateAuthChallenge";
1111 0 : case FD_BUNDLE_CLIENT_REQ_Auth_GenerateAuthTokens:
1112 0 : return "GenerateAuthTokens";
1113 0 : case FD_BUNDLE_CLIENT_REQ_Bundle_SubscribePackets:
1114 0 : return "SubscribePackets";
1115 0 : case FD_BUNDLE_CLIENT_REQ_Bundle_SubscribeBundles:
1116 0 : return "SubscribeBundles";
1117 0 : case FD_BUNDLE_CLIENT_REQ_Bundle_GetBlockBuilderFeeInfo:
1118 0 : return "GetBlockBuilderFeeInfo";
1119 0 : default:
1120 0 : return "unknown";
1121 0 : }
1122 0 : }
|
