diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b296a16d..16015f4d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -61,6 +61,31 @@ jobs: - name: Test run: go test ./... + consensus-lab: + name: Consensus & performance lab + runs-on: ubuntu-latest + steps: + - name: Checkout + uses: actions/checkout@v6 + + - name: Set up Go + uses: actions/setup-go@v7 + with: + go-version-file: go.mod + cache: false + + - name: Multi-validator conformance gate + run: go test ./internal/api -run '^TestLab' -count=1 -timeout=90s + + - name: Partition recovery stress gate + run: go test ./internal/api -run '^TestLabSevenValidatorsStallWithoutQuorumThenRecoverWithPeerSync$' -count=3 -timeout=90s + + - name: Seven-validator finalized-throughput sample + run: go test ./internal/api -run '^$' -bench '^BenchmarkLabConsensusFinality7Validators$' -benchtime=3x -count=1 -timeout=120s + + - name: P-256 verification baseline + run: go test ./internal/api -run '^$' -bench '^BenchmarkLabP256TransactionVerification$' -benchtime=1s -count=1 + wallet: name: Wallet build runs-on: ubuntu-latest @@ -85,4 +110,4 @@ jobs: run: npm audit --audit-level=high - name: Type-check and build - run: npm run build + run: npm run build \ No newline at end of file diff --git a/docs/performance-lab.md b/docs/performance-lab.md new file mode 100644 index 00000000..d64b24a1 --- /dev/null +++ b/docs/performance-lab.md @@ -0,0 +1,238 @@ +# Zephyr Consensus & Performance Lab + +## Purpose + +The Consensus & Performance Lab turns Zephyr's scalability target into a repeatable engineering program with two independent responsibilities: + +1. **protocol conformance**: prove safety and liveness under deterministic multi-validator faults; +2. **performance measurement**: measure finalized throughput and finality through the real transaction, consensus, state and persistence paths. + +A performance change is not successful if it weakens the conformance matrix. + +## Canonical meaning of TPS + +For Zephyr, the headline TPS number means: + +> **transactions finalized by validator consensus per second**. + +The benchmark does not count HTTP requests accepted, mempool insertions, unsigned synthetic operations, or batches that were not individually executed as transactions. A transaction counts only after it appears in a committed block protected by the configured quorum-certificate rules. + +## Canonical transfer workload + +The first reference workload is a native ZPH transfer with: + +- a real P-256 signature; +- the normal transaction domain and chain ID; +- normal static and stateful validation; +- a real mempool entry; +- real state execution; +- a deterministic state commitment; +- proposal and vote dissemination; +- quorum-certificate formation; +- committed block persistence. + +Client-side key generation/signing is prepared outside the timed consensus benchmark. Signature **verification** remains inside the node path and is benchmarked separately as well. + +## Validator matrix + +The target matrix is: + +- 1 validator: local execution/control baseline; +- 4 validators: smallest useful multi-validator BFT-style lab; +- 7 validators: primary development baseline; +- 16 validators: scaling and dissemination-pressure baseline. + +The first checked-in gate focuses on 4 and 7 validators. The harness is parameterized so 1- and 16-validator scenarios can use the same machinery later. + +The 7-validator reference configuration uses equal voting power: `10,000` per validator, `70,000` total, with Zephyr's normal quorum calculation producing a `46,667` voting-power threshold. + +## Required metrics + +Every publishable Zephyr performance result should report at least: + +- sustained finalized transactions per second; +- time to finality p50, p95 and p99; +- validator count; +- transactions per finalized block; +- finalized block payload size; +- protocol payload bytes per finalized transaction; +- persisted state size per node; +- CPU profile; +- heap/allocation profile; +- machine CPU, RAM, operating system and Go version; +- network topology and latency assumptions. + +The repository benchmark currently emits: + +- `finalized-tx/s`; +- `finality-p50-ms`; +- `finality-p95-ms`; +- `finality-p99-ms`; +- `protocol-payload-B/finalized-tx`; +- `finalized-block-B`; +- `state-B/node`. + +CPU, heap, mutex and block profiles come from the standard Go benchmark profiler. + +## First CI baseline + +The first successful 7-validator measurement on a GitHub-hosted Ubuntu runner is a **development baseline only**, not a Zephyr performance claim and not a controlled-hardware result. + +With 32 finalized transfers per block, one observed run on an AMD EPYC 7763 hosted runner reported approximately: + +- `42.92 finalized-tx/s`; +- `628 ms` p50 finality; +- `1.128 s` p95/p99 finality; +- `23,045 B` finalized block size; +- `17,468 B` measured protocol payload per finalized transaction; +- `162,138 B` persisted state per node after the sample; +- approximately `88.8 us/op` for the separate P-256 transaction-validation baseline. + +A separate verification run was in the same rough range at about `44.3 finalized-tx/s`. Shared-runner variance is expected. These numbers establish a measurable starting point and must not be presented as production capacity. + +## Running the lab + +Run the complete protocol conformance gate: + +```bash +go test ./internal/api -run '^TestLab' -count=1 -timeout=90s +``` + +Stress the partition/heal recovery path repeatedly: + +```bash +go test ./internal/api \ + -run '^TestLabSevenValidatorsStallWithoutQuorumThenRecoverWithPeerSync$' \ + -count=5 \ + -timeout=120s +``` + +Run the 7-validator finalized-throughput benchmark across consecutive finalized blocks: + +```bash +go test ./internal/api \ + -run '^$' \ + -bench '^BenchmarkLabConsensusFinality7Validators$' \ + -benchtime=5x \ + -count=1 \ + -timeout=120s +``` + +Measure P-256 transaction verification separately: + +```bash +go test ./internal/api \ + -run '^$' \ + -bench '^BenchmarkLabP256TransactionVerification$' \ + -benchtime=2s \ + -count=1 +``` + +Generate profiles for the end-to-end benchmark: + +```bash +go test ./internal/api \ + -run '^$' \ + -bench '^BenchmarkLabConsensusFinality7Validators$' \ + -benchtime=5x \ + -cpuprofile=cpu.out \ + -memprofile=mem.out \ + -mutexprofile=mutex.out \ + -blockprofile=block.out \ + -count=1 \ + -timeout=120s +``` + +Then inspect a profile, for example: + +```bash +go tool pprof -http=:8081 cpu.out +``` + +## Fault-injection contract + +The lab transport wraps Zephyr's existing `peerTransport`; it does not replace consensus or ledger logic. Faults are injected at the transport boundary while transactions, proposals, votes, certificates, blocks, state roots and persistence remain production implementations. + +The initial gate covers: + +- 7-validator certified happy-path finality; +- a 4/3 partition where neither side has quorum: no block may commit while partitioned; after heal, quorum finality must resume and lagging validators must catch up; +- a 5/2 partition where the quorum side commits and the minority later catches up through normal peer recovery; +- delayed, duplicated and deliberately vote-before-proposal delivery: all nodes must still converge on one committed tip. + +The 4/3 scenario exposed a real recovery gap. After heal, enough validators can sign the same proposal to finalize a block even though only a subset has already materialized that block. Snapshot recovery alone cannot solve every such state immediately because fewer than 2/3 of validators may have the new committed snapshot available. + +## Certified block catch-up + +Zephyr therefore has a signed-evidence catch-up path before snapshot fallback: + +1. an authenticated internal endpoint exposes retained **proposal/vote fragments for a height even before the serving node has materialized the block**; +2. each peer may contribute only a partial fragment; +3. the receiver groups fragments by the same canonical proposal/round/block and never combines votes from competing proposals or rounds; +4. malformed or conflicting peer fragments are ignored without poisoning compatible evidence from other peers; +5. matching valid votes already persisted by the recovering validator are combined with compatible remote fragments; +6. every P-256 proposal/vote signature, validator identity and scheduled proposer is independently validated against the receiver's local validator set; +7. the receiver locally recomputes voting power and requires the normal `2/3+` quorum before any state mutation; +8. the block is independently executed and its chain continuity, transaction validity, state root and hash are checked before atomic import; +9. only after those checks does the receiver derive its local commit certificate; +10. quorum-validated snapshot recovery remains the fallback for deeper repair. + +This means a peer can transport evidence but cannot manufacture finality: a recovery import still requires the same validator signatures that would have been necessary for consensus. The transport capability is optional, so the future libp2p/QUIC transport can implement the same contract while HTTP remains the reference transport. + +The unit and lab gates explicitly verify that: + +- evidence below quorum cannot mutate state; +- a tampered signature is rejected; +- a locally persisted matching vote can contribute to the quorum together with remote evidence; +- partial compatible evidence from multiple peers can be aggregated; +- repeated 4/3 partition/heal recovery converges without depending on message arrival order. + +## Next conformance cases + +The matrix should expand to cover: + +- validator offline/restart before and after vote; +- proposer crash during a round; +- conflicting proposals; +- conflicting votes; +- explicit Byzantine peer payloads; +- corrupted snapshots; +- wrong-chain validators; +- longer partitions and repeated heal/fail cycles; +- the same conformance suite over HTTP and future libp2p/QUIC transport. + +## Performance-gate policy + +Correctness is a hard gate now. Numerical performance thresholds are deliberately not hard-coded against GitHub-hosted runners because shared-runner variance would make the gate noisy. + +The next performance step is to establish a controlled reference machine and retain benchmark history. Once variance is understood, Zephyr can add regression budgets such as: + +- no more than N% sustained-TPS regression; +- no more than N% p95 finality regression; +- no unbounded growth in protocol bytes per finalized transaction; +- no unexpected state-size or allocation regression. + +## Optimization decision rule + +No major performance architecture is selected before profiling the canonical benchmark. + +The first profile should attribute time and resource pressure across: + +`signature verification -> transaction validation -> mempool -> state execution -> state root -> block serialization -> proposal dissemination -> votes -> persistence` + +Parallel signature verification, serialization changes, storage replacement, lock reduction, incremental state commitments and transport/dissemination changes are hypotheses until the profile identifies the actual bottleneck. + +## Path toward 1M TPS + +The engineering sequence is: + +1. consensus and performance lab; +2. profiling and evidence-backed performance architecture; +3. scalable storage/state execution; +4. production libp2p/QUIC transport, while keeping HTTP as the reference transport; +5. deterministic Rust-first WASM execution and fee metering; +6. staking/governance; +7. public devnet; +8. confidential compute marketplace. + +At every stage the canonical finalized-TPS/finality benchmark and the consensus conformance matrix remain the comparison point. diff --git a/docs/performance-profile-2026-08-19.md b/docs/performance-profile-2026-08-19.md new file mode 100644 index 00000000..a48932ea --- /dev/null +++ b/docs/performance-profile-2026-08-19.md @@ -0,0 +1,100 @@ +# Zephyr Canonical Performance Profile — 2026-08-19 + +## Scope + +This profile uses the canonical Consensus & Performance Lab benchmark on a GitHub-hosted Ubuntu 24.04 runner with Go 1.22.12 and an AMD EPYC 7763 CPU. + +The measured workload uses 7 validators, real P-256 transaction validation, real HTTP peer replication, deterministic state execution/state root, quorum-certificate consensus and persisted committed blocks. The profiling run used 5 consecutive benchmark iterations with 32 finalized transfers per block. + +This document records engineering evidence, not a production performance claim. + +## Observed benchmark sample + +The profiling run reported approximately: + +- `38.52 finalized-tx/s`; +- `971.5 ms` p50 finality; +- `1.106 s` p95/p99 finality; +- `23,054 B` finalized block size; +- `15,071 B` measured protocol payload per finalized transaction; +- `244,576 B` persisted state per node after the five-iteration sample. + +The lower TPS than shorter CI samples is useful: the sustained multi-block run increases persisted state and exposes costs that a one-block microbenchmark hides. + +## CPU profile + +The strongest CPU signal is persistence/serialization: + +- `ledger.(*Store).writeState`: about **58.3% cumulative CPU**; +- `encoding/json.MarshalIndent`: about **54.5% cumulative CPU**; +- `encoding/json.appendIndent`: about **28.6% flat CPU**; +- `ledger.(*Store).Accept`: about **31.9% cumulative CPU**; +- P-256 `tx.VerifySignature`: about **10.1% cumulative CPU**. + +The conclusion is that signature verification is material but is not the first bottleneck. Full-state JSON persistence currently costs substantially more CPU than P-256 verification. + +## Allocation profile + +The allocation profile reinforces the same result. Roughly 2.0 GB were allocated during the profiled process, with: + +- `encoding/json.MarshalIndent`: about **40.8% flat allocations** and **73.8% cumulative** through its call tree; +- `ledger.(*Store).writeState`: about **71.4% cumulative allocations**; +- `encoding/json.Marshal`: about **20.4% flat allocations**; +- `bytes.growSlice`: about **11.6% flat allocations**; +- repeated cloning/snapshot construction also contributes materially as the persisted state grows. + +The current persistence model serializes and atomically rewrites a broad `persistedState` structure after high-frequency operations such as transaction acceptance, funding and consensus vote recording. That design is excellent for simple correctness/restart guarantees but does not scale as the hot-path persistence architecture. + +## Lock contention + +The mutex profile identifies the ledger write lock as another direct consequence of the persistence model: + +- `ledger.(*Store).Accept`: about **89.7% cumulative mutex delay**; +- `sync.(*Mutex).Unlock`: about **82.7% flat mutex delay**; +- `handleBroadcastTransaction`: about **98.8% cumulative through the affected request path**. + +Parallel transaction ingress therefore serializes behind a state-wide critical section that also performs expensive cloning/serialization/persistence work. + +## Blocking profile + +The blocking profile shows two major classes: + +1. ledger lock waiting during concurrent transaction acceptance; +2. synchronous HTTP transaction fan-out to peers. + +Notable cumulative blocking signals include: + +- HTTP client/send/round-trip paths around **35–37%**; +- `httpPeerTransport.postJSON` around **25%**; +- `Server.broadcastTransaction` around **23%**; +- ledger `Store.Accept` around **13%**. + +This confirms that transaction-by-transaction synchronous replication will become a later networking/dissemination bottleneck, but the persistence/lock problem should be addressed first because it dominates both CPU/allocations and local contention. + +## Evidence-backed optimization order + +The first optimization sequence is therefore: + +1. **Persistence hot path** + - remove human-readable `MarshalIndent` from machine state persistence immediately; + - stop treating full-state JSON rewrite as the long-term write path; + - introduce a durable append/batch-oriented journal or structured state backend so mempool/vote/transaction mutations do not reserialize the whole node state; + - preserve atomic committed checkpoints and restart validation. +2. **Ledger concurrency** + - reduce the amount of work performed while holding the global store mutex; + - separate validation/read preparation from the serialized commit section where deterministic safety allows; + - move toward deterministic batch execution/state updates rather than one durable full-state rewrite per transaction. +3. **State commitment/storage architecture** + - make state commitments incremental rather than rebuilding broad state structures as the chain grows; + - introduce a backend abstraction suitable for structured key/value state and deterministic snapshots. +4. **Networking/dissemination** + - replace synchronous transaction-by-transaction HTTP fan-out with batched/asynchronous dissemination semantics; + - retain the HTTP transport as the reference implementation while adding libp2p/QUIC later. +5. **Signature verification parallelism** + - parallelize P-256 verification once persistence/locking no longer masks its cost. + +## Immediate next experiment + +The first low-risk performance change should replace `json.MarshalIndent` with compact deterministic-equivalent JSON for `state.json` persistence and rerun the exact same 7-validator benchmark/profiles. This does not change consensus semantics or on-disk JSON meaning, but it directly tests the largest CPU/allocation signal. + +If the expected improvement appears, the next architectural change should replace repeated full-state rewrites with a journal/checkpoint persistence layer. No storage engine should be selected until that benchmark establishes how much of the remaining cost is serialization, file I/O, state cloning and lock hold time. diff --git a/internal/api/block_evidence.go b/internal/api/block_evidence.go new file mode 100644 index 00000000..32b9341e --- /dev/null +++ b/internal/api/block_evidence.go @@ -0,0 +1,37 @@ +package api + +import ( + "net/http" + "strconv" + "strings" + + "github.com/zephyr-chain/zephyr-chain/internal/ledger" +) + +type BlockEvidenceResponse struct { + Evidence []ledger.CertifiedBlockEvidence `json:"evidence"` +} + +func (s *Server) handleBlockEvidence(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + w.WriteHeader(http.StatusMethodNotAllowed) + return + } + if err := s.validatePeerRequest(r); err != nil { + writeJSON(w, statusForError(err), map[string]string{"error": err.Error()}) + return + } + + rawHeight := strings.TrimPrefix(r.URL.Path, "/v1/internal/block-evidence/") + height, err := strconv.ParseUint(rawHeight, 10, 64) + if err != nil || height == 0 { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid block evidence height"}) + return + } + fragments := s.ledger.CertifiedBlockEvidenceFragments(height) + if len(fragments) == 0 { + writeJSON(w, http.StatusNotFound, map[string]string{"error": "certified block evidence not found"}) + return + } + writeJSON(w, http.StatusOK, BlockEvidenceResponse{Evidence: fragments}) +} diff --git a/internal/api/certified_block_recovery.go b/internal/api/certified_block_recovery.go new file mode 100644 index 00000000..600440d1 --- /dev/null +++ b/internal/api/certified_block_recovery.go @@ -0,0 +1,150 @@ +package api + +import ( + "errors" + "fmt" + "sort" + + "github.com/zephyr-chain/zephyr-chain/internal/consensus" + "github.com/zephyr-chain/zephyr-chain/internal/ledger" +) + +func (s *Server) recoverCertifiedBlockFromPeers(primaryPeerURL string, height uint64, block ledger.Block) error { + transport, ok := s.transport.(certifiedBlockEvidenceTransport) + if !ok { + return fmt.Errorf("peer transport does not support certified block evidence") + } + + peerURLs := make([]string, 0, len(s.config.PeerURLs)+1) + seenPeers := make(map[string]struct{}, len(s.config.PeerURLs)+1) + appendPeer := func(peerURL string) { + if peerURL == "" { + return + } + if _, exists := seenPeers[peerURL]; exists { + return + } + seenPeers[peerURL] = struct{}{} + peerURLs = append(peerURLs, peerURL) + } + appendPeer(primaryPeerURL) + for _, peerURL := range s.config.PeerURLs { + appendPeer(peerURL) + } + + bundles := make(map[string]ledger.CertifiedBlockEvidence) + var lastErr error = ledger.ErrCertifiedEvidenceQuorum + for _, peerURL := range peerURLs { + fragments, err := transport.FetchBlockEvidence(peerURL, height) + if err != nil { + lastErr = err + continue + } + for _, fragment := range fragments { + proposal := fragment.Proposal + if proposal.Height != block.Height || proposal.BlockHash != block.Hash || proposal.StateRoot != block.StateRoot { + continue + } + key := certifiedProposalKey(proposal) + bundle := bundles[key] + if err := mergeCertifiedBlockEvidence(&bundle, fragment, block, s.config.ChainID); err != nil { + // Malformed/conflicting evidence from one peer must not poison a + // compatible bundle collected from other peers. + lastErr = err + continue + } + bundles[key] = bundle + + if err := s.ledger.ImportBlockWithEvidence(block, bundle); err != nil { + if errors.Is(err, ledger.ErrCertifiedEvidenceQuorum) { + lastErr = err + continue + } + // Another valid round/proposal fragment may still provide the + // evidence for the committed block, so keep collecting unless the + // error is a block/state invariant failure. + if errors.Is(err, ledger.ErrCertifiedEvidenceInvalid) || errors.Is(err, ledger.ErrConflictingProposal) || errors.Is(err, ledger.ErrConflictingVote) { + lastErr = err + continue + } + return err + } + return nil + } + } + return lastErr +} + +func certifiedProposalKey(proposal consensus.Proposal) string { + // Canonical payload already binds chain/domain/height/round/block/state and + // proposer identity. Public key is included explicitly; the ECDSA signature + // bytes are not part of the grouping key because two valid signatures over + // the same canonical proposal are semantically equivalent. + return proposal.Payload + "\x00" + proposal.Proposer + "\x00" + proposal.PublicKey +} + +func mergeCertifiedBlockEvidence(dst *ledger.CertifiedBlockEvidence, fragment ledger.CertifiedBlockEvidence, block ledger.Block, chainID string) error { + proposal := fragment.Proposal + if err := proposal.ValidateForChain(chainID); err != nil { + return fmt.Errorf("invalid certified block proposal: %w", err) + } + if proposal.Height != block.Height || proposal.BlockHash != block.Hash || proposal.StateRoot != block.StateRoot { + return ledger.ErrCertifiedEvidenceInvalid + } + + if dst.Proposal.Height == 0 { + dst.Proposal = proposal + } else if !sameCertifiedProposal(dst.Proposal, proposal) { + return ledger.ErrCertifiedEvidenceInvalid + } + + votesByValidator := make(map[string]consensus.Vote, len(dst.Votes)+len(fragment.Votes)) + for _, vote := range dst.Votes { + votesByValidator[vote.Voter] = vote + } + for _, vote := range fragment.Votes { + if err := vote.ValidateForChain(chainID); err != nil { + return fmt.Errorf("invalid certified block vote: %w", err) + } + if vote.Height != proposal.Height || vote.Round != proposal.Round || vote.BlockHash != proposal.BlockHash { + return ledger.ErrCertifiedEvidenceInvalid + } + if existing, ok := votesByValidator[vote.Voter]; ok { + if !sameCertifiedVote(existing, vote) { + return ledger.ErrCertifiedEvidenceInvalid + } + continue + } + votesByValidator[vote.Voter] = vote + } + + dst.Votes = dst.Votes[:0] + for _, vote := range votesByValidator { + dst.Votes = append(dst.Votes, vote) + } + sort.Slice(dst.Votes, func(i, j int) bool { return dst.Votes[i].Voter < dst.Votes[j].Voter }) + return nil +} + +func sameCertifiedProposal(left consensus.Proposal, right consensus.Proposal) bool { + return left.ChainID == right.ChainID && + left.Domain == right.Domain && + left.Height == right.Height && + left.Round == right.Round && + left.BlockHash == right.BlockHash && + left.StateRoot == right.StateRoot && + left.Proposer == right.Proposer && + left.PublicKey == right.PublicKey && + left.Payload == right.Payload +} + +func sameCertifiedVote(left consensus.Vote, right consensus.Vote) bool { + return left.ChainID == right.ChainID && + left.Domain == right.Domain && + left.Height == right.Height && + left.Round == right.Round && + left.BlockHash == right.BlockHash && + left.Voter == right.Voter && + left.PublicKey == right.PublicKey && + left.Payload == right.Payload +} diff --git a/internal/api/peer_sync.go b/internal/api/peer_sync.go index d7031d76..27ae26a6 100644 --- a/internal/api/peer_sync.go +++ b/internal/api/peer_sync.go @@ -162,6 +162,16 @@ func (s *Server) syncFromPeer(peerURL string, localHeight uint64, remoteHeight u result.ImportFailureHeight = block.Height result.ImportFailureBlockHash = block.Hash s.recordBlockImportFailure("peer_sync", block, err, peerURL) + + var evidenceErr error + if s.config.RequireConsensusCertificates { + if recoveryErr := s.recoverCertifiedBlockFromPeers(peerURL, height, block); recoveryErr != nil { + evidenceErr = recoveryErr + } else { + continue + } + } + restore, restoreErr := s.restoreSnapshotFromPeer(peerURL, "import_repair") if restore.Applied { result.UsedSnapshot = true @@ -171,9 +181,15 @@ func (s *Server) syncFromPeer(peerURL string, localHeight uint64, remoteHeight u result.SnapshotRestoreReason = restore.Reason } if restoreErr != nil { + if evidenceErr != nil { + return result, fmt.Errorf("certified block recovery failed: %v; snapshot recovery failed: %w", evidenceErr, restoreErr) + } return result, restoreErr } if !restore.Applied { + if evidenceErr != nil { + return result, fmt.Errorf("certified block recovery failed: %v; peer snapshot from %s is older than local state", evidenceErr, peerURL) + } return result, fmt.Errorf("peer snapshot from %s is older than local state", peerURL) } return result, nil diff --git a/internal/api/peer_transport.go b/internal/api/peer_transport.go index fcdbcaaa..f4194c46 100644 --- a/internal/api/peer_transport.go +++ b/internal/api/peer_transport.go @@ -23,6 +23,10 @@ type peerTransport interface { PostVote(peerURL string, vote consensus.Vote) error } +type certifiedBlockEvidenceTransport interface { + FetchBlockEvidence(peerURL string, height uint64) ([]ledger.CertifiedBlockEvidence, error) +} + type httpPeerTransport struct { client *http.Client sourceNode string @@ -67,6 +71,31 @@ func (t *httpPeerTransport) FetchBlock(peerURL string, height uint64) (ledger.Bl return payload.Block, nil } +func (t *httpPeerTransport) FetchBlockEvidence(peerURL string, height uint64) ([]ledger.CertifiedBlockEvidence, error) { + request, err := http.NewRequest(http.MethodGet, fmt.Sprintf("%s/v1/internal/block-evidence/%d", peerURL, height), nil) + if err != nil { + return nil, err + } + if err := t.applyPeerHeaders(request, nil); err != nil { + return nil, err + } + + response, err := t.client.Do(request) + if err != nil { + return nil, err + } + defer response.Body.Close() + if response.StatusCode != http.StatusOK { + return nil, fmt.Errorf("peer returned status %d", response.StatusCode) + } + + var payload BlockEvidenceResponse + if err := json.NewDecoder(response.Body).Decode(&payload); err != nil { + return nil, err + } + return payload.Evidence, nil +} + func (t *httpPeerTransport) FetchSnapshot(peerURL string) (ledger.Snapshot, error) { request, err := http.NewRequest(http.MethodGet, peerURL+"/v1/internal/snapshot", nil) if err != nil { diff --git a/internal/api/performance_lab_test.go b/internal/api/performance_lab_test.go new file mode 100644 index 00000000..f2b8fba6 --- /dev/null +++ b/internal/api/performance_lab_test.go @@ -0,0 +1,844 @@ +package api + +import ( + "bytes" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/x509" + "encoding/base64" + "encoding/json" + "fmt" + "io" + "math" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "sort" + "sync" + "testing" + "time" + + "github.com/zephyr-chain/zephyr-chain/internal/consensus" + "github.com/zephyr-chain/zephyr-chain/internal/dpos" + "github.com/zephyr-chain/zephyr-chain/internal/ledger" + "github.com/zephyr-chain/zephyr-chain/internal/protocol" + "github.com/zephyr-chain/zephyr-chain/internal/tx" +) + +const ( + labVotingPower = uint64(10_000) + labMaxTransactions = 4_096 + labConsensusRoundLimit = 1 * time.Second +) + +type labSigner struct { + privateKey *ecdsa.PrivateKey + address string + publicKey string +} + +type labNode struct { + server *Server + http *httptest.Server + signer labSigner + faults *labFaultTransport +} + +type labCluster struct { + tb testing.TB + nodes []*labNode +} + +type labFaultTransport struct { + base peerTransport + + mu sync.Mutex + blocked map[string]bool + delay time.Duration + duplicate bool + reorderConsensus bool + heldProposals map[string]consensus.Proposal + outboundPayloadLen uint64 +} + +func newLabCluster(tb testing.TB, validatorCount int) *labCluster { + tb.Helper() + if validatorCount <= 0 { + tb.Fatalf("validator count must be positive") + } + + cluster := &labCluster{tb: tb, nodes: make([]*labNode, 0, validatorCount)} + for index := 0; index < validatorCount; index++ { + signer := newLabSigner(tb) + server, err := NewServerWithConfig(Config{ + ChainID: protocol.DefaultChainID, + DataDir: tb.TempDir(), + NodeID: fmt.Sprintf("lab-validator-%02d", index+1), + ValidatorPrivateKey: encodeLabPrivateKey(tb, signer.privateKey), + BlockInterval: 0, + ConsensusInterval: 0, + ConsensusRoundTimeout: labConsensusRoundLimit, + SyncInterval: 0, + MaxTransactionsPerBlock: labMaxTransactions, + EnableBlockProduction: true, + EnableConsensusAutomation: false, + EnablePeerSync: false, + RequirePeerIdentity: false, + EnforceProposerSchedule: true, + RequireConsensusCertificates: true, + }) + if err != nil { + cluster.Close() + tb.Fatalf("create lab node %d: %v", index, err) + } + httpServer := httptest.NewServer(server.Handler()) + cluster.nodes = append(cluster.nodes, &labNode{server: server, http: httpServer, signer: signer}) + } + + validators := make([]dpos.Validator, 0, validatorCount) + for index, node := range cluster.nodes { + validators = append(validators, dpos.Validator{ + Rank: index + 1, + Address: node.signer.address, + VotingPower: labVotingPower, + SelfStake: labVotingPower, + DelegatedStake: 0, + }) + } + + for index, node := range cluster.nodes { + peers := make([]string, 0, validatorCount-1) + for peerIndex, peer := range cluster.nodes { + if peerIndex == index { + continue + } + peers = append(peers, peer.http.URL) + } + node.server.config.PeerURLs = peers + node.faults = newLabFaultTransport(node.server.transport) + node.server.transport = node.faults + if _, err := node.server.ledger.SetValidators(validators, dpos.ElectionConfig{ + MaxValidators: validatorCount, + MinSelfStake: 1, + MaxMissedBlocks: 100, + }); err != nil { + cluster.Close() + tb.Fatalf("set validator snapshot on node %d: %v", index, err) + } + view := node.server.ledger.Consensus() + expectedTotal := uint64(validatorCount) * labVotingPower + expectedQuorum := (expectedTotal/3)*2 + ((expectedTotal%3)*2)/3 + 1 + if view.ValidatorCount != validatorCount || view.TotalVotingPower != expectedTotal || view.QuorumVotingPower != expectedQuorum { + cluster.Close() + tb.Fatalf("unexpected validator quorum on node %d: count=%d total=%d quorum=%d, expected count=%d total=%d quorum=%d", index, view.ValidatorCount, view.TotalVotingPower, view.QuorumVotingPower, validatorCount, expectedTotal, expectedQuorum) + } + } + + return cluster +} + +func (c *labCluster) Close() { + if c == nil { + return + } + for _, node := range c.nodes { + if node == nil { + continue + } + if node.http != nil { + node.http.Close() + node.http = nil + } + if node.server != nil { + node.server.Close() + node.server = nil + } + } +} + +func (c *labCluster) prepareTransactions(count int) []tx.Envelope { + c.tb.Helper() + transactions := make([]tx.Envelope, 0, count) + for index := 0; index < count; index++ { + signer := newLabSigner(c.tb) + envelope := tx.Envelope{ + ChainID: protocol.DefaultChainID, + Domain: protocol.TransactionDomain, + From: signer.address, + To: "zph_lab_receiver", + Amount: 1, + Nonce: 1, + Memo: fmt.Sprintf("lab-%06d", index), + PublicKey: signer.publicKey, + } + envelope.Payload = envelope.CanonicalPayload() + signature, err := tx.SignPayload(signer.privateKey, envelope.Payload) + if err != nil { + c.tb.Fatalf("sign lab transaction %d: %v", index, err) + } + envelope.Signature = signature + transactions = append(transactions, envelope) + } + return transactions +} + +func (c *labCluster) fundTransactions(transactions []tx.Envelope) { + c.tb.Helper() + for _, node := range c.nodes { + for _, envelope := range transactions { + if _, err := node.server.ledger.Credit(envelope.From, envelope.Amount); err != nil { + c.tb.Fatalf("fund %s on %s: %v", envelope.From, node.server.nodeID, err) + } + } + } +} + +func (c *labCluster) submitTransactions(transactions []tx.Envelope, workers int) { + c.tb.Helper() + if len(transactions) == 0 { + return + } + if workers <= 0 { + workers = 1 + } + if workers > len(transactions) { + workers = len(transactions) + } + + jobs := make(chan tx.Envelope) + errs := make(chan error, len(transactions)) + var wg sync.WaitGroup + for worker := 0; worker < workers; worker++ { + wg.Add(1) + go func() { + defer wg.Done() + for envelope := range jobs { + body, err := json.Marshal(envelope) + if err != nil { + errs <- err + continue + } + response, err := http.Post(c.nodes[0].http.URL+"/v1/transactions", "application/json", bytes.NewReader(body)) + if err != nil { + errs <- err + continue + } + _, _ = io.Copy(io.Discard, response.Body) + _ = response.Body.Close() + if response.StatusCode != http.StatusAccepted { + errs <- fmt.Errorf("transaction ingress returned %d", response.StatusCode) + } + } + }() + } + for _, envelope := range transactions { + jobs <- envelope + } + close(jobs) + wg.Wait() + close(errs) + for err := range errs { + if err != nil { + c.tb.Fatalf("submit lab transaction: %v", err) + } + } +} + +func (c *labCluster) waitForMempools(size int, timeout time.Duration) { + c.tb.Helper() + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + ready := true + for _, node := range c.nodes { + if node.server.ledger.MempoolSize() != size { + ready = false + break + } + } + if ready { + return + } + time.Sleep(5 * time.Millisecond) + } + counts := make([]int, 0, len(c.nodes)) + for _, node := range c.nodes { + counts = append(counts, node.server.ledger.MempoolSize()) + } + c.tb.Fatalf("mempools did not converge to %d before timeout: %v", size, counts) +} + +func (c *labCluster) driveUntilHeight(indices []int, height uint64, timeout time.Duration) time.Duration { + c.tb.Helper() + startedAt := time.Now() + deadline := startedAt.Add(timeout) + for time.Now().Before(deadline) { + for _, node := range c.nodes { + if err := node.server.runConsensusAutomation(); err != nil && !ignoreConsensusAutomationError(err) { + c.tb.Fatalf("drive consensus on %s: %v", node.server.nodeID, err) + } + } + if c.indicesAtHeight(indices, height) { + return time.Since(startedAt) + } + time.Sleep(2 * time.Millisecond) + } + summaries := make([]string, 0, len(c.nodes)) + for index, node := range c.nodes { + status := node.server.ledger.Status() + view := node.server.ledger.Consensus() + round := node.server.ledger.RoundState() + proposals := node.server.ledger.ProposalsForHeight(view.NextHeight) + certificates := node.server.ledger.CertificatesForHeight(view.NextHeight) + tallies := node.server.ledger.VoteTalliesAt(view.NextHeight, view.CurrentRound) + summaries = append(summaries, fmt.Sprintf("node=%d height=%d mempool=%d next=%d round=%d roundHeight=%d proposer=%s proposals=%d tallies=%+v certs=%d", index, status.Height, status.MempoolSize, view.NextHeight, view.CurrentRound, round.Height, view.NextProposer, len(proposals), tallies, len(certificates))) + } + c.tb.Fatalf("target height %d not reached before timeout; consensus=%v", height, summaries) + return 0 +} + +func (c *labCluster) driveAndSyncUntilHeight(indices []int, height uint64, timeout time.Duration) { + c.tb.Helper() + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + for _, node := range c.nodes { + if err := node.server.runConsensusAutomation(); err != nil && !ignoreConsensusAutomationError(err) { + c.tb.Fatalf("drive consensus on %s: %v", node.server.nodeID, err) + } + } + for _, node := range c.nodes { + node.server.syncPeers() + } + if c.indicesAtHeight(indices, height) { + return + } + time.Sleep(5 * time.Millisecond) + } + c.driveUntilHeight(indices, height, time.Millisecond) +} + +func (c *labCluster) driveFor(duration time.Duration) { + c.tb.Helper() + deadline := time.Now().Add(duration) + for time.Now().Before(deadline) { + for _, node := range c.nodes { + if err := node.server.runConsensusAutomation(); err != nil && !ignoreConsensusAutomationError(err) { + c.tb.Fatalf("drive consensus on %s: %v", node.server.nodeID, err) + } + } + time.Sleep(2 * time.Millisecond) + } +} + +func (c *labCluster) indicesAtHeight(indices []int, height uint64) bool { + for _, index := range indices { + if c.nodes[index].server.ledger.Status().Height < height { + return false + } + } + return true +} + +func (c *labCluster) allIndices() []int { + indices := make([]int, len(c.nodes)) + for index := range c.nodes { + indices[index] = index + } + return indices +} + +func (c *labCluster) assertSameTip(indices []int, height uint64) { + c.tb.Helper() + var expected string + for _, index := range indices { + status := c.nodes[index].server.ledger.Status() + if status.Height != height { + c.tb.Fatalf("node %d expected height %d, got %d", index, height, status.Height) + } + if expected == "" { + expected = status.LatestBlockHash + continue + } + if status.LatestBlockHash != expected { + c.tb.Fatalf("safety violation: node %d tip %s differs from %s", index, status.LatestBlockHash, expected) + } + } +} + +func (c *labCluster) latestBlock(index int) ledger.Block { + c.tb.Helper() + response, err := http.Get(c.nodes[index].http.URL + "/v1/blocks/latest") + if err != nil { + c.tb.Fatalf("fetch latest block: %v", err) + } + defer response.Body.Close() + if response.StatusCode != http.StatusOK { + c.tb.Fatalf("fetch latest block returned %d", response.StatusCode) + } + var payload LatestBlockResponse + if err := json.NewDecoder(response.Body).Decode(&payload); err != nil { + c.tb.Fatalf("decode latest block: %v", err) + } + return payload.Block +} + +func (c *labCluster) partition(groups ...[]int) { + c.tb.Helper() + membership := make(map[int]int, len(c.nodes)) + for groupIndex, group := range groups { + for _, nodeIndex := range group { + membership[nodeIndex] = groupIndex + } + } + for nodeIndex, node := range c.nodes { + node.faults.clearBlocked() + groupIndex, grouped := membership[nodeIndex] + if !grouped { + continue + } + for peerIndex, peer := range c.nodes { + if peerIndex == nodeIndex { + continue + } + peerGroup, peerGrouped := membership[peerIndex] + if peerGrouped && peerGroup != groupIndex { + node.faults.blockPeer(peer.http.URL) + } + } + } +} + +func (c *labCluster) heal() { + for _, node := range c.nodes { + node.faults.clearBlocked() + } +} + +func (c *labCluster) enableConsensusFaults(delay time.Duration, duplicate bool, reorder bool) { + for _, node := range c.nodes { + node.faults.setBehavior(delay, duplicate, reorder) + } +} + +func (c *labCluster) outboundPayloadBytes() uint64 { + var total uint64 + for _, node := range c.nodes { + total += node.faults.payloadBytes() + } + return total +} + +func (c *labCluster) averageStateBytes() float64 { + var total int64 + for _, node := range c.nodes { + info, err := os.Stat(filepath.Join(node.server.ledger.DataDir(), "state.json")) + if err != nil { + c.tb.Fatalf("stat node state: %v", err) + } + total += info.Size() + } + return float64(total) / float64(len(c.nodes)) +} + +func newLabSigner(tb testing.TB) labSigner { + tb.Helper() + privateKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + tb.Fatalf("generate P-256 key: %v", err) + } + publicKeyBytes, err := x509.MarshalPKIXPublicKey(&privateKey.PublicKey) + if err != nil { + tb.Fatalf("marshal public key: %v", err) + } + publicKey := base64.StdEncoding.EncodeToString(publicKeyBytes) + address, err := tx.DeriveAddressFromPublicKey(publicKey) + if err != nil { + tb.Fatalf("derive Zephyr address: %v", err) + } + return labSigner{privateKey: privateKey, address: address, publicKey: publicKey} +} + +func encodeLabPrivateKey(tb testing.TB, privateKey *ecdsa.PrivateKey) string { + tb.Helper() + encoded, err := x509.MarshalPKCS8PrivateKey(privateKey) + if err != nil { + tb.Fatalf("marshal private key: %v", err) + } + return base64.StdEncoding.EncodeToString(encoded) +} + +func newLabFaultTransport(base peerTransport) *labFaultTransport { + return &labFaultTransport{ + base: base, + blocked: make(map[string]bool), + heldProposals: make(map[string]consensus.Proposal), + } +} + +func (t *labFaultTransport) blockPeer(peerURL string) { + t.mu.Lock() + defer t.mu.Unlock() + t.blocked[peerURL] = true +} + +func (t *labFaultTransport) clearBlocked() { + t.mu.Lock() + defer t.mu.Unlock() + clear(t.blocked) +} + +func (t *labFaultTransport) setBehavior(delay time.Duration, duplicate bool, reorder bool) { + t.mu.Lock() + defer t.mu.Unlock() + t.delay = delay + t.duplicate = duplicate + t.reorderConsensus = reorder + if !reorder { + clear(t.heldProposals) + } +} + +func (t *labFaultTransport) payloadBytes() uint64 { + t.mu.Lock() + defer t.mu.Unlock() + return t.outboundPayloadLen +} + +func (t *labFaultTransport) before(peerURL string) error { + t.mu.Lock() + blocked := t.blocked[peerURL] + delay := t.delay + t.mu.Unlock() + if blocked { + return fmt.Errorf("lab partition blocks %s", peerURL) + } + if delay > 0 { + time.Sleep(delay) + } + return nil +} + +func (t *labFaultTransport) recordPayload(payload any) { + encoded, err := json.Marshal(payload) + if err != nil { + return + } + t.mu.Lock() + t.outboundPayloadLen += uint64(len(encoded)) + t.mu.Unlock() +} + +func (t *labFaultTransport) duplicateEnabled() bool { + t.mu.Lock() + defer t.mu.Unlock() + return t.duplicate +} + +func (t *labFaultTransport) FetchStatus(peerURL string) (StatusResponse, error) { + if err := t.before(peerURL); err != nil { + return StatusResponse{}, err + } + return t.base.FetchStatus(peerURL) +} + +func (t *labFaultTransport) FetchBlock(peerURL string, height uint64) (ledger.Block, error) { + if err := t.before(peerURL); err != nil { + return ledger.Block{}, err + } + return t.base.FetchBlock(peerURL, height) +} + +func (t *labFaultTransport) FetchBlockEvidence(peerURL string, height uint64) ([]ledger.CertifiedBlockEvidence, error) { + if err := t.before(peerURL); err != nil { + return nil, err + } + evidenceTransport, ok := t.base.(certifiedBlockEvidenceTransport) + if !ok { + return nil, fmt.Errorf("lab base transport does not support certified block evidence") + } + return evidenceTransport.FetchBlockEvidence(peerURL, height) +} + +func (t *labFaultTransport) FetchSnapshot(peerURL string) (ledger.Snapshot, error) { + if err := t.before(peerURL); err != nil { + return ledger.Snapshot{}, err + } + return t.base.FetchSnapshot(peerURL) +} + +func (t *labFaultTransport) PostTransaction(peerURL string, envelope tx.Envelope) error { + if err := t.before(peerURL); err != nil { + return err + } + t.recordPayload(envelope) + if err := t.base.PostTransaction(peerURL, envelope); err != nil { + return err + } + if t.duplicateEnabled() { + _ = t.base.PostTransaction(peerURL, envelope) + } + return nil +} + +func (t *labFaultTransport) PostBlock(peerURL string, block ledger.Block) error { + if err := t.before(peerURL); err != nil { + return err + } + t.recordPayload(block) + if err := t.base.PostBlock(peerURL, block); err != nil { + return err + } + if t.duplicateEnabled() { + _ = t.base.PostBlock(peerURL, block) + } + return nil +} + +func (t *labFaultTransport) PostFaucet(peerURL string, request FaucetRequest) error { + if err := t.before(peerURL); err != nil { + return err + } + t.recordPayload(request) + return t.base.PostFaucet(peerURL, request) +} + +func (t *labFaultTransport) PostProposal(peerURL string, proposal consensus.Proposal) error { + if err := t.before(peerURL); err != nil { + return err + } + t.recordPayload(proposal) + t.mu.Lock() + reorder := t.reorderConsensus + if reorder { + t.heldProposals[peerURL] = proposal + } + t.mu.Unlock() + if reorder { + return nil + } + if err := t.base.PostProposal(peerURL, proposal); err != nil { + return err + } + if t.duplicateEnabled() { + _ = t.base.PostProposal(peerURL, proposal) + } + return nil +} + +func (t *labFaultTransport) PostVote(peerURL string, vote consensus.Vote) error { + if err := t.before(peerURL); err != nil { + return err + } + t.recordPayload(vote) + + t.mu.Lock() + proposal, hasProposal := t.heldProposals[peerURL] + if hasProposal { + delete(t.heldProposals, peerURL) + } + t.mu.Unlock() + + if hasProposal { + _ = t.base.PostVote(peerURL, vote) + if err := t.base.PostProposal(peerURL, proposal); err != nil { + return err + } + return nil + } + if err := t.base.PostVote(peerURL, vote); err != nil { + return err + } + if t.duplicateEnabled() { + _ = t.base.PostVote(peerURL, vote) + } + return nil +} + +func TestLabSevenValidatorsCertifiedFinality(t *testing.T) { + cluster := newLabCluster(t, 7) + defer cluster.Close() + + transactions := cluster.prepareTransactions(24) + cluster.fundTransactions(transactions) + startedAt := time.Now() + cluster.submitTransactions(transactions, 12) + cluster.waitForMempools(len(transactions), 3*time.Second) + cluster.driveUntilHeight(cluster.allIndices(), 1, 5*time.Second) + finality := time.Since(startedAt) + + cluster.assertSameTip(cluster.allIndices(), 1) + block := cluster.latestBlock(0) + if block.TransactionCount != len(transactions) { + t.Fatalf("expected %d finalized transactions, got %d", len(transactions), block.TransactionCount) + } + t.Logf("7-validator certified finality: tx=%d finality=%s finalized_tps=%.2f", len(transactions), finality, float64(len(transactions))/finality.Seconds()) +} + +func TestLabSevenValidatorsStallWithoutQuorumThenRecoverWithPeerSync(t *testing.T) { + cluster := newLabCluster(t, 7) + defer cluster.Close() + + transactions := cluster.prepareTransactions(8) + cluster.fundTransactions(transactions) + cluster.submitTransactions(transactions, 8) + cluster.waitForMempools(len(transactions), 3*time.Second) + + cluster.partition([]int{0, 1, 2, 3}, []int{4, 5, 6}) + cluster.driveFor(4 * labConsensusRoundLimit) + for index, node := range cluster.nodes { + if height := node.server.ledger.Status().Height; height != 0 { + t.Fatalf("node %d committed without 2/3+ quorum: height=%d", index, height) + } + } + + cluster.heal() + cluster.driveAndSyncUntilHeight(cluster.allIndices(), 1, 8*time.Second) + cluster.assertSameTip(cluster.allIndices(), 1) +} + +func TestLabSevenValidatorsMinorityPartitionRecoversFromCommittedMajority(t *testing.T) { + cluster := newLabCluster(t, 7) + defer cluster.Close() + + transactions := cluster.prepareTransactions(8) + cluster.fundTransactions(transactions) + cluster.submitTransactions(transactions, 8) + cluster.waitForMempools(len(transactions), 3*time.Second) + + majority := []int{0, 1, 2, 3, 4} + minority := []int{5, 6} + cluster.partition(majority, minority) + cluster.driveUntilHeight(majority, 1, 5*time.Second) + cluster.assertSameTip(majority, 1) + for _, index := range minority { + if height := cluster.nodes[index].server.ledger.Status().Height; height != 0 { + t.Fatalf("minority node %d unexpectedly committed during partition: height=%d", index, height) + } + } + + cluster.heal() + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) && !cluster.indicesAtHeight(minority, 1) { + for _, index := range minority { + cluster.nodes[index].server.syncPeers() + } + time.Sleep(20 * time.Millisecond) + } + if !cluster.indicesAtHeight(minority, 1) { + t.Fatalf("minority did not recover after partition heal") + } + cluster.assertSameTip(cluster.allIndices(), 1) +} + +func TestLabFourValidatorsDuplicateDelayedOutOfOrderMessagesPreserveSafety(t *testing.T) { + cluster := newLabCluster(t, 4) + defer cluster.Close() + cluster.enableConsensusFaults(2*time.Millisecond, true, true) + + transactions := cluster.prepareTransactions(8) + cluster.fundTransactions(transactions) + cluster.submitTransactions(transactions, 8) + cluster.waitForMempools(len(transactions), 3*time.Second) + cluster.driveUntilHeight(cluster.allIndices(), 1, 6*time.Second) + cluster.assertSameTip(cluster.allIndices(), 1) +} + +func BenchmarkLabConsensusFinality7Validators(b *testing.B) { + const transactionsPerBlock = 32 + cluster := newLabCluster(b, 7) + defer cluster.Close() + + finalitySamples := make([]time.Duration, 0, b.N) + var totalPayloadBytes uint64 + var totalBlockBytes uint64 + var totalStateBytes float64 + + for iteration := 0; iteration < b.N; iteration++ { + b.StopTimer() + transactions := cluster.prepareTransactions(transactionsPerBlock) + cluster.fundTransactions(transactions) + payloadBefore := cluster.outboundPayloadBytes() + + b.StartTimer() + startedAt := time.Now() + cluster.submitTransactions(transactions, 16) + cluster.waitForMempools(len(transactions), 3*time.Second) + cluster.driveUntilHeight(cluster.allIndices(), uint64(iteration+1), 5*time.Second) + finality := time.Since(startedAt) + b.StopTimer() + + cluster.assertSameTip(cluster.allIndices(), uint64(iteration+1)) + block := cluster.latestBlock(0) + if block.TransactionCount != transactionsPerBlock { + b.Fatalf("iteration %d: expected %d finalized transactions, got %d", iteration, transactionsPerBlock, block.TransactionCount) + } + encodedBlock, err := json.Marshal(block) + if err != nil { + b.Fatalf("marshal finalized block: %v", err) + } + finalitySamples = append(finalitySamples, finality) + totalPayloadBytes += cluster.outboundPayloadBytes() - payloadBefore + totalBlockBytes += uint64(len(encodedBlock)) + totalStateBytes += cluster.averageStateBytes() + } + + if len(finalitySamples) == 0 { + return + } + var totalFinality time.Duration + for _, sample := range finalitySamples { + totalFinality += sample + } + finalizedTransactions := float64(transactionsPerBlock * len(finalitySamples)) + b.ReportMetric(finalizedTransactions/totalFinality.Seconds(), "finalized-tx/s") + b.ReportMetric(durationMillis(percentileDuration(finalitySamples, 0.50)), "finality-p50-ms") + b.ReportMetric(durationMillis(percentileDuration(finalitySamples, 0.95)), "finality-p95-ms") + b.ReportMetric(durationMillis(percentileDuration(finalitySamples, 0.99)), "finality-p99-ms") + b.ReportMetric(float64(totalPayloadBytes)/finalizedTransactions, "protocol-payload-B/finalized-tx") + b.ReportMetric(float64(totalBlockBytes)/float64(len(finalitySamples)), "finalized-block-B") + b.ReportMetric(totalStateBytes/float64(len(finalitySamples)), "state-B/node") +} + +func BenchmarkLabP256TransactionVerification(b *testing.B) { + signer := newLabSigner(b) + envelope := tx.Envelope{ + ChainID: protocol.DefaultChainID, Domain: protocol.TransactionDomain, + From: signer.address, To: "zph_lab_receiver", Amount: 1, Nonce: 1, Memo: "verify", PublicKey: signer.publicKey, + } + envelope.Payload = envelope.CanonicalPayload() + var err error + envelope.Signature, err = tx.SignPayload(signer.privateKey, envelope.Payload) + if err != nil { + b.Fatal(err) + } + + b.ResetTimer() + for iteration := 0; iteration < b.N; iteration++ { + if err := envelope.ValidateForChain(protocol.DefaultChainID); err != nil { + b.Fatal(err) + } + } +} + +func percentileDuration(samples []time.Duration, percentile float64) time.Duration { + if len(samples) == 0 { + return 0 + } + ordered := append([]time.Duration(nil), samples...) + sort.Slice(ordered, func(i, j int) bool { return ordered[i] < ordered[j] }) + index := int(math.Ceil(percentile*float64(len(ordered)))) - 1 + if index < 0 { + index = 0 + } + if index >= len(ordered) { + index = len(ordered) - 1 + } + return ordered[index] +} + +func durationMillis(value time.Duration) float64 { + return float64(value) / float64(time.Millisecond) +} + +var _ peerTransport = (*labFaultTransport)(nil) diff --git a/internal/api/server.go b/internal/api/server.go index 7706a221..3334f3b8 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -316,6 +316,7 @@ func (s *Server) routes() { s.mux.HandleFunc("/v1/dev/block-template", s.handleBlockTemplate) s.mux.HandleFunc("/v1/dev/produce-block", s.handleProduceBlock) s.mux.HandleFunc("/v1/internal/blocks", s.handleImportBlock) + s.mux.HandleFunc("/v1/internal/block-evidence/", s.handleBlockEvidence) s.mux.HandleFunc("/v1/internal/snapshot", s.handleSnapshot) } diff --git a/internal/ledger/certified_import.go b/internal/ledger/certified_import.go new file mode 100644 index 00000000..ef917cef --- /dev/null +++ b/internal/ledger/certified_import.go @@ -0,0 +1,276 @@ +package ledger + +import ( + "errors" + "sort" + "time" + + "github.com/zephyr-chain/zephyr-chain/internal/consensus" +) + +var ( + ErrCertifiedEvidenceInvalid = errors.New("invalid certified block evidence") + ErrCertifiedEvidenceQuorum = errors.New("certified block evidence does not reach quorum") +) + +// CertifiedBlockEvidence carries signed consensus artifacts that can be used +// to independently prove that a block reached validator quorum. Sources may +// expose partial evidence; only the receiving node decides whether the merged +// signatures reach quorum. +type CertifiedBlockEvidence struct { + Proposal consensus.Proposal `json:"proposal"` + Votes []consensus.Vote `json:"votes"` +} + +// CertifiedBlockEvidenceFragments returns valid proposal/vote fragments the +// node has retained for a height, even when it has not materialized the block +// locally yet. This is important during partition recovery: a validator may +// have cast a durable vote that contributed to finality without receiving the +// eventual block commit before the network split/healed. +func (s *Store) CertifiedBlockEvidenceFragments(height uint64) []CertifiedBlockEvidence { + s.mu.RLock() + defer s.mu.RUnlock() + return certifiedBlockEvidenceFragmentsFromState(s.snapshotLocked(), s.chainID, height) +} + +// CertifiedBlockEvidenceAt keeps the committed-block-oriented collector used +// by tests and callers that want the evidence matching a local committed block. +func (s *Store) CertifiedBlockEvidenceAt(height uint64) (CertifiedBlockEvidence, bool) { + s.mu.RLock() + defer s.mu.RUnlock() + + if height == 0 || height > uint64(len(s.blocks)) { + return CertifiedBlockEvidence{}, false + } + state := s.snapshotLocked() + block := state.Blocks[height-1] + for _, fragment := range certifiedBlockEvidenceFragmentsFromState(state, s.chainID, height) { + proposal := fragment.Proposal + if proposal.BlockHash != block.Hash || proposal.StateRoot != block.StateRoot { + continue + } + if matchProposalForBlock([]consensus.Proposal{proposal}, block) != nil { + return fragment, true + } + } + return CertifiedBlockEvidence{}, false +} + +func certifiedBlockEvidenceFragmentsFromState(state persistedState, chainID string, height uint64) []CertifiedBlockEvidence { + state = normalizeState(state) + proposals := proposalsForHeight(state.Proposals, height) + fragments := make([]CertifiedBlockEvidence, 0, len(proposals)) + for _, proposal := range proposals { + if proposal.ValidateForChain(chainID) != nil { + continue + } + if _, ok := validatorVotingPower(state.ValidatorSnapshot, proposal.Proposer); !ok { + continue + } + + seen := make(map[string]struct{}) + votes := make([]consensus.Vote, 0) + for _, record := range state.Votes { + vote := record.Vote + if vote.Height != proposal.Height || vote.Round != proposal.Round || vote.BlockHash != proposal.BlockHash { + continue + } + if _, duplicate := seen[vote.Voter]; duplicate { + continue + } + if _, ok := validatorVotingPower(state.ValidatorSnapshot, vote.Voter); !ok { + continue + } + if vote.ValidateForChain(chainID) != nil { + continue + } + seen[vote.Voter] = struct{}{} + votes = append(votes, vote) + } + if len(votes) == 0 { + continue + } + sort.Slice(votes, func(i, j int) bool { return votes[i].Voter < votes[j].Voter }) + fragments = append(fragments, CertifiedBlockEvidence{ + Proposal: cloneProposal(proposal), + Votes: votes, + }) + } + + sort.Slice(fragments, func(i, j int) bool { + left := fragments[i].Proposal + right := fragments[j].Proposal + if left.Round != right.Round { + return left.Round < right.Round + } + if left.BlockHash != right.BlockHash { + return left.BlockHash < right.BlockHash + } + return left.Proposer < right.Proposer + }) + return fragments +} + +// ImportBlockWithEvidence imports the next block after independently +// validating a quorum of signed votes for its signed proposal. Matching valid +// votes already retained by the recovering node are combined with incoming +// signed evidence before quorum is evaluated. +func (s *Store) ImportBlockWithEvidence(block Block, evidence CertifiedBlockEvidence) error { + s.mu.Lock() + defer s.mu.Unlock() + + state := s.snapshotLocked() + if len(state.ValidatorSnapshot.Validators) == 0 { + return ErrNoValidatorSet + } + + nextState, err := importBlockIntoState(state, block, s.chainID) + if err != nil { + return err + } + + evidenceState, err := attachCertifiedBlockEvidence(state, block, evidence, s.chainID) + if err != nil { + return err + } + + nextState.Proposals = cloneProposals(evidenceState.Proposals) + nextState.Votes = cloneVoteRecords(evidenceState.Votes) + nextState.CommitCertificates = cloneCommitCertificates(evidenceState.CommitCertificates) + nextState = completeConsensusActionsForHeightInState(nextState, block.Height, time.Now().UTC(), "certified block evidence imported") + if err := s.writeState(nextState); err != nil { + return err + } + s.applyStateLocked(nextState) + return nil +} + +func attachCertifiedBlockEvidence(state persistedState, block Block, evidence CertifiedBlockEvidence, chainID string) (persistedState, error) { + state = normalizeState(state) + proposal := cloneProposal(evidence.Proposal) + if err := proposal.ValidateForChain(chainID); err != nil { + return state, ErrCertifiedEvidenceInvalid + } + if proposal.Height != block.Height || proposal.BlockHash != block.Hash || proposal.StateRoot != block.StateRoot { + return state, ErrCertifiedEvidenceInvalid + } + if matchProposalForBlock([]consensus.Proposal{proposal}, block) == nil { + return state, ErrCertifiedEvidenceInvalid + } + if _, ok := validatorVotingPower(state.ValidatorSnapshot, proposal.Proposer); !ok { + return state, ErrCertifiedEvidenceInvalid + } + if expected := proposerForHeightRound(state.ValidatorSnapshot.Validators, block.Height, proposal.Round); expected == "" || proposal.Proposer != expected { + return state, ErrCertifiedEvidenceInvalid + } + + for _, existing := range state.Proposals { + if existing.Height != proposal.Height || existing.Round != proposal.Round { + continue + } + if existing.BlockHash != proposal.BlockHash || existing.Proposer != proposal.Proposer { + return state, ErrConflictingProposal + } + } + + quorum := quorumVotingPower(totalVotingPower(state.ValidatorSnapshot)) + if quorum == 0 { + return state, ErrCertifiedEvidenceQuorum + } + + seen := make(map[string]struct{}, len(state.Votes)+len(evidence.Votes)) + providedSeen := make(map[string]struct{}, len(evidence.Votes)) + voters := make([]string, 0, len(state.Votes)+len(evidence.Votes)) + validatedVotes := make([]VoteRecord, 0, len(evidence.Votes)) + var signedPower uint64 + + for _, record := range state.Votes { + vote := record.Vote + if vote.Height != proposal.Height || vote.Round != proposal.Round || vote.BlockHash != proposal.BlockHash { + continue + } + if _, duplicate := seen[vote.Voter]; duplicate { + continue + } + if err := vote.ValidateForChain(chainID); err != nil { + return state, ErrCertifiedEvidenceInvalid + } + power, ok := validatorVotingPower(state.ValidatorSnapshot, vote.Voter) + if !ok { + return state, ErrCertifiedEvidenceInvalid + } + nextPower, ok := addUint64(signedPower, power) + if !ok { + return state, ErrVotingPowerOverflow + } + signedPower = nextPower + seen[vote.Voter] = struct{}{} + voters = append(voters, vote.Voter) + } + + for _, vote := range evidence.Votes { + if _, duplicate := providedSeen[vote.Voter]; duplicate { + return state, ErrCertifiedEvidenceInvalid + } + providedSeen[vote.Voter] = struct{}{} + if err := vote.ValidateForChain(chainID); err != nil { + return state, ErrCertifiedEvidenceInvalid + } + if vote.Height != proposal.Height || vote.Round != proposal.Round || vote.BlockHash != proposal.BlockHash { + return state, ErrCertifiedEvidenceInvalid + } + power, ok := validatorVotingPower(state.ValidatorSnapshot, vote.Voter) + if !ok { + return state, ErrCertifiedEvidenceInvalid + } + if existingVote := findVoteByValidator(state.Votes, vote.Height, vote.Round, vote.Voter); existingVote != nil && existingVote.BlockHash != vote.BlockHash { + return state, ErrConflictingVote + } + if _, alreadyCounted := seen[vote.Voter]; alreadyCounted { + continue + } + nextPower, ok := addUint64(signedPower, power) + if !ok { + return state, ErrVotingPowerOverflow + } + signedPower = nextPower + seen[vote.Voter] = struct{}{} + voters = append(voters, vote.Voter) + validatedVotes = append(validatedVotes, VoteRecord{Vote: vote, VotingPower: power, RecordedAt: time.Now().UTC()}) + } + if signedPower < quorum { + return state, ErrCertifiedEvidenceQuorum + } + sort.Strings(voters) + + proposalPresent := false + for _, existing := range state.Proposals { + if existing.Height == proposal.Height && existing.Round == proposal.Round && existing.BlockHash == proposal.BlockHash && existing.Proposer == proposal.Proposer { + proposalPresent = true + break + } + } + if !proposalPresent { + state.Proposals = append(state.Proposals, proposal) + } + + for _, record := range validatedVotes { + if findVoteByValidator(state.Votes, record.Vote.Height, record.Vote.Round, record.Vote.Voter) == nil { + state.Votes = append(state.Votes, record) + } + } + + if findCertificate(state.CommitCertificates, proposal.Height, proposal.Round, proposal.BlockHash) == nil { + state.CommitCertificates = append(state.CommitCertificates, CommitCertificate{ + Height: proposal.Height, + Round: proposal.Round, + BlockHash: proposal.BlockHash, + VotingPower: signedPower, + QuorumVotingPower: quorum, + VoterCount: len(voters), + Voters: voters, + CreatedAt: time.Now().UTC(), + }) + } + return normalizeState(state), nil +} diff --git a/internal/ledger/certified_import_test.go b/internal/ledger/certified_import_test.go new file mode 100644 index 00000000..a983b861 --- /dev/null +++ b/internal/ledger/certified_import_test.go @@ -0,0 +1,227 @@ +package ledger + +import ( + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/x509" + "encoding/base64" + "errors" + "fmt" + "testing" + "time" + + "github.com/zephyr-chain/zephyr-chain/internal/consensus" + "github.com/zephyr-chain/zephyr-chain/internal/dpos" + "github.com/zephyr-chain/zephyr-chain/internal/protocol" + "github.com/zephyr-chain/zephyr-chain/internal/tx" +) + +type certifiedImportSigner struct { + privateKey *ecdsa.PrivateKey + publicKey string + address string +} + +func TestImportBlockWithEvidenceRequiresSignedQuorum(t *testing.T) { + validators := make([]dpos.Validator, 0, 7) + signers := make(map[string]certifiedImportSigner, 7) + for index := 0; index < 7; index++ { + signer := newCertifiedImportSigner(t) + signers[signer.address] = signer + validators = append(validators, dpos.Validator{ + Rank: index + 1, + Address: signer.address, + VotingPower: 10_000, + SelfStake: 10_000, + }) + } + config := dpos.ElectionConfig{MaxValidators: 7, MinSelfStake: 1, MaxMissedBlocks: 100} + + source, err := NewStoreWithChainID(t.TempDir(), protocol.DefaultChainID) + if err != nil { + t.Fatal(err) + } + if _, err := source.SetValidators(validators, config); err != nil { + t.Fatal(err) + } + + txSigner := newCertifiedImportSigner(t) + envelope := tx.Envelope{ + ChainID: protocol.DefaultChainID, + Domain: protocol.TransactionDomain, + From: txSigner.address, + To: "zph_certified_receiver", + Amount: 1, + Nonce: 1, + Memo: "certified catch-up", + PublicKey: txSigner.publicKey, + } + envelope.Payload = envelope.CanonicalPayload() + envelope.Signature, err = tx.SignPayload(txSigner.privateKey, envelope.Payload) + if err != nil { + t.Fatal(err) + } + if _, err := source.Credit(envelope.From, 10); err != nil { + t.Fatal(err) + } + if _, err := source.Accept(envelope); err != nil { + t.Fatal(err) + } + + producedAt := time.Now().UTC() + candidate, err := source.BuildNextBlock(100, producedAt) + if err != nil { + t.Fatal(err) + } + view := source.Consensus() + proposer, ok := signers[view.NextProposer] + if !ok { + t.Fatalf("scheduled proposer %s has no signer", view.NextProposer) + } + proposal := consensus.Proposal{ + ChainID: protocol.DefaultChainID, + Domain: protocol.ConsensusProposalDomain, + Height: candidate.Height, + Round: view.CurrentRound, + BlockHash: candidate.Hash, + PreviousHash: candidate.PreviousHash, + StateRoot: candidate.StateRoot, + ProducedAt: candidate.ProducedAt, + TransactionIDs: append([]string(nil), candidate.TransactionIDs...), + Transactions: append([]tx.Envelope(nil), candidate.Transactions...), + Proposer: proposer.address, + PublicKey: proposer.publicKey, + ProposedAt: time.Now().UTC(), + } + proposal.Payload = proposal.CanonicalPayload() + proposal.Signature, err = tx.SignPayload(proposer.privateKey, proposal.Payload) + if err != nil { + t.Fatal(err) + } + if err := source.RecordProposal(proposal); err != nil { + t.Fatal(err) + } + + for index := 0; index < 5; index++ { + validator := validators[index] + signer := signers[validator.Address] + vote := consensus.Vote{ + ChainID: protocol.DefaultChainID, + Domain: protocol.ConsensusVoteDomain, + Height: proposal.Height, + Round: proposal.Round, + BlockHash: proposal.BlockHash, + Voter: signer.address, + PublicKey: signer.publicKey, + VotedAt: time.Now().UTC(), + } + vote.Payload = vote.CanonicalPayload() + vote.Signature, err = tx.SignPayload(signer.privateKey, vote.Payload) + if err != nil { + t.Fatal(err) + } + if _, _, err := source.RecordVote(vote); err != nil { + t.Fatal(err) + } + } + + block, err := source.ProduceBlockWithOptions(100, candidate.ProducedAt, true) + if err != nil { + t.Fatal(err) + } + evidence, ok := source.CertifiedBlockEvidenceAt(block.Height) + if !ok { + t.Fatal("expected certified evidence for committed block") + } + if len(evidence.Votes) < 5 { + t.Fatalf("expected at least 5 signed votes, got %d", len(evidence.Votes)) + } + + newTarget := func() *Store { + t.Helper() + store, err := NewStoreWithChainID(t.TempDir(), protocol.DefaultChainID) + if err != nil { + t.Fatal(err) + } + if _, err := store.SetValidators(validators, config); err != nil { + t.Fatal(err) + } + if _, err := store.Credit(envelope.From, 10); err != nil { + t.Fatal(err) + } + if _, err := store.Accept(envelope); err != nil { + t.Fatal(err) + } + return store + } + + insufficient := newTarget() + fourVotes := evidence + fourVotes.Votes = append([]consensus.Vote(nil), evidence.Votes[:4]...) + if err := insufficient.ImportBlockWithEvidence(block, fourVotes); !errors.Is(err, ErrCertifiedEvidenceQuorum) { + t.Fatalf("expected insufficient evidence quorum, got %v", err) + } + if height := insufficient.Status().Height; height != 0 { + t.Fatalf("insufficient evidence mutated target height to %d", height) + } + + localPlusRemote := newTarget() + if err := localPlusRemote.RecordProposal(proposal); err != nil { + t.Fatalf("record local recovery proposal: %v", err) + } + if _, _, err := localPlusRemote.RecordVote(evidence.Votes[0]); err != nil { + t.Fatalf("record local recovery vote: %v", err) + } + fourRemoteVotes := evidence + fourRemoteVotes.Votes = append([]consensus.Vote(nil), evidence.Votes[1:5]...) + if err := localPlusRemote.ImportBlockWithEvidence(block, fourRemoteVotes); err != nil { + t.Fatalf("expected local vote plus four remote votes to reach quorum: %v", err) + } + if height := localPlusRemote.Status().Height; height != 1 { + t.Fatalf("expected local plus remote evidence to import height 1, got %d", height) + } + + tampered := newTarget() + badEvidence := evidence + badEvidence.Votes = append([]consensus.Vote(nil), evidence.Votes...) + badEvidence.Votes[0].Signature = base64.StdEncoding.EncodeToString(make([]byte, 64)) + if err := tampered.ImportBlockWithEvidence(block, badEvidence); !errors.Is(err, ErrCertifiedEvidenceInvalid) { + t.Fatalf("expected invalid evidence error, got %v", err) + } + if height := tampered.Status().Height; height != 0 { + t.Fatalf("invalid evidence mutated target height to %d", height) + } + + valid := newTarget() + if err := valid.ImportBlockWithEvidence(block, evidence); err != nil { + t.Fatalf("import certified block: %v", err) + } + if height := valid.Status().Height; height != 1 { + t.Fatalf("expected imported height 1, got %d", height) + } + if latest, ok := valid.LatestBlock(); !ok || latest.Hash != block.Hash { + t.Fatalf("expected imported block %s, got %+v", block.Hash, latest) + } +} + +func newCertifiedImportSigner(t *testing.T) certifiedImportSigner { + t.Helper() + privateKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatal(err) + } + publicBytes, err := x509.MarshalPKIXPublicKey(&privateKey.PublicKey) + if err != nil { + t.Fatal(err) + } + publicKey := base64.StdEncoding.EncodeToString(publicBytes) + address, err := tx.DeriveAddressFromPublicKey(publicKey) + if err != nil { + t.Fatal(err) + } + if address == "" { + t.Fatal(fmt.Errorf("derived empty validator address")) + } + return certifiedImportSigner{privateKey: privateKey, publicKey: publicKey, address: address} +}