492: New Database Implementation

* new database trait with new data manager

* using new data manager in app client

* removing unused functions

* data manager uses more tightly defined database

* changed generic type

* using new function from data manager to store app data

* trying implementation without generics

* wrapping errors

* using new data manager functions in sync client

* added fns for handling confidence data

* sync client completely uses new data manager

* light client uses new data manager

* added new functions for sync checkpoint storage

* sync finality uses new data manager

* renamed generics

* using data manager for subscriptions

* fat client uses data manager

* properly initializing rpc with data manager

* data manager is used in v1 api handlers

* using data manager in v2 api handlers

* v2 routes are using new data manager

* mocking for unit tests with data manager

* v1 routes are capable of using data manager

* fixed references and unused params

* fixed lifetime

* renamed generics and methods

* fixed dropped binding

* removed unused code

* api compat tests now use data manager

* initializing data manager and passing it where needed

* clippy fix

* Add alternative database implementation into db mod.

* partially solved trait bounds

* Fix trait bounds in build.

* Specialize DB trait to use Key enum.

* Switch to serde instead of custom decode/encode.

* Use scale for rocks DB, remove obsolete modules.

* light client implemented with new db trait

* properly interfaced finality client to use new db

* added new boundary

* renames

* enabled mocks and tests

* re-enabled unit tests

* moved rocks keys into same mod

* mocking client for unit tests

* new instantiation for reworked clients

* Fix clippy warnings.

* Remove DB mocking from light_client.

* Remove DB mocking from fat client.

* Remove generics from handlers.

* using db implementation directly

* renamed key

* key left out

* lock update

* import fix

---------

Co-authored-by: Aleksandar Terentić <[email protected]>

src/api/server.rs

@@ -9,6 +9,7 @@
 //! * `/v1/appdata/{block_number}` - returns decoded extrinsic data for configured app_id and given block number
 
 use crate::api::v2;
