diff --git a/weed/filer/leveldb/leveldb_store.go b/weed/filer/leveldb/leveldb_store.go index 341327e433..9a343d18eb 100644 --- a/weed/filer/leveldb/leveldb_store.go +++ b/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 diff --git a/weed/filer/leveldb/leveldb_store_test.go b/weed/filer/leveldb/leveldb_store_test.go index b55586ec98..4f9c13719e 100644 --- a/weed/filer/leveldb/leveldb_store_test.go +++ b/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") + } +} diff --git a/weed/filer/leveldb2/leveldb2_store.go b/weed/filer/leveldb2/leveldb2_store.go index 7a6771a5fc..7e29f8e632 100644 --- a/weed/filer/leveldb2/leveldb2_store.go +++ b/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 diff --git a/weed/filer/leveldb2/leveldb2_store_test.go b/weed/filer/leveldb2/leveldb2_store_test.go index 8d1d46ccc3..7e8854ca43 100644 --- a/weed/filer/leveldb2/leveldb2_store_test.go +++ b/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") + } +} diff --git a/weed/filer/leveldb3/leveldb3_store.go b/weed/filer/leveldb3/leveldb3_store.go index 5484383713..634a2878db 100644 --- a/weed/filer/leveldb3/leveldb3_store.go +++ b/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 diff --git a/weed/filer/leveldb3/leveldb3_store_test.go b/weed/filer/leveldb3/leveldb3_store_test.go index 1f57acd4c1..c51bbaeb8a 100644 --- a/weed/filer/leveldb3/leveldb3_store_test.go +++ b/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") + } +} diff --git a/weed/filer/leveldb_handle/leveldb_handle.go b/weed/filer/leveldb_handle/leveldb_handle.go new file mode 100644 index 0000000000..20813ebea5 --- /dev/null +++ b/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 + } +} diff --git a/weed/filer/leveldb_handle/leveldb_handle_test.go b/weed/filer/leveldb_handle/leveldb_handle_test.go new file mode 100644 index 0000000000..d22d319124 --- /dev/null +++ b/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) + } +}