feltdb 0.5.0

/// Phase 6 Integration Tests — Persistence + Phase 5 Gates
///
/// Critical invariant: Phase 5 guarantees (convergence, causality, contracts)
/// must survive persistence and crash/recovery cycles.

#[cfg(test)]
mod tests {
    use crate::causal_dependency_barrier::CausalDependencyBarrier;
    use crate::consistency_contract::{CommutativeVerificationResult, ConsistencyContractEnforcer};
    use crate::convergence::VectorClock;
    use crate::persistence_reality::{
        DurablePersistenceLayer, PersistenceContractVerifier, PersistenceCrashPoint, RecoveryResult,
    };

    #[test]
    fn test_phase6_gate7_persistence_reality_fsync_safety() {
        // Gate 7: Verify fsync actually makes data durable
        let mut wal = DurablePersistenceLayer::new();

        // Scenario A: Crash before fsync
        wal.write_with_crash(
            "critical_op".to_string(),
            vec![1, 2, 3],
            PersistenceCrashPoint::AfterBufferBeforeFsync,
        );

        let recovery_a = wal.recover();
        // Data should NOT be recovered (not fsynced)
        assert_eq!(recovery_a.recovered_operations.len(), 0);

        // Scenario B: Crash after fsync
        let mut wal2 = DurablePersistenceLayer::new();
        wal2.write_with_crash(
            "critical_op".to_string(),
            vec![1, 2, 3],
            PersistenceCrashPoint::AfterFsyncBeforeReturn,
        );

        let recovery_b = wal2.recover();
        // Data SHOULD be recovered (fsynced)
        assert_eq!(recovery_b.recovered_operations.len(), 1);

        println!("✅ Gate 7: Fsync safety verified");
    }

    #[test]
    fn test_phase6_gate7_corruption_detection() {
        // Gate 7: Verify corruption is detected, not silently returned
        let mut wal = DurablePersistenceLayer::new();

        wal.write_with_crash(
            "op1".to_string(),
            vec![1, 2, 3],
            PersistenceCrashPoint::AfterWriteReturns,
        );
        wal.write_with_crash(
            "op2".to_string(),
            vec![4, 5, 6],
            PersistenceCrashPoint::AfterWriteReturns,
        );

        // Corrupt the second entry
        wal.corrupt_tail(1);

        let recovery = wal.recover();

        // Corrupted entry should be detected
        assert!(recovery.corrupted_entries.len() > 0);

        // Good entry should still be recovered
        assert!(recovery.recovered_operations.contains(&"op1".to_string()));

        // Corrupted entry must NOT be in recovered (not silently returned)
        assert!(!recovery.recovered_operations.contains(&"op2".to_string()));

        println!("✅ Gate 7: Corruption detection verified");
    }

    #[test]
    fn test_phase6_gates_5_7_partition_with_persistence() {
        // Gates 5 + 7: Partition scenario with crash during reconciliation
        let mut wal = DurablePersistenceLayer::new();

        // Simulate partition A's operations
        wal.write_with_crash(
            "partition_a_op1".to_string(),
            vec![1, 1, 1],
            PersistenceCrashPoint::AfterWriteReturns,
        );

        // Simulate partition B's operations (arriving later)
        wal.write_with_crash(
            "partition_b_op1".to_string(),
            vec![2, 2, 2],
            PersistenceCrashPoint::AfterFsyncBeforeReturn,
        );

        // Crash during reconciliation
        wal.corrupt_tail(1);

        let recovery = wal.recover();

        // Partition A's operation should survive
        assert!(recovery
            .recovered_operations
            .contains(&"partition_a_op1".to_string()));

        // Partition B's operation was corrupted during crash
        // Should not silently appear in recovery
        assert!(!recovery
            .recovered_operations
            .contains(&"partition_b_op1".to_string()));

        println!("✅ Gates 5+7: Partition persistence verified");
    }

    #[test]
    fn test_phase6_gates_4_7_causal_ordering_survives_crash() {
        // Gates 4 + 7: Causal ordering must survive crash/recovery
        let mut barrier = CausalDependencyBarrier::new();

        // Op1 has no dependencies
        let op1_required = VectorClock::new();
        let result1 = barrier.receive_operation("op1".to_string(), op1_required);
        assert!(matches!(
            result1,
            crate::causal_dependency_barrier::ReceiveResult::ReadyToApply { .. }
        ));

        // Op2 requires {node: 1} to be satisfied
        let mut op2_required = VectorClock::new();
        op2_required.increment("node");

        let result2 = barrier.receive_operation("op2".to_string(), op2_required.clone());
        assert!(matches!(
            result2,
            crate::causal_dependency_barrier::ReceiveResult::BufferedWaitingForDependencies { .. }
        ));

        // Gate 7: Simulate persistence
        let mut wal = DurablePersistenceLayer::new();
        wal.write_with_crash(
            "op1".to_string(),
            vec![1],
            PersistenceCrashPoint::AfterWriteReturns,
        );

        let recovery = wal.recover();
        assert!(recovery.recovered_operations.contains(&"op1".to_string()));

        // After recovery, Op1 should be replayed, advancing frontier
        let mut barrier_recovered = CausalDependencyBarrier::new();
        barrier_recovered.receive_operation("op1".to_string(), VectorClock::new());

        let mut op1_clock = VectorClock::new();
        op1_clock.increment("node");
        barrier_recovered.apply_operation("op1", &op1_clock);

        // Now Op2 should unblock
        let result_op2 = barrier_recovered.receive_operation("op2".to_string(), op2_required);

        // Op2 should now be ready (frontier is {node: 1}, op2 requires {node: 1})
        assert!(matches!(
            result_op2,
            crate::causal_dependency_barrier::ReceiveResult::ReadyToApply { .. }
        ));

        println!("✅ Gates 4+7: Causal ordering survives crash/recovery");
    }

