Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Empty file added .changelog/6555.trivial.md
Empty file.
8 changes: 7 additions & 1 deletion go/oasis-node/cmd/debug/byzantine/byzantine.go
Original file line number Diff line number Diff line change
Expand Up @@ -57,12 +57,14 @@ const (
ModeExecutorRunaway ExecutorMode = 2
ModeExecutorStraggler ExecutorMode = 3
ModeExecutorFailureIndicating ExecutorMode = 4
ModeExecutorInvalidBatchHash ExecutorMode = 5

modeExecutorHonestString = "executor_honest"
modeExecutorDishonestString = "executor_dishonest"
modeExecutorRunawayString = "executor_runaway"
modeExecutorStragglerString = "executor_straggler"
modeExecutorFailureIndicatingString = "executor_failure_indicating"
modeExecutorInvalidBatchHashString = "executor_invalid_batch_hash"
)

// String returns a string representation of a executor mode.
Expand All @@ -78,6 +80,8 @@ func (m ExecutorMode) String() string {
return modeExecutorStragglerString
case ModeExecutorFailureIndicating:
return modeExecutorFailureIndicatingString
case ModeExecutorInvalidBatchHash:
return modeExecutorInvalidBatchHashString
default:
return "[unsupported runtime kind]"
}
Expand All @@ -96,8 +100,10 @@ func (m *ExecutorMode) FromString(str string) error {
*m = ModeExecutorStraggler
case modeExecutorFailureIndicatingString:
*m = ModeExecutorFailureIndicating
case modeExecutorInvalidBatchHashString:
*m = ModeExecutorInvalidBatchHash
default:
return fmt.Errorf("invalid executor mode kind: %s", m)
return fmt.Errorf("invalid executor mode kind: %s", str)
}

