engine_validate.go

package dasmon

import (
	"bytes"
	"fmt"

	"github.com/OffchainLabs/prysm/v7/consensus-types/primitives"
	eth "github.com/OffchainLabs/prysm/v7/proto/prysm/v1alpha1"
	"github.com/OffchainLabs/prysm/v7/time/slots"
	"github.com/ethp2p/dasmon/metrics"
	"github.com/ethp2p/dasmon/store"
)

// processCompletedJob processes a completed job and emits sampling events
func (m *Engine) processCompletedJob(job *Job) {
	// Record job duration
	metrics.SamplingDuration.Observe(job.Result.Duration.Seconds())

	// Prepare batch results for database
	dbResults := make([]*store.SamplingResult, 0)

	// If the job had an error, emit failure events for each slot/column.
	if job.Result.Error != nil {
		job.log.Warn("job completed with error", "error", job.Result.Error)

		// Extract error kind for metrics
		errKind := ErrUnknown
		if de, ok := job.Result.Error.(DasmonError); ok {
			errKind = de.Kind
		}

		metrics.SamplingErrorsTotal.WithLabelValues(errKind.String()).Inc()

		// Convert to DasmonError for event emission
		var dasmonErr *DasmonError
		if de, ok := job.Result.Error.(DasmonError); ok {
			dasmonErr = &de
		} else {
			// Wrap unknown error types
			err := DasmonError{
				Kind:    ErrUnknown,
				Details: job.Result.Error.Error(),
				Inner:   job.Result.Error,
			}
			dasmonErr = &err
		}

		for slot := job.StartSlot; slot <= job.EndSlot; slot++ {
			// Find the block hash for this slot from commitments
			var blockHash string
			if blockInfo, ok := m.commTracker.Get(uint64(slot)); ok {
				blockHash = fmt.Sprintf("0x%x", blockInfo.BlockRoot)
			}

			for _, colIdx := range job.Columns {
				metrics.SamplingResultsTotal.WithLabelValues("failure").Inc()

				samplingEvt := &EvtSampling{
					PeerID:            &job.PeerID,
					Epoch:             uint64(slots.ToEpoch(primitives.Slot(slot))),
					Slot:              slot,
					BlockHash:         blockHash,
					Column:            colIdx,
					Result:            SamplingResultFailure,
					Error:             dasmonErr,
					Duration:          job.Result.Duration,
					JobStartTimestamp: job.StartedAt,
				}
				m.emitEvent(Event{Sampling: samplingEvt})

				// Add to database batch
				dbResults = append(dbResults, &store.SamplingResult{
					PeerID:   job.PeerID,
					Slot:     slot,
					Column:   colIdx,
					Result:   "failure",
					TestedAt: job.StartedAt,
				})
			}
		}

		// Persist all results to database
		if err := m.store.RecordSamplingResults(m.ctx, dbResults); err != nil {
			log.Error("failed to persist sampling results to database", "error", err)
		}

		return
	}

	// Create the sidecars map
	// slot -> column index -> sidecar
	indexedSidecars := make(map[uint64]map[uint64]*eth.DataColumnSidecar)
	for _, sc := range job.Result.Sidecars {
		slot := uint64(sc.SignedBlockHeader.Header.Slot)
		if _, exists := indexedSidecars[slot]; !exists {
			indexedSidecars[slot] = make(map[uint64]*eth.DataColumnSidecar)
		}
		indexedSidecars[slot][sc.Index] = sc
	}

	// Emit sampling event for each slot in the job
	for slot := job.StartSlot; slot <= job.EndSlot; slot++ {
		// Emit event for each column tested
		for _, colIdx := range job.Columns {
			samplingEvt := &EvtSampling{
				PeerID:            &job.PeerID,
				Epoch:             uint64(slots.ToEpoch(primitives.Slot(slot))),
				Slot:              slot,
				Column:            colIdx,
				Duration:          job.Result.Duration,
				JobStartTimestamp: job.StartedAt,
			}

			var resultStr string
			if sidecar, ok := indexedSidecars[uint64(slot)][colIdx]; ok {
				samplingEvt.Result = SamplingResultSuccess
				samplingEvt.BlockHash = fmt.Sprintf("0x%x", sidecar.SignedBlockHeader.Header.BodyRoot)
				samplingEvt.ColumnSize = len(sidecar.Column)
				resultStr = "success"

				metrics.SamplingResultsTotal.WithLabelValues("success").Inc()
			} else {
				samplingEvt.Result = SamplingResultMissing
				resultStr = "missing"

				metrics.SamplingResultsTotal.WithLabelValues("missing").Inc()
			}

			m.emitEvent(Event{Sampling: samplingEvt})

			// Add to database batch
			dbResults = append(dbResults, &store.SamplingResult{
				PeerID:   job.PeerID,
				Slot:     slot,
				Column:   colIdx,
				Result:   resultStr,
				TestedAt: job.StartedAt,
			})
		}
	}

	// Persist all results to database
	if err := m.store.RecordSamplingResults(m.ctx, dbResults); err != nil {
		log.Error("failed to persist sampling results to database", "error", err)
	}

	// Update peer's in-memory history
	m.peersMu.RLock()
	peer, exists := m.peers[job.PeerID]
	m.peersMu.RUnlock()

	if exists && peer.history != nil {
		for _, result := range dbResults {
			peer.history.RecordTest(result.Slot, result.Column, result.Result, result.TestedAt)
		}
	}
}

