| 1 | //! Public storage admission is independent of a paid subscription. Reserve |
| 2 | //! bytes before contacting the git store; a crash retains the reservation |
| 3 | //! rather than admitting the same capacity twice. A successful HTTP push |
| 4 | //! releases only the difference between its reservation and measured pack. |
| 5 | use g1t_contracts::repos::Repo; |
| 6 | use serde::Deserialize; |
| 7 | use worker::Result; |
| 8 | |
| 9 | use crate::registry::Registry; |
| 10 | |
| 11 | pub(crate) struct StreamQuota { |
| 12 | pub registry: Registry, |
| 13 | pub repo: Repo, |
| 14 | pub cap: i64, |
| 15 | } |
| 16 | |
| 17 | impl<S: crate::store::GitStore> crate::Repos<S> { |
| 18 | pub(crate) fn public_quota(&self, repo: &Repo) -> Option<std::rc::Rc<StreamQuota>> { |
| 19 | Some(std::rc::Rc::new(StreamQuota { |
| 20 | registry: Registry { db: self.registry.db.clone() }, repo: repo.clone(), cap: self.free_public_bytes, |
| 21 | })) |
| 22 | } |
| 23 | } |
| 24 | |
| 25 | pub(crate) const RESERVE: &str = "UPDATE repos SET storage_reserved_bytes = storage_reserved_bytes + ?2 |
| 26 | WHERE id = ?1 AND fork_of IS NULL AND ?2 >= 0 |
| 27 | AND NOT EXISTS (SELECT 1 FROM storage_copy_locks WHERE repo_id = ?1) |
| 28 | AND (is_private = 1 OR (?2 <= ?3 AND COALESCE((SELECT SUM(stored_bytes + storage_reserved_bytes) FROM repos r |
| 29 | WHERE r.namespace = repos.namespace AND r.is_private = 0 AND r.fork_of IS NULL), 0) <= ?3 - ?2)) |
| 30 | RETURNING storage_reserved_bytes AS stored_bytes"; |
| 31 | |
| 32 | pub(crate) const MAKE_PUBLIC: &str = "UPDATE repos SET is_private = 0 WHERE id = ?1 AND is_private = 1 |
| 33 | AND stored_bytes + storage_reserved_bytes <= ?2 - COALESCE((SELECT SUM(r.stored_bytes + r.storage_reserved_bytes) FROM repos r |
| 34 | WHERE r.namespace = repos.namespace AND r.is_private = 0 AND r.fork_of IS NULL), 0) |
| 35 | RETURNING id"; |
| 36 | |
| 37 | pub(crate) const TRANSFER: &str = "UPDATE repos SET namespace = ?2 WHERE id = ?1 AND namespace = ?3 |
| 38 | AND (is_private = 1 OR stored_bytes + storage_reserved_bytes <= ?4 - COALESCE((SELECT SUM(r.stored_bytes + r.storage_reserved_bytes) FROM repos r |
| 39 | WHERE r.namespace = ?2 AND r.is_private = 0 AND r.fork_of IS NULL), 0)) |
| 40 | RETURNING id"; |
| 41 | |
| 42 | impl Registry { |
| 43 | pub(crate) async fn copy_size(&self, repo: &Repo) -> Result<(u64, u64)> { |
| 44 | #[derive(Deserialize)] |
| 45 | struct Size { stored_bytes: f64, storage_reserved_bytes: f64 } |
| 46 | let row = self.db.prepare("SELECT stored_bytes, storage_reserved_bytes FROM repos WHERE id = ?") |
| 47 | .bind(&[repo.id.as_str().into()])?.first::<Size>(None).await? |
| 48 | .ok_or_else(|| worker::Error::RustError("Copy source disappeared".into()))?; |
| 49 | Ok((row.stored_bytes.max(0.0) as u64, row.storage_reserved_bytes.max(0.0) as u64)) |
| 50 | } |
| 51 | pub(crate) async fn lock_copy(&self, source: &Repo, destination: &Repo) -> Result<bool> { |
| 52 | #[derive(Deserialize)] |
| 53 | struct Locked { #[serde(rename = "repo_id")] _id: String } |
| 54 | Ok(self.db.prepare("INSERT INTO storage_copy_locks (repo_id, destination_id, created_at) VALUES (?1, ?2, ?3) ON CONFLICT DO NOTHING RETURNING repo_id") |
| 55 | .bind(&[source.id.as_str().into(), destination.id.as_str().into(), destination.created_at.as_str().into()])? |
| 56 | .first::<Locked>(None).await?.is_some()) |
| 57 | } |
| 58 | |
| 59 | pub(crate) async fn unlock_copy(&self, source: &Repo, destination: &Repo) -> Result<()> { |
| 60 | self.db.prepare("DELETE FROM storage_copy_locks WHERE repo_id = ?1 AND destination_id = ?2") |
| 61 | .bind(&[source.id.as_str().into(), destination.id.as_str().into()])?.run().await?; |
| 62 | Ok(()) |
| 63 | } |
| 64 | pub(crate) async fn make_public(&self, repo: &Repo, cap: i64) -> Result<bool> { |
| 65 | #[derive(Deserialize)] |
| 66 | struct Changed { #[serde(rename = "id")] _id: String } |
| 67 | let results = self.db.batch(vec![ |
| 68 | self.db.prepare(MAKE_PUBLIC).bind(&[repo.id.as_str().into(), (cap as f64).into()])?, |
| 69 | self.db.prepare("UPDATE repos SET is_private = 0 WHERE fork_of = ?1 AND EXISTS (SELECT 1 FROM repos root WHERE root.id = ?1 AND root.is_private = 0)") |
| 70 | .bind(&[repo.id.as_str().into()])?, |
| 71 | ]).await?; |
| 72 | Ok(!results[0].results::<Changed>()?.is_empty()) |
| 73 | } |
| 74 | /// Reserves against the canonical root, also counting retained deleted |
| 75 | /// repositories. Returns false when the workspace cannot fit the write. |
| 76 | pub(crate) async fn reserve_public(&self, repo: &Repo, bytes: u64, cap: i64) -> Result<bool> { |
| 77 | #[derive(Deserialize)] |
| 78 | struct Reserved { #[serde(rename = "stored_bytes")] _bytes: f64 } |
| 79 | let root = repo.fork_of.as_deref().unwrap_or(&repo.id); |
| 80 | Ok(self.db.prepare(RESERVE) |
| 81 | .bind(&[root.into(), (bytes as f64).into(), (cap as f64).into()])? |
| 82 | .first::<Reserved>(None).await?.is_some()) |
| 83 | } |
| 84 | |
| 85 | pub(crate) async fn release_public(&self, repo: &Repo, bytes: u64) -> Result<()> { |
| 86 | if bytes == 0 { return Ok(()); } |
| 87 | let root = repo.fork_of.as_deref().unwrap_or(&repo.id); |
| 88 | self.db.prepare("UPDATE repos SET storage_reserved_bytes = MAX(0, storage_reserved_bytes - ?2) WHERE id = ?1") |
| 89 | .bind(&[root.into(), (bytes as f64).into()])?.run().await?; |
| 90 | Ok(()) |
| 91 | } |
| 92 | |
| 93 | /// Convert admitted capacity to measured storage in one mutation. |
| 94 | pub(crate) async fn finish_public(&self, repo: &Repo, reserved: u64, measured: u64) -> Result<()> { |
| 95 | if measured > reserved { return Err(worker::Error::RustError("Push exceeded its storage reservation".into())); } |
| 96 | let root = repo.fork_of.as_deref().unwrap_or(&repo.id); |
| 97 | self.db.prepare("UPDATE repos SET stored_bytes = stored_bytes + ?3, storage_reserved_bytes = storage_reserved_bytes - ?2 WHERE id = ?1 AND storage_reserved_bytes >= ?2") |
| 98 | .bind(&[root.into(), (reserved as f64).into(), (measured as f64).into()])?.run().await?; |
| 99 | Ok(()) |
| 100 | } |
| 101 | } |