images.rs 92.1%

139 of 151 lines run, 32 of 80 functions · run #524 on main · run missed

//! Images by hash over the API (oxygenflow#39): `flow rootfs push` sends
//! an image's missing chunks and then its manifest; runners, and `flow
//! rootfs pull`, fetch a manifest and the chunks they lack.

use super::{caller, not_found, ApiError};
use crate::Shared;
use axum::body::Bytes;
use axum::extract::{Path, State};
use axum::http::{header, HeaderMap, StatusCode};
use axum::response::{IntoResponse, Response};
use axum::Json;
use jet::images::{is_hash, Manifest};
use serde::{Deserialize, Serialize};

/// Most chunk names one `missing` question may ask about.
const MAX_ASK: usize = 200_000;
/// The body that many names make: 64 hex digits, quotes and a comma each,
/// with room to spare (#147: the 2 MB default refused an 11 GB image's).
pub const MAX_ASK_BYTES: usize = MAX_ASK * 80;

/// Who sends images: an account whose token may write repositories.
async fn sender(app: &Shared, headers: &HeaderMap) -> Result<String, ApiError> {
    let c = caller(app, headers).await?;
    Ok(c.require("repo:write")?.username.clone())
}

/// Who fetches images: a runner, or any account (for `flow rootfs pull`).
async fn fetcher(app: &Shared, headers: &HeaderMap) -> Result<(), ApiError> {
    if crate::flow::runner_auth(app, headers).await.is_ok() {
        return Ok(());
    }
    let c = caller(app, headers).await?;
    c.account.as_ref().map(|_| ()).ok_or_else(|| ApiError(StatusCode::UNAUTHORIZED, "a runner or personal access token is required".into()))
}

#[derive(Deserialize)]
pub struct Ask {
    chunks: Vec<String>,
}

#[derive(Serialize)]
pub struct Missing {
    missing: Vec<String>,
}

/// `POST /flow/images/missing {"chunks": [..]}`: which of these chunks the
/// hub lacks.
pub async fn missing(State(app): State<Shared>, headers: HeaderMap, Json(ask): Json<Ask>) -> Result<Json<Missing>, ApiError> {
    sender(&app, &headers).await?;
    if ask.chunks.len() > MAX_ASK {
        return Err(ApiError(StatusCode::BAD_REQUEST, format!("ask about at most {MAX_ASK} chunks at once")));
    }
    let store = app.images();
    let missing = tokio::task::spawn_blocking(move || {
        let mut seen = std::collections::HashSet::new();
        ask.chunks.into_iter().filter(|h| is_hash(h) && seen.insert(h.clone()) && !store.has_chunk(h)).collect()
    })
    .await
    .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
    Ok(Json(Missing { missing }))
}

/// `PUT /flow/images/chunks/{hash}`: one chunk, checked against its name.
pub async fn put_chunk(
    State(app): State<Shared>,
    headers: HeaderMap,
    Path(hash): Path<String>,
    body: Bytes,
) -> Result<StatusCode, ApiError> {
    sender(&app, &headers).await?;
    if !is_hash(&hash) || body.is_empty() || body.len() > oxy_core::chunk::MAX_SIZE {
        return Err(ApiError(StatusCode::BAD_REQUEST, "not a chunk: a name is 64 hex digits, a chunk 1 byte to 256 KiB".into()));
    }
    if blake3::hash(&body).to_hex().as_str() != hash {
        return Err(ApiError(StatusCode::BAD_REQUEST, format!("those bytes are not chunk {}", &hash[..12])));
    }
    let store = app.images();
    tokio::task::spawn_blocking(move || store.put_chunk(&hash, &body))
        .await
        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?
        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("keep the chunk: {e}")))?;
    Ok(StatusCode::NO_CONTENT)
}

