Build a complete tenant backup envelope by fanning out across the cluster, gathering each node's slice, and framing the result. Single-node and cluster paths converge here — a single-node server produces a one-section envelope with origin = self.
(state: &Arc<SharedState>, tenant_id: u64)
| 35 | /// Single-node and cluster paths converge here — a single-node server |
| 36 | /// produces a one-section envelope with origin = self. |
| 37 | pub async fn backup_tenant(state: &Arc<SharedState>, tenant_id: u64) -> Result<Bytes, Error> { |
| 38 | let nodes = unique_origin_nodes(state); |
| 39 | let snapshot_plan = PhysicalPlan::Meta(MetaOp::CreateTenantSnapshot { tenant_id }); |
| 40 | |
| 41 | // Collect per-node sections first. The orchestrator's own |
| 42 | // dispatches advance the tenant write-HLC high-water via |
| 43 | // `dispatch_async`; capturing the envelope watermark AFTER the |
| 44 | // fan-out guarantees `envelope.watermark ≥ tenant_write_hlc` |
| 45 | // at backup time, so a subsequent restore of this envelope into |
| 46 | // the same (unchanged) cluster passes the staleness gate. |
| 47 | let mut sections = Vec::with_capacity(nodes.len()); |
| 48 | for node_id in nodes { |
| 49 | let body = if is_self(state, node_id) { |
| 50 | snapshot_self(state, tenant_id, &snapshot_plan).await? |
| 51 | } else { |
| 52 | snapshot_remote(state, node_id, tenant_id, &snapshot_plan).await? |
| 53 | }; |
| 54 | sections.push((node_id, body)); |
| 55 | } |
| 56 | |
| 57 | // Capture a cluster-wide logical instant for the envelope via the |
| 58 | // HLC. `hlc_clock.now()` advances past any previously observed |
| 59 | // local or remote HLC — the wall-ns component is the scalar |
| 60 | // watermark we stamp into the header. Restore compares this |
| 61 | // against the destination's `tenant_write_hlc` to detect stale |
| 62 | // envelopes. |
| 63 | let snapshot_watermark = state.hlc_clock.now().wall_ns; |
| 64 | let meta = EnvelopeMeta { |
| 65 | tenant_id, |
| 66 | source_vshard_count: VSHARD_COUNT as u16, |
| 67 | hash_seed: 0, // VSHARD_COUNT-derived hash; no seed today |
| 68 | snapshot_watermark, |
| 69 | }; |
| 70 | let mut writer = EnvelopeWriter::new(meta); |
| 71 | |
| 72 | for (node_id, body) in sections { |
| 73 | writer |
| 74 | .push_section(node_id, body) |
| 75 | .map_err(|e| Error::Internal { |
| 76 | detail: format!("backup envelope: {e}"), |
| 77 | })?; |
| 78 | } |
| 79 | |
| 80 | // Metadata sections: catalog rows + source-side tombstones. These |
| 81 | // live in dedicated sections with sentinel origin_node_ids so the |
| 82 | // restore path can distinguish them from per-node engine data. |
| 83 | // Without these, a backup taken during a collection's retention |
| 84 | // window loses its soft-deleted row (UNDROP can't work after |
| 85 | // restore), and a restore whose source has already purged a |
| 86 | // collection can resurrect rows that were properly reaped. |
| 87 | if let Some(catalog) = state.credentials.catalog() { |
| 88 | if let Ok(all) = catalog.load_all_collections(DatabaseId::DEFAULT) { |
| 89 | let mut blobs: Vec<nodedb_types::backup_envelope::StoredCollectionBlob> = Vec::new(); |
| 90 | for coll in all.iter().filter(|c| c.tenant_id == tenant_id) { |
| 91 | if let Ok(bytes) = zerompk::to_msgpack_vec(coll) { |
| 92 | blobs.push(nodedb_types::backup_envelope::StoredCollectionBlob { |
| 93 | name: coll.name.clone(), |
| 94 | bytes, |
no test coverage detected