mirror of
https://tangled.org/tranquil.farm/tranquil-pds
synced 2026-08-25 02:36:06 +00:00
Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
  LewisandTangled
|
b1d86caa78 | ||
|
Lewis
|
9b2cfb3a7e | ||
|
Lewis
|
efd499bb26 |
@@ -68,6 +68,10 @@ test-group = "serial-env-tests"
|
||||
filter = "package(tranquil-signal)"
|
||||
test-group = "serial-env-tests"
|
||||
|
||||
[[profile.default.overrides]]
|
||||
filter = "package(tranquil-config)"
|
||||
test-group = "serial-env-tests"
|
||||
|
||||
[[profile.default.overrides]]
|
||||
filter = "binary(whole_story)"
|
||||
test-group = "heavy-load-tests"
|
||||
@@ -118,6 +122,10 @@ test-group = "serial-env-tests"
|
||||
filter = "package(tranquil-signal)"
|
||||
test-group = "serial-env-tests"
|
||||
|
||||
[[profile.ci.overrides]]
|
||||
filter = "package(tranquil-config)"
|
||||
test-group = "serial-env-tests"
|
||||
|
||||
[[profile.ci.overrides]]
|
||||
filter = "binary(whole_story)"
|
||||
test-group = "heavy-load-tests"
|
||||
|
||||
Generated
+110
-30
@@ -9,7 +9,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "087113bd50d9adce24850eed5d0476c7d199d532fce8fab5173650331e09033a"
|
||||
dependencies = [
|
||||
"abnf-core",
|
||||
"nom",
|
||||
"nom 7.1.3",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -18,7 +18,7 @@ version = "0.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c44e09c43ae1c368fb91a03a566472d0087c26cf7e1b9e8e289c14ede681dd7d"
|
||||
dependencies = [
|
||||
"nom",
|
||||
"nom 7.1.3",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -213,7 +213,7 @@ dependencies = [
|
||||
"asn1-rs-derive",
|
||||
"asn1-rs-impl",
|
||||
"displaydoc",
|
||||
"nom",
|
||||
"nom 7.1.3",
|
||||
"num-traits",
|
||||
"rusticata-macros",
|
||||
"thiserror 1.0.69",
|
||||
@@ -1972,7 +1972,7 @@ checksum = "5cd0a5c643689626bec213c4d8bd4d96acc8ffdb4ad4bb6bc16abf27d5f4b553"
|
||||
dependencies = [
|
||||
"asn1-rs",
|
||||
"displaydoc",
|
||||
"nom",
|
||||
"nom 7.1.3",
|
||||
"num-bigint",
|
||||
"num-traits",
|
||||
"rusticata-macros",
|
||||
@@ -2216,6 +2216,22 @@ dependencies = [
|
||||
"zeroize",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "email-encoding"
|
||||
version = "0.4.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9298e6504d9b9e780ed3f7dfd43a61be8cd0e09eb07f7706a945b0072b6670b6"
|
||||
dependencies = [
|
||||
"base64 0.22.1",
|
||||
"memchr",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "email_address"
|
||||
version = "0.2.9"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e079f19b08ca6239f47f8ba8509c11cf3ea30095831f7fed61441475edd8c449"
|
||||
|
||||
[[package]]
|
||||
name = "embedded-io"
|
||||
version = "0.4.0"
|
||||
@@ -3780,6 +3796,37 @@ version = "0.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2"
|
||||
|
||||
[[package]]
|
||||
name = "lettre"
|
||||
version = "0.11.21"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "dabda5859ee7c06b995b9d1165aa52c39110e079ef609db97178d86aeb051fa7"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"base64 0.22.1",
|
||||
"ed25519-dalek",
|
||||
"email-encoding",
|
||||
"email_address",
|
||||
"fastrand",
|
||||
"futures-io",
|
||||
"futures-util",
|
||||
"httpdate",
|
||||
"idna",
|
||||
"mime",
|
||||
"nom 8.0.0",
|
||||
"percent-encoding",
|
||||
"quoted_printable",
|
||||
"rsa",
|
||||
"rustls 0.23.37",
|
||||
"sha2",
|
||||
"socket2 0.6.3",
|
||||
"tokio",
|
||||
"tokio-rustls 0.26.4",
|
||||
"tracing",
|
||||
"url",
|
||||
"webpki-roots 1.0.6",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "libc"
|
||||
version = "0.2.183"
|
||||
@@ -4409,6 +4456,15 @@ dependencies = [
|
||||
"minimal-lexical",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "nom"
|
||||
version = "8.0.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "df9761775871bdef83bee530e60050f7e54b1105350d6884eb0fb4f46c2f9405"
|
||||
dependencies = [
|
||||
"memchr",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "nonzero_ext"
|
||||
version = "0.3.0"
|
||||
@@ -4839,7 +4895,7 @@ checksum = "9114f9c1683dd09c5f4fa024c89fdad783eaae21d3d52dd23ddaaffa29ffb168"
|
||||
dependencies = [
|
||||
"either",
|
||||
"fnv",
|
||||
"nom",
|
||||
"nom 7.1.3",
|
||||
"once_cell",
|
||||
"postcard",
|
||||
"quick-xml",
|
||||
@@ -5427,6 +5483,12 @@ dependencies = [
|
||||
"proc-macro2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "quoted_printable"
|
||||
version = "0.5.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "478e0585659a122aa407eb7e3c0e1fa51b1d8a870038bd29f0cf4a8551eea972"
|
||||
|
||||
[[package]]
|
||||
name = "r-efi"
|
||||
version = "5.3.0"
|
||||
@@ -5833,7 +5895,7 @@ version = "4.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "faf0c4a6ece9950b9abdb62b1cfcf2a68b3b67a10ba445b3bb85be2a293d0632"
|
||||
dependencies = [
|
||||
"nom",
|
||||
"nom 7.1.3",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -6050,6 +6112,16 @@ dependencies = [
|
||||
"zeroize",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "secrecy"
|
||||
version = "0.10.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e891af845473308773346dc847b2c23ee78fe442e0472ac50e22a18a93d3ae5a"
|
||||
dependencies = [
|
||||
"serde",
|
||||
"zeroize",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "security-framework"
|
||||
version = "3.7.0"
|
||||
@@ -7455,7 +7527,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tranquil-api"
|
||||
version = "0.5.7"
|
||||
version = "0.6.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"axum",
|
||||
@@ -7506,7 +7578,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tranquil-auth"
|
||||
version = "0.5.7"
|
||||
version = "0.6.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"base32",
|
||||
@@ -7529,7 +7601,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tranquil-cache"
|
||||
version = "0.5.7"
|
||||
version = "0.6.0"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"base64 0.22.1",
|
||||
@@ -7543,11 +7615,19 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tranquil-comms"
|
||||
version = "0.5.7"
|
||||
version = "0.6.0"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"base64 0.22.1",
|
||||
"chrono",
|
||||
"ed25519-dalek",
|
||||
"futures",
|
||||
"hickory-resolver",
|
||||
"lettre",
|
||||
"rand 0.8.5",
|
||||
"reqwest",
|
||||
"rsa",
|
||||
"secrecy",
|
||||
"serde_json",
|
||||
"sqlx",
|
||||
"thiserror 2.0.18",
|
||||
@@ -7561,7 +7641,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tranquil-config"
|
||||
version = "0.5.7"
|
||||
version = "0.6.0"
|
||||
dependencies = [
|
||||
"confique",
|
||||
"serde",
|
||||
@@ -7569,7 +7649,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tranquil-crypto"
|
||||
version = "0.5.7"
|
||||
version = "0.6.0"
|
||||
dependencies = [
|
||||
"aes-gcm",
|
||||
"base64 0.22.1",
|
||||
@@ -7585,7 +7665,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tranquil-db"
|
||||
version = "0.5.7"
|
||||
version = "0.6.0"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"chrono",
|
||||
@@ -7602,7 +7682,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tranquil-db-traits"
|
||||
version = "0.5.7"
|
||||
version = "0.6.0"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"base64 0.22.1",
|
||||
@@ -7618,7 +7698,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tranquil-infra"
|
||||
version = "0.5.7"
|
||||
version = "0.6.0"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"bytes",
|
||||
@@ -7629,7 +7709,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tranquil-lexicon"
|
||||
version = "0.5.7"
|
||||
version = "0.6.0"
|
||||
dependencies = [
|
||||
"chrono",
|
||||
"futures",
|
||||
@@ -7648,7 +7728,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tranquil-oauth"
|
||||
version = "0.5.7"
|
||||
version = "0.6.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"axum",
|
||||
@@ -7671,7 +7751,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tranquil-oauth-server"
|
||||
version = "0.5.7"
|
||||
version = "0.6.0"
|
||||
dependencies = [
|
||||
"axum",
|
||||
"base64 0.22.1",
|
||||
@@ -7704,7 +7784,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tranquil-pds"
|
||||
version = "0.5.7"
|
||||
version = "0.6.0"
|
||||
dependencies = [
|
||||
"aes-gcm",
|
||||
"anyhow",
|
||||
@@ -7796,7 +7876,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tranquil-repo"
|
||||
version = "0.5.7"
|
||||
version = "0.6.0"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"cid",
|
||||
@@ -7808,7 +7888,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tranquil-ripple"
|
||||
version = "0.5.7"
|
||||
version = "0.6.0"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"backon",
|
||||
@@ -7833,7 +7913,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tranquil-scopes"
|
||||
version = "0.5.7"
|
||||
version = "0.6.0"
|
||||
dependencies = [
|
||||
"axum",
|
||||
"futures",
|
||||
@@ -7849,7 +7929,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tranquil-server"
|
||||
version = "0.5.7"
|
||||
version = "0.6.0"
|
||||
dependencies = [
|
||||
"axum",
|
||||
"clap",
|
||||
@@ -7870,7 +7950,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tranquil-signal"
|
||||
version = "0.5.7"
|
||||
version = "0.6.0"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"chrono",
|
||||
@@ -7893,7 +7973,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tranquil-storage"
|
||||
version = "0.5.7"
|
||||
version = "0.6.0"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"aws-config",
|
||||
@@ -7910,7 +7990,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tranquil-store"
|
||||
version = "0.5.7"
|
||||
version = "0.6.0"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"bytes",
|
||||
@@ -7959,7 +8039,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tranquil-sync"
|
||||
version = "0.5.7"
|
||||
version = "0.6.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"axum",
|
||||
@@ -7981,7 +8061,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tranquil-types"
|
||||
version = "0.5.7"
|
||||
version = "0.6.0"
|
||||
dependencies = [
|
||||
"chrono",
|
||||
"cid",
|
||||
@@ -8496,7 +8576,7 @@ dependencies = [
|
||||
"base64urlsafedata",
|
||||
"der-parser",
|
||||
"hex",
|
||||
"nom",
|
||||
"nom 7.1.3",
|
||||
"openssl",
|
||||
"openssl-sys",
|
||||
"rand 0.9.2",
|
||||
@@ -9068,7 +9148,7 @@ dependencies = [
|
||||
"data-encoding",
|
||||
"der-parser",
|
||||
"lazy_static",
|
||||
"nom",
|
||||
"nom 7.1.3",
|
||||
"oid-registry",
|
||||
"rusticata-macros",
|
||||
"thiserror 1.0.69",
|
||||
|
||||
+4
-1
@@ -26,7 +26,7 @@ members = [
|
||||
]
|
||||
|
||||
[workspace.package]
|
||||
version = "0.5.7"
|
||||
version = "0.6.0"
|
||||
edition = "2024"
|
||||
license = "AGPL-3.0-or-later"
|
||||
|
||||
@@ -93,6 +93,7 @@ ipld-core = "0.4"
|
||||
iroh-car = "0.5"
|
||||
jacquard-common = { version = "0.9", features = ["crypto-k256"] }
|
||||
jacquard-repo = "0.9"
|
||||
lettre = { version = "0.11", default-features = false, features = ["builder", "smtp-transport", "tokio1", "tokio1-rustls-tls", "pool", "dkim", "tracing"] }
|
||||
jsonwebtoken = { version = "10.2", features = ["rust_crypto"] }
|
||||
k256 = { version = "0.13", features = ["ecdsa", "pem", "pkcs8"] }
|
||||
metrics = "0.24"
|
||||
@@ -105,6 +106,8 @@ p384 = { version = "0.13", features = ["ecdsa"] }
|
||||
rand = "0.8"
|
||||
redis = { version = "1.0", features = ["tokio-comp", "connection-manager"] }
|
||||
regex = "1"
|
||||
rsa = "0.9"
|
||||
secrecy = { version = "0.10", features = ["serde"] }
|
||||
reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls-webpki-roots", "http2", "charset", "macos-system-configuration"] }
|
||||
serde = { version = "1.0", features = ["derive"] }
|
||||
serde_bytes = "0.11"
|
||||
|
||||
@@ -10,7 +10,14 @@ tranquil-signal = { workspace = true }
|
||||
|
||||
async-trait = { workspace = true }
|
||||
base64 = { workspace = true }
|
||||
ed25519-dalek = { workspace = true }
|
||||
futures = { workspace = true }
|
||||
hickory-resolver = { workspace = true }
|
||||
lettre = { workspace = true }
|
||||
rand = { workspace = true }
|
||||
reqwest = { workspace = true }
|
||||
rsa = { workspace = true }
|
||||
secrecy = { workspace = true }
|
||||
serde_json = { workspace = true }
|
||||
sqlx = { workspace = true }
|
||||
thiserror = { workspace = true }
|
||||
@@ -18,3 +25,7 @@ tokio = { workspace = true }
|
||||
tracing = { workspace = true }
|
||||
tranquil-db-traits = { workspace = true }
|
||||
uuid = { workspace = true }
|
||||
|
||||
[dev-dependencies]
|
||||
chrono = { workspace = true }
|
||||
tokio = { workspace = true, features = ["macros", "rt-multi-thread", "time", "io-util", "net"] }
|
||||
|
||||
@@ -0,0 +1,291 @@
|
||||
use std::path::PathBuf;
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum ParseError {
|
||||
#[error("empty value")]
|
||||
Empty,
|
||||
#[error("invalid character {0:?}")]
|
||||
InvalidChar(char),
|
||||
#[error("zero {0}")]
|
||||
Zero(&'static str),
|
||||
#[error("invalid TLS mode {0:?}")]
|
||||
InvalidTlsMode(String),
|
||||
}
|
||||
|
||||
fn parse_token(raw: &str, lowercase: bool, strip_trailing_dot: bool) -> Result<String, ParseError> {
|
||||
let mut s = raw.trim();
|
||||
if strip_trailing_dot {
|
||||
s = s.trim_end_matches('.');
|
||||
}
|
||||
match s {
|
||||
"" => Err(ParseError::Empty),
|
||||
_ if s.chars().any(char::is_whitespace) => Err(ParseError::InvalidChar(' ')),
|
||||
_ => Ok(match lowercase {
|
||||
true => s.to_lowercase(),
|
||||
false => s.to_string(),
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
|
||||
pub struct SmtpHost(String);
|
||||
|
||||
impl SmtpHost {
|
||||
pub fn parse(raw: &str) -> Result<Self, ParseError> {
|
||||
parse_token(raw, true, false).map(Self)
|
||||
}
|
||||
|
||||
pub fn as_str(&self) -> &str {
|
||||
&self.0
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
|
||||
pub struct SmtpPort(u16);
|
||||
|
||||
impl SmtpPort {
|
||||
pub fn parse(raw: u16) -> Result<Self, ParseError> {
|
||||
match raw {
|
||||
0 => Err(ParseError::Zero("smtp port")),
|
||||
n => Ok(Self(n)),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn as_u16(self) -> u16 {
|
||||
self.0
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
|
||||
pub struct HeloName(String);
|
||||
|
||||
impl HeloName {
|
||||
pub fn parse(raw: &str) -> Result<Self, ParseError> {
|
||||
parse_token(raw, false, false).map(Self)
|
||||
}
|
||||
|
||||
pub fn as_str(&self) -> &str {
|
||||
&self.0
|
||||
}
|
||||
|
||||
pub fn into_inner(self) -> String {
|
||||
self.0
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
|
||||
pub struct EmailDomain(String);
|
||||
|
||||
impl EmailDomain {
|
||||
pub fn parse(raw: &str) -> Result<Self, ParseError> {
|
||||
parse_token(raw, true, true).map(Self)
|
||||
}
|
||||
|
||||
pub fn as_str(&self) -> &str {
|
||||
&self.0
|
||||
}
|
||||
|
||||
pub fn into_inner(self) -> String {
|
||||
self.0
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
|
||||
pub struct MxHost(String);
|
||||
|
||||
impl MxHost {
|
||||
pub fn parse(raw: &str) -> Result<Self, ParseError> {
|
||||
parse_token(raw, true, true).map(Self)
|
||||
}
|
||||
|
||||
pub fn as_str(&self) -> &str {
|
||||
&self.0
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
|
||||
pub struct MxPriority(u16);
|
||||
|
||||
impl MxPriority {
|
||||
pub fn new(value: u16) -> Self {
|
||||
Self(value)
|
||||
}
|
||||
|
||||
pub fn as_u16(self) -> u16 {
|
||||
self.0
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct MxRecord {
|
||||
pub priority: MxPriority,
|
||||
pub host: MxHost,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
|
||||
pub struct DkimSelector(String);
|
||||
|
||||
impl DkimSelector {
|
||||
pub fn parse(raw: &str) -> Result<Self, ParseError> {
|
||||
let trimmed = raw.trim();
|
||||
let valid = !trimmed.is_empty() && trimmed.split('.').all(valid_subdomain);
|
||||
match valid {
|
||||
true => Ok(Self(trimmed.to_string())),
|
||||
false => Err(ParseError::InvalidChar('?')),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn into_inner(self) -> String {
|
||||
self.0
|
||||
}
|
||||
}
|
||||
|
||||
fn valid_subdomain(seg: &str) -> bool {
|
||||
let starts_alnum = seg
|
||||
.chars()
|
||||
.next()
|
||||
.is_some_and(|c| c.is_ascii_alphanumeric());
|
||||
let ends_alnum = seg
|
||||
.chars()
|
||||
.next_back()
|
||||
.is_some_and(|c| c.is_ascii_alphanumeric());
|
||||
let body_ok = seg.chars().all(|c| c.is_ascii_alphanumeric() || c == '-');
|
||||
starts_alnum && ends_alnum && body_ok
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct DkimKeyPath(PathBuf);
|
||||
|
||||
impl DkimKeyPath {
|
||||
pub fn parse(raw: &str) -> Result<Self, ParseError> {
|
||||
let trimmed = raw.trim();
|
||||
match trimmed.is_empty() {
|
||||
true => Err(ParseError::Empty),
|
||||
false => Ok(Self(PathBuf::from(trimmed))),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn as_path(&self) -> &std::path::Path {
|
||||
&self.0
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct SmtpUsername(String);
|
||||
|
||||
impl SmtpUsername {
|
||||
pub fn parse(raw: &str) -> Result<Self, ParseError> {
|
||||
match raw.is_empty() {
|
||||
true => Err(ParseError::Empty),
|
||||
false => Ok(Self(raw.to_string())),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn into_inner(self) -> String {
|
||||
self.0
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct SmtpPassword(secrecy::SecretString);
|
||||
|
||||
impl SmtpPassword {
|
||||
pub fn parse(raw: &str) -> Result<Self, ParseError> {
|
||||
match raw.is_empty() {
|
||||
true => Err(ParseError::Empty),
|
||||
false => Ok(Self(secrecy::SecretString::from(raw.to_string()))),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn expose(&self) -> &str {
|
||||
use secrecy::ExposeSecret;
|
||||
self.0.expose_secret()
|
||||
}
|
||||
}
|
||||
|
||||
impl std::fmt::Debug for SmtpPassword {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
f.write_str("SmtpPassword(***)")
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum TlsMode {
|
||||
Implicit,
|
||||
Starttls,
|
||||
None,
|
||||
}
|
||||
|
||||
impl TlsMode {
|
||||
pub fn parse(raw: &str) -> Result<Self, ParseError> {
|
||||
match raw.to_ascii_lowercase().as_str() {
|
||||
"implicit" => Ok(Self::Implicit),
|
||||
"starttls" => Ok(Self::Starttls),
|
||||
"none" => Ok(Self::None),
|
||||
other => Err(ParseError::InvalidTlsMode(other.to_string())),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn smtp_host_lowercases_and_trims() {
|
||||
let h = SmtpHost::parse(" SMTP.NEL.PET ").unwrap();
|
||||
assert_eq!(h.as_str(), "smtp.nel.pet");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn smtp_host_rejects_whitespace() {
|
||||
assert!(SmtpHost::parse("a b").is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn smtp_host_rejects_empty() {
|
||||
assert!(SmtpHost::parse("").is_err());
|
||||
assert!(SmtpHost::parse(" ").is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn smtp_port_rejects_zero() {
|
||||
assert!(SmtpPort::parse(0).is_err());
|
||||
assert_eq!(SmtpPort::parse(587).unwrap().as_u16(), 587);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn email_domain_strips_trailing_dot() {
|
||||
assert_eq!(EmailDomain::parse("Nel.pet.").unwrap().as_str(), "nel.pet");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn dkim_selector_validates() {
|
||||
assert!(DkimSelector::parse("default").is_ok());
|
||||
assert!(DkimSelector::parse("s1.nel.pet").is_ok());
|
||||
assert!(DkimSelector::parse("s2024-q1").is_ok());
|
||||
assert!(DkimSelector::parse("mailo-2024.nel.pet").is_ok());
|
||||
assert!(DkimSelector::parse("a-b").is_ok());
|
||||
assert!(DkimSelector::parse("").is_err());
|
||||
assert!(DkimSelector::parse("a..b").is_err());
|
||||
assert!(DkimSelector::parse("-leading").is_err());
|
||||
assert!(DkimSelector::parse("trailing-").is_err());
|
||||
assert!(DkimSelector::parse("s_under").is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn tls_mode_parses_known_modes() {
|
||||
assert_eq!(TlsMode::parse("STARTTLS").unwrap(), TlsMode::Starttls);
|
||||
assert_eq!(TlsMode::parse("implicit").unwrap(), TlsMode::Implicit);
|
||||
assert_eq!(TlsMode::parse("none").unwrap(), TlsMode::None);
|
||||
assert!(TlsMode::parse("garbage").is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn smtp_password_redacts_in_debug() {
|
||||
let p = SmtpPassword::parse("hunter2").unwrap();
|
||||
let dbg = format!("{:?}", p);
|
||||
assert_eq!(dbg, "SmtpPassword(***)");
|
||||
assert!(!dbg.contains("hunter2"));
|
||||
}
|
||||
}
|
||||
@@ -5,6 +5,14 @@ use std::sync::OnceLock;
|
||||
|
||||
static CONFIG: OnceLock<TranquilConfig> = OnceLock::new();
|
||||
|
||||
const REMOVED_ENV_VARS: &[(&str, &str)] = &[(
|
||||
"SENDMAIL_PATH",
|
||||
"the sendmail-binary transport was replaced with native SMTP. \
|
||||
Configure MAIL_SMARTHOST_HOST for relay delivery, or leave it unset to \
|
||||
deliver directly via recipient MX records. See example.toml for the full \
|
||||
MAIL_* surface.",
|
||||
)];
|
||||
|
||||
/// Errors discovered during configuration validation.
|
||||
#[derive(Debug)]
|
||||
pub struct ConfigError {
|
||||
@@ -162,6 +170,14 @@ impl TranquilConfig {
|
||||
pub fn validate(&self, ignore_secrets: bool) -> Result<(), ConfigError> {
|
||||
let mut errors = Vec::new();
|
||||
|
||||
// -- removed config ---------------------------------------------------
|
||||
errors.extend(
|
||||
REMOVED_ENV_VARS
|
||||
.iter()
|
||||
.filter(|(var, _)| std::env::var_os(var).is_some())
|
||||
.map(|(var, guidance)| format!("{var} is no longer supported: {guidance}")),
|
||||
);
|
||||
|
||||
// -- secrets ----------------------------------------------------------
|
||||
if !ignore_secrets && !self.secrets.allow_insecure && !cfg!(test) {
|
||||
if let Some(ref s) = self.secrets.jwt_secret {
|
||||
@@ -210,6 +226,85 @@ impl TranquilConfig {
|
||||
}
|
||||
}
|
||||
|
||||
// -- email smarthost --------------------------------------------------
|
||||
match self.email.smarthost.tls.to_ascii_lowercase().as_str() {
|
||||
"implicit" | "starttls" => {}
|
||||
"none" => {
|
||||
if self.email.smarthost.password.is_some() {
|
||||
errors.push(
|
||||
"email.smarthost.tls = \"none\" with email.smarthost.password set \
|
||||
would transmit credentials in plaintext; use \"starttls\" or \"implicit\""
|
||||
.to_string(),
|
||||
);
|
||||
}
|
||||
}
|
||||
other => errors.push(format!(
|
||||
"email.smarthost.tls must be \"implicit\", \"starttls\", or \"none\", got \"{other}\""
|
||||
)),
|
||||
}
|
||||
|
||||
let smarthost_host_set = self
|
||||
.email
|
||||
.smarthost
|
||||
.host
|
||||
.as_deref()
|
||||
.is_some_and(|h| !h.is_empty());
|
||||
let username_set = self.email.smarthost.username.is_some();
|
||||
let password_set = self.email.smarthost.password.is_some();
|
||||
if !smarthost_host_set && (username_set || password_set) {
|
||||
errors.push(
|
||||
"email.smarthost.username or email.smarthost.password is set but \
|
||||
email.smarthost.host is empty; credentials would be silently ignored"
|
||||
.to_string(),
|
||||
);
|
||||
}
|
||||
if smarthost_host_set && username_set != password_set {
|
||||
errors.push(
|
||||
"email.smarthost.username and email.smarthost.password must both be set or \
|
||||
both unset; otherwise authentication would silently degrade to anonymous"
|
||||
.to_string(),
|
||||
);
|
||||
}
|
||||
|
||||
if self.email.smarthost.command_timeout_secs == 0 {
|
||||
errors.push("email.smarthost.command_timeout_secs must be at least 1".to_string());
|
||||
}
|
||||
if self.email.smarthost.total_timeout_secs == 0 {
|
||||
errors.push("email.smarthost.total_timeout_secs must be at least 1".to_string());
|
||||
}
|
||||
if self.email.smarthost.pool_size == 0 {
|
||||
errors.push("email.smarthost.pool_size must be at least 1".to_string());
|
||||
}
|
||||
|
||||
if self.email.direct_mx.max_concurrent_sends == 0 {
|
||||
errors.push("email.direct_mx.max_concurrent_sends must be at least 1".to_string());
|
||||
}
|
||||
if self.email.direct_mx.command_timeout_secs == 0 {
|
||||
errors.push("email.direct_mx.command_timeout_secs must be at least 1".to_string());
|
||||
}
|
||||
if self.email.direct_mx.total_timeout_secs == 0 {
|
||||
errors.push("email.direct_mx.total_timeout_secs must be at least 1".to_string());
|
||||
}
|
||||
|
||||
let dkim_set = self.email.dkim.selector.is_some()
|
||||
|| self.email.dkim.domain.is_some()
|
||||
|| self.email.dkim.private_key_path.is_some();
|
||||
if dkim_set {
|
||||
if self.email.dkim.selector.is_none() {
|
||||
errors
|
||||
.push("email.dkim.selector is required when any DKIM field is set".to_string());
|
||||
}
|
||||
if self.email.dkim.domain.is_none() {
|
||||
errors.push("email.dkim.domain is required when any DKIM field is set".to_string());
|
||||
}
|
||||
if self.email.dkim.private_key_path.is_none() {
|
||||
errors.push(
|
||||
"email.dkim.private_key_path is required when any DKIM field is set"
|
||||
.to_string(),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// -- telegram ---------------------------------------------------------
|
||||
if self.telegram.bot_token.is_some() && self.telegram.webhook_secret.is_none() {
|
||||
errors.push(
|
||||
@@ -754,9 +849,98 @@ pub struct EmailConfig {
|
||||
#[config(env = "MAIL_FROM_NAME", default = "Tranquil PDS")]
|
||||
pub from_name: String,
|
||||
|
||||
/// Path to the `sendmail` binary.
|
||||
#[config(env = "SENDMAIL_PATH", default = "/usr/sbin/sendmail")]
|
||||
pub sendmail_path: String,
|
||||
/// HELO/EHLO name announced to remote SMTP servers. Applies to both
|
||||
/// smarthost and direct-MX modes. Defaults to the server hostname.
|
||||
#[config(env = "MAIL_HELO_NAME")]
|
||||
pub helo_name: Option<String>,
|
||||
|
||||
#[config(nested)]
|
||||
pub smarthost: SmarthostConfig,
|
||||
|
||||
#[config(nested)]
|
||||
pub direct_mx: DirectMxConfig,
|
||||
|
||||
#[config(nested)]
|
||||
pub dkim: DkimConfig,
|
||||
}
|
||||
|
||||
#[derive(Debug, Config)]
|
||||
pub struct SmarthostConfig {
|
||||
/// SMTP relay host. When set, mail is delivered through this host
|
||||
/// instead of resolving recipient MX records directly.
|
||||
#[config(env = "MAIL_SMARTHOST_HOST")]
|
||||
pub host: Option<String>,
|
||||
|
||||
/// SMTP relay port.
|
||||
#[config(env = "MAIL_SMARTHOST_PORT", default = 587)]
|
||||
pub port: u16,
|
||||
|
||||
/// SMTP authentication username.
|
||||
#[config(env = "MAIL_SMARTHOST_USERNAME")]
|
||||
pub username: Option<String>,
|
||||
|
||||
/// SMTP authentication password.
|
||||
#[config(env = "MAIL_SMARTHOST_PASSWORD")]
|
||||
pub password: Option<String>,
|
||||
|
||||
/// TLS mode. Valid values: "implicit", "starttls", "none". Setting "none"
|
||||
/// alongside a password is rejected at startup to prevent transmitting
|
||||
/// credentials in plaintext.
|
||||
#[config(env = "MAIL_SMARTHOST_TLS", default = "starttls")]
|
||||
pub tls: String,
|
||||
|
||||
/// Max size of the connection pool.
|
||||
#[config(env = "MAIL_SMARTHOST_POOL_SIZE", default = 4)]
|
||||
pub pool_size: u32,
|
||||
|
||||
/// Per-command SMTP timeout in seconds. Bounds the security handshake.
|
||||
#[config(env = "MAIL_SMARTHOST_COMMAND_TIMEOUT_SECS", default = 30)]
|
||||
pub command_timeout_secs: u64,
|
||||
|
||||
/// Total per-message timeout in seconds. Wraps the entire send so a
|
||||
/// stuck relay cannot stall the comms queue.
|
||||
#[config(env = "MAIL_SMARTHOST_TOTAL_TIMEOUT_SECS", default = 60)]
|
||||
pub total_timeout_secs: u64,
|
||||
}
|
||||
|
||||
#[derive(Debug, Config)]
|
||||
pub struct DirectMxConfig {
|
||||
/// Per-command SMTP timeout in seconds.
|
||||
#[config(env = "MAIL_COMMAND_TIMEOUT_SECS", default = 30)]
|
||||
pub command_timeout_secs: u64,
|
||||
|
||||
/// Total per-message timeout across all MX attempts in seconds.
|
||||
#[config(env = "MAIL_TOTAL_TIMEOUT_SECS", default = 60)]
|
||||
pub total_timeout_secs: u64,
|
||||
|
||||
/// Max number of concurrent direct-MX sends. Limits the load placed
|
||||
/// on any single recipient MX during a backlog drain.
|
||||
#[config(env = "MAIL_MAX_CONCURRENT_SENDS", default = 8)]
|
||||
pub max_concurrent_sends: usize,
|
||||
|
||||
/// Require STARTTLS on every MX hop. When false, TLS is
|
||||
/// attempted opportunistically and the session falls back to plaintext
|
||||
/// if the remote does not advertise STARTTLS. Set true to refuse
|
||||
/// plaintext delivery, at the cost of failing sends to MX hosts that
|
||||
/// do not support TLS.
|
||||
#[config(env = "MAIL_REQUIRE_TLS", default = false)]
|
||||
pub require_tls: bool,
|
||||
}
|
||||
|
||||
#[derive(Debug, Config)]
|
||||
pub struct DkimConfig {
|
||||
/// DKIM selector. When unset, outgoing mail is not signed.
|
||||
#[config(env = "MAIL_DKIM_SELECTOR")]
|
||||
pub selector: Option<String>,
|
||||
|
||||
/// DKIM signing domain.
|
||||
#[config(env = "MAIL_DKIM_DOMAIN")]
|
||||
pub domain: Option<String>,
|
||||
|
||||
/// Path to the DKIM private key in PEM format. Supports RSA and
|
||||
/// Ed25519 keys.
|
||||
#[config(env = "MAIL_DKIM_KEY_PATH")]
|
||||
pub private_key_path: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Config)]
|
||||
@@ -1196,3 +1380,64 @@ pub struct TranquilStoreConfig {
|
||||
pub fn template() -> String {
|
||||
confique::toml::template::<TranquilConfig>(confique::toml::FormatOptions::default())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn seed_required_env() {
|
||||
let required = [
|
||||
("PDS_HOSTNAME", "test.local"),
|
||||
("DATABASE_URL", "postgres://localhost/test"),
|
||||
("TRANQUIL_PDS_ALLOW_INSECURE_SECRETS", "1"),
|
||||
("INVITE_CODE_REQUIRED", "false"),
|
||||
("ENABLE_PDS_HOSTED_DID_WEB", "true"),
|
||||
("TRANQUIL_LEXICON_OFFLINE", "1"),
|
||||
];
|
||||
required
|
||||
.iter()
|
||||
.filter(|(k, _)| std::env::var_os(k).is_none())
|
||||
.for_each(|(k, v)| unsafe { std::env::set_var(k, v) });
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn serial_validate_rejects_legacy_sendmail_path() {
|
||||
seed_required_env();
|
||||
unsafe { std::env::set_var("SENDMAIL_PATH", "/usr/sbin/sendmail") };
|
||||
let config = TranquilConfig::builder()
|
||||
.env()
|
||||
.load()
|
||||
.expect("load fresh config");
|
||||
let result = config.validate(true);
|
||||
unsafe { std::env::remove_var("SENDMAIL_PATH") };
|
||||
|
||||
let err = result.expect_err("validate must reject SENDMAIL_PATH");
|
||||
let mentions_sendmail = err.errors.iter().any(|e| e.contains("SENDMAIL_PATH"));
|
||||
assert!(
|
||||
mentions_sendmail,
|
||||
"errors did not mention SENDMAIL_PATH: {:?}",
|
||||
err.errors
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn serial_validate_passes_when_no_legacy_env_set() {
|
||||
seed_required_env();
|
||||
unsafe { std::env::remove_var("SENDMAIL_PATH") };
|
||||
let config = TranquilConfig::builder()
|
||||
.env()
|
||||
.load()
|
||||
.expect("load fresh config");
|
||||
let result = config.validate(true);
|
||||
let leaked_legacy = result
|
||||
.as_ref()
|
||||
.err()
|
||||
.map(|e| e.errors.iter().any(|s| s.contains("SENDMAIL_PATH")))
|
||||
.unwrap_or(false);
|
||||
assert!(
|
||||
!leaked_legacy,
|
||||
"validate spuriously flagged SENDMAIL_PATH when unset: {:?}",
|
||||
result
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -523,13 +523,16 @@ fn wire_tranquil_store(
|
||||
let metastore =
|
||||
Metastore::open(&metastore_dir, metastore_config).expect("failed to open metastore");
|
||||
|
||||
let blockstore = TranquilBlockStore::open(BlockStoreConfig {
|
||||
data_dir: blockstore_data_dir,
|
||||
index_dir: blockstore_index_dir,
|
||||
max_file_size: store_cfg.max_blockstore_file_size,
|
||||
group_commit: Default::default(),
|
||||
shard_count: tranquil_store::blockstore::DEFAULT_SHARD_COUNT,
|
||||
})
|
||||
let blockstore = TranquilBlockStore::open_with_retry(
|
||||
BlockStoreConfig {
|
||||
data_dir: blockstore_data_dir,
|
||||
index_dir: blockstore_index_dir,
|
||||
max_file_size: store_cfg.max_blockstore_file_size,
|
||||
group_commit: Default::default(),
|
||||
shard_count: tranquil_store::blockstore::DEFAULT_SHARD_COUNT,
|
||||
},
|
||||
tranquil_store::blockstore::OpenRetryPolicy::default(),
|
||||
)
|
||||
.expect("failed to open blockstore");
|
||||
|
||||
let event_log = EventLog::open(
|
||||
|
||||
@@ -64,6 +64,16 @@ enum Cmd {
|
||||
#[arg(long)]
|
||||
config: Option<PathBuf>,
|
||||
|
||||
/// Tempdir parent for `IoBackend::Real` seeds only - ignored for
|
||||
/// flaky-mount and simulated backends. Repeatable; each rayon worker
|
||||
/// thread is pinned to one root so concurrent seeds on different
|
||||
/// threads land on different mounts. Default `/tmp`. Also reads
|
||||
/// colon-separated paths from `GAUNTLET_SCRATCH_ROOTS`. Set
|
||||
/// `RAYON_NUM_THREADS=N` to cap workers; for full distribution pass
|
||||
/// one root per worker.
|
||||
#[arg(long)]
|
||||
scratch_root: Vec<PathBuf>,
|
||||
|
||||
/// Skip shrinking when dumping regressions.
|
||||
#[arg(long)]
|
||||
no_shrink: bool,
|
||||
@@ -95,6 +105,13 @@ enum Cmd {
|
||||
#[arg(long)]
|
||||
dump_regressions: Option<PathBuf>,
|
||||
|
||||
/// Same as `farm --scratch-root`: tempdir parent for
|
||||
/// `IoBackend::Real` seeds only, pinned per worker thread. Ignored
|
||||
/// for flaky-mount and simulated backends. Repeatable; reads
|
||||
/// colon-separated paths from `GAUNTLET_SCRATCH_ROOTS`.
|
||||
#[arg(long)]
|
||||
scratch_root: Vec<PathBuf>,
|
||||
|
||||
/// Skip shrinking when dumping regressions.
|
||||
#[arg(long)]
|
||||
no_shrink: bool,
|
||||
@@ -159,6 +176,8 @@ struct ConfigFile {
|
||||
#[serde(default)]
|
||||
dump_regressions: Option<PathBuf>,
|
||||
#[serde(default)]
|
||||
scratch_roots: Vec<PathBuf>,
|
||||
#[serde(default)]
|
||||
overrides: ConfigOverrides,
|
||||
}
|
||||
|
||||
@@ -178,6 +197,8 @@ struct SweepConfigFile {
|
||||
#[serde(default)]
|
||||
dump_regressions: Option<PathBuf>,
|
||||
#[serde(default)]
|
||||
scratch_roots: Vec<PathBuf>,
|
||||
#[serde(default)]
|
||||
base_overrides: ConfigOverrides,
|
||||
#[serde(default)]
|
||||
axes: SweepAxes,
|
||||
@@ -405,6 +426,7 @@ struct FarmPlan {
|
||||
seeds: u64,
|
||||
hours: Option<f64>,
|
||||
dump_regressions: Option<PathBuf>,
|
||||
scratch_roots: Vec<PathBuf>,
|
||||
overrides: ConfigOverrides,
|
||||
shrink: bool,
|
||||
shrink_budget: usize,
|
||||
@@ -418,6 +440,7 @@ fn resolve_farm(
|
||||
hours: Option<f64>,
|
||||
dump_regressions: Option<PathBuf>,
|
||||
config: Option<PathBuf>,
|
||||
scratch_root: Vec<PathBuf>,
|
||||
shrink: bool,
|
||||
shrink_budget: usize,
|
||||
) -> Result<FarmPlan, String> {
|
||||
@@ -443,6 +466,11 @@ fn resolve_farm(
|
||||
}
|
||||
let dump_regressions =
|
||||
dump_regressions.or_else(|| file.as_ref().and_then(|f| f.dump_regressions.clone()));
|
||||
let file_scratch_roots = file
|
||||
.as_ref()
|
||||
.map(|f| f.scratch_roots.clone())
|
||||
.unwrap_or_default();
|
||||
let scratch_roots = resolve_scratch_roots(scratch_root, file_scratch_roots)?;
|
||||
let overrides = file.map(|f| f.overrides).unwrap_or_default();
|
||||
Ok(FarmPlan {
|
||||
scenario,
|
||||
@@ -450,12 +478,50 @@ fn resolve_farm(
|
||||
seeds,
|
||||
hours,
|
||||
dump_regressions,
|
||||
scratch_roots,
|
||||
overrides,
|
||||
shrink,
|
||||
shrink_budget,
|
||||
})
|
||||
}
|
||||
|
||||
const SCRATCH_ROOTS_ENV: &str = "GAUNTLET_SCRATCH_ROOTS";
|
||||
|
||||
fn resolve_scratch_roots(
|
||||
cli: Vec<PathBuf>,
|
||||
config_file: Vec<PathBuf>,
|
||||
) -> Result<Vec<PathBuf>, String> {
|
||||
let env_roots: Vec<PathBuf> = std::env::var(SCRATCH_ROOTS_ENV)
|
||||
.ok()
|
||||
.filter(|s| !s.is_empty())
|
||||
.map(|s| s.split(':').map(PathBuf::from).collect())
|
||||
.unwrap_or_default();
|
||||
let candidate: Vec<PathBuf> = if !cli.is_empty() {
|
||||
cli
|
||||
} else if !config_file.is_empty() {
|
||||
config_file
|
||||
} else {
|
||||
env_roots
|
||||
};
|
||||
candidate
|
||||
.into_iter()
|
||||
.map(|p| validate_scratch_root(&p).map(|_| p))
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn validate_scratch_root(path: &Path) -> Result<(), String> {
|
||||
match path.metadata() {
|
||||
Ok(m) if m.is_dir() => {}
|
||||
Ok(_) => return Err(format!("scratch root not a directory: {}", path.display())),
|
||||
Err(e) => return Err(format!("scratch root {}: {e}", path.display())),
|
||||
}
|
||||
tempfile::Builder::new()
|
||||
.prefix(".tranquil-gauntlet-probe-")
|
||||
.tempfile_in(path)
|
||||
.map(|_| ())
|
||||
.map_err(|e| format!("scratch root {} not writable: {e}", path.display()))
|
||||
}
|
||||
|
||||
fn validate_hours(h: f64) -> Result<(), String> {
|
||||
if !h.is_finite() || h <= 0.0 {
|
||||
return Err(format!("invalid --hours={h}: must be positive and finite"));
|
||||
@@ -564,6 +630,7 @@ fn main() -> ExitCode {
|
||||
hours,
|
||||
dump_regressions,
|
||||
config,
|
||||
scratch_root,
|
||||
no_shrink,
|
||||
shrink_budget,
|
||||
} => {
|
||||
@@ -574,6 +641,7 @@ fn main() -> ExitCode {
|
||||
hours,
|
||||
dump_regressions,
|
||||
config,
|
||||
scratch_root,
|
||||
!no_shrink,
|
||||
shrink_budget,
|
||||
) {
|
||||
@@ -625,6 +693,7 @@ fn main() -> ExitCode {
|
||||
seed_start,
|
||||
seeds,
|
||||
dump_regressions,
|
||||
scratch_root,
|
||||
no_shrink,
|
||||
shrink_budget,
|
||||
max_runs,
|
||||
@@ -634,6 +703,7 @@ fn main() -> ExitCode {
|
||||
seed_start,
|
||||
seeds,
|
||||
dump_regressions,
|
||||
scratch_root,
|
||||
!no_shrink,
|
||||
shrink_budget,
|
||||
max_runs,
|
||||
@@ -659,17 +729,20 @@ struct SweepPlan {
|
||||
seed_start: u64,
|
||||
seeds: u64,
|
||||
dump_regressions: Option<PathBuf>,
|
||||
scratch_roots: Vec<PathBuf>,
|
||||
shrink: bool,
|
||||
shrink_budget: usize,
|
||||
base_overrides: ConfigOverrides,
|
||||
axes: Vec<SweepAxisValues>,
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
fn resolve_sweep(
|
||||
config: PathBuf,
|
||||
seed_start: Option<u64>,
|
||||
seeds: Option<u64>,
|
||||
dump_regressions: Option<PathBuf>,
|
||||
scratch_root: Vec<PathBuf>,
|
||||
shrink: bool,
|
||||
shrink_budget: usize,
|
||||
max_runs: u64,
|
||||
@@ -687,6 +760,7 @@ fn resolve_sweep(
|
||||
return Err("--shrink-budget must be greater than zero".to_string());
|
||||
}
|
||||
let dump_regressions = dump_regressions.or(file.dump_regressions.clone());
|
||||
let scratch_roots = resolve_scratch_roots(scratch_root, file.scratch_roots.clone())?;
|
||||
let axes = file.axes.axis_values();
|
||||
if axes.is_empty() {
|
||||
return Err("sweep produced no combinations".to_string());
|
||||
@@ -705,6 +779,7 @@ fn resolve_sweep(
|
||||
seed_start,
|
||||
seeds,
|
||||
dump_regressions,
|
||||
scratch_roots,
|
||||
shrink,
|
||||
shrink_budget,
|
||||
base_overrides: file.base_overrides,
|
||||
@@ -749,6 +824,7 @@ fn run_sweep(plan: SweepPlan, rt: &Runtime, interrupt: Arc<AtomicBool>) -> ExitC
|
||||
seed_start,
|
||||
seeds,
|
||||
dump_regressions,
|
||||
scratch_roots,
|
||||
shrink,
|
||||
shrink_budget,
|
||||
base_overrides,
|
||||
@@ -782,12 +858,13 @@ fn run_sweep(plan: SweepPlan, rt: &Runtime, interrupt: Arc<AtomicBool>) -> ExitC
|
||||
axis_values.apply_to(&mut overrides);
|
||||
let combo_start = Instant::now();
|
||||
let overrides_for_farm = overrides.clone();
|
||||
let reports = farm::run_many_timed(
|
||||
let reports = farm::run_many_timed_with_scratch_roots(
|
||||
move |s| {
|
||||
let mut cfg = config_for(scenario, s);
|
||||
overrides_for_farm.apply_to(&mut cfg);
|
||||
cfg
|
||||
},
|
||||
&scratch_roots,
|
||||
(seed_start..end).map(Seed),
|
||||
);
|
||||
let combo_wall = combo_start.elapsed();
|
||||
@@ -840,6 +917,7 @@ fn run_farm(plan: FarmPlan, rt: &Runtime, interrupt: Arc<AtomicBool>) -> ExitCod
|
||||
seeds,
|
||||
hours,
|
||||
dump_regressions,
|
||||
scratch_roots,
|
||||
overrides,
|
||||
shrink,
|
||||
shrink_budget,
|
||||
@@ -871,12 +949,13 @@ fn run_farm(plan: FarmPlan, rt: &Runtime, interrupt: Arc<AtomicBool>) -> ExitCod
|
||||
};
|
||||
let overrides_ref = &overrides;
|
||||
let batch_start = Instant::now();
|
||||
let reports = farm::run_many_timed(
|
||||
let reports = farm::run_many_timed_with_scratch_roots(
|
||||
|s| {
|
||||
let mut cfg = config_for(scenario, s);
|
||||
overrides_ref.apply_to(&mut cfg);
|
||||
cfg
|
||||
},
|
||||
&scratch_roots,
|
||||
(next_seed..end).map(Seed),
|
||||
);
|
||||
let batch_wall = batch_start.elapsed();
|
||||
|
||||
@@ -222,7 +222,8 @@ fn stream_compact<S: StorageIO>(
|
||||
.io()
|
||||
.sync_dir(manager.data_dir())
|
||||
.map_err(CompactionError::from)
|
||||
});
|
||||
})
|
||||
.and_then(|()| manager.io().barrier().map_err(CompactionError::from));
|
||||
|
||||
let _ = manager.io().close(hint_fd);
|
||||
|
||||
|
||||
@@ -1343,6 +1343,10 @@ fn process_batch<S: StorageIO>(
|
||||
)
|
||||
.map_err(|e| rollback_on_err(CommitError::from(e)))?;
|
||||
hint_writer.sync().map_err(|e| rollback_on_err(e.into()))?;
|
||||
manager
|
||||
.io()
|
||||
.barrier()
|
||||
.map_err(|e| rollback_on_err(e.into()))?;
|
||||
let sync_nanos = t.elapsed().as_nanos() as u64;
|
||||
|
||||
if !rotations.is_empty() {
|
||||
|
||||
@@ -27,7 +27,7 @@ pub use hint::{
|
||||
pub use manager::{CachedHandle, DEFAULT_MAX_FILE_SIZE, DataFileManager};
|
||||
pub use reader::{BlockStoreReader, ReadError};
|
||||
pub use store::QuiesceGuard;
|
||||
pub use store::{BlockStoreConfig, DEFAULT_SHARD_COUNT, TranquilBlockStore};
|
||||
pub use store::{BlockStoreConfig, DEFAULT_SHARD_COUNT, OpenRetryPolicy, TranquilBlockStore};
|
||||
pub use types::{
|
||||
BlockLength, BlockLocation, BlockOffset, BlockstoreSnapshot, CidBytes, CollectionResult,
|
||||
CommitEpoch, CompactionResult, DataFileId, EpochCounter, HintOffset, IndexEntry, LivenessInfo,
|
||||
|
||||
@@ -1,7 +1,9 @@
|
||||
use std::collections::HashMap;
|
||||
use std::io;
|
||||
use std::num::NonZeroU8;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use bytes::Bytes;
|
||||
use cid::Cid;
|
||||
@@ -150,6 +152,24 @@ impl Drop for WriterHandle {
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug)]
|
||||
pub struct OpenRetryPolicy {
|
||||
pub max_attempts: NonZeroU8,
|
||||
pub initial_backoff: Duration,
|
||||
pub max_backoff: Duration,
|
||||
}
|
||||
|
||||
impl Default for OpenRetryPolicy {
|
||||
fn default() -> Self {
|
||||
const DEFAULT_MAX_ATTEMPTS: NonZeroU8 = NonZeroU8::new(5).unwrap();
|
||||
Self {
|
||||
max_attempts: DEFAULT_MAX_ATTEMPTS,
|
||||
initial_backoff: Duration::from_millis(100),
|
||||
max_backoff: Duration::from_secs(2),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl TranquilBlockStore<RealIO> {
|
||||
pub fn open(config: BlockStoreConfig) -> Result<Self, RepoError> {
|
||||
Self::open_with_hook(config, None)
|
||||
@@ -161,6 +181,50 @@ impl TranquilBlockStore<RealIO> {
|
||||
) -> Result<Self, RepoError> {
|
||||
Self::open_with_io_hook(config, RealIO::new, post_sync_hook)
|
||||
}
|
||||
|
||||
pub fn open_with_retry(
|
||||
config: BlockStoreConfig,
|
||||
policy: OpenRetryPolicy,
|
||||
) -> Result<Self, RepoError> {
|
||||
retry_with_backoff(policy, &mut |_| Self::open(config.clone()))
|
||||
}
|
||||
}
|
||||
|
||||
fn retry_with_backoff<T, F>(policy: OpenRetryPolicy, op: &mut F) -> Result<T, RepoError>
|
||||
where
|
||||
F: FnMut(u8) -> Result<T, RepoError>,
|
||||
{
|
||||
retry_attempt(policy, op, 0, policy.initial_backoff)
|
||||
}
|
||||
|
||||
fn retry_attempt<T, F>(
|
||||
policy: OpenRetryPolicy,
|
||||
op: &mut F,
|
||||
attempt: u8,
|
||||
backoff: Duration,
|
||||
) -> Result<T, RepoError>
|
||||
where
|
||||
F: FnMut(u8) -> Result<T, RepoError>,
|
||||
{
|
||||
match op(attempt) {
|
||||
Ok(t) => Ok(t),
|
||||
Err(e) if attempt + 1 >= policy.max_attempts.get() => Err(e),
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
attempt,
|
||||
error = %e,
|
||||
backoff_ms = u64::try_from(backoff.as_millis()).unwrap_or(u64::MAX),
|
||||
"blockstore open failed, retrying"
|
||||
);
|
||||
std::thread::sleep(backoff);
|
||||
retry_attempt(
|
||||
policy,
|
||||
op,
|
||||
attempt + 1,
|
||||
(backoff * 2).min(policy.max_backoff),
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<S: StorageIO + Send + Sync + 'static> TranquilBlockStore<S> {
|
||||
@@ -331,15 +395,7 @@ impl<S: StorageIO + Send + Sync + 'static> TranquilBlockStore<S> {
|
||||
let scan_pos = &mut { start_offset };
|
||||
let (scanned_entries, last_valid_end) = std::iter::from_fn(|| {
|
||||
match super::data_file::decode_block_record(io, fd, *scan_pos, file_size) {
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
file_id = %file_id,
|
||||
offset = scan_pos.raw(),
|
||||
error = %e,
|
||||
"IO error during recovery scan, stopping"
|
||||
);
|
||||
None
|
||||
}
|
||||
Err(e) => Some(Err(e)),
|
||||
Ok(None) => None,
|
||||
Ok(Some(ReadBlockRecord::Valid {
|
||||
offset,
|
||||
@@ -354,7 +410,7 @@ impl<S: StorageIO + Send + Sync + 'static> TranquilBlockStore<S> {
|
||||
let record_size = BLOCK_RECORD_OVERHEAD as u64 + u64::from(raw_len);
|
||||
let new_end = offset.advance(record_size);
|
||||
*scan_pos = new_end;
|
||||
Some((
|
||||
Some(Ok((
|
||||
cid_bytes,
|
||||
BlockLocation {
|
||||
file_id,
|
||||
@@ -362,20 +418,30 @@ impl<S: StorageIO + Send + Sync + 'static> TranquilBlockStore<S> {
|
||||
length,
|
||||
},
|
||||
new_end,
|
||||
))
|
||||
)))
|
||||
}
|
||||
Ok(Some(ReadBlockRecord::Corrupted { .. } | ReadBlockRecord::Truncated { .. })) => {
|
||||
None
|
||||
}
|
||||
}
|
||||
})
|
||||
.fold(
|
||||
.try_fold(
|
||||
(Vec::new(), start_offset),
|
||||
|(mut entries, _), (cid, loc, new_end)| {
|
||||
|(mut entries, _), item: io::Result<_>| {
|
||||
let (cid, loc, new_end) = item?;
|
||||
entries.push((cid, loc));
|
||||
(entries, new_end)
|
||||
Ok::<_, io::Error>((entries, new_end))
|
||||
},
|
||||
);
|
||||
)
|
||||
.map_err(|e| {
|
||||
tracing::warn!(
|
||||
file_id = %file_id,
|
||||
offset = scan_pos.raw(),
|
||||
error = %e,
|
||||
"IO error during recovery scan, aborting to preserve durable tail"
|
||||
);
|
||||
RepoError::storage(e)
|
||||
})?;
|
||||
|
||||
if file_size > last_valid_end.raw() {
|
||||
tracing::info!(
|
||||
@@ -713,3 +779,213 @@ impl<S: StorageIO + Send + Sync + 'static> TranquilBlockStore<S> {
|
||||
Ok(self.index.get(&cid_bytes).map(|entry| entry.refcount.raw()))
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Mutex;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
|
||||
use crate::blockstore::data_file::{
|
||||
BLOCK_FORMAT_VERSION, BLOCK_HEADER_SIZE, BLOCK_MAGIC, encode_block_record,
|
||||
};
|
||||
use crate::blockstore::manager::DATA_FILE_EXTENSION;
|
||||
use crate::io::FileId;
|
||||
|
||||
struct EioOnReadAtRange {
|
||||
inner: RealIO,
|
||||
target_path: PathBuf,
|
||||
target_min: u64,
|
||||
target_max: u64,
|
||||
fired: AtomicBool,
|
||||
fd_paths: Mutex<HashMap<FileId, PathBuf>>,
|
||||
}
|
||||
|
||||
impl StorageIO for EioOnReadAtRange {
|
||||
fn open(&self, path: &Path, opts: OpenOptions) -> io::Result<FileId> {
|
||||
let fd = self.inner.open(path, opts)?;
|
||||
self.fd_paths
|
||||
.lock()
|
||||
.unwrap()
|
||||
.insert(fd, path.to_path_buf());
|
||||
Ok(fd)
|
||||
}
|
||||
|
||||
fn close(&self, fd: FileId) -> io::Result<()> {
|
||||
self.fd_paths.lock().unwrap().remove(&fd);
|
||||
self.inner.close(fd)
|
||||
}
|
||||
|
||||
fn read_at(&self, fd: FileId, offset: u64, buf: &mut [u8]) -> io::Result<usize> {
|
||||
let path_match = self.fd_paths.lock().unwrap().get(&fd).cloned();
|
||||
let in_target_range = path_match.as_ref() == Some(&self.target_path)
|
||||
&& offset >= self.target_min
|
||||
&& offset <= self.target_max;
|
||||
if in_target_range && !self.fired.swap(true, Ordering::SeqCst) {
|
||||
return Err(io::Error::other("simulated EIO on read"));
|
||||
}
|
||||
self.inner.read_at(fd, offset, buf)
|
||||
}
|
||||
|
||||
fn write_at(&self, fd: FileId, offset: u64, buf: &[u8]) -> io::Result<usize> {
|
||||
self.inner.write_at(fd, offset, buf)
|
||||
}
|
||||
|
||||
fn sync(&self, fd: FileId) -> io::Result<()> {
|
||||
self.inner.sync(fd)
|
||||
}
|
||||
|
||||
fn file_size(&self, fd: FileId) -> io::Result<u64> {
|
||||
self.inner.file_size(fd)
|
||||
}
|
||||
|
||||
fn truncate(&self, fd: FileId, size: u64) -> io::Result<()> {
|
||||
self.inner.truncate(fd, size)
|
||||
}
|
||||
|
||||
fn rename(&self, from: &Path, to: &Path) -> io::Result<()> {
|
||||
self.inner.rename(from, to)
|
||||
}
|
||||
|
||||
fn delete(&self, path: &Path) -> io::Result<()> {
|
||||
self.inner.delete(path)
|
||||
}
|
||||
|
||||
fn mkdir(&self, path: &Path) -> io::Result<()> {
|
||||
self.inner.mkdir(path)
|
||||
}
|
||||
|
||||
fn sync_dir(&self, path: &Path) -> io::Result<()> {
|
||||
self.inner.sync_dir(path)
|
||||
}
|
||||
|
||||
fn list_dir(&self, path: &Path) -> io::Result<Vec<PathBuf>> {
|
||||
self.inner.list_dir(path)
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scan_and_index_does_not_truncate_acked_block_on_transient_eio() {
|
||||
let tmp = tempfile::TempDir::new().unwrap();
|
||||
let data_dir = tmp.path().join("data");
|
||||
let index_dir = tmp.path().join("index");
|
||||
std::fs::create_dir_all(&data_dir).unwrap();
|
||||
std::fs::create_dir_all(&index_dir).unwrap();
|
||||
|
||||
let file_id = DataFileId::new(0);
|
||||
let file_path = data_dir.join(format!("{file_id}.{DATA_FILE_EXTENSION}"));
|
||||
|
||||
let setup = RealIO::new();
|
||||
let fd = setup.open(&file_path, OpenOptions::read_write()).unwrap();
|
||||
let mut header = [0u8; BLOCK_HEADER_SIZE];
|
||||
header[..4].copy_from_slice(&BLOCK_MAGIC);
|
||||
header[4] = BLOCK_FORMAT_VERSION;
|
||||
setup.write_all_at(fd, 0, &header).unwrap();
|
||||
|
||||
let cid_a = [0xAAu8; CID_SIZE];
|
||||
let data_a = vec![1u8; 64];
|
||||
let block_a_offset = BlockOffset::new(BLOCK_HEADER_SIZE as u64);
|
||||
let len_a =
|
||||
encode_block_record(&setup, fd, block_a_offset, &cid_a, &data_a).unwrap();
|
||||
|
||||
let block_b_offset_raw = BLOCK_HEADER_SIZE as u64 + len_a;
|
||||
let block_b_offset = BlockOffset::new(block_b_offset_raw);
|
||||
let cid_b = [0xBBu8; CID_SIZE];
|
||||
let data_b = vec![2u8; 64];
|
||||
let len_b = encode_block_record(&setup, fd, block_b_offset, &cid_b, &data_b).unwrap();
|
||||
|
||||
setup.sync(fd).unwrap();
|
||||
setup.close(fd).unwrap();
|
||||
drop(setup);
|
||||
|
||||
let total_size = block_b_offset_raw + len_b;
|
||||
assert_eq!(std::fs::metadata(&file_path).unwrap().len(), total_size);
|
||||
|
||||
let wrapper = EioOnReadAtRange {
|
||||
inner: RealIO::new(),
|
||||
target_path: file_path.clone(),
|
||||
target_min: block_b_offset_raw,
|
||||
target_max: block_b_offset_raw + (BLOCK_RECORD_OVERHEAD as u64) - 1,
|
||||
fired: AtomicBool::new(false),
|
||||
fd_paths: Mutex::new(HashMap::new()),
|
||||
};
|
||||
|
||||
let index = BlockIndex::open(&index_dir).unwrap();
|
||||
|
||||
let result = TranquilBlockStore::<EioOnReadAtRange>::replay_single_file(
|
||||
&wrapper,
|
||||
&data_dir,
|
||||
&index,
|
||||
file_id,
|
||||
BlockOffset::new(BLOCK_HEADER_SIZE as u64),
|
||||
);
|
||||
|
||||
assert!(
|
||||
result.is_err(),
|
||||
"replay must surface transient EIO instead of silently truncating"
|
||||
);
|
||||
|
||||
let post_size = std::fs::metadata(&file_path).unwrap().len();
|
||||
assert_eq!(
|
||||
post_size, total_size,
|
||||
"scan truncated durable acked block past EIO point: expected {total_size} bytes, got {post_size}"
|
||||
);
|
||||
}
|
||||
|
||||
fn instant_policy(max_attempts: u8) -> OpenRetryPolicy {
|
||||
OpenRetryPolicy {
|
||||
max_attempts: NonZeroU8::new(max_attempts).expect("max_attempts must be nonzero"),
|
||||
initial_backoff: Duration::ZERO,
|
||||
max_backoff: Duration::ZERO,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn retry_with_backoff_succeeds_on_first_attempt() {
|
||||
let calls = std::sync::atomic::AtomicUsize::new(0);
|
||||
let result = retry_with_backoff(instant_policy(5), &mut |_| {
|
||||
calls.fetch_add(1, Ordering::Relaxed);
|
||||
Ok::<u8, RepoError>(42)
|
||||
});
|
||||
assert_eq!(result.expect("ok"), 42);
|
||||
assert_eq!(calls.load(Ordering::Relaxed), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn retry_with_backoff_recovers_after_transient_failures() {
|
||||
let calls = std::sync::atomic::AtomicUsize::new(0);
|
||||
let result = retry_with_backoff(instant_policy(5), &mut |_| {
|
||||
let n = calls.fetch_add(1, Ordering::Relaxed);
|
||||
if n >= 2 {
|
||||
Ok::<u8, RepoError>(7)
|
||||
} else {
|
||||
Err(RepoError::storage(io::Error::other("transient EIO")))
|
||||
}
|
||||
});
|
||||
assert_eq!(result.expect("ok"), 7);
|
||||
assert_eq!(calls.load(Ordering::Relaxed), 3);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn retry_with_backoff_gives_up_after_max_attempts() {
|
||||
let calls = std::sync::atomic::AtomicUsize::new(0);
|
||||
let result: Result<u8, RepoError> = retry_with_backoff(instant_policy(3), &mut |_| {
|
||||
calls.fetch_add(1, Ordering::Relaxed);
|
||||
Err(RepoError::storage(io::Error::other("permanent EIO")))
|
||||
});
|
||||
assert!(result.is_err(), "expected exhaustion error");
|
||||
assert_eq!(calls.load(Ordering::Relaxed), 3);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn retry_with_backoff_passes_attempt_index_to_op() {
|
||||
let observed = std::sync::Mutex::new(Vec::<u8>::new());
|
||||
let _result: Result<(), RepoError> = retry_with_backoff(instant_policy(4), &mut |attempt| {
|
||||
observed.lock().unwrap().push(attempt);
|
||||
Err(RepoError::storage(io::Error::other("EIO")))
|
||||
});
|
||||
assert_eq!(*observed.lock().unwrap(), vec![0, 1, 2, 3]);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,16 +3,27 @@ use std::sync::Arc;
|
||||
|
||||
use tracing::warn;
|
||||
|
||||
use crate::io::StorageIO;
|
||||
use crate::io::{FileId, StorageIO};
|
||||
|
||||
use super::manager::SegmentManager;
|
||||
use super::segment_file::{SEGMENT_HEADER_SIZE, SegmentWriter, ValidEvent};
|
||||
use super::segment_file::{
|
||||
SEGMENT_HEADER_SIZE, SEGMENT_MAGIC, SegmentWriter, ValidEvent, ValidateEventRecord,
|
||||
validate_event_record,
|
||||
};
|
||||
use super::segment_index::{DEFAULT_INDEX_INTERVAL, SegmentIndex, rebuild_from_segment};
|
||||
use super::sidecar::build_sidecar_from_segment;
|
||||
use super::types::{
|
||||
DidHash, EventSequence, EventTypeTag, SegmentId, SegmentOffset, TimestampMicros,
|
||||
};
|
||||
|
||||
const VALIDATE_RETRY_ATTEMPTS: u32 = 32;
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
struct PendingAppend {
|
||||
event: ValidEvent,
|
||||
offset: SegmentOffset,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct SyncResult {
|
||||
pub synced_through: EventSequence,
|
||||
@@ -31,7 +42,8 @@ pub struct EventLogWriter<S: StorageIO> {
|
||||
max_payload: u32,
|
||||
event_count_in_segment: usize,
|
||||
last_event_offset: Option<SegmentOffset>,
|
||||
pending_events: Vec<ValidEvent>,
|
||||
pending: Vec<PendingAppend>,
|
||||
poisoned: bool,
|
||||
}
|
||||
|
||||
impl<S: StorageIO> EventLogWriter<S> {
|
||||
@@ -83,10 +95,25 @@ impl<S: StorageIO> EventLogWriter<S> {
|
||||
max_payload,
|
||||
event_count_in_segment: 0,
|
||||
last_event_offset: None,
|
||||
pending_events: Vec::new(),
|
||||
pending: Vec::new(),
|
||||
poisoned: false,
|
||||
})
|
||||
}
|
||||
|
||||
fn truncate_and_init_fresh(
|
||||
manager: Arc<SegmentManager<S>>,
|
||||
fd: FileId,
|
||||
active_id: SegmentId,
|
||||
prev_segments: &[SegmentId],
|
||||
index_interval: usize,
|
||||
max_payload: u32,
|
||||
) -> io::Result<Self> {
|
||||
manager.io().truncate(fd, 0)?;
|
||||
let next_seq = find_last_seq_from_segments(&manager, prev_segments, max_payload)?
|
||||
.map_or(EventSequence::new(1), |s| s.next());
|
||||
Self::init_fresh(manager, active_id, next_seq, index_interval, max_payload)
|
||||
}
|
||||
|
||||
fn recover_active(
|
||||
manager: Arc<SegmentManager<S>>,
|
||||
segments: &[SegmentId],
|
||||
@@ -97,6 +124,19 @@ impl<S: StorageIO> EventLogWriter<S> {
|
||||
let handle = manager.open_for_append(active_id)?;
|
||||
let fd = handle.fd();
|
||||
|
||||
let prev_segments = &segments[..segments.len().saturating_sub(1)];
|
||||
|
||||
if highest_segment_has_torn_header(manager.io(), fd)? {
|
||||
return Self::truncate_and_init_fresh(
|
||||
Arc::clone(&manager),
|
||||
fd,
|
||||
active_id,
|
||||
prev_segments,
|
||||
index_interval,
|
||||
max_payload,
|
||||
);
|
||||
}
|
||||
|
||||
let (index, last_seq_in_active) = match rebuild_from_segment(
|
||||
manager.io(),
|
||||
fd,
|
||||
@@ -107,15 +147,11 @@ impl<S: StorageIO> EventLogWriter<S> {
|
||||
Err(rebuild_err) => {
|
||||
let file_size = manager.io().file_size(fd)?;
|
||||
if file_size <= SEGMENT_HEADER_SIZE as u64 {
|
||||
manager.io().truncate(fd, 0)?;
|
||||
let prev_segments = &segments[..segments.len().saturating_sub(1)];
|
||||
let next_seq =
|
||||
find_last_seq_from_segments(&manager, prev_segments, max_payload)?
|
||||
.map_or(EventSequence::new(1), |s| s.next());
|
||||
return Self::init_fresh(
|
||||
return Self::truncate_and_init_fresh(
|
||||
Arc::clone(&manager),
|
||||
fd,
|
||||
active_id,
|
||||
next_seq,
|
||||
prev_segments,
|
||||
index_interval,
|
||||
max_payload,
|
||||
);
|
||||
@@ -131,8 +167,6 @@ impl<S: StorageIO> EventLogWriter<S> {
|
||||
|
||||
let position = SegmentOffset::new(manager.io().file_size(fd)?);
|
||||
|
||||
let prev_segments = &segments[..segments.len().saturating_sub(1)];
|
||||
|
||||
let next_seq = match last_seq_in_active {
|
||||
Some(seq) => {
|
||||
if let Some(sealed_last) =
|
||||
@@ -196,7 +230,8 @@ impl<S: StorageIO> EventLogWriter<S> {
|
||||
max_payload,
|
||||
event_count_in_segment,
|
||||
last_event_offset,
|
||||
pending_events: Vec::new(),
|
||||
pending: Vec::new(),
|
||||
poisoned: false,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -227,28 +262,20 @@ impl<S: StorageIO> EventLogWriter<S> {
|
||||
payload,
|
||||
};
|
||||
|
||||
let offset = self.active_writer.append_event(self.manager.io(), &event)?;
|
||||
|
||||
let should_index = self.event_count_in_segment == 0
|
||||
|| self
|
||||
.event_count_in_segment
|
||||
.is_multiple_of(self.index_interval);
|
||||
if should_index {
|
||||
self.active_index.record(seq, offset);
|
||||
}
|
||||
|
||||
self.event_count_in_segment = self
|
||||
.event_count_in_segment
|
||||
.checked_add(1)
|
||||
.expect("event_count_in_segment overflow");
|
||||
self.last_event_offset = Some(offset);
|
||||
self.next_seq = seq.next();
|
||||
self.pending_events.push(event);
|
||||
|
||||
Ok(seq)
|
||||
self.append_inner(event).map(|_| seq)
|
||||
}
|
||||
|
||||
pub fn append_valid_event(&mut self, event: ValidEvent) -> io::Result<()> {
|
||||
self.append_inner(event)
|
||||
}
|
||||
|
||||
fn append_inner(&mut self, event: ValidEvent) -> io::Result<()> {
|
||||
if self.poisoned {
|
||||
return Err(io::Error::other(
|
||||
"writer poisoned by partial-valid sync; reopen required",
|
||||
));
|
||||
}
|
||||
|
||||
let offset = self.active_writer.append_event(self.manager.io(), &event)?;
|
||||
|
||||
let should_index = self.event_count_in_segment == 0
|
||||
@@ -265,21 +292,52 @@ impl<S: StorageIO> EventLogWriter<S> {
|
||||
.expect("event_count_in_segment overflow");
|
||||
self.last_event_offset = Some(offset);
|
||||
self.next_seq = event.seq.next();
|
||||
self.pending_events.push(event);
|
||||
self.pending.push(PendingAppend { event, offset });
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn peek_pending_event(&self, seq: EventSequence) -> Option<&ValidEvent> {
|
||||
self.pending_events.iter().find(|e| e.seq == seq)
|
||||
}
|
||||
|
||||
pub fn sync(&mut self) -> io::Result<SyncResult> {
|
||||
if !self.pending_events.is_empty() {
|
||||
self.active_writer.sync(self.manager.io())?;
|
||||
if self.poisoned {
|
||||
return Err(io::Error::other(
|
||||
"writer poisoned by partial-valid sync; reopen required",
|
||||
));
|
||||
}
|
||||
|
||||
let flushed = std::mem::take(&mut self.pending_events);
|
||||
if !self.pending.is_empty() {
|
||||
self.active_writer.sync(self.manager.io())?;
|
||||
self.manager.io().barrier()?;
|
||||
}
|
||||
|
||||
let pending = std::mem::take(&mut self.pending);
|
||||
|
||||
let fd = self.active_writer.fd();
|
||||
let file_size = self.manager.io().file_size(fd)?;
|
||||
|
||||
let valid_count = pending
|
||||
.iter()
|
||||
.take_while(|p| {
|
||||
validate_with_retry(
|
||||
self.manager.io(),
|
||||
fd,
|
||||
p.offset,
|
||||
file_size,
|
||||
self.max_payload,
|
||||
p.event.seq,
|
||||
)
|
||||
})
|
||||
.count();
|
||||
|
||||
if valid_count < pending.len() {
|
||||
self.poisoned = true;
|
||||
}
|
||||
|
||||
let flushed: Vec<ValidEvent> = pending
|
||||
.into_iter()
|
||||
.take(valid_count)
|
||||
.map(|p| p.event)
|
||||
.collect();
|
||||
|
||||
self.synced_seq = flushed.last().map(|e| e.seq).unwrap_or(self.synced_seq);
|
||||
|
||||
Ok(SyncResult {
|
||||
@@ -290,12 +348,22 @@ impl<S: StorageIO> EventLogWriter<S> {
|
||||
})
|
||||
}
|
||||
|
||||
pub fn is_poisoned(&self) -> bool {
|
||||
self.poisoned
|
||||
}
|
||||
|
||||
pub fn rotate_if_needed(&mut self) -> io::Result<Option<SegmentId>> {
|
||||
if self.poisoned {
|
||||
return Err(io::Error::other(
|
||||
"writer poisoned by partial-valid sync; reopen required",
|
||||
));
|
||||
}
|
||||
|
||||
if !self.manager.should_rotate(self.active_writer.position()) {
|
||||
return Ok(None);
|
||||
}
|
||||
|
||||
if !self.pending_events.is_empty() {
|
||||
if !self.pending.is_empty() {
|
||||
return Ok(None);
|
||||
}
|
||||
|
||||
@@ -386,6 +454,40 @@ impl<S: StorageIO> EventLogWriter<S> {
|
||||
}
|
||||
}
|
||||
|
||||
fn validate_with_retry<S: StorageIO>(
|
||||
io: &S,
|
||||
fd: FileId,
|
||||
offset: SegmentOffset,
|
||||
file_size: u64,
|
||||
max_payload: u32,
|
||||
expected_seq: EventSequence,
|
||||
) -> bool {
|
||||
(0..VALIDATE_RETRY_ATTEMPTS).any(|_| {
|
||||
matches!(
|
||||
validate_event_record(io, fd, offset, file_size, max_payload),
|
||||
Ok(Some(ValidateEventRecord::Valid { seq, .. })) if seq == expected_seq
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
fn highest_segment_has_torn_header<S: StorageIO>(io: &S, fd: FileId) -> io::Result<bool> {
|
||||
let file_size = io.file_size(fd)?;
|
||||
if file_size < SEGMENT_HEADER_SIZE as u64 {
|
||||
return Ok(true);
|
||||
}
|
||||
let outcomes: Vec<bool> = (0..VALIDATE_RETRY_ATTEMPTS)
|
||||
.filter_map(|_| {
|
||||
let mut header = [0u8; SEGMENT_MAGIC.len()];
|
||||
io.read_exact_at(fd, 0, &mut header)
|
||||
.ok()
|
||||
.map(|()| header == SEGMENT_MAGIC)
|
||||
})
|
||||
.collect();
|
||||
let saw_match = outcomes.iter().any(|&ok| ok);
|
||||
let saw_mismatch = outcomes.iter().any(|&ok| !ok);
|
||||
Ok(!saw_match && saw_mismatch)
|
||||
}
|
||||
|
||||
fn find_last_seq_from_segments<S: StorageIO>(
|
||||
manager: &SegmentManager<S>,
|
||||
segments: &[SegmentId],
|
||||
@@ -1094,4 +1196,62 @@ mod tests {
|
||||
|
||||
assert!(writer.rotate_if_needed().unwrap().is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sync_must_not_certify_durability_when_io_sync_silently_drops() {
|
||||
use crate::sim::{FaultConfig, Probability};
|
||||
|
||||
let sim = Arc::new(SimulatedIO::new(
|
||||
0,
|
||||
FaultConfig {
|
||||
sync_failure_probability: Probability::new(1.0),
|
||||
..FaultConfig::none()
|
||||
},
|
||||
));
|
||||
sim.set_pristine_mode(true);
|
||||
|
||||
let mgr = Arc::new(
|
||||
SegmentManager::new(Arc::clone(&sim), PathBuf::from("/segments"), 64 * 1024).unwrap(),
|
||||
);
|
||||
|
||||
let mut writer =
|
||||
EventLogWriter::open(Arc::clone(&mgr), DEFAULT_INDEX_INTERVAL, MAX_EVENT_PAYLOAD)
|
||||
.unwrap();
|
||||
|
||||
sim.set_pristine_mode(false);
|
||||
|
||||
writer
|
||||
.append(
|
||||
DidHash::from_did("did:plc:bug2"),
|
||||
EventTypeTag::COMMIT,
|
||||
b"bug2-payload".to_vec(),
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
assert!(
|
||||
writer.sync().is_err(),
|
||||
"sync must surface dropped fsync as an error"
|
||||
);
|
||||
let claimed_synced = writer.synced_seq();
|
||||
assert_eq!(
|
||||
claimed_synced.raw(),
|
||||
0,
|
||||
"synced_seq must not advance past a failed sync"
|
||||
);
|
||||
drop(writer);
|
||||
|
||||
mgr.shutdown();
|
||||
sim.crash();
|
||||
sim.set_pristine_mode(true);
|
||||
|
||||
let reopened =
|
||||
EventLogWriter::open(Arc::clone(&mgr), DEFAULT_INDEX_INTERVAL, MAX_EVENT_PAYLOAD)
|
||||
.unwrap();
|
||||
let actually_durable = reopened.current_seq();
|
||||
|
||||
assert!(
|
||||
actually_durable >= claimed_synced,
|
||||
"writer claimed sync through {claimed_synced} but post-crash recovery only reaches {actually_durable}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
use std::cell::RefCell;
|
||||
use std::panic::{AssertUnwindSafe, catch_unwind};
|
||||
use std::path::PathBuf;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use rayon::prelude::*;
|
||||
@@ -44,6 +45,17 @@ pub fn run_many_timed<F>(
|
||||
make_config: F,
|
||||
seeds: impl IntoIterator<Item = Seed>,
|
||||
) -> Vec<(GauntletReport, Duration)>
|
||||
where
|
||||
F: Fn(Seed) -> GauntletConfig + Sync + Send,
|
||||
{
|
||||
run_many_timed_with_scratch_roots(make_config, &[], seeds)
|
||||
}
|
||||
|
||||
pub fn run_many_timed_with_scratch_roots<F>(
|
||||
make_config: F,
|
||||
scratch_roots: &[PathBuf],
|
||||
seeds: impl IntoIterator<Item = Seed>,
|
||||
) -> Vec<(GauntletReport, Duration)>
|
||||
where
|
||||
F: Fn(Seed) -> GauntletConfig + Sync + Send,
|
||||
{
|
||||
@@ -51,10 +63,14 @@ where
|
||||
seeds
|
||||
.into_par_iter()
|
||||
.map(|s| {
|
||||
let scratch = scratch_for_thread(scratch_roots, rayon::current_thread_index());
|
||||
let start = Instant::now();
|
||||
let outcome = catch_unwind(AssertUnwindSafe(|| {
|
||||
let cfg = make_config(s);
|
||||
let gauntlet = Gauntlet::new(cfg).expect("build gauntlet");
|
||||
let mut gauntlet = Gauntlet::new(cfg).expect("build gauntlet");
|
||||
if let Some(root) = scratch {
|
||||
gauntlet = gauntlet.with_scratch_root(root);
|
||||
}
|
||||
with_runtime(|rt| rt.block_on(gauntlet.run()))
|
||||
}));
|
||||
let report = outcome.unwrap_or_else(|payload| {
|
||||
@@ -66,6 +82,14 @@ where
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn scratch_for_thread(roots: &[PathBuf], thread_idx: Option<usize>) -> Option<PathBuf> {
|
||||
if roots.is_empty() {
|
||||
None
|
||||
} else {
|
||||
Some(roots[thread_idx.unwrap_or(0) % roots.len()].clone())
|
||||
}
|
||||
}
|
||||
|
||||
fn panic_report(seed: Seed, payload: Box<dyn std::any::Any + Send>) -> GauntletReport {
|
||||
let msg = payload
|
||||
.downcast_ref::<&'static str>()
|
||||
@@ -84,3 +108,62 @@ fn panic_report(seed: Seed, payload: Box<dyn std::any::Any + Send>) -> GauntletR
|
||||
ops: OpStream::empty(),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn scratch_for_thread_returns_none_when_roots_empty() {
|
||||
assert!(scratch_for_thread(&[], Some(0)).is_none());
|
||||
assert!(scratch_for_thread(&[], Some(7)).is_none());
|
||||
assert!(scratch_for_thread(&[], None).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scratch_for_thread_round_robins_across_roots() {
|
||||
let roots = vec![
|
||||
PathBuf::from("/scratch/a"),
|
||||
PathBuf::from("/scratch/b"),
|
||||
PathBuf::from("/scratch/c"),
|
||||
];
|
||||
let assigned: Vec<PathBuf> = (0..7)
|
||||
.map(|i| scratch_for_thread(&roots, Some(i)).expect("scratch path"))
|
||||
.collect();
|
||||
assert_eq!(
|
||||
assigned,
|
||||
vec![
|
||||
PathBuf::from("/scratch/a"),
|
||||
PathBuf::from("/scratch/b"),
|
||||
PathBuf::from("/scratch/c"),
|
||||
PathBuf::from("/scratch/a"),
|
||||
PathBuf::from("/scratch/b"),
|
||||
PathBuf::from("/scratch/c"),
|
||||
PathBuf::from("/scratch/a"),
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scratch_for_thread_with_single_root_returns_same_path() {
|
||||
let roots = vec![PathBuf::from("/scratch/only")];
|
||||
(0..5).for_each(|i| {
|
||||
assert_eq!(
|
||||
scratch_for_thread(&roots, Some(i)),
|
||||
Some(PathBuf::from("/scratch/only"))
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scratch_for_thread_falls_back_to_root_zero_outside_pool() {
|
||||
let roots = vec![
|
||||
PathBuf::from("/scratch/a"),
|
||||
PathBuf::from("/scratch/b"),
|
||||
];
|
||||
assert_eq!(
|
||||
scratch_for_thread(&roots, None),
|
||||
Some(PathBuf::from("/scratch/a"))
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -405,6 +405,8 @@ fn mount_ext4(device: &Path, target: &Path) -> Result<(), FlakyError> {
|
||||
let out = Command::new("mount")
|
||||
.arg("-t")
|
||||
.arg("ext4")
|
||||
.arg("-o")
|
||||
.arg("errors=continue")
|
||||
.arg(device)
|
||||
.arg(target)
|
||||
.output()?;
|
||||
|
||||
@@ -23,7 +23,7 @@ use crate::eventlog::{
|
||||
SegmentManager, SegmentReader, TimestampMicros, ValidEvent,
|
||||
};
|
||||
use crate::io::{RealIO, StorageIO};
|
||||
use crate::sim::{FaultConfig, SimulatedIO};
|
||||
use crate::sim::{FaultConfig, PristineGuard, SimulatedIO};
|
||||
|
||||
#[derive(Debug, Clone, Copy)]
|
||||
pub enum IoBackend {
|
||||
@@ -178,6 +178,7 @@ pub struct SharedState<S: StorageIO + Send + Sync + 'static> {
|
||||
|
||||
pub struct Gauntlet {
|
||||
config: GauntletConfig,
|
||||
scratch_root: Option<PathBuf>,
|
||||
}
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
@@ -185,7 +186,15 @@ pub enum GauntletBuildError {}
|
||||
|
||||
impl Gauntlet {
|
||||
pub fn new(config: GauntletConfig) -> Result<Self, GauntletBuildError> {
|
||||
Ok(Self { config })
|
||||
Ok(Self {
|
||||
config,
|
||||
scratch_root: None,
|
||||
})
|
||||
}
|
||||
|
||||
pub fn with_scratch_root(mut self, root: PathBuf) -> Self {
|
||||
self.scratch_root = Some(root);
|
||||
self
|
||||
}
|
||||
|
||||
pub fn generate_ops(&self) -> OpStream {
|
||||
@@ -211,6 +220,7 @@ impl Gauntlet {
|
||||
let ops_counter = Arc::new(AtomicUsize::new(0));
|
||||
let op_errors_counter = Arc::new(AtomicUsize::new(0));
|
||||
let restarts_counter = Arc::new(AtomicUsize::new(0));
|
||||
let scratch_root = self.scratch_root;
|
||||
let fut: std::pin::Pin<Box<dyn std::future::Future<Output = GauntletReport> + Send>> =
|
||||
match self.config.io {
|
||||
IoBackend::Real => Box::pin(run_inner_real(
|
||||
@@ -219,6 +229,7 @@ impl Gauntlet {
|
||||
ops_counter.clone(),
|
||||
op_errors_counter.clone(),
|
||||
restarts_counter.clone(),
|
||||
scratch_root,
|
||||
)),
|
||||
IoBackend::RealWithFlaky { flaky } => Box::pin(run_inner_real_with_flaky(
|
||||
self.config,
|
||||
@@ -269,8 +280,12 @@ async fn run_inner_real(
|
||||
ops_counter: Arc<AtomicUsize>,
|
||||
op_errors_counter: Arc<AtomicUsize>,
|
||||
restarts_counter: Arc<AtomicUsize>,
|
||||
scratch_root: Option<PathBuf>,
|
||||
) -> GauntletReport {
|
||||
let dir = tempfile::TempDir::new().expect("tempdir");
|
||||
let dir = match scratch_root.as_deref() {
|
||||
Some(parent) => tempfile::TempDir::new_in(parent).expect("tempdir in scratch root"),
|
||||
None => tempfile::TempDir::new().expect("tempdir"),
|
||||
};
|
||||
let root = dir.path().to_path_buf();
|
||||
let report = run_inner_real_on_root(
|
||||
config,
|
||||
@@ -364,7 +379,7 @@ async fn run_inner_real_on_root(
|
||||
let segments_dir = segments_subdir(&root);
|
||||
let open = {
|
||||
let segments_dir = segments_dir.clone();
|
||||
move || -> Result<Harness<RealIO>, String> {
|
||||
move |_attempt: usize| -> Result<Harness<RealIO>, String> {
|
||||
let store = TranquilBlockStore::open(cfg.clone())
|
||||
.map(Arc::new)
|
||||
.map_err(|e| e.to_string())?;
|
||||
@@ -424,7 +439,8 @@ async fn run_inner_simulated(
|
||||
let sim_for_open = Arc::clone(&sim);
|
||||
let open = {
|
||||
let segments_dir = segments_dir.clone();
|
||||
move || -> Result<Harness<Arc<SimulatedIO>>, String> {
|
||||
move |attempt: usize| -> Result<Harness<Arc<SimulatedIO>>, String> {
|
||||
let _pristine = PristineGuard::new(Arc::clone(&sim_for_open), attempt > 0);
|
||||
let factory_sim = Arc::clone(&sim_for_open);
|
||||
let make_io = move || Arc::clone(&factory_sim);
|
||||
let store = TranquilBlockStore::<Arc<SimulatedIO>>::open_with_io(cfg.clone(), make_io)
|
||||
@@ -512,28 +528,30 @@ async fn run_inner_generic<S, Open, Crash>(
|
||||
) -> GauntletReport
|
||||
where
|
||||
S: StorageIO + Send + Sync + 'static,
|
||||
Open: FnMut() -> Result<Harness<S>, String>,
|
||||
Open: FnMut(usize) -> Result<Harness<S>, String>,
|
||||
Crash: FnMut(),
|
||||
{
|
||||
let mut oracle = Oracle::new();
|
||||
let mut violations: Vec<InvariantViolation> = Vec::new();
|
||||
|
||||
let mut harness: Option<Harness<S>> = match open() {
|
||||
Ok(h) => Some(h),
|
||||
Err(e) => {
|
||||
return GauntletReport {
|
||||
seed: config.seed,
|
||||
ops_executed: OpsExecuted(0),
|
||||
op_errors: OpErrorCount(op_errors_counter.load(Ordering::Relaxed)),
|
||||
restarts: RestartCount(0),
|
||||
violations: vec![InvariantViolation {
|
||||
invariant: "OpenStore",
|
||||
detail: format!("initial open: {e}"),
|
||||
}],
|
||||
ops: OpStream::empty(),
|
||||
};
|
||||
}
|
||||
};
|
||||
let mut harness: Option<Harness<S>> =
|
||||
match reopen_with_recovery(&mut open, &mut crash, tolerate_op_errors, reopen_backoff).await
|
||||
{
|
||||
Ok(h) => Some(h),
|
||||
Err(e) => {
|
||||
return GauntletReport {
|
||||
seed: config.seed,
|
||||
ops_executed: OpsExecuted(0),
|
||||
op_errors: OpErrorCount(op_errors_counter.load(Ordering::Relaxed)),
|
||||
restarts: RestartCount(0),
|
||||
violations: vec![InvariantViolation {
|
||||
invariant: "OpenStore",
|
||||
detail: format!("initial open: {e}"),
|
||||
}],
|
||||
ops: OpStream::empty(),
|
||||
};
|
||||
}
|
||||
};
|
||||
let mut root: Option<Cid> = None;
|
||||
let mut restart_rng = Lcg::new(Seed(config.seed.0 ^ 0xA5A5_A5A5_A5A5_A5A5));
|
||||
let mut sample_rng = Lcg::new(Seed(config.seed.0 ^ 0x5A5A_5A5A_5A5A_5A5A));
|
||||
@@ -750,7 +768,7 @@ async fn reopen_with_recovery<S, Open, Crash>(
|
||||
) -> Result<Harness<S>, String>
|
||||
where
|
||||
S: StorageIO + Send + Sync + 'static,
|
||||
Open: FnMut() -> Result<Harness<S>, String>,
|
||||
Open: FnMut(usize) -> Result<Harness<S>, String>,
|
||||
Crash: FnMut(),
|
||||
{
|
||||
let mut errors: Vec<String> = Vec::new();
|
||||
@@ -758,7 +776,7 @@ where
|
||||
if attempt > 0 && !backoff.is_zero() {
|
||||
tokio::time::sleep(backoff).await;
|
||||
}
|
||||
match open() {
|
||||
match open(attempt) {
|
||||
Ok(h) => return Ok(h),
|
||||
Err(e) => {
|
||||
errors.push(format!("attempt {attempt}: {e}"));
|
||||
@@ -1191,6 +1209,10 @@ fn run_retention<S: StorageIO + Send + Sync + 'static>(
|
||||
max_age: RetentionSecs,
|
||||
) -> Result<(), String> {
|
||||
let sync_result = el.writer.sync().map_err(|e| e.to_string())?;
|
||||
el.manager
|
||||
.io()
|
||||
.sync_dir(el.segments_dir.as_path())
|
||||
.map_err(|e| e.to_string())?;
|
||||
let _ = el.writer.rotate_if_needed();
|
||||
oracle.record_event_sync(sync_result.synced_through);
|
||||
let active_id = sync_result.segment_id;
|
||||
@@ -1564,7 +1586,7 @@ async fn run_inner_generic_concurrent<S, Open, Crash>(
|
||||
) -> GauntletReport
|
||||
where
|
||||
S: StorageIO + Send + Sync + 'static,
|
||||
Open: FnMut() -> Result<Harness<S>, String>,
|
||||
Open: FnMut(usize) -> Result<Harness<S>, String>,
|
||||
Crash: FnMut(),
|
||||
{
|
||||
let ops: Vec<Op> = op_stream.into_vec();
|
||||
@@ -1577,22 +1599,24 @@ where
|
||||
let mut sample_rng = Lcg::new(Seed(config.seed.0 ^ 0x5A5A_5A5A_5A5A_5A5A));
|
||||
let chunks = compute_chunks(config.restart_policy, total_ops, &mut restart_rng);
|
||||
|
||||
let mut harness: Option<Harness<S>> = match open() {
|
||||
Ok(h) => Some(h),
|
||||
Err(e) => {
|
||||
return GauntletReport {
|
||||
seed: config.seed,
|
||||
ops_executed: OpsExecuted(0),
|
||||
op_errors: OpErrorCount(op_errors_counter.load(Ordering::Relaxed)),
|
||||
restarts: RestartCount(0),
|
||||
violations: vec![InvariantViolation {
|
||||
invariant: "OpenStore",
|
||||
detail: format!("initial open: {e}"),
|
||||
}],
|
||||
ops: OpStream::empty(),
|
||||
};
|
||||
}
|
||||
};
|
||||
let mut harness: Option<Harness<S>> =
|
||||
match reopen_with_recovery(&mut open, &mut crash, tolerate_op_errors, reopen_backoff).await
|
||||
{
|
||||
Ok(h) => Some(h),
|
||||
Err(e) => {
|
||||
return GauntletReport {
|
||||
seed: config.seed,
|
||||
ops_executed: OpsExecuted(0),
|
||||
op_errors: OpErrorCount(op_errors_counter.load(Ordering::Relaxed)),
|
||||
restarts: RestartCount(0),
|
||||
violations: vec![InvariantViolation {
|
||||
invariant: "OpenStore",
|
||||
detail: format!("initial open: {e}"),
|
||||
}],
|
||||
ops: OpStream::empty(),
|
||||
};
|
||||
}
|
||||
};
|
||||
let mut root: Option<Cid> = None;
|
||||
let mut oracle = Oracle::new();
|
||||
let mut halt_ops = false;
|
||||
@@ -1801,3 +1825,125 @@ where
|
||||
ops: OpStream::empty(),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn minimal_config() -> GauntletConfig {
|
||||
GauntletConfig {
|
||||
seed: Seed(0),
|
||||
io: IoBackend::Real,
|
||||
workload: WorkloadModel::default(),
|
||||
op_count: OpCount(0),
|
||||
invariants: InvariantSet::EMPTY,
|
||||
limits: RunLimits {
|
||||
max_wall_ms: Some(WallMs(30_000)),
|
||||
},
|
||||
restart_policy: RestartPolicy::Never,
|
||||
store: StoreConfig {
|
||||
max_file_size: MaxFileSize(8 * 1024),
|
||||
group_commit: GroupCommitConfig::default(),
|
||||
shard_count: ShardCount(1),
|
||||
},
|
||||
eventlog: None,
|
||||
writer_concurrency: WriterConcurrency(1),
|
||||
}
|
||||
}
|
||||
|
||||
fn flaky_open(
|
||||
attempts: Arc<AtomicUsize>,
|
||||
sim: Arc<SimulatedIO>,
|
||||
store_cfg: BlockStoreConfig,
|
||||
) -> impl FnMut(usize) -> Result<Harness<Arc<SimulatedIO>>, String> + Send + 'static {
|
||||
move |_attempt: usize| -> Result<Harness<Arc<SimulatedIO>>, String> {
|
||||
let n = attempts.fetch_add(1, Ordering::Relaxed);
|
||||
if n == 0 {
|
||||
return Err("simulated EIO on initial open".to_string());
|
||||
}
|
||||
let factory_sim = Arc::clone(&sim);
|
||||
let make_io = move || Arc::clone(&factory_sim);
|
||||
TranquilBlockStore::<Arc<SimulatedIO>>::open_with_io(store_cfg.clone(), make_io)
|
||||
.map(|s| Harness {
|
||||
store: Arc::new(s),
|
||||
eventlog: None,
|
||||
})
|
||||
.map_err(|e| e.to_string())
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn run_inner_generic_retries_initial_open_on_transient_io_error() {
|
||||
let dir = tempfile::TempDir::new().expect("tempdir");
|
||||
let cfg = minimal_config();
|
||||
let store_cfg = blockstore_config(dir.path(), &cfg.store);
|
||||
let sim: Arc<SimulatedIO> = Arc::new(SimulatedIO::pristine(0));
|
||||
let attempts = Arc::new(AtomicUsize::new(0));
|
||||
|
||||
let report = run_inner_generic::<Arc<SimulatedIO>, _, _>(
|
||||
cfg,
|
||||
OpStream::empty(),
|
||||
Arc::new(AtomicUsize::new(0)),
|
||||
Arc::new(AtomicUsize::new(0)),
|
||||
Arc::new(AtomicUsize::new(0)),
|
||||
flaky_open(Arc::clone(&attempts), Arc::clone(&sim), store_cfg),
|
||||
|| {},
|
||||
true,
|
||||
Duration::ZERO,
|
||||
)
|
||||
.await;
|
||||
|
||||
let opens: Vec<&InvariantViolation> = report
|
||||
.violations
|
||||
.iter()
|
||||
.filter(|v| v.invariant == "OpenStore")
|
||||
.collect();
|
||||
assert!(
|
||||
opens.is_empty(),
|
||||
"expected initial open to retry, got OpenStore violations: {opens:?}"
|
||||
);
|
||||
let total = attempts.load(Ordering::Relaxed);
|
||||
assert!(
|
||||
total >= 2,
|
||||
"expected at least one retry after first failure, attempts={total}"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn run_inner_generic_concurrent_retries_initial_open_on_transient_io_error() {
|
||||
let dir = tempfile::TempDir::new().expect("tempdir");
|
||||
let mut cfg = minimal_config();
|
||||
cfg.writer_concurrency = WriterConcurrency(2);
|
||||
let store_cfg = blockstore_config(dir.path(), &cfg.store);
|
||||
let sim: Arc<SimulatedIO> = Arc::new(SimulatedIO::pristine(0));
|
||||
let attempts = Arc::new(AtomicUsize::new(0));
|
||||
|
||||
let report = run_inner_generic_concurrent::<Arc<SimulatedIO>, _, _>(
|
||||
cfg,
|
||||
OpStream::empty(),
|
||||
Arc::new(AtomicUsize::new(0)),
|
||||
Arc::new(AtomicUsize::new(0)),
|
||||
Arc::new(AtomicUsize::new(0)),
|
||||
flaky_open(Arc::clone(&attempts), Arc::clone(&sim), store_cfg),
|
||||
|| {},
|
||||
true,
|
||||
Duration::ZERO,
|
||||
)
|
||||
.await;
|
||||
|
||||
let opens: Vec<&InvariantViolation> = report
|
||||
.violations
|
||||
.iter()
|
||||
.filter(|v| v.invariant == "OpenStore")
|
||||
.collect();
|
||||
assert!(
|
||||
opens.is_empty(),
|
||||
"expected initial open to retry, got OpenStore violations: {opens:?}"
|
||||
);
|
||||
let total = attempts.load(Ordering::Relaxed);
|
||||
assert!(
|
||||
total >= 2,
|
||||
"expected at least one retry after first failure, attempts={total}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -9,7 +9,7 @@ use super::workload::{
|
||||
ByteRange, DidSpaceSize, KeySpaceSize, OpCount, OpWeights, RetentionMaxSecs, SizeDistribution,
|
||||
ValueBytes, WorkloadModel,
|
||||
};
|
||||
use crate::blockstore::GroupCommitConfig;
|
||||
use crate::blockstore::{GroupCommitConfig, MAX_BLOCK_SIZE};
|
||||
use crate::sim::FaultConfig;
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
@@ -402,7 +402,7 @@ fn huge_values(seed: Seed) -> GauntletConfig {
|
||||
workload: block_workload(
|
||||
block_weights(85, 5, 8, 2),
|
||||
SizeDistribution::HeavyTail(
|
||||
ByteRange::new(ValueBytes(256), ValueBytes(16 * 1024 * 1024))
|
||||
ByteRange::new(ValueBytes(256), ValueBytes(MAX_BLOCK_SIZE))
|
||||
.expect("huge_values ByteRange"),
|
||||
),
|
||||
KeySpaceSize(64),
|
||||
|
||||
@@ -104,6 +104,10 @@ pub trait StorageIO: Send + Sync {
|
||||
fn sync_dir(&self, path: &Path) -> io::Result<()>;
|
||||
fn list_dir(&self, path: &Path) -> io::Result<Vec<PathBuf>>;
|
||||
|
||||
fn barrier(&self) -> io::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn write_all_at(&self, fd: FileId, offset: u64, buf: &[u8]) -> io::Result<()> {
|
||||
let written = Cell::new(0usize);
|
||||
std::iter::from_fn(|| (written.get() < buf.len()).then_some(()))
|
||||
@@ -190,6 +194,9 @@ impl<S: StorageIO> StorageIO for Arc<S> {
|
||||
fn list_dir(&self, path: &Path) -> io::Result<Vec<PathBuf>> {
|
||||
(**self).list_dir(path)
|
||||
}
|
||||
fn barrier(&self) -> io::Result<()> {
|
||||
(**self).barrier()
|
||||
}
|
||||
fn mmap_file(&self, fd: FileId) -> io::Result<MappedFile> {
|
||||
(**self).mmap_file(fd)
|
||||
}
|
||||
|
||||
@@ -28,7 +28,7 @@ pub use record::{
|
||||
};
|
||||
#[cfg(any(test, feature = "test-harness"))]
|
||||
pub use sim::{
|
||||
FaultConfig, LatencyNs, OpRecord, Probability, SimulatedIO, SyncReorderWindow,
|
||||
FaultConfig, LatencyNs, OpRecord, Probability, PristineGuard, SimulatedIO, SyncReorderWindow,
|
||||
sim_proptest_cases, sim_seed_count, sim_seed_range, sim_single_seed,
|
||||
};
|
||||
|
||||
|
||||
@@ -2,7 +2,8 @@ use std::collections::{HashMap, HashSet, VecDeque};
|
||||
use std::io;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::Mutex;
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use crate::io::{FileId, OpenOptions, StorageIO};
|
||||
@@ -226,6 +227,7 @@ pub enum OpRecord {
|
||||
SyncDir {
|
||||
path: PathBuf,
|
||||
},
|
||||
Barrier,
|
||||
}
|
||||
|
||||
struct PendingSync {
|
||||
@@ -326,6 +328,7 @@ impl SimState {
|
||||
pub struct SimulatedIO {
|
||||
state: Mutex<SimState>,
|
||||
fault_config: FaultConfig,
|
||||
pristine_mode: AtomicBool,
|
||||
rng_seed: u64,
|
||||
latency_counter: AtomicU64,
|
||||
}
|
||||
@@ -346,13 +349,26 @@ impl SimulatedIO {
|
||||
pending_deletes: Vec::new(),
|
||||
}),
|
||||
fault_config,
|
||||
pristine_mode: AtomicBool::new(false),
|
||||
rng_seed: seed,
|
||||
latency_counter: AtomicU64::new(0),
|
||||
}
|
||||
}
|
||||
|
||||
fn effective_fault_config(&self) -> FaultConfig {
|
||||
if self.pristine_mode.load(Ordering::Relaxed) {
|
||||
FaultConfig::none()
|
||||
} else {
|
||||
self.fault_config
|
||||
}
|
||||
}
|
||||
|
||||
pub fn set_pristine_mode(&self, on: bool) {
|
||||
self.pristine_mode.store(on, Ordering::Relaxed);
|
||||
}
|
||||
|
||||
fn jitter(&self) {
|
||||
let max_ns = self.fault_config.latency_distribution_ns.0;
|
||||
let max_ns = self.effective_fault_config().latency_distribution_ns.0;
|
||||
if max_ns == 0 {
|
||||
return;
|
||||
}
|
||||
@@ -429,12 +445,30 @@ impl SimulatedIO {
|
||||
}
|
||||
}
|
||||
|
||||
pub struct PristineGuard {
|
||||
sim: Arc<SimulatedIO>,
|
||||
}
|
||||
|
||||
impl PristineGuard {
|
||||
pub fn new(sim: Arc<SimulatedIO>, on: bool) -> Self {
|
||||
sim.set_pristine_mode(on);
|
||||
Self { sim }
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for PristineGuard {
|
||||
fn drop(&mut self) {
|
||||
self.sim.set_pristine_mode(false);
|
||||
}
|
||||
}
|
||||
|
||||
impl StorageIO for SimulatedIO {
|
||||
fn open(&self, path: &Path, opts: OpenOptions) -> io::Result<FileId> {
|
||||
let fault = self.effective_fault_config();
|
||||
let mut state = self.state.lock().unwrap();
|
||||
let seed = self.rng_seed;
|
||||
|
||||
if state.should_fault(seed, self.fault_config.io_error_probability) {
|
||||
if state.should_fault(seed, fault.io_error_probability) {
|
||||
return Err(io::Error::other("simulated EIO on open"));
|
||||
}
|
||||
|
||||
@@ -514,6 +548,7 @@ impl StorageIO for SimulatedIO {
|
||||
|
||||
fn read_at(&self, id: FileId, offset: u64, buf: &mut [u8]) -> io::Result<usize> {
|
||||
self.jitter();
|
||||
let fault = self.effective_fault_config();
|
||||
let mut state = self.state.lock().unwrap();
|
||||
let sid = state.require_readable(id)?;
|
||||
let seed = self.rng_seed;
|
||||
@@ -522,12 +557,12 @@ impl StorageIO for SimulatedIO {
|
||||
return Err(io::Error::other("simulated EIO after delayed sync fault"));
|
||||
}
|
||||
|
||||
if state.should_fault(seed, self.fault_config.io_error_probability) {
|
||||
if state.should_fault(seed, fault.io_error_probability) {
|
||||
return Err(io::Error::other("simulated EIO on read"));
|
||||
}
|
||||
|
||||
let read_offset =
|
||||
if state.should_fault(seed, self.fault_config.misdirected_read_probability) {
|
||||
if state.should_fault(seed, fault.misdirected_read_probability) {
|
||||
let drift_sectors = state.next_random_usize(seed, 8) + 1;
|
||||
let drift = (drift_sectors * SECTOR_BYTES) as u64;
|
||||
if state.next_random(seed) < 0.5 {
|
||||
@@ -556,7 +591,7 @@ impl StorageIO for SimulatedIO {
|
||||
let to_read = buf.len().min(available);
|
||||
buf[..to_read].copy_from_slice(&storage.buffered[off..off + to_read]);
|
||||
|
||||
if state.should_fault(seed, self.fault_config.bit_flip_on_read_probability) && to_read > 0 {
|
||||
if state.should_fault(seed, fault.bit_flip_on_read_probability) && to_read > 0 {
|
||||
let flip_pos = state.next_random_usize(seed, to_read);
|
||||
let flip_bit = state.next_random_usize(seed, 8);
|
||||
buf[flip_pos] ^= 1 << flip_bit;
|
||||
@@ -572,6 +607,7 @@ impl StorageIO for SimulatedIO {
|
||||
|
||||
fn write_at(&self, id: FileId, offset: u64, buf: &[u8]) -> io::Result<usize> {
|
||||
self.jitter();
|
||||
let fault = self.effective_fault_config();
|
||||
let mut state = self.state.lock().unwrap();
|
||||
let sid = state.require_writable(id)?;
|
||||
let seed = self.rng_seed;
|
||||
@@ -580,12 +616,12 @@ impl StorageIO for SimulatedIO {
|
||||
return Err(io::Error::other("simulated EIO after delayed sync fault"));
|
||||
}
|
||||
|
||||
if state.should_fault(seed, self.fault_config.io_error_probability) {
|
||||
if state.should_fault(seed, fault.io_error_probability) {
|
||||
return Err(io::Error::other("simulated EIO on write"));
|
||||
}
|
||||
|
||||
let torn_len =
|
||||
if buf.len() > 1 && state.should_fault(seed, self.fault_config.torn_page_probability) {
|
||||
if buf.len() > 1 && state.should_fault(seed, fault.torn_page_probability) {
|
||||
let page_base = (offset as usize) - ((offset as usize) % TORN_PAGE_BYTES);
|
||||
let page_end = page_base + TORN_PAGE_BYTES;
|
||||
let cap = page_end.saturating_sub(offset as usize).min(buf.len());
|
||||
@@ -601,7 +637,7 @@ impl StorageIO for SimulatedIO {
|
||||
let actual_len = match torn_len {
|
||||
Some(n) => n,
|
||||
None if buf.len() > 1
|
||||
&& state.should_fault(seed, self.fault_config.partial_write_probability) =>
|
||||
&& state.should_fault(seed, fault.partial_write_probability) =>
|
||||
{
|
||||
let partial = state.next_random_usize(seed, buf.len());
|
||||
partial.max(1)
|
||||
@@ -609,7 +645,7 @@ impl StorageIO for SimulatedIO {
|
||||
None => buf.len(),
|
||||
};
|
||||
|
||||
let misdirected = state.should_fault(seed, self.fault_config.misdirected_write_probability);
|
||||
let misdirected = state.should_fault(seed, fault.misdirected_write_probability);
|
||||
let write_offset = if misdirected {
|
||||
let drift_sectors = state.next_random_usize(seed, 8) + 1;
|
||||
let drift = (drift_sectors * SECTOR_BYTES) as u64;
|
||||
@@ -643,6 +679,7 @@ impl StorageIO for SimulatedIO {
|
||||
|
||||
fn sync(&self, id: FileId) -> io::Result<()> {
|
||||
self.jitter();
|
||||
let fault = self.effective_fault_config();
|
||||
let mut state = self.state.lock().unwrap();
|
||||
let sid = state.require_open(id)?;
|
||||
let seed = self.rng_seed;
|
||||
@@ -651,16 +688,22 @@ impl StorageIO for SimulatedIO {
|
||||
return Err(io::Error::other("simulated EIO after delayed sync fault"));
|
||||
}
|
||||
|
||||
if state.should_fault(seed, self.fault_config.io_error_probability) {
|
||||
if state.should_fault(seed, fault.io_error_probability) {
|
||||
return Err(io::Error::other("simulated EIO on sync"));
|
||||
}
|
||||
|
||||
let sync_succeeded = !state.should_fault(seed, self.fault_config.sync_failure_probability);
|
||||
let poison_after = sync_succeeded
|
||||
&& state.should_fault(seed, self.fault_config.delayed_io_error_probability);
|
||||
let reorder_window = self.fault_config.sync_reorder_window.0 as usize;
|
||||
if state.should_fault(seed, fault.sync_failure_probability) {
|
||||
state.op_log.push(OpRecord::Sync {
|
||||
fd: id,
|
||||
succeeded: false,
|
||||
});
|
||||
return Err(io::Error::other("simulated dropped fsync"));
|
||||
}
|
||||
|
||||
let evicted = if sync_succeeded && reorder_window > 0 {
|
||||
let poison_after = state.should_fault(seed, fault.delayed_io_error_probability);
|
||||
let reorder_window = fault.sync_reorder_window.0 as usize;
|
||||
|
||||
let evicted = if reorder_window > 0 {
|
||||
let snapshot = state.storage.get(&sid).unwrap().buffered.clone();
|
||||
state.pending_syncs.push_back(PendingSync {
|
||||
storage_id: sid,
|
||||
@@ -686,7 +729,7 @@ impl StorageIO for SimulatedIO {
|
||||
|
||||
let storage = state.storage.get_mut(&sid).unwrap();
|
||||
|
||||
if sync_succeeded && reorder_window == 0 {
|
||||
if reorder_window == 0 {
|
||||
storage.durable = storage.buffered.clone();
|
||||
}
|
||||
if poison_after {
|
||||
@@ -695,7 +738,7 @@ impl StorageIO for SimulatedIO {
|
||||
|
||||
state.op_log.push(OpRecord::Sync {
|
||||
fd: id,
|
||||
succeeded: sync_succeeded,
|
||||
succeeded: true,
|
||||
});
|
||||
Ok(())
|
||||
}
|
||||
@@ -774,17 +817,31 @@ impl StorageIO for SimulatedIO {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn barrier(&self) -> io::Result<()> {
|
||||
self.jitter();
|
||||
let mut state = self.state.lock().unwrap();
|
||||
let drained: Vec<PendingSync> = state.pending_syncs.drain(..).collect();
|
||||
drained.into_iter().for_each(|p| {
|
||||
if let Some(storage) = state.storage.get_mut(&p.storage_id) {
|
||||
storage.durable = p.snapshot;
|
||||
}
|
||||
});
|
||||
state.op_log.push(OpRecord::Barrier);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn sync_dir(&self, path: &Path) -> io::Result<()> {
|
||||
let fault = self.effective_fault_config();
|
||||
let mut state = self.state.lock().unwrap();
|
||||
let seed = self.rng_seed;
|
||||
|
||||
if state.should_fault(seed, self.fault_config.io_error_probability) {
|
||||
if state.should_fault(seed, fault.io_error_probability) {
|
||||
return Err(io::Error::other("simulated EIO on sync_dir"));
|
||||
}
|
||||
|
||||
let dir_path = path.to_path_buf();
|
||||
let actually_persisted =
|
||||
!state.should_fault(seed, self.fault_config.dir_sync_failure_probability);
|
||||
!state.should_fault(seed, fault.dir_sync_failure_probability);
|
||||
|
||||
if actually_persisted {
|
||||
state.dirs_durable.insert(dir_path.clone());
|
||||
|
||||
@@ -460,3 +460,115 @@ async fn mst_restart_churn_single_seed() {
|
||||
assert_clean(&report);
|
||||
assert!(report.restarts.0 >= 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn torn_pages_only_completes_within_budget() {
|
||||
let cfg = GauntletConfig {
|
||||
seed: Seed(0),
|
||||
io: IoBackend::Simulated {
|
||||
fault: FaultConfig::torn_pages_only(),
|
||||
},
|
||||
workload: WorkloadModel {
|
||||
weights: OpWeights {
|
||||
add: 80,
|
||||
delete: 10,
|
||||
compact: 5,
|
||||
checkpoint: 5,
|
||||
..OpWeights::default()
|
||||
},
|
||||
size_distribution: SizeDistribution::Fixed(ValueBytes(128)),
|
||||
collections: vec![
|
||||
CollectionName("app.bsky.feed.post".to_string()),
|
||||
CollectionName("app.bsky.feed.like".to_string()),
|
||||
],
|
||||
key_space: KeySpaceSize(500),
|
||||
did_space: DidSpaceSize(32),
|
||||
retention_max_secs: RetentionMaxSecs(3600),
|
||||
},
|
||||
op_count: OpCount(2_000),
|
||||
invariants: InvariantSet::REFCOUNT_CONSERVATION
|
||||
| InvariantSet::REACHABILITY
|
||||
| InvariantSet::ACKED_WRITE_PERSISTENCE
|
||||
| InvariantSet::READ_AFTER_WRITE
|
||||
| InvariantSet::RESTART_IDEMPOTENT,
|
||||
limits: RunLimits {
|
||||
max_wall_ms: Some(WallMs(60_000)),
|
||||
},
|
||||
restart_policy: RestartPolicy::CrashAtSyscall(OpInterval(500)),
|
||||
store: StoreConfig {
|
||||
max_file_size: MaxFileSize(16 * 1024),
|
||||
group_commit: GroupCommitConfig {
|
||||
verify_persisted_blocks: true,
|
||||
..GroupCommitConfig::default()
|
||||
},
|
||||
shard_count: ShardCount(1),
|
||||
},
|
||||
eventlog: None,
|
||||
writer_concurrency: WriterConcurrency(1),
|
||||
};
|
||||
let report = Gauntlet::new(cfg).expect("build gauntlet").run().await;
|
||||
let budget_violations: Vec<&str> = report
|
||||
.violations
|
||||
.iter()
|
||||
.filter(|v| v.invariant == "WallClockBudget")
|
||||
.map(|v| v.detail.as_str())
|
||||
.collect();
|
||||
assert!(
|
||||
budget_violations.is_empty(),
|
||||
"torn-pages exceeded budget: {budget_violations:?}; ops_executed={}",
|
||||
report.ops_executed.0
|
||||
);
|
||||
assert_eq!(
|
||||
report.ops_executed.0, 2_000,
|
||||
"expected all ops to execute under torn-pages-only faults"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn real_io_gauntlet_uses_scratch_root_for_tempdir() {
|
||||
let scratch = tempfile::TempDir::new().expect("scratch dir");
|
||||
let scratch_path = scratch.path().to_path_buf();
|
||||
let cfg = fast_sanity_config(Seed(11));
|
||||
let report = Gauntlet::new(cfg)
|
||||
.expect("build gauntlet")
|
||||
.with_scratch_root(scratch_path.clone())
|
||||
.run()
|
||||
.await;
|
||||
assert_clean(&report);
|
||||
let entries: Vec<std::path::PathBuf> = std::fs::read_dir(&scratch_path)
|
||||
.expect("read scratch")
|
||||
.filter_map(|e| e.ok().map(|e| e.path()))
|
||||
.collect();
|
||||
assert!(
|
||||
entries.is_empty(),
|
||||
"scratch root must be empty after gauntlet drop, found: {entries:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn farm_run_many_timed_with_scratch_roots_honors_assignment() {
|
||||
let scratch = tempfile::TempDir::new().expect("scratch dir");
|
||||
let root_a = scratch.path().join("a");
|
||||
let root_b = scratch.path().join("b");
|
||||
std::fs::create_dir_all(&root_a).expect("mkdir a");
|
||||
std::fs::create_dir_all(&root_b).expect("mkdir b");
|
||||
let roots = vec![root_a.clone(), root_b.clone()];
|
||||
let reports = farm::run_many_timed_with_scratch_roots(
|
||||
|seed| fast_sanity_config(seed),
|
||||
&roots,
|
||||
(0..2).map(Seed),
|
||||
);
|
||||
assert_eq!(reports.len(), 2);
|
||||
reports.iter().for_each(|(r, _)| assert_clean(r));
|
||||
[&root_a, &root_b].iter().for_each(|root| {
|
||||
let leftover: Vec<std::path::PathBuf> = std::fs::read_dir(root)
|
||||
.expect("read scratch root")
|
||||
.filter_map(|e| e.ok().map(|e| e.path()))
|
||||
.collect();
|
||||
assert!(
|
||||
leftover.is_empty(),
|
||||
"scratch root {} must be empty after farm completes, found: {leftover:?}",
|
||||
root.display()
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
@@ -12,7 +12,9 @@ use tranquil_store::blockstore::{
|
||||
GroupCommitConfig, HINT_RECORD_SIZE, HintFileWriter, HintOffset, TranquilBlockStore,
|
||||
WallClockMs, WriteCursor, hint_file_path,
|
||||
};
|
||||
use tranquil_store::{FaultConfig, OpenOptions, SimulatedIO, StorageIO, sim_seed_range};
|
||||
use tranquil_store::{
|
||||
FaultConfig, OpenOptions, SimulatedIO, StorageIO, SyncReorderWindow, sim_seed_range,
|
||||
};
|
||||
|
||||
use common::{Rng, advance_epoch, block_data, test_cid, with_runtime};
|
||||
|
||||
@@ -691,3 +693,60 @@ fn sim_multi_file_rotation_crash_recovery() {
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sim_sync_reorder_loses_first_commit_durability() {
|
||||
with_runtime(|| {
|
||||
let dir = tempfile::TempDir::new().unwrap();
|
||||
let config = BlockStoreConfig {
|
||||
data_dir: dir.path().join("data"),
|
||||
index_dir: dir.path().join("index"),
|
||||
max_file_size: DEFAULT_MAX_FILE_SIZE,
|
||||
group_commit: GroupCommitConfig::default(),
|
||||
shard_count: 1,
|
||||
};
|
||||
|
||||
let fault = FaultConfig {
|
||||
sync_reorder_window: SyncReorderWindow(4),
|
||||
..FaultConfig::none()
|
||||
};
|
||||
let sim: Arc<SimulatedIO> = Arc::new(SimulatedIO::new(706, fault));
|
||||
|
||||
let cid = test_cid(0);
|
||||
let data = block_data(0);
|
||||
|
||||
{
|
||||
let s = Arc::clone(&sim);
|
||||
let store = TranquilBlockStore::<Arc<SimulatedIO>>::open_with_io(
|
||||
config.clone(),
|
||||
move || Arc::clone(&s),
|
||||
)
|
||||
.unwrap();
|
||||
store
|
||||
.put_blocks_blocking(vec![(cid, data.clone())])
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
sim.crash();
|
||||
|
||||
let s = Arc::clone(&sim);
|
||||
let store = TranquilBlockStore::<Arc<SimulatedIO>>::open_with_io(config, move || {
|
||||
Arc::clone(&s)
|
||||
})
|
||||
.unwrap();
|
||||
|
||||
match store.get_block_sync(&cid) {
|
||||
Ok(Some(d)) => assert_eq!(
|
||||
&d[..],
|
||||
&data[..],
|
||||
"block content mismatch after crash"
|
||||
),
|
||||
Ok(None) => panic!(
|
||||
"durability bug: put_blocks_blocking returned Ok but block missing after crash"
|
||||
),
|
||||
Err(e) => panic!(
|
||||
"durability bug: block read failed after crash: {e}"
|
||||
),
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@@ -5,10 +5,12 @@ use std::sync::Arc;
|
||||
|
||||
use rayon::prelude::*;
|
||||
use tranquil_store::eventlog::{
|
||||
DidHash, EVENT_RECORD_OVERHEAD, EventLogWriter, EventSequence, EventTypeTag, MAX_EVENT_PAYLOAD,
|
||||
SEGMENT_HEADER_SIZE, SegmentId, SegmentManager, SegmentReader, ValidEvent,
|
||||
DidHash, EVENT_HEADER_SIZE, EVENT_RECORD_OVERHEAD, EventLogWriter, EventSequence, EventTypeTag,
|
||||
MAX_EVENT_PAYLOAD, SEGMENT_HEADER_SIZE, SegmentId, SegmentManager, SegmentReader, ValidEvent,
|
||||
};
|
||||
use tranquil_store::{
|
||||
FaultConfig, OpenOptions, Probability, SimulatedIO, StorageIO, sim_seed_range,
|
||||
};
|
||||
use tranquil_store::{FaultConfig, Probability, SimulatedIO, StorageIO, sim_seed_range};
|
||||
|
||||
use common::Rng;
|
||||
|
||||
@@ -204,13 +206,13 @@ fn crash_mid_rotation_with_faults() {
|
||||
EventLogWriter::open(Arc::clone(&mgr), 256, MAX_EVENT_PAYLOAD)
|
||||
}));
|
||||
|
||||
if let Ok(Ok(writer)) = recovery
|
||||
&& let Ok(synced_before) = write_result
|
||||
{
|
||||
if let Ok(Ok(writer)) = recovery {
|
||||
let recovered = writer.synced_seq().raw();
|
||||
assert!(
|
||||
writer.synced_seq().raw() <= synced_before,
|
||||
"seed {seed}: recovered more events than were synced"
|
||||
recovered <= events_per_seg as u64,
|
||||
"seed {seed}: recovered {recovered} > written {events_per_seg}"
|
||||
);
|
||||
let _ = write_result;
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -1020,3 +1022,169 @@ fn aggressive_faults_group_sync_recovery() {
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sync_synced_seq_must_match_durable_valid_prefix() {
|
||||
let asserted = std::sync::atomic::AtomicU64::new(0);
|
||||
let range = sim_seed_range();
|
||||
let total = range.end - range.start;
|
||||
range.into_par_iter().for_each(|seed| {
|
||||
let fault_config = FaultConfig {
|
||||
partial_write_probability: Probability::new(0.05),
|
||||
torn_page_probability: Probability::new(0.01),
|
||||
misdirected_write_probability: Probability::new(0.01),
|
||||
sync_failure_probability: Probability::new(0.03),
|
||||
sync_reorder_window: tranquil_store::SyncReorderWindow(4),
|
||||
..FaultConfig::none()
|
||||
};
|
||||
let sim = SimulatedIO::new(seed, fault_config);
|
||||
let mgr = setup_manager(sim, 64 * 1024);
|
||||
|
||||
let Ok(mut writer) = EventLogWriter::open(Arc::clone(&mgr), 256, MAX_EVENT_PAYLOAD) else {
|
||||
return;
|
||||
};
|
||||
|
||||
let event_count = 10u64;
|
||||
(1..=event_count).for_each(|i| {
|
||||
let _ = append_test_event(&mut writer, i, seed);
|
||||
});
|
||||
|
||||
let synced_through = match writer.sync() {
|
||||
Ok(r) => r.synced_through.raw(),
|
||||
Err(_) => return,
|
||||
};
|
||||
let _ = mgr.io().sync_dir(Path::new(SEGMENTS_DIR));
|
||||
|
||||
if synced_through == 0 {
|
||||
return;
|
||||
}
|
||||
|
||||
let Ok(handle) = mgr.open_for_read(SegmentId::new(1)) else {
|
||||
return;
|
||||
};
|
||||
let Ok(reader) = SegmentReader::open(mgr.io(), handle.fd(), MAX_EVENT_PAYLOAD) else {
|
||||
return;
|
||||
};
|
||||
let Ok(valid) = reader.valid_prefix() else {
|
||||
return;
|
||||
};
|
||||
|
||||
let durable_max = valid.last().map(|e| e.seq.raw()).unwrap_or(0);
|
||||
|
||||
asserted.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
|
||||
assert!(
|
||||
synced_through <= durable_max,
|
||||
"seed {seed}: sync acked seq {synced_through} but durable valid prefix only reaches {durable_max}, events written: {event_count}, valid_prefix.len()={}",
|
||||
valid.len()
|
||||
);
|
||||
});
|
||||
|
||||
let asserted = asserted.load(std::sync::atomic::Ordering::Relaxed);
|
||||
if total >= 50 {
|
||||
assert!(
|
||||
asserted * 2 >= total,
|
||||
"fewer than half of {total} seeds reached the durability assertion: {asserted}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reopen_recovers_from_torn_segment_header() {
|
||||
let sim = SimulatedIO::pristine(0);
|
||||
let mgr = setup_manager(sim, 64 * 1024);
|
||||
|
||||
{
|
||||
let mut writer = EventLogWriter::open(Arc::clone(&mgr), 256, MAX_EVENT_PAYLOAD).unwrap();
|
||||
(1..=3).for_each(|i| {
|
||||
let _ = append_test_event(&mut writer, i, 0);
|
||||
});
|
||||
writer.sync().unwrap();
|
||||
}
|
||||
mgr.shutdown();
|
||||
|
||||
let path = mgr.segment_path(SegmentId::new(1));
|
||||
let fd = mgr
|
||||
.io()
|
||||
.open(&path, OpenOptions::read_write_existing())
|
||||
.unwrap();
|
||||
mgr.io().write_all_at(fd, 0, &[0u8; 4]).unwrap();
|
||||
mgr.io().sync(fd).unwrap();
|
||||
mgr.io().sync_dir(Path::new(SEGMENTS_DIR)).unwrap();
|
||||
mgr.io().close(fd).unwrap();
|
||||
|
||||
let writer = EventLogWriter::open(Arc::clone(&mgr), 256, MAX_EVENT_PAYLOAD)
|
||||
.expect("reopen with torn header on highest-numbered segment must succeed");
|
||||
assert_eq!(writer.active_segment_id(), SegmentId::new(1));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn partial_valid_sync_poisons_writer_and_acks_only_valid_prefix() {
|
||||
let sim = SimulatedIO::pristine(0);
|
||||
let mgr = setup_manager(sim, 64 * 1024);
|
||||
let mut writer = EventLogWriter::open(Arc::clone(&mgr), 256, MAX_EVENT_PAYLOAD).unwrap();
|
||||
|
||||
let payload = b"payload-x".to_vec();
|
||||
let payload_size = payload.len();
|
||||
let record_size = EVENT_RECORD_OVERHEAD + payload_size;
|
||||
|
||||
(1..=5u64).for_each(|i| {
|
||||
writer
|
||||
.append(
|
||||
DidHash::from_did(&format!("did:plc:user{i}")),
|
||||
EventTypeTag::COMMIT,
|
||||
payload.clone(),
|
||||
)
|
||||
.unwrap();
|
||||
});
|
||||
|
||||
let event_3_start = SEGMENT_HEADER_SIZE + 2 * record_size;
|
||||
let event_3_checksum_offset = event_3_start + EVENT_HEADER_SIZE + payload_size;
|
||||
|
||||
let segment_path = mgr.segment_path(SegmentId::new(1));
|
||||
let corrupt_fd = mgr
|
||||
.io()
|
||||
.open(&segment_path, OpenOptions::read_write_existing())
|
||||
.unwrap();
|
||||
mgr.io()
|
||||
.write_all_at(corrupt_fd, event_3_checksum_offset as u64, &[0xFFu8; 4])
|
||||
.unwrap();
|
||||
mgr.io().close(corrupt_fd).unwrap();
|
||||
|
||||
let result = writer.sync().unwrap();
|
||||
assert_eq!(
|
||||
result.synced_through,
|
||||
EventSequence::new(2),
|
||||
"sync must ack only events 1..=2 with corrupt event 3"
|
||||
);
|
||||
assert_eq!(result.flushed_events.len(), 2);
|
||||
assert!(writer.is_poisoned(), "writer must be poisoned after partial sync");
|
||||
|
||||
let append_after_poison = writer.append(
|
||||
DidHash::from_did("did:plc:after"),
|
||||
EventTypeTag::COMMIT,
|
||||
payload.clone(),
|
||||
);
|
||||
assert!(
|
||||
append_after_poison.is_err(),
|
||||
"append must fail on poisoned writer"
|
||||
);
|
||||
|
||||
let sync_after_poison = writer.sync();
|
||||
assert!(
|
||||
sync_after_poison.is_err(),
|
||||
"sync must fail on poisoned writer"
|
||||
);
|
||||
|
||||
drop(writer);
|
||||
let recovered = EventLogWriter::open(Arc::clone(&mgr), 256, MAX_EVENT_PAYLOAD).unwrap();
|
||||
assert_eq!(
|
||||
recovered.synced_seq(),
|
||||
EventSequence::new(2),
|
||||
"reopen must observe synced_seq matching disk's valid prefix"
|
||||
);
|
||||
|
||||
let valid = read_all_events(&mgr, 0);
|
||||
assert_eq!(valid.len(), 2);
|
||||
assert_eq!(valid[0].seq, EventSequence::new(1));
|
||||
assert_eq!(valid[1].seq, EventSequence::new(2));
|
||||
}
|
||||
|
||||
+109
-4
@@ -373,12 +373,117 @@
|
||||
# Default value: "Tranquil PDS"
|
||||
#from_name = "Tranquil PDS"
|
||||
|
||||
# Path to the `sendmail` binary.
|
||||
# HELO/EHLO name announced to remote SMTP servers. Applies to both
|
||||
# smarthost and direct-MX modes. Defaults to the server hostname.
|
||||
#
|
||||
# Can also be specified via environment variable `SENDMAIL_PATH`.
|
||||
# Can also be specified via environment variable `MAIL_HELO_NAME`.
|
||||
#helo_name =
|
||||
|
||||
[email.smarthost]
|
||||
# SMTP relay host. When set, mail is delivered through this host
|
||||
# instead of resolving recipient MX records directly.
|
||||
#
|
||||
# Default value: "/usr/sbin/sendmail"
|
||||
#sendmail_path = "/usr/sbin/sendmail"
|
||||
# Can also be specified via environment variable `MAIL_SMARTHOST_HOST`.
|
||||
#host =
|
||||
|
||||
# SMTP relay port.
|
||||
#
|
||||
# Can also be specified via environment variable `MAIL_SMARTHOST_PORT`.
|
||||
#
|
||||
# Default value: 587
|
||||
#port = 587
|
||||
|
||||
# SMTP authentication username.
|
||||
#
|
||||
# Can also be specified via environment variable `MAIL_SMARTHOST_USERNAME`.
|
||||
#username =
|
||||
|
||||
# SMTP authentication password.
|
||||
#
|
||||
# Can also be specified via environment variable `MAIL_SMARTHOST_PASSWORD`.
|
||||
#password =
|
||||
|
||||
# TLS mode. Valid values: "implicit", "starttls", "none". Setting "none"
|
||||
# alongside a password is rejected at startup to prevent transmitting
|
||||
# credentials in plaintext.
|
||||
#
|
||||
# Can also be specified via environment variable `MAIL_SMARTHOST_TLS`.
|
||||
#
|
||||
# Default value: "starttls"
|
||||
#tls = "starttls"
|
||||
|
||||
# Max size of the connection pool.
|
||||
#
|
||||
# Can also be specified via environment variable `MAIL_SMARTHOST_POOL_SIZE`.
|
||||
#
|
||||
# Default value: 4
|
||||
#pool_size = 4
|
||||
|
||||
# Per-command SMTP timeout in seconds. Bounds the security handshake.
|
||||
#
|
||||
# Can also be specified via environment variable `MAIL_SMARTHOST_COMMAND_TIMEOUT_SECS`.
|
||||
#
|
||||
# Default value: 30
|
||||
#command_timeout_secs = 30
|
||||
|
||||
# Total per-message timeout in seconds. Wraps the entire send so a
|
||||
# stuck relay cannot stall the comms queue.
|
||||
#
|
||||
# Can also be specified via environment variable `MAIL_SMARTHOST_TOTAL_TIMEOUT_SECS`.
|
||||
#
|
||||
# Default value: 60
|
||||
#total_timeout_secs = 60
|
||||
|
||||
[email.direct_mx]
|
||||
# Per-command SMTP timeout in seconds.
|
||||
#
|
||||
# Can also be specified via environment variable `MAIL_COMMAND_TIMEOUT_SECS`.
|
||||
#
|
||||
# Default value: 30
|
||||
#command_timeout_secs = 30
|
||||
|
||||
# Total per-message timeout across all MX attempts in seconds.
|
||||
#
|
||||
# Can also be specified via environment variable `MAIL_TOTAL_TIMEOUT_SECS`.
|
||||
#
|
||||
# Default value: 60
|
||||
#total_timeout_secs = 60
|
||||
|
||||
# Max number of concurrent direct-MX sends. Limits the load placed
|
||||
# on any single recipient MX during a backlog drain.
|
||||
#
|
||||
# Can also be specified via environment variable `MAIL_MAX_CONCURRENT_SENDS`.
|
||||
#
|
||||
# Default value: 8
|
||||
#max_concurrent_sends = 8
|
||||
|
||||
# Require STARTTLS on every MX hop. When false, TLS is
|
||||
# attempted opportunistically and the session falls back to plaintext
|
||||
# if the remote does not advertise STARTTLS. Set true to refuse
|
||||
# plaintext delivery, at the cost of failing sends to MX hosts that
|
||||
# do not support TLS.
|
||||
#
|
||||
# Can also be specified via environment variable `MAIL_REQUIRE_TLS`.
|
||||
#
|
||||
# Default value: false
|
||||
#require_tls = false
|
||||
|
||||
[email.dkim]
|
||||
# DKIM selector. When unset, outgoing mail is not signed.
|
||||
#
|
||||
# Can also be specified via environment variable `MAIL_DKIM_SELECTOR`.
|
||||
#selector =
|
||||
|
||||
# DKIM signing domain.
|
||||
#
|
||||
# Can also be specified via environment variable `MAIL_DKIM_DOMAIN`.
|
||||
#domain =
|
||||
|
||||
# Path to the DKIM private key in PEM format. Supports RSA and
|
||||
# Ed25519 keys.
|
||||
#
|
||||
# Can also be specified via environment variable `MAIL_DKIM_KEY_PATH`.
|
||||
#private_key_path =
|
||||
|
||||
[discord]
|
||||
# Discord bot token. When unset, Discord integration is disabled.
|
||||
|
||||
Reference in New Issue
Block a user