test

test2

test

test

test

test

test

test

test

test

test

test

Author: richard-ramos

cmd/storemsgcounter/execute.go

@@ -150,15 +150,15 @@ func Execute(ctx context.Context, options Options) error {
 	missingMessagesTimer := time.NewTimer(0)
 	defer missingMessagesTimer.Stop()
 
-	syncCheckTimer := time.NewTicker(30 * time.Minute)
+	syncCheckTimer := time.NewTicker(1 * time.Minute)
 	defer syncCheckTimer.Stop()
 
 	for {
 		select {
 		case <-ctx.Done():
 			return nil
 		case <-missingMessagesTimer.C:
-			tmpUUID := uuid.New()
+			/*tmpUUID := uuid.New()
 			runId := hex.EncodeToString(tmpUUID[:])
 			runIdLogger := logger.With(zap.String("runId", runId))
 
@@ -169,21 +169,21 @@ func Execute(ctx context.Context, options Options) error {
 			}
 			runIdLogger.Info("verification complete")
 
-			missingMessagesTimer.Reset(timeInterval)
+			missingMessagesTimer.Reset(timeInterval)*/
 		case <-syncCheckTimer.C:
 			go func() {
 				tmpUUID := uuid.New()
 				runId := hex.EncodeToString(tmpUUID[:])
 				runIdLogger := logger.With(zap.String("syncRunId", runId))
-				runIdLogger.Info("rechecking missing messages status")
+				runIdLogger.Info("rechecking missing messages status", zap.Stringers("X", storenodeIDs))
 
-				err := application.checkMissingMessageStatus(ctx, runId, runIdLogger)
+				err := application.checkMissingMessageStatus(ctx, storenodeIDs, runId, runIdLogger)
 				if err != nil {
 					logger.Error("could not recheck the status of missing messages", zap.Error(err))
 					return
 				}
 
-				err = application.countMissingMessages()
+				err = application.countMissingMessages(storenodeIDs)
 				if err != nil {
 					logger.Error("could not count missing messages", zap.Error(err))
 					return
@@ -338,29 +338,43 @@ func (app *Application) verifyHistory(ctx context.Context, runId string, storeno
 	return nil
 }
 
-func (app *Application) checkMissingMessageStatus(ctx context.Context, runId string, logger *zap.Logger) error {
+func (app *Application) checkMissingMessageStatus(ctx context.Context, storenodes []peer.ID, runId string, logger *zap.Logger) error {
 	now := app.node.Timesource().Now()
 
 	// Get all messages whose status is missing or does not exist, and the column found_on_recheck is false
 	// if found, set found_on_recheck to true
 	missingMessages, err := app.db.GetMissingMessages(now.Add(-2*time.Hour), now.Add(-time.Hour), options.ClusterID)
+	if err == nil {
+		for k, t := range missingMessages {
+			logger.Info("TTTTTX", zap.String("perID", k.String()), zap.Any("missingMessages", len(t)))
+
+		}
+	}
 	if err != nil {
+		logger.Info("TTT2 error?", zap.Error(err))
 		return err
 	}
 
 	wg := sync.WaitGroup{}
-	for storenodeID, messageHashes := range missingMessages {
+
+	logger.Info("VVVVV1", zap.Any("V1", storenodes))
+	for _, storenodeID := range storenodes {
+		logger.Info("VVVVV2")
+
 		wg.Add(1)
 		go func(peerID peer.ID, messageHashes []pb.MessageHash) {
 			defer wg.Done()
 
 			foundMissingMessages := make(map[pb.MessageHash]struct{})
+			logger.Info("CCCCC Verifying message existence")
 			app.verifyMessageExistence(ctx, runId, peerID, messageHashes, func(result *store.Result) {
 				for _, mkv := range result.Messages() {
 					foundMissingMessages[mkv.WakuMessageHash()] = struct{}{}
 				}
 			}, logger)
 
+			logger.Info("DDD Marking as found", zap.Int("X", len(foundMissingMessages)))
+
 			err := app.db.MarkMessagesAsFound(peerID, maps.Keys(foundMissingMessages), options.ClusterID)
 			if err != nil {
 				logger.Error("could not mark messages as found", zap.Error(err))
@@ -369,15 +383,19 @@ func (app *Application) checkMissingMessageStatus(ctx context.Context, runId str
 
 			app.metrics.RecordMissingMessagesPrevHour(peerID, len(messageHashes)-len(foundMissingMessages))
 
-		}(storenodeID, messageHashes)
+		}(storenodeID, missingMessages[storenodeID])
+		logger.Info("VVVVV4")
+
 	}
+	logger.Info("VVVVV5")
 
 	wg.Wait()
+	logger.Info("VVVVV6")
 
 	return nil
 }
 
-func (app *Application) countMissingMessages() error {
+func (app *Application) countMissingMessages(storenodes []peer.ID) error {
 
 	// not including last two hours in now to let sync work
 	now := app.node.Timesource().Now().Add(-2 * time.Hour)
@@ -396,8 +414,8 @@ func (app *Application) countMissingMessages() error {
 	if err != nil {
 		return err
 	}
-	for storenode, cnt := range results {
-		app.metrics.RecordMissingMessagesLastWeek(storenode, cnt)
+	for _, storenodeID := range storenodes {
+		app.metrics.RecordMissingMessagesLastWeek(storenodeID, results[storenodeID])
 	}
 	return nil
 }
@@ -539,6 +557,10 @@ func (app *Application) verifyMessageExistence(ctx context.Context, runId string
 	var result *store.Result
 	var err error
 
+	if len(messageHashes) == 0 {
+		return
+	}
+
 	peerInfo := app.node.Host().Peerstore().PeerInfo(peerID)
 
 	queryLogger := logger.With(zap.Stringer("storenode", peerID))

internal/persistence/database.go

@@ -240,6 +240,7 @@ func (d *DBStore) GetTopicSyncStatus(ctx context.Context, clusterID uint, pubsub
 }
 
 func (d *DBStore) GetMissingMessages(from time.Time, to time.Time, clusterID uint) (map[peer.ID][]pb.MessageHash, error) {
+	// And not found on rehcekc
 	rows, err := d.db.Query("SELECT messageHash, storenode FROM missingMessages WHERE storedAt >= $1 AND storedAt <= $2 AND clusterId = $3 AND msgStatus = 'does_not_exist'", from.UnixNano(), to.UnixNano(), clusterID)
 	if err != nil {
 		return nil, err
@@ -307,6 +308,10 @@ func (d *DBStore) RecordMessage(uuid string, tx *sql.Tx, msgHash pb.MessageHash,
 }
 
 func (d *DBStore) MarkMessagesAsFound(peerID peer.ID, messageHashes []pb.MessageHash, clusterID uint) error {
+	if len(messageHashes) == 0 {
+		return nil
+	}
+
 	query := "UPDATE missingMessages SET foundOnRecheck = true WHERE clusterID = $1 AND messageHash IN ("
 	for i := range messageHashes {
 		if i > 0 {
@@ -315,7 +320,7 @@ func (d *DBStore) MarkMessagesAsFound(peerID peer.ID, messageHashes []pb.Message
 		query += fmt.Sprintf("$%d", i+2)
 	}
 	query += ")"
-
+	d.log.Info("THE QUERY: ", zap.String("Q", query))
 	args := []interface{}{clusterID}
 	for _, messageHash := range messageHashes {
 		args = append(args, messageHash)
@@ -343,7 +348,7 @@ func (d *DBStore) RecordStorenodeUnavailable(uuid string, storenode peer.ID) err
 }
 
 func (d *DBStore) CountMissingMessages(from time.Time, to time.Time, clusterID uint) (map[peer.ID]int, error) {
-	rows, err := d.db.Query("SELECT storenode, count(1) as cnt FROM missingMessages WHERE storedAt >= $1 AND storedAt <= $2 AND clusterId = $3 AND msgStatus = 'does_not_exist' GROUP BY storenode", from.UnixNano(), to.UnixNano(), clusterID)
+	rows, err := d.db.Query("SELECT storenode, count(1) as cnt FROM missingMessages WHERE storedAt >= $1 AND storedAt <= $2 AND clusterId = $3 AND msgStatus = 'does_not_exist' AND foundOnRecheck = false GROUP BY storenode", from.UnixNano(), to.UnixNano(), clusterID)
 	if err != nil {
 		return nil, err
 	}