| 10 | 10 | | //! `FinalizeCacheEntryUpload`, `GetCacheEntryDownloadURL`) and the |
| 11 | 11 | | //! artifacts' (`CreateArtifact`, `FinalizeArtifact`, `ListArtifacts`, |
| 12 | 12 | | //! `GetSignedArtifactURL`, `DeleteArtifact`). JSON, the toolkit's field |
| 13 | | − | //! names. |
| 13 | + | //! names; the cache's methods also in protobuf (`application/protobuf`), |
| 14 | + | //! which other clients of the protocol send (sccache, through OpenDAL). |
| 14 | 15 | | //! - The cache's older protocol, at `{ACTIONS_CACHE_URL}_apis/artifactcache/…`, |
| 15 | 16 | | //! which the toolkit's client uses whenever the server it runs against is |
| 16 | | − | //! not github.com: on g1t, that is the one it uses. |
| 17 | + | //! not github.com: on g1t, that is the one it uses, and sccache's too. |
| 18 | + | //! Its entries are sent in 32 MB chunks, or in one chunk of any size. |
| 17 | 19 | | //! - Blobs, at `/actions/toolkit/blobs/{token}`: the signed links those |
| 18 | | − | //! hand out. Downloads are a plain GET. Uploads speak the part of Azure |
| 19 | | − | //! Blob Storage's protocol the toolkit's client uses (Put Blob, Put |
| 20 | | − | //! Block, Put Block List), mapped onto an R2 multipart upload: a block's |
| 21 | | − | //! id ends in its index, which is its part's number. |
| 20 | + | //! hand out. Downloads are a GET, of the whole blob or of one byte range |
| 21 | + | //! (`Range`), as the toolkit's client fetches large entries in segments. |
| 22 | + | //! Uploads speak the part of Azure Blob Storage's protocol the toolkit's |
| 23 | + | //! client uses (Put Blob, Put Block, Put Block List), mapped onto an R2 |
| 24 | + | //! multipart upload: a block's id ends in its index, which is its part's |
| 25 | + | //! number. An upload link carries a query, as an Azure SAS link does, |
| 26 | + | //! which clients that sign their requests with it need. |
| 22 | 27 | | //! |
| 23 | 28 | | //! Every call carries the job's runtime token; the actions service checks |
| 24 | 29 | | //! it and keeps the entries (cache.rs, artifacts.rs, runtime.rs there). |
| ⋯ |
| 85 | 90 | | |
| 86 | 91 | | // ── Twirp ─────────────────────────────────────────────────────────────────── |
| 87 | 92 | | |
| 93 | + | /// Twirp's binary encoding. |
| 94 | + | const PROTOBUF: &str = "application/protobuf"; |
| 95 | + | |
| 96 | + | /// Whether a request's `Content-Type` is Twirp's protobuf encoding. |
| 97 | + | fn is_protobuf(content_type: &str) -> bool { |
| 98 | + | let kind = content_type.split(';').next().unwrap_or_default().trim().to_ascii_lowercase(); |
| 99 | + | kind == PROTOBUF || kind == "application/x-protobuf" |
| 100 | + | } |
| 101 | + | |
| 102 | + | /// The cache service's messages in protobuf, read into and written from the |
| 103 | + | /// JSON the handlers use (`results/api/v1/cache.proto`, field numbers as |
| 104 | + | /// there). Only what the cache's three methods carry: strings, a repeated |
| 105 | + | /// string, an int64 and a bool. `metadata` (field 1 of each request) is |
| 106 | + | /// skipped: the runtime token says whose cache it is. |
| 107 | + | mod proto { |
| 108 | + | use serde_json::{Map, Value}; |
| 109 | + | |
| 110 | + | #[derive(Clone, Copy)] |
| 111 | + | enum Kind { |
| 112 | + | Text, |
| 113 | + | Texts, |
| 114 | + | Int, |
| 115 | + | Bool, |
| 116 | + | } |
| 117 | + | |
| 118 | + | /// A message's fields: number, JSON name, kind. |
| 119 | + | type Fields = &'static [(u64, &'static str, Kind)]; |
| 120 | + | |
| 121 | + | fn request_fields(method: &str) -> Option<Fields> { |
| 122 | + | Some(match method { |
| 123 | + | "CreateCacheEntry" => &[(2, "key", Kind::Text), (3, "version", Kind::Text)], |
| 124 | + | "FinalizeCacheEntryUpload" => &[(2, "key", Kind::Text), (3, "size_bytes", Kind::Int), (4, "version", Kind::Text)], |
| 125 | + | "GetCacheEntryDownloadURL" => &[(2, "key", Kind::Text), (3, "restore_keys", Kind::Texts), (4, "version", Kind::Text)], |
| 126 | + | _ => return None, |
| 127 | + | }) |
| 128 | + | } |
| 129 | + | |
| 130 | + | fn response_fields(method: &str) -> Fields { |
| 131 | + | match method { |
| 132 | + | "CreateCacheEntry" => &[(1, "ok", Kind::Bool), (2, "signed_upload_url", Kind::Text), (3, "message", Kind::Text)], |
| 133 | + | "FinalizeCacheEntryUpload" => &[(1, "ok", Kind::Bool), (2, "entry_id", Kind::Int), (3, "message", Kind::Text)], |
| 134 | + | "GetCacheEntryDownloadURL" => &[(1, "ok", Kind::Bool), (2, "signed_download_url", Kind::Text), (3, "matched_key", Kind::Text)], |
| 135 | + | _ => &[], |
| 136 | + | } |
| 137 | + | } |
| 138 | + | |
| 139 | + | fn varint(bytes: &[u8], at: &mut usize) -> Option<u64> { |
| 140 | + | let mut value = 0u64; |
| 141 | + | for shift in (0..64).step_by(7) { |
| 142 | + | let byte = *bytes.get(*at)?; |
| 143 | + | *at += 1; |
| 144 | + | value |= u64::from(byte & 0x7f) << shift; |
| 145 | + | if byte & 0x80 == 0 { |
| 146 | + | return Some(value); |
| 147 | + | } |
| 148 | + | } |
| 149 | + | None |
| 150 | + | } |
| 151 | + | |
| 152 | + | fn put_varint(out: &mut Vec<u8>, mut value: u64) { |
| 153 | + | while value >= 0x80 { |
| 154 | + | out.push((value as u8 & 0x7f) | 0x80); |
| 155 | + | value >>= 7; |
| 156 | + | } |
| 157 | + | out.push(value as u8); |
| 158 | + | } |
| 159 | + | |
| 160 | + | /// A request of `method` as JSON, or None when it is not one. |
| 161 | + | pub fn request(method: &str, bytes: &[u8]) -> Option<Value> { |
| 162 | + | let fields = request_fields(method)?; |
| 163 | + | let mut out = Map::new(); |
| 164 | + | let mut at = 0; |
| 165 | + | while at < bytes.len() { |
| 166 | + | let tag = varint(bytes, &mut at)?; |
| 167 | + | let (number, wire) = (tag >> 3, tag & 7); |
| 168 | + | let known = fields.iter().find(|(n, _, _)| *n == number); |
| 169 | + | match wire { |
| 170 | + | 0 => { |
| 171 | + | let value = varint(bytes, &mut at)?; |
| 172 | + | if let Some((_, name, Kind::Int)) = known { |
| 173 | + | // An int64 is sent as its two's complement. |
| 174 | + | out.insert((*name).to_owned(), Value::String((value as i64).to_string())); |
| 175 | + | } |
| 176 | + | } |
| 177 | + | 2 => { |
| 178 | + | let length = usize::try_from(varint(bytes, &mut at)?).ok()?; |
| 179 | + | let end = at.checked_add(length).filter(|end| *end <= bytes.len())?; |
| 180 | + | let raw = &bytes[at..end]; |
| 181 | + | at = end; |
| 182 | + | match known { |
| 183 | + | Some((_, name, Kind::Text)) => { |
| 184 | + | out.insert((*name).to_owned(), Value::String(String::from_utf8(raw.to_vec()).ok()?)); |
| 185 | + | } |
| 186 | + | Some((_, name, Kind::Texts)) => { |
| 187 | + | let text = Value::String(String::from_utf8(raw.to_vec()).ok()?); |
| 188 | + | match out.entry((*name).to_owned()).or_insert_with(|| Value::Array(Vec::new())) { |
| 189 | + | Value::Array(list) => list.push(text), |
| 190 | + | _ => return None, |
| 191 | + | } |
| 192 | + | } |
| 193 | + | _ => {} |
| 194 | + | } |
| 195 | + | } |
| 196 | + | 1 => at = at.checked_add(8).filter(|end| *end <= bytes.len())?, |
| 197 | + | 5 => at = at.checked_add(4).filter(|end| *end <= bytes.len())?, |
| 198 | + | _ => return None, |
| 199 | + | } |
| 200 | + | } |
| 201 | + | Some(Value::Object(out)) |
| 202 | + | } |
| 203 | + | |
| 204 | + | /// A response of `method` from its JSON. Defaults are left out, as |
| 205 | + | /// proto3 does. |
| 206 | + | pub fn response(method: &str, value: &Value) -> Vec<u8> { |
| 207 | + | let mut out = Vec::new(); |
| 208 | + | for (number, name, kind) in response_fields(method) { |
| 209 | + | let field = &value[*name]; |
| 210 | + | match kind { |
| 211 | + | Kind::Bool if field.as_bool() == Some(true) => { |
| 212 | + | put_varint(&mut out, number << 3); |
| 213 | + | put_varint(&mut out, 1); |
| 214 | + | } |
| 215 | + | Kind::Int => { |
| 216 | + | let n = field.as_i64().or_else(|| field.as_str().and_then(|s| s.parse().ok())).unwrap_or(0); |
| 217 | + | if n != 0 { |
| 218 | + | put_varint(&mut out, number << 3); |
| 219 | + | put_varint(&mut out, n as u64); |
| 220 | + | } |
| 221 | + | } |
| 222 | + | Kind::Text => { |
| 223 | + | let text = field.as_str().unwrap_or_default(); |
| 224 | + | if !text.is_empty() { |
| 225 | + | put_varint(&mut out, (number << 3) | 2); |
| 226 | + | put_varint(&mut out, text.len() as u64); |
| 227 | + | out.extend_from_slice(text.as_bytes()); |
| 228 | + | } |
| 229 | + | } |
| 230 | + | _ => {} |
| 231 | + | } |
| 232 | + | } |
| 233 | + | out |
| 234 | + | } |
| 235 | + | } |
| 236 | + | |
| 88 | 237 | | /// A Twirp error: its code and message, at the status Twirp gives it. |
| 89 | 238 | | fn twirp_error(code: &str, message: &str) -> Result<Response> { |
| 90 | 239 | | let status = match code { |
| ⋯ |
| 92 | 241 | | "permission_denied" => 403, |
| 93 | 242 | | "not_found" => 404, |
| 94 | 243 | | "already_exists" => 409, |
| 95 | | − | "invalid_argument" => 400, |
| 244 | + | "invalid_argument" | "malformed" => 400, |
| 245 | + | "bad_route" => 404, |
| 96 | 246 | | "failed_precondition" => 412, |
| 97 | 247 | | "resource_exhausted" => 429, |
| 98 | 248 | | _ => 500, |
| ⋯ |
| 155 | 305 | | }) |
| 156 | 306 | | } |
| 157 | 307 | | |
| 308 | + | /// The Azure Storage version g1t's blob links answer as. |
| 309 | + | const AZURE_VERSION: &str = "2024-11-04"; |
| 310 | + | |
| 311 | + | /// An upload link: the blob's, with a query as an Azure SAS link has one. |
| 312 | + | /// A client that treats it as a container, a blob and a SAS token (OpenDAL, |
| 313 | + | /// which sccache uses) refuses a link without one; the token in the path |
| 314 | + | /// is what g1t checks. |
| 315 | + | pub fn upload_url(api: &str, blob: &str) -> String { |
| 316 | + | format!("{}?sv={AZURE_VERSION}", blob_url(api, blob)) |
| 317 | + | } |
| 318 | + | |
| 158 | 319 | | /// Starts an R2 upload for an entry the service reserved, and the signed |
| 159 | 320 | | /// link the toolkit sends it to. |
| 160 | 321 | | async fn start_upload(bucket: &Bucket, services: &Services, job: &str, token: &str, kind: &str, id: &str, object: &str) -> Result<Outcome<String>> { |
| ⋯ |
| 167 | 328 | | ) |
| 168 | 329 | | .await?; |
| 169 | 330 | | Ok(match signed { |
| 170 | | − | Outcome::Ok(blob) => Outcome::Ok(blob_url(&services.addresses.api, &blob)), |
| 331 | + | Outcome::Ok(blob) => Outcome::Ok(upload_url(&services.addresses.api, &blob)), |
| 171 | 332 | | Outcome::Fail(refused) => Outcome::Fail(refused), |
| 172 | 333 | | }) |
| 173 | 334 | | } |
| ⋯ |
| 178 | 339 | | let Some(job) = runtime_job(&token) else { |
| 179 | 340 | | return twirp_error("unauthenticated", "Send the job's ACTIONS_RUNTIME_TOKEN as a bearer token."); |
| 180 | 341 | | }; |
| 181 | | − | let body: Value = request.json().await.unwrap_or(Value::Null); |
| 342 | + | // Twirp clients send JSON or protobuf, and are answered in kind. |
| 343 | + | let binary = is_protobuf(&request.headers().get("content-type")?.unwrap_or_default()); |
| 344 | + | let body: Value = if binary { |
| 345 | + | match proto::request(method, &request.bytes().await.unwrap_or_default()) { |
| 346 | + | Some(body) => body, |
| 347 | + | None => return twirp_error("malformed", "That is not a protobuf message this method takes."), |
| 348 | + | } |
| 349 | + | } else { |
| 350 | + | request.json().await.unwrap_or(Value::Null) |
| 351 | + | }; |
| 352 | + | let answer = |value: Value| -> Result<Response> { |
| 353 | + | if binary { |
| 354 | + | let mut response = Response::from_bytes(proto::response(method, &value))?; |
| 355 | + | response.headers_mut().set("content-type", PROTOBUF)?; |
| 356 | + | Ok(response) |
| 357 | + | } else { |
| 358 | + | Response::from_json(&value) |
| 359 | + | } |
| 360 | + | }; |
| 182 | 361 | | let bucket = env.bucket("ACTIONS_CACHE")?; |
| 183 | 362 | | let actions = &services.actions; |
| 184 | 363 | | let (run, own_job) = backend_ids(&token); |
| ⋯ |
| 201 | 380 | | let found: Outcome<Option<CacheHit>> = g1t_kit::call(actions, "cache_lookup", &args).await?; |
| 202 | 381 | | match found { |
| 203 | 382 | | Outcome::Ok(Some(CacheHit { key, blob: Some(blob), .. })) => { |
| 204 | | − | Response::from_json(&json!({ "ok": true, "signed_download_url": blob_url(&services.addresses.api, &blob), "matched_key": key })) |
| 383 | + | answer(json!({ "ok": true, "signed_download_url": blob_url(&services.addresses.api, &blob), "matched_key": key })) |
| 205 | 384 | | } |
| 206 | | − | Outcome::Ok(_) => Response::from_json(&json!({ "ok": false, "signed_download_url": "", "matched_key": "" })), |
| 385 | + | Outcome::Ok(_) => answer(json!({ "ok": false, "signed_download_url": "", "matched_key": "" })), |
| 207 | 386 | | Outcome::Fail(refused) => twirp_failure(&refused), |
| 208 | 387 | | } |
| 209 | 388 | | } |
| ⋯ |
| 212 | 391 | | let reserved: Outcome<CacheReservation> = g1t_kit::call(actions, "cache_reserve", &args).await?; |
| 213 | 392 | | let reserved = match reserved { |
| 214 | 393 | | Outcome::Ok(reserved) => reserved, |
| 215 | | − | // The client warns with this and goes on, as for a key |
| 216 | | − | // another job is saving. |
| 217 | | − | Outcome::Fail(refused) => return Response::from_json(&json!({ "ok": false, "signed_upload_url": "", "message": refused.message })), |
| 394 | + | // A key already saved, or being saved by another job, to a |
| 395 | + | // protobuf client (OpenDAL's) is Twirp's `already_exists` |
| 396 | + | // (409), which it takes as "someone else has it": sccache |
| 397 | + | // then still writes. An `ok: false` would read as a broken |
| 398 | + | // cache, and sccache would only read from it. |
| 399 | + | Outcome::Fail(refused) if binary && refused.code == FailureCode::Conflict => return twirp_failure(&refused), |
| 400 | + | // The toolkit's client logs this ("another job may be |
| 401 | + | // creating this cache") and goes on. |
| 402 | + | Outcome::Fail(refused) => return answer(json!({ "ok": false, "signed_upload_url": "", "message": refused.message })), |
| 218 | 403 | | }; |
| 219 | 404 | | match start_upload(&bucket, services, &job, &token, "cache", &reserved.id, &reserved.object).await? { |
| 220 | | − | Outcome::Ok(url) => Response::from_json(&json!({ "ok": true, "signed_upload_url": url })), |
| 221 | | − | Outcome::Fail(refused) => Response::from_json(&json!({ "ok": false, "signed_upload_url": "", "message": refused.message })), |
| 405 | + | Outcome::Ok(url) => answer(json!({ "ok": true, "signed_upload_url": url })), |
| 406 | + | Outcome::Fail(refused) => answer(json!({ "ok": false, "signed_upload_url": "", "message": refused.message })), |
| 222 | 407 | | } |
| 223 | 408 | | } |
| 224 | 409 | | (CACHE_SERVICE, "FinalizeCacheEntryUpload") => { |
| ⋯ |
| 226 | 411 | | let pending: Outcome<CacheReservation> = g1t_kit::call(actions, "cache_upload", &args).await?; |
| 227 | 412 | | let pending = match pending { |
| 228 | 413 | | Outcome::Ok(pending) => pending, |
| 229 | | − | Outcome::Fail(refused) => return Response::from_json(&json!({ "ok": false, "entry_id": "0", "message": refused.message })), |
| 414 | + | Outcome::Fail(refused) => return answer(json!({ "ok": false, "entry_id": "0", "message": refused.message })), |
| 230 | 415 | | }; |
| 231 | 416 | | let Some(object) = bucket.head(&pending.object).await? else { |
| 232 | | − | return Response::from_json(&json!({ "ok": false, "entry_id": "0", "message": "Nothing was uploaded for that entry." })); |
| 417 | + | return answer(json!({ "ok": false, "entry_id": "0", "message": "Nothing was uploaded for that entry." })); |
| 233 | 418 | | }; |
| 234 | 419 | | match commit_cache(&bucket, services, &job, &token, &pending.id, object.size()).await? { |
| 235 | | − | Outcome::Ok(()) => Response::from_json(&json!({ "ok": true, "entry_id": pending.number.to_string() })), |
| 236 | | − | Outcome::Fail(refused) => Response::from_json(&json!({ "ok": false, "entry_id": "0", "message": refused.message })), |
| 420 | + | Outcome::Ok(()) => answer(json!({ "ok": true, "entry_id": pending.number.to_string() })), |
| 421 | + | Outcome::Fail(refused) => answer(json!({ "ok": false, "entry_id": "0", "message": refused.message })), |
| 237 | 422 | | } |
| 238 | 423 | | } |
| 239 | 424 | | (ARTIFACT_SERVICE, "CreateArtifact") => { |
| ⋯ |
| 321 | 506 | | } |
| 322 | 507 | | |
| 323 | 508 | | /// The part a chunk of the older protocol is, from its `Content-Range`: |
| 324 | | − | /// chunks are `CACHE_PART_BYTES` apart, as the toolkit sends them. |
| 509 | + | /// chunks are `CACHE_PART_BYTES` apart, as the toolkit sends them. A first |
| 510 | + | /// chunk may be larger, up to `MAX_BLOCK_BYTES`: a client that sends an |
| 511 | + | /// entry in one request (sccache does) sends a single chunk from 0. |
| 325 | 512 | | pub fn chunk_part(range: &str) -> Option<(u16, u64)> { |
| 326 | 513 | | let range = range.trim().strip_prefix("bytes ")?; |
| 327 | 514 | | let (span, _) = range.split_once('/')?; |
| 328 | 515 | | let (start, end) = span.split_once('-')?; |
| 329 | 516 | | let (start, end): (u64, u64) = (start.trim().parse().ok()?, end.trim().parse().ok()?); |
| 330 | | − | if end < start || start % CACHE_PART_BYTES != 0 || end - start + 1 > CACHE_PART_BYTES { |
| 517 | + | if end < start { |
| 331 | 518 | | return None; |
| 332 | 519 | | } |
| 333 | | − | Some(((start / CACHE_PART_BYTES + 1) as u16, end - start + 1)) |
| 520 | + | let length = end - start + 1; |
| 521 | + | if start == 0 && length <= MAX_BLOCK_BYTES { |
| 522 | + | return Some((1, length)); |
| 523 | + | } |
| 524 | + | if start % CACHE_PART_BYTES != 0 || length > CACHE_PART_BYTES { |
| 525 | + | return None; |
| 526 | + | } |
| 527 | + | Some(((start / CACHE_PART_BYTES + 1) as u16, length)) |
| 334 | 528 | | } |
| 335 | 529 | | |
| 530 | + | /// The bytes a download's `Range` header asks for, out of `size`: first and |
| 531 | + | /// last, inclusive. None to send the whole blob (no header, or one this |
| 532 | + | /// does not read, such as several ranges); `Some(None)` when the range is |
| 533 | + | /// past the end (416). |
| 534 | + | pub fn byte_range(header: &str, size: u64) -> Option<Option<(u64, u64)>> { |
| 535 | + | let spec = header.trim().strip_prefix("bytes=")?.trim(); |
| 536 | + | if spec.contains(',') { |
| 537 | + | return None; |
| 538 | + | } |
| 539 | + | let (first, last) = spec.split_once('-')?; |
| 540 | + | let (first, last) = (first.trim(), last.trim()); |
| 541 | + | let range = if first.is_empty() { |
| 542 | + | // The last `n` bytes. |
| 543 | + | let n: u64 = last.parse().ok()?; |
| 544 | + | if n == 0 || size == 0 { |
| 545 | + | return Some(None); |
| 546 | + | } |
| 547 | + | (size.saturating_sub(n), size - 1) |
| 548 | + | } else { |
| 549 | + | let first: u64 = first.parse().ok()?; |
| 550 | + | let last: u64 = if last.is_empty() { u64::MAX } else { last.parse().ok()? }; |
| 551 | + | if last < first { |
| 552 | + | return None; |
| 553 | + | } |
| 554 | + | if first >= size { |
| 555 | + | return Some(None); |
| 556 | + | } |
| 557 | + | (first, last.min(size - 1)) |
| 558 | + | }; |
| 559 | + | Some(Some(range)) |
| 560 | + | } |
| 561 | + | |
| 336 | 562 | | /// `{ACTIONS_CACHE_URL}_apis/artifactcache/…`. `rest` is the path after it. |
| 337 | 563 | | pub async fn cache_v1(mut request: Request, env: &Env, services: &Services, method: &str, rest: &str) -> Result<Response> { |
| 338 | 564 | | let token = bearer(&request); |
| ⋯ |
| 527 | 753 | | headers.set("content-length", &size.to_string())?; |
| 528 | 754 | | headers.set("content-type", grant.content_type.as_deref().unwrap_or("application/octet-stream"))?; |
| 529 | 755 | | headers.set("x-ms-blob-type", "BlockBlob")?; |
| 756 | + | headers.set("accept-ranges", "bytes")?; |
| 530 | 757 | | if let Some(name) = &grant.filename { |
| 531 | 758 | | headers.set("content-disposition", &format!("attachment; filename=\"{}\"", name.replace('"', "")))?; |
| 532 | 759 | | } |
| ⋯ |
| 538 | 765 | | headers(&mut response, object.size())?; |
| 539 | 766 | | return Ok(response); |
| 540 | 767 | | } |
| 768 | + | // One byte range (`Range`, or Azure's `x-ms-range`): the |
| 769 | + | // toolkit's client fetches a large entry in segments, side by |
| 770 | + | // side, and writes each where its range says. |
| 771 | + | let asked = match request.headers().get("x-ms-range")? { |
| 772 | + | Some(range) => Some(range), |
| 773 | + | None => request.headers().get("range")?, |
| 774 | + | }; |
| 775 | + | if let Some(asked) = asked.filter(|r| !r.trim().is_empty()) { |
| 776 | + | let Some(object) = bucket.head(&grant.object).await? else { return azure_error(404, "BlobNotFound", "It is gone.") }; |
| 777 | + | let size = object.size(); |
| 778 | + | match byte_range(&asked, size) { |
| 779 | + | Some(Some((first, last))) => { |
| 780 | + | let length = last - first + 1; |
| 781 | + | let Some(object) = bucket.get(&grant.object).range(worker::Range::OffsetWithLength { offset: first, length }).execute().await? else { |
| 782 | + | return azure_error(404, "BlobNotFound", "It is gone."); |
| 783 | + | }; |
| 784 | + | let Some(body) = object.body() else { return azure_error(404, "BlobNotFound", "It is gone.") }; |
| 785 | + | let mut response = Response::from_body(body.response_body()?)?.with_status(206); |
| 786 | + | headers(&mut response, length)?; |
| 787 | + | response.headers_mut().set("content-range", &format!("bytes {first}-{last}/{size}"))?; |
| 788 | + | return Ok(response); |
| 789 | + | } |
| 790 | + | Some(None) => { |
| 791 | + | let mut response = azure_error(416, "InvalidRange", "The range is past the end of the blob.")?; |
| 792 | + | response.headers_mut().set("content-range", &format!("bytes */{size}"))?; |
| 793 | + | return Ok(response); |
| 794 | + | } |
| 795 | + | // Not a range this reads: the whole blob. |
| 796 | + | None => {} |
| 797 | + | } |
| 798 | + | } |
| 541 | 799 | | let Some(object) = bucket.get(&grant.object).execute().await? else { return azure_error(404, "BlobNotFound", "It is gone.") }; |
| 542 | 800 | | let size = object.size(); |
| 543 | 801 | | let Some(body) = object.body() else { return azure_error(404, "BlobNotFound", "It is gone.") }; |
| ⋯ |
| 674 | 932 | | let mb32 = CACHE_PART_BYTES; |
| 675 | 933 | | assert_eq!(chunk_part(&format!("bytes 0-{}/*", mb32 - 1)), Some((1, mb32))); |
| 676 | 934 | | assert_eq!(chunk_part(&format!("bytes {}-{}/*", mb32 * 2, mb32 * 2 + 99)), Some((3, 100))); |
| 677 | | − | // Not on a chunk's boundary, too long, or not a range. |
| 935 | + | // A whole entry in one chunk, as sccache (OpenDAL) sends it: its |
| 936 | + | // check file is 13 bytes, a compiled crate can be well over 32 MB. |
| 937 | + | assert_eq!(chunk_part("bytes 0-12/*"), Some((1, 13))); |
| 938 | + | assert_eq!(chunk_part(&format!("bytes 0-{}/*", mb32)), Some((1, mb32 + 1))); |
| 939 | + | assert_eq!(chunk_part(&format!("bytes 0-{}/*", MAX_BLOCK_BYTES - 1)), Some((1, MAX_BLOCK_BYTES))); |
| 940 | + | // Not on a chunk's boundary, too long, backwards, or not a range. |
| 678 | 941 | | assert_eq!(chunk_part("bytes 5-10/*"), None); |
| 679 | | − | assert_eq!(chunk_part(&format!("bytes 0-{}/*", mb32)), None); |
| 942 | + | assert_eq!(chunk_part(&format!("bytes {mb32}-{}/*", mb32 * 2)), None); |
| 943 | + | assert_eq!(chunk_part(&format!("bytes 0-{}/*", MAX_BLOCK_BYTES)), None); |
| 944 | + | assert_eq!(chunk_part("bytes 10-5/*"), None); |
| 680 | 945 | | assert_eq!(chunk_part("0-10"), None); |
| 681 | 946 | | } |
| 682 | 947 | | |
| 948 | + | #[test] |
| 949 | + | fn downloads_read_one_byte_range() { |
| 950 | + | // OpenDAL's stat: the first byte. |
| 951 | + | assert_eq!(byte_range("bytes=0-0", 100), Some(Some((0, 0)))); |
| 952 | + | // The toolkit's segments, the last one cut at the end. |
| 953 | + | assert_eq!(byte_range("bytes=0-49", 100), Some(Some((0, 49)))); |
| 954 | + | assert_eq!(byte_range("bytes=50-999", 100), Some(Some((50, 99)))); |
| 955 | + | assert_eq!(byte_range("bytes=90-", 100), Some(Some((90, 99)))); |
| 956 | + | assert_eq!(byte_range("bytes=-10", 100), Some(Some((90, 99)))); |
| 957 | + | assert_eq!(byte_range("bytes=-1000", 100), Some(Some((0, 99)))); |
| 958 | + | // Past the end: 416. |
| 959 | + | assert_eq!(byte_range("bytes=100-200", 100), Some(None)); |
| 960 | + | assert_eq!(byte_range("bytes=0-0", 0), Some(None)); |
| 961 | + | assert_eq!(byte_range("bytes=-0", 100), Some(None)); |
| 962 | + | // Not read: the whole blob. |
| 963 | + | assert_eq!(byte_range("bytes=0-1,5-6", 100), None); |
| 964 | + | assert_eq!(byte_range("bytes=9-3", 100), None); |
| 965 | + | assert_eq!(byte_range("items=0-1", 100), None); |
| 966 | + | assert_eq!(byte_range("bytes=a-b", 100), None); |
| 967 | + | } |
| 968 | + | |
| 969 | + | #[test] |
| 970 | + | fn upload_links_carry_a_query_as_sas_links_do() { |
| 971 | + | let url = upload_url("https://api.g1t.sh", "tok.sig"); |
| 972 | + | assert_eq!(url, "https://api.g1t.sh/actions/toolkit/blobs/tok.sig?sv=2024-11-04"); |
| 973 | + | // How OpenDAL reads a signed upload link: a container, a blob in |
| 974 | + | // it, and a SAS query, all of which must be there. |
| 975 | + | let rest = url.strip_prefix("https://api.g1t.sh/").unwrap(); |
| 976 | + | let (path, query) = rest.split_once('?').unwrap(); |
| 977 | + | let (container, blob) = path.split_once('/').unwrap(); |
| 978 | + | assert_eq!((container, blob, query), ("actions", "toolkit/blobs/tok.sig", "sv=2024-11-04")); |
| 979 | + | } |
| 980 | + | |
| 981 | + | /// A protobuf length-delimited field, as prost writes it. |
| 982 | + | fn pb_text(number: u8, text: &str) -> Vec<u8> { |
| 983 | + | let mut out = vec![(number << 3) | 2, text.len() as u8]; |
| 984 | + | out.extend_from_slice(text.as_bytes()); |
| 985 | + | out |
| 986 | + | } |
| 987 | + | |
| 988 | + | /// The requests sccache 0.18 sends (OpenDAL 0.58's `ghac` service, with |
| 989 | + | /// prost): fields in number order, defaults left out, no metadata. |
| 990 | + | #[test] |
| 991 | + | fn twirp_reads_sccaches_protobuf_requests() { |
| 992 | + | assert!(is_protobuf("application/protobuf")); |
| 993 | + | assert!(is_protobuf("Application/Protobuf; charset=utf-8")); |
| 994 | + | assert!(!is_protobuf("application/json")); |
| 995 | + | assert!(!is_protobuf("")); |
| 996 | + | |
| 997 | + | let key = "sccache/f/c/b/fcb0a1d2e3"; |
| 998 | + | let version = "sccache-v0.18.0"; |
| 999 | + | let create = [pb_text(2, key), pb_text(3, version)].concat(); |
| 1000 | + | let read = proto::request("CreateCacheEntry", &create).unwrap(); |
| 1001 | + | assert_eq!((text(&read, "key"), text(&read, "version")), (key.to_owned(), version.to_owned())); |
| 1002 | + | |
| 1003 | + | // size_bytes is field 3, a varint: 300 is 0xac 0x02. |
| 1004 | + | let finalize = [pb_text(2, key), vec![0x18, 0xac, 0x02], pb_text(4, version)].concat(); |
| 1005 | + | let read = proto::request("FinalizeCacheEntryUpload", &finalize).unwrap(); |
| 1006 | + | assert_eq!(number(&read, "size_bytes"), Some(300)); |
| 1007 | + | assert_eq!(text(&read, "version"), version); |
| 1008 | + | |
| 1009 | + | let lookup = [pb_text(2, key), pb_text(4, version)].concat(); |
| 1010 | + | let read = proto::request("GetCacheEntryDownloadURL", &lookup).unwrap(); |
| 1011 | + | assert_eq!(text(&read, "key"), key); |
| 1012 | + | assert!(field(&read, "restore_keys").is_null()); |
| 1013 | + | // The toolkit's own lookup, with metadata (skipped) and restore keys. |
| 1014 | + | let metadata = vec![0x0a, 0x02, 0x08, 0x07]; |
| 1015 | + | let with_restore = [metadata, pb_text(2, "k"), pb_text(3, "k-"), pb_text(3, "x-"), pb_text(4, "v")].concat(); |
| 1016 | + | let read = proto::request("GetCacheEntryDownloadURL", &with_restore).unwrap(); |
| 1017 | + | assert_eq!(field(&read, "restore_keys"), &json!(["k-", "x-"])); |
| 1018 | + | assert_eq!(text(&read, "version"), "v"); |
| 1019 | + | |
| 1020 | + | // The same three, as prost 0.14 encodes them with OpenDAL's |
| 1021 | + | // generated types (the `ghac` crate, 0.3.0), byte for byte. |
| 1022 | + | let recorded = |hex: &str| -> Vec<u8> { (0..hex.len()).step_by(2).map(|i| u8::from_str_radix(&hex[i..i + 2], 16).unwrap()).collect() }; |
| 1023 | + | let prefix = "1218736363616368652f662f632f622f66636230613164326533"; |
| 1024 | + | let suffix = "0f736363616368652d76302e31382e30"; |
| 1025 | + | assert_eq!(recorded(&format!("{prefix}1a{suffix}")), create); |
| 1026 | + | assert_eq!(recorded(&format!("{prefix}18ac0222{suffix}")), finalize); |
| 1027 | + | assert_eq!(recorded(&format!("{prefix}22{suffix}")), lookup); |
| 1028 | + | |
| 1029 | + | // Cut short, or not a cache method. |
| 1030 | + | assert!(proto::request("CreateCacheEntry", &create[..create.len() - 1]).is_none()); |
| 1031 | + | assert!(proto::request("CreateCacheEntry", &[0x12, 0xff]).is_none()); |
| 1032 | + | assert!(proto::request("CreateArtifact", &create).is_none()); |
| 1033 | + | assert_eq!(proto::request("CreateCacheEntry", &[]), Some(json!({}))); |
| 1034 | + | } |
| 1035 | + | |
| 1036 | + | #[test] |
| 1037 | + | fn twirp_answers_in_protobuf_as_prost_reads_it() { |
| 1038 | + | let url = "https://api.g1t.sh/actions/toolkit/blobs/t?sv=2024-11-04"; |
| 1039 | + | let created = proto::response("CreateCacheEntry", &json!({ "ok": true, "signed_upload_url": url })); |
| 1040 | + | assert_eq!(created, [vec![0x08, 0x01], pb_text(2, url)].concat()); |
| 1041 | + | // Refused: ok false is the default, so only the message is sent. |
| 1042 | + | let refused = proto::response("CreateCacheEntry", &json!({ "ok": false, "signed_upload_url": "", "message": "no" })); |
| 1043 | + | assert_eq!(refused, pb_text(3, "no")); |
| 1044 | + | // entry_id is an int64, given as a string in JSON. |
| 1045 | + | let finalized = proto::response("FinalizeCacheEntryUpload", &json!({ "ok": true, "entry_id": "300" })); |
| 1046 | + | assert_eq!(finalized, vec![0x08, 0x01, 0x10, 0xac, 0x02]); |
| 1047 | + | let found = proto::response("GetCacheEntryDownloadURL", &json!({ "ok": true, "signed_download_url": "u", "matched_key": "k" })); |
| 1048 | + | assert_eq!(found, [vec![0x08, 0x01], pb_text(2, "u"), pb_text(3, "k")].concat()); |
| 1049 | + | // A miss is an empty message: ok false. |
| 1050 | + | assert!(proto::response("GetCacheEntryDownloadURL", &json!({ "ok": false, "signed_download_url": "", "matched_key": "" })).is_empty()); |
| 1051 | + | } |
| 1052 | + | |
| 683 | 1053 | | /// The toolkit's requests, as `@actions/cache` 4 and `@actions/artifact` |
| 684 | 1054 | | /// 2 send them (protobuf-ts, proto field names, no defaults). |
| 685 | 1055 | | #[test] |