Feat: Refactor Preferred / Liked logic (#428)

* Feat: initial commit

* Feat: added setPreferred to TransactionMetadata

* Feat: added a Conflicting() method to the transactionMetadata

* Fix: fixed logic bug

* Feat: refactored fcob

* Refactor: refactored additional code

* Fix: fixed a bug in ForeachConsumers

* Refactor: cleaned up code

* Feat: implemented FCOB consensus into the valuetransfer dapp

* Refactor: refactored FCOB

* Docs: added some additional comments

* Docs: fixed comments

* Refactor: commit before branch change

* Feat: added PayloadLiked Event

* Refactor: fixed some missing comments + added liked to marshal

* Feat: reworked the preferred and liked propagation

* Refactor: cleaned up some logic

* Refactor: simplified code

* Refactor: cleaned up more stuff :P

* Refactor: refactor

* Feat: moved test + refactored fcob

* Fix: fixed a few bugs in liked propagation

* adapt to new hive.go version

* upgrade hive.go

* Feat: started implementing a wallet

* Feat: extended wallet files

* use store backed sequence

* add option to use in-memory database

* address review comments

* Fix: fixed missing events in branchmanaer

* Feat: propagate changes from branch to transaction

* Feat: started implementing confirmed propagation

* Fix: fixed unreachable code

* Refactor: refactored some code according to wolfgangs review

* Refactor: cleaned up the code according to DRY

* Refactor: refactored according to wollac

* Refactor: refactored according to wollac

* Refactor: refactored according to wollac

* Refactor: refactored the code to make it more readable

* Refactor: added some doc comments + cleaned up some more code

Co-authored-by: Wolfgang Welz <welzwo@gmail.com>

dapps/valuetransfers/dapp.go

@@ -9,10 +9,9 @@ import (
 	"github.com/iotaledger/hive.go/logger"
 	"github.com/iotaledger/hive.go/node"
 
-	"github.com/iotaledger/goshimmer/dapps/valuetransfers/packages/branchmanager"
+	"github.com/iotaledger/goshimmer/dapps/valuetransfers/packages/consensus"
 	valuepayload "github.com/iotaledger/goshimmer/dapps/valuetransfers/packages/payload"
 	"github.com/iotaledger/goshimmer/dapps/valuetransfers/packages/tangle"
-	"github.com/iotaledger/goshimmer/dapps/valuetransfers/packages/transaction"
 	"github.com/iotaledger/goshimmer/packages/binary/messagelayer/message"
 	messageTangle "github.com/iotaledger/goshimmer/packages/binary/messagelayer/tangle"
 	"github.com/iotaledger/goshimmer/packages/shutdown"
@@ -25,7 +24,7 @@ const (
 	PluginName = "ValueTransfers"
 
 	// AverageNetworkDelay contains the average time it takes for a network to propagate through gossip.
-	AverageNetworkDelay = 6 * time.Second
+	AverageNetworkDelay = 5 * time.Second
 )
 
 var (
@@ -35,6 +34,9 @@ var (
 	// Tangle represents the value tangle that is used to express votes on value transactions.
 	Tangle *tangle.Tangle
 
+	// FCOB contains the fcob consensus logic.
+	FCOB *consensus.FCOB
+
 	// LedgerState represents the ledger state, that keeps track of the liked branches and offers an API to access funds.
 	LedgerState *tangle.LedgerState
 
@@ -43,44 +45,35 @@ var (
 )
 
 func configure(_ *node.Plugin) {
+	// configure logger
 	log = logger.NewLogger(PluginName)
 
-	log.Debug("configuring ValueTransfers")
-
-	// create instances
+	// configure Tangle
 	Tangle = tangle.New(database.Store())
+	Tangle.Events.Error.Attach(events.NewClosure(func(err error) {
+		log.Error(err)
+	}))
 
-	// subscribe to message-layer
-	messagelayer.Tangle.Events.MessageSolid.Attach(events.NewClosure(onReceiveMessageFromMessageLayer))
-
-	// setup behavior of package instances
-	Tangle.Events.TransactionBooked.Attach(events.NewClosure(onTransactionBooked))
-	Tangle.Events.Fork.Attach(events.NewClosure(onForkOfFirstConsumer))
-
-	configureFPC()
-	// TODO: DECIDE WHAT WE SHOULD DO IF FPC FAILS -> cry
-	// voter.Events().Failed.Attach(events.NewClosure(panic))
-	voter.Events().Finalized.Attach(events.NewClosure(func(id string, opinion vote.Opinion) {
-		branchID, err := branchmanager.BranchIDFromBase58(id)
-		if err != nil {
+	// configure FCOB consensus rules
+	FCOB = consensus.NewFCOB(Tangle, AverageNetworkDelay)
+	FCOB.Events.Vote.Attach(events.NewClosure(func(id string, initOpn vote.Opinion) {
+		if err := voter.Vote(id, initOpn); err != nil {
 			log.Error(err)
-
-			return
 		}
+	}))
+	FCOB.Events.Error.Attach(events.NewClosure(func(err error) {
+		log.Error(err)
+	}))
 
-		switch opinion {
-		case vote.Like:
-			if _, err := Tangle.BranchManager().SetBranchPreferred(branchID, true); err != nil {
-				panic(err)
-			}
-			// TODO: merge branch mutations into the parent branch
-		case vote.Dislike:
-			if _, err := Tangle.BranchManager().SetBranchPreferred(branchID, false); err != nil {
-				panic(err)
-			}
-			// TODO: merge branch mutations into the parent branch / cleanup
-		}
+	// configure FPC + link to consensus
+	configureFPC()
+	voter.Events().Finalized.Attach(events.NewClosure(FCOB.ProcessVoteResult))
+	voter.Events().Failed.Attach(events.NewClosure(func(id string, lastOpinion vote.Opinion) {
+		log.Errorf("FPC failed for transaction with id '%s' - last opinion: '%s'", id, lastOpinion)
 	}))
+
+	// subscribe to message-layer
+	messagelayer.Tangle.Events.MessageSolid.Attach(events.NewClosure(onReceiveMessageFromMessageLayer))
 }
 
 func run(*node.Plugin) {
@@ -98,103 +91,22 @@ func onReceiveMessageFromMessageLayer(cachedMessage *message.CachedMessage, cach
 
 	solidMessage := cachedMessage.Unwrap()
 	if solidMessage == nil {
-		// TODO: LOG ERROR?
+		log.Debug("failed to unpack solid message from message layer")
 
 		return
 	}
 
 	messagePayload := solidMessage.Payload()
 	if messagePayload.Type() != valuepayload.Type {
-		// TODO: LOG ERROR?
-
 		return
 	}
 
 	valuePayload, ok := messagePayload.(*valuepayload.Payload)
 	if !ok {
-		// TODO: LOG ERROR?
+		log.Debug("could not cast payload to value payload")
 
 		return
 	}
 
 	Tangle.AttachPayload(valuePayload)
 }
-
-func onTransactionBooked(cachedTransaction *transaction.CachedTransaction, cachedTransactionMetadata *tangle.CachedTransactionMetadata, cachedBranch *branchmanager.CachedBranch, conflictingInputs []transaction.OutputID, decisionPending bool) {
-	defer cachedTransaction.Release()
-	defer cachedTransactionMetadata.Release()
-	defer cachedBranch.Release()
-
-	if len(conflictingInputs) >= 1 {
-		// abort if the previous consumers where finalized already
-		if !decisionPending {
-			return
-		}
-
-		branch := cachedBranch.Unwrap()
-		if branch == nil {
-			log.Error("failed to unpack branch")
-
-			return
-		}
-
-		err := voter.Vote(branch.ID().String(), vote.Dislike)
-		if err != nil {
-			log.Error(err)
-		}
-
-		return
-	}
-
-	// If the transaction is not conflicting, then we apply the fcob rule (we finalize after 2 network delays).
-	// Note: We do not set a liked flag after 1 network delay because that can be derived by the retriever later.
-	cachedTransactionMetadata.Retain()
-	time.AfterFunc(2*AverageNetworkDelay, func() {
-		defer cachedTransactionMetadata.Release()
-
-		transactionMetadata := cachedTransactionMetadata.Unwrap()
-		if transactionMetadata == nil {
-			return
-		}
-
-		// TODO: check that the booking goroutine in the UTXO DAG and this check is somehow synchronized
-		if transactionMetadata.BranchID() == branchmanager.NewBranchID(transactionMetadata.ID()) {
-			return
-		}
-
-		transactionMetadata.SetFinalized(true)
-	})
-}
-
-// TODO: clarify what we do here
-func onForkOfFirstConsumer(cachedTransaction *transaction.CachedTransaction, cachedTransactionMetadata *tangle.CachedTransactionMetadata, cachedBranch *branchmanager.CachedBranch, conflictingInputs []transaction.OutputID) {
-	defer cachedTransaction.Release()
-	defer cachedTransactionMetadata.Release()
-	defer cachedBranch.Release()
-
-	transactionMetadata := cachedTransactionMetadata.Unwrap()
-	if transactionMetadata == nil {
-		return
-	}
-
-	branch := cachedBranch.Unwrap()
-	if branch == nil {
-		return
-	}
-
-	if time.Since(transactionMetadata.SoldificationTime()) < AverageNetworkDelay {
-		if err := voter.Vote(branch.ID().String(), vote.Dislike); err != nil {
-			log.Error(err)
-		}
-
-		return
-	}
-
-	if _, err := Tangle.BranchManager().SetBranchPreferred(branch.ID(), true); err != nil {
-		log.Error(err)
-	}
-
-	if err := voter.Vote(branch.ID().String(), vote.Like); err != nil {
-		log.Error(err)
-	}
-}

dapps/valuetransfers/packages/branchmanager/branch.go

@@ -21,11 +21,15 @@ type Branch struct {
 	conflicts      map[ConflictID]types.Empty
 	preferred      bool
 	liked          bool
+	finalized      bool
+	confirmed      bool
 
 	parentBranchesMutex sync.RWMutex
 	conflictsMutex      sync.RWMutex
 	preferredMutex      sync.RWMutex
 	likedMutex          sync.RWMutex
+	finalizedMutex      sync.RWMutex
+	confirmedMutex      sync.RWMutex
 }
 
 // NewBranch is the constructor of a Branch and creates a new Branch object from the given details.
@@ -211,7 +215,7 @@ func (branch *Branch) setPreferred(preferred bool) (modified bool) {
 	branch.SetModified()
 	modified = true
 
-	return branch.preferred
+	return
 }
 
 // Liked returns if the branch is liked (it is preferred and all of its parents are liked).
@@ -246,6 +250,76 @@ func (branch *Branch) setLiked(liked bool) (modified bool) {
 	return branch.liked
 }
 
+// Finalized returns true if the branch has been marked as finalized.
+func (branch *Branch) Finalized() bool {
+	branch.finalizedMutex.RLock()
+	defer branch.finalizedMutex.RUnlock()
+
+	return branch.finalized
+}
+
+// setFinalized is the setter for the finalized flag. It returns true if the value of the flag has been updated.
+// A branch is finalized if a decisions regarding its preference has been made.
+// Note: Just because a branch has been finalized, does not mean that all transactions it contains have also been
+//       finalized but only that the underlying conflict that created the Branch has been finalized.
+func (branch *Branch) setFinalized(finalized bool) (modified bool) {
+	branch.finalizedMutex.RLock()
+	if branch.finalized == finalized {
+		branch.finalizedMutex.RUnlock()
+
+		return
+	}
+
+	branch.finalizedMutex.RUnlock()
+	branch.finalizedMutex.Lock()
+	defer branch.finalizedMutex.Unlock()
+
+	if branch.finalized == finalized {
+		return
+	}
+
+	branch.finalized = finalized
+	branch.SetModified()
+	modified = true
+
+	return
+}
+
+// Confirmed returns true if the branch has been accepted to be part of the ledger state.
+func (branch *Branch) Confirmed() bool {
+	branch.confirmedMutex.RLock()
+	defer branch.confirmedMutex.RUnlock()
+
+	return branch.confirmed
+}
+
+// setConfirmed is the setter for the confirmed flag. It returns true if the value of the flag has been updated.
+// A branch is confirmed if it is considered to have been accepted to be part of the ledger state.
+// Note: Just because a branch has been confirmed, does not mean that all transactions it contains have also been
+//       confirmed but only that the underlying conflict that created the Branch has been decided.
+func (branch *Branch) setConfirmed(confirmed bool) (modified bool) {
+	branch.confirmedMutex.RLock()
+	if branch.confirmed == confirmed {
+		branch.confirmedMutex.RUnlock()
+
+		return
+	}
+
+	branch.confirmedMutex.RUnlock()
+	branch.confirmedMutex.Lock()
+	defer branch.confirmedMutex.Unlock()
+
+	if branch.confirmed == confirmed {
+		return
+	}
+
+	branch.confirmed = confirmed
+	branch.SetModified()
+	modified = true
+
+	return
+}
+
 // Bytes returns a marshaled version of this Branch.
 func (branch *Branch) Bytes() []byte {
 	return marshalutil.New().
@@ -282,9 +356,10 @@ func (branch *Branch) ObjectStorageValue() []byte {
 	parentBranches := branch.ParentBranches()
 	parentBranchCount := len(parentBranches)
 
-	marshalUtil := marshalutil.New(2*marshalutil.BOOL_SIZE + marshalutil.UINT32_SIZE + parentBranchCount*BranchIDLength)
-	marshalUtil.WriteBool(branch.preferred)
-	marshalUtil.WriteBool(branch.liked)
+	marshalUtil := marshalutil.New(3*marshalutil.BOOL_SIZE + marshalutil.UINT32_SIZE + parentBranchCount*BranchIDLength)
+	marshalUtil.WriteBool(branch.Preferred())
+	marshalUtil.WriteBool(branch.Liked())
+	marshalUtil.WriteBool(branch.Confirmed())
 	marshalUtil.WriteUint32(uint32(parentBranchCount))
 	for _, branchID := range parentBranches {
 		marshalUtil.WriteBytes(branchID.Bytes())
@@ -304,6 +379,10 @@ func (branch *Branch) UnmarshalObjectStorageValue(valueBytes []byte) (consumedBy
 	if err != nil {
 		return
 	}
+	branch.confirmed, err = marshalUtil.ReadBool()
+	if err != nil {
+		return
+	}
 	parentBranchCount, err := marshalUtil.ReadUint32()
 	if err != nil {
 		return

dapps/valuetransfers/packages/branchmanager/branchmanager.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"sort"
 
+	"github.com/iotaledger/hive.go/events"
 	"github.com/iotaledger/hive.go/kvstore"
 	"github.com/iotaledger/hive.go/marshalutil"
 	"github.com/iotaledger/hive.go/objectstorage"
@@ -47,6 +48,12 @@ func New(store kvstore.KVStore) (branchManager *BranchManager) {
 		childBranchStorage:    osFactory.New(osChildBranch, osChildBranchFactory, osChildBranchOptions...),
 		conflictStorage:       osFactory.New(osConflict, osConflictFactory, osConflictOptions...),
 		conflictMemberStorage: osFactory.New(osConflictMember, osConflictMemberFactory, osConflictMemberOptions...),
+		Events: &Events{
+			BranchPreferred:   events.NewEvent(branchCaller),
+			BranchUnpreferred: events.NewEvent(branchCaller),
+			BranchLiked:       events.NewEvent(branchCaller),
+			BranchDisliked:    events.NewEvent(branchCaller),
+		},
 	}
 	branchManager.init()
 
@@ -337,6 +344,10 @@ func (branchManager *BranchManager) SetBranchLiked(branchID BranchID, liked bool
 	return branchManager.setBranchLiked(branchManager.Branch(branchID), liked)
 }
 
+func (branchManager *BranchManager) SetBranchFinalized(branchID BranchID) (modified bool, err error) {
+	return
+}
+
 // Prune resets the database and deletes all objects (for testing or "node resets").
 func (branchManager *BranchManager) Prune() (err error) {
 	for _, storage := range []*objectstorage.ObjectStorage{
@@ -472,6 +483,23 @@ func (branchManager *BranchManager) setBranchLiked(cachedBranch *CachedBranch, l
 	return
 }
 
+// IsBranchLiked returns true if the Branch is currently marked as liked.
+func (branchManager *BranchManager) IsBranchLiked(id BranchID) (liked bool) {
+	if id == UndefinedBranchID {
+		return
+	}
+
+	if id == MasterBranchID {
+		return true
+	}
+
+	branchManager.Branch(id).Consume(func(branch *Branch) {
+		liked = branch.Liked()
+	})
+
+	return
+}
+
 func (branchManager *BranchManager) propagateLike(cachedBranch *CachedBranch) (err error) {
 	// unpack CachedBranch and abort of the branch doesn't exist or isn't preferred
 	defer cachedBranch.Release()

dapps/valuetransfers/packages/branchmanager/events.go

@@ -18,3 +18,7 @@ type Events struct {
 	// BranchLiked gets triggered whenever a Branch becomes preferred that was not preferred before.
 	BranchDisliked *events.Event
 }
+
+func branchCaller(handler interface{}, params ...interface{}) {
+	handler.(func(branch *CachedBranch))(params[0].(*CachedBranch).Retain())
+}

dapps/valuetransfers/packages/consensus/fcob.go

+package consensus
+
+import (
+	"time"
+
+	"github.com/iotaledger/hive.go/events"
+
+	"github.com/iotaledger/goshimmer/dapps/valuetransfers/packages/tangle"
+	"github.com/iotaledger/goshimmer/dapps/valuetransfers/packages/transaction"
+	"github.com/iotaledger/goshimmer/packages/vote"
+)
+
+// FCOB defines the "Fast Consensus of Barcelona" rules that are used to form the initial opinions of nodes. It uses a
+// local modifier based approach to reach approximate consensus within the network by waiting 1 network delay before
+// setting a transaction to preferred (if it didnt see a conflict) and another network delay to set it to finalized (if
+// it still didn't see a conflict).
+type FCOB struct {
+	Events *FCOBEvents
+
+	tangle              *tangle.Tangle
+	averageNetworkDelay time.Duration
+}
+
+// NewFCOB is the constructor for an FCOB consensus instance. It automatically attaches to the passed in Tangle and
+// calls the corresponding Events if it needs to trigger a vote.
+func NewFCOB(tangle *tangle.Tangle, averageNetworkDelay time.Duration) (fcob *FCOB) {
+	fcob = &FCOB{
+		tangle:              tangle,
+		averageNetworkDelay: averageNetworkDelay,
+		Events: &FCOBEvents{
+			Error: events.NewEvent(events.ErrorCaller),
+		},
+	}
+
+	// setup behavior of package instances
+	tangle.Events.TransactionBooked.Attach(events.NewClosure(fcob.onTransactionBooked))
+	tangle.Events.Fork.Attach(events.NewClosure(fcob.onFork))
+
+	return
+}
+
+// ProcessVoteResult allows an external voter to hand in the results of the voting process.
+func (fcob *FCOB) ProcessVoteResult(id string, opinion vote.Opinion) {
+	transactionID, err := transaction.IDFromBase58(id)
+	if err != nil {
+		fcob.Events.Error.Trigger(err)
+
+		return
+	}
+
+	if _, err := fcob.tangle.SetTransactionPreferred(transactionID, opinion == vote.Like); err != nil {
+		fcob.Events.Error.Trigger(err)
+	}
+}
+
+// onTransactionBooked analyzes the transaction that was booked by the Tangle and initiates the FCOB rules if it is not
+// conflicting. If it is conflicting and a decision is still pending we trigger a voting process.
+func (fcob *FCOB) onTransactionBooked(cachedTransaction *transaction.CachedTransaction, cachedTransactionMetadata *tangle.CachedTransactionMetadata, decisionPending bool) {
+	defer cachedTransaction.Release()
+
+	cachedTransactionMetadata.Consume(func(transactionMetadata *tangle.TransactionMetadata) {
+		if transactionMetadata.Conflicting() {
+			// abort if the previous consumers where finalized already
+			if !decisionPending {
+				return
+			}
+
+			fcob.Events.Vote.Trigger(transactionMetadata.BranchID().String(), vote.Dislike)
+
+			return
+		}
+
+		fcob.scheduleSetPreferred(cachedTransactionMetadata.Retain())
+	})
+}
+
+// scheduleSetPreferred schedules the setPreferred logic after 1 network delay.
+func (fcob *FCOB) scheduleSetPreferred(cachedTransactionMetadata *tangle.CachedTransactionMetadata) {
+	if fcob.averageNetworkDelay == 0 {
+		fcob.setPreferred(cachedTransactionMetadata)
+	} else {
+		time.AfterFunc(fcob.averageNetworkDelay, func() {
+			fcob.setPreferred(cachedTransactionMetadata)
+		})
+	}
+}
+
+// setPreferred sets the Transaction to preferred if it is not conflicting.
+func (fcob *FCOB) setPreferred(cachedTransactionMetadata *tangle.CachedTransactionMetadata) {
+	cachedTransactionMetadata.Consume(func(transactionMetadata *tangle.TransactionMetadata) {
+		if transactionMetadata.Conflicting() {
+			return
+		}
+
+		modified, err := fcob.tangle.SetTransactionPreferred(transactionMetadata.ID(), true)
+		if err != nil {
+			fcob.Events.Error.Trigger(err)
+
+			return
+		}
+
+		if modified {
+			fcob.scheduleSetFinalized(cachedTransactionMetadata.Retain())
+		}
+	})
+}
+
+// scheduleSetFinalized schedules the setFinalized logic after 2 network delays.
+// Note: it is 2 network delays because this function gets triggered at the end of the first delay that sets a
+// Transaction to preferred (see setPreferred).
+func (fcob *FCOB) scheduleSetFinalized(cachedTransactionMetadata *tangle.CachedTransactionMetadata) {
+	if fcob.averageNetworkDelay == 0 {
+		fcob.setFinalized(cachedTransactionMetadata)
+	} else {
+		time.AfterFunc(fcob.averageNetworkDelay, func() {
+			fcob.setFinalized(cachedTransactionMetadata)
+		})
+	}
+}
+
+// setFinalized sets the Transaction to finalized if it is not conflicting.
+func (fcob *FCOB) setFinalized(cachedTransactionMetadata *tangle.CachedTransactionMetadata) {
+	cachedTransactionMetadata.Consume(func(transactionMetadata *tangle.TransactionMetadata) {
+		if transactionMetadata.Conflicting() {
+			return
+		}
+
+		transactionMetadata.SetFinalized(true)
+	})
+}
+
+// onFork triggers a voting process whenever a Transaction gets forked into a new Branch. The initial opinion is derived
+// from the preferred flag that was set using the FCOB rule.
+func (fcob *FCOB) onFork(cachedTransaction *transaction.CachedTransaction, cachedTransactionMetadata *tangle.CachedTransactionMetadata) {
+	defer cachedTransaction.Release()
+	defer cachedTransactionMetadata.Release()
+
+	transactionMetadata := cachedTransactionMetadata.Unwrap()
+	if transactionMetadata == nil {
+		return
+	}
+
+	switch transactionMetadata.Preferred() {
+	case true:
+		fcob.Events.Vote.Trigger(transactionMetadata.ID().String(), vote.Like)
+	case false:
+		fcob.Events.Vote.Trigger(transactionMetadata.ID().String(), vote.Dislike)
+	}
+}
+
+// FCOBEvents acts as a dictionary for events of an FCOB instance.
+type FCOBEvents struct {
+	// Error gets called when FCOB faces an error.
+	Error *events.Event
+
+	// Vote gets called when FCOB needs to vote on a transaction.
+	Vote *events.Event
+}

dapps/valuetransfers/packages/tangle/events.go

@@ -13,6 +13,8 @@ type Events struct {
 	// Get's called whenever a transaction
 	PayloadAttached        *events.Event
 	PayloadSolid           *events.Event
+	PayloadLiked           *events.Event
+	PayloadDisliked        *events.Event
 	MissingPayloadReceived *events.Event
 	PayloadMissing         *events.Event
 	PayloadUnsolidifiable  *events.Event
@@ -34,6 +36,8 @@ func newEvents() *Events {
 	return &Events{
 		PayloadAttached:        events.NewEvent(cachedPayloadEvent),
 		PayloadSolid:           events.NewEvent(cachedPayloadEvent),
+		PayloadLiked:           events.NewEvent(cachedPayloadEvent),
+		PayloadDisliked:        events.NewEvent(cachedPayloadEvent),
 		MissingPayloadReceived: events.NewEvent(cachedPayloadEvent),
 		PayloadMissing:         events.NewEvent(payloadIDEvent),
 		PayloadUnsolidifiable:  events.NewEvent(payloadIDEvent),
@@ -56,12 +60,10 @@ func cachedPayloadEvent(handler interface{}, params ...interface{}) {
 }
 
 func transactionBookedEvent(handler interface{}, params ...interface{}) {
-	handler.(func(*transaction.CachedTransaction, *CachedTransactionMetadata, *branchmanager.CachedBranch, []transaction.OutputID, bool))(
+	handler.(func(*transaction.CachedTransaction, *CachedTransactionMetadata, bool))(
 		params[0].(*transaction.CachedTransaction).Retain(),
 		params[1].(*CachedTransactionMetadata).Retain(),
-		params[2].(*branchmanager.CachedBranch).Retain(),
-		params[3].([]transaction.OutputID),
-		params[4].(bool),
+		params[2].(bool),
 	)
 }
 

dapps/valuetransfers/packages/tangle/objectstorage.go

@@ -27,7 +27,7 @@ const (
 
 var (
 	osLeakDetectionOption = objectstorage.LeakDetectionEnabled(true, objectstorage.LeakDetectionOptions{
-		MaxConsumersPerObject: 10,
+		MaxConsumersPerObject: 20,
 		MaxConsumerHoldTime:   10 * time.Second,
 	})
 )

dapps/valuetransfers/packages/tangle/payloadmetadata.go

@@ -20,10 +20,12 @@ type PayloadMetadata struct {
 	payloadID          payload.ID
 	solid              bool
 	solidificationTime time.Time
+	liked              bool
 	branchID           branchmanager.BranchID
 
 	solidMutex              sync.RWMutex
 	solidificationTimeMutex sync.RWMutex
+	likedMutex              sync.RWMutex
 	branchIDMutex           sync.RWMutex
 }
 
@@ -139,6 +141,38 @@ func (payloadMetadata *PayloadMetadata) SoldificationTime() time.Time {
 	return payloadMetadata.solidificationTime
 }
 
+// Liked returns true if the Payload was marked as liked.
+func (payloadMetadata *PayloadMetadata) Liked() bool {
+	payloadMetadata.likedMutex.RLock()
+	defer payloadMetadata.likedMutex.RUnlock()
+
+	return payloadMetadata.liked
+}
+
+// SetLiked modifies the liked flag of the given Payload. It returns true if the value has been updated.
+func (payloadMetadata *PayloadMetadata) SetLiked(liked bool) (modified bool) {
+	payloadMetadata.likedMutex.RLock()
+	if payloadMetadata.liked == liked {
+		payloadMetadata.likedMutex.RUnlock()
+
+		return
+	}
+
+	payloadMetadata.likedMutex.RUnlock()
+	payloadMetadata.likedMutex.Lock()
+	defer payloadMetadata.likedMutex.Unlock()
+
+	if payloadMetadata.liked == liked {
+		return
+	}
+
+	payloadMetadata.liked = liked
+	payloadMetadata.SetModified()
+	modified = true
+
+	return
+}
+
 // BranchID returns the identifier of the Branch that this Payload was booked into.
 func (payloadMetadata *PayloadMetadata) BranchID() branchmanager.BranchID {
 	payloadMetadata.branchIDMutex.RLock()
@@ -173,7 +207,7 @@ func (payloadMetadata *PayloadMetadata) SetBranchID(branchID branchmanager.Branc
 
 // Bytes marshals the metadata into a sequence of bytes.
 func (payloadMetadata *PayloadMetadata) Bytes() []byte {
-	return marshalutil.New(payload.IDLength + marshalutil.TIME_SIZE + marshalutil.BOOL_SIZE + branchmanager.BranchIDLength).
+	return marshalutil.New(payload.IDLength + marshalutil.TIME_SIZE + 2*marshalutil.BOOL_SIZE + branchmanager.BranchIDLength).
 		WriteBytes(payloadMetadata.ObjectStorageKey()).
 		WriteBytes(payloadMetadata.ObjectStorageValue()).
 		Bytes()
@@ -202,9 +236,10 @@ func (payloadMetadata *PayloadMetadata) Update(other objectstorage.StorableObjec
 
 // ObjectStorageValue is required to match the encoding.BinaryMarshaler interface.
 func (payloadMetadata *PayloadMetadata) ObjectStorageValue() []byte {
-	return marshalutil.New(marshalutil.TIME_SIZE + marshalutil.BOOL_SIZE).
+	return marshalutil.New(marshalutil.TIME_SIZE + 2*marshalutil.BOOL_SIZE).
 		WriteTime(payloadMetadata.solidificationTime).
 		WriteBool(payloadMetadata.solid).
+		WriteBool(payloadMetadata.liked).
 		WriteBytes(payloadMetadata.branchID.Bytes()).
 		Bytes()
 }
@@ -218,6 +253,9 @@ func (payloadMetadata *PayloadMetadata) UnmarshalObjectStorageValue(data []byte)
 	if payloadMetadata.solid, err = marshalUtil.ReadBool(); err != nil {
 		return
 	}
+	if payloadMetadata.liked, err = marshalUtil.ReadBool(); err != nil {
+		return
+	}
 	if payloadMetadata.branchID, err = branchmanager.ParseBranchID(marshalUtil); err != nil {
 		return
 	}

dapps/valuetransfers/packages/tangle/transactionmetadata.go

@@ -20,12 +20,14 @@ type TransactionMetadata struct {
 	id                 transaction.ID
 	branchID           branchmanager.BranchID
 	solid              bool
+	preferred          bool
 	finalized          bool
 	solidificationTime time.Time
 	finalizationTime   time.Time
 
 	branchIDMutex           sync.RWMutex
 	solidMutex              sync.RWMutex
+	preferredMutex          sync.RWMutex
 	finalizedMutex          sync.RWMutex
 	solidificationTimeMutex sync.RWMutex
 }
@@ -129,6 +131,11 @@ func (transactionMetadata *TransactionMetadata) SetBranchID(branchID branchmanag
 	return
 }
 
+// Conflicting returns true if the Transaction has been forked into its own Branch and there is a vote going on.
+func (transactionMetadata *TransactionMetadata) Conflicting() bool {
+	return transactionMetadata.BranchID() == branchmanager.NewBranchID(transactionMetadata.ID())
+}
+
 // Solid returns true if the Transaction has been marked as solid.
 func (transactionMetadata *TransactionMetadata) Solid() (result bool) {
 	transactionMetadata.solidMutex.RLock()
@@ -167,6 +174,40 @@ func (transactionMetadata *TransactionMetadata) SetSolid(solid bool) (modified b
 	return
 }
 
+// Preferred returns true if the transaction is considered to be the first valid spender of all of its Inputs.
+func (transactionMetadata *TransactionMetadata) Preferred() (result bool) {
+	transactionMetadata.preferredMutex.RLock()
+	defer transactionMetadata.preferredMutex.RUnlock()
+
+	return transactionMetadata.preferred
+}
+
+// setPreferred updates the preferred flag of the transaction. It is defined as a private setter because updating the
+// preferred flag causes changes in other transactions and branches as well. This means that we need additional logic
+// in the tangle. To update the preferred flag of a transaction, we need to use Tangle.SetTransactionPreferred(bool).
+func (transactionMetadata *TransactionMetadata) setPreferred(preferred bool) (modified bool) {
+	transactionMetadata.preferredMutex.RLock()
+	if transactionMetadata.preferred == preferred {
+		transactionMetadata.preferredMutex.RUnlock()
+
+		return
+	}
+
+	transactionMetadata.preferredMutex.RUnlock()
+	transactionMetadata.preferredMutex.Lock()
+	defer transactionMetadata.preferredMutex.Unlock()
+
+	if transactionMetadata.preferred == preferred {
+		return
+	}
+
+	transactionMetadata.preferred = preferred
+	transactionMetadata.SetModified()
+	modified = true
+
+	return
+}
+
 // SetFinalized allows us to set the finalized flag on the transactions. Finalized transactions will not be forked when
 // a conflict arrives later.
 func (transactionMetadata *TransactionMetadata) SetFinalized(finalized bool) (modified bool) {
@@ -186,6 +227,7 @@ func (transactionMetadata *TransactionMetadata) SetFinalized(finalized bool) (mo
 	}
 
 	transactionMetadata.finalized = finalized
+	transactionMetadata.SetModified()
 	if finalized {
 		transactionMetadata.finalizationTime = time.Now()
 	}
@@ -220,12 +262,13 @@ func (transactionMetadata *TransactionMetadata) SoldificationTime() time.Time {
 
 // Bytes marshals the TransactionMetadata object into a sequence of bytes.
 func (transactionMetadata *TransactionMetadata) Bytes() []byte {
-	return marshalutil.New(branchmanager.BranchIDLength + 2*marshalutil.TIME_SIZE + 2*marshalutil.BOOL_SIZE).
+	return marshalutil.New(branchmanager.BranchIDLength + 2*marshalutil.TIME_SIZE + 3*marshalutil.BOOL_SIZE).
 		WriteBytes(transactionMetadata.BranchID().Bytes()).
-		WriteTime(transactionMetadata.solidificationTime).
-		WriteTime(transactionMetadata.finalizationTime).
-		WriteBool(transactionMetadata.solid).
-		WriteBool(transactionMetadata.finalized).
+		WriteTime(transactionMetadata.SoldificationTime()).
+		WriteTime(transactionMetadata.FinalizationTime()).
+		WriteBool(transactionMetadata.Solid()).
+		WriteBool(transactionMetadata.Preferred()).
+		WriteBool(transactionMetadata.Finalized()).
 		Bytes()
 }
 
@@ -271,6 +314,9 @@ func (transactionMetadata *TransactionMetadata) UnmarshalObjectStorageValue(data
 	if transactionMetadata.solid, err = marshalUtil.ReadBool(); err != nil {
 		return
 	}
+	if transactionMetadata.preferred, err = marshalUtil.ReadBool(); err != nil {
+		return
+	}
 	if transactionMetadata.finalized, err = marshalUtil.ReadBool(); err != nil {
 		return
 	}

dapps/valuetransfers/packages/test/valuetransfers_test.go → dapps/valuetransfers/packages/test/tangle_test.go

@@ -2,62 +2,66 @@ package test
 
 import (
 	"testing"
-	"time"
 
-	"github.com/iotaledger/hive.go/crypto/ed25519"
+	"github.com/iotaledger/hive.go/events"
 	"github.com/iotaledger/hive.go/kvstore/mapdb"
+	"github.com/iotaledger/hive.go/types"
+
 	"github.com/stretchr/testify/assert"
 
 	"github.com/iotaledger/goshimmer/dapps/valuetransfers/packages/address"
 	"github.com/iotaledger/goshimmer/dapps/valuetransfers/packages/balance"
+	"github.com/iotaledger/goshimmer/dapps/valuetransfers/packages/consensus"
 	"github.com/iotaledger/goshimmer/dapps/valuetransfers/packages/payload"
 	"github.com/iotaledger/goshimmer/dapps/valuetransfers/packages/tangle"
 	"github.com/iotaledger/goshimmer/dapps/valuetransfers/packages/transaction"
+	"github.com/iotaledger/goshimmer/dapps/valuetransfers/packages/wallet"
 )
 
 func TestTangle_ValueTransfer(t *testing.T) {
-	// initialize tangle + ledgerstate
+	// initialize tangle
 	valueTangle := tangle.New(mapdb.NewMapDB())
-	if err := valueTangle.Prune(); err != nil {
-		t.Error(err)
+	defer valueTangle.Shutdown()
 
-		return
-	}
+	// initialize ledger state
 	ledgerState := tangle.NewLedgerState(valueTangle)
 
-	//
-	addressKeyPair1 := ed25519.GenerateKeyPair()
-	addressKeyPair2 := ed25519.GenerateKeyPair()
-	address1 := address.FromED25519PubKey(addressKeyPair1.PublicKey)
-	address2 := address.FromED25519PubKey(addressKeyPair2.PublicKey)
+	// initialize seed
+	seed := wallet.NewSeed()
+
+	// setup consensus rules
+	consensus.NewFCOB(valueTangle, 0)
 
 	// check if ledger empty first
-	assert.Equal(t, map[balance.Color]int64{}, ledgerState.Balances(address1))
-	assert.Equal(t, map[balance.Color]int64{}, ledgerState.Balances(address2))
+	assert.Equal(t, map[balance.Color]int64{}, ledgerState.Balances(seed.Address(0)))
+	assert.Equal(t, map[balance.Color]int64{}, ledgerState.Balances(seed.Address(1)))
 
 	// load snapshot
 	valueTangle.LoadSnapshot(map[transaction.ID]map[address.Address][]*balance.Balance{
 		transaction.GenesisID: {
-			address1: []*balance.Balance{
+			seed.Address(0): []*balance.Balance{
 				balance.New(balance.ColorIOTA, 337),
 			},
 
-			address2: []*balance.Balance{
+			seed.Address(1): []*balance.Balance{
 				balance.New(balance.ColorIOTA, 1000),
 			},
 		},
 	})
 
 	// check if balance exists after loading snapshot
-	assert.Equal(t, map[balance.Color]int64{balance.ColorIOTA: 337}, ledgerState.Balances(address1))
-	assert.Equal(t, map[balance.Color]int64{balance.ColorIOTA: 1000}, ledgerState.Balances(address2))
+	assert.Equal(t, map[balance.Color]int64{balance.ColorIOTA: 337}, ledgerState.Balances(seed.Address(0)))
+	assert.Equal(t, map[balance.Color]int64{balance.ColorIOTA: 1000}, ledgerState.Balances(seed.Address(1)))
+
+	// introduce logic to record liked payloads
+	recordedLikedPayloads, resetRecordedLikedPayloads := recordLikedPayloads(valueTangle)
 
 	// attach first spend
 	outputAddress1 := address.Random()
-	valueTangle.AttachPayload(payload.New(payload.GenesisID, payload.GenesisID, transaction.New(
+	attachedPayload1 := payload.New(payload.GenesisID, payload.GenesisID, transaction.New(
 		transaction.NewInputs(
-			transaction.NewOutputID(address1, transaction.GenesisID),
-			transaction.NewOutputID(address2, transaction.GenesisID),
+			transaction.NewOutputID(seed.Address(0), transaction.GenesisID),
+			transaction.NewOutputID(seed.Address(1), transaction.GenesisID),
 		),
 
 		transaction.NewOutputs(map[address.Address][]*balance.Balance{
@@ -65,24 +69,24 @@ func TestTangle_ValueTransfer(t *testing.T) {
 				balance.New(balance.ColorIOTA, 1337),
 			},
 		}),
-	)))
-
-	// wait for async task to run (TODO: REPLACE TIME BASED APPROACH WITH A WG)
-	time.Sleep(500 * time.Millisecond)
-
-	// check if old addresses are empty
-	assert.Equal(t, map[balance.Color]int64{}, ledgerState.Balances(address1))
-	assert.Equal(t, map[balance.Color]int64{}, ledgerState.Balances(address2))
+	))
+	valueTangle.AttachPayloadSync(attachedPayload1)
 
-	// check if new addresses are filled
+	// check if old addresses are empty and new addresses are filled
+	assert.Equal(t, map[balance.Color]int64{}, ledgerState.Balances(seed.Address(0)))
+	assert.Equal(t, map[balance.Color]int64{}, ledgerState.Balances(seed.Address(1)))
 	assert.Equal(t, map[balance.Color]int64{balance.ColorIOTA: 1337}, ledgerState.Balances(outputAddress1))
+	assert.Equal(t, 1, len(recordedLikedPayloads))
+	assert.Contains(t, recordedLikedPayloads, attachedPayload1.ID())
+
+	resetRecordedLikedPayloads()
 
 	// attach double spend
 	outputAddress2 := address.Random()
-	valueTangle.AttachPayload(payload.New(payload.GenesisID, payload.GenesisID, transaction.New(
+	valueTangle.AttachPayloadSync(payload.New(payload.GenesisID, payload.GenesisID, transaction.New(
 		transaction.NewInputs(
-			transaction.NewOutputID(address1, transaction.GenesisID),
-			transaction.NewOutputID(address2, transaction.GenesisID),
+			transaction.NewOutputID(seed.Address(0), transaction.GenesisID),
+			transaction.NewOutputID(seed.Address(1), transaction.GenesisID),
 		),
 
 		transaction.NewOutputs(map[address.Address][]*balance.Balance{
@@ -91,7 +95,22 @@ func TestTangle_ValueTransfer(t *testing.T) {
 			},
 		}),
 	)))
+}
+
+func recordLikedPayloads(valueTangle *tangle.Tangle) (recordedLikedPayloads map[payload.ID]types.Empty, resetFunc func()) {
+	recordedLikedPayloads = make(map[payload.ID]types.Empty)
+
+	valueTangle.Events.PayloadLiked.Attach(events.NewClosure(func(cachedPayload *payload.CachedPayload, cachedPayloadMetadata *tangle.CachedPayloadMetadata) {
+		defer cachedPayloadMetadata.Release()
+
+		cachedPayload.Consume(func(payload *payload.Payload) {
+			recordedLikedPayloads[payload.ID()] = types.Void
+		})
+	}))
+
+	resetFunc = func() {
+		recordedLikedPayloads = make(map[payload.ID]types.Empty)
+	}
 
-	// shutdown tangle
-	valueTangle.Shutdown()
+	return
 }

dapps/valuetransfers/packages/wallet/seed.go

+package wallet
+
+import (
+	"github.com/iotaledger/hive.go/crypto/ed25519"
+
+	"github.com/iotaledger/goshimmer/dapps/valuetransfers/packages/address"
+)
+
+// Seed represents a seed for IOTA wallets. A seed allows us to generate a deterministic sequence of Addresses and their
+// corresponding KeyPairs.
+type Seed struct {
+	*ed25519.Seed
+}
+
+// NewSeed is the factory method for an IOTA seed. It either generates a new one or imports an existing  marshaled seed.
+// before.
+func NewSeed(optionalSeedBytes ...[]byte) *Seed {
+	return &Seed{
+		ed25519.NewSeed(optionalSeedBytes...),
+	}
+}
+
+// Address returns an ed25519 address which can be used for receiving or sending funds.
+func (seed *Seed) Address(index uint64) address.Address {
+	return address.FromED25519PubKey(seed.Seed.KeyPair(index).PublicKey)
+}

dapps/valuetransfers/packages/wallet/wallet.go

+package wallet
+
+// Wallet represents a simple cryptocurrency wallet for the IOTA tangle. It contains the logic to manage the movement of
+// funds.
+type Wallet struct {
+	seed *Seed
+}
+
+// New is the factory method of the wallet. It either creates a new wallet or restores the wallet backup that is handed
+// in as an optional parameter.
+func New(optionalRecoveryBytes ...[]byte) *Wallet {
+	return &Wallet{
+		seed: NewSeed(optionalRecoveryBytes...),
+	}
+}
+
+// Seed returns the seed of this wallet that is used to generate all of the wallets addresses and private keys.
+func (wallet *Wallet) Seed() *Seed {
+	return wallet.seed
+}