replication log compaction

Author: MarinPostma

Cargo.lock

@@ -54,6 +54,9 @@ tower-http = { version = "0.3.5", features = ["compression-full", "cors", "trace
 tracing = "0.1.37"
 tracing-subscriber = { version = "0.3.16", features = ["env-filter"] }
 uuid = { version = "1.3", features = ["v4", "serde"] }
+nix = { version = "0.26.2", features = ["fs"] }
+tempfile = "3.3.0"
+memmap = "0.7.0"
 
 [dev-dependencies]
 proptest = "1.0.0"

sqld/Cargo.toml

@@ -24,4 +24,5 @@ message Frame {
 service ReplicationLog {
     rpc Hello(HelloRequest) returns (HelloResponse) {}
     rpc LogEntries(LogOffset) returns (stream Frame) {}
+    rpc Snapshot(LogOffset) returns (stream Frame) {}
 }

sqld/proto/replication_log.proto

@@ -13,7 +13,7 @@ use database::write_proxy::WriteProxyDbFactory;
 use once_cell::sync::Lazy;
 #[cfg(feature = "mwal_backend")]
 use once_cell::sync::OnceCell;
-use replication::logger::{ReplicationLogger, ReplicationLoggerHook};
+use replication::{ReplicationLogger, ReplicationLoggerHook};
 use rpc::run_rpc_server;
 use tokio::sync::{mpsc, Notify};
 use tokio::task::JoinSet;
@@ -89,6 +89,7 @@ pub struct Config {
     pub create_local_http_tunnel: bool,
     pub idle_shutdown_timeout: Option<Duration>,
     pub load_from_dump: Option<PathBuf>,
+    pub max_log_size: u64,
 }
 
 async fn run_service(
@@ -214,7 +215,10 @@ async fn start_primary(
     idle_shutdown_layer: Option<IdleShutdownLayer>,
 ) -> anyhow::Result<()> {
     let is_fresh_db = check_fresh_db(&config.db_path);
-    let logger = Arc::new(ReplicationLogger::open(&config.db_path)?);
+    let logger = Arc::new(ReplicationLogger::open(
+        &config.db_path,
+        config.max_log_size,
+    )?);
     let logger_clone = logger.clone();
     let path_clone = config.db_path.clone();
     #[cfg(feature = "bottomless")]

sqld/src/lib.rs

@@ -110,6 +110,11 @@ struct Cli {
     /// Requires that the node is not in replica mode
     #[clap(long, env = "SQLD_LOAD_DUMP_PATH", conflicts_with = "primary_grpc_url")]
     load_from_dump: Option<PathBuf>,
+
+    /// Maximum size the replication log is allowed to grow (in MB).
+    /// defaults to 200MB.
+    #[clap(long, env = "SQLD_MAX_LOG_SIZE", default_value = "200")]
+    max_log_size: u64,
 }
 
 impl Cli {
@@ -198,6 +203,7 @@ fn config_from_args(args: Cli) -> Result<Config> {
         create_local_http_tunnel: args.create_local_http_tunnel,
         idle_shutdown_timeout: args.idle_shutdown_timeout_s.map(Duration::from_secs),
         load_from_dump: args.load_from_dump,
+        max_log_size: args.max_log_size,
     })
 }