OXYGENHUB
# 30644f1ecb7b oxygen / crates / oxy-strata / src / lib.rs
36.7 KiB · Rust · last changed Bucket stores count pack listings and kept listings apart 18 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 
static BY_OP: [std::sync::atomic::AtomicU64; Op::ALL.len()] =
110 
    [const { std::sync::atomic::AtomicU64::new(0) }; Op::ALL.len()];
111
112 
/// The kinds of request a cloud store makes, counted apart so a server can
113 
/// tell which of them its time goes to.
114 
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
115 
pub enum Op {
116 
    /// `has` that found the object.
117 
    Has,
118 
    /// `has` that did not: an absent object is never cached.
119 
    HasMissing,
120 
    Get,
121 
    Range,
122 
    Put,
123 
    List,
124 
    /// A listing of a repository's packs, sent.
125 
    ListPacks,
126 
    /// A listing answered from [`Listings`], not sent.
127 
    ListKept,
128 
    Delete,
129 
    ReadRef,
130 
    WriteRef,
131 
}
132
133 
impl Op {
134 
    pub const ALL: [Op; 11] = [
135 
        Op::Has,
136 
        Op::HasMissing,
137 
        Op::Get,
138 
        Op::Range,
139 
        Op::Put,
140 
        Op::List,
141 
        Op::ListPacks,
142 
        Op::ListKept,
143 
        Op::Delete,
144 
        Op::ReadRef,
145 
        Op::WriteRef,
146 
    ];
147
148 
    pub fn name(self) -> &'static str {
149 
        match self {
150 
            Op::Has => "has",
151 
            Op::HasMissing => "has_missing",
152 
            Op::Get => "get",
153 
            Op::Range => "range",
154 
            Op::Put => "put",
155 
            Op::List => "list",
156 
            Op::ListPacks => "list_packs",
157 
            Op::ListKept => "list_kept",
158 
            Op::Delete => "delete",
159 
            Op::ReadRef => "read_ref",
160 
            Op::WriteRef => "write_ref",
161 
        }
162 
    }
163 
}
164
165 
fn counted(op: Op) {
166 
    BY_OP[op as usize].fetch_add(1, std::sync::atomic::Ordering::Relaxed);
167 
}
168
169 
/// What the cloud stores have done since the process started: clients
170 
/// opened, requests made and the time spent in them, for a server to show
171 
/// (a hub's /health).
172 
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
173 
pub struct Stats {
174 
    pub opens: u64,
175 
    pub calls: u64,
176 
    pub millis: u64,
177 
    /// Requests by kind, in [`Op::ALL`]'s order.
178 
    pub by_op: [u64; Op::ALL.len()],
179 
}
180
181 
impl Stats {
182 
    /// Each kind of request with how many were made.
183 
    pub fn ops(&self) -> impl Iterator<Item = (&'static str, u64)> + '_ {
184 
        Op::ALL
185 
            .iter()
186 
            .map(|op| (op.name(), self.by_op[*op as usize]))
187 
    }
188 
}
189
190 
pub fn stats() -> Stats {
191 
    use std::sync::atomic::Ordering::Relaxed;
192 
    Stats {
193 
        opens: OPENS.load(Relaxed),
194 
        calls: CALLS.load(Relaxed),
195 
        millis: MICROS.load(Relaxed) / 1000,
196 
        by_op: std::array::from_fn(|i| BY_OP[i].load(Relaxed)),
197 
    }
198 
}
199
200 
/// Credentials a program keeps itself and hands over, instead of the
201 
/// provider's usual places: a hub keeps each repository's sealed. When a
202 
/// store is opened with them, nothing from the environment or the
203 
/// provider's files is used.
204 
#[derive(Clone, PartialEq, Eq, Hash)]
205 
pub enum Credentials {
206 
    /// An S3 access key.
207 
    Aws {
208 
        access_key_id: String,
209 
        secret_access_key: String,
210 
        session_token: Option<String>,
211 
    },
212 
    /// A Google Cloud service account key, as the JSON Google issues.
213 
    Gcp { service_account_key: String },
214 
    /// An Azure service principal.
215 
    Azure {
216 
        tenant_id: String,
217 
        client_id: String,
218 
        client_secret: String,
219 
    },
220 
}
221
222 
impl std::fmt::Debug for Credentials {
223 
    /// Names the kind and the public part only: never the secret.
224 
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
225 
        match self {
226 
            Credentials::Aws { access_key_id, .. } => {
227 
                write!(f, "Aws({access_key_id}, secret hidden)")
228 
            }
229 
            Credentials::Gcp { .. } => write!(f, "Gcp(service account key hidden)"),
230 
            Credentials::Azure { client_id, .. } => write!(f, "Azure({client_id}, secret hidden)"),
231 
        }
232 
    }
233 
}
234
235 
impl Credentials {
236 
    /// The scheme of the stores these are for.
237 
    pub fn scheme(&self) -> &'static str {
238 
        match self {
239 
            Credentials::Aws { .. } => "s3",
240 
            Credentials::Gcp { .. } => "gs",
241 
            Credentials::Azure { .. } => "az",
242 
        }
243 
    }
