OXYGENHUB
# 2ddd471651cd oxygen / crates / oxy-strata / src / lib.rs
28.9 KiB · Rust · last changed Strata: one client per cloud store, its requests side by side, and counts 19 days ago source blame history raw
1 
//! Oxygen Strata: repositories kept on cloud object storage (docs/stores.md).
2 
//!
3 
//! This crate gives `oxy` the stores that need an HTTP client and the
4 
//! providers' signing: Amazon S3 (`s3://BUCKET/PREFIX`), Google Cloud
5 
//! Storage (`gs://BUCKET/PREFIX`) and Azure Blob Storage
6 
//! (`az://ACCOUNT/CONTAINER/PREFIX`), each as a [`Store`] over the
7 
//! `object_store` crate. Credentials come from each provider's standard
8 
//! places, with nothing oxy-specific to set up; a URL may carry settings
9 
//! as query parameters (`?profile=work&region=eu-west-2`), which is where
10 
//! `oxy remote add --profile` keeps them.
11 
//!
12 
//! [`register`] hands these stores to oxy-core's remote opener; a program
13 
//! that links this crate calls it once at start.
14
15 
mod aws;
16 
mod azure;
17 
mod gcp;
18 
mod http;
19 
mod ini;
20
21 
use futures::TryStreamExt;
22 
use object_store::path::Path as ObjectPath;
23 
use object_store::{
24 
    GetOptions, ObjectStore, ObjectStoreExt, PutMode, PutOptions, PutPayload, UpdateVersion,
25 
};
26 
use oxy_core::strata::{check_key, Entry, Error, Result, Store, Version};
27 
use std::collections::BTreeMap;
28 
use std::future::Future;
29 
use std::ops::Range;
30 
use std::sync::{Arc, Mutex};
31
32 
/// Make the cloud stores available to `oxy_core::protocol::open_remote`.
33 
/// Safe to call more than once.
34 
pub fn register() {
35 
    // object_store's reqwest is built with rustls and no bundled crypto
36 
    // provider; oxy links ring already, so that is the one to install.
37 
    let _ = rustls::crypto::ring::default_provider().install_default();
38 
    oxy_core::protocol::register_store_opener(open_url);
39 
}
40
41 
fn open_url(
42 
    url: &str,
43 
    _options: &oxy_core::protocol::RemoteOptions,
44 
) -> Option<Result<Box<dyn Store>>> {
45 
    let scheme = url.split_once("://")?.0;
46 
    if !matches!(scheme, "s3" | "gs" | "az") {
47 
        return None;
48 
    }
49 
    let lookup = *CREDENTIALS.lock().unwrap_or_else(|p| p.into_inner());
50 
    let given = lookup.and_then(|f| f(url));
51 
    Some(shared(url, given).map(|s| Box::new(Shared(s)) as Box<dyn Store>))
52 
}
53
54 
/// One client per store and credentials, kept for the life of the process
55 
/// (oxygenhub#234): a server opens a repository for every request, and a
56 
/// new client each time meant a new runtime, new connections and signing
57 
/// in to the provider again before the first byte.
58 
type OpenStores = std::collections::HashMap<(String, Option<Credentials>), Arc<CloudStore>>;
59
60 
fn shared(url: &str, given: Option<Credentials>) -> Result<Arc<CloudStore>> {
61 
    static OPEN: Mutex<Option<OpenStores>> = Mutex::new(None);
62 
    let key = (url.to_string(), given);
63 
    let mut open = OPEN.lock().unwrap_or_else(|p| p.into_inner());
64 
    let map = open.get_or_insert_with(Default::default);
65 
    if let Some(s) = map.get(&key) {
66 
        return Ok(s.clone());
67 
    }
68 
    let store = Arc::new(CloudStore::open_with(url, key.1.as_ref())?);
69 
    map.insert(key, store.clone());
70 
    Ok(store)
71 
}
72
73 
/// A shared [`CloudStore`] as a [`Store`].
74 
struct Shared(Arc<CloudStore>);
75
76 
impl Store for Shared {
77 
    fn describe(&self) -> String {
78 
        self.0.describe()
79 
    }
80 
    fn has(&self, key: &str) -> Result<bool> {
81 
        self.0.has(key)
82 
    }
83 
    fn get(&self, key: &str) -> Result<Vec<u8>> {
84 
        self.0.get(key)
85 
    }
86 
    fn get_range(&self, key: &str, range: Range<u64>) -> Result<Vec<u8>> {
87 
        self.0.get_range(key, range)
88 
    }
89 
    fn put_new(&self, key: &str, bytes: &[u8]) -> Result<()> {
90 
        self.0.put_new(key, bytes)
91 
    }
92 
    fn list(&self, prefix: &str) -> Result<Vec<Entry>> {
93 
        self.0.list(prefix)
94 
    }
95 
    fn delete(&self, key: &str) -> Result<()> {
96 
        self.0.delete(key)
97 
    }
98 
    fn read_ref(&self, key: &str) -> Result<Option<(Vec<u8>, Version)>> {
99 
        self.0.read_ref(key)
100 
    }
101 
    fn write_ref(&self, key: &str, bytes: &[u8], expected: Option<&Version>) -> Result<Version> {
102 
        self.0.write_ref(key, bytes, expected)
103 
    }
104 
}
105
106 
static OPENS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
107 
static CALLS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
108 
static MICROS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
109
110 
/// What the cloud stores have done since the process started: clients
111 
/// opened, requests made and the time spent in them, for a server to show
112 
/// (a hub's /health).
113 
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
114 
pub struct Stats {
115 
    pub opens: u64,
116 
    pub calls: u64,
117 
    pub millis: u64,
118 
}
119
120 
pub fn stats() -> Stats {
121 
    use std::sync::atomic::Ordering::Relaxed;
122 
    Stats {
123 
        opens: OPENS.load(Relaxed),
124 
        calls: CALLS.load(Relaxed),
125 
        millis: MICROS.load(Relaxed) / 1000,
126 
    }
127 
}
128
129 
/// Credentials a program keeps itself and hands over, instead of the
130 
/// provider's usual places: a hub keeps each repository's sealed. When a
131 
/// store is opened with them, nothing from the environment or the
132 
/// provider's files is used.
133 
#[derive(Clone, PartialEq, Eq, Hash)]
134 
pub enum Credentials {
135 
    /// An S3 access key.
136 
    Aws {
137 
        access_key_id: String,
138 
        secret_access_key: String,
139 
        session_token: Option<String>,
140 
    },
141 
    /// A Google Cloud service account key, as the JSON Google issues.
142 
    Gcp { service_account_key: String },
143 
    /// An Azure service principal.
144 
    Azure {
145 
        tenant_id: String,
146 
        client_id: String,
147 
        client_secret: String,
148 
    },
149 
}
150
151 
impl std::fmt::Debug for Credentials {
152 
    /// Names the kind and the public part only: never the secret.
153 
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
154 
        match self {
155 
            Credentials::Aws { access_key_id, .. } => {
156 
                write!(f, "Aws({access_key_id}, secret hidden)")
157 
            }
158 
            Credentials::Gcp { .. } => write!(f, "Gcp(service account key hidden)"),
159 
            Credentials::Azure { client_id, .. } => write!(f, "Azure({client_id}, secret hidden)"),
160 
        }
161 
    }
