gnosis/lighthouse
1
0
Fork 0
mirror of https://github.com/sigp/lighthouse.git synced 2026-04-17 04:48:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Optimise slasher DB layout and switch to MDBX (#2776)

Browse Source 
## Issue Addressed

Closes #2286
Closes #2538
Closes #2342

## Proposed Changes

Part II of major slasher optimisations after #2767

These changes will be backwards-incompatible due to the move to MDBX (and the schema change) 😱 

* [x] Shrink attester keys from 16 bytes to 7 bytes.
* [x] Shrink attester records from 64 bytes to 6 bytes.
* [x] Separate `DiskConfig` from regular `Config`.
* [x] Add configuration for the LRU cache size.
* [x] Add a "migration" that deletes any legacy LMDB database.
This commit is contained in:
Michael Sproul
2021-12-21 08:23:17 +00:00
parent a290a3c537
commit 3b61ac9cbf
26 changed files with 963 additions and 566 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
slasher/src/array.rs
View File

@@ -1,8 +1,9 @@
use crate::metrics::{self, SLASHER_COMPRESSION_RATIO, SLASHER_NUM_CHUNKS_UPDATED};
use crate::RwTransaction;
use crate::{AttesterSlashingStatus, Config, Error, IndexedAttesterRecord, SlasherDB};
use flate2::bufread::{ZlibDecoder, ZlibEncoder};
use lmdb::{RwTransaction, Transaction};
use serde_derive::{Deserialize, Serialize};
use std::borrow::{Borrow, Cow};
use std::collections::{btree_map::Entry, BTreeMap, HashSet};
use std::convert::TryFrom;
use std::io::Read;
@@ -146,7 +147,10 @@ pub trait TargetArrayChunk: Sized + serde::Serialize + serde::de::DeserializeOwn
fn next_start_epoch(start_epoch: Epoch, config: &Config) -> Epoch;
fn select_db<E: EthSpec>(db: &SlasherDB<E>) -> lmdb::Database;
fn select_db<'txn, E: EthSpec>(
db: &SlasherDB<E>,
txn: &'txn RwTransaction<'txn>,
) -> Result<mdbx::Database<'txn>, Error>;
fn load<E: EthSpec>(
db: &SlasherDB<E>,
@@ -156,13 +160,13 @@ pub trait TargetArrayChunk: Sized + serde::Serialize + serde::de::DeserializeOwn
config: &Config,
) -> Result<Option<Self>, Error> {
let disk_key = config.disk_key(validator_chunk_index, chunk_index);
let chunk_bytes = match txn.get(Self::select_db(db), &disk_key.to_be_bytes()) {
Ok(chunk_bytes) => chunk_bytes,
Err(lmdb::Error::NotFound) => return Ok(None),
Err(e) => return Err(e.into()),
};
let chunk_bytes: Cow<[u8]> =
match txn.get(&Self::select_db(db, txn)?, &disk_key.to_be_bytes())? {
Some(chunk_bytes) => chunk_bytes,
None => return Ok(None),
};
let chunk = bincode::deserialize_from(ZlibDecoder::new(chunk_bytes))?;
let chunk = bincode::deserialize_from(ZlibDecoder::new(chunk_bytes.borrow()))?;
Ok(Some(chunk))
}
@@ -185,7 +189,7 @@ pub trait TargetArrayChunk: Sized + serde::Serialize + serde::de::DeserializeOwn
metrics::set_float_gauge(&SLASHER_COMPRESSION_RATIO, compression_ratio);
txn.put(
Self::select_db(db),
&Self::select_db(db, txn)?,
&disk_key.to_be_bytes(),
&compressed_value,
SlasherDB::<E>::write_flags(),
@@ -292,8 +296,11 @@ impl TargetArrayChunk for MinTargetChunk {
start_epoch / chunk_size * chunk_size - 1
}
fn select_db<E: EthSpec>(db: &SlasherDB<E>) -> lmdb::Database {
db.min_targets_db
fn select_db<'txn, E: EthSpec>(
db: &SlasherDB<E>,
txn: &'txn RwTransaction<'txn>,
) -> Result<mdbx::Database<'txn>, Error> {
db.min_targets_db(txn)
}
}
@@ -391,8 +398,11 @@ impl TargetArrayChunk for MaxTargetChunk {
(start_epoch / chunk_size + 1) * chunk_size
}
fn select_db<E: EthSpec>(db: &SlasherDB<E>) -> lmdb::Database {
db.max_targets_db
fn select_db<'txn, E: EthSpec>(
db: &SlasherDB<E>,
txn: &'txn RwTransaction<'txn>,
) -> Result<mdbx::Database<'txn>, Error> {
db.max_targets_db(txn)
}
}

59
slasher/src/attester_record.rs
View File

