Adds rate setter in data flow (#1557)

* feat: Move scheduler after booker and disable orderer

* adjust flow

* refactor: Update parentsMap right after MessageOrdererd event

* fix: Disable wait group in fifoscheduler

* feat: Remove FIFO-scheduler

* feat: Add Error event in scheduler

* feat: Increase scheduler rate if node's not synced at start

* fix: Await messages get scheduled in unit test

* fix TestOpinionFormer_Scenario2 test

* feat: Move TipManager after Orderer

* refactor: Remove unused chan in tangle.go

* fix: Avoid logging logged errors in scheduler

* fix: Attach tip manager to OpinionFormed event

* fix: Adjust unit test component setup wrt the scheduler moving

* feat: Make booking running in a single go-routine

* fix: Await all messages to be booked in the booker

* plug in ratesetter

* add rate setter to info API

* refactor: rename MessageRated event

* dummy rate setter estimate

* slow down spammer

* refactor ratesetter.estimate

* add estimate to info endpoint

* fix rate setter initial rate

* Add estimate to dashboard

* fix goimports

* update dashboard

* refactor

* add estimate to /data

* revert to use message instead of payload

* pkger

* fixes compilation errors

* update pkged

* Make rate setter work, adapt parameters.

* Add rate setter metrics to Grafana.

* Update dashboard.

* Add ratesetter info endpoint and other updates.

* Fix unit tests.

* Disable rate setter in integration tests.

* Fix integration tests

* Change type of variable of rounds to skip

* Change type of variable of rounds to skip

* Disable ratesetter webapi endpoint in entrynode

* Fix negative deficit issue

* Merge develop to feat/rate-setter

* Merge develop to feat/rate-setter

* Fix stuff

* Remove println

* Fix feature network and print

* Add paramater to enable debugger

* Update dashboard

* Fix negative deficit problem

* Update evil wallet and cli wallet to use rate setter. Fix rate setter.

* Remove blocking from API endpoints.

* Fix integration tests.

* Update dashboard

* Fix integration tests

* Revert "Remove blocking from API endpoints."

This reverts commit ae62a24.

* Revert API changes and test fixes

* Revert test running script.

* Add rate setter integration to activity plugin.

* Set default PoW difficulty to zero and update scheduling rate.

* Adjust rate setter parameters. Fix tests.

* Add scheduler metrics to dashboard and webapi.

* AccessMana is now Mana1

* package/mana/access* all gone
* consensusbase renamed to manabase implementing base interface for both access and consensus mana
* consensusbasevector renamed to manabasevector implementing basevector interface for both access and consensus mana
* both mana vectors are now maintained using 2 instances of ManaBaseVector (former ConsensusBaseVector)
* update methods removed from both interfaces and implementations
* SetCoefficients gone, along with corresponding mana plugin parameters
    *
* EffectiveValue methods removed, only BaseValue remains
* manarefresher is gone
    * DelegationAddress stuff managed by the plugin also gone
        * Reclaim functionality removed
    * corresponding WebAPI endpoints and response fields removed
    * cli-wallet delegation & reclaim removed
    * occurrences removed from Ansible and docker-compose files
* GetMana and GetManaMap methods do not receive and optional time anymore, but they still return a timestamp which is always set to time.Now(). This timestamp should be tied to the epoch timestamp for the committed mana vector.

 Conflicts:
	packages/mana/accessbasevector.go
	packages/mana/manabasevector.go
	packages/mana/parameters.go
	plugins/core.go

* Remove references to --mana.snapshotResetTime=true

* Address PR review

* Fixes after merge

* Fix rate seter params and bugs

* Adjust rate setter params

* More parameter adjustment.

* Fix integration tests.

* Update hive.go dependency

* Address review comments.

* Mana initializer plugin and faucet web API endpoint (#2296)

* Add manainitializer plugin and API endpoint.

* Address review comments.

* Simplify code

* Add !

Co-authored-by: jonastheis <4181434+jonastheis@users.noreply.github.com>

Co-authored-by: jkrvivian <jkrvivian@gmail.com>
Co-authored-by: Luca Moser <moser.luca@gmail.com>
Co-authored-by: Piotr Macek <piotr.macek@iota.org>
Co-authored-by: Andrea V <1577639+karimodm@users.noreply.github.com>
Co-authored-by: jonastheis <4181434+jonastheis@users.noreply.github.com>

.github/workflows/feature-network-deploy.yml

@@ -7,9 +7,13 @@ on:
         required: false
         default: ""
       powDifficulty:
-        description: 'PoW difficulty (default: 21):'
+        description: 'PoW difficulty (default: 0):'
         required: false
         default: ""
+      remoteDebugging:
+        description: 'Enable debugger (0/1):'
+        required: true
+        default: "0"
 jobs:
   deploy:
     environment: feature
@@ -45,7 +49,7 @@ jobs:
           build-args: |
             CUSTOM_SNAPSHOT_URL=${{github.event.inputs.snapshotUrl}}
             DEFAULT_SNAPSHOT_URL=https://dbfiles-goshimmer.s3.eu-central-1.amazonaws.com/snapshots/feature/snapshot.bin
-            REMOTE_DEBUGGING=1
+            REMOTE_DEBUGGING=${{github.event.inputs.remoteDebugging}}
           cache-from: type=local,src=/tmp/.buildx-cache
           cache-to: type=local,mode=max,dest=/tmp/.buildx-cache-new
 

client/evilspammer/errors.go

@@ -2,14 +2,16 @@ package evilspammer
 
 import (
 	"fmt"
+	"sync"
+
 	"github.com/cockroachdb/errors"
 	"go.uber.org/atomic"
-	"sync"
 )
 
 var (
 	ErrFailPostTransaction = errors.New("failed to post transaction")
 	ErrFailSendDataMessage = errors.New("failed to send a data message")
+
 	ErrTransactionIsNil    = errors.New("provided transaction is nil")
 	ErrFailToPrepareBatch  = errors.New("custom conflict batch could not be prepared")
 	ErrInsufficientClients = errors.New("insufficient clients to send conflicts")

client/evilspammer/options.go

@@ -36,6 +36,13 @@ func WithSpamDuration(maxDuration time.Duration) Options {
 	}
 }
 
+// WithRateSetter enables setting rate of spammer.
+func WithRateSetter(enable bool) Options {
+	return func(s *Spammer) {
+		s.UseRateSetter = enable
+	}
+}
+
 // WithBatchesSent provides spammer with options regarding rate, time unit, and finishing spam criteria. Provide 0 to one of max parameters to skip it.
 func WithBatchesSent(maxBatchesSent int) Options {
 	return func(s *Spammer) {

client/evilspammer/spammer.go

@@ -35,14 +35,14 @@ type State struct {
 // Not mandatory options, if not provided spammer will use default settings:
 // WithSpamDetails, WithEvilWallet, WithErrorCounter, WithLogTickerInterval
 type Spammer struct {
-	SpamDetails *SpamDetails
-	State       *State
-
-	Clients      evilwallet.Connector
-	EvilWallet   *evilwallet.EvilWallet
-	EvilScenario *evilwallet.EvilScenario
-	ErrCounter   *ErrorCounter
-	log          Logger
+	SpamDetails   *SpamDetails
+	State         *State
+	UseRateSetter bool
+	Clients       evilwallet.Connector
+	EvilWallet    *evilwallet.EvilWallet
+	EvilScenario  *evilwallet.EvilScenario
+	ErrCounter    *ErrorCounter
+	log           Logger
 
 	// accessed from spamming functions
 	done     chan bool
@@ -65,6 +65,7 @@ func NewSpammer(options ...Options) *Spammer {
 		spamFunc:       CustomConflictSpammingFunc,
 		State:          state,
 		EvilScenario:   evilwallet.NewEvilScenario(),
+		UseRateSetter:  true,
 		done:           make(chan bool),
 		shutdown:       make(chan types.Empty),
 		NumberOfSpends: 2,
@@ -148,6 +149,7 @@ func (s *Spammer) Spam() {
 	timeExceeded := time.After(s.SpamDetails.MaxDuration)
 
 	go func() {
+		goroutineCount := atomic.NewInt32(0)
 		for {
 			select {
 			case <-s.State.logTicker.C:
@@ -160,7 +162,14 @@ func (s *Spammer) Spam() {
 				s.StopSpamming()
 				return
 			case <-s.State.spamTicker.C:
-				go s.spamFunc(s)
+				if goroutineCount.Load() > 100 {
+					break
+				}
+				go func() {
+					goroutineCount.Inc()
+					defer goroutineCount.Dec()
+					s.spamFunc(s)
+				}()
 			}
 		}
 	}()
@@ -200,6 +209,9 @@ func (s *Spammer) PostTransaction(tx *devnetvm.Transaction, clt evilwallet.Clien
 
 	var err error
 	var txID utxo.TransactionID
+	if err = evilwallet.RateSetterSleep(clt, s.UseRateSetter); err != nil {
+		return
+	}
 	txID, err = clt.PostTransaction(tx)
 	if err != nil {
 		s.log.Debug(ErrFailPostTransaction)

client/evilspammer/spamming_functions.go

@@ -2,7 +2,9 @@ package evilspammer
 
 import (
 	"fmt"
+	"math/rand"
 	"sync"
+	"time"
 
 	"github.com/cockroachdb/errors"
 
@@ -12,10 +14,16 @@ import (
 
 func DataSpammingFunction(s *Spammer) {
 	clt := s.Clients.GetClient()
+	// sleep randomly to avoid issuing messages in different goroutines at once
+	time.Sleep(time.Duration(rand.Float64()*20) * time.Millisecond)
+	if err := evilwallet.RateSetterSleep(clt, s.UseRateSetter); err != nil {
+		s.ErrCounter.CountError(err)
+	}
 	msgID, err := clt.PostData([]byte(fmt.Sprintf("SPAM")))
 	if err != nil {
 		s.ErrCounter.CountError(ErrFailSendDataMessage)
 	}
+
 	count := s.State.txSent.Add(1)
 	if count%int64(s.SpamDetails.Rate*2) == 0 {
 		s.log.Debugf("Last sent message, ID: %s; msgCount: %d", msgID, count)
@@ -44,6 +52,12 @@ func CustomConflictSpammingFunc(s *Spammer) {
 			wg.Add(1)
 			go func(clt evilwallet.Client, tx *devnetvm.Transaction) {
 				defer wg.Done()
+
+				// sleep randomly to avoid issuing messages in different goroutines at once
+				time.Sleep(time.Duration(rand.Float64()*100) * time.Millisecond)
+				if err = evilwallet.RateSetterSleep(clt, s.UseRateSetter); err != nil {
+					s.ErrCounter.CountError(err)
+				}
 				s.PostTransaction(tx, clt)
 			}(clients[i], tx)
 		}

client/evilwallet/connector.go

@@ -2,6 +2,7 @@ package evilwallet
 
 import (
 	"sync"
+	"time"
 
 	"github.com/iotaledger/hive.go/identity"
 
@@ -166,6 +167,10 @@ func (c *WebClients) SetPledgeID(id *identity.ID) {
 type Client interface {
 	// Url returns a client API url.
 	Url() (cltID string)
+	// GetRateSetterEstimate returns a rate setter estimate.
+	GetRateSetterEstimate() (estimate time.Duration, err error)
+	// SleepRateSetterEstimate sleeps for rate setter estimate.
+	SleepRateSetterEstimate() (err error)
 	// PostTransaction sends a transaction to the Tangle via a given client.
 	PostTransaction(tx *devnetvm.Transaction) (utxo.TransactionID, error)
 	// PostData sends the given data (payload) by creating a message in the backend.
@@ -180,8 +185,8 @@ type Client interface {
 	GetOutput(outputID utxo.OutputID) devnetvm.Output
 	// GetOutputGoF gets the first unspent outputs of a given address.
 	GetOutputGoF(outputID utxo.OutputID) gof.GradeOfFinality
-	// SendFaucetRequest requests funds from the faucet and returns the faucet request message ID.
-	SendFaucetRequest(address string) error
+	// BroadcastFaucetRequest requests funds from the faucet and returns the faucet request message ID.
+	BroadcastFaucetRequest(address string) error
 	// GetTransactionOutputs returns the outputs the transaction created.
 	GetTransactionOutputs(txID string) (outputs devnetvm.Outputs, err error)
 	// GetTransaction gets the transaction.
@@ -209,9 +214,27 @@ func NewWebClient(url string, setters ...client.Option) *WebClient {
 	}
 }
 
-// SendFaucetRequest requests funds from the faucet and returns the faucet request message ID.
-func (c *WebClient) SendFaucetRequest(address string) (err error) {
-	_, err = c.api.SendFaucetRequest(address, -1)
+// GetRateSetterEstimate returns a rate setter estimate.
+func (c *WebClient) GetRateSetterEstimate() (estimate time.Duration, err error) {
+	response, err := c.api.RateSetter()
+	if err != nil {
+		return time.Duration(0), err
+	}
+	return response.Estimate, nil
+}
+
+// SleepRateSetterEstimate returns a rate setter estimate.
+func (c *WebClient) SleepRateSetterEstimate() (err error) {
+	err = c.api.SleepRateSetterEstimate()
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// BroadcastFaucetRequest requests funds from the faucet and returns the faucet request message ID.
+func (c *WebClient) BroadcastFaucetRequest(address string) (err error) {
+	_, err = c.api.BroadcastFaucetRequest(address, -1)
 	return
 }
 

client/evilwallet/evilwallet.go

@@ -188,7 +188,10 @@ func (e *EvilWallet) requestAndSplitFaucetFunds(initWallet *Wallet, splitNum int
 func (e *EvilWallet) requestFaucetFunds(wallet *Wallet) (outputID utxo.OutputID, err error) {
 	addr := wallet.Address()
 	clt := e.connector.GetClient()
-	err = clt.SendFaucetRequest(addr.Base58())
+	if err = RateSetterSleep(clt, true); err != nil {
+		return
+	}
+	err = clt.BroadcastFaucetRequest(addr.Base58())
 	if err != nil {
 		return
 	}
@@ -252,6 +255,9 @@ func (e *EvilWallet) splitOutputs(inputWallet, outputWallet *Wallet, splitNumber
 				}
 
 				clt := e.connector.GetClient()
+				if err = RateSetterSleep(clt, true); err != nil {
+					return
+				}
 				txID, err := clt.PostTransaction(tx)
 				if err != nil {
 					return
@@ -334,6 +340,9 @@ func (e *EvilWallet) SendCustomConflicts(conflictsMaps []ConflictSlice) (err err
 			wg.Add(1)
 			go func(clt Client, tx *devnetvm.Transaction) {
 				defer wg.Done()
+				if err = RateSetterSleep(clt, true); err != nil {
+					return
+				}
 				_, _ = clt.PostTransaction(tx)
 			}(clients[i], tx)
 		}

client/evilwallet/utils.go

@@ -52,3 +52,14 @@ func getIotaColorAmount(balance *devnetvm.ColoredBalances) uint64 {
 	})
 	return outBalance
 }
+
+// RateSetterSleep sleeps for the given rate.
+func RateSetterSleep(clt Client, useRateSetter bool) error {
+	if useRateSetter {
+		err := clt.SleepRateSetterEstimate()
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}

client/faucet.go

@@ -6,7 +6,6 @@ import (
 
 	"github.com/cockroachdb/errors"
 	"github.com/iotaledger/hive.go/identity"
-	"github.com/mr-tron/base58"
 
 	"github.com/iotaledger/goshimmer/packages/faucet"
 	"github.com/iotaledger/goshimmer/packages/jsonmodels"
@@ -16,16 +15,17 @@ import (
 )
 
 const (
-	routeFaucet = "faucet"
+	routeFaucetRequestBroadcast = "faucetrequest"
+	routeFaucetRequestAPI       = "faucet"
 )
 
 var (
 	defaultPOWTarget = 25
 	powWorker        = pow.New(1)
 )
 
-// SendFaucetRequest requests funds from faucet nodes by sending a faucet request payload message.
-func (api *GoShimmerAPI) SendFaucetRequest(base58EncodedAddr string, powTarget int, pledgeIDs ...string) (*jsonmodels.FaucetResponse, error) {
+// BroadcastFaucetRequest requests funds from faucet nodes by sending a faucet request payload message.
+func (api *GoShimmerAPI) BroadcastFaucetRequest(base58EncodedAddr string, powTarget int, pledgeIDs ...string) (*jsonmodels.FaucetRequestResponse, error) {
 	var aManaPledgeID identity.ID
 	var cManaPledgeID identity.ID
 	if len(pledgeIDs) > 1 {
@@ -49,12 +49,52 @@ func (api *GoShimmerAPI) SendFaucetRequest(base58EncodedAddr string, powTarget i
 		return nil, errors.Errorf("could not compute faucet PoW: %w", err)
 	}
 
-	res := &jsonmodels.FaucetResponse{}
-	if err := api.do(http.MethodPost, routeFaucet,
+	res := &jsonmodels.FaucetRequestResponse{}
+	if err := api.do(http.MethodPost, routeFaucetRequestBroadcast,
 		&jsonmodels.FaucetRequest{
 			Address:               base58EncodedAddr,
-			AccessManaPledgeID:    base58.Encode(aManaPledgeID.Bytes()),
-			ConsensusManaPledgeID: base58.Encode(cManaPledgeID.Bytes()),
+			AccessManaPledgeID:    aManaPledgeID.EncodeBase58(),
+			ConsensusManaPledgeID: cManaPledgeID.EncodeBase58(),
+			Nonce:                 nonce,
+		}, res); err != nil {
+		return nil, err
+	}
+
+	return res, nil
+}
+
+// SendFaucetRequestAPI requests funds from faucet nodes by sending a faucet request directly to the faucet node.
+func (api *GoShimmerAPI) SendFaucetRequestAPI(base58EncodedAddr string, powTarget int, accessPledgeID, consensusPledgeID string) (*jsonmodels.FaucetAPIResponse, error) {
+	var aManaPledgeID identity.ID
+	var cManaPledgeID identity.ID
+	if accessPledgeID == "" && consensusPledgeID == "" {
+		return nil, errors.Errorf("accessPledgeID and consensusPledgeID must not be empty")
+	}
+	aManaPledgeIDFromString, err := mana.IDFromStr(accessPledgeID)
+	if err == nil {
+		aManaPledgeID = aManaPledgeIDFromString
+	}
+	cManaPledgeIDFromString, err := mana.IDFromStr(consensusPledgeID)
+	if err == nil {
+		cManaPledgeID = cManaPledgeIDFromString
+	}
+
+	address, err := devnetvm.AddressFromBase58EncodedString(base58EncodedAddr)
+	if err != nil {
+		return nil, errors.Errorf("could not decode address from string: %w", err)
+	}
+
+	nonce, err := computeFaucetPoW(address, aManaPledgeID, cManaPledgeID, powTarget)
+	if err != nil {
+		return nil, errors.Errorf("could not compute faucet PoW: %w", err)
+	}
+
+	res := &jsonmodels.FaucetAPIResponse{}
+	if err := api.do(http.MethodPost, routeFaucetRequestAPI,
+		&jsonmodels.FaucetRequest{
+			Address:               base58EncodedAddr,
+			AccessManaPledgeID:    aManaPledgeID.EncodeBase58(),
+			ConsensusManaPledgeID: cManaPledgeID.EncodeBase58(),
 			Nonce:                 nonce,
 		}, res); err != nil {
 		return nil, err

client/ratesetter.go

@@ -0,0 +1,31 @@
+package client
+
+import (
+	"net/http"
+	"time"
+
+	"github.com/iotaledger/goshimmer/packages/jsonmodels"
+)
+
+const (
+	rateSetterInfo = "ratesetter"
+)
+
+// RateSetter gets the ratesetter estimate and the rate-setter info.
+func (api *GoShimmerAPI) RateSetter() (*jsonmodels.RateSetter, error) {
+	res := &jsonmodels.RateSetter{}
+	if err := api.do(http.MethodGet, rateSetterInfo, nil, res); err != nil {
+		return nil, err
+	}
+	return res, nil
+}
+
+// SleepRateSetterEstimate gets the rate-setter estimate and the rate-setter info and later sleeps the estimated amount of time.
+func (api *GoShimmerAPI) SleepRateSetterEstimate() error {
+	res, err := api.RateSetter()
+	if err != nil {
+		return err
+	}
+	time.Sleep(res.Estimate)
+	return nil
+}