(OraklNode) Refresh fetcher when global aggregate and local aggregate price differ by more than 30%

node/pkg/aggregator/aggregator.go

@@ -4,9 +4,10 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
-
+	"math"
 	"time"
 
+	"bisonai.com/miko/node/pkg/bus"
 	"bisonai.com/miko/node/pkg/chain/helper"
 	errorSentinel "bisonai.com/miko/node/pkg/error"
 	"bisonai.com/miko/node/pkg/raft"
@@ -16,7 +17,15 @@ import (
 	"github.com/rs/zerolog/log"
 )
 
-func NewAggregator(h host.Host, ps *pubsub.PubSub, topicString string, config Config, signHelper *helper.Signer, latestLocalAggregates *LatestLocalAggregates) (*Aggregator, error) {
+func NewAggregator(
+	h host.Host,
+	ps *pubsub.PubSub,
+	topicString string,
+	config Config,
+	signHelper *helper.Signer,
+	latestLocalAggregates *LatestLocalAggregates,
+	mb *bus.MessageBus,
+) (*Aggregator, error) {
 	if h == nil || ps == nil || topicString == "" {
 		return nil, errorSentinel.ErrAggregatorInvalidInitValue
 	}
@@ -53,6 +62,9 @@ func NewAggregator(h host.Host, ps *pubsub.PubSub, topicString string, config Co
 		RoundID:               1,
 		Signer:                signHelper,
 		LatestLocalAggregates: latestLocalAggregates,
+		bus:                   mb,
+
+		roundLocalAggregate: map[int32]int64{},
 	}
 	aggregator.Raft.LeaderJob = aggregator.LeaderJob
 	aggregator.Raft.HandleCustomMessage = aggregator.HandleCustomMessage
@@ -102,16 +114,14 @@ func (n *Aggregator) HandleTriggerMessage(ctx context.Context, msg raft.Message)
 		return err
 	}
 	defer n.leaveOnlyLast10Entries(triggerMessage.RoundID)
-
+	n.mu.Lock()
+	defer n.mu.Unlock()
 	if msg.SentFrom != n.Raft.GetHostId() {
 		// follower can be changed into leader unexpectedly before recieving the message
 		// increase round id before checking the message sent from leader
 		// so that the next round will be triggered with larger round id
 		// prevents already handled message error
-
-		n.mu.Lock()
 		n.RoundID = max(triggerMessage.RoundID, n.RoundID)
-		n.mu.Unlock()
 	}
 
 	if triggerMessage.RoundID == 0 {
@@ -142,6 +152,7 @@ func (n *Aggregator) HandleTriggerMessage(ctx context.Context, msg raft.Message)
 		// if not enough messages collected from HandleSyncReplyMessage, it will hang in certain round
 		value = -1
 	} else {
+		n.roundLocalAggregate[triggerMessage.RoundID] = localAggregate.Value
 		value = localAggregate.Value
 	}
 
@@ -282,11 +293,33 @@ func (n *Aggregator) HandleProofMessage(ctx context.Context, msg raft.Message) e
 		concatProof := bytes.Join(n.roundProofs.proofs[proofMessage.RoundID], nil)
 		proof := Proof{ConfigID: n.ID, Round: proofMessage.RoundID, Proof: concatProof}
 
+		n.mu.Lock()
+		defer n.mu.Unlock()
+		if localAggregate, ok := n.roundLocalAggregate[proofMessage.RoundID]; ok {
+			if math.Abs(float64(localAggregate-globalAggregate.Value))/float64(globalAggregate.Value) > 0.3 {
+				log.Warn().Str("Player", "Aggregator").Str("Name", n.Name).Int32("roundId", proofMessage.RoundID).Int64("localAggregate", localAggregate).Int64("globalAggregate", globalAggregate.Value).Msg("local aggregate and global aggregate mismatch")
+				msg := bus.Message{
+					From: bus.AGGREGATOR,
+					To:   bus.FETCHER,
+					Content: bus.MessageContent{
+						Command: bus.REFRESH_FETCHER_APP,
+						Args:    nil,
+					},
+				}
+				err = n.bus.Publish(msg)
+				if err != nil {
+					log.Warn().Str("Player", "Aggregator").Err(err).Msg("failed to publish fetcher refresh bus message")
+				}
+
+				return errorSentinel.ErrAggregatorGlobalLocalPriceMismatch
+			}
+		}
+
 		err := PublishGlobalAggregateAndProof(ctx, n.Name, globalAggregate, proof)
 		if err != nil {
 			log.Error().Str("Player", "Aggregator").Err(err).Msg("failed to publish global aggregate and proof")
 		}
-
+		delete(n.roundLocalAggregate, proofMessage.RoundID)
 	}
 	return nil
 }

