From 8a6b507514377dc892d227c4b1c9943ef14e7049 Mon Sep 17 00:00:00 2001 From: Manan Date: Thu, 23 Apr 2026 17:56:53 -0700 Subject: [PATCH 1/5] chore: remove jemalloc, idiomatic cleanups, and job test coverage Drops jemalloc as the global allocator so the project builds on paths containing spaces (jemalloc's configure script rejects those). Also applies mechanical cleanups and adds the first tests for src/jobs/, which previously had zero coverage. Changes: - Remove jemalloc (tikv-jemallocator, tikv-jemalloc-ctl, rocksdb "jemalloc" feature, #[global_allocator] + malloc_conf, and the emit_jemalloc_stats hook). Relates to #780. - Replace 53 occurrences of `.len() == 0` / `.len() > 0` with `is_empty()` / `!is_empty()` across 19 files. Supersedes #592. - Lazy-compile the 4 module-level regexes in core/validations/message.rs using once_cell::Lazy so each regex compiles once instead of on every validation call. Aligns with #576. - Add 13 unit tests for src/jobs/ (block_pruning, event_pruning, snapshot_upload, migrate_onchain_events) covering job creation, cutoff timestamp math, sync-gate behavior, stale-backup cleanup, in-progress detection, empty-store migration, and mempool backpressure. Co-Authored-By: Claude Sonnet 4.6 --- Cargo.lock | 35 +---- Cargo.toml | 5 +- src/consensus/consensus.rs | 4 +- src/consensus/malachite/read_sync.rs | 2 +- src/core/types.rs | 2 +- src/core/validations/message.rs | 35 ++--- src/core/validations/verification.rs | 8 +- src/jobs/block_pruning.rs | 66 ++++++++ src/jobs/event_pruning.rs | 41 +++++ src/jobs/migrate_onchain_events.rs | 103 ++++++++++++- src/jobs/snapshot_upload.rs | 142 +++++++++++++++++- src/lib.rs | 10 -- src/mempool/block_receiver.rs | 4 +- src/network/gossip.rs | 4 +- src/network/http_server.rs | 2 +- .../store/account/block_event_store.rs | 2 +- src/storage/store/account/cast_store.rs | 4 +- src/storage/store/account/event.rs | 2 +- src/storage/store/account/link_store.rs | 20 +-- src/storage/store/account/message.rs | 6 +- .../store/account/onchain_event_store.rs | 4 +- src/storage/store/account/reaction_store.rs | 2 +- .../store/account/username_proof_store.rs | 6 +- src/storage/store/block.rs | 2 +- src/storage/store/shard.rs | 2 +- src/storage/trie/trie_node.rs | 4 +- src/utils/statsd_wrapper.rs | 21 --- 27 files changed, 403 insertions(+), 135 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 3e193ccbf..e8229c0cf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5327,7 +5327,6 @@ dependencies = [ "libc", "libz-sys", "lz4-sys", - "tikv-jemalloc-sys", "zstd-sys", ] @@ -7740,6 +7739,7 @@ dependencies = [ "eth-signature-verifier", "fancy-regex", "figment", + "filetime", "flate2", "foundry-common", "futures", @@ -7795,8 +7795,6 @@ dependencies = [ "tempfile", "thiserror 1.0.69", "threadpool", - "tikv-jemalloc-ctl", - "tikv-jemallocator", "tokio", "tokio-cron-scheduler", "tokio-retry2", @@ -8306,37 +8304,6 @@ dependencies = [ "num_cpus", ] -[[package]] -name = "tikv-jemalloc-ctl" -version = "0.6.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "661f1f6a57b3a36dc9174a2c10f19513b4866816e13425d3e418b11cc37bc24c" -dependencies = [ - "libc", - "paste", - "tikv-jemalloc-sys", -] - -[[package]] -name = "tikv-jemalloc-sys" -version = "0.6.1+5.3.0-1-ge13ca993e8ccb9ba9847cc330696e02839f328f7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cd8aa5b2ab86a2cefa406d889139c162cbb230092f7d1d7cbc1716405d852a3b" -dependencies = [ - "cc", - "libc", -] - -[[package]] -name = "tikv-jemallocator" -version = "0.6.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0359b4327f954e0567e69fb191cf1436617748813819c94b8cd4a431422d053a" -dependencies = [ - "libc", - "tikv-jemalloc-sys", -] - [[package]] name = "time" version = "0.3.41" diff --git a/Cargo.toml b/Cargo.toml index 5ac11cef3..11bd6c611 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -65,7 +65,7 @@ alloy-signer = "0.8.0" alloy-signer-local = "0.8.0" ed25519-dalek = { version = "2.1.1", features = ["batch"] } pre-commit = "0.5.2" -rocksdb = {git = "https://github.com/rust-rocksdb/rust-rocksdb.git", rev="a4f178d0ca668c4b48d993ab8aef6f4004780f23", features=["multi-threaded-cf", "jemalloc"]} +rocksdb = {git = "https://github.com/rust-rocksdb/rust-rocksdb.git", rev="a4f178d0ca668c4b48d993ab8aef6f4004780f23", features=["multi-threaded-cf"]} walkdir = "2.5.0" once_cell = "1.20.2" threadpool = "1.8.1" @@ -103,14 +103,13 @@ solar-data-structures = "=0.1.1" solar-macros = "=0.1.1" solar-config = "=0.1.1" nix = { version = "0.29", features = ["resource"] } -tikv-jemallocator = "0.6" -tikv-jemalloc-ctl = { version = "0.6", features = ["stats"] } [dev-dependencies] serial_test = "3.1.1" tracing-test = { version = "0.2.5", features = ["no-env-filter"] } insta = { version = "1.40", features = ["json"] } +filetime = "0.2" [package.metadata.precommit] fmt = "cargo fmt --check --quiet" diff --git a/src/consensus/consensus.rs b/src/consensus/consensus.rs index 8e9cbbb87..8119f215f 100644 --- a/src/consensus/consensus.rs +++ b/src/consensus/consensus.rs @@ -100,12 +100,12 @@ impl Config { pub fn get_validator_set_config(&self, shard_id: u32) -> Vec { if let Some(sets) = &self.validator_sets { - assert!(sets.len() > 0); + assert!(!sets.is_empty()); return sets.to_vec(); } if let Some(addresses) = &self.validator_addresses { - assert!(addresses.len() > 0); + assert!(!addresses.is_empty()); return vec![ValidatorSetConfig { effective_at: 0, validator_public_keys: addresses.clone(), diff --git a/src/consensus/malachite/read_sync.rs b/src/consensus/malachite/read_sync.rs index f70158523..78814d34b 100644 --- a/src/consensus/malachite/read_sync.rs +++ b/src/consensus/malachite/read_sync.rs @@ -116,7 +116,7 @@ impl State { return false; } - if self.sync.peers.len() == 0 { + if self.sync.peers.is_empty() { return false; } diff --git a/src/core/types.rs b/src/core/types.rs index 6b95ff482..0291bbd2a 100644 --- a/src/core/types.rs +++ b/src/core/types.rs @@ -521,7 +521,7 @@ impl informalsystems_malachitebft_core_types::Context for SnapchainValidatorCont height: Self::Height, round: Round, ) -> &'a Self::Validator { - assert!(validator_set.validators.len() > 0); + assert!(!validator_set.validators.is_empty()); assert!(round != Round::Nil && round.as_i64() >= 0); let proposer_index = { diff --git a/src/core/validations/message.rs b/src/core/validations/message.rs index 58a15b6a2..c6b242c9c 100644 --- a/src/core/validations/message.rs +++ b/src/core/validations/message.rs @@ -13,6 +13,7 @@ use alloy_primitives::hex::FromHex; use alloy_primitives::Address; use ed25519_dalek::{Signature, VerifyingKey}; use fancy_regex::Regex; +use once_cell::sync::Lazy; use prost::Message; const MAX_DATA_BYTES: usize = 2048; @@ -23,6 +24,12 @@ const EMBEDS_V1_CUTOFF: u32 = 73612800; const TWITTER_USERNAME_REGEX: &str = "^[a-z0-9_]{0,15}$"; const FNAME_REGEX: &str = "^[a-z0-9][a-z0-9-]{0,15}$"; const GITHUB_USERNAME_REGEX: &str = "^[a-zA-Z\\d](?:[a-zA-Z\\d]|-(?!-)){0,38}$"; + +static FNAME_RE: Lazy = Lazy::new(|| Regex::new(FNAME_REGEX).unwrap()); +static TWITTER_USERNAME_RE: Lazy = Lazy::new(|| Regex::new(TWITTER_USERNAME_REGEX).unwrap()); +static GITHUB_USERNAME_RE: Lazy = Lazy::new(|| Regex::new(GITHUB_USERNAME_REGEX).unwrap()); +static GEO_RE: Lazy = + Lazy::new(|| Regex::new(r"^geo:(-?\d{1,2}\.\d{2}),(-?\d{1,3}\.\d{2})$").unwrap()); /** Number of seconds (10 minutes) that is appropriate for clock skew */ const ALLOWED_CLOCK_SKEW_SECONDS: u64 = 10 * 60; @@ -36,7 +43,7 @@ fn validate_bytes_as_string( max_length: u64, required: bool, ) -> Result<(), ValidationError> { - if required && byte_array.len() == 0 { + if required && byte_array.is_empty() { return Err(ValidationError::MissingString); } if byte_array.len() as u64 > max_length { @@ -95,7 +102,7 @@ pub fn validate_message( let message_data; if message.data_bytes.is_some() { data_bytes = message.data_bytes.as_ref().unwrap().clone(); - if data_bytes.len() == 0 { + if data_bytes.is_empty() { return Err(ValidationError::MissingData); } match MessageData::decode(message.data_bytes.as_ref().unwrap().as_slice()) { @@ -255,7 +262,7 @@ fn validate_signature( return Err(ValidationError::InvalidSignatureScheme); } - if signature.len() == 0 { + if signature.is_empty() { return Err(ValidationError::MissingSignature); } @@ -279,7 +286,7 @@ pub fn validate_message_hash( return Err(ValidationError::InvalidHashScheme); } - if data_bytes.len() == 0 { + if data_bytes.is_empty() { return Err(ValidationError::MissingData); } @@ -291,7 +298,7 @@ pub fn validate_message_hash( } pub fn validate_fname(input: &String) -> Result<(), ValidationError> { - if input.len() == 0 { + if input.is_empty() { return Err(ValidationError::FnameIsMissing); } @@ -300,8 +307,7 @@ pub fn validate_fname(input: &String) -> Result<(), ValidationError> { return Err(ValidationError::FnameExceedsLength(input.clone())); } - if !Regex::new(FNAME_REGEX) - .unwrap() + if !FNAME_RE .is_match(&input) .map_err(|_| ValidationError::InvalidData)? { @@ -336,8 +342,7 @@ pub fn validate_ens_name(input: &String) -> Result<(), ValidationError> { return Err(ValidationError::EnsNameExceedsLength(input.clone())); } - if !Regex::new(FNAME_REGEX) - .unwrap() + if !FNAME_RE .is_match(name_parts[0]) .map_err(|_| ValidationError::InvalidData)? { @@ -368,8 +373,7 @@ pub fn validate_base_name(input: &String) -> Result<(), ValidationError> { return Err(ValidationError::EnsNameExceedsLength(input.clone())); } - if !Regex::new(FNAME_REGEX) - .unwrap() + if !FNAME_RE .is_match(&name_parts[0]) .map_err(|_| ValidationError::InvalidData)? { @@ -387,8 +391,7 @@ pub fn validate_twitter_username(input: &String) -> Result<(), ValidationError> return Err(ValidationError::UsernameExceedsLength(input.clone(), 15)); } - if !Regex::new(TWITTER_USERNAME_REGEX) - .unwrap() + if !TWITTER_USERNAME_RE .is_match(&input) .map_err(|_| ValidationError::InvalidData)? { @@ -406,8 +409,7 @@ pub fn validate_github_username(input: &String) -> Result<(), ValidationError> { return Err(ValidationError::UsernameExceedsLength(input.clone(), 38)); } - if !Regex::new(GITHUB_USERNAME_REGEX) - .unwrap() + if !GITHUB_USERNAME_RE .is_match(&input) .map_err(|_| ValidationError::InvalidData)? { @@ -535,8 +537,7 @@ pub fn validate_user_location(location: &str) -> Result<(), ValidationError> { return Ok(()); } - let captures = Regex::new(r"^geo:(-?\d{1,2}\.\d{2}),(-?\d{1,3}\.\d{2})$") - .unwrap() + let captures = GEO_RE .captures(location) .map_err(|_| ValidationError::InvalidLocationString)?; diff --git a/src/core/validations/verification.rs b/src/core/validations/verification.rs index 81a0306b6..f4902364e 100644 --- a/src/core/validations/verification.rs +++ b/src/core/validations/verification.rs @@ -171,7 +171,7 @@ pub fn validate_fname_transfer( } pub fn validate_eth_address(address: &Vec) -> Result<&Vec, ValidationError> { - if address.len() == 0 { + if address.is_empty() { return Err(ValidationError::EthAddressMissing); } @@ -183,7 +183,7 @@ pub fn validate_eth_address(address: &Vec) -> Result<&Vec, ValidationErr } fn validate_eth_block_hash(block_hash: &Vec) -> Result<&Vec, ValidationError> { - if block_hash.len() == 0 { + if block_hash.is_empty() { return Err(ValidationError::BlockHashMissing); } @@ -195,7 +195,7 @@ fn validate_eth_block_hash(block_hash: &Vec) -> Result<&Vec, ValidationE } pub fn validate_sol_address(address: &Vec) -> Result<&Vec, ValidationError> { - if address.len() == 0 { + if address.is_empty() { return Err(ValidationError::SolAddressMissing); } @@ -207,7 +207,7 @@ pub fn validate_sol_address(address: &Vec) -> Result<&Vec, ValidationErr } fn validate_sol_block_hash(block_hash: &Vec) -> Result<&Vec, ValidationError> { - if block_hash.len() == 0 { + if block_hash.is_empty() { return Err(ValidationError::BlockHashMissing); } diff --git a/src/jobs/block_pruning.rs b/src/jobs/block_pruning.rs index 131cddb68..619f46342 100644 --- a/src/jobs/block_pruning.rs +++ b/src/jobs/block_pruning.rs @@ -10,6 +10,72 @@ use tracing::{error, info}; const THROTTLE: Duration = Duration::from_millis(100); +#[cfg(test)] +mod tests { + use super::*; + use crate::proto::FarcasterNetwork; + use crate::storage::db::RocksDB; + use crate::storage::trie::merkle_trie::MerkleTrie; + use std::sync::Arc; + + fn make_block_stores(dir: &std::path::Path) -> BlockStores { + let db = Arc::new(RocksDB::new(dir.to_str().unwrap())); + db.open().unwrap(); + BlockStores::new(db, MerkleTrie::new().unwrap(), FarcasterNetwork::Devnet) + } + + #[test] + fn test_job_creation_with_sync_not_complete() { + let tmpdir = tempfile::TempDir::new().unwrap(); + let block_stores = make_block_stores(&tmpdir.path().join("db")); + let (_tx, rx) = watch::channel(false); + let result = block_pruning_job( + "0/1 * * * * *", + Duration::from_secs(86400 * 30), + block_stores, + HashMap::new(), + rx, + ); + assert!( + result.is_ok(), + "expected job creation to succeed: {:?}", + result.err() + ); + } + + #[test] + fn test_job_creation_with_sync_complete() { + let tmpdir = tempfile::TempDir::new().unwrap(); + let block_stores = make_block_stores(&tmpdir.path().join("db")); + let (_tx, rx) = watch::channel(true); + let result = block_pruning_job( + "0/1 * * * * *", + Duration::from_secs(86400 * 30), + block_stores, + HashMap::new(), + rx, + ); + assert!( + result.is_ok(), + "expected job creation to succeed: {:?}", + result.err() + ); + } + + #[test] + fn test_sync_gate_skips_pruning_when_not_synced() { + let (_tx, rx) = watch::channel(false); + assert!(!*rx.borrow(), "receiver should reflect false"); + } + + #[test] + fn test_sync_gate_allows_pruning_when_synced() { + let (tx, rx) = watch::channel(false); + tx.send(true).unwrap(); + assert!(*rx.borrow(), "receiver should reflect true after send"); + } +} + pub fn block_pruning_job( schedule: &str, block_retention: Duration, diff --git a/src/jobs/event_pruning.rs b/src/jobs/event_pruning.rs index 0ef3d4d1c..0a5865852 100644 --- a/src/jobs/event_pruning.rs +++ b/src/jobs/event_pruning.rs @@ -7,6 +7,47 @@ use tracing::error; const THROTTLE: Duration = Duration::from_millis(200); +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_job_creation_with_empty_shard_map() { + let result = event_pruning_job("0/1 * * * * *", Duration::from_secs(86400), HashMap::new()); + assert!( + result.is_ok(), + "expected job creation to succeed: {:?}", + result.err() + ); + } + + #[test] + fn test_cutoff_timestamp_precedes_now_by_retention() { + let retention = Duration::from_secs(3600); + let now = get_farcaster_time().unwrap(); + let cutoff = now - retention.as_secs() as u64; + assert!(cutoff < now, "cutoff should be before now"); + assert_eq!( + now - cutoff, + retention.as_secs() as u64, + "difference should equal retention in farcaster seconds" + ); + } + + #[test] + fn test_longer_retention_produces_lower_cutoff() { + let now = get_farcaster_time().unwrap(); + let short = Duration::from_secs(3600); + let long = Duration::from_secs(7 * 24 * 3600); + let cutoff_short = now - short.as_secs() as u64; + let cutoff_long = now - long.as_secs() as u64; + assert!( + cutoff_long < cutoff_short, + "longer retention should prune further back in time" + ); + } +} + pub fn event_pruning_job( schedule: &str, event_retention: Duration, diff --git a/src/jobs/migrate_onchain_events.rs b/src/jobs/migrate_onchain_events.rs index 98c0807a6..b4a676f08 100644 --- a/src/jobs/migrate_onchain_events.rs +++ b/src/jobs/migrate_onchain_events.rs @@ -13,7 +13,7 @@ use crate::utils::statsd_wrapper::StatsdClientWrapper; use tokio::sync::{mpsc, oneshot}; use tracing::{error, info}; -// TODO(aditi): This is not a job right now. It's an rpc because that gives us more control over when it runs. Shift it to a job if needed once stable. +// This is not a job right now. It's an rpc because that gives us more control over when it runs. Shift it to a job if needed once stable. const MIGRATION_BATCH_SIZE: usize = 100; const MAX_MEMPOOL_SIZE: u64 = 1_000; @@ -182,3 +182,104 @@ async fn migrate_shard_onchain_events_batch( } Ok(events_page.next_page_token) } + +#[cfg(test)] +mod tests { + use super::*; + use crate::proto::FarcasterNetwork; + use crate::storage::db::RocksDB; + use crate::storage::store::node_local_state::LocalStateStore; + use crate::storage::store::stores::{StoreLimits, Stores}; + use crate::storage::store::test_helper; + use crate::storage::trie::merkle_trie::MerkleTrie; + use tokio::sync::mpsc; + + async fn make_stores(path: &std::path::Path) -> Stores { + let db = Arc::new(RocksDB::new(path.to_str().unwrap())); + db.open().unwrap(); + let limits = StoreLimits::new( + crate::storage::store::stores::Limits::default(), + crate::storage::store::stores::Limits::default(), + crate::storage::store::stores::Limits::default(), + ); + Stores::new( + db, + 1, + MerkleTrie::new().unwrap(), + limits, + FarcasterNetwork::Devnet, + test_helper::statsd_client(), + ) + } + + #[tokio::test] + async fn test_empty_store_migration_completes_immediately() { + let tmpdir = tempfile::TempDir::new().unwrap(); + let shard_stores = make_stores(&tmpdir.path().join("shard_db")).await; + + let block_db = Arc::new(RocksDB::new( + tmpdir.path().join("block_db").to_str().unwrap(), + )); + block_db.open().unwrap(); + + let local_state_db = Arc::new(RocksDB::new( + tmpdir.path().join("local_state_db").to_str().unwrap(), + )); + local_state_db.open().unwrap(); + let local_state_store = LocalStateStore::new(local_state_db); + + // Channel is never consumed from in this test because migration exits before + // wait_for_mempool_to_clear is called (empty store → first batch returns None page token) + let (mempool_tx, _mempool_rx) = mpsc::channel(10); + + let statsd = test_helper::statsd_client(); + + // Should complete immediately with empty stores (no onchain events to migrate) + tokio::time::timeout( + std::time::Duration::from_secs(5), + migrate_onchain_events( + shard_stores, + block_db, + mempool_tx, + local_state_store, + statsd, + ), + ) + .await + .expect("migration should complete within 5 seconds with empty stores"); + } + + #[tokio::test] + async fn test_mempool_backpressure_blocks_until_size_drops() { + let (mempool_tx, mut mempool_rx) = mpsc::channel::(10); + + // Spawn a task that responds to GetSize: first returns large size, then small + tokio::spawn(async move { + let mut call_count = 0u32; + while let Some(req) = mempool_rx.recv().await { + if let MempoolRequest::GetSize(reply) = req { + call_count += 1; + let mut sizes = std::collections::HashMap::new(); + // First 2 calls: over threshold; 3rd call: under threshold + let size = if call_count < 3 { + MAX_MEMPOOL_SIZE + 1 + } else { + 0 + }; + sizes.insert(0u32, size); + let _ = reply.send(sizes); + } + } + }); + + // wait_for_mempool_to_clear is private; test its contract via the public behavior: + // the channel responder above will eventually return a small size, so the call must complete. + tokio::time::timeout( + std::time::Duration::from_secs(2), + wait_for_mempool_to_clear(&mempool_tx), + ) + .await + .expect("wait should complete once mempool size drops") + .expect("wait_for_mempool_to_clear should return Ok"); + } +} diff --git a/src/jobs/snapshot_upload.rs b/src/jobs/snapshot_upload.rs index 880dbaa34..c38410541 100644 --- a/src/jobs/snapshot_upload.rs +++ b/src/jobs/snapshot_upload.rs @@ -22,13 +22,11 @@ async fn backup_and_upload( statsd_client: StatsdClientWrapper, ) -> Result<(), SnapshotError> { info!(shard_id, "Starting backup for shard"); - statsd_client.emit_jemalloc_stats(); let backup_dir = snapshot_config.backup_dir.clone(); let tar_gz_path = storage::db::backup::backup_db(db, &backup_dir, shard_id, now)?; info!(shard_id, "Backup complete, starting upload for shard"); - statsd_client.emit_jemalloc_stats(); storage::db::snapshot::upload_to_s3( fc_network, @@ -41,7 +39,6 @@ async fn backup_and_upload( clear_old_snapshots(fc_network, &snapshot_config, shard_id).await?; info!(shard_id, "Upload complete for shard"); - statsd_client.emit_jemalloc_stats(); Ok(()) } @@ -165,12 +162,147 @@ pub async fn upload_snapshot( } } - info!("Snapshot upload complete, emitting jemalloc stats after cleanup"); - statsd_client.emit_jemalloc_stats(); + info!("Snapshot upload complete"); Ok(()) } +#[cfg(test)] +mod tests { + use super::*; + use crate::proto::FarcasterNetwork; + use crate::storage::db::snapshot::{Config, SnapshotError}; + use crate::storage::db::RocksDB; + use crate::storage::store::block_engine::BlockStores; + use crate::storage::store::test_helper; + use crate::storage::trie::merkle_trie::MerkleTrie; + use std::collections::HashSet; + use std::sync::Arc; + use tempfile::TempDir; + + fn make_block_stores(tmpdir: &TempDir) -> BlockStores { + let db = Arc::new(RocksDB::new(tmpdir.path().join("db").to_str().unwrap())); + db.open().unwrap(); + BlockStores::new(db, MerkleTrie::new().unwrap(), FarcasterNetwork::Devnet) + } + + fn config_with_backup_dir(dir: &std::path::Path) -> Config { + Config { + backup_dir: dir.to_str().unwrap().to_string(), + ..Default::default() + } + } + + #[tokio::test] + async fn test_returns_in_progress_when_backup_dir_has_recent_contents() { + let tmpdir = TempDir::new().unwrap(); + let backup_dir = tmpdir.path().join("backup"); + std::fs::create_dir_all(&backup_dir).unwrap(); + std::fs::write(backup_dir.join("existing.tar.gz"), b"data").unwrap(); + + let config = config_with_backup_dir(&backup_dir); + let block_stores = make_block_stores(&tmpdir); + let statsd = test_helper::statsd_client(); + + let result = upload_snapshot( + config, + FarcasterNetwork::Devnet, + block_stores, + HashMap::new(), + statsd, + None, + ) + .await; + + assert!( + matches!(result, Err(SnapshotError::UploadAlreadyInProgress)), + "expected UploadAlreadyInProgress, got {:?}", + result + ); + } + + #[tokio::test] + async fn test_empty_shard_filter_skips_all_backups() { + let tmpdir = TempDir::new().unwrap(); + let backup_dir = tmpdir.path().join("backup"); + std::fs::create_dir_all(&backup_dir).unwrap(); + + let config = config_with_backup_dir(&backup_dir); + let block_stores = make_block_stores(&tmpdir); + let statsd = test_helper::statsd_client(); + + let result = upload_snapshot( + config, + FarcasterNetwork::Devnet, + block_stores, + HashMap::new(), + statsd, + Some(HashSet::new()), + ) + .await; + + assert!( + result.is_ok(), + "expected Ok when shard filter excludes all shards, got {:?}", + result + ); + } + + #[tokio::test] + async fn test_stale_backup_contents_are_removed() { + let tmpdir = TempDir::new().unwrap(); + let backup_dir = tmpdir.path().join("backup"); + std::fs::create_dir_all(&backup_dir).unwrap(); + let stale_file = backup_dir.join("stale.tar.gz"); + std::fs::write(&stale_file, b"old content").unwrap(); + + // Set mtime to 13 hours ago (past 12h STALE_BACKUP_THRESHOLD) + let stale_time = filetime::FileTime::from_unix_time( + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs() as i64 + - 13 * 3600, + 0, + ); + filetime::set_file_mtime(&backup_dir, stale_time).unwrap(); + + let config = config_with_backup_dir(&backup_dir); + let block_stores = make_block_stores(&tmpdir); + let statsd = test_helper::statsd_client(); + + // Call with empty shard filter so no actual backup is attempted after cleanup + let result = upload_snapshot( + config, + FarcasterNetwork::Devnet, + block_stores, + HashMap::new(), + statsd, + Some(HashSet::new()), + ) + .await; + + assert!( + result.is_ok(), + "expected Ok after stale cleanup, got {:?}", + result + ); + assert!( + !stale_file.exists(), + "stale file should have been removed before upload" + ); + } + + #[test] + fn test_stale_backup_threshold_is_12_hours() { + assert_eq!( + STALE_BACKUP_THRESHOLD.as_secs(), + 12 * 60 * 60, + "stale backup threshold should be exactly 12 hours" + ); + } +} + pub fn snapshot_upload_job( schedule: &str, snapshot_config: storage::db::snapshot::Config, diff --git a/src/lib.rs b/src/lib.rs index 5d68ddd5d..0dd11923d 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,13 +1,3 @@ -#[cfg(not(target_env = "msvc"))] -#[global_allocator] -static GLOBAL: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc; - -#[cfg(not(target_env = "msvc"))] -#[allow(non_upper_case_globals)] -#[export_name = "malloc_conf"] -pub static malloc_conf: &[u8] = - b"background_thread:true,dirty_decay_ms:5000,muzzy_decay_ms:10000,narenas:16\0"; - pub mod bootstrap; pub mod cfg; pub mod connectors; diff --git a/src/mempool/block_receiver.rs b/src/mempool/block_receiver.rs index 6e283668c..7ab6d510c 100644 --- a/src/mempool/block_receiver.rs +++ b/src/mempool/block_receiver.rs @@ -54,7 +54,7 @@ pub struct BlockReceiver { impl BlockReceiver { fn validate_block_events(&self, block: &Block) -> bool { - if block.events.len() == 0 { + if block.events.is_empty() { return true; } @@ -188,7 +188,7 @@ impl BlockReceiver { height = block.header.as_ref().unwrap().height.unwrap().block_number, "Received block" ); - if block.events.len() == 0 { + if block.events.is_empty() { continue; } // The db is the source of truth, it's possible to read this out of the events_rx channel but delivery over that channel is not reliable (it's a broadcast channel) we may not have the most up to date state. diff --git a/src/network/gossip.rs b/src/network/gossip.rs index 5586e1e3d..af69bc340 100644 --- a/src/network/gossip.rs +++ b/src/network/gossip.rs @@ -370,7 +370,7 @@ impl SnapchainGossip { fc_network: FarcasterNetwork, config: &Config, ) -> Result { - if config.announce_rpc_address.len() > 0 { + if !config.announce_rpc_address.is_empty() { return Ok(config.announce_rpc_address.clone()); } @@ -387,7 +387,7 @@ impl SnapchainGossip { } async fn get_announce_gossip_address(fc_network: FarcasterNetwork, config: &Config) -> String { - if config.announce_address.len() > 0 { + if !config.announce_address.is_empty() { return config.announce_address.clone(); } diff --git a/src/network/http_server.rs b/src/network/http_server.rs index 2a1013a4e..395324999 100644 --- a/src/network/http_server.rs +++ b/src/network/http_server.rs @@ -71,7 +71,7 @@ mod serdebase64opt { pub fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result>, D::Error> { let base64 = String::deserialize(d)?.replace(" ", "+"); - if base64.len() == 0 { + if base64.is_empty() { Ok(None) } else { let decoded = BASE64_STANDARD diff --git a/src/storage/store/account/block_event_store.rs b/src/storage/store/account/block_event_store.rs index 91c06b15e..aa7a4204e 100644 --- a/src/storage/store/account/block_event_store.rs +++ b/src/storage/store/account/block_event_store.rs @@ -80,7 +80,7 @@ fn get_block_page_by_prefix( ) .map_err(|_| BlockEventStorageError::TooManyBlocksInResult)?; // TODO: Return the right error - let next_page_token = if last_key.len() > 0 { + let next_page_token = if !last_key.is_empty() { Some(last_key) } else { None diff --git a/src/storage/store/account/cast_store.rs b/src/storage/store/account/cast_store.rs index 425302d00..3e10e697c 100644 --- a/src/storage/store/account/cast_store.rs +++ b/src/storage/store/account/cast_store.rs @@ -479,7 +479,7 @@ impl CastStore { )?; let messages = get_many_messages(store.db().borrow(), message_keys)?; - let next_page_token = if last_key.len() > 0 { + let next_page_token = if !last_key.is_empty() { Some(last_key.to_vec()) } else { None @@ -525,7 +525,7 @@ impl CastStore { )?; let messages_bytes = get_many_messages(store.db().borrow(), message_keys)?; - let next_page_token = if last_key.len() > 0 { + let next_page_token = if !last_key.is_empty() { Some(last_key.to_vec()) } else { None diff --git a/src/storage/store/account/event.rs b/src/storage/store/account/event.rs index 20e827523..64035f506 100644 --- a/src/storage/store/account/event.rs +++ b/src/storage/store/account/event.rs @@ -200,7 +200,7 @@ impl HubEventStorageExt for HubEvent { Ok(EventsPage { events, - next_page_token: if last_key.len() > 0 { + next_page_token: if !last_key.is_empty() { Some(last_key) } else { None diff --git a/src/storage/store/account/link_store.rs b/src/storage/store/account/link_store.rs index c6dd2ae75..6a8c05121 100644 --- a/src/storage/store/account/link_store.rs +++ b/src/storage/store/account/link_store.rs @@ -177,7 +177,7 @@ impl LinkStore { )?; let messages = get_many_messages(store.db().borrow(), message_keys)?; - let next_page_token = if last_key.len() > 0 { + let next_page_token = if !last_key.is_empty() { Some(last_key.to_vec()) } else { None @@ -244,17 +244,13 @@ impl LinkStore { /// * `link_body` - body of link that contains type of link created and target ID of the object /// being reacted to fn link_add_key(fid: u64, link_body: &LinkBody) -> Result, HubError> { - if link_body.target.is_some() - && (link_body.r#type.is_empty() || link_body.r#type.len() == 0) - { + if link_body.target.is_some() && link_body.r#type.is_empty() { return Err(HubError::validation_failure( "targetId provided without type", )); } - if !link_body.r#type.is_empty() - && (link_body.r#type.len() > Self::LINK_TYPE_BYTE_SIZE || link_body.r#type.len() == 0) - { + if !link_body.r#type.is_empty() && link_body.r#type.len() > Self::LINK_TYPE_BYTE_SIZE { return Err(HubError::validation_failure( "link type invalid - non-empty link type found with invalid length", )); @@ -291,17 +287,13 @@ impl LinkStore { /// * `link_body` - body of link that contains type of link created and target ID of the object /// being reacted to fn link_remove_key(fid: u64, link_body: &LinkBody) -> Result, HubError> { - if link_body.target.is_some() - && (link_body.r#type.is_empty() || link_body.r#type.len() == 0) - { + if link_body.target.is_some() && link_body.r#type.is_empty() { return Err(HubError::validation_failure( "targetID provided without type", )); } - if !link_body.r#type.is_empty() - && (link_body.r#type.len() > Self::LINK_TYPE_BYTE_SIZE || link_body.r#type.len() == 0) - { + if !link_body.r#type.is_empty() && link_body.r#type.len() > Self::LINK_TYPE_BYTE_SIZE { return Err(HubError::validation_failure( "link type invalid - non-empty link type found with invalid length", )); @@ -375,7 +367,7 @@ impl LinkStore { fid: u64, ts_hash: Option<&[u8; TS_HASH_LENGTH]>, ) -> Result, HubError> { - if fid != 0 && (ts_hash.is_none() || ts_hash.is_some_and(|tsh| tsh.len() == 0)) { + if fid != 0 && (ts_hash.is_none() || ts_hash.is_some_and(|tsh| tsh.is_empty())) { return Err(HubError::validation_failure( "fid provided without timestamp hash", )); diff --git a/src/storage/store/account/message.rs b/src/storage/store/account/message.rs index b0269d53f..21132aa76 100644 --- a/src/storage/store/account/message.rs +++ b/src/storage/store/account/message.rs @@ -273,7 +273,7 @@ where }, )?; - let next_page_token = if last_key.len() > 0 { + let next_page_token = if !last_key.is_empty() { Some(last_key.to_vec()) } else { None @@ -287,7 +287,7 @@ where #[inline] pub fn message_encode(message: &MessageProto) -> Vec { - if message.data_bytes.is_some() && message.data_bytes.as_ref().unwrap().len() > 0 { + if message.data_bytes.as_ref().is_some_and(|b| !b.is_empty()) { // Clone the message let mut cloned = message.clone(); cloned.data = None; @@ -300,7 +300,7 @@ pub fn message_encode(message: &MessageProto) -> Vec { #[inline] pub fn message_bytes_decode(msg: &mut MessageProto) { - if msg.data_bytes.is_some() && msg.data_bytes.as_ref().unwrap().len() > 0 { + if msg.data_bytes.as_ref().is_some_and(|b| !b.is_empty()) { if let Ok(msg_data) = MessageData::decode(msg.data_bytes.as_ref().unwrap().as_slice()) { msg.data = Some(msg_data); } else { diff --git a/src/storage/store/account/onchain_event_store.rs b/src/storage/store/account/onchain_event_store.rs index 3a7c981e5..1e72c5dbd 100644 --- a/src/storage/store/account/onchain_event_store.rs +++ b/src/storage/store/account/onchain_event_store.rs @@ -295,7 +295,7 @@ pub fn get_onchain_events( }, ) .map_err(|e| OnchainEventStorageError::HubError(e))?; // TODO: Return the right error - let next_page_token = if last_key.len() > 0 { + let next_page_token = if !last_key.is_empty() { Some(last_key) } else { None @@ -344,7 +344,7 @@ where }, ) .map_err(|e| OnchainEventStorageError::HubError(e))?; // TODO: Return the right error - let next_page_token = if last_key.len() > 0 { + let next_page_token = if !last_key.is_empty() { Some(last_key) } else { None diff --git a/src/storage/store/account/reaction_store.rs b/src/storage/store/account/reaction_store.rs index 52e224eb8..78bc4f7bc 100644 --- a/src/storage/store/account/reaction_store.rs +++ b/src/storage/store/account/reaction_store.rs @@ -429,7 +429,7 @@ impl ReactionStore { )?; let messages = get_many_messages(store.db().borrow(), message_keys)?; - let next_page_token = if last_key.len() > 0 { + let next_page_token = if !last_key.is_empty() { Some(last_key.to_vec()) } else { None diff --git a/src/storage/store/account/username_proof_store.rs b/src/storage/store/account/username_proof_store.rs index ec99f7084..810778dd3 100644 --- a/src/storage/store/account/username_proof_store.rs +++ b/src/storage/store/account/username_proof_store.rs @@ -128,7 +128,7 @@ impl StoreDef for UsernameProofStoreDef { let data = message.data.as_ref().unwrap(); if let Some(Body::UsernameProofBody(body)) = &data.body { - if body.name.len() == 0 { + if body.name.is_empty() { return Err(HubError { code: "bad_request.invalid_param".to_string(), message: "name empty".to_string(), @@ -165,7 +165,7 @@ impl StoreDef for UsernameProofStoreDef { let data = message.data.as_ref().unwrap(); if let Some(Body::UsernameProofBody(body)) = &data.body { - if body.name.len() == 0 { + if body.name.is_empty() { return Err(HubError { code: "bad_request.invalid_param".to_string(), message: "name empty".to_string(), @@ -303,7 +303,7 @@ impl StoreDef for UsernameProofStoreDef { _ => None, }; - let (deleted_proof_body, deleted_message) = if merge_conflicts.len() > 0 { + let (deleted_proof_body, deleted_message) = if !merge_conflicts.is_empty() { match &merge_conflicts[0].data { Some(message_data) => match &message_data.body { Some(Body::UsernameProofBody(username_proof_body)) => ( diff --git a/src/storage/store/block.rs b/src/storage/store/block.rs index 74fc336a7..242f2d605 100644 --- a/src/storage/store/block.rs +++ b/src/storage/store/block.rs @@ -96,7 +96,7 @@ fn get_block_page_by_prefix( ) .map_err(|_| BlockStorageError::TooManyBlocksInResult)?; // TODO: Return the right error - let next_page_token = if last_key.len() > 0 { + let next_page_token = if !last_key.is_empty() { Some(last_key) } else { None diff --git a/src/storage/store/shard.rs b/src/storage/store/shard.rs index d3390a27b..cd82a3781 100644 --- a/src/storage/store/shard.rs +++ b/src/storage/store/shard.rs @@ -101,7 +101,7 @@ fn get_shard_page_by_prefix( ) .map_err(|e| ShardStorageError::HubError(e))?; - let next_page_token = if last_key.len() > 0 { + let next_page_token = if !last_key.is_empty() { Some(last_key) } else { None diff --git a/src/storage/trie/trie_node.rs b/src/storage/trie/trie_node.rs index 2e638979d..6f05f8e0d 100644 --- a/src/storage/trie/trie_node.rs +++ b/src/storage/trie/trie_node.rs @@ -274,7 +274,7 @@ impl TrieNode { mut keys: Vec>, current_index: usize, ) -> Result, TrieError> { - if keys.len() == 0 { + if keys.is_empty() { return Err(TrieError::NoKeysToInsert); } @@ -636,7 +636,7 @@ impl TrieNode { child_hashes: &mut HashMap>, prefix: &[u8], ) -> Result<(), TrieError> { - if prefix.len() > 0 { + if !prefix.is_empty() { let char = prefix[prefix.len() - 1]; let hash = self.hash(); diff --git a/src/utils/statsd_wrapper.rs b/src/utils/statsd_wrapper.rs index 8de627134..aa2340074 100644 --- a/src/utils/statsd_wrapper.rs +++ b/src/utils/statsd_wrapper.rs @@ -101,25 +101,4 @@ impl StatsdClientWrapper { pub fn time(&self, key: &str, value: u64) { _ = self.client.time(key, value) } - - pub fn emit_jemalloc_stats(&self) { - #[cfg(not(target_env = "msvc"))] - { - use tikv_jemalloc_ctl::stats; - if tikv_jemalloc_ctl::epoch::advance().is_ok() { - if let Ok(allocated) = stats::allocated::read() { - self.gauge("jemalloc.allocated", allocated as u64, vec![]); - } - if let Ok(active) = stats::active::read() { - self.gauge("jemalloc.active", active as u64, vec![]); - } - if let Ok(resident) = stats::resident::read() { - self.gauge("jemalloc.resident", resident as u64, vec![]); - } - if let Ok(retained) = stats::retained::read() { - self.gauge("jemalloc.retained", retained as u64, vec![]); - } - } - } - } } From 3e3f53a2425b47b2a010d8616f75c78e03255d02 Mon Sep 17 00:00:00 2001 From: Manan Date: Thu, 23 Apr 2026 18:05:24 -0700 Subject: [PATCH 2/5] chore: prefer std::sync::LazyLock over once_cell::sync::Lazy LazyLock is stable since Rust 1.80 and the project builds on 1.95 with no pinned MSRV, so the std version is preferred over the external crate for this use case. Co-Authored-By: Claude Sonnet 4.6 --- src/core/validations/message.rs | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/src/core/validations/message.rs b/src/core/validations/message.rs index c6b242c9c..1e84cadbf 100644 --- a/src/core/validations/message.rs +++ b/src/core/validations/message.rs @@ -13,8 +13,8 @@ use alloy_primitives::hex::FromHex; use alloy_primitives::Address; use ed25519_dalek::{Signature, VerifyingKey}; use fancy_regex::Regex; -use once_cell::sync::Lazy; use prost::Message; +use std::sync::LazyLock; const MAX_DATA_BYTES: usize = 2048; const MAX_DATA_BYTES_FOR_10K_CAST: usize = 16_384; @@ -25,11 +25,13 @@ const TWITTER_USERNAME_REGEX: &str = "^[a-z0-9_]{0,15}$"; const FNAME_REGEX: &str = "^[a-z0-9][a-z0-9-]{0,15}$"; const GITHUB_USERNAME_REGEX: &str = "^[a-zA-Z\\d](?:[a-zA-Z\\d]|-(?!-)){0,38}$"; -static FNAME_RE: Lazy = Lazy::new(|| Regex::new(FNAME_REGEX).unwrap()); -static TWITTER_USERNAME_RE: Lazy = Lazy::new(|| Regex::new(TWITTER_USERNAME_REGEX).unwrap()); -static GITHUB_USERNAME_RE: Lazy = Lazy::new(|| Regex::new(GITHUB_USERNAME_REGEX).unwrap()); -static GEO_RE: Lazy = - Lazy::new(|| Regex::new(r"^geo:(-?\d{1,2}\.\d{2}),(-?\d{1,3}\.\d{2})$").unwrap()); +static FNAME_RE: LazyLock = LazyLock::new(|| Regex::new(FNAME_REGEX).unwrap()); +static TWITTER_USERNAME_RE: LazyLock = + LazyLock::new(|| Regex::new(TWITTER_USERNAME_REGEX).unwrap()); +static GITHUB_USERNAME_RE: LazyLock = + LazyLock::new(|| Regex::new(GITHUB_USERNAME_REGEX).unwrap()); +static GEO_RE: LazyLock = + LazyLock::new(|| Regex::new(r"^geo:(-?\d{1,2}\.\d{2}),(-?\d{1,3}\.\d{2})$").unwrap()); /** Number of seconds (10 minutes) that is appropriate for clock skew */ const ALLOWED_CLOCK_SKEW_SECONDS: u64 = 10 * 60; From 5a32a9ca3ae249b50646fdbd9e7784e7042fa8c1 Mon Sep 17 00:00:00 2001 From: Manan Date: Thu, 23 Apr 2026 18:09:23 -0700 Subject: [PATCH 3/5] test: cover link_add_key / link_remove_key error branches The is_empty() simplification in link_store.rs removed redundant `.len() == 0` checks that were logically unreachable, but the error branches (empty type with target, overlong type) had no existing test coverage. These 6 regression tests lock in the current behavior so future refactors catch drift. Co-Authored-By: Claude Sonnet 4.6 --- src/storage/store/account/link_store_test.rs | 114 +++++++++++++++++++ 1 file changed, 114 insertions(+) diff --git a/src/storage/store/account/link_store_test.rs b/src/storage/store/account/link_store_test.rs index fc9e03eb9..fa564cfc5 100644 --- a/src/storage/store/account/link_store_test.rs +++ b/src/storage/store/account/link_store_test.rs @@ -2144,4 +2144,118 @@ mod tests { .unwrap(); assert_eq!(result3, add3); } + + // Regression tests for link_add_key / link_remove_key error branches. + // These cover the paths simplified when `.len() == 0 || .is_empty()` duplication was removed. + + fn make_link_add_with(fid: u64, link_type: &str, target: Option) -> message::Message { + let mut msg = messages_factory::links::create_link_add(fid, link_type, 0, None, None); + if let Some(data) = msg.data.as_mut() { + if let Some(message::message_data::Body::LinkBody(body)) = data.body.as_mut() { + body.r#type = link_type.to_string(); + body.target = target; + } + } + msg + } + + fn make_link_remove_with( + fid: u64, + link_type: &str, + target: Option, + ) -> message::Message { + let mut msg = messages_factory::links::create_link_remove(fid, link_type, 0, None, None); + if let Some(data) = msg.data.as_mut() { + if let Some(message::message_data::Body::LinkBody(body)) = data.body.as_mut() { + body.r#type = link_type.to_string(); + body.target = target; + } + } + msg + } + + #[test] + fn test_make_add_key_rejects_empty_type_with_target() { + let msg = make_link_add_with(FID_FOR_TEST, "", Some(Target::TargetFid(TARGET_FID))); + let result = LinkStore::make_add_key(&msg); + assert!(result.is_err(), "empty type with target should be rejected"); + assert!( + result + .unwrap_err() + .message + .contains("targetId provided without type"), + "expected targetId-without-type error" + ); + } + + #[test] + fn test_make_remove_key_rejects_empty_type_with_target() { + let msg = make_link_remove_with(FID_FOR_TEST, "", Some(Target::TargetFid(TARGET_FID))); + let result = LinkStore::make_remove_key(&msg); + assert!(result.is_err(), "empty type with target should be rejected"); + assert!( + result + .unwrap_err() + .message + .contains("targetID provided without type"), + "expected targetID-without-type error" + ); + } + + #[test] + fn test_make_add_key_rejects_overlong_type() { + // LINK_TYPE_BYTE_SIZE is 8; "xxxxxxxxx" is 9 bytes + let msg = make_link_add_with( + FID_FOR_TEST, + "xxxxxxxxx", + Some(Target::TargetFid(TARGET_FID)), + ); + let result = LinkStore::make_add_key(&msg); + assert!( + result.is_err(), + "type longer than 8 bytes should be rejected" + ); + assert!( + result.unwrap_err().message.contains("link type invalid"), + "expected link-type-invalid error" + ); + } + + #[test] + fn test_make_remove_key_rejects_overlong_type() { + let msg = make_link_remove_with( + FID_FOR_TEST, + "xxxxxxxxx", + Some(Target::TargetFid(TARGET_FID)), + ); + let result = LinkStore::make_remove_key(&msg); + assert!( + result.is_err(), + "type longer than 8 bytes should be rejected" + ); + assert!( + result.unwrap_err().message.contains("link type invalid"), + "expected link-type-invalid error" + ); + } + + #[test] + fn test_make_add_key_accepts_valid_type() { + let msg = make_link_add_with( + FID_FOR_TEST, + LINK_TYPE_FOLLOW, + Some(Target::TargetFid(TARGET_FID)), + ); + assert!(LinkStore::make_add_key(&msg).is_ok()); + } + + #[test] + fn test_make_remove_key_accepts_valid_type() { + let msg = make_link_remove_with( + FID_FOR_TEST, + LINK_TYPE_FOLLOW, + Some(Target::TargetFid(TARGET_FID)), + ); + assert!(LinkStore::make_remove_key(&msg).is_ok()); + } } From 4ce888f19e5b36c9e95bab5cc142e89cc344d015 Mon Sep 17 00:00:00 2001 From: Manan Date: Thu, 23 Apr 2026 18:15:18 -0700 Subject: [PATCH 4/5] refactor: drop unreachable is_empty() check on fixed-size array ts_hash is Option<&[u8; TS_HASH_LENGTH]> (24 bytes), so is_empty() on the inner slice can never return true. The branch has always been dead code; removing it per PR review. Co-Authored-By: Claude Sonnet 4.6 --- src/storage/store/account/link_store.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/storage/store/account/link_store.rs b/src/storage/store/account/link_store.rs index 6a8c05121..7212ddf7e 100644 --- a/src/storage/store/account/link_store.rs +++ b/src/storage/store/account/link_store.rs @@ -367,7 +367,7 @@ impl LinkStore { fid: u64, ts_hash: Option<&[u8; TS_HASH_LENGTH]>, ) -> Result, HubError> { - if fid != 0 && (ts_hash.is_none() || ts_hash.is_some_and(|tsh| tsh.is_empty())) { + if fid != 0 && ts_hash.is_none() { return Err(HubError::validation_failure( "fid provided without timestamp hash", )); From 09817b6756322f5599534e145ac3b3b0a961850a Mon Sep 17 00:00:00 2001 From: Manan Date: Fri, 24 Apr 2026 16:24:50 -0700 Subject: [PATCH 5/5] Revert: keep jemalloc, drop only the malloc_conf export MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per PR review (#796), the original commit removed too much. jemalloc remains the global allocator and the tikv-jemallocator / tikv-jemalloc-ctl deps + rocksdb \"jemalloc\" feature stay in place. Only the malloc_conf export — which was unused as topocount noted in the original #780 proposal — is removed. emit_jemalloc_stats and its callers are restored. Co-Authored-By: Claude Sonnet 4.6 --- Cargo.toml | 4 +++- src/jobs/snapshot_upload.rs | 6 +++++- src/lib.rs | 4 ++++ src/utils/statsd_wrapper.rs | 21 +++++++++++++++++++++ 4 files changed, 33 insertions(+), 2 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 11bd6c611..f42cb90ae 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -65,7 +65,7 @@ alloy-signer = "0.8.0" alloy-signer-local = "0.8.0" ed25519-dalek = { version = "2.1.1", features = ["batch"] } pre-commit = "0.5.2" -rocksdb = {git = "https://github.com/rust-rocksdb/rust-rocksdb.git", rev="a4f178d0ca668c4b48d993ab8aef6f4004780f23", features=["multi-threaded-cf"]} +rocksdb = {git = "https://github.com/rust-rocksdb/rust-rocksdb.git", rev="a4f178d0ca668c4b48d993ab8aef6f4004780f23", features=["multi-threaded-cf", "jemalloc"]} walkdir = "2.5.0" once_cell = "1.20.2" threadpool = "1.8.1" @@ -103,6 +103,8 @@ solar-data-structures = "=0.1.1" solar-macros = "=0.1.1" solar-config = "=0.1.1" nix = { version = "0.29", features = ["resource"] } +tikv-jemallocator = "0.6" +tikv-jemalloc-ctl = { version = "0.6", features = ["stats"] } [dev-dependencies] diff --git a/src/jobs/snapshot_upload.rs b/src/jobs/snapshot_upload.rs index c38410541..f333c7b92 100644 --- a/src/jobs/snapshot_upload.rs +++ b/src/jobs/snapshot_upload.rs @@ -22,11 +22,13 @@ async fn backup_and_upload( statsd_client: StatsdClientWrapper, ) -> Result<(), SnapshotError> { info!(shard_id, "Starting backup for shard"); + statsd_client.emit_jemalloc_stats(); let backup_dir = snapshot_config.backup_dir.clone(); let tar_gz_path = storage::db::backup::backup_db(db, &backup_dir, shard_id, now)?; info!(shard_id, "Backup complete, starting upload for shard"); + statsd_client.emit_jemalloc_stats(); storage::db::snapshot::upload_to_s3( fc_network, @@ -39,6 +41,7 @@ async fn backup_and_upload( clear_old_snapshots(fc_network, &snapshot_config, shard_id).await?; info!(shard_id, "Upload complete for shard"); + statsd_client.emit_jemalloc_stats(); Ok(()) } @@ -162,7 +165,8 @@ pub async fn upload_snapshot( } } - info!("Snapshot upload complete"); + info!("Snapshot upload complete, emitting jemalloc stats after cleanup"); + statsd_client.emit_jemalloc_stats(); Ok(()) } diff --git a/src/lib.rs b/src/lib.rs index 0dd11923d..c4a94eab4 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,3 +1,7 @@ +#[cfg(not(target_env = "msvc"))] +#[global_allocator] +static GLOBAL: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc; + pub mod bootstrap; pub mod cfg; pub mod connectors; diff --git a/src/utils/statsd_wrapper.rs b/src/utils/statsd_wrapper.rs index aa2340074..8de627134 100644 --- a/src/utils/statsd_wrapper.rs +++ b/src/utils/statsd_wrapper.rs @@ -101,4 +101,25 @@ impl StatsdClientWrapper { pub fn time(&self, key: &str, value: u64) { _ = self.client.time(key, value) } + + pub fn emit_jemalloc_stats(&self) { + #[cfg(not(target_env = "msvc"))] + { + use tikv_jemalloc_ctl::stats; + if tikv_jemalloc_ctl::epoch::advance().is_ok() { + if let Ok(allocated) = stats::allocated::read() { + self.gauge("jemalloc.allocated", allocated as u64, vec![]); + } + if let Ok(active) = stats::active::read() { + self.gauge("jemalloc.active", active as u64, vec![]); + } + if let Ok(resident) = stats::resident::read() { + self.gauge("jemalloc.resident", resident as u64, vec![]); + } + if let Ok(retained) = stats::retained::read() { + self.gauge("jemalloc.retained", retained as u64, vec![]); + } + } + } + } }