mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-09-23 16:34:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

volume: load the .ecj deletion journal in chunks, and repair a torn tail (#11408)

Browse Source 
* volume: load the .ecj deletion journal in chunks, and repair a torn tail

Two independent defects in the EC deletion journal's load path.

1. The loader issued one NEEDLE_ID_SIZE-byte positional read per entry.

   That is fine for a healthy journal -- kilobytes -- and pathological for a
   large one. A `.ecj` is semantically a SET of deleted needle ids but is
   written as an append-only log that nothing dedupes, and several paths append
   a peer's ENTIRE journal onto the local one (VolumeEcShardsCopy with
   copy_ecj_file, EC index recovery, and ec_decode's deliberate cross-holder
   merge), so a volume whose shards are repeatedly balanced between two servers
   grows the file without bound.

   Observed in production: 1.51 TB and 1.30 TB on the two holders of one 10+4
   volume containing ~100 distinct ids. At that size the per-entry loop is
   ~188e9 syscalls, run synchronously while holding the deleted_needles write
   lock and before the HTTP port opens. The process sits at 100% of one core
   with a small RSS -- the set stays tiny because the ids repeat -- reading at a
   few MiB/s because 8-byte reads defeat readahead, logs nothing after "Adding
   storage location", and ignores SIGTERM. The master then unregisters every
   volume it holds and reads of them fail. 4.46 and 4.47 are both affected.

   Read in 1 MiB chunks and build into a local set, merging once at the end so
   the write lock is not held for the whole scan. Measured on a 256 MiB journal
   of 100 distinct ids: 33,554,500 syscalls -> 257, identical resulting set.

2. A torn tail silently corrupted later deletes.

   The journal handle is in append mode, so writes land at the physical end
   regardless of alignment. A trailing partial record therefore pushed every
   later append out of alignment: the loader skipped the partial bytes, but the
   next mount decoded them together with the leading bytes of the following
   entry, producing one garbage id and dropping the delete that came after the
   tear -- after acknowledging it.

   Truncate to a whole number of records at mount, before anything can append.
   The repair uses its own read+write (non-append) handle: on Windows,
   append(true) requests FILE_APPEND_DATA without FILE_WRITE_DATA (and
   .write(true) is subsumed by .append(true)), so SetEndOfFile through the
   journal handle fails with ERROR_ACCESS_DENIED.

   The same trap exists in journal_delete's recovery path, which calls set_len
   on the append handle to roll back a partial write whose sync failed. It is
   error-handled rather than fatal, so on Windows that rollback silently does
   not happen. Untouched here; worth a separate fix.

Bounding the journal's growth needs compaction, which is deliberately not in
this change: replacing the file under a store that can hold several EcVolume
instances for one volume id requires coordinating with the other holders, and
that belongs at the store layer. Sent separately.

Tests: a journal spanning several read chunks loads every entry; a trailing
partial record is ignored rather than panicking; a torn tail is truncated at
mount and a delete taken afterwards survives a remount.

* volume: roll back a failed .ecj append through a dedicated write handle

The append handle lacks FILE_WRITE_DATA on Windows, so the set_len
rollback after a failed sync silently did nothing and the journal could
drift one record past deleted_needles. Same trap as the torn-tail repair
in this file; fix it the same way. Also format the new tests.

* volume: mirror chunked .ecj load and torn-tail repair in Go

---------

