mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-31 16:22:06 +00:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 overtrue
|
5983594da9 |
@@ -2633,403 +2633,6 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
fn u27b_rows(bytes: Vec<u8>, expected: &[Vec<(Vec<u8>, Vec<u8>)>], cut: LogId, member: LogId) {
|
||||
let mut sink = U27aSink::default();
|
||||
let header = snapshot::decode(io::Cursor::new(bytes), &mut sink, u24_limits()).expect("decode captured bytes");
|
||||
assert_eq!(header.applied, Some(cut), "U27B_SAME_TX");
|
||||
assert_eq!(header.membership, u17a_membership(member.index).1, "U27B_SAME_TX");
|
||||
assert_eq!(header.cluster_id, identity().cluster_id, "U27B_SAME_TX");
|
||||
assert_eq!(header.binding, u17a_binding(), "U27B_SAME_TX");
|
||||
for (section, rows) in [
|
||||
snapshot::Section::Heads,
|
||||
snapshot::Section::Versions,
|
||||
snapshot::Section::Events,
|
||||
snapshot::Section::Operations,
|
||||
]
|
||||
.into_iter()
|
||||
.zip(&expected[3..7])
|
||||
{
|
||||
let actual: Vec<_> = sink
|
||||
.0
|
||||
.iter()
|
||||
.filter(|row| row.0 == section)
|
||||
.map(|row| (row.1.clone(), row.2.clone()))
|
||||
.collect();
|
||||
assert_eq!(&actual, rows, "U27B_SAME_TX");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn u27b_checkpoint_same_tx_cut() {
|
||||
let (_dir, _backend, store, _) = u27a_fixture();
|
||||
let before = u17b_snapshot(&store);
|
||||
let writer = NativeLogWriter::new(store);
|
||||
let owner = writer.store.clone();
|
||||
let _hook = native_snapshot::after_capture(move || {
|
||||
let mut store = owner.lock().expect("live corruption owner");
|
||||
let tx = store.write_transaction().expect("live corruption transaction");
|
||||
tx.open_table(META)
|
||||
.expect("live metadata")
|
||||
.remove("committed")
|
||||
.expect("remove live marker");
|
||||
store.commit(tx).expect("durable live corruption");
|
||||
assert!(
|
||||
matches!(store.read_committed(), Err(StoreError::CorruptRecord)),
|
||||
"live transaction sees missing committed"
|
||||
);
|
||||
});
|
||||
let old = NativeSnapshotView::acquire(writer.store.clone());
|
||||
assert_eq!(old.is_ok(), true, "U27B_SAME_TX");
|
||||
u27b_rows(
|
||||
old.expect("asserted old view")
|
||||
.stream(Vec::new(), u24_limits())
|
||||
.expect("old stream"),
|
||||
&before,
|
||||
u17a_id(10),
|
||||
u17a_id(1),
|
||||
);
|
||||
assert!(
|
||||
matches!(
|
||||
NativeSnapshotView::acquire(writer.store.clone()),
|
||||
Err(NativeSnapshotError::Store(StoreError::CorruptRecord))
|
||||
),
|
||||
"U27B_SAME_TX"
|
||||
);
|
||||
drop(_hook);
|
||||
|
||||
let (_dir2, _backend2, store, _) = u27a_fixture();
|
||||
let before = u17b_snapshot(&store);
|
||||
let writer = NativeLogWriter::new(store);
|
||||
let old = NativeSnapshotView::acquire(writer.store.clone()).expect("normal old view");
|
||||
let after = {
|
||||
let mut store = writer.store.lock().expect("normal writer");
|
||||
let mut command = u17a_command();
|
||||
command.operation_id = [81; 16];
|
||||
command.object.key = b"u27b-next-object".to_vec();
|
||||
command.prepared.version_id = [82; 16];
|
||||
u24_apply(&mut store, &[u17a_membership(11).0, u17a_created(12, command)]);
|
||||
u17b_snapshot(&store)
|
||||
};
|
||||
for i in 3..7 {
|
||||
assert_ne!(before[i], after[i], "U27B_SAME_TX new table changes");
|
||||
}
|
||||
u27b_rows(
|
||||
old.stream(Vec::new(), u24_limits()).expect("normal old stream"),
|
||||
&before,
|
||||
u17a_id(10),
|
||||
u17a_id(1),
|
||||
);
|
||||
u27b_rows(
|
||||
NativeSnapshotView::acquire(writer.store.clone())
|
||||
.expect("normal fresh view")
|
||||
.stream(Vec::new(), u24_limits())
|
||||
.expect("normal fresh stream"),
|
||||
&after,
|
||||
u17a_id(12),
|
||||
u17a_id(11),
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn u27b_checkpoint_bounds() {
|
||||
for case in 0..4 {
|
||||
let (_dir, _backend, mut store, _) = u27a_fixture();
|
||||
let tx = store.write_transaction().expect("bounds fixture transaction");
|
||||
{
|
||||
let mut meta = tx.open_table(META).expect("bounds metadata");
|
||||
match case {
|
||||
0 => {
|
||||
meta.insert("committed", encode_id(u17a_id(9)).as_slice())
|
||||
.expect("committed below applied");
|
||||
}
|
||||
1 => {
|
||||
let mut applied = vec![1];
|
||||
applied.extend(encode_id(LogId { term: 12, ..u17a_id(10) }));
|
||||
meta.insert("last_applied", applied.as_slice())
|
||||
.expect("applied term mismatch");
|
||||
}
|
||||
2 => {
|
||||
let mut applied = vec![1];
|
||||
applied.extend(encode_id(LogId {
|
||||
leader_node: 8,
|
||||
..u17a_id(10)
|
||||
}));
|
||||
meta.insert("last_applied", applied.as_slice())
|
||||
.expect("applied leader mismatch");
|
||||
}
|
||||
_ => {
|
||||
meta.remove("committed").expect("missing committed fixture");
|
||||
}
|
||||
}
|
||||
}
|
||||
store.commit(tx).expect("persist bounds fixture");
|
||||
let expected_applied = match case {
|
||||
1 => LogId { term: 12, ..u17a_id(10) },
|
||||
2 => LogId {
|
||||
leader_node: 8,
|
||||
..u17a_id(10)
|
||||
},
|
||||
_ => u17a_id(10),
|
||||
};
|
||||
assert_eq!(
|
||||
store.native_application_state().expect("valid applied encoding").0,
|
||||
Some(expected_applied),
|
||||
"U27B_FULL_CUT decoded fixture"
|
||||
);
|
||||
let before = u17b_snapshot(&store);
|
||||
let writer = NativeLogWriter::new(store);
|
||||
let result = NativeSnapshotView::acquire(writer.store.clone());
|
||||
assert_eq!(
|
||||
matches!(result, Err(NativeSnapshotError::Store(StoreError::CorruptRecord))),
|
||||
true,
|
||||
"U27B_FULL_CUT case {case}"
|
||||
);
|
||||
assert_eq!(
|
||||
u17b_snapshot(&writer.store.lock().expect("unchanged bounds owner")),
|
||||
before,
|
||||
"U27B_FULL_CUT zero writes"
|
||||
);
|
||||
}
|
||||
let (_dir, _backend, store, _) = u27a_fixture();
|
||||
let writer = NativeLogWriter::new(store);
|
||||
assert!(NativeSnapshotView::acquire(writer.store.clone()).is_ok(), "U27B_FULL_CUT equal full ID");
|
||||
u19_runtime()
|
||||
.block_on(writer.write(|store| {
|
||||
store.append_native(&[u17a_blank(11)])?;
|
||||
store.save_committed(u17a_id(11))
|
||||
}))
|
||||
.expect("advance committed only");
|
||||
assert!(
|
||||
NativeSnapshotView::acquire(writer.store.clone()).is_ok(),
|
||||
"U27B_FULL_CUT applied below committed"
|
||||
);
|
||||
|
||||
for schema6 in [false, true] {
|
||||
let (_dir, _backend, mut store) = u17a_store();
|
||||
if schema6 {
|
||||
provenance_promote(&mut store);
|
||||
}
|
||||
let schema = if schema6 { 6 } else { 4 };
|
||||
let before = u17b_snapshot(&store);
|
||||
let writer = NativeLogWriter::new(store);
|
||||
let bytes = NativeSnapshotView::acquire(writer.store.clone())
|
||||
.expect("legal empty native view")
|
||||
.stream(Vec::new(), u24_limits())
|
||||
.expect("empty bare stream");
|
||||
let mut sink = U27aSink::default();
|
||||
let header = snapshot::decode(io::Cursor::new(bytes), &mut sink, u24_limits()).expect("empty decode");
|
||||
assert_eq!(header.applied, None, "U27B_FULL_CUT empty applied");
|
||||
assert!(
|
||||
header.membership.log_id().is_none()
|
||||
&& header.membership.configs().is_empty()
|
||||
&& header.membership.nodes().is_empty(),
|
||||
"U27B_FULL_CUT empty membership"
|
||||
);
|
||||
assert!(sink.0.is_empty(), "U27B_FULL_CUT empty rows");
|
||||
let capability = NativeSnapshotView::acquire(writer.store.clone())
|
||||
.expect("empty capability view")
|
||||
.stream_capability_artifact(Vec::new(), u24_limits());
|
||||
if schema6 {
|
||||
assert!(capability.is_ok(), "U27B_FULL_CUT explicit schema6");
|
||||
} else {
|
||||
assert!(
|
||||
matches!(capability, Err(NativeSnapshotError::Store(StoreError::UnsupportedSchema))),
|
||||
"U27B_FULL_CUT schema4 gate"
|
||||
);
|
||||
}
|
||||
assert_eq!(
|
||||
committed_schema(&writer.store.lock().expect("schema owner")),
|
||||
schema,
|
||||
"U27B_FULL_CUT no promotion"
|
||||
);
|
||||
assert_eq!(
|
||||
u17b_snapshot(&writer.store.lock().expect("empty unchanged owner")),
|
||||
before,
|
||||
"U27B_FULL_CUT zero writes"
|
||||
);
|
||||
u19_runtime()
|
||||
.block_on(writer.write(|store| {
|
||||
store.append_native(&[u17a_blank(0)])?;
|
||||
store.save_committed(u17a_id(0))
|
||||
}))
|
||||
.expect("empty applied with committed log");
|
||||
assert!(
|
||||
NativeSnapshotView::acquire(writer.store.clone()).is_ok(),
|
||||
"U27B_FULL_CUT committed unapplied accepted"
|
||||
);
|
||||
}
|
||||
let (_dir, _backend, mut store) = u17a_store();
|
||||
let tx = store.write_transaction().expect("missing applied fixture");
|
||||
tx.open_table(META)
|
||||
.expect("empty metadata")
|
||||
.remove("last_applied")
|
||||
.expect("remove empty applied encoding");
|
||||
store.commit(tx).expect("commit missing applied");
|
||||
let writer = NativeLogWriter::new(store);
|
||||
assert!(
|
||||
matches!(
|
||||
NativeSnapshotView::acquire(writer.store.clone()),
|
||||
Err(NativeSnapshotError::Store(StoreError::CorruptRecord))
|
||||
),
|
||||
"U27B_FULL_CUT missing empty encoding"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn u27b_checkpoint_retained_log_identity() {
|
||||
for case in 0..5 {
|
||||
let (_dir, _backend, mut store, _) = u27a_fixture();
|
||||
let tx = store.write_transaction().expect("retained log fixture");
|
||||
{
|
||||
let mut logs = tx.open_table(LOGS).expect("retained log table");
|
||||
match case {
|
||||
0 => {
|
||||
logs.remove(10).expect("missing committed log");
|
||||
}
|
||||
1 => {
|
||||
logs.insert(10, [0xff].as_slice()).expect("malformed committed log");
|
||||
}
|
||||
2 | 3 => {
|
||||
let raw = logs
|
||||
.get(10)
|
||||
.expect("original committed log")
|
||||
.expect("retained log")
|
||||
.value()
|
||||
.to_vec();
|
||||
let mut entry = decode_log(&raw).expect("valid original log");
|
||||
if case == 2 {
|
||||
entry.id.term += 1;
|
||||
} else {
|
||||
entry.id.leader_node += 1;
|
||||
}
|
||||
logs.insert(10, encode_log(&entry).expect("encoded mismatching log").as_slice())
|
||||
.expect("mismatching committed log");
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
store.commit(tx).expect("persist retained log fixture");
|
||||
let before = u17b_snapshot(&store);
|
||||
let writer = NativeLogWriter::new(store);
|
||||
let result = NativeSnapshotView::acquire(writer.store.clone());
|
||||
if case == 4 {
|
||||
assert!(result.is_ok(), "U27B_COMMITTED_RECORD complete identity");
|
||||
} else {
|
||||
assert_eq!(
|
||||
matches!(result, Err(NativeSnapshotError::Store(StoreError::CorruptRecord))),
|
||||
true,
|
||||
"U27B_COMMITTED_RECORD case {case}"
|
||||
);
|
||||
}
|
||||
assert_eq!(
|
||||
u17b_snapshot(&writer.store.lock().expect("retained log owner")),
|
||||
before,
|
||||
"U27B_COMMITTED_RECORD zero writes"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn u27b_checkpoint_owner_and_read_lifetime() {
|
||||
let (_dir, _backend, store, _) = u27a_fixture();
|
||||
let before = u17b_snapshot(&store);
|
||||
let writer = NativeLogWriter::new(store);
|
||||
let view = NativeSnapshotView::acquire(writer.store.clone()).expect("owned captured view");
|
||||
drop(writer);
|
||||
u27b_rows(
|
||||
view.stream(Vec::new(), u24_limits()).expect("view outlives external writer"),
|
||||
&before,
|
||||
u17a_id(10),
|
||||
u17a_id(1),
|
||||
);
|
||||
|
||||
struct PausedOutput {
|
||||
entered: Option<mpsc::Sender<()>>,
|
||||
release: mpsc::Receiver<()>,
|
||||
bytes: Vec<u8>,
|
||||
}
|
||||
impl Write for PausedOutput {
|
||||
fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
|
||||
if let Some(entered) = self.entered.take() {
|
||||
entered.send(()).map_err(io::Error::other)?;
|
||||
self.release.recv_timeout(Duration::from_secs(10)).map_err(io::Error::other)?;
|
||||
}
|
||||
self.bytes.extend_from_slice(bytes);
|
||||
Ok(bytes.len())
|
||||
}
|
||||
fn flush(&mut self) -> io::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
let (_dir2, _backend2, store, _) = u27a_fixture();
|
||||
let before = u17b_snapshot(&store);
|
||||
let writer = NativeLogWriter::new(store);
|
||||
let view = NativeSnapshotView::acquire(writer.store.clone()).expect("paused view");
|
||||
let (entered_tx, entered_rx) = mpsc::channel();
|
||||
let (release_tx, release_rx) = mpsc::channel();
|
||||
let stream = std::thread::spawn(move || {
|
||||
view.stream(
|
||||
PausedOutput {
|
||||
entered: Some(entered_tx),
|
||||
release: release_rx,
|
||||
bytes: Vec::new(),
|
||||
},
|
||||
u24_limits(),
|
||||
)
|
||||
});
|
||||
let entered = entered_rx.recv_timeout(Duration::from_secs(5));
|
||||
let owner_available = writer.store.try_lock().is_ok();
|
||||
let (done_tx, done_rx) = mpsc::channel();
|
||||
let writer_for_thread = writer.clone();
|
||||
let write = std::thread::spawn(move || {
|
||||
let result = u19_runtime().block_on(writer_for_thread.write(|store| {
|
||||
u24_apply(store, &[u17a_blank(11)]);
|
||||
Ok(())
|
||||
}));
|
||||
let _ = done_tx.send(result.is_ok());
|
||||
});
|
||||
let done = done_rx.recv_timeout(Duration::from_secs(5));
|
||||
let _ = release_tx.send(());
|
||||
let streamed = stream.join().expect("paused stream joined");
|
||||
write.join().expect("writer joined after release");
|
||||
assert!(entered.is_ok(), "U27B_OWNER output rendezvous");
|
||||
assert!(owner_available, "U27B_OWNER owner try_lock while output paused: {owner_available}");
|
||||
assert!(
|
||||
matches!(done, Ok(true)),
|
||||
"U27B_OWNER writer ACK before release: done={done:?}, owner_available={owner_available}"
|
||||
);
|
||||
u27b_rows(streamed.expect("paused stream completed").bytes, &before, u17a_id(10), u17a_id(1));
|
||||
struct Broken;
|
||||
impl Write for Broken {
|
||||
fn write(&mut self, _: &[u8]) -> io::Result<usize> {
|
||||
Err(io::Error::new(io::ErrorKind::BrokenPipe, "u27b output failure"))
|
||||
}
|
||||
fn flush(&mut self) -> io::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
let result = NativeSnapshotView::acquire(writer.store.clone())
|
||||
.expect("failed output view")
|
||||
.stream(Broken, u24_limits());
|
||||
assert!(
|
||||
matches!(result, Err(NativeSnapshotError::Snapshot(snapshot::SnapshotError::Io(ref error))) if error.kind() == io::ErrorKind::BrokenPipe),
|
||||
"U27B_OWNER typed output failure"
|
||||
);
|
||||
u19_runtime()
|
||||
.block_on(writer.write(|store| {
|
||||
u24_apply(store, &[u17a_blank(12)]);
|
||||
Ok(())
|
||||
}))
|
||||
.expect("write after failed output");
|
||||
assert!(
|
||||
NativeSnapshotView::acquire(writer.store.clone())
|
||||
.expect("capture after failed output")
|
||||
.stream(Vec::new(), u24_limits())
|
||||
.is_ok(),
|
||||
"U27B_OWNER capture after failed output"
|
||||
);
|
||||
}
|
||||
|
||||
fn u24_limits() -> snapshot::Limits {
|
||||
snapshot::Limits {
|
||||
total_bytes: 1_000_000,
|
||||
@@ -3163,7 +2766,7 @@ mod tests {
|
||||
let streamed = stream.join().expect("stream joined");
|
||||
write.join().expect("writer joined");
|
||||
assert!(entered.is_ok(), "output entered rendezvous");
|
||||
assert!(matches!(done, Ok(true)), "U24_STREAM_RELEASES_WRITER done={done:?}");
|
||||
assert!(matches!(done, Ok(true)), "U24_STREAM_RELEASES_WRITER");
|
||||
assert!(streamed.is_ok(), "stream completed");
|
||||
}
|
||||
#[test]
|
||||
|
||||
@@ -6,30 +6,6 @@ use std::{
|
||||
sync::{Arc, Mutex},
|
||||
};
|
||||
|
||||
#[cfg(test)]
|
||||
std::thread_local! {
|
||||
static AFTER_CAPTURE: std::cell::RefCell<Option<Box<dyn FnOnce()>>> = const { std::cell::RefCell::new(None) };
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) struct CaptureHook;
|
||||
#[cfg(test)]
|
||||
impl Drop for CaptureHook {
|
||||
fn drop(&mut self) {
|
||||
AFTER_CAPTURE.with(|hook| {
|
||||
hook.borrow_mut().take();
|
||||
});
|
||||
}
|
||||
}
|
||||
#[cfg(test)]
|
||||
pub(crate) fn after_capture(hook: impl FnOnce() + 'static) -> CaptureHook {
|
||||
AFTER_CAPTURE.with(|slot| {
|
||||
assert!(slot.borrow().is_none(), "one capture hook per test thread");
|
||||
*slot.borrow_mut() = Some(Box::new(hook));
|
||||
});
|
||||
CaptureHook
|
||||
}
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum NativeSnapshotError {
|
||||
#[error(transparent)]
|
||||
@@ -59,21 +35,9 @@ impl NativeSnapshotView {
|
||||
}
|
||||
store.database.begin_read().map_err(StoreError::from)?
|
||||
};
|
||||
#[cfg(test)]
|
||||
if let Some(hook) = AFTER_CAPTURE.with(|slot| slot.borrow_mut().take()) {
|
||||
hook();
|
||||
}
|
||||
let (applied, membership) = application::native_application_state(&transaction)?;
|
||||
let (header, schema) = {
|
||||
let meta = transaction.open_table(crate::META).map_err(StoreError::from)?;
|
||||
let logs = transaction.open_table(crate::LOGS).map_err(StoreError::from)?;
|
||||
let committed = crate::committed::validate_committed(&meta, &logs)?;
|
||||
if let Some(applied) = applied {
|
||||
let committed = committed.ok_or(StoreError::CorruptRecord)?;
|
||||
if applied.index > committed.index || (applied.index == committed.index && applied != committed) {
|
||||
return Err(StoreError::CorruptRecord.into());
|
||||
}
|
||||
}
|
||||
let identity = meta
|
||||
.get("identity")
|
||||
.map_err(StoreError::from)?
|
||||
|
||||
@@ -372,15 +372,6 @@ pub mod config {
|
||||
with_server_config_write_lock,
|
||||
};
|
||||
|
||||
/// Observe persisted config bytes and their reader-associated metadata within a byte limit.
|
||||
pub async fn read_config_observation(
|
||||
api: std::sync::Arc<crate::store::ECStore>,
|
||||
file: &str,
|
||||
max_bytes: usize,
|
||||
) -> crate::error::Result<(Vec<u8>, crate::object_api::ObjectInfo)> {
|
||||
crate::config::com::read_config_observation(api, file, max_bytes).await
|
||||
}
|
||||
|
||||
/// Read config bytes and metadata without treating an empty payload as absent.
|
||||
pub async fn read_config_preserve_empty_with_metadata_opts(
|
||||
api: std::sync::Arc<crate::store::ECStore>,
|
||||
|
||||
@@ -449,22 +449,6 @@ where
|
||||
Ok(data)
|
||||
}
|
||||
|
||||
pub(crate) async fn read_config_observation<S>(api: Arc<S>, file: &str, max_bytes: usize) -> Result<(Vec<u8>, ObjectInfo)>
|
||||
where
|
||||
S: EcstoreObjectIO,
|
||||
{
|
||||
let opts = ObjectOptions {
|
||||
suppress_read_repair: true,
|
||||
no_lock: false,
|
||||
skip_verify_bitrot: false,
|
||||
..Default::default()
|
||||
};
|
||||
crate::object_api::without_get_object_body_cache_hook(read_config_limited_preserve_empty_with_metadata_opts(
|
||||
api, file, &opts, max_bytes,
|
||||
))
|
||||
.await
|
||||
}
|
||||
|
||||
pub(crate) async fn read_config_limited_preserve_empty<S>(api: Arc<S>, file: &str, max_bytes: usize) -> Result<Vec<u8>>
|
||||
where
|
||||
S: EcstoreObjectIO,
|
||||
@@ -5441,166 +5425,6 @@ mod tests {
|
||||
assert_eq!(store.write_calls.load(Ordering::SeqCst), 0);
|
||||
}
|
||||
|
||||
struct ObservationStream {
|
||||
inner: Cursor<Vec<u8>>,
|
||||
consumed: Arc<AtomicUsize>,
|
||||
fail: bool,
|
||||
}
|
||||
|
||||
impl AsyncRead for ObservationStream {
|
||||
fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
|
||||
if self.fail {
|
||||
return Poll::Ready(Err(std::io::Error::other("observation stream failed")));
|
||||
}
|
||||
let before = buf.filled().len();
|
||||
let result = Pin::new(&mut self.inner).poll_read(cx, buf);
|
||||
self.consumed.fetch_add(buf.filled().len() - before, Ordering::SeqCst);
|
||||
result
|
||||
}
|
||||
}
|
||||
|
||||
struct ObservationStore {
|
||||
body: Vec<u8>,
|
||||
declared_size: i64,
|
||||
error: Option<Error>,
|
||||
stream_error: bool,
|
||||
consumed: Arc<AtomicUsize>,
|
||||
options: std::sync::Mutex<Option<(bool, bool, bool)>>,
|
||||
writes: AtomicUsize,
|
||||
}
|
||||
|
||||
impl Debug for ObservationStore {
|
||||
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
|
||||
f.debug_struct("ObservationStore").finish()
|
||||
}
|
||||
}
|
||||
|
||||
impl ObservationStore {
|
||||
fn new(body: Vec<u8>, declared_size: i64) -> Self {
|
||||
Self {
|
||||
body,
|
||||
declared_size,
|
||||
error: None,
|
||||
stream_error: false,
|
||||
consumed: Arc::new(AtomicUsize::new(0)),
|
||||
options: std::sync::Mutex::new(None),
|
||||
writes: AtomicUsize::new(0),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl crate::storage_api_contracts::object::ObjectIO for ObservationStore {
|
||||
type Error = Error;
|
||||
type RangeSpec = HTTPRangeSpec;
|
||||
type HeaderMap = HeaderMap;
|
||||
type ObjectOptions = ObjectOptions;
|
||||
type ObjectInfo = ObjectInfo;
|
||||
type GetObjectReader = GetObjectReader;
|
||||
type PutObjectReader = PutObjReader;
|
||||
|
||||
async fn get_object_reader(
|
||||
&self,
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
range: Option<HTTPRangeSpec>,
|
||||
_headers: HeaderMap,
|
||||
opts: &ObjectOptions,
|
||||
) -> Result<GetObjectReader> {
|
||||
assert_eq!(bucket, RUSTFS_META_BUCKET);
|
||||
assert_eq!(object, "config/observation.bin");
|
||||
assert!(range.is_none());
|
||||
*self.options.lock().expect("observation options") =
|
||||
Some((opts.suppress_read_repair, opts.no_lock, opts.skip_verify_bitrot));
|
||||
if let Some(error) = &self.error {
|
||||
return Err(error.clone());
|
||||
}
|
||||
Ok(GetObjectReader {
|
||||
stream: Box::new(ObservationStream {
|
||||
inner: Cursor::new(self.body.clone()),
|
||||
consumed: self.consumed.clone(),
|
||||
fail: self.stream_error,
|
||||
}),
|
||||
object_info: ObjectInfo {
|
||||
size: self.declared_size,
|
||||
etag: Some("observation-etag".to_string()),
|
||||
..Default::default()
|
||||
},
|
||||
buffered_body: None,
|
||||
body_source: Default::default(),
|
||||
})
|
||||
}
|
||||
|
||||
async fn put_object(
|
||||
&self,
|
||||
_bucket: &str,
|
||||
_object: &str,
|
||||
_data: &mut PutObjReader,
|
||||
_opts: &ObjectOptions,
|
||||
) -> Result<ObjectInfo> {
|
||||
self.writes.fetch_add(1, Ordering::SeqCst);
|
||||
Err(Error::other("observation must not write"))
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn observation_preserves_options_and_errors() {
|
||||
for body in [Vec::new(), b"raw config".to_vec()] {
|
||||
let store = Arc::new(ObservationStore::new(body.clone(), i64::try_from(body.len()).expect("fixture size")));
|
||||
let result = super::read_config_observation(store.clone(), "config/observation.bin", 32).await;
|
||||
assert!(result.is_ok(), "observation must preserve empty and nonempty objects: {result:?}");
|
||||
let (observed, info) = result.expect("successful observation");
|
||||
assert_eq!(observed, body);
|
||||
assert_eq!(info.etag.as_deref(), Some("observation-etag"));
|
||||
assert_eq!(
|
||||
*store.options.lock().expect("options"),
|
||||
Some((true, false, false)),
|
||||
"I05_REPAIR_SUPPRESSED"
|
||||
);
|
||||
assert_eq!(store.writes.load(Ordering::SeqCst), 0);
|
||||
}
|
||||
for (source, expected) in [
|
||||
(Error::FileNotFound, Error::ConfigNotFound),
|
||||
(
|
||||
Error::ObjectNotFound(RUSTFS_META_BUCKET.to_string(), "config/observation.bin".to_string()),
|
||||
Error::ConfigNotFound,
|
||||
),
|
||||
(Error::ErasureReadQuorum, Error::ErasureReadQuorum),
|
||||
] {
|
||||
let mut fixture = ObservationStore::new(Vec::new(), 0);
|
||||
fixture.error = Some(source);
|
||||
let store = Arc::new(fixture);
|
||||
let result = super::read_config_observation(store.clone(), "config/observation.bin", 32).await;
|
||||
assert_eq!(result.expect_err("read error must propagate"), expected);
|
||||
assert_eq!(store.writes.load(Ordering::SeqCst), 0);
|
||||
}
|
||||
let mut fixture = ObservationStore::new(Vec::new(), 0);
|
||||
fixture.stream_error = true;
|
||||
let store = Arc::new(fixture);
|
||||
let result = super::read_config_observation(store.clone(), "config/observation.bin", 32).await;
|
||||
let error = result.expect_err("stream error must propagate");
|
||||
assert!(error.to_string().contains("observation stream failed"));
|
||||
assert_eq!(store.writes.load(Ordering::SeqCst), 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn observation_enforces_byte_limit() {
|
||||
let store = Arc::new(ObservationStore::new(vec![7; 64], 0));
|
||||
let result = super::read_config_observation(store.clone(), "config/observation.bin", 8).await;
|
||||
assert_eq!(result.is_err(), true, "I05_BOUND_ENFORCED");
|
||||
assert_eq!(
|
||||
result.expect_err("bounded observation error"),
|
||||
Error::CorruptedFormat,
|
||||
"I05_BOUND_ENFORCED"
|
||||
);
|
||||
assert_eq!(store.consumed.load(Ordering::SeqCst), 9, "I05_BOUND_ENFORCED: drain exactly limit + 1");
|
||||
let store = Arc::new(ObservationStore::new(vec![7; 8], 8));
|
||||
let result = super::read_config_observation(store.clone(), "config/observation.bin", 8).await;
|
||||
assert!(result.is_ok(), "exact byte limit must succeed: {result:?}");
|
||||
assert_eq!(result.expect("exact limit").0, vec![7; 8]);
|
||||
assert_eq!(store.consumed.load(Ordering::SeqCst), 8);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn read_config_preserve_empty_distinguishes_empty_object() {
|
||||
let store = Arc::new(RecoveryMockStore::new(RecoveryReadState::Blob(Vec::new()), None));
|
||||
|
||||
@@ -1095,217 +1095,6 @@ mod prepared_get_object_metadata_tests {
|
||||
use http::HeaderMap;
|
||||
use tokio::io::AsyncReadExt;
|
||||
|
||||
const OBSERVATION_HOOK_BODY_LEN: usize = 1024 * 1024 + 1;
|
||||
|
||||
struct ObservationBodyHook {
|
||||
object: &'static str,
|
||||
calls: Arc<AtomicU64>,
|
||||
body: bytes::Bytes,
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl crate::object_api::GetObjectBodyCacheHook for ObservationBodyHook {
|
||||
async fn lookup(&self, bucket: &str, object: &str, info: &ObjectInfo) -> Option<bytes::Bytes> {
|
||||
if bucket == crate::disk::RUSTFS_META_BUCKET && object == self.object {
|
||||
assert_eq!(info.size, i64::try_from(self.body.len()).expect("hook fixture size"));
|
||||
self.calls.fetch_add(1, Ordering::SeqCst);
|
||||
Some(self.body.clone())
|
||||
} else {
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
struct ObservationHookGuard;
|
||||
|
||||
impl Drop for ObservationHookGuard {
|
||||
fn drop(&mut self) {
|
||||
crate::object_api::unregister_get_object_body_cache_hook();
|
||||
}
|
||||
}
|
||||
|
||||
async fn ordinary_observation_get(set: &SetDisks, object: &str) -> Vec<u8> {
|
||||
let mut reader = set
|
||||
.get_object_reader(crate::disk::RUSTFS_META_BUCKET, object, None, HeaderMap::new(), &ObjectOptions::default())
|
||||
.await
|
||||
.expect("ordinary GET positive control");
|
||||
let mut bytes = Vec::new();
|
||||
reader.stream.read_to_end(&mut bytes).await.expect("ordinary GET body");
|
||||
bytes
|
||||
}
|
||||
|
||||
fn observation_register_hook(object: &'static str, body: Vec<u8>) -> (ObservationHookGuard, Arc<AtomicU64>) {
|
||||
let calls = Arc::new(AtomicU64::new(0));
|
||||
crate::object_api::register_get_object_body_cache_hook(Arc::new(ObservationBodyHook {
|
||||
object,
|
||||
calls: calls.clone(),
|
||||
body: bytes::Bytes::from(body),
|
||||
}));
|
||||
(ObservationHookGuard, calls)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial_test::serial(body_cache_hook)]
|
||||
async fn observation_real_backend_boundaries() {
|
||||
let object = "config/observation-real.bin";
|
||||
let payload = b"persisted config";
|
||||
let (_dirs, set) = make_local_set_disks(4, 2).await;
|
||||
let mut reader = PutObjReader::from_vec(payload.to_vec());
|
||||
let written = set
|
||||
.put_object(crate::disk::RUSTFS_META_BUCKET, object, &mut reader, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("persist real config fixture");
|
||||
let result = crate::config::com::read_config_observation(set.clone(), object, payload.len()).await;
|
||||
assert_eq!(result.is_ok(), true, "I05_OBSERVATION_SUCCESS: {result:?}");
|
||||
let (bytes, info) = result.expect("real observation success");
|
||||
assert_eq!(bytes, payload);
|
||||
assert_eq!(info.size, written.size);
|
||||
assert_eq!(info.etag, written.etag);
|
||||
assert_eq!(info.data_dir, written.data_dir);
|
||||
assert_eq!(info.name, object);
|
||||
let over = crate::config::com::read_config_observation(set.clone(), object, payload.len() - 1).await;
|
||||
assert_eq!(over.is_err(), true, "I05_BOUND_ENFORCED");
|
||||
assert_eq!(over.expect_err("over limit must fail"), Error::CorruptedFormat, "I05_BOUND_ENFORCED");
|
||||
let absent = crate::config::com::read_config_observation(set.clone(), "config/observation-absent.bin", 32).await;
|
||||
assert_eq!(absent.expect_err("absent config"), Error::ConfigNotFound);
|
||||
let empty_object = "config/observation-empty.bin";
|
||||
let mut empty = PutObjReader::from_vec(Vec::new());
|
||||
set.put_object(crate::disk::RUSTFS_META_BUCKET, empty_object, &mut empty, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("persist empty config");
|
||||
let empty = crate::config::com::read_config_observation(set, empty_object, 0).await;
|
||||
assert!(empty.is_ok(), "existing empty config must succeed: {empty:?}");
|
||||
let (bytes, info) = empty.expect("empty observation");
|
||||
assert!(bytes.is_empty());
|
||||
assert_eq!(info.size, 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial_test::serial(body_cache_hook)]
|
||||
async fn observation_bypasses_registered_body_hook() {
|
||||
let object = "config/observation-hook.bin";
|
||||
let (_dirs, set) = make_local_set_disks(4, 2).await;
|
||||
let mut reader = PutObjReader::from_vec(vec![b'p'; OBSERVATION_HOOK_BODY_LEN]);
|
||||
set.put_object(crate::disk::RUSTFS_META_BUCKET, object, &mut reader, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("persist real config fixture");
|
||||
let (_hook, calls) = observation_register_hook(object, vec![b'x'; OBSERVATION_HOOK_BODY_LEN]);
|
||||
assert_eq!(ordinary_observation_get(&set, object).await, vec![b'x'; OBSERVATION_HOOK_BODY_LEN]);
|
||||
assert_eq!(calls.swap(0, Ordering::SeqCst), 1);
|
||||
let result = crate::config::com::read_config_observation(set, object, OBSERVATION_HOOK_BODY_LEN).await;
|
||||
assert_eq!(result.is_ok(), true, "I05_HOOK_BYPASS: {result:?}");
|
||||
assert_eq!(
|
||||
result.expect("observed persisted body").0,
|
||||
vec![b'p'; OBSERVATION_HOOK_BODY_LEN],
|
||||
"I05_HOOK_BYPASS"
|
||||
);
|
||||
assert_eq!(calls.load(Ordering::SeqCst), 0, "I05_HOOK_BYPASS");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial_test::serial(body_cache_hook)]
|
||||
async fn observation_hook_scope_restores_after_error() {
|
||||
let object = "config/observation-hook-error.bin";
|
||||
let (_dirs, set) = make_local_set_disks(4, 2).await;
|
||||
let mut reader = PutObjReader::from_vec(vec![b'p'; OBSERVATION_HOOK_BODY_LEN]);
|
||||
set.put_object(crate::disk::RUSTFS_META_BUCKET, object, &mut reader, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("persist real config fixture");
|
||||
let (_hook, calls) = observation_register_hook(object, vec![b'x'; OBSERVATION_HOOK_BODY_LEN]);
|
||||
let result = crate::config::com::read_config_observation(set.clone(), object, OBSERVATION_HOOK_BODY_LEN - 1).await;
|
||||
assert_eq!(result.expect_err("bounded observation error"), Error::CorruptedFormat);
|
||||
assert_eq!(calls.load(Ordering::SeqCst), 0);
|
||||
assert_eq!(ordinary_observation_get(&set, object).await, vec![b'x'; OBSERVATION_HOOK_BODY_LEN]);
|
||||
assert_eq!(calls.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
|
||||
struct PendingObservationStore {
|
||||
set: Arc<SetDisks>,
|
||||
ready: std::sync::Mutex<Option<tokio::sync::oneshot::Sender<()>>>,
|
||||
}
|
||||
|
||||
impl std::fmt::Debug for PendingObservationStore {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
f.debug_struct("PendingObservationStore").finish()
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl crate::storage_api_contracts::object::ObjectIO for PendingObservationStore {
|
||||
type Error = Error;
|
||||
type RangeSpec = HTTPRangeSpec;
|
||||
type HeaderMap = HeaderMap;
|
||||
type ObjectOptions = ObjectOptions;
|
||||
type ObjectInfo = ObjectInfo;
|
||||
type GetObjectReader = GetObjectReader;
|
||||
type PutObjectReader = PutObjReader;
|
||||
|
||||
async fn get_object_reader(
|
||||
&self,
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
range: Option<HTTPRangeSpec>,
|
||||
headers: HeaderMap,
|
||||
opts: &ObjectOptions,
|
||||
) -> Result<GetObjectReader> {
|
||||
let mut reader = self.set.get_object_reader(bucket, object, range, headers, opts).await?;
|
||||
let mut body = Vec::new();
|
||||
reader.stream.read_to_end(&mut body).await.expect("pending producer real GET");
|
||||
assert_eq!(
|
||||
body,
|
||||
vec![b'p'; OBSERVATION_HOOK_BODY_LEN],
|
||||
"I05_HOOK_BYPASS: producer must enter its own scope"
|
||||
);
|
||||
self.ready
|
||||
.lock()
|
||||
.expect("pending signal")
|
||||
.take()
|
||||
.expect("one GET")
|
||||
.send(())
|
||||
.expect("producer entry signal");
|
||||
std::future::pending::<Result<GetObjectReader>>().await
|
||||
}
|
||||
|
||||
async fn put_object(
|
||||
&self,
|
||||
_bucket: &str,
|
||||
_object: &str,
|
||||
_data: &mut PutObjReader,
|
||||
_opts: &ObjectOptions,
|
||||
) -> Result<ObjectInfo> {
|
||||
assert!(false, "observation must not PUT");
|
||||
Err(Error::other("unexpected observation PUT"))
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial_test::serial(body_cache_hook)]
|
||||
async fn observation_hook_scope_restores_after_cancel() {
|
||||
let object = "config/observation-hook-cancel.bin";
|
||||
let (_dirs, set) = make_local_set_disks(4, 2).await;
|
||||
let mut reader = PutObjReader::from_vec(vec![b'p'; OBSERVATION_HOOK_BODY_LEN]);
|
||||
set.put_object(crate::disk::RUSTFS_META_BUCKET, object, &mut reader, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("persist real config fixture");
|
||||
let (_hook, calls) = observation_register_hook(object, vec![b'x'; OBSERVATION_HOOK_BODY_LEN]);
|
||||
assert_eq!(ordinary_observation_get(&set, object).await, vec![b'x'; OBSERVATION_HOOK_BODY_LEN]);
|
||||
assert_eq!(calls.swap(0, Ordering::SeqCst), 1);
|
||||
let (ready_tx, ready_rx) = tokio::sync::oneshot::channel();
|
||||
let adapter = Arc::new(PendingObservationStore {
|
||||
set: set.clone(),
|
||||
ready: std::sync::Mutex::new(Some(ready_tx)),
|
||||
});
|
||||
let mut scope = Box::pin(crate::config::com::read_config_observation(adapter, object, OBSERVATION_HOOK_BODY_LEN));
|
||||
tokio::select! {
|
||||
result = ready_rx => { result.expect("producer reached pending GET"); }
|
||||
result = &mut scope => { assert!(false, "producer must remain pending until cancelled: {result:?}"); }
|
||||
}
|
||||
assert_eq!(calls.load(Ordering::SeqCst), 0);
|
||||
drop(scope);
|
||||
assert_eq!(ordinary_observation_get(&set, object).await, vec![b'x'; OBSERVATION_HOOK_BODY_LEN]);
|
||||
assert_eq!(calls.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
|
||||
const READ_VERSION_BARRIER_GUARD: std::time::Duration = std::time::Duration::from_secs(10);
|
||||
|
||||
fn object_with_initial_data_shards(bucket: &str, prefix: &str) -> String {
|
||||
|
||||
Reference in New Issue
Block a user