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.
This commit is contained in:
2026-07-09 10:40:12 -04:00
parent ba4c6aa732
commit d7e6d64967
3 changed files with 200 additions and 6 deletions
+60 -6
View File
@@ -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")