mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-07-19 14:32:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(ec): read chunk-manifest chunks stored on EC volumes (rust volume server) (#10187)

Browse Source 
* fix(ec): read chunk-manifest chunks stored on EC volumes

Chunk-manifest expansion read every chunk through store.read_volume_needle,
which only resolves a local regular volume. Once a chunk's volume is
EC-encoded, that lookup returns NotFound and the GET fails 500 with
"read chunk ...: not found", so a chunked object over an EC tier is
unreadable even though its parity is intact and reconstructable.

Resolve each chunk to wherever it lives — a local regular volume, a
local EC volume (reconstruct-on-read from the surviving shards), or a
peer via master lookup — matching Go's ChunkedFileReader, which never
assumes chunks are local regular needles.

* fix(ec): validate the chunk cookie on local manifest chunk reads

A chunk fetched from a peer is cookie-checked by that peer's GET handler,
but the local regular and EC reads returned data without comparing the
needle's cookie to the one in the chunk fid. Check it, matching the main
GET paths, so a stale or guessed id can't serve another needle's bytes.

* fix(ec): clamp manifest chunk copy to its declared size

Expansion writes each chunk into result[offset..] by offset, so a chunk
whose bytes exceed its declared size could overwrite the next chunk's
window. Clamp the copy to chunk.size (and reject a negative size) so an
over-long or malformed chunk stays within its own range.
This commit is contained in:
Chris Lu
2026-07-01 13:38:34 -07:00
committed by GitHub
parent 746ed82662
commit f6032cf23d
2 changed files with 257 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
seaweed-volume/src/server/handlers.rs
+157 -69
View File
@@ -1262,7 +1262,9 @@ async fn get_or_head_handler_inner(
&query,
&etag,
&last_modified_str,
) {
)
.await
{
return resp;
}
// If manifest expansion fails (invalid JSON etc.), fall through to raw data
@@ -3179,12 +3181,11 @@ struct ChunkManifest {
struct ChunkInfo {
fid: String,
offset: i64,
#[allow(dead_code)]
size: i64,
}
/// Try to expand a chunk manifest needle. Returns None if manifest can't be parsed.
fn try_expand_chunk_manifest(
async fn try_expand_chunk_manifest(
state: &Arc<VolumeServerState>,
n: &Needle,
_headers: &HeaderMap,
@@ -3224,29 +3225,26 @@ fn try_expand_chunk_manifest(
return None;
}
// Read and concatenate all chunks
// Read and concatenate all chunks. Each chunk is resolved to wherever it
// lives — a local regular volume, a local EC volume (reconstruct-on-read),
// or a peer via master lookup — mirroring Go's ChunkedFileReader, which
// never assumes chunks are local regular needles.
let mut result = vec![0u8; manifest.size as usize];
let store = state.store.read().unwrap();
for chunk in &manifest.chunks {
let (chunk_vid, chunk_nid, chunk_cookie) = match parse_url_path(&chunk.fid) {
Some(p) => p,
None => {
return Some(
(
StatusCode::INTERNAL_SERVER_ERROR,
format!("invalid chunk fid: {}", chunk.fid),
)
.into_response(),
// Validate the attacker-controlled chunk offset before indexing: a
// negative value would wrap to a huge usize, and an out-of-range one has
// nowhere to land.
if chunk.offset < 0 || chunk.size < 0 {
return Some(
(
StatusCode::INTERNAL_SERVER_ERROR,
format!("invalid negative chunk offset/size in {}", chunk.fid),
)
}
};
let mut chunk_needle = Needle {
id: chunk_nid,
cookie: chunk_cookie,
..Needle::default()
};
match store.read_volume_needle(chunk_vid, &mut chunk_needle) {
Ok(_) => {}
.into_response(),
);
}
let data = match read_chunk_needle(state, &chunk.fid).await {
Ok(d) => d,
Err(e) => {
return Some(
(
@@ -3256,56 +3254,16 @@ fn try_expand_chunk_manifest(
.into_response(),
)
}
}
// Validate the attacker-controlled chunk offset before indexing: a
// negative value would wrap to a huge usize, and an out-of-range one has
// nowhere to land.
if chunk.offset < 0 {
return Some(
(
StatusCode::INTERNAL_SERVER_ERROR,
format!("invalid negative chunk offset in {}", chunk.fid),
)
.into_response(),
);
}
};
let offset = chunk.offset as usize;
if offset >= result.len() {
continue;
}
// Write the chunk into its window in `result`. Compressed chunks inflate
// directly into the destination slice (bounded by the remaining window),
// so a chunk never allocates a second large buffer — peak memory stays at
// ~manifest.size instead of doubling. Bytes past the window are dropped,
// matching the prior truncation behavior.
if chunk_needle.is_compressed() {
use flate2::read::GzDecoder;
use std::io::Read as _;
let mut decoder = GzDecoder::new(&chunk_needle.data[..]);
let window = &mut result[offset..];
let mut written = 0usize;
let mut decode_failed = false;
while written < window.len() {
match decoder.read(&mut window[written..]) {
Ok(0) => break,
Ok(n) => written += n,
Err(_) => {
decode_failed = true;
break;
}
}
}
// On any decode failure, drop the partial output and fall back to the
// chunk's raw bytes (truncated to the window), preserving prior behavior.
if decode_failed {
window[..written].fill(0);
let copy_len = chunk_needle.data.len().min(window.len());
window[..copy_len].copy_from_slice(&chunk_needle.data[..copy_len]);
}
} else {
let copy_len = chunk_needle.data.len().min(result.len() - offset);
result[offset..offset + copy_len].copy_from_slice(&chunk_needle.data[..copy_len]);
}
// Clamp to the chunk's declared size so an over-long chunk can't bleed
// into the next chunk's window; also drop bytes past the buffer end.
let bound = (chunk.size as usize).min(result.len() - offset);
let copy_len = data.len().min(bound);
result[offset..offset + copy_len].copy_from_slice(&data[..copy_len]);
}
// Determine filename: URL path filename, then manifest name
@@ -3451,6 +3409,136 @@ fn try_expand_chunk_manifest(
Some((StatusCode::OK, response_headers, result).into_response())
}
/// Read one chunk-manifest chunk's final (decompressed) content bytes from
/// wherever it lives: a local regular volume, a local EC volume
/// (reconstruct-on-read from surviving shards), or a peer resolved via the
/// master. Mirrors Go's ChunkedFileReader, which looks every chunk up through
/// the master instead of assuming a local regular needle.
async fn read_chunk_needle(
state: &Arc<VolumeServerState>,
fid: &str,
) -> Result<Vec<u8>, String> {
let (vid, nid, cookie) =
parse_url_path(fid).ok_or_else(|| format!("invalid chunk fid: {}", fid))?;
// Decide where the chunk lives under one store read lock; drop it before any
// await (the EC and remote paths are async).
enum Placement {
Ec,
Remote,
}
let placement = {
let store = state.store.read().unwrap();
if store.find_volume(vid).is_some() {
let mut n = Needle {
id: nid,
cookie,
..Needle::default()
};
return store
.read_volume_needle(vid, &mut n)
.map_err(|e| format!("{}", e))
.and_then(|_| cookie_checked_chunk(n, cookie));
} else if store.find_ec_volume(vid).is_some() {
Placement::Ec
} else {
Placement::Remote
}
};
match placement {
Placement::Ec => {
match crate::server::store_ec::read_ec_shard_needle_distributed(state, vid, nid).await {
Ok(Some(n)) => cookie_checked_chunk(n, cookie),
Ok(None) => Err("not found".to_string()),
Err(e) => Err(format!("{}", e)),
}
}
// The peer serves through its own GET handler, which validates the cookie.
Placement::Remote => read_remote_chunk_needle(state, vid, fid).await,
}
}
/// Validate a locally-read chunk's cookie against the one in its fid, then return
/// its content bytes. The main GET paths check the cookie after a read; a chunk
/// read must do the same so a stale/guessed id can't serve another needle's data.
fn cookie_checked_chunk(n: Needle, cookie: Cookie) -> Result<Vec<u8>, String> {
if n.cookie != cookie {
return Err("not found".to_string());
}
decompress_chunk(n)
}
/// Return a needle's content bytes, decompressing gzip payloads the way the read
/// handler does for a client that did not ask for gzip.
fn decompress_chunk(n: Needle) -> Result<Vec<u8>, String> {
if n.is_compressed() {
match maybe_decompress_gzip(&n.data) {
Ok(d) => Ok(d),
Err(GunzipError::TooLarge) => {
Err("compressed chunk exceeds decompression limit".to_string())
}
// Not valid gzip; keep the raw bytes, matching the prior fallback.
Err(GunzipError::Decode) => Ok(n.data),
}
} else {
Ok(n.data)
}
}
/// Fetch a chunk that is not hosted locally from a peer volume server. The peer
/// serves the final (decompressed) bytes whether the chunk is on a regular or EC
/// volume, so a chunk whose EC shards live elsewhere is still reconstructed on
/// the holder's side. Mirrors Go's ChunkedFileReader.readChunkNeedle.
async fn read_remote_chunk_needle(
state: &Arc<VolumeServerState>,
vid: VolumeId,
fid: &str,
) -> Result<Vec<u8>, String> {
let locations = lookup_volume(
&state.http_client,
&state.outgoing_http_scheme,
&state.master_url,
vid.0,
)
.await?;
if locations.is_empty() {
return Err("not found".to_string());
}
let mut last_err = String::new();
for loc in &locations {
// Skip self: the local paths already ruled this server out.
if loc.url.contains(&state.self_url) {
continue;
}
let target_http = to_http_address(&loc.url);
let url = match normalize_outgoing_http_url(
&state.outgoing_http_scheme,
&format!("{}/{}?proxied=true", target_http, fid),
) {
Ok(u) => u,
Err(e) => {
last_err = e;
continue;
}
};
match state.http_client.get(&url).send().await {
Ok(resp) if resp.status().is_success() => match resp.bytes().await {
Ok(b) => return Ok(b.to_vec()),
Err(e) => last_err = format!("read body from {}: {}", url, e),
},
Ok(resp) => last_err = format!("{} returned {}", url, resp.status()),
Err(e) => last_err = format!("request to {} failed: {}", url, e),
}
}
Err(if last_err.is_empty() {
"not found".to_string()
} else {
last_err
})
}
// ============================================================================
// Helpers
// ============================================================================
seaweed-volume/tests/http_integration.rs
+100
View File
@@ -834,3 +834,103 @@ async fn replicate_write_does_not_re_replicate() {
"plain write to a multi-copy volume must attempt replication"
);
}
// ============================================================================
// Chunk-manifest expansion resolves chunks on EC volumes
//
// A chunked object whose data chunks live on an EC-encoded volume must expand
// by reconstruct-on-read from the shards, not by a local regular-volume lookup
// (which finds nothing once the volume is EC-encoded). Mirrors Go's
// ChunkedFileReader, which resolves every chunk through the master.
// ============================================================================
#[tokio::test]
async fn chunk_manifest_expands_chunk_stored_on_ec_volume() {
use seaweed_volume::storage::erasure_coding::ec_encoder::write_ec_files;
use seaweed_volume::storage::needle::needle::{FileId, Needle};
use seaweed_volume::storage::types::{Cookie, NeedleId};
use seaweed_volume::storage::volume::Volume;
let (state, tmp) = test_state();
let dir = tmp.path().to_str().unwrap();
// A chunk large enough to be worth EC-encoding; its bytes are the payload we
// expect the manifest GET to return.
let chunk_data: Vec<u8> = (0..4096u32).map(|i| (i % 251) as u8).collect();
let chunk_nid = NeedleId(0x2a);
let chunk_cookie = Cookie(0x1234abcd);
// Build regular volume 2 holding the chunk, then EC-encode it and mount all
// 14 shards locally so reconstruct-on-read is a pure local read.
{
let mut v = Volume::new(
dir,
dir,
"",
VolumeId(2),
NeedleMapKind::InMemory,
None,
None,
0,
Version::current(),
)
.unwrap();
let mut n = Needle {
id: chunk_nid,
cookie: chunk_cookie,
data: chunk_data.clone(),
data_size: chunk_data.len() as u32,
..Needle::default()
};
v.write_needle(&mut n, true).unwrap();
v.sync_to_disk().unwrap();
v.close();
}
write_ec_files(dir, dir, "", VolumeId(2), 10, 4).unwrap();
// Volume 2 was built standalone (never registered in the store), so it only
// exists as EC shards — the chunk resolves through the EC path, as it would
// after ec.encode retired the regular volume.
{
let mut store = state.store.write().unwrap();
let shard_ids: Vec<u32> = (0..14).collect();
store.mount_ec_shards(VolumeId(2), "", &shard_ids).unwrap();
}
// Write the chunk-manifest needle to regular volume 1.
let chunk_fid = FileId::new(VolumeId(2), chunk_nid, chunk_cookie).to_string();
let manifest = format!(
r#"{{"name":"big.bin","mime":"application/octet-stream","size":{},"chunks":[{{"fid":"{}","offset":0,"size":{}}}]}}"#,
chunk_data.len(),
chunk_fid,
chunk_data.len()
);
let manifest_nid = NeedleId(0x7);
let manifest_cookie = Cookie(0x55667788);
{
let mut store = state.store.write().unwrap();
let mut n = Needle {
id: manifest_nid,
cookie: manifest_cookie,
data: manifest.into_bytes(),
..Needle::default()
};
n.data_size = n.data.len() as u32;
n.set_is_chunk_manifest();
store.write_volume_needle(VolumeId(1), &mut n).unwrap();
}
// GET the manifest object; expect the reconstructed chunk bytes.
let manifest_fid = FileId::new(VolumeId(1), manifest_nid, manifest_cookie).to_string();
let app = build_admin_router(state.clone());
let response = app
.oneshot(
Request::builder()
.uri(format!("/{}", manifest_fid))
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(body_bytes(response).await, chunk_data);
}