boatramp_node/
blobs.rs

1//! Blob (object-store) backend construction: build the configured object store
2//! (fs/S3/GCS/Azure, with optional blob-change notification provisioning) from a
3//! resolved [`BlobArgs`]. Each cloud backend is feature-gated; a disabled one
4//! returns an explanatory error rather than a misleading no-op. Moved out of the
5//! binary (node-library N2b.2c); the binary populates `BlobArgs` from its CLI
6//! `ServeArgs`.
7
8use std::path::Path;
9use std::sync::Arc;
10#[cfg(feature = "fallback")]
11use std::time::Duration;
12
13use boatramp_core::Storage;
14
15use crate::backends::BlobBackend;
16use crate::error::{Error, Result};
17
18#[cfg(feature = "fs")]
19use boatramp_storage::FsStorage;
20
21/// The boatramp read-fallback allowlist predicate (blob-backend migration Part 2): a primary miss
22/// falls through to the read-only secondary ONLY for a boatramp-OWNED key —
23/// - a content-addressed deploy blob (`{2hex}/{64hex}`, via [`boatramp_core::deploy::is_blob_key`]),
24/// - a guest object (`hblob/…`),
25/// - a messaging PAYLOAD — both the work-queue payload (`mqp/…`) and the grouped/retained payload
26///   (`mqgp/…`).
27///
28/// Any OTHER key (e.g. a stray `config/…` that should never live in the blob store — the mutable
29/// control-plane records are in the KV, not `Storage`) is primary-only, so it can NEVER silently
30/// resurrect off a secondary during a transition (Security Finding F4/G5-cp, defense-in-depth). The
31/// predicate is supplied to the generic [`FallbackStorage`](boatramp_storage::FallbackStorage), which
32/// hardcodes no app prefixes.
33///
34/// v0.31.0: `mqp/` was added alongside `mqgp/` so that, during the default-scope re-home soak
35/// (Finding 8), a work-queue payload re-homed from `mqp/{topic}/…` to `mqp/default/{topic}/…` is
36/// fallback-eligible too — without it a default-scoped work-queue payload would 404 on the new
37/// spelling during a concurrent backend switch.
38#[cfg(feature = "fallback")]
39pub fn blob_fallback_when() -> boatramp_storage::FallbackWhen {
40    Arc::new(|k: &str| {
41        boatramp_core::deploy::is_blob_key(k)
42            || k.starts_with("hblob/")
43            || k.starts_with("mqp/")
44            || k.starts_with("mqgp/")
45    })
46}
47
48/// The guest-blob data families whose keys the default-scope re-home (v0.31.0 Finding 8) re-homes:
49/// the guest object store and both messaging payload stores. `hkv/` + the `mq*/` INDICES live in the
50/// KV, not `Storage`, so they migrate via `boatramp migrate`, not here.
51#[cfg(feature = "fallback")]
52pub const DEFAULT_SCOPE_BLOB_FAMILIES: &[&str] = &["hblob/", "mqp/", "mqgp/"];
53
54/// Timeout bounding each translated secondary read in the default-scope blob fallback (a wedged
55/// backend degrades a translated miss to `NotFound`, never a serve-path hang). Matches the default
56/// used by the backend-switch fallback.
57#[cfg(feature = "fallback")]
58const DEFAULT_SCOPE_FALLBACK_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
59
60/// The **new→old** translator for the default-scope blob fallback: for a post-flip key
61/// `{family}default/{rest}` under a known data family, name the pre-flip bare key `{family}{rest}` to
62/// fall back to. Purely syntactic (no project set needed): only a `…/default/…` key is translatable,
63/// so a named project's `{family}{project}/…` key never falls back and no cross-scope resurrection is
64/// possible.
65#[cfg(feature = "fallback")]
66pub fn default_scope_to_old() -> boatramp_storage::KeyTranslate {
67    Arc::new(|k: &str| {
68        for fam in DEFAULT_SCOPE_BLOB_FAMILIES {
69            if let Some(rest) = k.strip_prefix(fam) {
70                let suffix = rest.strip_prefix("default/")?;
71                return Some(format!("{fam}{suffix}"));
72            }
73        }
74        None
75    })
76}
77
78/// The **old→new** translator (project-aware) for the default-scope blob fallback's `list` union and
79/// the [`blob_rehome`](boatramp_storage::blob_rehome) drain/retire: for a bare default-scoped key
80/// `{family}{base}/…` (whose scope segment is NOT a known project), name the re-homed
81/// `{family}default/{base}/…`. A named project's key and an already-`default/` key both yield `None`.
82#[cfg(feature = "fallback")]
83pub fn default_scope_to_new(
84    projects: std::sync::Arc<std::collections::HashSet<String>>,
85) -> boatramp_storage::KeyTranslate {
86    Arc::new(move |k: &str| {
87        for fam in DEFAULT_SCOPE_BLOB_FAMILIES {
88            if let Some(rest) = k.strip_prefix(fam) {
89                let first = rest.split('/').next().unwrap_or("");
90                if first.is_empty()
91                    || first == boatramp_core::project::DEFAULT_PROJECT
92                    || projects.contains(first)
93                {
94                    return None;
95                }
96                return Some(format!(
97                    "{fam}{}/{rest}",
98                    boatramp_core::project::DEFAULT_PROJECT
99                ));
100            }
101        }
102        None
103    })
104}
105
106/// A [`BlobBackends`](boatramp_core::blob_backends::BlobBackends) decorator that wraps each resolved
107/// backend in a [`TranslatingFallbackStorage`](boatramp_storage::TranslatingFallbackStorage) — the
108/// serve-path half of the default-scope blob re-home (Finding 8). Enabled ONLY while the store is in
109/// the default-scope migration's `dual` soak (see [`default_scope_blob_fallback`]); removed (by
110/// restarting after `boatramp migrate --finalize` + a verified blob retire) once the soak ends.
111#[cfg(feature = "fallback")]
112pub struct DefaultScopeFallbackBackends {
113    inner: Arc<dyn boatramp_core::blob_backends::BlobBackends>,
114    to_old: boatramp_storage::KeyTranslate,
115    to_new: boatramp_storage::KeyTranslate,
116}
117
118#[cfg(feature = "fallback")]
119impl boatramp_core::blob_backends::BlobBackends for DefaultScopeFallbackBackends {
120    fn backend(&self, project: boatramp_core::project::ProjectRef<'_>) -> Arc<dyn Storage> {
121        Arc::new(boatramp_storage::TranslatingFallbackStorage::new(
122            self.inner.backend(project),
123            self.to_old.clone(),
124            self.to_new.clone(),
125            DEFAULT_SCOPE_FALLBACK_TIMEOUT,
126        ))
127    }
128    fn default_backend(&self) -> Arc<dyn Storage> {
129        Arc::new(boatramp_storage::TranslatingFallbackStorage::new(
130            self.inner.default_backend(),
131            self.to_old.clone(),
132            self.to_new.clone(),
133            DEFAULT_SCOPE_FALLBACK_TIMEOUT,
134        ))
135    }
136}
137
138/// If the control-plane store `kv` is in the **default-scope migration's `dual` soak** (its forward
139/// re-key has run but the destructive finalize is deferred), wrap `inner` so every `wasi:blobstore`
140/// backend serves a not-yet-drained bare `hblob/{site}/…`/`mqp/…`/`mqgp/…` object at the new
141/// `…/default/…` spelling the post-flip runtime composes — the zero-downtime blob transition. Returns
142/// `None` (the inner resolver unchanged) when the store is not in that soak, so a fully-migrated or
143/// pre-migration node is byte-for-byte unchanged. The named-project set (for the `list`-union / drain
144/// translator) is read once here from `projectmeta/`.
145#[cfg(feature = "fallback")]
146pub async fn default_scope_blob_fallback(
147    kv: &dyn boatramp_core::kv::KvStore,
148    inner: Arc<dyn boatramp_core::blob_backends::BlobBackends>,
149) -> Arc<dyn boatramp_core::blob_backends::BlobBackends> {
150    let in_soak = boatramp_core::migrate::read_state(kv)
151        .await
152        .map(|s| {
153            s.unfinalized
154                .contains(&boatramp_core::migrate::DEFAULT_SCOPE_REHOME_VERSION)
155        })
156        .unwrap_or(false);
157    if !in_soak {
158        return inner;
159    }
160    // The named projects (for the project-aware old→new translator). Read once at wiring.
161    let mut projects = std::collections::HashSet::new();
162    if let Ok(keys) = kv.list_prefix(boatramp_core::project::POINTER_PREFIX).await {
163        for key in keys {
164            if let Some(name) = key.strip_prefix(boatramp_core::project::POINTER_PREFIX)
165                && !name.is_empty()
166                && !name.contains('/')
167            {
168                projects.insert(name.to_string());
169            }
170        }
171    }
172    tracing::warn!(
173        "default-scope blob read-fallback ACTIVE (migration {} in dual soak): serving bare \
174         default-scoped blobs at their new default/ spelling. Run the blob re-home drain, then \
175         `boatramp migrate --finalize` + retire the old blobs, and restart to disable it.",
176        boatramp_core::migrate::DEFAULT_SCOPE_REHOME_VERSION
177    );
178    Arc::new(DefaultScopeFallbackBackends {
179        inner,
180        to_old: default_scope_to_old(),
181        to_new: default_scope_to_new(std::sync::Arc::new(projects)),
182    })
183}
184
185/// The resolved blob-backend selection — the binary populates this from its CLI
186/// `ServeArgs` (the credential/endpoint flags), keeping clap out of the library.
187///
188/// `s3_credential` (#505) is the optional node-level **sealed base S3 credential** the operator
189/// resolved from the `[secrets]` store; when present it is injected into the S3 client's SDK config
190/// instead of the ambient env chain. It is a redacted [`SealedS3Credential`], so `BlobArgs` can keep its
191/// derived `Debug` without leaking the secret.
192#[derive(Debug, Clone)]
193pub struct BlobArgs {
194    pub blobs: BlobBackend,
195    pub s3_bucket: Option<String>,
196    pub s3_endpoint: Option<String>,
197    pub s3_region: Option<String>,
198    pub s3_path_style: bool,
199    /// The node-level sealed base S3 credential (#505). `None` ⇒ the ambient AWS env chain (unchanged).
200    pub s3_credential: Option<crate::s3_credential::SealedS3Credential>,
201    pub gcs_bucket: Option<String>,
202    pub gcs_endpoint: Option<String>,
203    pub gcs_anonymous: bool,
204    pub azure_account: Option<String>,
205    pub azure_container: Option<String>,
206    pub azure_access_key: Option<String>,
207    pub azure_emulator: bool,
208}
209
210/// The blob backend plus, on a cloud object store with notification provisioning
211/// configured, its blob-change [`WatchProvider`](boatramp_core::blob_provision::WatchProvider)
212/// and operator tier (FA-5b2). The provider/tier are consumed only by the handler
213/// runtime, so they are dead code in a `--no-default-features` (no `handlers`) build.
214pub struct BuiltBlobs {
215    pub storage: Arc<dyn Storage>,
216    #[cfg_attr(not(feature = "handlers"), allow(dead_code))]
217    pub watch_provider: Option<Arc<dyn boatramp_core::blob_provision::WatchProvider>>,
218    #[cfg_attr(not(feature = "handlers"), allow(dead_code))]
219    pub provision_tier: boatramp_core::blob_notify::ProvisionTier,
220}
221
222/// Build the object store for the selected [`BlobBackend`]. `data_dir` is used
223/// only by the `fs` backend (unused when `fs` is off).
224#[cfg_attr(not(feature = "fs"), allow(unused_variables))]
225pub async fn build_blobs(
226    args: &BlobArgs,
227    data_dir: &Path,
228    notify_tier: Option<boatramp_core::blob_notify::ProvisionTier>,
229    notify_account: Option<String>,
230) -> Result<BuiltBlobs> {
231    match args.blobs {
232        #[cfg(feature = "fs")]
233        BlobBackend::Fs => Ok(BuiltBlobs {
234            storage: Arc::new(FsStorage::new(data_dir.join("blobs"))),
235            watch_provider: None,
236            provision_tier: boatramp_core::blob_notify::ProvisionTier::default(),
237        }),
238        #[cfg(not(feature = "fs"))]
239        BlobBackend::Fs => Err(Error::NoFsSupport),
240        BlobBackend::S3 => build_s3(args, notify_tier, notify_account).await,
241        BlobBackend::Gcs => build_gcs(args, notify_tier, notify_account).await,
242        BlobBackend::Azure => build_azure(args, notify_tier, notify_account).await,
243    }
244}
245
246/// Build the PRIMARY blob backend (with its notify provisioning) and, when `fallback` is supplied,
247/// build a read-only SECONDARY and wrap the pair in a [`FallbackStorage`](boatramp_storage::FallbackStorage)
248/// — the zero-downtime backend switch (blob-backend migration Part 2).
249///
250/// The secondary is built via the SAME [`build_blobs`] path but with **NO watcher provisioning**
251/// (`notify_tier: None`, `notify_account: None`) — a read-only drain source never watches — and its
252/// `watch_provider`/`provision_tier` are discarded. The returned [`BuiltBlobs`] keeps the PRIMARY's
253/// `watch_provider`/`provision_tier` (watching is the primary's capability). The secondary is NEVER
254/// handed to the blob-upload/STS minter (that path takes the primary `BlobArgs` only).
255///
256/// CRITICAL ordering: if a read-through cache is ever wrapped around the result, it must wrap the
257/// OUTSIDE — `cache(fallback(primary, secondary))` — so the cache's `allows_prune` delegates through
258/// the fallback's `false`. `FallbackStorage` is therefore the INNERMOST composite here; the returned
259/// `storage` is the fallback (or the bare primary when no fallback is configured).
260#[cfg(feature = "fallback")]
261pub async fn build_blobs_with_fallback(
262    args: &BlobArgs,
263    fallback: Option<(&BlobArgs, Duration)>,
264    data_dir: &Path,
265    notify_tier: Option<boatramp_core::blob_notify::ProvisionTier>,
266    notify_account: Option<String>,
267) -> Result<BuiltBlobs> {
268    // The PRIMARY carries its notify provisioning (watching stays a primary capability).
269    let primary = build_blobs(args, data_dir, notify_tier, notify_account).await?;
270    let Some((secondary_args, timeout)) = fallback else {
271        return Ok(primary);
272    };
273    // The read-only SECONDARY: same build path, NO watcher provisioning; discard its
274    // watch_provider/provision_tier (a drain source never watches).
275    let secondary = build_blobs(secondary_args, data_dir, None, None).await?;
276    let composite: Arc<dyn Storage> = Arc::new(boatramp_storage::FallbackStorage::new(
277        primary.storage,
278        secondary.storage,
279        blob_fallback_when(),
280        timeout,
281    ));
282    Ok(BuiltBlobs {
283        storage: composite,
284        // Keep the PRIMARY's watcher/tier — watching is the primary's capability.
285        watch_provider: primary.watch_provider,
286        provision_tier: primary.provision_tier,
287    })
288}
289
290// Azure storage + optional blob-change notification (Event Grid → Storage Queue,
291// FA-5b2). When a notify tier is configured the backend is consumer-wired and
292// paired with the AzureWatchProvider (the Event Grid subscription is an operator
293// step — see the provider recipe).
294#[cfg(feature = "azure")]
295async fn build_azure(
296    args: &BlobArgs,
297    notify_tier: Option<boatramp_core::blob_notify::ProvisionTier>,
298    _notify_account: Option<String>,
299) -> Result<BuiltBlobs> {
300    let (Some(account), Some(container)) =
301        (args.azure_account.clone(), args.azure_container.clone())
302    else {
303        return Err(Error::AzureConfigRequired);
304    };
305    let opts = boatramp_storage::AzureOptions {
306        account,
307        container,
308        access_key: args.azure_access_key.clone(),
309        emulator: args.azure_emulator,
310    };
311    match notify_tier {
312        Some(tier) => {
313            let (storage, provider) = boatramp_storage::AzureStorage::connect_with_notify(opts)
314                .map_err(|err| Error::AzureConnect(err.to_string()))?;
315            Ok(BuiltBlobs {
316                storage: Arc::new(storage),
317                watch_provider: Some(Arc::new(provider)),
318                provision_tier: tier,
319            })
320        }
321        None => {
322            let storage = boatramp_storage::AzureStorage::connect(opts)
323                .map_err(|err| Error::AzureConnect(err.to_string()))?;
324            Ok(BuiltBlobs {
325                storage: Arc::new(storage),
326                watch_provider: None,
327                provision_tier: boatramp_core::blob_notify::ProvisionTier::default(),
328            })
329        }
330    }
331}
332
333#[cfg(not(feature = "azure"))]
334async fn build_azure(
335    _args: &BlobArgs,
336    _notify_tier: Option<boatramp_core::blob_notify::ProvisionTier>,
337    _notify_account: Option<String>,
338) -> Result<BuiltBlobs> {
339    Err(Error::NoAzureSupport)
340}
341
342// GCS storage + optional blob-change notification (GCS→Pub/Sub, FA-5b2). When a
343// notify tier is configured the backend is consumer-wired and paired with the
344// GcsWatchProvider; `blob_notify_account_id` is read as the GCP project id.
345#[cfg(feature = "gcs")]
346async fn build_gcs(
347    args: &BlobArgs,
348    notify_tier: Option<boatramp_core::blob_notify::ProvisionTier>,
349    notify_account: Option<String>,
350) -> Result<BuiltBlobs> {
351    let bucket = args.gcs_bucket.clone().ok_or(Error::GcsBucketRequired)?;
352    let opts = boatramp_storage::GcsOptions {
353        bucket,
354        endpoint: args.gcs_endpoint.clone(),
355        anonymous: args.gcs_anonymous,
356    };
357    match notify_tier {
358        Some(tier) => {
359            let project = notify_account.unwrap_or_default();
360            let (storage, provider) =
361                boatramp_storage::GcsStorage::connect_with_notify(opts, project)
362                    .await
363                    .map_err(|err| Error::GcsConnect(err.to_string()))?;
364            Ok(BuiltBlobs {
365                storage: Arc::new(storage),
366                watch_provider: Some(Arc::new(provider)),
367                provision_tier: tier,
368            })
369        }
370        None => {
371            let storage = boatramp_storage::GcsStorage::connect(opts)
372                .await
373                .map_err(|err| Error::GcsConnect(err.to_string()))?;
374            Ok(BuiltBlobs {
375                storage: Arc::new(storage),
376                watch_provider: None,
377                provision_tier: boatramp_core::blob_notify::ProvisionTier::default(),
378            })
379        }
380    }
381}
382
383#[cfg(not(feature = "gcs"))]
384async fn build_gcs(
385    _args: &BlobArgs,
386    _notify_tier: Option<boatramp_core::blob_notify::ProvisionTier>,
387    _notify_account: Option<String>,
388) -> Result<BuiltBlobs> {
389    Err(Error::NoGcsSupport)
390}
391
392#[cfg(feature = "s3")]
393async fn build_s3(
394    args: &BlobArgs,
395    notify_tier: Option<boatramp_core::blob_notify::ProvisionTier>,
396    notify_account: Option<String>,
397) -> Result<BuiltBlobs> {
398    let bucket = args.s3_bucket.clone().ok_or(Error::S3BucketRequired)?;
399    let opts = boatramp_storage::S3Options {
400        bucket,
401        endpoint: args.s3_endpoint.clone(),
402        region: args.s3_region.clone(),
403        force_path_style: args.s3_path_style,
404        // #505: when the operator configured a node-level sealed base credential, inject it so the S3
405        // client signs with the sealed key rather than the ambient `AWS_ACCESS_KEY_ID`/`_SECRET`.
406        credential: args.s3_credential.as_ref().map(|c| c.as_pair()),
407    };
408    match notify_tier {
409        // Blob-change notification provisioning is enabled: build the
410        // consumer-wired storage + the S3→SQS provider from one AWS config.
411        Some(tier) => {
412            let account = notify_account.unwrap_or_default();
413            let (storage, provider) =
414                boatramp_storage::S3Storage::connect_with_notify(opts, account).await;
415            Ok(BuiltBlobs {
416                storage: Arc::new(storage),
417                watch_provider: Some(Arc::new(provider)),
418                provision_tier: tier,
419            })
420        }
421        // No provisioning configured: a plain S3 backend (blob triggers refuse).
422        None => Ok(BuiltBlobs {
423            storage: Arc::new(boatramp_storage::S3Storage::connect(opts).await),
424            watch_provider: None,
425            provision_tier: boatramp_core::blob_notify::ProvisionTier::default(),
426        }),
427    }
428}
429
430#[cfg(not(feature = "s3"))]
431async fn build_s3(
432    _args: &BlobArgs,
433    _notify_tier: Option<boatramp_core::blob_notify::ProvisionTier>,
434    _notify_account: Option<String>,
435) -> Result<BuiltBlobs> {
436    Err(Error::NoS3Support)
437}
boatramp_node/blobs.rs

boatramp_node/
blobs.rs
