switching to rocksdb as storage backend

Niko PLP 10 months ago
parent eac2c26a87
commit bf157b25fc
  1. 155
      Cargo.lock
  2. 1
      Cargo.toml
  3. 2
      ng-app/src/App.svelte
  4. 8
      ng-app/src/lib/Home.svelte
  5. 198
      ng-app/src/lib/Install.svelte
  6. 3
      ng-app/src/routes/Home.svelte
  7. 205
      ng-app/src/routes/Install.svelte
  8. 1
      ngone/.gitignore
  9. 2
      ngone/Cargo.toml
  10. BIN
      ngone/data/data.mdb
  11. BIN
      ngone/data/lock.mdb
  12. 10
      ngone/src/main.rs
  13. 2
      ngone/web/src/App.svelte
  14. 28
      ngone/web/src/routes/Install.svelte
  15. 83
      p2p-broker/src/broker_store/account.rs
  16. 2
      p2p-broker/src/broker_store/invitation.rs
  17. 6
      p2p-repo/src/kcv_store.rs
  18. 1
      p2p-repo/src/store.rs
  19. 12
      stores-lmdb/src/kcv_store.rs
  20. 20
      stores-rocksdb/Cargo.toml
  21. 424
      stores-rocksdb/src/kcv_store.rs
  22. 5
      stores-rocksdb/src/lib.rs
  23. 992
      stores-rocksdb/src/repo_store.rs

155
Cargo.lock generated