Co-authored-by: chrislusf <chrislusf@users.noreply.github.com>
Co-authored-by: Devin <devin@cognition.ai>
This commit is contained in:
Eliah Rusin
2026-09-20 23:26:41 -07:00
committed by GitHub
co-authored by chrislusf Devin
parent 4bb40732bb
commit ca62d4297b
3 changed files with 327 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
seaweed-volume/src/storage/erasure_coding/ec_volume.rs
+210 -15
View File
@@ -44,6 +44,10 @@ pub(crate) struct ShardLocationCache {
stale: bool,
}
/// Bytes read per positional read when seeding `deleted_needles` from `.ecj`.
/// A multiple of `NEEDLE_ID_SIZE`; 1 MiB is 131072 entries per syscall.
const ECJ_LOAD_CHUNK_BYTES: usize = 1 << 20;
/// An erasure-coded volume managing its local shards and index.
pub struct EcVolume {
pub volume_id: VolumeId,
@@ -494,6 +498,45 @@ impl EcVolume {
let ecj_base =
crate::storage::volume::volume_file_name(&vol.ecx_actual_dir, collection, volume_id);
let ecj_path = format!("{}.ecj", ecj_base);
// Repair a torn tail BEFORE the append handle exists.
//
// The file is a flat array of fixed-size records and the journal handle
// is in append mode, so every write lands at the physical end. A
// trailing partial record therefore knocks every later append out of
// alignment: the loader below skips the partial bytes, but the next
// mount decodes those bytes together with the leading bytes of a real
// entry, yielding one garbage id and silently dropping the delete that
// followed the tear. Truncating to a whole number of records costs at
// most one incomplete id that was never readable anyway.
//
// This deliberately uses its own read+write handle rather than the
// append handle opened below: on Windows, `append(true)` requests
// FILE_APPEND_DATA *without* FILE_WRITE_DATA (and `.write(true)` is
// subsumed by `.append(true)`), so SetEndOfFile through that handle
// fails with ERROR_ACCESS_DENIED. The handle is scoped so it is closed
// again before the append handle opens.
{
let repair = open_volume_file(
OpenOptions::new().read(true).write(true).create(true),
&ecj_path,
)?;
let on_disk = repair.metadata()?.len() as i64;
let ragged = on_disk % NEEDLE_ID_SIZE as i64;
if ragged != 0 {
let whole = on_disk - ragged;
tracing::warn!(
volume_id = volume_id.0,
collection = %collection,
on_disk_bytes = on_disk,
truncated_to = whole,
"truncating torn .ecj tail so later appends stay aligned",
);
repair.set_len(whole as u64)?;
repair.sync_all()?;
}
}
let ecj_file = open_volume_file(
OpenOptions::new()
.read(true)
@@ -693,6 +736,14 @@ impl EcVolume {
/// from `new()` under exclusive ownership of the just-constructed
/// EcVolume, so locking is not strictly required — but we take the
/// write lock anyway for symmetry with later mutations.
///
/// Read in large chunks. This previously issued one `NEEDLE_ID_SIZE`-byte
/// positional read per entry, which is fine for a healthy journal (a few
/// KB) and catastrophic for a bloated one: at the 1.51 TB seen in
/// production that is ~188e9 syscalls, so the server spun at 100% of one
/// core with a 31 MB RSS — the set stays small because the ids repeat —
/// and never opened its HTTP port, which made the master unregister every
/// volume it held. Chunked reads cut that by ~`ECJ_LOAD_CHUNK_BYTES / 8`.
fn load_deleted_needles_from_ecj(&mut self) -> io::Result<()> {
let ecj_file = match self.ecj_file.as_ref() {
Some(f) => f,
@@ -701,35 +752,51 @@ impl EcVolume {
if self.ecj_file_size < NEEDLE_ID_SIZE as i64 {
return Ok(());
}
let mut buf = [0u8; NEEDLE_ID_SIZE];
let mut set = self
.deleted_needles
.write()
.map_err(|_| io::Error::other("deleted_needles lock poisoned"))?;
let mut off: i64 = 0;
while off + NEEDLE_ID_SIZE as i64 <= self.ecj_file_size {
// Build into a local set and merge at the end. The previous version
// held the `deleted_needles` write lock for the whole scan, which on a
// bloated journal is the entire (unbounded) startup.
let mut loaded: HashSet<NeedleId> = HashSet::new();
let mut buf = vec![0u8; ECJ_LOAD_CHUNK_BYTES];
let end = self.ecj_file_size as u64;
let mut off: u64 = 0;
while off + NEEDLE_ID_SIZE as u64 <= end {
// Whole entries only; a trailing partial record is ignored, as the
// per-entry loop did by construction.
let mut want = std::cmp::min(ECJ_LOAD_CHUNK_BYTES as u64, end - off) as usize;
want -= want % NEEDLE_ID_SIZE;
if want == 0 {
break;
}
#[cfg(unix)]
{
use std::os::unix::fs::FileExt;
ecj_file.read_exact_at(&mut buf, off as u64)?;
ecj_file.read_exact_at(&mut buf[..want], off)?;
}
#[cfg(windows)]
{
// Positional read so concurrent readers of the shared .ecj
// handle can't interleave seek/read. Mirrors the
// read_exact_at helper at the bottom of this file.
read_exact_at(ecj_file, &mut buf, off as u64)?;
read_exact_at(ecj_file, &mut buf[..want], off)?;
}
#[cfg(not(any(unix, windows)))]
{
compile_error!("Platform not supported: only unix and windows are supported");
}
set.insert(NeedleId::from_bytes(&buf));
off += NEEDLE_ID_SIZE as i64;
for entry in buf[..want].chunks_exact(NEEDLE_ID_SIZE) {
loaded.insert(NeedleId::from_bytes(entry));
}
off += want as u64;
}
let mut set = self
.deleted_needles
.write()
.map_err(|_| io::Error::other("deleted_needles lock poisoned"))?;
set.extend(loaded);
Ok(())
}
/// Returns (file_count, delete_count) for this EC volume. Mirrors Go's
/// `EcVolume.FileAndDeleteCount`:
///
@@ -1553,9 +1620,21 @@ impl EcVolume {
// write_all may have extended the file on disk before
// sync_all failed; truncate back to the known-good size so
// the on-disk journal never drifts past `deleted_needles`.
if let Some(ecj) = self.ecj_file.as_mut()
&& let Err(trunc_err) = ecj.set_len(prev_ecj_size as u64)
{
// Uses its own write handle: on Windows the append handle
// lacks FILE_WRITE_DATA, so set_len through it fails with
// ERROR_ACCESS_DENIED and the rollback would silently not
// happen.
let ecj_path = format!(
"{}.ecj",
crate::storage::volume::volume_file_name(
&self.ecx_actual_dir,
&self.collection,
self.volume_id,
)
);
let rollback = open_volume_file(OpenOptions::new().write(true), &ecj_path)
.and_then(|f| f.set_len(prev_ecj_size as u64).and_then(|_| f.sync_all()));
if let Err(trunc_err) = rollback {
tracing::error!(
volume_id = self.volume_id.0,
needle_id = needle_id.0,
@@ -2333,6 +2412,122 @@ mod tests {
assert_eq!((fc, dc), (2, 2));
}
/// Write a raw `.ecj` containing `ids` repeated `repeats` times, i.e. the
/// shape the append paths produce when a peer's whole journal is
/// concatenated onto this one over and over.
fn write_bloated_ecj(
dir: &str,
collection: &str,
vid: VolumeId,
ids: &[NeedleId],
repeats: usize,
) {
let base = crate::storage::volume::volume_file_name(dir, collection, vid);
let mut one = vec![0u8; ids.len() * NEEDLE_ID_SIZE];
for (i, id) in ids.iter().enumerate() {
id.to_bytes(&mut one[i * NEEDLE_ID_SIZE..(i + 1) * NEEDLE_ID_SIZE]);
}
let mut f = File::create(format!("{}.ecj", base)).unwrap();
for _ in 0..repeats {
f.write_all(&one).unwrap();
}
f.sync_all().unwrap();
}
/// A journal whose length is not a whole number of entries must not lose
/// the entries that ARE complete, and must not panic.
#[test]
fn test_ecj_with_trailing_partial_entry() {
let tmp = TempDir::new().unwrap();
let dir = tmp.path().to_str().unwrap();
write_ecx_file(dir, "", VolumeId(40), &[]);
let base = crate::storage::volume::volume_file_name(dir, "", VolumeId(40));
let ids: Vec<NeedleId> = (1..=3).map(NeedleId).collect();
let mut bytes = vec![0u8; ids.len() * NEEDLE_ID_SIZE];
for (i, id) in ids.iter().enumerate() {
id.to_bytes(&mut bytes[i * NEEDLE_ID_SIZE..(i + 1) * NEEDLE_ID_SIZE]);
}
bytes.extend_from_slice(&[0xAB, 0xCD, 0xEF]); // torn tail
std::fs::write(format!("{}.ecj", base), &bytes).unwrap();
let vol = EcVolume::new(dir, dir, "", VolumeId(40)).unwrap();
assert_eq!(vol.read_deleted_needles().unwrap(), ids);
}
/// A torn tail must be truncated at mount, not merely skipped. The handle
/// is in append mode, so leaving the partial bytes in place would push
/// every later append out of alignment: the delete taken after the tear
/// would decode as garbage on the next mount and be silently lost.
#[test]
fn test_torn_ecj_tail_is_repaired_so_later_deletes_survive() {
let tmp = TempDir::new().unwrap();
let dir = tmp.path().to_str().unwrap();
// journal_delete only appends on a live->tombstone transition.
write_ecx_file(
dir,
"",
VolumeId(42),
&[(NeedleId(7), Offset::from_actual_offset(8), Size(10))],
);
let base = crate::storage::volume::volume_file_name(dir, "", VolumeId(42));
let ecj_path = format!("{}.ecj", base);
let ids: Vec<NeedleId> = (1..=3).map(NeedleId).collect();
let mut bytes = vec![0u8; ids.len() * NEEDLE_ID_SIZE];
for (i, id) in ids.iter().enumerate() {
id.to_bytes(&mut bytes[i * NEEDLE_ID_SIZE..(i + 1) * NEEDLE_ID_SIZE]);
}
bytes.extend_from_slice(&[0xAB, 0xCD, 0xEF]); // torn tail
std::fs::write(&ecj_path, &bytes).unwrap();
let mut vol = EcVolume::new(dir, dir, "", VolumeId(42)).unwrap();
// The tear is gone from disk, not just ignored in memory.
assert_eq!(
std::fs::metadata(&ecj_path).unwrap().len(),
(ids.len() * NEEDLE_ID_SIZE) as u64,
"torn tail should have been truncated at mount",
);
vol.journal_delete(NeedleId(7)).unwrap();
drop(vol);
// The delete taken after the repair must survive a remount.
let vol2 = EcVolume::new(dir, dir, "", VolumeId(42)).unwrap();
let deleted = vol2.read_deleted_needles().unwrap();
assert!(
deleted.contains(&NeedleId(7)),
"delete after a torn tail was lost: {:?}",
deleted,
);
assert_eq!(
deleted.len(),
ids.len() + 1,
"misaligned decode: {:?}",
deleted
);
}
/// A journal spanning several read chunks must load every entry — guards
/// the chunk-boundary arithmetic in the buffered loader.
#[test]
fn test_ecj_spanning_multiple_read_chunks() {
let tmp = TempDir::new().unwrap();
let dir = tmp.path().to_str().unwrap();
write_ecx_file(dir, "", VolumeId(41), &[]);
// 2.5 chunks' worth of DISTINCT ids, so nothing is masked by dedup and
// the total is not a chunk multiple.
let n = (ECJ_LOAD_CHUNK_BYTES / NEEDLE_ID_SIZE) * 5 / 2;
let ids: Vec<NeedleId> = (1..=n as u64).map(NeedleId).collect();
write_bloated_ecj(dir, "", VolumeId(41), &ids, 1);
let vol = EcVolume::new(dir, dir, "", VolumeId(41)).unwrap();
let deleted = vol.read_deleted_needles().unwrap();
assert_eq!(deleted.len(), n, "entries lost across a chunk boundary");
assert_eq!(deleted, ids);
}
#[test]
fn test_journal_delete_wrong_cookie() {
let tmp = TempDir::new().unwrap();
weed/storage/erasure_coding/ec_volume.go
+31 -5
View File
@@ -26,6 +26,11 @@ var (
destroyDelaySeconds int64 = 0
)
// ecjLoadChunkBytes bounds each positional read when seeding deletedNeedles
// from .ecj. A multiple of NeedleIdSize; 1 MiB is 131072 entries per syscall,
// which keeps a bloated journal from spending mount in per-entry reads.
const ecjLoadChunkBytes = 1 << 20
type EcVolume struct {
VolumeId needle.VolumeId
Collection string
@@ -207,6 +212,20 @@ func NewEcVolume(diskType types.DiskType, dir string, dirIdx string, collection
} else {
glog.Warningf("stat ec volume journal %s.ecj: %v", indexBaseFileName, statErr)
}
// Truncate a torn tail before the loader runs: appends land at the
// physical end, so a trailing partial record would misalign every later
// delete and lose it on the next mount.
if ragged := ev.ecjFileSize % int64(types.NeedleIdSize); ragged != 0 {
whole := ev.ecjFileSize - ragged
glog.Warningf("ec volume %d: truncating torn .ecj tail %d -> %d bytes", vid, ev.ecjFileSize, whole)
if truncErr := ev.ecjFile.Truncate(whole); truncErr != nil {
return nil, fmt.Errorf("ec volume %d: repair torn .ecj tail: %w", vid, truncErr)
}
if syncErr := ev.ecjFile.Sync(); syncErr != nil {
return nil, fmt.Errorf("ec volume %d: sync .ecj after tail repair: %w", vid, syncErr)
}
ev.ecjFileSize = whole
}
ev.deletedNeedles = make(map[types.NeedleId]struct{})
if loadErr := ev.loadDeletedNeedlesFromEcj(); loadErr != nil {
glog.Warningf("ec volume %d: load deleted needles from .ecj: %v", vid, loadErr)
@@ -636,13 +655,20 @@ func (ev *EcVolume) loadDeletedNeedlesFromEcj() error {
if ev.ecjFile == nil || ev.ecjFileSize < int64(types.NeedleIdSize) {
return nil
}
buf := make([]byte, types.NeedleIdSize)
for off := int64(0); off+int64(types.NeedleIdSize) <= ev.ecjFileSize; off += int64(types.NeedleIdSize) {
if _, err := ev.ecjFile.ReadAt(buf, off); err != nil {
buf := make([]byte, ecjLoadChunkBytes)
for off := int64(0); off+int64(types.NeedleIdSize) <= ev.ecjFileSize; {
want := min(int64(ecjLoadChunkBytes), ev.ecjFileSize-off)
want -= want % int64(types.NeedleIdSize)
if want == 0 {
break
}
if _, err := ev.ecjFile.ReadAt(buf[:want], off); err != nil {
return fmt.Errorf("read ecj at %d: %w", off, err)
}
id := types.BytesToNeedleId(buf)
ev.deletedNeedles[id] = struct{}{}
for i := int64(0); i+int64(types.NeedleIdSize) <= want; i += int64(types.NeedleIdSize) {
ev.deletedNeedles[types.BytesToNeedleId(buf[i:i+types.NeedleIdSize])] = struct{}{}
}
off += want
}
return nil
}
weed/storage/erasure_coding/ec_volume_ecj_test.go
+86
View File
@@ -0,0 +1,86 @@
package erasure_coding_test
import (
"os"
"testing"
erasure_coding "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
"github.com/seaweedfs/seaweedfs/weed/storage/types"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func ecjBytes(ids ...types.NeedleId) []byte {
b := make([]byte, 0, len(ids)*types.NeedleIdSize)
rec := make([]byte, types.NeedleIdSize)
for _, id := range ids {
types.NeedleIdToBytes(rec, id)
b = append(b, rec...)
}
return b
}
func mountEcVolume(t *testing.T, dir string, ecx, ecj []byte) (*erasure_coding.EcVolume, string) {
t.Helper()
base := erasure_coding.EcShardFileName("", dir, 7)
require.NoError(t, os.WriteFile(base+".ecx", ecx, 0644))
if ecj != nil {
require.NoError(t, os.WriteFile(base+".ecj", ecj, 0644))
}
require.NoError(t, os.WriteFile(base+".vif", []byte{}, 0644))
ev, err := erasure_coding.NewEcVolume("hdd", dir, dir, "", 7)
require.NoError(t, err)
return ev, base
}
// A crash mid-append leaves a partial record at the end of .ecj. Mount must
// drop it: deletes append at the physical end, so keeping the fragment would
// misalign every later record and lose those deletes on the next mount.
func TestEcjTornTailTruncatedOnMount(t *testing.T) {
dir := t.TempDir()
ecx := append(makeNeedleMapEntry(types.NeedleId(1), types.ToOffset(0), types.Size(100)),
makeNeedleMapEntry(types.NeedleId(4), types.ToOffset(8), types.Size(100))...)
ecj := append(ecjBytes(1, 2, 3), []byte{1, 2, 3, 4, 5}...)
ev, base := mountEcVolume(t, dir, ecx, ecj)
fi, err := os.Stat(base + ".ecj")
require.NoError(t, err)
assert.Equal(t, int64(3*types.NeedleIdSize), fi.Size())
for _, id := range []types.NeedleId{1, 2, 3} {
assert.True(t, ev.IsNeedleDeleted(id), "id %d", id)
}
assert.False(t, ev.IsNeedleDeleted(4))
require.NoError(t, ev.DeleteNeedleFromEcx(4))
ev.Close()
ev, _ = mountEcVolume(t, dir, ecx, nil)
defer ev.Close()
for _, id := range []types.NeedleId{1, 2, 3, 4} {
assert.True(t, ev.IsNeedleDeleted(id), "id %d", id)
}
}
// A journal larger than one load chunk must seed every record, including the
// ones straddling and following the chunk boundary.
func TestEcjLoadsAcrossChunkBoundary(t *testing.T) {
dir := t.TempDir()
const count = types.NeedleId(200_000) // 1.6 MiB > 1 MiB chunk
ecj := make([]byte, 0, int(count)*types.NeedleIdSize)
for id := types.NeedleId(1); id <= count; id++ {
rec := make([]byte, types.NeedleIdSize)
types.NeedleIdToBytes(rec, id)
ecj = append(ecj, rec...)
}
ev, _ := mountEcVolume(t, dir, nil, ecj)
defer ev.Close()
for _, id := range []types.NeedleId{1, 131072, 131073, count} {
assert.True(t, ev.IsNeedleDeleted(id), "id %d", id)
}
assert.False(t, ev.IsNeedleDeleted(count+1))
}