velesdb_memory/migration/
execute.rs

1//! The operator's path into the rebuild (#1762, PR C2b).
2//!
3//! [`super::rebuild`] takes staged handles and a journal and drives the pass;
4//! this module is everything an operator's `migrate-embeddings` invocation
5//! needs BEFORE that call can be made honestly: the diagnosis, the regime
6//! resolution, a destination that provably is not somebody else's data, a
7//! journal workspace outside both stores, and the lock. Each refusal here is
8//! distinct on purpose — an operator has to know whether they were refused by
9//! the regime, by the destination, or by another migration's lock, because the
10//! recovery for each is different.
11//!
12//! # Where the pieces live
13//!
14//! The journal workspace is a SIBLING of the destination, named after it
15//! (`<destination>.migration-journal`). Inside the source it would violate the
16//! read-only contract; inside the destination it would be swept along by the
17//! eventual switch rename (C3), which must move the rebuilt store and nothing
18//! else. A sibling survives the switch, which matters because the phases
19//! after the switch are journalled too.
20//!
21//! # What execute stops short of
22//!
23//! The pass ends with the journal at [`Phase::Prepared`] and every collection
24//! `Complete`. Validation of the destination and the switch itself are the
25//! next PR's work; see [`NOT_YET_SWITCHABLE`](super::not_yet_switchable).
26
27use super::query_error;
28use std::path::{Path, PathBuf};
29
30use velesdb_core::agent::AgentMemory;
31use velesdb_core::Database;
32
33use super::diagnosis::{diagnose, DiagnosisReport, TargetContract};
34use super::rebuild::{
35    rebuild, RebuildDestination, RebuildJournal, RebuildOutcome, RebuildSource, VectorPolicy,
36};
37use super::state::{CollectionProgress, MigrationLock, MigrationState, Phase};
38use super::strategy::Resolution;
39use crate::embedder::Embedder;
40
41/// What one `execute` run did, and where its artefacts live.
42#[derive(Debug)]
43pub struct ExecuteOutcome {
44    /// The diagnosis that gated the run.
45    pub report: DiagnosisReport,
46    /// What the pass wrote.
47    pub rebuild: RebuildOutcome,
48    /// The rebuilt store, still unswitched.
49    pub destination: PathBuf,
50    /// Where the journal (and the lock evidence) lives.
51    pub workspace: PathBuf,
52}
53
54/// Diagnose, stage, lock, rebuild, release — the whole non-dry-run path.
55///
56/// A pre-existing journal at the derived workspace is resumed, provided it
57/// describes this exact source, fingerprint and target; anything else about it
58/// is a refusal, never an overwrite.
59///
60/// # Errors
61/// Returns [`crate::MemoryError`] when the regime resolution refuses, the
62/// destination holds data no journal accounts for, the journal describes a
63/// different migration, the lock is held or left from a crash, or the pass
64/// itself fails.
65pub fn execute(
66    store: &Path,
67    scratch_parent: &Path,
68    target: &TargetContract,
69    destination: &Path,
70    embedder: &dyn Embedder,
71    batch: usize,
72) -> Result<ExecuteOutcome, crate::MemoryError> {
73    let report = diagnose(store, scratch_parent, target, Some(destination))?;
74    let staging = stage(&report, destination)?;
75
76    let lock =
77        MigrationLock::acquire(&staging.workspace, "migrate-embeddings").map_err(query_error)?;
78    let result = execute_locked(
79        &report,
80        target,
81        destination,
82        &staging.workspace,
83        &lock,
84        &ExecutePass {
85            embedder,
86            batch,
87            resuming: staging.resuming,
88            settled_fingerprint: &staging.settled_fingerprint,
89        },
90    );
91    // Release on BOTH paths. The fail-closed evidence a dropped lock leaves is
92    // for crashes — a run that reached a clean `Err` has nothing for the
93    // operator to acknowledge, and making them `rm` a lock file after every
94    // refusal would train them to do it after real crashes too.
95    let rebuild = reconcile(result, lock.release())?;
96    Ok(ExecuteOutcome {
97        report,
98        rebuild,
99        destination: destination.to_path_buf(),
100        workspace: staging.workspace,
101    })
102}
103
104/// Fold a locked pass's result and the lock release into one verdict.
105///
106/// When BOTH fail, both are reported: an operator who fixes the pass's error
107/// and reruns deserves to know beforehand that the lock evidence remains, not
108/// to discover it as a second surprise. Shared by every locked entry point in
109/// this module tree — execute, validate, switch — because the first rewrite of
110/// this pattern silently dropped the release error, and the second and third
111/// would have too.
112pub(super) fn reconcile<T>(
113    result: Result<T, crate::MemoryError>,
114    released: Result<(), String>,
115) -> Result<T, crate::MemoryError> {
116    match (result, released) {
117        (Ok(value), Ok(())) => Ok(value),
118        (Ok(_), Err(release_error)) => Err(query_error(format!(
119            "the pass completed, but releasing the migration lock failed: \
120             {release_error}. The canonical lock record remains and must be \
121             removed by hand before the next run"
122        ))),
123        (Err(error), Ok(())) => Err(error),
124        (Err(error), Err(release_error)) => Err(query_error(format!(
125            "{error}; additionally, releasing the migration lock failed: \
126             {release_error} — the canonical lock record remains and must be \
127             removed by hand before the next run"
128        ))),
129    }
130}
131
132/// What the pre-lock staging established.
133struct Staging {
134    workspace: PathBuf,
135    resuming: bool,
136    settled_fingerprint: String,
137}
138
139/// Everything between the diagnosis and the lock: the regime gate, the settle,
140/// the journal workspace and the destination checks.
141fn stage(report: &DiagnosisReport, destination: &Path) -> Result<Staging, crate::MemoryError> {
142    if let Resolution::Refuse { because, requested } = &report.resolution {
143        return Err(query_error(format!(
144            "the requested regime '{}' cannot run: {because:?}. Nothing was \
145             created; re-run --dry-run for the full report",
146            regime_word(*requested),
147        )));
148    }
149    // Settle the source BEFORE fingerprinting it: the first open of a store
150    // compacts its WAL into materialised index files, so the tree after an
151    // open is not the tree before it — and the rebuild below opens it. A
152    // fingerprint taken pre-settle would therefore never match on resume,
153    // refusing every legitimately interrupted migration as "source changed".
154    // A second open is proven to change nothing (see
155    // `settling_a_store_is_idempotent_which_the_resume_fingerprint_rests_on`),
156    // which is what makes the settled fingerprint stable. The settle itself is
157    // what any daemon start performs; it runs only after the regime gate, so a
158    // refusal leaves the source byte-identical.
159    {
160        let _settle = Database::open(&report.source_path)?;
161    }
162    let settled_fingerprint = super::filesystem::fingerprint(&report.source_path)?;
163    let workspace = journal_workspace(destination)?;
164    let resuming = workspace.join(super::state::STATE_FILE).exists();
165    ensure_destination(destination, resuming)?;
166    Ok(Staging {
167        workspace,
168        resuming,
169        settled_fingerprint,
170    })
171}
172
173/// The run's inputs beyond the diagnosis: how to embed, how much per batch,
174/// whether a journal already existed, and the post-settle fingerprint the
175/// journal carries (the diagnosis's own fingerprint predates the settle and
176/// would never match on resume).
177struct ExecutePass<'a> {
178    embedder: &'a dyn Embedder,
179    batch: usize,
180    resuming: bool,
181    settled_fingerprint: &'a str,
182}
183
184fn execute_locked(
185    report: &DiagnosisReport,
186    target: &TargetContract,
187    destination: &Path,
188    workspace: &Path,
189    lock: &MigrationLock,
190    pass: &ExecutePass<'_>,
191) -> Result<RebuildOutcome, crate::MemoryError> {
192    let mut state = journal_entry(report, target, workspace, lock, pass)?;
193    let policy = match report.resolution {
194        Resolution::Reuse => VectorPolicy::Reuse,
195        Resolution::Reembed { .. } => VectorPolicy::Reembed(pass.embedder),
196        Resolution::Refuse { .. } => {
197            unreachable!("execute gated Refuse before the lock was taken")
198        }
199    };
200    let Some(source_dimension) = report.source_dimension else {
201        return Err(query_error(
202            "the source collections do not establish one shared dimension, so \
203             no AgentMemory view can open them; the diagnosis carries the \
204             details",
205        ));
206    };
207
208    let source_db = std::sync::Arc::new(Database::open(&report.source_path)?);
209    let source_memory =
210        AgentMemory::with_dimension(std::sync::Arc::clone(&source_db), source_dimension)?;
211    let destination_db = std::sync::Arc::new(Database::open(destination)?);
212    let destination_memory =
213        AgentMemory::with_dimension(std::sync::Arc::clone(&destination_db), target.dimension)?;
214
215    rebuild(
216        &RebuildSource {
217            db: &source_db,
218            memory: &source_memory,
219        },
220        &RebuildDestination {
221            db: &destination_db,
222            memory: &destination_memory,
223        },
224        &mut state,
225        &RebuildJournal { workspace, lock },
226        &policy,
227        pass.batch,
228    )
229}
230
231/// The sentence every `reembed` migration embeds once at prepare and once at
232/// every resume. Its content is arbitrary; its STABILITY is the contract —
233/// change it and every in-flight migration's witness stops matching.
234const WITNESS_SENTENCE: &str =
235    "velesdb embedder witness v1: one fixed sentence, embedded at prepare and at every resume";
236
237/// What the target embedder actually produces, as a digest — `Some` under
238/// `reembed`, `None` under `reuse` (where the embedder is never called).
239fn embedder_witness(
240    resolution: Resolution,
241    embedder: &dyn Embedder,
242) -> Result<Option<String>, crate::MemoryError> {
243    match resolution {
244        Resolution::Reuse => Ok(None),
245        Resolution::Reembed { .. } => target_embedder_witness(embedder).map(Some),
246        Resolution::Refuse { .. } => {
247            unreachable!("execute gated Refuse before the witness was computed")
248        }
249    }
250}
251
252pub(crate) fn target_embedder_witness(
253    embedder: &dyn Embedder,
254) -> Result<String, crate::MemoryError> {
255    use sha2::Digest;
256    let vector = embedder.embed(WITNESS_SENTENCE).map_err(|err| {
257        query_error(format!(
258            "the target embedder cannot embed the witness: {err}"
259        ))
260    })?;
261    let mut hash = sha2::Sha256::new();
262    for value in &vector {
263        hash.update(value.to_le_bytes());
264    }
265    Ok(format!(
266        "sha256:{}",
267        super::filesystem::encode_hex(&hash.finalize())
268    ))
269}
270
271/// Read-and-verify the existing journal, or write the first entry.
272///
273/// Both directions use the SETTLED fingerprint, never the diagnosis's: the
274/// diagnosis fingerprinted the tree before the settle compacted it. And both
275/// carry the embedder WITNESS, not just the model name: `may_resume` checks
276/// what the embedder is CALLED, the witness what it PRODUCES, and only the
277/// second survives a model updated in place under a stable name — the replayed
278/// batches would keep run-one vectors while the remaining batches got
279/// run-two's, one store with two incompatible vector spaces.
280fn journal_entry(
281    report: &DiagnosisReport,
282    target: &TargetContract,
283    workspace: &Path,
284    lock: &MigrationLock,
285    pass: &ExecutePass<'_>,
286) -> Result<MigrationState, crate::MemoryError> {
287    let witness = embedder_witness(report.resolution, pass.embedder)?;
288    if pass.resuming {
289        return resume_journal(report, target, workspace, pass, witness.as_deref());
290    }
291    let state = MigrationState {
292        format_version: super::state::STATE_FORMAT_VERSION,
293        phase: Phase::Prepared,
294        source_path: report.source_path.clone(),
295        source_fingerprint: pass.settled_fingerprint.to_owned(),
296        target_model: target.model.clone(),
297        target_dimension: target.dimension,
298        progress: super::enumeration::AGENT_COLLECTIONS
299            .iter()
300            .map(|name| {
301                (
302                    (*name).to_owned(),
303                    CollectionProgress::Facts { cursor: None },
304                )
305            })
306            .collect(),
307        embedder_witness: witness,
308    };
309    state.write(workspace, lock).map_err(query_error)?;
310    Ok(state)
311}
312
313/// Verify an existing journal against the run in front of us.
314fn resume_journal(
315    report: &DiagnosisReport,
316    target: &TargetContract,
317    workspace: &Path,
318    pass: &ExecutePass<'_>,
319    witness: Option<&str>,
320) -> Result<MigrationState, crate::MemoryError> {
321    let state = MigrationState::read(workspace)
322        .map_err(query_error)?
323        .ok_or_else(|| {
324            query_error(format!(
325                "the journal at {} disappeared between inspection and locking",
326                workspace.display()
327            ))
328        })?;
329    state
330        .may_resume(
331            &report.source_path,
332            pass.settled_fingerprint,
333            &target.model,
334            target.dimension,
335        )
336        .map_err(query_error)?;
337    if state.embedder_witness.as_deref() != witness {
338        return Err(query_error(format!(
339            "this migration was prepared with an embedder whose witness was \
340             {:?}, and the embedder answering to '{}' now produces {:?}. Same \
341             name, different vectors — the model was updated in place, or the \
342             regime changed between runs. Resuming would mix two vector spaces \
343             in one store; start a fresh migration",
344            state.embedder_witness, target.model, witness,
345        )));
346    }
347    Ok(state)
348}
349
350/// The operator's word for a regime, as they typed it on the CLI.
351fn regime_word(strategy: super::strategy::Strategy) -> &'static str {
352    match strategy {
353        super::strategy::Strategy::Auto => "auto",
354        super::strategy::Strategy::Reuse => "reuse",
355        super::strategy::Strategy::Reembed => "reembed",
356    }
357}
358
359/// The journal's home: a sibling of the destination, named after it.
360pub(crate) fn journal_workspace(destination: &Path) -> Result<PathBuf, crate::MemoryError> {
361    let name = destination
362        .file_name()
363        .and_then(|name| name.to_str())
364        .ok_or_else(|| {
365            query_error(format!(
366                "the destination {} has no usable directory name to derive the \
367                 journal workspace from",
368                destination.display()
369            ))
370        })?;
371    let workspace = destination.with_file_name(format!("{name}.migration-journal"));
372    std::fs::create_dir_all(&workspace).map_err(|err| {
373        query_error(format!(
374            "cannot create the journal workspace {}: {err}",
375            workspace.display()
376        ))
377    })?;
378    Ok(workspace)
379}
380
381/// Create the destination, or verify that what is there is ours to continue.
382fn ensure_destination(destination: &Path, resuming: bool) -> Result<(), crate::MemoryError> {
383    if !destination.exists() {
384        std::fs::create_dir_all(destination).map_err(|err| {
385            query_error(format!(
386                "cannot create the destination {}: {err}",
387                destination.display()
388            ))
389        })?;
390        return Ok(());
391    }
392    if resuming {
393        // The journal accounts for whatever the interrupted run left here, and
394        // `reinsert_batch`'s collision refusal is what protects each id.
395        return Ok(());
396    }
397    let mut entries = std::fs::read_dir(destination).map_err(|err| {
398        query_error(format!(
399            "cannot inspect the destination {}: {err}",
400            destination.display()
401        ))
402    })?;
403    if entries.next().is_some() {
404        return Err(query_error(format!(
405            "the destination {} already holds data and no migration journal \
406             accounts for it; rebuilding into it could mix two stores, so \
407             choose an empty destination or remove it deliberately",
408            destination.display()
409        )));
410    }
411    Ok(())
412}
velesdb_memory/migration/execute.rs

velesdb_memory/migration/
execute.rs
