diff --git a/Makefile b/Makefile index cba36f5..efbc20f 100644 --- a/Makefile +++ b/Makefile @@ -1,5 +1,5 @@ BINARY=syncserver -VERSION?=1.0.9 +VERSION?=1.0.10 GO?=go LDFLAGS=-s -w -X main.version=$(VERSION) -X main.commit=$(shell git rev-parse --short HEAD 2>/dev/null || echo unknown) BUILD_FLAGS=CGO_ENABLED=0 diff --git a/cmd/server/main.go b/cmd/server/main.go index 4ed1b01..e692697 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -20,7 +20,7 @@ import ( "github.com/syncserver/internal/syncengine" ) -var version = "1.0.9" +var version = "1.0.10" func main() { cfgPath := flag.String("config", "", "Path to config.yaml") diff --git a/internal/syncengine/engine.go b/internal/syncengine/engine.go index 534e674..45c3a2a 100644 --- a/internal/syncengine/engine.go +++ b/internal/syncengine/engine.go @@ -26,12 +26,13 @@ type Engine struct { } type Event struct { - Type string - JobID int64 - Key string - Value string - Line string - Stream string + Type string + JobID int64 + MachineID int64 + Key string + Value string + Line string + Stream string } func New(database interface{ SQLDB() *sql.DB }, cfg *config.Config) *Engine { @@ -116,9 +117,6 @@ func (e *Engine) Run(ctx context.Context, jobID int64, pairID int64) error { f.Close() } - e.setJobStatus(jobID, "waking_up") - e.emit(Event{Type: "status", JobID: jobID, Key: "status", Value: "waking_up"}) - var targetMachine *models.Machine var remotePort int if pair.Direction == "pull" && srcMachine != nil { @@ -129,32 +127,52 @@ func (e *Engine) Run(ctx context.Context, jobID int64, pairID int64) error { remotePort = dstMachine.Port } - if targetMachine != nil && targetMachine.WoLEnabled && targetMachine.MACAddress != nil { - mac, err := wol.ParseMAC(*targetMachine.MACAddress) - if err == nil { - bcast := "" - if targetMachine.BroadcastAddr != nil { - bcast = *targetMachine.BroadcastAddr - } - wolErr := wol.Send(mac, bcast) - if wolErr != nil { - slog.Warn("WoL send failed", "host", targetMachine.Host, "error", wolErr) - } else { - slog.Info("WoL magic packet sent", "host", targetMachine.Host, "mac", *targetMachine.MACAddress) - } + wolShouldRun := targetMachine != nil && + targetMachine.WoLEnabled && + targetMachine.MACAddress != nil - timeout := time.Duration(targetMachine.WakeTimeoutSeconds) * time.Second - interval := time.Duration(targetMachine.WakeCheckIntervalSeconds) * time.Second - if err := wol.WaitUntilReady(jobCtx, targetMachine.Host, remotePort, timeout, interval); err != nil { - e.setJobStatus(jobID, "failed") - if wolErr != nil { - e.setJobError(jobID, "wol_send_failed", wolErr.Error()) - e.emit(Event{Type: "status", JobID: jobID, Key: "status", Value: "failed", Line: fmt.Sprintf("WoL send failed: %v", wolErr)}) - return fmt.Errorf("WoL send failed: %w", wolErr) + if wolShouldRun { + if wol.IsReachable(jobCtx, targetMachine.Host, remotePort, 3*time.Second) { + slog.Info("machine already reachable, skipping WoL", "host", targetMachine.Host) + if targetMachine.ID > 0 { + e.setMachineStatus(targetMachine.ID, "online") + } + } else { + e.setJobStatus(jobID, "waking_up") + e.emit(Event{Type: "status", JobID: jobID, Key: "status", Value: "waking_up"}) + + mac, err := wol.ParseMAC(*targetMachine.MACAddress) + if err == nil { + bcast := "" + if targetMachine.BroadcastAddr != nil { + bcast = *targetMachine.BroadcastAddr + } + wolErr := wol.Send(mac, bcast) + if wolErr != nil { + slog.Warn("WoL send failed", "host", targetMachine.Host, "error", wolErr) + } else { + slog.Info("WoL magic packet sent", "host", targetMachine.Host, "mac", *targetMachine.MACAddress) + } + + timeout := time.Duration(targetMachine.WakeTimeoutSeconds) * time.Second + interval := time.Duration(targetMachine.WakeCheckIntervalSeconds) * time.Second + if err := wol.WaitUntilReady(jobCtx, targetMachine.Host, remotePort, timeout, interval); err != nil { + e.setJobStatus(jobID, "failed") + if wolErr != nil { + e.setJobError(jobID, "wol_send_failed", wolErr.Error()) + e.emit(Event{Type: "status", JobID: jobID, Key: "status", Value: "failed", Line: fmt.Sprintf("WoL send failed: %v", wolErr)}) + return fmt.Errorf("WoL send failed: %w", wolErr) + } + if targetMachine.ID > 0 { + e.setMachineStatus(targetMachine.ID, "offline") + } + e.setJobError(jobID, "wol_timeout", err.Error()) + e.emit(Event{Type: "status", JobID: jobID, Key: "status", Value: "failed", Line: err.Error()}) + return fmt.Errorf("machine not ready: %w", err) + } + if targetMachine.ID > 0 { + e.setMachineStatus(targetMachine.ID, "online") } - e.setJobError(jobID, "wol_timeout", err.Error()) - e.emit(Event{Type: "status", JobID: jobID, Key: "status", Value: "failed", Line: err.Error()}) - return fmt.Errorf("machine not ready: %w", err) } } } @@ -281,6 +299,21 @@ func (e *Engine) setJobError(jobID int64, code, message string) { jobRepo.SetError(jobID, code, message) } +func (e *Engine) setMachineStatus(machineID int64, status string) { + machineRepo := models.NewMachineRepository(e.db) + if err := machineRepo.UpdateStatus(machineID, status); err != nil { + slog.Warn("failed to update machine status", + "machine_id", machineID, "status", status, "error", err) + return + } + e.emit(Event{ + Type: "machine_status", + MachineID: machineID, + Key: "status", + Value: status, + }) +} + func (e *Engine) emit(evt Event) { e.eventBus.Publish(evt) } diff --git a/internal/syncengine/engine_status_test.go b/internal/syncengine/engine_status_test.go new file mode 100644 index 0000000..a4f8037 --- /dev/null +++ b/internal/syncengine/engine_status_test.go @@ -0,0 +1,39 @@ +package syncengine + +import ( + "context" + "net" + "testing" + "time" + + "github.com/syncserver/internal/wol" +) + +func TestIsReachable(t *testing.T) { + ln, err := net.ListenTCP("tcp4", &net.TCPAddr{IP: net.ParseIP("127.0.0.1"), Port: 0}) + if err != nil { + t.Skipf("skipping test (no TCP listener available): %v", err) + } + defer ln.Close() + addr := ln.Addr().(*net.TCPAddr) + + ctx := context.Background() + + if !wol.IsReachable(ctx, "127.0.0.1", addr.Port, 500*time.Millisecond) { + t.Error("expected reachable on listening port") + } + if wol.IsReachable(ctx, "127.0.0.1", addr.Port+1, 500*time.Millisecond) { + t.Error("expected unreachable on non-listening port") + } +} + +func TestIsReachableTimeout(t *testing.T) { + ctx := context.Background() + start := time.Now() + if wol.IsReachable(ctx, "192.0.2.1", 12345, 500*time.Millisecond) { + t.Error("expected unreachable for non-routable IP") + } + if d := time.Since(start); d < 400*time.Millisecond { + t.Errorf("IsReachable returned too early: %v", d) + } +} diff --git a/internal/wol/waiter.go b/internal/wol/waiter.go index affa7df..3f9ff3b 100644 --- a/internal/wol/waiter.go +++ b/internal/wol/waiter.go @@ -45,3 +45,16 @@ func WaitUntilReady(ctx context.Context, host string, sshPort int, maxWait, chec } } } + +func IsReachable(ctx context.Context, host string, port int, timeout time.Duration) bool { + ctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + addr := fmt.Sprintf("%s:%d", host, port) + dialer := net.Dialer{Timeout: timeout} + conn, err := dialer.DialContext(ctx, "tcp", addr) + if err != nil { + return false + } + conn.Close() + return true +} diff --git a/web/src/lib/sse.ts b/web/src/lib/sse.ts new file mode 100644 index 0000000..9704c6c --- /dev/null +++ b/web/src/lib/sse.ts @@ -0,0 +1,22 @@ +export interface MachineStatusEvent { + type: 'machine_status'; + machine_id: number; + status: string; + key?: string; + value?: string; +} + +export function subscribeMachineStatus( + onUpdate: (evt: MachineStatusEvent) => void +): () => void { + const es = new EventSource('/api/jobs/stream'); + es.addEventListener('machine_status', (ev) => { + try { + const data = JSON.parse((ev as MessageEvent).data); + onUpdate(data as MachineStatusEvent); + } catch { + // ignore parse errors + } + }); + return () => es.close(); +} diff --git a/web/src/pages/Dashboard.tsx b/web/src/pages/Dashboard.tsx index 71476f4..50efecd 100644 --- a/web/src/pages/Dashboard.tsx +++ b/web/src/pages/Dashboard.tsx @@ -11,6 +11,7 @@ import { EmptyState } from '@/components/ui/EmptyState'; import { Skeleton } from '@/components/ui/Skeleton'; import { statusVariant, statusLabel } from '@/lib/status'; import { formatRelativeTime } from '@/lib/utils'; +import { subscribeMachineStatus } from '@/lib/sse'; export default function Dashboard() { const [machines, setMachines] = useState([]); @@ -33,6 +34,17 @@ export default function Dashboard() { .finally(() => setLoading(false)); }, []); + useEffect(() => { + const unsub = subscribeMachineStatus((evt) => { + setMachines((prev) => + prev.map((m) => + m.id === evt.machine_id ? { ...m, status: evt.status } : m + ) + ); + }); + return unsub; + }, []); + const pairName = (id: number) => pairs.find(p => p.id === id)?.name ?? `Pair ${id}`; const online = machines.filter(m => m.status.startsWith('online')).length; diff --git a/web/src/pages/Machines.tsx b/web/src/pages/Machines.tsx index b38aff2..d534e53 100644 --- a/web/src/pages/Machines.tsx +++ b/web/src/pages/Machines.tsx @@ -29,6 +29,7 @@ import { Card } from '@/components/ui/Card'; import { Pencil, Trash2, Plus, Server, Zap } from 'lucide-react'; import { toast } from 'sonner'; import { cn } from '@/lib/utils'; +import { subscribeMachineStatus } from '@/lib/sse'; type MachineForm = { id: number | undefined; @@ -70,6 +71,17 @@ export default function Machines() { load(); }, []); + useEffect(() => { + const unsub = subscribeMachineStatus((evt) => { + setMachines((prev) => + prev.map((m) => + m.id === evt.machine_id ? { ...m, status: evt.status } : m + ) + ); + }); + return unsub; + }, []); + async function load() { try { const [ms, ks] = await Promise.all([