Merge #293

293: log compaction r=MarinPostma a=MarinPostma

# Log compaction and snapshotting

This PR enables log compaction and snapshotting.

## Motivation

The replication log follows sqlite WAL and grows indefinitely. Fortunately, it contains a lot of duplicate data, so we can compress it by getting by keeping only the most recent version of each page. This is what this PR does: whenever the replication log grows above some threshold, a new log is created, and the old log is compacted. This operation is done atomically, so the compaction can happen in the background, while we keep writing to the old log.

## Log compaction:

The log compaction is very straightforward. We iterate backwards through the replication log, and write the frames to the snapshot file. We keep track of what pages we have already seen, and ignore older version of them. When the snapshot is finished, we remove the old log file.

Notice how the frames are now in a reverse order in the snapshot: starting with the most recent, and ending with the oldest.

## Snapshoting:

Whenever a replica asks for a frame that is not present in the current log (i.e, it asks for a frame index less than the log starting frame id), then it sends the replica an error, asking it to ask for a snapshot.

The replica receives this message, and immediately asks for a snapshot, sending over the frame id that got it rejected in the first place.
The primary looks for a snapshot containing the request frame, and starts iterating through it, until it reaches a frame that is less than the requested one.

This mechanism allows us to send partial snapshot: the replica gets minimal amount of frames required to get up to speed.

The replica writes the snapshot frames to a file, and then `mmap` that file to build a chained list of pages to append to the WAL.


## Future work:

This PR was getting a bit too long, so I left out some work for followup PRs:

- Even though the snapshot significantly compresses the size of the log, a new log is created, that will lead to the creation of a new snapshot. Those snapshot will pile up and we're back at step one. The next step is to merge snapshots into bigger snapshots, and get rid of the older snapshots.
- Every query for a frame causes a read to the snapshot/log. To speed things up, let's add a MRU cache.
- Explore compression


Co-authored-by: ad hoc <[email protected]>

Author: bors[bot]

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,
     })
 }