Method load_event_file

crates/hyperqueue/src/server/restore.rs:208–436

(&mut self, path: &Path)

    }

    pub fn load_event_file(&mut self, path: &Path) -> crate::Result<()> {
        log::debug!("Loading event file {}", path.display());
        let mut event_reader = JournalReader::open(path)?;
        for event in &mut event_reader {
            let event = event.map_err(|error| {
                crate::Error::DeserializationError(format!(
                    "Journal load error: {error:?}.\nIt appears that the journal file is corrupted."
                ))
            })?;
            match event.payload {
                EventPayload::WorkerConnected(worker_id, config) => {
                    log::debug!("Replaying: WorkerConnected {worker_id}");
                    self.max_worker_id = self.max_worker_id.max(worker_id.as_num());

                    // If we see a worker connected from an allocation, it should have occurred
                    // in the log *after* the corresponding allocation has been submitted from a
                    // queue.
                    if let Some(info) = config.get_manager_info()
                        && let Some(queue_id) = self.allocation_to_queue_id.get(&info.allocation_id)
                    {
                        self.queue_to_worker_resources
                            .insert(*queue_id, config.resources);
                    }
                }
                EventPayload::WorkerLost(worker_id, reason) => {
                    if reason.is_failure() {
                        for job in self.jobs.values_mut() {
                            job.increase_crash_counters(worker_id);
                        }
                    }
                }
                EventPayload::WorkerOverviewReceived(_) => {}
                EventPayload::Submit {
                    job_id,
                    closed_job,
                    serialized_desc,
                } => {
                    log::debug!("Replaying: JobTasksCreated {job_id}");
                    let submit_request: SubmitRequest = serialized_desc.deserialize()?;
                    if closed_job {
                        let mut job = RestorerJob::new(submit_request.job_desc, false);
                        job.add_submit(SubmittedJobDescription::at(
                            event.time,
                            submit_request.submit_desc,
                        ));
                        self.add_job(job_id, job);
                    } else if let Some(job) = self.get_job_mut(job_id) {
                        job.add_submit(SubmittedJobDescription::at(
                            event.time,
                            submit_request.submit_desc,
                        ));
                    } else {
                        log::warn!("Ignoring submit attachment to an non-existing job")
                    }
                }
                EventPayload::JobCompleted(job_id) => {
                    log::debug!("Replaying: JobCompleted {job_id}");
                    self.jobs.remove(&job_id);