OXYGENHUB
# 27322cd2755a oxygenhub / crates / hub / src / api / images.rs
8.0 KiB · Rust · last changed Peer-held images: the hub keeps runners' manifests of images it lacks and places their jobs where a peer in the site holds them (closes #342; pins OxygenFlow #94) 9 days ago source blame history raw
1 
//! Images by hash over the API (oxygenflow#39): `flow rootfs push` sends
2 
//! an image's missing chunks and then its manifest; runners, and `flow
3 
//! rootfs pull`, fetch a manifest and the chunks they lack.
4
5 
use super::{caller, not_found, ApiError};
6 
use crate::Shared;
7 
use axum::body::Bytes;
8 
use axum::extract::{Path, State};
9 
use axum::http::{header, HeaderMap, StatusCode};
10 
use axum::response::{IntoResponse, Response};
11 
use axum::Json;
12 
use jet::images::{is_hash, Manifest};
13 
use serde::{Deserialize, Serialize};
14
15 
/// Most chunk names one `missing` question may ask about.
16 
const MAX_ASK: usize = 200_000;
17 
/// The body that many names make: 64 hex digits, quotes and a comma each,
18 
/// with room to spare (#147: the 2 MB default refused an 11 GB image's).
19 
pub const MAX_ASK_BYTES: usize = MAX_ASK * 80;
20
21 
/// Who sends images: an account whose token may write repositories.
22 
async fn sender(app: &Shared, headers: &HeaderMap) -> Result<String, ApiError> {
23 
    let c = caller(app, headers).await?;
24 
    Ok(c.require("repo:write")?.username.clone())
25 
}
26
27 
/// Who fetches images: a runner, or any account (for `flow rootfs pull`).
28 
async fn fetcher(app: &Shared, headers: &HeaderMap) -> Result<(), ApiError> {
29 
    if crate::flow::runner_auth(app, headers).await.is_ok() {
30 
        return Ok(());
31 
    }
32 
    let c = caller(app, headers).await?;
33 
    c.account.as_ref().map(|_| ()).ok_or_else(|| ApiError(StatusCode::UNAUTHORIZED, "a runner or personal access token is required".into()))
34 
}
35
36 
#[derive(Deserialize)]
37 
pub struct Ask {
38 
    chunks: Vec<String>,
39 
}
40
41 
#[derive(Serialize)]
42 
pub struct Missing {
43 
    missing: Vec<String>,
44 
}
45
46 
/// `POST /flow/images/missing {"chunks": [..]}`: which of these chunks the
47 
/// hub lacks.
48 
pub async fn missing(State(app): State<Shared>, headers: HeaderMap, Json(ask): Json<Ask>) -> Result<Json<Missing>, ApiError> {
49 
    sender(&app, &headers).await?;
50 
    if ask.chunks.len() > MAX_ASK {
51 
        return Err(ApiError(StatusCode::BAD_REQUEST, format!("ask about at most {MAX_ASK} chunks at once")));
52 
    }
53 
    let store = app.images();
54 
    let missing = tokio::task::spawn_blocking(move || {
55 
        let mut seen = std::collections::HashSet::new();
56 
        ask.chunks.into_iter().filter(|h| is_hash(h) && seen.insert(h.clone()) && !store.has_chunk(h)).collect()
57 
    })
58 
    .await
59 
    .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
60 
    Ok(Json(Missing { missing }))
61 
}
62
63 
/// `PUT /flow/images/chunks/{hash}`: one chunk, checked against its name.
64 
pub async fn put_chunk(
65 
    State(app): State<Shared>,
66 
    headers: HeaderMap,
67 
    Path(hash): Path<String>,
68 
    body: Bytes,
69 
) -> Result<StatusCode, ApiError> {
70 
    sender(&app, &headers).await?;
71 
    if !is_hash(&hash) || body.is_empty() || body.len() > oxy_core::chunk::MAX_SIZE {
72 
        return Err(ApiError(StatusCode::BAD_REQUEST, "not a chunk: a name is 64 hex digits, a chunk 1 byte to 256 KiB".into()));
73 
    }
74 
    if blake3::hash(&body).to_hex().as_str() != hash {
75 
        return Err(ApiError(StatusCode::BAD_REQUEST, format!("those bytes are not chunk {}", &hash[..12])));
76 
    }
77 
    let store = app.images();
78 
    tokio::task::spawn_blocking(move || store.put_chunk(&hash, &body))
79 
        .await
80 
        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?
81 
        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("keep the chunk: {e}")))?;
82 
    Ok(StatusCode::NO_CONTENT)
