Coverage / crates/flow/src/images.rs 31.8%

    fn fail(url: &str, e: ureq::Error) -> anyhow::Error {

        match e {

            ureq::Error::Status(code, r) => {

                let text = r.into_string().unwrap_or_default();

                let detail = serde_json::from_str::<serde_json::Value>(&text)

                    .ok()

                    .and_then(|v| v["detail"].as_str().map(String::from))

                    .unwrap_or(text);

                anyhow::anyhow!("{url}: HTTP {code}: {detail}")

            e => anyhow::anyhow!("{url}: {e}"),

    }

            Err(ureq::Error::Status(404, _)) => Ok(None),

            Err(e) => Err(Self::fail(&url, e)),

    pub fn push(&self, home: &Home, hash: &str) -> Result<Moved> {

        let start = Instant::now();

        let path = home.rootfs(hash);

        if !path.is_file() {

            bail!("{} is not in this Flow home ({}); import it first", &hash[..12.min(hash.len())], path.display());

        }

        let m = images::scan_file(&path, home.image_arch(hash), |_, _| Ok(()))?;

        if m.hash != hash {

            bail!("{} holds an image named {}, not {}: the file was changed", path.display(), &m.hash[..12], &hash[..12]);

        }

        let mut missing: HashSet<String> = HashSet::new();

        for ask in m.needed().chunks(ASK) {

            let url = format!("{}/flow/images/missing", self.base);

            let r: serde_json::Value = self

                .agent

                .post(&url)

                .set("Authorization", &self.auth())

                .send_json(serde_json::json!({ "chunks": ask }))

                .map_err(|e| Self::fail(&url, e))?

                .into_json()?;

            missing.extend(r["missing"].as_array().into_iter().flatten().filter_map(|h| h.as_str().map(String::from)));

        let (count, bytes) = (missing.len(), Mutex::new(0u64));

        let queue: Mutex<Vec<(String, Vec<u8>)>> = Mutex::new(Vec::new());

        let failed: Mutex<Option<anyhow::Error>> = Mutex::new(None);

        let sent = Mutex::new(HashSet::new());

        let send = |c: &str, b: &[u8]| -> Result<()> {

            let url = format!("{}/flow/images/chunks/{c}", self.base);

            self.agent.put(&url).set("Authorization", &self.auth()).send_bytes(b).map_err(|e| Self::fail(&url, e))?;

            *bytes.lock().unwrap() += b.len() as u64;

            Ok(())

        };

        images::scan_file(&path, None, |c: &Chunk, b: &[u8]| {

            if missing.contains(c.hash()) && sent.lock().unwrap().insert(c.hash().to_string()) {

                queue.lock().unwrap().push((c.hash().to_string(), b.to_vec()));

            }

            if queue.lock().unwrap().len() >= PARALLEL * 4 {

                let batch: Vec<_> = std::mem::take(&mut *queue.lock().unwrap());

                send_batch(&batch, &send, &failed);

            }

            match failed.lock().unwrap().take() {

                Some(e) => Err(images::Error::Bad(format!("{e:#}"))),

                None => Ok(()),

        })?;

        let rest: Vec<_> = std::mem::take(&mut *queue.lock().unwrap());

        send_batch(&rest, &send, &failed);

        if let Some(e) = failed.lock().unwrap().take() {

            return Err(e);

        }

        let url = format!("{}/flow/images/{hash}", self.base);

        self.agent

            .put(&url)

            .timeout(Duration::from_secs(3600))

            .set("Authorization", &self.auth())

            .send_json(&m)

            .map_err(|e| Self::fail(&url, e))?;

        let bytes = *bytes.lock().unwrap();

        Ok(Moved { chunks: count, bytes, peer_bytes: 0, elapsed: start.elapsed() })

    }

    pub fn fetch(&self, home: &Home, hash: &str) -> Result<Option<Moved>> {

        let start = Instant::now();

        let Some(m) = self.manifest(hash)? else { return Ok(None) };

        if m.hash != hash {

            bail!("the hub's manifest for {} names {}", &hash[..12], &m.hash[..12]);

        }

        let cache = chunk_cache(home);

        save_manifest(&home.root, &m)?;

        let wanted: Vec<String> = m.needed().into_iter().filter(|h| !cache_path(&cache, h).is_file()).map(String::from).collect();

        let (bytes, failed) = (Mutex::new(0u64), Mutex::new(None::<anyhow::Error>));

        let next = Mutex::new(wanted.iter());

        std::thread::scope(|s| {

            for _ in 0..PARALLEL {

                s.spawn(|| loop {

                    let Some(h) = next.lock().unwrap().next() else { break };

                    if failed.lock().unwrap().is_some() {

                        break;

                    }

                    let got = self.chunk(h).and_then(|b| {

                        if blake3::hash(&b).to_hex().as_str() != h {

                            bail!("chunk {} came back with other bytes", &h[..12]);

                        }

                        let p = cache_path(&cache, h);

                        std::fs::create_dir_all(p.parent().expect("a chunk directory"))?;

                        let tmp = p.with_extension("part");

                        std::fs::write(&tmp, &b)?;

                        std::fs::rename(&tmp, &p)?;

                        *bytes.lock().unwrap() += b.len() as u64;

                        Ok(())

                    });

                    if let Err(e) = got {

                        *failed.lock().unwrap() = Some(e);

                        break;

                    }

                });

        });

        if let Some(e) = failed.into_inner().unwrap() {

            return Err(e);

        }

        write_image(home, &m)?;

        Ok(Some(Moved { chunks: wanted.len(), bytes: bytes.into_inner().unwrap(), peer_bytes: 0, elapsed: start.elapsed() }))

    }

fn send_batch(batch: &[(String, Vec<u8>)], send: &(dyn Fn(&str, &[u8]) -> Result<()> + Sync), failed: &Mutex<Option<anyhow::Error>>) {

    std::thread::scope(|s| {

        for part in batch.chunks(batch.len().div_ceil(PARALLEL).max(1)) {

            s.spawn(move || {

                for (h, b) in part {

                    if let Err(e) = send(h, b) {

                        *failed.lock().unwrap() = Some(e);

                        return;

                    }

            });

    });

}

        format!("{:.1} GiB", bytes as f64 / (1u64 << 30) as f64)