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
branchbc985d6bbd321be340e3750b3da4734ee1553cbd ▾
main All branches →
zuka/src/control/mod.rs
RSmod.rs26.2 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 _ => Err(crate::error::Error::NotFound("route")),
428 }
429}
430
431#[cfg(test)]
432mod tests {
433 use super::*;
434 use provision::FakeProvisioner;
435 use std::sync::atomic::Ordering;
436
437 fn setup() -> (tempfile::TempDir, AccountStore, Arc<FakeProvisioner>) {
438 let dir = tempfile::tempdir().unwrap();
439 let store = AccountStore::open(dir.path()).unwrap();
440 (dir, store, Arc::new(FakeProvisioner::new()))
441 }
442
443 #[test]
444 fn a_new_account_is_driven_to_ready_without_a_request_waiting() {
445 let (_dir, store, fake) = setup();
446 let alice = crate::git::validate::name("alice").unwrap();
447
448 // Creation records intent and returns; nothing is provisioned yet.
449 store.create(&alice).unwrap();
450 assert_eq!(
451 store.get(&alice).unwrap().unwrap().state,
452 State::Provisioning
453 );
454
455 // First sweep creates it, second observes it running.
456 reconcile_once(&store, fake.as_ref());
457 reconcile_once(&store, fake.as_ref());
458
459 let record = store.get(&alice).unwrap().unwrap();
460 assert_eq!(record.state, State::Ready);
461 assert!(record.endpoint.is_some(), "a ready tenant must be routable");
462 }
463
464 #[test]
465 fn a_provisioning_failure_is_recorded_with_its_reason() {
466 let (_dir, store, fake) = setup();
467 let alice = crate::git::validate::name("alice").unwrap();
468 store.create(&alice).unwrap();
469
470 fake.fail_next.store(true, Ordering::SeqCst);
471 reconcile_once(&store, fake.as_ref());
472
473 let record = store.get(&alice).unwrap().unwrap();
474 assert_eq!(record.state, State::Failed);
475 assert!(record.detail.is_some(), "a failure must say why");
476 }
477
478 #[test]
479 fn a_failed_account_is_not_retried_until_asked() {
480 let (_dir, store, fake) = setup();
481 let alice = crate::git::validate::name("alice").unwrap();
482 store.create(&alice).unwrap();
483
484 fake.fail_next.store(true, Ordering::SeqCst);
485 reconcile_once(&store, fake.as_ref());
486 assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Failed);
487
488 // Sweeping again must not relaunch: a container that cannot start would be
489 // relaunched forever.
490 assert_eq!(reconcile_once(&store, fake.as_ref()), 0);
491 assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Failed);
492
493 retry(&store, &alice).unwrap();
494 reconcile_once(&store, fake.as_ref());
495 reconcile_once(&store, fake.as_ref());
496 assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Ready);
497 }
498
499 #[test]
500 fn a_stopped_tenant_is_started_again() {
501 let (_dir, store, fake) = setup();
502 let alice = crate::git::validate::name("alice").unwrap();
503 store.create(&alice).unwrap();
504 reconcile_once(&store, fake.as_ref());
505 reconcile_once(&store, fake.as_ref());
506
507 fake.stop(&alice).unwrap();
508 let mut record = store.get(&alice).unwrap().unwrap();
509 record.state = State::Stopped;
510 store.put(&record).unwrap();
511
512 reconcile_once(&store, fake.as_ref());
513 assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Ready);
514 }
515
516 /// Drive an account all the way to Ready on the current version.
517 fn settle(store: &AccountStore, fake: &FakeProvisioner, name: &str) {
518 let account = crate::git::validate::name(name).unwrap();
519 store.create(&account).unwrap();
520 reconcile_once(store, fake);
521 reconcile_once(store, fake);
522 let mut record = store.get(&account).unwrap().unwrap();
523 record.last_seen_at = crate::account::token::now_secs();
524 store.put(&record).unwrap();
525 }
526
527 #[test]
528 fn a_new_tenant_records_the_version_it_was_built_with() {
529 let (_dir, store, fake) = setup();
530 settle(&store, &fake, "alice");
531
532 let record = store
533 .get(&crate::git::validate::name("alice").unwrap())
534 .unwrap()
535 .unwrap();
536 assert_eq!(record.state, State::Ready);
537 assert_eq!(
538 record.binary_version.as_deref(),
539 Some(crate::brand::build())
540 );
541 assert!(
542 fake.upgrades.lock().unwrap().is_empty(),
543 "a container built from the current binary needs no upgrade"
544 );
545 }
546
547 #[test]
548 fn a_tenant_on_an_old_binary_is_upgraded_in_place() {
549 let (_dir, store, fake) = setup();
550 settle(&store, &fake, "alice");
551 let alice = crate::git::validate::name("alice").unwrap();
552
553 // What a deploy looks like: the control plane is newer than the tenant.
554 let mut record = store.get(&alice).unwrap().unwrap();
555 record.binary_version = Some("0.0.1-old".into());
556 store.put(&record).unwrap();
557
558 reconcile_once(&store, fake.as_ref());
559
560 assert_eq!(fake.upgrades.lock().unwrap().as_slice(), ["alice"]);
561 assert_eq!(
562 store
563 .get(&alice)
564 .unwrap()
565 .unwrap()
566 .binary_version
567 .as_deref(),
568 Some(crate::brand::build())
569 );
570 // Upgrading must not have disturbed the tenant's state.
571 assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Ready);
572 }
573
574 #[test]
575 fn a_failed_upgrade_leaves_the_old_version_so_the_next_sweep_retries() {
576 use std::sync::atomic::Ordering;
577 let (_dir, store, fake) = setup();
578 settle(&store, &fake, "alice");
579 let alice = crate::git::validate::name("alice").unwrap();
580
581 let mut record = store.get(&alice).unwrap().unwrap();
582 record.binary_version = Some("0.0.1-old".into());
583 store.put(&record).unwrap();
584
585 fake.fail_next.store(true, Ordering::SeqCst);
586 reconcile_once(&store, fake.as_ref());
587 assert_eq!(
588 store
589 .get(&alice)
590 .unwrap()
591 .unwrap()
592 .binary_version
593 .as_deref(),
594 Some("0.0.1-old"),
595 "a failed upgrade must not be recorded as done"
596 );
597
598 reconcile_once(&store, fake.as_ref());
599 assert_eq!(
600 store
601 .get(&alice)
602 .unwrap()
603 .unwrap()
604 .binary_version
605 .as_deref(),
606 Some(crate::brand::build()),
607 "the next sweep must retry"
608 );
609 }
610
611 #[test]
612 fn upgrades_are_staggered_so_a_deploy_is_not_an_outage() {
613 let (_dir, store, fake) = setup();
614 let names = ["a1", "a2", "a3", "a4", "a5"];
615
616 // Settle every account first: `settle` reconciles, so backdating inside the
617 // loop would let those sweeps do the upgrading and hide the stagger.
618 for name in names {
619 settle(&store, &fake, name);
620 }
621 for name in names {
622 let account = crate::git::validate::name(name).unwrap();
623 let mut record = store.get(&account).unwrap().unwrap();
624 record.binary_version = Some("0.0.1-old".into());
625 store.put(&record).unwrap();
626 }
627 fake.upgrades.lock().unwrap().clear();
628
629 reconcile_once(&store, fake.as_ref());
630 assert_eq!(
631 fake.upgrades.lock().unwrap().len(),
632 UPGRADES_PER_SWEEP,
633 "restarting every tenant at once turns a deploy into an outage"
634 );
635
636 // Successive sweeps finish the rest.
637 for _ in 0..3 {
638 reconcile_once(&store, fake.as_ref());
639 }
640 assert_eq!(fake.upgrades.lock().unwrap().len(), 5);
641 }
642
643 #[test]
644 fn a_stopped_tenant_is_not_upgraded_until_it_is_needed() {
645 let (_dir, store, fake) = setup();
646 settle(&store, &fake, "alice");
647 let alice = crate::git::validate::name("alice").unwrap();
648
649 let mut record = store.get(&alice).unwrap().unwrap();
650 record.state = State::Stopped;
651 record.binary_version = Some("0.0.1-old".into());
652 store.put(&record).unwrap();
653
654 reconcile_once(&store, fake.as_ref());
655 assert!(
656 fake.upgrades.lock().unwrap().is_empty(),
657 "a stopped tenant picks up the new binary when it starts; \
658 restarting it now would wake it for nothing"
659 );
660 }
661
662 #[test]
663 fn an_idle_tenant_is_stopped_and_woken_by_the_next_request() {
664 let (_dir, store, fake) = setup();
665 let alice = crate::git::validate::name("alice").unwrap();
666 store.create(&alice).unwrap();
667 reconcile_once(&store, fake.as_ref());
668 reconcile_once(&store, fake.as_ref());
669 assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Ready);
670
671 // Long enough that nobody has used it.
672 let later = crate::account::token::now_secs() + IDLE_STOP.as_secs() + 1;
673 reconcile_at(&store, fake.as_ref(), later);
674 assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Stopped);
675
676 // Asking for it refreshes the timestamp and reports "come back shortly".
677 let err = store.endpoint(&alice).unwrap_err();
678 assert_eq!(err.status(), 429);
679
680 reconcile_once(&store, fake.as_ref());
681 assert_eq!(
682 store.get(&alice).unwrap().unwrap().state,
683 State::Ready,
684 "a request must bring a stopped tenant back"
685 );
686 }
687
688 #[test]
689 fn a_busy_tenant_is_not_stopped() {
690 let (_dir, store, fake) = setup();
691 let alice = crate::git::validate::name("alice").unwrap();
692 store.create(&alice).unwrap();
693 reconcile_once(&store, fake.as_ref());
694 reconcile_once(&store, fake.as_ref());
695
696 // Used just now.
697 let mut record = store.get(&alice).unwrap().unwrap();
698 record.last_seen_at = crate::account::token::now_secs();
699 store.put(&record).unwrap();
700
701 reconcile_once(&store, fake.as_ref());
702 assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Ready);
703 }
704
705 #[test]
706 fn a_ready_account_is_left_alone() {
707 let (_dir, store, fake) = setup();
708 let alice = crate::git::validate::name("alice").unwrap();
709 store.create(&alice).unwrap();
710 reconcile_once(&store, fake.as_ref());
711 reconcile_once(&store, fake.as_ref());
712
713 assert_eq!(
714 reconcile_once(&store, fake.as_ref()),
715 0,
716 "a settled account must not be touched every sweep"
717 );
718 }
719}