fix: cancelling multiple tasks (#84)

hacks/watch.sh

@@ -2,5 +2,5 @@
 
 # in foreground, continously run app
 while true; do
-    _build/pug -w ./demo/configs -d
+    _build/pug -w ./demo -d
 done

internal/app/task_test.go

@@ -1,11 +1,17 @@
 package app
 
 import (
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
 	"strings"
 	"testing"
 
 	tea "github.com/charmbracelet/bubbletea"
 	"github.com/leg100/pug/internal"
+	"github.com/stretchr/testify/require"
 )
 
 func TestTaskList_Split(t *testing.T) {
@@ -165,3 +171,139 @@ func TestTask_RetryMultiple(t *testing.T) {
 		return matchPattern(t, "TaskGroup.*retry.*3/3", s)
 	})
 }
+func TestTask_Cancel(t *testing.T) {
+	t.Parallel()
+
+	tm := setup(t, "./testdata/cancel_single/")
+
+	// Go to modules listing
+	tm.Type("m")
+
+	// Expect single module to be listed
+	waitFor(t, tm, func(s string) bool {
+		return strings.Contains(s, "modules/a")
+	})
+
+	// Initialize module
+	tm.Type("i")
+	waitFor(t, tm, func(s string) bool {
+		return matchPattern(t, "Task.*init.*modules/a.*exited", s)
+	})
+
+	// Go back to modules listing
+	tm.Send(tea.KeyMsg{Type: tea.KeyEsc})
+
+	// Expect single module to be listed, along with its default workspace.
+	waitFor(t, tm, func(s string) bool {
+		return matchPattern(t, "modules/a.*default", s)
+	})
+
+	// Stand up http server to receive request from terraform plan
+	setupHTTPServer(t, tm.workdir, "a")
+
+	// Invoke plan on module
+	tm.Type("p")
+
+	// Wait for something that never arrives
+	waitFor(t, tm, func(s string) bool {
+		// Remove bold formatting
+		s = internal.StripAnsi(s)
+		return strings.Contains(s, "data.http.forever: Reading...")
+	})
+
+	// Cancel plan
+	tm.Type("c")
+	waitFor(t, tm, func(s string) bool {
+		return strings.Contains(s, "Cancel task? (y/N):")
+	})
+	tm.Type("y")
+
+	// Wait for footer to report signal sent, and for process to receive signal
+	waitFor(t, tm, func(s string) bool {
+		return strings.Contains(s, "sent cancel signal to task") &&
+			strings.Contains(s, "Interrupt received")
+	})
+}
+
+func TestTask_CancelMultiple(t *testing.T) {
+	t.Parallel()
+
+	tm := setup(t, "./testdata/cancel_multiple/")
+
+	// Go to modules listing
+	tm.Type("m")
+
+	// Expect three modules to be listed
+	waitFor(t, tm, func(s string) bool {
+		return strings.Contains(s, "modules/a") &&
+			strings.Contains(s, "modules/b") &&
+			strings.Contains(s, "modules/c")
+	})
+
+	// Select all modules and init
+	tm.Send(tea.KeyMsg{Type: tea.KeyCtrlA})
+	tm.Type("i")
+	waitFor(t, tm, func(s string) bool {
+		return matchPattern(t, "TaskGroup.*init", s) &&
+			matchPattern(t, `modules/a.*exited`, s) &&
+			matchPattern(t, `modules/b.*exited`, s) &&
+			matchPattern(t, `modules/c.*exited`, s)
+	})
+
+	// Go back to modules listing
+	tm.Send(tea.KeyMsg{Type: tea.KeyEsc})
+
+	// Expect three modules to be listed, along with their default workspace.
+	waitFor(t, tm, func(s string) bool {
+		return matchPattern(t, "modules/a.*default", s) &&
+			matchPattern(t, "modules/b.*default", s) &&
+			matchPattern(t, "modules/c.*default", s)
+	})
+
+	// Stand up http server to receive request from terraform plans
+	setupHTTPServer(t, tm.workdir, "a", "b", "c")
+
+	// Invoke plans on modules
+	tm.Type("p")
+
+	// Wait for plan tasks to enter running state.
+	waitFor(t, tm, func(s string) bool {
+		return matchPattern(t, `modules/a.*default.*plan.*running`, s) &&
+			matchPattern(t, `modules/b.*default.*plan.*running`, s) &&
+			matchPattern(t, `modules/c.*default.*plan.*running`, s)
+	})
+
+	// Cancel plans
+	tm.Send(tea.KeyMsg{Type: tea.KeyCtrlA})
+	tm.Type("c")
+	waitFor(t, tm, func(s string) bool {
+		return strings.Contains(s, "Cancel 3 tasks? (y/N):")
+	})
+	tm.Type("y")
+
+	// Wait for footer to report signals sent, and for processes to receive signal
+	waitFor(t, tm, func(s string) bool {
+		return strings.Contains(s, "sent cancel signal to 3 tasks")
+	})
+}
+
+// Stand up http server to receive http request, and write out its URL to a
+// tfvars file in each of the given modules.
+func setupHTTPServer(t *testing.T, workdir string, mods ...string) {
+	ch := make(chan struct{})
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		// Block request
+		<-ch
+	}))
+	t.Cleanup(srv.Close)
+	t.Cleanup(func() { close(ch) })
+
+	// Write out a tfvars files with the url variable set to the url of the http
+	// server
+	for _, mod := range mods {
+		path := filepath.Join(workdir, "modules", mod, "default.tfvars")
+		contents := fmt.Sprintf(`url = "%s"`, srv.URL)
+		err := os.WriteFile(path, []byte(contents), 0o644)
+		require.NoError(t, err)
+	}
+}

