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:54:07 UTC
(opendal-yinyang) branch xuanwo/managed-sync-reconciliation updated: fix: map publication conflicts, retire resolved work, complete inspection evidence
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 4d87a66 fix: map publication conflicts, retire resolved work, complete inspection evidence
4d87a66 is described below
commit 4d87a662848ee84fcf4294e32186da1bbeb9aff3
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Tue Oct 27 03:53:59 2026 +0800
fix: map publication conflicts, retire resolved work, complete inspection evidence
- Translate a publication predicate Conflict into a durable directory conflict
instead of aborting the run; keep Retryable and Unknown distinct.
- Exclude paths owned by unresolved or terminal work from reconciliation so the
exact original request is never replaced by a fresh edit.
- Select the newest non-retired work request per path; supersede work belonging
to a resolved conflict while keeping its history.
- Report LocalPending with a RemoteCompleted outcome when the prior receipt
committed and a newer local edit is pending.
- Remove a whole baseline subtree when a remote-deleted directory is
quarantined as one unit.
- Stage parent creates/renames against the dependency-affected parent revision.
- Expose artifact physical path and exportability, and add a local-only conflict
side export that needs no Authority.
Co-authored-by: Chojan Shang <ps...@apache.org>
---
src/replica/directory.rs | 285 +++++++++++++++++++++++++++++++++++------------
src/replica/linux.rs | 50 +++++----
src/replica/state.rs | 51 +++++++--
3 files changed, 285 insertions(+), 101 deletions(-)
diff --git a/src/replica/directory.rs b/src/replica/directory.rs
index 4cdeecb..9507cfc 100644
--- a/src/replica/directory.rs
+++ b/src/replica/directory.rs
@@ -529,10 +529,18 @@ impl DirectorySync {
)
} else if local_action != Action::None {
pending += 1;
+ // A local change is pending, but a previous request may already
+ // have completed: report that receipt separately from the new
+ // local edit so a completed v1 is not shown as merely Pending.
+ let previous_completed = baseline_node.is_some() && remote_action == Action::None;
(
EntryState::LocalPending,
OutcomeEvidence {
- state: OutcomeState::Pending,
+ state: if previous_completed {
+ OutcomeState::RemoteCompleted
+ } else {
+ OutcomeState::Pending
+ },
error: None,
},
)
@@ -628,28 +636,9 @@ impl DirectorySync {
side: &str,
output: &mut dyn Write,
) -> Result<()> {
- let record = self
- .replica
- .store()
- .conflicts()?
- .into_iter()
- .find(|record| record.id == conflict_id)
- .ok_or_else(|| crate::replica::Error::invalid("unknown conflict").to_string())?;
+ let record = self.load_conflict_record(conflict_id)?;
match side {
- "local" => {
- if let Some(artifact) = record.local_artifact {
- return Ok(self.replica.export_artifact(&artifact, output)?);
- }
- // The live local file for the conflict's current path.
- let path = record
- .desired_path
- .as_deref()
- .unwrap_or(record.path.as_str());
- let file = self.replica.open_local_file(path)?;
- let mut file = file;
- std::io::copy(&mut file, output).map_err(|error| error.to_string())?;
- Ok(())
- }
+ "local" => self.export_conflict_local(conflict_id, output),
"remote" => {
let node = record
.remote_node
@@ -681,6 +670,37 @@ impl DirectorySync {
}
}
+ /// Export only the local side of a conflict. The bytes are local (a
+ /// retained artifact or the live local file), so no Authority is needed;
+ /// it keeps working offline.
+ pub fn export_conflict_local(&self, conflict_id: &str, output: &mut dyn Write) -> Result<()> {
+ let record = self.load_conflict_record(conflict_id)?;
+ if let Some(artifact) = record.local_artifact {
+ return Ok(self.replica.export_artifact(&artifact, output)?);
+ }
+ let path = record
+ .desired_path
+ .as_deref()
+ .unwrap_or(record.path.as_str());
+ let file = self.replica.open_local_file(path)?;
+ let mut file = file;
+ std::io::copy(&mut file, output).map_err(|error| error.to_string())?;
+ Ok(())
+ }
+
+ fn load_conflict_record(&self, conflict_id: &str) -> Result<ConflictRecord> {
+ match self
+ .replica
+ .store()
+ .conflicts()?
+ .into_iter()
+ .find(|record| record.id == conflict_id)
+ {
+ Some(record) => Ok(record),
+ None => self.synthetic_conflict(conflict_id),
+ }
+ }
+
pub async fn run(&mut self) -> Result<RunOutcome> {
let mut outcome = RunOutcome::default();
self.retry_associated(&mut outcome).await?;
@@ -694,6 +714,20 @@ impl DirectorySync {
.await?;
}
+ // Paths owned by unresolved or terminal work are excluded from this
+ // pass so the exact original request is never silently superseded by a
+ // freshly staged edit.
+ let blocked: BTreeSet<String> = self
+ .replica
+ .store()
+ .work()?
+ .into_iter()
+ .filter_map(|record| match record.state.as_str() {
+ "prepared" | "unknown" | "retryable" | "conflict" | "failed" => Some(record.path),
+ _ => None,
+ })
+ .collect();
+
if let Some(baseline) = self.replica.store().load_baseline()? {
let mut ctx = Context {
sync: &mut self.sync,
@@ -706,6 +740,7 @@ impl DirectorySync {
nodes: BTreeMap::new(),
revisions: BTreeMap::new(),
dirty: BTreeSet::new(),
+ blocked,
};
ctx.reconcile().await?;
}
@@ -717,16 +752,7 @@ impl DirectorySync {
/// Apply one durable decision to an active conflict, then reconcile again.
/// The conflict stays active until its derived request actually commits.
pub async fn resolve(&mut self, conflict_id: &str, decision: Resolution) -> Result<RunOutcome> {
- let record = match self
- .replica
- .store()
- .conflicts()?
- .into_iter()
- .find(|record| record.id == conflict_id)
- {
- Some(record) => record,
- None => self.synthetic_conflict(conflict_id)?,
- };
+ let record = self.load_conflict_record(conflict_id)?;
let observed = self.sync.authority().observe_latest().await?;
let remote = crate::directory_nodes(&observed).await?;
let baseline = self.replica.store().load_baseline()?;
@@ -754,7 +780,12 @@ impl DirectorySync {
.find(|artifact| artifact_conflict_identity(&artifact.id) == conflict_id);
match artifact {
Some(artifact) => self.replica.acknowledge_retained_artifact(&artifact.id)?,
- None => self.replica.store_mut().remove_conflict(conflict_id)?,
+ None => {
+ self.replica
+ .store_mut()
+ .supersede_conflict_work(conflict_id)?;
+ self.replica.store_mut().remove_conflict(conflict_id)?;
+ }
}
Ok(())
}
@@ -1169,7 +1200,10 @@ impl DirectorySync {
async fn retry_associated(&mut self, outcome: &mut RunOutcome) -> Result<()> {
let work = self.replica.store().work()?;
for record in work {
- if matches!(record.state.as_str(), "committed" | "conflict" | "failed") {
+ if matches!(
+ record.state.as_str(),
+ "committed" | "superseded" | "conflict" | "failed"
+ ) {
continue;
}
let edit = EditId::from_bytes(record.id);
@@ -1302,6 +1336,8 @@ struct Context<'a> {
/// Paths whose remote namespace we changed earlier in this run; their
/// staging predicate must reflect those receipts.
dirty: BTreeSet<String>,
+ /// Paths owned by unresolved or terminal work; never re-planned here.
+ blocked: BTreeSet<String>,
}
impl<'a> Context<'a> {
@@ -1339,7 +1375,11 @@ impl<'a> Context<'a> {
let mut dirs: Vec<String> = paths.iter().cloned().collect();
dirs.sort_by_key(|path| (depth(path), path.clone()));
for path in &dirs {
- if path.is_empty() || handled.contains(path) || under_any_inclusive(&retained, path) {
+ if path.is_empty()
+ || handled.contains(path)
+ || self.blocked.contains(path)
+ || under_any_inclusive(&retained, path)
+ {
continue;
}
let is_remote_deleted_dir = self
@@ -1355,6 +1395,7 @@ impl<'a> Context<'a> {
continue;
}
self.materialize_delete(path).await?;
+ self.replica.store_mut().remove_baseline_subtree(path)?;
self.mark_subtree(&mut handled, path);
handled.insert(path.clone());
}
@@ -1362,7 +1403,7 @@ impl<'a> Context<'a> {
let mut ascending: Vec<String> = paths.iter().cloned().collect();
ascending.sort_by_key(|path| (depth(path), path.clone()));
for path in &ascending {
- if path.is_empty() || handled.contains(path) {
+ if path.is_empty() || handled.contains(path) || self.blocked.contains(path) {
continue;
}
if under_any(&retained, path) {
@@ -1375,7 +1416,7 @@ impl<'a> Context<'a> {
descending.sort_by_key(|path| (depth(path), path.clone()));
descending.reverse();
for path in &descending {
- if handled.contains(path) || under_any(&retained, path) {
+ if handled.contains(path) || self.blocked.contains(path) || under_any(&retained, path) {
continue;
}
let local = local_action(self.baseline.local.get(path), self.current.get(path));
@@ -1440,6 +1481,9 @@ impl<'a> Context<'a> {
.cloned()
.collect();
for source in roots {
+ if self.blocked.contains(&source) {
+ continue;
+ }
let rm = remote_moves.get(&source);
let lm = local_moves.get(&source);
match (rm, lm) {
@@ -1480,9 +1524,12 @@ impl<'a> Context<'a> {
remote_action(self.baseline.remote.get(&source), self.remote.get(&source))
== Action::None;
if remote_unchanged && !self.remote.contains_key(destination) {
- let revision = self.local_rename(&source, destination).await?;
- self.migrate_subtree(handled, &source, destination, revision)
- .await?;
+ if let Some(revision) = self.local_rename(&source, destination).await? {
+ self.migrate_subtree(handled, &source, destination, revision)
+ .await?;
+ } else {
+ self.mark_subtree(handled, &source);
+ }
} else {
self.mark_subtree(handled, &source);
self.conflict(&source, Some(&source), Some(&source)).await?;
@@ -1773,7 +1820,7 @@ impl<'a> Context<'a> {
.get(parent)
.copied()
.ok_or_else(|| format!("directory parent {parent:?} is missing"))?;
- let parent_revision = self.revisions.get(parent).copied().unwrap_or(self.revision);
+ let parent_revision = self.entry_revision(parent);
let edit = self
.sync
.stage_create_directory(parent_node, name, parent_revision)
@@ -1788,7 +1835,9 @@ impl<'a> Context<'a> {
Some(parent_node),
parent_revision,
)?;
- let revision = self.sync.publish(edit).await?;
+ let Some(revision) = self.publish_edit(edit, path).await? else {
+ return Ok(());
+ };
self.revision = revision;
let node_id = self.namespace_node(edit).await?;
self.nodes.insert(path.to_owned(), node_id);
@@ -1812,7 +1861,7 @@ impl<'a> Context<'a> {
.get(parent)
.copied()
.ok_or_else(|| format!("file parent {parent:?} is missing"))?;
- let parent_revision = self.revisions.get(parent).copied().unwrap_or(self.revision);
+ let parent_revision = self.entry_revision(parent);
let captured = self.replica.capture_file(path, node)?;
let mut source =
tokio::fs::File::from_std(self.replica.open_artifact(&captured.artifact.id)?);
@@ -1836,7 +1885,9 @@ impl<'a> Context<'a> {
Some(parent_node),
parent_revision,
)?;
- let revision = self.sync.publish(edit).await?;
+ let Some(revision) = self.publish_edit(edit, path).await? else {
+ return Ok(());
+ };
self.revision = revision;
let node_id = self.namespace_node(edit).await?;
self.nodes.insert(path.to_owned(), node_id);
@@ -1872,7 +1923,9 @@ impl<'a> Context<'a> {
Some(remote_node),
predicate,
)?;
- let revision = self.sync.publish(edit).await?;
+ let Some(revision) = self.publish_edit(edit, path).await? else {
+ return Ok(());
+ };
self.revision = revision;
self.revisions.insert(path.to_owned(), revision);
self.accept_local(path, node, remote_node, revision).await?;
@@ -1902,7 +1955,9 @@ impl<'a> Context<'a> {
Some(node),
predicate,
)?;
- let revision = self.sync.publish(edit).await?;
+ let Some(revision) = self.publish_edit(edit, path).await? else {
+ return Ok(());
+ };
self.revision = revision;
self.nodes.remove(path);
self.revisions.remove(path);
@@ -1916,7 +1971,7 @@ impl<'a> Context<'a> {
Ok(())
}
- async fn local_rename(&mut self, from: &str, to: &str) -> Result<Revision> {
+ async fn local_rename(&mut self, from: &str, to: &str) -> Result<Option<Revision>> {
let node = self
.nodes
.get(from)
@@ -1943,7 +1998,9 @@ impl<'a> Context<'a> {
Some(node),
predicate,
)?;
- let revision = self.sync.publish(edit).await?;
+ let Some(revision) = self.publish_edit(edit, from).await? else {
+ return Ok(None);
+ };
self.revision = revision;
// The rename receipt proves the namespace move, never a concurrent
// content edit. Preserve the previously accepted fingerprint for `to`.
@@ -1955,7 +2012,7 @@ impl<'a> Context<'a> {
.update_work(edit.as_bytes(), "committed", Some(revision), None)?;
self.outcome.published_local += 1;
self.outcome.revision = Some(revision);
- Ok(revision)
+ Ok(Some(revision))
}
async fn materialize_create(&mut self, path: &str, node: &Node) -> Result<()> {
@@ -2076,6 +2133,101 @@ impl<'a> Context<'a> {
}
}
+ /// Publish an associated edit and translate a predicate Conflict into a
+ /// durable directory conflict instead of aborting the run.
+ async fn publish_edit(&mut self, edit: EditId, path: &str) -> Result<Option<Revision>> {
+ match self.sync.publish(edit).await {
+ Ok(revision) => Ok(Some(revision)),
+ Err(error) if error.kind() == crate::runtime::ErrorKind::Conflict => {
+ self.replica.store_mut().update_work(
+ edit.as_bytes(),
+ "conflict",
+ None,
+ Some(&error.to_string()),
+ )?;
+ self.record_publish_conflict(path).await?;
+ Ok(None)
+ }
+ Err(error) if error.kind() == crate::runtime::ErrorKind::Retryable => {
+ self.replica.store_mut().update_work(
+ edit.as_bytes(),
+ "retryable",
+ None,
+ Some(&error.to_string()),
+ )?;
+ Ok(None)
+ }
+ Err(error) => {
+ if let Some(commit) = error.commit_id() {
+ self.replica
+ .store_mut()
+ .set_work_commit_id(edit.as_bytes(), *commit.as_bytes())?;
+ }
+ self.replica.store_mut().update_work(
+ edit.as_bytes(),
+ work_state_for(&error),
+ None,
+ Some(&error.to_string()),
+ )?;
+ Err(error.into())
+ }
+ }
+ }
+
+ /// Record the current remote evidence for a path whose publication
+ /// predicate no longer held.
+ async fn record_publish_conflict(&mut self, path: &str) -> Result<()> {
+ let observed = self.sync.authority().observe_latest().await?;
+ let remote_node = observed.resolve(path).await?;
+ let local = self.local_side(path);
+ let remote = match &remote_node {
+ Some(node) if node.is_directory() => "current_directory",
+ Some(_) => "current_file",
+ None => "deleted",
+ };
+ let baseline_node = self.baseline.remote.get(path).map(|entry| entry.node);
+ let remote_id = remote_node.as_ref().map(Node::id);
+ let id = conflict_identity(
+ path,
+ Some(path),
+ Some(path),
+ local,
+ remote,
+ baseline_node.map(|node| *node.as_bytes()),
+ remote_id,
+ );
+ let existing_artifact = self
+ .replica
+ .store()
+ .conflicts()?
+ .into_iter()
+ .find(|conflict| conflict.id == id)
+ .and_then(|conflict| conflict.local_artifact);
+ let local_artifact = match existing_artifact {
+ Some(artifact) => Some(artifact),
+ None => self
+ .current
+ .get(path)
+ .filter(|node| matches!(node.kind, LocalKind::File { .. }))
+ .and_then(|node| self.replica.capture_file(path, node).ok())
+ .map(|captured| captured.artifact.id),
+ };
+ record_conflict_side(
+ self.outcome,
+ self.replica,
+ &id,
+ path,
+ Some(path),
+ Some(path),
+ local,
+ remote,
+ local_artifact.as_deref(),
+ remote_id,
+ Some(observed.revision()),
+ "a concurrent remote edit changed the publication predicate",
+ )
+ }
+
fn entry_revision(&self, path: &str) -> Revision {
// Preserve each path's own accepted revision, except where our earlier
// namespace receipts in this run changed that path; then stage against
@@ -2330,18 +2482,6 @@ fn work_state_for(error: &crate::runtime::Error) -> &'static str {
}
}
-fn intent_priority(state: Option<IntentState>) -> u8 {
- match state {
- Some(IntentState::Unknown) => 6,
- Some(IntentState::Prepared) => 5,
- Some(IntentState::Retryable) => 4,
- Some(IntentState::Conflict) => 3,
- Some(IntentState::Failed) => 2,
- Some(IntentState::Committed) => 1,
- None => 0,
- }
-}
-
/// The durable state of one associated work record, preferring the Sync
/// journal's own outcome when it is present.
fn effective_intent_state(
@@ -2370,18 +2510,18 @@ fn effective_intent_state(
fn select_work<'a>(
work: &'a [WorkRecord],
path: &str,
- inspect: Option<&crate::sync::SyncInspect>,
+ _inspect: Option<&crate::sync::SyncInspect>,
) -> Option<&'a WorkRecord> {
- let mut best: Option<&WorkRecord> = None;
- let mut best_rank = 0_u8;
- for record in work.iter().filter(|record| record.path == path) {
- let rank = intent_priority(effective_intent_state(record, inspect));
- if best.is_none() || rank > best_rank {
- best = Some(record);
- best_rank = rank;
- }
- }
- best
+ // The newest non-retired request is the current one. History remains
+ // stored but never hides a later request for the same path.
+ work.iter()
+ .filter(|record| record.path == path && record.state != "superseded")
+ .max_by_key(|record| record.seq)
+ .or_else(|| {
+ work.iter()
+ .filter(|record| record.path == path)
+ .max_by_key(|record| record.seq)
+ })
}
/// Count each associated logical request once by its effective durable state.
@@ -2695,6 +2835,7 @@ fn intent_state_from_work(state: &str) -> IntentState {
"prepared" => IntentState::Prepared,
"unknown" => IntentState::Unknown,
"committed" => IntentState::Committed,
+ "superseded" => IntentState::Committed,
"conflict" => IntentState::Conflict,
"retryable" => IntentState::Retryable,
_ => IntentState::Failed,
diff --git a/src/replica/linux.rs b/src/replica/linux.rs
index 36a848b..c396768 100644
--- a/src/replica/linux.rs
+++ b/src/replica/linux.rs
@@ -145,6 +145,8 @@ pub struct Artifact {
pub reason: String,
pub phase: String,
pub error: Option<String>,
+ pub path: PathBuf,
+ pub exportable: bool,
}
#[derive(Clone, Debug, Eq, PartialEq)]
@@ -363,7 +365,7 @@ impl Replica {
.store
.artifacts()?
.into_iter()
- .map(Artifact::from)
+ .map(|record| self.artifact_from(record))
.collect();
let conflicts = self
.store
@@ -437,7 +439,11 @@ impl Replica {
fn inspect_status_readonly(&self) -> Result<ReplicaStatus> {
let records = self.store.artifacts()?;
let installs = self.store.installs()?;
- let mut artifacts: Vec<Artifact> = records.iter().cloned().map(Artifact::from).collect();
+ let mut artifacts: Vec<Artifact> = records
+ .iter()
+ .cloned()
+ .map(|record| self.artifact_from(record))
+ .collect();
for (artifact, record) in artifacts.iter_mut().zip(records) {
let install = installs
.iter()
@@ -654,7 +660,7 @@ impl Replica {
self.store
.update_artifact(&artifact.id, &artifact.phase, artifact.error.as_deref())?;
Ok(CapturedContent {
- artifact: artifact.clone().into(),
+ artifact: self.artifact_from(artifact.clone()),
length,
digest,
executable: expected.executable,
@@ -1130,7 +1136,7 @@ impl Replica {
}
self.store
.update_artifact(&record.id, &record.phase, record.error.as_deref())?;
- Ok(record.into())
+ Ok(self.artifact_from(record))
}
pub fn export_artifact(&self, id: &str, output: &mut (impl Write + ?Sized)) -> Result<()> {
@@ -1174,6 +1180,24 @@ impl Replica {
Ok(self.state.join("quarantine").join(artifact.name))
}
+ /// Canonical domain evidence for a retained artifact: physical path and
+ /// whether one byte stream can represent it.
+ fn artifact_from(&self, record: ArtifactRecord) -> Artifact {
+ let path = self.state.join("quarantine").join(&record.name);
+ let exportable = !matches!(record.kind.as_str(), "quarantined-directory");
+ Artifact {
+ id: record.id,
+ origin: record.origin,
+ name: record.name,
+ kind: record.kind,
+ reason: record.reason,
+ phase: record.phase,
+ error: record.error,
+ path,
+ exportable,
+ }
+ }
+
fn scan_directory(
&self,
directory: &OwnedFd,
@@ -1935,7 +1959,7 @@ impl PreparedInstall<'_> {
self.replica
.store
.update_artifact(&record.id, &record.phase, record.error.as_deref())?;
- Ok(record.into())
+ Ok(self.replica.artifact_from(record))
}
fn retain_displaced(&mut self, reason: &str, phase: InstallPhase) -> Result<()> {
@@ -2022,25 +2046,11 @@ impl PreparedInstall<'_> {
.artifacts()?
.into_iter()
.find(|artifact| artifact.id == self.record.id)
- .map(Artifact::from)
+ .map(|artifact| self.replica.artifact_from(artifact))
.ok_or_else(|| Error::corrupt("retained artifact record disappeared"))
}
}
-impl From<ArtifactRecord> for Artifact {
- fn from(value: ArtifactRecord) -> Self {
- Self {
- id: value.id,
- origin: value.origin,
- name: value.name,
- kind: value.kind,
- reason: value.reason,
- phase: value.phase,
- error: value.error,
- }
- }
-}
-
impl From<ConflictRecord> for Conflict {
fn from(value: ConflictRecord) -> Self {
Self {
diff --git a/src/replica/state.rs b/src/replica/state.rs
index 51de162..77e399a 100644
--- a/src/replica/state.rs
+++ b/src/replica/state.rs
@@ -144,6 +144,8 @@ pub(super) struct ConflictRecord {
/// associated work and keep orphaned staging unpublished.
#[derive(Clone, Debug)]
pub(super) struct WorkRecord {
+ /// Monotonic insertion order, used to select the current request.
+ pub seq: u64,
pub id: [u8; 32],
pub path: String,
pub original_path: Option<String>,
@@ -258,6 +260,7 @@ impl Store {
);
CREATE TABLE IF NOT EXISTS work(
id BLOB PRIMARY KEY,
+ seq INTEGER NOT NULL DEFAULT 0,
path TEXT NOT NULL,
original_path TEXT,
desired_path TEXT,
@@ -283,6 +286,8 @@ impl Store {
.map_err(|e| Error::io("initialize replica state", path, e))?;
ensure_column(&connection, "baseline", "remote_revision", "BLOB")
.map_err(|e| Error::io("migrate replica baseline", path, e))?;
+ ensure_column(&connection, "work", "seq", "INTEGER NOT NULL DEFAULT 0")
+ .map_err(|e| Error::io("migrate replica work", path, e))?;
ensure_column(&connection, "work", "conflict", "TEXT")
.map_err(|e| Error::io("migrate replica work", path, e))?;
ensure_column(&connection, "work", "source_artifact", "TEXT")
@@ -591,6 +596,19 @@ impl Store {
Ok(())
}
+ /// Remove a baseline directory entry and every entry below it, used when a
+ /// whole unchanged subtree is quarantined as one unit.
+ pub fn remove_baseline_subtree(&mut self, path: &str) -> Result<()> {
+ self.connection
+ .execute(
+ "DELETE FROM baseline
+ WHERE path=?1 OR substr(path,1,length(?1)+1)=?1||'/'",
+ params![path],
+ )
+ .map_err(|e| Error::corrupt(format!("remove baseline subtree: {e}")))?;
+ Ok(())
+ }
+
pub fn set_baseline_revision(&mut self, revision: Revision) -> Result<()> {
self.connection
.execute(
@@ -822,6 +840,18 @@ impl Store {
Ok(())
}
+ /// Retire work that belonged to a resolved conflict while keeping its
+ /// history for audit; a later request at the path becomes the current one.
+ pub fn supersede_conflict_work(&mut self, conflict: &str) -> Result<()> {
+ self.connection
+ .execute(
+ "UPDATE work SET state='superseded' WHERE conflict=?1 AND state!='committed'",
+ params![conflict],
+ )
+ .map_err(|e| Error::corrupt(format!("supersede conflict work: {e}")))?;
+ Ok(())
+ }
+
/// Refresh volatile evidence of an active conflict without replacing its
/// retained local artifact or changing its stable identity.
pub fn refresh_conflict(&mut self, conflict: &ConflictRecord) -> Result<()> {
@@ -855,11 +885,12 @@ impl Store {
self.connection
.execute(
"INSERT OR REPLACE INTO work
- (id,path,original_path,desired_path,operation,local_kind,local_digest,
+ (seq,id,path,original_path,desired_path,operation,local_kind,local_digest,
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,
+ VALUES((SELECT COALESCE(MAX(seq),0)+1 FROM work),
+ ?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(),
@@ -929,7 +960,7 @@ impl Store {
let mut query = self
.connection
.prepare(
- "SELECT id,path,original_path,desired_path,operation,local_kind,local_digest,
+ "SELECT seq,id,path,original_path,desired_path,operation,local_kind,local_digest,
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
@@ -950,12 +981,14 @@ impl Store {
let remote_revision: Option<Vec<u8>> = row.get(15)?;
let commit_id: Option<Vec<u8>> = row.get(16)?;
Ok(WorkRecord {
+ seq: u64::try_from(row.get::<_, i64>(0)?)
+ .map_err(|_| corrupt_sql("negative work sequence"))?,
id: fixed(&id, "work id")?,
- path: row.get(1)?,
- original_path: row.get(2)?,
- desired_path: row.get(3)?,
- operation: row.get(4)?,
- local_kind: row.get(5)?,
+ path: row.get(2)?,
+ original_path: row.get(3)?,
+ desired_path: row.get(4)?,
+ operation: row.get(5)?,
+ local_kind: row.get(6)?,
local_digest: local_digest
.as_deref()
.map(|value| fixed(value, "work local digest"))
@@ -965,7 +998,7 @@ impl Store {
u64::try_from(value).map_err(|_| corrupt_sql("negative work length"))
})
.transpose()?,
- local_executable: row.get(8)?,
+ local_executable: row.get(9)?,
local_dev: local_dev
.as_deref()
.map(|value| number(value, "work local device"))