+use crate::data::Database;
 use crate::shutdown::Controller;
 use crate::types::IdentityConfig;
 use crate::{
@@ -18,7 +19,6 @@ use crate::{
 };
 use color_eyre::eyre::WrapErr;
 use futures::{Future, FutureExt};
-use rocksdb::DB;
 use std::{
 	net::SocketAddr,
 	str::FromStr,
@@ -27,8 +27,8 @@ use std::{
 use tracing::info;
 use warp::{Filter, Reply};
 
-pub struct Server {
-	pub db: Arc<DB>,
+pub struct Server<T: Database> {
+	pub db: T,
 	pub cfg: RuntimeConfig,
 	pub identity_cfg: IdentityConfig,
 	pub state: Arc<Mutex<State>>,
@@ -46,7 +46,7 @@ fn health_route() -> impl Filter<Extract = impl Reply, Error = warp::Rejection>
 		.map(|_| warp::reply::with_status("", warp::http::StatusCode::OK))
 }
 
-impl Server {
+impl<T: Database + Clone + Send + Sync + 'static> Server<T> {
 	/// Creates a HTTP server that needs to be spawned into a runtime
 	pub fn bind(self) -> impl Future<Output = ()> {
 		let RuntimeConfig {
@@ -65,7 +65,7 @@ impl Server {
 			self.identity_cfg,
 			self.node_client.clone(),
 			self.ws_clients.clone(),
-			crate::data::RocksDB(self.db.clone()),
+			self.db.clone(),
 		);
 
 		let cors = warp::cors()

src/api/v1/handlers.rs

@@ -1,7 +1,7 @@
 use super::types::{AppDataQuery, ClientResponse, ConfidenceResponse, LatestBlockResponse, Status};
 use crate::{
 	api::v1::types::{Extrinsics, ExtrinsicsDataResponse},
-	data::{get_confidence_from_db, get_decoded_data_from_db},
+	data::{Database, Key},
 	types::{Mode, OptionBlockRange, State},
 	utils::calculate_confidence,
 };
@@ -13,7 +13,6 @@ use base64::{engine::general_purpose, Engine};
 use codec::Decode;
 use color_eyre::{eyre::WrapErr, Result};
 use num::{BigUint, FromPrimitive};
-use rocksdb::DB;
 use std::sync::{Arc, Mutex};
 use tracing::{debug, info};
 
@@ -30,11 +29,11 @@ pub fn mode(app_id: Option<u32>) -> ClientResponse<Mode> {
 
 pub fn confidence(
 	block_num: u32,
-	db: Arc<DB>,
+	db: impl Database,
 	state: Arc<Mutex<State>>,
 ) -> ClientResponse<ConfidenceResponse> {
 	info!("Got request for confidence for block {block_num}");
-	let res = match get_confidence_from_db(db, block_num) {
+	let res = match db.get(Key::VerifiedCellCount(block_num)) {
 		Ok(Some(count)) => {
 			let confidence = calculate_confidence(count);
 			let serialised_confidence = serialised_confidence(block_num, confidence);
@@ -66,13 +65,13 @@ pub fn confidence(
 pub fn status(
 	app_id: Option<u32>,
 	state: Arc<Mutex<State>>,
-	db: Arc<DB>,
+	db: impl Database,
 ) -> ClientResponse<Status> {
 	let state = state.lock().unwrap();
 	let Some(last) = state.confidence_achieved.last() else {
 		return ClientResponse::NotFound;
 	};
-	let res = match get_confidence_from_db(db, last) {
+	let res = match db.get(Key::VerifiedCellCount(last)) {
 		Ok(Some(count)) => {
 			let confidence = calculate_confidence(count);
 			ClientResponse::Normal(Status {
@@ -101,7 +100,7 @@ pub fn latest_block(state: Arc<Mutex<State>>) -> ClientResponse<LatestBlockRespo
 pub fn appdata(
 	block_num: u32,
 	query: AppDataQuery,
-	db: Arc<DB>,
+	db: impl Database,
 	app_id: Option<u32>,
 	state: Arc<Mutex<State>>,
 ) -> ClientResponse<ExtrinsicsDataResponse> {
@@ -130,11 +129,9 @@ pub fn appdata(
 	let state = state.lock().unwrap();
 	let last = state.confidence_achieved.last();
 	let decode = query.decode.unwrap_or(false);
-	let res = match decode_app_data_to_extrinsics(get_decoded_data_from_db(
-		db,
-		app_id.unwrap_or(0u32),
-		block_num,
-	)) {
+	let res = match decode_app_data_to_extrinsics(
+		db.get(Key::AppData(app_id.unwrap_or(0u32), block_num)),
+	) {
 		Ok(Some(data)) => {
 			if !decode {
 				ClientResponse::Normal(ExtrinsicsDataResponse {

src/api/v1/mod.rs

@@ -1,7 +1,6 @@
-use crate::types::State;
+use crate::{data::Database, types::State};
 
 use self::types::AppDataQuery;
-use rocksdb::DB;
 use std::{
 	convert::Infallible,
 	sync::{Arc, Mutex},
@@ -17,7 +16,9 @@ fn with_state(
 	warp::any().map(move || state.clone())
 }
 
-fn with_db(db: Arc<DB>) -> impl Filter<Extract = (Arc<DB>,), Error = Infallible> + Clone {
+fn with_db<T: Database + Clone + Send>(
+	db: T,
+) -> impl Filter<Extract = (T,), Error = Infallible> + Clone {
 	warp::any().map(move || db.clone())
 }
 
@@ -28,7 +29,7 @@ fn with_app_id(
 }
 
 pub fn routes(
-	db: Arc<DB>,
+	db: impl Database + Clone + Send,
 	app_id: Option<u32>,
 	state: Arc<Mutex<State>>,
 ) -> impl Filter<Extract = (impl Reply,), Error = Rejection> + Clone {

src/api/v2/handlers.rs

@@ -10,9 +10,11 @@ use super::{
 use crate::{
 	api::v2::types::{ErrorCode, InternalServerError},
 	data::Database,
+	data::Key,
 	types::{RuntimeConfig, State},
 	utils::calculate_confidence,
 };
+use avail_subxt::primitives;
 use color_eyre::{eyre::eyre, Result};
 use hyper::StatusCode;
 use std::{
@@ -100,7 +102,7 @@ pub async fn block(
 	};
 
 	let confidence = db
-		.get_confidence(block_number)
+		.get(Key::VerifiedCellCount(block_number))
 		.map_err(Error::internal_server_error)?
 		.map(calculate_confidence);
 
@@ -126,7 +128,7 @@ pub async fn block_header(
 		return Err(Error::bad_request_unknown("Block header is not available"));
 	};
 
-	db.get_header(block_number)
+	db.get::<primitives::Header>(Key::BlockHeader(block_number))
 		.and_then(|header| header.ok_or_else(|| eyre!("Header not found")))
 		.and_then(|header| header.try_into())
 		.map_err(Error::internal_server_error)
@@ -154,7 +156,7 @@ pub async fn block_data(
 	};
 
 	let data = db
-		.get_data(app_id, block_number)
+		.get::<Vec<Vec<u8>>>(Key::AppData(app_id, block_number))
 		.map_err(Error::internal_server_error)?;
 
 	let Some(data) = data else {