83 
}
84
85 
/// `PUT /flow/images/{hash}`: the manifest, accepted once every chunk is
86 
/// here and the image they make has this hash.
87 
pub async fn put_manifest(
88 
    State(app): State<Shared>,
89 
    headers: HeaderMap,
90 
    Path(hash): Path<String>,
91 
    Json(m): Json<Manifest>,
92 
) -> Result<(StatusCode, Json<serde_json::Value>), ApiError> {
93 
    let who = sender(&app, &headers).await?;
94 
    if m.hash != hash {
95 
        return Err(ApiError(StatusCode::BAD_REQUEST, format!("the manifest is for {}, not {}", m.hash, hash)));
96 
    }
97 
    // In a task of its own (#341): reading a big image can outlast the
98 
    // sender's wait, and a sender that gives up must not leave an image
99 
    // kept but never offered.
100 
    let m = tokio::spawn(async move {
101 
        let store = app.images();
102 
        // A manifest kept before was verified then; only its row may be missing.
103 
        let m = match store.manifest(&m.hash).filter(|k| k.chunks == m.chunks) {
104 
            Some(kept) => kept,
105 
            // Kept with the architecture its own files name (oxygenhub#173).
106 
            None => tokio::task::spawn_blocking(move || store.keep_reading_arch(&m))
107 
                .await
108 
                .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?
109 
                .map_err(|e| ApiError(StatusCode::CONFLICT, e))?,
110 
        };
111 
        crate::images::record(&app.db, &m, &who).await?;
112 
        crate::db::accounts::audit(&app.db, None, "flow.image", &who, &format!("{} ({} MiB)", &m.hash[..12], m.size >> 20)).await.ok();
113 
        Ok::<_, ApiError>(m)
114 
    })
115 
    .await
116 
    .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))??;
117 
    Ok((StatusCode::CREATED, Json(serde_json::json!({ "hash": m.hash, "size": m.size, "chunks": m.chunks.len(), "arch": m.arch }))))
118 
}
119
120 
/// Most images one `held` question may name.
121 
const MAX_HELD: usize = 1000;
122
123 
/// `POST /flow/images/held` (a runner's token): the images it holds; the
124 
/// answer names those the hub keeps no manifest of, for the runner to send
125 
/// with `PUT /flow/images/{hash}/held` (oxygenflow#94).
126 
pub async fn held(
127 
    State(app): State<Shared>,
128 
    headers: HeaderMap,
129 
    Json(ask): Json<flow_core::api::HeldImages>,
130 
) -> Result<Json<flow_core::api::WantedManifests>, ApiError> {
131 
    crate::flow::runner_auth(&app, &headers).await?;
132 
    let kept = crate::images::held(&app.db).await?;
133 
    let mut wanted: Vec<String> =
134 
        ask.images.into_iter().take(MAX_HELD).map(|h| h.to_ascii_lowercase()).filter(|h| is_hash(h) && !kept.contains(h)).collect();
135 
    wanted.sort();
136 
    wanted.dedup();
137 
    Ok(Json(flow_core::api::WantedManifests { wanted }))
138 
}
139
140 
/// `PUT /flow/images/{hash}/held` (a runner's token): the manifest of an
141 
/// image the runner holds and the hub does not. Kept as held by runners
142 
/// only: the hub has none of its chunks to check it against, and a fetcher
143 
/// checks what it assembles against the pinned hash (jetstream.md §8.1.1).
144 
pub async fn put_held(
145 
    State(app): State<Shared>,
146 
    headers: HeaderMap,
147 
    Path(hash): Path<String>,
148 
    Json(m): Json<Manifest>,
149 
) -> Result<StatusCode, ApiError> {
150 
    let runner = crate::flow::runner_auth(&app, &headers).await?;
151 
    if !is_hash(&hash) || m.hash != hash {
152 
        return Err(ApiError(StatusCode::BAD_REQUEST, format!("the manifest is for {}, not {}", m.hash, hash)));
153 
    }
154 
    if crate::images::held(&app.db).await?.contains(&hash) {
155 
        return Ok(StatusCode::NO_CONTENT);
156 
    }
157 
    let store = app.images();
158 
    let kept = m.clone();
159 
    tokio::task::spawn_blocking(move || store.keep_peer_manifest(&kept))
160 
        .await
161 
        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?
162 
        .map_err(|e| ApiError(StatusCode::BAD_REQUEST, e))?;
163 
    crate::images::record_peer(&app.db, &m, &runner.name).await?;
164 
    Ok(StatusCode::CREATED)
165 
}
166
167 
/// `GET /flow/images/{hash}`: an image's manifest.
168 
pub async fn get_manifest(State(app): State<Shared>, headers: HeaderMap, Path(hash): Path<String>) -> Result<Json<Manifest>, ApiError> {
169 
    fetcher(&app, &headers).await?;
170 
    app.images().manifest(&hash).map(Json).ok_or_else(|| not_found("this hub does not hold that image; send it with flow rootfs push"))
171 
}
172
173 
/// `GET /flow/images/chunks/{hash}`: one chunk's bytes.
174 
pub async fn get_chunk(State(app): State<Shared>, headers: HeaderMap, Path(hash): Path<String>) -> Result<Response, ApiError> {
175 
    fetcher(&app, &headers).await?;
176 
    let store = app.images();
177 
    let bytes = tokio::task::spawn_blocking(move || store.chunk(&hash))
178 
        .await
179 
        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?
180 
        .map_err(|_| not_found("no such chunk"))?;
181 
    Ok(([(header::CONTENT_TYPE, "application/octet-stream"), (header::CACHE_CONTROL, "private, max-age=31536000, immutable")], bytes)
182 
        .into_response())
183 
}