fix overutil CPU

statefun/cache/backup_write_barrier.go

@@ -49,11 +49,14 @@ func (cs *Store) checkBackupBarrierInfoBeforeWrite(opTime int64) error {
 		}
 	}
 
-	lg.Logf(lg.InfoLevel, "clearBackupBarrier: backup barrier cleared")
-
 	return nil
 }
 
+func (cs *Store) IsBackupBarrierActive() bool {
+	_, status := cs.getBackupBarrierState()
+	return status == BackupBarrierStatusLocked || status == BackupBarrierStatusLocking
+}
+
 func (cs *Store) getBackupBarrierState() (timestamp int64, status int32) {
 	if cs.shouldRefreshBackupBarrier() {
 		cs.refreshBackupBarrierFromKV()
@@ -73,6 +76,7 @@ func (cs *Store) refreshBackupBarrierFromKV() {
 	barrier, err := cs.getBackupBarrierInfo()
 	if err != nil {
 		lg.Logf(lg.ErrorLevel, "refreshBackupBarrierFromKV: failed to read barrier from KV: %s", err)
+		atomic.StoreInt64(&cs.backupBarrierLastChecked, system.GetCurrentTimeNs())
 		return
 	}
 

statefun/cache/cache.go

@@ -510,11 +510,17 @@ func NewCacheStore(ctx context.Context, cacheConfig *Config, js nats.JetStreamCo
 		defer system.GlobalPrometrics.GetRoutinesCounter().Stopped("cache.storeUpdatesHandler")
 		if w, err := kv.Watch(cacheConfig.kvStorePrefix+".>", nats.IgnoreDeletes()); err == nil {
 			defer system.MsgOnErrorReturn(w.Stop())
-			for {
+			activeKVSync := true
+			for activeKVSync {
 				select {
 				case <-cs.ctx.Done():
-					return
-				case entry := <-w.Updates():
+					activeKVSync = false
+				case entry, ok := <-w.Updates():
+					if !ok {
+						le.Warnf(ctx, "storeUpdatesHandler: KV watcher channel closed unexpectedly")
+						activeKVSync = false
+						break
+					}
 					if entry != nil {
 						key := cs.fromStoreKey(entry.Key())
 						valueBytes := entry.Value()

statefun/domain.go

@@ -10,7 +10,6 @@ import (
 
 	"github.com/nats-io/nats.go"
 
-	"github.com/foliagecp/easyjson"
 	"github.com/foliagecp/sdk/embedded/nats/kv"
 	"github.com/foliagecp/sdk/statefun/cache"
 	lg "github.com/foliagecp/sdk/statefun/logger"
@@ -681,15 +680,8 @@ func (dm *Domain) GenerateTransactionID() string {
 }
 
 func (dm *Domain) isBackupBarrierActive() bool {
-	entry, err := dm.kv.Get(cache.BackupBarrierLockKey)
-	if err != nil {
-		lg.Logf(lg.ErrorLevel, "IsBackupBarrierActive: failed to get backup barrier lock entry: %s", err)
-		return false
-	}
-	barrier, ok := easyjson.JSONFromBytes(entry.Value())
-	if !ok {
+	if dm.cache == nil {
 		return false
 	}
-	status := int32(barrier.GetByPath("status").AsNumericDefault(cache.BackupBarrierStatusUnlocked))
-	return status == cache.BackupBarrierStatusLocked
+	return dm.cache.IsBackupBarrierActive()
 }