Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 20 additions & 8 deletions bridge/setu/processor/checkpoint.go
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ import (
"math"
"math/big"
"strconv"
"sync"
"time"

cliContext "github.com/cosmos/cosmos-sdk/client/context"
Expand Down Expand Up @@ -94,18 +95,30 @@ func (cp *CheckpointProcessor) RegisterTasks() {
}

func (cp *CheckpointProcessor) startPolling(ctx context.Context) {
tickerForNoAck := time.NewTicker(helper.GetConfig().NoACKPollInterval)
now := time.Now()
baseTime := time.Unix(0, 0)
// no-ack ticker interval keep same with checkpoint interval
noAckInterval := helper.GetConfig().CheckpointerPollInterval
// adjust no-ack ticker to tick at the middle of checkpoint interval
firstIntervalForNoAck := noAckInterval - (now.UTC().Sub(baseTime) % noAckInterval) - noAckInterval/2 // nolint: gomnd
if firstIntervalForNoAck <= 0 {
firstIntervalForNoAck += noAckInterval
}

tickerForNoAck := time.NewTicker(firstIntervalForNoAck)
syncInterval := helper.GetConfig().CheckpointerPollInterval / 2
tickerForSync := time.NewTicker(syncInterval)
// stop ticker when everything done
defer tickerForNoAck.Stop()
defer tickerForSync.Stop()

cp.Logger.Info("Start polling", "no-ack-interval", helper.GetConfig().NoACKPollInterval, "checkpoint-sync-interval", syncInterval)
cp.Logger.Info("Start polling", "no-ack-interval", noAckInterval, "checkpoint-sync-interval", syncInterval)

adjustOnce := sync.Once{}
for {
select {
case <-tickerForNoAck.C:
adjustOnce.Do(func() { tickerForNoAck.Reset(noAckInterval) })
go cp.handleCheckpointNoAck()
case <-tickerForSync.C:
go cp.handleCheckpointSync()
Expand Down Expand Up @@ -392,11 +405,11 @@ func (cp *CheckpointProcessor) handleCheckpointNoAck() {
return
}

isNoAckRequired, count := cp.checkIfNoAckIsRequired(checkpointContext, lastCreatedAt)
isNoAckRequired, _ := cp.checkIfNoAckIsRequired(checkpointContext, lastCreatedAt)
if isNoAckRequired {
var isProposer bool

if isProposer, err = util.IsProposerByIndex(cp.cliCtx, count); err != nil {
if isProposer, err = util.IsValidator(cp.cliCtx); err != nil {
cp.Logger.Error("Error checking IsProposer while proposing Checkpoint No-Ack ", "error", err)
return
}
Expand Down Expand Up @@ -797,10 +810,10 @@ func (cp *CheckpointProcessor) checkIfNoAckIsRequired(checkpointContext *Checkpo

checkpointCreationTime := time.Unix(lastCreatedAt, 0)
currentTime := time.Now().UTC()

timeDiff := currentTime.Sub(checkpointCreationTime)
// check if last checkpoint was < NoACK wait time
if timeDiff.Seconds() >= helper.GetConfig().NoACKWaitTime.Seconds() && index == 0 {
index = math.Floor(timeDiff.Seconds() / helper.GetConfig().NoACKWaitTime.Seconds())
if timeDiff.Seconds() >= helper.GetConfig().CheckpointerPollInterval.Seconds() && index == 0 {
index = math.Floor(timeDiff.Seconds() / helper.GetConfig().CheckpointerPollInterval.Seconds())
}

if index == 0 {
Expand All @@ -812,7 +825,6 @@ func (cp *CheckpointProcessor) checkIfNoAckIsRequired(checkpointContext *Checkpo

// check if difference between no-ack time and current time
lastNoAck := cp.getLastNoAckTime()

lastNoAckTime := time.Unix(int64(lastNoAck), 0)
// if last no ack == 0 , first no-ack to be sent
if currentTime.Sub(lastNoAckTime).Seconds() < checkpointParams.CheckpointBufferTime.Seconds() && lastNoAck != 0 {
Expand Down
28 changes: 28 additions & 0 deletions bridge/setu/util/common.go
Original file line number Diff line number Diff line change
Expand Up @@ -120,6 +120,34 @@ func IsProposerByIndex(cliCtx cliContext.CLIContext, index uint64) (bool, error)
return false, nil
}

func IsValidator(cliCtx cliContext.CLIContext) (bool, error) {
var validatorSet hmtypes.ValidatorSet

result, err := helper.FetchFromAPI(cliCtx,
helper.GetHeimdallServerEndpoint(CurrentValidatorSetURL),
)
if err != nil {
logger.Error("Error fetching proposers", "url", CurrentValidatorSetURL, "error", err)

return false, fmt.Errorf("failed to query heimdall server: %w", err)
}

err = json.Unmarshal(result.Result, &validatorSet)
if err != nil {
logger.Error("error unmarshalling proposer slice", "error", err)

return false, fmt.Errorf("failed to unmarshal validatorSet: %w", err)
}

for _, validator := range validatorSet.Validators {
if bytes.Equal(validator.Signer.Bytes(), helper.GetAddress()) {
return true, nil
}
}

return false, nil
}

// IsInProposerList checks if we are in current proposer
func IsInProposerList(cliCtx cliContext.CLIContext, count uint64) (bool, error) {
logger.Debug("Skipping proposers", "count", strconv.FormatUint(count, 10))
Expand Down