ceremonyclient/consensus/notifications/pubsub/vote_aggregation_violation_consumer.go
Cassandra Heart c797d482f9
v2.1.0.5 (#457)
* wip: conversion of hotstuff from flow into Q-oriented model

* bulk of tests

* remaining non-integration tests

* add integration test, adjust log interface, small tweaks

* further adjustments, restore full pacemaker shape

* add component lifecycle management+supervisor

* further refinements

* resolve timeout hanging

* mostly finalized state for consensus

* bulk of engine swap out

* lifecycle-ify most types

* wiring nearly complete, missing needed hooks for proposals

* plugged in, vetting message validation paths

* global consensus, plugged in and verified

* app shard now wired in too

* do not decode empty keys.yml (#456)

* remove obsolete engine.maxFrames config parameter (#454)

* default to Info log level unless debug is enabled (#453)

* respect config's  "logging" section params, remove obsolete single-file logging (#452)

* Trivial code cleanup aiming to reduce Go compiler warnings (#451)

* simplify range traversal

* simplify channel read for single select case

* delete rand.Seed() deprecated in Go 1.20 and no-op as of Go 1.24

* simplify range traversal

* simplify channel read for single select case

* remove redundant type from array

* simplify range traversal

* simplify channel read for single select case

* RC slate

* finalize 2.1.0.5

* Update comments in StrictMonotonicCounter

Fix comment formatting and clarify description.

---------

Co-authored-by: Black Swan <3999712+blacks1ne@users.noreply.github.com>
2025-11-11 05:00:17 -06:00

76 lines
2.0 KiB
Go

package pubsub
import (
"sync"
"source.quilibrium.com/quilibrium/monorepo/consensus"
"source.quilibrium.com/quilibrium/monorepo/consensus/models"
)
// VoteAggregationViolationDistributor ingests notifications about vote
// aggregation violations and distributes them to consumers. Such notifications
// are produced by the vote aggregation logic. Concurrency safe.
type VoteAggregationViolationDistributor[
StateT models.Unique,
VoteT models.Unique,
] struct {
consumers []consensus.VoteAggregationViolationConsumer[StateT, VoteT]
lock sync.RWMutex
}
var _ consensus.VoteAggregationViolationConsumer[*nilUnique, *nilUnique] = (*VoteAggregationViolationDistributor[*nilUnique, *nilUnique])(nil)
func NewVoteAggregationViolationDistributor[
StateT models.Unique,
VoteT models.Unique,
]() *VoteAggregationViolationDistributor[StateT, VoteT] {
return &VoteAggregationViolationDistributor[StateT, VoteT]{}
}
func (d *VoteAggregationViolationDistributor[
StateT,
VoteT,
]) AddVoteAggregationViolationConsumer(
consumer consensus.VoteAggregationViolationConsumer[StateT, VoteT],
) {
d.lock.Lock()
defer d.lock.Unlock()
d.consumers = append(d.consumers, consumer)
}
func (d *VoteAggregationViolationDistributor[
StateT,
VoteT,
]) OnDoubleVotingDetected(vote1, vote2 *VoteT) {
d.lock.RLock()
defer d.lock.RUnlock()
for _, subscriber := range d.consumers {
subscriber.OnDoubleVotingDetected(vote1, vote2)
}
}
func (d *VoteAggregationViolationDistributor[
StateT,
VoteT,
]) OnInvalidVoteDetected(err models.InvalidVoteError[VoteT]) {
d.lock.RLock()
defer d.lock.RUnlock()
for _, subscriber := range d.consumers {
subscriber.OnInvalidVoteDetected(err)
}
}
func (d *VoteAggregationViolationDistributor[
StateT,
VoteT,
]) OnVoteForInvalidStateDetected(
vote *VoteT,
invalidProposal *models.SignedProposal[StateT, VoteT],
) {
d.lock.RLock()
defer d.lock.RUnlock()
for _, subscriber := range d.consumers {
subscriber.OnVoteForInvalidStateDetected(vote, invalidProposal)
}
}