mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-27 22:37:13 +00:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| b7fe40f4b3 |
@@ -1,3 +1,40 @@
|
||||
fn read_native_range_in<R: std::ops::RangeBounds<u64>>(
|
||||
tx: &redb::ReadTransaction,
|
||||
range: R,
|
||||
) -> Result<Vec<crate::CaptureRaftEntry>, StoreError> {
|
||||
use std::ops::Bound::{Excluded, Included, Unbounded};
|
||||
let logs = tx.open_table(LOGS)?;
|
||||
let start = match range.start_bound() {
|
||||
Included(value) => *value,
|
||||
Excluded(value) => match value.checked_add(1) {
|
||||
Some(value) => value,
|
||||
None => return Ok(Vec::new()),
|
||||
},
|
||||
Unbounded => 0,
|
||||
};
|
||||
let end = match range.end_bound() {
|
||||
Included(value) => *value,
|
||||
Excluded(value) => match value.checked_sub(1) {
|
||||
Some(value) => value,
|
||||
None => return Ok(Vec::new()),
|
||||
},
|
||||
Unbounded => u64::MAX,
|
||||
};
|
||||
if start > end {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
let mut entries = Vec::new();
|
||||
for row in logs.range(start..=end)? {
|
||||
let (key, value) = row?;
|
||||
let entry = decode_log(value.value())?;
|
||||
if entry.id.index != key.value() {
|
||||
return Err(StoreError::CorruptRecord);
|
||||
}
|
||||
entries.push(decode_native_entry(entry)?);
|
||||
}
|
||||
Ok(entries)
|
||||
}
|
||||
|
||||
use crate::{
|
||||
LOGS, LogId, META, Store, StoreError, committed::validate_committed, created::*, decode_log, encode_id, encode_log,
|
||||
next_index, number,
|
||||
@@ -368,6 +405,14 @@ impl Store {
|
||||
})
|
||||
.transpose()
|
||||
}
|
||||
/// Read typed native entries within the requested bounds.
|
||||
pub fn read_native_range<R: std::ops::RangeBounds<u64>>(&self, range: R) -> Result<Vec<crate::CaptureRaftEntry>, StoreError> {
|
||||
let tx = self.database.begin_read()?;
|
||||
if number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
|
||||
return Err(StoreError::UnsupportedSchema);
|
||||
}
|
||||
read_native_range_in(&tx, range)
|
||||
}
|
||||
pub fn read_native_entries(&self, start: u64, end: u64) -> Result<Vec<crate::CaptureRaftEntry>, StoreError> {
|
||||
let tx = self.database.begin_read()?;
|
||||
if number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
|
||||
@@ -376,18 +421,9 @@ impl Store {
|
||||
if start > end {
|
||||
return Err(StoreError::InvalidLogSequence);
|
||||
}
|
||||
let logs = tx.open_table(LOGS)?;
|
||||
let mut entries = Vec::new();
|
||||
for row in logs.range(start..end)? {
|
||||
let (key, value) = row?;
|
||||
let entry = decode_log(value.value())?;
|
||||
if entry.id.index != key.value() {
|
||||
return Err(StoreError::CorruptRecord);
|
||||
}
|
||||
entries.push(decode_native_entry(entry)?);
|
||||
}
|
||||
Ok(entries)
|
||||
read_native_range_in(&tx, start..end)
|
||||
}
|
||||
|
||||
pub fn apply_next_native(&mut self, id: LogId) -> Result<crate::CaptureResponse, StoreError> {
|
||||
let tx = self.write_transaction()?;
|
||||
let response;
|
||||
|
||||
@@ -535,6 +535,89 @@ mod tests {
|
||||
)
|
||||
}
|
||||
#[test]
|
||||
fn u18_native_read_roundtrip() {
|
||||
let (_dir, _backend, mut store) = u17a_store();
|
||||
let command = u17a_command();
|
||||
let entries = vec![u17a_blank(0), u17a_membership(1).0, u17a_created(2, command.clone())];
|
||||
store.append_native(&entries).expect("append native fixture");
|
||||
let result = store.read_native_range(..);
|
||||
assert!(result.is_ok(), "U18_NATIVE_READ_SUCCESS: {result:?}");
|
||||
let read = result.expect("native read");
|
||||
assert_eq!(read, entries, "full identities and payloads");
|
||||
assert_eq!(
|
||||
store.read_entries(2, 3).expect("canonical row")[0].payload[14..],
|
||||
command.encode().expect("canonical command")
|
||||
);
|
||||
}
|
||||
#[test]
|
||||
fn u18_native_read_range_bounds() {
|
||||
use std::ops::Bound::{Excluded, Included, Unbounded};
|
||||
let (_dir, _backend, mut store) = u17a_store();
|
||||
let entries = vec![u17a_blank(0), u17a_membership(1).0, u17a_created(2, u17a_command())];
|
||||
store.append_native(&entries).expect("append range fixture");
|
||||
assert_eq!(store.read_native_range(1..2).expect("bounded read"), entries[1..2], "U18_RANGE_BOUNDARY");
|
||||
assert_eq!(store.read_native_range(..=1).expect("inclusive end"), entries[..2]);
|
||||
assert_eq!(store.read_native_range((Excluded(0), Included(2))).expect("excluded start"), entries[1..]);
|
||||
assert_eq!(store.read_native_range(2..).expect("unbounded end"), entries[2..]);
|
||||
assert_eq!(store.read_native_range(..).expect("all"), entries);
|
||||
for range in [
|
||||
(Included(2), Excluded(2)),
|
||||
(Included(2), Included(1)),
|
||||
(Unbounded, Excluded(0)),
|
||||
(Excluded(u64::MAX), Unbounded),
|
||||
(Included(u64::MAX), Included(u64::MAX)),
|
||||
] {
|
||||
assert!(store.read_native_range(range).expect("empty mathematical range").is_empty());
|
||||
}
|
||||
}
|
||||
#[test]
|
||||
fn u18_native_read_reopen() {
|
||||
let (dir, backend, mut store) = u17a_store();
|
||||
let entries = vec![u17a_blank(0), u17a_membership(1).0, u17a_created(2, u17a_command())];
|
||||
store.append_native(&entries).expect("append reopen fixture");
|
||||
drop(store);
|
||||
drop(backend);
|
||||
let store = Store::open(dir.path().join("db"), identity()).expect("reopen native file");
|
||||
assert_eq!(store.read_native_range(..).expect("reopened native read"), entries);
|
||||
}
|
||||
#[test]
|
||||
fn u18_native_read_legacy_rejected() {
|
||||
let (_dir, _backend, store) = u17a_bound();
|
||||
assert!(
|
||||
matches!(store.read_native_range(0..0), Err(StoreError::UnsupportedSchema)),
|
||||
"U18_SCHEMA_GUARD"
|
||||
);
|
||||
assert!(matches!(store.read_native_range(..), Err(StoreError::UnsupportedSchema)));
|
||||
}
|
||||
#[test]
|
||||
fn u18_native_read_corrupt_row() {
|
||||
let (_dir, _backend, mut store) = u17a_store();
|
||||
store.append_native(&[u17a_blank(0)]).expect("append corruption fixture");
|
||||
let original = store.read_entry(0).expect("read stored entry").expect("present entry");
|
||||
let tx = store.database.begin_write().expect("corruption transaction");
|
||||
{
|
||||
let mut logs = tx.open_table(LOGS).expect("logs");
|
||||
let bytes = logs.get(0).expect("get row").expect("present row").value().to_vec();
|
||||
logs.insert(1, bytes.as_slice()).expect("mismatched index row");
|
||||
}
|
||||
tx.commit().expect("persist mismatched row");
|
||||
assert!(matches!(store.read_native_range(1..=1), Err(StoreError::CorruptRecord)));
|
||||
let tx = store.database.begin_write().expect("malformed transaction");
|
||||
{
|
||||
let mut logs = tx.open_table(LOGS).expect("logs");
|
||||
let mut entry = original;
|
||||
entry.payload = b"CENR\x01".to_vec();
|
||||
let bytes = encode_log(&entry).expect("encode malformed native envelope");
|
||||
logs.insert(0, bytes.as_slice()).expect("malformed native row");
|
||||
}
|
||||
tx.commit().expect("persist malformed native row");
|
||||
assert!(matches!(store.read_native_range(..1), Err(StoreError::CorruptRecord)));
|
||||
let tx = store.database.begin_write().expect("missing-table transaction");
|
||||
assert!(tx.delete_table(LOGS).expect("delete logs table"));
|
||||
tx.commit().expect("persist missing logs table");
|
||||
assert!(store.read_native_entries(0, 0).is_err(), "U18_EMPTY_RANGE_TABLE_ERROR");
|
||||
}
|
||||
#[test]
|
||||
fn u17a_native_enable_empty_idempotent() {
|
||||
let (_dir, backend, mut store) = u17a_bound();
|
||||
assert_eq!(committed_schema(&store), 3);
|
||||
|
||||
Reference in New Issue
Block a user