Connected telemetry and slashing consumers to consensus distributors.

cmd/access/node_builder/access_node_builder.go

@@ -609,7 +609,7 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionDataRequester() *FlowAccessN
 
 func FlowAccessNode(nodeBuilder *cmd.FlowNodeBuilder) *FlowAccessNodeBuilder {
 	dist := consensuspubsub.NewFollowerDistributor()
-	dist.AddProtocolViolationConsumer(notifications.NewSlashingViolationsConsumer(nodeBuilder.Logger))
+	dist.AddProposalViolationConsumer(notifications.NewSlashingViolationsConsumer(nodeBuilder.Logger))
 	return &FlowAccessNodeBuilder{
 		AccessNodeConfig:    DefaultAccessNodeConfig(),
 		FlowNodeBuilder:     nodeBuilder,

cmd/collection/main.go

@@ -173,7 +173,7 @@ func main() {
 		PreInit(cmd.DynamicStartPreInit).
 		Module("follower distributor", func(node *cmd.NodeConfig) error {
 			followerDistributor = pubsub.NewFollowerDistributor()
-			followerDistributor.AddProtocolViolationConsumer(notifications.NewSlashingViolationsConsumer(node.Logger))
+			followerDistributor.AddProposalViolationConsumer(notifications.NewSlashingViolationsConsumer(node.Logger))
 			return nil
 		}).
 		Module("mutable follower state", func(node *cmd.NodeConfig) error {

cmd/consensus/main.go

@@ -366,7 +366,6 @@ func main() {
 		}).
 		Module("follower distributor", func(node *cmd.NodeConfig) error {
 			followerDistributor = pubsub.NewFollowerDistributor()
-			followerDistributor.AddProtocolViolationConsumer(notifications.NewSlashingViolationsConsumer(nodeBuilder.Logger))
 			return nil
 		}).
 		Module("machine account config", func(node *cmd.NodeConfig) error {
@@ -554,12 +553,17 @@ func main() {
 			// create consensus logger
 			logger := createLogger(node.Logger, node.RootChainID)
 
+			telemetryConsumer := notifications.NewTelemetryConsumer(logger)
+			slashingViolationConsumer := notifications.NewSlashingViolationsConsumer(nodeBuilder.Logger)
+			followerDistributor.AddProposalViolationConsumer(slashingViolationConsumer)
+
 			// initialize a logging notifier for hotstuff
 			notifier := createNotifier(
 				logger,
 				mainMetrics,
 			)
 
+			notifier.AddParticipantConsumer(telemetryConsumer)
 			notifier.AddFollowerConsumer(followerDistributor)
 
 			// initialize the persister
@@ -582,8 +586,11 @@ func main() {
 				return nil, err
 			}
 
-			// TODO: connect to slashing violation consumer
+			// create producer and connect it to consumers
 			voteAggregationDistributor := pubsub.NewVoteAggregationDistributor()
+			voteAggregationDistributor.AddVoteCollectorConsumer(telemetryConsumer)
+			voteAggregationDistributor.AddVoteAggregationViolationConsumer(slashingViolationConsumer)
+
 			validator := consensus.NewValidator(mainMetrics, wrappedCommittee)
 			voteProcessorFactory := votecollector.NewCombinedVoteProcessorFactory(wrappedCommittee, voteAggregationDistributor.OnQcConstructedFromVotes)
 			lowestViewForVoteProcessing := finalizedBlock.View + 1
@@ -600,8 +607,11 @@ func main() {
 				return nil, fmt.Errorf("could not initialize vote aggregator: %w", err)
 			}
 
-			// TODO: connect to slashing violation consumer
+			// create producer and connect it to consumers
 			timeoutAggregationDistributor := pubsub.NewTimeoutAggregationDistributor()
+			timeoutAggregationDistributor.AddTimeoutCollectorConsumer(telemetryConsumer)
+			timeoutAggregationDistributor.AddTimeoutAggregationViolationConsumer(slashingViolationConsumer)
+
 			timeoutProcessorFactory := timeoutcollector.NewTimeoutProcessorFactory(
 				logger,
 				timeoutAggregationDistributor,
@@ -628,7 +638,7 @@ func main() {
 				Committee:                   wrappedCommittee,
 				Signer:                      signer,
 				Persist:                     persist,
-				QCCreatedDistributor:        voteAggregationDistributor.VoteCollectorDistributor,
+				VoteCollectorDistributor:    voteAggregationDistributor.VoteCollectorDistributor,
 				FollowerDistributor:         followerDistributor,
 				TimeoutCollectorDistributor: timeoutAggregationDistributor.TimeoutCollectorDistributor,
 				Forks:                       forks,

cmd/consensus/notifier.go

@@ -17,11 +17,9 @@ func createLogger(log zerolog.Logger, chainID flow.ChainID) zerolog.Logger {
 
 // createNotifier creates a pubsub distributor and connects it to consensus consumers.
 func createNotifier(log zerolog.Logger, metrics module.HotstuffMetrics) *pubsub.Distributor {
-	telemetryConsumer := notifications.NewTelemetryConsumer(log)
 	metricsConsumer := metricsconsumer.NewMetricsConsumer(metrics)
 	logsConsumer := notifications.NewLogConsumer(log)
 	dis := pubsub.NewDistributor()
-	dis.AddConsumer(telemetryConsumer)
 	dis.AddConsumer(metricsConsumer)
 	dis.AddConsumer(logsConsumer)
 	return dis

cmd/execution_builder.go

@@ -274,7 +274,7 @@ func (exeNode *ExecutionNode) LoadExecutionReceiptsStorage(
 
 func (exeNode *ExecutionNode) LoadFollowerDistributor(node *NodeConfig) error {
 	exeNode.followerDistributor = pubsub.NewFollowerDistributor()
-	exeNode.followerDistributor.AddProtocolViolationConsumer(notifications.NewSlashingViolationsConsumer(node.Logger))
+	exeNode.followerDistributor.AddProposalViolationConsumer(notifications.NewSlashingViolationsConsumer(node.Logger))
 	return nil
 }
 

cmd/observer/node_builder/observer_builder.go

@@ -566,7 +566,7 @@ func NewFlowObserverServiceBuilder(opts ...Option) *ObserverServiceBuilder {
 		FlowNodeBuilder:       cmd.FlowNode("observer"),
 		FollowerDistributor:   pubsub.NewFollowerDistributor(),
 	}
-	anb.FollowerDistributor.AddProtocolViolationConsumer(notifications.NewSlashingViolationsConsumer(anb.Logger))
+	anb.FollowerDistributor.AddProposalViolationConsumer(notifications.NewSlashingViolationsConsumer(anb.Logger))
 	// the observer gets a version of the root snapshot file that does not contain any node addresses
 	// hence skip all the root snapshot validations that involved an identity address
 	anb.FlowNodeBuilder.SkipNwAddressBasedValidations = true

cmd/verification_builder.go

@@ -179,7 +179,7 @@ func (v *VerificationNodeBuilder) LoadComponentsAndModules() {
 		}).
 		Module("follower distributor", func(node *NodeConfig) error {
 			followerDistributor = pubsub.NewFollowerDistributor()
-			followerDistributor.AddProtocolViolationConsumer(notifications.NewSlashingViolationsConsumer(node.Logger))
+			followerDistributor.AddProposalViolationConsumer(notifications.NewSlashingViolationsConsumer(node.Logger))
 			return nil
 		}).
 		Module("sync core", func(node *NodeConfig) error {

consensus/config.go

@@ -12,17 +12,17 @@ import (
 // HotstuffModules is a helper structure to encapsulate dependencies to create
 // a hotStuff participant.
 type HotstuffModules struct {
-	Committee                   hotstuff.DynamicCommittee        // consensus committee
-	Signer                      hotstuff.Signer                  // signer of proposal & votes
-	Persist                     hotstuff.Persister               // last state of consensus participant
-	Notifier                    *pubsub.Distributor              // observer for hotstuff events
-	FollowerDistributor         *pubsub.FollowerDistributor      // observer for finalization events, used by compliance engine
-	QCCreatedDistributor        *pubsub.VoteCollectorDistributor // observer for qc created event, used by leader
-	TimeoutCollectorDistributor *pubsub.TimeoutCollectorDistributor
-	Forks                       hotstuff.Forks             // information about multiple forks
-	Validator                   hotstuff.Validator         // validator of proposals & votes
-	VoteAggregator              hotstuff.VoteAggregator    // aggregator of votes, used by leader
-	TimeoutAggregator           hotstuff.TimeoutAggregator // aggregator of `TimeoutObject`s, used by every replica
+	Committee                   hotstuff.DynamicCommittee           // consensus committee
+	Signer                      hotstuff.Signer                     // signer of proposal & votes
+	Persist                     hotstuff.Persister                  // last state of consensus participant
+	Notifier                    *pubsub.Distributor                 // observer for hotstuff events
+	FollowerDistributor         *pubsub.FollowerDistributor         // observer for finalization events, used by compliance engine
+	VoteCollectorDistributor    *pubsub.VoteCollectorDistributor    // observer for vote aggregation events, used by leader
+	TimeoutCollectorDistributor *pubsub.TimeoutCollectorDistributor // observer for timeout aggregation events
+	Forks                       hotstuff.Forks                      // information about multiple forks
+	Validator                   hotstuff.Validator                  // validator of proposals & votes
+	VoteAggregator              hotstuff.VoteAggregator             // aggregator of votes, used by leader
+	TimeoutAggregator           hotstuff.TimeoutAggregator          // aggregator of `TimeoutObject`s, used by every replica
 }
 
 type ParticipantConfig struct {

consensus/hotstuff/notifications/pubsub/distributor.go

@@ -47,7 +47,7 @@ func NewFollowerDistributor() *FollowerDistributor {
 
 func (d *FollowerDistributor) AddFollowerConsumer(consumer hotstuff.FollowerConsumer) {
 	d.FinalizationDistributor.AddFinalizationConsumer(consumer)
-	d.ProtocolViolationDistributor.AddProtocolViolationConsumer(consumer)
+	d.ProtocolViolationDistributor.AddProposalViolationConsumer(consumer)
 }
 
 type TimeoutAggregationDistributor struct {
@@ -85,5 +85,5 @@ func NewVoteAggregationDistributor() *VoteAggregationDistributor {
 
 func (d *VoteAggregationDistributor) AddVoteAggregationConsumer(consumer hotstuff.VoteAggregationConsumer) {
 	d.VoteAggregationViolationDistributor.AddVoteAggregationViolationConsumer(consumer)
-	d.VoteCollectorDistributor.AddQCCreatedConsumer(consumer)
+	d.VoteCollectorDistributor.AddVoteCollectorConsumer(consumer)
 }

consensus/hotstuff/notifications/pubsub/protocol_violation_distributor.go

@@ -22,7 +22,7 @@ func NewProtocolViolationDistributor() *ProtocolViolationDistributor {
 	return &ProtocolViolationDistributor{}
 }
 
-func (d *ProtocolViolationDistributor) AddProtocolViolationConsumer(consumer hotstuff.ProposalViolationConsumer) {
+func (d *ProtocolViolationDistributor) AddProposalViolationConsumer(consumer hotstuff.ProposalViolationConsumer) {
 	d.lock.Lock()
 	defer d.lock.Unlock()
 	d.subscribers = append(d.subscribers, consumer)

consensus/hotstuff/notifications/pubsub/vote_collector_distributor.go

@@ -23,7 +23,7 @@ func NewQCCreatedDistributor() *VoteCollectorDistributor {
 	return &VoteCollectorDistributor{}
 }
 
-func (d *VoteCollectorDistributor) AddQCCreatedConsumer(consumer hotstuff.VoteCollectorConsumer) {
+func (d *VoteCollectorDistributor) AddVoteCollectorConsumer(consumer hotstuff.VoteCollectorConsumer) {
 	d.lock.Lock()
 	defer d.lock.Unlock()
 	d.consumers = append(d.consumers, consumer)

consensus/hotstuff/notifications/telemetry.go

@@ -32,12 +32,15 @@ import (
 //
 // Telemetry does NOT capture slashing notifications
 type TelemetryConsumer struct {
-	NoopConsumer
+	NoopTimeoutCollectorConsumer
+	NoopVoteCollectorConsumer
 	pathHandler  *PathHandler
 	noPathLogger zerolog.Logger
 }
 
-var _ hotstuff.Consumer = (*TelemetryConsumer)(nil)
+var _ hotstuff.ParticipantConsumer = (*TelemetryConsumer)(nil)
+var _ hotstuff.VoteCollectorConsumer = (*TelemetryConsumer)(nil)
+var _ hotstuff.TimeoutCollectorConsumer = (*TelemetryConsumer)(nil)
 
 // NewTelemetryConsumer creates consumer that reports telemetry events using logger backend.
 // Logger MUST include `chain` parameter as part of log context with corresponding chain ID to correctly map telemetry events to chain.
@@ -240,6 +243,13 @@ func (t *TelemetryConsumer) OnCurrentViewDetails(currentView, finalizedView uint
 		Msg("OnCurrentViewDetails")
 }
 
+func (t *TelemetryConsumer) OnViewChange(oldView, newView uint64) {
+	t.pathHandler.NextStep().
+		Uint64("old_view", oldView).
+		Uint64("new_view", newView).
+		Msg("OnViewChange")
+}
+
 // PathHandler maintains a notion of the current path through the state machine.
 // It allows to close a path and open new path. Each path is identified by a unique
 // (randomly generated) uuid. Along each path, we can capture information about relevant

consensus/integration/nodes_test.go

@@ -566,7 +566,7 @@ func createNode(
 		Committee:                   committee,
 		Signer:                      signer,
 		Persist:                     persist,
-		QCCreatedDistributor:        qcDistributor,
+		VoteCollectorDistributor:    qcDistributor,
 		TimeoutCollectorDistributor: timeoutCollectorDistributor,
 		VoteAggregator:              voteAggregator,
 		TimeoutAggregator:           timeoutAggregator,

consensus/participant.go

@@ -122,7 +122,7 @@ func NewParticipant(
 	}
 
 	// add observer, event loop needs to receive events from distributor
-	modules.QCCreatedDistributor.AddQCCreatedConsumer(loop)
+	modules.VoteCollectorDistributor.AddVoteCollectorConsumer(loop)
 	modules.TimeoutCollectorDistributor.AddTimeoutCollectorConsumer(loop)
 
 	return loop, nil

engine/collection/epochmgr/factories/hotstuff.go

@@ -79,7 +79,7 @@ func (f *HotStuffFactory) CreateModules(
 	notifier.AddConsumer(notifications.NewLogConsumer(log))
 	notifier.AddConsumer(hotmetrics.NewMetricsConsumer(metrics))
 	notifier.AddConsumer(notifications.NewTelemetryConsumer(log))
-	notifier.AddProtocolViolationConsumer(notifications.NewSlashingViolationsConsumer(log))
+	notifier.AddProposalViolationConsumer(notifications.NewSlashingViolationsConsumer(log))
 
 	var (
 		err       error
@@ -159,7 +159,7 @@ func (f *HotStuffFactory) CreateModules(
 		Persist:                     persister.New(f.db, cluster.ChainID()),
 		VoteAggregator:              voteAggregator,
 		TimeoutAggregator:           timeoutAggregator,
-		QCCreatedDistributor:        qcDistributor,
+		VoteCollectorDistributor:    qcDistributor,
 		TimeoutCollectorDistributor: timeoutCollectorDistributor,
 		FollowerDistributor:         notifier.FollowerDistributor,
 	}, metrics, nil

follower/follower_builder.go

@@ -357,7 +357,7 @@ func FlowConsensusFollowerService(opts ...FollowerOption) *FollowerServiceBuilde
 		FlowNodeBuilder:     cmd.FlowNode(flow.RoleAccess.String(), config.baseOptions...),
 		FollowerDistributor: pubsub.NewFollowerDistributor(),
 	}
-	ret.FollowerDistributor.AddProtocolViolationConsumer(notifications.NewSlashingViolationsConsumer(ret.Logger))
+	ret.FollowerDistributor.AddProposalViolationConsumer(notifications.NewSlashingViolationsConsumer(ret.Logger))
 	// the observer gets a version of the root snapshot file that does not contain any node addresses
 	// hence skip all the root snapshot validations that involved an identity address
 	ret.FlowNodeBuilder.SkipNwAddressBasedValidations = true