test and fix engine goroutine leak, fixes #14

client_test.go

@@ -1,13 +1,28 @@
 package lungo
 
 import (
+	"runtime"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/mongo"
 )
 
+func TestOpenGoroutineLeak(t *testing.T) {
+	num := runtime.NumGoroutine()
+
+	for i := 0; i < 10; i++ {
+		_, engine, err := Open(nil, Options{
+			Store: NewMemoryStore(),
+		})
+		assert.NoError(t, err)
+		engine.Close()
+	}
+
+	assert.Equal(t, num, runtime.NumGoroutine())
+}
+
 func TestClientListDatabasesAndNames(t *testing.T) {
 	clientTest(t, func(t *testing.T, c IClient) {
 		err := c.Database(testDB).Drop(nil)

engine.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"go.mongodb.org/mongo-driver/bson/primitive"
+	"gopkg.in/tomb.v2"
 
 	"github.com/256dpi/lungo/bsonkit"
 	"github.com/256dpi/lungo/dbkit"
@@ -52,7 +53,7 @@ type Engine struct {
 	streams map[*Stream]struct{}
 	token   *dbkit.Semaphore
 	txn     *Transaction
-	closed  bool
+	tomb    tomb.Tomb
 	mutex   sync.Mutex
 }
 
@@ -98,7 +99,10 @@ func CreateEngine(opts Options) (*Engine, error) {
 	e.catalog = data
 
 	// run expiry
-	go e.expire(opts.ExpireInterval, opts.ExpireErrors)
+	e.tomb.Go(func() error {
+		e.expire(opts.ExpireInterval, opts.ExpireErrors)
+		return nil
+	})
 
 	return e, nil
 }
@@ -123,7 +127,7 @@ func (e *Engine) Begin(ctx context.Context, lock bool) (*Transaction, error) {
 	defer e.mutex.Unlock()
 
 	// check if closed
-	if e.closed {
+	if !e.tomb.Alive() {
 		return nil, ErrEngineClosed
 	}
 
@@ -173,7 +177,7 @@ func (e *Engine) Commit(txn *Transaction) error {
 	defer e.mutex.Unlock()
 
 	// check if closed
-	if e.closed {
+	if !e.tomb.Alive() {
 		return ErrEngineClosed
 	}
 
@@ -225,7 +229,7 @@ func (e *Engine) Abort(txn *Transaction) {
 	defer e.mutex.Unlock()
 
 	// check if closed
-	if e.closed {
+	if !e.tomb.Alive() {
 		return
 	}
 
@@ -248,7 +252,7 @@ func (e *Engine) Watch(handle Handle, pipeline bsonkit.List, resumeAfter, startA
 	defer e.mutex.Unlock()
 
 	// check if closed
-	if e.closed {
+	if !e.tomb.Alive() {
 		return nil, ErrEngineClosed
 	}
 
@@ -346,7 +350,7 @@ func (e *Engine) Close() {
 	defer e.mutex.Unlock()
 
 	// check if closed
-	if e.closed {
+	if !e.tomb.Alive() {
 		return
 	}
 
@@ -355,14 +359,23 @@ func (e *Engine) Close() {
 		close(stream.signal)
 	}
 
-	// set flag
-	e.closed = true
+	// kill tomb
+	e.tomb.Kill(nil)
+	_ = e.tomb.Wait()
 }
 
 func (e *Engine) expire(interval time.Duration, reporter func(error)) {
+	// prepare ticker
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+
 	for {
 		// await next interval
-		time.Sleep(interval)
+		select {
+		case <-e.tomb.Dying():
+			return
+		case <-ticker.C:
+		}
 
 		// get transaction
 		txn, err := e.Begin(nil, true)

go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/stretchr/testify v1.8.4
 	github.com/tidwall/btree v1.7.0
 	go.mongodb.org/mongo-driver v1.14.0
+	gopkg.in/tomb.v2 v2.0.0-20161208151619-d5d1b5820637
 )
 
 require (
@@ -20,6 +21,7 @@ require (
 	github.com/xdg-go/stringprep v1.0.4 // indirect
 	github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d // indirect
 	golang.org/x/crypto v0.17.0 // indirect
+	golang.org/x/net v0.10.0 // indirect
 	golang.org/x/sync v0.1.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect

go.sum

@@ -34,6 +34,8 @@ golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
+golang.org/x/net v0.10.0 h1:X2//UzNDwYmtCLn7To6G58Wr6f5ahEAQgKNzv9Y951M=
+golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o=
@@ -58,5 +60,7 @@ golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8T
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/tomb.v2 v2.0.0-20161208151619-d5d1b5820637 h1:yiW+nvdHb9LVqSHQBXfZCieqV4fzYhNBql77zY0ykqs=
+gopkg.in/tomb.v2 v2.0.0-20161208151619-d5d1b5820637/go.mod h1:BHsqpu/nsuzkT5BpiH1EMZPLyqSMM8JbIavyFACoFNk=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=