You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2026/10/26 19:16:57 UTC
(opendal-yinyang) branch xuanwo/managed-sync-reconciliation updated: fix: persist accepted identity and commit id, order completion safely
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch xuanwo/managed-sync-reconciliation
in repository https://gitbox.apache.org/repos/asf/opendal-yinyang.git
The following commit(s) were added to refs/heads/xuanwo/managed-sync-reconciliation by this push:
new 7af6c3d fix: persist accepted identity and commit id, order completion safely
7af6c3d is described below
commit 7af6c3d53fb9908e052075c8181e4ed186912908
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Tue Oct 27 03:16:51 2026 +0800
fix: persist accepted identity and commit id, order completion safely
- Persist the accepted local inode and parent identity with the work
association so a delivered v1 is not lost when the local file was removed
while its response was held.
- Persist the original commit identity before a replay clears it, so status
keeps the same ID through a successful Unknown retry.
- Preserve Retryable in inspection instead of collapsing it to Failed.
- Apply baseline and clear the linked conflict before marking work complete;
idempotent publication makes a crash there recoverable on the next run.
Co-authored-by: Chojan Shang <ps...@apache.org>
---
src/replica/directory.rs | 83 ++++++++++++++++++++++++++++++++++----------
src/replica/mod.rs | 4 +--
src/replica/state.rs | 90 +++++++++++++++++++++++++++++++++++++++++-------
3 files changed, 144 insertions(+), 33 deletions(-)
diff --git a/src/replica/directory.rs b/src/replica/directory.rs
index a946713..b1fd0d6 100644
--- a/src/replica/directory.rs
+++ b/src/replica/directory.rs
@@ -23,7 +23,7 @@
//! read-only domain evidence for the adapter to project.
use crate::replica::state::{BaselineNode, BaselineRecord, ConflictRecord, WorkRecord};
-use crate::replica::{Artifact, Conflict, LocalKind, LocalNode, LocalSnapshot, Replica};
+use crate::replica::{Artifact, Conflict, Identity, LocalKind, LocalNode, LocalSnapshot, Replica};
use crate::sync::{EditId, FileMetadata, NamespaceState, RenameMode, Sync};
use std::collections::{BTreeMap, BTreeSet};
use std::io::{Seek as _, SeekFrom, Write};
@@ -709,7 +709,7 @@ impl DirectorySync {
let observed = self.sync.authority().observe_latest().await?;
let remote = crate::directory_nodes(&observed).await?;
let baseline = self.replica.store().load_baseline()?;
- let applied = match decision {
+ let _applied = match decision {
Resolution::KeepLocal | Resolution::Content(_) => {
self.resolve_keep_local(&record, &remote, baseline.as_ref(), decision)
.await?
@@ -719,9 +719,6 @@ impl DirectorySync {
.await?
}
};
- if applied {
- self.clear_resolved_conflict(conflict_id)?;
- }
self.run().await
}
@@ -831,9 +828,30 @@ impl DirectorySync {
local_digest: Some(*blake3::hash(&bytes).as_bytes()),
local_length: Some(bytes.len() as u64),
local_executable: Some(false),
+ local_dev: self
+ .replica
+ .scan()?
+ .get(&path)
+ .map(|node| node.identity.dev),
+ local_ino: self
+ .replica
+ .scan()?
+ .get(&path)
+ .map(|node| node.identity.ino),
+ local_parent_dev: self
+ .replica
+ .scan()?
+ .get(&path)
+ .and_then(|node| node.parent.map(|parent| parent.dev)),
+ local_parent_ino: self
+ .replica
+ .scan()?
+ .get(&path)
+ .and_then(|node| node.parent.map(|parent| parent.ino)),
baseline_node: remote.get(&path).map(|node| *node.id().as_bytes()),
baseline_revision: predicate.to_bytes(),
remote_revision: None,
+ commit_id: None,
conflict: Some(record.id.clone()),
source_artifact: Some(artifact),
state: "prepared".to_owned(),
@@ -844,6 +862,7 @@ impl DirectorySync {
self.install_local_bytes(&path, &bytes)?;
self.apply_resolved_baseline(record, &path, revision)
.await?;
+ self.clear_resolved_conflict(&record.id)?;
self.replica.store_mut().update_work(
edit.as_bytes(),
"committed",
@@ -900,6 +919,7 @@ impl DirectorySync {
self.replica
.store_mut()
.remove_baseline_path(&record.path)?;
+ self.clear_resolved_conflict(&record.id)?;
return Ok(true);
};
if remote.get(&path).is_none() {
@@ -946,6 +966,7 @@ impl DirectorySync {
}
self.apply_resolved_baseline(record, &path, snapshot.revision())
.await?;
+ self.clear_resolved_conflict(&record.id)?;
Ok(true)
}
@@ -1142,15 +1163,23 @@ impl DirectorySync {
continue;
}
let edit = EditId::from_bytes(record.id);
+ // Capture the original commit identity before a replay clears it.
+ if let Ok(crate::sync::EditStatus::Content(status)) = self.sync.edit_status(edit).await
+ && let Some(commit) = status.frozen_commit_id
+ {
+ self.replica
+ .store_mut()
+ .set_work_commit_id(&record.id, *commit.as_bytes())?;
+ }
match self.sync.publish(edit).await {
Ok(revision) => {
outcome.published_local += 1;
outcome.revision = Some(revision);
- // Apply baseline from the returned revision before marking
- // the work complete, so a crash cannot skip receipt
- // application or lose the conflict link.
+ // Apply baseline and clear the linked conflict before
+ // marking the work complete. Publication is idempotent, so
+ // a crash here is recovered by re-running this retry.
if let Some(node) = self.accepted_node_at(&record, revision).await? {
- let accepted = accepted_local(&self.replica, &record)?;
+ let accepted = accepted_local(&record)?;
self.replica.store_mut().save_baseline_path(
&record.path,
&node,
@@ -1158,18 +1187,23 @@ impl DirectorySync {
revision,
)?;
}
+ if let Some(conflict) = &record.conflict {
+ self.clear_resolved_conflict(conflict)?;
+ }
self.replica.store_mut().update_work(
&record.id,
"committed",
Some(revision),
None,
)?;
- if let Some(conflict) = &record.conflict {
- self.clear_resolved_conflict(conflict)?;
- }
}
Err(error) if error.kind() == crate::runtime::ErrorKind::Unknown => {
outcome.unknown += 1;
+ if let Some(commit) = error.commit_id() {
+ self.replica
+ .store_mut()
+ .set_work_commit_id(&record.id, *commit.as_bytes())?;
+ }
self.replica.store_mut().update_work(
&record.id,
"unknown",
@@ -1204,10 +1238,7 @@ impl DirectorySync {
}
}
-fn accepted_local(replica: &Replica, record: &WorkRecord) -> Result<Option<LocalNode>> {
- let fresh = replica.scan()?;
- let identity = fresh.get(&record.path).map(|node| node.identity);
- let parent = fresh.get(&record.path).and_then(|node| node.parent);
+fn accepted_local(record: &WorkRecord) -> Result<Option<LocalNode>> {
let Some(kind) = record.local_kind.as_deref() else {
return Ok(None);
};
@@ -1223,9 +1254,14 @@ fn accepted_local(replica: &Replica, record: &WorkRecord) -> Result<Option<Local
},
other => return Err(format!("unknown work local kind {other:?}").into()),
};
- let Some(identity) = identity else {
+ let (Some(dev), Some(ino)) = (record.local_dev, record.local_ino) else {
return Ok(None);
};
+ let identity = Identity { dev, ino };
+ let parent = match (record.local_parent_dev, record.local_parent_ino) {
+ (Some(dev), Some(ino)) => Some(Identity { dev, ino }),
+ _ => None,
+ };
Ok(Some(LocalNode {
path: record.path.clone(),
kind,
@@ -2070,9 +2106,14 @@ impl<'a> Context<'a> {
local_digest,
local_length,
local_executable,
+ local_dev: local.map(|node| node.identity.dev),
+ local_ino: local.map(|node| node.identity.ino),
+ local_parent_dev: local.and_then(|node| node.parent.map(|parent| parent.dev)),
+ local_parent_ino: local.and_then(|node| node.parent.map(|parent| parent.ino)),
baseline_node: baseline_node.map(|node| *node.as_bytes()),
baseline_revision: baseline_revision.to_bytes(),
remote_revision: None,
+ commit_id: None,
conflict: None,
source_artifact: None,
state: "prepared".to_owned(),
@@ -2581,7 +2622,7 @@ fn intent_evidence(
id: Some(work.id),
operation: Some(operation),
state: Some(intent_state_from_work(&work.state)),
- commit_id: None,
+ commit_id: work.commit_id.map(CommitId::from_bytes),
receipt_revision: work.remote_revision.map(Revision::from_bytes),
error: work.error.clone(),
};
@@ -2600,7 +2641,10 @@ fn intent_evidence(
evidence.error = status.failure.as_ref().map(|failure| failure.to_string());
} else if let Some(content) = inspect.content.iter().find(|content| content.id == edit) {
evidence.state = Some(intent_state_from_handle(&content.handle));
- evidence.commit_id = content.handle.frozen_commit_id;
+ evidence.commit_id = content
+ .handle
+ .frozen_commit_id
+ .or(work.commit_id.map(CommitId::from_bytes));
evidence.error = content
.handle
.error
@@ -2639,6 +2683,7 @@ fn intent_state_from_handle(handle: &crate::runtime::HandleStatus) -> IntentStat
}
match handle.error.as_ref().map(|failure| failure.kind()) {
Some(crate::runtime::ErrorKind::Unknown) => IntentState::Unknown,
+ Some(crate::runtime::ErrorKind::Retryable) => IntentState::Retryable,
Some(crate::runtime::ErrorKind::Conflict) => IntentState::Conflict,
Some(_) => IntentState::Failed,
None if handle.pending || handle.frozen => IntentState::Prepared,
diff --git a/src/replica/mod.rs b/src/replica/mod.rs
index 1579f37..2e6d08a 100644
--- a/src/replica/mod.rs
+++ b/src/replica/mod.rs
@@ -34,6 +34,6 @@ pub use directory::{
};
pub use error::{Error, ErrorKind, Result};
pub use linux::{
- Artifact, ArtifactKind, CapturedContent, Conflict, ContentHash, InstallOutcome, LocalKind,
- LocalNode, LocalSnapshot, Replica, ReplicaStatus,
+ Artifact, ArtifactKind, CapturedContent, Conflict, ContentHash, Identity, InstallOutcome,
+ LocalKind, LocalNode, LocalSnapshot, Replica, ReplicaStatus,
};
diff --git a/src/replica/state.rs b/src/replica/state.rs
index 79bd1a5..51de162 100644
--- a/src/replica/state.rs
+++ b/src/replica/state.rs
@@ -153,9 +153,15 @@ pub(super) struct WorkRecord {
pub local_digest: Option<[u8; 32]>,
pub local_length: Option<u64>,
pub local_executable: Option<bool>,
+ pub local_dev: Option<u64>,
+ pub local_ino: Option<u64>,
+ pub local_parent_dev: Option<u64>,
+ pub local_parent_ino: Option<u64>,
pub baseline_node: Option<[u8; 16]>,
pub baseline_revision: [u8; 24],
pub remote_revision: Option<[u8; 24]>,
+ /// Original commit identity, retained before a replay can clear it.
+ pub commit_id: Option<[u8; 16]>,
/// Conflict this edit resolves, if it was created by an explicit decision.
pub conflict: Option<String>,
/// Retained artifact holding the exact local input published by this edit.
@@ -260,9 +266,14 @@ impl Store {
local_digest BLOB,
local_length INTEGER,
local_executable INTEGER,
+ local_dev BLOB,
+ local_ino BLOB,
+ local_parent_dev BLOB,
+ local_parent_ino BLOB,
baseline_node BLOB,
baseline_revision BLOB NOT NULL,
remote_revision BLOB,
+ commit_id BLOB,
conflict TEXT,
source_artifact TEXT,
state TEXT NOT NULL,
@@ -276,6 +287,16 @@ impl Store {
.map_err(|e| Error::io("migrate replica work", path, e))?;
ensure_column(&connection, "work", "source_artifact", "TEXT")
.map_err(|e| Error::io("migrate replica work", path, e))?;
+ for column in [
+ "local_dev",
+ "local_ino",
+ "local_parent_dev",
+ "local_parent_ino",
+ "commit_id",
+ ] {
+ ensure_column(&connection, "work", column, "BLOB")
+ .map_err(|e| Error::io("migrate replica work", path, e))?;
+ }
Ok(Self { connection })
}
@@ -835,9 +856,11 @@ impl Store {
.execute(
"INSERT OR REPLACE INTO work
(id,path,original_path,desired_path,operation,local_kind,local_digest,
- local_length,local_executable,baseline_node,baseline_revision,
- remote_revision,conflict,source_artifact,state,error)
- VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13,?14,?15,?16)",
+ local_length,local_executable,local_dev,local_ino,local_parent_dev,
+ local_parent_ino,baseline_node,baseline_revision,remote_revision,
+ commit_id,conflict,source_artifact,state,error)
+ VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13,?14,?15,
+ ?16,?17,?18,?19,?20,?21)",
params![
work.id.as_slice(),
work.path,
@@ -851,9 +874,14 @@ impl Store {
.transpose()
.map_err(|_| Error::invalid("work length exceeds SQLite range"))?,
work.local_executable,
+ work.local_dev.map(|v| v.to_be_bytes()),
+ work.local_ino.map(|v| v.to_be_bytes()),
+ work.local_parent_dev.map(|v| v.to_be_bytes()),
+ work.local_parent_ino.map(|v| v.to_be_bytes()),
work.baseline_node.as_ref().map(|v| v.as_slice()),
work.baseline_revision.as_slice(),
work.remote_revision.as_ref().map(|v| v.as_slice()),
+ work.commit_id.as_ref().map(|v| v.as_slice()),
work.conflict,
work.source_artifact,
work.state,
@@ -885,13 +913,26 @@ impl Store {
Ok(())
}
+ /// Persist the original commit identity of an edit before a replay can
+ /// clear it from the staging handle.
+ pub fn set_work_commit_id(&mut self, id: &[u8; 32], commit_id: [u8; 16]) -> Result<()> {
+ self.connection
+ .execute(
+ "UPDATE work SET commit_id=?2 WHERE id=?1",
+ params![id.as_slice(), commit_id.as_slice()],
+ )
+ .map_err(|e| Error::corrupt(format!("set work commit id: {e}")))?;
+ Ok(())
+ }
+
pub fn work(&self) -> Result<Vec<WorkRecord>> {
let mut query = self
.connection
.prepare(
"SELECT id,path,original_path,desired_path,operation,local_kind,local_digest,
- local_length,local_executable,baseline_node,baseline_revision,
- remote_revision,conflict,source_artifact,state,error
+ local_length,local_executable,local_dev,local_ino,local_parent_dev,
+ local_parent_ino,baseline_node,baseline_revision,remote_revision,
+ commit_id,conflict,source_artifact,state,error
FROM work ORDER BY path,id",
)
.map_err(|e| Error::corrupt(format!("read work associations: {e}")))?;
@@ -900,9 +941,14 @@ impl Store {
let id: Vec<u8> = row.get(0)?;
let local_digest: Option<Vec<u8>> = row.get(6)?;
let local_length: Option<i64> = row.get(7)?;
- let baseline_node: Option<Vec<u8>> = row.get(9)?;
- let baseline_revision: Vec<u8> = row.get(10)?;
- let remote_revision: Option<Vec<u8>> = row.get(11)?;
+ let local_dev: Option<Vec<u8>> = row.get(9)?;
+ let local_ino: Option<Vec<u8>> = row.get(10)?;
+ let local_parent_dev: Option<Vec<u8>> = row.get(11)?;
+ let local_parent_ino: Option<Vec<u8>> = row.get(12)?;
+ let baseline_node: Option<Vec<u8>> = row.get(13)?;
+ let baseline_revision: Vec<u8> = row.get(14)?;
+ let remote_revision: Option<Vec<u8>> = row.get(15)?;
+ let commit_id: Option<Vec<u8>> = row.get(16)?;
Ok(WorkRecord {
id: fixed(&id, "work id")?,
path: row.get(1)?,
@@ -920,6 +966,22 @@ impl Store {
})
.transpose()?,
local_executable: row.get(8)?,
+ local_dev: local_dev
+ .as_deref()
+ .map(|value| number(value, "work local device"))
+ .transpose()?,
+ local_ino: local_ino
+ .as_deref()
+ .map(|value| number(value, "work local inode"))
+ .transpose()?,
+ local_parent_dev: local_parent_dev
+ .as_deref()
+ .map(|value| number(value, "work parent device"))
+ .transpose()?,
+ local_parent_ino: local_parent_ino
+ .as_deref()
+ .map(|value| number(value, "work parent inode"))
+ .transpose()?,
baseline_node: baseline_node
.as_deref()
.map(|value| fixed(value, "work baseline node"))
@@ -929,10 +991,14 @@ impl Store {
.as_deref()
.map(|value| fixed(value, "work remote revision"))
.transpose()?,
- conflict: row.get(12)?,
- source_artifact: row.get(13)?,
- state: row.get(14)?,
- error: row.get(15)?,
+ commit_id: commit_id
+ .as_deref()
+ .map(|value| fixed(value, "work commit id"))
+ .transpose()?,
+ conflict: row.get(17)?,
+ source_artifact: row.get(18)?,
+ state: row.get(19)?,
+ error: row.get(20)?,
})
})
.map_err(|e| Error::corrupt(format!("read work associations: {e}")))?