fix: hold close lock when creating transactions

Otherwise, we might not clean things up properly.

datastore.go

@@ -370,8 +370,8 @@ func (d *Datastore) Delete(key ds.Key) error {
 
 func (d *Datastore) Query(q dsq.Query) (dsq.Results, error) {
 	d.closeLk.RLock()
-	defer d.closeLk.RUnlock()
 	if d.closed {
+		d.closeLk.RUnlock()
 		return nil, ErrClosed
 	}
 
@@ -680,14 +680,15 @@ func (t *txn) delete(key ds.Key) error {
 
 func (t *txn) Query(q dsq.Query) (dsq.Results, error) {
 	t.ds.closeLk.RLock()
-	defer t.ds.closeLk.RUnlock()
 	if t.ds.closed {
+		t.ds.closeLk.RUnlock()
 		return nil, ErrClosed
 	}
 
 	return t.query(q)
 }
 
+// must be called with the close lock. this function will unlock it internally.
 func (t *txn) query(q dsq.Query) (dsq.Results, error) {
 	opt := badger.DefaultIteratorOptions
 	opt.PrefetchValues = !q.KeysOnly
@@ -737,7 +738,6 @@ func (t *txn) query(q dsq.Query) (dsq.Results, error) {
 	it := t.txn.NewIterator(opt)
 	qrb := dsq.NewResultBuilder(q)
 	qrb.Process.Go(func(worker goprocess.Process) {
-		t.ds.closeLk.RLock()
 		closedEarly := false
 		defer func() {
 			t.ds.closeLk.RUnlock()
@@ -751,10 +751,6 @@ func (t *txn) query(q dsq.Query) (dsq.Results, error) {
 			}
 
 		}()
-		if t.ds.closed {
-			closedEarly = true
-			return
-		}
 
 		// this iterator is part of an implicit transaction, so when
 		// we're done we must discard the transaction. It's safe to