zuka
zuka/src/control/mod.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
branch5ae1edf5f64f7ed5c59a5d767931619d4cb931cc ▾
main All branches →
zuka/src/control/mod.rs
RSmod.rs33.0 KBDownload
1// The control plane.
2//
3// Owns accounts, provisions a container per account, and proxies every data-plane
4// request to the right one. It holds no repositories itself.
5//
6// Two rules shape everything here:
7//
8// * **Provisioning is never awaited by a request.** Creating an account writes a
9// `Provisioning` record and returns; a reconciler drives it to `Ready` or
10// `Failed`. Waiting synchronously on container boot ties a user-facing request
11// to a multi-second operation that can fail halfway.
12// * **The tenant is told who is calling by a signed, single-use, request-bound
13// assertion.** Not a shared secret: a tenant runs the account's own CI, so
14// anything symmetric in there forges every other tenant.
15
16pub mod accounts;
17pub mod provision;
18pub mod proxy;
19
20use crate::error::Result;
21use crate::git::validate::Name;
22use accounts::{AccountRecord, AccountStore};
23use provision::{Provisioner, State};
24use std::sync::Arc;
25use std::time::Duration;
26
27/// How often the reconciler sweeps accounts that are not yet `Ready`.
28const RECONCILE_INTERVAL: Duration = Duration::from_secs(10);
29
30/// Tenants upgraded per sweep.
31///
32/// Staggered rather than all at once: an upgrade restarts the tenant, and restarting
33/// every account simultaneously turns a routine deploy into an outage.
34const UPGRADES_PER_SWEEP: usize = 2;
35
36/// How long a tenant may sit unused before it is stopped.
37///
38/// Always-on would mean one permanently running container per account, including for
39/// people who push twice a year. Stopping is cheap and starting is a few seconds,
40/// which the caller sees as a `429` with a `Retry-After`.
41const IDLE_STOP: Duration = Duration::from_secs(1800);
42
43/// Drive every non-ready account forward, forever.
44pub async fn reconcile_forever(store: AccountStore, provisioner: Arc<dyn Provisioner>) {
45 eprintln!("[control] reconciler running every {RECONCILE_INTERVAL:?}");
46 loop {
47 let store = store.clone();
48 let provisioner = Arc::clone(&provisioner);
49 let outcome =
50 tokio::task::spawn_blocking(move || reconcile_once(&store, provisioner.as_ref())).await;
51
52 if let Err(e) = outcome {
53 // A panic in one sweep must not stop the reconciler; the next sweep is
54 // the retry.
55 eprintln!("[control] reconcile sweep failed: {e}");
56 }
57 tokio::time::sleep(RECONCILE_INTERVAL).await;
58 }
59}
60
61/// One reconciliation sweep. Returns how many accounts changed state.
62pub fn reconcile_once(store: &AccountStore, provisioner: &dyn Provisioner) -> usize {
63 reconcile_at(store, provisioner, crate::account::token::now_secs())
64}
65
66/// As [`reconcile_once`], with the clock injected so idle behaviour is testable.
67pub fn reconcile_at(store: &AccountStore, provisioner: &dyn Provisioner, now: u64) -> usize {
68 let mut changed = 0;
69 let mut upgraded = 0;
70
71 for mut record in store.list() {
72 if record.state == State::Ready {
73 // A tenant running an older binary is upgraded in place. Stopped tenants
74 // are skipped: they pick the new binary up when they are next started,
75 // because `create` is idempotent and re-pushes.
76 if record.binary_version.as_deref() != Some(crate::brand::build())
77 && upgraded < UPGRADES_PER_SWEEP
78 {
79 let Ok(account) = crate::git::validate::name(&record.name) else {
80 continue;
81 };
82 match provisioner.upgrade(&account) {
83 Ok(()) => {
84 record.binary_version = Some(crate::brand::build().to_string());
85 if store.put(&record).is_ok() {
86 upgraded += 1;
87 changed += 1;
88 eprintln!(
89 "[control] {} upgraded to {}",
90 record.name,
91 crate::brand::build()
92 );
93 }
94 }
95 // Left at the old version so the next sweep retries. A failed
96 // upgrade must not mark the tenant as upgraded.
97 Err(e) => eprintln!("[control] could not upgrade {}: {e}", record.name),
98 }
99 continue;
100 }
101
102 // Stop a tenant nobody has used. It comes back on the next request.
103 if now.saturating_sub(record.last_seen_at) >= IDLE_STOP.as_secs() {
104 let Ok(account) = crate::git::validate::name(&record.name) else {
105 continue;
106 };
107 match provisioner.stop(&account) {
108 Ok(()) => {
109 record.state = State::Stopped;
110 if store.put(&record).is_ok() {
111 changed += 1;
112 eprintln!("[control] {} -> Stopped (idle)", record.name);
113 }
114 }
115 Err(e) => eprintln!("[control] could not stop {}: {e}", record.name),
116 }
117 }
118 continue;
119 }
120
121 // A stopped tenant is only restarted once someone asks for it again.
122 if record.state == State::Stopped
123 && now.saturating_sub(record.last_seen_at) >= IDLE_STOP.as_secs()
124 {
125 continue;
126 }
127 let Ok(account) = crate::git::validate::name(&record.name) else {
128 continue;
129 };
130
131 let next = drive(&account, &record, provisioner);
132 if next.state != record.state
133 || next.endpoint != record.endpoint
134 || next.version != record.binary_version
135 {
136 record.state = next.state;
137 record.endpoint = next.endpoint;
138 record.detail = next.detail;
139 record.binary_version = next.version;
140 if store.put(&record).is_ok() {
141 changed += 1;
142 eprintln!("[control] {} -> {:?}", record.name, record.state);
143 }
144 }
145 }
146 changed
147}
148
149/// What one account's record should become.
150struct Next {
151 state: State,
152 endpoint: Option<String>,
153 detail: Option<String>,
154 version: Option<String>,
155}
156
157fn drive(account: &Name, record: &AccountRecord, provisioner: &dyn Provisioner) -> Next {
158 // A failed account is not retried automatically: repeatedly relaunching a
159 // container that cannot start burns the host. It is retried when someone asks,
160 // through `retry`.
161 if record.state == State::Failed {
162 return Next {
163 state: State::Failed,
164 endpoint: record.endpoint.clone(),
165 detail: record.detail.clone(),
166 version: record.binary_version.clone(),
167 };
168 }
169
170 match provisioner.status(account) {
171 // Already up: adopt whatever the provisioner reports.
172 Ok(Some(instance)) if instance.state == State::Ready => Next {
173 state: State::Ready,
174 endpoint: Some(instance.endpoint),
175 detail: None,
176 version: record.binary_version.clone(),
177 },
178 // Exists but stopped, and someone wants it: start it.
179 Ok(Some(_)) => match provisioner.start(account) {
180 Ok(instance) => Next {
181 state: State::Ready,
182 endpoint: Some(instance.endpoint),
183 detail: None,
184 version: record.binary_version.clone(),
185 },
186 Err(e) => Next {
187 state: State::Provisioning,
188 endpoint: record.endpoint.clone(),
189 detail: Some(e.detail_for_user()),
190 version: record.binary_version.clone(),
191 },
192 },
193 // Not there at all: create it.
194 // A fresh container gets the binary we are running, by construction.
195 Ok(None) => match provisioner.create(account) {
196 Ok(instance) => Next {
197 state: instance.state,
198 endpoint: Some(instance.endpoint),
199 detail: None,
200 version: Some(crate::brand::build().to_string()),
201 },
202 Err(e) => Next {
203 state: State::Failed,
204 endpoint: None,
205 detail: Some(e.detail_for_user()),
206 version: None,
207 },
208 },
209 Err(e) => Next {
210 state: State::Provisioning,
211 endpoint: record.endpoint.clone(),
212 detail: Some(e.detail_for_user()),
213 version: record.binary_version.clone(),
214 },
215 }
216}
217
218/// Move a failed account back into the queue.
219pub fn retry(store: &AccountStore, account: &Name) -> Result<()> {
220 let mut record = store
221 .get(account)?
222 .ok_or(crate::error::Error::NotFound("account"))?;
223 record.state = State::Provisioning;
224 record.detail = None;
225 store.put(&record)
226}
227
228// ── serving ────────────────────────────────────────────────────────────────
229
230use crate::account::assertion::Signer_;
231use crate::config::Config;
232use crate::http::response::{self, Body};
233use hyper::body::Incoming;
234use hyper::service::service_fn;
235use hyper::{Method, Request, Response, StatusCode};
236use hyper_util::rt::{TokioIo, TokioTimer};
237use std::convert::Infallible;
238use tokio::net::TcpListener;
239
240/// What the control plane needs to answer a request.
241pub struct ControlState {
242 pub config: Config,
243 pub accounts: AccountStore,
244 pub signer: Signer_,
245 pub client: reqwest::Client,
246 pub provisioner: Arc<dyn Provisioner>,
247 /// The control plane is the authentication authority.
248 ///
249 /// A tenant holds no tokens and cannot check one — it trusts the signed
250 /// assertion instead. So the token store lives here, and this is where a
251 /// credential is turned into an account.
252 tokens: crate::store::cached::Cached<crate::account::token::TokenFile>,
253}
254
255impl ControlState {
256 /// Authorise an operator action: creating, listing or destroying accounts.
257 ///
258 /// These are not tenant operations. Creating an account provisions a container
259 /// and deleting one destroys its repositories, so they cannot be reachable with
260 /// an ordinary account's credential — a tenant admin token must not be able to
261 /// delete a different tenant — and they certainly cannot be reachable with none.
262 ///
263 /// The operator is named by `<PREFIX>OPERATOR`. When it is unset every account
264 /// endpoint is refused, because the alternative default is the one this replaced:
265 /// a public host where a stranger could enumerate every tenant and delete them.
266 fn require_operator(&self, headers: &hyper::HeaderMap) -> Result<()> {
267 let configured = std::env::var(crate::brand::env_name("OPERATOR")).ok();
268 let tokens = self.tokens.get();
269 let identity =
270 crate::http::auth::resolve_local(&tokens, &crate::http::auth::extract(headers))?;
271 operator_allows(configured.as_deref(), &identity)
272 }
273
274 /// Resolve a caller to the account whose tenant should serve them.
275 fn account_for(&self, headers: &hyper::HeaderMap) -> Result<crate::git::validate::Name> {
276 let tokens = self.tokens.get();
277 let identity =
278 crate::http::auth::resolve_local(&tokens, &crate::http::auth::extract(headers))?;
279 Ok(identity.account)
280 }
281}
282
283/// Whether an identity may perform an operator action.
284///
285/// Separated from the request so the rule can be tested without a running control
286/// plane, an Incus host or a token file — the thing that must not be wrong here is
287/// the decision, not the plumbing that reaches it.
288fn operator_allows(configured: Option<&str>, identity: &crate::account::Identity) -> Result<()> {
289 let Some(operator) = configured.filter(|v| !v.is_empty()) else {
290 // No operator configured means nobody may manage accounts. The alternative
291 // default is the one this replaced: a public host on which a stranger could
292 // enumerate every tenant and delete them.
293 return Err(crate::error::Error::Forbidden);
294 };
295 let operator = crate::git::validate::name(operator)?;
296
297 // `require_account_admin` carries both remaining conditions: the identity must
298 // own the account, and a token confined to a repository list never qualifies —
299 // a repo-scoped token has no business provisioning containers.
300 identity.require_account_admin(&operator)
301}
302
303/// Serve the control plane.
304pub async fn serve(
305 config: Config,
306 accounts: AccountStore,
307 signer: Signer_,
308 provisioner: Arc<dyn Provisioner>,
309) -> anyhow::Result<()> {
310 let bind = config.bind;
311 let client = proxy::client(config.proxy_timeout)?;
312 let tokens = crate::store::cached::Cached::load(
313 config.tokens_file(),
314 crate::account::token::TokenFile::load,
315 )?;
316 let state = Arc::new(ControlState {
317 config,
318 accounts,
319 signer,
320 client,
321 provisioner,
322 tokens,
323 });
324
325 let listener = TcpListener::bind(bind).await?;
326 eprintln!("[control] listening on http://{bind}");
327
328 loop {
329 let (stream, _peer) = listener.accept().await?;
330 let state = Arc::clone(&state);
331 tokio::spawn(async move {
332 let io = TokioIo::new(stream);
333 let service = service_fn(move |req| handle(req, Arc::clone(&state)));
334 if let Err(e) = hyper::server::conn::http1::Builder::new()
335 .timer(TokioTimer::new())
336 .header_read_timeout(state_header_timeout())
337 .serve_connection(io, service)
338 .await
339 {
340 let message = e.to_string();
341 if !message.contains("closed") && !message.contains("reset") {
342 eprintln!("[control] conn: {message}");
343 }
344 }
345 });
346 }
347}
348
349fn state_header_timeout() -> Duration {
350 Duration::from_secs(15)
351}
352
353async fn handle(
354 request: Request<Incoming>,
355 state: Arc<ControlState>,
356) -> std::result::Result<Response<Body>, Infallible> {
357 let path = request.uri().path().to_string();
358 let method = request.method().clone();
359
360 let outcome = route(request, &state).await;
361 Ok(match outcome {
362 Ok(response) => response,
363 Err(error) => {
364 if let Some(cause) = error.cause() {
365 eprintln!("[control] {method} {path} -> {} {cause:#}", error.status());
366 }
367 response::problem(&error)
368 }
369 })
370}
371
372async fn route(request: Request<Incoming>, state: &ControlState) -> Result<Response<Body>> {
373 let path = request.uri().path().to_string();
374 let segments: Vec<&str> = path.split('/').filter(|s| !s.is_empty()).collect();
375
376 match (request.method(), segments.as_slice()) {
377 (&Method::GET, ["healthz"]) => Ok(response::json(
378 StatusCode::OK,
379 &serde_json::json!({
380 "status": "ok",
381 "mode": "control",
382 "version": crate::brand::VERSION,
383 "build": crate::brand::build(),
384 "accounts": state.accounts.list().len(),
385 }),
386 )),
387
388 // Accounts are the control plane's own resource; everything else is proxied.
389 (&Method::GET, ["v1", "accounts"]) => {
390 state.require_operator(request.headers())?;
391 Ok(response::json(
392 StatusCode::OK,
393 &serde_json::json!({ "items": state.accounts.list(), "truncated": false }),
394 ))
395 }
396 (&Method::POST, ["v1", "accounts", name]) => {
397 state.require_operator(request.headers())?;
398 let account = crate::git::validate::name(name)?;
399 let record = state.accounts.create(&account)?;
400 Ok(response::json(StatusCode::ACCEPTED, &record))
401 }
402 (&Method::GET, ["v1", "accounts", name]) => {
403 state.require_operator(request.headers())?;
404 let account = crate::git::validate::name(name)?;
405 let record = state
406 .accounts
407 .get(&account)?
408 .ok_or(crate::error::Error::NotFound("account"))?;
409 Ok(response::json(StatusCode::OK, &record))
410 }
411 (&Method::DELETE, ["v1", "accounts", name]) => {
412 state.require_operator(request.headers())?;
413 let account = crate::git::validate::name(name)?;
414 if !state.accounts.delete(&account)? {
415 return Err(crate::error::Error::NotFound("account"));
416 }
417 // The container goes with the record. Provisioning is idempotent, so a
418 // failure here is retried by the next sweep rather than blocking the
419 // response.
420 let provisioner = Arc::clone(&state.provisioner);
421 let name = account.clone();
422 tokio::task::spawn_blocking(move || {
423 if let Err(e) = provisioner.delete(&name) {
424 eprintln!("[control] could not remove {name}'s tenant: {e}");
425 }
426 });
427 Ok(response::no_content())
428 }
429 (&Method::POST, ["v1", "accounts", name, "retry"]) => {
430 state.require_operator(request.headers())?;
431 let account = crate::git::validate::name(name)?;
432 retry(&state.accounts, &account)?;
433 Ok(response::no_content())
434 }
435
436 // Everything else belongs to a tenant. Which one is decided by the
437 // credential, not by the path: a path segment is a claim, a token is not.
438 (_, ["v1", ..]) | (_, ["mcp"]) => {
439 let account = state.account_for(request.headers())?;
440 let endpoint = state.accounts.endpoint(&account)?;
441 proxy::forward(
442 &state.client,
443 &state.signer,
444 &endpoint,
445 account.as_str(),
446 request,
447 state.config.max_body_bytes,
448 )
449 .await
450 }
451
452 // Git paths name the account, which must match the credential's — otherwise
453 // any valid token could reach any account's repositories.
454 (_, [path_account, repo, ..]) if repo.ends_with(".git") => {
455 let claimed = crate::git::validate::name(path_account)?;
456 let name = state.account_for(request.headers())?;
457 if claimed != name {
458 return Err(crate::error::Error::NotFound("repository"));
459 }
460 let endpoint = state.accounts.endpoint(&name)?;
461 proxy::forward(
462 &state.client,
463 &state.signer,
464 &endpoint,
465 name.as_str(),
466 request,
467 state.config.max_body_bytes,
468 )
469 .await
470 }
471
472 // The stylesheet is static and identical in every build, so the control
473 // plane answers it rather than waking a tenant for it.
474 (&Method::GET, _) if crate::web::is_style_path(&path) => {
475 Ok(response::css(crate::web::STYLE))
476 }
477
478 // `/` features one repository, which lives in a tenant like any other.
479 (&Method::GET, []) => match state.config.home_repo() {
480 Some((account, repo)) => Ok(response::redirect(&format!("/{account}/{repo}"))),
481 None => Ok(crate::web::welcome()),
482 },
483
484 // The browser surface. Unlike the git and API paths, the account comes from
485 // the path rather than the credential, because the whole point is that a
486 // stranger with no credential can reach a public repository.
487 (&Method::GET, [path_account, _repo, ..]) if crate::web::owns(&path) => {
488 let claimed = crate::git::validate::name(path_account)?;
489 let endpoint = state.accounts.endpoint(&claimed)?;
490
491 // Vouch only when the credential belongs to the account being browsed.
492 // A token is worth nothing in another account's tenant, so presenting
493 // one must not turn into an assertion naming that other account — and
494 // an anonymous forward is exactly right for browsing someone else's
495 // public repository.
496 let vouch = match state.account_for(request.headers()) {
497 Ok(name) if name == claimed => true,
498 Ok(_) => false,
499 // No credential at all is the anonymous case. A credential that was
500 // presented and rejected is still an error.
501 Err(_) if !crate::http::auth::presented(request.headers()) => false,
502 Err(e) => return Err(e),
503 };
504
505 if vouch {
506 proxy::forward(
507 &state.client,
508 &state.signer,
509 &endpoint,
510 claimed.as_str(),
511 request,
512 state.config.max_body_bytes,
513 )
514 .await
515 } else {
516 proxy::forward_anonymous(
517 &state.client,
518 &endpoint,
519 request,
520 state.config.max_body_bytes,
521 )
522 .await
523 }
524 }
525
526 _ => Err(crate::error::Error::NotFound("route")),
527 }
528}
529
530#[cfg(test)]
531mod tests {
532 use super::*;
533 use provision::FakeProvisioner;
534 use std::sync::atomic::Ordering;
535
536 fn setup() -> (tempfile::TempDir, AccountStore, Arc<FakeProvisioner>) {
537 let dir = tempfile::tempdir().unwrap();
538 let store = AccountStore::open(dir.path()).unwrap();
539 (dir, store, Arc::new(FakeProvisioner::new()))
540 }
541
542 fn identity_for(
543 account: &str,
544 scopes: Vec<crate::account::token::Scope>,
545 ) -> crate::account::Identity {
546 crate::account::Identity {
547 account: crate::git::validate::name(account).unwrap(),
548 token_id: "t".into(),
549 scopes,
550 repos: Vec::new(),
551 }
552 }
553
554 #[test]
555 fn account_management_is_refused_when_no_operator_is_named() {
556 use crate::account::token::Scope;
557 // The default has to be closed. Creating an account provisions a container
558 // and deleting one destroys its repositories, and both were once reachable
559 // with no credential at all.
560 let admin = identity_for("alice", vec![Scope::Admin]);
561 assert!(operator_allows(None, &admin).is_err());
562 assert!(operator_allows(Some(""), &admin).is_err());
563 }
564
565 #[test]
566 fn only_the_named_operator_may_manage_accounts() {
567 use crate::account::token::Scope;
568 assert!(operator_allows(Some("ops"), &identity_for("ops", vec![Scope::Admin])).is_ok());
569
570 // A tenant's own admin token must not reach account management, or any
571 // account could destroy any other.
572 assert!(operator_allows(Some("ops"), &identity_for("alice", vec![Scope::Admin])).is_err());
573
574 // The operator account still needs admin scope.
575 assert!(operator_allows(Some("ops"), &identity_for("ops", vec![Scope::RepoRead])).is_err());
576 assert!(
577 operator_allows(Some("ops"), &identity_for("ops", vec![Scope::RepoWrite])).is_err()
578 );
579 }
580
581 #[test]
582 fn an_operator_token_confined_to_repositories_is_refused() {
583 use crate::account::token::Scope;
584 // Confinement must not be escapable by an operation that names no
585 // repository — provisioning containers is not repository work.
586 let confined = crate::account::Identity {
587 repos: vec![crate::git::validate::name("site").unwrap()],
588 ..identity_for("ops", vec![Scope::Admin])
589 };
590 assert!(operator_allows(Some("ops"), &confined).is_err());
591 }
592
593 #[test]
594 fn a_new_account_is_driven_to_ready_without_a_request_waiting() {
595 let (_dir, store, fake) = setup();
596 let alice = crate::git::validate::name("alice").unwrap();
597
598 // Creation records intent and returns; nothing is provisioned yet.
599 store.create(&alice).unwrap();
600 assert_eq!(
601 store.get(&alice).unwrap().unwrap().state,
602 State::Provisioning
603 );
604
605 // First sweep creates it, second observes it running.
606 reconcile_once(&store, fake.as_ref());
607 reconcile_once(&store, fake.as_ref());
608
609 let record = store.get(&alice).unwrap().unwrap();
610 assert_eq!(record.state, State::Ready);
611 assert!(record.endpoint.is_some(), "a ready tenant must be routable");
612 }
613
614 #[test]
615 fn a_provisioning_failure_is_recorded_with_its_reason() {
616 let (_dir, store, fake) = setup();
617 let alice = crate::git::validate::name("alice").unwrap();
618 store.create(&alice).unwrap();
619
620 fake.fail_next.store(true, Ordering::SeqCst);
621 reconcile_once(&store, fake.as_ref());
622
623 let record = store.get(&alice).unwrap().unwrap();
624 assert_eq!(record.state, State::Failed);
625 assert!(record.detail.is_some(), "a failure must say why");
626 }
627
628 #[test]
629 fn a_failed_account_is_not_retried_until_asked() {
630 let (_dir, store, fake) = setup();
631 let alice = crate::git::validate::name("alice").unwrap();
632 store.create(&alice).unwrap();
633
634 fake.fail_next.store(true, Ordering::SeqCst);
635 reconcile_once(&store, fake.as_ref());
636 assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Failed);
637
638 // Sweeping again must not relaunch: a container that cannot start would be
639 // relaunched forever.
640 assert_eq!(reconcile_once(&store, fake.as_ref()), 0);
641 assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Failed);
642
643 retry(&store, &alice).unwrap();
644 reconcile_once(&store, fake.as_ref());
645 reconcile_once(&store, fake.as_ref());
646 assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Ready);
647 }
648
649 #[test]
650 fn a_stopped_tenant_is_started_again() {
651 let (_dir, store, fake) = setup();
652 let alice = crate::git::validate::name("alice").unwrap();
653 store.create(&alice).unwrap();
654 reconcile_once(&store, fake.as_ref());
655 reconcile_once(&store, fake.as_ref());
656
657 fake.stop(&alice).unwrap();
658 let mut record = store.get(&alice).unwrap().unwrap();
659 record.state = State::Stopped;
660 store.put(&record).unwrap();
661
662 reconcile_once(&store, fake.as_ref());
663 assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Ready);
664 }
665
666 /// Drive an account all the way to Ready on the current version.
667 fn settle(store: &AccountStore, fake: &FakeProvisioner, name: &str) {
668 let account = crate::git::validate::name(name).unwrap();
669 store.create(&account).unwrap();
670 reconcile_once(store, fake);
671 reconcile_once(store, fake);
672 let mut record = store.get(&account).unwrap().unwrap();
673 record.last_seen_at = crate::account::token::now_secs();
674 store.put(&record).unwrap();
675 }
676
677 #[test]
678 fn a_new_tenant_records_the_version_it_was_built_with() {
679 let (_dir, store, fake) = setup();
680 settle(&store, &fake, "alice");
681
682 let record = store
683 .get(&crate::git::validate::name("alice").unwrap())
684 .unwrap()
685 .unwrap();
686 assert_eq!(record.state, State::Ready);
687 assert_eq!(
688 record.binary_version.as_deref(),
689 Some(crate::brand::build())
690 );
691 assert!(
692 fake.upgrades.lock().unwrap().is_empty(),
693 "a container built from the current binary needs no upgrade"
694 );
695 }
696
697 #[test]
698 fn a_tenant_on_an_old_binary_is_upgraded_in_place() {
699 let (_dir, store, fake) = setup();
700 settle(&store, &fake, "alice");
701 let alice = crate::git::validate::name("alice").unwrap();
702
703 // What a deploy looks like: the control plane is newer than the tenant.
704 let mut record = store.get(&alice).unwrap().unwrap();
705 record.binary_version = Some("0.0.1-old".into());
706 store.put(&record).unwrap();
707
708 reconcile_once(&store, fake.as_ref());
709
710 assert_eq!(fake.upgrades.lock().unwrap().as_slice(), ["alice"]);
711 assert_eq!(
712 store
713 .get(&alice)
714 .unwrap()
715 .unwrap()
716 .binary_version
717 .as_deref(),
718 Some(crate::brand::build())
719 );
720 // Upgrading must not have disturbed the tenant's state.
721 assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Ready);
722 }
723
724 #[test]
725 fn a_failed_upgrade_leaves_the_old_version_so_the_next_sweep_retries() {
726 use std::sync::atomic::Ordering;
727 let (_dir, store, fake) = setup();
728 settle(&store, &fake, "alice");
729 let alice = crate::git::validate::name("alice").unwrap();
730
731 let mut record = store.get(&alice).unwrap().unwrap();
732 record.binary_version = Some("0.0.1-old".into());
733 store.put(&record).unwrap();
734
735 fake.fail_next.store(true, Ordering::SeqCst);
736 reconcile_once(&store, fake.as_ref());
737 assert_eq!(
738 store
739 .get(&alice)
740 .unwrap()
741 .unwrap()
742 .binary_version
743 .as_deref(),
744 Some("0.0.1-old"),
745 "a failed upgrade must not be recorded as done"
746 );
747
748 reconcile_once(&store, fake.as_ref());
749 assert_eq!(
750 store
751 .get(&alice)
752 .unwrap()
753 .unwrap()
754 .binary_version
755 .as_deref(),
756 Some(crate::brand::build()),
757 "the next sweep must retry"
758 );
759 }
760
761 #[test]
762 fn upgrades_are_staggered_so_a_deploy_is_not_an_outage() {
763 let (_dir, store, fake) = setup();
764 let names = ["a1", "a2", "a3", "a4", "a5"];
765
766 // Settle every account first: `settle` reconciles, so backdating inside the
767 // loop would let those sweeps do the upgrading and hide the stagger.
768 for name in names {
769 settle(&store, &fake, name);
770 }
771 for name in names {
772 let account = crate::git::validate::name(name).unwrap();
773 let mut record = store.get(&account).unwrap().unwrap();
774 record.binary_version = Some("0.0.1-old".into());
775 store.put(&record).unwrap();
776 }
777 fake.upgrades.lock().unwrap().clear();
778
779 reconcile_once(&store, fake.as_ref());
780 assert_eq!(
781 fake.upgrades.lock().unwrap().len(),
782 UPGRADES_PER_SWEEP,
783 "restarting every tenant at once turns a deploy into an outage"
784 );
785
786 // Successive sweeps finish the rest.
787 for _ in 0..3 {
788 reconcile_once(&store, fake.as_ref());
789 }
790 assert_eq!(fake.upgrades.lock().unwrap().len(), 5);
791 }
792
793 #[test]
794 fn a_stopped_tenant_is_not_upgraded_until_it_is_needed() {
795 let (_dir, store, fake) = setup();
796 settle(&store, &fake, "alice");
797 let alice = crate::git::validate::name("alice").unwrap();
798
799 let mut record = store.get(&alice).unwrap().unwrap();
800 record.state = State::Stopped;
801 record.binary_version = Some("0.0.1-old".into());
802 store.put(&record).unwrap();
803
804 reconcile_once(&store, fake.as_ref());
805 assert!(
806 fake.upgrades.lock().unwrap().is_empty(),
807 "a stopped tenant picks up the new binary when it starts; \
808 restarting it now would wake it for nothing"
809 );
810 }
811
812 #[test]
813 fn an_idle_tenant_is_stopped_and_woken_by_the_next_request() {
814 let (_dir, store, fake) = setup();
815 let alice = crate::git::validate::name("alice").unwrap();
816 store.create(&alice).unwrap();
817 reconcile_once(&store, fake.as_ref());
818 reconcile_once(&store, fake.as_ref());
819 assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Ready);
820
821 // Long enough that nobody has used it.
822 let later = crate::account::token::now_secs() + IDLE_STOP.as_secs() + 1;
823 reconcile_at(&store, fake.as_ref(), later);
824 assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Stopped);
825
826 // Asking for it refreshes the timestamp and reports "come back shortly".
827 let err = store.endpoint(&alice).unwrap_err();
828 assert_eq!(err.status(), 429);
829
830 reconcile_once(&store, fake.as_ref());
831 assert_eq!(
832 store.get(&alice).unwrap().unwrap().state,
833 State::Ready,
834 "a request must bring a stopped tenant back"
835 );
836 }
837
838 #[test]
839 fn a_busy_tenant_is_not_stopped() {
840 let (_dir, store, fake) = setup();
841 let alice = crate::git::validate::name("alice").unwrap();
842 store.create(&alice).unwrap();
843 reconcile_once(&store, fake.as_ref());
844 reconcile_once(&store, fake.as_ref());
845
846 // Used just now.
847 let mut record = store.get(&alice).unwrap().unwrap();
848 record.last_seen_at = crate::account::token::now_secs();
849 store.put(&record).unwrap();
850
851 reconcile_once(&store, fake.as_ref());
852 assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Ready);
853 }
854
855 #[test]
856 fn a_ready_account_is_left_alone() {
857 let (_dir, store, fake) = setup();
858 let alice = crate::git::validate::name("alice").unwrap();
859 store.create(&alice).unwrap();
860 reconcile_once(&store, fake.as_ref());
861 reconcile_once(&store, fake.as_ref());
862
863 assert_eq!(
864 reconcile_once(&store, fake.as_ref()),
865 0,
866 "a settled account must not be touched every sweep"
867 );
868 }
869}