consensus/engine/gpu_batch_pipeline.go

// Copyright (C) 2019-2025, Lux Industries Inc. All rights reserved.
// See the file LICENSE for licensing terms.

package engine

import (
	"context"
	"crypto/ed25519"
	"crypto/sha256"
	"errors"
	"fmt"
	"sync"
	"sync/atomic"
	"time"

	"github.com/luxfi/accel"
	"github.com/luxfi/consensus/types"
	"github.com/luxfi/crypto/bls"
	"github.com/luxfi/crypto/mldsa"
	"github.com/luxfi/crypto/secp256k1"
)

// GPU Batch Pipeline for ultra-low-latency consensus processing.
// Implements double-buffered pipeline with GPU-resident state to minimize
// CPU-GPU transfers. Designed for maximum throughput with parallel:
// - Signature verification (ECDSA/Ed25519/BLS batch)
// - Merkle/Verkle tree updates
// - Vote aggregation

// Pipeline errors
var (
	ErrPipelineStopped    = errors.New("pipeline stopped")
	ErrBatchTooLarge      = errors.New("batch exceeds maximum size")
	ErrGPUUnavailable     = errors.New("GPU unavailable, using CPU fallback")
	ErrInvalidTransaction = errors.New("invalid transaction in batch")
	ErrBufferFull         = errors.New("buffer full, backpressure applied")
)

// SignatureType identifies cryptographic signature algorithms.
type SignatureType uint8

const (
	SigECDSA SignatureType = iota
	SigEd25519
	SigBLS
	SigMLDSA // Post-quantum
)

// Transaction represents a consensus transaction with signature data.
type Transaction struct {
	ID        types.ID
	Hash      [32]byte
	Signature []byte
	PublicKey []byte
	SigType   SignatureType
	Payload   []byte
	Timestamp time.Time
}

// BatchResult contains the results of processing a transaction batch.
type BatchResult struct {
	BatchID         uint64
	ProcessedCount  int
	ValidCount      int
	InvalidCount    int
	MerkleRoot      [32]byte
	ProcessingTime  time.Duration
	GPUTime         time.Duration
	Errors          []TransactionError
	SignatureProofs []SignatureProof
}

// TransactionError records a failed transaction.
type TransactionError struct {
	TxID  types.ID
	Error error
}

// SignatureProof contains verification proof for a signature.
type SignatureProof struct {
	TxID     types.ID
	Valid    bool
	SigType  SignatureType
	Duration time.Duration
}

// GPUBuffer holds transaction data in GPU-resident memory.
// Uses pinned memory for efficient CPU-GPU transfers.
type GPUBuffer struct {
	id       int
	capacity int

	// Transaction data (CPU-side, copied to GPU)
	txHashes   [][32]byte
	signatures [][]byte
	publicKeys [][]byte
	sigTypes   []SignatureType

	// Results (GPU-side, copied back)
	validFlags  []bool
	merkleNodes [][32]byte

	// GPU handles (opaque pointers to device memory)
	gpuHashes     uintptr
	gpuSignatures uintptr
	gpuPubKeys    uintptr
	gpuResults    uintptr
	gpuMerkle     uintptr

	// State
	count    int
	inUse    atomic.Bool
	uploaded atomic.Bool
}

// GPUStream represents an async GPU execution stream.
type GPUStream struct {
	id     int
	handle uintptr
}

// PipelineConfig configures the GPU batch pipeline.
type PipelineConfig struct {
	// BatchSize is the number of transactions per batch
	BatchSize int

	// BufferCount is the number of GPU buffers (2 for double-buffering)
	BufferCount int

	// MaxPendingBatches limits queued batches for backpressure
	MaxPendingBatches int

	// EnableGPU enables GPU acceleration (falls back to CPU if unavailable)
	EnableGPU bool

	// GPUDeviceID specifies which GPU to use (-1 for auto-select)
	GPUDeviceID int

	// ParallelVerify enables parallel signature verification
	ParallelVerify bool

	// VerifyWorkers is the number of CPU verification workers (when GPU unavailable)
	VerifyWorkers int

	// MerkleTreeDepth is the depth of the Merkle tree maintained on GPU
	MerkleTreeDepth int

	// VerkleEnabled uses Verkle trees instead of Merkle
	VerkleEnabled bool
}