244
245 
    /// As JSON, for keeping (sealed) and reading back with [`Credentials::from_json`].
246 
    pub fn to_json(&self) -> String {
247 
        let v = match self {
248 
            Credentials::Aws {
249 
                access_key_id,
250 
                secret_access_key,
251 
                session_token,
252 
            } => serde_json::json!({
253 
                "kind": "aws", "access_key_id": access_key_id, "secret_access_key": secret_access_key,
254 
                "session_token": session_token,
255 
            }),
256 
            Credentials::Gcp {
257 
                service_account_key,
258 
            } => {
259 
                serde_json::json!({ "kind": "gcp", "service_account_key": service_account_key })
260 
            }
261 
            Credentials::Azure {
262 
                tenant_id,
263 
                client_id,
264 
                client_secret,
265 
            } => serde_json::json!({
266 
                "kind": "azure", "tenant_id": tenant_id, "client_id": client_id, "client_secret": client_secret,
267 
            }),
268 
        };
269 
        v.to_string()
270 
    }
271
272 
    pub fn from_json(text: &str) -> Result<Credentials> {
273 
        let v: serde_json::Value = serde_json::from_str(text)
274 
            .map_err(|e| Error::Backend(format!("credentials are not JSON: {e}")))?;
275 
        let field = |k: &str| -> Result<String> {
276 
            v.get(k)
277 
                .and_then(|x| x.as_str())
278 
                .filter(|x| !x.trim().is_empty())
279 
                .map(|x| x.trim().to_string())
280 
                .ok_or_else(|| Error::Backend(format!("credentials need {k}")))
281 
        };
282 
        match v.get("kind").and_then(|k| k.as_str()) {
283 
            Some("aws") => Ok(Credentials::Aws {
284 
                access_key_id: field("access_key_id")?,
285 
                secret_access_key: field("secret_access_key")?,
286 
                session_token: field("session_token").ok(),
287 
            }),
288 
            Some("gcp") => {
289 
                let key = field("service_account_key")?;
290 
                let parsed: serde_json::Value = serde_json::from_str(&key).map_err(|_| {
291 
                    Error::Backend("the service account key is not the JSON Google issues".into())
292 
                })?;
293 
                if parsed.get("type").and_then(|t| t.as_str()) != Some("service_account") {
294 
                    return Err(Error::Backend(
295 
                        "the key is not a service account key (its type is not service_account)"
296 
                            .into(),
297 
                    ));
298 
                }
299 
                Ok(Credentials::Gcp {
300 
                    service_account_key: key,
301 
                })
302 
            }
303 
            Some("azure") => Ok(Credentials::Azure {
304 
                tenant_id: field("tenant_id")?,
305 
                client_id: field("client_id")?,
306 
                client_secret: field("client_secret")?,
307 
            }),
308 
            _ => Err(Error::Backend(
309 
                "credentials need a kind: aws, gcp or azure".into(),
310 
            )),
311 
        }
312 
    }
