Perform a copy-up: write `source_row_bytes` into target storage with a fresh surrogate and record the mapping in `clone_copyups`. Returns the fresh target surrogate so the caller can apply the pending UPDATE to it.
(params: CopyUpParams<'_>)
| 141 | /// Returns the fresh target surrogate so the caller can apply the pending |
| 142 | /// UPDATE to it. |
| 143 | pub async fn perform_clone_copyup(params: CopyUpParams<'_>) -> crate::Result<Surrogate> { |
| 144 | let CopyUpParams { |
| 145 | state, |
| 146 | tenant_id, |
| 147 | target_db_id, |
| 148 | target_collection, |
| 149 | origin, |
| 150 | source_surrogate, |
| 151 | source_doc_id, |
| 152 | source_row_bytes, |
| 153 | } = params; |
| 154 | |
| 155 | // Allocate a fresh target surrogate using the (collection, doc_id) key. |
| 156 | let target_coll_qualified = crate::control::planner::sql_plan_convert::convert::db_qualified( |
| 157 | target_db_id, |
| 158 | target_collection, |
| 159 | ); |
| 160 | let target_surrogate = state |
| 161 | .surrogate_assigner |
| 162 | .assign(&target_coll_qualified, source_doc_id.as_bytes()) |
| 163 | .map_err(|e| crate::Error::Storage { |
| 164 | engine: "clone_copyup".into(), |
| 165 | detail: format!("surrogate alloc failed: {e}"), |
| 166 | })?; |
| 167 | |
| 168 | // Catalog mapping is recorded BEFORE the KV put so that the catalog is |
| 169 | // always at least as informed as target storage. If the put fails we then |
| 170 | // compensate by removing the just-written mapping; this guarantees we |
| 171 | // never end up with an orphaned row in target (row present, no mapping) |
| 172 | // which would later cause duplicate surrogates on retry. |
| 173 | let catalog_arc = state.credentials.catalog(); |
| 174 | let catalog = catalog_arc.as_ref().ok_or(crate::Error::Storage { |
| 175 | engine: "clone_copyup".into(), |
| 176 | detail: "catalog unavailable".into(), |
| 177 | })?; |
| 178 | |
| 179 | let source_coll_qualified = crate::control::planner::sql_plan_convert::convert::db_qualified( |
| 180 | origin.source_database, |
| 181 | &origin.source_collection, |
| 182 | ); |
| 183 | catalog |
| 184 | .put_clone_copyup( |
| 185 | &source_coll_qualified, |
| 186 | source_surrogate.as_u32(), |
| 187 | target_surrogate.as_u32(), |
| 188 | ) |
| 189 | .map_err(|e| crate::Error::Storage { |
| 190 | engine: "clone_copyup".into(), |
| 191 | detail: format!("put_clone_copyup catalog write failed: {e}"), |
| 192 | })?; |
| 193 | |
| 194 | // Helper: roll the catalog mapping back if any subsequent step fails. |
| 195 | // Logged-but-not-fatal if the rollback itself fails — the mapping then |
| 196 | // points at a target row that does not exist; the read path treats a |
| 197 | // missing target row as "fall through to source", which is the same |
| 198 | // semantics as having no mapping at all. |
| 199 | let rollback_mapping = |reason: &str| { |
| 200 | if let Err(e) = |
no test coverage detected