Clear Background portal state when compositor IPC disconnects

Notify consumers when cached apps disappear and repopulate state from the next compositor snapshot. Cover disconnects, reconnects and duplicate signals on a private D-Bus session, and run the regression in CI.

Fixes #56

.github/workflows/workspace-checks.yml

@@ -50,6 +50,7 @@ jobs:
       - name: Test installer, session recovery and private portal contracts
         env:
           FERESE_TEST_PORTAL_INTERFACES: ${{ github.workspace }}/build/portal-contracts/data
+          FERESE_TEST_PORTAL_BACKGROUND: "1"
         run: python3 -m unittest discover -s scripts/tests -v
       - name: Test capture privacy, resize dependencies and cursor composition
         env:

portal/src/permissions.rs

@@ -149,6 +149,17 @@ impl Background {
                 }
             }
             .await;
+            // An interrupted generation is no longer authoritative. Clear it
+            // before waiting to reconnect, and signal only an actual change.
+            let had_apps = {
+                let mut states = self.states.write().await;
+                let had_apps = !states.is_empty();
+                states.clear();
+                had_apps
+            };
+            if had_apps {
+                let _ = Self::running_applications_changed(&emitter).await;
+            }
             if let Err(error) = result {
                 eprintln!("ferese portal: background event stream: {error}");
             }

scripts/tests/test_portal_background_isolated.py

