replication: add HTTP implementation for /frames endpoint

psarna · psarna · commit 8da8af7df28a · 2023-07-20T12:12:04.000+02:00
It still does not perform any kind of handshake, just asks
for frames starting N.

Tested with:
 $ cargo run -- --http-replication-listen-addr 127.0.0.1:8081
 $ curl -d '{"next_offset": 0}' -v localhost:8081/frames
diff --git a/sqld/src/lib.rs b/sqld/src/lib.rs
@@ -497,14 +497,14 @@ async fn start_primary(
             config.rpc_server_key.clone(),
             config.rpc_server_ca_cert.clone(),
             db_factory.clone(),
-            logger,
+            logger.clone(),
             idle_shutdown_layer.clone(),
         ));
     }
 
     if let Some(ref addr) = config.http_replication_addr {
         let auth = get_auth(config)?;
-        join_set.spawn(replication::http::run(auth, *addr));
+        join_set.spawn(replication::http::run(auth, *addr, logger));
     }
 
     run_service(
diff --git a/sqld/src/replication/frame.rs b/sqld/src/replication/frame.rs
@@ -27,7 +27,7 @@ pub struct FrameHeader {
     pub size_after: u32,
 }
 
-#[derive(Clone)]
+#[derive(Clone, serde::Serialize, serde::Deserialize)]
 /// The owned version of a replication frame.
 /// Cloning this is cheap.
 pub struct Frame {
diff --git a/sqld/src/replication/http.rs b/sqld/src/replication/http.rs
@@ -1,3 +1,4 @@
+use crate::replication::{frame::Frame, primary::frame_stream::FrameStream, ReplicationLogger};
 use crate::Auth;
 use anyhow::{Context, Result};
 use hyper::server::conn::AddrIncoming;
@@ -9,7 +10,11 @@ use tower_http::trace::DefaultOnResponse;
 use tower_http::{compression::CompressionLayer, cors};
 use tracing::{Level, Span};
 
-pub(crate) async fn run(auth: Arc<Auth>, addr: SocketAddr) -> Result<()> {
+pub(crate) async fn run(
+    auth: Arc<Auth>,
+    addr: SocketAddr,
+    logger: Arc<ReplicationLogger>,
+) -> Result<()> {
     tracing::info!("listening for HTTP requests on {addr}");
 
     fn trace_request<B>(req: &Request<B>, _span: &Span) {
@@ -34,7 +39,8 @@ pub(crate) async fn run(auth: Arc<Auth>, addr: SocketAddr) -> Result<()> {
         )
         .service_fn(move |req| {
             let auth = auth.clone();
-            handle_request(auth, req)
+            let logger = logger.clone();
+            handle_request(auth, req, logger)
         });
 
     let listener = tokio::net::TcpListener::bind(&addr).await?;
@@ -47,7 +53,11 @@ pub(crate) async fn run(auth: Arc<Auth>, addr: SocketAddr) -> Result<()> {
     Ok(())
 }
 
-async fn handle_request(auth: Arc<Auth>, req: Request<Body>) -> Result<Response<Body>> {
+async fn handle_request(
+    auth: Arc<Auth>,
+    req: Request<Body>,
+    logger: Arc<ReplicationLogger>,
+) -> Result<Response<Body>> {
     let auth_header = req.headers().get(hyper::header::AUTHORIZATION);
     let auth = match auth.authenticate_http(auth_header) {
         Ok(auth) => auth,
@@ -60,7 +70,7 @@ async fn handle_request(auth: Arc<Auth>, req: Request<Body>) -> Result<Response<
     };
 
     match (req.method(), req.uri().path()) {
-        (&Method::POST, "/frames") => handle_query(req, auth).await,
+        (&Method::POST, "/frames") => handle_query(req, auth, logger).await,
         _ => Ok(Response::builder().status(404).body(Body::empty()).unwrap()),
     }
 }
@@ -70,6 +80,25 @@ pub struct FramesRequest {
     pub next_offset: u64,
 }
 
+#[derive(Debug, serde::Deserialize, serde::Serialize)]
+pub struct Frames {
+    pub frames: Vec<Frame>,
+}
+
+impl Frames {
+    pub fn new() -> Self {
+        Self { frames: Vec::new() }
+    }
+
+    pub fn push(&mut self, frame: Frame) {
+        self.frames.push(frame);
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.frames.is_empty()
+    }
+}
+
 fn error(msg: &str, code: hyper::StatusCode) -> Response<Body> {
     let err = serde_json::json!({ "error": msg });
     Response::builder()
@@ -81,18 +110,58 @@ fn error(msg: &str, code: hyper::StatusCode) -> Response<Body> {
 async fn handle_query(
     mut req: Request<Body>,
     _auth: crate::auth::Authenticated,
+    logger: Arc<ReplicationLogger>,
 ) -> Result<Response<Body>> {
     let bytes = hyper::body::to_bytes(req.body_mut()).await?;
     let FramesRequest { next_offset } = match serde_json::from_slice(&bytes) {
         Ok(req) => req,
         Err(resp) => return Ok(error(&resp.to_string(), hyper::StatusCode::BAD_REQUEST)),
     };
 
+    let mut frame_stream = FrameStream::new(logger, next_offset);
+
+    if frame_stream.max_available_frame_no < next_offset {
+        tracing::trace!("No frames available starting {next_offset}, returning 204 No Content");
+        return Ok(Response::builder()
+            .status(hyper::StatusCode::NO_CONTENT)
+            .body(Body::empty())
+            .unwrap());
+    }
+
+    let mut frames = Frames::new();
+    loop {
+        use futures::StreamExt;
+
+        match frame_stream.next().await {
+            Some(Ok(frame)) => {
+                tracing::trace!("Read frame {}", frame_stream.current_frame_no);
+                frames.push(frame);
+            }
+            Some(Err(e)) => {
+                tracing::error!("Error reading frame: {}", e);
+                return Ok(Response::builder()
+                    .status(hyper::StatusCode::INTERNAL_SERVER_ERROR)
+                    .body(Body::empty())
+                    .unwrap());
+            }
+            None => break,
+        }
+
+        // FIXME: also stop when we have enough frames to fill a large buffer
+        if frame_stream.max_available_frame_no <= frame_stream.current_frame_no {
+            break;
+        }
+    }
+
+    if frames.is_empty() {
+        return Ok(Response::builder()
+            .status(hyper::StatusCode::NO_CONTENT)
+            .body(Body::empty())
+            .unwrap());
+    }
+
     Ok(Response::builder()
         .status(hyper::StatusCode::OK)
-        .body(Body::from(format!(
-            "{{\"comment\":\"thx for sending the request\", \"next_offset\":{}}}",
-            next_offset + 1
-        )))
+        .body(Body::from(serde_json::to_string(&frames)?))
         .unwrap())
 }
diff --git a/sqld/src/replication/primary/frame_stream.rs b/sqld/src/replication/primary/frame_stream.rs
@@ -11,8 +11,8 @@ use crate::replication::{FrameNo, LogReadError, ReplicationLogger};
 /// Streams frames from the replication log starting at `current_frame_no`.
 /// Only stops if the current frame is not in the log anymore.
 pub struct FrameStream {
-    current_frame_no: FrameNo,
-    max_available_frame_no: FrameNo,
+    pub(crate) current_frame_no: FrameNo,
+    pub(crate) max_available_frame_no: FrameNo,
     logger: Arc<ReplicationLogger>,
     state: FrameStreamState,
 }

Original file line number	Diff line number	Diff line change
`@@ -27,7 +27,7 @@ pub struct FrameHeader {`
`27`	`27`	`pub size_after: u32,`
`28`	`28`	`}`
`29`	`29`
`30`		`-#[derive(Clone)]`
	`30`	`+#[derive(Clone, serde::Serialize, serde::Deserialize)]`
`31`	`31`	`/// The owned version of a replication frame.`
`32`	`32`	`/// Cloning this is cheap.`
`33`	`33`	`pub struct Frame {`