WIP refactor with generic tlog-tiles

Author: lukevalenta

Cargo.lock

@@ -14,6 +14,7 @@ use std::{
     collections::{HashMap, HashSet},
     time::Duration,
 };
+use tlog_tiles::PendingLogEntryTrait;
 use tokio::sync::watch::{self, Sender};
 #[allow(clippy::wildcard_imports)]
 use worker::*;

crates/ct_worker/src/batcher_do.rs

@@ -30,15 +30,18 @@ use log::{debug, error, info, trace, warn};
 use p256::ecdsa::SigningKey as EcdsaSigningKey;
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
-use static_ct_api::{LogEntry, PendingLogEntry, TileIterator, TreeWithTimestamp};
+use static_ct_api::{PendingLogEntry, TileIterator, TreeWithTimestamp};
 use std::collections::HashMap;
 use std::time::Duration;
 use std::{
     cmp::{Ord, Ordering},
     sync::LazyLock,
 };
 use thiserror::Error;
-use tlog_tiles::{Hash, HashReader, PathElem, Tile, TlogError, TlogTile, HASH_SIZE};
+use tlog_tiles::{
+    Hash, HashReader, LogEntryTrait, PathElem, PendingLogEntryTrait, Tile, TlogError,
+    TlogIteratorTrait, TlogTile, HASH_SIZE,
+};
 use tokio::sync::watch::{channel, Receiver, Sender};
 
 /// The maximum tile level is 63 (<c2sp.org/static-ct-api>), so safe to use [`u8::MAX`] as
@@ -354,6 +357,7 @@ impl SequenceState {
                 edge_tiles.get(&DATA_TILE_KEY).unwrap().b.clone(),
                 data_tile.tile.width() as usize,
             )
