Switch to fixed emitter rate (#323)

gossip/common_test.go

@@ -140,6 +140,7 @@ func newTestEnv(firstEpoch idx.Epoch, validatorsNum idx.Validator, tb testing.TB
 	rules := opera.FakeNetRules()
 	rules.Epochs.MaxEpochDuration = inter.Timestamp(maxEpochDuration)
 	rules.Blocks.MaxEmptyBlockSkipPeriod = 0
+	rules.Emitter.Interval = 0
 
 	genStore := makefakegenesis.FakeGenesisStoreWithRulesAndStart(validatorsNum, utils.ToFtm(genesisBalance), utils.ToFtm(genesisStake), rules, firstEpoch, 2)
 	genesis := genStore.Genesis()

gossip/emitter/control.go

@@ -9,6 +9,7 @@ import (
 	"github.com/Fantom-foundation/lachesis-base/utils/piecefunc"
 
 	"github.com/Fantom-foundation/go-opera/inter"
+	"github.com/Fantom-foundation/go-opera/opera"
 )
 
 func scalarUpdMetric(diff idx.Event, weight pos.Weight, totalWeight pos.Weight) ancestor.Metric {
@@ -46,6 +47,12 @@ func (em *Emitter) isAllowedToEmit(e inter.EventI, eTxs bool, metric ancestor.Me
 	if passedTime < 0 {
 		passedTime = 0
 	}
+
+	// If a emitter interval is defined, all other heuristics are ignored.
+	if interval, enabled := em.getEmitterIntervalLimit(); enabled {
+		return passedTime >= interval
+	}
+
 	passedTimeIdle := e.CreationTime().Time().Sub(em.prevIdleTime)
 	if passedTimeIdle < 0 {
 		passedTimeIdle = 0
@@ -143,3 +150,41 @@ func (em *Emitter) recheckIdleTime() {
 		em.prevIdleTime = time.Now()
 	}
 }
+
+func (em *Emitter) getEmitterIntervalLimit() (interval time.Duration, enabled bool) {
+	rules := em.world.GetRules().Emitter
+
+	var lastConfirmationTime time.Time
+	if last := em.lastTimeAnEventWasConfirmed.Load(); last != nil {
+		lastConfirmationTime = *last
+	} else {
+		// If we have not seen any event confirmed so far, we take the current time
+		// as the last confirmation time. Thus, during start-up we would not unnecessarily
+		// slow down the event emission for the very first event. The switch into the stall
+		// mode is delayed by the stall-threshold.
+		now := time.Now()
+		em.lastTimeAnEventWasConfirmed.Store(&now)
+		lastConfirmationTime = now
+	}
+
+	return getEmitterIntervalLimit(rules, time.Since(lastConfirmationTime))
+}
+
+func getEmitterIntervalLimit(
+	rules opera.EmitterRules,
+	delayOfLastConfirmedEvent time.Duration,
+) (interval time.Duration, enabled bool) {
+	// Check whether the fixed-interval emitter should be enabled.
+	if rules.Interval == 0 {
+		return 0, false
+	}
+
+	// Check for a network-stall situation in which events emitting should be slowed down.
+	stallThreshold := time.Duration(rules.StallThreshold)
+	if delayOfLastConfirmedEvent > stallThreshold {
+		return time.Duration(rules.StalledInterval), true
+	}
+
+	// Use the regular emitter interval.
+	return time.Duration(rules.Interval), true
+}

gossip/emitter/control_test.go

@@ -0,0 +1,50 @@
+package emitter
+
+import (
+	"testing"
+	"time"
+
+	"github.com/Fantom-foundation/go-opera/inter"
+	"github.com/Fantom-foundation/go-opera/opera"
+)
+
+func TestGetEmitterIntervalLimit_IsOffWhenIntervalIsZero(t *testing.T) {
+	ms := time.Microsecond
+	rules := opera.EmitterRules{
+		Interval:       0,
+		StallThreshold: inter.Timestamp(200 * ms),
+	}
+	for _, delay := range []time.Duration{0, 100 * ms, 199 * ms, 200 * ms, 201 * ms} {
+		_, enabled := getEmitterIntervalLimit(rules, delay)
+		if enabled {
+			t.Fatal("should be disabled")
+		}
+	}
+}
+
+func TestGetEmitterIntervalLimit_SwitchesToStallIfDelayed(t *testing.T) {
+	ms := time.Millisecond
+	regular := 100 * ms
+	stallThreshold := 200 * ms
+	stalled := 300 * ms
+
+	rules := opera.EmitterRules{
+		Interval:        inter.Timestamp(regular),
+		StallThreshold:  inter.Timestamp(stallThreshold),
+		StalledInterval: inter.Timestamp(stalled),
+	}
+
+	for _, delay := range []time.Duration{0, 100 * ms, 199 * ms, 200 * ms, 201 * ms, 60 * time.Minute} {
+		got, enabled := getEmitterIntervalLimit(rules, delay)
+		if !enabled {
+			t.Fatalf("should be enabled for delay %v", delay)
+		}
+		want := regular
+		if delay > stallThreshold {
+			want = stalled
+		}
+		if want != got {
+			t.Errorf("for delay %v, want %v, got %v", delay, want, got)
+		}
+	}
+}

gossip/emitter/emitter.go

@@ -8,6 +8,7 @@ import (
 	"os"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/Fantom-foundation/go-opera/utils"
@@ -107,6 +108,8 @@ type Emitter struct {
 	logger.Periodic
 
 	baseFeeSource BaseFeeSource
+
+	lastTimeAnEventWasConfirmed atomic.Pointer[time.Time]
 }
 
 type BaseFeeSource interface {

gossip/emitter/hooks.go

@@ -1,9 +1,10 @@
 package emitter
 
 import (
-	"github.com/Fantom-foundation/go-opera/utils/txtime"
 	"time"
 
+	"github.com/Fantom-foundation/go-opera/utils/txtime"
+
 	"github.com/Fantom-foundation/lachesis-base/emitter/ancestor"
 	"github.com/Fantom-foundation/lachesis-base/inter/idx"
 	"github.com/Fantom-foundation/lachesis-base/inter/pos"
@@ -117,6 +118,8 @@ func (em *Emitter) OnEventConfirmed(he inter.EventI) {
 	if !em.isValidator() {
 		return
 	}
+	now := time.Now()
+	em.lastTimeAnEventWasConfirmed.Store(&now)
 	if em.pendingGas > he.GasPowerUsed() {
 		em.pendingGas -= he.GasPowerUsed()
 	} else {

opera/rules.go

@@ -65,6 +65,9 @@ type RulesRLP struct {
 	// Graph options
 	Dag DagRules
 
+	// Emitter options
+	Emitter EmitterRules
+
 	// Epochs options
 	Epochs EpochsRules
 
@@ -115,6 +118,44 @@ type DagRules struct {
 	MaxExtraData   uint32
 }
 
+// EmitterRules contains options for the emitter of Lachesis events.
+type EmitterRules struct {
+	// Interval defines the length of the period
+	// between events produced by the emitter in milliseconds.
+	// If set to zero, a heuristic is used producing irregular
+	// intervals.
+	//
+	// The Interval is used to control the rate of event
+	// production by the emitter. It thus indirectly controls
+	// the rate of blocks production on the network, by providing
+	// a lower bound. The actual block production rate is also
+	// influenced by the number of validators, their weighting,
+	// and the inter-connection of events. However, the Interval
+	// should provide an effective mean to control the block
+	// production rate.
+	Interval inter.Timestamp
+
+	// StallThreshold defines a maximum time the confirmation of
+	// new events may be delayed before the emitter considers the
+	// network stalled.
+	//
+	// The emitter has two modes: normal and stalled. In normal
+	// mode, the emitter produces events at a regular interval, as
+	// defined by the Interval option. In stalled mode, the emitter
+	// produces events at a much lower rate, to avoid building up
+	// a backlog of events. The StallThreshold defines the upper
+	// limit of delay seen for new confirmed events before the emitter
+	// switches to stalled mode.
+	//
+	// This option is disabled if Interval is set to 0.
+	StallThreshold inter.Timestamp
+
+	// StallInterval defines the length of the period between
+	// events produced by the emitter in milliseconds when the
+	// network is stalled.
+	StalledInterval inter.Timestamp
+}
+
 // BlocksMissed is information about missed blocks from a staker
 type BlocksMissed struct {
 	BlocksNum idx.Block
@@ -250,6 +291,7 @@ func MainNetRules() Rules {
 		Name:      "main",
 		NetworkID: MainNetworkID,
 		Dag:       DefaultDagRules(),
+		Emitter:   DefaultEmitterRules(),
 		Epochs:    DefaultEpochsRules(),
 		Economy:   DefaultEconomyRules(),
 		Blocks: BlocksRules{
@@ -264,6 +306,7 @@ func FakeNetRules() Rules {
 		Name:      "fake",
 		NetworkID: FakeNetworkID,
 		Dag:       DefaultDagRules(),
+		Emitter:   DefaultEmitterRules(),
 		Epochs:    FakeNetEpochsRules(),
 		Economy:   FakeEconomyRules(),
 		Blocks: BlocksRules{
@@ -308,6 +351,14 @@ func DefaultDagRules() DagRules {
 	}
 }
 
+func DefaultEmitterRules() EmitterRules {
+	return EmitterRules{
+		Interval:        inter.Timestamp(600 * time.Millisecond),
+		StallThreshold:  inter.Timestamp(30 * time.Second),
+		StalledInterval: inter.Timestamp(60 * time.Second),
+	}
+}
+
 func DefaultEpochsRules() EpochsRules {
 	return EpochsRules{
 		MaxEpochGas:      1500000000,

tests/block_header_test.go

@@ -279,10 +279,21 @@ func testHeaders_MixDigestDiffersForAllBlocks(t *testing.T, headers []*types.Hea
 	seen := map[common.Hash]struct{}{}
 
 	for i := 1; i < len(headers); i++ {
-		_, ok := seen[headers[i].MixDigest]
-		require.False(ok, "mix digest is not unique")
-		seen[headers[i].MixDigest] = struct{}{}
+		// We skip empty blocks, since in those cases the MixDigest value is not
+		// consumed by any transaction. For those cases, values may be reused.
+		// Since the prev-randao value filling this field is computed based on
+		// the hash of non-empty lachesis events, the value used for empty blocks
+		// is always the same.
+		header := headers[i]
+		if header.GasUsed == 0 {
+			continue
+		}
+		digest := header.MixDigest
+		_, found := seen[digest]
+		require.False(found, "mix digest is not unique, block %d, value %x", i, digest)
+		seen[digest] = struct{}{}
 	}
+	require.NotZero(len(seen), "no non-empty blocks in the chain")
 }
 
 func testHeaders_LastBlockOfEpochContainsSealingTransaction(t *testing.T, headers []*types.Header, client *ethclient.Client) {

tests/gas_price_suggestion_test.go

@@ -37,10 +37,10 @@ func TestGasPrice_SuggestedGasPricesApproximateActualBaseFees(t *testing.T) {
 	}
 
 	// Suggestions should over-estimate the actual prices by ~10%
-	for i := range suggestions {
-		ratio := float64(suggestions[i]) / float64(fees[i])
-		require.Less(1.09, ratio)
-		require.Less(ratio, 1.11)
+	for i := 1; i < int(len(suggestions)); i++ {
+		ratio := float64(suggestions[i]) / float64(fees[i-1])
+		require.Less(1.09, ratio, "step %d, suggestion %d, fees %d", i, suggestions[i], fees[i-1])
+		require.Less(ratio, 1.11, "step %d, suggestion %d, fees %d", i, suggestions[i], fees[i-1])
 	}
 }