P3: Complete Phase 3 Task Registry + Persistence

Implements the 14-table task-store schema from plan §4 with both SQLite
and Redis backends, enabling pod restart resilience and multi-replica HA.

## Changes

- SqliteTaskStore: Full TaskStore trait implementation for all 14 tables
  - Tables 1-7: tasks, node_settings_version, aliases, sessions,
    idempotency_cache, jobs, leader_lease
  - Tables 8-14: canaries, canary_runs, cdc_cursors, tenant_map,
    rollover_policies, search_ui_config, admin_sessions
  - WAL mode + busy_timeout for concurrent access
  - Idempotent migrations with schema version tracking

- RedisTaskStore: Complete TaskStore trait implementation
  - Mirrors SQLite keyspace with hash + _index pattern for O(1) lookups
  - Uses SET NX/EX for leader leases, ZADD for canary runs
  - Pub/Sub for instant admin session revocation
  - Rate limiting helpers (search_ui, admin_login with backoff)
  - CDC overflow buffer with byte tracking

- Schema migrations: 3-migration system (001_initial, 002_feature_tables,
  003_task_registry_fields)

- Tests:
  - SQLite: 36 tests including property tests (proptest)
  - Redis: 20+ integration tests using testcontainers
  - Restart resilience: tasks survive DB close/reopen cycles

- Helm validation: values.schema.json enforces replicas > 1 requires
  taskStore.backend: redis

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
jedarden 2026-05-02 17:27:48 -04:00
parent a39f0ad9c9
commit 63a9207051

View file

@ -53,24 +53,25 @@ impl RedisPool {
/// Block on an async future using a dedicated runtime.
/// Always spawns a fresh thread with its own single-threaded runtime to
/// avoid "cannot start a runtime from within a runtime" panics.
/// Spawns a dedicated thread with its own single-threaded runtime to avoid
/// "cannot start a runtime from within a runtime" panics when called from
/// within an existing tokio runtime (e.g., in tests).
fn block_on<F>(&self, future: F) -> F::Output
where
F: std::future::Future + Send + 'static,
F::Output: Send + 'static,
{
std::thread::scope(|s| {
s.spawn(|| {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("Failed to create runtime in thread");
rt.block_on(future)
})
.join()
.unwrap_or_else(|_| panic!("block_on thread panicked"))
// Spawn a dedicated thread to run the async future
// This avoids conflicts with any existing tokio runtime
std::thread::spawn(move || {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("Failed to create runtime in thread");
rt.block_on(future)
})
.join()
.unwrap_or_else(|_| panic!("block_on thread panicked"))
}
}
@ -112,6 +113,15 @@ impl RedisTaskStore {
self.pool.block_on(future)
}
/// Helper: run an async future and return a Result, for use in methods that return Result.
fn block_on_result<F, T>(&self, future: F) -> Result<T>
where
F: std::future::Future<Output = Result<T>> + Send + 'static,
T: Send + 'static,
{
self.pool.block_on(future)
}
/// Helper: parse a hash row into a TaskRow.
fn task_from_hash(
miroir_id: String,