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..ed082b511118 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"`
@@ -712,6 +718,7 @@ func NewConfig() *Config {
 		ExperimentalMemoryMlock:             false,
 		ExperimentalStopGRPCServiceOnDefrag: false,
 		MaxLearners:                         membership.DefaultMaxLearners,
+		DefragJournalMaxOps:                 backend.DefaultDefragJournalMaxOps,
 
 		ExperimentalTxnModeWriteWithSharedBuffer:  DefaultExperimentalTxnModeWriteWithSharedBuffer,
 		ExperimentalDistributedTracingAddress:     DefaultDistributedTracingAddress,
@@ -950,6 +957,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/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/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/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 275064f083b2..19fac4159eb5 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"
 )
 
@@ -127,6 +129,16 @@ 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
+
+	// 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
 }
 
@@ -150,6 +162,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)
@@ -235,6 +255,9 @@ func newBackend(bcfg BackendConfig) *backend {
 		stopc: make(chan struct{}),
 		donec: make(chan struct{}),
 
+		nonBlockingDefrag:   bcfg.NonBlockingDefrag,
+		defragJournalMaxOps: bcfg.DefragJournalMaxOps,
+
 		lg: bcfg.Logger,
 	}
 
@@ -460,35 +483,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)
-
-	// 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.
+// 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{}
@@ -500,74 +507,120 @@ 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)
+	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
+		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",
+	b.lg.Info("defragmenting",
 		zap.String("path", dbp),
 		zap.Int64("current-db-size-bytes", size1),
 		zap.String("current-db-size", humanize.Bytes(uint64(size1))),
 		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()
+
+	// 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() {
-		// 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))
 		}
 	}()
 
-	// Commit/stop and then reset current transactions (including the readTx)
 	b.batchTx.unsafeCommit(true)
 	b.batchTx.tx = nil
 
-	// gofail: var defragBeforeCopy struct{}
-	err = defragdb(b.db, tmpdb, defragLimit)
+	// 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()
-		if rmErr := os.RemoveAll(tmpdb.Path()); rmErr != nil {
-			b.lg.Error("failed to remove db.tmp after defragmentation completed", zap.Error(rmErr))
+		if rmErr := os.RemoveAll(tdbp); rmErr != nil {
+			b.lg.Error("failed to remove tmp database", zap.String("path", tdbp), zap.Error(rmErr))
 		}
+		return err
+	}
 
-		// restore the bbolt transactions if defragmentation fails
+	// 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 tmp database", zap.String("path", tdbp), zap.Error(rmErr))
+		}
 		b.batchTx.tx = b.unsafeBegin(true)
 		b.readTx.tx = b.unsafeBegin(false)
-
 		return err
 	}
 
-	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))
 	}
 
@@ -587,27 +640,174 @@ func (b *backend) defrag() error {
 
 	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()
+
+	readTx, err := b.defragPrepare()
+	if err != nil {
+		tmpdb.Close()
+		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{}
+	if b.defragCopyFailHook != nil {
+		err = b.defragCopyFailHook()
+	} else {
+		err = defragFromTx(readTx, tmpdb, defragLimit)
+	}
+	readTx.Rollback()
+
+	if err != nil {
+		tmpdb.Close()
+		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(tmpdb, dbp, tdbp)
+	if err != nil {
+		return err
+	}
+
+	took := time.Since(now)
+	defragSec.Observe(took.Seconds())
+	b.defragLogFinish(dbp, size1, sizeInUse1, took)
+	return nil
+}
+
+// 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()
+	readTx, err := b.db.Begin(false)
+	b.mu.RUnlock()
+	if err != nil {
+		return nil, fmt.Errorf("failed to begin read tx for defrag: %w", err)
+	}
+
+	b.batchTx.defragJournal.Store(newDefragJournal(b.defragJournalMaxOps))
+	return readTx, nil
+}
+
+// defragCancelJournal removes the journal from the batch transaction
+// 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()
+	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(tmpdb *bolt.DB, dbp, tdbp string) error {
+	b.batchTx.LockOutsideApply()
+	defer b.batchTx.Unlock()
+
+	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)))
+	}
+
+	if err := replayJournal(tmpdb, ops, defragLimit); err != nil {
+		tmpdb.Close()
+		if rmErr := os.RemoveAll(tdbp); rmErr != nil {
+			b.lg.Error("failed to remove tmp database", zap.String("path", tdbp), zap.Error(rmErr))
+		}
+		return err
+	}
+
+	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
+	// 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
+
+	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))
+	}
+
+	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))
+	}
+	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)))
 
-	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
 }
 
