Add partial tile cleanup utility, fixes #1

The tlog-tiles and static-ct-api specs allow partial tiles to be deleted
when the corresponding full tile is available. This helps to reduce R2
storage costs, but will incur extra cost for the R2 list and delete
operations.

- Add a `Cleaner` Durable Object, which iterates over a log and for each
  full tile available in the public bucket, lists and deletes any
  corresponding partial tiles. The cleaner tracks subrequests to prevent
  an alarm from exceeding the 1000 subrequests limit per invocation.
- (bonus) Lays the groundwork for implementing a tlog-witness or
  tlog-mirror as a service that periodically updates based on a target
  log's latest checkpoint.

Other changes:
- Update to worker 0.6.1 for R2 `delete_multiple` support and
  `PartialEq` support for  ObjectIds.
- Move more Durable Object intialization logic from the MTC and CT
  applications to the `generic_log_worker` crate to deduplicate code.
- Fix tree time metric to seconds.

Author: lukevalenta

Cargo.lock

@@ -71,7 +71,7 @@ thiserror = "2.0"
 tlog_tiles = { path = "crates/tlog_tiles", version = "0.2.0" }
 tokio = { version = "1", features = ["sync"] }
 url = "2.2"
-worker = "0.6.0"
+worker = "0.6.1"
 x509-cert = "0.2.5"
 x509-verify = { version = "0.4.4", features = [
     "md2",

Cargo.toml

@@ -57,7 +57,6 @@ tlog_tiles.workspace = true
 worker.workspace = true
 x509-cert.workspace = true
 x509_util.workspace = true
-prometheus.workspace = true
 chrono.workspace = true
 base64ct.workspace = true
 csv.workspace = true

crates/ct_worker/Cargo.toml

@@ -113,6 +113,12 @@
                             "type": "boolean",
                             "default": true,
                             "description": "Enables loading root store trusted roots from the CCADB list, in addition to any roots configured in `roots.<env>.pem`. If enabled, requires a KV namespace with the binding `ccadb_roots` to be configured in `wrangler.jsonc`, as well as a cron trigger so that the CCADB list auto-updates."
+                        },
+                        "clean_interval_secs": {
+                            "type": "integer",
+                            "minimum": 1,
+                            "default": 60,
+                            "description": "How long to wait in between runs of the partial tile cleaner. For static CT, the cleaner can clean 498 tiles (127,488 entries) per run before hitting the Workers limit of 1000 subrequests, so the default of once every 60 seconds should keep up with a log that grows at 2000 entries/second."
                         }
                     },
                     "required": [

crates/ct_worker/config.schema.json

@@ -42,6 +42,8 @@ pub struct LogParams {
     pub enable_dedup: bool,
     #[serde(default = "default_bool::<true>")]
     pub enable_ccadb_roots: bool,
+    #[serde(default = "default_u64::<60>")]
+    pub clean_interval_secs: u64,
 }
 
 fn default_bool<const V: bool>() -> bool {

crates/ct_worker/config/src/lib.rs

@@ -1,5 +1,5 @@
 use crate::CONFIG;
-use generic_log_worker::{get_durable_object_stub, load_cache_kv, BatcherConfig, GenericBatcher};
+use generic_log_worker::{get_durable_object_name, BatcherConfig, GenericBatcher, BATCHER_BINDING};
 #[allow(clippy::wildcard_imports)]
 use worker::*;
 
@@ -8,46 +8,24 @@ struct Batcher(GenericBatcher);
 
 impl DurableObject for Batcher {
     fn new(state: State, env: Env) -> Self {
-        // Find the Durable Object name by enumerating all possibilities.
-        // TODO after update to worker > 0.6.0 use ObjectId::equals for comparison.
-        let id = state.id().to_string();
-        let namespace = env.durable_object("BATCHER").unwrap();
-        let (name, params) = CONFIG
-            .logs
-            .iter()
-            .find(|(name, params)| {
-                for shard_id in 0..params.num_batchers {
-                    if id
-                        == namespace
-                            .id_from_name(&format!("{name}_{shard_id:x}"))
-                            .unwrap()
-                            .to_string()
-                    {
-                        return true;
-                    }
-                }
-                false
-            })
-            .expect("unable to find batcher name");
-        let kv = if params.enable_dedup {
-            Some(load_cache_kv(&env, name).unwrap())
-        } else {
-            None
-        };
-        let sequencer = get_durable_object_stub(
+        let name = get_durable_object_name(
             &env,
-            name,
-            None,
-            "SEQUENCER",
-            params.location_hint.as_deref(),
-        )
-        .unwrap();
+            &state,
+            BATCHER_BINDING,
+            &mut CONFIG
+                .logs
+                .iter()
+                .map(|(name, params)| (name.as_str(), params.num_batchers)),
+        );
+        let params = &CONFIG.logs[name];
         let config = BatcherConfig {
             name: name.to_string(),
             max_batch_entries: params.max_batch_entries,
             batch_timeout_millis: params.batch_timeout_millis,
+            enable_dedup: params.enable_dedup,
+            location_hint: params.location_hint.clone(),
         };
-        Batcher(GenericBatcher::new(config, kv, sequencer))
+        Batcher(GenericBatcher::new(&env, config))
     }
 
     async fn fetch(&self, req: Request) -> Result<Response> {

crates/ct_worker/src/batcher_do.rs

@@ -0,0 +1,48 @@
+use std::time::Duration;
+
+use crate::{load_checkpoint_signers, load_origin, CONFIG};
+use generic_log_worker::{get_durable_object_name, CleanerConfig, GenericCleaner, CLEANER_BINDING};
+use signed_note::VerifierList;
+use static_ct_api::StaticCTPendingLogEntry;
+use tlog_tiles::PendingLogEntry;
+#[allow(clippy::wildcard_imports)]
+use worker::*;
+
+#[durable_object(alarm)]
+struct Cleaner(GenericCleaner);
+
+impl DurableObject for Cleaner {
+    fn new(state: State, env: Env) -> Self {
+        let name = get_durable_object_name(
+            &env,
+            &state,
+            CLEANER_BINDING,
+            &mut CONFIG.logs.keys().map(|name| (name.as_str(), 0)),
+        );
+        let params = &CONFIG.logs[name];
+
+        let config = CleanerConfig {
+            name: name.to_string(),
+            origin: load_origin(name),
+            data_path: StaticCTPendingLogEntry::DATA_TILE_PATH,
+            aux_path: StaticCTPendingLogEntry::AUX_TILE_PATH,
+            verifiers: VerifierList::new(
+                load_checkpoint_signers(&env, name)
+                    .iter()
+                    .map(|s| s.verifier())
+                    .collect(),
+            ),
+            clean_interval: Duration::from_secs(params.clean_interval_secs),
+        };
+
+        Cleaner(GenericCleaner::new(&state, &env, config))
+    }
+
+    async fn fetch(&self, req: Request) -> Result<Response> {
+        self.0.fetch(req).await
+    }
+
+    async fn alarm(&self) -> Result<Response> {
+        self.0.alarm().await
+    }
+}

crates/ct_worker/src/cleaner_do.rs

@@ -18,6 +18,7 @@ use crate::ccadb_roots_cron::update_ccadb_roots;
 
 mod batcher_do;
 mod ccadb_roots_cron;
+mod cleaner_do;
 mod frontend_worker;
 mod sequencer_do;
 

crates/ct_worker/src/lib.rs

@@ -6,8 +6,9 @@
 use std::time::Duration;
 
 use crate::{load_checkpoint_signers, load_origin, CONFIG};
-use generic_log_worker::{load_public_bucket, GenericSequencer, SequencerConfig};
-use prometheus::Registry;
+use generic_log_worker::{
+    get_durable_object_name, GenericSequencer, SequencerConfig, SEQUENCER_BINDING,
+};
 use static_ct_api::StaticCTLogEntry;
 #[allow(clippy::wildcard_imports)]
 use worker::*;
@@ -17,38 +18,27 @@ struct Sequencer(GenericSequencer<StaticCTLogEntry>);
 
 impl DurableObject for Sequencer {
     fn new(state: State, env: Env) -> Self {
-        // Find the Durable Object name by enumerating all possibilities.
-        // TODO after update to worker > 0.6.0 use ObjectId::equals for comparison.
-        let id = state.id().to_string();
-        let namespace = env.durable_object("SEQUENCER").unwrap();
-        let (name, params) = CONFIG
-            .logs
-            .iter()
-            .find(|(name, _)| id == namespace.id_from_name(name).unwrap().to_string())
-            .expect("unable to find sequencer name");
-
-        let origin = load_origin(name);
-        let sequence_interval = Duration::from_millis(params.sequence_interval_millis);
-
-        // We don't use checkpoint extensions for CT
-        let checkpoint_extension = Box::new(|_| vec![]);
-
-        let checkpoint_signers = load_checkpoint_signers(&env, name);
-        let bucket = load_public_bucket(&env, name).unwrap();
-        let registry = Registry::new();
+        let name = get_durable_object_name(
+            &env,
+            &state,
+            SEQUENCER_BINDING,
+            &mut CONFIG.logs.keys().map(|name| (name.as_str(), 0)),
+        );
+        let params = &CONFIG.logs[name];
 
         let config = SequencerConfig {
             name: name.to_string(),
-            origin,
-            checkpoint_signers,
-            checkpoint_extension,
-            sequence_interval,
+            origin: load_origin(name),
+            checkpoint_signers: load_checkpoint_signers(&env, name),
+            checkpoint_extension: Box::new(|_| vec![]), // no checkpoint extensions for CT
+            sequence_interval: Duration::from_millis(params.sequence_interval_millis),
             max_sequence_skips: params.max_sequence_skips,
             enable_dedup: params.enable_dedup,
             sequence_skip_threshold_millis: params.sequence_skip_threshold_millis,
+            location_hint: params.location_hint.clone(),
         };
 
-        Sequencer(GenericSequencer::new(config, state, bucket, registry))
+        Sequencer(GenericSequencer::new(state, env, config))
     }
 
     async fn fetch(&self, req: Request) -> Result<Response> {

crates/ct_worker/src/sequencer_do.rs

@@ -55,6 +55,10 @@
                     {
                         "name": "BATCHER",
                         "class_name": "Batcher"
+                    },
+                    {
+                        "name": "CLEANER",
+                        "class_name": "Cleaner"
                     }
                 ]
             },
@@ -66,6 +70,12 @@
                         "Sequencer",
                         "Batcher"
                     ]
+                },
+                {
+                    "tag": "v2",
+                    "new_sqlite_classes": [
+                        "Cleaner"
+                    ]
                 }
             ]
         },
@@ -112,6 +122,10 @@
                     {
                         "name": "BATCHER",
                         "class_name": "Batcher"
+                    },
+                    {
+                        "name": "CLEANER",
+                        "class_name": "Cleaner"
                     }
                 ]
             },
@@ -123,6 +137,12 @@
                         "Sequencer",
                         "Batcher"
                     ]
+                },
+                {
+                    "tag": "v2",
+                    "new_sqlite_classes": [
+                        "Cleaner"
+                    ]
                 }
             ]
         }