(
&self,
database: Database,
host_type: HostType,
replica_id: u64,
program_bytes: Box<[u8]>,
policy: MigrationPolicy,
)
| 389 | /// the host keeps running. |
| 390 | #[tracing::instrument(level = "trace", skip_all, err)] |
| 391 | pub async fn update_module_host( |
| 392 | &self, |
| 393 | database: Database, |
| 394 | host_type: HostType, |
| 395 | replica_id: u64, |
| 396 | program_bytes: Box<[u8]>, |
| 397 | policy: MigrationPolicy, |
| 398 | ) -> anyhow::Result<UpdateDatabaseResult> { |
| 399 | let program = Program::from_bytes(host_type.into(), program_bytes); |
| 400 | trace!( |
| 401 | "update module host {}/{}: genesis={} update-to={}", |
| 402 | database.database_identity, |
| 403 | replica_id, |
| 404 | database.initial_program, |
| 405 | program.hash |
| 406 | ); |
| 407 | |
| 408 | let Ok(mut guard) = self.acquire_write_lock(replica_id).await else { |
| 409 | bail!("unable to lock database {} for update", database.database_identity); |
| 410 | }; |
| 411 | |
| 412 | // `HostController::clone` is fast, |
| 413 | // as all of its fields are either `Copy` or wrapped in `Arc`. |
| 414 | let this = self.clone(); |
| 415 | |
| 416 | // `try_init_host` is not cancel safe, as it will spawn other async tasks |
| 417 | // which hold a filesystem lock past when `try_init_host` returns or is cancelled. |
| 418 | // This means that, if `try_init_host` is cancelled, subsequent calls will fail. |
| 419 | // |
| 420 | // The rest of this future is also not cancel safe, as it will `Option::take` out of the guard |
| 421 | // at the start of the block and then store back into it at the end. |
| 422 | // |
| 423 | // This is problematic because Axum will cancel its handler tasks if the client disconnects, |
| 424 | // and this method is called from Axum handlers, e.g. for the publish route. |
| 425 | // `tokio::spawn` a task to update the contents of `guard`, |
| 426 | // so that it will run to completion even if the caller goes away. |
| 427 | // |
| 428 | // Note that `tokio::spawn` only cancels its tasks when the runtime shuts down, |
| 429 | // at which point we won't be calling `try_init_host` again anyways. |
| 430 | let update_result = tokio::spawn(async move { |
| 431 | let mut host = match guard.take() { |
| 432 | None => { |
| 433 | trace!("host not running, try_init"); |
| 434 | this.try_init_host(database, replica_id).await? |
| 435 | } |
| 436 | Some(host) => { |
| 437 | trace!("host found, updating"); |
| 438 | host |
| 439 | } |
| 440 | }; |
| 441 | let update_result = host |
| 442 | .update_module( |
| 443 | this.runtimes.clone(), |
| 444 | program, |
| 445 | policy, |
| 446 | this.energy_monitor.clone(), |
| 447 | this.unregister_fn(replica_id), |
| 448 | this.db_cores.take(), |
no test coverage detected