fix(backup,restore): harden the branch fixes and prove them with tests

Backup: a container that vanishes between the docker ps listing and the
swarm-task inspect (--rm one-shots, task-history GC) no longer aborts the
whole backup run; it counts as not stoppable and is skipped.

Restore: the postgres replay streams the dump through a spooled temp file
instead of buffering it three times in memory (multi-GB dumps OOMed the
restore mid-replay), and the superuser-only line filter is COPY-aware: data
rows inside COPY ... FROM stdin blocks pass through untouched, so a row
that happens to start with COMMENT ON EXTENSION or ALTER DEFAULT PRIVILEGES
is no longer silently dropped.

The e2e runner talks to the DinD daemon through docker exec instead of a
host-published tcp://127.0.0.1:2375: port publishing is unreachable from
sandboxed runners and from hosts with broken loopback publishing, and the
unencrypted root API port disappears from the host. The debug tmp dump
shrinks to tar plus docker cp against the DinD container itself.

New coverage: an e2e reproducing the swarm flake end to end (service task
on the volume, nothing whitelisted: the backup must succeed, the very same
task container must keep running, and the service must never replace a
task), unit tests for the COPY-aware filter, the swarm-task probe including
the vanished-container path, filter_stoppable ordering, and the one-session
FOREIGN_KEY_CHECKS drop assembly. Full suite: 35 unit, 9 integration,
30 e2e green.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

tests/unit/restore/test_mariadb_empty_drop.py

@@ -0,0 +1,44 @@
+import tempfile
+import unittest
+from unittest.mock import MagicMock, patch
+
+from baudolo.restore.db import mariadb as mariadb_mod
+
+
+class TestMariadbEmptyDrop(unittest.TestCase):
+    def test_drops_run_in_one_session_with_fk_checks_off(self) -> None:
+        calls = []
+
+        def _capture(container, argv, **kwargs):
+            calls.append(argv)
+            result = MagicMock()
+            result.stdout = b"users\nfetches\n"
+            return result
+
+        with tempfile.NamedTemporaryFile(suffix=".sql") as sql:
+            sql.write(b"CREATE TABLE t (id int);\n")
+            sql.flush()
+            with (
+                patch.object(mariadb_mod, "docker_exec", side_effect=_capture),
+                patch.object(mariadb_mod, "_pick_client", return_value="mariadb"),
+            ):
+                mariadb_mod.restore_mariadb_sql(
+                    container="db",
+                    db_name="mailu",
+                    user="mailu",
+                    password="pw",
+                    sql_path=sql.name,
+                    empty=True,
+                )
+
+        drop_calls = [argv for argv in calls if any("DROP TABLE" in a for a in argv)]
+        self.assertEqual(len(drop_calls), 1, f"expected ONE drop session: {calls}")
+        drop_sql = drop_calls[0][-1]
+        self.assertTrue(drop_sql.startswith("SET FOREIGN_KEY_CHECKS=0; "))
+        self.assertIn("DROP TABLE IF EXISTS `mailu`.`users`;", drop_sql)
+        self.assertIn("DROP TABLE IF EXISTS `mailu`.`fetches`;", drop_sql)
+        self.assertTrue(drop_sql.endswith("SET FOREIGN_KEY_CHECKS=1;"))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/restore/test_postgres_filter.py

@@ -0,0 +1,63 @@
+import unittest
+
+from baudolo.restore.db.postgres import filter_superuser_only_lines
+
+
+def _filter(raw: bytes) -> bytes:
+    return b"".join(filter_superuser_only_lines(raw.splitlines(keepends=True)))
+
+
+class TestFilterSuperuserOnlyLines(unittest.TestCase):
+    def test_drops_superuser_only_statements(self) -> None:
+        raw = (
+            b"CREATE TABLE t (id int);\n"
+            b"COMMENT ON EXTENSION pg_trgm IS 'trigram';\n"
+            b"ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT ALL ON TABLES TO x;\n"
+            b"INSERT INTO t VALUES (1);\n"
+        )
+        self.assertEqual(
+            _filter(raw),
+            b"CREATE TABLE t (id int);\nINSERT INTO t VALUES (1);\n",
+        )
+
+    def test_copy_data_rows_are_never_filtered(self) -> None:
+        raw = (
+            b"COPY public.snippets (body) FROM stdin;\n"
+            b"COMMENT ON EXTENSION looks like sql but is data\n"
+            b"ALTER DEFAULT PRIVILEGES stored as text\n"
+            b"\\.\n"
+            b"ALTER DEFAULT PRIVILEGES IN SCHEMA public REVOKE ALL ON TABLES FROM y;\n"
+        )
+        self.assertEqual(
+            _filter(raw),
+            b"COPY public.snippets (body) FROM stdin;\n"
+            b"COMMENT ON EXTENSION looks like sql but is data\n"
+            b"ALTER DEFAULT PRIVILEGES stored as text\n"
+            b"\\.\n",
+        )
+
+    def test_consecutive_copy_blocks_keep_state(self) -> None:
+        raw = (
+            b"COPY public.a (v) FROM stdin;\n"
+            b"row-a\n"
+            b"\\.\n"
+            b"COMMENT ON EXTENSION dropme IS 'x';\n"
+            b"COPY public.b (v) FROM stdin;\n"
+            b"COMMENT ON EXTENSION kept-as-data\n"
+            b"\\.\n"
+        )
+        out = _filter(raw)
+        self.assertNotIn(b"dropme", out)
+        self.assertIn(b"COMMENT ON EXTENSION kept-as-data\n", out)
+
+    def test_everything_else_passes_through_verbatim(self) -> None:
+        raw = (
+            b"SET statement_timeout = 0;\n"
+            b"CREATE EXTENSION IF NOT EXISTS pg_trgm;\n"
+            b"GRANT ALL ON SCHEMA public TO app;\n"
+        )
+        self.assertEqual(_filter(raw), raw)
+
+
+if __name__ == "__main__":
+    unittest.main()