From d7e6d64967836d37978ab91dc780578fedb15768 Mon Sep 17 00:00:00 2001 From: Daniel Arroyo Date: Thu, 9 Jul 2026 10:40:12 -0400 Subject: [PATCH] fix: reject + handle remote-to-remote sync pairs Option A (validation): - handlers_syncpairs.go: reject create/update when both SourceMachineID and DestMachineID are set; clear error message explains the constraint - engine.go: detect 'both remote' rsync error at runtime and surface it as error_code=remote_to_remote_unsupported with a human-readable message Option B (remote-to-remote support): - rsync_runner.go: add RunRemote() method that SSHs to the source machine and runs rsync locally there (src=local path, dst=user@host:/path), streaming output back through the onLine callback - engine.go: when both srcMachine and dstMachine are non-nil, use RunRemote() instead of Run(), SSHing to srcMachine and running rsync from there. Also wake dstMachine via WoL when both sides are remote. --- internal/api/handlers_syncpairs.go | 8 ++ internal/syncengine/engine.go | 66 ++++++++++++-- internal/syncengine/rsync_runner.go | 132 ++++++++++++++++++++++++++++ 3 files changed, 200 insertions(+), 6 deletions(-) diff --git a/internal/api/handlers_syncpairs.go b/internal/api/handlers_syncpairs.go index 9d3fbe7..79ae23d 100644 --- a/internal/api/handlers_syncpairs.go +++ b/internal/api/handlers_syncpairs.go @@ -75,6 +75,10 @@ func (h *SyncPairHandler) Create(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, "direction must be push, pull, or mirror") return } + if req.SourceMachineID != nil && req.DestMachineID != nil { + writeError(w, http.StatusBadRequest, "both source and destination cannot be remote machines; one side must be the server (set one MachineID to null)") + return + } if req.RsyncFlags == "" { req.RsyncFlags = "-aP" } @@ -127,6 +131,10 @@ func (h *SyncPairHandler) Update(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, "direction must be push, pull, or mirror") return } + if req.SourceMachineID != nil && req.DestMachineID != nil { + writeError(w, http.StatusBadRequest, "both source and destination cannot be remote machines; one side must be the server (set one MachineID to null)") + return + } repo := models.NewSyncPairRepository(h.db) existing, err := repo.GetByID(id) diff --git a/internal/syncengine/engine.go b/internal/syncengine/engine.go index 57b854d..95b6db8 100644 --- a/internal/syncengine/engine.go +++ b/internal/syncengine/engine.go @@ -7,6 +7,7 @@ import ( "log/slog" "os" "path/filepath" + "strings" "sync" "sync/atomic" "time" @@ -57,6 +58,28 @@ func (e *Engine) SubscribeGlobal() (chan Event, func()) { return e.eventBus.SubscribeGlobal() } +func (e *Engine) wakeMachine(ctx context.Context, m *models.Machine) { + if m == nil || !m.WoLEnabled || m.MACAddress == nil { + return + } + if wol.IsReachable(ctx, m.Host, m.Port, 3*time.Second) { + return + } + mac, err := wol.ParseMAC(*m.MACAddress) + if err != nil { + return + } + bcast := "" + if m.BroadcastAddr != nil { + bcast = *m.BroadcastAddr + } + if err := wol.Send(mac, bcast); err != nil { + slog.Warn("WoL send failed", "host", m.Host, "error", err) + return + } + slog.Info("WoL magic packet sent (remote-to-remote)", "host", m.Host, "mac", *m.MACAddress) +} + func (e *Engine) Run(ctx context.Context, jobID int64, pairID int64) error { if e.queue.IsRunning(pairID) { existingJobID, _ := e.queue.GetJobID(pairID) @@ -107,8 +130,13 @@ func (e *Engine) Run(ctx context.Context, jobID int64, pairID int64) error { RsyncFlags: pair.RsyncFlags, ExcludePatterns: pair.ExcludePatternsList(), } - cfg.Source = src - cfg.Dest = dst + if isRemoteToRemote(srcMachine, dstMachine) { + cfg.Source = pair.SourcePath + cfg.Dest = buildPath(pair.DestPath, dstMachine) + } else { + cfg.Source = src + cfg.Dest = dst + } logPath := filepath.Join(e.cfg.LogsDir(), fmt.Sprintf("%d.log", jobID)) f, err := os.Create(logPath) @@ -120,7 +148,10 @@ func (e *Engine) Run(ctx context.Context, jobID int64, pairID int64) error { var targetMachine *models.Machine var remotePort int - if pair.Direction == "pull" && srcMachine != nil { + if isRemoteToRemote(srcMachine, dstMachine) { + targetMachine = srcMachine + remotePort = srcMachine.Port + } else if pair.Direction == "pull" && srcMachine != nil { targetMachine = srcMachine remotePort = srcMachine.Port } else if dstMachine != nil { @@ -178,6 +209,9 @@ func (e *Engine) Run(ctx context.Context, jobID int64, pairID int64) error { } } + if isRemoteToRemote(srcMachine, dstMachine) && dstMachine != nil && dstMachine.WoLEnabled && dstMachine.MACAddress != nil { + e.wakeMachine(jobCtx, dstMachine) + } e.setJobStatus(jobID, "running") e.setJobLogFile(jobID, logPath) e.emit(Event{Type: "status", JobID: jobID, Key: "status", Value: "running"}) @@ -225,7 +259,17 @@ func (e *Engine) Run(ctx context.Context, jobID int64, pairID int64) error { } } - result, err := runner.Run(jobCtx, cfg, onLine) + var result *RsyncResult + if isRemoteToRemote(srcMachine, dstMachine) { + result, err = runner.RunRemote(jobCtx, cfg, RemoteMachine{ + Host: srcMachine.Host, + Port: srcMachine.Port, + SSHUser: srcMachine.SSHUser, + PrivKey: privKey, + }, onLine) + } else { + result, err = runner.Run(jobCtx, cfg, onLine) + } flush() if err != nil { @@ -248,9 +292,15 @@ func (e *Engine) Run(ctx context.Context, jobID int64, pairID int64) error { } if result.ExitCode != 0 { + errMsg := result.Stderr + errCode := "exit_code" + if strings.Contains(errMsg, "cannot both be remote") { + errCode = "remote_to_remote_unsupported" + errMsg = "sync pair has both source and destination as remote machines; rsync requires one side to be the server (edit the sync pair to set one MachineID to null)" + } e.setJobStatus(jobID, "failed") - e.setJobError(jobID, "exit_code", result.Stderr) - e.emit(Event{Type: "status", JobID: jobID, Key: "status", Value: "failed", Line: result.Stderr}) + e.setJobError(jobID, errCode, errMsg) + e.emit(Event{Type: "status", JobID: jobID, Key: "status", Value: "failed", Line: errMsg}) return fmt.Errorf("rsync exited with code %d: %s", result.ExitCode, result.Stderr) } @@ -326,6 +376,10 @@ func buildPath(path string, machine *models.Machine) string { return fmt.Sprintf("%s@%s:%s", machine.SSHUser, machine.Host, path) } +func isRemoteToRemote(srcMachine, dstMachine *models.Machine) bool { + return srcMachine != nil && dstMachine != nil +} + func (e *Engine) CreateJob(syncPairID int64, triggerType string) (int64, error) { jobRepo := models.NewJobRepository(e.db) id, err := jobRepo.Create(syncPairID, triggerType, "queued") diff --git a/internal/syncengine/rsync_runner.go b/internal/syncengine/rsync_runner.go index 89d797f..17f313a 100644 --- a/internal/syncengine/rsync_runner.go +++ b/internal/syncengine/rsync_runner.go @@ -155,3 +155,135 @@ func (r *RsyncRunner) buildArgs(pair *SyncPairConfig) []string { return args } + +func (r *RsyncRunner) buildRemoteArgs(pair *SyncPairConfig) []string { + var args []string + + flags := strings.Fields(pair.RsyncFlags) + args = append(args, flags...) + + for _, pattern := range pair.ExcludePatterns { + args = append(args, "--exclude="+pattern) + } + + if pair.Direction == "mirror" { + args = append(args, "--delete") + } + + if pair.Direction == "pull" { + args = append(args, pair.Dest, pair.Source) + } else { + args = append(args, pair.Source, pair.Dest) + } + + return args +} + +type RemoteMachine struct { + Host string + Port int + SSHUser string + PrivKey string +} + +func (r *RsyncRunner) RunRemote(ctx context.Context, pair *SyncPairConfig, remote RemoteMachine, onLine func(stream string, line string)) (*RsyncResult, error) { + args := r.buildRemoteArgs(pair) + + sshArgs := []string{ + "ssh", + "-i", remote.PrivKey, + "-o", "StrictHostKeyChecking=accept-new", + "-o", "UserKnownHostsFile=" + strings.TrimRight(r.sshDir, "/") + "/known_hosts", + "-tt", + "-p", fmt.Sprintf("%d", remote.Port), + fmt.Sprintf("%s@%s", remote.SSHUser, remote.Host), + "rsync", + } + sshArgs = append(sshArgs, args...) + + cmd := exec.CommandContext(ctx, "ssh", sshArgs...) + + stdout, err := cmd.StdoutPipe() + if err != nil { + return nil, fmt.Errorf("stdout pipe: %w", err) + } + stderr, err := cmd.StderrPipe() + if err != nil { + return nil, fmt.Errorf("stderr pipe: %w", err) + } + + if err := cmd.Start(); err != nil { + return nil, fmt.Errorf("starting ssh: %w", err) + } + + var outLines, errLines []string + done := make(chan struct{}) + go func() { + br := io.Reader(stdout) + buf := make([]byte, 4096) + for { + n, err := br.Read(buf) + if n > 0 { + line := strings.TrimRight(string(buf[:n]), "\r\n") + if line != "" { + outLines = append(outLines, line) + if onLine != nil { + onLine("stdout", line) + } + } + } + if err != nil { + break + } + } + select { + case <-done: + default: + close(done) + } + }() + + go func() { + br := io.Reader(stderr) + buf := make([]byte, 4096) + for { + n, err := br.Read(buf) + if n > 0 { + line := strings.TrimRight(string(buf[:n]), "\r\n") + if line != "" { + errLines = append(errLines, line) + if onLine != nil { + onLine("stderr", line) + } + } + } + if err != nil { + break + } + } + select { + case <-done: + default: + close(done) + } + }() + + err = cmd.Wait() + <-done + + result := &RsyncResult{ + ExitCode: 0, + Stdout: strings.Join(outLines, "\n"), + Stderr: strings.Join(errLines, "\n"), + Stats: ParseFinalStats(strings.Join(outLines, "\n")), + } + if err != nil { + if exitErr, ok := err.(*exec.ExitError); ok { + result.ExitCode = exitErr.ExitCode() + } else { + result.ExitCode = -1 + } + } + + return result, nil +}