func (e *Executor) Invalidate(task *Task, event InvalidationEvent) {
	execution := e.tasks[task]
	if didInvalidate := execution.Invalidate(event); !didInvalidate {
		return
	}
	e.invalidationCh <- struct{}{}
}
Invalidate marks a task and its dependencies as invalidated. If any
tasks become invalidated from this call, Invalidate() will also
schedule a re-execution of the DAG.
package taskrunner
import (
	"context"
	"errors"
	"sync"
	"time"
	"github.com/samsarahq/go/oops"
	"github.com/samsarahq/taskrunner/config"
	"github.com/samsarahq/taskrunner/shell"
	"github.com/samsarahq/taskrunner/watcher"
	"go.uber.org/multierr"
	"golang.org/x/sync/errgroup"
	"mvdan.cc/sh/interp"
)
type Executor struct {
	ctx    context.Context
	config *config.Config
	tasks taskSet
	watchMode bool
	mu sync.Mutex
	wg errgroup.Group
	invalidationCh chan struct{}
	taskRegistry map[string]*Task
	shellRunOptions []shell.RunOption
	eventsChs []chan ExecutorEvent
	watcherEnhancers []WatcherEnhancer
}
type WatcherEnhancer func(watcher.Watcher) watcher.Watcher
var errUndefinedTaskName = errors.New("undefined task name")
type ExecutorOption func(*Executor)
func WithWatcherEnhancer(we WatcherEnhancer) ExecutorOption {
	return func(e *Executor) {
		e.watcherEnhancers = append(e.watcherEnhancers, we)
	}
}
func WithWatchMode(watchMode bool) ExecutorOption {
	return func(e *Executor) {
		e.watchMode = watchMode
	}
}
func ShellRunOptions(opts ...shell.RunOption) ExecutorOption {
	return func(e *Executor) {
		e.shellRunOptions = append(e.shellRunOptions, opts...)
	}
}
func NewExecutor(config *config.Config, tasks []*Task, opts ...ExecutorOption) *Executor {
	executor := &Executor{
		config:         config,
		invalidationCh: make(chan struct{}, 1),
		taskRegistry:   make(map[string]*Task),
	}
	for _, opt := range opts {
		opt(executor)
	}
	for _, task := range tasks {
		executor.taskRegistry[task.Name] = task
	}
	return executor
}
func (e *Executor) Config() *config.Config { return e.config }
func (e *Executor) Subscribe() (events <-chan ExecutorEvent) {
	ch := make(chan ExecutorEvent, 1024)
	e.eventsChs = append(e.eventsChs, ch)
	return ch
}
func (e *Executor) publishEvent(event ExecutorEvent) {
	for _, ch := range e.eventsChs {
		ch <- event
	}
}
func (e *Executor) runInvalidationLoop() {
	timer := time.NewTimer(time.Second)
	timer.Stop()
	go func() {
		for {
			select {
			case <-e.invalidationCh:
				timer.Reset(time.Second)
			case <-e.ctx.Done():
				return
			case <-timer.C:
				e.evaluateInvalidationPlan()
				go e.runPass()
			}
		}
	}()
}
func (e *Executor) evaluateInvalidationPlan() {
	time.Sleep(time.Millisecond * 1000)
	e.mu.Lock()
	defer e.mu.Unlock()
	var toInvalidate []*taskExecution
	for _, execution := range e.tasks {
		if len(execution.pendingInvalidations) == 0 || execution.state == taskExecutionState_invalid {
			continue
		}
		var reasons []InvalidationEvent
		for reason := range execution.pendingInvalidations {
			reasons = append(reasons, reason)
		}
		e.publishEvent(&TaskInvalidatedEvent{
			simpleEvent: execution.simpleEvent(),
			Reasons:     reasons,
		})
		toInvalidate = append(toInvalidate, execution)
	}
	for _, execution := range toInvalidate {
		execution.invalidate(e.ctx)
	}
}