diff --git a/Cargo.lock b/Cargo.lock index 26251d6aa..10b63b541 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -107,6 +107,12 @@ dependencies = [ "memchr", ] +[[package]] +name = "aliasable" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "250f629c0161ad8107cf89319e990051fae62832fd343083bea452d93e2205fd" + [[package]] name = "alien-ai-gateway" version = "3.3.18" @@ -1251,7 +1257,7 @@ dependencies = [ "lambda_http", "lambda_runtime", "libc", - "lru", + "lru 0.16.4", "once_cell", "opentelemetry", "opentelemetry-appender-tracing", @@ -1690,6 +1696,15 @@ dependencies = [ "num-traits", ] +[[package]] +name = "atomic" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89cbf775b137e9b968e67227ef7f775587cde3fd31b0d8599dbd0f598a48340" +dependencies = [ + "bytemuck", +] + [[package]] name = "atomic-waker" version = "1.1.2" @@ -2786,7 +2801,7 @@ dependencies = [ "num-traits", "serde", "wasm-bindgen", - "windows-link", + "windows-link 0.2.1", ] [[package]] @@ -3276,6 +3291,16 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "crossbeam-skiplist" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df29de440c58ca2cc6e587ec3d22347551a32435fbde9d2bff64e78a9ffa151b" +dependencies = [ + "crossbeam-epoch", + "crossbeam-utils", +] + [[package]] name = "crossbeam-utils" version = "0.8.22" @@ -3871,6 +3896,19 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" +[[package]] +name = "duration-str" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f88959de2d447fd3eddcf1909d1f19fe084e27a056a6904203dc5d8b9e771c1e" +dependencies = [ + "rust_decimal", + "serde", + "thiserror 2.0.20", + "time", + "winnow 0.6.26", +] + [[package]] name = "dyn-clone" version = "1.0.20" @@ -4093,6 +4131,18 @@ dependencies = [ "tokio", ] +[[package]] +name = "fail-parallel" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6f4a147ba57fcd64323c54c8f69fd4e045456a99574163010ccf5eea3168aaa" +dependencies = [ + "log", + "once_cell", + "rand 0.9.5", + "tokio", +] + [[package]] name = "fallible-iterator" version = "0.2.0" @@ -4138,6 +4188,22 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "64cd1e32ddd350061ae6edb1b082d7c54915b5c672c389143b9a63403a109f24" +[[package]] +name = "figment" +version = "0.10.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8cb01cd46b0cf372153850f4c6c272d9cbea2da513e07538405148f95bd789f3" +dependencies = [ + "atomic", + "pear", + "serde", + "serde_json", + "serde_yaml", + "toml 0.8.23", + "uncased", + "version_check", +] + [[package]] name = "filetime" version = "0.2.29" @@ -4166,6 +4232,16 @@ version = "0.5.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d674e81391d1e1ab681a28d99df07927c6d4aa5b027d7da16ba32d1d21ecd99" +[[package]] +name = "flatbuffers" +version = "25.12.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "35f6839d7b3b98adde531effaf34f0c2badc6f4735d26fe74709d8e513a96ef3" +dependencies = [ + "bitflags 2.13.1", + "rustc_version", +] + [[package]] name = "flate2" version = "1.1.9" @@ -4438,8 +4514,8 @@ dependencies = [ "libc", "log", "rustversion", - "windows-link", - "windows-result", + "windows-link 0.2.1", + "windows-result 0.4.1", ] [[package]] @@ -4644,6 +4720,11 @@ name = "hashbrown" version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" +dependencies = [ + "allocator-api2 0.2.21", + "equivalent", + "foldhash 0.2.0", +] [[package]] name = "hashlink" @@ -4823,7 +4904,7 @@ checksum = "617aaa3557aef3810a6369d0a99fac8a080891b68bd9f9812a1eeda0c0730cbd" dependencies = [ "cfg-if", "libc", - "windows-link", + "windows-link 0.2.1", ] [[package]] @@ -5165,7 +5246,7 @@ dependencies = [ "js-sys", "log", "wasm-bindgen", - "windows-core", + "windows-core 0.62.2", ] [[package]] @@ -5372,6 +5453,12 @@ dependencies = [ "syn 3.0.3", ] +[[package]] +name = "inlinable_string" +version = "0.1.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c8fae54786f62fb2918dcfae3d568594e50eb9b5c25bf04371af6fe7516452fb" + [[package]] name = "inotify" version = "0.10.2" @@ -5531,7 +5618,7 @@ dependencies = [ "simd_cesu8", "thiserror 2.0.20", "walkdir", - "windows-link", + "windows-link 0.2.1", ] [[package]] @@ -5895,7 +5982,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d7c4b02199fee7c5d21a5ae7d8cfa79a6ef5bb2fc834d6e9058e89c825efdc55" dependencies = [ "cfg-if", - "windows-link", + "windows-link 0.2.1", ] [[package]] @@ -5905,7 +5992,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "754ca22de805bb5744484a5b151a9e1a8e837d5dc232c2d7d8c2e3492edc8b60" dependencies = [ "cfg-if", - "windows-link", + "windows-link 0.2.1", ] [[package]] @@ -6007,6 +6094,15 @@ dependencies = [ "hashbrown 0.16.1", ] +[[package]] +name = "lru" +version = "0.18.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5d2f2f9b4ba7e6b24d95e7e899329d35be83bcded72c8540cdd5368932d1d90a" +dependencies = [ + "hashbrown 0.17.1", +] + [[package]] name = "lru-slab" version = "0.1.2" @@ -6228,6 +6324,26 @@ dependencies = [ "num-traits", ] +[[package]] +name = "moka" +version = "0.12.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4293f18e7567a1caf3c584855554377025c65e0aa445344d04171f5ad63d19b9" +dependencies = [ + "async-lock", + "crossbeam-channel", + "crossbeam-epoch", + "crossbeam-utils", + "equivalent", + "event-listener 5.4.2", + "futures-util", + "parking_lot", + "portable-atomic", + "smallvec", + "tagptr", + "uuid", +] + [[package]] name = "multimap" version = "0.10.1" @@ -6436,6 +6552,15 @@ dependencies = [ "instant", ] +[[package]] +name = "ntapi" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3b335231dfd352ffb0f8017f3b6027a4917f7df785ea2143d8af2adc66980ae" +dependencies = [ + "winapi", +] + [[package]] name = "nu-ansi-term" version = "0.50.3" @@ -6535,6 +6660,16 @@ dependencies = [ "bitflags 2.13.1", ] +[[package]] +name = "objc2-io-kit" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "33fafba39597d6dc1fb709123dfa8289d39406734be322956a69f0931c73bb15" +dependencies = [ + "libc", + "objc2-core-foundation", +] + [[package]] name = "objc2-system-configuration" version = "0.3.2" @@ -6897,6 +7032,30 @@ dependencies = [ "hashbrown 0.14.5", ] +[[package]] +name = "ouroboros" +version = "0.18.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e0f050db9c44b97a94723127e6be766ac5c340c48f2c4bb3ffa11713744be59" +dependencies = [ + "aliasable", + "ouroboros_macro", + "static_assertions", +] + +[[package]] +name = "ouroboros_macro" +version = "0.18.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3c7028bdd3d43083f6d8d4d5187680d0d3560d54df4cc9d752005268b41e64d0" +dependencies = [ + "heck 0.4.1", + "proc-macro2", + "proc-macro2-diagnostics", + "quote", + "syn 2.0.119", +] + [[package]] name = "outref" version = "0.5.2" @@ -6988,7 +7147,7 @@ dependencies = [ "libc", "redox_syscall", "smallvec", - "windows-link", + "windows-link 0.2.1", ] [[package]] @@ -7003,6 +7162,29 @@ version = "0.2.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "df94ce210e5bc13cb6651479fa48d14f601d9858cfe0467f43ae157023b938d3" +[[package]] +name = "pear" +version = "0.2.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bdeeaa00ce488657faba8ebf44ab9361f9365a97bd39ffb8a60663f57ff4b467" +dependencies = [ + "inlinable_string", + "pear_codegen", + "yansi", +] + +[[package]] +name = "pear_codegen" +version = "0.2.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4bab5b985dc082b345f812b7df84e1bef27e7207b39e448439ba8bd69c93f147" +dependencies = [ + "proc-macro2", + "proc-macro2-diagnostics", + "quote", + "syn 2.0.119", +] + [[package]] name = "pem" version = "3.0.6" @@ -7314,6 +7496,12 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4c2749dcd0984ec1be3c01001bb1d83623a58c3c0049a99b9afec61464fa98e7" +[[package]] +name = "portable-atomic" +version = "1.15.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05c8b63e8d9609db387f0324918f81d68fe27748f084ef092fb35954d0539a85" + [[package]] name = "portpicker" version = "0.1.1" @@ -7566,6 +7754,19 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "proc-macro2-diagnostics" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "af066a9c399a26e020ada66a034357a868728e72cd426f3adcd35f80d88d88c8" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", + "version_check", + "yansi", +] + [[package]] name = "progenitor" version = "0.11.2" @@ -7673,7 +7874,7 @@ version = "0.13.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf" dependencies = [ - "heck 0.4.1", + "heck 0.5.0", "itertools 0.14.0", "log", "multimap", @@ -7979,6 +8180,15 @@ dependencies = [ "rand_core 0.10.1", ] +[[package]] +name = "rand_xoshiro" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f703f4665700daf5512dcca5f43afa6af89f09db47fb56be587f80636bda2d41" +dependencies = [ + "rand_core 0.9.5", +] + [[package]] name = "rapidhash" version = "4.5.1" @@ -8079,7 +8289,7 @@ dependencies = [ "cfg-if", "libc", "rustix 1.1.4", - "windows", + "windows 0.62.2", ] [[package]] @@ -8428,6 +8638,16 @@ dependencies = [ "ordered-multimap", ] +[[package]] +name = "rust_decimal" +version = "1.42.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "be2a24f50780bc85f09cc6ac299bdf1424302742d77221106859c9d8b102126a" +dependencies = [ + "arrayvec", + "num-traits", +] + [[package]] name = "rustc-hash" version = "1.1.0" @@ -9309,6 +9529,103 @@ version = "0.4.12" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" +[[package]] +name = "slatedb" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "286ec5657d17f3ba70c060becd57a5550ad43973b8480811ad27d3fa4d620a49" +dependencies = [ + "async-channel 2.5.0", + "async-trait", + "atomic", + "backon", + "bitflags 2.13.1", + "bytes", + "chrono", + "crc32fast", + "crossbeam-skiplist", + "dotenvy", + "duration-str", + "fail-parallel", + "figment", + "flatbuffers", + "futures", + "log", + "lru 0.18.2", + "moka", + "object_store", + "once_cell", + "ouroboros", + "parking_lot", + "rand 0.9.5", + "rand_xoshiro", + "serde", + "serde_json", + "siphasher", + "slatedb-common", + "slatedb-txn-obj", + "sysinfo", + "thiserror 1.0.69", + "thread_local", + "tokio", + "tokio-util", + "tracing", + "ulid", + "url", + "uuid", + "walkdir", +] + +[[package]] +name = "slatedb-common" +version = "0.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "396a08a34b93becfc72e7d7286f620ac887b48ad37bdaa4768b024df7c0ea7f5" +dependencies = [ + "chrono", + "log", + "serde", + "tokio", +] + +[[package]] +name = "slatedb-trace-store" +version = "0.1.0" +dependencies = [ + "alien-error", + "alien-sdk", + "axum 0.8.9", + "bytes", + "chrono", + "hex", + "object_store", + "serde", + "serde_json", + "sha2 0.10.9", + "slatedb", + "tempfile", + "tokio", + "tower-http 0.6.11", + "tracing", + "tracing-subscriber", +] + +[[package]] +name = "slatedb-txn-obj" +version = "0.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "37172cf2255b0f0c56f0d2e98dbc865b2f6dd56333ed297147e9bba5d9ecf97a" +dependencies = [ + "async-trait", + "bytes", + "chrono", + "futures", + "log", + "object_store", + "slatedb-common", + "thiserror 1.0.69", +] + [[package]] name = "slotmap" version = "1.1.1" @@ -9363,7 +9680,7 @@ version = "0.8.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c1c97747dbf44bb1ca44a561ece23508e99cb592e862f22222dcf42f51d1e451" dependencies = [ - "heck 0.4.1", + "heck 0.5.0", "proc-macro2", "quote", "syn 2.0.119", @@ -9759,6 +10076,20 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "sysinfo" +version = "0.35.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3c3ffa3e4ff2b324a57f7aeb3c349656c7b127c3c189520251a648102a92496e" +dependencies = [ + "libc", + "memchr", + "ntapi", + "objc2-core-foundation", + "objc2-io-kit", + "windows 0.61.3", +] + [[package]] name = "system-configuration" version = "0.7.0" @@ -9780,6 +10111,12 @@ dependencies = [ "libc", ] +[[package]] +name = "tagptr" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b2093cf4c8eb1e67749a6762251bc9cd836b6fc171623bd0a9d324d37af2417" + [[package]] name = "tap" version = "1.0.1" @@ -9830,7 +10167,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.3.4", + "getrandom 0.4.3", "once_cell", "rustix 1.1.4", "windows-sys 0.61.2", @@ -10163,6 +10500,7 @@ dependencies = [ "futures-io", "futures-sink", "futures-util", + "hashbrown 0.15.5", "libc", "pin-project-lite", "tokio", @@ -10824,6 +11162,17 @@ version = "0.1.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2896d95c02a80c6d6a5d6e953d479f5ddf2dfdb6a244441010e373ac0fb88971" +[[package]] +name = "ulid" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "470dbf6591da1b39d43c14523b2b469c86879a53e8b758c8e090a470fe7b1fbe" +dependencies = [ + "rand 0.9.5", + "serde", + "web-time", +] + [[package]] name = "uncased" version = "0.9.10" @@ -11342,7 +11691,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.48.0", + "windows-sys 0.61.2", ] [[package]] @@ -11351,16 +11700,38 @@ version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" +[[package]] +name = "windows" +version = "0.61.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9babd3a767a4c1aef6900409f85f5d53ce2544ccdfaa86dad48c91782c6d6893" +dependencies = [ + "windows-collections 0.2.0", + "windows-core 0.61.2", + "windows-future 0.2.1", + "windows-link 0.1.3", + "windows-numerics 0.2.0", +] + [[package]] name = "windows" version = "0.62.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "527fadee13e0c05939a6a05d5bd6eec6cd2e3dbd648b9f8e447c6518133d8580" dependencies = [ - "windows-collections", - "windows-core", - "windows-future", - "windows-numerics", + "windows-collections 0.3.2", + "windows-core 0.62.2", + "windows-future 0.3.2", + "windows-numerics 0.3.1", +] + +[[package]] +name = "windows-collections" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3beeceb5e5cfd9eb1d76b381630e82c4241ccd0d27f1a39ed41b2760b255c5e8" +dependencies = [ + "windows-core 0.61.2", ] [[package]] @@ -11369,7 +11740,20 @@ version = "0.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "23b2d95af1a8a14a3c7367e1ed4fc9c20e0a26e79551b1454d72583c97cc6610" dependencies = [ - "windows-core", + "windows-core 0.62.2", +] + +[[package]] +name = "windows-core" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0fdd3ddb90610c7638aa2b3a3ab2904fb9e5cdbecc643ddb3647212781c4ae3" +dependencies = [ + "windows-implement", + "windows-interface", + "windows-link 0.1.3", + "windows-result 0.3.4", + "windows-strings 0.4.2", ] [[package]] @@ -11380,9 +11764,20 @@ checksum = "b8e83a14d34d0623b51dce9581199302a221863196a1dde71a7663a4c2be9deb" dependencies = [ "windows-implement", "windows-interface", - "windows-link", - "windows-result", - "windows-strings", + "windows-link 0.2.1", + "windows-result 0.4.1", + "windows-strings 0.5.1", +] + +[[package]] +name = "windows-future" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc6a41e98427b19fe4b73c550f060b59fa592d7d686537eebf9385621bfbad8e" +dependencies = [ + "windows-core 0.61.2", + "windows-link 0.1.3", + "windows-threading 0.1.0", ] [[package]] @@ -11391,9 +11786,9 @@ version = "0.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e1d6f90251fe18a279739e78025bd6ddc52a7e22f921070ccdc67dde84c605cb" dependencies = [ - "windows-core", - "windows-link", - "windows-threading", + "windows-core 0.62.2", + "windows-link 0.2.1", + "windows-threading 0.2.1", ] [[package]] @@ -11418,20 +11813,36 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "windows-link" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e6ad25900d524eaabdbbb96d20b4311e1e7ae1699af4fb28c17ae66c80d798a" + [[package]] name = "windows-link" version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" +[[package]] +name = "windows-numerics" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9150af68066c4c5c07ddc0ce30421554771e528bde427614c61038bc2c92c2b1" +dependencies = [ + "windows-core 0.61.2", + "windows-link 0.1.3", +] + [[package]] name = "windows-numerics" version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6e2e40844ac143cdb44aead537bbf727de9b044e107a0f1220392177d15b0f26" dependencies = [ - "windows-core", - "windows-link", + "windows-core 0.62.2", + "windows-link 0.2.1", ] [[package]] @@ -11440,9 +11851,18 @@ version = "0.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "02752bf7fbdcce7f2a27a742f798510f3e5ad88dbe84871e5168e2120c3d5720" dependencies = [ - "windows-link", - "windows-result", - "windows-strings", + "windows-link 0.2.1", + "windows-result 0.4.1", + "windows-strings 0.5.1", +] + +[[package]] +name = "windows-result" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "56f42bd332cc6c8eac5af113fc0c1fd6a8fd2aa08a0119358686e5160d0586c6" +dependencies = [ + "windows-link 0.1.3", ] [[package]] @@ -11451,7 +11871,7 @@ version = "0.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7781fa89eaf60850ac3d2da7af8e5242a5ea78d1a11c49bf2910bb5a73853eb5" dependencies = [ - "windows-link", + "windows-link 0.2.1", ] [[package]] @@ -11465,13 +11885,22 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "windows-strings" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "56e6c93f3a0c3b36176cb1327a4958a0353d5d166c2a35cb268ace15e91d3b57" +dependencies = [ + "windows-link 0.1.3", +] + [[package]] name = "windows-strings" version = "0.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7837d08f69c77cf6b07689544538e017c1bfcf57e34b4c0ff58e6c2cd3b37091" dependencies = [ - "windows-link", + "windows-link 0.2.1", ] [[package]] @@ -11507,7 +11936,7 @@ version = "0.61.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" dependencies = [ - "windows-link", + "windows-link 0.2.1", ] [[package]] @@ -11556,13 +11985,22 @@ dependencies = [ "windows_x86_64_msvc 0.52.6", ] +[[package]] +name = "windows-threading" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b66463ad2e0ea3bbf808b7f1d371311c80e115c0b71d60efc142cafbcfb057a6" +dependencies = [ + "windows-link 0.1.3", +] + [[package]] name = "windows-threading" version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3949bd5b99cafdf1c7ca86b43ca564028dfe27d66958f2470940f73d86d75b37" dependencies = [ - "windows-link", + "windows-link 0.2.1", ] [[package]] @@ -11697,6 +12135,15 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" +[[package]] +name = "winnow" +version = "0.6.26" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e90edd2ac1aa278a5c4599b1d89cf03074b610800f866d4026dc199d7929a28" +dependencies = [ + "memchr", +] + [[package]] name = "winnow" version = "0.7.15" @@ -11807,6 +12254,12 @@ dependencies = [ "hashlink", ] +[[package]] +name = "yansi" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cfe53a6657fd280eaa890a3bc59152892ffa3e30101319d168b781ed6529b049" + [[package]] name = "yasna" version = "0.5.2" diff --git a/Cargo.toml b/Cargo.toml index fd6ba0ffe..1f548d3c1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -40,6 +40,7 @@ members = [ "examples/basic-worker-rs", "examples/endpoint-agent", "examples/byoc-database", + "examples/slatedb-trace-store", "crates/alien-test", "tests/e2e/test-apps/comprehensive-rust", "tests/e2e/test-apps/container-rust", @@ -209,6 +210,7 @@ zip = { version = "2", default-features = false, features = ["deflate"] } pem = "3.0" oci-tar-builder = "0.4" sha2 = "0.10" +slatedb = { version = "=0.13.0", default-features = false, features = ["moka"] } oauth2 = { version = "5.0", default-features = false, features = ["reqwest"] } cacache = { version = "13.1", default-features = false, features = ["tokio-runtime"] } strip-ansi-escapes = "0.2.1" diff --git a/crates/alien-bindings/src/providers/storage/local.rs b/crates/alien-bindings/src/providers/storage/local.rs index a3fd1d7b1..64d9f7b60 100644 --- a/crates/alien-bindings/src/providers/storage/local.rs +++ b/crates/alien-bindings/src/providers/storage/local.rs @@ -253,8 +253,7 @@ impl ObjectStore for LocalStorage { ) -> ObjectStoreResult { let dst = prefixed_path(&self.base_dir, location); if !opts.attributes.is_empty() { - return Err(object_store::Error::Generic { - store: "LocalStorage", + return Err(object_store::Error::NotSupported { source: Box::new(AlienError::new(ErrorData::OperationNotSupported { operation: "storage.put object attributes".to_string(), reason: "the local filesystem backend cannot persist object attributes" diff --git a/crates/alien-bindings/tests/storage.rs b/crates/alien-bindings/tests/storage.rs index 106fdb92c..ebb9edac7 100644 --- a/crates/alien-bindings/tests/storage.rs +++ b/crates/alien-bindings/tests/storage.rs @@ -112,7 +112,7 @@ async fn local_storage_rejects_object_attributes_without_writing_the_payload() { .await .expect_err("local storage must not silently discard object attributes"); - assert!(matches!(&error, object_store::Error::Generic { .. })); + assert!(matches!(&error, object_store::Error::NotSupported { .. })); assert!(error .to_string() .contains("the local filesystem backend cannot persist object attributes")); diff --git a/examples/package.json b/examples/package.json index c4b927cbb..eff112ad0 100644 --- a/examples/package.json +++ b/examples/package.json @@ -4,6 +4,6 @@ "packageManager": "pnpm@10.11.0", "scripts": { "test": "pnpm run test:projects", - "test:projects": "pnpm -C byob-storage-ts check && pnpm -C basic-worker-ts test && pnpm -C remote-worker-ts test && pnpm -C data-connector-ts test && pnpm -C event-pipeline-ts test && pnpm -C command-routing-ts test:ts" + "test:projects": "pnpm -C byob-storage-ts check && pnpm -C basic-worker-ts test && pnpm -C remote-worker-ts test && pnpm -C data-connector-ts test && pnpm -C event-pipeline-ts test && pnpm -C command-routing-ts test:ts && pnpm -C slatedb-trace-store test" } } diff --git a/examples/pnpm-lock.yaml b/examples/pnpm-lock.yaml index 28b07d178..7f178d5c6 100644 --- a/examples/pnpm-lock.yaml +++ b/examples/pnpm-lock.yaml @@ -532,6 +532,24 @@ importers: specifier: ^3.1.4 version: 3.2.7(@types/debug@4.1.13)(@types/node@24.12.0)(jiti@2.7.0)(lightningcss@1.32.0) + slatedb-trace-store: + devDependencies: + '@alienplatform/core': + specifier: file:../../packages/core + version: file:../packages/core(@types/json-schema@7.0.15)(openapi-types@12.1.3) + '@alienplatform/testing': + specifier: file:../../packages/testing + version: file:../packages/testing(@aws-sdk/client-ssm@3.1004.0)(@azure/identity@4.13.0)(@azure/keyvault-secrets@4.10.0(@azure/core-client@1.10.1))(@google-cloud/secret-manager@5.6.0)(@types/json-schema@7.0.15)(openapi-types@12.1.3) + '@types/node': + specifier: ^22.10.5 + version: 22.19.17 + typescript: + specifier: ^5.7.3 + version: 5.8.3 + vitest: + specifier: ^3.2.6 + version: 3.2.7(@types/debug@4.1.13)(@types/node@22.19.17)(jiti@2.7.0)(lightningcss@1.32.0) + webhook-api-ts: dependencies: '@alienplatform/core': diff --git a/examples/pnpm-workspace.yaml b/examples/pnpm-workspace.yaml index 51983d7df..40fef162f 100644 --- a/examples/pnpm-workspace.yaml +++ b/examples/pnpm-workspace.yaml @@ -12,6 +12,7 @@ packages: - webhook-api-ts - nextjs-app - byoc-database + - slatedb-trace-store - endpoint-agent - full-stack-microservices - full-stack-microservices/services/* diff --git a/examples/slatedb-trace-store/Cargo.toml b/examples/slatedb-trace-store/Cargo.toml new file mode 100644 index 000000000..740aae802 --- /dev/null +++ b/examples/slatedb-trace-store/Cargo.toml @@ -0,0 +1,25 @@ +[package] +name = "slatedb-trace-store" +version = "0.1.0" +edition = "2021" +publish = false + +[dependencies] +alien-error = { workspace = true, features = ["axum"] } +alien-sdk = { workspace = true, default-features = false, features = ["all-platforms"] } +axum = { workspace = true, features = ["tokio", "http1", "json", "query"] } +bytes = { workspace = true } +chrono = { workspace = true, features = ["serde"] } +hex = { workspace = true } +object_store = { workspace = true } +serde = { workspace = true, features = ["derive"] } +serde_json = { workspace = true } +sha2 = { workspace = true } +slatedb = { workspace = true } +tokio = { workspace = true, features = ["full"] } +tower-http = { version = "0.6.2", features = ["trace"] } +tracing = { workspace = true } +tracing-subscriber = { workspace = true, features = ["env-filter", "json"] } + +[dev-dependencies] +tempfile = { workspace = true } diff --git a/examples/slatedb-trace-store/README.md b/examples/slatedb-trace-store/README.md new file mode 100644 index 000000000..0d011f6ba --- /dev/null +++ b/examples/slatedb-trace-store/README.md @@ -0,0 +1,54 @@ +# SlateDB trace store + +This example deploys durable AI trace history into one customer's cloud. A horizontally scalable HTTP API accepts and queries traces. One background writer commits them to [SlateDB](https://slatedb.io), backed by the customer's object storage. + +The deployment is the tenant boundary. There is intentionally no organization ID in the API, queue messages, or storage keys. Deploy another stack for another customer. + +## Architecture + +| Resource | Lifecycle | Purpose | +| --- | --- | --- | +| `data` Storage | Frozen | SlateDB files, staged traces, and rejected-ingestion metadata | +| `ingestion` Queue | Frozen | Durable handoff from API replicas to the writer | +| `api` Container | Live, 2–4 replicas | `POST` and indexed `GET` requests | +| `writer` Container | Live, 1 replica | The single SlateDB writer and queue consumer | + +The API returns `202 Accepted` after the canonical trace object and queue pointer are durable. The trace becomes queryable after the writer commits it and a reader observes the latest SlateDB manifest and WAL. Readers poll once per second. + +## Run locally + +```bash +alien dev +``` + +Submit a trace: + +```bash +curl -i http://localhost:8080/v1/traces \ + -H 'content-type: application/json' \ + -d '{ + "traceId": "run-01", + "agent": "researcher", + "status": "completed", + "model": "claude-sonnet", + "startedAt": "2026-08-26T18:00:00Z", + "finishedAt": "2026-08-26T18:00:04Z", + "payload": {"events": [{"type": "tool", "name": "search"}]} + }' +``` + +Read it after it becomes visible: + +```bash +curl http://localhost:8080/v1/traces/run-01 +curl 'http://localhost:8080/v1/traces?agent=researcher&status=completed&limit=25' +``` + +Run the behavior tests: + +```bash +cargo nextest run -p slatedb-trace-store +pnpm test +``` + +See the [complete guide](https://alien.dev/docs/examples/trace-history) for the data model, guarantees, failure behavior, and production integration points. diff --git a/examples/slatedb-trace-store/alien.ts b/examples/slatedb-trace-store/alien.ts new file mode 100644 index 000000000..f77702af9 --- /dev/null +++ b/examples/slatedb-trace-store/alien.ts @@ -0,0 +1,62 @@ +import * as alien from "@alienplatform/core" + +const data = new alien.Storage("data").lifecycleRules([{ prefix: "staging/v1/", days: 7 }]).build() +const ingestion = new alien.Queue("ingestion").build() + +const code = { + type: "source" as const, + src: ".", + toolchain: { type: "rust" as const, binaryName: "slatedb-trace-store" }, +} + +const api = new alien.Container("api") + .code(code) + .cpu(0.5) + .memory("512Mi") + .port(8080) + .autoScale({ + min: 2, + desired: 2, + max: 4, + targetHttpInFlightPerReplica: 100, + }) + .publicEndpoint("api", 8080, "http") + .healthCheck({ path: "/health", method: "GET", timeoutSeconds: 2, failureThreshold: 3 }) + .environment({ TRACE_STORE_MODE: "api", PORT: "8080", RUST_LOG: "info" }) + .permissions("api") + .link(data) + .link(ingestion) + .build() + +const writer = new alien.Container("writer") + .code(code) + .cpu(1) + .memory("1Gi") + .port(8081) + .replicas(1) + .healthCheck({ path: "/health", method: "GET", timeoutSeconds: 2, failureThreshold: 3 }) + .environment({ TRACE_STORE_MODE: "writer", PORT: "8081", RUST_LOG: "info" }) + .permissions("writer") + .link(data) + .link(ingestion) + .build() + +export default new alien.Stack("slatedb-trace-store") + .platforms(["aws", "gcp", "azure"]) + .add(data, "frozen") + .add(ingestion, "frozen") + .add(api, "live") + .add(writer, "live") + .permissions({ + profiles: { + api: { + data: ["storage/data-read", "storage/data-write"], + ingestion: ["queue/data-write"], + }, + writer: { + data: ["storage/data-read", "storage/data-write"], + ingestion: ["queue/data-read", "queue/data-write"], + }, + }, + }) + .build() diff --git a/examples/slatedb-trace-store/package.json b/examples/slatedb-trace-store/package.json new file mode 100644 index 000000000..25123e879 --- /dev/null +++ b/examples/slatedb-trace-store/package.json @@ -0,0 +1,16 @@ +{ + "name": "slatedb-trace-store", + "version": "0.1.0", + "private": true, + "type": "module", + "scripts": { + "test": "vitest run" + }, + "devDependencies": { + "@alienplatform/core": "^1.7.0", + "@alienplatform/testing": "^0.1.0", + "@types/node": "^22.10.5", + "typescript": "^5.7.3", + "vitest": "^3.2.6" + } +} diff --git a/examples/slatedb-trace-store/src/error.rs b/examples/slatedb-trace-store/src/error.rs new file mode 100644 index 000000000..b9c2da677 --- /dev/null +++ b/examples/slatedb-trace-store/src/error.rs @@ -0,0 +1,177 @@ +use alien_error::AlienErrorData; +use axum::{ + http::StatusCode, + response::{IntoResponse, Response}, + Json, +}; +use serde::{Deserialize, Serialize}; + +/// Infrastructure and startup errors that benefit from Alien's retry and +/// internal-detail metadata. +#[derive(Debug, Clone, AlienErrorData, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub enum ErrorData { + /// The process configuration is incomplete or invalid. + #[error( + code = "CONFIGURATION_INVALID", + message = "Invalid configuration: {message}", + retryable = "false", + internal = "false" + )] + ConfigurationInvalid { + /// Configuration failure. + message: String, + }, + + /// An Alien binding operation failed. + #[error( + code = "BINDING_OPERATION_FAILED", + message = "Binding operation '{operation}' failed", + retryable = "inherit", + internal = "true" + )] + BindingOperationFailed { + /// Operation being performed. + operation: String, + }, + + /// An object-storage operation failed. + #[error( + code = "STORAGE_OPERATION_FAILED", + message = "Storage operation '{operation}' failed for '{path}'", + retryable = "inherit", + internal = "true" + )] + StorageOperationFailed { + /// Operation being performed. + operation: String, + /// Object path involved. + path: String, + }, + + /// A queue operation failed. + #[error( + code = "QUEUE_OPERATION_FAILED", + message = "Queue operation '{operation}' failed", + retryable = "inherit", + internal = "true" + )] + QueueOperationFailed { + /// Operation being performed. + operation: String, + }, + + /// A SlateDB operation failed. + #[error( + code = "DATABASE_OPERATION_FAILED", + message = "Database operation '{operation}' failed", + retryable = "inherit", + internal = "true" + )] + DatabaseOperationFailed { + /// Operation being performed. + operation: String, + }, + + /// JSON encoding or decoding failed inside the service. + #[error( + code = "SERIALIZATION_FAILED", + message = "Serialization operation '{operation}' failed", + retryable = "false", + internal = "true" + )] + SerializationFailed { + /// Operation being performed. + operation: String, + }, +} + +pub type Error = alien_error::AlienError; +pub type Result = alien_error::Result; + +/// Small, application-owned errors for expected HTTP outcomes. +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ApiError { + pub code: &'static str, + pub message: String, + #[serde(skip)] + pub status: StatusCode, +} + +impl ApiError { + pub fn invalid(message: impl Into) -> Self { + Self { + code: "TRACE_INVALID", + message: message.into(), + status: StatusCode::BAD_REQUEST, + } + } + + pub fn not_found(trace_id: &str) -> Self { + Self { + code: "TRACE_NOT_FOUND", + message: format!("Trace '{trace_id}' was not found"), + status: StatusCode::NOT_FOUND, + } + } + + pub fn conflict(trace_id: &str) -> Self { + Self { + code: "TRACE_CONFLICT", + message: format!("Trace '{trace_id}' already exists with different content"), + status: StatusCode::CONFLICT, + } + } + + pub fn invalid_cursor() -> Self { + Self { + code: "CURSOR_INVALID", + message: "Invalid pagination cursor".to_string(), + status: StatusCode::BAD_REQUEST, + } + } + + pub fn not_ready() -> Self { + Self { + code: "TRACE_STORE_NOT_READY", + message: "The trace store is starting; try again shortly".to_string(), + status: StatusCode::SERVICE_UNAVAILABLE, + } + } +} + +impl IntoResponse for ApiError { + fn into_response(self) -> Response { + (self.status, Json(self)).into_response() + } +} + +#[derive(Debug)] +pub enum AppError { + Api(ApiError), + Internal(Error), +} + +impl From for AppError { + fn from(error: ApiError) -> Self { + Self::Api(error) + } +} + +impl From for AppError { + fn from(error: Error) -> Self { + Self::Internal(error) + } +} + +impl IntoResponse for AppError { + fn into_response(self) -> Response { + match self { + Self::Api(error) => error.into_response(), + Self::Internal(error) => error.into_response(), + } + } +} + +pub type AppResult = std::result::Result; diff --git a/examples/slatedb-trace-store/src/keys.rs b/examples/slatedb-trace-store/src/keys.rs new file mode 100644 index 000000000..c9bfda649 --- /dev/null +++ b/examples/slatedb-trace-store/src/keys.rs @@ -0,0 +1,115 @@ +use crate::{ + error::ApiError, + models::{Trace, TraceQuery}, +}; + +const AGENT_BIT: u8 = 1; +const STATUS_BIT: u8 = 2; +const MODEL_BIT: u8 = 4; + +pub fn primary_key(trace_id: &str) -> String { + format!("t/{}", hex::encode(trace_id)) +} + +pub fn index_keys(trace: &Trace) -> impl Iterator + '_ { + (0..8).map(|mask| index_key(mask, trace)) +} + +fn index_key(mask: u8, trace: &Trace) -> String { + format!( + "{}{time:016x}/{}", + index_prefix( + mask, + (mask & AGENT_BIT != 0).then_some(trace.agent.as_str()), + (mask & STATUS_BIT != 0).then_some(trace.status.as_str()), + (mask & MODEL_BIT != 0).then_some(trace.model.as_str()), + ), + hex::encode(&trace.trace_id), + time = trace.started_at.timestamp_millis() as u64, + ) +} + +pub fn query_prefix(query: &TraceQuery) -> String { + let mask = (u8::from(query.agent.is_some()) * AGENT_BIT) + | (u8::from(query.status.is_some()) * STATUS_BIT) + | (u8::from(query.model.is_some()) * MODEL_BIT); + index_prefix( + mask, + query.agent.as_deref(), + query.status.as_deref(), + query.model.as_deref(), + ) +} + +fn index_prefix( + mask: u8, + agent: Option<&str>, + status: Option<&str>, + model: Option<&str>, +) -> String { + let mut prefix = format!("i/{mask:x}/"); + for value in [agent, status, model].into_iter().flatten() { + prefix.push_str(&hex::encode(value)); + prefix.push('/'); + } + prefix +} + +pub fn timestamp_from_index_key(key: &[u8], prefix: &str) -> std::result::Result { + let key = std::str::from_utf8(key).map_err(|_| ApiError::invalid_cursor())?; + let suffix = key + .strip_prefix(prefix) + .ok_or_else(ApiError::invalid_cursor)?; + let encoded = suffix + .split('/') + .next() + .ok_or_else(ApiError::invalid_cursor)?; + u64::from_str_radix(encoded, 16) + .map(|value| value as i64) + .map_err(|_| ApiError::invalid_cursor()) +} + +pub fn decode_cursor(cursor: Option<&str>) -> std::result::Result>, ApiError> { + cursor + .map(|value| hex::decode(value).map_err(|_| ApiError::invalid_cursor())) + .transpose() +} + +#[cfg(test)] +mod tests { + use super::*; + use chrono::TimeZone; + use serde_json::json; + + fn trace() -> Trace { + Trace { + trace_id: "trace/one".to_string(), + agent: "research/agent".to_string(), + status: "complete".to_string(), + model: "claude-sonnet".to_string(), + started_at: chrono::Utc.timestamp_millis_opt(1_700_000_000_000).unwrap(), + finished_at: None, + payload: json!({}), + } + } + + #[test] + fn writes_one_index_for_every_filter_subset() { + let keys = index_keys(&trace()).collect::>(); + + assert_eq!(keys.len(), 8); + assert_eq!(keys.iter().filter(|key| key.starts_with("i/7/")).count(), 1); + } + + #[test] + fn query_prefix_matches_corresponding_index() { + let query = TraceQuery { + agent: Some("research/agent".to_string()), + model: Some("claude-sonnet".to_string()), + ..TraceQuery::default() + }; + let prefix = query_prefix(&query); + + assert!(index_keys(&trace()).any(|key| key.starts_with(&prefix))); + } +} diff --git a/examples/slatedb-trace-store/src/main.rs b/examples/slatedb-trace-store/src/main.rs new file mode 100644 index 000000000..e5cf3ac83 --- /dev/null +++ b/examples/slatedb-trace-store/src/main.rs @@ -0,0 +1,136 @@ +#![allow(clippy::result_large_err)] + +mod error; +mod keys; +mod models; +mod service; +mod store; + +use crate::{ + error::{ErrorData, Result}, + service::{ + get_trace, health, ingest, list_traces, open_writer, run_writer, writer_health, ApiState, + WriterHealth, + }, +}; +use alien_error::{Context, IntoAlienError}; +use alien_sdk::Bindings; +use axum::{routing::get, Router}; +use std::{net::SocketAddr, str::FromStr, sync::Arc}; +use tower_http::trace::TraceLayer; +use tracing_subscriber::{fmt, layer::SubscriberExt, util::SubscriberInitExt, EnvFilter}; + +#[derive(Debug, Clone, Copy)] +enum Mode { + Api, + Writer, +} + +impl FromStr for Mode { + type Err = crate::error::Error; + + fn from_str(value: &str) -> Result { + match value { + "api" => Ok(Self::Api), + "writer" => Ok(Self::Writer), + other => Err(alien_error::AlienError::new( + ErrorData::ConfigurationInvalid { + message: format!("TRACE_STORE_MODE must be 'api' or 'writer', got '{other}'"), + }, + )), + } + } +} + +#[tokio::main] +async fn main() -> Result<()> { + tracing_subscriber::registry() + .with( + fmt::layer() + .json() + .with_ansi(false) + .with_target(false) + .with_current_span(false), + ) + .with(EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"))) + .init(); + + let mode = std::env::var("TRACE_STORE_MODE") + .into_alien_error() + .context(ErrorData::ConfigurationInvalid { + message: "TRACE_STORE_MODE is required".to_string(), + })? + .parse()?; + let port = std::env::var("PORT") + .unwrap_or_else(|_| "8080".to_string()) + .parse::() + .into_alien_error() + .context(ErrorData::ConfigurationInvalid { + message: "PORT must be a valid TCP port".to_string(), + })?; + let bindings = Bindings::from_env().context(ErrorData::BindingOperationFailed { + operation: "load bindings".to_string(), + })?; + let storage = bindings + .storage("data") + .await + .context(ErrorData::BindingOperationFailed { + operation: "load data storage".to_string(), + })?; + let queue = bindings + .queue("ingestion") + .await + .context(ErrorData::BindingOperationFailed { + operation: "load ingestion queue".to_string(), + })?; + + let (app, writer) = match mode { + Mode::Api => { + let state = Arc::new(ApiState::new(storage, queue)); + ( + Router::new() + .route("/health", get(health)) + .route("/v1/traces", get(list_traces).post(ingest)) + .route("/v1/traces/{trace_id}", get(get_trace)) + .with_state(state), + None, + ) + } + Mode::Writer => { + let (writer, storage, queue) = open_writer(storage, queue).await?; + let health = Arc::new(WriterHealth::new()); + ( + Router::new() + .route("/health", get(writer_health)) + .with_state(Arc::clone(&health)), + Some(run_writer(writer, storage, queue, health)), + ) + } + }; + let app = app.layer(TraceLayer::new_for_http()); + let address = SocketAddr::from(([0, 0, 0, 0], port)); + let listener = tokio::net::TcpListener::bind(address) + .await + .into_alien_error() + .context(ErrorData::ConfigurationInvalid { + message: format!("could not bind HTTP server to {address}"), + })?; + let server = axum::serve(listener, app); + tracing::info!(%address, ?mode, "trace-history service started"); + + if let Some(writer) = writer { + tokio::select! { + result = server => result.into_alien_error().context(ErrorData::ConfigurationInvalid { + message: "HTTP server stopped".to_string(), + }), + result = writer => result, + } + } else { + server + .await + .into_alien_error() + .context(ErrorData::ConfigurationInvalid { + message: "HTTP server stopped".to_string(), + }) + } +} diff --git a/examples/slatedb-trace-store/src/models.rs b/examples/slatedb-trace-store/src/models.rs new file mode 100644 index 000000000..7a7ccddbf --- /dev/null +++ b/examples/slatedb-trace-store/src/models.rs @@ -0,0 +1,146 @@ +use crate::error::ApiError; +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +pub const MAX_TRACE_BYTES: usize = 100 * 1024; +pub const DEFAULT_PAGE_SIZE: usize = 50; +pub const MAX_PAGE_SIZE: usize = 100; + +/// A durable AI trace accepted by the service. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct Trace { + /// Stable, caller-assigned identifier used for idempotency. + pub trace_id: String, + /// Agent or engine that produced the trace. + pub agent: String, + /// Final or current trace status. + pub status: String, + /// Model used by the agent. + pub model: String, + /// Time at which execution began. + pub started_at: DateTime, + /// Time at which execution ended, when known. + pub finished_at: Option>, + /// Application-specific trace content. + pub payload: Value, +} + +impl Trace { + pub fn validate(&self, encoded_len: usize) -> std::result::Result<(), ApiError> { + for (field, value) in [ + ("traceId", self.trace_id.as_str()), + ("agent", self.agent.as_str()), + ("status", self.status.as_str()), + ("model", self.model.as_str()), + ] { + if value.trim().is_empty() { + return Err(ApiError::invalid(format!("{field} must not be empty"))); + } + if value.len() > 512 { + return Err(ApiError::invalid(format!( + "{field} must not exceed 512 bytes" + ))); + } + } + if encoded_len > MAX_TRACE_BYTES { + return Err(ApiError::invalid(format!( + "encoded trace must not exceed {MAX_TRACE_BYTES} bytes" + ))); + } + if self.started_at.timestamp_millis() < 0 { + return Err(ApiError::invalid( + "startedAt must be on or after 1970-01-01", + )); + } + if self + .finished_at + .is_some_and(|finished| finished < self.started_at) + { + return Err(ApiError::invalid("finishedAt must not precede startedAt")); + } + Ok(()) + } +} + +/// Trace plus database-managed commit metadata. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct StoredTrace { + #[serde(flatten)] + pub trace: Trace, + pub content_hash: String, + pub committed_at: DateTime, +} + +/// Response returned after an ingestion request becomes durable. +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct AcceptedTrace { + pub trace_id: String, + pub content_hash: String, + pub status: &'static str, +} + +/// Filters and pagination accepted by the list endpoint. +#[derive(Debug, Default, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct TraceQuery { + pub agent: Option, + pub status: Option, + pub model: Option, + pub started_after: Option>, + pub started_before: Option>, + pub cursor: Option, + pub limit: Option, +} + +impl TraceQuery { + pub fn page_size(&self) -> std::result::Result { + let limit = self.limit.unwrap_or(DEFAULT_PAGE_SIZE); + if !(1..=MAX_PAGE_SIZE).contains(&limit) { + return Err(ApiError::invalid(format!( + "limit must be between 1 and {MAX_PAGE_SIZE}" + ))); + } + if self + .started_after + .zip(self.started_before) + .is_some_and(|(after, before)| after > before) + { + return Err(ApiError::invalid( + "startedAfter must not follow startedBefore", + )); + } + Ok(limit) + } +} + +/// A page of committed traces. +#[derive(Debug, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct TracePage { + pub traces: Vec, + pub next_cursor: Option, +} + +/// Small queue message pointing at a staged trace object. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct IngestionPointer { + pub trace_id: String, + pub content_hash: String, + pub staging_path: String, +} + +/// Metadata retained when a queued trace cannot be committed. +#[derive(Debug, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct IngestionFailure { + pub trace_id: Option, + pub staging_path: Option, + pub reason: String, + pub failed_at: DateTime, + pub queue_attempt: u32, +} diff --git a/examples/slatedb-trace-store/src/service.rs b/examples/slatedb-trace-store/src/service.rs new file mode 100644 index 000000000..22b2f9ef4 --- /dev/null +++ b/examples/slatedb-trace-store/src/service.rs @@ -0,0 +1,440 @@ +use crate::{ + error::{AppError, AppResult, ErrorData, Result}, + models::{AcceptedTrace, IngestionFailure, IngestionPointer, Trace, TracePage, TraceQuery}, + store::{TraceReader, TraceWriter}, +}; +use alien_error::{Context, ContextError, IntoAlienError, IntoAlienErrorDirect}; +use alien_sdk::{ + traits::{MessagePayload, Storage}, + Queue, +}; +use axum::{ + extract::{Path as AxumPath, Query, State}, + http::StatusCode, + response::IntoResponse, + Json, +}; +use bytes::Bytes; +use object_store::{path::Path, ObjectStore}; +use serde_json::json; +use sha2::{Digest, Sha256}; +use std::sync::{ + atomic::{AtomicBool, Ordering}, + Arc, +}; +use tokio::{sync::RwLock, time::Duration}; + +const MAX_QUEUE_ATTEMPTS: usize = 5; + +pub struct WriterHealth { + // Health is process-wide: can this writer still receive work? A failed + // per-message ack is not a liveness failure. The message remains durable + // in the queue and is retried after its lease, while the writer continues + // processing other traces. Restarting the writer would not resolve that + // delivery and would unnecessarily interrupt healthy work. + receive: AtomicBool, +} + +impl WriterHealth { + pub fn new() -> Self { + Self { + receive: AtomicBool::new(true), + } + } + + fn is_healthy(&self) -> bool { + self.receive.load(Ordering::Relaxed) + } +} + +pub struct ApiState { + storage: Arc, + queue: Queue, + reader: RwLock>>, +} + +impl ApiState { + pub fn new(storage: Arc, queue: Queue) -> Self { + Self { + storage, + queue, + reader: RwLock::new(None), + } + } + + async fn reader(&self) -> AppResult> { + if let Some(reader) = self.reader.read().await.as_ref() { + return Ok(Arc::clone(reader)); + } + let mut slot = self.reader.write().await; + if let Some(reader) = slot.as_ref() { + return Ok(Arc::clone(reader)); + } + let object_store: Arc = Arc::clone(&self.storage) as Arc; + let reader = Arc::new(TraceReader::open(object_store).await.map_err(|error| { + // On a brand-new deployment the writer may not have created SlateDB's + // first manifest yet. Keep this expected startup race friendly to callers, + // while retaining the underlying detail in the service log. + tracing::info!(code = %error.code, "trace store reader is not ready"); + crate::error::ApiError::not_ready() + })?); + *slot = Some(Arc::clone(&reader)); + Ok(reader) + } +} + +pub async fn health() -> impl IntoResponse { + (StatusCode::OK, Json(json!({ "status": "ok" }))) +} + +pub async fn writer_health(State(health): State>) -> impl IntoResponse { + if health.is_healthy() { + (StatusCode::OK, Json(json!({ "status": "ok" }))) + } else { + ( + StatusCode::SERVICE_UNAVAILABLE, + Json(json!({ "status": "degraded" })), + ) + } +} + +pub async fn ingest( + State(state): State>, + Json(trace): Json, +) -> AppResult<(StatusCode, Json)> { + let encoded = + serde_json::to_vec(&trace) + .into_alien_error() + .context(ErrorData::SerializationFailed { + operation: "encode submitted trace".to_string(), + })?; + trace.validate(encoded.len())?; + + let content_hash = hex::encode(Sha256::digest(&encoded)); + let staging_path = format!( + "staging/v1/{}/{}.json", + hex::encode(&trace.trace_id), + content_hash + ); + state + .storage + .put( + &Path::from(staging_path.as_str()), + Bytes::from(encoded).into(), + ) + .await + .into_alien_error() + .context(ErrorData::StorageOperationFailed { + operation: "stage trace".to_string(), + path: staging_path.clone(), + })?; + + let pointer = IngestionPointer { + trace_id: trace.trace_id.clone(), + content_hash: content_hash.clone(), + staging_path, + }; + let pointer = serde_json::to_value(pointer).into_alien_error().context( + ErrorData::SerializationFailed { + operation: "encode ingestion pointer".to_string(), + }, + )?; + state + .queue + .send(MessagePayload::Json(pointer)) + .await + .context(ErrorData::QueueOperationFailed { + operation: "enqueue trace".to_string(), + })?; + + Ok(( + StatusCode::ACCEPTED, + Json(AcceptedTrace { + trace_id: trace.trace_id, + content_hash, + status: "accepted", + }), + )) +} + +pub async fn get_trace( + State(state): State>, + AxumPath(trace_id): AxumPath, +) -> AppResult> { + Ok(Json(state.reader().await?.get(&trace_id).await?)) +} + +pub async fn list_traces( + State(state): State>, + Query(query): Query, +) -> AppResult> { + Ok(Json(state.reader().await?.list(&query).await?)) +} + +pub async fn open_writer( + storage: Arc, + queue: Queue, +) -> Result<(TraceWriter, Arc, Queue)> { + let object_store: Arc = Arc::clone(&storage) as Arc; + let writer = TraceWriter::open(object_store).await?; + Ok((writer, storage, queue)) +} + +pub async fn run_writer( + writer: TraceWriter, + storage: Arc, + queue: Queue, + health: Arc, +) -> Result<()> { + let mut receive_failures = 0; + loop { + let messages = match queue.receive(10).await { + Ok(messages) => { + receive_failures = 0; + health.receive.store(true, Ordering::Relaxed); + messages + } + Err(error) => { + receive_failures += 1; + if receive_failures >= MAX_QUEUE_ATTEMPTS { + health.receive.store(false, Ordering::Relaxed); + } + tracing::warn!(%error, attempt = receive_failures, "could not receive traces; retrying"); + tokio::time::sleep(Duration::from_secs(1)).await; + continue; + } + }; + if messages.is_empty() { + tokio::time::sleep(Duration::from_millis(250)).await; + continue; + } + + for message in messages { + let receipt_handle = message.receipt_handle.clone(); + let pointer = decode_pointer(message.payload); + let outcome = match pointer { + Ok(pointer) => process_pointer(&writer, &storage, &pointer).await, + Err(reason) => Err(ProcessError::Permanent(PermanentFailure { + trace_id: None, + staging_path: None, + reason, + })), + }; + + match outcome { + Ok(staging_path) => { + let acknowledged = + ack_with_retry(&queue, &receipt_handle, "ack committed trace").await; + if acknowledged { + if let Err(error) = storage.delete(&Path::from(staging_path.as_str())).await + { + tracing::warn!(path = %staging_path, %error, "failed to delete committed staging object"); + } + } + } + Err(ProcessError::Permanent(failure)) => { + let recorded = + record_failure_with_retry(&storage, &failure, message.attempt).await; + if recorded { + ack_with_retry(&queue, &receipt_handle, "ack rejected trace").await; + } + } + Err(ProcessError::Retryable(error)) => { + tracing::warn!( + trace_id = ?error.trace_id, + code = %error.error.code, + "trace ingestion will be retried" + ); + if let Err(nack_error) = queue.nack(&receipt_handle).await { + // Some queues, including the current SQS binding, cannot + // shorten a message lease. Leaving it unacknowledged still + // allows redelivery after the lease expires, and must not + // terminate the sole writer. + tracing::warn!(%nack_error, "could not release trace early; waiting for lease-based redelivery"); + } + } + } + } + } +} + +async fn ack_with_retry(queue: &Queue, receipt_handle: &str, operation: &str) -> bool { + for attempt in 1..=MAX_QUEUE_ATTEMPTS { + match queue.ack(receipt_handle).await { + Ok(()) => return true, + Err(error) => { + if attempt >= MAX_QUEUE_ATTEMPTS { + tracing::error!(%error, %operation, "queue operation remains unavailable; leaving message for redelivery"); + return false; + } + tracing::warn!(%error, attempt, %operation, "queue operation failed; retrying"); + tokio::time::sleep(Duration::from_secs(1)).await; + } + } + } + unreachable!("the bounded retry loop always returns") +} + +async fn record_failure_with_retry( + storage: &Arc, + failure: &PermanentFailure, + attempt: u32, +) -> bool { + for storage_attempt in 1..=MAX_QUEUE_ATTEMPTS { + match record_failure(storage, failure, attempt).await { + Ok(()) => return true, + Err(error) => { + if storage_attempt >= MAX_QUEUE_ATTEMPTS { + tracing::error!(%error, "failure record remains unavailable; leaving message for redelivery"); + return false; + } + tracing::warn!(%error, attempt = storage_attempt, "could not record rejected trace; retrying"); + tokio::time::sleep(Duration::from_secs(1)).await; + } + } + } + unreachable!("the bounded retry loop always returns") +} + +struct PermanentFailure { + trace_id: Option, + staging_path: Option, + reason: String, +} + +struct RetryableFailure { + trace_id: Option, + error: crate::error::Error, +} + +enum ProcessError { + Permanent(PermanentFailure), + Retryable(RetryableFailure), +} + +fn decode_pointer(payload: MessagePayload) -> std::result::Result { + match payload { + MessagePayload::Json(value) => serde_json::from_value(value), + MessagePayload::Text(value) => serde_json::from_str(&value), + } + .map_err(|error| format!("invalid ingestion pointer: {error}")) +} + +async fn process_pointer( + writer: &TraceWriter, + storage: &Arc, + pointer: &IngestionPointer, +) -> std::result::Result { + let path = Path::from(pointer.staging_path.as_str()); + let result = storage.get(&path).await.map_err(|error| { + if matches!(error, object_store::Error::NotFound { .. }) { + ProcessError::Permanent(PermanentFailure { + trace_id: Some(pointer.trace_id.clone()), + staging_path: Some(pointer.staging_path.clone()), + reason: "staged trace object was not found".to_string(), + }) + } else { + ProcessError::Retryable(RetryableFailure { + trace_id: Some(pointer.trace_id.clone()), + error: error + .into_alien_error() + .context(ErrorData::StorageOperationFailed { + operation: "read staged trace".to_string(), + path: pointer.staging_path.clone(), + }), + }) + } + })?; + let bytes = result.bytes().await.map_err(|error| { + ProcessError::Retryable(RetryableFailure { + trace_id: Some(pointer.trace_id.clone()), + error: error + .into_alien_error() + .context(ErrorData::StorageOperationFailed { + operation: "read staged trace body".to_string(), + path: pointer.staging_path.clone(), + }), + }) + })?; + let actual_hash = hex::encode(Sha256::digest(&bytes)); + if actual_hash != pointer.content_hash { + return Err(ProcessError::Permanent(PermanentFailure { + trace_id: Some(pointer.trace_id.clone()), + staging_path: Some(pointer.staging_path.clone()), + reason: "staged trace content hash does not match queue pointer".to_string(), + })); + } + let trace: Trace = serde_json::from_slice(&bytes).map_err(|error| { + ProcessError::Permanent(PermanentFailure { + trace_id: Some(pointer.trace_id.clone()), + staging_path: Some(pointer.staging_path.clone()), + reason: format!("staged trace is invalid JSON: {error}"), + }) + })?; + if trace.trace_id != pointer.trace_id { + return Err(ProcessError::Permanent(PermanentFailure { + trace_id: Some(pointer.trace_id.clone()), + staging_path: Some(pointer.staging_path.clone()), + reason: "staged trace ID does not match queue pointer".to_string(), + })); + } + trace.validate(bytes.len()).map_err(|error| { + ProcessError::Permanent(PermanentFailure { + trace_id: Some(pointer.trace_id.clone()), + staging_path: Some(pointer.staging_path.clone()), + reason: error.message, + }) + })?; + if let Err(error) = writer.commit(trace, pointer.content_hash.clone()).await { + match error { + AppError::Api(error) => { + return Err(ProcessError::Permanent(PermanentFailure { + trace_id: Some(pointer.trace_id.clone()), + staging_path: Some(pointer.staging_path.clone()), + reason: format!("{}: {}", error.code, error.message), + })); + } + AppError::Internal(error) => { + return Err(ProcessError::Retryable(RetryableFailure { + trace_id: Some(pointer.trace_id.clone()), + error, + })); + } + } + } + Ok(pointer.staging_path.clone()) +} + +async fn record_failure( + storage: &Arc, + failure: &PermanentFailure, + attempt: u32, +) -> Result<()> { + let failed_at = chrono::Utc::now(); + let identity = serde_json::to_vec(&(&failure.trace_id, &failure.staging_path, &failure.reason)) + .into_alien_error() + .context(ErrorData::SerializationFailed { + operation: "encode ingestion failure identity".to_string(), + })?; + let path = format!("failures/v1/{}.json", hex::encode(Sha256::digest(identity))); + let body = serde_json::to_vec(&IngestionFailure { + trace_id: failure.trace_id.clone(), + staging_path: failure.staging_path.clone(), + reason: failure.reason.clone(), + failed_at, + queue_attempt: attempt, + }) + .into_alien_error() + .context(ErrorData::SerializationFailed { + operation: "encode ingestion failure".to_string(), + })?; + storage + .put(&Path::from(path.as_str()), Bytes::from(body).into()) + .await + .into_alien_error() + .context(ErrorData::StorageOperationFailed { + operation: "record ingestion failure".to_string(), + path, + })?; + Ok(()) +} diff --git a/examples/slatedb-trace-store/src/store.rs b/examples/slatedb-trace-store/src/store.rs new file mode 100644 index 000000000..2846a3e16 --- /dev/null +++ b/examples/slatedb-trace-store/src/store.rs @@ -0,0 +1,376 @@ +use crate::{ + error::{ApiError, AppResult, ErrorData, Result}, + keys::{decode_cursor, index_keys, primary_key, query_prefix, timestamp_from_index_key}, + models::{StoredTrace, Trace, TracePage, TraceQuery}, +}; +use alien_error::{Context, IntoAlienError}; +use object_store::ObjectStore; +use slatedb::{ + config::{DbReaderOptions, WriteOptions}, + db_cache::moka::{MokaCache, MokaCacheOptions}, + Db, DbReader, IsolationLevel, +}; +use std::{sync::Arc, time::Duration}; + +const DATABASE_PATH: &str = "db/v1"; +const CACHE_CAPACITY_BYTES: u64 = 128 * 1024 * 1024; + +#[derive(Debug, PartialEq)] +pub enum CommitResult { + Inserted, + AlreadyExists, +} + +pub struct TraceWriter { + db: Db, +} + +impl TraceWriter { + pub async fn open(object_store: Arc) -> Result { + let db = Db::builder(DATABASE_PATH, object_store) + .with_db_cache(Arc::new(MokaCache::new_with_opts(MokaCacheOptions { + max_capacity: CACHE_CAPACITY_BYTES, + time_to_live: None, + time_to_idle: None, + }))) + .build() + .await + .into_alien_error() + .context(ErrorData::DatabaseOperationFailed { + operation: "open writer".to_string(), + })?; + Ok(Self { db }) + } + + pub async fn commit(&self, trace: Trace, content_hash: String) -> AppResult { + let key = primary_key(&trace.trace_id); + let transaction = self + .db + .begin(IsolationLevel::SerializableSnapshot) + .await + .into_alien_error() + .context(ErrorData::DatabaseOperationFailed { + operation: "begin trace transaction".to_string(), + })?; + + if let Some(existing) = transaction + .get(key.as_bytes()) + .await + .into_alien_error() + .context(ErrorData::DatabaseOperationFailed { + operation: "check trace idempotency".to_string(), + })? + { + let existing: StoredTrace = serde_json::from_slice(&existing) + .into_alien_error() + .context(ErrorData::SerializationFailed { + operation: "decode existing trace".to_string(), + })?; + if existing.content_hash == content_hash { + return Ok(CommitResult::AlreadyExists); + } + return Err(ApiError::conflict(&trace.trace_id).into()); + } + + let stored = StoredTrace { + trace: trace.clone(), + content_hash, + committed_at: chrono::Utc::now(), + }; + let value = serde_json::to_vec(&stored).into_alien_error().context( + ErrorData::SerializationFailed { + operation: "encode committed trace".to_string(), + }, + )?; + + transaction + .put(key.as_bytes(), &value) + .into_alien_error() + .context(ErrorData::DatabaseOperationFailed { + operation: "buffer primary trace".to_string(), + })?; + for index_key in index_keys(&trace) { + transaction + .put(index_key.as_bytes(), key.as_bytes()) + .into_alien_error() + .context(ErrorData::DatabaseOperationFailed { + operation: "buffer trace index".to_string(), + })?; + } + transaction + .commit_with_options(&WriteOptions { + await_durable: true, + ..WriteOptions::default() + }) + .await + .into_alien_error() + .context(ErrorData::DatabaseOperationFailed { + operation: "durably commit trace".to_string(), + })?; + Ok(CommitResult::Inserted) + } +} + +pub struct TraceReader { + db: DbReader, +} + +impl TraceReader { + pub async fn open(object_store: Arc) -> Result { + let db = DbReader::builder(DATABASE_PATH, object_store) + .with_options(DbReaderOptions { + manifest_poll_interval: Duration::from_secs(1), + ..DbReaderOptions::default() + }) + .with_db_cache(Arc::new(MokaCache::new_with_opts(MokaCacheOptions { + max_capacity: CACHE_CAPACITY_BYTES, + time_to_live: None, + time_to_idle: None, + }))) + .build() + .await + .into_alien_error() + .context(ErrorData::DatabaseOperationFailed { + operation: "open reader".to_string(), + })?; + Ok(Self { db }) + } + + pub async fn get(&self, trace_id: &str) -> AppResult { + let bytes = self + .db + .get(primary_key(trace_id)) + .await + .into_alien_error() + .context(ErrorData::DatabaseOperationFailed { + operation: "read trace".to_string(), + })? + .ok_or_else(|| ApiError::not_found(trace_id))?; + Ok(decode_stored_trace(&bytes)?) + } + + pub async fn list(&self, query: &TraceQuery) -> AppResult { + let limit = query.page_size()?; + let prefix = query_prefix(query); + let cursor = decode_cursor(query.cursor.as_deref())?; + if cursor + .as_deref() + .is_some_and(|cursor| !cursor.starts_with(prefix.as_bytes())) + { + return Err(ApiError::invalid_cursor().into()); + } + + let mut iterator = self + .db + .scan_prefix(prefix.as_bytes()) + .await + .into_alien_error() + .context(ErrorData::DatabaseOperationFailed { + operation: "scan trace index".to_string(), + })?; + let mut traces = Vec::with_capacity(limit); + let mut last_returned_key = None; + let mut has_more = false; + + while let Some(entry) = iterator.next().await.into_alien_error().context( + ErrorData::DatabaseOperationFailed { + operation: "read trace index page".to_string(), + }, + )? { + if cursor + .as_deref() + .is_some_and(|cursor| entry.key.as_ref() <= cursor) + { + continue; + } + let timestamp = timestamp_from_index_key(&entry.key, &prefix)?; + if query + .started_after + .is_some_and(|after| timestamp < after.timestamp_millis()) + { + continue; + } + if query + .started_before + .is_some_and(|before| timestamp > before.timestamp_millis()) + { + break; + } + if traces.len() == limit { + has_more = true; + break; + } + let primary = self + .db + .get(&entry.value) + .await + .into_alien_error() + .context(ErrorData::DatabaseOperationFailed { + operation: "read indexed trace".to_string(), + })? + .ok_or_else(|| { + crate::error::Error::new(ErrorData::DatabaseOperationFailed { + operation: "resolve dangling trace index".to_string(), + }) + })?; + traces.push(decode_stored_trace(&primary)?); + last_returned_key = Some(entry.key); + } + + Ok(TracePage { + traces, + next_cursor: has_more.then(|| hex::encode(last_returned_key.unwrap_or_default())), + }) + } +} + +fn decode_stored_trace(bytes: &[u8]) -> Result { + serde_json::from_slice(bytes) + .into_alien_error() + .context(ErrorData::SerializationFailed { + operation: "decode committed trace".to_string(), + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::error::AppError; + use chrono::{TimeZone, Utc}; + use object_store::memory::InMemory; + use serde_json::json; + + fn trace(id: &str, agent: &str, status: &str, model: &str, timestamp: i64) -> Trace { + Trace { + trace_id: id.to_string(), + agent: agent.to_string(), + status: status.to_string(), + model: model.to_string(), + started_at: Utc.timestamp_millis_opt(timestamp).unwrap(), + finished_at: None, + payload: json!({ "id": id }), + } + } + + #[tokio::test] + async fn commit_is_idempotent_and_rejects_different_content() { + let storage: Arc = Arc::new(InMemory::new()); + let writer = TraceWriter::open(storage) + .await + .expect("writer should open"); + let original = trace("trace-1", "agent-a", "ok", "model-a", 1_000); + + let inserted = writer + .commit(original.clone(), "hash-a".to_string()) + .await + .expect("first commit should succeed"); + let duplicate = writer + .commit(original, "hash-a".to_string()) + .await + .expect("identical commit should succeed"); + let conflict = writer + .commit( + trace("trace-1", "agent-a", "failed", "model-a", 1_000), + "hash-b".to_string(), + ) + .await + .expect_err("different content should conflict"); + + assert_eq!(inserted, CommitResult::Inserted); + assert_eq!(duplicate, CommitResult::AlreadyExists); + assert!(matches!( + conflict, + AppError::Api(ApiError { + code: "TRACE_CONFLICT", + .. + }) + )); + } + + #[tokio::test] + async fn every_filter_combination_returns_only_matching_traces() { + let storage: Arc = Arc::new(InMemory::new()); + let writer = TraceWriter::open(Arc::clone(&storage)) + .await + .expect("writer should open"); + for (trace, hash) in [ + (trace("one", "a", "ok", "m1", 1_000), "h1"), + (trace("two", "a", "failed", "m2", 2_000), "h2"), + (trace("three", "b", "ok", "m2", 3_000), "h3"), + ] { + writer + .commit(trace, hash.to_string()) + .await + .expect("trace should commit"); + } + writer.db.flush().await.expect("writer should flush"); + let reader = TraceReader::open(storage) + .await + .expect("reader should open"); + + for (agent, status, model, expected) in [ + (None, None, None, 3), + (Some("a"), None, None, 2), + (None, Some("ok"), None, 2), + (None, None, Some("m2"), 2), + (Some("a"), Some("ok"), None, 1), + (Some("a"), None, Some("m2"), 1), + (None, Some("ok"), Some("m2"), 1), + (Some("a"), Some("failed"), Some("m2"), 1), + ] { + let page = reader + .list(&TraceQuery { + agent: agent.map(str::to_string), + status: status.map(str::to_string), + model: model.map(str::to_string), + ..TraceQuery::default() + }) + .await + .expect("filtered query should succeed"); + assert_eq!(page.traces.len(), expected); + } + } + + #[tokio::test] + async fn cursor_pages_without_duplicates() { + let storage: Arc = Arc::new(InMemory::new()); + let writer = TraceWriter::open(Arc::clone(&storage)) + .await + .expect("writer should open"); + for index in 0..3 { + writer + .commit( + trace(&format!("trace-{index}"), "a", "ok", "m", index), + format!("hash-{index}"), + ) + .await + .expect("trace should commit"); + } + writer.db.flush().await.expect("writer should flush"); + let reader = TraceReader::open(storage) + .await + .expect("reader should open"); + let first = reader + .list(&TraceQuery { + limit: Some(2), + ..TraceQuery::default() + }) + .await + .expect("first page should load"); + let second = reader + .list(&TraceQuery { + limit: Some(2), + cursor: first.next_cursor.clone(), + ..TraceQuery::default() + }) + .await + .expect("second page should load"); + + assert_eq!(first.traces.len(), 2); + assert_eq!(second.traces.len(), 1); + assert_ne!( + first.traces[1].trace.trace_id, + second.traces[0].trace.trace_id + ); + } +} diff --git a/examples/slatedb-trace-store/tests/trace-store.test.ts b/examples/slatedb-trace-store/tests/trace-store.test.ts new file mode 100644 index 000000000..58ba1813f --- /dev/null +++ b/examples/slatedb-trace-store/tests/trace-store.test.ts @@ -0,0 +1,119 @@ +import { type Deployment, deploy } from "@alienplatform/testing" +import { afterAll, beforeAll, describe, expect, it } from "vitest" + +type AcceptedTrace = { + traceId: string + contentHash: string + status: "accepted" +} + +type StoredTrace = { + traceId: string + agent: string + status: string + model: string + payload: unknown +} + +type TracePage = { + traces: StoredTrace[] + nextCursor?: string +} + +async function waitForTrace(url: string, traceId: string): Promise { + for (let attempt = 0; attempt < 30; attempt++) { + const response = await fetch(`${url}/v1/traces/${traceId}`) + if (response.ok) return response.json() as Promise + expect([404, 503]).toContain(response.status) + await new Promise(resolve => setTimeout(resolve, 500)) + } + throw new Error(`trace '${traceId}' did not become visible`) +} + +describe("SlateDB trace store", () => { + let deployment: Deployment + + beforeAll(async () => { + deployment = await deploy({ app: ".", platform: "local" }) + }) + + afterAll(async () => { + await deployment?.destroy() + }) + + it("durably accepts, commits, reads, and filters a trace", async () => { + const trace = { + traceId: "integration-trace-1", + agent: "researcher", + status: "completed", + model: "claude-sonnet", + startedAt: "2026-08-26T18:00:00Z", + finishedAt: "2026-08-26T18:00:04Z", + payload: { answer: 42 }, + } + const acceptedResponse = await fetch(`${deployment.url}/v1/traces`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(trace), + }) + + expect(acceptedResponse.status).toBe(202) + const accepted = (await acceptedResponse.json()) as AcceptedTrace + expect(accepted.traceId).toBe(trace.traceId) + expect(accepted.status).toBe("accepted") + expect(accepted.contentHash).toMatch(/^[a-f0-9]{64}$/) + + const stored = await waitForTrace(deployment.url!, trace.traceId) + expect(stored).toMatchObject(trace) + + const query = new URLSearchParams({ + agent: trace.agent, + status: trace.status, + model: trace.model, + }) + const listResponse = await fetch(`${deployment.url}/v1/traces?${query}`) + expect(listResponse.ok).toBe(true) + const page = (await listResponse.json()) as TracePage + expect(page.traces.map(item => item.traceId)).toContain(trace.traceId) + }) + + it("treats an identical submission as idempotent", async () => { + const trace = { + traceId: "integration-trace-2", + agent: "operator", + status: "running", + model: "claude-haiku", + startedAt: "2026-08-26T19:00:00Z", + payload: { step: 1 }, + } + const submit = () => + fetch(`${deployment.url}/v1/traces`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(trace), + }) + + expect((await submit()).status).toBe(202) + expect((await submit()).status).toBe(202) + const stored = await waitForTrace(deployment.url!, trace.traceId) + expect(stored.payload).toEqual(trace.payload) + }) + + it("rejects invalid traces before staging them", async () => { + const response = await fetch(`${deployment.url}/v1/traces`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + traceId: "", + agent: "researcher", + status: "completed", + model: "claude-sonnet", + startedAt: "2026-08-26T18:00:00Z", + payload: {}, + }), + }) + + expect(response.status).toBe(400) + await expect(response.json()).resolves.toMatchObject({ code: "TRACE_INVALID" }) + }) +}) diff --git a/examples/slatedb-trace-store/tsconfig.json b/examples/slatedb-trace-store/tsconfig.json new file mode 100644 index 000000000..cbacaf555 --- /dev/null +++ b/examples/slatedb-trace-store/tsconfig.json @@ -0,0 +1,10 @@ +{ + "compilerOptions": { + "module": "ESNext", + "moduleResolution": "Bundler", + "target": "ES2022", + "strict": true, + "types": ["node", "vitest/globals"] + }, + "include": ["alien.ts", "tests/**/*.ts", "vitest.config.ts"] +} diff --git a/examples/slatedb-trace-store/vitest.config.ts b/examples/slatedb-trace-store/vitest.config.ts new file mode 100644 index 000000000..ce5a8df41 --- /dev/null +++ b/examples/slatedb-trace-store/vitest.config.ts @@ -0,0 +1,9 @@ +import { defineConfig } from "vitest/config" + +export default defineConfig({ + test: { + testTimeout: 60_000, + hookTimeout: 600_000, + sequence: { concurrent: false }, + }, +})