    #[test]
    fn test_phase6_gates_3_7_exactly_once_with_persistence() {
        // Gates 3 + 7: Exactly-once must hold across crash boundary
        use crate::crash_atomic_boundary::{CrashAtomicExecutor, CrashBoundaryPoint};

        let mut executor = CrashAtomicExecutor::new();

        // Apply operation across crash boundary
        let crash_state = executor.apply_with_simulated_crash(
            "exactly_once_op",
            "rec1",
            42,
            CrashBoundaryPoint::AfterDedupPersist,
        );

        // Gate 3: Verify invariant holds
        let invariant = executor.verify_invariant(&crash_state);
        assert!(matches!(
            invariant,
            crate::crash_atomic_boundary::InvariantCheckResult::Satisfied
        ));

        // Gate 7: Simulate persistence of the dedup entry
        let mut wal = DurablePersistenceLayer::new();
        wal.write_with_crash(
            "exactly_once_op".to_string(),
            vec![42],
            PersistenceCrashPoint::AfterWriteReturns,
        );

        let recovery = wal.recover();
        assert!(recovery
            .recovered_operations
            .contains(&"exactly_once_op".to_string()));

        // Critical: After recovery, duplicate application must be prevented
        let result = executor.apply_with_crash("exactly_once_op", "rec1", 99);
        assert!(result.is_err());
        assert!(result.unwrap_err().contains("already applied"));

        println!("✅ Gates 3+7: Exactly-once with persistence verified");
    }

    #[test]
    fn test_phase6_gates_6_7_consistency_contract_survives_crash() {
        // Gates 6 + 7: Consistency contracts must not be silently downgraded by crashes
        let mut enforcer = ConsistencyContractEnforcer::new();

        // Gate 6: Record commutative operation
        let result = enforcer.verify_commutative("Increment", &[5, 10, 15]);
        assert!(matches!(
            result,
            CommutativeVerificationResult::Satisfied { .. }
        ));

        // Gate 7: Simulate persistence
        let mut wal = DurablePersistenceLayer::new();

        wal.write_with_crash(
            "increment_op1".to_string(),
            vec![5],
            PersistenceCrashPoint::AfterWriteReturns,
        );
        wal.write_with_crash(
            "increment_op2".to_string(),
            vec![10],
            PersistenceCrashPoint::AfterWriteReturns,
        );
        wal.write_with_crash(
            "increment_op3".to_string(),
            vec![15],
            PersistenceCrashPoint::AfterWriteReturns,
        );

        let recovery = wal.recover();
        assert_eq!(recovery.recovered_operations.len(), 3);

        // After recovery, consistency contract must STILL apply
        let mut enforcer_recovered = ConsistencyContractEnforcer::new();
        let result_recovered = enforcer_recovered.verify_commutative("Increment", &[5, 10, 15]);

        assert!(matches!(
            result_recovered,
            CommutativeVerificationResult::Satisfied { .. }
        ));

        println!("✅ Gates 6+7: Consistency contracts survive crash/recovery");
    }

    #[test]
    fn test_phase6_gate7_arbitr_truncation_recovery() {
        // Gate 7: Recovery must handle arbitrary byte truncation
        let mut wal = DurablePersistenceLayer::new();

        // Write multiple operations
        for i in 1..=5 {
            wal.write_with_crash(
                format!("op{}", i),
                vec![i as u8, i as u8, i as u8],
                PersistenceCrashPoint::AfterWriteReturns,
            );
        }

        assert_eq!(wal.get_log().len(), 5);

        // Truncate at arbitrary byte offset
        wal.truncate_at_byte(8);

        let recovery = wal.recover();

        // Should recover what's still valid
        // The key property: recovery should NOT panic and should be deterministic
        println!(
            "  Recovered {} operations after arbitrary truncation",
            recovery.recovered_operations.len()
        );

        // Truncate again at different point
        let mut wal2 = DurablePersistenceLayer::new();
        for i in 1..=5 {
            wal2.write_with_crash(
                format!("op{}", i),
                vec![i as u8, i as u8, i as u8],
                PersistenceCrashPoint::AfterWriteReturns,
            );
        }

        wal2.truncate_at_byte(12);
        let recovery2 = wal2.recover();

        // Both truncations should produce deterministic results
        println!(
            "  Recovered {} operations at different truncation point",
            recovery2.recovered_operations.len()
        );

        println!("✅ Gate 7: Arbitrary truncation recovery verified");
    }

