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
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
//! Orphan group-file garbage collection after checkpoints.
//!
//! File-level cleanup only: removes group files whose group no longer exists.
//! Data-level orphan convergence (authority or property rows without topology)
//! lives in `core::owner::rebuild_owner_map_with_stats` and the load-time
//! `copy_audit` refusal. The two must never be confused: this module counts
//! deleted files, the audit counts converged or refused data rows.
use super::super::core::EdgeStore;
use super::layout::parse_group_file;
use std::collections::HashSet;
use std::path::Path;
impl EdgeStore {
/// Delete every residual file of one owner group and count removals.
///
/// Invalid-group recovery after a torn epoch: the manifest lists the
/// group but a companion file is missing or corrupt, so no cross-epoch
/// stitching is attempted. Residual base, sidecar, timestamp, property
/// and snapshot files are removed and the group is rebuilt empty by the
/// caller, leaving the table writable.
pub(crate) fn remove_invalid_group_files(&self, dir: &Path, gid: u32) -> usize {
use super::layout::{
in_append_path, in_group_path, out_append_path, out_group_path, props_group_path,
ts_group_path,
};
use super::snapshot::snapshot_path_for;
let mut removed = 0usize;
let candidates = [
out_group_path(dir, gid as usize),
in_group_path(dir, gid as usize),
out_append_path(dir, gid as usize),
in_append_path(dir, gid as usize),
ts_group_path(dir, gid),
props_group_path(dir, gid),
];
for path in candidates {
if path.exists() && std::fs::remove_file(&path).is_ok() {
removed += 1;
}
let snapshot = snapshot_path_for(&path);
if snapshot.exists() && std::fs::remove_file(&snapshot).is_ok() {
removed += 1;
}
}
if removed > 0 {
log::warn!(
"EdgeTable[{}] cleaned residual files of invalid group {}: {} files",
self.label,
gid,
removed
);
}
removed
}
/// Remove group files whose group no longer exists, returning the removed
/// count for observability.
///
/// Timestamp and property shards follow the owner direction's groups, so
/// orphan shard convergence is visible here as a count instead of silent
/// deletion. Snapshot sidecars track their base group and drop with it.
pub(crate) fn remove_orphan_group_files(&self, dir: &Path) -> usize {
let out_existing: HashSet<u32> = self
.out_csr
.existing_group_ids()
.into_iter()
.map(|gid| gid as u32)
.collect();
let in_existing: HashSet<u32> = self
.in_csr
.existing_group_ids()
.into_iter()
.map(|gid| gid as u32)
.collect();
let owner_existing: HashSet<u32> =
if self.schema.oe_strategy != crate::edge::EdgeStrategy::None {
out_existing.clone()
} else {
in_existing.clone()
};
let mut removed = 0usize;
let mut remove = |path: &std::path::Path| {
if std::fs::remove_file(path).is_ok() {
removed += 1;
}
};
if let Ok(entries) = std::fs::read_dir(dir) {
for entry in entries.flatten() {
let name = entry.file_name().to_string_lossy().to_string();
let path = entry.path();
if name.starts_with("out_g") && name.ends_with(".bin") && !name.contains(".append")
{
if let Some(gid) = parse_group_file(&name, "out_g") {
if !out_existing.contains(&gid) {
remove(&path);
}
}
} else if name.starts_with("in_g")
&& name.ends_with(".bin")
&& !name.contains(".append")
{
if let Some(gid) = parse_group_file(&name, "in_g") {
if !in_existing.contains(&gid) {
remove(&path);
}
}
} else if name.starts_with("out_g") && name.contains(".append") {
if let Some(gid) = parse_group_file(&name, "out_g") {
if !out_existing.contains(&gid) {
remove(&path);
}
}
} else if name.starts_with("in_g") && name.contains(".append") {
if let Some(gid) = parse_group_file(&name, "in_g") {
if !in_existing.contains(&gid) {
remove(&path);
}
}
} else if name.ends_with(".snapshot") {
// Snapshot sidecars track their base group: drop the cache
// when the group itself is gone.
let base = name.strip_suffix(".snapshot").unwrap_or("");
if base.starts_with("out_g") {
if let Some(gid) = parse_group_file(base, "out_g") {
if !out_existing.contains(&gid) {
remove(&path);
}
}
} else if base.starts_with("in_g") {
if let Some(gid) = parse_group_file(base, "in_g") {
if !in_existing.contains(&gid) {
remove(&path);
}
}
}
} else if name.starts_with("ts_g") && name.ends_with(".bin") {
if let Some(gid) = parse_group_file(&name, "ts_g") {
if !owner_existing.contains(&gid) {
remove(&path);
}
}
} else if name.starts_with("props_g") && name.ends_with(".bin") {
if let Some(gid) = parse_group_file(&name, "props_g") {
if !owner_existing.contains(&gid) {
remove(&path);
}
}
}
}
}
if removed > 0 {
log::debug!(
"remove_orphan_group_files: removed {} orphan files",
removed
);
}
removed
}
}