162 
}
163
164 
impl Credentials {
165 
    /// The scheme of the stores these are for.
166 
    pub fn scheme(&self) -> &'static str {
167 
        match self {
168 
            Credentials::Aws { .. } => "s3",
169 
            Credentials::Gcp { .. } => "gs",
170 
            Credentials::Azure { .. } => "az",
171 
        }
172 
    }
173
174 
    /// As JSON, for keeping (sealed) and reading back with [`Credentials::from_json`].
175 
    pub fn to_json(&self) -> String {
176 
        let v = match self {
177 
            Credentials::Aws {
178 
                access_key_id,
179 
                secret_access_key,
180 
                session_token,
181 
            } => serde_json::json!({
182 
                "kind": "aws", "access_key_id": access_key_id, "secret_access_key": secret_access_key,
183 
                "session_token": session_token,
184 
            }),
185 
            Credentials::Gcp {
186 
                service_account_key,
187 
            } => {
188 
                serde_json::json!({ "kind": "gcp", "service_account_key": service_account_key })
189 
            }
190 
            Credentials::Azure {
191 
                tenant_id,
192 
                client_id,
193 
                client_secret,
194 
            } => serde_json::json!({
195 
                "kind": "azure", "tenant_id": tenant_id, "client_id": client_id, "client_secret": client_secret,
196 
            }),
197 
        };
198 
        v.to_string()
199 
    }
