gian/seaweedfs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(ec): don't mix EC shards from different encode runs (#9880)

Browse Source 
* feat(ec): add encode_ts_ns to EC shard metadata and the shard read RPC

EcShardConfig and VolumeEcShardReadRequest gain an int64 encode_ts_ns
(encode time in unix nanos). It rides in .vif and the read request so a
read can be scoped to the encode run that produced the index.

* fix(ec): stamp each encode and reject cross-run shard reads

Generate stamps EncodeTsNs into the volume's .vif. Reads carry it to the
shard's owning volume (resolved together via FindEcVolumeWithShard, so a
multi-disk server validates the disk that actually serves the bytes) and
reject a shard from a different encode run, recovering from parity. A
zero on either side (pre-upgrade volume) skips the guard.

* fix(ec): stamp the encode identity on the worker-generated .vif

The worker-local encode path now writes EncodeTsNs (and the resolved EC
ratio) into the .vif, so the read guard is not silently off for volumes
encoded by the maintenance worker.

* fix(ec): wipe stale EC artifacts before re-encoding

VolumeEcShardsGenerate evicts any in-memory EcVolume for the volume and
removes its on-disk shard/index/sidecar files before writing fresh ones,
so a retried encode never builds on a partial prior run and the unlink
frees the inodes instead of leaving open fds serving old bytes.

* fix(ec): unmount EC shards across all disks

UnmountEcShards walked only the first disk holding the shard, leaving a
duplicate copy mounted on a sibling disk (split-disk reconciled volumes)
still serving and heartbeating. Traverse every disk and emit one
deletion delta per disk.

* fix(ec): delete orphan shards without a local .ecx

deleteEcShardIdsForEachLocation gated shard-file removal on a local .ecx,
so it could not clean an orphan .ecNN left by a failed copy on a disk
with no index. Delete the requested shard files unconditionally; the
index-file (.ecx/.ecj/.vif) routing stays gated as before.

* fix(ec): clear stale EC shards cluster-wide before re-encoding

ec.encode unmounts and deletes EC shards for the target volumes on every
node before regenerating: fatal for the shards the topology reports
(mounted leftovers), best-effort for the rest (a sweep that catches
unmounted failed-copy orphans). A down node is a no-op.

* fix(ec): don't nil EC fds on close so reads can't race eviction

A reader resolves an EcVolume/shard under the lock then reads after it is
released, so an eviction that nils ecxFile/ecdFile would race that read
and panic. Close the fds without nilling the fields: the field is now
write-once (no data race) and a concurrent read hits a closed fd, getting
a clean error that the caller recovers from parity.

* fix(ec): wipe stale EC artifacts on every disk and surface failures

The pre-encode wipe only deleted beside the source volume, so a stale
shard on a sibling disk survived and could be mounted against the new
index at reconcile. Sweep every disk. Removal also ignored os.Remove
errors, reporting a failed cleanup as success and letting a stale shard
join the next generation; surface the first real failure (treating
already-gone as success) from removeStaleEcArtifacts and the shard delete.

* fix(ec): log when a local shard is skipped for a different encode run

The cross-run guard returned errShardNotLocal, indistinguishable in logs
from a genuinely-absent shard. Add a V(1) line naming both EncodeTsNs so
operators can tell "wrong encode generation" from "shard not here".

* fix(ec): surface metadata removal failures in the shard delete path

deleteEcShardIdsForEachLocation still dropped os.Remove errors on the
.ecx/.ecj/.vif/sidecar cleanup. A surviving stale .ecx is the orphan-index
condition this path prevents, so route those through removeFileIfExists and
return the first real failure instead of reporting cleanup as success.

* fix(ec): fail orphan cleanup when a reachable node's delete fails

The pre-encode orphan sweep swallowed every error for unreported (node,
volume) pairs. That is only safe for an unreachable node, which cannot
receive this encode's new generation. A reachable node whose delete
genuinely failed (permission/IO) keeps an orphan shard that a later copy
re-stamps with the new run's volume-level .vif identity, so the read guard
would accept stale data. Surface those; stay best-effort only for
unreachable nodes (gRPC Unavailable / no status).

* fix(ec): guard ecjFile under its lock in the EC delete path

EcVolume.Close nils ecjFile under ecjFileAccessLock; a delete that resolved
its .ecx lookup before a concurrent eviction (the generate-time
UnloadEcVolume) could then reach the journal append with a nil fd. Bail
with a clear "volume closed" error under the lock instead.

* fix(ec): reject an unstamped shard when the caller has an encode identity

The read guard required both identities nonzero, so a current (stamped)
caller accepted a holder with identity 0 and could be served a stale
pre-upgrade shard. Reject when the caller is stamped and the holder
differs (including unstamped); stay lenient only when the caller itself
has no identity (pre-upgrade reader). A skipped shard recovers from parity.

* fix(ec): full-teardown delete so cluster cleanup wipes a whole generation

The pre-encode cluster sweep deleted only the listed canonical shards on
remote nodes, leaving index/sidecar (and, on builds with versioned
generations, those too) behind. Add a full_teardown flag to
VolumeEcShardsDelete that evicts the volume and wipes every EC artifact for
it on every disk via removeStaleEcArtifacts; the shell and worker pre-encode
cleanup paths set it. Other delete callers (balance/decode/repair) are
unchanged.

* fix(ec): take ecjFileAccessLock before the nil-check in Sync and Close

Sync and Close read ev.ecjFile before acquiring ecjFileAccessLock while
Close nils it under the lock, a data race on the field. Take the lock
first, then nil-check inside, in both.

* fix(ec): acknowledge full_teardown so a pre-upgrade server can't fake success

An old volume server silently ignores full_teardown and returns success
for an ordinary delete, so the caller wrongly believes the generation was
wiped and copies a fresh gen-0 onto an unwiped node. Echo full_teardown_done
in the response; the worker destination cleanup fails when it is absent, and
the shell cluster sweep fails for a reported (mounted) leftover while staying
best-effort for an unreported node. encode_ts_ns stays an accepted transient
(an old server just skips the new read guard, no regression).

* fix(ec): fail the pre-encode sweep for any reachable node that can't ack teardown

A reachable pre-upgrade server ignores full_teardown and returns success
without wiping an orphan, which a later copy then folds into the new
generation. Treat a missing full_teardown_done ack as fatal for every
reachable node (best-effort only for a gRPC-unreachable one), not just for
topology-reported pairs.

* fix(ec): return the served shard identity and validate it client-side

The encode identity was only enforced server-side, so a pre-upgrade server
ignored the request field and served bytes unchecked. Echo the served
shard's EncodeTsNs on every read response chunk and have the client reject a
mismatch (including 0 from an old server), so the guard holds regardless of
server version; a rejected read recovers from parity.

* fix(ec): reject a short/empty remote shard read instead of serving zeros

doReadRemoteEcShardInterval accepted an immediate EOF or a short stream and
returned success with a partly zero-filled, unvalidated buffer (the server
stamps the identity only on chunks that carry bytes). A non-deleted interval
must arrive whole: require n == len(buf), exempting the is_deleted
short-circuit (n=0), matching readLocalEcShardInterval's local check. A short
read now fails so the caller recovers from parity.

* test(ec): fake volume server echoes the full_teardown acknowledgement

The worker now fails a teardown delete that isn't acknowledged (so a
pre-upgrade server can't silently skip the wipe). The fake server's no-op
VolumeEcShardsDelete returned an empty response, which the worker read as a
skipped teardown and aborted the encode. Echo full_teardown_done.

* feat(ec): mirror the encode-run identity guard + full_teardown into the Rust volume server

The Go volume server stamps an encode-run identity (encode_ts_ns) into the .vif
and rejects a read served from a shard of a different run; full_teardown wipes a
whole generation and acknowledges it. The Rust volume server had none of it.
Mirror the shared logic: load encode_ts_ns from the .vif onto the EcVolume,
stamp it on every read response, and reject a request/response mismatch on both
the server and the distributed-read client (recovering from parity); handle
full_teardown by evicting the volume and wiping every EC artifact on each disk,
echoing full_teardown_done so the caller can detect a server that ignored it.

* fix(ec): remove a stale .vif on full teardown of a shard-only node

A shard copy installs shards + .ecx before .vif, so an interrupted copy after a
teardown could mount the new files under the previous run's identity / version /
shard ratio / dat_file_size carried by the surviving .vif. Remove .vif during
full teardown, gated on .idx absence so a source-volume holder keeps its live
.vif. In Rust this lives in a teardown-only helper so the reconcile / load-
fallback paths (which share the base removal) still preserve .vif.

* fix(ec): treat a missing teardown ack as fatal, not as an unreachable node

isNodeUnreachable returned true for any non-gRPC-status error, so a reachable
pre-upgrade server's missing full_teardown_done ack (a plain error) was
classified unreachable and the unreported pair was silently skipped. Classify
only a real codes.Unavailable as unreachable, and wrap the missing ack in a
sentinel the sweep treats as fatal regardless. A genuinely down node still
surfaces as Unavailable from the RPC and stays best-effort.

* fix(ec): reject a short shard read in the local EC needle reader

read_ec_shard_needle ignored the byte count from shard.read_at and appended the
whole pre-sized buffer, so a truncated shard's zero-filled tail passed the later
length check and parsed as garbage. Require n == buf.len() per interval, erroring
on a short read like the local interval reader already does.

* fix(ec): probe reachability before skipping a node that returns Unavailable

The pre-encode sweep skipped any node whose teardown delete returned
codes.Unavailable, but a reachable volume server in maintenance mode also
returns that code for the maintenance-gated delete, so its stale EC files were
left behind on a node that can still receive the new generation. Confirm with a
non-maintenance-gated empty-target Ping: skip only when the node fails the probe
too (genuinely unreachable).

* fix(ec): use try_exists for the teardown .vif .idx guard

The teardown-only .vif removal gated on Path::exists(), which returns false on a
permission/IO stat error, so a stat failure on a present .idx would read as a
shard-only node and delete the live source volume's .vif. Gate on
try_exists() == Ok(false) instead, preserving the sidecar on any stat error.

* fix(ec): only skip a sweep node when a Ping confirms it is transport-down

The pre-encode sweep skipped a node whenever its teardown delete and a liveness
Ping both failed, but it treated ANY Ping error as down — an application-level
Internal/ResourceExhausted, or Unimplemented from a pre-Ping server, left a
reachable node's stale generation in place. Classify the Ping tri-state and skip
only when it transport-fails with codes.Unavailable; a reachable or inconclusive
node stays fatal.

* fix(ec): exclude sweep-skipped nodes from the encode's rebalance

The pre-encode sweep skips a genuinely-down node best-effort, but the rebalance
then recollected the current topology — a node that recovered between the two
could become a copy target and receive the new generation while still holding
its stale, never-cleared shards. Have the sweep return the skipped set and
exclude those nodes from the rebalance for this encode, so a node we could not
clean cannot receive the new generation. Standalone ec.balance is unaffected.

* fix(ec): re-sweep recovered nodes before generation so they aren't stranded

A node skipped as down by the pre-encode sweep is excluded from the rebalance,
but it can recover and become the generation host — mounting all shards locally,
then being excluded from distribution. Union-only verification accepts all
shards on one node and deletes the originals: a single point of failure. Re-sweep
the skipped nodes just before generation; one whose teardown now succeeds leaves
the skipped set and rebalances normally, while a node still down stays skipped.

* fix(ec): abort the encode if a selected source is still skipped after re-sweep

The re-sweep un-skips a recovered node, but the source was selected before it and
a node can stay down through the re-sweep then recover just in time to be the
generation host — mounting all shards locally while still excluded from the
rebalance, which union-only verification accepts before deleting the originals.
Abort the encode when a selected source remains skipped after the re-sweep.

* fix(ec): batch delete returns retriable 503 when a volume became EC mid-batch

If a volume is not EC at the batch-delete classification but is encoded to EC and
its .dat deleted before the regular-volume mutation, the mutation returns an exact
"not found" that the filer chunk-GC treats as completed, dropping the delete.
Recheck EC presence under the mutation lock and return a retriable 503 with the
"try again" token so the filer requeues it onto the EC path.

* fix(ec): recheck EC state before the regular batch-delete mutation

ec.encode mounts EC shards (copied from the .dat) before deleting the originals,
so a volume can be EC while its .dat still exists. The batch delete only rechecked
EC after a NotFound, so a successful regular-volume delete in that window wrote a
tombstone to the soon-removed .dat — the delete was lost and the needle resurrected
from the pre-tombstone shards. Recheck has_ec_volume under the write lock before
delete_volume_needle and return a retriable 503 so the filer requeues onto the EC path.

* fix(volume): make the metrics push test independent of test order

test_push_metrics_once asserted the pushed body contains the request-counter
family without ever touching the counter — a CounterVec with no children emits
nothing, so the assertion only held when another test had already created a
labelset in the shared registry. Create one in the test itself.
This commit is contained in:
Chris Lu 2026-06-10 22:31:18 -07:00 committed by GitHub
parent 1dd292fb84
commit 79ac279fe1
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
25 changed files with 1077 additions and 122 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
seaweed-volume/proto/volume_server.proto
View File