// DefaultPipelineConfig returns production-ready defaults.
func DefaultPipelineConfig() PipelineConfig {
	return PipelineConfig{
		BatchSize:         1024,
		BufferCount:       2, // Double-buffer
		MaxPendingBatches: 8,
		EnableGPU:         true,
		GPUDeviceID:       -1, // Auto-select
		ParallelVerify:    true,
		VerifyWorkers:     4,
		MerkleTreeDepth:   20, // 2^20 = 1M leaves
		VerkleEnabled:     false,
	}
}

// GPUBatchPipeline processes transaction batches with GPU acceleration.
// Implements double-buffered pipeline: while batch N processes, batch N+1 loads.
type GPUBatchPipeline struct {
	config PipelineConfig

	// Double-buffered GPU memory
	buffers      []*GPUBuffer
	activeBuffer int
	bufferMu     sync.Mutex

	// GPU execution streams for parallel load/execute
	loadStream    *GPUStream
	executeStream *GPUStream

	// Batch tracking
	nextBatchID   atomic.Uint64
	pendingCount  atomic.Int32
	processedStat atomic.Uint64

	// GPU-resident Merkle/Verkle tree state
	merkleTree *GPUMerkleTree

	// Results channel
	results chan *BatchResult

	// Lifecycle
	ctx    context.Context
	cancel context.CancelFunc
	wg     sync.WaitGroup

	// State
	running  atomic.Bool
	gpuReady atomic.Bool

	// Metrics
	metrics *PipelineMetrics
}

// PipelineMetrics tracks performance statistics.
type PipelineMetrics struct {
	TotalBatches      atomic.Uint64
	TotalTransactions atomic.Uint64
	TotalValid        atomic.Uint64
	TotalInvalid      atomic.Uint64
	TotalGPUTime      atomic.Int64 // nanoseconds
	TotalCPUTime      atomic.Int64
	BufferSwaps       atomic.Uint64
	BackpressureCount atomic.Uint64
}

// GPUMerkleTree maintains Merkle tree state in GPU memory.
type GPUMerkleTree struct {
	depth     int
	nodeCount int

	// GPU-resident nodes (bottom-up layout)
	gpuNodes uintptr

	// Root cached on CPU after computation
	root   [32]byte
	rootMu sync.RWMutex

	// Pending leaves to insert
	pending [][32]byte
	pendMu  sync.Mutex
}

// NewGPUBatchPipeline creates a new GPU-accelerated batch pipeline.
func NewGPUBatchPipeline(config PipelineConfig) (*GPUBatchPipeline, error) {
	ctx, cancel := context.WithCancel(context.Background())

	p := &GPUBatchPipeline{
		config:  config,
		buffers: make([]*GPUBuffer, config.BufferCount),
		results: make(chan *BatchResult, config.MaxPendingBatches),
		ctx:     ctx,
		cancel:  cancel,
		metrics: &PipelineMetrics{},
	}

	// Initialize GPU buffers
	for i := 0; i < config.BufferCount; i++ {
		p.buffers[i] = newGPUBuffer(i, config.BatchSize)
	}

	// Initialize GPU if enabled
	if config.EnableGPU {
		if err := p.initGPU(); err != nil {
			// Log warning but continue with CPU fallback
			p.gpuReady.Store(false)
		} else {
			p.gpuReady.Store(true)
		}
	}

	// Initialize Merkle tree
	p.merkleTree = newGPUMerkleTree(config.MerkleTreeDepth, p.gpuReady.Load())

	return p, nil
}

// newGPUBuffer allocates a GPU buffer with pinned memory.
func newGPUBuffer(id, capacity int) *GPUBuffer {
	return &GPUBuffer{
		id:          id,
		capacity:    capacity,
		txHashes:    make([][32]byte, capacity),
		signatures:  make([][]byte, capacity),
		publicKeys:  make([][]byte, capacity),
		sigTypes:    make([]SignatureType, capacity),
		validFlags:  make([]bool, capacity),
		merkleNodes: make([][32]byte, capacity),
	}
}

