Runners offer the manifests of images the hub lacks; a fetch the hub cannot seed asks peers alone (refs #94) 32f54b393396

- **The swarm moves an image nobody pushed (#94).** A runner that serves its peers tells the hub, at start and every ten minutes, which images it holds, and sends the manifest of each the hub keeps none of: the manifest alone, not the image. With a hub that keeps such manifests (oxygenhub#342), runners in the same site fetch the image from it. A fetch told the hub holds no chunks asks peers alone, and fails plainly when none can serve it. Design: docs/jetstream.md §8.1.1.

    /// Whether the hub holds the image's chunks and so may be asked for

    /// them (jetstream.md §8.1.1): `Some(false)` for an image only runners

    /// hold; absent from a hub that predates it, which always seeds.

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub seed: Option<bool>,

}

/// `POST /flow/images/held`: the images a runner holds, by hash

/// (oxygenflow#94).

#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]

pub struct HeldImages {

    pub images: Vec<String>,

}

/// The hub's answer to [`HeldImages`]: those it keeps no manifest for, which

/// the runner sends with `PUT /flow/images/{hash}/held`.

#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]

pub struct WantedManifests {

    pub wanted: Vec<String>,

impl Client {

    /// Tell the hub which images this home holds, and send the manifest of

    /// each it keeps none for (oxygenflow#94, jetstream.md §8.1.1): the

    /// swarm can then carry an image nobody pushed. A manifest not kept here

    /// yet is made by reading the image once, and kept. How many were sent;

    /// none from a hub that predates it.

    pub fn offer_held(&self, home: &Home) -> Result<usize> {

        let images = home.images();

        if images.is_empty() {

            return Ok(0);

        }

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

        let wanted: flow_core::api::WantedManifests =

            match self.agent.post(&url).set("Authorization", &self.auth()).send_json(flow_core::api::HeldImages { images }) {

                Ok(r) => r.into_json()?,

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

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

            };

        let mut sent = 0;

        for hash in wanted.wanted.iter().filter(|h| images::is_hash(h)) {

            let path = home.rootfs(hash);

            if !path.is_file() {

                continue;

            }

            let kept = std::fs::read(manifest_path(&home.root, hash)).ok().and_then(|b| serde_json::from_slice::<Manifest>(&b).ok());

            let m = match kept.filter(|m| m.hash == *hash) {

                Some(m) => m,

                None => {

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

                    if m.hash != *hash {

                        continue;

                    }

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

                    m

                }

            };

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

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

            sent += 1;

        }

        Ok(sent)

    }

}

    // The manifests of images held here that the hub keeps none of, so

    // peers in this runner's site can fetch them without anyone pushing

    // them (oxygenflow#94): at start and every ten minutes, which also

    // catches an image imported while it runs. Only a runner that serves

    // its peers has anything to offer.

    if swarm.is_some() {

        let (root, cfg) = (home.root.clone(), cfg.clone());

        std::thread::spawn(move || loop {

            let offered = Home::locate(Some(&root))

                .and_then(|home| crate::images::Client::new(&cfg.hub, &cfg.token, cfg.ca.as_deref())?.offer_held(&home));

            match offered {

                Ok(0) => {}

                Ok(n) => eprintln!("flow runner: sent the hub the manifests of {n} image(s) only runners hold"),

                Err(e) => eprintln!("flow runner: offering held images: {e:#}"),

            }

            std::thread::sleep(Duration::from_secs(600));

        });

    }

            // An image only runners hold (jetstream.md §8.1.1): the hub

            // has its manifest but none of its chunks, so only peers are asked.

            let hub_seeds = announced.as_ref().is_none_or(|a| a.seed != Some(false));

                if hub_seeds {

                    for _ in 0..HUB_WORKERS {

                        s.spawn(move || worker(Source::Hub));

                    }

                let mut peerless = Instant::now();

                    if !hub_seeds {

                        if peers.lock().unwrap().iter().any(|p| p.alive.load(Ordering::Relaxed)) {

                            peerless = Instant::now();

                        } else if peerless.elapsed() >= interval * 2 {

                            *failed.lock().unwrap() = Some(anyhow::anyhow!(

                                "image {} is held only by runners, and none in this site could serve it for {} s; send it to the hub with `flow rootfs push`",

                                &hash[..12],

                                (interval * 2).as_secs()

                            ));

                        }

                    }

        hub_with(m, image, site, Duration::ZERO, true)

        hub_with(m, image, site, delay, true)

    }

    /// A hub that keeps the image's manifest but none of its chunks, as for

    /// an image only runners hold (jetstream.md §8.1.1).

    fn peer_only_hub(m: Manifest, image: Vec<u8>, site: &str) -> FakeHub {

        hub_with(m, image, site, Duration::ZERO, false)

    }

    fn hub_with(m: Manifest, image: Vec<u8>, site: &str, delay: Duration, seeds: bool) -> FakeHub {

                        } else if let Some(c) = path.strip_prefix("/api/v1/flow/images/chunks/").filter(|_| seeds) {

                            let a = flow_core::api::Announced {

                                ticket: ticket.seal(key, sk),

                                peers,

                                interval_s: 5,

                                seed: (!seeds).then_some(false),

                            };

    }

    #[test]

    fn an_image_only_runners_hold_comes_from_a_peer_and_fails_plainly_without_one() {

        let bytes = image();

        let m = jet::images::scan(&bytes[..], Some("x86_64".into()), |_, _| Ok(())).unwrap();

        let hub = peer_only_hub(m.clone(), bytes.clone(), "lab");

        // No peer yet: no hub to fall back on, so the fetch says why.

        let lone = tempfile::tempdir().unwrap();

        let (lone_home, lone_runner) = runner(lone.path(), &hub, 3, "tc", free_port(), "lab");

        let e = lone_runner.fetch(&lone_home, &m.hash).err().expect("a fetch with no peer and no seed fails");

        assert!(e.to_string().contains("held only by runners"), "{e:#}");

        // With a peer holding it, it all comes from the peer.

        let (a_dir, b_dir) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());

        std::fs::create_dir_all(a_dir.path().join("cas")).unwrap();

        std::fs::write(a_dir.path().join("cas").join(format!("{}.ext4", m.hash)), &bytes).unwrap();

        let (_a_home, a) = runner(a_dir.path(), &hub, 1, "ta", free_port(), "lab");

        let deadline = Instant::now() + Duration::from_secs(20);

        while !a.state.offers.lock().unwrap().contains_key(&m.hash) {

            assert!(Instant::now() < deadline, "runner A never offered its image");

            std::thread::sleep(Duration::from_millis(20));

        }

        let (b_home, b) = runner(b_dir.path(), &hub, 2, "tb", free_port(), "lab");

        let moved = b.fetch(&b_home, &m.hash).unwrap().unwrap();

        assert_eq!(std::fs::read(b_home.rootfs(&m.hash)).unwrap(), bytes);

        assert_eq!(moved.peer_bytes, moved.bytes, "every chunk came from the peer");

        assert_eq!(hub.chunks_served.load(Ordering::Relaxed), 0);

- **Runners say what they hold.** At start, every ten minutes and after an

  import, a runner sends the hashes of the images it holds

  (`POST /flow/images/held {images}`); the hub answers with those it keeps

  no manifest for (`{wanted}`), and the runner sends each manifest alone

  chunks. (Not in the claim: an idle runner's claim gets no answer at all.)