200
201 
    pub fn from_json(text: &str) -> Result<Credentials> {
202 
        let v: serde_json::Value = serde_json::from_str(text)
203 
            .map_err(|e| Error::Backend(format!("credentials are not JSON: {e}")))?;
204 
        let field = |k: &str| -> Result<String> {
205 
            v.get(k)
206 
                .and_then(|x| x.as_str())
207 
                .filter(|x| !x.trim().is_empty())
208 
                .map(|x| x.trim().to_string())
209 
                .ok_or_else(|| Error::Backend(format!("credentials need {k}")))
210 
        };
211 
        match v.get("kind").and_then(|k| k.as_str()) {
212 
            Some("aws") => Ok(Credentials::Aws {
213 
                access_key_id: field("access_key_id")?,
214 
                secret_access_key: field("secret_access_key")?,
215 
                session_token: field("session_token").ok(),
216 
            }),
217 
            Some("gcp") => {
218 
                let key = field("service_account_key")?;
219 
                let parsed: serde_json::Value = serde_json::from_str(&key).map_err(|_| {
220 
                    Error::Backend("the service account key is not the JSON Google issues".into())
221 
                })?;
222 
                if parsed.get("type").and_then(|t| t.as_str()) != Some("service_account") {
223 
                    return Err(Error::Backend(
224 
                        "the key is not a service account key (its type is not service_account)"
225 
                            .into(),
226 
                    ));
227 
                }
228 
                Ok(Credentials::Gcp {
229 
                    service_account_key: key,
230 
                })
231 
            }
232 
            Some("azure") => Ok(Credentials::Azure {
233 
                tenant_id: field("tenant_id")?,
234 
                client_id: field("client_id")?,
235 
                client_secret: field("client_secret")?,
236 
            }),
237 
            _ => Err(Error::Backend(
238 
                "credentials need a kind: aws, gcp or azure".into(),
239 
            )),
240 
        }
241 
    }
