wdev 0.1.4

//! sync 持久化契约：全局 sync 覆盖他线程写入、跨线程 fsync 可行性实验、
//! 删段/截断免责与守护位图清空。
//!
//! 对标 libs/storage/Tsavorite/cs/test/test.hlog/DeviceTests.cs:LocalStorageDevice：句柄表进程级共享（`SafeConcurrentDictionary<int,
//! SafeFileHandle>` 按段键控）、任意线程可对全设备 sync。Rust 侧可行性依赖
//! `compio-driver/sync` feature（SharedFd 为 `Arc`，fd 无线程亲和，fsync 按 inode 全量生效）。
//!
//! 所有跨线程场景均挂看门狗（超时强制退出进程），防运行时互等挂死。

use std::{
  fs,
  sync::{Arc, mpsc},
  thread,
};

use aok::{OK, Void};
use compio::{buf::BufResult, fs::File, io::AsyncWriteAt, runtime::Runtime};
use log::info;
use tempfile::tempdir;
use wbase::AlignedBuf;
use wdev::{Device, SegmentedDevice};

use crate::support::Watchdog;

const SECTOR: usize = 4096;
const SEG_SIZE: u64 = 64 * 1024;

/// 跨线程 fsync 可行性测试：线程 A 在其独立 Runtime 写入数据，线程 B 在
/// 自己的独立 Runtime 对同一路径打开并执行 sync_all / sync_data。
/// 验证 POSIX 文件系统按 inode 脏页落盘契约与多 Runtime 协同可靠性。
#[test]
fn raw_cross_thread_fsync_feasibility() -> Void {
  let dir = tempdir()?;
  let path = dir.path().join("raw_ct_fsync.bin");
  let _wd = Watchdog::start(60);

  let path_a = path.clone();
  let path_b = path.clone();
  let (tx, rx) = mpsc::channel::<()>();

  // 线程 A：打开文件并写入 4096 字节
  let ta = thread::spawn(move || -> Void {
    let rt = Runtime::new()?;
    rt.block_on(async {
      let mut file = File::create(&path_a).await?;
      let data: Vec<u8> = (0..SECTOR).map(|j| (j & 0xFF) as u8).collect();
      let buf = AlignedBuf::from_slice(&data, 4096)?;
      let BufResult(res, _) = file.write_at(buf, 0).await;
      assert_eq!(res?, SECTOR);
      tx.send(()).expect("接收端存活");
      aok::Result::<()>::Ok(())
    })?;
    OK
  });

  // 线程 B：独立 Runtime，对同一路径文件执行全量与数据级 fsync
  let tb = thread::spawn(move || -> Void {
    rx.recv().expect("发送端存活");
    let rt = Runtime::new()?;
    rt.block_on(async {
      let file = File::open(&path_b).await?;
      file.sync_all().await?;
      file.sync_data().await?;
      aok::Result::<()>::Ok(())
    })?;
    OK
  });

  ta.join().unwrap()?;
  tb.join().unwrap()?;

  // 常规 std 读路径复验数据完整
  let disk = fs::read(&path)?;
  assert_eq!(disk.len(), SECTOR);
  assert_eq!(disk[0], 0);
  assert_eq!(disk[SECTOR - 1], ((SECTOR - 1) & 0xFF) as u8);

  info!("跨线程 fsync 可行性实验通过：他线程 fd 的 sync_all/sync_data 全部成功");
  OK
}

