diff options
| -rw-r--r-- | src/main.rs | 165 |
1 files changed, 94 insertions, 71 deletions
diff --git a/src/main.rs b/src/main.rs index 97bc034..80b61a8 100644 --- a/src/main.rs +++ b/src/main.rs @@ -281,68 +281,100 @@ fn update_images(db: &Connection, file: &str, auth: &mut Auth, img: DockerRef, e let cache_stale = now - last_checked > cache_max_age; - if let Some(ref tag_str) = tag { - if cache_stale { - let mut url = format!("/v2/{}/tags/list", &image); - loop { - let mut response = match perform_registry_request(registry, &url, "application/vnd.oci.image.index.v1+json", auth) { - Ok(r) => r, - Err(RegistryError::NotFound) => { - println!("Warning: image not found, skipping: {}", image); - return Ok(()); - }, - Err(RegistryError::Other(msg)) => return Err(msg), - }; + let cached_tags: Vec<String> = { + let _timer = metrics::get().db_query_duration.start_timer(); + let mut stmt = db.prepare("SELECT tag FROM tags WHERE image_id = ?1 ORDER BY tag").unwrap(); + let mut rows = stmt.query((image_id,)).unwrap(); + let mut tags = vec![]; + while let Some(row) = rows.next().unwrap() { + tags.push(row.get(0).unwrap()); + } + tags + }; - let content_type = response.headers()["Content-Type"].to_str().unwrap(); - if !content_type.starts_with("application/json") { - return Err(format!("Unexpected Content-Type: {}", content_type)); - } + let (tags, cached_digest): (Vec<String>, Option<String>) = if cache_stale { + let mut url = format!("/v2/{}/tags/list", &image); + let mut fetched_tags = vec![]; + loop { + let mut response = match perform_registry_request(registry, &url, "application/vnd.oci.image.index.v1+json", auth) { + Ok(r) => r, + Err(RegistryError::NotFound) => { + println!("Warning: image not found, skipping: {}", image); + return Ok(()); + }, + Err(RegistryError::Other(msg)) => return Err(msg), + }; - let body = response.body_mut().read_to_string().unwrap(); - let body = body.parse::<tinyjson::JsonValue>().unwrap(); + let content_type = response.headers()["Content-Type"].to_str().unwrap(); + if !content_type.starts_with("application/json") { + return Err(format!("Unexpected Content-Type: {}", content_type)); + } - for it in body["tags"].get::<Vec<tinyjson::JsonValue>>().unwrap().iter() { - let candidate_str = it.get::<String>().unwrap(); + let body = response.body_mut().read_to_string().unwrap(); + let body = body.parse::<tinyjson::JsonValue>().unwrap(); - let _timer = metrics::get().db_query_duration.start_timer(); - db.execute(" - INSERT OR IGNORE INTO tags(image_id, tag, fetched_at) VALUES (?1, ?2, ?3) - ", (image_id, candidate_str, now)).unwrap(); - } + for it in body["tags"].get::<Vec<tinyjson::JsonValue>>().unwrap().iter() { + fetched_tags.push(it.get::<String>().unwrap().clone()); + } - if let Some(link_header) = response.headers().get("link") { - let link_str = link_header.to_str().unwrap(); - let link = extract_next_page(&link_str).unwrap(); - url = link_str[link.next_uri.unwrap().clone()].to_string(); - } else { - break; - } + if let Some(link_header) = response.headers().get("link") { + let link_str = link_header.to_str().unwrap(); + let link = extract_next_page(&link_str).unwrap(); + url = link_str[link.next_uri.unwrap().clone()].to_string(); + } else { + break; } + } + fetched_tags.sort(); + let mut existing = cached_tags.iter().peekable(); + for t in &fetched_tags { + while existing.peek().is_some_and(|e| *e < t) { + let gone = existing.next().unwrap(); + let _timer = metrics::get().db_query_duration.start_timer(); + db.execute("DELETE FROM tags WHERE image_id = ?1 AND tag = ?2", + (image_id, gone)).unwrap(); + } + if existing.peek() == Some(&t) { + existing.next(); + } else { + let _timer = metrics::get().db_query_duration.start_timer(); + db.execute("INSERT INTO tags(image_id, tag, fetched_at) VALUES (?1, ?2, ?3)", + (image_id, t, now)).unwrap(); + } + } + for gone in existing { + let _timer = metrics::get().db_query_duration.start_timer(); + db.execute("DELETE FROM tags WHERE image_id = ?1 AND tag = ?2", + (image_id, gone)).unwrap(); + } + { let _timer = metrics::get().db_query_duration.start_timer(); db.execute("UPDATE images SET last_checked = ?1 WHERE id = ?2", (now, image_id)).unwrap(); } - let tags: Vec<String> = { + (fetched_tags, None) + } else { + let tag_for_digest = tag.as_deref().unwrap_or("latest"); + let digest = { let _timer = metrics::get().db_query_duration.start_timer(); - let mut stmt = db.prepare("SELECT tag FROM tags WHERE image_id = ?1").unwrap(); - let mut rows = stmt.query((image_id,)).unwrap(); - let mut tags = vec![]; - while let Some(row) = rows.next().unwrap() { - tags.push(row.get(0).unwrap()); - } - tags + db.query_one(" + SELECT digest FROM tags WHERE image_id = ?1 AND tag = ?2 AND digest IS NOT NULL + ", (image_id, tag_for_digest), |row| row.get::<_, String>(0)).optional().unwrap() }; + (cached_tags, digest) + }; + + if let Some(ref tag_str) = tag { let mut current = VersionPattern::parse(&tag_str); let mut new_tag = None; - for candidate_str in tags { + for candidate_str in &tags { let candidate = VersionPattern::parse(&candidate_str); match current.compare(&candidate) { CompareOutcome::Higher => { - new_tag = Some(candidate_str); + new_tag = Some(candidate_str.clone()); current = candidate; }, CompareOutcome::Lower => {}, @@ -364,39 +396,30 @@ fn update_images(db: &Connection, file: &str, auth: &mut Auth, img: DockerRef, e if let Some(ref digest) = img.digest { let tag_for_digest = tag.as_deref().unwrap_or("latest"); - let mut digest_string = None; - if !cache_stale { - let _timer = metrics::get().db_query_duration.start_timer(); - digest_string = db.query_one(" - SELECT digest FROM tags WHERE image_id = ?1 AND tag = ?2 AND digest IS NOT NULL - ", (image_id, tag_for_digest), |row| row.get::<_, String>(0)).optional().unwrap(); - } + let digest_string = match cached_digest { + Some(d) => d, + None => { + let url = format!("/v2/{}/manifests/{}", &file[img.image.clone()], tag_for_digest); + let response = match perform_registry_request(registry, &url, "application/vnd.oci.image.manifest.v1+json,application/vnd.oci.image.index.v1+json", auth) { + Ok(r) => r, + Err(RegistryError::NotFound) => { + println!("Warning: image not found, skipping: {}", image); + return Ok(()); + }, + Err(RegistryError::Other(msg)) => return Err(msg), + }; - if digest_string.is_none() { - let url = format!("/v2/{}/manifests/{}", &file[img.image.clone()], tag_for_digest); - let response = match perform_registry_request(registry, &url, "application/vnd.oci.image.manifest.v1+json,application/vnd.oci.image.index.v1+json", auth) { - Ok(r) => r, - Err(RegistryError::NotFound) => { - println!("Warning: image not found, skipping: {}", image); - return Ok(()); - }, - Err(RegistryError::Other(msg)) => return Err(msg), - }; + let fetched = response.headers()["docker-content-digest"].to_str().unwrap().to_string(); - let fetched = response.headers()["docker-content-digest"].to_str().unwrap().to_string(); + { + let _timer = metrics::get().db_query_duration.start_timer(); + db.execute("UPDATE tags SET digest = ?1, fetched_at = ?2 WHERE image_id = ?3 AND tag = ?4", + (&fetched, now, image_id, tag_for_digest)).unwrap(); + } - { - let _timer = metrics::get().db_query_duration.start_timer(); - db.execute(" - INSERT INTO tags(image_id, tag, digest, fetched_at) VALUES (?1, ?2, ?3, ?4) - ON CONFLICT(image_id, tag) DO UPDATE SET digest = ?3, fetched_at = ?4 - ", (image_id, tag_for_digest, &fetched, now)).unwrap(); + fetched } - - digest_string = Some(fetched); - } - - let digest_string = digest_string.unwrap(); + }; if file[digest.clone()] != digest_string { edits.push(FilePatch { |