Wrap the code indexer (#9476)

* Wrap the code indexer

In order to prevent a data race in the code indexer it must be wrapped
with a holder otherwise it is possible to Search/Index on an
incompletely initialised indexer, and search will fail with a nil
pointer until the repository indexer is initialised.

Further a completely initialised repository indexer should not be closed
until Termination otherwise actions in Hammer/Shutdown phases could
block or be lost.

Finally, there is a complex dance of shutdown etiquette should the index
initialisation fail. This PR restores that.

* Always return err if closed whilst waiting

Co-authored-by: techknowlogick <matti@mdranta.net>

modules/indexer/code/indexer.go

@@ -5,17 +5,15 @@
 package code
 
 import (
+	"context"
+	"os"
 	"time"
 
 	"code.gitea.io/gitea/modules/graceful"
 	"code.gitea.io/gitea/modules/log"
 	"code.gitea.io/gitea/modules/setting"
 )
 
-var (
-	indexer Indexer
-)
-
 // SearchResult result of performing a search in a repo
 type SearchResult struct {
 	RepoID     int64
@@ -36,28 +34,45 @@ type Indexer interface {
 // Init initialize the repo indexer
 func Init() {
 	if !setting.Indexer.RepoIndexerEnabled {
+		indexer.Close()
 		return
 	}
 
+	ctx, cancel := context.WithCancel(context.Background())
+
+	graceful.GetManager().RunAtTerminate(ctx, func() {
+		log.Debug("Closing repository indexer")
+		indexer.Close()
+		log.Info("PID: %d Repository Indexer closed", os.Getpid())
+	})
+
 	waitChannel := make(chan time.Duration)
 	go func() {
 		start := time.Now()
-		log.Info("Initializing Repository Indexer")
-		var created bool
-		var err error
-		indexer, created, err = NewBleveIndexer(setting.Indexer.RepoPath)
+		log.Info("PID: %d Initializing Repository Indexer at: %s", os.Getpid(), setting.Indexer.RepoPath)
+		bleveIndexer, created, err := NewBleveIndexer(setting.Indexer.RepoPath)
 		if err != nil {
+			if bleveIndexer != nil {
+				bleveIndexer.Close()
+			}
+			cancel()
 			indexer.Close()
-			log.Fatal("indexer.Init: %v", err)
+			close(waitChannel)
+			log.Fatal("PID: %d Unable to initialize the Repository Indexer at path: %s Error: %v", os.Getpid(), setting.Indexer.RepoPath, err)
 		}
+		indexer.set(bleveIndexer)
 
 		go processRepoIndexerOperationQueue(indexer)
 
 		if created {
 			go populateRepoIndexer()
 		}
+		select {
+		case waitChannel <- time.Since(start):
+		case <-graceful.GetManager().IsShutdown():
+		}
 
-		waitChannel <- time.Since(start)
+		close(waitChannel)
 	}()
 
 	if setting.Indexer.StartupTimeout > 0 {
@@ -67,9 +82,21 @@ func Init() {
 				timeout += setting.GracefulHammerTime
 			}
 			select {
-			case duration := <-waitChannel:
+			case <-graceful.GetManager().IsShutdown():
+				log.Warn("Shutdown before Repository Indexer completed initialization")
+				cancel()
+				indexer.Close()
+			case duration, ok := <-waitChannel:
+				if !ok {
+					log.Warn("Repository Indexer Initialization failed")
+					cancel()
+					indexer.Close()
+					return
+				}
 				log.Info("Repository Indexer Initialization took %v", duration)
 			case <-time.After(timeout):
+				cancel()
+				indexer.Close()
 				log.Fatal("Repository Indexer Initialization Timed-Out after: %v", timeout)
 			}
 		}()

modules/indexer/code/queue.go

@@ -22,8 +22,6 @@ type repoIndexerOperation struct {
 var repoIndexerOperationQueue chan repoIndexerOperation
 
 func processRepoIndexerOperationQueue(indexer Indexer) {
-	defer indexer.Close()
-
 	repoIndexerOperationQueue = make(chan repoIndexerOperation, setting.Indexer.UpdateQueueLength)
 	for {
 		select {

modules/indexer/code/wrapped.go

@@ -0,0 +1,94 @@
+// Copyright 2019 The Gitea Authors. All rights reserved.
+// Use of this source code is governed by a MIT-style
+// license that can be found in the LICENSE file.
+
+package code
+
+import (
+	"fmt"
+	"sync"
+)
+
+var (
+	indexer = newWrappedIndexer()
+)
+
+// ErrWrappedIndexerClosed is the error returned if the indexer was closed before it was ready
+var ErrWrappedIndexerClosed = fmt.Errorf("Indexer closed before ready")
+
+type wrappedIndexer struct {
+	internal Indexer
+	lock     sync.RWMutex
+	cond     *sync.Cond
+	closed   bool
+}
+
+func newWrappedIndexer() *wrappedIndexer {
+	w := &wrappedIndexer{}
+	w.cond = sync.NewCond(w.lock.RLocker())
+	return w
+}
+
+func (w *wrappedIndexer) set(indexer Indexer) {
+	w.lock.Lock()
+	defer w.lock.Unlock()
+	if w.closed {
+		// Too late!
+		indexer.Close()
+	}
+	w.internal = indexer
+	w.cond.Broadcast()
+}
+
+func (w *wrappedIndexer) get() (Indexer, error) {
+	w.lock.RLock()
+	defer w.lock.RUnlock()
+	if w.internal == nil {
+		if w.closed {
+			return nil, ErrWrappedIndexerClosed
+		}
+		w.cond.Wait()
+		if w.closed {
+			return nil, ErrWrappedIndexerClosed
+		}
+	}
+	return w.internal, nil
+}
+
+func (w *wrappedIndexer) Index(repoID int64) error {
+	indexer, err := w.get()
+	if err != nil {
+		return err
+	}
+	return indexer.Index(repoID)
+}
+
+func (w *wrappedIndexer) Delete(repoID int64) error {
+	indexer, err := w.get()
+	if err != nil {
+		return err
+	}
+	return indexer.Delete(repoID)
+}
+
+func (w *wrappedIndexer) Search(repoIDs []int64, keyword string, page, pageSize int) (int64, []*SearchResult, error) {
+	indexer, err := w.get()
+	if err != nil {
+		return 0, nil, err
+	}
+	return indexer.Search(repoIDs, keyword, page, pageSize)
+
+}
+
+func (w *wrappedIndexer) Close() {
+	w.lock.Lock()
+	defer w.lock.Unlock()
+	if w.closed {
+		return
+	}
+	w.closed = true
+	w.cond.Broadcast()
+	if w.internal != nil {
+		w.internal.Close()
+	}
+}