evidence: new evidence event subscription (#5108)

CHANGELOG_PENDING.md

@@ -101,6 +101,7 @@ Friendly reminder, we have a [bug bounty program](https://hackerone.com/tendermi
 - [rpc] \#4979 Support EXISTS operator in `/tx_search` query (@melekes)
 - [rpc] \#5017 Add `/check_tx` endpoint to check transactions without executing them or adding them to the mempool (@melekes)
 - [statesync] Add state sync support, where a new node can be rapidly bootstrapped by fetching state snapshots from peers instead of replaying blocks. See the `[statesync]` config section.
+- [rpc] [\#5108](https://github.com/tendermint/tendermint/pull/5108) Subscribe using the websocket for new evidence events (@cmwaters)
 
 ### IMPROVEMENTS:
 

state/execution.go

@@ -481,6 +481,15 @@ func fireEvents(
 		ResultEndBlock:   *abciResponses.EndBlock,
 	})
 
+	if len(block.Evidence.Evidence) != 0 {
+		for _, ev := range block.Evidence.Evidence {
+			eventBus.PublishEventNewEvidence(types.EventDataNewEvidence{
+				Evidence: ev,
+				Height:   block.Height,
+			})
+		}
+	}
+
 	for i, tx := range block.Data.Txs {
 		eventBus.PublishEventTx(types.EventDataTx{TxResult: abci.TxResult{
 			Height: block.Height,

types/event_bus.go

@@ -156,6 +156,10 @@ func (b *EventBus) PublishEventNewBlockHeader(data EventDataNewBlockHeader) erro
 	return b.pubsub.PublishWithEvents(ctx, data, events)
 }
 
+func (b *EventBus) PublishEventNewEvidence(evidence EventDataNewEvidence) error {
+	return b.Publish(EventNewEvidence, evidence)
+}
+
 func (b *EventBus) PublishEventVote(data EventDataVote) error {
 	return b.Publish(EventVote, data)
 }
@@ -249,6 +253,10 @@ func (NopEventBus) PublishEventNewBlockHeader(data EventDataNewBlockHeader) erro
 	return nil
 }
 
+func (NopEventBus) PublishEventNewEvidence(evidence EventDataNewEvidence) error {
+	return nil
+}
+
 func (NopEventBus) PublishEventVote(data EventDataVote) error {
 	return nil
 }

types/event_bus_test.go

@@ -259,6 +259,40 @@ func TestEventBusPublishEventNewBlockHeader(t *testing.T) {
 	}
 }
 
+func TestEventBusPublishEventNewEvidence(t *testing.T) {
+	eventBus := NewEventBus()
+	err := eventBus.Start()
+	require.NoError(t, err)
+	defer eventBus.Stop()
+
+	ev := NewMockDuplicateVoteEvidence(1, time.Now(), "test-chain-id")
+
+	query := "tm.event='NewEvidence'"
+	evSub, err := eventBus.Subscribe(context.Background(), "test", tmquery.MustParse(query))
+	require.NoError(t, err)
+
+	done := make(chan struct{})
+	go func() {
+		msg := <-evSub.Out()
+		edt := msg.Data().(EventDataNewEvidence)
+		assert.Equal(t, ev, edt.Evidence)
+		assert.Equal(t, int64(4), edt.Height)
+		close(done)
+	}()
+
+	err = eventBus.PublishEventNewEvidence(EventDataNewEvidence{
+		Evidence: ev,
+		Height:   4,
+	})
+	assert.NoError(t, err)
+
+	select {
+	case <-done:
+	case <-time.After(1 * time.Second):
+		t.Fatal("did not receive a block header after 1 sec.")
+	}
+}
+
 func TestEventBusPublish(t *testing.T) {
 	eventBus := NewEventBus()
 	err := eventBus.Start()

types/events.go

@@ -18,6 +18,7 @@ const (
 	// All of this data can be fetched through the rpc.
 	EventNewBlock            = "NewBlock"
 	EventNewBlockHeader      = "NewBlockHeader"
+	EventNewEvidence         = "NewEvidence"
 	EventTx                  = "Tx"
 	EventValidatorSetUpdates = "ValidatorSetUpdates"
 
@@ -49,6 +50,7 @@ type TMEventData interface {
 func init() {
 	tmjson.RegisterType(EventDataNewBlock{}, "tendermint/event/NewBlock")
 	tmjson.RegisterType(EventDataNewBlockHeader{}, "tendermint/event/NewBlockHeader")
+	tmjson.RegisterType(EventDataNewEvidence{}, "tendermint/event/NewEvidence")
 	tmjson.RegisterType(EventDataTx{}, "tendermint/event/Tx")
 	tmjson.RegisterType(EventDataRoundState{}, "tendermint/event/RoundState")
 	tmjson.RegisterType(EventDataNewRound{}, "tendermint/event/NewRound")
@@ -76,6 +78,12 @@ type EventDataNewBlockHeader struct {
 	ResultEndBlock   abci.ResponseEndBlock   `json:"result_end_block"`
 }
 
+type EventDataNewEvidence struct {
+	Evidence Evidence `json:"evidence"`
+
+	Height int64 `json:"height"`
+}
+
 // All txs fire EventDataTx
 type EventDataTx struct {
 	abci.TxResult
@@ -139,6 +147,7 @@ var (
 	EventQueryLock                = QueryForEvent(EventLock)
 	EventQueryNewBlock            = QueryForEvent(EventNewBlock)
 	EventQueryNewBlockHeader      = QueryForEvent(EventNewBlockHeader)
+	EventQueryNewEvidence         = QueryForEvent(EventNewEvidence)
 	EventQueryNewRound            = QueryForEvent(EventNewRound)
 	EventQueryNewRoundStep        = QueryForEvent(EventNewRoundStep)
 	EventQueryPolka               = QueryForEvent(EventPolka)
@@ -164,6 +173,7 @@ func QueryForEvent(eventType string) tmpubsub.Query {
 type BlockEventPublisher interface {
 	PublishEventNewBlock(block EventDataNewBlock) error
 	PublishEventNewBlockHeader(header EventDataNewBlockHeader) error
+	PublishEventNewEvidence(evidence EventDataNewEvidence) error
 	PublishEventTx(EventDataTx) error
 	PublishEventValidatorSetUpdates(EventDataValidatorSetUpdates) error
 }

types/events_test.go

@@ -20,4 +20,8 @@ func TestQueryForEvent(t *testing.T) {
 		"tm.event='NewBlock'",
 		QueryForEvent(EventNewBlock).String(),
 	)
+	assert.Equal(t,
+		"tm.event='NewEvidence'",
+		QueryForEvent(EventNewEvidence).String(),
+	)
 }