return nil
Expand Down
14 changes: 13 additions & 1 deletion go/oasis-node/cmd/debug/byzantine/node.go
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import (

beacon "github.com/oasisprotocol/oasis-core/go/beacon/api"
"github.com/oasisprotocol/oasis-core/go/common"
"github.com/oasisprotocol/oasis-core/go/common/crypto/hash"
"github.com/oasisprotocol/oasis-core/go/common/crypto/signature"
"github.com/oasisprotocol/oasis-core/go/common/entity"
"github.com/oasisprotocol/oasis-core/go/common/identity"
Expand Down Expand Up @@ -76,7 +77,18 @@ func (b *byzantine) receiveAndScheduleTransactions(ctx context.Context, cbc *com
panic(fmt.Sprintf("executor proposing batch: %+v", err))
}

if mode == ModeExecutorFailureIndicating {
switch mode {
case ModeExecutorInvalidBatchHash:
validBatchHash := cbc.proposal.Header.BatchHash
cbc.proposal.Header.BatchHash = hash.NewFromBytes([]byte("invalid batch hash"))
if err := cbc.proposal.Sign(b.identity.NodeSigner, block.Header.Namespace); err != nil {
return false, fmt.Errorf("failed to re-sign invalid batch hash proposal: %w", err)
}
logger.Debug("invalid batch hash: invalidating batch hash",
"valid_batch_hash", validBatchHash,
"invalid_batch_hash", cbc.proposal.Header.BatchHash,
)
case ModeExecutorFailureIndicating:
// Submit failure indicating commitment and stop.
logger.Debug("executor failure indicating: submitting commitment and stopping")
schedulerID := b.identity.NodeSigner.Public()
Expand Down
29 changes: 29 additions & 0 deletions go/oasis-test-runner/scenario/e2e/runtime/byzantine.go
Original file line number Diff line number Diff line change
Expand Up @@ -164,6 +164,35 @@ var (
Index: primarySchedulerIndex,
},
)
ByzantineExecutorSchedulerInvalidBatchHash scenario.Scenario = newByzantineImpl(
"primary-worker/primary-scheduler/invalid-batch-hash",
"executor",
[]log.WatcherHandlerFactory{
// The Byzantine node will publish a proposal with an invalid batch hash.
// Other workers will reject the proposal after determining that the I/O root
// does not match. As a result, the second-ranked proposer will prepare a new
// proposal but will not have enough commitments to finalize it. This will
// trigger the round timeout, leading to discrepancy resolution, which will
// succeed once the remaining backup workers submit their commitments.
oasis.LogAssertTimeouts(),
oasis.LogAssertExecutionDiscrepancyDetected(),
},
oasis.ByzantineDefaultIdentitySeed,
false,
// Byzantine node entity should be slashed once for liveness.
map[staking.SlashReason]uint64{
staking.SlashRuntimeLiveness: 1,
},
[]oasis.Argument{
{Name: byzantine.CfgPrimarySchedulerExpected},
{Name: byzantine.CfgExecutorMode, Values: []string{byzantine.ModeExecutorInvalidBatchHash.String()}},
},
scheduler.ForceElectCommitteeRole{
Kind: scheduler.KindComputeExecutor,
Roles: []scheduler.Role{scheduler.RoleWorker},
Index: primarySchedulerIndex,
},
)
// ByzantineExecutorStraggler is a scenario in which the Byzantine node acts
// as the primary worker, backup scheduler, and a straggler.
ByzantineExecutorStraggler scenario.Scenario = newByzantineImpl(
Expand Down
1 change: 1 addition & 0 deletions go/oasis-test-runner/scenario/e2e/runtime/scenario.go
Original file line number Diff line number Diff line change
Expand Up @@ -333,6 +333,7 @@ func RegisterScenarios() error {
ByzantineExecutorDishonest,
ByzantineExecutorSchedulerRunaway,
ByzantineExecutorSchedulerBogus,
ByzantineExecutorSchedulerInvalidBatchHash,
ByzantineExecutorStraggler,
ByzantineExecutorSchedulerStraggler,
ByzantineExecutorStragglerAllowed,
Expand Down
52 changes: 52 additions & 0 deletions go/worker/compute/executor/committee/node.go
Original file line number Diff line number Diff line change
Expand Up @@ -588,6 +588,25 @@ func (n *Node) runtimeExecuteTxBatch(
func (n *Node) startProcessingBatch(ctx context.Context, proposal *commitment.Proposal, rank uint64, batch transaction.RawBatch) {
// This method runs within its own goroutine and is always stopped before the runtime
// worker finishes. Therefore, it is safe to read local round variables (block info, ...).
n.logger.Debug("verifying batch")

ioRoot, err := n.computeIORoot(ctx, proposal.Header.Round, batch)
if err != nil {
n.logger.Error("failed to compute I/O root",
"err", err,
)
// Notify the round worker that the execution failed.
n.processedBatchCh <- nil
return
}
if !ioRoot.Equal(&proposal.Header.BatchHash) {
n.logger.Debug("batch I/O root mismatch")
n.proposals.Reject(proposal, rank)
// Notify the round worker that the execution failed.
n.processedBatchCh <- nil
return
}

n.logger.Debug("processing batch",
"batch_size", len(batch),
)
Expand Down Expand Up @@ -624,6 +643,39 @@ func (n *Node) startProcessingBatch(ctx context.Context, proposal *commitment.Pr
}
}

func (n *Node) computeIORoot(ctx context.Context, round uint64, batch transaction.RawBatch) (hash.Hash, error) {
txs := make([]*transaction.Transaction, 0, len(batch))
for idx, tx := range batch {
txs = append(txs, &transaction.Transaction{
Input: tx,
BatchOrder: uint32(idx),
})
}

emptyRoot := storage.Root{
Namespace: n.rt.ID(),
Version: round,
Type: storage.RootTypeIO,
}
emptyRoot.Hash.Empty()

ioTree := transaction.NewTree(nil, emptyRoot)
defer ioTree.Close()

for _, tx := range txs {
if err := ioTree.AddTransaction(ctx, *tx, nil); err != nil {
return hash.Hash{}, fmt.Errorf("failed to add transaction to tree: %w", err)
}
}

_, ioRoot, err := ioTree.Commit(ctx)
if err != nil {
return hash.Hash{}, fmt.Errorf("failed to commit I/O tree: %w", err)
}

return ioRoot, nil
}

func (n *Node) abortBatch(state *StateProcessingBatch) {
n.logger.Warn("aborting processing batch")

Expand Down
5 changes: 4 additions & 1 deletion go/worker/compute/executor/committee/p2p.go
Original file line number Diff line number Diff line change
Expand Up @@ -91,7 +91,10 @@ func (h *committeeMsgHandler) HandleMessage(_ context.Context, _ signature.Publi
"batch_size", len(proposal.Batch),
)

// Add to the queue.
if _, ok := h.n.proposals.Get(proposal.Header.Round, rank); ok {
return nil
}

if err := h.n.proposals.Add(proposal, rank); err != nil {
return err
}
Expand Down
46 changes: 46 additions & 0 deletions go/worker/compute/executor/committee/proposals.go
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ const maxPendingProposals = 32
type proposalInfo struct {
proposal *commitment.Proposal
rank uint64
invalid bool
}

// proposalQueue is a priority queue of pending proposals, ordered by round and rank.
Expand Down Expand Up @@ -61,6 +62,8 @@ func (q *proposalQueue) Best(round uint64, minRank uint64, maxRank uint64, exclu
return true
case pi.rank > maxRank:
return false
case pi.invalid:
return true
default:
if _, skip := exclude[pi.rank]; skip {
return true
Expand All @@ -75,6 +78,28 @@ func (q *proposalQueue) Best(round uint64, minRank uint64, maxRank uint64, exclu
return proposal, rank, ok
}

// Get returns a proposal for the given round and rank.
func (q *proposalQueue) Get(round uint64, rank uint64) (*commitment.Proposal, bool) {
q.l.RLock()
defer q.l.RUnlock()

info := proposalInfo{
proposal: &commitment.Proposal{
Header: commitment.ProposalHeader{
Round: round,
},
},
rank: rank,
}

pi, ok := q.q.Get(&info)
if !ok {
return nil, false
}

return pi.proposal, true
}

// Add adds a new pending proposal that MUST HAVE already undergone basic validity checks
// and is therefore considered a valid proposal for the given round, but the node's
// local consensus view may not yet be ready to process the proposal.
Expand All @@ -91,6 +116,11 @@ func (q *proposalQueue) Add(proposal *commitment.Proposal, rank uint64) error {
proposal: proposal,
rank: rank,
}

if _, ok := q.q.Get(&info); ok {
return fmt.Errorf("proposal already exists")
}

q.q.ReplaceOrInsert(&info)

// In case of overflows, remove the proposal that is the most in the future.
Expand All @@ -105,6 +135,22 @@ func (q *proposalQueue) Add(proposal *commitment.Proposal, rank uint64) error {
return nil
}

// Reject marks a proposal as invalid.
func (q *proposalQueue) Reject(proposal *commitment.Proposal, rank uint64) {
q.l.Lock()
defer q.l.Unlock()

info, ok := q.q.Get(&proposalInfo{
proposal: proposal,
rank: rank,
})
if !ok {
return
}

info.invalid = true
}

// Prune prunes any proposals which are not valid anymore.
func (q *proposalQueue) Prune(round uint64) {
q.l.Lock()
Expand Down
Loading