[Fix] Bound Iota storage, relay and transport resources
This commit is contained in:
Alex Emmet
parent
46078cbc4a
commit
e19c3c3d12
19 changed files with 609 additions and 48 deletions
|
|
@ -21,6 +21,7 @@ sha2 = "0.10"
|
|||
thiserror = "2"
|
||||
rand = "0.8"
|
||||
rusqlite = "0.40.0"
|
||||
fs2 = "0.4.3"
|
||||
tokio = { version = "1.50.0", features = ["full"] }
|
||||
|
||||
[dev-dependencies]
|
||||
|
|
|
|||
|
|
@ -14,6 +14,8 @@ pub enum StorageError {
|
|||
RevisionConflict,
|
||||
#[error("pending relay ownership is unknown")]
|
||||
PendingRelayOwnershipUnknown,
|
||||
#[error("asset resource limit: {0}")]
|
||||
AssetResourceLimit(&'static str),
|
||||
#[error("{0}")]
|
||||
Other(String),
|
||||
}
|
||||
|
|
|
|||
|
|
@ -40,8 +40,14 @@ pub enum ConfigError {
|
|||
InvalidRelayRouterKey(String),
|
||||
#[error("relay router certificate path must not be empty")]
|
||||
MissingRelayRouterCertificate,
|
||||
#[error("web.max_mtp_sessions must be greater than zero")]
|
||||
InvalidMaxMtpSessions,
|
||||
#[error("max_ipc_clients must be greater than zero")]
|
||||
InvalidMaxIpcClients,
|
||||
#[error("invalid storage limit: {0}")]
|
||||
InvalidStorageLimit(&'static str),
|
||||
#[error("invalid max_relay_future_skew_millis")]
|
||||
InvalidRelayFutureSkew,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
|
|
@ -71,6 +77,44 @@ pub struct IotaConfig {
|
|||
pub read_receipts_enabled: bool,
|
||||
#[serde(default = "default_max_ipc_clients")]
|
||||
pub max_ipc_clients: usize,
|
||||
#[serde(default)]
|
||||
pub storage_limits: StorageLimits,
|
||||
#[serde(default = "default_max_relay_future_skew_millis")]
|
||||
pub max_relay_future_skew_millis: u64,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
pub struct StorageLimits {
|
||||
pub max_asset_bytes: i64,
|
||||
pub max_user_asset_bytes: i64,
|
||||
#[serde(default = "default_max_user_blobs")]
|
||||
pub max_user_blobs: i64,
|
||||
pub max_active_asset_uploads_per_user: usize,
|
||||
pub min_free_asset_storage_bytes: u64,
|
||||
#[serde(default = "default_max_asset_io_workers")]
|
||||
pub max_asset_io_workers: usize,
|
||||
}
|
||||
|
||||
const fn default_max_user_blobs() -> i64 {
|
||||
4096
|
||||
}
|
||||
|
||||
const fn default_max_asset_io_workers() -> usize {
|
||||
4
|
||||
}
|
||||
|
||||
impl Default for StorageLimits {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
max_asset_bytes: 256 * 1024 * 1024,
|
||||
max_user_asset_bytes: 2 * 1024 * 1024 * 1024,
|
||||
max_user_blobs: default_max_user_blobs(),
|
||||
max_active_asset_uploads_per_user: 4,
|
||||
min_free_asset_storage_bytes: 512 * 1024 * 1024,
|
||||
max_asset_io_workers: default_max_asset_io_workers(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
|
|
@ -109,11 +153,16 @@ pub struct WebSettings {
|
|||
pub key: Option<String>,
|
||||
#[serde(default)]
|
||||
pub required: bool,
|
||||
#[serde(default = "default_max_mtp_sessions")]
|
||||
pub max_mtp_sessions: usize,
|
||||
#[serde(default)]
|
||||
pub direct_endpoints: Vec<String>,
|
||||
#[serde(default)]
|
||||
pub relay_hints: Vec<String>,
|
||||
}
|
||||
const fn default_max_mtp_sessions() -> usize {
|
||||
256
|
||||
}
|
||||
fn default_web_bind() -> String {
|
||||
"127.0.0.1".into()
|
||||
}
|
||||
|
|
@ -130,6 +179,7 @@ impl Default for WebSettings {
|
|||
certificate: None,
|
||||
key: None,
|
||||
required: false,
|
||||
max_mtp_sessions: default_max_mtp_sessions(),
|
||||
direct_endpoints: Vec::new(),
|
||||
relay_hints: Vec::new(),
|
||||
}
|
||||
|
|
@ -148,6 +198,10 @@ const fn default_max_ipc_clients() -> usize {
|
|||
64
|
||||
}
|
||||
|
||||
const fn default_max_relay_future_skew_millis() -> u64 {
|
||||
5 * 60 * 1_000
|
||||
}
|
||||
|
||||
impl Default for IotaConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
|
|
@ -163,6 +217,8 @@ impl Default for IotaConfig {
|
|||
private_key: None,
|
||||
read_receipts_enabled: default_read_receipts_enabled(),
|
||||
max_ipc_clients: default_max_ipc_clients(),
|
||||
storage_limits: StorageLimits::default(),
|
||||
max_relay_future_skew_millis: default_max_relay_future_skew_millis(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -211,9 +267,33 @@ pub fn validate_config(config: &IotaConfig) -> Result<(), ConfigError> {
|
|||
bind: config.web.bind.clone(),
|
||||
source,
|
||||
})?;
|
||||
if config.storage_limits.max_user_blobs <= 0 {
|
||||
return Err(ConfigError::InvalidStorageLimit("max_user_blobs"));
|
||||
}
|
||||
if config.web.max_mtp_sessions == 0 {
|
||||
return Err(ConfigError::InvalidMaxMtpSessions);
|
||||
}
|
||||
if config.max_ipc_clients == 0 {
|
||||
return Err(ConfigError::InvalidMaxIpcClients);
|
||||
}
|
||||
if config.max_relay_future_skew_millis > super::relay_replay::MAX_RELAY_FUTURE_SKEW_MILLIS {
|
||||
return Err(ConfigError::InvalidRelayFutureSkew);
|
||||
}
|
||||
let limits = &config.storage_limits;
|
||||
if limits.max_asset_bytes <= 0 {
|
||||
return Err(ConfigError::InvalidStorageLimit("max_asset_bytes"));
|
||||
}
|
||||
if limits.max_user_asset_bytes < limits.max_asset_bytes {
|
||||
return Err(ConfigError::InvalidStorageLimit("max_user_asset_bytes"));
|
||||
}
|
||||
if limits.max_active_asset_uploads_per_user == 0 {
|
||||
return Err(ConfigError::InvalidStorageLimit(
|
||||
"max_active_asset_uploads_per_user",
|
||||
));
|
||||
}
|
||||
if limits.max_asset_io_workers == 0 {
|
||||
return Err(ConfigError::InvalidStorageLimit("max_asset_io_workers"));
|
||||
}
|
||||
for endpoint in config
|
||||
.web
|
||||
.direct_endpoints
|
||||
|
|
@ -357,6 +437,30 @@ mod tests {
|
|||
use super::{ConfigError, IotaConfig, RelayRouterSettings, parse_config, validate_config};
|
||||
use std::path::Path;
|
||||
|
||||
#[test]
|
||||
fn asset_limits_reject_invalid_quota_and_upload_count() {
|
||||
let mut config = IotaConfig::default();
|
||||
config.storage_limits.max_asset_bytes = 0;
|
||||
assert!(matches!(
|
||||
validate_config(&config),
|
||||
Err(ConfigError::InvalidStorageLimit("max_asset_bytes"))
|
||||
));
|
||||
config.storage_limits.max_asset_bytes = 10;
|
||||
config.storage_limits.max_user_asset_bytes = 9;
|
||||
assert!(matches!(
|
||||
validate_config(&config),
|
||||
Err(ConfigError::InvalidStorageLimit("max_user_asset_bytes"))
|
||||
));
|
||||
config.storage_limits.max_user_asset_bytes = 10;
|
||||
config.storage_limits.max_active_asset_uploads_per_user = 0;
|
||||
assert!(matches!(
|
||||
validate_config(&config),
|
||||
Err(ConfigError::InvalidStorageLimit(
|
||||
"max_active_asset_uploads_per_user"
|
||||
))
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn malformed_yaml_is_rejected() {
|
||||
assert!(matches!(
|
||||
|
|
|
|||
|
|
@ -2,6 +2,9 @@ use crate::storage_error::StorageError;
|
|||
use crate::util::db;
|
||||
use rusqlite::{OptionalExtension, params};
|
||||
|
||||
pub const RELAY_RETENTION_MILLIS: i64 = 30 * 24 * 60 * 60 * 1_000;
|
||||
pub const MAX_RELAY_FUTURE_SKEW_MILLIS: u64 = 5 * 60 * 1_000;
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct DeliveredRelay {
|
||||
pub signer_id: i64,
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
use crate::storage_error::StorageError;
|
||||
use crate::util::{db, sync};
|
||||
use crate::util::{config_util::CONFIG, db, sync};
|
||||
use rand::RngCore;
|
||||
use rusqlite::{OptionalExtension, Row, params};
|
||||
use sha2::{Digest, Sha256};
|
||||
|
|
@ -118,7 +118,8 @@ fn reconcile_offset(
|
|||
if status.state != "uploading" {
|
||||
return Ok(status);
|
||||
}
|
||||
let actual = fs::metadata(asset_dir()?.join(file_name))?.len() as i64;
|
||||
let actual = i64::try_from(fs::metadata(asset_dir()?.join(file_name))?.len())
|
||||
.map_err(|_| StorageError::Other("upload file exceeds supported size".into()))?;
|
||||
if actual > status.size {
|
||||
return Err(StorageError::Other(
|
||||
"upload file exceeds declared size".into(),
|
||||
|
|
@ -151,6 +152,11 @@ pub fn start(
|
|||
if size < 0 || sha256.len() != 32 || mime_type.is_empty() || mime_type.len() > 255 {
|
||||
return Err(StorageError::Other("invalid asset metadata".into()));
|
||||
}
|
||||
let config = CONFIG.load();
|
||||
let limits = &config.storage_limits;
|
||||
if size > limits.max_asset_bytes {
|
||||
return Err(StorageError::AssetResourceLimit("max_asset_bytes"));
|
||||
}
|
||||
cleanup_if_due()?;
|
||||
if let Some((status, file_name)) = load_upload(user_id, upload_id)? {
|
||||
if status.asset_id != asset_id
|
||||
|
|
@ -166,15 +172,52 @@ pub fn start(
|
|||
}
|
||||
|
||||
let file_name = random_file_name();
|
||||
let path = asset_dir()?.join(&file_name);
|
||||
let directory = asset_dir()?;
|
||||
let required_free = limits
|
||||
.min_free_asset_storage_bytes
|
||||
.checked_add(
|
||||
u64::try_from(size).map_err(|_| StorageError::Other("invalid asset size".into()))?,
|
||||
)
|
||||
.ok_or_else(|| StorageError::Other("filesystem reservation overflow".into()))?;
|
||||
if fs2::available_space(&directory)? < required_free {
|
||||
return Err(StorageError::AssetResourceLimit(
|
||||
"min_free_asset_storage_bytes",
|
||||
));
|
||||
}
|
||||
let path = directory.join(&file_name);
|
||||
OpenOptions::new()
|
||||
.write(true)
|
||||
.create_new(true)
|
||||
.open(&path)?
|
||||
.sync_all()?;
|
||||
let now = sync::now_millis();
|
||||
let inserted = db::with_db(|conn| {
|
||||
conn.execute(
|
||||
let inserted = db::with_immediate_transaction(|tx| {
|
||||
let committed: i64 = tx.query_row(
|
||||
"SELECT COALESCE(SUM(size), 0) FROM user_assets WHERE user_id = ?1 AND deleted = 0",
|
||||
[user_id],
|
||||
|row| row.get(0),
|
||||
)?;
|
||||
let (reserved, active): (i64, i64) = tx.query_row(
|
||||
"SELECT COALESCE(SUM(size), 0), COUNT(*) FROM user_asset_uploads WHERE user_id = ?1 AND state = 'uploading'",
|
||||
[user_id],
|
||||
|row| Ok((row.get(0)?, row.get(1)?)),
|
||||
)?;
|
||||
if usize::try_from(active)
|
||||
.ok()
|
||||
.is_none_or(|count| count >= limits.max_active_asset_uploads_per_user)
|
||||
{
|
||||
return Err(StorageError::AssetResourceLimit(
|
||||
"max_active_asset_uploads_per_user",
|
||||
));
|
||||
}
|
||||
let proposed = committed
|
||||
.checked_add(reserved)
|
||||
.and_then(|value| value.checked_add(size))
|
||||
.ok_or_else(|| StorageError::Other("asset quota accounting overflow".into()))?;
|
||||
if proposed > limits.max_user_asset_bytes {
|
||||
return Err(StorageError::AssetResourceLimit("max_user_asset_bytes"));
|
||||
}
|
||||
tx.execute(
|
||||
"INSERT INTO user_asset_uploads (upload_id, user_id, asset_id, file_name, size, sha256, mime_type, received, state, created_at, updated_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, 0, 'uploading', ?8, ?8)",
|
||||
params![upload_id, user_id, asset_id, file_name, size, sha256, mime_type, now],
|
||||
)?;
|
||||
|
|
@ -219,10 +262,12 @@ pub fn append(
|
|||
return Err(StorageError::Other("upload not found".into()));
|
||||
};
|
||||
let status = reconcile_offset(user_id, status, &file_name)?;
|
||||
if status.state != "uploading"
|
||||
|| offset != status.offset
|
||||
|| offset + chunk.len() as i64 > status.size
|
||||
{
|
||||
let chunk_len = i64::try_from(chunk.len())
|
||||
.map_err(|_| StorageError::Other("asset chunk length overflow".into()))?;
|
||||
let end = offset
|
||||
.checked_add(chunk_len)
|
||||
.ok_or_else(|| StorageError::Other("asset chunk offset overflow".into()))?;
|
||||
if status.state != "uploading" || offset != status.offset || end > status.size {
|
||||
return Err(StorageError::Other(
|
||||
"asset chunk offset does not match acknowledged offset".into(),
|
||||
));
|
||||
|
|
@ -231,7 +276,7 @@ pub fn append(
|
|||
let mut file = OpenOptions::new().append(true).open(path)?;
|
||||
file.write_all(chunk)?;
|
||||
file.sync_data()?;
|
||||
let received = offset + chunk.len() as i64;
|
||||
let received = end;
|
||||
let now = sync::now_millis();
|
||||
db::with_db(|conn| {
|
||||
conn.execute(
|
||||
|
|
@ -362,6 +407,34 @@ pub fn abort(user_id: i64, upload_id: &str) -> Result<bool, StorageError> {
|
|||
}
|
||||
}
|
||||
|
||||
pub const ASSET_PAGE_SIZE: i64 = 128;
|
||||
|
||||
pub fn list_page(
|
||||
user_id: i64,
|
||||
after_id: Option<i64>,
|
||||
) -> Result<(Vec<UserAssetMetadata>, Option<i64>), StorageError> {
|
||||
if user_id <= 0 || after_id.is_some_and(|id| id <= 0) {
|
||||
return Err(StorageError::Other("invalid asset list cursor".into()));
|
||||
}
|
||||
cleanup_if_due()?;
|
||||
db::with_db(|conn| {
|
||||
let mut statement = conn.prepare("SELECT id, user_id, asset_id, file_name, size, sha256, mime_type, revision, created_at, updated_at FROM user_assets WHERE user_id = ?1 AND deleted = 0 AND id < ?2 ORDER BY id DESC LIMIT ?3")?;
|
||||
let mut items = statement
|
||||
.query_map(
|
||||
params![user_id, after_id.unwrap_or(i64::MAX), ASSET_PAGE_SIZE + 1],
|
||||
metadata_from_row,
|
||||
)?
|
||||
.collect::<Result<Vec<_>, _>>()?;
|
||||
let next = if items.len() > ASSET_PAGE_SIZE as usize {
|
||||
items.pop();
|
||||
items.last().map(|asset| asset.id)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
Ok((items, next))
|
||||
})
|
||||
}
|
||||
|
||||
pub fn list(user_id: i64) -> Result<Vec<UserAssetMetadata>, StorageError> {
|
||||
cleanup_if_due()?;
|
||||
db::with_db(|conn| {
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
/* Opaque client-owned bytes. Confidentiality is provided by the client-side blob format. */
|
||||
use crate::storage_error::StorageError;
|
||||
use crate::util::{db, sync};
|
||||
use crate::util::{config_util::CONFIG, db, sync};
|
||||
use rusqlite::{Connection, OptionalExtension, Row, Transaction, params};
|
||||
|
||||
pub const MAX_BLOB_ID_BYTES: usize = 256;
|
||||
|
|
@ -97,6 +97,16 @@ pub fn put(
|
|||
.filter(|(_, _, deleted, _)| !*deleted)
|
||||
.map(|(_, _, _, size)| size)
|
||||
.unwrap_or(0);
|
||||
if existing.is_none_or(|(_, _, deleted, _)| deleted) {
|
||||
let count: i64 = tx.query_row(
|
||||
"SELECT COUNT(*) FROM user_blobs WHERE user_id = ?1 AND deleted = 0",
|
||||
[user_id],
|
||||
|row| row.get(0),
|
||||
)?;
|
||||
if count >= CONFIG.load().storage_limits.max_user_blobs {
|
||||
return Err(StorageError::Other("user blob count quota exceeded".into()));
|
||||
}
|
||||
}
|
||||
let aggregate: i64 = tx.query_row("SELECT COALESCE(SUM(length(blob)), 0) FROM user_blobs WHERE user_id = ?1 AND deleted = 0", [user_id], |r| r.get(0))?;
|
||||
let proposed_size = aggregate
|
||||
.checked_sub(current_size)
|
||||
|
|
@ -145,6 +155,46 @@ pub fn list(user_id: i64) -> Result<Vec<UserBlob>, StorageError> {
|
|||
)
|
||||
}
|
||||
|
||||
pub const BLOB_PAGE_SIZE: i64 = 128;
|
||||
|
||||
pub fn list_metadata_page(
|
||||
user_id: i64,
|
||||
after_id: Option<i64>,
|
||||
) -> Result<(Vec<UserBlobMetadata>, Option<i64>), StorageError> {
|
||||
if user_id <= 0 || after_id.is_some_and(|id| id <= 0) {
|
||||
return Err(StorageError::Other("invalid blob list cursor".into()));
|
||||
}
|
||||
db::with_db(|conn| {
|
||||
let mut statement = conn.prepare("SELECT id, blob_id, revision, updated_at FROM user_blobs WHERE user_id = ?1 AND deleted = 0 AND id < ?2 ORDER BY id DESC LIMIT ?3")?;
|
||||
let mut rows = statement.query(params![
|
||||
user_id,
|
||||
after_id.unwrap_or(i64::MAX),
|
||||
BLOB_PAGE_SIZE + 1
|
||||
])?;
|
||||
let mut items = Vec::new();
|
||||
while let Some(row) = rows.next()? {
|
||||
items.push((
|
||||
row.get::<_, i64>(0)?,
|
||||
UserBlobMetadata {
|
||||
blob_id: row.get(1)?,
|
||||
revision: row.get(2)?,
|
||||
updated_at: row.get(3)?,
|
||||
},
|
||||
));
|
||||
}
|
||||
let next = if items.len() > BLOB_PAGE_SIZE as usize {
|
||||
items.pop();
|
||||
items.last().map(|(id, _)| *id)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
Ok((
|
||||
items.into_iter().map(|(_, metadata)| metadata).collect(),
|
||||
next,
|
||||
))
|
||||
})
|
||||
}
|
||||
|
||||
pub fn list_metadata(user_id: i64) -> Result<Vec<UserBlobMetadata>, StorageError> {
|
||||
if user_id <= 0 {
|
||||
return Err(StorageError::Other("invalid blob owner".into()));
|
||||
|
|
|
|||
50
iota-storage/tests/asset_quota.rs
Normal file
50
iota-storage/tests/asset_quota.rs
Normal file
|
|
@ -0,0 +1,50 @@
|
|||
use iota_storage::util::{config_util, db, user_assets};
|
||||
use sha2::{Digest, Sha256};
|
||||
use std::sync::Arc;
|
||||
|
||||
#[test]
|
||||
fn upload_quota_reserves_and_releases_capacity() -> Result<(), Box<dyn std::error::Error>> {
|
||||
let directory = tempfile::tempdir()?;
|
||||
iota_util::file_util::configure_storage_directory(directory.path().into());
|
||||
db::initialize_database()?;
|
||||
let mut config = config_util::IotaConfig::default();
|
||||
config.storage_limits.max_asset_bytes = 5;
|
||||
config.storage_limits.max_user_asset_bytes = 8;
|
||||
config.storage_limits.max_active_asset_uploads_per_user = 2;
|
||||
config.storage_limits.min_free_asset_storage_bytes = 0;
|
||||
config_util::CONFIG.store(Arc::new(config));
|
||||
|
||||
let hash = Sha256::digest(b"data");
|
||||
assert!(user_assets::start(42, "oversize", "a", 6, &hash, "text/plain").is_err());
|
||||
user_assets::start(42, "first", "a", 4, &hash, "text/plain")?;
|
||||
user_assets::start(42, "second", "b", 4, &hash, "text/plain")?;
|
||||
assert!(user_assets::start(42, "third", "c", 1, &hash, "text/plain").is_err());
|
||||
assert!(user_assets::abort(42, "second")?);
|
||||
user_assets::start(42, "third", "c", 1, &hash, "text/plain")?;
|
||||
user_assets::append(42, "first", 0, b"data")?;
|
||||
user_assets::commit(42, "first")?;
|
||||
assert!(user_assets::start(42, "fourth", "d", 4, &hash, "text/plain").is_err());
|
||||
assert!(user_assets::abort(42, "third")?);
|
||||
user_assets::start(42, "fourth", "d", 4, &hash, "text/plain")?;
|
||||
|
||||
let barrier = Arc::new(std::sync::Barrier::new(2));
|
||||
let attempts: Vec<_> = ["concurrent-a", "concurrent-b"]
|
||||
.into_iter()
|
||||
.map(|upload_id| {
|
||||
let barrier = barrier.clone();
|
||||
std::thread::spawn(move || {
|
||||
barrier.wait();
|
||||
user_assets::start(77, upload_id, upload_id, 5, &hash, "text/plain").is_ok()
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
assert_eq!(
|
||||
attempts
|
||||
.into_iter()
|
||||
.filter_map(|handle| handle.join().ok())
|
||||
.filter(|ok| *ok)
|
||||
.count(),
|
||||
1
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
Loading…
Reference in a new issue