add data delay logs to node and dal

node/pkg/aggregator/types.go

@@ -36,6 +36,7 @@ type GlobalAggregate types.GlobalAggregate
 type SubmissionData struct {
 	GlobalAggregate GlobalAggregate
 	Proof           Proof
+	PublishTime     time.Time
 }
 
 type App struct {

node/pkg/aggregator/utils.go

@@ -35,6 +35,13 @@ func PublishGlobalAggregateAndProof(ctx context.Context, globalAggregate GlobalA
 	data := SubmissionData{
 		GlobalAggregate: globalAggregate,
 		Proof:           proof,
+		PublishTime:     time.Now(),
+	}
+
+	diff := time.Since(globalAggregate.Timestamp)
+	if diff > 1*time.Second {
+		log.Info().Dur("duration", diff).Int32("config_id", globalAggregate.ConfigID).Int64("value", globalAggregate.Value).Msg("published global aggregate")
+
 	}
 
 	return db.Publish(ctx, keys.SubmissionDataStreamKey(globalAggregate.ConfigID), data)

node/pkg/dal/collector/collector.go

@@ -6,6 +6,7 @@ import (
 	"os"
 	"strconv"
 	"sync"
+	"time"
 
 	"bisonai.com/orakl/node/pkg/aggregator"
 	"bisonai.com/orakl/node/pkg/chain/websocketchainreader"
@@ -157,6 +158,13 @@ func (c *Collector) receiveEach(ctx context.Context, configId int32) error {
 }
 
 func (c *Collector) processIncomingData(ctx context.Context, data aggregator.SubmissionData) {
+	symbol := c.Symbols[data.GlobalAggregate.ConfigID]
+	diff := time.Since(data.GlobalAggregate.Timestamp).Seconds()
+	diffFromPublish := time.Since(data.PublishTime).Seconds()
+	if diffFromPublish >= 1 || diff >= 1 {
+		log.Warn().Msgf("dataDiff: %v, diffFromPublish: %v for symbol %s", diff, diffFromPublish, symbol)
+	}
+
 	result, err := c.IncomingDataToOutgoingData(ctx, data)
 	if err != nil {
 		log.Error().Err(err).Str("Player", "DalCollector").Msg("failed to convert incoming data to outgoing data")