@ -450,8 +450,10 @@ message VolumeEcShardsDeleteRequest {
uint32 volume_id = 1;
string collection = 2;
repeated uint32 shard_ids = 3;
bool full_teardown = 4; // pre-encode cleanup: wipe every EC artifact + generation for this volume, not just shard_ids
}
message VolumeEcShardsDeleteResponse {
bool full_teardown_done = 1; // set by a new server that performed full_teardown; absent from an old server lets the caller detect the silent no-op
}
message VolumeEcShardsMountRequest {
@ -476,10 +478,13 @@ message VolumeEcShardReadRequest {
int64 offset = 3;
int64 size = 4;
uint64 file_key = 5;
reserved 6;
int64 encode_ts_ns = 7; // caller's expected encode time; the server rejects a shard from a different encode run
}
message VolumeEcShardReadResponse {
bytes data = 1;
bool is_deleted = 2;
int64 encode_ts_ns = 3; // identity of the shard actually served; client rejects a mismatch (0 = pre-upgrade server)
}
message VolumeEcBlobDeleteRequest {
@ -577,6 +582,7 @@ message VolumeInfo {
message EcShardConfig {
uint32 data_shards = 1; // Number of data shards (e.g., 10)
uint32 parity_shards = 2; // Number of parity shards (e.g., 4)
int64 encode_ts_ns = 3; // encode time (unix nanos); a read served from a shard of a different encode run is rejected
}
message OldVersionVolumeInfo {
repeated RemoteFile files = 1;

4
seaweed-volume/src/metrics.rs
View File

@ -424,6 +424,10 @@ mod tests {
#[tokio::test]
async fn test_push_metrics_once() {
register_metrics();
// A CounterVec with no children emits nothing, so create a labelset
// instead of depending on another test having touched the counter
// first (test order is nondeterministic).
REQUEST_COUNTER.with_label_values(&["GET", "200"]).inc();
let captured = Arc::new(Mutex::new(None::<String>));
let captured_clone = captured.clone();

105
seaweed-volume/src/server/grpc_server.rs
View File

@ -399,6 +399,27 @@ impl VolumeServer for VolumeGrpcService {
if !is_ec_volume {
let mut store = self.state.store.write().unwrap();
// Recheck EC state under the write lock before mutating the .dat. The
// is_ec_volume snapshot was taken earlier under a separate read lock;
// ec.encode mounts EC shards (copied from the .dat) BEFORE deleting the
// originals, so the vid can be EC now while the .dat still exists. A
// delete_volume_needle here would tombstone that .dat, which the encode
// then removes — the delete is lost and the needle resurrected from the
// pre-tombstone shards. If the vid is now EC, return a retriable 503 so
// the filer requeues the delete onto the EC path.
if store.has_ec_volume(file_id.volume_id) {
results.push(volume_server_pb::DeleteResult {
file_id: fid_str.clone(),
status: 503,
error: format!(
"volume {} became ec during delete, try again",
file_id.volume_id
),
size: 0,
version: 0,
});
continue;
}
match store.delete_volume_needle(file_id.volume_id, &mut n) {
Ok(size) => {
if size.0 == 0 {
@ -420,13 +441,35 @@ impl VolumeServer for VolumeGrpcService {
}
}
Err(e) => {
results.push(volume_server_pb::DeleteResult {
file_id: fid_str.clone(),
status: 500,
error: e.to_string(),
size: 0,
version: 0,
});
// The volume vanished between the is_ec_volume snapshot and
// this mutation. If an EC volume now occupies the vid (ec.encode
// mounted EC then deleted the .dat under us), the delete belongs
// on the EC journal, not here — return a retriable 503 with the
// "try again" token so the filer requeues it and the retry hits
// the EC path. A bare exact "not found" would be dropped
// permanently by the filer chunk-GC.
if matches!(e, crate::storage::volume::VolumeError::NotFound)
&& store.has_ec_volume(file_id.volume_id)
{
results.push(volume_server_pb::DeleteResult {
file_id: fid_str.clone(),
status: 503,
error: format!(
"volume {} became ec during delete, try again",
file_id.volume_id
),
size: 0,
version: 0,
});
} else {
results.push(volume_server_pb::DeleteResult {
file_id: fid_str.clone(),
status: 500,
error: e.to_string(),
size: 0,
version: 0,
});
}
}
}
} else {
@ -2133,6 +2176,12 @@ impl VolumeServer for VolumeGrpcService {
ec_shard_config: Some(crate::storage::volume::VifEcShardConfig {
data_shards: data_shards,
parity_shards: parity_shards,
// This run's identity; the read path rejects a shard from a
// different encode run.
encode_ts_ns: std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_nanos() as i64,
}),
..Default::default()
};
@ -2593,12 +2642,35 @@ impl VolumeServer for VolumeGrpcService {
self.state.check_maintenance()?;
let req = request.into_inner();
let vid = VolumeId(req.volume_id);
if req.full_teardown {
// Pre-encode cleanup: evict the volume and wipe every EC artifact for it
// on every disk, not just the listed shards, so a remote node retains no
// stale generation that a fresh gen-0 copy would collide with. Echo the
// acknowledgement so the caller can tell a pre-upgrade server apart.
{
let mut store = self.state.store.write().unwrap();
let _ = store.remove_ec_volume(vid);
for loc in &store.locations {
loc.remove_ec_volume_files_full_teardown(&req.collection, vid);
}
}
self.state.volume_state_notify.notify_one();
return Ok(Response::new(
volume_server_pb::VolumeEcShardsDeleteResponse {
full_teardown_done: true,
},
));
}
let mut store = self.state.store.write().unwrap();
store.delete_ec_shards(vid, &req.collection, &req.shard_ids);
drop(store);
self.state.volume_state_notify.notify_one();
Ok(Response::new(
volume_server_pb::VolumeEcShardsDeleteResponse {},
volume_server_pb::VolumeEcShardsDeleteResponse {
full_teardown_done: false,
},
))
}
@ -2672,6 +2744,21 @@ impl VolumeServer for VolumeGrpcService {
))
})?;
// Reject a shard whose identity doesn't match the caller's index; the caller
// then recovers from parity. Lenient only when the caller has no identity
// (pre-upgrade reader): a known caller must not accept an unstamped holder,
// which would serve a stale shard from a different encode run.
if req.encode_ts_ns != 0 && req.encode_ts_ns != ec_vol.encode_ts_ns {
return Err(Status::failed_precondition(format!(
"ec shard {}.{} belongs to a different encode run",
req.volume_id, req.shard_id
)));
}
// Identity of the shard actually served, echoed on every response chunk so
// the client can reject a different encode run even from a pre-upgrade server.
let served_encode_ts_ns = ec_vol.encode_ts_ns;
// Check if the requested needle is deleted (via .ecx index, matching Go)
if req.file_key > 0 {
let needle_id = NeedleId(req.file_key);
@ -2682,6 +2769,7 @@ impl VolumeServer for VolumeGrpcService {
if size.is_deleted() {
let results = vec![Ok(volume_server_pb::VolumeEcShardReadResponse {
is_deleted: true,
encode_ts_ns: served_encode_ts_ns,
..Default::default()
})];
return Ok(Response::new(Box::pin(tokio_stream::iter(results))));
@ -2730,6 +2818,7 @@ impl VolumeServer for VolumeGrpcService {
results.push(Ok(volume_server_pb::VolumeEcShardReadResponse {
data: buf,
is_deleted: false,
encode_ts_ns: served_encode_ts_ns,
}));
if n < chunk_size {
break; // short read means EOF

27
seaweed-volume/src/server/store_ec.rs
View File

@ -70,6 +70,10 @@ struct Snapshot {
intervals: Vec<IntervalResult>,
cached_locations: HashMap<ShardId, Vec<String>>,
cache_refreshed_at: Option<Instant>,
/// This volume's encode identity, carried to peers on remote shard reads so a
/// shard from a different encode run is rejected rather than served at a
/// mismatched offset. 0 for a pre-feature volume (lenient).
encode_ts_ns: i64,
}
/// Top-level entry point. Returns `Ok(None)` for "not found" (matches
@ -147,6 +151,7 @@ pub async fn read_ec_shard_needle_distributed(
&shard_locations,
snapshot.data_shards as usize,
snapshot.parity_shards as usize,
snapshot.encode_ts_ns,
)
.await?;
assembled.push(buf);
@ -259,6 +264,7 @@ fn snapshot_under_lock(
intervals: interval_results,
cached_locations,
cache_refreshed_at,
encode_ts_ns: ecv.encode_ts_ns,
}))
}
@ -383,6 +389,7 @@ async fn fetch_one_interval(
shard_locations: &HashMap<ShardId, Vec<String>>,
data_shards: usize,
parity_shards: usize,
expected_encode_ts_ns: i64,
) -> io::Result<Vec<u8>> {
// Direct peer read against the cached locations for this shard.
if let Some(sources) = shard_locations.get(&shard_id) {
@ -395,6 +402,7 @@ async fn fetch_one_interval(
shard_id,
shard_offset,
size,
expected_encode_ts_ns,
)
.await
{
@ -424,6 +432,7 @@ async fn fetch_one_interval(
shard_locations,
data_shards,
parity_shards,
expected_encode_ts_ns,
)
.await
}
@ -436,6 +445,7 @@ async fn read_remote_ec_shard_interval(
shard_id: ShardId,
shard_offset: i64,
size: usize,
expected_encode_ts_ns: i64,
) -> io::Result<Vec<u8>> {
let mut last_err: Option<io::Error> = None;
for src in sources {
@ -447,6 +457,7 @@ async fn read_remote_ec_shard_interval(
shard_id,
shard_offset,
size,
expected_encode_ts_ns,
)
.await
{
@ -470,6 +481,7 @@ async fn do_read_remote_ec_shard_interval(
shard_id: ShardId,
shard_offset: i64,
size: usize,
expected_encode_ts_ns: i64,
) -> io::Result<Vec<u8>> {
let grpc_addr =
parse_grpc_address(source).map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?;
@ -505,6 +517,7 @@ async fn do_read_remote_ec_shard_interval(
offset: shard_offset,
size: size as i64,
file_key: needle_id.0,
encode_ts_ns: expected_encode_ts_ns,
};
let resp = client
.volume_ec_shard_read(Request::new(req))
@ -523,6 +536,18 @@ async fn do_read_remote_ec_shard_interval(
.await
.map_err(|e| io::Error::new(io::ErrorKind::Other, format!("recv: {}", e)))?
{
// Validate the served shard's identity client-side, so the guard holds even
// against a pre-upgrade server that ignored the request field (returns 0).
// A mismatch fails the read; the caller recovers from parity.
if expected_encode_ts_ns != 0 && msg.encode_ts_ns != expected_encode_ts_ns {
return Err(io::Error::new(
io::ErrorKind::Other,
format!(
"ec shard {}.{} from {} belongs to a different encode run (want {} got {})",
vid.0, shard_id, source, expected_encode_ts_ns, msg.encode_ts_ns
),
));
}
if !msg.data.is_empty() {
out.extend_from_slice(&msg.data);
}
@ -554,6 +579,7 @@ async fn recover_one_remote_ec_shard_interval(
shard_locations: &HashMap<ShardId, Vec<String>>,
data_shards: usize,
parity_shards: usize,
expected_encode_ts_ns: i64,
) -> io::Result<Vec<u8>> {
let total_shards = data_shards + parity_shards;
let rs = ReedSolomon::new(data_shards, parity_shards).map_err(|e| {
@ -614,6 +640,7 @@ async fn recover_one_remote_ec_shard_interval(
sid,
shard_offset,
size,
expected_encode_ts_ns,
)
.await;
(sid, res)

31
seaweed-volume/src/storage/disk_location.rs
View File

@ -344,6 +344,37 @@ impl DiskLocation {
}
}
/// Full-teardown variant: everything remove_ec_volume_files clears, PLUS the
/// normal-volume <base>.vif on a SHARD-ONLY node. An interrupted shard copy
/// (which installs shards + .ecx before .vif) could otherwise mount a fresh
/// generation under the prior run's identity / ratio / dat_file_size carried by
/// a stale .vif. Gated on .idx absence so a source-volume holder keeps its live
/// .vif. Reconcile/load-fallback call remove_ec_volume_files directly and
/// intentionally preserve it, mirroring Go's removeEcVolumeFiles (reconcile) vs
/// removeStaleEcArtifacts (teardown) split.
pub(crate) fn remove_ec_volume_files_full_teardown(&self, collection: &str, vid: VolumeId) {
self.remove_ec_volume_files(collection, vid);
let base = volume_file_name(&self.directory, collection, vid);
let idx_base = volume_file_name(&self.idx_directory, collection, vid);
// try_exists, not exists(): a stat error on a PRESENT .idx must not read as
// "shard-only" and delete a live source volume's .vif. Only Ok(false) (genuine
// NotFound) clears it; on a stat error stay conservative and keep the sidecar.
if matches!(
std::path::Path::new(&format!("{}.idx", idx_base)).try_exists(),
Ok(false)
) {
let _ = fs::remove_file(format!("{}.vif", idx_base));
if self.idx_directory != self.directory
&& matches!(
std::path::Path::new(&format!("{}.idx", base)).try_exists(),
Ok(false)
)
{
let _ = fs::remove_file(format!("{}.vif", base));
}
}
}
/// Find a volume by ID.
pub fn find_volume(&self, vid: VolumeId) -> Option<&Volume> {
self.volumes.get(&vid)

38
seaweed-volume/src/storage/erasure_coding/ec_volume.rs
View File

@ -57,6 +57,10 @@ pub struct EcVolume {
pub shard_locations_refresh_time: std::sync::Mutex<Option<std::time::Instant>>,
/// EC volume expiration time (unix epoch seconds), set during EC encode from TTL.
pub expire_at_sec: u64,
/// Encode-run identity (unix nanos) loaded from the .vif EcShardConfig. A read
/// served from a shard of a different encode run is rejected (server- and
/// client-side); 0 for a pre-feature volume, which is treated leniently.
pub encode_ts_ns: i64,
}
/// Locate the `.vif` for a (collection, vid) by preferring the data dir
@ -137,7 +141,7 @@ impl EcVolume {
// for shard-size math and by the Store-level prune in
// `store_ec_reconcile.rs` to verify a sibling-disk .dat is
// plausibly the encoding source (#9478).
let (expire_at_sec, vif_version, vif_dat_file_size) = {
let (expire_at_sec, vif_version, vif_dat_file_size, encode_ts_ns) = {
let vif_path = locate_vif_path(dir, dir_idx, collection, volume_id);
if let Ok(vif_content) = std::fs::read_to_string(&vif_path) {
if let Ok(vif_info) =
@ -148,12 +152,21 @@ impl EcVolume {
} else {
Version::current()
};
(vif_info.expire_at_sec, ver, vif_info.dat_file_size)
let cfg_encode_ts_ns = vif_info
.ec_shard_config
.as_ref()
.map_or(0, |c| c.encode_ts_ns);
(
vif_info.expire_at_sec,
ver,
vif_info.dat_file_size,
cfg_encode_ts_ns,
)
} else {
(0, Version::current(), 0)
(0, Version::current(), 0, 0)
}
} else {
(0, Version::current(), 0)
(0, Version::current(), 0, 0)
}
};
@ -177,6 +190,7 @@ impl EcVolume {
shard_locations: std::sync::RwLock::new(HashMap::new()),
shard_locations_refresh_time: std::sync::Mutex::new(None),
expire_at_sec,
encode_ts_ns,
};
// Open .ecx file (sorted index) in read/write mode for in-place deletion marking.
@ -605,7 +619,19 @@ impl EcVolume {
})?;
let mut buf = vec![0u8; interval.size as usize];
shard.read_at(&mut buf, shard_offset as u64)?;
let n = shard.read_at(&mut buf, shard_offset as u64)?;
if n != buf.len() {
return Err(io::Error::new(
io::ErrorKind::UnexpectedEof,
format!(
"short read on ec shard {}: read {} of {} bytes for needle {}",
shard_id,
n,
buf.len(),
needle_id
),
));
}
bytes.extend_from_slice(&buf);
}
@ -1243,6 +1269,7 @@ mod tests {
ec_shard_config: Some(crate::storage::volume::VifEcShardConfig {
data_shards: 6,
parity_shards: 3,
..Default::default()
}),
..Default::default()
};
@ -1268,6 +1295,7 @@ mod tests {
ec_shard_config: Some(crate::storage::volume::VifEcShardConfig {
data_shards: 10,
parity_shards: 10,
..Default::default()
}),
..Default::default()
};

1
seaweed-volume/src/storage/store_ec_mirror.rs
View File

@ -327,6 +327,7 @@ mod tests {
ec_shard_config: Some(VifEcShardConfig {
data_shards,
parity_shards,
..Default::default()
}),
..Default::default()
};

2
seaweed-volume/src/storage/store_ec_reconcile.rs
View File

@ -455,6 +455,7 @@ mod tests {
ec_shard_config: Some(VifEcShardConfig {
data_shards,
parity_shards,
..Default::default()
}),
..Default::default()
};
@ -1165,6 +1166,7 @@ mod tests {
ec_shard_config: Some(VifEcShardConfig {
data_shards: 10,
parity_shards: 4,
..Default::default()
}),
..Default::default()
};

6
seaweed-volume/src/storage/volume.rs
View File

@ -189,6 +189,10 @@ pub struct VifEcShardConfig {
pub data_shards: u32,
#[serde(default, rename = "parityShards")]
pub parity_shards: u32,
/// Encode time (unix nanos). Shards and .ecx of one encode run share it,
/// so a read served from a different run's shard is rejected.
#[serde(default, rename = "encodeTsNs", with = "string_or_i64")]
pub encode_ts_ns: i64,
}
/// Serde-compatible representation of OldVersionVolumeInfo for legacy .vif JSON deserialization.
@ -279,6 +283,7 @@ impl VifVolumeInfo {
ec_shard_config: pb.ec_shard_config.as_ref().map(|c| VifEcShardConfig {
data_shards: c.data_shards,
parity_shards: c.parity_shards,
encode_ts_ns: c.encode_ts_ns,
}),
}
}
@ -309,6 +314,7 @@ impl VifVolumeInfo {
crate::pb::volume_server_pb::EcShardConfig {
data_shards: c.data_shards,
parity_shards: c.parity_shards,
encode_ts_ns: c.encode_ts_ns,
}
}),
}

6
test/plugin_workers/fake_volume_server.go
View File

@ -309,9 +309,11 @@ func (v *VolumeServer) VolumeEcShardsUnmount(ctx context.Context, req *volume_se
// VolumeEcShardsDelete is a no-op stub paired with VolumeEcShardsUnmount
// above; the fake server doesn't persist shard files beyond what
// ReceiveFile wrote, so there's nothing to remove.
// ReceiveFile wrote, so there's nothing to remove. It still echoes the
// full_teardown acknowledgement so the worker doesn't treat the fake as a
// pre-upgrade server that silently skipped the teardown.
func (v *VolumeServer) VolumeEcShardsDelete(ctx context.Context, req *volume_server_pb.VolumeEcShardsDeleteRequest) (*volume_server_pb.VolumeEcShardsDeleteResponse, error) {
return &volume_server_pb.VolumeEcShardsDeleteResponse{}, nil
return &volume_server_pb.VolumeEcShardsDeleteResponse{FullTeardownDone: req.FullTeardown}, nil
}
func (v *VolumeServer) VolumeEcShardsInfo(ctx context.Context, req *volume_server_pb.VolumeEcShardsInfoRequest) (*volume_server_pb.VolumeEcShardsInfoResponse, error) {

6
weed/pb/volume_server.proto
View File

@ -452,8 +452,10 @@ message VolumeEcShardsDeleteRequest {
uint32 volume_id = 1;
string collection = 2;
repeated uint32 shard_ids = 3;
bool full_teardown = 4; // pre-encode cleanup: wipe every EC artifact + generation for this volume, not just shard_ids
}
message VolumeEcShardsDeleteResponse {
bool full_teardown_done = 1; // set by a new server that performed full_teardown; absent from an old server lets the caller detect the silent no-op
}
message VolumeEcShardsMountRequest {
@ -478,10 +480,13 @@ message VolumeEcShardReadRequest {
int64 offset = 3;
int64 size = 4;
uint64 file_key = 5;
reserved 6;
int64 encode_ts_ns = 7; // caller's expected encode time; the server rejects a shard from a different encode run
}
message VolumeEcShardReadResponse {
bytes data = 1;
bool is_deleted = 2;
int64 encode_ts_ns = 3; // identity of the shard actually served; client rejects a mismatch (0 = pre-upgrade server)
}
message VolumeEcBlobDeleteRequest {
@ -580,6 +585,7 @@ message VolumeInfo {
message EcShardConfig {
uint32 data_shards = 1; // Number of data shards (e.g., 10)
uint32 parity_shards = 2; // Number of parity shards (e.g., 4)
int64 encode_ts_ns = 3; // encode time (unix nanos); a read served from a shard of a different encode run is rejected
}
// EcBitrotProtection is the entire content of a bitrot checksum sidecar

70
weed/pb/volume_server_pb/volume_server.pb.go
View File

@ -3605,6 +3605,7 @@ type VolumeEcShardsDeleteRequest struct {
VolumeId uint32 `protobuf:"varint,1,opt,name=volume_id,json=volumeId,proto3" json:"volume_id,omitempty"`
Collection string `protobuf:"bytes,2,opt,name=collection,proto3" json:"collection,omitempty"`
ShardIds []uint32 `protobuf:"varint,3,rep,packed,name=shard_ids,json=shardIds,proto3" json:"shard_ids,omitempty"`
FullTeardown bool `protobuf:"varint,4,opt,name=full_teardown,json=fullTeardown,proto3" json:"full_teardown,omitempty"` // pre-encode cleanup: wipe every EC artifact + generation for this volume, not just shard_ids
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
@ -3660,10 +3661,18 @@ func (x *VolumeEcShardsDeleteRequest) GetShardIds() []uint32 {
return nil
}
func (x *VolumeEcShardsDeleteRequest) GetFullTeardown() bool {
if x != nil {
return x.FullTeardown
}
return false
}
type VolumeEcShardsDeleteResponse struct {
state protoimpl.MessageState `protogen:"open.v1"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
state protoimpl.MessageState `protogen:"open.v1"`
FullTeardownDone bool `protobuf:"varint,1,opt,name=full_teardown_done,json=fullTeardownDone,proto3" json:"full_teardown_done,omitempty"` // set by a new server that performed full_teardown; absent from an old server lets the caller detect the silent no-op
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *VolumeEcShardsDeleteResponse) Reset() {
@ -3696,6 +3705,13 @@ func (*VolumeEcShardsDeleteResponse) Descriptor() ([]byte, []int) {
return file_volume_server_proto_rawDescGZIP(), []int{65}
}
func (x *VolumeEcShardsDeleteResponse) GetFullTeardownDone() bool {
if x != nil {
return x.FullTeardownDone
}
return false
}
type VolumeEcShardsMountRequest struct {
state protoimpl.MessageState `protogen:"open.v1"`
VolumeId uint32 `protobuf:"varint,1,opt,name=volume_id,json=volumeId,proto3" json:"volume_id,omitempty"`
@ -3895,6 +3911,7 @@ type VolumeEcShardReadRequest struct {
Offset int64 `protobuf:"varint,3,opt,name=offset,proto3" json:"offset,omitempty"`
Size int64 `protobuf:"varint,4,opt,name=size,proto3" json:"size,omitempty"`
FileKey uint64 `protobuf:"varint,5,opt,name=file_key,json=fileKey,proto3" json:"file_key,omitempty"`
EncodeTsNs int64 `protobuf:"varint,7,opt,name=encode_ts_ns,json=encodeTsNs,proto3" json:"encode_ts_ns,omitempty"` // caller's expected encode time; the server rejects a shard from a different encode run
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
@ -3964,10 +3981,18 @@ func (x *VolumeEcShardReadRequest) GetFileKey() uint64 {
return 0
}
func (x *VolumeEcShardReadRequest) GetEncodeTsNs() int64 {
if x != nil {
return x.EncodeTsNs
}
return 0
}
type VolumeEcShardReadResponse struct {
state protoimpl.MessageState `protogen:"open.v1"`
Data []byte `protobuf:"bytes,1,opt,name=data,proto3" json:"data,omitempty"`
IsDeleted bool `protobuf:"varint,2,opt,name=is_deleted,json=isDeleted,proto3" json:"is_deleted,omitempty"`
EncodeTsNs int64 `protobuf:"varint,3,opt,name=encode_ts_ns,json=encodeTsNs,proto3" json:"encode_ts_ns,omitempty"` // identity of the shard actually served; client rejects a mismatch (0 = pre-upgrade server)
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
@ -4016,6 +4041,13 @@ func (x *VolumeEcShardReadResponse) GetIsDeleted() bool {
return false
}
func (x *VolumeEcShardReadResponse) GetEncodeTsNs() int64 {
if x != nil {
return x.EncodeTsNs
}
return 0
}
type VolumeEcBlobDeleteRequest struct {
state protoimpl.MessageState `protogen:"open.v1"`
VolumeId uint32 `protobuf:"varint,1,opt,name=volume_id,json=volumeId,proto3" json:"volume_id,omitempty"`
@ -4946,6 +4978,7 @@ type EcShardConfig struct {
state protoimpl.MessageState `protogen:"open.v1"`
DataShards uint32 `protobuf:"varint,1,opt,name=data_shards,json=dataShards,proto3" json:"data_shards,omitempty"` // Number of data shards (e.g., 10)
ParityShards uint32 `protobuf:"varint,2,opt,name=parity_shards,json=parityShards,proto3" json:"parity_shards,omitempty"` // Number of parity shards (e.g., 4)
EncodeTsNs int64 `protobuf:"varint,3,opt,name=encode_ts_ns,json=encodeTsNs,proto3" json:"encode_ts_ns,omitempty"` // encode time (unix nanos); a read served from a shard of a different encode run is rejected
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
@ -4994,6 +5027,13 @@ func (x *EcShardConfig) GetParityShards() uint32 {
return 0
}
func (x *EcShardConfig) GetEncodeTsNs() int64 {
if x != nil {
return x.EncodeTsNs
}
return 0
}
// EcBitrotProtection is the entire content of a bitrot checksum sidecar
// (<base>.ecsum for the legacy generation, <base>.ecsum.v<N> for vacuum
// generation N). On disk it is wrapped in a fixed header carrying a CRC32C
@ -7228,14 +7268,16 @@ const file_volume_server_proto_rawDesc = "" +
"\rcopy_vif_file\x18\a \x01(\bR\vcopyVifFile\x12\x17\n" +
"\adisk_id\x18\b \x01(\rR\x06diskId\x12&\n" +
"\x0fcopy_ecsum_file\x18\t \x01(\bR\rcopyEcsumFile\"\x1c\n" +
"\x1aVolumeEcShardsCopyResponse\"w\n" +
"\x1aVolumeEcShardsCopyResponse\"\x9c\x01\n" +
"\x1bVolumeEcShardsDeleteRequest\x12\x1b\n" +
"\tvolume_id\x18\x01 \x01(\rR\bvolumeId\x12\x1e\n" +
"\n" +
"collection\x18\x02 \x01(\tR\n" +
"collection\x12\x1b\n" +
"\tshard_ids\x18\x03 \x03(\rR\bshardIds\"\x1e\n" +
"\x1cVolumeEcShardsDeleteResponse\"\xa0\x01\n" +
"\tshard_ids\x18\x03 \x03(\rR\bshardIds\x12#\n" +
"\rfull_teardown\x18\x04 \x01(\bR\ffullTeardown\"L\n" +
"\x1cVolumeEcShardsDeleteResponse\x12,\n" +
"\x12full_teardown_done\x18\x01 \x01(\bR\x10fullTeardownDone\"\xa0\x01\n" +
"\x1aVolumeEcShardsMountRequest\x12\x1b\n" +
"\tvolume_id\x18\x01 \x01(\rR\bvolumeId\x12\x1e\n" +
"\n" +
@ -7247,17 +7289,21 @@ const file_volume_server_proto_rawDesc = "" +
"\x1cVolumeEcShardsUnmountRequest\x12\x1b\n" +
"\tvolume_id\x18\x01 \x01(\rR\bvolumeId\x12\x1b\n" +
"\tshard_ids\x18\x03 \x03(\rR\bshardIds\"\x1f\n" +
"\x1dVolumeEcShardsUnmountResponse\"\x99\x01\n" +
"\x1dVolumeEcShardsUnmountResponse\"\xc1\x01\n" +
"\x18VolumeEcShardReadRequest\x12\x1b\n" +
"\tvolume_id\x18\x01 \x01(\rR\bvolumeId\x12\x19\n" +
"\bshard_id\x18\x02 \x01(\rR\ashardId\x12\x16\n" +
"\x06offset\x18\x03 \x01(\x03R\x06offset\x12\x12\n" +
"\x04size\x18\x04 \x01(\x03R\x04size\x12\x19\n" +
"\bfile_key\x18\x05 \x01(\x04R\afileKey\"N\n" +
"\bfile_key\x18\x05 \x01(\x04R\afileKey\x12 \n" +
"\fencode_ts_ns\x18\a \x01(\x03R\n" +
"encodeTsNsJ\x04\b\x06\x10\a\"p\n" +
"\x19VolumeEcShardReadResponse\x12\x12\n" +
"\x04data\x18\x01 \x01(\fR\x04data\x12\x1d\n" +
"\n" +
"is_deleted\x18\x02 \x01(\bR\tisDeleted\"\x8d\x01\n" +
"is_deleted\x18\x02 \x01(\bR\tisDeleted\x12 \n" +
"\fencode_ts_ns\x18\x03 \x01(\x03R\n" +
"encodeTsNs\"\x8d\x01\n" +
"\x19VolumeEcBlobDeleteRequest\x12\x1b\n" +
"\tvolume_id\x18\x01 \x01(\rR\bvolumeId\x12\x1e\n" +
"\n" +
@ -7346,11 +7392,13 @@ const file_volume_server_proto_rawDesc = "" +
"\rdat_file_size\x18\x05 \x01(\x03R\vdatFileSize\x12\"\n" +
"\rexpire_at_sec\x18\x06 \x01(\x04R\vexpireAtSec\x12\x1b\n" +
"\tread_only\x18\a \x01(\bR\breadOnly\x12G\n" +
"\x0fec_shard_config\x18\b \x01(\v2\x1f.volume_server_pb.EcShardConfigR\recShardConfig\"U\n" +
"\x0fec_shard_config\x18\b \x01(\v2\x1f.volume_server_pb.EcShardConfigR\recShardConfig\"w\n" +
"\rEcShardConfig\x12\x1f\n" +
"\vdata_shards\x18\x01 \x01(\rR\n" +
"dataShards\x12#\n" +
"\rparity_shards\x18\x02 \x01(\rR\fparityShards\"\xbc\x02\n" +
"\rparity_shards\x18\x02 \x01(\rR\fparityShards\x12 \n" +
"\fencode_ts_ns\x18\x03 \x01(\x03R\n" +
"encodeTsNs\"\xbc\x02\n" +
"\x12EcBitrotProtection\x12A\n" +
"\talgorithm\x18\x01 \x01(\x0e2#.volume_server_pb.ChecksumAlgorithmR\talgorithm\x12\x1d\n" +
"\n" +

179
weed/server/volume_grpc_erasure_coding.go
View File

@ -88,6 +88,20 @@ func (vs *VolumeServer) VolumeEcShardsGenerate(ctx context.Context, req *volume_
os.Remove(erasure_coding.BitrotSidecarPath(baseFileName, 0))
}()
// Wipe any EC artifacts from a prior encode so a retry never mixes two runs.
// Evict the in-memory EcVolume first so the unlink frees the inodes instead
// of leaving open fds serving the old bytes. Sweep every disk: stale shards
// can sit on a sibling disk and would otherwise survive and be mounted
// against the new index at reconcile. Scans the cap for custom ratios.
vs.store.UnloadEcVolume(needle.VolumeId(req.VolumeId))
for _, loc := range vs.store.Locations {
dataBase := storage.VolumeFileName(loc.Directory, req.Collection, int(req.VolumeId))
idxBase := storage.VolumeFileName(loc.IdxDirectory, req.Collection, int(req.VolumeId))
if err := removeStaleEcArtifacts(dataBase, idxBase, erasure_coding.MaxShardCount); err != nil {
return nil, fmt.Errorf("wipe stale EC artifacts for volume %d on %s: %w", req.VolumeId, loc.Directory, err)
}
}
// IMPORTANT: Generate .ecx BEFORE EC shards to prevent a race condition.
// If .ecx were generated after EC shards, any write (e.g. from WriteNeedleBlob
// during replica sync) between the two steps would add entries to .idx that
@ -141,10 +155,12 @@ func (vs *VolumeServer) VolumeEcShardsGenerate(ctx context.Context, req *volume_
ecCtx.DataShards, ecCtx.ParityShards, ecCtx.Total(), erasure_coding.MaxShardCount)
}
// Save EC configuration to VolumeInfo
// EncodeTsNs stamps this run's identity into the .vif (copied with the
// shards), so a read served from a different run's shard is rejected.
volumeInfo.EcShardConfig = &volume_server_pb.EcShardConfig{
DataShards: uint32(ecCtx.DataShards),
ParityShards: uint32(ecCtx.ParityShards),
EncodeTsNs: time.Now().UnixNano(),
}
glog.V(1).Infof("Saving EC config to .vif for volume %d: %d+%d (total: %d)",
req.VolumeId, ecCtx.DataShards, ecCtx.ParityShards, ecCtx.Total())
@ -401,6 +417,23 @@ func (vs *VolumeServer) VolumeEcShardsDelete(ctx context.Context, req *volume_se
bName := erasure_coding.EcShardBaseFileName(req.Collection, int(req.VolumeId))
if req.FullTeardown {
// Pre-encode cleanup: evict the volume and wipe every EC artifact for it on
// every disk (the same teardown the generator does locally), not just the
// listed shards, so a remote node retains no stale generation that a fresh
// gen-0 copy would later collide with.
glog.V(0).Infof("ec volume %s full teardown", bName)
vs.store.UnloadEcVolume(needle.VolumeId(req.VolumeId))
for _, location := range vs.store.Locations {
dataBase := storage.VolumeFileName(location.Directory, req.Collection, int(req.VolumeId))
idxBase := storage.VolumeFileName(location.IdxDirectory, req.Collection, int(req.VolumeId))
if err := removeStaleEcArtifacts(dataBase, idxBase, erasure_coding.MaxShardCount); err != nil {
return nil, fmt.Errorf("full teardown of ec volume %d on %s: %w", req.VolumeId, location.Directory, err)
}
}
return &volume_server_pb.VolumeEcShardsDeleteResponse{FullTeardownDone: true}, nil
}
glog.V(0).Infof("ec volume %s shard delete %v", bName, req.ShardIds)
for diskId, location := range vs.store.Locations {
@ -420,16 +453,15 @@ func deleteEcShardIdsForEachLocation(bName string, location *storage.DiskLocatio
indexBaseFilename := path.Join(location.IdxDirectory, bName)
dataBaseFilename := path.Join(location.Directory, bName)
ecxExists := util.FileExists(path.Join(location.IdxDirectory, bName+".ecx"))
if !ecxExists && location.IdxDirectory != location.Directory {
ecxExists = util.FileExists(path.Join(location.Directory, bName+".ecx"))
}
if ecxExists {
for _, shardId := range shardIds {
shardFileName := dataBaseFilename + erasure_coding.ToExt(int(shardId))
if util.FileExists(shardFileName) {
found = true
os.Remove(shardFileName)
// Delete the requested shard files unconditionally. Gating on a local .ecx
// (still used for index-file routing below) would leak an orphan shard left
// by a failed copy that reconciliation later mounts under a foreign index.
for _, shardId := range shardIds {
shardFileName := dataBaseFilename + erasure_coding.ToExt(int(shardId))
if util.FileExists(shardFileName) {
found = true
if err := removeFileIfExists(shardFileName); err != nil {
return fmt.Errorf("remove ec shard %s: %w", shardFileName, err)
}
}
}
@ -453,26 +485,38 @@ func deleteEcShardIdsForEachLocation(bName string, location *storage.DiskLocatio
// leaks. The per-shard-id delete that ec.rebuild uses for
// copied-survivor cleanup leaves shards behind, so this guard does not
// fire there.
removeBitrotSidecars(dataBaseFilename)
if err := removeBitrotSidecars(dataBaseFilename); err != nil {
return err
}
if location.IdxDirectory != location.Directory {
removeBitrotSidecars(indexBaseFilename)
if err := removeBitrotSidecars(indexBaseFilename); err != nil {
return err
}
}
if hasEcxFile {
// Remove .ecx/.ecj from both idx and data directories
// since they may be in either location depending on when -dir.idx was configured
if err := os.Remove(indexBaseFilename + ".ecx"); err != nil && !os.IsNotExist(err) {
return err
// since they may be in either location depending on when -dir.idx was configured.
// A surviving stale .ecx is the orphan-index condition this path prevents,
// so surface a real removal failure instead of reporting cleanup as success.
for _, p := range []string{indexBaseFilename + ".ecx", indexBaseFilename + ".ecj"} {
if err := removeFileIfExists(p); err != nil {
return err
}
}
os.Remove(indexBaseFilename + ".ecj")
if location.IdxDirectory != location.Directory {
os.Remove(dataBaseFilename + ".ecx")
os.Remove(dataBaseFilename + ".ecj")
for _, p := range []string{dataBaseFilename + ".ecx", dataBaseFilename + ".ecj"} {
if err := removeFileIfExists(p); err != nil {
return err
}
}
}
if !hasIdxFile {
// .vif is used for ec volumes and normal volumes
os.Remove(dataBaseFilename + ".vif")
if err := removeFileIfExists(dataBaseFilename + ".vif"); err != nil {
return err
}
}
}
}
@ -480,15 +524,71 @@ func deleteEcShardIdsForEachLocation(bName string, location *storage.DiskLocatio
return nil
}
// removeBitrotSidecars removes the legacy <base>.ecsum and any versioned
// <base>.ecsum.v<N> sidecars. Best-effort; logs nothing on absence.
func removeBitrotSidecars(baseFilename string) {
os.Remove(baseFilename + erasure_coding.BitrotSidecarExt)
if matches, _ := filepath.Glob(baseFilename + erasure_coding.BitrotSidecarExt + ".v*"); matches != nil {
for _, m := range matches {
os.Remove(m)
// removeFileIfExists removes path, treating "already gone" as success and
// returning only a real failure (so a stale shard left behind is not silently
// reported as cleaned).
func removeFileIfExists(path string) error {
if err := os.Remove(path); err != nil && !os.IsNotExist(err) {
return err
}
return nil
}
// removeStaleEcArtifacts deletes the shard, index, journal, and bitrot sidecar
// files of a prior encode so a fresh encode never mixes runs. total is the
// shard-id range to scan (pass the cap for custom ratios). Returns the first
// real removal failure; does not touch the source .dat/.idx/.vif.
func removeStaleEcArtifacts(dataBaseFileName, indexBaseFileName string, total int) error {
var firstErr error
record := func(err error) {
if err != nil && firstErr == nil {
firstErr = err
}
}
for i := 0; i < total; i++ {
record(removeFileIfExists(dataBaseFileName + erasure_coding.ToExt(i)))
}
// .ecx/.ecj/.ecsum may sit in either dir depending on -dir.idx; clear both.
record(removeFileIfExists(indexBaseFileName + ".ecx"))
record(removeFileIfExists(indexBaseFileName + ".ecj"))
record(removeBitrotSidecars(indexBaseFileName))
if dataBaseFileName != indexBaseFileName {
record(removeFileIfExists(dataBaseFileName + ".ecx"))
record(removeFileIfExists(dataBaseFileName + ".ecj"))
record(removeBitrotSidecars(dataBaseFileName))
}
// Canonical <base>.vif. A shard copy installs shards + .ecx before .vif, so an
// interrupted copy can leave a stale .vif whose run identity / shard ratio /
// dat_file_size a fresh generation would inherit. Remove it only on a shard-only
// EC node: where a normal <base>.idx exists this is the source volume holder and
// the .vif belongs to that live volume — keep it. This mirrors the !hasIdxFile
// gate in the per-shard delete path.
if _, statErr := os.Stat(indexBaseFileName + ".idx"); os.IsNotExist(statErr) {
record(removeFileIfExists(indexBaseFileName + ".vif"))
if dataBaseFileName != indexBaseFileName {
if _, dStatErr := os.Stat(dataBaseFileName + ".idx"); os.IsNotExist(dStatErr) {
record(removeFileIfExists(dataBaseFileName + ".vif"))
}
}
}
return firstErr
}
// removeBitrotSidecars removes the legacy <base>.ecsum and any versioned
// <base>.ecsum.v<N> sidecars, returning the first real removal failure.
func removeBitrotSidecars(baseFilename string) error {
var firstErr error
if err := removeFileIfExists(baseFilename + erasure_coding.BitrotSidecarExt); err != nil {
firstErr = err
}
matches, _ := filepath.Glob(baseFilename + erasure_coding.BitrotSidecarExt + ".v*")
for _, m := range matches {
if err := removeFileIfExists(m); err != nil && firstErr == nil {
firstErr = err
}
}
return firstErr
}
func checkEcVolumeStatus(bName string, location *storage.DiskLocation) (hasEcxFile bool, hasIdxFile bool, existingShardCount int, err error) {
@ -581,22 +681,28 @@ func (vs *VolumeServer) VolumeEcShardsUnmount(ctx context.Context, req *volume_s
func (vs *VolumeServer) VolumeEcShardRead(req *volume_server_pb.VolumeEcShardReadRequest, stream volume_server_pb.VolumeServer_VolumeEcShardReadServer) error {
ecVolume, found := vs.store.FindEcVolume(needle.VolumeId(req.VolumeId))
if !found {
return fmt.Errorf("VolumeEcShardRead not found ec volume id %d", req.VolumeId)
}
// shard may live on a sibling disk of this server; walk all of them
// under ecVolumesLock.
_, ecShard, found := vs.store.FindEcShard(needle.VolumeId(req.VolumeId), erasure_coding.ShardId(req.ShardId))
// Resolve the shard together with the EcVolume on the disk that owns it,
// rather than a first-match volume on a sibling disk: on a multi-disk server
// those can belong to different encode generations, and the guard must
// validate the identity of the volume whose bytes we serve.
ecVolume, ecShard, found := vs.store.FindEcVolumeWithShard(needle.VolumeId(req.VolumeId), erasure_coding.ShardId(req.ShardId))
if !found {
return fmt.Errorf("not found ec shard %d.%d", req.VolumeId, req.ShardId)
}
// Reject a shard whose identity doesn't match the caller's index; the caller
// then recovers from parity. Lenient only when the caller has no identity
// (pre-upgrade reader): a known caller must not accept an unstamped holder,
// which would serve a stale pre-upgrade shard.
if req.EncodeTsNs != 0 && req.EncodeTsNs != ecVolume.EncodeTsNs {
return fmt.Errorf("ec shard %d.%d belongs to a different encode run", req.VolumeId, req.ShardId)
}
if req.FileKey != 0 {
_, size, _ := ecVolume.FindNeedleFromEcx(types.Uint64ToNeedleId(req.FileKey))
if size.IsDeleted() {
return stream.Send(&volume_server_pb.VolumeEcShardReadResponse{
IsDeleted: true,
IsDeleted: true,
EncodeTsNs: ecVolume.EncodeTsNs,
})
}
}
@ -624,7 +730,8 @@ func (vs *VolumeServer) VolumeEcShardRead(req *volume_server_pb.VolumeEcShardRea
bytesread = int(bytesToRead)
}
err = stream.Send(&volume_server_pb.VolumeEcShardReadResponse{
Data: buffer[:bytesread],
Data: buffer[:bytesread],
EncodeTsNs: ecVolume.EncodeTsNs,
})
if err != nil {
// println("sending", bytesread, "bytes err", err.Error())

89
weed/server/volume_grpc_erasure_coding_test.go
View File

@ -64,6 +64,95 @@ func TestCheckEcVolumeStatusCountOnlyDataShards(t *testing.T) {
}
}
// TestRemoveStaleEcArtifacts: a fresh encode deletes every prior EC artifact
// (incl. shard ids beyond the default ratio and versioned bitrot sidecars)
// while leaving the source .dat/.idx/.vif untouched.
func TestRemoveStaleEcArtifacts(t *testing.T) {
tempDir := t.TempDir()
dataDir := filepath.Join(tempDir, "data")
idxDir := filepath.Join(tempDir, "idx")
for _, d := range []string{dataDir, idxDir} {
if err := os.MkdirAll(d, 0o755); err != nil {
t.Fatalf("mkdir %s: %v", d, err)
}
}
const baseName = "7"
dataBase := filepath.Join(dataDir, baseName)
idxBase := filepath.Join(idxDir, baseName)
// EC artifacts that must be removed, including a shard id past the default
// 10+4 ratio (proves the MaxShardCount scan) and a versioned sidecar.
var ecFiles []string
for _, id := range []int{0, 9, 13, 20, erasure_coding.MaxShardCount - 1} {
ecFiles = append(ecFiles, dataBase+erasure_coding.ToExt(id))
}
ecFiles = append(ecFiles,
dataBase+".ecx", dataBase+".ecj",
idxBase+".ecx", idxBase+".ecj",
dataBase+erasure_coding.BitrotSidecarExt, // .ecsum (generation 0)
dataBase+erasure_coding.BitrotSidecarExt+".v2", // versioned sidecar
)
// Source files that must survive — the authoritative input for the encode.
srcFiles := []string{dataBase + ".dat", idxBase + ".idx", dataBase + ".vif"}
for _, f := range append(append([]string{}, ecFiles...), srcFiles...) {
if err := os.WriteFile(f, []byte("x"), 0o644); err != nil {
t.Fatalf("create %s: %v", f, err)
}
}
removeStaleEcArtifacts(dataBase, idxBase, erasure_coding.MaxShardCount)
for _, f := range ecFiles {
if util.FileExists(f) {
t.Errorf("expected EC artifact removed: %s", f)
}
}
for _, f := range srcFiles {
if !util.FileExists(f) {
t.Errorf("expected source file preserved: %s", f)
}
}
}
// TestDeleteEcShardsWithoutLocalEcx: the delete handler removes the requested
// shard files even on a disk with no local .ecx, so a failed-copy orphan stays
// cleanable rather than being mounted later under a foreign index.
func TestDeleteEcShardsWithoutLocalEcx(t *testing.T) {
tempDir := t.TempDir()
dataDir := filepath.Join(tempDir, "data")
idxDir := filepath.Join(tempDir, "idx")
for _, d := range []string{dataDir, idxDir} {
if err := os.MkdirAll(d, 0o755); err != nil {
t.Fatalf("mkdir %s: %v", d, err)
}
}
const baseName = "7"
// Orphan shard files with NO .ecx/.idx anywhere — a failed-copy leftover.
orphans := []string{
filepath.Join(dataDir, baseName+".ec03"),
filepath.Join(dataDir, baseName+".ec11"),
}
for _, f := range orphans {
if err := os.WriteFile(f, []byte("x"), 0o644); err != nil {
t.Fatalf("create %s: %v", f, err)
}
}
location := &storage.DiskLocation{Directory: dataDir, IdxDirectory: idxDir}
if err := deleteEcShardIdsForEachLocation(baseName, location, []uint32{3, 11}); err != nil {
t.Fatalf("deleteEcShardIdsForEachLocation: %v", err)
}
for _, f := range orphans {
if util.FileExists(f) {
t.Errorf("expected orphan shard removed without a local .ecx: %s", f)
}
}
}
// TestVolumeEcShardsInfo_AggregatesAcrossDisks pins the multi-disk path:
// when a volume server mounts EC shards for the same volume on more than
// one disk (each disk holds its own EcVolume entry — Store.FindEcVolume

2
weed/shell/command_ec_balance.go
View File

@ -75,5 +75,5 @@ func (c *commandEcBalance) Do(args []string, commandEnv *CommandEnv, writer io.W
diskType := types.ToDiskType(*diskTypeStr)
return EcBalance(commandEnv, collections, *dc, rp, diskType, *maxParallelization, *applyBalancing)
return EcBalance(commandEnv, collections, *dc, rp, diskType, *maxParallelization, *applyBalancing, nil)
}

78
weed/shell/command_ec_common.go
View File

@ -2,6 +2,7 @@ package shell
import (
"context"
"errors"
"fmt"
"regexp"
"slices"
@ -536,6 +537,58 @@ func sourceServerDeleteEcShards(grpcDialOption grpc.DialOption, collection strin
}
// errFullTeardownNotAcked marks a reachable server that completed the delete RPC
// but did not report full_teardown_done (a pre-upgrade volume server). The orphan
// sweep must treat this as fatal: the node may still hold an orphan that a later
// copy would re-stamp into the new generation.
var errFullTeardownNotAcked = errors.New("delete did not perform full teardown (pre-upgrade volume server?); a stale EC generation may remain")
// pingVolumeServer probes node liveness with an empty-target Ping, which is never
// maintenance-gated, and returns the raw Ping error (nil on success). It lets the
// orphan sweep disambiguate a delete codes.Unavailable: a Rust volume server in
// maintenance mode fails the maintenance-gated delete with Unavailable yet answers
// Ping, whereas a genuinely-down node fails Ping with a transport Unavailable too.
// A Go server returns Unknown for maintenance, which isNodeUnreachable already
// treats as fatal. The caller classifies the result with classifyNodeLiveness:
// only a Ping that itself transport-failed (codes.Unavailable) confirms the node
// is down; a nil error (reachable) or any other Ping error (inconclusive — e.g. a
// pre-Ping server returning Unimplemented, which means the node is up) is fatal.
func pingVolumeServer(grpcDialOption grpc.DialOption, location pb.ServerAddress) error {
return operation.WithVolumeServerClient(false, location, grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
_, pingErr := client.Ping(context.Background(), &volume_server_pb.PingRequest{})
return pingErr
})
}
// unmountAndDeleteEcShardsQuiet unmounts then deletes shards on one server in a
// single connection, without the per-call logging the interactive helpers emit.
// Used by the orphan sweep, which fans out to every node x volume and would
// otherwise flood the shell with no-op lines.
func unmountAndDeleteEcShardsQuiet(grpcDialOption grpc.DialOption, collection string, volumeId needle.VolumeId, location pb.ServerAddress, shardIds []erasure_coding.ShardId) error {
ids := erasure_coding.ShardIdsToUint32(shardIds)
return operation.WithVolumeServerClient(false, location, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
if _, err := volumeServerClient.VolumeEcShardsUnmount(context.Background(), &volume_server_pb.VolumeEcShardsUnmountRequest{
VolumeId: uint32(volumeId),
ShardIds: ids,
}); err != nil {
return fmt.Errorf("unmount: %w", err)
}
resp, err := volumeServerClient.VolumeEcShardsDelete(context.Background(), &volume_server_pb.VolumeEcShardsDeleteRequest{
VolumeId: uint32(volumeId),
Collection: collection,
ShardIds: ids,
FullTeardown: true,
})
if err != nil {
return fmt.Errorf("delete: %w", err)
}
if !resp.GetFullTeardownDone() {
return fmt.Errorf("delete on %s: %w", location, errFullTeardownNotAcked)
}
return nil
})
}
func unmountEcShards(grpcDialOption grpc.DialOption, volumeId needle.VolumeId, sourceLocation pb.ServerAddress, toBeUnmountedShardIds []erasure_coding.ShardId) error {
fmt.Printf("unmount %d.%v from %s\n", volumeId, toBeUnmountedShardIds, sourceLocation)
@ -741,12 +794,35 @@ type ecBalancer struct {
diskType types.DiskType
}
func EcBalance(commandEnv *CommandEnv, collections []string, dc string, ecReplicaPlacement *super_block.ReplicaPlacement, diskType types.DiskType, maxParallelization int, applyBalancing bool) (err error) {
// excludeNodes is a set of server addresses kept out of the balance as copy/move
// targets and sources. ec.encode passes the nodes its orphan sweep could not
// reach: such a node may still hold a stale-generation shard orphan, and pairing
// it with a new-generation shard from a balance copy would mix generations on one
// node. The standalone ec.balance command passes nil.
func EcBalance(commandEnv *CommandEnv, collections []string, dc string, ecReplicaPlacement *super_block.ReplicaPlacement, diskType types.DiskType, maxParallelization int, applyBalancing bool, excludeNodes map[pb.ServerAddress]struct{}) (err error) {
// collect all ec nodes
allEcNodes, totalFreeEcSlots, err := collectEcNodesForDC(commandEnv, dc, diskType)
if err != nil {
return err
}
// Drop excluded nodes (and the slots they contribute) before planning so they
// can be neither a target nor a source for any move this balance plans.
if len(excludeNodes) > 0 {
kept := allEcNodes[:0]
var excludedFreeSlots int
for _, en := range allEcNodes {
if _, skip := excludeNodes[pb.NewServerAddressFromDataNode(en.info)]; skip {
excludedFreeSlots += en.freeEcSlot
glog.V(0).Infof("EC balance excluding node %s: skipped as unreachable by the encode orphan sweep", en.info.Id)
continue
}
kept = append(kept, en)
}
allEcNodes = kept
totalFreeEcSlots -= excludedFreeSlots
}
if totalFreeEcSlots < 1 {
return fmt.Errorf("no free ec shard slots. only %d left", totalFreeEcSlots)
}

269
weed/shell/command_ec_encode.go
View File

@ -2,11 +2,14 @@ package shell
import (
"context"
"errors"
"flag"
"fmt"
"io"
"regexp"
"sort"
"strconv"
"sync"
"time"
"github.com/seaweedfs/seaweedfs/weed/storage/types"
@ -16,6 +19,8 @@ import (
"github.com/seaweedfs/seaweedfs/weed/wdclient"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
"github.com/seaweedfs/seaweedfs/weed/operation"
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
@ -193,11 +198,13 @@ func (c *commandEcEncode) Do(args []string, commandEnv *CommandEnv, writer io.Wr
}
// encode all requested volumes...
if err = doEcEncode(commandEnv, writer, volumeIdToCollection, volumeIds, *maxParallelization, topologyInfo); err != nil {
skippedNodes, err := doEcEncode(commandEnv, writer, volumeIdToCollection, volumeIds, *maxParallelization, topologyInfo)
if err != nil {
return fmt.Errorf("ec encode for volumes %v: %w", volumeIds, err)
}
// ...re-balance ec shards...
if err := EcBalance(commandEnv, balanceCollections, "", rp, diskType, *maxParallelization, *applyBalancing); err != nil {
// ...re-balance ec shards, excluding nodes the orphan sweep could not reach so
// a recovered node's stale orphan is never paired with a new-generation shard...
if err := EcBalance(commandEnv, balanceCollections, "", rp, diskType, *maxParallelization, *applyBalancing, skippedNodes); err != nil {
return fmt.Errorf("re-balance ec shards for collection(s) %v: %w", balanceCollections, err)
}
// A partial encode followed by source deletion is unrecoverable.
@ -227,13 +234,23 @@ func volumeLocations(commandEnv *CommandEnv, volumeIds []needle.VolumeId) (map[n
return res, nil
}
func doEcEncode(commandEnv *CommandEnv, writer io.Writer, volumeIdToCollection map[needle.VolumeId]string, volumeIds []needle.VolumeId, maxParallelization int, topologyInfo *master_pb.TopologyInfo) error {
func doEcEncode(commandEnv *CommandEnv, writer io.Writer, volumeIdToCollection map[needle.VolumeId]string, volumeIds []needle.VolumeId, maxParallelization int, topologyInfo *master_pb.TopologyInfo) (skippedNodes map[pb.ServerAddress]struct{}, err error) {
if !commandEnv.isLocked() {
return fmt.Errorf("lock is lost")
return nil, fmt.Errorf("lock is lost")
}
locations, err := volumeLocations(commandEnv, volumeIds)
if err != nil {
return fmt.Errorf("failed to get volume locations for EC encoding: %w", err)
return nil, fmt.Errorf("failed to get volume locations for EC encoding: %w", err)
}
// Clear EC shards left by a previous failed/partial encode so a retry
// starts clean and never mixes two encode runs. A node skipped here as
// unreachable is excluded from the later balance: it may still hold a stale
// orphan that, paired with a new-generation shard from a balance copy, would
// mix generations on that node.
skippedNodes, err = clearPreexistingEcShards(commandEnv, topologyInfo, volumeIds, volumeIdToCollection, maxParallelization)
if err != nil {
return nil, fmt.Errorf("clear pre-existing ec shards before encoding: %w", err)
}
// Build a map of (volumeId, serverAddress) -> freeVolumeCount.
@ -267,7 +284,7 @@ func doEcEncode(commandEnv *CommandEnv, writer io.Writer, volumeIdToCollection m
}
}
if len(filteredLocs) == 0 {
return fmt.Errorf("no healthy replicas (FreeVolumeCount >= 2) found for volume %d to use as source for EC encoding", vid)
return nil, fmt.Errorf("no healthy replicas (FreeVolumeCount >= 2) found for volume %d to use as source for EC encoding", vid)
}
filteredLocations[vid] = filteredLocs
}
@ -285,7 +302,7 @@ func doEcEncode(commandEnv *CommandEnv, writer io.Writer, volumeIdToCollection m
}
}
if err := ewg.Wait(); err != nil {
return err
return nil, err
}
// Sync replicas and select the best one for each volume (with highest file count)
@ -298,11 +315,36 @@ func doEcEncode(commandEnv *CommandEnv, writer io.Writer, volumeIdToCollection m
// Sync missing entries between replicas, then select the best one
bestLoc, selectErr := volume_replica.SyncAndSelectBestReplica(commandEnv.option.GrpcDialOption, vid, collection, filteredLocations[vid], "", writer)
if selectErr != nil {
return fmt.Errorf("failed to sync and select replica for volume %d: %v", vid, selectErr)
return nil, fmt.Errorf("failed to sync and select replica for volume %d: %v", vid, selectErr)
}
bestReplicas[vid] = bestLoc
}
// Re-attempt the orphan sweep on the nodes skipped as unreachable, now that
// any node that recovered during readonly-marking and replica sync answers
// again. A node whose teardown now succeeds is clean (and the generation host
// re-wipes its own disks regardless), so it leaves the skipped set and can be
// a balance source/target — otherwise its shards would never distribute off
// it. A node that is still down stays skipped and excluded, preserving the
// leniency for a genuinely-down node; such a node also cannot be the
// generation host below, since VolumeEcShardsGenerate would fail to read .dat.
if err := resweepSkippedNodes(commandEnv, skippedNodes, volumeIds, volumeIdToCollection, maxParallelization); err != nil {
return nil, err
}
// A selected generation host still in skippedNodes after the re-sweep was
// transport-down when we tried to clean it, so its stale orphans were never
// removed and EcBalance excludes it as both source and target. If it recovers
// just in time for generation, all shards land on a node we can neither clean
// nor balance off — a single point of failure that union-only verification
// still accepts, after which the originals are deleted. Abort instead.
for _, vid := range volumeIds {
genHost := bestReplicas[vid].ServerAddress()
if _, stillSkipped := skippedNodes[genHost]; stillSkipped {
return nil, fmt.Errorf("generate ec shards for volume %d aborted: selected source %s is still skipped after the orphan re-sweep", vid, genHost)
}
}
// generate ec shards using the best replica for each volume
ewg.Reset()
for _, vid := range volumeIds {
@ -316,7 +358,7 @@ func doEcEncode(commandEnv *CommandEnv, writer io.Writer, volumeIdToCollection m
})
}
if err := ewg.Wait(); err != nil {
return err
return nil, err
}
// mount all ec shards for the converted volume
@ -333,13 +375,220 @@ func doEcEncode(commandEnv *CommandEnv, writer io.Writer, volumeIdToCollection m
return nil
})
}
if err := ewg.Wait(); err != nil {
return nil, err
}
return skippedNodes, nil
}
// clearPreexistingEcShards removes EC shards and index files left over from a
// previous (failed or partial) encode of the given volume ids, on every node
// that still reports them, so a fresh encode regenerates from a clean slate.
// Scans all disk types. The normal .dat/.idx — the source of truth for this
// encode — is untouched; only orphaned EC artifacts are deleted.
//
// Returns the set of nodes skipped as unreachable. A skipped node may still hold
// an un-deleted orphan from a prior run; if it recovers it must be kept out of
// this encode's shard distribution, or the balance could install the new
// generation alongside the stale orphan and mix generations on one node.
func clearPreexistingEcShards(commandEnv *CommandEnv, topologyInfo *master_pb.TopologyInfo, volumeIds []needle.VolumeId, volumeIdToCollection map[needle.VolumeId]string, maxParallelization int) (skipped map[pb.ServerAddress]struct{}, err error) {
wanted := make(map[uint32]bool, len(volumeIds))
for _, vid := range volumeIds {
wanted[uint32(vid)] = true
}
// Note which (node, vid) pairs the topology already reports EC shards for:
// those are mounted leftovers and cleaning them is required (fatal on
// error). Every other (node, vid) is swept best-effort to catch UNMOUNTED
// orphans left by a failed copy — invisible to the heartbeat, so absent
// here. A node that is down or holds nothing is a harmless no-op; a node
// unreachable now also cannot receive this encode's new generation, so a
// surviving orphan there keeps its old identity and the read guard rejects
// it. Always delete the full shard-id range so a wider custom ratio's
// leftovers are covered too.
reportedKey := func(addr pb.ServerAddress, vid uint32) string {
return string(addr) + "\x00" + strconv.Itoa(int(vid))
}
reported := make(map[string]struct{})
var nodes []pb.ServerAddress
eachDataNode(topologyInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) {
addr := pb.NewServerAddressFromDataNode(dn)
nodes = append(nodes, addr)
for _, diskInfo := range dn.DiskInfos {
for _, ecInfo := range diskInfo.EcShardInfos {
if wanted[ecInfo.Id] {
reported[reportedKey(addr, ecInfo.Id)] = struct{}{}
}
}
}
})
allShardIds := make([]erasure_coding.ShardId, erasure_coding.MaxShardCount)
for i := range allShardIds {
allShardIds[i] = erasure_coding.ShardId(i)
}
if len(reported) > 0 {
fmt.Printf("clearing stale EC shards reported for %d (node,volume) pair(s) before regenerating...\n", len(reported))
}
// Nodes skipped as unreachable, accumulated across the concurrent sweep tasks.
skipped = make(map[pb.ServerAddress]struct{})
var skippedMu sync.Mutex
ewg := NewErrorWaitGroup(maxParallelization)
for _, addr := range nodes {
for _, vid := range volumeIds {
fatal := false
if _, ok := reported[reportedKey(addr, uint32(vid))]; ok {
fatal = true
}
collection := volumeIdToCollection[vid]
ewg.Add(func() error {
if err := unmountAndDeleteEcShardsQuiet(commandEnv.option.GrpcDialOption, collection, vid, addr, allShardIds); err != nil {
// Surface a reachable node whose delete genuinely failed (its orphan would
// be re-stamped by a later copy installing the new .vif). A missing
// full_teardown ack from a reachable pre-upgrade node is fatal too: it may
// still hold an orphan a later copy would re-stamp into the new generation.
// Stay best-effort only for a node that is truly unreachable: codes.Unavailable
// alone is ambiguous — a genuinely-down node and a reachable Rust volume
// server in maintenance mode both return it (a Go server returns Unknown for
// maintenance, already fatal above). Confirm with a non-maintenance-gated Ping
// before skipping; skip only when the Ping itself transport-failed (nodeDown).
// A reachable maintenance node (nodeUp) CAN receive this generation, and an
// inconclusive Ping (nodeLivenessUnknown, e.g. a pre-Ping server returning
// Unimplemented — which means the node is up) does not prove the node is down,
// so both stay fatal rather than silently leaving a stale EC generation.
if fatal || errors.Is(err, errFullTeardownNotAcked) || !isNodeUnreachable(err) ||
classifyNodeLiveness(pingVolumeServer(commandEnv.option.GrpcDialOption, addr)) != nodeDown {
return fmt.Errorf("clear stale ec shards for volume %d on %s: %w", vid, addr, err)
}
glog.V(1).Infof("orphan sweep: volume %d on %s skipped (unreachable): %v", vid, addr, err)
skippedMu.Lock()
skipped[addr] = struct{}{}
skippedMu.Unlock()
}
return nil
})
}
}
if err := ewg.Wait(); err != nil {
return nil, err
}
return skipped, nil
}
// resweepSkippedNodes re-attempts the orphan teardown on the nodes that the
// initial sweep skipped as unreachable, just before shard generation. A node
// that recovered in the meantime — and is therefore eligible to host this
// encode's generation — has its teardown retried; if it now fully succeeds it is
// removed from skipped so the rebalance can use it as a source and move its
// shards off, instead of stranding all shards on the single generation host and
// collapsing fault tolerance. A node still transport-down stays skipped (the
// same leniency the initial sweep grants), and a node that came back reachable
// but whose delete genuinely failed is fatal, exactly as in the initial sweep,
// so a stale generation is never silently left behind. Mutates skipped in place.
func resweepSkippedNodes(commandEnv *CommandEnv, skipped map[pb.ServerAddress]struct{}, volumeIds []needle.VolumeId, volumeIdToCollection map[needle.VolumeId]string, maxParallelization int) error {
if len(skipped) == 0 {
return nil
}
allShardIds := make([]erasure_coding.ShardId, erasure_coding.MaxShardCount)
for i := range allShardIds {
allShardIds[i] = erasure_coding.ShardId(i)
}
addrs := make([]pb.ServerAddress, 0, len(skipped))
for addr := range skipped {
addrs = append(addrs, addr)
}
fmt.Printf("re-checking %d node(s) skipped by the orphan sweep before generating shards...\n", len(addrs))
// A node still down on every retried vid stays skipped; one that fully
// succeeds is un-skipped. Track per-node whether any retry still failed
// (down) so a node whose state is mixed across vids never gets un-skipped.
stillDown := make(map[pb.ServerAddress]struct{})
var mu sync.Mutex
ewg := NewErrorWaitGroup(maxParallelization)
for _, addr := range addrs {
for _, vid := range volumeIds {
collection := volumeIdToCollection[vid]
ewg.Add(func() error {
if err := unmountAndDeleteEcShardsQuiet(commandEnv.option.GrpcDialOption, collection, vid, addr, allShardIds); err != nil {
// Same decision as the initial sweep: a reachable node whose delete
// genuinely failed (or did not ack a full teardown, or whose liveness is
// inconclusive) is fatal, since it could hold an orphan a later copy
// re-stamps into this generation. Only a node still transport-down stays
// skipped.
if errors.Is(err, errFullTeardownNotAcked) || !isNodeUnreachable(err) ||
classifyNodeLiveness(pingVolumeServer(commandEnv.option.GrpcDialOption, addr)) != nodeDown {
return fmt.Errorf("re-clear stale ec shards for volume %d on %s: %w", vid, addr, err)
}
glog.V(1).Infof("orphan re-sweep: volume %d on %s still skipped (unreachable): %v", vid, addr, err)
mu.Lock()
stillDown[addr] = struct{}{}
mu.Unlock()
}
return nil
})
}
}
if err := ewg.Wait(); err != nil {
return err
}
for _, addr := range addrs {
if _, down := stillDown[addr]; !down {
delete(skipped, addr)
glog.V(0).Infof("orphan re-sweep: node %s recovered and was cleaned; it will participate in the EC rebalance", addr)
}
}
return nil
}
// isNodeUnreachable reports whether err means the volume server could not be
// reached at all, as opposed to an RPC that reached the node and failed. Only an
// unreachable node is safe to skip in the orphan sweep. A dead peer surfaces as
// a gRPC codes.Unavailable from the RPC (the dial is lazy, so it never fails at
// connect time); any non-status error reached node logic and is treated as
// reachable, so the sweep stays fatal rather than silently leaving stale state.
func isNodeUnreachable(err error) bool {
if err == nil {
return false
}
st, ok := status.FromError(err)
return ok && st.Code() == codes.Unavailable
}
// nodeLiveness is the tri-state result of a pingVolumeServer probe.
type nodeLiveness int
const (
// nodeUp: Ping succeeded — the node is reachable (e.g. a Rust volume server
// in maintenance mode that fails the delete but answers Ping).
nodeUp nodeLiveness = iota
// nodeDown: Ping itself transport-failed with codes.Unavailable — the node is
// confirmed unreachable. The only state the orphan sweep may skip.
nodeDown
// nodeLivenessUnknown: Ping reached failing logic with any non-Unavailable
// code (Internal, ResourceExhausted, Unimplemented from a pre-Ping server, …)
// or a non-status error. This does NOT prove the node is down, so it is fatal.
nodeLivenessUnknown
)
// classifyNodeLiveness maps a pingVolumeServer error into the tri-state. A nil
// error is nodeUp, a transport codes.Unavailable is nodeDown (reusing the same
// rule as isNodeUnreachable), and every other Ping failure is nodeLivenessUnknown.
func classifyNodeLiveness(pingErr error) nodeLiveness {
if pingErr == nil {
return nodeUp
}
if isNodeUnreachable(pingErr) {
return nodeDown
}
return nodeLivenessUnknown
}
func verifyEcShardsBeforeDelete(commandEnv *CommandEnv, volumeIds []needle.VolumeId, diskType types.DiskType) error {
// Shard relocations from the preceding EC balance reach the master via
// volume-server heartbeats, so freshly distributed shards may not all be

6
weed/storage/erasure_coding/ec_shard.go
View File

@ -134,9 +134,13 @@ func EcShardBaseFileName(collection string, id int) (baseFileName string) {
}
func (shard *EcVolumeShard) Close() {
// Close the fd but do NOT nil it: a reader resolves the shard under the
// ecVolumesLock then calls ReadAt after the lock is released, so a
// concurrent eviction that nils the field would race that read and panic.
// Leaving the field set means ReadAt hits a closed fd and returns a clean
// error (the caller recovers from parity) with no data race on the field.
if shard.ecdFile != nil {
_ = shard.ecdFile.Close()
shard.ecdFile = nil
}
}

18
weed/storage/erasure_coding/ec_volume.go
View File

@ -44,6 +44,10 @@ type EcVolume struct {
ExpireAtSec uint64 //ec volume destroy time, calculated from the ec volume was created
ECContext *ECContext // EC encoding parameters
// EncodeTsNs is the encode time (unix nanos) loaded from .vif; reads carry it
// so a shard from a different encode run is rejected. 0 for pre-upgrade volumes.
EncodeTsNs int64
// ecjFileSize mirrors the on-disk size of the .ecj deletion journal and
// is maintained under ecjFileAccessLock. It is only used by IO helpers
// (seek/truncate) — the authoritative runtime delete count comes from
@ -150,6 +154,7 @@ func NewEcVolume(diskType types.DiskType, dir string, dirIdx string, collection
if volumeInfo.EcShardConfig != nil {
ds := int(volumeInfo.EcShardConfig.DataShards)
ps := int(volumeInfo.EcShardConfig.ParityShards)
ev.EncodeTsNs = volumeInfo.EcShardConfig.GetEncodeTsNs()
// Validate shard counts to prevent zero or invalid values
if ds <= 0 || ps <= 0 || ds+ps > MaxShardCount {
@ -227,16 +232,19 @@ func (ev *EcVolume) Close() {
for _, s := range ev.Shards {
s.Close()
}
ev.ecjFileAccessLock.Lock()
if ev.ecjFile != nil {
ev.ecjFileAccessLock.Lock()
_ = ev.ecjFile.Close()
ev.ecjFile = nil
ev.ecjFileAccessLock.Unlock()
}
ev.ecjFileAccessLock.Unlock()
if ev.ecxFile != nil {
_ = ev.ecxFile.Sync()
// Do NOT nil ecxFile: LocateEcShardNeedle reads it without the
// ecVolumesLock after the resolving lookup released it, so a concurrent
// eviction that nils the field would race that read. A closed-but-set fd
// yields a clean read error (recovered from parity) and no data race.
_ = ev.ecxFile.Close()
ev.ecxFile = nil
}
}
@ -244,13 +252,13 @@ func (ev *EcVolume) Close() {
// This ensures that deletions made via DeleteNeedleFromEcx are visible
// to other processes/file handles that may read these files.
func (ev *EcVolume) Sync() {
ev.ecjFileAccessLock.Lock()
if ev.ecjFile != nil {
ev.ecjFileAccessLock.Lock()
if err := ev.ecjFile.Sync(); err != nil {
glog.Warningf("failed to sync ecj file for volume %d: %v", ev.VolumeId, err)
}
ev.ecjFileAccessLock.Unlock()
}
ev.ecjFileAccessLock.Unlock()
if ev.ecxFile != nil {
if err := ev.ecxFile.Sync(); err != nil {
glog.Warningf("failed to sync ecx file for volume %d: %v", ev.VolumeId, err)

8
weed/storage/erasure_coding/ec_volume_delete.go
View File

@ -64,6 +64,14 @@ func (ev *EcVolume) DeleteNeedleFromEcx(needleId types.NeedleId) (err error) {
return nil
}
// Close nils ecjFile under this same lock, so a delete that resolved its
// .ecx lookup before an eviction (e.g. the generate-time UnloadEcVolume)
// can reach here with no journal fd. Bail with a clear error rather than
// operating on the closed/nil handle.
if ev.ecjFile == nil {
return fmt.Errorf("ec volume %d closed", ev.VolumeId)
}
b := make([]byte, types.NeedleIdSize)
types.NeedleIdToBytes(b, needleId)

39
weed/storage/erasure_coding/ec_volume_test.go
View File

@ -7,8 +7,10 @@ import (
"github.com/stretchr/testify/assert"
"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
"github.com/seaweedfs/seaweedfs/weed/storage/types"
"github.com/seaweedfs/seaweedfs/weed/storage/volume_info"
)
func TestPositioning(t *testing.T) {
@ -52,3 +54,40 @@ func TestPositioning(t *testing.T) {
}
}
// TestNewEcVolumeLoadsEncodeTsNs pins that the per-encode identity stamped into
// .vif is loaded onto the EcVolume, so reads can reject a shard from a different
// encode run.
func TestNewEcVolumeLoadsEncodeTsNs(t *testing.T) {
dir := t.TempDir()
const vid = needle.VolumeId(123)
base := EcShardFileName("", dir, int(vid))
// A 0-byte .ecx is a valid index (no live needles) and lets NewEcVolume mount.
if err := os.WriteFile(base+".ecx", nil, 0o644); err != nil {
t.Fatalf("write .ecx: %v", err)
}
const tsNs int64 = 1717000000000000123
vi := &volume_server_pb.VolumeInfo{
Version: uint32(needle.Version3),
EcShardConfig: &volume_server_pb.EcShardConfig{
DataShards: 10,
ParityShards: 4,
EncodeTsNs: tsNs,
},
}
if err := volume_info.SaveVolumeInfo(base+".vif", vi); err != nil {
t.Fatalf("save .vif: %v", err)
}
ev, err := NewEcVolume(types.HardDriveType, dir, dir, "", vid)
if err != nil {
t.Fatalf("NewEcVolume: %v", err)
}
defer ev.Close()
if ev.EncodeTsNs != tsNs {
t.Errorf("EncodeTsNs = %d, want %d", ev.EncodeTsNs, tsNs)
}
}

124
weed/storage/store_ec.go
View File

@ -252,31 +252,39 @@ func (s *Store) MountEcShards(collection string, vid needle.VolumeId, shardId er
}
func (s *Store) UnmountEcShards(vid needle.VolumeId, shardId erasure_coding.ShardId) error {
diskId, ecShard, found := s.findEcShard(vid, shardId)
if !found {
return nil
// Walk every disk: a split-disk reconciled volume can mount the same vid on
// more than one disk, so a first-match unmount would leave a sibling copy
// mounted and heartbeating. Emit one deletion delta per disk.
unmountedAny := false
var lastErr error
for diskId, location := range s.Locations {
ecShard, found := location.FindEcShard(vid, shardId)
if !found {
continue
}
if deleted := location.UnloadEcShard(vid, shardId); deleted {
si := erasure_coding.NewShardsInfo()
si.Set(erasure_coding.NewShardInfo(shardId, 0))
s.DeletedEcShardsChan <- &master_pb.VolumeEcShardInformationMessage{
Id: uint32(vid),
Collection: ecShard.Collection,
EcIndexBits: si.Bitmap(),
ShardSizes: si.SizesInt64(),
DiskType: string(ecShard.DiskType),
DiskId: uint32(diskId),
}
glog.V(0).Infof("UnmountEcShards %d.%d disk_id:%d", vid, shardId, diskId)
unmountedAny = true
} else {
lastErr = fmt.Errorf("UnmountEcShards %d.%d not found on disk %d", vid, shardId, diskId)
}
}
si := erasure_coding.NewShardsInfo()
si.Set(erasure_coding.NewShardInfo(shardId, 0))
message := master_pb.VolumeEcShardInformationMessage{
Id: uint32(vid),
Collection: ecShard.Collection,
EcIndexBits: si.Bitmap(),
ShardSizes: si.SizesInt64(),
DiskType: string(ecShard.DiskType),
DiskId: diskId,
// nil when no disk held the shard (idempotent re-unmount).
if !unmountedAny {
return lastErr
}
location := s.Locations[diskId]
if deleted := location.UnloadEcShard(vid, shardId); deleted {
glog.V(0).Infof("UnmountEcShards %d.%d disk_id:%d", vid, shardId, diskId)
s.DeletedEcShardsChan <- &message
return nil
}
return fmt.Errorf("UnmountEcShards %d.%d not found on disk %d", vid, shardId, diskId)
return nil
}
func (s *Store) findEcShard(vid needle.VolumeId, shardId erasure_coding.ShardId) (diskId uint32, shard *erasure_coding.EcVolumeShard, found bool) {
@ -294,6 +302,21 @@ func (s *Store) FindEcShard(vid needle.VolumeId, shardId erasure_coding.ShardId)
return s.findEcShard(vid, shardId)
}
// FindEcVolumeWithShard returns the EcVolume on the disk that owns the given
// shard, plus the shard. The read guard must check the identity of the volume
// that owns the bytes served: on a multi-disk server one vid can hold shards
// from different encode runs across disks, so a first-match volume can differ.
func (s *Store) FindEcVolumeWithShard(vid needle.VolumeId, shardId erasure_coding.ShardId) (*erasure_coding.EcVolume, *erasure_coding.EcVolumeShard, bool) {
for _, location := range s.Locations {
if shard, found := location.FindEcShard(vid, shardId); found {
if ev, ok := location.FindEcVolume(vid); ok {
return ev, shard, true
}
}
}
return nil, nil, false
}
func (s *Store) FindEcVolume(vid needle.VolumeId) (*erasure_coding.EcVolume, bool) {
for _, location := range s.Locations {
if s, found := location.FindEcVolume(vid); found {
@ -334,6 +357,14 @@ func (s *Store) DestroyEcVolume(vid needle.VolumeId) {
}
}
// UnloadEcVolume drops any in-memory EcVolume for vid from every disk and closes
// its fds without deleting files, so a following unlink frees the inodes.
func (s *Store) UnloadEcVolume(vid needle.VolumeId) {
for _, location := range s.Locations {
location.unloadEcVolume(vid)
}
}
func (s *Store) ReadEcShardNeedle(vid needle.VolumeId, n *needle.Needle, onReadSizeFn func(size types.Size)) (int, error) {
for _, location := range s.Locations {
if localEcVolume, found := location.FindEcVolume(vid); found {
@ -422,7 +453,7 @@ func (s *Store) readOneEcShardInterval(needleId types.NeedleId, ecVolume *erasur
// try reading directly
if hasShardIdLocation {
_, is_deleted, err = s.readRemoteEcShardInterval(sourceDataNodes, needleId, ecVolume.VolumeId, shardId, data, actualOffset)
_, is_deleted, err = s.readRemoteEcShardInterval(sourceDataNodes, needleId, ecVolume.VolumeId, shardId, data, actualOffset, ecVolume.EncodeTsNs)
if err == nil {
return
}
@ -490,12 +521,20 @@ func (s *Store) cachedLookupEcShardLocations(ecVolume *erasure_coding.EcVolume)
}
func (s *Store) readLocalEcShardInterval(ecVolume *erasure_coding.EcVolume, shardId erasure_coding.ShardId, buf []byte, offset int64) error {
// findEcShard walks every DiskLocation under ecVolumesLock; the
// Resolve the shard together with the EcVolume on the disk that owns it; the
// shard may live on a sibling disk of this server.
_, shard, found := s.findEcShard(ecVolume.VolumeId, shardId)
ownerVolume, shard, found := s.FindEcVolumeWithShard(ecVolume.VolumeId, shardId)
if !found {
return fmt.Errorf("shard %d for volume %d: %w", shardId, ecVolume.VolumeId, errShardNotLocal)
}
// Skip a local shard whose identity doesn't match the caller's index, so the
// read recovers from the correct generation. Lenient only when the caller has
// no identity (pre-upgrade): a known caller must not accept an unstamped local
// shard, which would serve a stale pre-upgrade generation.
if ecVolume.EncodeTsNs != 0 && ecVolume.EncodeTsNs != ownerVolume.EncodeTsNs {
glog.V(1).Infof("skip local ec shard %d.%d from a different encode run: caller EncodeTsNs %d, local %d", ecVolume.VolumeId, shardId, ecVolume.EncodeTsNs, ownerVolume.EncodeTsNs)
return fmt.Errorf("shard %d for volume %d: %w", shardId, ecVolume.VolumeId, errShardNotLocal)
}
readBytes, err := shard.ReadAt(buf, offset)
if err != nil {
@ -508,7 +547,7 @@ func (s *Store) readLocalEcShardInterval(ecVolume *erasure_coding.EcVolume, shar
return nil
}
func (s *Store) readRemoteEcShardInterval(sourceDataNodes []pb.ServerAddress, needleId types.NeedleId, vid needle.VolumeId, shardId erasure_coding.ShardId, buf []byte, offset int64) (n int, is_deleted bool, err error) {
func (s *Store) readRemoteEcShardInterval(sourceDataNodes []pb.ServerAddress, needleId types.NeedleId, vid needle.VolumeId, shardId erasure_coding.ShardId, buf []byte, offset int64, expectedEncodeTsNs int64) (n int, is_deleted bool, err error) {
if len(sourceDataNodes) == 0 {
return 0, false, fmt.Errorf("failed to find ec shard %d.%d", vid, shardId)
@ -516,7 +555,7 @@ func (s *Store) readRemoteEcShardInterval(sourceDataNodes []pb.ServerAddress, ne
for _, sourceDataNode := range sourceDataNodes {
glog.V(3).Infof("read remote ec shard %d.%d from %s", vid, shardId, sourceDataNode)
n, is_deleted, err = s.doReadRemoteEcShardInterval(sourceDataNode, needleId, vid, shardId, buf, offset)
n, is_deleted, err = s.doReadRemoteEcShardInterval(sourceDataNode, needleId, vid, shardId, buf, offset, expectedEncodeTsNs)
if err == nil {
return
}
@ -526,17 +565,18 @@ func (s *Store) readRemoteEcShardInterval(sourceDataNodes []pb.ServerAddress, ne
return
}
func (s *Store) doReadRemoteEcShardInterval(sourceDataNode pb.ServerAddress, needleId types.NeedleId, vid needle.VolumeId, shardId erasure_coding.ShardId, buf []byte, offset int64) (n int, is_deleted bool, err error) {
func (s *Store) doReadRemoteEcShardInterval(sourceDataNode pb.ServerAddress, needleId types.NeedleId, vid needle.VolumeId, shardId erasure_coding.ShardId, buf []byte, offset int64, expectedEncodeTsNs int64) (n int, is_deleted bool, err error) {
err = operation.WithVolumeServerClient(false, sourceDataNode, s.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
// copy data slice
shardReadClient, err := client.VolumeEcShardRead(context.Background(), &volume_server_pb.VolumeEcShardReadRequest{
VolumeId: uint32(vid),
ShardId: uint32(shardId),
Offset: offset,
Size: int64(len(buf)),
FileKey: uint64(needleId),
VolumeId: uint32(vid),
ShardId: uint32(shardId),
Offset: offset,
Size: int64(len(buf)),
FileKey: uint64(needleId),
EncodeTsNs: expectedEncodeTsNs,
})
if err != nil {
return fmt.Errorf("failed to start reading ec shard %d.%d from %s: %v", vid, shardId, sourceDataNode, err)
@ -550,6 +590,12 @@ func (s *Store) doReadRemoteEcShardInterval(sourceDataNode pb.ServerAddress, nee
if receiveErr != nil {
return fmt.Errorf("receiving ec shard %d.%d from %s: %v", vid, shardId, sourceDataNode, receiveErr)
}
// Validate the served shard's identity client-side, so the guard holds
// even against a pre-upgrade server that ignored the request field (it
// returns 0). A mismatch fails the read; the caller recovers from parity.
if expectedEncodeTsNs != 0 && resp.EncodeTsNs != expectedEncodeTsNs {
return fmt.Errorf("ec shard %d.%d from %s belongs to a different encode run (want %d, got %d)", vid, shardId, sourceDataNode, expectedEncodeTsNs, resp.EncodeTsNs)
}
if resp.IsDeleted {
is_deleted = true
}
@ -563,6 +609,16 @@ func (s *Store) doReadRemoteEcShardInterval(sourceDataNode pb.ServerAddress, nee
return 0, is_deleted, fmt.Errorf("read ec shard %d.%d from %s: %v", vid, shardId, sourceDataNode, err)
}
// A non-deleted interval must arrive whole: the server stamps EncodeTsNs only
// on chunks that carry bytes, so a short or empty stream (e.g. immediate EOF
// from a pre-upgrade or stale server) leaves the buffer partly zero-filled and
// unvalidated. Reject it so the caller recovers from parity. The is_deleted
// short-circuit legitimately returns n=0 with no data and is exempt, matching
// readLocalEcShardInterval's got==len(buf) rule for the local path.
if !is_deleted && n != len(buf) {
return n, is_deleted, fmt.Errorf("short read ec shard %d.%d from %s: got %d want %d", vid, shardId, sourceDataNode, n, len(buf))
}
return
}
@ -595,7 +651,7 @@ func (s *Store) recoverOneRemoteEcShardInterval(needleId types.NeedleId, ecVolum
go func(shardId erasure_coding.ShardId, locations []pb.ServerAddress) {
defer wg.Done()
data := make([]byte, len(buf))
nRead, isDeleted, readErr := s.readRemoteEcShardInterval(locations, needleId, ecVolume.VolumeId, shardId, data, offset)
nRead, isDeleted, readErr := s.readRemoteEcShardInterval(locations, needleId, ecVolume.VolumeId, shardId, data, offset, ecVolume.EncodeTsNs)
if readErr != nil {
glog.V(3).Infof("recover: readRemoteEcShardInterval %d.%d %d bytes from %+v: %v", ecVolume.VolumeId, shardId, nRead, locations, readErr)
forgetShardId(ecVolume, shardId)

2
weed/storage/store_ec_scrub.go
View File

@ -54,7 +54,7 @@ func (s *Store) ScrubEcVolume(vid needle.VolumeId) (int64, []*volume_server_pb.E
sourceDataNodes, ok := ecv.ShardLocations[shardId]
ecv.ShardLocationsLock.RUnlock()
if ok {
if _, _, err := s.readRemoteEcShardInterval(sourceDataNodes, id, ecv.VolumeId, shardId, chunk, offset); err == nil {
if _, _, err := s.readRemoteEcShardInterval(sourceDataNodes, id, ecv.VolumeId, shardId, chunk, offset, ecv.EncodeTsNs); err == nil {
data = append(data, chunk...)
continue
}

33
weed/worker/tasks/erasure_coding/ec_task.go
View File

@ -616,10 +616,24 @@ func (t *ErasureCodingTask) generateEcShardsLocally(localFiles map[string]string
}).Info("EC journal file generated")
}
// Generate .vif file (volume info)
// Always stamp the encode identity into the .vif so the read guard stays on.
// The ratio is the resolved one from the encoder's protection, defaulting to
// the context this path encodes with (not t.dataShards, which this path does
// not pass to the encoder).
vifFile := baseName + ".vif"
defaultCtx := erasure_coding.NewDefaultECContext("", 0)
ecShardConfig := &volume_server_pb.EcShardConfig{
DataShards: uint32(defaultCtx.DataShards),
ParityShards: uint32(defaultCtx.ParityShards),
EncodeTsNs: time.Now().UnixNano(),
}
if ecBitrot != nil && ecBitrot.EcShardConfig != nil {
ecShardConfig.DataShards = ecBitrot.EcShardConfig.DataShards
ecShardConfig.ParityShards = ecBitrot.EcShardConfig.ParityShards
}
volumeInfo := &volume_server_pb.VolumeInfo{
Version: uint32(needle.GetCurrentVersion()),
Version: uint32(needle.GetCurrentVersion()),
EcShardConfig: ecShardConfig,
}
if err := volume_info.SaveVolumeInfo(vifFile, volumeInfo); err != nil {
glog.Warningf("Failed to create .vif file: %v", err)
@ -963,13 +977,18 @@ func unmountAndDeleteEcShards(
}); err != nil {
return fmt.Errorf("unmount: %w", err)
}
if _, err := client.VolumeEcShardsDelete(ctx, &volume_server_pb.VolumeEcShardsDeleteRequest{
VolumeId: volumeID,
Collection: collection,
ShardIds: shardIds,
}); err != nil {
resp, err := client.VolumeEcShardsDelete(ctx, &volume_server_pb.VolumeEcShardsDeleteRequest{
VolumeId: volumeID,
Collection: collection,
ShardIds: shardIds,
FullTeardown: true,
})
if err != nil {
return fmt.Errorf("delete: %w", err)
}
if !resp.GetFullTeardownDone() {
return fmt.Errorf("delete: %s did not perform full teardown (pre-upgrade volume server?); a stale EC generation may remain", destination)
}
return nil
})
}

50
weed/worker/tasks/erasure_coding/ec_task_test.go
View File

@ -11,6 +11,7 @@ import (
"github.com/seaweedfs/seaweedfs/test/volume_server/framework"
"github.com/seaweedfs/seaweedfs/test/volume_server/matrix"
"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
"github.com/seaweedfs/seaweedfs/weed/storage/volume_info"
"github.com/stretchr/testify/require"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure"
@ -77,6 +78,55 @@ func TestCopyVolumeFilesToWorkerUsesCurrentCompactionRevision(t *testing.T) {
require.Equal(t, int64(fileStatus.GetIdxFileSize()), idxInfo.Size())
}
// The worker-local encode path must stamp the EC ratio and an encode identity
// into the .vif, or the read guard is silently off for worker-encoded volumes.
func TestGenerateEcShardsLocallyStampsEncodeIdentity(t *testing.T) {
if testing.Short() {
t.Skip("skipping integration test in short mode")
}
clusterHarness := framework.StartVolumeCluster(t, matrix.P1())
conn, grpcClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress())
defer conn.Close()
const volumeID = uint32(952)
framework.AllocateVolume(t, grpcClient, volumeID, "")
httpClient := framework.NewHTTPClient()
fid := framework.NewFileID(volumeID, 2001, 0x3333CCCC)
resp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), fid, []byte("payload-for-ec-encode-identity"))
_ = framework.ReadAllAndClose(t, resp)
require.Equal(t, http.StatusCreated, resp.StatusCode)
task := NewErasureCodingTask(
"ec-encode-identity",
clusterHarness.VolumeServerAddress(),
volumeID,
"",
grpc.WithTransportCredentials(insecure.NewCredentials()),
)
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
require.NoError(t, task.markReplicasReadonly(ctx))
localFiles, err := task.copyVolumeFilesToWorker(ctx, t.TempDir())
require.NoError(t, err)
shardFiles, err := task.generateEcShardsLocally(localFiles, t.TempDir())
require.NoError(t, err)
vifPath := shardFiles["vif"]
require.NotEmpty(t, vifPath, "generate must produce a .vif")
vi, _, found, err := volume_info.MaybeLoadVolumeInfo(vifPath)
require.NoError(t, err)
require.True(t, found)
require.NotNil(t, vi.EcShardConfig, "worker .vif must record the EC config")
require.Greater(t, vi.EcShardConfig.DataShards, uint32(0))
require.Greater(t, vi.EcShardConfig.ParityShards, uint32(0))
require.NotZero(t, vi.EcShardConfig.EncodeTsNs, "worker-encoded .vif must carry an encode identity")
}
func compactVolumeOnce(t *testing.T, grpcClient volume_server_pb.VolumeServerClient, volumeID uint32) {
t.Helper()