diff --git a/Cargo.lock b/Cargo.lock index 5af34f023..c9a79ba8e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1088,16 +1088,17 @@ dependencies = [ "bencher_config", "bencher_endpoint", "bencher_json", + "bencher_litestream", "bencher_logger", "bencher_otel", "bencher_otel_provider", "bencher_schema", + "camino", "dropshot", "futures-concurrency", "futures-util", "libsqlite3-sys", "sentry", - "serde_yaml", "slog", "thiserror 2.0.18", "tokio", @@ -1248,6 +1249,7 @@ dependencies = [ "bencher_google_client", "bencher_json", "bencher_license", + "bencher_litestream", "bencher_oci_storage", "bencher_rbac", "bencher_recaptcha", @@ -1346,18 +1348,17 @@ name = "bencher_json" version = "0.6.7" dependencies = [ "bencher_context", + "bencher_litestream", "bencher_valid", "camino", "derive_more", "diesel", "ordered-float", "percent-encoding", - "pretty_assertions", "schemars", "serde", "serde_json", "serde_urlencoded", - "serde_yaml", "tabled", "thiserror 2.0.18", "typeshare", @@ -1377,6 +1378,21 @@ dependencies = [ "thiserror 2.0.18", ] +[[package]] +name = "bencher_litestream" +version = "0.6.7" +dependencies = [ + "bencher_valid", + "camino", + "pretty_assertions", + "schemars", + "serde", + "serde_yaml", + "slog", + "thiserror 2.0.18", + "tokio", +] + [[package]] name = "bencher_logger" version = "0.6.7" diff --git a/Cargo.toml b/Cargo.toml index 5a0d5f6e0..a4ef8e5e3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -61,6 +61,7 @@ bencher_github_client = { path = "plus/bencher_github_client" } bencher_google_client = { path = "plus/bencher_google_client" } bencher_google_index = { path = "plus/bencher_google_index" } bencher_license = { path = "plus/bencher_license" } +bencher_litestream = { path = "plus/bencher_litestream" } bencher_otel = { path = "plus/bencher_otel" } bencher_otel_provider = { path = "plus/bencher_otel_provider" } bencher_oci_storage = { path = "plus/bencher_oci_storage" } diff --git a/docker/bench.Dockerfile b/docker/bench.Dockerfile index 2fa46be1d..0ee92bd76 100644 --- a/docker/bench.Dockerfile +++ b/docker/bench.Dockerfile @@ -63,6 +63,7 @@ RUN cargo init --lib bencher_recaptcha RUN cargo init --lib bencher_rootfs RUN cargo init --lib bencher_runner RUN cargo init --lib bencher_init +COPY plus/bencher_litestream bencher_litestream WORKDIR /usr/src/bencher/tasks RUN cargo init --bin bin_version @@ -90,7 +91,7 @@ RUN cargo init runner COPY services/api/openapi.json /usr/src/bencher/services/api/openapi.json WORKDIR /usr/src/bencher -RUN cargo bench --package bencher_adapter --no-run --message-format=json \ +RUN cargo bench --package bencher_adapter --no-run --features server --message-format=json \ | jq -r 'select(.executable != null and .target.kind == ["bench"]) | .executable' \ | xargs -I {} cp {} /usr/local/bin/bench-adapter diff --git a/lib/bencher_adapter/Cargo.toml b/lib/bencher_adapter/Cargo.toml index 63de9b1ed..68dfbcc88 100644 --- a/lib/bencher_adapter/Cargo.toml +++ b/lib/bencher_adapter/Cargo.toml @@ -8,10 +8,12 @@ publish = false [features] default = [] +client = ["bencher_json/client"] +server = ["bencher_json/server"] plus = ["bencher_json/plus"] [dependencies] -bencher_json = { workspace = true, features = ["server"] } +bencher_json.workspace = true literally.workspace = true nom.workspace = true ordered-float.workspace = true diff --git a/lib/bencher_config/Cargo.toml b/lib/bencher_config/Cargo.toml index 828ea2257..9213d44f0 100644 --- a/lib/bencher_config/Cargo.toml +++ b/lib/bencher_config/Cargo.toml @@ -11,6 +11,7 @@ default = [] plus = [ "dep:bencher_github_client", "dep:bencher_google_client", + "dep:bencher_litestream", "dep:bencher_oci_storage", "bencher_billing/plus", "bencher_endpoint/plus", @@ -30,6 +31,7 @@ bencher_github_client = { workspace = true, optional = true } bencher_google_client = { workspace = true, optional = true } bencher_json.workspace = true bencher_license = { workspace = true, optional = true } +bencher_litestream = { workspace = true, optional = true } bencher_oci_storage = { workspace = true, optional = true } bencher_rbac.workspace = true bencher_recaptcha = { workspace = true, optional = true } diff --git a/lib/bencher_config/src/config_tx.rs b/lib/bencher_config/src/config_tx.rs index c4100567c..33db7d40c 100644 --- a/lib/bencher_config/src/config_tx.rs +++ b/lib/bencher_config/src/config_tx.rs @@ -419,7 +419,7 @@ fn run_litestream(database: &mut DbConnection) -> Result<(), ConfigTxError> { // https://litestream.io/tips/#disable-autocheckpoints-for-high-write-load-servers // https://sqlite.org/wal.html#automatic_checkpoint database - .batch_execute("PRAGMA wal_autocheckpoint = 0") + .batch_execute(bencher_litestream::DISABLE_AUTOCHECKPOINT_PRAGMA) .map_err(ConfigTxError::Pragma)?; Ok(()) diff --git a/lib/bencher_json/Cargo.toml b/lib/bencher_json/Cargo.toml index 1c34124b3..15e4934d8 100644 --- a/lib/bencher_json/Cargo.toml +++ b/lib/bencher_json/Cargo.toml @@ -8,16 +8,22 @@ publish = false [features] default = [] -client = ["bencher_context/client", "bencher_valid/client"] +client = ["bencher_context/client", "bencher_valid/client", "bencher_litestream?/client"] table = ["dep:tabled"] -server = ["bencher_context/server", "bencher_valid/server"] -schema = ["dep:schemars", "bencher_context/schema", "ordered-float/schemars"] -db = ["dep:diesel", "dep:serde_yaml", "bencher_valid/db"] -plus = ["dep:camino", "bencher_valid/plus"] +server = ["bencher_context/server", "bencher_valid/server", "bencher_litestream?/server"] +schema = [ + "dep:schemars", + "bencher_context/schema", + "ordered-float/schemars", + "bencher_litestream?/schema", +] +db = ["dep:diesel", "bencher_valid/db", "bencher_litestream?/yaml"] +plus = ["dep:camino", "dep:bencher_litestream", "bencher_valid/plus"] test-clock = ["bencher_valid/test-clock"] [dependencies] bencher_context.workspace = true +bencher_litestream = { workspace = true, optional = true } bencher_valid = { workspace = true, features = ["schema"] } camino = { workspace = true, features = ["serde1"], optional = true } derive_more.workspace = true @@ -28,15 +34,11 @@ schemars = { workspace = true, optional = true, features = ["chrono", "url"] } serde.workspace = true serde_json.workspace = true serde_urlencoded.workspace = true -serde_yaml = { workspace = true, optional = true } tabled = { workspace = true, optional = true } thiserror.workspace = true typeshare.workspace = true url = { workspace = true, features = ["serde"] } uuid = { workspace = true, features = ["v4", "serde"] } -[dev-dependencies] -pretty_assertions.workspace = true - [lints] workspace = true diff --git a/lib/bencher_json/src/lib.rs b/lib/bencher_json/src/lib.rs index 9581c76a1..5c63bd52d 100644 --- a/lib/bencher_json/src/lib.rs +++ b/lib/bencher_json/src/lib.rs @@ -17,8 +17,6 @@ pub use bencher_valid::{ #[cfg(feature = "schema")] use schemars::JsonSchema; use serde::{Deserialize, Serialize}; -#[cfg(all(not(feature = "plus"), feature = "db"))] -use serde_yaml as _; mod auth; pub mod big_int; diff --git a/lib/bencher_json/src/system/config/plus/litestream.rs b/lib/bencher_json/src/system/config/plus/litestream.rs index 3dc1ac043..b2b4751e6 100644 --- a/lib/bencher_json/src/system/config/plus/litestream.rs +++ b/lib/bencher_json/src/system/config/plus/litestream.rs @@ -1,598 +1,20 @@ -use std::path::PathBuf; - -use bencher_valid::{Sanitize, Secret}; -#[cfg(feature = "schema")] -use schemars::JsonSchema; -use serde::{Deserialize, Serialize}; - -#[derive(Debug, Clone, Serialize, Deserialize)] -#[cfg_attr(feature = "schema", derive(JsonSchema))] -// https://litestream.io/reference/config/ -pub struct JsonLitestream { - /// Disaster recovery replica - pub replica: JsonReplica, - /// Snapshot configuration - #[serde(skip_serializing_if = "Option::is_none")] - pub snapshot: Option, - /// Validation configuration - #[serde(skip_serializing_if = "Option::is_none")] - pub validation: Option, - /// Checkpoint configuration - #[serde(skip_serializing_if = "Option::is_none")] - pub checkpoint: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -#[cfg_attr(feature = "schema", derive(JsonSchema))] -pub struct JsonSnapshot { - /// How often new snapshots are created - #[serde(skip_serializing_if = "Option::is_none")] - pub interval: Option, - /// How long snapshot & WAL files are kept - #[serde(skip_serializing_if = "Option::is_none")] - pub retention: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -#[cfg_attr(feature = "schema", derive(JsonSchema))] -pub struct JsonValidation { - /// How often to restore and validate replica data - #[serde(skip_serializing_if = "Option::is_none")] - pub interval: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -#[cfg_attr(feature = "schema", derive(JsonSchema))] -pub struct JsonCheckpoint { - /// How often to perform a non-blocking PASSIVE checkpoint - #[serde(skip_serializing_if = "Option::is_none")] - pub interval: Option, - /// Minimum WAL pages before a PASSIVE checkpoint triggers - #[serde(skip_serializing_if = "Option::is_none")] - pub min_page_count: Option, - /// Page threshold for blocking TRUNCATE checkpoint (0 to disable) - #[serde(skip_serializing_if = "Option::is_none")] - pub truncate_page_n: Option, -} - -impl JsonCheckpoint { - /// Default value for `truncate_page_n`: disables blocking TRUNCATE checkpoints. - /// When set to 0, Litestream will never perform a blocking TRUNCATE checkpoint, - /// only non-blocking PASSIVE checkpoints (controlled by `interval` / `min_page_count`). - pub const TRUNCATE_DISABLED: u64 = 0; -} - -impl Sanitize for JsonLitestream { - fn sanitize(&mut self) { - self.replica.sanitize(); - } -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -#[cfg_attr(feature = "schema", derive(JsonSchema))] -#[serde(tag = "scheme", rename_all = "snake_case")] -pub enum JsonReplica { - // https://litestream.io/reference/config/#file-replica - File { - path: PathBuf, - #[serde(skip_serializing_if = "Option::is_none")] - sync_interval: Option, - }, - // https://litestream.io/guides/sftp/ - Sftp { - host: String, - port: u16, - user: String, - #[serde(skip_serializing_if = "Option::is_none")] - password: Option, - #[serde(skip_serializing_if = "Option::is_none")] - path: Option, - #[serde(skip_serializing_if = "Option::is_none")] - key_path: Option, - #[serde(skip_serializing_if = "Option::is_none")] - sync_interval: Option, - }, - // https://litestream.io/guides/s3/ - S3 { - bucket: String, - #[serde(skip_serializing_if = "Option::is_none")] - path: Option, - #[serde(skip_serializing_if = "Option::is_none")] - endpoint: Option, - #[serde(skip_serializing_if = "Option::is_none")] - region: Option, - access_key_id: String, - secret_access_key: Secret, - #[serde(skip_serializing_if = "Option::is_none")] - sync_interval: Option, - }, -} - -impl Sanitize for JsonReplica { - fn sanitize(&mut self) { - match self { - Self::File { .. } => {}, - Self::Sftp { password, .. } => password.sanitize(), - Self::S3 { - secret_access_key, .. - } => secret_access_key.sanitize(), +//! Litestream configuration types for Bencher Plus. +//! +//! The types now live in the [`bencher_litestream`] crate so they can be shared by +//! path. They are re-exported here to keep the historical +//! `bencher_json::system::config::...` paths (and the `OpenAPI` schema) stable. + +pub use bencher_litestream::{JsonCheckpoint, JsonLitestream, JsonReplica, LitestreamLevel}; + +use crate::system::config::LogLevel; + +impl From for LitestreamLevel { + fn from(level: LogLevel) -> Self { + match level { + LogLevel::Trace | LogLevel::Debug => Self::Debug, + LogLevel::Info => Self::Info, + LogLevel::Warn => Self::Warn, + LogLevel::Error | LogLevel::Critical => Self::Error, } } } - -#[cfg(feature = "db")] -mod db { - use std::path::PathBuf; - - use bencher_valid::Secret; - use serde::Serialize; - - use crate::system::config::LogLevel; - - use super::{JsonCheckpoint, JsonLitestream, JsonReplica, JsonSnapshot, JsonValidation}; - - impl JsonLitestream { - pub fn into_yaml( - self, - path: PathBuf, - log_level: LogLevel, - ) -> Result { - let Self { - replica, - snapshot, - validation, - checkpoint, - } = self; - let replica = LitestreamReplica::from(replica); - let snapshot = snapshot.map(LitestreamSnapshot::from); - let validation = validation.map(LitestreamValidation::from); - let (min_checkpoint_page_count, checkpoint_interval, truncate_page_n) = checkpoint - .map_or((None, None, JsonCheckpoint::TRUNCATE_DISABLED), |c| { - let JsonCheckpoint { - interval, - min_page_count, - truncate_page_n, - } = c; - ( - min_page_count, - interval, - truncate_page_n.unwrap_or(JsonCheckpoint::TRUNCATE_DISABLED), - ) - }); - let dbs = vec![LitestreamDb { - path, - replica, - min_checkpoint_page_count, - checkpoint_interval, - truncate_page_n, - }]; - let logging = Some(LitestreamLogging { - level: Some(log_level.into()), - }); - let litestream = Litestream { - dbs, - snapshot, - validation, - logging, - }; - serde_yaml::to_string(&litestream) - } - } - - #[derive(Debug, Clone, Serialize)] - #[serde(rename_all = "kebab-case")] - pub struct Litestream { - pub dbs: Vec, - #[serde(skip_serializing_if = "Option::is_none")] - pub snapshot: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub validation: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub logging: Option, - } - - #[derive(Debug, Clone, Serialize)] - #[serde(rename_all = "kebab-case")] - pub struct LitestreamSnapshot { - #[serde(skip_serializing_if = "Option::is_none")] - pub interval: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub retention: Option, - } - - #[derive(Debug, Clone, Serialize)] - #[serde(rename_all = "kebab-case")] - pub struct LitestreamValidation { - #[serde(skip_serializing_if = "Option::is_none")] - pub interval: Option, - } - - #[derive(Debug, Clone, Serialize)] - #[serde(rename_all = "kebab-case")] - pub struct LitestreamDb { - pub path: PathBuf, - pub replica: LitestreamReplica, - #[serde(skip_serializing_if = "Option::is_none")] - pub min_checkpoint_page_count: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub checkpoint_interval: Option, - pub truncate_page_n: u64, - } - - #[derive(Debug, Clone, Serialize)] - #[serde( - tag = "type", - rename_all = "kebab-case", - rename_all_fields = "kebab-case" - )] - pub enum LitestreamReplica { - File { - path: PathBuf, - #[serde(skip_serializing_if = "Option::is_none")] - sync_interval: Option, - }, - Sftp { - host: String, - user: String, - #[serde(skip_serializing_if = "Option::is_none")] - password: Option, - #[serde(skip_serializing_if = "Option::is_none")] - path: Option, - #[serde(skip_serializing_if = "Option::is_none")] - key_path: Option, - #[serde(skip_serializing_if = "Option::is_none")] - sync_interval: Option, - }, - S3 { - bucket: String, - #[serde(skip_serializing_if = "Option::is_none")] - path: Option, - #[serde(skip_serializing_if = "Option::is_none")] - endpoint: Option, - #[serde(skip_serializing_if = "Option::is_none")] - region: Option, - access_key_id: String, - secret_access_key: Secret, - #[serde(skip_serializing_if = "Option::is_none")] - sync_interval: Option, - }, - } - - impl From for LitestreamReplica { - fn from(replica: JsonReplica) -> Self { - match replica { - JsonReplica::File { - path, - sync_interval, - } => Self::File { - path, - sync_interval, - }, - JsonReplica::Sftp { - host, - port, - user, - password, - path, - key_path, - sync_interval, - } => Self::Sftp { - host: format!("{host}:{port}"), - user, - password, - path, - key_path, - sync_interval, - }, - JsonReplica::S3 { - bucket, - path, - endpoint, - region, - access_key_id, - secret_access_key, - sync_interval, - } => Self::S3 { - bucket, - path, - endpoint, - region, - access_key_id, - secret_access_key, - sync_interval, - }, - } - } - } - - impl From for LitestreamSnapshot { - fn from(snapshot: JsonSnapshot) -> Self { - let JsonSnapshot { - interval, - retention, - } = snapshot; - Self { - interval, - retention, - } - } - } - - impl From for LitestreamValidation { - fn from(validation: JsonValidation) -> Self { - let JsonValidation { interval } = validation; - Self { interval } - } - } - - #[derive(Debug, Clone, Serialize)] - #[serde(rename_all = "kebab-case")] - pub struct LitestreamLogging { - #[serde(skip_serializing_if = "Option::is_none")] - pub level: Option, - } - - #[derive(Debug, Clone, Serialize)] - #[serde(rename_all = "kebab-case")] - pub enum LitestreamLevel { - Debug, - Info, - Warn, - Error, - } - - impl From for LitestreamLevel { - fn from(level: LogLevel) -> Self { - match level { - LogLevel::Trace | LogLevel::Debug => Self::Debug, - LogLevel::Info => Self::Info, - LogLevel::Warn => Self::Warn, - LogLevel::Error | LogLevel::Critical => Self::Error, - } - } - } - - #[test] - fn into_yaml() { - let json_litestream = JsonLitestream { - replica: JsonReplica::S3 { - bucket: "bucket".to_owned(), - path: Some("/path/to/backup".to_owned()), - endpoint: None, - region: None, - access_key_id: "access_key_id".to_owned(), - secret_access_key: "secret_access_key".parse().unwrap(), - sync_interval: None, - }, - snapshot: None, - validation: None, - checkpoint: None, - }; - let path = PathBuf::from("/path/to/db"); - let log_level = LogLevel::Info; - let yaml = json_litestream.into_yaml(path, log_level).unwrap(); - pretty_assertions::assert_eq!( - yaml, - "dbs: -- path: /path/to/db - replica: - type: s3 - bucket: bucket - path: /path/to/backup - access-key-id: access_key_id - secret-access-key: secret_access_key - truncate-page-n: 0 -logging: - level: info -" - ); - } - - #[test] - fn into_yaml_with_snapshot_and_validation() { - let json_litestream = JsonLitestream { - replica: JsonReplica::S3 { - bucket: "bucket".to_owned(), - path: Some("/path/to/backup".to_owned()), - endpoint: None, - region: None, - access_key_id: "access_key_id".to_owned(), - secret_access_key: "secret_access_key".parse().unwrap(), - sync_interval: None, - }, - snapshot: Some(JsonSnapshot { - interval: Some("1h".to_owned()), - retention: Some("24h".to_owned()), - }), - validation: Some(JsonValidation { - interval: Some("6h".to_owned()), - }), - checkpoint: None, - }; - let path = PathBuf::from("/path/to/db"); - let log_level = LogLevel::Info; - let yaml = json_litestream.into_yaml(path, log_level).unwrap(); - pretty_assertions::assert_eq!( - yaml, - "dbs: -- path: /path/to/db - replica: - type: s3 - bucket: bucket - path: /path/to/backup - access-key-id: access_key_id - secret-access-key: secret_access_key - truncate-page-n: 0 -snapshot: - interval: 1h - retention: 24h -validation: - interval: 6h -logging: - level: info -" - ); - } - - #[test] - fn into_yaml_with_checkpoint_config() { - let json_litestream = JsonLitestream { - replica: JsonReplica::S3 { - bucket: "bucket".to_owned(), - path: Some("/path/to/backup".to_owned()), - endpoint: None, - region: None, - access_key_id: "access_key_id".to_owned(), - secret_access_key: "secret_access_key".parse().unwrap(), - sync_interval: None, - }, - snapshot: None, - validation: None, - checkpoint: Some(JsonCheckpoint { - interval: Some("5m".to_owned()), - min_page_count: Some(2000), - truncate_page_n: Some(121_359), - }), - }; - let path = PathBuf::from("/path/to/db"); - let log_level = LogLevel::Info; - let yaml = json_litestream.into_yaml(path, log_level).unwrap(); - pretty_assertions::assert_eq!( - yaml, - "dbs: -- path: /path/to/db - replica: - type: s3 - bucket: bucket - path: /path/to/backup - access-key-id: access_key_id - secret-access-key: secret_access_key - min-checkpoint-page-count: 2000 - checkpoint-interval: 5m - truncate-page-n: 121359 -logging: - level: info -" - ); - } - - #[test] - fn into_yaml_file() { - let json_litestream = JsonLitestream { - replica: JsonReplica::File { - path: PathBuf::from("/path/to/replica"), - sync_interval: Some("5s".to_owned()), - }, - snapshot: None, - validation: None, - checkpoint: None, - }; - let path = PathBuf::from("/path/to/db"); - let log_level = LogLevel::Info; - let yaml = json_litestream.into_yaml(path, log_level).unwrap(); - pretty_assertions::assert_eq!( - yaml, - "dbs: -- path: /path/to/db - replica: - type: file - path: /path/to/replica - sync-interval: 5s - truncate-page-n: 0 -logging: - level: info -" - ); - } - - #[test] - fn into_yaml_sftp() { - let json_litestream = JsonLitestream { - replica: JsonReplica::Sftp { - host: "example.com".to_owned(), - port: 22, - user: "user".to_owned(), - password: Some("pass".parse().unwrap()), - path: Some("/backup".to_owned()), - key_path: None, - sync_interval: None, - }, - snapshot: None, - validation: None, - checkpoint: None, - }; - let path = PathBuf::from("/path/to/db"); - let log_level = LogLevel::Info; - let yaml = json_litestream.into_yaml(path, log_level).unwrap(); - pretty_assertions::assert_eq!( - yaml, - "dbs: -- path: /path/to/db - replica: - type: sftp - host: example.com:22 - user: user - password: pass - path: /backup - truncate-page-n: 0 -logging: - level: info -" - ); - } - - #[test] - fn sanitize_file() { - use bencher_valid::Sanitize as _; - let mut replica = JsonReplica::File { - path: PathBuf::from("/path/to/replica"), - sync_interval: None, - }; - replica.sanitize(); - let JsonReplica::File { path, .. } = replica else { - panic!("expected File") - }; - assert_eq!(path, PathBuf::from("/path/to/replica")); - } - - #[test] - fn sanitize_sftp() { - use bencher_valid::Sanitize as _; - let mut replica = JsonReplica::Sftp { - host: "example.com".to_owned(), - port: 22, - user: "user".to_owned(), - password: Some("secret_pass".parse().unwrap()), - path: None, - key_path: None, - sync_interval: None, - }; - replica.sanitize(); - let JsonReplica::Sftp { password, .. } = replica else { - panic!("expected Sftp") - }; - assert_eq!(password.unwrap().as_ref(), "************"); - } - - #[test] - fn sanitize_s3() { - use bencher_valid::Sanitize as _; - let mut replica = JsonReplica::S3 { - bucket: "mybucket".to_owned(), - path: None, - endpoint: None, - region: None, - access_key_id: "access_key_id".to_owned(), - secret_access_key: "my_secret".parse().unwrap(), - sync_interval: None, - }; - replica.sanitize(); - let JsonReplica::S3 { - secret_access_key, .. - } = replica - else { - panic!("expected S3") - }; - assert_eq!(secret_access_key.as_ref(), "************"); - } -} diff --git a/plus/bencher_litestream/Cargo.toml b/plus/bencher_litestream/Cargo.toml new file mode 100644 index 000000000..ecbc5f139 --- /dev/null +++ b/plus/bencher_litestream/Cargo.toml @@ -0,0 +1,31 @@ +[package] +name = "bencher_litestream" +version.workspace = true +authors.workspace = true +edition.workspace = true +license-file.workspace = true +publish = false + +[features] +default = [] +server = ["bencher_valid/server"] +client = ["bencher_valid/client"] +schema = ["dep:schemars", "bencher_valid/schema"] +yaml = ["dep:serde_yaml", "dep:camino"] +runtime = ["yaml", "dep:tokio", "dep:slog", "dep:thiserror"] + +[dependencies] +bencher_valid.workspace = true +camino = { workspace = true, optional = true } +schemars = { workspace = true, optional = true } +serde.workspace = true +serde_yaml = { workspace = true, optional = true } +slog = { workspace = true, optional = true } +thiserror = { workspace = true, optional = true } +tokio = { workspace = true, optional = true, features = ["process", "rt", "sync"] } + +[dev-dependencies] +pretty_assertions.workspace = true + +[lints] +workspace = true diff --git a/plus/bencher_litestream/src/lib.rs b/plus/bencher_litestream/src/lib.rs new file mode 100644 index 000000000..9bcaa253a --- /dev/null +++ b/plus/bencher_litestream/src/lib.rs @@ -0,0 +1,219 @@ +//! Litestream configuration and runtime for Bencher Plus. +//! +//! The wire types ([`JsonLitestream`] and friends) render to Litestream's YAML +//! config (`yaml` feature), and [`run_litestream`] supervises `litestream restore` +//! then `litestream replicate` around a `SQLite` database replicated to +//! S3-compatible storage (`runtime` feature). The types live in this crate, rather +//! than `bencher_json`, so they can be shared by path without pulling in `bencher_json`. +//! +//! + +use std::path::PathBuf; + +use bencher_valid::{Sanitize, Secret}; +#[cfg(feature = "schema")] +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; + +#[cfg(feature = "runtime")] +mod runtime; +#[cfg(feature = "yaml")] +mod yaml; + +#[cfg(feature = "runtime")] +pub use runtime::{LitestreamError, run_litestream}; + +/// `SQLite` PRAGMA that disables automatic checkpoints so that Litestream manages them. +/// +/// Apply this on the writer connection whenever Litestream is enabled. +/// +/// +/// +pub const DISABLE_AUTOCHECKPOINT_PRAGMA: &str = "PRAGMA wal_autocheckpoint = 0"; + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[cfg_attr(feature = "schema", derive(JsonSchema))] +// https://litestream.io/reference/config/ +pub struct JsonLitestream { + /// Disaster recovery replica + pub replica: JsonReplica, + /// Snapshot configuration + #[serde(skip_serializing_if = "Option::is_none")] + pub snapshot: Option, + /// Validation configuration + #[serde(skip_serializing_if = "Option::is_none")] + pub validation: Option, + /// Checkpoint configuration + #[serde(skip_serializing_if = "Option::is_none")] + pub checkpoint: Option, +} + +impl Sanitize for JsonLitestream { + fn sanitize(&mut self) { + self.replica.sanitize(); + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[cfg_attr(feature = "schema", derive(JsonSchema))] +pub struct JsonSnapshot { + /// How often new snapshots are created + #[serde(skip_serializing_if = "Option::is_none")] + pub interval: Option, + /// How long snapshot & WAL files are kept + #[serde(skip_serializing_if = "Option::is_none")] + pub retention: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[cfg_attr(feature = "schema", derive(JsonSchema))] +pub struct JsonValidation { + /// How often to restore and validate replica data + #[serde(skip_serializing_if = "Option::is_none")] + pub interval: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[cfg_attr(feature = "schema", derive(JsonSchema))] +pub struct JsonCheckpoint { + /// How often to perform a non-blocking PASSIVE checkpoint + #[serde(skip_serializing_if = "Option::is_none")] + pub interval: Option, + /// Minimum WAL pages before a PASSIVE checkpoint triggers + #[serde(skip_serializing_if = "Option::is_none")] + pub min_page_count: Option, + /// Page threshold for blocking TRUNCATE checkpoint (0 to disable) + #[serde(skip_serializing_if = "Option::is_none")] + pub truncate_page_n: Option, +} + +impl JsonCheckpoint { + /// Default value for `truncate_page_n`: disables blocking TRUNCATE checkpoints. + /// When set to 0, Litestream will never perform a blocking TRUNCATE checkpoint, + /// only non-blocking PASSIVE checkpoints (controlled by `interval` / `min_page_count`). + pub const TRUNCATE_DISABLED: u64 = 0; +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[cfg_attr(feature = "schema", derive(JsonSchema))] +#[serde(tag = "scheme", rename_all = "snake_case")] +pub enum JsonReplica { + // https://litestream.io/reference/config/#file-replica + File { + path: PathBuf, + #[serde(skip_serializing_if = "Option::is_none")] + sync_interval: Option, + }, + // https://litestream.io/guides/sftp/ + Sftp { + host: String, + port: u16, + user: String, + #[serde(skip_serializing_if = "Option::is_none")] + password: Option, + #[serde(skip_serializing_if = "Option::is_none")] + path: Option, + #[serde(skip_serializing_if = "Option::is_none")] + key_path: Option, + #[serde(skip_serializing_if = "Option::is_none")] + sync_interval: Option, + }, + // https://litestream.io/guides/s3/ + S3 { + bucket: String, + #[serde(skip_serializing_if = "Option::is_none")] + path: Option, + #[serde(skip_serializing_if = "Option::is_none")] + endpoint: Option, + #[serde(skip_serializing_if = "Option::is_none")] + region: Option, + access_key_id: String, + secret_access_key: Secret, + #[serde(skip_serializing_if = "Option::is_none")] + sync_interval: Option, + }, +} + +impl Sanitize for JsonReplica { + fn sanitize(&mut self) { + match self { + Self::File { .. } => {}, + Self::Sftp { password, .. } => password.sanitize(), + Self::S3 { + secret_access_key, .. + } => secret_access_key.sanitize(), + } + } +} + +/// Litestream log level. +/// +/// Mirrors the values Litestream accepts for `logging.level`. Construct it directly, +/// or, in `bencher`, from a `bencher_json` `LogLevel` via its `From` impl. +#[derive(Debug, Clone, Copy, Serialize)] +#[serde(rename_all = "kebab-case")] +pub enum LitestreamLevel { + Debug, + Info, + Warn, + Error, +} + +#[cfg(test)] +mod tests { + use bencher_valid::Sanitize as _; + + use super::JsonReplica; + + #[test] + fn sanitize_file() { + let mut replica = JsonReplica::File { + path: std::path::PathBuf::from("/path/to/replica"), + sync_interval: None, + }; + replica.sanitize(); + let JsonReplica::File { path, .. } = replica else { + panic!("expected File") + }; + assert_eq!(path, std::path::PathBuf::from("/path/to/replica")); + } + + #[test] + fn sanitize_sftp() { + let mut replica = JsonReplica::Sftp { + host: "example.com".to_owned(), + port: 22, + user: "user".to_owned(), + password: Some("secret_pass".parse().unwrap()), + path: None, + key_path: None, + sync_interval: None, + }; + replica.sanitize(); + let JsonReplica::Sftp { password, .. } = replica else { + panic!("expected Sftp") + }; + assert_eq!(password.unwrap().as_ref(), "************"); + } + + #[test] + fn sanitize_s3() { + let mut replica = JsonReplica::S3 { + bucket: "mybucket".to_owned(), + path: None, + endpoint: None, + region: None, + access_key_id: "access_key_id".to_owned(), + secret_access_key: "my_secret".parse().unwrap(), + sync_interval: None, + }; + replica.sanitize(); + let JsonReplica::S3 { + secret_access_key, .. + } = replica + else { + panic!("expected S3") + }; + assert_eq!(secret_access_key.as_ref(), "************"); + } +} diff --git a/plus/bencher_litestream/src/runtime.rs b/plus/bencher_litestream/src/runtime.rs new file mode 100644 index 000000000..e8acc1d66 --- /dev/null +++ b/plus/bencher_litestream/src/runtime.rs @@ -0,0 +1,101 @@ +use camino::Utf8PathBuf; +use slog::Logger; +use tokio::{process::Command, sync::oneshot, task::JoinHandle}; + +use crate::{JsonLitestream, LitestreamLevel}; + +/// Render the Litestream config to `config_path`, then supervise `litestream restore` +/// followed by `litestream replicate` around the `SQLite` database at `db_path`. +/// +/// `db_path` must be absolute. Once the one-shot restore completes, `restore_tx` is +/// signaled (the database file now exists) and replication continues on the returned +/// [`JoinHandle`] for the lifetime of the process. The `litestream` binary is resolved +/// on `PATH`. +pub fn run_litestream( + log: &Logger, + litestream: JsonLitestream, + db_path: Utf8PathBuf, + config_path: Utf8PathBuf, + log_level: LitestreamLevel, + restore_tx: oneshot::Sender<()>, +) -> Result>, LitestreamError> { + let yaml = litestream + .into_yaml(db_path.clone(), log_level) + .map_err(LitestreamError::Yaml)?; + std::fs::write(&config_path, yaml) + .map_err(|e| LitestreamError::WriteYaml(config_path.clone(), e))?; + + let litestream_logger = log.clone(); + Ok(tokio::spawn(async move { + // https://litestream.io/reference/restore/ + let restore = Command::new("litestream") + .arg("restore") + .arg("-if-replica-exists") + .arg("-if-db-not-exists") + .arg("-config") + .arg(&config_path) + .arg("-no-expand-env") + .arg(&db_path) + .output() + .await + .map_err(LitestreamError::Restore)?; + slog::info!(litestream_logger, "Litestream restore: {restore:?}"); + if !restore.status.success() { + return Err(LitestreamError::RestoreExit { + status: restore.status, + stdout: String::from_utf8_lossy(&restore.stdout).into_owned(), + stderr: String::from_utf8_lossy(&restore.stderr).into_owned(), + }); + } + + // Signal the caller that restore is complete (the DB file exists) + restore_tx.send(()).map_err(LitestreamError::RestoreSend)?; + + // https://litestream.io/reference/replicate/ + let mut replicate = Command::new("litestream") + .arg("replicate") + .arg("-config") + .arg(&config_path) + .arg("-no-expand-env") + .spawn() + .map_err(LitestreamError::Replicate)?; + // Litestream should run indefinitely + Err(LitestreamError::ReplicateExit( + replicate.wait().await.map_err(LitestreamError::Replicate)?, + )) + })) +} + +#[derive(Debug, thiserror::Error)] +pub enum LitestreamError { + #[error("Database path is not valid UTF-8: {0:?}")] + DatabasePathNotUtf8(std::path::PathBuf), + #[error("Failed to absolutize the database path: {0}")] + Database(std::io::Error), + #[error( + "Failed to convert Bencher config to Litestream config. This is likely a bug. Please report this: {0}" + )] + Yaml(serde_yaml::Error), + #[error("Failed to write Litestream config ({0}): {1}")] + WriteYaml(Utf8PathBuf, std::io::Error), + #[error("Failed to run `litestream restore`: {0}")] + Restore(std::io::Error), + #[error("Failed to run `litestream replicate`: {0}")] + Replicate(std::io::Error), + #[error("Failed to restore (exit status {status})\nstdout: {stdout}\nstderr: {stderr}")] + RestoreExit { + status: std::process::ExitStatus, + stdout: String, + stderr: String, + }, + #[error( + "Failed to send restore completion message: receiver dropped, server likely crashed during startup" + )] + RestoreSend(()), + #[error("Failed to receive restore completion message")] + RestoreRecv(oneshot::error::RecvError), + #[error("Failed to replicate: {0}")] + ReplicateExit(std::process::ExitStatus), + #[error("Failed to join Litestream handle: {0}")] + JoinHandle(tokio::task::JoinError), +} diff --git a/plus/bencher_litestream/src/yaml.rs b/plus/bencher_litestream/src/yaml.rs new file mode 100644 index 000000000..a0e598fb4 --- /dev/null +++ b/plus/bencher_litestream/src/yaml.rs @@ -0,0 +1,413 @@ +use std::path::PathBuf; + +use bencher_valid::Secret; +use camino::Utf8PathBuf; +use serde::Serialize; + +use crate::{ + JsonCheckpoint, JsonLitestream, JsonReplica, JsonSnapshot, JsonValidation, LitestreamLevel, +}; + +impl JsonLitestream { + pub fn into_yaml( + self, + path: Utf8PathBuf, + log_level: LitestreamLevel, + ) -> Result { + let Self { + replica, + snapshot, + validation, + checkpoint, + } = self; + let replica = LitestreamReplica::from(replica); + let snapshot = snapshot.map(LitestreamSnapshot::from); + let validation = validation.map(LitestreamValidation::from); + let (min_checkpoint_page_count, checkpoint_interval, truncate_page_n) = + checkpoint.map_or((None, None, JsonCheckpoint::TRUNCATE_DISABLED), |c| { + let JsonCheckpoint { + interval, + min_page_count, + truncate_page_n, + } = c; + ( + min_page_count, + interval, + truncate_page_n.unwrap_or(JsonCheckpoint::TRUNCATE_DISABLED), + ) + }); + let dbs = vec![LitestreamDb { + path, + replica, + min_checkpoint_page_count, + checkpoint_interval, + truncate_page_n, + }]; + let logging = Some(LitestreamLogging { + level: Some(log_level), + }); + let litestream = Litestream { + dbs, + snapshot, + validation, + logging, + }; + serde_yaml::to_string(&litestream) + } +} + +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "kebab-case")] +struct Litestream { + dbs: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + snapshot: Option, + #[serde(skip_serializing_if = "Option::is_none")] + validation: Option, + #[serde(skip_serializing_if = "Option::is_none")] + logging: Option, +} + +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "kebab-case")] +struct LitestreamDb { + path: Utf8PathBuf, + replica: LitestreamReplica, + #[serde(skip_serializing_if = "Option::is_none")] + min_checkpoint_page_count: Option, + #[serde(skip_serializing_if = "Option::is_none")] + checkpoint_interval: Option, + truncate_page_n: u64, +} + +#[derive(Debug, Clone, Serialize)] +#[serde( + tag = "type", + rename_all = "kebab-case", + rename_all_fields = "kebab-case" +)] +enum LitestreamReplica { + File { + path: PathBuf, + #[serde(skip_serializing_if = "Option::is_none")] + sync_interval: Option, + }, + Sftp { + host: String, + user: String, + #[serde(skip_serializing_if = "Option::is_none")] + password: Option, + #[serde(skip_serializing_if = "Option::is_none")] + path: Option, + #[serde(skip_serializing_if = "Option::is_none")] + key_path: Option, + #[serde(skip_serializing_if = "Option::is_none")] + sync_interval: Option, + }, + S3 { + bucket: String, + #[serde(skip_serializing_if = "Option::is_none")] + path: Option, + #[serde(skip_serializing_if = "Option::is_none")] + endpoint: Option, + #[serde(skip_serializing_if = "Option::is_none")] + region: Option, + access_key_id: String, + secret_access_key: Secret, + #[serde(skip_serializing_if = "Option::is_none")] + sync_interval: Option, + }, +} + +impl From for LitestreamReplica { + fn from(replica: JsonReplica) -> Self { + match replica { + JsonReplica::File { + path, + sync_interval, + } => Self::File { + path, + sync_interval, + }, + JsonReplica::Sftp { + host, + port, + user, + password, + path, + key_path, + sync_interval, + } => Self::Sftp { + host: format!("{host}:{port}"), + user, + password, + path, + key_path, + sync_interval, + }, + JsonReplica::S3 { + bucket, + path, + endpoint, + region, + access_key_id, + secret_access_key, + sync_interval, + } => Self::S3 { + bucket, + path, + endpoint, + region, + access_key_id, + secret_access_key, + sync_interval, + }, + } + } +} + +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "kebab-case")] +struct LitestreamSnapshot { + #[serde(skip_serializing_if = "Option::is_none")] + interval: Option, + #[serde(skip_serializing_if = "Option::is_none")] + retention: Option, +} + +impl From for LitestreamSnapshot { + fn from(snapshot: JsonSnapshot) -> Self { + let JsonSnapshot { + interval, + retention, + } = snapshot; + Self { + interval, + retention, + } + } +} + +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "kebab-case")] +struct LitestreamValidation { + #[serde(skip_serializing_if = "Option::is_none")] + interval: Option, +} + +impl From for LitestreamValidation { + fn from(validation: JsonValidation) -> Self { + let JsonValidation { interval } = validation; + Self { interval } + } +} + +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "kebab-case")] +struct LitestreamLogging { + #[serde(skip_serializing_if = "Option::is_none")] + level: Option, +} + +#[cfg(test)] +mod tests { + use camino::Utf8PathBuf; + + use crate::{ + JsonCheckpoint, JsonLitestream, JsonReplica, JsonSnapshot, JsonValidation, LitestreamLevel, + }; + + #[test] + fn into_yaml() { + let json_litestream = JsonLitestream { + replica: JsonReplica::S3 { + bucket: "bucket".to_owned(), + path: Some("/path/to/backup".to_owned()), + endpoint: None, + region: None, + access_key_id: "access_key_id".to_owned(), + secret_access_key: "secret_access_key".parse().unwrap(), + sync_interval: None, + }, + snapshot: None, + validation: None, + checkpoint: None, + }; + let path = Utf8PathBuf::from("/path/to/db"); + let yaml = json_litestream + .into_yaml(path, LitestreamLevel::Info) + .unwrap(); + pretty_assertions::assert_eq!( + yaml, + "dbs: +- path: /path/to/db + replica: + type: s3 + bucket: bucket + path: /path/to/backup + access-key-id: access_key_id + secret-access-key: secret_access_key + truncate-page-n: 0 +logging: + level: info +" + ); + } + + #[test] + fn into_yaml_with_snapshot_and_validation() { + let json_litestream = JsonLitestream { + replica: JsonReplica::S3 { + bucket: "bucket".to_owned(), + path: Some("/path/to/backup".to_owned()), + endpoint: None, + region: None, + access_key_id: "access_key_id".to_owned(), + secret_access_key: "secret_access_key".parse().unwrap(), + sync_interval: None, + }, + snapshot: Some(JsonSnapshot { + interval: Some("1h".to_owned()), + retention: Some("24h".to_owned()), + }), + validation: Some(JsonValidation { + interval: Some("6h".to_owned()), + }), + checkpoint: None, + }; + let path = Utf8PathBuf::from("/path/to/db"); + let yaml = json_litestream + .into_yaml(path, LitestreamLevel::Info) + .unwrap(); + pretty_assertions::assert_eq!( + yaml, + "dbs: +- path: /path/to/db + replica: + type: s3 + bucket: bucket + path: /path/to/backup + access-key-id: access_key_id + secret-access-key: secret_access_key + truncate-page-n: 0 +snapshot: + interval: 1h + retention: 24h +validation: + interval: 6h +logging: + level: info +" + ); + } + + #[test] + fn into_yaml_with_checkpoint_config() { + let json_litestream = JsonLitestream { + replica: JsonReplica::S3 { + bucket: "bucket".to_owned(), + path: Some("/path/to/backup".to_owned()), + endpoint: None, + region: None, + access_key_id: "access_key_id".to_owned(), + secret_access_key: "secret_access_key".parse().unwrap(), + sync_interval: None, + }, + snapshot: None, + validation: None, + checkpoint: Some(JsonCheckpoint { + interval: Some("5m".to_owned()), + min_page_count: Some(2000), + truncate_page_n: Some(121_359), + }), + }; + let path = Utf8PathBuf::from("/path/to/db"); + let yaml = json_litestream + .into_yaml(path, LitestreamLevel::Info) + .unwrap(); + pretty_assertions::assert_eq!( + yaml, + "dbs: +- path: /path/to/db + replica: + type: s3 + bucket: bucket + path: /path/to/backup + access-key-id: access_key_id + secret-access-key: secret_access_key + min-checkpoint-page-count: 2000 + checkpoint-interval: 5m + truncate-page-n: 121359 +logging: + level: info +" + ); + } + + #[test] + fn into_yaml_file() { + let json_litestream = JsonLitestream { + replica: JsonReplica::File { + path: std::path::PathBuf::from("/path/to/replica"), + sync_interval: Some("5s".to_owned()), + }, + snapshot: None, + validation: None, + checkpoint: None, + }; + let path = Utf8PathBuf::from("/path/to/db"); + let yaml = json_litestream + .into_yaml(path, LitestreamLevel::Info) + .unwrap(); + pretty_assertions::assert_eq!( + yaml, + "dbs: +- path: /path/to/db + replica: + type: file + path: /path/to/replica + sync-interval: 5s + truncate-page-n: 0 +logging: + level: info +" + ); + } + + #[test] + fn into_yaml_sftp() { + let json_litestream = JsonLitestream { + replica: JsonReplica::Sftp { + host: "example.com".to_owned(), + port: 22, + user: "user".to_owned(), + password: Some("pass".parse().unwrap()), + path: Some("/backup".to_owned()), + key_path: None, + sync_interval: None, + }, + snapshot: None, + validation: None, + checkpoint: None, + }; + let path = Utf8PathBuf::from("/path/to/db"); + let yaml = json_litestream + .into_yaml(path, LitestreamLevel::Info) + .unwrap(); + pretty_assertions::assert_eq!( + yaml, + "dbs: +- path: /path/to/db + replica: + type: sftp + host: example.com:22 + user: user + password: pass + path: /backup + truncate-page-n: 0 +logging: + level: info +" + ); + } +} diff --git a/services/api/Cargo.toml b/services/api/Cargo.toml index b14f14ab9..7c74aef65 100644 --- a/services/api/Cargo.toml +++ b/services/api/Cargo.toml @@ -12,6 +12,8 @@ plus = [ "dep:api_oci", "dep:api_runners", "dep:api_specs", + "dep:bencher_litestream", + "dep:camino", "api_auth/plus", "api_checkout/plus", "api_oci/plus", @@ -76,16 +78,17 @@ api_users.workspace = true bencher_config.workspace = true bencher_endpoint.workspace = true bencher_json = { workspace = true, features = ["server", "schema", "db"] } +bencher_litestream = { workspace = true, optional = true, features = ["runtime", "server"] } bencher_logger.workspace = true bencher_schema.workspace = true bencher_otel = { workspace = true, optional = true } bencher_otel_provider = { workspace = true, optional = true } +camino = { workspace = true, optional = true } dropshot.workspace = true futures-concurrency.workspace = true futures-util.workspace = true libsqlite3-sys.workspace = true sentry = { workspace = true, optional = true } -serde_yaml.workspace = true slog.workspace = true thiserror.workspace = true tokio = { workspace = true, features = ["fs", "rt-multi-thread"] } diff --git a/services/api/Dockerfile b/services/api/Dockerfile index 26656c870..15ab0134f 100644 --- a/services/api/Dockerfile +++ b/services/api/Dockerfile @@ -68,6 +68,7 @@ COPY plus/bencher_github_client bencher_github_client COPY plus/bencher_google_client bencher_google_client COPY plus/bencher_google_index bencher_google_index COPY plus/bencher_license bencher_license +COPY plus/bencher_litestream bencher_litestream COPY plus/bencher_oci_storage bencher_oci_storage COPY plus/bencher_otel bencher_otel COPY plus/bencher_otel_provider bencher_otel_provider diff --git a/services/api/src/lib.rs b/services/api/src/lib.rs index 58b6f4410..6f97071bf 100644 --- a/services/api/src/lib.rs +++ b/services/api/src/lib.rs @@ -3,16 +3,19 @@ use api_checkout as _; use bencher_config as _; use bencher_json as _; +#[cfg(feature = "plus")] +use bencher_litestream as _; use bencher_logger as _; #[cfg(feature = "otel")] use bencher_otel as _; #[cfg(any(feature = "plus", feature = "otel"))] use bencher_otel_provider as _; +#[cfg(feature = "plus")] +use camino as _; use futures_concurrency as _; use futures_util as _; #[cfg(feature = "sentry")] use sentry as _; -use serde_yaml as _; use slog as _; use thiserror as _; use tokio as _; diff --git a/services/api/src/main.rs b/services/api/src/main.rs index e7c88096c..601d02a1b 100644 --- a/services/api/src/main.rs +++ b/services/api/src/main.rs @@ -3,16 +3,16 @@ reason = "dependencies used by lib but not binary" )] -#[cfg(feature = "sentry")] -use std::path::PathBuf; use std::sync::Arc; use bencher_api::api::Api; use bencher_config::{Config, ConfigTx}; use bencher_json::BENCHER_API_VERSION; #[cfg(feature = "plus")] -use bencher_json::system::config::JsonLitestream; +use bencher_litestream::{LitestreamError, LitestreamLevel, run_litestream}; use bencher_schema::context::ApiContext; +#[cfg(feature = "plus")] +use camino::Utf8PathBuf; use dropshot::HttpServer; use futures_concurrency::future::Race as _; use futures_util::FutureExt as _; @@ -20,11 +20,7 @@ use futures_util::FutureExt as _; use sentry::ClientInitGuard; use slog::{Logger, error, info}; #[cfg(feature = "plus")] -use tokio::process::Command; -#[cfg(feature = "plus")] use tokio::sync; -#[cfg(feature = "plus")] -use tokio::task::JoinHandle; use tokio_rustls::rustls::crypto::{CryptoProvider, aws_lc_rs}; #[derive(Debug, thiserror::Error)] @@ -103,8 +99,25 @@ async fn run( .as_ref() .and_then(|plus| plus.litestream.clone()) { + // Absolutize the database path so Litestream resolves it independently of the CWD. + let db_path = if config.database.file.is_absolute() { + config.database.file.clone() + } else { + std::env::current_dir() + .map_err(LitestreamError::Database)? + .join(&config.database.file) + }; + let db_path = + Utf8PathBuf::from_path_buf(db_path).map_err(LitestreamError::DatabasePathNotUtf8)?; + #[cfg(debug_assertions)] + let config_path = Utf8PathBuf::from("etc/litestream.yml"); + #[cfg(not(debug_assertions))] + let config_path = Utf8PathBuf::from("/etc/litestream.yml"); + let log_level = LitestreamLevel::from(config.logging.log.level()); + let (restore_tx, restore_rx) = sync::oneshot::channel(); - let litestream_handle = run_litestream(log, &config, litestream, restore_tx)?; + let litestream_handle = + run_litestream(log, litestream, db_path, config_path, log_level, restore_tx)?; // Wait for Litestream restore to complete (replicate starts in background) restore_rx.await.map_err(LitestreamError::RestoreRecv)?; @@ -179,105 +192,6 @@ fn init_sentry(log: &Logger, config: &Config) -> Option { }) } -#[cfg(feature = "plus")] -#[derive(Debug, thiserror::Error)] -pub enum LitestreamError { - #[error("Failed to absolutize the database path: {0}")] - Database(std::io::Error), - #[error( - "Failed to convert Bencher config to Litestream config. This is likely a bug. Please report this: {0}" - )] - Yaml(serde_yaml::Error), - #[error("Failed to write Litestream config ({0}): {1}")] - WriteYaml(PathBuf, std::io::Error), - #[error("Failed to run `litestream restore`: {0}")] - Restore(std::io::Error), - #[error("Failed to run `litestream replicate`: {0}")] - Replicate(std::io::Error), - #[error("Failed to restore (exit status {status})\nstdout: {stdout}\nstderr: {stderr}")] - RestoreExit { - status: std::process::ExitStatus, - stdout: String, - stderr: String, - }, - #[error( - "Failed to send restore completion message: receiver dropped, server likely crashed during startup" - )] - RestoreSend(()), - #[error("Failed to receive restore completion message")] - RestoreRecv(sync::oneshot::error::RecvError), - #[error("Failed to replicate: {0}")] - ReplicateExit(std::process::ExitStatus), - #[error("Failed to join Litestream handle: {0}")] - JoinHandle(tokio::task::JoinError), -} - -#[cfg(feature = "plus")] -fn run_litestream( - log: &Logger, - config: &Config, - litestream: JsonLitestream, - restore_tx: sync::oneshot::Sender<()>, -) -> Result>, LitestreamError> { - // Get the absolute database path from the config - let db_path = if config.database.file.is_absolute() { - config.database.file.clone() - } else { - std::env::current_dir() - .map_err(LitestreamError::Database)? - .join(&config.database.file) - }; - #[cfg(debug_assertions)] - let config_path = PathBuf::from("etc/litestream.yml"); - #[cfg(not(debug_assertions))] - let config_path = PathBuf::from("/etc/litestream.yml"); - let yaml = litestream - .into_yaml(db_path.clone(), config.logging.log.level()) - .map_err(LitestreamError::Yaml)?; - std::fs::write(&config_path, yaml) - .map_err(|e| LitestreamError::WriteYaml(config_path.clone(), e))?; - - let litestream_logger = log.clone(); - Ok(tokio::spawn(async move { - // https://litestream.io/reference/restore/ - let restore = Command::new("litestream") - .arg("restore") - .arg("-if-replica-exists") - .arg("-if-db-not-exists") - .arg("-config") - .arg(&config_path) - .arg("-no-expand-env") - .arg(&db_path) - .output() - .await - .map_err(LitestreamError::Restore)?; - slog::info!(litestream_logger, "Litestream restore: {restore:?}"); - if !restore.status.success() { - return Err(LitestreamError::RestoreExit { - status: restore.status, - stdout: String::from_utf8_lossy(&restore.stdout).into_owned(), - stderr: String::from_utf8_lossy(&restore.stderr).into_owned(), - }); - } - - // Signal the server that restore is complete (DB file exists) - restore_tx.send(()).map_err(LitestreamError::RestoreSend)?; - - // https://litestream.io/reference/replicate/ - let mut replicate = Command::new("litestream") - .arg("replicate") - .arg("-config") - .arg(&config_path) - .arg("-no-expand-env") - .spawn() - .map_err(LitestreamError::Replicate)?; - // Litestream should run indefinitely - Err(LitestreamError::ReplicateExit( - replicate.wait().await.map_err(LitestreamError::Replicate)?, - )) - })) -} - async fn create_api_server(config: Config) -> Result, ApiError> { #[cfg(feature = "otel")] bencher_otel::ApiMeter::increment(bencher_otel::ApiCounter::ServerStartup); diff --git a/services/cli/Dockerfile b/services/cli/Dockerfile index 97336e0ab..50252b38b 100644 --- a/services/cli/Dockerfile +++ b/services/cli/Dockerfile @@ -63,6 +63,7 @@ RUN cargo init --lib bencher_recaptcha RUN cargo init --lib bencher_rootfs RUN cargo init --lib bencher_runner RUN cargo init --lib bencher_init +COPY plus/bencher_litestream bencher_litestream WORKDIR /usr/src/bencher/tasks RUN cargo init --bin bin_version diff --git a/services/console/Dockerfile b/services/console/Dockerfile index e896de659..8e87dec05 100644 --- a/services/console/Dockerfile +++ b/services/console/Dockerfile @@ -37,6 +37,7 @@ RUN cargo init --lib bencher_github_client RUN cargo init --lib bencher_google_client RUN cargo init --lib bencher_google_index RUN cargo init --lib bencher_license +RUN cargo init --lib bencher_litestream RUN cargo init --lib bencher_oci RUN cargo init --lib bencher_oci_storage RUN cargo init --lib bencher_output_protocol