feat(messaging): implement Track 0 fault tolerance (B-14, B-15) and add 10 regression guards (G-1 to G-10)

2026-08-20 12:16:40 +09:00
parent 4025623958
commit c6b6c77ce4
6 changed files with 692 additions and 38 deletions
@@ -47,15 +47,51 @@ TERMINAL_EVENTS = ("completed", "error")


 def _format_line(topic: str, payload: Dict[str, Any]) -> str:
+    source_tag = f" [{payload['source']}]" if payload.get("source") else ""
    return (
        f"{payload.get('timestamp','-')}  "
        f"job={payload.get('job_id','?')}  "
        f"seq={payload.get('seq','?')}  "
-        f"{payload.get('event','?'):<20}  "
+        f"{payload.get('event','?') + source_tag:<20}  "
        f"{payload.get('detail','')}"
    )


+def _check_disk_fallback(
+    pending: Set[str],
+    registry_dir: str,
+    terminal: Dict[str, str],
+) -> None:
+    """Check local on-disk registry and audit logs for terminal status (B-15)."""
+    for jid in list(pending):
+        disk_status = None
+        try:
+            job_rec = load_job(jid, registry_dir)
+            disk_status = job_rec.get("status")
+        except Exception:
+            pass
+        if not disk_status or disk_status not in ("completed", "error", "cancelled"):
+            try:
+                disk_status = mqtt_common.read_logged_status(jid, mqtt_common.get_logs_dir())
+            except Exception:
+                pass
+
+        if disk_status in ("completed", "error", "cancelled"):
+            synth_event = "completed" if disk_status == "completed" else "error"
+            synth_payload = {
+                "job_id": jid,
+                "event": synth_event,
+                "seq": -1,
+                "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+                "detail": f"terminal status '{disk_status}' resolved via disk-fallback",
+                "source": "disk-fallback",
+            }
+            synth_topic = f"mam/{jid}/events"
+            print(_format_line(synth_topic, synth_payload), flush=True)
+            terminal[jid] = synth_event
+            pending.discard(jid)
+
+
 class _Watcher:
    """Holds the shared queue + the set of job_ids we accept events for."""

@@ -125,24 +161,7 @@ def _collect_jobs(args) -> List[Dict[str, Any]]:
    return [job]


-def main(argv=None) -> int:
-    parser = argparse.ArgumentParser(description="Subscribe to Job events on MQTT")
-    target = parser.add_mutually_exclusive_group(required=True)
-    target.add_argument("--job", help="job id to watch")
-    target.add_argument("--wait-any", action="store_true",
-                        help="watch every pending/running job in the registry")
-    parser.add_argument("--timeout", type=float, default=None,
-                        help="wall-clock budget in seconds (default: job.timeout_sec or 3600)")
-    parser.add_argument("--idle-timeout", type=float, default=None,
-                        help="max seconds with no new event (default: job.idle_timeout_sec or 120)")
-    parser.add_argument("--expect-retention", action="store_true",
-                        help="warn if no retained terminal event arrives promptly")
-    parser.add_argument("--registry-dir", default=DEFAULT_REGISTRY_DIR)
-    parser.add_argument("-v", "--verbose", action="store_true")
-    args = parser.parse_args(argv)
-
-    mqtt_common.setup_logging(logging.DEBUG if args.verbose else logging.WARNING)
-
+def _run_subscriber(args) -> int:
    try:
        jobs = _collect_jobs(args)
    except FileNotFoundError as exc:
@@ -197,28 +216,55 @@ def main(argv=None) -> int:
    client.on_disconnect = on_disconnect
    client.on_subscribe = on_subscribe
    client.reconnect_delay_set(min_delay=1, max_delay=16)
-    mqtt_common.with_retry(
-        lambda: client.connect(config.host, config.port, config.keepalive),
-        attempts=5, base_delay=1.0, max_delay=16.0
-    )()
-    client.loop_start()

    terminal: Dict[str, str] = {}        # job_id -> "completed"/"error"
    pending: Set[str] = set(expected_ids)
    start = time.monotonic()
    wall_deadline = start + wall_timeout
    last_event = start
