Refactor sharding logic; coming next: full replication with epidemic dissemination

8 files changed, 221 insertions, 165 deletions

diff --git a/src/main.rs b/src/main.rs
index 3e91b21d..cc9da8e2 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -4,6 +4,7 @@ mod error;
 mod background;
 mod membership;
 mod table;
+mod table_sharded;
 mod table_sync;
 
 mod block;
@@ -22,12 +23,15 @@ use std::collections::HashSet;
 use std::net::SocketAddr;
 use std::path::PathBuf;
 use std::sync::Arc;
+use std::time::Duration;
 use structopt::StructOpt;
 
 use error::Error;
 use membership::*;
 use rpc_client::*;
-use server::{TlsConfig, DEFAULT_TIMEOUT};
+use server::TlsConfig;
+
+const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10);
 
 #[derive(StructOpt, Debug)]
 #[structopt(name = "garage")]
@@ -158,11 +162,8 @@ async fn cmd_status(rpc_cli: RpcAddrClient<Message>, rpc_host: SocketAddr) -> Re
 	for adv in status.iter() {
 		if let Some(cfg) = config.members.get(&adv.id) {
 			println!(
-				"{}\t{}\t{}\t{}",
-				hex::encode(&adv.id),
-				cfg.datacenter,
-				cfg.n_tokens,
-				adv.addr
+				"{:?}\t{}\t{}\t{}",
+				adv.id, cfg.datacenter, cfg.n_tokens, adv.addr
 			);
 		}
 	}
@@ -176,7 +177,7 @@ async fn cmd_status(rpc_cli: RpcAddrClient<Message>, rpc_host: SocketAddr) -> Re
 		println!("\nFailed nodes:");
 		for (id, cfg) in config.members.iter() {
 			if !status.iter().any(|x| x.id == *id) {
-				println!("{}\t{}\t{}", hex::encode(&id), cfg.datacenter, cfg.n_tokens);
+				println!("{:?}\t{}\t{}", id, cfg.datacenter, cfg.n_tokens);
 			}
 		}
 	}
@@ -188,7 +189,7 @@ async fn cmd_status(rpc_cli: RpcAddrClient<Message>, rpc_host: SocketAddr) -> Re
 		println!("\nUnconfigured nodes:");
 		for adv in status.iter() {
 			if !config.members.contains_key(&adv.id) {
-				println!("{}\t{}", hex::encode(&adv.id), adv.addr);
+				println!("{:?}\t{}", adv.id, adv.addr);
 			}
 		}
 	}
diff --git a/src/membership.rs b/src/membership.rs
index e2f4bb90..08dd5f2f 100644
--- a/src/membership.rs
+++ b/src/membership.rs
@@ -198,7 +198,7 @@ impl Ring {
 		self.walk_ring_from_pos(start, n)
 	}
 
