From cd081280bfd57831dad546abae0a52432a70383d Mon Sep 17 00:00:00 2001 From: Bojan Serafimov Date: Sun, 12 Jun 2022 12:23:43 -0400 Subject: [PATCH] Improve errors --- control_plane/src/storage.rs | 4 ++-- pageserver/src/page_service.rs | 13 ++++++++----- 2 files changed, 10 insertions(+), 7 deletions(-) diff --git a/control_plane/src/storage.rs b/control_plane/src/storage.rs index 8ef01d7116..82256f3f8a 100644 --- a/control_plane/src/storage.rs +++ b/control_plane/src/storage.rs @@ -543,7 +543,7 @@ impl PageServerNode { let base_tarfile = File::open(base_tarfile_path)?; let mut base_reader = BufReader::new(base_tarfile); - // If there's any wal, extend the reader and end_lsn + // Init wal reader if necessary let (end_lsn, wal_reader) = if let Some((end_lsn, wal_tarfile_path)) = pg_wal { let wal_tarfile = File::open(wal_tarfile_path)?; let wal_reader = BufReader::new(wal_tarfile); @@ -559,7 +559,7 @@ impl PageServerNode { io::copy(&mut base_reader, &mut writer)?; writer.finish()?; - // Import wal + // Import wal if necessary if let Some(mut wal_reader) = wal_reader { let import_cmd = format!("import wal {tenant_id} {timeline_id} {start_lsn} {end_lsn}"); let mut writer = client.copy_in(&import_cmd)?; diff --git a/pageserver/src/page_service.rs b/pageserver/src/page_service.rs index 7341bab1cc..05f1713359 100644 --- a/pageserver/src/page_service.rs +++ b/pageserver/src/page_service.rs @@ -261,9 +261,11 @@ impl<'a> Read for CopyInReader<'a> { let copy_data_bytes = match message { FeMessage::CopyData(bytes) => bytes, FeMessage::CopyDone => return Ok(0), + FeMessage::Sync => continue, m => { - info!("unexpected copy in client message {:?}", m); - continue; + let msg = format!("unexpected message {:?}", m); + self.pgb.write_message(&BeMessage::ErrorResponse(&msg))?; + return Err(io::Error::new(io::ErrorKind::Other, msg)); } }; @@ -274,12 +276,13 @@ impl<'a> Read for CopyInReader<'a> { return Ok(bytes_read); } Ok(None) => { - // Is this ok? - return Ok(0); + let msg = "client closed connection"; + self.pgb.write_message(&BeMessage::ErrorResponse(msg))?; + return Err(io::Error::new(io::ErrorKind::Other, msg)); } Err(e) => { if !is_socket_read_timed_out(&e) { - todo!("return io::Error"); + return Err(io::Error::new(io::ErrorKind::Other, e)); } } }