Guard writer units against activation and fix backup cancellation

tests/test_backup.py

@@ -9,6 +9,7 @@ import json
 import os
 from pathlib import Path
 import shutil
+import signal
 import subprocess
 import tarfile
 import tempfile
@@ -126,6 +127,8 @@ class Integration(unittest.TestCase):
         (self.wp/'nested/data').mkdir(parents=True)
         (self.wp/'nested/data/keep.txt').write_text('nested data must survive')
         (self.nc/'occ').write_text('test stub')
+        self.runtime=self.base/'runtime'
+        self.runtime.mkdir()
         self.state=self.base/'state.json'
         self.initial=dict(services={s:'active' for s in ['httpd','php-fpm','gitea','postfix','dovecot','nc-cron']}, maintenance=False)
         self.state.write_text(json.dumps(self.initial))
@@ -142,16 +145,36 @@ with open(os.environ['TEST_EVENTS'],'a') as f: f.write(name+' '+' '.join(args)+'
 if name=='systemctl':
  cmd=args[0]
  service=args[1] if len(args)>1 else ''
+ runtime=Path(os.environ['TEST_RUNTIME'])
+ unit=service if service.endswith(('.service','.timer','.socket','.path')) else service+'.service'
+ files=sorted((runtime/'systemd/system'/f'{unit}.d').glob('*.conf'))
+ guards=[]
+ for file in files:
+  for line in file.read_text().splitlines():
+   if line.startswith('ConditionPathExists=!'): guards.append(Path(line.split('=!',1)[1]))
+ stamp_file=Path(str(p)+'.stamps')
+ stamps=json.loads(stamp_file.read_text()) if stamp_file.exists() else {}
  if cmd=='show':
-  if '--property=LoadState' in args: print('loaded' if service in s['services'] else 'not-found')
+  if '--property=LoadState' in args: print('masked' if service in s.get('masked',[]) else ('loaded' if service in s['services'] else 'not-found'))
+  elif '--property=Id' in args: print(unit)
+  elif '--property=DropInPaths' in args: print(' '.join(map(str,files)) if not os.environ.get('TEST_IGNORE_GUARD') else '')
+  elif '--property=ActiveEnterTimestampMonotonic' in args: print('ActiveEnterTimestampMonotonic='+str(stamps.get(service,1))+'\\nInactiveExitTimestampMonotonic='+str(stamps.get(service,1)))
   elif '--property=Environment' in args or '--property=TriggeredBy' in args: print('')
   else: print(s['services'].get(service,'inactive'))
  elif cmd=='is-active': sys.exit(0 if s['services'].get(args[-1])=='active' else 3)
  elif cmd in ('stop','start'):
   if os.environ.get('TEST_FAIL_SERVICE')==cmd+':'+service: sys.exit(1)
+  if cmd=='start':
+   if service in s.get('masked',[]): sys.exit(1)
+   if any(g.exists() for g in guards):
+    with open(os.environ['TEST_EVENTS'],'a') as f: f.write('guard blocked '+service+'\\n')
+    sys.exit(0) # systemd skips a failed Condition without starting the unit
+   stamps[service]=stamps.get(service,1)+1
+   stamp_file.write_text(json.dumps(stamps))
   s['services'][service]='active' if cmd=='start' else 'inactive'
  elif cmd=='enable': s['services'][args[-1]]='active'
- elif cmd=='daemon-reload': pass
+ elif cmd=='daemon-reload':
+  if os.environ.get('TEST_FAIL_GUARD_RELEASE') and list(runtime.glob('app-backup-guard-*')) and not list(runtime.glob('systemd/system/*/zzzz-app-backup-*.conf')): sys.exit(1)
 elif name=='sudo':
  os.execvp(args[2],args[2:])
 elif name=='php':
@@ -181,7 +204,8 @@ exec {real} "$@"
         self.cnf.chmod(0o600)
         self.sql('DROP DATABASE IF EXISTS backup_test; CREATE DATABASE backup_test; CREATE TABLE backup_test.t (id INT PRIMARY KEY, body TEXT) ENGINE=MyISAM; INSERT INTO backup_test.t VALUES (1,"original");')
         self.config=self.base/'config'
-        self.config.write_text(f'''WORKDIR='{self.base}/work'
+        self.config.write_text(f'''SERVICE_GUARD_RUNTIME_ROOT='{self.runtime}'
+WORKDIR='{self.base}/work'
 LOG_DIR='{self.base}/logs'
 LOCKFILE='{self.base}/lock'
 MIN_FREE_GB=0
@@ -212,7 +236,7 @@ GITEA_DB_CNF='{self.cnf}'
 QUIESCE_SERVICES=(nc-cron)
 ''')
         self.config.chmod(0o600)
-        self.env=dict(os.environ, PATH=str(self.bin)+':'+os.environ['PATH'],TEST_STATE=str(self.state),TEST_EVENTS=str(self.events))
+        self.env=dict(os.environ,TEST_RUNTIME=str(self.runtime), PATH=str(self.bin)+':'+os.environ['PATH'],TEST_STATE=str(self.state),TEST_EVENTS=str(self.events))
 
     def run_script(self, script, *args, ok=True, **env):
         proc=subprocess.run(['bash',str(REPO/'usr-local-sbin'/script),'--config',str(self.config),*map(str,args)],env=dict(self.env,**env),text=True,stdout=subprocess.PIPE,stderr=subprocess.STDOUT,timeout=90)
@@ -312,6 +336,77 @@ if '--exclude=/*' in sys.argv:
         run=self.backup()
         self.run_script('app-restore.sh','--local-run',run,'--components','wordpress','--apply','--yes')
 
+    def test_start_guard_blocks_monitor_during_copy_and_preserves_inactive_writer(self):
+        state=dict(services=dict(self.initial['services'],gitea='inactive'),maintenance=False)
+        self.state.write_text(json.dumps(state))
+        wrapper=self.bin/'rsync'
+        wrapper.write_text('''#!/usr/bin/env bash
+if [[ " $* " != *" --list-only "* && " $* " != *" --exclude=/* "* ]]; then
+ systemctl start httpd
+ systemctl start php-fpm
+ systemctl start gitea
+fi
+exec /usr/bin/rsync "$@"
+''')
+        self.backup()
+        self.assertEqual(json.loads(self.state.read_text()),state)
+        for unit in ['httpd','php-fpm','gitea']:
+            self.assertIn('guard blocked '+unit,self.events.read_text())
+        self.assertFalse(list(self.runtime.glob('systemd/system/*/zzzz-app-backup-*.conf')))
+        self.assertFalse(list(self.runtime.glob('app-backup-guard-*')))
+
+    def inject_writer_restart(self, final_state):
+        wrapper=self.bin/'rsync'
+        wrapper.write_text('''#!/usr/bin/env python3
+import json,os,subprocess,sys
+from pathlib import Path
+result=subprocess.run(['/usr/bin/rsync',*sys.argv[1:]])
+if result.returncode: sys.exit(result.returncode)
+if '--list-only' not in sys.argv and '--exclude=/*' not in sys.argv:
+ p=Path(os.environ['TEST_STATE'])
+ state=json.loads(p.read_text())
+ state['services']['httpd']='''+repr(final_state)+'''
+ p.write_text(json.dumps(state))
+ p=Path(str(p)+'.stamps')
+ stamps=json.loads(p.read_text()) if p.exists() else {}
+ stamps['httpd']=stamps.get('httpd',1)+1
+ p.write_text(json.dumps(stamps))
+''')
+        result=self.run_script('app-backup.sh',ok=False)
+        self.assertIn('Consistency lost:',result.stdout)
+        self.assertFalse(list((self.base/'work/archives').glob('appbackup_*')))
+        self.assertEqual(json.loads(self.state.read_text()),self.initial)
+        self.assertFalse(list(self.runtime.glob('systemd/system/*/zzzz-app-backup-*.conf')))
+
+    def test_running_writer_prevents_backup_publication(self):
+        self.inject_writer_restart('active')
+
+    def test_restarted_then_stopped_writer_prevents_backup_publication(self):
+        self.inject_writer_restart('inactive')
+
+    def test_unloaded_start_guard_aborts_before_service_stop(self):
+        result=self.run_script('app-backup.sh',ok=False,TEST_IGNORE_GUARD='1')
+        self.assertIn('Start guard not loaded last',result.stdout)
+        self.assertNotIn('systemctl stop',self.events.read_text())
+        self.assertEqual(json.loads(self.state.read_text()),self.initial)
+        self.assertFalse(list(self.runtime.glob('app-backup-guard-*')))
+
+    def test_existing_service_mask_is_preserved(self):
+        state=dict(services=dict(self.initial['services'],gitea='inactive'),maintenance=False,masked=['gitea'])
+        self.state.write_text(json.dumps(state))
+        self.backup()
+        self.assertEqual(json.loads(self.state.read_text()),state)
+        self.assertNotIn('systemctl start gitea',self.events.read_text())
+
+    def test_guard_release_failure_keeps_recovery_and_blocks_publication(self):
+        result=self.run_script('app-backup.sh',ok=False,TEST_FAIL_GUARD_RELEASE='1')
+        self.assertIn('Could not release writer start guards',result.stdout)
+        self.assertFalse(list((self.base/'work/archives').glob('appbackup_*')))
+        self.assertTrue(json.loads(self.state.read_text())['maintenance'])
+        self.assertTrue(all(s=='inactive' for s in json.loads(self.state.read_text())['services'].values()))
+        recovery=json.loads(next((self.base/'work').glob('recovery-*.json')).read_text())
+        self.assertTrue(Path(recovery['service_guard_directory']).is_dir())
+
     def test_maintenance_and_inactive_service_preserved(self):
         state=dict(services=dict(self.initial['services'],gitea='inactive'),maintenance=True)
         self.state.write_text(json.dumps(state))
@@ -341,6 +436,9 @@ if '--exclude=/*' in sys.argv:
         self.assertIn('Partial restore',result.stdout)
         self.assertTrue(json.loads(self.state.read_text())['maintenance'])
         self.assertEqual(json.loads(self.state.read_text())['services']['httpd'],'inactive')
+        recovery=json.loads(next((self.base/'work').glob('recovery-*.json')).read_text())
+        self.assertTrue(recovery['service_guard_files'])
+        self.assertTrue(all(Path(p).is_file() for p in recovery['service_guard_files']))
 
     def test_check_does_not_claim_successful_backup(self):
         self.run_script('app-backup.sh','--check')
@@ -675,6 +773,7 @@ from pathlib import Path
 args=sys.argv[1:]
 assert '--unit=app-backup-upgrade' in args
 assert '--property=Type=oneshot' in args
+assert '--property=KillMode=control-group' in args
 assert '--property=SELinuxContext=system_u:system_r:unconfined_service_t:s0' in args
 command=args[args.index('/bin/bash'):]
 state=Path(command[-1])
@@ -682,6 +781,7 @@ if os.environ.get('TEST_SYSTEMD_SUBMIT') == 'reject':
  sys.exit('systemd refused submission')
 with open(state/'test-worker.log','w') as log:
  worker=subprocess.Popen(command,stdin=subprocess.DEVNULL,stdout=log,stderr=subprocess.STDOUT,env=os.environ,close_fds=True,start_new_session=True)
+(state/'test-worker.pid').write_text(str(worker.pid))
 if os.environ.get('TEST_SYSTEMD_SUBMIT') == 'accepted-error':
  sys.exit('simulated reply failure after systemd accepted the job')
 # Type=oneshot remains starting until ExecStart exits. Real systemd-run waits
@@ -696,6 +796,52 @@ if '--no-block' not in args:
 ''')
         launcher.chmod(0o755)
 
+    def assert_group_term_recovers_services(self, tool):
+        self.configure_upgrade()
+        self.systemd_launcher()
+        wrapper=self.bin/tool
+        if tool=='rsync':
+            match='[[ " $* " != *" --list-only "* && " $* " != *" --exclude=/* "* ]]'
+        else:
+            match='[[ " $* " == *" copy "* ]]'
+        wrapper.write_text('#!/usr/bin/env bash\nif '+match+'; then\n echo waiting > "$TEST_EVENTS.waiting"\n sleep 120\nfi\nexec /usr/bin/'+tool+' "$@"\n')
+        package=self.build_package()
+        result=subprocess.run(['bash',str(package)],env=self.env,capture_output=True,text=True,timeout=60)
+        self.assertEqual(result.returncode,0,result.stdout+result.stderr)
+        state=Path(result.stdout.split('Update-Sicherung und Protokoll: ')[1].splitlines()[0])
+        pid=int((state/'test-worker.pid').read_text())
+        try:
+            for _ in range(400):
+                if Path(str(self.events)+'.waiting').exists(): break
+                time.sleep(.1)
+            else: self.fail('Worker did not reach '+tool+': '+(state/'test-worker.log').read_text())
+            os.killpg(pid,signal.SIGTERM) # emulate systemd KillMode=control-group
+            for _ in range(200):
+                status=(state/'result.txt').read_text()
+                if status.startswith('FAILED'): break
+                time.sleep(.1)
+            self.assertTrue(status.startswith('FAILED'),status+(state/'test-worker.log').read_text())
+            actual=json.loads(self.state.read_text())
+            self.assertFalse(actual['maintenance'])
+            for unit in self.initial['services']:
+                self.assertEqual(actual['services'][unit],'active',unit)
+            self.assertEqual(actual['services']['app-backup.timer'],'inactive')
+            if tool=='rsync':
+                self.assertFalse(list((self.base/'work/archives').glob('appbackup_*')))
+            else:
+                self.assertFalse(list((self.base/'work/archives').glob('appbackup_*/UPLOADED')))
+            self.assertFalse(list(self.runtime.glob('app-backup-guard-*')))
+            self.assertIn('Service started: httpd',(state/'update.log').read_text())
+        finally:
+            try: os.killpg(pid,signal.SIGKILL)
+            except ProcessLookupError: pass
+
+    def test_group_term_interrupts_worker_copy_and_recovers_services(self):
+        self.assert_group_term_recovers_services('rsync')
+
+    def test_group_term_interrupts_worker_upload_and_keeps_timer_stopped(self):
+        self.assert_group_term_recovers_services('rclone')
+
     def test_detached_package_handoff_finishes_after_starter_exits(self):
         self.configure_upgrade()
         self.systemd_launcher()