diff --git a/Data_Get/README.md b/Data_Get/README.md index 7062b9d..ad91fff 100644 --- a/Data_Get/README.md +++ b/Data_Get/README.md @@ -4,5 +4,6 @@ `manifest.json`、`READY`。数据来自机器人 Nvidia 上独立的项目录制服务，不影响厂家 `record_bag_node`。运行数据、临时同步目录及 `sync_status.json` 均不会提交到 Git。 -只有通过大小和 SHA-256 校验的目录才会从 `.incoming` 原子移动到这里。机器人端保留 -ready 副本，传输中断后会自动续传，不要把 `.incoming` 当作完整数据集。 +只有通过大小和 SHA-256 校验、fsync 和原子发布的目录才会从 `.incoming` 进入这里， +并写入持久化 `VERIFIED` 凭据。之后才按 episode ID 与 manifest SHA 精确删除机器人端 +对应 `ready` 副本。传输或删除确认中断会自动重试，不要把 `.incoming` 当作完整数据集。 diff --git a/README.md b/README.md index 20e9a3f..c6bbc96 100644 --- a/README.md +++ b/README.md @@ -12,7 +12,7 @@ - `tg3_local_teleop/`：部署到机器人 Nvidia；直接接收 OmniSocket 数据，发布双臂、 双手及 `/hric/robot/cmd_vel`，提供限速回 Home，并独立录制控制数据 MCAP。 - `tg3_data_collection/`：部署在本机；通过本地 SSH/rsync 拉取已完整收尾并校验的 - episode，原子保存到 `Data_Get/`。不会删除机器人上的备份。 + episode，原子保存到 `Data_Get/`；持久化验真凭据后自动回收机器人上的对应副本。 - `Data_Get/`：最终数采目录。实际 MCAP、清单和同步状态默认不提交 Git。 - `docs/`：可提交的跨机器人迁移步骤。含实测帧和现场拓扑的汇报/证据文档只保留在 当前本地工作副本，不同步到匿名可读的远端仓库。 @@ -54,8 +54,9 @@ TOML 解析检查。若还要核对外置 OmniSocketGo 版本，可设置任务 Nvidia 独立服务录制明确白名单内的机器人实测状态、双臂/双手命令与反馈、行走、 IMU、电源状态、xTELE 应用帧以及头/腰 Orbbec 的 RGB-D。RGB 使用标准 JPEG 压缩流， -16 位深度使用无损 compressedDepth PNG，并同时保存相机内参与静态 TF；腰部相机是 -必采，头部相机未启动时不阻塞，之后启动即可被同一录制自动发现。它不接管或停止厂家 +16 位深度使用无损 compressedDepth PNG，并同时保存相机内参与静态 TF。头部和腰部 +相机都是可选流：服务运行就自动采集，未运行不阻塞核心 episode；manifest 会分别标记 +`absent/healthy/partial/low_rate`。它不接管或停止厂家 `/record_bag_node`，也不录点云。正常结束后先校验 MCAP、生成 SHA-256 `manifest.json` 和 `READY`，再由 本机服务拉取到： @@ -64,8 +65,11 @@ IMU、电源状态、xTELE 应用帧以及头/腰 Orbbec 的 RGB-D。RGB 使用 /home/ps/Desktop/TG3_TS1P_OmniSocket_Teleop/Data_Get// ``` -传输中断只会留在 `Data_Get/.incoming/`，不会显示为完成 episode；机器人端 ready -副本不会自动删除。部署、状态检查和恢复步骤见 `tg3_data_collection/README.md`。 +传输中断只会留在 `Data_Get/.incoming/`，不会显示为完成 episode。机器人在录制期间 +必须临时写 `active/ready`；只有本机完成全文件 SHA-256、fsync、原子发布并写入 +`VERIFIED` 后，才会自动删除机器人上同 ID、同 manifest 哈希的 `ready` 目录。EAI +从不保存 MCAP 或图像 payload。部署、状态检查和恢复步骤见 +`tg3_data_collection/README.md`。 ## 迁移前必须修改 diff --git a/docs/天工3.0本地同构臂遥操迁移部署指南.md b/docs/天工3.0本地同构臂遥操迁移部署指南.md index a262409..916019c 100644 --- a/docs/天工3.0本地同构臂遥操迁移部署指南.md +++ b/docs/天工3.0本地同构臂遥操迁移部署指南.md @@ -1031,6 +1031,7 @@ tg3_local_teleop.py data_collection.py data_recorder_protocol.py data_recorder_node.py +delete_ready_episode.py config.toml run.sh run_data_recorder.sh @@ -1061,11 +1062,13 @@ rosbag 加载 `RobotState/ArmCtrl` 时出现 undefined symbol。`run.sh` 仍需 迁移时在 `config.toml [data_collection]` 核对 `base_directory`、100 GiB 余量、30 分钟 上限、明确 topic 白名单和 required topics。不要改成 `ros2 bag record -a`，点云需另行 估算带宽和磁盘后再加入。当前白名单已经包含头/腰 Orbbec 的 JPEG RGB、无损 PNG 深度、 -CameraInfo、Metadata 和 `/tf_static`；腰部 RGB-D/Metadata/静态 TF 为 required，两路 -图像收尾时还要求各自平均频率至少 `20 Hz`。头部未启动时是 optional，以后启动 -`orbbec_head.service` 即可由正在录制的 rosbag 自动发现。 +CameraInfo、Metadata 和 `/tf_static`。头、腰相机都是 optional：服务已启动时 rosbag 自动 +发现并采集；未启动时不影响核心 episode 开始或进入 READY。收尾后 manifest 会分别把 +两台相机标成 `absent`、`healthy`、`partial` 或 `low_rate`，后两种附带质量警告但不阻塞 +核心数据保存。 -新机器人先确认相机服务和发布者。相机必须在按 L3 开始 episode 之前启动： +需要采集某台相机的完整 episode 时，先确认相机服务和发布者；不需要相机时可保持服务 +停止，L3 数采仍可正常启动： ```bash systemctl is-active orbbec_waist.service orbbec_head.service @@ -1077,7 +1080,8 @@ ros2 topic info /ob_camera_waist/color/image_raw/compressed ros2 topic info /ob_camera_waist/depth/image_raw/compressedDepth ``` -两条腰部话题都必须显示 `Publisher count: 1`。原始 1280x720 RGB+深度约 +希望采集腰部图像时，两条腰部话题应显示 `Publisher count: 1`；头部同理。原始 +1280x720 RGB+深度约 `8.3 GB/min/相机`，所以默认不重复录 raw，而是录标准 image_transport 压缩流；现场 单相机约 `1.3–1.5 GB/min`，实际随画面变化。头部和腰部都运行时应按约两倍预留空间与 同步时间。`minimum_free_gib=100` 是停止下限，不是整条 episode 的容量预算；双相机 @@ -1113,12 +1117,17 @@ systemctl --user restart tg3-data-get-sync.service 2. 左摇杆按下（L3）连续 1 秒开始数采；该键是 `button_joystick.left`，不是 X/Y/Z； 3. 松开 L3 至少 0.5 秒，再长按 1 秒结束；或正常 Z+C 结束遥操自动收尾； 4. 查看 Nvidia `/tg3/data_collection/status`，必须先到 `ready`，不能从 `active/` 取数据； - recorder 会同时要求 required topics 有实时发布者且最终消息计数大于零； + recorder 会要求核心 required topics 有实时发布者且最终消息计数大于零；相机质量看 + manifest 的 `optional_topic_groups`，相机未启动显示 `absent`，不会阻塞； 5. 查看本机 `Data_Get/sync_status.json`，最终目录必须有 `READY`、`manifest.json`、 - `bag/metadata.yaml` 和非空 `*.mcap`； + `VERIFIED`、`bag/metadata.yaml` 和非空 `*.mcap`； 6. 执行 `ros2 bag info /bag` 并确认所需 topic 有消息； -7. 断开 PS 网络再采一条，确认 Nvidia 保留数据；恢复网络后应续传且 SHA-256 通过。 +7. 断开 PS 网络再采一条，确认 Nvidia 暂存 `ready/`；恢复网络后应续传，完成 SHA-256、 + fsync 和 `VERIFIED` 后，只删除机器人上 manifest 完全匹配的该 episode。 每个新会话先要求 L3 稳定松开 0.5 秒，同一次持续按压只切换一次。数采故障不得解除 -遥操或延迟 STOP/Home；桥心跳中断超过 3 秒时 recorder 会自行 SIGINT 收尾。机器人 -`ready/` 不自动删除，确认本机和外部备份后才按具体 episode 清理。 +遥操或延迟 STOP/Home；桥心跳中断超过 3 秒时 recorder 会自行 SIGINT 收尾。当前架构在 +录制期间必须先写 Nvidia 的项目专用 `active/`，PS 离线时也会暂存在 `ready/`；它不是 +长期副本。本机完成全文件验真并持久化 `VERIFIED` 后，固定根目录删除助手才会原子移走并 +删除该 exact episode，不能触碰 `active/`、`failed/` 或根目录。`retain_failed_episodes=false` +时失败录包也不在机器人保留。EAI 只转发控制数据，不保存 MCAP 或相机 payload。 diff --git a/tg3_data_collection/README.md b/tg3_data_collection/README.md index 59ef92b..37e140a 100644 --- a/tg3_data_collection/README.md +++ b/tg3_data_collection/README.md @@ -2,8 +2,10 @@ 机器人只在 `/home/nvidia/tg3_data_collection/ready` 暴露已经收到 SIGINT、写完 `metadata.yaml`、通过 `ros2 bag info` 并生成 SHA-256 清单的 episode。本机服务用免密 SSH/rsync 复制到 -`Data_Get/.incoming`，逐文件校验后再原子改名为 `Data_Get/`。中断的复制不会 -显示成完成数据，也不会删除机器人上的备份。 +`Data_Get/.incoming`，逐文件校验并 fsync 后再原子改名为 `Data_Get/`。本机 +`VERIFIED` 收据持久化成功后，正式服务会再次完整哈希本机 payload，再让 Nvidia 侧固定根目录 +helper 比对原始 manifest SHA-256，原子移入 `deleting/` 后清理。因此完整数据最终只保留在 PS 本机，Nvidia 只是临时 +staging；中断、哈希错误或本机落盘失败都不会触发远端删除。 数据只走本机到 Nvidia 的 SSH 链路，与公网或 EAI 本地 OmniSocket Hub 无关。当前默认： @@ -12,6 +14,8 @@ nvidia@192.168.41.2:/home/nvidia/tg3_data_collection/ready/ -> /home/ps/Desktop/TG3_TS1P_OmniSocket_Teleop/Data_Get/ ``` +EAI 不运行 recorder/sync，也不保存 MCAP 或图像数据。 + 先确认免密与依赖： ```bash @@ -20,6 +24,18 @@ command -v rsync ssh nvidia@192.168.41.2 command -v rsync ``` +先把固定根目录删除 helper 部署到 Nvidia。它的 CLI 只接受安全 episode ID 和 64 位 +manifest SHA-256，不接受 root 参数，也不会枚举或触碰 `active/`、`failed/`。helper 在 +`deleting/` 树之外的 `delete_ledger/` 先持久化独立删除授权；即使递归删除中断且 tombstone +里的 READY/manifest 已消失，仍能凭该授权安全续删，最后才删除授权文件： + +```bash +scp tg3_local_teleop/delete_ready_episode.py \ + nvidia@192.168.41.2:/home/nvidia/tg3_local_teleop/ +ssh nvidia@192.168.41.2 \ + chmod 755 /home/nvidia/tg3_local_teleop/delete_ready_episode.py +``` + 安装本机用户服务： ```bash @@ -39,10 +55,20 @@ python3 -m json.tool Data_Get/sync_status.json 也可以只执行一次，便于首次部署验收： ```bash -python3 tg3_data_collection/data_get_sync.py --once +systemctl --user stop tg3-data-get-sync.service +python3 tg3_data_collection/data_get_sync.py --once --delete-remote-after-sync +systemctl --user start tg3-data-get-sync.service ``` +同步器使用 `Data_Get/.data_get_sync.lock` 做 flock；正式服务运行时，另一个 `--once` 会明确 +报“已有同步进程”，不会与服务同时复制或删除。不带 `--delete-remote-after-sync` 的手动命令 +只复制、验真并保留远端，正式 unit 已显式启用自动回收。 + 日常轮询不会每 2 秒重新读取并哈希全部历史 MCAP；首次原子发布时已经完成深度校验。 +如果进程在远端删除 ACK 返回前退出，`VERIFIED` 会保持 `pending`。每次真正调用 helper +之前都会重新完整哈希本机 payload；失败后从 30 秒开始指数退避，最大 15 分钟，进程重启也 +遵守持久化的 `next_retry_unix_s`，避免每 2 秒重读大型 MCAP。helper 使用独立 600 秒超时， +能继续清理同摘要 tombstone/ledger，也会把已经不存在的同一目标视为成功。 需要定期审计已有本地数据时单独执行（可能耗时较长）： ```bash @@ -56,6 +82,7 @@ python3 tg3_data_collection/data_get_sync.py --once --verify-existing ```text / READY + VERIFIED manifest.json bag/metadata.yaml bag/*.mcap @@ -64,10 +91,32 @@ python3 tg3_data_collection/data_get_sync.py --once --verify-existing ``` 同步端会再次核对 `READY`、episode ID、每个 MCAP/metadata 的长度和 SHA-256；远端 -manifest 在传输中发生变化也会拒绝发布最终目录。`.incoming`、`sync_status.json` 和全部 -episode 已由 `Data_Get/.gitignore` 排除。服务绝不自动删除机器人端 `ready/` 数据；确认 -本机备份后如需清理，必须由操作者明确指定具体 episode，不能删除整个项目目录。 +manifest 在传输中发生变化也会拒绝发布最终目录。随后 fsync 所有复制文件和目录、以 +NOREPLACE 语义原子发布、fsync 父目录，最后写入并 fsync `VERIFIED`。收据绑定 episode、 +远端身份、原始 manifest SHA-256、已校验文件数和字节数，并记录 `pending/deleted`。 +`.incoming`、`sync_status.json` 和全部 episode 已由 `Data_Get/.gitignore` 排除。 -迁移后需要同步修改 service 中的本机项目路径、机器人 SSH 地址和机器人 ready 路径。 -如果 Nvidia 的 `192.168.41.2` 改变，只改该 unit 的 `--remote`；它不在机器人 -`config.toml` 或 OmniSocket Peer 设置中。 +临时目录按 `..partial` 隔离，rsync 同时使用删除同步；发布前还会 +拒绝 manifest 未列出的任何 `*.mcap` 或 `metadata.yaml`，避免旧中断文件混入新 episode。 + +`sync_status.json` 的 `state=delete_pending` 表示本机数据已完整发布，但远端回收尚未收到 +成功 ACK；`pending_remote_cleanup`、`last_delete_error`、`delete_count` 和 +`last_deleted_episode` 可用于排查。此状态下不要手工删除本机数据，服务会在下一轮重试。 + +已有本机同名目录不会直接授权删除：没有可信 `VERIFIED` 时必须重新做完整哈希，并要求 +本机与当前远端 manifest 原始摘要一致，之后才补收据并回收远端。同名碰撞、损坏、符号 +链接、错误 root 或摘要不符都保留远端并报错。 + +历史 `retained/deleted` 收据可以来自旧机器人 IP，不会阻塞新地址的同步；仍为 `pending` 的 +跨 IP 收据不会静默授权新机器人删除，只会逐项报错且不影响其他新 episode 继续同步。 + +自动回收 helper 严格只覆盖 `ready/`，录制中的 `active/` 永远不能由 PS 同步器处理， +也不能对 `failed/` 使用 READY 凭据。当前正式配置为 +`retain_failed_episodes=false`：录制器只把失败原因写入状态和 journal，随后精确删除自己 +刚创建的失败 payload，因此机器人不会长期保留失败 MCAP。若以后改回 `true` 以保留故障 +诊断，必须另建独立验真与回收协议，不能整目录清空。 + +迁移后需要同步修改 service 中的本机项目路径和机器人 SSH 地址。如果 Nvidia 的 +`192.168.41.2` 改变，只改 unit 的 `--remote` 并重新部署 helper；它不在机器人 +`config.toml` 或 OmniSocket Peer 设置中。启用自动删除时远端路径必须保持固定的 +`/home/nvidia/tg3_data_collection/ready`，同步器会拒绝对其他 root 启用删除。 diff --git a/tg3_data_collection/data_get_sync.py b/tg3_data_collection/data_get_sync.py index ab7450b..6f6d84a 100755 --- a/tg3_data_collection/data_get_sync.py +++ b/tg3_data_collection/data_get_sync.py @@ -4,40 +4,205 @@ from __future__ import annotations import argparse +import ctypes +import errno +import fcntl import hashlib import json +import math import os import re import shlex import shutil import signal +import stat import subprocess import time +import uuid +from contextlib import contextmanager +from dataclasses import dataclass from pathlib import Path -from typing import Any, Sequence +from typing import Any, Iterator, Sequence SAFE_EPISODE_NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,127}$") +SAFE_SHA256 = re.compile(r"^[0-9a-f]{64}$") +VERIFIED_RECEIPT_NAME = "VERIFIED" +VERIFIED_RECEIPT_VERSION = 1 +FIXED_REMOTE_READY = "/home/nvidia/tg3_data_collection/ready" +DEFAULT_REMOTE_DELETE_HELPER = ( + "/home/nvidia/tg3_local_teleop/delete_ready_episode.py" +) +AT_FDCWD = -100 +RENAME_NOREPLACE = 1 -def atomic_write_json(path: Path, payload: dict[str, Any]) -> None: +def _fsync_directory(path: Path) -> None: + flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0) + descriptor = os.open(path, flags) + try: + if not stat.S_ISDIR(os.fstat(descriptor).st_mode): + raise ValueError(f"not a directory: {path}") + os.fsync(descriptor) + finally: + os.close(descriptor) + + +def atomic_rename_noreplace(source: Path, destination: Path) -> None: + """Atomically publish a directory without replacing a colliding target.""" + + libc = ctypes.CDLL(None, use_errno=True) + renameat2 = getattr(libc, "renameat2", None) + if renameat2 is not None: + renameat2.argtypes = [ + ctypes.c_int, + ctypes.c_char_p, + ctypes.c_int, + ctypes.c_char_p, + ctypes.c_uint, + ] + renameat2.restype = ctypes.c_int + result = renameat2( + AT_FDCWD, + os.fsencode(source), + AT_FDCWD, + os.fsencode(destination), + RENAME_NOREPLACE, + ) + if result != 0: + error_number = ctypes.get_errno() + raise OSError( + error_number, + os.strerror(error_number), + str(destination), + ) + return + # Linux targets used by TG3 expose renameat2. This fallback remains safe + # against the supported sync processes because they share the flock. + if os.path.lexists(destination): + raise FileExistsError(errno.EEXIST, "destination already exists", destination) + os.rename(source, destination) + + +def atomic_write_json( + path: Path, + payload: dict[str, Any], + *, + durable: bool = False, +) -> None: path.parent.mkdir(parents=True, exist_ok=True) - temporary = path.with_suffix(path.suffix + ".tmp") - temporary.write_text( - json.dumps(payload, ensure_ascii=False, indent=2) + "\n", - encoding="utf-8", - ) - os.replace(temporary, path) + temporary = path.parent / f".{path.name}.{uuid.uuid4().hex}.tmp" + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0) + descriptor = os.open(temporary, flags, 0o600) + try: + data = (json.dumps(payload, ensure_ascii=False, indent=2) + "\n").encode( + "utf-8" + ) + with os.fdopen(descriptor, "wb", closefd=False) as stream: + stream.write(data) + stream.flush() + if durable: + os.fsync(descriptor) + except Exception: + try: + temporary.unlink() + except FileNotFoundError: + pass + raise + finally: + os.close(descriptor) + try: + os.replace(temporary, path) + if durable: + _fsync_directory(path.parent) + finally: + try: + temporary.unlink() + except FileNotFoundError: + pass def sha256_file(path: Path) -> str: digest = hashlib.sha256() - with path.open("rb") as stream: - for block in iter(lambda: stream.read(1024 * 1024), b""): - digest.update(block) + flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) + descriptor = os.open(path, flags) + try: + if not stat.S_ISREG(os.fstat(descriptor).st_mode): + raise ValueError(f"not a regular file: {path}") + stream = os.fdopen(descriptor, "rb", closefd=False) + with stream: + for block in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(block) + finally: + os.close(descriptor) return digest.hexdigest() +def _regular_file(path: Path, description: str) -> None: + try: + metadata = path.lstat() + except FileNotFoundError as exc: + raise ValueError(f"missing {description}: {path}") from exc + if not stat.S_ISREG(metadata.st_mode): + raise ValueError(f"{description} is not a regular file: {path}") + + +def fsync_episode_tree(directory: Path) -> None: + """Reject links/special entries, fsync every file, then every directory.""" + + try: + root_metadata = directory.lstat() + except FileNotFoundError as exc: + raise ValueError(f"episode directory is missing: {directory}") from exc + if not stat.S_ISDIR(root_metadata.st_mode): + raise ValueError(f"episode path is not a real directory: {directory}") + + directories: list[Path] = [] + for root, child_directories, child_files in os.walk( + directory, topdown=True, followlinks=False + ): + root_path = Path(root) + directories.append(root_path) + for name in child_directories: + child = root_path / name + if not stat.S_ISDIR(child.lstat().st_mode): + raise ValueError(f"unsafe non-directory entry: {child}") + for name in child_files: + child = root_path / name + metadata = child.lstat() + if not stat.S_ISREG(metadata.st_mode): + raise ValueError(f"unsafe non-regular episode entry: {child}") + flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) + descriptor = os.open(child, flags) + try: + if not stat.S_ISREG(os.fstat(descriptor).st_mode): + raise ValueError(f"unsafe episode file: {child}") + os.fsync(descriptor) + finally: + os.close(descriptor) + for child in reversed(directories): + _fsync_directory(child) + + +@dataclass(frozen=True) +class ManifestDocument: + raw: bytes + payload: dict[str, Any] + sha256: str + + @classmethod + def from_bytes(cls, raw: bytes, expected_episode: str) -> "ManifestDocument": + try: + text = raw.decode("utf-8") + except UnicodeDecodeError as exc: + raise ValueError(f"manifest is not UTF-8: {exc}") from exc + return cls( + raw=raw, + payload=parse_manifest(text, expected_episode), + sha256=hashlib.sha256(raw).hexdigest(), + ) + + def safe_episode_name(value: str) -> bool: return bool(SAFE_EPISODE_NAME.fullmatch(value)) and value not in (".", "..") @@ -63,10 +228,15 @@ def parse_manifest(raw: str, expected_episode: str) -> dict[str, Any]: if not isinstance(files, list) or not files: raise ValueError("manifest files must be a non-empty list") saw_mcap = False + seen_paths: set[str] = set() for entry in files: if not isinstance(entry, dict): raise ValueError("manifest file entry must be an object") relative = safe_relative_path(entry.get("path")) + normalized = relative.as_posix() + if normalized in seen_paths: + raise ValueError(f"duplicate manifest file path: {relative}") + seen_paths.add(normalized) saw_mcap = saw_mcap or relative.suffix == ".mcap" size = entry.get("size") digest = entry.get("sha256") @@ -85,17 +255,29 @@ def parse_manifest(raw: str, expected_episode: str) -> dict[str, Any]: def validate_episode_dir(directory: Path, manifest: dict[str, Any]) -> None: ready = directory / "READY" - if not ready.is_file() or ready.is_symlink(): - raise ValueError("episode has no regular READY marker") + _regular_file(ready, "READY marker") + _regular_file(directory / "manifest.json", "manifest") + listed_payloads: set[str] = set() for entry in manifest["files"]: relative = safe_relative_path(entry["path"]) + listed_payloads.add(relative.as_posix()) path = directory / relative - if not path.is_file() or path.is_symlink(): - raise ValueError(f"missing copied file: {relative}") + _regular_file(path, f"copied file {relative}") if path.stat().st_size != entry["size"]: raise ValueError(f"size mismatch: {relative}") if sha256_file(path) != entry["sha256"]: raise ValueError(f"sha256 mismatch: {relative}") + # A resumed staging directory must never smuggle an old bag segment into a + # newly verified episode. Logs/control markers may be unlisted, but every + # MCAP and rosbag metadata payload must be closed over by the manifest. + for root, _directories, files in os.walk(directory, followlinks=False): + root_path = Path(root) + for name in files: + relative = (root_path / name).relative_to(directory).as_posix() + if (name.endswith(".mcap") or name == "metadata.yaml") and ( + relative not in listed_payloads + ): + raise ValueError(f"unlisted bag payload: {relative}") class CommandRunner: @@ -126,6 +308,9 @@ class DataGetSync: ssh_timeout_s: float = 8.0, reserve_bytes: int = 1_073_741_824, verify_existing: bool = False, + delete_remote_after_sync: bool = False, + remote_delete_helper: str = DEFAULT_REMOTE_DELETE_HELPER, + remote_delete_timeout_s: float = 600.0, ) -> None: self.remote = remote self.remote_ready = remote_ready.rstrip("/") @@ -135,11 +320,31 @@ class DataGetSync: self.ssh_timeout_s = ssh_timeout_s self.reserve_bytes = reserve_bytes self.verify_existing = verify_existing + self.delete_remote_after_sync = delete_remote_after_sync + self.remote_delete_helper = remote_delete_helper + self.remote_delete_timeout_s = remote_delete_timeout_s + if self.delete_remote_after_sync and self.remote_ready != FIXED_REMOTE_READY: + raise ValueError( + "automatic deletion is restricted to the fixed Nvidia ready root: " + f"{FIXED_REMOTE_READY}" + ) + if self.delete_remote_after_sync and not self.remote_delete_helper.startswith("/"): + raise ValueError("remote delete helper path must be absolute") + if ( + not math.isfinite(self.remote_delete_timeout_s) + or self.remote_delete_timeout_s <= 0.0 + ): + raise ValueError("remote delete timeout must be positive and finite") self.stop_requested = False self.started_at = time.time() self.sync_count = 0 + self.delete_count = 0 self.last_episode: str | None = None + self.last_deleted_episode: str | None = None self.last_error = "" + self.last_delete_error = "" + self.pending_remote_cleanup = 0 + self._verified_this_process: set[tuple[str, str]] = set() @property def ssh_base(self) -> list[str]: @@ -148,6 +353,8 @@ class DataGetSync: "-o", "BatchMode=yes", "-o", + "StrictHostKeyChecking=yes", + "-o", f"ConnectTimeout={max(1, int(self.ssh_timeout_s))}", self.remote, ] @@ -155,6 +362,10 @@ class DataGetSync: def request_stop(self, _signum: int, _frame: object) -> None: self.stop_requested = True + @property + def remote_identity(self) -> str: + return f"{self.remote}:{self.remote_ready}" + def _write_status(self, state: str) -> None: atomic_write_json( self.status_file, @@ -164,22 +375,53 @@ class DataGetSync: "remote_ready": self.remote_ready, "destination": str(self.destination), "sync_count": self.sync_count, + "delete_count": self.delete_count, "last_episode": self.last_episode, + "last_deleted_episode": self.last_deleted_episode, "last_error": self.last_error, + "last_delete_error": self.last_delete_error, + "pending_remote_cleanup": self.pending_remote_cleanup, "uptime_s": round(time.time() - self.started_at, 1), "updated_unix_s": time.time(), }, ) - def _remote_command(self, command: str) -> str: + def _remote_command(self, command: str, *, timeout: float | None = None) -> str: result = self.runner.run( - [*self.ssh_base, command], timeout=self.ssh_timeout_s + 2.0 + [*self.ssh_base, command], + timeout=(self.ssh_timeout_s + 2.0 if timeout is None else timeout), ) if result.returncode != 0: detail = result.stderr.strip() or result.stdout.strip() raise RuntimeError(f"remote command failed: {detail}") return result.stdout + @contextmanager + def process_lock(self) -> Iterator[None]: + """Serialize the daemon and any manual ``--once`` invocation.""" + + self.destination.mkdir(parents=True, exist_ok=True) + lock_path = self.destination / ".data_get_sync.lock" + flags = os.O_RDWR | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0) + try: + descriptor = os.open(lock_path, flags, 0o600) + except OSError as exc: + raise RuntimeError(f"cannot open safe sync lock {lock_path}: {exc}") from exc + try: + if not stat.S_ISREG(os.fstat(descriptor).st_mode): + raise RuntimeError(f"sync lock is not a regular file: {lock_path}") + try: + fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError as exc: + raise RuntimeError( + "another tg3 data sync process already holds the destination lock" + ) from exc + os.ftruncate(descriptor, 0) + os.write(descriptor, f"{os.getpid()}\n".encode("ascii")) + yield + finally: + os.close(descriptor) + def list_remote_episodes(self) -> list[str]: root = shlex.quote(self.remote_ready) output = self._remote_command( @@ -193,53 +435,440 @@ class DataGetSync: raise RuntimeError(f"remote returned unsafe episode names: {unsafe!r}") return names - def get_remote_manifest(self, episode: str) -> dict[str, Any]: + def get_remote_manifest(self, episode: str) -> ManifestDocument: if not safe_episode_name(episode): raise ValueError(f"unsafe episode name: {episode!r}") episode_path = f"{self.remote_ready}/{episode}" manifest_path = f"{episode_path}/manifest.json" ready_path = f"{episode_path}/READY" raw = self._remote_command( - f"test -f {shlex.quote(ready_path)} && cat {shlex.quote(manifest_path)}" + f"test -d {shlex.quote(episode_path)} && " + f"test ! -L {shlex.quote(episode_path)} && " + f"test -f {shlex.quote(ready_path)} && " + f"test ! -L {shlex.quote(ready_path)} && " + f"test -f {shlex.quote(manifest_path)} && " + f"test ! -L {shlex.quote(manifest_path)} && " + f"cat {shlex.quote(manifest_path)}" ) - return parse_manifest(raw, episode) + return ManifestDocument.from_bytes(raw.encode("utf-8"), episode) def _enough_local_space(self, manifest: dict[str, Any]) -> bool: required = sum(int(entry["size"]) for entry in manifest["files"]) free = shutil.disk_usage(self.destination).free return free >= required + self.reserve_bytes + @staticmethod + def _manifest_totals(manifest: dict[str, Any]) -> tuple[int, int]: + return ( + len(manifest["files"]), + sum(int(entry["size"]) for entry in manifest["files"]), + ) + + @staticmethod + def _read_manifest_document(directory: Path, episode: str) -> ManifestDocument: + manifest_path = directory / "manifest.json" + _regular_file(manifest_path, "manifest") + return ManifestDocument.from_bytes(manifest_path.read_bytes(), episode) + + def _receipt_payload( + self, + final: Path, + document: ManifestDocument, + *, + delete_state: str, + attempts: int = 0, + last_error: str = "", + deleted_at_unix_s: float | None = None, + next_retry_unix_s: float = 0.0, + retry_delay_s: float = 0.0, + ) -> dict[str, Any]: + file_count, total_bytes = self._manifest_totals(document.payload) + remote_delete: dict[str, Any] = { + "state": delete_state, + "attempts": attempts, + "last_error": last_error, + "next_retry_unix_s": next_retry_unix_s, + "retry_delay_s": retry_delay_s, + } + if deleted_at_unix_s is not None: + remote_delete["deleted_at_unix_s"] = deleted_at_unix_s + return { + "schema_version": VERIFIED_RECEIPT_VERSION, + "state": "VERIFIED", + "episode_id": document.payload["episode_id"], + "manifest_sha256": document.sha256, + "remote_identity": self.remote_identity, + "remote": self.remote, + "remote_ready": self.remote_ready, + "local_final": str(final.resolve(strict=True)), + "verified_files": file_count, + "verified_bytes": total_bytes, + "verified_at_unix_s": time.time(), + "remote_delete": remote_delete, + } + + @staticmethod + def _receipt_path(final: Path) -> Path: + return final / VERIFIED_RECEIPT_NAME + + def _read_receipt(self, final: Path) -> dict[str, Any] | None: + path = self._receipt_path(final) + try: + metadata = path.lstat() + except FileNotFoundError: + return None + if not stat.S_ISREG(metadata.st_mode): + raise RuntimeError(f"VERIFIED receipt is not a regular file: {path}") + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise RuntimeError(f"invalid VERIFIED receipt {path}: {exc}") from exc + if not isinstance(payload, dict): + raise RuntimeError(f"invalid VERIFIED receipt root: {path}") + return payload + + def _validate_receipt( + self, + final: Path, + document: ManifestDocument, + receipt: dict[str, Any], + ) -> None: + file_count, total_bytes = self._manifest_totals(document.payload) + expected = { + "schema_version": VERIFIED_RECEIPT_VERSION, + "state": "VERIFIED", + "episode_id": document.payload["episode_id"], + "manifest_sha256": document.sha256, + "remote_identity": self.remote_identity, + "remote": self.remote, + "remote_ready": self.remote_ready, + "local_final": str(final.resolve(strict=True)), + "verified_files": file_count, + "verified_bytes": total_bytes, + } + for key, value in expected.items(): + if receipt.get(key) != value: + raise RuntimeError(f"VERIFIED receipt mismatch for {key}") + remote_delete = receipt.get("remote_delete") + if not isinstance(remote_delete, dict): + raise RuntimeError("VERIFIED receipt has no remote_delete state") + if remote_delete.get("state") not in ("retained", "pending", "deleted"): + raise RuntimeError("VERIFIED receipt has invalid remote_delete state") + attempts = remote_delete.get("attempts") + if isinstance(attempts, bool) or not isinstance(attempts, int) or attempts < 0: + raise RuntimeError("VERIFIED receipt has invalid delete attempts") + for key in ("next_retry_unix_s", "retry_delay_s"): + value = remote_delete.get(key, 0.0) + if ( + isinstance(value, bool) + or not isinstance(value, (int, float)) + or not math.isfinite(float(value)) + or float(value) < 0.0 + ): + raise RuntimeError(f"VERIFIED receipt has invalid {key}") + + def _write_receipt(self, final: Path, receipt: dict[str, Any]) -> None: + atomic_write_json(self._receipt_path(final), receipt, durable=True) + + @staticmethod + def _receipt_delete_state(receipt: dict[str, Any]) -> str: + remote_delete = receipt.get("remote_delete") + if not isinstance(remote_delete, dict): + raise RuntimeError("VERIFIED receipt has no remote_delete state") + state = remote_delete.get("state") + if state not in ("retained", "pending", "deleted"): + raise RuntimeError("VERIFIED receipt has invalid remote_delete state") + return str(state) + + @staticmethod + def _delete_retry_due(receipt: dict[str, Any], now: float | None = None) -> bool: + remote_delete = receipt["remote_delete"] + next_retry = float(remote_delete.get("next_retry_unix_s", 0.0)) + return (time.time() if now is None else now) >= next_retry + + def _deep_verify_local( + self, + final: Path, + document: ManifestDocument, + ) -> None: + validate_episode_dir(final, document.payload) + # Besides making the data durable, this traversal rejects symlinks in + # intermediate directories that a leaf-only manifest check cannot see. + fsync_episode_tree(final) + self._verified_this_process.add( + (document.payload["episode_id"], document.sha256) + ) + + def _ensure_existing_verified( + self, + final: Path, + remote_document: ManifestDocument, + ) -> tuple[ManifestDocument, dict[str, Any]]: + try: + final_metadata = final.lstat() + except FileNotFoundError as exc: + raise RuntimeError(f"existing destination disappeared: {final}") from exc + if not stat.S_ISDIR(final_metadata.st_mode): + raise RuntimeError(f"existing destination is not a real directory: {final}") + + local_document = self._read_manifest_document( + final, remote_document.payload["episode_id"] + ) + if ( + local_document.sha256 != remote_document.sha256 + or local_document.payload != remote_document.payload + ): + raise RuntimeError( + "existing local episode does not match the current remote manifest" + ) + + try: + receipt = self._read_receipt(final) + except RuntimeError: + # A malformed regular receipt is not a credential. The full tree + # verification below must succeed before it can be replaced. A + # symlink receipt is rejected by fsync_episode_tree. + receipt = None + receipt_trusted = False + if receipt is not None: + state = self._receipt_delete_state(receipt) + if ( + state == "pending" + and receipt.get("remote_identity") != self.remote_identity + ): + raise RuntimeError( + "pending remote deletion belongs to a different remote identity" + ) + try: + self._validate_receipt(final, local_document, receipt) + receipt_trusted = True + except RuntimeError: + receipt = None + + if receipt_trusted: + if self.verify_existing: + self._deep_verify_local(final, local_document) + return local_document, receipt + + # Missing/malformed/stale non-pending receipts are never deletion + # credentials. Rebuild one only after a current full payload hash. + if self.verify_existing or not receipt_trusted: + self._deep_verify_local(final, local_document) + + desired_state = "pending" if self.delete_remote_after_sync else "retained" + receipt = self._receipt_payload( + final, local_document, delete_state=desired_state + ) + self._write_receipt(final, receipt) + return local_document, receipt + + def _set_delete_state( + self, + final: Path, + document: ManifestDocument, + receipt: dict[str, Any], + *, + state: str, + attempts: int, + last_error: str, + deleted_at_unix_s: float | None = None, + next_retry_unix_s: float = 0.0, + retry_delay_s: float = 0.0, + ) -> dict[str, Any]: + updated = self._receipt_payload( + final, + document, + delete_state=state, + attempts=attempts, + last_error=last_error, + deleted_at_unix_s=deleted_at_unix_s, + next_retry_unix_s=next_retry_unix_s, + retry_delay_s=retry_delay_s, + ) + self._write_receipt(final, updated) + return updated + + def _invoke_delete_helper( + self, + episode: str, + manifest_sha256: str, + ) -> dict[str, Any]: + command = " ".join( + shlex.quote(part) + for part in (self.remote_delete_helper, episode, manifest_sha256) + ) + output = self._remote_command( + command, timeout=self.remote_delete_timeout_s + ) + lines = [line for line in output.splitlines() if line.strip()] + if not lines: + raise RuntimeError("remote delete helper returned no acknowledgement") + try: + acknowledgement = json.loads(lines[-1]) + except json.JSONDecodeError as exc: + raise RuntimeError("remote delete helper returned invalid JSON") from exc + if not isinstance(acknowledgement, dict): + raise RuntimeError("remote delete helper acknowledgement is not an object") + if acknowledgement.get("state") not in ( + "deleted", + "resumed_delete", + "already_absent", + ): + raise RuntimeError( + f"remote delete helper refused: {acknowledgement!r}" + ) + if acknowledgement.get("episode_id") != episode: + raise RuntimeError("remote delete acknowledgement episode mismatch") + if acknowledgement.get("manifest_sha256") != manifest_sha256: + raise RuntimeError("remote delete acknowledgement manifest mismatch") + return acknowledgement + + def _delete_verified_remote( + self, + final: Path, + document: ManifestDocument, + receipt: dict[str, Any], + ) -> bool: + self._validate_receipt(final, document, receipt) + # A VERIFIED receipt is a durable recovery cursor, not proof that the + # local payload has remained intact. Re-hash on every real helper call, + # including retries after an earlier remote deletion failure. + self._deep_verify_local(final, document) + + attempts = int(receipt["remote_delete"]["attempts"]) + 1 + retry_delay_s = min(900.0, 30.0 * (2 ** min(attempts - 1, 5))) + next_retry_unix_s = time.time() + retry_delay_s + # This durable pending write is the recovery cursor if the SSH ACK is + # lost after the robot has already renamed or removed its staging copy. + pending = self._set_delete_state( + final, + document, + receipt, + state="pending", + attempts=attempts, + last_error="", + next_retry_unix_s=next_retry_unix_s, + retry_delay_s=retry_delay_s, + ) + try: + self._invoke_delete_helper(document.payload["episode_id"], document.sha256) + self._set_delete_state( + final, + document, + pending, + state="deleted", + attempts=attempts, + last_error="", + deleted_at_unix_s=time.time(), + next_retry_unix_s=0.0, + retry_delay_s=0.0, + ) + except Exception as exc: + detail = str(exc) + try: + self._set_delete_state( + final, + document, + pending, + state="pending", + attempts=attempts, + last_error=detail, + next_retry_unix_s=next_retry_unix_s, + retry_delay_s=retry_delay_s, + ) + except Exception as receipt_exc: + detail += f"; could not persist pending receipt: {receipt_exc}" + self.last_delete_error = detail + return False + self.delete_count += 1 + self.last_deleted_episode = document.payload["episode_id"] + self.last_delete_error = "" + return True + + def _pending_receipts( + self, + ) -> tuple[ + list[tuple[Path, ManifestDocument, dict[str, Any]]], + list[str], + int, + ]: + pending: list[tuple[Path, ManifestDocument, dict[str, Any]]] = [] + errors: list[str] = [] + pending_count = 0 + if not self.destination.is_dir(): + return pending, errors, pending_count + for final in sorted(self.destination.iterdir(), key=lambda path: path.name): + if not safe_episode_name(final.name): + continue + if not stat.S_ISDIR(final.lstat().st_mode): + continue + try: + receipt = self._read_receipt(final) + if receipt is None: + continue + # Historical retained/deleted receipts may legitimately name a + # previous robot IP. They are not pending work and must not be + # validated against (or block) the current remote identity. + if self._receipt_delete_state(receipt) != "pending": + continue + pending_count += 1 + document = self._read_manifest_document(final, final.name) + self._validate_receipt(final, document, receipt) + except Exception as exc: + errors.append(f"{final.name}: invalid pending receipt: {exc}") + continue + pending.append((final, document, receipt)) + return pending, errors, pending_count + + def _retry_pending_deletions(self, exclude: set[str]) -> list[str]: + if not self.delete_remote_after_sync: + return [] + pending, errors, _pending_count = self._pending_receipts() + for final, document, receipt in pending: + episode = document.payload["episode_id"] + if episode in exclude: + continue + if not self._delete_retry_due(receipt): + continue + if not self._delete_verified_remote(final, document, receipt): + errors.append(f"{episode}: {self.last_delete_error}") + return errors + def sync_episode(self, episode: str) -> bool: if not safe_episode_name(episode): raise ValueError(f"unsafe episode name: {episode!r}") + remote_document = self.get_remote_manifest(episode) final = self.destination / episode - if final.exists(): - manifest_path = final / "manifest.json" - ready_path = final / "READY" - if ( - not manifest_path.is_file() - or manifest_path.is_symlink() - or not ready_path.is_file() - or ready_path.is_symlink() - ): - raise RuntimeError(f"existing destination is incomplete: {final}") - manifest = parse_manifest(manifest_path.read_text(encoding="utf-8"), episode) - # The first transfer already hashed every listed artifact before - # the atomic rename. Re-reading all historical MCAP files every - # two seconds would eventually saturate the workstation disk. - if self.verify_existing: - validate_episode_dir(final, manifest) + copied = False + if os.path.lexists(final): + local_document, receipt = self._ensure_existing_verified( + final, remote_document + ) + if self.delete_remote_after_sync and self._delete_retry_due(receipt): + self._delete_verified_remote(final, local_document, receipt) return False - manifest = self.get_remote_manifest(episode) - if not self._enough_local_space(manifest): + if not self._enough_local_space(remote_document.payload): raise RuntimeError("not enough local disk space for episode") - incoming = self.destination / ".incoming" / episode - incoming.mkdir(parents=True, exist_ok=True) + incoming_root = self.destination / ".incoming" + try: + incoming_root_metadata = incoming_root.lstat() + except FileNotFoundError: + incoming_root.mkdir(mode=0o700) + incoming_root_metadata = incoming_root.lstat() + if not stat.S_ISDIR(incoming_root_metadata.st_mode): + raise RuntimeError(f"incoming root is not a real directory: {incoming_root}") + incoming = incoming_root / f"{episode}.{remote_document.sha256}.partial" + if os.path.lexists(incoming): + if not stat.S_ISDIR(incoming.lstat().st_mode): + raise RuntimeError(f"incoming episode collision: {incoming}") + else: + incoming.mkdir(mode=0o700) remote_source = f"{self.remote}:{self.remote_ready}/{episode}/" ssh_transport = ( "ssh -o BatchMode=yes " + "-o StrictHostKeyChecking=yes " f"-o ConnectTimeout={max(1, int(self.ssh_timeout_s))}" ) result = self.runner.run( @@ -247,6 +876,7 @@ class DataGetSync: "rsync", "-a", "--partial", + "--delete-delay", "--protect-args", "-e", ssh_transport, @@ -260,27 +890,49 @@ class DataGetSync: raise RuntimeError(f"rsync failed: {detail}") copied_manifest_path = incoming / "manifest.json" - if not copied_manifest_path.is_file(): - raise RuntimeError("copied episode has no manifest.json") - copied_manifest = parse_manifest( - copied_manifest_path.read_text(encoding="utf-8"), episode + _regular_file(copied_manifest_path, "copied manifest") + copied_document = ManifestDocument.from_bytes( + copied_manifest_path.read_bytes(), episode ) - if copied_manifest != manifest: + if ( + copied_document.sha256 != remote_document.sha256 + or copied_document.payload != remote_document.payload + ): raise RuntimeError("remote manifest changed during transfer") - validate_episode_dir(incoming, copied_manifest) - os.replace(incoming, final) + validate_episode_dir(incoming, copied_document.payload) + fsync_episode_tree(incoming) + if os.path.lexists(final): + raise RuntimeError(f"destination collision before publish: {final}") + atomic_rename_noreplace(incoming, final) + _fsync_directory(incoming_root) + _fsync_directory(self.destination) + self._verified_this_process.add((episode, copied_document.sha256)) + receipt = self._receipt_payload( + final, + copied_document, + delete_state=("pending" if self.delete_remote_after_sync else "retained"), + ) + self._write_receipt(final, receipt) self.sync_count += 1 self.last_episode = episode - return True + copied = True + if self.delete_remote_after_sync and self._delete_retry_due(receipt): + self._delete_verified_remote(final, copied_document, receipt) + return copied - def run_once(self) -> int: + def _run_once_unlocked(self) -> int: self.destination.mkdir(parents=True, exist_ok=True) - (self.destination / ".incoming").mkdir(parents=True, exist_ok=True) + incoming_root = self.destination / ".incoming" + incoming_root.mkdir(parents=True, exist_ok=True) + if not stat.S_ISDIR(incoming_root.lstat().st_mode): + raise RuntimeError(f"incoming root is not a real directory: {incoming_root}") copied = 0 errors: list[str] = [] + processed: set[str] = set() try: episodes = self.list_remote_episodes() for episode in episodes: + processed.add(episode) try: if self.sync_episode(episode): copied += 1 @@ -288,34 +940,59 @@ class DataGetSync: # One damaged historical episode must not starve newer # ready data. Keep its error visible and continue. errors.append(f"{episode}: {exc}") + delete_errors = self._retry_pending_deletions(processed) + if self.delete_remote_after_sync: + pending, pending_scan_errors, pending_count = self._pending_receipts() + delete_errors.extend(pending_scan_errors) + else: + pending, pending_count = [], 0 + self.pending_remote_cleanup = pending_count + if pending and not self.last_delete_error: + pending_details = [ + str(receipt["remote_delete"].get("last_error", "")) + for _final, _document, receipt in pending + if receipt["remote_delete"].get("last_error") + ] + self.last_delete_error = "; ".join(pending_details) + if delete_errors and not self.last_delete_error: + self.last_delete_error = "; ".join(delete_errors) if errors: raise RuntimeError("; ".join(errors)) self.last_error = "" - self._write_status("idle") + if self.pending_remote_cleanup: + self._write_status("delete_pending") + else: + self.last_delete_error = "" + self._write_status("idle") except Exception as exc: self.last_error = str(exc) self._write_status("error") raise return copied + def run_once(self) -> int: + with self.process_lock(): + return self._run_once_unlocked() + def run_forever(self, poll_seconds: float) -> None: self.destination.mkdir(parents=True, exist_ok=True) - while not self.stop_requested: - try: - self.run_once() - except Exception: - pass - deadline = time.monotonic() + poll_seconds - while not self.stop_requested and time.monotonic() < deadline: - time.sleep(min(0.2, max(0.0, deadline - time.monotonic()))) - self._write_status("stopped") + with self.process_lock(): + while not self.stop_requested: + try: + self._run_once_unlocked() + except Exception: + pass + deadline = time.monotonic() + poll_seconds + while not self.stop_requested and time.monotonic() < deadline: + time.sleep(min(0.2, max(0.0, deadline - time.monotonic()))) + self._write_status("stopped") def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--remote", default="nvidia@192.168.41.2") parser.add_argument( - "--remote-ready", default="/home/nvidia/tg3_data_collection/ready" + "--remote-ready", default=FIXED_REMOTE_READY ) parser.add_argument( "--destination", @@ -331,6 +1008,25 @@ def parse_args() -> argparse.Namespace: action="store_true", help="rehash already published local episodes (slow; intended for audits)", ) + parser.add_argument( + "--delete-remote-after-sync", + action="store_true", + help=( + "after durable local verification, use the fixed Nvidia helper to " + "remove only the matching remote ready episode" + ), + ) + parser.add_argument( + "--remote-delete-helper", + default=DEFAULT_REMOTE_DELETE_HELPER, + help="absolute path of the fixed-root deletion helper on Nvidia", + ) + parser.add_argument( + "--remote-delete-timeout-seconds", + type=float, + default=600.0, + help="timeout for one remote recursive cleanup (default: 600)", + ) parser.add_argument("--once", action="store_true") return parser.parse_args() @@ -347,6 +1043,9 @@ def main() -> int: destination=destination, status_file=status_file, verify_existing=args.verify_existing, + delete_remote_after_sync=args.delete_remote_after_sync, + remote_delete_helper=args.remote_delete_helper, + remote_delete_timeout_s=args.remote_delete_timeout_seconds, ) signal.signal(signal.SIGINT, syncer.request_stop) signal.signal(signal.SIGTERM, syncer.request_stop) @@ -356,6 +1055,12 @@ def main() -> int: except Exception as exc: print(f"data sync failed: {exc}") return 1 + if syncer.pending_remote_cleanup: + print( + "data sync completed locally, but remote cleanup remains pending: " + f"{syncer.last_delete_error}" + ) + return 2 return 0 syncer.run_forever(args.poll_seconds) return 0 diff --git a/tg3_data_collection/test_data_get_sync.py b/tg3_data_collection/test_data_get_sync.py index a3c3a70..c187cfc 100755 --- a/tg3_data_collection/test_data_get_sync.py +++ b/tg3_data_collection/test_data_get_sync.py @@ -3,12 +3,20 @@ from __future__ import annotations import hashlib import json +import os +import shutil +import subprocess import tempfile import unittest from pathlib import Path +from unittest import mock from data_get_sync import ( DataGetSync, + FIXED_REMOTE_READY, + ManifestDocument, + VERIFIED_RECEIPT_NAME, + fsync_episode_tree, parse_manifest, safe_episode_name, safe_relative_path, @@ -16,6 +24,99 @@ from data_get_sync import ( ) +class LocalRsyncRunner: + def __init__(self, remote_root: Path) -> None: + self.remote_root = remote_root + self.mode = "normal" + self.rsync_calls = 0 + + def run( + self, command: list[str], *, timeout: float + ) -> subprocess.CompletedProcess[str]: + del timeout + if not command or command[0] != "rsync": + return subprocess.CompletedProcess(command, 1, "", "unexpected command") + self.rsync_calls += 1 + remote_source = command[-2].rstrip("/") + episode = remote_source.rsplit("/", 1)[-1] + destination = Path(command[-1]) + shutil.copytree( + self.remote_root / episode, + destination, + dirs_exist_ok=True, + symlinks=True, + ) + if self.mode == "truncate": + (destination / "bag" / "bag_0.mcap").write_bytes(b"truncated") + elif self.mode == "manifest_changed": + manifest = destination / "manifest.json" + manifest.write_bytes(manifest.read_bytes() + b"\n") + elif self.mode == "manifest_symlink": + manifest = destination / "manifest.json" + manifest.unlink() + manifest.symlink_to("bag/metadata.yaml") + return subprocess.CompletedProcess(command, 0, "", "") + + +class LocalSync(DataGetSync): + def __init__( + self, + *, + remote_root: Path, + destination: Path, + runner: LocalRsyncRunner, + delete: bool = True, + remote: str = "nvidia@test", + ) -> None: + super().__init__( + remote=remote, + remote_ready=FIXED_REMOTE_READY, + destination=destination, + status_file=destination / "sync_status.json", + runner=runner, + reserve_bytes=0, + delete_remote_after_sync=delete, + remote_delete_helper="/fixed/delete_ready_episode.py", + ) + self.remote_root = remote_root + self.delete_calls: list[tuple[str, str]] = [] + self.delete_outcomes: list[str] = [] + + def list_remote_episodes(self) -> list[str]: + return sorted( + path.name + for path in self.remote_root.iterdir() + if path.is_dir() and not path.is_symlink() + ) + + def get_remote_manifest(self, episode: str) -> ManifestDocument: + raw = (self.remote_root / episode / "manifest.json").read_bytes() + return ManifestDocument.from_bytes(raw, episode) + + def _invoke_delete_helper( + self, episode: str, manifest_sha256: str + ) -> dict[str, object]: + self.delete_calls.append((episode, manifest_sha256)) + outcome = self.delete_outcomes.pop(0) if self.delete_outcomes else "success" + target = self.remote_root / episode + if outcome == "failure": + raise RuntimeError("injected delete failure") + if outcome == "ack_lost_after_delete": + if target.exists(): + shutil.rmtree(target) + raise RuntimeError("injected lost acknowledgement") + if target.exists(): + shutil.rmtree(target) + state = "deleted" + else: + state = "already_absent" + return { + "state": state, + "episode_id": episode, + "manifest_sha256": manifest_sha256, + } + + class DataGetSyncTests(unittest.TestCase): def test_safe_episode_name(self) -> None: self.assertTrue(safe_episode_name("episode_20260810T120000000_deadbeef")) @@ -93,28 +194,377 @@ class DataGetSyncTests(unittest.TestCase): with self.assertRaises(ValueError): validate_episode_dir(episode, manifest) - def test_existing_episode_skips_expensive_hash_unless_requested(self) -> None: + def test_fsync_episode_tree_rejects_intermediate_symlink(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) - name = "episode_20260810T120000000_deadbeef" - episode, _manifest = self._episode(root, name) - (episode / "bag" / "bag_0.mcap").write_bytes(b"changed") - fast = DataGetSync( - remote="unused", - remote_ready="/unused", - destination=root, - status_file=root / "status.json", - ) - self.assertFalse(fast.sync_episode(name)) - deep = DataGetSync( - remote="unused", - remote_ready="/unused", - destination=root, - status_file=root / "status.json", - verify_existing=True, - ) + episode = root / "episode" + episode.mkdir() + outside = root / "outside" + outside.mkdir() + (outside / "file").write_bytes(b"sentinel") + (episode / "linked").symlink_to(outside, target_is_directory=True) with self.assertRaises(ValueError): - deep.sync_episode(name) + fsync_episode_tree(episode) + + def _local_sync( + self, root: Path, name: str = "episode_20260810T120000000_deadbeef" + ) -> tuple[LocalSync, LocalRsyncRunner, Path, Path]: + remote = root / "remote_ready" + destination = root / "Data_Get" + remote.mkdir() + destination.mkdir() + self._episode(remote, name) + runner = LocalRsyncRunner(remote) + syncer = LocalSync( + remote_root=remote, + destination=destination, + runner=runner, + ) + return syncer, runner, remote, destination + + def test_copy_is_verified_receipted_then_remote_is_deleted(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + name = "episode_20260810T120000000_deadbeef" + syncer, _runner, remote, destination = self._local_sync( + Path(temporary), name + ) + self.assertEqual(syncer.run_once(), 1) + final = destination / name + receipt = json.loads((final / VERIFIED_RECEIPT_NAME).read_text()) + self.assertEqual(receipt["state"], "VERIFIED") + self.assertEqual(receipt["remote_identity"], syncer.remote_identity) + self.assertEqual(receipt["verified_files"], 2) + self.assertGreater(receipt["verified_bytes"], 0) + self.assertEqual(receipt["remote_delete"]["state"], "deleted") + self.assertFalse((remote / name).exists()) + self.assertEqual(syncer.delete_count, 1) + status = json.loads((destination / "sync_status.json").read_text()) + self.assertEqual(status["state"], "idle") + self.assertEqual(status["pending_remote_cleanup"], 0) + + def test_truncated_copy_never_calls_delete(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + syncer, runner, remote, _destination = self._local_sync(Path(temporary)) + runner.mode = "truncate" + with self.assertRaises(RuntimeError): + syncer.run_once() + self.assertEqual(syncer.delete_calls, []) + self.assertEqual(len(list(remote.iterdir())), 1) + + def test_manifest_change_during_copy_never_calls_delete(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + syncer, runner, remote, _destination = self._local_sync(Path(temporary)) + runner.mode = "manifest_changed" + with self.assertRaises(RuntimeError): + syncer.run_once() + self.assertEqual(syncer.delete_calls, []) + self.assertEqual(len(list(remote.iterdir())), 1) + + def test_copied_manifest_symlink_never_calls_delete(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + syncer, runner, remote, _destination = self._local_sync(Path(temporary)) + runner.mode = "manifest_symlink" + with self.assertRaises(RuntimeError): + syncer.run_once() + self.assertEqual(syncer.delete_calls, []) + self.assertEqual(len(list(remote.iterdir())), 1) + + def test_delete_failure_is_pending_and_retried(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + name = "episode_20260810T120000000_deadbeef" + syncer, _runner, remote, destination = self._local_sync( + Path(temporary), name + ) + syncer.delete_outcomes = ["failure", "success"] + self.assertEqual(syncer.run_once(), 1) + receipt_path = destination / name / VERIFIED_RECEIPT_NAME + receipt = json.loads(receipt_path.read_text()) + self.assertEqual(receipt["remote_delete"]["state"], "pending") + self.assertTrue((remote / name).exists()) + status = json.loads((destination / "sync_status.json").read_text()) + self.assertEqual(status["state"], "delete_pending") + self.assertEqual(status["pending_remote_cleanup"], 1) + + # Persistent backoff prevents a 2-second poll loop from repeatedly + # hashing a large MCAP after a network/helper failure. + self.assertEqual(syncer.run_once(), 0) + self.assertEqual(len(syncer.delete_calls), 1) + receipt = json.loads(receipt_path.read_text()) + receipt["remote_delete"]["next_retry_unix_s"] = 0.0 + receipt_path.write_text(json.dumps(receipt), encoding="utf-8") + + self.assertEqual(syncer.run_once(), 0) + receipt = json.loads(receipt_path.read_text()) + self.assertEqual(receipt["remote_delete"]["state"], "deleted") + self.assertFalse((remote / name).exists()) + self.assertEqual(len(syncer.delete_calls), 2) + + def test_lost_delete_ack_is_idempotent_after_process_restart(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + name = "episode_20260810T120000000_deadbeef" + root = Path(temporary) + first, runner, remote, destination = self._local_sync(root, name) + first.delete_outcomes = ["ack_lost_after_delete"] + self.assertEqual(first.run_once(), 1) + self.assertFalse((remote / name).exists()) + receipt_path = destination / name / VERIFIED_RECEIPT_NAME + self.assertEqual( + json.loads(receipt_path.read_text())["remote_delete"]["state"], + "pending", + ) + + restarted = LocalSync( + remote_root=remote, + destination=destination, + runner=runner, + ) + with mock.patch( + "data_get_sync.validate_episode_dir", + wraps=validate_episode_dir, + ) as deep_validator: + self.assertEqual(restarted.run_once(), 0) + self.assertEqual(deep_validator.call_count, 0) + self.assertEqual(restarted.delete_calls, []) + + receipt = json.loads(receipt_path.read_text()) + receipt["remote_delete"]["next_retry_unix_s"] = 0.0 + receipt_path.write_text(json.dumps(receipt), encoding="utf-8") + self.assertEqual(restarted.run_once(), 0) + self.assertGreaterEqual(deep_validator.call_count, 1) + self.assertEqual( + json.loads(receipt_path.read_text())["remote_delete"]["state"], + "deleted", + ) + self.assertEqual(restarted.delete_calls[0][0], name) + + def test_retry_rehashes_and_refuses_delete_after_local_tamper(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + name = "episode_20260810T120000000_deadbeef" + syncer, _runner, remote, destination = self._local_sync( + Path(temporary), name + ) + syncer.delete_outcomes = ["failure", "success"] + self.assertEqual(syncer.run_once(), 1) + (destination / name / "bag" / "bag_0.mcap").write_bytes(b"tampered") + receipt_path = destination / name / VERIFIED_RECEIPT_NAME + receipt = json.loads(receipt_path.read_text()) + receipt["remote_delete"]["next_retry_unix_s"] = 0.0 + receipt_path.write_text(json.dumps(receipt), encoding="utf-8") + + with self.assertRaises(RuntimeError): + syncer.run_once() + self.assertEqual(len(syncer.delete_calls), 1) + self.assertTrue((remote / name).exists()) + + def test_unlisted_stale_mcap_is_never_published_or_deleted(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + name = "episode_20260810T120000000_deadbeef" + syncer, _runner, remote, destination = self._local_sync( + Path(temporary), name + ) + (remote / name / "bag" / "stale_old_payload.mcap").write_bytes( + b"stale" + ) + with self.assertRaises(RuntimeError): + syncer.run_once() + self.assertFalse((destination / name).exists()) + self.assertEqual(syncer.delete_calls, []) + self.assertTrue((remote / name).exists()) + + def test_reused_hash_staging_cannot_smuggle_old_mcap(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + name = "episode_20260810T120000000_deadbeef" + syncer, _runner, remote, destination = self._local_sync( + Path(temporary), name + ) + document = syncer.get_remote_manifest(name) + staging = ( + destination + / ".incoming" + / f"{name}.{document.sha256}.partial" + / "bag" + ) + staging.mkdir(parents=True) + (staging / "stale_old_payload.mcap").write_bytes(b"old interrupted bag") + with self.assertRaises(RuntimeError): + syncer.run_once() + self.assertFalse((destination / name).exists()) + self.assertEqual(syncer.delete_calls, []) + self.assertTrue((remote / name).exists()) + + def test_deleted_receipt_from_old_ip_does_not_block_new_ip(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + name = "episode_20260810T120000000_deadbeef" + root = Path(temporary) + first, runner, remote, destination = self._local_sync(root, name) + self.assertEqual(first.run_once(), 1) + receipt_path = destination / name / VERIFIED_RECEIPT_NAME + receipt = json.loads(receipt_path.read_text()) + receipt["remote"] = "nvidia@old-ip" + receipt["remote_identity"] = ( + f"nvidia@old-ip:{FIXED_REMOTE_READY}" + ) + receipt_path.write_text(json.dumps(receipt), encoding="utf-8") + + migrated = LocalSync( + remote_root=remote, + destination=destination, + runner=runner, + remote="nvidia@new-ip", + ) + self.assertEqual(migrated.run_once(), 0) + status = json.loads((destination / "sync_status.json").read_text()) + self.assertEqual(status["state"], "idle") + self.assertEqual(status["pending_remote_cleanup"], 0) + + def test_bad_pending_old_ip_does_not_starve_new_episode(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + old_name = "episode_old" + first, _runner, _old_remote, destination = self._local_sync( + root, old_name + ) + first.delete_outcomes = ["failure"] + self.assertEqual(first.run_once(), 1) + + new_remote = root / "new_remote_ready" + new_remote.mkdir() + new_name = "episode_new" + self._episode(new_remote, new_name) + new_runner = LocalRsyncRunner(new_remote) + migrated = LocalSync( + remote_root=new_remote, + destination=destination, + runner=new_runner, + remote="nvidia@new-ip", + ) + self.assertEqual(migrated.run_once(), 1) + self.assertFalse((new_remote / new_name).exists()) + self.assertTrue((destination / new_name).is_dir()) + status = json.loads((destination / "sync_status.json").read_text()) + self.assertEqual(status["state"], "delete_pending") + self.assertEqual(status["pending_remote_cleanup"], 1) + self.assertIn("remote_identity", status["last_delete_error"]) + + def test_delete_helper_uses_independent_long_timeout(self) -> None: + class AckRunner: + def __init__(self) -> None: + self.timeouts: list[float] = [] + + def run( + self, command: list[str], *, timeout: float + ) -> subprocess.CompletedProcess[str]: + self.timeouts.append(timeout) + payload = { + "state": "already_absent", + "episode_id": "episode_timeout", + "manifest_sha256": "0" * 64, + } + return subprocess.CompletedProcess( + command, 0, json.dumps(payload) + "\n", "" + ) + + with tempfile.TemporaryDirectory() as temporary: + runner = AckRunner() + syncer = DataGetSync( + remote="nvidia@test", + remote_ready=FIXED_REMOTE_READY, + destination=Path(temporary), + status_file=Path(temporary) / "status.json", + runner=runner, + delete_remote_after_sync=True, + remote_delete_timeout_s=600.0, + ) + syncer._invoke_delete_helper("episode_timeout", "0" * 64) + self.assertEqual(runner.timeouts, [600.0]) + + def test_existing_final_without_receipt_is_deep_verified_before_delete(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + name = "episode_20260810T120000000_deadbeef" + root = Path(temporary) + syncer, _runner, remote, destination = self._local_sync(root, name) + shutil.copytree(remote / name, destination / name) + with mock.patch( + "data_get_sync.validate_episode_dir", + wraps=validate_episode_dir, + ) as deep_validator: + self.assertEqual(syncer.run_once(), 0) + self.assertGreaterEqual(deep_validator.call_count, 1) + self.assertTrue((destination / name / VERIFIED_RECEIPT_NAME).is_file()) + self.assertFalse((remote / name).exists()) + + def test_existing_corrupt_final_without_receipt_never_deletes_remote(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + name = "episode_20260810T120000000_deadbeef" + root = Path(temporary) + syncer, _runner, remote, destination = self._local_sync(root, name) + shutil.copytree(remote / name, destination / name) + (destination / name / "bag" / "bag_0.mcap").write_bytes(b"corrupt") + with self.assertRaises(RuntimeError): + syncer.run_once() + self.assertEqual(syncer.delete_calls, []) + self.assertTrue((remote / name).exists()) + + def test_existing_manifest_collision_never_deletes_remote(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + name = "episode_20260810T120000000_deadbeef" + root = Path(temporary) + syncer, _runner, remote, destination = self._local_sync(root, name) + self._episode(destination, name) + local_manifest = destination / name / "manifest.json" + local_manifest.write_bytes(local_manifest.read_bytes() + b"\n") + with self.assertRaises(RuntimeError): + syncer.run_once() + self.assertEqual(syncer.delete_calls, []) + self.assertTrue((remote / name).exists()) + + def test_receipt_write_failure_never_calls_delete(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + syncer, _runner, remote, _destination = self._local_sync(Path(temporary)) + with mock.patch( + "data_get_sync.atomic_write_json", + side_effect=RuntimeError("injected receipt fsync failure"), + ): + with self.assertRaises(RuntimeError): + syncer.sync_episode(next(remote.iterdir()).name) + self.assertEqual(syncer.delete_calls, []) + self.assertEqual(len(list(remote.iterdir())), 1) + + def test_sync_flock_rejects_concurrent_once(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + first, runner, remote, destination = self._local_sync(root) + second = LocalSync( + remote_root=remote, + destination=destination, + runner=runner, + ) + with first.process_lock(): + with self.assertRaisesRegex(RuntimeError, "already holds"): + second.run_once() + + def test_symlink_lock_is_refused(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + syncer, _runner, _remote, destination = self._local_sync(root) + sentinel = root / "sentinel" + sentinel.write_text("do not touch", encoding="utf-8") + (destination / ".data_get_sync.lock").symlink_to(sentinel) + with self.assertRaises(RuntimeError): + syncer.run_once() + self.assertEqual(sentinel.read_text(encoding="utf-8"), "do not touch") + + def test_delete_mode_rejects_nonfixed_remote_ready(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + with self.assertRaises(ValueError): + DataGetSync( + remote="nvidia@test", + remote_ready="/tmp/not-ready", + destination=root, + status_file=root / "status.json", + delete_remote_after_sync=True, + ) def test_bad_old_episode_does_not_starve_newer_episode(self) -> None: with tempfile.TemporaryDirectory() as temporary: diff --git a/tg3_data_collection/tg3-data-get-sync.service b/tg3_data_collection/tg3-data-get-sync.service index 6c092d7..90c19ca 100644 --- a/tg3_data_collection/tg3-data-get-sync.service +++ b/tg3_data_collection/tg3-data-get-sync.service @@ -4,7 +4,7 @@ Description=TG3 completed data episode sync to Data_Get [Service] Type=simple WorkingDirectory=/home/ps/Desktop/TG3_TS1P_OmniSocket_Teleop -ExecStart=/usr/bin/python3 /home/ps/Desktop/TG3_TS1P_OmniSocket_Teleop/tg3_data_collection/data_get_sync.py --remote nvidia@192.168.41.2 --remote-ready /home/nvidia/tg3_data_collection/ready --destination /home/ps/Desktop/TG3_TS1P_OmniSocket_Teleop/Data_Get --poll-seconds 2 +ExecStart=/usr/bin/python3 /home/ps/Desktop/TG3_TS1P_OmniSocket_Teleop/tg3_data_collection/data_get_sync.py --remote nvidia@192.168.41.2 --remote-ready /home/nvidia/tg3_data_collection/ready --destination /home/ps/Desktop/TG3_TS1P_OmniSocket_Teleop/Data_Get --delete-remote-after-sync --poll-seconds 2 Restart=always RestartSec=2 KillSignal=SIGINT diff --git a/tg3_local_teleop/README.md b/tg3_local_teleop/README.md index e1f93af..0522efe 100644 --- a/tg3_local_teleop/README.md +++ b/tg3_local_teleop/README.md @@ -167,21 +167,19 @@ OmniSocket 进程；在此之前桥会保持最后一帧目标。未 START 或 为避免 1280x720 原始 RGB+深度每台相机约 `8.3 GB/min` 的持续写盘负载，默认录制 `color/image_raw/compressed`（JPEG）和 `depth/image_raw/compressedDepth`（16UC1 无损 PNG）。现场单台腰部相机实测合计约 `22–25 MB/s`（约 `1.3–1.5 GB/min`，随 -场景变化）。腰部图像、内参、Metadata 和 `/tf_static` 是 required topics；两路图像还 -必须在整条 bag 内分别达到至少 `20 Hz` 平均频率。相机服务未发布或图像严重掉流时， -L3 启动或最终校验会失败，不会生成缺图像却标记 READY 的 episode。头部话题在白名单 -中但不是 required：当前 -`orbbec_head.service` 未启动不影响腰部数采；将来在 episode 开始前启动后，rosbag 会 -自动发现并采集，无需再改配置。静态 TF 会被保存用于相机外参，但当前收尾只验证该 -Topic 非空，不解析每条 TF 的 frame 语义。 +场景变化）。头、腰两组相机都不是 required topics：服务在运行就由 rosbag 自动发现并 +采集，未运行不阻塞 L3 或核心 READY。manifest 对每组记录 `absent`、`healthy`、 +`partial` 或 `low_rate`，并保存各话题消息数和图像平均频率；相机异常会产生明确质量 +警告，但不会丢弃其余机器人控制数据。静态 TF 也尽力采集，不作为启动门槛。 机器人暂存目录： ```text /home/nvidia/tg3_data_collection/ - active/ # 尚未完成，不能取走 - ready/ # 已正常收尾、校验并生成 READY - failed/ # 启动、磁盘、进程或校验失败，保留供诊断 + active/ # 录制期间的临时数据，不能取走 + ready/ # 等待 PS 本机完成持久化验真；随后自动回收 + failed/ # 当前策略不保留失败 payload + deleting/ # 固定根删除助手的短暂原子隔离目录 ``` MCAP 使用 `zstd_fast`、64 MiB cache、每 300 秒分片；每条 episode 最长 30 分钟， @@ -191,6 +189,12 @@ MCAP 使用 `zstd_fast`、64 MiB cache、每 300 秒分片；每条 episode 最 `manifest.json` 和 `READY`，最后才原子进入 `ready/`。Nvidia 需已安装 `python3-yaml` （现场已验证 PyYAML 6.0.1）。 +当前 `retain_failed_episodes=false`：失败原因仍进入 recorder 状态和 journal，但失败 +MCAP 不留在机器人磁盘。成功 episode 必须先在 PS 本机完成逐文件 SHA-256、fsync、 +原子发布并生成 `VERIFIED`，然后由固定根目录删除助手按 episode ID + manifest SHA +精确回收；它不能访问 `active/`、`failed/` 或数据根目录。PS 断线时 `ready/` 会临时保留， +避免尚未落到本机的数据丢失，恢复后自动续传和回收。 + 相机由厂家系统服务管理，本项目只订阅、不自动启停。录制前检查或按需启动： ```bash @@ -224,7 +228,8 @@ find /home/nvidia/tg3_data_collection/ready -mindepth 1 -maxdepth 1 -type d ``` 完成 episode 由 PS 本机服务校验后保存到项目 `Data_Get`，详见 -`../tg3_data_collection/README.md`。机器人 `ready/` 中的副本不会自动删除。 +`../tg3_data_collection/README.md`。本机 `VERIFIED` 持久化后，机器人对应 `ready/` +副本会自动删除；EAI 不保存数采 payload。 ## 完整重启顺序 diff --git a/tg3_local_teleop/config.toml b/tg3_local_teleop/config.toml index 82b7dea..3b1131a 100644 --- a/tg3_local_teleop/config.toml +++ b/tg3_local_teleop/config.toml @@ -75,20 +75,17 @@ control_topic = "/tg3/data_collection/control" status_topic = "/tg3/data_collection/status" iarm_frame_topic = "/tg3/data_collection/iarm_frame" base_directory = "/home/nvidia/tg3_data_collection" +# Failed/incomplete payloads are discarded after the error is reported through +# recorder status and journal. Completed READY episodes remain only until the +# PS sync has durably verified its local copy and requests exact-episode removal. +retain_failed_episodes = false minimum_free_gib = 100.0 max_duration_seconds = 1800.0 -# A camera endpoint merely existing is insufficient: completed episodes must -# also sustain this conservative average rate across the rosbag data span. -# The driver targets 30 Hz; 20 Hz tolerates startup/discovery jitter while -# rejecting a stream that only delivered a few frames or stopped mid-episode. -minimum_topic_rates_hz."/ob_camera_waist/color/image_raw/compressed" = 20.0 -minimum_topic_rates_hz."/ob_camera_waist/depth/image_raw/compressedDepth" = 20.0 # Explicit data whitelist; never replace this with `-a` on the live robot. # RGB is recorded through image_transport JPEG and 16UC1 depth through its -# lossless compressedDepth PNG transport. Waist RGB-D is required because that -# camera is enabled on this robot. Head RGB-D stays optional: rosbag discovers -# and records it automatically whenever orbbec_head.service is running. +# lossless compressedDepth PNG transport. Both RGB-D cameras are optional: +# rosbag discovers and records either one whenever its service is running. # /robot_state is the authoritative measured robot state. topics = [ "/robot_state", @@ -137,18 +134,43 @@ required_topics = [ "/left_hand/motor_status", "/right_hand/motor_status", "/hric/robot/rl_state", + "/tg3/data_collection/control", + "/tg3/data_collection/status", + "/tg3/data_collection/iarm_frame", +] + +# Optional sensor groups never block capture or core READY. Their message +# counts and whole-episode average rates are recorded in manifest.json as +# absent/healthy/partial/low_rate quality observations. The Orbbec driver +# targets 30 Hz; 20 Hz leaves room for discovery and compression jitter. +[data_collection.optional_topic_groups.head_rgbd] +topics = [ + "/ob_camera_head/color/image_raw/compressed", + "/ob_camera_head/color/camera_info", + "/ob_camera_head/color/metadata", + "/ob_camera_head/depth/image_raw/compressedDepth", + "/ob_camera_head/depth/camera_info", + "/ob_camera_head/depth/metadata", +] + +[data_collection.optional_topic_groups.head_rgbd.minimum_topic_rates_hz] +"/ob_camera_head/color/image_raw/compressed" = 20.0 +"/ob_camera_head/depth/image_raw/compressedDepth" = 20.0 + +[data_collection.optional_topic_groups.waist_rgbd] +topics = [ "/ob_camera_waist/color/image_raw/compressed", "/ob_camera_waist/color/camera_info", "/ob_camera_waist/color/metadata", "/ob_camera_waist/depth/image_raw/compressedDepth", "/ob_camera_waist/depth/camera_info", "/ob_camera_waist/depth/metadata", - "/tf_static", - "/tg3/data_collection/control", - "/tg3/data_collection/status", - "/tg3/data_collection/iarm_frame", ] +[data_collection.optional_topic_groups.waist_rgbd.minimum_topic_rates_hz] +"/ob_camera_waist/color/image_raw/compressed" = 20.0 +"/ob_camera_waist/depth/image_raw/compressedDepth" = 20.0 + [locomotion] # The two original immediate bindings are independent: right C + left-stick # vertical controls translation; left Z + right-stick horizontal controls diff --git a/tg3_local_teleop/data_collection.py b/tg3_local_teleop/data_collection.py index 279bed7..15b6ff6 100755 --- a/tg3_local_teleop/data_collection.py +++ b/tg3_local_teleop/data_collection.py @@ -32,6 +32,15 @@ import yaml ToggleAction = Literal["start", "stop"] +def _valid_ros_topic(topic: Any) -> bool: + return ( + isinstance(topic, str) + and topic.startswith("/") + and topic.strip() == topic + and not any(character.isspace() for character in topic) + ) + + def left_joystick_pressed(data: Mapping[str, Any]) -> bool | None: """Strictly parse the live-frame ``button_joystick.left`` value. @@ -195,6 +204,55 @@ class RecordingToggleGate: return "active" if self.active else "idle" +@dataclass(frozen=True) +class OptionalTopicGroupConfig: + """Non-fatal completeness and rate checks for an optional sensor group.""" + + topics: Sequence[str] + minimum_topic_rates_hz: Mapping[str, float] = field(default_factory=dict) + + def __post_init__(self) -> None: + topics = ( + tuple(self.topics) + if isinstance(self.topics, Sequence) + and not isinstance(self.topics, str) + else () + ) + if not topics: + raise ValueError("optional topic group must contain at least one topic") + if any(not _valid_ros_topic(topic) for topic in topics): + raise ValueError( + "optional topic group topics must be absolute ROS topic names" + ) + if len(set(topics)) != len(topics): + raise ValueError("optional topic group topics must not contain duplicates") + if not isinstance(self.minimum_topic_rates_hz, Mapping): + raise ValueError("optional minimum topic rates must be a mapping") + rates: dict[str, float] = {} + for topic, rate in self.minimum_topic_rates_hz.items(): + if topic not in topics: + raise ValueError( + "optional minimum-rate topics must be a subset of group topics" + ) + if ( + isinstance(rate, bool) + or not isinstance(rate, (int, float)) + or not math.isfinite(rate) + or rate <= 0.0 + ): + raise ValueError( + f"optional minimum topic rate for {topic!r} must be " + "positive and finite" + ) + rates[topic] = float(rate) + object.__setattr__(self, "topics", topics) + object.__setattr__( + self, + "minimum_topic_rates_hz", + MappingProxyType(rates), + ) + + @dataclass(frozen=True) class RecorderConfig: """Static configuration for :class:`DataRecorderManager`.""" @@ -203,6 +261,10 @@ class RecorderConfig: topics: Sequence[str] = () required_topics: Sequence[str] = () minimum_topic_rates_hz: Mapping[str, float] = field(default_factory=dict) + optional_topic_groups: Mapping[str, OptionalTopicGroupConfig] = field( + default_factory=dict + ) + retain_failed_episodes: bool = True minimum_free_bytes: int = 5 * 1024**3 max_duration_seconds: float = 30 * 60.0 poll_interval_seconds: float = 0.1 @@ -219,25 +281,13 @@ class RecorderConfig: minimum_topic_rates: dict[str, float] = {} if not topics: raise ValueError("at least one recording topic is required") - if any( - not isinstance(topic, str) - or not topic.startswith("/") - or topic.strip() != topic - or any(character.isspace() for character in topic) - for topic in topics - ): + if any(not _valid_ros_topic(topic) for topic in topics): raise ValueError("recording topics must be absolute ROS topic names") if len(set(topics)) != len(topics): raise ValueError("recording topics must not contain duplicates") if len(set(required_topics)) != len(required_topics): raise ValueError("required topics must not contain duplicates") - if any( - not isinstance(topic, str) - or not topic.startswith("/") - or topic.strip() != topic - or any(character.isspace() for character in topic) - for topic in required_topics - ): + if any(not _valid_ros_topic(topic) for topic in required_topics): raise ValueError("required topics must be absolute ROS topic names") if not set(required_topics).issubset(topics): raise ValueError("required topics must be a subset of recording topics") @@ -258,6 +308,45 @@ class RecorderConfig: f"minimum topic rate for {topic!r} must be positive and finite" ) minimum_topic_rates[topic] = float(rate) + if not isinstance(self.optional_topic_groups, Mapping): + raise ValueError("optional topic groups must be a mapping") + optional_groups: dict[str, OptionalTopicGroupConfig] = {} + grouped_topics: set[str] = set() + for name, group in self.optional_topic_groups.items(): + if ( + not isinstance(name, str) + or not name + or len(name) > 64 + or any( + character + not in "-_0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz" + for character in name + ) + ): + raise ValueError( + "optional topic group names must contain 1-64 safe characters" + ) + if not isinstance(group, OptionalTopicGroupConfig): + raise ValueError( + f"optional topic group {name!r} has an invalid configuration" + ) + group_topics = set(group.topics) + if not group_topics.issubset(topics): + raise ValueError( + f"optional topic group {name!r} must be a subset of recording topics" + ) + if group_topics.intersection(required_topics): + raise ValueError( + f"optional topic group {name!r} must be disjoint from required topics" + ) + overlap = group_topics.intersection(grouped_topics) + if overlap: + raise ValueError( + "optional topic groups must be disjoint; repeated topics: " + + ", ".join(sorted(overlap)) + ) + grouped_topics.update(group_topics) + optional_groups[name] = group if ( isinstance(self.minimum_free_bytes, bool) or not isinstance(self.minimum_free_bytes, int) @@ -277,6 +366,8 @@ class RecorderConfig: raise ValueError("ROS 2 executable must be a non-empty string") if type(self.validate_bag_info) is not bool: raise ValueError("validate bag info must be a bool") + if type(self.retain_failed_episodes) is not bool: + raise ValueError("retain failed episodes must be a bool") object.__setattr__(self, "base_directory", base) object.__setattr__(self, "topics", topics) object.__setattr__(self, "required_topics", required_topics) @@ -285,6 +376,11 @@ class RecorderConfig: "minimum_topic_rates_hz", MappingProxyType(minimum_topic_rates), ) + object.__setattr__( + self, + "optional_topic_groups", + MappingProxyType(optional_groups), + ) class RecorderProcess(Protocol): @@ -325,6 +421,8 @@ class _Episode: required_topic_message_counts: dict[str, int] | None = None minimum_topic_message_counts: dict[str, int] | None = None observed_topic_rates_hz: dict[str, float] | None = None + optional_topic_groups: dict[str, Any] | None = None + data_quality_warnings: list[str] | None = None metadata_duration_nanoseconds: int | None = None process: RecorderProcess | None = None stdout_stream: Any = None @@ -767,6 +865,8 @@ class DataRecorderManager: episode.minimum_topic_message_counts ), "observed_topic_rates_hz": episode.observed_topic_rates_hz, + "optional_topic_groups": episode.optional_topic_groups, + "data_quality_warnings": episode.data_quality_warnings, "metadata_duration_nanoseconds": ( episode.metadata_duration_nanoseconds ), @@ -859,6 +959,7 @@ class DataRecorderManager: topic: all_counts.get(topic, 0) for topic in self.config.topics } episode.required_topic_message_counts = required_counts + self._observe_optional_topic_groups(episode, information, all_counts) missing = [ topic for topic in self.config.required_topics if topic not in all_counts ] @@ -877,6 +978,101 @@ class DataRecorderManager: self._validate_minimum_topic_rates(episode, information, all_counts) return required_counts + def _observe_optional_topic_groups( + self, + episode: _Episode, + information: Mapping[str, Any], + all_counts: Mapping[str, int], + ) -> None: + """Classify optional sensor quality without invalidating core data.""" + + duration_nanoseconds: int | None = None + duration = information.get("duration") + if isinstance(duration, Mapping): + candidate = duration.get("nanoseconds") + if ( + type(candidate) is int + and candidate > 0 + ): + duration_nanoseconds = candidate + episode.metadata_duration_nanoseconds = candidate + duration_seconds = ( + None + if duration_nanoseconds is None + else duration_nanoseconds / 1_000_000_000.0 + ) + + observations: dict[str, Any] = {} + warnings: list[str] = [] + for name, group in self.config.optional_topic_groups.items(): + counts = { + topic: int(all_counts.get(topic, 0)) for topic in group.topics + } + empty_topics = [ + topic for topic, count in counts.items() if count <= 0 + ] + all_absent = len(empty_topics) == len(group.topics) + minimum_counts: dict[str, int | None] = {} + observed_rates: dict[str, float | None] = {} + below_rate: list[str] = [] + for topic, minimum_rate in group.minimum_topic_rates_hz.items(): + if duration_seconds is None: + minimum_counts[topic] = None + observed_rates[topic] = None + if not all_absent: + below_rate.append(topic) + continue + minimum_count = math.floor(duration_seconds * minimum_rate) + observed_rate = counts[topic] / duration_seconds + minimum_counts[topic] = minimum_count + observed_rates[topic] = observed_rate + if not all_absent and counts[topic] < minimum_count: + below_rate.append(topic) + + if all_absent: + state = "absent" + elif empty_topics: + state = "partial" + warnings.append( + f"optional topic group {name!r} is partial; zero-message " + "topics: " + ", ".join(empty_topics) + ) + elif below_rate: + state = "low_rate" + if duration_seconds is None: + warnings.append( + f"optional topic group {name!r} rate could not be " + "validated because bag duration is unavailable" + ) + else: + details = [ + f"{topic}={observed_rates[topic]:.3f}Hz<" + f"{group.minimum_topic_rates_hz[topic]:.3f}Hz" + for topic in below_rate + ] + warnings.append( + f"optional topic group {name!r} is below its observed " + "minimum rate: " + "; ".join(details) + ) + else: + state = "healthy" + + observations[name] = { + "state": state, + "topics": list(group.topics), + "topic_message_counts": counts, + "minimum_topic_rates_hz": dict( + group.minimum_topic_rates_hz + ), + "minimum_topic_message_counts": minimum_counts, + "observed_topic_rates_hz": observed_rates, + "zero_message_topics": empty_topics, + "below_minimum_rate_topics": below_rate, + } + + episode.optional_topic_groups = observations + episode.data_quality_warnings = warnings + def _validate_minimum_topic_rates( self, episode: _Episode, @@ -1036,6 +1232,31 @@ class DataRecorderManager: ) -> tuple[Path | None, str]: if not episode.active_directory.exists(): return None, "active episode directory is missing" + if not self.config.retain_failed_episodes: + try: + active_root = Path(self.config.base_directory) / "active" + target = episode.active_directory + if ( + target.parent != active_root + or target.name != episode.request.episode_id + or target.is_symlink() + or active_root.is_symlink() + ): + raise RecordingError( + "refusing to discard a failed episode outside its " + "fixed active root" + ) + shutil.rmtree(target) + try: + self._sync_directory(active_root) + except OSError: + pass + return None, "" + except Exception as discard_error: + return None, ( + "failed to discard project-owned failed episode: " + f"{type(discard_error).__name__}: {discard_error}" + ) try: ready_marker = episode.active_directory / "READY" if ready_marker.exists(): @@ -1063,6 +1284,8 @@ class DataRecorderManager: episode.minimum_topic_message_counts ), "observed_topic_rates_hz": episode.observed_topic_rates_hz, + "optional_topic_groups": episode.optional_topic_groups, + "data_quality_warnings": episode.data_quality_warnings, "metadata_duration_nanoseconds": ( episode.metadata_duration_nanoseconds ), @@ -1242,6 +1465,7 @@ class DataRecorderManager: __all__ = [ "DataRecorderManager", + "OptionalTopicGroupConfig", "RecorderConfig", "RecordingError", "RecordingToggleGate", diff --git a/tg3_local_teleop/data_recorder_node.py b/tg3_local_teleop/data_recorder_node.py index 43282bd..dbbac6b 100755 --- a/tg3_local_teleop/data_recorder_node.py +++ b/tg3_local_teleop/data_recorder_node.py @@ -29,7 +29,11 @@ from rclpy.qos import ( from rclpy.utilities import remove_ros_args from std_msgs.msg import String -from data_collection import DataRecorderManager, RecorderConfig +from data_collection import ( + DataRecorderManager, + OptionalTopicGroupConfig, + RecorderConfig, +) from data_recorder_protocol import ( RecorderControlProtocol, topics_without_publishers, @@ -102,6 +106,41 @@ def _minimum_topic_rates(value: Any) -> dict[str, float]: return result +def _optional_topic_groups( + value: Any, +) -> dict[str, OptionalTopicGroupConfig]: + name = "optional_topic_groups" + if not isinstance(value, dict): + raise ValueError(f"data_collection.{name} must be a TOML table") + result: dict[str, OptionalTopicGroupConfig] = {} + for group_name, raw_group in value.items(): + if not isinstance(raw_group, dict): + raise ValueError( + f"data_collection.{name}.{group_name} must be a TOML table" + ) + unknown = sorted( + set(raw_group) - {"topics", "minimum_topic_rates_hz"} + ) + if unknown: + raise ValueError( + f"data_collection.{name}.{group_name} has unknown keys: " + + ", ".join(unknown) + ) + topics = _topic_list( + raw_group.get("topics", []), + f"{name}.{group_name}.topics", + allow_empty=False, + ) + rates = _minimum_topic_rates( + raw_group.get("minimum_topic_rates_hz", {}) + ) + result[group_name] = OptionalTopicGroupConfig( + topics=topics, + minimum_topic_rates_hz=rates, + ) + return result + + @dataclass(frozen=True) class SupervisorSettings: enabled: bool @@ -150,6 +189,9 @@ def load_settings(path: Path | str) -> SupervisorSettings: minimum_topic_rates = _minimum_topic_rates( section.get("minimum_topic_rates_hz", {}) ) + optional_topic_groups = _optional_topic_groups( + section.get("optional_topic_groups", {}) + ) unknown_rate_topics = sorted( set(minimum_topic_rates) - set(required_topics) ) @@ -187,6 +229,11 @@ def load_settings(path: Path | str) -> SupervisorSettings: topics=topics, required_topics=required_topics, minimum_topic_rates_hz=minimum_topic_rates, + optional_topic_groups=optional_topic_groups, + retain_failed_episodes=_boolean( + section.get("retain_failed_episodes", True), + "retain_failed_episodes", + ), minimum_free_bytes=int(minimum_free_gib * 1024**3), max_duration_seconds=max_duration, poll_interval_seconds=0.1, diff --git a/tg3_local_teleop/delete_ready_episode.py b/tg3_local_teleop/delete_ready_episode.py new file mode 100755 index 0000000..52d7bc0 --- /dev/null +++ b/tg3_local_teleop/delete_ready_episode.py @@ -0,0 +1,537 @@ +#!/usr/bin/env python3 +"""Safely and idempotently remove one verified recorder ``ready`` episode. + +This helper intentionally has no command-line option for either root directory. +It can only move a direct child of the fixed recorder ``ready`` directory into +the fixed ``deleting`` staging directory and remove that tombstone. The caller +must provide both the episode id and the SHA-256 of the exact manifest that was +verified and durably published on the PS workstation. + +Before recursive removal, an authorization ledger is fsynced in the fixed +``delete_ledger`` sibling directory. It survives partial tombstone deletion and +is removed only after the tree cleanup itself is durable. +""" + +from __future__ import annotations + +import argparse +import fcntl +import hashlib +import json +import os +import re +import stat +import sys +import uuid +from pathlib import Path +from typing import Any + + +BASE_ROOT = Path("/home/nvidia/tg3_data_collection") +READY_ROOT = BASE_ROOT / "ready" +DELETING_ROOT = BASE_ROOT / "deleting" +LEDGER_ROOT = BASE_ROOT / "delete_ledger" +SAFE_EPISODE_NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,127}$") +SAFE_SHA256 = re.compile(r"^[0-9a-f]{64}$") +MAX_MANIFEST_BYTES = 16 * 1024 * 1024 +MAX_LEDGER_BYTES = 64 * 1024 +LEDGER_SCHEMA_VERSION = 1 + + +class DeleteSafetyError(RuntimeError): + """The requested deletion did not satisfy the fixed safety contract.""" + + +class DeleteCollisionError(DeleteSafetyError): + """A conflicting ready/deleting episode exists and was left untouched.""" + + +def safe_episode_name(value: str) -> bool: + return bool(SAFE_EPISODE_NAME.fullmatch(value)) and value not in (".", "..") + + +def _open_directory(path: Path) -> int: + flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0) + try: + descriptor = os.open(path, flags) + except OSError as exc: + raise DeleteSafetyError(f"unsafe or missing directory: {path}: {exc}") from exc + try: + metadata = os.fstat(descriptor) + if not stat.S_ISDIR(metadata.st_mode): + raise DeleteSafetyError(f"not a directory: {path}") + return descriptor + except Exception: + os.close(descriptor) + raise + + +def _open_child_directory(parent_fd: int, name: str) -> int: + flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0) + try: + descriptor = os.open(name, flags, dir_fd=parent_fd) + except OSError as exc: + raise DeleteSafetyError(f"unsafe or missing episode directory: {name}: {exc}") from exc + try: + if not stat.S_ISDIR(os.fstat(descriptor).st_mode): + raise DeleteSafetyError(f"episode is not a directory: {name}") + return descriptor + except Exception: + os.close(descriptor) + raise + + +def _read_regular_file(parent_fd: int, name: str, *, max_bytes: int) -> bytes: + flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) + try: + descriptor = os.open(name, flags, dir_fd=parent_fd) + except OSError as exc: + raise DeleteSafetyError(f"missing or unsafe regular file {name!r}: {exc}") from exc + try: + metadata = os.fstat(descriptor) + if not stat.S_ISREG(metadata.st_mode): + raise DeleteSafetyError(f"not a regular file: {name}") + if metadata.st_size > max_bytes: + raise DeleteSafetyError(f"file is too large: {name}") + chunks: list[bytes] = [] + remaining = max_bytes + 1 + while remaining: + block = os.read(descriptor, min(1024 * 1024, remaining)) + if not block: + break + chunks.append(block) + remaining -= len(block) + payload = b"".join(chunks) + if len(payload) > max_bytes: + raise DeleteSafetyError(f"file is too large: {name}") + return payload + finally: + os.close(descriptor) + + +def _validate_tree_entries(directory_fd: int) -> None: + """Require the quarantined recorder tree to contain only dirs/files.""" + + for child in os.listdir(directory_fd): + if child in (".", "..") or "/" in child or "\x00" in child: + raise DeleteSafetyError(f"unsafe directory entry: {child!r}") + metadata = os.stat(child, dir_fd=directory_fd, follow_symlinks=False) + if stat.S_ISDIR(metadata.st_mode): + child_fd = _open_child_directory(directory_fd, child) + try: + _validate_tree_entries(child_fd) + finally: + os.close(child_fd) + elif not stat.S_ISREG(metadata.st_mode): + raise DeleteSafetyError(f"episode contains a symlink/special entry: {child}") + + +def _ledger_name(episode: str, manifest_sha256: str) -> str: + return f"{episode}.{manifest_sha256}.authorized.json" + + +def _ledger_payload( + episode: str, manifest_sha256: str, tombstone: str +) -> dict[str, Any]: + return { + "schema_version": LEDGER_SCHEMA_VERSION, + "state": "AUTHORIZED_DELETE", + "episode_id": episode, + "manifest_sha256": manifest_sha256, + "tombstone": tombstone, + } + + +def _validate_ledger( + ledger_fd: int, + ledger_name: str, + *, + episode: str, + manifest_sha256: str, + tombstone: str, +) -> None: + raw = _read_regular_file(ledger_fd, ledger_name, max_bytes=MAX_LEDGER_BYTES) + try: + payload: Any = json.loads(raw.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise DeleteSafetyError(f"invalid deletion authorization ledger: {exc}") from exc + if payload != _ledger_payload(episode, manifest_sha256, tombstone): + raise DeleteSafetyError("deletion authorization ledger identity mismatch") + + +def _write_ledger( + ledger_fd: int, + ledger_name: str, + *, + episode: str, + manifest_sha256: str, + tombstone: str, +) -> None: + payload = ( + json.dumps( + _ledger_payload(episode, manifest_sha256, tombstone), + sort_keys=True, + separators=(",", ":"), + ) + + "\n" + ).encode("utf-8") + temporary = f".{ledger_name}.{uuid.uuid4().hex}.tmp" + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0) + descriptor = os.open(temporary, flags, 0o600, dir_fd=ledger_fd) + try: + view = memoryview(payload) + while view: + written = os.write(descriptor, view) + if written <= 0: + raise OSError("short write while creating deletion ledger") + view = view[written:] + os.fsync(descriptor) + finally: + os.close(descriptor) + try: + # A hard-link publish is atomic and refuses to replace any existing + # sidecar. The helper flock serializes legitimate writers as well. + os.link( + temporary, + ledger_name, + src_dir_fd=ledger_fd, + dst_dir_fd=ledger_fd, + follow_symlinks=False, + ) + os.fsync(ledger_fd) + finally: + try: + os.unlink(temporary, dir_fd=ledger_fd) + os.fsync(ledger_fd) + except FileNotFoundError: + pass + + +def _validate_episode( + parent_fd: int, + directory_name: str, + *, + episode: str, + expected_manifest_sha256: str, +) -> None: + episode_fd = _open_child_directory(parent_fd, directory_name) + try: + # READY must be a real regular file. Its contents are deliberately not + # trusted as a deletion credential; the exact manifest digest is. + _read_regular_file(episode_fd, "READY", max_bytes=4096) + raw_manifest = _read_regular_file( + episode_fd, "manifest.json", max_bytes=MAX_MANIFEST_BYTES + ) + actual_digest = hashlib.sha256(raw_manifest).hexdigest() + if actual_digest != expected_manifest_sha256: + raise DeleteSafetyError( + "manifest SHA-256 mismatch: " + f"expected {expected_manifest_sha256}, got {actual_digest}" + ) + try: + manifest: Any = json.loads(raw_manifest.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise DeleteSafetyError(f"invalid manifest JSON: {exc}") from exc + if not isinstance(manifest, dict): + raise DeleteSafetyError("manifest root is not an object") + if manifest.get("state") != "complete": + raise DeleteSafetyError("manifest state is not complete") + if manifest.get("episode_id") != episode: + raise DeleteSafetyError("manifest episode_id does not match target") + _validate_tree_entries(episode_fd) + finally: + os.close(episode_fd) + + +def _remove_tree_at(parent_fd: int, name: str) -> None: + """Remove one child tree without following any symbolic link.""" + + directory_fd = _open_child_directory(parent_fd, name) + try: + for child in os.listdir(directory_fd): + if child in (".", "..") or "/" in child or "\x00" in child: + raise DeleteSafetyError(f"unsafe directory entry: {child!r}") + metadata = os.stat(child, dir_fd=directory_fd, follow_symlinks=False) + if stat.S_ISDIR(metadata.st_mode): + _remove_tree_at(directory_fd, child) + else: + # Symlinks and all non-directory entries are unlinked, never + # followed. Recorder output is expected to contain only files. + os.unlink(child, dir_fd=directory_fd) + os.fsync(directory_fd) + finally: + os.close(directory_fd) + os.rmdir(name, dir_fd=parent_fd) + + +def _child_kind(parent_fd: int, name: str) -> str: + try: + metadata = os.stat(name, dir_fd=parent_fd, follow_symlinks=False) + except FileNotFoundError: + return "absent" + if stat.S_ISDIR(metadata.st_mode): + return "directory" + if stat.S_ISLNK(metadata.st_mode): + return "symlink" + if stat.S_ISREG(metadata.st_mode): + return "regular" + return "other" + + +def delete_ready_episode( + episode: str, + manifest_sha256: str, + *, + ready_root: Path = READY_ROOT, + deleting_root: Path = DELETING_ROOT, + ledger_root: Path | None = None, +) -> dict[str, str]: + """Compare, atomically quarantine, and remove exactly one ready episode. + + Root arguments are injectable only for unit tests. The CLI below always + uses the immutable production constants, including a ledger directory that + is outside the recursively removed ``deleting`` tree. + """ + + if not safe_episode_name(episode): + raise DeleteSafetyError(f"unsafe episode id: {episode!r}") + if not SAFE_SHA256.fullmatch(manifest_sha256): + raise DeleteSafetyError("manifest SHA-256 must be 64 lowercase hex characters") + + ready_root = Path(ready_root) + deleting_root = Path(deleting_root) + ledger_root = Path(ledger_root or (deleting_root.parent / LEDGER_ROOT.name)) + if not ( + ready_root.parent == deleting_root.parent == ledger_root.parent + ): + raise DeleteSafetyError("ready, deleting, and ledger roots must be siblings") + base_root = ready_root.parent + + # Refuse a symlinked/wrong parent even in tests. resolve(strict=True) also + # proves that all existing parent components resolve to this exact path. + try: + if base_root.resolve(strict=True) != base_root.absolute(): + raise DeleteSafetyError("base root must not contain symlinks") + if ready_root.resolve(strict=True) != ready_root.absolute(): + raise DeleteSafetyError("ready root must not contain symlinks") + except FileNotFoundError as exc: + raise DeleteSafetyError(f"fixed recorder root is missing: {exc}") from exc + + base_fd = _open_directory(base_root) + ready_fd = _open_directory(ready_root) + deleting_fd: int | None = None + ledger_fd: int | None = None + lock_fd: int | None = None + try: + base_stat = os.fstat(base_fd) + ready_stat = os.fstat(ready_fd) + if base_stat.st_dev != ready_stat.st_dev: + raise DeleteSafetyError("ready root is not on the recorder filesystem") + + deleting_kind = _child_kind(base_fd, deleting_root.name) + if deleting_kind == "absent": + os.mkdir(deleting_root.name, mode=0o750, dir_fd=base_fd) + os.fsync(base_fd) + elif deleting_kind != "directory": + raise DeleteSafetyError("deleting root is not a real directory") + deleting_fd = _open_directory(deleting_root) + if os.fstat(deleting_fd).st_dev != ready_stat.st_dev: + raise DeleteSafetyError("deleting root is not on the ready filesystem") + + ledger_kind = _child_kind(base_fd, ledger_root.name) + if ledger_kind == "absent": + os.mkdir(ledger_root.name, mode=0o750, dir_fd=base_fd) + os.fsync(base_fd) + elif ledger_kind != "directory": + raise DeleteSafetyError("delete ledger root is not a real directory") + ledger_fd = _open_directory(ledger_root) + if os.fstat(ledger_fd).st_dev != ready_stat.st_dev: + raise DeleteSafetyError("delete ledger is not on the ready filesystem") + + lock_flags = os.O_RDWR | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0) + lock_fd = os.open(".delete_ready.lock", lock_flags, 0o600, dir_fd=deleting_fd) + if not stat.S_ISREG(os.fstat(lock_fd).st_mode): + raise DeleteSafetyError("deletion lock is not a regular file") + fcntl.flock(lock_fd, fcntl.LOCK_EX) + + tombstone = f"{episode}.{manifest_sha256}.deleting" + authorization = _ledger_name(episode, manifest_sha256) + ready_kind = _child_kind(ready_fd, episode) + tombstone_kind = _child_kind(deleting_fd, tombstone) + authorization_kind = _child_kind(ledger_fd, authorization) + + # Any other tombstone for this episode is a manifest-identity collision. + prefix = f"{episode}." + collisions = [ + name + for name in os.listdir(deleting_fd) + if name.startswith(prefix) + and name.endswith(".deleting") + and name != tombstone + ] + if collisions: + raise DeleteCollisionError( + f"conflicting deleting tombstone(s) exist: {sorted(collisions)!r}" + ) + ledger_collisions = [ + name + for name in os.listdir(ledger_fd) + if name.startswith(prefix) + and name.endswith(".authorized.json") + and name != authorization + ] + if ledger_collisions: + raise DeleteCollisionError( + "conflicting deletion authorization(s) exist: " + f"{sorted(ledger_collisions)!r}" + ) + if ready_kind not in ("absent", "directory"): + raise DeleteSafetyError(f"ready target is unsafe: {ready_kind}") + if tombstone_kind not in ("absent", "directory"): + raise DeleteSafetyError(f"deleting target is unsafe: {tombstone_kind}") + if authorization_kind not in ("absent", "regular"): + raise DeleteSafetyError( + f"deletion authorization is unsafe: {authorization_kind}" + ) + if ready_kind == "directory" and tombstone_kind == "directory": + raise DeleteCollisionError("ready target and deleting tombstone both exist") + if ready_kind == "directory" and authorization_kind == "regular": + raise DeleteCollisionError( + "ready target exists alongside a stale deletion authorization" + ) + + if ready_kind == "directory": + _validate_episode( + ready_fd, + episode, + episode=episode, + expected_manifest_sha256=manifest_sha256, + ) + os.rename( + episode, + tombstone, + src_dir_fd=ready_fd, + dst_dir_fd=deleting_fd, + ) + os.fsync(ready_fd) + os.fsync(deleting_fd) + # Re-validate after the rename. If the source name was exchanged + # between the first comparison and rename, the unexpected tree is + # quarantined but is never unlinked. + _validate_episode( + deleting_fd, + tombstone, + episode=episode, + expected_manifest_sha256=manifest_sha256, + ) + _write_ledger( + ledger_fd, + authorization, + episode=episode, + manifest_sha256=manifest_sha256, + tombstone=tombstone, + ) + authorization_kind = "regular" + result_state = "deleted" + elif tombstone_kind == "directory": + if authorization_kind == "regular": + # The durable out-of-tree ledger remains usable even if an + # interrupted recursive removal already removed READY and the + # manifest from inside the tombstone. + _validate_ledger( + ledger_fd, + authorization, + episode=episode, + manifest_sha256=manifest_sha256, + tombstone=tombstone, + ) + else: + # A crash between rename and ledger publication is recoverable + # only while the tombstone still proves its original identity. + _validate_episode( + deleting_fd, + tombstone, + episode=episode, + expected_manifest_sha256=manifest_sha256, + ) + _write_ledger( + ledger_fd, + authorization, + episode=episode, + manifest_sha256=manifest_sha256, + tombstone=tombstone, + ) + authorization_kind = "regular" + result_state = "resumed_delete" + else: + if authorization_kind == "regular": + # The tree was fully removed and only the final ledger cleanup + # (or its ACK) was interrupted. + _validate_ledger( + ledger_fd, + authorization, + episode=episode, + manifest_sha256=manifest_sha256, + tombstone=tombstone, + ) + os.unlink(authorization, dir_fd=ledger_fd) + os.fsync(ledger_fd) + return { + "state": "already_absent", + "episode_id": episode, + "manifest_sha256": manifest_sha256, + } + + _remove_tree_at(deleting_fd, tombstone) + os.fsync(deleting_fd) + # Delete authorization is deliberately the final artifact removed. + # If this unlink/ACK is lost, the next invocation validates it and + # safely returns already_absent. + _validate_ledger( + ledger_fd, + authorization, + episode=episode, + manifest_sha256=manifest_sha256, + tombstone=tombstone, + ) + os.unlink(authorization, dir_fd=ledger_fd) + os.fsync(ledger_fd) + return { + "state": result_state, + "episode_id": episode, + "manifest_sha256": manifest_sha256, + } + finally: + if lock_fd is not None: + os.close(lock_fd) + if deleting_fd is not None: + os.close(deleting_fd) + if ledger_fd is not None: + os.close(ledger_fd) + os.close(ready_fd) + os.close(base_fd) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("episode_id") + parser.add_argument("manifest_sha256") + return parser.parse_args() + + +def main() -> int: + args = parse_args() + try: + result = delete_ready_episode(args.episode_id, args.manifest_sha256) + except DeleteCollisionError as exc: + print(json.dumps({"state": "collision", "error": str(exc)})) + return 3 + except (DeleteSafetyError, OSError) as exc: + print(json.dumps({"state": "refused", "error": str(exc)})) + return 2 + print(json.dumps(result, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tg3_local_teleop/test_data_collection.py b/tg3_local_teleop/test_data_collection.py index 48caf57..467a512 100755 --- a/tg3_local_teleop/test_data_collection.py +++ b/tg3_local_teleop/test_data_collection.py @@ -15,12 +15,45 @@ from typing import Any from data_collection import ( DataRecorderManager, + OptionalTopicGroupConfig, RecorderConfig, RecordingToggleGate, left_joystick_pressed, ) +HEAD_CAMERA_TOPICS = ( + "/ob_camera_head/color/image_raw/compressed", + "/ob_camera_head/color/camera_info", + "/ob_camera_head/color/metadata", + "/ob_camera_head/depth/image_raw/compressedDepth", + "/ob_camera_head/depth/camera_info", + "/ob_camera_head/depth/metadata", +) +WAIST_CAMERA_TOPICS = tuple( + topic.replace("_head", "_waist") for topic in HEAD_CAMERA_TOPICS +) + + +def optional_camera_groups() -> dict[str, OptionalTopicGroupConfig]: + return { + "head_rgbd": OptionalTopicGroupConfig( + topics=HEAD_CAMERA_TOPICS, + minimum_topic_rates_hz={ + HEAD_CAMERA_TOPICS[0]: 20.0, + HEAD_CAMERA_TOPICS[3]: 20.0, + }, + ), + "waist_rgbd": OptionalTopicGroupConfig( + topics=WAIST_CAMERA_TOPICS, + minimum_topic_rates_hz={ + WAIST_CAMERA_TOPICS[0]: 20.0, + WAIST_CAMERA_TOPICS[3]: 20.0, + }, + ), + } + + class LeftJoystickParserTest(unittest.TestCase): def test_accepts_only_live_bool_or_binary_integer(self) -> None: self.assertIs( @@ -291,6 +324,10 @@ class DataRecorderManagerTest(unittest.TestCase): topics: Sequence[str] = ("/joint_states", "/tf"), required_topics: Sequence[str] = ("/joint_states", "/tf"), minimum_topic_rates_hz: Mapping[str, float] | None = None, + optional_topic_groups: Mapping[ + str, OptionalTopicGroupConfig + ] | None = None, + retain_failed_episodes: bool = True, ) -> tuple[DataRecorderManager, Any]: if factory is None: factory = FakeProcessFactory(**(process_options or {})) @@ -305,6 +342,8 @@ class DataRecorderManagerTest(unittest.TestCase): topics=topics, required_topics=required_topics, minimum_topic_rates_hz=minimum_topic_rates_hz or {}, + optional_topic_groups=optional_topic_groups or {}, + retain_failed_episodes=retain_failed_episodes, minimum_free_bytes=minimum_free, max_duration_seconds=max_duration, poll_interval_seconds=0.005, @@ -330,6 +369,36 @@ class DataRecorderManagerTest(unittest.TestCase): self.fail("condition did not become true before timeout") time.sleep(0.002) + def record_with_optional_cameras( + self, topic_counts: Mapping[str, int] + ) -> tuple[dict[str, Any], dict[str, Any]]: + topics = ( + "/joint_states", + "/tf", + *HEAD_CAMERA_TOPICS, + *WAIST_CAMERA_TOPICS, + ) + manager, _ = self.make_manager( + topics=topics, + required_topics=("/joint_states", "/tf"), + optional_topic_groups=optional_camera_groups(), + process_options={ + "topic_counts": topic_counts, + "duration_nanoseconds": 1_000_000_000, + }, + ) + self.assertTrue(manager.request_start("ca" * 16)) + self.wait_for(lambda: manager.status()["recording"]) + self.assertTrue(manager.request_stop()) + self.assertTrue(manager.wait_until_idle(2.0)) + status = manager.status() + manifest = json.loads( + ( + Path(status["last_episode_directory"]) / "manifest.json" + ).read_text() + ) + return status, manifest + def test_manual_stop_uses_sigint_and_atomically_creates_ready_manifest( self, ) -> None: @@ -627,6 +696,103 @@ class DataRecorderManagerTest(unittest.TestCase): 0 if head_count is None else head_count, ) + def test_optional_cameras_absent_do_not_block_core_ready(self) -> None: + status, manifest = self.record_with_optional_cameras( + {"/joint_states": 25, "/tf": 50} + ) + self.assertEqual(status["last_result"], "ready") + self.assertEqual(manifest["data_quality_warnings"], []) + self.assertEqual( + { + name: observation["state"] + for name, observation in manifest[ + "optional_topic_groups" + ].items() + }, + {"head_rgbd": "absent", "waist_rgbd": "absent"}, + ) + self.assertEqual( + manifest["optional_topic_groups"]["head_rgbd"][ + "below_minimum_rate_topics" + ], + [], + ) + self.assertTrue((Path(status["last_episode_directory"]) / "READY").exists()) + + def test_each_active_optional_camera_is_observed_as_healthy(self) -> None: + for active_name, active_topics in ( + ("head_rgbd", HEAD_CAMERA_TOPICS), + ("waist_rgbd", WAIST_CAMERA_TOPICS), + ): + with self.subTest(active_name=active_name): + counts = {"/joint_states": 25, "/tf": 50} + counts.update({topic: 1 for topic in active_topics}) + counts[active_topics[0]] = 25 + counts[active_topics[3]] = 25 + status, manifest = self.record_with_optional_cameras(counts) + self.assertEqual(status["last_result"], "ready") + observation = manifest["optional_topic_groups"][active_name] + self.assertEqual(observation["state"], "healthy") + self.assertEqual( + observation["observed_topic_rates_hz"], + {active_topics[0]: 25.0, active_topics[3]: 25.0}, + ) + inactive_name = ( + "waist_rgbd" + if active_name == "head_rgbd" + else "head_rgbd" + ) + self.assertEqual( + manifest["optional_topic_groups"][inactive_name]["state"], + "absent", + ) + self.assertEqual(manifest["data_quality_warnings"], []) + + def test_partial_optional_camera_warns_but_remains_ready(self) -> None: + counts = {"/joint_states": 25, "/tf": 50} + counts.update( + { + HEAD_CAMERA_TOPICS[0]: 25, + HEAD_CAMERA_TOPICS[1]: 1, + HEAD_CAMERA_TOPICS[2]: 1, + } + ) + status, manifest = self.record_with_optional_cameras(counts) + self.assertEqual(status["last_result"], "ready") + observation = manifest["optional_topic_groups"]["head_rgbd"] + self.assertEqual(observation["state"], "partial") + self.assertEqual( + observation["zero_message_topics"], + list(HEAD_CAMERA_TOPICS[3:]), + ) + self.assertTrue(manifest["data_quality_warnings"]) + self.assertIn("is partial", manifest["data_quality_warnings"][0]) + self.assertTrue((Path(status["last_episode_directory"]) / "READY").exists()) + + def test_low_rate_optional_camera_warns_but_remains_ready(self) -> None: + counts = {"/joint_states": 25, "/tf": 50} + counts.update({topic: 1 for topic in WAIST_CAMERA_TOPICS}) + counts[WAIST_CAMERA_TOPICS[0]] = 5 + counts[WAIST_CAMERA_TOPICS[3]] = 10 + status, manifest = self.record_with_optional_cameras(counts) + self.assertEqual(status["last_result"], "ready") + observation = manifest["optional_topic_groups"]["waist_rgbd"] + self.assertEqual(observation["state"], "low_rate") + self.assertEqual( + observation["observed_topic_rates_hz"], + { + WAIST_CAMERA_TOPICS[0]: 5.0, + WAIST_CAMERA_TOPICS[3]: 10.0, + }, + ) + self.assertEqual( + observation["below_minimum_rate_topics"], + [WAIST_CAMERA_TOPICS[0], WAIST_CAMERA_TOPICS[3]], + ) + self.assertTrue(manifest["data_quality_warnings"]) + self.assertIn("below", manifest["data_quality_warnings"][0]) + self.assertTrue((Path(status["last_episode_directory"]) / "READY").exists()) + def test_topic_that_stops_mid_episode_fails_average_rate(self) -> None: bag_info = FakeBagInfoRunner() manager, _ = self.make_manager( @@ -779,6 +945,20 @@ class DataRecorderManagerTest(unittest.TestCase): ) self.assertEqual(failure["status"], "failed") + def test_failed_payload_is_discarded_when_retention_is_disabled(self) -> None: + manager, factory = self.make_manager( + free_bytes=lambda _path: 0, + retain_failed_episodes=False, + ) + self.assertTrue(manager.request_start("d0" * 16)) + self.assertTrue(manager.wait_until_idle(2.0)) + status = manager.status() + self.assertEqual(status["last_result"], "failed") + self.assertIsNone(status["last_episode_directory"]) + self.assertEqual(factory.processes, []) + self.assertEqual(list((self.base / "active").iterdir()), []) + self.assertEqual(list((self.base / "failed").iterdir()), []) + def test_configuration_rejects_unsafe_or_ambiguous_values(self) -> None: with self.assertRaises(ValueError): RecorderConfig(self.base, ()) @@ -790,6 +970,8 @@ class DataRecorderManagerTest(unittest.TestCase): RecorderConfig(self.base, ("/ok",), minimum_free_bytes=-1) with self.assertRaises(ValueError): RecorderConfig(self.base, ("/ok",), validate_bag_info=1) + with self.assertRaises(ValueError): + RecorderConfig(self.base, ("/ok",), retain_failed_episodes=1) with self.assertRaises(ValueError): RecorderConfig(self.base, ("/ok",), bag_info_timeout_seconds=0) with self.assertRaises(ValueError): @@ -827,6 +1009,45 @@ class DataRecorderManagerTest(unittest.TestCase): required_topics=("/recorded",), minimum_topic_rates_hz=[], # type: ignore[arg-type] ) + with self.assertRaises(ValueError): + OptionalTopicGroupConfig(topics=()) + with self.assertRaises(ValueError): + OptionalTopicGroupConfig( + topics=("/camera/image",), + minimum_topic_rates_hz={"/other": 20.0}, + ) + camera_group = OptionalTopicGroupConfig( + topics=("/camera/image",), + minimum_topic_rates_hz={"/camera/image": 20.0}, + ) + with self.assertRaises(ValueError): + RecorderConfig( + self.base, + ("/recorded",), + optional_topic_groups={"camera": camera_group}, + ) + with self.assertRaises(ValueError): + RecorderConfig( + self.base, + ("/recorded", "/camera/image"), + required_topics=("/camera/image",), + optional_topic_groups={"camera": camera_group}, + ) + with self.assertRaises(ValueError): + RecorderConfig( + self.base, + ("/recorded", "/camera/image"), + optional_topic_groups={ + "camera_a": camera_group, + "camera_b": camera_group, + }, + ) + with self.assertRaises(ValueError): + RecorderConfig( + self.base, + ("/recorded", "/camera/image"), + optional_topic_groups={"bad name": camera_group}, + ) if __name__ == "__main__": diff --git a/tg3_local_teleop/test_delete_ready_episode.py b/tg3_local_teleop/test_delete_ready_episode.py new file mode 100755 index 0000000..377290c --- /dev/null +++ b/tg3_local_teleop/test_delete_ready_episode.py @@ -0,0 +1,376 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import hashlib +import json +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +from delete_ready_episode import ( + DeleteCollisionError, + DeleteSafetyError, + delete_ready_episode, +) + + +class DeleteReadyEpisodeTests(unittest.TestCase): + def _roots(self, root: Path) -> tuple[Path, Path, Path]: + base = root / "tg3_data_collection" + ready = base / "ready" + deleting = base / "deleting" + active = base / "active" + failed = base / "failed" + for path in (ready, active, failed): + path.mkdir(parents=True, exist_ok=True) + (active / "ACTIVE_SENTINEL").write_text("active", encoding="utf-8") + (failed / "FAILED_SENTINEL").write_text("failed", encoding="utf-8") + return ready, deleting, base + + def _episode( + self, + ready: Path, + name: str = "episode_20260810T120000000_deadbeef", + *, + state: str = "complete", + ) -> tuple[Path, str]: + episode = ready / name + (episode / "bag").mkdir(parents=True) + (episode / "bag" / "bag_0.mcap").write_bytes(b"mcap") + manifest = {"state": state, "episode_id": name, "files": []} + raw = (json.dumps(manifest, sort_keys=True) + "\n").encode("utf-8") + (episode / "manifest.json").write_bytes(raw) + (episode / "READY").write_text("ready\n", encoding="ascii") + return episode, hashlib.sha256(raw).hexdigest() + + def _authorization( + self, + base: Path, + episode: str, + digest: str, + *, + payload_digest: str | None = None, + ) -> Path: + ledger = base / "delete_ledger" + ledger.mkdir(exist_ok=True) + tombstone = f"{episode}.{digest}.deleting" + path = ledger / f"{episode}.{digest}.authorized.json" + path.write_text( + json.dumps( + { + "schema_version": 1, + "state": "AUTHORIZED_DELETE", + "episode_id": episode, + "manifest_sha256": payload_digest or digest, + "tombstone": tombstone, + } + ), + encoding="utf-8", + ) + return path + + def test_compare_rename_delete_and_absent_retry_are_idempotent(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + ready, deleting, base = self._roots(Path(temporary)) + episode, digest = self._episode(ready) + name = episode.name + result = delete_ready_episode( + name, + digest, + ready_root=ready, + deleting_root=deleting, + ) + self.assertEqual(result["state"], "deleted") + self.assertFalse(episode.exists()) + self.assertEqual( + delete_ready_episode( + name, + digest, + ready_root=ready, + deleting_root=deleting, + )["state"], + "already_absent", + ) + self.assertEqual( + (base / "active" / "ACTIVE_SENTINEL").read_text(), "active" + ) + self.assertEqual( + (base / "failed" / "FAILED_SENTINEL").read_text(), "failed" + ) + + def test_resume_matching_tombstone_after_lost_ack(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + ready, deleting, _base = self._roots(Path(temporary)) + episode, digest = self._episode(ready) + deleting.mkdir() + tombstone = deleting / f"{episode.name}.{digest}.deleting" + episode.rename(tombstone) + result = delete_ready_episode( + "episode_20260810T120000000_deadbeef", + digest, + ready_root=ready, + deleting_root=deleting, + ) + self.assertEqual(result["state"], "resumed_delete") + self.assertFalse(tombstone.exists()) + + def test_partial_tree_without_manifest_resumes_from_durable_ledger(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + ready, deleting, base = self._roots(Path(temporary)) + episode, digest = self._episode(ready) + tombstone = deleting / f"{episode.name}.{digest}.deleting" + ledger = ( + base + / "delete_ledger" + / f"{episode.name}.{digest}.authorized.json" + ) + + def interrupt_after_control_files_removed( + _parent_fd: int, _name: str + ) -> None: + (tombstone / "READY").unlink() + (tombstone / "manifest.json").unlink() + raise RuntimeError("injected power loss during recursive delete") + + with mock.patch( + "delete_ready_episode._remove_tree_at", + side_effect=interrupt_after_control_files_removed, + ): + with self.assertRaises(RuntimeError): + delete_ready_episode( + episode.name, + digest, + ready_root=ready, + deleting_root=deleting, + ) + self.assertTrue(tombstone.is_dir()) + self.assertFalse((tombstone / "manifest.json").exists()) + self.assertTrue(ledger.is_file()) + + result = delete_ready_episode( + episode.name, + digest, + ready_root=ready, + deleting_root=deleting, + ) + self.assertEqual(result["state"], "resumed_delete") + self.assertFalse(tombstone.exists()) + self.assertFalse(ledger.exists()) + + def test_partial_tombstone_without_ledger_is_refused(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + ready, deleting, _base = self._roots(Path(temporary)) + episode, digest = self._episode(ready) + deleting.mkdir() + tombstone = deleting / f"{episode.name}.{digest}.deleting" + episode.rename(tombstone) + (tombstone / "manifest.json").unlink() + with self.assertRaises(DeleteSafetyError): + delete_ready_episode( + episode.name, + digest, + ready_root=ready, + deleting_root=deleting, + ) + self.assertTrue(tombstone.exists()) + + def test_wrong_digest_or_symlink_ledger_is_refused(self) -> None: + for mode in ("wrong_digest", "symlink"): + with self.subTest(mode=mode), tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + ready, deleting, base = self._roots(root) + episode, digest = self._episode(ready) + deleting.mkdir() + tombstone = deleting / f"{episode.name}.{digest}.deleting" + episode.rename(tombstone) + (tombstone / "manifest.json").unlink() + if mode == "wrong_digest": + ledger = self._authorization( + base, episode.name, digest, payload_digest="f" * 64 + ) + else: + ledger_root = base / "delete_ledger" + ledger_root.mkdir() + ledger = ( + ledger_root + / f"{episode.name}.{digest}.authorized.json" + ) + outside = root / "outside" + outside.write_text("keep", encoding="utf-8") + ledger.symlink_to(outside) + with self.assertRaises(DeleteSafetyError): + delete_ready_episode( + episode.name, + digest, + ready_root=ready, + deleting_root=deleting, + ) + self.assertTrue(tombstone.exists()) + self.assertTrue(ledger.exists()) + + def test_ledger_only_after_tree_removal_is_finalized_idempotently(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + ready, deleting, base = self._roots(Path(temporary)) + deleting.mkdir() + episode = "episode_20260810T120000000_deadbeef" + digest = "a" * 64 + ledger = self._authorization(base, episode, digest) + result = delete_ready_episode( + episode, + digest, + ready_root=ready, + deleting_root=deleting, + ) + self.assertEqual(result["state"], "already_absent") + self.assertFalse(ledger.exists()) + + def test_conflicting_ledger_digest_refuses_ready_target(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + ready, deleting, base = self._roots(Path(temporary)) + episode, digest = self._episode(ready) + collision = self._authorization(base, episode.name, "b" * 64) + with self.assertRaises(DeleteCollisionError): + delete_ready_episode( + episode.name, + digest, + ready_root=ready, + deleting_root=deleting, + ) + self.assertTrue(episode.exists()) + self.assertTrue(collision.exists()) + + def test_wrong_manifest_digest_refuses_and_preserves_target(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + ready, deleting, _base = self._roots(Path(temporary)) + episode, _digest = self._episode(ready) + with self.assertRaises(DeleteSafetyError): + delete_ready_episode( + episode.name, + "0" * 64, + ready_root=ready, + deleting_root=deleting, + ) + self.assertTrue(episode.exists()) + + def test_noncomplete_manifest_refuses_and_preserves_target(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + ready, deleting, _base = self._roots(Path(temporary)) + episode, digest = self._episode(ready, state="failed") + with self.assertRaises(DeleteSafetyError): + delete_ready_episode( + episode.name, + digest, + ready_root=ready, + deleting_root=deleting, + ) + self.assertTrue(episode.exists()) + + def test_unsafe_episode_and_digest_are_rejected(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + ready, deleting, _base = self._roots(Path(temporary)) + for episode, digest in (("../active", "0" * 64), ("safe", "BAD")): + with self.subTest(episode=episode, digest=digest): + with self.assertRaises(DeleteSafetyError): + delete_ready_episode( + episode, + digest, + ready_root=ready, + deleting_root=deleting, + ) + + def test_symlink_episode_is_refused_without_touching_outside(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + ready, deleting, _base = self._roots(root) + outside = root / "outside" + outside.mkdir() + sentinel = outside / "sentinel" + sentinel.write_text("keep", encoding="utf-8") + name = "episode_20260810T120000000_deadbeef" + (ready / name).symlink_to(outside, target_is_directory=True) + with self.assertRaises(DeleteSafetyError): + delete_ready_episode( + name, + "0" * 64, + ready_root=ready, + deleting_root=deleting, + ) + self.assertEqual(sentinel.read_text(), "keep") + + def test_symlink_manifest_ready_or_nested_entry_is_refused(self) -> None: + for target in ("manifest", "ready", "nested"): + with self.subTest(target=target), tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + ready, deleting, _base = self._roots(root) + episode, digest = self._episode(ready) + outside = root / "outside" + outside.write_text("keep", encoding="utf-8") + if target == "manifest": + (episode / "manifest.json").unlink() + (episode / "manifest.json").symlink_to(outside) + elif target == "ready": + (episode / "READY").unlink() + (episode / "READY").symlink_to(outside) + else: + (episode / "nested_link").symlink_to(outside) + with self.assertRaises(DeleteSafetyError): + delete_ready_episode( + episode.name, + digest, + ready_root=ready, + deleting_root=deleting, + ) + self.assertEqual(outside.read_text(), "keep") + + def test_conflicting_tombstone_refuses_without_moving_ready(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + ready, deleting, _base = self._roots(Path(temporary)) + episode, digest = self._episode(ready) + deleting.mkdir() + collision = deleting / f"{episode.name}.{'1' * 64}.deleting" + collision.mkdir() + with self.assertRaises(DeleteCollisionError): + delete_ready_episode( + episode.name, + digest, + ready_root=ready, + deleting_root=deleting, + ) + self.assertTrue(episode.exists()) + self.assertTrue(collision.exists()) + + def test_ready_and_same_tombstone_collision_refuses_both(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + ready, deleting, _base = self._roots(Path(temporary)) + episode, digest = self._episode(ready) + deleting.mkdir() + tombstone = deleting / f"{episode.name}.{digest}.deleting" + tombstone.mkdir() + with self.assertRaises(DeleteCollisionError): + delete_ready_episode( + episode.name, + digest, + ready_root=ready, + deleting_root=deleting, + ) + self.assertTrue(episode.exists()) + self.assertTrue(tombstone.exists()) + + def test_wrong_parent_relationship_is_refused(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + ready, _deleting, _base = self._roots(root) + elsewhere = root / "elsewhere" / "deleting" + elsewhere.mkdir(parents=True) + with self.assertRaises(DeleteSafetyError): + delete_ready_episode( + "episode_safe", + "0" * 64, + ready_root=ready, + deleting_root=elsewhere, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/verify.sh b/verify.sh index b01e05f..cf96057 100755 --- a/verify.sh +++ b/verify.sh @@ -8,6 +8,7 @@ for executable in \ "$repo_dir/tg3_omnisocket_transport/omnisocket_xtele_sender.py" \ "$repo_dir/tg3_local_teleop/tg3_local_teleop.py" \ "$repo_dir/tg3_local_teleop/data_recorder_node.py" \ + "$repo_dir/tg3_local_teleop/delete_ready_episode.py" \ "$repo_dir/tg3_local_teleop/run.sh" \ "$repo_dir/tg3_local_teleop/run_data_recorder.sh" \ "$repo_dir/tg3_local_teleop/wait_ros_ready.sh" \ @@ -27,6 +28,7 @@ python3 -m py_compile \ "$repo_dir/tg3_local_teleop/data_collection.py" \ "$repo_dir/tg3_local_teleop/data_recorder_protocol.py" \ "$repo_dir/tg3_local_teleop/data_recorder_node.py" \ + "$repo_dir/tg3_local_teleop/delete_ready_episode.py" \ "$repo_dir/tg3_data_collection/data_get_sync.py" python3 "$repo_dir/tg3_omnisocket_transport/test_session_gate.py" python3 "$repo_dir/tg3_local_teleop/test_session_gate.py" @@ -34,6 +36,7 @@ python3 "$repo_dir/tg3_local_teleop/test_gesture_toggle.py" python3 "$repo_dir/tg3_local_teleop/test_idle_session_refresh.py" python3 "$repo_dir/tg3_local_teleop/test_data_collection.py" python3 "$repo_dir/tg3_local_teleop/test_data_recorder_protocol.py" +python3 "$repo_dir/tg3_local_teleop/test_delete_ready_episode.py" python3 "$repo_dir/tg3_data_collection/test_data_get_sync.py" python3 - "$repo_dir/tg3_local_teleop/config.toml" <<'PY' import sys @@ -47,29 +50,39 @@ required = set(section["required_topics"]) head_rgbd = { "/ob_camera_head/color/image_raw/compressed", "/ob_camera_head/color/camera_info", + "/ob_camera_head/color/metadata", "/ob_camera_head/depth/image_raw/compressedDepth", "/ob_camera_head/depth/camera_info", + "/ob_camera_head/depth/metadata", } waist_rgbd = { "/ob_camera_waist/color/image_raw/compressed", "/ob_camera_waist/color/camera_info", + "/ob_camera_waist/color/metadata", "/ob_camera_waist/depth/image_raw/compressedDepth", "/ob_camera_waist/depth/camera_info", + "/ob_camera_waist/depth/metadata", } assert head_rgbd <= topics, "head RGB-D topics must stay in the recording whitelist" assert waist_rgbd <= topics, "waist RGB-D topics must stay in the recording whitelist" -assert waist_rgbd <= required, "live waist RGB-D topics must stay required" -assert head_rgbd.isdisjoint(required), "inactive head RGB-D must not block recording" -assert { - "/ob_camera_waist/color/metadata", - "/ob_camera_waist/depth/metadata", - "/tf_static", -} <= required -assert section["minimum_topic_rates_hz"] == { +assert head_rgbd.isdisjoint(required), "head RGB-D must not block recording" +assert waist_rgbd.isdisjoint(required), "waist RGB-D must not block recording" +assert "/tf_static" in topics and "/tf_static" not in required +assert section.get("minimum_topic_rates_hz", {}) == {} +groups = section["optional_topic_groups"] +assert set(groups) == {"head_rgbd", "waist_rgbd"} +assert set(groups["head_rgbd"]["topics"]) == head_rgbd +assert set(groups["waist_rgbd"]["topics"]) == waist_rgbd +assert groups["head_rgbd"]["minimum_topic_rates_hz"] == { + "/ob_camera_head/color/image_raw/compressed": 20.0, + "/ob_camera_head/depth/image_raw/compressedDepth": 20.0, +} +assert groups["waist_rgbd"]["minimum_topic_rates_hz"] == { "/ob_camera_waist/color/image_raw/compressed": 20.0, "/ob_camera_waist/depth/image_raw/compressedDepth": 20.0, } assert section["minimum_free_gib"] >= 100.0 +assert section["retain_failed_episodes"] is False print("config.toml parse passed") PY