From c68562f5a9256c2645069d104a0146aa0a2a9b28 Mon Sep 17 00:00:00 2001 From: Allen Ray Date: Wed, 10 Jun 2026 09:39:54 -0400 Subject: [PATCH 1/5] UPSTREAM: : backend: refactor defrag to minimize gRPC disruption Split defrag into three phases: unlocked bulk copy against a bbolt snapshot, locked journal replay, and atomic DB switchover. This reduces the gRPC blocking window from O(db_size) to O(writes_during_copy). Co-Authored-By: Claude Opus 4.6 (1M context) --- server/etcdserver/server.go | 7 +- server/storage/backend/backend.go | 203 ++- server/storage/backend/backend_test.go | 1118 +++++++++++++++++ server/storage/backend/batch_tx.go | 20 + server/storage/backend/defrag_journal.go | 122 ++ server/storage/backend/defrag_journal_test.go | 88 ++ 6 files changed, 1500 insertions(+), 58 deletions(-) create mode 100644 server/storage/backend/defrag_journal.go create mode 100644 server/storage/backend/defrag_journal_test.go diff --git a/server/etcdserver/server.go b/server/etcdserver/server.go index 5338815ae9d8..79e5d545cfe4 100644 --- a/server/etcdserver/server.go +++ b/server/etcdserver/server.go @@ -260,6 +260,7 @@ type EtcdServer struct { kv mvcc.WatchableKV lessor lease.Lessor bemu sync.RWMutex + defragMu sync.Mutex be backend.Backend beHooks *serverstorage.BackendHooks authStore auth.AuthStore @@ -975,8 +976,10 @@ func (s *EtcdServer) Cleanup() { } func (s *EtcdServer) Defragment() error { - s.bemu.Lock() - defer s.bemu.Unlock() + s.defragMu.Lock() + defer s.defragMu.Unlock() + s.bemu.RLock() + defer s.bemu.RUnlock() return s.be.Defrag() } diff --git a/server/storage/backend/backend.go b/server/storage/backend/backend.go index 275064f083b2..5b031364c8c8 100644 --- a/server/storage/backend/backend.go +++ b/server/storage/backend/backend.go @@ -15,6 +15,7 @@ package backend import ( + "errors" "fmt" "hash/crc32" "io" @@ -28,6 +29,7 @@ import ( "go.uber.org/zap" bolt "go.etcd.io/bbolt" + bolterrors "go.etcd.io/bbolt/errors" "go.etcd.io/etcd/client/pkg/v3/verify" ) @@ -469,20 +471,6 @@ func (b *backend) defrag() error { isDefragActive.Set(1) defer isDefragActive.Set(0) - // TODO: make this non-blocking? - // lock batchTx to ensure nobody is using previous tx, and then - // close previous ongoing tx. - b.batchTx.LockOutsideApply() - defer b.batchTx.Unlock() - - // lock database after lock tx to avoid deadlock. - b.mu.Lock() - defer b.mu.Unlock() - - // block concurrent read requests while resetting tx - b.readTx.Lock() - defer b.readTx.Unlock() - // Create a temporary file to ensure we start with a clean slate. // Snapshotter.cleanupSnapdir cleans up any of these that are found during startup. dir := filepath.Dir(b.db.Path()) @@ -500,27 +488,23 @@ func (b *backend) defrag() error { // return nil, fmt.Errorf(defragOpenFileError) return temp, nil } - // Don't load tmp db into memory regardless of opening options options.Mlock = false tdbp := temp.Name() tmpdb, err := bolt.Open(tdbp, 0o600, &options) if err != nil { temp.Close() if rmErr := os.Remove(temp.Name()); rmErr != nil { - b.lg.Error( - "failed to remove temporary file", + b.lg.Error("failed to remove temporary file", zap.String("path", temp.Name()), zap.Error(rmErr), ) } - return err } dbp := b.db.Path() size1, sizeInUse1 := b.Size(), b.SizeInUse() - b.lg.Info( - "defragmenting", + b.lg.Info("defragmenting", zap.String("path", dbp), zap.Int64("current-db-size-bytes", size1), zap.String("current-db-size", humanize.Bytes(uint64(size1))), @@ -528,35 +512,88 @@ func (b *backend) defrag() error { zap.String("current-db-size-in-use", humanize.Bytes(uint64(sizeInUse1))), ) - defer func() { - // NOTE: We should exit as soon as possible because that tx - // might be closed. The inflight request might use invalid - // tx and then panic as well. The real panic reason might be - // shadowed by new panic. So, we should fatal here with lock. - if rerr := recover(); rerr != nil { - b.lg.Fatal("unexpected panic during defrag", zap.Any("panic", rerr)) - } - }() + // ============================================================ + // PHASE 1: Commit pending writes, take a snapshot, install the + // journal, then unlock so writes can continue during the copy. + // ============================================================ + b.batchTx.LockOutsideApply() - // Commit/stop and then reset current transactions (including the readTx) - b.batchTx.unsafeCommit(true) - b.batchTx.tx = nil + b.batchTx.commit(false) + + b.mu.RLock() + snapTx, snapErr := b.db.Begin(false) + b.mu.RUnlock() + if snapErr != nil { + b.batchTx.Unlock() + tmpdb.Close() + os.Remove(tdbp) + return fmt.Errorf("failed to begin snapshot tx for defrag: %w", snapErr) + } + + journal := newDefragJournal() + b.batchTx.defragJournal = journal + b.batchTx.Unlock() + + b.lg.Info("defrag: copying data (writes unlocked)") // gofail: var defragBeforeCopy struct{} - err = defragdb(b.db, tmpdb, defragLimit) + err = defragFromTx(snapTx, tmpdb, defragLimit) + snapTx.Rollback() + if err != nil { + b.batchTx.LockOutsideApply() + b.batchTx.defragJournal = nil + b.batchTx.Unlock() + journal.close() tmpdb.Close() - if rmErr := os.RemoveAll(tmpdb.Path()); rmErr != nil { - b.lg.Error("failed to remove db.tmp after defragmentation completed", zap.Error(rmErr)) - } + os.RemoveAll(tdbp) + return err + } + + // ============================================================ + // PHASE 2: Lock writes, drain and replay the journal into the + // temp DB. This should be fast since it's only the delta. + // ============================================================ + b.lg.Info("defrag: replaying journal") - // restore the bbolt transactions if defragmentation fails - b.batchTx.tx = b.unsafeBegin(true) - b.readTx.tx = b.unsafeBegin(false) + b.batchTx.LockOutsideApply() + b.batchTx.defragJournal = nil + journal.close() + ops := journal.drain() + + if len(ops) > 0 { + b.lg.Info("defrag: replaying journal ops", zap.Int("count", len(ops))) + } + err = replayJournal(tmpdb, ops, defragLimit) + if err != nil { + b.batchTx.Unlock() + tmpdb.Close() + os.RemoveAll(tdbp) return err } + // ============================================================ + // PHASE 3: Atomic switchover — close old db, rename, reopen. + // ============================================================ + b.lg.Info("defrag: switching database") + + b.mu.Lock() + b.readTx.Lock() + + // NOTE: We should exit as soon as possible because that tx + // might be closed. The inflight request might use invalid + // tx and then panic as well. The real panic reason might be + // shadowed by new panic. So, we should fatal here with lock. + defer func() { + if rerr := recover(); rerr != nil { + b.lg.Fatal("unexpected panic during defrag", zap.Any("panic", rerr)) + } + }() + + b.batchTx.unsafeCommit(true) + b.batchTx.tx = nil + err = b.db.Close() if err != nil { b.lg.Fatal("failed to close database", zap.Error(err)) @@ -585,12 +622,15 @@ func (b *backend) defrag() error { atomic.StoreInt64(&b.size, size) atomic.StoreInt64(&b.sizeInUse, size-(int64(db.Stats().FreePageN)*int64(db.Info().PageSize))) + b.readTx.Unlock() + b.mu.Unlock() + b.batchTx.Unlock() + took := time.Since(now) defragSec.Observe(took.Seconds()) size2, sizeInUse2 := b.Size(), b.SizeInUse() - b.lg.Info( - "finished defragmenting directory", + b.lg.Info("finished defragmenting directory", zap.String("path", dbp), zap.Int64("current-db-size-bytes-diff", size2-size1), zap.Int64("current-db-size-bytes", size2), @@ -603,11 +643,7 @@ func (b *backend) defrag() error { return nil } -func defragdb(odb, tmpdb *bolt.DB, limit int) error { - // gofail: var defragdbFail string - // return fmt.Errorf(defragdbFail) - - // open a tx on tmpdb for writes +func defragFromTx(srcTx *bolt.Tx, tmpdb *bolt.DB, limit int) error { tmptx, err := tmpdb.Begin(true) if err != nil { return err @@ -618,18 +654,10 @@ func defragdb(odb, tmpdb *bolt.DB, limit int) error { } }() - // open a tx on old db for read - tx, err := odb.Begin(false) - if err != nil { - return err - } - defer tx.Rollback() - - c := tx.Cursor() - + c := srcTx.Cursor() count := 0 for next, _ := c.First(); next != nil; next, _ = c.Next() { - b := tx.Bucket(next) + b := srcTx.Bucket(next) if b == nil { return fmt.Errorf("backend: cannot defrag bucket %s", next) } @@ -665,6 +693,69 @@ func defragdb(odb, tmpdb *bolt.DB, limit int) error { return tmptx.Commit() } +func replayJournal(tmpdb *bolt.DB, ops []defragJournalOp, limit int) error { + if len(ops) == 0 { + return nil + } + + tx, err := tmpdb.Begin(true) + if err != nil { + return err + } + defer func() { + if err != nil { + tx.Rollback() + } + }() + + count := 0 + for _, op := range ops { + count++ + if count > limit { + if err = tx.Commit(); err != nil { + return err + } + tx, err = tmpdb.Begin(true) + if err != nil { + return err + } + count = 0 + } + + switch op.opType { + case opCreateBucket: + if _, err = tx.CreateBucketIfNotExists(op.bucketName); err != nil { + return fmt.Errorf("replay: create bucket %s: %w", op.bucketName, err) + } + case opDeleteBucket: + if delErr := tx.DeleteBucket(op.bucketName); delErr != nil && !errors.Is(delErr, bolterrors.ErrBucketNotFound) { + return fmt.Errorf("replay: delete bucket %s: %w", op.bucketName, delErr) + } + case opPut: + b := tx.Bucket(op.bucketName) + if b == nil { + return fmt.Errorf("replay: bucket %s not found for put", op.bucketName) + } + if op.seq { + b.FillPercent = 0.9 + } + if err = b.Put(op.key, op.value); err != nil { + return fmt.Errorf("replay: put in bucket %s: %w", op.bucketName, err) + } + case opDelete: + b := tx.Bucket(op.bucketName) + if b == nil { + continue + } + if err = b.Delete(op.key); err != nil { + return fmt.Errorf("replay: delete from bucket %s: %w", op.bucketName, err) + } + } + } + + return tx.Commit() +} + func (b *backend) begin(write bool) *bolt.Tx { b.mu.RLock() tx := b.unsafeBegin(write) diff --git a/server/storage/backend/backend_test.go b/server/storage/backend/backend_test.go index fc024b88bc1f..2ccbadb0b4e3 100644 --- a/server/storage/backend/backend_test.go +++ b/server/storage/backend/backend_test.go @@ -18,6 +18,8 @@ import ( "fmt" "os" "reflect" + "sync" + "sync/atomic" "testing" "time" @@ -349,3 +351,1119 @@ func TestBackendWritebackForEach(t *testing.T) { t.Fatalf("expected %q, got %q", seq, partialSeq) } } + +func TestBackendDefragConcurrentWrites(t *testing.T) { + b, _ := betesting.NewDefaultTmpBackend(t) + defer betesting.Close(t, b) + + largeVal := make([]byte, 1024) + tx := b.BatchTx() + tx.Lock() + tx.UnsafeCreateBucket(schema.Test) + for i := 0; i < 5000; i++ { + tx.UnsafePut(schema.Test, []byte(fmt.Sprintf("pre_%05d", i)), largeVal) + } + tx.Unlock() + b.ForceCommit() + + // delete some keys to create reclaimable space + tx.Lock() + for i := 0; i < 2500; i++ { + tx.UnsafeDelete(schema.Test, []byte(fmt.Sprintf("pre_%05d", i))) + } + tx.Unlock() + b.ForceCommit() + + // Writer waits until defrag is about to start, then writes concurrently. + var wg sync.WaitGroup + stop := make(chan struct{}) + defragStarted := make(chan struct{}) + var opsDone atomic.Int32 + + wg.Add(1) + go func() { + defer wg.Done() + <-defragStarted + for i := 0; ; i++ { + select { + case <-stop: + return + default: + } + tx := b.BatchTx() + tx.Lock() + tx.UnsafePut(schema.Test, []byte(fmt.Sprintf("concurrent_%04d", i)), []byte("new")) + tx.Unlock() + opsDone.Add(1) + } + }() + + defragDone := make(chan error, 1) + close(defragStarted) + go func() { + defragDone <- b.Defrag() + }() + err := <-defragDone + close(stop) + wg.Wait() + + require.NoError(t, err) + require.Greater(t, opsDone.Load(), int32(0), "at least one write must complete during defrag") + + // verify pre-existing keys that weren't deleted still exist + rtx := b.BatchTx() + rtx.Lock() + for i := 2500; i < 5000; i++ { + keys, vals := rtx.UnsafeRange(schema.Test, []byte(fmt.Sprintf("pre_%05d", i)), nil, 0) + require.Lenf(t, keys, 1, "key pre_%05d should exist", i) + assert.Equal(t, largeVal, vals[0]) + } + // verify deleted keys are gone + for i := 0; i < 2500; i++ { + keys, _ := rtx.UnsafeRange(schema.Test, []byte(fmt.Sprintf("pre_%05d", i)), nil, 0) + assert.Emptyf(t, keys, "key pre_%05d should be deleted", i) + } + rtx.Unlock() + + // verify at least some concurrent writes are present + b.ForceCommit() + rtx.Lock() + var found int + for i := 0; i < 10000; i++ { + keys, _ := rtx.UnsafeRange(schema.Test, []byte(fmt.Sprintf("concurrent_%04d", i)), nil, 0) + if len(keys) > 0 { + found++ + } + } + rtx.Unlock() + assert.Greater(t, found, 0, "at least some concurrent writes should be present after defrag") +} + +func TestBackendDefragConcurrentReads(t *testing.T) { + b, _ := betesting.NewDefaultTmpBackend(t) + defer betesting.Close(t, b) + + tx := b.BatchTx() + tx.Lock() + tx.UnsafeCreateBucket(schema.Test) + for i := 0; i < 1000; i++ { + tx.UnsafePut(schema.Test, []byte(fmt.Sprintf("key_%04d", i)), []byte("value")) + } + tx.Unlock() + b.ForceCommit() + + var wg sync.WaitGroup + stop := make(chan struct{}) + readErrors := make(chan error, 100) + + // concurrent reader + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-stop: + return + default: + } + rtx := b.ConcurrentReadTx() + rtx.RLock() + keys, _ := rtx.UnsafeRange(schema.Test, []byte("key_0500"), nil, 0) + if len(keys) == 0 { + readErrors <- fmt.Errorf("key_0500 not found during defrag") + } + rtx.RUnlock() + } + }() + + err := b.Defrag() + close(stop) + wg.Wait() + close(readErrors) + + require.NoError(t, err) + for readErr := range readErrors { + t.Error(readErr) + } +} + +func TestBackendDefragWriteAvailability(t *testing.T) { + b, _ := betesting.NewDefaultTmpBackend(t) + defer betesting.Close(t, b) + + tx := b.BatchTx() + tx.Lock() + tx.UnsafeCreateBucket(schema.Test) + for i := 0; i < backend.DefragLimitForTest()+100; i++ { + tx.UnsafePut(schema.Test, []byte(fmt.Sprintf("key_%06d", i)), []byte("value")) + } + tx.Unlock() + b.ForceCommit() + + // delete half the keys to make defrag meaningful + tx.Lock() + for i := 0; i < backend.DefragLimitForTest()/2; i++ { + tx.UnsafeDelete(schema.Test, []byte(fmt.Sprintf("key_%06d", i))) + } + tx.Unlock() + b.ForceCommit() + + // track write latencies during defrag + var wg sync.WaitGroup + stop := make(chan struct{}) + var maxWriteLatency time.Duration + var mu sync.Mutex + + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; ; i++ { + select { + case <-stop: + return + default: + } + start := time.Now() + wtx := b.BatchTx() + wtx.Lock() + wtx.UnsafePut(schema.Test, []byte(fmt.Sprintf("during_%06d", i)), []byte("val")) + wtx.Unlock() + elapsed := time.Since(start) + + mu.Lock() + if elapsed > maxWriteLatency { + maxWriteLatency = elapsed + } + mu.Unlock() + } + }() + + err := b.Defrag() + close(stop) + wg.Wait() + + require.NoError(t, err) + t.Logf("max write latency during defrag: %v", maxWriteLatency) + // The max latency should be well under a second since the copy phase + // doesn't hold the batchTx lock. Writes only block during the brief + // replay + switchover phases. + assert.Less(t, maxWriteLatency, 5*time.Second, + "write latency during defrag should be bounded") +} + +func TestBackendDefragConcurrentReadsAndWrites(t *testing.T) { + b, _ := betesting.NewDefaultTmpBackend(t) + defer betesting.Close(t, b) + + tx := b.BatchTx() + tx.Lock() + tx.UnsafeCreateBucket(schema.Test) + for i := 0; i < 500; i++ { + tx.UnsafePut(schema.Test, []byte(fmt.Sprintf("key_%04d", i)), []byte("original")) + } + tx.Unlock() + b.ForceCommit() + + var wg sync.WaitGroup + stop := make(chan struct{}) + + // concurrent writer + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; ; i++ { + select { + case <-stop: + return + default: + } + wtx := b.BatchTx() + wtx.Lock() + wtx.UnsafePut(schema.Test, []byte(fmt.Sprintf("write_%04d", i)), []byte("new")) + wtx.Unlock() + } + }() + + // concurrent reader using single-key lookups (schema.Test is not a safe-range bucket) + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-stop: + return + default: + } + rtx := b.ConcurrentReadTx() + rtx.RLock() + rtx.UnsafeRange(schema.Test, []byte("key_0250"), nil, 0) + rtx.RUnlock() + } + }() + + err := b.Defrag() + close(stop) + wg.Wait() + + require.NoError(t, err) + + // verify original data survived + b.ForceCommit() + rtx := b.BatchTx() + rtx.Lock() + for i := 0; i < 500; i++ { + keys, vals := rtx.UnsafeRange(schema.Test, []byte(fmt.Sprintf("key_%04d", i)), nil, 0) + require.Lenf(t, keys, 1, "key_%04d should exist", i) + assert.Equal(t, []byte("original"), vals[0]) + } + rtx.Unlock() +} + +func TestBackendDefragDeletesDuringDefrag(t *testing.T) { + b, _ := betesting.NewDefaultTmpBackend(t) + defer betesting.Close(t, b) + + tx := b.BatchTx() + tx.Lock() + tx.UnsafeCreateBucket(schema.Test) + for i := 0; i < 100; i++ { + tx.UnsafePut(schema.Test, []byte(fmt.Sprintf("key_%04d", i)), []byte("value")) + } + tx.Unlock() + b.ForceCommit() + + // delete keys concurrently during defrag + var wg sync.WaitGroup + stop := make(chan struct{}) + + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; i < 50; i++ { + select { + case <-stop: + return + default: + } + tx := b.BatchTx() + tx.Lock() + tx.UnsafeDelete(schema.Test, []byte(fmt.Sprintf("key_%04d", i))) + tx.Unlock() + } + }() + + err := b.Defrag() + close(stop) + wg.Wait() + + require.NoError(t, err) + + // keys 50-99 should still exist + b.ForceCommit() + rtx := b.BatchTx() + rtx.Lock() + for i := 50; i < 100; i++ { + keys, _ := rtx.UnsafeRange(schema.Test, []byte(fmt.Sprintf("key_%04d", i)), nil, 0) + require.Lenf(t, keys, 1, "key_%04d should exist", i) + } + rtx.Unlock() +} + +func TestBackendDefragLogicalConsistency(t *testing.T) { + b, _ := betesting.NewDefaultTmpBackend(t) + defer betesting.Close(t, b) + + tx := b.BatchTx() + tx.Lock() + tx.UnsafeCreateBucket(schema.Test) + tx.UnsafeCreateBucket(schema.Key) + for i := 0; i < 500; i++ { + tx.UnsafePut(schema.Test, []byte(fmt.Sprintf("test_%04d", i)), []byte(fmt.Sprintf("val_%04d", i))) + } + for i := 0; i < 300; i++ { + tx.UnsafeSeqPut(schema.Key, []byte(fmt.Sprintf("rev_%06d", i)), []byte(fmt.Sprintf("data_%06d", i))) + } + tx.Unlock() + b.ForceCommit() + + // delete some keys to create fragmentation + tx.Lock() + for i := 0; i < 200; i++ { + tx.UnsafeDelete(schema.Test, []byte(fmt.Sprintf("test_%04d", i))) + } + tx.Unlock() + b.ForceCommit() + + // capture logical hash before defrag + hashBefore, err := b.Hash(nil) + require.NoError(t, err) + sizeBefore := b.Size() + + err = b.Defrag() + require.NoError(t, err) + + // hash must be identical — same logical content + hashAfter, err := b.Hash(nil) + require.NoError(t, err) + assert.Equal(t, hashBefore, hashAfter, "logical hash must match after defrag") + + // physical size should have shrunk + sizeAfter := b.Size() + assert.Less(t, sizeAfter, sizeBefore, "defrag should reclaim space") + + // verify all surviving keys individually + b.ForceCommit() + rtx := b.BatchTx() + rtx.Lock() + for i := 200; i < 500; i++ { + keys, vals := rtx.UnsafeRange(schema.Test, []byte(fmt.Sprintf("test_%04d", i)), nil, 0) + require.Lenf(t, keys, 1, "test_%04d should exist", i) + assert.Equal(t, []byte(fmt.Sprintf("val_%04d", i)), vals[0]) + } + for i := 0; i < 300; i++ { + keys, vals := rtx.UnsafeRange(schema.Key, []byte(fmt.Sprintf("rev_%06d", i)), nil, 0) + require.Lenf(t, keys, 1, "rev_%06d should exist", i) + assert.Equal(t, []byte(fmt.Sprintf("data_%06d", i)), vals[0]) + } + // deleted keys must be gone + for i := 0; i < 200; i++ { + keys, _ := rtx.UnsafeRange(schema.Test, []byte(fmt.Sprintf("test_%04d", i)), nil, 0) + assert.Emptyf(t, keys, "test_%04d should be deleted", i) + } + rtx.Unlock() +} + +func TestBackendDefragOverwriteDuringCopy(t *testing.T) { + b, _ := betesting.NewDefaultTmpBackend(t) + defer betesting.Close(t, b) + + tx := b.BatchTx() + tx.Lock() + tx.UnsafeCreateBucket(schema.Test) + for i := 0; i < 100; i++ { + tx.UnsafePut(schema.Test, []byte(fmt.Sprintf("key_%04d", i)), []byte("original")) + } + tx.Unlock() + b.ForceCommit() + + // overwrite keys concurrently during defrag + var wg sync.WaitGroup + stop := make(chan struct{}) + + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-stop: + return + default: + } + wtx := b.BatchTx() + wtx.Lock() + for i := 0; i < 50; i++ { + wtx.UnsafePut(schema.Test, []byte(fmt.Sprintf("key_%04d", i)), []byte("updated")) + } + wtx.Unlock() + } + }() + + err := b.Defrag() + close(stop) + wg.Wait() + require.NoError(t, err) + + // all 50 overwritten keys must have the new value + b.ForceCommit() + rtx := b.BatchTx() + rtx.Lock() + for i := 0; i < 50; i++ { + keys, vals := rtx.UnsafeRange(schema.Test, []byte(fmt.Sprintf("key_%04d", i)), nil, 0) + require.Lenf(t, keys, 1, "key_%04d should exist", i) + assert.Equalf(t, []byte("updated"), vals[0], "key_%04d should have updated value", i) + } + // keys 50-99 should still have original value + for i := 50; i < 100; i++ { + keys, vals := rtx.UnsafeRange(schema.Test, []byte(fmt.Sprintf("key_%04d", i)), nil, 0) + require.Lenf(t, keys, 1, "key_%04d should exist", i) + assert.Equalf(t, []byte("original"), vals[0], "key_%04d should have original value", i) + } + rtx.Unlock() +} + +func TestBackendDefragNewBucketDuringCopy(t *testing.T) { + b, _ := betesting.NewDefaultTmpBackend(t) + defer betesting.Close(t, b) + + tx := b.BatchTx() + tx.Lock() + tx.UnsafeCreateBucket(schema.Test) + for i := 0; i < 100; i++ { + tx.UnsafePut(schema.Test, []byte(fmt.Sprintf("key_%04d", i)), []byte("value")) + } + tx.Unlock() + b.ForceCommit() + + // create a new bucket and write to it during defrag + var wg sync.WaitGroup + done := make(chan struct{}) + + wg.Add(1) + go func() { + defer wg.Done() + wtx := b.BatchTx() + wtx.Lock() + wtx.UnsafeCreateBucket(schema.Key) + for i := 0; i < 50; i++ { + wtx.UnsafeSeqPut(schema.Key, []byte(fmt.Sprintf("newkey_%04d", i)), []byte("newval")) + } + wtx.Unlock() + close(done) + }() + + err := b.Defrag() + <-done + wg.Wait() + require.NoError(t, err) + + // verify original bucket data + b.ForceCommit() + rtx := b.BatchTx() + rtx.Lock() + for i := 0; i < 100; i++ { + keys, _ := rtx.UnsafeRange(schema.Test, []byte(fmt.Sprintf("key_%04d", i)), nil, 0) + require.Lenf(t, keys, 1, "key_%04d should exist in Test bucket", i) + } + // verify new bucket and its data exist + for i := 0; i < 50; i++ { + keys, vals := rtx.UnsafeRange(schema.Key, []byte(fmt.Sprintf("newkey_%04d", i)), nil, 0) + require.Lenf(t, keys, 1, "newkey_%04d should exist in Key bucket", i) + assert.Equal(t, []byte("newval"), vals[0]) + } + rtx.Unlock() +} + +func TestBackendDefragMultipleSequential(t *testing.T) { + b, _ := betesting.NewDefaultTmpBackend(t) + defer betesting.Close(t, b) + + tx := b.BatchTx() + tx.Lock() + tx.UnsafeCreateBucket(schema.Test) + for i := 0; i < 200; i++ { + tx.UnsafePut(schema.Test, []byte(fmt.Sprintf("key_%04d", i)), []byte("value")) + } + tx.Unlock() + b.ForceCommit() + + for round := 0; round < 3; round++ { + // delete some keys before each defrag + tx.Lock() + for i := round * 20; i < (round+1)*20; i++ { + tx.UnsafeDelete(schema.Test, []byte(fmt.Sprintf("key_%04d", i))) + } + tx.Unlock() + b.ForceCommit() + + hashBefore, err := b.Hash(nil) + require.NoError(t, err) + + err = b.Defrag() + require.NoErrorf(t, err, "defrag round %d failed", round) + + hashAfter, err := b.Hash(nil) + require.NoError(t, err) + assert.Equalf(t, hashBefore, hashAfter, "hash mismatch after defrag round %d", round) + + // add new keys after each defrag + tx.Lock() + tx.UnsafeCreateBucket(schema.Test) + for i := 0; i < 10; i++ { + tx.UnsafePut(schema.Test, []byte(fmt.Sprintf("round%d_%04d", round, i)), []byte("new")) + } + tx.Unlock() + b.ForceCommit() + } + + // verify round keys from all 3 rounds exist + rtx := b.BatchTx() + rtx.Lock() + for round := 0; round < 3; round++ { + for i := 0; i < 10; i++ { + keys, _ := rtx.UnsafeRange(schema.Test, []byte(fmt.Sprintf("round%d_%04d", round, i)), nil, 0) + require.Lenf(t, keys, 1, "round%d_%04d should exist", round, i) + } + } + rtx.Unlock() +} + +func TestBackendDefragEmptyDatabase(t *testing.T) { + b, _ := betesting.NewDefaultTmpBackend(t) + defer betesting.Close(t, b) + + err := b.Defrag() + require.NoError(t, err) + + // verify we can still use the database after defragging an empty one + tx := b.BatchTx() + tx.Lock() + tx.UnsafeCreateBucket(schema.Test) + tx.UnsafePut(schema.Test, []byte("after_defrag"), []byte("works")) + tx.Unlock() + b.ForceCommit() + + rtx := b.BatchTx() + rtx.Lock() + keys, vals := rtx.UnsafeRange(schema.Test, []byte("after_defrag"), nil, 0) + require.Len(t, keys, 1) + assert.Equal(t, []byte("works"), vals[0]) + rtx.Unlock() +} + +func TestBackendDefragLargeJournalReplay(t *testing.T) { + b, _ := betesting.NewDefaultTmpBackend(t) + defer betesting.Close(t, b) + + tx := b.BatchTx() + tx.Lock() + tx.UnsafeCreateBucket(schema.Test) + // pre-populate enough data to slow down Phase 1 copy, giving the + // concurrent writer time to exceed defragLimit + largeVal := make([]byte, 1024) + for i := 0; i < 5000; i++ { + tx.UnsafePut(schema.Test, []byte(fmt.Sprintf("pre_%05d", i)), largeVal) + } + tx.Unlock() + b.ForceCommit() + + // write enough keys during defrag to exceed defragLimit (10000) in the replay, + // exercising the batched commit path in replayJournal + var wg sync.WaitGroup + stop := make(chan struct{}) + totalWritten := make(chan int, 1) + + wg.Add(1) + go func() { + defer wg.Done() + var count int + for i := 0; ; i++ { + select { + case <-stop: + totalWritten <- count + return + default: + } + wtx := b.BatchTx() + wtx.Lock() + for j := 0; j < 10; j++ { + wtx.UnsafePut(schema.Test, []byte(fmt.Sprintf("journal_%06d", count)), []byte("jval")) + count++ + } + wtx.Unlock() + } + }() + + err := b.Defrag() + close(stop) + wg.Wait() + written := <-totalWritten + + require.NoError(t, err) + require.Greater(t, written, 0, + "at least some writes must occur during defrag") + t.Logf("journal ops written during defrag: %d", written) + + // verify pre-existing keys survived + b.ForceCommit() + rtx := b.BatchTx() + rtx.Lock() + for i := 0; i < 5000; i++ { + keys, _ := rtx.UnsafeRange(schema.Test, []byte(fmt.Sprintf("pre_%05d", i)), nil, 0) + require.Lenf(t, keys, 1, "pre_%05d should exist", i) + } + // verify at least some journal keys exist + var found int + for i := 0; i < written; i++ { + keys, _ := rtx.UnsafeRange(schema.Test, []byte(fmt.Sprintf("journal_%06d", i)), nil, 0) + if len(keys) > 0 { + found++ + } + } + rtx.Unlock() + assert.Greater(t, found, 0, "journal writes should be present after defrag") +} + +func TestBackendDefragReadsDuringDefragSeeBufferedPuts(t *testing.T) { + b, _ := betesting.NewDefaultTmpBackend(t) + defer betesting.Close(t, b) + + tx := b.BatchTx() + tx.Lock() + tx.UnsafeCreateBucket(schema.Test) + tx.UnsafePut(schema.Test, []byte("pre_key"), []byte("pre_val")) + tx.Unlock() + b.ForceCommit() + + // During defrag, writes go to journal+buffer only (not bbolt). + // Readers via ConcurrentReadTx should still see these puts + // because readTx.buf is preserved (commits are skipped). + wroteKey := make(chan struct{}) + readResult := make(chan bool, 1) + stop := make(chan struct{}) + var wg sync.WaitGroup + + // writer: put a key during defrag and signal + wg.Add(1) + go func() { + defer wg.Done() + // wait a moment for defrag to start + time.Sleep(10 * time.Millisecond) + wtx := b.BatchTx() + wtx.Lock() + wtx.UnsafePut(schema.Test, []byte("during_defrag"), []byte("visible")) + wtx.Unlock() + close(wroteKey) + <-stop + }() + + // reader: after the write, check if the key is visible via ConcurrentReadTx + wg.Add(1) + go func() { + defer wg.Done() + <-wroteKey + time.Sleep(5 * time.Millisecond) + rtx := b.ConcurrentReadTx() + rtx.RLock() + keys, _ := rtx.UnsafeRange(schema.Test, []byte("during_defrag"), nil, 0) + rtx.RUnlock() + readResult <- len(keys) > 0 + <-stop + }() + + err := b.Defrag() + close(stop) + wg.Wait() + require.NoError(t, err) + + visible := <-readResult + assert.True(t, visible, "puts during defrag should be visible to ConcurrentReadTx via buffer") + + // verify key is also present after defrag + b.ForceCommit() + rtx := b.BatchTx() + rtx.Lock() + keys, vals := rtx.UnsafeRange(schema.Test, []byte("during_defrag"), nil, 0) + rtx.Unlock() + require.Len(t, keys, 1) + assert.Equal(t, []byte("visible"), vals[0]) +} + +func TestBackendDefragExceedBatchLimit(t *testing.T) { + // Use a small batch limit so we can exceed it without writing too many keys. + b, _ := betesting.NewTmpBackend(t, time.Hour, 100) + defer betesting.Close(t, b) + + largeVal := make([]byte, 1024) + tx := b.BatchTx() + tx.Lock() + tx.UnsafeCreateBucket(schema.Test) + for i := 0; i < 5000; i++ { + tx.UnsafePut(schema.Test, []byte(fmt.Sprintf("seed_%05d", i)), largeVal) + } + tx.Unlock() + b.ForceCommit() + + // Write more than batchLimit keys during defrag to exercise the + // Unlock path where pending >= batchLimit but commits are skipped. + var wg sync.WaitGroup + stop := make(chan struct{}) + totalWritten := make(chan int, 1) + + wg.Add(1) + go func() { + defer wg.Done() + var count int + for i := 0; ; i++ { + select { + case <-stop: + totalWritten <- count + return + default: + } + wtx := b.BatchTx() + wtx.Lock() + for j := 0; j < 10; j++ { + wtx.UnsafePut(schema.Test, []byte(fmt.Sprintf("batch_%06d", count)), []byte("v")) + count++ + } + wtx.Unlock() + } + }() + + err := b.Defrag() + close(stop) + wg.Wait() + written := <-totalWritten + + require.NoError(t, err) + t.Logf("wrote %d keys during defrag (batch limit = 100)", written) + require.Greater(t, written, 100, "should have exceeded the batch limit during defrag") + + // verify all keys are present after defrag + b.ForceCommit() + rtx := b.BatchTx() + rtx.Lock() + var found int + for i := 0; i < written; i++ { + keys, _ := rtx.UnsafeRange(schema.Test, []byte(fmt.Sprintf("batch_%06d", i)), nil, 0) + if len(keys) > 0 { + found++ + } + } + rtx.Unlock() + assert.Equal(t, written, found, "all keys written during defrag should be present") +} + +func TestBackendDefragCommitDuringDefragPreservesData(t *testing.T) { + b, _ := betesting.NewDefaultTmpBackend(t) + defer betesting.Close(t, b) + + tx := b.BatchTx() + tx.Lock() + tx.UnsafeCreateBucket(schema.Test) + tx.UnsafePut(schema.Test, []byte("seed"), []byte("value")) + tx.Unlock() + b.ForceCommit() + + // Simulate what backend.run() does: call Commit() periodically. + // During defrag, Commit() should be skipped so readTx.buf isn't cleared. + wroteKeys := make(chan struct{}) + commitDone := make(chan struct{}) + stop := make(chan struct{}) + var wg sync.WaitGroup + + // writer: put keys during defrag + wg.Add(1) + go func() { + defer wg.Done() + time.Sleep(10 * time.Millisecond) + for i := 0; i < 50; i++ { + wtx := b.BatchTx() + wtx.Lock() + wtx.UnsafePut(schema.Test, []byte(fmt.Sprintf("commit_test_%04d", i)), []byte("val")) + wtx.Unlock() + } + close(wroteKeys) + <-stop + }() + + // committer: call ForceCommit after writes, simulating periodic timer + wg.Add(1) + go func() { + defer wg.Done() + <-wroteKeys + b.ForceCommit() + close(commitDone) + <-stop + }() + + // reader: after commit, verify data is still visible + readResult := make(chan int, 1) + wg.Add(1) + go func() { + defer wg.Done() + <-commitDone + rtx := b.ConcurrentReadTx() + rtx.RLock() + var found int + for i := 0; i < 50; i++ { + keys, _ := rtx.UnsafeRange(schema.Test, []byte(fmt.Sprintf("commit_test_%04d", i)), nil, 0) + if len(keys) > 0 { + found++ + } + } + rtx.RUnlock() + readResult <- found + <-stop + }() + + err := b.Defrag() + close(stop) + wg.Wait() + require.NoError(t, err) + + found := <-readResult + assert.Equal(t, 50, found, "all puts should remain visible after Commit() during defrag") +} + +func TestBackendDefragConcurrentReadTxSurvivesSwitchover(t *testing.T) { + b, _ := betesting.NewDefaultTmpBackend(t) + defer betesting.Close(t, b) + + tx := b.BatchTx() + tx.Lock() + tx.UnsafeCreateBucket(schema.Test) + for i := 0; i < 100; i++ { + tx.UnsafePut(schema.Test, []byte(fmt.Sprintf("key_%04d", i)), []byte("value")) + } + tx.Unlock() + b.ForceCommit() + + // Acquire a ConcurrentReadTx BEFORE defrag starts. This holds a + // reference to the old DB's bbolt read tx via txWg. Phase 3 must + // wait for this reader to finish before it can close the old DB. + longReader := b.ConcurrentReadTx() + longReader.RLock() + + // Verify the long reader can read data + keys, vals := longReader.UnsafeRange(schema.Test, []byte("key_0050"), nil, 0) + require.Len(t, keys, 1) + assert.Equal(t, []byte("value"), vals[0]) + + defragDone := make(chan error, 1) + go func() { + defragDone <- b.Defrag() + }() + + // Hold the read tx open for a bit — Phase 3 should block on db.Close() + // waiting for this reader to release. + time.Sleep(50 * time.Millisecond) + + select { + case err := <-defragDone: + t.Fatalf("Defrag finished before longReader.RUnlock(): %v", err) + default: + } + + // The reader should still be able to read from the old snapshot + keys, vals = longReader.UnsafeRange(schema.Test, []byte("key_0000"), nil, 0) + require.Len(t, keys, 1) + assert.Equal(t, []byte("value"), vals[0]) + + // Release the long reader — this unblocks Phase 3 + longReader.RUnlock() + + err := <-defragDone + require.NoError(t, err) + + // After defrag, verify new reads work on the new DB + b.ForceCommit() + rtx := b.ConcurrentReadTx() + rtx.RLock() + for i := 0; i < 100; i++ { + keys, _ := rtx.UnsafeRange(schema.Test, []byte(fmt.Sprintf("key_%04d", i)), nil, 0) + require.Lenf(t, keys, 1, "key_%04d should exist after defrag", i) + } + rtx.RUnlock() +} + +func TestBackendDefragNoDeadlock(t *testing.T) { + b, _ := betesting.NewDefaultTmpBackend(t) + defer betesting.Close(t, b) + + tx := b.BatchTx() + tx.Lock() + tx.UnsafeCreateBucket(schema.Test) + for i := range 1000 { + tx.UnsafePut(schema.Test, []byte(fmt.Sprintf("key_%04d", i)), make([]byte, 512)) + } + tx.Unlock() + b.ForceCommit() + + done := make(chan struct{}) + go func() { + defer close(done) + var wg sync.WaitGroup + stop := make(chan struct{}) + + // Goroutine doing writes (acquires batchTx lock) + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; ; i++ { + select { + case <-stop: + return + default: + } + tx := b.BatchTx() + tx.Lock() + tx.UnsafePut(schema.Test, []byte(fmt.Sprintf("dl_%04d", i%100)), []byte("w")) + tx.Unlock() + } + }() + + // Goroutine doing reads (acquires readTx + mu) + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-stop: + return + default: + } + rtx := b.ConcurrentReadTx() + rtx.RLock() + rtx.UnsafeRange(schema.Test, []byte("key_0000"), nil, 0) + rtx.RUnlock() + } + }() + + // Goroutine doing commits (acquires batchTx lock + mu) + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-stop: + return + default: + } + b.ForceCommit() + } + }() + + // Run 5 sequential defrags (acquires batchTx + mu + readTx) + for range 5 { + require.NoError(t, b.Defrag()) + } + + close(stop) + wg.Wait() + }() + + select { + case <-done: + case <-time.After(30 * time.Second): + t.Fatal("deadlock detected: defrag with concurrent writes, reads, and commits did not complete within 30s") + } +} + +func TestBackendDefragMultipleConcurrentReadTxDuringSwitchover(t *testing.T) { + b, _ := betesting.NewDefaultTmpBackend(t) + defer betesting.Close(t, b) + + tx := b.BatchTx() + tx.Lock() + tx.UnsafeCreateBucket(schema.Test) + for i := 0; i < 200; i++ { + tx.UnsafePut(schema.Test, []byte(fmt.Sprintf("key_%04d", i)), []byte("value")) + } + tx.Unlock() + b.ForceCommit() + + // Simulate many concurrent readers that continuously create and + // release ConcurrentReadTx during defrag, including across the + // Phase 3 switchover. Each reader creates short-lived snapshots + // that must not panic or return errors when the DB is swapped. + var wg sync.WaitGroup + stop := make(chan struct{}) + readErrors := make(chan error, 1000) + + for r := range 5 { + wg.Add(1) + go func(id int) { + defer wg.Done() + for { + select { + case <-stop: + return + default: + } + rtx := b.ConcurrentReadTx() + rtx.RLock() + key := []byte(fmt.Sprintf("key_%04d", id*20)) + keys, vals := rtx.UnsafeRange(schema.Test, key, nil, 0) + if len(keys) != 1 { + readErrors <- fmt.Errorf("reader %d: expected 1 key, got %d", id, len(keys)) + } else if string(vals[0]) != "value" { + readErrors <- fmt.Errorf("reader %d: expected 'value', got %q", id, vals[0]) + } + rtx.RUnlock() + } + }(r) + } + + err := b.Defrag() + close(stop) + wg.Wait() + close(readErrors) + + require.NoError(t, err) + + var errs []error + for e := range readErrors { + errs = append(errs, e) + } + assert.Empty(t, errs, "no read errors should occur during defrag switchover") + + // Verify data integrity after defrag + b.ForceCommit() + rtx := b.BatchTx() + rtx.Lock() + for i := 0; i < 200; i++ { + keys, _ := rtx.UnsafeRange(schema.Test, []byte(fmt.Sprintf("key_%04d", i)), nil, 0) + require.Lenf(t, keys, 1, "key_%04d should exist after defrag", i) + } + rtx.Unlock() +} + +// TestBackendDefragWriteVisibilityDuringCopy verifies that writes made during +// the defrag copy phase are immediately visible via UnsafeRange on the batch +// transaction. This reproduces a crash (range failed to find revision pair) +// that occurred when writes were only sent to the journal and not to bbolt. +func TestBackendDefragWriteVisibilityDuringCopy(t *testing.T) { + b, _ := betesting.NewDefaultTmpBackend(t) + defer betesting.Close(t, b) + + tx := b.BatchTx() + tx.Lock() + tx.UnsafeCreateBucket(schema.Test) + for i := 0; i < 100; i++ { + tx.UnsafePut(schema.Test, []byte(fmt.Sprintf("init_%04d", i)), []byte("val")) + } + tx.Unlock() + b.ForceCommit() + + // Start defrag — during Phase 1 the journal is active. + defragDone := make(chan error, 1) + go func() { + defragDone <- b.Defrag() + }() + + // Write keys and immediately read them back while defrag is running. + // If the write only goes to the journal (not bbolt), UnsafeRange + // returns 0 results — reproducing the fatal crash. + for attempt := 0; attempt < 50; attempt++ { + key := []byte(fmt.Sprintf("during_%04d", attempt)) + val := []byte(fmt.Sprintf("v%d", attempt)) + + tx = b.BatchTx() + tx.Lock() + tx.UnsafePut(schema.Test, key, val) + keys, vals := tx.UnsafeRange(schema.Test, key, nil, 0) + tx.Unlock() + + require.Lenf(t, keys, 1, "key %s written during defrag must be readable", key) + require.Equal(t, val, vals[0]) + } + + err := <-defragDone + require.NoError(t, err) + + // After defrag completes, all writes must survive in the new database. + b.ForceCommit() + tx = b.BatchTx() + tx.Lock() + for i := 0; i < 50; i++ { + key := []byte(fmt.Sprintf("during_%04d", i)) + keys, _ := tx.UnsafeRange(schema.Test, key, nil, 0) + require.Lenf(t, keys, 1, "key %s must survive defrag", key) + } + for i := 0; i < 100; i++ { + keys, _ := tx.UnsafeRange(schema.Test, []byte(fmt.Sprintf("init_%04d", i)), nil, 0) + require.Lenf(t, keys, 1, "initial key init_%04d must survive defrag", i) + } + tx.Unlock() +} diff --git a/server/storage/backend/batch_tx.go b/server/storage/backend/batch_tx.go index 5af557cb4283..26b2d857609f 100644 --- a/server/storage/backend/batch_tx.go +++ b/server/storage/backend/batch_tx.go @@ -291,6 +291,7 @@ type batchTxBuffered struct { batchTx buf txWriteBuffer pendingDeleteOperations int + defragJournal *defragJournal } func newBatchTxBuffered(backend *backend) *batchTxBuffered { @@ -385,22 +386,41 @@ func (t *batchTxBuffered) unsafeCommit(stop bool) { } } +func (t *batchTxBuffered) UnsafeCreateBucket(bucket Bucket) { + if j := t.defragJournal; j != nil { + j.appendCreateBucket(bucket.Name()) + } + t.batchTx.UnsafeCreateBucket(bucket) +} + func (t *batchTxBuffered) UnsafePut(bucket Bucket, key []byte, value []byte) { + if j := t.defragJournal; j != nil { + j.appendPut(bucket.Name(), key, value, false) + } t.batchTx.UnsafePut(bucket, key, value) t.buf.put(bucket, key, value) } func (t *batchTxBuffered) UnsafeSeqPut(bucket Bucket, key []byte, value []byte) { + if j := t.defragJournal; j != nil { + j.appendPut(bucket.Name(), key, value, true) + } t.batchTx.UnsafeSeqPut(bucket, key, value) t.buf.putSeq(bucket, key, value) } func (t *batchTxBuffered) UnsafeDelete(bucketType Bucket, key []byte) { + if j := t.defragJournal; j != nil { + j.appendDelete(bucketType.Name(), key) + } t.batchTx.UnsafeDelete(bucketType, key) t.pendingDeleteOperations++ } func (t *batchTxBuffered) UnsafeDeleteBucket(bucket Bucket) { + if j := t.defragJournal; j != nil { + j.appendDeleteBucket(bucket.Name()) + } t.batchTx.UnsafeDeleteBucket(bucket) t.pendingDeleteOperations++ } diff --git a/server/storage/backend/defrag_journal.go b/server/storage/backend/defrag_journal.go new file mode 100644 index 000000000000..940891b7df25 --- /dev/null +++ b/server/storage/backend/defrag_journal.go @@ -0,0 +1,122 @@ +// Copyright 2024 The etcd Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package backend + +import "sync" + +type defragOpType uint8 + +const ( + opPut defragOpType = iota + opDelete + opCreateBucket + opDeleteBucket +) + +type defragJournalOp struct { + opType defragOpType + bucketName []byte + key []byte + value []byte + seq bool +} + +type defragJournal struct { + mu sync.Mutex + ops []defragJournalOp + closed bool +} + +func newDefragJournal() *defragJournal { + return &defragJournal{ + ops: make([]defragJournalOp, 0, 1024), + } +} + +func (j *defragJournal) appendPut(bucketName, key, value []byte, seq bool) { + j.mu.Lock() + defer j.mu.Unlock() + if j.closed { + return + } + j.ops = append(j.ops, defragJournalOp{ + opType: opPut, + bucketName: cloneBytes(bucketName), + key: cloneBytes(key), + value: cloneBytes(value), + seq: seq, + }) +} + +func (j *defragJournal) appendDelete(bucketName, key []byte) { + j.mu.Lock() + defer j.mu.Unlock() + if j.closed { + return + } + j.ops = append(j.ops, defragJournalOp{ + opType: opDelete, + bucketName: cloneBytes(bucketName), + key: cloneBytes(key), + }) +} + +func (j *defragJournal) appendCreateBucket(bucketName []byte) { + j.mu.Lock() + defer j.mu.Unlock() + if j.closed { + return + } + j.ops = append(j.ops, defragJournalOp{ + opType: opCreateBucket, + bucketName: cloneBytes(bucketName), + }) +} + +func (j *defragJournal) appendDeleteBucket(bucketName []byte) { + j.mu.Lock() + defer j.mu.Unlock() + if j.closed { + return + } + j.ops = append(j.ops, defragJournalOp{ + opType: opDeleteBucket, + bucketName: cloneBytes(bucketName), + }) +} + +// drain returns all accumulated ops and resets the journal. +func (j *defragJournal) drain() []defragJournalOp { + j.mu.Lock() + defer j.mu.Unlock() + ops := j.ops + j.ops = nil + return ops +} + +func (j *defragJournal) close() { + j.mu.Lock() + defer j.mu.Unlock() + j.closed = true +} + +func cloneBytes(b []byte) []byte { + if b == nil { + return nil + } + cp := make([]byte, len(b)) + copy(cp, b) + return cp +} diff --git a/server/storage/backend/defrag_journal_test.go b/server/storage/backend/defrag_journal_test.go new file mode 100644 index 000000000000..b5e8ccae3c86 --- /dev/null +++ b/server/storage/backend/defrag_journal_test.go @@ -0,0 +1,88 @@ +// Copyright 2024 The etcd Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package backend + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestDefragJournalAppendAndDrain(t *testing.T) { + j := newDefragJournal() + + j.appendCreateBucket([]byte("bucket1")) + j.appendPut([]byte("bucket1"), []byte("key1"), []byte("val1"), false) + j.appendPut([]byte("bucket1"), []byte("key2"), []byte("val2"), true) + j.appendDelete([]byte("bucket1"), []byte("key1")) + j.appendDeleteBucket([]byte("bucket1")) + + ops := j.drain() + require.Len(t, ops, 5) + + assert.Equal(t, opCreateBucket, ops[0].opType) + assert.Equal(t, []byte("bucket1"), ops[0].bucketName) + + assert.Equal(t, opPut, ops[1].opType) + assert.Equal(t, []byte("key1"), ops[1].key) + assert.Equal(t, []byte("val1"), ops[1].value) + assert.False(t, ops[1].seq) + + assert.Equal(t, opPut, ops[2].opType) + assert.Equal(t, []byte("key2"), ops[2].key) + assert.True(t, ops[2].seq) + + assert.Equal(t, opDelete, ops[3].opType) + assert.Equal(t, []byte("key1"), ops[3].key) + + assert.Equal(t, opDeleteBucket, ops[4].opType) + assert.Equal(t, []byte("bucket1"), ops[4].bucketName) + + // drain again should be empty + ops = j.drain() + assert.Empty(t, ops) +} + +func TestDefragJournalClose(t *testing.T) { + j := newDefragJournal() + + j.appendPut([]byte("b"), []byte("k"), []byte("v"), false) + j.close() + + // appends after close are no-ops + j.appendPut([]byte("b"), []byte("k2"), []byte("v2"), false) + j.appendDelete([]byte("b"), []byte("k")) + j.appendCreateBucket([]byte("b2")) + j.appendDeleteBucket([]byte("b")) + + ops := j.drain() + require.Len(t, ops, 1) + assert.Equal(t, []byte("k"), ops[0].key) +} + +func TestDefragJournalDeepCopies(t *testing.T) { + j := newDefragJournal() + + key := []byte("original") + j.appendPut([]byte("b"), key, []byte("v"), false) + + // mutate the source slice + key[0] = 'X' + + ops := j.drain() + require.Len(t, ops, 1) + assert.Equal(t, []byte("original"), ops[0].key, "journal should deep-copy byte slices") +} From d444792819f81365fce85491d6c11813f3126738 Mon Sep 17 00:00:00 2001 From: Allen Ray Date: Mon, 15 Jun 2026 12:18:44 -0400 Subject: [PATCH 2/5] separate phases into individual functions --- server/storage/backend/backend.go | 124 ++++++++++++++++-------------- 1 file changed, 68 insertions(+), 56 deletions(-) diff --git a/server/storage/backend/backend.go b/server/storage/backend/backend.go index 5b031364c8c8..c6849942cb6e 100644 --- a/server/storage/backend/backend.go +++ b/server/storage/backend/backend.go @@ -512,28 +512,13 @@ func (b *backend) defrag() error { zap.String("current-db-size-in-use", humanize.Bytes(uint64(sizeInUse1))), ) - // ============================================================ - // PHASE 1: Commit pending writes, take a snapshot, install the - // journal, then unlock so writes can continue during the copy. - // ============================================================ - b.batchTx.LockOutsideApply() - - b.batchTx.commit(false) - - b.mu.RLock() - snapTx, snapErr := b.db.Begin(false) - b.mu.RUnlock() - if snapErr != nil { - b.batchTx.Unlock() + journal, snapTx, err := b.defragSetupSnapshot() + if err != nil { tmpdb.Close() os.Remove(tdbp) - return fmt.Errorf("failed to begin snapshot tx for defrag: %w", snapErr) + return err } - journal := newDefragJournal() - b.batchTx.defragJournal = journal - b.batchTx.Unlock() - b.lg.Info("defrag: copying data (writes unlocked)") // gofail: var defragBeforeCopy struct{} @@ -541,22 +526,72 @@ func (b *backend) defrag() error { snapTx.Rollback() if err != nil { - b.batchTx.LockOutsideApply() - b.batchTx.defragJournal = nil - b.batchTx.Unlock() - journal.close() + b.defragCancelJournal(journal) tmpdb.Close() os.RemoveAll(tdbp) return err } - // ============================================================ - // PHASE 2: Lock writes, drain and replay the journal into the - // temp DB. This should be fast since it's only the delta. - // ============================================================ b.lg.Info("defrag: replaying journal") + err = b.defragReplayAndSwap(journal, tmpdb, dbp, tdbp) + if err != nil { + return err + } + took := time.Since(now) + defragSec.Observe(took.Seconds()) + + size2, sizeInUse2 := b.Size(), b.SizeInUse() + b.lg.Info("finished defragmenting directory", + zap.String("path", dbp), + zap.Int64("current-db-size-bytes-diff", size2-size1), + zap.Int64("current-db-size-bytes", size2), + zap.String("current-db-size", humanize.Bytes(uint64(size2))), + zap.Int64("current-db-size-in-use-bytes-diff", sizeInUse2-sizeInUse1), + zap.Int64("current-db-size-in-use-bytes", sizeInUse2), + zap.String("current-db-size-in-use", humanize.Bytes(uint64(sizeInUse2))), + zap.Duration("took", took), + ) + return nil +} + +// defragSetupSnapshot commits pending writes, takes a read-only +// snapshot, and installs a journal to capture writes during the copy. +func (b *backend) defragSetupSnapshot() (*defragJournal, *bolt.Tx, error) { b.batchTx.LockOutsideApply() + defer b.batchTx.Unlock() + + b.batchTx.commit(false) + + b.mu.RLock() + snapTx, err := b.db.Begin(false) + b.mu.RUnlock() + if err != nil { + return nil, nil, fmt.Errorf("failed to begin snapshot tx for defrag: %w", err) + } + + journal := newDefragJournal() + b.batchTx.defragJournal = journal + return journal, snapTx, nil +} + +// defragCancelJournal removes the journal from the batch transaction +// and closes it. Called when the copy phase fails. +func (b *backend) defragCancelJournal(journal *defragJournal) { + b.batchTx.LockOutsideApply() + defer b.batchTx.Unlock() + b.batchTx.defragJournal = nil + journal.close() +} + +// defragReplayAndSwap drains the journal, replays it into the temp +// database, then atomically swaps the old database for the new one. +// batchTx is held for the entire operation to prevent writes between +// journal drain and database swap. +func (b *backend) defragReplayAndSwap(journal *defragJournal, tmpdb *bolt.DB, dbp, tdbp string) error { + b.batchTx.LockOutsideApply() + defer b.batchTx.Unlock() + b.batchTx.defragJournal = nil journal.close() ops := journal.drain() @@ -565,21 +600,18 @@ func (b *backend) defrag() error { b.lg.Info("defrag: replaying journal ops", zap.Int("count", len(ops))) } - err = replayJournal(tmpdb, ops, defragLimit) - if err != nil { - b.batchTx.Unlock() + if err := replayJournal(tmpdb, ops, defragLimit); err != nil { tmpdb.Close() os.RemoveAll(tdbp) return err } - // ============================================================ - // PHASE 3: Atomic switchover — close old db, rename, reopen. - // ============================================================ b.lg.Info("defrag: switching database") b.mu.Lock() + defer b.mu.Unlock() b.readTx.Lock() + defer b.readTx.Unlock() // NOTE: We should exit as soon as possible because that tx // might be closed. The inflight request might use invalid @@ -594,20 +626,18 @@ func (b *backend) defrag() error { b.batchTx.unsafeCommit(true) b.batchTx.tx = nil - err = b.db.Close() - if err != nil { + if err := b.db.Close(); err != nil { b.lg.Fatal("failed to close database", zap.Error(err)) } - err = tmpdb.Close() - if err != nil { + if err := tmpdb.Close(); err != nil { b.lg.Fatal("failed to close tmp database", zap.Error(err)) } // gofail: var defragBeforeRename struct{} - err = os.Rename(tdbp, dbp) - if err != nil { + if err := os.Rename(tdbp, dbp); err != nil { b.lg.Fatal("failed to rename tmp database", zap.Error(err)) } + var err error b.db, err = bolt.Open(dbp, 0o600, b.bopts) if err != nil { b.lg.Fatal("failed to open database", zap.String("path", dbp), zap.Error(err)) @@ -622,24 +652,6 @@ func (b *backend) defrag() error { atomic.StoreInt64(&b.size, size) atomic.StoreInt64(&b.sizeInUse, size-(int64(db.Stats().FreePageN)*int64(db.Info().PageSize))) - b.readTx.Unlock() - b.mu.Unlock() - b.batchTx.Unlock() - - took := time.Since(now) - defragSec.Observe(took.Seconds()) - - size2, sizeInUse2 := b.Size(), b.SizeInUse() - b.lg.Info("finished defragmenting directory", - zap.String("path", dbp), - zap.Int64("current-db-size-bytes-diff", size2-size1), - zap.Int64("current-db-size-bytes", size2), - zap.String("current-db-size", humanize.Bytes(uint64(size2))), - zap.Int64("current-db-size-in-use-bytes-diff", sizeInUse2-sizeInUse1), - zap.Int64("current-db-size-in-use-bytes", sizeInUse2), - zap.String("current-db-size-in-use", humanize.Bytes(uint64(sizeInUse2))), - zap.Duration("took", took), - ) return nil } From 7379d1a0cbd1deb132676d420c91b95b61420db2 Mon Sep 17 00:00:00 2001 From: Allen Ray Date: Wed, 24 Jun 2026 16:03:43 -0400 Subject: [PATCH 3/5] backend: restore defragdbFail gofail failpoint Restore the defragdbFail failpoint that was in upstream defragdb() but dropped during the refactor to defragFromTx(). Co-Authored-By: Claude Opus 4.6 (1M context) --- server/storage/backend/backend.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/server/storage/backend/backend.go b/server/storage/backend/backend.go index c6849942cb6e..306d87c8b12b 100644 --- a/server/storage/backend/backend.go +++ b/server/storage/backend/backend.go @@ -656,6 +656,9 @@ func (b *backend) defragReplayAndSwap(journal *defragJournal, tmpdb *bolt.DB, db } func defragFromTx(srcTx *bolt.Tx, tmpdb *bolt.DB, limit int) error { + // gofail: var defragdbFail string + // return fmt.Errorf(defragdbFail) + tmptx, err := tmpdb.Begin(true) if err != nil { return err From 828d7117e38196f1d7593822352a70b2a1abe8c8 Mon Sep 17 00:00:00 2001 From: Allen Ray Date: Fri, 17 Jul 2026 09:03:57 -0400 Subject: [PATCH 4/5] backend: add NonBlockingDefrag feature gate, journal backpressure, and --defrag-journal-max-ops flag Gate the non-blocking defrag behind a new Alpha feature gate (NonBlockingDefrag, default false). When disabled, the legacy blocking defrag path is used. When enabled, the journal-based three-phase defrag allows writes to continue during the bulk copy. Add backpressure to the defrag journal using a sync.Cond so that writers block in LockInsideApply() when the journal reaches the configured limit, avoiding unbounded memory growth. The defrag goroutine acquires the mutex via LockOutsideApply() which bypasses the backpressure check, preventing deadlock. Add --defrag-journal-max-ops CLI flag (default 50000) to configure the backpressure threshold. Set to 0 for unlimited. Thread the config through: CLI flag -> embed.Config -> config.ServerConfig -> backend.BackendConfig -> backend struct --- server/config/config.go | 6 + server/embed/config.go | 7 + server/embed/etcd.go | 1 + server/features/etcd_features.go | 9 + server/storage/backend.go | 5 + server/storage/backend/backend.go | 179 +++++++++++++++--- server/storage/backend/backend_test.go | 87 +++++++++ server/storage/backend/batch_tx.go | 7 + server/storage/backend/defrag_journal.go | 37 +++- server/storage/backend/defrag_journal_test.go | 133 ++++++++++++- server/storage/backend/export_test.go | 10 + 11 files changed, 449 insertions(+), 32 deletions(-) diff --git a/server/config/config.go b/server/config/config.go index 182dec2ae441..84bc214a6426 100644 --- a/server/config/config.go +++ b/server/config/config.go @@ -201,6 +201,12 @@ type ServerConfig struct { // consider running defrag during bootstrap. Needs to be set to non-zero value to take effect. BootstrapDefragThresholdMegabytes uint `json:"bootstrap-defrag-threshold-megabytes"` + // DefragJournalMaxOps sets the maximum number of write operations buffered + // in the non-blocking defrag journal before backpressure is applied. Only + // effective when NonBlockingDefrag feature gate is enabled. 0 means + // unlimited (no backpressure). + DefragJournalMaxOps int `json:"defrag-journal-max-ops"` + // MaxLearners sets a limit to the number of learner members that can exist in the cluster membership. MaxLearners int `json:"max-learners"` diff --git a/server/embed/config.go b/server/embed/config.go index fcb282d32592..07980632a633 100644 --- a/server/embed/config.go +++ b/server/embed/config.go @@ -52,6 +52,7 @@ import ( "go.etcd.io/etcd/server/v3/etcdserver/api/v3compactor" "go.etcd.io/etcd/server/v3/etcdserver/api/v3discovery" "go.etcd.io/etcd/server/v3/features" + "go.etcd.io/etcd/server/v3/storage/backend" ) const ( @@ -459,6 +460,11 @@ type Config struct { ExperimentalBootstrapDefragThresholdMegabytes uint `json:"experimental-bootstrap-defrag-threshold-megabytes"` // BootstrapDefragThresholdMegabytes is the minimum number of megabytes needed to be freed for etcd server to BootstrapDefragThresholdMegabytes uint `json:"bootstrap-defrag-threshold-megabytes"` + // DefragJournalMaxOps sets the maximum number of write operations buffered + // in the defrag journal before backpressure is applied to writers. Only + // effective when the NonBlockingDefrag feature gate is enabled. 0 means + // unlimited (no backpressure). + DefragJournalMaxOps int `json:"defrag-journal-max-ops"` // WarningUnaryRequestDuration is the time duration after which a warning is generated if applying // unary request takes more time than this value. WarningUnaryRequestDuration time.Duration `json:"warning-unary-request-duration"` @@ -950,6 +956,7 @@ func (cfg *Config) AddFlags(fs *flag.FlagSet) { // TODO: delete in v3.7 fs.UintVar(&cfg.ExperimentalBootstrapDefragThresholdMegabytes, "experimental-bootstrap-defrag-threshold-megabytes", 0, "Enable the defrag during etcd server bootstrap on condition that it will free at least the provided threshold of disk space. Needs to be set to non-zero value to take effect. It's deprecated, and will be decommissioned in v3.7. Use --bootstrap-defrag-threshold-megabytes instead.") fs.UintVar(&cfg.BootstrapDefragThresholdMegabytes, "bootstrap-defrag-threshold-megabytes", 0, "Enable the defrag during etcd server bootstrap on condition that it will free at least the provided threshold of disk space. Needs to be set to non-zero value to take effect.") + fs.IntVar(&cfg.DefragJournalMaxOps, "defrag-journal-max-ops", backend.DefaultDefragJournalMaxOps, "Maximum number of write operations buffered in the non-blocking defrag journal before backpressure is applied. Requires --feature-gates=NonBlockingDefrag=true. Set to 0 for unlimited.") // TODO: delete in v3.7 fs.IntVar(&cfg.MaxLearners, "max-learners", membership.DefaultMaxLearners, "Sets the maximum number of learners that can be available in the cluster membership.") fs.Uint64Var(&cfg.ExperimentalSnapshotCatchUpEntries, "experimental-snapshot-catchup-entries", cfg.ExperimentalSnapshotCatchUpEntries, "Number of entries for a slow follower to catch up after compacting the raft storage entries. Deprecated in v3.6 and will be decommissioned in v3.7. Use --snapshot-catchup-entries instead.") diff --git a/server/embed/etcd.go b/server/embed/etcd.go index ef25652022a2..24d4f60371bf 100644 --- a/server/embed/etcd.go +++ b/server/embed/etcd.go @@ -233,6 +233,7 @@ func StartEtcd(inCfg *Config) (e *Etcd, err error) { WarningUnaryRequestDuration: cfg.WarningUnaryRequestDuration, MemoryMlock: cfg.MemoryMlock, BootstrapDefragThresholdMegabytes: cfg.BootstrapDefragThresholdMegabytes, + DefragJournalMaxOps: cfg.DefragJournalMaxOps, MaxLearners: cfg.MaxLearners, V2Deprecation: cfg.V2DeprecationEffective(), ExperimentalLocalAddress: cfg.InferLocalAddr(), diff --git a/server/features/etcd_features.go b/server/features/etcd_features.go index 230b37c703a3..a87fd8a84ade 100644 --- a/server/features/etcd_features.go +++ b/server/features/etcd_features.go @@ -35,6 +35,14 @@ const ( // of code conflicts because changes are more likely to be scattered // across the file. + // NonBlockingDefrag enables a non-blocking defragmentation algorithm that + // allows writes to continue during the bulk data copy phase. A write journal + // captures mutations during the copy; only the journal replay and database + // swap require the write lock, reducing disruption from O(db_size) to + // O(writes_during_copy). + // owner: @dusk125 + // alpha: v3.7 + NonBlockingDefrag featuregate.Feature = "NonBlockingDefrag" // StopGRPCServiceOnDefrag enables etcd gRPC service to stop serving client requests on defragmentation. // owner: @chaochn47 // alpha: v3.6 @@ -78,6 +86,7 @@ const ( var ( DefaultEtcdServerFeatureGates = map[featuregate.Feature]featuregate.FeatureSpec{ + NonBlockingDefrag: {Default: false, PreRelease: featuregate.Alpha}, StopGRPCServiceOnDefrag: {Default: false, PreRelease: featuregate.Alpha}, InitialCorruptCheck: {Default: false, PreRelease: featuregate.Alpha}, CompactHashCheck: {Default: false, PreRelease: featuregate.Alpha}, diff --git a/server/storage/backend.go b/server/storage/backend.go index 7db61f9fae56..e3c037f5f507 100644 --- a/server/storage/backend.go +++ b/server/storage/backend.go @@ -23,6 +23,7 @@ import ( "go.etcd.io/etcd/server/v3/config" "go.etcd.io/etcd/server/v3/etcdserver/api/snap" + "go.etcd.io/etcd/server/v3/features" "go.etcd.io/etcd/server/v3/storage/backend" "go.etcd.io/etcd/server/v3/storage/schema" "go.etcd.io/raft/v3/raftpb" @@ -52,6 +53,10 @@ func newBackend(cfg config.ServerConfig, hooks backend.Hooks) backend.Backend { } bcfg.Mlock = cfg.MemoryMlock bcfg.Hooks = hooks + if cfg.ServerFeatureGate != nil { + bcfg.NonBlockingDefrag = cfg.ServerFeatureGate.Enabled(features.NonBlockingDefrag) + } + bcfg.DefragJournalMaxOps = cfg.DefragJournalMaxOps return backend.New(bcfg) } diff --git a/server/storage/backend/backend.go b/server/storage/backend/backend.go index 306d87c8b12b..47fa41934fea 100644 --- a/server/storage/backend/backend.go +++ b/server/storage/backend/backend.go @@ -129,6 +129,12 @@ type backend struct { // txPostLockInsideApplyHook is called each time right after locking the tx. txPostLockInsideApplyHook func() + // nonBlockingDefrag enables the journal-based non-blocking defrag path. + nonBlockingDefrag bool + // defragJournalMaxOps is the journal backpressure limit. + // 0 means unlimited (no backpressure). + defragJournalMaxOps int + lg *zap.Logger } @@ -152,6 +158,14 @@ type BackendConfig struct { // Hooks are getting executed during lifecycle of Backend's transactions. Hooks Hooks + + // NonBlockingDefrag enables the journal-based non-blocking defrag + // algorithm. When false, the legacy blocking defrag is used. + NonBlockingDefrag bool + // DefragJournalMaxOps is the backpressure threshold for the defrag + // journal. Writers block when the journal reaches this size. + // 0 means unlimited (no backpressure). + DefragJournalMaxOps int } type BackendConfigOption func(*BackendConfig) @@ -237,6 +251,9 @@ func newBackend(bcfg BackendConfig) *backend { stopc: make(chan struct{}), donec: make(chan struct{}), + nonBlockingDefrag: bcfg.NonBlockingDefrag, + defragJournalMaxOps: bcfg.DefragJournalMaxOps, + lg: bcfg.Logger, } @@ -462,21 +479,19 @@ func (b *backend) Commits() int64 { } func (b *backend) Defrag() error { - return b.defrag() + if b.nonBlockingDefrag { + return b.defragNonBlocking() + } + return b.defragBlocking() } -func (b *backend) defrag() error { - verify.Assert(b.lg != nil, "the logger should not be nil") - now := time.Now() - isDefragActive.Set(1) - defer isDefragActive.Set(0) - - // Create a temporary file to ensure we start with a clean slate. - // Snapshotter.cleanupSnapdir cleans up any of these that are found during startup. +// defragOpenTmpDB creates a temporary bbolt database in the same +// directory as the current database file for use during defrag. +func (b *backend) defragOpenTmpDB() (tmpdb *bolt.DB, tdbp string, err error) { dir := filepath.Dir(b.db.Path()) temp, err := os.CreateTemp(dir, "db.tmp.*") if err != nil { - return err + return nil, "", err } options := bolt.Options{} @@ -489,8 +504,8 @@ func (b *backend) defrag() error { return temp, nil } options.Mlock = false - tdbp := temp.Name() - tmpdb, err := bolt.Open(tdbp, 0o600, &options) + tdbp = temp.Name() + tmpdb, err = bolt.Open(tdbp, 0o600, &options) if err != nil { temp.Close() if rmErr := os.Remove(temp.Name()); rmErr != nil { @@ -499,9 +514,12 @@ func (b *backend) defrag() error { zap.Error(rmErr), ) } - return err + return nil, "", err } + return tmpdb, tdbp, nil +} +func (b *backend) defragLogStart() (string, int64, int64) { dbp := b.db.Path() size1, sizeInUse1 := b.Size(), b.SizeInUse() b.lg.Info("defragmenting", @@ -511,6 +529,126 @@ func (b *backend) defrag() error { zap.Int64("current-db-size-in-use-bytes", sizeInUse1), zap.String("current-db-size-in-use", humanize.Bytes(uint64(sizeInUse1))), ) + return dbp, size1, sizeInUse1 +} + +func (b *backend) defragLogFinish(dbp string, size1, sizeInUse1 int64, took time.Duration) { + size2, sizeInUse2 := b.Size(), b.SizeInUse() + b.lg.Info("finished defragmenting directory", + zap.String("path", dbp), + zap.Int64("current-db-size-bytes-diff", size2-size1), + zap.Int64("current-db-size-bytes", size2), + zap.String("current-db-size", humanize.Bytes(uint64(size2))), + zap.Int64("current-db-size-in-use-bytes-diff", sizeInUse2-sizeInUse1), + zap.Int64("current-db-size-in-use-bytes", sizeInUse2), + zap.String("current-db-size-in-use", humanize.Bytes(uint64(sizeInUse2))), + zap.Duration("took", took), + ) +} + +// defragBlocking is the legacy defrag path that holds the write lock +// for the entire duration of the copy. +func (b *backend) defragBlocking() error { + verify.Assert(b.lg != nil, "the logger should not be nil") + now := time.Now() + isDefragActive.Set(1) + defer isDefragActive.Set(0) + + b.batchTx.LockOutsideApply() + defer b.batchTx.Unlock() + + b.mu.Lock() + defer b.mu.Unlock() + + b.readTx.Lock() + defer b.readTx.Unlock() + + tmpdb, tdbp, err := b.defragOpenTmpDB() + if err != nil { + return err + } + + dbp, size1, sizeInUse1 := b.defragLogStart() + + defer func() { + if rerr := recover(); rerr != nil { + b.lg.Fatal("unexpected panic during defrag", zap.Any("panic", rerr)) + } + }() + + b.batchTx.unsafeCommit(true) + b.batchTx.tx = nil + + // Open a read-only transaction for the copy. We can't use + // b.readTx.tx because unsafeCommit(true) rolled it back. + srcTx, err := b.db.Begin(false) + if err != nil { + b.batchTx.tx = b.unsafeBegin(true) + b.readTx.tx = b.unsafeBegin(false) + tmpdb.Close() + os.RemoveAll(tdbp) + return err + } + + // gofail: var defragBeforeCopy struct{} + err = defragFromTx(srcTx, tmpdb, defragLimit) + srcTx.Rollback() + if err != nil { + tmpdb.Close() + if rmErr := os.RemoveAll(tdbp); rmErr != nil { + b.lg.Error("failed to remove db.tmp after defragmentation completed", zap.Error(rmErr)) + } + b.batchTx.tx = b.unsafeBegin(true) + b.readTx.tx = b.unsafeBegin(false) + return err + } + + if err = b.db.Close(); err != nil { + b.lg.Fatal("failed to close database", zap.Error(err)) + } + if err = tmpdb.Close(); err != nil { + b.lg.Fatal("failed to close tmp database", zap.Error(err)) + } + // gofail: var defragBeforeRename struct{} + if err = os.Rename(tdbp, dbp); err != nil { + b.lg.Fatal("failed to rename tmp database", zap.Error(err)) + } + + b.db, err = bolt.Open(dbp, 0o600, b.bopts) + if err != nil { + b.lg.Fatal("failed to open database", zap.String("path", dbp), zap.Error(err)) + } + b.batchTx.tx = b.unsafeBegin(true) + + b.readTx.reset() + b.readTx.tx = b.unsafeBegin(false) + + size := b.readTx.tx.Size() + db := b.readTx.tx.DB() + atomic.StoreInt64(&b.size, size) + atomic.StoreInt64(&b.sizeInUse, size-(int64(db.Stats().FreePageN)*int64(db.Info().PageSize))) + + took := time.Since(now) + defragSec.Observe(took.Seconds()) + b.defragLogFinish(dbp, size1, sizeInUse1, took) + return nil +} + +// defragNonBlocking uses a journal to capture writes during the copy +// phase, allowing writes to continue unblocked. Only the journal +// replay and database swap hold the write lock. +func (b *backend) defragNonBlocking() error { + verify.Assert(b.lg != nil, "the logger should not be nil") + now := time.Now() + isDefragActive.Set(1) + defer isDefragActive.Set(0) + + tmpdb, tdbp, err := b.defragOpenTmpDB() + if err != nil { + return err + } + + dbp, size1, sizeInUse1 := b.defragLogStart() journal, snapTx, err := b.defragSetupSnapshot() if err != nil { @@ -540,18 +678,7 @@ func (b *backend) defrag() error { took := time.Since(now) defragSec.Observe(took.Seconds()) - - size2, sizeInUse2 := b.Size(), b.SizeInUse() - b.lg.Info("finished defragmenting directory", - zap.String("path", dbp), - zap.Int64("current-db-size-bytes-diff", size2-size1), - zap.Int64("current-db-size-bytes", size2), - zap.String("current-db-size", humanize.Bytes(uint64(size2))), - zap.Int64("current-db-size-in-use-bytes-diff", sizeInUse2-sizeInUse1), - zap.Int64("current-db-size-in-use-bytes", sizeInUse2), - zap.String("current-db-size-in-use", humanize.Bytes(uint64(sizeInUse2))), - zap.Duration("took", took), - ) + b.defragLogFinish(dbp, size1, sizeInUse1, took) return nil } @@ -570,7 +697,7 @@ func (b *backend) defragSetupSnapshot() (*defragJournal, *bolt.Tx, error) { return nil, nil, fmt.Errorf("failed to begin snapshot tx for defrag: %w", err) } - journal := newDefragJournal() + journal := newDefragJournal(b.defragJournalMaxOps) b.batchTx.defragJournal = journal return journal, snapTx, nil } diff --git a/server/storage/backend/backend_test.go b/server/storage/backend/backend_test.go index 2ccbadb0b4e3..08066c0c36d0 100644 --- a/server/storage/backend/backend_test.go +++ b/server/storage/backend/backend_test.go @@ -355,6 +355,7 @@ func TestBackendWritebackForEach(t *testing.T) { func TestBackendDefragConcurrentWrites(t *testing.T) { b, _ := betesting.NewDefaultTmpBackend(t) defer betesting.Close(t, b) + backend.SetNonBlockingDefragForTest(b, true) largeVal := make([]byte, 1024) tx := b.BatchTx() @@ -442,6 +443,7 @@ func TestBackendDefragConcurrentWrites(t *testing.T) { func TestBackendDefragConcurrentReads(t *testing.T) { b, _ := betesting.NewDefaultTmpBackend(t) defer betesting.Close(t, b) + backend.SetNonBlockingDefragForTest(b, true) tx := b.BatchTx() tx.Lock() @@ -490,6 +492,7 @@ func TestBackendDefragConcurrentReads(t *testing.T) { func TestBackendDefragWriteAvailability(t *testing.T) { b, _ := betesting.NewDefaultTmpBackend(t) defer betesting.Close(t, b) + backend.SetNonBlockingDefragForTest(b, true) tx := b.BatchTx() tx.Lock() @@ -554,6 +557,7 @@ func TestBackendDefragWriteAvailability(t *testing.T) { func TestBackendDefragConcurrentReadsAndWrites(t *testing.T) { b, _ := betesting.NewDefaultTmpBackend(t) defer betesting.Close(t, b) + backend.SetNonBlockingDefragForTest(b, true) tx := b.BatchTx() tx.Lock() @@ -622,6 +626,7 @@ func TestBackendDefragConcurrentReadsAndWrites(t *testing.T) { func TestBackendDefragDeletesDuringDefrag(t *testing.T) { b, _ := betesting.NewDefaultTmpBackend(t) defer betesting.Close(t, b) + backend.SetNonBlockingDefragForTest(b, true) tx := b.BatchTx() tx.Lock() @@ -672,6 +677,7 @@ func TestBackendDefragDeletesDuringDefrag(t *testing.T) { func TestBackendDefragLogicalConsistency(t *testing.T) { b, _ := betesting.NewDefaultTmpBackend(t) defer betesting.Close(t, b) + backend.SetNonBlockingDefragForTest(b, true) tx := b.BatchTx() tx.Lock() @@ -736,6 +742,7 @@ func TestBackendDefragLogicalConsistency(t *testing.T) { func TestBackendDefragOverwriteDuringCopy(t *testing.T) { b, _ := betesting.NewDefaultTmpBackend(t) defer betesting.Close(t, b) + backend.SetNonBlockingDefragForTest(b, true) tx := b.BatchTx() tx.Lock() @@ -794,6 +801,7 @@ func TestBackendDefragOverwriteDuringCopy(t *testing.T) { func TestBackendDefragNewBucketDuringCopy(t *testing.T) { b, _ := betesting.NewDefaultTmpBackend(t) defer betesting.Close(t, b) + backend.SetNonBlockingDefragForTest(b, true) tx := b.BatchTx() tx.Lock() @@ -846,6 +854,7 @@ func TestBackendDefragNewBucketDuringCopy(t *testing.T) { func TestBackendDefragMultipleSequential(t *testing.T) { b, _ := betesting.NewDefaultTmpBackend(t) defer betesting.Close(t, b) + backend.SetNonBlockingDefragForTest(b, true) tx := b.BatchTx() tx.Lock() @@ -900,6 +909,7 @@ func TestBackendDefragMultipleSequential(t *testing.T) { func TestBackendDefragEmptyDatabase(t *testing.T) { b, _ := betesting.NewDefaultTmpBackend(t) defer betesting.Close(t, b) + backend.SetNonBlockingDefragForTest(b, true) err := b.Defrag() require.NoError(t, err) @@ -923,6 +933,7 @@ func TestBackendDefragEmptyDatabase(t *testing.T) { func TestBackendDefragLargeJournalReplay(t *testing.T) { b, _ := betesting.NewDefaultTmpBackend(t) defer betesting.Close(t, b) + backend.SetNonBlockingDefragForTest(b, true) tx := b.BatchTx() tx.Lock() @@ -996,6 +1007,7 @@ func TestBackendDefragLargeJournalReplay(t *testing.T) { func TestBackendDefragReadsDuringDefragSeeBufferedPuts(t *testing.T) { b, _ := betesting.NewDefaultTmpBackend(t) defer betesting.Close(t, b) + backend.SetNonBlockingDefragForTest(b, true) tx := b.BatchTx() tx.Lock() @@ -1062,6 +1074,7 @@ func TestBackendDefragExceedBatchLimit(t *testing.T) { // Use a small batch limit so we can exceed it without writing too many keys. b, _ := betesting.NewTmpBackend(t, time.Hour, 100) defer betesting.Close(t, b) + backend.SetNonBlockingDefragForTest(b, true) largeVal := make([]byte, 1024) tx := b.BatchTx() @@ -1127,6 +1140,7 @@ func TestBackendDefragExceedBatchLimit(t *testing.T) { func TestBackendDefragCommitDuringDefragPreservesData(t *testing.T) { b, _ := betesting.NewDefaultTmpBackend(t) defer betesting.Close(t, b) + backend.SetNonBlockingDefragForTest(b, true) tx := b.BatchTx() tx.Lock() @@ -1199,6 +1213,7 @@ func TestBackendDefragCommitDuringDefragPreservesData(t *testing.T) { func TestBackendDefragConcurrentReadTxSurvivesSwitchover(t *testing.T) { b, _ := betesting.NewDefaultTmpBackend(t) defer betesting.Close(t, b) + backend.SetNonBlockingDefragForTest(b, true) tx := b.BatchTx() tx.Lock() @@ -1260,6 +1275,7 @@ func TestBackendDefragConcurrentReadTxSurvivesSwitchover(t *testing.T) { func TestBackendDefragNoDeadlock(t *testing.T) { b, _ := betesting.NewDefaultTmpBackend(t) defer betesting.Close(t, b) + backend.SetNonBlockingDefragForTest(b, true) tx := b.BatchTx() tx.Lock() @@ -1343,6 +1359,7 @@ func TestBackendDefragNoDeadlock(t *testing.T) { func TestBackendDefragMultipleConcurrentReadTxDuringSwitchover(t *testing.T) { b, _ := betesting.NewDefaultTmpBackend(t) defer betesting.Close(t, b) + backend.SetNonBlockingDefragForTest(b, true) tx := b.BatchTx() tx.Lock() @@ -1416,6 +1433,7 @@ func TestBackendDefragMultipleConcurrentReadTxDuringSwitchover(t *testing.T) { func TestBackendDefragWriteVisibilityDuringCopy(t *testing.T) { b, _ := betesting.NewDefaultTmpBackend(t) defer betesting.Close(t, b) + backend.SetNonBlockingDefragForTest(b, true) tx := b.BatchTx() tx.Lock() @@ -1467,3 +1485,72 @@ func TestBackendDefragWriteVisibilityDuringCopy(t *testing.T) { } tx.Unlock() } + +// TestBackendDefragBackpressure verifies that a small journal limit +// causes writers to be throttled (not deadlocked) during defrag, and +// that all data is consistent after defrag completes. +func TestBackendDefragBackpressure(t *testing.T) { + b, _ := betesting.NewDefaultTmpBackend(t) + defer betesting.Close(t, b) + backend.SetNonBlockingDefragForTest(b, true) + + // Set a very small journal limit to force backpressure during defrag. + backend.SetDefragJournalMaxOpsForTest(b, 50) + + tx := b.BatchTx() + tx.Lock() + tx.UnsafeCreateBucket(schema.Test) + // Pre-populate enough data to make the copy phase take time. + for i := 0; i < 5000; i++ { + tx.UnsafePut(schema.Test, []byte(fmt.Sprintf("pre_%05d", i)), make([]byte, 256)) + } + tx.Unlock() + b.ForceCommit() + + // Concurrent writer that will exceed the journal limit many times over. + var wg sync.WaitGroup + stop := make(chan struct{}) + var written atomic.Int32 + + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; ; i++ { + select { + case <-stop: + return + default: + } + wtx := b.BatchTx() + wtx.Lock() + wtx.UnsafePut(schema.Test, []byte(fmt.Sprintf("bp_%06d", i)), []byte("val")) + wtx.Unlock() + written.Add(1) + } + }() + + // Defrag must complete without deadlocking. + err := b.Defrag() + close(stop) + wg.Wait() + + require.NoError(t, err) + totalWritten := int(written.Load()) + require.Greater(t, totalWritten, 50, + "writer should have produced more ops than the journal limit") + t.Logf("wrote %d ops with journal limit 50", totalWritten) + + // Verify all pre-existing and concurrent writes survived. + b.ForceCommit() + rtx := b.BatchTx() + rtx.Lock() + for i := 0; i < 5000; i++ { + keys, _ := rtx.UnsafeRange(schema.Test, []byte(fmt.Sprintf("pre_%05d", i)), nil, 0) + require.Lenf(t, keys, 1, "pre_%05d must survive defrag", i) + } + for i := 0; i < totalWritten; i++ { + keys, _ := rtx.UnsafeRange(schema.Test, []byte(fmt.Sprintf("bp_%06d", i)), nil, 0) + require.Lenf(t, keys, 1, "bp_%06d must survive defrag", i) + } + rtx.Unlock() +} diff --git a/server/storage/backend/batch_tx.go b/server/storage/backend/batch_tx.go index 26b2d857609f..ad4c1ce62752 100644 --- a/server/storage/backend/batch_tx.go +++ b/server/storage/backend/batch_tx.go @@ -306,6 +306,13 @@ func newBatchTxBuffered(backend *backend) *batchTxBuffered { return tx } +func (t *batchTxBuffered) LockInsideApply() { + if j := t.defragJournal; j != nil { + j.waitForSpace() + } + t.batchTx.LockInsideApply() +} + func (t *batchTxBuffered) Unlock() { if t.pending != 0 { t.backend.readTx.Lock() // blocks txReadBuffer for writing. diff --git a/server/storage/backend/defrag_journal.go b/server/storage/backend/defrag_journal.go index 940891b7df25..7a0b91a092fe 100644 --- a/server/storage/backend/defrag_journal.go +++ b/server/storage/backend/defrag_journal.go @@ -25,6 +25,13 @@ const ( opDeleteBucket ) +// DefaultDefragJournalMaxOps is the default maximum number of operations +// the journal will buffer before applying backpressure to writers. +// At ~700 bytes per op this caps memory at roughly 35 MB and bounds +// Phase 3 replay to well under a second. The CLI flag +// --defrag-journal-max-ops defaults to this value. +const DefaultDefragJournalMaxOps = 50_000 + type defragJournalOp struct { opType defragOpType bucketName []byte @@ -35,13 +42,34 @@ type defragJournalOp struct { type defragJournal struct { mu sync.Mutex + cond *sync.Cond ops []defragJournalOp closed bool + maxOps int +} + +// newDefragJournal creates a journal that buffers write operations +// during the defrag copy phase. maxOps sets the backpressure threshold: +// writers block when the journal reaches this size. A value of 0 means +// no limit. +func newDefragJournal(maxOps int) *defragJournal { + j := &defragJournal{ + ops: make([]defragJournalOp, 0, 1024), + maxOps: maxOps, + } + j.cond = sync.NewCond(&j.mu) + return j } -func newDefragJournal() *defragJournal { - return &defragJournal{ - ops: make([]defragJournalOp, 0, 1024), +// waitForSpace blocks until the journal has room for more operations +// or is closed. Must be called BEFORE acquiring the batchTx mutex to +// avoid deadlock: writers wait here without holding the mutex, so +// the defrag goroutine can still acquire the mutex to drain the journal. +func (j *defragJournal) waitForSpace() { + j.mu.Lock() + defer j.mu.Unlock() + for j.maxOps > 0 && len(j.ops) >= j.maxOps && !j.closed { + j.cond.Wait() } } @@ -98,11 +126,13 @@ func (j *defragJournal) appendDeleteBucket(bucketName []byte) { } // drain returns all accumulated ops and resets the journal. +// It broadcasts to wake any writers blocked in waitForSpace. func (j *defragJournal) drain() []defragJournalOp { j.mu.Lock() defer j.mu.Unlock() ops := j.ops j.ops = nil + j.cond.Broadcast() return ops } @@ -110,6 +140,7 @@ func (j *defragJournal) close() { j.mu.Lock() defer j.mu.Unlock() j.closed = true + j.cond.Broadcast() } func cloneBytes(b []byte) []byte { diff --git a/server/storage/backend/defrag_journal_test.go b/server/storage/backend/defrag_journal_test.go index b5e8ccae3c86..3a670436cccd 100644 --- a/server/storage/backend/defrag_journal_test.go +++ b/server/storage/backend/defrag_journal_test.go @@ -15,14 +15,16 @@ package backend import ( + "sync" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) func TestDefragJournalAppendAndDrain(t *testing.T) { - j := newDefragJournal() + j := newDefragJournal(0) j.appendCreateBucket([]byte("bucket1")) j.appendPut([]byte("bucket1"), []byte("key1"), []byte("val1"), false) @@ -57,7 +59,7 @@ func TestDefragJournalAppendAndDrain(t *testing.T) { } func TestDefragJournalClose(t *testing.T) { - j := newDefragJournal() + j := newDefragJournal(0) j.appendPut([]byte("b"), []byte("k"), []byte("v"), false) j.close() @@ -73,8 +75,133 @@ func TestDefragJournalClose(t *testing.T) { assert.Equal(t, []byte("k"), ops[0].key) } +func TestDefragJournalBackpressureBlocksWhenFull(t *testing.T) { + j := newDefragJournal(5) + + // Fill to capacity. + for i := 0; i < 5; i++ { + j.appendPut([]byte("b"), []byte("k"), []byte("v"), false) + } + require.Len(t, j.drain(), 5) + + // Fill again. + for i := 0; i < 5; i++ { + j.appendPut([]byte("b"), []byte("k"), []byte("v"), false) + } + + // waitForSpace should block because the journal is at capacity. + unblocked := make(chan struct{}) + go func() { + j.waitForSpace() + close(unblocked) + }() + + select { + case <-unblocked: + t.Fatal("waitForSpace returned while journal is full") + case <-time.After(50 * time.Millisecond): + } + + // Drain frees space and wakes the blocked goroutine. + ops := j.drain() + assert.Len(t, ops, 5) + + select { + case <-unblocked: + case <-time.After(time.Second): + t.Fatal("waitForSpace did not unblock after drain") + } +} + +func TestDefragJournalBackpressureUnblocksOnClose(t *testing.T) { + j := newDefragJournal(3) + + for i := 0; i < 3; i++ { + j.appendPut([]byte("b"), []byte("k"), []byte("v"), false) + } + + unblocked := make(chan struct{}) + go func() { + j.waitForSpace() + close(unblocked) + }() + + select { + case <-unblocked: + t.Fatal("waitForSpace returned while journal is full") + case <-time.After(50 * time.Millisecond): + } + + // Close wakes blocked writers even without draining. + j.close() + + select { + case <-unblocked: + case <-time.After(time.Second): + t.Fatal("waitForSpace did not unblock after close") + } +} + +func TestDefragJournalBackpressureMultipleWriters(t *testing.T) { + j := newDefragJournal(2) + + for i := 0; i < 2; i++ { + j.appendPut([]byte("b"), []byte("k"), []byte("v"), false) + } + + // Block two writers. + var wg sync.WaitGroup + wg.Add(2) + for i := 0; i < 2; i++ { + go func() { + defer wg.Done() + j.waitForSpace() + }() + } + + // Give goroutines time to block. + time.Sleep(50 * time.Millisecond) + + // Drain should wake both. + j.drain() + + done := make(chan struct{}) + go func() { + wg.Wait() + close(done) + }() + + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("not all writers unblocked after drain") + } +} + +func TestDefragJournalBackpressureDisabledWhenZero(t *testing.T) { + j := newDefragJournal(0) + + // Fill well beyond what would normally be a limit. + for i := 0; i < 100; i++ { + j.appendPut([]byte("b"), []byte("k"), []byte("v"), false) + } + + // waitForSpace should return immediately when maxOps is 0 (unlimited). + done := make(chan struct{}) + go func() { + j.waitForSpace() + close(done) + }() + + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("waitForSpace blocked with maxOps=0") + } +} + func TestDefragJournalDeepCopies(t *testing.T) { - j := newDefragJournal() + j := newDefragJournal(0) key := []byte("original") j.appendPut([]byte("b"), key, []byte("v"), false) diff --git a/server/storage/backend/export_test.go b/server/storage/backend/export_test.go index e9f5ad38d6a1..30435ad8a6aa 100644 --- a/server/storage/backend/export_test.go +++ b/server/storage/backend/export_test.go @@ -27,3 +27,13 @@ func DefragLimitForTest() int { func CommitsForTest(b Backend) int64 { return b.(*backend).Commits() } + +func SetDefragJournalMaxOpsForTest(b Backend, maxOps int) { + be := b.(*backend) + be.defragJournalMaxOps = maxOps +} + +func SetNonBlockingDefragForTest(b Backend, enabled bool) { + be := b.(*backend) + be.nonBlockingDefrag = enabled +} From 476ee855724be3ade02e06717ca48f619e1ae3b8 Mon Sep 17 00:00:00 2001 From: Allen Ray Date: Fri, 17 Jul 2026 22:38:56 -0400 Subject: [PATCH 5/5] backend: address PR review comments and add defrag tests Address review feedback from Thomas, Filip, and Ben: - Rename defragSetupSnapshot to defragPrepare; 'snapshot' is overloaded in etcd (raft snapshots) - Drop FillPercent=0.9 from replayJournal; no meaningful effect on the small number of journal replay ops in a freshly compacted db - Error on missing bucket in replayJournal delete path, consistent with put path - Panic on append-after-close instead of silently dropping ops; this path is unreachable in normal operation - Combine close() and drain() into closeAndDrain() to eliminate the two-step state machine - Use atomic.Pointer[defragJournal] to fix data race between LockInsideApply (reads before mutex) and defrag lifecycle (writes under mutex); use Swap(nil) for atomic load-and-clear - Load defragJournal from batchTx in cancel/replay instead of passing as parameter; defer defragCancelJournal for cleanup - Drop bytes.Clone from journal appends; callers provide Go heap slices same as txWriteBuffer which doesn't clone - Strip copyright headers from new files - Consistent os.Remove error handling across all defrag error paths - Add defragBeforeReplay gofail failpoint between copy and replay - Add defragCopyFailHook for test injection of copy failures Tests: - E2E: TestNonBlockingDefragNoSpace and TestNonBlockingDefragSuccess - Unit: TestBackendDefragConcurrentLockInsideApply (race detector), TestBackendDefragCopyFailurePreservesData (no data loss on copy failure), TestDefragJournalAppendAfterClosePanics --- server/embed/config.go | 1 + server/etcdmain/help.go | 2 + server/storage/backend/backend.go | 79 ++++++---- server/storage/backend/backend_test.go | 140 ++++++++++++++++++ server/storage/backend/batch_tx.go | 18 ++- server/storage/backend/defrag_journal.go | 57 ++----- server/storage/backend/defrag_journal_test.go | 73 +++------ server/storage/backend/export_test.go | 5 + tests/e2e/non_blocking_defrag_test.go | 121 +++++++++++++++ tests/robustness/failpoint/failpoint.go | 2 +- tests/robustness/failpoint/gofail.go | 1 + 11 files changed, 367 insertions(+), 132 deletions(-) create mode 100644 tests/e2e/non_blocking_defrag_test.go diff --git a/server/embed/config.go b/server/embed/config.go index 07980632a633..ed082b511118 100644 --- a/server/embed/config.go +++ b/server/embed/config.go @@ -718,6 +718,7 @@ func NewConfig() *Config { ExperimentalMemoryMlock: false, ExperimentalStopGRPCServiceOnDefrag: false, MaxLearners: membership.DefaultMaxLearners, + DefragJournalMaxOps: backend.DefaultDefragJournalMaxOps, ExperimentalTxnModeWriteWithSharedBuffer: DefaultExperimentalTxnModeWriteWithSharedBuffer, ExperimentalDistributedTracingAddress: DefaultDistributedTracingAddress, diff --git a/server/etcdmain/help.go b/server/etcdmain/help.go index 558ddb23e2d9..41063a2ffbbf 100644 --- a/server/etcdmain/help.go +++ b/server/etcdmain/help.go @@ -319,6 +319,8 @@ Experimental feature: Enable the defrag during etcd server bootstrap on condition that it will free at least the provided threshold of disk space. Needs to be set to non-zero value to take effect. Deprecated in v3.6 and will be decommissioned in v3.7. Use '--bootstrap-defrag-threshold-megabytes' instead. --bootstrap-defrag-threshold-megabytes Enable the defrag during etcd server bootstrap on condition that it will free at least the provided threshold of disk space. Needs to be set to non-zero value to take effect. + --defrag-journal-max-ops '50000' + Maximum number of write operations buffered in the non-blocking defrag journal before backpressure is applied. Requires --feature-gates=NonBlockingDefrag=true. Set to 0 for unlimited. --experimental-warning-unary-request-duration '300ms' Set time duration after which a warning is generated if a unary request takes more than this duration. Deprecated in v3.6 and will be decommissioned in v3.7. Use '--warning-unary-request-duration' instead. --max-learners '1' diff --git a/server/storage/backend/backend.go b/server/storage/backend/backend.go index 47fa41934fea..19fac4159eb5 100644 --- a/server/storage/backend/backend.go +++ b/server/storage/backend/backend.go @@ -135,6 +135,10 @@ type backend struct { // 0 means unlimited (no backpressure). defragJournalMaxOps int + // defragCopyFailHook, if non-nil, is called instead of defragFromTx + // during the copy phase. Used by tests to simulate copy failures. + defragCopyFailHook func() error + lg *zap.Logger } @@ -570,6 +574,10 @@ func (b *backend) defragBlocking() error { dbp, size1, sizeInUse1 := b.defragLogStart() + // NOTE: We should exit as soon as possible because that tx + // might be closed. The inflight request might use invalid + // tx and then panic as well. The real panic reason might be + // shadowed by new panic. So, we should fatal here with lock. defer func() { if rerr := recover(); rerr != nil { b.lg.Fatal("unexpected panic during defrag", zap.Any("panic", rerr)) @@ -586,7 +594,9 @@ func (b *backend) defragBlocking() error { b.batchTx.tx = b.unsafeBegin(true) b.readTx.tx = b.unsafeBegin(false) tmpdb.Close() - os.RemoveAll(tdbp) + if rmErr := os.RemoveAll(tdbp); rmErr != nil { + b.lg.Error("failed to remove tmp database", zap.String("path", tdbp), zap.Error(rmErr)) + } return err } @@ -596,7 +606,7 @@ func (b *backend) defragBlocking() error { if err != nil { tmpdb.Close() if rmErr := os.RemoveAll(tdbp); rmErr != nil { - b.lg.Error("failed to remove db.tmp after defragmentation completed", zap.Error(rmErr)) + b.lg.Error("failed to remove tmp database", zap.String("path", tdbp), zap.Error(rmErr)) } b.batchTx.tx = b.unsafeBegin(true) b.readTx.tx = b.unsafeBegin(false) @@ -650,28 +660,37 @@ func (b *backend) defragNonBlocking() error { dbp, size1, sizeInUse1 := b.defragLogStart() - journal, snapTx, err := b.defragSetupSnapshot() + readTx, err := b.defragPrepare() if err != nil { tmpdb.Close() - os.Remove(tdbp) + if rmErr := os.Remove(tdbp); rmErr != nil { + b.lg.Error("failed to remove tmp database", zap.String("path", tdbp), zap.Error(rmErr)) + } return err } + defer b.defragCancelJournal() b.lg.Info("defrag: copying data (writes unlocked)") // gofail: var defragBeforeCopy struct{} - err = defragFromTx(snapTx, tmpdb, defragLimit) - snapTx.Rollback() + if b.defragCopyFailHook != nil { + err = b.defragCopyFailHook() + } else { + err = defragFromTx(readTx, tmpdb, defragLimit) + } + readTx.Rollback() if err != nil { - b.defragCancelJournal(journal) tmpdb.Close() - os.RemoveAll(tdbp) + if rmErr := os.RemoveAll(tdbp); rmErr != nil { + b.lg.Error("failed to remove tmp database", zap.String("path", tdbp), zap.Error(rmErr)) + } return err } + // gofail: var defragBeforeReplay struct{} b.lg.Info("defrag: replaying journal") - err = b.defragReplayAndSwap(journal, tmpdb, dbp, tdbp) + err = b.defragReplayAndSwap(tmpdb, dbp, tdbp) if err != nil { return err } @@ -682,46 +701,47 @@ func (b *backend) defragNonBlocking() error { return nil } -// defragSetupSnapshot commits pending writes, takes a read-only -// snapshot, and installs a journal to capture writes during the copy. -func (b *backend) defragSetupSnapshot() (*defragJournal, *bolt.Tx, error) { +// defragPrepare commits pending writes, opens a read-only bbolt +// transaction for the copy phase, and installs a journal to capture +// writes that occur during the copy. +func (b *backend) defragPrepare() (*bolt.Tx, error) { b.batchTx.LockOutsideApply() defer b.batchTx.Unlock() b.batchTx.commit(false) b.mu.RLock() - snapTx, err := b.db.Begin(false) + readTx, err := b.db.Begin(false) b.mu.RUnlock() if err != nil { - return nil, nil, fmt.Errorf("failed to begin snapshot tx for defrag: %w", err) + return nil, fmt.Errorf("failed to begin read tx for defrag: %w", err) } - journal := newDefragJournal(b.defragJournalMaxOps) - b.batchTx.defragJournal = journal - return journal, snapTx, nil + b.batchTx.defragJournal.Store(newDefragJournal(b.defragJournalMaxOps)) + return readTx, nil } // defragCancelJournal removes the journal from the batch transaction -// and closes it. Called when the copy phase fails. -func (b *backend) defragCancelJournal(journal *defragJournal) { +// and closes it. Safe to call even if the journal was already +// drained or never installed. +func (b *backend) defragCancelJournal() { b.batchTx.LockOutsideApply() defer b.batchTx.Unlock() - b.batchTx.defragJournal = nil - journal.close() + if journal := b.batchTx.defragJournal.Swap(nil); journal != nil { + journal.closeAndDrain() + } } // defragReplayAndSwap drains the journal, replays it into the temp // database, then atomically swaps the old database for the new one. // batchTx is held for the entire operation to prevent writes between // journal drain and database swap. -func (b *backend) defragReplayAndSwap(journal *defragJournal, tmpdb *bolt.DB, dbp, tdbp string) error { +func (b *backend) defragReplayAndSwap(tmpdb *bolt.DB, dbp, tdbp string) error { b.batchTx.LockOutsideApply() defer b.batchTx.Unlock() - b.batchTx.defragJournal = nil - journal.close() - ops := journal.drain() + journal := b.batchTx.defragJournal.Swap(nil) + ops := journal.closeAndDrain() if len(ops) > 0 { b.lg.Info("defrag: replaying journal ops", zap.Int("count", len(ops))) @@ -729,7 +749,9 @@ func (b *backend) defragReplayAndSwap(journal *defragJournal, tmpdb *bolt.DB, db if err := replayJournal(tmpdb, ops, defragLimit); err != nil { tmpdb.Close() - os.RemoveAll(tdbp) + if rmErr := os.RemoveAll(tdbp); rmErr != nil { + b.lg.Error("failed to remove tmp database", zap.String("path", tdbp), zap.Error(rmErr)) + } return err } @@ -878,16 +900,13 @@ func replayJournal(tmpdb *bolt.DB, ops []defragJournalOp, limit int) error { if b == nil { return fmt.Errorf("replay: bucket %s not found for put", op.bucketName) } - if op.seq { - b.FillPercent = 0.9 - } if err = b.Put(op.key, op.value); err != nil { return fmt.Errorf("replay: put in bucket %s: %w", op.bucketName, err) } case opDelete: b := tx.Bucket(op.bucketName) if b == nil { - continue + return fmt.Errorf("replay: bucket %s not found for delete", op.bucketName) } if err = b.Delete(op.key); err != nil { return fmt.Errorf("replay: delete from bucket %s: %w", op.bucketName, err) diff --git a/server/storage/backend/backend_test.go b/server/storage/backend/backend_test.go index 08066c0c36d0..6bc18480f485 100644 --- a/server/storage/backend/backend_test.go +++ b/server/storage/backend/backend_test.go @@ -1554,3 +1554,143 @@ func TestBackendDefragBackpressure(t *testing.T) { } rtx.Unlock() } + +// TestBackendDefragConcurrentLockInsideApply exercises the data race +// between LockInsideApply reading t.defragJournal without the mutex +// and defrag setting/clearing it under the mutex. Run with -race to +// verify there is no unsynchronized pointer access. +func TestBackendDefragConcurrentLockInsideApply(t *testing.T) { + b, _ := betesting.NewDefaultTmpBackend(t) + defer betesting.Close(t, b) + backend.SetNonBlockingDefragForTest(b, true) + + tx := b.BatchTx() + tx.Lock() + tx.UnsafeCreateBucket(schema.Test) + for i := 0; i < 1000; i++ { + tx.UnsafePut(schema.Test, []byte(fmt.Sprintf("key_%04d", i)), make([]byte, 256)) + } + tx.Unlock() + b.ForceCommit() + + // Concurrent goroutines calling LockInsideApply while defrag + // sets and clears the defragJournal pointer. + var wg sync.WaitGroup + stop := make(chan struct{}) + + for g := 0; g < 4; g++ { + wg.Add(1) + go func(id int) { + defer wg.Done() + for i := 0; ; i++ { + select { + case <-stop: + return + default: + } + tx := b.BatchTx() + tx.LockInsideApply() + tx.UnsafePut(schema.Test, []byte(fmt.Sprintf("w%d_%06d", id, i)), []byte("v")) + tx.Unlock() + } + }(g) + } + + err := b.Defrag() + close(stop) + wg.Wait() + require.NoError(t, err) +} + +// TestBackendDefragCopyFailurePreservesData proves that when the copy +// phase fails during non-blocking defrag, no data is lost. Writes +// during the copy phase go to both the live database and the journal; +// discarding the journal on failure is safe because the live database +// already has every write. +func TestBackendDefragCopyFailurePreservesData(t *testing.T) { + b, _ := betesting.NewDefaultTmpBackend(t) + defer betesting.Close(t, b) + backend.SetNonBlockingDefragForTest(b, true) + + tx := b.BatchTx() + tx.Lock() + tx.UnsafeCreateBucket(schema.Test) + for i := 0; i < 100; i++ { + tx.UnsafePut(schema.Test, []byte(fmt.Sprintf("pre_%04d", i)), []byte("value")) + } + tx.Unlock() + b.ForceCommit() + + // Inject a copy failure. Writes will still flow into the live + // database and the journal during the (simulated) copy phase. + var wg sync.WaitGroup + stop := make(chan struct{}) + var written atomic.Int32 + + backend.SetDefragCopyFailHookForTest(b, func() error { + // Give the concurrent writer time to produce writes while + // the journal is active. + time.Sleep(50 * time.Millisecond) + return fmt.Errorf("simulated copy failure") + }) + + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; ; i++ { + select { + case <-stop: + return + default: + } + wtx := b.BatchTx() + wtx.Lock() + wtx.UnsafePut(schema.Test, []byte(fmt.Sprintf("during_%06d", i)), []byte("val")) + wtx.Unlock() + written.Add(1) + } + }() + + err := b.Defrag() + close(stop) + wg.Wait() + + // Defrag should have failed. + require.Error(t, err) + require.Contains(t, err.Error(), "simulated copy failure") + + totalWritten := int(written.Load()) + require.Greater(t, totalWritten, 0, + "at least some writes must have occurred during the failed defrag") + t.Logf("wrote %d ops during failed defrag copy", totalWritten) + + // All pre-existing data must survive. + b.ForceCommit() + rtx := b.BatchTx() + rtx.Lock() + for i := 0; i < 100; i++ { + keys, _ := rtx.UnsafeRange(schema.Test, []byte(fmt.Sprintf("pre_%04d", i)), nil, 0) + require.Lenf(t, keys, 1, "pre_%04d must survive failed defrag", i) + } + // All writes during the failed copy must also survive — they went + // to the live database, not just the journal. + for i := 0; i < totalWritten; i++ { + keys, _ := rtx.UnsafeRange(schema.Test, []byte(fmt.Sprintf("during_%06d", i)), nil, 0) + require.Lenf(t, keys, 1, "during_%06d must survive failed defrag", i) + } + rtx.Unlock() + + // The server should remain fully functional after the failed defrag. + tx = b.BatchTx() + tx.Lock() + tx.UnsafePut(schema.Test, []byte("after_failure"), []byte("ok")) + tx.Unlock() + b.ForceCommit() + + rtx = b.BatchTx() + rtx.Lock() + keys, vals := rtx.UnsafeRange(schema.Test, []byte("after_failure"), nil, 0) + rtx.Unlock() + require.Len(t, keys, 1) + require.Equal(t, []byte("ok"), vals[0]) +} diff --git a/server/storage/backend/batch_tx.go b/server/storage/backend/batch_tx.go index ad4c1ce62752..07f22bb9ea1e 100644 --- a/server/storage/backend/batch_tx.go +++ b/server/storage/backend/batch_tx.go @@ -291,7 +291,11 @@ type batchTxBuffered struct { batchTx buf txWriteBuffer pendingDeleteOperations int - defragJournal *defragJournal + // defragJournal captures write operations during the non-blocking + // defrag copy phase. Accessed atomically because LockInsideApply + // reads it before acquiring the batchTx mutex (for backpressure), + // while defragPrepare/defragReplayAndSwap set it under the mutex. + defragJournal atomic.Pointer[defragJournal] } func newBatchTxBuffered(backend *backend) *batchTxBuffered { @@ -307,7 +311,7 @@ func newBatchTxBuffered(backend *backend) *batchTxBuffered { } func (t *batchTxBuffered) LockInsideApply() { - if j := t.defragJournal; j != nil { + if j := t.defragJournal.Load(); j != nil { j.waitForSpace() } t.batchTx.LockInsideApply() @@ -394,14 +398,14 @@ func (t *batchTxBuffered) unsafeCommit(stop bool) { } func (t *batchTxBuffered) UnsafeCreateBucket(bucket Bucket) { - if j := t.defragJournal; j != nil { + if j := t.defragJournal.Load(); j != nil { j.appendCreateBucket(bucket.Name()) } t.batchTx.UnsafeCreateBucket(bucket) } func (t *batchTxBuffered) UnsafePut(bucket Bucket, key []byte, value []byte) { - if j := t.defragJournal; j != nil { + if j := t.defragJournal.Load(); j != nil { j.appendPut(bucket.Name(), key, value, false) } t.batchTx.UnsafePut(bucket, key, value) @@ -409,7 +413,7 @@ func (t *batchTxBuffered) UnsafePut(bucket Bucket, key []byte, value []byte) { } func (t *batchTxBuffered) UnsafeSeqPut(bucket Bucket, key []byte, value []byte) { - if j := t.defragJournal; j != nil { + if j := t.defragJournal.Load(); j != nil { j.appendPut(bucket.Name(), key, value, true) } t.batchTx.UnsafeSeqPut(bucket, key, value) @@ -417,7 +421,7 @@ func (t *batchTxBuffered) UnsafeSeqPut(bucket Bucket, key []byte, value []byte) } func (t *batchTxBuffered) UnsafeDelete(bucketType Bucket, key []byte) { - if j := t.defragJournal; j != nil { + if j := t.defragJournal.Load(); j != nil { j.appendDelete(bucketType.Name(), key) } t.batchTx.UnsafeDelete(bucketType, key) @@ -425,7 +429,7 @@ func (t *batchTxBuffered) UnsafeDelete(bucketType Bucket, key []byte) { } func (t *batchTxBuffered) UnsafeDeleteBucket(bucket Bucket) { - if j := t.defragJournal; j != nil { + if j := t.defragJournal.Load(); j != nil { j.appendDeleteBucket(bucket.Name()) } t.batchTx.UnsafeDeleteBucket(bucket) diff --git a/server/storage/backend/defrag_journal.go b/server/storage/backend/defrag_journal.go index 7a0b91a092fe..83d5c747e314 100644 --- a/server/storage/backend/defrag_journal.go +++ b/server/storage/backend/defrag_journal.go @@ -1,17 +1,3 @@ -// Copyright 2024 The etcd Authors -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - package backend import "sync" @@ -77,13 +63,13 @@ func (j *defragJournal) appendPut(bucketName, key, value []byte, seq bool) { j.mu.Lock() defer j.mu.Unlock() if j.closed { - return + panic("defragJournal: append to closed journal") } j.ops = append(j.ops, defragJournalOp{ opType: opPut, - bucketName: cloneBytes(bucketName), - key: cloneBytes(key), - value: cloneBytes(value), + bucketName: bucketName, + key: key, + value: value, seq: seq, }) } @@ -92,12 +78,12 @@ func (j *defragJournal) appendDelete(bucketName, key []byte) { j.mu.Lock() defer j.mu.Unlock() if j.closed { - return + panic("defragJournal: append to closed journal") } j.ops = append(j.ops, defragJournalOp{ opType: opDelete, - bucketName: cloneBytes(bucketName), - key: cloneBytes(key), + bucketName: bucketName, + key: key, }) } @@ -105,11 +91,11 @@ func (j *defragJournal) appendCreateBucket(bucketName []byte) { j.mu.Lock() defer j.mu.Unlock() if j.closed { - return + panic("defragJournal: append to closed journal") } j.ops = append(j.ops, defragJournalOp{ opType: opCreateBucket, - bucketName: cloneBytes(bucketName), + bucketName: bucketName, }) } @@ -117,37 +103,24 @@ func (j *defragJournal) appendDeleteBucket(bucketName []byte) { j.mu.Lock() defer j.mu.Unlock() if j.closed { - return + panic("defragJournal: append to closed journal") } j.ops = append(j.ops, defragJournalOp{ opType: opDeleteBucket, - bucketName: cloneBytes(bucketName), + bucketName: bucketName, }) } -// drain returns all accumulated ops and resets the journal. -// It broadcasts to wake any writers blocked in waitForSpace. -func (j *defragJournal) drain() []defragJournalOp { +// closeAndDrain marks the journal as closed, wakes any writers +// blocked in waitForSpace, and returns all accumulated ops. +func (j *defragJournal) closeAndDrain() []defragJournalOp { j.mu.Lock() defer j.mu.Unlock() + j.closed = true ops := j.ops j.ops = nil j.cond.Broadcast() return ops } -func (j *defragJournal) close() { - j.mu.Lock() - defer j.mu.Unlock() - j.closed = true - j.cond.Broadcast() -} -func cloneBytes(b []byte) []byte { - if b == nil { - return nil - } - cp := make([]byte, len(b)) - copy(cp, b) - return cp -} diff --git a/server/storage/backend/defrag_journal_test.go b/server/storage/backend/defrag_journal_test.go index 3a670436cccd..c1112dc8f90a 100644 --- a/server/storage/backend/defrag_journal_test.go +++ b/server/storage/backend/defrag_journal_test.go @@ -1,17 +1,3 @@ -// Copyright 2024 The etcd Authors -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - package backend import ( @@ -23,7 +9,7 @@ import ( "github.com/stretchr/testify/require" ) -func TestDefragJournalAppendAndDrain(t *testing.T) { +func TestDefragJournalCloseAndDrain(t *testing.T) { j := newDefragJournal(0) j.appendCreateBucket([]byte("bucket1")) @@ -32,7 +18,7 @@ func TestDefragJournalAppendAndDrain(t *testing.T) { j.appendDelete([]byte("bucket1"), []byte("key1")) j.appendDeleteBucket([]byte("bucket1")) - ops := j.drain() + ops := j.closeAndDrain() require.Len(t, ops, 5) assert.Equal(t, opCreateBucket, ops[0].opType) @@ -53,26 +39,19 @@ func TestDefragJournalAppendAndDrain(t *testing.T) { assert.Equal(t, opDeleteBucket, ops[4].opType) assert.Equal(t, []byte("bucket1"), ops[4].bucketName) - // drain again should be empty - ops = j.drain() + // closeAndDrain again should be empty + ops = j.closeAndDrain() assert.Empty(t, ops) } -func TestDefragJournalClose(t *testing.T) { +func TestDefragJournalAppendAfterClosePanics(t *testing.T) { j := newDefragJournal(0) + j.closeAndDrain() - j.appendPut([]byte("b"), []byte("k"), []byte("v"), false) - j.close() - - // appends after close are no-ops - j.appendPut([]byte("b"), []byte("k2"), []byte("v2"), false) - j.appendDelete([]byte("b"), []byte("k")) - j.appendCreateBucket([]byte("b2")) - j.appendDeleteBucket([]byte("b")) - - ops := j.drain() - require.Len(t, ops, 1) - assert.Equal(t, []byte("k"), ops[0].key) + assert.Panics(t, func() { j.appendPut([]byte("b"), []byte("k"), []byte("v"), false) }) + assert.Panics(t, func() { j.appendDelete([]byte("b"), []byte("k")) }) + assert.Panics(t, func() { j.appendCreateBucket([]byte("b")) }) + assert.Panics(t, func() { j.appendDeleteBucket([]byte("b")) }) } func TestDefragJournalBackpressureBlocksWhenFull(t *testing.T) { @@ -82,7 +61,9 @@ func TestDefragJournalBackpressureBlocksWhenFull(t *testing.T) { for i := 0; i < 5; i++ { j.appendPut([]byte("b"), []byte("k"), []byte("v"), false) } - require.Len(t, j.drain(), 5) + require.Len(t, j.closeAndDrain(), 5) + // Re-create since we just closed it. + j = newDefragJournal(5) // Fill again. for i := 0; i < 5; i++ { @@ -102,14 +83,14 @@ func TestDefragJournalBackpressureBlocksWhenFull(t *testing.T) { case <-time.After(50 * time.Millisecond): } - // Drain frees space and wakes the blocked goroutine. - ops := j.drain() + // closeAndDrain frees space and wakes the blocked goroutine. + ops := j.closeAndDrain() assert.Len(t, ops, 5) select { case <-unblocked: case <-time.After(time.Second): - t.Fatal("waitForSpace did not unblock after drain") + t.Fatal("waitForSpace did not unblock after closeAndDrain") } } @@ -132,13 +113,13 @@ func TestDefragJournalBackpressureUnblocksOnClose(t *testing.T) { case <-time.After(50 * time.Millisecond): } - // Close wakes blocked writers even without draining. - j.close() + // closeAndDrain wakes blocked writers. + j.closeAndDrain() select { case <-unblocked: case <-time.After(time.Second): - t.Fatal("waitForSpace did not unblock after close") + t.Fatal("waitForSpace did not unblock after closeAndDrain") } } @@ -162,8 +143,8 @@ func TestDefragJournalBackpressureMultipleWriters(t *testing.T) { // Give goroutines time to block. time.Sleep(50 * time.Millisecond) - // Drain should wake both. - j.drain() + // closeAndDrain should wake both. + j.closeAndDrain() done := make(chan struct{}) go func() { @@ -200,16 +181,4 @@ func TestDefragJournalBackpressureDisabledWhenZero(t *testing.T) { } } -func TestDefragJournalDeepCopies(t *testing.T) { - j := newDefragJournal(0) - key := []byte("original") - j.appendPut([]byte("b"), key, []byte("v"), false) - - // mutate the source slice - key[0] = 'X' - - ops := j.drain() - require.Len(t, ops, 1) - assert.Equal(t, []byte("original"), ops[0].key, "journal should deep-copy byte slices") -} diff --git a/server/storage/backend/export_test.go b/server/storage/backend/export_test.go index 30435ad8a6aa..7055cccca6af 100644 --- a/server/storage/backend/export_test.go +++ b/server/storage/backend/export_test.go @@ -37,3 +37,8 @@ func SetNonBlockingDefragForTest(b Backend, enabled bool) { be := b.(*backend) be.nonBlockingDefrag = enabled } + +func SetDefragCopyFailHookForTest(b Backend, hook func() error) { + be := b.(*backend) + be.defragCopyFailHook = hook +} diff --git a/tests/e2e/non_blocking_defrag_test.go b/tests/e2e/non_blocking_defrag_test.go new file mode 100644 index 000000000000..9b1f096ea9a9 --- /dev/null +++ b/tests/e2e/non_blocking_defrag_test.go @@ -0,0 +1,121 @@ +package e2e + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "go.etcd.io/etcd/tests/v3/framework/config" + "go.etcd.io/etcd/tests/v3/framework/e2e" +) + +// TestNonBlockingDefragNoSpace verifies that a failed non-blocking defrag +// does not corrupt data and the server continues to operate. +func TestNonBlockingDefragNoSpace(t *testing.T) { + tests := []struct { + name string + failpoint string + err string + }{ + { + name: "defragOpenFileError", + failpoint: "defragOpenFileError", + err: "no space", + }, + { + name: "defragdbFail", + failpoint: "defragdbFail", + err: "some random error", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + e2e.BeforeTest(t) + + clus, err := e2e.NewEtcdProcessCluster(context.TODO(), t, + e2e.WithClusterSize(1), + e2e.WithGoFailEnabled(true), + e2e.WithServerFeatureGate("NonBlockingDefrag", true), + ) + require.NoError(t, err) + t.Cleanup(func() { clus.Stop() }) + + member := clus.Procs[0] + + // Write data before defrag + for i := 0; i < 10; i++ { + require.NoError(t, member.Etcdctl().Put(context.Background(), fmt.Sprintf("key_%d", i), "value", config.PutOptions{})) + } + + // Trigger a failed defrag + require.NoError(t, member.Failpoints().SetupHTTP(context.Background(), tc.failpoint, fmt.Sprintf(`return("%s")`, tc.err))) + require.ErrorContains(t, member.Etcdctl().Defragment(context.Background(), config.DefragOption{Timeout: time.Minute}), tc.err) + + // Verify server is still functional after failed defrag + require.NoError(t, member.Etcdctl().Put(context.Background(), "after_defrag", "ok", config.PutOptions{})) + value, err := member.Etcdctl().Get(context.Background(), "after_defrag", config.GetOptions{}) + require.NoError(t, err) + require.Len(t, value.Kvs, 1) + require.Equal(t, "ok", string(value.Kvs[0].Value)) + + // Verify pre-defrag data survived + for i := 0; i < 10; i++ { + value, err := member.Etcdctl().Get(context.Background(), fmt.Sprintf("key_%d", i), config.GetOptions{}) + require.NoError(t, err) + require.Lenf(t, value.Kvs, 1, "key_%d should exist after failed defrag", i) + } + }) + } +} + +// TestNonBlockingDefragSuccess verifies that a successful non-blocking +// defrag preserves all data written before and during the defrag. +func TestNonBlockingDefragSuccess(t *testing.T) { + e2e.BeforeTest(t) + clus, err := e2e.NewEtcdProcessCluster(context.TODO(), t, + e2e.WithClusterSize(1), + e2e.WithServerFeatureGate("NonBlockingDefrag", true), + ) + require.NoError(t, err) + t.Cleanup(func() { clus.Stop() }) + + member := clus.Procs[0] + + // Write data before defrag + for i := 0; i < 100; i++ { + require.NoError(t, member.Etcdctl().Put(context.Background(), fmt.Sprintf("key_%04d", i), "value", config.PutOptions{})) + } + + // Delete half to make defrag meaningful + for i := 0; i < 50; i++ { + _, err := member.Etcdctl().Delete(context.Background(), fmt.Sprintf("key_%04d", i), config.DeleteOptions{}) + require.NoError(t, err) + } + + // Defrag should succeed + require.NoError(t, member.Etcdctl().Defragment(context.Background(), config.DefragOption{Timeout: time.Minute})) + + // Verify surviving keys + for i := 50; i < 100; i++ { + value, err := member.Etcdctl().Get(context.Background(), fmt.Sprintf("key_%04d", i), config.GetOptions{}) + require.NoError(t, err) + require.Lenf(t, value.Kvs, 1, "key_%04d should survive defrag", i) + } + + // Verify deleted keys are gone + for i := 0; i < 50; i++ { + value, err := member.Etcdctl().Get(context.Background(), fmt.Sprintf("key_%04d", i), config.GetOptions{}) + require.NoError(t, err) + require.Emptyf(t, value.Kvs, "key_%04d should be deleted after defrag", i) + } + + // Write new data after defrag + require.NoError(t, member.Etcdctl().Put(context.Background(), "post_defrag", "ok", config.PutOptions{})) + value, err := member.Etcdctl().Get(context.Background(), "post_defrag", config.GetOptions{}) + require.NoError(t, err) + require.Equal(t, "ok", string(value.Kvs[0].Value)) +} diff --git a/tests/robustness/failpoint/failpoint.go b/tests/robustness/failpoint/failpoint.go index 4aef09ca4060..7b322daa3c4c 100644 --- a/tests/robustness/failpoint/failpoint.go +++ b/tests/robustness/failpoint/failpoint.go @@ -37,7 +37,7 @@ const ( var allFailpoints = []Failpoint{ KillFailpoint, BeforeCommitPanic, AfterCommitPanic, RaftBeforeSavePanic, RaftAfterSavePanic, - DefragBeforeCopyPanic, DefragBeforeRenamePanic, BackendBeforePreCommitHookPanic, BackendAfterPreCommitHookPanic, + DefragBeforeCopyPanic, DefragBeforeReplayPanic, DefragBeforeRenamePanic, BackendBeforePreCommitHookPanic, BackendAfterPreCommitHookPanic, BackendBeforeStartDBTxnPanic, BackendAfterStartDBTxnPanic, BackendBeforeWritebackBufPanic, BackendAfterWritebackBufPanic, CompactBeforeCommitScheduledCompactPanic, CompactAfterCommitScheduledCompactPanic, CompactBeforeSetFinishedCompactPanic, CompactAfterSetFinishedCompactPanic, CompactBeforeCommitBatchPanic, diff --git a/tests/robustness/failpoint/gofail.go b/tests/robustness/failpoint/gofail.go index 3ce2ff39a535..871b6001052d 100644 --- a/tests/robustness/failpoint/gofail.go +++ b/tests/robustness/failpoint/gofail.go @@ -32,6 +32,7 @@ import ( var ( DefragBeforeCopyPanic Failpoint = goPanicFailpoint{"defragBeforeCopy", triggerDefrag{}, AnyMember} + DefragBeforeReplayPanic Failpoint = goPanicFailpoint{"defragBeforeReplay", triggerDefrag{}, AnyMember} DefragBeforeRenamePanic Failpoint = goPanicFailpoint{"defragBeforeRename", triggerDefrag{}, AnyMember} BeforeCommitPanic Failpoint = goPanicFailpoint{"beforeCommit", nil, AnyMember} AfterCommitPanic Failpoint = goPanicFailpoint{"afterCommit", nil, AnyMember}