Disk-full owner email (shard-initiated event, #76)

agents.md

@@ -36,6 +36,7 @@ shard_core/
     freeshard_controller.py  API calls to controller
     telemetry.py        Usage metrics reporting
     disk.py             Disk space monitoring
+    disk_full_notification.py  Owner email when disk crosses threshold (via controller relay, deduped)
     migration.py        Database schema migrations
     websocket.py        WebSocket connection lifecycle
   database/           → PostgreSQL access layer (per-entity modules, conn-first-arg pattern)
@@ -86,6 +87,7 @@ Started at app lifespan startup, stopped at shutdown:
 - `InstallationWorker` — async task queue for app install/uninstall
 - `PeriodicTask(control_apps, 30s)` — app idle lifecycle. With `apps.lifecycle.pause_enabled` (default off): RUNNING pauses after `idle_for_pause` (cgroup freeze + page-out to swap), PAUSED stops after `idle_for_stop`, and high memory PSI demotes the LRU app one tier per cycle. Flag off: legacy stop-only
 - `PeriodicTask(update_disk_space, 30s)` — disk monitoring
+- `PeriodicTask(disk_full_notification.run_check, 30s)` — owner email when disk usage ≥ `event_notifications.disk_full.threshold_percent` (default 90%), deduped via kv_store (sent once, re-armed when usage drops below threshold), opt-out via `event_notifications.disk_full.enabled`. Runs as its own task so a slow controller relay call can never stall disk monitoring
 - `CronTask(start_backup, "0 3 * * *")` — daily backup with random delay
 - `CronTask(docker_prune_images, daily)` — image cleanup
 - Various telemetry and peer key refresh tasks

config.toml

@@ -57,6 +57,10 @@ enabled = false
 enabled = false
 send_interval_seconds = 300
 
+[event_notifications.disk_full]
+enabled = true
+threshold_percent = 90
+
 [management]
 api_url = "https://ptlfunctionapp.azurewebsites.net/api/management"
 

shard_core/app_factory.py

@@ -24,6 +24,7 @@
     portal_controller,
     backup,
     disk,
+    disk_full_notification,
     telemetry,
 )
 from .service.app_installation.util import (
@@ -133,6 +134,7 @@ def _make_background_tasks() -> List[BackgroundTask]:
             max_random_delay=s.services.backup.timing.max_random_delay,
         ),
         PeriodicTask(disk.update_disk_space, 30),
+        PeriodicTask(disk_full_notification.run_check, 30),
         websocket.ws_worker,
         PeriodicTask(telemetry.send_telemetry, s.telemetry.send_interval_seconds),
     ]

shard_core/service/disk_full_notification.py