node/pkg/aggregator/aggregator_test.go

@@ -23,14 +23,14 @@ func TestNewAggregator(t *testing.T) {
 		}
 	}()
 
-	_, err = NewAggregator(testItems.app.Host, testItems.app.Pubsub, testItems.topicString, testItems.tmpData.config, testItems.signer, testItems.latestLocalAggMap)
+	_, err = NewAggregator(testItems.app.Host, testItems.app.Pubsub, testItems.topicString, testItems.tmpData.config, testItems.signer, testItems.latestLocalAggMap, testItems.app.Bus)
 	if err != nil {
 		t.Fatal("error creating new node")
 	}
 }
 
 func TestNewAggregator_Error(t *testing.T) {
-	_, err := NewAggregator(nil, nil, "", Config{}, nil, nil)
+	_, err := NewAggregator(nil, nil, "", Config{}, nil, nil, nil)
 	assert.NotNil(t, err, "Expected error when creating new aggregator with nil parameters")
 }
 
@@ -46,7 +46,7 @@ func TestLeaderJob(t *testing.T) {
 		}
 	}()
 
-	node, err := NewAggregator(testItems.app.Host, testItems.app.Pubsub, testItems.topicString, testItems.tmpData.config, testItems.signer, testItems.latestLocalAggMap)
+	node, err := NewAggregator(testItems.app.Host, testItems.app.Pubsub, testItems.topicString, testItems.tmpData.config, testItems.signer, testItems.latestLocalAggMap, testItems.app.Bus)
 	if err != nil {
 		t.Fatal("error creating new node")
 	}
@@ -70,7 +70,7 @@ func TestGetLatestRoundId(t *testing.T) {
 		}
 	}()
 
-	node, err := NewAggregator(testItems.app.Host, testItems.app.Pubsub, testItems.topicString, testItems.tmpData.config, testItems.signer, testItems.latestLocalAggMap)
+	node, err := NewAggregator(testItems.app.Host, testItems.app.Pubsub, testItems.topicString, testItems.tmpData.config, testItems.signer, testItems.latestLocalAggMap, testItems.app.Bus)
 	if err != nil {
 		t.Fatal("error creating new node")
 	}
@@ -98,7 +98,7 @@ func TestPublishGlobalAggregateAndProof(t *testing.T) {
 		}
 	}()
 
-	node, err := NewAggregator(testItems.app.Host, testItems.app.Pubsub, testItems.topicString, testItems.tmpData.config, testItems.signer, testItems.latestLocalAggMap)
+	node, err := NewAggregator(testItems.app.Host, testItems.app.Pubsub, testItems.topicString, testItems.tmpData.config, testItems.signer, testItems.latestLocalAggMap, testItems.app.Bus)
 	if err != nil {
 		t.Fatal("error creating new node")
 	}

node/pkg/aggregator/app.go

@@ -127,7 +127,7 @@ func (a *App) initializeLoadedAggregators(ctx context.Context, loadedConfigs []C
 		}
 
 		topicString := config.Name + "-global-aggregator-topic-" + strconv.Itoa(int(config.AggregateInterval))
-		tmpNode, err := NewAggregator(h, ps, topicString, config, signer, a.LatestLocalAggregates)
+		tmpNode, err := NewAggregator(h, ps, topicString, config, signer, a.LatestLocalAggregates, a.Bus)
 		if err != nil {
 			return err
 		}

node/pkg/aggregator/globalaggregatebulkwriter_test.go