-	pub fn walk_ring_from_pos(&self, start: usize, n: usize) -> Vec<UUID> {
+	fn walk_ring_from_pos(&self, start: usize, n: usize) -> Vec<UUID> {
 		if n >= self.config.members.len() {
 			return self.config.members.keys().cloned().collect::<Vec<_>>();
 		}
diff --git a/src/object_table.rs b/src/object_table.rs
index 82a64cd1..59ce3b7f 100644
--- a/src/object_table.rs
+++ b/src/object_table.rs
@@ -5,6 +5,7 @@ use std::sync::Arc;
 use crate::background::BackgroundRunner;
 use crate::data::*;
 use crate::table::*;
+use crate::table_sharded::*;
 
 use crate::version_table::*;
 
@@ -90,7 +91,7 @@ impl Entry<String, String> for Object {
 
 pub struct ObjectTable {
 	pub background: Arc<BackgroundRunner>,
-	pub version_table: Arc<Table<VersionTable>>,
+	pub version_table: Arc<Table<VersionTable, TableShardedReplication>>,
 }
 
 #[async_trait]
diff --git a/src/server.rs b/src/server.rs
index e728c667..6b4b5b6b 100644
--- a/src/server.rs
+++ b/src/server.rs
@@ -2,7 +2,6 @@ use std::io::{Read, Write};
 use std::net::SocketAddr;
 use std::path::PathBuf;
 use std::sync::Arc;
-use std::time::Duration;
 
 pub use futures_util::future::FutureExt;
 use serde::Deserialize;
@@ -14,6 +13,7 @@ use crate::error::Error;
 use crate::membership::System;
 use crate::rpc_server::RpcServer;
 use crate::table::*;
+use crate::table_sharded::*;
 
 use crate::block::*;
 use crate::block_ref_table::*;
@@ -22,8 +22,6 @@ use crate::version_table::*;
 
 use crate::api_server;
 
-pub const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10);
-
 #[derive(Deserialize, Debug, Clone)]
 pub struct Config {
 	pub metadata_dir: PathBuf,
@@ -59,9 +57,9 @@ pub struct Garage {
 	pub system: Arc<System>,
 	pub block_manager: Arc<BlockManager>,
 
-	pub object_table: Arc<Table<ObjectTable>>,
-	pub version_table: Arc<Table<VersionTable>>,
-	pub block_ref_table: Arc<Table<BlockRefTable>>,
+	pub object_table: Arc<Table<ObjectTable, TableShardedReplication>>,
+	pub version_table: Arc<Table<VersionTable, TableShardedReplication>>,
+	pub block_ref_table: Arc<Table<BlockRefTable, TableShardedReplication>>,
 }
 
 impl Garage {
@@ -79,18 +77,16 @@ impl Garage {
 		let block_manager =
 			BlockManager::new(&db, config.data_dir.clone(), system.clone(), rpc_server);
 
-		let data_rep_param = TableReplicationParams {
+		let data_rep_param = TableShardedReplication {
 			replication_factor: system.config.data_replication_factor,
 			write_quorum: (system.config.data_replication_factor + 1) / 2,
 			read_quorum: 1,
-			timeout: DEFAULT_TIMEOUT,
 		};
 
-		let meta_rep_param = TableReplicationParams {
+		let meta_rep_param = TableShardedReplication {
 			replication_factor: system.config.meta_replication_factor,
 			write_quorum: (system.config.meta_replication_factor + 1) / 2,
 			read_quorum: (system.config.meta_replication_factor + 1) / 2,
-			timeout: DEFAULT_TIMEOUT,
 		};
 
 		println!("Initialize block_ref_table...");
@@ -99,10 +95,10 @@ impl Garage {
 				background: background.clone(),
 				block_manager: block_manager.clone(),
 			},
+			data_rep_param.clone(),
 			system.clone(),
 			&db,
 			"block_ref".to_string(),
-			data_rep_param.clone(),
 			rpc_server,
 		)
 		.await;
@@ -113,10 +109,10 @@ impl Garage {
 				background: background.clone(),
 				block_ref_table: block_ref_table.clone(),
 			},
+			meta_rep_param.clone(),
 			system.clone(),
 			&db,
 			"version".to_string(),
-			meta_rep_param.clone(),
 			rpc_server,
 		)
 		.await;
@@ -127,10 +123,10 @@ impl Garage {
 				background: background.clone(),
 				version_table: version_table.clone(),
 			},
+			meta_rep_param.clone(),
 			system.clone(),
 			&db,
 			"object".to_string(),
-			meta_rep_param.clone(),
 			rpc_server,
 		)
 		.await;
diff --git a/src/table.rs b/src/table.rs
index f7354376..d5357277 100644
--- a/src/table.rs
+++ b/src/table.rs
@@ -10,30 +10,23 @@ use serde_bytes::ByteBuf;
 
 use crate::data::*;
 use crate::error::Error;
-use crate::membership::System;
+use crate::membership::{Ring, System};
 use crate::rpc_client::*;
 use crate::rpc_server::*;
 use crate::table_sync::*;
 
-pub struct Table<F: TableSchema> {
+const TABLE_RPC_TIMEOUT: Duration = Duration::from_secs(10);
+
+pub struct Table<F: TableSchema, R: TableReplication> {
 	pub instance: F,
+	pub replication: R,
 
 	pub name: String,
 	pub rpc_client: Arc<RpcClient<TableRPC<F>>>,
 
 	pub system: Arc<System>,
 	pub store: sled::Tree,
-	pub syncer: ArcSwapOption<TableSyncer<F>>,
-
-	pub param: TableReplicationParams,
-}
-
-#[derive(Clone)]
-pub struct TableReplicationParams {
-	pub replication_factor: usize,
-	pub read_quorum: usize,
-	pub write_quorum: usize,
-	pub timeout: Duration,
+	pub syncer: ArcSwapOption<TableSyncer<F, R>>,
 }
 
 #[derive(Serialize, Deserialize)]
@@ -112,15 +105,38 @@ pub trait TableSchema: Send + Sync {
 	}
 }
 
-impl<F: TableSchema + 'static> Table<F> {
+pub trait TableReplication: Send + Sync {
+	// See examples in table_sharded.rs and table_fullcopy.rs
+	// To understand various replication methods
+
+	// Which nodes to send reads from
+	fn read_nodes(&self, hash: &Hash, system: &System) -> Vec<UUID>;
+	fn read_quorum(&self) -> usize;
+
+	// Which nodes to send writes to
+	fn write_nodes(&self, hash: &Hash, system: &System) -> Vec<UUID>;
+	fn write_quorum(&self) -> usize;
+	fn max_write_errors(&self) -> usize;
+	fn epidemic_writes(&self) -> bool;
+
+	// Which are the nodes that do actually replicate the data
+	fn replication_nodes(&self, hash: &Hash, ring: &Ring) -> Vec<UUID>;
+	fn split_points(&self, ring: &Ring) -> Vec<Hash>;
+}
+
+impl<F, R> Table<F, R>
+where
+	F: TableSchema + 'static,
+	R: TableReplication + 'static,
+{
 	// =============== PUBLIC INTERFACE FUNCTIONS (new, insert, get, etc) ===============
 
 	pub async fn new(
 		instance: F,
+		replication: R,
 		system: Arc<System>,
 		db: &sled::Db,
 		name: String,
-		param: TableReplicationParams,
 		rpc_server: &mut RpcServer,
 	) -> Arc<Self> {
 		let store = db.open_tree(&name).expect("Unable to open DB tree");
@@ -130,11 +146,11 @@ impl<F: TableSchema + 'static> Table<F> {
 
 		let table = Arc::new(Self {
 			instance,
+			replication,
 			name,
 			rpc_client,
 			system,
 			store,
-			param,
 			syncer: ArcSwapOption::from(None),
 		});
 		table.clone().register_handler(rpc_server, rpc_path);
@@ -147,15 +163,19 @@ impl<F: TableSchema + 'static> Table<F> {
 
 	pub async fn insert(&self, e: &F::E) -> Result<(), Error> {
 		let hash = e.partition_key().hash();
-		let ring = self.system.ring.borrow().clone();
-		let who = ring.walk_ring(&hash, self.param.replication_factor);
+		let who = self.replication.write_nodes(&hash, &self.system);
 		//eprintln!("insert who: {:?}", who);
 
 		let e_enc = Arc::new(ByteBuf::from(rmp_to_vec_all_named(e)?));
 		let rpc = TableRPC::<F>::Update(vec![e_enc]);
 
 		self.rpc_client
-			.try_call_many(&who[..], rpc, self.param.write_quorum, self.param.timeout)
+			.try_call_many(
+				&who[..],
+				rpc,
+				self.replication.write_quorum(),
+				TABLE_RPC_TIMEOUT,
+			)
 			.await?;
 		Ok(())
 	}
@@ -165,8 +185,7 @@ impl<F: TableSchema + 'static> Table<F> {
 
 		for entry in entries.iter() {
 			let hash = entry.partition_key().hash();
-			let ring = self.system.ring.borrow().clone();
-			let who = ring.walk_ring(&hash, self.param.replication_factor);
+			let who = self.replication.write_nodes(&hash, &self.system);
 			let e_enc = Arc::new(ByteBuf::from(rmp_to_vec_all_named(entry)?));
 			for node in who {
 				if !call_list.contains_key(&node) {
@@ -179,7 +198,7 @@ impl<F: TableSchema + 'static> Table<F> {
 		let call_futures = call_list.drain().map(|(node, entries)| async move {
 			let rpc = TableRPC::<F>::Update(entries);
 
-			let resp = self.rpc_client.call(&node, rpc, self.param.timeout).await?;
+			let resp = self.rpc_client.call(&node, rpc, TABLE_RPC_TIMEOUT).await?;
 			Ok::<_, Error>((node, resp))
 		});
 		let mut resps = call_futures.collect::<FuturesUnordered<_>>();
@@ -190,7 +209,7 @@ impl<F: TableSchema + 'static> Table<F> {
 				errors.push(e);
 			}
 		}
-		if errors.len() > self.param.replication_factor - self.param.write_quorum {
+		if errors.len() > self.replication.max_write_errors() {
 			Err(Error::Message("Too many errors".into()))
 		} else {
 			Ok(())
@@ -203,14 +222,18 @@ impl<F: TableSchema + 'static> Table<F> {
 		sort_key: &F::S,
 	) -> Result<Option<F::E>, Error> {
 		let hash = partition_key.hash();
-		let ring = self.system.ring.borrow().clone();
-		let who = ring.walk_ring(&hash, self.param.replication_factor);
+		let who = self.replication.read_nodes(&hash, &self.system);
 		//eprintln!("get who: {:?}", who);
 
 		let rpc = TableRPC::<F>::ReadEntry(partition_key.clone(), sort_key.clone());
 		let resps = self
 			.rpc_client
-			.try_call_many(&who[..], rpc, self.param.read_quorum, self.param.timeout)
+			.try_call_many(
+				&who[..],
+				rpc,
+				self.replication.read_quorum(),
+				TABLE_RPC_TIMEOUT,
+			)
 			.await?;
 
 		let mut ret = None;
@@ -254,14 +277,18 @@ impl<F: TableSchema + 'static> Table<F> {
 		limit: usize,
 	) -> Result<Vec<F::E>, Error> {
 		let hash = partition_key.hash();
-		let ring = self.system.ring.borrow().clone();
-		let who = ring.walk_ring(&hash, self.param.replication_factor);
+		let who = self.replication.read_nodes(&hash, &self.system);
 
 		let rpc =
 			TableRPC::<F>::ReadRange(partition_key.clone(), begin_sort_key.clone(), filter, limit);
 		let resps = self
 			.rpc_client
-			.try_call_many(&who[..], rpc, self.param.read_quorum, self.param.timeout)
+			.try_call_many(
+				&who[..],
+				rpc,
+				self.replication.read_quorum(),
+				TABLE_RPC_TIMEOUT,
+			)
 			.await?;
 
 		let mut ret = BTreeMap::new();
@@ -315,7 +342,7 @@ impl<F: TableSchema + 'static> Table<F> {
 				&who[..],
 				TableRPC::<F>::Update(vec![what_enc]),
 				who.len(),
-				self.param.timeout,
+				TABLE_RPC_TIMEOUT,
 			)
 			.await?;
 		Ok(())
diff --git a/src/table_sharded.rs b/src/table_sharded.rs
new file mode 100644
index 00000000..485a9212
--- /dev/null
+++ b/src/table_sharded.rs
@@ -0,0 +1,55 @@
+use crate::data::*;
+use crate::membership::{System, Ring};
+use crate::table::*;
+
+#[derive(Clone)]
+pub struct TableShardedReplication {
+	pub replication_factor: usize,
+	pub read_quorum: usize,
+	pub write_quorum: usize,
+}
+
+impl TableReplication for TableShardedReplication {
+	// Sharded replication schema:
+	// - based on the ring of nodes, a certain set of neighbors
+	//   store entries, given as a function of the position of the
+	//   entry's hash in the ring
+	// - reads are done on all of the nodes that replicate the data
+	// - writes as well
+
+	fn read_nodes(&self, hash: &Hash, system: &System) -> Vec<UUID> {
+		let ring = system.ring.borrow().clone();
+		ring.walk_ring(&hash, self.replication_factor)
+	}
+	fn read_quorum(&self) -> usize {
+		self.read_quorum
+	}
+
+	fn write_nodes(&self, hash: &Hash, system: &System) -> Vec<UUID> {
+		let ring = system.ring.borrow().clone();
+		ring.walk_ring(&hash, self.replication_factor)
+	}
+	fn write_quorum(&self) -> usize {
+		self.write_quorum
+	}
+	fn max_write_errors(&self) -> usize {
+		self.replication_factor - self.write_quorum
+	}
+	fn epidemic_writes(&self) -> bool {
+		false
+	}
+
+	fn replication_nodes(&self, hash: &Hash, ring: &Ring) -> Vec<UUID> {
+		ring.walk_ring(&hash, self.replication_factor)
+	}
+	fn split_points(&self, ring: &Ring) -> Vec<Hash> {
+		let mut ret = vec![];
+
+		ret.push([0u8; 32].into());
+		for entry in ring.ring.iter() {
+			ret.push(entry.location.clone());
+		}
+		ret.push([0xFFu8; 32].into());
+		ret
+	}
+}
diff --git a/src/table_sync.rs b/src/table_sync.rs
index 3ba2fc6a..b4555a77 100644
--- a/src/table_sync.rs
+++ b/src/table_sync.rs
@@ -1,5 +1,5 @@
 use rand::Rng;
-use std::collections::{BTreeMap, BTreeSet, VecDeque};
+use std::collections::{BTreeMap, VecDeque};
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
@@ -21,10 +21,12 @@ const MAX_DEPTH: usize = 16;
 const SCAN_INTERVAL: Duration = Duration::from_secs(3600);
 const CHECKSUM_CACHE_TIMEOUT: Duration = Duration::from_secs(1800);
 
-pub struct TableSyncer<F: TableSchema> {
-	pub table: Arc<Table<F>>,
-	pub todo: Mutex<SyncTodo>,
-	pub cache: Vec<Mutex<BTreeMap<SyncRange, RangeChecksum>>>,
+const TABLE_SYNC_RPC_TIMEOUT: Duration = Duration::from_secs(10);
+
+pub struct TableSyncer<F: TableSchema, R: TableReplication> {
+	table: Arc<Table<F, R>>,
+	todo: Mutex<SyncTodo>,
+	cache: Vec<Mutex<BTreeMap<SyncRange, RangeChecksum>>>,
 }
 
 #[derive(Serialize, Deserialize)]
@@ -36,21 +38,21 @@ pub enum SyncRPC {
 }
 
 pub struct SyncTodo {
-	pub todo: Vec<Partition>,
+	todo: Vec<TodoPartition>,
 }
 
 #[derive(Debug, Clone)]
-pub struct Partition {
-	pub begin: Hash,
-	pub end: Hash,
-	pub retain: bool,
+struct TodoPartition {
+	begin: Hash,
+	end: Hash,
+	retain: bool,
 }
 
 #[derive(Hash, PartialEq, Eq, Debug, Clone, Serialize, Deserialize)]
 pub struct SyncRange {
-	pub begin: Vec<u8>,
-	pub end: Vec<u8>,
-	pub level: usize,
+	begin: Vec<u8>,
+	end: Vec<u8>,
+	level: usize,
 }
 
 impl std::cmp::PartialOrd for SyncRange {
@@ -66,16 +68,20 @@ impl std::cmp::Ord for SyncRange {
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct RangeChecksum {
-	pub bounds: SyncRange,
-	pub children: Vec<(SyncRange, Hash)>,
-	pub found_limit: Option<Vec<u8>>,
+	bounds: SyncRange,
+	children: Vec<(SyncRange, Hash)>,
+	found_limit: Option<Vec<u8>>,
 
 	#[serde(skip, default = "std::time::Instant::now")]
-	pub time: Instant,
+	time: Instant,
 }
 
-impl<F: TableSchema + 'static> TableSyncer<F> {
-	pub async fn launch(table: Arc<Table<F>>) -> Arc<Self> {
+impl<F, R> TableSyncer<F, R>
+where
+	F: TableSchema + 'static,
+	R: TableReplication + 'static,
+{
+	pub async fn launch(table: Arc<Table<F, R>>) -> Arc<Self> {
 		let todo = SyncTodo { todo: Vec::new() };
 		let syncer = Arc::new(TableSyncer {
 			table: table.clone(),
@@ -166,7 +172,7 @@ impl<F: TableSchema + 'static> TableSyncer<F> {
 
 	async fn sync_partition(
 		self: Arc<Self>,
-		partition: &Partition,
+		partition: &TodoPartition,
 		must_exit: &mut watch::Receiver<bool>,
 	) -> Result<(), Error> {
 		eprintln!("({}) Preparing to sync {:?}...", self.table.name, partition);
@@ -175,8 +181,10 @@ impl<F: TableSchema + 'static> TableSyncer<F> {
 			.await?;
 
 		let my_id = self.table.system.id.clone();
-		let ring = self.table.system.ring.borrow().clone();
-		let nodes = ring.walk_ring(&partition.begin, self.table.param.replication_factor);
+		let nodes = self
+			.table
+			.replication
+			.write_nodes(&partition.begin, &self.table.system);
 		let mut sync_futures = nodes
 			.iter()
 			.filter(|node| **node != my_id)
@@ -349,7 +357,7 @@ impl<F: TableSchema + 'static> TableSyncer<F> {
 
 	async fn do_sync_with(
 		self: Arc<Self>,
-		partition: Partition,
+		partition: TodoPartition,
 		root_ck: RangeChecksum,
 		who: UUID,
 		retain: bool,
@@ -367,7 +375,7 @@ impl<F: TableSchema + 'static> TableSyncer<F> {
 					partition.begin.clone(),
 					partition.end.clone(),
 				)),
-				self.table.param.timeout,
+				TABLE_SYNC_RPC_TIMEOUT,
 			)
 			.await?;
 		if let TableRPC::<F>::SyncRPC(SyncRPC::RootChecksumRange(range)) = root_cks_resp {
@@ -398,7 +406,7 @@ impl<F: TableSchema + 'static> TableSyncer<F> {
 				.call(
 					&who,
 					&TableRPC::<F>::SyncRPC(SyncRPC::Checksums(step, retain)),
-					self.table.param.timeout,
+					TABLE_SYNC_RPC_TIMEOUT,
 				)
 				.await?;
 			if let TableRPC::<F>::SyncRPC(SyncRPC::Difference(mut diff_ranges, diff_items)) =
@@ -456,11 +464,7 @@ impl<F: TableSchema + 'static> TableSyncer<F> {
 		let rpc_resp = self
 			.table
 			.rpc_client
-			.call(
-				&who,
-				&TableRPC::<F>::Update(values),
-				self.table.param.timeout,
-			)
+			.call(&who, &TableRPC::<F>::Update(values), TABLE_SYNC_RPC_TIMEOUT)
 			.await?;
 		if let TableRPC::<F>::Ok = rpc_resp {
 			Ok(())
@@ -490,7 +494,7 @@ impl<F: TableSchema + 'static> TableSyncer<F> {
 		}
 	}
 
-	pub async fn handle_checksums_rpc(
+	async fn handle_checksums_rpc(
 		self: &Arc<Self>,
 		checksums: &[RangeChecksum],
 		retain: bool,
@@ -589,99 +593,80 @@ impl<F: TableSchema + 'static> TableSyncer<F> {
 }
 
 impl SyncTodo {
-	fn add_full_scan<F: TableSchema>(&mut self, table: &Table<F>) {
+	fn add_full_scan<F: TableSchema, R: TableReplication>(&mut self, table: &Table<F, R>) {
 		let my_id = table.system.id.clone();
 
 		self.todo.clear();
 
-		let ring: Arc<Ring> = table.system.ring.borrow().clone();
-
-		for i in 0..ring.ring.len() {
-			let nodes = ring.walk_ring_from_pos(i, table.param.replication_factor);
-			let begin = ring.ring[i].location.clone();
-
-			if i == 0 {
-				self.add_full_scan_aux(table, [0u8; 32].into(), begin.clone(), &nodes[..], &my_id);
+		let ring = table.system.ring.borrow().clone();
+		let split_points = table.replication.split_points(&ring);
+
+		for i in 0..split_points.len() - 1 {
+			let begin = split_points[i].clone();
+			let end = split_points[i + 1].clone();
+			let nodes = table.replication.write_nodes_from_ring(&begin, &ring);
+
+			let retain = nodes.contains(&my_id);
+			if !retain {
+				// Check if we have some data to send, otherwise skip
+				if table
+					.store
+					.range(begin.clone()..end.clone())
+					.next()
+					.is_none()
+				{
+					continue;
+				}
 			}
 
-			if i == ring.ring.len() - 1 {
-				self.add_full_scan_aux(table, begin, [0xffu8; 32].into(), &nodes[..], &my_id);
-			} else {
-				let end = ring.ring[i + 1].location.clone();
-				self.add_full_scan_aux(table, begin, end, &nodes[..], &my_id);
-			}
+			self.todo.push(TodoPartition { begin, end, retain });
 		}
 	}
 
-	fn add_full_scan_aux<F: TableSchema>(
+	fn add_ring_difference<F: TableSchema, R: TableReplication>(
 		&mut self,
-		table: &Table<F>,
-		begin: Hash,
-		end: Hash,
-		nodes: &[UUID],
-		my_id: &UUID,
+		table: &Table<F, R>,
+		old_ring: &Ring,
+		new_ring: &Ring,
 	) {
-		let retain = nodes.contains(my_id);
-		if !retain {
-			// Check if we have some data to send, otherwise skip
-			if table
-				.store
-				.range(begin.clone()..end.clone())
-				.next()
-				.is_none()
-			{
-				return;
-			}
-		}
-
-		self.todo.push(Partition { begin, end, retain });
-	}
-
-	fn add_ring_difference<F: TableSchema>(&mut self, table: &Table<F>, old: &Ring, new: &Ring) {
 		let my_id = table.system.id.clone();
 
-		let old_ring = ring_points(old);
-		let new_ring = ring_points(new);
-		let both_ring = old_ring.union(&new_ring).cloned().collect::<BTreeSet<_>>();
-
-		let prev_todo_begin = self
-			.todo
-			.iter()
-			.map(|x| x.begin.clone())
-			.collect::<BTreeSet<_>>();
-		let prev_todo_end = self
-			.todo
-			.iter()
-			.map(|x| x.end.clone())
-			.collect::<BTreeSet<_>>();
-		let prev_todo = prev_todo_begin
-			.union(&prev_todo_end)
-			.cloned()
-			.collect::<BTreeSet<_>>();
-
-		let all_points = both_ring.union(&prev_todo).cloned().collect::<Vec<_>>();
-
-		self.todo.sort_by(|x, y| x.begin.cmp(&y.begin));
+		let mut all_points = None
+			.into_iter()
+			.chain(table.replication.split_points(old_ring).drain(..))
+			.chain(table.replication.split_points(new_ring).drain(..))
+			.chain(self.todo.iter().map(|x| x.begin.clone()))
+			.chain(self.todo.iter().map(|x| x.end.clone()))
+			.collect::<Vec<_>>();
+		all_points.sort();
+		all_points.dedup();
+
+		let mut old_todo = std::mem::replace(&mut self.todo, vec![]);
+		old_todo.sort_by(|x, y| x.begin.cmp(&y.begin));
 		let mut new_todo = vec![];
+
 		for i in 0..all_points.len() - 1 {
 			let begin = all_points[i].clone();
 			let end = all_points[i + 1].clone();
-			let was_ours = old
-				.walk_ring(&begin, table.param.replication_factor)
+			let was_ours = table
+				.replication
+				.write_nodes_from_ring(&begin, &old_ring)
 				.contains(&my_id);
-			let is_ours = new
-				.walk_ring(&begin, table.param.replication_factor)
+			let is_ours = table
+				.replication
+				.write_nodes_from_ring(&begin, &new_ring)
 				.contains(&my_id);
-			let was_todo = match self.todo.binary_search_by(|x| x.begin.cmp(&begin)) {
+
+			let was_todo = match old_todo.binary_search_by(|x| x.begin.cmp(&begin)) {
 				Ok(_) => true,
 				Err(j) => {
-					(j > 0 && self.todo[j - 1].begin < end && begin < self.todo[j - 1].end)
-						|| (j < self.todo.len()
-							&& self.todo[j].begin < end && begin < self.todo[j].end)
+					(j > 0 && old_todo[j - 1].begin < end && begin < old_todo[j - 1].end)
+						|| (j < old_todo.len()
+							&& old_todo[j].begin < end && begin < old_todo[j].end)
 				}
 			};
 			if was_todo || (is_ours && !was_ours) || (was_ours && !is_ours) {
-				new_todo.push(Partition {
+				new_todo.push(TodoPartition {
 					begin,
 					end,
 					retain: is_ours,
@@ -692,7 +677,7 @@ impl SyncTodo {
 		self.todo = new_todo;
 	}
 
-	fn pop_task(&mut self) -> Option<Partition> {
+	fn pop_task(&mut self) -> Option<TodoPartition> {
 		if self.todo.is_empty() {
 			return None;
 		}
@@ -707,13 +692,3 @@ impl SyncTodo {
 		}
 	}
 }
-
-fn ring_points(ring: &Ring) -> BTreeSet<Hash> {
-	let mut ret = BTreeSet::new();
-	ret.insert([0u8; 32].into());
-	ret.insert([0xFFu8; 32].into());
-	for i in 0..ring.ring.len() {
-		ret.insert(ring.ring[i].location.clone());
-	}
-	ret
-}
diff --git a/src/version_table.rs b/src/version_table.rs
index 7e7623fc..24109981 100644
--- a/src/version_table.rs
+++ b/src/version_table.rs
@@ -5,6 +5,7 @@ use std::sync::Arc;
 use crate::background::BackgroundRunner;
 use crate::data::*;
 use crate::table::*;
+use crate::table_sharded::*;
 
 use crate::block_ref_table::*;
 
@@ -56,7 +57,7 @@ impl Entry<Hash, EmptySortKey> for Version {
 
 pub struct VersionTable {
 	pub background: Arc<BackgroundRunner>,
-	pub block_ref_table: Arc<Table<BlockRefTable>>,
+	pub block_ref_table: Arc<Table<BlockRefTable, TableShardedReplication>>,
 }
 
 #[async_trait]