lux/consensus
2
0
Fork 0
mirror of https://github.com/luxfi/consensus.git synced 2026-09-24 00:26:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(integration): align integration.go with BlockProposer API

Browse Source 
- Rename VoteEmitter -> BlockProposer
- Rename gossiperEmitter -> gossiperProposer
- Rename NetID -> NetworkID for clarity
- Use SetProposer instead of SetEmitter
- Add HandleIncomingBlock for gossip handling
- Update all method signatures to match engine.go

Fixes node build failure with missing methods.
This commit is contained in:
zeekay
2026-01-01 19:18:22 -08:00
parent 1e7c66ee2d
commit 1011cbfcfd
3 changed files with 819 additions and 432 deletions
Download Patch File Download Diff File Expand all files Collapse all files
engine/chain/engine.go
+431 -343
View File
@@ -2,6 +2,7 @@ package chain
import (
"context"
"errors"
"fmt"
"sync"
"time"
@@ -12,25 +13,51 @@ import (
"github.com/luxfi/ids"
)
// Engine defines the chain consensus engine
// -----------------------------------------------------------------------------
// Errors
// -----------------------------------------------------------------------------
var (
ErrNotStarted = errors.New("engine not started")
ErrAlreadyStarted = errors.New("engine already started")
)
// -----------------------------------------------------------------------------
// Interfaces
// -----------------------------------------------------------------------------
// Engine is the chain consensus engine interface.
type Engine interface {
// Start starts the engine
Start(context.Context, bool) error
// StopWithError stops the engine with an error
StopWithError(context.Context, error) error
// Context returns the engine's context
Context() context.Context
// HealthCheck performs a health check
HealthCheck(context.Context) (interface{}, error)
// IsBootstrapped returns whether the chain is bootstrapped
IsBootstrapped() bool
}
// Re-export message types for convenience
// BlockBuilder is the minimal interface consensus needs from a VM.
// Intentionally small: VMs may implement more, but consensus only needs these.
type BlockBuilder interface {
BuildBlock(context.Context) (block.Block, error)
GetBlock(context.Context, ids.ID) (block.Block, error)
ParseBlock(context.Context, []byte) (block.Block, error)
LastAccepted(context.Context) (ids.ID, error)
}
// BlockProposer submits blocks to validators.
// Consensus expresses WHAT (propose block); implementation decides HOW.
type BlockProposer interface {
Propose(ctx context.Context, proposal BlockProposal) error
RequestVotes(ctx context.Context, req VoteRequest) error
}
// VoteEmitter is deprecated. Use BlockProposer.
type VoteEmitter = BlockProposer
// -----------------------------------------------------------------------------
// Message types
// -----------------------------------------------------------------------------
type (
MessageType = engine.MessageType
Message = engine.Message
@@ -41,41 +68,11 @@ const (
StateSyncDone = engine.StateSyncDone
)
// BlockBuilder is the minimal interface consensus needs from a VM.
// It provides the four core operations required for block consensus:
// building, fetching, parsing, and tracking the chain head.
//
// This interface is intentionally small - VMs may implement the full
// block.ChainVM interface, but consensus only depends on these methods.
type BlockBuilder interface {
// BuildBlock creates a new block from pending transactions.
// Returns an error if no transactions are pending or block creation fails.
BuildBlock(context.Context) (block.Block, error)
// -----------------------------------------------------------------------------
// Protocol types
// -----------------------------------------------------------------------------
// GetBlock retrieves a block by its ID.
// Returns an error if the block is not found.
GetBlock(context.Context, ids.ID) (block.Block, error)
// ParseBlock deserializes a block from its byte representation.
// Used when receiving blocks from the network.
ParseBlock(context.Context, []byte) (block.Block, error)
// LastAccepted returns the ID of the most recently accepted block.
// This is the current chain head from the VM's perspective.
LastAccepted(context.Context) (ids.ID, error)
}
// PendingBlock tracks a block pending consensus decision
type PendingBlock struct {
ConsensusBlock *Block // Block in consensus
VMBlock block.Block // Original VM block (for Accept/Reject callbacks)
ProposedAt time.Time // When block was proposed
VoteCount int // Votes received
Decided bool // Whether consensus decision was made
}
// BlockProposal contains all data needed to propose a block for consensus.
// This is what consensus produces; how it reaches validators is not its concern.
// BlockProposal contains data needed to propose a block.
type BlockProposal struct {
BlockID ids.ID
BlockData []byte
@@ -83,14 +80,13 @@ type BlockProposal struct {
ParentID ids.ID
}
// VoteRequest is a request for specific validators to vote on a block.
// Used when consensus needs targeted vote collection (e.g., missing votes).
// VoteRequest asks specific validators to vote.
type VoteRequest struct {
BlockID ids.ID
Validators []ids.NodeID
}
// Vote represents a validator's decision on a block.
// Vote represents a validator's decision.
type Vote struct {
BlockID ids.ID
NodeID ids.NodeID
@@ -98,62 +94,130 @@ type Vote struct {
SignedAt time.Time
}
// BlockProposer is the interface consensus uses to propose blocks.
// Consensus expresses WHAT it wants (propose block, request votes);
// the implementation decides HOW (gossip, direct send, Warp, etc.).
//
// This separation enables:
// - Testing consensus without network
// - Swapping transport layers (P2P, Warp, local)
// - Composing with different network topologies
//
// Design principles:
// - Small interface (2 methods)
// - Intent-based naming (Propose, not Emit/Broadcast)
// - No network terminology in consensus layer
type BlockProposer interface {
// Propose submits a block for validators to vote on.
// Returns nil if the proposal was accepted for delivery.
// The actual delivery mechanism is implementation-defined.
Propose(ctx context.Context, proposal BlockProposal) error
// RequestVotes asks specific validators to vote on a block.
// Used for vote collection when some validators haven't responded.
// Returns nil if the request was accepted for delivery.
RequestVotes(ctx context.Context, req VoteRequest) error
// PendingBlock tracks a block awaiting consensus.
type PendingBlock struct {
ConsensusBlock *Block
VMBlock block.Block
ProposedAt time.Time
VoteCount int
Decided bool
}
// VoteEmitter is deprecated. Use BlockProposer instead.
// Kept for backward compatibility during migration.
//
// Deprecated: Use BlockProposer interface.
type VoteEmitter = BlockProposer
// -----------------------------------------------------------------------------
// Configuration
// -----------------------------------------------------------------------------
// Config holds engine dependencies and settings.
type Config struct {
Params config.Parameters
VM BlockBuilder
Proposer BlockProposer
// Channel buffer sizes (defaults applied if zero)
VoteRequestBuffer int
VoteBuffer int
}
// DefaultConfig returns sensible defaults.
func DefaultConfig() Config {
return Config{
Params: config.DefaultParams(),
VoteRequestBuffer: 100,
VoteBuffer: 1000,
}
}
// Validate checks config validity.
func (c Config) Validate() error {
return c.Params.Valid()
}
// -----------------------------------------------------------------------------
// Functional options
// -----------------------------------------------------------------------------
// Option configures the engine.
type Option func(*Transitive)
// WithParams sets consensus parameters.
func WithParams(p config.Parameters) Option {
return func(t *Transitive) {
t.params = p
t.consensus = NewChainConsensus(p.K, p.AlphaPreference, int(p.Beta))
}
}
// WithVM sets the block builder.
func WithVM(vm BlockBuilder) Option {
return func(t *Transitive) {
t.vm = vm
}
}
// WithProposer sets the block proposer.
func WithProposer(p BlockProposer) Option {
return func(t *Transitive) {
t.proposer = p
}
}
// WithEmitter is deprecated. Use WithProposer.
func WithEmitter(e BlockProposer) Option {
return WithProposer(e)
}
// WithVoteBuffers sets channel buffer sizes.
func WithVoteBuffers(requests, votes int) Option {
return func(t *Transitive) {
if requests > 0 {
t.voteRequests = make(chan VoteRequest, requests)
}
if votes > 0 {
t.votes = make(chan Vote, votes)
}
}
}
// -----------------------------------------------------------------------------
// Transitive consensus engine
// -----------------------------------------------------------------------------
// Transitive implements chain consensus using Lux protocols.
// It coordinates block building, proposal, voting, and finalization.
//
// Construction:
//
// New() // defaults
// New(WithVM(vm), WithProposer(p)) // with options
// NewWithConfig(cfg) // explicit config
// NewWithConfig(cfg, WithVM(vm)) // config + option overrides
//
// Lifecycle: New -> Start -> (running) -> Stop
type Transitive struct {
mu sync.RWMutex
consensus *ChainConsensus
params config.Parameters
bootstrapped bool
ctx context.Context
cancel context.CancelFunc
pendingBuildBlocks int
vm BlockBuilder // The VM to build blocks from
// Core consensus
consensus *ChainConsensus
params config.Parameters
// Pending blocks awaiting consensus decision
pendingBlocks map[ids.ID]*PendingBlock
// Vote handling channels
voteRequests chan VoteRequest
votes chan Vote
// Block proposer for submitting blocks to validators.
// Consensus expresses intent; proposer handles delivery.
// Dependencies
vm BlockBuilder
proposer BlockProposer
// Metrics for tracking consensus progress
// Runtime state
ctx context.Context
cancel context.CancelFunc
bootstrapped bool
started bool
// Block management
pendingBlocks map[ids.ID]*PendingBlock
pendingBuildBlocks int
// Vote channels
voteRequests chan VoteRequest
votes chan Vote
// Metrics
blocksBuilt uint64
blocksAccepted uint64
blocksRejected uint64
@@ -161,163 +225,85 @@ type Transitive struct {
votesReceived uint64
}
// Transport handles message transport for consensus
type Transport[ID comparable] interface {
// Send sends a message
Send(ctx context.Context, to string, msg interface{}) error
// Receive receives messages
Receive(ctx context.Context) (interface{}, error)
}
// New creates a new chain consensus engine with real Lux consensus
func New() *Transitive {
return NewWithParams(config.DefaultParams())
}
// NewWithParams creates an engine with specific parameters
func NewWithParams(params config.Parameters) *Transitive {
return &Transitive{
consensus: NewChainConsensus(params.K, params.AlphaPreference, int(params.Beta)),
params: params,
bootstrapped: false,
pendingBlocks: make(map[ids.ID]*PendingBlock),
voteRequests: make(chan VoteRequest, 100),
votes: make(chan Vote, 1000),
}
}
// SetProposer sets the block proposer for delivering proposals to validators.
func (t *Transitive) SetProposer(proposer BlockProposer) {
t.mu.Lock()
defer t.mu.Unlock()
t.proposer = proposer
}
// SetEmitter is deprecated. Use SetProposer instead.
// New creates an engine with default parameters.
// Apply options to configure dependencies.
//
// Deprecated: Use SetProposer.
func (t *Transitive) SetEmitter(emitter VoteEmitter) {
t.SetProposer(emitter)
// Example:
//
// engine := New(WithVM(vm), WithProposer(proposer))
func New(opts ...Option) *Transitive {
return NewWithConfig(DefaultConfig(), opts...)
}
// NewWithConfig creates an engine from explicit config plus options.
// Options are applied after config, allowing overrides.
//
// Example:
//
// cfg := Config{Params: config.MainnetParams(), VM: vm}
// engine := NewWithConfig(cfg, WithProposer(proposer))
func NewWithConfig(cfg Config, opts ...Option) *Transitive {
if cfg.VoteRequestBuffer == 0 {
cfg.VoteRequestBuffer = 100
}
if cfg.VoteBuffer == 0 {
cfg.VoteBuffer = 1000
}
t := &Transitive{
consensus: NewChainConsensus(cfg.Params.K, cfg.Params.AlphaPreference, int(cfg.Params.Beta)),
params: cfg.Params,
vm: cfg.VM,
proposer: cfg.Proposer,
pendingBlocks: make(map[ids.ID]*PendingBlock),
voteRequests: make(chan VoteRequest, cfg.VoteRequestBuffer),
votes: make(chan Vote, cfg.VoteBuffer),
}
for _, opt := range opts {
opt(t)
}
return t
}
// NewWithParams creates an engine with specific parameters.
// Deprecated: Use NewWithConfig or New(WithParams(p)).
func NewWithParams(params config.Parameters) *Transitive {
cfg := DefaultConfig()
cfg.Params = params
return NewWithConfig(cfg)
}
// -----------------------------------------------------------------------------
// Lifecycle
// -----------------------------------------------------------------------------
// Start starts the engine.
func (t *Transitive) Start(ctx context.Context, startReqID bool) error {
func (t *Transitive) Start(ctx context.Context, _ bool) error {
t.mu.Lock()
defer t.mu.Unlock()
if t.started {
return ErrAlreadyStarted
}
t.ctx, t.cancel = context.WithCancel(ctx)
t.bootstrapped = true
t.started = true
// Start the consensus polling loop
go t.consensusPollLoop()
// Start the vote handler
go t.pollLoop()
go t.voteHandler()
return nil
}
// consensusPollLoop runs the Wave → Focus consensus phases
// It periodically polls pending blocks and finalizes accepted ones
func (t *Transitive) consensusPollLoop() {
ticker := time.NewTicker(50 * time.Millisecond) // Fast polling for low latency
defer ticker.Stop()
for {
select {
case <-t.ctx.Done():
return
case <-ticker.C:
t.processPendingBlocks()
}
}
}
// processPendingBlocks checks all pending blocks for consensus decisions
func (t *Transitive) processPendingBlocks() {
t.mu.Lock()
defer t.mu.Unlock()
for blockID, pending := range t.pendingBlocks {
if pending.Decided {
continue
}
// Check if consensus has been reached (Focus convergence)
if t.consensus.IsAccepted(blockID) {
pending.Decided = true
t.blocksAccepted++
// Call Accept on the VM block - this finalizes it in the chain
if pending.VMBlock != nil {
if err := pending.VMBlock.Accept(t.ctx); err != nil {
// Log error but continue - the consensus decision was made
fmt.Printf("warning: failed to accept VM block %s: %v\n", blockID, err)
}
}
// Remove from pending (keep for a bit for late votes)
delete(t.pendingBlocks, blockID)
} else if t.consensus.IsRejected(blockID) {
pending.Decided = true
t.blocksRejected++
// Call Reject on the VM block
if pending.VMBlock != nil {
if err := pending.VMBlock.Reject(t.ctx); err != nil {
fmt.Printf("warning: failed to reject VM block %s: %v\n", blockID, err)
}
}
delete(t.pendingBlocks, blockID)
}
}
}
// voteHandler processes incoming votes from validators.
func (t *Transitive) voteHandler() {
for {
select {
case <-t.ctx.Done():
return
case vote := <-t.votes:
t.handleVote(vote)
}
}
}
// handleVote processes a single vote from a validator.
func (t *Transitive) handleVote(vote Vote) {
t.mu.Lock()
defer t.mu.Unlock()
t.votesReceived++
// Record the vote in consensus
if err := t.consensus.ProcessVote(t.ctx, vote.BlockID, vote.Accept); err != nil {
// Block might not exist yet - that's OK for late votes
return
}
// Update pending block vote count
if pending, ok := t.pendingBlocks[vote.BlockID]; ok {
pending.VoteCount++
// Build poll response for consensus
responses := map[ids.ID]int{vote.BlockID: pending.VoteCount}
if err := t.consensus.Poll(t.ctx, responses); err != nil {
return
}
}
}
// StartWithID starts the engine with a specific request ID
// StartWithID starts with a request ID.
func (t *Transitive) StartWithID(ctx context.Context, requestID uint32) error {
return t.Start(ctx, requestID > 0)
}
// Stop stops the engine
// Stop stops the engine.
func (t *Transitive) Stop(ctx context.Context) error {
t.mu.Lock()
defer t.mu.Unlock()
@@ -326,16 +312,16 @@ func (t *Transitive) Stop(ctx context.Context) error {
t.cancel()
}
t.bootstrapped = false
t.started = false
return nil
}
// StopWithError stops the engine with an error (for graceful shutdown on errors)
func (t *Transitive) StopWithError(ctx context.Context, err error) error {
// StopWithError stops with an error.
func (t *Transitive) StopWithError(ctx context.Context, _ error) error {
return t.Stop(ctx)
}
// Context returns the engine's context
// Context returns the engine's context.
func (t *Transitive) Context() context.Context {
t.mu.RLock()
defer t.mu.RUnlock()
@@ -345,7 +331,14 @@ func (t *Transitive) Context() context.Context {
return t.ctx
}
// HealthCheck performs a health check
// IsBootstrapped returns bootstrap status.
func (t *Transitive) IsBootstrapped() bool {
t.mu.RLock()
defer t.mu.RUnlock()
return t.bootstrapped
}
// HealthCheck returns health stats.
func (t *Transitive) HealthCheck(ctx context.Context) (interface{}, error) {
t.mu.RLock()
defer t.mu.RUnlock()
@@ -355,57 +348,70 @@ func (t *Transitive) HealthCheck(ctx context.Context) (interface{}, error) {
stats["k"] = t.params.K
stats["alpha"] = t.params.AlphaPreference
stats["beta"] = t.params.Beta
return stats, nil
}
// IsBootstrapped returns whether the chain is bootstrapped
func (t *Transitive) IsBootstrapped() bool {
t.mu.RLock()
defer t.mu.RUnlock()
return t.bootstrapped
// -----------------------------------------------------------------------------
// Deprecated setters (backward compatibility)
// -----------------------------------------------------------------------------
// SetProposer sets the block proposer.
// Deprecated: Use WithProposer option at construction.
func (t *Transitive) SetProposer(proposer BlockProposer) {
t.mu.Lock()
defer t.mu.Unlock()
t.proposer = proposer
}
// AddBlock adds a block to consensus
func (t *Transitive) AddBlock(ctx context.Context, block *Block) error {
return t.consensus.AddBlock(ctx, block)
// SetEmitter sets the proposer.
// Deprecated: Use WithProposer option at construction.
func (t *Transitive) SetEmitter(e BlockProposer) {
t.SetProposer(e)
}
// ProcessVote processes a vote for a block
func (t *Transitive) ProcessVote(ctx context.Context, blockID ids.ID, accept bool) error {
return t.consensus.ProcessVote(ctx, blockID, accept)
}
// Poll conducts a consensus poll
func (t *Transitive) Poll(ctx context.Context, responses map[ids.ID]int) error {
return t.consensus.Poll(ctx, responses)
}
// IsAccepted checks if a block is accepted
func (t *Transitive) IsAccepted(blockID ids.ID) bool {
return t.consensus.IsAccepted(blockID)
}
// Preference returns the current preferred block
func (t *Transitive) Preference() ids.ID {
return t.consensus.Preference()
}
// GetBlock gets a block by ID
func (t *Transitive) GetBlock(ctx context.Context, nodeID ids.NodeID, requestID uint32, blockID ids.ID) error {
// In the real implementation, this would fetch the block and add to consensus
// For now, just return success
return nil
}
// SetVM sets the block builder (VM) for the engine
// SetVM sets the block builder.
// Deprecated: Use WithVM option at construction.
func (t *Transitive) SetVM(vm BlockBuilder) {
t.mu.Lock()
defer t.mu.Unlock()
t.vm = vm
}
// Notify handles VM notifications (e.g., pending transactions)
// -----------------------------------------------------------------------------
// Consensus operations
// -----------------------------------------------------------------------------
// AddBlock adds a block to consensus.
func (t *Transitive) AddBlock(ctx context.Context, blk *Block) error {
return t.consensus.AddBlock(ctx, blk)
}
// ProcessVote processes a vote.
func (t *Transitive) ProcessVote(ctx context.Context, blockID ids.ID, accept bool) error {
return t.consensus.ProcessVote(ctx, blockID, accept)
}
// Poll conducts a poll.
func (t *Transitive) Poll(ctx context.Context, responses map[ids.ID]int) error {
return t.consensus.Poll(ctx, responses)
}
// IsAccepted checks if block is accepted.
func (t *Transitive) IsAccepted(blockID ids.ID) bool {
return t.consensus.IsAccepted(blockID)
}
// Preference returns preferred block.
func (t *Transitive) Preference() ids.ID {
return t.consensus.Preference()
}
// GetBlock handles a block request.
func (t *Transitive) GetBlock(ctx context.Context, nodeID ids.NodeID, requestID uint32, blockID ids.ID) error {
return nil
}
// Notify handles VM notifications.
func (t *Transitive) Notify(ctx context.Context, msg Message) error {
t.mu.Lock()
defer t.mu.Unlock()
@@ -415,84 +421,21 @@ func (t *Transitive) Notify(ctx context.Context, msg Message) error {
t.pendingBuildBlocks++
return t.buildBlocksLocked(ctx)
case StateSyncDone:
// State sync completed, nothing to do
return nil
}
return nil
}
// buildBlocksLocked builds pending blocks and submits them to consensus.
// Must be called with lock held.
func (t *Transitive) buildBlocksLocked(ctx context.Context) error {
if t.vm == nil {
return nil
}
// Build blocks until we have no more pending
for t.pendingBuildBlocks > 0 {
t.pendingBuildBlocks--
vmBlock, err := t.vm.BuildBlock(ctx)
if err != nil {
// Failed to build block, but this is not fatal
// The VM might not have any transactions ready
return nil
}
t.blocksBuilt++
// Create a consensus block from the VM block
consensusBlock := &Block{
id: vmBlock.ID(),
parentID: vmBlock.ParentID(),
height: vmBlock.Height(),
timestamp: vmBlock.Timestamp().Unix(),
data: vmBlock.Bytes(),
}
// Add the block to consensus for voting
if err := t.consensus.AddBlock(ctx, consensusBlock); err != nil {
// Block might already exist (reorg or duplicate notification)
continue
}
// Track as pending - waiting for consensus decision
t.pendingBlocks[vmBlock.ID()] = &PendingBlock{
ConsensusBlock: consensusBlock,
VMBlock: vmBlock,
ProposedAt: time.Now(),
VoteCount: 0,
Decided: false,
}
// Propose block to validators for voting
if t.proposer != nil {
proposal := BlockProposal{
BlockID: vmBlock.ID(),
BlockData: vmBlock.Bytes(),
Height: vmBlock.Height(),
ParentID: vmBlock.ParentID(),
}
if err := t.proposer.Propose(ctx, proposal); err != nil {
// Failed to propose, but block is still tracked locally
fmt.Printf("warning: failed to propose block %s: %v\n", vmBlock.ID(), err)
}
}
}
return nil
}
// ReceiveVote receives a vote from a validator.
// This is called by the network layer when a vote arrives.
// ReceiveVote queues a vote for processing.
func (t *Transitive) ReceiveVote(vote Vote) {
select {
case t.votes <- vote:
default:
// Channel full, drop vote (it will be resent)
// Channel full; vote will be resent
}
}
// Stats returns engine statistics including consensus metrics
// Stats returns engine statistics.
func (t *Transitive) Stats() map[string]interface{} {
t.mu.RLock()
defer t.mu.RUnlock()
@@ -505,13 +448,158 @@ func (t *Transitive) Stats() map[string]interface{} {
stats["votes_received"] = t.votesReceived
stats["pending_blocks"] = len(t.pendingBlocks)
stats["bootstrapped"] = t.bootstrapped
return stats
}
// PendingBuildBlocks returns the number of pending block builds
// PendingBuildBlocks returns pending build count.
func (t *Transitive) PendingBuildBlocks() int {
t.mu.RLock()
defer t.mu.RUnlock()
return t.pendingBuildBlocks
}
// -----------------------------------------------------------------------------
// Internal
// -----------------------------------------------------------------------------
func (t *Transitive) pollLoop() {
ticker := time.NewTicker(50 * time.Millisecond)
defer ticker.Stop()
for {
select {
case <-t.ctx.Done():
return
case <-ticker.C:
t.processPendingBlocks()
}
}
}
func (t *Transitive) processPendingBlocks() {
t.mu.Lock()
defer t.mu.Unlock()
for blockID, pending := range t.pendingBlocks {
if pending.Decided {
continue
}
if t.consensus.IsAccepted(blockID) {
pending.Decided = true
t.blocksAccepted++
if pending.VMBlock != nil {
if err := pending.VMBlock.Accept(t.ctx); err != nil {
fmt.Printf("warning: accept failed for %s: %v\n", blockID, err)
}
}
delete(t.pendingBlocks, blockID)
} else if t.consensus.IsRejected(blockID) {
pending.Decided = true
t.blocksRejected++
if pending.VMBlock != nil {
if err := pending.VMBlock.Reject(t.ctx); err != nil {
fmt.Printf("warning: reject failed for %s: %v\n", blockID, err)
}
}
delete(t.pendingBlocks, blockID)
}
}
}
func (t *Transitive) voteHandler() {
for {
select {
case <-t.ctx.Done():
return
case vote := <-t.votes:
t.handleVote(vote)
}
}
}
func (t *Transitive) handleVote(vote Vote) {
t.mu.Lock()
defer t.mu.Unlock()
t.votesReceived++
if err := t.consensus.ProcessVote(t.ctx, vote.BlockID, vote.Accept); err != nil {
return
}
if pending, ok := t.pendingBlocks[vote.BlockID]; ok {
pending.VoteCount++
responses := map[ids.ID]int{vote.BlockID: pending.VoteCount}
_ = t.consensus.Poll(t.ctx, responses)
}
}
func (t *Transitive) buildBlocksLocked(ctx context.Context) error {
if t.vm == nil {
return nil
}
for t.pendingBuildBlocks > 0 {
t.pendingBuildBlocks--
vmBlock, err := t.vm.BuildBlock(ctx)
if err != nil {
return nil
}
t.blocksBuilt++
consensusBlock := &Block{
id: vmBlock.ID(),
parentID: vmBlock.ParentID(),
height: vmBlock.Height(),
timestamp: vmBlock.Timestamp().Unix(),
data: vmBlock.Bytes(),
}
if err := t.consensus.AddBlock(ctx, consensusBlock); err != nil {
continue
}
t.pendingBlocks[vmBlock.ID()] = &PendingBlock{
ConsensusBlock: consensusBlock,
VMBlock: vmBlock,
ProposedAt: time.Now(),
VoteCount: 0,
Decided: false,
}
if t.proposer != nil {
proposal := BlockProposal{
BlockID: vmBlock.ID(),
BlockData: vmBlock.Bytes(),
Height: vmBlock.Height(),
ParentID: vmBlock.ParentID(),
}
if err := t.proposer.Propose(ctx, proposal); err != nil {
fmt.Printf("warning: propose failed for %s: %v\n", vmBlock.ID(), err)
}
// Request votes from all validators (nil = all validators)
// This sends PullQuery messages asking validators to vote on this block
voteReq := VoteRequest{
BlockID: vmBlock.ID(),
Validators: nil, // nil means request from all validators
}
if err := t.proposer.RequestVotes(ctx, voteReq); err != nil {
fmt.Printf("warning: request votes failed for %s: %v\n", vmBlock.ID(), err)
}
}
}
return nil
}
// -----------------------------------------------------------------------------
// Transport (network layer interface)
// -----------------------------------------------------------------------------
// Transport handles message transport.
type Transport[ID comparable] interface {
Send(ctx context.Context, to string, msg interface{}) error
Receive(ctx context.Context) (interface{}, error)
}
engine/chain/engine_test.go
+262
View File
@@ -6,6 +6,7 @@ import (
"testing"
"time"
"github.com/luxfi/consensus/config"
"github.com/luxfi/consensus/engine/chain/block"
"github.com/luxfi/ids"
)
@@ -303,3 +304,264 @@ func TestNotifyBuildBlockError(t *testing.T) {
t.Errorf("Expected 1 BuildBlock call, got %d", vm.buildBlockCalls)
}
}
// -----------------------------------------------------------------------------
// Quorum-based acceptance tests
// -----------------------------------------------------------------------------
// trackingMockBlock tracks Accept/Reject calls for testing
type trackingMockBlock struct {
id ids.ID
parentID ids.ID
height uint64
timestamp time.Time
bytes []byte
acceptCalled int
rejectCalled int
}
func (b *trackingMockBlock) ID() ids.ID { return b.id }
func (b *trackingMockBlock) Parent() ids.ID { return b.parentID }
func (b *trackingMockBlock) ParentID() ids.ID { return b.parentID }
func (b *trackingMockBlock) Height() uint64 { return b.height }
func (b *trackingMockBlock) Timestamp() time.Time { return b.timestamp }
func (b *trackingMockBlock) Status() uint8 { return 0 }
func (b *trackingMockBlock) Verify(context.Context) error { return nil }
func (b *trackingMockBlock) Accept(context.Context) error {
b.acceptCalled++
return nil
}
func (b *trackingMockBlock) Reject(context.Context) error {
b.rejectCalled++
return nil
}
func (b *trackingMockBlock) Bytes() []byte { return b.bytes }
// trackingMockVM returns trackingMockBlocks for acceptance tracking
type trackingMockVM struct {
blocks []*trackingMockBlock
buildBlockIdx int
lastAcceptedID ids.ID
}
func (m *trackingMockVM) BuildBlock(ctx context.Context) (block.Block, error) {
if m.buildBlockIdx >= len(m.blocks) {
return nil, errors.New("no more blocks")
}
blk := m.blocks[m.buildBlockIdx]
m.buildBlockIdx++
return blk, nil
}
func (m *trackingMockVM) GetBlock(ctx context.Context, id ids.ID) (block.Block, error) {
for _, blk := range m.blocks {
if blk.id == id {
return blk, nil
}
}
return nil, errors.New("block not found")
}
func (m *trackingMockVM) ParseBlock(ctx context.Context, bytes []byte) (block.Block, error) {
return &trackingMockBlock{bytes: bytes}, nil
}
func (m *trackingMockVM) LastAccepted(ctx context.Context) (ids.ID, error) {
return m.lastAcceptedID, nil
}
// TestEngine_DoesNotAcceptWithoutQuorum verifies blocks are NOT accepted without sufficient votes
func TestEngine_DoesNotAcceptWithoutQuorum(t *testing.T) {
engine := New()
ctx := context.Background()
// Create a tracking block
blk := &trackingMockBlock{
id: ids.GenerateTestID(),
parentID: ids.Empty,
height: 1,
timestamp: time.Now(),
bytes: []byte("block-data"),
}
vm := &trackingMockVM{
blocks: []*trackingMockBlock{blk},
}
engine.SetVM(vm)
if err := engine.Start(ctx, true); err != nil {
t.Fatalf("Start failed: %v", err)
}
defer engine.Stop(ctx)
// Trigger block build
if err := engine.Notify(ctx, Message{Type: PendingTxs}); err != nil {
t.Fatalf("Notify failed: %v", err)
}
// Verify block was added to pending
engine.mu.RLock()
pending, exists := engine.pendingBlocks[blk.id]
engine.mu.RUnlock()
if !exists {
t.Fatal("Block should be in pending blocks")
}
if pending.VMBlock == nil {
t.Fatal("VMBlock should be set")
}
// Process a single vote (below quorum - default K=20, Alpha=15)
// One vote is not enough for quorum
engine.ReceiveVote(Vote{
BlockID: blk.id,
NodeID: ids.GenerateTestNodeID(),
Accept: true,
SignedAt: time.Now(),
})
// Give vote handler time to process
time.Sleep(100 * time.Millisecond)
// Block should NOT be accepted yet (insufficient votes)
if blk.acceptCalled > 0 {
t.Errorf("Block Accept() should NOT be called with insufficient votes, but was called %d times", blk.acceptCalled)
}
// Check consensus state - should not be accepted
if engine.IsAccepted(blk.id) {
t.Error("Block should not be marked as accepted with insufficient votes")
}
}
// TestEngine_AcceptsAfterQuorum verifies blocks ARE accepted after sufficient votes
func TestEngine_AcceptsAfterQuorum(t *testing.T) {
// Use smaller parameters for testing (K=5, Alpha=3, Beta=2)
// This means we need 3 out of 5 votes for quorum
engine := NewWithParams(config.Parameters{
K: 5,
AlphaPreference: 3,
AlphaConfidence: 3,
Beta: 2,
})
ctx := context.Background()
// Create a tracking block
blk := &trackingMockBlock{
id: ids.GenerateTestID(),
parentID: ids.Empty,
height: 1,
timestamp: time.Now(),
bytes: []byte("block-data"),
}
vm := &trackingMockVM{
blocks: []*trackingMockBlock{blk},
}
engine.SetVM(vm)
if err := engine.Start(ctx, true); err != nil {
t.Fatalf("Start failed: %v", err)
}
defer engine.Stop(ctx)
// Trigger block build
if err := engine.Notify(ctx, Message{Type: PendingTxs}); err != nil {
t.Fatalf("Notify failed: %v", err)
}
// Verify block was added
engine.mu.RLock()
_, exists := engine.pendingBlocks[blk.id]
engine.mu.RUnlock()
if !exists {
t.Fatal("Block should be in pending blocks")
}
// Send enough votes to reach quorum (3 votes for Alpha=3)
for i := 0; i < 4; i++ {
engine.ReceiveVote(Vote{
BlockID: blk.id,
NodeID: ids.GenerateTestNodeID(),
Accept: true,
SignedAt: time.Now(),
})
}
// Give poll loop time to process the acceptance
// The poll loop runs every 50ms
time.Sleep(300 * time.Millisecond)
// Block SHOULD be accepted after quorum
if blk.acceptCalled != 1 {
t.Errorf("Block Accept() should be called exactly once after quorum, but was called %d times", blk.acceptCalled)
}
// Block should no longer be in pending
engine.mu.RLock()
_, stillPending := engine.pendingBlocks[blk.id]
engine.mu.RUnlock()
if stillPending {
t.Error("Block should be removed from pending after acceptance")
}
}
// TestEngine_RejectsWithInsufficientSupport verifies blocks can be rejected
// TODO: This test is skipped because reject vote tracking is not yet implemented.
// Currently the consensus only tracks accept votes via RecordVote().
// Future enhancement: Track both accept and reject votes for proper rejection handling.
func TestEngine_RejectsWithInsufficientSupport(t *testing.T) {
t.Skip("Reject vote tracking not yet implemented - only accept votes are tracked")
// Use smaller parameters (K=5, Alpha=3, Beta=2)
engine := NewWithParams(config.Parameters{
K: 5,
AlphaPreference: 3,
AlphaConfidence: 3,
Beta: 2,
})
ctx := context.Background()
blk := &trackingMockBlock{
id: ids.GenerateTestID(),
parentID: ids.Empty,
height: 1,
timestamp: time.Now(),
bytes: []byte("block-data"),
}
vm := &trackingMockVM{
blocks: []*trackingMockBlock{blk},
}
engine.SetVM(vm)
if err := engine.Start(ctx, true); err != nil {
t.Fatalf("Start failed: %v", err)
}
defer engine.Stop(ctx)
// Trigger block build
if err := engine.Notify(ctx, Message{Type: PendingTxs}); err != nil {
t.Fatalf("Notify failed: %v", err)
}
// Send reject votes (Accept=false)
for i := 0; i < 4; i++ {
engine.ReceiveVote(Vote{
BlockID: blk.id,
NodeID: ids.GenerateTestNodeID(),
Accept: false, // Reject votes
SignedAt: time.Now(),
})
}
// Give poll loop time to process
time.Sleep(300 * time.Millisecond)
// Block should NOT have Accept called
if blk.acceptCalled > 0 {
t.Errorf("Block Accept() should NOT be called for rejected block, but was called %d times", blk.acceptCalled)
}
}
engine/chain/integration.go
+126 -89
View File
@@ -15,10 +15,12 @@ import (
// NetworkConfig holds parameters for integrating consensus with the node network.
type NetworkConfig struct {
// ChainID is the blockchain ID
// ChainID is the ID of this chain (used for chain-scoped messages like Put, PullQuery)
ChainID ids.ID
// NetID is the subnet/network ID
NetID ids.ID
// NetworkID is the ID of the network whose validators secure this chain
// For primary network chains (P/X/C), this equals constants.PrimaryNetworkID
// For L1 chains, this is the L1's validator set ID
NetworkID ids.ID
// Logger for consensus events
Logger log.Logger
// Gossiper broadcasts messages to validators
@@ -29,6 +31,9 @@ type NetworkConfig struct {
// Gossiper abstracts the network layer for consensus message broadcasting.
// This minimal interface avoids importing the full node/network package.
//
// Gossiper is the network-level interface. BlockProposer (in engine.go)
// is the consensus-level interface. The gossiperAdapter bridges them.
type Gossiper interface {
// GossipPut broadcasts a Put message with block data to validators.
// Returns the number of validators the message was sent to.
@@ -49,18 +54,18 @@ type IntegratedEngine struct {
// This is the single, canonical way to create a chain consensus engine for node integration.
// It:
// 1. Creates the Transitive engine with default parameters
// 2. Wires the network gossiper as the VoteEmitter
// 2. Wires the network gossiper as the BlockProposer
// 3. Registers the VM for block building
// 4. Returns a ready-to-start engine
//
// Usage in manager.go:
//
// engine := chain.NewIntegratedEngine(chain.NetworkConfig{
// ChainID: chainParams.ID,
// NetID: chainParams.ChainID,
// Logger: m.Log,
// Gossiper: &networkGossiper{net: m.Net, msgCreator: m.MsgCreator},
// VM: vm.(chain.BlockBuilder),
// ChainID: chainParams.ID,
// NetworkID: chainParams.ChainID, // PrimaryNetworkID for P/X/C
// Logger: m.Log,
// Gossiper: &networkGossiper{net: m.Net, msgCreator: m.MsgCreator},
// VM: vm.(chain.BlockBuilder),
// })
// if err := engine.Start(ctx, true); err != nil { return err }
// go engine.ForwardVMNotifications(toEngine)
@@ -72,12 +77,12 @@ func NewIntegratedEngine(cfg NetworkConfig) *IntegratedEngine {
config: cfg,
}
// Wire the emitter (adapts Gossiper to VoteEmitter interface)
engine.SetEmitter(&gossiperEmitter{
gossiper: cfg.Gossiper,
chainID: cfg.ChainID,
netID: cfg.NetID,
logger: cfg.Logger,
// Wire the proposer (adapts Gossiper to BlockProposer interface)
engine.SetProposer(&gossiperProposer{
gossiper: cfg.Gossiper,
chainID: cfg.ChainID,
networkID: cfg.NetworkID,
logger: cfg.Logger,
})
// Set the VM for block building
@@ -123,110 +128,142 @@ func (ie *IntegratedEngine) ForwardVMNotifications(toEngine <-chan block.Message
}
}
// gossiperEmitter adapts a Gossiper to the VoteEmitter interface.
// This is an internal implementation detail - users don't see it.
type gossiperEmitter struct {
gossiper Gossiper
chainID ids.ID
netID ids.ID
logger log.Logger
// gossiperProposer adapts a Gossiper to the BlockProposer interface.
// This bridges the network layer (Gossiper) to the consensus layer (BlockProposer).
type gossiperProposer struct {
gossiper Gossiper
chainID ids.ID
networkID ids.ID
logger log.Logger
}
var _ VoteEmitter = (*gossiperEmitter)(nil)
var _ BlockProposer = (*gossiperProposer)(nil)
func (e *gossiperEmitter) EmitBlockProposal(ctx context.Context, req VoteRequest) error {
if e.gossiper == nil {
if e.logger != nil {
e.logger.Warn("cannot emit block proposal - gossiper is nil",
log.Stringer("blockID", req.BlockID))
// Propose broadcasts a block proposal to validators via the network gossiper.
func (p *gossiperProposer) Propose(ctx context.Context, proposal BlockProposal) error {
if p.gossiper == nil {
if p.logger != nil {
p.logger.Warn("cannot propose block - gossiper is nil",
log.Stringer("blockID", proposal.BlockID))
}
return nil // Not fatal - local acceptance still works
}
sentTo := e.gossiper.GossipPut(e.chainID, e.netID, req.BlockData)
if e.logger != nil {
e.logger.Info("emitted block proposal to validators (Photon phase)",
log.Stringer("blockID", req.BlockID),
log.Uint64("height", req.Height),
sentTo := p.gossiper.GossipPut(p.chainID, p.networkID, proposal.BlockData)
if p.logger != nil {
p.logger.Info("proposed block to validators",
log.Stringer("blockID", proposal.BlockID),
log.Uint64("height", proposal.Height),
log.Int("sentTo", sentTo))
}
return nil
}
func (e *gossiperEmitter) EmitVoteRequest(ctx context.Context, blockID ids.ID, validators []ids.NodeID) error {
if e.gossiper == nil {
if e.logger != nil {
e.logger.Warn("cannot emit vote request - gossiper is nil",
log.Stringer("blockID", blockID))
// RequestVotes asks specific validators to vote on a block.
func (p *gossiperProposer) RequestVotes(ctx context.Context, req VoteRequest) error {
if p.gossiper == nil {
if p.logger != nil {
p.logger.Warn("cannot request votes - gossiper is nil",
log.Stringer("blockID", req.BlockID))
}
return nil
}
sentTo := e.gossiper.SendPullQuery(e.chainID, e.netID, blockID, validators)
if e.logger != nil {
e.logger.Debug("requested votes from validators (Wave phase)",
log.Stringer("blockID", blockID),
log.Int("requested", len(validators)),
sentTo := p.gossiper.SendPullQuery(p.chainID, p.networkID, req.BlockID, req.Validators)
if p.logger != nil {
p.logger.Debug("requested votes from validators",
log.Stringer("blockID", req.BlockID),
log.Int("requested", len(req.Validators)),
log.Int("sentTo", sentTo))
}
return nil
}
// SetVM sets the block builder VM on the engine.
func (t *Transitive) SetVM(vm BlockBuilder) {
t.mu.Lock()
defer t.mu.Unlock()
t.vm = vm
}
// Notify handles VM notifications (PendingTxs triggers block building).
func (t *Transitive) Notify(ctx context.Context, msg Message) error {
if msg.Type == PendingTxs {
return t.buildAndProposeBlock(ctx)
}
return nil
}
// buildAndProposeBlock builds a block from the VM and adds it to consensus.
func (t *Transitive) buildAndProposeBlock(ctx context.Context) error {
t.mu.Lock()
defer t.mu.Unlock()
if t.vm == nil {
return nil // No VM set, skip
// HandleIncomingBlock processes a block received from network gossip.
// Instead of auto-accepting, it:
// 1. Parses and verifies the block
// 2. Adds it to pending blocks for consensus tracking
// 3. Votes in favor of the block
// Returns the parsed block if successful, nil otherwise.
func (ie *IntegratedEngine) HandleIncomingBlock(ctx context.Context, blockData []byte, fromNodeID ids.NodeID) (block.Block, error) {
if ie.config.VM == nil {
return nil, nil
}
// Build the block
blk, err := t.vm.BuildBlock(ctx)
// Parse the block
blk, err := ie.config.VM.ParseBlock(ctx, blockData)
if err != nil {
return err
if ie.config.Logger != nil {
ie.config.Logger.Debug("failed to parse incoming block",
log.Stringer("from", fromNodeID),
log.Err(err))
}
return nil, nil // Don't return error - just skip invalid blocks
}
t.blocksBuilt++
blockID := blk.ID()
// Create consensus block
consensusBlk, err := t.consensus.Add(ctx, blockID, blk.Parent(), blk.Height())
if err != nil {
return err
// Verify the block
if err := blk.Verify(ctx); err != nil {
if ie.config.Logger != nil {
ie.config.Logger.Debug("incoming block failed verification",
log.Stringer("blockID", blk.ID()),
log.Stringer("from", fromNodeID),
log.Err(err))
}
return nil, nil // Don't return error - just skip invalid blocks
}
// Track pending block
t.pendingBlocks[blockID] = &PendingBlock{
ConsensusBlock: consensusBlk,
VMBlock: blk,
ProposedAt: time.Now(),
if ie.config.Logger != nil {
ie.config.Logger.Info("received and verified block from gossip",
log.Stringer("blockID", blk.ID()),
log.Uint64("height", blk.Height()),
log.Stringer("from", fromNodeID))
}
// Emit block proposal (Photon phase)
if t.emitter != nil {
t.emitter.EmitBlockProposal(ctx, VoteRequest{
BlockID: blockID,
BlockData: blk.Bytes(),
Height: blk.Height(),
ParentID: blk.Parent(),
})
// Create consensus block and add to tracking
consensusBlock := &Block{
id: blk.ID(),
parentID: blk.ParentID(),
height: blk.Height(),
timestamp: blk.Timestamp().Unix(),
data: blk.Bytes(),
}
return nil
// Add to consensus engine
if err := ie.Transitive.consensus.AddBlock(ctx, consensusBlock); err != nil {
if ie.config.Logger != nil {
ie.config.Logger.Debug("failed to add block to consensus",
log.Stringer("blockID", blk.ID()),
log.Err(err))
}
// Continue anyway - block may already exist
}
// Add to pending blocks for tracking
ie.Transitive.mu.Lock()
if _, exists := ie.Transitive.pendingBlocks[blk.ID()]; !exists {
ie.Transitive.pendingBlocks[blk.ID()] = &PendingBlock{
ConsensusBlock: consensusBlock,
VMBlock: blk,
ProposedAt: time.Now(),
VoteCount: 1, // Count our own vote
Decided: false,
}
} else {
// Block already tracked, increment vote count
ie.Transitive.pendingBlocks[blk.ID()].VoteCount++
}
ie.Transitive.mu.Unlock()
// Vote in favor of the block (process our vote)
responses := map[ids.ID]int{blk.ID(): 1}
if err := ie.Transitive.consensus.Poll(ctx, responses); err != nil {
if ie.config.Logger != nil {
ie.config.Logger.Debug("failed to vote on block",
log.Stringer("blockID", blk.ID()),
log.Err(err))
}
}
return blk, nil
}