feat: Add save_snapshot_delayed service with optional delay before snapshot

- Add rosbag_blackbox_interfaces package with SaveSnapshotDelayed.srv service definition (float64 delay request, bool success + string message response)
- Update rosbag_blackbox package.xml to depend on rosbag_blackbox_interfaces
- Add ~/save_snapshot_delayed service to RingBufferNode that returns immediately and runs the snapshot in a background ThreadPoolExecutor after the specified delay
- Add 8 pure-Python tests for the delayed snapshot logic

Co-authored-by: marc-hanheide <1153084+marc-hanheide@users.noreply.github.com>

Author: Copilot

src/rosbag_blackbox/package.xml

@@ -17,6 +17,7 @@
   <exec_depend>rosbag2_storage_sqlite3</exec_depend>
   <exec_depend>rosidl_runtime_py</exec_depend>
   <exec_depend>std_srvs</exec_depend>
+  <exec_depend>rosbag_blackbox_interfaces</exec_depend>
   <exec_depend>sensor_msgs</exec_depend>
   <exec_depend>python3-yaml</exec_depend>
 

src/rosbag_blackbox/rosbag_blackbox/ringbuffer_node.py

@@ -19,6 +19,10 @@
 A ``std_srvs/srv/Trigger`` service triggers a snapshot: the buffered
 messages are written to a rosbag file.
 
+A second service (``~/save_snapshot_delayed``) accepts an optional
+``delay`` (seconds).  It returns immediately and performs the snapshot
+in a background thread once the delay has elapsed.
+
 Topics are **dynamically discovered** at runtime, mirroring the
 behaviour of the ``ros2 bag record`` CLI.  Pattern matching follows
 the zenoh-selector convention (``*`` = one segment, ``**`` = any depth).
@@ -44,7 +48,9 @@
 is written into each snapshot.
 """
 
+import concurrent.futures
 import os
+import time
 from datetime import datetime
 from typing import Any, Dict, List, Optional, Set
 
@@ -59,6 +65,7 @@
     ReliabilityPolicy,
 )
 from rclpy.serialization import serialize_message
+from rosbag_blackbox_interfaces.srv import SaveSnapshotDelayed
 from rosidl_runtime_py.utilities import get_message
 from std_srvs.srv import Trigger
 
@@ -224,6 +231,18 @@ def __init__(self) -> None:
             self._on_save_snapshot,
         )
 
+        # Background executor for delayed snapshots ---------------------------
+        self._snapshot_executor = concurrent.futures.ThreadPoolExecutor(
+            max_workers=4, thread_name_prefix='snapshot_bg'
+        )
+
+        # Service to trigger a delayed snapshot (returns immediately) ----------
+        self._delayed_snapshot_srv = self.create_service(
+            SaveSnapshotDelayed,
+            '~/save_snapshot_delayed',
+            self._on_save_snapshot_delayed,
+        )
+
         # Periodic topic discovery --------------------------------------------
         self._discovery_timer = self.create_timer(2.0, self._discover_topics)
 
@@ -564,6 +583,51 @@ def _on_auto_snapshot(self) -> None:
         except Exception as exc:  # noqa: BLE001
             self.get_logger().warning(f'Auto-snapshot skipped: {exc}')
 
+    def _on_save_snapshot_delayed(
+        self,
+        request: SaveSnapshotDelayed.Request,
+        response: SaveSnapshotDelayed.Response,
+    ) -> SaveSnapshotDelayed.Response:
+        """Schedule a snapshot after an optional delay, returning immediately.
+
+        The caller receives an acknowledgement right away.  The actual
+        snapshot is written by the background executor once *delay*
+        seconds have elapsed.
+
+        :param request: Service request containing ``delay`` (float64, seconds).
+        :param response: Service response with ``success`` and ``message``.
+        :returns: Populated response indicating the snapshot was scheduled.
+        """
+        delay = request.delay
+        if delay < 0.0:
+            response.success = False
+            response.message = f'Invalid delay {delay}s: must be >= 0.'
+            self.get_logger().error(response.message)
+            return response
+
+        self._snapshot_executor.submit(self._write_snapshot_background, delay=delay)
+
+        if delay > 0.0:
+            response.message = f'Snapshot scheduled in {delay:.3g}s'
+        else:
+            response.message = 'Snapshot scheduled immediately'
+        response.success = True
+        self.get_logger().info(response.message)
+        return response
+
+    def _write_snapshot_background(self, delay: float = 0.0) -> None:
+        """Write a snapshot in a background thread, optionally after a delay.
+
+        :param delay: Seconds to wait before taking the snapshot.
+        """
+        if delay > 0.0:
+            time.sleep(delay)
+        try:
+            path = self._write_snapshot(trigger='manual')
+            self.get_logger().info(f'Delayed snapshot saved: {path}')
+        except Exception as exc:  # noqa: BLE001
+            self.get_logger().error(f'Delayed snapshot failed: {exc}')
+
     def _write_snapshot(self, trigger: str = 'manual') -> str:
         """Serialise the ring buffer to a new rosbag directory.
 

