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

little_durable_objects/
storage_urls.rs
