Skip to content

Commit 26002a5

Browse files
serprexserprex
committed
refactor
1 parent 8db7154 commit 26002a5

17 files changed

Lines changed: 252 additions & 446 deletions

File tree

‎src/config/mod.rs‎

Lines changed: 114 additions & 62 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,31 @@ pub enum StorageSettings {
5959
Gcs(gcs::GcsConfig),
6060
}
6161

62+
impl Default for Settings {
63+
/// Convenience defaults: single-worker fs pipeline at zstd-3, no throttling
64+
/// or encryption. Production constructs via [`Settings::from_env`]; this
65+
/// lets tests vary only the fields they exercise via `..Default::default()`
66+
fn default() -> Self {
67+
Settings {
68+
storage: StorageSettings::Fs {
69+
path: String::new(),
70+
},
71+
compression: compression::Method::Zstd,
72+
compression_level: 3,
73+
upload_concurrency: 1,
74+
upload_queue: 1,
75+
download_concurrency: 1,
76+
prevent_wal_overwrite: false,
77+
use_wal_delta: false,
78+
retry: RetryPolicy::default(),
79+
network_rate_limit: 0,
80+
disk_rate_limit: 0,
81+
delta: DeltaSettings::default(),
82+
crypter: None,
83+
}
84+
}
85+
}
86+
6287
impl Settings {
6388
pub fn from_env() -> Result<Self> {
6489
let storage = detect_storage()?;
@@ -201,46 +226,12 @@ fn storage_from_uri(uri: &str, src: &StorageSettings) -> Result<StorageSettings>
201226
if let Some(rest) = uri.strip_prefix("s3://") {
202227
let (bucket, prefix) = split_bucket_prefix(rest);
203228
let s3_src = match src {
204-
StorageSettings::S3(c) => Some(c.clone()),
229+
StorageSettings::S3(c) => Some(c),
205230
_ => None,
206231
};
207-
let region = s3_src
208-
.as_ref()
209-
.map(|c| c.region.clone())
210-
.or_else(|| std::env::var("AWS_REGION").ok())
211-
.unwrap_or_else(|| "us-east-1".into());
212-
let access_key = s3_src
213-
.as_ref()
214-
.map(|c| c.access_key.clone())
215-
.or_else(|| std::env::var("AWS_ACCESS_KEY_ID").ok())
216-
.ok_or_else(|| anyhow!("AWS_ACCESS_KEY_ID not set"))?;
217-
let secret_key = s3_src
218-
.as_ref()
219-
.map(|c| c.secret_key.clone())
220-
.or_else(|| std::env::var("AWS_SECRET_ACCESS_KEY").ok())
221-
.ok_or_else(|| anyhow!("AWS_SECRET_ACCESS_KEY not set"))?;
222-
let session_token = s3_src
223-
.as_ref()
224-
.and_then(|c| c.session_token.clone())
225-
.or_else(|| std::env::var("AWS_SESSION_TOKEN").ok());
226-
let endpoint = s3_src
227-
.as_ref()
228-
.and_then(|c| c.endpoint.clone())
229-
.or_else(|| std::env::var("AWS_ENDPOINT_URL").ok());
230-
let force_path_style = s3_src
231-
.as_ref()
232-
.map(|c| c.force_path_style)
233-
.unwrap_or(endpoint.is_some());
234-
return Ok(StorageSettings::S3(s3::S3Config {
235-
bucket,
236-
prefix,
237-
region,
238-
access_key,
239-
secret_key,
240-
session_token,
241-
endpoint,
242-
force_path_style,
243-
}));
232+
return Ok(StorageSettings::S3(s3_config_from_env(
233+
bucket, prefix, s3_src,
234+
)?));
244235
}
245236
if let Some(rest) = uri.strip_prefix("gs://") {
246237
let (bucket, prefix) = split_bucket_prefix(rest);
@@ -268,6 +259,55 @@ fn split_bucket_prefix(rest: &str) -> (String, String) {
268259
}
269260
}
270261

262+
/// Resolve an `S3Config` for `bucket`/`prefix`, layering credential fields.
263+
/// `src` (an existing S3 source for `backup-copy`) takes priority; otherwise
264+
/// fall back to env honoring every wal-g alias so detection & destination
265+
/// resolution read the same names: AWS_REGION/WALG_S3_REGION,
266+
/// AWS_ACCESS_KEY_ID/AWS_ACCESS_KEY, AWS_SECRET_ACCESS_KEY/AWS_SECRET_KEY,
267+
/// AWS_SESSION_TOKEN, AWS_ENDPOINT_URL/WALG_S3_ENDPOINT, WALG_S3_FORCE_PATH_STYLE
268+
fn s3_config_from_env(
269+
bucket: String,
270+
prefix: String,
271+
src: Option<&s3::S3Config>,
272+
) -> Result<s3::S3Config> {
273+
let region = src
274+
.map(|c| c.region.clone())
275+
.or_else(|| std::env::var("AWS_REGION").ok())
276+
.or_else(|| std::env::var("WALG_S3_REGION").ok())
277+
.unwrap_or_else(|| "us-east-1".into());
278+
let access_key = src
279+
.map(|c| c.access_key.clone())
280+
.or_else(|| std::env::var("AWS_ACCESS_KEY_ID").ok())
281+
.or_else(|| std::env::var("AWS_ACCESS_KEY").ok())
282+
.ok_or_else(|| anyhow!("AWS_ACCESS_KEY_ID not set"))?;
283+
let secret_key = src
284+
.map(|c| c.secret_key.clone())
285+
.or_else(|| std::env::var("AWS_SECRET_ACCESS_KEY").ok())
286+
.or_else(|| std::env::var("AWS_SECRET_KEY").ok())
287+
.ok_or_else(|| anyhow!("AWS_SECRET_ACCESS_KEY not set"))?;
288+
let session_token = src
289+
.and_then(|c| c.session_token.clone())
290+
.or_else(|| std::env::var("AWS_SESSION_TOKEN").ok());
291+
let endpoint = src
292+
.and_then(|c| c.endpoint.clone())
293+
.or_else(|| std::env::var("AWS_ENDPOINT_URL").ok())
294+
.or_else(|| std::env::var("WALG_S3_ENDPOINT").ok());
295+
let force_path_style = match src {
296+
Some(c) => c.force_path_style,
297+
None => parse_env_bool("WALG_S3_FORCE_PATH_STYLE", endpoint.is_some())?,
298+
};
299+
Ok(s3::S3Config {
300+
bucket,
301+
prefix,
302+
region,
303+
access_key,
304+
secret_key,
305+
session_token,
306+
endpoint,
307+
force_path_style,
308+
})
309+
}
310+
271311
impl DeltaSettings {
272312
pub fn from_env() -> Result<Self> {
273313
let max_steps = parse_env_int("WALG_DELTA_MAX_STEPS", 0)?.max(0) as u32;
@@ -298,30 +338,9 @@ fn detect_storage() -> Result<StorageSettings> {
298338
}
299339
if let Ok(s3_prefix) = std::env::var("WALG_S3_PREFIX") {
300340
let (bucket, prefix) = parse_uri_prefix(&s3_prefix, "s3://")?;
301-
let region = std::env::var("AWS_REGION")
302-
.or_else(|_| std::env::var("WALG_S3_REGION"))
303-
.unwrap_or_else(|_| "us-east-1".into());
304-
let access_key = std::env::var("AWS_ACCESS_KEY_ID")
305-
.or_else(|_| std::env::var("AWS_ACCESS_KEY"))
306-
.map_err(|_| anyhow!("AWS_ACCESS_KEY_ID not set"))?;
307-
let secret_key = std::env::var("AWS_SECRET_ACCESS_KEY")
308-
.or_else(|_| std::env::var("AWS_SECRET_KEY"))
309-
.map_err(|_| anyhow!("AWS_SECRET_ACCESS_KEY not set"))?;
310-
let session_token = std::env::var("AWS_SESSION_TOKEN").ok();
311-
let endpoint = std::env::var("AWS_ENDPOINT_URL")
312-
.or_else(|_| std::env::var("WALG_S3_ENDPOINT"))
313-
.ok();
314-
let force_path_style = parse_env_bool("WALG_S3_FORCE_PATH_STYLE", endpoint.is_some())?;
315-
return Ok(StorageSettings::S3(s3::S3Config {
316-
bucket,
317-
prefix,
318-
region,
319-
access_key,
320-
secret_key,
321-
session_token,
322-
endpoint,
323-
force_path_style,
324-
}));
341+
return Ok(StorageSettings::S3(s3_config_from_env(
342+
bucket, prefix, None,
343+
)?));
325344
}
326345
if let Ok(gs_prefix) = std::env::var("WALG_GS_PREFIX") {
327346
let (bucket, prefix) = parse_uri_prefix(&gs_prefix, "gs://")?;
@@ -508,6 +527,39 @@ mod tests {
508527
assert!(parse_uri_prefix("s3:///prefix", "s3://").is_err());
509528
}
510529

530+
#[test]
531+
fn s3_dst_from_non_s3_src_honors_walg_aliases() {
532+
// file://->s3:// copy: no S3 source to inherit, so credential fields
533+
// come from env. Must read the same aliases as detect_storage, not just
534+
// the bare AWS_* names
535+
let vars = [
536+
("AWS_REGION", None),
537+
("WALG_S3_REGION", Some("eu-west-2")),
538+
("AWS_ACCESS_KEY_ID", None),
539+
("AWS_ACCESS_KEY", Some("AKIA_ALIAS")),
540+
("AWS_SECRET_ACCESS_KEY", None),
541+
("AWS_SECRET_KEY", Some("secret_alias")),
542+
("AWS_SESSION_TOKEN", None),
543+
("AWS_ENDPOINT_URL", None),
544+
("WALG_S3_ENDPOINT", Some("http://minio:9000")),
545+
("WALG_S3_FORCE_PATH_STYLE", Some("true")),
546+
];
547+
let _g = EnvGuard::new(&vars);
548+
let src = StorageSettings::Fs { path: "/x".into() };
549+
match storage_from_uri("s3://bkt/pre/fix", &src).unwrap() {
550+
StorageSettings::S3(c) => {
551+
assert_eq!(c.bucket, "bkt");
552+
assert_eq!(c.prefix, "pre/fix");
553+
assert_eq!(c.region, "eu-west-2");
554+
assert_eq!(c.access_key, "AKIA_ALIAS");
555+
assert_eq!(c.secret_key, "secret_alias");
556+
assert_eq!(c.endpoint.as_deref(), Some("http://minio:9000"));
557+
assert!(c.force_path_style);
558+
}
559+
other => panic!("expected S3, got {other:?}"),
560+
}
561+
}
562+
511563
#[test]
512564
fn parse_env_int_default_valid_and_malformed() {
513565
let key = "WALRS_TEST_PARSE_INT";

‎src/crypto/libsodium.rs‎

Lines changed: 19 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -163,6 +163,23 @@ pub fn from_env() -> Result<Option<DynCrypter>> {
163163
// append plaintext to `out`, drain into caller
164164
// - FINAL tag flips `finalized = true`; subsequent reads return EOF
165165

166+
/// Copy queued bytes from `out[*out_pos..]` into `buf`, resetting the buffer
167+
/// once fully drained. Returns true when there were bytes pending (caller then
168+
/// yields `Ready`), mirroring the secretstream readers' drain-first step
169+
fn drain_into(out: &mut Vec<u8>, out_pos: &mut usize, buf: &mut ReadBuf<'_>) -> bool {
170+
if *out_pos >= out.len() {
171+
return false;
172+
}
173+
let want = buf.remaining().min(out.len() - *out_pos);
174+
buf.put_slice(&out[*out_pos..*out_pos + want]);
175+
*out_pos += want;
176+
if *out_pos == out.len() {
177+
out.clear();
178+
*out_pos = 0;
179+
}
180+
true
181+
}
182+
166183
struct EncryptReader {
167184
inner: AsyncReader,
168185
stream: Option<DryocStream<Push>>,
@@ -242,14 +259,7 @@ impl AsyncRead for EncryptReader {
242259

243260
loop {
244261
// 1) Drain any ready ciphertext (or header bytes)
245-
if me.out_pos < me.out.len() {
246-
let want = buf.remaining().min(me.out.len() - me.out_pos);
247-
buf.put_slice(&me.out[me.out_pos..me.out_pos + want]);
248-
me.out_pos += want;
249-
if me.out_pos == me.out.len() {
250-
me.out.clear();
251-
me.out_pos = 0;
252-
}
262+
if drain_into(&mut me.out, &mut me.out_pos, buf) {
253263
return Poll::Ready(Ok(()));
254264
}
255265
if me.finalized {
@@ -350,14 +360,7 @@ impl AsyncRead for DecryptReader {
350360
let me = &mut *self;
351361
loop {
352362
// 1) Drain plaintext
353-
if me.out_pos < me.out.len() {
354-
let want = buf.remaining().min(me.out.len() - me.out_pos);
355-
buf.put_slice(&me.out[me.out_pos..me.out_pos + want]);
356-
me.out_pos += want;
357-
if me.out_pos == me.out.len() {
358-
me.out.clear();
359-
me.out_pos = 0;
360-
}
363+
if drain_into(&mut me.out, &mut me.out_pos, buf) {
361364
return Poll::Ready(Ok(()));
362365
}
363366
if me.finalized {

‎src/daemon/uploader.rs‎

Lines changed: 1 addition & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -277,17 +277,8 @@ mod tests {
277277
path: store.to_string_lossy().into(),
278278
},
279279
compression: crate::compression::Method::None,
280-
compression_level: 3,
281280
upload_concurrency: concurrency,
282-
upload_queue: 1,
283-
download_concurrency: 1,
284-
prevent_wal_overwrite: false,
285-
use_wal_delta: false,
286-
retry: crate::retry::RetryPolicy::default(),
287-
network_rate_limit: 0,
288-
disk_rate_limit: 0,
289-
delta: Default::default(),
290-
crypter: None,
281+
..Default::default()
291282
}
292283
}
293284

‎src/pg/backup/delta.rs‎

Lines changed: 3 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -700,16 +700,11 @@ mod tests {
700700
#[tokio::test]
701701
async fn delta_parent_carries_increment_format() {
702702
use crate::config::DeltaSettings;
703-
use crate::pg::backup::{
704-
BackupSentinelDto, METADATA_DATETIME_FORMAT, format_backup_name, sentinel_key,
705-
};
703+
use crate::pg::backup::{BackupSentinelDto, format_backup_name, sentinel_key};
706704
use crate::storage::fs::FsStorage;
707705
use std::sync::Arc;
708706

709707
let seg = DEFAULT_WAL_SEG_SIZE;
710-
let ts = chrono::DateTime::parse_from_rfc3339("2024-01-01T00:00:00Z")
711-
.unwrap()
712-
.with_timezone(&chrono::Utc);
713708

714709
let sentinel = |from: Option<&str>, fmt: increment::Format| BackupSentinelDtoV2 {
715710
sentinel: BackupSentinelDto {
@@ -721,23 +716,11 @@ mod tests {
721716
increment_format: fmt,
722717
pg_version: 170000,
723718
backup_finish_lsn: Some(seg + 1),
724-
system_identifier: None,
725-
uncompressed_size: 0,
726-
compressed_size: 0,
727-
data_catalog_size: 0,
728-
user_data: None,
729-
files_metadata_disabled: true,
730-
tablespace_spec: None,
731-
backup_start_chkp_num: Some(0),
732-
increment_from_chkp_num: None,
719+
..Default::default()
733720
},
734-
version: 2,
735-
start_time: ts,
736-
finish_time: ts,
737-
date_fmt: METADATA_DATETIME_FORMAT.into(),
738721
hostname: "h".into(),
739722
data_dir: "/d".into(),
740-
is_permanent: false,
723+
..Default::default()
741724
};
742725

743726
// Parent is itself a delta → its format constrains the new push

0 commit comments

Comments
 (0)