diff --git a/src/fenris/collector.py b/src/fenris/collector.py index 540633f..6c549c4 100644 --- a/src/fenris/collector.py +++ b/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() diff --git a/src/fenris/derive.py b/src/fenris/derive.py index ab62889..6c87da3 100644 --- a/src/fenris/derive.py +++ b/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() diff --git a/src/fenris/monitor.py b/src/fenris/monitor.py index 414cdc7..47f8693 100644 --- a/src/fenris/monitor.py +++ b/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"]) diff --git a/src/fenris/monitoring_periods.py b/src/fenris/monitoring_periods.py index ae16385..622e6f8 100644 --- a/src/fenris/monitoring_periods.py +++ b/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( diff --git a/src/fenris/tz_util.py b/src/fenris/tz_util.py index 40dced0..55fc79a 100644 --- a/src/fenris/tz_util.py +++ b/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" diff --git a/tests/test_collector_history_tracer.py b/tests/test_collector_history_tracer.py index 37624e2..13a3a10 100644 --- a/tests/test_collector_history_tracer.py +++ b/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 # --------------------------------------------------------------------------- diff --git a/tests/test_init_system.py b/tests/test_init_system.py index d8d60c9..a76d724 100644 --- a/tests/test_init_system.py +++ b/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 diff --git a/tests/test_tz_util.py b/tests/test_tz_util.py new file mode 100644 index 0000000..ddabc15 --- /dev/null +++ b/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"