src/rosbag_blackbox/test/test_delayed_snapshot.py

@@ -0,0 +1,251 @@
+# Copyright 2024 Maintainer
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Tests for the delayed-snapshot helpers – no ROS2 runtime needed.
+
+The methods under test are extracted from RingBufferNode and re-bound onto a
+lightweight fake node so we can exercise the logic without a live ROS2
+installation or compiled interface packages.
+"""
+
+import concurrent.futures
+import threading
+import time
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _make_request(delay: float) -> SimpleNamespace:
+    return SimpleNamespace(delay=delay)
+
+
+def _make_response() -> SimpleNamespace:
+    return SimpleNamespace(success=None, message='')
+
+
+# ---------------------------------------------------------------------------
+# The two methods under test, kept in sync with RingBufferNode so the
+# tests remain independent of the ROS2 interface package.
+# ---------------------------------------------------------------------------
+
+def _on_save_snapshot_delayed(self, request, response):
+    """Schedule a snapshot after an optional delay, returning immediately."""
+    delay = request.delay
+    if delay < 0.0:
+        response.success = False
+        response.message = f'Invalid delay {delay}s: must be >= 0.'
+        self.get_logger().error(response.message)
+        return response
+
+    self._snapshot_executor.submit(self._write_snapshot_background, delay=delay)
+
+    if delay > 0.0:
+        response.message = f'Snapshot scheduled in {delay:.3g}s'
+    else:
+        response.message = 'Snapshot scheduled immediately'
+    response.success = True
+    self.get_logger().info(response.message)
+    return response
+
+
+def _write_snapshot_background(self, delay: float = 0.0) -> None:
+    """Write a snapshot in a background thread, optionally after a delay."""
+    if delay > 0.0:
+        time.sleep(delay)
+    try:
+        path = self._write_snapshot(trigger='manual')
+        self.get_logger().info(f'Delayed snapshot saved: {path}')
+    except Exception as exc:  # noqa: BLE001
+        self.get_logger().error(f'Delayed snapshot failed: {exc}')
+
+
+# ---------------------------------------------------------------------------
+# Fake node that hosts the methods above
+# ---------------------------------------------------------------------------
+
+class _FakeNode:
+    """Minimal stand-in for RingBufferNode."""
+
+    def __init__(self, write_result='/tmp/snap', write_raises=None):
+        self._write_result = write_result
+        self._write_raises = write_raises
+        self._log_messages = []
+        self._snapshot_executor = concurrent.futures.ThreadPoolExecutor(
+            max_workers=4, thread_name_prefix='snapshot_bg'
+        )
+
+    def get_logger(self):
+        node = self
+
+        class _Logger:
+            def info(self, msg):
+                node._log_messages.append(('info', msg))
+
+            def error(self, msg):
+                node._log_messages.append(('error', msg))
+
+            def warning(self, msg):
+                node._log_messages.append(('warning', msg))
+
+        return _Logger()
+
+    def _write_snapshot(self, trigger: str = 'manual') -> str:
+        if self._write_raises:
+            raise self._write_raises
+        return self._write_result
+
+    _on_save_snapshot_delayed = _on_save_snapshot_delayed
+    _write_snapshot_background = _write_snapshot_background
+
+
+# ---------------------------------------------------------------------------
+# Tests for _on_save_snapshot_delayed
+# ---------------------------------------------------------------------------
+
+class TestOnSaveSnapshotDelayed:
+
+    def test_zero_delay_returns_immediately(self):
+        node = _FakeNode()
+        req = _make_request(delay=0.0)
+        resp = _make_response()
+
+        with patch.object(node._snapshot_executor, 'submit') as mock_submit:
+            result = node._on_save_snapshot_delayed(req, resp)
+
+        assert result.success is True
+        assert 'immediately' in result.message.lower()
+        mock_submit.assert_called_once()
+
+    def test_positive_delay_schedules_background_task(self):
+        node = _FakeNode()
+        req = _make_request(delay=5.0)
+        resp = _make_response()
+
+        with patch.object(node._snapshot_executor, 'submit') as mock_submit:
+            result = node._on_save_snapshot_delayed(req, resp)
+
+        assert result.success is True
+        assert '5' in result.message
+        mock_submit.assert_called_once_with(
+            node._write_snapshot_background, delay=5.0
+        )
+
+    def test_negative_delay_returns_error(self):
+        node = _FakeNode()
+        req = _make_request(delay=-1.0)
+        resp = _make_response()
+
+        result = node._on_save_snapshot_delayed(req, resp)
+
+        assert result.success is False
+        assert 'invalid' in result.message.lower() or '-1' in result.message
+
+    def test_returns_immediately_before_snapshot_completes(self):
+        """The response must arrive before the snapshot is written."""
+        write_event = threading.Event()
+
+        class _SlowNode(_FakeNode):
+            def _write_snapshot(self, trigger='manual'):
+                time.sleep(0.05)
+                write_event.set()
+                return '/tmp/snap'
+
+        node = _SlowNode()
+        req = _make_request(delay=0.0)
+        resp = _make_response()
+
+        start = time.monotonic()
+        result = node._on_save_snapshot_delayed(req, resp)
+        elapsed = time.monotonic() - start
+
+        # Wait for the background task to finish
+        write_event.wait(timeout=1.0)
+
+        # The response should arrive well before the 50 ms sleep in _write_snapshot
+        assert elapsed < 0.04, f'Response took {elapsed:.3f}s – not immediate'
+        assert result.success is True
+
+
+# ---------------------------------------------------------------------------
+# Tests for _write_snapshot_background
+# ---------------------------------------------------------------------------
+
+class TestWriteSnapshotBackground:
+
+    def test_calls_write_snapshot_immediately_when_no_delay(self):
+        node = _FakeNode(write_result='/tmp/snap')
+        calls = []
+        original = node._write_snapshot
+
+        def _capture(**kw):
+            calls.append(kw)
+            return original(**kw)
+
+        node._write_snapshot = _capture
+        node._write_snapshot_background(delay=0.0)
+
+        assert len(calls) == 1
+
+    def test_calls_write_snapshot_after_delay(self):
+        node = _FakeNode(write_result='/tmp/snap')
+        calls = []
+        original = node._write_snapshot
+
+        def _capture(**kw):
+            calls.append(time.monotonic())
+            return original(**kw)
+
+        node._write_snapshot = _capture
+        start = time.monotonic()
+        node._write_snapshot_background(delay=0.05)
+        elapsed = time.monotonic() - start
+
+        assert len(calls) == 1
+        assert elapsed >= 0.05, f'Did not wait: elapsed={elapsed:.4f}s'
+
+    def test_logs_error_when_write_snapshot_raises(self):
+        node = _FakeNode(write_raises=ValueError('buffer empty'))
+        node._write_snapshot_background(delay=0.0)
+
+        error_logs = [msg for level, msg in node._log_messages if level == 'error']
+        assert any('failed' in m.lower() for m in error_logs)
+
+    def test_executor_manages_background_tasks(self):
+        """Verify that tasks are submitted to the executor, not raw threads."""
+        node = _FakeNode()
+        req = _make_request(delay=0.0)
+        resp = _make_response()
+
+        submitted = []
+
+        original_submit = node._snapshot_executor.submit
+
+        def _capture_submit(fn, **kwargs):
+            submitted.append((fn, kwargs))
+            return original_submit(fn, **kwargs)
+
+        node._snapshot_executor.submit = _capture_submit
+        node._on_save_snapshot_delayed(req, resp)
+
+        # Wait for background work to finish
+        node._snapshot_executor.shutdown(wait=True)
+
+        assert len(submitted) == 1
+        assert submitted[0][1] == {'delay': 0.0}
+
+

src/rosbag_blackbox_interfaces/CMakeLists.txt

@@ -0,0 +1,17 @@
+cmake_minimum_required(VERSION 3.8)
+project(rosbag_blackbox_interfaces)
+
+if(CMAKE_COMPILER_IS_GNUCXX OR CMAKE_CXX_COMPILER_ID MATCHES "Clang")
+  add_compile_options(-Wall -Wextra -Wpedantic)
+endif()
+
+find_package(ament_cmake REQUIRED)
+find_package(rosidl_default_generators REQUIRED)
+
+rosidl_generate_interfaces(${PROJECT_NAME}
+  "srv/SaveSnapshotDelayed.srv"
+)
+
+ament_export_dependencies(rosidl_default_runtime)
+
+ament_package()

src/rosbag_blackbox_interfaces/package.xml

@@ -0,0 +1,20 @@
+<?xml version="1.0"?>
+<?xml-model href="http://download.ros.org/schema/package_format3.xsd" schematypens="http://www.w3.org/2001/XMLSchema"?>
+<package format="3">
+  <name>rosbag_blackbox_interfaces</name>
+  <version>0.1.0</version>
+  <description>ROS2 service/message interfaces for rosbag_blackbox.</description>
+  <maintainer email="maintainer@example.com">Maintainer</maintainer>
+  <license>Apache-2.0</license>
+
+  <buildtool_depend>ament_cmake</buildtool_depend>
+  <buildtool_depend>rosidl_default_generators</buildtool_depend>
+
+  <exec_depend>rosidl_default_runtime</exec_depend>
+
+  <member_of_group>rosidl_interface_packages</member_of_group>
+
+  <export>
+    <build_type>ament_cmake</build_type>
+  </export>
+</package>