fix(task): Windows task cancellation and SSH tunnel lifecycle hardening

.github/workflows/pr-ci.yml

@@ -302,6 +302,12 @@ jobs:
             install-kind: false
             requires-secret: false
 
+          - label: task-cancel
+            runner: ubuntu-latest
+            free-disk-space: false
+            install-kind: false
+            requires-secret: false
+
           - label: ssh-credentials-server-race
             runner: ubuntu-latest
             free-disk-space: false
@@ -592,6 +598,17 @@ jobs:
             free-disk-space: false
             install-kind: true
             requires-secret: false
+          - label: ssh-tunnel-mode
+            runner: windows-latest
+            free-disk-space: false
+            install-kind: true
+            requires-secret: false
+
+          - label: task-cancel
+            runner: windows-latest
+            free-disk-space: false
+            install-kind: true
+            requires-secret: false
 
 
     runs-on: ${{ matrix.runner }}

Taskfile.yml

@@ -116,10 +116,20 @@ tasks:
         host_os="$(go env GOOS)"
         host_arch="$(go env GOARCH)"
         host_binary="devsy-${host_os}-${host_arch}"
-        host_dir="dist/devsy-dev_${host_os}_${host_arch}_v1"
         if [ "$host_os" = "windows" ]; then
           host_binary="${host_binary}.exe"
         fi
+        host_dir=""
+        for candidate in dist/devsy-dev_${host_os}_${host_arch}_*; do
+          if [ -d "$candidate" ]; then
+            if [ -n "$host_dir" ]; then
+              echo "multiple host build directories match ${host_os}/${host_arch}" >&2
+              exit 1
+            fi
+            host_dir="$candidate"
+          fi
+        done
+        test -n "$host_dir"
         test -f "${host_dir}/${host_binary}"
         cp "${host_dir}/${host_binary}" "e2e/bin/${host_binary}"
 

cmd/workspace/stop.go

