job.go

package dasmon

import (
	"context"
	"fmt"
	"log/slog"
	"sync/atomic"
	"time"

	"github.com/OffchainLabs/prysm/v7/consensus-types/primitives"
	eth "github.com/OffchainLabs/prysm/v7/proto/prysm/v1alpha1"
	"github.com/libp2p/go-libp2p/core/peer"
)

var (
	startTimestamp = time.Now().UnixNano()
	counter        atomic.Uint64
)

// Status represents the status of a job
type Status string

const (
	StatusPending   Status = "pending"
	StatusRunning   Status = "running"
	StatusSuccess   Status = "success"
	StatusFailed    Status = "failed"
	StatusCancelled Status = "cancelled"
)

// JobResult holds the outcome of a job execution
type JobResult struct {
	Sidecars []*eth.DataColumnSidecar
	Duration time.Duration
	Error    error
}

// Job represents a data column fetch request to a peer.
// It starts with just the request parameters (PeerID, slots, columns),
// and tracking fields are populated when Prepare() is called before execution.
type Job struct {
	// Request parameters (set when job is created)
	PeerID    peer.ID
	StartSlot uint64
	EndSlot   uint64
	Columns   []uint64

	// Tracking fields (set when Prepare is called)
	ID        string
	Status    Status
	CreatedAt time.Time
	StartedAt time.Time
	EndedAt   time.Time

	// Execution state
	Context context.Context
	Cancel  context.CancelFunc
	Request *eth.DataColumnSidecarsByRangeRequest
	Result  *JobResult

	log *slog.Logger
}

// Prepare initializes tracking fields before dispatching to a worker.
// Must be called before Execute.
func (j *Job) Prepare(ctx context.Context) {
	j.ID = fmt.Sprintf("%s-%d-%d", j.PeerID.String(), startTimestamp, counter.Add(1))
	j.Status = StatusPending
	j.CreatedAt = time.Now()
	j.Context, j.Cancel = context.WithCancel(ctx)
	j.Result = new(JobResult)
	j.log = log.With("job_id", j.ID, "peer", j.PeerID.String())
}

// Execute performs a DataColumnSidecarsByRange request to the peer and validates the response.
// It validates the response according to the Ethereum consensus specification and logs any violations.
// It also verifies that the returned columns match the expected blob KZG commitments.
func (j *Job) Execute(client *RpcClient) error {
	startTime := time.Now()
	defer func() {
		j.EndedAt = time.Now()
	}()

	// Build the request
	if j.EndSlot < j.StartSlot {
		return fmt.Errorf("invalid slot range: [%d, %d]", j.StartSlot, j.EndSlot)
	}

	count := j.EndSlot - j.StartSlot + 1
	j.Request = &eth.DataColumnSidecarsByRangeRequest{
		StartSlot: primitives.Slot(j.StartSlot),
		Count:     count,
		Columns:   j.Columns,
	}

	j.log.Debug("starting DataColumnSidecarsByRange request")
	sidecars, err := client.DataColumnByRange(j.Context, j.PeerID, j.Request)
	j.Result.Duration = time.Since(startTime)

	if err != nil {
		j.Result.Error = WrapError(ErrRPCFailure, err, "start: %d, end: %d, cols: %v", j.StartSlot, j.EndSlot, j.Columns)
		return nil
	}

	j.log.Debug("completed DataColumnSidecarsByRange request",
		"duration", j.Result.Duration,
		"chunks_rcvd", len(sidecars),
	)

	// Store result for validation by engine
	j.Result.Sidecars = sidecars

	j.log.Info("successfully fetched data column sidecars",
		"sidecars", len(sidecars),
	)

	return nil
}