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