// newGPUMerkleTree creates a GPU-resident Merkle tree.
func newGPUMerkleTree(depth int, gpuEnabled bool) *GPUMerkleTree {
	nodeCount := (1 << (depth + 1)) - 1 // 2^(d+1) - 1 nodes
	return &GPUMerkleTree{
		depth:     depth,
		nodeCount: nodeCount,
		pending:   make([][32]byte, 0, 1024),
	}
}

// initGPU initializes GPU resources.
func (p *GPUBatchPipeline) initGPU() error {
	// Check GPU availability via luxcpp/gpu
	if !gpuAvailable() {
		return ErrGPUUnavailable
	}

	// Create execution streams
	p.loadStream = &GPUStream{id: 0}
	p.executeStream = &GPUStream{id: 1}

	// Allocate GPU memory for each buffer
	for _, buf := range p.buffers {
		if err := p.allocateGPUMemory(buf); err != nil {
			return fmt.Errorf("failed to allocate GPU memory: %w", err)
		}
	}

	// Allocate GPU memory for Merkle tree
	if err := p.allocateMerkleGPU(); err != nil {
		return fmt.Errorf("failed to allocate Merkle GPU memory: %w", err)
	}

	return nil
}

// allocateGPUMemory allocates device memory for a buffer.
// Actual GPU allocation is in gpu_batch_metal.go; this is the fallback path.
func (p *GPUBatchPipeline) allocateGPUMemory(buf *GPUBuffer) error {
	return nil
}

// allocateMerkleGPU allocates GPU memory for the Merkle tree.
func (p *GPUBatchPipeline) allocateMerkleGPU() error {
	// GPU memory allocation for Merkle nodes
	return nil
}

// Start begins the pipeline processing loop.
func (p *GPUBatchPipeline) Start() error {
	if p.running.Swap(true) {
		return errors.New("pipeline already running")
	}

	p.wg.Add(1)
	go p.processLoop()

	return nil
}

// Stop gracefully shuts down the pipeline.
func (p *GPUBatchPipeline) Stop() error {
	if !p.running.Swap(false) {
		return errors.New("pipeline not running")
	}

	p.cancel()

	// Wait for all pending batches to drain (bounded to prevent hang)
	drainDeadline := time.Now().Add(5 * time.Second)
	for p.pendingCount.Load() > 0 && time.Now().Before(drainDeadline) {
		time.Sleep(time.Millisecond)
	}

	p.wg.Wait()

	// Close results channel after all senders are done
	close(p.results)

	// Free GPU resources
	p.freeGPUResources()

	return nil
}

// freeGPUResources releases all GPU memory.
func (p *GPUBatchPipeline) freeGPUResources() {
	// Release buffer GPU memory
	for _, buf := range p.buffers {
		p.freeBufferGPU(buf)
	}
	// Release Merkle tree GPU memory
	p.freeMerkleGPU()
}

func (p *GPUBatchPipeline) freeBufferGPU(buf *GPUBuffer) {
	// GPU memory deallocation
}

func (p *GPUBatchPipeline) freeMerkleGPU() {
	// Merkle GPU memory deallocation
}

// ProcessBatch submits a batch of transactions for processing.
// Returns immediately; results delivered via Results() channel.
func (p *GPUBatchPipeline) ProcessBatch(txs []Transaction) (uint64, error) {
	if !p.running.Load() {
		return 0, ErrPipelineStopped
	}

	if len(txs) > p.config.BatchSize {
		return 0, ErrBatchTooLarge
	}

	// Apply backpressure if too many pending batches
	if int(p.pendingCount.Load()) >= p.config.MaxPendingBatches {
		p.metrics.BackpressureCount.Add(1)
		return 0, ErrBufferFull
	}

	batchID := p.nextBatchID.Add(1)
	p.pendingCount.Add(1)

	// Async load to inactive buffer while active buffer processes
	go p.loadAndProcess(batchID, txs)

	return batchID, nil
}

