kv/bulk: close bulk-adder-monitor on error

Previously, we would have left the `bulk-adder-monitor` not closed if we
hit an error in `MakeBulkAdder`, and this is now fixed. (The only error
possible is actually not enough memory budget to reserve the buffer
size.)

Release note: None

Author: yuzefovich

pkg/kv/bulk/buffering_adder.go

@@ -79,6 +79,9 @@ var _ kvserverbase.BulkAdder = &BufferingAdder{}
 // passed to add into SSTs that are then ingested. rangeCache if set is
 // consulted to avoid generating an SST that will span a range boundary and thus
 // encounter an error and need to be split and retired to be applied.
+//
+// The BulkAdder takes ownership of the memory monitor which must be non-nil. In
+// case of an error, the monitor will be stopped.
 func MakeBulkAdder(
 	ctx context.Context,
 	db *kv.DB,
@@ -88,7 +91,15 @@ func MakeBulkAdder(
 	opts kvserverbase.BulkAdderOptions,
 	bulkMon *mon.BytesMonitor,
 	sendLimiter limit.ConcurrentRequestLimiter,
-) (*BufferingAdder, error) {
+) (_ *BufferingAdder, retErr error) {
+	if bulkMon == nil {
+		return nil, errors.New("bulkMon must be non-nil")
+	}
+	defer func() {
+		if retErr != nil {
+			bulkMon.Stop(ctx)
+		}
+	}()
 	if opts.MinBufferSize == 0 {
 		opts.MinBufferSize = 32 << 20
 	}
@@ -166,11 +177,8 @@ func (b *BufferingAdder) Close(ctx context.Context) {
 		b.sink.mu.Unlock()
 	}
 	b.sink.Close(ctx)
-
-	if b.bulkMon != nil {
-		b.memAcc.Close(ctx)
-		b.bulkMon.Stop(ctx)
-	}
+	b.memAcc.Close(ctx)
+	b.bulkMon.Stop(ctx)
 }
 
 // Add adds a key to the buffer and checks if it needs to flush.