-func defragdb(odb, tmpdb *bolt.DB, limit int) error {
+func defragFromTx(srcTx *bolt.Tx, tmpdb *bolt.DB, limit int) error {
 	// gofail: var defragdbFail string
 	// return fmt.Errorf(defragdbFail)
 
-	// open a tx on tmpdb for writes
 	tmptx, err := tmpdb.Begin(true)
 	if err != nil {
 		return err
@@ -618,18 +818,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 +857,66 @@ 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 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 {
+				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)
+			}
+		}
+	}
+
+	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..6bc18480f485 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,1346 @@ 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)
+	backend.SetNonBlockingDefragForTest(b, true)
+
+	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)
+	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)), []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)
+	backend.SetNonBlockingDefragForTest(b, true)
+
+	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)
+	backend.SetNonBlockingDefragForTest(b, true)
+
+	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)
+	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("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)
+	backend.SetNonBlockingDefragForTest(b, true)
+
+	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)
+	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("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)
+	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("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)
+	backend.SetNonBlockingDefragForTest(b, true)
+
+	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)
+	backend.SetNonBlockingDefragForTest(b, true)
+
+	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)
+	backend.SetNonBlockingDefragForTest(b, true)
+
+	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)
+	backend.SetNonBlockingDefragForTest(b, true)
+
+	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)
+	backend.SetNonBlockingDefragForTest(b, true)
+
+	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)
+	backend.SetNonBlockingDefragForTest(b, true)
+
+	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)
+	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("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)
+	backend.SetNonBlockingDefragForTest(b, true)
+
+	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)
+	backend.SetNonBlockingDefragForTest(b, true)
+
+	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)
+	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("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()
+}
+
+// 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()
+}
+
+// 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 5af557cb4283..07f22bb9ea1e 100644
--- a/server/storage/backend/batch_tx.go
+++ b/server/storage/backend/batch_tx.go
@@ -291,6 +291,11 @@ type batchTxBuffered struct {
 	batchTx
 	buf                     txWriteBuffer
 	pendingDeleteOperations int
+	// 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 {
@@ -305,6 +310,13 @@ func newBatchTxBuffered(backend *backend) *batchTxBuffered {
 	return tx
 }
 
+func (t *batchTxBuffered) LockInsideApply() {
+	if j := t.defragJournal.Load(); j != nil {
+		j.waitForSpace()
+	}
+	t.batchTx.LockInsideApply()
+}
+
 func (t *batchTxBuffered) Unlock() {
 	if t.pending != 0 {
 		t.backend.readTx.Lock() // blocks txReadBuffer for writing.
@@ -385,22 +397,41 @@ func (t *batchTxBuffered) unsafeCommit(stop bool) {
 	}
 }
 
+func (t *batchTxBuffered) UnsafeCreateBucket(bucket Bucket) {
+	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.Load(); 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.Load(); 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.Load(); j != nil {
+		j.appendDelete(bucketType.Name(), key)
+	}
 	t.batchTx.UnsafeDelete(bucketType, key)
 	t.pendingDeleteOperations++
 }
 
 func (t *batchTxBuffered) UnsafeDeleteBucket(bucket Bucket) {
+	if j := t.defragJournal.Load(); 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..83d5c747e314
--- /dev/null
+++ b/server/storage/backend/defrag_journal.go
@@ -0,0 +1,126 @@
+package backend
+
+import "sync"
+
+type defragOpType uint8
+
+const (
+	opPut defragOpType = iota
+	opDelete
+	opCreateBucket
+	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
+	key        []byte
+	value      []byte
+	seq        bool
+}
+
+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
+}
+
+// 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()
+	}
+}
+
+func (j *defragJournal) appendPut(bucketName, key, value []byte, seq bool) {
+	j.mu.Lock()
+	defer j.mu.Unlock()
+	if j.closed {
+		panic("defragJournal: append to closed journal")
+	}
+	j.ops = append(j.ops, defragJournalOp{
+		opType:     opPut,
+		bucketName: bucketName,
+		key:        key,
+		value:      value,
+		seq:        seq,
+	})
+}
+
+func (j *defragJournal) appendDelete(bucketName, key []byte) {
+	j.mu.Lock()
+	defer j.mu.Unlock()
+	if j.closed {
+		panic("defragJournal: append to closed journal")
+	}
+	j.ops = append(j.ops, defragJournalOp{
+		opType:     opDelete,
+		bucketName: bucketName,
+		key:        key,
+	})
+}
+
+func (j *defragJournal) appendCreateBucket(bucketName []byte) {
+	j.mu.Lock()
+	defer j.mu.Unlock()
+	if j.closed {
+		panic("defragJournal: append to closed journal")
+	}
+	j.ops = append(j.ops, defragJournalOp{
+		opType:     opCreateBucket,
+		bucketName: bucketName,
+	})
+}
+
+func (j *defragJournal) appendDeleteBucket(bucketName []byte) {
+	j.mu.Lock()
+	defer j.mu.Unlock()
+	if j.closed {
+		panic("defragJournal: append to closed journal")
+	}
+	j.ops = append(j.ops, defragJournalOp{
+		opType:     opDeleteBucket,
+		bucketName: bucketName,
+	})
+}
+
+// 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
+}
+
+
diff --git a/server/storage/backend/defrag_journal_test.go b/server/storage/backend/defrag_journal_test.go
new file mode 100644
index 000000000000..c1112dc8f90a
--- /dev/null
+++ b/server/storage/backend/defrag_journal_test.go
@@ -0,0 +1,184 @@
+package backend
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDefragJournalCloseAndDrain(t *testing.T) {
+	j := newDefragJournal(0)
+
+	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.closeAndDrain()
+	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)
+
+	// closeAndDrain again should be empty
+	ops = j.closeAndDrain()
+	assert.Empty(t, ops)
+}
+
+func TestDefragJournalAppendAfterClosePanics(t *testing.T) {
+	j := newDefragJournal(0)
+	j.closeAndDrain()
+
+	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) {
+	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.closeAndDrain(), 5)
+	// Re-create since we just closed it.
+	j = newDefragJournal(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):
+	}
+
+	// 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 closeAndDrain")
+	}
+}
+
+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):
+	}
+
+	// closeAndDrain wakes blocked writers.
+	j.closeAndDrain()
+
+	select {
+	case <-unblocked:
+	case <-time.After(time.Second):
+		t.Fatal("waitForSpace did not unblock after closeAndDrain")
+	}
+}
+
+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)
+
+	// closeAndDrain should wake both.
+	j.closeAndDrain()
+
+	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")
+	}
+}
+
+
diff --git a/server/storage/backend/export_test.go b/server/storage/backend/export_test.go
index e9f5ad38d6a1..7055cccca6af 100644
--- a/server/storage/backend/export_test.go
+++ b/server/storage/backend/export_test.go
@@ -27,3 +27,18 @@ 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
+}
+
+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}