From 4d5a41301dd84cc0b5bf8949978a2e1b2e7370ce Mon Sep 17 00:00:00 2001
From: Stas Kelvich <stas.kelvich@gmail.com>
Date: Wed, 19 May 2021 15:56:12 +0300
Subject: [PATCH] Support returning errors from page service

---
 pageserver/src/branches.rs     |  4 ++--
 pageserver/src/page_service.rs | 22 ++++++++++++++++++----
 2 files changed, 20 insertions(+), 6 deletions(-)

diff --git a/pageserver/src/branches.rs b/pageserver/src/branches.rs
index d63dddc678..00702ed7fe 100644
--- a/pageserver/src/branches.rs
+++ b/pageserver/src/branches.rs
@@ -4,7 +4,7 @@
 // TODO: move all paths construction to conf impl
 //
 
-use anyhow::{anyhow, Context, Result};
+use anyhow::{anyhow, bail, Context, Result};
 use bytes::Bytes;
 use fs::File;
 use fs_extra;
@@ -274,7 +274,7 @@ fn parse_point_in_time(conf: &PageServerConf, s: &str) -> Result<PointInTime> {
         }
     }
 
-    panic!("could not parse point-in-time {}", s);
+    bail!("could not parse point-in-time {}", s);
 }
 
 // If control file says the cluster was shut down cleanly, modify it, to mark
diff --git a/pageserver/src/page_service.rs b/pageserver/src/page_service.rs
index 6562d3fe07..8ac1d19e7d 100644
--- a/pageserver/src/page_service.rs
+++ b/pageserver/src/page_service.rs
@@ -60,6 +60,7 @@ enum BeMessage {
     CommandComplete,
     ControlFile,
     CopyData(Bytes),
+    ErrorResponse(String),
 }
 
 // Wrapped in libpq CopyData
@@ -575,6 +576,16 @@ impl Connection {
                 self.stream.write_u32::<BE>(4 + data.len() as u32)?;
                 self.stream.write_all(&data)?;
             }
+
+            BeMessage::ErrorResponse(error_msg) => {
+                self.stream.write_u8(b'E')?;
+                self.stream
+                    .write_u32::<BE>(4 + 1 + error_msg.len() as u32 + 2)?;
+                self.stream.write_u8(b'M')?; /* primary human-readable error message */
+                self.stream.write_all(error_msg.as_bytes())?;
+                self.stream.write_u8(0)?; /* end of M field */
+                self.stream.write_u8(0)?; /* end of all fields */
+            }
         }
 
         Ok(())
@@ -612,7 +623,12 @@ impl Connection {
                     }
                 }
                 Some(FeMessage::Query(m)) => {
-                    self.process_query(m.body)?;
+                    if let Err(e) = self.process_query(m.body) {
+                        self.write_message_noflush(&BeMessage::ErrorResponse(format!(
+                            "ERROR: {}",
+                            e
+                        )))?;
+                    }
                     self.write_message(&BeMessage::ReadyForQuery)?;
                 }
                 Some(FeMessage::Parse(m)) => {
@@ -752,9 +768,7 @@ impl Connection {
             self.write_message_noflush(&BeMessage::DataRow(Bytes::from(system_id)))?;
             self.write_message_noflush(&BeMessage::CommandComplete)?;
         } else {
-            self.write_message_noflush(&BeMessage::RowDescription)?;
-            self.write_message_noflush(&HELLO_WORLD_ROW)?;
-            self.write_message_noflush(&BeMessage::CommandComplete)?;
+            bail!("unknown command");
         }
 
         Ok(())