intelligently close marksets and signal errors in concurrent operations

blockstore/splitstore/markset.go

@@ -1,11 +1,15 @@
 package splitstore
 
 import (
+	"errors"
+
 	"golang.org/x/xerrors"
 
 	cid "github.com/ipfs/go-cid"
 )
 
+var errMarkSetClosed = errors.New("markset closed")
+
 // MarkSet is a utility to keep track of seen CID, and later query for them.
 //
 // * If the expected dataset is large, it can be backed by a datastore (e.g. bbolt).

blockstore/splitstore/markset_bloom.go

@@ -74,6 +74,10 @@ func (s *BloomMarkSet) Mark(cid cid.Cid) error {
 		defer s.mx.Unlock()
 	}
 
+	if s.bf == nil {
+		return errMarkSetClosed
+	}
+
 	s.bf.Add(s.saltedKey(cid))
 	return nil
 }
@@ -84,9 +88,18 @@ func (s *BloomMarkSet) Has(cid cid.Cid) (bool, error) {
 		defer s.mx.Unlock()
 	}
 
+	if s.bf == nil {
+		return false, errMarkSetClosed
+	}
+
 	return s.bf.Has(s.saltedKey(cid)), nil
 }
 
 func (s *BloomMarkSet) Close() error {
+	if s.ts {
+		s.mx.Lock()
+		defer s.mx.Unlock()
+	}
+	s.bf = nil
 	return nil
 }

blockstore/splitstore/markset_map.go

@@ -42,6 +42,10 @@ func (s *MapMarkSet) Mark(cid cid.Cid) error {
 		defer s.mx.Unlock()
 	}
 
+	if s.set == nil {
+		return errMarkSetClosed
+	}
+
 	s.set[string(cid.Hash())] = struct{}{}
 	return nil
 }
@@ -52,10 +56,19 @@ func (s *MapMarkSet) Has(cid cid.Cid) (bool, error) {
 		defer s.mx.Unlock()
 	}
 
+	if s.set == nil {
+		return false, errMarkSetClosed
+	}
+
 	_, ok := s.set[string(cid.Hash())]
 	return ok, nil
 }
 
 func (s *MapMarkSet) Close() error {
+	if s.ts {
+		s.mx.Lock()
+		defer s.mx.Unlock()
+	}
+	s.set = nil
 	return nil
 }