@@ -0,0 +1,155 @@
+"""Background portal against a private bus and a fake compositor event stream.
+
+FERESE_TEST_PORTAL_BACKGROUND=1 python3 -m unittest discover -s scripts/tests -p test_portal_background_isolated.py
+No Wayland session or public D-Bus service is used.
+"""
+import json
+import os
+from pathlib import Path
+import socket
+import struct
+import subprocess
+import tempfile
+import threading
+import time
+import unittest
+
+
+@unittest.skipUnless(os.environ.get("FERESE_TEST_PORTAL_BACKGROUND") == "1", "private Background portal test is opt-in")
+class BackgroundPortal(unittest.TestCase):
+    def test_disconnect_clears_apps_signals_once_and_reconnect_replaces_state(self):
+        import gi
+        from gi.repository import Gio, GLib
+
+        repo = Path(__file__).resolve().parents[2]
+        binary = repo / os.environ.get("FERESE_TEST_PORTAL_BINARY", "target/debug/xdg-desktop-portal-ferese")
+        with tempfile.TemporaryDirectory(prefix="ferese-background-") as temporary:
+            root = Path(temporary)
+            runtime = root / "runtime"
+            runtime.mkdir(mode=0o700)
+            address_file = root / "control.sock"
+            listener = socket.socket(socket.AF_UNIX)
+            listener.bind(str(address_file))
+            listener.listen()
+            listener.settimeout(.1)
+            stop = threading.Event()
+            streams = []
+            failures = []
+
+            def read_exact(peer, count):
+                data = b""
+                while len(data) < count:
+                    part = peer.recv(count - len(data))
+                    if not part:
+                        raise EOFError("IPC peer closed")
+                    data += part
+                return data
+
+            def send(peer, value):
+                body = json.dumps(value).encode()
+                peer.sendall(struct.pack(">I", len(body)) + body)
+
+            def accept():
+                while not stop.is_set():
+                    try:
+                        peer, _ = listener.accept()
+                    except socket.timeout:
+                        continue
+                    except OSError:
+                        return
+                    try:
+                        peer.settimeout(2)
+                        size = struct.unpack(">I", read_exact(peer, 4))[0]
+                        self.assertLessEqual(size, 1024 * 1024)
+                        request = json.loads(read_exact(peer, size))
+                        if request["command"] != "event-stream":
+                            send(peer, {"version": 1, "id": request["id"], "error": {"code": "unsupported", "message": "test peer"}})
+                            peer.close()
+                            continue
+                        send(peer, {"version": 1, "id": request["id"], "result": {"version": 1}})
+                        streams.append(peer)
+                    except Exception as error:
+                        peer.close()
+                        failures.append(error)
+
+            worker = threading.Thread(target=accept, daemon=True)
+            worker.start()
+            env = dict(os.environ, XDG_RUNTIME_DIR=str(runtime), FERESE_SOCKET=str(address_file),
+                       XDG_CONFIG_HOME=str(root / "config"), XDG_CACHE_HOME=str(root / "cache"))
+            with (root / "portal.log").open("w+") as log:
+                bus = subprocess.Popen(["dbus-daemon", "--session", "--nofork", "--print-address=1"],
+                                       stdout=subprocess.PIPE, stderr=log, text=True)
+                env["DBUS_SESSION_BUS_ADDRESS"] = bus.stdout.readline().strip()
+                bus.stdout.close()
+                connection = Gio.DBusConnection.new_for_address_sync(env["DBUS_SESSION_BUS_ADDRESS"],
+                    Gio.DBusConnectionFlags.AUTHENTICATION_CLIENT | Gio.DBusConnectionFlags.MESSAGE_BUS_CONNECTION, None, None)
+                dbus = "org.freedesktop.DBus"
+                # GetAppState is restricted to the frontend's bus owner.
+                connection.call_sync(dbus, "/org/freedesktop/DBus", dbus, "RequestName",
+                    GLib.Variant("(su)", ("org.freedesktop.portal.Desktop", 0)), GLib.VariantType("(u)"),
+                    Gio.DBusCallFlags.NONE, 2000, None)
+                backend = subprocess.Popen([str(binary)], env=env, stdout=log, stderr=log)
+                name, path, interface = "org.freedesktop.impl.portal.desktop.ferese", "/org/freedesktop/portal/desktop", "org.freedesktop.impl.portal.Background"
+                changes = []
+                subscription = connection.signal_subscribe(name, interface, "RunningApplicationsChanged", path, None,
+                    Gio.DBusSignalFlags.NONE, lambda *args: changes.append(True))
+
+                def until(predicate, timeout=5):
+                    deadline = time.monotonic() + timeout
+                    while time.monotonic() < deadline:
+                        while GLib.MainContext.default().iteration(False):
+                            pass
+                        self.assertFalse(failures, str(failures))
+                        self.assertIsNone(backend.poll(), (root / "portal.log").read_text())
+                        value = predicate()
+                        if value:
+                            return value
+                        time.sleep(.01)
+                    self.fail((root / "portal.log").read_text())
+
+                def state():
+                    return connection.call_sync(name, path, interface, "GetAppState", GLib.Variant("()", ()),
+                        GLib.VariantType("(a{sv})"), Gio.DBusCallFlags.NO_AUTO_START, 2000, None).unpack()[0]
+
+                def snapshot(peer, app):
+                    send(peer, {"version": 1, "generation": 0, "last": True, "type": "snapshot", "desktop": {
+                        "outputs": [], "workspaces": [], "windows": [{"id": 1, "app_id": app, "focused": True}],
+                        "focus": {"window": 1}, "config": {"revision": 0, "error": None}, "theme": None,
+                        "lock": {"phase": "unlocked", "sleeping": False}}})
+
+                try:
+                    until(lambda: streams)
+                    snapshot(streams[0], "org.test.Gone")
+                    until(lambda: state().get("org.test.Gone") == 2 and len(changes) == 1)
+                    streams[0].shutdown(socket.SHUT_RDWR)
+                    streams[0].close()
+                    # Keep the replacement subscription idle until its snapshot.
+                    # The previous desktop must already be empty during this gap.
+                    until(lambda: not state() and len(changes) == 2)
+                    until(lambda: len(streams) >= 2)
+                    self.assertEqual(state(), {})
+                    self.assertEqual(len(changes), 2)
+                    snapshot(streams[1], "org.test.New")
+                    until(lambda: state() == {"org.test.New": 2} and len(changes) == 3)
+                    streams[1].shutdown(socket.SHUT_RDWR)
+                    streams[1].close()
+                    until(lambda: not state() and len(changes) == 4)
+                    # A second disconnect before receiving any snapshot is a no-op.
+                    until(lambda: len(streams) >= 3)
+                    streams[2].shutdown(socket.SHUT_RDWR)
+                    streams[2].close()
+                    until(lambda: len(streams) >= 4)
+                    self.assertEqual(state(), {})
+                    self.assertEqual(len(changes), 4)
+                finally:
+                    backend.terminate()
+                    backend.wait(timeout=5)
+                    connection.signal_unsubscribe(subscription)
+                    connection.close_sync(None)
+                    bus.terminate()
+                    bus.wait(timeout=5)
+                    stop.set()
+                    listener.close()
+                    worker.join(timeout=3)
+                    for peer in streams:
+                        peer.close()