store: lock-refresh-write pattern to prevent duplicate UUIDs
All write paths (upsert_node, upsert_provenance, delete_node, rename_node, ingest_units) now hold StoreLock across the full refresh→check→write cycle. This prevents the race where two concurrent processes both see a key as "new" and create separate UUIDs for it. Adds append_nodes_unlocked() and append_relations_unlocked() for callers already holding the lock. Adds refresh_nodes() to replay log tail under lock before deciding create vs update. Also adds find_duplicates() for detecting existing duplicates in the log (replays full log, groups live nodes by key).
This commit is contained in:
parent
8bbc246b3d
commit
37ae37667b
3 changed files with 121 additions and 10 deletions
|
|
@ -169,12 +169,58 @@ impl Store {
|
|||
Ok(())
|
||||
}
|
||||
|
||||
/// Find all duplicate keys: keys with multiple live UUIDs in the log.
|
||||
/// Returns a map from key → vec of all live Node versions (one per UUID).
|
||||
/// The "winner" in self.nodes is always one of them.
|
||||
pub fn find_duplicates(&self) -> Result<HashMap<String, Vec<Node>>, String> {
|
||||
let path = nodes_path();
|
||||
if !path.exists() { return Ok(HashMap::new()); }
|
||||
|
||||
let file = fs::File::open(&path)
|
||||
.map_err(|e| format!("open {}: {}", path.display(), e))?;
|
||||
let mut reader = BufReader::new(file);
|
||||
|
||||
// Track latest version of each UUID
|
||||
let mut by_uuid: HashMap<[u8; 16], Node> = HashMap::new();
|
||||
|
||||
while let Ok(msg) = serialize::read_message(&mut reader, message::ReaderOptions::new()) {
|
||||
let log = msg.get_root::<memory_capnp::node_log::Reader>()
|
||||
.map_err(|e| format!("read node log: {}", e))?;
|
||||
for node_reader in log.get_nodes()
|
||||
.map_err(|e| format!("get nodes: {}", e))? {
|
||||
let node = Node::from_capnp(node_reader)?;
|
||||
let dominated = by_uuid.get(&node.uuid)
|
||||
.map(|n| node.version >= n.version)
|
||||
.unwrap_or(true);
|
||||
if dominated {
|
||||
by_uuid.insert(node.uuid, node);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Group live (non-deleted) nodes by key
|
||||
let mut by_key: HashMap<String, Vec<Node>> = HashMap::new();
|
||||
for node in by_uuid.into_values() {
|
||||
if !node.deleted {
|
||||
by_key.entry(node.key.clone()).or_default().push(node);
|
||||
}
|
||||
}
|
||||
|
||||
// Keep only duplicates
|
||||
by_key.retain(|_, nodes| nodes.len() > 1);
|
||||
Ok(by_key)
|
||||
}
|
||||
|
||||
/// Append nodes to the log file.
|
||||
/// Serializes to a Vec first, then does a single write() syscall
|
||||
/// so the append is atomic with O_APPEND even without flock.
|
||||
pub fn append_nodes(&mut self, nodes: &[Node]) -> Result<(), String> {
|
||||
let _lock = StoreLock::acquire()?;
|
||||
self.append_nodes_unlocked(nodes)
|
||||
}
|
||||
|
||||
/// Append nodes without acquiring the lock. Caller must hold StoreLock.
|
||||
pub(crate) fn append_nodes_unlocked(&mut self, nodes: &[Node]) -> Result<(), String> {
|
||||
let mut msg = message::Builder::new_default();
|
||||
{
|
||||
let log = msg.init_root::<memory_capnp::node_log::Builder>();
|
||||
|
|
@ -199,11 +245,55 @@ impl Store {
|
|||
Ok(())
|
||||
}
|
||||
|
||||
/// Replay only new entries appended to the node log since we last loaded.
|
||||
/// Call under StoreLock to catch writes from concurrent processes.
|
||||
pub(crate) fn refresh_nodes(&mut self) -> Result<(), String> {
|
||||
let path = nodes_path();
|
||||
let current_size = fs::metadata(&path).map(|m| m.len()).unwrap_or(0);
|
||||
if current_size <= self.loaded_nodes_size {
|
||||
return Ok(()); // no new data
|
||||
}
|
||||
|
||||
let file = fs::File::open(&path)
|
||||
.map_err(|e| format!("open {}: {}", path.display(), e))?;
|
||||
let mut reader = BufReader::new(file);
|
||||
reader.seek(std::io::SeekFrom::Start(self.loaded_nodes_size))
|
||||
.map_err(|e| format!("seek nodes log: {}", e))?;
|
||||
|
||||
while let Ok(msg) = serialize::read_message(&mut reader, message::ReaderOptions::new()) {
|
||||
let log = msg.get_root::<memory_capnp::node_log::Reader>()
|
||||
.map_err(|e| format!("read node log delta: {}", e))?;
|
||||
for node_reader in log.get_nodes()
|
||||
.map_err(|e| format!("get nodes delta: {}", e))? {
|
||||
let node = Node::from_capnp(node_reader)?;
|
||||
let dominated = self.nodes.get(&node.key)
|
||||
.map(|n| node.version >= n.version)
|
||||
.unwrap_or(true);
|
||||
if dominated {
|
||||
if node.deleted {
|
||||
self.nodes.remove(&node.key);
|
||||
self.uuid_to_key.remove(&node.uuid);
|
||||
} else {
|
||||
self.uuid_to_key.insert(node.uuid, node.key.clone());
|
||||
self.nodes.insert(node.key.clone(), node);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
self.loaded_nodes_size = current_size;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Append relations to the log file.
|
||||
/// Single write() syscall for atomic O_APPEND.
|
||||
pub fn append_relations(&mut self, relations: &[Relation]) -> Result<(), String> {
|
||||
let _lock = StoreLock::acquire()?;
|
||||
self.append_relations_unlocked(relations)
|
||||
}
|
||||
|
||||
/// Append relations without acquiring the lock. Caller must hold StoreLock.
|
||||
pub(crate) fn append_relations_unlocked(&mut self, relations: &[Relation]) -> Result<(), String> {
|
||||
let mut msg = message::Builder::new_default();
|
||||
{
|
||||
let log = msg.init_root::<memory_capnp::relation_log::Builder>();
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue