fix(indeehub): prove idle legacy worker termination before backup

docs/managed-update-recovery-implementation.md

@@ -293,3 +293,35 @@ or widening manager write access. The controller uses that fixed command now;
 25 pure controller tests pass, including refusal before fence creation on dump
 failure. A new helper hash requires a matching backend rebuild. Candidate-helper
 qualification remains separate from actual backend transaction acceptance.
+
+### Legacy worker termination qualification — 2026-10-07
+
+The held disposable-VM operation `4320fe90-8ab5-4496-a4e7-cf11ca0fd376`
+exposed the original worker's Node-as-PID1 SIGTERM behavior. A no-network,
+no-volume probe from its exact recovery image exited 137 without init and 143
+with init. This is not graceful shutdown or proof of completed jobs.
+
+The controller now requires an exact six-field nonnegative integer queue
+observation; absent `active` can no longer imply idle. Its narrowly scoped legacy
+worker path records **forced idle termination**, with `graceful=false` and
+`completed_work_claim=false`, only after paused all-zero observations before and
+after, closed ingress/frontend, exact operation/original/recovery image and known
+command, unchanged saved unit, original stop intent and died event, and proof the
+process is dead. Other writers' 137 exits remain refused. All 31 pure controller
+regressions passed, including missing/nonzero counts, reopened queue, changed
+identity, command, unit and process state. Actual worker-only classification passed
+under the hardened service → user scope with the lifecycle lock held.
+
+The same fixture's API has npm as PID1 and one direct `node dist/main` child.
+After fresh empty-business-state proof and durable stop intent, an exact-command,
+parent-validated SIGTERM to that child stopped the original container; npm emitted
+exit 1. The existing clean-exit gate correctly retained the hold. API termination
+classification is still under review; no generic exit-1 allowance was added.
+Frontend/worker are confirmed stopped; API is stopped but unconfirmed; storage
+members remain running. Backup/fresh-restore, full RPC rollback/success and live
+activation are **not passed**. Installed pinned helper remains unchanged; this is
+private candidate-helper qualification only. Live Yaya remains unchanged.
+
+Future worker image source now includes idempotent SIGTERM/SIGINT shutdown in app
+commit `29627fc` with four passing Jest tests. Its image has not been built; the
+previous frontend/API-only candidate catalog cannot cover that new worker image.

scripts/indeehub-maintenance-controller.py

@@ -7,6 +7,9 @@ import datetime, hashlib, json, os, pathlib, re, shutil, subprocess, sys, time,
 NAMES = ('indeedhub','indeedhub-api','indeedhub-ffmpeg','indeedhub-minio','indeedhub-postgres','indeedhub-redis','indeedhub-relay')
 VOLUMES = ('indeedhub-minio-data','indeedhub-postgres-data','indeedhub-redis-data','indeedhub-relay-data')
 DATA = pathlib.Path('/var/lib/archipelago')
