#Persistence Ops Runbook

# Cargo.toml
[features]
persistence-postgres = ["ranvier-runtime/persistence-postgres"]

let pool = sqlx::PgPool::connect(&env::var("DATABASE_URL")?).await?;
let store = PostgresPersistenceStore::new(pool.clone());
store.ensure_schema().await?;   // idempotent -- call once at startup

[features]
persistence-redis = ["ranvier-runtime/persistence-redis"]

let store = RedisPersistenceStore::connect(&env::var("REDIS_URL")?).await?;

// On restart, load the interrupted trace
let trace_id = "order-trace-abc123";
let persisted = store.load(trace_id).await?;

if let Some(trace) = persisted {
    if trace.completion.is_none() {
        // Find the last successfully completed step
        let last_step = trace.events.last().map(|e| e.step).unwrap_or(0);
        let cursor = store.resume(trace_id, last_step).await?;
        
        // Re-run the pipeline from the cursor position
        tracing::info!(
            trace_id,
            resume_from = cursor.next_step,
            "resuming interrupted workflow"
        );
        // ... re-execute with the cursor
    }
}

match store.append(envelope).await {
    Ok(()) => { /* continue */ }
    Err(e) => {
        tracing::error!(
            trace_id = %envelope.trace_id,
            step = envelope.step,
            error = %e,
            "checkpoint failed -- applying circuit breaker"
        );
        // Option A: Halt the pipeline (safest -- no partial progress)
        return Outcome::Fault(anyhow::anyhow!("persistence unavailable: {e}"));
        // Option B: Continue without checkpoint (only for idempotent workflows)
        // tracing::warn!("continuing without checkpoint");
    }
}

if let Err(error) = store.append(envelope.clone()).await {
    // Reconcile the deterministic step key before deciding whether to retry.
    let committed = store
        .load(&envelope.trace_id)
        .await?
        .is_some_and(|trace| {
            trace.events.iter().any(|event| {
                event.step == envelope.step
                    && event.outcome_kind == envelope.outcome_kind
                    && event.payload_hash == envelope.payload_hash
            })
        });
    if !committed {
        return Outcome::Fault(anyhow::anyhow!(
            "persistence append is unresolved: {error}"
        ));
    }
}

.\scripts\dependency_failure_smoke_podman.ps1

BEGIN;
CREATE TEMP TABLE expired_ranvier_traces ON COMMIT DROP AS
SELECT state.trace_id
FROM ranvier_persistence_state AS state
JOIN (
    SELECT trace_id, MAX(timestamp_ms) AS latest_event_ms
    FROM ranvier_persistence_events
    GROUP BY trace_id
) AS event_age USING (trace_id)
WHERE state.completion IS NOT NULL
  AND event_age.latest_event_ms <
      EXTRACT(EPOCH FROM (NOW() - INTERVAL '30 days')) * 1000;

DELETE FROM ranvier_persistence_interventions
WHERE trace_id IN (SELECT trace_id FROM expired_ranvier_traces);
DELETE FROM ranvier_persistence_events
WHERE trace_id IN (SELECT trace_id FROM expired_ranvier_traces);
DELETE FROM ranvier_persistence_state
WHERE trace_id IN (SELECT trace_id FROM expired_ranvier_traces);
COMMIT;

let cutoff_ms = (Utc::now() - Duration::days(30)).timestamp_millis();
let purged = idempotency_store.purge_older_than_ms(cutoff_ms).await?;
tracing::info!(purged_rows = purged, "idempotency cleanup complete");

let pool = sqlx::postgres::PgPoolOptions::new()
    .max_connections(20)     // match your DB max_connections / expected concurrency
    .min_connections(5)      // keep warm connections ready
    .acquire_timeout(Duration::from_secs(3))
    .connect(&database_url).await?;

-- Partition events table into 16 buckets
CREATE TABLE ranvier_persistence_events (
    trace_id TEXT NOT NULL,
    ...
) PARTITION BY HASH (trace_id);

CREATE TABLE ranvier_persistence_events_0
    PARTITION OF ranvier_persistence_events
    FOR VALUES WITH (MODULUS 16, REMAINDER 0);
-- ... repeat for 1..15

# Full dump (for small databases)
pg_dump -U postgres -d mydb -t 'ranvier_persistence_*' \
  --format=custom -f ranvier_persistence_$(date +%Y%m%d).dump

# Continuous WAL archiving (for production)
# Configure postgresql.conf:
# wal_level = replica
# archive_mode = on
# archive_command = 'cp %p /mnt/backup/wal/%f'

# Restore from dump
pg_restore -U postgres -d mydb ranvier_persistence_20260226.dump

# Verify row counts after restore
psql -U postgres -d mydb -c \
  "SELECT COUNT(*) FROM ranvier_persistence_state WHERE completion IS NOT NULL;"

# Trigger immediate save
redis-cli BGSAVE

# Copy the dump file
cp /var/lib/redis/dump.rdb /mnt/backup/redis_$(date +%Y%m%d).rdb

# Stop Redis, replace dump.rdb, restart
systemctl stop redis
cp /mnt/backup/redis_20260226.rdb /var/lib/redis/dump.rdb
systemctl start redis

# Verify persistence keys exist
redis-cli --scan --pattern "ranvier:persistence:*" | wc -l