xavierk/Fenris
1
0
Fork 0
Code Issues 7 Pull Requests Actions Packages Projects Releases 11 Wiki Activity

feat: implement repair and retention for observation history (issue #74)

Browse Source
This commit is contained in:
xavierk
2026-09-14 03:54:46 +05:30
parent d790ff84c5
commit 6917a658cb
4 changed files with 1033 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/fenris/pruning.py
+133 -2
View File
@@ -2,6 +2,7 @@
Raw samples are pruned opportunistically to 14 days.
Hour observations and day aggregates are retained indefinitely.
Boundary anchors required for successor evidence are retained.
"""
import sqlite3
from datetime import datetime, timedelta, timezone
@@ -10,6 +11,117 @@ from datetime import datetime, timedelta, timezone
RAW_SAMPLE_RETENTION_DAYS = 14
def needs_boundary_anchor(
conn: sqlite3.Connection,
sample_ts: str,
now: datetime,
) -> bool:
"""Check if a sample is needed as a boundary anchor for derivation.
A sample is a boundary anchor if:
1. It's older than retention_days (strictly before cutoff)
2. It has a next sample that forms an interval spanning the retention boundary
3. The interval hasn't been derived yet
The interval spans the boundary if:
- The sample is before the cutoff, AND
- The next sample is strictly after the cutoff (or within retention)
"""
from .derive import _parse_ts
sample_dt = _parse_ts(sample_ts)
retention_cutoff = now - timedelta(days=RAW_SAMPLE_RETENTION_DAYS)
# If sample is within retention (strictly after cutoff), not an anchor
if sample_dt > retention_cutoff:
return False
# Check if this sample has a next sample
cursor = conn.execute(
"""SELECT ts, segment_id FROM samples WHERE ts > ? ORDER BY ts LIMIT 1""",
(sample_ts,),
)
next_row = cursor.fetchone()
if next_row is None:
# No next sample - this is the last sample
# It's not needed for derivation (no interval to derive)
return False
next_ts_str = next_row[0]
next_segment_id = next_row[1]
next_dt = _parse_ts(next_ts_str)
# Check if the next sample is strictly after the cutoff (i.e., interval spans boundary)
if next_dt > retention_cutoff:
# The interval spans the retention boundary
# Check if the interval needs derivation
# Get current sample's segment_id
cursor = conn.execute(
"SELECT segment_id FROM samples WHERE ts = ?",
(sample_ts,),
)
current_segment_row = cursor.fetchone()
current_segment_id = current_segment_row[0] if current_segment_row else None
# If different segments, no interval to derive
if current_segment_id != next_segment_id:
return False
# Check if the interval [sample_ts, next_ts] needs derivation
# It needs derivation if any hour in the span lacks an observation
current_hour = sample_dt.replace(minute=0, second=0, microsecond=0)
end_hour = next_dt.replace(minute=0, second=0, microsecond=0)
while current_hour <= end_hour:
cursor = conn.execute(
"SELECT id FROM hour_observations WHERE hour = ?",
(current_hour.isoformat(),),
)
if cursor.fetchone() is None:
# This hour lacks an observation - interval needs derivation
return True
current_hour += timedelta(hours=1)
# All hours in the span have observations - interval is derived
return False
else:
# The interval doesn't span the boundary (both samples are old)
# Check if the interval needs derivation
# Get current sample's segment_id
cursor = conn.execute(
"SELECT segment_id FROM samples WHERE ts = ?",
(sample_ts,),
)
current_segment_row = cursor.fetchone()
current_segment_id = current_segment_row[0] if current_segment_row else None
# If different segments, no interval to derive
if current_segment_id != next_segment_id:
return False
# Check if the interval [sample_ts, next_ts] needs derivation
current_hour = sample_dt.replace(minute=0, second=0, microsecond=0)
end_hour = next_dt.replace(minute=0, second=0, microsecond=0)
while current_hour <= end_hour:
cursor = conn.execute(
"SELECT id FROM hour_observations WHERE hour = ?",
(current_hour.isoformat(),),
)
if cursor.fetchone() is None:
# This hour lacks an observation - interval needs derivation
# But only keep if the interval is significant (spans multiple hours)
# or if the next sample is the last sample before a gap
gap = (next_dt - sample_dt).total_seconds()
if gap > 24 * 3600: # Significant gap (> 24 hours)
return True
current_hour += timedelta(hours=1)
# All hours in the span have observations or gap is not significant
return False
def prune_old_samples(
conn: sqlite3.Connection,
now: datetime,
@@ -17,6 +129,8 @@ def prune_old_samples(
) -> int:
"""Remove raw samples older than retention_days.
Retains boundary anchors required for successor evidence.
Args:
conn: Connection to the observation store.
now: Current UTC time.
@@ -26,6 +140,23 @@ def prune_old_samples(
Number of samples removed.
"""
cutoff = (now - timedelta(days=retention_days)).isoformat()
cursor = conn.execute("DELETE FROM samples WHERE ts < ?", (cutoff,))
# Get all samples older than cutoff
cursor = conn.execute(
"SELECT id, ts FROM samples WHERE ts < ? ORDER BY ts",
(cutoff,),
)
old_samples = cursor.fetchall()
removed = 0
for sample_id, sample_ts in old_samples:
# Check if this sample is a boundary anchor
if needs_boundary_anchor(conn, sample_ts, now):
continue # Skip - it's a boundary anchor
# Remove the sample
conn.execute("DELETE FROM samples WHERE id = ?", (sample_id,))
removed += 1
conn.commit()
return cursor.rowcount
return removed
src/fenris/repair.py
+448
View File
@@ -0,0 +1,448 @@
"""Historical repair and retention (issue #74).
Re-derives hour observations and day aggregates from surviving raw samples,
with idempotent and interruption-safe guarantees. Preserves import markers,
boundary anchors, and valid historical summaries.
Contracts:
- Idempotent: running repair multiple times produces no duplicates
- Interruption-safe: partial repair preserves prior valid history
- Preserves existing valid data: never overwrites valid derived data
- Surfaces failures explicitly for retry
"""
import logging
import sqlite3
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from typing import Optional, List, Tuple
from .derive import find_previous_sample, derive_hours_from_interval, _parse_ts
logger = logging.getLogger(__name__)
@dataclass
class RepairResult:
"""Result of a repair operation."""
ok: bool
hours_created: int = 0
hours_updated: int = 0
days_created: int = 0
days_updated: int = 0
intervals_derived: int = 0
boundary_anchors_retained: int = 0
legacy_summaries_preserved: int = 0
error: Optional[str] = None
@dataclass
class RepairStatus:
"""Current repair status for read-only views."""
last_repair: Optional[str] = None # ISO timestamp of last successful repair
repair_in_progress: bool = False
hours_derived: int = 0
days_derived: int = 0
def _ensure_repair_metadata(conn: sqlite3.Connection) -> None:
"""Ensure metadata table exists for tracking repair state."""
conn.execute("""
CREATE TABLE IF NOT EXISTS store_metadata (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
)
""")
def is_repair_in_progress(conn: sqlite3.Connection) -> bool:
"""Check if a repair operation is currently in progress."""
_ensure_repair_metadata(conn)
cursor = conn.execute(
"SELECT value FROM store_metadata WHERE key = 'repair_in_progress'"
)
row = cursor.fetchone()
return row is not None and row[0] == "true"
def _set_repair_in_progress(conn: sqlite3.Connection, in_progress: bool) -> None:
"""Mark repair as in progress or complete."""
_ensure_repair_metadata(conn)
conn.execute(
"INSERT OR REPLACE INTO store_metadata (key, value) VALUES (?, ?)",
("repair_in_progress", "true" if in_progress else "false"),
)
conn.commit()
def _update_repair_status(conn: sqlite3.Connection, result: RepairResult) -> None:
"""Update repair status after successful completion."""
_ensure_repair_metadata(conn)
now = datetime.now(timezone.utc).isoformat()
# Update last repair timestamp
conn.execute(
"INSERT OR REPLACE INTO store_metadata (key, value) VALUES (?, ?)",
("last_repair", now),
)
# Update derived counts
cursor = conn.execute("SELECT COUNT(*) FROM hour_observations")
hours = cursor.fetchone()[0]
cursor = conn.execute("SELECT COUNT(*) FROM day_aggregates")
days = cursor.fetchone()[0]
conn.execute(
"INSERT OR REPLACE INTO store_metadata (key, value) VALUES (?, ?)",
("hours_derived", str(hours)),
)
conn.execute(
"INSERT OR REPLACE INTO store_metadata (key, value) VALUES (?, ?)",
("days_derived", str(days)),
)
conn.commit()
def get_repair_status(conn: sqlite3.Connection) -> RepairStatus:
"""Get current repair status for read-only views."""
_ensure_repair_metadata(conn)
last_repair = None
cursor = conn.execute(
"SELECT value FROM store_metadata WHERE key = 'last_repair'"
)
row = cursor.fetchone()
if row:
last_repair = row[0]
in_progress = is_repair_in_progress(conn)
hours_derived = 0
cursor = conn.execute(
"SELECT value FROM store_metadata WHERE key = 'hours_derived'"
)
row = cursor.fetchone()
if row:
hours_derived = int(row[0])
days_derived = 0
cursor = conn.execute(
"SELECT value FROM store_metadata WHERE key = 'days_derived'"
)
row = cursor.fetchone()
if row:
days_derived = int(row[0])
return RepairStatus(
last_repair=last_repair,
repair_in_progress=in_progress,
hours_derived=hours_derived,
days_derived=days_derived,
)
def needs_boundary_anchor(
conn: sqlite3.Connection,
sample_ts: str,
now: datetime,
) -> bool:
"""Check if a sample is needed as a boundary anchor for derivation.
A sample is a boundary anchor if:
1. It's older than retention_days
2. It has no derived hour observation for its hour
3. It's the last sample before a gap that needs derivation (gap > 24 hours)
"""
from .pruning import RAW_SAMPLE_RETENTION_DAYS
sample_dt = _parse_ts(sample_ts)
retention_cutoff = now - timedelta(days=RAW_SAMPLE_RETENTION_DAYS)
# If sample is within retention, not an anchor (will be kept anyway)
if sample_dt >= retention_cutoff:
return False
# Check if this sample's hour already has a derived observation
hour_start = sample_dt.replace(minute=0, second=0, microsecond=0).isoformat()
hour_end = (sample_dt + timedelta(hours=1)).replace(minute=0, second=0, microsecond=0).isoformat()
cursor = conn.execute(
"""SELECT COUNT(*) FROM hour_observations
WHERE hour >= ? AND hour < ?""",
(hour_start, hour_end),
)
# If there's an hour observation in this sample's hour, it's been derived
if cursor.fetchone()[0] > 0:
return False
# Check if this sample is the last sample before a gap
# (i.e., the next sample is significantly later)
cursor = conn.execute(
"""SELECT ts FROM samples WHERE ts > ? ORDER BY ts LIMIT 1""",
(sample_ts,),
)
next_row = cursor.fetchone()
if next_row is None:
# No next sample - this is the last sample, might be needed
# But if it's old and fully derived, it's not needed
return False
next_ts = _parse_ts(next_row[0])
gap = (next_ts - sample_dt).total_seconds()
# If gap > 24 hours, this sample is a boundary anchor
# (needed to derive the interval spanning the gap)
return gap > 24 * 3600
def _get_unlinked_intervals(conn: sqlite3.Connection) -> List[Tuple[dict, dict]]:
"""Find sample pairs that form intervals but have no hour observations."""
cursor = conn.execute(
"""SELECT id, ts, bytes_written, bytes_read, power_on_hours,
temperature_c, data_units_written, data_units_read, segment_id
FROM samples ORDER BY ts"""
)
all_samples = []
for row in cursor.fetchall():
all_samples.append({
"id": row[0], "ts": row[1], "bytes_written": row[2],
"bytes_read": row[3], "power_on_hours": row[4],
"temperature_c": row[5], "data_units_written": row[6],
"data_units_read": row[7], "segment_id": row[8],
})
intervals = []
for i in range(len(all_samples) - 1):
prev = all_samples[i]
next_s = all_samples[i + 1]
# Skip if different segments
if prev["segment_id"] != next_s["segment_id"]:
continue
# Check if the interval spans hours that need derivation
prev_dt = _parse_ts(prev["ts"])
next_dt = _parse_ts(next_s["ts"])
# Check if any hour in the span lacks an observation
current = prev_dt.replace(minute=0, second=0, microsecond=0)
end = next_dt.replace(minute=0, second=0, microsecond=0)
needs_derivation = False
while current <= end:
cursor2 = conn.execute(
"SELECT id FROM hour_observations WHERE hour = ?",
(current.isoformat(),),
)
if cursor2.fetchone() is None:
needs_derivation = True
break
current += timedelta(hours=1)
if needs_derivation:
intervals.append((prev, next_s))
return intervals
def _derive_day_aggregate_from_hours(
conn: sqlite3.Connection,
day: str,
) -> Optional[dict]:
"""Derive a day aggregate from its hour observations."""
cursor = conn.execute(
"""SELECT SUM(active_seconds), SUM(idle_seconds),
SUM(powered_off_seconds), SUM(unknown_seconds),
SUM(bytes_written_delta), SUM(bytes_read_delta),
SUM(sample_count)
FROM hour_observations WHERE hour LIKE ?""",
(day + "T%",),
)
row = cursor.fetchone()
if row is None or row[0] is None:
return None
return {
"day": day,
"active_seconds": row[0] or 0,
"idle_seconds": row[1] or 0,
"powered_off_seconds": row[2] or 0,
"unknown_seconds": row[3] or 0,
"bytes_written_delta": row[4] or 0,
"bytes_read_delta": row[5] or 0,
"sample_count": row[6] or 0,
}
def _upsert_day_aggregate(conn: sqlite3.Connection, day_data: dict) -> bool:
"""Insert or update a day aggregate. Returns True if created."""
existing = conn.execute(
"SELECT id FROM day_aggregates WHERE day = ?",
(day_data["day"],),
).fetchone()
if existing is None:
# Calculate coverage
total_seconds = (day_data["active_seconds"] + day_data["idle_seconds"] +
day_data["powered_off_seconds"] + day_data["unknown_seconds"])
coverage = (day_data["active_seconds"] + day_data["idle_seconds"] +
day_data["powered_off_seconds"]) / total_seconds if total_seconds > 0 else 0.0
conn.execute(
"""INSERT INTO day_aggregates
(day, active_seconds, idle_seconds, powered_off_seconds, unknown_seconds,
bytes_written_delta, bytes_read_delta, sample_count, coverage)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(day_data["day"], day_data["active_seconds"], day_data["idle_seconds"],
day_data["powered_off_seconds"], day_data["unknown_seconds"],
day_data["bytes_written_delta"], day_data["bytes_read_delta"],
day_data["sample_count"], coverage),
)
return True
else:
# Update existing (but only if new data is more complete)
# This implements the "don't overwrite valid older history" rule
cursor = conn.execute(
"""SELECT bytes_written_delta, sample_count
FROM day_aggregates WHERE day = ?""",
(day_data["day"],),
)
existing_data = cursor.fetchone()
# Only update if new data has more samples or more bytes
if (day_data["sample_count"] > existing_data[1] or
day_data["bytes_written_delta"] > existing_data[0]):
total_seconds = (day_data["active_seconds"] + day_data["idle_seconds"] +
day_data["powered_off_seconds"] + day_data["unknown_seconds"])
coverage = (day_data["active_seconds"] + day_data["idle_seconds"] +
day_data["powered_off_seconds"]) / total_seconds if total_seconds > 0 else 0.0
conn.execute(
"""UPDATE day_aggregates SET
active_seconds = ?, idle_seconds = ?, powered_off_seconds = ?,
unknown_seconds = ?, bytes_written_delta = ?, bytes_read_delta = ?,
sample_count = ?, coverage = ?
WHERE day = ?""",
(day_data["active_seconds"], day_data["idle_seconds"],
day_data["powered_off_seconds"], day_data["unknown_seconds"],
day_data["bytes_written_delta"], day_data["bytes_read_delta"],
day_data["sample_count"], coverage, day_data["day"]),
)
return False # Updated, not created
else:
return False # No update needed
def repair_derivation(
conn: sqlite3.Connection,
clock=None,
) -> RepairResult:
"""Repair hour observations and day aggregates from surviving samples.
This is the main entry point for historical repair. It:
1. Finds sample pairs that need interval derivation
2. Derives hour observations from those intervals
3. Updates day aggregates from the hour observations
4. Preserves existing valid data
5. Is idempotent and interruption-safe
Args:
conn: Connection to the observation store
clock: Injected clock (for testing)
Returns:
RepairResult with operation details
"""
if clock is None:
clock = datetime.now(timezone.utc)
elif hasattr(clock, 'utcnow'):
clock = clock.utcnow()
# Check if repair is already in progress
if is_repair_in_progress(conn):
return RepairResult(
ok=False,
error="Repair already in progress",
)
# Mark repair as in progress
_set_repair_in_progress(conn, True)
result = RepairResult(ok=True)
try:
# Begin transaction
conn.execute("BEGIN IMMEDIATE")
# 1. Find and derive intervals from sample pairs
intervals = _get_unlinked_intervals(conn)
for prev, next_s in intervals:
try:
derived_hours = derive_hours_from_interval(conn, prev, next_s)
result.intervals_derived += 1
result.hours_created += len([h for h in derived_hours if h.get("attributed", True)])
except Exception as e:
logger.warning("Failed to derive interval %s -> %s: %s",
prev["ts"], next_s["ts"], e)
# Continue with other intervals (resilient)
# 2. Update day aggregates from hour observations
cursor = conn.execute(
"SELECT DISTINCT substr(hour, 1, 10) as day FROM hour_observations ORDER BY day"
)
days = [row[0] for row in cursor.fetchall()]
for day in days:
day_data = _derive_day_aggregate_from_hours(conn, day)
if day_data is not None:
created = _upsert_day_aggregate(conn, day_data)
if created:
result.days_created += 1
else:
result.days_updated += 1
# 3. Count boundary anchors retained
now = clock if isinstance(clock, datetime) else datetime.now(timezone.utc)
cursor = conn.execute("SELECT ts FROM samples ORDER BY ts")
anchor_count = 0
for row in cursor.fetchall():
if needs_boundary_anchor(conn, row[0], now):
anchor_count += 1
result.boundary_anchors_retained = anchor_count
# 4. Count preserved legacy summaries
# Legacy summaries are day aggregates without corresponding hour observations
cursor = conn.execute(
"""SELECT COUNT(*) FROM day_aggregates d
WHERE NOT EXISTS (
SELECT 1 FROM hour_observations h
WHERE h.hour LIKE d.day || 'T%'
)"""
)
result.legacy_summaries_preserved = cursor.fetchone()[0]
# Commit transaction
conn.commit()
# Update repair status
_update_repair_status(conn, result)
except Exception as e:
conn.rollback()
logger.error("Repair failed: %s", e)
return RepairResult(
ok=False,
error=str(e),
)
finally:
# Mark repair as complete
_set_repair_in_progress(conn, False)
return result
tests/test_pruning.py
+54 -3
View File
@@ -1,7 +1,9 @@
"""Raw sample pruning tests.
Spec §3.4, ST-5: Raw samples pruned to 14 days; hour observations and
day aggregates retained indefinitely.
day aggregates are retained indefinitely.
Issue #74: Boundary anchors required for successor evidence are retained.
"""
import sqlite3
import sys
@@ -51,6 +53,9 @@ class TestPruneOldSamples:
def test_removes_old_samples(self, store_conn):
now = datetime(2026, 9, 15, 12, 0, 0, tzinfo=timezone.utc)
# Insert samples at 10, 14, and 15 days ago
# All three are before the cutoff (2026-09-01T12:00:00)
# The 15-day-old sample is not a boundary anchor because
# the next sample (14 days ago) is also before the cutoff
for days_ago in [10, 14, 15]:
ts = (now - timedelta(days=days_ago)).isoformat()
_insert_sample(store_conn, ts)
@@ -82,6 +87,7 @@ class TestPruneOldSamples:
"""Hour observations are retained indefinitely."""
now = datetime(2026, 9, 15, 12, 0, 0, tzinfo=timezone.utc)
# Insert an old sample and a recent sample
# The old sample is a boundary anchor (needed for derivation)
_insert_sample(store_conn, (now - timedelta(days=20)).isoformat())
_insert_sample(store_conn, (now - timedelta(days=1)).isoformat())
@@ -95,10 +101,55 @@ class TestPruneOldSamples:
prune_old_samples(store_conn, now, retention_days=14)
# Sample removed
# Old sample is retained as boundary anchor (needed for derivation)
cursor = store_conn.execute("SELECT COUNT(*) FROM samples")
assert cursor.fetchone()[0] == 1
assert cursor.fetchone()[0] == 2
# Hour observation retained
cursor = store_conn.execute("SELECT COUNT(*) FROM hour_observations")
assert cursor.fetchone()[0] == 1
def test_boundary_anchor_retained(self, store_conn):
"""Boundary anchors required for derivation are retained."""
now = datetime(2026, 9, 30, 12, 0, 0, tzinfo=timezone.utc)
# Old sample before boundary (2026-09-14T23:55:00)
# is 15 days and 0.75 hours old (before cutoff at 2026-09-16T12:00:00)
_insert_sample(store_conn, "2026-09-14T23:55:00+00:00", 1000000)
# Sample after boundary (2026-09-16T12:30:00)
# is 13 days and 23.5 hours old (within retention)
_insert_sample(store_conn, "2026-09-16T12:30:00+00:00", 2000000)
# Run pruning
pruned = prune_old_samples(store_conn, now, retention_days=14)
# The boundary anchor should be retained
cursor = store_conn.execute(
"SELECT COUNT(*) FROM samples WHERE ts = '2026-09-14T23:55:00+00:00'"
)
assert cursor.fetchone()[0] == 1
def test_old_sample_with_derived_interval_removed(self, store_conn):
"""Old samples with fully derived intervals are removed."""
now = datetime(2026, 9, 30, 12, 0, 0, tzinfo=timezone.utc)
# Old sample with derived interval
_insert_sample(store_conn, "2026-09-10T10:00:00+00:00", 1000000)
_insert_sample(store_conn, "2026-09-10T10:30:00+00:00", 2000000)
# Hour observation exists for the interval
store_conn.execute(
"INSERT INTO hour_observations (hour, active_seconds, bytes_written_delta, bytes_read_delta, sample_count, coverage) "
"VALUES ('2026-09-10T10:00:00+00:00', 3600, 1000000, 0, 1, 1.0)",
)
store_conn.commit()
# Run pruning
pruned = prune_old_samples(store_conn, now, retention_days=14)
# Old sample should be removed (interval is derived)
cursor = store_conn.execute(
"SELECT COUNT(*) FROM samples WHERE ts = '2026-09-10T10:00:00+00:00'"
)
assert cursor.fetchone()[0] == 0
tests/test_repair.py
+398
View File
@@ -0,0 +1,398 @@
"""Repair and retention tests (issue #74).
Tests the idempotent, safe repair of hour observations and day aggregates
from surviving raw samples, boundary anchor retention, and legacy summary
handling at actual precision.
"""
import sqlite3
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from fenris.store import init_store
from fenris.repair import (
repair_derivation,
is_repair_in_progress,
get_repair_status,
)
from fenris.pruning import prune_old_samples, needs_boundary_anchor
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
def store_conn(tmp_path: Path):
"""Create a fresh store for each test."""
db_path = tmp_path / "test.db"
conn = init_store(db_path)
yield conn
conn.close()
def _insert_sample(conn, ts_iso, bytes_written, bytes_read=0, power_on_hours=100,
device="/dev/nvme0", segment_id=None):
"""Insert a raw sample."""
conn.execute(
"""INSERT INTO samples
(ts, device, bytes_written, bytes_read, power_on_hours,
data_units_written, data_units_read, segment_id)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
(ts_iso, device, bytes_written, bytes_read, power_on_hours,
bytes_written // 512000, bytes_read // 512000, segment_id),
)
conn.commit()
def _insert_hour(conn, hour_iso, bytes_written_delta=0, active_seconds=3600,
idle_seconds=0, powered_off_seconds=0, unknown_seconds=0,
sample_count=1, coverage=1.0):
"""Insert an hour observation."""
conn.execute(
"""INSERT INTO hour_observations
(hour, active_seconds, idle_seconds, powered_off_seconds, unknown_seconds,
bytes_written_delta, bytes_read_delta, sample_count, coverage)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(hour_iso, active_seconds, idle_seconds, powered_off_seconds, unknown_seconds,
bytes_written_delta, 0, sample_count, coverage),
)
conn.commit()
def _insert_day_aggregate(conn, day, bytes_written_delta=0, coverage=1.0):
"""Insert a day aggregate."""
conn.execute(
"""INSERT INTO day_aggregates
(day, active_seconds, bytes_written_delta, coverage, sample_count)
VALUES (?, 3600, ?, ?, 1)""",
(day, bytes_written_delta, coverage),
)
conn.commit()
def _open_period(conn, start_iso, end_iso=None, end_cause=None):
"""Insert a monitoring period."""
conn.execute(
"""INSERT INTO monitoring_periods (started_at, ended_at, end_cause)
VALUES (?, ?, ?)""",
(start_iso, end_iso, end_cause),
)
conn.commit()
# ---------------------------------------------------------------------------
# AC1: Normal collection, legacy import and recovery use same evidence rules
# ---------------------------------------------------------------------------
class TestRepairUsesSameEvidenceRules:
"""AC1: Repair derives only surviving supported evidence, transactionally
and idempotently."""
def test_repair_idempotent_on_empty_store(self, store_conn):
"""Repair on empty store succeeds and does nothing."""
result = repair_derivation(store_conn)
assert result.ok is True
assert result.hours_created == 0
assert result.days_created == 0
def test_repair_idempotent_on_fully_derived(self, store_conn):
"""Repair on store with existing derived data does not duplicate."""
now = datetime(2026, 9, 15, 12, 0, 0, tzinfo=timezone.utc)
_open_period(store_conn, "2026-09-14T00:00:00+00:00")
# Insert samples that span an hour
_insert_sample(store_conn, "2026-09-14T10:00:00+00:00", 1000000)
_insert_sample(store_conn, "2026-09-14T10:30:00+00:00", 2000000)
# Pre-existing hour observation for the 10:00 hour
# This hour observation captures the interval [10:00, 10:30]
_insert_hour(store_conn, "2026-09-14T10:00:00+00:00",
bytes_written_delta=1000000)
_insert_day_aggregate(store_conn, "2026-09-14",
bytes_written_delta=1000000)
# Run repair
result = repair_derivation(store_conn)
# Should not create new observations (already exist)
assert result.hours_created == 0
assert result.days_created == 0
# Verify no duplicates
cursor = store_conn.execute(
"SELECT COUNT(*) FROM hour_observations WHERE hour = '2026-09-14T10:00:00+00:00'"
)
assert cursor.fetchone()[0] == 1
def test_repair_preserves_import_markers(self, store_conn):
"""Repair does not remove legacy import markers."""
# Set legacy import marker
store_conn.execute(
"INSERT INTO store_metadata (key, value) VALUES ('legacy_imported', 'true')"
)
store_conn.commit()
# Run repair
result = repair_derivation(store_conn)
# Verify marker preserved
cursor = store_conn.execute(
"SELECT value FROM store_metadata WHERE key = 'legacy_imported'"
)
assert cursor.fetchone()[0] == "true"
# ---------------------------------------------------------------------------
# AC2: Rerunning or interrupting repair produces no duplicates
# ---------------------------------------------------------------------------
class TestRepairIdempotency:
"""AC2: No duplicated intervals or totals from repeated repair."""
def test_repair_no_duplicate_hours(self, store_conn):
"""Running repair twice produces no duplicate hour observations."""
_open_period(store_conn, "2026-09-14T00:00:00+00:00")
_insert_sample(store_conn, "2026-09-14T10:00:00+00:00", 1000000)
_insert_sample(store_conn, "2026-09-14T10:30:00+00:00", 2000000)
# First repair
result1 = repair_derivation(store_conn)
assert result1.hours_created == 1
# Second repair
result2 = repair_derivation(store_conn)
assert result2.hours_created == 0 # No new hours
# Verify only one hour observation
cursor = store_conn.execute("SELECT COUNT(*) FROM hour_observations")
assert cursor.fetchone()[0] == 1
def test_repair_no_duplicate_days(self, store_conn):
"""Running repair twice produces no duplicate day aggregates."""
_open_period(store_conn, "2026-09-14T00:00:00+00:00")
_insert_sample(store_conn, "2026-09-14T10:00:00+00:00", 1000000)
_insert_sample(store_conn, "2026-09-14T10:30:00+00:00", 2000000)
# First repair
result1 = repair_derivation(store_conn)
assert result1.days_created == 1
# Second repair
result2 = repair_derivation(store_conn)
assert result2.days_created == 0 # No new days
# Verify only one day aggregate
cursor = store_conn.execute("SELECT COUNT(*) FROM day_aggregates")
assert cursor.fetchone()[0] == 1
def test_interrupted_repair_preserves_evidence(self, store_conn):
"""If repair fails, prior valid history is preserved."""
_open_period(store_conn, "2026-09-14T00:00:00+00:00")
# Insert valid existing data
_insert_hour(store_conn, "2026-09-14T10:00:00+00:00",
bytes_written_delta=500000)
_insert_day_aggregate(store_conn, "2026-09-14",
bytes_written_delta=500000)
# Run repair (should succeed but not modify existing valid data)
result = repair_derivation(store_conn)
assert result.ok is True
# Verify existing data preserved
cursor = store_conn.execute(
"SELECT bytes_written_delta FROM hour_observations "
"WHERE hour = '2026-09-14T10:00:00+00:00'"
)
assert cursor.fetchone()[0] == 500000
# ---------------------------------------------------------------------------
# AC3: Boundary anchor retention
# ---------------------------------------------------------------------------
class TestBoundaryAnchorRetention:
"""AC3: Prune samples only after durable derivation; retain anchors."""
def test_needs_boundary_anchor_sample(self, store_conn):
"""Sample before 14-day boundary is needed for derivation."""
now = datetime(2026, 9, 30, 12, 0, 0, tzinfo=timezone.utc)
# Sample just before 14-day boundary (2026-09-15T23:55:00)
# is 14 days and 0.75 hours old (before cutoff at 2026-09-16T12:00:00)
_insert_sample(store_conn, "2026-09-15T23:55:00+00:00", 1000000)
# Sample just after boundary (2026-09-16T12:30:00)
# is 13 days and 23.5 hours old (within retention)
_insert_sample(store_conn, "2026-09-16T12:30:00+00:00", 2000000)
# The sample at 2026-09-15 is a boundary anchor because
# the interval spans the retention boundary
assert needs_boundary_anchor(store_conn, "2026-09-15T23:55:00+00:00", now)
def test_not_boundary_anchor_if_fully_derived(self, store_conn):
"""Sample that's fully derived is not a boundary anchor."""
now = datetime(2026, 9, 30, 12, 0, 0, tzinfo=timezone.utc)
# Insert sample and fully derive its interval
_insert_sample(store_conn, "2026-09-14T10:00:00+00:00", 1000000)
_insert_sample(store_conn, "2026-09-14T10:30:00+00:00", 2000000)
# Hour observation already exists for this interval
_insert_hour(store_conn, "2026-09-14T10:00:00+00:00",
bytes_written_delta=1000000)
# Not a boundary anchor
assert not needs_boundary_anchor(store_conn, "2026-09-14T10:00:00+00:00", now)
def test_pruning_retains_boundary_anchors(self, store_conn):
"""Pruning keeps samples needed as boundary anchors."""
now = datetime(2026, 9, 30, 12, 0, 0, tzinfo=timezone.utc)
# Old sample before boundary (2026-09-14T23:55:00)
# is 15 days and 0.75 hours old (before cutoff at 2026-09-16T12:00:00)
_insert_sample(store_conn, "2026-09-14T23:55:00+00:00", 1000000)
# Sample after boundary (2026-09-16T12:30:00)
# is 13 days and 23.5 hours old (within retention)
_insert_sample(store_conn, "2026-09-16T12:30:00+00:00", 2000000)
# Recent sample
_insert_sample(store_conn, "2026-09-29T12:00:00+00:00", 3000000)
# Run pruning
pruned = prune_old_samples(store_conn, now, retention_days=14)
# The boundary anchor should be retained
cursor = store_conn.execute(
"SELECT COUNT(*) FROM samples WHERE ts = '2026-09-14T23:55:00+00:00'"
)
assert cursor.fetchone()[0] == 1
def test_pruning_removes_old_sample_with_derived_interval(self, store_conn):
"""Pruning removes old samples when interval is fully derived."""
now = datetime(2026, 9, 30, 12, 0, 0, tzinfo=timezone.utc)
# Old sample with derived interval
_insert_sample(store_conn, "2026-09-10T10:00:00+00:00", 1000000)
_insert_sample(store_conn, "2026-09-10T10:30:00+00:00", 2000000)
# Hour observation exists for the interval
_insert_hour(store_conn, "2026-09-10T10:00:00+00:00",
bytes_written_delta=1000000)
# Run pruning
pruned = prune_old_samples(store_conn, now, retention_days=14)
# Old sample should be removed (interval is derived)
cursor = store_conn.execute(
"SELECT COUNT(*) FROM samples WHERE ts = '2026-09-10T10:00:00+00:00'"
)
assert cursor.fetchone()[0] == 0
# ---------------------------------------------------------------------------
# AC4: Legacy day-only summaries retain actual precision
# ---------------------------------------------------------------------------
class TestLegacySummaryPrecision:
"""AC4: Legacy summaries at actual precision, no interpolation."""
def test_legacy_summary_not_reconstructed(self, store_conn):
"""Legacy day-only summaries are not interpolated to hour detail."""
# Insert a legacy-style day aggregate without hour observations
_insert_day_aggregate(store_conn, "2026-08-01",
bytes_written_delta=5000000)
# Run repair
result = repair_derivation(store_conn)
# Should not create hour observations for legacy day
cursor = store_conn.execute(
"SELECT COUNT(*) FROM hour_observations WHERE hour LIKE '2026-08-01%'"
)
assert cursor.fetchone()[0] == 0
def test_legacy_summary_no_double_counting(self, store_conn):
"""Legacy summaries and derived intervals don't double-count."""
# Insert legacy day aggregate
_insert_day_aggregate(store_conn, "2026-08-01",
bytes_written_delta=5000000)
# Run repair
result = repair_derivation(store_conn)
# Day aggregate should not be modified
cursor = store_conn.execute(
"SELECT bytes_written_delta FROM day_aggregates WHERE day = '2026-08-01'"
)
assert cursor.fetchone()[0] == 5000000
# ---------------------------------------------------------------------------
# AC5: Status distinguishes evidence states
# ---------------------------------------------------------------------------
class TestStatusEvidenceDistingushing:
"""AC5: Read-only views distinguish evidence states."""
def test_repair_status_available(self, store_conn):
"""Repair status is available for read-only views."""
status = get_repair_status(store_conn)
assert hasattr(status, 'last_repair')
assert hasattr(status, 'repair_in_progress')
assert hasattr(status, 'hours_derived')
assert hasattr(status, 'days_derived')
def test_repair_in_progress_flag(self, store_conn):
"""Repair in progress flag is trackable."""
assert not is_repair_in_progress(store_conn)
# ---------------------------------------------------------------------------
# AC6: Migration then collection then reader consumption
# ---------------------------------------------------------------------------
class TestMigrationCollectionReader:
"""AC6: End-to-end migration, collection, and reader consumption."""
def test_store_with_raw_evidence_and_summaries(self, store_conn):
"""Store with raw evidence and old summaries works correctly."""
# Set up store with mixed data
_open_period(store_conn, "2026-08-01T00:00:00+00:00")
# Old day-only summary (legacy)
_insert_day_aggregate(store_conn, "2026-08-01",
bytes_written_delta=5000000)
# Recent raw samples
_insert_sample(store_conn, "2026-09-14T10:00:00+00:00", 1000000)
_insert_sample(store_conn, "2026-09-14T10:30:00+00:00", 2000000)
# Run repair
result = repair_derivation(store_conn)
assert result.ok is True
# Verify legacy summary preserved
cursor = store_conn.execute(
"SELECT bytes_written_delta FROM day_aggregates WHERE day = '2026-08-01'"
)
assert cursor.fetchone()[0] == 5000000
# Verify new hour observation created
cursor = store_conn.execute(
"SELECT COUNT(*) FROM hour_observations WHERE hour LIKE '2026-09-14%'"
)
assert cursor.fetchone()[0] == 1
def test_concurrent_reader_consistency(self, store_conn):
"""Reader sees consistent snapshot during repair."""
# This is more of a documentation test - SQLite WAL mode handles this
# We verify the store is in WAL mode
cursor = store_conn.execute("PRAGMA journal_mode")
assert cursor.fetchone()[0] == "wal"