Merge pull request 'Refactor resync manager' (#1476) from krtab/garage:refactor_resync_manager into main-v2

Reviewed-on: https://git.deuxfleurs.fr/Deuxfleurs/garage/pulls/1476
Reviewed-by: Armael <armael@noreply.localhost>

src/api/admin/node.rs

@@ -161,7 +161,7 @@ impl RequestHandler for LocalGetNodeStatisticsRequest {
 		let block_manager_stats = NodeBlockManagerStats {
 			rc_entries: garage.block_manager.rc_approximate_len()? as u64,
 			resync_queue_len: garage.block_manager.resync.queue_approximate_len()? as u64,
-			resync_errors: garage.block_manager.resync.errors_approximate_len()? as u64,
+			resync_errors: garage.block_manager.resync.errored() as u64,
 		};
 
 		// Gather block manager statistics

src/block/manager.rs

@@ -148,7 +148,7 @@ impl BlockManager {
 			.expect("Unable to open block_local_rc tree");
 		let rc = BlockRc::new(rc);
 
-		let resync = BlockResyncManager::new(db, &system);
+		let resync = BlockResyncManager::new(db, &system)?;
 
 		let endpoint = system
 			.netapp
@@ -159,8 +159,7 @@ impl BlockManager {
 		let metrics = BlockManagerMetrics::new(
 			config.compression_level,
 			rc.rc_table.untyped().clone(),
-			resync.queue.untyped().clone(),
-			resync.errors.untyped().clone(),
+			resync.idxqueue.clone(),
 			buffer_kb_semaphore.clone(),
 		);
 
@@ -446,15 +445,18 @@ impl BlockManager {
 
 	/// List all resync errors
 	pub fn list_resync_errors(&self) -> Result<Vec<BlockResyncErrorInfo>, Error> {
-		let mut blocks = Vec::with_capacity(self.resync.errors.approximate_len()?);
-		for ent in self.resync.errors.iter()? {
-			let (hash, cnt) = ent?;
+		let mut blocks = Vec::with_capacity(self.resync.errored());
+		for ent in self.resync.idxqueue.lock().unwrap().iter()? {
+			let (hash, ResyncEntry { when, errors }) = ent?;
+			if errors == 0 {
+				continue;
+			}
 			blocks.push(BlockResyncErrorInfo {
 				hash,
 				refcount: 0,
-				error_count: cnt.errors,
-				last_try: cnt.last_try,
-				next_try: cnt.next_try(),
+				error_count: errors,
+				last_try: when - retry_delay_ms(errors - 1),
+				next_try: when,
 			});
 		}
 		for block in blocks.iter_mut() {
@@ -483,7 +485,7 @@ impl BlockManager {
 			tokio::spawn(async move {
 				if let Err(e) = this
 					.resync
-					.put_to_resync(&hash, 2 * this.system.rpc_helper().rpc_timeout())
+					.put_to_resync_after(&hash, 2 * this.system.rpc_helper().rpc_timeout())
 				{
 					error!("Block {:?} could not be put in resync queue: {}.", hash, e);
 				}
@@ -508,7 +510,7 @@ impl BlockManager {
 			tokio::spawn(async move {
 				if let Err(e) = this
 					.resync
-					.put_to_resync(&hash, BLOCK_GC_DELAY + Duration::from_secs(10))
+					.put_to_resync_after(&hash, BLOCK_GC_DELAY + Duration::from_secs(10))
 				{
 					error!("Block {:?} could not be put in resync queue: {}.", hash, e);
 				}
@@ -622,7 +624,8 @@ impl BlockManager {
 				.await
 				.move_block_to_corrupted(block_path)
 				.await?;
-			self.resync.put_to_resync(hash, Duration::from_millis(0))?;
+			self.resync
+				.put_to_resync_after(hash, Duration::from_millis(0))?;
 
 			return Err(Error::CorruptData(*hash));
 		}

src/block/metrics.rs

@@ -1,4 +1,4 @@
-use std::sync::Arc;
+use std::{convert::TryInto, sync::Arc};
 
 use tokio::sync::Semaphore;
 
@@ -6,6 +6,8 @@ use opentelemetry::{global, metrics::*};
 
 use garage_db as db;
 
+use crate::resync::IndexedQueue;
+
 /// `TableMetrics` reference all counter used for metrics
 pub struct BlockManagerMetrics {
 	pub(crate) _compression_level: ValueObserver<u64>,
@@ -34,8 +36,7 @@ impl BlockManagerMetrics {
 	pub fn new(
 		compression_level: Option<i32>,
 		rc_tree: db::Tree,
-		resync_queue: db::Tree,
-		resync_errors: db::Tree,
+		resync_queue: Arc<std::sync::Mutex<IndexedQueue>>,
 		buffer_semaphore: Arc<Semaphore>,
 	) -> Self {
 		let meter = global::meter("garage_model/block");
@@ -57,21 +58,22 @@ impl BlockManagerMetrics {
 				})
 				.with_description("Number of blocks known to the reference counter")
 				.init(),
-			_resync_queue_len: meter
+			_resync_queue_len: {
+				let resync_queue = resync_queue.clone();
+				meter
 				.u64_value_observer("block.resync_queue_length", move |observer| {
-					if let Ok(value) = resync_queue.approximate_len() {
-						observer.observe(value as u64, &[]);
-					}
+					let len = resync_queue.lock().unwrap().approximate_len().unwrap_or_default();
+						observer.observe(len.try_into().unwrap(), &[]);
 				})
 				.with_description(
 					"Number of block hashes queued for local check and possible resync",
 				)
-				.init(),
+				.init()
+			},
 			_resync_errored_blocks: meter
 				.u64_value_observer("block.resync_errored_blocks", move |observer| {
-					if let Ok(value) = resync_errors.approximate_len() {
-						observer.observe(value as u64, &[]);
-					}
+					let errs = resync_queue.lock().unwrap().errored();
+						observer.observe(errs, &[]);
 				})
 				.with_description("Number of block hashes whose last resync resulted in an error")
 				.init(),

src/block/repair.rs

@@ -125,7 +125,7 @@ impl Worker for RepairWorker {
 				for hash in batch_of_hashes.into_iter() {
 					self.manager
 						.resync
-						.put_to_resync(&hash, Duration::from_secs(0))?;
+						.put_to_resync_after(&hash, Duration::from_secs(0))?;
 					self.next_start = Some(hash);
 				}
 
@@ -139,7 +139,7 @@ impl Worker for RepairWorker {
 				if let Some((_path, hash)) = bi.next().await? {
 					self.manager
 						.resync
-						.put_to_resync(&hash, Duration::from_secs(0))?;
+						.put_to_resync_after(&hash, Duration::from_secs(0))?;
 					Ok(WorkerState::Busy)
 				} else {
 					Ok(WorkerState::Done)

src/block/resync.rs

@@ -1,4 +1,4 @@
-use std::collections::HashSet;
+use std::collections::{BTreeSet, HashSet};
 use std::convert::TryInto;
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
@@ -37,64 +37,216 @@ pub(crate) const RESYNC_RETRY_DELAY: Duration = Duration::from_secs(60);
 // The maximum retry delay is 60 seconds * 2^6 = 60 seconds << 6 = 64 minutes (~1 hour)
 pub(crate) const RESYNC_RETRY_DELAY_MAX_BACKOFF_POWER: u64 = 6;
 
+// Double the retry delay for each error, but no more than
+// RESYNC_RETRY_DELAY_MAX_BACKOFF_POWER times.
+// Used to implement exponential backoff
+pub(crate) fn retry_delay_ms(errors: u64) -> u64 {
+	(RESYNC_RETRY_DELAY.as_millis() as u64)
+		<< u64::min(errors, RESYNC_RETRY_DELAY_MAX_BACKOFF_POWER)
+}
+
 // No more than 4 resync workers can be running in the system
 pub(crate) const MAX_RESYNC_WORKERS: usize = 8;
 // Resync tranquility is initially set to 2, but can be changed in the CLI
 // and the updated version is persisted over Garage restarts
 const INITIAL_RESYNC_TRANQUILITY: u32 = 2;
 
-pub struct BlockResyncManager {
-	pub(crate) queue: db::TypedTree<ResyncQueueKey, Hash>,
-	pub(crate) notify: Arc<Notify>,
-	pub(crate) errors: db::TypedTree<Hash, ErrorCounter>,
-
-	busy_set: BusySet,
-
-	persister: PersisterShared<ResyncPersistedConfig>,
-}
-
-/// Key of the resync queue tree: blocks are resynced in order of increasing
-/// `when` (msec timestamp of the next try), with the block hash as tie-breaker.
-///
-// CAREFUL: this type implements `DbOrdKey`, so its byte encoding must be
-// order-preserving.
-// The derived `Ord` compares fields in declaration order, which must match
-// the order in which `encode()` writes them; and `when` must remain an
-// *unsigned* integer, as the big-endian encoding is only order-preserving
-// for unsigned types.
-#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
-pub(crate) struct ResyncQueueKey {
+#[derive(Clone, Copy, Debug)]
+pub(crate) struct ResyncEntry {
 	pub(crate) when: u64,
-	pub(crate) hash: Hash,
+	pub(crate) errors: u64,
 }
 
-impl db::DbBytes for ResyncQueueKey {
+impl ResyncEntry {
+	const ENCODE_LEN: usize =
+		{
+			let res = 2 * size_of::<u64>();
+			if res != size_of::<Self>() {
+				panic!("Size of ResyncEntry has changed, you likely want to change the encoding as well")
+			}
+			res
+		};
+}
+
+impl db::DbBytes for ResyncEntry {
 	fn encode(&self) -> Vec<u8> {
-		let mut v = Vec::with_capacity(40);
-		v.extend_from_slice(&u64::to_be_bytes(self.when));
-		v.extend_from_slice(self.hash.as_slice());
-		v
+		let res = [u64::to_be_bytes(self.when), u64::to_be_bytes(self.errors)].concat();
+		assert_eq!(res.len(), Self::ENCODE_LEN);
+		res
 	}
 
 	fn decode(bytes: &[u8]) -> std::result::Result<Self, db::DecodeError> {
-		if bytes.len() != 40 {
+		if bytes.len() != Self::ENCODE_LEN {
 			return Err(db::DecodeError(
 				format!(
-					"invalid resync queue key: expected 40 bytes, got {}",
+					"invalid error counter: expected {}  bytes, got {}",
+					Self::ENCODE_LEN,
 					bytes.len()
 				)
 				.into(),
 			));
 		}
-		Ok(ResyncQueueKey {
-			when: u64::from_be_bytes(bytes[..8].try_into().unwrap()),
-			hash: Hash::try_from(&bytes[8..])
-				.ok_or_else(|| db::DecodeError("invalid resync queue key: bad hash".into()))?,
+		// Split the encoding as [when || errors]
+		let parts: [[u8; 8]; 2] = bytes.as_chunks().0.try_into().unwrap();
+		// Decode each word
+		let deser: [u64; 2] = parts.map(u64::from_be_bytes);
+		Ok(Self {
+			when: deser[0],
+			errors: deser[1],
 		})
 	}
 }
 
-impl db::DbOrdKey for ResyncQueueKey {}
+#[derive(PartialEq, Eq, PartialOrd, Ord, Debug)]
+// /!\ Order of fields matter for the derivation of Ord!
+struct WhenIndexEntry {
+	when: u64,
+	hash: Hash,
+}
+
+pub struct IndexedQueue {
+	queue: db::TypedTree<Hash, ResyncEntry>,
+	// Index by when in the queue (the same set of hashes is contained in both)
+	when_index: BTreeSet<WhenIndexEntry>,
+	// Numbers of elements of queue that have errors > 0
+	errored: u64,
+	// This is the set of blocks being resynced by one of the workers.
+	// All the hashes here must appear in queue as well.
+	busy_set: HashSet<Hash>,
+}
+
+impl IndexedQueue {
+	fn entry(&mut self, hash: Hash) -> Result<IndexedQueueEntry<'_>, db::Error> {
+		Ok(match self.queue.get(&hash)? {
+			Some(value) => IndexedQueueEntry::Occupied(OccupiedEntry {
+				origin: self,
+				hash,
+				value,
+			}),
+			None => IndexedQueueEntry::Vacant(VacantEntry { origin: self, hash }),
+		})
+	}
+
+	fn clear(&mut self) -> Result<(), db::Error> {
+		self.queue.clear()?;
+		self.when_index.clear();
+		Ok(())
+	}
+
+	/// Hash order
+	pub(crate) fn iter(
+		&self,
+	) -> Result<
+		impl Iterator<Item = Result<(FixedBytes32, ResyncEntry), db::Error>> + use<'_>,
+		db::Error,
+	> {
+		self.queue.iter()
+	}
+
+	pub fn errored(&self) -> u64 {
+		self.errored
+	}
+
+	// May actually be exact in our case because of the mutex
+	pub fn approximate_len(&self) -> Result<usize, garage_db::DbError> {
+		self.queue.approximate_len()
+	}
+}
+
+struct OccupiedEntry<'idxqueue> {
+	origin: &'idxqueue mut IndexedQueue,
+	hash: Hash,
+	value: ResyncEntry,
+}
+
+impl<'idxqueue> OccupiedEntry<'idxqueue> {
+	fn set_when_and_errors(&mut self, new_when: u64, new_errors: u64) -> Result<(), db::Error> {
+		self.origin.queue.insert(
+			&self.hash,
+			&ResyncEntry {
+				when: new_when,
+				errors: new_errors,
+			},
+		)?;
+		let was_there = self.origin.when_index.remove(&WhenIndexEntry {
+			when: self.value.when,
+			hash: self.hash,
+		});
+		debug_assert!(
+			was_there,
+			"The entry was not in the when index anymore (set_when_and_errors)"
+		);
+		self.origin.when_index.insert(WhenIndexEntry {
+			when: new_when,
+			hash: self.hash,
+		});
+		match (self.value.errors, new_errors) {
+			(0, 0) => (),
+			(0, _) => self.origin.errored = self.origin.errored.checked_add(1).unwrap(),
+			(_, 0) => self.origin.errored = self.origin.errored.checked_sub(1).unwrap(),
+			(_, _) => (),
+		}
+		self.value.when = new_when;
+		self.value.errors = new_errors;
+		Ok(())
+	}
+
+	fn set_when(&mut self, new_when: u64) -> Result<(), db::Error> {
+		self.set_when_and_errors(new_when, self.errors())
+	}
+
+	fn remove(self) -> Result<(), db::Error> {
+		self.origin.queue.remove(&self.hash)?;
+		let was_there = self.origin.when_index.remove(&WhenIndexEntry {
+			when: self.when(),
+			hash: self.hash,
+		});
+		debug_assert!(
+			was_there,
+			"The entry was not in the when index anymore (remove)"
+		);
+		if self.value.errors > 0 {
+			self.origin.errored = self.origin.errored.checked_sub(1).unwrap();
+		}
+		Ok(())
+	}
+
+	fn errors(&self) -> u64 {
+		self.value.errors
+	}
+
+	fn when(&self) -> u64 {
+		self.value.when
+	}
+}
+
+struct VacantEntry<'idxqueue> {
+	origin: &'idxqueue mut IndexedQueue,
+	hash: Hash,
+}
+impl<'idxqueue> VacantEntry<'idxqueue> {
+	fn insert(&mut self, when: u64) -> Result<(), db::Error> {
+		self.origin
+			.queue
+			.insert(&self.hash, &ResyncEntry { when, errors: 0 })?;
+		self.origin.when_index.insert(WhenIndexEntry {
+			when,
+			hash: self.hash,
+		});
+		Ok(())
+	}
+}
+
+enum IndexedQueueEntry<'idxqueue> {
+	Occupied(OccupiedEntry<'idxqueue>),
+	Vacant(VacantEntry<'idxqueue>),
+}
+
+pub struct BlockResyncManager {
+	pub(crate) idxqueue: Arc<Mutex<IndexedQueue>>,
+	pub(crate) notify: Arc<Notify>,
+	persister: PersisterShared<ResyncPersistedConfig>,
+}
 
 #[derive(Serialize, Deserialize, Clone, Copy)]
 struct ResyncPersistedConfig {
@@ -113,56 +265,59 @@ impl Default for ResyncPersistedConfig {
 
 enum ResyncIterResult {
 	BusyDidSomething,
-	BusyDidNothing,
 	IdleFor(Duration),
 }
 
-type BusySet = Arc<Mutex<HashSet<ResyncQueueKey>>>;
-
-struct BusyBlock {
-	key: ResyncQueueKey,
-	busy_set: BusySet,
-}
-
 impl BlockResyncManager {
-	pub(crate) fn new(db: &db::Db, system: &System) -> Self {
+	pub(crate) fn new(db: &db::Db, system: &System) -> Result<Self, Error> {
+		v1::migrate_resync_queue_v1_to_v2(db)?;
 		let queue = db
-			.open_typed_tree("block_local_resync_queue")
+			.open_typed_tree::<Hash, ResyncEntry, _>("block_local_resync_queue_v2")
 			.expect("Unable to open block_local_resync_queue tree");
 
-		let errors = db
-			.open_typed_tree("block_local_resync_errors")
-			.expect("Unable to open block_local_resync_errors tree");
-
 		let persister = PersisterShared::new(&system.metadata_dir, "resync_cfg");
 
-		Self {
-			queue,
-			notify: Arc::new(Notify::new()),
-			errors,
-			busy_set: Arc::new(Mutex::new(HashSet::new())),
+		let (when_index, errored) =
+			queue
+				.iter()?
+				.try_fold((BTreeSet::new(), 0), |(mut index, errored), tree_row| {
+					let (hash, ResyncEntry { when, errors }) = tree_row?;
+					index.insert(WhenIndexEntry { when, hash });
+					let errored = errored + if errors > 0 { 1 } else { 0 };
+					Ok::<_, Error>((index, errored))
+				})?;
+
+		Ok(Self {
 			persister,
-		}
+			notify: Arc::new(Notify::new()),
+			idxqueue: Arc::new(Mutex::new(IndexedQueue {
+				queue,
+				when_index,
+				errored,
+				busy_set: HashSet::new(),
+			})),
+		})
 	}
 
 	/// Get length of resync queue
 	pub fn queue_approximate_len(&self) -> Result<usize, Error> {
-		Ok(self.queue.approximate_len()?)
+		let idxqueue = self.idxqueue.lock().unwrap();
+		Ok(idxqueue.approximate_len()?)
 	}
 
-	/// Get number of blocks that have an error
-	pub fn errors_approximate_len(&self) -> Result<usize, Error> {
-		Ok(self.errors.approximate_len()?)
+	/// Get length of resync queue
+	pub fn errored(&self) -> usize {
+		let idxqueue = self.idxqueue.lock().unwrap();
+		idxqueue.errored().try_into().unwrap()
 	}
 
 	/// Clear the error counter for a block and put it in queue immediately
 	pub fn clear_backoff(&self, hash: &Hash) -> Result<(), Error> {
 		let now = now_msec();
-		if let Some(mut ec) = self.errors.get(hash)? {
-			if ec.errors > 0 {
-				ec.last_try = now - ec.delay_msec();
-				self.errors.insert(hash, &ec)?;
-				self.put_to_resync_at(hash, now)?;
+		let mut idxqueue = self.idxqueue.lock().unwrap();
+		if let IndexedQueueEntry::Occupied(mut resync_entry) = idxqueue.entry(*hash)? {
+			if resync_entry.errors() > 0 {
+				resync_entry.set_when_and_errors(now, 0)?;
 				return Ok(());
 			}
 		}
@@ -175,8 +330,8 @@ impl BlockResyncManager {
 	/// Clear the entire resync queue and list of errored blocks
 	/// Corresponds to `garage repair clear-resync-queue`
 	pub fn clear_resync_queue(&self) -> Result<(), Error> {
-		self.queue.clear()?;
-		self.errors.clear()?;
+		let mut idxqueue = self.idxqueue.lock().unwrap();
+		idxqueue.clear()?;
 		Ok(())
 	}
 
@@ -212,163 +367,58 @@ impl BlockResyncManager {
 		);
 	}
 
-	// ---- Resync loop ----
-
-	// This part manages a queue of blocks that need to be
-	// "resynchronized", i.e. that need to have a check that
-	// they are at present if we need them, or that they are
-	// deleted once the garbage collection delay has passed.
-	//
-	// Here are some explanations on how the resync queue works.
-	// There are two db trees that are used to have information
-	// about the status of blocks that need to be resynchronized:
-	//
-	// - resync.queue: a tree that is ordered first by a timestamp
-	//   (in milliseconds since Unix epoch) that is the time at which
-	//   the resync must be done, and second by block hash.
-	//   The key in this tree is just:
-	//       concat(timestamp (8 bytes), hash (32 bytes))
-	//   The value is the same 32-byte hash.
-	//
-	// - resync.errors: a tree that indicates for each block
-	//   if the last resync resulted in an error, and if so,
-	//   the following two information (see the ErrorCounter struct):
-	//   - how many consecutive resync errors for this block?
-	//   - when was the last try?
-	//   These two information are used to implement an
-	//   exponential backoff retry strategy.
-	//   The key in this tree is the 32-byte hash of the block,
-	//   and the value is the encoded ErrorCounter value.
-	//
-	// We need to have these two trees, because the resync queue
-	// is not just a queue of items to process, but a set of items
-	// that are waiting a specific delay until we can process them
-	// (the delay being necessary both internally for the exponential
-	// backoff strategy, and exposed as a parameter when adding items
-	// to the queue, e.g. to wait until the GC delay has passed).
-	// This is why we need one tree ordered by time, and one
-	// ordered by identifier of item to be processed (block hash).
-	//
-	// When the worker wants to process an item it takes from
-	// resync.queue, it checks in resync.errors that if there is an
-	// exponential back-off delay to await, it has passed before we
-	// process the item. If not, the item in the queue is skipped
-	// (but added back for later processing after the time of the
-	// delay).
-	//
-	// An alternative that would have seemed natural is to
-	// only add items to resync.queue with a processing time that is
-	// after the delay, but there are several issues with this:
-	// - This requires to synchronize updates to resync.queue and
-	//   resync.errors (with the current model, there is only one thread,
-	//   the worker thread, that accesses resync.errors,
-	//   so no need to synchronize) by putting them both in a lock.
-	//   This would mean that block_incref might need to take a lock
-	//   before doing its thing, meaning it has much more chances of
-	//   not completing successfully if something bad happens to Garage.
-	//   Currently Garage is not able to recover from block_incref that
-	//   doesn't complete successfully, because it is necessary to ensure
-	//   the consistency between the state of the block manager and
-	//   information in the BlockRef table.
-	// - If a resync fails, we put that block in the resync.errors table,
-	//   and also add it back to resync.queue to be processed after
-	//   the exponential back-off delay,
-	//   but maybe the block is already scheduled to be resynced again
-	//   at another time that is before the exponential back-off delay,
-	//   and we have no way to check that easily. This means that
-	//   in all cases, we need to check the resync.errors table
-	//   in the resync loop at the time when a block is popped from
-	//   the resync.queue.
-	// Overall, the current design is therefore simpler and more robust
-	// because it tolerates inconsistencies between the resync.queue
-	// and resync.errors table (items being scheduled in resync.queue
-	// for times that are earlier than the exponential back-off delay
-	// is a natural condition that is handled properly).
-
-	pub(crate) fn put_to_resync(&self, hash: &Hash, delay: Duration) -> db::Result<()> {
+	pub(crate) fn put_to_resync_after(&self, hash: &Hash, delay: Duration) -> Result<(), Error> {
 		let when = now_msec() + delay.as_millis() as u64;
-		self.put_to_resync_at(hash, when)
+		self.put_to_resync_at_or_later(hash, when)
 	}
 
-	pub(crate) fn put_to_resync_at(&self, hash: &Hash, when: u64) -> db::Result<()> {
+	pub(crate) fn put_to_resync_at_or_later(&self, hash: &Hash, when: u64) -> Result<(), Error> {
 		trace!("Put resync_queue: {} {:?}", when, hash);
-		let qkey = ResyncQueueKey { when, hash: *hash };
-		self.queue.insert(&qkey, hash)?;
+		let mut idxqueue = self.idxqueue.lock().unwrap();
+		match idxqueue.entry(*hash)? {
+			IndexedQueueEntry::Occupied(mut occupied_entry) => {
+				let old_when = occupied_entry.when();
+				// We consider that `when` is a constraint indicating that a resync should be done
+				// but would lead to worth performance or maybe incorrect behavior if done before the given date,
+				// so we give precedence to the latest of the two.
+				occupied_entry.set_when(u64::max(old_when, when))?;
+			}
+			IndexedQueueEntry::Vacant(mut vacant_entry) => {
+				vacant_entry.insert(when)?;
+			}
+		}
 		self.notify.notify_waiters();
 		Ok(())
 	}
 
 	async fn resync_iter(&self, manager: &BlockManager) -> Result<ResyncIterResult, db::Error> {
-		if let Some(block) = self.get_block_to_resync()? {
-			let time_msec = block.key.when;
-			let now = now_msec();
-
-			if now >= time_msec {
-				let hash = block.key.hash;
-
-				if let Some(ec) = self.errors.get(&hash)? {
-					if now < ec.next_try() {
-						// if next retry after an error is not yet,
-						// don't do resync and return early, but still
-						// make sure the item is still in queue at expected time
-						self.put_to_resync_at(&hash, ec.next_try())?;
-						// ec.next_try() > now >= time_msec, so this remove
-						// is not removing the one we added just above
-						// (we want to do the remove after the insert to ensure
-						// that the item is not lost if we crash in-between)
-						self.queue.remove(&block.key)?;
-						return Ok(ResyncIterResult::BusyDidNothing);
+		let block = {
+			let idxqueue = &mut *self.idxqueue.lock().unwrap();
+			let mut iter = idxqueue.when_index.iter();
+			loop {
+				match iter.next() {
+					None => break None,
+					Some(&WhenIndexEntry { when, hash }) if !idxqueue.busy_set.contains(&hash) => {
+						idxqueue.busy_set.insert(hash);
+						break Some((when, hash));
 					}
+					Some(_) => continue,
 				}
-
-				let tracer = opentelemetry::global::tracer("garage");
-				let trace_id = gen_uuid();
-				let span = tracer
-					.span_builder("Resync block")
-					.with_trace_id(
-						opentelemetry::trace::TraceId::from_hex(&hex::encode(
-							&trace_id.as_slice()[..16],
-						))
-						.unwrap(),
-					)
-					.with_attributes(vec![KeyValue::new("block", format!("{:?}", hash))])
-					.start(&tracer);
-
-				let res = self
-					.resync_block(manager, &hash)
-					.with_context(Context::current_with_span(span))
-					.bound_record_duration(&manager.metrics.resync_duration)
-					.await;
-
-				manager.metrics.resync_counter.add(1);
-
-				if let Err(e) = &res {
-					manager.metrics.resync_error_counter.add(1);
-					error!("Error when resyncing {:?}: {}", hash, e);
-
-					let err_counter = match self.errors.get(&hash)? {
-						Some(ec) => ec.add1(now + 1),
-						None => ErrorCounter::new(now + 1),
-					};
-
-					self.errors.insert(&hash, &err_counter)?;
-
-					self.put_to_resync_at(&hash, err_counter.next_try())?;
-					// err_counter.next_try() >= now + 1 > now,
-					// the entry we remove from the queue is not
-					// the entry we inserted with put_to_resync_at
-					self.queue.remove(&block.key)?;
-				} else {
-					self.errors.remove(&hash)?;
-					self.queue.remove(&block.key)?;
-				}
-
-				Ok(ResyncIterResult::BusyDidSomething)
-			} else {
-				Ok(ResyncIterResult::IdleFor(Duration::from_millis(
-					time_msec - now,
-				)))
 			}
+		};
+		if let Some((when, hash)) = block {
+			let res = self
+				.resync_iter_process_one_block(when, hash, manager)
+				.await;
+			// Moving this line up will cause deadlock. (resync_iter_process_one_block takes
+			// the (non re entrant) lock as well and there is an await point)
+			let mut idxqueue = self.idxqueue.lock().unwrap();
+			// /!\ This "lock" (ie removing from the busy set) will not be
+			// released in case of early return in resync_iter.
+			// Be careful when adding `?` in this function ;)
+			let was_there = idxqueue.busy_set.remove(&hash);
+			debug_assert!(was_there);
+			res
 		} else {
 			// Here we wait either for a notification that an item has been
 			// added to the queue, or for a constant delay of 10 secs to expire.
@@ -380,19 +430,67 @@ impl BlockResyncManager {
 		}
 	}
 
-	fn get_block_to_resync(&self) -> Result<Option<BusyBlock>, db::Error> {
-		let mut busy = self.busy_set.lock().unwrap();
-		for it in self.queue.iter()? {
-			let (key, _) = it?;
-			if !busy.contains(&key) {
-				busy.insert(key);
-				return Ok(Some(BusyBlock {
-					key,
-					busy_set: self.busy_set.clone(),
-				}));
+	async fn resync_iter_process_one_block(
+		&self,
+		when: u64,
+		hash: FixedBytes32,
+		manager: &BlockManager,
+	) -> Result<ResyncIterResult, garage_db::Error> {
+		let time_msec = when;
+		let now = now_msec();
+
+		if now >= time_msec {
+			let tracer = opentelemetry::global::tracer("garage");
+			let trace_id = gen_uuid();
+			let span = tracer
+				.span_builder("Resync block")
+				.with_trace_id(
+					opentelemetry::trace::TraceId::from_hex(&hex::encode(
+						&trace_id.as_slice()[..16],
+					))
+					.unwrap(),
+				)
+				.with_attributes(vec![KeyValue::new("block", format!("{:?}", hash))])
+				.start(&tracer);
+
+			let res = self
+				.resync_block(manager, &hash)
+				.with_context(Context::current_with_span(span))
+				.bound_record_duration(&manager.metrics.resync_duration)
+				.await;
+
+			manager.metrics.resync_counter.add(1);
+
+			if let Err(e) = &res {
+				manager.metrics.resync_error_counter.add(1);
+				error!("Error when resyncing {:?}: {}", hash, e);
 			}
+
+			let mut idxqueue = self.idxqueue.lock().unwrap();
+			match idxqueue.entry(hash)? {
+				IndexedQueueEntry::Vacant(_) => {
+					// The resync queue was cleared concurrently while this block was
+					// being processed (e.g. via `garage repair clear-resync-queue`).
+					// There is nothing left to update.
+				}
+				IndexedQueueEntry::Occupied(mut entry) => {
+					if res.is_err() {
+						entry.set_when_and_errors(
+							now + retry_delay_ms(entry.errors()),
+							entry.errors() + 1,
+						)?;
+					} else {
+						entry.remove()?;
+					}
+				}
+			}
+
+			Ok(ResyncIterResult::BusyDidSomething)
+		} else {
+			Ok(ResyncIterResult::IdleFor(Duration::from_millis(
+				time_msec - now,
+			)))
 		}
-		Ok(None)
 	}
 
 	async fn resync_block(&self, manager: &BlockManager, hash: &Hash) -> Result<(), Error> {
@@ -541,13 +639,6 @@ impl BlockResyncManager {
 	}
 }
 
-impl Drop for BusyBlock {
-	fn drop(&mut self) {
-		let mut busy = self.busy_set.lock().unwrap();
-		busy.remove(&self.key);
-	}
-}
-
 pub(crate) struct ResyncWorker {
 	index: usize,
 	manager: Arc<BlockManager>,
@@ -590,9 +681,7 @@ impl Worker for ResyncWorker {
 		WorkerStatus {
 			queue_length: Some(self.manager.resync.queue_approximate_len().unwrap_or(0) as u64),
 			tranquility: Some(tranquility),
-			persistent_errors: Some(
-				self.manager.resync.errors_approximate_len().unwrap_or(0) as u64
-			),
+			persistent_errors: Some(self.manager.resync.errored().try_into().unwrap()),
 			..Default::default()
 		}
 	}
@@ -609,7 +698,6 @@ impl Worker for ResyncWorker {
 			Ok(ResyncIterResult::BusyDidSomething) => {
 				Ok(self.tranquilizer.tranquilize_worker(tranquility))
 			}
-			Ok(ResyncIterResult::BusyDidNothing) => Ok(WorkerState::Busy),
 			Ok(ResyncIterResult::IdleFor(delay)) => {
 				self.next_delay = delay;
 				Ok(WorkerState::Idle)
@@ -650,62 +738,129 @@ impl Worker for ResyncWorker {
 	}
 }
 
-/// Counts the number of errors when resyncing a block,
-/// and the time of the last try.
-///
-/// Used to implement exponential backoff.
-#[derive(Clone, Copy, Debug)]
-pub(crate) struct ErrorCounter {
-	pub(crate) errors: u64,
-	pub(crate) last_try: u64,
-}
+mod v1 {
+	use std::convert::TryInto as _;
 
-impl db::DbBytes for ErrorCounter {
-	fn encode(&self) -> Vec<u8> {
-		let mut v = Vec::with_capacity(16);
-		v.extend_from_slice(&u64::to_be_bytes(self.errors));
-		v.extend_from_slice(&u64::to_be_bytes(self.last_try));
-		v
+	use garage_db::{self as db, Error};
+	use garage_util::data::Hash;
+
+	use super::ResyncEntry;
+
+	/// Key of the resync queue tree: blocks are resynced in order of increasing
+	/// `when` (msec timestamp of the next try), with the block hash as tie-breaker.
+	///
+	// CAREFUL: this type implements `DbOrdKey`, so its byte encoding must be
+	// order-preserving.
+	// The derived `Ord` compares fields in declaration order, which must match
+	// the order in which `encode()` writes them; and `when` must remain an
+	// *unsigned* integer, as the big-endian encoding is only order-preserving
+	// for unsigned types.
+	#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
+	struct ResyncQueueKey {
+		when: u64,
+		hash: Hash,
 	}
 
-	fn decode(bytes: &[u8]) -> std::result::Result<Self, db::DecodeError> {
-		if bytes.len() != 16 {
-			return Err(db::DecodeError(
-				format!(
-					"invalid error counter: expected 16 bytes, got {}",
-					bytes.len()
-				)
-				.into(),
-			));
+	impl db::DbBytes for ResyncQueueKey {
+		fn encode(&self) -> Vec<u8> {
+			let mut v = Vec::with_capacity(40);
+			v.extend_from_slice(&u64::to_be_bytes(self.when));
+			v.extend_from_slice(self.hash.as_slice());
+			v
 		}
-		Ok(Self {
-			errors: u64::from_be_bytes(bytes[..8].try_into().unwrap()),
-			last_try: u64::from_be_bytes(bytes[8..].try_into().unwrap()),
-		})
-	}
-}
 
-impl ErrorCounter {
-	fn new(now: u64) -> Self {
-		Self {
-			errors: 1,
-			last_try: now,
+		fn decode(bytes: &[u8]) -> std::result::Result<Self, db::DecodeError> {
+			if bytes.len() != 40 {
+				return Err(db::DecodeError(
+					format!(
+						"invalid resync queue key: expected 40 bytes, got {}",
+						bytes.len()
+					)
+					.into(),
+				));
+			}
+			Ok(ResyncQueueKey {
+				when: u64::from_be_bytes(bytes[..8].try_into().unwrap()),
+				hash: Hash::try_from(&bytes[8..])
+					.ok_or_else(|| db::DecodeError("invalid resync queue key: bad hash".into()))?,
+			})
 		}
 	}
 
-	fn add1(self, now: u64) -> Self {
-		Self {
-			errors: self.errors + 1,
-			last_try: now,
+	impl db::DbOrdKey for ResyncQueueKey {}
+
+	/// Counts the number of errors when resyncing a block,
+	/// and the time of the last try.
+	///
+	/// Used to implement exponential backoff.
+	#[derive(Clone, Copy, Debug)]
+	struct ErrorCounter {
+		errors: u64,
+		last_try: u64,
+	}
+
+	impl db::DbBytes for ErrorCounter {
+		fn encode(&self) -> Vec<u8> {
+			let mut v = Vec::with_capacity(16);
+			v.extend_from_slice(&u64::to_be_bytes(self.errors));
+			v.extend_from_slice(&u64::to_be_bytes(self.last_try));
+			v
+		}
+
+		fn decode(bytes: &[u8]) -> std::result::Result<Self, db::DecodeError> {
+			if bytes.len() != 16 {
+				return Err(db::DecodeError(
+					format!(
+						"invalid error counter: expected 16 bytes, got {}",
+						bytes.len()
+					)
+					.into(),
+				));
+			}
+			Ok(Self {
+				errors: u64::from_be_bytes(bytes[..8].try_into().unwrap()),
+				last_try: u64::from_be_bytes(bytes[8..].try_into().unwrap()),
+			})
 		}
 	}
 
-	fn delay_msec(&self) -> u64 {
-		(RESYNC_RETRY_DELAY.as_millis() as u64)
-			<< std::cmp::min(self.errors - 1, RESYNC_RETRY_DELAY_MAX_BACKOFF_POWER)
-	}
+	/// Migrate resync queue and error table from v1 (two trees) to v2 (one merged tree).
+	///
+	/// v1 format:
+	///   `block_local_resync_queue`  — `TypedTree<ResyncQueueKey, Hash>`
+	///   `block_local_resync_errors` — `TypedTree<Hash, ErrorCounter>`
+	///
+	/// v2 format:
+	///   `block_local_resync_queue_v2` — `TypedTree<Hash, ResyncEntry>`
+	///
+	/// Safe to call again after an interrupted run. If the old queue is empty (either
+	/// never populated or already cleared by a prior completed migration) the function is a no-op.
+	/// The old trees are emptied rather than dropped because the `Db` API has no drop-tree primitive.
+	pub(crate) fn migrate_resync_queue_v1_to_v2(db: &db::Db) -> Result<(), Error> {
+		let old_queue =
+			db.open_typed_tree::<ResyncQueueKey, Hash, _>("block_local_resync_queue")?;
+		if old_queue.is_empty()? {
+			return Ok(());
+		}
 
-	pub(crate) fn next_try(&self) -> u64 {
-		self.last_try + self.delay_msec()
+		let new_queue =
+			db.open_typed_tree::<Hash, ResyncEntry, _>("block_local_resync_queue_v2")?;
+		let old_errors =
+			db.open_typed_tree::<Hash, ErrorCounter, _>("block_local_resync_errors")?;
+
+		// The old queue is ordered by (when, hash). For a given hash, its occurrences are
+		// therefore visited in ascending `when` order, so overwriting on every match
+		// means the last (and thus highest) `when` we see for a hash is the one that survives
+		// in the v2 tree — consistent with the max-wins semantics used when updating an
+		// already-queued entry elsewhere (see `put_to_resync_at_or_later`).
+		for row in old_queue.iter()? {
+			let (ResyncQueueKey { when, hash }, _) = row?;
+			let errors = old_errors.get(&hash)?.map(|ec| ec.errors).unwrap_or(0);
+			new_queue.insert(&hash, &ResyncEntry { when, errors })?;
+		}
+
+		old_queue.clear()?;
+		old_errors.clear()?;
+		Ok(())
 	}
 }