zuka
zuka/src/control/accounts.rs

worklyn / zukapublic

Agent-first git hosting. One Rust binary: git over HTTP and SSH, a REST API, MCP, CI, and multi-tenant isolation.

Get a copy: git clone https://zuka.worklyn.com/worklyn/zuka.git
branchbc985d6bbd321be340e3750b3da4734ee1553cbd ▾
main All branches →
zuka/src/control/accounts.rs
RSaccounts.rs10.2 KBDownload
1// Accounts, as the control plane sees them.
2//
3// One JSON file per account. The control plane holds no repositories, so this is the
4// whole of its durable state: who exists, where their tenant is, and whether it is
5// up.
6
7use crate::control::provision::State;
8use crate::error::{Error, Result};
9use crate::git::validate::Name;
10use anyhow::Context;
11use serde::{Deserialize, Serialize};
12use std::path::PathBuf;
13
14#[derive(Debug, Clone, Serialize, Deserialize)]
15pub struct AccountRecord {
16 /// Canonical account name.
17 pub name: String,
18 pub state: State,
19 /// Where to proxy, once there is somewhere to proxy to.
20 #[serde(default, skip_serializing_if = "Option::is_none")]
21 pub endpoint: Option<String>,
22 /// Why the last transition failed, when it did.
23 #[serde(default, skip_serializing_if = "Option::is_none")]
24 pub detail: Option<String>,
25 pub created_at: u64,
26 /// When this tenant was last proxied to. Drives idle stop.
27 #[serde(default)]
28 pub last_seen_at: u64,
29 /// Version of the binary currently inside this tenant's container.
30 ///
31 /// Recorded rather than probed: asking each container its version would be a
32 /// subprocess per tenant per sweep, and the control plane already knows what it
33 /// pushed. A mismatch with our own version is what triggers an upgrade.
34 #[serde(default)]
35 pub binary_version: Option<String>,
36}
37
38#[derive(Debug, Clone)]
39pub struct AccountStore {
40 root: PathBuf,
41}
42
43impl AccountStore {
44 pub fn open(root: impl Into<PathBuf>) -> Result<Self> {
45 let root = root.into();
46 std::fs::create_dir_all(&root)
47 .with_context(|| format!("create accounts dir {}", root.display()))?;
48 Ok(AccountStore { root })
49 }
50
51 fn path(&self, account: &Name) -> PathBuf {
52 self.root.join(format!("{}.json", account.as_str()))
53 }
54
55 /// Register an account. Returns the record as it now stands.
56 ///
57 /// Records intent and returns immediately: the reconciler does the provisioning.
58 /// Creating an account must not block on a container booting.
59 pub fn create(&self, account: &Name) -> Result<AccountRecord> {
60 if let Some(existing) = self.get(account)? {
61 return Ok(existing);
62 }
63 let record = AccountRecord {
64 name: account.as_str().to_string(),
65 state: State::Provisioning,
66 endpoint: None,
67 detail: None,
68 created_at: crate::account::token::now_secs(),
69 last_seen_at: crate::account::token::now_secs(),
70 binary_version: None,
71 };
72 self.put(&record)?;
73 Ok(record)
74 }
75
76 pub fn get(&self, account: &Name) -> Result<Option<AccountRecord>> {
77 let path = self.path(account);
78 match std::fs::read(&path) {
79 Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
80 Err(e) => Err(Error::Internal(
81 anyhow::Error::from(e).context(format!("read {}", path.display())),
82 )),
83 Ok(bytes) => Ok(Some(
84 serde_json::from_slice(&bytes)
85 .with_context(|| format!("parse {}", path.display()))?,
86 )),
87 }
88 }
89
90 pub fn put(&self, record: &AccountRecord) -> Result<()> {
91 let account = crate::git::validate::name(&record.name)?;
92 let path = self.path(&account);
93 let body = serde_json::to_vec_pretty(record).context("serialize account")?;
94
95 let temp = path.with_extension("json.tmp");
96 std::fs::write(&temp, &body).with_context(|| format!("write {}", temp.display()))?;
97 std::fs::rename(&temp, &path).with_context(|| format!("rename into {}", path.display()))?;
98 Ok(())
99 }
100
101 pub fn delete(&self, account: &Name) -> Result<bool> {
102 match std::fs::remove_file(self.path(account)) {
103 Ok(()) => Ok(true),
104 Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false),
105 Err(e) => Err(Error::Internal(
106 anyhow::Error::from(e).context("remove account"),
107 )),
108 }
109 }
110
111 /// Every account, oldest first. A record that fails to parse is skipped and
112 /// logged rather than failing the sweep.
113 pub fn list(&self) -> Vec<AccountRecord> {
114 let Ok(entries) = std::fs::read_dir(&self.root) else {
115 return Vec::new();
116 };
117 let mut records: Vec<AccountRecord> = entries
118 .flatten()
119 .filter(|e| e.path().extension().and_then(|x| x.to_str()) == Some("json"))
120 .filter_map(|e| match std::fs::read(e.path()) {
121 Ok(bytes) => match serde_json::from_slice(&bytes) {
122 Ok(record) => Some(record),
123 Err(err) => {
124 eprintln!(
125 "[control] skipping unreadable {}: {err}",
126 e.path().display()
127 );
128 None
129 }
130 },
131 Err(_) => None,
132 })
133 .collect();
134 records.sort_by_key(|r: &AccountRecord| r.created_at);
135 records
136 }
137
138 /// Where to proxy a request for this account, if anywhere.
139 ///
140 /// Touches `last_seen_at` on the way through, which is what keeps a tenant in
141 /// use from being stopped as idle.
142 pub fn endpoint(&self, account: &Name) -> Result<String> {
143 let mut record = self.get(account)?.ok_or(Error::NotFound("account"))?;
144
145 // A stopped tenant is woken by asking for it: the reconciler sees the
146 // refreshed timestamp and starts it.
147 if record.state == State::Stopped {
148 record.last_seen_at = crate::account::token::now_secs();
149 let _ = self.put(&record);
150 return Err(Error::RateLimited { retry_after: 5 });
151 }
152
153 if record.state == State::Ready {
154 let now = crate::account::token::now_secs();
155 // Written at most once a minute: a busy tenant would otherwise rewrite
156 // its record on every request.
157 if now.saturating_sub(record.last_seen_at) > 60 {
158 let mut touched = record.clone();
159 touched.last_seen_at = now;
160 let _ = self.put(&touched);
161 }
162 }
163
164 match record.state {
165 State::Ready => record.endpoint.ok_or_else(|| {
166 Error::Internal(anyhow::anyhow!("account is ready with no endpoint"))
167 }),
168 // Not an error the caller can fix, and not permanent: tell them to
169 // come back rather than reporting a fault.
170 State::Provisioning | State::Stopped => Err(Error::RateLimited { retry_after: 5 }),
171 State::Failed => Err(Error::conflict(
172 "account-unavailable",
173 record
174 .detail
175 .unwrap_or_else(|| "this account could not be provisioned".into()),
176 )),
177 }
178 }
179}
180
181#[cfg(test)]
182mod tests {
183 use super::*;
184 use crate::git::validate::name;
185
186 fn store() -> (tempfile::TempDir, AccountStore) {
187 let dir = tempfile::tempdir().unwrap();
188 let store = AccountStore::open(dir.path()).unwrap();
189 (dir, store)
190 }
191
192 #[test]
193 fn creating_an_account_records_intent_and_returns() {
194 let (_dir, store) = store();
195 let record = store.create(&name("alice").unwrap()).unwrap();
196
197 assert_eq!(record.state, State::Provisioning);
198 assert!(
199 record.endpoint.is_none(),
200 "nothing is provisioned yet; the reconciler does that"
201 );
202 }
203
204 #[test]
205 fn creating_twice_returns_the_existing_account() {
206 let (_dir, store) = store();
207 let alice = name("alice").unwrap();
208 let first = store.create(&alice).unwrap();
209 let second = store.create(&alice).unwrap();
210 assert_eq!(first.created_at, second.created_at);
211 }
212
213 #[test]
214 fn routing_reports_provisioning_as_come_back_rather_than_a_fault() {
215 let (_dir, store) = store();
216 let alice = name("alice").unwrap();
217 store.create(&alice).unwrap();
218
219 let err = store.endpoint(&alice).unwrap_err();
220 assert_eq!(
221 err.status(),
222 429,
223 "a booting tenant is a wait, not an error"
224 );
225 }
226
227 #[test]
228 fn routing_reports_a_failed_account_with_its_reason() {
229 let (_dir, store) = store();
230 let alice = name("alice").unwrap();
231 let mut record = store.create(&alice).unwrap();
232 record.state = State::Failed;
233 record.detail = Some("image not found".into());
234 store.put(&record).unwrap();
235
236 let err = store.endpoint(&alice).unwrap_err();
237 assert_eq!(err.status(), 409);
238 assert!(err.detail_for_user().contains("image not found"));
239 }
240
241 #[test]
242 fn routing_an_unknown_account_is_absent() {
243 let (_dir, store) = store();
244 assert_eq!(
245 store
246 .endpoint(&name("nobody").unwrap())
247 .unwrap_err()
248 .status(),
249 404
250 );
251 }
252
253 #[test]
254 fn a_ready_account_routes_to_its_endpoint() {
255 let (_dir, store) = store();
256 let alice = name("alice").unwrap();
257 let mut record = store.create(&alice).unwrap();
258 record.state = State::Ready;
259 record.endpoint = Some("http://10.0.0.5:8790".into());
260 store.put(&record).unwrap();
261
262 assert_eq!(store.endpoint(&alice).unwrap(), "http://10.0.0.5:8790");
263 }
264
265 #[test]
266 fn listing_is_oldest_first_and_survives_a_corrupt_record() {
267 let (dir, store) = store();
268 let mut old = store.create(&name("old").unwrap()).unwrap();
269 old.created_at = 100;
270 store.put(&old).unwrap();
271
272 let mut new = store.create(&name("new").unwrap()).unwrap();
273 new.created_at = 200;
274 store.put(&new).unwrap();
275
276 std::fs::write(dir.path().join("broken.json"), b"{not json").unwrap();
277
278 let listed = store.list();
279 assert_eq!(listed.len(), 2, "a corrupt record must not fail the sweep");
280 assert_eq!(listed[0].name, "old");
281 }
282
283 #[test]
284 fn an_account_name_cannot_become_a_path() {
285 let (_dir, store) = store();
286 let mut record = store.create(&name("alice").unwrap()).unwrap();
287 record.name = "../../escape".into();
288 assert!(store.put(&record).is_err());
289 }
290}