mirror of
https://github.com/LGUG2Z/komorebi.git
synced 2026-11-15 13:22:19 +01:00
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.
This commit is contained in:
LGUG2Z committed
1 parent
0de0f82840
commit
c748237741
12 files changed
+2454
No files matched your search
Generated
+397
@@ -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"
|
||||
|
||||
@@ -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",
|
||||
[
|
||||
|
||||
@@ -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 |
|
||||
@@ -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.<UTC date>` in the data directory | `%TEMP%\komorebi_plaintext.log.<UTC date>` |
|
||||
| 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
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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<String>,
|
||||
}
|
||||
|
||||
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<dyn std::error::Error>>(())
|
||||
/// ```
|
||||
#[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(),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<dyn std::error::Error>>(())
|
||||
//! ```
|
||||
|
||||
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<u32>,
|
||||
}
|
||||
|
||||
/// 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<u32>,
|
||||
/// When it happened, in nanoseconds on the master clock, if that is not now.
|
||||
pub mach_ns: Option<u64>,
|
||||
}
|
||||
|
||||
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<String>) -> 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<dyn std::error::Error>>(())
|
||||
/// ```
|
||||
pub struct IncidentLog {
|
||||
writer: BufWriter<File>,
|
||||
recent: VecDeque<Incident>,
|
||||
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<Self> {
|
||||
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<Item = Draft>) {
|
||||
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<Incident> {
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -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 (`<local data dir>/komorebi`);
|
||||
//! each message is one compact JSON object, `{"type": "<Name>", "content": <payload>}`, 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 `<local data dir>/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<PathBuf> = 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": "<Variant>"}`
|
||||
/// or `{"type": "<Variant>", "content": <payload>}`, 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<String> {
|
||||
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<String, StateError> {
|
||||
query_state_with(
|
||||
|| send_query(&Message::State),
|
||||
STATE_ATTEMPTS,
|
||||
STATE_RETRY_DELAY,
|
||||
)
|
||||
}
|
||||
|
||||
fn query_state_with(
|
||||
mut ask: impl FnMut() -> std::io::Result<String>,
|
||||
attempts: u32,
|
||||
delay: Duration,
|
||||
) -> Result<String, StateError> {
|
||||
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<dyn std::error::Error>>(())
|
||||
/// ```
|
||||
///
|
||||
/// 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<UnixListener> {
|
||||
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}"#);
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
@@ -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<dyn std::error::Error>>(())
|
||||
//! ```
|
||||
|
||||
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<dyn std::error::Error>>(())
|
||||
/// ```
|
||||
pub fn acquire(dir: &Path) -> std::io::Result<Self> {
|
||||
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<u32>),
|
||||
/// 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<dyn std::error::Error>>(())
|
||||
/// ```
|
||||
///
|
||||
/// 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<u32> {
|
||||
[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());
|
||||
}
|
||||
}
|
||||
@@ -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/<id>.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<u8>,
|
||||
}
|
||||
|
||||
/// 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::<EventRow>(&written)?.seq, 7);
|
||||
/// # Ok::<(), Box<dyn std::error::Error>>(())
|
||||
/// ```
|
||||
#[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<Value>,
|
||||
/// The id of the stored state snapshot (`states/<id>.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<u64>,
|
||||
/// 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<String>,
|
||||
/// 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<String>,
|
||||
}
|
||||
|
||||
/// 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<Value>),
|
||||
/// 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/<id>.json`, with `<id>` 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-<pid>`, 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<dyn std::error::Error>>(())
|
||||
/// ```
|
||||
pub struct EventRecorder {
|
||||
name: String,
|
||||
socket: PathBuf,
|
||||
subscribed: Arc<AtomicBool>,
|
||||
stop: Arc<AtomicBool>,
|
||||
counters: Arc<EventCounters>,
|
||||
accept: Option<JoinHandle<()>>,
|
||||
writer: Option<JoinHandle<()>>,
|
||||
}
|
||||
|
||||
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<Sender<Discovery>>) -> eyre::Result<Self> {
|
||||
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::<RawNotification>();
|
||||
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<EventCounters> {
|
||||
self.counters.clone()
|
||||
}
|
||||
|
||||
/// The name this recorder registers under with komorebi, `komorebi-recorder-<pid>` 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<RawNotification>, 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<RawNotification>,
|
||||
events_file: File,
|
||||
states_dir: &Path,
|
||||
counters: &EventCounters,
|
||||
discovery: Option<&Sender<Discovery>>,
|
||||
) {
|
||||
let mut events = BufWriter::new(events_file);
|
||||
let mut previous_state: Option<Value> = 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::<Value>(&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<Value> {
|
||||
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<EventRow> = 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<_>>(),
|
||||
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());
|
||||
}
|
||||
}
|
||||
@@ -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/<id>.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;
|
||||
Reference in new issue
Block a user