From ab013d97861b4237fd857c41b6434eade3ed99f3 Mon Sep 17 00:00:00 2001 From: Tee8z Date: Tue, 20 Oct 2026 17:30:38 -0400 Subject: [PATCH] feat(agents): verify sandbox checkpoint evidence against S3 Add the reaper's checkpoint evidence module. A key is recorded only when it sits directly under the session's own / prefix, with the prefix built from the reaper configuration and a validated (32 lowercase hex) Sandbox generation, never from the Pod, and an S3 ListObjectsV2 confirms it. Three sources name a key: the termination-log receipt, the newest object listed in the session prefix (for a receipt that was never observed), and a recovered session's inherited restore key, which may belong to an older ancestor generation. An existence check lists with the key itself as the prefix, so it needs one request and no pagination. The newest-object listing follows all pages within a 25-second deadline and rejects missing or repeated continuation tokens. A regression test finds the newest key after twelve listing pages. Any store failure defers the decision instead of recording a missing checkpoint, and buzz_sandbox_reaper_s3_errors_total carries a bounded kind label (timeout, access_denied, not_found_bucket, other). The S3 adapter is tested against a local HTTP endpoint. This brings in the AWS SDK dependencies and documents the checkpoint evidence rules in docs/remote-agents.md. Signed-off-by: Tee8z --- Cargo.lock | 578 +++++++++++++++-- crates/buzz-sandbox-reaper/Cargo.toml | 7 + crates/buzz-sandbox-reaper/src/evidence.rs | 721 +++++++++++++++++++++ crates/buzz-sandbox-reaper/src/main.rs | 10 +- crates/buzz-sandbox-reaper/src/plan.rs | 58 +- docs/remote-agents.md | 24 + 6 files changed, 1316 insertions(+), 82 deletions(-) create mode 100644 crates/buzz-sandbox-reaper/src/evidence.rs diff --git a/Cargo.lock b/Cargo.lock index fa15535ee26..0be94ad1609 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -501,7 +501,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "16e2cdb6d5ed835199484bb92bb8b3edd526effe995c61732580439c1a67e2e9" dependencies = [ "base64 0.22.1", - "http", + "http 1.4.2", "log", "rustls", "serde", @@ -516,6 +516,44 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" +[[package]] +name = "aws-config" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8d7b388a9fc3a6db15a5ec778c38b354eff1364882c94d08e0252f7a47dcaa4" +dependencies = [ + "aws-credential-types", + "aws-runtime", + "aws-sdk-sts", + "aws-smithy-async", + "aws-smithy-http", + "aws-smithy-json", + "aws-smithy-runtime", + "aws-smithy-runtime-api", + "aws-smithy-schema", + "aws-smithy-types", + "aws-types", + "bytes", + "fastrand", + "http 1.4.2", + "time", + "tokio", + "tracing", + "url", +] + +[[package]] +name = "aws-credential-types" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e93964ffdaf57857f544be3666a5f57570bb699e934700f11b49708f61bb556e" +dependencies = [ + "aws-smithy-async", + "aws-smithy-runtime-api", + "aws-smithy-types", + "zeroize", +] + [[package]] name = "aws-creds" version = "0.39.1" @@ -565,6 +603,358 @@ dependencies = [ "thiserror 2.0.18", ] +[[package]] +name = "aws-runtime" +version = "1.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2b8a9911551b4ea6ca13805ef52ed96f7d2bbb43cc3b4a14cb0776a71f33cfaa" +dependencies = [ + "aws-credential-types", + "aws-sigv4", + "aws-smithy-async", + "aws-smithy-eventstream", + "aws-smithy-http", + "aws-smithy-runtime", + "aws-smithy-runtime-api", + "aws-smithy-types", + "aws-types", + "bytes", + "bytes-utils", + "fastrand", + "http 0.2.12", + "http 1.4.2", + "http-body 0.4.6", + "http-body 1.0.1", + "percent-encoding", + "pin-project-lite", + "tracing", + "uuid", +] + +[[package]] +name = "aws-sdk-s3" +version = "1.143.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a0ade5433c9561daac7c0c6bc910f1240b4f8ec0d6148b0b463aac0691d747c9" +dependencies = [ + "arc-swap", + "aws-credential-types", + "aws-runtime", + "aws-sigv4", + "aws-smithy-async", + "aws-smithy-checksums", + "aws-smithy-eventstream", + "aws-smithy-http", + "aws-smithy-json", + "aws-smithy-observability", + "aws-smithy-runtime", + "aws-smithy-runtime-api", + "aws-smithy-schema", + "aws-smithy-types", + "aws-smithy-xml", + "aws-types", + "bytes", + "fastrand", + "hex", + "hmac 0.13.0", + "http 0.2.12", + "http 1.4.2", + "http-body 1.0.1", + "lru 0.16.4", + "percent-encoding", + "regex-lite", + "sha2 0.11.0", + "tracing", + "url", +] + +[[package]] +name = "aws-sdk-sts" +version = "1.119.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "03e490aa904849b38e770922faac779dd245f5cf8be81f19065acf18d276e4ad" +dependencies = [ + "arc-swap", + "aws-credential-types", + "aws-runtime", + "aws-smithy-async", + "aws-smithy-http", + "aws-smithy-json", + "aws-smithy-observability", + "aws-smithy-query", + "aws-smithy-runtime", + "aws-smithy-runtime-api", + "aws-smithy-schema", + "aws-smithy-types", + "aws-smithy-xml", + "aws-types", + "fastrand", + "http 1.4.2", + "regex-lite", + "tracing", +] + +[[package]] +name = "aws-sigv4" +version = "1.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2312577f088c9fbf4206dfdb884cf1de9407b43e1a923cbed5237775116fc24b" +dependencies = [ + "aws-credential-types", + "aws-smithy-eventstream", + "aws-smithy-http", + "aws-smithy-runtime-api", + "aws-smithy-types", + "bytes", + "form_urlencoded", + "hex", + "hmac 0.13.0", + "http 1.4.2", + "percent-encoding", + "sha2 0.11.0", + "time", + "tracing", +] + +[[package]] +name = "aws-smithy-async" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f02e407fb3b54891734224b9ffac8a71fdd35f542500fa1af95754a6b2beb316" +dependencies = [ + "futures-util", + "pin-project-lite", + "tokio", +] + +[[package]] +name = "aws-smithy-checksums" +version = "0.65.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b67ecd999972b58e67cab052f5129906c08c25883bd0788ceefc55ef97d61307" +dependencies = [ + "aws-smithy-http", + "aws-smithy-types", + "bytes", + "crc-fast", + "hex", + "http 1.4.2", + "http-body 1.0.1", + "http-body-util", + "md-5", + "pin-project-lite", + "sha1 0.11.0", + "sha2 0.11.0", + "tracing", +] + +[[package]] +name = "aws-smithy-eventstream" +version = "0.61.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "80c2051c2f1016fb8e6548dd07b8bc2ac9c3fe583721444b92f515e856d31609" +dependencies = [ + "aws-smithy-types", + "bytes", + "crc32fast", +] + +[[package]] +name = "aws-smithy-http" +version = "0.64.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "639b4d8f8555f24a9be649811c3eb0b4d4616f4d61daf0c32e28873bc1ea9af1" +dependencies = [ + "aws-smithy-eventstream", + "aws-smithy-runtime-api", + "aws-smithy-types", + "bytes", + "bytes-utils", + "futures-core", + "futures-util", + "http 1.4.2", + "http-body 1.0.1", + "http-body-util", + "percent-encoding", + "pin-project-lite", + "pin-utils", + "tracing", +] + +[[package]] +name = "aws-smithy-http-client" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "51c89cc3f1f281d659a67a519a1b5c6d445b5ce09fa7e5aee40c2c2707e9509d" +dependencies = [ + "aws-smithy-async", + "aws-smithy-runtime-api", + "aws-smithy-types", + "h2", + "http 1.4.2", + "hyper", + "hyper-rustls", + "hyper-util", + "pin-project-lite", + "rustls", + "rustls-native-certs", + "rustls-pki-types", + "tokio", + "tokio-rustls", + "tower 0.5.3", + "tracing", +] + +[[package]] +name = "aws-smithy-json" +version = "0.63.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3385d469edbe8b60cc72002784652b5efca39178192aa9cc4b44c9875c6bdc18" +dependencies = [ + "aws-smithy-runtime-api", + "aws-smithy-schema", + "aws-smithy-types", +] + +[[package]] +name = "aws-smithy-observability" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e86338c869539a581bf161247762a6e87f92c5c075060057b5ed6d06632ed0c" +dependencies = [ + "aws-smithy-runtime-api", +] + +[[package]] +name = "aws-smithy-query" +version = "0.62.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f1d1d71f6562be974caa85442ecd90194c40fdb5df045f182a6c2e872ce95056" +dependencies = [ + "aws-smithy-runtime-api", + "aws-smithy-schema", + "aws-smithy-types", + "aws-smithy-xml", + "urlencoding", +] + +[[package]] +name = "aws-smithy-runtime" +version = "1.16.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6e302ac1d88b99652489df31abdec6ac42a2ab2ac3982ad0ac49f64dfaf28ba" +dependencies = [ + "aws-smithy-async", + "aws-smithy-http", + "aws-smithy-http-client", + "aws-smithy-observability", + "aws-smithy-runtime-api", + "aws-smithy-schema", + "aws-smithy-types", + "bytes", + "fastrand", + "http 1.4.2", + "http-body 1.0.1", + "http-body-util", + "pin-project-lite", + "pin-utils", + "tokio", + "tracing", +] + +[[package]] +name = "aws-smithy-runtime-api" +version = "1.19.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0730c16f91124c6a2abb4932c77e299288b3dd9f967ea2e9ec48cc6731e87a4" +dependencies = [ + "aws-smithy-async", + "aws-smithy-runtime-api-macros", + "aws-smithy-types", + "bytes", + "http 0.2.12", + "http 1.4.2", + "pin-project-lite", + "tokio", + "tracing", + "zeroize", +] + +[[package]] +name = "aws-smithy-runtime-api-macros" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "221eaa237ddf1ca79b60d1372aad77e47f9c0ea5b3ce5099da8c61d027dc77b3" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.117", +] + +[[package]] +name = "aws-smithy-schema" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e8f395d93304280b64b7632fea798d177e74897fe7f063416ce627cd6fa24829" +dependencies = [ + "aws-smithy-runtime-api", + "aws-smithy-types", + "http 1.4.2", +] + +[[package]] +name = "aws-smithy-types" +version = "1.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69bb407740a197147da48238ecc94498493c9e85445732360cec180296ca45f1" +dependencies = [ + "base64-simd", + "bytes", + "bytes-utils", + "futures-core", + "http 0.2.12", + "http 1.4.2", + "http-body 0.4.6", + "http-body 1.0.1", + "http-body-util", + "itoa", + "num-integer", + "pin-project-lite", + "pin-utils", + "ryu", + "serde", + "time", + "tokio", + "tokio-util", +] + +[[package]] +name = "aws-smithy-xml" +version = "0.62.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b932c8d6dc127fc980eecd78f8694ae9b9551b69a93a7def2a199c1c0033daf" +dependencies = [ + "aws-smithy-runtime-api", + "aws-smithy-schema", + "aws-smithy-types", + "xmlparser", +] + +[[package]] +name = "aws-types" +version = "1.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "209f3a6d82a6e9e5f94abbed94c7a26e1c052341002bf57a5fb5481f625896fc" +dependencies = [ + "aws-credential-types", + "aws-smithy-async", + "aws-smithy-runtime-api", + "aws-smithy-schema", + "aws-smithy-types", + "rustc_version", + "tracing", +] + [[package]] name = "axum" version = "0.8.9" @@ -577,8 +967,8 @@ dependencies = [ "bytes", "form_urlencoded", "futures-util", - "http", - "http-body", + "http 1.4.2", + "http-body 1.0.1", "http-body-util", "hyper", "hyper-util", @@ -610,8 +1000,8 @@ checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1" dependencies = [ "bytes", "futures-core", - "http", - "http-body", + "http 1.4.2", + "http-body 1.0.1", "http-body-util", "mime", "pin-project-lite", @@ -673,6 +1063,16 @@ version = "0.23.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5" +[[package]] +name = "base64-simd" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "339abbe78e73178762e23bea9dfd08e697eb3f3301cd4be981c0f78ba5859195" +dependencies = [ + "outref", + "vsimd", +] + [[package]] name = "base64ct" version = "1.8.3" @@ -1091,7 +1491,7 @@ dependencies = [ "buzz-sdk", "chrono", "hex", - "http", + "http 1.4.2", "http-body-util", "k8s-openapi", "kube", @@ -1262,7 +1662,7 @@ name = "buzz-feature-flags" version = "0.1.0" dependencies = [ "buzz-core", - "http", + "http 1.4.2", "launchdarkly-server-sdk", "serde_json", "tempfile", @@ -1452,7 +1852,7 @@ dependencies = [ "futures-util", "hex", "hmac 0.13.0", - "http-body", + "http-body 1.0.1", "http-body-util", "infer", "jsonwebtoken", @@ -1520,11 +1920,16 @@ dependencies = [ name = "buzz-sandbox-reaper" version = "0.1.0" dependencies = [ + "aws-config", + "aws-sdk-s3", "buzz-backend-kubernetes", "chrono", + "metrics", "proptest", "serde", "serde_json", + "tokio", + "tracing", ] [[package]] @@ -1672,6 +2077,16 @@ version = "1.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" +[[package]] +name = "bytes-utils" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7dafe3a8757b027e2be6e4e5601ed563c55989fcf1546e933c66c8eb3a058d35" +dependencies = [ + "bytes", + "either", +] + [[package]] name = "bzip2" version = "0.4.4" @@ -2208,6 +2623,16 @@ version = "2.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "217698eaf96b4a3f0bc4f3662aaa55bdf913cd54d7204591faa790070c6d0853" +[[package]] +name = "crc-fast" +version = "1.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e75b2483e97a5a7da73ac68a05b629f9c53cff58d8ed1c77866079e18b00dba5" +dependencies = [ + "digest 0.10.7", + "spin 0.10.1", +] + [[package]] name = "crc32fast" version = "1.5.0" @@ -3117,7 +3542,7 @@ dependencies = [ "base64 0.22.1", "bytes", "futures", - "http", + "http 1.4.2", "launchdarkly-sdk-transport", "log", "pin-project", @@ -3662,7 +4087,7 @@ dependencies = [ "fnv", "futures-core", "futures-sink", - "http", + "http 1.4.2", "indexmap 2.14.0", "slab", "tokio", @@ -3759,7 +4184,7 @@ dependencies = [ "base64 0.22.1", "bytes", "headers-core", - "http", + "http 1.4.2", "httpdate", "mime", "sha1 0.10.6", @@ -3771,7 +4196,7 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "54b4a22553d4242c49fddb9ba998a99962b5cc6f22cb5a3482bec22522403ce4" dependencies = [ - "http", + "http 1.4.2", ] [[package]] @@ -3835,7 +4260,7 @@ dependencies = [ "anyhow", "async-trait", "bytes", - "http", + "http 1.4.2", "more-asserts", "serde", "serde_json", @@ -3866,7 +4291,7 @@ dependencies = [ "futures-util", "h2", "hickory-proto", - "http", + "http 1.4.2", "idna", "ipnet", "jni 0.22.4", @@ -3964,6 +4389,17 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "http" +version = "0.2.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "601cbb57e577e2f5ef5be8e7b83f0f63994f25aa94d673e54a92d5c516d101f1" +dependencies = [ + "bytes", + "fnv", + "itoa", +] + [[package]] name = "http" version = "1.4.2" @@ -3974,6 +4410,17 @@ dependencies = [ "itoa", ] +[[package]] +name = "http-body" +version = "0.4.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7ceab25649e9960c0311ea418d17bee82c0dcec1bd053b5f9a66e265a693bed2" +dependencies = [ + "bytes", + "http 0.2.12", + "pin-project-lite", +] + [[package]] name = "http-body" version = "1.0.1" @@ -3981,7 +4428,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1efedce1fb8e6913f23e0c92de8e62cd5b772a67e7b3946df930a62566c93184" dependencies = [ "bytes", - "http", + "http 1.4.2", ] [[package]] @@ -3992,8 +4439,8 @@ checksum = "23169fe34a5fbcdd3f3862e78fb9b6fccd5f02a6dc6f732547005d45631ce71c" dependencies = [ "bytes", "futures-core", - "http", - "http-body", + "http 1.4.2", + "http-body 1.0.1", "pin-project-lite", ] @@ -4041,8 +4488,8 @@ dependencies = [ "futures-channel", "futures-core", "h2", - "http", - "http-body", + "http 1.4.2", + "http-body 1.0.1", "httparse", "httpdate", "itoa", @@ -4061,7 +4508,7 @@ dependencies = [ "bytes", "futures-util", "headers", - "http", + "http 1.4.2", "hyper", "hyper-rustls", "hyper-util", @@ -4077,7 +4524,7 @@ version = "0.27.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "33ca68d021ef39cf6463ab54c1d0f5daf03377b70561305bb89a8f83aab66e0f" dependencies = [ - "http", + "http 1.4.2", "hyper", "hyper-util", "log", @@ -4112,8 +4559,8 @@ dependencies = [ "bytes", "futures-channel", "futures-util", - "http", - "http-body", + "http 1.4.2", + "http-body 1.0.1", "hyper", "ipnet", "libc", @@ -4292,7 +4739,7 @@ dependencies = [ "attohttpc", "bytes", "futures", - "http", + "http 1.4.2", "http-body-util", "hyper", "hyper-util", @@ -4478,7 +4925,7 @@ dependencies = [ "futures-util", "getrandom 0.4.3", "hickory-resolver", - "http", + "http 1.4.2", "ipnet", "iroh-base", "iroh-dns", @@ -4595,7 +5042,7 @@ dependencies = [ "derive_more", "getrandom 0.4.3", "hickory-resolver", - "http", + "http 1.4.2", "http-body-util", "hyper", "hyper-util", @@ -4913,8 +5360,8 @@ dependencies = [ "either", "futures", "home", - "http", - "http-body", + "http 1.4.2", + "http-body 1.0.1", "http-body-util", "hyper", "hyper-rustls", @@ -4946,7 +5393,7 @@ dependencies = [ "chrono", "derive_more", "form_urlencoded", - "http", + "http 1.4.2", "k8s-openapi", "serde", "serde-value", @@ -4968,7 +5415,7 @@ checksum = "032d66802c461d7254ed1de887be6ca98ebbe126867d9a60c754baae81c4d10d" dependencies = [ "bytes", "futures", - "http", + "http 1.4.2", "http-body-util", "hyper", "hyper-http-proxy", @@ -4995,7 +5442,7 @@ dependencies = [ "eventsource-client", "flate2", "futures", - "http", + "http 1.4.2", "launchdarkly-sdk-transport", "launchdarkly-server-sdk-evaluation", "log", @@ -5195,6 +5642,9 @@ name = "lru" version = "0.16.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f66e8d5d03f609abc3a39e6f08e4164ebf1447a732906d39eb9b99b7919ef39" +dependencies = [ + "hashbrown 0.16.1", +] [[package]] name = "lru" @@ -5529,7 +5979,7 @@ dependencies = [ "flate2", "futures-util", "hex", - "http", + "http 1.4.2", "http-body-util", "httparse", "iroh", @@ -6948,7 +7398,7 @@ checksum = "5683015d09e2df236ef005b17f6f196f0d5f6313c4fa43a7b6a53b52776e4331" dependencies = [ "async-trait", "bytes", - "http", + "http 1.4.2", "opentelemetry", "reqwest 0.13.4", ] @@ -6959,7 +7409,7 @@ version = "0.32.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9966929966d17620d7c316c643ba62631826e10021409357772d5eea84f62c35" dependencies = [ - "http", + "http 1.4.2", "opentelemetry", "opentelemetry-http", "opentelemetry-proto", @@ -7086,6 +7536,12 @@ dependencies = [ "memchr", ] +[[package]] +name = "outref" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1a80800c0488c3a21695ea981a54918fbb37abf04f4d0720c453632255e2ff0e" + [[package]] name = "p256" version = "0.13.2" @@ -7406,6 +7862,12 @@ version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" +[[package]] +name = "pin-utils" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "13bee6c73da26345c729282832b60b0363cf3dd9f4bfd81d8551b7a1c889a113" + [[package]] name = "piper" version = "0.2.5" @@ -8468,8 +8930,8 @@ dependencies = [ "bytes", "futures-core", "futures-util", - "http", - "http-body", + "http 1.4.2", + "http-body 1.0.1", "http-body-util", "hyper", "hyper-rustls", @@ -8512,8 +8974,8 @@ dependencies = [ "futures-core", "futures-util", "h2", - "http", - "http-body", + "http 1.4.2", + "http-body 1.0.1", "http-body-util", "hyper", "hyper-rustls", @@ -8553,7 +9015,7 @@ checksum = "07bc3f1384cffa4f274dad2d4ddd73aed32fed8f786d96c6be8aa4e5fd3c3b58" dependencies = [ "anyhow", "async-trait", - "http", + "http 1.4.2", "reqwest 0.13.4", "thiserror 2.0.18", "tower-service", @@ -8620,8 +9082,8 @@ dependencies = [ "bytes", "chrono", "futures", - "http", - "http-body", + "http 1.4.2", + "http-body 1.0.1", "http-body-util", "indexmap 2.14.0", "pastey", @@ -8715,7 +9177,7 @@ dependencies = [ "futures-util", "hex", "hmac 0.12.1", - "http", + "http 1.4.2", "log", "maybe-async", "md5", @@ -10011,7 +10473,7 @@ checksum = "c25ac7aff0abd1dbc474536e40416e1102c7dd9bfba0b9861c6d357f835dcfb4" dependencies = [ "bytes", "futures-util", - "http-body", + "http-body 1.0.1", "http-body-util", "pin-project-lite", ] @@ -10753,7 +11215,7 @@ dependencies = [ "futures-core", "futures-sink", "getrandom 0.4.3", - "http", + "http 1.4.2", "httparse", "rand 0.10.1", "ring", @@ -10852,8 +11314,8 @@ dependencies = [ "base64 0.22.1", "bytes", "h2", - "http", - "http-body", + "http 1.4.2", + "http-body 1.0.1", "http-body-util", "hyper", "hyper-timeout", @@ -10935,8 +11397,8 @@ dependencies = [ "bytes", "futures-core", "futures-util", - "http", - "http-body", + "http 1.4.2", + "http-body 1.0.1", "http-body-util", "http-range-header", "httpdate", @@ -11083,7 +11545,7 @@ checksum = "8628dcc84e5a09eb3d8423d6cb682965dea9133204e8fb3efee74c2a0c259442" dependencies = [ "bytes", "data-encoding", - "http", + "http 1.4.2", "httparse", "log", "rand 0.9.4", @@ -11102,7 +11564,7 @@ checksum = "6c01152af293afb9c7c2a57e4b559c5620b421f6d133261c60dd2d0cdb38e6b8" dependencies = [ "bytes", "data-encoding", - "http", + "http 1.4.2", "httparse", "log", "rand 0.9.4", @@ -11380,6 +11842,12 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" +[[package]] +name = "vsimd" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c3082ca00d5a5ef149bb8b555a72ae84c9c59f7250f013ac822ac2e49b19c64" + [[package]] name = "vtparse" version = "0.6.2" @@ -12271,7 +12739,7 @@ dependencies = [ "bytes", "crc32fast", "futures", - "http", + "http 1.4.2", "hyper", "more-asserts", "rand 0.10.1", @@ -12339,7 +12807,7 @@ dependencies = [ "bytes", "chrono", "gearhash", - "http", + "http 1.4.2", "itertools", "more-asserts", "rand 0.10.1", @@ -12406,6 +12874,12 @@ version = "0.8.28" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3ae8337f8a065cfc972643663ea4279e04e7256de865aa66fe25cec5fb912d3f" +[[package]] +name = "xmlparser" +version = "0.13.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "66fee0b777b0f5ac1c69bb06d361268faafa61cd4682ae064a171c16c433e9e4" + [[package]] name = "xmltree" version = "0.10.3" diff --git a/crates/buzz-sandbox-reaper/Cargo.toml b/crates/buzz-sandbox-reaper/Cargo.toml index af73d23bbd4..7e700db090b 100644 --- a/crates/buzz-sandbox-reaper/Cargo.toml +++ b/crates/buzz-sandbox-reaper/Cargo.toml @@ -10,11 +10,18 @@ description = "In-cluster reaper for ended and orphaned Buzz Agent Sandbox sessi name = "buzz-sandbox-reaper" path = "src/main.rs" +# Heavy, cluster-only dependencies (aws-sdk-s3, axum, metrics) live here and +# never in buzz-backend-kubernetes, whose binary ships inside the desktop app. [dependencies] buzz-backend-kubernetes = { path = "../buzz-backend-kubernetes" } +tokio = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } chrono = { workspace = true } +metrics = { workspace = true } +tracing = { workspace = true } +aws-config = { version = "1", default-features = false, features = ["rt-tokio", "default-https-client"] } +aws-sdk-s3 = { version = "1", default-features = false, features = ["rt-tokio", "default-https-client"] } [dev-dependencies] proptest = { workspace = true } diff --git a/crates/buzz-sandbox-reaper/src/evidence.rs b/crates/buzz-sandbox-reaper/src/evidence.rs new file mode 100644 index 00000000000..04f53efe8b4 --- /dev/null +++ b/crates/buzz-sandbox-reaper/src/evidence.rs @@ -0,0 +1,721 @@ +//! Checkpoint evidence: what S3 confirms about a session's checkpoint. +//! +//! A key is recorded on a tombstone only when S3 lists it under the session's +//! own prefix, `/`. That prefix comes from the +//! reaper's config and the Sandbox's generation annotation, never from the +//! Pod, so a key for another developer or generation is refused before S3 is +//! asked. Three sources name a key: +//! +//! - [`from_receipt`]: the harness's termination-message receipt. The receipt +//! alone is written by the (untrusted) Pod, so S3 must also list it. +//! - [`latest`]: the newest object in the session prefix, for a session whose +//! receipt was never observed (the reconcile engine decides when). +//! - [`inherited`]: the provider-written restore key of a recovered session. +//! +//! Any store failure is [`Evidence::Unavailable`]: the caller defers, and +//! never records `missing` because S3 could not answer. + +use std::collections::HashSet; +use std::fmt; +use std::future::Future; +use std::time::Duration; + +use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError}; +use buzz_backend_kubernetes::lifecycle::{is_session_key, receipt_key}; + +use crate::plan::Evidence; + +/// Bound a complete listing by elapsed time, not by the session's object count. +const LISTING_TIMEOUT: Duration = Duration::from_secs(25); + +/// Why the bucket could not answer: the bounded `kind` label of +/// `buzz_sandbox_reaper_s3_errors_total`, so an outage (`timeout`, `other`) +/// can be told apart from misconfiguration (`access_denied`, +/// `not_found_bucket`). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum StoreErrorKind { + Timeout, + AccessDenied, + NotFoundBucket, + Other, +} + +impl StoreErrorKind { + /// Metric label value. + pub fn as_str(self) -> &'static str { + match self { + Self::Timeout => "timeout", + Self::AccessDenied => "access_denied", + Self::NotFoundBucket => "not_found_bucket", + Self::Other => "other", + } + } +} + +/// A failed bucket request. +#[derive(Debug)] +pub struct StoreError { + pub kind: StoreErrorKind, + pub detail: String, +} + +impl StoreError { + fn other(detail: String) -> Self { + Self { + kind: StoreErrorKind::Other, + detail, + } + } +} + +impl fmt::Display for StoreError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{} ({})", self.detail, self.kind.as_str()) + } +} + +/// Read-only view of the checkpoint bucket. A trait because S3 is a real +/// external boundary: production uses [`S3Store`], tests an in-memory bucket. +pub trait CheckpointStore: Send + Sync { + /// Does exactly `key` exist? + fn contains(&self, key: &str) -> impl Future> + Send; + + /// The most recently written checkpoint object directly under `prefix` + /// (one session's `/`). + fn newest( + &self, + prefix: &str, + ) -> impl Future, StoreError>> + Send; +} + +/// S3 store using `ListObjectsV2` only (IAM: `s3:ListBucket`). +pub struct S3Store { + client: aws_sdk_s3::Client, + bucket: String, +} + +impl S3Store { + /// Build from the ambient AWS credential chain (IRSA in-cluster). + pub async fn new(bucket: &str, region: &str) -> Self { + // Bounded so one slow probe cannot stall a namespace pass (and with + // it shutdown) for longer than a few attempts. + let timeouts = aws_config::timeout::TimeoutConfig::builder() + .connect_timeout(Duration::from_secs(5)) + .operation_attempt_timeout(Duration::from_secs(10)) + .operation_timeout(Duration::from_secs(25)) + .build(); + let shared = aws_config::defaults(aws_config::BehaviorVersion::latest()) + .region(aws_config::Region::new(region.to_string())) + .timeout_config(timeouts) + .load() + .await; + Self::from_client(aws_sdk_s3::Client::new(&shared), bucket) + } + + fn from_client(client: aws_sdk_s3::Client, bucket: &str) -> Self { + Self { + client, + bucket: bucket.to_string(), + } + } +} + +fn store_error(error: SdkError) -> StoreError +where + E: ProvideErrorMetadata + std::error::Error + 'static, + R: fmt::Debug, +{ + let kind = match &error { + SdkError::TimeoutError(_) => StoreErrorKind::Timeout, + SdkError::DispatchFailure(failure) if failure.is_timeout() => StoreErrorKind::Timeout, + SdkError::ServiceError(service) => match service.err().code() { + Some("AccessDenied") => StoreErrorKind::AccessDenied, + Some("NoSuchBucket") => StoreErrorKind::NotFoundBucket, + _ => StoreErrorKind::Other, + }, + _ => StoreErrorKind::Other, + }; + StoreError { + kind, + detail: format!( + "list checkpoints: {}", + aws_sdk_s3::error::DisplayErrorContext(error) + ), + } +} + +impl CheckpointStore for S3Store { + async fn contains(&self, key: &str) -> Result { + // S3 lists keys in UTF-8 binary order, and a key sorts before every + // longer key it prefixes. So the first object listed under `key` is + // `key` itself when it exists: one request, no pagination. + let page = self + .client + .list_objects_v2() + .bucket(&self.bucket) + .prefix(key) + .max_keys(1) + .send() + .await + .map_err(store_error)?; + Ok(page.contents().first().and_then(|object| object.key()) == Some(key)) + } + + async fn newest(&self, prefix: &str) -> Result, StoreError> { + let mut newest: Option<((i64, u32), String)> = None; + let mut token: Option = None; + let deadline = tokio::time::Instant::now() + LISTING_TIMEOUT; + let mut seen_tokens = HashSet::new(); + loop { + let request = self + .client + .list_objects_v2() + .bucket(&self.bucket) + .prefix(prefix) + .set_continuation_token(token.take()) + .send(); + let page = tokio::time::timeout_at(deadline, request) + .await + .map_err(|_| StoreError { + kind: StoreErrorKind::Timeout, + detail: format!("checkpoint listing of {prefix} exceeded its time budget"), + })? + .map_err(store_error)?; + for object in page.contents() { + let (Some(key), Some(at)) = (object.key(), object.last_modified()) else { + continue; + }; + let at = (at.secs(), at.subsec_nanos()); + if is_session_key(key, prefix) + && newest + .as_ref() + .is_none_or(|(best, best_key)| (at, key) > (*best, best_key.as_str())) + { + newest = Some((at, key.to_string())); + } + } + if page.is_truncated() != Some(true) { + return Ok(newest.map(|(_, key)| key)); + } + // A truncated page without a way to continue is not an answer. + let Some(next) = page.next_continuation_token() else { + return Err(StoreError::other(format!( + "checkpoint listing of {prefix} is truncated without a continuation token" + ))); + }; + if !seen_tokens.insert(next.to_string()) { + return Err(StoreError::other(format!( + "checkpoint listing of {prefix} repeated a continuation token" + ))); + } + token = Some(next.to_string()); + } + } +} + +/// Is `generation` a provider-minted generation (32 lowercase hex)? +fn is_generation(generation: &str) -> bool { + generation.len() == 32 + && generation + .bytes() + .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)) +} + +/// The session prefix `/`, or `None` when the +/// generation is not one the provider could have written. The one generation +/// check for every prefix the reaper lists. +pub fn session_prefix(developer_prefix: &str, generation: &str) -> Option { + is_generation(generation).then(|| format!("{developer_prefix}{generation}/")) +} + +/// Evidence for a terminated container's `message` within `session_prefix` +/// (`/`). +pub async fn from_receipt( + store: &S, + message: Option<&str>, + session_prefix: &str, +) -> Evidence { + let Some(key) = message.and_then(|m| receipt_key(m, session_prefix)) else { + return Evidence::Missing; + }; + probe(store, key, Evidence::Verified).await +} + +/// Evidence for a session whose receipt was never observed: the newest object +/// directly under its own `session_prefix`, as [`Evidence::Listed`]. Nothing +/// listed is [`Evidence::Missing`]; a failed listing is +/// [`Evidence::Unavailable`], never `Missing`. +pub async fn latest(store: &S, session_prefix: &str) -> Evidence { + match store.newest(session_prefix).await { + Ok(Some(key)) if is_session_key(&key, session_prefix) => Evidence::Listed(key), + Ok(_) => Evidence::Missing, + Err(error) => unavailable(&error), + } +} + +/// Evidence for a recovered session that left no verified checkpoint of its +/// own: the checkpoint it was restored from, if that object still exists +/// directly under `/` for a valid generation. +/// The generation can be an older ancestor than `recovered-from` after +/// repeated failed recovery. `restore` comes from the provider-written +/// Sandbox annotation, never the Pod, so carrying it forward keeps the +/// original work recoverable after a failed restore or stop. +pub async fn inherited( + store: &S, + restore: &str, + developer_prefix: &str, +) -> Evidence { + let prefix = restore + .strip_prefix(developer_prefix) + .and_then(|rest| rest.split_once('/')) + .and_then(|(generation, _)| session_prefix(developer_prefix, generation)); + match prefix { + Some(prefix) if is_session_key(restore, &prefix) => { + probe(store, restore.to_string(), Evidence::Inherited).await + } + _ => Evidence::Missing, + } +} + +async fn probe( + store: &S, + key: String, + found: fn(String) -> Evidence, +) -> Evidence { + match store.contains(&key).await { + Ok(true) => found(key), + Ok(false) => Evidence::Missing, + Err(error) => unavailable(&error), + } +} + +fn unavailable(error: &StoreError) -> Evidence { + tracing::warn!(%error, "checkpoint store failed; deferring decision"); + metrics::counter!("buzz_sandbox_reaper_s3_errors_total", "kind" => error.kind.as_str()) + .increment(1); + Evidence::Unavailable +} + +#[cfg(test)] +pub mod tests { + use super::*; + use std::sync::{Arc, Mutex}; + + /// In-memory bucket recording every probed key or listed prefix. + /// Insertion order stands in for `LastModified`. + #[derive(Default)] + pub struct FakeStore { + pub keys: Vec, + pub fail: bool, + pub probes: Mutex>, + } + + impl FakeStore { + fn outage(&self) -> Result<(), StoreError> { + if self.fail { + return Err(StoreError::other("synthetic outage".into())); + } + Ok(()) + } + } + + impl CheckpointStore for FakeStore { + async fn contains(&self, key: &str) -> Result { + self.probes.lock().unwrap().push(key.to_string()); + self.outage()?; + Ok(self.keys.iter().any(|k| k == key)) + } + + async fn newest(&self, prefix: &str) -> Result, StoreError> { + self.probes.lock().unwrap().push(prefix.to_string()); + self.outage()?; + Ok(self + .keys + .iter() + .rev() + .find(|k| is_session_key(k, prefix)) + .cloned()) + } + } + + const PREFIX: &str = "dev/0123456789abcdef0123456789abcdef/"; + + fn receipt(key: &str) -> String { + serde_json::json!({"version": 1, "checkpoint": key}).to_string() + } + + #[tokio::test] + async fn receipt_and_listing_must_agree() { + let key = format!("{PREFIX}6f9619ff-8b86-d011-b42d-00cf4fc964ff.tar.gz"); + let store = FakeStore { + keys: vec![key.clone()], + ..Default::default() + }; + assert_eq!( + from_receipt(&store, Some(&receipt(&key)), PREFIX).await, + Evidence::Verified(key.clone()) + ); + // Receipt names an object S3 does not have. + let other = format!("{PREFIX}00000000-0000-0000-0000-000000000000.tar.gz"); + assert_eq!( + from_receipt(&store, Some(&receipt(&other)), PREFIX).await, + Evidence::Missing + ); + // No receipt, even though S3 has a key. + assert_eq!(from_receipt(&store, None, PREFIX).await, Evidence::Missing); + assert_eq!( + from_receipt(&store, Some("garbage"), PREFIX).await, + Evidence::Missing + ); + } + + /// A compromised Pod pointing at another generation's (existing) object + /// is refused without even asking S3. + #[tokio::test] + async fn foreign_generation_receipts_are_refused_before_probing() { + let foreign = "dev/ffffffffffffffffffffffffffffffff/x.tar.gz".to_string(); + let store = FakeStore { + keys: vec![foreign.clone()], + ..Default::default() + }; + assert_eq!( + from_receipt(&store, Some(&receipt(&foreign)), PREFIX).await, + Evidence::Missing + ); + assert!(store.probes.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn an_s3_outage_defers_rather_than_records_missing() { + let key = format!("{PREFIX}x.tar.gz"); + let store = FakeStore { + keys: vec![key.clone()], + fail: true, + ..Default::default() + }; + assert_eq!( + from_receipt(&store, Some(&receipt(&key)), PREFIX).await, + Evidence::Unavailable + ); + assert_eq!(latest(&store, PREFIX).await, Evidence::Unavailable); + assert_eq!( + store.probes.lock().unwrap().as_slice(), + [key.as_str(), PREFIX] + ); + } + + #[tokio::test] + async fn the_newest_object_in_the_session_prefix_is_listed() { + let older = format!("{PREFIX}00000000-0000-0000-0000-000000000001.tar.gz"); + let newer = format!("{PREFIX}00000000-0000-0000-0000-000000000002.tar.gz"); + let store = FakeStore { + keys: vec![ + older, + newer.clone(), + "dev/ffffffffffffffffffffffffffffffff/x.tar.gz".into(), + format!("{PREFIX}nested/x.tar.gz"), + ], + ..Default::default() + }; + assert_eq!(latest(&store, PREFIX).await, Evidence::Listed(newer)); + assert_eq!( + latest(&FakeStore::default(), PREFIX).await, + Evidence::Missing + ); + } + + #[test] + fn session_prefixes_require_a_provider_generation() { + let generation = "0123456789abcdef0123456789abcdef"; + assert_eq!(session_prefix("dev/", generation).as_deref(), Some(PREFIX)); + for bad in [ + "", + "0123456789ABCDEF0123456789ABCDEF", + "0123456789abcdef0123456789abcde", + "0123456789abcdef0123456789abcdef0", + "0123456789abcdef0123456789abcdeg", + "../3456789abcdef0123456789abcdef0", + "0123456789abcdef/123456789abcdef", + ] { + assert_eq!(session_prefix("dev/", bad), None, "{bad:?}"); + } + } + + #[tokio::test] + async fn a_restore_key_is_inherited_only_within_its_developer_prefix() { + let key = format!("{PREFIX}6f9619ff-8b86-d011-b42d-00cf4fc964ff.tar.gz"); + let store = FakeStore { + keys: vec![key.clone()], + ..Default::default() + }; + assert_eq!( + inherited(&store, &key, "dev/").await, + Evidence::Inherited(key.clone()) + ); + // Expired or deleted since the recovery. + let gone = format!("{PREFIX}00000000-0000-0000-0000-000000000000.tar.gz"); + assert_eq!(inherited(&store, &gone, "dev/").await, Evidence::Missing); + // Another developer's prefix, an invalid generation, or a nested key + // is refused before S3 is asked. + let probes = store.probes.lock().unwrap().len(); + assert_eq!( + inherited(&store, &key, "another-developer/").await, + Evidence::Missing + ); + for bad in [ + "dev/NOT-A-GENERATION/x.tar.gz", + &format!("{PREFIX}a/x.tar.gz"), + ] { + let store = FakeStore { + keys: vec![bad.to_string()], + ..Default::default() + }; + assert_eq!(inherited(&store, bad, "dev/").await, Evidence::Missing); + assert!(store.probes.lock().unwrap().is_empty(), "{bad}"); + } + assert_eq!(store.probes.lock().unwrap().len(), probes); + let outage = FakeStore { + fail: true, + ..Default::default() + }; + assert_eq!( + inherited(&outage, &key, "dev/").await, + Evidence::Unavailable + ); + } + + // ── The real S3 adapter against a local endpoint ────────────────────── + + /// One canned S3 reply; `None` never answers (holds the connection). + type Reply = Option<(u16, String)>; + + /// Serve `reply(query)` for every request on a local port, recording + /// each decoded query. Returns the endpoint URL. + async fn s3_endpoint( + reply: impl Fn(&str) -> Reply + Send + Sync + 'static, + ) -> (String, Arc>>) { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + let seen = Arc::new(Mutex::new(Vec::new())); + let (reply, log) = (Arc::new(reply), Arc::clone(&seen)); + tokio::spawn(async move { + while let Ok((mut socket, _)) = listener.accept().await { + let (reply, log) = (Arc::clone(&reply), Arc::clone(&log)); + tokio::spawn(async move { + let mut request = Vec::new(); + let mut buf = [0u8; 4096]; + while !request.windows(4).any(|w| w == b"\r\n\r\n") { + match socket.read(&mut buf).await { + Ok(0) | Err(_) => return, + Ok(n) => request.extend_from_slice(&buf[..n]), + } + } + let head = String::from_utf8_lossy(&request); + let target = head.split_whitespace().nth(1).unwrap_or_default(); + let query = decode(target.split_once('?').map_or("", |(_, q)| q)); + log.lock().unwrap().push(query.clone()); + let Some((status, body)) = reply(&query) else { + std::future::pending::<()>().await; + return; + }; + let response = format!( + "HTTP/1.1 {status} S3\r\nContent-Type: application/xml\r\n\ + Content-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + ); + let _ = socket.write_all(response.as_bytes()).await; + let _ = socket.shutdown().await; + }); + } + }); + (url, seen) + } + + fn decode(query: &str) -> String { + let mut out = Vec::new(); + let mut bytes = query.bytes(); + while let Some(b) = bytes.next() { + if b == b'%' { + let hex: String = bytes.by_ref().take(2).map(char::from).collect(); + out.push(u8::from_str_radix(&hex, 16).unwrap()); + } else { + out.push(b); + } + } + String::from_utf8(out).unwrap() + } + + fn page(keys: &[(&str, &str)], next: Option<&str>, truncated: bool) -> Reply { + let contents: String = keys + .iter() + .map(|(key, at)| { + format!("{key}{at}1") + }) + .collect(); + let next = next + .map(|t| format!("{t}")) + .unwrap_or_default(); + Some(( + 200, + format!( + "\ + \ + bucket{}1000\ + {truncated}{next}{contents}", + keys.len() + ), + )) + } + + fn s3_error(status: u16, code: &str) -> Reply { + Some(( + status, + format!("{code}m"), + )) + } + + fn s3(url: &str, timeout: Duration) -> S3Store { + let config = aws_sdk_s3::Config::builder() + .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest()) + .region(aws_sdk_s3::config::Region::new("us-west-2")) + .credentials_provider(aws_sdk_s3::config::Credentials::new( + "AKID", "secret", None, None, "test", + )) + .endpoint_url(url) + .force_path_style(true) + .retry_config(aws_sdk_s3::config::retry::RetryConfig::disabled()) + .timeout_config( + aws_sdk_s3::config::timeout::TimeoutConfig::builder() + .operation_timeout(timeout) + .build(), + ) + .build(); + S3Store::from_client(aws_sdk_s3::Client::from_conf(config), "bucket") + } + + const AT: &str = "2026-10-20T00:00:00.000Z"; + const LATER: &str = "2026-10-20T00:00:01.000Z"; + + /// `contains` asks for the key itself as the prefix and accepts only an + /// exact match, never a longer key that merely starts with it. + #[tokio::test] + async fn s3_contains_lists_the_key_as_its_own_prefix() { + let key = format!("{PREFIX}a.tar.gz"); + let partial = format!("{PREFIX}b.tar.gz"); + let (stored, extended) = (key.clone(), format!("{partial}.tmp")); + let (url, seen) = s3_endpoint(move |query| { + if query.contains(&format!("prefix={stored}")) { + page(&[(&stored, AT)], None, false) + } else { + page(&[(&extended, AT)], None, false) + } + }) + .await; + let store = s3(&url, Duration::from_secs(5)); + assert!(store.contains(&key).await.unwrap()); + assert!(!store.contains(&partial).await.unwrap()); + let seen = seen.lock().unwrap(); + assert!(seen.iter().all(|q| q.contains("max-keys=1")), "{seen:?}"); + } + + #[tokio::test] + async fn s3_newest_follows_continuation_to_a_later_page() { + let first = format!("{PREFIX}first.tar.gz"); + let second = format!("{PREFIX}second.tar.gz"); + let nested = format!("{PREFIX}nested/x.tar.gz"); + let (a, b) = (first.clone(), second.clone()); + let (url, seen) = s3_endpoint(move |query| { + if query.contains("continuation-token=page-2") { + page( + &[(&b, LATER), (&nested, "2027-01-01T00:00:00.000Z")], + None, + false, + ) + } else { + page(&[(&a, AT)], Some("page-2"), true) + } + }) + .await; + let store = s3(&url, Duration::from_secs(5)); + assert_eq!(store.newest(PREFIX).await.unwrap(), Some(second)); + assert_eq!(seen.lock().unwrap().len(), 2); + let (url, _) = s3_endpoint(|_| page(&[], None, false)).await; + assert_eq!( + s3(&url, Duration::from_secs(5)) + .newest(PREFIX) + .await + .unwrap(), + None + ); + } + + #[tokio::test] + async fn s3_newest_can_select_a_checkpoint_after_ten_pages() { + let key = format!("{PREFIX}last.tar.gz"); + let expected = key.clone(); + let (url, seen) = s3_endpoint(move |query| { + let index = query + .split('&') + .find_map(|part| part.strip_prefix("continuation-token=")) + .map_or(0, |value| value.parse::().unwrap()); + if index == 11 { + page(&[(&key, LATER)], None, false) + } else { + page(&[], Some(&(index + 1).to_string()), true) + } + }) + .await; + assert_eq!( + s3(&url, Duration::from_secs(5)) + .newest(PREFIX) + .await + .unwrap(), + Some(expected) + ); + assert_eq!(seen.lock().unwrap().len(), 12); + } + + #[tokio::test] + async fn s3_newest_refuses_partial_listings() { + // A repeated token is rejected instead of walking forever. + let (url, seen) = s3_endpoint(|_| page(&[], Some("again"), true)).await; + let error = s3(&url, Duration::from_secs(5)) + .newest(PREFIX) + .await + .unwrap_err(); + assert_eq!(error.kind, StoreErrorKind::Other); + assert_eq!(seen.lock().unwrap().len(), 2); + // Truncated with no way to continue is a failure, not "nothing found". + let (url, _) = s3_endpoint(|_| page(&[], None, true)).await; + let error = s3(&url, Duration::from_secs(5)) + .newest(PREFIX) + .await + .unwrap_err(); + assert!(error.detail.contains("continuation token"), "{error}"); + } + + #[tokio::test] + async fn s3_failures_are_classified_for_operators() { + let cases = [ + (s3_error(403, "AccessDenied"), StoreErrorKind::AccessDenied), + ( + s3_error(404, "NoSuchBucket"), + StoreErrorKind::NotFoundBucket, + ), + (s3_error(500, "InternalError"), StoreErrorKind::Other), + (s3_error(503, "SlowDown"), StoreErrorKind::Other), + (None, StoreErrorKind::Timeout), + ]; + for (reply, kind) in cases { + let (url, _) = s3_endpoint(move |_| reply.clone()).await; + let store = s3(&url, Duration::from_millis(300)); + let error = store + .contains(&format!("{PREFIX}x.tar.gz")) + .await + .unwrap_err(); + assert_eq!(error.kind, kind, "{error}"); + assert_eq!(store.newest(PREFIX).await.unwrap_err().kind, kind); + } + } +} diff --git a/crates/buzz-sandbox-reaper/src/main.rs b/crates/buzz-sandbox-reaper/src/main.rs index 6038455d579..18d1c6a407b 100644 --- a/crates/buzz-sandbox-reaper/src/main.rs +++ b/crates/buzz-sandbox-reaper/src/main.rs @@ -1,14 +1,12 @@ //! `buzz-sandbox-reaper`: the in-cluster sandbox reaper for Buzz Agent //! Sandbox sessions (`docs/remote-agents.md` §Sandbox lifecycle). -#![cfg_attr( - not(test), - expect( - dead_code, - reason = "stack glue: the modules are exercised by their tests until the reconcile subcommand wires them up" - ) +#![expect( + dead_code, + reason = "stack glue: the modules are exercised by their tests until the reconcile subcommand wires them up (S3Store::new has no test caller)" )] mod config; +mod evidence; mod plan; fn main() {} diff --git a/crates/buzz-sandbox-reaper/src/plan.rs b/crates/buzz-sandbox-reaper/src/plan.rs index 2355d552751..f5ea43174bf 100644 --- a/crates/buzz-sandbox-reaper/src/plan.rs +++ b/crates/buzz-sandbox-reaper/src/plan.rs @@ -21,6 +21,10 @@ pub enum Evidence { Unconfigured, /// No receipt, or the receipt and S3 disagree. Missing, + /// No receipt was observed, but S3 lists this newest object in the + /// session's own prefix. Not attested by the harness, so a Completed + /// session still holds before recording it. + Listed(String), /// Missing, but the session was recovered from this (still present) key, /// which the tombstone carries forward. Inherited(String), @@ -32,7 +36,7 @@ impl Evidence { /// Tombstone `checkpoint` annotation value; `None` while undecidable. fn annotation(&self) -> Option { match self { - Self::Verified(key) | Self::Inherited(key) => Some(key.clone()), + Self::Verified(key) | Self::Listed(key) | Self::Inherited(key) => Some(key.clone()), Self::Unconfigured => Some(CHECKPOINT_UNCONFIGURED.into()), Self::Missing => Some(CHECKPOINT_MISSING.into()), Self::Unavailable => None, @@ -101,7 +105,7 @@ pub fn plan( State::Replaced => tombstone(EndedReason::Replaced, evidence), State::Lost { confirmed: true } => tombstone(EndedReason::Lost, evidence), State::Completed { .. } => match evidence { - Evidence::Missing | Evidence::Inherited(_) => { + Evidence::Missing | Evidence::Listed(_) | Evidence::Inherited(_) => { let until = hold_since + Duration::seconds(MISSING_CHECKPOINT_HOLD_SECS); if now >= until { tombstone(EndedReason::Completed, evidence) @@ -233,34 +237,39 @@ mod tests { } } - /// A recovered session that saved nothing of its own still holds, then - /// carries its restore checkpoint forward instead of `missing`. + /// A recovered session that saved nothing of its own, or a session whose + /// key only a listing found, still holds, then records that key instead + /// of `missing`. A session whose Pod is gone records it at once. #[test] - fn completed_with_an_inherited_checkpoint_holds_then_carries_it_forward() { + fn completed_with_an_unattested_checkpoint_holds_then_carries_it_forward() { let completed = State::Completed { finished_at: Some(t(0)), message: None, }; - let inherited = Evidence::Inherited("dev/g/x.tar.gz".into()); let until = t(MISSING_CHECKPOINT_HOLD_SECS); - assert_eq!( - plan(&completed, &inherited, t(1), t(0)), - Action::Hold { until } - ); - assert_eq!( - plan(&completed, &inherited, until, t(0)), - Action::Tombstone { - reason: EndedReason::Completed, - checkpoint: "dev/g/x.tar.gz".into() - } - ); - assert_eq!( - plan(&State::Lost { confirmed: true }, &inherited, t(0), t(0)), - Action::Tombstone { - reason: EndedReason::Lost, - checkpoint: "dev/g/x.tar.gz".into() - } - ); + for evidence in [ + Evidence::Inherited("dev/g/x.tar.gz".into()), + Evidence::Listed("dev/g/x.tar.gz".into()), + ] { + assert_eq!( + plan(&completed, &evidence, t(1), t(0)), + Action::Hold { until } + ); + assert_eq!( + plan(&completed, &evidence, until, t(0)), + Action::Tombstone { + reason: EndedReason::Completed, + checkpoint: "dev/g/x.tar.gz".into() + } + ); + assert_eq!( + plan(&State::Lost { confirmed: true }, &evidence, t(0), t(0)), + Action::Tombstone { + reason: EndedReason::Lost, + checkpoint: "dev/g/x.tar.gz".into() + } + ); + } } #[test] @@ -317,6 +326,7 @@ mod tests { prop_oneof![ "[a-z]{1,8}".prop_map(Evidence::Verified), "[a-z]{1,8}".prop_map(Evidence::Inherited), + "[a-z]{1,8}".prop_map(Evidence::Listed), Just(Evidence::Unconfigured), Just(Evidence::Missing), Just(Evidence::Unavailable), diff --git a/docs/remote-agents.md b/docs/remote-agents.md index f75bb34885a..8b79c48c386 100644 --- a/docs/remote-agents.md +++ b/docs/remote-agents.md @@ -1554,6 +1554,30 @@ configuration lists for the namespace. It records a Warning Event for a managed Sandbox that it leaves alone (foreign owner, foreign Pod, inconsistent lifecycle). +**Checkpoint evidence.** After a successful checkpoint on stop, the harness +writes `{"version":1,"checkpoint":""}` to `/dev/termination-log` (the +Pod keeps `terminationMessagePolicy: File`). The reaper reads the bound +Pod's termination message. A Pod watch keeps the message after a stop or a +replacement deletes the Pod. The reaper accepts a key only if it is under +`/` (prefix from the reaper configuration, +generation from the Sandbox, 32 lowercase hex characters) and an S3 +`ListObjectsV2` (`s3:ListBucket` only) finds exactly that key. Otherwise it +records `missing`. A compromised Pod therefore cannot point a tombstone at +the checkpoint of a different session. If S3 cannot be reached, the reaper +waits for a later pass. It never records `missing` because S3 failed. +`buzz_sandbox_reaper_s3_errors_total{kind}` counts these failures by kind: +`timeout` or `other` for an outage, `access_denied` or `not_found_bucket` for +a misconfigured role or bucket. + +A recovered session that has no verified checkpoint of its own (for example, +its restore failed) keeps the checkpoint it was recovered from. The reaper +takes the key from the provider-written `restore-checkpoint` annotation, +requires a valid generation and an object directly under the configured +developer prefix, and confirms it with S3. The key can belong to an older +ancestor than `recovered-from` after repeated failed recovery. The 24-hour +hold still applies. The tombstone then records that key instead of `missing`, +so the developer can retry recovery. + **Recovery.** An ordinary deploy to an ended session is refused with `session_ended` (§Sandbox error codes). The desktop then offers two actions in the thread. Each sends `provider_config.sandbox.recovery` once, for that