diff --git a/backend/cronjobs/cronjobs.go b/backend/cronjobs/cronjobs.go index 60e27b18..796a9362 100644 --- a/backend/cronjobs/cronjobs.go +++ b/backend/cronjobs/cronjobs.go @@ -1,7 +1,6 @@ package cronjobs import ( - "github.com/pocketbase/pocketbase" "github.com/pocketbase/pocketbase/core" "github.com/robfig/cron/v3" "github.com/seriousm4x/upsnap/iptracking" @@ -56,11 +55,14 @@ func SetPingJobs(app core.App) { for _, device := range devices { // ping device go func(d *core.Record) { + // only write status changes so concurrent writers to other + // fields (e.g. ip tracking) are never clobbered + d.IgnoreUnchangedFields(true) status := d.GetString("status") if status == "pending" { return } - isUp, err := networking.PingDevice(d) + isUp, err := networking.PingDevice(d, nil) if err != nil { logger.Error.Println(err) } @@ -119,7 +121,7 @@ func SetPingJobs(app core.App) { } } -func SetWakeShutdownJobs(app *pocketbase.PocketBase) { +func SetWakeShutdownJobs(app core.App) { // remove existing jobs for _, job := range CronWakeShutdown.Entries() { CronWakeShutdown.Remove(job.ID) @@ -143,10 +145,11 @@ func SetWakeShutdownJobs(app *pocketbase.PocketBase) { logger.Error.Println(err) return } + d.IgnoreUnchangedFields(true) if d.GetString("status") == "pending" { return } - isOnline, err := networking.PingDevice(d) + isOnline, err := networking.PingDevice(d, nil) if err != nil { logger.Error.Println(err) return @@ -159,7 +162,13 @@ func SetWakeShutdownJobs(app *pocketbase.PocketBase) { logger.Error.Println("Failed to save record:", err) return } - if err := networking.WakeDevice(d); err != nil { + // refresh the save baseline so a revert to the load-time + // status below isn't dropped as unchanged + if err := d.PostScan(); err != nil { + logger.Error.Println(err) + } + iptracking.TrackDeviceAfterWake(app, d) + if err := networking.WakeDevice(d, networking.DeviceIPFunc(app, d)); err != nil { logger.Error.Println(err) d.Set("status", "offline") } else { @@ -181,10 +190,20 @@ func SetWakeShutdownJobs(app *pocketbase.PocketBase) { logger.Error.Println(err) return } + d.IgnoreUnchangedFields(true) + if d.GetString("status") == "pending" { + return + } + // refresh a tracked ip first so the online check and the + // shutdown command use the device's current address + d = iptracking.TrackDevice(app, d) + d.IgnoreUnchangedFields(true) + // the scan can take a while: bail if a wake was initiated + // in the meantime if d.GetString("status") == "pending" { return } - isOnline, err := networking.PingDevice(d) + isOnline, err := networking.PingDevice(d, nil) if err != nil { logger.Error.Println(err) return @@ -192,13 +211,11 @@ func SetWakeShutdownJobs(app *pocketbase.PocketBase) { if !isOnline { return } - status := d.GetString("status") - if status != "online" { - return - } d.Set("status", "pending") if err := app.Save(d); err != nil { logger.Error.Println("Failed to save record:", err) + } else if err := d.PostScan(); err != nil { + logger.Error.Println(err) } if err := networking.ShutdownDevice(d); err != nil { logger.Error.Println(err) diff --git a/backend/cronjobs/cronjobs_test.go b/backend/cronjobs/cronjobs_test.go new file mode 100644 index 00000000..e67478f6 --- /dev/null +++ b/backend/cronjobs/cronjobs_test.go @@ -0,0 +1,59 @@ +package cronjobs + +import ( + "testing" + + "github.com/pocketbase/pocketbase/core" + "github.com/pocketbase/pocketbase/tests" + + // register the app migrations so test apps get the real schema + _ "github.com/seriousm4x/upsnap/migrations" +) + +// A failed scheduled wake must persist the revert to the pre-wake status. +// With IgnoreUnchangedFields set, that only works because the save baseline +// is refreshed after the intermediate pending write — otherwise the revert +// equals the load-time status and is dropped from the update, leaving the +// device wedged at "pending" and skipped by every cron from then on. +func TestWakeCronPersistsRevertedStatus(t *testing.T) { + app, err := tests.NewTestApp(t.TempDir()) + if err != nil { + t.Fatalf("Got unexpected error: %v", err) + } + t.Cleanup(app.Cleanup) + + collection, err := app.FindCollectionByNameOrId("devices") + if err != nil { + t.Fatalf("Got unexpected error: %v", err) + } + device := core.NewRecord(collection) + device.Set("name", "wake-fails") + device.Set("ip", "127.0.0.1") + device.Set("netmask", "255.255.255.0") + device.Set("mac", "AA:BB:CC:DD:0A:01") + device.Set("status", "offline") + device.Set("wake_cron", "0 0 * * *") + device.Set("wake_cron_enabled", true) + // the wake command fails immediately, the ping command reports offline + device.Set("wake_cmd", "exit 1") + device.Set("wake_timeout", 1) + device.Set("ping_cmd", "exit 1") + if err := app.Save(device); err != nil { + t.Fatalf("Got unexpected error: %v", err) + } + + SetWakeShutdownJobs(app) + entries := CronWakeShutdown.Entries() + if len(entries) != 1 { + t.Fatalf("Expected one wake cron entry, got %d", len(entries)) + } + entries[0].Job.Run() + + fresh, err := app.FindRecordById("devices", device.Id) + if err != nil { + t.Fatalf("Got unexpected error: %v", err) + } + if status := fresh.GetString("status"); status != "offline" { + t.Errorf("Status mismatch: expected offline, got %s", status) + } +} diff --git a/backend/go.mod b/backend/go.mod index e8ec24c8..a40b42d0 100644 --- a/backend/go.mod +++ b/backend/go.mod @@ -8,6 +8,7 @@ require ( github.com/pocketbase/pocketbase v0.39.9 github.com/prometheus-community/pro-bing v0.9.1 github.com/robfig/cron/v3 v3.0.1 + golang.org/x/sync v0.22.0 golang.org/x/sys v0.47.0 kernel.org/pub/linux/libs/security/libcap/cap v1.2.78 ) @@ -42,7 +43,6 @@ require ( golang.org/x/image v0.44.0 // indirect golang.org/x/net v0.57.0 // indirect golang.org/x/oauth2 v0.36.0 // indirect - golang.org/x/sync v0.22.0 // indirect golang.org/x/text v0.40.0 // indirect kernel.org/pub/linux/libs/security/libcap/psx v1.2.78 // indirect modernc.org/libc v1.74.1 // indirect diff --git a/backend/iptracking/iptracking.go b/backend/iptracking/iptracking.go index ff5e9f3b..aa660485 100644 --- a/backend/iptracking/iptracking.go +++ b/backend/iptracking/iptracking.go @@ -5,10 +5,12 @@ package iptracking import ( "net" "sync/atomic" + "time" "github.com/pocketbase/pocketbase/core" "github.com/seriousm4x/upsnap/logger" "github.com/seriousm4x/upsnap/networking" + "golang.org/x/sync/singleflight" ) var sweepRunning atomic.Bool @@ -25,6 +27,13 @@ func init() { // nmapScan is a seam for tests to stub out the privileged nmap invocation var nmapScan = networking.NmapScan +// scanGroup joins concurrent scans of the same subnet into one nmap run +var scanGroup singleflight.Group + +// wakeScanDelay gives a woken device time to boot and renew its dhcp lease +// before its subnet is scanned; a variable so tests can shorten it +var wakeScanDelay = 15 * time.Second + // TrackAllSubnets scans the local subnets of devices with ip tracking enabled // and updates their ip address if their mac address is found at a different // one. Non-scannable subnets are silently skipped, but their devices can @@ -85,6 +94,53 @@ func CatchUpSweep(app core.App) { TrackAllSubnets(app) } +// TrackDevice scans the device's subnet and returns the device re-read from +// the database, so an action about to use its ip address sees the address +// the scan found. Returns the given record unchanged when ip tracking is +// disabled globally or for the device, or when the scan fails. +func TrackDevice(app core.App, device *core.Record) *core.Record { + if !networking.DeviceTrackingEnabled(app, device) { + return device + } + subnet, err := networking.DeviceSubnet(device.GetString("ip"), device.GetString("netmask")) + if err != nil { + logger.Error.Println("Ip tracking for", device.GetString("name")+":", err) + return device + } + if err := TrackOneSubnet(app, subnet); err != nil { + logger.Error.Println("Ip tracking scan for", subnet.String()+":", err) + return device + } + fresh, err := app.FindRecordById("devices", device.Id) + if err != nil { + logger.Error.Println(err) + return device + } + return fresh +} + +// TrackDeviceAfterWake schedules a scan of the device's subnet to pick up +// the ip address the device acquired while booting. Does nothing unless ip +// tracking is enabled globally and for the device. +func TrackDeviceAfterWake(app core.App, device *core.Record) { + if !networking.DeviceTrackingEnabled(app, device) { + return + } + subnet, err := networking.DeviceSubnet(device.GetString("ip"), device.GetString("netmask")) + if err != nil { + logger.Error.Println("Ip tracking for", device.GetString("name")+":", err) + return + } + // the timer fires regardless of how the wake attempt ends and is not + // cancelled on app shutdown: a scan is harmless when the device never + // came up, and at worst runs once against a closing app + time.AfterFunc(wakeScanDelay, func() { + if err := TrackOneSubnet(app, subnet); err != nil { + logger.Error.Println("Ip tracking scan for", subnet.String()+":", err) + } + }) +} + // TrackOneSubnet runs an nmap scan of the given subnet and updates the ip // address of any ip-tracked device whose mac address is found at a new // address within its own subnet. Returns an error for a non-scannable subnet. @@ -93,6 +149,17 @@ func TrackOneSubnet(app core.App, subnet *net.IPNet) error { return err } + // a caller whose subnet is already being scanned waits for that scan's + // result instead of spawning another nmap run. The joined scan may have + // started before the caller's trigger and miss a very recent change; + // the periodic sweep covers that gap + _, err, _ := scanGroup.Do(subnet.String(), func() (any, error) { + return nil, scanSubnet(app, subnet) + }) + return err +} + +func scanSubnet(app core.App, subnet *net.IPNet) error { scan, err := nmapScan(subnet.String()) if err != nil { return err diff --git a/backend/iptracking/iptracking_test.go b/backend/iptracking/iptracking_test.go index f3221b89..98ec9668 100644 --- a/backend/iptracking/iptracking_test.go +++ b/backend/iptracking/iptracking_test.go @@ -5,7 +5,9 @@ import ( "fmt" "net" "strings" + "sync/atomic" "testing" + "time" "github.com/pocketbase/pocketbase/core" "github.com/pocketbase/pocketbase/tests" @@ -50,6 +52,29 @@ func newDevice(t *testing.T, app core.App, name, ip, netmask, mac string, trackI return device } +// enableTracking creates a settings record with a tracking interval, which +// enables ip tracking globally. +func enableTracking(t *testing.T, app core.App) { + t.Helper() + collection, err := app.FindCollectionByNameOrId("settings_private") + if err != nil { + t.Fatalf("Got unexpected error: %v", err) + } + settings := core.NewRecord(collection) + settings.Set("track_ip_interval", "@every 60s") + if err := app.SaveNoValidate(settings); err != nil { + t.Fatalf("Got unexpected error: %v", err) + } +} + +// shortenWakeScanDelay makes the post-wake scan fire almost immediately. +func shortenWakeScanDelay(t *testing.T) { + t.Helper() + orig := wakeScanDelay + wakeScanDelay = 10 * time.Millisecond + t.Cleanup(func() { wakeScanDelay = orig }) +} + // newLegacyDevice saves a device bypassing field validation, like rows // created before the ip and mac format rules existed. func newLegacyDevice(t *testing.T, app core.App, name, ip, netmask, mac string, trackIp bool) *core.Record { @@ -274,6 +299,58 @@ func TestTrackOneSubnetSameSubnetGuard(t *testing.T) { } } +// A wake schedules a delayed scan of the woken device's subnet; devices +// without track_ip never schedule one. +func TestTrackDeviceAfterWake(t *testing.T) { + app := newTestApp(t) + enableTracking(t, app) + shortenWakeScanDelay(t) + device := newDevice(t, app, "woken", "127.0.0.50", testNetmask, "AA:BB:CC:DD:05:01", true) + // in a different subnet, so a scheduling bug would show up as a second scan + untracked := newDevice(t, app, "untracked", "127.0.5.5", "255.255.255.0", "AA:BB:CC:DD:05:02", false) + scanned := stubScan(t, map[string]string{"AA:BB:CC:DD:05:01": "127.0.0.99"}) + + TrackDeviceAfterWake(app, untracked) + TrackDeviceAfterWake(app, device) + + deadline := time.Now().Add(2 * time.Second) + for { + fresh, err := app.FindRecordById("devices", device.Id) + if err != nil { + t.Fatalf("Got unexpected error: %v", err) + } + if fresh.GetString("ip") == "127.0.0.99" { + break + } + if time.Now().After(deadline) { + t.Fatalf("Ip not updated by the post-wake scan, scans: %v", *scanned) + } + time.Sleep(10 * time.Millisecond) + } + + // give a wrongly scheduled scan for the untracked device time to fire + time.Sleep(50 * time.Millisecond) + if len(*scanned) != 1 || (*scanned)[0] != testSubnet { + t.Errorf("Expected a single scan of %s, got %v", testSubnet, *scanned) + } +} + +// Without the global interval setting, a wake schedules no scan even for a +// tracked device. +func TestTrackDeviceAfterWakeRequiresGlobalEnable(t *testing.T) { + app := newTestApp(t) + shortenWakeScanDelay(t) + device := newDevice(t, app, "woken", "127.0.0.50", testNetmask, "AA:BB:CC:DD:06:01", true) + scanned := stubScan(t, map[string]string{"AA:BB:CC:DD:06:01": "127.0.0.99"}) + + TrackDeviceAfterWake(app, device) + + time.Sleep(100 * time.Millisecond) + if len(*scanned) != 0 { + t.Errorf("Expected no scans, got %v", *scanned) + } +} + // A paused periodic sweep scans nothing and owes a catch-up sweep, which // runs at most once until the next pause; an unpaused sweep scans directly. func TestPeriodicSweepAndCatchUp(t *testing.T) { @@ -307,6 +384,76 @@ func TestPeriodicSweepAndCatchUp(t *testing.T) { } } +// TrackDevice scans synchronously and returns the re-read device; when the +// subnet can't be scanned it returns the given record. +func TestTrackDevice(t *testing.T) { + app := newTestApp(t) + enableTracking(t, app) + device := newDevice(t, app, "refreshed", "127.0.0.50", testNetmask, "AA:BB:CC:DD:07:01", true) + stubScan(t, map[string]string{"AA:BB:CC:DD:07:01": "127.0.0.99"}) + + if ip := TrackDevice(app, device).GetString("ip"); ip != "127.0.0.99" { + t.Errorf("Ip mismatch: expected 127.0.0.99, got %s", ip) + } + + // TEST-NET-1 (RFC 5737) is never assigned to an interface + unscannable := newDevice(t, app, "unscannable", "192.0.2.5", "255.255.255.0", "AA:BB:CC:DD:07:02", true) + if got := TrackDevice(app, unscannable); got != unscannable { + t.Error("Expected the original record back for an unscannable subnet") + } +} + +// Without the global interval setting, TrackDevice never scans and returns +// the given record. +func TestTrackDeviceRequiresGlobalEnable(t *testing.T) { + app := newTestApp(t) + device := newDevice(t, app, "untouched", "127.0.0.50", testNetmask, "AA:BB:CC:DD:08:01", true) + scanned := stubScan(t, map[string]string{"AA:BB:CC:DD:08:01": "127.0.0.99"}) + + if got := TrackDevice(app, device); got != device { + t.Error("Expected the original record back when tracking is disabled") + } + if len(*scanned) != 0 { + t.Errorf("Expected no scans, got %v", *scanned) + } +} + +// A scan of a subnet already being scanned joins the in-flight scan instead +// of running nmap again. +func TestTrackOneSubnetCoalesces(t *testing.T) { + app := newTestApp(t) + + entered := make(chan struct{}, 2) + release := make(chan struct{}) + var calls atomic.Int32 + orig := nmapScan + nmapScan = func(scanRange string) (networking.Nmaprun, error) { + calls.Add(1) + entered <- struct{}{} + <-release + return networking.Nmaprun{}, nil + } + t.Cleanup(func() { nmapScan = orig }) + + subnet := mustSubnet(t, testSubnet) + errs := make(chan error, 2) + go func() { errs <- TrackOneSubnet(app, subnet) }() + <-entered // the first scan is now in flight + go func() { errs <- TrackOneSubnet(app, subnet) }() + // give the second call time to reach the flight group, then finish the scan + time.Sleep(100 * time.Millisecond) + close(release) + + for range 2 { + if err := <-errs; err != nil { + t.Fatalf("Got unexpected error: %v", err) + } + } + if n := calls.Load(); n != 1 { + t.Errorf("Expected the second scan to join the first, got %d nmap runs", n) + } +} + // TrackAllSubnets scans each unique scannable subnet once, silently skips the // rest, and updates any tracked device found at a new ip within its own // subnet — even one whose own subnet couldn't be scanned. diff --git a/backend/networking/ping_test.go b/backend/networking/ping_test.go index a89161ab..ee05c6ea 100644 --- a/backend/networking/ping_test.go +++ b/backend/networking/ping_test.go @@ -37,7 +37,7 @@ func TestPingDevice(t *testing.T) { device.Set("ip", tc.ip) device.Set("ping_cmd", tc.ping_cmd) - _, err := PingDevice(device) + _, err := PingDevice(device, nil) if err == nil && tc.wantError { t.Errorf("Expected error but got none") } else if err != nil && !tc.wantError { diff --git a/backend/networking/pingdevice_linux.go b/backend/networking/pingdevice_linux.go index 4b202bb5..074ccd97 100644 --- a/backend/networking/pingdevice_linux.go +++ b/backend/networking/pingdevice_linux.go @@ -14,10 +14,18 @@ import ( "kernel.org/pub/linux/libs/security/libcap/cap" ) -func PingDevice(device *core.Record) (bool, error) { +// PingDevice reports whether the device answers a ping. getIp is called for +// the address to ping, so it can follow ip changes written by concurrent +// tracking scans (see DeviceIPFunc); a nil getIp pings the record's ip. A +// custom ping_cmd runs verbatim and ignores the ip either way. +func PingDevice(device *core.Record, getIp func() string) (bool, error) { ping_cmd := device.GetString("ping_cmd") if ping_cmd == "" { - pinger, err := probing.NewPinger(device.GetString("ip")) + ip := device.GetString("ip") + if getIp != nil { + ip = getIp() + } + pinger, err := probing.NewPinger(ip) if err != nil { return false, err } diff --git a/backend/networking/pingdevice_other.go b/backend/networking/pingdevice_other.go index 9ff79799..3d01aeb9 100644 --- a/backend/networking/pingdevice_other.go +++ b/backend/networking/pingdevice_other.go @@ -13,10 +13,18 @@ import ( probing "github.com/prometheus-community/pro-bing" ) -func PingDevice(device *core.Record) (bool, error) { +// PingDevice reports whether the device answers a ping. getIp is called for +// the address to ping, so it can follow ip changes written by concurrent +// tracking scans (see DeviceIPFunc); a nil getIp pings the record's ip. A +// custom ping_cmd runs verbatim and ignores the ip either way. +func PingDevice(device *core.Record, getIp func() string) (bool, error) { ping_cmd := device.GetString("ping_cmd") if ping_cmd == "" { - pinger, err := probing.NewPinger(device.GetString("ip")) + ip := device.GetString("ip") + if getIp != nil { + ip = getIp() + } + pinger, err := probing.NewPinger(ip) if err != nil { return false, err } diff --git a/backend/networking/scan_linux.go b/backend/networking/scan_linux.go index 685f7ed2..11f0ad61 100644 --- a/backend/networking/scan_linux.go +++ b/backend/networking/scan_linux.go @@ -4,11 +4,20 @@ package networking import ( "fmt" + "sync" "kernel.org/pub/linux/libs/security/libcap/cap" ) +// nmapMu serializes scans: raising and restoring NET_RAW mutates +// process-wide state, so concurrent scans would race the capability +// dance and could leave it raised for unrelated child processes +var nmapMu sync.Mutex + func NmapScan(scanRange string) (Nmaprun, error) { + nmapMu.Lock() + defer nmapMu.Unlock() + orig := cap.GetProc() defer orig.SetProc() // restore original caps on exit. diff --git a/backend/networking/shutdown.go b/backend/networking/shutdown.go index 17945f87..ee92d5b8 100644 --- a/backend/networking/shutdown.go +++ b/backend/networking/shutdown.go @@ -77,7 +77,7 @@ func ShutdownDevice(device *core.Record) error { return fmt.Errorf("%s", stderr.String()) } else { for { - isOnline, err := PingDevice(device) + isOnline, err := PingDevice(device, nil) if err != nil { logger.Error.Println(err) return err @@ -94,7 +94,7 @@ func ShutdownDevice(device *core.Record) error { } } default: - isOnline, err := PingDevice(device) + isOnline, err := PingDevice(device, nil) if err != nil { logger.Error.Println(err) return err diff --git a/backend/networking/trackip.go b/backend/networking/trackip.go index 7837f57e..d8f6c066 100644 --- a/backend/networking/trackip.go +++ b/backend/networking/trackip.go @@ -3,8 +3,39 @@ package networking import ( "errors" "net" + + "github.com/pocketbase/pocketbase/core" ) +// DeviceTrackingEnabled reports whether ip tracking is enabled both +// globally and for the device. +func DeviceTrackingEnabled(app core.App, device *core.Record) bool { + if !device.GetBool("track_ip") { + return false + } + settings, err := app.FindFirstRecordByFilter("settings_private", "") + return err == nil && settings.GetString("track_ip_interval") != "" +} + +// DeviceIPFunc returns a function that returns the device's current ip +// address. When ip tracking is enabled globally and for the device, each +// call re-reads the device from the database to pick up concurrent tracking +// updates; otherwise it always returns the given record's ip. The returned +// function is not safe for concurrent use. +func DeviceIPFunc(app core.App, device *core.Record) func() string { + if !DeviceTrackingEnabled(app, device) { + return func() string { + return device.GetString("ip") + } + } + return func() string { + if fresh, err := app.FindRecordById("devices", device.Id); err == nil { + device = fresh + } + return device.GetString("ip") + } +} + // DeviceSubnet returns the device's IPv4 subnet computed from its ip and netmask. func DeviceSubnet(ipStr, maskStr string) (*net.IPNet, error) { ip := net.ParseIP(ipStr) diff --git a/backend/networking/trackip_app_test.go b/backend/networking/trackip_app_test.go new file mode 100644 index 00000000..8d598be1 --- /dev/null +++ b/backend/networking/trackip_app_test.go @@ -0,0 +1,131 @@ +package networking + +import ( + "testing" + + "github.com/pocketbase/pocketbase/core" + "github.com/pocketbase/pocketbase/tests" + + // register the app migrations so test apps get the real schema + _ "github.com/seriousm4x/upsnap/migrations" +) + +// newTestApp returns a throwaway app with all migrations applied. +func newTestApp(t *testing.T) *tests.TestApp { + t.Helper() + app, err := tests.NewTestApp(t.TempDir()) + if err != nil { + t.Fatalf("Got unexpected error: %v", err) + } + t.Cleanup(app.Cleanup) + return app +} + +func newDevice(t *testing.T, app core.App, name, ip, mac string, trackIp bool) *core.Record { + t.Helper() + collection, err := app.FindCollectionByNameOrId("devices") + if err != nil { + t.Fatalf("Got unexpected error: %v", err) + } + device := core.NewRecord(collection) + device.Set("name", name) + device.Set("ip", ip) + device.Set("netmask", "255.255.0.0") + device.Set("mac", mac) + device.Set("track_ip", trackIp) + if err := app.Save(device); err != nil { + t.Fatalf("Got unexpected error: %v", err) + } + return device +} + +func newSettings(t *testing.T, app core.App, trackIpInterval string) *core.Record { + t.Helper() + collection, err := app.FindCollectionByNameOrId("settings_private") + if err != nil { + t.Fatalf("Got unexpected error: %v", err) + } + settings := core.NewRecord(collection) + settings.Set("track_ip_interval", trackIpInterval) + if err := app.SaveNoValidate(settings); err != nil { + t.Fatalf("Got unexpected error: %v", err) + } + return settings +} + +func TestDeviceTrackingEnabled(t *testing.T) { + app := newTestApp(t) + tracked := newDevice(t, app, "tracked", "127.0.0.50", "AA:BB:CC:DD:0B:01", true) + untracked := newDevice(t, app, "untracked", "127.0.0.51", "AA:BB:CC:DD:0B:02", false) + + // without a settings record tracking is off + if DeviceTrackingEnabled(app, tracked) { + t.Error("Expected tracking to be disabled without a settings record") + } + + // an empty interval disables tracking globally + settings := newSettings(t, app, "") + if DeviceTrackingEnabled(app, tracked) { + t.Error("Expected tracking to be disabled with an empty interval") + } + + // the global interval and the device toggle together enable tracking + settings.Set("track_ip_interval", "@every 60s") + if err := app.SaveNoValidate(settings); err != nil { + t.Fatalf("Got unexpected error: %v", err) + } + if !DeviceTrackingEnabled(app, tracked) { + t.Error("Expected tracking to be enabled") + } + if DeviceTrackingEnabled(app, untracked) { + t.Error("Expected tracking to be disabled for a device without track_ip") + } +} + +func TestDeviceIPFunc(t *testing.T) { + app := newTestApp(t) + newSettings(t, app, "@every 60s") + tracked := newDevice(t, app, "tracked", "127.0.0.50", "AA:BB:CC:DD:0C:01", true) + untracked := newDevice(t, app, "untracked", "127.0.0.60", "AA:BB:CC:DD:0C:02", false) + + trackedIp := DeviceIPFunc(app, tracked) + untrackedIp := DeviceIPFunc(app, untracked) + if ip := trackedIp(); ip != "127.0.0.50" { + t.Errorf("Ip mismatch: expected 127.0.0.50, got %s", ip) + } + if ip := untrackedIp(); ip != "127.0.0.60" { + t.Errorf("Ip mismatch: expected 127.0.0.60, got %s", ip) + } + + // change both ips in the database behind the records' backs + for _, d := range []*core.Record{tracked, untracked} { + fresh, err := app.FindRecordById("devices", d.Id) + if err != nil { + t.Fatalf("Got unexpected error: %v", err) + } + fresh.Set("ip", "127.0.9.9") + if err := app.Save(fresh); err != nil { + t.Fatalf("Got unexpected error: %v", err) + } + } + + // only the tracked device's getter follows the change + if ip := trackedIp(); ip != "127.0.9.9" { + t.Errorf("Ip mismatch: expected 127.0.9.9, got %s", ip) + } + if ip := untrackedIp(); ip != "127.0.0.60" { + t.Errorf("Ip mismatch: expected 127.0.0.60, got %s", ip) + } + + // a deleted device falls back to the last known ip + fresh, err := app.FindRecordById("devices", tracked.Id) + if err != nil { + t.Fatalf("Got unexpected error: %v", err) + } + if err := app.Delete(fresh); err != nil { + t.Fatalf("Got unexpected error: %v", err) + } + if ip := trackedIp(); ip != "127.0.9.9" { + t.Errorf("Ip mismatch after delete: expected 127.0.9.9, got %s", ip) + } +} diff --git a/backend/networking/wake.go b/backend/networking/wake.go index 03731f66..c9e618e0 100644 --- a/backend/networking/wake.go +++ b/backend/networking/wake.go @@ -14,7 +14,11 @@ import ( "github.com/seriousm4x/upsnap/logger" ) -func WakeDevice(device *core.Record) error { +// WakeDevice wakes the device and waits for it to come online. getIp is +// called for the address to ping on each attempt, so it can follow ip +// changes written by concurrent tracking scans (see DeviceIPFunc); a nil +// getIp always pings the record's ip. +func WakeDevice(device *core.Record, getIp func() string) error { logger.Info.Println("Wake triggered for", device.GetString("name")) wakeTimeout := device.GetInt("wake_timeout") @@ -79,7 +83,7 @@ func WakeDevice(device *core.Record) error { } return fmt.Errorf("%s not online after %d seconds", device.GetString("name"), wakeTimeout) } - isOnline, err := PingDevice(device) + isOnline, err := PingDevice(device, getIp) if err != nil { logger.Error.Println(err) return err @@ -113,7 +117,7 @@ func WakeDevice(device *core.Record) error { start := time.Now() for { time.Sleep(1 * time.Second) - isOnline, err := PingDevice(device) + isOnline, err := PingDevice(device, getIp) if err != nil { logger.Error.Println(err) return err diff --git a/backend/networking/wake_test.go b/backend/networking/wake_test.go new file mode 100644 index 00000000..d6b5d6b8 --- /dev/null +++ b/backend/networking/wake_test.go @@ -0,0 +1,28 @@ +package networking + +import ( + "testing" + + "github.com/pocketbase/pocketbase/core" +) + +// WakeDevice consults getIp for the address of every ping attempt during +// its wait. The wait's outcome depends on the environment's ping +// privileges, so only the consultation itself is asserted. +func TestWakeDeviceConsultsGetIp(t *testing.T) { + device := core.NewRecord(&core.Collection{}) + device.Set("name", "test") + device.Set("ip", "127.0.0.2") + device.Set("netmask", "255.255.255.0") + device.Set("mac", "AA:BB:CC:DD:EE:0F") + device.Set("wake_timeout", 1) + + calls := 0 + _ = WakeDevice(device, func() string { + calls++ + return "127.0.0.2" + }) + if calls == 0 { + t.Error("Expected getIp to be consulted during the wake wait") + } +} diff --git a/backend/pb/handlers.go b/backend/pb/handlers.go index 794c3fb3..4691ae9e 100644 --- a/backend/pb/handlers.go +++ b/backend/pb/handlers.go @@ -13,6 +13,7 @@ import ( "github.com/pocketbase/pocketbase/core" "github.com/pocketbase/pocketbase/tools/router" "github.com/robfig/cron/v3" + "github.com/seriousm4x/upsnap/iptracking" "github.com/seriousm4x/upsnap/logger" "github.com/seriousm4x/upsnap/networking" ) @@ -22,14 +23,28 @@ func HandlerWake(e *core.RequestEvent) error { if err != nil { return apis.NewNotFoundError("The device does not exist.", err) } + // only write status changes so concurrent writers to other fields + // (e.g. ip tracking) are never clobbered; same in the handlers below + record.IgnoreUnchangedFields(true) + + // a pending device already has an action in progress: report the + // current state instead of starting another one; same below + if record.GetString("status") == "pending" { + return e.JSON(http.StatusOK, record) + } + // the PostScan calls here and below refresh the save baseline so a + // later revert to the load-time status isn't dropped as unchanged record.Set("status", "pending") if err := e.App.Save(record); err != nil { logger.Error.Println("Failed to save record:", err) + } else if err := record.PostScan(); err != nil { + logger.Error.Println(err) } if err := asyncCall(e, func() *router.ApiError { - if err := networking.WakeDevice(record); err != nil { + iptracking.TrackDeviceAfterWake(e.App, record) + if err := networking.WakeDevice(record, networking.DeviceIPFunc(e.App, record)); err != nil { logger.Error.Println(err) record.Set("status", "offline") if err := e.App.Save(record); err != nil { @@ -56,25 +71,35 @@ func HandlerSleep(e *core.RequestEvent) error { if err != nil { return apis.NewNotFoundError("The device does not exist.", err) } + record.IgnoreUnchangedFields(true) + + if record.GetString("status") == "pending" { + return e.JSON(http.StatusOK, record) + } record.Set("status", "pending") if err := e.App.Save(record); err != nil { logger.Error.Println("Failed to save record:", err) + } else if err := record.PostScan(); err != nil { + logger.Error.Println(err) } if err := asyncCall(e, func() *router.ApiError { - resp, err := networking.SleepDevice(record) + device := iptracking.TrackDevice(e.App, record) + device.IgnoreUnchangedFields(true) + + resp, err := networking.SleepDevice(device) if err != nil { logger.Error.Println(err) - record.Set("status", "online") - if err := e.App.Save(record); err != nil { + device.Set("status", "online") + if err := e.App.Save(device); err != nil { logger.Error.Println("Failed to save record:", err) } return apis.NewBadRequestError(resp.Message, nil) } - record.Set("status", "offline") - if err := e.App.Save(record); err != nil { + device.Set("status", "offline") + if err := e.App.Save(device); err != nil { logger.Error.Println("Failed to save record:", err) } @@ -91,17 +116,27 @@ func HandlerReboot(e *core.RequestEvent) error { if err != nil { return apis.NewNotFoundError("The device does not exist.", err) } + record.IgnoreUnchangedFields(true) + + if record.GetString("status") == "pending" { + return e.JSON(http.StatusOK, record) + } record.Set("status", "pending") if err := e.App.Save(record); err != nil { logger.Error.Println("Failed to save record:", err) + } else if err := record.PostScan(); err != nil { + logger.Error.Println(err) } if err := asyncCall(e, func() *router.ApiError { - if err := networking.ShutdownDevice(record); err != nil { + device := iptracking.TrackDevice(e.App, record) + device.IgnoreUnchangedFields(true) + + if err := networking.ShutdownDevice(device); err != nil { logger.Error.Println(err) - record.Set("status", "online") - if err := e.App.Save(record); err != nil { + device.Set("status", "online") + if err := e.App.Save(device); err != nil { logger.Error.Println("Failed to save record:", err) } return apis.NewBadRequestError(err.Error(), nil) @@ -112,17 +147,18 @@ func HandlerReboot(e *core.RequestEvent) error { // so we wait a little to make sure the device has shut down completely and is ready to receive wake requests. time.Sleep(15 * time.Second) - if err := networking.WakeDevice(record); err != nil { + iptracking.TrackDeviceAfterWake(e.App, device) + if err := networking.WakeDevice(device, networking.DeviceIPFunc(e.App, device)); err != nil { logger.Error.Println(err) - record.Set("status", "offline") - if err := e.App.Save(record); err != nil { + device.Set("status", "offline") + if err := e.App.Save(device); err != nil { logger.Error.Println("Failed to save record:", err) } return apis.NewBadRequestError(err.Error(), nil) } - record.Set("status", "online") - if err := e.App.Save(record); err != nil { + device.Set("status", "online") + if err := e.App.Save(device); err != nil { logger.Error.Println("Failed to save record:", err) } @@ -131,6 +167,11 @@ func HandlerReboot(e *core.RequestEvent) error { return err } + // re-read so a synchronous request reports the action's outcome + // instead of the pending state + if fresh, err := e.App.FindRecordById("devices", record.Id); err == nil { + return e.JSON(http.StatusOK, fresh) + } return e.JSON(http.StatusOK, record) } @@ -139,24 +180,34 @@ func HandlerShutdown(e *core.RequestEvent) error { if err != nil { return apis.NewNotFoundError("The device does not exist.", err) } + record.IgnoreUnchangedFields(true) + + if record.GetString("status") == "pending" { + return e.JSON(http.StatusOK, record) + } record.Set("status", "pending") if err := e.App.Save(record); err != nil { logger.Error.Println("Failed to save record:", err) + } else if err := record.PostScan(); err != nil { + logger.Error.Println(err) } if err := asyncCall(e, func() *router.ApiError { - if err := networking.ShutdownDevice(record); err != nil { + device := iptracking.TrackDevice(e.App, record) + device.IgnoreUnchangedFields(true) + + if err := networking.ShutdownDevice(device); err != nil { logger.Error.Println(strings.ReplaceAll(err.Error(), "\n", "")) - record.Set("status", "online") - if err := e.App.Save(record); err != nil { + device.Set("status", "online") + if err := e.App.Save(device); err != nil { logger.Error.Println("Failed to save record:", err) } return apis.NewBadRequestError(err.Error(), nil) } - record.Set("status", "offline") - if err := e.App.Save(record); err != nil { + device.Set("status", "offline") + if err := e.App.Save(device); err != nil { logger.Error.Println("Failed to save record:", err) } @@ -165,6 +216,11 @@ func HandlerShutdown(e *core.RequestEvent) error { return err } + // re-read so a synchronous request reports the action's outcome + // instead of the pending state + if fresh, err := e.App.FindRecordById("devices", record.Id); err == nil { + return e.JSON(http.StatusOK, fresh) + } return e.JSON(http.StatusOK, record) } @@ -178,12 +234,16 @@ func HandlerWakeGroup(e *core.RequestEvent) error { for _, record := range records { go func() { + record.IgnoreUnchangedFields(true) record.Set("status", "pending") if err := e.App.Save(record); err != nil { logger.Error.Println("Failed to save record:", err) + } else if err := record.PostScan(); err != nil { + logger.Error.Println(err) } - if err := networking.WakeDevice(record); err != nil { + iptracking.TrackDeviceAfterWake(e.App, record) + if err := networking.WakeDevice(record, networking.DeviceIPFunc(e.App, record)); err != nil { logger.Error.Println(err) record.Set("status", "offline") if err := e.App.Save(record); err != nil { @@ -212,22 +272,28 @@ func HandlerShutdownGroup(e *core.RequestEvent) error { for _, record := range records { go func() { + record.IgnoreUnchangedFields(true) record.Set("status", "pending") if err := e.App.Save(record); err != nil { logger.Error.Println("Failed to save record:", err) + } else if err := record.PostScan(); err != nil { + logger.Error.Println(err) } - if err := networking.ShutdownDevice(record); err != nil { + device := iptracking.TrackDevice(e.App, record) + device.IgnoreUnchangedFields(true) + + if err := networking.ShutdownDevice(device); err != nil { logger.Error.Println(err) - record.Set("status", "online") - if err := e.App.Save(record); err != nil { + device.Set("status", "online") + if err := e.App.Save(device); err != nil { logger.Error.Println("Failed to save record:", err) } return } - record.Set("status", "offline") - if err := e.App.Save(record); err != nil { + device.Set("status", "offline") + if err := e.App.Save(device); err != nil { logger.Error.Println("Failed to save record:", err) } }() diff --git a/backend/pb/pb.go b/backend/pb/pb.go index 2834adb1..5291cbbb 100644 --- a/backend/pb/pb.go +++ b/backend/pb/pb.go @@ -321,6 +321,8 @@ func resetDeviceStates(app *pocketbase.PocketBase) error { return err } for _, device := range devices { + // only write the status so concurrent writers are never clobbered + device.IgnoreUnchangedFields(true) device.Set("status", "offline") if err := app.Save(device); err != nil { return err diff --git a/frontend/src/lib/components/DeviceCard.svelte b/frontend/src/lib/components/DeviceCard.svelte index be719e36..743f43c2 100644 --- a/frontend/src/lib/components/DeviceCard.svelte +++ b/frontend/src/lib/components/DeviceCard.svelte @@ -60,7 +60,7 @@ }); function sleep() { - fetch(`${backendUrl}api/upsnap/sleep/${device.id}`, { + fetch(`${backendUrl}api/upsnap/sleep/${device.id}?async=true`, { headers: { Authorization: $pocketbase.authStore.token } @@ -70,7 +70,7 @@ } function reboot() { - fetch(`${backendUrl}api/upsnap/reboot/${device.id}`, { + fetch(`${backendUrl}api/upsnap/reboot/${device.id}?async=true`, { headers: { Authorization: $pocketbase.authStore.token } diff --git a/frontend/src/lib/components/DeviceCardNic.svelte b/frontend/src/lib/components/DeviceCardNic.svelte index 92a5cd19..3c20b68c 100644 --- a/frontend/src/lib/components/DeviceCardNic.svelte +++ b/frontend/src/lib/components/DeviceCardNic.svelte @@ -80,18 +80,17 @@ countdown(Date.now(), 'shutdown'); device.status = 'pending'; - fetch(`${backendUrl}api/upsnap/shutdown/${device.id}`, { + // the shutdown runs asynchronously in the backend; the outcome + // arrives as a realtime status update + fetch(`${backendUrl}api/upsnap/shutdown/${device.id}?async=true`, { headers: { Authorization: $pocketbase.authStore.token } }) - .then((resp) => resp.json()) - .then(async (data) => { - if (data.status !== 200) { + .then((resp) => { + if (!resp.ok) { device.status = 'online'; - return; } - device = data as Device; }) .catch((err) => { toast.error(err.message); @@ -118,6 +117,16 @@ clearInterval(interval); interval = 0; + + if (timeout <= 0 && device.status === 'pending') { + // the countdown outlived the action (e.g. a tracking scan + // ran before it) or a realtime update was missed: re-sync + $pocketbase + .collection('devices') + .getOne(device.id) + .then((data) => (device = data)) + .catch(() => {}); + } } }, 1000); }