From 47c191b5957be882afea5e5da86895e1ac558af0 Mon Sep 17 00:00:00 2001 From: Yacov Manevich Date: Thu, 2 Jul 2026 16:09:52 +0200 Subject: [PATCH 1/2] Move approval store into MSM This commit moves the approval store into the MSM, instead of being an external dependency. The reason is that the approval store needs to be initialized with the validator set of the *next* epoch and not the *current epoch*, which can only be computed inside the MSM. Signed-off-by: Yacov Manevich --- msm/approvals.go | 55 +++++++++++--- msm/approvals_test.go | 146 +++++++++++++++++++++++++++++++++++++ msm/build_decision.go | 13 ++-- msm/build_decision_test.go | 16 ++-- msm/fake_node_test.go | 40 +++------- msm/fuzz_test.go | 9 +-- msm/msm.go | 67 +++++++++++++---- msm/msm_test.go | 46 +++++------- msm/util_test.go | 1 - 9 files changed, 292 insertions(+), 101 deletions(-) diff --git a/msm/approvals.go b/msm/approvals.go index 5c6fe3cb..973be34c 100644 --- a/msm/approvals.go +++ b/msm/approvals.go @@ -5,6 +5,7 @@ package metadata import ( "fmt" + "sync" "github.com/ava-labs/simplex/common" "go.uber.org/zap" @@ -27,8 +28,12 @@ type ApprovalStore struct { validators NodeBLSMappings logger common.Logger pkByNodeID map[nodeID][]byte - approvalsByNodes map[nodeID]approvalsByPChainHeightAndAuxInfoDigest - storedCount int + // lock guards the mutable state below (approvalsByNodes, storedCount). The + // store is accessed concurrently: approvals are handled as they arrive while + // the block builder reads the accumulated approvals when building a block. + lock sync.RWMutex + approvalsByNodes map[nodeID]approvalsByPChainHeightAndAuxInfoDigest + storedCount int } func NewApprovalStore(signatureVerifier SignatureVerifier, validators NodeBLSMappings, logger common.Logger) *ApprovalStore { @@ -52,6 +57,9 @@ func NewApprovalStore(signatureVerifier SignatureVerifier, validators NodeBLSMap } func (as *ApprovalStore) Approvals() ValidatorSetApprovals { + as.lock.RLock() + defer as.lock.RUnlock() + approvals := make(ValidatorSetApprovals, 0, as.storedCount) for _, approvalsByHeight := range as.approvalsByNodes { for _, approval := range approvalsByHeight { @@ -70,17 +78,21 @@ func (as *ApprovalStore) HandleApproval(approval *ValidatorSetApproval, timestam return nil } - // Second thing we check is if we already have an approval for this height from this node. - if as.approvalExistsAndUpToDate(approval, timestamp) { - as.logger.Debug("Already have an approval from the node", zap.String("nodeID", + // Second thing we check is if the signature of the approval is valid. + // We need it to be valid in order for nodes to be able to aggregate it later on along with other approvals. + // This is checked before taking the lock, as it only reads immutable state. + if err := as.checkApprovalSignature(approval, pk); err != nil { + as.logger.Debug("Received an approval with an invalid signature", zap.String("nodeID", fmt.Sprintf("%x", approval.NodeID)), zap.Uint64("pChainHeight", approval.PChainHeight)) return nil } - // Third thing we check is if the signature of the approval is valid. - // We need it to be valid in order for nodes to be able to aggregate it later on along with other approvals. - if err := as.checkApprovalSignature(approval, pk); err != nil { - as.logger.Debug("Received an approval with an invalid signature", zap.String("nodeID", + as.lock.Lock() + defer as.lock.Unlock() + + // Third thing we check is if we already have an approval for this height from this node. + if as.approvalExistsAndUpToDate(approval, timestamp) { + as.logger.Debug("Already have an approval from the node", zap.String("nodeID", fmt.Sprintf("%x", approval.NodeID)), zap.Uint64("pChainHeight", approval.PChainHeight)) return nil } @@ -167,3 +179,28 @@ func (as *ApprovalStore) approvalExistsAndUpToDate(approval *ValidatorSetApprova return existingApproval.Timestamp >= timestamp } + +func (as *ApprovalStore) PutApprovals(approvalStore *ApprovalStore) { + // Snapshot the approvals under our lock, then hand them to the destination + // store (which takes its own lock). Copying first avoids holding two store + // locks at once. + as.lock.Lock() + type approvalWithTimestamp struct { + approval ValidatorSetApproval + timestamp uint64 + } + snapshot := make([]approvalWithTimestamp, 0, as.storedCount) + for _, approvalsByHeight := range as.approvalsByNodes { + for _, approval := range approvalsByHeight { + snapshot = append(snapshot, approvalWithTimestamp{ + approval: approval.ValidatorSetApproval, + timestamp: approval.Timestamp, + }) + } + } + as.lock.Unlock() + + for _, a := range snapshot { + approvalStore.HandleApproval(&a.approval, a.timestamp) + } +} diff --git a/msm/approvals_test.go b/msm/approvals_test.go index c24e5f37..865bf6c7 100644 --- a/msm/approvals_test.go +++ b/msm/approvals_test.go @@ -222,6 +222,152 @@ func TestApprovalStoreHandleApproval(t *testing.T) { } } +func TestApprovalStorePutApprovals(t *testing.T) { + // findApproval locates the approval for a given (NodeID, PChainHeight) in a slice returned by + // Approvals(). It is used to assert which of two competing approvals (source vs. destination) + // survived a merge, since Approvals() does not carry the timestamp but does carry the Signature, + // and signApproval produces a distinct signature per call. + findApproval := func(got ValidatorSetApprovals, node nodeID, height uint64) (ValidatorSetApproval, bool) { + for _, a := range got { + if a.NodeID == node && a.PChainHeight == height { + return a, true + } + } + return ValidatorSetApproval{}, false + } + + for _, tc := range []struct { + name string + // srcValidators/dstValidators size the two stores via makeValidators; NodeIDs are makeNodeID(i+1). + srcValidators int + dstValidators int + // srcApprovals are loaded into the source store; dstApprovals are pre-loaded into the destination + // store before PutApprovals is called. + srcApprovals []approvalAndTimestamp + dstApprovals []approvalAndTimestamp + // verify asserts on the destination (and source) store after src.PutApprovals(dst). + verify func(t *testing.T, dst, src *ApprovalStore, srcSent, dstSent []approvalAndTimestamp) + }{ + { + // Copies every source approval into an empty destination that shares the same validator set, + // and leaves the source store untouched (PutApprovals copies, it does not move). + name: "copies all approvals into empty destination", + srcValidators: 3, + dstValidators: 3, + srcApprovals: []approvalAndTimestamp{ + {ValidatorSetApproval{NodeID: makeNodeID(1), PChainHeight: 7, Signature: signApproval(7, [32]byte{})}, 100}, + {ValidatorSetApproval{NodeID: makeNodeID(2), PChainHeight: 8, Signature: signApproval(8, [32]byte{})}, 100}, + }, + verify: func(t *testing.T, dst, src *ApprovalStore, srcSent, _ []approvalAndTimestamp) { + got := dst.Approvals() + require.Len(t, got, 2) + require.Equal(t, 2, dst.storedCount) + require.ElementsMatch(t, []ValidatorSetApproval{srcSent[0].ValidatorSetApproval, srcSent[1].ValidatorSetApproval}, got) + // The source store is unchanged. + require.Len(t, src.Approvals(), 2) + require.Equal(t, 2, src.storedCount) + }, + }, + { + // Approvals from nodes absent from the destination's (smaller) validator set are dropped on + // carry-over. This is the epoch-change case: a validator that is not part of the new set does + // not have its approval carried into the new store. + name: "drops approvals from nodes not in destination validator set", + srcValidators: 3, + dstValidators: 2, + srcApprovals: []approvalAndTimestamp{ + {ValidatorSetApproval{NodeID: makeNodeID(1), PChainHeight: 1, Signature: signApproval(1, [32]byte{})}, 10}, + {ValidatorSetApproval{NodeID: makeNodeID(2), PChainHeight: 1, Signature: signApproval(1, [32]byte{})}, 10}, + {ValidatorSetApproval{NodeID: makeNodeID(3), PChainHeight: 1, Signature: signApproval(1, [32]byte{})}, 10}, + }, + verify: func(t *testing.T, dst, _ *ApprovalStore, _, _ []approvalAndTimestamp) { + got := dst.Approvals() + require.Len(t, got, 2, "only approvals from nodes in the destination set carry over") + require.Equal(t, 2, dst.storedCount) + _, ok := findApproval(got, makeNodeID(3), 1) + require.False(t, ok, "node 3 is not in the destination validator set") + }, + }, + { + // Merges into a non-empty destination: a pre-existing destination approval with a newer + // timestamp is kept over an older one carried from the source, while a brand-new node's + // approval from the source is added. + name: "keeps newer destination approval and adds new node", + srcValidators: 2, + dstValidators: 2, + dstApprovals: []approvalAndTimestamp{ + {ValidatorSetApproval{NodeID: makeNodeID(1), PChainHeight: 7, Signature: signApproval(7, [32]byte{})}, 200}, // newer, already present + }, + srcApprovals: []approvalAndTimestamp{ + {ValidatorSetApproval{NodeID: makeNodeID(1), PChainHeight: 7, Signature: signApproval(7, [32]byte{})}, 100}, // older, must not overwrite + {ValidatorSetApproval{NodeID: makeNodeID(2), PChainHeight: 7, Signature: signApproval(7, [32]byte{})}, 100}, // new node, added + }, + verify: func(t *testing.T, dst, _ *ApprovalStore, _, dstSent []approvalAndTimestamp) { + got := dst.Approvals() + require.Len(t, got, 2) + require.Equal(t, 2, dst.storedCount) + kept, ok := findApproval(got, makeNodeID(1), 7) + require.True(t, ok) + require.Equal(t, dstSent[0].ValidatorSetApproval, kept, "the newer destination approval is retained") + _, ok = findApproval(got, makeNodeID(2), 7) + require.True(t, ok, "the new node's approval is carried over") + }, + }, + { + // A source approval with a newer timestamp replaces the stale destination approval at the + // same (NodeID, PChainHeight). + name: "newer source approval replaces stale destination approval", + srcValidators: 2, + dstValidators: 2, + dstApprovals: []approvalAndTimestamp{ + {ValidatorSetApproval{NodeID: makeNodeID(1), PChainHeight: 7, Signature: signApproval(7, [32]byte{})}, 100}, // older, present + }, + srcApprovals: []approvalAndTimestamp{ + {ValidatorSetApproval{NodeID: makeNodeID(1), PChainHeight: 7, Signature: signApproval(7, [32]byte{})}, 200}, // newer, replaces + }, + verify: func(t *testing.T, dst, _ *ApprovalStore, srcSent, _ []approvalAndTimestamp) { + got := dst.Approvals() + require.Len(t, got, 1) + require.Equal(t, 1, dst.storedCount) + require.Equal(t, srcSent[0].ValidatorSetApproval, got[0], "the newer source approval replaces the stale one") + }, + }, + { + // An empty source store is a no-op: the destination is left exactly as it was. + name: "empty source is a no-op", + srcValidators: 2, + dstValidators: 2, + dstApprovals: []approvalAndTimestamp{ + {ValidatorSetApproval{NodeID: makeNodeID(1), PChainHeight: 1, Signature: signApproval(1, [32]byte{})}, 10}, + }, + verify: func(t *testing.T, dst, _ *ApprovalStore, _, dstSent []approvalAndTimestamp) { + got := dst.Approvals() + require.Len(t, got, 1) + require.Equal(t, 1, dst.storedCount) + require.Equal(t, dstSent[0].ValidatorSetApproval, got[0]) + }, + }, + } { + t.Run(tc.name, func(t *testing.T) { + src := NewApprovalStore(&signatureVerifier{}, makeValidators(tc.srcValidators), testutil.MakeLogger(t)) + dst := NewApprovalStore(&signatureVerifier{}, makeValidators(tc.dstValidators), testutil.MakeLogger(t)) + + for _, a := range tc.srcApprovals { + require.NoError(t, src.HandleApproval(&a.ValidatorSetApproval, a.Timestamp)) + } + for _, a := range tc.dstApprovals { + require.NoError(t, dst.HandleApproval(&a.ValidatorSetApproval, a.Timestamp)) + } + + src.PutApprovals(dst) + + // storedCount must always stay in sync with the number of retrievable approvals. + require.Len(t, dst.Approvals(), dst.storedCount) + tc.verify(t, dst, src, tc.srcApprovals, tc.dstApprovals) + }) + } +} + func TestApprovalStoreHandleApprovalStoredCountStaysConsistent(t *testing.T) { // Runs a mixed workload (insert, duplicate, replace, new height, prune) // and asserts that storedCount equals len(Approvals()) after every step. diff --git a/msm/build_decision.go b/msm/build_decision.go index cfac2d98..252be698 100644 --- a/msm/build_decision.go +++ b/msm/build_decision.go @@ -18,6 +18,7 @@ type blockBuildingDecision struct { buildInnerBlock bool transitionEpoch bool pChainHeight uint64 + validatorSet NodeBLSMappings } // PChainProgressListener listens for changes in the P-chain height. @@ -33,7 +34,7 @@ type blockBuildingDecider struct { waitForPendingBlock func(ctx context.Context) // hasValidatorSetChanged should return whether the validator set has changed since the // P-chain height referenced by the last block in the chain and until the provided P-chain height. - hasValidatorSetChanged func(pChainHeight uint64) (bool, error) + hasValidatorSetChanged func(pChainHeight uint64) (bool, NodeBLSMappings, error) getPChainHeight func() uint64 } @@ -48,14 +49,14 @@ func (bbd *blockBuildingDecider) shouldBuildBlock( for { pChainHeight := bbd.getPChainHeight() - shouldTransitionEpoch, err := bbd.hasValidatorSetChanged(pChainHeight) + shouldTransitionEpoch, newValidatorSet, err := bbd.hasValidatorSetChanged(pChainHeight) if err != nil { return blockBuildingDecision{}, err } if shouldTransitionEpoch { // If we should transition to a new epoch, maybe we can also build a block along the way. - return bbd.buildBlockWithEpochTransition(ctx, pChainHeight) + return bbd.buildBlockWithEpochTransition(ctx, pChainHeight, newValidatorSet) } // Else, we don't need to transition to a new epoch, but maybe we should build a block. @@ -109,7 +110,7 @@ func (bbd *blockBuildingDecider) waitForPChainChangeOrPendingBlock(ctx context.C // It waits up to a limited amount of time (bbd.maxBlockBuildingWaitTime) for a block to be ready to be built, // and if no block is ready by then, it returns the decision to transition epoch without building a block. // Otherwise, it returns the decision to build a block and transition epoch along the way. -func (bbd *blockBuildingDecider) buildBlockWithEpochTransition(ctx context.Context, pChainHeight uint64) (blockBuildingDecision, error) { +func (bbd *blockBuildingDecider) buildBlockWithEpochTransition(ctx context.Context, pChainHeight uint64, validatorSet NodeBLSMappings) (blockBuildingDecision, error) { impatientContext, cancel := context.WithTimeout(ctx, bbd.maxBlockBuildingWaitTime) defer cancel() @@ -124,9 +125,9 @@ func (bbd *blockBuildingDecider) buildBlockWithEpochTransition(ctx context.Conte if impatientContext.Err() != nil { // We have returned from waitForPendingBlock because impatientContext has timed out, // which means we don't need to build a block. - return blockBuildingDecision{transitionEpoch: true, pChainHeight: pChainHeight}, nil + return blockBuildingDecision{transitionEpoch: true, pChainHeight: pChainHeight, validatorSet: validatorSet}, nil } // Block is ready to be built - return blockBuildingDecision{buildInnerBlock: true, transitionEpoch: true, pChainHeight: pChainHeight}, nil + return blockBuildingDecision{buildInnerBlock: true, transitionEpoch: true, pChainHeight: pChainHeight, validatorSet: validatorSet}, nil } diff --git a/msm/build_decision_test.go b/msm/build_decision_test.go index b1b13402..6b34b067 100644 --- a/msm/build_decision_test.go +++ b/msm/build_decision_test.go @@ -31,7 +31,7 @@ func TestShouldBuildBlock_VMSignalsBlock(t *testing.T) { }, }, waitForPendingBlock: func(ctx context.Context) {}, - hasValidatorSetChanged: func(uint64) (bool, error) { return false, nil }, + hasValidatorSetChanged: func(uint64) (bool, NodeBLSMappings, error) { return false, nil, nil }, getPChainHeight: func() uint64 { return 100 }, } @@ -55,7 +55,7 @@ func TestShouldBuildBlock_ContextCanceled(t *testing.T) { cancel() <-ctx.Done() }, - hasValidatorSetChanged: func(uint64) (bool, error) { return false, nil }, + hasValidatorSetChanged: func(uint64) (bool, NodeBLSMappings, error) { return false, nil, nil }, getPChainHeight: func() uint64 { return 100 }, } @@ -85,8 +85,8 @@ func TestShouldBuildBlock_PChainHeightChangeTriggersEpochTransition(t *testing.T waitForPendingBlock: func(ctx context.Context) { <-ctx.Done() }, - hasValidatorSetChanged: func(height uint64) (bool, error) { - return height == 200, nil + hasValidatorSetChanged: func(height uint64) (bool, NodeBLSMappings, error) { + return height == 200, nil, nil }, getPChainHeight: func() uint64 { return pChainHeight.Load() }, } @@ -122,7 +122,7 @@ func TestShouldBuildBlock_PChainHeightChangeButNoEpochTransition(t *testing.T) { <-ctx.Done() } }, - hasValidatorSetChanged: func(uint64) (bool, error) { return false, nil }, + hasValidatorSetChanged: func(uint64) (bool, NodeBLSMappings, error) { return false, nil, nil }, getPChainHeight: func() uint64 { return pChainHeight.Load() }, } @@ -141,7 +141,7 @@ func TestShouldBuildBlock_EpochTransitionWithVMBlock(t *testing.T) { }, }, waitForPendingBlock: func(ctx context.Context) {}, - hasValidatorSetChanged: func(uint64) (bool, error) { return true, nil }, + hasValidatorSetChanged: func(uint64) (bool, NodeBLSMappings, error) { return true, nil, nil }, getPChainHeight: func() uint64 { return 100 }, } @@ -162,7 +162,7 @@ func TestShouldBuildBlock_EpochTransitionWithoutVMBlock(t *testing.T) { waitForPendingBlock: func(ctx context.Context) { <-ctx.Done() }, - hasValidatorSetChanged: func(uint64) (bool, error) { return true, nil }, + hasValidatorSetChanged: func(uint64) (bool, NodeBLSMappings, error) { return true, nil, nil }, getPChainHeight: func() uint64 { return 100 }, } @@ -187,7 +187,7 @@ func TestShouldBuildBlock_EpochTransitionContextCanceled(t *testing.T) { cancel() <-ctx.Done() }, - hasValidatorSetChanged: func(uint64) (bool, error) { return true, nil }, + hasValidatorSetChanged: func(uint64) (bool, NodeBLSMappings, error) { return true, nil, nil }, getPChainHeight: func() uint64 { return 100 }, } diff --git a/msm/fake_node_test.go b/msm/fake_node_test.go index 9e8c69b9..be6fc29d 100644 --- a/msm/fake_node_test.go +++ b/msm/fake_node_test.go @@ -60,13 +60,9 @@ func TestFakeNodeEpochChangesDespiteEmptyMempool(t *testing.T) { node.tryFinalizeNextBlock() } if flipCoin() { - node.sm.ApprovalsRetriever = &approvalsRetriever{ - result: []ValidatorSetApproval{{NodeID: [20]byte{1}, PChainHeight: 200, Signature: signApproval(200, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}}, - } + require.NoError(t, node.sm.HandleApproval(&ValidatorSetApproval{NodeID: [20]byte{1}, PChainHeight: 200, Signature: signApproval(200, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}, 1)) } else { - node.sm.ApprovalsRetriever = &approvalsRetriever{ - result: []ValidatorSetApproval{{NodeID: [20]byte{2}, PChainHeight: 200, Signature: signApproval(200, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}}, - } + require.NoError(t, node.sm.HandleApproval(&ValidatorSetApproval{NodeID: [20]byte{2}, PChainHeight: 200, Signature: signApproval(200, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}, 1)) } if node.isLastBlockSealing() { @@ -111,13 +107,9 @@ func TestFakeNode(t *testing.T) { for node.Epoch() == epoch { node.act() if flipCoin() { - node.sm.ApprovalsRetriever = &approvalsRetriever{ - result: []ValidatorSetApproval{{NodeID: [20]byte{1}, PChainHeight: 200, Signature: signApproval(200, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}}, - } + require.NoError(t, node.sm.HandleApproval(&ValidatorSetApproval{NodeID: [20]byte{1}, PChainHeight: 200, Signature: signApproval(200, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}, 1)) } else { - node.sm.ApprovalsRetriever = &approvalsRetriever{ - result: []ValidatorSetApproval{{NodeID: [20]byte{2}, PChainHeight: 200, Signature: signApproval(200, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}}, - } + require.NoError(t, node.sm.HandleApproval(&ValidatorSetApproval{NodeID: [20]byte{2}, PChainHeight: 200, Signature: signApproval(200, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}, 1)) } } @@ -132,13 +124,9 @@ func TestFakeNode(t *testing.T) { for node.Epoch() == epoch { node.act() if flipCoin() { - node.sm.ApprovalsRetriever = &approvalsRetriever{ - result: []ValidatorSetApproval{{NodeID: [20]byte{2}, PChainHeight: 300, Signature: signApproval(300, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}}, - } + require.NoError(t, node.sm.HandleApproval(&ValidatorSetApproval{NodeID: [20]byte{2}, PChainHeight: 300, Signature: signApproval(300, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}, 1)) } else { - node.sm.ApprovalsRetriever = &approvalsRetriever{ - result: []ValidatorSetApproval{{NodeID: [20]byte{3}, PChainHeight: 300, Signature: signApproval(300, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}}, - } + require.NoError(t, node.sm.HandleApproval(&ValidatorSetApproval{NodeID: [20]byte{3}, PChainHeight: 300, Signature: signApproval(300, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}, 1)) } } @@ -185,13 +173,9 @@ func TestFakeNodeEmptyMempool(t *testing.T) { for node.lastFinalizedBlock().Metadata.SimplexEpochInfo.BlockValidationDescriptor == nil { node.act() if flipCoin() { - node.sm.ApprovalsRetriever = &approvalsRetriever{ - result: []ValidatorSetApproval{{NodeID: [20]byte{1}, PChainHeight: 200, Signature: signApproval(200, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}}, - } + require.NoError(t, node.sm.HandleApproval(&ValidatorSetApproval{NodeID: [20]byte{1}, PChainHeight: 200, Signature: signApproval(200, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}, 1)) } else { - node.sm.ApprovalsRetriever = &approvalsRetriever{ - result: []ValidatorSetApproval{{NodeID: [20]byte{2}, PChainHeight: 200, Signature: signApproval(200, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}}, - } + require.NoError(t, node.sm.HandleApproval(&ValidatorSetApproval{NodeID: [20]byte{2}, PChainHeight: 200, Signature: signApproval(200, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}, 1)) } } @@ -218,13 +202,9 @@ func TestFakeNodeEmptyMempool(t *testing.T) { for node.Height() < 30 { node.act() if flipCoin() { - node.sm.ApprovalsRetriever = &approvalsRetriever{ - result: []ValidatorSetApproval{{NodeID: [20]byte{2}, PChainHeight: 300, Signature: signApproval(300, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}}, - } + require.NoError(t, node.sm.HandleApproval(&ValidatorSetApproval{NodeID: [20]byte{2}, PChainHeight: 300, Signature: signApproval(300, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}, 1)) } else { - node.sm.ApprovalsRetriever = &approvalsRetriever{ - result: []ValidatorSetApproval{{NodeID: [20]byte{3}, PChainHeight: 300, Signature: signApproval(300, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}}, - } + require.NoError(t, node.sm.HandleApproval(&ValidatorSetApproval{NodeID: [20]byte{3}, PChainHeight: 300, Signature: signApproval(300, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}, 1)) } } diff --git a/msm/fuzz_test.go b/msm/fuzz_test.go index 12a4e489..de4dc402 100644 --- a/msm/fuzz_test.go +++ b/msm/fuzz_test.go @@ -203,9 +203,6 @@ func buildEpochChain(tb testing.TB, logger common.Logger) ([]*StateMachineBlock, sm.LastNonSimplexInnerBlock = genesis.InnerBlock tc.blockStore[0] = &outerBlock{block: genesis} - var approvalsResult ValidatorSetApprovals - sm.ApprovalsRetriever = &dynamicApprovalsRetriever{approvals: &approvalsResult} - ctx := context.Background() addBlock := func(seq uint64, b *StateMachineBlock, fin *common.Finalization) { @@ -260,20 +257,20 @@ func buildEpochChain(tb testing.TB, logger common.Logger) ([]*StateMachineBlock, auxInfoDigest := sha256.Sum256(nil) // block4 & block5: collecting-approvals blocks (1/3 then 2/3, not enough to seal). - approvalsResult = ValidatorSetApprovals{{NodeID: node1, PChainHeight: pChainHeight2, AuxInfoDigest: auxInfoDigest, Signature: signApproval(pChainHeight2, auxInfoDigest)}} + require.NoError(tb, sm.HandleApproval(&ValidatorSetApproval{NodeID: node1, PChainHeight: pChainHeight2, AuxInfoDigest: auxInfoDigest, Signature: signApproval(pChainHeight2, auxInfoDigest)}, 1)) currentTime = startTime.Add(time.Second + 4*time.Millisecond) tc.blockBuilder.block = nextInner(4) block4 := build(4, 3, 1, block3) addBlock(4, block4, nil) - approvalsResult = ValidatorSetApprovals{{NodeID: node2, PChainHeight: pChainHeight2, AuxInfoDigest: auxInfoDigest, Signature: signApproval(pChainHeight2, auxInfoDigest)}} + require.NoError(tb, sm.HandleApproval(&ValidatorSetApproval{NodeID: node2, PChainHeight: pChainHeight2, AuxInfoDigest: auxInfoDigest, Signature: signApproval(pChainHeight2, auxInfoDigest)}, 2)) currentTime = startTime.Add(time.Second + 5*time.Millisecond) tc.blockBuilder.block = nextInner(5) block5 := build(5, 4, 1, block4) addBlock(5, block5, nil) // block6: the sealing block (3/3 approvals). Its successor is in stateBuildBlockEpochSealed. - approvalsResult = ValidatorSetApprovals{{NodeID: node3, PChainHeight: pChainHeight2, AuxInfoDigest: auxInfoDigest, Signature: signApproval(pChainHeight2, auxInfoDigest)}} + require.NoError(tb, sm.HandleApproval(&ValidatorSetApproval{NodeID: node3, PChainHeight: pChainHeight2, AuxInfoDigest: auxInfoDigest, Signature: signApproval(pChainHeight2, auxInfoDigest)}, 3)) currentTime = startTime.Add(time.Second + 6*time.Millisecond) tc.blockBuilder.block = nextInner(6) block6 := build(6, 5, 1, block5) diff --git a/msm/msm.go b/msm/msm.go index bb7797a2..f86a446d 100644 --- a/msm/msm.go +++ b/msm/msm.go @@ -12,6 +12,7 @@ import ( "fmt" "math" "slices" + "sync" "time" "github.com/ava-labs/simplex/common" @@ -131,11 +132,6 @@ type ICMEpochInput struct { // ICMEpochTransition computes the next ICM epoch given the current upgrade configuration and epoch input. type ICMEpochTransition func(ICMEpochInput) ICMEpochInfo -// ApprovalsRetriever retrieves the approvals from validators of the next epoch for the epoch change. -type ApprovalsRetriever interface { - Approvals() ValidatorSetApprovals -} - // KeyAggregator combines multiple public keys into a single aggregated public key. type KeyAggregator interface { AggregateKeys(keys ...[]byte) ([]byte, error) @@ -191,6 +187,9 @@ type AuxiliaryInfoGenVerifier interface { // StateMachine manages block building and verification across epoch transitions. type StateMachine struct { *Config + lock sync.RWMutex + approvalStore *ApprovalStore + approvalStoreValidatorSet NodeBLSMappings } // Config contains the dependencies and configuration parameters needed to initialize the StateMachine. @@ -216,8 +215,6 @@ type Config struct { GetValidatorSet ValidatorSetRetriever // GetBlock retrieves a previously built or finalized block. GetBlock BlockRetriever - // ApprovalsRetriever retrieves validator approvals for epoch transitions. - ApprovalsRetriever ApprovalsRetriever // SignatureAggregatorCreator creates a new SignatureAggregator for aggregating validator signatures for epoch transitions. SignatureAggregatorCreator common.SignatureAggregatorCreator // KeyAggregator aggregates public keys from validators. @@ -264,6 +261,40 @@ func NewStateMachine(config *Config) (*StateMachine, error) { return &sm, nil } +func (sm *StateMachine) HandleApproval(approval *ValidatorSetApproval, timestamp uint64) error { + sm.lock.Lock() + approvalStore := sm.approvalStore + sm.lock.Unlock() + + if approvalStore == nil { + sm.Logger.Debug("Approval store is not initialized, ignoring approval", + zap.String("nodeID", fmt.Sprintf("%x", approval.NodeID)), + zap.Uint64("pChainHeight", approval.PChainHeight)) + return nil + } + + return approvalStore.HandleApproval(approval, timestamp) +} + +func (sm *StateMachine) maybeInitializeApprovalStore(validatorSet NodeBLSMappings) error { + sm.lock.Lock() + defer sm.lock.Unlock() + + // If the approval store is not initialized or the validator set has changed, create a new approval store. + if sm.approvalStore == nil || !validatorSet.Equal(sm.approvalStoreValidatorSet) { + // We first save the old approval store to copy over any existing approvals to the new approval store. + oldApprovalStore := sm.approvalStore + sm.approvalStore = NewApprovalStore(sm.SignatureVerifier, validatorSet, sm.Logger) + sm.approvalStoreValidatorSet = validatorSet + if oldApprovalStore != nil { + oldApprovalStore.PutApprovals(sm.approvalStore) + } + return nil + } + + return nil +} + // BuildBlock constructs the next block on top of the given parent block, and passes in the provided simplex metadata and blacklist. func (sm *StateMachine) BuildBlock(ctx context.Context, metadata common.ProtocolMetadata, blacklist *common.Blacklist) (*StateMachineBlock, error) { // The zero sequence number is reserved for the genesis block, which should never be built. @@ -468,6 +499,7 @@ func (sm *StateMachine) buildBlockOrTransitionEpoch(ctx context.Context, parentB if decisionToBuildBlock.transitionEpoch && isSealingBlockFinalized { sm.Logger.Debug("Transitioning epoch after building block", zap.Uint64("newPChainRefHeight", decisionToBuildBlock.pChainHeight)) newSimplexEpochInfo.NextPChainReferenceHeight = decisionToBuildBlock.pChainHeight + sm.maybeInitializeApprovalStore(decisionToBuildBlock.validatorSet) } now := sm.GetTime() @@ -618,6 +650,12 @@ func (sm *StateMachine) verifyNextPChainRefHeightNormal(prevMD StateMachineMetad errValidatorSetUnchanged, next.NextPChainReferenceHeight, prev.PChainReferenceHeight) } + // Else, ! currentValidatorSet.Equal(newValidatorSet) || next.NextPChainReferenceHeight == 0 + // so if next.NextPChainReferenceHeight > 0, we should initialize the approval store for the new validator set. + if next.NextPChainReferenceHeight > 0 { + sm.maybeInitializeApprovalStore(newValidatorSet) + } + // Else, either the validator set has changed, or the next P-chain reference height is still 0. // Both of these cases are fine. @@ -666,6 +704,8 @@ func (sm *StateMachine) verifyNextPChainRefHeightForNewEpoch(expectedEpochInfo S errValidatorSetUnchanged, next.NextPChainReferenceHeight, expectedEpochInfo.PChainReferenceHeight) } + sm.maybeInitializeApprovalStore(newValidatorSet) + return nil } @@ -676,7 +716,7 @@ func (sm *StateMachine) createBlockBuildingDecider(pChainReferenceHeight uint64) pChainListener: sm.PChainProgressListener, getPChainHeight: sm.GetPChainHeightForProposing, waitForPendingBlock: sm.BlockBuilder.WaitForPendingBlock, - hasValidatorSetChanged: func(pChainHeight uint64) (bool, error) { + hasValidatorSetChanged: func(pChainHeight uint64) (bool, NodeBLSMappings, error) { // The given pChainHeight was sampled by the caller of shouldTransitionEpoch(). // We compare between the current validator set, defined by the P-chain reference height in the parent block, // and the new validator set defined by the given pChainHeight. @@ -684,12 +724,12 @@ func (sm *StateMachine) createBlockBuildingDecider(pChainReferenceHeight uint64) currentValidatorSet, err := sm.GetValidatorSet(pChainReferenceHeight) if err != nil { - return false, err + return false, nil, err } newValidatorSet, err := sm.GetValidatorSet(pChainHeight) if err != nil { - return false, err + return false, nil, err } if !currentValidatorSet.Equal(newValidatorSet) { @@ -698,9 +738,9 @@ func (sm *StateMachine) createBlockBuildingDecider(pChainReferenceHeight uint64) zap.String("newValidatorSet", fmt.Sprintf("%v", newValidatorSet.NodeWeights())), zap.Uint64("currentPChainRefHeight", pChainReferenceHeight), zap.Uint64("newPChainHeight", pChainHeight)) - return true, nil + return true, newValidatorSet, nil } - return false, nil + return false, nil, nil }, } return blockBuildingDecider @@ -1069,7 +1109,8 @@ func (sm *StateMachine) computeNewApprovals(parentBlock StateMachineBlock, valid // We retrieve approvals that validators have sent us for the next epoch. // These approvals are signed by validators of the next epoch. - approvalsFromPeers := sm.ApprovalsRetriever.Approvals() + sm.maybeInitializeApprovalStore(validators) + approvalsFromPeers := sm.approvalStore.Approvals() sm.Logger.Debug("Retrieved approvals from peers", zap.Int("numApprovals", len(approvalsFromPeers))) // Optimistically sign the epoch transition even if we have already did so in a previous round. diff --git a/msm/msm_test.go b/msm/msm_test.go index 6d7fca94..8baa3117 100644 --- a/msm/msm_test.go +++ b/msm/msm_test.go @@ -701,19 +701,13 @@ func TestMSMFullEpochLifecycle(t *testing.T) { // ----- Step 4: First collecting block (1/3 approvals, not enough to seal) ----- - // Override ApprovalsRetriever to use our dynamic approvals. - var approvalsResult ValidatorSetApprovals - sm.ApprovalsRetriever = &dynamicApprovalsRetriever{approvals: &approvalsResult} - sig1 := signApproval(pChainHeight2, emptyAuxInfoDigest) - approvalsResult = ValidatorSetApprovals{ - { - NodeID: node1, - PChainHeight: pChainHeight2, - AuxInfoDigest: emptyAuxInfoDigest, - Signature: sig1, - }, - } + require.NoError(t, sm.HandleApproval(&ValidatorSetApproval{ + NodeID: node1, + PChainHeight: pChainHeight2, + AuxInfoDigest: emptyAuxInfoDigest, + Signature: sig1, + }, 1)) // node1 is at index 0 in validatorSet2 → bitmask bit 0 → {1} bitmask := []byte{1} @@ -750,14 +744,12 @@ func TestMSMFullEpochLifecycle(t *testing.T) { // ----- Step 5: Second collecting block (2/3 approvals, still not enough since threshold is strictly > 2/3) ----- sig2 := signApproval(pChainHeight2, emptyAuxInfoDigest) - approvalsResult = ValidatorSetApprovals{ - { - NodeID: node2, - PChainHeight: pChainHeight2, - AuxInfoDigest: emptyAuxInfoDigest, - Signature: sig2, - }, - } + require.NoError(t, sm.HandleApproval(&ValidatorSetApproval{ + NodeID: node2, + PChainHeight: pChainHeight2, + AuxInfoDigest: emptyAuxInfoDigest, + Signature: sig2, + }, 2)) // node2 is at index 1 → bitmask bits 0,1 → {3} sig, err = aggr.AppendSignatures(sig, sig2) @@ -794,14 +786,12 @@ func TestMSMFullEpochLifecycle(t *testing.T) { // ----- Step 6: Sealing block (3/3 approvals, enough to seal) ----- sig3 := signApproval(pChainHeight2, emptyAuxInfoDigest) - approvalsResult = ValidatorSetApprovals{ - { - NodeID: node3, - PChainHeight: pChainHeight2, - AuxInfoDigest: emptyAuxInfoDigest, - Signature: sig3, - }, - } + require.NoError(t, sm.HandleApproval(&ValidatorSetApproval{ + NodeID: node3, + PChainHeight: pChainHeight2, + AuxInfoDigest: emptyAuxInfoDigest, + Signature: sig3, + }, 3)) // node3 is at index 2 → bitmask bits 0,1,2 → {7} sig6, err := aggr.AppendSignatures(sig, sig3) diff --git a/msm/util_test.go b/msm/util_test.go index c0465a12..6a1dc8ea 100644 --- a/msm/util_test.go +++ b/msm/util_test.go @@ -369,7 +369,6 @@ func newStateMachineWithLogger(tb testing.TB, logger common.Logger) (*StateMachi Logger: logger, GetBlock: testConfig.blockStore.getBlock, MaxBlockBuildingWaitTime: time.Second, - ApprovalsRetriever: &testConfig.approvalsRetriever, SignatureVerifier: &testConfig.signatureVerifier, SignatureAggregatorCreator: newSignatureAggregatorCreator(), BlockBuilder: &testConfig.blockBuilder, From 8bfb075f387d342eb8589ff5cc7dee5cc74a6a32 Mon Sep 17 00:00:00 2001 From: Yacov Manevich Date: Thu, 2 Jul 2026 22:33:50 +0200 Subject: [PATCH 2/2] Address code review comments Signed-off-by: Yacov Manevich --- msm/approvals.go | 29 ++++++++++------------------- msm/approvals_test.go | 20 ++++++++++++++++++++ msm/msm.go | 2 +- 3 files changed, 31 insertions(+), 20 deletions(-) diff --git a/msm/approvals.go b/msm/approvals.go index 973be34c..db478e64 100644 --- a/msm/approvals.go +++ b/msm/approvals.go @@ -27,7 +27,7 @@ type ApprovalStore struct { signatureVerifier SignatureVerifier validators NodeBLSMappings logger common.Logger - pkByNodeID map[nodeID][]byte + nodeIDToPK map[nodeID][]byte // lock guards the mutable state below (approvalsByNodes, storedCount). The // store is accessed concurrently: approvals are handled as they arrive while // the block builder reads the accumulated approvals when building a block. @@ -50,7 +50,7 @@ func NewApprovalStore(signatureVerifier SignatureVerifier, validators NodeBLSMap return &ApprovalStore{ signatureVerifier: signatureVerifier, validators: validators, - pkByNodeID: pkByNodeID, + nodeIDToPK: pkByNodeID, logger: logger, approvalsByNodes: approvalsByNodes, } @@ -71,7 +71,7 @@ func (as *ApprovalStore) Approvals() ValidatorSetApprovals { func (as *ApprovalStore) HandleApproval(approval *ValidatorSetApproval, timestamp uint64) error { // First thing we check is if the node that sent this approval is a validator. - pk, exists := as.getPKOfNode(approval.NodeID) + pk, exists := as.nodeIDToPK[approval.NodeID] if !exists { as.logger.Debug("Received an approval from a node that is not a validator", zap.String("nodeID", fmt.Sprintf("%x", approval.NodeID)), zap.Uint64("pChainHeight", approval.PChainHeight)) @@ -157,11 +157,6 @@ func (as *ApprovalStore) checkApprovalSignature(approval *ValidatorSetApproval, return as.signatureVerifier.VerifySignature(approval.Signature, toBeSigned, pk) } -func (as *ApprovalStore) getPKOfNode(nodeID nodeID) ([]byte, bool) { - pk, exists := as.pkByNodeID[nodeID] - return pk, exists -} - func (as *ApprovalStore) approvalExistsAndUpToDate(approval *ValidatorSetApproval, timestamp uint64) bool { if as.approvalsByNodes[approval.NodeID] == nil { return false @@ -184,23 +179,19 @@ func (as *ApprovalStore) PutApprovals(approvalStore *ApprovalStore) { // Snapshot the approvals under our lock, then hand them to the destination // store (which takes its own lock). Copying first avoids holding two store // locks at once. - as.lock.Lock() - type approvalWithTimestamp struct { - approval ValidatorSetApproval - timestamp uint64 - } - snapshot := make([]approvalWithTimestamp, 0, as.storedCount) + as.lock.RLock() + snapshot := make([]approvalAndTimestamp, 0, as.storedCount) for _, approvalsByHeight := range as.approvalsByNodes { for _, approval := range approvalsByHeight { - snapshot = append(snapshot, approvalWithTimestamp{ - approval: approval.ValidatorSetApproval, - timestamp: approval.Timestamp, + snapshot = append(snapshot, approvalAndTimestamp{ + ValidatorSetApproval: approval.ValidatorSetApproval, + Timestamp: approval.Timestamp, }) } } - as.lock.Unlock() + as.lock.RUnlock() for _, a := range snapshot { - approvalStore.HandleApproval(&a.approval, a.timestamp) + approvalStore.HandleApproval(&a.ValidatorSetApproval, a.Timestamp) } } diff --git a/msm/approvals_test.go b/msm/approvals_test.go index 865bf6c7..5f9409db 100644 --- a/msm/approvals_test.go +++ b/msm/approvals_test.go @@ -347,6 +347,26 @@ func TestApprovalStorePutApprovals(t *testing.T) { require.Equal(t, dstSent[0].ValidatorSetApproval, got[0]) }, }, + { + // A source approval with the same timestamp as an existing destination approval at the same + // (NodeID, PChainHeight) does not overwrite it: the >= tie in approvalExistsAndUpToDate goes + // to the current destination. + name: "equal-timestamp source does not overwrite destination", + srcValidators: 2, + dstValidators: 2, + dstApprovals: []approvalAndTimestamp{ + {ValidatorSetApproval{NodeID: makeNodeID(1), PChainHeight: 7, Signature: signApproval(7, [32]byte{})}, 100}, + }, + srcApprovals: []approvalAndTimestamp{ + {ValidatorSetApproval{NodeID: makeNodeID(1), PChainHeight: 7, Signature: signApproval(7, [32]byte{})}, 100}, + }, + verify: func(t *testing.T, dst, _ *ApprovalStore, _, dstSent []approvalAndTimestamp) { + got := dst.Approvals() + require.Len(t, got, 1) + require.Equal(t, 1, dst.storedCount) + require.Equal(t, dstSent[0].ValidatorSetApproval, got[0], "the destination approval is retained on a timestamp tie") + }, + }, } { t.Run(tc.name, func(t *testing.T) { src := NewApprovalStore(&signatureVerifier{}, makeValidators(tc.srcValidators), testutil.MakeLogger(t)) diff --git a/msm/msm.go b/msm/msm.go index f86a446d..dd831bc8 100644 --- a/msm/msm.go +++ b/msm/msm.go @@ -650,7 +650,7 @@ func (sm *StateMachine) verifyNextPChainRefHeightNormal(prevMD StateMachineMetad errValidatorSetUnchanged, next.NextPChainReferenceHeight, prev.PChainReferenceHeight) } - // Else, ! currentValidatorSet.Equal(newValidatorSet) || next.NextPChainReferenceHeight == 0 + // Else, !currentValidatorSet.Equal(newValidatorSet) || next.NextPChainReferenceHeight == 0 // so if next.NextPChainReferenceHeight > 0, we should initialize the approval store for the new validator set. if next.NextPChainReferenceHeight > 0 { sm.maybeInitializeApprovalStore(newValidatorSet)