jcoffey-dev
212cd77cd3
--export skipped three things, so a move from one database to another
(RocksDB to PostgreSQL, say) lost them without a word:
- archived items (subspace j), the records behind undelete;
- spam training samples (subspace w);
- the trained spam classifier and its trainer state, blobs stored under
fixed names that no blob link points at, so the walk over links never
reached them.
j and w now travel with the registry family, where their indexes and id
counters already were, so EXPORT_TYPES=registry keeps them consistent.
The two named blobs travel with the blob family. The file format is
unchanged and import reads any subspace it is given, so an export made
by an older binary still imports.
The full-text index (subspace z) stays out, on purpose. It belongs to one
search backend: PostgreSQL and MySQL index into their own tables and have
no z table at all, and external engines keep the index themselves. So
--import now returns the subspaces it wrote, and boot queues the
reindexAccounts and reindexTelemetry store maintenance tasks, the same
ones an administrator can queue by hand, to rebuild the index for
whichever search store the server runs with once it starts.
The round trip also turned up a loss in import itself: the SQL stores
add a negative amount with an UPDATE, which does nothing to a row that
isn't there yet, so every negative counter or quota vanished on import
into PostgreSQL, MySQL or SQLite. Import now creates the row first.
The in-memory subspaces (m, y) stay out: rate limits, locks, greylisting,
ACME challenge tokens and OAuth codes, all short-lived. Issued
certificates are registry objects and travel.
The store test now writes archived items, spam samples, directory
entries, the fork's own subspace and the named blobs, checks they come
back in place, then imports the same export into a fresh store of the
other local backend (RocksDB to SQLite, or SQLite to RocksDB), compares
it key for key and counter for counter, and checks the queued reindex.
It fails on the old export code ("Subspace j was not exported").
--help now says what an export holds.
359 lines
12 KiB
Rust
359 lines
12 KiB
Rust
/*
|
|
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
|
*
|
|
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
|
|
*
|
|
* Modified by Coffey Labs in 2026 for INBUXA.
|
|
*/
|
|
|
|
use super::backup::MAGIC_MARKER;
|
|
use crate::{Core, DATABASE_SCHEMA_VERSION};
|
|
use lz4_flex::frame::FrameDecoder;
|
|
use registry::{
|
|
schema::{
|
|
enums::{CompressionAlgo, TaskStoreMaintenanceType},
|
|
structs::{Task, TaskStatus, TaskStoreMaintenance},
|
|
},
|
|
types::EnumImpl,
|
|
};
|
|
use std::{
|
|
fs::File,
|
|
io::{BufReader, ErrorKind, Read},
|
|
path::{Path, PathBuf},
|
|
};
|
|
use store::{
|
|
BlobStore, IterateParams, SUBSPACE_BLOBS, SUBSPACE_COUNTER, SUBSPACE_INDEXES,
|
|
SUBSPACE_PROPERTY, SUBSPACE_QUOTA, SUBSPACE_REGISTRY_PK, SUBSPACE_TELEMETRY_SPAN, Store,
|
|
U32_LEN,
|
|
write::{
|
|
AnyClass, AnyKey, BatchBuilder, ValueClass,
|
|
key::{DeserializeBigEndian, is_node_id_key},
|
|
},
|
|
};
|
|
use types::{collection::Collection, field::Field};
|
|
use utils::{UnwrapFailure, failed};
|
|
|
|
impl Core {
|
|
/// Imports an export into an empty store and returns the subspaces it
|
|
/// wrote. inbuxa: the caller hands them to [`Core::queue_reindex`].
|
|
pub async fn restore(&self, src: PathBuf) -> Vec<u8> {
|
|
// Backup the core
|
|
let paths = if src.is_dir() {
|
|
let mut paths = Vec::new();
|
|
for entry in std::fs::read_dir(&src).failed("Failed to read directory") {
|
|
let entry = entry.failed("Failed to read entry");
|
|
let path = entry.path();
|
|
if path.is_file() {
|
|
paths.push(path);
|
|
}
|
|
}
|
|
paths
|
|
} else {
|
|
vec![src]
|
|
};
|
|
|
|
let mut conflicts = Vec::new();
|
|
for path in &paths {
|
|
let subspace = KeyValueReader::new(path).subspace;
|
|
if subspace_has_data(&self.storage.data, subspace).await {
|
|
conflicts.push(path.clone());
|
|
}
|
|
}
|
|
|
|
if !conflicts.is_empty() {
|
|
eprintln!(
|
|
"Cannot import: the target database already contains data in the key ranges being \
|
|
imported. This usually means the server was started before the import ran, which \
|
|
can create duplicate entries. Import into a fresh, empty database and do not \
|
|
start the server before importing. Conflicting dumps:"
|
|
);
|
|
for path in conflicts {
|
|
eprintln!(" {}", path.display());
|
|
}
|
|
std::process::exit(1);
|
|
}
|
|
|
|
let mut imported = paths
|
|
.iter()
|
|
.map(|path| KeyValueReader::new(path).subspace)
|
|
.collect::<Vec<_>>();
|
|
imported.sort_unstable();
|
|
imported.dedup();
|
|
|
|
let mut tasks = Vec::new();
|
|
for path in paths {
|
|
let storage = self.storage.clone();
|
|
let blob_store = self.storage.blob.clone();
|
|
tasks.push(tokio::spawn(async move {
|
|
restore_file(storage.data, blob_store, &path).await;
|
|
}));
|
|
}
|
|
|
|
for task in tasks {
|
|
task.await.failed("Failed to wait for task");
|
|
}
|
|
|
|
imported
|
|
}
|
|
|
|
/// inbuxa: an export never carries the full-text index. It is built by
|
|
/// and for one search backend (the SQL stores index into their own
|
|
/// tables, the key-value stores into a subspace, external engines keep it
|
|
/// themselves), so it would be wrong or unreadable after a move to
|
|
/// another one. Instead, an import queues the same reindex tasks an
|
|
/// administrator can queue by hand (`reindexAccounts` and
|
|
/// `reindexTelemetry` store maintenance), and the server rebuilds the
|
|
/// index for whatever search store it is configured with once it starts.
|
|
pub async fn queue_reindex(&self, imported: &[u8]) -> Vec<TaskStoreMaintenanceType> {
|
|
let mut queued = Vec::new();
|
|
if imported.contains(&SUBSPACE_PROPERTY) {
|
|
queued.push(TaskStoreMaintenanceType::ReindexAccounts);
|
|
}
|
|
if imported.contains(&SUBSPACE_TELEMETRY_SPAN) {
|
|
queued.push(TaskStoreMaintenanceType::ReindexTelemetry);
|
|
}
|
|
if queued.is_empty() {
|
|
return queued;
|
|
}
|
|
|
|
let mut batch = BatchBuilder::new();
|
|
for maintenance_type in &queued {
|
|
batch.schedule_task(Task::StoreMaintenance(TaskStoreMaintenance {
|
|
maintenance_type: *maintenance_type,
|
|
status: TaskStatus::now(),
|
|
shard_index: None,
|
|
}));
|
|
}
|
|
self.storage
|
|
.data
|
|
.write(batch.build_all())
|
|
.await
|
|
.failed("Failed to queue the reindex tasks");
|
|
|
|
println!(
|
|
"Queued {} to rebuild the search index; it runs when the server starts.",
|
|
queued
|
|
.iter()
|
|
.map(|t| t.as_str())
|
|
.collect::<Vec<_>>()
|
|
.join(" and ")
|
|
);
|
|
|
|
queued
|
|
}
|
|
}
|
|
|
|
async fn subspace_has_data(store: &Store, subspace: u8) -> bool {
|
|
let mut has_data = false;
|
|
store
|
|
.iterate(
|
|
IterateParams::new(
|
|
AnyKey {
|
|
subspace,
|
|
key: vec![0u8],
|
|
},
|
|
AnyKey {
|
|
subspace,
|
|
key: vec![u8::MAX; 32],
|
|
},
|
|
)
|
|
.no_values(),
|
|
|key, _| {
|
|
if subspace == SUBSPACE_REGISTRY_PK && is_node_id_key(key) {
|
|
Ok(true)
|
|
} else {
|
|
has_data = true;
|
|
Ok(false)
|
|
}
|
|
},
|
|
)
|
|
.await
|
|
.failed("Failed to inspect target database");
|
|
has_data
|
|
}
|
|
|
|
async fn restore_file(store: Store, blob_store: BlobStore, path: &Path) {
|
|
println!("Importing database dump from {}.", path.to_str().unwrap());
|
|
|
|
let mut reader = KeyValueReader::new(path);
|
|
let mut batch = BatchBuilder::new();
|
|
|
|
match reader.subspace {
|
|
SUBSPACE_BLOBS => {
|
|
while let Some((key, value)) = reader.next() {
|
|
blob_store
|
|
.put_blob(&key, &value, CompressionAlgo::Lz4)
|
|
.await
|
|
.failed("Failed to write blob");
|
|
}
|
|
}
|
|
SUBSPACE_COUNTER | SUBSPACE_QUOTA => {
|
|
while let Some((key, value)) = reader.next() {
|
|
let class = ValueClass::Any(AnyClass {
|
|
subspace: reader.subspace,
|
|
key,
|
|
});
|
|
let value = u64::from_le_bytes(
|
|
value
|
|
.try_into()
|
|
.expect("Failed to deserialize counter/quota"),
|
|
) as i64;
|
|
// inbuxa: the SQL stores add a negative amount with an UPDATE,
|
|
// which does nothing to a row that isn't there yet, so a
|
|
// negative counter vanished on import. Create the row first.
|
|
if value < 0 {
|
|
batch.add(class.clone(), 0);
|
|
}
|
|
batch.add(class, value);
|
|
if batch.is_large_batch() {
|
|
store
|
|
.write(batch.build_all())
|
|
.await
|
|
.failed("Failed to write batch");
|
|
batch = BatchBuilder::new();
|
|
}
|
|
}
|
|
}
|
|
SUBSPACE_INDEXES => {
|
|
while let Some((key, _)) = reader.next() {
|
|
let account_id = key
|
|
.as_slice()
|
|
.deserialize_be_u32(0)
|
|
.failed("Failed to deserialize account ID");
|
|
let collection = *key.get(U32_LEN).failed("Missing collection byte");
|
|
let field = *key.get(U32_LEN + 1).failed("Missing field byte");
|
|
let value = key
|
|
.get(U32_LEN + 2..key.len() - U32_LEN)
|
|
.failed("Missing index key")
|
|
.to_vec();
|
|
let document_id = key
|
|
.as_slice()
|
|
.deserialize_be_u32(key.len() - U32_LEN)
|
|
.failed("Failed to deserialize document ID");
|
|
|
|
batch
|
|
.with_account_id(account_id)
|
|
.with_collection(Collection::from(collection))
|
|
.with_document(document_id)
|
|
.index(Field::new(field), value);
|
|
|
|
if batch.is_large_batch() {
|
|
store
|
|
.write(batch.build_all())
|
|
.await
|
|
.failed("Failed to write batch");
|
|
batch = BatchBuilder::new();
|
|
}
|
|
}
|
|
}
|
|
_ => {
|
|
while let Some((key, value)) = reader.next() {
|
|
batch.set(
|
|
ValueClass::Any(AnyClass {
|
|
subspace: reader.subspace,
|
|
key,
|
|
}),
|
|
value,
|
|
);
|
|
if batch.is_large_batch() {
|
|
store
|
|
.write(batch.build_all())
|
|
.await
|
|
.failed("Failed to write batch");
|
|
batch = BatchBuilder::new();
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
if !batch.is_empty() {
|
|
store
|
|
.write(batch.build_all())
|
|
.await
|
|
.failed("Failed to write batch");
|
|
}
|
|
}
|
|
|
|
struct KeyValueReader {
|
|
subspace: u8,
|
|
file: FrameDecoder<BufReader<File>>,
|
|
}
|
|
|
|
impl KeyValueReader {
|
|
fn new(path: &Path) -> Self {
|
|
let mut file = FrameDecoder::new(BufReader::new(
|
|
File::open(path).failed("Failed to open file"),
|
|
));
|
|
let mut buf = [0u8; 1];
|
|
file.read_exact(&mut buf)
|
|
.failed(&format!("Failed to read magic marker from {path:?}"));
|
|
|
|
if buf[0] != MAGIC_MARKER {
|
|
failed(&format!("Invalid magic marker in {path:?}"));
|
|
}
|
|
|
|
file.read_exact(&mut buf)
|
|
.failed(&format!("Failed to read subspace from {path:?}"));
|
|
let subspace = buf[0];
|
|
|
|
let mut buf = [0u8; 4];
|
|
file.read_exact(&mut buf)
|
|
.failed(&format!("Failed to read version from {path:?}"));
|
|
let version = u32::from_le_bytes(buf);
|
|
|
|
if version != DATABASE_SCHEMA_VERSION {
|
|
failed(&format!(
|
|
"Invalid database schema version in {path:?}: Expected {DATABASE_SCHEMA_VERSION}, found {version}"
|
|
));
|
|
}
|
|
|
|
Self { file, subspace }
|
|
}
|
|
|
|
fn next(&mut self) -> Option<(Vec<u8>, Vec<u8>)> {
|
|
let size = self.read_size()?;
|
|
|
|
let mut key = vec![0; size as usize];
|
|
self.file
|
|
.read_exact(&mut key)
|
|
.failed("Failed to read bytes");
|
|
let value = self.expect_sized_bytes();
|
|
|
|
Some((key, value))
|
|
}
|
|
|
|
fn read_size(&mut self) -> Option<u32> {
|
|
let mut result = 0;
|
|
let mut buf = [0u8; 1];
|
|
|
|
for shift in [0, 7, 14, 21, 28] {
|
|
if let Err(err) = self.file.read_exact(&mut buf) {
|
|
if err.kind() == ErrorKind::UnexpectedEof {
|
|
return None;
|
|
} else {
|
|
failed(&format!("Failed to read file: {err:?}"));
|
|
}
|
|
}
|
|
|
|
let byte = buf[0];
|
|
if (byte & 0x80) == 0 {
|
|
result |= (byte as u32) << shift;
|
|
return Some(result);
|
|
} else {
|
|
result |= ((byte & 0x7F) as u32) << shift;
|
|
}
|
|
}
|
|
|
|
failed("Invalid leb128 sequence")
|
|
}
|
|
|
|
fn expect_sized_bytes(&mut self) -> Vec<u8> {
|
|
let len = self.read_size().failed("Missing leb128 value sequence") as usize;
|
|
let mut bytes = vec![0; len];
|
|
self.file
|
|
.read_exact(&mut bytes)
|
|
.failed("Failed to read bytes");
|
|
bytes
|
|
}
|
|
}
|