@ -477,6 +477,27 @@ dependencies = [
"serde",
]
[[package]]
name = "bindgen"
version = "0.65.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cfdf7b466f9a4903edc73f95d6d2bcd5baf8ae620638762244d3f60143643cc5"
dependencies = [
"bitflags",
"cexpr",
"clang-sys",
"lazy_static",
"lazycell",
"peeking_take_while",
"prettyplease",
"proc-macro2",
"quote",
"regex",
"rustc-hash",
"shlex",
"syn 2.0.18",
]
[[package]]
name = "bit_field"
version = "0.10.2"
@ -621,6 +642,17 @@ dependencies = [
"serde",
]
[[package]]
name = "bzip2-sys"
version = "0.1.11+1.0.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "736a955f3fa7875102d57c82b8cac37ec45224a07fd32d58f9f7a186b6cd4cdc"
dependencies = [
"cc",
"libc",
"pkg-config",
]
[[package]]
name = "cairo-rs"
version = "0.16.7"
@ -661,6 +693,9 @@ name = "cc"
version = "1.0.79"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "50d30906286121d95be3d479533b458f87493b30a4b5f79a607db8f5d11aa91f"
dependencies = [
"jobserver",
]
[[package]]
name = "cesu8"
@ -668,6 +703,15 @@ version = "1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6d43a04d8753f35258c91f8ec639f792891f748a1edbd759cf1dcea3382ad83c"
[[package]]
name = "cexpr"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6fac387a98bb7c37292057cffc56d62ecb629900026402633ae9160df93a8766"
dependencies = [
"nom",
]
[[package]]
name = "cfb"
version = "0.7.3"
@ -746,6 +790,17 @@ dependencies = [
"zeroize",
]
[[package]]
name = "clang-sys"
version = "1.6.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c688fc74432808e3eb684cae8830a86be1d66a2bd58e1f248ed0960a590baf6f"
dependencies = [
"glob",
"libc",
"libloading",
]
[[package]]
name = "clap"
version = "4.3.5"
@ -2370,6 +2425,15 @@ version = "0.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8eaf4bc02d17cbdd7ff4c7438cafcdf7fb9a4613313ad11b4f8fefe7d3fa0130"
[[package]]
name = "jobserver"
version = "0.1.26"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "936cfd212a0155903bcbc060e316fb6cc7cbf2e1907329391ebadc1fe0ce77c2"
dependencies = [
"libc",
]
[[package]]
name = "jpeg-decoder"
version = "0.3.0"
@ -2427,6 +2491,12 @@ version = "1.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e2abad23fbc42b3700f2f279844dc832adb2b2eb069b2df918f455c4e18cc646"
[[package]]
name = "lazycell"
version = "1.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "830d08ce1d1d941e6b30645f1a0eb5643013d835ce3779a5fc208261dbe10f55"
[[package]]
name = "lebe"
version = "0.5.2"
@ -2439,12 +2509,46 @@ version = "0.2.146"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f92be4933c13fd498862a9e02a3055f8a8d9c039ce33db97306fd5a6caa7f29b"
[[package]]
name = "libloading"
version = "0.7.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b67380fd3b2fbe7527a606e18729d21c6f3951633d0500574c4dc22d2d638b9f"
dependencies = [
"cfg-if",
"winapi",
]
[[package]]
name = "libm"
version = "0.1.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7fc7aa29613bd6a620df431842069224d8bc9011086b1db4c0e0cd47fa03ec9a"
[[package]]
name = "librocksdb-sys"
version = "0.11.0+8.3.2"
source = "git+https://git.nextgraph.org/NextGraph/rust-rocksdb.git?branch=master#520e09b1ada58e90075d0d462876afd3b3f9f6a9"
dependencies = [
"bindgen",
"bzip2-sys",
"cc",
"glob",
"libc",
"libz-sys",
]
[[package]]
name = "libz-sys"
version = "1.1.12"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d97137b25e321a73eef1418d1d5d2eda4d77e12813f8e6dead84bc52c5870a7b"
dependencies = [
"cc",
"pkg-config",
"vcpkg",
]
[[package]]
name = "line-wrap"
version = "0.1.1"
@ -2912,7 +3016,7 @@ dependencies = [
"serde_bytes",
"serde_json",
"slice_as_array",
"stores-lmdb",
"stores-rocksdb",
"tokio",
"warp",
"warp-embed",
@ -3339,6 +3443,12 @@ version = "1.0.12"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9f746c4065a8fa3fe23974dd82f15431cc8d40779821001404d10d2e79ca7d79"
[[package]]
name = "peeking_take_while"
version = "0.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "19b17cddbe7ec3f8bc800887bab5e717348c95ea2ca0b1bf0837fb964dc67099"
[[package]]
name = "percent-encoding"
version = "2.3.0"
@ -3603,6 +3713,16 @@ dependencies = [
"termtree",
]
[[package]]
name = "prettyplease"
version = "0.2.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9825a04601d60621feed79c4e6b56d65db77cdca55cef43b46b0de1096d1c282"
dependencies = [
"proc-macro2",
"syn 2.0.18",
]
[[package]]
name = "proc-macro-crate"
version = "1.3.1"
@ -3925,6 +4045,15 @@ dependencies = [
"uuid 0.8.2",
]
[[package]]
name = "rocksdb"
version = "0.21.0"
source = "git+https://git.nextgraph.org/NextGraph/rust-rocksdb.git?branch=master#520e09b1ada58e90075d0d462876afd3b3f9f6a9"
dependencies = [
"libc",
"librocksdb-sys",
]
[[package]]
name = "rust-embed"
version = "6.7.0"
@ -3970,6 +4099,12 @@ dependencies = [
"num-traits",
]
[[package]]
name = "rustc-hash"
version = "1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "08d43f7aa6b08d49f382cde6a7982047c3426db949b1424bc4b7ec9ae12c6ce2"
[[package]]
name = "rustc_version"
version = "0.4.0"
@ -4328,6 +4463,12 @@ dependencies = [
"lazy_static",
]
[[package]]
name = "shlex"
version = "1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "43b2853a4d09f215c24cc5489c992ce46052d359b5109343cbafbf26bc62f8a3"
[[package]]
name = "signal-hook"
version = "0.3.15"
@ -4460,6 +4601,18 @@ dependencies = [
"tempfile",
]
[[package]]
name = "stores-rocksdb"
version = "0.1.0"
dependencies = [
"hex",
"p2p-repo",
"rocksdb",
"serde",
"serde_bare",
"tempfile",
]
[[package]]
name = "string_cache"
version = "0.8.7"

@ -6,6 +6,7 @@ members = [
"p2p-client-ws",
"p2p-verifier",
"stores-lmdb",
"stores-rocksdb",
"ngcli",
"ngd",
"ngone",

@ -29,7 +29,7 @@
import WalletCreate from "./routes/WalletCreate.svelte";
import WalletLogin from "./routes/WalletLogin.svelte";
import UserRegistered from "./routes/UserRegistered.svelte";
import Install from "./routes/Install.svelte";
import Install from "./lib/Install.svelte";
import ng from "./api";

@ -14,14 +14,12 @@
import { link } from "svelte-spa-router";
// @ts-ignore
import Logo from "../assets/nextgraph.svg?component";
import Test from "./Test.svelte";
import { has_wallets, active_wallet } from "../store";
import { onMount } from "svelte";
export let display_login_create = false;
</script>
{#if !$has_wallets || !$active_wallet || display_login_create}
{#if display_login_create}
<main class="container3">
<div class="row">
<Logo class="logo block h-40" alt="NextGraph Logo" />
@ -86,8 +84,6 @@
{:else}
<main class="container3">
<h1>Welcome to test</h1>
<div class="row">
<Test />
</div>
<div class="row" />
</main>
{/if}

File diff suppressed because one or more lines are too long

@ -24,6 +24,7 @@
derived,
} from "../store";
let display_login_create = !$has_wallets || !$active_wallet;
let unsubscribe;
onMount(() => {
const combined = derived([active_wallet, has_wallets], ([$s1, $s2]) => [
@ -43,4 +44,4 @@
});
</script>
<Home />
<Home {display_login_create} />

File diff suppressed because one or more lines are too long

1
ngone/.gitignore vendored

@ -0,0 +1 @@
data

@ -14,7 +14,7 @@ warp-embed = "0.4"
rust-embed = "6"
log = "0.4"
env_logger = "0.10"
stores-lmdb = { path = "../stores-lmdb" }
stores-rocksdb = { path = "../stores-rocksdb" }
p2p-repo = { path = "../p2p-repo", features = ["server_log_output"] }
p2p-net = { path = "../p2p-net" }
ng-wallet = { path = "../ng-wallet" }

Binary file not shown.

Binary file not shown.

@ -29,14 +29,14 @@ use p2p_net::types::{APP_NG_ONE_URL, NG_ONE_URL};
use p2p_repo::log::*;
use p2p_repo::types::*;
use p2p_repo::utils::{generate_keypair, sign, verify};
use stores_lmdb::kcv_store::LmdbKCVStore;
use stores_rocksdb::kcv_store::RocksdbKCVStore;
#[derive(RustEmbed)]
#[folder = "web/dist"]
struct Static;
struct Server {
store: LmdbKCVStore,
store: RocksdbKCVStore,
}
impl Server {
@ -158,7 +158,7 @@ async fn main() {
let key: [u8; 32] = [0; 32];
log_debug!("data directory: {}", dir.to_str().unwrap());
fs::create_dir_all(dir.clone()).unwrap();
let store = LmdbKCVStore::open(&dir, key);
let store = RocksdbKCVStore::open(&dir, key);
if store.is_err() {
return;
}
@ -217,8 +217,8 @@ async fn main() {
log_debug!("CORS: any origin");
cors = cors.allow_any_origin();
}
log::info!("Starting server on http://localhost:3030");
log::info!("Starting server on http://localhost:3032");
warp::serve(api_v1.or(static_files).with(cors))
.run(([127, 0, 0, 1], 3030))
.run(([127, 0, 0, 1], 3032))
.await;
}

@ -14,7 +14,7 @@
import Home from "./routes/Home.svelte";
import WalletCreate from "./routes/WalletCreate.svelte";
import Install from "../../../ng-app/src/routes/Install.svelte";
import Install from "./routes/Install.svelte";
//import URI from "./routes/URI.svelte";
import NotFound from "./routes/NotFound.svelte";

@ -0,0 +1,28 @@
<!--
// Copyright (c) 2022-2023 Niko Bonnieure, Par le Peuple, NextGraph.org developers
// All rights reserved.
// Licensed under the Apache License, Version 2.0
// <LICENSE-APACHE2 or http://www.apache.org/licenses/LICENSE-2.0>
// or the MIT license <LICENSE-MIT or http://opensource.org/licenses/MIT>,
// at your option. All files in the project carrying such
// notice may not be copied, modified, or distributed except
// according to those terms.
-->
<script type="ts">
import { Button } from "flowbite-svelte";
import { link } from "svelte-spa-router";
import Install from "../../../../ng-app/src/lib/Install.svelte";
import { push } from "svelte-spa-router";
import { onMount, onDestroy } from "svelte";
let display_has_wallets_warning = false;
let unsubscribe;
onMount(() => {});
onDestroy(() => {
unsubscribe();
});
</script>
<Install {display_has_wallets_warning} />

@ -27,24 +27,16 @@ pub struct Account<'a> {
}
impl<'a> Account<'a> {
const PREFIX: u8 = b"u"[0];
const PREFIX_CLIENT: u8 = b"d"[0];
// propertie's suffixes
const CLIENT: u8 = b"c"[0];
const ADMIN: u8 = b"a"[0];
//const OVERLAY: u8 = b"o"[0];
const PREFIX_ACCOUNT: u8 = b"a"[0];
const PREFIX_CLIENT: u8 = b"c"[0];
const PREFIX_CLIENT_PROPERTY: u8 = b"d"[0];
// propertie's client suffixes
const INFO: u8 = b"i"[0];
const LAST_SEEN: u8 = b"l"[0];
const ALL_PROPERTIES: [u8; 2] = [Self::CLIENT, Self::ADMIN];
const ALL_CLIENT_PROPERTIES: [u8; 2] = [Self::INFO, Self::LAST_SEEN];
const SUFFIX_FOR_EXIST_CHECK: u8 = Self::ADMIN;
pub fn open(id: &UserId, store: &'a dyn KCVStore) -> Result<Account<'a>, StorageError> {
let opening = Account {
id: id.clone(),
@ -67,12 +59,7 @@ impl<'a> Account<'a> {
if acc.exists() {
return Err(StorageError::AlreadyExists);
}
store.put(
Self::PREFIX,
&to_vec(&id)?,
Some(Self::ADMIN),
to_vec(&admin)?,
)?;
store.put(Self::PREFIX_ACCOUNT, &to_vec(&id)?, None, to_vec(&admin)?)?;
Ok(acc)
}
pub fn get_all_users(
@ -81,7 +68,7 @@ impl<'a> Account<'a> {
) -> Result<Vec<UserId>, StorageError> {
let size = to_vec(&UserId::nil())?.len();
let mut res: Vec<UserId> = vec![];
for user in store.get_all_keys_and_values(Self::PREFIX, size, Some(Self::ADMIN))? {
for user in store.get_all_keys_and_values(Self::PREFIX_ACCOUNT, size, vec![], None)? {
let admin: bool = from_slice(&user.1)?;
if admin == admins {
let id: UserId = from_slice(&user.0[1..user.0.len() - 1])?;
@ -92,11 +79,7 @@ impl<'a> Account<'a> {
}
pub fn exists(&self) -> bool {
self.store
.get(
Self::PREFIX,
&to_vec(&self.id).unwrap(),
Some(Self::SUFFIX_FOR_EXIST_CHECK),
)
.get(Self::PREFIX_ACCOUNT, &to_vec(&self.id).unwrap(), None)
.is_ok()
}
pub fn id(&self) -> UserId {
@ -112,39 +95,31 @@ impl<'a> Account<'a> {
let hash = s.finish();
let client_key = (client.clone(), hash);
let client_key_ser = to_vec(&client_key)?;
let mut client_key_ser = to_vec(&client_key)?;
let info_ser = to_vec(info)?;
self.store.write_transaction(&mut |tx| {
let mut id_and_client = to_vec(&self.id)?;
id_and_client.append(&mut client_key_ser);
if tx
.has_property_value(
Self::PREFIX,
&to_vec(&self.id)?,
Some(Self::CLIENT),
&client_key_ser,
)
.has_property_value(Self::PREFIX_CLIENT, &id_and_client, None, &vec![])
.is_err()
{
tx.put(
Self::PREFIX,
&to_vec(&self.id)?,
Some(Self::CLIENT),
&client_key_ser,
)?;
tx.put(Self::PREFIX_CLIENT, &id_and_client, None, &vec![])?;
}
if tx
.has_property_value(
Self::PREFIX_CLIENT,
&client_key_ser,
Self::PREFIX_CLIENT_PROPERTY,
&id_and_client,
Some(Self::INFO),
&info_ser,
)
.is_err()
{
tx.put(
Self::PREFIX_CLIENT,
&client_key_ser,
Self::PREFIX_CLIENT_PROPERTY,
&id_and_client,
Some(Self::INFO),
&info_ser,
)?;
@ -154,8 +129,8 @@ impl<'a> Account<'a> {
.unwrap()
.as_secs();
tx.replace(
Self::PREFIX_CLIENT,
&client_key_ser,
Self::PREFIX_CLIENT_PROPERTY,
&id_and_client,
Some(Self::LAST_SEEN),
&to_vec(&now)?,
)?;
@ -205,9 +180,9 @@ impl<'a> Account<'a> {
if self
.store
.has_property_value(
Self::PREFIX,
Self::PREFIX_ACCOUNT,
&to_vec(&self.id)?,
Some(Self::ADMIN),
None,
&to_vec(&true)?,
)
.is_ok()
@ -219,12 +194,26 @@ impl<'a> Account<'a> {
pub fn del(&self) -> Result<(), StorageError> {
self.store.write_transaction(&mut |tx| {
if let Ok(clients) = tx.get_all(Self::PREFIX, &to_vec(&self.id)?, Some(Self::CLIENT)) {
let id = to_vec(&self.id)?;
// let mut id_and_client = to_vec(&self.id)?;
// let client_key = (client.clone(), hash);
// let mut client_key_ser = to_vec(&client_key)?;
let client_key = (ClientId::nil(), 0u64);
let mut client_key_ser = to_vec(&client_key)?;
let size = client_key_ser.len() + id.len();
if let Ok(clients) = tx.get_all_keys_and_values(Self::PREFIX_CLIENT, size, id, None) {
for client in clients {
tx.del_all(Self::PREFIX_CLIENT, &client, &Self::ALL_CLIENT_PROPERTIES)?;
tx.del(Self::PREFIX_CLIENT, &client.0, None)?;
tx.del_all(
Self::PREFIX_CLIENT_PROPERTY,
&client.0,
&Self::ALL_CLIENT_PROPERTIES,
)?;
}
}
tx.del_all(Self::PREFIX, &to_vec(&self.id)?, &Self::ALL_PROPERTIES)?;
tx.del(Self::PREFIX_ACCOUNT, &to_vec(&self.id)?, None)?;
Ok(())
})
}

@ -94,7 +94,7 @@ impl<'a> Invitation<'a> {
unique = true;
multi = true;
}
for invite in store.get_all_keys_and_values(Self::PREFIX, size, None)? {
for invite in store.get_all_keys_and_values(Self::PREFIX, size, vec![], None)? {
if invite.0.len() == size + 2 {
let code: [u8; 32] = from_slice(&invite.0[1..invite.0.len() - 1])?;
if invite.0[size + 1] == Self::TYPE {

@ -8,6 +8,8 @@
use crate::store::StorageError;
// TODO:remove mut on self for trait WriteTransaction methods
pub trait WriteTransaction: ReadTransaction {
/// Save a property value to the store.
fn put(
@ -53,6 +55,9 @@ pub trait ReadTransaction {
fn get(&self, prefix: u8, key: &Vec<u8>, suffix: Option<u8>) -> Result<Vec<u8>, StorageError>;
/// Load all the values of a property from the store.
#[deprecated(
note = "KVStore has unique values (since switch from lmdb to rocksdb) use get() instead"
)]
fn get_all(
&self,
prefix: u8,
@ -74,6 +79,7 @@ pub trait ReadTransaction {
&self,
prefix: u8,
key_size: usize,
key_prefix: Vec<u8>,
suffix: Option<u8>,
) -> Result<Vec<(Vec<u8>, Vec<u8>)>, StorageError>;
}

@ -38,6 +38,7 @@ pub trait RepoStore {
pub enum StorageError {
NotFound,
InvalidValue,
DifferentValue,
BackendError,
SerializationError,
AlreadyExists,

@ -47,6 +47,7 @@ impl<'a> ReadTransaction for LmdbTransaction<'a> {
&self,
prefix: u8,
key_size: usize,
key_prefix: Vec<u8>,
suffix: Option<u8>,
) -> Result<Vec<(Vec<u8>, Vec<u8>)>, StorageError> {
unimplemented!();
@ -253,10 +254,17 @@ impl ReadTransaction for LmdbKCVStore {
&self,
prefix: u8,
key_size: usize,
key_prefix: Vec<u8>,
suffix: Option<u8>,
) -> Result<Vec<(Vec<u8>, Vec<u8>)>, StorageError> {
let vec_key_start = vec![0u8; key_size];
let vec_key_end = vec![255u8; key_size];
let mut vec_key_start = key_prefix.clone();
let mut trailing_zeros = vec![0u8; key_size - key_prefix.len()];
vec_key_start.append(&mut trailing_zeros);
let mut vec_key_end = key_prefix.clone();
let mut trailing_max = vec![255u8; key_size - key_prefix.len()];
vec_key_end.append(&mut trailing_max);
let property_start = Self::compute_property(prefix, &vec_key_start, suffix);
let property_end =
Self::compute_property(prefix, &vec_key_end, Some(suffix.unwrap_or(255u8)));

@ -0,0 +1,20 @@
[package]
name = "stores-rocksdb"
version = "0.1.0"
edition = "2021"
license = "MIT/Apache-2.0"
authors = ["Niko PLP <niko@nextgraph.org>"]
description = "P2P stores based on LMDB for NextGraph"
repository = "https://git.nextgraph.org/NextGraph/nextgraph-rs"
[dependencies]
p2p-repo = { path = "../p2p-repo" }
serde = { version = "1.0.142", features = ["derive"] }
serde_bare = "0.5.0"
tempfile = "3"
hex = "0.4.3"
[target.'cfg(not(target_arch = "wasm32"))'.dependencies.rocksdb]
git = "https://git.nextgraph.org/NextGraph/rust-rocksdb.git"
branch = "master"
features = [ ]

@ -0,0 +1,424 @@
// Copyright (c) 2022-2023 Niko Bonnieure, Par le Peuple, NextGraph.org developers
// All rights reserved.
// Licensed under the Apache License, Version 2.0
// <LICENSE-APACHE2 or http://www.apache.org/licenses/LICENSE-2.0>
// or the MIT license <LICENSE-MIT or http://opensource.org/licenses/MIT>,
// at your option. All files in the project carrying such
// notice may not be copied, modified, or distributed except
// according to those terms.
use p2p_repo::kcv_store::*;
use p2p_repo::store::*;
use p2p_repo::types::*;
use p2p_repo::utils::*;
use p2p_repo::log::*;
use std::path::Path;
use std::path::PathBuf;
use std::sync::RwLockReadGuard;
use std::sync::{Arc, RwLock};
use serde::{Deserialize, Serialize};
use serde_bare::error::Error;
use rocksdb::{
ColumnFamilyDescriptor, Direction, Env, ErrorKind, IteratorMode, Options, SingleThreaded,
TransactionDB, TransactionDBOptions, DB,
};
pub struct RocksdbTransaction<'a> {
store: &'a RocksdbKCVStore,
tx: Option<rocksdb::Transaction<'a, TransactionDB>>,
}
impl<'a> RocksdbTransaction<'a> {
fn commit(&mut self) {
self.tx.take().unwrap().commit().unwrap();
}
fn tx(&self) -> &rocksdb::Transaction<'a, TransactionDB> {
self.tx.as_ref().unwrap()
}
}
impl<'a> ReadTransaction for RocksdbTransaction<'a> {
fn get_all_keys_and_values(
&self,
prefix: u8,
key_size: usize,
key_prefix: Vec<u8>,
suffix: Option<u8>,
) -> Result<Vec<(Vec<u8>, Vec<u8>)>, StorageError> {
self.store
.get_all_keys_and_values(prefix, key_size, key_prefix, suffix)
}
/// Load a single value property from the store.
fn get(&self, prefix: u8, key: &Vec<u8>, suffix: Option<u8>) -> Result<Vec<u8>, StorageError> {
let property = RocksdbKCVStore::compute_property(prefix, key, suffix);
let mut res = self
.tx()
.get_for_update(property, true)
.map_err(|e| StorageError::BackendError)?;
match res {
Some(val) => Ok(val),
None => Err(StorageError::NotFound),
}
}
/// Load all the values of a property from the store.
fn get_all(
&self,
prefix: u8,
key: &Vec<u8>,
suffix: Option<u8>,
) -> Result<Vec<Vec<u8>>, StorageError> {
unimplemented!();
}
/// Check if a specific value exists for a property from the store.
fn has_property_value(
&self,
prefix: u8,
key: &Vec<u8>,
suffix: Option<u8>,
value: &Vec<u8>,
) -> Result<(), StorageError> {
let property = RocksdbKCVStore::compute_property(prefix, key, suffix);
let exists = self
.tx()
.get_for_update(property, true)
.map_err(|e| StorageError::BackendError)?;
match exists {
Some(stored_value) => {
if stored_value.eq(value) {
Ok(())
} else {
Err(StorageError::DifferentValue)
}
}
None => Err(StorageError::NotFound),
}
}
}
impl<'a> WriteTransaction for RocksdbTransaction<'a> {
/// Save a property value to the store.
fn put(
&mut self,
prefix: u8,
key: &Vec<u8>,
suffix: Option<u8>,
value: &Vec<u8>,
) -> Result<(), StorageError> {
let property = RocksdbKCVStore::compute_property(prefix, key, suffix);
self.tx()
.put(property, value)
.map_err(|e| StorageError::BackendError)?;
Ok(())
}
/// Replace the property of a key (single value) to the store.
fn replace(
&mut self,
prefix: u8,
key: &Vec<u8>,
suffix: Option<u8>,
value: &Vec<u8>,
) -> Result<(), StorageError> {
let property = RocksdbKCVStore::compute_property(prefix, key, suffix);
self.tx()
.put(property, value)
.map_err(|e| StorageError::BackendError)?;
Ok(())
}
/// Delete a property from the store.
fn del(&mut self, prefix: u8, key: &Vec<u8>, suffix: Option<u8>) -> Result<(), StorageError> {
let property = RocksdbKCVStore::compute_property(prefix, key, suffix);
let res = self.tx().delete(property);
if res.is_err() {
if let ErrorKind::NotFound = res.unwrap_err().kind() {
return Ok(());
}
return Err(StorageError::BackendError);
}
Ok(())
}
/// Delete a specific value for a property from the store.
fn del_property_value(
&mut self,
prefix: u8,
key: &Vec<u8>,
suffix: Option<u8>,
value: &Vec<u8>,
) -> Result<(), StorageError> {
let property = RocksdbKCVStore::compute_property(prefix, key, suffix);
let exists = self
.tx()
.get_for_update(property.clone(), true)
.map_err(|e| StorageError::BackendError)?;
match exists {
Some(val) => {
if val.eq(value) {
self.tx()
.delete(property)
.map_err(|e| StorageError::BackendError)?;
}
}
None => return Err(StorageError::DifferentValue),
}
Ok(())
}
/// Delete all properties of a key from the store.
fn del_all(
&mut self,
prefix: u8,
key: &Vec<u8>,
all_suffixes: &[u8],
) -> Result<(), StorageError> {
for suffix in all_suffixes {
self.del(prefix, key, Some(*suffix))?;
}
if all_suffixes.is_empty() {
self.del(prefix, key, None)?;
}
Ok(())
}
}
pub struct RocksdbKCVStore {
/// the main store where all the properties of keys are stored
main_db: TransactionDB,
/// path for the storage backend data
path: String,
}
fn compare<T: Ord>(a: &[T], b: &[T]) -> std::cmp::Ordering {
let mut iter_b = b.iter();
for v in a {
match iter_b.next() {
Some(w) => match v.cmp(w) {
std::cmp::Ordering::Equal => continue,
ord => return ord,
},
None => break,
}
}
return a.len().cmp(&b.len());
}
impl ReadTransaction for RocksdbKCVStore {
fn get_all_keys_and_values(
&self,
prefix: u8,
key_size: usize,
key_prefix: Vec<u8>,
suffix: Option<u8>,
) -> Result<Vec<(Vec<u8>, Vec<u8>)>, StorageError> {
if key_prefix.len() > key_size {
return Err(StorageError::InvalidValue);
}
let mut vec_key_start = key_prefix.clone();
let mut trailing_zeros = vec![0u8; key_size - key_prefix.len()];
vec_key_start.append(&mut trailing_zeros);
let mut vec_key_end = key_prefix.clone();
let mut trailing_max = vec![255u8; key_size - key_prefix.len()];
vec_key_end.append(&mut trailing_max);
let property_start = Self::compute_property(prefix, &vec_key_start, suffix);
let property_end =
Self::compute_property(prefix, &vec_key_end, Some(suffix.unwrap_or(255u8)));
let mut iter = self
.main_db
.iterator(IteratorMode::From(&property_start, Direction::Forward));
let mut vector: Vec<(Vec<u8>, Vec<u8>)> = vec![];
while let res = iter.next() {
match res {
Some(Ok(val)) => {
match compare(&val.0, property_end.as_slice()) {
std::cmp::Ordering::Less | std::cmp::Ordering::Equal => {
if suffix.is_some() {
if val.0.len() < (key_size + 2)
|| val.0[1 + key_size] != suffix.unwrap()
{
continue;
}
// } else if val.0.len() > (key_size + 1) {
// continue;
}
vector.push((val.0.to_vec(), val.1.to_vec()));
}
_ => {} //,
}
}
Some(Err(_e)) => return Err(StorageError::BackendError),
None => {
break;
}
}
}
Ok(vector)
}
/// Load a single value property from the store.
fn get(&self, prefix: u8, key: &Vec<u8>, suffix: Option<u8>) -> Result<Vec<u8>, StorageError> {
let property = Self::compute_property(prefix, key, suffix);
let mut res = self
.main_db
.get(property)
.map_err(|e| StorageError::BackendError)?;
match res {
Some(val) => Ok(val),
None => Err(StorageError::NotFound),
}
}
/// Load all the values of a property from the store.
fn get_all(
&self,
prefix: u8,
key: &Vec<u8>,
suffix: Option<u8>,
) -> Result<Vec<Vec<u8>>, StorageError> {
unimplemented!();
}
/// Check if a specific value exists for a property from the store.
fn has_property_value(
&self,
prefix: u8,
key: &Vec<u8>,
suffix: Option<u8>,
value: &Vec<u8>,
) -> Result<(), StorageError> {
let property = Self::compute_property(prefix, key, suffix);
let exists = self
.main_db
.get(property)
.map_err(|e| StorageError::BackendError)?;
match exists {
Some(stored_value) => {
if stored_value.eq(value) {
Ok(())
} else {
Err(StorageError::DifferentValue)
}
}
None => Err(StorageError::NotFound),
}
}
}
impl KCVStore for RocksdbKCVStore {
fn write_transaction(
&self,
method: &mut dyn FnMut(&mut dyn WriteTransaction) -> Result<(), StorageError>,
) -> Result<(), StorageError> {
let tx = self.main_db.transaction();
let mut transaction = RocksdbTransaction {
store: self,
tx: Some(tx),
};
let res = method(&mut transaction);
if res.is_ok() {
transaction.commit();
//lock.sync(true);
}
res
}
/// Save a property value to the store.
fn put(
&self,
prefix: u8,
key: &Vec<u8>,
suffix: Option<u8>,
value: Vec<u8>,
) -> Result<(), StorageError> {
self.write_transaction(&mut |tx| tx.put(prefix, key, suffix, &value))
}
/// Replace the property of a key (single value) to the store.
fn replace(
&self,
prefix: u8,
key: &Vec<u8>,
suffix: Option<u8>,
value: Vec<u8>,
) -> Result<(), StorageError> {
self.write_transaction(&mut |tx| tx.replace(prefix, key, suffix, &value))
}
/// Delete a property from the store.
fn del(&self, prefix: u8, key: &Vec<u8>, suffix: Option<u8>) -> Result<(), StorageError> {
self.write_transaction(&mut |tx| tx.del(prefix, key, suffix))
}
/// Delete a specific value for a property from the store.
fn del_property_value(
&self,
prefix: u8,
key: &Vec<u8>,
suffix: Option<u8>,
value: Vec<u8>,
) -> Result<(), StorageError> {
self.write_transaction(&mut |tx| tx.del_property_value(prefix, key, suffix, &value))
}
/// Delete all properties of a key from the store.
fn del_all(&self, prefix: u8, key: &Vec<u8>, all_suffixes: &[u8]) -> Result<(), StorageError> {
for suffix in all_suffixes {
self.del(prefix, key, Some(*suffix))?;
}
if all_suffixes.is_empty() {
self.del(prefix, key, None)?;
}
Ok(())
}
}
impl RocksdbKCVStore {
pub fn path(&self) -> PathBuf {
PathBuf::from(&self.path)
}
fn compute_property(prefix: u8, key: &Vec<u8>, suffix: Option<u8>) -> Vec<u8> {
let mut new: Vec<u8> = Vec::with_capacity(key.len() + 2);
new.push(prefix);
new.extend(key);
if suffix.is_some() {
new.push(suffix.unwrap())
}
new
}
/// Opens the store and returns a KCVStore object that should be kept and used to manipulate the properties
/// The key is the encryption key for the data at rest.
pub fn open<'a>(path: &Path, key: [u8; 32]) -> Result<RocksdbKCVStore, StorageError> {
let mut opts = Options::default();
opts.create_if_missing(true);
opts.create_missing_column_families(true);
let env = Env::enc_env(key).unwrap();
opts.set_env(&env);
let tx_options = TransactionDBOptions::new();
let db: TransactionDB =
TransactionDB::open_cf(&opts, &tx_options, &path, vec!["cf0", "cf1"]).unwrap();
log_info!("created db with Rocksdb Version: {}", Env::version());
Ok(RocksdbKCVStore {
main_db: db,
path: path.to_str().unwrap().to_string(),
})
}
}

@ -0,0 +1,5 @@
// #[cfg(not(target_arch = "wasm32"))]
// pub mod repo_store;
#[cfg(not(target_arch = "wasm32"))]
pub mod kcv_store;

@ -0,0 +1,992 @@
// Copyright (c) 2022-2023 Niko Bonnieure, Par le Peuple, NextGraph.org developers
// All rights reserved.
// Licensed under the Apache License, Version 2.0
// <LICENSE-APACHE2 or http://www.apache.org/licenses/LICENSE-2.0>
// or the MIT license <LICENSE-MIT or http://opensource.org/licenses/MIT>,
// at your option. All files in the project carrying such
// notice may not be copied, modified, or distributed except
// according to those terms.
use p2p_repo::store::*;
use p2p_repo::types::*;
use p2p_repo::utils::*;
use p2p_repo::log::*;
use std::path::Path;
use std::sync::{Arc, RwLock};
use serde::{Deserialize, Serialize};
use serde_bare::error::Error;
#[derive(Debug)]
pub struct LmdbRepoStore {
/// the main store where all the repo blocks are stored
main_store: SingleStore<LmdbDatabase>,
/// store for the pin boolean, recently_used timestamp, and synced boolean
meta_store: SingleStore<LmdbDatabase>,
/// store for the expiry timestamp
expiry_store: MultiIntegerStore<LmdbDatabase, u32>,
/// store for the LRU list
recently_used_store: MultiIntegerStore<LmdbDatabase, u32>,
/// the opened environment so we can create new transactions
environment: Arc<RwLock<Rkv<LmdbEnvironment>>>,
}
// TODO: versioning V0
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
struct BlockMeta {
pub pin: bool,
pub last_used: Timestamp,
pub synced: bool,
}
impl RepoStore for LmdbRepoStore {
/// Retrieves a block from the storage backend.
fn get(&self, block_id: &BlockId) -> Result<Block, StorageError> {
let lock = self.environment.read().unwrap();
let reader = lock.read().unwrap();
let block_id_ser = serde_bare::to_vec(&block_id).unwrap();
let block_ser_res = self.main_store.get(&reader, block_id_ser.clone());
match block_ser_res {
Err(e) => Err(StorageError::BackendError),
Ok(None) => Err(StorageError::NotFound),
Ok(Some(block_ser)) => {
// updating recently_used
// first getting the meta for this BlockId
let meta_ser = self.meta_store.get(&reader, block_id_ser.clone()).unwrap();
match meta_ser {
Some(meta_value) => {
let mut meta =
serde_bare::from_slice::<BlockMeta>(&meta_value.to_bytes().unwrap())
.unwrap();
if meta.synced {
let mut writer = lock.write().unwrap();
let now = now_timestamp();
if !meta.pin {
// we remove the previous timestamp (last_used) from recently_used_store
self.remove_from_lru(&mut writer, &block_id_ser, &meta.last_used)
.unwrap();
// we add an entry to recently_used_store with now
self.add_to_lru(&mut writer, &block_id_ser, &now).unwrap();
}
// we save the new meta (with last_used:now)
meta.last_used = now;
let new_meta_ser = serde_bare::to_vec(&meta).unwrap();
self.meta_store
.put(
&mut writer,
block_id_ser,
&Value::Blob(new_meta_ser.as_slice()),
)
.unwrap();
// commit
writer.commit().unwrap();
}
}
_ => {} // there is no meta. we do nothing since we start to record LRU only once synced == true.
}
match serde_bare::from_slice::<Block>(&block_ser.to_bytes().unwrap()) {
Err(_e) => Err(StorageError::InvalidValue),
Ok(o) => {
if o.id() != *block_id {
log_debug!(
"Invalid ObjectId.\nExp: {:?}\nGot: {:?}\nContent: {:?}",
block_id,
o.id(),
o
);
panic!("CORRUPTION OF DATA !");
}
Ok(o)
}
}
}
}
}
/// Adds a block in the storage backend.
/// The block is persisted to disk.
/// Returns the BlockId of the Block.
fn put(&self, block: &Block) -> Result<BlockId, StorageError> {
let block_ser = serde_bare::to_vec(&block).unwrap();
let block_id = block.id();
let block_id_ser = serde_bare::to_vec(&block_id).unwrap();
let lock = self.environment.read().unwrap();
let mut writer = lock.write().unwrap();
// TODO: check if the block is already in store? if yes, don't put it again.
// I didnt do it yet because it is extra cost. surely a get on the store is lighter than a put
// but doing a get in additing to a put for every call, is probably even costlier. better to deal with that at the higher level
self.main_store
.put(
&mut writer,
&block_id_ser,
&Value::Blob(block_ser.as_slice()),
)
.unwrap();
// if it has an expiry, adding the BlockId to the expiry_store