Make LogEntry::Metadata a generic associated type

crates/bootstrap_mtc_api/src/lib.rs

@@ -174,6 +174,16 @@ impl LogEntry for BootstrapMtcLogEntry {
     const REQUIRE_CHECKPOINT_TIMESTAMP: bool = false;
     type Pending = BootstrapMtcPendingLogEntry;
     type ParseError = MtcError;
+    type Metadata = SequenceMetadata;
+
+    fn make_metadata(
+        leaf_index: LeafIndex,
+        timestamp: UnixTimestamp,
+        _old_tree_size: u64,
+        _new_tree_size: u64,
+    ) -> Self::Metadata {
+        (leaf_index, timestamp)
+    }
 
     fn initial_entry() -> Option<Self::Pending> {
         Some(Self::Pending {
@@ -184,7 +194,7 @@ impl LogEntry for BootstrapMtcLogEntry {
         })
     }
 
-    fn new(pending: Self::Pending, metadata: SequenceMetadata) -> Self {
+    fn new(pending: Self::Pending, metadata: Self::Metadata) -> Self {
         Self(TlogTilesLogEntry::new(pending.entry, metadata))
     }
 

crates/bootstrap_mtc_worker/src/batcher_do.rs

@@ -4,7 +4,7 @@ use generic_log_worker::{get_durable_object_name, BatcherConfig, GenericBatcher,
 use worker::*;
 
 #[durable_object(fetch)]
-struct Batcher(GenericBatcher);
+struct Batcher(GenericBatcher<tlog_tiles::SequenceMetadata>);
 
 impl DurableObject for Batcher {
     fn new(state: State, env: Env) -> Self {
@@ -25,7 +25,7 @@ impl DurableObject for Batcher {
             enable_dedup: false, // deduplication is not currently supported
             location_hint: params.location_hint.clone(),
         };
-        Batcher(GenericBatcher::new(state, env, config))
+        Batcher(GenericBatcher::<tlog_tiles::SequenceMetadata>::new(state, env, config))
     }
 
     async fn fetch(&self, req: Request) -> Result<Response> {

crates/bootstrap_mtc_worker/src/sequencer_do.rs

@@ -86,7 +86,7 @@ fn checkpoint_callback(env: &Env, name: &str) -> CheckpointCallbacker {
     let params = &CONFIG.logs[name];
     let bucket = load_public_bucket(env, name).unwrap();
     Box::new(
-        move |old_time: UnixTimestamp, new_time: UnixTimestamp, new_checkpoint_bytes: &[u8]| {
+        move |old_time: UnixTimestamp, new_time: UnixTimestamp, _old_tree_size: u64, _new_tree_size: u64, new_checkpoint_bytes: &[u8]| {
             let new_checkpoint = {
                 // TODO: Make more efficient. There are two unnecessary allocations here.
 

crates/ct_worker/src/batcher_do.rs

@@ -4,7 +4,7 @@ use generic_log_worker::{get_durable_object_name, BatcherConfig, GenericBatcher,
 use worker::*;
 
 #[durable_object(fetch)]
-struct Batcher(GenericBatcher);
+struct Batcher(GenericBatcher<tlog_tiles::SequenceMetadata>);
 
 impl DurableObject for Batcher {
     fn new(state: State, env: Env) -> Self {
@@ -25,7 +25,7 @@ impl DurableObject for Batcher {
             enable_dedup: params.enable_dedup,
             location_hint: params.location_hint.clone(),
         };
-        Batcher(GenericBatcher::new(state, env, config))
+        Batcher(GenericBatcher::<tlog_tiles::SequenceMetadata>::new(state, env, config))
     }
 
     async fn fetch(&self, req: Request) -> Result<Response> {

crates/generic_log_worker/src/batcher_do.rs

@@ -7,9 +7,10 @@
 //! Entries are assigned to Batcher shards with consistent hashing on the cache key.
 
 use crate::{
-    deserialize, get_durable_object_stub, load_cache_kv, obs, serialize, LookupKey,
-    SequenceMetadata, BATCH_ENDPOINT, ENTRY_ENDPOINT, SEQUENCER_BINDING,
+    deserialize, get_durable_object_stub, load_cache_kv, obs, serialize, CacheSerialize,
+    LookupKey, BATCH_ENDPOINT, ENTRY_ENDPOINT, SEQUENCER_BINDING,
 };
+use serde::{de::DeserializeOwned, Serialize};
 use base64::prelude::*;
 use futures_util::future::{join_all, select, Either};
 use std::{
@@ -23,12 +24,17 @@ use worker::kv::KvStore;
 #[allow(clippy::wildcard_imports)]
 use worker::*;
 
-pub struct GenericBatcher {
+/// A Durable Object that buffers incoming log entries and submits them to the
+/// Sequencer in batches.
+///
+/// `M` is the sequence metadata type produced for each entry after sequencing
+/// (i.e., [`LogEntry::Metadata`](tlog_tiles::LogEntry::Metadata)).
+pub struct GenericBatcher<M> {
     env: Env,
     config: BatcherConfig,
     state: State,
     kv: Option<KvStore>,
-    batch: RefCell<Batch>,
+    batch: RefCell<Batch<M>>,
     in_flight: RefCell<usize>,
     processed: RefCell<usize>,
     wshim: Option<obs::Wshim>,
@@ -42,14 +48,15 @@ pub struct BatcherConfig {
     pub location_hint: Option<String>,
 }
 
-// A batch of entries to be submitted to the Sequencer together.
-struct Batch {
+// A batch of entries to be submitted to the Sequencer together. `M` is the
+// sequence metadata type returned to waiters once the batch is sequenced.
+struct Batch<M> {
     entries: Vec<PendingLogEntryBlob>,
     by_hash: HashSet<LookupKey>,
-    done: Sender<HashMap<LookupKey, SequenceMetadata>>,
+    done: Sender<HashMap<LookupKey, M>>,
 }
 
-impl Default for Batch {
+impl<M: Clone + Default> Default for Batch<M> {
     /// Returns a batch initialized with a watch channel.
     fn default() -> Self {
         Self {
@@ -60,7 +67,7 @@ impl Default for Batch {
     }
 }
 
-impl GenericBatcher {
+impl<M: CacheSerialize + Serialize + DeserializeOwned + Clone + Default + Copy + 'static> GenericBatcher<M> {
     /// Returns a new batcher with the given config.
     ///
     /// # Panics
@@ -189,7 +196,7 @@ impl GenericBatcher {
     }
 }
 
-impl GenericBatcher {
+impl<M: CacheSerialize + Serialize + DeserializeOwned + Clone + Default + Copy + 'static> GenericBatcher<M> {
     /// Submit the current pending batch to be sequenced.
     ///
     /// # Errors
@@ -217,8 +224,8 @@ impl GenericBatcher {
                 ..Default::default()
             },
         )?;
-        let sequenced_entries: HashMap<LookupKey, SequenceMetadata> =
-            deserialize::<Vec<(LookupKey, SequenceMetadata)>>(
+        let sequenced_entries: HashMap<LookupKey, M> =
+            deserialize::<Vec<(LookupKey, M)>>(
                 &get_durable_object_stub(
                     &self.env,
                     &self.config.name,
@@ -244,7 +251,7 @@ impl GenericBatcher {
                 .map(|(k, v)| {
                     Ok(kv
                         .put(&BASE64_STANDARD.encode(k), "")?
-                        .metadata::<SequenceMetadata>(v)?
+                        .metadata::<M>(v)?
                         .execute())
                 })
                 .collect::<Result<Vec<_>>>()?;