Bugfix NGT flush logic

pkg/agent/core/ngt/handler/grpc/index.go

@@ -64,6 +64,9 @@
 					},
 				}, info.Get())...)
 			code = codes.FailedPrecondition
+		case errors.Is(err, errors.ErrFlushingIsInProgress):
+			err = status.WrapWithAborted("CreateIndex API aborted to process create indexes request due to flushing indices is in progress", err, details...)
+			code = codes.Aborted
 		case errors.Is(err, context.Canceled):
 			err = status.WrapWithCanceled(fmt.Sprintf("CreateIndex API canceled to create indexes pool_size = %d, error: %v", c.GetPoolSize(), err), err, details...)
 			code = codes.Canceled
@@ -149,6 +152,9 @@
 					},
 				}, info.Get())...)
 			code = codes.FailedPrecondition
+		case errors.Is(err, errors.ErrFlushingIsInProgress):
+			err = status.WrapWithAborted("CreateAndSaveIndex API aborted to process create indexes request due to flushing indices is in progress", err, details...)
+			code = codes.Aborted
 		case errors.Is(err, context.Canceled):
 			err = status.WrapWithCanceled(fmt.Sprintf("CreateAndSaveIndex API canceled to create indexes pool_size = %d, error: %v", c.GetPoolSize(), err), err, details...)
 			code = codes.Canceled

pkg/agent/core/ngt/service/ngt.go

@@ -908,7 +908,7 @@
 				}
 				return ctx.Err()
 			case <-tick.C:
-				if n.vq.IVQLen() >= n.alen {
+				if n != nil && n.vq != nil && !n.IsFlushing() && n.vq.IVQLen() >= n.alen {
 					err = n.CreateIndex(ctx, n.poolSize)
 				}
 			case <-limit.C:
@@ -1249,7 +1249,7 @@
 	runtime.GC()
 	atomic.AddUint64(&n.nogce, 1)
 
-	if n.inMem {
+	if !n.inMem {
 		// delete file
 		err = file.DeleteDir(ctx, n.path)
 		if err != nil {
@@ -1265,20 +1265,20 @@
 		}
 	}
 
-	nkvs := kvs.New(kvs.WithConcurrency(n.kvsdbConcurrency))
-
-	nvq, err := vqueue.New()
-	if err != nil {
-		log.Errorf("failed to create new vector vector queue. error: %v", err)
-	}
+	// nkvs := kvs.New(kvs.WithConcurrency(n.kvsdbConcurrency))
+	//
+	// nvq, err := vqueue.New()
+	// if err != nil {
+	// 	log.Errorf("failed to create new vector vector queue. error: %v", err)
+	// }
 
 	// renew instance
 	nn, err := newNGT(n.cfg, n.opts...)
 	if err != nil {
 		return err
 	}
-	nn.kvs = nkvs
-	nn.vq = nvq
+	// nn.kvs = nkvs
+	// nn.vq = nvq
 
 	// Regenerate with flags set
 	nn.flushing.Store(true)
@@ -1299,8 +1299,11 @@
 		}
 	}()
 
-	if n.isReadReplica {
+	switch {
+	case n.isReadReplica:
 		return errors.ErrWriteOperationToReadReplica
+	case n.IsFlushing():
+		return errors.ErrFlushingIsInProgress
 	}
 
 	ic := n.vq.IVQLen() + n.vq.DVQLen()
@@ -1941,15 +1944,24 @@
 }
 
 func (n *ngt) Len() uint64 {
-	return n.kvs.Len()
+	if n != nil && n.kvs != nil && !n.IsFlushing() {
+		return n.kvs.Len()
+	}
+	return 0
 }
 
 func (n *ngt) InsertVQueueBufferLen() uint64 {
-	return uint64(n.vq.IVQLen())
+	if n != nil && n.vq != nil && !n.IsFlushing() {
+		return uint64(n.vq.IVQLen())
+	}
+	return 0
 }
 
 func (n *ngt) DeleteVQueueBufferLen() uint64 {
-	return uint64(n.vq.DVQLen())
+	if n != nil && n.vq != nil && !n.IsFlushing() {
+		return uint64(n.vq.DVQLen())
+	}
+	return 0
 }
 
 func (n *ngt) GetDimensionSize() int {