WIP

bootstrap/bootstrap.go

@@ -271,9 +271,17 @@ func Boot(ctx context.Context, opts BootOpts) (Bootstrap, error) {
 	return bs, nil
 }
 
-func BootScheduler(ctx context.Context, bs Bootstrap) (*scheduler.Scheduler, error) {
+type SchedulerOpts struct {
+	Approver scheduler.Approver
+}
+
+func BootScheduler(ctx context.Context, bs Bootstrap, opts SchedulerOpts) (*scheduler.Scheduler, error) {
 	fins := &finalizers.Finalizers{}
 
+	if opts.Approver == nil {
+		opts.Approver = scheduler.StaticApprover{Value: false}
+	}
+
 	localCache, err := lcache.NewState(bs.Root, bs.Pool, bs.Graph.Targets(), bs.Observability, fins, bs.Config.Engine.GC, bs.Config.Engine.ParallelCaching)
 	if err != nil {
 		return nil, err
@@ -321,6 +329,7 @@ func BootScheduler(ctx context.Context, bs Bootstrap) (*scheduler.Scheduler, err
 		BackgroundTracker: worker2.NewRunningTracker(),
 		Finalizers:        fins,
 		Runner:            runner,
+		Approver:          opts.Approver,
 	})
 
 	if bs.Config.Engine.GitCacheHints {
@@ -346,16 +355,16 @@ type SchedulerBootstrap struct {
 	Scheduler *scheduler.Scheduler
 }
 
-func BootWithScheduler(ctx context.Context, opts BootOpts) (SchedulerBootstrap, error) {
+func BootWithScheduler(ctx context.Context, bootOpts BootOpts, schedOpts SchedulerOpts) (SchedulerBootstrap, error) {
 	ebs := SchedulerBootstrap{}
 
-	bs, err := Boot(ctx, opts)
+	bs, err := Boot(ctx, bootOpts)
 	if err != nil {
 		return ebs, err
 	}
 	ebs.Bootstrap = bs
 
-	e, err := BootScheduler(ctx, bs)
+	e, err := BootScheduler(ctx, bs, schedOpts)
 	if err != nil {
 		return ebs, err
 	}

bootstrap/rrs.go

@@ -125,7 +125,7 @@ func RunGen(ctx context.Context, e *scheduler.Scheduler, plain bool, filterFacto
 			return err
 		}
 
-		err = poolwait.Wait(ctx, fmt.Sprintf("Gen run %v", i), e.Pool, deps, plain, e.Config.ProgressInterval)
+		err = poolwait.Wait(ctx, fmt.Sprintf("Gen run %v", i), e.Pool, deps, plain, e.Config.ProgressInterval, e.Approver)
 		if err != nil {
 			return err
 		}

bootstrap/run.go

@@ -78,7 +78,7 @@ func RunMode(ctx context.Context, e *scheduler.Scheduler, rrs targetrun.Requests
 	tdeps := tdepsMap.All()
 	tdeps.AddDep(tracker.Group())
 
-	err = poolwait.Wait(ctx, "Run", e.Pool, tdeps, runopts.Plain, e.Config.ProgressInterval)
+	err = poolwait.Wait(ctx, "Run", e.Pool, tdeps, runopts.Plain, e.Config.ProgressInterval, e.Approver)
 	if err != nil {
 		return err
 	}

bootstrapwatch/bootstrap.go

@@ -324,7 +324,7 @@ func (s *State) trigger(ctx context.Context, events []fsEvent) error {
 
 	usingBs := false
 	if bs.Scheduler == nil {
-		cbs, err := bootstrap.BootWithScheduler(ctx, s.bootopts)
+		cbs, err := bootstrap.BootWithScheduler(ctx, s.bootopts, bootstrap.SchedulerOpts{})
 		if err != nil {
 			return fmt.Errorf("boot: %w", err)
 		}
@@ -379,7 +379,7 @@ func (s *State) trigger(ctx context.Context, events []fsEvent) error {
 	runDeps.AddDep(tdepsMap.All())
 	runDeps.AddDep(tracker.Group())
 
-	err = poolwait.Wait(ctx, "Change", bs.Pool, runDeps, s.runopts.Plain, bs.Config.ProgressInterval)
+	err = poolwait.Wait(ctx, "Change", bs.Pool, runDeps, s.runopts.Plain, bs.Config.ProgressInterval, bs.Scheduler.Approver)
 	if err != nil {
 		return err
 	}

cmd/heph/init.go

@@ -8,6 +8,7 @@ import (
 	"github.com/hephbuild/heph/scheduler"
 	"github.com/hephbuild/heph/specs"
 	"github.com/hephbuild/heph/utils/tuistatus"
+	"github.com/hephbuild/heph/worker2/poolui"
 	"os"
 	"strings"
 	"time"
@@ -51,7 +52,14 @@ func schedulerInit(ctx context.Context, postBoot func(bootstrap.BaseBootstrap) e
 	// This allows to block reading stdin right after a potential CheckAndUpgrade has run
 	opts.PostBootBase = postBoot
 
-	bs, err := bootstrap.BootWithScheduler(ctx, opts)
+	schedOpts := bootstrap.SchedulerOpts{
+		Approver: poolui.NewApprover(),
+	}
+	if autoApprove {
+		schedOpts.Approver = scheduler.StaticApprover{Value: true}
+	}
+
+	bs, err := bootstrap.BootWithScheduler(ctx, opts, schedOpts)
 	if err != nil {
 		return bs, err
 	}

cmd/heph/query.go

@@ -809,7 +809,7 @@ var hashinCmd = &cobra.Command{
 			return err
 		}
 
-		err = poolwait.Wait(ctx, "Run", bs.Scheduler.Pool, tdeps.All(), *plain, bs.Config.ProgressInterval)
+		err = poolwait.Wait(ctx, "Run", bs.Scheduler.Pool, tdeps.All(), *plain, bs.Config.ProgressInterval, bs.Scheduler.Approver)
 		if err != nil {
 			return err
 		}

cmd/heph/root.go

@@ -43,6 +43,7 @@ var summary *bool
 var summaryGen *bool
 var jaegerEndpoint *string
 var check *bool
+var autoApprove bool
 
 func getRunOpts() bootstrap.RunOpts {
 	return bootstrap.RunOpts{
@@ -80,6 +81,7 @@ func init() {
 	alwaysOut = runCmd.Flags().Bool("always-out", false, "Ensure output will be present in cache")
 	runCmd.Flags().BoolVarP(&all, "all", "a", false, "Force run all")
 	runCmd.Flags().MarkHidden("all")
+	runCmd.Flags().BoolVar(&autoApprove, "auto-approve", false, "Auto-approve dangerous targets")
 
 	check = fmtCmd.Flags().Bool("check", false, "Only check formatting")
 

hbuiltin/predeclared.go

@@ -184,6 +184,7 @@ func target(thread *starlark.Thread, fn *starlark.Builtin, args starlark.Tuple,
 		"gen_deps_meta?", &sargs.GenDepsMeta,
 		"annotations?", &sargs.Annotations,
 		"requests?", &sargs.Requests,
+		"dangerous?", &sargs.Dangerous,
 	); err != nil {
 		if sargs.Name != "" {
 			return nil, fmt.Errorf("%v: %w", pkg.TargetAddr(sargs.Name), err)

hbuiltin/predeclared_types.go

@@ -42,6 +42,7 @@ type TargetArgs struct {
 	GenDepsMeta         bool
 	Annotations         xstarlark.Distruct
 	Requests            xstarlark.Distruct
+	Dangerous           bool
 }
 
 type TargetArgsTransitive struct {

hbuiltin/target_spec_parser.go

@@ -57,6 +57,7 @@ func specFromArgs(args TargetArgs, pkg *packages.Package) (specs.Target, error)
 		OutEnv:      args.OutEnv,
 		HashFile:    args.HashFile,
 		GenDepsMeta: args.GenDepsMeta,
+		Dangerous:   args.Dangerous,
 	}
 
 	var err error

scheduler/engine.go

@@ -22,6 +22,18 @@ import (
 	"sync"
 )
 
+type Approver interface {
+	Approve(ctx context.Context, t specs.Target) bool
+}
+
+type StaticApprover struct {
+	Value bool
+}
+
+func (a StaticApprover) Approve(ctx context.Context, t specs.Target) bool {
+	return a.Value
+}
+
 type Scheduler struct {
 	Cwd               string
 	Root              *hroot.State
@@ -39,6 +51,7 @@ type Scheduler struct {
 	Finalizers        *finalizers.Finalizers
 	Runner            *targetrun.Runner
 	GitStatus         *gitstatus.GitStatus
+	Approver          Approver
 
 	toolsLock locks.Locker
 }

scheduler/target_run.go

@@ -24,6 +24,13 @@ func (e *Scheduler) Run(ctx context.Context, rr targetrun.Request, iocfg sandbox
 
 	target := rr.Target
 
+	if target.Dangerous {
+		approved := e.Approver.Approve(ctx, target.Spec())
+		if !approved {
+			return fmt.Errorf("dangerous target, pass --auto-approve to run")
+		}
+	}
+
 	done := log.TraceTiming("run " + target.Addr)
 	defer done()
 

specs/target_spec.go

@@ -136,6 +136,7 @@ type Target struct {
 	GenDepsMeta         bool
 	Annotations         map[string]interface{}
 	Requests            map[string]float64
+	Dangerous           bool
 }
 
 func (t Target) MarshalJSON() ([]byte, error) {

utils/locks/flock.go

@@ -150,10 +150,13 @@ func (l *Flock) lock(ctx context.Context, ro bool) error {
 			lockCh <- flock.Flock(f, ro, true)
 		}()
 
-		err := worker2.WaitChanE(ctx, lockCh)
+		lockErr, err := worker2.WaitChanReceive(ctx, lockCh)
 		if err != nil {
 			return false, err
 		}
+		if lockErr != nil {
+			return false, lockErr
+		}
 
 		return true, nil
 	})

worker2/poolui/approver.go

@@ -0,0 +1,77 @@
+package poolui
+
+import (
+	"context"
+	"github.com/hephbuild/heph/specs"
+	"github.com/hephbuild/heph/worker2"
+	"sync"
+)
+
+func NewApprover() *Approver {
+	return &Approver{
+		queue: make(chan ApproveRequest),
+	}
+}
+
+type ApproveRequest struct {
+	Target  specs.Target
+	Respond func(bool)
+}
+
+type Approver struct {
+	m         sync.Mutex
+	connected bool
+	queue     chan ApproveRequest
+}
+
+func (a *Approver) SetConnected(connected bool) {
+	a.m.Lock()
+	defer a.m.Unlock()
+
+	a.connected = connected
+
+	if !connected {
+		for {
+			select {
+			case req := <-a.queue:
+				req.Respond(false)
+			default:
+				return
+			}
+		}
+	}
+}
+
+func (a *Approver) Approve(ctx context.Context, t specs.Target) bool {
+	a.m.Lock()
+
+	if !a.connected {
+		a.m.Unlock()
+		return false
+	}
+	a.m.Unlock()
+
+	valueCh := make(chan bool)
+	defer close(valueCh)
+
+	err := worker2.WaitChanSend(ctx, a.queue, ApproveRequest{
+		Target: t,
+		Respond: func(approved bool) {
+			valueCh <- approved
+		},
+	})
+	if err != nil {
+		return false
+	}
+
+	res, err := worker2.WaitChanReceive(ctx, valueCh)
+	if err != nil {
+		return false
+	}
+
+	return res
+}
+
+func (a *Approver) Next() ApproveRequest {
+	return <-a.queue
+}