@@ -22,6 +22,7 @@ import (
 type StopCmd struct {
 	*flags.GlobalFlags
 	client2.StopOptions
+	quiescer workspace2.UpTaskQuiescer
 }
 
 // NewStopCmd creates a new destroy command.
@@ -123,6 +124,12 @@ func (cmd *StopCmd) run(
 	devsyConfig *config.Config,
 	client client2.BaseWorkspaceClient,
 ) error {
+	// Quiesce before waiting for the workspace lock: a still-provisioning up
+	// worker can be the one holding it.
+	if err := cmd.quiesce(ctx, client.Workspace()); err != nil {
+		return err
+	}
+
 	// lock workspace
 	if !cmd.Platform.Enabled {
 		err := status.RunStep(
@@ -137,6 +144,13 @@ func (cmd *StopCmd) run(
 		defer client.Unlock()
 	}
 
+	// Quiesce again under the lock: a task may have become visible while
+	// the lock wait blocked, and a surviving up worker can restart the
+	// workspace after the stop.
+	if err := cmd.quiesce(ctx, client.Workspace()); err != nil {
+		return err
+	}
+
 	// get instance status
 	var instanceStatus client2.Status
 	err := status.RunStep(
@@ -255,3 +269,18 @@ func otherWorkspaceUsesMachine(
 	}
 	return false
 }
+
+func (cmd *StopCmd) quiesce(ctx context.Context, workspaceID string) error {
+	quiescer := cmd.quiescer
+	if quiescer == nil {
+		quiescer = workspace2.StoreUpTaskQuiescer{}
+	}
+	if err := quiescer.Quiesce(ctx, workspaceID); err != nil {
+		return fmt.Errorf(
+			"cancel detached up tasks for workspace %s: %w",
+			workspaceID,
+			err,
+		)
+	}
+	return nil
+}

cmd/workspace/stop_test.go

@@ -0,0 +1,131 @@
+package workspace
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/devsy-org/devsy/cmd/flags"
+	client2 "github.com/devsy-org/devsy/pkg/client"
+	"github.com/devsy-org/devsy/pkg/config"
+	"github.com/devsy-org/devsy/pkg/provider"
+	"github.com/devsy-org/devsy/pkg/task"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+type stopRecordingClient struct {
+	*fakeWorkspaceClient
+	calls *[]string
+}
+
+type quiescerFunc func(context.Context, string) error
+
+func (f quiescerFunc) Quiesce(ctx context.Context, workspaceID string) error {
+	return f(ctx, workspaceID)
+}
+
+func (c *stopRecordingClient) Stop(ctx context.Context, opts client2.StopOptions) error {
+	*c.calls = append(*c.calls, "stop")
+	return c.fakeWorkspaceClient.Stop(ctx, opts)
+}
+
+func newStopTestConfig() *config.Config {
+	return &config.Config{
+		DefaultContext: testContext,
+		Contexts:       map[string]*config.ContextConfig{testContext: {}},
+	}
+}
+
+func TestStopCancelsPersistedUpTaskBeforeProviderStop(t *testing.T) {
+	var calls []string
+	fake := &fakeWorkspaceClient{
+		workspace: testWorkspaceName,
+		context:   testContext,
+		provider:  testProvider,
+		config:    &provider.Workspace{ID: testWorkspaceName, Context: testContext},
+		status:    client2.StatusRunning,
+	}
+	client := &stopRecordingClient{fakeWorkspaceClient: fake, calls: &calls}
+	cmd := &StopCmd{GlobalFlags: &flags.GlobalFlags{ResultFormat: formatPlain}}
+	cmd.quiescer = quiescerFunc(func(_ context.Context, workspaceID string) error {
+		assert.Equal(t, testWorkspaceName, workspaceID)
+		calls = append(calls, "quiesce")
+		return nil
+	})
+
+	require.NoError(t, cmd.run(t.Context(), newStopTestConfig(), client))
+
+	require.NotEmpty(t, calls)
+	assert.Equal(t, "quiesce", calls[0])
+	assert.Contains(t, calls, "stop")
+	stopIndex := -1
+	for i, call := range calls {
+		if call == "stop" {
+			stopIndex = i
+			break
+		}
+	}
+	quiesceBeforeStop := 0
+	for _, call := range calls[:stopIndex] {
+		if call == "quiesce" {
+			quiesceBeforeStop++
+		}
+	}
+	assert.Equal(t, 2, quiesceBeforeStop, "stop must quiesce before and after the workspace lock")
+}
+
+func TestStopDoesNotProceedWhenActiveTaskCannotBeKilled(t *testing.T) {
+	fake := &fakeWorkspaceClient{
+		workspace: testWorkspaceName,
+		context:   testContext,
+		provider:  testProvider,
+		config:    &provider.Workspace{ID: testWorkspaceName, Context: testContext},
+		status:    client2.StatusRunning,
+	}
+	var calls []string
+	client := &stopRecordingClient{fakeWorkspaceClient: fake, calls: &calls}
+	cmd := &StopCmd{GlobalFlags: &flags.GlobalFlags{ResultFormat: formatPlain}}
+	cmd.quiescer = quiescerFunc(func(context.Context, string) error {
+		return errors.New("boom")
+	})
+
+	err := cmd.run(t.Context(), newStopTestConfig(), client)
+	require.Error(t, err)
+	assert.NotContains(t, calls, "stop")
+}
+
+func TestStopFindsTaskWithoutDesktopState(t *testing.T) {
+	home := t.TempDir()
+	t.Setenv("DEVSY_HOME", home)
+	t.Setenv("HOME", home)
+	t.Setenv("USERPROFILE", home)
+	config.ResetPathManager()
+	t.Cleanup(config.ResetPathManager)
+
+	// A persisted up task with no live worker, as a Desktop restart leaves
+	// behind.
+	store, err := task.NewStore()
+	require.NoError(t, err)
+	tk, err := store.Create(task.CreateOptions{Command: "up", WorkspaceID: testWorkspaceName})
+	require.NoError(t, err)
+
+	fake := &fakeWorkspaceClient{
+		workspace: testWorkspaceName,
+		context:   testContext,
+		provider:  testProvider,
+		config:    &provider.Workspace{ID: testWorkspaceName, Context: testContext},
+		status:    client2.StatusRunning,
+	}
+	var calls []string
+	client := &stopRecordingClient{fakeWorkspaceClient: fake, calls: &calls}
+	cmd := &StopCmd{GlobalFlags: &flags.GlobalFlags{ResultFormat: formatPlain}}
+
+	require.NoError(t, cmd.run(t.Context(), newStopTestConfig(), client))
+	assert.Contains(t, calls, "stop")
+
+	state, err := store.Get(tk.ID())
+	require.NoError(t, err)
+	assert.Equal(t, task.StatusFailed, state.Status)
+	assert.Equal(t, task.ErrCanceled.Error(), state.Error)
+}

cmd/workspace/up/detach.go

@@ -1,16 +1,20 @@
 package up
 
 import (
+	"errors"
 	"fmt"
 	"os"
 	"os/exec"
 	"strings"
+	"time"
 
 	"github.com/devsy-org/devsy/pkg/command"
 	config2 "github.com/devsy-org/devsy/pkg/devcontainer/config"
+	"github.com/devsy-org/devsy/pkg/file"
 	"github.com/devsy-org/devsy/pkg/flags/names"
 	"github.com/devsy-org/devsy/pkg/output"
 	"github.com/devsy-org/devsy/pkg/task"
+	workspace2 "github.com/devsy-org/devsy/pkg/workspace"
 )
 
 // runDetached submits this invocation as a background task and returns
@@ -20,15 +24,24 @@ func (cmd *UpCmd) runDetached(args []string) error {
 	if err != nil {
 		return err
 	}
+	workspaceID, err := cmd.detachWorkspaceLabel(args)
+	if err != nil {
+		return err
+	}
 	t, err := store.Create(task.CreateOptions{
 		Command:     "up",
-		WorkspaceID: cmd.detachWorkspaceLabel(args),
+		WorkspaceID: workspaceID,
 	})
 	if err != nil {
 		return err
 	}
 
-	if err := launchDetached(t.ID()); err != nil {
+	if err := t.BeginLaunch(); err != nil {
+		_ = t.Fail(err)
+		return fmt.Errorf("prepare detached up: %w", err)
+	}
+	if err := launchDetached(t); err != nil {
+		_ = t.FinishLaunch()
 		_ = t.Fail(err)
 		return fmt.Errorf("launch detached up: %w", err)
 	}
@@ -46,14 +59,19 @@ func (cmd *UpCmd) runDetached(args []string) error {
 	return err
 }
 
-func launchDetached(taskID string) error {
+func launchDetached(t *task.Task) error {
 	execPath, err := os.Executable()
 	if err != nil {
 		return err
 	}
 
-	args := append(detachedArgs(os.Args[1:]), names.Flag(names.TaskID), taskID)
-	return command.StartBackground(task.WorkerProcessName(taskID), func() (*exec.Cmd, error) {
+	args := append(detachedArgs(os.Args[1:]), names.Flag(names.TaskID), t.ID())
+	return command.StartSupervisedBackground(command.SupervisedStartOptions{
+		Name: task.WorkerProcessName(t.ID()),
+		OnStarted: func(ref command.ProcessRef) error {
+			return t.SetProcess(ref)
+		},
+	}, func() (*exec.Cmd, error) {
 		return &exec.Cmd{
 			Path: execPath,
 			Args: append([]string{execPath}, args...),
@@ -76,15 +94,22 @@ func detachedArgs(args []string) []string {
 	return out
 }
 
-// detachWorkspaceLabel is a best-effort label for task list.
-func (cmd *UpCmd) detachWorkspaceLabel(args []string) string {
+// detachWorkspaceLabel derives the ID stop and delete use before worker startup.
+func (cmd *UpCmd) detachWorkspaceLabel(args []string) (string, error) {
+	if cmd.FromSnapshot != "" {
+		if _, err := cmd.resolveExplicitSource(); err != nil {
+			return "", err
+		}
+	}
 	if cmd.ID != "" {
-		return cmd.ID
+		return cmd.ID, nil
 	}
+	args = cmd.ensureArgs(args)
 	if len(args) > 0 {
-		return args[0]
+		_, source := file.IsLocalDir(args[0])
+		return workspace2.ToID(source), nil
 	}
-	return ""
+	return "", nil
 }
 
 func wd() string {
@@ -105,16 +130,45 @@ func (cmd *UpCmd) openTask() (*task.Task, error) {
 		return nil, err
 	}
 	t := store.Open(cmd.taskID)
-	// obtain the worker lock first.
 	if err := t.HoldWorkerLock(); err != nil {
 		failTask(t, err)
 		return nil, err
 	}
-	if err := t.SetPID(os.Getpid()); err != nil {
-		failTask(t, err)
-		return nil, err
+	deadline := time.NewTimer(5 * time.Second)
+	defer deadline.Stop()
+	ticker := time.NewTicker(20 * time.Millisecond)
+	defer ticker.Stop()
+	for {
+		state, err := store.Get(cmd.taskID)
+		if err != nil {
+			failTask(t, err)
+			_ = t.ReleaseWorkerLock()
+			return nil, err
+		}
+		if state.Status.Terminal() || state.CancelRequested {
+			if err := t.ReleaseWorkerLock(); err != nil {
+				return nil, fmt.Errorf("release canceled worker lock: %w", err)
+			}
+			return nil, task.ErrCanceled
+		}
+		if !state.LaunchPending {
+			if _, ok := state.ProcessReference(); !ok {
+				err := errors.New("detached worker launch did not publish process metadata")
+				failTask(t, err)
+				_ = t.ReleaseWorkerLock()
+				return nil, err
+			}
+			return t, nil
+		}
+		select {
+		case <-deadline.C:
+			err := errors.New("timed out waiting for detached worker launch metadata")
+			failTask(t, err)
+			_ = t.ReleaseWorkerLock()
+			return nil, err
+		case <-ticker.C:
+		}
 	}
-	return t, nil
 }
 
 func failTask(t *task.Task, err error) {