type Key struct {
	Slot   uint64
	Column uint64
}

func (k Key) WithExtra(format string, a ...any) Detail {
	return Detail{
		Key:   k,
		Extra: fmt.Sprintf(format, a...),
	}
}

func (k Key) WithError(err error) Detail {
	return Detail{
		Key:   k,
		Extra: err.Error(),
	}
}

func (k Key) NoExtra() Detail {
	return Detail{Key: k}
}

type Detail struct {
	Key
	Extra string
}

type ValidationErrors map[ErrKind][]Detail

// validateResponse validates the DataColumnSidecar response according to the consensus specs.
func (m *Engine) validateResponse(job *Job) error {
	ret := make(ValidationErrors)

	var (
		reqRange  = RangeLength(uint64(job.Request.StartSlot), uint64(job.Request.Count))
		seenTimes = make(map[Key]int)
	)

	// Initialize the sidecars map with the expected keys
	for slot := range reqRange.Values() {
		for _, column := range job.Request.Columns {
			k := Key{Slot: slot, Column: column}
			seenTimes[k] = 0
		}
	}

	// Record missing slots.
	for _, scar := range job.Result.Sidecars {
		// TODO verify columns in order?
		// TODO check max columns respected.

		var (
			slot = uint64(scar.SignedBlockHeader.Header.Slot)
			key  = Key{Slot: slot, Column: scar.Index}
		)

		if _, exists := seenTimes[key]; exists {
			seenTimes[key]++
		} else {
			ret[ErrUnexpectedSidecar] = append(ret[ErrUnexpectedSidecar], key.NoExtra())
			continue
		}

		if err := verifyDataColumn(scar); err != nil {
			ret[ErrVerificationFailed] = append(ret[ErrVerificationFailed], key.WithError(err))
			continue
		}

		blkInfo, ok := m.commTracker.Get(slot)
		if !ok {
			ret[ErrNoCommitmentInfo] = append(ret[ErrNoCommitmentInfo], key.NoExtra())
			continue
		}

		var (
			nComms  = len(scar.KzgCommitments)
			nCells  = len(scar.Column)
			nProofs = len(scar.KzgProofs)
		)

		switch {
		case nComms != len(blkInfo.BlobCommitments):
			d := key.WithExtra("%d != %d (got, expected)", nComms, len(blkInfo.BlobCommitments))
			ret[ErrCommitmentCountMismatch] = append(ret[ErrCommitmentCountMismatch], d)
		case nCells != nComms:
			d := key.WithExtra("%d != %d (got, expected)", nCells, nComms)
			ret[ErrCellCountMismatch] = append(ret[ErrCellCountMismatch], d)
		case nProofs != nComms:
			d := key.WithExtra("%d != %d (got, expected)", nProofs, nComms)
			ret[ErrProofCountMismatch] = append(ret[ErrProofCountMismatch], d)
		}

		if err := verifyInclusionProof(scar); err != nil {
			ret[ErrInclusionProofInvalid] = append(ret[ErrInclusionProofInvalid], key.WithError(err))
		}

		for i := range scar.KzgCommitments {
			left, right := scar.KzgCommitments[i], blkInfo.BlobCommitments[i]
			if !bytes.Equal(left, right) {
				ret[ErrCommitmentMismatch] = append(ret[ErrCommitmentMismatch], key.WithExtra("%s != %s", left, right))
			}
		}
	}

	for k, count := range seenTimes {
		switch {
		case count == 0:
			ret[ErrMissingSidecar] = append(ret[ErrMissingSidecar], k.NoExtra())
		case count > 1:
			ret[ErrDuplicateSidecar] = append(ret[ErrDuplicateSidecar], k.NoExtra())
		}
	}

	return nil
}