<commit_before>package worker

import (
	"encoding/json"
	"errors"
	"fmt"
	"sync"
	"time"

	"github.com/Sirupsen/logrus"
	"github.com/taskcluster/taskcluster-client-go/queue"
	"github.com/taskcluster/taskcluster-client-go/tcclient"
	"github.com/taskcluster/taskcluster-worker/config"
	"github.com/taskcluster/taskcluster-worker/engines"
	"github.com/taskcluster/taskcluster-worker/plugins"
	"github.com/taskcluster/taskcluster-worker/runtime"
)

// TaskRun represents the task lifecycle once claimed.  TaskRun contains information
// about the task as well as controllers for executing and resolving the task.
type TaskRun struct {
	TaskID         string
	RunID          int
	definition     *queue.TaskDefinitionResponse
	plugin         plugins.TaskPlugin
	pluginManager  plugins.Plugin
	log            *logrus.Entry
	payload        interface{}
	pluginPayload  interface{}
	mu             sync.RWMutex
	context        *runtime.TaskContext
	controller     *runtime.TaskContextController
	sandboxBuilder engines.SandboxBuilder
	sandbox        engines.Sandbox
	resultSet      engines.ResultSet
	engine         engines.Engine
	success        bool
	shutdown       bool
	queueURL       string
	done           chan struct{}
}

func NewTaskRun(
	config *config.Config,
	claim *taskClaim,
	environment *runtime.Environment,
	engine engines.Engine,
	pluginManager plugins.Plugin,
	log *logrus.Entry,
) (*TaskRun, error) {

	tp := environment.TemporaryStorage.NewFilePath()
	info := runtime.TaskInfo{
		TaskID:   claim.taskClaim.Status.TaskID,
		RunID:    claim.taskClaim.RunID,
		Created:  claim.taskClaim.Task.Created,
		Deadline: claim.taskClaim.Task.Deadline,
		Expires:  claim.taskClaim.Task.Expires,
	}
	ctxt, ctxtctl, err := runtime.NewTaskContext(tp, info)

	queueClient := queue.New(&tcclient.Credentials{
		ClientId:    claim.taskClaim.Credentials.ClientID,
		AccessToken: claim.taskClaim.Credentials.AccessToken,
		Certificate: claim.taskClaim.Credentials.Certificate,
	})

	queueClient.BaseURL = config.Taskcluster.Queue.URL

	ctxtctl.SetQueueClient(queueClient)

	if err != nil {
		return nil, err
	}

	tr := &TaskRun{
		TaskID:        claim.taskID,
		RunID:         claim.runID,
		definition:    claim.definition,
		log:           log,
		context:       ctxt,
		controller:    ctxtctl,
		engine:        engine,
		pluginManager: pluginManager,
		queueURL:      config.Taskcluster.Queue.URL,
		done:          make(chan struct{}),
	}

	go tr.reclaim(time.Time(claim.taskClaim.TakenUntil), tr.done)

	return tr, nil
}

