refactor: use mount type instead of strings (#48)

Refactors how container mounts (volumes) are managed within the system. Instead of relying on simple string representations, a new, more explicit Mount data structure is now used. This change enhances type safety, improves the clarity of mount configurations, makes the system easier to extend, and streamlines the underlying logic for volume management across various components.

Author: w-h-a

internal/engine/clients/runner/docker/runner.go

@@ -52,17 +52,12 @@ func (r *dockerRunner) Run(ctx context.Context, opts ...runner.RunOption) (strin
 
 	var mounts []mount.Mount
 
-	for _, v := range options.Volumes {
-		vol := strings.Split(v, ":")
-		if len(vol) != 2 {
-			return "", runner.ErrInvalidVolumeName
-		}
-		mount := mount.Mount{
+	for _, m := range options.Mounts {
+		mounts = append(mounts, mount.Mount{
 			Type:   mount.TypeVolume,
-			Source: vol[0],
-			Target: vol[1],
-		}
-		mounts = append(mounts, mount)
+			Source: m["source"],
+			Target: m["target"],
+		})
 	}
 
 	hc := container.HostConfig{

internal/engine/clients/runner/errors.go

@@ -3,9 +3,8 @@ package runner
 import "errors"
 
 var (
-	ErrPullingImage      = errors.New("failed to pull image")
-	ErrVolumeNotFound    = errors.New("volume not found")
-	ErrInvalidVolumeName = errors.New("invalid volume name")
-	ErrBadExitCode       = errors.New("bad exit code")
-	ErrRunnerClosing     = errors.New("runner is closing")
+	ErrPullingImage   = errors.New("failed to pull image")
+	ErrVolumeNotFound = errors.New("volume not found")
+	ErrBadExitCode    = errors.New("bad exit code")
+	ErrRunnerClosing  = errors.New("runner is closing")
 )

internal/engine/clients/runner/options.go

@@ -48,7 +48,7 @@ type RunOptions struct {
 	Image    string
 	Cmd      []string
 	Env      []string
-	Volumes  []string
+	Mounts   []map[string]string
 	Networks []string
 	Context  context.Context
 }
@@ -77,9 +77,9 @@ func RunWithEnv(env []string) RunOption {
 	}
 }
 
-func RunWithVolumes(volumes []string) RunOption {
+func RunWithMounts(mounts []map[string]string) RunOption {
 	return func(o *RunOptions) {
-		o.Volumes = volumes
+		o.Mounts = mounts
 	}
 }
 

internal/engine/services/coordinator/service.go

@@ -143,43 +143,18 @@ func (s *Service) RestartTask(ctx context.Context, id string) (*task.Task, error
 		return nil, task.ErrTaskNotRestartable
 	}
 
-	if len(t.Volumes) > 0 {
-		var cleanedVolumes []string
-		for _, v := range t.Volumes {
-			parts := strings.Split(v, ":")
-			if len(parts) == 2 {
-				cleanedVolumes = append(cleanedVolumes, parts[1])
-			} else {
-				cleanedVolumes = append(cleanedVolumes, v)
-			}
+	if len(t.Mounts) > 0 {
+		for _, m := range t.Mounts {
+			m.Source = ""
 		}
-		t.Volumes = cleanedVolumes
 	}
 
 	for _, pre := range t.Pre {
-		var cleanedPreVolumes []string
-		for _, v := range pre.Volumes {
-			parts := strings.Split(v, ":")
-			if len(parts) == 2 {
-				cleanedPreVolumes = append(cleanedPreVolumes, parts[1])
-			} else {
-				cleanedPreVolumes = append(cleanedPreVolumes, v)
-			}
-		}
-		pre.Volumes = cleanedPreVolumes
+		pre.Mounts = nil
 	}
 
 	for _, post := range t.Post {
-		var cleanedPostVolumes []string
-		for _, v := range post.Volumes {
-			parts := strings.Split(v, ":")
-			if len(parts) == 2 {
-				cleanedPostVolumes = append(cleanedPostVolumes, parts[1])
-			} else {
-				cleanedPostVolumes = append(cleanedPostVolumes, v)
-			}
-		}
-		post.Volumes = cleanedPostVolumes
+		post.Mounts = nil
 	}
 
 	now := time.Now()

internal/engine/services/worker/service.go

@@ -3,7 +3,6 @@ package worker
 import (
 	"context"
 	"encoding/json"
-	"fmt"
 	"log/slog"
 	"strings"
 	"sync"
@@ -113,9 +112,9 @@ func (s *Service) runTask(ctx context.Context, t *task.Task) error {
 		return err
 	}
 
-	var vs []string
+	var ms []*task.Mount
 
-	for _, v := range t.Volumes {
+	for _, m := range t.Mounts {
 		volName := strings.ReplaceAll(uuid.NewString(), "-", "")
 		opts := []runner.CreateVolumeOption{
 			runner.CreateVolumeWithName(volName),
@@ -147,14 +146,17 @@ func (s *Service) runTask(ctx context.Context, t *task.Task) error {
 				slog.ErrorContext(ctx, "failed to delete volume", "name", volName)
 			}
 		}(volName)
-		vs = append(vs, fmt.Sprintf("%s:%s", volName, v))
+		ms = append(ms, &task.Mount{
+			Source: volName,
+			Target: m.Target,
+		})
 	}
 
-	t.Volumes = vs
+	t.Mounts = ms
 
 	for _, pre := range t.Pre {
 		pre.ID = strings.ReplaceAll(uuid.NewString(), "-", "")
-		pre.Volumes = t.Volumes
+		pre.Mounts = t.Mounts
 		pre.Networks = t.Networks
 		result, err := s.run(ctx, pre)
 		finished := time.Now()
@@ -202,7 +204,7 @@ func (s *Service) runTask(ctx context.Context, t *task.Task) error {
 
 	for _, post := range t.Post {
 		post.ID = strings.ReplaceAll(uuid.NewString(), "-", "")
-		post.Volumes = t.Volumes
+		post.Mounts = t.Mounts
 		post.Networks = t.Networks
 		result, err := s.run(ctx, post)
 		finished := time.Now()
@@ -251,12 +253,21 @@ func (s *Service) run(ctx context.Context, t *task.Task) (string, error) {
 		ctx = timeoutCtx
 	}
 
+	var mounts []map[string]string
+
+	for _, m := range t.Mounts {
+		mounts = append(mounts, map[string]string{
+			"source": m.Source,
+			"target": m.Target,
+		})
+	}
+
 	runOpts := []runner.RunOption{
 		runner.RunWithID(t.ID),
 		runner.RunWithImage(t.Image),
 		runner.RunWithCmd(t.Cmd),
 		runner.RunWithEnv(t.Env),
-		runner.RunWithVolumes(t.Volumes),
+		runner.RunWithMounts(mounts),
 		runner.RunWithNetworks(t.Networks),
 	}
 

internal/task/domain.go

@@ -30,12 +30,17 @@ type Task struct {
 	Error       string     `json:"error,omitempty"`
 	Pre         []*Task    `json:"pre,omitempty"`
 	Post        []*Task    `json:"post,omitempty"`
-	Volumes     []string   `json:"volumes,omitempty"`
+	Mounts      []*Mount   `json:"mounts,omitempty"`
 	Networks    []string   `json:"networks,omitempty"`
 	Retry       *Retry     `json:"retry,omitempty"`
 	Timeout     string     `json:"timeout,omitempty"`
 }
 
+type Mount struct {
+	Source string `json:"source,omitempty"`
+	Target string `json:"target,omitempty"`
+}
+
 type Retry struct {
 	Limit    int `json:"limit,omitempty"`
 	Attempts int `json:"attempts,omitempty"`

internal/task/errors.go

@@ -5,6 +5,7 @@ import "errors"
 var (
 	ErrTaskNotFound       = errors.New("task not found")
 	ErrMissingImage       = errors.New("task is missing image")
+	ErrMountMissingTarget = errors.New("mount missing target")
 	ErrAttemptsSpecified  = errors.New("may not specify retry attempts")
 	ErrExcessiveLimit     = errors.New("may not specify retry limit > 10")
 	ErrInvalidTimeout     = errors.New("invalid timeout")

internal/task/task.go

@@ -19,6 +19,12 @@ func Factory(bs []byte) (*Task, error) {
 
 	// TODO: validate queue field
 
+	for _, m := range t.Mounts {
+		if m == nil || len(strings.TrimSpace(m.Target)) == 0 {
+			return nil, ErrMountMissingTarget
+		}
+	}
+
 	if t.Retry != nil {
 		if t.Retry.Attempts != 0 {
 			return nil, ErrAttemptsSpecified

tests/testdata/convert.json

@@ -8,7 +8,7 @@
     "5",
     "/tmp/output.mp4"
   ],
-  "volumes": ["/tmp"],
+  "mounts": [{"target": "/tmp"}],
   "pre": [{
     "name": "download the remote file",
     "image": "alpine:3.18.3",