@@ -0,0 +1,99 @@
+import json
+import logging
+
+from shard_core.database import database
+from shard_core.service import disk
+from shard_core.service.freeshard_controller import call_freeshard_controller
+from shard_core.settings import settings
+
+log = logging.getLogger(__name__)
+
+KV_KEY_DISK_FULL_NOTIFICATION_SENT = "disk_full_notification_sent"
+
+ACTIVE_LANGUAGE = "en"
+
+_TEMPLATES = {
+    "en": {
+        "subject": "Your Freeshard is running low on disk space",
+        "body": [
+            "Hello,",
+            "Your Freeshard has used {used_percent:.0f}% of its disk space.",
+            "When the disk fills up, apps may stop working correctly and backups "
+            "may fail. Please free up space or expand your storage.",
+            "You will not receive another disk space warning until usage drops "
+            "back below the threshold.",
+            "— Your Freeshard",
+        ],
+    },
+    "de": {
+        "subject": "Der Speicherplatz deines Freeshards wird knapp",
+        "body": [
+            "Hallo,",
+            "Dein Freeshard hat {used_percent:.0f}% seines Speicherplatzes belegt.",
+            "Wenn der Speicher voll ist, funktionieren Apps möglicherweise nicht "
+            "mehr richtig und Backups können fehlschlagen. Bitte gib Speicher frei "
+            "oder erweitere deinen Speicher.",
+            "Du erhältst erst wieder eine Warnung, wenn die Belegung unter den "
+            "Schwellwert fällt.",
+            "— Dein Freeshard",
+        ],
+    },
+}
+
+
+async def run_check() -> None:
+    """Background-task entry point: check the latest disk snapshot."""
+    await check_disk_full(disk.current_disk_usage)
+
+
+async def check_disk_full(usage: disk.DiskUsage) -> None:
+    """Send a one-off email when disk usage crosses the configured threshold.
+
+    Deduplicated via a persistent flag in the kv_store: the email is sent once
+    when usage rises above the threshold and re-armed only after usage falls
+    back below it.
+    """
+    config = settings().event_notifications.disk_full
+    if not config.enabled:
+        return
+    if usage.total_gb <= 0:
+        return
+
+    used_percent = (usage.total_gb - usage.free_gb) / usage.total_gb * 100
+    over_threshold = used_percent >= config.threshold_percent
+    already_notified = await _was_notified()
+
+    if over_threshold and not already_notified:
+        try:
+            await _send_disk_full_email(used_percent)
+        except Exception as e:
+            log.error(f"failed to send disk-full notification email: {e}")
+            return
+        await database.set_value(KV_KEY_DISK_FULL_NOTIFICATION_SENT, True)
+    elif not over_threshold and already_notified:
+        await database.set_value(KV_KEY_DISK_FULL_NOTIFICATION_SENT, False)
+
+
+async def _was_notified() -> bool:
+    try:
+        return bool(await database.get_value(KV_KEY_DISK_FULL_NOTIFICATION_SENT))
+    except KeyError:
+        return False
+
+
+async def _send_disk_full_email(used_percent: float) -> None:
+    template = _TEMPLATES[ACTIVE_LANGUAGE]
+    payload = {
+        "subject": template["subject"],
+        "body": [line.format(used_percent=used_percent) for line in template["body"]],
+    }
+    response = await call_freeshard_controller(
+        "api/email_relay",
+        method="POST",
+        body=json.dumps(payload).encode(),
+    )
+    try:
+        response.raise_for_status()
+    finally:
+        response.close()
+    log.info(f"sent disk-full notification email (disk {used_percent:.0f}% full)")

shard_core/settings.py

@@ -90,6 +90,15 @@ class FreeshardControllerSettings(BaseModel):
     base_url: str
 
 
+class DiskFullNotificationSettings(BaseModel):
+    enabled: bool = True
+    threshold_percent: float = 90.0
+
+
+class EventNotificationsSettings(BaseModel):
+    disk_full: DiskFullNotificationSettings = DiskFullNotificationSettings()
+
+
 class DatabaseSettings(BaseModel):
     host: str = "postgres"
     port: int = 5432
@@ -121,6 +130,7 @@ class Settings(BaseSettings):
     traefik: TraefikSettings
     apps: AppsSettings
     telemetry: TelemetrySettings = TelemetrySettings()
+    event_notifications: EventNotificationsSettings = EventNotificationsSettings()
     management: ManagementSettings
     freeshard_controller: FreeshardControllerSettings
     log: LogSettings = LogSettings()

tests/config.toml

@@ -19,3 +19,6 @@ shard_core = "info"
 
 [telemetry]
 enabled = true
+
+[event_notifications.disk_full]
+enabled = false

tests/test_disk_full_notification.py

