[{"_1":2,"_3":-5,"_4":-5},"loaderData",{"_5":6,"_7":8,"_9":10,"_11":12},"actionData","errors","root",{"_179":-5,"_182":183,"_184":185},"routes/shell",{"_41":42,"_178":-5,"_179":-5,"_180":181},"routes/repo",{"_34":-7,"_35":36,"_37":38,"_39":40,"_41":42,"_43":32,"_44":32,"_45":32,"_46":32,"_47":32,"_48":32,"_49":50,"_51":50,"_52":32,"_53":-7,"_54":-7,"_55":56,"_57":58,"_59":60,"_61":-5,"_62":63,"_64":65,"_66":67,"_68":69},"routes/repo.blob",{"_13":14,"_15":16,"_17":18,"_19":20,"_21":22,"_23":24,"_25":-5,"_26":27,"_28":29,"_29":30,"_31":32,"_33":14},"ref","main","path","crates/ltx/src/client/object_store.rs","oid","e10329d44d0a7e57ea7d97f5a0c7fe7a56c4d48b","size",54380,"commit","90b43017241f81189453d326d05948f388b34652","mounts",[],"source","plain",{"_13":14,"_15":16},"kind","text","//! client::object_store — S3/R2/MinIO `ReplicaClient` via the `object_store` crate.\n//!\n//! Ported (behavior only) from litestream@v0.5.11 `s3/replica_client.go`. We do\n//! **not** use the Go AWS SDK; instead the five `ReplicaClient` operations are\n//! mapped onto `object_store::ObjectStore` (the `s3` cargo feature wires the\n//! `object_store::aws::AmazonS3Builder` backend). The behavioral invariants from\n//! the upstream conformance suite are kept:\n//!\n//! * key/path scheme `{path}/{level:04x}/{min}-{max}.ltx`\n//! (s3/replica_client.go:629, 677, 1040-1042);\n//! * the 5 MiB single-PUT vs multipart threshold\n//! (s3/replica_client.go:99 + the Go uploader default);\n//! * list + seek-skip on `min_txid < seek`, ascending TXID order\n//! (s3/replica_client.go:1530-1533);\n//! * `NoSuchKey` → `os.ErrNotExist` error mapping\n//! (s3/replica_client.go:647-649, 1662-1668);\n//! * batch DELETE up to 1000 keys per call with per-key error surfacing\n//! (s3/replica_client.go:1028-1101).\n//!\n//! The provider-defaults table (`ParseHost`, path-style flags, endpoint env\n//! var) is a faithful port of `NewReplicaClientFromURL`\n//! (s3/replica_client.go:133-314) so a `s3://…` URL configures the same way.\n//!\n//! This whole module is gated behind `#[cfg(feature = \"s3\")]` because it needs\n//! `object_store`'s AWS backend.\n\n#![cfg(feature = \"s3\")]\n\nuse std::sync::Arc;\n\nuse chrono::{DateTime, SecondsFormat};\nuse futures_util::stream::{StreamExt, TryStreamExt};\nuse object_store::aws::{\n AmazonS3, AmazonS3Builder, AmazonS3ConfigKey, AwsAuthorizer, AwsCredential,\n};\nuse object_store::client::HttpRequestBody;\nuse object_store::path::Path as ObjPath;\nuse object_store::{\n Attribute, AttributeValue, Attributes, ClientOptions, GetOptions, GetRange, ObjectStore,\n PutMultipartOptions, PutOptions, PutPayload, RetryConfig,\n};\n\nuse crate::paged::RangeReader;\n\nuse crate::error::{Error, Result};\nuse crate::ltx::{self, FileInfo};\nuse crate::replica_url::{\n self, bool_query_value, ensure_endpoint_scheme, region_from_s3_arn, ParsedReplicaUrl,\n};\nuse crate::TXID;\n\nuse super::ReplicaClient;\n\n/// The standard Litestream S3 metadata key for an LTX header timestamp.\nconst METADATA_KEY_TIMESTAMP: &str = \"litestream-timestamp\";\n\n/// The retry policy every replica store uses.\n///\n/// object_store defaults to 10 retries under a 180-second ceiling. That suits\n/// a client whose caller has nowhere else to go. A replica store always has\n/// somewhere else to go: a failed segment read is retried by the restore that\n/// asked for it, and a failed upload is retried by the next replication turn,\n/// so a long ladder inside one request buys nothing and delays the answer.\n///\n/// The ceiling matters most on a request a person is waiting for. A read-only\n/// cell inspection reaches this store, and a definite failure under the\n/// default policy takes seconds of pure backoff before the caller learns\n/// anything. Three attempts still absorb an ordinary transient error.\npub fn replica_retry_config() -> RetryConfig {\n RetryConfig {\n max_retries: 3,\n retry_timeout: std::time::Duration::from_secs(30),\n ..RetryConfig::default()\n }\n}\n\n/// The same key with an underscore, for Azure Blob Storage. An Azure blob\n/// metadata name must be a C# identifier, so it cannot hold a hyphen, and\n/// the standard key is refused there.\nconst METADATA_KEY_TIMESTAMP_UNDERSCORE: &str = \"litestream_timestamp\";\n\n/// Which object-metadata name carries the LTX header timestamp.\n///\n/// The name is per backend, not per deployment: Azure Blob Storage\n/// refuses a hyphen in a metadata name, and every other supported store\n/// accepts the standard Litestream key. The value and its format do not\n/// change.\n///\n/// The key exists for external Litestream tooling, which reads\n/// `litestream-timestamp` to do a timestamp restore without downloading\n/// every candidate file. A renamed key therefore ends that interop on\n/// Azure. celld itself never reads the key back.\n#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]\npub enum TimestampMetadataKey {\n /// `litestream-timestamp`, the standard Litestream key.\n #[default]\n Litestream,\n /// `litestream_timestamp`, for a store that refuses a hyphen.\n Underscore,\n}\n\nimpl TimestampMetadataKey {\n /// The metadata name this variant writes.\n pub fn as_str(self) -> &'static str {\n match self {\n Self::Litestream => METADATA_KEY_TIMESTAMP,\n Self::Underscore => METADATA_KEY_TIMESTAMP_UNDERSCORE,\n }\n }\n}\n\n/// Max keys S3 operates on per batch DELETE. `MaxKeys` (s3/replica_client.go:56).\npub const MAX_KEYS: usize = 1000;\n\n/// Region used when none is specified. `DefaultRegion` (s3/replica_client.go:59).\npub const DEFAULT_REGION: &str = \"us-east-1\";\n\n/// Multipart upload threshold: data at or above this size is uploaded with\n/// `put_multipart`; below it, a single `put`. Matches the Go uploader's 5 MiB\n/// `PartSize` default (s3/replica_client.go:99).\npub const MULTIPART_THRESHOLD: usize = 5 * 1024 * 1024;\n\n// ── Config ────────────────────────────────────────────────────────────────────\n\n/// Configuration for the S3/R2/MinIO backend.\n///\n/// Maps to the public fields of Go's `ReplicaClient` struct\n/// (s3/replica_client.go:78-116). Zero/`None` values mean \"use the backend\n/// default\".\n#[derive(Debug, Clone, Default)]\npub struct ObjectStoreConfig {\n /// Bucket name (required).\n pub bucket: String,\n /// Key prefix within the bucket.\n pub path: String,\n /// AWS region.\n pub region: String,\n /// Custom endpoint (MinIO, R2, …); empty = native AWS.\n pub endpoint: String,\n /// Static access key id; empty = ambient credential chain.\n pub access_key_id: String,\n /// Static secret access key; empty = ambient credential chain.\n pub secret_access_key: String,\n /// Session token for temporary/scoped credentials (STS, R2 API tokens);\n /// empty = none. Required alongside temporary keys or signing fails.\n pub session_token: String,\n /// Force path-style addressing (required for MinIO/Backblaze/Supabase/Filebase).\n pub force_path_style: bool,\n /// Skip TLS verification (allows self-signed endpoints).\n pub skip_verify: bool,\n /// Multipart part size in bytes; 0 = default (5 MiB).\n pub part_size: u64,\n /// The metadata name for the LTX header timestamp. The default is the\n /// standard Litestream key; a host that injects an Azure store must\n /// select [`TimestampMetadataKey::Underscore`], because Azure refuses\n /// a hyphen in a metadata name.\n pub timestamp_metadata_key: TimestampMetadataKey,\n}\n\nimpl ObjectStoreConfig {\n /// Construct from a parsed `s3://` URL, mirroring `NewReplicaClientFromURL`\n /// (s3/replica_client.go:133-314): host → bucket/region/endpoint/path-style\n /// (or ARN), query-param overrides (camelCase ↔ hyphenated aliases), the\n /// `AWS_*`/`LITESTREAM_*` env credentials, the `LITESTREAM_S3_ENDPOINT` env\n /// fallback, and the provider-specific path-style defaults for\n /// MinIO/Backblaze/Filebase/Supabase.\n pub fn from_url(parsed: &ParsedReplicaUrl) -> Result {\n let host = &parsed.host;\n let query = &parsed.query;\n\n // Host → bucket/region/endpoint/forcePathStyle (or ARN access point).\n let (bucket, mut region, mut endpoint, mut force_path_style) = if host.starts_with(\"arn:\") {\n (host.clone(), region_from_s3_arn(host), String::new(), false)\n } else {\n parse_host(host)\n };\n\n let q = Some(query);\n\n // endpoint query param: ensure scheme, default to path-style for custom\n // endpoints unless force-path-style is explicitly set to false.\n let q_endpoint = query.get(\"endpoint\");\n if !q_endpoint.is_empty() {\n let (ep, _) = ensure_endpoint_scheme(q_endpoint);\n endpoint = ep;\n match bool_query_value(q, &[\"forcePathStyle\", \"force-path-style\"]) {\n Some(false) => {}\n _ => force_path_style = true,\n }\n }\n let q_region = query.get(\"region\");\n if !q_region.is_empty() {\n region = q_region.to_string();\n }\n if let Some(v) = bool_query_value(q, &[\"forcePathStyle\", \"force-path-style\"]) {\n force_path_style = v;\n }\n let mut skip_verify = false;\n if let Some(v) = bool_query_value(q, &[\"skipVerify\", \"skip-verify\"]) {\n skip_verify = v;\n }\n\n let mut part_size: u64 = 0;\n let v = query.get(\"partSize\");\n let v2 = query.get(\"part-size\");\n if !v.is_empty() {\n if let Ok(n) = v.parse::() {\n if n > 0 {\n part_size = n;\n }\n }\n } else if !v2.is_empty() {\n if let Ok(n) = v2.parse::() {\n if n > 0 {\n part_size = n;\n }\n }\n }\n\n if bucket.is_empty() {\n return Err(Error::Other(\"bucket required for s3 replica URL\".into()));\n }\n\n // Track whether forcePathStyle was explicitly set via query param\n // (s3/replica_client.go:208) — this gates the env-var/provider defaults.\n let force_path_style_set =\n !query.get(\"forcePathStyle\").is_empty() || !query.get(\"force-path-style\").is_empty();\n\n // Static credentials from env (AWS_* preferred, then LITESTREAM_*).\n let mut access_key_id = String::new();\n let mut secret_access_key = String::new();\n if let Some(v) = nonempty_env(\"AWS_ACCESS_KEY_ID\") {\n access_key_id = v;\n } else if let Some(v) = nonempty_env(\"LITESTREAM_ACCESS_KEY_ID\") {\n access_key_id = v;\n }\n if let Some(v) = nonempty_env(\"AWS_SECRET_ACCESS_KEY\") {\n secret_access_key = v;\n } else if let Some(v) = nonempty_env(\"LITESTREAM_SECRET_ACCESS_KEY\") {\n secret_access_key = v;\n }\n let session_token = nonempty_env(\"AWS_SESSION_TOKEN\")\n .or_else(|| nonempty_env(\"LITESTREAM_SESSION_TOKEN\"))\n .unwrap_or_default();\n\n // LITESTREAM_S3_ENDPOINT env fallback (only when no endpoint yet).\n if endpoint.is_empty() {\n if let Some(v) = nonempty_env(\"LITESTREAM_S3_ENDPOINT\") {\n let (ep, _) = ensure_endpoint_scheme(&v);\n endpoint = ep;\n if !force_path_style_set {\n force_path_style = true;\n }\n }\n }\n\n // Provider detection for applying defaults.\n let is_filebase = replica_url::is_filebase_endpoint(&endpoint);\n let is_backblaze = replica_url::is_backblaze_endpoint(&endpoint);\n let is_minio = replica_url::is_minio_endpoint(&endpoint);\n let is_supabase = replica_url::is_supabase_endpoint(&endpoint);\n if !force_path_style_set && (is_filebase || is_backblaze || is_minio || is_supabase) {\n force_path_style = true;\n }\n\n Ok(ObjectStoreConfig {\n bucket,\n path: parsed.path.clone(),\n region,\n endpoint,\n access_key_id,\n secret_access_key,\n session_token,\n force_path_style,\n skip_verify,\n part_size,\n timestamp_metadata_key: TimestampMetadataKey::default(),\n })\n }\n\n /// Build the backing `Arc` for this config. Public so a host\n /// can build one store for a bucket and share it across many\n /// [`ObjectStoreClient::with_store`] clients that differ only by key prefix\n /// — one connection pool for every cell on a node.\n ///\n /// An empty `access_key_id` and `secret_access_key` select the ambient AWS\n /// credential chain, which reads web identity, ECS task credentials, EKS\n /// Pod Identity, and instance metadata from the process environment. Set\n /// both fields to keep a caller's own credentials authoritative.\n pub fn build_store(&self) -> Result> {\n Ok(self.build_s3()?)\n }\n\n /// The concrete S3 store, kept beside the erased one so the paged fault\n /// path can borrow its credential.\n fn build_s3(&self) -> Result> {\n if self.bucket.is_empty() {\n return Err(Error::Other(\"s3: bucket name is required\".into()));\n }\n\n let region = if self.region.is_empty() {\n DEFAULT_REGION.to_string()\n } else {\n self.region.clone()\n };\n\n // Forward the credential inputs that object_store reads. A bare\n // `new()` reads no environment, so it falls through to IMDS on EKS,\n // and `from_env()` would also import endpoint and request settings\n // that belong to this explicit ObjectStoreConfig. So the allowlist\n // stops at the credential boundary, and the config below stays\n // authoritative.\n //\n // object_store 0.12.5 takes AWS_WEB_IDENTITY_TOKEN_FILE and\n // AWS_ROLE_ARN from the process environment rather than from these\n // builder fields, so that pair changes nothing today. It stays because\n // the session name and the STS endpoint beside it *are* read from the\n // builder, and a release that moves the pair to the builder must not\n // drop this client back to IMDS.\n let mut builder = AmazonS3Builder::new();\n for (name, key) in [\n (\n \"AWS_CONTAINER_CREDENTIALS_RELATIVE_URI\",\n AmazonS3ConfigKey::ContainerCredentialsRelativeUri,\n ),\n (\n \"AWS_CONTAINER_CREDENTIALS_FULL_URI\",\n AmazonS3ConfigKey::ContainerCredentialsFullUri,\n ),\n (\n \"AWS_CONTAINER_AUTHORIZATION_TOKEN_FILE\",\n AmazonS3ConfigKey::ContainerAuthorizationTokenFile,\n ),\n (\n \"AWS_WEB_IDENTITY_TOKEN_FILE\",\n AmazonS3ConfigKey::WebIdentityTokenFile,\n ),\n (\"AWS_ROLE_ARN\", AmazonS3ConfigKey::RoleArn),\n (\"AWS_ROLE_SESSION_NAME\", AmazonS3ConfigKey::RoleSessionName),\n (\"AWS_ENDPOINT_URL_STS\", AmazonS3ConfigKey::StsEndpoint),\n (\"AWS_METADATA_ENDPOINT\", AmazonS3ConfigKey::MetadataEndpoint),\n ] {\n if let Some(value) = nonempty_env(name) {\n builder = builder.with_config(key, value);\n }\n }\n let mut builder = builder\n .with_bucket_name(&self.bucket)\n .with_region(region)\n .with_retry(replica_retry_config())\n // Path-style ⇔ NOT virtual-hosted-style (s3/replica_client.go:258-263).\n .with_virtual_hosted_style_request(!self.force_path_style)\n // object_store disables S3 conditional puts by default. Keep the\n // established ETagMatch behavior, so a user of the returned store\n // can send PutMode::Create or PutMode::Update. Ordinary replica\n // PUTs remain unconditional.\n .with_conditional_put(object_store::aws::S3ConditionalPut::ETagMatch);\n\n if !self.endpoint.is_empty() {\n // A plaintext or local endpoint must allow HTTP. skip_verify applies\n // only to TLS certificate validation.\n let allow_http = self.endpoint.starts_with(\"http://\")\n || replica_url::is_local_endpoint(&self.endpoint);\n let client_options = ClientOptions::new()\n .with_allow_http(allow_http)\n .with_allow_invalid_certificates(self.skip_verify);\n builder = builder\n .with_endpoint(&self.endpoint)\n .with_client_options(client_options);\n }\n\n if !self.access_key_id.is_empty() {\n builder = builder.with_access_key_id(&self.access_key_id);\n }\n if !self.secret_access_key.is_empty() {\n builder = builder.with_secret_access_key(&self.secret_access_key);\n }\n if !self.session_token.is_empty() {\n builder = builder.with_token(&self.session_token);\n }\n\n let store = builder\n .build()\n .map_err(|e| Error::Other(format!(\"s3: build store: {e}\").into()))?;\n Ok(Arc::new(store))\n }\n\n /// Effective multipart part size (`part_size`, or the 5 MiB default).\n fn effective_part_size(&self) -> usize {\n if self.part_size > 0 {\n self.part_size as usize\n } else {\n MULTIPART_THRESHOLD\n }\n }\n}\n\n/// Returns `Some(value)` for a non-empty env var, else `None`. Mirrors Go's\n/// `if v := os.Getenv(k); v != \"\"` pattern (s3/replica_client.go:211-224).\nfn nonempty_env(key: &str) -> Option {\n match std::env::var(key) {\n Ok(v) if !v.is_empty() => Some(v),\n _ => None,\n }\n}\n\n// ── ParseHost ─────────────────────────────────────────────────────────────────\n\n/// Parse an S3 host into `(bucket, region, endpoint, force_path_style)`.\n///\n/// Direct port of `ParseHost` (s3/replica_client.go:1608-1652): MinIO-style\n/// `bucket.host:port`, then the AWS / DigitalOcean / Backblaze / Filebase /\n/// Scaleway provider patterns, falling back to \"host *is* the bucket\".\npub fn parse_host(host: &str) -> (String, String, String, bool) {\n // MinIO-style hosts: `bucket.host:port` (a colon and not a \".com\").\n if host.contains(':') && !host.contains(\".com\") {\n // SplitN(host, \".\", 2)\n if let Some((bucket, rest)) = host.split_once('.') {\n return (\n bucket.to_string(),\n DEFAULT_REGION.to_string(),\n format!(\"http://{rest}\"),\n true,\n );\n }\n // No bucket in host, just host:port.\n return (String::new(), String::new(), format!(\"http://{host}\"), true);\n }\n\n // AWS S3: `^(.+)\\.s3(?:\\.([^.]+))?\\.amazonaws\\.com$`\n if let Some((bucket, region)) = match_aws_s3(host) {\n return (bucket, region, String::new(), false);\n }\n // DigitalOcean: `^(?:(.+)\\.)?([^.]+)\\.digitaloceanspaces.com$`\n if let Some((bucket, region)) = match_two_label_suffix(host, \".digitaloceanspaces.com\") {\n return (\n bucket,\n region.clone(),\n format!(\"https://{region}.digitaloceanspaces.com\"),\n false,\n );\n }\n // Backblaze: `^(?:(.+)\\.)?s3.([^.]+)\\.backblazeb2.com$`\n if let Some((bucket, region)) = match_s3_region_suffix(host, \".backblazeb2.com\") {\n return (\n bucket,\n region.clone(),\n format!(\"https://s3.{region}.backblazeb2.com\"),\n true,\n );\n }\n // Filebase: `^(?:(.+)\\.)?s3.filebase.com$`\n if let Some(bucket) = match_filebase(host) {\n return (bucket, String::new(), \"s3.filebase.com\".to_string(), false);\n }\n // Scaleway: `^(?:(.+)\\.)?s3.([^.]+)\\.scw\\.cloud$`\n if let Some((bucket, region)) = match_s3_region_suffix(host, \".scw.cloud\") {\n return (\n bucket,\n region.clone(),\n format!(\"s3.{region}.scw.cloud\"),\n false,\n );\n }\n\n // Standard S3: the host is the bucket name.\n (host.to_string(), String::new(), String::new(), false)\n}\n\n/// `^(.+)\\.s3(?:\\.([^.]+))?\\.amazonaws\\.com$` → (bucket, region).\nfn match_aws_s3(host: &str) -> Option<(String, String)> {\n let rest = host.strip_suffix(\".amazonaws.com\")?;\n // rest = \".s3\" or \".s3.\"\n if let Some(bucket) = rest.strip_suffix(\".s3\") {\n if bucket.is_empty() {\n return None;\n }\n return Some((bucket.to_string(), String::new()));\n }\n // \".s3.\": find the \".s3.\" separator; region is a single\n // label ([^.]+) — i.e. the remainder after \".s3.\" must contain no dot.\n let idx = rest.find(\".s3.\")?;\n let bucket = &rest[..idx];\n let region = &rest[idx + 4..];\n if bucket.is_empty() || region.is_empty() || region.contains('.') {\n return None;\n }\n Some((bucket.to_string(), region.to_string()))\n}\n\n/// `^(?:(.+)\\.)?([^.]+)\\.$` → (bucket, region). `suffix` starts with '.'.\nfn match_two_label_suffix(host: &str, suffix: &str) -> Option<(String, String)> {\n let rest = host.strip_suffix(suffix)?;\n if rest.is_empty() {\n return None;\n }\n // The last label before the suffix is the region; anything before it\n // (optionally) is the bucket.\n match rest.rfind('.') {\n Some(i) => {\n let bucket = &rest[..i];\n let region = &rest[i + 1..];\n if region.is_empty() {\n return None;\n }\n Some((bucket.to_string(), region.to_string()))\n }\n None => Some((String::new(), rest.to_string())),\n }\n}\n\n/// `^(?:(.+)\\.)?s3.([^.]+)\\.$` → (bucket, region). `suffix` starts with '.'.\nfn match_s3_region_suffix(host: &str, suffix: &str) -> Option<(String, String)> {\n let rest = host.strip_suffix(suffix)?;\n // rest = \"[bucket.]s3.\"; region is one label ([^.]+).\n // Bucket-less form: rest == \"s3.\".\n if let Some(region) = rest.strip_prefix(\"s3.\") {\n if region.is_empty() || region.contains('.') {\n return None;\n }\n return Some((String::new(), region.to_string()));\n }\n // Bucketed form: rest == \".s3.\".\n let sep = rest.find(\".s3.\")?;\n let bucket = &rest[..sep];\n let region = &rest[sep + 4..];\n if bucket.is_empty() || region.is_empty() || region.contains('.') {\n return None;\n }\n Some((bucket.to_string(), region.to_string()))\n}\n\n/// `^(?:(.+)\\.)?s3.filebase.com$` → bucket.\nfn match_filebase(host: &str) -> Option {\n if host == \"s3.filebase.com\" {\n return Some(String::new());\n }\n let bucket = host.strip_suffix(\".s3.filebase.com\")?;\n if bucket.is_empty() {\n None\n } else {\n Some(bucket.to_string())\n }\n}\n\n// ── Client ────────────────────────────────────────────────────────────────────\n\n/// Concrete S3/R2/MinIO backend, wrapping a lazily-initialised\n/// `Arc`. The config-driven path builds an S3 store, but\n/// [`Self::with_store`] accepts any prebuilt `ObjectStore` — the five\n/// replica operations are provider-neutral, and a host can inject e.g. a\n/// GCS or in-memory store.\n///\n/// Mirrors Go `ReplicaClient` (s3/replica_client.go:78-116). The inner store is\n/// created on the first call that needs it (`OnceCell`, mirroring `Init`,\n/// s3/replica_client.go:322-477), so construction is infallible and race-free.\npub struct ObjectStoreClient {\n store: tokio::sync::OnceCell>,\n /// The config's S3 store, initialized for replica I/O or when a shared\n /// store's paged reader needs an ambient credential provider.\n s3: tokio::sync::OnceCell>,\n config: ObjectStoreConfig,\n}\n\nimpl std::fmt::Debug for ObjectStoreClient {\n fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {\n f.debug_struct(\"ObjectStoreClient\")\n .field(\"config\", &self.config)\n .field(\"initialized\", &self.store.initialized())\n .finish()\n }\n}\n\nimpl ObjectStoreClient {\n /// Create a client from config (no I/O; the store is built on first use).\n pub fn new(config: ObjectStoreConfig) -> Self {\n ObjectStoreClient {\n store: tokio::sync::OnceCell::new(),\n s3: tokio::sync::OnceCell::new(),\n config,\n }\n }\n\n /// Create a client directly from an already-built `ObjectStore`, such as an\n /// in-memory store or a pre-configured backend.\n pub fn with_store(config: ObjectStoreConfig, store: Arc) -> Self {\n let cell = tokio::sync::OnceCell::new();\n cell.set(store).ok();\n ObjectStoreClient {\n store: cell,\n s3: tokio::sync::OnceCell::new(),\n config,\n }\n }\n\n /// Get-or-build the inner store, once.\n async fn store(&self) -> Result<&Arc> {\n self.store\n .get_or_try_init(|| async { Ok(self.s3().await?.clone() as Arc) })\n .await\n }\n\n async fn s3(&self) -> Result<&Arc> {\n self.s3\n .get_or_try_init(|| async { self.config.build_s3() })\n .await\n }\n\n /// A blocking ranged reader over this client's objects for the paged\n /// VFS. It snapshots the store's credential now (an async fetch on the\n /// caller's runtime) so each fault can sign its own request on whatever\n /// thread SQLite is on, with no runtime in the loop.\n pub async fn blocking_range_reader(&self) -> Result> {\n let config = &self.config;\n // Only native S3 is read by a signed request of our own; any other\n // store — the deterministic simulation's, a directory, memory —\n // is read through the store on a helper thread.\n let store = self.store().await?;\n if !store.to_string().starts_with(\"AmazonS3\") {\n return Ok(Box::new(StoreRangeReader::new(\n store.clone(),\n config.path.clone(),\n )));\n }\n let mut provider = None;\n let credential = if !config.access_key_id.is_empty() {\n Arc::new(AwsCredential {\n key_id: config.access_key_id.clone(),\n secret_key: config.secret_access_key.clone(),\n token: (!config.session_token.is_empty()).then(|| config.session_token.clone()),\n })\n } else {\n // A shared store erases AmazonS3's credential provider. Waiting\n // for replica I/O to install it leaves every paged restore with\n // ambient credentials failing. Build it from the same config on\n // demand; retain its provider so expired role credentials can\n // refresh during page faults. Static keys need no second store.\n let s3 = self.s3().await?;\n provider = Some(s3.credentials().clone());\n s3.credentials()\n .get_credential()\n .await\n .map_err(map_os_error)?\n };\n let region = if config.region.is_empty() {\n DEFAULT_REGION.to_string()\n } else {\n config.region.clone()\n };\n // Mirror the store's own addressing so the signed host matches what\n // the bucket expects: virtual-hosted unless path-style was forced.\n let base = if config.endpoint.is_empty() {\n format!(\"https://{}.s3.{region}.amazonaws.com\", config.bucket)\n } else if config.force_path_style {\n format!(\n \"{}/{}\",\n config.endpoint.trim_end_matches('/'),\n config.bucket\n )\n } else {\n let (scheme, host) = config\n .endpoint\n .split_once(\"://\")\n .ok_or_else(|| Error::Other(\"s3: endpoint has no scheme\".into()))?;\n format!(\n \"{scheme}://{}.{}\",\n config.bucket,\n host.trim_end_matches('/')\n )\n };\n let tls = ureq::tls::TlsConfig::builder()\n .disable_verification(config.skip_verify)\n .build();\n // A fault runs on the caller's thread — for the actor's queries the\n // node's core — so this budget is how long one slow object read can\n // stall every cell on the node. A normal fault is 65-200 ms; 10 s is\n // fifty times that, and the single retry below bounds a stall at\n // about 20 s instead of the 90 s three 30 s attempts allowed.\n let agent = ureq::Agent::config_builder()\n .timeout_global(Some(FAULT_TIMEOUT))\n .http_status_as_error(false)\n .tls_config(tls)\n .build()\n .new_agent();\n Ok(Box::new(S3RangeReader {\n credential: std::sync::RwLock::new(credential),\n provider,\n region,\n base,\n prefix: config.path.clone(),\n agent,\n }))\n }\n\n /// Returns whether the replica epoch prefix contains an object.\n ///\n /// The caller needs only an existence proof, so it stops the recursive,\n /// delimiter-less listing after the first object. An empty or non-empty\n /// prefix therefore costs one list request instead of one request for each\n /// compaction level.\n pub async fn has_any_object(&self) -> Result {\n let store = self.store().await?;\n let prefix = ObjPath::from(self.root_prefix());\n store\n .list(Some(&prefix))\n .next()\n .await\n .transpose()\n .map(|object| object.is_some())\n .map_err(map_os_error)\n }\n\n /// Build the S3 key for an LTX file: `{path}/{level:04x}/{min}-{max}.ltx`.\n /// Ported from s3/replica_client.go:629, 677, 1040-1042.\n fn ltx_key(&self, level: i32, min_txid: TXID, max_txid: TXID) -> String {\n let filename = ltx::format_filename(min_txid, max_txid);\n format!(\"{}/{:04x}/{}\", self.config.path, level, filename)\n }\n\n /// Prefix for listing a level: `{path}/{level:04x}/`.\n /// Ported from s3/replica_client.go:1363.\n fn level_prefix(&self, level: i32) -> String {\n format!(\"{}/{:04x}/\", self.config.path, level)\n }\n\n /// Root prefix for delete-all: `{path}/`. (s3/replica_client.go:1114).\n fn root_prefix(&self) -> String {\n format!(\"{}/\", self.config.path)\n }\n\n /// The highest transaction any level holds, from one listing of the\n /// whole path. A caller that only needs the watermark, such as a log\n /// recovery deciding which gathered rows the bucket already covers, pays\n /// one round trip here instead of one per level.\n pub async fn max_txid_all_levels(&self) -> Result {\n let store = self.store().await?;\n let prefix = ObjPath::from(self.root_prefix());\n let mut listed = store.list(Some(&prefix));\n let mut max = TXID(0);\n while let Some(meta) = listed.try_next().await.map_err(map_os_error)? {\n let name = meta.location.filename().unwrap_or(\"\");\n if let Ok((_, max_txid)) = ltx::parse_filename(name) {\n max = max.max(max_txid);\n }\n }\n Ok(max)\n }\n}\n\n/// Map an `object_store::Error` to `crate::Error`, preserving NotFound as\n/// `io::ErrorKind::NotFound` so callers keep working with the std error kind.\n/// Mirrors `isNotExists` → `os.ErrNotExist` (s3/replica_client.go:647-649,\n/// 1662-1668).\nfn map_os_error(e: object_store::Error) -> Error {\n match e {\n object_store::Error::NotFound { .. } => {\n Error::Io(std::io::Error::new(std::io::ErrorKind::NotFound, e))\n }\n other => Error::Other(Box::new(other)),\n }\n}\n\n#[async_trait::async_trait]\nimpl ReplicaClient for ObjectStoreClient {\n async fn ltx_files(&self, level: i32, seek: TXID) -> Result> {\n self.ltx_files_bounded(level, seek, usize::MAX).await\n }\n\n async fn ltx_files_bounded(\n &self,\n level: i32,\n seek: TXID,\n limit: usize,\n ) -> Result> {\n if limit == 0 {\n return Ok(Vec::new());\n }\n let store = self.store().await?;\n let prefix = ObjPath::from(self.level_prefix(level));\n // The offset ends immediately before every filename with this minimum\n // TXID. S3 and GCS can push it into the listing request, so an additive\n // compaction does not scan the complete retained L0 history. The\n // supported production stores return each page in lexical order. The\n // compactor validates continuity and fails closed for an unordered\n // custom store.\n let offset = ObjPath::from(format!(\"{}{:016x}\", self.level_prefix(level), seek.0));\n let mut listed = store.list_with_offset(Some(&prefix), &offset);\n let mut infos = Vec::with_capacity(limit.min(256));\n while let Some(meta) = listed.try_next().await.map_err(map_os_error)? {\n let name = meta.location.filename().unwrap_or(\"\");\n let (min_txid, max_txid) = match ltx::parse_filename(name) {\n Ok(t) => t,\n Err(_) => continue, // skip non-LTX keys\n };\n if min_txid < seek {\n continue;\n }\n infos.push(FileInfo {\n level,\n min_txid,\n max_txid,\n size: meta.size as i64,\n created_at: Some(std::time::SystemTime::from(meta.last_modified)),\n ..Default::default()\n });\n if infos.len() == limit {\n break;\n }\n }\n\n // Iterator contract: ascending by (level, min_txid, max_txid).\n infos.sort_by(|a, b| {\n (a.level, a.min_txid.0, a.max_txid.0).cmp(&(b.level, b.min_txid.0, b.max_txid.0))\n });\n Ok(infos)\n }\n\n async fn open_ltx_file(&self, level: i32, min_txid: TXID, max_txid: TXID) -> Result> {\n let store = self.store().await?;\n let key = ObjPath::from(self.ltx_key(level, min_txid, max_txid));\n\n let result = match store.get(&key).await {\n Ok(r) => r,\n Err(object_store::Error::NotFound { .. }) => {\n return Err(Error::Io(std::io::Error::new(\n std::io::ErrorKind::NotFound,\n format!(\"replica: get object {key}: not found\"),\n )));\n }\n Err(e) => return Err(map_os_error(e)),\n };\n\n let bytes = result.bytes().await.map_err(map_os_error)?;\n Ok(bytes.to_vec())\n }\n\n async fn read_range(\n &self,\n level: i32,\n min_txid: TXID,\n max_txid: TXID,\n offset: u64,\n len: u64,\n ) -> Result> {\n let store = self.store().await?;\n let key = ObjPath::from(self.ltx_key(level, min_txid, max_txid));\n let options = GetOptions {\n range: Some(GetRange::Bounded(offset..offset.saturating_add(len))),\n ..Default::default()\n };\n let result = match store.get_opts(&key, options).await {\n Ok(r) => r,\n Err(object_store::Error::NotFound { .. }) => {\n return Err(Error::Io(std::io::Error::new(\n std::io::ErrorKind::NotFound,\n format!(\"replica: get range {key}: not found\"),\n )));\n }\n Err(e) => return Err(map_os_error(e)),\n };\n let bytes = result.bytes().await.map_err(map_os_error)?;\n Ok(bytes.to_vec())\n }\n\n async fn write_ltx_file(\n &self,\n level: i32,\n min_txid: TXID,\n max_txid: TXID,\n data: &[u8],\n ) -> Result {\n let store = self.store().await?;\n\n // Preserve the LTX header timestamp in the standard Litestream object\n // metadata. This costs no extra request and lets Litestream perform an\n // accurate timestamp restore without downloading every candidate file.\n // The name is per backend, because Azure refuses a hyphen in it.\n let header = ltx::Header::parse(data)?;\n let created_at = std::time::UNIX_EPOCH\n + std::time::Duration::from_millis(header.timestamp.max(0) as u64);\n let mut attributes = Attributes::new();\n attributes.insert(\n Attribute::Metadata(self.config.timestamp_metadata_key.as_str().into()),\n AttributeValue::from(format_rfc3339_nano(header.timestamp)?),\n );\n let key = ObjPath::from(self.ltx_key(level, min_txid, max_txid));\n\n // Multipart threshold: < 5 MiB → single PUT; ≥ 5 MiB → multipart with\n // fixed-size parts. Ported from the Go uploader's 5 MiB PartSize default\n // (s3/replica_client.go:99, brief §5.1).\n let part_size = self.config.effective_part_size();\n if data.len() < MULTIPART_THRESHOLD {\n let payload = PutPayload::from(data.to_vec());\n let options = PutOptions {\n attributes,\n ..Default::default()\n };\n store\n .put_opts(&key, payload, options)\n .await\n .map_err(|e| Error::Other(format!(\"replica: upload to {key}: {e}\").into()))?;\n } else {\n let options = PutMultipartOptions {\n attributes,\n ..Default::default()\n };\n let mut upload = store\n .put_multipart_opts(&key, options)\n .await\n .map_err(|e| Error::Other(format!(\"replica: upload to {key}: {e}\").into()))?;\n // Keep every fallible step inside one result. Otherwise, a new\n // early return can leave completed parts in the storage service.\n let upload_result: Result<()> = async {\n // Upload in fixed-size parts (each ≥ 5 MiB except possibly the\n // last, matching object_store's part-size requirement).\n for chunk in data.chunks(part_size.max(MULTIPART_THRESHOLD)) {\n upload\n .put_part(PutPayload::from(chunk.to_vec()))\n .await\n .map_err(|e| {\n Error::Other(format!(\"replica: upload part to {key}: {e}\").into())\n })?;\n }\n upload.complete().await.map_err(|e| {\n Error::Other(format!(\"replica: complete upload to {key}: {e}\").into())\n })?;\n Ok(())\n }\n .await;\n if let Err(upload_error) = upload_result {\n if let Err(abort_error) = upload.abort().await {\n tracing::warn!(\n %abort_error,\n key = %key,\n \"failed LTX multipart upload could not be aborted\"\n );\n }\n return Err(upload_error);\n }\n }\n\n Ok(FileInfo {\n level,\n min_txid,\n max_txid,\n size: data.len() as i64,\n created_at: Some(created_at),\n ..Default::default()\n })\n }\n\n async fn write_ltx_file_from_file(\n &self,\n level: i32,\n min_txid: TXID,\n max_txid: TXID,\n file: crate::host::HostFile,\n host: crate::LtxHost,\n ) -> Result {\n let part_size = self.config.effective_part_size().max(MULTIPART_THRESHOLD);\n let (mut file, first, size) = super::read_upload_chunk(file, &host, 0, part_size).await?;\n if size < MULTIPART_THRESHOLD as u64 {\n return self.write_ltx_file(level, min_txid, max_txid, &first).await;\n }\n let header = ltx::Header::parse(&first)?;\n let mut attributes = Attributes::new();\n attributes.insert(\n Attribute::Metadata(self.config.timestamp_metadata_key.as_str().into()),\n AttributeValue::from(format_rfc3339_nano(header.timestamp)?),\n );\n let store = self.store().await?;\n let key = ObjPath::from(self.ltx_key(level, min_txid, max_txid));\n let mut upload = store\n .put_multipart_opts(\n &key,\n PutMultipartOptions {\n attributes,\n ..Default::default()\n },\n )\n .await\n .map_err(map_os_error)?;\n let upload_result: Result<()> = async {\n let mut offset = first.len() as u64;\n upload\n .put_part(PutPayload::from(first))\n .await\n .map_err(map_os_error)?;\n while offset < size {\n let (next_file, bytes, _) =\n super::read_upload_chunk(file, &host, offset, part_size).await?;\n file = next_file;\n offset += bytes.len() as u64;\n upload\n .put_part(PutPayload::from(bytes))\n .await\n .map_err(map_os_error)?;\n }\n upload.complete().await.map_err(map_os_error)?;\n Ok(())\n }\n .await;\n if let Err(error) = upload_result {\n if let Err(abort_error) = upload.abort().await {\n tracing::warn!(%abort_error, %key, \"failed LTX multipart upload could not be aborted\");\n }\n return Err(error);\n }\n Ok(FileInfo {\n level,\n min_txid,\n max_txid,\n size: size as i64,\n created_at: Some(\n std::time::UNIX_EPOCH\n + std::time::Duration::from_millis(header.timestamp.max(0) as u64),\n ),\n ..Default::default()\n })\n }\n\n async fn delete_ltx_files(&self, files: &[FileInfo]) -> Result<()> {\n if files.is_empty() {\n return Ok(());\n }\n let store = self.store().await?;\n\n // Build the key list, then delete in batches of MAX_KEYS via the\n // store's delete_stream, surfacing per-key errors (brief §5.5).\n let keys: Vec = files\n .iter()\n .map(|info| ObjPath::from(self.ltx_key(info.level, info.min_txid, info.max_txid)))\n .collect();\n\n for batch in keys.chunks(MAX_KEYS) {\n delete_batch(store.as_ref(), batch, /*ignore_missing=*/ true).await?;\n }\n Ok(())\n }\n\n async fn delete_all(&self) -> Result<()> {\n let store = self.store().await?;\n let prefix = ObjPath::from(self.root_prefix());\n\n // List everything under the path prefix, then batch-delete.\n // (s3/replica_client.go:1104-1148).\n let keys: Vec = store\n .list(Some(&prefix))\n .map_ok(|m| m.location)\n .map_err(map_os_error)\n .try_collect()\n .await?;\n\n for batch in keys.chunks(MAX_KEYS) {\n delete_batch(store.as_ref(), batch, /*ignore_missing=*/ true).await?;\n }\n Ok(())\n }\n}\n\n/// Delete a batch of keys via `delete_stream`, surfacing per-key errors.\n///\n/// When `ignore_missing` is set, `NotFound` is tolerated (delete is idempotent —\n/// the file client swallows ENOENT the same way), but every other per-key error\n/// is returned (brief §5.5: do not silently swallow partial failures).\nasync fn delete_batch(\n store: &dyn ObjectStore,\n keys: &[ObjPath],\n ignore_missing: bool,\n) -> Result<()> {\n if keys.is_empty() {\n return Ok(());\n }\n let owned: Vec = keys.to_vec();\n let stream = futures_util::stream::iter(owned.into_iter().map(Ok));\n let mut results = store.delete_stream(stream.boxed());\n while let Some(res) = results.next().await {\n match res {\n Ok(_) => {}\n Err(object_store::Error::NotFound { .. }) if ignore_missing => {}\n Err(e) => return Err(map_os_error(e)),\n }\n }\n Ok(())\n}\n\n/// Format a Unix-millisecond timestamp like Go's `time.RFC3339Nano`.\nfn format_rfc3339_nano(unix_millis: i64) -> Result {\n let mut timestamp = DateTime::from_timestamp_millis(unix_millis.max(0))\n .ok_or_else(|| Error::Other(\"LTX timestamp is outside the RFC3339 range\".into()))?\n .to_rfc3339_opts(SecondsFormat::AutoSi, true);\n\n // Chrono keeps millisecond precision as three digits. Go removes trailing\n // zeros, so `.500Z` becomes `.5Z`.\n if timestamp.contains('.') {\n // The pop must stay outside debug_assert!, which vanishes in release\n // builds and would leave the Z in place to be doubled below.\n let suffix = timestamp.pop();\n debug_assert_eq!(suffix, Some('Z'));\n while timestamp.ends_with('0') {\n timestamp.pop();\n }\n timestamp.push('Z');\n }\n Ok(timestamp)\n}\n\n#[doc(hidden)]\npub mod internal {\n use super::*;\n\n pub const METADATA_KEY_TIMESTAMP: &str = super::METADATA_KEY_TIMESTAMP;\n pub const METADATA_KEY_TIMESTAMP_UNDERSCORE: &str = super::METADATA_KEY_TIMESTAMP_UNDERSCORE;\n\n pub fn format_rfc3339_nano(unix_millis: i64) -> Result {\n super::format_rfc3339_nano(unix_millis)\n }\n\n pub fn ltx_key(client: &ObjectStoreClient, level: i32, min: TXID, max: TXID) -> String {\n client.ltx_key(level, min, max)\n }\n\n pub fn level_prefix(client: &ObjectStoreClient, level: i32) -> String {\n client.level_prefix(level)\n }\n\n pub fn root_prefix(client: &ObjectStoreClient) -> String {\n client.root_prefix()\n }\n\n pub fn map_os_error(error: object_store::Error) -> Error {\n super::map_os_error(error)\n }\n}\n\n/// Bytes the SigV4 canonical URI leaves unencoded, plus the segment\n/// delimiter: A-Z a-z 0-9 `-` `.` `_` `~` and `/`.\nconst KEY_ENCODE_SET: &percent_encoding::AsciiSet = &percent_encoding::NON_ALPHANUMERIC\n .remove(b'-')\n .remove(b'.')\n .remove(b'_')\n .remove(b'~')\n .remove(b'/');\n\n/// One ranged read's budget on the fault path (see the agent below).\nconst FAULT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);\n\n/// The paged VFS's fault-path reader: a SigV4-signed ranged GET over a\n/// blocking socket. See [`RangeReader`] for why it must not touch a runtime.\n///\n/// The credential is the snapshot taken at activation; a credential that\n/// rotates during a long residency makes later faults fail with 403, which\n/// surfaces as an I/O error on that read. Refreshing it needs the async\n/// provider and belongs to a re-activation, not to a fault.\n/// The fault path over any [`ObjectStore`] that is not native S3: the\n/// deterministic simulation's store, a directory, a memory store. Reads run\n/// on one helper thread with its own runtime; the calling thread blocks on\n/// the reply, which is the fault path's contract. The helper is a thread,\n/// not a task on any caller's runtime, so the same-runtime deadlocks that\n/// sank every bridge on the fleet do not apply (`refresh_aws_credential`\n/// takes the same shape).\npub struct StoreRangeReader {\n requests: std::sync::mpsc::Sender,\n prefix: String,\n}\n\nstruct RangeRequest {\n key: String,\n offset: u64,\n len: u64,\n reply: std::sync::mpsc::Sender>>,\n}\n\nimpl StoreRangeReader {\n pub fn new(store: Arc, prefix: String) -> Self {\n let (requests, inbox) = std::sync::mpsc::channel::();\n std::thread::Builder::new()\n .name(\"paged-store-reader\".into())\n .spawn(move || {\n let runtime = tokio::runtime::Builder::new_current_thread()\n .enable_all()\n .build()\n .expect(\"paged store reader runtime\");\n for request in inbox {\n let path = ObjPath::from(request.key);\n let range = request.offset..request.offset + request.len;\n let out = runtime\n .block_on(store.get_range(&path, range))\n .map(|bytes| bytes.to_vec())\n .map_err(map_os_error);\n let _ = request.reply.send(out);\n }\n })\n .expect(\"spawn the paged store reader\");\n Self { requests, prefix }\n }\n}\n\nimpl RangeReader for StoreRangeReader {\n fn read_range(&self, info: &FileInfo, offset: u64, len: u64) -> Result> {\n if len == 0 {\n return Ok(Vec::new());\n }\n let filename = ltx::format_filename(info.min_txid, info.max_txid);\n let key = format!(\"{}/{:04x}/{filename}\", self.prefix, info.level);\n let (reply, inbox) = std::sync::mpsc::channel();\n let stopped = || Error::Other(\"the paged store reader stopped\".into());\n self.requests\n .send(RangeRequest {\n key,\n offset,\n len,\n reply,\n })\n .map_err(|_| stopped())?;\n inbox.recv().map_err(|_| stopped())?\n }\n}\n\npub struct S3RangeReader {\n credential: std::sync::RwLock>,\n /// The store's provider when the credential came from an ambient chain\n /// (STS, instance roles), which rotates; `None` for static keys.\n provider: Option,\n region: String,\n base: String,\n prefix: String,\n agent: ureq::Agent,\n}\n\nimpl S3RangeReader {\n fn get(&self, url: &str, offset: u64, len: u64) -> Result> {\n let mut signed = http::Request::builder()\n .method(\"GET\")\n .uri(url)\n .header(\"range\", format!(\"bytes={offset}-{}\", offset + len - 1))\n .body(HttpRequestBody::empty())\n .map_err(|e| Error::Other(format!(\"paged: build request: {e}\").into()))?;\n let credential = self.credential.read().unwrap().clone();\n AwsAuthorizer::new(&credential, \"s3\", &self.region).authorize(&mut signed, None);\n let mut request = self.agent.get(url);\n for (name, value) in signed.headers() {\n // ureq sets Host from the URL itself, identically to the signed\n // value; setting it again is refused.\n if name != http::header::HOST {\n let value = value\n .to_str()\n .map_err(|e| Error::Other(format!(\"paged: header: {e}\").into()))?;\n request = request.header(name.as_str(), value);\n }\n }\n let mut response = request\n .call()\n .map_err(|e| Error::Io(std::io::Error::other(format!(\"paged get: {e}\"))))?;\n if response.status() == http::StatusCode::FORBIDDEN {\n return Err(Error::Forbidden);\n }\n if response.status() != http::StatusCode::PARTIAL_CONTENT {\n // S3's error body names the code and, for a signature mismatch,\n // the canonical request it computed — the only way to see it.\n let status = response.status();\n let detail = response\n .body_mut()\n .with_config()\n .limit(8192)\n .read_to_string()\n .unwrap_or_default();\n // Not transient: the retry below is for transport errors only.\n return Err(Error::Other(\n format!(\"paged get: {status} for range {offset}+{len} of {url}: {detail}\").into(),\n ));\n }\n let body = response\n .body_mut()\n .with_config()\n // ureq's limit is exclusive; the length check below is the real\n // bound.\n .limit(len + 1)\n .read_to_vec()\n .map_err(|e| Error::Io(std::io::Error::other(format!(\"paged body: {e}\"))))?;\n if body.len() as u64 != len {\n return Err(Error::Io(std::io::Error::other(format!(\n \"paged get: short range body {} of {len}\",\n body.len()\n ))));\n }\n Ok(body)\n }\n}\n\nimpl RangeReader for S3RangeReader {\n fn read_range(&self, info: &FileInfo, offset: u64, len: u64) -> Result> {\n if len == 0 {\n return Ok(Vec::new());\n }\n let filename = ltx::format_filename(info.min_txid, info.max_txid);\n // SigV4's canonical URI percent-encodes every byte outside the\n // unreserved set — a `:` in a cell name is `%3A` — and R2 canonicalizes\n // what it receives. Signing the raw path therefore signs a different\n // string than the bucket checks (every fault a 403), so the path is\n // encoded before it is signed, and sent encoded.\n let key = format!(\"{}/{:04x}/{filename}\", self.prefix, info.level);\n let key = percent_encoding::utf8_percent_encode(&key, KEY_ENCODE_SET);\n let url = format!(\"{}/{key}\", self.base);\n // A transport hiccup on a 4KiB read is cheap to retry; a status\n // failure is not transient and returns at once.\n let mut attempt = 0;\n let mut refreshed = false;\n loop {\n match self.get(&url, offset, len) {\n Ok(body) => return Ok(body),\n // A rotated ambient credential fails with 403; take a fresh\n // one from the provider once, then treat a second 403 as\n // final.\n Err(Error::Forbidden) if !refreshed && self.provider.is_some() => {\n refreshed = true;\n match refresh_aws_credential(self.provider.as_ref().unwrap()) {\n Some(fresh) => *self.credential.write().unwrap() = fresh,\n None => return Err(Error::Forbidden),\n }\n }\n Err(error) if attempt < 1 && matches!(error, Error::Io(_)) => {\n attempt += 1;\n std::thread::sleep(std::time::Duration::from_millis(100 * attempt));\n }\n Err(error) => return Err(error),\n }\n }\n }\n}\n\n/// Fetches a fresh credential from an ambient provider, from a thread that\n/// is not on any runtime: the fault path runs on whatever thread SQLite is\n/// on, which may be a runtime worker where `block_on` panics, so the async\n/// provider is driven on a helper thread with its own runtime.\npub fn refresh_aws_credential(\n provider: &object_store::aws::AwsCredentialProvider,\n) -> Option> {\n let provider = provider.clone();\n std::thread::spawn(move || {\n let runtime = tokio::runtime::Builder::new_current_thread()\n .enable_all()\n .build()\n .ok()?;\n runtime.block_on(provider.get_credential()).ok()\n })\n .join()\n .ok()\n .flatten()\n}\n","editable",false,"branch","documentNavigation","repo","denoland/celld","repoId","a23b8a10-92cf-4dc4-a7ab-a04e58298dde","ownerId","org:42048915","site","nexthub","private","admin","settings","maintainer","triage","starred","stars",1,"watchers","watching","pinned","notify","views",[134,135,136,137,138,139,140,141,142,143,144,145,146],"navigation",{"_55":56,"_91":92,"_93":94,"_95":96,"_97":98,"_99":100,"_101":102},"sources",[],"viewer","meta",{"_83":84,"_85":86,"_87":88,"_43":32,"_89":14,"_90":32,"_49":50,"_51":50},"branches",[14],"tags",[70,71,72,73,74,75,76,77,78,79,80,81,82],"cloneUrl","https://git.nexthub.sh/denoland/celld.git","v0.0.1","v0.0.2","v0.1.0","v0.2.0","v0.2.1","v0.3.0","v0.4.0","v0.4.1","v0.5.0","v0.5.1","v0.6.0","v0.6.1","v0.6.2","description","self-hosted, distributed Durable Objects","owner","denoland","section","","head","empty","pins",[116,117,118],"documents",{"_111":112,"_113":114},"code",{"_108":109,"_110":104},"automations",{"_106":50,"_107":-5},"delivery",{"_103":104,"_105":-5},"tag","7gTA52Z9a1vO6FP6eCQPix9zO4O","releases",0,"applications","workflows","deployments","commits",13,"changes","counts",{"_115":114},"total",38,"issue",{"_119":110,"_121":132,"_15":110,"_123":133,"_53":125,"_126":32,"_127":125,"_128":104},{"_119":129,"_121":130,"_15":129,"_123":131,"_53":125,"_126":32,"_127":125,"_128":104},{"_119":120,"_121":122,"_15":120,"_123":124,"_53":125,"_126":32,"_127":125,"_128":114},"id","issues","name","Issues","filters",{},true,"saved","pinnable","count","tasks","Tasks",{},"Changes",{},{"_119":110,"_121":132,"_15":110,"_123":177,"_53":125,"_126":32,"_127":125},{"_119":129,"_121":130,"_15":129,"_123":176,"_53":125,"_126":32,"_127":125},{"_119":120,"_121":122,"_15":120,"_123":175,"_53":125,"_126":32,"_127":125},{"_119":172,"_121":173,"_15":172,"_123":174,"_53":32,"_126":32,"_127":125},{"_119":169,"_121":170,"_15":169,"_123":171,"_53":32,"_126":32,"_127":125},{"_119":106,"_121":167,"_15":97,"_123":168,"_53":32,"_126":32,"_127":125},{"_119":107,"_121":164,"_15":165,"_123":166,"_53":32,"_126":32,"_127":125},{"_119":103,"_121":162,"_15":103,"_123":163,"_53":32,"_126":32,"_127":125},{"_119":158,"_121":159,"_15":160,"_123":161,"_53":32,"_126":32,"_127":125},{"_119":93,"_121":156,"_15":93,"_123":157,"_53":32,"_126":32,"_127":32},{"_119":153,"_121":154,"_15":153,"_123":155,"_53":32,"_126":32,"_127":32},{"_119":150,"_121":151,"_15":150,"_123":152,"_53":32,"_126":32,"_127":32},{"_119":147,"_121":148,"_15":147,"_123":149,"_53":32,"_126":32,"_127":32},"decisions","Decisions",{},"changelog","Changelog",{},"roadmap","Roadmap",{},"Documents",{},"activity","Activity","insights",{},"Releases",{},"Deployments","automations/deployments",{},"Workflows",{},"discussions","Discussions",{},"goals","Goals",{},{},{},{},"groups","user","agents",[],"now",1794455315201,"sidebar","expanded"]