linkrs_api/
migration_online.rs

1//! Online migration drain fence shared by embedded and server callers.
2//!
3//! Bridges `linkrs_transaction::issue_request_drain` into the migration
4//! engine's [`linkrs_migration::SchemaWriteFence`] hook so schema switches
5//! run inside a bounded write stall while the data backfill stays online.
6
7use std::sync::Arc;
8use std::time::Duration;
9
10use linkrs_core::event_dispatch::EventSubscriptions;
11
12/// [`linkrs_migration::SchemaWriteFence`] backed by the transaction
13/// checkpoint gate.
14pub struct CheckpointGateSchemaFence {
15    gate: Arc<linkrs_transaction::CheckpointGate>,
16    timeout: Duration,
17    held: parking_lot::Mutex<Option<linkrs_transaction::MigrationDrainFence>>,
18}
19
20impl CheckpointGateSchemaFence {
21    pub fn new(gate: Arc<linkrs_transaction::CheckpointGate>, timeout: Duration) -> Self {
22        Self {
23            gate,
24            timeout,
25            held: parking_lot::Mutex::new(None),
26        }
27    }
28}
29
30impl linkrs_migration::SchemaWriteFence for CheckpointGateSchemaFence {
31    fn hold(&self) -> Result<(), linkrs_migration::MigrationError> {
32        let fence = linkrs_transaction::maintenance::issue_request_drain(&self.gate, self.timeout)
33            .map_err(|e| linkrs_migration::MigrationError::Lock(e.to_string()))?;
34        *self.held.lock() = Some(fence);
35        Ok(())
36    }
37
38    fn release(&self) {
39        if let Some(fence) = self.held.lock().take() {
40            fence.complete();
41        }
42    }
43}
44
45/// Single online execution entry shared by embedded, HTTP and gRPC callers.
46///
47/// Holds a drain fence only across schema-modifying steps (data-only plans
48/// skip it entirely), runs the plan, and confirms the switch on success.
49/// Concurrency is guarded by the engine's own per-target lock; no separate
50/// begin guard is needed here. Metric recording and event-registry setup
51/// stay with the callers, which own their respective state sources.
52pub fn execute_online_migration<S>(
53    storage: &mut S,
54    plan: &linkrs_migration::MigrationPlan,
55    config: &linkrs_migration::MigrationConfig,
56    gate: &Arc<linkrs_transaction::CheckpointGate>,
57    event_registry: Option<&Arc<EventSubscriptions<linkrs_migration::MigrationEvent>>>,
58) -> Result<linkrs_migration::MigrationReport, linkrs_migration::MigrationError>
59where
60    S: linkrs_storage::StorageReader
61        + linkrs_storage::StorageWriter
62        + linkrs_storage::StorageSchemaOps
63        + linkrs_storage::AutoCommitGroupOps
64        + linkrs_storage::AutoCommitBatchOps
65        + ?Sized,
66{
67    let fence = if plan.has_schema_modifying_steps() {
68        Some(CheckpointGateSchemaFence::new(
69            Arc::clone(gate),
70            Duration::from_millis(config.drain_timeout_ms.max(1)),
71        ))
72    } else {
73        None
74    };
75    let report = linkrs_migration::execute_migration_plan_with_options(
76        storage,
77        plan,
78        linkrs_migration::ExecuteOptions {
79            config: Some(config),
80            event_registry,
81            schema_fence: fence
82                .as_ref()
83                .map(|f| f as &dyn linkrs_migration::SchemaWriteFence),
84            ..Default::default()
85        },
86    )?;
87    if report.success {
88        let (_, audit) =
89            linkrs_transaction::maintenance::issue_confirm_switch(plan.version_range.to);
90        log::info!("{audit}");
91    }
92    Ok(report)
93}
linkrs_api/migration_online.rs

linkrs_api/
migration_online.rs