/// `PUT /flow/images/{hash}`: the manifest, accepted once every chunk is
/// here and the image they make has this hash.
pub async fn put_manifest(
    State(app): State<Shared>,
    headers: HeaderMap,
    Path(hash): Path<String>,
    Json(m): Json<Manifest>,
) -> Result<(StatusCode, Json<serde_json::Value>), ApiError> {
    let who = sender(&app, &headers).await?;
    if m.hash != hash {
        return Err(ApiError(StatusCode::BAD_REQUEST, format!("the manifest is for {}, not {}", m.hash, hash)));
    }
    // In a task of its own (#341): reading a big image can outlast the
    // sender's wait, and a sender that gives up must not leave an image
    // kept but never offered.
    let m = tokio::spawn(async move {
        let store = app.images();
        // A manifest kept before was verified then; only its row may be missing.
        let m = match store.manifest(&m.hash).filter(|k| k.chunks == m.chunks) {
            Some(kept) => kept,
            // Kept with the architecture its own files name (oxygenhub#173).
            None => tokio::task::spawn_blocking(move || store.keep_reading_arch(&m))
                .await
                .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?
                .map_err(|e| ApiError(StatusCode::CONFLICT, e))?,
        };
        crate::images::record(&app.db, &m, &who).await?;
        crate::db::accounts::audit(&app.db, None, "flow.image", &who, &format!("{} ({} MiB)", &m.hash[..12], m.size >> 20)).await.ok();
        Ok::<_, ApiError>(m)
    })
    .await
    .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))??;
    Ok((StatusCode::CREATED, Json(serde_json::json!({ "hash": m.hash, "size": m.size, "chunks": m.chunks.len(), "arch": m.arch }))))
}

/// Most images one `held` question may name.
const MAX_HELD: usize = 1000;

/// `POST /flow/images/held` (a runner's token): the images it holds; the
/// answer names those the hub keeps no manifest of, for the runner to send
/// with `PUT /flow/images/{hash}/held` (oxygenflow#94).
pub async fn held(
    State(app): State<Shared>,
    headers: HeaderMap,
    Json(ask): Json<flow_core::api::HeldImages>,
) -> Result<Json<flow_core::api::WantedManifests>, ApiError> {
    crate::flow::runner_auth(&app, &headers).await?;
    let kept = crate::images::held(&app.db).await?;
    let mut wanted: Vec<String> =
        ask.images.into_iter().take(MAX_HELD).map(|h| h.to_ascii_lowercase()).filter(|h| is_hash(h) && !kept.contains(h)).collect();
    wanted.sort();
    wanted.dedup();
    Ok(Json(flow_core::api::WantedManifests { wanted }))
}

/// `PUT /flow/images/{hash}/held` (a runner's token): the manifest of an
/// image the runner holds and the hub does not. Kept as held by runners
/// only: the hub has none of its chunks to check it against, and a fetcher
/// checks what it assembles against the pinned hash (jetstream.md §8.1.1).
pub async fn put_held(
    State(app): State<Shared>,
    headers: HeaderMap,
    Path(hash): Path<String>,
    Json(m): Json<Manifest>,
) -> Result<StatusCode, ApiError> {
    let runner = crate::flow::runner_auth(&app, &headers).await?;
    if !is_hash(&hash) || m.hash != hash {
        return Err(ApiError(StatusCode::BAD_REQUEST, format!("the manifest is for {}, not {}", m.hash, hash)));
    }
    if crate::images::held(&app.db).await?.contains(&hash) {
        return Ok(StatusCode::NO_CONTENT);
    }
    let store = app.images();
    let kept = m.clone();
    tokio::task::spawn_blocking(move || store.keep_peer_manifest(&kept))
        .await
        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?
        .map_err(|e| ApiError(StatusCode::BAD_REQUEST, e))?;
    crate::images::record_peer(&app.db, &m, &runner.name).await?;
    Ok(StatusCode::CREATED)
}

/// `GET /flow/images/{hash}`: an image's manifest.
pub async fn get_manifest(State(app): State<Shared>, headers: HeaderMap, Path(hash): Path<String>) -> Result<Json<Manifest>, ApiError> {
    fetcher(&app, &headers).await?;
    app.images().manifest(&hash).map(Json).ok_or_else(|| not_found("this hub does not hold that image; send it with flow rootfs push"))
}

