hippogrief/pobsync
1
0
Fork 0
Code Issues 21 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

(bugfix) Avoid live backup data scans in web views

Browse Source 
Use stored snapshot storage metadata for dashboard and host backup data summaries instead of walking snapshot directories during request rendering. Snapshots without recorded storage metadata are counted as not measured so large backup targets cannot trigger unbounded filesystem scans from live-refresh views.

Closes #97
This commit is contained in:
Peter van Arkel
2026-06-08 22:48:22 +02:00
parent 522ae98bb1
commit f886a7b620
6 changed files with 163 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files

73
src/pobsync_backend/tests/test_views.py
View File

@@ -8,11 +8,9 @@ from unittest.mock import patch
from django.contrib.auth import get_user_model
from django.core.files.uploadedfile import SimpleUploadedFile
from django.template.defaultfilters import filesizeformat
from django.test import TestCase, override_settings
from django.urls import reverse
from pobsync.run_stats import tree_usage
from pobsync.util import write_yaml_atomic
from pobsync_backend.models import (
BackupRun,
@@ -464,23 +462,12 @@ class ViewTests(TestCase):
db = HostConfig.objects.create(host="db-01", address="db-01.example.test")
scheduled = self._snapshot(web, "20260519-021500Z__SCHED01", kind=SnapshotRecord.Kind.SCHEDULED)
manual = self._snapshot(web, "20260519-031500Z__MANUAL1", kind=SnapshotRecord.Kind.MANUAL)
incomplete = self._snapshot(db, "20260519-041500Z__BROKEN1", kind=SnapshotRecord.Kind.INCOMPLETE)
self._set_snapshot_storage(scheduled, allocated=100)
self._set_snapshot_storage(manual, allocated=200)
with TemporaryDirectory() as tmp:
incomplete_dir = Path(tmp) / db.host / ".incomplete" / "20260519-041500Z__BROKEN1"
data_dir = incomplete_dir / "data"
data_dir.mkdir(parents=True)
data_dir.joinpath("partial-file").write_text("interrupted backup data\n", encoding="utf-8")
expected_usage = tree_usage(data_dir)
SnapshotRecord.objects.create(
host=db,
kind=SnapshotRecord.Kind.INCOMPLETE,
dirname=incomplete_dir.name,
path=str(incomplete_dir),
status="failed",
)
self._set_snapshot_storage(incomplete, allocated=400)
response = self.client.get(reverse("dashboard_priority_live"))
response = self.client.get(reverse("dashboard_priority_live"))
self.assertEqual(response.status_code, 200)
self.assertContains(response, "Scheduled data")
@@ -489,8 +476,8 @@ class ViewTests(TestCase):
self.assertContains(response, "Total snapshot data")
self.assertContains(response, "100 bytes", html=True)
self.assertContains(response, "200 bytes", html=True)
self.assertContains(response, filesizeformat(expected_usage["allocated_size_bytes"]))
self.assertContains(response, filesizeformat(300 + expected_usage["allocated_size_bytes"]))
self.assertContains(response, "400 bytes", html=True)
self.assertContains(response, "700 bytes", html=True)
def test_dashboard_hosts_live_returns_hosts_partial(self) -> None:
self.client.force_login(self.staff_user)
@@ -510,23 +497,12 @@ class ViewTests(TestCase):
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
scheduled = self._snapshot(host, "20260519-021500Z__SCHED01", kind=SnapshotRecord.Kind.SCHEDULED)
manual = self._snapshot(host, "20260519-031500Z__MANUAL1", kind=SnapshotRecord.Kind.MANUAL)
incomplete = self._snapshot(host, "20260519-041500Z__BROKEN1", kind=SnapshotRecord.Kind.INCOMPLETE)
self._set_snapshot_storage(scheduled, allocated=100)
self._set_snapshot_storage(manual, allocated=200)
with TemporaryDirectory() as tmp:
incomplete_dir = Path(tmp) / host.host / ".incomplete" / "20260519-041500Z__BROKEN1"
data_dir = incomplete_dir / "data"
data_dir.mkdir(parents=True)
data_dir.joinpath("partial-file").write_text("interrupted backup data\n", encoding="utf-8")
expected_usage = tree_usage(data_dir)
SnapshotRecord.objects.create(
host=host,
kind=SnapshotRecord.Kind.INCOMPLETE,
dirname=incomplete_dir.name,
path=str(incomplete_dir),
status="failed",
)
self._set_snapshot_storage(incomplete, allocated=400)
response = self.client.get(reverse("dashboard_hosts_live"))
response = self.client.get(reverse("dashboard_hosts_live"))
self.assertEqual(response.status_code, 200)
self.assertContains(response, "Scheduled data")
@@ -535,32 +511,27 @@ class ViewTests(TestCase):
self.assertContains(response, "Total data")
self.assertContains(response, "100 bytes", html=True)
self.assertContains(response, "200 bytes", html=True)
self.assertContains(response, filesizeformat(expected_usage["allocated_size_bytes"]))
self.assertContains(response, filesizeformat(300 + expected_usage["allocated_size_bytes"]))
self.assertContains(response, "400 bytes", html=True)
self.assertContains(response, "700 bytes", html=True)
def test_dashboard_host_cards_measure_incomplete_data_without_snapshot_metadata(self) -> None:
def test_dashboard_host_cards_mark_incomplete_data_without_snapshot_metadata_unmeasured(self) -> None:
self.client.force_login(self.staff_user)
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
with TemporaryDirectory() as tmp:
incomplete_dir = Path(tmp) / host.host / ".incomplete" / "20260519-041500Z__BROKEN1"
data_dir = incomplete_dir / "data"
data_dir.mkdir(parents=True)
data_dir.joinpath("partial-file").write_text("interrupted backup data\n", encoding="utf-8")
expected_usage = tree_usage(data_dir)
SnapshotRecord.objects.create(
host=host,
kind=SnapshotRecord.Kind.INCOMPLETE,
dirname=incomplete_dir.name,
path=str(incomplete_dir),
status="failed",
metadata={},
)
SnapshotRecord.objects.create(
host=host,
kind=SnapshotRecord.Kind.INCOMPLETE,
dirname="20260519-041500Z__BROKEN1",
path="/backups/web-01/.incomplete/20260519-041500Z__BROKEN1",
status="failed",
metadata={},
)
response = self.client.get(reverse("dashboard_hosts_live"))
response = self.client.get(reverse("dashboard_hosts_live"))
self.assertEqual(response.status_code, 200)
self.assertContains(response, "Incomplete data")
self.assertContains(response, filesizeformat(expected_usage["allocated_size_bytes"]))
self.assertContains(response, "0 bytes", html=True)
self.assertContains(response, "1 not measured")
def test_hosts_list_renders_host_cards_and_controls(self) -> None:
self.client.force_login(self.staff_user)