From 4f8bf86438d27c152cba61a67483699134b01fcc Mon Sep 17 00:00:00 2001 From: 7layermagik <7layermagik@users.noreply.github.com> Date: Thu, 30 Jul 2026 00:47:56 -0500 Subject: [PATCH 1/9] fix(blockprod): target two complete FEC batches --- pkg/blockprod/bank_test.go | 13 +++++++++++++ pkg/costmodel/limits.go | 16 +++++++++++----- pkg/costmodel/limits_test.go | 13 +++++++++++++ 3 files changed, 37 insertions(+), 5 deletions(-) create mode 100644 pkg/costmodel/limits_test.go diff --git a/pkg/blockprod/bank_test.go b/pkg/blockprod/bank_test.go index fa219dde..4e793f15 100644 --- a/pkg/blockprod/bank_test.go +++ b/pkg/blockprod/bank_test.go @@ -455,6 +455,19 @@ func TestEntryBuilderFlush(t *testing.T) { assert.Greater(t, batchBytes, 0) } +func TestEntryBuilderDefaultTargetCoalescesTransactions(t *testing.T) { + builder := NewEntryBuilder(costmodel.DefaultLimits(), solana.Hash{0xcd}) + for seq := uint64(0); seq < 2; seq++ { + wire := txfixture.MustSignedTransferWire(seq) + tx, err := solana.TransactionFromBytes(wire) + require.NoError(t, err) + entries, _, flushed := builder.Append(*tx, len(wire)) + assert.False(t, flushed) + assert.Empty(t, entries) + } + assert.Equal(t, 2, builder.PendingCount()) +} + func TestControllerWorkingBank(t *testing.T) { controller := NewController() assert.Nil(t, controller.WorkingBank()) diff --git a/pkg/costmodel/limits.go b/pkg/costmodel/limits.go index fd73c275..bdbe4190 100644 --- a/pkg/costmodel/limits.go +++ b/pkg/costmodel/limits.go @@ -4,12 +4,12 @@ package costmodel const ( ComputeUnitToUSRatio = 30 - SignatureCost = ComputeUnitToUSRatio * 24 // 720 + SignatureCost = ComputeUnitToUSRatio * 24 // 720 Secp256k1VerifyCost = ComputeUnitToUSRatio * 223 Ed25519VerifyStrictCost = ComputeUnitToUSRatio * 80 Secp256r1VerifyCost = ComputeUnitToUSRatio * 160 - WriteLockUnits = ComputeUnitToUSRatio * 10 // 300 - InstructionDataBytesCost = 140 / ComputeUnitToUSRatio // ~4 CU per byte + WriteLockUnits = ComputeUnitToUSRatio * 10 // 300 + InstructionDataBytesCost = 140 / ComputeUnitToUSRatio // ~4 CU per byte MaxBlockUnitsSIMD0256 = 60_000_000 MaxBlockUnitsSIMD0286 = 100_000_000 @@ -20,9 +20,15 @@ const ( // DefaultMaxDataShredsPerSlot matches agave DEFAULT_MAX_DATA_SHREDS_PER_SLOT. DefaultMaxDataShredsPerSlot = 32 * 1024 - // TypicalDataShredPayloadBytes is the usable data bytes per entry-batch target. + // TypicalDataShredPayloadBytes is the usable data in one chained Merkle + // data shred for the standard 32+32 FEC layout. TypicalDataShredPayloadBytes = 963 - DefaultTargetBatchBytes = 2 * TypicalDataShredPayloadBytes + // DataShredsPerFECBlock and DefaultTargetBatchBytes mirror Agave's + // DATA_SHREDS_PER_FEC_BLOCK and get_target_batch_bytes_default. The target + // is two complete FEC payloads, not two individual shred payloads. + DataShredsPerFECBlock = 32 + TypicalFECDataBytes = DataShredsPerFECBlock * TypicalDataShredPayloadBytes + DefaultTargetBatchBytes = 2 * TypicalFECDataBytes ) // Limits configures per-slot cost and size budgets. diff --git a/pkg/costmodel/limits_test.go b/pkg/costmodel/limits_test.go new file mode 100644 index 00000000..e9d32ab2 --- /dev/null +++ b/pkg/costmodel/limits_test.go @@ -0,0 +1,13 @@ +package costmodel + +import "testing" + +func TestDefaultTargetBatchBytesMatchesTwoTypicalFECSets(t *testing.T) { + const want = 61_632 + if DefaultTargetBatchBytes != want { + t.Fatalf("DefaultTargetBatchBytes = %d, want %d", DefaultTargetBatchBytes, want) + } + if DefaultTargetBatchBytes != 2*DataShredsPerFECBlock*TypicalDataShredPayloadBytes { + t.Fatal("default batch target must remain two complete typical FEC payloads") + } +} From fe9851f7c7fd5c7360bf5e129038951eb8274b83 Mon Sep 17 00:00:00 2001 From: 7layermagik <7layermagik@users.noreply.github.com> Date: Thu, 30 Jul 2026 00:56:00 -0500 Subject: [PATCH 2/9] perf(turbine): eliminate generated shred copy churn --- pkg/turbine/generate.go | 182 +++++++++++++--------- pkg/turbine/generate_bench_test.go | 242 +++++++++++++++++++++++++++++ pkg/turbine/shred.go | 5 +- 3 files changed, 356 insertions(+), 73 deletions(-) create mode 100644 pkg/turbine/generate_bench_test.go diff --git a/pkg/turbine/generate.go b/pkg/turbine/generate.go index 14626e21..3eb29c0a 100644 --- a/pkg/turbine/generate.go +++ b/pkg/turbine/generate.go @@ -4,6 +4,7 @@ import ( "crypto/ed25519" "encoding/binary" "fmt" + "sync" "github.com/gagliardetto/solana-go" "github.com/klauspost/reedsolomon" @@ -14,6 +15,21 @@ const ( proofEntriesFor32x32 = 6 ) +var erasureEncoderPool sync.Pool + +func acquireErasureEncoder() (reedsolomon.Encoder, error) { + if encoder := erasureEncoderPool.Get(); encoder != nil { + return encoder.(reedsolomon.Encoder), nil + } + return reedsolomon.New(dataShredsPerFECBlock, codingShredsPerFECBlock) +} + +func releaseErasureEncoder(encoder reedsolomon.Encoder) { + if encoder != nil { + erasureEncoderPool.Put(encoder) + } +} + // ShredGenerator builds merkle FEC shreds from a serialized byte buffer. type ShredGenerator struct { Slot uint64 @@ -69,6 +85,14 @@ func (g *ShredGenerator) MakeShredsFromData( if g.Slot < g.ParentSlot || g.Slot-g.ParentSlot > uint64(^uint16(0)) { return nil, solana.Hash{}, nextShredIndex, nextCodeIndex, fmt.Errorf("invalid parent slot %d for slot %d", g.ParentSlot, g.Slot) } + // The 32+32 coding matrix is invariant across every FEC set in this + // operation. Building it requires a Vandermonde inversion, so retain the + // encoder for the whole payload rather than reconstructing it per set. + encoder, err := acquireErasureEncoder() + if err != nil { + return nil, solana.Hash{}, nextShredIndex, nextCodeIndex, err + } + defer releaseErasureEncoder(encoder) proofSize := uint8(proofEntriesFor32x32) unsignedCap := dataCapacity(proofSize, false) signedCap := dataCapacity(proofSize, true) @@ -99,7 +123,7 @@ func (g *ShredGenerator) MakeShredsFromData( for len(unsignedData) >= unsignedBatch { batch := unsignedData[:unsignedBatch] unsignedData = unsignedData[unsignedBatch:] - batchPackets, root, err := g.makeFECBatch(leader, batch, unsignedCap, proofSize, false, parentOffset, flags, false, chainedRoot, dataIndex, codeIndex) + batchPackets, root, err := g.makeFECBatch(encoder, leader, batch, unsignedCap, proofSize, false, parentOffset, flags, false, chainedRoot, dataIndex, codeIndex) if err != nil { return nil, solana.Hash{}, dataIndex, codeIndex, err } @@ -110,7 +134,7 @@ func (g *ShredGenerator) MakeShredsFromData( } if len(unsignedData) > 0 || (len(packets) == 0 && !isLastInSlot) { - batchPackets, root, err := g.makeFECBatch(leader, unsignedData, unsignedCap, proofSize, false, parentOffset, flags, false, chainedRoot, dataIndex, codeIndex) + batchPackets, root, err := g.makeFECBatch(encoder, leader, unsignedData, unsignedCap, proofSize, false, parentOffset, flags, false, chainedRoot, dataIndex, codeIndex) if err != nil { return nil, solana.Hash{}, dataIndex, codeIndex, err } @@ -121,7 +145,7 @@ func (g *ShredGenerator) MakeShredsFromData( } if len(signedData) > 0 || (len(packets) == 0 && isLastInSlot) { - batchPackets, root, err := g.makeFECBatch(leader, signedData, signedCap, proofSize, true, parentOffset, flags, isLastInSlot, chainedRoot, dataIndex, codeIndex) + batchPackets, root, err := g.makeFECBatch(encoder, leader, signedData, signedCap, proofSize, true, parentOffset, flags, isLastInSlot, chainedRoot, dataIndex, codeIndex) if err != nil { return nil, solana.Hash{}, dataIndex, codeIndex, err } @@ -135,6 +159,7 @@ func (g *ShredGenerator) MakeShredsFromData( } func (g *ShredGenerator) makeFECBatch( + encoder reedsolomon.Encoder, leader solana.PrivateKey, data []byte, dataCap int, @@ -200,7 +225,7 @@ func (g *ShredGenerator) makeFECBatch( dataPackets[len(dataPackets)-1][dataFlagsOffset] |= shredFlagDataComplete } - root, err := finishErasureBatch(leader, allPackets, chainedMerkleRoot, proofSize, resigned) + root, err := finishErasureBatch(encoder, leader, allPackets, chainedMerkleRoot, proofSize, resigned) if err != nil { return nil, solana.Hash{}, err } @@ -208,101 +233,70 @@ func (g *ShredGenerator) makeFECBatch( } func finishErasureBatch( + encoder reedsolomon.Encoder, leader solana.PrivateKey, packets [][]byte, chainedMerkleRoot solana.Hash, proofSize uint8, resigned bool, ) (solana.Hash, error) { - encoder, err := reedsolomon.New(dataShredsPerFECBlock, codingShredsPerFECBlock) + if len(packets) != dataShredsPerFECBlock+codingShredsPerFECBlock { + return solana.Hash{}, fmt.Errorf("invalid FEC packet count %d", len(packets)) + } + dataCap, err := merkleCapacity(dataPayloadSize, dataHeaderSize, proofSize, true, resigned) if err != nil { return solana.Hash{}, err } + codeCap, err := merkleCapacity(codingPayloadSize, codingHeaderSize, proofSize, true, resigned) + if err != nil { + return solana.Hash{}, err + } + dataVariant := chainedDataVariant(proofSize, resigned) + codeVariant := chainedCodeVariant(proofSize, resigned) + // These packets were constructed immediately above, so retain direct views + // of their erasure regions. ParseShred is intentionally a defensive, + // owning parser for untrusted network packets; using it here would allocate + // and copy every packet several times only to copy the same bytes back. shards := make([][]byte, len(packets)) for i, packet := range packets { - shred, err := ParseShred(packet) - if err != nil { - return solana.Hash{}, fmt.Errorf("parse batch shred %d: %w", i, err) + if i < dataShredsPerFECBlock { + if len(packet) < dataPayloadSize || packet[shredVariantOffset] != dataVariant { + return solana.Hash{}, fmt.Errorf("invalid generated data shred %d", i) + } + shards[i] = packet[shredSignatureSize : dataHeaderSize+dataCap] + continue } - shard, err := shred.erasureShard() - if err != nil { - return solana.Hash{}, fmt.Errorf("erasure shard %d: %w", i, err) + if len(packet) < codingPayloadSize || packet[shredVariantOffset] != codeVariant { + return solana.Hash{}, fmt.Errorf("invalid generated coding shred %d", i-dataShredsPerFECBlock) } - shards[i] = shard + shards[i] = packet[codingHeaderSize : codingHeaderSize+codeCap] } if err := encoder.Encode(shards); err != nil { return solana.Hash{}, fmt.Errorf("reed-solomon encode: %w", err) } for i, packet := range packets { - shred, err := ParseShred(packet) - if err != nil { - return solana.Hash{}, err - } - proofSizeInfo, chained, resignedFlag, ok := merkleVariantInfo(shred.Variant) - if !ok { - return solana.Hash{}, ErrUnsupportedShred - } - _ = proofSizeInfo - _ = chained - _ = resignedFlag - - capacity, err := merkleCapacity(len(packet), dataHeaderSize, proofSize, true, resigned) - if shred.Type == ShredTypeCode { - capacity, err = merkleCapacity(len(packet), codingHeaderSize, proofSize, true, resigned) - } - if err != nil { - return solana.Hash{}, err - } - rootOffset := dataHeaderSize + capacity - if shred.Type == ShredTypeCode { - rootOffset = codingHeaderSize + capacity + rootOffset := dataHeaderSize + dataCap + if i >= dataShredsPerFECBlock { + rootOffset = codingHeaderSize + codeCap } copy(packet[rootOffset:rootOffset+merkleRootSize], chainedMerkleRoot[:]) - - if shred.Type == ShredTypeCode { - start := codingHeaderSize - end := start + capacity - copy(packet[start:end], shards[i]) - } else { - start := shredSignatureSize - end := dataHeaderSize + capacity - copy(packet[start:end], shards[i]) - } } - nodes, err := buildMerkleTree(packets) - if err != nil { - return solana.Hash{}, err - } + nodes := buildGeneratedMerkleTree(packets, dataCap, codeCap) root := nodes[len(nodes)-1] sig := ed25519.Sign(ed25519.PrivateKey(leader), root[:]) - for _, packet := range packets { + for i, packet := range packets { copy(packet[shredSignatureOffset:shredSignatureSize], sig) - shred, err := ParseShred(packet) - if err != nil { - return solana.Hash{}, err + proofOffset := dataHeaderSize + dataCap + merkleRootSize + if i >= dataShredsPerFECBlock { + proofOffset = codingHeaderSize + codeCap + merkleRootSize } - leafIndex, err := shred.merkleLeafIndex() - if err != nil { - return solana.Hash{}, err - } - proof := makeMerkleProof(nodes, leafIndex, len(packets)) - capacity, err := merkleCapacity(len(packet), dataHeaderSize, proofSize, true, resigned) - if shred.Type == ShredTypeCode { - capacity, err = merkleCapacity(len(packet), codingHeaderSize, proofSize, true, resigned) - } - if err != nil { - return solana.Hash{}, err - } - proofOffset := dataHeaderSize + capacity + merkleRootSize - if shred.Type == ShredTypeCode { - proofOffset = codingHeaderSize + capacity + merkleRootSize - } - for j, entry := range proof { - copy(packet[proofOffset+j*merkleProofEntrySize:], entry[:]) + proofEntries := writeMerkleProof(packet[proofOffset:], nodes, i, len(packets)) + if proofEntries != int(proofSize) { + return solana.Hash{}, fmt.Errorf("generated merkle proof has %d entries, want %d", proofEntries, proofSize) } if resigned { retransmitOffset := proofOffset + int(proofSize)*merkleProofEntrySize @@ -312,6 +306,54 @@ func finishErasureBatch( return root, nil } +// buildGeneratedMerkleTree hashes the fixed packet order emitted by +// makeFECBatch: 32 data shreds followed by 32 coding shreds. Callers must have +// already validated the packet sizes and variants in finishErasureBatch. +func buildGeneratedMerkleTree(packets [][]byte, dataCap, codeCap int) []solana.Hash { + leaves := make([]solana.Hash, len(packets)) + for i, packet := range packets { + end := dataHeaderSize + dataCap + merkleRootSize + if i >= dataShredsPerFECBlock { + end = codingHeaderSize + codeCap + merkleRootSize + } + leaves[i] = merkleHashLeaf(packet[shredSignatureSize:end]) + } + + nodes := make([]solana.Hash, 0, merkleTreeSize(len(leaves))) + nodes = append(nodes, leaves...) + for size := len(leaves); size > 1; size = (size + 1) >> 1 { + offset := len(nodes) - size + for index := offset; index < offset+size; index += 2 { + other := index + 1 + if other >= offset+size { + other = offset + size - 1 + } + nodes = append(nodes, merkleHashNode(nodes[index][:merkleProofEntrySize], nodes[other][:merkleProofEntrySize])) + } + } + return nodes +} + +// writeMerkleProof writes the truncated sibling hashes directly into a packet. +// The generated FEC tree has fixed depth, so materializing a temporary proof +// slice for every one of its 64 packets only adds allocator and copy traffic. +func writeMerkleProof(dst []byte, nodes []solana.Hash, index, size int) int { + entries := 0 + offset := 0 + for size > 1 { + sibling := index ^ 1 + if sibling >= size { + sibling = size - 1 + } + copy(dst[entries*merkleProofEntrySize:], nodes[offset+sibling][:merkleProofEntrySize]) + entries++ + offset += size + size = (size + 1) >> 1 + index >>= 1 + } + return entries +} + func buildMerkleTree(packets [][]byte) ([]solana.Hash, error) { leaves := make([]solana.Hash, len(packets)) for i, packet := range packets { diff --git a/pkg/turbine/generate_bench_test.go b/pkg/turbine/generate_bench_test.go new file mode 100644 index 00000000..233392d3 --- /dev/null +++ b/pkg/turbine/generate_bench_test.go @@ -0,0 +1,242 @@ +package turbine + +import ( + "crypto/ed25519" + "crypto/sha256" + "encoding/hex" + "fmt" + "testing" + + "github.com/Overclock-Validator/mithril/pkg/costmodel" + "github.com/gagliardetto/solana-go" + "github.com/klauspost/reedsolomon" +) + +const ( + benchmarkTransactionCount = 50_000 + benchmarkTransactionBytes = 1_232 +) + +var ( + benchmarkEncoderSink reedsolomon.Encoder + benchmarkPacketsSink [][]byte + benchmarkRootSink solana.Hash + benchmarkByteSink byte +) + +func benchmarkLeaderKey() solana.PrivateKey { + var seed [ed25519.SeedSize]byte + for i := range seed { + seed[i] = byte(i + 1) + } + return solana.PrivateKey(ed25519.NewKeyFromSeed(seed[:])) +} + +func benchmarkPayload(size int) []byte { + payload := make([]byte, size) + var state uint64 = 0x9e3779b97f4a7c15 + for i := range payload { + // A deterministic, non-zero corpus avoids accidentally benchmarking a + // special all-zero input while keeping fixture construction out of the + // timed region. + state ^= state << 7 + state ^= state >> 9 + state ^= state << 8 + payload[i] = byte(state) + } + return payload +} + +// BenchmarkReedSolomonEncode32x32 isolates the arithmetic kernel used by one +// unsigned 32+32 chained FEC set. Encoder construction, shred parsing, Merkle +// hashing, signing, and packet copies are intentionally outside this result. +func BenchmarkReedSolomonEncode32x32(b *testing.B) { + const shardBytes = 987 // unsigned chained 32+32 shreds with proof size 6 + encoder, err := reedsolomon.New(dataShredsPerFECBlock, codingShredsPerFECBlock) + if err != nil { + b.Fatal(err) + } + shards := make([][]byte, dataShredsPerFECBlock+codingShredsPerFECBlock) + for i := range shards { + shards[i] = make([]byte, shardBytes) + if i < dataShredsPerFECBlock { + copy(shards[i], benchmarkPayload(shardBytes)) + } + } + + b.SetBytes(dataShredsPerFECBlock * shardBytes) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if err := encoder.Encode(shards); err != nil { + b.Fatal(err) + } + } + benchmarkByteSink = shards[len(shards)-1][shardBytes-1] +} + +// BenchmarkReedSolomonNew32x32 measures work that finishErasureBatch currently +// repeats for every FEC set even though the 32+32 shape never changes. +func BenchmarkReedSolomonNew32x32(b *testing.B) { + b.ReportAllocs() + for i := 0; i < b.N; i++ { + encoder, err := reedsolomon.New(dataShredsPerFECBlock, codingShredsPerFECBlock) + if err != nil { + b.Fatal(err) + } + benchmarkEncoderSink = encoder + } +} + +// BenchmarkMakeShredsFromData reports the complete current generator cost, +// including packet construction, Reed-Solomon coding, chained Merkle trees, +// one Ed25519 signature per FEC set, and proof materialization. +func BenchmarkMakeShredsFromData(b *testing.B) { + const blockBytes = benchmarkTransactionCount * benchmarkTransactionBytes + cases := []struct { + name string + size int + isLastInSlot bool + }{ + {name: "one-unsigned-fec", size: dataShredsPerFECBlock * dataCapacity(proofEntriesFor32x32, false)}, + {name: "block-50000x1232", size: blockBytes, isLastInSlot: true}, + } + + for _, tc := range cases { + b.Run(tc.name, func(b *testing.B) { + leader := benchmarkLeaderKey() + payload := benchmarkPayload(tc.size) + gen := ShredGenerator{Slot: 10, ParentSlot: 9, Version: 1} + + b.SetBytes(int64(tc.size)) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + packets, root, _, _, err := gen.MakeShredsFromData(leader, payload, tc.isLastInSlot, solana.Hash{}, 0, 0) + if err != nil { + b.Fatal(err) + } + benchmarkPacketsSink = packets + benchmarkRootSink = root + } + b.StopTimer() + if len(benchmarkPacketsSink) == 0 || len(benchmarkPacketsSink)%64 != 0 { + b.Fatalf("unexpected packet count %d", len(benchmarkPacketsSink)) + } + b.ReportMetric(float64(len(benchmarkPacketsSink)), "packets/op") + b.ReportMetric(float64(len(benchmarkPacketsSink)/64), "FEC-sets/op") + if tc.size == blockBytes { + b.ReportMetric(benchmarkTransactionCount, "transactions/op") + } + }) + } +} + +// BenchmarkMakeShreds50000TargetBatches1232 models the producer's target-sized +// component stream without retaining a multi-gigabyte output. It measures only +// the 61.6 MB transaction payload; entry framing is deliberately outside this +// erasure-coding benchmark. +func BenchmarkMakeShreds50000TargetBatches1232(b *testing.B) { + const inputBytes = benchmarkTransactionCount * benchmarkTransactionBytes + leader := benchmarkLeaderKey() + payload := benchmarkPayload(inputBytes) + gen := ShredGenerator{Slot: 10, ParentSlot: 9, Version: 1} + + b.SetBytes(inputBytes) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + var ( + root solana.Hash + dataIndex uint32 + codeIndex uint32 + ) + var ( + offset int + components int + packetCount int + fecSetCount int + ) + for offset < len(payload) { + end := min(offset+costmodel.DefaultTargetBatchBytes, len(payload)) + packets, nextRoot, nextData, nextCode, err := gen.MakeShredsFromData( + leader, + payload[offset:end], + end == len(payload), + root, + dataIndex, + codeIndex, + ) + if err != nil { + b.Fatal(err) + } + benchmarkPacketsSink = packets + components++ + packetCount += len(packets) + fecSetCount += len(packets) / (dataShredsPerFECBlock + codingShredsPerFECBlock) + root, dataIndex, codeIndex = nextRoot, nextData, nextCode + offset = end + } + b.ReportMetric(float64(components), "components/op") + b.ReportMetric(float64(fecSetCount), "FEC-sets/op") + b.ReportMetric(float64(packetCount), "packets/op") + benchmarkRootSink = root + } + b.StopTimer() + b.ReportMetric(benchmarkTransactionCount, "transactions/op") +} + +func TestBenchmarkBlockPayloadAccounting(t *testing.T) { + const blockBytes = benchmarkTransactionCount * benchmarkTransactionBytes + unsignedBatch := dataShredsPerFECBlock * dataCapacity(proofEntriesFor32x32, false) + signedBatch := dataShredsPerFECBlock * dataCapacity(proofEntriesFor32x32, true) + unsignedBytes := blockBytes - signedBatch + unsignedFECs := (unsignedBytes + unsignedBatch - 1) / unsignedBatch + totalFECs := unsignedFECs + 1 + if totalFECs != 2000 { + t.Fatalf("50k x 1232 payload maps to %d FEC sets, want 2000 (%s)", totalFECs, fmt.Sprintf("%d bytes", blockBytes)) + } +} + +func TestProducerTargetMatchesTwoTypicalFECPayloads(t *testing.T) { + want := 2 * dataShredsPerFECBlock * dataCapacity(proofEntriesFor32x32, false) + if costmodel.DefaultTargetBatchBytes != want { + t.Fatalf("producer target = %d, want two typical FEC payloads = %d", costmodel.DefaultTargetBatchBytes, want) + } +} + +func TestMakeShredsFromDataStableBytes(t *testing.T) { + unsignedBatch := dataShredsPerFECBlock * dataCapacity(proofEntriesFor32x32, false) + signedBatch := dataShredsPerFECBlock * dataCapacity(proofEntriesFor32x32, true) + tests := []struct { + name string + size int + isLastInSlot bool + want string + }{ + {name: "unsigned-one-fec", size: unsignedBatch, want: "f9334f1835240df21d4b48a09f35b3ff90578122d30d0527608c10d38d0911f7"}, + {name: "signed-one-fec", size: signedBatch, isLastInSlot: true, want: "bfa398c445509c5e1345553bbe84fea04e86001caf441008b4014d07c6e36ccd"}, + {name: "two-unsigned-one-signed", size: 2*unsignedBatch + signedBatch, isLastInSlot: true, want: "1374b3b1cc35dab1fddb248a8924be73b8a22d2c42a63ce87663b3ffff3762a5"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + gen := ShredGenerator{Slot: 10, ParentSlot: 9, Version: 1, ReferenceTick: 17} + packets, _, _, _, err := gen.MakeShredsFromData( + benchmarkLeaderKey(), benchmarkPayload(tt.size), tt.isLastInSlot, + solana.Hash{3}, 7, 11, + ) + if err != nil { + t.Fatal(err) + } + h := sha256.New() + for _, packet := range packets { + _, _ = h.Write(packet) + } + got := hex.EncodeToString(h.Sum(nil)) + if got != tt.want { + t.Fatalf("packet digest %s, want %s", got, tt.want) + } + }) + } +} diff --git a/pkg/turbine/shred.go b/pkg/turbine/shred.go index dae24ddd..605faf81 100644 --- a/pkg/turbine/shred.go +++ b/pkg/turbine/shred.go @@ -410,13 +410,12 @@ func merkleHashNode(left []byte, right []byte) solana.Hash { return hashv([][]byte{[]byte(merkleHashPrefixNode), left, right}) } -func hashv(parts [][]byte) solana.Hash { +func hashv(parts [][]byte) (out solana.Hash) { h := sha256.New() for _, part := range parts { _, _ = h.Write(part) } - var out solana.Hash - copy(out[:], h.Sum(nil)) + _ = h.Sum(out[:0]) return out } From d7edc1fbcb037336f7fbbdcca078ee1a1ecada0a Mon Sep 17 00:00:00 2001 From: 7layermagik <7layermagik@users.noreply.github.com> Date: Thu, 30 Jul 2026 01:17:46 -0500 Subject: [PATCH 3/9] experiment(turbine): specialize synthetic FEC recovery --- docs/erasure_recovery_experiments.md | 121 +++++ pkg/turbine/internal/rsrecover/doc.go | 8 + pkg/turbine/internal/rsrecover/recover.go | 476 ++++++++++++++++++ .../internal/rsrecover/recover_bench_test.go | 198 ++++++++ .../internal/rsrecover/recover_test.go | 312 ++++++++++++ 5 files changed, 1115 insertions(+) create mode 100644 docs/erasure_recovery_experiments.md create mode 100644 pkg/turbine/internal/rsrecover/doc.go create mode 100644 pkg/turbine/internal/rsrecover/recover.go create mode 100644 pkg/turbine/internal/rsrecover/recover_bench_test.go create mode 100644 pkg/turbine/internal/rsrecover/recover_test.go diff --git a/docs/erasure_recovery_experiments.md b/docs/erasure_recovery_experiments.md new file mode 100644 index 00000000..567db6cf --- /dev/null +++ b/docs/erasure_recovery_experiments.md @@ -0,0 +1,121 @@ +# Erasure recovery experiments + +Status: experimental; no production `SlotAssembler` dispatch uses these paths. + +This document separates two repair regimes that have different objectives. It +also records the fixed 32 data + 32 coding Reed-Solomon contract used by the +synthetic implementation in `pkg/turbine/internal/rsrecover`. + +## Regimes + +Near-tip repair minimizes the time until replay receives a particular blocking +data shred. Its primary candidate is direct recovery when exactly one data +shred is absent and at least one coding shred is present. + +Catch-up repair minimizes useful recovered-data time across many incomplete FEC +sets. Its candidate constructs only the reduced system induced by the missing +data columns and produces only missing data outputs. + +Slot age alone should not select the regime. A future policy experiment should +consume observed Turbine progress: + +- number and fraction of incomplete FEC sets; +- missing data shreds per FEC set; +- whether new shreds are still arriving; +- time or scheduling intervals since the last useful arrival; +- number of FEC sets that have crossed the recovery threshold; +- data-heavy versus coding-heavy availability. + +A progressing slot with one or two holes remains a near-tip workload even if it +is not the newest slot. A stalled slot with many incomplete FEC sets is a +catch-up workload even if wall-clock age is modest. Any eventual selector needs +hysteresis so bursty arrivals cannot oscillate the algorithm on every packet. + +## Matrix contract + +The experiment uses the systematic generator over `GF(256)/0x11d`: + +```text +V[x,j] = x^j +A = V[0:32,0:32] +G = V * A^-1 +G = [I_32; C] +``` + +For the fixed 32+32 shape, exhaustive scalar tests confirm all 1,024 entries: + +```text +C[r,c] = 0xa5 / (0x20 xor r xor c) +``` + +They also confirm `C*C=I`. Recovery tests remain differential against +`github.com/klauspost/reedsolomon`; the closed form is not the sole oracle. + +## Candidates + +### Near tip: direct one-data recovery + +For missing data position `m` and available coding position `r`: + +```text +D_m = C[r,m]^-1 * P_r + + sum(i != m, C[r,m]^-1 * C[r,i] * D_i) +``` + +This prepares one 32-source coefficient row and writes one destination. It does +not construct or invert a general 32x32 matrix. + +### Catch up: reduced missing-data system + +For missing data columns `M` and selected coding rows `R`, substitute every +known data shard and solve: + +```text +B[i,j] = C[R[i],M[j]] +B * D_M = adjusted_coding_rows +``` + +The experiment uses the Cauchy closed form to construct `B^-1` in `O(m^2)`, +expands direct rows over 32 selected sources, and proves each row satisfies the +requested systematic generator row before byte processing. An independently +implemented Gauss-Jordan inverse is the setup fallback. + +The byte kernel is intentionally portable and uses `reedsolomon.LowLevel`. +This isolates algorithm and plan costs; it is not evidence that a portable +kernel will beat the dependency's generated AVX2/GFNI kernels on amd64. + +## Synthetic coverage + +The tests cover: + +- every missing-data position with every coding-row choice for the direct path; +- every pair of missing data positions; +- deterministic mixed patterns at 2, 4, 8, 16, 24, and 32 missing data shreds; +- exactly-threshold and one-below-threshold availability; +- changed availability between setup and execution; +- destination failure atomicity; +- coefficient mutation detection; +- Cauchy inverses against independent Gauss-Jordan inversion; +- recovered bytes against the existing general decoder. + +Run: + +```bash +go test ./pkg/turbine/internal/rsrecover +go test -run '^$' \ + -bench '^(BenchmarkRecoverOneData|BenchmarkRecoverDataSubset)$' \ + -benchmem -benchtime=2s -count=6 \ + ./pkg/turbine/internal/rsrecover +``` + +Benchmark result interpretation must keep these cases separate: + +- `prepare`: cold or changing erasure pattern; +- `execute`: prepared/repeated pattern; +- `prepare-and-execute`: first useful output for a new pattern; +- general cache off: existing decoder with changing pattern cost; +- general cache on: existing decoder after the inversion is cached. + +No production dispatch threshold should be chosen from an Apple benchmark. +Final crossover decisions require the pinned amd64 target and synthetic arrival +traces for progressing, stalled, and bursty slots. diff --git a/pkg/turbine/internal/rsrecover/doc.go b/pkg/turbine/internal/rsrecover/doc.go new file mode 100644 index 00000000..adac8aad --- /dev/null +++ b/pkg/turbine/internal/rsrecover/doc.go @@ -0,0 +1,8 @@ +// Package rsrecover contains experimental, fixed-shape Reed-Solomon recovery +// plans for Solana's 32 data + 32 coding shred FEC sets. +// +// Nothing in the production turbine assembler imports this package. It exists +// to measure two workload-specific questions before either policy is wired in: +// near-tip latency for one missing data shred, and catch-up throughput for a +// subset of missing data shreds. +package rsrecover diff --git a/pkg/turbine/internal/rsrecover/recover.go b/pkg/turbine/internal/rsrecover/recover.go new file mode 100644 index 00000000..38958c79 --- /dev/null +++ b/pkg/turbine/internal/rsrecover/recover.go @@ -0,0 +1,476 @@ +package rsrecover + +import ( + "errors" + "fmt" + + "github.com/klauspost/reedsolomon" +) + +const ( + DataShards = 32 + CodingShards = 32 + TotalShards = DataShards + CodingShards + + cauchyGamma = byte(0xa5) +) + +var ( + ErrInvalidPattern = errors.New("invalid erasure pattern") + ErrPatternChanged = errors.New("availability differs from prepared plan") + ErrInvalidBuffers = errors.New("invalid recovery buffers") + + gfLog [256]byte + gfExp [512]byte +) + +func init() { + value := uint16(1) + for exponent := 0; exponent < 255; exponent++ { + gfExp[exponent] = byte(value) + gfLog[byte(value)] = byte(exponent) + value <<= 1 + if value&0x100 != 0 { + value ^= 0x11d + } + } + for exponent := 255; exponent < len(gfExp); exponent++ { + gfExp[exponent] = gfExp[exponent-255] + } +} + +// OneDataPlan recovers exactly one absent data shard from the other 31 data +// shards and one available coding shard. It avoids a general 32x32 decode +// matrix inversion. The plan is immutable and safe for concurrent execution +// when callers provide independent destinations. +type OneDataPlan struct { + presence uint64 + missing uint8 + sources [DataShards]uint8 + coefficients [DataShards]byte +} + +// DataSubsetPlan recovers every absent data shard using the present data and +// the lowest-indexed coding rows required to reach the 32-shard threshold. +// Setup solves only the m x m system induced by the missing data columns. +type DataSubsetPlan struct { + presence uint64 + missing []uint8 + sources [DataShards]uint8 + weights [][]byte +} + +// Presence reports which of the 64 input shards are non-empty. A zero-length +// shard is absent, matching reedsolomon.ReconstructSome. +func Presence(shards [][]byte) (uint64, error) { + if len(shards) != TotalShards { + return 0, fmt.Errorf("%w: got %d shards, want %d", ErrInvalidBuffers, len(shards), TotalShards) + } + var mask uint64 + for index, shard := range shards { + if len(shard) != 0 { + mask |= uint64(1) << index + } + } + return mask, nil +} + +// PrepareRecoverOneData constructs the direct coefficient row for a pattern +// with exactly one missing data shard. Additional coding shards may be present; +// the lowest-indexed one is selected deterministically. +func PrepareRecoverOneData(presence uint64, missingDataIndex int) (OneDataPlan, error) { + if missingDataIndex < 0 || missingDataIndex >= DataShards { + return OneDataPlan{}, fmt.Errorf("%w: missing data index %d", ErrInvalidPattern, missingDataIndex) + } + for index := 0; index < DataShards; index++ { + present := presence&(uint64(1)< Date: Thu, 30 Jul 2026 01:21:52 -0500 Subject: [PATCH 4/9] experiment(turbine): benchmark coding-only recovery --- docs/erasure_recovery_experiments.md | 36 +++++++++++++ pkg/turbine/internal/rsrecover/recover.go | 54 +++++++++++++++++++ .../internal/rsrecover/recover_bench_test.go | 24 +++++++++ .../internal/rsrecover/recover_test.go | 48 +++++++++++++++++ 4 files changed, 162 insertions(+) diff --git a/docs/erasure_recovery_experiments.md b/docs/erasure_recovery_experiments.md index 567db6cf..d2d67f7d 100644 --- a/docs/erasure_recovery_experiments.md +++ b/docs/erasure_recovery_experiments.md @@ -84,6 +84,14 @@ The byte kernel is intentionally portable and uses `reedsolomon.LowLevel`. This isolates algorithm and plan costs; it is not evidence that a portable kernel will beat the dependency's generated AVX2/GFNI kernels on amd64. +### Catch up edge: all coding rows + +When all data rows are missing and every coding row is present, `C*C=I` means +the existing optimized encoder can apply `C` to the coding rows and recover the +data directly. This is kept as a separate synthetic arm. It is simpler than a +general decoder, but a cached reduced-system plan may still have a faster byte +kernel; hardware decides between them. + ## Synthetic coverage The tests cover: @@ -119,3 +127,31 @@ Benchmark result interpretation must keep these cases separate: No production dispatch threshold should be chosen from an Apple benchmark. Final crossover decisions require the pinned amd64 target and synthetic arrival traces for progressing, stalled, and bursty slots. + +## Preliminary Apple M4 Pro diagnostic + +These single-sample medians use 987-byte shards and exist only to reject or +retain candidates before the amd64 gate. Times are microseconds per FEC set. + +| Missing data | Specialized first use | Specialized prepared | General uncached | General cached | +|---:|---:|---:|---:|---:| +| 1 | 1.97 | 1.07 | 8.36 | 2.31 | +| 2 | 4.19 | 2.10 | 10.78 | 3.39 | +| 4 | 8.84 | 4.19 | 17.00 | 5.91 | +| 8 | 18.94 | 8.24 | 25.54 | 10.06 | +| 16 | 39.90 | 16.44 | 44.93 | 19.78 | +| 24 | 62.14 | 24.71 | 63.43 | 28.55 | +| 32 | 79.50 | 32.76 | 89.53 | 37.53 | + +The all-coding involution arm measured approximately 36.5 microseconds, +compared with 82.7 microseconds for an uncached general decode and 37.5 +microseconds for its cached form. Its main possible value is avoiding plan +setup; the prepared reduced-system byte path was faster on this machine. + +The current interpretation is deliberately conditional: + +- direct one-data recovery is strong enough to require an amd64 prototype; +- reduced-system first use wins through most of the tested range, but the + 24-missing crossover is within noise on this machine; +- prepared reduced-system execution wins at every tested width; +- none of these figures establishes a production policy or Zen 5 result. diff --git a/pkg/turbine/internal/rsrecover/recover.go b/pkg/turbine/internal/rsrecover/recover.go index 38958c79..dd48e99d 100644 --- a/pkg/turbine/internal/rsrecover/recover.go +++ b/pkg/turbine/internal/rsrecover/recover.go @@ -3,6 +3,7 @@ package rsrecover import ( "errors" "fmt" + "sync" "github.com/klauspost/reedsolomon" ) @@ -22,6 +23,10 @@ var ( gfLog [256]byte gfExp [512]byte + + allCodingEncoderOnce sync.Once + allCodingEncoder reedsolomon.Encoder + allCodingEncoderErr error ) func init() { @@ -60,6 +65,14 @@ type DataSubsetPlan struct { weights [][]byte } +// AllCodingPlan recovers all 32 data rows from all 32 coding rows. For the +// fixed Solana matrix C*C=I, so the package's optimized encoder can apply C a +// second time instead of constructing a decode matrix. +type AllCodingPlan struct { + presence uint64 + encoder reedsolomon.Encoder +} + // Presence reports which of the 64 input shards are non-empty. A zero-length // shard is absent, matching reedsolomon.ReconstructSome. func Presence(shards [][]byte) (uint64, error) { @@ -227,6 +240,47 @@ func PrepareRecoverDataSubset(presence uint64) (DataSubsetPlan, error) { return plan, nil } +// PrepareRecoverAllDataFromCoding accepts the single coding-only threshold +// pattern: no data rows and all coding rows. Encoder construction is shared +// process-wide because the 32+32 matrix is immutable. +func PrepareRecoverAllDataFromCoding(presence uint64) (AllCodingPlan, error) { + want := uint64(0xffffffff) << DataShards + if presence != want { + return AllCodingPlan{}, fmt.Errorf("%w: all-coding recovery requires presence %#016x, got %#016x", ErrInvalidPattern, want, presence) + } + allCodingEncoderOnce.Do(func() { + allCodingEncoder, allCodingEncoderErr = reedsolomon.New(DataShards, CodingShards) + }) + if allCodingEncoderErr != nil { + return AllCodingPlan{}, allCodingEncoderErr + } + return AllCodingPlan{presence: presence, encoder: allCodingEncoder}, nil +} + +// Recover writes all 32 data destinations by applying the coding matrix to +// the 32 coding inputs. Validation completes before Encode writes any output. +func (plan *AllCodingPlan) Recover(shards, destinations [][]byte) error { + shardSize, err := validateExecution(plan.presence, shards, destinations) + if err != nil { + return err + } + if len(destinations) != DataShards { + return fmt.Errorf("%w: got %d destinations, want %d", ErrInvalidBuffers, len(destinations), DataShards) + } + var work [TotalShards][]byte + for index := 0; index < DataShards; index++ { + if len(destinations[index]) != shardSize { + return fmt.Errorf("%w: destination %d has %d bytes, want %d", ErrInvalidBuffers, index, len(destinations[index]), shardSize) + } + work[index] = shards[DataShards+index] + work[DataShards+index] = destinations[index] + } + if err := plan.encoder.Encode(work[:]); err != nil { + return fmt.Errorf("recover all data from coding: %w", err) + } + return nil +} + func subsetWeights(knownData, selectedCoding []uint8, inverse [][]byte) [][]byte { weights := make([][]byte, len(inverse)) for output := range inverse { diff --git a/pkg/turbine/internal/rsrecover/recover_bench_test.go b/pkg/turbine/internal/rsrecover/recover_bench_test.go index 1f96157f..6f6005d3 100644 --- a/pkg/turbine/internal/rsrecover/recover_bench_test.go +++ b/pkg/turbine/internal/rsrecover/recover_bench_test.go @@ -150,6 +150,30 @@ func BenchmarkRecoverDataSubset(b *testing.B) { } } +func BenchmarkRecoverAllDataFromCoding(b *testing.B) { + fixture := makeRecoveryBenchmarkFixture(b, DataShards) + plan, err := PrepareRecoverAllDataFromCoding(fixture.presence) + if err != nil { + b.Fatal(err) + } + dst := make([][]byte, DataShards) + for index := range dst { + dst[index] = make([]byte, len(fixture.original[0])) + } + b.Run("coding-involution/execute", func(b *testing.B) { + b.ReportAllocs() + b.SetBytes(int64(DataShards * len(fixture.original[0]))) + for b.Loop() { + if err := plan.Recover(fixture.available, dst); err != nil { + b.Fatal(err) + } + benchmarkBytesSink = dst[0] + } + }) + benchmarkGeneralRecovery(b, fixture, false) + benchmarkGeneralRecovery(b, fixture, true) +} + func benchmarkGeneralRecovery(b *testing.B, fixture recoveryBenchmarkFixture, inversionCache bool) { name := "general/cache-off" if inversionCache { diff --git a/pkg/turbine/internal/rsrecover/recover_test.go b/pkg/turbine/internal/rsrecover/recover_test.go index db8eb08f..00e3b5c0 100644 --- a/pkg/turbine/internal/rsrecover/recover_test.go +++ b/pkg/turbine/internal/rsrecover/recover_test.go @@ -310,3 +310,51 @@ func TestRecoverDataSubsetThresholdAndBufferFailures(t *testing.T) { } } } + +func TestRecoverAllDataFromCoding(t *testing.T) { + original := encodedFixture(t, 987) + available := availableFixture(original, makeRange(DataShards), makeRange(CodingShards)) + presence, err := Presence(available) + if err != nil { + t.Fatal(err) + } + plan, err := PrepareRecoverAllDataFromCoding(presence) + if err != nil { + t.Fatal(err) + } + dst := make([][]byte, DataShards) + for index := range dst { + dst[index] = make([]byte, len(original[index])) + } + if err := plan.Recover(available, dst); err != nil { + t.Fatal(err) + } + for index := range dst { + if !bytes.Equal(dst[index], original[index]) { + t.Fatalf("recovered data shard %d differs", index) + } + } + + changed := append([][]byte(nil), available...) + changed[0] = original[0] + if _, err := PrepareRecoverAllDataFromCoding(mustPresence(t, changed)); !errors.Is(err, ErrInvalidPattern) { + t.Fatalf("non-coding-only pattern error = %v, want ErrInvalidPattern", err) + } +} + +func makeRange(count int) []int { + result := make([]int, count) + for index := range result { + result[index] = index + } + return result +} + +func mustPresence(t testing.TB, shards [][]byte) uint64 { + t.Helper() + presence, err := Presence(shards) + if err != nil { + t.Fatal(err) + } + return presence +} From c3193e26cb8247cd8b9a19a709257493bb49c71d Mon Sep 17 00:00:00 2001 From: 7layermagik <7layermagik@users.noreply.github.com> Date: Thu, 30 Jul 2026 01:39:06 -0500 Subject: [PATCH 5/9] fix(turbine): complete multi-FEC components only once --- pkg/turbine/component_test.go | 30 ++++++++++++++++++++++++++++++ pkg/turbine/generate.go | 14 ++++++++++---- pkg/turbine/generate_bench_test.go | 2 +- 3 files changed, 41 insertions(+), 5 deletions(-) diff --git a/pkg/turbine/component_test.go b/pkg/turbine/component_test.go index 64b05760..2617715b 100644 --- a/pkg/turbine/component_test.go +++ b/pkg/turbine/component_test.go @@ -131,6 +131,36 @@ func TestShredEntryBatchRoundTrip(t *testing.T) { require.Equal(t, entry.NumHashes, components[0].EntryBatch[0].NumHashes) } +func TestShredMultiFECEntryBatchRoundTrip(t *testing.T) { + leader := testLeader(t) + entries := make([]turbine.Entry, 1300) + for i := range entries { + entries[i] = turbine.Entry{NumHashes: 1, Hash: solana.Hash{byte(i), byte(i >> 8)}} + } + component, err := turbine.NewEntryBatch(entries) + require.NoError(t, err) + + shredder := turbine.Shredder{Slot: 100, ParentSlot: 99, Version: 42, ReferenceTick: 63} + batch, _, _, err := shredder.MakeMerkleShredsFromComponent( + leader, component, true, solana.Hash{}, 0, 0, + ) + require.NoError(t, err) + require.Greater(t, len(batch.DataShreds), 32) + for i, shred := range batch.DataShreds[:len(batch.DataShreds)-1] { + require.False(t, shred.DataComplete(), "intermediate data shred %d ended the component", i) + } + require.True(t, batch.DataShreds[len(batch.DataShreds)-1].DataComplete()) + require.True(t, batch.DataShreds[len(batch.DataShreds)-1].LastInSlot()) + + components, err := turbine.DecodeComponentsFromDataShreds(batch.DataShreds) + require.NoError(t, err) + require.Len(t, components, 1) + require.Len(t, components[0].EntryBatch, len(entries)) + for i := range entries { + require.Equal(t, entries[i].Hash, components[0].EntryBatch[i].Hash) + } +} + func TestShredBlockHeaderMarkerRoundTrip(t *testing.T) { leader := testLeader(t) parentID := solana.Hash{8} diff --git a/pkg/turbine/generate.go b/pkg/turbine/generate.go index 3eb29c0a..9df66019 100644 --- a/pkg/turbine/generate.go +++ b/pkg/turbine/generate.go @@ -123,7 +123,11 @@ func (g *ShredGenerator) MakeShredsFromData( for len(unsignedData) >= unsignedBatch { batch := unsignedData[:unsignedBatch] unsignedData = unsignedData[unsignedBatch:] - batchPackets, root, err := g.makeFECBatch(encoder, leader, batch, unsignedCap, proofSize, false, parentOffset, flags, false, chainedRoot, dataIndex, codeIndex) + // DATA_COMPLETE marks the end of the serialized component, not the end + // of every FEC set. A full unsigned batch is complete only when no + // unsigned remainder or signed-last batch follows it. + dataComplete := len(unsignedData) == 0 && len(signedData) == 0 + batchPackets, root, err := g.makeFECBatch(encoder, leader, batch, unsignedCap, proofSize, false, parentOffset, flags, dataComplete, false, chainedRoot, dataIndex, codeIndex) if err != nil { return nil, solana.Hash{}, dataIndex, codeIndex, err } @@ -134,7 +138,8 @@ func (g *ShredGenerator) MakeShredsFromData( } if len(unsignedData) > 0 || (len(packets) == 0 && !isLastInSlot) { - batchPackets, root, err := g.makeFECBatch(encoder, leader, unsignedData, unsignedCap, proofSize, false, parentOffset, flags, false, chainedRoot, dataIndex, codeIndex) + dataComplete := len(signedData) == 0 + batchPackets, root, err := g.makeFECBatch(encoder, leader, unsignedData, unsignedCap, proofSize, false, parentOffset, flags, dataComplete, false, chainedRoot, dataIndex, codeIndex) if err != nil { return nil, solana.Hash{}, dataIndex, codeIndex, err } @@ -145,7 +150,7 @@ func (g *ShredGenerator) MakeShredsFromData( } if len(signedData) > 0 || (len(packets) == 0 && isLastInSlot) { - batchPackets, root, err := g.makeFECBatch(encoder, leader, signedData, signedCap, proofSize, true, parentOffset, flags, isLastInSlot, chainedRoot, dataIndex, codeIndex) + batchPackets, root, err := g.makeFECBatch(encoder, leader, signedData, signedCap, proofSize, true, parentOffset, flags, true, isLastInSlot, chainedRoot, dataIndex, codeIndex) if err != nil { return nil, solana.Hash{}, dataIndex, codeIndex, err } @@ -167,6 +172,7 @@ func (g *ShredGenerator) makeFECBatch( resigned bool, parentOffset uint16, flags byte, + dataComplete bool, isLastInSlot bool, chainedMerkleRoot solana.Hash, dataIndex uint32, @@ -221,7 +227,7 @@ func (g *ShredGenerator) makeFECBatch( dataPackets[i][dataFlagsOffset] |= shredFlagLastShredInSlot break } - } else if len(dataPackets) > 0 { + } else if dataComplete && len(dataPackets) > 0 { dataPackets[len(dataPackets)-1][dataFlagsOffset] |= shredFlagDataComplete } diff --git a/pkg/turbine/generate_bench_test.go b/pkg/turbine/generate_bench_test.go index 233392d3..2301eefa 100644 --- a/pkg/turbine/generate_bench_test.go +++ b/pkg/turbine/generate_bench_test.go @@ -216,7 +216,7 @@ func TestMakeShredsFromDataStableBytes(t *testing.T) { }{ {name: "unsigned-one-fec", size: unsignedBatch, want: "f9334f1835240df21d4b48a09f35b3ff90578122d30d0527608c10d38d0911f7"}, {name: "signed-one-fec", size: signedBatch, isLastInSlot: true, want: "bfa398c445509c5e1345553bbe84fea04e86001caf441008b4014d07c6e36ccd"}, - {name: "two-unsigned-one-signed", size: 2*unsignedBatch + signedBatch, isLastInSlot: true, want: "1374b3b1cc35dab1fddb248a8924be73b8a22d2c42a63ce87663b3ffff3762a5"}, + {name: "two-unsigned-one-signed", size: 2*unsignedBatch + signedBatch, isLastInSlot: true, want: "dffae840e4c247680b5e1667747a63138872a0080c51a6fcf4cc5002eb7778ac"}, } for _, tt := range tests { From b4b8ef474b28f70428904f6de684c73fa505cdc8 Mon Sep 17 00:00:00 2001 From: 7layermagik <7layermagik@users.noreply.github.com> Date: Thu, 30 Jul 2026 01:42:00 -0500 Subject: [PATCH 6/9] test(turbine): add deterministic repair simulation --- cmd/repair-sim/main.go | 132 ++++ docs/erasure_recovery_experiments.md | 3 + docs/repair_sim.md | 149 +++++ docs/results/repair-sim/RESULTS_TEMPLATE.md | 22 + pkg/turbine/repairsim/ledger.go | 257 +++++++ pkg/turbine/repairsim/sim.go | 700 ++++++++++++++++++++ pkg/turbine/repairsim/sim_test.go | 214 ++++++ 7 files changed, 1477 insertions(+) create mode 100644 cmd/repair-sim/main.go create mode 100644 docs/repair_sim.md create mode 100644 docs/results/repair-sim/RESULTS_TEMPLATE.md create mode 100644 pkg/turbine/repairsim/ledger.go create mode 100644 pkg/turbine/repairsim/sim.go create mode 100644 pkg/turbine/repairsim/sim_test.go diff --git a/cmd/repair-sim/main.go b/cmd/repair-sim/main.go new file mode 100644 index 00000000..3a95fead --- /dev/null +++ b/cmd/repair-sim/main.go @@ -0,0 +1,132 @@ +// repair-sim runs deterministic, single-node Turbine repair scenarios. +package main + +import ( + "encoding/json" + "flag" + "fmt" + "os" + "os/exec" + "runtime" + "strings" + "time" + + "github.com/Overclock-Validator/mithril/pkg/turbine/repairsim" +) + +type environment struct { + GoVersion string `json:"go_version"` + GOOS string `json:"goos"` + GOARCH string `json:"goarch"` + CPU string `json:"cpu"` +} + +type report struct { + Environment environment `json:"environment"` + Ledger repairsim.LedgerConfig `json:"ledger"` + Network repairsim.Config `json:"network"` + LedgerGenerationWall time.Duration `json:"ledger_generation_wall_ns"` + Result repairsim.Result `json:"result"` +} + +func main() { + var ( + scenarioFlag = flag.String("scenario", string(repairsim.ScenarioNearTip), "near-tip or deep-catchup") + slots = flag.Int("slots", 200, "number of deterministic slots") + fecSets = flag.Int("fec-sets", 4, "FEC sets generated per slot") + entries = flag.Int("entries", 0, "entries per slot (0 derives an exact FEC count)") + seed = flag.Int64("seed", 1, "deterministic content and network seed") + availability = flag.String("availability", "", "complete, near-loss, sparse, or mixed") + repair = flag.Bool("repair", true, "enable repair requests") + latency = flag.Duration("repair-latency", 20*time.Millisecond, "synthetic one-way response latency") + jitter = flag.Duration("repair-jitter", 2*time.Millisecond, "deterministic +/- response jitter") + loss = flag.Float64("packet-loss", 0, "repair response loss probability [0,1]") + duplicates = flag.Float64("duplicates", 0.02, "duplicate response probability [0,1]") + bandwidth = flag.Int64("repair-bandwidth", 100*1024*1024, "synthetic repair bytes/sec (0 is unlimited)") + concurrent = flag.Int("max-concurrent", 256, "maximum outstanding repair shreds") + corrupt = flag.Int("corrupt-responses", 0, "corrupt the first N repair responses") + naturalLate = flag.Bool("natural-late", true, "schedule selected late live shreds during repair") + spoolDir = flag.String("spool-dir", "", "persistent shred-spool directory (empty uses a temporary directory)") + output = flag.String("output", "", "write JSON to this file instead of stdout") + includeTrace = flag.Bool("trace", true, "include the logical event trace in JSON") + ) + flag.Parse() + + scenario := repairsim.Scenario(*scenarioFlag) + network := repairsim.DefaultConfig(scenario) + network.Availability = repairsim.Availability(*availability) + network.RepairEnabled = *repair + network.RepairLatency = *latency + network.RepairJitter = *jitter + network.PacketLoss = *loss + network.DuplicateProbability = *duplicates + network.BandwidthBytesPerSec = *bandwidth + network.MaxConcurrent = *concurrent + network.CorruptResponses = *corrupt + network.NaturalLateShreds = *naturalLate + network.CollectTrace = *includeTrace + network.Seed = *seed + network.SpoolDir = *spoolDir + if network.Availability == "" { + network.Availability = repairsim.DefaultConfig(scenario).Availability + } + + ledgerCfg := repairsim.LedgerConfig{ + StartSlot: 10_000, + Slots: *slots, + FECsPerSlot: *fecSets, + EntriesPerSlot: *entries, + Seed: *seed, + ShredVersion: 1, + ReferenceTick: 63, + } + started := time.Now() + ledger, err := repairsim.GenerateLedger(ledgerCfg) + if err != nil { + fatalf("generate ledger: %v", err) + } + generationWall := time.Since(started) + result, err := repairsim.Run(ledger, network) + if err != nil { + fatalf("run simulation: %v", err) + } + report := report{ + Environment: environment{GoVersion: runtime.Version(), GOOS: runtime.GOOS, GOARCH: runtime.GOARCH, CPU: cpuModel()}, + Ledger: ledger.Config, Network: network, LedgerGenerationWall: generationWall, Result: result, + } + encoded, err := json.MarshalIndent(report, "", " ") + if err != nil { + fatalf("marshal report: %v", err) + } + encoded = append(encoded, '\n') + if *output == "" { + _, _ = os.Stdout.Write(encoded) + return + } + if err := os.WriteFile(*output, encoded, 0o644); err != nil { + fatalf("write %s: %v", *output, err) + } +} + +func cpuModel() string { + if runtime.GOOS == "linux" { + if data, err := os.ReadFile("/proc/cpuinfo"); err == nil { + for _, line := range strings.Split(string(data), "\n") { + if key, value, ok := strings.Cut(line, ":"); ok && strings.TrimSpace(key) == "model name" { + return strings.TrimSpace(value) + } + } + } + } + if runtime.GOOS == "darwin" { + if out, err := exec.Command("sysctl", "-n", "machdep.cpu.brand_string").Output(); err == nil { + return strings.TrimSpace(string(out)) + } + } + return "unknown" +} + +func fatalf(format string, args ...any) { + _, _ = fmt.Fprintf(os.Stderr, format+"\n", args...) + os.Exit(1) +} diff --git a/docs/erasure_recovery_experiments.md b/docs/erasure_recovery_experiments.md index d2d67f7d..dc61d77c 100644 --- a/docs/erasure_recovery_experiments.md +++ b/docs/erasure_recovery_experiments.md @@ -2,6 +2,9 @@ Status: experimental; no production `SlotAssembler` dispatch uses these paths. +The end-to-end deterministic harness that drives production repair selection, +assembly, storage, and completion is documented in [repair_sim.md](repair_sim.md). + This document separates two repair regimes that have different objectives. It also records the fixed 32 data + 32 coding Reed-Solomon contract used by the synthetic implementation in `pkg/turbine/internal/rsrecover`. diff --git a/docs/repair_sim.md b/docs/repair_sim.md new file mode 100644 index 00000000..f70fc429 --- /dev/null +++ b/docs/repair_sim.md @@ -0,0 +1,149 @@ +# Deterministic repair simulation + +`cmd/repair-sim` is a single-process harness for measuring the local path from +an incomplete slot to a block that is available to replay. It exists because +near-tip repair and deep catch-up optimize different outcomes: + +- near the tip, latency of the first replay-blocking slot matters; +- during catch-up, sustained useful data and completed slots per second matter. + +The first implementation deliberately stops before UDP and transaction +execution. It establishes a deterministic, correctness-checked baseline before +network realism or alternative scheduling policies are introduced. + +## What is real and what is simulated + +| Stage | Implementation | +| --- | --- | +| block-component serialization | production `turbine.MarshalBlockComponent` | +| 32+32 FEC generation and Merkle signing | production `turbine.Shredder` | +| missing-shred selection | production `SlotAssembler.RepairRequests` | +| packet parsing and Merkle/signature validation | production `ParseShred` and `VerifySignature` | +| verified-shred insertion | production `ShredSpool` | +| threshold detection and Reed-Solomon recovery | production `SlotAssembler.AddShredFrom` | +| component decode and transaction-signature gate | production slot completion path | +| remote peer, latency, jitter, loss, duplication, bandwidth | deterministic in-process simulator | +| replay notification | block emission is recorded as “offered to replay” | +| transaction execution | not run in this version | + +Synthetic entries are valid Alpenglow entry-batch components but contain no +transactions. This isolates shred/FEC/repair/storage costs; it is not a replay +execution benchmark. + +## Scenarios + +### Near tip + +`near-loss` alternates two useful patterns across FEC sets: + +- 30 data + 1 coding shred: one repair response crosses the threshold and + reconstructs the other missing data shred; +- 31 data shreds: the one missing data shred must be fetched directly. + +Selected omitted shreds can also arrive through the simulated live path while +a repair response is outstanding. This measures cancellation/late-response +behavior without changing production scheduling. + +### Deep catch-up + +`mixed` begins every FEC set with 16 data + 15 coding shreds. One fetched data +shred crosses the threshold and reconstructs the remaining 15. + +`sparse` begins every FEC set with two data shreds and no coding layout. The +ordinary repair interface serves data shreds only, so nearly all missing data +must arrive over the simulated network. Comparing `mixed` with `sparse` +quantifies the network work avoided by already-held coding shreds. + +## Commands + +```sh +go test ./pkg/turbine/repairsim + +go run ./cmd/repair-sim \ + -scenario=near-tip \ + -slots=200 \ + -fec-sets=4 \ + -seed=1 \ + -output=/tmp/repair-near.json + +go run ./cmd/repair-sim \ + -scenario=deep-catchup \ + -availability=mixed \ + -slots=1000 \ + -fec-sets=4 \ + -seed=1 \ + -repair-latency=20ms \ + -repair-bandwidth=104857600 \ + -output=/tmp/repair-deep-mixed.json + +go run ./cmd/repair-sim \ + -scenario=deep-catchup \ + -availability=sparse \ + -slots=1000 \ + -fec-sets=4 \ + -seed=1 \ + -repair-latency=20ms \ + -repair-bandwidth=104857600 \ + -output=/tmp/repair-deep-sparse.json + +go test ./pkg/turbine/repairsim \ + -run '^$' -bench '^BenchmarkScenarios$' -benchmem -count=5 +``` + +Logical trace timestamps are deterministic. `wall_elapsed_ns`, allocations, +and `stage_cpu_ns` are actual local measurements and therefore are not expected +to be byte-identical across runs. + +## Correctness gates + +The current tests require: + +- exact requested FEC-set counts from authentic generated shreds; +- Merkle/signature validity for every canonical packet; +- no completed slot when loss is present and repair is disabled; +- complete, canonical entry streams after threshold recovery; +- a complete `ShredSpool` journal record before replay admission; +- rejection and retry of a corrupted repair response; +- deterministic logical traces for identical seeds; +- late repair responses to leave a completed block unchanged. + +The Turbine package also retains focused byte-for-byte Reed-Solomon recovery +tests. The simulator verifies the stronger end-to-end consequence: recovered +shreds must decode to the exact canonical entry sequence and pass the normal +completion gates. + +## Generator compatibility finding + +Building this harness exposed a multi-FEC component bug: the local generator +set `DATA_COMPLETE_SHRED` at every FEC boundary. The decoder correctly treats +that flag as the end of one serialized component, so a component spanning more +than one FEC set was truncated and failed to decode. The generator now follows +Agave's ordering: construct every FEC set, then mark only the final data shred +of the component complete (or last-in-slot). A regression test round-trips one +1,300-entry component across multiple FEC sets. + +## Future mode selection + +No production mode switch is added here. A later policy experiment should use +both replay distance and observed Turbine usefulness, with hysteresis: + +- stay in near-tip mode while the replay gap is small and Turbine supplies a + high fraction of useful shreds before repair deadlines; +- enter catch-up mode only when the replay-blocking gap is sustained and live + Turbine delivery is insufficient to approach FEC thresholds; +- return to near-tip mode only after both the gap and repair backlog fall below + lower thresholds. + +That signal is preferable to slot distance alone: a node may be numerically +close to the tip while receiving too few live shreds, or far behind while its +local spool already holds most FEC thresholds. + +## Next steps + +1. Add shallow catch-up and whole-block-pressure configurations. +2. Add a loopback-UDP transport without replacing the deterministic mode. +3. Expose internal FEC start/finish timing through opt-in instrumentation. +4. Run replay execution against a reusable synthetic bank fixture. +5. Compare ordinary requests with explicit test-only threshold-acquisition and + earliest-blocked-slot policies. + diff --git a/docs/results/repair-sim/RESULTS_TEMPLATE.md b/docs/results/repair-sim/RESULTS_TEMPLATE.md new file mode 100644 index 00000000..e11bce84 --- /dev/null +++ b/docs/results/repair-sim/RESULTS_TEMPLATE.md @@ -0,0 +1,22 @@ +# Repair simulation results template + +Record the commit, CPU, Go version, governor/pinning, command, configuration, +and SHA-256 of every raw JSON file. Do not combine logical network time with +measured local CPU time. + +| scenario | availability | slots | FEC/slot | completed | logical slots/s | CPU slots/s | requests | network data shreds | locally recovered shreds | FEC decodes | p50 completion | p95 completion | p99 completion | +| --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | +| near-tip | near-loss | | | | | | | | | | | | | +| deep-catchup | mixed | | | | | | | | | | | | | +| deep-catchup | sparse | | | | | | | | | | | | | + +Also report: + +- `stage_cpu_ns` by stage; +- `repair_bytes_requested` and `repair_bytes_received`; +- canceled/late, duplicate, lost, and rejected-corrupt responses; +- queue high-water mark; +- spool bytes and complete slots; +- allocations; +- every limitation emitted in the JSON result. + diff --git a/pkg/turbine/repairsim/ledger.go b/pkg/turbine/repairsim/ledger.go new file mode 100644 index 00000000..3510fcd6 --- /dev/null +++ b/pkg/turbine/repairsim/ledger.go @@ -0,0 +1,257 @@ +// Package repairsim provides a deterministic, single-process repair harness +// around Mithril's production Turbine shred generator and slot assembler. +// +// The synthetic ledger and network are test infrastructure. Shred parsing, +// Merkle/signature validation, repair selection, Reed-Solomon reconstruction, +// component decoding, transaction verification, and completion accounting are +// production code paths. +package repairsim + +import ( + "crypto/ed25519" + "crypto/sha256" + "fmt" + + "github.com/Overclock-Validator/mithril/pkg/turbine" + "github.com/gagliardetto/solana-go" +) + +const ( + dataShredsPerFEC = 32 + codeShredsPerFEC = 32 +) + +// LedgerConfig controls deterministic canonical-ledger generation. +type LedgerConfig struct { + StartSlot uint64 `json:"start_slot"` + Slots int `json:"slots"` + FECsPerSlot int `json:"fec_sets_per_slot"` + EntriesPerSlot int `json:"entries_per_slot,omitempty"` + Seed int64 `json:"seed"` + ShredVersion uint16 `json:"shred_version"` + ReferenceTick uint8 `json:"reference_tick"` +} + +// Packet is one canonical wire packet and its parsed routing metadata. +type Packet struct { + Bytes []byte + Slot uint64 + Type turbine.ShredType + Index uint32 + FECSetIndex uint32 + Position uint16 +} + +// FECSet contains the canonical packets for one 32+32 FEC set. +type FECSet struct { + Index uint32 + Data []Packet + Coding []Packet +} + +// Slot is the complete canonical source for one generated slot. +type Slot struct { + Number uint64 + ParentSlot uint64 + Entries []turbine.Entry + FECs []FECSet + Data map[uint32]Packet + Highest uint32 +} + +// Ledger is the complete data held by the in-process repair peer. +type Ledger struct { + Config LedgerConfig + Leader solana.PrivateKey + LeaderPub solana.PublicKey + Slots []Slot + bySlot map[uint64]*Slot +} + +// Slot returns a canonical slot by number. +func (l *Ledger) Slot(number uint64) (*Slot, bool) { + if l == nil { + return nil, false + } + s, ok := l.bySlot[number] + return s, ok +} + +// GenerateLedger creates authentic signed Merkle shreds using the production +// 32+32 generator. Entries contain no transactions: this isolates repair, +// reconstruction, parsing, and storage while still exercising the real block +// component codec and completion path. +func GenerateLedger(cfg LedgerConfig) (*Ledger, error) { + if cfg.Slots <= 0 { + return nil, fmt.Errorf("slots must be positive") + } + if cfg.FECsPerSlot <= 0 { + return nil, fmt.Errorf("FEC sets per slot must be positive") + } + if cfg.StartSlot == 0 { + cfg.StartSlot = 10_000 + } + if cfg.ReferenceTick == 0 { + cfg.ReferenceTick = 63 + } + + leader := deterministicLeader(cfg.Seed) + entryCount := cfg.EntriesPerSlot + if entryCount == 0 { + var err error + entryCount, err = findEntryCount(cfg, leader) + if err != nil { + return nil, err + } + } + + ledger := &Ledger{ + Config: cfg, + Leader: leader, + LeaderPub: leader.PublicKey(), + Slots: make([]Slot, 0, cfg.Slots), + bySlot: make(map[uint64]*Slot, cfg.Slots), + } + ledger.Config.EntriesPerSlot = entryCount + for i := 0; i < cfg.Slots; i++ { + number := cfg.StartSlot + uint64(i) + parent := number - 1 + entries := deterministicEntries(cfg.Seed, number, entryCount) + slot, err := generateSlot(cfg, leader, number, parent, entries) + if err != nil { + return nil, fmt.Errorf("generate slot %d: %w", number, err) + } + if got := len(slot.FECs); got != cfg.FECsPerSlot { + return nil, fmt.Errorf("slot %d produced %d FEC sets, want %d (entries=%d)", number, got, cfg.FECsPerSlot, entryCount) + } + ledger.Slots = append(ledger.Slots, slot) + ledger.bySlot[number] = &ledger.Slots[len(ledger.Slots)-1] + } + return ledger, nil +} + +func deterministicLeader(seed int64) solana.PrivateKey { + var input [16]byte + for i := range input { + input[i] = byte(uint64(seed)>>uint((i%8)*8)) ^ byte(i*29+7) + } + digest := sha256.Sum256(append([]byte("mithril-repair-sim-leader-v1"), input[:]...)) + return solana.PrivateKey(ed25519.NewKeyFromSeed(digest[:])) +} + +func deterministicEntries(seed int64, slot uint64, count int) []turbine.Entry { + entries := make([]turbine.Entry, count) + for i := range entries { + material := fmt.Sprintf("mithril-repair-sim-entry-v1:%d:%d:%d", seed, slot, i) + h := sha256.Sum256([]byte(material)) + entries[i] = turbine.Entry{NumHashes: 1, Hash: solana.Hash(h)} + } + return entries +} + +// findEntryCount uses the generator itself as the capacity oracle. This avoids +// duplicating signed-last-FEC payload constants in the harness. +func findEntryCount(cfg LedgerConfig, leader solana.PrivateKey) (int, error) { + lo, hi := 1, cfg.FECsPerSlot*900 + for lo < hi { + mid := lo + (hi-lo)/2 + entries := deterministicEntries(cfg.Seed, cfg.StartSlot, mid) + slot, err := generateSlot(cfg, leader, cfg.StartSlot, cfg.StartSlot-1, entries) + if err != nil { + return 0, err + } + if len(slot.FECs) < cfg.FECsPerSlot { + lo = mid + 1 + } else { + hi = mid + } + } + entries := deterministicEntries(cfg.Seed, cfg.StartSlot, lo) + slot, err := generateSlot(cfg, leader, cfg.StartSlot, cfg.StartSlot-1, entries) + if err != nil { + return 0, err + } + if len(slot.FECs) != cfg.FECsPerSlot { + return 0, fmt.Errorf("cannot derive %d FEC sets within %d entries (got %d)", cfg.FECsPerSlot, hi, len(slot.FECs)) + } + return lo, nil +} + +func generateSlot(cfg LedgerConfig, leader solana.PrivateKey, number, parent uint64, entries []turbine.Entry) (Slot, error) { + component, err := turbine.NewEntryBatch(entries) + if err != nil { + return Slot{}, err + } + shredder := turbine.Shredder{ + Slot: number, + ParentSlot: parent, + Version: cfg.ShredVersion, + ReferenceTick: cfg.ReferenceTick, + } + batch, _, _, err := shredder.MakeMerkleShredsFromComponent( + leader, component, true, solana.Hash{}, 0, 0, + ) + if err != nil { + return Slot{}, err + } + + byFEC := make(map[uint32]*FECSet) + packetByKey := make(map[packetKey][]byte, len(batch.Packets)) + for _, raw := range batch.Packets { + shred, err := turbine.ParseShred(raw) + if err != nil { + return Slot{}, err + } + key := keyForShred(shred) + packetByKey[key] = append([]byte(nil), raw...) + } + data := make(map[uint32]Packet, len(batch.DataShreds)) + var highest uint32 + for _, shred := range append(append([]*turbine.Shred(nil), batch.DataShreds...), batch.CodeShreds...) { + fec := byFEC[shred.FECSetIndex] + if fec == nil { + fec = &FECSet{Index: shred.FECSetIndex} + byFEC[shred.FECSetIndex] = fec + } + packet := Packet{ + Bytes: packetByKey[keyForShred(shred)], + Slot: shred.Slot, + Type: shred.Type, + Index: shred.Index, + FECSetIndex: shred.FECSetIndex, + Position: shred.Position, + } + if shred.Type == turbine.ShredTypeData { + fec.Data = append(fec.Data, packet) + data[shred.Index] = packet + if shred.Index > highest { + highest = shred.Index + } + } else { + fec.Coding = append(fec.Coding, packet) + } + } + fecs := make([]FECSet, 0, len(byFEC)) + for index := uint32(0); len(fecs) < len(byFEC); index += dataShredsPerFEC { + fec := byFEC[index] + if fec == nil { + return Slot{}, fmt.Errorf("non-contiguous FEC sets: missing %d", index) + } + if len(fec.Data) != dataShredsPerFEC || len(fec.Coding) != codeShredsPerFEC { + return Slot{}, fmt.Errorf("FEC %d has %d+%d shreds", index, len(fec.Data), len(fec.Coding)) + } + fecs = append(fecs, *fec) + } + return Slot{Number: number, ParentSlot: parent, Entries: entries, FECs: fecs, Data: data, Highest: highest}, nil +} + +type packetKey struct { + type_ turbine.ShredType + index uint32 + fec uint32 + position uint16 +} + +func keyForShred(shred *turbine.Shred) packetKey { + return packetKey{type_: shred.Type, index: shred.Index, fec: shred.FECSetIndex, position: shred.Position} +} diff --git a/pkg/turbine/repairsim/sim.go b/pkg/turbine/repairsim/sim.go new file mode 100644 index 00000000..dbabcd9e --- /dev/null +++ b/pkg/turbine/repairsim/sim.go @@ -0,0 +1,700 @@ +package repairsim + +import ( + "container/heap" + "errors" + "fmt" + "math/rand" + "os" + "runtime" + "sort" + "time" + + "github.com/Overclock-Validator/mithril/pkg/block" + "github.com/Overclock-Validator/mithril/pkg/turbine" +) + +type Scenario string + +const ( + ScenarioNearTip Scenario = "near-tip" + ScenarioDeepCatchup Scenario = "deep-catchup" +) + +type Availability string + +const ( + AvailabilityComplete Availability = "complete" + AvailabilityNearLoss Availability = "near-loss" + AvailabilitySparse Availability = "sparse" + AvailabilityMixed Availability = "mixed" +) + +// Config controls the deterministic network and local starting state. +type Config struct { + Scenario Scenario `json:"scenario"` + Availability Availability `json:"availability"` + RepairEnabled bool `json:"repair_enabled"` + RepairLatency time.Duration `json:"repair_latency_ns"` + RepairJitter time.Duration `json:"repair_jitter_ns"` + PacketLoss float64 `json:"packet_loss"` + DuplicateProbability float64 `json:"duplicate_probability"` + BandwidthBytesPerSec int64 `json:"bandwidth_bytes_per_sec"` + MaxConcurrent int `json:"max_concurrent_requests"` + MaxRequestSlots int `json:"max_request_slots"` + MaxMissingPerSlot int `json:"max_missing_per_slot"` + CorruptResponses int `json:"corrupt_responses"` + NaturalLateShreds bool `json:"natural_late_shreds"` + CollectTrace bool `json:"collect_trace"` + Seed int64 `json:"seed"` + SpoolDir string `json:"spool_dir,omitempty"` + SpoolMaxBytes int64 `json:"spool_max_bytes"` +} + +// DefaultConfig returns a deterministic starting point for a scenario. +func DefaultConfig(scenario Scenario) Config { + cfg := Config{ + Scenario: scenario, + RepairEnabled: true, + RepairLatency: 20 * time.Millisecond, + RepairJitter: 2 * time.Millisecond, + DuplicateProbability: 0.02, + BandwidthBytesPerSec: 100 * 1024 * 1024, + MaxConcurrent: 256, + MaxRequestSlots: 64, + MaxMissingPerSlot: 256, + NaturalLateShreds: scenario == ScenarioNearTip, + CollectTrace: true, + Seed: 1, + SpoolMaxBytes: 1 << 30, + } + if scenario == ScenarioDeepCatchup { + cfg.Availability = AvailabilityMixed + } else { + cfg.Availability = AvailabilityNearLoss + } + return cfg +} + +// TraceEvent is a deterministic logical-time event. CPU durations are kept in +// Result.StageCPU so trace equality does not depend on scheduler noise. +type TraceEvent struct { + Sequence int `json:"sequence"` + AtNanos int64 `json:"at_ns"` + Stage string `json:"stage"` + Slot uint64 `json:"slot,omitempty"` + FECSetIndex uint32 `json:"fec_set_index,omitempty"` + ShredIndex uint32 `json:"shred_index,omitempty"` + ShredType turbine.ShredType `json:"shred_type,omitempty"` + Bytes int `json:"bytes,omitempty"` + Detail string `json:"detail,omitempty"` +} + +type LatencySummary struct { + P50 time.Duration `json:"p50_ns"` + P95 time.Duration `json:"p95_ns"` + P99 time.Duration `json:"p99_ns"` +} + +// Result separates simulated-network time from actual local execution time. +type Result struct { + Scenario Scenario `json:"scenario"` + Availability Availability `json:"availability"` + Slots int `json:"slots"` + CompletedSlots int `json:"completed_slots"` + LogicalElapsed time.Duration `json:"logical_elapsed_ns"` + WallElapsed time.Duration `json:"wall_elapsed_ns"` + TimeToFirstReplayable time.Duration `json:"time_to_first_replayable_ns"` + TimeToFirstRecoveredData time.Duration `json:"time_to_first_recovered_data_ns"` + RepairEligibleToRecovery time.Duration `json:"repair_eligible_to_first_recovery_ns"` + CompletionLatency LatencySummary `json:"completion_latency"` + SlotsPerLogicalSecond float64 `json:"slots_per_logical_second"` + SlotsPerCPUSecond float64 `json:"slots_per_cpu_second"` + RepairRequests uint64 `json:"repair_requests"` + RepairResponses uint64 `json:"repair_responses"` + RepairBytesRequested uint64 `json:"repair_bytes_requested"` + RepairBytesReceived uint64 `json:"repair_bytes_received"` + UsefulNetworkDataShreds uint64 `json:"useful_network_data_shreds"` + LocallyRecoveredDataShreds uint64 `json:"locally_recovered_data_shreds"` + LocallyRecoveredDataBytes uint64 `json:"locally_recovered_data_bytes"` + InitialMissingDataShreds uint64 `json:"initial_missing_data_shreds"` + FractionRecoveredLocally float64 `json:"fraction_missing_recovered_locally"` + FECDecodes uint64 `json:"fec_decodes"` + DuplicateResponses uint64 `json:"duplicate_responses"` + CanceledOrLateResponses uint64 `json:"canceled_or_late_responses"` + LostResponses uint64 `json:"lost_responses"` + RejectedCorruptResponses uint64 `json:"rejected_corrupt_responses"` + QueueHighWater int `json:"queue_high_water"` + SpoolBytes int64 `json:"spool_bytes"` + SpoolCompleteSlots int `json:"spool_complete_slots"` + DataShredBytesReplayable uint64 `json:"data_shred_bytes_replayable"` + Allocations uint64 `json:"allocations"` + StageCPU map[string]time.Duration `json:"stage_cpu_ns"` + Trace []TraceEvent `json:"trace"` + Limitations []string `json:"limitations"` +} + +// Run executes the virtual network around production parsing, validation, +// repair selection, reconstruction, spool insertion, and block completion. +func Run(ledger *Ledger, cfg Config) (Result, error) { + if ledger == nil || len(ledger.Slots) == 0 { + return Result{}, errors.New("empty ledger") + } + if cfg.Scenario != ScenarioNearTip && cfg.Scenario != ScenarioDeepCatchup { + return Result{}, fmt.Errorf("unsupported scenario %q", cfg.Scenario) + } + if cfg.Availability == "" { + cfg.Availability = DefaultConfig(cfg.Scenario).Availability + } + if cfg.MaxConcurrent <= 0 { + cfg.MaxConcurrent = 1 + } + if cfg.MaxRequestSlots <= 0 { + cfg.MaxRequestSlots = 64 + } + if cfg.MaxMissingPerSlot <= 0 { + cfg.MaxMissingPerSlot = 256 + } + if cfg.SpoolMaxBytes <= 0 { + cfg.SpoolMaxBytes = 1 << 30 + } + + spoolDir := cfg.SpoolDir + if spoolDir == "" { + var err error + spoolDir, err = os.MkdirTemp("", "mithril-repair-sim-") + if err != nil { + return Result{}, err + } + defer os.RemoveAll(spoolDir) + } + spool, err := turbine.OpenShredSpool(spoolDir, cfg.SpoolMaxBytes) + if err != nil { + return Result{}, err + } + defer spool.Close() + + var memBefore runtime.MemStats + runtime.ReadMemStats(&memBefore) + wallStarted := time.Now() + s := &simulation{ + ledger: ledger, + cfg: cfg, + assembler: turbine.NewSlotAssembler(), + spool: spool, + rng: rand.New(rand.NewSource(cfg.Seed)), + firstShred: make(map[uint64]time.Duration), + completed: make(map[uint64]*block.Block), + replayableAt: make(map[uint64]time.Duration), + pending: make(map[repairKey]struct{}), + stageCPU: make(map[string]time.Duration), + } + s.assembler.SetRetentionFloor(ledger.Slots[0].Number) + s.assembler.SetOnComplete(spool.MarkComplete) + s.nextReplaySlot = ledger.Slots[0].Number + + if err := s.seedLocalState(); err != nil { + return Result{}, err + } + if cfg.RepairEnabled { + if err := s.repairUntilDone(); err != nil { + return Result{}, err + } + } + + var memAfter runtime.MemStats + runtime.ReadMemStats(&memAfter) + _, spoolBytes := spool.Stats() + result := s.result + result.Scenario = cfg.Scenario + result.Availability = cfg.Availability + result.Slots = len(ledger.Slots) + result.CompletedSlots = len(s.completed) + result.LogicalElapsed = s.now + result.WallElapsed = time.Since(wallStarted) + result.LocallyRecoveredDataShreds = s.assembler.RecoveredDataShreds() + result.InitialMissingDataShreds = initialMissingDataShreds(ledger, cfg.Availability) + if result.InitialMissingDataShreds > 0 { + result.FractionRecoveredLocally = float64(result.LocallyRecoveredDataShreds) / float64(result.InitialMissingDataShreds) + } + if len(ledger.Slots[0].FECs) > 0 && len(ledger.Slots[0].FECs[0].Data) > 0 { + result.LocallyRecoveredDataBytes = result.LocallyRecoveredDataShreds * uint64(len(ledger.Slots[0].FECs[0].Data[0].Bytes)) + } + if s.haveRecoverAt { + result.TimeToFirstRecoveredData = s.firstRecoverAt + if s.haveRepairAt && s.firstRecoverAt >= s.firstRepairAt { + result.RepairEligibleToRecovery = s.firstRecoverAt - s.firstRepairAt + } + } + result.SpoolBytes = spoolBytes + result.SpoolCompleteSlots = spool.CompleteSlots() + result.Allocations = memAfter.Mallocs - memBefore.Mallocs + result.StageCPU = s.stageCPU + result.Trace = s.trace + result.Limitations = []string{ + "remote peers and latency are simulated in process; no UDP/IP stack is measured", + "synthetic entries contain no transactions, so transaction execution is not measured", + "slot offered to replay means SlotAssembler emitted a verified block; replay execution is not run", + "FEC decode start is observed at the AddShredFrom call boundary, not inside the Reed-Solomon library", + } + latencies := make([]time.Duration, 0, len(s.replayableAt)) + for slot, completedAt := range s.replayableAt { + if first, ok := s.firstShred[slot]; ok { + latencies = append(latencies, completedAt-first) + } + } + result.CompletionLatency = summarizeLatencies(latencies) + if len(s.replayableAt) > 0 { + first := ledger.Slots[0].Number + result.TimeToFirstReplayable = s.replayableAt[first] + } + if result.LogicalElapsed > 0 { + result.SlotsPerLogicalSecond = float64(result.CompletedSlots) / result.LogicalElapsed.Seconds() + } + if result.WallElapsed > 0 { + result.SlotsPerCPUSecond = float64(result.CompletedSlots) / result.WallElapsed.Seconds() + } + return result, nil +} + +type simulation struct { + ledger *Ledger + cfg Config + assembler *turbine.SlotAssembler + spool *turbine.ShredSpool + rng *rand.Rand + now time.Duration + nextWireAt time.Duration + sequence int + trace []TraceEvent + queue deliveryHeap + pending map[repairKey]struct{} + firstShred map[uint64]time.Duration + completed map[uint64]*block.Block + replayableAt map[uint64]time.Duration + nextReplaySlot uint64 + stageCPU map[string]time.Duration + result Result + corruptLeft int + firstRepairAt time.Duration + haveRepairAt bool + firstRecoverAt time.Duration + haveRecoverAt bool +} + +func (s *simulation) seedLocalState() error { + s.corruptLeft = s.cfg.CorruptResponses + for ordinal := 0; ; ordinal++ { + added := false + for slotIdx := range s.ledger.Slots { + packets := initialPackets(&s.ledger.Slots[slotIdx], s.cfg.Availability) + if ordinal >= len(packets) { + continue + } + added = true + if err := s.ingest(packets[ordinal], false, false); err != nil { + return fmt.Errorf("seed slot %d: %w", s.ledger.Slots[slotIdx].Number, err) + } + s.now++ + } + if !added { + break + } + } + if s.cfg.NaturalLateShreds && s.cfg.Availability == AvailabilityNearLoss { + for i := range s.ledger.Slots { + slot := &s.ledger.Slots[i] + if len(slot.FECs) == 0 || len(slot.FECs[0].Data) < 31 { + continue + } + at := s.now + s.cfg.RepairLatency/2 + time.Duration(i)*time.Microsecond + heap.Push(&s.queue, delivery{at: at, sequence: s.sequence, packet: slot.FECs[0].Data[30]}) + s.sequence++ + } + } + return nil +} + +func initialPackets(slot *Slot, availability Availability) []Packet { + var out []Packet + for i := range slot.FECs { + fec := &slot.FECs[i] + switch availability { + case AvailabilityComplete: + out = append(out, fec.Data...) + case AvailabilityNearLoss: + if i%2 == 0 { + out = append(out, fec.Data[:30]...) + out = append(out, fec.Coding[0]) + } else { + out = append(out, fec.Data[:31]...) + } + case AvailabilitySparse: + out = append(out, fec.Data[:2]...) + case AvailabilityMixed: + out = append(out, fec.Data[:16]...) + out = append(out, fec.Coding[:15]...) + } + } + return out +} + +func (s *simulation) repairUntilDone() error { + const maxIterations = 10_000_000 + for iterations := 0; len(s.completed) < len(s.ledger.Slots); iterations++ { + if iterations >= maxIterations { + return errors.New("repair simulation exceeded iteration limit") + } + s.prioritizeHeadWindow() + s.scheduleRequests() + if len(s.queue) == 0 { + return fmt.Errorf("repair stalled with %d/%d completed", len(s.completed), len(s.ledger.Slots)) + } + event := heap.Pop(&s.queue).(delivery) + if event.at > s.now { + s.now = event.at + } + if event.primary { + delete(s.pending, event.key) + } + if event.drop { + s.result.LostResponses++ + s.record("repair_response_lost", event.packet, "") + continue + } + if event.duplicate { + s.result.DuplicateResponses++ + } + if event.fromRepair { + s.result.RepairResponses++ + s.result.RepairBytesReceived += uint64(len(event.packet.Bytes)) + } + beforeUseful := s.assembler.UsefulRepairShreds() + if err := s.ingest(event.packet, event.fromRepair, event.corrupt); err != nil { + if event.corrupt && errors.Is(err, turbine.ErrInvalidSignature) { + s.result.RejectedCorruptResponses++ + s.record("repair_response_rejected", event.packet, "invalid signature or Merkle proof") + continue + } + return err + } + if event.fromRepair { + afterUseful := s.assembler.UsefulRepairShreds() + if afterUseful == beforeUseful { + s.result.CanceledOrLateResponses++ + } else { + s.result.UsefulNetworkDataShreds += afterUseful - beforeUseful + } + } + } + return nil +} + +func (s *simulation) prioritizeHeadWindow() { + var head uint64 + found := false + for i := range s.ledger.Slots { + slot := s.ledger.Slots[i].Number + if _, complete := s.completed[slot]; !complete { + head, found = slot, true + break + } + } + if !found { + return + } + end := head + 63 + last := s.ledger.Slots[len(s.ledger.Slots)-1].Number + if end > last { + end = last + } + s.assembler.PrioritizeRepairRange(head, end) +} + +func (s *simulation) scheduleRequests() { + capacity := s.cfg.MaxConcurrent - len(s.pending) + if capacity <= 0 { + return + } + requests := s.assembler.RepairRequests(s.cfg.MaxRequestSlots, s.cfg.MaxMissingPerSlot) + for _, req := range requests { + if !s.haveRepairAt { + s.firstRepairAt = s.now + s.haveRepairAt = true + } + s.recordAt("repair_needed_decision", req.Slot, 0, 0, 0, 0, + fmt.Sprintf("missing=%d need_highest=%t", len(req.MissingDataShreds), req.NeedHighestDataShred)) + slot, ok := s.ledger.Slot(req.Slot) + if !ok { + continue + } + indexes := append([]uint32(nil), req.MissingDataShreds...) + if req.NeedHighestDataShred { + indexes = append(indexes, slot.Highest) + } + seen := make(map[uint32]struct{}, len(indexes)) + for _, index := range indexes { + if capacity == 0 { + return + } + if _, duplicate := seen[index]; duplicate { + continue + } + seen[index] = struct{}{} + packet, ok := slot.Data[index] + if !ok { + continue + } + key := repairKey{slot: req.Slot, index: index} + if _, outstanding := s.pending[key]; outstanding { + continue + } + s.pending[key] = struct{}{} + capacity-- + s.result.RepairRequests++ + s.result.RepairBytesRequested += uint64(len(packet.Bytes)) + s.record("repair_request_enqueue", packet, "data shred") + s.record("repair_request_send", packet, "data shred") + s.scheduleResponse(key, packet) + } + } +} + +func (s *simulation) scheduleResponse(key repairKey, packet Packet) { + jitter := time.Duration(0) + if s.cfg.RepairJitter > 0 { + span := int64(s.cfg.RepairJitter)*2 + 1 + jitter = time.Duration(s.rng.Int63n(span)) - s.cfg.RepairJitter + } + at := s.now + s.cfg.RepairLatency + jitter + if at < s.now { + at = s.now + } + if at < s.nextWireAt { + at = s.nextWireAt + } + if s.cfg.BandwidthBytesPerSec > 0 { + wire := time.Duration(float64(len(packet.Bytes)) / float64(s.cfg.BandwidthBytesPerSec) * float64(time.Second)) + if wire < time.Nanosecond { + wire = time.Nanosecond + } + at += wire + s.nextWireAt = at + } + d := delivery{at: at, sequence: s.sequence, packet: packet, key: key, primary: true, fromRepair: true} + s.sequence++ + if s.cfg.PacketLoss > 0 && s.rng.Float64() < s.cfg.PacketLoss { + d.drop = true + } + if s.corruptLeft > 0 { + d.corrupt = true + s.corruptLeft-- + } + heap.Push(&s.queue, d) + if !d.drop && s.cfg.DuplicateProbability > 0 && s.rng.Float64() < s.cfg.DuplicateProbability { + dup := d + dup.at++ + dup.sequence = s.sequence + dup.primary = false + dup.duplicate = true + dup.corrupt = false + s.sequence++ + heap.Push(&s.queue, dup) + } + if len(s.queue) > s.result.QueueHighWater { + s.result.QueueHighWater = len(s.queue) + } +} + +func (s *simulation) ingest(packet Packet, fromRepair, corrupt bool) error { + raw := packet.Bytes + if corrupt { + raw = append([]byte(nil), raw...) + if len(raw) > 200 { + raw[200] ^= 0x80 + } else if len(raw) > 0 { + raw[len(raw)-1] ^= 0x80 + } + } + started := time.Now() + shred, err := turbine.ParseShred(raw) + s.stageCPU["shred_parse"] += time.Since(started) + if err != nil { + return err + } + started = time.Now() + err = shred.VerifySignature(s.ledger.LeaderPub) + s.stageCPU["shred_validation"] += time.Since(started) + if err != nil { + return err + } + s.record("shred_validation", packet, "Merkle proof and leader signature valid") + if _, ok := s.firstShred[shred.Slot]; !ok { + s.firstShred[shred.Slot] = s.now + s.record("first_shred", packet, "") + } + started = time.Now() + spooled := s.spool.AppendShred(shred, raw) + s.stageCPU["blockstore_insert"] += time.Since(started) + if spooled { + s.record("blockstore_insert", packet, "verified shred spool") + } + beforeRecovered := s.assembler.RecoveredDataShreds() + started = time.Now() + blk, err := s.assembler.AddShredFrom(shred, fromRepair) + s.stageCPU["assembler_ingest_and_recovery"] += time.Since(started) + if err != nil { + return err + } + afterRecovered := s.assembler.RecoveredDataShreds() + if afterRecovered > beforeRecovered { + s.result.FECDecodes++ + if !s.haveRecoverAt { + s.firstRecoverAt = s.now + s.haveRecoverAt = true + } + s.record("fec_threshold_reached", packet, "observed at AddShredFrom call boundary") + s.record("fec_decode_complete", packet, fmt.Sprintf("recovered_data=%d", afterRecovered-beforeRecovered)) + } + if fromRepair { + s.record("repair_response_receive", packet, "") + } else { + s.record("live_shred_receive", packet, "") + } + if blk != nil { + canonical, ok := s.ledger.Slot(blk.Slot) + if !ok { + return fmt.Errorf("completed unknown slot %d", blk.Slot) + } + if err := compareBlock(canonical, blk); err != nil { + return err + } + if _, ok := s.spool.IsComplete(blk.Slot); !ok { + return fmt.Errorf("slot %d completed without spool completion record", blk.Slot) + } + s.completed[blk.Slot] = blk + for _, data := range canonical.Data { + s.result.DataShredBytesReplayable += uint64(len(data.Bytes)) + } + s.record("slot_offered_to_replay", packet, "verified block emitted") + s.advanceReplayable() + } + return nil +} + +func (s *simulation) advanceReplayable() { + for { + if _, ok := s.completed[s.nextReplaySlot]; !ok { + return + } + s.replayableAt[s.nextReplaySlot] = s.now + s.recordAt("slot_replayable", s.nextReplaySlot, 0, 0, 0, 0, "contiguous parent chain available") + s.nextReplaySlot++ + } +} + +func compareBlock(canonical *Slot, got *block.Block) error { + if got.Slot != canonical.Number || got.SourceParentSlot != canonical.ParentSlot { + return fmt.Errorf("block identity got slot=%d parent=%d, want slot=%d parent=%d", got.Slot, got.SourceParentSlot, canonical.Number, canonical.ParentSlot) + } + if len(got.Entries) != len(canonical.Entries) { + return fmt.Errorf("slot %d entries=%d, want %d", got.Slot, len(got.Entries), len(canonical.Entries)) + } + for i := range canonical.Entries { + want := canonical.Entries[i] + entry := got.Entries[i] + if entry.NumHashes != want.NumHashes || string(entry.Hash) != string(want.Hash[:]) || len(entry.Indices) != len(want.Txns) { + return fmt.Errorf("slot %d entry %d differs from canonical ledger", got.Slot, i) + } + } + if !got.TransactionSignaturesVerified() { + return fmt.Errorf("slot %d block was not signature-verified", got.Slot) + } + return nil +} + +func (s *simulation) record(stage string, packet Packet, detail string) { + s.recordAt(stage, packet.Slot, packet.FECSetIndex, packet.Index, packet.Type, len(packet.Bytes), detail) +} + +func (s *simulation) recordAt(stage string, slot uint64, fec, index uint32, typ turbine.ShredType, bytes int, detail string) { + if s.cfg.CollectTrace { + s.trace = append(s.trace, TraceEvent{ + Sequence: s.sequence, AtNanos: int64(s.now), Stage: stage, Slot: slot, + FECSetIndex: fec, ShredIndex: index, ShredType: typ, Bytes: bytes, Detail: detail, + }) + } + s.sequence++ +} + +func summarizeLatencies(values []time.Duration) LatencySummary { + if len(values) == 0 { + return LatencySummary{} + } + sort.Slice(values, func(i, j int) bool { return values[i] < values[j] }) + percentile := func(p float64) time.Duration { + index := int(float64(len(values)-1)*p + 0.5) + return values[index] + } + return LatencySummary{P50: percentile(.50), P95: percentile(.95), P99: percentile(.99)} +} + +func initialMissingDataShreds(ledger *Ledger, availability Availability) uint64 { + var heldPerFEC int + switch availability { + case AvailabilityComplete: + heldPerFEC = 32 + case AvailabilityNearLoss: + var missing uint64 + for i := range ledger.Slots { + for fec := range ledger.Slots[i].FECs { + if fec%2 == 0 { + missing += 2 + } else { + missing++ + } + } + } + return missing + case AvailabilitySparse: + heldPerFEC = 2 + case AvailabilityMixed: + heldPerFEC = 16 + } + return uint64(len(ledger.Slots) * ledger.Config.FECsPerSlot * (dataShredsPerFEC - heldPerFEC)) +} + +type repairKey struct { + slot uint64 + index uint32 +} + +type delivery struct { + at time.Duration + sequence int + packet Packet + key repairKey + primary bool + fromRepair bool + drop bool + duplicate bool + corrupt bool +} + +type deliveryHeap []delivery + +func (h deliveryHeap) Len() int { return len(h) } +func (h deliveryHeap) Less(i, j int) bool { + if h[i].at != h[j].at { + return h[i].at < h[j].at + } + return h[i].sequence < h[j].sequence +} +func (h deliveryHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] } +func (h *deliveryHeap) Push(x any) { *h = append(*h, x.(delivery)) } +func (h *deliveryHeap) Pop() any { + old := *h + last := old[len(old)-1] + *h = old[:len(old)-1] + return last +} diff --git a/pkg/turbine/repairsim/sim_test.go b/pkg/turbine/repairsim/sim_test.go new file mode 100644 index 00000000..00cdeed3 --- /dev/null +++ b/pkg/turbine/repairsim/sim_test.go @@ -0,0 +1,214 @@ +package repairsim + +import ( + "reflect" + "testing" + "time" + + "github.com/Overclock-Validator/mithril/pkg/turbine" +) + +func testLedger(t *testing.T, slots, fecs int) *Ledger { + t.Helper() + ledger, err := GenerateLedger(LedgerConfig{ + StartSlot: 20_000, + Slots: slots, + FECsPerSlot: fecs, + Seed: 7, + ShredVersion: 11, + ReferenceTick: 63, + }) + if err != nil { + t.Fatal(err) + } + return ledger +} + +func deterministicConfig(scenario Scenario) Config { + cfg := DefaultConfig(scenario) + cfg.RepairLatency = 10 * time.Millisecond + cfg.RepairJitter = time.Millisecond + cfg.DuplicateProbability = 0 + cfg.BandwidthBytesPerSec = 0 + cfg.MaxConcurrent = 32 + cfg.Seed = 19 + return cfg +} + +func TestGenerateLedgerHasExactFECCountAndValidPackets(t *testing.T) { + ledger := testLedger(t, 2, 3) + if ledger.Config.EntriesPerSlot == 0 { + t.Fatal("entry count was not resolved") + } + for _, slot := range ledger.Slots { + if len(slot.FECs) != 3 { + t.Fatalf("slot %d FECs=%d, want 3", slot.Number, len(slot.FECs)) + } + for _, fec := range slot.FECs { + if len(fec.Data) != 32 || len(fec.Coding) != 32 { + t.Fatalf("slot %d FEC %d shape=%d+%d", slot.Number, fec.Index, len(fec.Data), len(fec.Coding)) + } + for _, packet := range append(append([]Packet(nil), fec.Data...), fec.Coding...) { + shred, err := parseAndVerify(packet, ledger) + if err != nil { + t.Fatalf("slot %d FEC %d: %v", slot.Number, fec.Index, err) + } + if shred.Slot != slot.Number || shred.FECSetIndex != fec.Index { + t.Fatalf("packet routing got slot=%d FEC=%d", shred.Slot, shred.FECSetIndex) + } + } + } + } +} + +func TestNearTipRepairCompletesAndTraceIsDeterministic(t *testing.T) { + ledger := testLedger(t, 3, 2) + cfg := deterministicConfig(ScenarioNearTip) + cfg.NaturalLateShreds = true + + first, err := Run(ledger, cfg) + if err != nil { + t.Fatal(err) + } + second, err := Run(ledger, cfg) + if err != nil { + t.Fatal(err) + } + if first.CompletedSlots != 3 || first.SpoolCompleteSlots != 3 { + t.Fatalf("completed=%d spool=%d, want 3", first.CompletedSlots, first.SpoolCompleteSlots) + } + if first.LocallyRecoveredDataShreds == 0 || first.FECDecodes == 0 { + t.Fatalf("recovered=%d decodes=%d, want both nonzero", first.LocallyRecoveredDataShreds, first.FECDecodes) + } + if first.CanceledOrLateResponses == 0 { + t.Fatal("natural late-shred scenario did not produce canceled/late repair work") + } + if !reflect.DeepEqual(first.Trace, second.Trace) { + t.Fatal("same seed/config produced different logical traces") + } +} + +func TestNearTipWithoutRepairRemainsIncomplete(t *testing.T) { + ledger := testLedger(t, 2, 2) + cfg := deterministicConfig(ScenarioNearTip) + cfg.RepairEnabled = false + cfg.NaturalLateShreds = false + result, err := Run(ledger, cfg) + if err != nil { + t.Fatal(err) + } + if result.CompletedSlots != 0 || result.SpoolCompleteSlots != 0 { + t.Fatalf("completed=%d spool=%d without repair", result.CompletedSlots, result.SpoolCompleteSlots) + } +} + +func TestCompleteDeliveryEstablishesZeroRepairBaseline(t *testing.T) { + ledger := testLedger(t, 2, 2) + cfg := deterministicConfig(ScenarioNearTip) + cfg.Availability = AvailabilityComplete + cfg.RepairEnabled = false + cfg.NaturalLateShreds = false + result, err := Run(ledger, cfg) + if err != nil { + t.Fatal(err) + } + if result.CompletedSlots != 2 { + t.Fatalf("completed=%d, want 2", result.CompletedSlots) + } + if result.RepairRequests != 0 || result.LocallyRecoveredDataShreds != 0 { + t.Fatalf("baseline requests=%d recovered=%d, want zero", result.RepairRequests, result.LocallyRecoveredDataShreds) + } +} + +func TestDeepCatchupMixedUsesThresholdRecovery(t *testing.T) { + ledger := testLedger(t, 8, 2) + cfg := deterministicConfig(ScenarioDeepCatchup) + cfg.Availability = AvailabilityMixed + cfg.NaturalLateShreds = false + result, err := Run(ledger, cfg) + if err != nil { + t.Fatal(err) + } + if result.CompletedSlots != len(ledger.Slots) { + t.Fatalf("completed=%d, want %d", result.CompletedSlots, len(ledger.Slots)) + } + if result.LocallyRecoveredDataShreds <= result.UsefulNetworkDataShreds { + t.Fatalf("local recovery=%d, network data=%d; mixed threshold scenario should recover most losses locally", result.LocallyRecoveredDataShreds, result.UsefulNetworkDataShreds) + } + if result.RepairRequests == 0 || result.RepairBytesReceived == 0 { + t.Fatal("deep catch-up completed without exercising repair") + } +} + +func TestCorruptRepairResponseRejectedThenRetried(t *testing.T) { + ledger := testLedger(t, 1, 1) + cfg := deterministicConfig(ScenarioDeepCatchup) + cfg.Availability = AvailabilityMixed + cfg.CorruptResponses = 1 + result, err := Run(ledger, cfg) + if err != nil { + t.Fatal(err) + } + if result.CompletedSlots != 1 { + t.Fatalf("completed=%d, want 1", result.CompletedSlots) + } + if result.RejectedCorruptResponses != 1 { + t.Fatalf("rejected corrupt=%d, want 1", result.RejectedCorruptResponses) + } + if result.RepairRequests < 2 { + t.Fatalf("requests=%d, want retry after corruption", result.RepairRequests) + } +} + +func parseAndVerify(packet Packet, ledger *Ledger) (*turbine.Shred, error) { + shred, err := turbine.ParseShred(packet.Bytes) + if err != nil { + return nil, err + } + if err := shred.VerifySignature(ledger.LeaderPub); err != nil { + return nil, err + } + return shred, nil +} + +func BenchmarkScenarios(b *testing.B) { + ledger, err := GenerateLedger(LedgerConfig{ + StartSlot: 30_000, Slots: 8, FECsPerSlot: 2, Seed: 23, ShredVersion: 1, ReferenceTick: 63, + }) + if err != nil { + b.Fatal(err) + } + tests := []struct { + name string + scenario Scenario + availability Availability + }{ + {name: "near-tip", scenario: ScenarioNearTip, availability: AvailabilityNearLoss}, + {name: "deep-mixed", scenario: ScenarioDeepCatchup, availability: AvailabilityMixed}, + {name: "deep-sparse", scenario: ScenarioDeepCatchup, availability: AvailabilitySparse}, + } + for _, tt := range tests { + b.Run(tt.name, func(b *testing.B) { + cfg := DefaultConfig(tt.scenario) + cfg.Availability = tt.availability + cfg.RepairLatency = 0 + cfg.RepairJitter = 0 + cfg.DuplicateProbability = 0 + cfg.BandwidthBytesPerSec = 0 + cfg.NaturalLateShreds = false + cfg.CollectTrace = false + b.ReportAllocs() + for i := 0; i < b.N; i++ { + result, err := Run(ledger, cfg) + if err != nil { + b.Fatal(err) + } + if result.CompletedSlots != len(ledger.Slots) { + b.Fatalf("completed=%d", result.CompletedSlots) + } + b.ReportMetric(float64(result.RepairRequests), "repair-requests/op") + b.ReportMetric(float64(result.LocallyRecoveredDataShreds), "recovered-shreds/op") + } + }) + } +} From 06282a3cfadee85b3c265aa53239c36033a8420b Mon Sep 17 00:00:00 2001 From: 7layermagik <7layermagik@users.noreply.github.com> Date: Thu, 30 Jul 2026 01:45:19 -0500 Subject: [PATCH 7/9] test(turbine): mirror receiver signature caching in repair sim --- cmd/repair-sim/main.go | 7 ++++- docs/repair_sim.md | 4 +-- docs/results/repair-sim/RESULTS_TEMPLATE.md | 2 +- pkg/turbine/repairsim/sim.go | 6 ++++- pkg/turbine/repairsim/sim_test.go | 4 +++ pkg/turbine/sigcache.go | 29 ++++++++++++++++++--- 6 files changed, 43 insertions(+), 9 deletions(-) diff --git a/cmd/repair-sim/main.go b/cmd/repair-sim/main.go index 3a95fead..5eebf05e 100644 --- a/cmd/repair-sim/main.go +++ b/cmd/repair-sim/main.go @@ -47,6 +47,7 @@ func main() { corrupt = flag.Int("corrupt-responses", 0, "corrupt the first N repair responses") naturalLate = flag.Bool("natural-late", true, "schedule selected late live shreds during repair") spoolDir = flag.String("spool-dir", "", "persistent shred-spool directory (empty uses a temporary directory)") + cpuLabel = flag.String("cpu-label", "", "explicit CPU label when platform discovery is unavailable") output = flag.String("output", "", "write JSON to this file instead of stdout") includeTrace = flag.Bool("trace", true, "include the logical event trace in JSON") ) @@ -90,8 +91,12 @@ func main() { if err != nil { fatalf("run simulation: %v", err) } + cpu := *cpuLabel + if cpu == "" { + cpu = cpuModel() + } report := report{ - Environment: environment{GoVersion: runtime.Version(), GOOS: runtime.GOOS, GOARCH: runtime.GOARCH, CPU: cpuModel()}, + Environment: environment{GoVersion: runtime.Version(), GOOS: runtime.GOOS, GOARCH: runtime.GOARCH, CPU: cpu}, Ledger: ledger.Config, Network: network, LedgerGenerationWall: generationWall, Result: result, } encoded, err := json.MarshalIndent(report, "", " ") diff --git a/docs/repair_sim.md b/docs/repair_sim.md index f70fc429..2f36cd48 100644 --- a/docs/repair_sim.md +++ b/docs/repair_sim.md @@ -18,7 +18,7 @@ network realism or alternative scheduling policies are introduced. | block-component serialization | production `turbine.MarshalBlockComponent` | | 32+32 FEC generation and Merkle signing | production `turbine.Shredder` | | missing-shred selection | production `SlotAssembler.RepairRequests` | -| packet parsing and Merkle/signature validation | production `ParseShred` and `VerifySignature` | +| packet parsing and Merkle/signature validation | production `ParseShred` and `ShredSignatureVerifier` (the receiver's per-root cache) | | verified-shred insertion | production `ShredSpool` | | threshold detection and Reed-Solomon recovery | production `SlotAssembler.AddShredFrom` | | component decode and transaction-signature gate | production slot completion path | @@ -64,6 +64,7 @@ go run ./cmd/repair-sim \ -slots=200 \ -fec-sets=4 \ -seed=1 \ + -cpu-label='Ryzen 7 9700X' \ -output=/tmp/repair-near.json go run ./cmd/repair-sim \ @@ -146,4 +147,3 @@ local spool already holds most FEC thresholds. 4. Run replay execution against a reusable synthetic bank fixture. 5. Compare ordinary requests with explicit test-only threshold-acquisition and earliest-blocked-slot policies. - diff --git a/docs/results/repair-sim/RESULTS_TEMPLATE.md b/docs/results/repair-sim/RESULTS_TEMPLATE.md index e11bce84..530939d1 100644 --- a/docs/results/repair-sim/RESULTS_TEMPLATE.md +++ b/docs/results/repair-sim/RESULTS_TEMPLATE.md @@ -18,5 +18,5 @@ Also report: - queue high-water mark; - spool bytes and complete slots; - allocations; +- shred-signature cache hits and actual Ed25519 verifications; - every limitation emitted in the JSON result. - diff --git a/pkg/turbine/repairsim/sim.go b/pkg/turbine/repairsim/sim.go index dbabcd9e..438b5ecd 100644 --- a/pkg/turbine/repairsim/sim.go +++ b/pkg/turbine/repairsim/sim.go @@ -124,6 +124,8 @@ type Result struct { CanceledOrLateResponses uint64 `json:"canceled_or_late_responses"` LostResponses uint64 `json:"lost_responses"` RejectedCorruptResponses uint64 `json:"rejected_corrupt_responses"` + ShredSignatureCacheHits uint64 `json:"shred_signature_cache_hits"` + ShredEd25519Verifications uint64 `json:"shred_ed25519_verifications"` QueueHighWater int `json:"queue_high_water"` SpoolBytes int64 `json:"spool_bytes"` SpoolCompleteSlots int `json:"spool_complete_slots"` @@ -231,6 +233,7 @@ func Run(ledger *Ledger, cfg Config) (Result, error) { result.Allocations = memAfter.Mallocs - memBefore.Mallocs result.StageCPU = s.stageCPU result.Trace = s.trace + result.ShredSignatureCacheHits, result.ShredEd25519Verifications = s.shredVerifier.Stats() result.Limitations = []string{ "remote peers and latency are simulated in process; no UDP/IP stack is measured", "synthetic entries contain no transactions, so transaction execution is not measured", @@ -261,6 +264,7 @@ type simulation struct { ledger *Ledger cfg Config assembler *turbine.SlotAssembler + shredVerifier turbine.ShredSignatureVerifier spool *turbine.ShredSpool rng *rand.Rand now time.Duration @@ -523,7 +527,7 @@ func (s *simulation) ingest(packet Packet, fromRepair, corrupt bool) error { return err } started = time.Now() - err = shred.VerifySignature(s.ledger.LeaderPub) + err = s.shredVerifier.Verify(shred, s.ledger.LeaderPub) s.stageCPU["shred_validation"] += time.Since(started) if err != nil { return err diff --git a/pkg/turbine/repairsim/sim_test.go b/pkg/turbine/repairsim/sim_test.go index 00cdeed3..607c900f 100644 --- a/pkg/turbine/repairsim/sim_test.go +++ b/pkg/turbine/repairsim/sim_test.go @@ -118,6 +118,10 @@ func TestCompleteDeliveryEstablishesZeroRepairBaseline(t *testing.T) { if result.RepairRequests != 0 || result.LocallyRecoveredDataShreds != 0 { t.Fatalf("baseline requests=%d recovered=%d, want zero", result.RepairRequests, result.LocallyRecoveredDataShreds) } + if result.ShredEd25519Verifications != 4 || result.ShredSignatureCacheHits != 124 { + t.Fatalf("signature cache verifies=%d hits=%d, want 4/124 for four FEC roots", + result.ShredEd25519Verifications, result.ShredSignatureCacheHits) + } } func TestDeepCatchupMixedUsesThresholdRecovery(t *testing.T) { diff --git a/pkg/turbine/sigcache.go b/pkg/turbine/sigcache.go index 7f3b1313..46358b07 100644 --- a/pkg/turbine/sigcache.go +++ b/pkg/turbine/sigcache.go @@ -20,7 +20,13 @@ import ( // hit reproduces exactly the result of re-running it on the same inputs. // Tampered content can never hit — different bytes yield a different root, // hence a different key. Failures are never cached. -type shredSigCache struct { +// ShredSignatureVerifier authenticates Merkle shreds with the same bounded, +// per-root result cache used by UDPReceiver. The cache never stores failures; +// each packet's Merkle proof is still evaluated before a cache lookup. +// +// It is exported so deterministic and loopback ingress harnesses can exercise +// production validation without constructing a UDPReceiver. +type ShredSignatureVerifier struct { mu sync.Mutex cur map[shredSigCacheKey]struct{} prev map[shredSigCacheKey]struct{} @@ -29,6 +35,10 @@ type shredSigCache struct { verifies atomic.Uint64 } +// Keep the internal receiver/test name as an alias; there is one +// implementation and one cache contract. +type shredSigCache = ShredSignatureVerifier + type shredSigCacheKey struct { leader solana.PublicKey root solana.Hash @@ -42,7 +52,7 @@ const shredSigCacheGenCap = 4096 // verifyShred authenticates a shred exactly like Shred.VerifySignature, with // the per-root ed25519 result cached. -func (c *shredSigCache) verifyShred(s *Shred, leader solana.PublicKey) error { +func (c *ShredSignatureVerifier) verifyShred(s *Shred, leader solana.PublicKey) error { root, err := s.MerkleRoot() if err != nil { return err @@ -74,7 +84,13 @@ func (c *shredSigCache) verifyShred(s *Shred, leader solana.PublicKey) error { return nil } -func (c *shredSigCache) addLocked(key shredSigCacheKey) { +// Verify authenticates one shred and retains successful root/signature tuples +// for sibling shreds in the same FEC set. +func (c *ShredSignatureVerifier) Verify(s *Shred, leader solana.PublicKey) error { + return c.verifyShred(s, leader) +} + +func (c *ShredSignatureVerifier) addLocked(key shredSigCacheKey) { if c.cur == nil { c.cur = make(map[shredSigCacheKey]struct{}, shredSigCacheGenCap) } @@ -85,6 +101,11 @@ func (c *shredSigCache) addLocked(key shredSigCacheKey) { c.cur[key] = struct{}{} } -func (c *shredSigCache) stats() (hits, verifies uint64) { +func (c *ShredSignatureVerifier) stats() (hits, verifies uint64) { return c.hits.Load(), c.verifies.Load() } + +// Stats reports cache hits and actual Ed25519 verifications. +func (c *ShredSignatureVerifier) Stats() (hits, verifies uint64) { + return c.stats() +} From 4cec6916f0daf9a3fff3af92f452ec8be0380f11 Mon Sep 17 00:00:00 2001 From: 7layermagik <7layermagik@users.noreply.github.com> Date: Thu, 30 Jul 2026 15:14:26 -0500 Subject: [PATCH 8/9] perf(turbine): specialize one-missing FEC recovery --- pkg/turbine/assembler.go | 35 +++++- pkg/turbine/internal/rsrecover/recover.go | 69 +++++++++++ .../internal/rsrecover/recover_test.go | 15 +++ pkg/turbine/recover_one_data_test.go | 113 ++++++++++++++++++ 4 files changed, 226 insertions(+), 6 deletions(-) create mode 100644 pkg/turbine/recover_one_data_test.go diff --git a/pkg/turbine/assembler.go b/pkg/turbine/assembler.go index 9b69f437..5824f13c 100644 --- a/pkg/turbine/assembler.go +++ b/pkg/turbine/assembler.go @@ -10,6 +10,7 @@ import ( "github.com/Overclock-Validator/mithril/pkg/block" "github.com/Overclock-Validator/mithril/pkg/statsd" + "github.com/Overclock-Validator/mithril/pkg/turbine/internal/rsrecover" "github.com/gagliardetto/solana-go" "github.com/klauspost/reedsolomon" ) @@ -1300,21 +1301,43 @@ func (a *SlotAssembler) recoverFEC(state *slotState, fecSetIndex uint32) ([]*Shr } shards[int(layout.dataShreds)+int(pos)] = shard } - encoder, err := a.fecEncoder(layout) - if err != nil { - return nil, err - } - required := make([]bool, int(layout.dataShreds)+int(layout.codingShreds)) var missingData int + missingDataIndex := -1 for idx := 0; idx < int(layout.dataShreds); idx++ { if fec.data[uint32(idx)] == nil { - required[idx] = true missingData++ + missingDataIndex = idx } } if missingData == 0 { return nil, nil } + if missingData == 1 && + layout.dataShreds == rsrecover.DataShards && + layout.codingShreds == rsrecover.CodingShards { + presence, err := rsrecover.Presence(shards) + if err != nil { + return nil, err + } + dst := make([]byte, layout.shardSize) + if err := rsrecover.RecoverOneData(presence, missingDataIndex, shards, dst); err != nil { + return nil, fmt.Errorf("recover one FEC data shred slot %d fec_set=%d: %w", state.slot, fecSetIndex, err) + } + shred, err := fec.recoveredDataShred(uint32(missingDataIndex), dst) + if err != nil { + return nil, err + } + return []*Shred{shred}, nil + } + + required := make([]bool, int(layout.dataShreds)+int(layout.codingShreds)) + for idx := 0; idx < int(layout.dataShreds); idx++ { + required[idx] = fec.data[uint32(idx)] == nil + } + encoder, err := a.fecEncoder(layout) + if err != nil { + return nil, err + } if err := encoder.ReconstructSome(shards, required); err != nil { if errors.Is(err, reedsolomon.ErrTooFewShards) { return nil, nil diff --git a/pkg/turbine/internal/rsrecover/recover.go b/pkg/turbine/internal/rsrecover/recover.go index dd48e99d..6a16e91e 100644 --- a/pkg/turbine/internal/rsrecover/recover.go +++ b/pkg/turbine/internal/rsrecover/recover.go @@ -3,6 +3,7 @@ package rsrecover import ( "errors" "fmt" + "math/bits" "sync" "github.com/klauspost/reedsolomon" @@ -27,6 +28,12 @@ var ( allCodingEncoderOnce sync.Once allCodingEncoder reedsolomon.Encoder allCodingEncoderErr error + + // oneDataCoefficientRows[missing][coding] contains one coefficient for + // each data input followed by the selected coding input. Every one of the + // fixed 32x32 rows is exhaustively differential-tested against the general + // Reed-Solomon decoder. + oneDataCoefficientRows [DataShards][CodingShards][DataShards + 1]byte ) func init() { @@ -42,6 +49,22 @@ func init() { for exponent := 255; exponent < len(gfExp); exponent++ { gfExp[exponent] = gfExp[exponent-255] } + for missing := 0; missing < DataShards; missing++ { + for coding := 0; coding < CodingShards; coding++ { + // If c = sum(a_i*d_i), then the missing d_m is + // inv(a_m) * (c + sum(i != m, a_i*d_i)) over GF(2^8). + coefficientInv := reedsolomon.Inv(codingCoefficient(coding, missing)) + for data := 0; data < DataShards; data++ { + if data != missing { + oneDataCoefficientRows[missing][coding][data] = gfMul( + coefficientInv, + codingCoefficient(coding, data), + ) + } + } + oneDataCoefficientRows[missing][coding][DataShards] = coefficientInv + } + } } // OneDataPlan recovers exactly one absent data shard from the other 31 data @@ -175,6 +198,52 @@ func (plan *OneDataPlan) Recover(shards [][]byte, dst []byte) error { return nil } +// RecoverOneData uses a process-wide table of coefficient rows that are +// exhaustively differential-tested against the general decoder. +// for the fixed 32+32 matrix. It is the setup-free counterpart to +// PrepareRecoverOneData for a changing stream of one-missing FEC patterns. +// Validation completes before dst is modified. +func RecoverOneData(presence uint64, missingDataIndex int, shards [][]byte, dst []byte) error { + if missingDataIndex < 0 || missingDataIndex >= DataShards { + return fmt.Errorf("%w: missing data index %d", ErrInvalidPattern, missingDataIndex) + } + const dataMask = uint64(1)<> DataShards) + if codingMask == 0 { + return fmt.Errorf("%w: no coding shard is available", ErrInvalidPattern) + } + shardSize, err := validateExecution(presence, shards, [][]byte{dst}) + if err != nil { + return err + } + if len(dst) != shardSize { + return fmt.Errorf("%w: destination has %d bytes, want %d", ErrInvalidBuffers, len(dst), shardSize) + } + + codingPosition := bits.TrailingZeros32(codingMask) + row := &oneDataCoefficientRows[missingDataIndex][codingPosition] + var lowLevel reedsolomon.LowLevel + first := true + for dataIndex := 0; dataIndex < DataShards; dataIndex++ { + coefficient := row[dataIndex] + if coefficient == 0 { + continue + } + if first { + lowLevel.GalMulSlice(coefficient, shards[dataIndex], dst) + first = false + } else { + lowLevel.GalMulSliceXor(coefficient, shards[dataIndex], dst) + } + } + lowLevel.GalMulSliceXor(row[DataShards], shards[DataShards+codingPosition], dst) + return nil +} + // PrepareRecoverDataSubset constructs direct output rows for all absent data // shards. It first inverts only the reduced m x m coding/data matrix, then // expands those rows over exactly 32 selected input shards so byte execution diff --git a/pkg/turbine/internal/rsrecover/recover_test.go b/pkg/turbine/internal/rsrecover/recover_test.go index 00e3b5c0..3d722a48 100644 --- a/pkg/turbine/internal/rsrecover/recover_test.go +++ b/pkg/turbine/internal/rsrecover/recover_test.go @@ -145,6 +145,13 @@ func TestRecoverOneDataAllPositionsAndCodingRows(t *testing.T) { if !bytes.Equal(dst, original[missing]) { t.Fatalf("missing=%d coding=%d: recovered bytes differ", missing, coding) } + fastDst := bytes.Repeat([]byte{0x5a}, len(original[missing])) + if err := RecoverOneData(presence, missing, available, fastDst); err != nil { + t.Fatalf("fast missing=%d coding=%d: %v", missing, coding, err) + } + if !bytes.Equal(fastDst, dst) { + t.Fatalf("fast missing=%d coding=%d: direct rows differ", missing, coding) + } if coding == 0 { reference = append([]byte(nil), dst...) } else if !bytes.Equal(dst, reference) { @@ -176,6 +183,14 @@ func TestRecoverOneDataRejectsUnsafePatternsAtomically(t *testing.T) { if !bytes.Equal(dst, want) { t.Fatal("destination changed after a validation error") } + fastDst := bytes.Repeat([]byte{0x4e}, len(original[0])) + fastWant := append([]byte(nil), fastDst...) + if err := RecoverOneData(presence, 7, changed, fastDst); !errors.Is(err, ErrPatternChanged) { + t.Fatalf("fast changed pattern error = %v, want ErrPatternChanged", err) + } + if !bytes.Equal(fastDst, fastWant) { + t.Fatal("fast destination changed after a validation error") + } insufficient := availableFixture(original, []int{7, 8}, []int{3}) insufficientPresence, err := Presence(insufficient) diff --git a/pkg/turbine/recover_one_data_test.go b/pkg/turbine/recover_one_data_test.go new file mode 100644 index 00000000..982bed7b --- /dev/null +++ b/pkg/turbine/recover_one_data_test.go @@ -0,0 +1,113 @@ +package turbine + +import ( + "bytes" + "testing" + + "github.com/gagliardetto/solana-go" +) + +var benchmarkRecoveredShredsSink []*Shred + +type recoverFECOneMissingFixture struct { + assembler *SlotAssembler + state *slotState + fecSet uint32 + want *Shred +} + +func makeRecoverFECOneMissingFixture(tb testing.TB) recoverFECOneMissingFixture { + tb.Helper() + gen := ShredGenerator{Slot: 10, ParentSlot: 9, Version: 1} + payloadSize := dataShredsPerFECBlock * dataCapacity(proofEntriesFor32x32, false) + packets, _, _, _, err := gen.MakeShredsFromData( + benchmarkLeaderKey(), benchmarkPayload(payloadSize), false, solana.Hash{}, 0, 0, + ) + if err != nil { + tb.Fatal(err) + } + if len(packets) != dataShredsPerFECBlock+codingShredsPerFECBlock { + tb.Fatalf("fixture packets=%d", len(packets)) + } + + var fixture recoverFECOneMissingFixture + for _, packet := range packets { + shred, err := ParseShred(packet) + if err != nil { + tb.Fatal(err) + } + if fixture.state == nil { + fixture.state = &slotState{ + slot: shred.Slot, + shreds: make(map[uint32]*Shred), + fecSets: make(map[uint32]*fecState), + shredVer: shred.Version, + lastIndex: ^uint32(0), + } + fixture.fecSet = shred.FECSetIndex + } + switch shred.Type { + case ShredTypeData: + if shred.Index-shred.FECSetIndex == 1 { + fixture.want = shred + continue + } + if err := fixture.state.addDataShred(shred); err != nil { + tb.Fatal(err) + } + case ShredTypeCode: + if err := fixture.state.addCodingShred(shred); err != nil { + tb.Fatal(err) + } + } + } + if fixture.want == nil { + tb.Fatal("fixture did not omit one data shred") + } + fixture.assembler = NewSlotAssembler() + return fixture +} + +func TestRecoverFECOneMissingFixed32x32(t *testing.T) { + fixture := makeRecoverFECOneMissingFixture(t) + recovered, err := fixture.assembler.recoverFEC(fixture.state, fixture.fecSet) + if err != nil { + t.Fatal(err) + } + if len(recovered) != 1 { + t.Fatalf("recovered %d shreds, want 1", len(recovered)) + } + if !recovered[0].Recovered { + t.Fatal("recovered shred is not marked recovered") + } + gotShard, err := recovered[0].erasureShard() + if err != nil { + t.Fatal(err) + } + wantShard, err := fixture.want.erasureShard() + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(gotShard, wantShard) || !bytes.Equal(recovered[0].Data, fixture.want.Data) { + t.Fatal("recovered erasure shard differs from the original") + } +} + +func BenchmarkRecoverFECOneMissingBoundary(b *testing.B) { + fixture := makeRecoverFECOneMissingFixture(b) + if _, err := fixture.assembler.recoverFEC(fixture.state, fixture.fecSet); err != nil { + b.Fatal(err) + } + b.ReportAllocs() + b.ResetTimer() + for b.Loop() { + recovered, err := fixture.assembler.recoverFEC(fixture.state, fixture.fecSet) + if err != nil { + b.Fatal(err) + } + if len(recovered) != 1 { + b.Fatalf("recovered %d shreds", len(recovered)) + } + benchmarkRecoveredShredsSink = recovered + } +} From 847768930cde10b6c885e4a82648d192b34098a9 Mon Sep 17 00:00:00 2001 From: 7layermagik <7layermagik@users.noreply.github.com> Date: Thu, 30 Jul 2026 15:15:49 -0500 Subject: [PATCH 9/9] docs(turbine): record one-missing recovery gate --- docs/erasure_recovery_experiments.md | 27 ++++++++++++++++++++++++++- 1 file changed, 26 insertions(+), 1 deletion(-) diff --git a/docs/erasure_recovery_experiments.md b/docs/erasure_recovery_experiments.md index dc61d77c..dc2ed2d2 100644 --- a/docs/erasure_recovery_experiments.md +++ b/docs/erasure_recovery_experiments.md @@ -1,6 +1,8 @@ # Erasure recovery experiments -Status: experimental; no production `SlotAssembler` dispatch uses these paths. +Status: `SlotAssembler` dispatches the fixed 32+32, exactly-one-missing-data +case to the direct recovery path. Reduced multi-missing and all-coding paths +remain experimental and are not used by production dispatch. The end-to-end deterministic harness that drives production repair selection, assembly, storage, and completion is documented in [repair_sim.md](repair_sim.md). @@ -68,6 +70,11 @@ D_m = C[r,m]^-1 * P_r This prepares one 32-source coefficient row and writes one destination. It does not construct or invert a general 32x32 matrix. +Production uses a process-wide table containing every missing-data and +coding-row combination. This removes per-call plan construction while keeping +the same equation. The table is exhaustively differential-tested against the +general decoder across all 32 x 32 combinations. + ### Catch up: reduced missing-data system For missing data columns `M` and selected coding rows `R`, substitute every @@ -131,6 +138,24 @@ No production dispatch threshold should be chosen from an Apple benchmark. Final crossover decisions require the pinned amd64 target and synthetic arrival traces for progressing, stalled, and bursty slots. +## Zen 5 production gate + +The direct one-data path was measured on a Ryzen 7 9700X with Go 1.26.4, +`GOMAXPROCS=1`, and one pinned physical core. Medians below are from seven +sequential one-second samples unless otherwise noted. + +| Benchmark | General path | Direct one-data path | Change | +| --- | ---: | ---: | ---: | +| one-missing `SlotAssembler` boundary | 10.73 us/FEC | 2.82 us/FEC | -73.7% (3.8x) | +| near-tip repair simulation | 3.0295 ms/op | 2.8659 ms/op | -5.40% | +| deep-mixed repair simulation | 4.0343 ms/op | 4.0240 ms/op | -0.26% | +| deep-sparse repair simulation | 10.8489 ms/op | 10.8327 ms/op | -0.15% | + +The production dispatch is intentionally narrow. The deep scenarios do not +enter it and remain effectively neutral, while the near-tip workload benefits +from repeated exactly-one-missing recoveries. The one-missing boundary also +dropped from 144 to 5 allocations per operation. + ## Preliminary Apple M4 Pro diagnostic These single-sample medians use 987-byte shards and exist only to reject or