    #[test]
    fn test_phase6_gate7_crash_point_state_machine() {
        // Gate 7: Verify all crash points produce valid recovery states
        use crate::persistence_reality::PersistenceCrashPoint;

        let crash_points = vec![
            ("BeforeWrite", PersistenceCrashPoint::BeforeWrite),
            (
                "AfterBufferBeforeFsync",
                PersistenceCrashPoint::AfterBufferBeforeFsync,
            ),
            ("DuringFsync", PersistenceCrashPoint::DuringFsync),
            (
                "AfterFsyncBeforeReturn",
                PersistenceCrashPoint::AfterFsyncBeforeReturn,
            ),
            (
                "AfterWriteReturns",
                PersistenceCrashPoint::AfterWriteReturns,
            ),
        ];

        let mut verifier = PersistenceContractVerifier::new();

        for (name, crash_point) in crash_points {
            let mut wal = DurablePersistenceLayer::new();
            let result = wal.write_with_crash(format!("op_{}", name), vec![1, 2, 3], crash_point);

            let recovery = wal.recover();

            // Only count as "written" if actually fsynced
            let was_fsynced = matches!(
                result,
                crate::persistence_reality::WriteResult::Success { .. }
            ) || matches!(
                result,
                crate::persistence_reality::WriteResult::Crashed { fsynced: true, .. }
            );

            let written_count = if was_fsynced { 1 } else { 0 };
            let recovered_count = recovery.recovered_operations.len();

            let written_ops = if was_fsynced {
                vec![format!("op_{}", name)]
            } else {
                vec![]
            };

            verifier.record_scenario(
                format!("crash_point_{}", name),
                written_ops,
                recovery.recovered_operations.clone(),
                recovery.corrupted_entries.len(),
            );

            println!(
                "  {} → written(durable): {}, recovered: {}",
                name, written_count, recovered_count
            );
        }

        // Verify contracts
        let fsync_result = verifier.verify_fsync_durability();
        assert!(matches!(
            fsync_result,
            crate::persistence_reality::FsyncDurabilityResult::Satisfied { .. }
        ));

        let corruption_result = verifier.verify_corruption_detection();
        assert!(matches!(
            corruption_result,
            crate::persistence_reality::CorruptionDetectionResult::Satisfied { .. }
        ));

        println!("✅ Gate 7: All crash points produce valid states");
    }

    #[test]
    fn test_phase6_integration_all_gates_with_persistence() {
        // Comprehensive: All Phase 5 gates + Phase 6 Gate 7
        // Simulate: convergence → persistence → crash → recovery → re-convergence

        println!("=== Phase 6 Integration: All Gates + Persistence ===\n");

        // Phase 5 Gate 2: Algebra
        let registry = crate::operation_algebra::OperationAlgebraRegistry::new();
        let increment = registry.get("Increment").unwrap();
        assert!(increment.is_commutative());
        println!("  ✓ Gate 2 (Algebra): Increment is commutative");

        // Phase 6 Gate 7: Persistence
        let mut wal = DurablePersistenceLayer::new();
        wal.write_with_crash(
            "op1".to_string(),
            vec![1, 2, 3],
            PersistenceCrashPoint::AfterWriteReturns,
        );
        let recovery = wal.recover();
        assert!(recovery.recovered_operations.contains(&"op1".to_string()));
        println!("  ✓ Gate 7 (Persistence): Operation persisted and recovered");

        // Phase 5 Gate 4: Causal ordering
        let mut barrier = CausalDependencyBarrier::new();
        let op1 = VectorClock::new();
        barrier.receive_operation("op1".to_string(), op1);
        let mut op1_clock = VectorClock::new();
        op1_clock.increment("node");
        barrier.apply_operation("op1", &op1_clock);
        println!("  ✓ Gate 4 (Causality): Operations ordered correctly");

        // Phase 5 Gate 3: Exactly-once
        let mut enforcer = crate::crash_atomic_boundary::CrashAtomicExecutor::new();
        let _result = enforcer.apply_with_crash("op1", "rec1", 42);
        println!("  ✓ Gate 3 (Exactly-Once): Operation applied atomically");

        // Phase 5 Gate 6: Consistency contract
        let mut contract = ConsistencyContractEnforcer::new();
        let verify = contract.verify_commutative("Increment", &[1, 2, 3]);
        assert!(matches!(
            verify,
            CommutativeVerificationResult::Satisfied { .. }
        ));
        println!("  ✓ Gate 6 (Contract): Commutative semantics verified");

        println!("\n✅ Phase 6 Integration Complete: All gates + persistence");
    }
}