@@ -0,0 +1,152 @@
+import json
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from requests import HTTPError
+
+from shard_core.service import disk
+from shard_core.service import disk_full_notification as dfn
+from shard_core.service.disk import DiskUsage
+from tests.conftest import settings_override
+
+
+def _usage_at(used_percent: float, total_gb: float = 100.0) -> DiskUsage:
+    free_gb = total_gb * (1 - used_percent / 100)
+    return DiskUsage(total_gb=total_gb, free_gb=free_gb, disk_space_low=False)
+
+
+def _enabled_config(threshold_percent: float = 90):
+    return settings_override(
+        {
+            "event_notifications": {
+                "disk_full": {"enabled": True, "threshold_percent": threshold_percent}
+            }
+        }
+    )
+
+
+def _patch_relay():
+    """Patch the controller relay call with a 201-style success response."""
+    mock_call = AsyncMock()
+    mock_call.return_value = MagicMock()  # requests.Response with sync raise_for_status
+    return patch(
+        "shard_core.service.disk_full_notification.call_freeshard_controller",
+        mock_call,
+    )
+
+
+async def test_sends_email_when_over_threshold(app_client):
+    with _enabled_config(), _patch_relay() as mock_call:
+        await dfn.check_disk_full(_usage_at(95))
+
+    assert mock_call.await_count == 1
+    call = mock_call.await_args
+    assert call.args[0] == "api/email_relay"
+    assert call.kwargs["method"] == "POST"
+    payload = json.loads(call.kwargs["body"].decode())
+    assert isinstance(payload["body"], list)
+    assert "95%" in " ".join(payload["body"])
+
+
+async def test_triggers_at_exact_threshold(app_client):
+    with _enabled_config(90), _patch_relay() as mock_call:
+        await dfn.check_disk_full(_usage_at(90))
+
+    assert mock_call.await_count == 1
+
+
+async def test_does_not_trigger_just_below_threshold(app_client):
+    with _enabled_config(90), _patch_relay() as mock_call:
+        await dfn.check_disk_full(_usage_at(89))
+
+    assert not mock_call.called
+
+
+async def test_does_not_send_below_threshold(app_client):
+    with _enabled_config(), _patch_relay() as mock_call:
+        await dfn.check_disk_full(_usage_at(50))
+
+    assert not mock_call.called
+
+
+async def test_dedupe_only_sends_once(app_client):
+    with _enabled_config(), _patch_relay() as mock_call:
+        await dfn.check_disk_full(_usage_at(95))
+        await dfn.check_disk_full(_usage_at(96))
+        await dfn.check_disk_full(_usage_at(97))
+
+    assert mock_call.await_count == 1
+
+
+async def test_resets_after_dropping_below_threshold(app_client):
+    with _enabled_config(), _patch_relay() as mock_call:
+        await dfn.check_disk_full(_usage_at(95))  # sent
+        assert await dfn._was_notified() is True
+        await dfn.check_disk_full(_usage_at(50))  # re-armed
+        assert await dfn._was_notified() is False
+        await dfn.check_disk_full(_usage_at(95))  # sent again
+
+    assert mock_call.await_count == 2
+
+
+async def test_opt_out_when_disabled(app_client):
+    override = {"event_notifications": {"disk_full": {"enabled": False}}}
+    with settings_override(override), _patch_relay() as mock_call:
+        await dfn.check_disk_full(_usage_at(99))
+
+    assert not mock_call.called
+
+
+async def test_send_failure_does_not_set_flag_and_retries(app_client):
+    with _enabled_config(), _patch_relay() as mock_call:
+        mock_call.side_effect = RuntimeError("controller offline")
+        await dfn.check_disk_full(_usage_at(95))
+        assert await dfn._was_notified() is False
+
+        mock_call.side_effect = None
+        await dfn.check_disk_full(_usage_at(95))
+
+    assert mock_call.await_count == 2
+    assert await dfn._was_notified() is True
+
+
+async def test_error_status_does_not_set_flag(app_client):
+    with _enabled_config(), _patch_relay() as mock_call:
+        mock_call.return_value.raise_for_status.side_effect = HTTPError("429")
+        await dfn.check_disk_full(_usage_at(95))
+        assert await dfn._was_notified() is False
+
+
+async def test_ignores_unmeasured_disk(app_client):
+    with _enabled_config(), _patch_relay() as mock_call:
+        await dfn.check_disk_full(
+            DiskUsage(total_gb=0, free_gb=0, disk_space_low=False)
+        )
+
+    assert not mock_call.called
+
+
+async def test_run_check_uses_current_disk_snapshot(app_client, mocker):
+    """run_check is the background-task entry point; it must read the live snapshot."""
+    mocker.patch.object(disk, "current_disk_usage", _usage_at(95))
+    with _enabled_config(), _patch_relay() as mock_call:
+        await dfn.run_check()
+
+    assert mock_call.await_count == 1
+
+
+async def test_email_is_english(app_client):
+    with _enabled_config(), _patch_relay() as mock_call:
+        await dfn.check_disk_full(_usage_at(95))
+
+    payload = json.loads(mock_call.await_args.kwargs["body"].decode())
+    assert payload["subject"] == "Your Freeshard is running low on disk space"
+    assert payload["body"][0] == "Hello,"
+    assert any("will not receive another" in line for line in payload["body"])
+
+
+def test_both_templates_are_well_formed():
+    for lang in ("en", "de"):
+        template = dfn._TEMPLATES[lang]
+        assert template["subject"]
+        rendered = [line.format(used_percent=90) for line in template["body"]]
+        assert any("90%" in line for line in rendered)