velesdb_memory/migration/
state.rs

1use serde_json::Value;
2use std::io::Write;
3use std::path::{Path, PathBuf};
4
5mod lock;
6mod resume;
7mod switch;
8
9#[cfg(test)]
10pub(super) use lock::LOCK_GUARD_FILE;
11pub use lock::{MigrationLock, LOCK_FILE};
12pub use switch::{Phase, Recovery, SwitchState, PHASES};
13
14// ---------------------------------------------------------------------------
15// THE LOCK AND THE PHASE JOURNAL
16//
17// A rebuild is a sequence that can stop anywhere: between reading and writing,
18// between writing and validating, between archiving the source and activating
19// the destination. What matters is not that it never stops — it is that every
20// place it CAN stop has one defined action, and that an ambiguous stop changes
21// nothing at all.
22// ---------------------------------------------------------------------------
23
24/// The file a prepared migration records its state in.
25pub const STATE_FILE: &str = "migration-state.json";
26
27/// The fixed sibling staging file for an atomic state replacement.
28///
29/// Its presence is ambiguous evidence of an interrupted writer, so it is
30/// never overwritten or silently swept by a later run.
31pub const STATE_TEMP_FILE: &str = "migration-state.json.tmp";
32
33/// The shape of a [`MigrationState`].
34///
35/// Bumped when the state's meaning changes. Only the current version may
36/// resume: a newer state may contain unknown decisions, while an older one may
37/// rely on guarantees this build deliberately strengthened.
38///
39/// # v3 — per-collection rebuild progress (#1762, PR C2b)
40///
41/// v2 recorded the phase and nothing else, so a resumed rebuild had to start
42/// every collection from zero. v3 adds [`MigrationState::progress`], and with
43/// it two rules a v2 build never enforced: progress can only advance, and the
44/// phase cannot leave [`Phase::Prepared`] while any collection is unfinished.
45pub const STATE_FORMAT_VERSION: u32 = 3;
46
47/// How far one collection's rebuild got inside [`Phase::Prepared`].
48///
49/// Three stages, because a resume needs to answer three different questions.
50/// `Facts` says where the cursor walk stands — `cursor` is the last fact id
51/// reinserted, and `None` means the walk has not started. `Edges` says every
52/// fact landed and the edge pass is running; it carries no cursor because the
53/// pass is idempotent end to end (reinserting an existing edge answers with
54/// the same id, and the destination is verified by re-reading, so replaying it
55/// after a crash is safe where replaying half a fact walk would not be).
56/// `Complete` says both are done and a resume must not touch the collection.
57#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
58#[serde(tag = "stage", rename_all = "snake_case")]
59pub enum CollectionProgress {
60    /// The fact walk is under way, resumable strictly after `cursor`.
61    Facts {
62        /// The last fact id reinserted at the destination; `None` = not started.
63        cursor: Option<u64>,
64    },
65    /// Every fact landed; the (idempotent) edge pass runs until `Complete`.
66    Edges,
67    /// Facts and edges are both at the destination.
68    Complete,
69}
70
71impl CollectionProgress {
72    /// Whether `self` may be recorded after `previous` for one collection.
73    ///
74    /// Exactly the transitions the pass emits, and no others. An earlier
75    /// version tolerated skipping `Edges` ("a collection with no edges goes
76    /// straight to Complete") — which was FALSE: the pass journals `Edges`
77    /// unconditionally, edges or none. A tolerance the writer never uses only
78    /// widens what a BUGGY writer can make the journal swallow — a refactor
79    /// that lost the edge pass would have journalled `Complete` without a
80    /// sound. Requiring the passage through `Edges` costs the real writer
81    /// nothing and makes that bug un-journallable.
82    fn may_follow(self, previous: Self) -> bool {
83        match (previous, self) {
84            (Self::Facts { cursor: before }, Self::Facts { cursor: after }) => {
85                match (before, after) {
86                    (Some(before), Some(after)) => after >= before,
87                    (Some(_), None) => false,
88                    (None, _) => true,
89                }
90            }
91            (Self::Facts { .. } | Self::Edges, Self::Edges)
92            | (Self::Edges | Self::Complete, Self::Complete) => true,
93            _ => false,
94        }
95    }
96}
97
98/// What a prepared migration recorded, so a later run can decide whether to
99/// resume it.
100///
101/// Emphatically not a [`crate::migration::DiagnosisReport`]: a report answers
102/// "what is here", a
103/// state asserts "a migration is under way and got this far". A diagnosis never
104/// produces one.
105#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
106pub struct MigrationState {
107    /// The shape of this state — see [`STATE_FORMAT_VERSION`].
108    pub format_version: u32,
109    /// How far the migration got.
110    pub phase: Phase,
111    /// The store being migrated.
112    pub source_path: PathBuf,
113    /// The source's fingerprint when the migration was prepared.
114    pub source_fingerprint: String,
115    /// The model the migration is rebuilding against.
116    pub target_model: String,
117    /// The width that model produces.
118    pub target_dimension: usize,
119    /// How far each collection's rebuild got — exactly one entry per agent
120    /// collection, always. A missing key would silently skip a collection's
121    /// rebuild; an extra one would journal work nobody will do. Both are
122    /// refused by validation rather than tolerated.
123    pub progress: std::collections::BTreeMap<String, CollectionProgress>,
124    /// A digest of what the target embedder actually PRODUCES, not what it is
125    /// called — `sha256:` over the vector it answers for a fixed sentinel
126    /// sentence. `Some` exactly when the resolved regime is `reembed`, `None`
127    /// under `reuse`, so the field also records the regime without a second
128    /// field to drift from it.
129    ///
130    /// This exists because [`MigrationState::may_resume`]'s model check
131    /// compares NAMES, and a name is a claim: `ollama pull` updates a model's
132    /// weights in place under the same identifier. A run resumed across such
133    /// an update would collide its replayed batch into run-one vectors and
134    /// write run-two vectors after it — one store, one recorded model, two
135    /// incompatible vector spaces, which is exactly what the model check says
136    /// it prevents. Vectors do not lie about the embedder that made them.
137    pub embedder_witness: Option<String>,
138}
139
140impl MigrationState {
141    /// Whether this state may be resumed against the source and target now in
142    /// front of us.
143    ///
144    /// Every refusal names both sides. A resume that silently adapted to a
145    /// changed fingerprint would rebuild from a store that is no longer the one
146    /// it inventoried; one that adapted to a changed model would produce a
147    /// store whose vectors and whose recorded model disagree.
148    ///
149    /// # Errors
150    /// A message naming what changed and what the operator can do about it.
151    pub fn may_resume(
152        &self,
153        source_path: &Path,
154        source_fingerprint: &str,
155        target_model: &str,
156        target_dimension: usize,
157    ) -> Result<(), String> {
158        validate_current_state_version(self)?;
159        validate_state_semantics(self)?;
160        validate_migration_identity(
161            source_path,
162            source_fingerprint,
163            target_model,
164            target_dimension,
165        )
166        .map_err(|reason| {
167            format!("cannot resume against an invalid requested identity: {reason}")
168        })?;
169        resume::validate_resume_source(self, source_path)?;
170        resume::validate_resume_fingerprint(self, source_fingerprint)?;
171        resume::validate_resume_model(self, target_model)?;
172        resume::validate_resume_dimension(self, target_dimension)?;
173        Ok(())
174    }
175
176    /// Read a state from `workspace`, refusing one this build cannot act on.
177    ///
178    /// The version is read out of the raw JSON BEFORE the state is
179    /// deserialised, because a newer state may carry fields this build cannot
180    /// parse — and "cannot parse" would otherwise surface as a corruption error
181    /// instead of the version refusal it actually is.
182    ///
183    /// # Errors
184    /// The file is unreadable, is not JSON, or is stamped with a version newer
185    /// from [`STATE_FORMAT_VERSION`].
186    pub fn read(workspace: &Path) -> Result<Option<Self>, String> {
187        let Some(value) = read_state_value(workspace)? else {
188            return Ok(None);
189        };
190        let version = serialized_state_version(&value)?;
191        validate_serialized_state_version(version)?;
192        let state: Self = serde_json::from_value(value).map_err(|err| {
193            format!("{STATE_FILE} is version {version} but does not parse: {err}")
194        })?;
195        validate_state_semantics(&state)
196            .map_err(|reason| format!("{STATE_FILE} has invalid semantics: {reason}"))?;
197        Ok(Some(state))
198    }
199
200    /// Atomically and durably replace the state in `workspace`.
201    ///
202    /// The caller must hold `lock`. The complete JSON is written to a fixed
203    /// sibling staging file, flushed and synced before one atomic promotion.
204    /// The promotion is then made durable with the platform's directory or
205    /// write-through barrier. A pre-existing staging file is refused as
206    /// evidence of an interrupted writer; it is never overwritten.
207    ///
208    /// # Errors
209    /// The lock does not guard this workspace, an existing state is invalid,
210    /// staging is ambiguous, or any write/durability step fails.
211    pub fn write(&self, workspace: &Path, lock: &MigrationLock) -> Result<(), String> {
212        lock.verify_workspace(workspace)?;
213        validate_current_state_version(self)?;
214        validate_state_semantics(self)?;
215        let existing = validate_existing_state(workspace)?;
216        validate_state_update(existing.as_ref(), self)?;
217        let body = serde_json::to_string_pretty(self)
218            .map_err(|err| format!("cannot serialise the migration state: {err}"))?;
219        // Re-check ownership immediately before the first mutation. Validation
220        // above can be arbitrarily slow on a hostile filesystem; a stale
221        // handle must not create even the staging file after an ABA replacement.
222        lock.verify_workspace(workspace)?;
223        commit_state_with(
224            workspace,
225            body.as_bytes(),
226            promote_state,
227            state_durability_barrier,
228        )
229    }
230}
231
232fn read_state_value(workspace: &Path) -> Result<Option<Value>, String> {
233    let path = workspace.join(STATE_FILE);
234    let raw = match std::fs::read_to_string(path) {
235        Ok(raw) => raw,
236        Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None),
237        Err(err) => return Err(format!("cannot read {STATE_FILE}: {err}")),
238    };
239    serde_json::from_str(&raw)
240        .map(Some)
241        .map_err(|err| format!("{STATE_FILE} is not readable JSON: {err}"))
242}
243
244fn serialized_state_version(value: &Value) -> Result<u64, String> {
245    value
246        .get("format_version")
247        .and_then(Value::as_u64)
248        .ok_or_else(|| format!("{STATE_FILE} carries no format_version"))
249}
250
251fn validate_serialized_state_version(version: u64) -> Result<(), String> {
252    if version == u64::from(STATE_FORMAT_VERSION) {
253        return Ok(());
254    }
255    let action = if version < u64::from(STATE_FORMAT_VERSION) {
256        "This older state predates per-collection rebuild progress; start a fresh diagnosis."
257    } else {
258        "Use the version that wrote it."
259    };
260    Err(format!(
261        "{STATE_FILE} is version {version} and this build requires version {STATE_FORMAT_VERSION}. Refusing incompatible migration semantics. {action}"
262    ))
263}
264
265fn validate_current_state_version(state: &MigrationState) -> Result<(), String> {
266    if state.format_version == STATE_FORMAT_VERSION {
267        return Ok(());
268    }
269    let action = if state.format_version < STATE_FORMAT_VERSION {
270        "This older state predates per-collection rebuild progress. Start a fresh diagnosis."
271    } else {
272        "Use the version that wrote it."
273    };
274    Err(format!(
275        "this migration state is version {} and this build requires version {}. \
276         Resuming across incompatible state semantics is unsafe. {action}",
277        state.format_version, STATE_FORMAT_VERSION,
278    ))
279}
280
281fn validate_state_semantics(state: &MigrationState) -> Result<(), String> {
282    validate_migration_identity(
283        &state.source_path,
284        &state.source_fingerprint,
285        &state.target_model,
286        state.target_dimension,
287    )?;
288    validate_progress_keys(state)?;
289    validate_phase_against_progress(state)?;
290    validate_embedder_witness(state)
291}
292
293/// A present witness must be a well-formed digest, not free text an editor
294/// could plausibly have typed.
295fn validate_embedder_witness(state: &MigrationState) -> Result<(), String> {
296    let Some(witness) = &state.embedder_witness else {
297        return Ok(());
298    };
299    let digest = witness
300        .strip_prefix("sha256:")
301        .filter(|digest| digest.len() == 64)
302        .filter(|digest| {
303            digest
304                .bytes()
305                .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
306        });
307    if digest.is_none() {
308        return Err(
309            "embedder_witness must be exactly 'sha256:' followed by 64 lowercase hexadecimal \
310             characters"
311                .to_owned(),
312        );
313    }
314    Ok(())
315}
316
317/// The progress map must cover exactly the agent collections.
318fn validate_progress_keys(state: &MigrationState) -> Result<(), String> {
319    for name in super::enumeration::AGENT_COLLECTIONS {
320        if !state.progress.contains_key(*name) {
321            return Err(format!(
322                "progress carries no entry for collection '{name}'; a resume would \
323                 silently skip its rebuild"
324            ));
325        }
326    }
327    for name in state.progress.keys() {
328        if !super::enumeration::AGENT_COLLECTIONS.contains(&name.as_str()) {
329            return Err(format!(
330                "progress tracks '{name}', which is not an agent collection; this \
331                 journal describes work nobody will do"
332            ));
333        }
334    }
335    Ok(())
336}
337
338/// A phase past [`Phase::Prepared`] asserts the rebuild is finished, so every
339/// collection must say so too — otherwise the journal contradicts itself and
340/// a later phase would validate, archive, or activate a half-built store.
341fn validate_phase_against_progress(state: &MigrationState) -> Result<(), String> {
342    if state.phase == Phase::Prepared {
343        return Ok(());
344    }
345    for (name, progress) in &state.progress {
346        if *progress != CollectionProgress::Complete {
347            return Err(format!(
348                "phase {:?} asserts the rebuild is finished, but collection \
349                 '{name}' stands at {progress:?}; the phase cannot leave \
350                 {:?} while any collection is unfinished",
351                state.phase,
352                Phase::Prepared,
353            ));
354        }
355    }
356    Ok(())
357}
358
359fn validate_migration_identity(
360    source_path: &Path,
361    source_fingerprint: &str,
362    target_model: &str,
363    target_dimension: usize,
364) -> Result<(), String> {
365    if !source_path.is_absolute()
366        || source_path
367            .components()
368            .any(|component| matches!(component, std::path::Component::ParentDir))
369    {
370        return Err(
371            "source_path must be an absolute normalized path produced by diagnosis".to_owned(),
372        );
373    }
374    let digest = source_fingerprint
375        .strip_prefix("sha256-tree-v2:")
376        .filter(|digest| digest.len() == 64)
377        .filter(|digest| {
378            digest
379                .bytes()
380                .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
381        });
382    if digest.is_none() {
383        return Err(
384            "source_fingerprint must be exactly 'sha256-tree-v2:' followed by 64 lowercase hexadecimal characters"
385                .to_owned(),
386        );
387    }
388    if target_model.trim().is_empty() {
389        return Err("target_model must not be empty".to_owned());
390    }
391    if target_dimension == 0 {
392        return Err("target_dimension must be greater than zero".to_owned());
393    }
394    Ok(())
395}
396
397fn validate_state_update(
398    existing: Option<&MigrationState>,
399    candidate: &MigrationState,
400) -> Result<(), String> {
401    let Some(existing) = existing else {
402        if candidate.phase != Phase::Prepared {
403            return Err(format!(
404                "a new migration journal must start at {:?}, not {:?}; refusing to invent skipped work",
405                Phase::Prepared,
406                candidate.phase
407            ));
408        }
409        return Ok(());
410    };
411
412    let immutable_drift = if existing.source_path != candidate.source_path {
413        Some(format!(
414            "source_path changed from '{}' to '{}'",
415            existing.source_path.display(),
416            candidate.source_path.display()
417        ))
418    } else if existing.source_fingerprint != candidate.source_fingerprint {
419        Some(format!(
420            "source_fingerprint changed from '{}' to '{}'",
421            existing.source_fingerprint, candidate.source_fingerprint
422        ))
423    } else if existing.target_model != candidate.target_model {
424        Some(format!(
425            "target_model changed from '{}' to '{}'",
426            existing.target_model, candidate.target_model
427        ))
428    } else if existing.target_dimension != candidate.target_dimension {
429        Some(format!(
430            "target_dimension changed from {} to {}",
431            existing.target_dimension, candidate.target_dimension
432        ))
433    } else if existing.embedder_witness != candidate.embedder_witness {
434        Some(format!(
435            "embedder_witness changed from {:?} to {:?} — either the embedder's \
436             output drifted under a stable model name, or the resolved regime \
437             flipped between runs; both make the replayed and the remaining \
438             batches incompatible",
439            existing.embedder_witness, candidate.embedder_witness
440        ))
441    } else {
442        None
443    };
444    if let Some(drift) = immutable_drift {
445        return Err(format!(
446            "refusing to rewrite migration identity: {drift}. Start a fresh migration instead"
447        ));
448    }
449    if !candidate.phase.may_follow(existing.phase) {
450        return Err(format!(
451            "refusing migration phase transition from {:?} to {:?}: journal updates may be idempotent or advance exactly one phase, never regress or skip work",
452            existing.phase, candidate.phase
453        ));
454    }
455    validate_progress_advance(existing, candidate)
456}
457
458/// Progress may repeat or advance per collection, never regress.
459///
460/// Key equality between the two maps is already established: both states
461/// passed [`validate_progress_keys`] before reaching this comparison.
462fn validate_progress_advance(
463    existing: &MigrationState,
464    candidate: &MigrationState,
465) -> Result<(), String> {
466    for (name, after) in &candidate.progress {
467        let Some(before) = existing.progress.get(name) else {
468            continue;
469        };
470        if !after.may_follow(*before) {
471            return Err(format!(
472                "refusing progress regression on collection '{name}': the journal \
473                 records {before:?} and the update asserts {after:?}; a rebuild \
474                 journal may repeat or advance, never regress"
475            ));
476        }
477    }
478    Ok(())
479}
480
481fn validate_existing_state(workspace: &Path) -> Result<Option<MigrationState>, String> {
482    let path = workspace.join(STATE_FILE);
483    let metadata = match std::fs::symlink_metadata(&path) {
484        Ok(metadata) => metadata,
485        Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None),
486        Err(err) => return Err(format!("cannot inspect existing {STATE_FILE}: {err}")),
487    };
488    if metadata.file_type().is_symlink() || !metadata.is_file() {
489        return Err(format!(
490            "refusing to replace {STATE_FILE}: {} is a symlink, directory, or special file",
491            path.display()
492        ));
493    }
494    MigrationState::read(workspace)?
495        .map(Some)
496        .ok_or_else(|| format!("{STATE_FILE} disappeared while it was being validated"))
497}
498
499pub(super) fn commit_state_with<P, B>(
500    workspace: &Path,
501    body: &[u8],
502    promote: P,
503    durability_barrier: B,
504) -> Result<(), String>
505where
506    P: FnOnce(&Path, &Path) -> std::io::Result<()>,
507    B: FnOnce(&Path, &Path) -> std::io::Result<()>,
508{
509    let temporary = workspace.join(STATE_TEMP_FILE);
510    let final_path = workspace.join(STATE_FILE);
511    let mut file = match std::fs::OpenOptions::new()
512        .write(true)
513        .create_new(true)
514        .open(&temporary)
515    {
516        Ok(file) => file,
517        Err(err) if err.kind() == std::io::ErrorKind::AlreadyExists => {
518            return Err(format!(
519                "refusing to overwrite pre-existing {STATE_TEMP_FILE} at {}: it may be evidence of an interrupted state write; inspect and remove that exact file manually",
520                temporary.display()
521            ));
522        }
523        Err(err) => return Err(format!("cannot create {STATE_TEMP_FILE}: {err}")),
524    };
525
526    let write_result = (|| {
527        file.write_all(body)?;
528        file.flush()?;
529        file.sync_all()
530    })();
531    drop(file);
532    if let Err(err) = write_result {
533        return cleanup_uncommitted_temp(
534            &temporary,
535            format!("cannot write and sync {STATE_TEMP_FILE}: {err}"),
536        );
537    }
538
539    if let Err(err) = promote(&temporary, &final_path) {
540        return cleanup_uncommitted_temp(
541            &temporary,
542            format!("cannot atomically promote {STATE_TEMP_FILE} to {STATE_FILE}: {err}"),
543        );
544    }
545    durability_barrier(workspace, &final_path).map_err(|err| {
546        format!(
547            "{STATE_FILE} was replaced and is visible, but its durability could not be confirmed: {err}. Do not retry blindly; inspect the state before continuing"
548        )
549    })
550}
551
552fn cleanup_uncommitted_temp(temporary: &Path, primary: String) -> Result<(), String> {
553    match std::fs::remove_file(temporary) {
554        Ok(()) => Err(primary),
555        Err(err) if err.kind() == std::io::ErrorKind::NotFound => Err(primary),
556        Err(err) => Err(format!(
557            "{primary}; additionally, cannot remove {}: {err}",
558            temporary.display()
559        )),
560    }
561}
562
563#[cfg(unix)]
564fn promote_state(temporary: &Path, final_path: &Path) -> std::io::Result<()> {
565    // Keep promotion and the directory durability barrier as distinct failure
566    // domains: after this succeeds, a barrier error must report that the new
567    // state is already visible and must not be retried blindly.
568    std::fs::rename(temporary, final_path)
569}
570
571#[cfg(windows)]
572fn promote_state(temporary: &Path, final_path: &Path) -> std::io::Result<()> {
573    atomicwrites::replace_atomic(temporary, final_path)
574}
575
576#[cfg(not(any(unix, windows)))]
577fn promote_state(_temporary: &Path, _final_path: &Path) -> std::io::Result<()> {
578    Err(std::io::Error::new(
579        std::io::ErrorKind::Unsupported,
580        "durable migration-state replacement is supported only on Unix and Windows",
581    ))
582}
583
584#[cfg(unix)]
585fn state_durability_barrier(workspace: &Path, _final_path: &Path) -> std::io::Result<()> {
586    std::fs::File::open(workspace)?.sync_all()
587}
588
589#[cfg(windows)]
590fn state_durability_barrier(_workspace: &Path, _final_path: &Path) -> std::io::Result<()> {
591    // `atomicwrites::replace_atomic` uses MOVEFILE_WRITE_THROUGH on Windows.
592    Ok(())
593}
594
595#[cfg(not(any(unix, windows)))]
596fn state_durability_barrier(_workspace: &Path, _final_path: &Path) -> std::io::Result<()> {
597    Err(std::io::Error::new(
598        std::io::ErrorKind::Unsupported,
599        "no durable migration-state barrier is defined for this platform",
600    ))
601}
velesdb_memory/migration/state.rs

velesdb_memory/migration/
state.rs
