| 980 | } |
| 981 | |
| 982 | pub(super) async fn run(mut self) { |
| 983 | self.send(ComputeCommand::Hello { |
| 984 | // The nonce is protocol iteration-specific and will be set in |
| 985 | // `ReplicaTask::specialize_command`. |
| 986 | nonce: Uuid::default(), |
| 987 | }); |
| 988 | |
| 989 | let instance_config = InstanceConfig { |
| 990 | peek_stash_persist_location: self.peek_stash_persist_location.clone(), |
| 991 | // The remaining fields are replica-specific and will be set in |
| 992 | // `ReplicaTask::specialize_command` (logging, expiration, dictionary compression) and |
| 993 | // `Instance::specialize_command_for_replica` (the initial config snapshot). |
| 994 | logging: Default::default(), |
| 995 | expiration_offset: Default::default(), |
| 996 | arrangement_dictionary_compression: Default::default(), |
| 997 | initial_config: Default::default(), |
| 998 | }; |
| 999 | |
| 1000 | self.send(ComputeCommand::CreateInstance(Box::new(instance_config))); |
| 1001 | |
| 1002 | loop { |
| 1003 | tokio::select! { |
| 1004 | command = self.command_rx.recv() => match command { |
| 1005 | Some(cmd) => cmd(&mut self), |
| 1006 | None => break, |
| 1007 | }, |
| 1008 | response = self.replica_rx.recv() => match response { |
| 1009 | Some(response) => self.handle_response(response), |
| 1010 | None => unreachable!("self owns a sender side of the channel"), |
| 1011 | } |
| 1012 | } |
| 1013 | } |
| 1014 | } |
| 1015 | |
| 1016 | /// Update instance configuration. |
| 1017 | #[mz_ore::instrument(level = "debug")] |