feat: add atomic state store and legacy migration
This commit is contained in:
Media
@@ -0,0 +1,315 @@
|
||||
import concurrent.futures
|
||||
import hashlib
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
import yaml
|
||||
|
||||
from controller.config import StorageConfig
|
||||
from controller.models import MigrationError, SlotProtectedError
|
||||
from controller.store import ControllerStore, LEGACY_MIGRATION_NAME
|
||||
|
||||
|
||||
class MutableClock:
|
||||
def __init__(self, value=1_700_000_000.0):
|
||||
self.value = value
|
||||
|
||||
def __call__(self):
|
||||
return self.value
|
||||
|
||||
|
||||
class StoreTestCase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tempdir = tempfile.TemporaryDirectory()
|
||||
self.base = Path(self.tempdir.name)
|
||||
self.media = self.base / "media"
|
||||
self.data = self.base / "data"
|
||||
self.clock = MutableClock()
|
||||
self.config = StorageConfig(
|
||||
media_base=self.media,
|
||||
data_base=self.data,
|
||||
retention_seconds=100,
|
||||
uncertain_review_after_seconds=30,
|
||||
)
|
||||
|
||||
def tearDown(self):
|
||||
self.tempdir.cleanup()
|
||||
|
||||
def open_store(self):
|
||||
return ControllerStore(self.config, now=self.clock).open()
|
||||
|
||||
|
||||
class FreshStoreTests(StoreTestCase):
|
||||
def test_fresh_database_seeds_exact_stable_slots(self):
|
||||
with self.open_store() as store:
|
||||
slots = store.list_slots()
|
||||
self.assertEqual([slot.slot for slot in slots], list(range(1, 11)))
|
||||
self.assertTrue(all(slot.media_state == "empty" for slot in slots))
|
||||
self.assertEqual(store.player_observation().phase, "unknown")
|
||||
|
||||
def test_slot_reservation_is_committed_before_worker_transition(self):
|
||||
with self.open_store() as store:
|
||||
reservation = store.reserve_download(
|
||||
"https://www.youtube.com/playlist?list=one",
|
||||
"Alice",
|
||||
"Service",
|
||||
job_id="job-one",
|
||||
)
|
||||
self.assertEqual(reservation.status, "reserved")
|
||||
self.assertEqual(reservation.slot, 1)
|
||||
self.assertEqual(store.get_slot(1).media_state, "queued")
|
||||
job = store.get_job("job-one")
|
||||
self.assertIsNotNone(job)
|
||||
self.assertEqual(job.status, "reserved")
|
||||
self.assertTrue(store.mark_download_running("job-one", 1, 1))
|
||||
self.assertEqual(store.get_slot(1).media_state, "downloading")
|
||||
|
||||
def test_concurrent_requests_get_only_one_remaining_slot(self):
|
||||
with self.open_store() as store:
|
||||
for index in range(1, 10):
|
||||
result = store.reserve_download(
|
||||
f"https://example.test/{index}",
|
||||
None,
|
||||
None,
|
||||
job_id=f"seed-{index}",
|
||||
)
|
||||
self.assertEqual(result.slot, index)
|
||||
|
||||
def reserve(index):
|
||||
return store.reserve_download(
|
||||
f"https://example.test/concurrent-{index}",
|
||||
None,
|
||||
None,
|
||||
job_id=f"concurrent-{index}",
|
||||
)
|
||||
|
||||
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
|
||||
results = list(pool.map(reserve, range(8)))
|
||||
winners = [result for result in results if result.status == "reserved"]
|
||||
self.assertEqual(len(winners), 1)
|
||||
self.assertEqual(winners[0].slot, 10)
|
||||
self.assertEqual(
|
||||
sum(result.status == "queue_full" for result in results),
|
||||
7,
|
||||
)
|
||||
|
||||
def test_stale_worker_cannot_overwrite_reused_slot(self):
|
||||
with self.open_store() as store:
|
||||
first = store.reserve_download("https://example.test/one", None, None, job_id="old")
|
||||
store.mark_download_running("old", 1, first.generation)
|
||||
store.fail_download("old", 1, first.generation, "failed")
|
||||
deletion = store.begin_delete(1, reason="test")
|
||||
self.assertTrue(store.finish_delete(deletion))
|
||||
second = store.reserve_download("https://example.test/two", None, None, job_id="new")
|
||||
self.assertEqual(second.slot, 1)
|
||||
self.assertFalse(
|
||||
store.complete_download(
|
||||
"old",
|
||||
1,
|
||||
first.generation,
|
||||
title="stale",
|
||||
items=(("Stale", "stale.mp4"),),
|
||||
)
|
||||
)
|
||||
self.assertEqual(store.get_slot(1).url, "https://example.test/two")
|
||||
|
||||
def test_deleting_slot_does_not_renumber_other_slots(self):
|
||||
with self.open_store() as store:
|
||||
first = store.reserve_download("https://example.test/one", None, None, job_id="one")
|
||||
second = store.reserve_download("https://example.test/two", None, None, job_id="two")
|
||||
store.fail_download("one", 1, first.generation, "failed")
|
||||
store.fail_download("two", 2, second.generation, "failed")
|
||||
token = store.begin_delete(1, reason="manual")
|
||||
store.finish_delete(token)
|
||||
self.assertEqual(store.get_slot(1).media_state, "empty")
|
||||
self.assertEqual(store.get_slot(2).slot, 2)
|
||||
self.assertEqual(store.get_slot(2).url, "https://example.test/two")
|
||||
|
||||
|
||||
class GuardAndExpiryTests(StoreTestCase):
|
||||
def _ready_slot(self, store, *, slot_expected=1):
|
||||
reservation = store.reserve_download("https://example.test/ready", None, None, job_id="ready")
|
||||
self.assertEqual(reservation.slot, slot_expected)
|
||||
store.mark_download_running("ready", reservation.slot, reservation.generation)
|
||||
store.complete_download(
|
||||
"ready",
|
||||
reservation.slot,
|
||||
reservation.generation,
|
||||
title="Ready",
|
||||
items=(("One", "one.mp4"),),
|
||||
)
|
||||
return reservation
|
||||
|
||||
def test_active_and_review_guards_block_delete_and_expiry(self):
|
||||
with self.open_store() as store:
|
||||
reservation = self._ready_slot(store)
|
||||
guard = store.protect_slot(
|
||||
1,
|
||||
"restart_uncertain",
|
||||
guard_id="review-me",
|
||||
review_after=self.clock.value + 10,
|
||||
)
|
||||
self.clock.value += 200
|
||||
due = store.mark_due_guards_for_review()
|
||||
self.assertEqual(due[0].state, "review_required")
|
||||
with self.assertRaises(SlotProtectedError):
|
||||
store.begin_delete(1, reason="manual")
|
||||
self.assertEqual(store.begin_expired_deletions(), [])
|
||||
self.assertTrue(
|
||||
store.clear_guard(
|
||||
1,
|
||||
guard.guard_id,
|
||||
actor="operator",
|
||||
reason="Player host replaced",
|
||||
)
|
||||
)
|
||||
self.assertEqual(store.player_observation().phase, "unknown")
|
||||
expired = store.begin_expired_deletions()
|
||||
self.assertEqual([token.slot for token in expired], [reservation.slot])
|
||||
|
||||
def test_admin_clear_requires_current_token_actor_and_reason(self):
|
||||
with self.open_store() as store:
|
||||
self._ready_slot(store)
|
||||
guard = store.protect_slot(1, "player_active", guard_id="current")
|
||||
with self.assertRaises(ValueError):
|
||||
store.clear_guard(1, guard.guard_id, actor="", reason="because")
|
||||
self.assertFalse(
|
||||
store.clear_guard(
|
||||
1,
|
||||
"stale-token",
|
||||
actor="operator",
|
||||
reason="confirmed",
|
||||
)
|
||||
)
|
||||
self.assertEqual(store.list_live_guards()[0].guard_id, "current")
|
||||
|
||||
def test_restart_marks_last_active_slot_uncertain_without_claiming_idle(self):
|
||||
with self.open_store() as store:
|
||||
self._ready_slot(store)
|
||||
store.set_player_observation(
|
||||
availability="online",
|
||||
phase="playing",
|
||||
active_slot=1,
|
||||
active_path="/mnt/Media/YTPlayer/slot1/playlist.m3u",
|
||||
paused=False,
|
||||
reconciliation_state="confirmed",
|
||||
reconciliation_due_at=None,
|
||||
)
|
||||
guard = store.mark_restart_uncertain()
|
||||
self.assertEqual(guard.slot, 1)
|
||||
observation = store.player_observation()
|
||||
self.assertEqual(observation.phase, "unknown")
|
||||
self.assertEqual(observation.reconciliation_state, "awaiting_evidence")
|
||||
self.assertEqual(observation.active_slot, 1)
|
||||
|
||||
|
||||
class LegacyMigrationTests(StoreTestCase):
|
||||
def write_legacy(self, data):
|
||||
self.data.mkdir(parents=True, exist_ok=True)
|
||||
raw = yaml.safe_dump(data, sort_keys=True).encode("utf-8")
|
||||
self.config.legacy_yaml_path.write_bytes(raw)
|
||||
return raw
|
||||
|
||||
def test_migration_preserves_holes_timestamps_original_and_backup(self):
|
||||
raw = self.write_legacy(
|
||||
{
|
||||
"slots": {
|
||||
"2": {
|
||||
"slot": 2,
|
||||
"url": "https://example.test/two",
|
||||
"yt_id": "abc",
|
||||
"title": "Two",
|
||||
"path": "/unsafe/ignored",
|
||||
"m3u": "/unsafe/ignored/playlist.m3u",
|
||||
"created_at": "2025-01-02T03:04:05Z",
|
||||
"expires_at": "2025-01-04T03:04:05Z",
|
||||
"item_count": 3,
|
||||
"state": "playing",
|
||||
},
|
||||
"7": {
|
||||
"slot": 7,
|
||||
"url": "https://example.test/seven",
|
||||
"title": "Seven",
|
||||
"path": "/unsafe/seven",
|
||||
"m3u": "/unsafe/seven/playlist.m3u",
|
||||
"created_at": "2025-02-01T00:00:00Z",
|
||||
"expires_at": "2025-02-03T00:00:00Z",
|
||||
"item_count": 1,
|
||||
"state": "ready",
|
||||
},
|
||||
}
|
||||
}
|
||||
)
|
||||
digest = hashlib.sha256(raw).hexdigest()
|
||||
with self.open_store() as store:
|
||||
self.assertTrue(store.migration_report.migrated)
|
||||
self.assertEqual(store.migration_report.source_sha256, digest)
|
||||
self.assertEqual(store.get_slot(1).media_state, "empty")
|
||||
self.assertEqual(store.get_slot(2).title, "Two")
|
||||
self.assertEqual(store.get_slot(7).slot, 7)
|
||||
self.assertEqual(store.get_slot(2).created_at, 1_735_787_045.0)
|
||||
guards = store.list_live_guards()
|
||||
self.assertEqual([(guard.slot, guard.state) for guard in guards], [(2, "review_required")])
|
||||
marker = store.connection.execute(
|
||||
"SELECT * FROM data_migrations WHERE name = ?",
|
||||
(LEGACY_MIGRATION_NAME,),
|
||||
).fetchone()
|
||||
self.assertEqual(marker["source_sha256"], digest)
|
||||
backup = Path(marker["backup_path"])
|
||||
self.assertEqual(backup.read_bytes(), raw)
|
||||
self.assertEqual(self.config.legacy_yaml_path.read_bytes(), raw)
|
||||
|
||||
def test_completed_migration_is_idempotent_and_ignores_changed_yaml(self):
|
||||
original = self.write_legacy(
|
||||
{
|
||||
"slots": {
|
||||
"3": {
|
||||
"slot": 3,
|
||||
"url": "https://example.test/original",
|
||||
"title": "Original",
|
||||
"path": "/ignored",
|
||||
"m3u": "/ignored/playlist.m3u",
|
||||
"created_at": "2025-01-01T00:00:00Z",
|
||||
"expires_at": "2025-01-03T00:00:00Z",
|
||||
"item_count": 1,
|
||||
"state": "ready",
|
||||
}
|
||||
}
|
||||
}
|
||||
)
|
||||
with self.open_store() as store:
|
||||
generation = store.get_slot(3).generation
|
||||
backup_path = Path(store.migration_report.backup_path)
|
||||
self.config.legacy_yaml_path.write_text("not: [the, original]\n", encoding="utf-8")
|
||||
with self.open_store() as store:
|
||||
self.assertFalse(store.migration_report.migrated)
|
||||
self.assertEqual(store.get_slot(3).title, "Original")
|
||||
self.assertEqual(store.get_slot(3).generation, generation)
|
||||
self.assertEqual(backup_path.read_bytes(), original)
|
||||
|
||||
def test_failed_migration_keeps_original_and_verified_backup_recoverable(self):
|
||||
self.data.mkdir(parents=True, exist_ok=True)
|
||||
raw = b"slots:\n 1: [truncated\n"
|
||||
self.config.legacy_yaml_path.write_bytes(raw)
|
||||
store = ControllerStore(self.config, now=self.clock)
|
||||
with self.assertRaises(MigrationError):
|
||||
store.open()
|
||||
self.assertFalse(self.config.database_path.exists())
|
||||
self.assertEqual(self.config.legacy_yaml_path.read_bytes(), raw)
|
||||
backups = list(self.data.glob("playlists.yaml.pre-v2.*.bak"))
|
||||
self.assertEqual(len(backups), 1)
|
||||
self.assertEqual(backups[0].read_bytes(), raw)
|
||||
|
||||
def test_normalized_duplicate_slot_keys_fail_all_or_nothing(self):
|
||||
self.data.mkdir(parents=True, exist_ok=True)
|
||||
raw = b"slots:\n 1:\n slot: 1\n '01':\n slot: 1\n"
|
||||
self.config.legacy_yaml_path.write_bytes(raw)
|
||||
with self.assertRaisesRegex(MigrationError, "normalize to 1"):
|
||||
self.open_store()
|
||||
self.assertFalse(self.config.database_path.exists())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user