/// 全局 sync 覆盖他线程写入：线程 A 写段 0 与段 1 后退出（不再 sync），线程 B
/// 调用 `sync()` 必须覆盖 A 的全部在表句柄——守护位图清空（debug 构建），且
/// 全新设备实例（全新句柄）能读回全部字节，证明数据已过 fsync 落盘。
#[test]
fn global_sync_covers_foreign_thread_writes() -> Void {
  let dir = tempdir()?;
  let path = dir.path().join("global_sync.log");
  let device = Arc::new(SegmentedDevice::segmented(&path, SEG_SIZE)?);
  let _wd = Watchdog::start(60);

  // 线程 A：写段 0 首块与段 1 首块（跨段覆盖守护窗口内两个段），写完即退出
  let dev_a = Arc::clone(&device);
  let ta = thread::spawn(move || -> Void {
    let rt = Runtime::new()?;
    rt.block_on(async move {
      for seg in 0u32..2 {
        let data: Vec<u8> = (0..SECTOR).map(|j| (j ^ seg as usize) as u8).collect();
        let wbuf = AlignedBuf::from_slice(&data, 4096)?;
        let offset = u64::from(seg) * SEG_SIZE;
        let (res, _) = dev_a.write_aligned(offset, wbuf).await;
        assert_eq!(res?, SECTOR);
      }
      aok::Result::<()>::Ok(())
    })?;
    OK
  });
  ta.join().unwrap()?;

  // 写入已记账、尚未 sync：守护位图应含段 0 与段 1
  #[cfg(debug_assertions)]
  assert_eq!(device.debug_dirty_segments(), vec![0, 1]);

  // 线程 B：独立 Runtime，全局 sync（覆盖线程 A 的句柄）+ 全新设备实例读回复验
  let dev_b = Arc::clone(&device);
  let tb = thread::spawn(move || -> Void {
    let rt = Runtime::new()?;
    rt.block_on(async move {
      dev_b.sync().await?;

      // 全新设备实例（全新句柄）读回：字节已在文件上
      let fresh = SegmentedDevice::segmented(&path, SEG_SIZE)?;
      for seg in 0u32..2 {
        let expected: Vec<u8> = (0..SECTOR).map(|j| (j ^ seg as usize) as u8).collect();
        let check = AlignedBuf::new(SECTOR, 4096)?;
        let (res, check) = fresh.read_aligned(u64::from(seg) * SEG_SIZE, check).await;
        assert_eq!(res?, SECTOR);
        assert_eq!(check.as_slice(), &expected[..], "段 {seg} 落盘字节不匹配");
      }
      aok::Result::<()>::Ok(())
    })?;
    OK
  });
  tb.join().unwrap()?;

  // sync 覆盖后守护位图清空（他线程写入被全局 sync 背书）
  #[cfg(debug_assertions)]
  assert!(
    device.debug_dirty_segments().is_empty(),
    "全局 sync 后仍有在册脏段"
  );

  info!("全局 sync 覆盖他线程写入并通过全新句柄读回复验");
  OK
}

/// 全局 sync_data 与 sync 同覆盖口径：线程 A 写入后退出，线程 B 仅 fdatasync，
/// 守护位图同样必须清空（数据级 fsync 即构成持久化背书）。
#[test]
fn global_sync_data_covers_foreign_thread_writes() -> Void {
  let dir = tempdir()?;
  let path = dir.path().join("global_sync_data.log");
  let device = Arc::new(SegmentedDevice::segmented(&path, SEG_SIZE)?);
  let _wd = Watchdog::start(60);

  let dev_a = Arc::clone(&device);
  let ta = thread::spawn(move || -> Void {
    let rt = Runtime::new()?;
    rt.block_on(async move {
      let data: Vec<u8> = (0..SECTOR).map(|j| (j | 0x80) as u8).collect();
      let wbuf = AlignedBuf::from_slice(&data, 4096)?;
      let (res, _) = dev_a.write_aligned(0, wbuf).await;
      assert_eq!(res?, SECTOR);
      aok::Result::<()>::Ok(())
    })?;
    OK
  });
  ta.join().unwrap()?;

  let dev_b = Arc::clone(&device);
  let tb = thread::spawn(move || -> Void {
    let rt = Runtime::new()?;
    rt.block_on(async move { dev_b.sync_data().await })?;
    OK
  });
  tb.join().unwrap()?;

  #[cfg(debug_assertions)]
  assert!(device.debug_dirty_segments().is_empty());

  info!("全局 sync_data 与 sync 同覆盖口径");
  OK
}

/// 删段与截断免责：写入后显式 `remove_segment` / `truncate_until_segment` 的段
/// 无须 fsync 背书，随后的全局 sync 不得触发守护违约断言，位图清空。
#[test]
fn sync_contract_remove_and_truncate_immunity() -> Void {
  let dir = tempdir()?;
  let path = dir.path().join("immunity.log");
  let device = Arc::new(SegmentedDevice::segmented(&path, SEG_SIZE)?);
  let _wd = Watchdog::start(60);

  let rt = Runtime::new()?;
  rt.block_on(async {
    // 写段 0 与段 1
    for seg in 0u32..2 {
      let wbuf = AlignedBuf::from_slice(&vec![0xA5; SECTOR], 4096)?;
      let (res, _) = device.write_aligned(u64::from(seg) * SEG_SIZE, wbuf).await;
      assert_eq!(res?, SECTOR);
    }
    #[cfg(debug_assertions)]
    assert_eq!(device.debug_dirty_segments(), vec![0, 1]);

    // 显式删段 0：免责清位
    device.remove_segment(0).await?;
    #[cfg(debug_assertions)]
    assert_eq!(device.debug_dirty_segments(), vec![1]);

    // 截断至段 1：仅删除段 0 之前的段，段 1 仍有效，脏位保持待 sync 背书
    device.truncate_until_segment(1).await?;
    #[cfg(debug_assertions)]
    assert_eq!(device.debug_dirty_segments(), vec![1]);

    // 免责后 sync：不得触发守护违约断言
    device.sync().await?;

    // 免责段物理消失，start_segment 单调推进
    assert_eq!(device.start_segment(), 1);
    assert_eq!(device.get_file_size(0)?, 0);

    aok::Result::<()>::Ok(())
  })?;

  info!("删段与截断免责路径通过，sync 无守护违约");
  OK
}

