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 { .. } => {
246            use sha2::Digest;
247            let vector = embedder.embed(WITNESS_SENTENCE).map_err(|err| {
248                query_error(format!(
249                    "the target embedder cannot embed the witness: {err}"
250                ))
251            })?;
252            let mut hash = sha2::Sha256::new();
253            for value in &vector {
254                hash.update(value.to_le_bytes());
255            }
256            Ok(Some(format!(
257                "sha256:{}",
258                super::filesystem::encode_hex(&hash.finalize())
259            )))
260        }
261        Resolution::Refuse { .. } => {
262            unreachable!("execute gated Refuse before the witness was computed")
263        }
264    }
265}
266
267/// Read-and-verify the existing journal, or write the first entry.
268///
269/// Both directions use the SETTLED fingerprint, never the diagnosis's: the
270/// diagnosis fingerprinted the tree before the settle compacted it. And both
271/// carry the embedder WITNESS, not just the model name: `may_resume` checks
272/// what the embedder is CALLED, the witness what it PRODUCES, and only the
273/// second survives a model updated in place under a stable name — the replayed
274/// batches would keep run-one vectors while the remaining batches got
275/// run-two's, one store with two incompatible vector spaces.
276fn journal_entry(
277    report: &DiagnosisReport,
278    target: &TargetContract,
279    workspace: &Path,
280    lock: &MigrationLock,
281    pass: &ExecutePass<'_>,
282) -> Result<MigrationState, crate::MemoryError> {
283    let witness = embedder_witness(report.resolution, pass.embedder)?;
284    if pass.resuming {
285        return resume_journal(report, target, workspace, pass, witness.as_deref());
286    }
287    let state = MigrationState {
288        format_version: super::state::STATE_FORMAT_VERSION,
289        phase: Phase::Prepared,
290        source_path: report.source_path.clone(),
291        source_fingerprint: pass.settled_fingerprint.to_owned(),
292        target_model: target.model.clone(),
293        target_dimension: target.dimension,
294        progress: super::enumeration::AGENT_COLLECTIONS
295            .iter()
296            .map(|name| {
297                (
298                    (*name).to_owned(),
299                    CollectionProgress::Facts { cursor: None },
300                )
301            })
302            .collect(),
303        embedder_witness: witness,
304    };
305    state.write(workspace, lock).map_err(query_error)?;
306    Ok(state)
307}
308
309/// Verify an existing journal against the run in front of us.
310fn resume_journal(
311    report: &DiagnosisReport,
312    target: &TargetContract,
313    workspace: &Path,
314    pass: &ExecutePass<'_>,
315    witness: Option<&str>,
316) -> Result<MigrationState, crate::MemoryError> {
317    let state = MigrationState::read(workspace)
318        .map_err(query_error)?
319        .ok_or_else(|| {
320            query_error(format!(
321                "the journal at {} disappeared between inspection and locking",
322                workspace.display()
323            ))
324        })?;
325    state
326        .may_resume(
327            &report.source_path,
328            pass.settled_fingerprint,
329            &target.model,
330            target.dimension,
331        )
332        .map_err(query_error)?;
333    if state.embedder_witness.as_deref() != witness {
334        return Err(query_error(format!(
335            "this migration was prepared with an embedder whose witness was \
336             {:?}, and the embedder answering to '{}' now produces {:?}. Same \
337             name, different vectors — the model was updated in place, or the \
338             regime changed between runs. Resuming would mix two vector spaces \
339             in one store; start a fresh migration",
340            state.embedder_witness, target.model, witness,
341        )));
342    }
343    Ok(state)
344}
345
346/// The operator's word for a regime, as they typed it on the CLI.
347fn regime_word(strategy: super::strategy::Strategy) -> &'static str {
348    match strategy {
349        super::strategy::Strategy::Auto => "auto",
350        super::strategy::Strategy::Reuse => "reuse",
351        super::strategy::Strategy::Reembed => "reembed",
352    }
353}
354
355/// The journal's home: a sibling of the destination, named after it.
356pub(super) fn journal_workspace(destination: &Path) -> Result<PathBuf, crate::MemoryError> {
357    let name = destination
358        .file_name()
359        .and_then(|name| name.to_str())
360        .ok_or_else(|| {
361            query_error(format!(
362                "the destination {} has no usable directory name to derive the \
363                 journal workspace from",
364                destination.display()
365            ))
366        })?;
367    let workspace = destination.with_file_name(format!("{name}.migration-journal"));
368    std::fs::create_dir_all(&workspace).map_err(|err| {
369        query_error(format!(
370            "cannot create the journal workspace {}: {err}",
371            workspace.display()
372        ))
373    })?;
374    Ok(workspace)
375}
376
377/// Create the destination, or verify that what is there is ours to continue.
378fn ensure_destination(destination: &Path, resuming: bool) -> Result<(), crate::MemoryError> {
379    if !destination.exists() {
380        std::fs::create_dir_all(destination).map_err(|err| {
381            query_error(format!(
382                "cannot create the destination {}: {err}",
383                destination.display()
384            ))
385        })?;
386        return Ok(());
387    }
388    if resuming {
389        // The journal accounts for whatever the interrupted run left here, and
390        // `reinsert_batch`'s collision refusal is what protects each id.
391        return Ok(());
392    }
393    let mut entries = std::fs::read_dir(destination).map_err(|err| {
394        query_error(format!(
395            "cannot inspect the destination {}: {err}",
396            destination.display()
397        ))
398    })?;
399    if entries.next().is_some() {
400        return Err(query_error(format!(
401            "the destination {} already holds data and no migration journal \
402             accounts for it; rebuilding into it could mix two stores, so \
403             choose an empty destination or remove it deliberately",
404            destination.display()
405        )));
406    }
407    Ok(())
408}
velesdb_memory/migration/execute.rs

velesdb_memory/migration/
execute.rs
