Skip to content

Commit e3d2414

Browse files
authored
Merge pull request #1923 from levydsa/main
Prevent reporting HttpDispatch errors in `sync_offline`
2 parents 89cc8cd + 111c277 commit e3d2414

File tree

2 files changed

+55
-25
lines changed

2 files changed

+55
-25
lines changed

libsql/src/local/connection.rs

Lines changed: 27 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -495,7 +495,7 @@ impl Connection {
495495
Ok(buf)
496496
}
497497

498-
pub(crate) fn wal_insert_begin(&self) -> Result<()> {
498+
fn wal_insert_begin(&self) -> Result<()> {
499499
let rc = unsafe { libsql_sys::ffi::libsql_wal_insert_begin(self.handle()) };
500500
if rc != 0 {
501501
return Err(crate::errors::Error::SqliteFailure(
@@ -506,7 +506,7 @@ impl Connection {
506506
Ok(())
507507
}
508508

509-
pub(crate) fn wal_insert_end(&self) -> Result<()> {
509+
fn wal_insert_end(&self) -> Result<()> {
510510
let rc = unsafe { libsql_sys::ffi::libsql_wal_insert_end(self.handle()) };
511511
if rc != 0 {
512512
return Err(crate::errors::Error::SqliteFailure(
@@ -517,7 +517,7 @@ impl Connection {
517517
Ok(())
518518
}
519519

520-
pub(crate) fn wal_insert_frame(&self, frame: &[u8]) -> Result<()> {
520+
fn wal_insert_frame(&self, frame: &[u8]) -> Result<()> {
521521
let rc = unsafe {
522522
libsql_sys::ffi::libsql_wal_insert_frame(
523523
self.handle(),
@@ -534,6 +534,30 @@ impl Connection {
534534
}
535535
Ok(())
536536
}
537+
538+
pub(crate) fn wal_insert_handle(&self) -> Result<WalInsertHandle<'_>> {
539+
self.wal_insert_begin()?;
540+
Ok(WalInsertHandle { conn: self })
541+
}
542+
}
543+
544+
pub(crate) struct WalInsertHandle<'a> {
545+
conn: &'a Connection,
546+
}
547+
548+
impl WalInsertHandle<'_> {
549+
pub fn insert(&self, frame: &[u8]) -> Result<()> {
550+
self.conn.wal_insert_frame(frame)
551+
}
552+
}
553+
554+
impl Drop for WalInsertHandle<'_> {
555+
fn drop(&mut self) {
556+
if let Err(err) = self.conn.wal_insert_end() {
557+
tracing::error!("{:?}", err);
558+
Err(err).unwrap()
559+
}
560+
}
537561
}
538562

539563
impl fmt::Debug for Connection {

libsql/src/local/database.rs

Lines changed: 28 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -478,9 +478,7 @@ impl Database {
478478
Err(Error::Sync(err)) => {
479479
// Retry the sync because we are ahead of the server and we need to push some older
480480
// frames.
481-
if let Some(SyncError::InvalidPushFrameNoLow(_, _)) =
482-
err.downcast_ref::<SyncError>()
483-
{
481+
if let Some(SyncError::InvalidPushFrameNoLow(_, _)) = err.downcast_ref() {
484482
tracing::debug!("got InvalidPushFrameNo, retrying push");
485483
self.try_push(&mut sync_ctx, &conn).await
486484
} else {
@@ -492,6 +490,22 @@ impl Database {
492490
} else {
493491
self.try_pull(&mut sync_ctx, &conn).await
494492
}
493+
.or_else(|err| {
494+
let Error::Sync(err) = err else {
495+
return Err(err);
496+
};
497+
498+
// TODO(levy): upcasting should be done *only* at the API boundary, doing this in
499+
// internal code just sucks.
500+
let Some(SyncError::HttpDispatch(_)) = err.downcast_ref() else {
501+
return Err(Error::Sync(err));
502+
};
503+
504+
Ok(crate::database::Replicated {
505+
frame_no: None,
506+
frames_synced: 0,
507+
})
508+
})
495509
}
496510

497511
#[cfg(feature = "sync")]
@@ -557,37 +571,29 @@ impl Database {
557571
) -> Result<crate::database::Replicated> {
558572
let generation = sync_ctx.generation();
559573
let mut frame_no = sync_ctx.durable_frame_num() + 1;
560-
conn.wal_insert_begin()?;
561574

562-
let mut err = None;
575+
let insert_handle = conn.wal_insert_handle()?;
563576

564577
loop {
565578
match sync_ctx.pull_one_frame(generation, frame_no).await {
566579
Ok(Some(frame)) => {
567-
conn.wal_insert_frame(&frame)?;
580+
insert_handle.insert(&frame)?;
568581
frame_no += 1;
569582
}
570583
Ok(None) => {
571-
break;
584+
sync_ctx.write_metadata().await?;
585+
return Ok(crate::database::Replicated {
586+
frame_no: None,
587+
frames_synced: 1,
588+
});
572589
}
573-
Err(e) => {
574-
tracing::debug!("pull_one_frame error: {:?}", e);
575-
err.replace(e);
576-
break;
590+
Err(err) => {
591+
tracing::debug!("pull_one_frame error: {:?}", err);
592+
sync_ctx.write_metadata().await?;
593+
return Err(err);
577594
}
578595
}
579596
}
580-
conn.wal_insert_end()?;
581-
sync_ctx.write_metadata().await?;
582-
583-
if let Some(err) = err {
584-
Err(err)
585-
} else {
586-
Ok(crate::database::Replicated {
587-
frame_no: None,
588-
frames_synced: 1,
589-
})
590-
}
591597
}
592598

593599
pub(crate) fn path(&self) -> &str {

0 commit comments

Comments
 (0)