feat(replication): support application health and controlled two-node recovery

Allow explicit 200 health responses for application origins and share only
in-flight readiness probes without caching their results. Document the accepted
native primary/replica pattern over the existing Xray transport, with fenced
manual promotion and independent node repositories.

Validation: 18 Worker tests, 17 readiness tests, and git diff --check passed.
Production SSO migration and traffic activation remain separate rollout steps.

Disclosure: I'm an AI assistant helping the author; I'm not the author.

assets/composition-readiness/README.md

@@ -1,6 +1,6 @@
 # Authenticated composition readiness adapter
 
-This dependency-free Python adapter exposes one local HTTP endpoint for the readiness-gated Worker and origin proxy. It runs an operator-supplied executable for every authenticated probe; exit0 means ready, every other outcome means unavailable. The executable must establish the application's actual readiness contract. A successful TCP connection or container health flag is not enough for replicated SSO.
+This dependency-free Python adapter exposes one local HTTP endpoint for the readiness-gated Worker and origin proxy. By default it runs an operator-supplied executable for every authenticated probe; exit0 means ready, every other outcome means unavailable. The executable must establish the application's actual readiness contract. A successful TCP connection or container health flag is not enough for replicated SSO.
 
 Example private configuration:
 
@@ -12,7 +12,8 @@ Example private configuration:
   "token_file": "/private/readiness-token",
   "command": ["/usr/local/bin/python3", "/assets/probe.py", "--config", "/private/probe.json"],
   "timeout": 1.5,
