fix: ended page scopes start no timers; SSH transfers keep their connection from the idle sweep; linked files download whole; only the sent item unpacks on an SSH host, replace and skip as on a machine

internal/agent/sshfs.go

@@ -39,6 +39,28 @@ type sshfsConn struct {
 	statFmt string          // "gnu", "bsd" or "ls"
 	home    string
 	last    time.Time
+	busy    int // transfers under way: the sweep leaves the connection be
+}
+
+// hold marks the connection busy for a long transfer; the returned func ends
+// it and starts the idle clock again.
+func (c *sshfsConn) hold() func() {
+	c.mu.Lock()
+	c.busy++
+	c.mu.Unlock()
+	return func() {
+		c.mu.Lock()
+		c.busy--
+		c.last = time.Now()
+		c.mu.Unlock()
+	}
+}
+
+// idle: nothing under way and unused for sshfsIdle by now.
+func (c *sshfsConn) idle(now time.Time) bool {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.busy == 0 && now.Sub(c.last) > sshfsIdle
 }
 
 var (
@@ -151,11 +173,7 @@ func hasLine(lines []string, w string) bool {
 func sshfsSweep() {
 	for range time.Tick(time.Minute) {
 		sshfsConns.Range(func(k, v any) bool {
-			c := v.(*sshfsConn)
-			c.mu.Lock()
-			idle := time.Since(c.last) > sshfsIdle
-			c.mu.Unlock()
-			if idle {
+			if v.(*sshfsConn).idle(time.Now()) {
 				sshfsClose(k.(string))
 			}
 			return true
@@ -410,14 +428,20 @@ func writeCommand(p string) string {
 	return "cat > " + shq(tmp) + " && mv -f -- " + shq(tmp) + " " + shq(p) + " || { rm -f -- " + shq(tmp) + "; exit 1; }"
 }
 
-func (a *Agent) sshfsSize(c *sshfsConn, p string) (int64, error) {
-	cmd := "stat -c %s -- " + shq(p)
-	if c.statFmt == "bsd" {
-		cmd = "stat -f %z -- " + shq(p)
-	} else if c.statFmt == "ls" {
-		cmd = "wc -c < " + shq(p)
+// sizeCommand tells the size of p - of its target, if a link: what cat
+// reads.
+func sizeCommand(statFmt, p string) string {
+	switch statFmt {
+	case "bsd":
+		return "stat -L -f %z -- " + shq(p)
+	case "ls":
+		return "wc -c < " + shq(p)
 	}
-	out, err := a.sshfsRun(c, cmd, nil)
+	return "stat -L -c %s -- " + shq(p)
+}
+
+func (a *Agent) sshfsSize(c *sshfsConn, p string) (int64, error) {
+	out, err := a.sshfsRun(c, sizeCommand(c.statFmt, p), nil)
 	if err != nil {
 		return 0, err
 	}
@@ -483,6 +507,7 @@ func (a *Agent) doSSHFSSend(ctx context.Context, env proto.TaskEnvelope) proto.T
 	if err != nil {
 		return proto.TaskResult{Error: err.Error()}
 	}
+	defer c.hold()()
 	size, err := a.sshfsSize(c, args.Path)
 	if err != nil {
 		return proto.TaskResult{Error: err.Error()}
@@ -518,7 +543,8 @@ func (a *Agent) doSSHFSSend(ctx context.Context, env proto.TaskEnvelope) proto.T
 			return proto.TaskResult{Error: "передача прервана: " + err.Error()}
 		}
 	}
-	_ = s.Wait()
+	// All of it is read; the session closes (deferred) without waiting on a
+	// cat that may have more to say.
 	data, _ := json.Marshal(map[string]any{"path": args.Path, "size": size})
 	return proto.TaskResult{OK: true, Data: data}
 }
@@ -534,6 +560,7 @@ func (a *Agent) doSSHFSPut(ctx context.Context, env proto.TaskEnvelope) proto.Ta
 	if err != nil {
 		return proto.TaskResult{Error: err.Error()}
 	}
+	defer c.hold()()
 	s, err := a.sshfsSession(c)
 	if err != nil {
 		return proto.TaskResult{Error: err.Error()}
@@ -598,7 +625,8 @@ func renameTar(w io.Writer, r io.Reader, base, name string) error {
 		if i := strings.IndexByte(n, '/'); i >= 0 {
 			top, rest = n[:i], n[i:]
 		}
-		if top != base {
+		// Only base and what is under it - nothing that climbs out by "..".
+		if top != base || strings.Contains("/"+rest+"/", "/../") {
 			continue
 		}
 		h.Name = name + rest
@@ -615,6 +643,7 @@ func renameTar(w io.Writer, r io.Reader, base, name string) error {
 // packRemote writes src on the connection's host as a tar stream named name,
 // counting what goes for the progress.
 func (a *Agent) packRemote(ctx context.Context, w io.Writer, c *sshfsConn, src, name string, m *shipMeter) error {
+	defer c.hold()()
 	s, err := a.sshfsSession(c)
 	if err != nil {
 		return err
@@ -657,25 +686,34 @@ func (a *Agent) remoteSize(c *sshfsConn, src string) int64 {
 // unpackRemote puts a tar stream into dir on the connection's host, by the
 // conflict rule: tar keeps or replaces what is there, an "error" rule
 // refuses when name is there already.
-func (a *Agent) unpackRemote(c *sshfsConn, r io.Reader, dir, name, conflict string) error {
+func (a *Agent) unpackRemote(c *sshfsConn, r io.Reader, dir, name, conflict string) (skipped bool, err error) {
+	defer c.hold()()
 	target := strings.TrimSuffix(dir, "/") + "/" + name
 	out, _ := a.sshfsRun(c, "test -e "+shq(target)+" -o -L "+shq(target)+" && echo LPRA_EXISTS; tar --version 2>/dev/null | head -n 1", nil)
-	if strings.Contains(string(out), "LPRA_EXISTS") && conflict != "replace" && conflict != "skip" {
-		return errors.New(target + " уже существует")
-	}
-	gnu := strings.Contains(string(out), "GNU tar")
-	flags := ""
-	switch conflict {
-	case "skip":
-		flags = " -k"
-		if gnu {
-			flags = " --skip-old-files"
-		}
-	case "replace":
-		if gnu {
-			flags = " --overwrite"
+	if strings.Contains(string(out), "LPRA_EXISTS") {
+		switch conflict {
+		case "skip":
+			// Left as it was, as on a machine.
+			return true, nil
+		case "replace":
+		default:
+			return false, errors.New(target + " уже существует")
 		}
 	}
-	_, err := a.sshfsRun(c, "mkdir -p -- "+shq(dir)+" && cd -- "+shq(dir)+" && tar -xf -"+flags, r)
-	return err
+	// Only name and what is under it reaches the host's tar.
+	pr, pw := io.Pipe()
+	go func() { pw.CloseWithError(renameTar(pw, r, name, name)) }()
+	defer pr.Close()
+	_, err = a.sshfsRun(c, unpackCommand(dir, name, conflict, strings.Contains(string(out), "GNU tar")), pr)
+	return false, err
+}
+
+// unpackCommand unpacks a tar stream of name into dir; "replace" removes
+// what is there first, so the thing is replaced whole as on a machine.
+func unpackCommand(dir, name, conflict string, gnu bool) string {
+	cmd := "mkdir -p -- " + shq(dir) + " && cd -- " + shq(dir) + " && tar -xf -"
+	if conflict == "replace" {
+		cmd = "rm -rf -- " + shq(strings.TrimSuffix(dir, "/")+"/"+name) + " && " + cmd
+	}
+	return cmd
 }