Handle Batch full

Author: tv42

batch.go

@@ -22,15 +22,14 @@ func (db *DB) Batch(fn func(*Tx) error) error {
 	errCh := make(chan error, 1)
 
 	db.batchMu.Lock()
+
 	if db.batch == nil {
 		db.batch = &batch{
 			db: db,
 		}
 		db.batch.timer = time.AfterFunc(db.MaxBatchDelay, db.batch.trigger)
 	}
-	if len(db.batch.calls) < db.MaxBatchSize {
-		db.batch.calls = append(db.batch.calls, call{fn: fn, err: errCh})
-	}
+	db.batch.calls = append(db.batch.calls, call{fn: fn, err: errCh})
 	if len(db.batch.calls) >= db.MaxBatchSize {
 		// wake up batch, it's ready to run
 		go db.batch.trigger()

batch_test.go

@@ -2,6 +2,7 @@ package bolt_test
 
 import (
 	"testing"
+	"time"
 
 	"github.com/boltdb/bolt"
 )
@@ -72,3 +73,41 @@ func TestDB_Batch_Panic(t *testing.T) {
 		t.Fatalf("wrong error: %v != %v", g, e)
 	}
 }
+
+func TestDB_BatchFull(t *testing.T) {
+	db := NewTestDB()
+	defer db.Close()
+	db.MustCreateBucket([]byte("widgets"))
+
+	n := 3
+	// high enough to never trigger here
+	db.MaxBatchDelay = 1 * time.Hour
+	// make sure everything will not fit in one batch
+	db.MaxBatchSize = n - 1
+	ch := make(chan error)
+	for i := 0; i < n; i++ {
+		go func(i int) {
+			ch <- db.Batch(func(tx *bolt.Tx) error {
+				return tx.Bucket([]byte("widgets")).Put(ui64tob(uint64(i)), []byte{})
+			})
+		}(i)
+	}
+
+	// Check all responses to make sure there's no error.
+	for i := 0; i < n; i++ {
+		if err := <-ch; err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	// Ensure data is correct.
+	db.MustView(func(tx *bolt.Tx) error {
+		b := tx.Bucket([]byte("widgets"))
+		for i := 0; i < n; i++ {
+			if v := b.Get(ui64tob(uint64(i))); v == nil {
+				t.Errorf("key not found: %d", i)
+			}
+		}
+		return nil
+	})
+}