313 
}
314
315 
/// Finds the credentials a program keeps for a store URL.
316 
pub type CredentialsLookup = fn(&str) -> Option<Credentials>;
317
318 
static CREDENTIALS: Mutex<Option<CredentialsLookup>> = Mutex::new(None);
319
320 
/// Have stores opened by URL (as a repository's `storage.url` is) use the
321 
/// credentials `lookup` finds for it, where it finds any.
322 
pub fn set_credentials_lookup(lookup: CredentialsLookup) {
323 
    *CREDENTIALS.lock().unwrap_or_else(|p| p.into_inner()) = Some(lookup);
324 
}
325
326 
/// A parsed store URL: `SCHEME://HOST/PATH?PARAMS`.
327 
#[derive(Clone, Debug, PartialEq, Eq)]
328 
pub struct StoreUrl {
329 
    pub scheme: String,
330 
    /// The bucket (S3, GCS) or the storage account (Azure).
331 
    pub host: String,
332 
    /// The path under the host, without leading or trailing slashes; for
333 
    /// Azure the container is its first segment.
334 
    pub path: String,
335 
    pub params: BTreeMap<String, String>,
336 
}
337
338 
impl StoreUrl {
339 
    pub fn parse(url: &str) -> Result<StoreUrl> {
340 
        let bad = |why: &str| Error::Backend(format!("{url}: {why}"));
341 
        let (scheme, rest) = url.split_once("://").ok_or_else(|| bad("not a URL"))?;
342 
        let (rest, query) = match rest.split_once('?') {
343 
            Some((r, q)) => (r, Some(q)),
344 
            None => (rest, None),
345 
        };
346 
        let (host, path) = match rest.split_once('/') {
347 
            Some((h, p)) => (h, p.trim_matches('/')),
348 
            None => (rest, ""),
349 
        };
350 
        if host.is_empty() {
351 
            return Err(bad("names no bucket"));
352 
        }
353 
        let mut params = BTreeMap::new();
354 
        for pair in query.unwrap_or("").split('&').filter(|p| !p.is_empty()) {
355 
            let (k, v) = pair.split_once('=').unwrap_or((pair, ""));
356 
            params.insert(k.to_string(), percent_decode(v));
357 
        }
358 
        Ok(StoreUrl {
359 
            scheme: scheme.to_string(),
360 
            host: host.to_string(),
361 
            path: path.to_string(),
362 
            params,
363 
        })
364 
    }
365
366 
    /// The URL without its parameters, for messages.
367 
    pub fn plain(&self) -> String {
368 
        if self.path.is_empty() {
369 
            format!("{}://{}", self.scheme, self.host)
370 
        } else {
371 
            format!("{}://{}/{}", self.scheme, self.host, self.path)
372 
        }
373 
    }
374
375 
    pub fn param(&self, name: &str) -> Option<&str> {
376 
        self.params.get(name).map(String::as_str)
377 
    }
378 
}
379
380 
fn percent_decode(s: &str) -> String {
381 
    let bytes = s.as_bytes();
382 
    let mut out = Vec::with_capacity(bytes.len());
383 
    let mut i = 0;
384 
    while i < bytes.len() {
385 
        if bytes[i] == b'%' && i + 2 < bytes.len() {
386 
            if let Ok(v) = u8::from_str_radix(&s[i + 1..i + 3], 16) {
387 
                out.push(v);
388 
                i += 3;
389 
                continue;
390 
            }
391 
        }
392 
        out.push(bytes[i]);
393 
        i += 1;
394 
    }
395 
    String::from_utf8_lossy(&out).into_owned()
396 
}
397
398 
/// A store on a cloud provider: the provider's client, the prefix every
399 
/// key goes under, and a runtime to run the client on.
400 
pub struct CloudStore {
401 
    inner: Arc<dyn ObjectStore>,
402 
    prefix: String,
403 
    name: String,
404 
    hint: &'static str,
405 
    /// Always there until the store is dropped: two workers, so the
406 
    /// requests of a shared store run side by side.
407 
    runtime: Option<tokio::runtime::Runtime>,
408 
    /// Keys `has` found missing lately (see [`Misses`]).
409 
    misses: Misses,
410 
    /// Pack listings, kept a moment (see [`Listings`]).
411 
    listings: Listings,
412 
}
413
414 
/// How long a key found missing is taken as missing without asking again.
415 
const MISS_FOR: std::time::Duration = std::time::Duration::from_secs(30);
416
417 
/// Keys found missing in the last [`MISS_FOR`], so a server answering the
418 
/// same question again and again does not send a request each time
419 
/// (oxygenhub#228: a client's negotiate asked about ~190 objects the hub
420 
/// lacked every 8 s, 45 ms a request). This process forgets a key the
421 
/// moment it writes it, so its own writes are never hidden; another writer
422 
/// to the same keys is seen within `MISS_FOR`.
423 
#[derive(Default)]
424 
struct Misses(Mutex<std::collections::HashMap<String, std::time::Instant>>);
425
426 
impl Misses {
427 
    /// At most this many kept; past it the oldest half is let go.
428 
    const MAX: usize = 100_000;
429
430 
    fn known(&self, key: &str) -> bool {
431 
        let map = self.0.lock().unwrap_or_else(|p| p.into_inner());
432 
        map.get(key).is_some_and(|t| t.elapsed() < MISS_FOR)
433 
    }
434
435 
    fn missed(&self, key: &str) {
436 
        let mut map = self.0.lock().unwrap_or_else(|p| p.into_inner());
437 
        if map.len() >= Self::MAX {
438 
            map.retain(|_, t| t.elapsed() < MISS_FOR / 2);
439 
            if map.len() >= Self::MAX {
440 
                map.clear();
441 
            }
442 
        }
443 
        map.insert(key.to_string(), std::time::Instant::now());
444 
    }
445
446 
    fn written(&self, key: &str) {
447 
        self.0.lock().unwrap_or_else(|p| p.into_inner()).remove(key);
448 
    }
449 
}
450
451 
/// Listings kept for [`MISS_FOR`] (oxygen#116): the `pack-` keys, which
452 
/// every open of a repository's store lists, so a forge serving a
453 
/// repository from a bucket lists them once a while rather than once a
454 
/// request. Like [`Misses`], this process forgets a listing the moment it
455 
/// writes or deletes a key under it; another writer's pack is seen within
456 
/// `MISS_FOR`, as a missing key is.
457 
#[derive(Default)]
458 
struct Listings(Mutex<std::collections::HashMap<String, (std::time::Instant, Vec<Entry>)>>);
459
460 
impl Listings {
461 
    /// The prefixes kept: those every open lists.
462 
    fn kept(prefix: &str) -> bool {
463 
        prefix == "pack-"
464 
    }
465
466 
    fn get(&self, prefix: &str) -> Option<Vec<Entry>> {
467 
        if !Self::kept(prefix) {
468 
            return None;
469 
        }
470 
        let map = self.0.lock().unwrap_or_else(|p| p.into_inner());
471 
        map.get(prefix)
472 
            .filter(|(t, _)| t.elapsed() < MISS_FOR)
473 
            .map(|(_, e)| e.clone())
474 
    }
475
476 
    fn put(&self, prefix: &str, entries: &[Entry]) {
477 
        if Self::kept(prefix) {
478 
            self.0.lock().unwrap_or_else(|p| p.into_inner()).insert(
479 
                prefix.to_string(),
480 
                (std::time::Instant::now(), entries.to_vec()),
481 
            );
482 
        }
483 
    }
484
485 
    fn written(&self, key: &str) {
486 
        self.0
487 
            .lock()
488 
            .unwrap_or_else(|p| p.into_inner())
489 
            .retain(|prefix, _| !key.starts_with(prefix.as_str()));
490 
    }
491 
}
492
493 
impl Drop for CloudStore {
494 
    /// Without waiting, so a store can be dropped inside a runtime too.
495 
    fn drop(&mut self) {
496 
        if let Some(rt) = self.runtime.take() {
497 
            rt.shutdown_background();
498 
        }
499 
    }
500 
}
501
502 
impl CloudStore {
503 
    /// Open the store a URL names. Nothing is sent yet: credentials are
504 
    /// resolved here, but the bucket is first spoken to on the first call.
505 
    pub fn open(url: &str) -> Result<CloudStore> {
506 
        Self::open_with(url, None)
507 
    }
508
509 
    /// [`CloudStore::open`] with credentials given, which must be for
510 
    /// the URL's kind of store; then only they are used.
511 
    pub fn open_with(url: &str, given: Option<&Credentials>) -> Result<CloudStore> {
512 
        let parsed = StoreUrl::parse(url)?;
513 
        if let Some(c) = given.filter(|c| c.scheme() != parsed.scheme) {
514 
            return Err(Error::Backend(format!(
515 
                "{}: the credentials given are for {}:// stores",
516 
                parsed.plain(),
517 
                c.scheme()
518 
            )));
519 
        }
520 
        let aws_keys = match given {
521 
            Some(Credentials::Aws {
522 
                access_key_id,
523 
                secret_access_key,
524 
                session_token,
525 
            }) => Some((
526 
                access_key_id.as_str(),
527 
                secret_access_key.as_str(),
528 
                session_token.as_deref(),
529 
            )),
530 
            _ => None,
531 
        };
532 
        let gcp_key = match given {
533 
            Some(Credentials::Gcp {
534 
                service_account_key,
535 
            }) => Some(service_account_key.as_str()),
536 
            _ => None,
537 
        };
538 
        let principal = match given {
539 
            Some(Credentials::Azure {
540 
                tenant_id,
541 
                client_id,
542 
                client_secret,
543 
            }) => Some((
544 
                tenant_id.as_str(),
545 
                client_id.as_str(),
546 
                client_secret.as_str(),
547 
            )),
548 
            _ => None,
549 
        };
550 
        OPENS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
551 
        let runtime = tokio::runtime::Builder::new_multi_thread()
552 
            .worker_threads(2)
553 
            .thread_name("oxy-strata")
554 
            .enable_all()
555 
            .build()
556 
            .map_err(|e| Error::Backend(format!("starting the HTTP runtime: {e}")))?;
557 
        let (inner, prefix, hint): (Arc<dyn ObjectStore>, String, &'static str) =
558 
            match parsed.scheme.as_str() {
559 
                "s3" => (
560 
                    Arc::new(aws::open(&parsed, aws_keys)?),
561 
                    parsed.path.clone(),
562 
                    aws::HINT,
563 
                ),
564 
                "gs" => (
565 
                    Arc::new(gcp::open(&parsed, gcp_key)?),
566 
                    parsed.path.clone(),
567 
                    gcp::HINT,
568 
                ),
569 
                "az" => {
570 
                    let (store, prefix) = azure::open(&parsed, principal)?;
571 
                    (Arc::new(store), prefix, azure::HINT)
572 
                }
573 
                other => {
574 
                    return Err(Error::Backend(format!(
575 
                        "{url}: {other}:// is not a store oxy knows (s3://, gs://, az://, dir://)"
576 
                    )))
577 
                }
578 
            };
579 
        Ok(CloudStore {
580 
            inner,
581 
            prefix,
582 
            name: parsed.plain(),
583 
            hint,
584 
            runtime: Some(runtime),
585 
            misses: Misses::default(),
586 
            listings: Listings::default(),
587 
        })
588 
    }
589
590 
    /// Run a request to completion. Called from a thread that is itself
591 
    /// driving async tasks (a server's handler), where blocking on another
592 
    /// runtime is not allowed, the request runs on a thread of its own.
593 
    fn block<F>(&self, f: F) -> F::Output
594 
    where
595 
        F: Future + Send,
596 
        F::Output: Send,
597 
    {
598 
        let runtime = self
599 
            .runtime
600 
            .as_ref()
601 
            .expect("the runtime lives as long as the store");
602 
        let started = std::time::Instant::now();
603 
        let out = if tokio::runtime::Handle::try_current().is_ok() {
604 
            std::thread::scope(|s| {
605 
                s.spawn(|| runtime.block_on(f))
606 
                    .join()
607 
                    .unwrap_or_else(|p| std::panic::resume_unwind(p))
608 
            })
609 
        } else {
610 
            runtime.block_on(f)
611 
        };
612 
        CALLS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
613 
        MICROS.fetch_add(
614 
            started.elapsed().as_micros() as u64,
615 
            std::sync::atomic::Ordering::Relaxed,
616 
        );
617 
        out
618 
    }
619
620 
    fn path(&self, key: &str) -> Result<ObjectPath> {
621 
        check_key(key)?;
622 
        let full = if self.prefix.is_empty() {
623 
            key.to_string()
624 
        } else {
625 
            format!("{}/{key}", self.prefix)
626 
        };
627 
        ObjectPath::parse(&full).map_err(|e| Error::Backend(format!("{full}: {e}")))
628 
    }
629
630 
    /// Our error for the provider's, with what to do next where it helps.
631 
    fn map(&self, e: object_store::Error, key: &str) -> Error {
632 
        use object_store::Error as E;
633 
        match e {
634 
            E::NotFound { .. } => Error::NotFound(key.to_string()),
635 
            E::AlreadyExists { .. } => Error::Exists(key.to_string()),
636 
            E::Precondition { .. } | E::NotModified { .. } => Error::Conflict(key.to_string()),
637 
            other => {
638 
                let text = other.to_string();
639 
                let lower = text.to_ascii_lowercase();
640 
                let advice = if lower.contains("403")
641 
                    || lower.contains("accessdenied")
642 
                    || lower.contains("forbidden")
643 
                    || lower.contains("authorizationpermissionmismatch")
644 
                {
645 
                    format!(
646 
                        ": access denied; the credentials in use may not do this on {}: {}",
647 
                        self.name, self.hint
648 
                    )
649 
                } else if lower.contains("credential") || lower.contains("no valid credentials") {
650 
                    format!(": no usable credentials: {}", self.hint)
651 
                } else if lower.contains("nosuchbucket")
652 
                    || lower.contains("bucketnotfound")
653 
                    || lower.contains("containernotfound")
654 
                {
655 
                    format!(
656 
                        ": {} does not exist, or the credentials cannot see it",
657 
                        self.name
658 
                    )
659 
                } else if lower.contains("dns")
660 
                    || lower.contains("connect")
661 
                    || lower.contains("timed out")
662 
                {
663 
                    ": cannot reach the service; check the network, the endpoint and the region"
664 
                        .to_string()
665 
                } else {
666 
                    String::new()
667 
                };
668 
                Error::Backend(format!("{}: {text}{advice}", self.name))
669 
            }
670 
        }
671 
    }
672
673 
    fn version_of(e_tag: Option<String>, version: Option<String>) -> Version {
674 
        Version(format!(
675 
            "{}\t{}",
676 
            e_tag.unwrap_or_default(),
677 
            version.unwrap_or_default()
678 
        ))
679 
    }
680
681 
    fn update_version(v: &Version) -> UpdateVersion {
682 
        let (e_tag, version) = v.0.split_once('\t').unwrap_or((v.0.as_str(), ""));
683 
        UpdateVersion {
684 
            e_tag: (!e_tag.is_empty()).then(|| e_tag.to_string()),
685 
            version: (!version.is_empty()).then(|| version.to_string()),
686 
        }
687 
    }
688 
}
689
690 
impl Store for CloudStore {
691 
    fn describe(&self) -> String {
692 
        self.name.clone()
693 
    }
694
695 
    fn has(&self, key: &str) -> Result<bool> {
696 
        let path = self.path(key)?;
697 
        if self.misses.known(key) {
698 
            return Ok(false);
699 
        }
700 
        match self.block(self.inner.head(&path)) {
701 
            Ok(_) => {
702 
                counted(Op::Has);
703 
                Ok(true)
704 
            }
705 
            Err(object_store::Error::NotFound { .. }) => {
706 
                counted(Op::HasMissing);
707 
                self.misses.missed(key);
708 
                Ok(false)
709 
            }
710 
            Err(e) => Err(self.map(e, key)),
711 
        }
712 
    }
713
714 
    fn get(&self, key: &str) -> Result<Vec<u8>> {
715 
        let path = self.path(key)?;
716 
        counted(Op::Get);
717 
        self.block(async {
718 
            let r = self.inner.get(&path).await?;
719 
            r.bytes().await
720 
        })
721 
        .map(|b| b.to_vec())
722 
        .map_err(|e| self.map(e, key))
723 
    }
724
725 
    fn get_range(&self, key: &str, range: Range<u64>) -> Result<Vec<u8>> {
726 
        let path = self.path(key)?;
727 
        if range.start > range.end {
728 
            return Err(Error::Backend(format!(
729 
                "{key}: bytes {}..{} is not a range",
730 
                range.start, range.end
731 
            )));
732 
        }
733 
        if range.is_empty() {
734 
            return Ok(Vec::new());
735 
        }
736 
        let want = (range.end - range.start) as usize;
737 
        counted(Op::Range);
738 
        let bytes = self
739 
            .block(self.inner.get_range(&path, range.clone()))
740 
            .map_err(|e| self.map(e, key))?;
741 
        if bytes.len() != want {
742 
            return Err(Error::Backend(format!(
743 
                "{key}: bytes {}..{} asked, {} returned: the object is shorter",
744 
                range.start,
745 
                range.end,
746 
                bytes.len()
747 
            )));
748 
        }
749 
        Ok(bytes.to_vec())
750 
    }
751
752 
    fn put_new(&self, key: &str, bytes: &[u8]) -> Result<()> {
753 
        let path = self.path(key)?;
754 
        self.misses.written(key);
755 
        self.listings.written(key);
756 
        let opts = PutOptions {
757 
            mode: PutMode::Create,
758 
            ..Default::default()
759 
        };
760 
        counted(Op::Put);
761 
        self.block(
762 
            self.inner
763 
                .put_opts(&path, PutPayload::from(bytes.to_vec()), opts),
764 
        )
765 
        .map(|_| ())
766 
        .map_err(|e| self.map(e, key))
767 
    }
768
769 
    fn list(&self, prefix: &str) -> Result<Vec<Entry>> {
770 
        if let Some(kept) = self.listings.get(prefix) {
771 
            counted(Op::ListKept);
772 
            return Ok(kept);
773 
        }
774 
        // Providers list by whole path segments; the asked prefix may end
775 
        // mid-segment, so list its directory and filter here.
776 
        let dir = prefix.rsplit_once('/').map(|(d, _)| d).unwrap_or("");
777 
        let full_dir = match (self.prefix.as_str(), dir) {
778 
            ("", "") => None,
779 
            ("", d) => Some(d.to_string()),
780 
            (p, "") => Some(p.to_string()),
781 
            (p, d) => Some(format!("{p}/{d}")),
782 
        };
783 
        let full_dir = full_dir
784 
            .map(|d| ObjectPath::parse(&d).map_err(|e| Error::Backend(format!("{d}: {e}"))))
785 
            .transpose()?;
786 
        counted(if Listings::kept(prefix) {
787 
            Op::ListPacks
788 
        } else {
789 
            Op::List
790 
        });
791 
        let metas: Vec<object_store::ObjectMeta> = self
792 
            .block(self.inner.list(full_dir.as_ref()).try_collect())
793 
            .map_err(|e| self.map(e, prefix))?;
794 
        let strip = if self.prefix.is_empty() {
795 
            String::new()
796 
        } else {
797 
            format!("{}/", self.prefix)
798 
        };
799 
        let mut out: Vec<Entry> = metas
800 
            .into_iter()
801 
            .filter_map(|m| {
802 
                let loc = m.location.to_string();
803 
                let key = loc.strip_prefix(&strip)?;
804 
                key.starts_with(prefix).then(|| Entry {
805 
                    key: key.to_string(),
806 
                    size: m.size,
807 
                    modified_ms: u64::try_from(m.last_modified.timestamp_millis()).ok(),
808 
                })
809 
            })
810 
            .collect();
811 
        out.sort_by(|a, b| a.key.cmp(&b.key));
812 
        self.listings.put(prefix, &out);
813 
        Ok(out)
814 
    }
815
816 
    fn delete(&self, key: &str) -> Result<()> {
817 
        let path = self.path(key)?;
818 
        self.listings.written(key);
819 
        counted(Op::Delete);
820 
        match self.block(self.inner.delete(&path)) {
821 
            Ok(()) | Err(object_store::Error::NotFound { .. }) => Ok(()),
822 
            Err(e) => Err(self.map(e, key)),
823 
        }
824 
    }
825
826 
    fn read_ref(&self, key: &str) -> Result<Option<(Vec<u8>, Version)>> {
827 
        let path = self.path(key)?;
828 
        counted(Op::ReadRef);
829 
        let got = self.block(async {
830 
            let r = self.inner.get_opts(&path, GetOptions::default()).await?;
831 
            let meta = r.meta.clone();
832 
            let bytes = r.bytes().await?;
833 
            Ok::<_, object_store::Error>((bytes, meta))
834 
        });
835 
        match got {
836 
            Ok((bytes, meta)) => Ok(Some((
837 
                bytes.to_vec(),
838 
                Self::version_of(meta.e_tag, meta.version),
839 
            ))),
840 
            Err(object_store::Error::NotFound { .. }) => Ok(None),
841 
            Err(e) => Err(self.map(e, key)),
842 
        }
843 
    }
844
845 
    fn write_ref(&self, key: &str, bytes: &[u8], expected: Option<&Version>) -> Result<Version> {
846 
        let path = self.path(key)?;
847 
        self.misses.written(key);
848 
        self.listings.written(key);
849 
        let mode = match expected {
850 
            None => PutMode::Create,
851 
            Some(v) => PutMode::Update(Self::update_version(v)),
852 
        };
853 
        let opts = PutOptions {
854 
            mode,
855 
            ..Default::default()
856 
        };
857 
        counted(Op::WriteRef);
858 
        let put = self
859 
            .block(
860 
                self.inner
861 
                    .put_opts(&path, PutPayload::from(bytes.to_vec()), opts),
862 
            )
863 
            .map_err(|e| match self.map(e, key) {
864 
                // A record that exists when none was expected is a lost
865 
                // race too, not a misuse of put_new.
866 
                Error::Exists(k) => Error::Conflict(k),
867 
                other => other,
868 
            })?;
869 
        Ok(Self::version_of(put.e_tag, put.version))
870 
    }
871 
}
872
873 
#[cfg(test)]
874 
mod tests {
875
876 
    /// A miss is remembered until the key is written, or for MISS_FOR.
877 
    #[test]
878 
    fn a_miss_is_remembered_until_written() {
879 
        let m = super::Misses::default();
880 
        assert!(!m.known("loose/ab/cd"));
881 
        m.missed("loose/ab/cd");
882 
        assert!(m.known("loose/ab/cd") && !m.known("loose/ab/ce"));
883 
        m.written("loose/ab/cd");
884 
        assert!(!m.known("loose/ab/cd"), "a write forgets the miss");
885 
        m.missed("old");
886 
        m.0.lock()
887 
            .unwrap()
888 
            .insert("old".into(), std::time::Instant::now() - super::MISS_FOR);
889 
        assert!(!m.known("old"), "a miss expires");
890 
    }
891
892 
    /// Every kind of request is reported, once and by its own name.
893 
    /// The pack listing is kept until a pack key is written or deleted
894 
    /// here, or for MISS_FOR; other prefixes are never kept (oxygen#116).
895 
    #[test]
896 
    fn the_pack_listing_is_kept_until_a_pack_is_written() {
897 
        let l = super::Listings::default();
898 
        let e = |k: &str| oxy_core::strata::Entry {
899 
            key: k.into(),
900 
            size: 1,
901 
            modified_ms: None,
902 
        };
903 
        l.put("pack-", &[e("pack-1.idx")]);
904 
        assert_eq!(l.get("pack-").map(|v| v.len()), Some(1));
905 
        l.put("loose/", &[e("loose/ab")]);
906 
        assert!(l.get("loose/").is_none(), "only the pack listing is kept");
907 
        l.written("loose/cd");
908 
        assert!(l.get("pack-").is_some(), "a loose object leaves it");
909 
        l.written("pack-2.pack");
910 
        assert!(l.get("pack-").is_none(), "a new pack forgets it");
911 
        l.put("pack-", &[e("pack-1.idx")]);
912 
        l.0.lock().unwrap().get_mut("pack-").unwrap().0 -= super::MISS_FOR;
913 
        assert!(l.get("pack-").is_none(), "and it ages out");
914 
    }
915
916 
    #[test]
917 
    fn stats_name_every_kind_of_request() {
918 
        let names: Vec<&str> = super::stats().ops().map(|(n, _)| n).collect();
919 
        assert_eq!(names.len(), super::Op::ALL.len());
920 
        let mut unique = names.clone();
921 
        unique.sort();
922 
        unique.dedup();
923 
        assert_eq!(unique.len(), names.len());
924 
        assert!(names.contains(&"has_missing") && names.contains(&"range"));
925 
    }
926
927 
    /// One client per store and credentials (oxygenhub#234): opening the
928 
    /// same again shares it, and only a new one counts as opened.
929 
    #[test]
930 
    fn a_store_opened_again_shares_its_client() {
931 
        let creds = |k: &str| Credentials::Aws {
932 
            access_key_id: k.into(),
933 
            secret_access_key: "s".into(),
934 
            session_token: None,
935 
        };
936 
        let url = "s3://oxy-shared-test/repos/a?region=eu-west-2";
937 
        let before = super::stats().opens;
938 
        let a = super::shared(url, Some(creds("one"))).unwrap();
939 
        let b = super::shared(url, Some(creds("one"))).unwrap();
940 
        assert!(std::sync::Arc::ptr_eq(&a, &b));
941 
        let other_keys = super::shared(url, Some(creds("two"))).unwrap();
942 
        assert!(
943 
            !std::sync::Arc::ptr_eq(&a, &other_keys),
944 
            "other credentials, another client"
945 
        );
946 
        assert!(
947 
            super::stats().opens - before >= 2,
948 
            "two clients opened (other tests may open more)"
949 
        );
950 
    }
951
952 
    #[tokio::test]
953 
    async fn a_store_can_be_used_from_inside_a_runtime() {
954 
        // As a server's handler would: blocking on the store's own runtime
955 
        // from here must neither panic nor deadlock, and nor must dropping it.
956 
        let store = CloudStore::open(
957 
            "s3://oxy-no-such-bucket/p?region=eu-west-2&endpoint=http://127.0.0.1:9",
958 
        )
959 
        .unwrap();
960 
        assert_eq!(store.block(async { 7 }), 7);
961 
        drop(store);
962 
    }
963
964 
    #[test]
965 
    fn credentials_round_trip_and_never_show_their_secret() {
966 
        let aws = Credentials::Aws {
967 
            access_key_id: "AKIDEXAMPLE".into(),
968 
            secret_access_key: "wJalrXUtnFEMI".into(),
969 
            session_token: None,
970 
        };
971 
        assert_eq!(Credentials::from_json(&aws.to_json()).unwrap(), aws);
972 
        assert_eq!(format!("{aws:?}"), "Aws(AKIDEXAMPLE, secret hidden)");
973 
        let az = Credentials::Azure {
974 
            tenant_id: "t".into(),
975 
            client_id: "c".into(),
976 
            client_secret: "s3cr3t".into(),
977 
        };
978 
        assert_eq!(Credentials::from_json(&az.to_json()).unwrap(), az);
979 
        assert!(!format!("{az:?}").contains("s3cr3t"));
980 
        let key = r#"{"type":"service_account","client_email":"x@y","private_key":"k"}"#;
981 
        let gcp = Credentials::Gcp {
982 
            service_account_key: key.into(),
983 
        };
984 
        assert_eq!(Credentials::from_json(&gcp.to_json()).unwrap(), gcp);
985 
        assert!(!format!("{gcp:?}").contains("private_key"));
986 
        for (bad, why) in [
987 
            (r#"{"kind":"aws","access_key_id":"a"}"#, "secret_access_key"),
988 
            (
989 
                r#"{"kind":"gcp","service_account_key":"{\"type\":\"authorized_user\"}"}"#,
990 
                "not a service account key",
991 
            ),
992 
            (r#"{"kind":"ftp"}"#, "aws, gcp or azure"),
993 
            ("nope", "not JSON"),
994 
        ] {
995 
            let e = Credentials::from_json(bad).unwrap_err().to_string();
996 
            assert!(e.contains(why), "{bad}: {e}");
997 
        }
998 
        let e = CloudStore::open_with("gs://bucket/p", Some(&aws))
999 
            .err()
1000 
            .unwrap()
1001 
            .to_string();
1002 
        assert!(e.contains("are for s3:// stores"), "{e}");
1003 
    }
1004
1005 
    use super::*;
1006
1007 
    #[test]
1008 
    fn urls_parse_with_their_parameters() {
1009 
        let u =
1010 
            StoreUrl::parse("s3://my-bucket/repos/oxygen?profile=work&region=eu-west-2").unwrap();
1011 
        assert_eq!(u.scheme, "s3");
1012 
        assert_eq!(u.host, "my-bucket");
1013 
        assert_eq!(u.path, "repos/oxygen");
1014 
        assert_eq!(u.param("profile"), Some("work"));
1015 
        assert_eq!(u.param("region"), Some("eu-west-2"));
1016 
        assert_eq!(u.plain(), "s3://my-bucket/repos/oxygen");
1017 
        let u = StoreUrl::parse("gs://bucket").unwrap();
1018 
        assert_eq!((u.host.as_str(), u.path.as_str()), ("bucket", ""));
1019 
        assert_eq!(u.plain(), "gs://bucket");
1020 
        let u = StoreUrl::parse("az://acct/container/a/b/?endpoint=http%3A%2F%2F127.0.0.1%3A10000")
1021 
            .unwrap();
1022 
        assert_eq!(u.path, "container/a/b");
1023 
        assert_eq!(u.param("endpoint"), Some("http://127.0.0.1:10000"));
1024 
        assert!(StoreUrl::parse("s3://").is_err());
1025 
        assert!(StoreUrl::parse("nonsense").is_err());
1026 
    }
1027
1028 
    #[test]
1029 
    fn versions_carry_both_an_etag_and_a_generation() {
1030 
        let v = CloudStore::version_of(Some("\"abc\"".into()), Some("17".into()));
1031 
        let u = CloudStore::update_version(&v);
1032 
        assert_eq!(u.e_tag.as_deref(), Some("\"abc\""));
1033 
        assert_eq!(u.version.as_deref(), Some("17"));
1034 
        let u = CloudStore::update_version(&CloudStore::version_of(None, Some("3".into())));
1035 
        assert_eq!((u.e_tag, u.version.as_deref()), (None, Some("3")));
1036 
    }
1037 
}