-  "slots": 4
+  "slots": 4,
+  "single_flight": false
 }
 ```
 
@@ -22,6 +23,8 @@ Only GET at the exact canonical path, with one matching `X-APEX-Ready-Token` hea
 
 Probe timeout is bounded to1.5seconds so an origin can answer within the Worker's two-second deadline. A timeout kills the probe's process group. Probe slots and request threads are bounded; slow socket operations time out after two seconds. Operators must account for proxy overhead and actual probe execution time. Probe programs must not detach into a new session or launch background jobs.
 
-The SSO probe must verify the local Authelia instance/config generation, local authenticated SQL and directory access, completed migration, SQL Primary/Synced/write authority, and fresh session-controller authority plus actual verified TLS application access to its selected primary. It must reject uncertain/stale state and avoid inventing a competing election. A short-lived controller status file alone is not a consensus lease. The generic adapter supplies none of those application checks; do not deploy a placeholder-success probe or claim these tests establish HA.
+Optional `single_flight: true` lets overlapping authenticated requests share one running fixed-command probe. It accepts only a JSON boolean and defaults to false. Waiters use that probe's remaining monotonic deadline; they never queue another probe for the same request. The existing connection limit (`slots * 2 + 4`) also bounds waiting requests. Timeout or shutdown returns unavailable and terminates the probe group. Completed results are not cached: each subsequent request starts a fresh probe, with no success TTL or failure cache. Coalescing reduces concurrent work but does not make a slow individual probe meet its deadline.
+
+For the controlled active/standby SSO pattern, the probe must reject a node not explicitly configured as active, verify completed migration and unchanged Authelia configuration, and check local Authelia health, authenticated SQL access/schema with read_only disabled, directory access, and an authenticated session write/read/delete over verified TLS. All dependencies belong to the selected local identity instance. These checks do not establish exclusive primary ownership: the operator must fence the previous writer before promotion, and the standby application and ingress stay disabled. No quorum, Sentinel, or session-controller authority is required by this pattern. The generic adapter supplies none of those application checks; do not deploy a placeholder-success probe or claim these tests establish HA.
 
 Run `python3 test_adapter.py` for loopback HTTP and subprocess behavior checks. No third-party packages are required. Production native image/probe/proxy integration and location-loss tests remain separate acceptance gates.

assets/composition-readiness/adapter.py

@@ -9,6 +9,7 @@ import re
 import signal
 import subprocess
 import threading
+import time
 
 
 class ReadinessServer(http.server.ThreadingHTTPServer):
@@ -16,7 +17,7 @@ class ReadinessServer(http.server.ThreadingHTTPServer):
     block_on_close = False
     allow_reuse_address = True
 
-    def __init__(self, address, *, token, path, command, timeout=1.5, slots=4):
+    def __init__(self, address, *, token, path, command, timeout=1.5, slots=4, single_flight=False):
         if not isinstance(token, str) or not re.fullmatch(r'[!-~]{16,512}', token):
             raise ValueError('A private visible-ASCII token is required')
         if not isinstance(path, str) or not re.fullmatch(r'/[A-Za-z0-9._~/-]+', path) or any(part in ('', '.', '..') for part in path[1:].split('/')):
@@ -25,10 +26,14 @@ class ReadinessServer(http.server.ThreadingHTTPServer):
             raise ValueError('A fixed absolute executable and argv are required')
         if isinstance(timeout, bool) or not 0 < timeout <= 1.5 or isinstance(slots, bool) or not isinstance(slots, int) or not 1 <= slots <= 16:
             raise ValueError('Invalid probe bounds')
+        if not isinstance(single_flight, bool):
+            raise ValueError('single_flight must be a boolean')
         self.token = token.encode('ascii')
         self.ready_path = path
         self.command = command
         self.timeout = timeout
+        self.single_flight = single_flight
+        self.flight = None
         self.probe_slots = threading.BoundedSemaphore(slots)
         self.connection_slots = threading.BoundedSemaphore(slots * 2 + 4)
         self.active = set()
@@ -74,6 +79,31 @@ class ReadinessServer(http.server.ThreadingHTTPServer):
         process.wait()
 
     def probe(self):
+        if not self.single_flight:
+            return self.run_probe()
+        with self.active_lock:
+            if self.closing:
+                return False
+            flight = self.flight
+            leader = flight is None
+            if leader:
+                flight = {'deadline': time.monotonic() + self.timeout, 'done': threading.Event(), 'ready': False}
+                self.flight = flight
+        if not leader:
+            flight['done'].wait(max(0, flight['deadline'] - time.monotonic()))
+            with self.active_lock:
+                return flight['ready'] and not self.closing and time.monotonic() < flight['deadline']
+        ready = False
+        try:
+            ready = self.run_probe(flight['deadline'])
+        finally:
+            with self.active_lock:
+                flight['ready'] = ready and not self.closing and time.monotonic() < flight['deadline']
+                self.flight = None
+                flight['done'].set()
+        return flight['ready']
+
+    def run_probe(self, deadline=None):
         if not self.probe_slots.acquire(blocking=False):
             return False
         process = None
@@ -84,7 +114,8 @@ class ReadinessServer(http.server.ThreadingHTTPServer):
                 process = subprocess.Popen(self.command, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True, close_fds=True)
                 self.active.add(process)
             try:
-                return process.wait(timeout=self.timeout) == 0
+                timeout = self.timeout if deadline is None else max(0, deadline - time.monotonic())
+                return process.wait(timeout=timeout) == 0
             except subprocess.TimeoutExpired:
                 self.stop_probe(process)
                 return False
@@ -99,6 +130,9 @@ class ReadinessServer(http.server.ThreadingHTTPServer):
     def server_close(self):
         with self.active_lock:
             self.closing = True
+            if self.flight is not None:
+                self.flight['ready'] = False
+                self.flight['done'].set()
             active = list(self.active)
         for process in active:
             self.stop_probe(process)

assets/composition-readiness/test_adapter.py

@@ -6,9 +6,9 @@ class AdapterTests(unittest.TestCase):
  def tearDown(self):
   for s,t in self.servers:s.shutdown();s.server_close();t.join()
   self.tmp.cleanup()
- def server(self,code=None,timeout=.3,slots=2):
+ def server(self,code=None,timeout=.3,slots=2,**options):
   code=code or f"from pathlib import Path;Path({str(self.marker)!r}).write_text('yes')"
-  server=ReadinessServer(('127.0.0.1',0),token='private-test-token',path='/.well-known/apex-ready',command=[sys.executable,'-c',code],timeout=timeout,slots=slots)
+  server=ReadinessServer(('127.0.0.1',0),token='private-test-token',path='/.well-known/apex-ready',command=[sys.executable,'-c',code],timeout=timeout,slots=slots,**options)
   t=threading.Thread(target=server.serve_forever,kwargs={'poll_interval':.01},daemon=True);t.start();self.servers.append((server,t));return server
  def request(self,s,path='/.well-known/apex-ready',token='private-test-token',method='GET',extra_token=False):
   c=http.client.HTTPConnection(*s.server_address,timeout=3);c.putrequest(method,path)
@@ -50,4 +50,70 @@ class AdapterTests(unittest.TestCase):
   self.assertFalse(self.marker.exists())
  def test_no_cached_readiness(self):
   s=self.server();self.assertEqual(self.request(s)[0],204);s.command=[sys.executable,'-c','raise SystemExit(1)'];self.assertEqual(self.request(s)[0],503)
+ def overlapping_requests(self,s,count=4):
+  barrier=threading.Barrier(count)
+  def one():
+   barrier.wait(timeout=2);start=time.monotonic();result=self.request(s);return result,time.monotonic()-start
+  with concurrent.futures.ThreadPoolExecutor(max_workers=count) as pool:return list(pool.map(lambda _:one(),range(count)))
+ def test_single_flight_shares_success_and_sequential_request_reprobes(self):
+  code=f"import pathlib,time;p=pathlib.Path({str(self.marker)!r});f=p.open('a');f.write('probe\\n');f.close();time.sleep(.2)"
+  s=self.server(code,timeout=1,slots=1,single_flight=True)
+  results=self.overlapping_requests(s)
+  self.assertEqual([r for r,_ in results],[(204,'no-store',b'')]*4)
+  self.assertEqual(self.marker.read_text().splitlines(),['probe'])
+  self.assertEqual(self.request(s)[0],204)
+  self.assertEqual(self.marker.read_text().splitlines(),['probe','probe'])
+ def test_single_flight_shares_failure_without_caching_it(self):
+  code=f"import pathlib,time;p=pathlib.Path({str(self.marker)!r});f=p.open('a');f.write('probe\\n');f.close();time.sleep(.2);raise SystemExit(1)"
+  s=self.server(code,timeout=1,slots=4,single_flight=True)
+  self.assertEqual([r for r,_ in self.overlapping_requests(s)],[(503,'no-store',b'')]*4)
+  self.assertEqual(self.marker.read_text().splitlines(),['probe'])
+  s.command=[sys.executable,'-c','pass'];self.assertEqual(self.request(s)[0],204)
+ def test_single_flight_timeout_wakes_waiters_and_kills_descendant(self):
+  code=f"import subprocess,time,sys;subprocess.Popen([sys.executable,'-c',\"import time,pathlib;time.sleep(.8);pathlib.Path({str(self.marker)!r}).write_text('leaked')\"]);time.sleep(3)"
+  s=self.server(code,timeout=.25,slots=1,single_flight=True)
+  results=self.overlapping_requests(s)
+  self.assertEqual([r for r,_ in results],[(503,'no-store',b'')]*4)
+  self.assertTrue(all(.15<elapsed<.7 for _,elapsed in results),results)
+  time.sleep(.9);self.assertFalse(self.marker.exists())
+ def test_single_flight_shutdown_wakes_waiters_and_rejects_new_requests(self):
+  leaked=self.root/'descendant'
+  child=f"import time,pathlib;pathlib.Path({str(self.marker)!r}).write_text('started');time.sleep(.8);pathlib.Path({str(leaked)!r}).write_text('leaked')"
+  code=f"import subprocess,time,sys;subprocess.Popen([sys.executable,'-c',{child!r}]);time.sleep(3)"
+  s=self.server(code,timeout=1.5,slots=1,single_flight=True)
+  with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:
+   futures=[pool.submit(self.request,s) for _ in range(4)]
+   for _ in range(100):
+    if self.marker.exists():break
+    time.sleep(.005)
+   self.assertTrue(self.marker.exists());start=time.monotonic();s.server_close()
+   self.assertEqual([f.result()[0] for f in futures],[503]*4);self.assertLess(time.monotonic()-start,.7)
+  self.assertFalse(s.probe());time.sleep(.9);self.assertFalse(leaked.exists())
+ def test_single_flight_late_waiter_uses_remaining_deadline(self):
+  code=f"import pathlib,time;p=pathlib.Path({str(self.marker)!r});f=p.open('a');f.write('probe\\n');f.close();time.sleep(3)"
+  s=self.server(code,timeout=.4,slots=1,single_flight=True)
+  with concurrent.futures.ThreadPoolExecutor() as pool:
+   first=pool.submit(self.request,s)
+   for _ in range(100):
+    if self.marker.exists():break
+    time.sleep(.005)
+   self.assertTrue(self.marker.exists());time.sleep(.2);start=time.monotonic()
+   self.assertEqual(self.request(s)[0],503);self.assertLess(time.monotonic()-start,.3)
+   self.assertEqual(first.result()[0],503)
+  self.assertEqual(self.marker.read_text().splitlines(),['probe'])
+ def test_single_flight_connection_limit_remains_bounded(self):
+  s=self.server(slots=1,single_flight=True);connections=[]
+  try:
+   for _ in range(6):
+    sock=socket.create_connection(s.server_address,timeout=2);sock.sendall(b'GET /.well-known/apex-ready HTTP/1.1\r\n');connections.append(sock)
+   self.assertEqual(self.request(s),(503,'no-store',b''));self.assertFalse(self.marker.exists())
+  finally:
+   for sock in connections:sock.close()
+ def test_single_flight_still_authenticates_before_probe(self):
+  s=self.server(single_flight=True)
+  for token,duplicate in [(None,False),('bad',False),('private-test-token',True)]:self.assertEqual(self.request(s,token=token,extra_token=duplicate)[0],403)
+  self.assertFalse(self.marker.exists())
+ def test_single_flight_requires_explicit_boolean(self):
+  for value in [0,1,None,'true']:
+   with self.subTest(value=value),self.assertRaises(ValueError):self.server(single_flight=value)
 if __name__=='__main__':unittest.main()