add permits to merkle worker

src/table/data.rs

@@ -266,7 +266,13 @@ impl<F: TableSchema, R: TableReplication> TableData<F, R> {
 		// early return if nothing changed
 		let (new_entry, new_bytes_hash) = match changed {
 			Some((e, b)) => (e, b),
-			None => return Ok(None),
+			None => {
+				let maybe_bound = self.merkle_todo_bounded_queue.clone();
+				if let Some(b) = &maybe_bound {
+					b.add_permits(1);
+				}
+				return Ok(None);
+			}
 		};
 
 		// Handle GC in case of tombstone
@@ -313,6 +319,10 @@ impl<F: TableSchema, R: TableReplication> TableData<F, R> {
 			})?;
 
 		if !removed {
+			let maybe_bound = self.merkle_todo_bounded_queue.clone();
+			if let Some(b) = &maybe_bound {
+				b.add_permits(1);
+			}
 			return Ok(false);
 		}
 
@@ -343,6 +353,10 @@ impl<F: TableSchema, R: TableReplication> TableData<F, R> {
 			})?;
 
 		if !removed {
+			let maybe_bound = self.merkle_todo_bounded_queue.clone();
+			if let Some(b) = &maybe_bound {
+				b.add_permits(1);
+			}
 			return Ok(false);
 		}
 

src/table/merkle.rs

@@ -1,5 +1,5 @@
 use std::sync::Arc;
-use std::time::{Duration, SystemTime};
+use std::time::Duration;
 
 use async_trait::async_trait;
 use serde::{Deserialize, Serialize};
@@ -9,7 +9,7 @@ use tokio::sync::watch;
 use garage_db as db;
 
 use garage_util::background::*;
-use garage_util::config::{MerkleBackpressureEnum, MerkleFixedQueue};
+use garage_util::config::MerkleBackpressureEnum;
 use garage_util::data::*;
 use garage_util::encode::{nonversioned_decode, nonversioned_encode};
 use garage_util::error::Error;
@@ -135,6 +135,11 @@ impl<F: TableSchema, R: TableReplication> MerkleUpdater<F, R> {
 				k
 			);
 		}
+
+		let maybe_bound = self.data.merkle_todo_bounded_queue.clone();
+		if let Some(b) = &maybe_bound {
+			b.add_permits(1);
+		}
 		Ok(())
 	}
 

src/table/metrics.rs

@@ -89,7 +89,7 @@ impl TableMetrics {
                                     observer.observe(free, &[KeyValue::new("table_name", table_name)])
                                 }
                             )
-			    .with_description("Merkle tree updater TODO sleep backpressure to apply in ms")
+			    .with_description("Merkle TODO queue free slots")
 			    .init(),
 			_gc_todo_len: meter
 				.u64_value_observer(

src/table/sync.rs

@@ -11,7 +11,6 @@ use serde::{Deserialize, Serialize};
 use serde_bytes::ByteBuf;
 use tokio::select;
 use tokio::sync::{mpsc, watch, Notify};
-use tokio::time::sleep;
 
 use garage_util::background::*;
 use garage_util::data::*;