Skip to content
Sign in

flagon-io/g1t

public
0

Commit

Merge empty successes across services: Ok(None) and Ok(()) read back as themselves, so every cache miss is a miss and not a 500, and toolkit failures say why

syntaqxcommitted Parentsbff2c6d38e2e16Browse files
3 files+167−160/3 viewed
+113−15
333333 })
334334 }
335335
336−/// `POST /twirp/{service}/{method}`.
337−pub async fn twirp(mut request: Request, env: &Env, services: &Services, service: &str, method: &str) -> Result<Response> {
336+/// `POST /twirp/{service}/{method}`. A failure inside is logged and
337+/// answered as Twirp's `internal`, with its cause.
338+pub async fn twirp(request: Request, env: &Env, services: &Services, service: &str, method: &str) -> Result<Response> {
339+ match twirp_inner(request, env, services, service, method).await {
340+ Ok(response) => Ok(response),
341+ Err(error) => twirp_error("internal", &failed(&format!("POST /twirp/{service}/{method}"), &error)),
342+ }
343+}
344+
345+async fn twirp_inner(mut request: Request, env: &Env, services: &Services, service: &str, method: &str) -> Result<Response> {
338346 let token = bearer(&request);
339347 let Some(job) = runtime_job(&token) else {
340348 return twirp_error("unauthenticated", "Send the job's ACTIONS_RUNTIME_TOKEN as a bearer token.");
502510 }
503511
504512 fn query(request: &Request, name: &str) -> Option<String> {
505− request.url().ok()?.query_pairs().find(|(k, _)| k == name).map(|(_, v)| v.into_owned())
513+ query_in(&request.url().ok()?, name)
514+}
515+
516+fn query_in(url: &worker::Url, name: &str) -> Option<String> {
517+ url.query_pairs().find(|(k, _)| k == name).map(|(_, v)| v.into_owned())
506518 }
507519
508520 /// The part a chunk of the older protocol is, from its `Content-Range`:
559571 Some(Some(range))
560572 }
561573
574+/// What a lookup of the older protocol answers: 200 with the entry, 204
575+/// for a miss (which the toolkit's client and sccache read as "not
576+/// cached"), or the refusal's status.
577+pub fn lookup_answer(found: Outcome<Option<CacheHit>>, version: &str, api: &str) -> (u16, Option<Value>) {
578+ match found {
579+ Outcome::Ok(Some(CacheHit { key, blob: Some(blob), created_at, .. })) => (
580+ 200,
581+ Some(json!({
582+ "cacheKey": key,
583+ "cacheVersion": version,
584+ "scope": "",
585+ "creationTime": created_at,
586+ "archiveLocation": blob_url(api, &blob),
587+ })),
588+ ),
589+ // No entry, or one without a download link (no ACTIONS_KEY): a miss.
590+ Outcome::Ok(_) => (204, None),
591+ Outcome::Fail(refused) => (refused.code.http_status(), Some(json!({ "message": refused.message, "error": { "message": refused.message } }))),
592+ }
593+}
594+
595+/// Logs a toolkit request that failed inside g1t, and says what to tell
596+/// its client: the cause, so a job's log shows more than a bare 500.
597+fn failed(route: &str, error: &worker::Error) -> String {
598+ worker::console_error!("toolkit: {route} failed: {error}");
599+ format!("g1t could not answer this: {error}")
600+}
601+
562602 /// `{ACTIONS_CACHE_URL}_apis/artifactcache/…`. `rest` is the path after it.
563−pub async fn cache_v1(mut request: Request, env: &Env, services: &Services, method: &str, rest: &str) -> Result<Response> {
603+/// A failure inside is logged and answered as a 500 with its cause.
604+pub async fn cache_v1(request: Request, env: &Env, services: &Services, method: &str, rest: &str) -> Result<Response> {
605+ match cache_v1_inner(request, env, services, method, rest).await {
606+ Ok(response) => Ok(response),
607+ Err(error) => plain_error(500, &failed(&format!("{method} {CACHE_PATH}_apis/artifactcache/{rest}"), &error)),
608+ }
609+}
610+
611+async fn cache_v1_inner(mut request: Request, env: &Env, services: &Services, method: &str, rest: &str) -> Result<Response> {
564612 let token = bearer(&request);
565613 let Some(job) = runtime_job(&token) else {
566614 return plain_error(401, "Send the job's ACTIONS_RUNTIME_TOKEN as a bearer token.");
577625 let version = query(&request, "version").unwrap_or_default();
578626 let args = CacheLookupArgs { job, token, key: key.clone(), restore: restore.to_vec(), version: Some(version.clone()) };
579627 let found: Outcome<Option<CacheHit>> = g1t_kit::call(actions, "cache_lookup", &args).await?;
580− match found {
581− Outcome::Ok(Some(CacheHit { key, blob: Some(blob), created_at, .. })) => Response::from_json(&json!({
582− "cacheKey": key,
583− "cacheVersion": version,
584− "scope": "",
585− "creationTime": created_at,
586− "archiveLocation": blob_url(&services.addresses.api, &blob),
587− })),
588− Outcome::Ok(_) => Ok(Response::empty()?.with_status(204)),
589− Outcome::Fail(refused) => plain_error(refused.code.http_status(), &refused.message),
628+ match lookup_answer(found, &version, &services.addresses.api) {
629+ (status, Some(body)) => Ok(Response::from_json(&body)?.with_status(status)),
630+ (status, None) => Ok(Response::empty()?.with_status(status)),
590631 }
591632 }
592633 ("POST", ["caches"]) => {
736777 }
737778
738779 /// `/actions/toolkit/blobs/{token}`: GET or HEAD a download, PUT an upload.
739−pub async fn blob(mut request: Request, env: &Env, services: &Services, method: &str, token: &str) -> Result<Response> {
780+/// A failure inside is logged and answered as Azure's `InternalError`.
781+pub async fn blob(request: Request, env: &Env, services: &Services, method: &str, token: &str) -> Result<Response> {
782+ match blob_inner(request, env, services, method, token).await {
783+ Ok(response) => Ok(response),
784+ // The token is a credential: the route is logged without it.
785+ Err(error) => azure_error(500, "InternalError", &failed(&format!("{method} /actions/toolkit/blobs/…"), &error)),
786+ }
787+}
788+
789+async fn blob_inner(mut request: Request, env: &Env, services: &Services, method: &str, token: &str) -> Result<Response> {
740790 let opened: Outcome<BlobGrant> = g1t_kit::call(&services.actions, "blob_open", &BlobArgs { blob: token.to_owned(), ..BlobArgs::default() }).await?;
741791 let grant = match opened {
742792 Outcome::Ok(grant) => grant,
10861136 assert_eq!(runtime_job("deadbeef"), None);
10871137 }
10881138
1139+ /// sccache 0.18's storage check, at server start: a lookup of
1140+ /// `sccache/.sccache_check`. The actions service answers a miss with
1141+ /// `Ok(None)`, `{"ok":true,"value":null}`, which was read back as a
1142+ /// malformed outcome, and every lookup that missed was a 500
1143+ /// ("Server startup failed: cache storage failed to read").
1144+ #[test]
1145+ fn sccaches_first_lookup_misses_with_a_204() {
1146+ let url = worker::Url::parse(
1147+ "https://api.g1t.sh/actions/toolkit/_apis/artifactcache/cache?keys=sccache/.sccache_check&version=sccache-v0.18.0",
1148+ )
1149+ .unwrap();
1150+ assert_eq!(query_in(&url, "keys").as_deref(), Some("sccache/.sccache_check"));
1151+ assert_eq!(query_in(&url, "version").as_deref(), Some("sccache-v0.18.0"));
1152+
1153+ // As the actions service replies (`g1t_kit::reply`), and the API
1154+ // reads it (`g1t_kit::call`).
1155+ let wire = serde_json::to_string(&Outcome::<Option<CacheHit>>::Ok(None)).unwrap();
1156+ assert_eq!(wire, r#"{"ok":true,"value":null}"#);
1157+ let found: Outcome<Option<CacheHit>> = g1t_kit::read_answer("cache_lookup", &wire).unwrap();
1158+ assert_eq!(lookup_answer(found, "sccache-v0.18.0", "https://api.g1t.sh"), (204, None));
1159+
1160+ // Once saved, the same lookup is a hit with its download link.
1161+ let hit = CacheHit {
1162+ key: "sccache/.sccache_check".into(),
1163+ object: "c/repo_1/cache_1".into(),
1164+ size: 13,
1165+ created_at: "2026-10-08T12:00:00.000Z".into(),
1166+ blob: Some("tok.sig".into()),
1167+ };
1168+ let wire = serde_json::to_string(&Outcome::Ok(Some(hit))).unwrap();
1169+ let found: Outcome<Option<CacheHit>> = g1t_kit::read_answer("cache_lookup", &wire).unwrap();
1170+ let (status, body) = lookup_answer(found, "sccache-v0.18.0", "https://api.g1t.sh");
1171+ let body = body.unwrap();
1172+ assert_eq!(status, 200);
1173+ assert_eq!(body["cacheKey"], "sccache/.sccache_check");
1174+ assert_eq!(body["cacheVersion"], "sccache-v0.18.0");
1175+ assert_eq!(body["archiveLocation"], "https://api.g1t.sh/actions/toolkit/blobs/tok.sig");
1176+
1177+ // A refusal keeps its status and says why.
1178+ let refused = Outcome::<Option<CacheHit>>::fail(FailureCode::Unauthenticated, "That job is not running.");
1179+ let (status, body) = lookup_answer(refused, "v", "https://api.g1t.sh");
1180+ assert_eq!((status, body.unwrap()["message"].as_str()), (401, Some("That job is not running.")));
1181+
1182+ // An answer that does not read names its method and the cause.
1183+ let unread = g1t_kit::read_answer::<Outcome<CacheHit>>("cache_lookup", r#"{"ok":true,"value":null}"#).unwrap_err();
1184+ assert!(unread.to_string().contains("cache_lookup answered with what could not be read"), "{unread}");
1185+ }
1186+
10891187 #[test]
10901188 fn a_job_is_told_where_the_toolkit_s_services_are() {
10911189 let vars = runtime_variables("https://api.g1t.sh", "tok", false);
+42−0
11 use serde::de::DeserializeOwned;
2+use serde::de::value::UnitDeserializer;
23 use serde::{Deserialize, Deserializer, Serialize, Serializer};
34
45 #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
114115 match (wire.ok, wire.value, wire.error) {
115116 (true, Some(value), _) => Ok(Outcome::Ok(value)),
116117 (false, _, Some(error)) => Ok(Outcome::Fail(error)),
118+ // `"value": null`, or no value: what `Ok(None)` and `Ok(())` are
119+ // written as. `Option<Option<T>>` reads a null as the outer
120+ // `None`, so the value is read from nothing instead: `None` for
121+ // an `Option`, `()` for a unit, and still malformed for a type
122+ // that needs a value.
123+ (true, None, _) => T::deserialize(UnitDeserializer::<D::Error>::new())
124+ .map(Outcome::Ok)
125+ .map_err(|_| serde::de::Error::custom("malformed outcome: ok without a value")),
117126 _ => Err(serde::de::Error::custom("malformed outcome")),
118127 }
119128 }
120129 }
130+
131+#[cfg(test)]
132+mod tests {
133+ use super::*;
134+
135+ fn round_trip<T: Serialize + DeserializeOwned>(outcome: &Outcome<T>) -> Result<Outcome<T>, String> {
136+ // As a service replies (`g1t_kit::reply`) and its caller reads it
137+ // (`g1t_kit::call`).
138+ let wire = serde_json::to_string(outcome).map_err(|e| e.to_string())?;
139+ serde_json::from_str(&wire).map_err(|e| e.to_string())
140+ }
141+
142+ /// A cache miss is `Ok(None)`, written `{"ok":true,"value":null}`. It
143+ /// was read back as malformed, so every miss of the toolkit's cache
144+ /// (sccache's first lookup, `sccache/.sccache_check`) was a 500.
145+ #[test]
146+ fn ok_none_and_ok_unit_cross_the_wire() {
147+ assert_eq!(serde_json::to_string(&Outcome::<Option<u8>>::Ok(None)).unwrap(), r#"{"ok":true,"value":null}"#);
148+ assert!(matches!(round_trip(&Outcome::<Option<u8>>::Ok(None)), Ok(Outcome::Ok(None))));
149+ assert!(matches!(round_trip(&Outcome::Ok(Some(7u8))), Ok(Outcome::Ok(Some(7)))));
150+ assert!(matches!(round_trip(&Outcome::Ok(())), Ok(Outcome::Ok(()))));
151+ assert!(matches!(serde_json::from_str::<Outcome<Option<u8>>>(r#"{"ok":true}"#), Ok(Outcome::Ok(None))));
152+ let failed = round_trip(&Outcome::<Option<u8>>::fail(FailureCode::Unauthenticated, "no"));
153+ assert!(matches!(failed, Ok(Outcome::Fail(Failure { code: FailureCode::Unauthenticated, .. }))));
154+ }
155+
156+ #[test]
157+ fn an_ok_without_the_value_it_needs_is_still_malformed() {
158+ assert!(serde_json::from_str::<Outcome<u8>>(r#"{"ok":true,"value":null}"#).is_err());
159+ assert!(serde_json::from_str::<Outcome<String>>(r#"{"ok":true}"#).is_err());
160+ assert!(serde_json::from_str::<Outcome<u8>>(r#"{"ok":false}"#).is_err());
161+ }
162+}
+12−1
5757 response.text().await.unwrap_or_default()
5858 )));
5959 }
60− response.json().await
60+ // Say which call's answer did not read, and why: a bare serde error
61+ // ("JSON serialization error") is all a log would otherwise show. The
62+ // body is left out, as it may carry a signed link.
63+ let text = response.text().await?;
64+ read_answer(method, &text)
65+}
66+
67+/// A method's answer, read from its body.
68+pub fn read_answer<R: DeserializeOwned>(method: &str, text: &str) -> Result<R> {
69+ serde_json::from_str(text).map_err(|error| {
70+ worker::Error::RustError(format!("{method} answered with what could not be read: {error}"))
71+ })
6172 }
6273
6374 pub mod d1;