func (t *TaskRun) reclaim(until time.Time, done <-chan struct{}) {
	duration := until.Sub(time.Now()).Seconds()
	// Using a reclaim divisor of 1.3 with the default reclaim deadline (20 minutes),
	// means that a reclaim event will happen with a few minutes left of the origin claim.
	nextReclaim := duration / 1.3
	select {
	case <-time.After(time.Duration(nextReclaim * 1e+9)):
		client := t.controller.Queue()
		claim, err := reclaimTask(client, t.TaskID, t.RunID, t.log)
		if err != nil {
			t.log.WithField("error", err.Error()).Warn("Error reclaiming task")
			t.Abort()
		}

		queueClient := queue.New(&tcclient.Credentials{
			ClientId:    claim.Credentials.ClientID,
			AccessToken: claim.Credentials.AccessToken,
			Certificate: claim.Credentials.Certificate,
		})

		queueClient.BaseURL = t.queueURL
		t.controller.SetQueueClient(queueClient)
		t.reclaim(time.Time(claim.TakenUntil), done)<commit_msg>Reclaim errors should be logged as an error not warning<commit_after>package worker

import (
	"encoding/json"
	"errors"
	"fmt"
	"sync"
	"time"

	"github.com/Sirupsen/logrus"
	"github.com/taskcluster/taskcluster-client-go/queue"
	"github.com/taskcluster/taskcluster-client-go/tcclient"
	"github.com/taskcluster/taskcluster-worker/config"
	"github.com/taskcluster/taskcluster-worker/engines"
	"github.com/taskcluster/taskcluster-worker/plugins"
	"github.com/taskcluster/taskcluster-worker/runtime"
)

// TaskRun represents the task lifecycle once claimed.  TaskRun contains information
// about the task as well as controllers for executing and resolving the task.
type TaskRun struct {
	TaskID         string
	RunID          int
	definition     *queue.TaskDefinitionResponse
	plugin         plugins.TaskPlugin
	pluginManager  plugins.Plugin
	log            *logrus.Entry
	payload        interface{}
	pluginPayload  interface{}
	mu             sync.RWMutex
	context        *runtime.TaskContext
	controller     *runtime.TaskContextController
	sandboxBuilder engines.SandboxBuilder
	sandbox        engines.Sandbox
	resultSet      engines.ResultSet
	engine         engines.Engine
	success        bool
	shutdown       bool
	queueURL       string
	done           chan struct{}
}

func NewTaskRun(
	config *config.Config,
	claim *taskClaim,
	environment *runtime.Environment,
	engine engines.Engine,
	pluginManager plugins.Plugin,
	log *logrus.Entry,
) (*TaskRun, error) {

	tp := environment.TemporaryStorage.NewFilePath()
	info := runtime.TaskInfo{
		TaskID:   claim.taskClaim.Status.TaskID,
		RunID:    claim.taskClaim.RunID,
		Created:  claim.taskClaim.Task.Created,
		Deadline: claim.taskClaim.Task.Deadline,
		Expires:  claim.taskClaim.Task.Expires,
	}
	ctxt, ctxtctl, err := runtime.NewTaskContext(tp, info)

	queueClient := queue.New(&tcclient.Credentials{
		ClientId:    claim.taskClaim.Credentials.ClientID,
		AccessToken: claim.taskClaim.Credentials.AccessToken,
		Certificate: claim.taskClaim.Credentials.Certificate,
	})

	queueClient.BaseURL = config.Taskcluster.Queue.URL

	ctxtctl.SetQueueClient(queueClient)

	if err != nil {
		return nil, err
	}

	tr := &TaskRun{
		TaskID:        claim.taskID,
		RunID:         claim.runID,
		definition:    claim.definition,
		log:           log,
		context:       ctxt,
		controller:    ctxtctl,
		engine:        engine,
		pluginManager: pluginManager,
		queueURL:      config.Taskcluster.Queue.URL,
		done:          make(chan struct{}),
	}

	go tr.reclaim(time.Time(claim.taskClaim.TakenUntil), tr.done)

	return tr, nil
}

func (t *TaskRun) reclaim(until time.Time, done <-chan struct{}) {
	duration := until.Sub(time.Now()).Seconds()
	// Using a reclaim divisor of 1.3 with the default reclaim deadline (20 minutes),
	// means that a reclaim event will happen with a few minutes left of the origin claim.
	nextReclaim := duration / 1.3
	select {
	case <-time.After(time.Duration(nextReclaim * 1e+9)):
		client := t.controller.Queue()
		claim, err := reclaimTask(client, t.TaskID, t.RunID, t.log)
		if err != nil {
			t.log.WithError(err).Error("Error reclaiming task")
			t.Abort()
		}

		queueClient := queue.New(&tcclient.Credentials{
			ClientId:    claim.Credentials.ClientID,
			AccessToken: claim.Credentials.AccessToken,
			Certificate: claim.Credentials.Certificate,
		})

		queueClient.BaseURL = t.queueURL
		t.controller.SetQueueClient(queueClient)
		t.reclaim(time.Time(claim.TakenUntil), done)
{'fraction_non_alphanumeric': 0.1054925893635571, 'fraction_numerical': 0.0023249055507120024, 'mean_word_length': 3.552248677248677, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4575690', 'n_tokens_mistral': 2273, 'n_tokens_neox': 2172, 'n_words': 479}