Line data Source code
1 : #define _GNU_SOURCE
2 : #include "utils/fd_ssarchive.h"
3 : #include "utils/fd_ssctrl.h"
4 : #include "utils/fd_sshttp.h"
5 : #include "utils/fd_sspeer_selector.h"
6 : #include "utils/fd_wfs.h"
7 :
8 : #include "../../disco/topo/fd_topo.h"
9 : #include "../../disco/metrics/fd_metrics.h"
10 : #include "../../disco/waker/fd_waker.h"
11 : #include "../../disco/fd_clock_tile.h"
12 :
13 : #include <sys/mman.h> /* memfd_create */
14 : #include <errno.h>
15 : #include <fcntl.h>
16 : #include <unistd.h>
17 : #include <sys/socket.h>
18 : #include <sys/uio.h>
19 :
20 : #include <linux/futex.h>
21 : #include "generated/fd_snapld_tile_seccomp.h"
22 :
23 : #define NAME "snapld"
24 :
25 : /* download progress in each 10 second window must be at
26 : min_download_speed_mibs * 10 seconds or higher. Catches extremely
27 : slow download speeds where we may not get to 100 MiB downloaded for a
28 : while. */
29 0 : #define FD_SNAPLD_DOWNLOAD_WINDOW_NS (10L*1000L*1000L*1000L) /* 10 seconds */
30 :
31 0 : #define FD_SNAPLD_RX_BURST (16UL) /* data frags per receive syscall */
32 :
33 : /* The snapld tile is responsible for loading data from the local file
34 : or from an HTTP/TCP connection and sending it to the snapdc tile
35 : for later decompression. */
36 :
37 : typedef struct fd_snapld_tile {
38 :
39 : struct {
40 : char path[ PATH_MAX ];
41 : uint min_download_speed_mibs;
42 : } config;
43 :
44 : int state;
45 : int pipeline_ready;
46 : int load_full;
47 : int load_file;
48 : int sent_meta;
49 : int is_redirect;
50 : ulong gossip_slot;
51 : ulong file_sz;
52 :
53 : ulong bytes_in_batch;
54 : double download_speed_mibs;
55 : long start_batch;
56 : long end_batch;
57 :
58 : ulong bytes_in_window;
59 : ulong min_bytes_in_window;
60 : long window_deadline;
61 :
62 : int local_full_fd;
63 : int local_incr_fd;
64 : int sockfd;
65 :
66 : ulong waker_client_idx;
67 : ulong * waker_fseq;
68 :
69 : fd_clock_tile_t clock[1];
70 :
71 : fd_sshttp_t * sshttp;
72 :
73 : struct {
74 : void const * base;
75 : } in_rd;
76 :
77 : struct {
78 : fd_wksp_t * mem;
79 : ulong chunk0;
80 : ulong wmark;
81 : ulong chunk;
82 : ulong mtu;
83 : } out_dc;
84 :
85 : } fd_snapld_tile_t;
86 :
87 : static ulong
88 0 : scratch_align( void ) {
89 0 : ulong a = alignof(fd_snapld_tile_t);
90 0 : a = fd_ulong_max( a, fd_sshttp_align() );
91 0 : a = fd_ulong_max( a, fd_alloc_align() );
92 0 : return a;
93 0 : }
94 :
95 : static ulong
96 0 : scratch_footprint( fd_topo_tile_t const * tile FD_PARAM_UNUSED ) {
97 0 : ulong l = FD_LAYOUT_INIT;
98 0 : l = FD_LAYOUT_APPEND( l, alignof(fd_snapld_tile_t), sizeof(fd_snapld_tile_t) );
99 0 : l = FD_LAYOUT_APPEND( l, fd_sshttp_align(), fd_sshttp_footprint() );
100 0 : l = FD_LAYOUT_APPEND( l, fd_alloc_align(), fd_alloc_footprint() );
101 0 : return FD_LAYOUT_FINI( l, scratch_align() );
102 0 : }
103 :
104 :
105 : static void
106 : privileged_init( fd_topo_t const * topo,
107 0 : fd_topo_tile_t const * tile ) {
108 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
109 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
110 0 : fd_snapld_tile_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_snapld_tile_t), sizeof(fd_snapld_tile_t) );
111 0 : void * _sshttp = FD_SCRATCH_ALLOC_APPEND( l, fd_sshttp_align(), fd_sshttp_footprint() );
112 :
113 0 : ctx->sshttp = fd_sshttp_join( fd_sshttp_new( _sshttp, FD_WAKER_INNER_FD( tile->waker_client_idx ) ) );
114 0 : FD_TEST( ctx->sshttp );
115 :
116 0 : ulong full_slot = ULONG_MAX;
117 0 : ulong incr_slot = ULONG_MAX;
118 0 : int full_is_zstd = 0;
119 0 : int incr_is_zstd = 0;
120 0 : char full_path[ PATH_MAX ] = { 0 };
121 0 : char incr_path[ PATH_MAX ] = { 0 };
122 0 : uchar full_snapshot_hash[ FD_HASH_FOOTPRINT ] = { 0 };
123 0 : uchar incr_snapshot_hash[ FD_HASH_FOOTPRINT ] = { 0 };
124 0 : ctx->local_full_fd = -1;
125 0 : ctx->local_incr_fd = -1;
126 : /* fd_ssarchive_latest_best needs to be invoked here, irrespective
127 : of whether snapct may do the same, because this information is
128 : needed here during privileged_init. */
129 0 : int found = 0==fd_ssarchive_latest_best( tile->snapld.snapshots_path,
130 0 : tile->snapld.incremental_snapshots,
131 0 : fd_wfs_configured( tile->snapld.wfs_slot,
132 0 : tile->snapld.wfs_hash_is_zero,
133 0 : (ulong)tile->snapld.wfs_shred_version )
134 0 : ? tile->snapld.wfs_slot : 0UL,
135 0 : &full_slot, &incr_slot,
136 0 : full_path, incr_path,
137 0 : &full_is_zstd, &incr_is_zstd,
138 0 : full_snapshot_hash, incr_snapshot_hash );
139 :
140 0 : if( FD_LIKELY( found ) ) {
141 0 : FD_TEST( full_slot!=ULONG_MAX );
142 :
143 0 : ctx->local_full_fd = open( full_path, O_RDONLY|O_CLOEXEC|O_NONBLOCK );
144 0 : if( FD_UNLIKELY( -1==ctx->local_full_fd ) ) FD_LOG_ERR(( "open() failed `%s` (%i-%s)", full_path, errno, fd_io_strerror( errno ) ));
145 0 : posix_fadvise( ctx->local_full_fd, 0L, 0L, POSIX_FADV_SEQUENTIAL );
146 :
147 0 : if( FD_LIKELY( incr_slot!=ULONG_MAX ) ) {
148 0 : ctx->local_incr_fd = open( incr_path, O_RDONLY|O_CLOEXEC|O_NONBLOCK );
149 0 : if( FD_UNLIKELY( -1==ctx->local_incr_fd ) ) FD_LOG_ERR(( "open() failed `%s` (%i-%s)", incr_path, errno, fd_io_strerror( errno ) ));
150 0 : posix_fadvise( ctx->local_incr_fd, 0L, 0L, POSIX_FADV_SEQUENTIAL );
151 0 : }
152 0 : }
153 :
154 : /* Load CA trust store while we still have filesystem access
155 : (before seccomp sandbox locks down). */
156 0 : fd_sshttp_load_ca_store( ctx->sshttp );
157 :
158 : /* Create a temporary file descriptor for our socket file descriptor.
159 : It is closed later in unprivileged init so that the sandbox sees
160 : an existent file descriptor. */
161 0 : ctx->sockfd = memfd_create( "snapld.sockfd", 0 );
162 0 : if( FD_UNLIKELY( -1==ctx->sockfd ) ) FD_LOG_ERR(( "memfd_create() failed (%i-%s)", errno, fd_io_strerror( errno ) ));
163 0 : }
164 :
165 : static ulong
166 : populate_allowed_fds( fd_topo_t const * topo,
167 : fd_topo_tile_t const * tile,
168 : ulong out_fds_cnt,
169 0 : int * out_fds ) {
170 0 : if( FD_UNLIKELY( out_fds_cnt<7UL ) ) FD_LOG_ERR(( "out_fds_cnt %lu", out_fds_cnt ));
171 :
172 0 : ulong out_cnt = 0;
173 0 : out_fds[ out_cnt++ ] = 2UL; /* stderr */
174 0 : if( FD_LIKELY( -1!=fd_log_private_logfile_fd() ) ) {
175 0 : out_fds[ out_cnt++ ] = fd_log_private_logfile_fd();
176 0 : }
177 0 : out_fds[ out_cnt++ ] = FD_WAKER_OUTER_FD; /* waker outer epoll fd (rearm) */
178 0 : out_fds[ out_cnt++ ] = FD_WAKER_INNER_FD( tile->waker_client_idx ); /* waker inner epoll fd */
179 :
180 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
181 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
182 0 : fd_snapld_tile_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_snapld_tile_t), sizeof(fd_snapld_tile_t) );
183 0 : if( FD_LIKELY( -1!=ctx->local_full_fd ) ) out_fds[ out_cnt++ ] = ctx->local_full_fd;
184 0 : if( FD_LIKELY( -1!=ctx->local_incr_fd ) ) out_fds[ out_cnt++ ] = ctx->local_incr_fd;
185 0 : out_fds[ out_cnt++ ] = ctx->sockfd;
186 :
187 0 : return out_cnt;
188 0 : }
189 :
190 : static ulong
191 : populate_allowed_seccomp( fd_topo_t const * topo,
192 : fd_topo_tile_t const * tile,
193 : ulong out_cnt,
194 0 : struct sock_filter * out ) {
195 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
196 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
197 0 : fd_snapld_tile_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_snapld_tile_t), sizeof(fd_snapld_tile_t) );
198 :
199 0 : uint epoll_inner_fd = (uint)FD_WAKER_INNER_FD( tile->waker_client_idx );
200 0 : uint epoll_outer_fd = (uint)FD_WAKER_OUTER_FD;
201 0 : populate_sock_filter_policy_fd_snapld_tile( out_cnt, out, (uint)fd_log_private_logfile_fd(), (uint)ctx->local_full_fd, (uint)ctx->local_incr_fd, (uint)ctx->sockfd, epoll_inner_fd, epoll_outer_fd );
202 0 : return sock_filter_policy_fd_snapld_tile_instr_cnt;
203 0 : }
204 :
205 : static void
206 : unprivileged_init( fd_topo_t const * topo,
207 0 : fd_topo_tile_t const * tile ) {
208 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
209 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
210 0 : fd_snapld_tile_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_snapld_tile_t), sizeof(fd_snapld_tile_t) );
211 0 : FD_SCRATCH_ALLOC_APPEND( l, fd_sshttp_align(), fd_sshttp_footprint() );
212 0 : FD_SCRATCH_ALLOC_APPEND( l, fd_alloc_align(), fd_alloc_footprint() );
213 :
214 0 : fd_memcpy( ctx->config.path, tile->snapld.snapshots_path, PATH_MAX );
215 0 : ctx->config.min_download_speed_mibs = tile->snapld.min_download_speed_mibs;
216 :
217 0 : ctx->state = FD_SNAPSHOT_STATE_IDLE;
218 0 : ctx->pipeline_ready = 0;
219 :
220 0 : ctx->waker_client_idx = tile->waker_client_idx;
221 0 : FD_TEST( ctx->waker_client_idx!=ULONG_MAX );
222 0 : ctx->waker_fseq = fd_fseq_join( fd_topo_obj_laddr( topo, tile->waker_fseq_obj_id ) );
223 0 : FD_TEST( ctx->waker_fseq );
224 0 : fd_clock_tile_init( ctx->clock );
225 :
226 0 : ctx->download_speed_mibs = 0.0;
227 0 : ctx->bytes_in_batch = 0UL;
228 0 : ctx->start_batch = 0L;
229 0 : ctx->end_batch = 0L;
230 0 : ctx->bytes_in_window = 0UL;
231 0 : ctx->window_deadline = LONG_MAX;
232 0 : ctx->min_bytes_in_window = ((ulong)ctx->config.min_download_speed_mibs * (FD_SNAPLD_DOWNLOAD_WINDOW_NS / (ulong)1e9))<<20UL;
233 :
234 0 : FD_TEST( tile->in_cnt==1UL );
235 0 : fd_topo_link_t const * in_link = &topo->links[ tile->in_link_id[ 0 ] ];
236 0 : FD_TEST( 0==strcmp( in_link->name, "snapct_ld" ) );
237 0 : ctx->in_rd.base = fd_topo_obj_wksp_base( topo, in_link->dcache_obj_id );
238 :
239 0 : FD_TEST( tile->out_cnt==1UL );
240 0 : fd_topo_link_t const * out_link = &topo->links[ tile->out_link_id[ 0 ] ];
241 0 : FD_TEST( 0==strcmp( out_link->name, "snapld_dc" ) );
242 0 : ctx->out_dc.mem = fd_topo_obj_wksp_base( topo, out_link->dcache_obj_id );
243 0 : ctx->out_dc.chunk0 = fd_dcache_compact_chunk0( ctx->out_dc.mem, out_link->dcache );
244 0 : ctx->out_dc.wmark = fd_dcache_compact_wmark ( ctx->out_dc.mem, out_link->dcache, out_link->mtu );
245 0 : ctx->out_dc.chunk = ctx->out_dc.chunk0;
246 0 : ctx->out_dc.mtu = out_link->mtu;
247 :
248 0 : FD_TEST( sizeof(fd_ssctrl_meta_t)<=ctx->out_dc.mtu );
249 0 : FD_TEST( fd_dcache_compact_next( 0UL, ctx->out_dc.mtu, 0UL, ULONG_MAX )*FD_CHUNK_SZ==ctx->out_dc.mtu );
250 :
251 : /* We can only close the temporary socket file descriptor after
252 : entering the sandbox because the sandbox checks all file
253 : descriptors are existent. */
254 0 : if( -1==close( ctx->sockfd ) ) FD_LOG_ERR((" close() failed (%i-%s)", errno, fd_io_strerror( errno ) ));
255 :
256 0 : ulong scratch_top = FD_SCRATCH_ALLOC_FINI( l, scratch_align() );
257 0 : if( FD_UNLIKELY( scratch_top > (ulong)scratch + scratch_footprint( tile ) ) )
258 0 : FD_LOG_ERR(( "scratch overflow %lu %lu %lu", scratch_top - (ulong)scratch - scratch_footprint( tile ), scratch_top, (ulong)scratch + scratch_footprint( tile ) ));
259 0 : }
260 :
261 : static int
262 0 : should_shutdown( fd_snapld_tile_t * ctx ) {
263 0 : return ctx->state==FD_SNAPSHOT_STATE_SHUTDOWN;
264 0 : }
265 :
266 : static void
267 0 : during_housekeeping( fd_snapld_tile_t * ctx ) {
268 0 : if( FD_UNLIKELY( fd_clock_tile_recal_due( ctx->clock ) ) ) fd_clock_tile_recal( ctx->clock );
269 0 : }
270 :
271 : static long
272 0 : next_deadline( fd_snapld_tile_t * ctx ) {
273 0 : if( FD_LIKELY( ctx->state!=FD_SNAPSHOT_STATE_PROCESSING || ctx->load_file ) ) return LONG_MAX;
274 0 : long next = fd_long_min( fd_sshttp_deadline( ctx->sshttp ), ctx->window_deadline );
275 0 : return next==LONG_MAX ? LONG_MAX : fd_clock_tile_wallclock_to_tickcount( ctx->clock, next );
276 0 : }
277 :
278 : static void
279 0 : metrics_write( fd_snapld_tile_t * ctx ) {
280 0 : FD_MGAUGE_SET( SNAPLD, STREAM_STATE, (ulong)(ctx->state) );
281 0 : }
282 :
283 : static void
284 : transition_malformed( fd_snapld_tile_t * ctx,
285 0 : fd_stem_context_t * stem ) {
286 0 : if( FD_UNLIKELY( ctx->state==FD_SNAPSHOT_STATE_ERROR ) ) return;
287 0 : ctx->state = FD_SNAPSHOT_STATE_ERROR;
288 0 : fd_stem_publish( stem, 0UL, FD_SNAPSHOT_MSG_CTRL_ERROR, 0UL, 0UL, 0UL, 0UL, 0UL );
289 0 : }
290 :
291 : static int
292 : check_download_progress( fd_snapld_tile_t * ctx,
293 : fd_stem_context_t * stem,
294 : int downloading,
295 0 : long now ) {
296 0 : if( FD_UNLIKELY( ctx->window_deadline==LONG_MAX && downloading ) ) {
297 0 : ctx->window_deadline = now + FD_SNAPLD_DOWNLOAD_WINDOW_NS;
298 0 : ctx->bytes_in_window = 0UL;
299 0 : }
300 :
301 0 : if( FD_UNLIKELY( now>ctx->window_deadline ) ) {
302 0 : if( FD_UNLIKELY( ctx->bytes_in_window<ctx->min_bytes_in_window ) ) {
303 : /* cancel the download if the download progress speed in the last
304 : window is less than the minimum download speed. */
305 0 : double download_speed_mibs = (double)ctx->bytes_in_window / (double)(FD_SNAPLD_DOWNLOAD_WINDOW_NS / 1e9) / (double)(1<<20UL);
306 0 : FD_LOG_WARNING(( "download progress of %.2f MiB/s in the last %lu seconds for %s snapshot "
307 0 : "is below the minimum download speed %u MiB/s, cancelling download",
308 0 : download_speed_mibs, FD_SNAPLD_DOWNLOAD_WINDOW_NS / (ulong)1e9,
309 0 : ctx->load_full ? "full" : "incremental", ctx->config.min_download_speed_mibs ));
310 0 : transition_malformed( ctx, stem );
311 0 : fd_sshttp_cancel( ctx->sshttp );
312 0 : return -1;
313 0 : }
314 0 : ctx->window_deadline = now + FD_SNAPLD_DOWNLOAD_WINDOW_NS;
315 0 : ctx->bytes_in_window = 0UL;
316 0 : }
317 0 : return 0;
318 0 : }
319 :
320 : /* rx_iov covers the next burst MTU sized frag slots with at most two
321 : iovecs, split only at the wrap to chunk0. Returns the iov count. */
322 :
323 : static ulong
324 : rx_iov( fd_snapld_tile_t * ctx,
325 : struct iovec iov[ static 2 ],
326 0 : ulong burst ) {
327 0 : ulong mtu = ctx->out_dc.mtu;
328 0 : ulong step = fd_dcache_compact_next( 0UL, mtu, 0UL, ULONG_MAX );
329 0 : ulong pre = fd_ulong_min( burst, (ctx->out_dc.wmark-ctx->out_dc.chunk)/step + 1UL );
330 0 : iov[0].iov_base = fd_chunk_to_laddr( ctx->out_dc.mem, ctx->out_dc.chunk );
331 0 : iov[0].iov_len = pre*mtu;
332 0 : iov[1].iov_base = fd_chunk_to_laddr( ctx->out_dc.mem, ctx->out_dc.chunk0 );
333 0 : iov[1].iov_len = (burst-pre)*mtu;
334 0 : return 1UL + (burst>pre);
335 0 : }
336 :
337 : /* publish_data publishes sz bytes received via rx_iov as MTU sized
338 : frags. */
339 :
340 : static void
341 : publish_data( fd_snapld_tile_t * ctx,
342 : fd_stem_context_t * stem,
343 0 : ulong sz ) {
344 0 : while( sz ) {
345 0 : ulong frag_sz = fd_ulong_min( sz, ctx->out_dc.mtu );
346 0 : fd_stem_publish( stem, 0UL, FD_SNAPSHOT_MSG_DATA, ctx->out_dc.chunk, frag_sz, 0UL, 0UL, 0UL );
347 0 : ctx->out_dc.chunk = fd_dcache_compact_next( ctx->out_dc.chunk, frag_sz, ctx->out_dc.chunk0, ctx->out_dc.wmark );
348 0 : sz -= frag_sz;
349 0 : }
350 0 : }
351 :
352 : static void
353 : after_credit( fd_snapld_tile_t * ctx,
354 : fd_stem_context_t * stem,
355 : int * opt_poll_in FD_PARAM_UNUSED,
356 0 : int * charge_busy ) {
357 0 : if( ctx->state!=FD_SNAPSHOT_STATE_PROCESSING ) {
358 0 : if( FD_LIKELY( !stem->sleep ) ) fd_log_sleep( (long)1e6 );
359 0 : return;
360 0 : }
361 :
362 0 : if( FD_UNLIKELY( !ctx->pipeline_ready ) ) {
363 0 : return;
364 0 : }
365 :
366 0 : uchar * out = fd_chunk_to_laddr( ctx->out_dc.mem, ctx->out_dc.chunk );
367 :
368 0 : if( ctx->load_file ) {
369 0 : if( FD_UNLIKELY( !ctx->sent_meta ) ) {
370 0 : FD_TEST( sizeof(fd_ssctrl_meta_t)<=ctx->out_dc.mtu );
371 0 : fd_ssctrl_meta_t * meta = (fd_ssctrl_meta_t *)out;
372 0 : meta->total_sz = ctx->file_sz;
373 0 : meta->resolved_slot = ULONG_MAX;
374 0 : fd_memset( meta->resolved_hash, 0, FD_HASH_FOOTPRINT );
375 0 : meta->resolved_name[0] = '\0';
376 0 : ctx->sent_meta = 1;
377 0 : fd_stem_publish( stem, 0UL, FD_SNAPSHOT_MSG_META, ctx->out_dc.chunk, sizeof(fd_ssctrl_meta_t), 0UL, 0UL, 0UL );
378 0 : ctx->out_dc.chunk = fd_dcache_compact_next( ctx->out_dc.chunk, sizeof(fd_ssctrl_meta_t), ctx->out_dc.chunk0, ctx->out_dc.wmark );
379 0 : return;
380 0 : }
381 0 : struct iovec iov[2];
382 0 : ulong iov_cnt = rx_iov( ctx, iov, FD_SNAPLD_RX_BURST );
383 0 : long result = readv( ctx->load_full ? ctx->local_full_fd : ctx->local_incr_fd, iov, (int)iov_cnt );
384 0 : if( FD_UNLIKELY( result<=0L ) ) {
385 0 : if( result==0L ) {
386 0 : FD_LOG_INFO(( "finished reading %s snapshot from local file", ctx->load_full ? "full" : "incremental" ));
387 0 : ctx->state = FD_SNAPSHOT_STATE_FINISHING;
388 0 : fd_stem_publish( stem, 0UL, FD_SNAPSHOT_MSG_LOAD_COMPLETE, 0UL, 0UL, 0UL, 0UL, 0UL );
389 0 : } else if( FD_UNLIKELY( errno!=EAGAIN && errno!=EINTR ) ) {
390 0 : FD_LOG_WARNING(( "read() failed on %s snapshot file (%i-%s)", ctx->load_full ? "full" : "incremental", errno, fd_io_strerror( errno ) ));
391 0 : transition_malformed( ctx, stem );
392 0 : return; /* verbose return */
393 0 : }
394 0 : } else {
395 0 : publish_data( ctx, stem, (ulong)result );
396 0 : *charge_busy = 1;
397 0 : return; /* verbose return */
398 0 : }
399 0 : } else {
400 : /* Single chunk until META is sent, see the swap below. */
401 0 : struct iovec iov[2];
402 0 : ulong iov_cnt = rx_iov( ctx, iov, ctx->sent_meta ? FD_SNAPLD_RX_BURST : 1UL );
403 0 : int downloading = 0;
404 0 : ulong data_len = 0UL;
405 0 : long now = fd_clock_tile_now( ctx->clock );
406 0 : int fired = fd_fseq_query( ctx->waker_fseq )==1UL;
407 0 : if( FD_LIKELY( fired ) ) fd_fseq_update( ctx->waker_fseq, 0UL );
408 0 : int result = fd_sshttp_advance( ctx->sshttp, iov, iov_cnt, &data_len, &downloading, now );
409 0 : if( FD_LIKELY( fired ) ) fd_waker_client_rearm( ctx->waker_client_idx );
410 0 : switch( result ) {
411 0 : case FD_SSHTTP_ADVANCE_AGAIN:
412 : /* Return value ignored: on failure, check_download_progress
413 : already calls transition_malformed and fd_sshttp_cancel. */
414 0 : check_download_progress( ctx, stem, downloading, now );
415 0 : break;
416 0 : case FD_SSHTTP_ADVANCE_DATA: {
417 0 : ctx->bytes_in_window += data_len;
418 0 : if( FD_UNLIKELY( -1==check_download_progress( ctx, stem, downloading, now ) ) ) break;
419 0 : if( FD_UNLIKELY( !ctx->sent_meta ) ) {
420 : /* On the first DATA return, the HTTP headers are available
421 : for use. We need to send this metadata downstream, but
422 : need to do so before any data frags. So, we copy any data
423 : we received with the headers (if any) to the next dcache
424 : chunk and then publish both in order. */
425 0 : ctx->start_batch = fd_clock_tile_now( ctx->clock );
426 0 : FD_TEST( sizeof(fd_ssctrl_meta_t)<=ctx->out_dc.mtu );
427 0 : fd_ssctrl_meta_t * meta = (fd_ssctrl_meta_t *)out;
428 0 : ulong next_chunk = fd_dcache_compact_next( ctx->out_dc.chunk, sizeof(fd_ssctrl_meta_t), ctx->out_dc.chunk0, ctx->out_dc.wmark );
429 0 : memmove( fd_chunk_to_laddr( ctx->out_dc.mem, next_chunk ), out, data_len );
430 0 : meta->total_sz = fd_sshttp_content_len( ctx->sshttp );
431 0 : if( FD_UNLIKELY( meta->total_sz==ULONG_MAX ) ) {
432 0 : FD_LOG_WARNING(( "HTTP response for %s snapshot is missing Content-Length header", ctx->load_full ? "full" : "incremental" ));
433 0 : transition_malformed( ctx, stem );
434 0 : fd_sshttp_cancel( ctx->sshttp );
435 0 : break;
436 0 : }
437 :
438 : /* Populate resolved redirect fields in META */
439 0 : meta->resolved_slot = ULONG_MAX;
440 0 : meta->resolved_name[0] = '\0';
441 0 : fd_memset( meta->resolved_hash, 0, FD_HASH_FOOTPRINT );
442 :
443 0 : if( ctx->is_redirect ) {
444 0 : char const * resolved_name = fd_sshttp_snapshot_name( ctx->sshttp );
445 0 : if( FD_UNLIKELY( !resolved_name || resolved_name[0]=='\0' ) ) {
446 0 : FD_LOG_WARNING(( "redirect-based download did not resolve to a snapshot filename for %s snapshot",
447 0 : ctx->load_full ? "full" : "incremental" ));
448 0 : transition_malformed( ctx, stem );
449 0 : fd_sshttp_cancel( ctx->sshttp );
450 0 : break;
451 0 : }
452 0 : int is_full_filename = !strncmp( resolved_name, "snapshot-", 9 );
453 0 : if( FD_UNLIKELY( is_full_filename!=ctx->load_full ) ) {
454 0 : FD_LOG_WARNING(( "resolved snapshot type mismatch: expected %s but got %s filename `%s`",
455 0 : ctx->load_full ? "full" : "incremental", is_full_filename ? "full" : "incremental", resolved_name ));
456 0 : transition_malformed( ctx, stem );
457 0 : fd_sshttp_cancel( ctx->sshttp );
458 0 : break;
459 0 : }
460 0 : ulong resolved_slot = fd_sshttp_resolved_slot( ctx->sshttp );
461 0 : if( FD_UNLIKELY( resolved_slot<ctx->gossip_slot ) ) {
462 0 : FD_LOG_WARNING(( "resolved snapshot slot %lu is older than gossip slot %lu for %s snapshot, rejecting",
463 0 : resolved_slot, ctx->gossip_slot, ctx->load_full ? "full" : "incremental" ));
464 0 : transition_malformed( ctx, stem );
465 0 : fd_sshttp_cancel( ctx->sshttp );
466 0 : break;
467 0 : }
468 0 : if( FD_UNLIKELY( resolved_slot>=FD_SSPEER_PLAUSIBLE_MAX_SLOT ) ) {
469 0 : FD_LOG_WARNING(( "resolved snapshot slot %lu exceeds plausibility bound for %s snapshot, rejecting",
470 0 : resolved_slot, ctx->load_full ? "full" : "incremental" ));
471 0 : transition_malformed( ctx, stem );
472 0 : fd_sshttp_cancel( ctx->sshttp );
473 0 : break;
474 0 : }
475 0 : meta->resolved_slot = resolved_slot;
476 0 : fd_memcpy( meta->resolved_hash, fd_sshttp_resolved_hash( ctx->sshttp ), FD_HASH_FOOTPRINT );
477 0 : fd_cstr_ncpy( meta->resolved_name, resolved_name, PATH_MAX );
478 0 : FD_LOG_INFO(( "redirect resolved to `%s` (slot %lu) for %s snapshot",
479 0 : resolved_name, resolved_slot, ctx->load_full ? "full" : "incremental" ));
480 0 : }
481 :
482 0 : ctx->sent_meta = 1;
483 0 : fd_stem_publish( stem, 0UL, FD_SNAPSHOT_MSG_META, ctx->out_dc.chunk, sizeof(fd_ssctrl_meta_t), 0UL, 0UL, 0UL );
484 0 : ctx->out_dc.chunk = next_chunk;
485 0 : }
486 0 : if( FD_LIKELY( data_len!=0UL ) ) {
487 0 : publish_data( ctx, stem, data_len );
488 0 : ctx->bytes_in_batch += data_len;
489 :
490 : /* measure download speed every 100 MiB */
491 0 : if(ctx->bytes_in_batch>=100<<20UL) {
492 0 : ctx->end_batch = fd_clock_tile_now( ctx->clock );
493 : /* as a precaution, make sure elapsed_batch is positive
494 : and larger than zero (to avoid division by zero). */
495 0 : long elapsed_batch = fd_long_if( ctx->end_batch > ctx->start_batch, ctx->end_batch - ctx->start_batch, 1L );
496 : /* download speed in MiB/s = bytes/nanoseconds * 1e9/(1 second) * 1/(1MiB = 1<<20UL) = 1e9/(1024*1024) ~= 954 */
497 0 : ctx->download_speed_mibs = (double)(ctx->bytes_in_batch*954) / (double)elapsed_batch;
498 0 : if( FD_UNLIKELY( ctx->download_speed_mibs<ctx->config.min_download_speed_mibs ) ) {
499 : /* cancel the snapshot load if the download speed is less
500 : than the minimum download speed. */
501 0 : FD_LOG_WARNING(( "download speed %.2f MiB/s on a batch of %lu MiB for %s snapshot is below the minimum threshold %.2f MiB/s. "
502 0 : "cancelling snapshot download",
503 0 : ctx->download_speed_mibs, ctx->bytes_in_batch>>20UL, ctx->load_full ? "full" : "incremental",
504 0 : (double)(ctx->config.min_download_speed_mibs) ));
505 0 : transition_malformed( ctx, stem );
506 0 : fd_sshttp_cancel( ctx->sshttp );
507 0 : break;
508 0 : }
509 0 : ctx->start_batch = ctx->end_batch;
510 0 : ctx->bytes_in_batch = 0UL;
511 0 : }
512 0 : }
513 0 : *charge_busy = 1;
514 0 : break;
515 0 : }
516 0 : case FD_SSHTTP_ADVANCE_DONE:
517 0 : if( FD_UNLIKELY( !ctx->sent_meta ) ) {
518 0 : FD_LOG_WARNING(( "zero-length HTTP response for %s snapshot", ctx->load_full ? "full" : "incremental" ));
519 0 : transition_malformed( ctx, stem );
520 0 : fd_sshttp_cancel( ctx->sshttp );
521 0 : break;
522 0 : }
523 0 : FD_LOG_INFO(( "finished downloading %s snapshot", ctx->load_full ? "full" : "incremental" ));
524 0 : ctx->state = FD_SNAPSHOT_STATE_FINISHING;
525 0 : fd_stem_publish( stem, 0UL, FD_SNAPSHOT_MSG_LOAD_COMPLETE, 0UL, 0UL, 0UL, 0UL, 0UL );
526 0 : break;
527 0 : case FD_SSHTTP_ADVANCE_ERROR:
528 0 : FD_LOG_WARNING(( "HTTP advance error during %s snapshot download, entering error state",
529 0 : ctx->load_full ? "full" : "incremental" ));
530 0 : transition_malformed( ctx, stem );
531 0 : fd_sshttp_cancel( ctx->sshttp );
532 0 : break;
533 0 : default: FD_LOG_ERR(( "unexpected fd_sshttp_advance result %d for %s snapshot",
534 0 : result, ctx->load_full ? "full" : "incremental" ));
535 0 : }
536 0 : }
537 0 : }
538 :
539 : static int
540 : returnable_frag( fd_snapld_tile_t * ctx,
541 : ulong in_idx FD_PARAM_UNUSED,
542 : ulong seq FD_PARAM_UNUSED,
543 : ulong sig,
544 : ulong chunk,
545 : ulong sz,
546 : ulong ctl FD_PARAM_UNUSED,
547 : ulong tsorig FD_PARAM_UNUSED,
548 : ulong tspub FD_PARAM_UNUSED,
549 0 : fd_stem_context_t * stem ) {
550 0 : if( ctx->state==FD_SNAPSHOT_STATE_ERROR && sig!=FD_SNAPSHOT_MSG_CTRL_FAIL ) {
551 : /* Control messages move along the snapshot load pipeline. Since
552 : error conditions can be triggered by any tile in the pipeline,
553 : it is possible to be in error state and still receive otherwise
554 : valid messages. Only a fail message can revert this. */
555 0 : return 0;
556 0 : };
557 :
558 0 : int forward_msg = 1;
559 :
560 0 : switch( sig ) {
561 :
562 0 : case FD_SNAPSHOT_MSG_CTRL_INIT_FULL:
563 0 : case FD_SNAPSHOT_MSG_CTRL_INIT_INCR: {
564 0 : FD_TEST( ctx->state==FD_SNAPSHOT_STATE_IDLE );
565 0 : ctx->state = FD_SNAPSHOT_STATE_PROCESSING;
566 0 : ctx->pipeline_ready = 0;
567 0 : FD_TEST( sz==sizeof(fd_ssctrl_init_t) && sz<=ctx->out_dc.mtu );
568 0 : fd_ssctrl_init_t const * msg_in = fd_chunk_to_laddr_const( ctx->in_rd.base, chunk );
569 0 : ctx->load_full = sig==FD_SNAPSHOT_MSG_CTRL_INIT_FULL;
570 0 : ctx->load_file = msg_in->file;
571 0 : ctx->sent_meta = 0;
572 0 : ctx->gossip_slot = msg_in->slot;
573 0 : ctx->is_redirect = msg_in->is_redirect;
574 0 : ctx->file_sz = msg_in->file_sz;
575 :
576 0 : ctx->window_deadline = LONG_MAX;
577 0 : ctx->bytes_in_window = 0UL;
578 0 : ctx->bytes_in_batch = 0UL;
579 0 : if( ctx->load_file ) {
580 0 : if( FD_UNLIKELY( 0!=lseek( ctx->load_full ? ctx->local_full_fd : ctx->local_incr_fd, 0, SEEK_SET ) ) )
581 0 : FD_LOG_ERR(( "lseek(0) failed on %s snapshot file (%i-%s)",
582 0 : ctx->load_full ? "full" : "incremental", errno, fd_io_strerror( errno ) ));
583 0 : }
584 0 : fd_ssctrl_init_t * msg_out = fd_chunk_to_laddr( ctx->out_dc.mem, ctx->out_dc.chunk );
585 0 : fd_memcpy( msg_out, msg_in, sz );
586 0 : fd_stem_publish( stem, 0UL, sig, ctx->out_dc.chunk, sz, 0UL, 0UL, 0UL );
587 0 : ctx->out_dc.chunk = fd_dcache_compact_next( ctx->out_dc.chunk, ctx->out_dc.mtu, ctx->out_dc.chunk0, ctx->out_dc.wmark );
588 0 : forward_msg = 0; // we are forwarding the control message in the `fd_sstrl_init_t` message
589 0 : break;
590 0 : }
591 :
592 0 : case FD_SNAPSHOT_MSG_CTRL_START: {
593 0 : FD_TEST( ctx->state==FD_SNAPSHOT_STATE_PROCESSING );
594 0 : if( !ctx->load_file ) {
595 0 : FD_TEST( sz==sizeof(fd_ssctrl_start_t) );
596 0 : fd_ssctrl_start_t const * msg = fd_chunk_to_laddr_const( ctx->in_rd.base, chunk );
597 0 : if( FD_UNLIKELY( fd_sshttp_init( ctx->sshttp, msg->addr, msg->hostname, msg->is_https, msg->path, msg->path_len, 4UL, fd_clock_tile_now( ctx->clock ) ) ) ) {
598 0 : transition_malformed( ctx, stem );
599 0 : forward_msg = 0;
600 0 : break;
601 0 : }
602 0 : }
603 0 : ctx->pipeline_ready = 1;
604 0 : forward_msg = 0;
605 0 : break;
606 0 : }
607 :
608 0 : case FD_SNAPSHOT_MSG_CTRL_FINI: {
609 0 : FD_TEST( ctx->state==FD_SNAPSHOT_STATE_FINISHING );
610 0 : break;
611 0 : }
612 :
613 0 : case FD_SNAPSHOT_MSG_CTRL_NEXT:
614 0 : case FD_SNAPSHOT_MSG_CTRL_DONE: {
615 0 : FD_TEST( ctx->state==FD_SNAPSHOT_STATE_FINISHING );
616 0 : ctx->state = FD_SNAPSHOT_STATE_IDLE;
617 0 : break;
618 0 : }
619 :
620 0 : case FD_SNAPSHOT_MSG_CTRL_ERROR: {
621 0 : FD_TEST( ctx->state!=FD_SNAPSHOT_STATE_SHUTDOWN );
622 0 : fd_sshttp_cancel( ctx->sshttp );
623 0 : ctx->state = FD_SNAPSHOT_STATE_ERROR;
624 0 : break;
625 0 : }
626 :
627 0 : case FD_SNAPSHOT_MSG_CTRL_FAIL:
628 0 : FD_TEST( ctx->state!=FD_SNAPSHOT_STATE_SHUTDOWN );
629 0 : fd_sshttp_cancel( ctx->sshttp );
630 0 : ctx->state = FD_SNAPSHOT_STATE_IDLE;
631 0 : ctx->pipeline_ready = 0;
632 0 : break;
633 :
634 0 : case FD_SNAPSHOT_MSG_CTRL_SHUTDOWN: {
635 0 : FD_TEST( ctx->state==FD_SNAPSHOT_STATE_IDLE );
636 0 : ctx->state = FD_SNAPSHOT_STATE_SHUTDOWN;
637 0 : break;
638 0 : }
639 :
640 : /* FD_SNAPSHOT_MSG_DATA is not possible */
641 0 : default: {
642 0 : FD_LOG_ERR(( "unexpected control frag %s (%lu) in state %s (%lu)",
643 0 : fd_ssctrl_msg_ctrl_str( sig ), sig,
644 0 : fd_ssctrl_state_str( (ulong)ctx->state ), (ulong)ctx->state ));
645 0 : break;
646 0 : }
647 0 : }
648 :
649 : /* Forward the control message down the pipeline */
650 0 : if( FD_LIKELY( forward_msg ) ) {
651 0 : fd_stem_publish( stem, 0UL, sig, 0UL, 0UL, 0UL, 0UL, 0UL );
652 0 : }
653 :
654 0 : return 0;
655 0 : }
656 :
657 0 : #define STEM_BURST (FD_SNAPLD_RX_BURST+2UL)
658 :
659 0 : #define STEM_LAZY (128L*3000L)
660 :
661 0 : #define STEM_CALLBACK_CONTEXT_TYPE fd_snapld_tile_t
662 0 : #define STEM_CALLBACK_CONTEXT_ALIGN alignof(fd_snapld_tile_t)
663 :
664 0 : #define STEM_CALLBACK_SHOULD_SHUTDOWN should_shutdown
665 0 : #define STEM_CALLBACK_DURING_HOUSEKEEPING during_housekeeping
666 0 : #define STEM_CALLBACK_NEXT_DEADLINE next_deadline
667 0 : #define STEM_CALLBACK_METRICS_WRITE metrics_write
668 0 : #define STEM_CALLBACK_AFTER_CREDIT after_credit
669 0 : #define STEM_CALLBACK_RETURNABLE_FRAG returnable_frag
670 :
671 : #include "../../disco/stem/fd_stem.c"
672 :
673 : fd_topo_run_tile_t fd_tile_snapld = {
674 : .name = NAME,
675 : .populate_allowed_seccomp = populate_allowed_seccomp,
676 : .populate_allowed_fds = populate_allowed_fds,
677 : .scratch_align = scratch_align,
678 : .scratch_footprint = scratch_footprint,
679 : .privileged_init = privileged_init,
680 : .unprivileged_init = unprivileged_init,
681 : .run = stem_run,
682 : .keep_host_networking = 1,
683 : .allow_connect = 1,
684 : .rlimit_file_cnt = 5UL, /* stderr, log, http, full/incr local files */
685 : };
686 :
687 : #undef NAME
|
