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
branch8d539ae691178f667dc895389953e0990166f1a5 ▾
main All branches →
zuka/src/control/mod.rs
RSmod.rs28.6 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 /// Resolve a caller to the account whose tenant should serve them.
257 fn account_for(&self, headers: &hyper::HeaderMap) -> Result<crate::git::validate::Name> {
258 let tokens = self.tokens.get();
259 let identity =
260 crate::http::auth::resolve_local(&tokens, &crate::http::auth::extract(headers))?;
261 Ok(identity.account)
262 }
263}
264
265/// Serve the control plane.
266pub async fn serve(
267 config: Config,
268 accounts: AccountStore,
269 signer: Signer_,
270 provisioner: Arc<dyn Provisioner>,
271) -> anyhow::Result<()> {
272 let bind = config.bind;
273 let client = proxy::client(config.proxy_timeout)?;
274 let tokens = crate::store::cached::Cached::load(
275 config.tokens_file(),
276 crate::account::token::TokenFile::load,
277 )?;
278 let state = Arc::new(ControlState {
279 config,
280 accounts,
281 signer,
282 client,
283 provisioner,
284 tokens,
285 });
286
287 let listener = TcpListener::bind(bind).await?;
288 eprintln!("[control] listening on http://{bind}");
289
290 loop {
291 let (stream, _peer) = listener.accept().await?;
292 let state = Arc::clone(&state);
293 tokio::spawn(async move {
294 let io = TokioIo::new(stream);
295 let service = service_fn(move |req| handle(req, Arc::clone(&state)));
296 if let Err(e) = hyper::server::conn::http1::Builder::new()
297 .timer(TokioTimer::new())
298 .header_read_timeout(state_header_timeout())
299 .serve_connection(io, service)
300 .await
301 {
302 let message = e.to_string();
303 if !message.contains("closed") && !message.contains("reset") {
304 eprintln!("[control] conn: {message}");
305 }
306 }
307 });
308 }
309}
310
311fn state_header_timeout() -> Duration {
312 Duration::from_secs(15)
313}
314
315async fn handle(
316 request: Request<Incoming>,
317 state: Arc<ControlState>,
318) -> std::result::Result<Response<Body>, Infallible> {
319 let path = request.uri().path().to_string();
320 let method = request.method().clone();
321
322 let outcome = route(request, &state).await;
323 Ok(match outcome {
324 Ok(response) => response,
325 Err(error) => {
326 if let Some(cause) = error.cause() {
327 eprintln!("[control] {method} {path} -> {} {cause:#}", error.status());
328 }
329 response::problem(&error)
330 }
331 })
332}
333
334async fn route(request: Request<Incoming>, state: &ControlState) -> Result<Response<Body>> {
335 let path = request.uri().path().to_string();
336 let segments: Vec<&str> = path.split('/').filter(|s| !s.is_empty()).collect();
337
338 match (request.method(), segments.as_slice()) {
339 (&Method::GET, ["healthz"]) => Ok(response::json(
340 StatusCode::OK,
341 &serde_json::json!({
342 "status": "ok",
343 "mode": "control",
344 "version": crate::brand::VERSION,
345 "build": crate::brand::build(),
346 "accounts": state.accounts.list().len(),
347 }),
348 )),
349
350 // Accounts are the control plane's own resource; everything else is proxied.
351 (&Method::GET, ["v1", "accounts"]) => Ok(response::json(
352 StatusCode::OK,
353 &serde_json::json!({ "items": state.accounts.list(), "truncated": false }),
354 )),
355 (&Method::POST, ["v1", "accounts", name]) => {
356 let account = crate::git::validate::name(name)?;
357 let record = state.accounts.create(&account)?;
358 Ok(response::json(StatusCode::ACCEPTED, &record))
359 }
360 (&Method::GET, ["v1", "accounts", name]) => {
361 let account = crate::git::validate::name(name)?;
362 let record = state
363 .accounts
364 .get(&account)?
365 .ok_or(crate::error::Error::NotFound("account"))?;
366 Ok(response::json(StatusCode::OK, &record))
367 }
368 (&Method::DELETE, ["v1", "accounts", name]) => {
369 let account = crate::git::validate::name(name)?;
370 if !state.accounts.delete(&account)? {
371 return Err(crate::error::Error::NotFound("account"));
372 }
373 // The container goes with the record. Provisioning is idempotent, so a
374 // failure here is retried by the next sweep rather than blocking the
375 // response.
376 let provisioner = Arc::clone(&state.provisioner);
377 let name = account.clone();
378 tokio::task::spawn_blocking(move || {
379 if let Err(e) = provisioner.delete(&name) {
380 eprintln!("[control] could not remove {name}'s tenant: {e}");
381 }
382 });
383 Ok(response::no_content())
384 }
385 (&Method::POST, ["v1", "accounts", name, "retry"]) => {
386 let account = crate::git::validate::name(name)?;
387 retry(&state.accounts, &account)?;
388 Ok(response::no_content())
389 }
390
391 // Everything else belongs to a tenant. Which one is decided by the
392 // credential, not by the path: a path segment is a claim, a token is not.
393 (_, ["v1", ..]) | (_, ["mcp"]) => {
394 let account = state.account_for(request.headers())?;
395 let endpoint = state.accounts.endpoint(&account)?;
396 proxy::forward(
397 &state.client,
398 &state.signer,
399 &endpoint,
400 account.as_str(),
401 request,
402 state.config.max_body_bytes,
403 )
404 .await
405 }
406
407 // Git paths name the account, which must match the credential's — otherwise
408 // any valid token could reach any account's repositories.
409 (_, [path_account, repo, ..]) if repo.ends_with(".git") => {
410 let claimed = crate::git::validate::name(path_account)?;
411 let name = state.account_for(request.headers())?;
412 if claimed != name {
413 return Err(crate::error::Error::NotFound("repository"));
414 }
415 let endpoint = state.accounts.endpoint(&name)?;
416 proxy::forward(
417 &state.client,
418 &state.signer,
419 &endpoint,
420 name.as_str(),
421 request,
422 state.config.max_body_bytes,
423 )
424 .await
425 }
426
427 // The stylesheet is static and identical in every build, so the control
428 // plane answers it rather than waking a tenant for it.
429 (&Method::GET, _) if crate::web::is_style_path(&path) => {
430 Ok(response::css(crate::web::STYLE))
431 }
432
433 // `/` features one repository, which lives in a tenant like any other.
434 (&Method::GET, []) => match state.config.home_repo() {
435 Some((account, repo)) => Ok(response::redirect(&format!("/{account}/{repo}"))),
436 None => Ok(crate::web::welcome()),
437 },
438
439 // The browser surface. Unlike the git and API paths, the account comes from
440 // the path rather than the credential, because the whole point is that a
441 // stranger with no credential can reach a public repository.
442 (&Method::GET, [path_account, _repo, ..]) if crate::web::owns(&path) => {
443 let claimed = crate::git::validate::name(path_account)?;
444 let endpoint = state.accounts.endpoint(&claimed)?;
445
446 // Vouch only when the credential belongs to the account being browsed.
447 // A token is worth nothing in another account's tenant, so presenting
448 // one must not turn into an assertion naming that other account — and
449 // an anonymous forward is exactly right for browsing someone else's
450 // public repository.
451 let vouch = match state.account_for(request.headers()) {
452 Ok(name) if name == claimed => true,
453 Ok(_) => false,
454 // No credential at all is the anonymous case. A credential that was
455 // presented and rejected is still an error.
456 Err(_) if !crate::http::auth::presented(request.headers()) => false,
457 Err(e) => return Err(e),
458 };
459
460 if vouch {
461 proxy::forward(
462 &state.client,
463 &state.signer,
464 &endpoint,
465 claimed.as_str(),
466 request,
467 state.config.max_body_bytes,
468 )
469 .await
470 } else {
471 proxy::forward_anonymous(
472 &state.client,
473 &endpoint,
474 request,
475 state.config.max_body_bytes,
476 )
477 .await
478 }
479 }
480
481 _ => Err(crate::error::Error::NotFound("route")),
482 }
483}
484
485#[cfg(test)]
486mod tests {
487 use super::*;
488 use provision::FakeProvisioner;
489 use std::sync::atomic::Ordering;
490
491 fn setup() -> (tempfile::TempDir, AccountStore, Arc<FakeProvisioner>) {
492 let dir = tempfile::tempdir().unwrap();
493 let store = AccountStore::open(dir.path()).unwrap();
494 (dir, store, Arc::new(FakeProvisioner::new()))
495 }
496
497 #[test]
498 fn a_new_account_is_driven_to_ready_without_a_request_waiting() {
499 let (_dir, store, fake) = setup();
500 let alice = crate::git::validate::name("alice").unwrap();
501
502 // Creation records intent and returns; nothing is provisioned yet.
503 store.create(&alice).unwrap();
504 assert_eq!(
505 store.get(&alice).unwrap().unwrap().state,
506 State::Provisioning
507 );
508
509 // First sweep creates it, second observes it running.
510 reconcile_once(&store, fake.as_ref());
511 reconcile_once(&store, fake.as_ref());
512
513 let record = store.get(&alice).unwrap().unwrap();
514 assert_eq!(record.state, State::Ready);
515 assert!(record.endpoint.is_some(), "a ready tenant must be routable");
516 }
517
518 #[test]
519 fn a_provisioning_failure_is_recorded_with_its_reason() {
520 let (_dir, store, fake) = setup();
521 let alice = crate::git::validate::name("alice").unwrap();
522 store.create(&alice).unwrap();
523
524 fake.fail_next.store(true, Ordering::SeqCst);
525 reconcile_once(&store, fake.as_ref());
526
527 let record = store.get(&alice).unwrap().unwrap();
528 assert_eq!(record.state, State::Failed);
529 assert!(record.detail.is_some(), "a failure must say why");
530 }
531
532 #[test]
533 fn a_failed_account_is_not_retried_until_asked() {
534 let (_dir, store, fake) = setup();
535 let alice = crate::git::validate::name("alice").unwrap();
536 store.create(&alice).unwrap();
537
538 fake.fail_next.store(true, Ordering::SeqCst);
539 reconcile_once(&store, fake.as_ref());
540 assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Failed);
541
542 // Sweeping again must not relaunch: a container that cannot start would be
543 // relaunched forever.
544 assert_eq!(reconcile_once(&store, fake.as_ref()), 0);
545 assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Failed);
546
547 retry(&store, &alice).unwrap();
548 reconcile_once(&store, fake.as_ref());
549 reconcile_once(&store, fake.as_ref());
550 assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Ready);
551 }
552
553 #[test]
554 fn a_stopped_tenant_is_started_again() {
555 let (_dir, store, fake) = setup();
556 let alice = crate::git::validate::name("alice").unwrap();
557 store.create(&alice).unwrap();
558 reconcile_once(&store, fake.as_ref());
559 reconcile_once(&store, fake.as_ref());
560
561 fake.stop(&alice).unwrap();
562 let mut record = store.get(&alice).unwrap().unwrap();
563 record.state = State::Stopped;
564 store.put(&record).unwrap();
565
566 reconcile_once(&store, fake.as_ref());
567 assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Ready);
568 }
569
570 /// Drive an account all the way to Ready on the current version.
571 fn settle(store: &AccountStore, fake: &FakeProvisioner, name: &str) {
572 let account = crate::git::validate::name(name).unwrap();
573 store.create(&account).unwrap();
574 reconcile_once(store, fake);
575 reconcile_once(store, fake);
576 let mut record = store.get(&account).unwrap().unwrap();
577 record.last_seen_at = crate::account::token::now_secs();
578 store.put(&record).unwrap();
579 }
580
581 #[test]
582 fn a_new_tenant_records_the_version_it_was_built_with() {
583 let (_dir, store, fake) = setup();
584 settle(&store, &fake, "alice");
585
586 let record = store
587 .get(&crate::git::validate::name("alice").unwrap())
588 .unwrap()
589 .unwrap();
590 assert_eq!(record.state, State::Ready);
591 assert_eq!(
592 record.binary_version.as_deref(),
593 Some(crate::brand::build())
594 );
595 assert!(
596 fake.upgrades.lock().unwrap().is_empty(),
597 "a container built from the current binary needs no upgrade"
598 );
599 }
600
601 #[test]
602 fn a_tenant_on_an_old_binary_is_upgraded_in_place() {
603 let (_dir, store, fake) = setup();
604 settle(&store, &fake, "alice");
605 let alice = crate::git::validate::name("alice").unwrap();
606
607 // What a deploy looks like: the control plane is newer than the tenant.
608 let mut record = store.get(&alice).unwrap().unwrap();
609 record.binary_version = Some("0.0.1-old".into());
610 store.put(&record).unwrap();
611
612 reconcile_once(&store, fake.as_ref());
613
614 assert_eq!(fake.upgrades.lock().unwrap().as_slice(), ["alice"]);
615 assert_eq!(
616 store
617 .get(&alice)
618 .unwrap()
619 .unwrap()
620 .binary_version
621 .as_deref(),
622 Some(crate::brand::build())
623 );
624 // Upgrading must not have disturbed the tenant's state.
625 assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Ready);
626 }
627
628 #[test]
629 fn a_failed_upgrade_leaves_the_old_version_so_the_next_sweep_retries() {
630 use std::sync::atomic::Ordering;
631 let (_dir, store, fake) = setup();
632 settle(&store, &fake, "alice");
633 let alice = crate::git::validate::name("alice").unwrap();
634
635 let mut record = store.get(&alice).unwrap().unwrap();
636 record.binary_version = Some("0.0.1-old".into());
637 store.put(&record).unwrap();
638
639 fake.fail_next.store(true, Ordering::SeqCst);
640 reconcile_once(&store, fake.as_ref());
641 assert_eq!(
642 store
643 .get(&alice)
644 .unwrap()
645 .unwrap()
646 .binary_version
647 .as_deref(),
648 Some("0.0.1-old"),
649 "a failed upgrade must not be recorded as done"
650 );
651
652 reconcile_once(&store, fake.as_ref());
653 assert_eq!(
654 store
655 .get(&alice)
656 .unwrap()
657 .unwrap()
658 .binary_version
659 .as_deref(),
660 Some(crate::brand::build()),
661 "the next sweep must retry"
662 );
663 }
664
665 #[test]
666 fn upgrades_are_staggered_so_a_deploy_is_not_an_outage() {
667 let (_dir, store, fake) = setup();
668 let names = ["a1", "a2", "a3", "a4", "a5"];
669
670 // Settle every account first: `settle` reconciles, so backdating inside the
671 // loop would let those sweeps do the upgrading and hide the stagger.
672 for name in names {
673 settle(&store, &fake, name);
674 }
675 for name in names {
676 let account = crate::git::validate::name(name).unwrap();
677 let mut record = store.get(&account).unwrap().unwrap();
678 record.binary_version = Some("0.0.1-old".into());
679 store.put(&record).unwrap();
680 }
681 fake.upgrades.lock().unwrap().clear();
682
683 reconcile_once(&store, fake.as_ref());
684 assert_eq!(
685 fake.upgrades.lock().unwrap().len(),
686 UPGRADES_PER_SWEEP,
687 "restarting every tenant at once turns a deploy into an outage"
688 );
689
690 // Successive sweeps finish the rest.
691 for _ in 0..3 {
692 reconcile_once(&store, fake.as_ref());
693 }
694 assert_eq!(fake.upgrades.lock().unwrap().len(), 5);
695 }
696
697 #[test]
698 fn a_stopped_tenant_is_not_upgraded_until_it_is_needed() {
699 let (_dir, store, fake) = setup();
700 settle(&store, &fake, "alice");
701 let alice = crate::git::validate::name("alice").unwrap();
702
703 let mut record = store.get(&alice).unwrap().unwrap();
704 record.state = State::Stopped;
705 record.binary_version = Some("0.0.1-old".into());
706 store.put(&record).unwrap();
707
708 reconcile_once(&store, fake.as_ref());
709 assert!(
710 fake.upgrades.lock().unwrap().is_empty(),
711 "a stopped tenant picks up the new binary when it starts; \
712 restarting it now would wake it for nothing"
713 );
714 }
715
716 #[test]
717 fn an_idle_tenant_is_stopped_and_woken_by_the_next_request() {
718 let (_dir, store, fake) = setup();
719 let alice = crate::git::validate::name("alice").unwrap();
720 store.create(&alice).unwrap();
721 reconcile_once(&store, fake.as_ref());
722 reconcile_once(&store, fake.as_ref());
723 assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Ready);
724
725 // Long enough that nobody has used it.
726 let later = crate::account::token::now_secs() + IDLE_STOP.as_secs() + 1;
727 reconcile_at(&store, fake.as_ref(), later);
728 assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Stopped);
729
730 // Asking for it refreshes the timestamp and reports "come back shortly".
731 let err = store.endpoint(&alice).unwrap_err();
732 assert_eq!(err.status(), 429);
733
734 reconcile_once(&store, fake.as_ref());
735 assert_eq!(
736 store.get(&alice).unwrap().unwrap().state,
737 State::Ready,
738 "a request must bring a stopped tenant back"
739 );
740 }
741
742 #[test]
743 fn a_busy_tenant_is_not_stopped() {
744 let (_dir, store, fake) = setup();
745 let alice = crate::git::validate::name("alice").unwrap();
746 store.create(&alice).unwrap();
747 reconcile_once(&store, fake.as_ref());
748 reconcile_once(&store, fake.as_ref());
749
750 // Used just now.
751 let mut record = store.get(&alice).unwrap().unwrap();
752 record.last_seen_at = crate::account::token::now_secs();
753 store.put(&record).unwrap();
754
755 reconcile_once(&store, fake.as_ref());
756 assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Ready);
757 }
758
759 #[test]
760 fn a_ready_account_is_left_alone() {
761 let (_dir, store, fake) = setup();
762 let alice = crate::git::validate::name("alice").unwrap();
763 store.create(&alice).unwrap();
764 reconcile_once(&store, fake.as_ref());
765 reconcile_once(&store, fake.as_ref());
766
767 assert_eq!(
768 reconcile_once(&store, fake.as_ref()),
769 0,
770 "a settled account must not be touched every sweep"
771 );
772 }
773}