+            .into_entry_iter()
             .enumerate()
             {
                 let got = tlog_tiles::record_hash(&entry?.merkle_tree_leaf());
@@ -596,17 +600,18 @@ async fn sequence_entries(
     }
     let mut overlay = HashMap::new();
     let mut n = old_size;
-    let mut sequenced_entries: Vec<LogEntry> = Vec::with_capacity(entries.len());
     let mut sequenced_metadata = Vec::with_capacity(entries.len());
+    let mut cache_metadata = Vec::with_capacity(entries.len());
 
     for (entry, sender) in entries {
-        let sequenced_entry = LogEntry {
-            inner: entry,
-            leaf_index: n,
-            timestamp,
-        };
+        let metadata = (n, timestamp);
+        cache_metadata.push((entry.lookup_key(), metadata));
+        sequenced_metadata.push((sender, metadata));
+
+        let sequenced_entry = entry.into_log_entry(metadata);
         let tile_leaf = sequenced_entry.tile_leaf();
         let merkle_tree_leaf = sequenced_entry.merkle_tree_leaf();
+
         metrics.seq_leaf_size.observe(tile_leaf.len().as_f64());
         data_tile.extend(tile_leaf);
 
@@ -623,7 +628,7 @@ async fn sequence_entries(
         )
         .map_err(|e| {
             SequenceError::NonFatal(format!(
-                "couldn't compute new hashes for leaf {sequenced_entry:?}: {e}",
+                "couldn't compute new hashes for leaf at index {n}: {e}",
             ))
         })?;
         for (i, h) in hashes.iter().enumerate() {
@@ -639,12 +644,6 @@ async fn sequence_entries(
             metrics.seq_data_tile_size.observe(data_tile.len().as_f64());
             data_tile.clear();
         }
-
-        sequenced_metadata.push((
-            sender,
-            (sequenced_entry.leaf_index, sequenced_entry.timestamp),
-        ));
-        sequenced_entries.push(sequenced_entry);
     }
 
     // Stage leftover partial data tile, if any.
@@ -768,23 +767,10 @@ async fn sequence_entries(
     // only consequence of cache false negatives are duplicated leaves anyway. In fact, an
     // error might cause the clients to resubmit, producing more cache false negatives and
     // duplicates.
-    if let Err(e) = cache
-        .put_entries(
-            &sequenced_entries
-                .iter()
-                .map(|entry| {
-                    (
-                        entry.inner.lookup_key(),
-                        (entry.leaf_index, entry.timestamp),
-                    )
-                })
-                .collect::<Vec<_>>(),
-        )
-        .await
-    {
+    if let Err(e) = cache.put_entries(&cache_metadata).await {
         warn!(
             "{name}: Cache put failed (entries={}): {e}",
-            sequenced_entries.len()
+            cache_metadata.len()
         );
     }
 
@@ -2097,23 +2083,24 @@ mod tests {
                 {
                     let entry = entry.unwrap();
                     let idx = n * u64::from(TlogTile::FULL_WIDTH) + i as u64;
-                    assert_eq!(entry.leaf_index, idx);
-                    assert!(entry.timestamp <= sth_timestamp);
+                    let (leaf_index, timestamp) = entry.metadata();
+                    assert_eq!(leaf_index, idx);
+                    assert!(timestamp <= sth_timestamp);
                     assert_eq!(
                         leaf_hashes[usize::try_from(idx).unwrap()],
                         tlog_tiles::record_hash(&entry.merkle_tree_leaf())
                     );
 
-                    assert!(!entry.inner.certificate.is_empty());
-                    if entry.inner.is_precert {
-                        assert!(!entry.inner.pre_certificate.is_empty());
-                        assert_ne!(entry.inner.issuer_key_hash, [0; 32]);
+                    assert!(!entry.as_pending_entry().certificate.is_empty());
+                    if entry.as_pending_entry().is_precert {
+                        assert!(!entry.as_pending_entry().pre_certificate.is_empty());
+                        assert_ne!(entry.as_pending_entry().issuer_key_hash, [0; 32]);
                     } else {
-                        assert!(entry.inner.pre_certificate.is_empty());
-                        assert_eq!(entry.inner.issuer_key_hash, [0; 32]);
+                        assert!(entry.as_pending_entry().pre_certificate.is_empty());
+                        assert_eq!(entry.as_pending_entry().issuer_key_hash, [0; 32]);
                     }
 
-                    for fp in entry.inner.chain_fingerprints {
+                    for fp in &entry.as_pending_entry().chain_fingerprints {
                         let b = block_on(self.object.fetch(&format!("issuer/{}", hex::encode(fp))))
                             .unwrap()
                             .unwrap();

crates/ct_worker/src/ctlog.rs

@@ -14,8 +14,9 @@ use p256::pkcs8::EncodePublicKey;
 use serde::Serialize;
 use serde_with::{base64::Base64, serde_as};
 use sha2::{Digest, Sha256};
-use static_ct_api::{AddChainRequest, GetRootsResponse, LogEntry, PendingLogEntry, UnixTimestamp};
+use static_ct_api::{AddChainRequest, GetRootsResponse, PendingLogEntry};
 use std::str::FromStr;
+use tlog_tiles::PendingLogEntryTrait;
 #[allow(clippy::wildcard_imports)]
 use worker::*;
 
@@ -207,18 +208,14 @@ async fn add_chain_or_pre_chain(
 
     // Check if entry is cached and return right away if so.
     let kv = load_cache_kv(env, name)?;
-    if let Some((leaf_index, timestamp)) = kv
+    if let Some(metadata) = kv
         .get(&BASE64_STANDARD.encode(lookup_key))
         .bytes_with_metadata::<SequenceMetadata>()
         .await?
         .1
     {
         debug!("{name}: Entry is cached");
-        let entry = LogEntry {
-            inner: pending_entry,
-            leaf_index,
-            timestamp,
-        };
+        let entry = pending_entry.into_log_entry(metadata);
         let sct = static_ct_api::signed_certificate_timestamp(signing_key, &entry)
             .map_err(|e| e.to_string())?;
         return Response::from_json(&sct);
@@ -261,12 +258,8 @@ async fn add_chain_or_pre_chain(
         // Return the response from the Batcher directly to the client.
         return Ok(response);
     }
-    let (leaf_index, timestamp) = response.json::<(u64, UnixTimestamp)>().await?;
-    let entry = LogEntry {
-        inner: pending_entry,
-        leaf_index,
-        timestamp,
-    };
+    let metadata = response.json::<SequenceMetadata>().await?;
+    let entry = pending_entry.into_log_entry(metadata);
     let sct = static_ct_api::signed_certificate_timestamp(signing_key, &entry)
         .map_err(|e| e.to_string())?;
     Response::from_json(&sct)

crates/ct_worker/src/frontend_worker.rs

@@ -19,10 +19,11 @@ use p256::{ecdsa::SigningKey as EcdsaSigningKey, pkcs8::DecodePrivateKey};
 use serde::Deserialize;
 use serde_bytes::ByteBuf;
 use sha2::{Digest, Sha256};
-use static_ct_api::{LookupKey, UnixTimestamp};
+use static_ct_api::LookupKey;
 use std::collections::{HashMap, VecDeque};
 use std::io::Write;
 use std::sync::{LazyLock, OnceLock};
+use tlog_tiles::SequenceMetadata;
 use util::now_millis;
 #[allow(clippy::wildcard_imports)]
 use worker::*;
@@ -114,8 +115,6 @@ struct QueryParams {
     name: String,
 }
 
-type SequenceMetadata = (u64, UnixTimestamp);
-
 trait CacheWrite {
     /// Put the provided sequenced entries into the cache. This does NOT overwrite existing entries.
     async fn put_entries(&mut self, entries: &[(LookupKey, SequenceMetadata)]) -> Result<()>;

crates/ct_worker/src/lib.rs

@@ -15,6 +15,7 @@ use log::{info, warn, Level};
 use static_ct_api::PendingLogEntry;
 use std::str::FromStr;
 use std::time::Duration;
+use tlog_tiles::PendingLogEntryTrait;
 use tokio::sync::Mutex;
 #[allow(clippy::wildcard_imports)]
 use worker::*;

crates/ct_worker/src/sequencer_do.rs

@@ -21,7 +21,6 @@
 //! - [cert_checker_test.go](https://github.com/google/certificate-transparency-go/blob/74d106d3a25205b16d571354c64147c5f1f7dbc1/trillian/ctfe/cert_checker_test.go)
 
 use crate::StaticCTError;
-use crate::{UnixTimestamp, ValidatedChain};
 use der::{
     asn1::{Null, OctetString},
     oid::{
@@ -33,6 +32,7 @@ use der::{
 use serde::{Deserialize, Serialize};
 use serde_with::{base64::Base64, serde_as};
 use sha2::{Digest, Sha256};
+use tlog_tiles::UnixTimestamp;
 use x509_util::CertPool;
 use x509_verify::{
     x509_cert::{
@@ -79,6 +79,15 @@ pub struct GetRootsResponse {
     pub certificates: Vec<Vec<u8>>,
 }
 
+/// A chain that has been parsed and validated.
+pub struct ValidatedChain {
+    pub certificate: Vec<u8>,
+    pub is_precert: bool,
+    pub issuer_key_hash: [u8; 32],
+    pub issuers: Vec<Vec<u8>>,
+    pub pre_certificate: Vec<u8>,
+}
+
 /// Validates a certificate chain and returns a [`ValidatedChain`].
 ///
 /// # Errors