linkrs_migration/
backup.rs

1//! Pre-migration backup for destructive steps.
2//!
3//! `DropColumn` has no inverse step, so a plan containing drops can only be
4//! rolled back from a backup taken before execution. When the caller supplies
5//! a backup directory, the executor snapshots the affected rows first and
6//! rollback restores from that snapshot instead of failing outright.
7
8use std::path::Path;
9
10use linkrs_core::error::storage::StorageErrorKind;
11use linkrs_core::{Edge, Vertex};
12use linkrs_storage::{StorageReader, StorageSchemaOps, StorageWriter};
13use serde::{Deserialize, Serialize};
14
15use crate::error::MigrationError;
16use crate::plan::{push_escaped_path_component, MigrationPlan, MigrationReport, MigrationStep};
17
18#[derive(Debug, Clone, Serialize, Deserialize)]
19pub(crate) struct DestructiveBackup {
20    pub space: String,
21    pub label: String,
22    pub is_edge: bool,
23    pub plan_hash: String,
24    pub dropped_columns: Vec<String>,
25    pub dropped_label: Option<String>,
26    pub dropped_edge_type: Option<String>,
27    pub vertices: Vec<Vertex>,
28    pub edges: Vec<Edge>,
29}
30
31fn backup_file_name(plan: &MigrationPlan) -> String {
32    let hash = if plan.plan_hash.is_empty() {
33        plan.compute_hash()
34    } else {
35        plan.plan_hash.clone()
36    };
37    let mut name = String::from("backup_");
38    push_escaped_path_component(&mut name, &plan.target.space);
39    name.push('-');
40    push_escaped_path_component(&mut name, &plan.target.label);
41    name.push('-');
42    push_escaped_path_component(&mut name, &hash);
43    name.push_str(".json");
44    name
45}
46
47fn backup_path(plan: &MigrationPlan, dir: &Path) -> std::path::PathBuf {
48    dir.join(backup_file_name(plan))
49}
50
51/// Snapshot rows touched by destructive steps. Returns the row count.
52pub fn write_backup<S>(
53    storage: &S,
54    plan: &MigrationPlan,
55    dir: &Path,
56) -> Result<usize, MigrationError>
57where
58    S: StorageReader + ?Sized,
59{
60    std::fs::create_dir_all(dir)
61        .map_err(|e| MigrationError::Checkpoint(format!("cannot create backup directory: {e}")))?;
62    let mut backup = DestructiveBackup {
63        space: plan.target.space.clone(),
64        label: plan.target.label.clone(),
65        is_edge: plan.target.is_edge,
66        plan_hash: if plan.plan_hash.is_empty() {
67            plan.compute_hash()
68        } else {
69            plan.plan_hash.clone()
70        },
71        dropped_columns: plan.dropped_columns(),
72        dropped_label: None,
73        dropped_edge_type: None,
74        vertices: Vec::new(),
75        edges: Vec::new(),
76    };
77    for step in &plan.steps {
78        match step {
79            MigrationStep::DropLabel { label_name } => {
80                backup.dropped_label = Some(label_name.clone());
81            }
82            MigrationStep::DropEdgeType { edge_type_name } => {
83                backup.dropped_edge_type = Some(edge_type_name.clone());
84            }
85            _ => {}
86        }
87    }
88    if plan.target.is_edge {
89        backup.edges = storage
90            .scan_edges_by_type(&plan.target.space, &plan.target.label)
91            .map_err(MigrationError::from)?;
92    } else {
93        backup.vertices = storage
94            .scan_vertices_by_tag(&plan.target.space, &plan.target.label)
95            .map_err(MigrationError::from)?;
96    }
97    let rows = backup.vertices.len() + backup.edges.len();
98    let content = serde_json::to_string_pretty(&backup)
99        .map_err(|e| MigrationError::Checkpoint(format!("serialize backup: {e}")))?;
100    std::fs::write(backup_path(plan, dir), content)
101        .map_err(|e| MigrationError::Checkpoint(format!("write backup: {e}")))?;
102    Ok(rows)
103}
104
105fn load_backup(
106    plan: &MigrationPlan,
107    dir: &Path,
108) -> Result<Option<DestructiveBackup>, MigrationError> {
109    let path = backup_path(plan, dir);
110    if !path.exists() {
111        return Ok(None);
112    }
113    let content = std::fs::read_to_string(&path)
114        .map_err(|e| MigrationError::Checkpoint(format!("read backup: {e}")))?;
115    let backup: DestructiveBackup = serde_json::from_str(&content)
116        .map_err(|e| MigrationError::Checkpoint(format!("parse backup: {e}")))?;
117    if !plan.plan_hash.is_empty()
118        && !backup.plan_hash.is_empty()
119        && backup.plan_hash != plan.plan_hash
120    {
121        return Err(MigrationError::Plan(format!(
122            "backup hash mismatch for {}/{}: backup {} != plan {}",
123            plan.target.space, plan.target.label, backup.plan_hash, plan.plan_hash
124        )));
125    }
126    Ok(Some(backup))
127}
128
129/// Restore a destructive plan from its backup. Returns `None` when no backup
130/// file exists for the plan.
131pub fn restore_backup<S>(
132    storage: &mut S,
133    plan: &MigrationPlan,
134    dir: &Path,
135) -> Result<Option<MigrationReport>, MigrationError>
136where
137    S: StorageReader + StorageWriter + StorageSchemaOps + ?Sized,
138{
139    let backup = match load_backup(plan, dir)? {
140        Some(b) => b,
141        None => return Ok(None),
142    };
143    let mut restored: u64 = 0;
144    if plan.target.is_edge {
145        if backup.dropped_edge_type.is_some() {
146            let info = linkrs_core::types::EdgeTypeInfo::new(plan.target.label.clone());
147            match storage.create_edge_type(&plan.target.space, &info) {
148                Ok(_) => {}
149                Err(e) if e.kind() == StorageErrorKind::AlreadyExists => {}
150                Err(e) => return Err(MigrationError::Storage(Box::new(e))),
151            }
152            for edge in &backup.edges {
153                match storage.insert_edge(&plan.target.space, edge.clone()) {
154                    Ok(_) => restored += 1,
155                    Err(e) => log::warn!("Backup restore skipped one edge: {e}"),
156                }
157            }
158        } else {
159            let current = storage
160                .scan_edges_by_type(&plan.target.space, &plan.target.label)
161                .map_err(MigrationError::from)?;
162            for saved in &backup.edges {
163                let Some(mut live) = current
164                    .iter()
165                    .find(|e| {
166                        e.src == saved.src && e.dst == saved.dst && e.ranking == saved.ranking
167                    })
168                    .cloned()
169                else {
170                    log::warn!(
171                        "Backup restore skipped missing edge ({:?}->{:?})",
172                        saved.src,
173                        saved.dst
174                    );
175                    continue;
176                };
177                for col in &backup.dropped_columns {
178                    if let Some(v) = saved.props.get(col.as_str()) {
179                        live.props.insert(col.as_str().into(), v.clone());
180                    }
181                }
182                storage
183                    .update_edge(&plan.target.space, live)
184                    .map_err(MigrationError::from)?;
185                restored += 1;
186            }
187        }
188    } else if backup.dropped_label.is_some() {
189        let tag = linkrs_core::types::TagInfo::new(plan.target.label.clone());
190        match storage.create_tag(&plan.target.space, &tag) {
191            Ok(_) => {}
192            Err(e) if e.kind() == StorageErrorKind::AlreadyExists => {}
193            Err(e) => return Err(MigrationError::Storage(Box::new(e))),
194        }
195        for vertex in &backup.vertices {
196            match storage.insert_vertex(&plan.target.space, vertex.clone()) {
197                Ok(_) => restored += 1,
198                Err(e) => log::warn!("Backup restore skipped vertex {}: {e}", vertex.vid),
199            }
200        }
201    } else {
202        let current = storage
203            .scan_vertices_by_tag(&plan.target.space, &plan.target.label)
204            .map_err(MigrationError::from)?;
205        for saved in &backup.vertices {
206            let Some(mut live) = current.iter().find(|v| v.vid == saved.vid).cloned() else {
207                log::warn!("Backup restore skipped missing vertex {}", saved.vid);
208                continue;
209            };
210            for col in &backup.dropped_columns {
211                if let Some(v) = saved.tag.properties.get(col.as_str()) {
212                    live.tag.properties.insert(col.as_str().into(), v.clone());
213                }
214            }
215            storage
216                .update_vertex(&plan.target.space, live)
217                .map_err(MigrationError::from)?;
218            restored += 1;
219        }
220    }
221    let completed: Vec<usize> = (0..plan.steps.len()).collect();
222    Ok(Some(MigrationReport {
223        success: true,
224        steps_completed: completed.len(),
225        rows_migrated: restored,
226        errors: vec![],
227        completed_step_indices: completed,
228    }))
229}
linkrs_migration/backup.rs

linkrs_migration/
backup.rs
