vendor(qemu-ga): svos-dev's nh3-svos quiesce hook + its 6 controls beside the dispatcher

configs/qemu-ga/svos/50-svos

@@ -0,0 +1,121 @@
+#!/usr/bin/python3
+"""Stop the installed SVOS services for a VM backup, then restore their prior state."""
+import argparse
+import fcntl
+import json
+import os
+from pathlib import Path
+import stat
+import subprocess
+import sys
+import tempfile
+import time
+
+SERVICES = ('svos-kernel.service', 'svos-interface.service')
+
+class Quiesce:
+    def __init__(self, directory, *, run=None):
+        self.directory = Path(directory)
+        self.directory.mkdir(mode=0o700, exist_ok=True)
+        info = self.directory.lstat()
+        if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.geteuid() or stat.S_IMODE(info.st_mode) != 0o700:
+            raise RuntimeError('backup state directory must be private and owned by this user')
+        self.state = self.directory / 'state.json'
+        self.run = run or self._run
+
+    @staticmethod
+    def _run(args, **kwargs):
+        return subprocess.run(['systemctl', *args], check=True, text=True, capture_output=True, **kwargs)
+
+    def status(self, unit, deadline):
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            raise TimeoutError('service quiesce deadline exceeded')
+        answer = self.run(['show', unit, '--property=ActiveState,Result'], timeout=min(5, remaining))
+        fields = dict(line.split('=', 1) for line in answer.stdout.splitlines() if '=' in line)
+        return fields.get('ActiveState'), fields.get('Result')
+
+    def record(self, active):
+        with tempfile.NamedTemporaryFile(mode='w', dir=self.directory, delete=False) as output:
+            name = output.name
+            json.dump({'version': 1, 'services': active}, output)
+            output.flush()
+            os.fsync(output.fileno())
+        os.replace(name, self.state)
+        self.sync_directory()
+
+    def sync_directory(self):
+        fd = os.open(self.directory, os.O_RDONLY | os.O_DIRECTORY)
+        try:
+            os.fsync(fd)
+        finally:
+            os.close(fd)
+
+    def freeze(self):
+        if self.state.exists():
+            raise RuntimeError('previous backup state remains; thaw before retrying freeze')
+        deadline = time.monotonic() + 35
+        active = []
+        for unit in SERVICES:
+            state, _ = self.status(unit, deadline)
+            if state == 'active':
+                active.append(unit)
+            elif state not in ('inactive', 'failed'):
+                raise RuntimeError('service is transitioning: ' + unit)
+        self.record(active)
+        try:
+            if active:
+                self.run(['stop', *active], timeout=max(0.1, deadline - time.monotonic()))
+            for unit in active:
+                state, result = self.status(unit, deadline)
+                if state != 'inactive' or result != 'success':
+                    raise RuntimeError('service did not stop cleanly: ' + unit)
+        except BaseException:
+            if active:
+                try:
+                    self.run(['start', '--no-block', *active], timeout=5)
+                except Exception as error:
+                    print('backup recovery start refused: ' + type(error).__name__, file=sys.stderr)
+            # Keep the original state until thaw confirms recovery, even if freeze failed.
+            raise
+
+    def thaw(self):
+        if not self.state.exists():
+            return
+        record = json.loads(self.state.read_text())
+        active = record.get('services')
+        if (set(record) != {'version', 'services'} or record['version'] != 1 or
+                not isinstance(active, list) or any(unit not in SERVICES for unit in active) or
+                len(active) != len(set(active))):
+            raise RuntimeError('invalid backup recovery record')
+        deadline = time.monotonic() + 15
+        if active:
+            self.run(['start', *active], timeout=10)
+        for unit in active:
+            if self.status(unit, deadline)[0] != 'active':
+                raise RuntimeError('service did not resume: ' + unit)
+        self.state.unlink()
+        self.sync_directory()
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('action', choices=('freeze', 'thaw'))
+    parser.add_argument('--state-dir', default='/run/svos-backup')
+    args = parser.parse_args()
+    try:
+        hook = Quiesce(args.state_dir)
+        fd = os.open(hook.directory / 'lock', os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600)
+        try:
+            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            getattr(hook, args.action)()
+        finally:
+            os.close(fd)
+    except Exception as error:
+        print('SVOS backup ' + args.action + ' failed: ' + str(error), file=sys.stderr)
+        return 1
+    print('SVOS backup ' + args.action + ' complete')
+    return 0
+
+if __name__ == '__main__':
+    raise SystemExit(main())

configs/qemu-ga/svos/README.md

