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};
7use serde::{Deserialize, Serialize};
8
9use crate::{actor::ActorKey, placement::validate_region};
10
11const SIGNED_URL_TTL: Duration = Duration::from_secs(60);
12pub const STATE_CONTENT_TYPE: &str = "application/json";
13
14#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
15#[serde(rename_all = "camelCase")]
16pub struct StateWriteTicket {
17    pub state_version: u64,
18    pub object_name: String,
19    pub url: String,
20    pub expires_at_ms: i64,
21}
22
23#[async_trait]
24pub trait StorageUrlSigner: Send + Sync {
25    async fn read_url(&self, region: &str, object_name: &str) -> Result<String>;
26
27    async fn write_ticket(
28        &self,
29        region: &str,
30        actor: &ActorKey,
31        state_version: u64,
32    ) -> Result<StateWriteTicket>;
33
34    fn regions(&self) -> Vec<String>;
35}
36
37#[derive(Clone)]
38pub struct GcsStorageUrlSigner {
39    buckets: HashMap<String, String>,
40    signer: Signer,
41}
42
43impl GcsStorageUrlSigner {
44    pub fn from_adc(buckets: HashMap<String, String>) -> Result<Self> {
45        Self::new(buckets, credentials::Builder::default().build_signer()?)
46    }
47
48    pub fn new(buckets: HashMap<String, String>, signer: Signer) -> Result<Self> {
49        validate_buckets(&buckets)?;
50        Ok(Self { buckets, signer })
51    }
52}
53
54#[async_trait]
55impl StorageUrlSigner for GcsStorageUrlSigner {
56    async fn read_url(&self, region: &str, object_name: &str) -> Result<String> {
57        validate_object_name(object_name)?;
58        SignedUrlBuilder::for_object(self.bucket(region)?, object_name)
59            .with_method(Method::GET)
60            .with_expiration(SIGNED_URL_TTL)
61            .sign_with(&self.signer)
62            .await
63            .context("sign GCS actor-state read URL")
64    }
65
66    async fn write_ticket(
67        &self,
68        region: &str,
69        actor: &ActorKey,
70        state_version: u64,
71    ) -> Result<StateWriteTicket> {
72        actor.validate()?;
73        ensure!(state_version > 0, "actor state version must be positive");
74        let nonce = uuid::Uuid::new_v4().simple().to_string();
75        let object_name = snapshot_object_name(actor, state_version, &nonce)?;
76        let expires_at_ms = unix_millis()?
77            .checked_add(i64::try_from(SIGNED_URL_TTL.as_millis())?)
78            .context("signed state-write URL expiration overflow")?;
79        let url = SignedUrlBuilder::for_object(self.bucket(region)?, &object_name)
80            .with_method(Method::PUT)
81            .with_expiration(SIGNED_URL_TTL)
82            .with_header("content-type", STATE_CONTENT_TYPE)
83            .with_query_param("ifGenerationMatch", "0")
84            .sign_with(&self.signer)
85            .await
86            .context("sign GCS actor-state write URL")?;
87        Ok(StateWriteTicket {
88            state_version,
89            object_name,
90            url,
91            expires_at_ms,
92        })
93    }
94
95    fn regions(&self) -> Vec<String> {
96        let mut regions = self.buckets.keys().cloned().collect::<Vec<_>>();
97        regions.sort();
98        regions
99    }
100}
101
102impl GcsStorageUrlSigner {
103    fn bucket(&self, region: &str) -> Result<String> {
104        validate_region(region)?;
105        self.buckets
106            .get(region)
107            .map(|bucket| format!("projects/_/buckets/{bucket}"))
108            .with_context(|| format!("sandbox region {region:?} has no Standard bucket"))
109    }
110}
111
112pub fn snapshot_object_name(actor: &ActorKey, state_version: u64, nonce: &str) -> Result<String> {
113    actor.validate()?;
114    ensure!(state_version > 0, "actor state version must be positive");
115    ensure!(
116        nonce.len() == 32 && nonce.bytes().all(|byte| byte.is_ascii_hexdigit()),
117        "actor state object nonce is invalid"
118    );
119    Ok(format!(
120        "snapshots/{}/{}/{}/{}/{}/{}.json",
121        &nonce[..2],
122        nonce,
123        actor.namespace_id,
124        actor.actor_type,
125        actor.actor_id,
126        state_version,
127    ))
128}
129
130pub fn validate_snapshot_object_name(
131    actor: &ActorKey,
132    state_version: u64,
133    object_name: &str,
134) -> Result<()> {
135    validate_object_name(object_name)?;
136    let mut parts = object_name.split('/');
137    let valid = matches!(parts.next(), Some("snapshots"))
138        && matches!((parts.next(), parts.next()), (Some(prefix), Some(nonce)) if nonce.starts_with(prefix) && prefix.len() == 2 && nonce.len() == 32 && nonce.bytes().all(|byte| byte.is_ascii_hexdigit()))
139        && parts.next() == Some(actor.namespace_id.as_str())
140        && parts.next() == Some(actor.actor_type.as_str())
141        && parts.next() == Some(actor.actor_id.as_str())
142        && parts.next() == Some(format!("{state_version}.json").as_str())
143        && parts.next().is_none();
144    ensure!(valid, "actor state object name does not match its commit");
145    Ok(())
146}
147
148pub fn validate_buckets(buckets: &HashMap<String, String>) -> Result<()> {
149    ensure!(!buckets.is_empty(), "Standard bucket map must not be empty");
150    for (region, bucket) in buckets {
151        validate_region(region)?;
152        ensure!(
153            !bucket.is_empty()
154                && bucket.trim() == bucket
155                && bucket.len() <= 222
156                && !bucket.contains('/'),
157            "Standard bucket for region {region:?} is invalid"
158        );
159    }
160    Ok(())
161}
162
163fn validate_object_name(object_name: &str) -> Result<()> {
164    ensure!(
165        object_name.starts_with("snapshots/")
166            && object_name.len() <= 1024
167            && !object_name.chars().any(char::is_control),
168        "actor state object name is invalid"
169    );
170    Ok(())
171}
172
173fn unix_millis() -> Result<i64> {
174    i64::try_from(
175        std::time::SystemTime::now()
176            .duration_since(std::time::UNIX_EPOCH)
177            .context("system clock is before the Unix epoch")?
178            .as_millis(),
179    )
180    .context("system clock exceeds supported state-write timestamp range")
181}
182
183#[cfg(test)]
184mod tests {
185    use super::*;
186
187    #[test]
188    fn derives_a_randomly_distributed_immutable_snapshot_path() -> Result<()> {
189        assert_eq!(
190            snapshot_object_name(
191                &ActorKey {
192                    namespace_id: "project-1".into(),
193                    actor_type: "Counter".into(),
194                    actor_id: "account.42".into(),
195                },
196                7,
197                "0123456789abcdef0123456789abcdef",
198            )?,
199            "snapshots/01/0123456789abcdef0123456789abcdef/project-1/Counter/account.42/7.json"
200        );
201        validate_snapshot_object_name(
202            &ActorKey {
203                namespace_id: "project-1".into(),
204                actor_type: "Counter".into(),
205                actor_id: "account.42".into(),
206            },
207            7,
208            "snapshots/01/0123456789abcdef0123456789abcdef/project-1/Counter/account.42/7.json",
209        )?;
210        Ok(())
211    }
212
213    #[test]
214    fn validates_the_only_storage_configuration() {
215        assert!(
216            validate_buckets(&HashMap::from([(
217                "us-east".into(),
218                "objects-us-east".into()
219            )]))
220            .is_ok()
221        );
222        assert!(validate_buckets(&HashMap::new()).is_err());
223        assert!(
224            validate_buckets(&HashMap::from([("bad/region".into(), "objects".into())])).is_err()
225        );
226    }
227}
little_durable_objects/storage_urls.rs

little_durable_objects/
storage_urls.rs
