record: warn while dropping, not just at close

[jeff-hykin]

Warn >=10s while when writer queue is dropping messages

[greptile-apps]

Greptile Summary

This change makes recorder queue loss visible while a run is active and retains the close-time total. Focused execution found two P2 observability issues: simultaneous transport callbacks can produce duplicate warnings within one ten-second interval, and the first warning is suppressed when the first drop occurs during the first ten seconds of monotonic uptime.

Confidence Score: 4/5

Not merge-safe without addressing the two warning-observability issues.

There are two independent verified P2 findings and no P0 or P1 findings; the required score for P2-only findings is 4.

Files Needing Attention: dimos/memory/tap.py needs synchronization for the warning throttle and initialization that permits the first drop warning.

T-Rex Logs

What T-Rex did

• T-Rex produced a complete proof for the posted P2 finding and linked it to the review comment detailing the finding.
• T-Rex reproduced a focused transport recorder concurrency scenario and captured both the single full-queue callback baseline output and the overlapping full-queue callbacks reproduction output.
• T-Rex ran the focused TransportRecorder first-drop warning exercise to exercise drop behavior.
• T-Rex captured the queue-full drop events, including the drop before the 10-second interval and the drop at the 10-second interval.

_{Ran code and verified through T-Rex}

_{Reviews (1): Last reviewed commit: "record: warn while dropping, not just at..." | Re-trigger Greptile}

[codecov]

❌ 2 Tests Failed:

Tests completed	Failed	Passed	Skipped
4815	2	4813	87

View the top 1 failed test(s) by shortest run time

dimos.e2e_tests.test_manipulation_planning_groups::test_dual_arm_plans_and_dispatches_both_arms_through_control_coordinator

Stack Traces | 120s run time

lcm_spy = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7359d489c530>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x7359d2ffb420>

    def test_dual_arm_plans_and_dispatches_both_arms_through_control_coordinator(
        lcm_spy: LcmSpy,
        start_blueprint: Callable[..., DimosCliCall],
    ) -> None:
        """Plan one generated plan over both arms and dispatch through one trajectory task."""
>       _start_openarm_mock_planner(start_blueprint, lcm_spy)

lcm_spy    = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7359d489c530>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x7359d2ffb420>

dimos/e2e_tests/test_manipulation_planning_groups.py:190: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
dimos/e2e_tests/test_manipulation_planning_groups.py:151: in _start_openarm_mock_planner
    lcm_spy.wait_for_saved_topic(JOINT_STATE_TOPIC, timeout=120.0)
        lcm_spy    = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7359d489c530>
        start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x7359d2ffb420>
dimos/e2e_tests/lcm_spy.py:97: in wait_for_saved_topic
    wait_until(
        condition  = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x7359d2ffbd80>
        self       = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7359d489c530>
        timeout    = 120.0
        topic      = '/coordinator_joint_state#sensor_msgs.JointState'
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

predicate = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x7359d2ffbd80>

    def wait_until(
        predicate: Callable[[], bool],
        *,
        timeout: float,
        interval: float = 0.1,
        message: str | None = None,
    ) -> None:
        """Poll ``predicate`` until it returns truthy or ``timeout`` elapses."""
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            if predicate():
                return
            time.sleep(interval)
>       raise TimeoutError(message or f"Timed out after {timeout}s waiting for condition")
E       TimeoutError: Timeout waiting for topic /coordinator_joint_state#sensor_msgs.JointState

deadline   = 4314826.379825446
interval   = 0.1
message    = 'Timeout waiting for topic /coordinator_joint_state#sensor_msgs.JointState'
predicate  = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x7359d2ffbd80>
timeout    = 120.0

.../utils/testing/waiting.py:35: TimeoutError

View the full list of 1 ❄️ flaky test(s)

dimos.e2e_tests.test_manipulation_planning_groups::test_single_arm_plans_and_executes_through_control_coordinator

Flake rate in main: 7.69% (Passed 12 times, Failed 1 times)

Stack Traces | 120s run time

lcm_spy = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7359d29cffb0>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x7359d2ffae80>

    def test_single_arm_plans_and_executes_through_control_coordinator(
        lcm_spy: LcmSpy,
        start_blueprint: Callable[..., DimosCliCall],
    ) -> None:
        """Plan with one arm and execute through its trajectory task."""
>       _start_openarm_mock_planner(start_blueprint, lcm_spy)

lcm_spy    = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7359d29cffb0>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x7359d2ffae80>

dimos/e2e_tests/test_manipulation_planning_groups.py:159: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
dimos/e2e_tests/test_manipulation_planning_groups.py:151: in _start_openarm_mock_planner
    lcm_spy.wait_for_saved_topic(JOINT_STATE_TOPIC, timeout=120.0)
        lcm_spy    = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7359d29cffb0>
        start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x7359d2ffae80>
dimos/e2e_tests/lcm_spy.py:97: in wait_for_saved_topic
    wait_until(
        condition  = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x7359d2ff9580>
        self       = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7359d29cffb0>
        timeout    = 120.0
        topic      = '/coordinator_joint_state#sensor_msgs.JointState'
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

predicate = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x7359d2ff9580>

    def wait_until(
        predicate: Callable[[], bool],
        *,
        timeout: float,
        interval: float = 0.1,
        message: str | None = None,
    ) -> None:
        """Poll ``predicate`` until it returns truthy or ``timeout`` elapses."""
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            if predicate():
                return
            time.sleep(interval)
>       raise TimeoutError(message or f"Timed out after {timeout}s waiting for condition")
E       TimeoutError: Timeout waiting for topic /coordinator_joint_state#sensor_msgs.JointState

deadline   = 4314706.183199974
interval   = 0.1
message    = 'Timeout waiting for topic /coordinator_joint_state#sensor_msgs.JointState'
predicate  = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x7359d2ff9580>
timeout    = 120.0

.../utils/testing/waiting.py:35: TimeoutError

To view more test analytics, go to the Test Analytics Dashboard
_{📋 Got 3 mins? Take this short survey to help us improve Test Analytics.}

[spomichter]

no global vars that are not in global config has been repo convention