feat(pro): cancel active up tasks if new one is started

Author: pascalbreuninger

cmd/up.go

@@ -1430,7 +1430,7 @@ func WithSignals(ctx context.Context) (context.Context, func()) {
 	signal.Notify(signals, os.Interrupt, syscall.SIGHUP, syscall.SIGTERM, syscall.SIGQUIT)
 	go func() {
 		select {
-		case _ = <-signals:
+		case <-signals:
 			cancel()
 		case <-ctx.Done():
 		}

desktop/src/client/command.ts

@@ -200,7 +200,7 @@ export class Command implements TCommand<ChildProcess<string>> {
         }),
       })
 
-      await sleep(2_000)
+      await sleep(3_000)
       // the actual child process could be gone after sending a SIGINT
       // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
       if (this.childProcess) {

pkg/client/clientimplementation/daemonclient/client.go

@@ -6,8 +6,6 @@ import (
 	"fmt"
 	"io"
 	"net/netip"
-	"os"
-	"path/filepath"
 	"strings"
 	"sync"
 	"time"
@@ -62,8 +60,7 @@ func New(devPodConfig *config.Config, prov *provider.ProviderConfig, workspace *
 type client struct {
 	m sync.Mutex
 
-	workspaceLockOnce sync.Once
-	workspaceLock     *flock.Flock
+	workspaceLock *flock.Flock
 
 	devPodConfig *config.Config
 	config       *provider.ProviderConfig
@@ -74,27 +71,12 @@ type client struct {
 }
 
 func (c *client) Lock(ctx context.Context) error {
-	c.initLock()
-
-	// try to lock workspace
-	c.log.Debugf("Acquire workspace lock...")
-	err := tryLock(ctx, c.workspaceLock, "workspace", c.log)
-	if err != nil {
-		return fmt.Errorf("error locking workspace: %w", err)
-	}
-	c.log.Debugf("Acquired workspace lock...")
-
+	// noop
 	return nil
 }
 
 func (c *client) Unlock() {
-	c.initLock()
-
-	// try to unlock workspace
-	err := c.workspaceLock.Unlock()
-	if err != nil {
-		c.log.Warnf("Error unlocking workspace: %v", err)
-	}
+	// noop
 }
 
 func (c *client) Provider() string {
@@ -207,23 +189,6 @@ func (c *client) DirectTunnel(ctx context.Context, stdin io.Reader, stdout io.Wr
 	}
 }
 
-func (c *client) initLock() {
-	c.workspaceLockOnce.Do(func() {
-		c.m.Lock()
-		defer c.m.Unlock()
-
-		// get locks dir
-		workspaceLocksDir, err := provider.GetLocksDir(c.workspace.Context)
-		if err != nil {
-			panic(fmt.Errorf("get workspaces dir: %w", err))
-		}
-		_ = os.MkdirAll(workspaceLocksDir, 0777)
-
-		// create workspace lock
-		c.workspaceLock = flock.New(filepath.Join(workspaceLocksDir, c.workspace.ID+".workspace.lock"))
-	})
-}
-
 func (c *client) Ping(ctx context.Context, writer io.Writer) error {
 	wAddr, err := c.getWorkspaceAddress()
 	if err != nil {
@@ -291,43 +256,3 @@ func (c *client) getWorkspaceAddress() (ts.Addr, error) {
 
 	return ts.NewAddr(ts.GetWorkspaceHostname(c.workspace.Pro.InstanceName, c.workspace.Pro.Project), sshServer.DefaultUserPort), nil
 }
-
-func printLogMessagePeriodically(message string, log log.Logger) chan struct{} {
-	done := make(chan struct{})
-	go func() {
-		for {
-			select {
-			case <-done:
-				return
-			case <-time.After(time.Second * 5):
-				log.Info(message)
-			}
-		}
-	}()
-
-	return done
-}
-
-func tryLock(ctx context.Context, lock *flock.Flock, name string, log log.Logger) error {
-	done := printLogMessagePeriodically(fmt.Sprintf("Trying to lock %s, seems like another process is running that blocks this %s", name, name), log)
-	defer close(done)
-
-	now := time.Now()
-	for time.Since(now) < time.Minute*5 {
-		locked, err := lock.TryLock()
-		if err != nil {
-			return err
-		} else if locked {
-			return nil
-		}
-
-		select {
-		case <-time.After(time.Second):
-			continue
-		case <-ctx.Done():
-			return ctx.Err()
-		}
-	}
-
-	return fmt.Errorf("timed out waiting to lock %s, seems like there is another process running on this machine that blocks it", name)
-}

pkg/client/clientimplementation/daemonclient/up.go

@@ -59,6 +59,24 @@ func (c *client) Up(ctx context.Context, opt clientpkg.UpOptions) (*config.Resul
 		return nil, fmt.Errorf("error getting management client: %w", err)
 	}
 
+	// prompt user to attach to active task or start new one
+	c.log.Debug("Check active up task")
+	activeUpTask, err := findActiveUpTask(ctx, managementClient, instance)
+	if err != nil {
+		return nil, fmt.Errorf("find active up task: %w", err)
+	}
+
+	// if we have an active up task, cancel it before creating a new one
+	if activeUpTask != nil {
+		c.log.Warnf("Found active up task %s, attempting to cancel it", activeUpTask.ID)
+		_, err = managementClient.Loft().ManagementV1().DevPodWorkspaceInstances(instance.Namespace).Cancel(ctx, instance.Name, &managementv1.DevPodWorkspaceInstanceCancel{
+			TaskID: activeUpTask.ID,
+		}, metav1.CreateOptions{})
+		if err != nil {
+			return nil, fmt.Errorf("cancel task: %w", err)
+		}
+	}
+
 	// create up task
 	task, err := managementClient.Loft().ManagementV1().DevPodWorkspaceInstances(instance.Namespace).Up(ctx, instance.Name, &managementv1.DevPodWorkspaceInstanceUp{
 		Spec: managementv1.DevPodWorkspaceInstanceUpSpec{
@@ -72,7 +90,11 @@ func (c *client) Up(ctx context.Context, opt clientpkg.UpOptions) (*config.Resul
 		return nil, fmt.Errorf("no up task id returned from server")
 	}
 
-	exitCode, err := observeTask(ctx, managementClient, instance, task.Status.TaskID, c.log)
+	return waitTaskDone(ctx, managementClient, instance, task.Status.TaskID, c.log)
+}
+
+func waitTaskDone(ctx context.Context, managementClient kube.Interface, instance *managementv1.DevPodWorkspaceInstance, taskID string, log log.Logger) (*config.Result, error) {
+	exitCode, err := observeTask(ctx, managementClient, instance, taskID, log)
 	if err != nil {
 		return nil, fmt.Errorf("up: %w", err)
 	} else if exitCode != 0 {
@@ -87,7 +109,7 @@ func (c *client) Up(ctx context.Context, opt clientpkg.UpOptions) (*config.Resul
 		Name(instance.Name).
 		SubResource("tasks").
 		VersionedParams(&managementv1.DevPodWorkspaceInstanceTasksOptions{
-			TaskID: task.Status.TaskID,
+			TaskID: taskID,
 		}, builders.ParameterCodec).
 		Do(ctx).
 		Into(tasks)
@@ -267,3 +289,33 @@ func printLogs(ctx context.Context, managementClient kube.Interface, workspace *
 
 	return 0, nil
 }
+
+const (
+	TaskStatusRunning = "Running"
+	TaskStatusSucceed = "Succeeded"
+	TaskStatusFailed  = "Failed"
+)
+const (
+	TaskTypeUp     = "up"
+	TaskTypeStop   = "stop"
+	TaskTypeDelete = "delete"
+)
+
+func findActiveUpTask(ctx context.Context, managementClient kube.Interface, instance *managementv1.DevPodWorkspaceInstance) (*managementv1.DevPodWorkspaceInstanceTask, error) {
+	tasks := &managementv1.DevPodWorkspaceInstanceTasks{}
+	err := managementClient.Loft().ManagementV1().RESTClient().Get().
+		Namespace(instance.Namespace).
+		Resource("devpodworkspaceinstances").
+		Name(instance.Name).
+		SubResource("tasks").
+		Do(ctx).
+		Into(tasks)
+
+	for _, task := range tasks.Tasks {
+		if task.Status == TaskStatusRunning && task.Type == TaskTypeUp {
+			return &task, nil
+		}
+	}
+
+	return nil, err
+}

pkg/ts/workspace_server.go

@@ -377,20 +377,20 @@ func (s *WorkspaceServer) sendHeartbeat(ctx context.Context, client *http.Client
 
 	discoveredRunner, err := s.discoverRunner(ctx, lc)
 	if err != nil {
-		return fmt.Errorf("failed to discover runner: %v", err)
+		return fmt.Errorf("failed to discover runner: %w", err)
 	}
 
 	heartbeatURL := fmt.Sprintf("http://%s.ts.loft/devpod/%s/%s/heartbeat", discoveredRunner, projectName, workspaceName)
 	s.log.Infof("Sending heartbeat to %s, because there are %d active connections", heartbeatURL, connections)
 	req, err := http.NewRequestWithContext(ctx, "GET", heartbeatURL, nil)
 	if err != nil {
-		return fmt.Errorf("failed to create request for %s: %v", heartbeatURL, err)
+		return fmt.Errorf("failed to create request for %s: %w", heartbeatURL, err)
 	}
 
 	req.Header.Set("Authorization", "Bearer "+s.config.AccessKey)
 	resp, err := client.Do(req)
 	if err != nil {
-		return fmt.Errorf("request to %s failed: %v", heartbeatURL, err)
+		return fmt.Errorf("request to %s failed: %w", heartbeatURL, err)
 	}
 	defer resp.Body.Close()
 
@@ -405,7 +405,7 @@ func (s *WorkspaceServer) sendHeartbeat(ctx context.Context, client *http.Client
 func (s *WorkspaceServer) discoverRunner(ctx context.Context, lc *tailscale.LocalClient) (string, error) {
 	status, err := lc.Status(ctx)
 	if err != nil {
-		return "", fmt.Errorf("failed to get status: %v", err)
+		return "", fmt.Errorf("failed to get status: %w", err)
 	}
 
 	var runner string