Merge pull request #1408 from calypr/chore/fix_file_mapper

Fix input/output duplication in `file_mapper`

Author: lbeckman314

go.mod

@@ -113,6 +113,7 @@ require (
 	github.com/shirou/gopsutil/v4 v4.26.3 // indirect
 	github.com/shopspring/decimal v1.4.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.43.0 // indirect
+	mvdan.cc/sh/v3 v3.13.1 // indirect
 )
 
 require (

go.sum

@@ -954,6 +954,8 @@ k8s.io/kube-openapi v0.0.0-20260127142750-a19766b6e2d4 h1:HhDfevmPS+OalTjQRKbTHp
 k8s.io/kube-openapi v0.0.0-20260127142750-a19766b6e2d4/go.mod h1:kdmbQkyfwUagLfXIad1y2TdrjPFWp2Q89B3qkRwf/pQ=
 k8s.io/utils v0.0.0-20260210185600-b8788abfbbc2 h1:AZYQSJemyQB5eRxqcPky+/7EdBj0xi3g0ZcxxJ7vbWU=
 k8s.io/utils v0.0.0-20260210185600-b8788abfbbc2/go.mod h1:xDxuJ0whA3d0I4mf/C4ppKHxXynQ+fxnkmQH0vTHnuk=
+mvdan.cc/sh/v3 v3.13.1 h1:DP3TfgZhDkT7lerUdnp6PTGKyxxzz6T+cOlY/xEvfWk=
+mvdan.cc/sh/v3 v3.13.1/go.mod h1:lXJ8SexMvEVcHCoDvAGLZgFJ9Wsm2sulmoNEXGhYZD0=
 pgregory.net/rapid v1.2.0 h1:keKAYRcjm+e1F0oAuU5F5+YPAWcyxNNRK2wud503Gnk=
 pgregory.net/rapid v1.2.0/go.mod h1:PY5XlDGj0+V1FCq0o192FdRhpKHGTRIWBgqjDBTrq04=
 sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 h1:IpInykpT6ceI+QxKBbEflcR5EXP7sU1kvOlxwZh5txg=

worker/file_mapper.go

@@ -479,6 +479,27 @@ func (mapper *FileMapper) ContainerPath(src string) string {
 	return p
 }
 
+// mergeVolume inserts vol into vols using the same RW-deduplication rules as
+// AddVolume: skip if already covered by an existing mount, replace a narrower
+// existing mount with the wider one, otherwise append.
+func mergeVolume(vols []Volume, vol Volume, mapper *FileMapper) []Volume {
+	for i, v := range vols {
+		if vol == v {
+			return vols
+		}
+		if !vol.Readonly && !v.Readonly {
+			if mapper.IsSubpath(vol.ContainerPath, v.ContainerPath) {
+				return vols
+			}
+			if mapper.IsSubpath(v.ContainerPath, vol.ContainerPath) {
+				vols[i] = vol
+				return vols
+			}
+		}
+	}
+	return append(vols, vol)
+}
+
 // consolidateVolumes optimizes container mounts by replacing individual input
 // file mounts with a single read-write ancestor directory mount.
 //
@@ -528,11 +549,12 @@ func (mapper *FileMapper) consolidateVolumes() {
 		// All inputs share a common ancestor — one mount covers them all.
 		// WorkDir+ancestor is valid by the HostPath() construction invariant
 		// (every input host path == WorkDir + container path).
-		nonInputVols = append(nonInputVols, Volume{
+		inputAncestorVol := Volume{
 			HostPath:      filepath.Join(mapper.WorkDir, ancestor),
 			ContainerPath: ancestor,
 			Readonly:      false,
-		})
+		}
+		nonInputVols = mergeVolume(nonInputVols, inputAncestorVol, mapper)
 	} else {
 		// Inputs span disjoint subtrees; fall back to promoting each to its
 		// immediate parent directory to at least eliminate file-level mounts.
@@ -545,11 +567,11 @@ func (mapper *FileMapper) consolidateVolumes() {
 				continue
 			}
 			seen[key] = true
-			nonInputVols = append(nonInputVols, Volume{
+			nonInputVols = mergeVolume(nonInputVols, Volume{
 				HostPath:      hostParent,
 				ContainerPath: contParent,
 				Readonly:      false,
-			})
+			}, mapper)
 		}
 	}
 	mapper.Volumes = nonInputVols

worker/file_mapper_test.go

@@ -187,3 +187,139 @@ func TestMapTask(t *testing.T) {
 		t.Fatal("path unmapping failed")
 	}
 }