// loadAndProcess handles the double-buffered load/process cycle.
func (p *GPUBatchPipeline) loadAndProcess(batchID uint64, txs []Transaction) {
	defer p.pendingCount.Add(-1)

	start := time.Now()

	// 1. Get inactive buffer (nil if context cancelled)
	buf := p.getInactiveBuffer()
	if buf == nil {
		return
	}
	defer buf.inUse.Store(false)

	// 2. Load transactions to buffer (CPU-side)
	p.loadTransactions(buf, txs)

	// 3. Upload to GPU (async via load stream)
	if p.gpuReady.Load() {
		p.uploadToGPU(buf)
	}

	// 4. Wait for previous batch to complete, then swap
	p.swapBuffers()

	// 5. Process on GPU (or CPU fallback)
	gpuStart := time.Now()
	result := p.processBatch(buf, batchID, txs)
	result.GPUTime = time.Since(gpuStart)

	// 6. Update Merkle tree with valid transactions
	p.updateMerkleTree(buf, result)

	result.ProcessingTime = time.Since(start)

	// 7. Update metrics
	p.updateMetrics(result)

	// 8. Send result
	select {
	case p.results <- result:
	case <-p.ctx.Done():
	}
}

// getInactiveBuffer returns the buffer not currently being processed.
// Returns nil if the pipeline context is cancelled during the wait.
func (p *GPUBatchPipeline) getInactiveBuffer() *GPUBuffer {
	for {
		// Check context before each attempt
		select {
		case <-p.ctx.Done():
			return nil
		default:
		}

		p.bufferMu.Lock()
		// Find first available buffer
		for _, buf := range p.buffers {
			if !buf.inUse.Load() {
				buf.inUse.Store(true)
				p.bufferMu.Unlock()
				return buf
			}
		}
		p.bufferMu.Unlock()

		// All buffers busy - yield and retry without holding the lock
		time.Sleep(time.Microsecond)
	}
}

// loadTransactions copies transaction data into buffer.
func (p *GPUBatchPipeline) loadTransactions(buf *GPUBuffer, txs []Transaction) {
	buf.count = len(txs)

	for i, tx := range txs {
		buf.txHashes[i] = tx.Hash
		buf.signatures[i] = tx.Signature
		buf.publicKeys[i] = tx.PublicKey
		buf.sigTypes[i] = tx.SigType
	}
}

// uploadToGPU transfers buffer data to GPU memory.
func (p *GPUBatchPipeline) uploadToGPU(buf *GPUBuffer) {
	if !p.gpuReady.Load() {
		return
	}

	buf.uploaded.Store(true)
}

// swapBuffers switches active/inactive buffers.
func (p *GPUBatchPipeline) swapBuffers() {
	p.bufferMu.Lock()
	defer p.bufferMu.Unlock()

	p.activeBuffer = (p.activeBuffer + 1) % len(p.buffers)
	p.metrics.BufferSwaps.Add(1)
}

// processBatch executes signature verification and aggregation.
func (p *GPUBatchPipeline) processBatch(buf *GPUBuffer, batchID uint64, txs []Transaction) *BatchResult {
	result := &BatchResult{
		BatchID:         batchID,
		ProcessedCount:  buf.count,
		Errors:          make([]TransactionError, 0),
		SignatureProofs: make([]SignatureProof, 0, buf.count),
	}

	if p.gpuReady.Load() && buf.uploaded.Load() {
		// GPU path: batch verification
		p.verifySignaturesGPU(buf, result)
	} else {
		// CPU fallback: parallel verification
		p.verifySignaturesCPU(buf, txs, result)
	}

	return result
}

// verifySignaturesGPU runs batch signature verification on GPU.
func (p *GPUBatchPipeline) verifySignaturesGPU(buf *GPUBuffer, result *BatchResult) {
	// Launch GPU kernel for batch verification
	// Groups signatures by type for efficient SIMD processing

	// ECDSA batch verify
	ecdsaCount := p.verifyECDSABatchGPU(buf)

	// Ed25519 batch verify
	ed25519Count := p.verifyEd25519BatchGPU(buf)

	// BLS batch verify (aggregatable)
	blsCount := p.verifyBLSBatchGPU(buf)

	// Download results
	p.downloadResults(buf)

	// Process results
	for i := 0; i < buf.count; i++ {
		proof := SignatureProof{
			TxID:    types.ID{}, // Would be set from txHashes
			Valid:   buf.validFlags[i],
			SigType: buf.sigTypes[i],
		}
		result.SignatureProofs = append(result.SignatureProofs, proof)

		if buf.validFlags[i] {
			result.ValidCount++
		} else {
			result.InvalidCount++
		}
	}

	_ = ecdsaCount
	_ = ed25519Count
	_ = blsCount
}