/// `GET /flow/images/chunks/{hash}`: one chunk's bytes.
pub async fn get_chunk(State(app): State<Shared>, headers: HeaderMap, Path(hash): Path<String>) -> Result<Response, ApiError> {
    fetcher(&app, &headers).await?;
    let store = app.images();
    let bytes = tokio::task::spawn_blocking(move || store.chunk(&hash))
        .await
        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?
        .map_err(|_| not_found("no such chunk"))?;
    Ok(([(header::CONTENT_TYPE, "application/octet-stream"), (header::CACHE_CONTROL, "private, max-age=31536000, immutable")], bytes)
        .into_response())
}

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	23×	async fn sender(app: &Shared, headers: &HeaderMap) -> Result<String, ApiError> {
23	23×	let c = caller(app, headers).await?;
24	22×	Ok(c.require("repo:write")?.username.clone())
25	23×	}
26
27		/// Who fetches images: a runner, or any account (for `flow rootfs pull`).
28	9×	async fn fetcher(app: &Shared, headers: &HeaderMap) -> Result<(), ApiError> {
29	9×	if crate::flow::runner_auth(app, headers).await.is_ok() {
30	8×	return Ok(());
31	1×	}
32	1×	let c = caller(app, headers).await?;
33	1×	c.account.as_ref().map(\|_\| ()).ok_or_else(\|\| ApiError(StatusCode::UNAUTHORIZED, "a runner or personal access token is required".into()))
34	9×	}
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	3×	pub async fn missing(State(app): State<Shared>, headers: HeaderMap, Json(ask): Json<Ask>) -> Result<Json<Missing>, ApiError> {
49	3×	sender(&app, &headers).await?;
50	3×	if ask.chunks.len() > MAX_ASK {
51	0×	return Err(ApiError(StatusCode::BAD_REQUEST, format!("ask about at most {MAX_ASK} chunks at once")));
52	3×	}
53	3×	let store = app.images();
54	3×	let missing = tokio::task::spawn_blocking(move \|\| {
55	3×	let mut seen = std::collections::HashSet::new();
56	60006×	ask.chunks.into_iter().filter(\|h\| is_hash(h) && seen.insert(h.clone()) && !store.has_chunk(h)).collect()
57	3×	})
58	3×	.await
59	3×	.map_err(\|e\| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
60	3×	Ok(Json(Missing { missing }))
61	3×	}
62
63		/// `PUT /flow/images/chunks/{hash}`: one chunk, checked against its name.
64	13×	pub async fn put_chunk(
65	13×	State(app): State<Shared>,
66	13×	headers: HeaderMap,
67	13×	Path(hash): Path<String>,
68	13×	body: Bytes,
69	13×	) -> Result<StatusCode, ApiError> {
70	13×	sender(&app, &headers).await?;
71	12×	if !is_hash(&hash) \|\| body.is_empty() \|\| body.len() > oxy_core::chunk::MAX_SIZE {
72	0×	return Err(ApiError(StatusCode::BAD_REQUEST, "not a chunk: a name is 64 hex digits, a chunk 1 byte to 256 KiB".into()));
73	12×	}
74	12×	if blake3::hash(&body).to_hex().as_str() != hash {
75	1×	return Err(ApiError(StatusCode::BAD_REQUEST, format!("those bytes are not chunk {}", &hash[..12])));
76	11×	}
77	11×	let store = app.images();
78	11×	tokio::task::spawn_blocking(move \|\| store.put_chunk(&hash, &body))
79	11×	.await
80	11×	.map_err(\|e\| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?
81	11×	.map_err(\|e\| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("keep the chunk: {e}")))?;
82	11×	Ok(StatusCode::NO_CONTENT)
83	13×	}
84
85		/// `PUT /flow/images/{hash}`: the manifest, accepted once every chunk is
86		/// here and the image they make has this hash.
87	7×	pub async fn put_manifest(
88	7×	State(app): State<Shared>,
89	7×	headers: HeaderMap,
90	7×	Path(hash): Path<String>,
91	7×	Json(m): Json<Manifest>,
92	7×	) -> Result<(StatusCode, Json<serde_json::Value>), ApiError> {
93	7×	let who = sender(&app, &headers).await?;
94	7×	if m.hash != hash {
95	0×	return Err(ApiError(StatusCode::BAD_REQUEST, format!("the manifest is for {}, not {}", m.hash, hash)));
96	7×	}
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	7×	let m = tokio::spawn(async move {
101	7×	let store = app.images();
102		// A manifest kept before was verified then; only its row may be missing.
103	7×	let m = match store.manifest(&m.hash).filter(\|k\| k.chunks == m.chunks) {
104	2×	Some(kept) => kept,
105		// Kept with the architecture its own files name (oxygenhub#173).
106	5×	None => tokio::task::spawn_blocking(move \|\| store.keep_reading_arch(&m))
107	5×	.await
108	5×	.map_err(\|e\| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?
109	5×	.map_err(\|e\| ApiError(StatusCode::CONFLICT, e))?,
110		};
111	5×	crate::images::record(&app.db, &m, &who).await?;
112	5×	crate::db::accounts::audit(&app.db, None, "flow.image", &who, &format!("{} ({} MiB)", &m.hash[..12], m.size >> 20)).await.ok();
113	5×	Ok::<_, ApiError>(m)
114	7×	})
115	7×	.await
116	7×	.map_err(\|e\| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))??;
117	5×	Ok((StatusCode::CREATED, Json(serde_json::json!({ "hash": m.hash, "size": m.size, "chunks": m.chunks.len(), "arch": m.arch }))))
118	7×	}
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	3×	pub async fn held(
127	3×	State(app): State<Shared>,
128	3×	headers: HeaderMap,
129	3×	Json(ask): Json<flow_core::api::HeldImages>,
130	3×	) -> Result<Json<flow_core::api::WantedManifests>, ApiError> {
131	3×	crate::flow::runner_auth(&app, &headers).await?;
132	2×	let kept = crate::images::held(&app.db).await?;
133	2×	let mut wanted: Vec<String> =
134	4×	ask.images.into_iter().take(MAX_HELD).map(\|h\| h.to_ascii_lowercase()).filter(\|h\| is_hash(h) && !kept.contains(h)).collect();
135	2×	wanted.sort();
136	2×	wanted.dedup();
137	2×	Ok(Json(flow_core::api::WantedManifests { wanted }))
138	3×	}
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	2×	pub async fn put_held(
145	2×	State(app): State<Shared>,
146	2×	headers: HeaderMap,
147	2×	Path(hash): Path<String>,
148	2×	Json(m): Json<Manifest>,
149	2×	) -> Result<StatusCode, ApiError> {
150	2×	let runner = crate::flow::runner_auth(&app, &headers).await?;
151	2×	if !is_hash(&hash) \|\| m.hash != hash {
152	1×	return Err(ApiError(StatusCode::BAD_REQUEST, format!("the manifest is for {}, not {}", m.hash, hash)));
153	1×	}
154	1×	if crate::images::held(&app.db).await?.contains(&hash) {
155	0×	return Ok(StatusCode::NO_CONTENT);
156	1×	}
157	1×	let store = app.images();
158	1×	let kept = m.clone();
159	1×	tokio::task::spawn_blocking(move \|\| store.keep_peer_manifest(&kept))
160	1×	.await
161	1×	.map_err(\|e\| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?
162	1×	.map_err(\|e\| ApiError(StatusCode::BAD_REQUEST, e))?;
163	1×	crate::images::record_peer(&app.db, &m, &runner.name).await?;
164	1×	Ok(StatusCode::CREATED)
165	2×	}
166
167		/// `GET /flow/images/{hash}`: an image's manifest.
168	5×	pub async fn get_manifest(State(app): State<Shared>, headers: HeaderMap, Path(hash): Path<String>) -> Result<Json<Manifest>, ApiError> {
169	5×	fetcher(&app, &headers).await?;
170	4×	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	5×	}
172
173		/// `GET /flow/images/chunks/{hash}`: one chunk's bytes.
174	4×	pub async fn get_chunk(State(app): State<Shared>, headers: HeaderMap, Path(hash): Path<String>) -> Result<Response, ApiError> {
175	4×	fetcher(&app, &headers).await?;
176	4×	let store = app.images();
177	4×	let bytes = tokio::task::spawn_blocking(move \|\| store.chunk(&hash))
178	4×	.await
179	4×	.map_err(\|e\| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?
180	4×	.map_err(\|_\| not_found("no such chunk"))?;
181	3×	Ok(([(header::CONTENT_TYPE, "application/octet-stream"), (header::CACHE_CONTROL, "private, max-age=31536000, immutable")], bytes)
182	3×	.into_response())
183	4×	}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

Coverage / crates/hub/src/api/images.rs 92.1%

Keyboard