Repository: blevesearch/bleve
Pull request #1613 (2021-08-05 21:44:51)

Address panic on closing a previously closed channel
  + Fixes: https://github.com/blevesearch/bleve/issues/1609

  panic: close of closed channel

    goroutine 68 [running]:
    github.com/blevesearch/bleve/v2/index/scorch.(*closeChWrapper).close(0xc000346b80)
            /Users/runner/work/bleve/bleve/index/scorch/merge.go:228 +0x99
    github.com/blevesearch/bleve/v2/index/scorch.(*closeChWrapper).listen(0xc000346b80)
            /Users/runner/work/bleve/bleve/index/scorch/merge.go:235 +0x1cd
    created by github.com/blevesearch/bleve/v2/index/scorch.(*Scorch).planMergeAtSnapshot
            /Users/runner/work/bleve/bleve/index/scorch/merge.go:275 +0x4ba

abhinavdangeti: Ah I see. Ok so, will need to edit this fix. The panic occurs whenever the index is closed when force merge is in flight or if the operation is canceled - and then the defer cw.close() runs. 

Patch:
diff --git a/index.go b/index.go
index e08271e5..e297122f 100644
--- a/index.go
+++ b/index.go
@@ -16,6 +16,7 @@ package bleve
 
 import (
 	"context"
+
 	"github.com/blevesearch/bleve/v2/index/upsidedown"
 
 	"github.com/blevesearch/bleve/v2/document"
@@ -306,3 +307,15 @@ type Builder interface {
 func NewBuilder(path string, mapping mapping.IndexMapping, config map[string]interface{}) (Builder, error) {
 	return newBuilder(path, mapping, config)
 }
+
+// IndexCopyable is an index which supports an online copy operation
+// of the index.
+type IndexCopyable interface {
+	// CopyTo creates a fully functional copy of the index at the
+	// specified destination directory implementation.
+	CopyTo(d index.Directory) error
+}
+
+// FileSystemDirectory is the default implementation for the
+// index.Directory interface.
+type FileSystemDirectory string
diff --git a/index/scorch/builder.go b/index/scorch/builder.go
index 3311bd02..5e64230c 100644
--- a/index/scorch/builder.go
+++ b/index/scorch/builder.go
@@ -304,7 +304,7 @@ func (o *Builder) Close() error {
 	}
 
 	// fill the root bolt with this fake index snapshot
-	_, _, err = prepareBoltSnapshot(is, tx, o.path, o.segPlugin)
+	_, _, err = prepareBoltSnapshot(is, tx, o.path, o.segPlugin, nil)
 	if err != nil {
 		_ = tx.Rollback()
 		_ = rootBolt.Close()
diff --git a/index/scorch/merge.go b/index/scorch/merge.go
index 1e269afc..30f7a4f2 100644
--- a/index/scorch/merge.go
+++ b/index/scorch/merge.go
@@ -209,32 +209,32 @@ func (s *Scorch) parseMergePlannerOptions() (*mergeplan.MergePlanOptions,
 }
 
 type closeChWrapper struct {
-	ch1     chan struct{}
-	ctx     context.Context
-	closeCh chan struct{}
+	ch1      chan struct{}
+	ctx      context.Context
+	closeCh  chan struct{}
+	cancelCh chan struct{}
 }
 
 func newCloseChWrapper(ch1 chan struct{},
 	ctx context.Context) *closeChWrapper {
-	return &closeChWrapper{ch1: ch1,
-		ctx:     ctx,
-		closeCh: make(chan struct{})}
+	return &closeChWrapper{
+		ch1:      ch1,
+		ctx:      ctx,
+		closeCh:  make(chan struct{}),
+		cancelCh: make(chan struct{}),
+	}
 }
 
 func (w *closeChWrapper) close() {
-	select {
-	case <-w.closeCh:
-	default:
-		close(w.closeCh)
-	}
+	close(w.closeCh)
 }
 
 func (w *closeChWrapper) listen() {
 	select {
 	case <-w.ch1:
-		w.close()
+		close(w.cancelCh)
 	case <-w.ctx.Done():
-		w.close()
+		close(w.cancelCh)
 	case <-w.closeCh:
 	}
 }
@@ -320,7 +320,7 @@ func (s *Scorch) planMergeAtSnapshot(ctx context.Context,
 
 			atomic.AddUint64(&s.stats.TotFileMergeZapBeg, 1)
 			newDocNums, _, err := s.segPlugin.Merge(segmentsToMerge, docsToDrop, path,
-				cw.closeCh, s)
+				cw.cancelCh, s)
 			atomic.AddUint64(&s.stats.TotFileMergeZapEnd, 1)
 
 			fileMergeZapTime := uint64(time.Since(fileMergeZapStartTime))
diff --git a/index/scorch/persister.go b/index/scorch/persister.go
index 00c92429..1688edae 100644
--- a/index/scorch/persister.go
+++ b/index/scorch/persister.go
@@ -19,6 +19,7 @@ import (
 	"encoding/binary"
 	"encoding/json"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"log"
 	"math"
@@ -427,8 +428,59 @@ func (s *Scorch) persistSnapshotMaybeMerge(snapshot *IndexSnapshot) (
 	return true, nil
 }
 
+func copyToDirectory(srcPath string, d index.Directory) (int64, error) {
+	if d == nil {
+		return 0, nil
+	}
+
+	dest, err := d.GetWriter(filepath.Join("store", filepath.Base(srcPath)))
+	if err != nil {
+		return 0, fmt.Errorf("GetWriter err: %v", err)
+	}
+
+	sourceFileStat, err := os.Stat(srcPath)
+	if err != nil {
+		return 0, err
+	}
+
+	if !sourceFileStat.Mode().IsRegular() {
+		return 0, fmt.Errorf("%s is not a regular file", srcPath)
+	}
+
+	source, err := os.Open(srcPath)
+	if err != nil {
+		return 0, err
+	}
+	defer source.Close()
+	defer dest.Close()
+	return io.Copy(dest, source)
+}
+
+func persistToDirectory(seg segment.UnpersistedSegment, d index.Directory,
+	path string) error {
+	if d == nil {
+		return seg.Persist(path)
+	}
+
+	sg, ok := seg.(io.WriterTo)
+	if !ok {
+		return fmt.Errorf("no io.WriterTo segment implementation found")
+	}
+
+	w, err := d.GetWriter(filepath.Join("store", filepath.Base(path)))
+	if err != nil {
+		return err
+	}
+
+	_, err = sg.WriteTo(w)
+	w.Close()
+
+	return err
+}
+
 func prepareBoltSnapshot(snapshot *IndexSnapshot, tx *bolt.Tx, path string,
-	segPlugin SegmentPlugin) ([]string, map[uint64]string, error) {
+	segPlugin SegmentPlugin, d index.Directory) (
+	[]string, map[uint64]string, error) {
 	snapshotsBucket, err := tx.CreateBucketIfNotExists(boltSnapshotsBucket)
 	if err != nil {
 		return nil, nil, err
@@ -481,7 +533,11 @@ func prepareBoltSnapshot(snapshot *IndexSnapshot, tx *bolt.Tx, path string,
 		switch seg := segmentSnapshot.segment.(type) {
 		case segment.PersistedSegment:
 			segPath := seg.Path()
-			filename := strings.TrimPrefix(segPath, path+string(os.PathSeparator))
+			_, err = copyToDirectory(segPath, d)
+			if err != nil {
+				return nil, nil, fmt.Errorf("segment: %s copy err: %v", segPath, err)
+			}
+			filename := filepath.Base(segPath)
 			err = snapshotSegmentBucket.Put(boltPathKey, []byte(filename))
 			if err != nil {
 				return nil, nil, err
@@ -490,10 +546,10 @@ func prepareBoltSnapshot(snapshot *IndexSnapshot, tx *bolt.Tx, path string,
 		case segment.UnpersistedSegment:
 			// need to persist this to disk
 			filename := zapFileName(segmentSnapshot.id)
-			path := path + string(os.PathSeparator) + filename
-			err = seg.Persist(path)
+			path := filepath.Join(path, filename)
+			err := persistToDirectory(seg, d, path)
 			if err != nil {
-				return nil, nil, fmt.Errorf("error persisting segment: %v", err)
+				return nil, nil, fmt.Errorf("segment: %s persist err: %v", path, err)
 			}
 			newSegmentPaths[segmentSnapshot.id] = path
 			err = snapshotSegmentBucket.Put(boltPathKey, []byte(filename))
@@ -534,7 +590,7 @@ func (s *Scorch) persistSnapshotDirect(snapshot *IndexSnapshot) (err error) {
 		}
 	}()
 
-	filenames, newSegmentPaths, err := prepareBoltSnapshot(snapshot, tx, s.path, s.segPlugin)
+	filenames, newSegmentPaths, err := prepareBoltSnapshot(snapshot, tx, s.path, s.segPlugin, nil)
 	if err != nil {
 		return err
 	}
diff --git a/index/scorch/snapshot_index.go b/index/scorch/snapshot_index.go
index ac2f3441..98282eae 100644
--- a/index/scorch/snapshot_index.go
+++ b/index/scorch/snapshot_index.go
@@ -18,6 +18,8 @@ import (
 	"container/heap"
 	"encoding/binary"
 	"fmt"
+	"os"
+	"path/filepath"
 	"reflect"
 	"sort"
 	"sync"
@@ -29,6 +31,7 @@ import (
 	segment "github.com/blevesearch/scorch_segment_api/v2"
 	"github.com/blevesearch/vellum"
 	lev "github.com/blevesearch/vellum/levenshtein"
+	bolt "go.etcd.io/bbolt"
 )
 
 // re usable, threadsafe levenshtein builders
@@ -762,3 +765,46 @@ OUTER:
 	}
 	return rv
 }
+
+func (i *IndexSnapshot) CopyTo(d index.Directory) error {
+	// get the root bolt file.
+	w, err := d.GetWriter(filepath.Join("store", "root.bolt"))
+	if err != nil || w == nil {
+		return fmt.Errorf("failed to create the root.bolt file, err: %v", err)
+	}
+	rootFile, ok := w.(*os.File)
+	if !ok {
+		return fmt.Errorf("invalid root.bolt file found")
+	}
+
+	copyBolt, err := bolt.Open(rootFile.Name(), 0600, nil)
+	if err != nil {
+		return err
+	}
+	defer func() {
+		w.Close()
+		if cerr := copyBolt.Close(); cerr != nil && err == nil {
+			err = cerr
+		}
+	}()
+
+	// start a write transaction
+	tx, err := copyBolt.Begin(true)
+	if err != nil {
+		return err
+	}
+
+	_, _, err = prepareBoltSnapshot(i, tx, "", i.parent.segPlugin, d)
+	if err != nil {
+		_ = tx.Rollback()
+		return fmt.Errorf("error backing up index snapshot: %v", err)
+	}
+
+	// commit bolt data
+	err = tx.Commit()
+	if err != nil {
+		return fmt.Errorf("error commit tx to backup root bolt: %v", err)
+	}
+
+	return copyBolt.Sync()
+}
diff --git a/index_impl.go b/index_impl.go
index 879a3663..8a9cfd3b 100644
--- a/index_impl.go
+++ b/index_impl.go
@@ -18,7 +18,9 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
 	"os"
+	"path/filepath"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -910,3 +912,49 @@ func (m *searchHitSorter) Less(i, j int) bool {
 	c := m.sort.Compare(m.cachedScoring, m.cachedDesc, m.hits[i], m.hits[j])
 	return c < 0
 }
+
+func (i *indexImpl) CopyTo(d index.Directory) (err error) {
+	i.mutex.RLock()
+	defer i.mutex.RUnlock()
+
+	if !i.open {
+		return ErrorIndexClosed
+	}
+
+	indexReader, err := i.i.Reader()
+	if err != nil {
+		return err
+	}
+	defer func() {
+		if cerr := indexReader.Close(); err == nil && cerr != nil {
+			err = cerr
+		}
+	}()
+
+	irc, ok := indexReader.(IndexCopyable)
+	if !ok {
+		return fmt.Errorf("index implementation does not support copy")
+	}
+
+	err = irc.CopyTo(d)
+	if err != nil {
+		return fmt.Errorf("error copying index metadata: %v", err)
+	}
+
+	// copy the metadata
+	return i.meta.CopyTo(d)
+}
+
+func (f FileSystemDirectory) GetWriter(filePath string) (io.WriteCloser,
+	error) {
+	dir, file := filepath.Split(filePath)
+	if dir != "" {
+		err := os.MkdirAll(filepath.Join(string(f), dir), os.ModePerm)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return os.OpenFile(filepath.Join(string(f), dir, file),
+		os.O_RDWR|os.O_CREATE, 0600)
+}
diff --git a/index_meta.go b/index_meta.go
index fe0ddeba..711e0c1b 100644
--- a/index_meta.go
+++ b/index_meta.go
@@ -16,11 +16,13 @@ package bleve
 
 import (
 	"encoding/json"
+	"fmt"
 	"io/ioutil"
 	"os"
 	"path/filepath"
 
 	"github.com/blevesearch/bleve/v2/index/upsidedown"
+	index "github.com/blevesearch/bleve_index_api"
 )
 
 const metaFilename = "index_meta.json"
@@ -92,6 +94,23 @@ func (i *indexMeta) Save(path string) (err error) {
 	return nil
 }
 
+func (i *indexMeta) CopyTo(d index.Directory) (err error) {
+	metaBytes, err := json.Marshal(i)
+	if err != nil {
+		return err
+	}
+
+	w, err := d.GetWriter(metaFilename)
+	if w == nil || err != nil {
+		return fmt.Errorf("invalid writer for file: %s, err: %v",
+			metaFilename, err)
+	}
+	defer w.Close()
+
+	_, err = w.Write(metaBytes)
+	return err
+}
+
 func indexMetaPath(path string) string {
 	return filepath.Join(path, metaFilename)
 }


Test patch:
diff --git a/index_test.go b/index_test.go
index a3d0d1dc..c0af3689 100644
--- a/index_test.go
+++ b/index_test.go
@@ -2261,3 +2261,227 @@ func TestIndexMappingDocValuesDynamic(t *testing.T) {
 		t.Fatalf("Expected DocValuesDynamic to remain false after the index mapping edit")
 	}
 }
+
+func TestCopyIndex(t *testing.T) {
+	tmpIndexPath := createTmpIndexPath(t)
+	defer cleanupTmpIndexPath(t, tmpIndexPath)
+
+	idx, err := New(tmpIndexPath, NewIndexMapping())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		err := idx.Close()
+		if err != nil {
+			t.Fatal(err)
+		}
+	}()
+
+	doca := map[string]interface{}{
+		"name": "tester",
+		"desc": "gophercon india testing",
+	}
+	err = idx.Index("a", doca)
+	if err != nil {
+		t.Error(err)
+	}
+
+	docy := map[string]interface{}{
+		"name": "jasper",
+		"desc": "clojure",
+	}
+	err = idx.Index("y", docy)
+	if err != nil {
+		t.Error(err)
+	}
+
+	err = idx.Delete("y")
+	if err != nil {
+		t.Error(err)
+	}
+
+	docx := map[string]interface{}{
+		"name": "rose",
+		"desc": "xoogler",
+	}
+	err = idx.Index("x", docx)
+	if err != nil {
+		t.Error(err)
+	}
+
+	err = idx.SetInternal([]byte("status"), []byte("pending"))
+	if err != nil {
+		t.Error(err)
+	}
+
+	docb := map[string]interface{}{
+		"name": "sree",
+		"desc": "cbft janitor",
+	}
+	batch := idx.NewBatch()
+	err = batch.Index("b", docb)
+	if err != nil {
+		t.Error(err)
+	}
+	batch.Delete("x")
+	batch.SetInternal([]byte("batchi"), []byte("batchv"))
+	batch.DeleteInternal([]byte("status"))
+	err = idx.Batch(batch)
+	if err != nil {
+		t.Error(err)
+	}
+	val, err := idx.GetInternal([]byte("batchi"))
+	if err != nil {
+		t.Error(err)
+	}
+	if string(val) != "batchv" {
+		t.Errorf("expected 'batchv', got '%s'", val)
+	}
+	val, err = idx.GetInternal([]byte("status"))
+	if err != nil {
+		t.Error(err)
+	}
+	if val != nil {
+		t.Errorf("expected nil, got '%s'", val)
+	}
+
+	err = idx.SetInternal([]byte("seqno"), []byte("7"))
+	if err != nil {
+		t.Error(err)
+	}
+	err = idx.SetInternal([]byte("status"), []byte("ready"))
+	if err != nil {
+		t.Error(err)
+	}
+	err = idx.DeleteInternal([]byte("status"))
+	if err != nil {
+		t.Error(err)
+	}
+	val, err = idx.GetInternal([]byte("status"))
+	if err != nil {
+		t.Error(err)
+	}
+	if val != nil {
+		t.Errorf("expected nil, got '%s'", val)
+	}
+
+	val, err = idx.GetInternal([]byte("seqno"))
+	if err != nil {
+		t.Error(err)
+	}
+	if string(val) != "7" {
+		t.Errorf("expected '7', got '%s'", val)
+	}
+
+	count, err := idx.DocCount()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if count != 2 {
+		t.Errorf("expected doc count 2, got %d", count)
+	}
+
+	doc, err := idx.Document("a")
+	if err != nil {
+		t.Fatal(err)
+	}
+	foundNameField := false
+	doc.VisitFields(func(field index.Field) {
+		if field.Name() == "name" && string(field.Value()) == "tester" {
+			foundNameField = true
+		}
+	})
+	if !foundNameField {
+		t.Errorf("expected to find field named 'name' with value 'tester'")
+	}
+
+	fields, err := idx.Fields()
+	if err != nil {
+		t.Fatal(err)
+	}
+	expectedFields := map[string]bool{
+		"_all": false,
+		"name": false,
+		"desc": false,
+	}
+	if len(fields) < len(expectedFields) {
+		t.Fatalf("expected %d fields got %d", len(expectedFields), len(fields))
+	}
+	for _, f := range fields {
+		expectedFields[f] = true
+	}
+	for ef, efp := range expectedFields {
+		if !efp {
+			t.Errorf("field %s is missing", ef)
+		}
+	}
+
+	// now create a copy of the index, and repeat assertions on it
+	copyableIndex, ok := idx.(IndexCopyable)
+	if !ok {
+		t.Fatal("index doesn't support copy")
+	}
+	backupIndexPath := createTmpIndexPath(t)
+	defer cleanupTmpIndexPath(t, backupIndexPath)
+
+	err = copyableIndex.CopyTo(FileSystemDirectory(backupIndexPath))
+	if err != nil {
+		t.Fatalf("error copying the index: %v", err)
+	}
+
+	// open the copied  index
+	idxCopied, err := Open(backupIndexPath)
+	if err != nil {
+		t.Fatalf("unable to open copy index")
+	}
+	defer func() {
+		err := idxCopied.Close()
+		if err != nil {
+			t.Fatalf("error closing copy index: %v", err)
+		}
+	}()
+
+	// assertions on copied index
+	copyCount, err := idxCopied.DocCount()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if copyCount != 2 {
+		t.Errorf("expected doc count 2, got %d", copyCount)
+	}
+
+	copyDoc, err := idxCopied.Document("a")
+	if err != nil {
+		t.Fatal(err)
+	}
+	copyFoundNameField := false
+	copyDoc.VisitFields(func(field index.Field) {
+		if field.Name() == "name" && string(field.Value()) == "tester" {
+			copyFoundNameField = true
+		}
+	})
+	if !copyFoundNameField {
+		t.Errorf("expected copy index to find field named 'name' with value 'tester'")
+	}
+
+	copyFields, err := idx.Fields()
+	if err != nil {
+		t.Fatal(err)
+	}
+	copyExpectedFields := map[string]bool{
+		"_all": false,
+		"name": false,
+		"desc": false,
+	}
+	if len(copyFields) < len(copyExpectedFields) {
+		t.Fatalf("expected %d fields got %d", len(copyExpectedFields), len(copyFields))
+	}
+	for _, f := range copyFields {
+		copyExpectedFields[f] = true
+	}
+	for ef, efp := range copyExpectedFields {
+		if !efp {
+			t.Errorf("copy field %s is missing", ef)
+		}
+	}
+}