Fix/get open challenges (#1181)

* Fix

* Fix

* Fix

* Fix

* Fix

* Fix

* Fix

* Improved logging

* Fix challenge table

Author: Jayashsatolia403

code/go/0chain.net/blobbercore/challenge/challenge.go

@@ -24,25 +24,23 @@ type BCChallengeResponse struct {
 	Challenges []*ChallengeEntity `json:"challenges"`
 }
 
-var lastChallengeTimestamp int
+var lastChallengeRound int64
 
 func syncOpenChallenges(ctx context.Context) {
-	const incrOffset = 20
 	defer func() {
 		if r := recover(); r != nil {
 			logging.Logger.Error("[recover]challenge", zap.Any("err", r))
 		}
 	}()
 
-	offset := 0
 	params := make(map[string]string)
 	params["blobber"] = node.Self.ID
-	params["offset"] = strconv.Itoa(offset)
-	params["limit"] = "20"
-	if lastChallengeTimestamp > 0 {
-		params["from"] = strconv.Itoa(lastChallengeTimestamp)
+
+	params["limit"] = "50"
+	if lastChallengeRound > 0 {
+		params["from"] = strconv.FormatInt(lastChallengeRound, 10)
 	}
-	logging.Logger.Info("[challenge]sync:pull", zap.Any("params", params))
+
 	start := time.Now()
 
 	var downloadElapsed, jsonElapsed time.Duration
@@ -54,6 +52,9 @@ func syncOpenChallenges(ctx context.Context) {
 			return
 		default:
 		}
+
+		logging.Logger.Info("[challenge]sync:pull", zap.Any("params", params))
+
 		var challenges BCChallengeResponse
 		var challengeIDs []string
 		challenges.Challenges = make([]*ChallengeEntity, 0)
@@ -75,13 +76,13 @@ func syncOpenChallenges(ctx context.Context) {
 			break
 		}
 		sort.Slice(challenges.Challenges, func(i, j int) bool {
-			return challenges.Challenges[i].CreatedAt < challenges.Challenges[j].CreatedAt
+			return challenges.Challenges[i].RoundCreatedAt < challenges.Challenges[j].RoundCreatedAt
 		})
 		count += len(challenges.Challenges)
 		for _, c := range challenges.Challenges {
 			challengeIDs = append(challengeIDs, c.ChallengeID)
-			if c.CreatedAt > common.Timestamp(lastChallengeTimestamp) {
-				lastChallengeTimestamp = int(c.CreatedAt)
+			if c.RoundCreatedAt >= lastChallengeRound {
+				lastChallengeRound = c.RoundCreatedAt
 			}
 			toProcessChallenge <- c
 		}
@@ -93,8 +94,6 @@ func syncOpenChallenges(ctx context.Context) {
 		if len(challenges.Challenges) == 0 {
 			break
 		}
-		offset += incrOffset
-		params["offset"] = strconv.Itoa(offset)
 	}
 
 	dbTimeStart := time.Now()
@@ -110,6 +109,11 @@ func syncOpenChallenges(ctx context.Context) {
 
 func validateOnValidators(c *ChallengeEntity) {
 
+	logging.Logger.Info("[challenge]validate: ",
+		zap.Any("challenge", c),
+		zap.String("challenge_id", c.ChallengeID),
+	)
+
 	ctx := datastore.GetStore().CreateTransaction(context.TODO())
 	defer ctx.Done()
 
@@ -119,7 +123,7 @@ func validateOnValidators(c *ChallengeEntity) {
 		logging.Logger.Error("[challengetiming]add: ",
 			zap.String("challenge_id", c.ChallengeID),
 			zap.Error(err))
-		deleteChallenge(int64(c.CreatedAt))
+		deleteChallenge(c.RoundCreatedAt)
 		tx.Rollback()
 	}
 
@@ -150,7 +154,7 @@ func validateOnValidators(c *ChallengeEntity) {
 			zap.Time("created", createdTime),
 			zap.Error(err))
 		//TODO: Should we delete the challenge from map or send it back to the todo channel?
-		deleteChallenge(int64(c.CreatedAt))
+		deleteChallenge(c.RoundCreatedAt)
 		tx.Rollback()
 		return
 	}

code/go/0chain.net/blobbercore/challenge/entity.go

@@ -97,8 +97,9 @@ type ChallengeEntity struct {
 	Timestamp               common.Timestamp      `gorm:"column:timestamp;not null;default:0" json:"timestamp"`
 
 	// This time is taken from Blockchain challenge object.
-	CreatedAt common.Timestamp `gorm:"created_at" json:"created"`
-	UpdatedAt time.Time        `gorm:"updated_at;type:timestamp without time zone;not null;default:current_timestamp" json:"-"`
+	RoundCreatedAt int64            `gorm:"round_created_at" json:"round_created_at"`
+	CreatedAt      common.Timestamp `gorm:"created_at" json:"created"`
+	UpdatedAt      time.Time        `gorm:"updated_at;type:timestamp without time zone;not null;default:current_timestamp" json:"-"`
 }
 
 func (ChallengeEntity) TableName() string {

code/go/0chain.net/blobbercore/challenge/protocol.go

@@ -48,7 +48,7 @@ type ChallengeResponse struct {
 func (cr *ChallengeEntity) CancelChallenge(ctx context.Context, errReason error) {
 	cancellation := time.Now()
 	db := datastore.GetStore().GetTransaction(ctx)
-	deleteChallenge(int64(cr.CreatedAt))
+	deleteChallenge(cr.RoundCreatedAt)
 	cr.Status = Cancelled
 	cr.StatusMessage = errReason.Error()
 	cr.UpdatedAt = cancellation.UTC()

code/go/0chain.net/blobbercore/challenge/worker.go

@@ -81,6 +81,8 @@ func challengeProcessor(ctx context.Context) {
 			return
 
 		case it := <-toProcessChallenge:
+
+			logging.Logger.Info("processing_challenge", zap.Any("challenge_id", it.ChallengeID))
 			if ok := it.createChallenge(); !ok {
 				continue
 			}
@@ -142,7 +144,7 @@ func commitOnChainWorker(ctx context.Context) {
 					}()
 					err := challenge.VerifyChallengeTransaction(txn)
 					if err == nil || err != ErrEntityNotFound {
-						deleteChallenge(int64(challenge.CreatedAt))
+						deleteChallenge(int64(challenge.RoundCreatedAt))
 					}
 				}(&chall)
 			}
@@ -175,11 +177,11 @@ func getBatch(batchSize int) (chall []ChallengeEntity) {
 
 func (it *ChallengeEntity) createChallenge() bool {
 	challengeMapLock.Lock()
-	if _, ok := challengeMap.Get(int64(it.CreatedAt)); ok {
+	if _, ok := challengeMap.Get(it.RoundCreatedAt); ok {
 		challengeMapLock.Unlock()
 		return false
 	}
-	challengeMap.Put(int64(it.CreatedAt), it)
+	challengeMap.Put(it.RoundCreatedAt, it)
 	challengeMapLock.Unlock()
 	return true
 }

goose/migrations/001_blobber_meta.sql

@@ -140,6 +140,7 @@ CREATE TABLE public.challenges (
     object_path jsonb,
     sequence bigint,
     "timestamp" bigint DEFAULT 0 NOT NULL,
+    round_created_at bigint,
     created_at bigint,
     updated_at timestamp without time zone DEFAULT CURRENT_TIMESTAMP NOT NULL
 );