+
+// TestMapTaskInputOutputSameDirNoDuplicateVolume reproduces the bug where a
+// task with both an input and an output that resolve to the same container
+// directory produced a duplicate volume mount (invalid in Kubernetes).
+func TestMapTaskInputOutputSameDirNoDuplicateVolume(t *testing.T) {
+	tmp, err := os.MkdirTemp("", "funnel-test-mapper-dedup")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.RemoveAll(tmp)
+
+	f := FileMapper{WorkDir: tmp}
+
+	task := &tes.Task{
+		Inputs: []*tes.Input{
+			{
+				Name:    "in1",
+				Path:    "/data/input.txt",
+				Content: "hello",
+			},
+		},
+		Outputs: []*tes.Output{
+			{
+				Name: "out1",
+				Url:  "file:///out/output.txt",
+				Path: "/data/output.txt",
+				Type: tes.FileType_FILE,
+			},
+		},
+	}
+
+	if err := f.MapTask(task); err != nil {
+		t.Fatal(err)
+	}
+
+	// Verify no two volumes share the same ContainerPath (the K8s duplicate mount error).
+	seen := map[string]int{}
+	for _, v := range f.Volumes {
+		seen[v.ContainerPath]++
+	}
+	for path, count := range seen {
+		if count > 1 {
+			t.Errorf("duplicate ContainerPath %q appears %d times in Volumes (causes K8s invalid mountPath)", path, count)
+		}
+	}
+
+	// /data should appear exactly once and cover both the input and the output.
+	if seen["/data"] != 1 {
+		t.Errorf("expected /data to appear exactly once in Volumes, got %d; volumes: %+v", seen["/data"], f.Volumes)
+	}
+}
+
+// TestMapTaskInputOutputOverlapNoDuplicateVolume tests the variant where the
+// output directory is a parent of the input directory.
+func TestMapTaskInputOutputOverlapNoDuplicateVolume(t *testing.T) {
+	tmp, err := os.MkdirTemp("", "funnel-test-mapper-overlap")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.RemoveAll(tmp)
+
+	f := FileMapper{WorkDir: tmp}
+
+	// Input under /data/inputs/, output dir is /data (parent of input ancestor).
+	task := &tes.Task{
+		Inputs: []*tes.Input{
+			{
+				Name:    "in1",
+				Path:    "/data/inputs/file.txt",
+				Content: "hello",
+			},
+		},
+		Outputs: []*tes.Output{
+			{
+				Name: "out1",
+				Url:  "file:///out/result",
+				Path: "/data/result",
+				Type: tes.FileType_DIRECTORY,
+			},
+		},
+	}
+
+	if err := f.MapTask(task); err != nil {
+		t.Fatal(err)
+	}
+
+	seen := map[string]int{}
+	for _, v := range f.Volumes {
+		seen[v.ContainerPath]++
+	}
+	for path, count := range seen {
+		if count > 1 {
+			t.Errorf("duplicate ContainerPath %q appears %d times in Volumes", path, count)
+		}
+	}
+}
+
+// TestMapTaskMultipleInputsAndOutputs mirrors the concrete failure case from
+// the bug report: one input and one output with a /tmp volume also present.
+func TestMapTaskMultipleInputsAndOutputs(t *testing.T) {
+	tmp, err := os.MkdirTemp("", "funnel-test-mapper-multi")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.RemoveAll(tmp)
+
+	f := FileMapper{WorkDir: tmp}
+
+	task := &tes.Task{
+		Inputs: []*tes.Input{
+			{Name: "a", Path: "/data/a.txt", Content: "a"},
+			{Name: "b", Path: "/data/b.txt", Content: "b"},
+		},
+		Outputs: []*tes.Output{
+			{Name: "out", Url: "file:///out/c.txt", Path: "/data/c.txt", Type: tes.FileType_FILE},
+		},
+	}
+
+	if err := f.MapTask(task); err != nil {
+		t.Fatal(err)
+	}
+
+	seen := map[string]int{}
+	for _, v := range f.Volumes {
+		seen[v.ContainerPath]++
+	}
+	for path, count := range seen {
+		if count > 1 {
+			t.Errorf("duplicate ContainerPath %q appears %d times in Volumes", path, count)
+		}
+	}
+
+	if seen["/data"] != 1 {
+		t.Errorf("expected /data once in Volumes, got %d; volumes: %+v", seen["/data"], f.Volumes)
+	}
+}

worker/kubernetes.go

@@ -19,6 +19,7 @@ import (
 	"k8s.io/client-go/kubernetes/scheme"
 	batchv1 "k8s.io/client-go/kubernetes/typed/batch/v1"
 	"k8s.io/client-go/rest"
+	"mvdan.cc/sh/v3/syntax"
 )
 
 // KubernetesCommand is responsible for configuring and running a task in a Kubernetes cluster.
@@ -112,6 +113,14 @@ func (kcmd KubernetesCommand) Run(ctx context.Context) error {
 		cmd = []string{shellCmd}
 	}
 
+	// Validate single-element shell scripts before passing them to /bin/sh -c.
+	// Multi-element commands are exec'd directly and bypass the shell entirely.
+	if len(cmd) == 1 && !hasRedirects {
+		if err := validateShellSyntax(cmd[0]); err != nil {
+			return err
+		}
+	}
+
 	// Use a shell wrapper when the command is a single element (a shell script
 	// string) or when stdio redirects are present.
 	useShell := len(cmd) == 1 || hasRedirects
@@ -444,3 +453,13 @@ func getKubernetesClientset() (*kubernetes.Clientset, error) {
 	clientset, err := kubernetes.NewForConfig(kubeconfig)
 	return clientset, err
 }
+
+// validateShellSyntax returns an error if s is not valid POSIX shell syntax.
+// Used to catch issues like unterminated quotes before passing single-element
+// commands to /bin/sh -c. Multi-element commands bypass the shell entirely.
+func validateShellSyntax(s string) error {
+	if _, err := syntax.NewParser().Parse(strings.NewReader(s), ""); err != nil {
+		return fmt.Errorf("single-element command is not valid shell syntax (consider using a multi-element command array instead): %w", err)
+	}
+	return nil
+}