// verifyECDSABatchGPU verifies all ECDSA signatures in the buffer. The
// accel package does not yet expose a batch ECDSA kernel, so each signature
// is verified on CPU and the result is written into buf.validFlags.
func (p *GPUBatchPipeline) verifyECDSABatchGPU(buf *GPUBuffer) int {
	count := 0
	for i := 0; i < buf.count; i++ {
		if buf.sigTypes[i] != SigECDSA {
			continue
		}
		buf.validFlags[i] = verifyECDSA(buf.txHashes[i][:], buf.signatures[i], buf.publicKeys[i])
		count++
	}
	return count
}

// verifyEd25519BatchGPU verifies all Ed25519 signatures in the buffer using
// the standard library's batch-friendly verifier. Results are written into
// buf.validFlags.
func (p *GPUBatchPipeline) verifyEd25519BatchGPU(buf *GPUBuffer) int {
	count := 0
	for i := 0; i < buf.count; i++ {
		if buf.sigTypes[i] != SigEd25519 {
			continue
		}
		buf.validFlags[i] = verifyEd25519(buf.txHashes[i][:], buf.signatures[i], buf.publicKeys[i])
		count++
	}
	return count
}

// verifyBLSBatchGPU performs batch BLS verification on the accel backend.
// BLS signatures are aggregatable; accel.BLSBatchVerify dispatches to CUDA
// / Metal / WebGPU / CPU depending on what the runtime detected. Below the
// BLSBatchVerifyThreshold the pipeline falls back to per-signature CPU
// verify (see verifySignaturesCPU).
func (p *GPUBatchPipeline) verifyBLSBatchGPU(buf *GPUBuffer) int {
	indices := make([]int, 0, buf.count)
	pks := make([][]byte, 0, buf.count)
	sigs := make([][]byte, 0, buf.count)
	msgs := make([][]byte, 0, buf.count)
	for i := 0; i < buf.count; i++ {
		if buf.sigTypes[i] != SigBLS {
			continue
		}
		indices = append(indices, i)
		pks = append(pks, buf.publicKeys[i])
		sigs = append(sigs, buf.signatures[i])
		hash := buf.txHashes[i]
		msgs = append(msgs, hash[:])
	}
	if len(indices) == 0 {
		return 0
	}

	results, err := accel.BLSBatchVerify(pks, sigs, msgs)
	if err != nil || len(results) != len(indices) {
		// Fall back to single-sig BLS verify on each transaction.
		for i, idx := range indices {
			buf.validFlags[idx] = verifyBLS(buf.txHashes[idx][:], sigs[i], pks[i])
		}
		return len(indices)
	}
	for i, idx := range indices {
		buf.validFlags[idx] = results[i]
	}
	return len(indices)
}

// downloadResults copies verification results from GPU to CPU.
func (p *GPUBatchPipeline) downloadResults(buf *GPUBuffer) {
	// GPU-to-CPU transfer via execute stream
	// Results land in buf.validFlags
}

// verifySignaturesCPU performs parallel CPU signature verification.
func (p *GPUBatchPipeline) verifySignaturesCPU(buf *GPUBuffer, txs []Transaction, result *BatchResult) {
	var wg sync.WaitGroup
	results := make(chan SignatureProof, buf.count)

	// Worker pool for parallel verification
	workers := p.config.VerifyWorkers
	if workers < 1 {
		workers = 1
	}

	// Distribute work
	chunkSize := (buf.count + workers - 1) / workers
	for w := 0; w < workers; w++ {
		start := w * chunkSize
		end := start + chunkSize
		if end > buf.count {
			end = buf.count
		}
		if start >= buf.count {
			break
		}

		wg.Add(1)
		go func(start, end int) {
			defer wg.Done()
			for i := start; i < end; i++ {
				verifyStart := time.Now()
				valid := p.verifySingleSignature(txs[i])
				buf.validFlags[i] = valid

				results <- SignatureProof{
					TxID:     txs[i].ID,
					Valid:    valid,
					SigType:  txs[i].SigType,
					Duration: time.Since(verifyStart),
				}
			}
		}(start, end)
	}

	// Wait and collect
	go func() {
		wg.Wait()
		close(results)
	}()

	for proof := range results {
		result.SignatureProofs = append(result.SignatureProofs, proof)
		if proof.Valid {
			result.ValidCount++
		} else {
			result.InvalidCount++
		}
	}
}