internal/app/testdata/cancel_multiple/modules/a/main.tf

@@ -0,0 +1,11 @@
+terraform {
+  backend "local" {}
+}
+
+// Requires that an http server be setup locally on $url - if the server
+// doesn't respond to the request then this should hang indefinitely.
+data "http" "forever" {
+  url = var.url
+}
+
+variable "url" {}

internal/app/testdata/cancel_multiple/modules/b/main.tf

@@ -0,0 +1,11 @@
+terraform {
+  backend "local" {}
+}
+
+// Requires that an http server be setup locally on $url - if the server
+// doesn't respond to the request then this should hang indefinitely.
+data "http" "forever" {
+  url = var.url
+}
+
+variable "url" {}

internal/app/testdata/cancel_multiple/modules/c/main.tf

@@ -0,0 +1,11 @@
+terraform {
+  backend "local" {}
+}
+
+// Requires that an http server be setup locally on $url - if the server
+// doesn't respond to the request then this should hang indefinitely.
+data "http" "forever" {
+  url = var.url
+}
+
+variable "url" {}

internal/app/testdata/cancel_single/modules/a/main.tf

@@ -0,0 +1,11 @@
+terraform {
+  backend "local" {}
+}
+
+// Requires that an http server be setup locally on $url - if the server
+// doesn't respond to the request then this should hang indefinitely.
+data "http" "forever" {
+  url = var.url
+}
+
+variable "url" {}

internal/task/service.go

@@ -210,14 +210,18 @@ func (s *Service) Get(taskID resource.ID) (*Task, error) {
 }
 
 func (s *Service) Cancel(taskID resource.ID) (*Task, error) {
-	task, err := s.tasks.Get(taskID)
+	task, err := func() (*Task, error) {
+		task, err := s.tasks.Get(taskID)
+		if err != nil {
+			return nil, err
+		}
+		return task, task.cancel()
+	}()
 	if err != nil {
-		s.logger.Error("canceling task", "id", taskID)
+		s.logger.Error("canceling task", "id", taskID, "error", err)
 		return nil, err
 	}
 
-	task.cancel()
-
 	s.logger.Info("canceled task", "task", task)
 	return task, nil
 }

