filer: reopen a leveldb store wedged by a transient write error (#12565)

* filer/leveldb_handle: reopen a database whose journal latched a write error

goleveldb keeps the first journal write error and returns it for every
later write until the database is reopened, so one transient ENOSPC or
EIO fails all writes until a restart. The handle offers the leveldb.DB
methods the stores use; a write that fails on an OS error probes the
storage and, once it accepts writes and no call holds the database,
closes and reopens it and retries the write once.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* filer: open the leveldb stores through leveldb_handle

leveldb, leveldb2 and leveldb3 now recover from a transient storage
write error without a restart.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* filer: leveldb stores return iterator errors from listings and deletes

A listing on a database that cannot be read reported an empty
directory, and DeleteFolderChildren wrote an empty batch, because the
iterator error was never checked.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* filer/leveldb_handle: test a reopen that fails after the probe passes

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* filer/leveldb_handle: probe storage before draining readers

A probe that stalls in Create, Write or Sync held mu exclusively and
blocked every read on the handle. Reopen attempts are now serialized by
their own mutex, and the probe runs before the drain.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* filer/leveldb_handle: hold new calls back while a reopen drains

TryLock succeeds only at an instant with no reader, so overlapping
listings could keep a reopen from ever draining while each write spun
for the full drain timeout. The drain now holds new calls back for at
most drainTimeout; a call nested inside a running one waits that out
instead of deadlocking. A drain that times out holds off the next
attempt for busyReopenInterval, so new calls are not held back over and
over.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* filer/leveldb_handle: bound the wait for a running reopen

A write that hit the latched error waited on the reopen mutex for as long
as another caller's storage probe ran, so a stalled probe queued every
writer. Callers now wait at most reopenWait and return their error, and
still retry once a concurrent reopen succeeds.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* filer/leveldb_handle: tests wait for the drain and readers explicitly

The nested-read test polled for a drain that could start and end between
polls and then spun forever; it now syncs on the probe, widens the drain
and fails if the put returns first. The overlapping-readers test waits
until every reader holds an iterator before it writes.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

weed/filer/leveldb/leveldb_store.go

@@ -8,12 +8,12 @@ import (
 	"os"
 
 	"github.com/syndtr/goleveldb/leveldb"
-	leveldb_errors "github.com/syndtr/goleveldb/leveldb/errors"
 	"github.com/syndtr/goleveldb/leveldb/filter"
 	"github.com/syndtr/goleveldb/leveldb/opt"
 	leveldb_util "github.com/syndtr/goleveldb/leveldb/util"
 
 	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/filer/leveldb_handle"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	weed_util "github.com/seaweedfs/seaweedfs/weed/util"
@@ -32,7 +32,7 @@ func init() {
 }
 
 type LevelDBStore struct {
-	db *leveldb.DB
+	db *leveldb_handle.DB
 }
 
 func (store *LevelDBStore) GetName() string {
@@ -57,14 +57,8 @@ func (store *LevelDBStore) initialize(dir string) (err error) {
 		Filter:             filter.NewBloomFilter(8), // false positive rate 0.02
 	}
 
-	if store.db, err = leveldb.OpenFile(dir, opts); err != nil {
-		if leveldb_errors.IsCorrupted(err) {
-			store.db, err = leveldb.RecoverFile(dir, opts)
-		}
-		if err != nil {
-			glog.Infof("filer store open dir %s: %v", dir, err)
-			return
-		}
+	if store.db, err = leveldb_handle.OpenFile(dir, opts); err != nil {
+		glog.Infof("filer store open dir %s: %v", dir, err)
 	}
 	return
 }
@@ -202,7 +196,11 @@ func (store *LevelDBStore) DeleteFolderChildren(ctx context.Context, fullpath we
 		}
 		batch.Delete([]byte(genKey(string(fullpath), fileName)))
 	}
+	iterErr := iter.Error()
 	iter.Release()
+	if iterErr != nil {
+		return fmt.Errorf("delete %s : %v", fullpath, iterErr)
+	}
 
 	err = store.db.Write(batch, writeOptions(ctx))
 
@@ -262,6 +260,9 @@ func (store *LevelDBStore) ListDirectoryPrefixedEntries(ctx context.Context, dir
 			break
 		}
 	}
+	if err == nil {
+		err = iter.Error()
+	}
 	iter.Release()
 
 	return lastFileName, err

weed/filer/leveldb/leveldb_store_test.go

@@ -135,3 +135,28 @@ func TestListDirectoryPrefixedEntries_StartBelowPrefix(t *testing.T) {
 		t.Fatalf("want [item-b.versions], got %v", names)
 	}
 }
