little_durable_objects/
storage_urls.rs

1use std::{collections::HashMap, time::Duration};
2
3use anyhow::{Context, Result, ensure};
4use async_trait::async_trait;
5use google_cloud_auth::{credentials, signer::Signer};
6use google_cloud_storage::{builder::storage::SignedUrlBuilder, http::Method};
7
8use crate::{actor::ActorKey, placement::validate_region};
9
10const SIGNED_URL_TTL: Duration = Duration::from_secs(60);
11pub const STATE_CONTENT_TYPE: &str = "application/x-ndjson";
12
13#[async_trait]
14pub trait StorageUrlSigner: Send + Sync {
15    async fn read_url(&self, region: &str, actor: &ActorKey) -> Result<String>;
16
17    async fn write_url(
18        &self,
19        region: &str,
20        actor: &ActorKey,
21        expected_generation: &str,
22    ) -> Result<String>;
23
24    fn regions(&self) -> Vec<String>;
25}
26
27#[derive(Clone)]
28pub struct GcsStorageUrlSigner {
29    buckets: HashMap<String, String>,
30    signer: Signer,
31}
32
33impl GcsStorageUrlSigner {
34    pub fn from_adc(buckets: HashMap<String, String>) -> Result<Self> {
35        Self::new(buckets, credentials::Builder::default().build_signer()?)
36    }
37
38    pub fn new(buckets: HashMap<String, String>, signer: Signer) -> Result<Self> {
39        validate_buckets(&buckets)?;
40        Ok(Self { buckets, signer })
41    }
42}
43
44#[async_trait]
45impl StorageUrlSigner for GcsStorageUrlSigner {
46    async fn read_url(&self, region: &str, actor: &ActorKey) -> Result<String> {
47        actor.validate()?;
48        SignedUrlBuilder::for_object(self.bucket(region)?, object_name(actor))
49            .with_method(Method::GET)
50            .with_expiration(SIGNED_URL_TTL)
51            .sign_with(&self.signer)
52            .await
53            .context("sign GCS actor-state read URL")
54    }
55
56    async fn write_url(
57        &self,
58        region: &str,
59        actor: &ActorKey,
60        expected_generation: &str,
61    ) -> Result<String> {
62        actor.validate()?;
63        validate_generation(expected_generation)?;
64        SignedUrlBuilder::for_object(self.bucket(region)?, object_name(actor))
65            .with_method(Method::PUT)
66            .with_expiration(SIGNED_URL_TTL)
67            .with_header("content-type", STATE_CONTENT_TYPE)
68            .with_query_param("ifGenerationMatch", expected_generation)
69            .sign_with(&self.signer)
70            .await
71            .context("sign GCS actor-state write URL")
72    }
73
74    fn regions(&self) -> Vec<String> {
75        let mut regions = self.buckets.keys().cloned().collect::<Vec<_>>();
76        regions.sort();
77        regions
78    }
79}
80
81impl GcsStorageUrlSigner {
82    fn bucket(&self, region: &str) -> Result<String> {
83        validate_region(region)?;
84        self.buckets
85            .get(region)
86            .map(|bucket| format!("projects/_/buckets/{bucket}"))
87            .with_context(|| format!("sandbox region {region:?} has no Standard bucket"))
88    }
89}
90
91pub fn object_name(actor: &ActorKey) -> String {
92    format!(
93        "objects/{}/{}/{}.ndjson",
94        actor.namespace_id, actor.actor_type, actor.actor_id
95    )
96}
97
98pub fn validate_buckets(buckets: &HashMap<String, String>) -> Result<()> {
99    ensure!(!buckets.is_empty(), "Standard bucket map must not be empty");
100    for (region, bucket) in buckets {
101        validate_region(region)?;
102        ensure!(
103            !bucket.is_empty()
104                && bucket.trim() == bucket
105                && bucket.len() <= 222
106                && !bucket.contains('/'),
107            "Standard bucket for region {region:?} is invalid"
108        );
109    }
110    Ok(())
111}
112
113fn validate_generation(generation: &str) -> Result<()> {
114    ensure!(
115        !generation.is_empty()
116            && generation.len() <= 32
117            && generation.bytes().all(|byte| byte.is_ascii_digit()),
118        "GCS generation is invalid"
119    );
120    Ok(())
121}
122
123#[cfg(test)]
124mod tests {
125    use super::*;
126
127    #[test]
128    fn derives_one_readable_object_path() {
129        assert_eq!(
130            object_name(&ActorKey {
131                namespace_id: "project-1".into(),
132                actor_type: "Counter".into(),
133                actor_id: "account.42".into(),
134            }),
135            "objects/project-1/Counter/account.42.ndjson"
136        );
137    }
138
139    #[test]
140    fn validates_the_only_storage_configuration() {
141        assert!(
142            validate_buckets(&HashMap::from([(
143                "us-east".into(),
144                "objects-us-east".into()
145            )]))
146            .is_ok()
147        );
148        assert!(validate_buckets(&HashMap::new()).is_err());
149        assert!(
150            validate_buckets(&HashMap::from([("bad/region".into(), "objects".into())])).is_err()
151        );
152    }
153}
little_durable_objects/storage_urls.rs

little_durable_objects/
storage_urls.rs