// verifySingleSignature verifies one signature on CPU.
func (p *GPUBatchPipeline) verifySingleSignature(tx Transaction) bool {
	switch tx.SigType {
	case SigECDSA:
		return verifyECDSA(tx.Hash[:], tx.Signature, tx.PublicKey)
	case SigEd25519:
		return verifyEd25519(tx.Hash[:], tx.Signature, tx.PublicKey)
	case SigBLS:
		return verifyBLS(tx.Hash[:], tx.Signature, tx.PublicKey)
	case SigMLDSA:
		return verifyMLDSA(tx.Hash[:], tx.Signature, tx.PublicKey)
	default:
		return false
	}
}

// updateMerkleTree adds valid transaction hashes to the GPU-resident tree.
func (p *GPUBatchPipeline) updateMerkleTree(buf *GPUBuffer, result *BatchResult) {
	// Collect valid transaction hashes
	validHashes := make([][32]byte, 0, result.ValidCount)
	for i := 0; i < buf.count; i++ {
		if buf.validFlags[i] {
			validHashes = append(validHashes, buf.txHashes[i])
		}
	}

	if len(validHashes) == 0 {
		return
	}

	if p.gpuReady.Load() {
		// GPU path: parallel Merkle update
		result.MerkleRoot = p.merkleTree.updateGPU(validHashes)
	} else {
		// CPU fallback
		result.MerkleRoot = p.merkleTree.updateCPU(validHashes)
	}
}

// updateGPU performs parallel Merkle tree update on GPU.
func (mt *GPUMerkleTree) updateGPU(hashes [][32]byte) [32]byte {
	mt.pendMu.Lock()
	mt.pending = append(mt.pending, hashes...)
	pending := mt.pending
	mt.pending = mt.pending[:0]
	mt.pendMu.Unlock()

	if len(pending) == 0 {
		mt.rootMu.RLock()
		root := mt.root
		mt.rootMu.RUnlock()
		return root
	}

	// Fall back to CPU until GPU kernel is wired up
	return mt.updateCPU(pending)
}

// updateCPU performs sequential Merkle tree update on CPU.
func (mt *GPUMerkleTree) updateCPU(hashes [][32]byte) [32]byte {
	if len(hashes) == 0 {
		mt.rootMu.RLock()
		root := mt.root
		mt.rootMu.RUnlock()
		return root
	}

	// Build tree bottom-up
	current := hashes
	for len(current) > 1 {
		next := make([][32]byte, (len(current)+1)/2)
		for i := 0; i < len(current); i += 2 {
			if i+1 < len(current) {
				next[i/2] = hashPair(current[i], current[i+1])
			} else {
				next[i/2] = current[i] // Odd node promoted
			}
		}
		current = next
	}

	mt.rootMu.Lock()
	mt.root = current[0]
	mt.rootMu.Unlock()

	return current[0]
}

// hashPair computes hash of two concatenated hashes.
func hashPair(a, b [32]byte) [32]byte {
	// SHA256 of concatenation
	data := make([]byte, 64)
	copy(data[:32], a[:])
	copy(data[32:], b[:])
	return sha256Hash(data)
}

// sha256Hash computes SHA256.
func sha256Hash(data []byte) [32]byte {
	return sha256.Sum256(data)
}

// updateMetrics records processing statistics.
func (p *GPUBatchPipeline) updateMetrics(result *BatchResult) {
	p.metrics.TotalBatches.Add(1)
	p.metrics.TotalTransactions.Add(uint64(result.ProcessedCount))
	p.metrics.TotalValid.Add(uint64(result.ValidCount))
	p.metrics.TotalInvalid.Add(uint64(result.InvalidCount))
	p.metrics.TotalGPUTime.Add(int64(result.GPUTime))
}