+
+func TestUnavailableDatabaseFailsListAndDelete(t *testing.T) {
+	dir := t.TempDir()
+	store := &LevelDBStore{}
+	if err := store.initialize(dir); err != nil {
+		t.Fatal(err)
+	}
+	ctx := context.Background()
+	if err := store.InsertEntry(ctx, &filer.Entry{FullPath: "/dir/file"}); err != nil {
+		t.Fatal(err)
+	}
+	store.db.Close()
+
+	listed := 0
+	_, err := store.ListDirectoryEntries(ctx, "/dir", "", true, 10, func(*filer.Entry) (bool, error) {
+		listed++
+		return true, nil
+	})
+	if err == nil || listed != 0 {
+		t.Fatalf("list on a closed database: %d entries, err %v", listed, err)
+	}
+	if err := store.DeleteFolderChildren(ctx, "/dir"); err == nil {
+		t.Fatal("delete children on a closed database succeeded")
+	}
+}

weed/filer/leveldb2/leveldb2_store.go

@@ -10,12 +10,12 @@ import (
 	"os"
 
 	"github.com/syndtr/goleveldb/leveldb"
-	leveldb_errors "github.com/syndtr/goleveldb/leveldb/errors"
 	"github.com/syndtr/goleveldb/leveldb/filter"
 	"github.com/syndtr/goleveldb/leveldb/opt"
 	leveldb_util "github.com/syndtr/goleveldb/leveldb/util"
 
 	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/filer/leveldb_handle"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	weed_util "github.com/seaweedfs/seaweedfs/weed/util"
@@ -26,7 +26,7 @@ func init() {
 }
 
 type LevelDB2Store struct {
-	dbs      []*leveldb.DB
+	dbs      []*leveldb_handle.DB
 	dbCount  int
 	ReadOnly bool
 
@@ -108,10 +108,7 @@ func (store *LevelDB2Store) initialize(dir string, dbCount int) (err error) {
 	for d := 0; d < dbCount; d++ {
 		dbFolder := fmt.Sprintf("%s/%02d", dir, d)
 		os.MkdirAll(dbFolder, 0755)
-		db, dbErr := leveldb.OpenFile(dbFolder, opts)
-		if leveldb_errors.IsCorrupted(dbErr) {
-			db, dbErr = leveldb.RecoverFile(dbFolder, opts)
-		}
+		db, dbErr := leveldb_handle.OpenFile(dbFolder, opts)
 		if dbErr != nil {
 			glog.Errorf("filer store open dir %s: %v", dbFolder, dbErr)
 			return dbErr
@@ -226,7 +223,11 @@ func (store *LevelDB2Store) DeleteFolderChildren(ctx context.Context, fullpath w
 		}
 		batch.Delete(append(directoryPrefix, []byte(fileName)...))
 	}
+	iterErr := iter.Error()
 	iter.Release()
+	if iterErr != nil {
+		return fmt.Errorf("delete %s : %v", fullpath, iterErr)
+	}
 
 	err = store.dbs[partitionId].Write(batch, writeOptions(ctx))
 
@@ -288,6 +289,9 @@ func (store *LevelDB2Store) ListDirectoryPrefixedEntries(ctx context.Context, di
 			break
 		}
 	}
+	if err == nil {
+		err = iter.Error()
+	}
 	iter.Release()
 
 	return lastFileName, err

weed/filer/leveldb2/leveldb2_store_test.go

@@ -197,3 +197,30 @@ func TestEmptyRoot(t *testing.T) {
 	}
 
 }
+
+func TestUnavailableDatabaseFailsListAndDelete(t *testing.T) {
+	dir := t.TempDir()
+	store := &LevelDB2Store{}
+	if err := store.initialize(dir, 2); err != nil {
+		t.Fatal(err)
+	}
+	ctx := context.Background()
+	if err := store.InsertEntry(ctx, &filer.Entry{FullPath: "/dir/file"}); err != nil {
+		t.Fatal(err)
+	}
+	for _, db := range store.dbs {
+		db.Close()
+	}
+
+	listed := 0
+	_, err := store.ListDirectoryEntries(ctx, "/dir", "", true, 10, func(*filer.Entry) (bool, error) {
+		listed++
+		return true, nil
+	})
+	if err == nil || listed != 0 {
+		t.Fatalf("list on a closed database: %d entries, err %v", listed, err)
+	}
+	if err := store.DeleteFolderChildren(ctx, "/dir"); err == nil {
+		t.Fatal("delete children on a closed database succeeded")
+	}
+}

weed/filer/leveldb3/leveldb3_store.go

@@ -13,13 +13,13 @@ import (
 	"sync"
 
 	"github.com/syndtr/goleveldb/leveldb"
-	leveldb_errors "github.com/syndtr/goleveldb/leveldb/errors"
 	"github.com/syndtr/goleveldb/leveldb/filter"
 	"github.com/syndtr/goleveldb/leveldb/opt"
 	"github.com/syndtr/goleveldb/leveldb/storage"
 	leveldb_util "github.com/syndtr/goleveldb/leveldb/util"
 
 	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/filer/leveldb_handle"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	weed_util "github.com/seaweedfs/seaweedfs/weed/util"
@@ -35,10 +35,10 @@ func init() {
 
 type LevelDB3Store struct {
 	dir      string
-	dbs      map[string]*leveldb.DB
+	dbs      map[string]*leveldb_handle.DB
 	dbsLock  sync.RWMutex
 	ReadOnly bool
-	emptyDB  *leveldb.DB
+	emptyDB  *leveldb_handle.DB
 
 	blockCacheSizeMB      int
 	writeBufferSizeMB     int
@@ -122,21 +122,18 @@ func (store *LevelDB3Store) initialize(dir string) (err error) {
 	if loadDbErr != nil {
 		return loadDbErr
 	}
-	store.dbs = make(map[string]*leveldb.DB)
+	store.dbs = make(map[string]*leveldb_handle.DB)
 	store.dbs[DEFAULT] = db
 
 	return
 }
 
-func (store *LevelDB3Store) loadDB(name string) (*leveldb.DB, error) {
+func (store *LevelDB3Store) loadDB(name string) (*leveldb_handle.DB, error) {
 	opts := store.buildOptions(name)
 
 	dbFolder := fmt.Sprintf("%s/%s", store.dir, name)
 	os.MkdirAll(dbFolder, 0755)
-	db, dbErr := leveldb.OpenFile(dbFolder, opts)
-	if leveldb_errors.IsCorrupted(dbErr) {
-		db, dbErr = leveldb.RecoverFile(dbFolder, opts)
-	}
+	db, dbErr := leveldb_handle.OpenFile(dbFolder, opts)
 	if dbErr != nil {
 		glog.Errorf("filer store open dir %s: %v", dbFolder, dbErr)
 		return nil, dbErr
@@ -144,7 +141,7 @@ func (store *LevelDB3Store) loadDB(name string) (*leveldb.DB, error) {
 	return db, nil
 }
 
-func (store *LevelDB3Store) findDB(fullpath weed_util.FullPath, isForChildren bool) (*leveldb.DB, string, weed_util.FullPath, error) {
+func (store *LevelDB3Store) findDB(fullpath weed_util.FullPath, isForChildren bool) (*leveldb_handle.DB, string, weed_util.FullPath, error) {
 
 	store.dbsLock.RLock()
 
@@ -193,11 +190,11 @@ func (store *LevelDB3Store) findDB(fullpath weed_util.FullPath, isForChildren bo
 	return db, bucket, shortPath, err
 }
 
-func (store *LevelDB3Store) emptyBucketDB() (*leveldb.DB, error) {
+func (store *LevelDB3Store) emptyBucketDB() (*leveldb_handle.DB, error) {
 	store.dbsLock.Lock()
 	defer store.dbsLock.Unlock()
 	if store.emptyDB == nil {
-		db, err := leveldb.Open(storage.NewMemStorage(), nil)
+		db, err := leveldb_handle.Open(storage.NewMemStorage(), nil)
 		if err != nil {
 			return nil, err
 		}
@@ -214,7 +211,7 @@ func isBucketFolder(bucket string) bool {
 	return !strings.HasPrefix(bucket, ".") && filepath.IsLocal(bucket) && !strings.ContainsAny(bucket, "/"+string(os.PathSeparator))
 }
 
-func (store *LevelDB3Store) createDB(bucket string) (*leveldb.DB, error) {
+func (store *LevelDB3Store) createDB(bucket string) (*leveldb_handle.DB, error) {
 
 	store.dbsLock.Lock()
 	defer store.dbsLock.Unlock()
@@ -391,7 +388,11 @@ func (store *LevelDB3Store) DeleteFolderChildren(ctx context.Context, fullpath w
 		}
 		batch.Delete(append(directoryPrefix, []byte(fileName)...))
 	}
+	iterErr := iter.Error()
 	iter.Release()
+	if iterErr != nil {
+		return fmt.Errorf("delete %s : %v", fullpath, iterErr)
+	}
 
 	err = db.Write(batch, writeOptions(ctx))
 
@@ -458,6 +459,9 @@ func (store *LevelDB3Store) ListDirectoryPrefixedEntries(ctx context.Context, di
 			break
 		}
 	}
+	if err == nil {
+		err = iter.Error()
+	}
 	iter.Release()
 
 	return lastFileName, err

weed/filer/leveldb3/leveldb3_store_test.go

@@ -250,3 +250,28 @@ func TestBucketEventsStayInsideStoreDir(t *testing.T) {
 		t.Fatalf("default db unusable after bucket events: %v", err)
 	}
 }
+
+func TestUnavailableDatabaseFailsListAndDelete(t *testing.T) {
+	dir := t.TempDir()
+	store := &LevelDB3Store{}
+	if err := store.initialize(dir); err != nil {
+		t.Fatal(err)
+	}
+	ctx := context.Background()
+	if err := store.InsertEntry(ctx, &filer.Entry{FullPath: "/dir/file"}); err != nil {
+		t.Fatal(err)
+	}
+	store.dbs[DEFAULT].Close()
+
+	listed := 0
+	_, err := store.ListDirectoryEntries(ctx, "/dir", "", true, 10, func(*filer.Entry) (bool, error) {
+		listed++
+		return true, nil
+	})
+	if err == nil || listed != 0 {
+		t.Fatalf("list on a closed database: %d entries, err %v", listed, err)
+	}
+	if err := store.DeleteFolderChildren(ctx, "/dir"); err == nil {
+		t.Fatal("delete children on a closed database succeeded")
+	}
+}

weed/filer/leveldb_handle/leveldb_handle.go

@@ -0,0 +1,272 @@
+// Package leveldb_handle wraps a goleveldb database so a write that fails on
+// a storage error reopens it. goleveldb latches the first journal write error
+// and returns it for every later write until the database is reopened, so one
+// transient ENOSPC or EIO would otherwise fail writes until a restart.
+package leveldb_handle
+
+import (
+	"errors"
+	"sync"
+	"sync/atomic"
+	"syscall"
+	"time"
+
+	"github.com/syndtr/goleveldb/leveldb"
+	leveldb_errors "github.com/syndtr/goleveldb/leveldb/errors"
+	"github.com/syndtr/goleveldb/leveldb/iterator"
+	"github.com/syndtr/goleveldb/leveldb/opt"
+	"github.com/syndtr/goleveldb/leveldb/storage"
+	leveldb_util "github.com/syndtr/goleveldb/leveldb/util"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+)
+
+var (
+	reopenInterval     = time.Second
+	busyReopenInterval = 10 * time.Second
+	drainTimeout       = time.Second
+	reopenWait         = 5 * time.Second
+	probeFile          = storage.FileDesc{Type: storage.TypeTemp, Num: 1 << 40}
+	probeData          = make([]byte, 64*1024)
+)
+
+// DB offers the leveldb.DB methods the filer stores use. Every call holds mu
+// shared, an iterator until it is released. A reopen stops admitting calls
+// for at most drainTimeout and takes mu with TryLock only, so a call that
+// nests another one delays the reopen instead of deadlocking.
+type DB struct {
+	name        string
+	stor        storage.Storage
+	ownsStorage bool
+	opts        *opt.Options
+	closed      atomic.Bool
+	nextReopen  atomic.Int64
+	reopening   chan struct{}
+	draining    atomic.Pointer[chan struct{}]
+
+	mu      sync.RWMutex
+	db      *leveldb.DB
+	openErr error
+}
+
+func OpenFile(path string, o *opt.Options) (*DB, error) {
+	stor, err := storage.OpenFile(path, o.GetReadOnly())
+	if err != nil {
+		return nil, err
+	}
+	d, err := Open(stor, o)
+	if err != nil {
+		stor.Close()
+		return nil, err
+	}
+	d.name = path
+	d.ownsStorage = true
+	return d, nil
+}
+
+func Open(stor storage.Storage, o *opt.Options) (*DB, error) {
+	db, err := open(stor, o)
+	if err != nil {
+		return nil, err
+	}
+	return &DB{stor: stor, opts: o, db: db, reopening: make(chan struct{}, 1)}, nil
+}
+
+func open(stor storage.Storage, o *opt.Options) (*leveldb.DB, error) {
+	db, err := leveldb.Open(stor, o)
+	if leveldb_errors.IsCorrupted(err) {
+		db, err = leveldb.Recover(stor, o)
+	}
+	return db, err
+}
+
+func (d *DB) Get(key []byte, ro *opt.ReadOptions) ([]byte, error) {
+	db, err := d.acquire()
+	if err != nil {
+		return nil, err
+	}
+	defer d.mu.RUnlock()
+	return db.Get(key, ro)
+}
+
+func (d *DB) NewIterator(slice *leveldb_util.Range, ro *opt.ReadOptions) iterator.Iterator {
+	db, err := d.acquire()
+	if err != nil {
+		return iterator.NewEmptyIterator(err)
+	}
+	return &heldIterator{Iterator: db.NewIterator(slice, ro), unlock: d.mu.RUnlock}
+}
+
+func (d *DB) Put(key, value []byte, wo *opt.WriteOptions) error {
+	return d.write(func(db *leveldb.DB) error { return db.Put(key, value, wo) })
+}
+
+func (d *DB) Delete(key []byte, wo *opt.WriteOptions) error {
+	return d.write(func(db *leveldb.DB) error { return db.Delete(key, wo) })
+}
+
+func (d *DB) Write(batch *leveldb.Batch, wo *opt.WriteOptions) error {
+	return d.write(func(db *leveldb.DB) error { return db.Write(batch, wo) })
+}
+
+// Close does not wait for in-flight calls, matching leveldb.DB.Close.
+func (d *DB) Close() error {
+	if d.closed.Swap(true) {
+		return leveldb.ErrClosed
+	}
+	d.mu.RLock()
+	db := d.db
+	d.mu.RUnlock()
+	var err error
+	if db != nil {
+		err = db.Close()
+	}
+	if d.ownsStorage {
+		if closeErr := d.stor.Close(); err == nil {
+			err = closeErr
+		}
+	}
+	return err
+}
+
+func (d *DB) write(fn func(*leveldb.DB) error) error {
+	db, err := d.acquire()
+	if err != nil {
+		return err
+	}
+	err = fn(db)
+	d.mu.RUnlock()
+	if !isStorageError(err) || !d.reopen(db) {
+		return err
+	}
+	if db, err = d.acquire(); err != nil {
+		return err
+	}
+	defer d.mu.RUnlock()
+	return fn(db)
+}
+
+// acquire returns the open database with mu held shared.
+func (d *DB) acquire() (*leveldb.DB, error) {
+	for attempt := 0; ; attempt++ {
+		if draining := d.draining.Load(); draining != nil {
+			<-*draining
+		}
+		d.mu.RLock()
+		if d.closed.Load() {
+			d.mu.RUnlock()
+			return nil, leveldb.ErrClosed
+		}
+		if d.db != nil {
+			return d.db, nil
+		}
+		err := d.openErr
+		d.mu.RUnlock()
+		if attempt > 0 || !d.reopen(nil) {
+			return nil, err
+		}
+	}
+}
+
+// reopen replaces failed with a freshly opened database once no call holds
+// it, unless another caller already did. It waits at most reopenWait for a
+// running reopen and reports whether a retry can use an open database.
+func (d *DB) reopen(failed *leveldb.DB) bool {
+	select {
+	case d.reopening <- struct{}{}:
+	case <-time.After(reopenWait):
+		return false
+	}
+	defer func() { <-d.reopening }()
+	d.mu.RLock()
+	current := d.db
+	d.mu.RUnlock()
+	if current != failed {
+		return current != nil
+	}
+	if d.closed.Load() || time.Now().UnixNano() < d.nextReopen.Load() {
+		return false
+	}
+	d.holdOffReopen(reopenInterval)
+	// Reopening on storage that still fails would also lose reads.
+	if err := probe(d.stor); err != nil {
+		glog.Warningf("leveldb %s: storage still failing, not reopening: %v", d.name, err)
+		return false
+	}
+	if !d.lockWhenIdle() {
+		glog.Warningf("leveldb %s: calls still running after %v, not reopening", d.name, drainTimeout)
+		d.holdOffReopen(busyReopenInterval)
+		return false
+	}
+	defer d.mu.Unlock()
+	if d.closed.Load() {
+		return false
+	}
+	glog.Warningf("leveldb %s: reopening after a storage write error", d.name)
+	if failed != nil {
+		failed.Close()
+	}
+	d.db, d.openErr = open(d.stor, d.opts)
+	if d.openErr != nil {
+		glog.Errorf("leveldb %s: reopen: %v", d.name, d.openErr)
+		return false
+	}
+	return true
+}
+
+func (d *DB) holdOffReopen(interval time.Duration) {
+	d.nextReopen.Store(time.Now().Add(interval).UnixNano())
+}
+
+// lockWhenIdle holds new calls back while the running ones finish, then
+// takes mu, giving up after drainTimeout.
+func (d *DB) lockWhenIdle() bool {
+	draining := make(chan struct{})
+	d.draining.Store(&draining)
+	defer func() {
+		d.draining.Store(nil)
+		close(draining)
+	}()
+	deadline := time.Now().Add(drainTimeout)
+	for !d.mu.TryLock() {
+		if time.Now().After(deadline) {
+			return false
+		}
+		time.Sleep(time.Millisecond)
+	}
+	return true
+}
+
+func probe(stor storage.Storage) error {
+	w, err := stor.Create(probeFile)
+	if err != nil {
+		return err
+	}
+	_, err = w.Write(probeData)
+	if err == nil {
+		err = w.Sync()
+	}
+	if closeErr := w.Close(); err == nil {
+		err = closeErr
+	}
+	stor.Remove(probeFile)
+	return err
+}
+
+func isStorageError(err error) bool {
+	var errno syscall.Errno
+	return errors.As(err, &errno)
+}
+
+type heldIterator struct {
+	iterator.Iterator
+	unlock func()
+}
+
+func (it *heldIterator) Release() {
+	it.Iterator.Release()
+	if it.unlock != nil {
+		it.unlock()
+		it.unlock = nil
+	}
+}

weed/filer/leveldb_handle/leveldb_handle_test.go

@@ -0,0 +1,360 @@
+package leveldb_handle
+
+import (
+	"errors"
+	"sync"
+	"sync/atomic"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/syndtr/goleveldb/leveldb"
+	"github.com/syndtr/goleveldb/leveldb/storage"
+)
+
+type failingStorage struct {
+	storage.Storage
+	fail         atomic.Bool
+	failOpen     atomic.Bool
+	probeStarted chan struct{}
+	probeRelease chan struct{}
+}
+
+func (f *failingStorage) Create(fd storage.FileDesc) (storage.Writer, error) {
+	if f.failOpen.Load() && fd.Type != storage.TypeTemp {
+		return nil, syscall.ENOSPC
+	}
+	if fd.Type == storage.TypeTemp && f.probeRelease != nil {
+		f.probeStarted <- struct{}{}
+		<-f.probeRelease
+	}
+	w, err := f.Storage.Create(fd)
+	if err != nil {
+		return nil, err
+	}
+	return &failingWriter{Writer: w, s: f}, nil
+}
+
+type failingWriter struct {
+	storage.Writer
+	s *failingStorage
+}
+
+func (w *failingWriter) Write(p []byte) (int, error) {
+	if w.s.fail.Load() {
+		return 0, syscall.ENOSPC
+	}
+	return w.Writer.Write(p)
+}
+
+func openFailing(t *testing.T) (*DB, *failingStorage) {
+	t.Helper()
+	interval, busy, timeout, wait := reopenInterval, busyReopenInterval, drainTimeout, reopenWait
+	reopenInterval, busyReopenInterval, drainTimeout, reopenWait = 0, 0, 50*time.Millisecond, 50*time.Millisecond
+	t.Cleanup(func() {
+		reopenInterval, busyReopenInterval, drainTimeout, reopenWait = interval, busy, timeout, wait
+	})
+	st := &failingStorage{Storage: storage.NewMemStorage()}
+	db, err := Open(st, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { db.Close() })
+	return db, st
+}
+
+func expectValue(t *testing.T, db *DB, key, want string) {
+	t.Helper()
+	got, err := db.Get([]byte(key), nil)
+	if err != nil || string(got) != want {
+		t.Fatalf("get %s = %q, %v; want %q", key, got, err, want)
+	}
+}
+
+func TestWriteRecoversOnceStorageRecovers(t *testing.T) {
+	db, st := openFailing(t)
+	if err := db.Put([]byte("a"), []byte("1"), nil); err != nil {
+		t.Fatal(err)
+	}
+
+	st.fail.Store(true)
+	if err := db.Put([]byte("b"), []byte("2"), nil); !errors.Is(err, syscall.ENOSPC) {
+		t.Fatalf("put while storage fails: %v", err)
+	}
+	expectValue(t, db, "a", "1")
+
+	st.fail.Store(false)
+	if err := db.Put([]byte("c"), []byte("3"), nil); err != nil {
+		t.Fatalf("put after storage recovered: %v", err)
+	}
+	batch := new(leveldb.Batch)
+	batch.Delete([]byte("c"))
+	if err := db.Write(batch, nil); err != nil {
+		t.Fatalf("batch after storage recovered: %v", err)
+	}
+	expectValue(t, db, "a", "1")
+	if _, err := db.Get([]byte("c"), nil); err != leveldb.ErrNotFound {
+		t.Fatalf("get deleted key: %v", err)
+	}
+}
+
+func TestStalledProbeKeepsReadsAvailable(t *testing.T) {
+	db, st := openFailing(t)
+	if err := db.Put([]byte("a"), []byte("1"), nil); err != nil {
+		t.Fatal(err)
+	}
+	st.probeStarted, st.probeRelease = make(chan struct{}, 1), make(chan struct{})
+	st.fail.Store(true)
+	put := make(chan error, 1)
+	go func() { put <- db.Put([]byte("b"), []byte("2"), nil) }()
+	<-st.probeStarted
+
+	read := make(chan struct{})
+	go func() {
+		defer close(read)
+		expectValue(t, db, "a", "1")
+		db.NewIterator(nil, nil).Release()
+	}()
+	select {
+	case <-read:
+	case <-time.After(5 * time.Second):
+		t.Error("reads blocked behind a stalled probe")
+	}
+	close(st.probeRelease)
+	if err := <-put; !errors.Is(err, syscall.ENOSPC) {
+		t.Fatalf("put while storage fails: %v", err)
+	}
+	<-read
+}
+
+func TestWritesDoNotQueueBehindAStalledProbe(t *testing.T) {
+	db, st := openFailing(t)
+	if err := db.Put([]byte("a"), []byte("1"), nil); err != nil {
+		t.Fatal(err)
+	}
+	st.probeStarted, st.probeRelease = make(chan struct{}, 1), make(chan struct{})
+	st.fail.Store(true)
+	first := make(chan error, 1)
+	go func() { first <- db.Put([]byte("b"), []byte("2"), nil) }()
+	<-st.probeStarted
+
+	second := make(chan error, 1)
+	go func() { second <- db.Put([]byte("c"), []byte("3"), nil) }()
+	select {
+	case err := <-second:
+		if !errors.Is(err, syscall.ENOSPC) {
+			t.Errorf("put behind a stalled probe: %v", err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Error("put queued behind a stalled probe")
+	}
+	close(st.probeRelease)
+	if err := <-first; !errors.Is(err, syscall.ENOSPC) {
+		t.Fatalf("put while storage fails: %v", err)
+	}
+}
+
+func TestReopenWaitsForOpenIterator(t *testing.T) {
+	db, st := openFailing(t)
+	for _, k := range []string{"a", "b"} {
+		if err := db.Put([]byte(k), []byte(k), nil); err != nil {
+			t.Fatal(err)
+		}
+	}
+	st.fail.Store(true)
+	if err := db.Put([]byte("x"), []byte("x"), nil); err == nil {
+		t.Fatal("put while storage fails succeeded")
+	}
+	st.fail.Store(false)
+	busyReopenInterval = time.Hour
+
+	iter := db.NewIterator(nil, nil)
+	if !iter.Next() || string(iter.Key()) != "a" {
+		t.Fatalf("first key %q", iter.Key())
+	}
+	if err := db.Put([]byte("y"), []byte("y"), nil); !errors.Is(err, syscall.ENOSPC) {
+		t.Fatalf("put with an open iterator: %v", err)
+	}
+	if !iter.Next() || string(iter.Key()) != "b" || iter.Next() {
+		t.Fatalf("iterator after a refused reopen: key %q, err %v", iter.Key(), iter.Error())
+	}
+	iter.Release()
+	iter.Release()
+
+	if err := db.Put([]byte("y"), []byte("y"), nil); !errors.Is(err, syscall.ENOSPC) {
+		t.Fatalf("put right after a refused reopen: %v", err)
+	}
+	db.nextReopen.Store(0)
+	if err := db.Put([]byte("y"), []byte("y"), nil); err != nil {
+		t.Fatalf("put after the iterator was released: %v", err)
+	}
+	expectValue(t, db, "b", "b")
+}
+
+func TestNestedReadWaitsOutTheDrain(t *testing.T) {
+	db, st := openFailing(t)
+	drainTimeout = 500 * time.Millisecond
+	if err := db.Put([]byte("a"), []byte("1"), nil); err != nil {
+		t.Fatal(err)
+	}
+	st.fail.Store(true)
+	if err := db.Put([]byte("x"), []byte("x"), nil); err == nil {
+		t.Fatal("put while storage fails succeeded")
+	}
+	st.fail.Store(false)
+	st.probeStarted, st.probeRelease = make(chan struct{}, 1), make(chan struct{})
+
+	iter := db.NewIterator(nil, nil)
+	put := make(chan error, 1)
+	go func() { put <- db.Put([]byte("y"), []byte("y"), nil) }()
+	<-st.probeStarted
+	close(st.probeRelease)
+	for db.draining.Load() == nil {
+		select {
+		case err := <-put:
+			t.Fatalf("put returned before its drain was seen: %v", err)
+		case <-time.After(time.Millisecond):
+		}
+	}
+	expectValue(t, db, "a", "1")
+	iter.Release()
+	if err := <-put; !errors.Is(err, syscall.ENOSPC) {
+		t.Fatalf("put with an open iterator: %v", err)
+	}
+}
+
+func TestReopenIsNotStarvedByOverlappingReaders(t *testing.T) {
+	db, st := openFailing(t)
+	drainTimeout = 5 * time.Second
+	if err := db.Put([]byte("a"), []byte("1"), nil); err != nil {
+		t.Fatal(err)
+	}
+	st.fail.Store(true)
+	if err := db.Put([]byte("x"), []byte("x"), nil); err == nil {
+		t.Fatal("put while storage fails succeeded")
+	}
+	st.fail.Store(false)
+
+	stop := make(chan struct{})
+	var readers, started sync.WaitGroup
+	for i := 0; i < 4; i++ {
+		readers.Add(1)
+		started.Add(1)
+		go func() {
+			defer readers.Done()
+			for first := true; ; first = false {
+				select {
+				case <-stop:
+					return
+				default:
+				}
+				iter := db.NewIterator(nil, nil)
+				if first {
+					started.Done()
+				}
+				time.Sleep(2 * time.Millisecond)
+				iter.Release()
+			}
+		}()
+		time.Sleep(500 * time.Microsecond)
+	}
+	started.Wait()
+	err := db.Put([]byte("y"), []byte("y"), nil)
+	close(stop)
+	readers.Wait()
+	if err != nil {
+		t.Fatalf("put while readers overlap: %v", err)
+	}
+	expectValue(t, db, "y", "y")
+}
+
+func TestCloseIsNotUndoneByReopen(t *testing.T) {
+	db, st := openFailing(t)
+	st.fail.Store(true)
+	if err := db.Put([]byte("a"), []byte("1"), nil); err == nil {
+		t.Fatal("put while storage fails succeeded")
+	}
+	st.fail.Store(false)
+	db.Close()
+	if err := db.Put([]byte("a"), []byte("1"), nil); err != leveldb.ErrClosed {
+		t.Fatalf("put after close: %v", err)
+	}
+	if iter := db.NewIterator(nil, nil); iter.Next() || iter.Error() != leveldb.ErrClosed {
+		t.Fatalf("iterator after close: %v", iter.Error())
+	}
+}
+
+func TestReopenKeepsFileDatabase(t *testing.T) {
+	dir := t.TempDir()
+	db, err := OpenFile(dir, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Put([]byte("a"), []byte("1"), nil); err != nil {
+		t.Fatal(err)
+	}
+	if !db.reopen(db.db) {
+		t.Fatal("reopen failed")
+	}
+	if err := db.Put([]byte("b"), []byte("2"), nil); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Close(); err != nil {
+		t.Fatal(err)
+	}
+
+	db, err = OpenFile(dir, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	expectValue(t, db, "a", "1")
+	expectValue(t, db, "b", "2")
+}
+
+func TestReopenAttemptsAreRateLimited(t *testing.T) {
+	db, st := openFailing(t)
+	reopenInterval = time.Hour
+	st.fail.Store(true)
+	if err := db.Put([]byte("a"), []byte("1"), nil); err == nil {
+		t.Fatal("put while storage fails succeeded")
+	}
+	st.fail.Store(false)
+	if err := db.Put([]byte("a"), []byte("1"), nil); !errors.Is(err, syscall.ENOSPC) {
+		t.Fatalf("put within the reopen interval: %v", err)
+	}
+	db.nextReopen.Store(0)
+	if err := db.Put([]byte("a"), []byte("1"), nil); err != nil {
+		t.Fatalf("put after the reopen interval: %v", err)
+	}
+}
+
+func TestFailedReopenReportsErrorsUntilOpen(t *testing.T) {
+	db, st := openFailing(t)
+	if err := db.Put([]byte("a"), []byte("1"), nil); err != nil {
+		t.Fatal(err)
+	}
+	st.fail.Store(true)
+	if err := db.Put([]byte("b"), []byte("2"), nil); err == nil {
+		t.Fatal("put while storage fails succeeded")
+	}
+	st.fail.Store(false)
+	st.failOpen.Store(true)
+	if err := db.Put([]byte("b"), []byte("2"), nil); err == nil {
+		t.Fatal("put with a failed reopen succeeded")
+	}
+	if _, err := db.Get([]byte("a"), nil); !errors.Is(err, syscall.ENOSPC) {
+		t.Fatalf("get with a failed reopen: %v", err)
+	}
+	iter := db.NewIterator(nil, nil)
+	if iter.Next() || !errors.Is(iter.Error(), syscall.ENOSPC) {
+		t.Fatalf("iterator with a failed reopen: %v", iter.Error())
+	}
+	iter.Release()
+
+	st.failOpen.Store(false)
+	expectValue(t, db, "a", "1")
+	if err := db.Put([]byte("b"), []byte("2"), nil); err != nil {
+		t.Fatalf("put after the reopen succeeded: %v", err)
+	}
+}