From c748237741b5250522afa0dab22ea36bccce3fa1 Mon Sep 17 00:00:00 2001 From: LGUG2Z Date: Tue, 27 Oct 2026 06:14:14 -0700 Subject: [PATCH] feat(komorebi_link): record what komorebi says The recorder subscribes to komorebi over its socket (a unix socket on macOS, `uds_windows` on Windows) and writes every notification with the state it came with, stamped on the master clock. Its accept loop only hands off to a writer thread, since komorebi notifies subscribers while holding its locks. What happens to the recording itself (a komorebi that exits, a stream that dies) is written as incidents, and a lock file says whether a recorder is still running in a session. --- Cargo.lock | 397 +++++++++++ dependencies.json | 24 + docs/ARCHITECTURE.md | 1 + docs/PLATFORM.md | 18 + komorebi-recorder/Cargo.toml | 3 + komorebi-recorder/src/analysis/events.rs | 61 ++ komorebi-recorder/src/incidents.rs | 433 ++++++++++++ komorebi-recorder/src/komorebi_link.rs | 530 ++++++++++++++ komorebi-recorder/src/lib.rs | 8 + komorebi-recorder/src/lock.rs | 280 ++++++++ .../src/sources/komorebi_events.rs | 665 ++++++++++++++++++ komorebi-recorder/src/sources/mod.rs | 34 + 12 files changed, 2454 insertions(+) create mode 100644 komorebi-recorder/src/incidents.rs create mode 100644 komorebi-recorder/src/komorebi_link.rs create mode 100644 komorebi-recorder/src/lock.rs create mode 100644 komorebi-recorder/src/sources/komorebi_events.rs create mode 100644 komorebi-recorder/src/sources/mod.rs diff --git a/Cargo.lock b/Cargo.lock index ee8a0abd..e4217cf2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -67,6 +67,18 @@ dependencies = [ "windows-sys 0.60.2", ] +[[package]] +name = "anyhow" +version = "1.0.102" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" + +[[package]] +name = "autocfg" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" + [[package]] name = "backtrace" version = "0.3.76" @@ -168,12 +180,54 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" +[[package]] +name = "crossbeam-channel" +version = "0.5.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "82b8f8f868b36967f9606790d1903570de9ceaf870a7bf9fbbd3016d636a2cb2" +dependencies = [ + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-utils" +version = "0.8.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28" + +[[package]] +name = "dirs" +version = "6.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3e8aa94d75141228480295a7d0e7feb620b1a5ad9f12bc40be62411e38cce4e" +dependencies = [ + "dirs-sys", +] + +[[package]] +name = "dirs-sys" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e01a3366d27ee9890022452ee61b2b63a67e6f13f58900b651ff5665f0bb1fab" +dependencies = [ + "libc", + "option-ext", + "redox_users", + "windows-sys 0.59.0", +] + [[package]] name = "dyn-clone" version = "1.0.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555" +[[package]] +name = "equivalent" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" + [[package]] name = "errno" version = "0.3.14" @@ -194,24 +248,93 @@ dependencies = [ "once_cell", ] +[[package]] +name = "fastrand" +version = "2.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f1f227452a390804cdb637b74a86990f2a7d7ba4b7d5693aac9b4dd6defd8d6" + +[[package]] +name = "foldhash" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" + +[[package]] +name = "getrandom" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" +dependencies = [ + "cfg-if", + "libc", + "wasi", +] + +[[package]] +name = "getrandom" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0de51e6874e94e7bf76d726fc5d13ba782deca734ff60d5bb2fb2607c7406555" +dependencies = [ + "cfg-if", + "libc", + "r-efi", + "wasip2", + "wasip3", +] + [[package]] name = "gimli" version = "0.32.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e629b9b98ef3dd8afe6ca2bd0f89306cec16d43d907889945bc5d6687f2f13c7" +[[package]] +name = "hashbrown" +version = "0.15.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" +dependencies = [ + "foldhash", +] + +[[package]] +name = "hashbrown" +version = "0.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4f467dd6dccf739c208452f8014c75c18bb8301b050ad1cfb27153803edb0f51" + [[package]] name = "heck" version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" +[[package]] +name = "id-arena" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3d3067d79b975e8844ca9eb072e16b31c3c1c36928edf9c6789548c524d0d954" + [[package]] name = "indenter" version = "0.3.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "964de6e86d545b246d84badc0fef527924ace5134f30641c203ef52ba83f58d5" +[[package]] +name = "indexmap" +version = "2.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9" +dependencies = [ + "equivalent", + "hashbrown 0.17.0", + "serde", + "serde_core", +] + [[package]] name = "is_terminal_polyfill" version = "1.70.2" @@ -230,10 +353,13 @@ version = "0.1.0" dependencies = [ "clap", "color-eyre", + "crossbeam-channel", + "dirs", "libc", "schemars", "serde", "serde_json_lenient", + "uds_windows", "windows", ] @@ -243,24 +369,54 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" +[[package]] +name = "leb128fmt" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2" + [[package]] name = "libc" version = "0.2.186" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" +[[package]] +name = "libredox" +version = "0.1.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e02f3bb43d335493c96bf3fd3a321600bf6bd07ed34bc64118e9293bdffea46c" +dependencies = [ + "libc", +] + [[package]] name = "linux-raw-sys" version = "0.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53" +[[package]] +name = "log" +version = "0.4.29" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" + [[package]] name = "memchr" version = "2.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79" +[[package]] +name = "memoffset" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "488016bfae457b036d996092f6cb448677611ce4449e970ceaf42695203f218a" +dependencies = [ + "autocfg", +] + [[package]] name = "miniz_oxide" version = "0.8.9" @@ -291,6 +447,12 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" +[[package]] +name = "option-ext" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "04744f49eae99ab78e0d5c0b603ab218f515ea8cfe5a456d7629ad883a3b6e7d" + [[package]] name = "owo-colors" version = "4.3.0" @@ -303,6 +465,16 @@ version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" +[[package]] +name = "prettyplease" +version = "0.2.37" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "479ca8adacdd7ce8f1fb39ce9ecccbfe93a3f1344b3d0d97f20bc0196208f62b" +dependencies = [ + "proc-macro2", + "syn", +] + [[package]] name = "proc-macro2" version = "1.0.106" @@ -321,6 +493,23 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "r-efi" +version = "6.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" + +[[package]] +name = "redox_users" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4e608c6638b9c18977b00b475ac1f28d14e84b27d8d42f70e0bf1e3dec127ac" +dependencies = [ + "getrandom 0.2.17", + "libredox", + "thiserror", +] + [[package]] name = "ref-cast" version = "1.0.25" @@ -391,6 +580,12 @@ dependencies = [ "syn", ] +[[package]] +name = "semver" +version = "1.0.28" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8a7852d02fc848982e0c167ef163aaff9cd91dc640ba85e263cb1ce46fae51cd" + [[package]] name = "serde" version = "1.0.228" @@ -483,6 +678,19 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "tempfile" +version = "3.27.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" +dependencies = [ + "fastrand", + "getrandom 0.4.2", + "once_cell", + "rustix", + "windows-sys 0.52.0", +] + [[package]] name = "terminal_size" version = "0.4.4" @@ -493,6 +701,26 @@ dependencies = [ "windows-sys 0.59.0", ] +[[package]] +name = "thiserror" +version = "2.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4288b5bcbc7920c07a1149a35cf9590a2aa808e0bc1eafaade0b80947865fbc4" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "2.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebc4ee7f67670e9b64d05fa4253e753e016c6c95ff35b89b7941d6b856dec1d5" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "thread_local" version = "1.1.9" @@ -543,12 +771,29 @@ dependencies = [ "tracing-core", ] +[[package]] +name = "uds_windows" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2f6fb2847f6742cd76af783a2a2c49e9375d0a111c7bef6f71cd9e738c72d6e" +dependencies = [ + "memoffset", + "tempfile", + "windows-sys 0.60.2", +] + [[package]] name = "unicode-ident" version = "1.0.24" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" +[[package]] +name = "unicode-xid" +version = "0.2.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853" + [[package]] name = "utf8parse" version = "0.2.2" @@ -561,6 +806,64 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + +[[package]] +name = "wasip2" +version = "1.0.3+wasi-0.2.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "20064672db26d7cdc89c7798c48a0fdfac8213434a1186e5ef29fd560ae223d6" +dependencies = [ + "wit-bindgen 0.57.1", +] + +[[package]] +name = "wasip3" +version = "0.4.0+wasi-0.3.0-rc-2026-01-06" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5428f8bf88ea5ddc08faddef2ac4a67e390b88186c703ce6dbd955e1c145aca5" +dependencies = [ + "wit-bindgen 0.51.0", +] + +[[package]] +name = "wasm-encoder" +version = "0.244.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "990065f2fe63003fe337b932cfb5e3b80e0b4d0f5ff650e6985b1048f62c8319" +dependencies = [ + "leb128fmt", + "wasmparser", +] + +[[package]] +name = "wasm-metadata" +version = "0.244.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bb0e353e6a2fbdc176932bbaab493762eb1255a7900fe0fea1a2f96c296cc909" +dependencies = [ + "anyhow", + "indexmap", + "wasm-encoder", + "wasmparser", +] + +[[package]] +name = "wasmparser" +version = "0.244.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47b807c72e1bac69382b3a6fb3dbe8ea4c0ed87ff5629b8685ae6b9a611028fe" +dependencies = [ + "bitflags", + "hashbrown 0.15.5", + "indexmap", + "semver", +] + [[package]] name = "windows" version = "0.62.2" @@ -827,6 +1130,100 @@ version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6bbff5f0aada427a1e5a6da5f1f98158182f26556f345ac9e04d36d0ebed650" +[[package]] +name = "wit-bindgen" +version = "0.51.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d7249219f66ced02969388cf2bb044a09756a083d0fab1e566056b04d9fbcaa5" +dependencies = [ + "wit-bindgen-rust-macro", +] + +[[package]] +name = "wit-bindgen" +version = "0.57.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e" + +[[package]] +name = "wit-bindgen-core" +version = "0.51.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ea61de684c3ea68cb082b7a88508a8b27fcc8b797d738bfc99a82facf1d752dc" +dependencies = [ + "anyhow", + "heck", + "wit-parser", +] + +[[package]] +name = "wit-bindgen-rust" +version = "0.51.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7c566e0f4b284dd6561c786d9cb0142da491f46a9fbed79ea69cdad5db17f21" +dependencies = [ + "anyhow", + "heck", + "indexmap", + "prettyplease", + "syn", + "wasm-metadata", + "wit-bindgen-core", + "wit-component", +] + +[[package]] +name = "wit-bindgen-rust-macro" +version = "0.51.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c0f9bfd77e6a48eccf51359e3ae77140a7f50b1e2ebfe62422d8afdaffab17a" +dependencies = [ + "anyhow", + "prettyplease", + "proc-macro2", + "quote", + "syn", + "wit-bindgen-core", + "wit-bindgen-rust", +] + +[[package]] +name = "wit-component" +version = "0.244.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d66ea20e9553b30172b5e831994e35fbde2d165325bec84fc43dbf6f4eb9cb2" +dependencies = [ + "anyhow", + "bitflags", + "indexmap", + "log", + "serde", + "serde_derive", + "serde_json", + "wasm-encoder", + "wasm-metadata", + "wasmparser", + "wit-parser", +] + +[[package]] +name = "wit-parser" +version = "0.244.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ecc8ac4bc1dc3381b7f59c34f00b67e18f910c2c0f50015669dde7def656a736" +dependencies = [ + "anyhow", + "id-arena", + "indexmap", + "log", + "semver", + "serde", + "serde_derive", + "serde_json", + "unicode-xid", + "wasmparser", +] + [[package]] name = "zmij" version = "1.0.21" diff --git a/dependencies.json b/dependencies.json index 1276c738..e5d20f62 100644 --- a/dependencies.json +++ b/dependencies.json @@ -16,6 +16,7 @@ "anstyle-parse 1.0.0 registry+https://github.com/rust-lang/crates.io-index", "anstyle-query 1.1.5 registry+https://github.com/rust-lang/crates.io-index", "anstyle-wincon 3.0.11 registry+https://github.com/rust-lang/crates.io-index", + "autocfg 1.5.0 registry+https://github.com/rust-lang/crates.io-index", "backtrace 0.3.76 registry+https://github.com/rust-lang/crates.io-index", "bitflags 2.11.1 registry+https://github.com/rust-lang/crates.io-index", "cfg-if 1.0.4 registry+https://github.com/rust-lang/crates.io-index", @@ -26,9 +27,15 @@ "color-eyre 0.6.5 registry+https://github.com/rust-lang/crates.io-index", "color-spantrace 0.3.0 registry+https://github.com/rust-lang/crates.io-index", "colorchoice 1.0.5 registry+https://github.com/rust-lang/crates.io-index", + "crossbeam-channel 0.5.15 registry+https://github.com/rust-lang/crates.io-index", + "crossbeam-utils 0.8.21 registry+https://github.com/rust-lang/crates.io-index", + "dirs 6.0.0 registry+https://github.com/rust-lang/crates.io-index", + "dirs-sys 0.5.0 registry+https://github.com/rust-lang/crates.io-index", "dyn-clone 1.0.20 registry+https://github.com/rust-lang/crates.io-index", "errno 0.3.14 registry+https://github.com/rust-lang/crates.io-index", "eyre 0.6.12 registry+https://github.com/rust-lang/crates.io-index", + "fastrand 2.4.1 registry+https://github.com/rust-lang/crates.io-index", + "getrandom 0.4.2 registry+https://github.com/rust-lang/crates.io-index", "gimli 0.32.3 registry+https://github.com/rust-lang/crates.io-index", "heck 0.5.0 registry+https://github.com/rust-lang/crates.io-index", "indenter 0.3.4 registry+https://github.com/rust-lang/crates.io-index", @@ -55,6 +62,7 @@ "serde_json 1.0.149 registry+https://github.com/rust-lang/crates.io-index", "serde_json_lenient 0.2.4 registry+https://github.com/rust-lang/crates.io-index", "syn 2.0.117 registry+https://github.com/rust-lang/crates.io-index", + "tempfile 3.27.0 registry+https://github.com/rust-lang/crates.io-index", "terminal_size 0.4.4 registry+https://github.com/rust-lang/crates.io-index", "thread_local 1.1.9 registry+https://github.com/rust-lang/crates.io-index", "unicode-ident 1.0.24 registry+https://github.com/rust-lang/crates.io-index", @@ -107,6 +115,7 @@ "anstyle-parse 1.0.0 registry+https://github.com/rust-lang/crates.io-index", "anstyle-query 1.1.5 registry+https://github.com/rust-lang/crates.io-index", "anstyle-wincon 3.0.11 registry+https://github.com/rust-lang/crates.io-index", + "autocfg 1.5.0 registry+https://github.com/rust-lang/crates.io-index", "backtrace 0.3.76 registry+https://github.com/rust-lang/crates.io-index", "bitflags 2.11.1 registry+https://github.com/rust-lang/crates.io-index", "cfg-if 1.0.4 registry+https://github.com/rust-lang/crates.io-index", @@ -117,9 +126,15 @@ "color-eyre 0.6.5 registry+https://github.com/rust-lang/crates.io-index", "color-spantrace 0.3.0 registry+https://github.com/rust-lang/crates.io-index", "colorchoice 1.0.5 registry+https://github.com/rust-lang/crates.io-index", + "crossbeam-channel 0.5.15 registry+https://github.com/rust-lang/crates.io-index", + "crossbeam-utils 0.8.21 registry+https://github.com/rust-lang/crates.io-index", + "dirs 6.0.0 registry+https://github.com/rust-lang/crates.io-index", + "dirs-sys 0.5.0 registry+https://github.com/rust-lang/crates.io-index", "dyn-clone 1.0.20 registry+https://github.com/rust-lang/crates.io-index", "errno 0.3.14 registry+https://github.com/rust-lang/crates.io-index", "eyre 0.6.12 registry+https://github.com/rust-lang/crates.io-index", + "fastrand 2.4.1 registry+https://github.com/rust-lang/crates.io-index", + "getrandom 0.4.2 registry+https://github.com/rust-lang/crates.io-index", "gimli 0.32.3 registry+https://github.com/rust-lang/crates.io-index", "heck 0.5.0 registry+https://github.com/rust-lang/crates.io-index", "indenter 0.3.4 registry+https://github.com/rust-lang/crates.io-index", @@ -128,6 +143,7 @@ "lazy_static 1.5.0 registry+https://github.com/rust-lang/crates.io-index", "libc 0.2.186 registry+https://github.com/rust-lang/crates.io-index", "memchr 2.8.0 registry+https://github.com/rust-lang/crates.io-index", + "memoffset 0.9.1 registry+https://github.com/rust-lang/crates.io-index", "miniz_oxide 0.8.9 registry+https://github.com/rust-lang/crates.io-index", "object 0.37.3 registry+https://github.com/rust-lang/crates.io-index", "once_cell 1.21.4 registry+https://github.com/rust-lang/crates.io-index", @@ -151,12 +167,14 @@ "sharded-slab 0.1.7 registry+https://github.com/rust-lang/crates.io-index", "strsim 0.11.1 registry+https://github.com/rust-lang/crates.io-index", "syn 2.0.117 registry+https://github.com/rust-lang/crates.io-index", + "tempfile 3.27.0 registry+https://github.com/rust-lang/crates.io-index", "terminal_size 0.4.4 registry+https://github.com/rust-lang/crates.io-index", "thread_local 1.1.9 registry+https://github.com/rust-lang/crates.io-index", "tracing 0.1.44 registry+https://github.com/rust-lang/crates.io-index", "tracing-core 0.1.36 registry+https://github.com/rust-lang/crates.io-index", "tracing-error 0.2.1 registry+https://github.com/rust-lang/crates.io-index", "tracing-subscriber 0.3.23 registry+https://github.com/rust-lang/crates.io-index", + "uds_windows 1.2.1 registry+https://github.com/rust-lang/crates.io-index", "unicode-ident 1.0.24 registry+https://github.com/rust-lang/crates.io-index", "utf8parse 0.2.2 registry+https://github.com/rust-lang/crates.io-index", "windows 0.62.2 registry+https://github.com/rust-lang/crates.io-index", @@ -180,6 +198,12 @@ "zmij 1.0.21 registry+https://github.com/rust-lang/crates.io-index" ] ], + [ + "MPL-2.0", + [ + "option-ext 0.2.0 registry+https://github.com/rust-lang/crates.io-index" + ] + ], [ "Unicode-3.0", [ diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 32234603..7db49510 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -72,3 +72,4 @@ source and have not been checked against a recording of it: an analysis that rea | An implementor of `Rules` answers every question that has no default | the type system: a missing method is E0046, proved by a `compile_fail` example on `Rules` | | Code that reads komorebi's output asks `Rules` and does not match on `Flavor` | convention only | | Rules that were not checked against a recording of komorebi say so | the framework path: an analysis that reads such a session adds a warning | +| A running recorder can be told from one that died | the OS: an exclusive lock on `recorder.lock`, held for as long as the recorder runs and dropped by the OS when the process ends however it ends | diff --git a/docs/PLATFORM.md b/docs/PLATFORM.md index 1c7f9cb3..c9a1dd9e 100644 --- a/docs/PLATFORM.md +++ b/docs/PLATFORM.md @@ -46,6 +46,24 @@ Video frame times and the OS trace are on the master clock already on both hosts `machTimestamp` on a Mac; Windows.Graphics.Capture on Windows). The field names ending in `_mach_ns` are spelled that way on every host: see [the architecture](ARCHITECTURE.md#the-master-clock). +### Locking a session + +A recording holds an exclusive lock on `recorder.lock` in its session for as long as it runs, and writes its process id +to `recorder.pid`. The OS drops the lock when the process dies, however it dies, so a lock that is held means a live +recorder and one that is free means there is none. Windows file locks are mandatory, which is why the pid is not kept in +the locked file. + +### komorebi's log and how the recorder talks to it + +| | komorebi for Mac | komorebi for Windows | +|---|---|---| +| data directory | `~/Library/Application Support/komorebi` | `%LOCALAPPDATA%\komorebi` | +| log | `komorebi_plaintext.log.` in the data directory | `%TEMP%\komorebi_plaintext.log.` | +| command and subscriber sockets | unix domain sockets in the data directory | unix domain sockets in the data directory, through `uds_windows` | +| `--log-level debug` | `komorebi --log-level debug` | `komorebi.exe --log-level debug`; `RUST_LOG=debug` also prints every dependency, so name the crate: `RUST_LOG=komorebi=debug` | + +The wire protocol is the same on both. The recorder's own binary does not depend on either komorebi's crates. + ## Degraded modes - `record --no-video` records everything but the screen, on the same clock. It asks the system nothing about displays and diff --git a/komorebi-recorder/Cargo.toml b/komorebi-recorder/Cargo.toml index 30a1ac9d..072b9166 100644 --- a/komorebi-recorder/Cargo.toml +++ b/komorebi-recorder/Cargo.toml @@ -8,6 +8,8 @@ repository.workspace = true [dependencies] clap = { workspace = true } color-eyre = { workspace = true } +crossbeam-channel = { workspace = true } +dirs = { workspace = true } schemars = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } @@ -16,6 +18,7 @@ serde_json = { workspace = true } libc = "0.2" [target.'cfg(windows)'.dependencies] +uds_windows = "1" windows = { version = "0.62", features = [ "Foundation", "Graphics", diff --git a/komorebi-recorder/src/analysis/events.rs b/komorebi-recorder/src/analysis/events.rs index ed9d3f9c..eb621225 100644 --- a/komorebi-recorder/src/analysis/events.rs +++ b/komorebi-recorder/src/analysis/events.rs @@ -19,6 +19,8 @@ //! [`super::classify`] labels each [`EventRecord`]. Tying an event to the log lines of the handler //! that produced it is done by `blocks::pair`. Both read the events this module returns. +use super::flavor::Rules; +use crate::sources::komorebi_events::EventRow; use schemars::JsonSchema; use serde::Serialize; use serde_json::Value; @@ -173,3 +175,62 @@ pub struct EventRecord { /// [`Family::Unknown`] and the name `unreadable`. pub parse_error: Option, } + +impl EventRecord { + /// Returns the record for one line of `events.jsonl`, reading its event with `rules`. + /// + /// A row with no event (its notification could not be parsed as JSON) gives + /// [`Family::Unknown`] and the name `unreadable`, not an error; the row's other fields are + /// kept. + /// + /// # Example + /// + /// ```rust + /// use komorebi_recorder::analysis::events::{EventRecord, Family}; + /// use komorebi_recorder::analysis::flavor::Flavor; + /// use komorebi_recorder::sources::komorebi_events::EventRow; + /// + /// let row: EventRow = serde_json::from_str( + /// r#"{"seq": 7, "recv_wall_us": 0, "recv_mach_ns": 30, "state_id": 4, + /// "event": {"type": "Retile"}}"#, + /// )?; + /// let record = EventRecord::from_row(&row, Flavor::Macos.rules()); + /// assert_eq!(record.seq, 7); + /// assert_eq!(record.state_id, Some(4)); + /// assert_eq!(record.info.family, Family::Command); + /// + /// // a notification that was not JSON has no event: it is unreadable, and says why + /// let row: EventRow = serde_json::from_str( + /// r#"{"seq": 8, "recv_wall_us": 0, "recv_mach_ns": 40, "parse_error": "expected value"}"#, + /// )?; + /// let record = EventRecord::from_row(&row, Flavor::Macos.rules()); + /// assert_eq!(record.info.family, Family::Unknown); + /// assert_eq!(record.info.name, "unreadable"); + /// assert_eq!(record.parse_error.as_deref(), Some("expected value")); + /// # Ok::<(), Box>(()) + /// ``` + #[must_use] + pub fn from_row(row: &EventRow, rules: &dyn Rules) -> Self { + let info = row.event.as_ref().map_or_else( + || EventInfo { + family: Family::Unknown, + name: "unreadable".to_string(), + notification: None, + pid: None, + window_id: None, + exe: None, + argument: None, + }, + |event| rules.describe_event(event), + ); + + Self { + seq: row.seq, + recv_mach_ns: row.recv_mach_ns, + recv_wall_us: row.recv_wall_us, + state_id: row.state_id, + info, + parse_error: row.parse_error.clone(), + } + } +} diff --git a/komorebi-recorder/src/incidents.rs b/komorebi-recorder/src/incidents.rs new file mode 100644 index 00000000..a4db20c5 --- /dev/null +++ b/komorebi-recorder/src/incidents.rs @@ -0,0 +1,433 @@ +//! Things that happened to the *recording environment* while recording: a display was unplugged, +//! the machine slept and woke, the screen locked, a log stream was restarted, komorebi restarted. +//! +//! They are appended to `recorder.jsonl` the moment they are noticed (and flushed, so a crash +//! loses nothing) and indexed into the timeline as the `recorder` source. That puts "display 1 +//! disconnected" or "woke after 42 s asleep" right next to the log lines and OS trace around it, +//! which is usually the first thing needed to explain a gap or a strange layout. +//! +//! The same on both hosts: which incidents a host can produce differs (the OS announces a lid +//! closing on Windows, and nothing does on macOS), but a recording from either is read the same +//! way, and the file does not say which host wrote it. +//! +//! # Who writes what +//! +//! A part of the recorder that notices something builds a [`Draft`] with a [`kind`] and a +//! sentence, and the recording loop hands it to the [`IncidentLog`], which timestamps and writes +//! it. The kinds are produced by: +//! +//! | Kinds | Noticed by | +//! |---|---| +//! | [`kind::CAPTURE_STARTED`], [`kind::CAPTURE_STOPPED`], [`kind::CAPTURE_FAILED`], [`kind::CAPTURE_RECOVERED`], the `display-*` kinds | the capture manager, which follows the displays | +//! | [`kind::SUSPEND`], [`kind::WAKE`], [`kind::LID_CLOSED`], [`kind::LID_OPENED`], [`kind::DISPLAY_ASLEEP`], [`kind::DISPLAY_AWAKE`] | `PowerTracker`, from what the OS announces | +//! | [`kind::WAKE`], [`kind::CLOCK_STEP`] | the recording loop, from the clocks disagreeing (see [`crate::clock::JumpDetector`]) | +//! | the `screen-*` and `session-*` kinds | `SessionTracker`, polling the session | +//! | [`kind::KOMOREBI_EXITED`], [`kind::KOMOREBI_STARTED`] | the recording loop, following komorebi's process | +//! | [`kind::LOG_STREAM_RESTARTED`], [`kind::LOG_STREAM_LOST`] | the OS trace | +//! | [`kind::RECORDER`] | the recorder itself, about itself | +//! +//! # The file +//! +//! `recorder.jsonl` is UTF-8, one JSON object per line ([`Incident`]), in the order they were +//! written. That is not always the order they happened in, because a [`Draft`] can be dated before +//! it was noticed. The timeline index sorts them in with every other source by `mach_ns`. +//! +//! ```json +//! {"mach_ns":104500000000,"wall_us":1800000004500000,"kind":"display-disconnected","detail":"display 1 disconnected","display":1} +//! ``` +//! +//! # Example +//! +//! ```rust +//! use komorebi_recorder::incidents::{kind, Draft, Incident, IncidentLog, INCIDENTS_FILE}; +//! +//! let dir = std::env::temp_dir().join(format!("komorebi-recorder-doc-incidents-{}", std::process::id())); +//! std::fs::create_dir_all(&dir)?; +//! +//! let mut log = IncidentLog::create(&dir)?; +//! let incident = log.record( +//! Draft::new(kind::DISPLAY_DISCONNECTED, "display 1 disconnected").display(1), +//! ); +//! +//! // the line is on disk already, without a flush or a drop +//! let text = std::fs::read_to_string(dir.join(INCIDENTS_FILE))?; +//! let on_disk: Incident = serde_json::from_str(text.trim_end())?; +//! assert_eq!(on_disk, incident); +//! assert_eq!(on_disk.kind, "display-disconnected"); +//! assert_eq!(on_disk.display, Some(1)); +//! # std::fs::remove_dir_all(&dir)?; +//! # Ok::<(), Box>(()) +//! ``` + +use crate::clock; +use serde::Deserialize; +use serde::Serialize; +use std::collections::VecDeque; +use std::fs::File; +use std::io::BufWriter; +use std::io::Write; +use std::path::Path; + +/// The file in a session directory that incidents are appended to: `recorder.jsonl`. +/// +/// See the [module docs](self) for the format of a line, which is [`Incident`]. +pub const INCIDENTS_FILE: &str = "recorder.jsonl"; +/// The name of the timeline source that incidents are indexed as: `recorder`. +/// +/// It is the `source` of every incident's entry in the timeline index, and the name a query's +/// source filter uses to ask for them. +pub const SOURCE_RECORDER: &str = "recorder"; + +/// The kinds of incident, as written to `recorder.jsonl`. +/// +/// Each constant is the `kind` of an [`Incident`], spelled in lowercase words joined by hyphens. +/// The `kind` is a plain string in the file and in [`Incident`], and a [`Draft`] takes any +/// `&'static str`: these constants are the kinds this crate writes, and keep their spellings in one +/// place, but nothing stops a writer passing another string, or a reader meeting one. +pub mod kind { + /// A capture of a display started, as a new segment of its video. `display` is set. + pub const CAPTURE_STARTED: &str = "capture-started"; + /// A capture segment was finalized for a reason that is not about the display itself. + /// + /// Its stream stalled or failed, or the machine woke and every capture is being replaced. The + /// detail says why, and which segment file it was. + pub const CAPTURE_STOPPED: &str = "capture-stopped"; + /// A capture could not be started. It is tried again with growing delays, and this is + /// reported once per outage. + pub const CAPTURE_FAILED: &str = "capture-failed"; + /// A capture started again, as a new segment, after one that had stopped on a stream failure + /// or had failed to start. + pub const CAPTURE_RECOVERED: &str = "capture-recovered"; + /// A display appeared that had not been seen during this recording (a monitor was plugged + /// in, or a laptop was docked). Its capture starts after a short settling time. + pub const DISPLAY_CONNECTED: &str = "display-connected"; + /// A display went away (unplugged, undocked, lid closed), and its video segment was finalized + /// on the spot. + pub const DISPLAY_DISCONNECTED: &str = "display-disconnected"; + /// A display's size changed, so its segment was closed and a new one begun. + pub const DISPLAY_RECONFIGURED: &str = "display-reconfigured"; + /// A display is still attached but asleep or turned off. A sleeping display counts as absent: + /// its segment was finalized, and it is not captured until it wakes. + pub const DISPLAY_ASLEEP: &str = "display-asleep"; + /// A display is available again after being asleep or gone: one that had been seen before + /// returned, or the OS said the displays were turned on. + pub const DISPLAY_AWAKE: &str = "display-awake"; + /// The machine is about to sleep or hibernate, as the OS announced it. Nothing is recorded + /// until it wakes. + pub const SUSPEND: &str = "suspend"; + /// The machine woke, either as the OS announced it or as the clocks showed it (the wall clock + /// moved on while the master clock did not). The detail says how long the machine slept. + pub const WAKE: &str = "wake"; + /// The laptop lid was closed, as the OS announced it. + pub const LID_CLOSED: &str = "lid-closed"; + /// The laptop lid was opened, as the OS announced it. + pub const LID_OPENED: &str = "lid-opened"; + /// The wall clock was stepped backwards. (A step forwards cannot be told from a sleep, and is + /// reported as [`WAKE`].) + pub const CLOCK_STEP: &str = "clock-step"; + /// The screen was locked. Video may pause or restart until it is unlocked. + pub const SCREEN_LOCKED: &str = "screen-locked"; + /// The screen was unlocked. + pub const SCREEN_UNLOCKED: &str = "screen-unlocked"; + /// Another user's session came to the front (fast user switching), so this one is in the + /// background and cannot be captured. + pub const SESSION_INACTIVE: &str = "session-inactive"; + /// This user's session is in front again. + pub const SESSION_ACTIVE: &str = "session-active"; + /// The komorebi process exited. Its events and OS trace stop until it runs again. + pub const KOMOREBI_EXITED: &str = "komorebi-exited"; + /// A komorebi process started while recording, and the recorder follows it. + pub const KOMOREBI_STARTED: &str = "komorebi-started"; + /// An OS trace stream exited and was started again. Events in the gap between are missing. + pub const LOG_STREAM_RESTARTED: &str = "log-stream-restarted"; + /// An OS trace stream, or one of the taps that make it up, could not be started or restarted: + /// the OS trace is incomplete. + pub const LOG_STREAM_LOST: &str = "log-stream-lost"; + /// A note from the recorder itself: that recording started, or that it could not do something + /// (list the displays, subscribe to komorebi, trace it). + pub const RECORDER: &str = "recorder"; +} + +/// One line of `recorder.jsonl`: something that happened to the recording environment, and when. +/// +/// An incident is written by [`IncidentLog::record`], with the time it happened on both clocks. +/// The fields are, in the order they are written, `mach_ns`, `wall_us`, `kind`, `detail`, and +/// `display` when there is one. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct Incident { + /// When it happened, in nanoseconds on the master clock (see [`crate::clock`]). That can be + /// earlier than when it was noticed. + pub mach_ns: u64, + /// When it happened by the wall clock, in microseconds since the Unix epoch, worked out from + /// how long ago `mach_ns` was. + pub wall_us: i64, + /// What kind of incident, one of the constants in [`kind`]. + pub kind: String, + /// A sentence saying what happened, for a person. + pub detail: String, + /// The display the incident is about, when it is about one. Left out of the file when there is + /// none, and `None` when it is not there. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub display: Option, +} + +/// An incident that has not been timestamped yet. The time it *happened* can be earlier than the +/// time it was noticed (a stream that died, a wake detected a moment later). +/// +/// A draft is built with [`Draft::new`] and the builder methods [`display`](Draft::display) and +/// [`at`](Draft::at), then handed to [`IncidentLog::record`], which gives it its times. +/// +/// # Example +/// +/// ```rust +/// use komorebi_recorder::incidents::{kind, Draft}; +/// +/// let draft = Draft::new(kind::CAPTURE_STOPPED, "the stream stalled") +/// .display(2) +/// .at(5_000_000_000); +/// +/// assert_eq!(draft.kind, "capture-stopped"); +/// assert_eq!(draft.display, Some(2)); +/// assert_eq!(draft.mach_ns, Some(5_000_000_000)); +/// +/// // without `at`, the incident is dated when it is recorded +/// assert_eq!(Draft::new(kind::RECORDER, "recording started").mach_ns, None); +/// ``` +/// +/// # Example: the kind is a static string +/// +/// A kind is a `&'static str`, which a constant in [`kind`] is: it is part of the file's format, +/// not something to build at run time. A `String` is refused by the compiler: +/// +/// ```rust,compile_fail,E0308 +/// use komorebi_recorder::incidents::Draft; +/// +/// let kind = format!("display-{}", "disconnected"); +/// let _draft = Draft::new(kind, "display 1 disconnected"); +/// ``` +/// +/// A constant, or any string literal, is accepted: +/// +/// ```rust +/// use komorebi_recorder::incidents::{kind, Draft}; +/// +/// let _draft = Draft::new(kind::DISPLAY_DISCONNECTED, "display 1 disconnected"); +/// ``` +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Draft { + /// The kind, one of the constants in [`kind`]. + pub kind: &'static str, + /// A sentence saying what happened. + pub detail: String, + /// The display it is about, if it is about one. + pub display: Option, + /// When it happened, in nanoseconds on the master clock, if that is not now. + pub mach_ns: Option, +} + +impl Draft { + /// Returns a draft of an incident of `kind`, described by `detail`, that is about no display + /// in particular and happened when it is recorded. + #[must_use] + pub fn new(kind: &'static str, detail: impl Into) -> Self { + Self { + kind, + detail: detail.into(), + display: None, + mach_ns: None, + } + } + + /// Returns the draft, about the display `display`. + #[must_use] + pub fn display(mut self, display: u32) -> Self { + self.display = Some(display); + self + } + + /// Returns the draft, dated `mach_ns` nanoseconds on the master clock instead of when it is + /// recorded. A time in the future is not believed: [`IncidentLog::record`] dates it now. + #[must_use] + pub fn at(mut self, mach_ns: u64) -> Self { + self.mach_ns = Some(mach_ns); + self + } +} + +/// How many recent incidents to keep for the UI. +const RECENT: usize = 8; + +/// The incident file of a recording in progress, open for appending. +/// +/// Every incident is written and flushed the moment it is recorded, because incidents are rare and +/// a crash is exactly when they matter most. The log also keeps the last few incidents in memory, +/// and a count, for the recorder window to show. +/// +/// # Example: an incident dated before it was noticed +/// +/// ```rust +/// use komorebi_recorder::clock::mono_now_ns; +/// use komorebi_recorder::incidents::{kind, Draft, IncidentLog}; +/// +/// let dir = std::env::temp_dir().join(format!("komorebi-recorder-doc-dated-{}", std::process::id())); +/// std::fs::create_dir_all(&dir)?; +/// let mut log = IncidentLog::create(&dir)?; +/// +/// // a stream that died 5 seconds ago, noticed now +/// let died = mono_now_ns() - 5_000_000_000; +/// let incident = log.record(Draft::new(kind::CAPTURE_STOPPED, "stream died").at(died)); +/// assert_eq!(incident.mach_ns, died); +/// +/// // a time in the future is not believed +/// let future = log.record(Draft::new(kind::WAKE, "woke").at(u64::MAX)); +/// assert!(future.mach_ns <= mono_now_ns()); +/// +/// assert_eq!(log.count(), 2); +/// # std::fs::remove_dir_all(&dir)?; +/// # Ok::<(), Box>(()) +/// ``` +pub struct IncidentLog { + writer: BufWriter, + recent: VecDeque, + count: u64, +} + +impl IncidentLog { + /// Creates `recorder.jsonl` in `session_dir`, ready to append to. + /// + /// An existing `recorder.jsonl` is truncated. + /// + /// # Errors + /// + /// Returns the I/O error if the file cannot be created: `session_dir` does not exist, or it + /// cannot be written to. + pub fn create(session_dir: &Path) -> std::io::Result { + Ok(Self { + writer: BufWriter::new(File::create(session_dir.join(INCIDENTS_FILE))?), + recent: VecDeque::new(), + count: 0, + }) + } + + /// Timestamps, writes and flushes an incident, and returns it as it was written. + /// + /// The incident happened at the draft's [`mach_ns`](Draft::mach_ns), or now if the draft has + /// none, and never later than now. Its wall clock time is the current wall clock time minus + /// how long ago that was. + /// + /// This does not fail. An error while writing or flushing is ignored: the incident is still + /// returned, counted and kept for [`recent`](IncidentLog::recent), but may not be in the file. + pub fn record(&mut self, draft: Draft) -> Incident { + let now_mach = clock::mono_now_ns(); + let mach_ns = draft.mach_ns.unwrap_or(now_mach).min(now_mach); + // the wall time it happened at, derived from how long ago that was + let ago_us = i64::try_from(now_mach.saturating_sub(mach_ns) / 1_000).unwrap_or(0); + let incident = Incident { + mach_ns, + wall_us: clock::wall_now_us() - ago_us, + kind: draft.kind.to_string(), + detail: draft.detail, + display: draft.display, + }; + + if let Ok(line) = serde_json::to_string(&incident) { + // incidents are rare and precious: never buffer them + let _ = writeln!(self.writer, "{line}"); + let _ = self.writer.flush(); + } + + self.count += 1; + if self.recent.len() == RECENT { + self.recent.pop_front(); + } + self.recent.push_back(incident.clone()); + incident + } + + /// Records each draft in turn, as [`record`](IncidentLog::record) does. + pub fn record_all(&mut self, drafts: impl IntoIterator) { + for draft in drafts { + self.record(draft); + } + } + + /// Returns how many incidents this log has recorded. + #[must_use] + pub fn count(&self) -> u64 { + self.count + } + + /// Returns the last 8 incidents recorded, oldest first. + #[must_use] + pub fn recent(&self) -> Vec { + self.recent.iter().cloned().collect() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn temp(name: &str) -> std::path::PathBuf { + let dir = std::env::temp_dir().join(format!( + "komorebi-recorder-incidents-{name}-{}", + std::process::id() + )); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).expect("dir"); + dir + } + + #[test] + fn incidents_are_on_disk_the_moment_they_are_recorded() { + let dir = temp("flush"); + let mut log = IncidentLog::create(&dir).expect("create"); + + log.record(Draft::new(kind::DISPLAY_DISCONNECTED, "display 1 went away").display(1)); + + // no flush or drop needed: a crash right now must not lose it + let text = std::fs::read_to_string(dir.join(INCIDENTS_FILE)).expect("read"); + let incident: Incident = + serde_json::from_str(text.lines().next().expect("line")).expect("parse"); + assert_eq!(incident.kind, "display-disconnected"); + assert_eq!(incident.display, Some(1)); + assert_eq!(log.count(), 1); + } + + #[test] + fn an_incident_can_be_dated_before_it_was_noticed() { + let dir = temp("dated"); + let mut log = IncidentLog::create(&dir).expect("create"); + + let before = clock::mono_now_ns(); + let earlier = before - 5_000_000_000; + let incident = log.record(Draft::new(kind::CAPTURE_STOPPED, "stream died").at(earlier)); + + assert_eq!(incident.mach_ns, earlier); + // the wall time is five seconds before now + let skew = (clock::wall_now_us() - incident.wall_us - 5_000_000).abs(); + assert!(skew < 100_000, "wall time was {skew} us off"); + } + + #[test] + fn an_incident_cannot_be_dated_in_the_future() { + let dir = temp("future"); + let mut log = IncidentLog::create(&dir).expect("create"); + let incident = log.record(Draft::new(kind::WAKE, "x").at(u64::MAX)); + assert!(incident.mach_ns <= clock::mono_now_ns()); + } + + #[test] + fn only_the_most_recent_incidents_are_kept_for_the_ui() { + let dir = temp("recent"); + let mut log = IncidentLog::create(&dir).expect("create"); + for index in 0..20 { + log.record(Draft::new(kind::RECORDER, format!("incident {index}"))); + } + + let recent = log.recent(); + assert_eq!(recent.len(), RECENT); + assert_eq!(recent.last().expect("last").detail, "incident 19"); + assert_eq!(log.count(), 20); + } +} diff --git a/komorebi-recorder/src/komorebi_link.rs b/komorebi-recorder/src/komorebi_link.rs new file mode 100644 index 00000000..9dbfbff7 --- /dev/null +++ b/komorebi-recorder/src/komorebi_link.rs @@ -0,0 +1,530 @@ +//! Talking to a running komorebi, on either OS, without depending on either komorebi's crates. +//! +//! komorebi for Mac and komorebi for Windows speak the same wire protocol over a unix domain +//! socket. Commands go to `komorebi.sock` in komorebi's data directory (`/komorebi`); +//! each message is one compact JSON object, `{"type": "", "content": }`, and a query +//! is answered by writing the reply back and closing the connection. To receive notifications a +//! subscriber binds a socket of its own in that directory and asks komorebi to connect to it by +//! name: komorebi then connects once per notification, writes one JSON blob and disconnects. +//! +//! Only the handful of messages the recorder sends are defined here. Sockets are `std`'s on unix +//! and those of the [`uds_windows`] crate on Windows, which has the same interface. +//! +//! # Where this fits +//! +//! This is the lowest layer that touches komorebi, and it is the same on both hosts. Which komorebi +//! is answering (the *flavor*, see [`crate::analysis::flavor`]) is not something this module +//! knows or asks. komorebi's `SocketMessage` is what [`Message`] has to agree with on the wire, +//! and [`Message`] says how that is checked. +//! +//! | Function | Used for | +//! |---|---| +//! | [`data_dir`], [`socket_path`] | finding komorebi's sockets; also the default parent of recordings | +//! | [`query_state`] | asking for komorebi's `State` (the recorder, and `doctor`) | +//! | [`send_message`] | subscribing and unsubscribing (see [`crate::sources::komorebi_events`]) | +//! | [`send_query`] | any other query, such as `doctor --probe window` | +//! | [`bind_subscriber`] | listening for komorebi's notifications | +//! +//! # Timeouts and retries +//! +//! Every read and every write on a connection to komorebi times out after one second, so a komorebi +//! that stops answering fails the exchange. Connecting has no timeout. Nothing here +//! retries, except [`query_state`] when komorebi answers with nothing. +//! +//! # Example +//! +//! The wire format is pinned by what [`Message`] serializes to: +//! +//! ```rust +//! use komorebi_recorder::komorebi_link::{Message, SubscribeOptions}; +//! +//! assert_eq!(serde_json::to_string(&Message::State)?, r#"{"type":"State"}"#); +//! +//! let subscribe = Message::AddSubscriberSocketWithOptions( +//! "komorebi-recorder-7".to_string(), +//! SubscribeOptions { filter_state_changes: false }, +//! ); +//! assert_eq!( +//! serde_json::to_string(&subscribe)?, +//! r#"{"type":"AddSubscriberSocketWithOptions","content":["komorebi-recorder-7",{"filter_state_changes":false}]}"# +//! ); +//! # Ok::<(), serde_json::Error>(()) +//! ``` +//! +//! The functions that need a komorebi, or a socket, have `no_run` examples and name the tests that +//! cover them: see [`query_state`] and [`bind_subscriber`]. +//! +//! [`uds_windows`]: https://docs.rs/uds_windows + +use color_eyre::eyre; +use color_eyre::eyre::WrapErr; +use serde::Serialize; +use std::io::Read; +use std::io::Write; +use std::net::Shutdown; +use std::path::Path; +use std::path::PathBuf; +use std::sync::OnceLock; +use std::time::Duration; + +/// The unix domain socket listener of this host: `std`'s on unix. +#[cfg(unix)] +pub use std::os::unix::net::UnixListener; +/// The unix domain socket stream of this host: `std`'s on unix. +#[cfg(unix)] +pub use std::os::unix::net::UnixStream; +/// The unix domain socket listener of this host: the `uds_windows` crate's on Windows, which has +/// the same interface as `std`'s. +#[cfg(windows)] +pub use uds_windows::UnixListener; +/// The unix domain socket stream of this host: the `uds_windows` crate's on Windows, which has the +/// same interface as `std`'s. +#[cfg(windows)] +pub use uds_windows::UnixStream; + +/// The name of the socket komorebi listens for commands on, inside the data directory. +/// +/// Its path is [`socket_path`]`(KOMOREBI_SOCKET)`. It is `komorebi.sock` on both flavors of +/// komorebi. +pub const KOMOREBI_SOCKET: &str = "komorebi.sock"; + +/// How long to wait on komorebi before giving up on one message. +const IO_TIMEOUT: Duration = Duration::from_secs(1); + +/// Returns komorebi's data directory, where its sockets and every subscriber's socket live. +/// +/// This is `/komorebi` on both OSes (`~/Library/Application Support/komorebi`, +/// `%LOCALAPPDATA%\komorebi`). If the OS has no local data directory, the directory is `komorebi` +/// inside the temporary directory. The path is worked out on the first call, and is the same for +/// the life of the process. The directory is not created and need not exist. +/// +/// A session's recordings go under it, in `recordings/`, unless told otherwise +/// (`session::default_recordings_dir`). +/// +/// # Example +/// +/// ```rust +/// use komorebi_recorder::komorebi_link::{data_dir, socket_path, KOMOREBI_SOCKET}; +/// +/// assert!(data_dir().ends_with("komorebi")); +/// assert_eq!(socket_path(KOMOREBI_SOCKET), data_dir().join("komorebi.sock")); +/// ``` +#[must_use] +pub fn data_dir() -> &'static Path { + static DATA_DIR: OnceLock = OnceLock::new(); + DATA_DIR.get_or_init(|| { + dirs::data_local_dir() + .unwrap_or_else(std::env::temp_dir) + .join("komorebi") + }) +} + +/// Returns the path of the socket called `name` in komorebi's data directory ([`data_dir`]). +/// +/// The name is joined as given, without checking that it is a plain file name, and the socket +/// need not exist. A unix socket path is limited to about 104 bytes on macOS and 108 on Windows +/// and Linux, so a long data directory leaves room for only a short `name`: `doctor` checks that +/// the recorder's subscriber socket fits. +#[must_use] +pub fn socket_path(name: &str) -> PathBuf { + data_dir().join(name) +} + +/// What a subscriber asks komorebi for. +/// +/// The default is to ask for every notification. The recorder does too: it wants every event, not +/// only the ones that changed state. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize)] +pub struct SubscribeOptions { + /// Only send notifications that changed the window manager's state. + pub filter_state_changes: bool, +} + +/// A message to komorebi, in the form komorebi reads from its socket. +/// +/// Only the messages the recorder sends are defined. The type serializes, as `{"type": ""}` +/// or `{"type": "", "content": }`, to what komorebi's `SocketMessage` reads. +/// It is never read back, because the recorder never receives one. +/// +/// komorebi is not a dependency, so nothing in the build checks that the two agree: the variant +/// names and payload shapes are kept as komorebi spells them, and the test +/// `messages_are_spelled_as_komorebi_reads_them` pins the JSON each variant produces. +/// +/// # Example +/// +/// ```rust +/// use komorebi_recorder::komorebi_link::Message; +/// +/// let json = |message: &Message| serde_json::to_string(message); +/// +/// assert_eq!(json(&Message::State)?, r#"{"type":"State"}"#); +/// assert_eq!( +/// json(&Message::RemoveSubscriberSocket("komorebi-recorder-7".to_string()))?, +/// r#"{"type":"RemoveSubscriberSocket","content":"komorebi-recorder-7"}"# +/// ); +/// assert_eq!( +/// json(&Message::DebugWindow(131_072))?, +/// r#"{"type":"DebugWindow","content":131072}"# +/// ); +/// # Ok::<(), serde_json::Error>(()) +/// ``` +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[serde(tag = "type", content = "content")] +pub enum Message { + /// Reply with the full window manager state as JSON. + State, + /// Connect to the socket named `.0` in the data directory for every notification, with the + /// options `.1`. + AddSubscriberSocketWithOptions(String, SubscribeOptions), + /// Stop connecting to the socket named `.0` in the data directory. + RemoveSubscriberSocket(String), + /// Reply with komorebi's verdict on whether it would manage the window with this handle, and + /// why (Windows). + DebugWindow(u32), +} + +/// Sends one message and does not wait for an answer. +/// +/// This connects to komorebi's command socket ([`KOMOREBI_SOCKET`]), writes the message as one JSON +/// object, and closes the connection. The write times out after one second. Use [`send_query`] for +/// a message komorebi answers. +/// +/// # Errors +/// +/// Returns the I/O error if +/// +/// * komorebi's socket cannot be connected to: komorebi is not running, or its socket cannot be +/// used by this user; +/// * the write fails or takes longer than one second (`WouldBlock` or `TimedOut`, depending on the +/// OS). +/// +/// # Example +/// +/// ```rust,no_run +/// use komorebi_recorder::komorebi_link::{send_message, Message, SubscribeOptions}; +/// +/// // ask komorebi to connect to `komorebi-recorder-7` in its data directory, per notification +/// send_message(&Message::AddSubscriberSocketWithOptions( +/// "komorebi-recorder-7".to_string(), +/// SubscribeOptions::default(), +/// ))?; +/// # Ok::<(), std::io::Error>(()) +/// ``` +/// +/// This needs komorebi running. What is sent is covered by the test +/// `messages_are_spelled_as_komorebi_reads_them`. No test covers the exchange over the socket. +pub fn send_message(message: &Message) -> std::io::Result<()> { + let mut stream = UnixStream::connect(socket_path(KOMOREBI_SOCKET))?; + stream.set_write_timeout(Some(IO_TIMEOUT))?; + stream.write_all(serde_json::to_string(message)?.as_bytes()) +} + +/// Sends one message and reads komorebi's reply to the end of the connection. +/// +/// This connects to komorebi's command socket, writes the message, shuts down the writing half so +/// that komorebi sees the end of it, and reads until komorebi closes the connection. Reads and +/// writes each time out after one second: the limit applies to each, and not to the exchange as a +/// whole. The reply is returned as komorebi wrote it, which may be empty: komorebi for Windows +/// declines a state query by answering with nothing (see [`query_state`]). +/// +/// # Errors +/// +/// Returns the I/O error if +/// +/// * komorebi's socket cannot be connected to: komorebi is not running, or this user may not use +/// its socket; +/// * a write or a read fails or takes longer than one second; +/// * the reply is not valid UTF-8 (`InvalidData`). +/// +/// # Example +/// +/// ```rust,no_run +/// use komorebi_recorder::komorebi_link::{send_query, Message}; +/// +/// let verdict = send_query(&Message::DebugWindow(131_072))?; +/// println!("komorebi says: {verdict}"); +/// # Ok::<(), std::io::Error>(()) +/// ``` +/// +/// This needs komorebi running, and no test covers the exchange over the socket. +pub fn send_query(message: &Message) -> std::io::Result { + let mut stream = UnixStream::connect(socket_path(KOMOREBI_SOCKET))?; + stream.set_read_timeout(Some(IO_TIMEOUT))?; + stream.set_write_timeout(Some(IO_TIMEOUT))?; + stream.write_all(serde_json::to_string(message)?.as_bytes())?; + stream.shutdown(Shutdown::Write)?; + + let mut response = String::new(); + stream.read_to_string(&mut response)?; + Ok(response) +} + +/// Why komorebi did not hand over its state. +/// +/// The `Display` text is a sentence fit for a warning, with no trailing period. Neither case says +/// anything about the state komorebi holds: it may be perfectly healthy. +/// +/// # Example +/// +/// ```rust +/// use komorebi_recorder::komorebi_link::StateError; +/// +/// let error = StateError::Empty { attempts: 5 }; +/// assert_eq!( +/// error.to_string(), +/// "komorebi answered the state query with nothing 5 times in a row \ +/// (busy, paused, or on another virtual desktop)" +/// ); +/// ``` +#[derive(Debug)] +pub enum StateError { + /// komorebi could not be reached at all (not running, or its socket is unreadable). The + /// error is whatever [`send_query`] returned, including a timeout. + Unreachable(std::io::Error), + /// komorebi accepted the query and answered with nothing, every time we asked. On Windows that + /// is how it declines while it is busy, paused, or showing another virtual desktop. + Empty { + /// How many times the query was made, and answered with nothing. + attempts: u32, + }, +} + +impl std::fmt::Display for StateError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + StateError::Unreachable(error) => write!(f, "could not query komorebi: {error}"), + StateError::Empty { attempts } => write!( + f, + "komorebi answered the state query with nothing {attempts} times in a row \ + (busy, paused, or on another virtual desktop)" + ), + } + } +} + +impl std::error::Error for StateError {} + +/// How many times to ask for the state when komorebi answers with nothing, and how long to wait +/// between asks. +const STATE_ATTEMPTS: u32 = 5; +const STATE_RETRY_DELAY: Duration = Duration::from_millis(200); + +/// Asks komorebi for its state, asking again when it answers with nothing. +/// +/// Returns komorebi's `State` as the JSON text komorebi sent, unparsed. A reply that is empty, or +/// only whitespace, is asked for again, up to 5 times in all, 200 ms apart. A failure to reach +/// komorebi is not asked again. +/// +/// # Errors +/// +/// Returns [`StateError::Unreachable`] if [`send_query`] fails, on any attempt, and +/// [`StateError::Empty`] if every attempt was answered with nothing. +/// +/// # Example +/// +/// ```rust,no_run +/// use komorebi_recorder::komorebi_link::{query_state, StateError}; +/// +/// match query_state() { +/// Ok(state) => println!("komorebi's state is {} bytes", state.len()), +/// Err(StateError::Unreachable(error)) => println!("not running? {error}"), +/// Err(error @ StateError::Empty { .. }) => println!("{error}"), +/// } +/// ``` +/// +/// This needs komorebi running. The retry rule is covered by the tests +/// `an_empty_state_reply_is_asked_again`, +/// `a_komorebi_that_only_ever_answers_with_nothing_is_an_error_that_says_so` and +/// `an_unreachable_komorebi_is_not_asked_again`, which supply the answers themselves. No test covers +/// the exchange over the socket. +pub fn query_state() -> Result { + query_state_with( + || send_query(&Message::State), + STATE_ATTEMPTS, + STATE_RETRY_DELAY, + ) +} + +fn query_state_with( + mut ask: impl FnMut() -> std::io::Result, + attempts: u32, + delay: Duration, +) -> Result { + for attempt in 1..=attempts { + let response = ask().map_err(StateError::Unreachable)?; + if !response.trim().is_empty() { + return Ok(response); + } + + if attempt < attempts { + std::thread::sleep(delay); + } + } + + Err(StateError::Empty { attempts }) +} + +/// Binds a listener for komorebi's notifications at `path`, replacing a stale socket file left by +/// a subscriber that died. +/// +/// Whatever is at `path` is removed first. A socket another live process is listening on is taken +/// over, not refused: that process keeps its listener, which nothing can reach by name. This +/// function does not remove the socket file afterwards; the caller does. A subscriber's `path` is +/// [`socket_path`] of the name it then gives komorebi in [`Message::AddSubscriberSocketWithOptions`]. +/// +/// # Errors +/// +/// Returns an error if +/// +/// * something at `path` cannot be removed, for any reason other than there being nothing there; +/// * the socket cannot be bound: the directory does not exist, this user may not write to it, or +/// `path` is longer than a unix socket path may be (about 104 bytes on macOS, 108 elsewhere). +/// +/// # Example +/// +/// ```rust,no_run +/// use komorebi_recorder::komorebi_link::{bind_subscriber, socket_path}; +/// +/// let listener = bind_subscriber(&socket_path("komorebi-recorder-7"))?; +/// // komorebi connects here once per notification, writes one JSON blob and disconnects +/// let (_stream, _) = listener.accept()?; +/// # Ok::<(), Box>(()) +/// ``` +/// +/// This waits for komorebi, so it is not run. It is covered by the test +/// `a_subscriber_socket_receives_what_is_written_to_it`, which binds over a stale file, connects a +/// writer to the socket, and reads what was written. +pub fn bind_subscriber(path: &Path) -> eyre::Result { + match std::fs::remove_file(path) { + Ok(()) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => return Err(error).wrap_err("could not clear a stale subscriber socket"), + } + + UnixListener::bind(path).wrap_err_with(|| format!("could not listen on {}", path.display())) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn messages_are_spelled_as_komorebi_reads_them() { + let json = |message: &Message| serde_json::to_string(message).expect("serialize"); + + assert_eq!(json(&Message::State), r#"{"type":"State"}"#); + assert_eq!( + json(&Message::AddSubscriberSocketWithOptions( + "komorebi-recorder-7".to_string(), + SubscribeOptions { + filter_state_changes: false + } + )), + r#"{"type":"AddSubscriberSocketWithOptions","content":["komorebi-recorder-7",{"filter_state_changes":false}]}"# + ); + assert_eq!( + json(&Message::RemoveSubscriberSocket( + "komorebi-recorder-7".into() + )), + r#"{"type":"RemoveSubscriberSocket","content":"komorebi-recorder-7"}"# + ); + assert_eq!( + json(&Message::DebugWindow(131_072)), + r#"{"type":"DebugWindow","content":131072}"# + ); + } + + #[test] + fn the_data_dir_is_komorebis() { + assert!(data_dir().ends_with("komorebi")); + assert_eq!(socket_path("x.sock"), data_dir().join("x.sock")); + } + + #[test] + fn an_empty_state_reply_is_asked_again() { + let mut replies = vec![ + Ok(String::new()), + Ok(" \n".to_string()), + Ok(r#"{"monitors":{}}"#.to_string()), + ] + .into_iter(); + + let state = query_state_with( + || replies.next().expect("asked too often"), + 5, + Duration::ZERO, + ) + .expect("state"); + assert_eq!(state, r#"{"monitors":{}}"#); + } + + #[test] + fn a_komorebi_that_only_ever_answers_with_nothing_is_an_error_that_says_so() { + let mut asked = 0; + let error = query_state_with( + || { + asked += 1; + Ok(String::new()) + }, + 3, + Duration::ZERO, + ) + .expect_err("nothing to return"); + + assert_eq!(asked, 3); + assert!( + matches!(error, StateError::Empty { attempts: 3 }), + "{error}" + ); + assert!(error.to_string().contains("another virtual desktop")); + } + + #[test] + fn an_unreachable_komorebi_is_not_asked_again() { + let mut asked = 0; + let error = query_state_with( + || { + asked += 1; + Err(std::io::Error::from(std::io::ErrorKind::NotFound)) + }, + 5, + Duration::ZERO, + ) + .expect_err("unreachable"); + + assert_eq!(asked, 1); + assert!(matches!(error, StateError::Unreachable(_))); + } + + #[test] + fn a_subscriber_socket_receives_what_is_written_to_it() { + // a unix socket path is limited to ~104 bytes, which a nix shell's $TMPDIR already exceeds + let base = if cfg!(unix) { + PathBuf::from("/tmp") + } else { + std::env::temp_dir() + }; + let dir = base.join(format!("kl-{}", std::process::id())); + std::fs::create_dir_all(&dir).expect("dir"); + let path = dir.join("sub"); + + // binding over a stale file from a dead subscriber works + std::fs::write(&path, b"stale").expect("stale"); + let listener = bind_subscriber(&path).expect("bind"); + + let writer = std::thread::spawn({ + let path = path.clone(); + move || { + let mut stream = UnixStream::connect(path).expect("connect"); + stream.write_all(b"{\"a\":1}").expect("write"); + } + }); + + let (mut stream, _) = listener.accept().expect("accept"); + let mut bytes = Vec::new(); + stream.read_to_end(&mut bytes).expect("read"); + writer.join().expect("writer"); + let _ = std::fs::remove_dir_all(&dir); + assert_eq!(bytes, br#"{"a":1}"#); + } +} diff --git a/komorebi-recorder/src/lib.rs b/komorebi-recorder/src/lib.rs index 9cd6824a..3fc0f631 100644 --- a/komorebi-recorder/src/lib.rs +++ b/komorebi-recorder/src/lib.rs @@ -43,6 +43,10 @@ //! | [`clock`] | the master clock, anchors and the map between the wall clock and the master clock | //! | [`platform`] | everything that differs by host, behind one facade | //! | [`analysis`] | understanding a session without a window, starting with the komorebi flavors | +//! | [`komorebi_link`] | talking to a running komorebi over its socket | +//! | [`incidents`] | what happened to the recording itself, as a log of its own | +//! | [`lock`] | telling a running recorder from one that died | +//! | [`sources`] | the sources that write raw files into the session | //! //! # What is not supported //! @@ -57,5 +61,9 @@ pub mod analysis; pub mod clock; +pub mod incidents; +pub mod komorebi_link; +pub mod lock; pub mod platform; +pub mod sources; pub mod timecode; diff --git a/komorebi-recorder/src/lock.rs b/komorebi-recorder/src/lock.rs new file mode 100644 index 00000000..9b17b47c --- /dev/null +++ b/komorebi-recorder/src/lock.rs @@ -0,0 +1,280 @@ +//! Telling a recorder that is running from one that died. +//! +//! A session's manifest says `recording` both while a recorder is alive and after one was killed, +//! and the right thing to do differs: wait, or `reindex`. The recorder therefore holds an exclusive +//! lock on `recorder.lock` for as long as it runs. The OS drops the lock the instant the process +//! dies, however it dies, so a lock that is still held means a live recorder and one that is free +//! means there is none. +//! +//! The pid lives in a file of its own, `recorder.pid`. Windows file locks are mandatory, so a +//! reader could not read a pid out of the very file that is locked; keeping the lock file empty +//! works the same on every OS. +//! +//! # The two files +//! +//! | File | Holds | +//! |---|---| +//! | [`LOCK_FILE`], `recorder.lock` | nothing: it is empty, and what matters is whether a lock is held on it | +//! | [`PID_FILE`], `recorder.pid` | the recorder's process id as decimal text and a newline, replaced by each recorder that takes the lock | +//! +//! The lock is [`File::try_lock`](std::fs::File::try_lock), which is `flock` on unix and +//! `LockFileEx` on Windows. It is held by an open file, not recorded as a pid, so it cannot go +//! stale. +//! +//! # Who uses this +//! +//! A recorder takes the lock with [`RecorderLock::acquire`] when it begins to record. Anything that +//! meets a session whose manifest says `recording` asks [`recorder`] whether that is a recorder at +//! work (wait for it) or one that died (`reindex` the session). This module is the same on every +//! host. +//! +//! # Example +//! +//! ```rust +//! use komorebi_recorder::lock::{recorder, Recorder, RecorderLock}; +//! +//! let dir = std::env::temp_dir().join(format!("komorebi-recorder-doc-lock-{}", std::process::id())); +//! std::fs::create_dir_all(&dir)?; +//! +//! // no lock file: a session that predates it, or no session at all +//! assert_eq!(recorder(&dir), Recorder::Unknown); +//! +//! // a second recorder cannot take a session that one is recording +//! let lock = RecorderLock::acquire(&dir)?; +//! assert_eq!(recorder(&dir), Recorder::Running(Some(std::process::id()))); +//! assert!(RecorderLock::acquire(&dir).is_err()); +//! +//! // the lock is released when the recorder is dropped, or dies +//! drop(lock); +//! assert_eq!(recorder(&dir), Recorder::NotRunning); +//! # std::fs::remove_dir_all(&dir)?; +//! # Ok::<(), Box>(()) +//! ``` + +use std::fs::File; +use std::fs::OpenOptions; +use std::fs::TryLockError; +use std::path::Path; + +/// The file in a session directory that a recording holds a lock on: `recorder.lock`. It is empty. +pub const LOCK_FILE: &str = "recorder.lock"; +/// The file in a session directory that has the recorder's process id: `recorder.pid`. +pub const PID_FILE: &str = "recorder.pid"; + +/// The lock a running recorder holds on its session; released when dropped or when the process +/// ends. +/// +/// There is no way to hold it without a recorder: the value is made by [`RecorderLock::acquire`], +/// and keeps the lock file open for as long as it lives. Its only field is private, so no other +/// code can make one around a file of its choosing: +/// +/// ```rust,compile_fail,E0451 +/// use komorebi_recorder::lock::RecorderLock; +/// +/// let file = std::fs::File::create("recorder.lock").expect("a file"); +/// let _lock = RecorderLock { _file: file }; +/// ``` +#[derive(Debug)] +pub struct RecorderLock { + _file: File, +} + +impl RecorderLock { + /// Takes the lock for the session in `dir`, and writes this process's id next to it. + /// + /// The lock file is created if it does not exist. The pid file is only written once the lock + /// is held, so it is never replaced by a process that did not get the lock. + /// + /// # Errors + /// + /// Returns the I/O error if + /// + /// * the lock file cannot be created or opened: `dir` does not exist, or cannot be written to; + /// * another recorder holds the lock: the kind is [`WouldBlock`](std::io::ErrorKind::WouldBlock) + /// and the message is `another recorder is already recording this session`; + /// * the lock cannot be taken for another reason, such as a file system that does not support + /// locks; + /// * the pid file cannot be written. The lock is released again in that case. + /// + /// # Example + /// + /// ```rust + /// use komorebi_recorder::lock::RecorderLock; + /// + /// let dir = std::env::temp_dir().join(format!("komorebi-recorder-doc-acquire-{}", std::process::id())); + /// std::fs::create_dir_all(&dir)?; + /// + /// let _first = RecorderLock::acquire(&dir)?; + /// let error = RecorderLock::acquire(&dir).unwrap_err(); + /// assert_eq!(error.kind(), std::io::ErrorKind::WouldBlock); + /// assert_eq!(error.to_string(), "another recorder is already recording this session"); + /// # std::fs::remove_dir_all(&dir)?; + /// # Ok::<(), Box>(()) + /// ``` + pub fn acquire(dir: &Path) -> std::io::Result { + let file = OpenOptions::new() + .create(true) + .truncate(false) + .write(true) + .open(dir.join(LOCK_FILE))?; + + match file.try_lock() { + Ok(()) => {} + Err(TryLockError::WouldBlock) => { + return Err(std::io::Error::new( + std::io::ErrorKind::WouldBlock, + "another recorder is already recording this session", + )); + } + Err(TryLockError::Error(error)) => return Err(error), + } + + // only now that the lock is ours is it safe to replace what the previous owner wrote + std::fs::write(dir.join(PID_FILE), format!("{}\n", std::process::id()))?; + Ok(Self { _file: file }) + } +} + +/// What can be told about the recorder of a session from its lock. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Recorder { + /// A recorder holds the lock; the pid it wrote, if readable. + Running(Option), + /// The lock file exists and nobody holds it. + NotRunning, + /// There is no lock file: the session predates it. + Unknown, +} + +/// Returns whether a recorder is at work on the session in `dir`, by asking whether its lock is +/// held. +/// +/// The lock is tried with a shared lock, and let go of at once. [`Recorder::Running`] therefore +/// means that some open file holds the lock, which includes one held by the calling process. +/// [`Recorder::NotRunning`] means the lock file is there and free, and is also what a lock that +/// cannot be queried at all (a file system without locks, say) reads as: that is no evidence of a +/// live recorder, and calling a dead session live would make `reindex` refuse it. +/// [`Recorder::Unknown`] is for a lock file that is missing, or cannot be opened. +/// +/// A recorder that was killed leaves its lock file and pid file behind, but not the lock, so the +/// session reads as [`NotRunning`](Recorder::NotRunning). +/// +/// This never fails and never blocks. It is a snapshot: the answer can be out of date by the time +/// it is read. +/// +/// # Example +/// +/// ```rust +/// use komorebi_recorder::lock::{recorder, Recorder}; +/// +/// let dir = std::env::temp_dir().join(format!("komorebi-recorder-doc-recorder-{}", std::process::id())); +/// std::fs::create_dir_all(&dir)?; +/// assert_eq!(recorder(&dir), Recorder::Unknown); +/// # std::fs::remove_dir_all(&dir)?; +/// # Ok::<(), Box>(()) +/// ``` +/// +/// What each answer is for a live recorder, a dead one and a session with no lock file is covered +/// by the tests `the_lock_is_held_while_the_recorder_lives_and_free_after`, +/// `a_session_without_a_lock_file_is_unknown` and +/// `a_pid_written_into_the_lock_file_by_an_older_build_is_still_read`. +#[must_use] +pub fn recorder(dir: &Path) -> Recorder { + let path = dir.join(LOCK_FILE); + let Ok(file) = File::open(&path) else { + return Recorder::Unknown; + }; + + match file.try_lock_shared() { + Ok(()) => { + let _ = file.unlock(); + Recorder::NotRunning + } + // anything but "someone holds it" (a filesystem without locks, say) is not evidence of a + // live recorder, and calling a dead session live would make `reindex` refuse it + Err(TryLockError::Error(_)) => Recorder::NotRunning, + Err(TryLockError::WouldBlock) => Recorder::Running(pid(dir)), + } +} + +/// The pid a recorder wrote: in `recorder.pid`, or else in the lock file itself, where a recorder +/// that wrote the pid there left it (readable only on an OS whose locks are advisory). +fn pid(dir: &Path) -> Option { + [PID_FILE, LOCK_FILE].iter().find_map(|name| { + std::fs::read_to_string(dir.join(name)) + .ok() + .and_then(|text| text.trim().parse().ok()) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn temp(name: &str) -> std::path::PathBuf { + let dir = std::env::temp_dir().join(format!( + "komorebi-recorder-lock-{name}-{}", + std::process::id() + )); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).expect("dir"); + dir + } + + #[test] + fn a_session_without_a_lock_file_is_unknown() { + assert_eq!(recorder(&temp("none")), Recorder::Unknown); + } + + #[test] + fn the_lock_is_held_while_the_recorder_lives_and_free_after() { + let dir = temp("held"); + let lock = RecorderLock::acquire(&dir).expect("acquire"); + + assert_eq!(recorder(&dir), Recorder::Running(Some(std::process::id()))); + + drop(lock); + // a process spawned at that moment by another test briefly holds a copy of the descriptor, + // and a lock lasts until every copy of the descriptor is closed + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + while recorder(&dir) != Recorder::NotRunning && std::time::Instant::now() < deadline { + std::thread::sleep(std::time::Duration::from_millis(10)); + } + assert_eq!(recorder(&dir), Recorder::NotRunning); + } + + #[test] + fn the_pid_is_kept_out_of_the_locked_file() { + let dir = temp("pidfile"); + let _lock = RecorderLock::acquire(&dir).expect("acquire"); + + // Windows would refuse to read a file that is locked, so the pid cannot live in it + assert_eq!( + std::fs::read_to_string(dir.join(LOCK_FILE)).expect("lock file"), + "" + ); + assert_eq!( + std::fs::read_to_string(dir.join(PID_FILE)) + .expect("pid file") + .trim(), + std::process::id().to_string() + ); + } + + #[test] + fn a_pid_written_into_the_lock_file_by_an_older_build_is_still_read() { + let dir = temp("legacy"); + let lock = File::create(dir.join(LOCK_FILE)).expect("lock file"); + std::fs::write(dir.join(LOCK_FILE), "4242\n").expect("legacy pid"); + lock.try_lock().expect("held"); + + assert_eq!(recorder(&dir), Recorder::Running(Some(4242))); + } + + #[test] + fn two_recorders_cannot_share_a_session() { + let dir = temp("twice"); + let _first = RecorderLock::acquire(&dir).expect("first"); + assert!(RecorderLock::acquire(&dir).is_err()); + } +} diff --git a/komorebi-recorder/src/sources/komorebi_events.rs b/komorebi-recorder/src/sources/komorebi_events.rs new file mode 100644 index 00000000..39bd2813 --- /dev/null +++ b/komorebi-recorder/src/sources/komorebi_events.rs @@ -0,0 +1,665 @@ +//! komorebi's subscriber-socket notifications: every window manager event together with the full +//! `State` it produced. +//! +//! The daemon pushes each notification as one JSON blob over a fresh unix-socket connection to a +//! listener the recorder binds, and connects *synchronously* per subscriber while holding its own +//! locks, so a slow subscriber stalls komorebi. The accept thread therefore does the bare minimum +//! (stamp the receive time, read to EOF, hand the bytes to a channel) and all parsing and disk IO +//! happens on a separate writer thread. +//! +//! The payload carries no timestamp, so the receive time is stamped here on both clocks. komorebi +//! sends the notification *after* its handler has run, so the receive time marks the **end** of +//! handling the event, not its start: in komorebi for Mac's log the `processing event` line is +//! when the handler began (and, at DEBUG, the `pushed notification to subscriber` line is when it +//! sent this). `komorebi-recorder trace` ties the two together. +//! +//! A notification is a JSON object with an `event` (what komorebi was told or did) and the `state` +//! (komorebi's whole `State` after it). This module is the same on both hosts and for both flavors +//! of komorebi. It stores the payload as komorebi sent it and does not read it: what an event +//! means is the business of the [`Rules`](crate::analysis::flavor::Rules) of the flavor that +//! recorded it (see [`crate::analysis::events`]). +//! +//! # Files written +//! +//! | file | contents | +//! |---|---| +//! | `events.jsonl` | one [`EventRow`] per line, one line per notification, in arrival order | +//! | `states/.json` | komorebi's `State` as compact JSON, stored only when it differs from the last one stored; named by [`state_file_name`] | +//! | `state-start.json`, `state-end.json` | the `State` komorebi answered to a query at the start and the end of the recording ([`snapshot_state`]) | +//! +//! A line of `events.jsonl`, for a window manager event on komorebi for Mac: +//! +//! ```json +//! {"seq":7,"recv_wall_us":1793016000123456,"recv_mach_ns":5000000000,"event":{"content":[{"content":"AXWindowMoved","type":"Accessibility"},20934,3395],"type":"MoveEnd"},"state_id":2} +//! ``` +//! +//! The file is flushed whenever the writer has caught up with the notifications that have arrived, +//! so a recording that is killed loses only what the writer had not yet caught up on. +//! +//! # Threads +//! +//! [`EventRecorder`] runs two threads. The accept thread (`komorebi-events-accept`) numbers and +//! stamps each notification and does nothing else. The writer thread (`komorebi-events-write`) +//! parses it, stores a new `State`, tells the app discovery about it ([`Discovery`]), runs the +//! window prober when one was given (`--probe-windows`, see `winprobe`) and appends the row. The +//! prober never runs on the accept thread: komorebi delivers a notification while holding its own +//! locks, so a query sent to komorebi from the accept thread could wait on a lock that is itself +//! waiting for the notification to be accepted. + +use crate::clock; +use crate::komorebi_link; +use crate::komorebi_link::Message; +use crate::komorebi_link::SubscribeOptions; +use crate::komorebi_link::UnixListener; +use crate::komorebi_link::UnixStream; +use color_eyre::eyre; +use color_eyre::eyre::WrapErr; +use crossbeam_channel::Receiver; +use crossbeam_channel::Sender; +use serde::Deserialize; +use serde::Serialize; +use serde_json::Value; +use std::fs::File; +use std::io::BufWriter; +use std::io::Read; +use std::io::Write; +use std::path::Path; +use std::path::PathBuf; +use std::sync::Arc; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::AtomicU64; +use std::sync::atomic::Ordering; +use std::thread::JoinHandle; +use std::time::Duration; + +/// How long to keep accepting after asking komorebi to forget the recorder. +const UNSUBSCRIBE_DRAIN: Duration = Duration::from_millis(250); + +fn subscribe_to_komorebi(name: &str) -> std::io::Result<()> { + // unfiltered: every event is wanted, not just the ones that changed state + komorebi_link::send_message(&Message::AddSubscriberSocketWithOptions( + name.to_string(), + SubscribeOptions { + filter_state_changes: false, + }, + )) +} + +/// A notification as it arrived, before any parsing. +struct RawNotification { + seq: u64, + recv_wall_us: i64, + recv_mach_ns: u64, + bytes: Vec, +} + +/// One line of `events.jsonl`: a notification as komorebi sent it, with the time it arrived on +/// both clocks. +/// +/// The file is JSON Lines: one object per line, one line per notification, in the order the +/// recorder accepted them. `seq`, `recv_wall_us` and `recv_mach_ns` are always present. Every +/// other key is left out of the line when it has no value, and reads back as `None` when absent. +/// +/// A row is one of two shapes, and `event` and `parse_error` are never both present: +/// +/// - **read**: the payload was valid JSON. `event` is its `event` value (`None` if the payload had +/// no `event` key) and `parse_error` and `raw` are absent. +/// - **unreadable**: the payload was not valid JSON. `event` is absent, `parse_error` gives the +/// reason and `raw` keeps the payload as text, so a notification that cannot be read is not lost. +/// +/// # Example +/// +/// A row is one line of text, and reads back as it was written: +/// +/// ```rust +/// use komorebi_recorder::sources::komorebi_events::EventRow; +/// +/// let line = r#"{"seq":7,"recv_wall_us":1793016000123456,"recv_mach_ns":5000000000,"event":{"type":"Show","content":[null,1]},"state_id":2}"#; +/// let row: EventRow = serde_json::from_str(line)?; +/// assert_eq!((row.seq, row.state_id), (7, Some(2))); +/// assert_eq!(row.event.as_ref().and_then(|event| event["type"].as_str()), Some("Show")); +/// +/// // keys without a value are left out of the line: there is no `parse_error`, `raw` or `probe` +/// let written = serde_json::to_string(&row)?; +/// assert!(!written.contains("parse_error") && !written.contains("raw") && !written.contains("probe")); +/// assert_eq!(serde_json::from_str::(&written)?.seq, 7); +/// # Ok::<(), Box>(()) +/// ``` +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct EventRow { + /// The notification's place in arrival order, counting from 1 for each recorder. The accept + /// thread numbers every non-empty notification it receives. + pub seq: u64, + /// When the notification was accepted, in microseconds since the Unix epoch on the wall clock. + pub recv_wall_us: i64, + /// When the notification was accepted, in nanoseconds on the host's master monotonic clock + /// (see [`crate::clock::mono_now_ns`]). komorebi notifies after its handler has run, so this + /// is when handling the event ended. + pub recv_mach_ns: u64, + /// The `event` value of the payload, re-serialized from the parsed JSON (komorebi's key order + /// and whitespace are not kept). Its shape is komorebi's and differs by flavor: a window + /// manager event, a command or a monitor change. Absent in an unreadable row. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub event: Option, + /// The id of the stored state snapshot (`states/.json`) in effect after this event: the + /// most recent snapshot stored at or before this row. `None` until the first snapshot is + /// stored. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub state_id: Option, + /// Why the payload was not valid JSON, as `serde_json` words it. Present only in an + /// unreadable row. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub parse_error: Option, + /// The payload as text, with any invalid UTF-8 replaced. Present only in an unreadable row. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub raw: Option, +} + +/// Something the app-discovery logic wants to know about as it happens, sent by the writer thread. +pub enum Discovery { + /// The state changed: this is the `State` that was just stored as a new snapshot. + State(Arc), + /// A window manager event mentioned this process id (see `managed_apps::event_pid`). + Pid(i32), +} + +/// Running totals of what the writer thread has handled, readable from any thread while the +/// recording goes on. +/// +/// Each counter only grows and is updated with relaxed ordering, so a reader sees a recent value, +/// not a value that is consistent with the others. [`EventSummary`] is these totals at the end. +#[derive(Debug, Default)] +pub struct EventCounters { + /// Notifications the writer has dequeued, readable or not. + pub events: AtomicU64, + /// `State` snapshots the writer has stored in `states/`. + pub states: AtomicU64, + /// Bytes of payload in those notifications. + pub bytes: AtomicU64, + /// Notifications whose payload was not valid JSON. + pub parse_errors: AtomicU64, +} + +/// The totals of an [`EventCounters`], as the manifest (`session.json`, key `events`) stores them. +/// +/// The recorder stores the totals so far at each checkpoint and the final totals when it stops; +/// [`EventRecorder::stop`] returns them. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)] +pub struct EventSummary { + /// Notifications handled, readable or not: the number of lines in `events.jsonl`. + pub events: u64, + /// `State` snapshots stored: the number of files in `states/`. + pub state_snapshots: u64, + /// Bytes of payload received, before parsing. + pub bytes: u64, + /// Notifications whose payload was not valid JSON (their rows have `parse_error`). + pub parse_errors: u64, +} + +/// Returns the path of the file that stores state snapshot `id`, relative to the session +/// directory: `states/.json`, with `` zero padded to six digits. +/// +/// Ids count from 1 in the order the snapshots were stored, so file names sort in that order up to +/// 999999 snapshots; an id with more digits is not padded further. +/// +/// # Example +/// +/// ```rust +/// use komorebi_recorder::sources::komorebi_events::state_file_name; +/// +/// assert_eq!(state_file_name(1), "states/000001.json"); +/// assert_eq!(state_file_name(123_456), "states/123456.json"); +/// ``` +#[must_use] +pub fn state_file_name(id: u64) -> String { + format!("states/{id:06}.json") +} + +/// Records komorebi's subscriber notifications into a session directory until it is stopped. +/// +/// A recorder binds a listener socket in komorebi's data directory, asks komorebi to send it every +/// notification, and writes `events.jsonl` and `states/` (see [`crate::sources::komorebi_events`] +/// for the files and the threads). It registers under the name `komorebi-recorder-`, which +/// komorebi's DEBUG log repeats when it pushes a notification, so an event can be tied to its +/// handler. +/// +/// The subscription asks for every event, not only the ones that changed komorebi's state. +/// +/// Stopping, or dropping a recorder that was not stopped, asks komorebi to forget the subscriber +/// and then keeps accepting for a moment before it closes the listener. If komorebi was in the +/// middle of a notification when it was told to forget the recorder, a reader that had already gone +/// would make its write fail, and that failure aborts delivery of the event to komorebi's other +/// subscribers (a status bar, for one). +/// +/// # Example +/// +/// Recording needs a komorebi socket and a session directory, so this example only compiles. The +/// writer's behavior is covered by `the_writer_dedupes_unchanged_states_and_tolerates_garbage` +/// and `the_writer_stores_what_the_prober_read_beside_the_event_it_is_about` in this file's unit +/// tests. The accept thread and the subscription handshake have no unit test: they are exercised +/// by recording against a running komorebi. +/// +/// ```rust,no_run +/// use komorebi_recorder::sources::komorebi_events::EventRecorder; +/// use std::path::Path; +/// +/// let recorder = EventRecorder::start(Path::new("session"), None)?; +/// if !recorder.is_subscribed() { +/// // komorebi was not running: register again once it is +/// let _ = recorder.subscribe(); +/// } +/// +/// // ... record ... +/// +/// let summary = recorder.stop(); +/// println!("{} events, {} states", summary.events, summary.state_snapshots); +/// # Ok::<(), Box>(()) +/// ``` +pub struct EventRecorder { + name: String, + socket: PathBuf, + subscribed: Arc, + stop: Arc, + counters: Arc, + accept: Option>, + writer: Option>, +} + +impl EventRecorder { + /// Listens for komorebi's notifications, writing `events.jsonl` and `states/` into `dir`, and + /// subscribes to komorebi if it is running. + /// + /// komorebi not running is not an error: the listener is bound regardless, so that when it + /// starts (or restarts, which forgets every subscriber) a later [`subscribe`](Self::subscribe) + /// picks the notifications up without losing the listener or the events file. + /// [`is_subscribed`](Self::is_subscribed) reports whether the first attempt succeeded. + /// + /// # Errors + /// + /// Returns an error if + /// + /// - `dir/states` cannot be created, + /// - the listener cannot be bound: a stale socket file of an earlier subscriber that cannot be + /// removed, or komorebi's data directory does not exist (as when komorebi has never run) or + /// cannot be written to (a unix socket path is also limited in length, to about 104 bytes on + /// macOS), + /// - `dir/events.jsonl` cannot be created, or + /// - a thread cannot be spawned. + /// + /// An error from the subscription itself is not one of them. + pub fn start(dir: &Path, discovery: Option>) -> eyre::Result { + std::fs::create_dir_all(dir.join("states"))?; + + let name = format!("komorebi-recorder-{}", std::process::id()); + let socket = komorebi_link::socket_path(&name); + let listener = komorebi_link::bind_subscriber(&socket)?; + + let subscribed = Arc::new(AtomicBool::new(subscribe_to_komorebi(&name).is_ok())); + + let events_file = File::create(dir.join("events.jsonl"))?; + let (tx, rx) = crossbeam_channel::unbounded::(); + let stop = Arc::new(AtomicBool::new(false)); + let counters = Arc::new(EventCounters::default()); + + let accept_stop = stop.clone(); + let accept = std::thread::Builder::new() + .name("komorebi-events-accept".to_string()) + .spawn(move || accept_loop(&listener, &tx, &accept_stop))?; + + let writer_counters = counters.clone(); + let states_dir = dir.join("states"); + let writer = std::thread::Builder::new() + .name("komorebi-events-write".to_string()) + .spawn(move || { + write_loop( + &rx, + events_file, + &states_dir, + &writer_counters, + discovery.as_ref(), + ) + })?; + + Ok(Self { + name, + socket, + subscribed, + stop, + counters, + accept: Some(accept), + writer: Some(writer), + }) + } + + /// Returns the live counters, shared with the writer thread. They keep counting until the + /// recorder is stopped. + #[must_use] + pub fn counters(&self) -> Arc { + self.counters.clone() + } + + /// The name this recorder registers under with komorebi, `komorebi-recorder-` for this + /// process. It names the recorder's socket in komorebi's data directory and is what komorebi's + /// DEBUG log says when it pushes a notification here. + #[must_use] + pub fn subscriber_name(&self) -> &str { + &self.name + } + + /// Returns `true` if and only if the most recent attempt to register with komorebi succeeded. + /// + /// This is the outcome of the last [`start`](Self::start) or [`subscribe`](Self::subscribe), + /// not a live check: a komorebi that has restarted has forgotten the recorder, and this stays + /// `true` until the next call to [`subscribe`](Self::subscribe) says otherwise. + #[must_use] + pub fn is_subscribed(&self) -> bool { + self.subscribed.load(Ordering::Relaxed) + } + + /// (Re-)registers with komorebi. Safe to call at any time; komorebi forgets its subscribers + /// when it restarts, so the recorder calls this when it sees a new komorebi. + /// + /// # Errors + /// + /// Returns an error if komorebi's socket cannot be connected to (komorebi is not running), or + /// the message cannot be written within one second. [`is_subscribed`](Self::is_subscribed) + /// then reads `false`. + pub fn subscribe(&self) -> std::io::Result<()> { + let result = subscribe_to_komorebi(&self.name); + self.subscribed.store(result.is_ok(), Ordering::Relaxed); + result + } + + /// Unsubscribes, drains everything already received, closes the listener and returns the + /// totals. + /// + /// This takes about a quarter of a second longer than the writer needs: the recorder keeps + /// accepting for that long after asking komorebi to forget it (see [`EventRecorder`]). The + /// socket file is removed. When this returns, every notification accepted has been written to + /// `events.jsonl` and flushed. + pub fn stop(mut self) -> EventSummary { + self.shutdown(); + EventSummary { + events: self.counters.events.load(Ordering::Relaxed), + state_snapshots: self.counters.states.load(Ordering::Relaxed), + bytes: self.counters.bytes.load(Ordering::Relaxed), + parse_errors: self.counters.parse_errors.load(Ordering::Relaxed), + } + } + + fn shutdown(&mut self) { + // tell komorebi first so it stops connecting while we wind down + let _ = komorebi_link::send_message(&Message::RemoveSubscriberSocket(self.name.clone())); + + // Keep reading for a moment. If komorebi was mid-notification when we unsubscribed, a + // reader that has already gone away would make its write fail, which aborts delivery of + // that event to its *other* subscribers (e.g. the bar). Draining lets it finish cleanly. + std::thread::sleep(UNSUBSCRIBE_DRAIN); + + self.stop.store(true, Ordering::SeqCst); + // wake the blocking accept so it observes the stop flag + let _ = UnixStream::connect(&self.socket); + + if let Some(accept) = self.accept.take() { + let _ = accept.join(); + } + + // the accept thread owned the only sender, so the writer drains and ends + if let Some(writer) = self.writer.take() { + let _ = writer.join(); + } + + let _ = std::fs::remove_file(&self.socket); + } +} + +impl Drop for EventRecorder { + fn drop(&mut self) { + if self.accept.is_some() || self.writer.is_some() { + self.shutdown(); + } + } +} + +fn accept_loop(listener: &UnixListener, tx: &Sender, stop: &AtomicBool) { + let mut seq = 0_u64; + + loop { + let (mut stream, _) = match listener.accept() { + Ok(accepted) => accepted, + Err(_) if stop.load(Ordering::SeqCst) => return, + Err(_) => { + std::thread::sleep(Duration::from_millis(10)); + continue; + } + }; + + // stamp before anything else: this is the number that matters + let recv_mach_ns = clock::mono_now_ns(); + let recv_wall_us = clock::wall_now_us(); + + if stop.load(Ordering::SeqCst) { + return; + } + + let _ = stream.set_read_timeout(Some(Duration::from_secs(2))); + let mut bytes = Vec::with_capacity(64 * 1024); + let _ = stream.read_to_end(&mut bytes); + if bytes.is_empty() { + continue; + } + + seq += 1; + if tx + .send(RawNotification { + seq, + recv_wall_us, + recv_mach_ns, + bytes, + }) + .is_err() + { + return; + } + } +} + +fn write_loop( + rx: &Receiver, + events_file: File, + states_dir: &Path, + counters: &EventCounters, + discovery: Option<&Sender>, +) { + let mut events = BufWriter::new(events_file); + let mut previous_state: Option = None; + let mut state_id = 0_u64; + + while let Ok(raw) = rx.recv() { + counters.events.fetch_add(1, Ordering::Relaxed); + counters + .bytes + .fetch_add(raw.bytes.len() as u64, Ordering::Relaxed); + + let row = match serde_json::from_slice::(&raw.bytes) { + Ok(mut value) => { + let event = value.get_mut("event").map(Value::take); + let state = value.get_mut("state").map(Value::take); + + if let Some(state) = state + && previous_state.as_ref() != Some(&state) + { + state_id += 1; + write_state(states_dir, state_id, &state); + counters.states.fetch_add(1, Ordering::Relaxed); + + if let Some(discovery) = discovery { + let _ = discovery.send(Discovery::State(Arc::new(state.clone()))); + } + + previous_state = Some(state); + } + + EventRow { + seq: raw.seq, + recv_wall_us: raw.recv_wall_us, + recv_mach_ns: raw.recv_mach_ns, + event, + state_id: (state_id > 0).then_some(state_id), + parse_error: None, + raw: None, + } + } + Err(error) => { + counters.parse_errors.fetch_add(1, Ordering::Relaxed); + EventRow { + seq: raw.seq, + recv_wall_us: raw.recv_wall_us, + recv_mach_ns: raw.recv_mach_ns, + event: None, + state_id: (state_id > 0).then_some(state_id), + parse_error: Some(error.to_string()), + raw: Some(String::from_utf8_lossy(&raw.bytes).into_owned()), + } + } + }; + + if let Ok(line) = serde_json::to_string(&row) { + let _ = writeln!(events, "{line}"); + } + + // events are sparse; keep the file current so a crash loses nothing + if rx.is_empty() { + let _ = events.flush(); + } + } + + let _ = events.flush(); +} + +fn write_state(states_dir: &Path, id: u64, state: &Value) { + let path = states_dir.join(format!("{id:06}.json")); + if let Ok(text) = serde_json::to_string(state) { + let _ = std::fs::write(path, text); + } +} + +/// Asks komorebi for its current `State`, writes the reply to `path` as komorebi sent it, and +/// returns the parsed value. +/// +/// An empty reply is asked for again a few times (see [`crate::komorebi_link::query_state`]). The +/// file is written only after the reply has been parsed, so a reply that is not JSON leaves no +/// file behind. This is how the recorder writes `state-start.json` and `state-end.json`. +/// +/// This talks to a running komorebi, so it has no doctest. The query and its retries are covered +/// by `an_empty_state_reply_is_asked_again` in [`crate::komorebi_link`]'s unit tests. +/// +/// # Errors +/// +/// Returns an error if +/// +/// - komorebi cannot be reached (it is not running, or its socket cannot be read), +/// - komorebi answers with nothing every time it is asked (on komorebi for Windows it does that +/// while it is busy, paused, or showing another virtual desktop), +/// - the reply is not valid JSON, or +/// - `path` cannot be written. +/// +/// Asking again later is the workaround for the first two. +pub fn snapshot_state(path: &Path) -> eyre::Result { + let response = komorebi_link::query_state().map_err(|error| eyre::Report::new(error))?; + let value: Value = + serde_json::from_str(&response).wrap_err("komorebi returned invalid state JSON")?; + std::fs::write(path, &response)?; + Ok(value) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn state_ids_map_to_ordered_file_names() { + assert_eq!(state_file_name(1), "states/000001.json"); + assert_eq!(state_file_name(123_456), "states/123456.json"); + } + + #[test] + fn event_rows_roundtrip_and_skip_empty_fields() { + let row = EventRow { + seq: 3, + recv_wall_us: 10, + recv_mach_ns: 20, + event: Some(serde_json::json!({"type": "Show"})), + state_id: Some(2), + parse_error: None, + raw: None, + }; + let text = serde_json::to_string(&row).expect("serialize"); + assert!(!text.contains("parse_error")); + let back: EventRow = serde_json::from_str(&text).expect("parse"); + assert_eq!(back.seq, 3); + assert_eq!(back.state_id, Some(2)); + } + + #[test] + fn the_writer_dedupes_unchanged_states_and_tolerates_garbage() { + let dir = + std::env::temp_dir().join(format!("komorebi-recorder-events-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(dir.join("states")).expect("dir"); + + let (tx, rx) = crossbeam_channel::unbounded(); + let notification = |seq: u64, state: &str| RawNotification { + seq, + recv_wall_us: seq as i64, + recv_mach_ns: seq * 1000, + bytes: format!(r#"{{"event":{{"type":"Show","content":[null,1]}},"state":{state}}}"#) + .into_bytes(), + }; + + tx.send(notification(1, r#"{"a":1}"#)).expect("send"); + tx.send(notification(2, r#"{"a":1}"#)).expect("send"); + tx.send(notification(3, r#"{"a":2}"#)).expect("send"); + tx.send(RawNotification { + seq: 4, + recv_wall_us: 4, + recv_mach_ns: 4000, + bytes: b"not json".to_vec(), + }) + .expect("send"); + drop(tx); + + let counters = EventCounters::default(); + write_loop( + &rx, + File::create(dir.join("events.jsonl")).expect("file"), + &dir.join("states"), + &counters, + None, + ); + + assert_eq!(counters.events.load(Ordering::Relaxed), 4); + assert_eq!( + counters.states.load(Ordering::Relaxed), + 2, + "an unchanged state is not stored twice" + ); + assert_eq!(counters.parse_errors.load(Ordering::Relaxed), 1); + + let rows: Vec = std::fs::read_to_string(dir.join("events.jsonl")) + .expect("events") + .lines() + .map(|line| serde_json::from_str(line).expect("row")) + .collect(); + assert_eq!( + rows.iter().map(|row| row.state_id).collect::>(), + vec![Some(1), Some(1), Some(2), Some(2)] + ); + assert!(rows[3].parse_error.is_some()); + assert_eq!(rows[3].raw.as_deref(), Some("not json")); + + assert!(dir.join("states/000001.json").exists()); + assert!(dir.join("states/000002.json").exists()); + assert!(!dir.join("states/000003.json").exists()); + } +} diff --git a/komorebi-recorder/src/sources/mod.rs b/komorebi-recorder/src/sources/mod.rs new file mode 100644 index 00000000..5c870fba --- /dev/null +++ b/komorebi-recorder/src/sources/mod.rs @@ -0,0 +1,34 @@ +//! The non-video capture sources: what komorebi and the operating system said while a recording +//! ran. +//! +//! Each source writes its own raw files into the session directory and never rewrites them +//! afterwards. The merged index over them (`timeline.jsonl`) only points into these files, so the +//! file names and record formats written here are part of the session format: a session recorded +//! by any build is read by every other, and a session recorded on one host is investigated on the +//! other. +//! +//! | module | files it writes | recorded on | +//! |---|---|---| +//! | [`komorebi_events`] | `events.jsonl` (one JSON object per line), `states/.json` | both hosts | +//! +//! The types that a source reports about itself to the manifest (`session.json`) are defined +//! beside it: `EventSummary`, `LogSegment`, `StreamInfo`, `KomorebiVersion` and `AppRecord`. +//! +//! # What this layer does not do +//! +//! - It does not understand komorebi's vocabulary. A source copies bytes and stamps times; reading +//! an event or a log line is the job of [`crate::analysis`], through the rules of the +//! [`Flavor`](crate::analysis::flavor::Flavor) that the manifest records. +//! - It does not index or merge. `timeline` builds the index when a recording ends, and `reindex` +//! can rebuild it from these files alone. +//! - It does not decide when to start or stop. The `recorder` starts every source, checkpoints +//! their counters into the manifest and stops them in order. +//! +//! # Clocks +//! +//! A source stamps what it receives on the master clock ([`crate::clock`]) wherever komorebi's own +//! text carries no time of its own. The `*_mach_ns` names in the records keep their historical +//! spelling: they hold nanoseconds on the host's master monotonic clock, and the manifest's +//! `host.clock` says which clock that was. + +pub mod komorebi_events;