Enable interrupting tasks manually

Rather than having to do this externally in a shell.

It's actually not that meaningful a feature,
and primarily saves us from PID hunting.

acid.adoc

@@ -39,6 +39,8 @@ Commands
 *restart* [_ID_]...::
 	Schedule tasks with the given IDs to be rerun.
 	Run this command without arguments to pick up external database changes.
+*kill* _ID_...::
+	Interrupt running tasks with the given IDs.
 *reload*::
 	Reload configuration.
 

acid.go

@@ -655,6 +655,7 @@ func handlePush(w http.ResponseWriter, r *http.Request) {
 const rpcHeaderSignature = "X-ACID-Signature"
 
 var errWrongUsage = errors.New("wrong usage")
+var errTaskInterrupted = errors.New("task interrupted")
 
 func rpcRestartOne(ctx context.Context, id int64) error {
 	gRunningMutex.Lock()
@@ -683,6 +684,18 @@ func rpcRestartOne(ctx context.Context, id int64) error {
 	return nil
 }
 
+func rpcKillOne(ctx context.Context, id int64) error {
+	gRunningMutex.Lock()
+	defer gRunningMutex.Unlock()
+
+	rt, ok := gRunning[id]
+	if !ok {
+		return fmt.Errorf("%d: not a running task", id)
+	}
+	rt.cancel(errTaskInterrupted)
+	return nil
+}
+
 func rpcEnqueueOne(ctx context.Context,
 	owner, repo, hash, runner string) error {
 	tasks, err := getTasks(ctx, `WHERE owner = ? AND repo = ? AND hash = ?
@@ -798,6 +811,32 @@ func rpcRestart(ctx context.Context,
 	return nil
 }
 
+func rpcKill(ctx context.Context,
+	w io.Writer, fs *flag.FlagSet, args []string) error {
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+
+	ids := []int64{}
+	for _, arg := range fs.Args() {
+		id, err := strconv.ParseInt(arg, 10, 64)
+		if err != nil {
+			return fmt.Errorf("%w: %s", errWrongUsage, err)
+		}
+		ids = append(ids, id)
+	}
+	for _, id := range ids {
+		if err := rpcKillOne(ctx, id); err != nil {
+			fmt.Fprintln(w, err)
+		}
+	}
+
+	if len(ids) == 0 {
+		return errWrongUsage
+	}
+	return nil
+}
+
 func rpcReload(ctx context.Context,
 	w io.Writer, fs *flag.FlagSet, args []string) error {
 	if err := fs.Parse(args); err != nil {
@@ -821,6 +860,8 @@ var rpcCommands = map[string]struct {
 		"Create or restart tasks for the given reference."},
 	"restart": {rpcRestart, "[ID]...",
 		"Schedule tasks with the given IDs to be rerun."},
+	"kill": {rpcKill, "ID...",
+		"Interrupt running tasks with the given IDs."},
 	"reload": {rpcReload, "",
 		"Reload configuration."},
 }
@@ -1053,6 +1094,7 @@ type RunningTask struct {
 	DB            Task
 	Runner        ConfigRunner
 	ProjectRunner ConfigProjectRunner
+	cancel        context.CancelCauseFunc
 
 	RunLog    terminalWriter
 	TaskLog   terminalWriter
@@ -1418,6 +1460,9 @@ func executorRunTask(ctx context.Context, task Task) error {
 
 	ctx, cancelTimeout := context.WithTimeout(ctx, rt.timeout)
 	defer cancelTimeout()
+	ctxRunner, cancelRunner := context.WithCancelCause(ctx)
+	rt.cancel = cancelRunner
+	defer cancelRunner(context.Canceled)
 
 	// RunningTasks can be concurrently accessed by HTTP handlers.
 	locked := func(f func()) {
@@ -1451,7 +1496,7 @@ func executorRunTask(ctx context.Context, task Task) error {
 		}
 	}
 
-	cmd := exec.CommandContext(ctx, rt.Runner.Run)
+	cmd := exec.CommandContext(ctxRunner, rt.Runner.Run)
 	cmd.Env = rt.localEnv()
 
 	// Pushing the runner into a new process group that can be killed at once
@@ -1481,8 +1526,6 @@ func executorRunTask(ctx context.Context, task Task) error {
 		return err
 	}
 
-	ctxRunner, cancelRunner := context.WithCancelCause(ctx)
-	defer cancelRunner(context.Canceled)
 	go func() {
 		if err := cmd.Wait(); err != nil {
 			cancelRunner(err)