Line data Source code
1 : #define _GNU_SOURCE
2 : #include "fd_event_client.h"
3 :
4 : #include "../../waltz/resolv/fd_netdb.h"
5 : #include "../../waltz/http/fd_url.h"
6 : #include "../../waltz/grpc/fd_grpc_client.h"
7 : #include "../../waltz/grpc/fd_grpc_client_private.h"
8 : #include "../../ballet/pb/fd_pb_tokenize.h"
9 : #include "../../ballet/pb/fd_pb_encode.h"
10 : #include "../../ballet/hex/fd_hex.h"
11 : #include "../../util/net/fd_ip4.h"
12 : #include "../../util/log/fd_log.h"
13 : #include "../keyguard/fd_keyguard.h"
14 :
15 : #include "../../waltz/tlsrec/fd_tlsrec.h"
16 : #include "../../ballet/ed25519/fd_x25519.h"
17 :
18 : #define ZSTD_STATIC_LINKING_ONLY
19 : #include <zstd.h>
20 :
21 : #include <netinet/tcp.h>
22 : #include <unistd.h>
23 : #include <errno.h>
24 : #include <sys/epoll.h>
25 : #include <sys/socket.h>
26 : #include <netinet/in.h>
27 :
28 0 : #define DISCONNECT_REASON_IDENTITY_CHANGED (0)
29 0 : #define DISCONNECT_REASON_CONNECT_FAILED (1)
30 0 : #define DISCONNECT_REASON_DNS_RESOLVE_FAILED (2)
31 0 : #define DISCONNECT_REASON_TIMEOUT (3)
32 0 : #define DISCONNECT_REASON_TRANSPORT_FAILED (4)
33 0 : #define DISCONNECT_REASON_PEER_CLOSED (5)
34 0 : #define DISCONNECT_REASON_INVALID_CURSOR (6)
35 0 : #define DISCONNECT_REASON_AUTH_FAILED (7)
36 0 : #define DISCONNECT_REASON_INVALID_PROTOBUF (8)
37 :
38 0 : #define FD_EVENT_CLIENT_REQ_CTX_AUTHENTICATE (1UL)
39 0 : #define FD_EVENT_CLIENT_REQ_CTX_STREAM_EVENTS (2UL) /* low byte; stream generation above */
40 :
41 0 : #define FD_EVENT_CLIENT_HEARTBEAT_NANOS (15L*(long)1e9)
42 0 : #define FD_EVENT_CLIENT_RESPONSE_TIMEOUT_NANOS (60L*(long)1e9)
43 :
44 0 : #define FD_EVENT_CLIENT_TX_BURST (256L<<10)
45 0 : #define FD_EVENT_CLIENT_CREDIT_STALL_NANOS (20L*(long)1e9)
46 0 : #define FD_EVENT_CLIENT_TX_DRAIN_MAX (1024UL)
47 :
48 : /* The edge proxy fails a request after 100 MB of body. Rotate to a new
49 : StreamEvents request on the same connection before that. The server
50 : may override the cap in its Authenticate response. */
51 0 : #define FD_EVENT_CLIENT_STREAM_BODY_CAP (100UL*1000UL*1000UL)
52 0 : #define FD_EVENT_CLIENT_STREAM_BODY_MARGIN (64UL<<10)
53 :
54 : /* Sent in Authenticate; servers send newer response fields only to
55 : clients that advertise them. 1: max_stream_body */
56 0 : #define FD_EVENT_CLIENT_PROTOCOL_VERSION (1UL)
57 :
58 : /* StreamEventsZstd has one zstd stream per call. Each gRPC message is
59 : the compressor output at one flush; decompressed, the stream is
60 : varint(len) | StreamEventsRequest, repeated. A flush ends a batch
61 : always on an event boundary. */
62 0 : #define FD_EVENT_CLIENT_ZSTD_LEVEL (3)
63 0 : #define FD_EVENT_CLIENT_ZSTD_WLOG (20)
64 0 : #define FD_EVENT_CLIENT_ZSTD_BATCH (64UL<<10)
65 0 : #define FD_EVENT_CLIENT_ZSTD_FLUSH_NANOS (10L*1000L*1000L)
66 : FD_STATIC_ASSERT( FD_EVENT_CLIENT_ZSTD_LEVEL<=3 || FD_EVENT_CLIENT_ZSTD_LEVEL>=16, zstd_level );
67 0 : #define FD_EVENT_CLIENT_STREAM_DRAIN_NANOS (30L*(long)1e9)
68 :
69 0 : #define FD_EVENT_CLIENT_TOKEN_SZ (217UL)
70 :
71 : struct fd_event_client {
72 : fd_grpc_client_t * grpc_client;
73 : fd_grpc_client_metrics_t grpc_metrics[1];
74 : fd_grpc_h2_stream_t * event_stream;
75 :
76 : char client_version[ 10UL ];
77 : char commit_hash[ 41UL ];
78 : char action[ 16UL ];
79 : uchar identity_pubkey[ 32UL ];
80 :
81 : int has_genesis_hash;
82 : fd_hash_t genesis_hash[1];
83 :
84 : int connect_fail_logged;
85 :
86 : ushort has_shred_version;
87 : ushort shred_version;
88 :
89 : ulong event_id;
90 :
91 : ulong instance_id;
92 : ulong boot_id;
93 : ulong machine_id;
94 :
95 : int defer_disconnect;
96 : ulong consecutive_failure_count;
97 :
98 : long now; /* of the current poll */
99 :
100 : long last_stream_send_ns;
101 : long last_response_ns;
102 :
103 : long tx_tokens;
104 : long tx_tokens_ns;
105 : long tx_byte_rate;
106 : long stall_since;
107 : ulong stall_rem;
108 : ulong stall_events_sent;
109 : ulong stall_acked_id;
110 :
111 : int auth_send_pending;
112 :
113 : ulong stream_body_max; /* rotate before a stream's request body exceeds this */
114 :
115 : ZSTD_CCtx * zst;
116 : ZSTD_outBuffer zout;
117 : ulong zbatch_raw;
118 : ulong zbatch_cnt;
119 : long zbatch_ns;
120 : int zframe_open; /* frame started, epilogue not yet sent */
121 :
122 : ulong stream_gen; /* generation of event_stream, in its request_ctx */
123 : ulong stream_tx_sz; /* request body bytes sent on event_stream */
124 : fd_grpc_h2_stream_t * drain_stream; /* half-closed stream, NULL once the server ends it */
125 : ulong drain_gen; /* generation of the half-closed stream, 0 if none */
126 : ulong drain_last; /* last nonce sent on the half-closed stream */
127 : ulong held_ack; /* newer stream's ack held until the drain ends, ULONG_MAX if none */
128 :
129 : ulong state;
130 : union {
131 : struct {
132 : long reconnect_deadline;
133 : } disconnected;
134 :
135 : struct {
136 : long connect_deadline;
137 : } connecting;
138 :
139 : struct {
140 : long connected_timestamp;
141 : } connected;
142 : };
143 :
144 : int so_sndbuf;
145 : int sockfd;
146 :
147 : int epoll_fd;
148 : int epoll_out_armed;
149 :
150 : int use_tls;
151 : fd_tls_t tls[1];
152 : fd_chacha_rng_t tls_rng[1];
153 : fd_tlsrec_conn_t tls_conn[1];
154 :
155 : /* wallclock deadline for auth handshake, LONG_MAX if not
156 : authenticating. */
157 : long auth_deadline;
158 :
159 : char server_fqdn[ FD_FQDN_BUF_MAX ]; /* cstr */
160 : ulong server_fqdn_len;
161 : uint server_ip4_addr;
162 : ushort server_tcp_port;
163 :
164 : fd_rng_t * rng;
165 : fd_circq_t * circq;
166 : fd_keyguard_client_t * keyguard_client;
167 :
168 : /* Stateless-auth bearer value: "hex(challenge_token).hex(signature)",
169 : built from the challenge token returned by Authenticate and our
170 : ed25519 signature over it. Presented as the `authorization: Bearer
171 : <...>` header on the StreamEvents request. */
172 : char auth_bearer[ 2UL*FD_EVENT_CLIENT_TOKEN_SZ + 1UL + 2UL*64UL + 1UL ];
173 : ulong auth_bearer_len;
174 :
175 : fd_event_client_metrics_t metrics;
176 : };
177 :
178 :
179 : static void
180 : tls_init( fd_event_client_t * client,
181 0 : fd_x509_ca_store_t const * ca_store ) {
182 0 : fd_tls_t * tls = fd_tls_join( fd_tls_new( client->tls ) );
183 0 : FD_TEST( tls );
184 :
185 0 : uchar rng_key[ FD_CHACHA_KEY_SZ ];
186 0 : if( FD_UNLIKELY( !fd_rng_secure( rng_key, sizeof(rng_key) ) ) ) FD_LOG_CRIT(( "fd_rng_secure failed" ));
187 0 : fd_chacha_rng_init( client->tls_rng, rng_key, FD_CHACHA_RNG_ALGO_CHACHA8 );
188 0 : fd_memzero_explicit( rng_key, sizeof(rng_key) );
189 0 : tls->rng = client->tls_rng;
190 :
191 0 : static uchar const alpn[] = { 2, 'h', '2' };
192 0 : fd_memcpy( tls->alpn, alpn, sizeof(alpn) );
193 0 : tls->alpn_sz = sizeof(alpn);
194 :
195 0 : if( FD_UNLIKELY( client->server_fqdn_len>=sizeof(tls->server_name) ) ) {
196 0 : FD_LOG_ERR(( "Server name is %lu bytes, longer than the %lu byte maximum: check [tiles.event.url]",
197 0 : client->server_fqdn_len, sizeof(tls->server_name)-1UL ));
198 0 : }
199 0 : fd_memcpy( tls->server_name, client->server_fqdn, client->server_fqdn_len+1UL );
200 0 : tls->server_name_len = (ushort)client->server_fqdn_len;
201 :
202 0 : tls->ca_store = ca_store;
203 0 : }
204 :
205 : FD_FN_CONST ulong
206 0 : fd_event_client_align( void ) {
207 0 : return alignof( fd_event_client_t );
208 0 : }
209 :
210 : static ulong
211 0 : zstd_footprint( void ) {
212 0 : ZSTD_compressionParameters cp = ZSTD_getCParams( FD_EVENT_CLIENT_ZSTD_LEVEL, ZSTD_CONTENTSIZE_UNKNOWN, 0UL );
213 0 : cp.windowLog = FD_EVENT_CLIENT_ZSTD_WLOG;
214 0 : return ZSTD_estimateCStreamSize_usingCParams( cp );
215 0 : }
216 :
217 : ulong
218 0 : fd_event_client_footprint( ulong buf_max ) {
219 0 : ulong l;
220 0 : l = FD_LAYOUT_INIT;
221 0 : l = FD_LAYOUT_APPEND( l, alignof(fd_event_client_t), sizeof(fd_event_client_t) );
222 0 : l = FD_LAYOUT_APPEND( l, fd_grpc_client_align(), fd_grpc_client_footprint( buf_max ) );
223 0 : l = FD_LAYOUT_APPEND( l, 64UL, zstd_footprint() );
224 0 : return FD_LAYOUT_FINI( l, alignof(fd_event_client_t) );
225 0 : }
226 :
227 : void *
228 : fd_event_client_new( void * shmem,
229 : fd_keyguard_client_t * keyguard_client,
230 : fd_rng_t * rng,
231 : fd_circq_t * circq,
232 : int epoll_fd,
233 : int so_sndbuf,
234 : char const * _url,
235 : uchar const * identity_pubkey,
236 : char const * client_version,
237 : char const * commit_hash,
238 : char const * action,
239 : ulong instance_id,
240 : ulong boot_id,
241 : ulong machine_id,
242 : ulong buf_max,
243 : int use_tls,
244 : fd_x509_ca_store_t const * ca_store,
245 0 : ulong tx_rate_mbps ) {
246 0 : if( FD_UNLIKELY( !shmem ) ) {
247 0 : FD_LOG_WARNING(( "NULL shmem" ));
248 0 : return NULL;
249 0 : }
250 :
251 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shmem, fd_event_client_align() ) ) ) {
252 0 : FD_LOG_WARNING(( "misaligned shmem" ));
253 0 : return NULL;
254 0 : }
255 :
256 0 : FD_SCRATCH_ALLOC_INIT( l, shmem );
257 0 : fd_event_client_t * client = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_event_client_t), sizeof(fd_event_client_t) );
258 0 : void * grpc_client_mem = FD_SCRATCH_ALLOC_APPEND( l, fd_grpc_client_align(), fd_grpc_client_footprint( buf_max ) );
259 0 : void * zst_mem = FD_SCRATCH_ALLOC_APPEND( l, 64UL, zstd_footprint() );
260 :
261 0 : fd_url_t url[1];
262 0 : _Bool _is_ssl = 0;
263 0 : if( FD_UNLIKELY( fd_url_parse_endpoint( url,
264 0 : _url,
265 0 : strlen( _url ),
266 0 : &client->server_tcp_port,
267 0 : &_is_ssl,
268 0 : "[tiles.event.url]" ) ) ) {
269 0 : FD_LOG_ERR(( "Could not parse [tiles.event.url]" ));
270 0 : }
271 0 : if( FD_UNLIKELY( url->host_len>=FD_FQDN_BUF_MAX ) ) {
272 0 : FD_LOG_CRIT(( "Invalid url->host_len" )); /* unreachable */
273 0 : }
274 0 : fd_cstr_fini( fd_cstr_append_text( fd_cstr_init( client->server_fqdn ), url->host, url->host_len ) );
275 0 : client->server_fqdn_len = url->host_len;
276 :
277 0 : fd_memcpy( client->identity_pubkey, identity_pubkey, 32UL );
278 0 : fd_cstr_ncpy( client->client_version, client_version, sizeof( client->client_version ) );
279 0 : fd_cstr_fini( fd_cstr_append_text( fd_cstr_init( client->commit_hash ), commit_hash, fd_ulong_min( strlen( commit_hash ), sizeof( client->commit_hash )-1UL ) ) );
280 0 : fd_cstr_ncpy( client->action, action, sizeof( client->action ) );
281 :
282 0 : client->event_id = 0UL;
283 :
284 0 : client->instance_id = instance_id;
285 0 : client->boot_id = boot_id;
286 0 : client->machine_id = machine_id;
287 :
288 0 : client->has_genesis_hash = 0;
289 0 : client->has_shred_version = 0;
290 0 : client->connect_fail_logged = 0;
291 :
292 0 : client->so_sndbuf = so_sndbuf;
293 0 : client->sockfd = -1;
294 0 : FD_TEST( -1!=epoll_fd );
295 0 : client->epoll_fd = epoll_fd;
296 0 : client->use_tls = use_tls;
297 0 : if( use_tls ) {
298 0 : if( FD_UNLIKELY( !ca_store ) ) FD_LOG_ERR(( "NULL ca_store, but [tiles.event.url] requires TLS" ));
299 0 : tls_init( client, ca_store );
300 0 : }
301 0 : client->auth_deadline = LONG_MAX;
302 0 : client->auth_send_pending = 0;
303 0 : client->stream_body_max = FD_EVENT_CLIENT_STREAM_BODY_CAP-FD_EVENT_CLIENT_STREAM_BODY_MARGIN;
304 0 : client->zst = ZSTD_initStaticCStream( zst_mem, zstd_footprint() );
305 0 : FD_TEST( client->zst );
306 0 : FD_TEST( !ZSTD_isError( ZSTD_CCtx_setParameter( client->zst, ZSTD_c_compressionLevel, FD_EVENT_CLIENT_ZSTD_LEVEL ) ) );
307 0 : FD_TEST( !ZSTD_isError( ZSTD_CCtx_setParameter( client->zst, ZSTD_c_windowLog, FD_EVENT_CLIENT_ZSTD_WLOG ) ) );
308 0 : client->zbatch_raw = 0UL;
309 0 : client->zbatch_cnt = 0UL;
310 0 : client->zframe_open = 0;
311 0 : client->stream_gen = 0UL;
312 0 : client->stream_tx_sz = 0UL;
313 0 : client->drain_stream = NULL;
314 0 : client->drain_gen = 0UL;
315 0 : client->held_ack = ULONG_MAX;
316 0 : client->state = FD_EVENT_CLIENT_STATE_DISCONNECTED;
317 0 : client->disconnected.reconnect_deadline = 0L;
318 :
319 0 : client->defer_disconnect = INT_MAX;
320 0 : client->consecutive_failure_count = 7UL; /* Start high, so if server is down we don't keep retrying on boot */
321 0 : client->now = 0L;
322 0 : client->last_stream_send_ns = 0L;
323 0 : client->last_response_ns = 0L;
324 0 : client->tx_byte_rate = (long)( ( tx_rate_mbps ? tx_rate_mbps : FD_EVENT_CLIENT_TX_RATE_MBPS_DEFAULT )*125000UL );
325 0 : client->tx_tokens = FD_EVENT_CLIENT_TX_BURST;
326 0 : client->tx_tokens_ns = 0L;
327 0 : client->stall_since = 0L;
328 :
329 0 : client->circq = circq;
330 0 : client->rng = rng;
331 0 : client->keyguard_client = keyguard_client;
332 :
333 0 : extern fd_grpc_client_callbacks_t fd_event_client_grpc_callbacks;
334 0 : client->grpc_client = fd_grpc_client_new( grpc_client_mem, &fd_event_client_grpc_callbacks, client->grpc_metrics, client, buf_max, fd_rng_ulong( rng ) );
335 0 : FD_TEST( client->grpc_client );
336 :
337 0 : memset( &client->metrics, 0, sizeof(client->metrics) );
338 0 : memset( client->grpc_metrics, 0, sizeof(fd_grpc_client_metrics_t) );
339 :
340 0 : fd_grpc_client_set_version( client->grpc_client, client->client_version, strlen( client->client_version ) );
341 0 : fd_grpc_client_set_authority( client->grpc_client, client->server_fqdn, client->server_fqdn_len, client->server_tcp_port );
342 0 : fd_grpc_client_set_https( client->grpc_client, client->use_tls );
343 :
344 0 : return (void *)client;
345 0 : }
346 :
347 : fd_event_client_t *
348 0 : fd_event_client_join( void * shec ) {
349 0 : if( FD_UNLIKELY( !shec ) ) {
350 0 : FD_LOG_WARNING(( "NULL shec" ));
351 0 : return NULL;
352 0 : }
353 :
354 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shec, fd_event_client_align() ) ) ) {
355 0 : FD_LOG_WARNING(( "misaligned shec" ));
356 0 : return NULL;
357 0 : }
358 :
359 0 : fd_event_client_t * client = (fd_event_client_t *)shec;
360 :
361 0 : return client;
362 0 : }
363 :
364 : fd_event_client_metrics_t const *
365 0 : fd_event_client_metrics( fd_event_client_t const * client ) {
366 : /* Update bytes from grpc metrics */
367 0 : ((fd_event_client_t *)client)->metrics.bytes_written = client->grpc_metrics->stream_chunks_tx_bytes;
368 0 : ((fd_event_client_t *)client)->metrics.bytes_read = client->grpc_metrics->stream_chunks_rx_bytes;
369 0 : return &client->metrics;
370 0 : }
371 :
372 : ulong
373 0 : fd_event_client_state( fd_event_client_t const * client ) {
374 0 : return client->state;
375 0 : }
376 :
377 : ulong
378 0 : fd_event_client_id_reserve( fd_event_client_t * client ) {
379 0 : return client->event_id++;
380 0 : }
381 :
382 : void
383 : fd_event_client_init_genesis( fd_event_client_t * client,
384 0 : fd_genesis_meta_t const * meta ) {
385 0 : *client->genesis_hash = meta->genesis_hash;
386 0 : client->has_genesis_hash = 1;
387 0 : }
388 :
389 : void
390 : fd_event_client_init_shred_version( fd_event_client_t * client,
391 0 : ushort shred_version ) {
392 0 : client->shred_version = shred_version;
393 0 : client->has_shred_version = 1;
394 0 : }
395 :
396 : static void
397 : backoff( fd_event_client_t * client,
398 0 : long now ) {
399 0 : ulong backoff_base = 1UL << fd_ulong_min( client->consecutive_failure_count, 7UL ); /* max 4 mins */
400 0 : ulong backoff_jitter = fd_rng_ulong_roll( client->rng, backoff_base );
401 0 : client->disconnected.reconnect_deadline = now + (long)( backoff_base + backoff_jitter )*(long)1e9;
402 0 : if( FD_UNLIKELY( client->consecutive_failure_count < 8UL ) ) client->consecutive_failure_count++;
403 0 : }
404 :
405 : static void
406 : disconnect( fd_event_client_t * client,
407 : long now,
408 : int reason,
409 : int err,
410 0 : int _backoff ) {
411 0 : if( FD_LIKELY( -1!=client->sockfd ) ) {
412 0 : if( FD_UNLIKELY( -1==close( client->sockfd ) ) ) FD_LOG_ERR(( "close() failed (%d-%s)", errno, fd_io_strerror( errno ) ));
413 0 : client->sockfd = -1;
414 0 : client->epoll_out_armed = 0;
415 0 : client->state = FD_EVENT_CLIENT_STATE_DISCONNECTED;
416 0 : fd_circq_reset_cursor( client->circq );
417 0 : }
418 :
419 0 : fd_tlsrec_conn_fini( client->tls_conn );
420 0 : fd_memzero_explicit( client->tls->kex_private_key, 32UL );
421 :
422 0 : client->event_stream = NULL;
423 0 : client->defer_disconnect = INT_MAX;
424 0 : client->auth_deadline = LONG_MAX;
425 0 : client->auth_send_pending = 0;
426 0 : client->stall_since = 0L;
427 0 : client->stream_tx_sz = 0UL;
428 0 : client->zbatch_raw = 0UL;
429 0 : client->zbatch_cnt = 0UL;
430 0 : client->zframe_open = 0;
431 0 : client->drain_stream = NULL;
432 0 : client->drain_gen = 0UL;
433 0 : client->held_ack = ULONG_MAX;
434 :
435 0 : client->auth_bearer[ 0 ] = '\0';
436 0 : client->auth_bearer_len = 0UL;
437 :
438 0 : switch( reason ) {
439 0 : case DISCONNECT_REASON_IDENTITY_CHANGED:
440 0 : FD_LOG_INFO(( "disconnected: identity changed" ));
441 0 : break;
442 0 : case DISCONNECT_REASON_CONNECT_FAILED:
443 0 : if( FD_UNLIKELY( !client->connect_fail_logged ) ) FD_LOG_WARNING(( "connecting to telemetry server " FD_IP4_ADDR_FMT ":%u failed %s(%i-%s)%s", FD_IP4_ADDR_FMT_ARGS( client->server_ip4_addr ), client->server_tcp_port, fd_log_style_dim(), errno, fd_io_strerror( errno ), fd_log_style_normal() ));
444 0 : else FD_LOG_INFO(( "connecting to telemetry server " FD_IP4_ADDR_FMT ":%u failed (%i-%s)", FD_IP4_ADDR_FMT_ARGS( client->server_ip4_addr ), client->server_tcp_port, errno, fd_io_strerror( errno ) ));
445 0 : client->connect_fail_logged = 1;
446 0 : client->metrics.transport_fail_cnt++;
447 0 : break;
448 0 : case DISCONNECT_REASON_DNS_RESOLVE_FAILED:
449 0 : if( FD_UNLIKELY( !client->connect_fail_logged ) ) FD_LOG_WARNING(( "failed to resolve telemetry server host %.*s %s(%d-%s)%s", (int)client->server_fqdn_len, client->server_fqdn, fd_log_style_dim(), err, fd_gai_strerror( err ), fd_log_style_normal() ));
450 0 : else FD_LOG_INFO(( "failed to resolve telemetry server host %.*s (%d-%s)", (int)client->server_fqdn_len, client->server_fqdn, err, fd_gai_strerror( err ) ));
451 0 : client->connect_fail_logged = 1;
452 0 : client->metrics.transport_fail_cnt++;
453 0 : break;
454 0 : case DISCONNECT_REASON_TIMEOUT:
455 0 : FD_LOG_INFO(( "connection failed: timeout" ));
456 0 : client->metrics.transport_fail_cnt++;
457 0 : break;
458 0 : case DISCONNECT_REASON_TRANSPORT_FAILED:
459 0 : FD_LOG_WARNING(( "disconnected from telemetry server: transport failed %s(%d-%s)%s", fd_log_style_dim(), err, fd_io_strerror( err ), fd_log_style_normal() ));
460 0 : client->metrics.transport_fail_cnt++;
461 0 : break;
462 0 : case DISCONNECT_REASON_PEER_CLOSED:
463 0 : FD_LOG_WARNING(( "disconnected from telemetry server: peer closed connection" ));
464 0 : client->metrics.transport_fail_cnt++;
465 0 : break;
466 0 : case DISCONNECT_REASON_INVALID_CURSOR:
467 0 : FD_LOG_WARNING(( "disconnected from telemetry server: invalid cursor" ));
468 0 : client->metrics.transport_fail_cnt++;
469 0 : break;
470 0 : case DISCONNECT_REASON_AUTH_FAILED:
471 0 : FD_LOG_WARNING(( "disconnected from telemetry server: authentication failed" ));
472 0 : client->metrics.transport_fail_cnt++;
473 0 : break;
474 0 : case DISCONNECT_REASON_INVALID_PROTOBUF:
475 0 : FD_LOG_WARNING(( "disconnected from telemetry server: invalid protobuf message received" ));
476 0 : client->metrics.transport_fail_cnt++;
477 0 : break;
478 0 : default:
479 0 : FD_LOG_WARNING(( "disconnected from telemetry server: unknown reason %d", reason ));
480 0 : client->metrics.transport_fail_cnt++;
481 0 : break;
482 0 : }
483 :
484 0 : if( FD_LIKELY( _backoff ) ) backoff( client, now );
485 0 : }
486 :
487 : void
488 : fd_event_client_set_identity( fd_event_client_t * client,
489 0 : uchar const * identity_pubkey ) {
490 0 : fd_memcpy( client->identity_pubkey, identity_pubkey, 32UL );
491 0 : disconnect( client, client->now, DISCONNECT_REASON_IDENTITY_CHANGED, 0, 0 );
492 0 : }
493 :
494 : static void
495 : reconnect( fd_event_client_t * client,
496 : long now,
497 0 : int * charge_busy ) {
498 0 : FD_TEST( client->state==FD_EVENT_CLIENT_STATE_DISCONNECTED );
499 :
500 0 : if( FD_UNLIKELY( now<client->disconnected.reconnect_deadline ) ) return;
501 :
502 0 : *charge_busy = 1;
503 0 : client->metrics.connect_attempt_cnt++;
504 :
505 0 : FD_LOG_INFO(( "connecting to event server %s://%.*s:%u", client->use_tls ? "https" : "http", (int)client->server_fqdn_len, client->server_fqdn, client->server_tcp_port ));
506 :
507 : /* FIXME IPv6 support */
508 0 : fd_addrinfo_t hints = {0};
509 0 : hints.ai_family = AF_INET;
510 0 : fd_addrinfo_t * res = NULL;
511 0 : uchar scratch[ 4096 ];
512 0 : void * pscratch = scratch;
513 0 : int err = fd_getaddrinfo( client->server_fqdn, &hints, &res, &pscratch, sizeof(scratch) );
514 0 : if( FD_UNLIKELY( err ) ) {
515 0 : disconnect( client, now, DISCONNECT_REASON_DNS_RESOLVE_FAILED, err, 1 );
516 0 : return;
517 0 : }
518 :
519 0 : if( FD_UNLIKELY( !res || !res->ai_addr ) ) {
520 0 : disconnect( client, now, DISCONNECT_REASON_DNS_RESOLVE_FAILED, 0, 1 );
521 0 : return;
522 0 : }
523 :
524 0 : uint const ip4_addr = ((struct sockaddr_in *)res->ai_addr)->sin_addr.s_addr;
525 0 : client->server_ip4_addr = ip4_addr;
526 :
527 0 : client->sockfd = socket( AF_INET, SOCK_STREAM|SOCK_NONBLOCK, 0 );
528 0 : if( FD_UNLIKELY( -1==client->sockfd ) ) FD_LOG_ERR(( "socket() failed (%d-%s)", errno, fd_io_strerror( errno ) ));
529 :
530 0 : struct epoll_event ev = { .events = EPOLLIN|EPOLLOUT, .data.fd = client->sockfd };
531 0 : if( FD_UNLIKELY( -1==epoll_ctl( client->epoll_fd, EPOLL_CTL_ADD, client->sockfd, &ev ) ) ) FD_LOG_ERR(( "epoll_ctl(ADD,sockfd) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
532 0 : client->epoll_out_armed = 1;
533 :
534 0 : struct sockaddr_in addr;
535 0 : fd_memset( &addr, 0, sizeof( addr ) );
536 0 : addr.sin_family = AF_INET;
537 0 : addr.sin_port = fd_ushort_bswap( client->server_tcp_port );
538 0 : addr.sin_addr.s_addr = ip4_addr;
539 :
540 0 : int tcp_nodelay = 1;
541 0 : if( FD_UNLIKELY( -1==setsockopt( client->sockfd, SOL_TCP, TCP_NODELAY, &tcp_nodelay, sizeof(int) ) ) ) FD_LOG_ERR(( "setsockopt failed (%d-%s)", errno, fd_io_strerror( errno ) ));
542 0 : if( FD_UNLIKELY( -1==setsockopt( client->sockfd, SOL_SOCKET, SO_SNDBUF, &client->so_sndbuf, sizeof(int) ) ) ) FD_LOG_ERR(( "setsockopt(SOL_SOCKET,SO_SNDBUF,%i) failed (%i-%s)", client->so_sndbuf, errno, fd_io_strerror( errno ) ));
543 :
544 0 : if( FD_UNLIKELY( -1==connect( client->sockfd, fd_type_pun_const( &addr ), sizeof(struct sockaddr_in) ) && errno!=EINPROGRESS ) ) {
545 0 : disconnect( client, now, DISCONNECT_REASON_CONNECT_FAILED, errno, 1 );
546 0 : return;
547 0 : }
548 :
549 0 : if( client->use_tls ) {
550 0 : fd_tls_t * tls = client->tls;
551 0 : if( FD_UNLIKELY( !fd_rng_secure( tls->kex_private_key, 32UL ) ) ) FD_LOG_CRIT(( "fd_rng_secure failed" ));
552 0 : fd_x25519_public( tls->kex_public_key, tls->kex_private_key );
553 0 : fd_tlsrec_conn_init( client->tls_conn, tls, 0 );
554 0 : }
555 :
556 0 : fd_grpc_client_reset( client->grpc_client );
557 :
558 0 : client->state = FD_EVENT_CLIENT_STATE_CONNECTING;
559 0 : client->connecting.connect_deadline = now+(long)1L*(long)1e9; /* 1 second to connect */
560 0 : }
561 :
562 : static int
563 : fd_event_client_try_send_authenticate( fd_event_client_t * client,
564 0 : long now ) {
565 0 : if( FD_UNLIKELY( fd_grpc_client_request_is_blocked( client->grpc_client ) ) ) return 0;
566 0 : if( FD_UNLIKELY( fd_grpc_client_request_stream_busy( client->grpc_client ) ) ) return 0;
567 :
568 0 : fd_pb_encoder_t auth_req[1];
569 0 : uchar buffer[ 256UL ];
570 0 : fd_pb_encoder_init( auth_req, buffer, sizeof(buffer) );
571 :
572 0 : fd_pb_push_bytes( auth_req, 1U, client->identity_pubkey, 32UL );
573 0 : fd_pb_push_string( auth_req, 2U, client->client_version, strlen( client->client_version ) );
574 0 : fd_pb_push_string( auth_req, 3U, client->commit_hash, strlen( client->commit_hash ) );
575 0 : fd_pb_push_bytes( auth_req, 4U, client->genesis_hash, 32UL );
576 0 : fd_pb_push_uint64( auth_req, 5U, client->shred_version );
577 0 : fd_pb_push_uint64( auth_req, 6U, client->instance_id );
578 0 : fd_pb_push_uint64( auth_req, 7U, client->machine_id );
579 0 : fd_pb_push_uint64( auth_req, 8U, client->boot_id );
580 0 : fd_pb_push_string( auth_req, 9U, client->action, strlen( client->action ) );
581 0 : fd_pb_push_uint64( auth_req, 10U, FD_EVENT_CLIENT_PROTOCOL_VERSION );
582 :
583 0 : fd_grpc_h2_stream_t * stream = fd_grpc_client_request_start(
584 0 : client->grpc_client,
585 0 : "/events.v1.EventService/Authenticate", strlen("/events.v1.EventService/Authenticate"),
586 0 : FD_EVENT_CLIENT_REQ_CTX_AUTHENTICATE,
587 0 : buffer, fd_pb_encoder_out_sz( auth_req ),
588 0 : NULL, 0UL,
589 0 : 0 /* not streaming */ );
590 :
591 0 : if( FD_UNLIKELY( !stream ) ) return 0;
592 :
593 0 : fd_grpc_client_deadline_set( stream, FD_GRPC_DEADLINE_HEADER, now+(long)2e9 );
594 0 : fd_grpc_client_deadline_set( stream, FD_GRPC_DEADLINE_RX_END, now+(long)2e9 );
595 :
596 0 : client->auth_send_pending = 0;
597 0 : FD_LOG_INFO(( "Requesting auth challenge from event server " FD_IP4_ADDR_FMT ":%u (%.*s)",
598 0 : FD_IP4_ADDR_FMT_ARGS( client->server_ip4_addr ), client->server_tcp_port,
599 0 : (int)client->server_fqdn_len, client->server_fqdn ));
600 0 : return 1;
601 0 : }
602 :
603 : static void
604 0 : fd_event_client_grpc_conn_established( void * app_ctx ) {
605 0 : fd_event_client_t * client = app_ctx;
606 :
607 0 : long now = client->now;
608 0 : client->state = FD_EVENT_CLIENT_STATE_AUTHENTICATING;
609 0 : client->auth_deadline = now + (long)2e9;
610 0 : client->auth_send_pending = 1;
611 :
612 0 : fd_event_client_try_send_authenticate( client, now );
613 0 : }
614 :
615 : static void
616 : fd_event_client_handle_auth_challenge_resp( fd_event_client_t * client,
617 : void const * protobuf,
618 0 : ulong protobuf_sz ) {
619 0 : fd_pb_inbuf_t inbuf[1];
620 0 : fd_pb_inbuf_init( inbuf, protobuf, protobuf_sz );
621 :
622 : /* challenge (1) and optional max_stream_body (2), in any order */
623 0 : uchar const * challenge = NULL;
624 0 : ulong body_cap = FD_EVENT_CLIENT_STREAM_BODY_CAP;
625 0 : while( fd_pb_inbuf_sz( inbuf ) ) {
626 0 : fd_pb_tlv_t tlv;
627 0 : if( FD_UNLIKELY( !fd_pb_tlv_read( inbuf, &tlv ) ) ) {
628 0 : FD_LOG_WARNING(( "Failed to parse auth challenge response" ));
629 0 : client->defer_disconnect = DISCONNECT_REASON_AUTH_FAILED;
630 0 : return;
631 0 : }
632 0 : if( tlv.field_id==1U && tlv.wire_type==FD_PB_WIRE_TYPE_LEN ) {
633 0 : if( FD_UNLIKELY( tlv.len!=FD_EVENT_CLIENT_TOKEN_SZ || fd_pb_inbuf_sz( inbuf )<tlv.len ) ) {
634 0 : FD_LOG_WARNING(( "Invalid challenge token size: %lu bytes (expected %lu)", tlv.len, FD_EVENT_CLIENT_TOKEN_SZ ));
635 0 : client->defer_disconnect = DISCONNECT_REASON_AUTH_FAILED;
636 0 : return;
637 0 : }
638 0 : challenge = inbuf->cur;
639 0 : inbuf->cur += tlv.len;
640 0 : } else if( tlv.field_id==2U && tlv.wire_type==FD_PB_WIRE_TYPE_VARINT ) {
641 0 : body_cap = fd_ulong_min( tlv.varint, UINT_MAX );
642 0 : } else {
643 0 : FD_LOG_WARNING(( "Unexpected field in auth challenge response" ));
644 0 : client->defer_disconnect = DISCONNECT_REASON_AUTH_FAILED;
645 0 : return;
646 0 : }
647 0 : }
648 0 : if( FD_UNLIKELY( !challenge ) ) {
649 0 : FD_LOG_WARNING(( "Missing challenge in auth challenge response" ));
650 0 : client->defer_disconnect = DISCONNECT_REASON_AUTH_FAILED;
651 0 : return;
652 0 : }
653 0 : if( FD_UNLIKELY( body_cap<client->grpc_client->pb_tx_max+FD_EVENT_CLIENT_STREAM_BODY_MARGIN ) ) {
654 0 : FD_LOG_WARNING(( "telemetry server stream body cap %lu is below the largest event (%lu)", body_cap, client->grpc_client->pb_tx_max ));
655 0 : client->defer_disconnect = DISCONNECT_REASON_AUTH_FAILED;
656 0 : return;
657 0 : }
658 0 : client->stream_body_max = body_cap-FD_EVENT_CLIENT_STREAM_BODY_MARGIN;
659 :
660 0 : ulong challenge_len = FD_EVENT_CLIENT_TOKEN_SZ;
661 0 : uchar challenge_token[ FD_EVENT_CLIENT_TOKEN_SZ ];
662 0 : memcpy( challenge_token, challenge, challenge_len );
663 :
664 0 : uchar sign_request[ 100UL + FD_EVENT_CLIENT_TOKEN_SZ ];
665 0 : static char const sign_prefix[ 100 ] =
666 0 : " " /* 32 spaces */
667 0 : " " /* 32 spaces */
668 0 : "Firedancer event challenge-response";
669 0 : memcpy( sign_request, sign_prefix, sizeof(sign_prefix) );
670 0 : memcpy( sign_request+100, challenge_token, challenge_len );
671 :
672 0 : uchar signature[ 64UL ];
673 0 : fd_keyguard_client_sign( client->keyguard_client,
674 0 : signature,
675 0 : sign_request, 100UL+challenge_len,
676 0 : FD_KEYGUARD_SIGN_TYPE_ED25519 );
677 :
678 : /* Build "hex(challenge_token).hex(signature)" for the bearer token. */
679 0 : fd_hex_encode( client->auth_bearer, challenge_token, FD_EVENT_CLIENT_TOKEN_SZ );
680 0 : client->auth_bearer[ 2UL*FD_EVENT_CLIENT_TOKEN_SZ ] = '.';
681 0 : fd_hex_encode( client->auth_bearer + 2UL*FD_EVENT_CLIENT_TOKEN_SZ+1UL, signature, 64UL );
682 0 : client->auth_bearer_len = 2UL*FD_EVENT_CLIENT_TOKEN_SZ + 1UL + 2UL*64UL;
683 0 : client->auth_bearer[ client->auth_bearer_len ] = '\0';
684 :
685 0 : if( FD_LIKELY( client->state==FD_EVENT_CLIENT_STATE_CONNECTED ) ) return; /* stream rotation */
686 :
687 0 : client->event_stream = NULL;
688 0 : client->metrics.transport_success_cnt++;
689 0 : client->state = FD_EVENT_CLIENT_STATE_CONNECTED;
690 0 : client->connected.connected_timestamp = client->now;
691 0 : client->connect_fail_logged = 0;
692 0 : FD_LOG_NOTICE(( "connected to telemetry server %s%s://%.*s:%u%s",
693 0 : fd_log_style_bold(), client->use_tls ? "https" : "http",
694 0 : (int)client->server_fqdn_len, client->server_fqdn, client->server_tcp_port,
695 0 : fd_log_style_normal() ));
696 0 : }
697 :
698 : static void
699 : fd_event_client_grpc_conn_dead( void * app_ctx,
700 : uint h2_err,
701 0 : int closed_by ) {
702 0 : fd_event_client_t * client = app_ctx;
703 0 : FD_LOG_WARNING(( "telemetry connection closed %s %s(%u-%s)%s",
704 0 : closed_by ? "by peer" : "due to error",
705 0 : fd_log_style_dim(), h2_err, fd_h2_strerror( h2_err ), fd_log_style_normal() ));
706 0 : client->defer_disconnect = DISCONNECT_REASON_PEER_CLOSED;
707 0 : }
708 :
709 : static void
710 : fd_event_client_grpc_tx_complete( void * app_ctx,
711 0 : ulong request_ctx ) {
712 0 : (void)app_ctx; (void)request_ctx;
713 0 : }
714 :
715 : void
716 : fd_event_client_grpc_rx_start( void * app_ctx,
717 0 : ulong request_ctx ) {
718 0 : (void)app_ctx; (void)request_ctx;
719 0 : }
720 :
721 : static void
722 : ack_apply( fd_event_client_t * client,
723 0 : ulong nonce_ack ) {
724 0 : client->metrics.last_acked_id = nonce_ack;
725 0 : if( FD_UNLIKELY( -1==fd_circq_pop_until( client->circq, nonce_ack ) ) ) {
726 0 : FD_LOG_WARNING(( "Event gRPC rx msg: invalid cursor ack %lu", nonce_ack ));
727 0 : client->defer_disconnect = DISCONNECT_REASON_INVALID_CURSOR;
728 0 : }
729 0 : }
730 :
731 : static void
732 0 : drain_end( fd_event_client_t * client ) {
733 0 : client->drain_gen = 0UL;
734 0 : if( FD_UNLIKELY( client->held_ack!=ULONG_MAX ) ) ack_apply( client, client->held_ack );
735 0 : client->held_ack = ULONG_MAX;
736 0 : }
737 :
738 : static void
739 : fd_event_client_handle_stream_events_resp( fd_event_client_t * client,
740 : void const * protobuf,
741 : ulong protobuf_sz,
742 0 : ulong request_ctx ) {
743 0 : fd_pb_inbuf_t inbuf[1];
744 0 : fd_pb_inbuf_init( inbuf, protobuf, protobuf_sz );
745 :
746 0 : ulong nonce_ack = 0UL;
747 0 : if( FD_LIKELY( protobuf_sz ) ) {
748 0 : fd_pb_tlv_t event_id;
749 0 : if( FD_UNLIKELY( !fd_pb_tlv_read( inbuf, &event_id ) ||
750 0 : event_id.field_id!=1U /* event_id */ ||
751 0 : event_id.wire_type!=FD_PB_WIRE_TYPE_VARINT ) ) {
752 0 : FD_LOG_WARNING(( "Event gRPC rx msg: invalid Protobuf" ));
753 0 : client->defer_disconnect = DISCONNECT_REASON_INVALID_PROTOBUF;
754 0 : return;
755 0 : }
756 0 : nonce_ack = event_id.varint;
757 :
758 0 : if( FD_UNLIKELY( fd_pb_inbuf_sz( inbuf ) ) ) {
759 0 : FD_LOG_WARNING(( "Event gRPC rx msg: trailing data in StreamEventsResponse" ));
760 0 : client->defer_disconnect = DISCONNECT_REASON_INVALID_PROTOBUF;
761 0 : return;
762 0 : }
763 0 : }
764 :
765 0 : client->metrics.events_acked++;
766 0 : client->last_response_ns = client->now;
767 0 : if( FD_UNLIKELY( client->now-client->connected.connected_timestamp>=10L*(long)1e9 ) ) client->consecutive_failure_count = 0UL; /* responding and stayed up */
768 0 : if( FD_UNLIKELY( nonce_ack==ULONG_MAX ) ) return;
769 :
770 : /* Acks are cumulative over the circq, so the new stream's acks would
771 : also pop events still in flight on the draining stream. */
772 0 : ulong gen = request_ctx>>8;
773 0 : if( FD_UNLIKELY( client->drain_gen && gen!=client->drain_gen ) ) {
774 0 : client->held_ack = nonce_ack;
775 0 : return;
776 0 : }
777 0 : if( FD_UNLIKELY( !client->drain_gen && gen!=client->stream_gen ) ) return; /* stale watermark from a drained stream */
778 :
779 0 : ack_apply( client, nonce_ack );
780 0 : if( FD_UNLIKELY( client->drain_gen && nonce_ack>=client->drain_last ) ) drain_end( client );
781 0 : }
782 :
783 : void
784 : fd_event_client_grpc_rx_msg( void * app_ctx,
785 : void const * protobuf,
786 : ulong protobuf_sz,
787 0 : ulong request_ctx ) {
788 0 : fd_event_client_t * client = app_ctx;
789 :
790 0 : switch( request_ctx & 0xffUL ) {
791 0 : case FD_EVENT_CLIENT_REQ_CTX_AUTHENTICATE:
792 0 : fd_event_client_handle_auth_challenge_resp( client, protobuf, protobuf_sz );
793 0 : break;
794 0 : case FD_EVENT_CLIENT_REQ_CTX_STREAM_EVENTS:
795 0 : fd_event_client_handle_stream_events_resp( client, protobuf, protobuf_sz, request_ctx );
796 0 : break;
797 0 : default:
798 0 : FD_LOG_WARNING(( "Unknown request_ctx: %lu, disconnecting", request_ctx ));
799 0 : client->defer_disconnect = DISCONNECT_REASON_INVALID_PROTOBUF;
800 0 : break;
801 0 : }
802 0 : }
803 :
804 : void
805 : fd_event_client_grpc_rx_end( void * app_ctx,
806 : ulong request_ctx,
807 0 : fd_grpc_resp_hdrs_t * resp ) {
808 0 : fd_event_client_t * client = app_ctx;
809 :
810 0 : if( FD_UNLIKELY( client->drain_stream && request_ctx==client->drain_stream->request_ctx ) ) {
811 0 : client->drain_stream = NULL; /* released by the gRPC client */
812 0 : if( FD_LIKELY( !client->drain_gen ) ) return;
813 : /* Unacked events on the old stream sit behind the cursor: resend
814 : them by reconnecting. */
815 0 : FD_LOG_WARNING(( "telemetry server ended rotated stream before acking it (%lu<%lu)", client->metrics.last_acked_id, client->drain_last ));
816 0 : client->defer_disconnect = DISCONNECT_REASON_PEER_CLOSED;
817 0 : return;
818 0 : }
819 :
820 0 : if( FD_UNLIKELY( resp->h2_status!=200 ) ) {
821 0 : FD_LOG_WARNING(( "telemetry server request failed %s(HTTP status %u)%s", fd_log_style_dim(), resp->h2_status, fd_log_style_normal() ));
822 0 : client->defer_disconnect = DISCONNECT_REASON_TRANSPORT_FAILED;
823 0 : return;
824 0 : }
825 :
826 0 : resp->grpc_msg_len = (uint)fd_url_unescape( resp->grpc_msg, resp->grpc_msg_len );
827 0 : if( !resp->grpc_msg_len ) {
828 0 : fd_memcpy( resp->grpc_msg, "unknown error", 13 );
829 0 : resp->grpc_msg_len = 13;
830 0 : }
831 :
832 0 : if( FD_UNLIKELY( resp->grpc_status!=FD_GRPC_STATUS_OK ) ) {
833 0 : switch( request_ctx & 0xffUL ) {
834 0 : case FD_EVENT_CLIENT_REQ_CTX_AUTHENTICATE:
835 0 : FD_LOG_WARNING(( "telemetry server authentication failed: %.*s %s(%u-%s)%s",
836 0 : (int)resp->grpc_msg_len, resp->grpc_msg,
837 0 : fd_log_style_dim(), resp->grpc_status, fd_grpc_status_cstr( resp->grpc_status ), fd_log_style_normal() ));
838 0 : client->defer_disconnect = DISCONNECT_REASON_AUTH_FAILED;
839 0 : return;
840 0 : case FD_EVENT_CLIENT_REQ_CTX_STREAM_EVENTS:
841 0 : FD_LOG_WARNING(( "telemetry server event stream failed: %.*s %s(%u-%s)%s",
842 0 : (int)resp->grpc_msg_len, resp->grpc_msg,
843 0 : fd_log_style_dim(), resp->grpc_status, fd_grpc_status_cstr( resp->grpc_status ), fd_log_style_normal() ));
844 0 : client->defer_disconnect = DISCONNECT_REASON_PEER_CLOSED;
845 0 : return;
846 0 : default:
847 0 : FD_LOG_WARNING(( "telemetry server request failed: %.*s %s(%u-%s)%s",
848 0 : (int)resp->grpc_msg_len, resp->grpc_msg,
849 0 : fd_log_style_dim(), resp->grpc_status, fd_grpc_status_cstr( resp->grpc_status ), fd_log_style_normal() ));
850 0 : client->defer_disconnect = DISCONNECT_REASON_TRANSPORT_FAILED;
851 0 : return;
852 0 : }
853 0 : }
854 :
855 0 : if( (request_ctx & 0xffUL)==FD_EVENT_CLIENT_REQ_CTX_STREAM_EVENTS ) {
856 0 : FD_LOG_INFO(( "telemetry server event stream ended gracefully" ));
857 0 : client->defer_disconnect = DISCONNECT_REASON_PEER_CLOSED;
858 0 : }
859 :
860 0 : if( FD_UNLIKELY( (request_ctx & 0xffUL)==FD_EVENT_CLIENT_REQ_CTX_AUTHENTICATE &&
861 0 : client->state==FD_EVENT_CLIENT_STATE_CONNECTED &&
862 0 : !client->event_stream && !client->auth_bearer_len ) ) {
863 0 : FD_LOG_WARNING(( "telemetry server ended authentication without a challenge" ));
864 0 : client->defer_disconnect = DISCONNECT_REASON_AUTH_FAILED;
865 0 : }
866 0 : }
867 :
868 : void
869 : fd_event_client_grpc_rx_timeout( void * app_ctx,
870 : ulong request_ctx,
871 0 : int deadline_kind FD_PARAM_UNUSED ) {
872 0 : fd_event_client_t * client = (fd_event_client_t *)app_ctx;
873 0 : if( FD_UNLIKELY( client->drain_stream && request_ctx==client->drain_stream->request_ctx ) ) {
874 0 : client->drain_stream = NULL; /* released by the gRPC client */
875 0 : if( FD_LIKELY( !client->drain_gen ) ) return;
876 0 : }
877 0 : FD_LOG_WARNING(( "Event gRPC rx timeout" ));
878 0 : client->defer_disconnect = DISCONNECT_REASON_TRANSPORT_FAILED;
879 0 : client->event_stream = NULL;
880 0 : }
881 :
882 : static void
883 0 : fd_event_client_grpc_ping_ack( void * app_ctx ) {
884 0 : (void)app_ctx;
885 0 : FD_LOG_WARNING(( "Event gRPC ping ack" ));
886 0 : }
887 :
888 : static long
889 : pace_refill( fd_event_client_t * client,
890 0 : long now ) {
891 0 : ulong rate = (ulong)client->tx_byte_rate;
892 0 : long dt = fd_long_max( now-client->tx_tokens_ns, 0L );
893 0 : uint128 avail = (uint128)(ulong)dt*rate/1000000000UL;
894 0 : ulong room = (ulong)( FD_EVENT_CLIENT_TX_BURST-client->tx_tokens );
895 0 : long grant = (long)( avail<room ? (ulong)avail : room );
896 0 : client->tx_tokens += grant;
897 0 : client->tx_tokens_ns = client->tx_tokens==FD_EVENT_CLIENT_TX_BURST ? now : client->tx_tokens_ns+(long)( ( (uint128)(ulong)grant*1000000000UL+rate-1UL )/rate );
898 0 : return client->tx_tokens;
899 0 : }
900 :
901 :
902 : static void
903 : zstd_emit( fd_event_client_t * client,
904 : long now,
905 : int * charge_busy,
906 0 : ZSTD_EndDirective mode ) {
907 0 : ZSTD_inBuffer in = { .src = NULL, .size = 0UL, .pos = 0UL };
908 0 : ulong rem = ZSTD_compressStream2( client->zst, &client->zout, &in, mode );
909 0 : if( FD_UNLIKELY( ZSTD_isError( rem ) || rem ) ) FD_LOG_CRIT(( "ZSTD_compressStream2 failed: %s", ZSTD_isError( rem ) ? ZSTD_getErrorName( rem ) : "output full" ));
910 0 : if( mode==ZSTD_e_end ) client->zframe_open = 0;
911 0 : ulong sz = client->zout.pos;
912 0 : FD_TEST( fd_grpc_client_stream_send_msg( client->grpc_client, client->event_stream, client->zout.dst, sz ) );
913 0 : client->tx_tokens -= (long)sz;
914 0 : client->stream_tx_sz += sizeof(fd_grpc_hdr_t)+sz;
915 0 : client->metrics.events_sent += client->zbatch_cnt;
916 0 : client->last_stream_send_ns = now;
917 0 : client->zout.pos = 0UL;
918 0 : client->zbatch_raw = 0UL;
919 0 : client->zbatch_cnt = 0UL;
920 0 : *charge_busy = 1;
921 0 : }
922 :
923 : static void
924 : rotate( fd_event_client_t * client,
925 : long now,
926 0 : int * charge_busy ) {
927 0 : if( FD_UNLIKELY( client->drain_gen || client->drain_stream ) ) return; /* previous rotation still draining */
928 0 : if( FD_UNLIKELY( client->zframe_open ) ) { zstd_emit( client, now, charge_busy, ZSTD_e_end ); return; } /* close once the epilogue is out */
929 0 : if( FD_UNLIKELY( !fd_grpc_client_stream_close( client->grpc_client, client->event_stream ) ) ) return;
930 0 : FD_LOG_INFO(( "rotating telemetry event stream after %lu bytes", client->stream_tx_sz ));
931 0 : fd_grpc_client_deadline_set( client->event_stream, FD_GRPC_DEADLINE_RX_END, now+FD_EVENT_CLIENT_STREAM_DRAIN_NANOS );
932 0 : client->drain_stream = client->event_stream;
933 0 : client->event_stream = NULL;
934 0 : client->auth_send_pending = 1;
935 0 : if( FD_LIKELY( fd_circq_unsent_cnt( client->circq )<client->circq->cnt ) ) { /* unacked events in flight */
936 0 : client->drain_gen = client->stream_gen;
937 0 : client->drain_last = client->circq->cursor_seq-1UL;
938 0 : }
939 0 : *charge_busy = 1;
940 0 : }
941 :
942 : static void
943 : tx_zstd( fd_event_client_t * client,
944 : long now,
945 : long tokens,
946 0 : int * charge_busy ) {
947 0 : for(;;) {
948 0 : ulong next_sz;
949 0 : int has_next = fd_circq_cursor_peek( client->circq, &next_sz );
950 :
951 0 : if( client->zbatch_cnt && ( !has_next || tokens<=0L ||
952 0 : client->zbatch_raw>=FD_EVENT_CLIENT_ZSTD_BATCH ) ) {
953 0 : if( FD_UNLIKELY( now-client->zbatch_ns>=FD_EVENT_CLIENT_ZSTD_FLUSH_NANOS || client->zbatch_raw>=FD_EVENT_CLIENT_ZSTD_BATCH ) ) zstd_emit( client, now, charge_busy, ZSTD_e_flush );
954 0 : return;
955 0 : }
956 :
957 0 : if( !has_next ) {
958 0 : if( FD_UNLIKELY( now-client->last_stream_send_ns>FD_EVENT_CLIENT_HEARTBEAT_NANOS ) ) {
959 0 : if( FD_UNLIKELY( client->stream_tx_sz+sizeof(fd_grpc_hdr_t)>client->stream_body_max ) ) { rotate( client, now, charge_busy ); return; }
960 0 : if( FD_LIKELY( fd_grpc_client_stream_send_msg( client->grpc_client, client->event_stream, (uchar const *)"", 0UL ) ) ) {
961 0 : client->stream_tx_sz += sizeof(fd_grpc_hdr_t);
962 0 : client->last_stream_send_ns = now;
963 0 : *charge_busy = 1;
964 0 : }
965 0 : }
966 0 : return;
967 0 : }
968 0 : if( FD_UNLIKELY( tokens<=0L ) ) return;
969 :
970 : /* Worst case batch size on the wire with this event */
971 0 : ulong raw = client->zbatch_raw + fd_pb_varint64_sz_max + next_sz;
972 0 : ulong out = FD_EVENT_CLIENT_ZSTD_OUT_MAX( raw );
973 0 : if( FD_UNLIKELY( client->stream_tx_sz+sizeof(fd_grpc_hdr_t)+out>client->stream_body_max ||
974 0 : out>client->zout.size ) ) {
975 0 : if( client->zbatch_cnt ) { zstd_emit( client, now, charge_busy, ZSTD_e_flush ); return; }
976 0 : if( FD_UNLIKELY( out>client->zout.size ) ) FD_LOG_CRIT(( "%lu byte telemetry event does not fit the %lu byte gRPC buffer", next_sz, client->grpc_client->pb_tx_max ));
977 0 : rotate( client, now, charge_busy );
978 0 : return;
979 0 : }
980 :
981 0 : ulong msg_sz;
982 0 : uchar const * msg = fd_circq_cursor_advance( client->circq, &msg_sz );
983 0 : uchar prefix[ fd_pb_varint64_sz_max ];
984 0 : ulong prefix_sz = (ulong)( fd_pb_append_varint64( prefix, msg_sz )-prefix );
985 0 : ZSTD_inBuffer in0 = { .src = prefix, .size = prefix_sz, .pos = 0UL };
986 0 : ZSTD_inBuffer in1 = { .src = msg, .size = msg_sz, .pos = 0UL };
987 0 : ulong r0 = ZSTD_compressStream2( client->zst, &client->zout, &in0, ZSTD_e_continue );
988 0 : ulong r1 = ZSTD_compressStream2( client->zst, &client->zout, &in1, ZSTD_e_continue );
989 0 : if( FD_UNLIKELY( ZSTD_isError( r0 ) || ZSTD_isError( r1 ) || in0.pos!=in0.size || in1.pos!=in1.size ) ) FD_LOG_CRIT(( "ZSTD_compressStream2 failed" ));
990 0 : if( !client->zbatch_cnt ) client->zbatch_ns = now;
991 0 : client->zframe_open = 1;
992 0 : client->zbatch_raw += prefix_sz+msg_sz;
993 0 : client->zbatch_cnt++;
994 0 : *charge_busy = 1;
995 0 : }
996 0 : }
997 :
998 : static void
999 : tx( fd_event_client_t * client,
1000 : long now,
1001 0 : int * charge_busy ) {
1002 0 : FD_TEST( client->state==FD_EVENT_CLIENT_STATE_CONNECTED );
1003 :
1004 0 : long tokens = pace_refill( client, now );
1005 :
1006 : /* Fully acked, but the server has not ended it. Cancel before
1007 : anything that waits on a stream slot: it may hold the only one. */
1008 0 : if( FD_UNLIKELY( client->drain_stream && !client->drain_gen ) ) {
1009 0 : if( FD_UNLIKELY( !fd_grpc_client_stream_cancel( client->grpc_client, client->drain_stream ) ) ) return;
1010 0 : client->drain_stream = NULL;
1011 0 : *charge_busy = 1;
1012 0 : return;
1013 0 : }
1014 :
1015 0 : if( FD_UNLIKELY( client->event_stream && client->grpc_client->request_stream != NULL && client->grpc_client->request_stream!=client->event_stream ) ) return;
1016 :
1017 0 : if( FD_UNLIKELY( client->event_stream ) ) {
1018 0 : if( FD_UNLIKELY( fd_grpc_client_stream_send_is_blocked( client->grpc_client ) ) ) return;
1019 0 : } else {
1020 0 : if( FD_UNLIKELY( fd_grpc_client_request_is_blocked( client->grpc_client ) ) ) return;
1021 0 : }
1022 :
1023 0 : if( FD_UNLIKELY( !client->event_stream ) ) {
1024 0 : if( FD_UNLIKELY( !client->auth_bearer_len ) ) return; /* rotation awaiting a fresh token */
1025 0 : client->event_stream = fd_grpc_client_request_start(
1026 0 : client->grpc_client,
1027 0 : "/events.v1.EventService/StreamEventsZstd", strlen("/events.v1.EventService/StreamEventsZstd"),
1028 0 : FD_EVENT_CLIENT_REQ_CTX_STREAM_EVENTS | ((client->stream_gen+1UL)<<8),
1029 0 : NULL, 0UL, /* headers only; first message sent later */
1030 0 : client->auth_bearer, client->auth_bearer_len,
1031 0 : 1 /* streaming */ );
1032 0 : if( FD_UNLIKELY( !client->event_stream ) ) return; /* transient; retry next poll */
1033 0 : client->stream_gen++;
1034 0 : client->stream_tx_sz = 0UL;
1035 0 : client->auth_bearer_len = 0UL;
1036 0 : FD_TEST( !ZSTD_isError( ZSTD_CCtx_reset( client->zst, ZSTD_reset_session_only ) ) );
1037 0 : client->zout = (ZSTD_outBuffer){ .dst = client->grpc_client->pb_tx+sizeof(fd_grpc_hdr_t), .size = client->grpc_client->pb_tx_max-sizeof(fd_grpc_hdr_t), .pos = 0UL };
1038 0 : client->zbatch_raw = 0UL;
1039 0 : client->zbatch_cnt = 0UL;
1040 0 : client->zframe_open = 0;
1041 0 : fd_grpc_client_deadline_set( client->event_stream, FD_GRPC_DEADLINE_HEADER, now+(long)10e9 /* 10s */ );
1042 0 : client->last_stream_send_ns = now;
1043 0 : client->last_response_ns = now;
1044 0 : *charge_busy = 1;
1045 0 : return;
1046 0 : }
1047 :
1048 0 : tx_zstd( client, now, tokens, charge_busy );
1049 0 : }
1050 :
1051 : static void
1052 : epoll_out( fd_event_client_t * client,
1053 0 : int want ) {
1054 0 : if( FD_UNLIKELY( client->sockfd<0 ) ) return;
1055 0 : if( FD_LIKELY( want==client->epoll_out_armed ) ) return;
1056 0 : struct epoll_event ev = { .events = EPOLLIN | (want ? EPOLLOUT : 0U), .data.fd = client->sockfd };
1057 0 : if( FD_UNLIKELY( -1==epoll_ctl( client->epoll_fd, EPOLL_CTL_MOD, client->sockfd, &ev ) ) ) FD_LOG_ERR(( "epoll_ctl(MOD,sockfd) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
1058 0 : client->epoll_out_armed = want;
1059 0 : }
1060 :
1061 : long
1062 : fd_event_client_next_deadline( fd_event_client_t const * client,
1063 0 : long now ) {
1064 0 : if( FD_UNLIKELY( !client->has_genesis_hash || !client->has_shred_version ) ) return LONG_MAX; /* frag-driven */
1065 :
1066 0 : switch( client->state ) {
1067 0 : case FD_EVENT_CLIENT_STATE_DISCONNECTED:
1068 0 : return fd_long_max( client->disconnected.reconnect_deadline, now );
1069 0 : case FD_EVENT_CLIENT_STATE_CONNECTING:
1070 : /* Completion/failure arrives as an EPOLLOUT/EPOLLERR event. */
1071 0 : return client->connecting.connect_deadline;
1072 0 : default: break;
1073 0 : }
1074 :
1075 : /* Decrypted TLS bytes buffered in the gRPC client do not make the fd
1076 : readable: drain now or they wait for the next unrelated event.
1077 : Except while TX is blocked: then they are held because the HTTP/2
1078 : rings are full behind the parked send, and only the EPOLLOUT armed
1079 : for that send can make progress. */
1080 0 : if( FD_UNLIKELY( client->use_tls &&
1081 0 : fd_grpc_client_tls_rx_pending( client->grpc_client ) &&
1082 0 : !fd_grpc_client_tls_tx_pending( client->grpc_client ) ) ) return now;
1083 :
1084 0 : long deadline = fd_grpc_client_next_deadline( client->grpc_client );
1085 0 : if( FD_UNLIKELY( client->state==FD_EVENT_CLIENT_STATE_AUTHENTICATING ) ) deadline = fd_long_min( deadline, client->auth_deadline );
1086 0 : if( FD_UNLIKELY( client->auth_send_pending && !fd_grpc_client_request_is_blocked( client->grpc_client ) ) ) deadline = now;
1087 0 : if( FD_LIKELY( client->state==FD_EVENT_CLIENT_STATE_CONNECTED && client->event_stream ) ) {
1088 0 : long send_deadline = client->last_stream_send_ns+FD_EVENT_CLIENT_HEARTBEAT_NANOS;
1089 0 : if( FD_UNLIKELY( client->tx_tokens<=0L && fd_circq_unsent_cnt( client->circq ) ) ) { /* paced: no heartbeat until tokens return */
1090 0 : ulong rate = (ulong)client->tx_byte_rate;
1091 0 : send_deadline = client->tx_tokens_ns + (long)( ( (uint128)(ulong)(1L-client->tx_tokens)*1000000000UL+rate-1UL )/rate );
1092 0 : }
1093 0 : deadline = fd_long_min( deadline, fd_long_min( client->last_response_ns+FD_EVENT_CLIENT_RESPONSE_TIMEOUT_NANOS, send_deadline ) );
1094 0 : if( FD_UNLIKELY( client->stall_since ) ) deadline = fd_long_min( deadline, client->stall_since+FD_EVENT_CLIENT_CREDIT_STALL_NANOS );
1095 0 : if( FD_UNLIKELY( client->zbatch_cnt ) ) deadline = fd_long_min( deadline, client->zbatch_ns+FD_EVENT_CLIENT_ZSTD_FLUSH_NANOS );
1096 0 : }
1097 0 : return deadline;
1098 0 : }
1099 :
1100 : static int
1101 : credit_stall_check( fd_event_client_t * client,
1102 0 : long now ) {
1103 0 : ulong rem = fd_grpc_client_tx_starved( client->grpc_client );
1104 0 : if( FD_LIKELY( !rem ) ) {
1105 0 : client->stall_since = 0L;
1106 0 : return 0;
1107 0 : }
1108 :
1109 0 : if( !client->stall_since || rem!=client->stall_rem || client->metrics.events_sent!=client->stall_events_sent ||
1110 0 : client->metrics.last_acked_id!=client->stall_acked_id ) {
1111 0 : client->stall_since = now;
1112 0 : client->stall_rem = rem;
1113 0 : client->stall_events_sent = client->metrics.events_sent;
1114 0 : client->stall_acked_id = client->metrics.last_acked_id;
1115 0 : return 0;
1116 0 : }
1117 0 : if( FD_LIKELY( now-client->stall_since<=FD_EVENT_CLIENT_CREDIT_STALL_NANOS ) ) return 0;
1118 0 : FD_LOG_WARNING(( "telemetry server stopped granting flow-control credit for over %ld seconds, reconnecting", FD_EVENT_CLIENT_CREDIT_STALL_NANOS/(long)1e9 ));
1119 0 : client->metrics.credit_stall_cnt++;
1120 0 : disconnect( client, now, DISCONNECT_REASON_TIMEOUT, 0, 1 );
1121 0 : return 1;
1122 0 : }
1123 :
1124 : static void
1125 : poll1( fd_event_client_t * client,
1126 : long now,
1127 0 : int * charge_busy ) {
1128 0 : if( FD_UNLIKELY( !client->has_genesis_hash || !client->has_shred_version ) ) return;
1129 :
1130 0 : if( FD_UNLIKELY( client->state==FD_EVENT_CLIENT_STATE_DISCONNECTED ) ) reconnect( client, now, charge_busy );
1131 0 : if( FD_UNLIKELY( client->state==FD_EVENT_CLIENT_STATE_CONNECTING ) ) {
1132 0 : if( FD_UNLIKELY( now>client->connecting.connect_deadline ) ) {
1133 0 : disconnect( client, now, DISCONNECT_REASON_TIMEOUT, 0, 1 );
1134 0 : return;
1135 0 : }
1136 0 : }
1137 : /* Check auth handshake timeout */
1138 0 : if( FD_UNLIKELY( client->state==FD_EVENT_CLIENT_STATE_AUTHENTICATING && now>client->auth_deadline ) ) {
1139 0 : FD_LOG_WARNING(( "auth handshake timed out" ));
1140 0 : client->metrics.handshake_timeout_cnt++;
1141 0 : disconnect( client, now, DISCONNECT_REASON_TIMEOUT, 0, 1 );
1142 0 : return;
1143 0 : }
1144 0 : if( FD_LIKELY( client->state!=FD_EVENT_CLIENT_STATE_DISCONNECTED ) ) {
1145 0 : int rxtx_err;
1146 0 : if( client->use_tls )
1147 0 : rxtx_err = fd_grpc_client_rxtx_tls( client->grpc_client, client->tls_conn, client->sockfd, now, charge_busy );
1148 0 : else
1149 0 : rxtx_err = fd_grpc_client_rxtx_socket( client->grpc_client, client->sockfd, now, charge_busy );
1150 0 : if( FD_UNLIKELY( -1==rxtx_err ) ) {
1151 0 : disconnect( client, now, DISCONNECT_REASON_TRANSPORT_FAILED, errno, 1 );
1152 0 : return;
1153 0 : }
1154 0 : }
1155 :
1156 0 : if( FD_UNLIKELY( client->defer_disconnect!=INT_MAX ) ) {
1157 0 : int reason = client->defer_disconnect;
1158 0 : if( reason==DISCONNECT_REASON_AUTH_FAILED ) client->metrics.auth_fail_cnt++;
1159 0 : if( reason==DISCONNECT_REASON_INVALID_PROTOBUF ) client->metrics.invalid_msg_cnt++;
1160 0 : disconnect( client, now, reason, 0, 1 );
1161 0 : return;
1162 0 : }
1163 :
1164 0 : if( FD_UNLIKELY( client->auth_send_pending ) ) fd_event_client_try_send_authenticate( client, now );
1165 :
1166 0 : if( FD_LIKELY( client->state==FD_EVENT_CLIENT_STATE_CONNECTED ) ) {
1167 0 : if( FD_UNLIKELY( client->event_stream && now-client->last_response_ns>FD_EVENT_CLIENT_RESPONSE_TIMEOUT_NANOS ) ) {
1168 0 : FD_LOG_WARNING(( "no response from telemetry server in over %ld seconds", FD_EVENT_CLIENT_RESPONSE_TIMEOUT_NANOS/(long)1e9 ));
1169 0 : disconnect( client, now, DISCONNECT_REASON_TIMEOUT, 0, 1 );
1170 0 : return;
1171 0 : }
1172 0 : if( FD_UNLIKELY( credit_stall_check( client, now ) ) ) return;
1173 0 : tx( client, now, charge_busy );
1174 0 : }
1175 0 : }
1176 :
1177 : static int
1178 : tx_flush( fd_event_client_t * client,
1179 0 : long now ) {
1180 0 : if( FD_LIKELY( client->state==FD_EVENT_CLIENT_STATE_DISCONNECTED || !fd_grpc_client_tx_pending( client->grpc_client ) ) ) return 0;
1181 0 : int flush_err = client->use_tls ? fd_grpc_client_tls_flush ( client->grpc_client, client->sockfd )
1182 0 : : fd_grpc_client_tx_flush_socket( client->grpc_client, client->sockfd );
1183 0 : if( FD_UNLIKELY( -1==flush_err ) ) {
1184 0 : disconnect( client, now, DISCONNECT_REASON_TRANSPORT_FAILED, errno, 1 );
1185 0 : return -1;
1186 0 : }
1187 0 : return 0;
1188 0 : }
1189 :
1190 : void
1191 : fd_event_client_poll( fd_event_client_t * client,
1192 : long now,
1193 0 : int * charge_busy ) {
1194 0 : client->now = now;
1195 0 : poll1( client, now, charge_busy );
1196 :
1197 0 : for( ulong i=0UL; i<FD_EVENT_CLIENT_TX_DRAIN_MAX; i++ ) {
1198 0 : if( FD_UNLIKELY( client->state!=FD_EVENT_CLIENT_STATE_CONNECTED ) ) break;
1199 0 : if( FD_UNLIKELY( fd_grpc_client_stream_send_is_blocked( client->grpc_client ) ) ) {
1200 0 : if( FD_UNLIKELY( tx_flush( client, now ) ) ) return;
1201 0 : if( FD_UNLIKELY( fd_grpc_client_stream_send_is_blocked( client->grpc_client ) ) ) break;
1202 0 : }
1203 0 : if( FD_LIKELY( !fd_circq_unsent_cnt( client->circq ) ) ) break;
1204 0 : if( FD_UNLIKELY( client->tx_tokens<=0L ) ) break;
1205 0 : int progress = 0;
1206 0 : tx( client, now, &progress );
1207 0 : if( FD_UNLIKELY( !progress ) ) break;
1208 0 : *charge_busy = 1;
1209 0 : }
1210 :
1211 0 : if( FD_UNLIKELY( tx_flush( client, now ) ) ) return;
1212 :
1213 : /* Arm EPOLLOUT only on write demand: unsent HTTP/2 bytes, or a TLS
1214 : handshake blocked on write (which is also what an in-progress
1215 : TCP connect looks like: the handshake write EAGAINed, and connect
1216 : completion is the writability event). Anything else leaves an
1217 : idle-writable socket keeping the epoll set ready. */
1218 0 : int want = fd_grpc_client_tx_pending( client->grpc_client );
1219 0 : want |= client->use_tls && fd_grpc_client_tls_tx_pending( client->grpc_client );
1220 0 : epoll_out( client, want );
1221 0 : }
1222 :
1223 : fd_grpc_client_callbacks_t fd_event_client_grpc_callbacks = {
1224 : .conn_established = fd_event_client_grpc_conn_established,
1225 : .conn_dead = fd_event_client_grpc_conn_dead,
1226 : .tx_complete = fd_event_client_grpc_tx_complete,
1227 : .rx_start = fd_event_client_grpc_rx_start,
1228 : .rx_msg = fd_event_client_grpc_rx_msg,
1229 : .rx_end = fd_event_client_grpc_rx_end,
1230 : .rx_timeout = fd_event_client_grpc_rx_timeout,
1231 : .ping_ack = fd_event_client_grpc_ping_ack,
1232 : };
|
