commit bbolt transaction if there is any pending deleting operations

Signed-off-by: Siyuan Zhang <sizhang@google.com>

server/storage/backend/batch_tx.go

@@ -73,7 +73,8 @@ type batchTx struct {
 	tx      *bolt.Tx
 	backend *backend
 
-	pending int
+	pending                 int
+	pendingDeleteOperations int
 }
 
 // Lock is supposed to be called only by the unit test.
@@ -133,6 +134,7 @@ func (t *batchTx) UnsafeDeleteBucket(bucket Bucket) {
 		)
 	}
 	t.pending++
+	t.pendingDeleteOperations++
 }
 
 // UnsafePut must be called holding the lock on the tx.
@@ -223,6 +225,7 @@ func (t *batchTx) UnsafeDelete(bucketType Bucket, key []byte) {
 		)
 	}
 	t.pending++
+	t.pendingDeleteOperations++
 }
 
 // UnsafeForEach must be called holding the lock on the tx.
@@ -277,6 +280,7 @@ func (t *batchTx) commit(stop bool) {
 		atomic.AddInt64(&t.backend.commits, 1)
 
 		t.pending = 0
+		t.pendingDeleteOperations = 0
 		if err != nil {
 			t.backend.lg.Fatal("failed to commit tx", zap.Error(err))
 		}
@@ -310,7 +314,14 @@ func (t *batchTxBuffered) Unlock() {
 		t.buf.writeback(&t.backend.readTx.buf)
 		// gofail: var afterWritebackBuf struct{}
 		t.backend.readTx.Unlock()
-		if t.pending >= t.backend.batchLimit {
+		// We need to commit the transaction if
+		// 1. t.pending >= t.backend.batchLimit for performance,
+		// 	  because We don't want buffer to grow too big as it has worse complexity.
+		// 2. there is any pending deleting for correctness,
+		//    because buffer doesn't keep track of deletions.
+		// Please also refer to
+		// https://github.com/etcd-io/etcd/pull/17119#issuecomment-1857547158
+		if t.pending >= t.backend.batchLimit || t.pendingDeleteOperations > 0 {
 			t.commit(false)
 		}
 	}