+    last_disk_check = 0.0
    retention_checked = not args.expect_retention
+    connected = False
+
+    # Check if all jobs are already in terminal state on disk before connecting
+    _check_disk_fallback(pending, args.registry_dir, terminal)
+    if not pending:
+        if any(state == "error" for state in terminal.values()):
+            return 1
+        return 0
+
+    try:
+        mqtt_common.with_retry(
+            lambda: client.connect(config.host, config.port, config.keepalive),
+            attempts=3, base_delay=0.2, max_delay=1.0
+        )()
+        client.loop_start()
+        connected = True
+    except Exception as exc:
+        logger.warning("broker connection failed (%s); checking disk fallback", exc)
+        _check_disk_fallback(pending, args.registry_dir, terminal)
+        if not pending:
+            if any(state == "error" for state in terminal.values()):
+                return 1
+            return 0
+        logger.error("broker connection failed: %s", exc)
+        return 3

    try:
        while pending:
            now = time.monotonic()
            if now >= wall_deadline:
+                _check_disk_fallback(pending, args.registry_dir, terminal)
+                if not pending:
+                    break
                logger.error("wall-clock timeout (%.0fs); still pending: %s",
                             wall_timeout, ", ".join(sorted(pending)))
                return 2
            idle_left = idle_timeout - (now - last_event)
            if idle_left <= 0:
+                _check_disk_fallback(pending, args.registry_dir, terminal)
+                if not pending:
+                    break
                logger.error("idle timeout (%.0fs, no events); still pending: %s",
                             idle_timeout, ", ".join(sorted(pending)))
                return 2
@@ -230,6 +276,11 @@ def main(argv=None) -> int:
                    logger.warning("--expect-retention set but no retained "
                                   "terminal event observed yet")
                    retention_checked = True
+
+                # Step 2 (B-15): Local disk fallback throttled to every 3.0s
+                if (now - last_disk_check) >= 3.0:
+                    last_disk_check = now
+                    _check_disk_fallback(pending, args.registry_dir, terminal)
                continue

            last_event = time.monotonic()
@@ -246,11 +297,12 @@ def main(argv=None) -> int:
                terminal[jid] = event
                pending.discard(jid)
    finally:
-        client.loop_stop()
-        try:
-            client.disconnect()
-        except Exception:  # pragma: no cover
-            pass
+        if connected:
+            client.loop_stop()
+            try:
+                client.disconnect()
+            except Exception:  # pragma: no cover
+                pass

    # All jobs reached a terminal state. error wins over completed.
    if any(state == "error" for state in terminal.values()):
@@ -258,5 +310,30 @@ def main(argv=None) -> int:
    return 0


+def main(argv=None) -> int:
+    parser = argparse.ArgumentParser(description="Subscribe to Job events on MQTT")
+    target = parser.add_mutually_exclusive_group(required=True)
+    target.add_argument("--job", help="job id to watch")
+    target.add_argument("--wait-any", action="store_true",
+                        help="watch every pending/running job in the registry")
+    parser.add_argument("--timeout", type=float, default=None,
+                        help="wall-clock budget in seconds (default: job.timeout_sec or 3600)")
+    parser.add_argument("--idle-timeout", type=float, default=None,
+                        help="max seconds with no new event (default: job.idle_timeout_sec or 120)")
+    parser.add_argument("--expect-retention", action="store_true",
+                        help="warn if no retained terminal event arrives promptly")
+    parser.add_argument("--registry-dir", default=DEFAULT_REGISTRY_DIR)
+    parser.add_argument("-v", "--verbose", action="store_true")
+    args = parser.parse_args(argv)
+
+    mqtt_common.setup_logging(logging.DEBUG if args.verbose else logging.WARNING)
+
+    try:
+        return _run_subscriber(args)
+    except Exception as exc:
+        logger.error("subscriber fatal error: %s", exc)
+        return 3
+
+
 if __name__ == "__main__":
    sys.exit(main())