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
//! 哈希索引最新记录探针（Lookup 判活与 CAS 复核共用的地址解析路径）
use windex::CandidateAddresses;
use super::LogCompactor;
use crate::{
error::Result,
host::{CompactSession, CompactStore},
};
/// 最新记录探查结果
pub(super) struct LatestRecord {
/// 主日志中的有效逻辑地址
pub(super) main_addr: u64,
/// 索引槽位中的地址（若为 ReadCache 驻留则携带虚拟标记位）
pub(super) index_addr: u64,
/// 是否为墓碑记录
pub(super) is_tombstone: bool,
}
impl<S: CompactStore> LogCompactor<S> {
/// 探查指定键在 store 中的最新记录地址及墓碑状态
///
/// 1. 通过无堆分配栈列表 `lookup_candidates` 获取 Tag 匹配的所有候选槽位并降序排列；
/// 2. 剥离 ReadCache 虚拟标记位（若已落入 ReadCache，顺链跳跃获取底层主日志真实逻辑地址）；
/// 3. 若候选地址匹配调用方已知当前记录（`known_curr`），直接复用其键匹配与墓碑状态，零冗余 I/O；
/// 4. 清理已截断或失效的陈旧索引槽位引用；
/// 5. 返回首个匹配指定 Key 的最新记录信息（包含主日志真实地址与索引原槽位地址）。
pub(super) async fn find_latest_address(
&self,
session: &S::Session,
key: &[u8],
known_curr: Option<(u64, bool)>,
) -> Result<Option<LatestRecord>> {
// 索引探测持短守卫分段执行（共享内存结构访问口径对齐 wkv delete_raw_disk_slow）
let mut addrs = {
let _guard = session.enter_epoch();
self.store.index().lookup_candidates(key)
};
addrs.sort_descending();
let begin_addr = self.store.begin_address();
let mut found = None;
let mut duplicates = CandidateAddresses::new();
for &addr in addrs.iter() {
// 剥离 ReadCache 虚拟标记位
let main_addr = if self.store.is_read_cache_addr(addr) {
self.store.skip_read_cache(addr)
} else {
addr
};
// 地址为 0（已滑出 ReadCache 窗口）或已落入截断区，标记为失效陈旧槽位
if main_addr == 0 || main_addr < begin_addr {
duplicates.push(addr);
continue;
}
// 若当前候选地址与调用方已知记录地址一致，直接复用已校验结果，避免冗余 I/O
if let Some((known_addr, known_tombstone)) = known_curr
&& main_addr == known_addr
{
if found.is_none() {
found = Some(LatestRecord {
main_addr,
index_addr: addr,
is_tombstone: known_tombstone,
});
} else {
duplicates.push(addr);
}
continue;
}
// 纪元纪律（对齐 read_from_disk / delete_raw_disk_slow 冷读协议）：磁盘区候选
// 免纪元纯设备读，绝不持守卫跨越磁盘 I/O await——否则紧缩探针全程钉住本线程
// 纪元，阻塞其他会话的 safe_head 推进与页回收；内存驻留候选在短守卫内读取，
// 恰好闭环 probe_resident 无锁裸读契约要求的 LightEpoch 保护（内存命中路径
// 纯同步完成、无实际协程让出）
let record = if self.store.hlog().is_on_disk(main_addr) {
self.store.hlog().read_disk_record(main_addr).await
} else {
let _guard = session.enter_epoch();
self.store.hlog().read_record(main_addr).await
};
match record {
Ok(record) => {
match record.key() {
Ok(rec_key) if rec_key == key => {
if found.is_none() {
let is_tombstone = record.is_tombstone().unwrap_or(false);
found = Some(LatestRecord {
main_addr,
index_addr: addr,
is_tombstone,
});
} else {
duplicates.push(addr);
}
}
// 键解码失败：介质损坏或半帧，本轮候选弃迁必须显式暴露，
// 静默跳过会让存活记录在紧缩后凭空消失且无迹可查
Err(e) => {
log::warn!(
"紧缩探针：候选 {main_addr:#x} 键解码失败（{e}），本轮弃迁，存活记录可能丢失可见性"
);
}
Ok(_) => {}
}
}
Err(e) => {
// 读取间隙起始地址被并发紧缩推进，按陈旧槽位处理；其余错误快速上抛
if main_addr < self.store.begin_address() {
duplicates.push(addr);
continue;
}
return Err(e.into());
}
}
}
for &stale_addr in duplicates.iter() {
self.store.index().delete(key, stale_addr);
}
Ok(found)
}
}