From 0b78b79cdf689ef51f57120ad0d7bae69f16323e Mon Sep 17 00:00:00 2001 From: hamzakargin Date: Tue, 3 Nov 2026 13:12:13 +0400 Subject: [PATCH] feat: create tenant keys before tenant writes and check the naming key at start --- crates/control/src/error.rs | 1 + crates/control/src/lib.rs | 10 +-- crates/control/src/queue.rs | 3 + crates/control/src/record.rs | 1 + crates/control/src/writer.rs | 66 +++++++++++++++++++- crates/control/tests/queues.rs | 8 ++- crates/keys/src/ring.rs | 72 +++++++++++++++++----- crates/keys/tests/ring.rs | 28 +++++++++ crates/sched/src/error.rs | 2 + crates/sched/src/keyring.rs | 46 ++++++++++---- crates/sched/src/start.rs | 107 +++++++++++++++++++++++++++------ crates/sched/tests/index.rs | 107 +++++++++++++++++++++------------ crates/sched/tests/start.rs | 7 ++- 13 files changed, 368 insertions(+), 90 deletions(-) diff --git a/crates/control/src/error.rs b/crates/control/src/error.rs index 68ddd25..aff4d7b 100644 --- a/crates/control/src/error.rs +++ b/crates/control/src/error.rs @@ -33,6 +33,7 @@ impl fmt::Display for Error { KeyError::Unavailable => "unavailable", KeyError::Unknown => "unknown", KeyError::Failed => "failed", + KeyError::Conflict => "held by another generation", }; write!(f, "keys {what} for tenant {tenant} generation {generation}") } diff --git a/crates/control/src/lib.rs b/crates/control/src/lib.rs index df8b925..5635db7 100644 --- a/crates/control/src/lib.rs +++ b/crates/control/src/lib.rs @@ -13,13 +13,13 @@ pub use error::Error; pub use leases::{Ask, Gcra, IDLE, LIFETIME, LeaseStats, Leases, Taken, Tokens, WAIT, limit_core}; pub use names::{Call, Loaded as LoadedQueues, Naming, NamingStats, Queues, Unnamed}; pub use queue::{ - KeyError, KeySource, LABEL_SIZE, Label, MAX_DOMAIN, Name, Queue, Ready, Tenancy, domain_ok, - mix, naming_shard, queue_shard, tenant_keys, + Created, KeyError, KeySource, LABEL_SIZE, Label, MAX_DOMAIN, Name, Queue, Ready, Tenancy, + domain_ok, mix, naming_shard, queue_shard, tenant_keys, }; pub use record::{ - KIND_POOL, KIND_QUEUE, KIND_TENANT, Limits, MAX_ADDRESSES, MAX_POOLS, Object, Pool, SYSTEM, - State, Tenant, Update, + KIND_NODE, KIND_POOL, KIND_QUEUE, KIND_TENANT, Limits, MAX_ADDRESSES, MAX_POOLS, Object, Pool, + SYSTEM, State, Tenant, Update, }; pub use siphash::siphash; pub use table::{Note, Snapshot, Table, broadcast}; -pub use writer::{ENTRY, Loaded, Writer, entry_size}; +pub use writer::{ENTRY, Loaded, Writer, entry_size, naming_check}; diff --git a/crates/control/src/queue.rs b/crates/control/src/queue.rs index 89efef3..67ef807 100644 --- a/crates/control/src/queue.rs +++ b/crates/control/src/queue.rs @@ -145,13 +145,16 @@ pub enum KeyError { Unavailable, Unknown, Failed, + Conflict, } pub type Ready<'a> = Pin> + 'a>>; +pub type Created<'a> = Pin> + 'a>>; pub trait KeySource { fn get(&self, tenant: u32, generation: u32) -> Option; fn ready(&self, tenant: u32, generation: u32) -> Ready<'_>; + fn create(&self, tenant: u32, generation: u32) -> Created<'_>; } pub async fn tenant_keys( diff --git a/crates/control/src/record.rs b/crates/control/src/record.rs index 3927a5c..6a297b6 100644 --- a/crates/control/src/record.rs +++ b/crates/control/src/record.rs @@ -5,6 +5,7 @@ use crate::error::Error; pub const KIND_TENANT: u8 = 1; pub const KIND_POOL: u8 = 2; pub const KIND_QUEUE: u8 = 3; +pub const KIND_NODE: u8 = 4; pub const SYSTEM: u32 = 0; pub const MAX_POOLS: usize = 16; pub const MAX_ADDRESSES: usize = 64; diff --git a/crates/control/src/writer.rs b/crates/control/src/writer.rs index d05f19d..5f90c9f 100644 --- a/crates/control/src/writer.rs +++ b/crates/control/src/writer.rs @@ -6,11 +6,31 @@ use store::{ }; use crate::error::Error; -use crate::record::{KIND_POOL, KIND_TENANT, Object, Update}; +use crate::record::{KIND_NODE, KIND_POOL, KIND_TENANT, Object, SYSTEM, Update}; +use crate::siphash::siphash; use crate::table::Snapshot; pub const ENTRY: usize = 64; +const CHECK: usize = 16; +const CHECK_FORMAT: u8 = 1; + +pub fn naming_check(key: &[u8; 16]) -> u64 { + siphash(key, b"bunzzle naming check") +} + +fn encode_check(value: u64) -> Vec { + let mut out = vec![0; CHECK]; + out[0] = CHECK_FORMAT; + out[8..].copy_from_slice(&value.to_le_bytes()); + out +} + +fn decode_check(clear: &[u8]) -> Option { + let ok = clear.len() == CHECK && clear[0] == CHECK_FORMAT && clear[1..8] == [0; 7]; + ok.then(|| u64::from_le_bytes(clear[8..].try_into().expect("8 bytes"))) +} + pub fn entry_size(clear: usize) -> usize { ENTRY + clear.next_multiple_of(8) } @@ -22,9 +42,14 @@ pub struct Loaded { sizes: BTreeMap<(u8, u32), usize>, used: usize, skipped: u64, + naming: Option, } impl Loaded { + pub fn naming_check(&self) -> Option { + self.naming + } + pub fn writer(self, changes: Changes) -> Writer { Writer { changes, @@ -54,6 +79,7 @@ impl Writer { sizes: BTreeMap::new(), used: 0, skipped: 0, + naming: None, }; let mut objects = Vec::new(); for (key, item) in state.entries() { @@ -62,6 +88,16 @@ impl Writer { } let (boot, id) = split_key(key.message); let kind = u8::try_from(CONTROL_BOOT - boot).unwrap_or(u8::MAX); + if kind == KIND_NODE { + match decode_check(&item.clear).filter(|_| id == 0) { + Some(check) => { + loaded.naming = Some(check); + loaded.track(kind, id, item.clear.len()); + } + None => loaded.skipped += 1, + } + continue; + } if kind != KIND_TENANT && kind != KIND_POOL { continue; } @@ -95,6 +131,34 @@ impl Writer { self.loaded.skipped } + pub fn naming_check(&self) -> Option { + self.loaded.naming + } + + pub async fn set_naming_check(&mut self, check: u64) -> Result<(), Error> { + let clear = encode_check(check); + let batch = Batch { + shard: self.loaded.shard, + message: control_key(KIND_NODE, 0), + tenant: SYSTEM, + ops: vec![Op { + item: HEAD_ITEM, + change: Change::Put(Delivery { + status: 0, + attempts: 0, + due: 0, + }), + clear: clear.clone(), + bodies: Vec::new(), + sealed: Vec::new(), + }], + }; + self.changes.apply(batch).await?; + self.loaded.naming = Some(check); + self.loaded.track(KIND_NODE, 0, clear.len()); + Ok(()) + } + pub async fn write(&mut self, object: Object) -> Result { object.check()?; let update = Update { diff --git a/crates/control/tests/queues.rs b/crates/control/tests/queues.rs index 8e3c5a6..115abd7 100644 --- a/crates/control/tests/queues.rs +++ b/crates/control/tests/queues.rs @@ -7,8 +7,8 @@ use std::sync::Arc; use std::time::Duration; use control::{ - Cache, Call, Error, KIND_QUEUE, KeyError, KeySource, LABEL_SIZE, Label, LoadedQueues, Name, - Naming, Queue, Queues, Ready, Tenancy, domain_ok, naming_shard, queue_shard, siphash, + Cache, Call, Created, Error, KIND_QUEUE, KeyError, KeySource, LABEL_SIZE, Label, LoadedQueues, + Name, Naming, Queue, Queues, Ready, Tenancy, domain_ok, naming_shard, queue_shard, siphash, }; use crypto::{Secret, Tenant, TenantKeys}; use rt::sim::{Config, Disk, Machine, SimDriver, Stalls, check}; @@ -83,6 +83,10 @@ impl KeySource for Toy { None } + fn create(&self, _: u32, _: u32) -> Created<'_> { + Box::pin(std::future::ready(Ok(()))) + } + fn ready(&self, tenant: u32, generation: u32) -> Ready<'_> { Box::pin(std::future::ready(self.answer(tenant, generation))) } diff --git a/crates/keys/src/ring.rs b/crates/keys/src/ring.rs index 0b1f6f7..f211985 100644 --- a/crates/keys/src/ring.rs +++ b/crates/keys/src/ring.rs @@ -52,6 +52,7 @@ pub enum Fault { Unavailable, Unknown, Failed, + Conflict, } #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] @@ -219,9 +220,46 @@ impl Ring { } } + pub async fn create(&self, tenant: u32, generation: u32, retention: u16) -> Result<(), Fault> { + let request = Request::Create { + tenant, + generation, + retention, + }; + let working = self.answered(tenant, generation, self.call(&request).await)?; + let request = Request::Retention { + tenant, + generation, + days: retention, + }; + match self.call(&request).await { + Ok(Answer::Done) => {} + Ok(Answer::Refused(refused)) => return Err(self.fault(None, Some(refused))), + Ok(_) => return Err(self.fault(None, None)), + Err(error) => return Err(self.fault(Some(error), None)), + } + if !self.failed() { + self.inner + .tenants + .borrow_mut() + .put((tenant, generation), working); + } + Ok(()) + } + async fn fetch(&self, tenant: u32, generation: u32) -> Result { self.count(|s| s.opens += 1); - match self.call(&Request::Open { tenant, generation }).await { + let answer = self.call(&Request::Open { tenant, generation }).await; + self.answered(tenant, generation, answer) + } + + fn answered( + &self, + tenant: u32, + generation: u32, + answer: Result, + ) -> Result { + match answer { Ok(Answer::Keys { day, oldest, @@ -240,22 +278,28 @@ impl Ring { tomorrow, }) } - Ok(Answer::Refused(Error::Unknown | Error::Generation)) => { - self.count(|s| s.unknown += 1); - Err(Fault::Unknown) - } - Ok(Answer::Refused(Error::Failed | Error::Unreadable)) - | Err(Error::Failed | Error::Changed) => { - self.count(|s| s.failed += 1); - Err(Fault::Failed) - } - _ => { - self.count(|s| s.unavailable += 1); - Err(Fault::Unavailable) - } + Ok(Answer::Refused(refused)) => Err(self.fault(None, Some(refused))), + Ok(_) => Err(self.fault(None, None)), + Err(error) => Err(self.fault(Some(error), None)), } } + fn fault(&self, error: Option, refused: Option) -> Fault { + let fault = match (error, refused) { + (_, Some(Error::Unknown)) => Fault::Unknown, + (_, Some(Error::Generation)) => Fault::Conflict, + (_, Some(Error::Failed | Error::Unreadable)) + | (Some(Error::Failed | Error::Changed), _) => Fault::Failed, + _ => Fault::Unavailable, + }; + self.count(|s| match fault { + Fault::Unknown | Fault::Conflict => s.unknown += 1, + Fault::Failed => s.failed += 1, + Fault::Unavailable => s.unavailable += 1, + }); + fault + } + async fn client(&self) -> Result, Error> { if self.failed() { self.count(|s| s.refused += 1); diff --git a/crates/keys/tests/ring.rs b/crates/keys/tests/ring.rs index 8c47009..e164045 100644 --- a/crates/keys/tests/ring.rs +++ b/crates/keys/tests/ring.rs @@ -184,6 +184,34 @@ fn tenant_keys_open_once_on_first_use_and_stay_cached() { }); } +#[test] +fn create_sets_retention_adopts_a_retry_and_refuses_another_generation() { + let driver = Rc::new(SimDriver::new(8, quiet())); + run(&driver, { + let driver = Rc::clone(&driver); + async move { + let _server = serve("/a", 1).await; + let ring = Ring::new(hosts(&[1]).0, settings()); + assert_eq!(ring.create(4, 1, 30).await, Ok(())); + let first = ring.get(4, 1).expect("cached"); + assert_eq!(ring.stats().opens, 0); + assert_eq!(ring.create(4, 1, 60).await, Ok(())); + let again = ring.open(4, 1).await.expect("open"); + let label = [&4u32.to_le_bytes()[..], &1u32.to_le_bytes()].concat(); + let sealed = first.keys.seal(&label, b"mail").expect("seal"); + assert_eq!(again.keys.open(&label, &sealed).expect("open"), b"mail"); + assert_eq!(ring.create(4, 2, 30).await, Err(Fault::Conflict)); + sleep(Duration::from_secs(1)).await; + let disk = driver.disk(); + let audit = disk + .contents(&Path::new("/a").join("audit")) + .expect("audit"); + let text = String::from_utf8(audit.to_vec()).expect("text"); + assert_eq!(text.matches(" retention 4 1 ").count(), 2, "{text}"); + } + }); +} + #[test] fn keys_down_is_unavailable_and_retries_back_off() { let driver = Rc::new(SimDriver::new(2, quiet())); diff --git a/crates/sched/src/error.rs b/crates/sched/src/error.rs index d8673e9..82b9912 100644 --- a/crates/sched/src/error.rs +++ b/crates/sched/src/error.rs @@ -4,6 +4,7 @@ use std::fmt; pub enum Error { Spool(spool::Error), Control(control::Error), + Naming, Closed, } @@ -12,6 +13,7 @@ impl fmt::Display for Error { match self { Self::Spool(e) => write!(f, "{e}"), Self::Control(e) => write!(f, "{e}"), + Self::Naming => write!(f, "naming key does not match this spool"), Self::Closed => write!(f, "closed"), } } diff --git a/crates/sched/src/keyring.rs b/crates/sched/src/keyring.rs index ad6b5b3..2450fec 100644 --- a/crates/sched/src/keyring.rs +++ b/crates/sched/src/keyring.rs @@ -1,32 +1,54 @@ -use control::{KeyError, KeySource, Ready}; +use control::{Created, KeyError, KeySource, Ready}; use crypto::TenantKeys; use keys::{Fault, Ring}; -pub struct Keyring(Ring); +pub struct Keyring { + ring: Ring, + retention: u16, +} impl Keyring { - pub fn new(ring: Ring) -> Self { - Self(ring) + pub fn new(ring: Ring, retention: u16) -> Self { + Self { ring, retention } } pub fn ring(&self) -> &Ring { - &self.0 + &self.ring + } +} + +fn key_error(fault: Fault) -> KeyError { + match fault { + Fault::Unknown => KeyError::Unknown, + Fault::Unavailable => KeyError::Unavailable, + Fault::Failed => KeyError::Failed, + Fault::Conflict => KeyError::Conflict, } } impl KeySource for Keyring { fn get(&self, tenant: u32, generation: u32) -> Option { - self.0.get(tenant, generation).map(|working| working.keys) + self.ring + .get(tenant, generation) + .map(|working| working.keys) } fn ready(&self, tenant: u32, generation: u32) -> Ready<'_> { Box::pin(async move { - match self.0.open(tenant, generation).await { - Ok(working) => Ok(working.keys), - Err(Fault::Unknown) => Err(KeyError::Unknown), - Err(Fault::Unavailable) => Err(KeyError::Unavailable), - Err(Fault::Failed) => Err(KeyError::Failed), - } + self.ring + .open(tenant, generation) + .await + .map(|working| working.keys) + .map_err(key_error) + }) + } + + fn create(&self, tenant: u32, generation: u32) -> Created<'_> { + Box::pin(async move { + self.ring + .create(tenant, generation, self.retention) + .await + .map_err(key_error) }) } } diff --git a/crates/sched/src/start.rs b/crates/sched/src/start.rs index c47dc4e..1d6ccff 100644 --- a/crates/sched/src/start.rs +++ b/crates/sched/src/start.rs @@ -2,8 +2,8 @@ use std::cell::{Cell, Ref, RefCell}; use std::rc::Rc; use control::{ - Ask, Call, KeySource, Leases, Name, Naming, Object, Queues, Table, Update, Writer, broadcast, - limit_core, + Ask, Call, KeyError, KeySource, Leases, Name, Naming, Object, Queues, Table, Update, Writer, + broadcast, limit_core, }; use rt::{Ends, JoinHandle, Mesh, Notify, Peers, spawn}; use store::Tiering; @@ -77,6 +77,7 @@ pub struct Parts { pub(crate) struct Inner { pub(crate) settings: Settings, + keys: Rc, cores: u16, pub(crate) spool: spool::Spool, pub(crate) sched: Sched, @@ -132,6 +133,7 @@ impl Node { } else { None }; + let check = naming(control.as_ref().map(|(loaded, _)| loaded), &settings)?; let loaded = Queues::load(&core, settings.naming, parts.keys.as_ref()).await?; let table = Rc::new(RefCell::new(Table::default())); let clock = parts.clock; @@ -151,10 +153,13 @@ impl Node { applied.notify(); }) }; - let writer = control.map(|(loaded, snapshot)| { - broadcast(&config, cores, &control::Note::Snapshot(snapshot)); - loaded.writer(changes.clone()) - }); + let writer = match control { + Some((loaded, snapshot)) => { + broadcast(&config, cores, &control::Note::Snapshot(snapshot)); + Some(control_writer(loaded, changes.clone(), check).await?) + } + None => None, + }; let inputs = Inputs { keys: Rc::clone(&parts.keys), @@ -166,18 +171,11 @@ impl Node { }; let (suppressions, requests, mut index_done) = index_services(links.suppress, inputs).await?; - let (queues, names_done) = loaded.start(links.names, changes.clone(), parts.keys); - - let shards = settings.naming.shards; - let rules = Rc::clone(&table); - let (leases, leases_done) = Leases::start( - links.leases, - move |tenant| limit_core(*tenant, shards, cores), - move |tenant| rules.borrow().tenant(*tenant).map(|t| t.limits), - ); - leases.restore(spool.open_counts()); - let freed = leases.clone(); - spool.on_close(move |tenant, count| freed.free(&tenant, count)); + let (queues, names_done) = + loaded.start(links.names, changes.clone(), Rc::clone(&parts.keys)); + + let (leases, leases_done) = + start_leases(links.leases, &table, &spool, settings.naming.shards); while !table.borrow().loaded() { applied.wait().await; @@ -201,6 +199,7 @@ impl Node { let inner = Inner { settings, + keys: Rc::clone(&parts.keys), cores, spool, sched, @@ -282,6 +281,11 @@ impl Node { inner.writing.set(false); return Err(control::Error::Invalid("core").into()); }; + if let Err(error) = self.create_keys(&object).await { + *inner.writer.borrow_mut() = Some(writer); + inner.writing.set(false); + return Err(error.into()); + } let written = writer.write(object).await; *inner.writer.borrow_mut() = Some(writer); inner.writing.set(false); @@ -294,6 +298,31 @@ impl Node { Ok(update) } + async fn create_keys(&self, object: &Object) -> Result<(), control::Error> { + let Object::Tenant(tenant) = object else { + return Ok(()); + }; + let (id, generation) = (tenant.id, tenant.generation); + let known = self.inner.table.borrow().tenant(id).map(|t| t.generation); + match known { + Some(current) if current == generation => return Ok(()), + None if generation != 1 => return Err(control::Error::Refused), + _ => {} + } + self.inner + .keys + .create(id, generation) + .await + .map_err(|error| match error { + KeyError::Conflict => control::Error::Refused, + error => control::Error::Keys { + tenant: id, + generation, + error, + }, + }) + } + pub async fn rebind(&self, name: &Name, pool: u32) -> Result { Ok(self.inner.queues.rebind(name, pool).await?) } @@ -328,6 +357,48 @@ async fn index_services( Ok((suppressions, requests, vec![lists_done, requests_done])) } +fn start_leases( + ends: Ends>, + table: &Rc>, + spool: &spool::Spool, + shards: u32, +) -> (Leases, JoinHandle<()>) { + let cores = ends.cores(); + let rules = Rc::clone(table); + let (leases, done) = Leases::start( + ends, + move |tenant| limit_core(*tenant, shards, cores), + move |tenant| rules.borrow().tenant(*tenant).map(|t| t.limits), + ); + leases.restore(spool.open_counts()); + let freed = leases.clone(); + spool.on_close(move |tenant, count| freed.free(&tenant, count)); + (leases, done) +} + +async fn control_writer( + loaded: control::Loaded, + changes: Changes, + check: u64, +) -> Result { + let mut writer = loaded.writer(changes); + if writer.naming_check().is_none() { + writer.set_naming_check(check).await?; + } + Ok(writer) +} + +fn naming(control: Option<&control::Loaded>, settings: &Settings) -> Result { + let check = control::naming_check(&settings.naming.key); + if control + .and_then(control::Loaded::naming_check) + .is_some_and(|found| found != check) + { + return Err(Error::Naming); + } + Ok(check) +} + fn retain_current(core: &mut Core, table: &Rc>, clock: Clock) { let rules = Rc::clone(table); core.keeper.set_tiering(Tiering { diff --git a/crates/sched/tests/index.rs b/crates/sched/tests/index.rs index 0c69716..b838cd1 100644 --- a/crates/sched/tests/index.rs +++ b/crates/sched/tests/index.rs @@ -8,10 +8,9 @@ use std::rc::Rc; use std::task::Poll; use std::time::Duration; -use control::{KeyError, KeySource, Naming, Object, Pool, Ready, State, Table, Tenant}; +use control::{Created, KeyError, KeySource, Naming, Object, Pool, Ready, State, Table, Tenant}; use crypto::{Secret, Tenant as Root, TenantKeys}; use index::{Entry, Op}; -use keys::proto::{Answer as KeyAnswer, Request}; use keys::{ Audit, Client as KeyClient, Connect, Ring, RingSettings, Server, Settings as KeySettings, Vault, }; @@ -38,6 +37,7 @@ const OPTIONS: Options = Options { struct Toy; const KEYLESS: u32 = 199; +const NO_CREATE: u32 = 198; impl KeySource for Toy { fn get(&self, tenant: u32, generation: u32) -> Option { @@ -50,6 +50,15 @@ impl KeySource for Toy { Some(TenantKeys::derive(tenant, generation, &root, 1)) } + fn create(&self, tenant: u32, _: u32) -> Created<'_> { + let created = if tenant == NO_CREATE { + Err(KeyError::Unavailable) + } else { + Ok(()) + }; + Box::pin(std::future::ready(created)) + } + fn ready(&self, tenant: u32, generation: u32) -> Ready<'_> { let missing = if tenant == KEYLESS { KeyError::Unavailable @@ -124,7 +133,9 @@ const SMALL: Options = Options { shards: SHARDS as u32, }; -fn settings() -> Settings { +const NAMING: [u8; 16] = *b"naming key 16 by"; + +fn settings_named(key: [u8; 16]) -> Settings { Settings { node: 3, spool: spool::Limits { @@ -133,7 +144,7 @@ fn settings() -> Settings { body: 64 * 1024, }, naming: Naming { - key: *b"naming key 16 by", + key, shards: u32::from(SHARDS), cache: 16, }, @@ -167,11 +178,11 @@ fn started_tuned( behind: u64, tuning: Option<(u64, usize)>, ) -> Cluster { - try_started(driver, cores, behind, tuning, false).expect("start") + try_started(driver, cores, behind, tuning, false, NAMING).expect("start") } fn started_with_keys(driver: &Rc, cores: u16) -> Cluster { - try_started(driver, cores, 0, None, true).expect("start") + try_started(driver, cores, 0, None, true, NAMING).expect("start") } const KEYS_DIR: &str = "/keys"; @@ -196,32 +207,17 @@ fn keys_up(machine: &Machine) { fn keyring() -> Rc { let connect: Connect = Rc::new(|| Box::pin(async { Ok(KeyClient::new(TcpStream::connect(KEYS_AT).await?)) })); - Rc::new(Keyring::new(Ring::new(connect, RingSettings::new(1)))) + Rc::new(Keyring::new(Ring::new(connect, RingSettings::new(1)), 30)) } -async fn create_keys(tenant: u32, generation: u32) { - let mut tries = 0; - let stream = loop { - match TcpStream::connect(KEYS_AT).await { - Ok(stream) => break stream, - Err(_) if tries < 100 => { - tries += 1; - sleep(Duration::from_millis(10)).await; - } - Err(error) => panic!("connect: {error}"), +async fn keys_listening() { + for _ in 0..100 { + if TcpStream::connect(KEYS_AT).await.is_ok() { + return; } - }; - let client = KeyClient::new(stream); - let create = Request::Create { - tenant, - generation, - retention: 30, - }; - assert!(matches!( - client.call(&create).await.expect("create"), - KeyAnswer::Keys { .. } - )); - client.close(); + sleep(Duration::from_millis(10)).await; + } + panic!("keys not listening"); } fn try_started( @@ -230,6 +226,7 @@ fn try_started( behind: u64, tuning: Option<(u64, usize)>, real_keys: bool, + naming: [u8; 16], ) -> Option { let machine = Machine::new(Rc::clone(driver), usize::from(cores) + 1, Stalls::default()); let setup = Setup { @@ -273,7 +270,7 @@ fn try_started( transport: Idle, clock: Clock::at(rt::wall_ms().saturating_sub(behind), rt::now()), }; - let started = Node::start(core, links, settings(), parts); + let started = Node::start(core, links, settings_named(naming), parts); timeout(Duration::from_secs(60), started).await.ok()?.ok() })); } @@ -605,6 +602,20 @@ fn missing_keys_refuse_for_now_and_never_answer_not_suppressed() { Err(Refusal::Keys) ); assert!(Refusal::Keys.temporary()); + let control = nodes.iter().find(|n| n.controls()).expect("control core"); + assert!(matches!( + control.write(tenant(NO_CREATE, 1, State::Active)).await, + Err(sched::Error::Control(control::Error::Keys { + tenant: NO_CREATE, + generation: 1, + error: KeyError::Unavailable, + })) + )); + assert!(control.table().tenant(NO_CREATE).is_none()); + assert!(matches!( + control.write(tenant(7, 3, State::Active)).await, + Err(sched::Error::Control(control::Error::Refused)) + )); }); cluster.stop(); } @@ -615,8 +626,7 @@ fn suppressions_and_queues_use_keys_from_the_keys_process() { let cluster = started_with_keys(&driver, 2); let nodes = cluster.nodes.clone(); let queue = cluster.run(async move { - create_keys(1, 1).await; - create_keys(2, 1).await; + keys_listening().await; ready(&nodes).await; let lists = nodes[0].suppressions(); lists @@ -635,12 +645,20 @@ fn suppressions_and_queues_use_keys_from_the_keys_process() { ); let page = lists.list(1, None, 10).await.expect("list"); assert_eq!(page.listed[0].address, "kept@example.org"); + let control = nodes.iter().find(|n| n.controls()).expect("control core"); + assert!(matches!( + control.write(tenant(1, 2, State::Active)).await, + Err(sched::Error::Control(control::Error::Refused)) + )); + assert!(matches!( + control.write(tenant(5, 2, State::Active)).await, + Err(sched::Error::Control(control::Error::Refused)) + )); + assert_eq!(control.table().tenant(1).map(|t| t.generation), Some(1)); + assert!(control.table().tenant(5).is_none()); configure(&nodes, vec![tenant(3, 1, State::Active)]).await; wait_for(&nodes, 3, 1, State::Active).await; - assert_eq!( - lists.check(3, 1, &["a@example.org"]).await, - Err(Refusal::Keys) - ); + assert_eq!(lists.check(3, 1, &["a@example.org"]).await, Ok(vec![false])); let mut draft = nodes[0].draft(1, None).expect("draft"); nodes[0] .recipient(&mut draft, "to@example.org") @@ -671,6 +689,21 @@ fn suppressions_and_queues_use_keys_from_the_keys_process() { restarted.stop(); } +#[test] +fn a_different_naming_key_refuses_to_start() { + let driver = Rc::new(SimDriver::with_disk(33, Config::default(), created(33))); + let cluster = started(&driver, 2); + let nodes = cluster.nodes.clone(); + cluster.run(async move { ready(&nodes).await }); + cluster.stop(); + let disk = driver.crash(); + let other = Rc::new(SimDriver::with_disk(34, Config::default(), disk.clone())); + assert!(try_started(&other, 2, 0, None, false, *b"another key 16 b").is_none()); + let same = Rc::new(SimDriver::with_disk(35, Config::default(), disk)); + let restarted = started(&same, 2); + restarted.stop(); +} + #[test] fn listing_pages_through_every_core_in_hash_order() { listing(3, 4); @@ -1344,7 +1377,7 @@ fn index_session( tuning: (u64, usize), second: bool, ) -> bool { - let Some(cluster) = try_started(driver, cores, 0, Some(tuning), false) else { + let Some(cluster) = try_started(driver, cores, 0, Some(tuning), false, NAMING) else { return false; }; let nodes = cluster.nodes.clone(); diff --git a/crates/sched/tests/start.rs b/crates/sched/tests/start.rs index b9fa3bd..179457f 100644 --- a/crates/sched/tests/start.rs +++ b/crates/sched/tests/start.rs @@ -8,7 +8,8 @@ use std::task::Poll; use std::time::Duration; use control::{ - KeyError, KeySource, Limits as TenantLimits, Name, Naming, Object, Pool, Ready, State, Tenant, + Created, KeyError, KeySource, Limits as TenantLimits, Name, Naming, Object, Pool, Ready, State, + Tenant, }; use crypto::{Secret, Tenant as Root, TenantKeys}; use rt::sim::{Config, Disk, Faults, Machine, SimDriver, Stalls, check}; @@ -133,6 +134,10 @@ impl KeySource for Toy { Some(TenantKeys::derive(tenant, generation, &root, 1)) } + fn create(&self, _: u32, _: u32) -> Created<'_> { + Box::pin(std::future::ready(Ok(()))) + } + fn ready(&self, tenant: u32, generation: u32) -> Ready<'_> { Box::pin(std::future::ready( self.get(tenant, generation).ok_or(KeyError::Unknown),