go lint

.github/workflows/ci.yml

@@ -20,6 +20,8 @@ jobs:
       run: nix develop --command make build
     - name: Test
       run: nix develop --command make test-verbose
+    - name: Lint
+      run: nix develop --command make lint
     - name: Tidy
       run: nix develop --command go mod tidy
     - name: Check if tidy changed anything

Makefile

@@ -1,12 +1,15 @@
 default: test
 
-.PHONY: test test-verbose
+.PHONY: test test-verbose lint
 test:
 	go test -cover ./...
 
 test-verbose:
 	go test -cover -v ./...
 
+lint:
+	golangci-lint run
+
 .PHONY:
 build:
 	go build -v ./...

bench_test.go

@@ -4,10 +4,10 @@ import (
 	"fmt"
 	"os"
 	"strings"
-	"sync"
 	"testing"
 
 	"github.com/stretchr/testify/require"
+	"golang.org/x/sync/errgroup"
 
 	"github.com/klev-dev/klevdb/message"
 )
@@ -91,23 +91,22 @@ func benchmarkPublishMulti(b *testing.B) {
 	b.SetBytes(s.Size(msgs[0]) * 4)
 	b.ResetTimer()
 
-	var wg sync.WaitGroup
+	g := new(errgroup.Group)
 	for k := 0; k < 10; k++ {
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
+		g.Go(func() error {
 			for i := 0; i < b.N; i += 4 {
 				top := i + 4
 				if top > b.N {
 					top = b.N
 				}
 				if _, err := s.Publish(msgs[i:top]); err != nil {
-					b.Fatal(err)
+					return err
 				}
 			}
-		}()
+			return nil
+		})
 	}
-	wg.Wait()
+	require.NoError(b, g.Wait())
 
 	b.StopTimer()
 }
@@ -237,19 +236,18 @@ func benchmarkConsumeMulti(b *testing.B) {
 	b.SetBytes(s.Size(msgs[0]) * 4)
 	b.ResetTimer()
 
-	var wg sync.WaitGroup
+	g := new(errgroup.Group)
 	for k := 0; k < 10; k++ {
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
+		g.Go(func() error {
 			for i := 0; i < b.N; i += 4 {
 				if _, _, err := s.Consume(int64(i), 4); err != nil {
-					b.Fatal(err)
+					return err
 				}
 			}
-		}()
+			return nil
+		})
 	}
-	wg.Wait()
+	require.NoError(b, g.Wait())
 
 	b.StopTimer()
 }
@@ -398,19 +396,18 @@ func benchmarkGetKeyMulti(b *testing.B) {
 	b.SetBytes(s.Size(msgs[0]))
 	b.ResetTimer()
 
-	var wg sync.WaitGroup
+	g := new(errgroup.Group)
 	for k := 0; k < 10; k++ {
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
+		g.Go(func() error {
 			for i := 0; i < b.N; i++ {
 				if _, err := s.GetByKey(msgs[i].Key); err != nil {
-					b.Fatal(err)
+					return err
 				}
 			}
-		}()
+			return nil
+		})
 	}
-	wg.Wait()
+	require.NoError(b, g.Wait())
 
 	b.StopTimer()
 }
@@ -427,34 +424,33 @@ func benchmarkBaseMulti(b *testing.B) {
 
 	b.ResetTimer()
 
-	var wg sync.WaitGroup
-	wg.Add(2)
+	g := new(errgroup.Group)
 
-	go func() {
-		defer wg.Done()
+	g.Go(func() error {
 		for i := 0; i < b.N; i += 10 {
 			top := i + 10
 			if top > b.N {
 				top = b.N
 			}
 			if _, err := s.Publish(msgs[i:top]); err != nil {
-				b.Fatal(err)
+				return err
 			}
 		}
-	}()
-
-	go func() {
-		defer wg.Done()
+		return nil
+	})
 
+	g.Go(func() error {
 		offset := OffsetOldest
 		for offset < int64(len(msgs)) {
 			next, _, err := s.Consume(offset, 10)
-			require.NoError(b, err)
+			if err != nil {
+				return err
+			}
 			offset = next
 		}
-	}()
-
-	wg.Wait()
+		return nil
+	})
+	require.NoError(b, g.Wait())
 
 	b.StopTimer()
 }

flake.nix

@@ -18,6 +18,8 @@
         devShell = pkgs.mkShell {
           buildInputs = with pkgs; [
             go
+            gopls
+            golangci-lint
           ];
         };
       });

log_test.go

@@ -963,7 +963,8 @@ func testBackupSegment(t *testing.T) {
 	require.Equal(t, 1, stat.Segments)
 
 	bdir := t.TempDir()
-	l.Backup(bdir)
+	err = l.Backup(bdir)
+	require.NoError(t, err)
 
 	bl, err := Open(bdir, Options{TimeIndex: true, KeyIndex: true})
 	require.NoError(t, err)
@@ -1005,7 +1006,8 @@ func testBackupSegments(t *testing.T) {
 	require.Equal(t, 2, stat.Segments)
 
 	bdir := t.TempDir()
-	l.Backup(bdir)
+	err = l.Backup(bdir)
+	require.NoError(t, err)
 
 	bl, err := Open(bdir, Options{TimeIndex: true, KeyIndex: true})
 	require.NoError(t, err)
@@ -1469,7 +1471,6 @@ func TestConcurrent(t *testing.T) {
 	t.Run("Consume", testConcurrentConsume)
 	t.Run("Delete", testConcurrentDelete)
 	t.Run("GC", testConcurrentGC)
-	// t.Run("Close", testConcurrentClose)
 }
 
 func testConcurrentPubsubRecent(t *testing.T) {
@@ -1698,41 +1699,3 @@ func testConcurrentGC(t *testing.T) {
 	}
 	require.NoError(t, err)
 }
-
-func testConcurrentClose(t *testing.T) {
-	dir := t.TempDir()
-
-	msgs := message.Gen(100)
-	msgSize := message.Size(msgs[0])
-
-	l, err := Open(dir, Options{
-		Rollover: msgSize * 10,
-	})
-	require.NoError(t, err)
-	defer l.Close()
-
-	publishBatched(t, l, msgs, 10)
-
-	g, ctx := errgroup.WithContext(context.TODO())
-	g.Go(func() error {
-		for ctx.Err() == nil {
-			next, consumed, err := l.Consume(OffsetOldest, 32)
-			if err != nil {
-				return err
-			}
-			require.Equal(t, int64(10), next)
-			require.Equal(t, msgs[0:10], consumed)
-		}
-		return nil
-	})
-
-	g.Go(func() error {
-		return l.Close()
-	})
-
-	err = g.Wait()
-	if serr := kleverr.Get(err); serr != nil {
-		fmt.Println(serr.Print())
-	}
-	require.NoError(t, err)
-}