fix: publish collection history atomically

src/fenris/collector.py

@@ -5,7 +5,7 @@ This module implements the thinnest complete write path:
 - Acquire controller identity from sysfs
 - Normalize identity exactly once at write time
 - Validate every row against store invariants
-- Commit one well-formed sample
+- Publish the sample and derived evidence in one collection-owned transaction
 
 No code path outside the collector interrogates the device.
 """
@@ -200,7 +200,7 @@ def write_sample(
     """Write one sample to the observation store.
     
     Identity normalization happens exactly once here.
-    Returns segment info for the caller.
+    Returns segment info for the caller. Caller owns the transaction.
     """
     from .segment import find_current_segment, should_open_new_segment, open_segment
     
@@ -262,8 +262,6 @@ def write_sample(
         ),
     )
     
-    conn.commit()
-    
     return {
         "segment_opened": segment_opened,
         "segment_reason": reason,
@@ -284,6 +282,7 @@ def run_collection(
     This is the main entry point for the collector.
     Returns the run outcome.
     """
+    conn = None
     try:
         # Acquire counters and thermal evidence
         counters = acquire_from_smartctl(smartctl_data)
@@ -309,72 +308,55 @@ def run_collection(
         if history_path.exists():
             import_legacy_history(conn, history_path, clock=clock)
 
-        
-        try:
-            # Ensure monitoring period is open (issue #73 AC2)
-            ensure_period_open(conn, clock.utcnow())
-            
-            # Validate invariants
-            validate_sample_invariants(sample, conn)
-            
-            # Write sample and get segment info
-            seg_info = write_sample(sample, identity, conn, clock)
-            
-            # Derive hour observations from interval with previous sample
-            try:
-                # Find the sample we just wrote
-                cursor = conn.execute("SELECT id FROM samples ORDER BY id DESC LIMIT 1")
-                current_id = cursor.fetchone()[0]
+        # Collection owns one transaction for the sample and its evidence.
+        ensure_period_open(conn, clock.utcnow())
 
-                prev = find_previous_sample(conn, seg_info.get("segment_id"), current_id)
-                if prev is not None:
-                    # Build current sample dict for derivation
-                    current = {
-                        "id": current_id,
-                        "ts": sample["ts"],
-                        "bytes_written": sample["bytes_written"],
-                        "bytes_read": sample["bytes_read"],
-                        "power_on_hours": sample["power_on_hours"],
-                        "temperature_c": sample["temperature_c"],
-                        "data_units_written": sample["data_units_written"],
-                        "data_units_read": sample["data_units_read"],
-                    }
-                    derive_hours_from_interval(conn, prev, current)
+        validate_sample_invariants(sample, conn)
+        seg_info = write_sample(sample, identity, conn, clock)
 
-                # Rebuild day aggregates from hour observations
-                from .day_aggregate import derive_all_days, persist_day_aggregate
-                for agg in derive_all_days(conn):
-                    persist_day_aggregate(conn, agg)
-
-                # Derive local-day summary using system timezone (issue #90)
-                try:
-                    from .tz_util import detect_system_tz
-                    from .local_day import derive_local_day_summary, persist_local_day
-                    tz_name = detect_system_tz()
-                    clock_now = clock.utcnow()
-                    local_summary = derive_local_day_summary(conn, tz_name, clock_now)
-                    if local_summary is not None:
-                        persist_local_day(conn, local_summary)
-                except Exception:
-                    # Local-day derivation failure must not prevent publication
-                    pass
-
-                conn.commit()
-            except Exception:
-                # Derivation failure must not prevent sample persistence (issue #73 AC6)
-                pass
-            
-            return {
-                "ok": True,
-                "sample_count": 1,
-                "store_path": str(store_path),
+        cursor = conn.execute("SELECT id FROM samples ORDER BY id DESC LIMIT 1")
+        current_id = cursor.fetchone()[0]
+        prev = find_previous_sample(conn, seg_info.get("segment_id"), current_id)
+        if prev is not None:
+            current = {
+                "id": current_id,
+                "ts": sample["ts"],
+                "bytes_written": sample["bytes_written"],
+                "bytes_read": sample["bytes_read"],
+                "power_on_hours": sample["power_on_hours"],
+                "temperature_c": sample["temperature_c"],
+                "data_units_written": sample["data_units_written"],
+                "data_units_read": sample["data_units_read"],
             }
-        finally:
-            conn.close()
-    
-    except (AcquisitionError, InvariantViolationError) as e:
+            derive_hours_from_interval(conn, prev, current)
+
+        from .day_aggregate import derive_all_days, persist_day_aggregate
+        for agg in derive_all_days(conn):
+            persist_day_aggregate(conn, agg)
+
+        from .tz_util import detect_system_tz
+        from .local_day import derive_local_day_summary, persist_local_day
+        tz_name = detect_system_tz()
+        local_summary = derive_local_day_summary(conn, tz_name, clock.utcnow())
+        if local_summary is not None:
+            persist_local_day(conn, local_summary)
+
+        conn.commit()
+
+        return {
+            "ok": True,
+            "sample_count": 1,
+            "store_path": str(store_path),
+        }
+
+    except Exception as e:
+        if conn is not None:
+            conn.rollback()
         return {
             "ok": False,
             "error": str(e),
             "error_type": type(e).__name__,
         }
+    finally:
+        if conn is not None:
+            conn.close()

src/fenris/derive.py

@@ -96,6 +96,7 @@ def derive_hours_from_interval(
     """Derive hour observations from a sample pair interval.
 
     Returns list of hour observation dicts that were written/updated.
+    Caller owns the transaction.
     """
     prev_ts = _parse_ts(prev_sample["ts"])
     next_ts = _parse_ts(next_sample["ts"])
@@ -207,9 +208,6 @@ def _upsert_hour_observation(
             "WHERE id = ?",
             (new_bw, new_br, new_samples, existing[0]),
         )
-    conn.commit()
-
-
 def _add_unattributed_bytes(
     conn: sqlite3.Connection,
     prev_ts: datetime,
@@ -241,4 +239,3 @@ def _add_unattributed_bytes(
             "unattributed_bytes_read = unattributed_bytes_read + ? WHERE day = ?",
             (bw_delta, br_delta, day),
         )
-    conn.commit()

src/fenris/monitor.py

@@ -69,6 +69,7 @@ def cmd_enable(args: argparse.Namespace) -> None:
         open_period = get_open_period(conn)
         if open_period is None:
             ensure_period_open(conn, now)
+            conn.commit()
             print("Monitoring period opened at", now.isoformat())
         else:
             print("Monitoring period already open (id=%d)" % open_period["id"])

src/fenris/monitoring_periods.py

@@ -16,7 +16,7 @@ def ensure_period_open(conn: sqlite3.Connection, run_time: datetime) -> None:
     """Ensure a monitoring period is open.  If none exists, open one at run_time.
 
     Spec §9.8: A collection run finding no open monitoring period opens one
-    at the run moment, never backdated.
+    at the run moment, never backdated. Caller owns the transaction.
     """
     if get_open_period(conn) is not None:
         return  # Already open — no-op
@@ -26,7 +26,6 @@ def ensure_period_open(conn: sqlite3.Connection, run_time: datetime) -> None:
         "INSERT INTO monitoring_periods (started_at) VALUES (?)",
         (ts,),
     )
-    conn.commit()
 
 
 def close_period(

src/fenris/tz_util.py

@@ -25,11 +25,14 @@ def detect_system_tz() -> str:
 
     localtime = Path("/etc/localtime")
     if localtime.is_symlink():
-        target = os.readlink(str(localtime))
-        # Strip common prefixes: /usr/share/zoneinfo/, /usr/lib/zoneinfo/
-        for prefix in ("/usr/share/zoneinfo/", "/usr/lib/zoneinfo/"):
-            if target.startswith(prefix):
-                return target[len(prefix):]
+        target_path = Path(os.readlink(str(localtime)))
+        if not target_path.is_absolute():
+            target_path = localtime.parent / target_path
+        target = target_path.resolve().as_posix()
+        marker = "/zoneinfo/"
+        marker_index = target.find(marker)
+        if marker_index >= 0:
+            return target[marker_index + len(marker):]
         return target
 
     return "UTC"

tests/test_collector_history_tracer.py

@@ -356,6 +356,165 @@ class TestCollectorDerivation:
         assert count_after == 2
 
 
+class TestCollectionAtomicity:
+    """Collection exposes sample and derived evidence as one publication."""
+
+    @pytest.mark.asyncio
+    async def test_collection_is_visible_through_cli_and_tui_readers(
+        self, config_fixture, smartctl_fixture, sysfs_fixture_tree, monkeypatch
+    ):
+        """Ordinary readers observe matching published sample and activity evidence."""
+        from fenris.status import get_status, read_status
+        from fenris.tui import FenrisTuiApp
+
+        monkeypatch.setenv("TZ", "UTC")
+        service = {
+            "boot_enabled": True,
+            "timer_active": True,
+            "last_collect_ok": True,
+        }
+        monkeypatch.setattr("fenris.status.query_service_state", lambda: service)
+
+        now = datetime.now(timezone.utc).replace(second=0, microsecond=0)
+        first = {
+            **smartctl_fixture,
+            "nvme_smart_health_information_log": {
+                **smartctl_fixture["nvme_smart_health_information_log"],
+                "data_units_written": 12345678,
+                "data_units_read": 9876543,
+            },
+        }
+        second = {
+            **smartctl_fixture,
+            "nvme_smart_health_information_log": {
+                **smartctl_fixture["nvme_smart_health_information_log"],
+                "data_units_written": 12345698,
+                "data_units_read": 9876553,
+            },
+        }
+        sysfs_path = sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0"
+
+        assert run_collection(
+            first, sysfs_path, config_fixture, FakeClock(now - timedelta(minutes=5))
+        )["ok"] is True
+        assert run_collection(
+            second, sysfs_path, config_fixture, FakeClock(now)
+        )["ok"] is True
+
+        store_path = Path(config_fixture["store_path"])
+        with read_status(store_path, now, query_services=False) as (reader, composition):
+            assert reader is not None
+            assert composition.sample_count == 2
+            assert composition.day_count == 1
+            utc_day = reader.execute(
+                "SELECT bytes_written_delta, bytes_read_delta FROM day_aggregates"
+            ).fetchone()
+            local_day = reader.execute(
+                "SELECT bytes_written, bytes_read FROM local_days"
+            ).fetchone()
+            assert tuple(utc_day) == (10_240_000, 5_120_000)
+            assert tuple(local_day) == (10_240_000, 5_120_000)
+
+        cli_output = get_status(
+            store_path, now, query_services=True, query_journal=False
+        )
+        assert "Monitoring" in cli_output
+
+        app = FenrisTuiApp(store_path=store_path)
+        async with app.run_test(size=(100, 30)) as pilot:
+            await pilot.pause()
+            live_readout = str(app.query_one("#live-readout").render())
+            assert "W 0.010 GB" in live_readout
+            assert "R 0.005 GB" in live_readout
+
+        repeated_result = run_collection(
+            second, sysfs_path, config_fixture, FakeClock(now + timedelta(minutes=5))
+        )
+        assert repeated_result["ok"] is True
+        with read_status(
+            store_path, now + timedelta(minutes=5), query_services=False
+        ) as (reader, composition):
+            assert reader is not None
+            assert composition.sample_count == 3
+            utc_day = reader.execute(
+                "SELECT bytes_written_delta, bytes_read_delta FROM day_aggregates"
+            ).fetchone()
+            local_day = reader.execute(
+                "SELECT bytes_written, bytes_read FROM local_days"
+            ).fetchone()
+            assert tuple(utc_day) == (10_240_000, 5_120_000)
+            assert tuple(local_day) == (10_240_000, 5_120_000)
+
+    def test_failed_local_day_publication_keeps_previous_publication(
+        self, config_fixture, smartctl_fixture, sysfs_fixture_tree, monkeypatch
+    ):
+        """A failed final derivation step leaves all prior reader state intact."""
+        from fenris.status import read_status
+
+        monkeypatch.setenv("TZ", "UTC")
+        now = datetime.now(timezone.utc).replace(second=0, microsecond=0)
+        first_clock = FakeClock(now)
+        first = {
+            **smartctl_fixture,
+            "nvme_smart_health_information_log": {
+                **smartctl_fixture["nvme_smart_health_information_log"],
+                "data_units_written": 12345678,
+                "data_units_read": 9876543,
+            },
+        }
+        first_result = run_collection(
+            first,
+            sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0",
+            config_fixture,
+            first_clock,
+        )
+        assert first_result["ok"] is True, first_result
+
+        writer = sqlite3.connect(config_fixture["store_path"])
+        writer.execute(
+            "CREATE TRIGGER fail_local_day_publication "
+            "BEFORE INSERT ON local_days "
+            "BEGIN SELECT RAISE(ABORT, 'injected local-day publication failure'); END"
+        )
+        writer.commit()
+        writer.close()
+
+        next_time = now + timedelta(minutes=5)
+        second = {
+            **smartctl_fixture,
+            "nvme_smart_health_information_log": {
+                **smartctl_fixture["nvme_smart_health_information_log"],
+                "data_units_written": 12345698,
+                "data_units_read": 9876548,
+            },
+        }
+        failed_result = run_collection(
+            second,
+            sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0",
+            config_fixture,
+            FakeClock(next_time),
+        )
+
+        assert failed_result["ok"] is False
+        assert "injected local-day publication failure" in failed_result["error"]
+
+        with read_status(
+            Path(config_fixture["store_path"]), next_time, query_services=False
+        ) as (reader, composition):
+            assert reader is not None
+            assert composition.sample_count == 1
+            public_counts = reader.execute(
+                "SELECT (SELECT COUNT(*) FROM samples), "
+                "(SELECT COUNT(*) FROM controller_segments), "
+                "(SELECT COUNT(*) FROM monitoring_periods), "
+                "(SELECT COUNT(*) FROM hour_observations), "
+                "(SELECT COUNT(*) FROM day_aggregates), "
+                "(SELECT COUNT(*) FROM local_days)"
+            ).fetchone()
+
+        assert tuple(public_counts) == (1, 1, 1, 0, 0, 0)
+
+
 # ---------------------------------------------------------------------------
 # Display states: awaiting first sample, awaiting another sample
 # ---------------------------------------------------------------------------

tests/test_init_system.py

@@ -634,6 +634,14 @@ class TestMonitorIntegration:
             cmd_enable(args)
             mock_enable.assert_called_once_with(True)
 
+        conn = sqlite3.connect(store_path)
+        try:
+            assert conn.execute(
+                "SELECT COUNT(*) FROM monitoring_periods WHERE ended_at IS NULL"
+            ).fetchone()[0] == 1
+        finally:
+            conn.close()
+
     def test_cmd_disable_uses_init_system(self, tmp_path):
         """cmd_disable uses init_system.disable_timer."""
         from fenris.monitor import cmd_disable

tests/test_tz_util.py

@@ -0,0 +1,26 @@
+"""Timezone detection tests."""
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
+
+from fenris import tz_util
+
+
+def test_relative_localtime_symlink_returns_zoneinfo_key(tmp_path, monkeypatch):
+    zoneinfo = tmp_path / "usr" / "share" / "zoneinfo" / "Asia" / "Kolkata"
+    zoneinfo.parent.mkdir(parents=True)
+    zoneinfo.write_bytes(b"zoneinfo")
+    localtime = tmp_path / "etc" / "localtime"
+    localtime.parent.mkdir()
+    localtime.symlink_to("../usr/share/zoneinfo/Asia/Kolkata")
+
+    real_path = tz_util.Path
+    monkeypatch.setattr(
+        tz_util,
+        "Path",
+        lambda path: localtime if path == "/etc/localtime" else real_path(path),
+    )
+    monkeypatch.delenv("TZ", raising=False)
+
+    assert tz_util.detect_system_tz() == "Asia/Kolkata"