+QUEUE_COUNTS = frozenset(('active','waiting','paused','delayed','failed','completed'))
+def valid_queue_counts(counts):
+    return isinstance(counts,dict) and set(counts)==QUEUE_COUNTS and all(type(v) is int and v>=0 for v in counts.values())
 QUEUE_SCRIPT = r'''const {Queue}=require('bullmq');
 (async()=>{const q=new Queue('transcode',{connection:{host:process.env.QUEUE_HOST,port:Number(process.env.QUEUE_PORT||6379),password:process.env.QUEUE_PASSWORD,maxRetriesPerRequest:1}});
 try{const action=process.argv[1];if(action==='pause')await q.pause();else if(action==='resume')await q.resume();else if(action!=='status')throw Error('action');
@@ -193,8 +196,7 @@ class Controller:
         self.record['ingress_closed']=True;self.save()
     def queue(self, action):
         result=json.loads(self.run(['podman','exec','indeedhub-api','node','-e',QUEUE_SCRIPT,action]))
-        require(type(result.get('paused')) is bool and isinstance(result.get('counts'),dict),'Invalid queue observation')
+        require(type(result.get('paused')) is bool and valid_queue_counts(result.get('counts')),'Incomplete or invalid queue observation')
-        for value in result['counts'].values():require(type(value) is int and value>=0,'Invalid job count')
         return result
     def pause_queue(self):
         if 'queue_was_paused' not in self.record:
@@ -207,7 +209,7 @@ class Controller:
         # completed: this path requires a fresh empty store behind closed ingress.
         require(self.record.get('stopped',{}).get('indeedhub',{}).get('confirmed'),'Frontend ingress must already be stopped')
         require(self.record.get('stopped',{}).get('indeedhub-ffmpeg',{}).get('confirmed'),'Transcode worker must already be stopped')
-        require(self.record.get('queue_pause_confirmed') is True and self.record.get('last_queue_counts',{}).get('active')==0,'Worker queue is not proven idle')
+        require(self.record.get('queue_pause_confirmed') is True and valid_queue_counts(self.record.get('last_queue_counts')) and self.record['last_queue_counts']['active']==0,'Worker queue is not proven idle')
         tables=('projects','contents','payments','shareholders','subscriptions','library_items')
         fields=','.join("'%s',(SELECT count(*) FROM public.%s)"%(name,name) for name in tables)
         sql="SELECT json_build_object("+fields+",'other_active_transactions',(SELECT count(*) FROM pg_stat_activity WHERE datname=current_database() AND pid<>pg_backend_pid() AND state<>'idle'))"
@@ -215,6 +217,41 @@ class Controller:
         require(set(counts)==set(tables)|{'other_active_transactions'},'Legacy API business-state observation incomplete')
         require(all(type(value) is int and value==0 for value in counts.values()),'Legacy API has business work or active transactions; completion cannot be inferred')
         self.record['legacy_api_empty_state']=counts;self.save()
+    def legacy_idle_worker_termination(self, member, properties):
+        # Compatibility only for the observed original Node-as-PID1 worker.
+        # A timeout/SIGKILL is never renamed graceful or completed work.
+        require(member['name']=='indeedhub-ffmpeg','Forced termination is not allowed for this writer')
+        self.holds();self.fence_matches()
+        before=self.record.get('last_queue_counts')
+        require(self.record.get('ingress_closed') is True and self.record.get('stopped',{}).get('indeedhub',{}).get('confirmed') is True,'Legacy worker ingress is not closed')
+        require(self.record.get('queue_pause_confirmed') is True and valid_queue_counts(before) and all(v==0 for v in before.values()),'Legacy worker queue was not proven completely empty')
+        stopped=self.record['stopped'][member['name']]
+        require(stopped.get('container_id')==member['container_id'] and type(stopped.get('intent_at')) in (int,float),'Legacy worker stop intent changed')
+        state=dict(line.split('=',1) for line in properties.splitlines() if '=' in line)
+        require(state.get('ActiveState') in ('inactive','failed') and state.get('SubState') in ('dead','failed') and state.get('ExecMainStatus')=='137','Legacy worker process death is not established')
+        runtime=json.loads((self.data/'update-transactions'/'supervised'/(self.operation+'.json')).read_text())
+        require(runtime.get('id')==self.operation and runtime.get('phase') in ('Editing','Restoring') and runtime.get('target_startup_began') is False,'Legacy worker operation changed')
+        records=[m for m in runtime['members'] if m['original']['name']==member['name']]
+        require(len(records)==1,'Ambiguous legacy worker recovery identity')
+        original=records[0]['original'];recovery=records[0]['recovery_image']
+        require(original['container_id']==member['container_id'] and original['image'].removeprefix('sha256:')==member['image_id'].removeprefix('sha256:') and original['config_sha256']==member['config_sha256'] and hashlib.sha256(original['body'].encode()).hexdigest()==member['unit_sha256'],'Legacy worker original identity changed')
+        require(recovery['source_container_id']==member['container_id'] and recovery['operation_id']==self.operation,'Legacy worker recovery image belongs to another operation')
+        rows=json.loads(self.run(['podman','image','inspect',recovery['image']]))
+        require(len(rows)==1 and rows[0]['Id'].removeprefix('sha256:')==recovery['image'].removeprefix('sha256:'),'Legacy worker recovery image changed')
+        image=rows[0];config=image['Config']
+        require(config.get('Cmd')==['node','dist/ffmpeg-worker/worker.js'] and config.get('Entrypoint')==['docker-entrypoint.sh'],'Unrecognized legacy worker command')
+        created=datetime.datetime.fromisoformat(image['Created'].replace('Z','+00:00')).timestamp()
+        require(created<=stopped['intent_at'],'Legacy worker recovery image was not captured before stop')
+        source=pathlib.Path(self.run(['systemctl','--user','show',member['name']+'.service','--property=SourcePath','--value']).decode().strip())
+        require(source.is_file() and not source.is_symlink() and source.suffix=='.container' and source.stat().st_uid==os.getuid() and sha(source)==member['unit_sha256'],'Legacy worker saved unit changed')
+        rows=self.run(['podman','ps','--all','--no-trunc','--filter','id='+member['container_id'],'--format','{{.ID}} {{.State}}']).decode().splitlines()
+        require(not rows or rows==[member['container_id']+' exited'],'Legacy worker may still be running')
+        after=self.queue('status')
+        require(after['paused'] is True and all(v==0 for v in after['counts'].values()),'Legacy worker queue reopened or changed')
+        return {'classification':'legacy-idle-worker-forced-termination','graceful':False,'completed_work_claim':False,
+                'original_container_id':member['container_id'],'original_image_id':member['image_id'],
+                'recovery_image_id':recovery['image'],'unit_sha256':member['unit_sha256'],
+                'before_counts':before,'after_counts':after['counts'],'process_dead':True}
     def graceful_stop(self, name):
         # Save the obligation before systemd can remove an AutoRemove container.
         stopped=self.record.setdefault('stopped',{})
@@ -225,7 +262,6 @@ class Controller:
             stopped[name]={'intent_at':time.time(),'container_id':actual['Id']};self.save()
         self.run(['systemctl','--user','stop',name+'.service'],timeout=180)
         properties=self.run(['systemctl','--user','show',name+'.service','--property=ActiveState,SubState,Result,ExecMainStatus']).decode()
-        require('ActiveState=inactive' in properties and 'Result=success' in properties,'Service did not stop successfully')
         # --rm removes inspect state. Require a persisted Podman died event for
         # this exact original ID; a forced SIGKILL is never called completed work.
         events=self.run(['podman','events','--stream=false','--since',str(int(stopped[name]['intent_at'])-1),'--filter','container='+member['container_id'],'--filter','event=died','--format','json']).decode().splitlines()
@@ -233,10 +269,13 @@ class Controller:
         matching=[event for event in matching if event.get('ID',event.get('id'))==member['container_id']]
         require(matching,'Original process exit evidence unavailable; hold retained')
         code=matching[-1].get('ContainerExitCode',matching[-1].get('containerExitCode'))
-        idle_worker = name=='indeedhub-ffmpeg' and self.record.get('queue_pause_confirmed') is True and self.record.get('last_queue_counts',{}).get('active')==0
+        idle_worker = name=='indeedhub-ffmpeg' and self.record.get('queue_pause_confirmed') is True and valid_queue_counts(self.record.get('last_queue_counts')) and self.record['last_queue_counts']['active']==0
         empty_api = name=='indeedhub-api' and self.record.get('legacy_api_empty_state') is not None
-        require(str(code)=='0' or (str(code)=='143' and (idle_worker or empty_api)),'Original process did not exit cleanly; active work is not claimed completed')
+        forced=self.legacy_idle_worker_termination(member,properties) if str(code)=='137' else None
-        classification=('idle-worker-terminated-after-queue-drain' if idle_worker else 'empty-business-store-legacy-api-terminated') if str(code)=='143' else 'clean-process-exit'
+        require(forced is not None or ('ActiveState=inactive' in properties and 'Result=success' in properties),'Service did not stop successfully')
+        require(forced is not None or str(code)=='0' or (str(code)=='143' and (idle_worker or empty_api)),'Original process did not exit cleanly; active work is not claimed completed')
+        classification=forced['classification'] if forced else (('idle-worker-terminated-after-queue-drain' if idle_worker else 'empty-business-store-legacy-api-terminated') if str(code)=='143' else 'clean-process-exit')
+        if forced:stopped[name]['legacy_idle_termination']=forced
         stopped[name].update(confirmed=True,exit_code=int(code),classification=classification,confirmed_at=time.time());self.save()
     def volume_sources(self):
         expected=VOLUMES
@@ -313,7 +352,7 @@ class Controller:
             while True:
                 state=self.queue('status');require(state['paused'],'Worker admission reopened')
                 self.record['last_queue_counts']=state['counts'];self.save()
-                if state['counts'].get('active',0)==0:break
+                if state['counts']['active']==0:break
                 require(time.monotonic()<deadline,'Transcodes still active; retained job state, no forced completion')
                 time.sleep(1)
             self.graceful_stop('indeedhub-ffmpeg');self.legacy_api_idle();self.graceful_stop('indeedhub-api')

tests/regression/test_indeehub_maintenance_controller.py

@@ -135,7 +135,7 @@ class MaintenanceTests(unittest.TestCase):
    if argv[:2]==['podman','events']:return json.dumps({'ID':members()[0]['container_id'],'ContainerExitCode':137}).encode()
    raise AssertionError(argv)
   c.runner=command
-  with self.assertRaisesRegex(RuntimeError,'did not exit cleanly'):c.graceful_stop('indeedhub')
+  with self.assertRaisesRegex(RuntimeError,'not allowed for this writer'):c.graceful_stop('indeedhub')
   self.assertFalse(c.record['stopped']['indeedhub'].get('confirmed',False))
   self.assertTrue((self.root/'update-transactions'/'holds'/'indeedhub').exists())
  def test_failed_stop_can_restore_without_fabricating_completed_drain(self):
@@ -183,6 +183,82 @@ class MaintenanceTests(unittest.TestCase):
   c.fence.parent.mkdir(parents=True);c.fence.write_text(self.operation)
   with self.assertRaisesRegex(RuntimeError,'without journal'):c.release('aborted')
   self.assertTrue(c.fence.exists())
+ def test_queue_requires_complete_nonnegative_integer_observations(self):
+  valid={name:0 for name in module.QUEUE_COUNTS}
+  for counts in ({}, {k:v for k,v in valid.items() if k!='active'},
+                 dict(valid,active=-1),dict(valid,active=True),dict(valid,unknown=0)):
+   with self.subTest(counts=counts):
+    self.controller.runner=lambda argv,timeout,output:json.dumps({'paused':True,'counts':counts}).encode()
+    with self.assertRaisesRegex(RuntimeError,'queue observation'):self.controller.queue('status')
+    self.assertIsNone(self.controller.record)
+  self.controller.runner=lambda argv,timeout,output:json.dumps({'paused':True,'counts':valid}).encode()
+  self.assertEqual(self.controller.queue('status')['counts'],valid)
+ def legacy_forced_fixture(self):
+  c=self.controller;unit=self.root/'worker.container';unit.write_text('[Container]\nImage=original\n')
+  worker=next(m for m in members() if m['name']=='indeedhub-ffmpeg');worker['unit_sha256']=module.sha(unit)
+  counts={name:0 for name in module.QUEUE_COUNTS}
+  c.record={'ingress_closed':True,'queue_pause_confirmed':True,'last_queue_counts':counts,
+   'stopped':{'indeedhub':{'confirmed':True},worker['name']:{'container_id':worker['container_id'],'intent_at':1700000000}}}
+  c.fence.parent.mkdir(parents=True);c.fence.write_text(self.operation)
+  original={'name':worker['name'],'container_id':worker['container_id'],'image':worker['image_id'],'config_sha256':worker['config_sha256'],'body':unit.read_text()}
+  recovery={'source_container_id':worker['container_id'],'operation_id':self.operation,'image':'d'*64}
+  runtime={'id':self.operation,'phase':'Restoring','target_startup_began':False,'members':[{'original':original,'recovery_image':recovery}]}
+  runtimepath=c.data/'update-transactions'/'supervised'/(self.operation+'.json');module.atomic(runtimepath,runtime)
+  image={'Id':'d'*64,'Created':'2023-01-01T00:00:00Z','Config':{'Cmd':['node','dist/ffmpeg-worker/worker.js'],'Entrypoint':['docker-entrypoint.sh']}}
+  observed={'paused':True,'counts':dict(counts)};state={'ps':b''}
+  def runner(argv,timeout,output):
+   if argv[:3]==['podman','image','inspect']:return json.dumps([image]).encode()
+   if argv[:3]==['systemctl','--user','show']:return str(unit).encode()
+   if argv[:2]==['podman','ps']:return state['ps']
+   if argv[:2]==['podman','exec']:return json.dumps(observed).encode()
+   raise AssertionError(argv)
+  c.runner=runner
+  props='ActiveState=failed\nSubState=failed\nExecMainStatus=137\nResult=exit-code\n'
+  return c,worker,props,runtime,runtimepath,image,observed,state,unit
+ def test_legacy_forced_idle_proof_is_explicit_and_does_not_claim_graceful_work(self):
+  c,w,p,*_=self.legacy_forced_fixture();proof=c.legacy_idle_worker_termination(w,p)
+  self.assertEqual(proof['classification'],'legacy-idle-worker-forced-termination')
+  self.assertFalse(proof['graceful']);self.assertFalse(proof['completed_work_claim']);self.assertTrue(proof['process_dead'])
+ def test_legacy_forced_idle_rejects_missing_active_or_queued_work_before_stop(self):
+  c,w,p,*_=self.legacy_forced_fixture();valid=dict(c.record['last_queue_counts'])
+  for field in module.QUEUE_COUNTS:
+   for bad in ({k:v for k,v in valid.items() if k!=field},dict(valid,**{field:1})):
+    c.record['last_queue_counts']=bad
+    with self.assertRaises(RuntimeError):c.legacy_idle_worker_termination(w,p)
+ def test_legacy_forced_idle_rejects_reopened_missing_or_nonempty_after_queue(self):
+  c,w,p,r,rp,i,after,*_=self.legacy_forced_fixture();valid=dict(after['counts'])
+  after['paused']=False
+  with self.assertRaises(RuntimeError):c.legacy_idle_worker_termination(w,p)
+  after['paused']=True
+  for field in module.QUEUE_COUNTS:
+   for bad in ({k:v for k,v in valid.items() if k!=field},dict(valid,**{field:1})):
+    after['counts']=bad
+    with self.assertRaises(RuntimeError):c.legacy_idle_worker_termination(w,p)
+ def test_legacy_forced_idle_rejects_changed_original_recovery_and_operation(self):
+  import copy
+  c,w,p,r,rp,*_=self.legacy_forced_fixture()
+  mutations=[lambda d:d.update(id='foreign'),lambda d:d.update(target_startup_began=True),lambda d:d.update(phase='Committed')]
+  for key in ('container_id','image','config_sha256','body'):
+   mutations.append(lambda d,k=key:d['members'][0]['original'].update({k:'changed'}))
+  for key in ('source_container_id','operation_id'):
+   mutations.append(lambda d,k=key:d['members'][0]['recovery_image'].update({k:'changed'}))
+  for mutate in mutations:
+   damaged=copy.deepcopy(r);mutate(damaged);module.atomic(rp,damaged)
+   with self.assertRaises(RuntimeError):c.legacy_idle_worker_termination(w,p)
+ def test_legacy_forced_idle_rejects_unproven_process_command_unit_and_ingress(self):
+  c,w,p,r,rp,image,after,state,unit=self.legacy_forced_fixture()
+  for field,value in [('Id','e'*64),('Created','2030-01-01T00:00:00Z'),('Config',{'Cmd':['other'],'Entrypoint':['docker-entrypoint.sh']})]:
+   old=image[field];image[field]=value
+   with self.assertRaises(RuntimeError):c.legacy_idle_worker_termination(w,p)
+   image[field]=old
+  state['ps']=(w['container_id']+' running').encode()
+  with self.assertRaises(RuntimeError):c.legacy_idle_worker_termination(w,p)
+  state['ps']=b''
+  with self.assertRaises(RuntimeError):c.legacy_idle_worker_termination(w,p.replace('ExecMainStatus=137','ExecMainStatus=0'))
+  c.record['ingress_closed']=False
+  with self.assertRaises(RuntimeError):c.legacy_idle_worker_termination(w,p)
+  c.record['ingress_closed']=True;unit.write_text('changed')
+  with self.assertRaises(RuntimeError):c.legacy_idle_worker_termination(w,p)
  def test_legacy_worker_sigterm_requires_proven_paused_idle_queue(self):
   c=self.controller;c.record={'operation_id':self.operation,'phase':'Prepared','original_members':module.validate_members(members())};c.save()
   worker=next(m for m in members() if m['name']=='indeedhub-ffmpeg')
@@ -194,12 +270,12 @@ class MaintenanceTests(unittest.TestCase):
    raise AssertionError(argv)
   c.runner=command
   with self.assertRaises(RuntimeError):c.graceful_stop(worker['name'])
-  c.record['queue_pause_confirmed']=True;c.record['last_queue_counts']={'active':1}
+  c.record['queue_pause_confirmed']=True;c.record['last_queue_counts']={name:(1 if name=='active' else 0) for name in module.QUEUE_COUNTS}
   with self.assertRaises(RuntimeError):c.graceful_stop(worker['name'])
   c.record['last_queue_counts']['active']=0;c.graceful_stop(worker['name'])
   self.assertEqual(c.record['stopped'][worker['name']]['classification'],'idle-worker-terminated-after-queue-drain')
  def test_legacy_api_compatibility_requires_fresh_empty_business_state(self):
-  c=self.controller;c.record={'operation_id':self.operation,'phase':'Prepared','stopped':{'indeedhub':{'confirmed':True},'indeedhub-ffmpeg':{'confirmed':True}},'queue_pause_confirmed':True,'last_queue_counts':{'active':0}};c.save()
+  c=self.controller;c.record={'operation_id':self.operation,'phase':'Prepared','stopped':{'indeedhub':{'confirmed':True},'indeedhub-ffmpeg':{'confirmed':True}},'queue_pause_confirmed':True,'last_queue_counts':{name:0 for name in module.QUEUE_COUNTS}};c.save()
   counts={name:0 for name in ('projects','contents','payments','shareholders','subscriptions','library_items','other_active_transactions')}
   c.runner=lambda argv,timeout,output:json.dumps(counts).encode()
   counts['payments']=1