Update to worker 0.6.0, addressing breaking changes (#62)

- Durable Object methods no longer take a mutable reference, so we need
  to use interior mutability (RefCell) to update state, as recommended
  by the workers-rs maintainers. The tricky part is avoiding passing a
  borrowed Ref to an async function, as that could cause a panic if, for
  example, a concurrent async operation attempts to borrow an
  already-mutably-borrowed value.
- To simplify the above change, remove the Option wrapper from the
  GenericSequencer's 'sequence_state' field. Instead of loading the
  sequencer state just-in-time for the first sequencing, load it during
  the DO initialization (which happens after the first fetch or alarm).
  Similarly, when there is a fatal sequencing error requiring reloading
  the sequence state, load the state right away rather than waiting for
  the next sequencing.

Author: lukevalenta

Cargo.lock

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

Cargo.toml

@@ -7,7 +7,6 @@ use worker::*;
 #[durable_object]
 struct Batcher(GenericBatcher<StaticCTPendingLogEntry>);
 
-#[durable_object]
 impl DurableObject for Batcher {
     fn new(state: State, env: Env) -> Self {
         // Find the Durable Object name by enumerating all possibilities.
@@ -52,7 +51,7 @@ impl DurableObject for Batcher {
         Batcher(GenericBatcher::new(config, kv, sequencer))
     }
 
-    async fn fetch(&mut self, req: Request) -> Result<Response> {
+    async fn fetch(&self, req: Request) -> Result<Response> {
         self.0.fetch(req).await
     }
 }

crates/ct_worker/src/batcher_do.rs

@@ -300,7 +300,7 @@ fn batcher_id_from_lookup_key(key: &LookupKey, num_batchers: u8) -> u8 {
 }
 
 fn headers_from_http_metadata(meta: HttpMetadata) -> Headers {
-    let mut h = Headers::new();
+    let h = Headers::new();
     if let Some(hdr) = meta.cache_control {
         h.append("Cache-Control", &hdr).unwrap();
     }

crates/ct_worker/src/frontend_worker.rs

@@ -17,7 +17,6 @@ use worker::*;
 #[durable_object]
 struct Sequencer(GenericSequencer<StaticCTLogEntry>);
 
-#[durable_object]
 impl DurableObject for Sequencer {
     fn new(state: State, env: Env) -> Self {
         // Find the Durable Object name by enumerating all possibilities.
@@ -77,11 +76,11 @@ impl DurableObject for Sequencer {
         Sequencer(GenericSequencer::new(config, state, bucket, registry))
     }
 
-    async fn fetch(&mut self, req: Request) -> Result<Response> {
+    async fn fetch(&self, req: Request) -> Result<Response> {
         self.0.fetch(req).await
     }
 
-    async fn alarm(&mut self) -> Result<Response> {
+    async fn alarm(&self) -> Result<Response> {
         self.0.alarm().await
     }
 }

crates/ct_worker/src/sequencer_do.rs

@@ -10,6 +10,7 @@ use crate::{LookupKey, SequenceMetadata, BATCH_ENDPOINT, ENTRY_ENDPOINT};
 use base64::prelude::*;
 use futures_util::future::{join_all, select, Either};
 use std::{
+    cell::RefCell,
     collections::{HashMap, HashSet},
     time::Duration,
 };
@@ -23,9 +24,9 @@ pub struct GenericBatcher<E: PendingLogEntry> {
     config: BatcherConfig,
     kv: Option<KvStore>,
     sequencer: Stub,
-    batch: Batch<E>,
-    in_flight: usize,
-    processed: usize,
+    batch: RefCell<Batch<E>>,
+    in_flight: RefCell<usize>,
+    processed: RefCell<usize>,
 }
 
 pub struct BatcherConfig {
@@ -59,9 +60,9 @@ impl<E: PendingLogEntry> GenericBatcher<E> {
             config,
             kv,
             sequencer,
-            batch: Batch::default(),
-            in_flight: 0,
-            processed: 0,
+            batch: RefCell::new(Batch::default()),
+            in_flight: RefCell::new(0),
+            processed: RefCell::new(0),
         }
     }
 
@@ -73,26 +74,26 @@ impl<E: PendingLogEntry> GenericBatcher<E> {
     ///
     /// Returns an error if the request cannot be parsed or response cannot be
     /// constructed.
-    pub async fn fetch(&mut self, mut req: Request) -> Result<Response> {
+    pub async fn fetch(&self, mut req: Request) -> Result<Response> {
         match req.path().as_str() {
             ENTRY_ENDPOINT => {
                 let entry: E = req.json().await?;
                 let key = entry.lookup_key();
 
-                self.in_flight += 1;
-                self.processed += 1;
+                *self.in_flight.borrow_mut() += 1;
+                *self.processed.borrow_mut() += 1;
 
                 // Add entry to the current pending batch if it isn't already present.
                 // Rely on the Sequencer to deduplicate entries across batches.
-                if !self.batch.by_hash.contains(&key) {
-                    self.batch.by_hash.insert(key);
-                    self.batch.entries.push(entry);
+                if !self.batch.borrow().by_hash.contains(&key) {
+                    self.batch.borrow_mut().by_hash.insert(key);
+                    self.batch.borrow_mut().entries.push(entry);
                 }
 
-                let mut recv = self.batch.done.subscribe();
+                let mut recv = self.batch.borrow().done.subscribe();
 
                 // Submit the current pending batch if it's full.
-                if self.batch.entries.len() >= self.config.max_batch_entries {
+                if self.batch.borrow().entries.len() >= self.config.max_batch_entries {
                     if let Err(e) = self.submit_batch().await {
                         log::warn!("{} failed to submit full batch: {e}", self.config.name);
                     }
@@ -112,7 +113,7 @@ impl<E: PendingLogEntry> GenericBatcher<E> {
                         }
                         Either::Right(((), batch_done)) => {
                             // Batch timeout reached; submit this entry's batch if no-one has already.
-                            if self.batch.by_hash.contains(&key) {
+                            if self.batch.borrow().by_hash.contains(&key) {
                                 if let Err(e) = self.submit_batch().await {
                                     log::warn!(
                                         "{} failed to submit timed-out batch: {e}",
@@ -140,7 +141,7 @@ impl<E: PendingLogEntry> GenericBatcher<E> {
                     // submitting the batch or rate limiting at the Sequencer.
                     Response::error("rate limited", 429)
                 };
-                self.in_flight -= 1;
+                *self.in_flight.borrow_mut() -= 1;
 
                 resp
             }
@@ -156,16 +157,16 @@ impl<E: PendingLogEntry> GenericBatcher<E> {
     ///
     /// Returns an error if there are issues constructing or sending requests to
     /// the sequencer or deduplication cache.
-    pub async fn submit_batch(&mut self) -> Result<()> {
+    pub async fn submit_batch(&self) -> Result<()> {
         // Take the current pending batch and replace it with a new one.
-        let batch = std::mem::take(&mut self.batch);
+        let batch = std::mem::take(&mut *self.batch.borrow_mut());
 
         log::debug!(
             "{} submitting batch: leaves={} inflight={} processed={}",
             self.config.name,
             batch.entries.len(),
-            self.in_flight,
-            self.processed,
+            self.in_flight.borrow(),
+            self.processed.borrow(),
         );
 
         // Submit the batch, and wait for it to be sequenced.

crates/generic_log_worker/src/batcher_do.rs

@@ -19,6 +19,7 @@ use log_ops::UploadOptions;
 use metrics::{millis_diff_as_secs, AsF64, ObjectMetrics};
 use serde_bytes::ByteBuf;
 use sha2::{Digest, Sha256};
+use std::cell::RefCell;
 use std::collections::{HashMap, VecDeque};
 use std::io::Write;
 use std::str::FromStr;
@@ -148,7 +149,7 @@ pub async fn put_cache_entry_metadata<L: PendingLogEntry>(
 
 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<()>;
+    async fn put_entries(&self, entries: &[(LookupKey, SequenceMetadata)]) -> Result<()>;
 }
 
 trait CacheRead {
@@ -163,7 +164,7 @@ struct DedupCache {
 
 impl CacheWrite for DedupCache {
     /// Write entries to both the short-term deduplication cache and its backup in DO Storage.
-    async fn put_entries(&mut self, entries: &[(LookupKey, SequenceMetadata)]) -> Result<()> {
+    async fn put_entries(&self, entries: &[(LookupKey, SequenceMetadata)]) -> Result<()> {
         if entries.is_empty() {
             return Ok(());
         }
@@ -194,7 +195,7 @@ impl DedupCache {
     }
 
     // Load batches of cache entries from DO storage into the in-memory cache.
-    async fn load(&mut self) -> Result<()> {
+    async fn load(&self) -> Result<()> {
         let head = self
             .storage
             .get::<usize>(Self::FIFO_HEAD_KEY)
@@ -216,7 +217,7 @@ impl DedupCache {
     }
 
     // Store a batch of cache entries in DO storage.
-    async fn store(&mut self, entries: &[(LookupKey, SequenceMetadata)]) -> Result<()> {
+    async fn store(&self, entries: &[(LookupKey, SequenceMetadata)]) -> Result<()> {
         let head = self
             .storage
             .get::<usize>(Self::FIFO_HEAD_KEY)
@@ -272,38 +273,40 @@ fn deserialize_entries(buf: &[u8]) -> Result<Vec<(LookupKey, SequenceMetadata)>>
 // A fixed-size in-memory FIFO cache.
 struct MemoryCache {
     max_size: usize,
-    map: HashMap<LookupKey, SequenceMetadata>,
-    fifo: VecDeque<LookupKey>,
+    map: RefCell<HashMap<LookupKey, SequenceMetadata>>,
+    fifo: RefCell<VecDeque<LookupKey>>,
 }
 
 impl MemoryCache {
     fn new(max_size: usize) -> Self {
         assert_ne!(max_size, 0);
         Self {
             max_size,
-            fifo: VecDeque::with_capacity(max_size),
-            map: HashMap::with_capacity(max_size),
+            fifo: RefCell::new(VecDeque::with_capacity(max_size)),
+            map: RefCell::new(HashMap::with_capacity(max_size)),
         }
     }
 
     // Get an entry from the in-memory cache.
     fn get_entry(&self, key: &LookupKey) -> Option<SequenceMetadata> {
-        self.map.get(key).copied()
+        self.map.borrow().get(key).copied()
     }
 
     // Put a batch of entries into the in-memory cache,
     // evicting old entries to make room if necessary.
-    fn put_entries(&mut self, entries: &[(LookupKey, SequenceMetadata)]) {
+    fn put_entries(&self, entries: &[(LookupKey, SequenceMetadata)]) {
+        let mut map = self.map.borrow_mut();
+        let mut fifo = self.fifo.borrow_mut();
         for (key, value) in entries {
-            if self.map.contains_key(key) {
+            if map.contains_key(key) {
                 continue;
             }
-            if self.map.len() == self.max_size {
+            if map.len() == self.max_size {
                 // Evict oldest entry to make room.
-                self.map.remove(&self.fifo.pop_front().unwrap());
+                map.remove(&fifo.pop_front().unwrap());
             }
-            self.fifo.push_back(*key);
-            self.map.insert(*key, *value);
+            fifo.push_back(*key);
+            map.insert(*key, *value);
         }
     }
 }