invest removeInvalidIndex bug

Signed-off-by: kpango <[email protected]>
vdaas · Feb 20, 2024 · fbebc42 · fbebc42
1 parent f080436
commit fbebc42
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 0 deletions.
diff --git a/pkg/agent/core/ngt/service/ngt.go b/pkg/agent/core/ngt/service/ngt.go
@@ -1274,12 +1274,15 @@ func (n *ngt) CreateIndex(ctx context.Context, poolSize uint32) (err error) {
 }
 
 func (n *ngt) removeInvalidIndex(ctx context.Context) {
+	log.Debug("[CHECK20240220]\tremoveInvalidIndex started")
 	if n.kvs.Len() == 0 {
+		log.Debug("[CHECK20240220]\tremoveInvalidIndex kvs Len is empty")
 		return
 	}
 	var dcnt uint32
 	n.kvs.Range(ctx, func(uuid string, oid uint32, _ int64) bool {
 		vec, err := n.core.GetVector(uint(oid))
+		log.Debugf("[CHECK20240220]\tuuid: %s,\toid: %d,\tvec: %#v,\terr: %v", uuid, oid, vec, err)
 		if err != nil || vec == nil || len(vec) != n.dim {
 			log.Debugf("invalid index detected err: %v\tuuid: %s\toid: %d will remove", err, uuid, oid)
 			n.kvs.Delete(uuid)
@@ -1295,6 +1298,7 @@ func (n *ngt) removeInvalidIndex(ctx context.Context) {
 		return true
 	})
 	if atomic.LoadUint32(&dcnt) <= 0 {
+		log.Debug("[CHECK20240220]\tremoveInvalidIndex deleted count is empty")
 		return
 	}
 	var poolSize uint32
@@ -1303,6 +1307,8 @@ func (n *ngt) removeInvalidIndex(ctx context.Context) {
 	} else {
 		poolSize = atomic.LoadUint32(&dcnt)
 	}
+	log.Debug("[CHECK20240220]\tremoveInvalidIndex before cimu.Lock")
+	defer log.Debug("[CHECK20240220]\tremoveInvalidIndex after cimu.Lock")
 	n.cimu.Lock()
 	defer n.cimu.Unlock()
 	n.indexing.Store(true)

diff --git a/pkg/agent/internal/kvs/kvs.go b/pkg/agent/internal/kvs/kvs.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"sync/atomic"
 
+	"github.com/vdaas/vald/internal/errors"
 	"github.com/vdaas/vald/internal/safety"
 	"github.com/vdaas/vald/internal/sync"
 	"github.com/vdaas/vald/internal/sync/errgroup"
@@ -159,12 +160,17 @@ func (b *bidi) Range(ctx context.Context, f func(string, uint32, int64) bool) {
 			b.uo[idx].Range(func(uuid string, val ValueStructUo) bool {
 				select {
 				case <-ctx.Done():
+					err = ctx.Err()
 					return false
 				default:
 					return f(uuid, val.value, val.timestamp)
 				}
 			})
 			wg.Done()
+			if err != nil &&
+				(errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)) {
+				return err
+			}
 			return nil
 		}))
 	}