implement function timeout

dot/parachain/candidate-validation/worker.go

@@ -1,11 +1,12 @@
 package candidatevalidation
 
 import (
+	"time"
+
 	parachainruntime "github.com/ChainSafe/gossamer/dot/parachain/runtime"
 	parachaintypes "github.com/ChainSafe/gossamer/dot/parachain/types"
 )
 
-// TODO(ed): figure out a better name for this that describes what it does
 type worker struct {
 	workerID    parachaintypes.ValidationCodeHash
 	instance    *parachainruntime.Instance
@@ -16,6 +17,7 @@ type workerTask struct {
 	work             parachainruntime.ValidationParameters
 	maxPoVSize       uint32
 	candidateReceipt *parachaintypes.CandidateReceipt
+	timeoutKind      parachaintypes.PvfExecTimeoutKind
 }
 
 func newWorker(validationCode parachaintypes.ValidationCode) (*worker, error) {
@@ -31,6 +33,26 @@ func newWorker(validationCode parachaintypes.ValidationCode) (*worker, error) {
 	}, nil
 }
 
+type resultWithError struct {
+	result *parachainruntime.ValidationResult
+	err    error
+}
+
+func determineTimeout(timeoutKind parachaintypes.PvfExecTimeoutKind) time.Duration {
+	value, err := timeoutKind.Value()
+	if err != nil {
+		return 2 * time.Second
+	}
+	switch value.(type) {
+	case parachaintypes.Backing:
+		return 2 * time.Second
+	case parachaintypes.Approval:
+		return 12 * time.Second
+	default:
+		return 2 * time.Second
+	}
+}
+
 func (w *worker) executeRequest(task *workerTask) (*ValidationResult, error) {
 	logger.Debugf("[EXECUTING] worker %x task %v", w.workerID, task.work)
 	candidateHash, err := parachaintypes.GetCandidateHash(task.candidateReceipt)
@@ -43,10 +65,33 @@ func (w *worker) executeRequest(task *workerTask) (*ValidationResult, error) {
 		logger.Debugf("candidate %x already processed", candidateHash)
 		return processed, nil
 	}
-	validationResult, err := w.instance.ValidateBlock(task.work)
-	if err != nil {
-		logger.Errorf("executing validate_block: %w", err)
-		reasonForInvalidity := ExecutionError
+
+	var validationResult *parachainruntime.ValidationResult
+	validationResultCh := make(chan (*resultWithError))
+	timeoutDuration := determineTimeout(task.timeoutKind)
+
+	go func() {
+		result, err := w.instance.ValidateBlock(task.work)
+		if err != nil {
+			validationResultCh <- &resultWithError{result: nil, err: err}
+		}
+		validationResultCh <- &resultWithError{
+			result: result,
+		}
+	}()
+
+	select {
+	case validationResultWErr := <-validationResultCh:
+		if validationResultWErr.err != nil {
+			logger.Errorf("executing validate_block: %w", err)
+			reasonForInvalidity := ExecutionError
+			return &ValidationResult{InvalidResult: &reasonForInvalidity}, nil //nolint
+		}
+		validationResult = validationResultWErr.result
+
+	case <-time.After(timeoutDuration):
+		logger.Errorf("validation timed out")
+		reasonForInvalidity := Timeout
 		return &ValidationResult{InvalidResult: &reasonForInvalidity}, nil
 	}
 

dot/parachain/candidate-validation/worker_pool.go

@@ -8,8 +8,6 @@ import (
 )
 
 type workerPool struct {
-
-	// todo, make sure other functions work with paraID
 	workers map[parachaintypes.ValidationCodeHash]*worker
 }
 
@@ -150,10 +148,12 @@ func (v *workerPool) submitRequest(msg *ValidationTask) (*ValidationResult, erro
 		RelayParentNumber:      msg.PersistedValidationData.RelayParentNumber,
 		RelayParentStorageRoot: msg.PersistedValidationData.RelayParentStorageRoot,
 	}
+
 	workTask := &workerTask{
 		work:             validationParams,
 		maxPoVSize:       msg.PersistedValidationData.MaxPovSize,
 		candidateReceipt: msg.CandidateReceipt,
+		timeoutKind:      msg.PvfExecTimeoutKind,
 	}
 	return syncWorker.executeRequest(workTask)