// processLoop is the main pipeline processing goroutine.
func (p *GPUBatchPipeline) processLoop() {
	defer p.wg.Done()

	ticker := time.NewTicker(time.Millisecond)
	defer ticker.Stop()

	for {
		select {
		case <-p.ctx.Done():
			return
		case <-ticker.C:
			// Periodic maintenance: flush pending Merkle updates
			if p.merkleTree != nil {
				p.merkleTree.flushPending(p.gpuReady.Load())
			}
		}
	}
}

// flushPending processes any pending Merkle tree updates.
func (mt *GPUMerkleTree) flushPending(gpuReady bool) {
	mt.pendMu.Lock()
	if len(mt.pending) == 0 {
		mt.pendMu.Unlock()
		return
	}
	pending := mt.pending
	mt.pending = make([][32]byte, 0, 1024)
	mt.pendMu.Unlock()

	if gpuReady {
		mt.updateGPU(pending)
	} else {
		mt.updateCPU(pending)
	}
}

// Results returns the channel for receiving batch results.
func (p *GPUBatchPipeline) Results() <-chan *BatchResult {
	return p.results
}

// Metrics returns current pipeline metrics.
func (p *GPUBatchPipeline) Metrics() *PipelineMetrics {
	return p.metrics
}

// GetMetricsSnapshot returns a snapshot of current metrics.
func (p *GPUBatchPipeline) GetMetricsSnapshot() (batches, txs, valid, invalid, swaps, backpressure uint64, gpuNs, cpuNs int64) {
	return p.metrics.TotalBatches.Load(),
		p.metrics.TotalTransactions.Load(),
		p.metrics.TotalValid.Load(),
		p.metrics.TotalInvalid.Load(),
		p.metrics.BufferSwaps.Load(),
		p.metrics.BackpressureCount.Load(),
		p.metrics.TotalGPUTime.Load(),
		p.metrics.TotalCPUTime.Load()
}

// MerkleRoot returns the current Merkle tree root.
func (p *GPUBatchPipeline) MerkleRoot() [32]byte {
	if p.merkleTree == nil {
		return [32]byte{}
	}
	p.merkleTree.rootMu.RLock()
	defer p.merkleTree.rootMu.RUnlock()
	return p.merkleTree.root
}

// GPUReady returns whether GPU acceleration is available.
func (p *GPUBatchPipeline) GPUReady() bool {
	return p.gpuReady.Load()
}

// Running returns whether the pipeline is running.
func (p *GPUBatchPipeline) Running() bool {
	return p.running.Load()
}

// Signature verification using real crypto packages
func verifyECDSA(hash, sig, pubkey []byte) bool {
	if len(sig) < 64 {
		return false
	}
	return secp256k1.VerifySignature(pubkey, hash, sig[:64])
}

func verifyEd25519(hash, sig, pubkey []byte) bool {
	if len(sig) != ed25519.SignatureSize || len(pubkey) != ed25519.PublicKeySize {
		return false
	}
	return ed25519.Verify(pubkey, hash, sig)
}

func verifyBLS(hash, sig, pubkey []byte) bool {
	pk, err := bls.PublicKeyFromCompressedBytes(pubkey)
	if err != nil {
		return false
	}
	signature, err := bls.SignatureFromBytes(sig)
	if err != nil {
		return false
	}
	return bls.Verify(pk, signature, hash)
}

func verifyMLDSA(hash, sig, pubkey []byte) bool {
	// Auto-detect mode from public key size (FIPS 204 sizes)
	var mode mldsa.Mode
	switch len(pubkey) {
	case 1312:
		mode = mldsa.MLDSA44
	case 1952:
		mode = mldsa.MLDSA65
	case 2592:
		mode = mldsa.MLDSA87
	default:
		return false
	}
	pk, err := mldsa.PublicKeyFromBytes(pubkey, mode)
	if err != nil {
		return false
	}
	return pk.VerifySignature(hash, sig)
}

// gpuAvailable returns true iff at least one accel backend (CUDA / Metal /
// WebGPU / CPU) reports a usable device. Wired to luxfi/accel.Available so
// the pipeline only takes the GPU path when the C library reports capacity
// to dispatch a kernel.
func gpuAvailable() bool {
	return accel.Available()
}