WiP: Malfeasance v2: Handlers and Publishers

activation/e2e/atx_merge_test.go

@@ -276,6 +276,7 @@ func Test_MarryAndMerge(t *testing.T) {
 	require.NoError(t, err)
 
 	mFetch := smocks.NewMockFetcher(ctrl)
+	mMalPublish := activation.NewMockatxMalfeasancePublisher(ctrl)
 	mLegacyPublish := activation.NewMocklegacyMalfeasancePublisher(ctrl)
 	mBeacon := activation.NewMockatxReceiver(ctrl)
 	mTortoise := smocks.NewMockTortoise(ctrl)
@@ -290,6 +291,7 @@ func Test_MarryAndMerge(t *testing.T) {
 		mFetch,
 		goldenATX,
 		validator,
+		mMalPublish,
 		mLegacyPublish,
 		mBeacon,
 		mTortoise,

activation/e2e/builds_atx_v2_test.go

@@ -118,6 +118,7 @@ func TestBuilder_SwitchesToBuildV2(t *testing.T) {
 	edVerifier := signing.NewEdVerifier()
 	mpub := mocks.NewMockPublisher(ctrl)
 	mFetch := smocks.NewMockFetcher(ctrl)
+	mMalPublish := activation.NewMockatxMalfeasancePublisher(ctrl)
 	mLegacyPublish := activation.NewMocklegacyMalfeasancePublisher(ctrl)
 	mBeacon := activation.NewMockatxReceiver(ctrl)
 	mTortoise := smocks.NewMockTortoise(ctrl)
@@ -131,6 +132,7 @@ func TestBuilder_SwitchesToBuildV2(t *testing.T) {
 		mFetch,
 		goldenATX,
 		validator,
+		mMalPublish,
 		mLegacyPublish,
 		mBeacon,
 		mTortoise,

activation/e2e/checkpoint_merged_test.go

@@ -107,6 +107,7 @@ func Test_CheckpointAfterMerge(t *testing.T) {
 	require.NoError(t, err)
 
 	mFetch := smocks.NewMockFetcher(ctrl)
+	mMalPublish := activation.NewMockatxMalfeasancePublisher(ctrl)
 	mLegacyPublish := activation.NewMocklegacyMalfeasancePublisher(ctrl)
 	mBeacon := activation.NewMockatxReceiver(ctrl)
 	mTortoise := smocks.NewMockTortoise(ctrl)
@@ -120,6 +121,7 @@ func Test_CheckpointAfterMerge(t *testing.T) {
 		mFetch,
 		goldenATX,
 		validator,
+		mMalPublish,
 		mLegacyPublish,
 		mBeacon,
 		mTortoise,
@@ -299,6 +301,7 @@ func Test_CheckpointAfterMerge(t *testing.T) {
 		mFetch,
 		goldenATX,
 		validator,
+		mMalPublish,
 		mLegacyPublish,
 		mBeacon,
 		mTortoise,

activation/e2e/checkpoint_test.go

@@ -103,6 +103,7 @@ func TestCheckpoint_PublishingSoloATXs(t *testing.T) {
 	atxVersions := activation.AtxVersions{0: types.AtxV2}
 	edVerifier := signing.NewEdVerifier()
 	mFetch := smocks.NewMockFetcher(ctrl)
+	mMalPublish := activation.NewMockatxMalfeasancePublisher(ctrl)
 	mLegacyPublish := activation.NewMocklegacyMalfeasancePublisher(ctrl)
 	mBeacon := activation.NewMockatxReceiver(ctrl)
 	mTortoise := smocks.NewMockTortoise(ctrl)
@@ -116,6 +117,7 @@ func TestCheckpoint_PublishingSoloATXs(t *testing.T) {
 		mFetch,
 		goldenATX,
 		validator,
+		mMalPublish,
 		mLegacyPublish,
 		mBeacon,
 		mTortoise,
@@ -206,6 +208,7 @@ func TestCheckpoint_PublishingSoloATXs(t *testing.T) {
 		mFetch,
 		goldenATX,
 		validator,
+		mMalPublish,
 		mLegacyPublish,
 		mBeacon,
 		mTortoise,

activation/e2e/nipost_test.go

@@ -97,7 +97,7 @@ func launchPostSupervisor(
 	provingOpts := activation.DefaultPostProvingOpts()
 	provingOpts.RandomXMode = activation.PostRandomXModeLight
 
-	builder := activation.NewMockAtxBuilder(gomock.NewController(tb))
+	builder := activation.NewMockatxBuilder(gomock.NewController(tb))
 	builder.EXPECT().Register(gomock.Any())
 	ps := activation.NewPostSupervisor(log, postCfg, provingOpts, mgr, builder)
 	require.NoError(tb, ps.Start(cmdCfg, postOpts, sig))

activation/handler.go

@@ -20,8 +20,6 @@ import (
 	"github.com/spacemeshos/go-spacemesh/p2p"
 	"github.com/spacemeshos/go-spacemesh/p2p/pubsub"
 	"github.com/spacemeshos/go-spacemesh/signing"
-	"github.com/spacemeshos/go-spacemesh/sql"
-	"github.com/spacemeshos/go-spacemesh/sql/atxs"
 	"github.com/spacemeshos/go-spacemesh/system"
 )
 
@@ -103,6 +101,7 @@ func NewHandler(
 	fetcher system.Fetcher,
 	goldenATXID types.ATXID,
 	nipostValidator nipostValidator,
+	malPublisher atxMalfeasancePublisher,
 	legacyMalPublisher legacyMalfeasancePublisher,
 	beacon atxReceiver,
 	tortoise system.Tortoise,
@@ -144,7 +143,7 @@ func NewHandler(
 			fetcher:         fetcher,
 			beacon:          beacon,
 			tortoise:        tortoise,
-			malPublisher:    &MalfeasancePublisher{}, // TODO(mafa): pass real publisher when available
+			malPublisher:    malPublisher,
 		},
 	}
 
@@ -291,28 +290,3 @@ func (h *Handler) handleAtx(ctx context.Context, expHash types.Hash32, peer p2p.
 	h.inProgress.Forget(key)
 	return err
 }
-
-// Obtain the atxSignature of the given ATX.
-func atxSignature(ctx context.Context, db sql.Executor, id types.ATXID) (types.EdSignature, error) {
-	var blob sql.Blob
-	v, err := atxs.LoadBlob(ctx, db, id.Bytes(), &blob)
-	if err != nil {
-		return types.EmptyEdSignature, err
-	}
-
-	if len(blob.Bytes) == 0 {
-		// An empty blob indicates a golden ATX (after a checkpoint-recovery).
-		return types.EmptyEdSignature, fmt.Errorf("can't get signature for a golden (checkpointed) ATX: %s", id)
-	}
-
-	// TODO: implement for ATX V2
-	switch v {
-	case types.AtxV1:
-		var atx wire.ActivationTxV1
-		if err := codec.Decode(blob.Bytes, &atx); err != nil {
-			return types.EmptyEdSignature, fmt.Errorf("decoding atx v1: %w", err)
-		}
-		return atx.Signature, nil
-	}
-	return types.EmptyEdSignature, fmt.Errorf("unsupported ATX version: %v", v)
-}

activation/handler_test.go

@@ -215,8 +215,6 @@ func newTestHandler(tb testing.TB, goldenATXID types.ATXID, opts ...HandlerOptio
 	edVerifier := signing.NewEdVerifier()
 
 	mocks := newTestHandlerMocks(tb, goldenATXID)
-	// TODO(mafa): make mandatory parameter when real publisher is available
-	opts = append(opts, func(h *Handler) { h.v2.malPublisher = mocks.mMalPublish })
 	atxHdlr := NewHandler(
 		"localID",
 		cdb,
@@ -226,6 +224,7 @@ func newTestHandler(tb testing.TB, goldenATXID types.ATXID, opts ...HandlerOptio
 		mocks.mockFetch,
 		goldenATXID,
 		mocks.mValidator,
+		mocks.mMalPublish,
 		mocks.mLegacyMalPublish,
 		mocks.mBeacon,
 		mocks.mTortoise,

activation/handler_v1.go

@@ -438,7 +438,6 @@ func (h *HandlerV1) checkWrongPrevAtx(
 		return nil, err
 	}
 	if v != types.AtxV1 {
-		// TODO(mafa): update when V2 is introduced
 		return nil, fmt.Errorf("ATX %s with same prev ATX as %s is not version 1", atx2ID, atx.PrevATXID)
 	}
 
@@ -505,22 +504,19 @@ func (h *HandlerV1) storeAtx(ctx context.Context, atx *types.ActivationTx, watx
 	}
 
 	atxs.AtxAdded(h.cdb, atx)
-	if proof != nil {
-		if err := h.malPublisher.PublishProof(ctx, atx.SmesherID, proof); err != nil {
-			return fmt.Errorf("publishing malfeasance proof: %w", err)
-		}
-	}
-
-	added := h.cacheAtx(ctx, atx, malicious || proof != nil)
 	h.beacon.OnAtx(atx)
-	if added != nil {
+	if added := h.cacheAtx(ctx, atx, malicious || proof != nil); added != nil {
 		h.tortoise.OnAtx(atx.TargetEpoch(), atx.ID(), added)
 	}
-
 	h.logger.Debug("finished storing atx in epoch",
 		zap.Stringer("atx_id", atx.ID()),
 		zap.Uint32("epoch_id", atx.PublishEpoch.Uint32()),
 	)
+	if proof != nil {
+		if err := h.malPublisher.PublishProof(ctx, atx.SmesherID, proof); err != nil {
+			return fmt.Errorf("publishing malfeasance proof: %w", err)
+		}
+	}
 	return nil
 }
 
@@ -632,3 +628,28 @@ func collectAtxDeps(goldenAtxId types.ATXID, atx *wire.ActivationTxV1) (types.Ha
 
 	return types.BytesToHash(atx.NIPost.PostMetadata.Challenge), maps.Keys(filtered)
 }
+
+// Obtain the atxSignature of the given ATXv1.
+func atxSignature(ctx context.Context, db sql.Executor, id types.ATXID) (types.EdSignature, error) {
+	var blob sql.Blob
+	v, err := atxs.LoadBlob(ctx, db, id.Bytes(), &blob)
+	if err != nil {
+		return types.EmptyEdSignature, err
+	}
+
+	if len(blob.Bytes) == 0 {
+		// An empty blob indicates a golden ATX (after a checkpoint-recovery).
+		return types.EmptyEdSignature, fmt.Errorf("can't get signature for a golden (checkpointed) ATX: %s", id)
+	}
+
+	switch v {
+	case types.AtxV1:
+		var atx wire.ActivationTxV1
+		if err := codec.Decode(blob.Bytes, &atx); err != nil {
+			return types.EmptyEdSignature, fmt.Errorf("decoding atx v1: %w", err)
+		}
+		return atx.Signature, nil
+	default: // only needed for V1 ATXs
+		return types.EmptyEdSignature, fmt.Errorf("unsupported ATX version: %v", v)
+	}
+}

activation/handler_v2.go

@@ -495,8 +495,8 @@ func (n nipostSizes) sumUp() (units uint32, weight uint64, err error) {
 
 func (h *HandlerV2) verifyIncludedIDsUniqueness(atx *wire.ActivationTxV2) error {
 	seen := make(map[uint32]struct{})
-	for _, niposts := range atx.NIPosts {
-		for _, post := range niposts.Posts {
+	for _, niPosts := range atx.NIPosts {
+		for _, post := range niPosts.Posts {
 			if _, ok := seen[post.MarriageIndex]; ok {
 				return fmt.Errorf("ID present twice (duplicated marriage index): %d", post.MarriageIndex)
 			}
@@ -528,7 +528,7 @@ func (h *HandlerV2) syntacticallyValidateDeps(
 			return nil, fmt.Errorf("fetching previous atx: %w", err)
 		}
 		if prevAtx.PublishEpoch >= atx.PublishEpoch {
-			err := fmt.Errorf("previous atx is too new (%d >= %d) (%s) ", prevAtx.PublishEpoch, atx.PublishEpoch, prev)
+			err := fmt.Errorf("previous atx (%s) is too new (%d >= %d)", prev, prevAtx.PublishEpoch, atx.PublishEpoch)
 			return nil, err
 		}
 		previousAtxs[i] = prevAtx
@@ -541,9 +541,9 @@ func (h *HandlerV2) syntacticallyValidateDeps(
 
 	// validate previous ATXs
 	nipostSizes := make(nipostSizes, len(atx.NIPosts))
-	for i, niposts := range atx.NIPosts {
+	for i, niPosts := range atx.NIPosts {
 		nipostSizes[i] = new(nipostSize)
-		for _, post := range niposts.Posts {
+		for _, post := range niPosts.Posts {
 			if post.MarriageIndex >= uint32(len(equivocationSet)) {
 				err := fmt.Errorf("marriage index out of bounds: %d > %d", post.MarriageIndex, len(equivocationSet)-1)
 				return nil, err
@@ -563,11 +563,11 @@ func (h *HandlerV2) syntacticallyValidateDeps(
 	}
 
 	// validate poet membership proofs
-	for i, niposts := range atx.NIPosts {
+	for i, niPosts := range atx.NIPosts {
 		// verify PoET memberships in a single go
 		indexedChallenges := make(map[uint64][]byte)
 
-		for _, post := range niposts.Posts {
+		for _, post := range niPosts.Posts {
 			if _, ok := indexedChallenges[post.MembershipLeafIndex]; ok {
 				continue
 			}
@@ -591,10 +591,10 @@ func (h *HandlerV2) syntacticallyValidateDeps(
 		}
 
 		membership := types.MultiMerkleProof{
-			Nodes:       niposts.Membership.Nodes,
+			Nodes:       niPosts.Membership.Nodes,
 			LeafIndices: leafIndices,
 		}
-		leaves, err := h.nipostValidator.PoetMembership(ctx, &membership, niposts.Challenge, poetChallenges)
+		leaves, err := h.nipostValidator.PoetMembership(ctx, &membership, niPosts.Challenge, poetChallenges)
 		if err != nil {
 			return nil, fmt.Errorf("validating poet membership: %w", err)
 		}
@@ -606,7 +606,7 @@ func (h *HandlerV2) syntacticallyValidateDeps(
 		return nil, err
 	}
 
-	// validate all niposts
+	// validate all NIPoSTs
 	if atx.Initial != nil {
 		commitment := atx.Initial.CommitmentATX
 		nipostIdx := 0
@@ -625,8 +625,8 @@ func (h *HandlerV2) syntacticallyValidateDeps(
 	}
 
 	var smesherCommitment *types.ATXID
-	for idx, niposts := range atx.NIPosts {
-		for _, post := range niposts.Posts {
+	for idx, niPosts := range atx.NIPosts {
+		for _, post := range niPosts.Posts {
 			id := equivocationSet[post.MarriageIndex]
 			commitment, err := atxs.CommitmentATX(h.cdb, id)
 			if err != nil {
@@ -635,7 +635,7 @@ func (h *HandlerV2) syntacticallyValidateDeps(
 			if id == atx.SmesherID {
 				smesherCommitment = &commitment
 			}
-			if err := h.validatePost(ctx, id, atx, commitment, niposts.Challenge, post, idx); err != nil {
+			if err := h.validatePost(ctx, id, atx, commitment, niPosts.Challenge, post, idx); err != nil {
 				return nil, err
 			}
 			result.ids[id] = idData{
@@ -847,7 +847,7 @@ func (h *HandlerV2) checkDoubleMerge(ctx context.Context, tx sql.Transaction, at
 	if err != nil {
 		return true, fmt.Errorf("creating double merge proof: %w", err)
 	}
-	return true, h.malPublisher.Publish(ctx, atx.SmesherID, proof)
+	return true, h.malPublisher.Publish(ctx, atx.ActivationTxV2.SmesherID, proof)
 }
 
 func (h *HandlerV2) checkPrevAtx(ctx context.Context, tx sql.Transaction, atx *activationTx) (bool, error) {

activation/handler_v2_test.go

@@ -3,6 +3,7 @@ package activation
 import (
 	"context"
 	"errors"
+	"fmt"
 	"math"
 	"slices"
 	"testing"
@@ -996,7 +997,7 @@ func TestHandlerV2_ProcessMergedATX(t *testing.T) {
 			gomock.Any(),
 			merged.SmesherID,
 			gomock.AssignableToTypeOf(&wire.ProofDoubleMerge{}),
-		).DoAndReturn(func(ctx context.Context, id types.NodeID, proof wire.Proof) error {
+		).DoAndReturn(func(ctx context.Context, _ types.NodeID, proof wire.Proof) error {
 			malProof := proof.(*wire.ProofDoubleMerge)
 			nId, err := malProof.Valid(context.Background(), verifier)
 			require.NoError(t, err)
@@ -1504,7 +1505,7 @@ func TestHandlerV2_SyntacticallyValidateDeps(t *testing.T) {
 		atx.Sign(sig)
 
 		_, err := atxHandler.syntacticallyValidateDeps(context.Background(), atx)
-		require.ErrorContains(t, err, "previous atx is too new")
+		require.ErrorContains(t, err, fmt.Sprintf("previous atx (%s) is too new", prev.ID()))
 	})
 	t.Run("previous ATX by different smesher", func(t *testing.T) {
 		atxHandler := newV2TestHandler(t, golden)

activation/interface.go

@@ -109,6 +109,14 @@ type atxMalfeasancePublisher interface {
 	Publish(ctx context.Context, nodeID types.NodeID, proof wire.Proof) error
 }
 
+// malfeasancePublisher is an interface for publishing malfeasance proofs.
+//
+// Additionally the publisher will only gossip proofs when the node is in sync, otherwise it will only store them
+// and mark the associated identity as malfeasant.
+type malfeasancePublisher interface {
+	PublishATXProof(ctx context.Context, nodeID types.NodeID, proof []byte) error
+}
+
 type atxProvider interface {
 	GetAtx(id types.ATXID) (*types.ActivationTx, error)
 }
@@ -198,7 +206,7 @@ var (
 	ErrPostClientNotConnected = errors.New("post service not registered")
 )
 
-type AtxBuilder interface {
+type atxBuilder interface {
 	Register(sig *signing.EdSigner)
 }