@@ -86,7 +86,7 @@ func TestGlobalAggregateBulkWriterDataStore(t *testing.T) {
 	defer bulkWriter.Stop()
 	assert.NotEqual(t, nil, bulkWriter.ctx)
 
-	node, err := NewAggregator(testItems.app.Host, testItems.app.Pubsub, testItems.topicString, testItems.tmpData.config, testItems.signer, testItems.latestLocalAggMap)
+	node, err := NewAggregator(testItems.app.Host, testItems.app.Pubsub, testItems.topicString, testItems.tmpData.config, testItems.signer, testItems.latestLocalAggMap, testItems.app.Bus)
 	if err != nil {
 		t.Fatal("error creating new node")
 	}

node/pkg/aggregator/types.go

@@ -186,10 +186,12 @@ type Aggregator struct {
 	Raft *raft.Raft
 
 	LatestLocalAggregates *LatestLocalAggregates
-	roundTriggers         *RoundTriggers
-	roundPrices           *RoundPrices
-	roundPriceFixes       *RoundPriceFixes
-	roundProofs           *RoundProofs
+	roundLocalAggregate   map[int32]int64
+
+	roundTriggers   *RoundTriggers
+	roundPrices     *RoundPrices
+	roundPriceFixes *RoundPriceFixes
+	roundProofs     *RoundProofs
 
 	RoundID int32
 	Signer  *helper.Signer
@@ -198,7 +200,8 @@ type Aggregator struct {
 	nodeCancel context.CancelFunc
 	isRunning  bool
 
-	mu sync.RWMutex
+	mu  sync.RWMutex
+	bus *bus.MessageBus
 }
 
 type PriceDataMessage struct {

node/pkg/error/sentinel.go

@@ -92,6 +92,7 @@ var (
 	ErrAggregatorNotFound                 = &CustomError{Service: Aggregator, Code: InternalError, Message: "Aggregator not found"}
 	ErrAggregatorCancelNotFound           = &CustomError{Service: Aggregator, Code: InternalError, Message: "Aggregator cancel function not found"}
 	ErrAggregatorEmptyProof               = &CustomError{Service: Aggregator, Code: InternalError, Message: "Empty proof"}
+	ErrAggregatorGlobalLocalPriceMismatch = &CustomError{Service: Aggregator, Code: InternalError, Message: "Global and local price mismatch"}
 
 	ErrBootAPIDbPoolNotFound = &CustomError{Service: BootAPI, Code: InternalError, Message: "db pool not found"}
 
@@ -156,6 +157,7 @@ var (
 	ErrFetcherFailedToGetDexResultSlice       = &CustomError{Service: Fetcher, Code: InternalError, Message: "Failed to get dex result slice"}
 	ErrFetcherFailedBigIntConvert             = &CustomError{Service: Fetcher, Code: InternalError, Message: "Failed to convert to fetched data to big.Int"}
 	ErrFetcherFeedNotFound                    = &CustomError{Service: Fetcher, Code: InvalidInputError, Message: "Feed not found"}
+	ErrFetcherRefreshCooldown                 = &CustomError{Service: Fetcher, Code: InternalError, Message: "Failed to refresh data: cooldown period not over"}
 
 	ErrLibP2pEmptyNonLocalAddress = &CustomError{Service: Others, Code: InternalError, Message: "Host has no non-local addresses"}
 	ErrLibP2pAddressSplitFail     = &CustomError{Service: Others, Code: InternalError, Message: "Failed to split address"}

node/pkg/fetcher/app.go

@@ -25,6 +25,7 @@ func New(bus *bus.MessageBus) *App {
 		},
 		FeedDataDumpChannel: make(chan *FeedData, DefaultFeedDataDumpChannelSize),
 		Bus:                 bus,
+		LastRefreshTime:     time.Time{},
 	}
 }
 
@@ -63,7 +64,7 @@ func (a *App) subscribe(ctx context.Context) {
 }
 
 func (a *App) handleMessage(ctx context.Context, msg bus.Message) {
-	if msg.From != bus.ADMIN {
+	if msg.From != bus.ADMIN && msg.From != bus.AGGREGATOR {
 		log.Debug().Str("Player", "Fetcher").Msg("fetcher received message from non-admin")
 		return
 	}
@@ -93,6 +94,11 @@ func (a *App) handleMessage(ctx context.Context, msg bus.Message) {
 		}
 		msg.Response <- bus.MessageResponse{Success: true}
 	case bus.REFRESH_FETCHER_APP:
+		if !a.LastRefreshTime.IsZero() && time.Since(a.LastRefreshTime) < RefreshCooldownInterval {
+			bus.HandleMessageError(errorSentinel.ErrFetcherRefreshCooldown, msg, "refresh on cooldown")
+			return
+		}
+		a.LastRefreshTime = time.Now()
 		err := a.stopAll(ctx)
 		if err != nil {
 			log.Error().Err(err).Str("Player", "Fetcher").Msg("failed to stop all fetchers")

node/pkg/fetcher/app_test.go

@@ -6,6 +6,7 @@ import (
 	"testing"
 	"time"
 
+	"bisonai.com/miko/node/pkg/bus"
 	"bisonai.com/miko/node/pkg/db"
 	"github.com/stretchr/testify/assert"
 )
@@ -113,3 +114,128 @@ func TestAppRun(t *testing.T) {
 	}
 	assert.Greater(t, len(localAggregateResult), 0)
 }
+
+func TestAppRefresh(t *testing.T) {
+	ctx := context.Background()
+	clean, testItems, err := setup(ctx)
+	if err != nil {
+		t.Fatalf("error setting up test: %v", err)
+	}
+	defer func() {
+		if cleanupErr := clean(); cleanupErr != nil {
+			t.Logf("Cleanup failed: %v", cleanupErr)
+		}
+	}()
+
+	app := testItems.app
+
+	err = app.initialize(ctx)
+	if err != nil {
+		t.Fatalf("error initializing fetcher: %v", err)
+	}
+
+	err = app.Run(ctx)
+	if err != nil {
+		t.Fatalf("error running fetcher: %v", err)
+	}
+	for _, fetcher := range app.Fetchers {
+		assert.True(t, fetcher.isRunning)
+	}
+	for _, localAggregator := range app.LocalAggregators {
+		assert.True(t, localAggregator.isRunning)
+	}
+	assert.True(t, app.FeedDataBulkWriter.isRunning)
+
+	time.Sleep(WAIT_SECONDS)
+
+	resp := make(chan bus.MessageResponse)
+	err = testItems.messageBus.Publish(
+		bus.Message{
+			From: bus.AGGREGATOR,
+			To:   bus.FETCHER,
+			Content: bus.MessageContent{
+				Command: bus.REFRESH_FETCHER_APP,
+				Args:    nil,
+			},
+			Response: resp,
+		},
+	)
+	if err != nil {
+		t.Fatalf("error publishing message: %v", err)
+	}
+
+	result := <-resp
+	assert.Equal(t, result.Success, true)
+}
+
+func TestAppRefreshCooldown(t *testing.T) {
+	ctx := context.Background()
+	clean, testItems, err := setup(ctx)
+	if err != nil {
+		t.Fatalf("error setting up test: %v", err)
+	}
+	defer func() {
+		if cleanupErr := clean(); cleanupErr != nil {
+			t.Logf("Cleanup failed: %v", cleanupErr)
+		}
+	}()
+
+	app := testItems.app
+
+	err = app.initialize(ctx)
+	if err != nil {
+		t.Fatalf("error initializing fetcher: %v", err)
+	}
+
+	err = app.Run(ctx)
+	if err != nil {
+		t.Fatalf("error running fetcher: %v", err)
+	}
+	for _, fetcher := range app.Fetchers {
+		assert.True(t, fetcher.isRunning)
+	}
+	for _, localAggregator := range app.LocalAggregators {
+		assert.True(t, localAggregator.isRunning)
+	}
+	assert.True(t, app.FeedDataBulkWriter.isRunning)
+
+	time.Sleep(WAIT_SECONDS)
+
+	resp := make(chan bus.MessageResponse)
+	err = testItems.messageBus.Publish(
+		bus.Message{
+			From: bus.AGGREGATOR,
+			To:   bus.FETCHER,
+			Content: bus.MessageContent{
+				Command: bus.REFRESH_FETCHER_APP,
+				Args:    nil,
+			},
+			Response: resp,
+		},
+	)
+	if err != nil {
+		t.Fatalf("error publishing message: %v", err)
+	}
+
+	result := <-resp
+	assert.Equal(t, result.Success, true)
+
+	resp2 := make(chan bus.MessageResponse)
+	err = testItems.messageBus.Publish(
+		bus.Message{
+			From: bus.AGGREGATOR,
+			To:   bus.FETCHER,
+			Content: bus.MessageContent{
+				Command: bus.REFRESH_FETCHER_APP,
+				Args:    nil,
+			},
+			Response: resp2,
+		},
+	)
+	if err != nil {
+		t.Fatalf("error publishing message: %v", err)
+	}
+
+	result2 := <-resp2
+	assert.False(t, result2.Success)
+}