feat: create tenant keys before tenant writes and check the naming key at start

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}")
             }

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};

crates/control/src/queue.rs

@@ -145,13 +145,16 @@ pub enum KeyError {
     Unavailable,
     Unknown,
     Failed,
+    Conflict,
 }
 
 pub type Ready<'a> = Pin<Box<dyn Future<Output = Result<TenantKeys, KeyError>> + 'a>>;
+pub type Created<'a> = Pin<Box<dyn Future<Output = Result<(), KeyError>> + 'a>>;
 
 pub trait KeySource {
     fn get(&self, tenant: u32, generation: u32) -> Option<TenantKeys>;
     fn ready(&self, tenant: u32, generation: u32) -> Ready<'_>;
+    fn create(&self, tenant: u32, generation: u32) -> Created<'_>;
 }
 
 pub async fn tenant_keys(

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;

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<u8> {
+    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<u64> {
+    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<u64>,
 }
 
 impl Loaded {
+    pub fn naming_check(&self) -> Option<u64> {
+        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<u64> {
+        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<Update, Error> {
         object.check()?;
         let update = Update {

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)))
     }

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<Working, Fault> {
         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<Answer, Error>,
+    ) -> Result<Working, Fault> {
+        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<Error>, refused: Option<Error>) -> 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<Rc<Client>, Error> {
         if self.failed() {
             self.count(|s| s.refused += 1);

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()));

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"),
         }
     }

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<TenantKeys> {
-        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)
         })
     }
 }