use crate::storage_api::owner::NS_SCANNER_PROTOCOL_VERSION;
use crate::{
DATA_USAGE_BLOOM_NAME_PATH, Disk, EcstoreError, ScannerDiskExt as _, ScannerError, ScannerObjectIO,
is_reserved_or_invalid_bucket, read_config, resolve_scanner_object_store_handle,
};
use std::collections::HashMap;
use std::sync::{Arc, LazyLock, Mutex};
use std::time::Duration;
use tokio::sync::{Mutex as AsyncMutex, Semaphore};
use tokio::time::{Instant, MissedTickBehavior};
use uuid::Uuid;
const NS_SCANNER_MAX_FRAME_SIZE: usize = 2 * 1024 * 1024;
const NS_SCANNER_HEARTBEAT_INTERVAL: Duration = Duration::from_secs(1);
const NS_SCANNER_BUDGET_HEARTBEAT_INTERVAL: Duration = Duration::from_millis(250);
const NS_SCANNER_STALL_TIMEOUT: Duration = Duration::from_secs(15);
const NS_SCANNER_SEMANTIC_STALL_TIMEOUT: Duration = Duration::from_secs(5 * 60);
const NS_SCANNER_WRITE_TIMEOUT: Duration = Duration::from_secs(15);
const NS_SCANNER_MAX_RPC_LIFETIME: Duration = Duration::from_secs(24 * 60 * 60);
const NS_SCANNER_DISCONNECT_GRACE_MAX: Duration = Duration::from_secs(2 * 60);
const NS_SCANNER_LOCK_POLL_INTERVAL: Duration = Duration::from_millis(250);
const NS_SCANNER_FENCE_POLL_INTERVAL: Duration = Duration::from_secs(5);
const NS_SCANNER_LOCK_LOSS_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(30);
const NS_SCANNER_DISK_HEALTH_TIMEOUT: Duration = Duration::from_secs(5);
const NS_SCANNER_VALIDATED_CYCLE_TTL: Duration = Duration::from_secs(1);
const NS_SCANNER_MAX_REPLAY_SESSIONS: usize = 65_536;
const NS_SCANNER_MAX_ERROR_CHARS: usize = 4096;
const NS_SCANNER_RETRY_BUCKET_ERROR_PREFIX: &str = "retry_bucket:";
const NS_SCANNER_FRAME_AUTH_DOMAIN: &[u8] = b"rustfs-ns-scanner-frame-v3";
pub const NS_SCANNER_MAX_REQUEST_BODY_SIZE: usize = 16 * 1024;
static REMOTE_SCANNER_REPLAY_CACHE: LazyLock<Mutex<RemoteScannerReplayCache>> =
LazyLock::new(|| Mutex::new(RemoteScannerReplayCache::default()));
static REMOTE_SCANNER_ADMISSION: LazyLock<Mutex<HashMap<String, Arc<Semaphore>>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
static REMOTE_SCANNER_VALIDATED_CYCLE: LazyLock<Mutex<Option<RemoteScannerValidatedCycle>>> = LazyLock::new(|| Mutex::new(None));
static REMOTE_SCANNER_CYCLE_REFRESH: LazyLock<AsyncMutex<()>> = LazyLock::new(|| AsyncMutex::new(()));
mod stream;
#[cfg(test)]
pub(crate) use stream::checkpoint_fixture_partial_return;
pub use stream::{RemoteScannerAdmission, RemoteScannerRequest, serve_remote_scanner_request};
pub(crate) use stream::{RemoteScannerOutcome, RemoteScannerScanSpec, scan_remote_bucket};
use stream::{RemoteScannerReplayCache, RemoteScannerRequestWire, RemoteScannerValidatedCycle};
pub fn decode_remote_scanner_request(body: &[u8]) -> Result<RemoteScannerRequest, ScannerError> {
if body.is_empty() || body.len() > NS_SCANNER_MAX_REQUEST_BODY_SIZE {
return Err(ScannerError::Other("remote namespace scanner request size is invalid".to_string()));
}
let request: RemoteScannerRequestWire =
rmp_serde::from_slice(body).map_err(|_| ScannerError::Other("invalid remote namespace scanner request".to_string()))?;
if request.version != NS_SCANNER_PROTOCOL_VERSION {
return Err(ScannerError::Other(format!(
"unsupported remote namespace scanner protocol version: {}",
request.version
)));
}
if request.request_id.is_nil() {
return Err(ScannerError::Other("remote namespace scanner request ID is invalid".to_string()));
}
if request.server_epoch.is_nil() {
return Err(ScannerError::Other("remote namespace scanner server epoch is invalid".to_string()));
}
if request.session_id.is_nil() {
return Err(ScannerError::Other("remote namespace scanner session ID is invalid".to_string()));
}
if request.leader_epoch == 0 {
return Err(ScannerError::Other("remote namespace scanner leader epoch is invalid".to_string()));
}
if request.bucket.contains(['/', '\\']) || is_reserved_or_invalid_bucket(&request.bucket, false) {
return Err(ScannerError::Other("remote namespace scanner bucket is invalid".to_string()));
}
Ok(RemoteScannerRequest(request))
}
pub fn remote_scanner_request_matches_envelope(
request: &RemoteScannerRequest,
request_id: Uuid,
server_epoch: Uuid,
session_id: Uuid,
session_sequence: u64,
next_cycle: u64,
leader_epoch: u64,
) -> bool {
request.0.request_id == request_id
&& request.0.server_epoch == server_epoch
&& request.0.session_id == session_id
&& request.0.session_sequence == session_sequence
&& request.0.next_cycle == next_cycle
&& request.0.leader_epoch == leader_epoch
}
pub async fn validate_remote_scanner_request_fence(
requested_cycle: u64,
requested_leader_epoch: u64,
) -> Result<(), ScannerError> {
let store = resolve_scanner_object_store_handle()
.ok_or_else(|| ScannerError::Other("remote namespace scanner object layer is unavailable".to_string()))?;
validate_remote_scanner_request_fence_cached(
requested_cycle,
requested_leader_epoch,
store,
&REMOTE_SCANNER_VALIDATED_CYCLE,
&REMOTE_SCANNER_CYCLE_REFRESH,
NS_SCANNER_VALIDATED_CYCLE_TTL,
)
.await
}
async fn validate_remote_scanner_request_fence_cached(
requested_cycle: u64,
requested_leader_epoch: u64,
store: Arc<impl ScannerObjectIO>,
cache: &Mutex<Option<RemoteScannerValidatedCycle>>,
refresh: &AsyncMutex<()>,
cache_ttl: Duration,
) -> Result<(), ScannerError> {
if cached_remote_scanner_fence_matches(cache, requested_cycle, requested_leader_epoch, Instant::now())? {
return Ok(());
}
let _refresh = refresh.lock().await;
if cached_remote_scanner_fence_matches(cache, requested_cycle, requested_leader_epoch, Instant::now())? {
return Ok(());
}
let (persisted_cycle, persisted_leader_epoch) =
validate_remote_scanner_request_fence_with_store(requested_cycle, requested_leader_epoch, store).await?;
let valid_until = Instant::now()
.checked_add(cache_ttl)
.ok_or_else(|| ScannerError::Other("remote namespace scanner cycle cache expiry overflow".to_string()))?;
*cache
.lock()
.map_err(|_| ScannerError::Other("remote namespace scanner cycle cache is unavailable".to_string()))? =
Some(RemoteScannerValidatedCycle {
cycle: persisted_cycle,
leader_epoch: persisted_leader_epoch,
valid_until,
});
Ok(())
}
fn cached_remote_scanner_fence_matches(
cache: &Mutex<Option<RemoteScannerValidatedCycle>>,
requested_cycle: u64,
requested_leader_epoch: u64,
now: Instant,
) -> Result<bool, ScannerError> {
Ok(cache
.lock()
.map_err(|_| ScannerError::Other("remote namespace scanner cycle cache is unavailable".to_string()))?
.is_some_and(|cached| cached.matches(requested_cycle, requested_leader_epoch, now)))
}
async fn validate_remote_scanner_request_fence_with_store(
requested_cycle: u64,
requested_leader_epoch: u64,
store: Arc<impl ScannerObjectIO>,
) -> Result<(u64, u64), ScannerError> {
let (persisted_cycle, persisted_leader_epoch) = match read_config(store, &DATA_USAGE_BLOOM_NAME_PATH).await {
Ok(buf) => crate::scanner::decode_persisted_scanner_cycle_fence(&buf)?,
Err(EcstoreError::ConfigNotFound) => (0, 0),
Err(err) => {
return Err(ScannerError::Other(format!("failed to read persisted scanner cycle state: {err}")));
}
};
if requested_cycle != persisted_cycle || requested_leader_epoch != persisted_leader_epoch {
return Err(ScannerError::Other(format!(
"remote namespace scanner fence does not match persisted state: requested_cycle={requested_cycle}, \
persisted_cycle={persisted_cycle}, requested_epoch={requested_leader_epoch}, persisted_epoch={persisted_leader_epoch}"
)));
}
Ok((persisted_cycle, persisted_leader_epoch))
}
async fn watch_remote_scanner_request_fence(
requested_cycle: u64,
requested_leader_epoch: u64,
store: Arc<impl ScannerObjectIO>,
poll_interval: Duration,
) -> Result<(), ScannerError> {
watch_remote_scanner_request_fence_with_cache(
requested_cycle,
requested_leader_epoch,
store,
poll_interval,
&REMOTE_SCANNER_VALIDATED_CYCLE,
&REMOTE_SCANNER_CYCLE_REFRESH,
NS_SCANNER_VALIDATED_CYCLE_TTL,
)
.await
}
async fn watch_remote_scanner_request_fence_with_cache(
requested_cycle: u64,
requested_leader_epoch: u64,
store: Arc<impl ScannerObjectIO>,
poll_interval: Duration,
cache: &Mutex<Option<RemoteScannerValidatedCycle>>,
refresh: &AsyncMutex<()>,
cache_ttl: Duration,
) -> Result<(), ScannerError> {
if poll_interval.is_zero() {
return Err(ScannerError::Other(
"remote namespace scanner fence poll interval must be positive".to_string(),
));
}
let first_poll = Instant::now()
.checked_add(poll_interval)
.ok_or_else(|| ScannerError::Other("remote namespace scanner fence poll deadline overflow".to_string()))?;
let mut poll = tokio::time::interval_at(first_poll, poll_interval);
poll.set_missed_tick_behavior(MissedTickBehavior::Delay);
loop {
poll.tick().await;
validate_remote_scanner_request_fence_cached(
requested_cycle,
requested_leader_epoch,
store.clone(),
cache,
refresh,
cache_ttl,
)
.await?;
}
}
pub fn admit_remote_scanner_request(disk: &Disk) -> Result<RemoteScannerAdmission, ScannerError> {
if !disk.is_local() {
return Err(ScannerError::Other(
"remote namespace scanner admission requires a local disk".to_string(),
));
}
try_admit_remote_scanner_key(disk.path().to_string_lossy().into_owned())
}
pub fn preflight_remote_scanner_request(
disk: &Disk,
session_id: Uuid,
next_cycle: u64,
leader_epoch: u64,
session_sequence: u64,
) -> Result<(), ScannerError> {
if !disk.is_local() {
return Err(ScannerError::Other(
"remote namespace scanner replay preflight requires a local disk".to_string(),
));
}
REMOTE_SCANNER_REPLAY_CACHE
.lock()
.map_err(|_| ScannerError::Other("remote namespace scanner replay cache is unavailable".to_string()))?
.preflight(
disk.path().to_string_lossy().as_ref(),
session_id,
next_cycle,
leader_epoch,
session_sequence,
)
}
pub fn claim_remote_scanner_request(
disk: &Disk,
session_id: Uuid,
next_cycle: u64,
leader_epoch: u64,
session_sequence: u64,
) -> Result<(), ScannerError> {
if !disk.is_local() {
return Err(ScannerError::Other(
"remote namespace scanner replay claim requires a local disk".to_string(),
));
}
REMOTE_SCANNER_REPLAY_CACHE
.lock()
.map_err(|_| ScannerError::Other("remote namespace scanner replay cache is unavailable".to_string()))?
.claim(
disk.path().to_string_lossy().into_owned(),
session_id,
next_cycle,
leader_epoch,
session_sequence,
)
}
pub(crate) fn try_admit_remote_scanner(disk: &Disk) -> Result<RemoteScannerAdmission, ScannerError> {
if !disk.is_local() {
return Err(ScannerError::Other(
"remote namespace scanner admission requires a local disk".to_string(),
));
}
try_admit_remote_scanner_key(disk.path().to_string_lossy().into_owned())
}
fn try_admit_remote_scanner_key(disk_key: String) -> Result<RemoteScannerAdmission, ScannerError> {
let semaphore = REMOTE_SCANNER_ADMISSION
.lock()
.map_err(|_| ScannerError::Other("remote namespace scanner admission is unavailable".to_string()))?
.entry(disk_key)
.or_insert_with(|| Arc::new(Semaphore::new(1)))
.clone();
let permit = semaphore.try_acquire_owned().map_err(|_| ScannerError::RemoteDiskBusy)?;
Ok(RemoteScannerAdmission { _permit: permit })
}