internal/task/task.go

@@ -233,24 +233,20 @@ func (t *Task) LogValue() slog.Value {
 
 // cancel the task - if it is queued it'll skip the running state and enter the
 // exited state
-func (t *Task) cancel() {
+func (t *Task) cancel() error {
 	// lock task state so that cancelation can atomically both inspect current
 	// state and update state
 	t.mu.Lock()
 	defer t.mu.Unlock()
 
 	switch t.State {
 	case Exited, Errored, Canceled:
-		// silently take no action if already finished
-		return
+		return errors.New("task has already finished")
 	case Pending, Queued:
 		t.updateState(Canceled)
-		return
+		return nil
 	default: // running
-		// ignore any errors from signal; instead take a "best effort" approach
-		// to canceling
-		_ = t.proc.Signal(os.Interrupt)
-		return
+		return t.proc.Signal(os.Interrupt)
 	}
 }
 

internal/tui/task/cancel.go

@@ -0,0 +1,45 @@
+package task
+
+import (
+	"errors"
+	"fmt"
+
+	tea "github.com/charmbracelet/bubbletea"
+	"github.com/leg100/pug/internal/resource"
+	"github.com/leg100/pug/internal/tui"
+)
+
+// cancel task(s)
+func cancel(svc tui.TaskService, taskIDs ...resource.ID) tea.Cmd {
+	var (
+		prompt string
+		cmd    tea.Cmd
+	)
+	switch len(taskIDs) {
+	case 0:
+		return nil
+	case 1:
+		prompt = "Cancel task?"
+		cmd = func() tea.Msg {
+			if _, err := svc.Cancel(taskIDs[0]); err != nil {
+				return tui.ErrorMsg(fmt.Errorf("cancelling task: %w", err))
+			}
+			return tui.InfoMsg("sent cancel signal to task")
+		}
+	default:
+		prompt = fmt.Sprintf("Cancel %d tasks?", len(taskIDs))
+		cmd = func() tea.Msg {
+			var errored bool
+			for _, id := range taskIDs {
+				if _, err := svc.Cancel(id); err != nil {
+					errored = true
+				}
+			}
+			if errored {
+				return tui.ErrorMsg(errors.New("one or more cancel requests failed; see logs"))
+			}
+			return tui.InfoMsg(fmt.Sprintf("sent cancel signal to %d tasks", len(taskIDs)))
+		}
+	}
+	return tui.YesNoPrompt(prompt, cmd)
+}

internal/tui/task/list.go

@@ -136,7 +136,7 @@ func (m List) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
 		switch {
 		case key.Matches(msg, keys.Common.Cancel):
 			taskIDs := m.Table.SelectedOrCurrentIDs()
-			return m, m.helpers.CreateTasks("cancel", m.tasks.Cancel, taskIDs...)
+			return m, cancel(m.tasks, taskIDs...)
 		case key.Matches(msg, keys.Global.Enter):
 			if row, ok := m.Table.CurrentRow(); ok {
 				return m, tui.NavigateTo(tui.TaskKind, tui.WithParent(row.Value))

internal/tui/task/model.go

@@ -134,7 +134,7 @@ func (m model) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
 	case tea.KeyMsg:
 		switch {
 		case key.Matches(msg, keys.Common.Cancel):
-			return m, m.helpers.CreateTasks("cancel", m.svc.Cancel, m.task.ID)
+			return m, cancel(m.svc, m.task.ID)
 		case key.Matches(msg, keys.Common.Apply):
 			if m.run != nil {
 				// Only trigger an apply if run is in the planned state

mirror/providers.tf

@@ -6,5 +6,8 @@ terraform {
     time = {
       version = "= 0.11.1"
     }
+    http = {
+      version = "= 3.4.3"
+    }
   }
 }