neoengram 0.1.0

Version control for AI models and large datasets
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99

use std::{fs, path::Path};

use anyhow::{bail, ensure, Context, Result};

use crate::{
    commands::{checkout, rm},
    repository::Repository,
};

pub(crate) async fn execute(abort: bool) -> Result<()> {
    let current_dir = std::env::current_dir().context("无法确定当前工作目录")?;
    let repository = Repository::discover(&current_dir)?;
    let outcome = tokio::task::spawn_blocking(move || recover_repository(&repository, abort))
        .await
        .context("Recover 任务异常终止")??;

    match outcome {
        RecoveryReport::Nothing => println!("No unfinished transaction"),
        RecoveryReport::Checkout { target_id, aborted } => {
            if aborted {
                println!("Aborted checkout transaction for {target_id}");
            } else {
                println!("Recovered checkout transaction to {target_id}");
            }
        }
        RecoveryReport::Removal { id, aborted } => {
            if aborted {
                println!("Aborted rm transaction {id}");
            } else {
                println!("Recovered rm transaction {id}");
            }
        }
    }
    Ok(())
}

enum RecoveryReport {
    Nothing,
    Checkout { target_id: String, aborted: bool },
    Removal { id: String, aborted: bool },
}

fn recover_repository(repository: &Repository, abort: bool) -> Result<RecoveryReport> {
    // recovery lock 不会因 transactions 非空而拒绝；它仍与 add/commit/checkout 使用
    // 同一 OS advisory lock，因此恢复期间不会有另一个 NeoEngram 写命令交错执行。
    let mut lock = repository.acquire_recovery_lock()?;
    let transactions = repository.unfinished_transactions()?;
    if transactions.is_empty() {
        return Ok(RecoveryReport::Nothing);
    }
    ensure!(
        transactions.len() == 1,
        "检测到多个未完成事务，无法安全推断恢复顺序: {}",
        transactions
            .iter()
            .map(|path| path.display().to_string())
            .collect::<Vec<_>>()
            .join(", ")
    );

    let transaction = transactions.first().context("内部错误：事务列表为空")?;
    let operation = read_operation(transaction)?;
    match operation.as_str() {
        "checkout" => {
            let outcome = checkout::recover_transaction(repository, transaction, abort, &mut lock)?;
            Ok(RecoveryReport::Checkout {
                target_id: outcome.target_id,
                aborted: outcome.aborted,
            })
        }
        "rm" => {
            let id = transaction
                .file_name()
                .and_then(|name| name.to_str())
                .context("rm 事务目录名不是有效 UTF-8")?
                .to_owned();
            lock.set_operation("rm-recover", Some(&id))?;
            rm::recover_transaction(repository, transaction, abort)?;
            lock.set_operation("rm-recover", None)?;
            Ok(RecoveryReport::Removal { id, aborted: abort })
        }
        other => bail!("不支持的事务类型 {other:?}: {}", transaction.display()),
    }
}

fn read_operation(transaction: &Path) -> Result<String> {
    let path = transaction.join("OPERATION");
    let metadata = fs::symlink_metadata(&path)
        .with_context(|| format!("无法检查事务类型文件: {}", path.display()))?;
    ensure!(
        metadata.is_file() && !metadata.file_type().is_symlink(),
        "事务类型路径不是普通文件: {}",
        path.display()
    );
    let operation = fs::read_to_string(&path)
        .with_context(|| format!("无法读取事务类型: {}", path.display()))?;
    ensure!(!operation.trim().is_empty(), "事务类型不能为空");
    Ok(operation.trim().to_owned())
}