Treat missing watched sources as deletes

Author: seslly

internal/agent/replication.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
+	"net/url"
 	"os"
 	"os/exec"
 	"path"
@@ -72,6 +73,14 @@ func (ExecRunner) Run(ctx context.Context, spec CommandSpec) error {
 	return cmd.Run()
 }
 
+type sourceExistsResult int
+
+const (
+	sourceExistsUnknown sourceExistsResult = iota
+	sourceExistsYes
+	sourceExistsNo
+)
+
 func NormalizeEventPath(p string) (string, error) {
 	p = strings.TrimSpace(strings.ReplaceAll(p, "\\", "/"))
 	p = strings.TrimPrefix(p, "/")
@@ -241,18 +250,72 @@ func ApplyEventBatch(ctx context.Context, runner Runner, pool Pool, batch EventB
 		default:
 			sourcePath := path.Join(safeSegment(batch.Namespace), safeSegment(batch.Volume), event.Path)
 			targetPath := filepath.Join(targetRoot, filepath.FromSlash(event.Path))
+			exists, err := sourcePathExists(ctx, batch.Source, sourcePath)
+			if err != nil {
+				return err
+			}
+			if exists == sourceExistsNo {
+				if err := os.RemoveAll(targetPath); err != nil {
+					return err
+				}
+				continue
+			}
 			spec, err := BuildRcloneCopyToCommand(batch.Source, sourcePath, targetPath)
 			if err != nil {
 				return err
 			}
 			if err := runner.Run(ctx, spec); err != nil {
+				exists, existsErr := sourcePathExists(ctx, batch.Source, sourcePath)
+				if existsErr != nil {
+					return err
+				}
+				if exists == sourceExistsNo {
+					if removeErr := os.RemoveAll(targetPath); removeErr != nil {
+						return removeErr
+					}
+					continue
+				}
 				return err
 			}
 		}
 	}
 	return nil
 }
 
+func sourcePathExists(ctx context.Context, source SourceRef, sourcePath string) (sourceExistsResult, error) {
+	if strings.TrimSpace(source.WebDAVURL) == "" {
+		return sourceExistsUnknown, nil
+	}
+	base, err := url.Parse(strings.TrimRight(source.WebDAVURL, "/") + "/")
+	if err != nil {
+		return sourceExistsUnknown, err
+	}
+	rel, err := url.Parse(path.Join(strings.Trim(sourcePath, "/")))
+	if err != nil {
+		return sourceExistsUnknown, err
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodHead, base.ResolveReference(rel).String(), nil)
+	if err != nil {
+		return sourceExistsUnknown, err
+	}
+	if source.User != "" || source.Password != "" {
+		req.SetBasicAuth(source.User, source.Password)
+	}
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return sourceExistsUnknown, nil
+	}
+	defer resp.Body.Close()
+	switch resp.StatusCode {
+	case http.StatusOK, http.StatusNoContent, http.StatusPartialContent:
+		return sourceExistsYes, nil
+	case http.StatusNotFound, http.StatusGone:
+		return sourceExistsNo, nil
+	default:
+		return sourceExistsUnknown, nil
+	}
+}
+
 func ApplyFullSync(ctx context.Context, runner Runner, pool Pool, req FullSyncRequest) error {
 	if runner == nil {
 		runner = ExecRunner{}

internal/agent/replication_test.go

@@ -2,6 +2,9 @@ package agent
 
 import (
 	"context"
+	"errors"
+	"net/http"
+	"net/http/httptest"
 	"os"
 	"path/filepath"
 	"strings"
@@ -10,11 +13,12 @@ import (
 
 type recordingRunner struct {
 	specs []CommandSpec
+	err   error
 }
 
 func (r *recordingRunner) Run(_ context.Context, spec CommandSpec) error {
 	r.specs = append(r.specs, spec)
-	return nil
+	return r.err
 }
 
 func containsArg(args []string, want string) bool {
@@ -179,6 +183,76 @@ func TestApplyEventBatchRunsCopyAndDelete(t *testing.T) {
 	}
 }
 
+func TestApplyEventBatchRemovesTargetWhenSourceAlreadyGone(t *testing.T) {
+	dir := t.TempDir()
+	pool := Pool{Name: "default", Path: dir}
+	if err := EnsurePool(pool, false); err != nil {
+		t.Fatal(err)
+	}
+	target := filepath.Join(dir, "default", "demo", "save", "a.txt")
+	if err := writeFile(target, "old"); err != nil {
+		t.Fatal(err)
+	}
+	source := httptest.NewServer(http.NotFoundHandler())
+	defer source.Close()
+
+	runner := &recordingRunner{}
+	err := ApplyEventBatch(context.Background(), runner, pool, EventBatch{
+		Namespace: "default",
+		Volume:    "demo",
+		Source:    SourceRef{WebDAVURL: source.URL},
+		Events:    []FileEvent{{Path: "save/a.txt", Op: EventOpUpsert}},
+	})
+	if err != nil {
+		t.Fatalf("ApplyEventBatch returned error: %v", err)
+	}
+	if len(runner.specs) != 0 {
+		t.Fatalf("ran %d commands, want none", len(runner.specs))
+	}
+	if _, err := os.Stat(target); !os.IsNotExist(err) {
+		t.Fatalf("target was not removed after missing source: %v", err)
+	}
+}
+
+func TestApplyEventBatchRemovesTargetWhenSourceDisappearsAfterCopyFailure(t *testing.T) {
+	dir := t.TempDir()
+	pool := Pool{Name: "default", Path: dir}
+	if err := EnsurePool(pool, false); err != nil {
+		t.Fatal(err)
+	}
+	target := filepath.Join(dir, "default", "demo", "save", "a.txt")
+	if err := writeFile(target, "old"); err != nil {
+		t.Fatal(err)
+	}
+	var checks int
+	source := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		checks++
+		if checks == 1 {
+			w.WriteHeader(http.StatusOK)
+			return
+		}
+		http.NotFound(w, r)
+	}))
+	defer source.Close()
+
+	runner := &recordingRunner{err: errors.New("copy failed")}
+	err := ApplyEventBatch(context.Background(), runner, pool, EventBatch{
+		Namespace: "default",
+		Volume:    "demo",
+		Source:    SourceRef{WebDAVURL: source.URL},
+		Events:    []FileEvent{{Path: "save/a.txt", Op: EventOpUpsert}},
+	})
+	if err != nil {
+		t.Fatalf("ApplyEventBatch returned error: %v", err)
+	}
+	if len(runner.specs) != 1 {
+		t.Fatalf("ran %d commands, want 1", len(runner.specs))
+	}
+	if _, err := os.Stat(target); !os.IsNotExist(err) {
+		t.Fatalf("target was not removed after disappeared source: %v", err)
+	}
+}
+
 func TestApplyFullSyncRunsScopedRcloneSync(t *testing.T) {
 	dir := t.TempDir()
 	pool := Pool{Name: "default", Path: dir}