242 
}
243
244 
/// Finds the credentials a program keeps for a store URL.
245 
pub type CredentialsLookup = fn(&str) -> Option<Credentials>;
246
247 
static CREDENTIALS: Mutex<Option<CredentialsLookup>> = Mutex::new(None);
248
249 
/// Have stores opened by URL (as a repository's `storage.url` is) use the
250 
/// credentials `lookup` finds for it, where it finds any.
251 
pub fn set_credentials_lookup(lookup: CredentialsLookup) {
252 
    *CREDENTIALS.lock().unwrap_or_else(|p| p.into_inner()) = Some(lookup);
253 
}
254
255 
/// A parsed store URL: `SCHEME://HOST/PATH?PARAMS`.
256 
#[derive(Clone, Debug, PartialEq, Eq)]
257 
pub struct StoreUrl {
258 
    pub scheme: String,
259 
    /// The bucket (S3, GCS) or the storage account (Azure).
260 
    pub host: String,
261 
    /// The path under the host, without leading or trailing slashes; for
262 
    /// Azure the container is its first segment.
263 
    pub path: String,
264 
    pub params: BTreeMap<String, String>,
265 
}
266
267 
impl StoreUrl {
268 
    pub fn parse(url: &str) -> Result<StoreUrl> {
269 
        let bad = |why: &str| Error::Backend(format!("{url}: {why}"));
270 
        let (scheme, rest) = url.split_once("://").ok_or_else(|| bad("not a URL"))?;
271 
        let (rest, query) = match rest.split_once('?') {
272 
            Some((r, q)) => (r, Some(q)),
273 
            None => (rest, None),
274 
        };
275 
        let (host, path) = match rest.split_once('/') {
276 
            Some((h, p)) => (h, p.trim_matches('/')),
277 
            None => (rest, ""),
278 
        };
279 
        if host.is_empty() {
280 
            return Err(bad("names no bucket"));
281 
        }
282 
        let mut params = BTreeMap::new();
283 
        for pair in query.unwrap_or("").split('&').filter(|p| !p.is_empty()) {
284 
            let (k, v) = pair.split_once('=').unwrap_or((pair, ""));
285 
            params.insert(k.to_string(), percent_decode(v));
286 
        }
287 
        Ok(StoreUrl {
288 
            scheme: scheme.to_string(),
289 
            host: host.to_string(),
290 
            path: path.to_string(),
291 
            params,
292 
        })
293 
    }
294
295 
    /// The URL without its parameters, for messages.
296 
    pub fn plain(&self) -> String {
297 
        if self.path.is_empty() {
298 
            format!("{}://{}", self.scheme, self.host)
299 
        } else {
300 
            format!("{}://{}/{}", self.scheme, self.host, self.path)
301 
        }
302 
    }
303
304 
    pub fn param(&self, name: &str) -> Option<&str> {
305 
        self.params.get(name).map(String::as_str)
306 
    }
307 
}
308
309 
fn percent_decode(s: &str) -> String {
310 
    let bytes = s.as_bytes();
311 
    let mut out = Vec::with_capacity(bytes.len());
312 
    let mut i = 0;
313 
    while i < bytes.len() {
314 
        if bytes[i] == b'%' && i + 2 < bytes.len() {
315 
            if let Ok(v) = u8::from_str_radix(&s[i + 1..i + 3], 16) {
316 
                out.push(v);
317 
                i += 3;
318 
                continue;
319 
            }
320 
        }
321 
        out.push(bytes[i]);
322 
        i += 1;
323 
    }
324 
    String::from_utf8_lossy(&out).into_owned()
325 
}
326
327 
/// A store on a cloud provider: the provider's client, the prefix every
328 
/// key goes under, and a runtime to run the client on.
329 
pub struct CloudStore {
330 
    inner: Arc<dyn ObjectStore>,
331 
    prefix: String,
332 
    name: String,
333 
    hint: &'static str,
334 
    /// Always there until the store is dropped: two workers, so the
335 
    /// requests of a shared store run side by side.
336 
    runtime: Option<tokio::runtime::Runtime>,
337 
}
338
339 
impl Drop for CloudStore {
340 
    /// Without waiting, so a store can be dropped inside a runtime too.
341 
    fn drop(&mut self) {
342 
        if let Some(rt) = self.runtime.take() {
343 
            rt.shutdown_background();
344 
        }
345 
    }
346 
}
347
348 
impl CloudStore {
349 
    /// Open the store a URL names. Nothing is sent yet: credentials are
350 
    /// resolved here, but the bucket is first spoken to on the first call.
351 
    pub fn open(url: &str) -> Result<CloudStore> {
352 
        Self::open_with(url, None)
353 
    }
354
355 
    /// [`CloudStore::open`] with credentials given, which must be for
356 
    /// the URL's kind of store; then only they are used.
357 
    pub fn open_with(url: &str, given: Option<&Credentials>) -> Result<CloudStore> {
358 
        let parsed = StoreUrl::parse(url)?;
359 
        if let Some(c) = given.filter(|c| c.scheme() != parsed.scheme) {
360 
            return Err(Error::Backend(format!(
361 
                "{}: the credentials given are for {}:// stores",
362 
                parsed.plain(),
363 
                c.scheme()
364 
            )));
365 
        }
366 
        let aws_keys = match given {
367 
            Some(Credentials::Aws {
368 
                access_key_id,
369 
                secret_access_key,
370 
                session_token,
371 
            }) => Some((
372 
                access_key_id.as_str(),
373 
                secret_access_key.as_str(),
374 
                session_token.as_deref(),
375 
            )),
376 
            _ => None,
377 
        };
378 
        let gcp_key = match given {
379 
            Some(Credentials::Gcp {
380 
                service_account_key,
381 
            }) => Some(service_account_key.as_str()),
382 
            _ => None,
383 
        };
384 
        let principal = match given {
385 
            Some(Credentials::Azure {
386 
                tenant_id,
387 
                client_id,
388 
                client_secret,
389 
            }) => Some((
390 
                tenant_id.as_str(),
391 
                client_id.as_str(),
392 
                client_secret.as_str(),
393 
            )),
394 
            _ => None,
395 
        };
396 
        OPENS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
397 
        let runtime = tokio::runtime::Builder::new_multi_thread()
398 
            .worker_threads(2)
399 
            .thread_name("oxy-strata")
400 
            .enable_all()
401 
            .build()
402 
            .map_err(|e| Error::Backend(format!("starting the HTTP runtime: {e}")))?;
403 
        let (inner, prefix, hint): (Arc<dyn ObjectStore>, String, &'static str) =
404 
            match parsed.scheme.as_str() {
405 
                "s3" => (
406 
                    Arc::new(aws::open(&parsed, aws_keys)?),
407 
                    parsed.path.clone(),
408 
                    aws::HINT,
409 
                ),
410 
                "gs" => (
411 
                    Arc::new(gcp::open(&parsed, gcp_key)?),
412 
                    parsed.path.clone(),
413 
                    gcp::HINT,
414 
                ),
415 
                "az" => {
416 
                    let (store, prefix) = azure::open(&parsed, principal)?;
417 
                    (Arc::new(store), prefix, azure::HINT)
418 
                }
419 
                other => {
420 
                    return Err(Error::Backend(format!(
421 
                        "{url}: {other}:// is not a store oxy knows (s3://, gs://, az://, dir://)"
422 
                    )))
423 
                }
424 
            };
425 
        Ok(CloudStore {
426 
            inner,
427 
            prefix,
428 
            name: parsed.plain(),
429 
            hint,
430 
            runtime: Some(runtime),
431 
        })
432 
    }
433
434 
    /// Run a request to completion. Called from a thread that is itself
435 
    /// driving async tasks (a server's handler), where blocking on another
436 
    /// runtime is not allowed, the request runs on a thread of its own.
437 
    fn block<F>(&self, f: F) -> F::Output
438 
    where
439 
        F: Future + Send,
440 
        F::Output: Send,
441 
    {
442 
        let runtime = self
443 
            .runtime
444 
            .as_ref()
445 
            .expect("the runtime lives as long as the store");
446 
        let started = std::time::Instant::now();
447 
        let out = if tokio::runtime::Handle::try_current().is_ok() {
448 
            std::thread::scope(|s| {
449 
                s.spawn(|| runtime.block_on(f))
450 
                    .join()
451 
                    .unwrap_or_else(|p| std::panic::resume_unwind(p))
452 
            })
453 
        } else {
454 
            runtime.block_on(f)
455 
        };
456 
        CALLS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
457 
        MICROS.fetch_add(
458 
            started.elapsed().as_micros() as u64,
459 
            std::sync::atomic::Ordering::Relaxed,
460 
        );
461 
        out
462 
    }
463
464 
    fn path(&self, key: &str) -> Result<ObjectPath> {
465 
        check_key(key)?;
466 
        let full = if self.prefix.is_empty() {
467 
            key.to_string()
468 
        } else {
469 
            format!("{}/{key}", self.prefix)
470 
        };
471 
        ObjectPath::parse(&full).map_err(|e| Error::Backend(format!("{full}: {e}")))
472 
    }
473
474 
    /// Our error for the provider's, with what to do next where it helps.
475 
    fn map(&self, e: object_store::Error, key: &str) -> Error {
476 
        use object_store::Error as E;
477 
        match e {
478 
            E::NotFound { .. } => Error::NotFound(key.to_string()),
479 
            E::AlreadyExists { .. } => Error::Exists(key.to_string()),
480 
            E::Precondition { .. } | E::NotModified { .. } => Error::Conflict(key.to_string()),
481 
            other => {
482 
                let text = other.to_string();
483 
                let lower = text.to_ascii_lowercase();
484 
                let advice = if lower.contains("403")
485 
                    || lower.contains("accessdenied")
486 
                    || lower.contains("forbidden")
487 
                    || lower.contains("authorizationpermissionmismatch")
488 
                {
489 
                    format!(
490 
                        ": access denied; the credentials in use may not do this on {}: {}",
491 
                        self.name, self.hint
492 
                    )
493 
                } else if lower.contains("credential") || lower.contains("no valid credentials") {
494 
                    format!(": no usable credentials: {}", self.hint)
495 
                } else if lower.contains("nosuchbucket")
496 
                    || lower.contains("bucketnotfound")
497 
                    || lower.contains("containernotfound")
498 
                {
499 
                    format!(
500 
                        ": {} does not exist, or the credentials cannot see it",
501 
                        self.name
502 
                    )
503 
                } else if lower.contains("dns")
504 
                    || lower.contains("connect")
505 
                    || lower.contains("timed out")
506 
                {
507 
                    ": cannot reach the service; check the network, the endpoint and the region"
508 
                        .to_string()
509 
                } else {
510 
                    String::new()
511 
                };
512 
                Error::Backend(format!("{}: {text}{advice}", self.name))
513 
            }
514 
        }
515 
    }
516
517 
    fn version_of(e_tag: Option<String>, version: Option<String>) -> Version {
518 
        Version(format!(
519 
            "{}\t{}",
520 
            e_tag.unwrap_or_default(),
521 
            version.unwrap_or_default()
522 
        ))
523 
    }
524
525 
    fn update_version(v: &Version) -> UpdateVersion {
526 
        let (e_tag, version) = v.0.split_once('\t').unwrap_or((v.0.as_str(), ""));
527 
        UpdateVersion {
528 
            e_tag: (!e_tag.is_empty()).then(|| e_tag.to_string()),
529 
            version: (!version.is_empty()).then(|| version.to_string()),
530 
        }
531 
    }
532 
}
533
534 
impl Store for CloudStore {
535 
    fn describe(&self) -> String {
536 
        self.name.clone()
537 
    }
538
539 
    fn has(&self, key: &str) -> Result<bool> {
540 
        let path = self.path(key)?;
541 
        match self.block(self.inner.head(&path)) {
542 
            Ok(_) => Ok(true),
543 
            Err(object_store::Error::NotFound { .. }) => Ok(false),
544 
            Err(e) => Err(self.map(e, key)),
545 
        }
546 
    }
547
548 
    fn get(&self, key: &str) -> Result<Vec<u8>> {
549 
        let path = self.path(key)?;
550 
        self.block(async {
551 
            let r = self.inner.get(&path).await?;
552 
            r.bytes().await
553 
        })
554 
        .map(|b| b.to_vec())
555 
        .map_err(|e| self.map(e, key))
556 
    }
557
558 
    fn get_range(&self, key: &str, range: Range<u64>) -> Result<Vec<u8>> {
559 
        let path = self.path(key)?;
560 
        if range.start > range.end {
561 
            return Err(Error::Backend(format!(
562 
                "{key}: bytes {}..{} is not a range",
563 
                range.start, range.end
564 
            )));
565 
        }
566 
        if range.is_empty() {
567 
            return Ok(Vec::new());
568 
        }
569 
        let want = (range.end - range.start) as usize;
570 
        let bytes = self
571 
            .block(self.inner.get_range(&path, range.clone()))
572 
            .map_err(|e| self.map(e, key))?;
573 
        if bytes.len() != want {
574 
            return Err(Error::Backend(format!(
575 
                "{key}: bytes {}..{} asked, {} returned: the object is shorter",
576 
                range.start,
577 
                range.end,
578 
                bytes.len()
579 
            )));
580 
        }
581 
        Ok(bytes.to_vec())
582 
    }
583
584 
    fn put_new(&self, key: &str, bytes: &[u8]) -> Result<()> {
585 
        let path = self.path(key)?;
586 
        let opts = PutOptions {
587 
            mode: PutMode::Create,
588 
            ..Default::default()
589 
        };
590 
        self.block(
591 
            self.inner
592 
                .put_opts(&path, PutPayload::from(bytes.to_vec()), opts),
593 
        )
594 
        .map(|_| ())
595 
        .map_err(|e| self.map(e, key))
596 
    }
597
598 
    fn list(&self, prefix: &str) -> Result<Vec<Entry>> {
599 
        // Providers list by whole path segments; the asked prefix may end
600 
        // mid-segment, so list its directory and filter here.
601 
        let dir = prefix.rsplit_once('/').map(|(d, _)| d).unwrap_or("");
602 
        let full_dir = match (self.prefix.as_str(), dir) {
603 
            ("", "") => None,
604 
            ("", d) => Some(d.to_string()),
605 
            (p, "") => Some(p.to_string()),
606 
            (p, d) => Some(format!("{p}/{d}")),
607 
        };
608 
        let full_dir = full_dir
609 
            .map(|d| ObjectPath::parse(&d).map_err(|e| Error::Backend(format!("{d}: {e}"))))
610 
            .transpose()?;
611 
        let metas: Vec<object_store::ObjectMeta> = self
612 
            .block(self.inner.list(full_dir.as_ref()).try_collect())
613 
            .map_err(|e| self.map(e, prefix))?;
614 
        let strip = if self.prefix.is_empty() {
615 
            String::new()
616 
        } else {
617 
            format!("{}/", self.prefix)
618 
        };
619 
        let mut out: Vec<Entry> = metas
620 
            .into_iter()
621 
            .filter_map(|m| {
622 
                let loc = m.location.to_string();
623 
                let key = loc.strip_prefix(&strip)?;
624 
                key.starts_with(prefix).then(|| Entry {
625 
                    key: key.to_string(),
626 
                    size: m.size,
627 
                    modified_ms: u64::try_from(m.last_modified.timestamp_millis()).ok(),
628 
                })
629 
            })
630 
            .collect();
631 
        out.sort_by(|a, b| a.key.cmp(&b.key));
632 
        Ok(out)
633 
    }
634
635 
    fn delete(&self, key: &str) -> Result<()> {
636 
        let path = self.path(key)?;
637 
        match self.block(self.inner.delete(&path)) {
638 
            Ok(()) | Err(object_store::Error::NotFound { .. }) => Ok(()),
639 
            Err(e) => Err(self.map(e, key)),
640 
        }
641 
    }
642
643 
    fn read_ref(&self, key: &str) -> Result<Option<(Vec<u8>, Version)>> {
644 
        let path = self.path(key)?;
645 
        let got = self.block(async {
646 
            let r = self.inner.get_opts(&path, GetOptions::default()).await?;
647 
            let meta = r.meta.clone();
648 
            let bytes = r.bytes().await?;
649 
            Ok::<_, object_store::Error>((bytes, meta))
650 
        });
651 
        match got {
652 
            Ok((bytes, meta)) => Ok(Some((
653 
                bytes.to_vec(),
654 
                Self::version_of(meta.e_tag, meta.version),
655 
            ))),
656 
            Err(object_store::Error::NotFound { .. }) => Ok(None),
657 
            Err(e) => Err(self.map(e, key)),
658 
        }
659 
    }
660
661 
    fn write_ref(&self, key: &str, bytes: &[u8], expected: Option<&Version>) -> Result<Version> {
662 
        let path = self.path(key)?;
663 
        let mode = match expected {
664 
            None => PutMode::Create,
665 
            Some(v) => PutMode::Update(Self::update_version(v)),
666 
        };
667 
        let opts = PutOptions {
668 
            mode,
669 
            ..Default::default()
670 
        };
671 
        let put = self
672 
            .block(
673 
                self.inner
674 
                    .put_opts(&path, PutPayload::from(bytes.to_vec()), opts),
675 
            )
676 
            .map_err(|e| match self.map(e, key) {
677 
                // A record that exists when none was expected is a lost
678 
                // race too, not a misuse of put_new.
679 
                Error::Exists(k) => Error::Conflict(k),
680 
                other => other,
681 
            })?;
682 
        Ok(Self::version_of(put.e_tag, put.version))
683 
    }
684 
}
685
686 
#[cfg(test)]
687 
mod tests {
688
689 
    /// One client per store and credentials (oxygenhub#234): opening the
690 
    /// same again shares it, and only a new one counts as opened.
691 
    #[test]
692 
    fn a_store_opened_again_shares_its_client() {
693 
        let creds = |k: &str| Credentials::Aws {
694 
            access_key_id: k.into(),
695 
            secret_access_key: "s".into(),
696 
            session_token: None,
697 
        };
698 
        let url = "s3://oxy-shared-test/repos/a?region=eu-west-2";
699 
        let before = super::stats().opens;
700 
        let a = super::shared(url, Some(creds("one"))).unwrap();
701 
        let b = super::shared(url, Some(creds("one"))).unwrap();
702 
        assert!(std::sync::Arc::ptr_eq(&a, &b));
703 
        let other_keys = super::shared(url, Some(creds("two"))).unwrap();
704 
        assert!(
705 
            !std::sync::Arc::ptr_eq(&a, &other_keys),
706 
            "other credentials, another client"
707 
        );
708 
        assert!(
709 
            super::stats().opens - before >= 2,
710 
            "two clients opened (other tests may open more)"
711 
        );
712 
    }
713
714 
    #[tokio::test]
715 
    async fn a_store_can_be_used_from_inside_a_runtime() {
716 
        // As a server's handler would: blocking on the store's own runtime
717 
        // from here must neither panic nor deadlock, and nor must dropping it.
718 
        let store = CloudStore::open(
719 
            "s3://oxy-no-such-bucket/p?region=eu-west-2&endpoint=http://127.0.0.1:9",
720 
        )
721 
        .unwrap();
722 
        assert_eq!(store.block(async { 7 }), 7);
723 
        drop(store);
724 
    }
725
726 
    #[test]
727 
    fn credentials_round_trip_and_never_show_their_secret() {
728 
        let aws = Credentials::Aws {
729 
            access_key_id: "AKIDEXAMPLE".into(),
730 
            secret_access_key: "wJalrXUtnFEMI".into(),
731 
            session_token: None,
732 
        };
733 
        assert_eq!(Credentials::from_json(&aws.to_json()).unwrap(), aws);
734 
        assert_eq!(format!("{aws:?}"), "Aws(AKIDEXAMPLE, secret hidden)");
735 
        let az = Credentials::Azure {
736 
            tenant_id: "t".into(),
737 
            client_id: "c".into(),
738 
            client_secret: "s3cr3t".into(),
739 
        };
740 
        assert_eq!(Credentials::from_json(&az.to_json()).unwrap(), az);
741 
        assert!(!format!("{az:?}").contains("s3cr3t"));
742 
        let key = r#"{"type":"service_account","client_email":"x@y","private_key":"k"}"#;
743 
        let gcp = Credentials::Gcp {
744 
            service_account_key: key.into(),
745 
        };
746 
        assert_eq!(Credentials::from_json(&gcp.to_json()).unwrap(), gcp);
747 
        assert!(!format!("{gcp:?}").contains("private_key"));
748 
        for (bad, why) in [
749 
            (r#"{"kind":"aws","access_key_id":"a"}"#, "secret_access_key"),
750 
            (
751 
                r#"{"kind":"gcp","service_account_key":"{\"type\":\"authorized_user\"}"}"#,
752 
                "not a service account key",
753 
            ),
754 
            (r#"{"kind":"ftp"}"#, "aws, gcp or azure"),
755 
            ("nope", "not JSON"),
756 
        ] {
757 
            let e = Credentials::from_json(bad).unwrap_err().to_string();
758 
            assert!(e.contains(why), "{bad}: {e}");
759 
        }
760 
        let e = CloudStore::open_with("gs://bucket/p", Some(&aws))
761 
            .err()
762 
            .unwrap()
763 
            .to_string();
764 
        assert!(e.contains("are for s3:// stores"), "{e}");
765 
    }
766
767 
    use super::*;
768
769 
    #[test]
770 
    fn urls_parse_with_their_parameters() {
771 
        let u =
772 
            StoreUrl::parse("s3://my-bucket/repos/oxygen?profile=work&region=eu-west-2").unwrap();
773 
        assert_eq!(u.scheme, "s3");
774 
        assert_eq!(u.host, "my-bucket");
775 
        assert_eq!(u.path, "repos/oxygen");
776 
        assert_eq!(u.param("profile"), Some("work"));
777 
        assert_eq!(u.param("region"), Some("eu-west-2"));
778 
        assert_eq!(u.plain(), "s3://my-bucket/repos/oxygen");
779 
        let u = StoreUrl::parse("gs://bucket").unwrap();
780 
        assert_eq!((u.host.as_str(), u.path.as_str()), ("bucket", ""));
781 
        assert_eq!(u.plain(), "gs://bucket");
782 
        let u = StoreUrl::parse("az://acct/container/a/b/?endpoint=http%3A%2F%2F127.0.0.1%3A10000")
783 
            .unwrap();
784 
        assert_eq!(u.path, "container/a/b");
785 
        assert_eq!(u.param("endpoint"), Some("http://127.0.0.1:10000"));
786 
        assert!(StoreUrl::parse("s3://").is_err());
787 
        assert!(StoreUrl::parse("nonsense").is_err());
788 
    }
789
790 
    #[test]
791 
    fn versions_carry_both_an_etag_and_a_generation() {
792 
        let v = CloudStore::version_of(Some("\"abc\"".into()), Some("17".into()));
793 
        let u = CloudStore::update_version(&v);
794 
        assert_eq!(u.e_tag.as_deref(), Some("\"abc\""));
795 
        assert_eq!(u.version.as_deref(), Some("17"));
796 
        let u = CloudStore::update_version(&CloudStore::version_of(None, Some("3".into())));
797 
        assert_eq!((u.e_tag, u.version.as_deref()), (None, Some("3")));
798 
    }
799 
}