@@ -1,17 +1,53 @@
use crate::{database::IndexedAttestationId, Error};
use ssz_derive::{Decode, Encode};
use std::sync::Arc;
use std::borrow::Cow;
use std::sync::{
atomic::{AtomicU64, Ordering},
Arc,
};
use tree_hash::TreeHash as _;
use tree_hash_derive::TreeHash;
use types::{AggregateSignature, EthSpec, Hash256, IndexedAttestation, VariableList};
#[derive(Debug, Clone, Copy, Encode, Decode)]
#[derive(Debug, Clone, Copy)]
pub struct AttesterRecord {
/// The hash of the attestation data, for checking double-voting.
/// The hash of the attestation data, for de-duplication.
pub attestation_data_hash: Hash256,
/// The hash of the indexed attestation, so it can be loaded.
pub indexed_attestation_hash: Hash256,
}
#[derive(Debug, Clone, Copy)]
pub struct CompactAttesterRecord {
/// The ID of the `IndexedAttestation` signed by this validator.
pub indexed_attestation_id: IndexedAttestationId,
}
impl CompactAttesterRecord {
pub fn new(indexed_attestation_id: IndexedAttestationId) -> Self {
Self {
indexed_attestation_id,
}
}
pub fn null() -> Self {
Self::new(IndexedAttestationId::null())
}
pub fn parse(bytes: Cow<[u8]>) -> Result<Self, Error> {
let id = IndexedAttestationId::parse(bytes)?;
Ok(Self::new(IndexedAttestationId::new(id)))
}
pub fn is_null(&self) -> bool {
self.indexed_attestation_id.is_null()
}
pub fn as_bytes(&self) -> &[u8] {
self.indexed_attestation_id.as_ref()
}
}
/// Bundling of an `IndexedAttestation` with an `AttesterRecord`.
///
/// This struct gets `Arc`d and passed around between each stage of queueing and processing.
@@ -19,11 +55,26 @@ pub struct AttesterRecord {
pub struct IndexedAttesterRecord<E: EthSpec> {
pub indexed: IndexedAttestation<E>,
pub record: AttesterRecord,
pub indexed_attestation_id: AtomicU64,
}
impl<E: EthSpec> IndexedAttesterRecord<E> {
pub fn new(indexed: IndexedAttestation<E>, record: AttesterRecord) -> Arc<Self> {
Arc::new(IndexedAttesterRecord { indexed, record })
Arc::new(IndexedAttesterRecord {
indexed,
record,
indexed_attestation_id: AtomicU64::new(0),
})
}
pub fn set_id(&self, id: u64) {
self.indexed_attestation_id
.compare_exchange(0, id, Ordering::Relaxed, Ordering::Relaxed)
.expect("IDs should only be initialized once");
}
pub fn get_id(&self) -> u64 {
self.indexed_attestation_id.load(Ordering::Relaxed)
}
}

44
slasher/src/config.rs
View File

@@ -9,14 +9,11 @@ pub const DEFAULT_HISTORY_LENGTH: usize = 4096;
pub const DEFAULT_UPDATE_PERIOD: u64 = 12;
pub const DEFAULT_SLOT_OFFSET: f64 = 10.5;
pub const DEFAULT_MAX_DB_SIZE: usize = 256 * 1024; // 256 GiB
pub const DEFAULT_ATTESTATION_ROOT_CACHE_SIZE: usize = 100_000;
pub const DEFAULT_BROADCAST: bool = false;
/// Database size to use for tests.
///
/// Mostly a workaround for Windows due to a bug in LMDB, see:
///
/// https://github.com/sigp/lighthouse/issues/2342
pub const TESTING_MAX_DB_SIZE: usize = 16; // MiB
pub const MAX_HISTORY_LENGTH: usize = 1 << 16;
pub const MDBX_GROWTH_STEP: isize = 256 * (1 << 20); // 256 MiB
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Config {
@@ -28,16 +25,21 @@ pub struct Config {
/// Update frequency in seconds.
pub update_period: u64,
/// Offset from the start of the slot to begin processing.
#[serde(skip, default = "default_slot_offset")]
pub slot_offset: f64,
/// Maximum size of the LMDB database in megabytes.
/// Maximum size of the database in megabytes.
pub max_db_size_mbs: usize,
/// Maximum size of the in-memory cache for attestation roots.
pub attestation_root_cache_size: usize,
/// Whether to broadcast slashings found to the network.
pub broadcast: bool,
}
fn default_slot_offset() -> f64 {
DEFAULT_SLOT_OFFSET
/// Immutable configuration parameters which are stored on disk and checked for consistency.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct DiskConfig {
pub chunk_size: usize,
pub validator_chunk_size: usize,
pub history_length: usize,
}
impl Config {
@@ -50,16 +52,11 @@ impl Config {
update_period: DEFAULT_UPDATE_PERIOD,
slot_offset: DEFAULT_SLOT_OFFSET,
max_db_size_mbs: DEFAULT_MAX_DB_SIZE,
attestation_root_cache_size: DEFAULT_ATTESTATION_ROOT_CACHE_SIZE,
broadcast: DEFAULT_BROADCAST,
}
}
/// Use a smaller max DB size for testing.
pub fn for_testing(mut self) -> Self {
self.max_db_size_mbs = TESTING_MAX_DB_SIZE;
self
}
pub fn validate(&self) -> Result<(), Error> {
if self.chunk_size == 0
|| self.validator_chunk_size == 0
@@ -74,15 +71,22 @@ impl Config {
chunk_size: self.chunk_size,
history_length: self.history_length,
})
} else if self.history_length > MAX_HISTORY_LENGTH {
Err(Error::ConfigInvalidHistoryLength {
history_length: self.history_length,
max_history_length: MAX_HISTORY_LENGTH,
})
} else {
Ok(())
}
}
pub fn is_compatible(&self, other: &Config) -> bool {
self.chunk_size == other.chunk_size
&& self.validator_chunk_size == other.validator_chunk_size
&& self.history_length == other.history_length
pub fn disk_config(&self) -> DiskConfig {
DiskConfig {
chunk_size: self.chunk_size,
validator_chunk_size: self.validator_chunk_size,
history_length: self.history_length,
}
}
pub fn chunk_index(&self, epoch: Epoch) -> usize {

683
slasher/src/database.rs
View File

@@ -1,26 +1,41 @@
use crate::config::MDBX_GROWTH_STEP;
use crate::{
utils::{TxnMapFull, TxnOptional},
AttesterRecord, AttesterSlashingStatus, Config, Error, ProposerSlashingStatus,
metrics, utils::TxnMapFull, AttesterRecord, AttesterSlashingStatus, CompactAttesterRecord,
Config, Environment, Error, ProposerSlashingStatus, RwTransaction,
};
use byteorder::{BigEndian, ByteOrder};
use lmdb::{Cursor, Database, DatabaseFlags, Environment, RwTransaction, Transaction, WriteFlags};
use serde::Deserialize;
use lru::LruCache;
use mdbx::{Database, DatabaseFlags, Geometry, WriteFlags};
use parking_lot::Mutex;
use serde::de::DeserializeOwned;
use slog::{info, Logger};
use ssz::{Decode, Encode};
use std::borrow::{Borrow, Cow};
use std::marker::PhantomData;
use std::ops::Range;
use std::path::Path;
use std::sync::Arc;
use tree_hash::TreeHash;
use types::{
Epoch, EthSpec, Hash256, IndexedAttestation, ProposerSlashing, SignedBeaconBlockHeader, Slot,
};
/// Current database schema version, to check compatibility of on-disk DB with software.
pub const CURRENT_SCHEMA_VERSION: u64 = 2;
pub const CURRENT_SCHEMA_VERSION: u64 = 3;
/// Metadata about the slashing database itself.
const METADATA_DB: &str = "metadata";
/// Map from `(target_epoch, validator_index)` to `AttesterRecord`.
/// Map from `(target_epoch, validator_index)` to `CompactAttesterRecord`.
const ATTESTERS_DB: &str = "attesters";
/// Map from `(target_epoch, indexed_attestation_hash)` to `IndexedAttestation`.
/// Companion database for the attesters DB mapping `validator_index` to largest `target_epoch`
/// stored for that validator in the attesters DB.
///
/// Used to implement wrap-around semantics for target epochs modulo the history length.
const ATTESTERS_MAX_TARGETS_DB: &str = "attesters_max_targets";
/// Map from `indexed_attestation_id` to `IndexedAttestation`.
const INDEXED_ATTESTATION_DB: &str = "indexed_attestations";
/// Map from `(target_epoch, indexed_attestation_hash)` to `indexed_attestation_id`.
const INDEXED_ATTESTATION_ID_DB: &str = "indexed_attestation_ids";
/// Table of minimum targets for every source epoch within range.
const MIN_TARGETS_DB: &str = "min_targets";
/// Table of maximum targets for every source epoch within range.
@@ -32,31 +47,31 @@ const CURRENT_EPOCHS_DB: &str = "current_epochs";
/// Map from `(slot, validator_index)` to `SignedBeaconBlockHeader`.
const PROPOSERS_DB: &str = "proposers";
/// The number of DBs for LMDB to use (equal to the number of DBs defined above).
const LMDB_MAX_DBS: u32 = 7;
/// The number of DBs for MDBX to use (equal to the number of DBs defined above).
const MAX_NUM_DBS: usize = 9;
/// Filename for the legacy (LMDB) database file, so that it may be deleted.
const LEGACY_DB_FILENAME: &str = "data.mdb";
const LEGACY_DB_LOCK_FILENAME: &str = "lock.mdb";
/// Constant key under which the schema version is stored in the `metadata_db`.
const METADATA_VERSION_KEY: &[u8] = &[0];
/// Constant key under which the slasher configuration is stored in the `metadata_db`.
const METADATA_CONFIG_KEY: &[u8] = &[1];
const ATTESTER_KEY_SIZE: usize = 16;
const ATTESTER_KEY_SIZE: usize = 7;
const PROPOSER_KEY_SIZE: usize = 16;
const CURRENT_EPOCH_KEY_SIZE: usize = 8;
const INDEXED_ATTESTATION_KEY_SIZE: usize = 40;
const INDEXED_ATTESTATION_ID_SIZE: usize = 6;
const INDEXED_ATTESTATION_ID_KEY_SIZE: usize = 40;
const MEGABYTE: usize = 1 << 20;
#[derive(Debug)]
pub struct SlasherDB<E: EthSpec> {
pub(crate) env: Environment,
pub(crate) indexed_attestation_db: Database,
pub(crate) attesters_db: Database,
pub(crate) min_targets_db: Database,
pub(crate) max_targets_db: Database,
pub(crate) current_epochs_db: Database,
pub(crate) proposers_db: Database,
pub(crate) metadata_db: Database,
config: Arc<Config>,
/// LRU cache mapping indexed attestation IDs to their attestation data roots.
attestation_root_cache: Mutex<LruCache<IndexedAttestationId, Hash256>>,
pub(crate) config: Arc<Config>,
_phantom: PhantomData<E>,
}
@@ -64,27 +79,27 @@ pub struct SlasherDB<E: EthSpec> {
///
/// Stored as big-endian `(target_epoch, validator_index)` to enable efficient iteration
/// while pruning.
///
/// The target epoch is stored in 2 bytes modulo the `history_length`.
///
/// The validator index is stored in 5 bytes (validator registry limit is 2^40).
#[derive(Debug)]
pub struct AttesterKey {
data: [u8; ATTESTER_KEY_SIZE],
}
impl AttesterKey {
pub fn new(validator_index: u64, target_epoch: Epoch) -> Self {
pub fn new(validator_index: u64, target_epoch: Epoch, config: &Config) -> Self {
let mut data = [0; ATTESTER_KEY_SIZE];
data[0..8].copy_from_slice(&target_epoch.as_u64().to_be_bytes());
data[8..ATTESTER_KEY_SIZE].copy_from_slice(&validator_index.to_be_bytes());
AttesterKey { data }
}
pub fn parse(data: &[u8]) -> Result<(Epoch, u64), Error> {
if data.len() == ATTESTER_KEY_SIZE {
let target_epoch = Epoch::new(BigEndian::read_u64(&data[..8]));
let validator_index = BigEndian::read_u64(&data[8..]);
Ok((target_epoch, validator_index))
} else {
Err(Error::AttesterKeyCorrupt { length: data.len() })
}
BigEndian::write_uint(
&mut data[..2],
target_epoch.as_u64() % config.history_length as u64,
2,
);
BigEndian::write_uint(&mut data[2..], validator_index, 5);
AttesterKey { data }
}
}
@@ -111,7 +126,7 @@ impl ProposerKey {
ProposerKey { data }
}
pub fn parse(data: &[u8]) -> Result<(Slot, u64), Error> {
pub fn parse(data: Cow<[u8]>) -> Result<(Slot, u64), Error> {
if data.len() == PROPOSER_KEY_SIZE {
let slot = Slot::new(BigEndian::read_u64(&data[..8]));
let validator_index = BigEndian::read_u64(&data[8..]);
@@ -148,93 +163,213 @@ impl AsRef<[u8]> for CurrentEpochKey {
}
/// Key containing an epoch and an indexed attestation hash.
pub struct IndexedAttestationKey {
target_and_root: [u8; INDEXED_ATTESTATION_KEY_SIZE],
pub struct IndexedAttestationIdKey {
target_and_root: [u8; INDEXED_ATTESTATION_ID_KEY_SIZE],
}
impl IndexedAttestationKey {
impl IndexedAttestationIdKey {
pub fn new(target_epoch: Epoch, indexed_attestation_root: Hash256) -> Self {
let mut data = [0; INDEXED_ATTESTATION_KEY_SIZE];
let mut data = [0; INDEXED_ATTESTATION_ID_KEY_SIZE];
data[0..8].copy_from_slice(&target_epoch.as_u64().to_be_bytes());
data[8..INDEXED_ATTESTATION_KEY_SIZE].copy_from_slice(indexed_attestation_root.as_bytes());
data[8..INDEXED_ATTESTATION_ID_KEY_SIZE]
.copy_from_slice(indexed_attestation_root.as_bytes());
Self {
target_and_root: data,
}
}
pub fn parse(data: &[u8]) -> Result<(Epoch, Hash256), Error> {
if data.len() == INDEXED_ATTESTATION_KEY_SIZE {
pub fn parse(data: Cow<[u8]>) -> Result<(Epoch, Hash256), Error> {
if data.len() == INDEXED_ATTESTATION_ID_KEY_SIZE {
let target_epoch = Epoch::new(BigEndian::read_u64(&data[..8]));
let indexed_attestation_root = Hash256::from_slice(&data[8..]);
Ok((target_epoch, indexed_attestation_root))
} else {
Err(Error::IndexedAttestationKeyCorrupt { length: data.len() })
Err(Error::IndexedAttestationIdKeyCorrupt { length: data.len() })
}
}
}
impl AsRef<[u8]> for IndexedAttestationKey {
impl AsRef<[u8]> for IndexedAttestationIdKey {
fn as_ref(&self) -> &[u8] {
&self.target_and_root
}
}
/// Key containing a 6-byte indexed attestation ID.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct IndexedAttestationId {
id: [u8; INDEXED_ATTESTATION_ID_SIZE],
}
impl IndexedAttestationId {
pub fn new(id: u64) -> Self {
let mut data = [0; INDEXED_ATTESTATION_ID_SIZE];
BigEndian::write_uint(&mut data, id, INDEXED_ATTESTATION_ID_SIZE);
Self { id: data }
}
pub fn parse(data: Cow<[u8]>) -> Result<u64, Error> {
if data.len() == INDEXED_ATTESTATION_ID_SIZE {
Ok(BigEndian::read_uint(
data.borrow(),
INDEXED_ATTESTATION_ID_SIZE,
))
} else {
Err(Error::IndexedAttestationIdCorrupt { length: data.len() })
}
}
pub fn null() -> Self {
Self::new(0)
}
pub fn is_null(&self) -> bool {
self.id == [0, 0, 0, 0, 0, 0]
}
pub fn as_u64(&self) -> u64 {
BigEndian::read_uint(&self.id, INDEXED_ATTESTATION_ID_SIZE)
}
}
impl AsRef<[u8]> for IndexedAttestationId {
fn as_ref(&self) -> &[u8] {
&self.id
}
}
/// Bincode deserialization specialised to `Cow<[u8]>`.
fn bincode_deserialize<T: DeserializeOwned>(bytes: Cow<[u8]>) -> Result<T, Error> {
Ok(bincode::deserialize(bytes.borrow())?)
}
fn ssz_decode<T: Decode>(bytes: Cow<[u8]>) -> Result<T, Error> {
Ok(T::from_ssz_bytes(bytes.borrow())?)
}
impl<E: EthSpec> SlasherDB<E> {
pub fn open(config: Arc<Config>) -> Result<Self, Error> {
pub fn open(config: Arc<Config>, log: Logger) -> Result<Self, Error> {
// Delete any legacy LMDB database.
Self::delete_legacy_file(&config.database_path, LEGACY_DB_FILENAME, &log)?;
Self::delete_legacy_file(&config.database_path, LEGACY_DB_LOCK_FILENAME, &log)?;
std::fs::create_dir_all(&config.database_path)?;
let env = Environment::new()
.set_max_dbs(LMDB_MAX_DBS)
.set_map_size(config.max_db_size_mbs * MEGABYTE)
.set_max_dbs(MAX_NUM_DBS)
.set_geometry(Self::geometry(&config))
.open_with_permissions(&config.database_path, 0o600)?;
let indexed_attestation_db =
env.create_db(Some(INDEXED_ATTESTATION_DB), Self::db_flags())?;
let attesters_db = env.create_db(Some(ATTESTERS_DB), Self::db_flags())?;
let min_targets_db = env.create_db(Some(MIN_TARGETS_DB), Self::db_flags())?;
let max_targets_db = env.create_db(Some(MAX_TARGETS_DB), Self::db_flags())?;
let current_epochs_db = env.create_db(Some(CURRENT_EPOCHS_DB), Self::db_flags())?;
let proposers_db = env.create_db(Some(PROPOSERS_DB), Self::db_flags())?;
let metadata_db = env.create_db(Some(METADATA_DB), Self::db_flags())?;
let txn = env.begin_rw_txn()?;
txn.create_db(Some(INDEXED_ATTESTATION_DB), Self::db_flags())?;
txn.create_db(Some(INDEXED_ATTESTATION_ID_DB), Self::db_flags())?;
txn.create_db(Some(ATTESTERS_DB), Self::db_flags())?;
txn.create_db(Some(ATTESTERS_MAX_TARGETS_DB), Self::db_flags())?;
txn.create_db(Some(MIN_TARGETS_DB), Self::db_flags())?;
txn.create_db(Some(MAX_TARGETS_DB), Self::db_flags())?;
txn.create_db(Some(CURRENT_EPOCHS_DB), Self::db_flags())?;
txn.create_db(Some(PROPOSERS_DB), Self::db_flags())?;
txn.create_db(Some(METADATA_DB), Self::db_flags())?;
txn.commit()?;
#[cfg(windows)]
{
use filesystem::restrict_file_permissions;
let data = config.database_path.join("data.mdb");
let lock = config.database_path.join("lock.mdb");
let data = config.database_path.join("mdbx.dat");
let lock = config.database_path.join("mdbx.lck");
restrict_file_permissions(data).map_err(Error::DatabasePermissionsError)?;
restrict_file_permissions(lock).map_err(Error::DatabasePermissionsError)?;
}
let db = Self {
let attestation_root_cache = Mutex::new(LruCache::new(config.attestation_root_cache_size));
let mut db = Self {
env,
indexed_attestation_db,
attesters_db,
min_targets_db,
max_targets_db,
current_epochs_db,
proposers_db,
metadata_db,
attestation_root_cache,
config,
_phantom: PhantomData,
};
db = db.migrate()?;
let mut txn = db.begin_rw_txn()?;
db.migrate(&mut txn)?;
if let Some(on_disk_config) = db.load_config(&mut txn)? {
if !db.config.is_compatible(&on_disk_config) {
let current_disk_config = db.config.disk_config();
if current_disk_config != on_disk_config {
return Err(Error::ConfigIncompatible {
on_disk_config,
config: (*db.config).clone(),
config: current_disk_config,
});
}
}
db.store_config(&db.config, &mut txn)?;
txn.commit()?;
Ok(db)
}
fn delete_legacy_file(slasher_dir: &Path, filename: &str, log: &Logger) -> Result<(), Error> {
let path = slasher_dir.join(filename);
if path.is_file() {
info!(
log,
"Deleting legacy slasher DB";
"file" => ?path.display(),
);
std::fs::remove_file(&path)?;
}
Ok(())
}
fn open_db<'a>(&self, txn: &'a RwTransaction<'a>, name: &str) -> Result<Database<'a>, Error> {
Ok(txn.open_db(Some(name))?)
}
pub fn indexed_attestation_db<'a>(
&self,
txn: &'a RwTransaction<'a>,
) -> Result<Database<'a>, Error> {
self.open_db(txn, INDEXED_ATTESTATION_DB)
}
pub fn indexed_attestation_id_db<'a>(
&self,
txn: &'a RwTransaction<'a>,
) -> Result<Database<'a>, Error> {
self.open_db(txn, INDEXED_ATTESTATION_ID_DB)
}
pub fn attesters_db<'a>(&self, txn: &'a RwTransaction<'a>) -> Result<Database<'a>, Error> {
self.open_db(txn, ATTESTERS_DB)
}
pub fn attesters_max_targets_db<'a>(
&self,
txn: &'a RwTransaction<'a>,
) -> Result<Database<'a>, Error> {
self.open_db(txn, ATTESTERS_MAX_TARGETS_DB)
}
pub fn min_targets_db<'a>(&self, txn: &'a RwTransaction<'a>) -> Result<Database<'a>, Error> {
self.open_db(txn, MIN_TARGETS_DB)
}
pub fn max_targets_db<'a>(&self, txn: &'a RwTransaction<'a>) -> Result<Database<'a>, Error> {
self.open_db(txn, MAX_TARGETS_DB)
}
pub fn current_epochs_db<'a>(&self, txn: &'a RwTransaction<'a>) -> Result<Database<'a>, Error> {
self.open_db(txn, CURRENT_EPOCHS_DB)
}
pub fn proposers_db<'a>(&self, txn: &'a RwTransaction<'a>) -> Result<Database<'a>, Error> {
self.open_db(txn, PROPOSERS_DB)
}
pub fn metadata_db<'a>(&self, txn: &'a RwTransaction<'a>) -> Result<Database<'a>, Error> {
self.open_db(txn, METADATA_DB)
}
pub fn db_flags() -> DatabaseFlags {
DatabaseFlags::default()
}
@@ -247,17 +382,24 @@ impl<E: EthSpec> SlasherDB<E> {
Ok(self.env.begin_rw_txn()?)
}
pub fn geometry(config: &Config) -> Geometry<Range<usize>> {
Geometry {
size: Some(0..config.max_db_size_mbs * MEGABYTE),
growth_step: Some(MDBX_GROWTH_STEP),
shrink_threshold: None,
page_size: None,
}
}
pub fn load_schema_version(&self, txn: &mut RwTransaction<'_>) -> Result<Option<u64>, Error> {
Ok(txn
.get(self.metadata_db, &METADATA_VERSION_KEY)
.optional()?
.map(bincode::deserialize)
.transpose()?)
txn.get(&self.metadata_db(txn)?, METADATA_VERSION_KEY)?
.map(bincode_deserialize)
.transpose()
}
pub fn store_schema_version(&self, txn: &mut RwTransaction<'_>) -> Result<(), Error> {
txn.put(
self.metadata_db,
&self.metadata_db(txn)?,
&METADATA_VERSION_KEY,
&bincode::serialize(&CURRENT_SCHEMA_VERSION)?,
Self::write_flags(),
@@ -269,20 +411,18 @@ impl<E: EthSpec> SlasherDB<E> {
///
/// This is generic in order to allow loading of configs for different schema versions.
/// Care should be taken to ensure it is only called for `Config`-like `T`.
pub fn load_config<'a, T: Deserialize<'a>>(
pub fn load_config<T: DeserializeOwned>(
&self,
txn: &'a mut RwTransaction<'_>,
txn: &mut RwTransaction<'_>,
) -> Result<Option<T>, Error> {
Ok(txn
.get(self.metadata_db, &METADATA_CONFIG_KEY)
.optional()?
.map(bincode::deserialize)
.transpose()?)
txn.get(&self.metadata_db(txn)?, METADATA_CONFIG_KEY)?
.map(bincode_deserialize)
.transpose()
}
pub fn store_config(&self, config: &Config, txn: &mut RwTransaction<'_>) -> Result<(), Error> {
txn.put(
self.metadata_db,
&self.metadata_db(txn)?,
&METADATA_CONFIG_KEY,
&bincode::serialize(config)?,
Self::write_flags(),
@@ -290,19 +430,70 @@ impl<E: EthSpec> SlasherDB<E> {
Ok(())
}
pub fn get_attester_max_target(
&self,
validator_index: u64,
txn: &mut RwTransaction<'_>,
) -> Result<Option<Epoch>, Error> {
txn.get(
&self.attesters_max_targets_db(txn)?,
CurrentEpochKey::new(validator_index).as_ref(),
)?
.map(ssz_decode)
.transpose()
}
pub fn update_attester_max_target(
&self,
validator_index: u64,
previous_max_target: Option<Epoch>,
max_target: Epoch,
txn: &mut RwTransaction<'_>,
) -> Result<(), Error> {
// Don't update maximum if new target is less than or equal to previous. In the case of
// no previous we *do* want to update.
if previous_max_target.map_or(false, |prev_max| max_target <= prev_max) {
return Ok(());
}
// Zero out attester DB entries which are now older than the history length.
// Avoid writing the whole array on initialization (`previous_max_target == None`), and
// avoid overwriting the entire attesters array more than once.
if let Some(previous_max_target) = previous_max_target {
let start_epoch = std::cmp::max(
previous_max_target.as_u64() + 1,
(max_target.as_u64() + 1).saturating_sub(self.config.history_length as u64),
);
for target_epoch in (start_epoch..max_target.as_u64()).map(Epoch::new) {
txn.put(
&self.attesters_db(txn)?,
&AttesterKey::new(validator_index, target_epoch, &self.config),
&CompactAttesterRecord::null().as_bytes(),
Self::write_flags(),
)?;
}
}
txn.put(
&self.attesters_max_targets_db(txn)?,
&CurrentEpochKey::new(validator_index),
&max_target.as_ssz_bytes(),
Self::write_flags(),
)?;
Ok(())
}
pub fn get_current_epoch_for_validator(
&self,
validator_index: u64,
txn: &mut RwTransaction<'_>,
) -> Result<Option<Epoch>, Error> {
Ok(txn
.get(
self.current_epochs_db,
&CurrentEpochKey::new(validator_index),
)
.optional()?
.map(Epoch::from_ssz_bytes)
.transpose()?)
txn.get(
&self.current_epochs_db(txn)?,
CurrentEpochKey::new(validator_index).as_ref(),
)?
.map(ssz_decode)
.transpose()
}
pub fn update_current_epoch_for_validator(
@@ -312,7 +503,7 @@ impl<E: EthSpec> SlasherDB<E> {
txn: &mut RwTransaction<'_>,
) -> Result<(), Error> {
txn.put(
self.current_epochs_db,
&self.current_epochs_db(txn)?,
&CurrentEpochKey::new(validator_index),
&current_epoch.as_ssz_bytes(),
Self::write_flags(),
@@ -320,41 +511,128 @@ impl<E: EthSpec> SlasherDB<E> {
Ok(())
}
pub fn store_indexed_attestation(
fn get_indexed_attestation_id(
&self,
txn: &mut RwTransaction<'_>,
indexed_attestation_hash: Hash256,
indexed_attestation: &IndexedAttestation<E>,
) -> Result<(), Error> {
let key = IndexedAttestationKey::new(
indexed_attestation.data.target.epoch,
indexed_attestation_hash,
);
let data = indexed_attestation.as_ssz_bytes();
key: &IndexedAttestationIdKey,
) -> Result<Option<u64>, Error> {
txn.get(&self.indexed_attestation_id_db(txn)?, key.as_ref())?
.map(IndexedAttestationId::parse)
.transpose()
}
fn put_indexed_attestation_id(
&self,
txn: &mut RwTransaction<'_>,
key: &IndexedAttestationIdKey,
value: IndexedAttestationId,
) -> Result<(), Error> {
txn.put(
self.indexed_attestation_db,
&key,
&data,
&self.indexed_attestation_id_db(txn)?,
key,
&value,
Self::write_flags(),
)?;
Ok(())
}
/// Store an indexed attestation and return its ID.
///
/// If the attestation is already stored then the existing ID will be returned without a write.
pub fn store_indexed_attestation(
&self,
txn: &mut RwTransaction<'_>,
indexed_attestation_hash: Hash256,
indexed_attestation: &IndexedAttestation<E>,
) -> Result<u64, Error> {
// Look-up ID by hash.
let id_key = IndexedAttestationIdKey::new(
indexed_attestation.data.target.epoch,
indexed_attestation_hash,
);
if let Some(indexed_att_id) = self.get_indexed_attestation_id(txn, &id_key)? {
return Ok(indexed_att_id);
}
// Store the new indexed attestation at the end of the current table.
let mut cursor = txn.cursor(&self.indexed_attestation_db(txn)?)?;
let indexed_att_id = match cursor.last::<_, ()>()? {
// First ID is 1 so that 0 can be used to represent `null` in `CompactAttesterRecord`.
None => 1,
Some((key_bytes, _)) => IndexedAttestationId::parse(key_bytes)? + 1,
};
let attestation_key = IndexedAttestationId::new(indexed_att_id);
let data = indexed_attestation.as_ssz_bytes();
cursor.put(attestation_key.as_ref(), &data, Self::write_flags())?;
drop(cursor);
// Update the (epoch, hash) to ID mapping.
self.put_indexed_attestation_id(txn, &id_key, attestation_key)?;
Ok(indexed_att_id)
}
pub fn get_indexed_attestation(
&self,
txn: &mut RwTransaction<'_>,
target_epoch: Epoch,
indexed_attestation_hash: Hash256,
indexed_attestation_id: IndexedAttestationId,
) -> Result<IndexedAttestation<E>, Error> {
let key = IndexedAttestationKey::new(target_epoch, indexed_attestation_hash);
let bytes = txn
.get(self.indexed_attestation_db, &key)
.optional()?
.get(
&self.indexed_attestation_db(txn)?,
indexed_attestation_id.as_ref(),
)?
.ok_or(Error::MissingIndexedAttestation {
root: indexed_attestation_hash,
id: indexed_attestation_id.as_u64(),
})?;
Ok(IndexedAttestation::from_ssz_bytes(bytes)?)
ssz_decode(bytes)
}
fn get_attestation_data_root(
&self,
txn: &mut RwTransaction<'_>,
indexed_id: IndexedAttestationId,
) -> Result<(Hash256, Option<IndexedAttestation<E>>), Error> {
metrics::inc_counter(&metrics::SLASHER_NUM_ATTESTATION_ROOT_QUERIES);
// If the value already exists in the cache, return it.
let mut cache = self.attestation_root_cache.lock();
if let Some(attestation_data_root) = cache.get(&indexed_id) {
metrics::inc_counter(&metrics::SLASHER_NUM_ATTESTATION_ROOT_HITS);
return Ok((*attestation_data_root, None));
}
// Otherwise, load the indexed attestation, compute the root and cache it.
let indexed_attestation = self.get_indexed_attestation(txn, indexed_id)?;
let attestation_data_root = indexed_attestation.data.tree_hash_root();
cache.put(indexed_id, attestation_data_root);
Ok((attestation_data_root, Some(indexed_attestation)))
}
pub fn cache_attestation_data_root(
&self,
indexed_attestation_id: IndexedAttestationId,
attestation_data_root: Hash256,
) {
let mut cache = self.attestation_root_cache.lock();
cache.put(indexed_attestation_id, attestation_data_root);
}
fn delete_attestation_data_roots(&self, ids: impl IntoIterator<Item = IndexedAttestationId>) {
let mut cache = self.attestation_root_cache.lock();
for indexed_id in ids {
cache.pop(&indexed_id);
}
}
pub fn attestation_root_cache_size(&self) -> usize {
self.attestation_root_cache.lock().len()
}
pub fn check_and_update_attester_record(
@@ -362,41 +640,57 @@ impl<E: EthSpec> SlasherDB<E> {
txn: &mut RwTransaction<'_>,
validator_index: u64,
attestation: &IndexedAttestation<E>,
record: AttesterRecord,
record: &AttesterRecord,
indexed_attestation_id: IndexedAttestationId,
) -> Result<AttesterSlashingStatus<E>, Error> {
// See if there's an existing attestation for this attester.
let target_epoch = attestation.data.target.epoch;
let prev_max_target = self.get_attester_max_target(validator_index, txn)?;
if let Some(existing_record) =
self.get_attester_record(txn, validator_index, target_epoch)?
self.get_attester_record(txn, validator_index, target_epoch, prev_max_target)?
{
// If the existing attestation data is identical, then this attestation is not
// If the existing indexed attestation is identical, then this attestation is not
// slashable and no update is required.
if existing_record.attestation_data_hash == record.attestation_data_hash {
let existing_att_id = existing_record.indexed_attestation_id;
if existing_att_id == indexed_attestation_id {
return Ok(AttesterSlashingStatus::NotSlashable);
}
// Otherwise, load the indexed attestation so we can confirm that it's slashable.
let existing_attestation = self.get_indexed_attestation(
txn,
target_epoch,
existing_record.indexed_attestation_hash,
)?;
// Otherwise, load the attestation data root and check slashability via a hash root
// comparison.
let (existing_data_root, opt_existing_att) =
self.get_attestation_data_root(txn, existing_att_id)?;
if existing_data_root == record.attestation_data_hash {
return Ok(AttesterSlashingStatus::NotSlashable);
}
// If we made it this far, then the attestation is slashable. Ensure that it's
// loaded, double-check the slashing condition and return.
let existing_attestation = opt_existing_att
.map_or_else(|| self.get_indexed_attestation(txn, existing_att_id), Ok)?;
if attestation.is_double_vote(&existing_attestation) {
Ok(AttesterSlashingStatus::DoubleVote(Box::new(
existing_attestation,
)))
} else {
Err(Error::AttesterRecordInconsistentRoot)
Err(Error::InconsistentAttestationDataRoot)
}
}
// If no attestation exists, insert a record for this validator.
else {
self.update_attester_max_target(validator_index, prev_max_target, target_epoch, txn)?;
txn.put(
self.attesters_db,
&AttesterKey::new(validator_index, target_epoch),
&record.as_ssz_bytes(),
&self.attesters_db(txn)?,
&AttesterKey::new(validator_index, target_epoch, &self.config),
&indexed_attestation_id,
Self::write_flags(),
)?;
Ok(AttesterSlashingStatus::NotSlashable)
}
}
@@ -407,13 +701,15 @@ impl<E: EthSpec> SlasherDB<E> {
validator_index: u64,
target_epoch: Epoch,
) -> Result<IndexedAttestation<E>, Error> {
let max_target = self.get_attester_max_target(validator_index, txn)?;
let record = self
.get_attester_record(txn, validator_index, target_epoch)?
.get_attester_record(txn, validator_index, target_epoch, max_target)?
.ok_or(Error::MissingAttesterRecord {
validator_index,
target_epoch,
})?;
self.get_indexed_attestation(txn, target_epoch, record.indexed_attestation_hash)
self.get_indexed_attestation(txn, record.indexed_attestation_id)
}
pub fn get_attester_record(
@@ -421,13 +717,18 @@ impl<E: EthSpec> SlasherDB<E> {
txn: &mut RwTransaction<'_>,
validator_index: u64,
target: Epoch,
) -> Result<Option<AttesterRecord>, Error> {
let attester_key = AttesterKey::new(validator_index, target);
prev_max_target: Option<Epoch>,
) -> Result<Option<CompactAttesterRecord>, Error> {
if prev_max_target.map_or(true, |prev_max| target > prev_max) {
return Ok(None);
}
let attester_key = AttesterKey::new(validator_index, target, &self.config);
Ok(txn
.get(self.attesters_db, &attester_key)
.optional()?
.map(AttesterRecord::from_ssz_bytes)
.transpose()?)
.get(&self.attesters_db(txn)?, attester_key.as_ref())?
.map(CompactAttesterRecord::parse)
.transpose()?
.filter(|record| !record.is_null()))
}
pub fn get_block_proposal(
@@ -437,11 +738,9 @@ impl<E: EthSpec> SlasherDB<E> {
slot: Slot,
) -> Result<Option<SignedBeaconBlockHeader>, Error> {
let proposer_key = ProposerKey::new(proposer_index, slot);
Ok(txn
.get(self.proposers_db, &proposer_key)
.optional()?
.map(SignedBeaconBlockHeader::from_ssz_bytes)
.transpose()?)
txn.get(&self.proposers_db(txn)?, proposer_key.as_ref())?
.map(ssz_decode)
.transpose()
}
pub fn check_or_insert_block_proposal(
@@ -465,7 +764,7 @@ impl<E: EthSpec> SlasherDB<E> {
}
} else {
txn.put(
self.proposers_db,
&self.proposers_db(txn)?,
&ProposerKey::new(proposer_index, slot),
&block_header.as_ssz_bytes(),
Self::write_flags(),
@@ -491,7 +790,6 @@ impl<E: EthSpec> SlasherDB<E> {
txn: &mut RwTransaction<'_>,
) -> Result<(), Error> {
self.prune_proposers(current_epoch, txn)?;
self.prune_attesters(current_epoch, txn)?;
self.prune_indexed_attestations(current_epoch, txn)?;
Ok(())
}
@@ -506,80 +804,22 @@ impl<E: EthSpec> SlasherDB<E> {
.saturating_sub(self.config.history_length)
.start_slot(E::slots_per_epoch());
let mut cursor = txn.open_rw_cursor(self.proposers_db)?;
let mut cursor = txn.cursor(&self.proposers_db(txn)?)?;
// Position cursor at first key, bailing out if the database is empty.
if cursor
.get(None, None, lmdb_sys::MDB_FIRST)
.optional()?
.is_none()
{
if cursor.first::<(), ()>()?.is_none() {
return Ok(());
}
loop {
let key_bytes = cursor
.get(None, None, lmdb_sys::MDB_GET_CURRENT)?
.0
.ok_or(Error::MissingProposerKey)?;
let (key_bytes, ()) = cursor.get_current()?.ok_or(Error::MissingProposerKey)?;
let (slot, _) = ProposerKey::parse(key_bytes)?;
if slot < min_slot {
cursor.del(Self::write_flags())?;
// End the loop if there is no next entry.
if cursor
.get(None, None, lmdb_sys::MDB_NEXT)
.optional()?
.is_none()
{
break;
}
} else {
break;
}
}
Ok(())
}
fn prune_attesters(
&self,
current_epoch: Epoch,
txn: &mut RwTransaction<'_>,
) -> Result<(), Error> {
let min_epoch = current_epoch
.saturating_add(1u64)
.saturating_sub(self.config.history_length as u64);
let mut cursor = txn.open_rw_cursor(self.attesters_db)?;
// Position cursor at first key, bailing out if the database is empty.
if cursor
.get(None, None, lmdb_sys::MDB_FIRST)
.optional()?
.is_none()
{
return Ok(());
}
loop {
let key_bytes = cursor
.get(None, None, lmdb_sys::MDB_GET_CURRENT)?
.0
.ok_or(Error::MissingAttesterKey)?;
let (target_epoch, _) = AttesterKey::parse(key_bytes)?;
if target_epoch < min_epoch {
cursor.del(Self::write_flags())?;
// End the loop if there is no next entry.
if cursor
.get(None, None, lmdb_sys::MDB_NEXT)
.optional()?
.is_none()
{
if cursor.next::<(), ()>()?.is_none() {
break;
}
} else {
@@ -599,39 +839,46 @@ impl<E: EthSpec> SlasherDB<E> {
.saturating_add(1u64)
.saturating_sub(self.config.history_length as u64);
let mut cursor = txn.open_rw_cursor(self.indexed_attestation_db)?;
// Collect indexed attestation IDs to delete.
let mut indexed_attestation_ids = vec![];
let mut cursor = txn.cursor(&self.indexed_attestation_id_db(txn)?)?;
// Position cursor at first key, bailing out if the database is empty.
if cursor
.get(None, None, lmdb_sys::MDB_FIRST)
.optional()?
.is_none()
{
if cursor.first::<(), ()>()?.is_none() {
return Ok(());
}
loop {
let key_bytes = cursor
.get(None, None, lmdb_sys::MDB_GET_CURRENT)?
.0
.ok_or(Error::MissingAttesterKey)?;
let (key_bytes, value) = cursor
.get_current()?
.ok_or(Error::MissingIndexedAttestationIdKey)?;
let (target_epoch, _) = IndexedAttestationKey::parse(key_bytes)?;
let (target_epoch, _) = IndexedAttestationIdKey::parse(key_bytes)?;
if target_epoch < min_epoch {
indexed_attestation_ids.push(IndexedAttestationId::new(
IndexedAttestationId::parse(value)?,
));
cursor.del(Self::write_flags())?;
if cursor
.get(None, None, lmdb_sys::MDB_NEXT)
.optional()?
.is_none()
{
if cursor.next::<(), ()>()?.is_none() {
break;
}
} else {
break;
}
}
drop(cursor);
// Delete the indexed attestations.
// Optimisation potential: use a cursor here.
let indexed_attestation_db = self.indexed_attestation_db(txn)?;
for indexed_attestation_id in &indexed_attestation_ids {
txn.del(&indexed_attestation_db, indexed_attestation_id, None)?;
}
self.delete_attestation_data_roots(indexed_attestation_ids);
Ok(())
}

32
slasher/src/error.rs
View File

@@ -1,10 +1,10 @@
use crate::Config;
use crate::config::{Config, DiskConfig};
use std::io;
use types::{Epoch, Hash256};
use types::Epoch;
#[derive(Debug)]
pub enum Error {
DatabaseError(lmdb::Error),
DatabaseError(mdbx::Error),
DatabaseIOError(io::Error),
DatabasePermissionsError(filesystem::Error),
SszDecodeError(ssz::DecodeError),
@@ -19,12 +19,16 @@ pub enum Error {
chunk_size: usize,
history_length: usize,
},
ConfigInvalidHistoryLength {
history_length: usize,
max_history_length: usize,
},
ConfigInvalidZeroParameter {
config: Config,
},
ConfigIncompatible {
on_disk_config: Config,
config: Config,
on_disk_config: DiskConfig,
config: DiskConfig,
},
ConfigMissing,
DistanceTooLarge,
@@ -43,22 +47,26 @@ pub enum Error {
ProposerKeyCorrupt {
length: usize,
},
IndexedAttestationKeyCorrupt {
IndexedAttestationIdKeyCorrupt {
length: usize,
},
IndexedAttestationIdCorrupt {
length: usize,
},
MissingIndexedAttestation {
root: Hash256,
id: u64,
},
MissingAttesterKey,
MissingProposerKey,
MissingIndexedAttestationKey,
AttesterRecordInconsistentRoot,
MissingIndexedAttestationId,
MissingIndexedAttestationIdKey,
InconsistentAttestationDataRoot,
}
impl From<lmdb::Error> for Error {
fn from(e: lmdb::Error) -> Self {
impl From<mdbx::Error> for Error {
fn from(e: mdbx::Error) -> Self {
match e {
lmdb::Error::Other(os_error) => Error::from(io::Error::from_raw_os_error(os_error)),
mdbx::Error::Other(os_error) => Error::from(io::Error::from_raw_os_error(os_error)),
_ => Error::DatabaseError(e),
}
}

8
slasher/src/lib.rs
View File

@@ -16,14 +16,18 @@ mod utils;
pub use crate::slasher::Slasher;
pub use attestation_queue::{AttestationBatch, AttestationQueue, SimpleBatch};
pub use attester_record::{AttesterRecord, IndexedAttesterRecord};
pub use attester_record::{AttesterRecord, CompactAttesterRecord, IndexedAttesterRecord};
pub use block_queue::BlockQueue;
pub use config::Config;
pub use database::SlasherDB;
pub use database::{IndexedAttestationId, SlasherDB};
pub use error::Error;
use types::{AttesterSlashing, EthSpec, IndexedAttestation, ProposerSlashing};
/// LMDB-to-MDBX compatibility shims.
pub type Environment = mdbx::Environment<mdbx::NoWriteMap>;
pub type RwTransaction<'env> = mdbx::Transaction<'env, mdbx::RW, mdbx::NoWriteMap>;
#[derive(Debug, PartialEq)]
pub enum AttesterSlashingStatus<E: EthSpec> {
NotSlashable,

15
slasher/src/metrics.rs
View File

@@ -4,7 +4,7 @@ pub use lighthouse_metrics::*;
lazy_static! {
pub static ref SLASHER_DATABASE_SIZE: Result<IntGauge> = try_create_int_gauge(
"slasher_database_size",
"Size of the LMDB database backing the slasher, in bytes"
"Size of the database backing the slasher, in bytes"
);
pub static ref SLASHER_RUN_TIME: Result<Histogram> = try_create_histogram(
"slasher_process_batch_time",
@@ -40,4 +40,17 @@ lazy_static! {
"slasher_compression_ratio",
"Compression ratio for min-max array chunks (higher is better)"
);
pub static ref SLASHER_NUM_ATTESTATION_ROOT_QUERIES: Result<IntCounter> =
try_create_int_counter(
"slasher_num_attestation_root_queries",
"Number of requests for an attestation data root",
);
pub static ref SLASHER_NUM_ATTESTATION_ROOT_HITS: Result<IntCounter> = try_create_int_counter(
"slasher_num_attestation_root_hits",
"Number of requests for an attestation data root that hit the LRU cache",
);
pub static ref SLASHER_ATTESTATION_ROOT_CACHE_SIZE: Result<IntGauge> = try_create_int_gauge(
"slasher_attestation_root_cache_size",
"Number of attestation data roots cached in memory"
);
}

94
slasher/src/migrate.rs
View File

@@ -1,79 +1,29 @@
use crate::{
config::{DEFAULT_BROADCAST, DEFAULT_SLOT_OFFSET},
database::CURRENT_SCHEMA_VERSION,
Config, Error, SlasherDB,
};
use lmdb::RwTransaction;
use serde_derive::{Deserialize, Serialize};
use std::path::PathBuf;
use crate::{database::CURRENT_SCHEMA_VERSION, Error, SlasherDB};
use types::EthSpec;
/// Config from schema version 1, for migration to version 2+.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConfigV1 {
database_path: PathBuf,
chunk_size: usize,
validator_chunk_size: usize,
history_length: usize,
update_period: u64,
max_db_size_mbs: usize,
}
type ConfigV2 = Config;
impl Into<ConfigV2> for ConfigV1 {
fn into(self) -> ConfigV2 {
Config {
database_path: self.database_path,
chunk_size: self.chunk_size,
validator_chunk_size: self.validator_chunk_size,
history_length: self.history_length,
update_period: self.update_period,
slot_offset: DEFAULT_SLOT_OFFSET,
max_db_size_mbs: self.max_db_size_mbs,
broadcast: DEFAULT_BROADCAST,
}
}
}
impl<E: EthSpec> SlasherDB<E> {
/// If the database exists, and has a schema, attempt to migrate it to the current version.
pub fn migrate(&self, txn: &mut RwTransaction<'_>) -> Result<(), Error> {
if let Some(schema_version) = self.load_schema_version(txn)? {
match (schema_version, CURRENT_SCHEMA_VERSION) {
// The migration from v1 to v2 is a bit messy because v1.0.5 silently
// changed the schema to v2, so a v1 schema could have either a v1 or v2
// config.
(1, 2) => {
match self.load_config::<ConfigV1>(txn) {
Ok(Some(config_v1)) => {
// Upgrade to v2 config and store on disk.
let config_v2 = config_v1.into();
self.store_config(&config_v2, txn)?;
}
Ok(None) => {
// Impossible to have schema version and no config.
return Err(Error::ConfigMissing);
}
Err(_) => {
// If loading v1 config failed, ensure loading v2 config succeeds.
// No further action is needed.
let _config_v2 = self.load_config::<ConfigV2>(txn)?;
}
}
}
(x, y) if x == y => {}
(_, _) => {
return Err(Error::IncompatibleSchemaVersion {
database_schema_version: schema_version,
software_schema_version: CURRENT_SCHEMA_VERSION,
});
}
}
}
pub fn migrate(self) -> Result<Self, Error> {
let mut txn = self.begin_rw_txn()?;
let schema_version = self.load_schema_version(&mut txn)?;
drop(txn);
// If the migration succeeded, update the schema version on-disk.
self.store_schema_version(txn)?;
Ok(())
if let Some(schema_version) = schema_version {
match (schema_version, CURRENT_SCHEMA_VERSION) {
// Schema v3 changed the underlying database from LMDB to MDBX. Unless the user did
// some manual hacking it should be impossible to read an MDBX schema version < 3.
(from, _) if from < 3 => Err(Error::IncompatibleSchemaVersion {
database_schema_version: schema_version,
software_schema_version: CURRENT_SCHEMA_VERSION,
}),
(x, y) if x == y => Ok(self),
(_, _) => Err(Error::IncompatibleSchemaVersion {
database_schema_version: schema_version,
software_schema_version: CURRENT_SCHEMA_VERSION,
}),
}
} else {
Ok(self)
}
}
}

29
slasher/src/slasher.rs
View File

@@ -6,9 +6,8 @@ use crate::metrics::{
};
use crate::{
array, AttestationBatch, AttestationQueue, AttesterRecord, BlockQueue, Config, Error,
ProposerSlashingStatus, SimpleBatch, SlasherDB,
IndexedAttestationId, ProposerSlashingStatus, RwTransaction, SimpleBatch, SlasherDB,
};
use lmdb::{RwTransaction, Transaction};
use parking_lot::Mutex;
use slog::{debug, error, info, Logger};
use std::collections::HashSet;
@@ -32,7 +31,7 @@ impl<E: EthSpec> Slasher<E> {
pub fn open(config: Config, log: Logger) -> Result<Self, Error> {
config.validate()?;
let config = Arc::new(config);
let db = SlasherDB::open(config.clone())?;
let db = SlasherDB::open(config.clone(), log.clone())?;
let attester_slashings = Mutex::new(HashSet::new());
let proposer_slashings = Mutex::new(HashSet::new());
let attestation_queue = AttestationQueue::default();
@@ -159,11 +158,19 @@ impl<E: EthSpec> Slasher<E> {
let mut num_stored = 0;
for weak_record in &batch.attestations {
if let Some(indexed_record) = weak_record.upgrade() {
self.db.store_indexed_attestation(
let indexed_attestation_id = self.db.store_indexed_attestation(
txn,
indexed_record.record.indexed_attestation_hash,
&indexed_record.indexed,
)?;
indexed_record.set_id(indexed_attestation_id);
// Prime the attestation data root LRU cache.
self.db.cache_attestation_data_root(
IndexedAttestationId::new(indexed_attestation_id),
indexed_record.record.attestation_data_hash,
);
num_stored += 1;
}
}
@@ -184,6 +191,12 @@ impl<E: EthSpec> Slasher<E> {
for (subqueue_id, subqueue) in grouped_attestations.subqueues.into_iter().enumerate() {
self.process_batch(txn, subqueue_id, subqueue, current_epoch)?;
}
metrics::set_gauge(
&metrics::SLASHER_ATTESTATION_ROOT_CACHE_SIZE,
self.db.attestation_root_cache_size() as i64,
);
Ok(AttestationStats { num_processed })
}
@@ -197,11 +210,13 @@ impl<E: EthSpec> Slasher<E> {
) -> Result<(), Error> {
// First, check for double votes.
for attestation in &batch {
let indexed_attestation_id = IndexedAttestationId::new(attestation.get_id());
match self.check_double_votes(
txn,
subqueue_id,
&attestation.indexed,
attestation.record,
&attestation.record,
indexed_attestation_id,
) {
Ok(slashings) => {
if !slashings.is_empty() {
@@ -262,7 +277,8 @@ impl<E: EthSpec> Slasher<E> {
txn: &mut RwTransaction<'_>,
subqueue_id: usize,
attestation: &IndexedAttestation<E>,
attester_record: AttesterRecord,
attester_record: &AttesterRecord,
indexed_attestation_id: IndexedAttestationId,
) -> Result<HashSet<AttesterSlashing<E>>, Error> {
let mut slashings = HashSet::new();
@@ -275,6 +291,7 @@ impl<E: EthSpec> Slasher<E> {
validator_index,
attestation,
attester_record,
indexed_attestation_id,
)?;
if let Some(slashing) = slashing_status.into_slashing(attestation) {

17
slasher/src/utils.rs
View File

@@ -1,20 +1,5 @@
use crate::Error;
/// Mix-in trait for loading values from LMDB that may or may not exist.
pub trait TxnOptional<T, E> {
fn optional(self) -> Result<Option<T>, E>;
}
impl<T> TxnOptional<T, Error> for Result<T, lmdb::Error> {
fn optional(self) -> Result<Option<T>, Error> {
match self {
Ok(x) => Ok(Some(x)),
Err(lmdb::Error::NotFound) => Ok(None),
Err(e) => Err(e.into()),
}
}
}
/// Transform a transaction that would fail with a `MapFull` error into an optional result.
pub trait TxnMapFull<T, E> {
fn allow_map_full(self) -> Result<Option<T>, E>;
@@ -24,7 +9,7 @@ impl<T> TxnMapFull<T, Error> for Result<T, Error> {
fn allow_map_full(self) -> Result<Option<T>, Error> {
match self {
Ok(x) => Ok(Some(x)),
Err(Error::DatabaseError(lmdb::Error::MapFull)) => Ok(None),
Err(Error::DatabaseError(mdbx::Error::MapFull)) => Ok(None),
Err(e) => Err(e),
}
}