Add RetrieveAllTips from the Tangle (#673)

*  Add RetrieveAllTips from the Tangle

* Fix: Fix 



Co-authored-by: jkrvivian <jkrvivian@gmail.com>

packages/binary/messagelayer/tangle/tangle.go

@@ -353,3 +353,23 @@ func (tangle *Tangle) SolidifierWorkerPoolStatus() (name string, load int) {
 func (tangle *Tangle) StoreMessageWorkerPoolStatus() (name string, load int) {
 	return "StoreMessage", tangle.storeMessageWorkerPool.RunningWorkers()
 }
+
+// RetrieveAllTips returns the tips (i.e., solid messages that are not part of the approvers list).
+// It iterates over the messageMetadataStorage, thus only use this method if necessary.
+// TODO: improve this function.
+func (tangle *Tangle) RetrieveAllTips() (tips []message.Id) {
+	tangle.messageMetadataStorage.ForEach(func(key []byte, cachedMessage objectstorage.CachedObject) bool {
+		cachedMessage.Consume(func(object objectstorage.StorableObject) {
+			messageMetadata := object.(*MessageMetadata)
+			if messageMetadata != nil && messageMetadata.IsSolid() {
+				cachedApprovers := tangle.Approvers(messageMetadata.messageId)
+				if len(cachedApprovers) == 0 {
+					tips = append(tips, messageMetadata.messageId)
+				}
+				cachedApprovers.Consume(func(approver *Approver) {})
+			}
+		})
+		return true
+	})
+	return tips
+}

packages/binary/messagelayer/test/retrievealltips_test.go

+package test
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/iotaledger/goshimmer/packages/binary/messagelayer/message"
+	"github.com/iotaledger/goshimmer/packages/binary/messagelayer/payload"
+	"github.com/iotaledger/goshimmer/packages/binary/messagelayer/tangle"
+	"github.com/iotaledger/hive.go/crypto/ed25519"
+	"github.com/iotaledger/hive.go/events"
+	"github.com/iotaledger/hive.go/kvstore/mapdb"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestRetrieveAllTips(t *testing.T) {
+	messageTangle := tangle.New(mapdb.NewMapDB())
+
+	messageA := newTestMessage("A", message.EmptyId, message.EmptyId)
+	messageB := newTestMessage("B", messageA.Id(), message.EmptyId)
+	messageC := newTestMessage("C", messageA.Id(), message.EmptyId)
+
+	var wg sync.WaitGroup
+
+	messageTangle.Events.MessageSolid.Attach(events.NewClosure(func(cachedMessage *message.CachedMessage, cachedMessageMetadata *tangle.CachedMessageMetadata) {
+		cachedMessage.Release()
+		cachedMessageMetadata.Release()
+		wg.Done()
+	}))
+
+	wg.Add(3)
+	messageTangle.AttachMessage(messageA)
+	messageTangle.AttachMessage(messageB)
+	messageTangle.AttachMessage(messageC)
+
+	wg.Wait()
+
+	allTips := messageTangle.RetrieveAllTips()
+
+	assert.Equal(t, 2, len(allTips))
+
+	messageTangle.Shutdown()
+}
+
+func newTestMessage(payloadString string, parent1, parent2 message.Id) *message.Message {
+	return message.New(parent1, parent2, time.Now(), ed25519.PublicKey{}, 0, payload.NewData([]byte(payloadString)), 0, ed25519.Signature{})
+}

packages/binary/messagelayer/tipselector/tipselector.go

@@ -14,14 +14,27 @@ type TipSelector struct {
 }
 
 // New creates a new tip-selector.
-func New() *TipSelector {
-	return &TipSelector{
+func New(tips ...message.Id) *TipSelector {
+	tipSelector := &TipSelector{
 		tips: datastructure.NewRandomMap(),
 		Events: Events{
 			TipAdded:   events.NewEvent(messageIdEvent),
 			TipRemoved: events.NewEvent(messageIdEvent),
 		},
 	}
+
+	if tips != nil {
+		tipSelector.Set(tips...)
+	}
+
+	return tipSelector
+}
+
+// Set adds the given messageIDs as tips.
+func (tipSelector *TipSelector) Set(tips ...message.Id) {
+	for _, messageID := range tips {
+		tipSelector.tips.Set(messageID, messageID)
+	}
 }
 
 // AddTip adds the given message as a tip.

plugins/syncbeacon/plugin.go

@@ -7,6 +7,7 @@ import (
 	"github.com/iotaledger/goshimmer/packages/shutdown"
 	"github.com/iotaledger/goshimmer/plugins/config"
 	"github.com/iotaledger/goshimmer/plugins/issuer"
+	"github.com/iotaledger/goshimmer/plugins/messagelayer"
 	"github.com/iotaledger/goshimmer/plugins/syncbeacon/payload"
 	"github.com/iotaledger/goshimmer/plugins/syncbeaconfollower"
 	"github.com/iotaledger/hive.go/daemon"
@@ -53,6 +54,9 @@ func configure(_ *node.Plugin) {
 	log.Infof("starting node as sync beacon")
 
 	if config.Node().GetBool(CfgSyncBeaconStartSynced) {
+		log.Infof("Retrieving all the tips")
+		messagelayer.TipSelector().Set(messagelayer.Tangle().RetrieveAllTips()...)
+
 		syncbeaconfollower.OverwriteSyncedState(true)
 		log.Infof("overwriting synced state to 'true'")
 	}
@@ -60,6 +64,7 @@ func configure(_ *node.Plugin) {
 
 // broadcastSyncBeaconPayload broadcasts a sync beacon via communication layer.
 func broadcastSyncBeaconPayload() {
+
 	syncBeaconPayload := payload.NewSyncBeaconPayload(time.Now().UnixNano())
 	msg, err := issuer.IssuePayload(syncBeaconPayload)
 
@@ -73,6 +78,10 @@ func broadcastSyncBeaconPayload() {
 
 func run(_ *node.Plugin) {
 	if err := daemon.BackgroundWorker("Sync-Beacon", func(shutdownSignal <-chan struct{}) {
+		// wait CfgSyncBeaconBroadcastIntervalSec to possibly retrieve new beacons
+		if config.Node().GetBool(CfgSyncBeaconStartSynced) {
+			time.Sleep(config.Node().GetDuration(CfgSyncBeaconBroadcastIntervalSec) * time.Second)
+		}
 		ticker := time.NewTicker(config.Node().GetDuration(CfgSyncBeaconBroadcastIntervalSec) * time.Second)
 		defer ticker.Stop()
 		for {