From 82f0a825641e20948868ae01f9fd3325ff0424ac Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 21 Oct 2026 23:57:09 -0700 Subject: [PATCH] filer+volume: make acknowledged writes durable under the fsync rule (#11943) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * volume: never compact a volume tiered to remote storage A tiered volume's data backend is the remote object, but compaction reads the local .dat path. When tiering removed the local .dat, every vacuum of a tiered volume with garbage fails with "no such file or directory", and the master's sweep and the worker's vacuum task keep selecting it. When the local .dat was kept, the compaction succeeds against the stale copy and the commit reloads the volume to read the remote object (old layout) through the compacted index, so live needles stop reading. Mirror the Rust volume server (#11545): - CompactByIndex and CompactByVolumeData refuse a volume with a remote file before creating .cpd/.cpx ("volume N is tiered to remote storage, cannot compact"). - CommitCompact refuses under the data-file locks when the volume became tiered during the copy, and removes .cpd/.cpx. - Store.CheckCompactVolume reports no garbage for a tiered volume, so the master's vacuum and volume.vacuum do not select it. - The worker's vacuum detection skips a volume with a remote copy, as balance and EC detection already do. Tests (each fails on master): TestCompactRefusesTieredVolume, TestCommitCompactRefusesVolumeTieredDuringCopy (on master the commit succeeds and a surviving needle no longer reads), TestPickPrimaryReplicaSkipsTieredVolumes. * volume: skip tiered replicas in vacuum selection, keep their real ratio Reporting a zero garbage ratio for a tiered volume hid it from volume.tier.compact, whose preflight reads that ratio while the volume is still remote, and did not exclude it from a vacuum with a zero threshold. Report the real ratio again, as the Rust volume server does, and have the master's vacuum check skip a replica whose volume info names remote storage, whatever the threshold or request. When CommitCompact refuses a volume tiered during the copy, also close the temporary needle map and remove .cpldb: nothing else would, since no later vacuum selects the volume. The tiered-compaction test no longer depends on random needle sizes reaching a 0.3 garbage ratio. * volume tier: do not attach a copy of a .dat a compaction replaced The upload copies the .dat, then records the remote file in the .vif and switches the backend. A compaction commit in between swaps in a compacted .dat and index; the upload then served the old-layout remote object through the new index, so live needles stopped reading. CommitCompact's tiered check cannot see this: the volume is not tiered until the switch. AttachRemoteDatFile records the remote file and switches the backend under the data-file locks, and refuses when the backend is no longer the file that was copied. The upload then removes the unused remote copy. * volume tier: keep a committed .vif entry when only the dir fsync fails AttachRemoteDatFile rolled back the in-memory remote entry on any SaveVolumeInfo error. A NotCrashDurableError comes after the rename, so the .vif on disk already names the copy; with the entry gone, the upload handler deleted that object, and a reload would serve from a missing remote file. Keep the entry and switch to the copy as the .vif says, and still return the error so the caller keeps the local .dat. * vacuum worker: filter tiered replicas instead of skipping the volume A partially tiered volume is the leftover of an interrupted volume.tier.upload; skipping the whole volume left garbage on its local replicas unreclaimed. Per-replica filtering matches the master's batchVacuumVolumeCheck and is safe: a tiered replica refuses both compaction and commit. * filer: sync leveldb entry writes when the path rule asks for fsync The fsync storage rule only reached the volume upload; the metadata writes behind an acknowledged create, update or delete still went to the leveldb journal with Sync:false, so a power loss could roll back an acknowledged bucket or object entry. A write-side flag on the entry context carries the request: the filer sets it from the path's storage rule in CreateEntry, UpdateEntry, EnsureDirectoryEntry and DeleteEntryMetaAndData, and the HTTP write path sets it for an explicit ?fsync=true. The leveldb stores write with Sync:true when the flag is present. * volume: flush the .idx before acknowledging a durable write A write with fsync synced only the .dat: the index row written on publish stayed in the page cache, so a power loss could leave the needle's bytes durable with nothing indexing them. The durable paths now run one .idx sync after publish, before the requests complete — one per syncWrite and one per batch run, paired with the .dat sync, the same split the Rust volume server makes between flush_dat and flush_idx. A failed .idx sync cannot be rolled back, so it stops the volume taking writes and fails the requests instead of acknowledging an index entry that may not survive. --------- Co-authored-by: Javier de la Torre --- weed/filer/filer.go | 26 ++++++++++ weed/filer/filer_delete_entry.go | 2 + weed/filer/leveldb/leveldb_store.go | 15 ++++-- weed/filer/leveldb2/leveldb2_store.go | 15 ++++-- weed/filer/leveldb3/leveldb3_store.go | 15 ++++-- weed/server/filer_server_handlers_write.go | 5 ++ weed/storage/volume_write.go | 56 +++++++++++++++++++--- 7 files changed, 119 insertions(+), 15 deletions(-) diff --git a/weed/filer/filer.go b/weed/filer/filer.go index 249ad4acd7..522df042a9 100644 --- a/weed/filer/filer.go +++ b/weed/filer/filer.go @@ -413,6 +413,28 @@ func (f *Filer) RollbackTransaction(ctx context.Context) error { return f.Store.RollbackTransaction(ctx) } +type syncWriteKey struct{} + +// WithSyncWrite marks the entry writes under ctx durable-on-ack: a store that +// buffers writes by default must flush them before returning. +func WithSyncWrite(ctx context.Context) context.Context { + return context.WithValue(ctx, syncWriteKey{}, true) +} + +// SyncWrite reports whether ctx carries a durability request for entry writes. +func SyncWrite(ctx context.Context) bool { + return ctx.Value(syncWriteKey{}) != nil +} + +// syncWriteContext flags ctx when the storage rule on p asks for fsync, so the +// store write is held to the same durability as the data path. +func (f *Filer) syncWriteContext(ctx context.Context, p util.FullPath) context.Context { + if f.FilerConf != nil && f.FilerConf.MatchStorageRule(string(p)).Fsync { + return WithSyncWrite(ctx) + } + return ctx +} + // CreateEntry creates or replaces an entry. When existing is non-nil the caller // has already fetched the current entry at this path under a path lock, and it // is reused instead of looking the store up again; pass nil to have CreateEntry @@ -423,6 +445,8 @@ func (f *Filer) CreateEntry(ctx context.Context, entry *Entry, existing *Entry, return nil } + ctx = f.syncWriteContext(ctx, entry.FullPath) + if entry.FullPath.IsLongerFileName(maxFilenameLength) { return filer_pb.ErrEntryNameTooLong } @@ -628,6 +652,7 @@ func (f *Filer) DirectoryAttributes(ctx context.Context, dirPath util.FullPath) // deleted with, so a restore cannot hand back a directory more permissive than the one // it replaces - including when someone else already put it back with wider ones. func (f *Filer) EnsureDirectoryEntry(ctx context.Context, dirPath util.FullPath, attrs empty_folder_cleanup.DirectoryAttributes) error { + ctx = f.syncWriteContext(ctx, dirPath) existing, err := f.FindEntry(ctx, dirPath) if err != nil && !errors.Is(err, filer_pb.ErrNotFound) { return err @@ -677,6 +702,7 @@ func (f *Filer) EnsureDirectoryEntry(ctx context.Context, dirPath util.FullPath, } func (f *Filer) UpdateEntry(ctx context.Context, oldEntry, entry *Entry, isFromOtherCluster bool) (err error) { + ctx = f.syncWriteContext(ctx, entry.FullPath) if oldEntry != nil { entry.Attr.Crtime = oldEntry.Attr.Crtime if oldEntry.Attr.Inode != 0 { diff --git a/weed/filer/filer_delete_entry.go b/weed/filer/filer_delete_entry.go index 6f7949a615..899d337c05 100644 --- a/weed/filer/filer_delete_entry.go +++ b/weed/filer/filer_delete_entry.go @@ -122,6 +122,8 @@ func (f *Filer) DeleteEntryMetaAndData(ctx context.Context, p util.FullPath, isR return nil } + ctx = f.syncWriteContext(ctx, p) + entry, findErr := f.FindEntry(ctx, p) if findErr != nil { return findErr diff --git a/weed/filer/leveldb/leveldb_store.go b/weed/filer/leveldb/leveldb_store.go index a6d699d8be..606166da90 100644 --- a/weed/filer/leveldb/leveldb_store.go +++ b/weed/filer/leveldb/leveldb_store.go @@ -79,6 +79,15 @@ func (store *LevelDBStore) RollbackTransaction(ctx context.Context) error { return nil } +var syncWriteOptions = &opt.WriteOptions{Sync: true} + +func writeOptions(ctx context.Context) *opt.WriteOptions { + if filer.SyncWrite(ctx) { + return syncWriteOptions + } + return nil +} + func (store *LevelDBStore) InsertEntry(ctx context.Context, entry *filer.Entry) (err error) { key := genKey(entry.DirAndName()) @@ -91,7 +100,7 @@ func (store *LevelDBStore) InsertEntry(ctx context.Context, entry *filer.Entry) value = weed_util.MaybeGzipData(value) } - err = store.db.Put(key, value, nil) + err = store.db.Put(key, value, writeOptions(ctx)) if err != nil { return fmt.Errorf("persisting %s : %v", entry.FullPath, err) @@ -168,7 +177,7 @@ func (store *LevelDBStore) FindEntry(ctx context.Context, fullpath weed_util.Ful func (store *LevelDBStore) DeleteEntry(ctx context.Context, fullpath weed_util.FullPath) (err error) { key := genKey(fullpath.DirAndName()) - err = store.db.Delete(key, nil) + err = store.db.Delete(key, writeOptions(ctx)) if err != nil { return fmt.Errorf("delete %s : %v", fullpath, err) } @@ -195,7 +204,7 @@ func (store *LevelDBStore) DeleteFolderChildren(ctx context.Context, fullpath we } iter.Release() - err = store.db.Write(batch, nil) + err = store.db.Write(batch, writeOptions(ctx)) if err != nil { return fmt.Errorf("delete %s : %v", fullpath, err) diff --git a/weed/filer/leveldb2/leveldb2_store.go b/weed/filer/leveldb2/leveldb2_store.go index 93c587d703..5c05033b99 100644 --- a/weed/filer/leveldb2/leveldb2_store.go +++ b/weed/filer/leveldb2/leveldb2_store.go @@ -127,6 +127,15 @@ func (store *LevelDB2Store) RollbackTransaction(ctx context.Context) error { return nil } +var syncWriteOptions = &opt.WriteOptions{Sync: true} + +func writeOptions(ctx context.Context) *opt.WriteOptions { + if filer.SyncWrite(ctx) { + return syncWriteOptions + } + return nil +} + func (store *LevelDB2Store) InsertEntry(ctx context.Context, entry *filer.Entry) (err error) { dir, name := entry.DirAndName() key, partitionId := genKey(dir, name, store.dbCount) @@ -140,7 +149,7 @@ func (store *LevelDB2Store) InsertEntry(ctx context.Context, entry *filer.Entry) value = weed_util.MaybeGzipData(value) } - err = store.dbs[partitionId].Put(key, value, nil) + err = store.dbs[partitionId].Put(key, value, writeOptions(ctx)) if err != nil { return fmt.Errorf("persisting %s : %v", entry.FullPath, err) @@ -186,7 +195,7 @@ func (store *LevelDB2Store) DeleteEntry(ctx context.Context, fullpath weed_util. dir, name := fullpath.DirAndName() key, partitionId := genKey(dir, name, store.dbCount) - err = store.dbs[partitionId].Delete(key, nil) + err = store.dbs[partitionId].Delete(key, writeOptions(ctx)) if err != nil { return fmt.Errorf("delete %s : %v", fullpath, err) } @@ -213,7 +222,7 @@ func (store *LevelDB2Store) DeleteFolderChildren(ctx context.Context, fullpath w } iter.Release() - err = store.dbs[partitionId].Write(batch, nil) + err = store.dbs[partitionId].Write(batch, writeOptions(ctx)) if err != nil { return fmt.Errorf("delete %s : %v", fullpath, err) diff --git a/weed/filer/leveldb3/leveldb3_store.go b/weed/filer/leveldb3/leveldb3_store.go index 5e012ba673..029d902715 100644 --- a/weed/filer/leveldb3/leveldb3_store.go +++ b/weed/filer/leveldb3/leveldb3_store.go @@ -221,6 +221,15 @@ func (store *LevelDB3Store) RollbackTransaction(ctx context.Context) error { return nil } +var syncWriteOptions = &opt.WriteOptions{Sync: true} + +func writeOptions(ctx context.Context) *opt.WriteOptions { + if filer.SyncWrite(ctx) { + return syncWriteOptions + } + return nil +} + func (store *LevelDB3Store) InsertEntry(ctx context.Context, entry *filer.Entry) (err error) { db, _, shortPath, err := store.findDB(entry.FullPath, false) @@ -240,7 +249,7 @@ func (store *LevelDB3Store) InsertEntry(ctx context.Context, entry *filer.Entry) value = weed_util.MaybeGzipData(value) } - err = db.Put(key, value, nil) + err = db.Put(key, value, writeOptions(ctx)) if err != nil { return fmt.Errorf("persisting %s : %v", entry.FullPath, err) @@ -298,7 +307,7 @@ func (store *LevelDB3Store) DeleteEntry(ctx context.Context, fullpath weed_util. dir, name := shortPath.DirAndName() key := genKey(dir, name) - err = db.Delete(key, nil) + err = db.Delete(key, writeOptions(ctx)) if err != nil { return fmt.Errorf("delete %s : %v", fullpath, err) } @@ -339,7 +348,7 @@ func (store *LevelDB3Store) DeleteFolderChildren(ctx context.Context, fullpath w } iter.Release() - err = db.Write(batch, nil) + err = db.Write(batch, writeOptions(ctx)) if err != nil { return fmt.Errorf("delete %s : %v", fullpath, err) diff --git a/weed/server/filer_server_handlers_write.go b/weed/server/filer_server_handlers_write.go index 0864409c3c..177093c501 100644 --- a/weed/server/filer_server_handlers_write.go +++ b/weed/server/filer_server_handlers_write.go @@ -11,6 +11,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/operation" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" @@ -122,6 +123,10 @@ func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request, conte return } + if so.Fsync { + ctx = filer.WithSyncWrite(ctx) + } + if util.FullPath(r.URL.Path).IsLongerFileName(so.MaxFileNameLength) { glog.V(1).InfolnCtx(ctx, "post", r.RequestURI, ": ", "entry name too long") w.WriteHeader(http.StatusRequestURITooLong) diff --git a/weed/storage/volume_write.go b/weed/storage/volume_write.go index 443c7bc814..a04fa746cc 100644 --- a/weed/storage/volume_write.go +++ b/weed/storage/volume_write.go @@ -235,6 +235,33 @@ func (v *Volume) syncWrite(n *needle.Needle, checkCookie bool, fsync bool) (offs if !isUnchanged { err = v.publishWriteRequest(n, offset, fsync) } + // The .idx row goes down before the write is acknowledged, same as the + // .dat record above; the access lock is dropped for the fsync the same + // way, since the row is already published. + v.dataFileAccessLock.Unlock() + if idxErr := v.syncIndex(); err == nil { + err = idxErr + } + v.dataFileAccessLock.Lock() + return +} + +// syncIndex flushes the .idx, the second half of a durable write: the row is +// already published and cannot be taken back, so a failed sync stops the +// volume taking writes rather than acknowledge an index entry that may not +// survive — the same split the Rust volume server makes between flush_dat +// and flush_idx. +func (v *Volume) syncIndex() (err error) { + if v.nm == nil { + return nil + } + if err = v.nm.Sync(); err != nil { + v.checkReadWriteError(err) + v.noWriteLock.Lock() + v.noWriteOrDelete = true + v.noWriteLock.Unlock() + err = fmt.Errorf("sync volume %d index: %w", v.Id, err) + } return } @@ -503,6 +530,7 @@ func (v *Volume) processBatchRun(run []*needle.AsyncRequest) { stagedWriteOk := make([]bool, len(run)) stagedDeleteOffsets := make([]uint64, len(run)) stagedDeleteOk := make([]bool, len(run)) + needIndexSync := false for i := 0; i < len(run); i++ { // Every queued request meets the same refusal it would get sent on // its own: once a durable index update fails the volume stops taking @@ -514,9 +542,12 @@ func (v *Volume) processBatchRun(run []*needle.AsyncRequest) { run[i].UpdateResult(0, 0, false, fmt.Errorf("volume %d is read only", v.Id)) } else { offset, size, isUnchanged, err := v.appendWriteRequest(run[i].N, true, true) - if err == nil && !isUnchanged { - stagedWriteOffsets[i] = offset - stagedWriteOk[i] = true + if err == nil { + needIndexSync = true + if !isUnchanged { + stagedWriteOffsets[i] = offset + stagedWriteOk[i] = true + } } run[i].UpdateResult(offset, uint64(size), isUnchanged, err) } @@ -527,9 +558,12 @@ func (v *Volume) processBatchRun(run []*needle.AsyncRequest) { run[i].UpdateResult(0, 0, false, fmt.Errorf("volume %d is read only", v.Id)) } else { size, offset, tombstoned, err := v.appendDeleteRequest(run[i].N) - if err == nil && tombstoned { - stagedDeleteOffsets[i] = offset - stagedDeleteOk[i] = true + if err == nil { + needIndexSync = true + if tombstoned { + stagedDeleteOffsets[i] = offset + stagedDeleteOk[i] = true + } } run[i].UpdateResult(0, uint64(size), false, err) } @@ -588,6 +622,16 @@ func (v *Volume) processBatchRun(run []*needle.AsyncRequest) { } } v.dataFileAccessLock.Unlock() + // One .idx sync per run, paired with the .dat sync above: the published + // rows cannot be rolled back, so a failed sync fails every request in the + // run instead of acknowledging an index entry that may not survive. + if needIndexSync { + if idxErr := v.syncIndex(); idxErr != nil { + for i := 0; i < len(run); i++ { + run[i].UpdateResult(0, 0, false, idxErr) + } + } + } for i := 0; i < len(run); i++ { run[i].Submit() }