//! S3 backend: AWS or any S3-compatible store (Linode Object Storage,
//! MinIO, Backblaze B2, DigitalOcean Spaces, ...).
//!
//! URL form: `s3://bucket/prefix?region=...&endpoint=...&style=...`
//! where every parameter is optional:
//!
//! - `region`: falls back to `$AWS_REGION`. For non-AWS stores this is
//! whatever region string the provider signs with.
//! - `endpoint`: falls back to `$AWS_ENDPOINT_URL`; omitted entirely
//! means AWS proper. `https://` is assumed if no scheme is given.
//! - `style`: `path` or `vhost` addressing. Defaults to `vhost` for
//! AWS and `path` for custom endpoints, which is the combination
//! that works nearly everywhere; providers that require
//! virtual-hosted buckets on custom endpoints can say `style=vhost`.
//!
//! Credentials come from the standard AWS sources (environment
//! variables, profile files, instance metadata) — the lingua franca
//! that S3-compatible providers document as well.
//!
//! Examples:
//!
//! ```text
//! s3://backups/laptop?region=eu-central-1
//! s3://backups/laptop?endpoint=eu-central-1.linodeobjects.com®ion=eu-central-1
//! s3://backups/laptop?endpoint=http://localhost:9000®ion=minio
//! ```
//!
//! No connection pool is needed: each operation is an independent HTTP
//! request and the client is safe to share across threads. S3 `put` is
//! atomic and last-writer-wins by nature; since objects are write-once
//! and content-addressed, overwriting is idempotent and no existence
//! pre-check or rename dance is required.
use repository::{Backend, ChunkId, Error, ObjectKey, ObjectKind};
use s3::{Bucket, Region, creds::Credentials, error::S3Error};
use url::Url;
/// How the bucket name is carried in requests: as a URL path segment or
/// as part of the host name.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum AddressingStyle {
Path,
VirtualHost,
}
pub struct S3Backend {
bucket: Box<Bucket>,
prefix: String,
}
impl S3Backend {
pub fn from_url(url: &Url) -> Result<S3Backend, Error> {
let name = url
.host_str()
.ok_or_else(|| Error::Backend(format!("{url} has no bucket name")))?
.to_string();
let mut region = std::env::var("AWS_REGION").ok();
let mut endpoint = std::env::var("AWS_ENDPOINT_URL").ok();
let mut style = None;
for (key, value) in url.query_pairs() {
match key.as_ref() {
"region" => region = Some(value.into_owned()),
"endpoint" => endpoint = Some(value.into_owned()),
"style" => match value.as_ref() {
"path" => style = Some(AddressingStyle::Path),
"vhost" => style = Some(AddressingStyle::VirtualHost),
other => {
return Err(Error::Backend(format!(
"s3: style must be \"path\" or \"vhost\", not {other:?}"
)));
}
},
other => {
return Err(Error::Backend(format!(
"s3: unknown URL parameter {other:?}"
)));
}
}
}
let custom_endpoint = endpoint.is_some();
let region = match endpoint {
Some(endpoint) => Region::Custom {
region: region.unwrap_or_else(|| "us-east-1".to_string()),
endpoint: ensure_scheme(endpoint),
},
None => region
.ok_or_else(|| {
Error::Backend(
"s3: no region: set $AWS_REGION or add ?region= to the URL".to_string(),
)
})?
.parse()
.map_err(|err| Error::Backend(format!("s3: bad region: {err}")))?,
};
// Path-style is the near-universal default for S3-compatible
// stores; AWS itself prefers virtual-hosted buckets
let style = style.unwrap_or(if custom_endpoint {
AddressingStyle::Path
} else {
AddressingStyle::VirtualHost
});
let credentials = Credentials::default()
.map_err(|err| Error::Backend(format!("s3: credentials: {err}")))?;
let mut bucket = Bucket::new(&name, region, credentials).map_err(s3_error)?;
if style == AddressingStyle::Path {
bucket = bucket.with_path_style();
}
Ok(S3Backend {
bucket,
prefix: url.path().trim_matches('/').to_string(),
})
}
fn object_path(&self, key: &ObjectKey) -> String {
let mut path = self.prefix.clone();
for segment in key.segments() {
if !path.is_empty() {
path.push('/');
}
path.push_str(&segment);
}
path
}
fn prefix_path(&self, segments: &[&str]) -> String {
let mut path = self.prefix.clone();
for segment in segments {
if !path.is_empty() {
path.push('/');
}
path.push_str(segment);
}
path
}
}
impl Backend for S3Backend {
fn put(&self, key: &ObjectKey, data: &[u8]) -> Result<(), Error> {
let response = self
.bucket
.put_object(self.object_path(key), data)
.map_err(s3_error)?;
if (200..300).contains(&response.status_code()) {
Ok(())
} else {
Err(Error::Backend(format!(
"s3: put returned status {}",
response.status_code()
)))
}
}
fn get(&self, key: &ObjectKey) -> Result<Option<Vec<u8>>, Error> {
match self.bucket.get_object(self.object_path(key)) {
Ok(response) if response.status_code() == 404 => Ok(None),
Ok(response) if (200..300).contains(&response.status_code()) => {
Ok(Some(response.into_bytes().to_vec()))
}
Ok(response) => Err(Error::Backend(format!(
"s3: get returned status {}",
response.status_code()
))),
Err(err) if is_not_found(&err) => Ok(None),
Err(err) => Err(s3_error(err)),
}
}
fn contains(&self, key: &ObjectKey) -> Result<bool, Error> {
match self.bucket.head_object(self.object_path(key)) {
Ok((_, 404)) => Ok(false),
Ok((_, code)) if (200..300).contains(&code) => Ok(true),
Ok((_, code)) => Err(Error::Backend(format!("s3: head returned status {code}"))),
Err(err) if is_not_found(&err) => Ok(false),
Err(err) => Err(s3_error(err)),
}
}
fn list(
&self,
kind: ObjectKind,
visit: &mut dyn FnMut(ObjectKey) -> Result<(), Error>,
) -> Result<(), Error> {
match kind {
ObjectKind::Header => {
if self.contains(&ObjectKey::Header)? {
visit(ObjectKey::Header)?;
}
}
// The flat kinds: one prefix of hex-named objects each
ObjectKind::Snapshot | ObjectKind::Prune | ObjectKind::Lock => {
self.visit_names(&self.prefix_path(&[kind.directory()]), &mut |name| {
if let Some(key) = kind.key_for(name) {
return visit(key);
}
Ok(())
})?;
}
ObjectKind::Chunk => {
self.visit_names(&self.prefix_path(&["chunks"]), &mut |name| {
if let Ok(id) = ChunkId::from_hex(name) {
return visit(ObjectKey::Chunk(id));
}
Ok(())
})?;
}
}
Ok(())
}
/// Adds up every object under the repository prefix, a thousand
/// keys to a request.
///
/// An object store has no size of its own to speak of, but a
/// backup told to keep its repository under a certain size is
/// asking about the repository, and that is answerable.
fn used_space(&self) -> Result<Option<u64>, Error> {
// A repository at the bucket root has no prefix to append a
// separator to, and "/" would match nothing
let prefix = match self.prefix.is_empty() {
true => String::new(),
false => format!("{}/", self.prefix),
};
let pages = self.bucket.list(prefix, None).map_err(s3_error)?;
let total = pages
.iter()
.flat_map(|page| page.contents.iter())
.map(|object| object.size)
.sum();
Ok(Some(total))
}
fn delete(&self, key: &ObjectKey) -> Result<(), Error> {
match self.bucket.delete_object(self.object_path(key)) {
Ok(_) => Ok(()),
Err(err) if is_not_found(&err) => Ok(()),
Err(err) => Err(s3_error(err)),
}
}
}
impl S3Backend {
/// Visits the final name component of every object under `prefix`.
/// S3 listing is flat, so chunk fan-out directories need no special
/// handling here.
fn visit_names(
&self,
prefix: &str,
visit: &mut dyn FnMut(&str) -> Result<(), Error>,
) -> Result<(), Error> {
let pages = self
.bucket
.list(format!("{prefix}/"), None)
.map_err(s3_error)?;
for page in pages {
for object in page.contents {
let name = object
.key
.rsplit('/')
.next()
.expect("rsplit always yields at least one part");
visit(name)?;
}
}
Ok(())
}
}
/// Endpoints are commonly written without a scheme ("HTTPS is implied");
/// the client library wants a full URL.
fn ensure_scheme(endpoint: String) -> String {
if endpoint.contains("://") {
endpoint
} else {
format!("https://{endpoint}")
}
}
fn is_not_found(err: &S3Error) -> bool {
matches!(err, S3Error::HttpFailWithBody(404, _))
}
fn s3_error(err: S3Error) -> Error {
Error::Backend(format!("s3: {err}"))
}