/// 多段并发 sync 与 sync_data 压榨：跨多个段并发写入后执行全局并发刷盘，验证快速路径与数据落盘
#[test]
fn multi_segment_concurrent_sync() -> Void {
  let dir = tempdir()?;
  let path = dir.path().join("multi_seg_concurrent_sync.log");
  let device = Arc::new(SegmentedDevice::segmented(&path, SEG_SIZE)?);
  let _wd = Watchdog::start(60);

  let rt = Runtime::new()?;
  rt.block_on(async {
    // 1. 快速路径 1：空段直接返回 Ok
    device.sync().await?;
    device.sync_data().await?;

    // 2. 快速路径 2：单段场景
    let data0 = vec![0x42u8; SECTOR];
    let wbuf0 = AlignedBuf::from_slice(&data0, 4096)?;
    let (res, _) = device.write_aligned(0, wbuf0).await;
    assert_eq!(res?, SECTOR);
    device.sync().await?;
    #[cfg(debug_assertions)]
    assert!(device.debug_dirty_segments().is_empty());

    // 3. 多段并发场景（8段并发下发）
    const SEGS: u32 = 8;
    for seg in 0..SEGS {
      let data = vec![(seg & 0xFF) as u8; SECTOR];
      let wbuf = AlignedBuf::from_slice(&data, 4096)?;
      let (res, _) = device.write_aligned(u64::from(seg) * SEG_SIZE, wbuf).await;
      assert_eq!(res?, SECTOR);
    }

    #[cfg(debug_assertions)]
    assert_eq!(device.debug_dirty_segments().len(), SEGS as usize);

    // 并发 sync_data 与 sync 全部段
    device.sync_data().await?;
    device.sync().await?;

    #[cfg(debug_assertions)]
    assert!(device.debug_dirty_segments().is_empty());

    // 零分配验证所有段数据落盘
    for seg in 0..SEGS {
      let check = AlignedBuf::new(SECTOR, 4096)?;
      let (res, check) = device.read_aligned(u64::from(seg) * SEG_SIZE, check).await;
      assert_eq!(res?, SECTOR);
      let expected_byte = (seg & 0xFF) as u8;
      assert!(
        check.as_slice().iter().all(|&b| b == expected_byte),
        "段 {seg} 落盘数据不匹配"
      );
    }

    aok::Result::<()>::Ok(())
  })?;

  info!("多段并发 sync / sync_data 压榨测试通过");
  OK
}

/// sync 补开路径的空洞段防御：写入段 0 与段 2（段 1 从未写入）后执行全局 sync，
/// 空洞段 1 绝不因 sync 被幽灵创建（Rust 补齐语义：C# IDevice 无设备级 sync，
/// 补开仅针对磁盘上已存在的段文件——他线程写入过的段必然已创建文件）；
/// 已写入段数据经 sync 落盘可回读。
#[test]
fn sync_does_not_ghost_create_hole_segments() -> Void {
  let dir = tempdir()?;
  let path = dir.path().join("hole_sync.log");
  let device = Arc::new(SegmentedDevice::segmented(&path, SEG_SIZE)?);
  let _wd = Watchdog::start(60);

  let rt = Runtime::new()?;
  rt.block_on(async {
    // 写段 0 与段 2，段 1 保持空洞
    for seg in [0u32, 2] {
      let data = vec![(seg * 37 + 3) as u8; SECTOR];
      let wbuf = AlignedBuf::from_slice(&data, 4096)?;
      let (res, _) = device.write_aligned(u64::from(seg) * SEG_SIZE, wbuf).await;
      assert_eq!(res?, SECTOR);
    }
    assert!(
      !device.segment_path(1).exists(),
      "写入阶段空洞段 1 不应存在"
    );

    // 全局 sync：覆盖段 0、2，且绝不为空洞段 1 创建幽灵文件
    device.sync().await?;
    device.sync_data().await?;
    assert!(
      !device.segment_path(1).exists(),
      "sync 绝不能为从未写入的空洞段创建幽灵文件"
    );

    // 已写入段数据落盘完好（全新设备实例复验，排除页缓存直读）
    let fresh = SegmentedDevice::segmented(&path, SEG_SIZE)?;
    for seg in [0u32, 2] {
      let expected = vec![(seg * 37 + 3) as u8; SECTOR];
      let check = AlignedBuf::new(SECTOR, 4096)?;
      let (res, check) = fresh.read_aligned(u64::from(seg) * SEG_SIZE, check).await;
      assert_eq!(res?, SECTOR);
      assert_eq!(check.as_slice(), &expected[..], "段 {seg} 落盘数据不匹配");
    }

    aok::Result::<()>::Ok(())
  })?;

  info!("sync 空洞段幽灵创建防御与在册段覆盖校验通过");
  OK
}