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() }