diff --git a/Cargo.lock b/Cargo.lock index cea15e3..e395a5f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -36,6 +36,15 @@ dependencies = [ "zerocopy", ] +[[package]] +name = "aho-corasick" +version = "1.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ddd31a130427c27518df266943a5308ed92d4b226cc639f5a8f1002816174301" +dependencies = [ + "memchr", +] + [[package]] name = "aligned" version = "0.4.3" @@ -133,7 +142,7 @@ dependencies = [ "num-traits", "pastey", "rayon", - "thiserror", + "thiserror 2.0.18", "v_frame", "y4m", ] @@ -382,6 +391,15 @@ dependencies = [ "parking_lot_core", ] +[[package]] +name = "deranged" +version = "0.5.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7cd812cc2bc1d69d4764bd80df88b4317eaef9e773c75226407d9bc0876b211c" +dependencies = [ + "powerfmt", +] + [[package]] name = "digest" version = "0.10.7" @@ -392,6 +410,27 @@ dependencies = [ "crypto-common", ] +[[package]] +name = "dirs" +version = "5.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "44c45a9d03d6676652bcb5e724c7e988de1acad23a711b5217ab9cbecbec2225" +dependencies = [ + "dirs-sys", +] + +[[package]] +name = "dirs-sys" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "520f05a5cbd335fae5a99ff7a6ab8627577660ee5cfd6a94a6a929b52ff0321c" +dependencies = [ + "libc", + "option-ext", + "redox_users", + "windows-sys 0.48.0", +] + [[package]] name = "either" version = "1.15.0" @@ -431,7 +470,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -551,12 +590,52 @@ version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "aa9a19cbb55df58761df49b23516a86d432839add4af60fc256da840f66ed35b" +[[package]] +name = "fs2" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9564fc758e15025b46aa6643b1b77d047d1a56a1aea6e01002ac0c7026876213" +dependencies = [ + "libc", + "winapi", +] + [[package]] name = "futures-core" version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d" +[[package]] +name = "futures-macro" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "futures-task" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "037711b3d59c33004d3856fbdc83b99d4ff37a24768fa1be9ce3538a1cde4393" + +[[package]] +name = "futures-util" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" +dependencies = [ + "futures-core", + "futures-macro", + "futures-task", + "pin-project-lite", + "slab", +] + [[package]] name = "generic-array" version = "0.14.7" @@ -802,6 +881,12 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "lazy_static" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" + [[package]] name = "leb128fmt" version = "0.1.0" @@ -836,6 +921,15 @@ version = "0.2.16" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981" +[[package]] +name = "libredox" +version = "0.1.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e02f3bb43d335493c96bf3fd3a321600bf6bd07ed34bc64118e9293bdffea46c" +dependencies = [ + "libc", +] + [[package]] name = "libsqlite3-sys" version = "0.28.0" @@ -877,6 +971,15 @@ dependencies = [ "imgref", ] +[[package]] +name = "matchers" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9" +dependencies = [ + "regex-automata", +] + [[package]] name = "matrixmultiply" version = "0.3.10" @@ -921,7 +1024,7 @@ checksum = "50b7e5b27aa02a74bac8c3f23f448f8d87ff11f92d3aac1a6ed369ee08cc56c1" dependencies = [ "libc", "wasi", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -979,6 +1082,15 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0676bb32a98c1a483ce53e500a81ad9c3d5b3f7c920c28c24e9cb0980d0b5bc8" +[[package]] +name = "nu-ansi-term" +version = "0.50.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "num" version = "0.4.3" @@ -1012,6 +1124,12 @@ dependencies = [ "num-traits", ] +[[package]] +name = "num-conv" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6673768db2d862beb9b39a78fdcb1a69439615d5794a1be50caa9bc92c81967" + [[package]] name = "num-derive" version = "0.4.2" @@ -1070,6 +1188,12 @@ version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" +[[package]] +name = "option-ext" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "04744f49eae99ab78e0d5c0b603ab218f515ea8cfe5a456d7629ad883a3b6e7d" + [[package]] name = "owned_ttf_parser" version = "0.25.1" @@ -1129,6 +1253,12 @@ dependencies = [ "miniz_oxide", ] +[[package]] +name = "powerfmt" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "439ee305def115ba05938db6eb1644ff94165c5ab5e9420d1c1bcedbba909391" + [[package]] name = "ppv-lite86" version = "0.2.21" @@ -1351,7 +1481,7 @@ dependencies = [ "rand 0.9.4", "rand_chacha 0.9.0", "simd_helpers", - "thiserror", + "thiserror 2.0.18", "v_frame", "wasm-bindgen", ] @@ -1406,6 +1536,28 @@ dependencies = [ "bitflags", ] +[[package]] +name = "redox_users" +version = "0.4.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba009ff324d1fc1b900bd1fdb31564febe58a8ccc8a6fdbb93b543d33b13ca43" +dependencies = [ + "getrandom 0.2.17", + "libredox", + "thiserror 1.0.69", +] + +[[package]] +name = "regex-automata" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e1dd4122fc1595e8162618945476892eefca7b88c52820e74af6262213cae8f" +dependencies = [ + "aho-corasick", + "memchr", + "regex-syntax", +] + [[package]] name = "regex-syntax" version = "0.8.10" @@ -1442,7 +1594,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -1551,12 +1703,53 @@ dependencies = [ "digest", ] +[[package]] +name = "sharded-slab" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6" +dependencies = [ + "lazy_static", +] + [[package]] name = "shlex" version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" +[[package]] +name = "signal-hook" +version = "0.3.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d881a16cf4426aa584979d30bd82cb33429027e42122b169753d6ef1085ed6e2" +dependencies = [ + "libc", + "signal-hook-registry", +] + +[[package]] +name = "signal-hook-registry" +version = "1.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" +dependencies = [ + "errno", + "libc", +] + +[[package]] +name = "signal-hook-tokio" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "213241f76fb1e37e27de3b6aa1b068a2c333233b59cca6634f634b80a27ecf1e" +dependencies = [ + "futures-core", + "libc", + "signal-hook", + "tokio", +] + [[package]] name = "simba" version = "0.8.1" @@ -1585,6 +1778,12 @@ dependencies = [ "quote", ] +[[package]] +name = "slab" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + [[package]] name = "smallvec" version = "1.15.1" @@ -1598,7 +1797,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e" dependencies = [ "libc", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -1607,6 +1806,12 @@ version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" +[[package]] +name = "symlink" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7973cce6668464ea31f176d85b13c7ab3bba2cb3b77a2ed26abd7801688010a" + [[package]] name = "syn" version = "2.0.117" @@ -1628,7 +1833,16 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix", - "windows-sys", + "windows-sys 0.61.2", +] + +[[package]] +name = "thiserror" +version = "1.0.69" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52" +dependencies = [ + "thiserror-impl 1.0.69", ] [[package]] @@ -1637,7 +1851,18 @@ version = "2.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4288b5bcbc7920c07a1149a35cf9590a2aa808e0bc1eafaade0b80947865fbc4" dependencies = [ - "thiserror-impl", + "thiserror-impl 2.0.18", +] + +[[package]] +name = "thiserror-impl" +version = "1.0.69" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1" +dependencies = [ + "proc-macro2", + "quote", + "syn", ] [[package]] @@ -1651,6 +1876,15 @@ dependencies = [ "syn", ] +[[package]] +name = "thread_local" +version = "1.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f60246a4944f24f6e018aa17cdeffb7818b76356965d03b07d6a9886e8962185" +dependencies = [ + "cfg-if", +] + [[package]] name = "tiff" version = "0.11.3" @@ -1665,6 +1899,37 @@ dependencies = [ "zune-jpeg", ] +[[package]] +name = "time" +version = "0.3.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "743bd48c283afc0388f9b8827b976905fb217ad9e647fae3a379a9283c4def2c" +dependencies = [ + "deranged", + "itoa", + "num-conv", + "powerfmt", + "serde_core", + "time-core", + "time-macros", +] + +[[package]] +name = "time-core" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7694e1cfe791f8d31026952abf09c69ca6f6fa4e1a1229e18988f06a04a12dca" + +[[package]] +name = "time-macros" +version = "0.2.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2e70e4c5a0e0a8a4823ad65dfe1a6930e4f4d756dcd9dd7939022b5e8c501215" +dependencies = [ + "num-conv", + "time-core", +] + [[package]] name = "tinyvec" version = "1.11.0" @@ -1690,9 +1955,10 @@ dependencies = [ "libc", "mio", "pin-project-lite", + "signal-hook-registry", "socket2", "tokio-macros", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -1739,6 +2005,19 @@ dependencies = [ "tracing-core", ] +[[package]] +name = "tracing-appender" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "050686193eb999b4bb3bc2acfa891a13da00f79734704c4b8b4ef1a10b368a3c" +dependencies = [ + "crossbeam-channel", + "symlink", + "thiserror 2.0.18", + "time", + "tracing-subscriber", +] + [[package]] name = "tracing-attributes" version = "0.1.31" @@ -1757,6 +2036,49 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" dependencies = [ "once_cell", + "valuable", +] + +[[package]] +name = "tracing-log" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3" +dependencies = [ + "log", + "once_cell", + "tracing-core", +] + +[[package]] +name = "tracing-serde" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "704b1aeb7be0d0a84fc9828cae51dab5970fee5088f83d1dd7ee6f6246fc6ff1" +dependencies = [ + "serde", + "tracing-core", +] + +[[package]] +name = "tracing-subscriber" +version = "0.3.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb7f578e5945fb242538965c2d0b04418d38ec25c79d160cd279bf0731c8d319" +dependencies = [ + "matchers", + "nu-ansi-term", + "once_cell", + "regex-automata", + "serde", + "serde_json", + "sharded-slab", + "smallvec", + "thread_local", + "tracing", + "tracing-core", + "tracing-log", + "tracing-serde", ] [[package]] @@ -1821,6 +2143,12 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "valuable" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" + [[package]] name = "vcli-capture" version = "0.0.1" @@ -1828,7 +2156,7 @@ dependencies = [ "image", "screencapturekit", "tempfile", - "thiserror", + "thiserror 2.0.18", "tracing", "vcli-core", ] @@ -1841,11 +2169,42 @@ dependencies = [ "ryu", "serde", "serde_json", - "thiserror", + "thiserror 2.0.18", "unicode-normalization", "uuid", ] +[[package]] +name = "vcli-daemon" +version = "0.0.1" +dependencies = [ + "async-trait", + "crossbeam-channel", + "dirs", + "fs2", + "futures-util", + "libc", + "serde", + "serde_json", + "signal-hook-tokio", + "tempfile", + "thiserror 2.0.18", + "tokio", + "tokio-test", + "tracing", + "tracing-appender", + "tracing-subscriber", + "uuid", + "vcli-capture", + "vcli-core", + "vcli-dsl", + "vcli-input", + "vcli-ipc", + "vcli-perception", + "vcli-runtime", + "vcli-store", +] + [[package]] name = "vcli-dsl" version = "0.0.1" @@ -1853,7 +2212,7 @@ dependencies = [ "proptest", "serde", "serde_json", - "thiserror", + "thiserror 2.0.18", "vcli-core", ] @@ -1868,7 +2227,7 @@ dependencies = [ "serde", "serde_json", "tempfile", - "thiserror", + "thiserror 2.0.18", "vcli-core", ] @@ -1881,7 +2240,7 @@ dependencies = [ "serde", "serde_json", "tempfile", - "thiserror", + "thiserror 2.0.18", "tokio", "tokio-test", "uuid", @@ -1899,7 +2258,7 @@ dependencies = [ "proptest", "serde", "serde_json", - "thiserror", + "thiserror 2.0.18", "vcli-core", ] @@ -1911,7 +2270,7 @@ dependencies = [ "proptest", "serde_json", "tempfile", - "thiserror", + "thiserror 2.0.18", "tracing", "vcli-capture", "vcli-core", @@ -1931,7 +2290,7 @@ dependencies = [ "serde_json", "sha2", "tempfile", - "thiserror", + "thiserror 2.0.18", "vcli-core", ] @@ -2075,12 +2434,43 @@ dependencies = [ "safe_arch", ] +[[package]] +name = "winapi" +version = "0.3.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419" +dependencies = [ + "winapi-i686-pc-windows-gnu", + "winapi-x86_64-pc-windows-gnu", +] + +[[package]] +name = "winapi-i686-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6" + +[[package]] +name = "winapi-x86_64-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" + [[package]] name = "windows-link" version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" +[[package]] +name = "windows-sys" +version = "0.48.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "677d2418bec65e3338edb076e806bc1ec15693c5d0104683f2efe857f61056a9" +dependencies = [ + "windows-targets", +] + [[package]] name = "windows-sys" version = "0.61.2" @@ -2090,6 +2480,63 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-targets" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a2fa6e2155d7247be68c096456083145c183cbbbc2764150dda45a87197940c" +dependencies = [ + "windows_aarch64_gnullvm", + "windows_aarch64_msvc", + "windows_i686_gnu", + "windows_i686_msvc", + "windows_x86_64_gnu", + "windows_x86_64_gnullvm", + "windows_x86_64_msvc", +] + +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2b38e32f0abccf9987a4e3079dfb67dcd799fb61361e53e2882c3cbaf0d905d8" + +[[package]] +name = "windows_aarch64_msvc" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc35310971f3b2dbbf3f0690a219f40e2d9afcf64f9ab7cc1be722937c26b4bc" + +[[package]] +name = "windows_i686_gnu" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a75915e7def60c94dcef72200b9a8e58e5091744960da64ec734a6c6e9b3743e" + +[[package]] +name = "windows_i686_msvc" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f55c233f70c4b27f66c523580f78f1004e8b5a8b659e05a4eb49d4166cca406" + +[[package]] +name = "windows_x86_64_gnu" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53d40abd2583d23e4718fddf1ebec84dbff8381c07cae67ff7768bbf19c6718e" + +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b7b52767868a23d5bab768e390dc5f5c55825b6d30b86c844ff2dc7414044cc" + +[[package]] +name = "windows_x86_64_msvc" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed94fce61571a4006852b7389a063ab983c02eb1bb37b47f8272ce92d06d9538" + [[package]] name = "wit-bindgen" version = "0.51.0" diff --git a/Cargo.toml b/Cargo.toml index bda3c0f..4c6bcad 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,6 +3,7 @@ resolver = "2" members = [ "crates/vcli-core", "crates/vcli-capture", + "crates/vcli-daemon", "crates/vcli-dsl", "crates/vcli-input", "crates/vcli-ipc", @@ -29,10 +30,18 @@ ryu = "1" unicode-normalization = "0.1" # Added by vcli-ipc lane -tokio = { version = "1", features = ["net", "io-util", "macros", "rt-multi-thread", "sync", "time"] } +tokio = { version = "1", features = ["net", "io-util", "macros", "rt-multi-thread", "sync", "time", "signal"] } async-trait = "0.1" futures-core = "0.3" +# Added by vcli-daemon lane +tracing-subscriber = { version = "0.3", features = ["fmt", "env-filter", "json"] } +tracing-appender = "0.2" +fs2 = "0.4" +dirs = "5" +signal-hook-tokio = { version = "0.3", features = ["futures-v0_3"] } +futures-util = "0.3" + rusqlite = { version = "0.31", features = ["bundled"] } sha2 = "0.10" hex = "0.4" diff --git a/crates/vcli-core/src/clock.rs b/crates/vcli-core/src/clock.rs index 5cf6c4f..046fc25 100644 --- a/crates/vcli-core/src/clock.rs +++ b/crates/vcli-core/src/clock.rs @@ -109,6 +109,25 @@ impl Clock for TestClock { } } +/// Convenience for callers that just want wall-clock ms and do not hold a +/// `Clock` handle. Equivalent to `SystemClock::new().unix_ms()`. Non-test code +/// outside the scheduler uses this for event timestamps. +#[must_use] +pub fn now_unix_ms() -> UnixMs { + SystemClock::new().unix_ms() +} + +#[cfg(test)] +mod now_unix_ms_tests { + use super::*; + + #[test] + fn now_unix_ms_is_positive_and_recent() { + let t = now_unix_ms(); + assert!(t > 1_700_000_000_000, "got {t}"); + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/vcli-core/src/lib.rs b/crates/vcli-core/src/lib.rs index 6d617ad..231a6af 100644 --- a/crates/vcli-core/src/lib.rs +++ b/crates/vcli-core/src/lib.rs @@ -21,7 +21,7 @@ pub use frame::{Frame, FrameFormat}; pub mod clock; -pub use clock::{Clock, SystemClock, TestClock, UnixMs}; +pub use clock::{now_unix_ms, Clock, SystemClock, TestClock, UnixMs}; pub mod region; diff --git a/crates/vcli-daemon/Cargo.toml b/crates/vcli-daemon/Cargo.toml new file mode 100644 index 0000000..a9eaeb5 --- /dev/null +++ b/crates/vcli-daemon/Cargo.toml @@ -0,0 +1,47 @@ +[package] +name = "vcli-daemon" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +repository.workspace = true +authors.workspace = true +description = "The vcli daemon binary — tokio reactor + scheduler bridge + socket server." + +[lib] +path = "src/lib.rs" + +[[bin]] +name = "vcli-daemon" +path = "src/bin/vcli-daemon.rs" + +[dependencies] +vcli-core = { path = "../vcli-core" } +vcli-dsl = { path = "../vcli-dsl" } +vcli-store = { path = "../vcli-store" } +vcli-ipc = { path = "../vcli-ipc" } +vcli-capture = { path = "../vcli-capture" } +vcli-input = { path = "../vcli-input" } +vcli-perception = { path = "../vcli-perception" } +vcli-runtime = { path = "../vcli-runtime" } + +serde = { workspace = true } +serde_json = { workspace = true } +thiserror = { workspace = true } +async-trait = { workspace = true } +tokio = { workspace = true } +tracing = { workspace = true } +tracing-subscriber = { workspace = true } +tracing-appender = { workspace = true } +crossbeam-channel = { workspace = true } +fs2 = { workspace = true } +dirs = { workspace = true } +futures-util = { workspace = true } +signal-hook-tokio = { workspace = true } +uuid = { workspace = true } +libc = { workspace = true } + +[dev-dependencies] +tempfile = { workspace = true } +tokio-test = { workspace = true } +tokio = { workspace = true, features = ["test-util"] } diff --git a/crates/vcli-daemon/build.rs b/crates/vcli-daemon/build.rs new file mode 100644 index 0000000..885803f --- /dev/null +++ b/crates/vcli-daemon/build.rs @@ -0,0 +1,13 @@ +fn main() { + #[cfg(target_os = "macos")] + { + // vcli-daemon pulls in vcli-capture → screencapturekit transitively. + // screencapturekit's Swift bridge references libswift_Concurrency.dylib. + // vcli-capture sets this rpath for its own artifacts; we repeat it here + // so vcli-daemon's test + binary output can load the Swift dylib too. + let xcode_swift55_path = + "/Applications/Xcode.app/Contents/Developer/Toolchains/XcodeDefault.xctoolchain\ + /usr/lib/swift-5.5/macosx"; + println!("cargo:rustc-link-arg=-Wl,-rpath,{xcode_swift55_path}"); + } +} diff --git a/crates/vcli-daemon/src/bin/vcli-daemon.rs b/crates/vcli-daemon/src/bin/vcli-daemon.rs new file mode 100644 index 0000000..ed411e0 --- /dev/null +++ b/crates/vcli-daemon/src/bin/vcli-daemon.rs @@ -0,0 +1,73 @@ +//! vcli-daemon entrypoint. Minimal argv surface: +//! * `--version` → print crate version, exit 0 +//! * `--help` → print usage, exit 0 +//! +//! All real work lives in `vcli_daemon::run_foreground`. + +use std::process::ExitCode; + +use vcli_daemon::{run_foreground, Config, DaemonError, RuntimeBackends}; + +fn main() -> ExitCode { + let args: Vec = std::env::args().collect(); + if args.iter().any(|a| a == "--version" || a == "-V") { + println!("vcli-daemon {}", env!("CARGO_PKG_VERSION")); + return ExitCode::SUCCESS; + } + if args.iter().any(|a| a == "--help" || a == "-h") { + println!("usage: vcli-daemon"); + println!(" no flags — runs in the foreground until SIGTERM/SIGINT"); + return ExitCode::SUCCESS; + } + + let rt = match tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + { + Ok(r) => r, + Err(e) => { + eprintln!("vcli-daemon: failed to build tokio runtime: {e}"); + return ExitCode::from(1); + } + }; + + let cfg = match Config::from_platform_defaults() { + Ok(c) => c, + Err(e) => { + eprintln!("vcli-daemon: {e}"); + return ExitCode::from(1); + } + }; + + let factory: vcli_daemon::RuntimeFactory = Box::new(default_runtime_factory); + + match rt.block_on(run_foreground(cfg, factory)) { + Ok(()) => ExitCode::SUCCESS, + Err(DaemonError::AlreadyRunning { pid, .. }) => { + eprintln!("vcli-daemon: already running (pid {pid})"); + ExitCode::from(1) + } + Err(e) => { + eprintln!("vcli-daemon: {e}"); + ExitCode::from(1) + } + } +} + +/// Real-backend factory used in production. For v0 the daemon binary falls +/// back to mock backends on every platform — a future lane wires the real +/// macOS capture + input. +fn default_runtime_factory() -> Result { + let capture: Box = Box::new(vcli_capture::MockCapture::empty()); + let input: std::sync::Arc = + std::sync::Arc::new(vcli_input::MockInputSink::new()); + let perception = vcli_perception::Perception::default(); + let clock: std::sync::Arc = + std::sync::Arc::new(vcli_runtime::SystemRuntimeClock::new()); + Ok(RuntimeBackends { + capture, + input, + perception, + clock, + }) +} diff --git a/crates/vcli-daemon/src/bridge.rs b/crates/vcli-daemon/src/bridge.rs new file mode 100644 index 0000000..ddc1c03 --- /dev/null +++ b/crates/vcli-daemon/src/bridge.rs @@ -0,0 +1,97 @@ +//! Tokio ↔ scheduler bridge. +//! +//! The scheduler thread is pure sync; the tokio reactor is async. We route +//! data between them via two `crossbeam_channel`s: +//! * `cmd_tx` (tokio → sched): bounded, backpressure-free; handlers send +//! [`SchedulerCommand`] values. +//! * `event_rx` (sched → tokio): unbounded. A dedicated tokio task drains +//! it and pushes each [`vcli_core::Event`] through the persistence sink +//! and then into the [`tokio::sync::broadcast`] channel for connected +//! IPC subscribers. +//! +//! `event_tx` (broadcast side) is cloned per IPC subscription; each streaming +//! handler subscribes, reads frames, and hangs up when the client drops. The +//! capacity of the broadcast (Decision 1.7) is 1024; overflowing clients +//! receive a `stream.dropped` notification inside the handler. + +use crossbeam_channel::{bounded, unbounded, Receiver, Sender}; +use tokio::sync::broadcast; +use vcli_core::Event; + +pub use vcli_runtime::SchedulerCommand; + +/// Capacity of the broadcast channel fanned out to IPC subscribers. +pub const EVENT_BROADCAST_CAPACITY: usize = 1024; + +/// Capacity of the command channel tokio → scheduler. +pub const CMD_CAPACITY: usize = 256; + +/// All the channel endpoints together. Cheap to clone; references passed into +/// the handler and startup modules. +#[derive(Clone)] +pub struct CommandChannel { + /// Tokio → scheduler commands. + pub cmd_tx: Sender, + /// Broadcast fanning out persisted events to IPC subscribers. + pub event_tx: broadcast::Sender, +} + +/// Construct a matched set of endpoints. The `cmd_rx` and `event_rx` returned +/// are handed to the scheduler thread + the event-pump task respectively. +#[must_use] +pub fn new_channels() -> ( + CommandChannel, + Receiver, + Receiver, + Sender, +) { + let (cmd_tx, cmd_rx) = bounded::(CMD_CAPACITY); + let (sched_event_tx, event_rx) = unbounded::(); + let (bcast_tx, _) = broadcast::channel::(EVENT_BROADCAST_CAPACITY); + ( + CommandChannel { + cmd_tx, + event_tx: bcast_tx, + }, + cmd_rx, + event_rx, + sched_event_tx, + ) +} + +#[cfg(test)] +mod tests { + use super::*; + use vcli_core::{EventData, ProgramId}; + + #[test] + fn command_channel_is_cloneable_and_reaches_scheduler() { + let (chan, cmd_rx, _event_rx, _sched_event_tx) = new_channels(); + let chan2 = chan.clone(); + chan2 + .cmd_tx + .send(SchedulerCommand::Cancel { + program_id: ProgramId::new(), + reason: "test".into(), + }) + .unwrap(); + match cmd_rx.recv().unwrap() { + SchedulerCommand::Cancel { .. } => {} + other => panic!("unexpected cmd: {other:?}"), + } + } + + #[tokio::test] + async fn broadcast_fanout_reaches_multiple_subscribers() { + let (chan, _cmd_rx, _event_rx, _sched_event_tx) = new_channels(); + let mut rx1 = chan.event_tx.subscribe(); + let mut rx2 = chan.event_tx.subscribe(); + let ev = Event { + at: 1, + data: EventData::DaemonStopped, + }; + chan.event_tx.send(ev.clone()).unwrap(); + assert_eq!(rx1.recv().await.unwrap(), ev); + assert_eq!(rx2.recv().await.unwrap(), ev); + } +} diff --git a/crates/vcli-daemon/src/config.rs b/crates/vcli-daemon/src/config.rs new file mode 100644 index 0000000..e9ec09e --- /dev/null +++ b/crates/vcli-daemon/src/config.rs @@ -0,0 +1,165 @@ +//! Resolved on-disk layout. One `Config` value, produced at startup, threaded +//! everywhere that needs a path. Every caller is driven by the struct — nothing +//! touches `$HOME` directly so tests stay hermetic. +//! +//! Layout (spec §Persistence → Data layout): +//! `data_root` = `~/Library/Application Support/vcli` (macOS) +//! | `~/.local/share/vcli` (Linux / XDG) +//! db = `/vcli.db` +//! pidfile = `/daemon.pid` +//! socket = `vcli_ipc::default_socket_path()` +//! `log_dir` = `~/Library/Logs/vcli` (macOS) +//! | `~/.cache/vcli/logs` (Linux / XDG) +//! `log_file` = `/daemon.log` (rotated daily, 7-day retention) + +use std::path::{Path, PathBuf}; + +use vcli_ipc::SocketPath; + +use crate::error::{DaemonError, DaemonResult}; + +/// Every path the daemon needs to know. Built once at startup. +#[derive(Debug, Clone)] +pub struct Config { + /// Directory holding `vcli.db`, `daemon.pid`, and the assets tree. + pub data_root: PathBuf, + /// Resolved socket path + its resolution origin (for `vcli health`). + pub socket: SocketPath, + /// Directory for rotating log files. + pub log_dir: PathBuf, +} + +impl Config { + /// Resolve from the platform defaults. macOS uses Apple locations; Linux uses XDG. + /// + /// # Errors + /// [`DaemonError::Paths`] if `$HOME` (or its platform equivalent) cannot be resolved. + pub fn from_platform_defaults() -> DaemonResult { + let data_root = platform_data_root()?; + let log_dir = platform_log_dir()?; + let socket = vcli_ipc::default_socket_path() + .map_err(|e| DaemonError::Paths(format!("socket path: {e}")))?; + Ok(Self { + data_root, + socket, + log_dir, + }) + } + + /// Build a config rooted at an explicit directory — used by tests and + /// by anyone wiring a non-default install. + #[must_use] + pub fn with_roots( + data_root: impl Into, + log_dir: impl Into, + socket: SocketPath, + ) -> Self { + Self { + data_root: data_root.into(), + socket, + log_dir: log_dir.into(), + } + } + + /// Absolute pidfile path (`/daemon.pid`). + #[must_use] + pub fn pidfile_path(&self) -> PathBuf { + self.data_root.join("daemon.pid") + } + + /// Absolute log file path (`/daemon.log`). + #[must_use] + pub fn log_file_path(&self) -> PathBuf { + self.log_dir.join("daemon.log") + } +} + +#[cfg(target_os = "macos")] +fn platform_data_root() -> DaemonResult { + let home = dirs::home_dir().ok_or_else(|| DaemonError::Paths("no home dir".into()))?; + Ok(home + .join("Library") + .join("Application Support") + .join("vcli")) +} + +#[cfg(target_os = "macos")] +fn platform_log_dir() -> DaemonResult { + let home = dirs::home_dir().ok_or_else(|| DaemonError::Paths("no home dir".into()))?; + Ok(home.join("Library").join("Logs").join("vcli")) +} + +#[cfg(not(target_os = "macos"))] +fn platform_data_root() -> DaemonResult { + let d = dirs::data_local_dir().ok_or_else(|| DaemonError::Paths("no XDG_DATA_HOME".into()))?; + Ok(d.join("vcli")) +} + +#[cfg(not(target_os = "macos"))] +fn platform_log_dir() -> DaemonResult { + let d = dirs::cache_dir().ok_or_else(|| DaemonError::Paths("no XDG_CACHE_HOME".into()))?; + Ok(d.join("vcli").join("logs")) +} + +/// Ensure every directory in `cfg` exists. Idempotent. +/// +/// # Errors +/// IO errors from `fs::create_dir_all`. +pub fn ensure_dirs(cfg: &Config) -> DaemonResult<()> { + mkdir_p(&cfg.data_root)?; + mkdir_p(&cfg.log_dir)?; + Ok(()) +} + +fn mkdir_p(p: &Path) -> DaemonResult<()> { + std::fs::create_dir_all(p).map_err(DaemonError::Io) +} + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::tempdir; + use vcli_ipc::socket_path::SocketPathOrigin; + + fn fake_sock(p: &Path) -> SocketPath { + SocketPath { + path: p.to_path_buf(), + origin: SocketPathOrigin::Override, + } + } + + #[test] + fn with_roots_builds_a_usable_config() { + let d = tempdir().unwrap(); + let cfg = Config::with_roots( + d.path().join("data"), + d.path().join("logs"), + fake_sock(&d.path().join("vcli.sock")), + ); + assert_eq!(cfg.pidfile_path(), d.path().join("data").join("daemon.pid")); + assert_eq!( + cfg.log_file_path(), + d.path().join("logs").join("daemon.log") + ); + } + + #[test] + fn ensure_dirs_creates_nested_paths() { + let d = tempdir().unwrap(); + let cfg = Config::with_roots( + d.path().join("a").join("b"), + d.path().join("c").join("d"), + fake_sock(&d.path().join("vcli.sock")), + ); + ensure_dirs(&cfg).unwrap(); + assert!(cfg.data_root.is_dir()); + assert!(cfg.log_dir.is_dir()); + } + + #[test] + #[cfg(target_os = "macos")] + fn platform_defaults_use_library_on_macos() { + let got = platform_data_root().unwrap(); + assert!(got.ends_with("Library/Application Support/vcli"), "{got:?}"); + } +} diff --git a/crates/vcli-daemon/src/error.rs b/crates/vcli-daemon/src/error.rs new file mode 100644 index 0000000..b152706 --- /dev/null +++ b/crates/vcli-daemon/src/error.rs @@ -0,0 +1,102 @@ +//! Top-level daemon error. Maps to `vcli_core::ErrorCode` at the IPC boundary. + +use std::io; +use std::path::PathBuf; + +use thiserror::Error; +use vcli_core::ErrorCode; + +/// Convenience alias. +pub type DaemonResult = Result; + +/// Errors the daemon produces outside the per-request handler path. +#[derive(Debug, Error)] +pub enum DaemonError { + /// PID file already held by a live process. + #[error("another vcli-daemon is already running (pid {pid}, lockfile {path})")] + AlreadyRunning { + /// Owning PID. + pid: u32, + /// Absolute pidfile path. + path: PathBuf, + }, + + /// Could not acquire / write the pidfile. + #[error("pidfile {path}: {source}")] + Pidfile { + /// Absolute path. + path: PathBuf, + /// Underlying IO cause. + #[source] + source: io::Error, + }, + + /// Could not resolve the daemon's data root, socket path, or log dir. + #[error("path resolution: {0}")] + Paths(String), + + /// Tracing subscriber failed to install. + #[error("logging init: {0}")] + Logging(String), + + /// Store open / migrate failed. + #[error("store: {0}")] + Store(#[from] vcli_store::StoreError), + + /// IPC transport setup / serve failure. + #[error("ipc: {0}")] + Ipc(#[from] vcli_ipc::IpcError), + + /// Generic IO error during startup / shutdown. + #[error("io: {0}")] + Io(#[from] io::Error), + + /// DSL validation of a submitted program failed before scheduler touched it. + #[error("invalid program: {0}")] + InvalidProgram(String), +} + +impl DaemonError { + /// Stable error code for IPC responses. + #[must_use] + pub fn code(&self) -> ErrorCode { + match self { + Self::InvalidProgram(_) => ErrorCode::InvalidProgram, + Self::AlreadyRunning { .. } + | Self::Pidfile { .. } + | Self::Paths(_) + | Self::Logging(_) + | Self::Store(_) + | Self::Ipc(_) + | Self::Io(_) => ErrorCode::Internal, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn invalid_program_maps_to_invalid_program_code() { + let e = DaemonError::InvalidProgram("bad path".into()); + assert_eq!(e.code(), ErrorCode::InvalidProgram); + } + + #[test] + fn paths_maps_to_internal() { + let e = DaemonError::Paths("no home".into()); + assert_eq!(e.code(), ErrorCode::Internal); + } + + #[test] + fn already_running_display_contains_pid_and_path() { + let e = DaemonError::AlreadyRunning { + pid: 42, + path: "/tmp/x.pid".into(), + }; + let s = e.to_string(); + assert!(s.contains("42"), "{s}"); + assert!(s.contains("/tmp/x.pid"), "{s}"); + } +} diff --git a/crates/vcli-daemon/src/handler.rs b/crates/vcli-daemon/src/handler.rs new file mode 100644 index 0000000..b497c0a --- /dev/null +++ b/crates/vcli-daemon/src/handler.rs @@ -0,0 +1,790 @@ +//! `DaemonHandler` implements `vcli_ipc::Handler`. One `impl` method per +//! [`RequestOp`] variant. The handler owns clones of the bridge endpoints and +//! an `Arc>` (sync — reached via `spawn_blocking` when we need to +//! call `SQLite` from inside an async method). + +use std::sync::{Arc, Mutex}; +use std::time::Instant; + +use async_trait::async_trait; +use tokio::sync::{broadcast, oneshot}; +use tracing::{debug, error}; +use vcli_core::{ErrorCode, ErrorPayload, Event, ProgramId}; +use vcli_ipc::{ + Handler, IpcResult, RequestId, RequestOp, Response, StreamFrame, StreamKind, StreamSender, +}; +use vcli_store::Store; + +use crate::bridge::{CommandChannel, SchedulerCommand}; + +/// Shared boundary between the tokio handler and the scheduler/store. +#[derive(Clone)] +pub struct DaemonHandler { + /// Async-side store handle. All DB ops run inside `spawn_blocking`. + pub store: Arc>, + /// Command + broadcast endpoints. + pub bridge: CommandChannel, + /// Wall-clock start time (for `health.uptime_ms`). + pub started_at: Instant, + /// Graceful-shutdown trigger, set when a client sends `Shutdown`. The + /// `run_foreground` task awaits the receiver. + pub shutdown_tx: Arc>>>, +} + +impl DaemonHandler { + /// Fire the shutdown oneshot if still armed. Idempotent. + /// + /// # Panics + /// Panics if the internal mutex is poisoned. + pub fn trigger_shutdown(&self) { + if let Some(tx) = self.shutdown_tx.lock().unwrap().take() { + let _ = tx.send(()); + } + } + + async fn handle_submit(&self, id: RequestId, program_json: serde_json::Value) -> Response { + let program = match vcli_dsl::validate_value(&program_json) { + Ok(v) => v.program, + Err(e) => { + return Response::err(id, e.to_payload()); + } + }; + + let program_id = program.id.unwrap_or_else(ProgramId::new); + let name = program.name.clone(); + let canonical_bytes = match vcli_core::canonicalize(&program_json) { + Ok(b) => b, + Err(e) => { + return Response::err( + id, + ErrorPayload::simple(ErrorCode::Internal, format!("canonicalize: {e}")), + ); + } + }; + let canonical = String::from_utf8(canonical_bytes) + .expect("canonicalize produces UTF-8 by construction"); + + let store = self.store.clone(); + let pid = program_id; + let submitted_at = vcli_core::clock::now_unix_ms(); + let name_for_insert = name.clone(); + let canonical_str = canonical.clone(); + let insert_result = tokio::task::spawn_blocking(move || { + let mut s = store.lock().unwrap(); + s.insert_program(&vcli_store::NewProgram { + id: pid, + name: &name_for_insert, + source_json: &canonical_str, + state: vcli_core::ProgramState::Pending, + submitted_at, + labels_json: "{}", + }) + }) + .await; + let insert = match insert_result { + Ok(v) => v, + Err(e) => { + error!(error = %e, "submit join error"); + return Response::err( + id, + ErrorPayload::simple(ErrorCode::Internal, format!("{e}")), + ); + } + }; + if let Err(e) = insert { + return Response::err( + id, + ErrorPayload::simple(ErrorCode::Internal, format!("{e}")), + ); + } + + if let Err(e) = self.bridge.cmd_tx.send(SchedulerCommand::SubmitValidated { + program_id: pid, + program, + }) { + error!(error = %e, "cmd_tx full"); + return Response::err( + id, + ErrorPayload::simple(ErrorCode::DaemonBusy, "scheduler command queue full"), + ); + } + + Response::ok( + id, + serde_json::json!({ "program_id": pid.to_string(), "name": name }), + ) + } + + async fn stream_events( + &self, + id: RequestId, + filter_program: Option, + follow: bool, + tx: StreamSender, + ) -> IpcResult<()> { + let mut rx = self.bridge.event_tx.subscribe(); + if !follow { + let store = self.store.clone(); + let history = tokio::task::spawn_blocking(move || { + let s = store.lock().unwrap(); + s.stream_events(0, 10_000) + }) + .await + .ok() + .and_then(Result::ok) + .unwrap_or_default(); + for row in history { + if filter_program.is_some_and(|p| p != row.program_id) { + continue; + } + if let Ok(ev) = serde_json::from_str::(&row.data_json) { + let frame = StreamFrame::event(id, ev); + if tx.send(frame).await.is_err() { + return Ok(()); + } + } + } + return Ok(()); + } + loop { + match rx.recv().await { + Ok(ev) => { + if let Some(p) = filter_program { + if crate::persist::program_id_of(&ev.data) != Some(p) { + continue; + } + } + if tx.send(StreamFrame::event(id, ev)).await.is_err() { + return Ok(()); + } + } + Err(broadcast::error::RecvError::Lagged(n)) => { + let now = vcli_core::clock::now_unix_ms(); + let dropped = StreamFrame::event( + id, + Event { + at: now, + data: vcli_core::EventData::StreamDropped { + count: u32::try_from(n).unwrap_or(u32::MAX), + since: now, + }, + }, + ); + if tx.send(dropped).await.is_err() { + return Ok(()); + } + } + Err(broadcast::error::RecvError::Closed) => return Ok(()), + } + } + } + + async fn handle_gc(&self, id: RequestId) -> Response { + let store = self.store.clone(); + let now = vcli_core::clock::now_unix_ms(); + let cutoff = now - i64::from(vcli_store::RETENTION_DAYS) * 24 * 60 * 60 * 1000; + let report = tokio::task::spawn_blocking(move || { + let mut s = store.lock().unwrap(); + s.gc_all(cutoff) + }) + .await; + match report { + Ok(Ok(r)) => Response::ok( + id, + serde_json::json!({ + "programs_deleted": r.programs_deleted, + "assets_deleted": r.assets_deleted, + "blobs_deleted": r.blobs_deleted, + "orphan_blobs_deleted": r.orphan_blobs_deleted, + }), + ), + Ok(Err(e)) => Response::err( + id, + ErrorPayload::simple(ErrorCode::Internal, format!("{e}")), + ), + Err(e) => Response::err( + id, + ErrorPayload::simple(ErrorCode::Internal, format!("{e}")), + ), + } + } + + async fn handle_resume(&self, id: RequestId, pid: ProgramId, from_start: bool) -> Response { + let store = self.store.clone(); + let now_ms = vcli_core::clock::now_unix_ms(); + let resume_result = tokio::task::spawn_blocking(move || { + let mut s = store.lock().unwrap(); + let outcome = s.resume_program(pid, from_start, now_ms)?; + let row = s.get_program(pid)?; + let value: serde_json::Value = serde_json::from_str(&row.source_json)?; + let program = vcli_dsl::validate_value(&value) + .map(|v| v.program) + .map_err(|e| vcli_store::StoreError::Io { + path: "".into(), + source: std::io::Error::new(std::io::ErrorKind::InvalidData, format!("{e}")), + })?; + Ok::<_, vcli_store::StoreError>((outcome, program)) + }) + .await; + + match resume_result { + Ok(Ok((out, program))) => { + let _ = self.bridge.cmd_tx.send(SchedulerCommand::ResumeRunning { + program_id: pid, + from_step: out.from_step, + program, + }); + Response::ok( + id, + serde_json::json!({ "program_id": pid.to_string(), "from_step": out.from_step }), + ) + } + Ok(Err(vcli_store::StoreError::NotResumable(m))) => { + Response::err(id, ErrorPayload::simple(ErrorCode::NotResumable, m)) + } + Ok(Err(vcli_store::StoreError::UnknownProgram(_))) => Response::err( + id, + ErrorPayload::simple(ErrorCode::UnknownProgram, "not found"), + ), + Ok(Err(e)) => Response::err( + id, + ErrorPayload::simple(ErrorCode::Internal, format!("{e}")), + ), + Err(e) => Response::err( + id, + ErrorPayload::simple(ErrorCode::Internal, format!("{e}")), + ), + } + } + + fn handle_cancel(&self, id: RequestId, pid: ProgramId) -> Response { + match self.bridge.cmd_tx.send(SchedulerCommand::Cancel { + program_id: pid, + reason: "user".into(), + }) { + Ok(()) => Response::ok(id, serde_json::json!({ "program_id": pid.to_string() })), + Err(_) => Response::err( + id, + ErrorPayload::simple(ErrorCode::DaemonBusy, "cmd queue full"), + ), + } + } + + fn handle_start(&self, id: RequestId, pid: ProgramId) -> Response { + match self + .bridge + .cmd_tx + .send(SchedulerCommand::Start { program_id: pid }) + { + Ok(()) => Response::ok(id, serde_json::json!({ "program_id": pid.to_string() })), + Err(_) => Response::err( + id, + ErrorPayload::simple(ErrorCode::DaemonBusy, "cmd queue full"), + ), + } + } + + async fn handle_list(&self, id: RequestId, state: Option) -> Response { + let filter = state.and_then(|s| s.parse::().ok()); + let store = self.store.clone(); + let rows = tokio::task::spawn_blocking(move || { + let s = store.lock().unwrap(); + s.list_programs(filter) + }) + .await + .unwrap_or_else(|e| { + Err(vcli_store::StoreError::Io { + path: "".into(), + source: std::io::Error::other(format!("{e}")), + }) + }); + match rows { + Ok(rows) => { + let items: Vec<_> = rows + .into_iter() + .map(|r| { + serde_json::json!({ + "id": r.id.to_string(), + "name": r.name, + "state": r.state.as_str(), + "submitted_at": r.submitted_at, + }) + }) + .collect(); + Response::ok(id, serde_json::json!({ "items": items })) + } + Err(e) => Response::err( + id, + ErrorPayload::simple(ErrorCode::Internal, format!("{e}")), + ), + } + } + + async fn handle_status(&self, id: RequestId, program_id: ProgramId) -> Response { + let store = self.store.clone(); + let row = tokio::task::spawn_blocking(move || { + let s = store.lock().unwrap(); + s.get_program(program_id) + }) + .await; + match row { + Ok(Ok(r)) => Response::ok( + id, + serde_json::json!({ + "id": r.id.to_string(), + "name": r.name, + "state": r.state.as_str(), + "body_cursor": r.body_cursor, + "submitted_at": r.submitted_at, + "started_at": r.started_at, + "finished_at": r.finished_at, + }), + ), + Ok(Err(vcli_store::StoreError::UnknownProgram(_))) => Response::err( + id, + ErrorPayload::simple(ErrorCode::UnknownProgram, "not found"), + ), + Ok(Err(e)) => Response::err( + id, + ErrorPayload::simple(ErrorCode::Internal, format!("{e}")), + ), + Err(e) => Response::err( + id, + ErrorPayload::simple(ErrorCode::Internal, format!("{e}")), + ), + } + } + + fn handle_health(&self, id: RequestId) -> Response { + let uptime_ms = u64::try_from(self.started_at.elapsed().as_millis()).unwrap_or(u64::MAX); + Response::ok( + id, + serde_json::json!({ + "version": env!("CARGO_PKG_VERSION"), + "uptime_ms": uptime_ms, + "socket_origin": "resolved", + }), + ) + } + + fn handle_shutdown(&self, id: RequestId) -> Response { + self.trigger_shutdown(); + let _ = self.bridge.cmd_tx.send(SchedulerCommand::Shutdown); + Response::ok(id, serde_json::json!({ "bye": true })) + } +} + +#[async_trait] +impl Handler for DaemonHandler { + async fn handle(&self, id: RequestId, op: RequestOp) -> IpcResult { + let resp = match op { + RequestOp::Health => self.handle_health(id), + RequestOp::Shutdown => self.handle_shutdown(id), + RequestOp::Submit { program } => self.handle_submit(id, program).await, + RequestOp::List { state } => self.handle_list(id, state).await, + RequestOp::Status { program_id } => self.handle_status(id, program_id).await, + RequestOp::Cancel { program_id } => self.handle_cancel(id, program_id), + RequestOp::Start { program_id } => self.handle_start(id, program_id), + RequestOp::Resume { + program_id, + from_start, + } => self.handle_resume(id, program_id, from_start).await, + RequestOp::Gc => self.handle_gc(id).await, + other => Response::err( + id, + ErrorPayload::simple(ErrorCode::Internal, format!("op not yet wired: {other:?}")), + ), + }; + Ok(resp) + } + + async fn handle_stream(&self, id: RequestId, op: RequestOp, tx: StreamSender) -> IpcResult<()> { + match op { + RequestOp::Events { follow } => self.stream_events(id, None, follow, tx).await, + RequestOp::Logs { program_id, follow } => { + self.stream_events(id, Some(program_id), follow, tx).await + } + RequestOp::Trace { program_id: _ } => { + // v0 minimum: empty trace; server writes end_of_stream on return. + let _ = tx; + Ok(()) + } + other => { + debug!("stream op {other:?} not supported"); + let _ = tx + .send(StreamFrame::end_of_stream(id, StreamKind::Events)) + .await; + Ok(()) + } + } + } +} + +#[cfg(test)] +pub(crate) mod test_support { + use super::*; + use tempfile::TempDir; + + /// Bundle returned by `fresh_handler` that keeps the tempdir alive and + /// exposes a few handles follow-up tasks consume directly. + pub struct Fixture { + // `dir` is held purely for Drop — the TempDir must outlive the Store. + // Why: `#[allow(dead_code)]` — required storage for the RAII tempdir + // even though no test reads the field. + #[allow(dead_code)] + pub dir: TempDir, + pub handler: DaemonHandler, + pub shutdown_rx: oneshot::Receiver<()>, + pub cmd_rx: crossbeam_channel::Receiver, + } + + pub fn fresh_handler() -> Fixture { + let dir = tempfile::tempdir().unwrap(); + let (store, _) = Store::open(dir.path()).unwrap(); + let (bridge, cmd_rx, _event_rx, _sched_event_tx) = crate::bridge::new_channels(); + let (stx, srx) = oneshot::channel(); + let handler = DaemonHandler { + store: Arc::new(Mutex::new(store)), + bridge, + started_at: Instant::now(), + shutdown_tx: Arc::new(Mutex::new(Some(stx))), + }; + Fixture { + dir, + handler, + shutdown_rx: srx, + cmd_rx, + } + } +} + +#[cfg(test)] +mod tests { + use super::test_support::fresh_handler; + use super::*; + + #[tokio::test] + async fn health_returns_version_and_uptime() { + let f = fresh_handler(); + let id = RequestId::new(); + let resp = f.handler.handle(id, RequestOp::Health).await.unwrap(); + let body = serde_json::to_value(&resp).unwrap(); + assert_eq!(body["id"], id.to_string()); + assert_eq!(body["ok"], true); + assert!(body["result"]["version"].as_str().is_some()); + assert!(body["result"]["uptime_ms"].as_u64().is_some()); + } + + #[tokio::test] + async fn gc_returns_report_shape() { + let f = fresh_handler(); + let resp = f + .handler + .handle(RequestId::new(), RequestOp::Gc) + .await + .unwrap(); + let body = serde_json::to_value(&resp).unwrap(); + assert_eq!(body["ok"], true); + assert!(body["result"]["programs_deleted"].as_u64().is_some()); + assert!(body["result"]["assets_deleted"].as_u64().is_some()); + } + + #[tokio::test] + async fn events_stream_drains_history_then_closes_when_not_following() { + use tokio::sync::mpsc; + let f = fresh_handler(); + let pid = ProgramId::new(); + { + let mut s = f.handler.store.lock().unwrap(); + s.insert_program(&vcli_store::NewProgram { + id: pid, + name: "x", + source_json: "{}", + state: vcli_core::ProgramState::Pending, + submitted_at: 0, + labels_json: "{}", + }) + .unwrap(); + let ev = Event { + at: 7, + data: vcli_core::EventData::ProgramCompleted { + program_id: pid, + emit: None, + }, + }; + s.append_event(pid, &ev).unwrap(); + } + + let (tx, mut rx) = mpsc::channel::(8); + let sender = StreamSender(tx); + let handler = f.handler.clone(); + let task = tokio::spawn(async move { + handler + .handle_stream( + RequestId::new(), + RequestOp::Events { follow: false }, + sender, + ) + .await + }); + + let frame = tokio::time::timeout(std::time::Duration::from_millis(500), rx.recv()) + .await + .expect("stream_events should deliver history in <500ms") + .expect("channel closed before delivering frame"); + assert!(frame.event.is_some()); + + let _ = tokio::time::timeout(std::time::Duration::from_millis(500), task) + .await + .expect("handler should return after draining history"); + } + + fn minimal_valid_program_json() -> String { + serde_json::json!({ + "version": "0.1", + "name": "r", + "trigger": { "kind": "on_submit" }, + "predicates": {}, + "watches": [], + "body": [], + }) + .to_string() + } + + #[tokio::test] + async fn resume_transitions_store_and_sends_command() { + let f = fresh_handler(); + let pid = ProgramId::new(); + let path = f.dir.path().to_path_buf(); + let src = minimal_valid_program_json(); + { + let mut s = f.handler.store.lock().unwrap(); + s.insert_program(&vcli_store::NewProgram { + id: pid, + name: "r", + source_json: &src, + state: vcli_core::ProgramState::Pending, + submitted_at: 0, + labels_json: "{}", + }) + .unwrap(); + s.update_state(pid, vcli_core::ProgramState::Running, 1) + .unwrap(); + s.set_body_cursor(pid, 3).unwrap(); + } + // Trigger a recovery cycle by reopening the same DB. + let (_, _) = vcli_store::Store::open(&path).unwrap(); + + let resp = f + .handler + .handle( + RequestId::new(), + RequestOp::Resume { + program_id: pid, + from_start: false, + }, + ) + .await + .unwrap(); + let body = serde_json::to_value(&resp).unwrap(); + assert_eq!(body["ok"], true); + assert_eq!(body["result"]["from_step"], 3); + match f.cmd_rx.try_recv().unwrap() { + SchedulerCommand::ResumeRunning { + program_id, + from_step, + .. + } => { + assert_eq!(program_id, pid); + assert_eq!(from_step, 3); + } + other => panic!("wrong cmd: {other:?}"), + } + } + + #[tokio::test] + async fn resume_rejects_non_resumable_program() { + let f = fresh_handler(); + let pid = ProgramId::new(); + f.handler + .store + .lock() + .unwrap() + .insert_program(&vcli_store::NewProgram { + id: pid, + name: "r", + source_json: "{}", + state: vcli_core::ProgramState::Pending, + submitted_at: 0, + labels_json: "{}", + }) + .unwrap(); + let resp = f + .handler + .handle( + RequestId::new(), + RequestOp::Resume { + program_id: pid, + from_start: false, + }, + ) + .await + .unwrap(); + let body = serde_json::to_value(&resp).unwrap(); + assert_eq!(body["ok"], false); + assert_eq!(body["error"]["code"], "not_resumable"); + } + + #[tokio::test] + async fn cancel_sends_command_and_returns_ok() { + let f = fresh_handler(); + let pid = ProgramId::new(); + let resp = f + .handler + .handle(RequestId::new(), RequestOp::Cancel { program_id: pid }) + .await + .unwrap(); + let body = serde_json::to_value(&resp).unwrap(); + assert_eq!(body["ok"], true); + match f.cmd_rx.try_recv().unwrap() { + SchedulerCommand::Cancel { program_id, .. } => assert_eq!(program_id, pid), + other => panic!("wrong cmd: {other:?}"), + } + } + + #[tokio::test] + async fn start_sends_command_and_returns_ok() { + let f = fresh_handler(); + let pid = ProgramId::new(); + let resp = f + .handler + .handle(RequestId::new(), RequestOp::Start { program_id: pid }) + .await + .unwrap(); + let body = serde_json::to_value(&resp).unwrap(); + assert_eq!(body["ok"], true); + match f.cmd_rx.try_recv().unwrap() { + SchedulerCommand::Start { program_id } => assert_eq!(program_id, pid), + other => panic!("wrong cmd: {other:?}"), + } + } + + #[tokio::test] + async fn list_returns_all_programs_when_no_filter() { + let f = fresh_handler(); + for name in ["a", "b"] { + let id = ProgramId::new(); + f.handler + .store + .lock() + .unwrap() + .insert_program(&vcli_store::NewProgram { + id, + name, + source_json: "{}", + state: vcli_core::ProgramState::Pending, + submitted_at: 0, + labels_json: "{}", + }) + .unwrap(); + } + let resp = f + .handler + .handle(RequestId::new(), RequestOp::List { state: None }) + .await + .unwrap(); + let body = serde_json::to_value(&resp).unwrap(); + let items = body["result"]["items"].as_array().unwrap(); + assert_eq!(items.len(), 2); + } + + #[tokio::test] + async fn status_returns_row_for_known_id() { + let f = fresh_handler(); + let pid = ProgramId::new(); + f.handler + .store + .lock() + .unwrap() + .insert_program(&vcli_store::NewProgram { + id: pid, + name: "s", + source_json: "{}", + state: vcli_core::ProgramState::Pending, + submitted_at: 0, + labels_json: "{}", + }) + .unwrap(); + let resp = f + .handler + .handle(RequestId::new(), RequestOp::Status { program_id: pid }) + .await + .unwrap(); + let body = serde_json::to_value(&resp).unwrap(); + assert_eq!(body["result"]["name"], "s"); + assert_eq!(body["result"]["state"], "pending"); + } + + #[tokio::test] + async fn status_returns_unknown_program_for_missing_id() { + let f = fresh_handler(); + let resp = f + .handler + .handle( + RequestId::new(), + RequestOp::Status { + program_id: ProgramId::new(), + }, + ) + .await + .unwrap(); + let body = serde_json::to_value(&resp).unwrap(); + assert_eq!(body["ok"], false); + assert_eq!(body["error"]["code"], "unknown_program"); + } + + #[tokio::test] + async fn submit_validates_and_enqueues() { + let f = fresh_handler(); + let program = serde_json::json!({ + "version": "0.1", + "name": "noop", + "trigger": { "kind": "on_submit" }, + "predicates": {}, + "watches": [], + "body": [], + }); + let id = RequestId::new(); + let resp = f + .handler + .handle( + id, + RequestOp::Submit { + program: program.clone(), + }, + ) + .await + .unwrap(); + let body = serde_json::to_value(&resp).unwrap(); + assert_eq!(body["ok"], true); + let pid: ProgramId = body["result"]["program_id"] + .as_str() + .unwrap() + .parse() + .unwrap(); + let row = f.handler.store.lock().unwrap().get_program(pid).unwrap(); + assert_eq!(row.name, "noop"); + match f.cmd_rx.try_recv().unwrap() { + SchedulerCommand::SubmitValidated { program_id, .. } => assert_eq!(program_id, pid), + other => panic!("wrong cmd: {other:?}"), + } + } + + #[tokio::test] + async fn shutdown_triggers_oneshot_and_cmd() { + let f = fresh_handler(); + let id = RequestId::new(); + let _resp = f.handler.handle(id, RequestOp::Shutdown).await.unwrap(); + assert!(f.shutdown_rx.await.is_ok()); + } +} diff --git a/crates/vcli-daemon/src/lib.rs b/crates/vcli-daemon/src/lib.rs new file mode 100644 index 0000000..e4dcf35 --- /dev/null +++ b/crates/vcli-daemon/src/lib.rs @@ -0,0 +1,54 @@ +//! vcli-daemon — the binary that wires IPC, the store, and the scheduler. +//! +//! See the v0 design spec at `docs/superpowers/specs/2026-04-16-vcli-design.md` +//! §Threading model, §IPC, and §Restart semantics for the authoritative behavior. +//! +//! The library surface exists so that the binary in `src/bin/vcli-daemon.rs` is +//! a thin `main`, and so that integration tests can exercise the assembly with +//! mock capture + input backends. + +#![forbid(unsafe_code)] +#![deny(rust_2018_idioms)] +#![warn(missing_docs)] +#![warn(clippy::pedantic)] +#![allow(clippy::module_name_repetitions)] + +pub mod error; + +pub use error::{DaemonError, DaemonResult}; + +pub mod config; + +pub use config::{ensure_dirs, Config}; + +pub mod pidfile; + +pub use pidfile::PidFile; + +pub mod logging; + +pub use logging::{init as init_logging, LogGuard}; + +pub mod bridge; + +pub use bridge::{new_channels, CommandChannel, SchedulerCommand, EVENT_BROADCAST_CAPACITY}; + +pub mod persist; + +pub use persist::{program_id_of, spawn_event_pump}; + +pub mod handler; + +pub use handler::DaemonHandler; + +pub mod startup; + +pub use startup::{emit_daemon_started, emit_recovery_events, reload_waiting_programs}; + +pub mod shutdown; + +pub use shutdown::{emit_daemon_stopped, install_signal_handlers}; + +pub mod run; + +pub use run::{run_foreground, RuntimeBackends, RuntimeFactory}; diff --git a/crates/vcli-daemon/src/logging.rs b/crates/vcli-daemon/src/logging.rs new file mode 100644 index 0000000..8ed097a --- /dev/null +++ b/crates/vcli-daemon/src/logging.rs @@ -0,0 +1,71 @@ +//! tracing-subscriber setup with a daily-rolling JSON file + stderr fallback. +//! +//! Returns a `WorkerGuard` the caller must keep alive for the lifetime of the +//! process — dropping it flushes any buffered records. Emits JSON into +//! `/daemon.log.`; `tracing-appender` handles rotation and +//! keeps the rolling logic out of the daemon. + +use std::path::Path; + +use tracing_appender::rolling; +use tracing_subscriber::{fmt, prelude::*, EnvFilter}; + +use crate::error::{DaemonError, DaemonResult}; + +/// Non-drop-until-shutdown handle to the background log writer thread. +pub struct LogGuard(#[allow(dead_code)] tracing_appender::non_blocking::WorkerGuard); + +/// Install the tracing subscriber. Respects `RUST_LOG` via `EnvFilter`; defaults +/// to `info` if unset. +/// +/// # Errors +/// Returns [`DaemonError::Logging`] if the subscriber cannot be installed +/// (typically: a subscriber is already set in the current process). +pub fn init(log_dir: &Path) -> DaemonResult { + let file_appender = rolling::daily(log_dir, "daemon.log"); + let (nb, guard) = tracing_appender::non_blocking(file_appender); + + let env = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")); + + let subscriber = tracing_subscriber::registry().with(env).with( + fmt::layer() + .json() + .with_writer(nb) + .with_current_span(false) + .with_span_list(false), + ); + + subscriber + .try_init() + .map_err(|e| DaemonError::Logging(format!("{e}")))?; + + Ok(LogGuard(guard)) +} + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::tempdir; + + /// Smoke test: constructing the appender does not panic and produces a file + /// on first log write. We cannot call `init()` from a test (global + /// subscriber), so this exercises the appender alone. + #[test] + fn daily_appender_writes_to_log_dir() { + use std::io::Write; + let d = tempdir().unwrap(); + let app = rolling::daily(d.path(), "daemon.log"); + let (mut nb, guard) = tracing_appender::non_blocking(app); + nb.write_all(b"hello\n").unwrap(); + nb.flush().unwrap(); + drop(guard); + let mut found = false; + for entry in std::fs::read_dir(d.path()).unwrap() { + let e = entry.unwrap(); + if e.file_name().to_string_lossy().starts_with("daemon.log") { + found = true; + } + } + assert!(found, "expected a daemon.log.* file"); + } +} diff --git a/crates/vcli-daemon/src/persist.rs b/crates/vcli-daemon/src/persist.rs new file mode 100644 index 0000000..66c380e --- /dev/null +++ b/crates/vcli-daemon/src/persist.rs @@ -0,0 +1,131 @@ +//! Event pipeline. The scheduler pushes `vcli_core::Event`s into +//! `crossbeam_channel::Receiver`. A dedicated tokio task drains that +//! receiver, calls `store.append_event` for every program-scoped variant +//! (`DaemonStarted` / `DaemonStopped` / `DaemonPressure` / `StreamDropped` +//! etc. are broadcast-only — no program id, no DB row), then forwards the +//! event into the `broadcast::Sender`. +//! +//! Ordering invariant (Decision 1.7): an event is in the DB before any +//! subscriber can see it. + +use std::sync::{Arc, Mutex}; + +use crossbeam_channel::Receiver; +use tokio::sync::broadcast; +use tracing::{error, trace}; +use vcli_core::{Event, EventData, ProgramId}; +use vcli_store::Store; + +/// Drain `event_rx` on a background tokio task. Returns a `JoinHandle` so the +/// run loop can await shutdown. +/// +/// # Panics +/// The background task panics if the store mutex is poisoned. +pub fn spawn_event_pump( + store: Arc>, + event_rx: Receiver, + broadcast_tx: broadcast::Sender, +) -> tokio::task::JoinHandle<()> { + tokio::task::spawn_blocking(move || { + while let Ok(ev) = event_rx.recv() { + if let Some(pid) = program_id_of(&ev.data) { + let mut guard = store.lock().expect("store mutex poisoned"); + if let Err(e) = guard.append_event(pid, &ev) { + error!(error = %e, "failed to append event"); + } + } + let _ = broadcast_tx.send(ev); + trace!("event pump iteration"); + } + }) +} + +/// Extract the program id from an event payload, if any. Returns `None` for +/// daemon-scoped events that are broadcast-only. +#[must_use] +pub fn program_id_of(d: &EventData) -> Option { + match d { + EventData::ProgramSubmitted { program_id, .. } + | EventData::ProgramStateChanged { program_id, .. } + | EventData::ProgramCompleted { program_id, .. } + | EventData::ProgramFailed { program_id, .. } + | EventData::ProgramResumed { program_id, .. } + | EventData::WatchFired { program_id, .. } + | EventData::ActionDispatched { program_id, .. } + | EventData::ActionDeferred { program_id, .. } => Some(*program_id), + EventData::DaemonStarted { .. } + | EventData::DaemonStopped + | EventData::DaemonPressure { .. } + | EventData::StreamDropped { .. } + | EventData::TickFrameSkipped { .. } + | EventData::CapturePermissionMissing { .. } => None, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crossbeam_channel::unbounded; + use tempfile::tempdir; + use vcli_core::EventData; + use vcli_store::{NewProgram, Store}; + + #[tokio::test] + async fn program_event_is_appended_before_broadcast() { + let d = tempdir().unwrap(); + let (mut store, _) = Store::open(d.path()).unwrap(); + let pid = ProgramId::new(); + store + .insert_program(&NewProgram { + id: pid, + name: "x", + source_json: "{}", + state: vcli_core::ProgramState::Pending, + submitted_at: 0, + labels_json: "{}", + }) + .unwrap(); + let store = Arc::new(Mutex::new(store)); + let (sched_tx, sched_rx) = unbounded::(); + let (bcast_tx, mut bcast_rx) = broadcast::channel::(16); + + let pump = spawn_event_pump(store.clone(), sched_rx, bcast_tx); + + let ev = Event { + at: 7, + data: EventData::ProgramCompleted { + program_id: pid, + emit: None, + }, + }; + sched_tx.send(ev.clone()).unwrap(); + let got = bcast_rx.recv().await.unwrap(); + assert_eq!(got, ev); + let rows = store.lock().unwrap().stream_events(0, 10).unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].type_tag, "program.completed"); + + drop(sched_tx); + pump.await.unwrap(); + } + + #[test] + fn daemon_started_has_no_program_id() { + let d = EventData::DaemonStarted { + version: "0.0.1".into(), + }; + assert!(program_id_of(&d).is_none()); + } + + #[test] + fn state_changed_has_program_id() { + let p = ProgramId::new(); + let d = EventData::ProgramStateChanged { + program_id: p, + from: vcli_core::ProgramState::Waiting, + to: vcli_core::ProgramState::Running, + reason: "trigger".into(), + }; + assert_eq!(program_id_of(&d), Some(p)); + } +} diff --git a/crates/vcli-daemon/src/pidfile.rs b/crates/vcli-daemon/src/pidfile.rs new file mode 100644 index 0000000..25bd2de --- /dev/null +++ b/crates/vcli-daemon/src/pidfile.rs @@ -0,0 +1,191 @@ +//! Advisory pidfile. On acquire: +//! 1. Open (create if missing) `` read-write. +//! 2. `fs2::try_lock_exclusive` — if another process holds the lock, read +//! its contents and return [`DaemonError::AlreadyRunning`]. +//! 3. Rewind, truncate, write current PID + `\n`. +//! +//! On drop / explicit `release()`: unlock + unlink (best effort). +//! +//! The advisory lock is flock(2)-based via `fs2`; it's unaffected by file +//! copies and released if the holder dies without cleanup (kernel releases +//! the lock on fd close). That covers the SIGKILL case: a stale PID file on +//! disk has no lock, so the next daemon acquires it cleanly. + +use std::fs::{File, OpenOptions}; +use std::io::{Read, Seek, SeekFrom, Write}; +use std::path::{Path, PathBuf}; + +use fs2::FileExt; + +use crate::error::{DaemonError, DaemonResult}; + +/// Owning handle on the pidfile. Lock released when dropped. +pub struct PidFile { + path: PathBuf, + file: Option, +} + +impl std::fmt::Debug for PidFile { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("PidFile") + .field("path", &self.path) + .finish_non_exhaustive() + } +} + +impl PidFile { + /// Try to acquire the lock and write the current PID. + /// + /// # Errors + /// - [`DaemonError::AlreadyRunning`] if another process holds the lock. + /// - [`DaemonError::Pidfile`] on any underlying IO failure. + pub fn acquire(path: impl AsRef) -> DaemonResult { + let path = path.as_ref().to_path_buf(); + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent).map_err(|e| DaemonError::Pidfile { + path: path.clone(), + source: e, + })?; + } + let mut file = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(&path) + .map_err(|e| DaemonError::Pidfile { + path: path.clone(), + source: e, + })?; + + if file.try_lock_exclusive().is_err() { + let mut s = String::new(); + let _ = file.read_to_string(&mut s); + let pid: u32 = s.trim().parse().unwrap_or(0); + return Err(DaemonError::AlreadyRunning { pid, path }); + } + + file.set_len(0).map_err(|e| DaemonError::Pidfile { + path: path.clone(), + source: e, + })?; + file.seek(SeekFrom::Start(0)) + .map_err(|e| DaemonError::Pidfile { + path: path.clone(), + source: e, + })?; + writeln!(file, "{}", std::process::id()).map_err(|e| DaemonError::Pidfile { + path: path.clone(), + source: e, + })?; + file.flush().map_err(|e| DaemonError::Pidfile { + path: path.clone(), + source: e, + })?; + + Ok(Self { + path, + file: Some(file), + }) + } + + /// The path this lock is anchored to. + #[must_use] + pub fn path(&self) -> &Path { + &self.path + } + + /// PID recorded inside the file (always this process on a live handle). + #[must_use] + pub fn pid(&self) -> u32 { + std::process::id() + } + + /// Explicitly release (unlock + unlink). Drop does the same, but in an + /// ordered shutdown we want errors logged rather than swallowed. + /// + /// # Errors + /// IO errors during unlink. + pub fn release(mut self) -> DaemonResult<()> { + if let Some(file) = self.file.take() { + let _ = fs2::FileExt::unlock(&file); + drop(file); + } + std::fs::remove_file(&self.path).or_else(|e| { + if e.kind() == std::io::ErrorKind::NotFound { + Ok(()) + } else { + Err(DaemonError::Pidfile { + path: self.path.clone(), + source: e, + }) + } + }) + } +} + +impl Drop for PidFile { + fn drop(&mut self) { + if let Some(file) = self.file.take() { + let _ = fs2::FileExt::unlock(&file); + } + let _ = std::fs::remove_file(&self.path); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::tempdir; + + #[test] + fn acquire_creates_pidfile_with_current_pid() { + let d = tempdir().unwrap(); + let p = d.path().join("daemon.pid"); + let lock = PidFile::acquire(&p).unwrap(); + assert!(p.exists()); + let s = std::fs::read_to_string(&p).unwrap(); + assert_eq!(s.trim().parse::().unwrap(), std::process::id()); + drop(lock); + } + + #[test] + fn second_acquire_in_same_process_returns_already_running() { + let d = tempdir().unwrap(); + let p = d.path().join("daemon.pid"); + let _first = PidFile::acquire(&p).unwrap(); + let err = PidFile::acquire(&p).unwrap_err(); + match err { + DaemonError::AlreadyRunning { pid, .. } => assert_eq!(pid, std::process::id()), + other => panic!("wrong err: {other:?}"), + } + } + + #[test] + fn release_unlinks_pidfile() { + let d = tempdir().unwrap(); + let p = d.path().join("daemon.pid"); + let lock = PidFile::acquire(&p).unwrap(); + lock.release().unwrap(); + assert!(!p.exists()); + } + + #[test] + fn drop_unlinks_pidfile() { + let d = tempdir().unwrap(); + let p = d.path().join("daemon.pid"); + { + let _lock = PidFile::acquire(&p).unwrap(); + assert!(p.exists()); + } + assert!(!p.exists()); + } + + #[test] + fn parent_dir_is_created_on_acquire() { + let d = tempdir().unwrap(); + let p = d.path().join("nested").join("dir").join("daemon.pid"); + let _lock = PidFile::acquire(&p).unwrap(); + assert!(p.exists()); + } +} diff --git a/crates/vcli-daemon/src/run.rs b/crates/vcli-daemon/src/run.rs new file mode 100644 index 0000000..81547c9 --- /dev/null +++ b/crates/vcli-daemon/src/run.rs @@ -0,0 +1,129 @@ +//! Daemon assembly. `run_foreground` is the function the binary calls. All +//! meaningful side effects (opening the store, binding the socket, spawning +//! the scheduler thread, installing signals) happen inside here, in exactly +//! the order spec §Restart semantics Phase A requires. + +use std::sync::{Arc, Mutex}; +use std::thread; +use std::time::Instant; + +use tokio::sync::oneshot; +use tracing::{error, info}; + +use vcli_capture::Capture; +use vcli_input::InputSink; +use vcli_ipc::IpcServer; +use vcli_perception::Perception; +use vcli_store::Store; + +use crate::bridge::{new_channels, SchedulerCommand}; +use crate::config::{ensure_dirs, Config}; +use crate::error::{DaemonError, DaemonResult}; +use crate::handler::DaemonHandler; +use crate::persist::spawn_event_pump; +use crate::pidfile::PidFile; +use crate::shutdown::{emit_daemon_stopped, install_signal_handlers}; +use crate::startup::{emit_daemon_started, emit_recovery_events, reload_waiting_programs}; + +/// Bundle of backend implementations the daemon will hand to the scheduler. +pub struct RuntimeBackends { + /// Capture backend. + pub capture: Box, + /// Input sink. + pub input: Arc, + /// Perception façade. + pub perception: Perception, + /// Runtime clock (usually `vcli_runtime::SystemRuntimeClock`). The runtime + /// crate exposes its own `RuntimeClock` trait because Rust 1.75 lacks + /// trait-object upcasting, so we cannot pass a `vcli_core::Clock` here. + pub clock: Arc, +} + +/// Function that produces `RuntimeBackends` at startup. Lets tests inject +/// mocks without entering `run_foreground` through platform backends. +pub type RuntimeFactory = Box DaemonResult + Send>; + +/// Foreground entrypoint. Blocks until a signal / IPC shutdown drains the +/// process. Use this from `fn main` after argv parsing. +/// +/// # Errors +/// Startup failures (path resolution, already-running, store open, socket bind). +pub async fn run_foreground(cfg: Config, factory: RuntimeFactory) -> DaemonResult<()> { + ensure_dirs(&cfg)?; + let _log_guard = crate::logging::init(&cfg.log_dir)?; + info!( + data_root = %cfg.data_root.display(), + socket = %cfg.socket.path.display(), + "starting vcli-daemon" + ); + + let pid = PidFile::acquire(cfg.pidfile_path())?; + info!(pid = pid.pid(), pidfile = %pid.path().display(), "pidfile acquired"); + + let (store, recovered) = Store::open(&cfg.data_root)?; + let store = Arc::new(Mutex::new(store)); + + let (bridge, cmd_rx, event_rx, sched_event_tx) = new_channels(); + + let pump = spawn_event_pump(store.clone(), event_rx, bridge.event_tx.clone()); + + emit_recovery_events(&recovered, &sched_event_tx); + emit_daemon_started(&sched_event_tx); + + let _ = reload_waiting_programs(&store, &bridge.cmd_tx); + + let RuntimeBackends { + capture, + input, + perception, + clock, + } = factory()?; + let sched_event_tx_for_thread = sched_event_tx.clone(); + let scheduler_join = thread::Builder::new() + .name("vcli-scheduler".into()) + .spawn(move || { + let scheduler = vcli_runtime::Scheduler::new( + vcli_runtime::SchedulerConfig::default(), + capture, + input, + perception, + clock, + cmd_rx, + sched_event_tx_for_thread, + ); + scheduler.run_until_shutdown(); + }) + .map_err(DaemonError::Io)?; + + let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>(); + let shutdown_slot = Arc::new(Mutex::new(Some(shutdown_tx))); + let handler = DaemonHandler { + store: store.clone(), + bridge: bridge.clone(), + started_at: Instant::now(), + shutdown_tx: shutdown_slot.clone(), + }; + + install_signal_handlers(shutdown_slot.clone()) + .await + .map_err(DaemonError::Io)?; + + let server = IpcServer::bind(&cfg.socket.path, Arc::new(handler))?; + info!(socket = %cfg.socket.path.display(), "listening"); + + let serve_result = server.serve(shutdown_rx).await; + if let Err(e) = serve_result { + error!(error = %e, "ipc server terminated with error"); + } + + let _ = bridge.cmd_tx.send(SchedulerCommand::Shutdown); + emit_daemon_stopped(&sched_event_tx); + drop(sched_event_tx); + if let Err(e) = scheduler_join.join() { + error!("scheduler thread panicked: {e:?}"); + } + let _ = pump.await; + pid.release()?; + info!("vcli-daemon exited cleanly"); + Ok(()) +} diff --git a/crates/vcli-daemon/src/shutdown.rs b/crates/vcli-daemon/src/shutdown.rs new file mode 100644 index 0000000..63a574c --- /dev/null +++ b/crates/vcli-daemon/src/shutdown.rs @@ -0,0 +1,70 @@ +//! Signal handling. On Unix, SIGTERM and SIGINT both trigger graceful shutdown +//! by firing the oneshot the IPC server is listening on. The second signal is +//! not special — `tokio::select!` races the loop already. A third signal +//! should escalate via the OS (we don't install SIGKILL handling). + +use std::sync::{Arc, Mutex}; + +use futures_util::stream::StreamExt; +use tokio::sync::oneshot; +use tracing::{info, warn}; + +/// Install a signal handler on the current tokio runtime that fires +/// `shutdown_tx` on the first SIGTERM or SIGINT. `shutdown_tx` is shared with +/// `DaemonHandler` so an IPC `shutdown` also counts. +/// +/// # Errors +/// Returns `std::io::Error` if `signal_hook_tokio::Signals::new` fails. +/// +/// # Panics +/// The background signal task panics if the shutdown mutex is poisoned. +pub async fn install_signal_handlers( + shutdown_tx: Arc>>>, +) -> std::io::Result<()> { + use signal_hook_tokio::Signals; + let mut signals = Signals::new([libc::SIGTERM, libc::SIGINT])?; + tokio::spawn(async move { + if let Some(sig) = signals.next().await { + info!(signal = ?sig, "received shutdown signal"); + if let Some(tx) = shutdown_tx.lock().unwrap().take() { + let _ = tx.send(()); + } else { + warn!("shutdown already triggered"); + } + } + }); + Ok(()) +} + +/// Emit `daemon.stopped` on the scheduler event channel so IPC subscribers and +/// the DB see the transition. +pub fn emit_daemon_stopped(sched_event_tx: &crossbeam_channel::Sender) { + use vcli_core::{Event, EventData}; + let _ = sched_event_tx.send(Event { + at: vcli_core::clock::now_unix_ms(), + data: EventData::DaemonStopped, + }); +} + +#[cfg(test)] +mod tests { + use super::*; + use crossbeam_channel::unbounded; + + #[test] + fn emit_daemon_stopped_sends_one_event() { + let (tx, rx) = unbounded::(); + emit_daemon_stopped(&tx); + assert!(matches!( + rx.recv().unwrap().data, + vcli_core::EventData::DaemonStopped + )); + } + + #[tokio::test] + async fn install_handler_is_idempotent_without_signal() { + let (tx, _rx) = oneshot::channel(); + let slot = Arc::new(Mutex::new(Some(tx))); + install_signal_handlers(slot.clone()).await.unwrap(); + } +} diff --git a/crates/vcli-daemon/src/startup.rs b/crates/vcli-daemon/src/startup.rs new file mode 100644 index 0000000..97087ee --- /dev/null +++ b/crates/vcli-daemon/src/startup.rs @@ -0,0 +1,131 @@ +//! Startup tasks: emit events for orphan recovery, reload waiting programs, +//! publish `daemon.started`. Runs once on every boot; idempotent given the +//! `RecoveredProgram` set from `Store::open` is fresh each call. + +use std::sync::{Arc, Mutex}; + +use crossbeam_channel::Sender; +use tracing::info; + +use vcli_core::{Event, EventData, ProgramState}; +use vcli_store::{RecoveredProgram, Store}; + +/// Emit one `program.state_changed` + one `program.failed` for every program +/// that `Store::open` rewrote from `running` → `failed(daemon_restart)`. +/// The events are sent on `sched_event_tx` (scheduler → pump side), so the +/// persistence pipeline sees them first. +pub fn emit_recovery_events(recovered: &[RecoveredProgram], sched_event_tx: &Sender) { + let now = vcli_core::clock::now_unix_ms(); + for r in recovered { + let ev1 = Event { + at: now, + data: EventData::ProgramStateChanged { + program_id: r.id, + from: ProgramState::Running, + to: ProgramState::Failed, + reason: "daemon_restart".into(), + }, + }; + let ev2 = Event { + at: now, + data: EventData::ProgramFailed { + program_id: r.id, + reason: "daemon restarted during execution".into(), + step: Some(format!("body[{}]", r.body_cursor)), + emit: None, + }, + }; + let _ = sched_event_tx.send(ev1); + let _ = sched_event_tx.send(ev2); + } + info!(count = recovered.len(), "emitted recovery events"); +} + +/// Emit a `daemon.started` event. +pub fn emit_daemon_started(sched_event_tx: &Sender) { + let ev = Event { + at: vcli_core::clock::now_unix_ms(), + data: EventData::DaemonStarted { + version: env!("CARGO_PKG_VERSION").into(), + }, + }; + let _ = sched_event_tx.send(ev); +} + +/// Reload programs currently in `waiting` state: enqueue a scheduler command +/// per program so the scheduler re-installs its trigger. This is the second +/// half of spec §Restart semantics step 5. +/// +/// # Panics +/// Panics if the store mutex is poisoned. +pub fn reload_waiting_programs( + store: &Arc>, + cmd_tx: &crossbeam_channel::Sender, +) -> usize { + let rows = { + let s = store.lock().unwrap(); + s.list_programs(Some(ProgramState::Waiting)) + .unwrap_or_default() + }; + let mut sent = 0; + for row in &rows { + let Ok(value) = serde_json::from_str::(&row.source_json) else { + continue; + }; + let Ok(program) = vcli_dsl::validate_value(&value).map(|v| v.program) else { + continue; + }; + if cmd_tx + .send(crate::bridge::SchedulerCommand::SubmitValidated { + program_id: row.id, + program, + }) + .is_ok() + { + sent += 1; + } + } + info!(count = sent, "reloaded waiting programs"); + sent +} + +#[cfg(test)] +mod tests { + use super::*; + use crossbeam_channel::unbounded; + use tempfile::tempdir; + + #[test] + fn emits_two_events_per_recovered_program() { + let (tx, rx) = unbounded::(); + let recovered = vec![RecoveredProgram { + id: vcli_core::ProgramId::new(), + body_cursor: 2, + }]; + emit_recovery_events(&recovered, &tx); + let first = rx.recv().unwrap(); + let second = rx.recv().unwrap(); + assert!(matches!(first.data, EventData::ProgramStateChanged { .. })); + assert!(matches!(second.data, EventData::ProgramFailed { .. })); + } + + #[test] + fn emit_daemon_started_sends_one_event() { + let (tx, rx) = unbounded::(); + emit_daemon_started(&tx); + assert!(matches!( + rx.recv().unwrap().data, + EventData::DaemonStarted { .. } + )); + } + + #[test] + fn reload_waiting_does_nothing_when_none_exist() { + let d = tempdir().unwrap(); + let (store, _) = Store::open(d.path()).unwrap(); + let store = Arc::new(Mutex::new(store)); + let (cmd_tx, _cmd_rx) = unbounded::(); + let n = reload_waiting_programs(&store, &cmd_tx); + assert_eq!(n, 0); + } +} diff --git a/crates/vcli-daemon/tests/graceful_shutdown.rs b/crates/vcli-daemon/tests/graceful_shutdown.rs new file mode 100644 index 0000000..d4a56ab --- /dev/null +++ b/crates/vcli-daemon/tests/graceful_shutdown.rs @@ -0,0 +1,61 @@ +//! On shutdown, the socket file and pidfile should both be unlinked. + +use std::sync::Arc; +use std::time::Duration; + +use tempfile::tempdir; + +use vcli_daemon::{run_foreground, Config, RuntimeBackends}; +use vcli_ipc::socket_path::SocketPathOrigin; +use vcli_ipc::{IpcClient, RequestOp, ResponseBody}; + +fn mocks() -> Result { + Ok(RuntimeBackends { + capture: Box::new(vcli_capture::MockCapture::empty()), + input: Arc::new(vcli_input::MockInputSink::new()), + perception: vcli_perception::Perception::default(), + clock: Arc::new(vcli_runtime::SystemRuntimeClock::new()), + }) +} + +#[tokio::test] +async fn shutdown_unlinks_socket_file() { + let dir = tempdir().unwrap(); + let sock = dir.path().join("vcli.sock"); + let cfg = Config::with_roots( + dir.path().join("data"), + dir.path().join("logs"), + vcli_ipc::SocketPath { + path: sock.clone(), + origin: SocketPathOrigin::Override, + }, + ); + + let factory: vcli_daemon::RuntimeFactory = Box::new(mocks); + let run = tokio::spawn(run_foreground(cfg, factory)); + + for _ in 0..50 { + if sock.exists() { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert!(sock.exists()); + + let mut client = IpcClient::connect(&sock).await.unwrap(); + let r = client.request(RequestOp::Shutdown).await.unwrap(); + assert!(matches!(r.body, ResponseBody::Ok { .. })); + + tokio::time::timeout(Duration::from_secs(2), run) + .await + .unwrap() + .unwrap() + .unwrap(); + + assert!(!sock.exists(), "socket should be unlinked after shutdown"); + let pidfile = dir.path().join("data").join("daemon.pid"); + assert!( + !pidfile.exists(), + "pidfile should be removed after shutdown" + ); +} diff --git a/crates/vcli-daemon/tests/startup_orphan_recovery.rs b/crates/vcli-daemon/tests/startup_orphan_recovery.rs new file mode 100644 index 0000000..be472b3 --- /dev/null +++ b/crates/vcli-daemon/tests/startup_orphan_recovery.rs @@ -0,0 +1,88 @@ +//! Startup integration: plant a `running` row in a store before the daemon +//! starts, then drive `run_foreground` until the socket is listening and +//! shutdown cleanly, asserting: +//! * the row is now `failed(daemon_restart)` +//! * the event log contains `program.state_changed` and `program.failed`. + +use std::sync::Arc; +use std::time::Duration; + +use tempfile::tempdir; +use tokio::time::timeout; + +use vcli_core::ProgramState; +use vcli_daemon::{run_foreground, Config, RuntimeBackends}; +use vcli_ipc::socket_path::SocketPathOrigin; +use vcli_store::{NewProgram, Store}; + +fn noop_backends() -> Result { + Ok(RuntimeBackends { + capture: Box::new(vcli_capture::MockCapture::empty()), + input: Arc::new(vcli_input::MockInputSink::new()), + perception: vcli_perception::Perception::default(), + clock: Arc::new(vcli_runtime::SystemRuntimeClock::new()), + }) +} + +#[tokio::test] +async fn running_programs_transition_to_failed_on_startup() { + let dir = tempdir().unwrap(); + let data_root = dir.path().join("data"); + let log_dir = dir.path().join("logs"); + let sock = dir.path().join("vcli.sock"); + + let pid = vcli_core::ProgramId::new(); + { + let (mut store, _) = Store::open(&data_root).unwrap(); + store + .insert_program(&NewProgram { + id: pid, + name: "orphan", + source_json: "{}", + state: ProgramState::Pending, + submitted_at: 0, + labels_json: "{}", + }) + .unwrap(); + store.update_state(pid, ProgramState::Running, 1).unwrap(); + store.set_body_cursor(pid, 2).unwrap(); + } + + let cfg = Config::with_roots( + &data_root, + &log_dir, + vcli_ipc::SocketPath { + path: sock.clone(), + origin: SocketPathOrigin::Override, + }, + ); + + let factory: vcli_daemon::RuntimeFactory = Box::new(noop_backends); + let run = tokio::spawn(run_foreground(cfg, factory)); + + for _ in 0..50 { + if sock.exists() { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert!(sock.exists(), "daemon never bound the socket"); + + let mut client = vcli_ipc::IpcClient::connect(&sock).await.unwrap(); + let resp = client.request(vcli_ipc::RequestOp::Shutdown).await.unwrap(); + assert!(matches!(resp.body, vcli_ipc::ResponseBody::Ok { .. })); + + timeout(Duration::from_secs(2), run) + .await + .unwrap() + .unwrap() + .unwrap(); + + let (store, _) = Store::open(&data_root).unwrap(); + let row = store.get_program(pid).unwrap(); + assert_eq!(row.state, ProgramState::Failed); + let events = store.stream_events(0, 100).unwrap(); + let types: Vec<&str> = events.iter().map(|e| e.type_tag.as_str()).collect(); + assert!(types.contains(&"program.state_changed")); + assert!(types.contains(&"program.failed")); +} diff --git a/crates/vcli-daemon/tests/submit_and_run.rs b/crates/vcli-daemon/tests/submit_and_run.rs new file mode 100644 index 0000000..9088d7b --- /dev/null +++ b/crates/vcli-daemon/tests/submit_and_run.rs @@ -0,0 +1,73 @@ +//! End-to-end: start the daemon with mock backends, submit a minimal program +//! over IPC, assert the daemon returns a program_id, then shut down. + +use std::sync::Arc; +use std::time::Duration; + +use tempfile::tempdir; +use tokio::time::timeout; + +use vcli_daemon::{run_foreground, Config, RuntimeBackends}; +use vcli_ipc::socket_path::SocketPathOrigin; +use vcli_ipc::{IpcClient, RequestOp, ResponseBody}; + +fn mocks() -> Result { + Ok(RuntimeBackends { + capture: Box::new(vcli_capture::MockCapture::empty()), + input: Arc::new(vcli_input::MockInputSink::new()), + perception: vcli_perception::Perception::default(), + clock: Arc::new(vcli_runtime::SystemRuntimeClock::new()), + }) +} + +#[tokio::test] +async fn submit_creates_program_row_and_returns_id() { + let dir = tempdir().unwrap(); + let sock = dir.path().join("vcli.sock"); + let cfg = Config::with_roots( + dir.path().join("data"), + dir.path().join("logs"), + vcli_ipc::SocketPath { + path: sock.clone(), + origin: SocketPathOrigin::Override, + }, + ); + + let factory: vcli_daemon::RuntimeFactory = Box::new(mocks); + let run = tokio::spawn(run_foreground(cfg, factory)); + + for _ in 0..50 { + if sock.exists() { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert!(sock.exists()); + + let mut client = IpcClient::connect(&sock).await.unwrap(); + let program = serde_json::json!({ + "version": "0.1", + "name": "noop", + "trigger": { "kind": "on_submit" }, + "predicates": {}, + "watches": [], + "body": [], + }); + let resp = client.request(RequestOp::Submit { program }).await.unwrap(); + match resp.body { + ResponseBody::Ok { result, .. } => { + let pid = result["program_id"].as_str().unwrap(); + assert!(!pid.is_empty()); + } + ResponseBody::Err { error, .. } => panic!("submit failed: {error:?}"), + } + + let shut = client.request(RequestOp::Shutdown).await.unwrap(); + assert!(matches!(shut.body, ResponseBody::Ok { .. })); + + timeout(Duration::from_secs(2), run) + .await + .unwrap() + .unwrap() + .unwrap(); +} diff --git a/crates/vcli-store/src/store.rs b/crates/vcli-store/src/store.rs index bb494a9..c17bddf 100644 --- a/crates/vcli-store/src/store.rs +++ b/crates/vcli-store/src/store.rs @@ -197,6 +197,50 @@ impl Store { )?; Ok(()) } + + /// List programs, optionally filtered by a single [`ProgramState`]. + /// Ordered by `submitted_at ASC` then by rowid for stability. + /// + /// # Errors + /// Surfaces `SQLite` errors. + /// + /// # Panics + /// Panics if a stored id or state value cannot be parsed (should never + /// happen for rows written by this crate). + pub fn list_programs( + &self, + state_filter: Option, + ) -> StoreResult> { + let state_str = state_filter.map(|s| s.as_str().to_string()); + let mut stmt = self.conn.prepare( + "SELECT id, name, source_json, state, submitted_at, started_at, + finished_at, last_error_code, last_error_msg, labels_json, + body_cursor, body_entered_at + FROM programs + WHERE (?1 IS NULL) OR (state = ?1) + ORDER BY submitted_at ASC, rowid ASC", + )?; + let rows = stmt.query_map(rusqlite::params![state_str], |r| { + let id: String = r.get(0)?; + let state: String = r.get(3)?; + let body_cursor: i64 = r.get(10)?; + Ok(ProgramRow { + id: id.parse().unwrap(), + name: r.get(1)?, + source_json: r.get(2)?, + state: state.parse().unwrap(), + submitted_at: r.get(4)?, + started_at: r.get(5)?, + finished_at: r.get(6)?, + last_error_code: r.get(7)?, + last_error_msg: r.get(8)?, + labels_json: r.get(9)?, + body_cursor: u32::try_from(body_cursor).unwrap_or(0), + body_entered_at: r.get(11)?, + }) + })?; + rows.collect::, _>>().map_err(Into::into) + } } /// Values a caller supplies when inserting a new program row. @@ -366,6 +410,49 @@ mod tests { assert_eq!(s.get_program(id).unwrap().body_cursor, 7); } + #[test] + fn list_programs_returns_all_when_no_filter() { + let d = tempdir().unwrap(); + let (mut s, _) = Store::open(d.path()).unwrap(); + for (i, name) in ["a", "b", "c"].into_iter().enumerate() { + s.insert_program(&NewProgram { + id: ProgramId::new(), + name, + source_json: "{}", + state: ProgramState::Pending, + submitted_at: i64::try_from(i).unwrap(), + labels_json: "{}", + }) + .unwrap(); + } + let all = s.list_programs(None).unwrap(); + assert_eq!(all.len(), 3); + assert_eq!(all[0].name, "a"); + } + + #[test] + fn list_programs_filters_by_state() { + let d = tempdir().unwrap(); + let (mut s, _) = Store::open(d.path()).unwrap(); + let running = ProgramId::new(); + let pending = ProgramId::new(); + for (id, name) in [(running, "r"), (pending, "p")] { + s.insert_program(&NewProgram { + id, + name, + source_json: "{}", + state: ProgramState::Pending, + submitted_at: 0, + labels_json: "{}", + }) + .unwrap(); + } + s.update_state(running, ProgramState::Running, 100).unwrap(); + let runs = s.list_programs(Some(ProgramState::Running)).unwrap(); + assert_eq!(runs.len(), 1); + assert_eq!(runs[0].name, "r"); + } + #[test] fn reopen_preserves_rows() { let d = tempdir().unwrap();