@@ -0,0 +1,12 @@
+# nh3-svos application-consistent backup hook (svos-dev, vendored 2026-10-05)
+
+- `50-svos` → installed as `/etc/qemu/fsfreeze-hook.d/50-svos` (root 0755) on nh3-svos. Authored by svos-dev
+  (SHA256 `f94cbba36fc410db2334a7c1fd552ffef60f16831f4e03501c53a4785970b556`, identical to the installed copy). On
+  freeze it records which SVOS services are active (root-private `/run/svos-backup/state.json`), stops them, and
+  requires `inactive` + `Result=success` within 35 s. On failure it queues a restart and keeps the record. On thaw
+  it restarts only those services, within 15 s, and keeps the record if a restart fails.
+- `test_quiesce.py`: the hook's six failure/recovery controls (`python3 -m unittest test_quiesce` from this
+  directory; the loader points at `50-svos` beside it).
+- Runs under `../fsfreeze-dispatch`, which qemu-ga calls via `/etc/qemu/qemu-ga.conf`. Without that config,
+  Debian's qemu-ga runs no hooks at all.
+- If the voice service is ever enabled, add it to `SERVICES` before enabling it (svos-dev).

configs/qemu-ga/svos/test_quiesce.py

@@ -0,0 +1,61 @@
+import importlib.machinery,importlib.util,json,subprocess,tempfile,unittest
+from pathlib import Path
+
+class QuiesceTests(unittest.TestCase):
+ def setUp(self):
+  self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup)
+  self.directory=Path(self.tmp.name)
+  # the installed hook has no .py suffix (50-svos), so give the loader explicitly; path is relative to this file
+  loader=importlib.machinery.SourceFileLoader('svos_quiesce',str(Path(__file__).with_name('50-svos')))
+  spec=importlib.util.spec_from_loader('svos_quiesce',loader)
+  self.module=importlib.util.module_from_spec(spec);spec.loader.exec_module(self.module)
+  self.states={'svos-kernel.service':('active','success'),'svos-interface.service':('inactive','success')}
+  self.calls=[];self.stop_fail=False;self.start_fail=False;self.dirty_stop=False
+  def run(args,**kwargs):
+   self.calls.append(args)
+   if args[0]=='show':
+    state,result=self.states[args[1]]
+    return subprocess.CompletedProcess(args,0,f'ActiveState={state}\nResult={result}\n','')
+   units=[x for x in args[1:] if not x.startswith('--')]
+   if args[0]=='stop':
+    self.states[units[0]]=('inactive','timeout' if self.dirty_stop else 'success')
+    if self.stop_fail:raise subprocess.TimeoutExpired(args,35)
+    for unit in units:self.states[unit]=('inactive','timeout' if self.dirty_stop else 'success')
+   elif args[0]=='start':
+    if self.start_fail:raise subprocess.CalledProcessError(1,args)
+    for unit in units:self.states[unit]=('active','success')
+   return subprocess.CompletedProcess(args,0,'','')
+  self.q=self.module.Quiesce(self.directory,run=run)
+ def test_restore_only_previously_active_services(self):
+  self.q.freeze();self.assertEqual(self.states['svos-kernel.service'][0],'inactive')
+  self.assertEqual(self.states['svos-interface.service'][0],'inactive')
+  self.q.thaw();self.assertEqual(self.states['svos-kernel.service'][0],'active')
+  self.assertEqual(self.states['svos-interface.service'][0],'inactive')
+  self.assertFalse(self.q.state.exists());self.q.thaw()
+ def test_stop_timeout_refuses_freeze_and_schedules_recovery(self):
+  self.stop_fail=True
+  with self.assertRaises(subprocess.TimeoutExpired):self.q.freeze()
+  self.assertEqual(self.states['svos-kernel.service'][0],'active')
+  self.assertTrue(self.q.state.exists())
+  self.q.thaw();self.assertFalse(self.q.state.exists())
+ def test_unclean_stop_is_not_reported_as_consistent(self):
+  self.dirty_stop=True
+  with self.assertRaises(RuntimeError):self.q.freeze()
+  self.assertEqual(self.states['svos-kernel.service'][0],'active')
+ def test_failed_thaw_preserves_recovery_record(self):
+  self.q.freeze();self.start_fail=True
+  with self.assertRaises(subprocess.CalledProcessError):self.q.thaw()
+  self.assertTrue(self.q.state.exists());self.start_fail=False
+  self.q.thaw();self.assertFalse(self.q.state.exists())
+ def test_repeated_freeze_cannot_overwrite_original_running_set(self):
+  self.q.freeze();before=self.q.state.read_bytes()
+  with self.assertRaises(RuntimeError):self.q.freeze()
+  self.assertEqual(self.q.state.read_bytes(),before)
+  self.q.thaw();self.assertEqual(self.states['svos-kernel.service'][0],'active')
+ def test_transition_refuses_before_any_stop(self):
+  self.states['svos-interface.service']=('deactivating','success')
+  with self.assertRaises(RuntimeError):self.q.freeze()
+  self.assertFalse(self.q.state.exists())
+  self.assertFalse(any(x[0]=='stop' for x in self.calls))
+
+if __name__=='__main__':unittest.main()