@@ -320,7 +320,10 @@ impl Database {
320
320
321
321
#[ cfg( feature = "replication" ) ]
322
322
/// Sync with primary at least to a given replication index
323
- pub async fn sync_until ( & self , replication_index : FrameNo ) -> Result < crate :: database:: Replicated > {
323
+ pub async fn sync_until (
324
+ & self ,
325
+ replication_index : FrameNo ,
326
+ ) -> Result < crate :: database:: Replicated > {
324
327
if let Some ( ctx) = & self . replication_ctx {
325
328
let mut frame_no: Option < FrameNo > = ctx. replicator . committed_frame_no ( ) . await ;
326
329
let mut frames_synced: usize = 0 ;
@@ -384,15 +387,17 @@ impl Database {
384
387
let conn = self . connect ( ) ?;
385
388
386
389
let page_size = {
387
- let rows = conn. query ( "PRAGMA page_size" , crate :: params:: Params :: None ) ?. unwrap ( ) ;
390
+ let rows = conn
391
+ . query ( "PRAGMA page_size" , crate :: params:: Params :: None ) ?
392
+ . unwrap ( ) ;
388
393
let row = rows. next ( ) ?. unwrap ( ) ;
389
394
let page_size = row. get :: < u32 > ( 0 ) ?;
390
395
page_size
391
396
} ;
392
397
393
398
let mut max_frame_no: std:: os:: raw:: c_uint = 0 ;
394
399
unsafe { libsql_sys:: ffi:: libsql_wal_frame_count ( conn. handle ( ) , & mut max_frame_no) } ;
395
-
400
+
396
401
let generation = 1 ; // TODO: Probe from WAL.
397
402
let start_frame_no = sync_ctx. durable_frame_num + 1 ;
398
403
let end_frame_no = max_frame_no;
@@ -402,44 +407,81 @@ impl Database {
402
407
// The server returns its maximum frame number. To avoid resending
403
408
// frames the server already knows about, we need to update the
404
409
// frame number to the one returned by the server.
405
- let max_frame_no = self . push_one_frame ( & conn, & sync_ctx, generation, frame_no, page_size) . await ?;
410
+ let max_frame_no = self
411
+ . push_one_frame ( & conn, & sync_ctx, generation, frame_no, page_size)
412
+ . await ?;
406
413
if max_frame_no > frame_no {
407
414
frame_no = max_frame_no;
408
415
}
409
416
frame_no += 1 ;
410
417
}
411
418
412
419
let frame_count = end_frame_no - start_frame_no + 1 ;
413
- Ok ( crate :: database:: Replicated {
420
+ Ok ( crate :: database:: Replicated {
414
421
frame_no : None ,
415
422
frames_synced : frame_count as usize ,
416
423
} )
417
424
}
418
425
419
426
#[ cfg( feature = "sync" ) ]
420
- async fn push_one_frame ( & self , conn : & Connection , sync_ctx : & SyncContext , generation : u32 , frame_no : u32 , page_size : u32 ) -> Result < u32 > {
421
- let frame_size: usize = 24 +page_size as usize ;
427
+ async fn push_one_frame (
428
+ & self ,
429
+ conn : & Connection ,
430
+ sync_ctx : & SyncContext ,
431
+ generation : u32 ,
432
+ frame_no : u32 ,
433
+ page_size : u32 ,
434
+ ) -> Result < u32 > {
435
+ let frame_size: usize = 24 + page_size as usize ;
422
436
let frame = vec ! [ 0 ; frame_size] ;
423
437
let rc = unsafe {
424
- libsql_sys:: ffi:: libsql_wal_get_frame ( conn. handle ( ) , frame_no, frame. as_ptr ( ) as * mut _ , frame_size as u32 )
438
+ libsql_sys:: ffi:: libsql_wal_get_frame (
439
+ conn. handle ( ) ,
440
+ frame_no,
441
+ frame. as_ptr ( ) as * mut _ ,
442
+ frame_size as u32 ,
443
+ )
425
444
} ;
426
445
if rc != 0 {
427
- return Err ( crate :: errors:: Error :: SqliteFailure ( rc as std:: ffi:: c_int , format ! ( "Failed to get frame: {}" , frame_no) ) ) ;
446
+ return Err ( crate :: errors:: Error :: SqliteFailure (
447
+ rc as std:: ffi:: c_int ,
448
+ format ! ( "Failed to get frame: {}" , frame_no) ,
449
+ ) ) ;
428
450
}
429
- let uri = format ! ( "{}/sync/{}/{}/{}" , sync_ctx. sync_url, generation, frame_no, frame_no+1 ) ;
430
- let max_frame_no = self . push_with_retry ( uri, & sync_ctx. auth_token , frame. to_vec ( ) , sync_ctx. max_retries ) . await ?;
451
+ let uri = format ! (
452
+ "{}/sync/{}/{}/{}" ,
453
+ sync_ctx. sync_url,
454
+ generation,
455
+ frame_no,
456
+ frame_no + 1
457
+ ) ;
458
+ let max_frame_no = self
459
+ . push_with_retry (
460
+ uri,
461
+ & sync_ctx. auth_token ,
462
+ frame. to_vec ( ) ,
463
+ sync_ctx. max_retries ,
464
+ )
465
+ . await ?;
431
466
Ok ( max_frame_no)
432
467
}
433
468
434
469
#[ cfg( feature = "sync" ) ]
435
- async fn push_with_retry ( & self , uri : String , auth_token : & Option < String > , frame : Vec < u8 > , max_retries : usize ) -> Result < u32 > {
470
+ async fn push_with_retry (
471
+ & self ,
472
+ uri : String ,
473
+ auth_token : & Option < String > ,
474
+ frame : Vec < u8 > ,
475
+ max_retries : usize ,
476
+ ) -> Result < u32 > {
436
477
let mut nr_retries = 0 ;
437
478
loop {
438
479
let client = reqwest:: Client :: new ( ) ;
439
480
let mut builder = client. post ( uri. to_owned ( ) ) ;
440
- match auth_token {
481
+ match auth_token {
441
482
Some ( ref auth_token) => {
442
- builder = builder. header ( "Authorization" , format ! ( "Bearer {}" , auth_token. to_owned( ) ) ) ;
483
+ builder = builder
484
+ . header ( "Authorization" , format ! ( "Bearer {}" , auth_token. to_owned( ) ) ) ;
443
485
}
444
486
None => { }
445
487
}
@@ -450,7 +492,10 @@ impl Database {
450
492
return Ok ( max_frame_no as u32 ) ;
451
493
}
452
494
if nr_retries > max_retries {
453
- return Err ( crate :: errors:: Error :: ConnectionFailed ( format ! ( "Failed to push frame: {}" , res. status( ) ) ) ) ;
495
+ return Err ( crate :: errors:: Error :: ConnectionFailed ( format ! (
496
+ "Failed to push frame: {}" ,
497
+ res. status( )
498
+ ) ) ) ;
454
499
}
455
500
let delay = std:: time:: Duration :: from_millis ( 100 * ( 1 << nr_retries) ) ;
456
501
tokio:: time:: sleep ( delay) . await ;
0 commit comments