Challenged blocks monitoring

cmd/bots/internal/common/environment.go

@@ -724,6 +724,8 @@ func executeAlertTemplate(
 		msg, err = executeBaseTargetTemplate(alertJSON, nodesAliases, extension)
 	case entities.InternalErrorAlertType:
 		msg, err = executeInternalErrorTemplate(alertJSON, extension)
+	case entities.ChallengedBlockAlertType:
+		msg, err = executeChallengedBlockTemplate(alertJSON, extension)
 	case entities.L2StuckAlertType:
 		msg, err = executeL2StuckAlertTemplate(alertJSON, extension)
 	default:
@@ -749,6 +751,20 @@ func executeSimpleAlertTemplate(alertJSON []byte, extension ExpectedExtension) (
 	return msg, nil
 }
 
+func executeChallengedBlockTemplate(alertJSON []byte, extension ExpectedExtension) (string, error) {
+	var challengedBlockAlert entities.ChallengedBlockAlert
+	err := json.Unmarshal(alertJSON, &challengedBlockAlert)
+	if err != nil {
+		return "", err
+	}
+	_ = fmt.Stringer(challengedBlockAlert.BlockID) // to check if it implements Stringer
+	msg, err := executeTemplate("templates/alerts/challenged_block_alert", challengedBlockAlert, extension)
+	if err != nil {
+		return "", err
+	}
+	return msg, nil
+}
+
 func executeInternalErrorTemplate(alertJSON []byte, extension ExpectedExtension) (string, error) {
 	var internalErrorAlert entities.InternalErrorAlert
 	err := json.Unmarshal(alertJSON, &internalErrorAlert)

cmd/bots/internal/common/templates/alerts/challenged_block_alert.html

@@ -0,0 +1,4 @@
+🚨 The block <code>{{.BlockID}}</code> has been challenged. Found on the following nodes:
+{{range .Nodes}}
+- <code>{{.}}</code>
+{{end}}

cmd/bots/internal/common/templates/alerts/challenged_block_alert.md

@@ -0,0 +1,6 @@
+```yaml
+🚨 The block {{ .BlockID }} has been challenged. Found on the following nodes:
+{{range .Nodes}}
+- {{.}}
+{{end}}
+```

cmd/bots/internal/common/templates_internal_test.go

@@ -565,3 +565,18 @@ func TestL2StuckTemplate(t *testing.T) {
 		assert.Equal(t, expected, actual)
 	}
 }
+
+func TestChallengedBlockTemplate(t *testing.T) {
+	data := &entities.ChallengedBlockAlert{
+		Timestamp: 100,
+		BlockID:   proto.NewBlockIDFromDigest(crypto.Digest{1, 2, 3, 4, 5}),
+		Nodes:     []string{"node1", "node2", "node3"},
+	}
+	for _, f := range expectedFormats() {
+		const template = "templates/alerts/challenged_block_alert"
+		actual, err := executeTemplate(template, data, f)
+		require.NoError(t, err)
+		expected := goldenValue(t, template, f, actual)
+		assert.Equal(t, expected, actual)
+	}
+}

cmd/bots/internal/common/testdata/templates/alerts/challenged_block_alert.html.golden

@@ -0,0 +1,8 @@
+🚨 The block <code>4wBqpZLLi8G2oXgScS3ipLzLdrJ1MFu6ghJzHNV6ym1</code> has been challenged. Found on the following nodes:
+
+- <code>node1</code>
+
+- <code>node2</code>
+
+- <code>node3</code>
+

cmd/bots/internal/common/testdata/templates/alerts/challenged_block_alert.md.golden

@@ -0,0 +1,10 @@
+```yaml
+🚨 The block 4wBqpZLLi8G2oXgScS3ipLzLdrJ1MFu6ghJzHNV6ym1 has been challenged. Found on the following nodes:
+
+- node1
+
+- node2
+
+- node3
+
+```

pkg/analysis/alerts_storage.go

@@ -65,16 +65,17 @@ type alertConfirmationsValue struct {
 
 func newAlertConfirmations(customConfirmations ...alertConfirmationsValue) alertConfirmations {
 	confirmations := alertConfirmations{
-		entities.SimpleAlertType:        0,
-		entities.UnreachableAlertType:   0,
-		entities.IncompleteAlertType:    0,
-		entities.InvalidHeightAlertType: 0,
-		entities.HeightAlertType:        0,
-		entities.StateHashAlertType:     0,
-		entities.AlertFixedType:         0,
-		entities.BaseTargetAlertType:    0,
-		entities.InternalErrorAlertType: 0,
-		entities.L2StuckAlertType:       0,
+		entities.SimpleAlertType:          0,
+		entities.UnreachableAlertType:     0,
+		entities.IncompleteAlertType:      0,
+		entities.InvalidHeightAlertType:   0,
+		entities.HeightAlertType:          0,
+		entities.StateHashAlertType:       0,
+		entities.AlertFixedType:           0,
+		entities.BaseTargetAlertType:      0,
+		entities.InternalErrorAlertType:   0,
+		entities.ChallengedBlockAlertType: 0,
+		entities.L2StuckAlertType:         0,
 	}
 	for _, cc := range customConfirmations {
 		confirmations[cc.alertType] = cc.confirmations

pkg/analysis/analyzer.go

@@ -2,6 +2,7 @@ package analysis
 
 import (
 	"context"
+	"iter"
 	"sync"
 	"time"
 
@@ -22,6 +23,7 @@ type AnalyzerOptions struct {
 	HeightCriteriaOpts      *criteria.HeightCriterionOptions
 	StateHashCriteriaOpts   *criteria.StateHashCriterionOptions
 	BaseTargetCriterionOpts *criteria.BaseTargetCriterionOptions
+	ChallengeCriterionOpts  *criteria.ChallengedBlockCriterionOptions
 }
 
 type Analyzer struct {
@@ -116,6 +118,20 @@ func (a *Analyzer) analyze(alerts chan<- entities.Alert, pollingResult entities.
 	return nil
 }
 
+func joinSlicesSeq2[Slice ~[]E, E any](slices ...Slice) iter.Seq2[int, E] {
+	return func(yield func(int, E) bool) {
+		var i int
+		for _, slice := range slices {
+			for _, v := range slice {
+				if !yield(i, v) {
+					return
+				}
+				i++
+			}
+		}
+	}
+}
+
 func (a *Analyzer) criteriaRoutines(
 	statements entities.NodeStatements,
 	timestamp int64,
@@ -135,6 +151,12 @@ func (a *Analyzer) criteriaRoutines(
 			}
 			return nil
 		},
+		func(in chan<- entities.Alert) error {
+			criterion := criteria.NewChallengedBlockCriterion(a.opts.ChallengeCriterionOpts, a.zap)
+			statementsToAnalyze := joinSlicesSeq2(statusSplit[entities.Incomplete], statusSplit[entities.OK])
+			criterion.Analyze(in, timestamp, statementsToAnalyze)
+			return nil
+		},
 		func(in chan<- entities.Alert) error {
 			criterion := criteria.NewUnreachableCriterion(a.es, a.opts.UnreachableCriteriaOpts, a.zap)
 			return criterion.Analyze(in, timestamp, statusSplit[entities.Unreachable])

pkg/analysis/analyzer_test.go

@@ -66,7 +66,7 @@ func mkEvents(node string, startHeight uint64, shs ...shInfo) []entities.Event {
 		h := startHeight + uint64(i)
 		ts := mkTimestamp(h)
 		sh := shs[i].sh
-		r[i] = entities.NewStateHashEvent(node, ts, "V", h, &sh, 1, &sh.BlockID, nil)
+		r[i] = entities.NewStateHashEvent(node, ts, "V", h, &sh, 1, &sh.BlockID, nil, false)
 	}
 	return r
 }

pkg/analysis/criteria/base_target_test.go

@@ -24,7 +24,7 @@ func mkBaseTargetStatements(baseTargetInfo []baseTargetInfo) entities.NodeStatem
 	var statements entities.NodeStatements
 	for _, info := range baseTargetInfo {
 		statement := entities.NewStateHashEvent(info.node, info.ts, info.v,
-			info.h, nil, info.baseTarget, nil, nil).Statement()
+			info.h, nil, info.baseTarget, nil, nil, false).Statement()
 		statements = append(statements, statement)
 	}
 	return statements

pkg/analysis/criteria/challenged_block.go

@@ -0,0 +1,50 @@
+package criteria
+
+import (
+	"iter"
+
+	"github.com/wavesplatform/gowaves/pkg/proto"
+	"go.uber.org/zap"
+
+	"nodemon/pkg/entities"
+)
+
+type ChallengedBlockCriterionOptions struct {
+	// no options for now
+}
+
+type ChallengedBlockCriterion struct {
+	opts   *ChallengedBlockCriterionOptions
+	logger *zap.Logger
+}
+
+func NewChallengedBlockCriterion(opts *ChallengedBlockCriterionOptions, logger *zap.Logger) *ChallengedBlockCriterion {
+	if opts == nil { // default
+		opts = &ChallengedBlockCriterionOptions{}
+	}
+	return &ChallengedBlockCriterion{opts: opts, logger: logger}
+}
+
+type statementsSeq2 = iter.Seq2[int, entities.NodeStatement]
+
+func (c *ChallengedBlockCriterion) Analyze(alerts chan<- entities.Alert, timestamp int64, statements statementsSeq2) {
+	challengedBlocks := make(map[proto.BlockID]entities.Nodes)
+	for _, statement := range statements {
+		if statement.Challenged && statement.BlockID != nil {
+			blockID := *statement.BlockID
+			challengedBlocks[blockID] = append(challengedBlocks[blockID], statement.Node)
+		}
+	}
+	for blockID, nodes := range challengedBlocks {
+		sortedNodes := nodes.Sort()
+		c.logger.Info("ChallengedBlockCriterion: challenged block detected",
+			zap.Stringer("block ID", blockID),
+			zap.Strings("nodes", sortedNodes),
+		)
+		alerts <- &entities.ChallengedBlockAlert{
+			Timestamp: timestamp,
+			BlockID:   blockID,
+			Nodes:     sortedNodes,
+		}
+	}
+}

pkg/analysis/criteria/challenged_block_test.go

@@ -0,0 +1,103 @@
+package criteria_test
+
+import (
+	"encoding/binary"
+	"fmt"
+	"log"
+	"slices"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"github.com/wavesplatform/gowaves/pkg/crypto"
+	"github.com/wavesplatform/gowaves/pkg/proto"
+	"go.uber.org/zap"
+
+	"nodemon/pkg/analysis/criteria"
+	"nodemon/pkg/entities"
+)
+
+func mkBlockID(i int) *proto.BlockID {
+	var d crypto.Digest
+	binary.BigEndian.PutUint64(d[:8], uint64(i))
+	blockID := proto.NewBlockIDFromDigest(d)
+	return &blockID
+}
+
+func TestChallengedBlockCriterion_Analyze(t *testing.T) {
+	logger, err := zap.NewDevelopment()
+	if err != nil {
+		log.Fatalf("can't initialize zap logger: %v", err)
+	}
+	defer func(zap *zap.Logger) {
+		if syncErr := zap.Sync(); syncErr != nil {
+			log.Println(syncErr)
+		}
+	}(logger)
+
+	tests := []struct {
+		opts           *criteria.ChallengedBlockCriterionOptions
+		data           entities.NodeStatements
+		expectedAlerts []entities.ChallengedBlockAlert
+	}{
+		{
+			opts: &criteria.ChallengedBlockCriterionOptions{},
+			data: entities.NodeStatements{
+				{Node: "a", BlockID: mkBlockID(1), Challenged: true},
+				{Node: "b", BlockID: mkBlockID(1), Challenged: true},
+				{Node: "c", BlockID: mkBlockID(2), Challenged: true},
+				{Node: "d", BlockID: mkBlockID(0), Challenged: false},
+				{Node: "e", BlockID: nil, Challenged: true},
+			},
+			expectedAlerts: []entities.ChallengedBlockAlert{
+				{Nodes: entities.Nodes{"a", "b"}, BlockID: *mkBlockID(1)},
+				{Nodes: entities.Nodes{"c"}, BlockID: *mkBlockID(2)},
+			},
+		},
+		{
+			opts: nil,
+			data: entities.NodeStatements{
+				{Node: "a", BlockID: mkBlockID(1), Challenged: true},
+				{Node: "b", BlockID: mkBlockID(1), Challenged: true},
+				{Node: "c", BlockID: mkBlockID(1), Challenged: true},
+				{Node: "c", BlockID: mkBlockID(2), Challenged: true},
+				{Node: "d", BlockID: mkBlockID(2), Challenged: false},
+				{Node: "e", BlockID: nil, Challenged: true},
+			},
+			expectedAlerts: []entities.ChallengedBlockAlert{
+				{Nodes: entities.Nodes{"a", "b", "c"}, BlockID: *mkBlockID(1)},
+				{Nodes: entities.Nodes{"c"}, BlockID: *mkBlockID(2)},
+			},
+		},
+	}
+	for i := range tests {
+		test := tests[i]
+		t.Run(fmt.Sprintf("TestCase#%d", i+1), func(t *testing.T) {
+			done := make(chan struct{})
+			defer func() {
+				select {
+				case <-done:
+					return // test passed
+				case <-time.After(5 * time.Second):
+					require.Fail(t, "timeout exceeded")
+				}
+			}()
+
+			alerts := make(chan entities.Alert)
+			go func() {
+				defer close(done)
+				criterion := criteria.NewChallengedBlockCriterion(test.opts, logger)
+				criterion.Analyze(alerts, 0, slices.All(test.data))
+			}()
+			for j := range test.expectedAlerts {
+				select {
+				case actualAlert := <-alerts:
+					challengedBlockAlert := *actualAlert.(*entities.ChallengedBlockAlert)
+					require.Contains(t, test.expectedAlerts, challengedBlockAlert, "test case #%d", j+1)
+				case <-time.After(5 * time.Second):
+					require.Fail(t, "timeout exceeded")
+				}
+			}
+		})
+	}
+}

pkg/analysis/criteria/state_hash_test.go

@@ -66,7 +66,7 @@ func mkEvents(node string, startHeight uint64, shs ...shInfo) []entities.Event {
 		h := startHeight + uint64(i)
 		ts := mkTimestamp(h)
 		sh := shs[i].sh
-		r[i] = entities.NewStateHashEvent(node, ts, "V", h, &sh, 1, &sh.BlockID, nil)
+		r[i] = entities.NewStateHashEvent(node, ts, "V", h, &sh, 1, &sh.BlockID, nil, false)
 	}
 	return r
 }

pkg/analysis/finders/fork_finder_test.go

@@ -66,7 +66,7 @@ func mkEvents(node string, startHeight uint64, shs ...shInfo) []entities.Event {
 		h := startHeight + uint64(i)
 		ts := int64(100 + i*100)
 		sh := shs[i].sh
-		r[i] = entities.NewStateHashEvent(node, ts, "V", h, &sh, 1, &sh.BlockID, nil)
+		r[i] = entities.NewStateHashEvent(node, ts, "V", h, &sh, 1, &sh.BlockID, nil, false)
 	}
 	return r
 }

pkg/api/api.go

@@ -232,6 +232,7 @@ func (a *API) specificNodesHandler(w http.ResponseWriter, r *http.Request) {
 		zeroBT,
 		&statehash.BlockID,
 		nil,
+		false,
 	)
 	a.privateNodesEvents.Write(stateHashEvent)