Merge pull request MaterializeInc#11720 from danhhz/persist_cleanup

danhhz · web-flow · commit faddb582841c · 2022-04-11T11:00:41.000-07:00
persist: small cleanups
diff --git a/src/persist-client/src/impl/machine.rs b/src/persist-client/src/impl/machine.rs
@@ -26,7 +26,7 @@ use crate::error::{InvalidUsage, NoOp};
 use crate::r#impl::state::{ReadCapability, State, WriteCapability};
 use crate::read::ReaderId;
 use crate::write::WriterId;
-use crate::Id;
+use crate::ShardId;
 
 #[derive(Debug)]
 pub struct Machine<K, V, T, D> {
@@ -54,16 +54,16 @@ where
     T: Timestamp + Lattice + Codec64,
     D: Semigroup + Codec64,
 {
-    pub fn new(id: Id, consensus: Arc<dyn Consensus + Send + Sync>) -> Self {
+    pub fn new(shard_id: ShardId, consensus: Arc<dyn Consensus + Send + Sync>) -> Self {
         Machine {
             consensus,
             seqno: None,
-            state: State::new(id),
+            state: State::new(shard_id),
         }
     }
 
-    pub fn id(&self) -> Id {
-        self.state.id()
+    pub fn shard_id(&self) -> ShardId {
+        self.state.shard_id()
     }
 
     pub async fn register(
@@ -240,7 +240,7 @@ where
         mut work_fn: WorkFn,
     ) -> Result<Result<(SeqNo, R), E>, ExternalError> {
         loop {
-            let id = self.state.id();
+            let shard_id = self.state.shard_id();
 
             let new_seqno = self.seqno.unwrap_or_default().next();
             let mut new_state = self.state.clone();
@@ -254,7 +254,7 @@ where
             let cas_res = self
                 .consensus
                 .compare_and_set(
-                    &id.to_string(),
+                    &shard_id.to_string(),
                     deadline,
                     self.seqno,
                     VersionedData {
@@ -289,8 +289,8 @@ where
     }
 
     async fn fetch_and_update_state(&mut self, deadline: Instant) -> Result<(), ExternalError> {
-        let id = self.id();
-        let current = self.consensus.head(&id.to_string(), deadline).await?;
+        let shard_id = self.shard_id();
+        let current = self.consensus.head(&shard_id.to_string(), deadline).await?;
         self.update_state(current).await
     }
 
@@ -299,7 +299,7 @@ where
             Some(x) => x,
             None => {
                 self.seqno = None;
-                self.state = State::new(self.state.id());
+                self.state = State::new(self.state.shard_id());
                 return Ok(());
             }
         };
diff --git a/src/persist-client/src/impl/state.rs b/src/persist-client/src/impl/state.rs
@@ -22,7 +22,7 @@ use timely::PartialOrder;
 use crate::error::{InvalidUsage, NoOp};
 use crate::read::ReaderId;
 use crate::write::WriterId;
-use crate::Id;
+use crate::ShardId;
 
 #[derive(Clone, Debug, PartialEq)]
 pub struct ReadCapability<T> {
@@ -38,7 +38,7 @@ pub struct WriteCapability<T> {
 // TODO: Document invariants.
 #[derive(Debug)]
 pub struct State<K, V, T, D> {
-    id: Id,
+    shard_id: ShardId,
 
     writers: HashMap<WriterId, WriteCapability<T>>,
     readers: HashMap<ReaderId, ReadCapability<T>>,
@@ -53,7 +53,7 @@ pub struct State<K, V, T, D> {
 impl<K, V, T: Clone, D> Clone for State<K, V, T, D> {
     fn clone(&self) -> Self {
         Self {
-            id: self.id.clone(),
+            shard_id: self.shard_id.clone(),
             writers: self.writers.clone(),
             readers: self.readers.clone(),
             since: self.since.clone(),
@@ -70,9 +70,9 @@ where
     T: Timestamp + Lattice + Codec64,
     D: Codec64,
 {
-    pub fn new(id: Id) -> Self {
+    pub fn new(shard_id: ShardId) -> Self {
         State {
-            id,
+            shard_id,
             writers: HashMap::new(),
             readers: HashMap::new(),
             since: Antichain::from_elem(T::minimum()),
@@ -81,8 +81,8 @@ where
         }
     }
 
-    pub fn id(&self) -> Id {
-        self.id
+    pub fn shard_id(&self) -> ShardId {
+        self.shard_id
     }
 
     pub fn register(
@@ -348,7 +348,7 @@ impl<T: Timestamp + Codec64> From<&DescriptionMeta> for Description<T> {
 
 #[derive(Debug, Serialize, Deserialize)]
 struct StateRollupMeta {
-    id: Id,
+    shard_id: ShardId,
     key_codec: String,
     val_codec: String,
     ts_codec: String,
@@ -369,7 +369,7 @@ where
 {
     fn from(x: &State<K, V, T, D>) -> Self {
         StateRollupMeta {
-            id: x.id,
+            shard_id: x.shard_id,
             key_codec: K::codec_name(),
             val_codec: V::codec_name(),
             ts_codec: T::codec_name(),
@@ -433,7 +433,7 @@ where
             ));
         }
         Ok(State {
-            id: x.id,
+            shard_id: x.shard_id,
             writers: x
                 .writers
                 .iter()
diff --git a/src/persist-client/src/lib.rs b/src/persist-client/src/lib.rs
@@ -112,20 +112,26 @@ impl Location {
 }
 
 /// An opaque identifier for a persist durable TVC (aka shard).
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Deserialize, Serialize)]
-pub struct Id([u8; 16]);
+#[derive(Clone, Copy, PartialEq, Eq, Hash, Deserialize, Serialize)]
+pub struct ShardId([u8; 16]);
 
-impl std::fmt::Display for Id {
+impl std::fmt::Display for ShardId {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        std::fmt::Display::fmt(&Uuid::from_bytes(self.0), f)
+        write!(f, "s{}", Uuid::from_bytes(self.0))
     }
 }
 
-impl Id {
-    /// Returns a random [Id] that is reasonably likely to have never been
+impl std::fmt::Debug for ShardId {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "ShardId({})", Uuid::from_bytes(self.0))
+    }
+}
+
+impl ShardId {
+    /// Returns a random [ShardId] that is reasonably likely to have never been
     /// generated before.
     pub fn new() -> Self {
-        Id(Uuid::new_v4().as_bytes().to_owned())
+        ShardId(Uuid::new_v4().as_bytes().to_owned())
     }
 }
 
@@ -160,8 +166,8 @@ impl Client {
         Ok(Client { blob, consensus })
     }
 
-    /// Provides capabilities for the durable TVC identified by `id` at its
-    /// current since and upper frontiers.
+    /// Provides capabilities for the durable TVC identified by `shard_id` at
+    /// its current since and upper frontiers.
     ///
     /// This method is a best-effort attempt to regain control of the frontiers
     /// of a shard. Its most common uses are to recover capabilities that have
@@ -170,23 +176,23 @@ impl Client {
     /// released by all other parties, this call may result in capabilities with
     /// empty frontiers (which are useless).
     ///
-    /// If `id` has never been used before, initializes a new shard and returns
-    /// handles with `since` and `upper` frontiers set to initial values of
-    /// `Antichain::from_elem(T::minimum())`.
+    /// If `shard_id` has never been used before, initializes a new shard and
+    /// returns handles with `since` and `upper` frontiers set to initial values
+    /// of `Antichain::from_elem(T::minimum())`.
     pub async fn open<K, V, T, D>(
         &self,
         timeout: Duration,
-        id: Id,
+        shard_id: ShardId,
     ) -> Result<(WriteHandle<K, V, T, D>, ReadHandle<K, V, T, D>), ExternalError>
     where
         K: Debug + Codec,
         V: Debug + Codec,
         T: Timestamp + Lattice + Codec64,
         D: Semigroup + Codec64,
     {
-        trace!("Client::open timeout={:?} id={:?}", timeout, id);
+        trace!("Client::open timeout={:?} shard_id={:?}", timeout, shard_id);
         let deadline = Instant::now() + timeout;
-        let mut machine = Machine::new(id, Arc::clone(&self.consensus));
+        let mut machine = Machine::new(shard_id, Arc::clone(&self.consensus));
         let (writer_id, reader_id) = (WriterId::new(), ReaderId::new());
         let (write_cap, read_cap) = machine.register(deadline, &writer_id, &reader_id).await?;
         let writer = WriteHandle {
@@ -257,7 +263,7 @@ mod tests {
 
         let (mut write, mut read) = new_test_client()
             .await?
-            .open::<String, String, u64, i64>(NO_TIMEOUT, Id::new())
+            .open::<String, String, u64, i64>(NO_TIMEOUT, ShardId::new())
             .await?;
         assert_eq!(write.upper(), &Antichain::from_elem(u64::minimum()));
         assert_eq!(read.since(), &Antichain::from_elem(u64::minimum()));
@@ -308,13 +314,13 @@ mod tests {
         let client = new_test_client().await?;
 
         let (mut write1, read1) = client
-            .open::<String, String, u64, i64>(NO_TIMEOUT, Id::new())
+            .open::<String, String, u64, i64>(NO_TIMEOUT, ShardId::new())
             .await?;
 
         // Different types, so that checks would fail in case we were not separating these
         // collections internally.
         let (mut write2, read2) = client
-            .open::<String, (), u64, i64>(NO_TIMEOUT, Id::new())
+            .open::<String, (), u64, i64>(NO_TIMEOUT, ShardId::new())
             .await?;
 
         let res = write1
@@ -349,7 +355,7 @@ mod tests {
         let client = new_test_client().await?;
 
         let (write, read) = client
-            .open::<String, String, u64, i64>(NO_TIMEOUT, Id::new())
+            .open::<String, String, u64, i64>(NO_TIMEOUT, ShardId::new())
             .await?;
 
         assert!(is_send_sync(client));
@@ -369,7 +375,7 @@ mod tests {
             (("3".to_owned(), "three".to_owned()), 3, 1),
         ];
 
-        let id = Id::new();
+        let id = ShardId::new();
         let client = new_test_client().await?;
         let (mut write1, read) = client
             .open::<String, String, u64, i64>(NO_TIMEOUT, id)
@@ -414,4 +420,32 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn fmt_ids() {
+        assert_eq!(
+            format!("{}", ShardId([0u8; 16])),
+            "s00000000-0000-0000-0000-000000000000"
+        );
+        assert_eq!(
+            format!("{:?}", ShardId([0u8; 16])),
+            "ShardId(00000000-0000-0000-0000-000000000000)"
+        );
+        assert_eq!(
+            format!("{}", WriterId([0u8; 16])),
+            "w00000000-0000-0000-0000-000000000000"
+        );
+        assert_eq!(
+            format!("{:?}", WriterId([0u8; 16])),
+            "WriterId(00000000-0000-0000-0000-000000000000)"
+        );
+        assert_eq!(
+            format!("{}", ReaderId([0u8; 16])),
+            "r00000000-0000-0000-0000-000000000000"
+        );
+        assert_eq!(
+            format!("{:?}", ReaderId([0u8; 16])),
+            "ReaderId(00000000-0000-0000-0000-000000000000)"
+        );
+    }
 }
diff --git a/src/persist-client/src/read.rs b/src/persist-client/src/read.rs
@@ -29,15 +29,21 @@ use uuid::Uuid;
 
 use crate::error::InvalidUsage;
 use crate::r#impl::machine::Machine;
-use crate::Id;
+use crate::ShardId;
 
 /// An opaque identifier for a reader of a persist durable TVC (aka shard).
-#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
+#[derive(Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
 pub struct ReaderId(pub(crate) [u8; 16]);
 
 impl std::fmt::Display for ReaderId {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        std::fmt::Display::fmt(&Uuid::from_bytes(self.0), f)
+        write!(f, "r{}", Uuid::from_bytes(self.0))
+    }
+}
+
+impl std::fmt::Debug for ReaderId {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "ReaderId({})", Uuid::from_bytes(self.0))
     }
 }
 
@@ -57,7 +63,7 @@ impl ReaderId {
 /// See [ReadHandle::snapshot] for details.
 #[derive(Debug, Serialize, Deserialize)]
 pub struct SnapshotSplit {
-    id: Id,
+    shard_id: ShardId,
     as_of: Vec<[u8; 8]>,
     batches: Vec<String>,
 }
@@ -352,6 +358,9 @@ where
     /// system. A `new_since` of the empty antichain "finishes" this shard,
     /// promising that no more data will ever be read by this handle.
     ///
+    /// It is possible to heartbeat a reader lease by calling this with
+    /// `new_since` equal to `self.since()` (making the call a no-op).
+    ///
     /// The clunky two-level Result is to enable more obvious error handling in
     /// the caller. See <http://sled.rs/errors.html> for details.
     pub async fn downgrade_since(
@@ -456,7 +465,7 @@ where
         };
         let mut splits = (0..num_splits.get())
             .map(|_| SnapshotSplit {
-                id: self.machine.id(),
+                shard_id: self.machine.shard_id(),
                 as_of: as_of.iter().map(|x| T::encode(x)).collect(),
                 batches: Vec::new(),
             })
@@ -479,11 +488,11 @@ where
             timeout,
             split
         );
-        if split.id != self.machine.id() {
+        if split.shard_id != self.machine.shard_id() {
             return Ok(Err(InvalidUsage(anyhow!(
                 "snapshot shard id {} doesn't match handle id {}",
-                split.id,
-                self.machine.id()
+                split.shard_id,
+                self.machine.shard_id()
             ))));
         }
         let iter = SnapshotIter {
diff --git a/src/persist-client/src/write.rs b/src/persist-client/src/write.rs
