diff --git a/frontend/package-lock.json b/frontend/package-lock.json index c25caf7d..8a078b6a 100644 --- a/frontend/package-lock.json +++ b/frontend/package-lock.json @@ -2331,9 +2331,9 @@ } }, "node_modules/baseline-browser-mapping": { - "version": "2.10.42", - "resolved": "https://registry.npmjs.org/baseline-browser-mapping/-/baseline-browser-mapping-2.10.42.tgz", - "integrity": "sha512-c/jurFrDLyui7o1J86yLkRu4LMsTYcBohveus7/I2Hzdn9KIP2bdJPTue/lR1KH46enoPbD77GKeSYNdyPoD3Q==", + "version": "2.11.23", + "resolved": "https://registry.npmjs.org/baseline-browser-mapping/-/baseline-browser-mapping-2.11.23.tgz", + "integrity": "sha512-le521dGVfxM7yRX0EikCoSz+rOK+hHzdDt/E7mG1jOJB/6WAAUuwVroLwaB7ApaUsz5Q0kFlDXLSA9MheUIfRQ==", "dev": true, "license": "Apache-2.0", "bin": { @@ -2367,9 +2367,9 @@ } }, "node_modules/browserslist": { - "version": "4.28.5", - "resolved": "https://registry.npmjs.org/browserslist/-/browserslist-4.28.5.tgz", - "integrity": "sha512-Cu2E6QejHWzuDMTkuwgpABFgDfZrXLQq5V13YOACZx4mFAG4IwGTbTfHPMr4WtxlHoXSM8FIuRwYYCz5XiabaQ==", + "version": "4.28.9", + "resolved": "https://registry.npmjs.org/browserslist/-/browserslist-4.28.9.tgz", + "integrity": "sha512-EWazOblFYUvlGZcfGhPUPmYh3nikUxBVb+y9MJun5f3hBi812X+8MSQTujLBtgK3cf51fJWbWfOjyeO954d+Eg==", "dev": true, "funding": [ { @@ -2387,11 +2387,11 @@ ], "license": "MIT", "dependencies": { - "baseline-browser-mapping": "^2.10.42", - "caniuse-lite": "^1.0.30001800", - "electron-to-chromium": "^1.5.387", - "node-releases": "^2.0.50", - "update-browserslist-db": "^1.2.3" + "baseline-browser-mapping": "^2.11.20", + "caniuse-lite": "^1.0.30001810", + "electron-to-chromium": "^1.5.420", + "node-releases": "^2.0.54", + "update-browserslist-db": "^1.3.2" }, "bin": { "browserslist": "cli.js" @@ -2401,9 +2401,9 @@ } }, "node_modules/caniuse-lite": { - "version": "1.0.30001803", - "resolved": "https://registry.npmjs.org/caniuse-lite/-/caniuse-lite-1.0.30001803.tgz", - "integrity": "sha512-g/uHREV2ZpK9qMalCsWaxmA6ol+DX8GYhuf3T40RKoP+oL7vhRJh8LNt73PCjpnR6l14FzfPrB5Yux4PKm2meg==", + "version": "1.0.30001810", + "resolved": "https://registry.npmjs.org/caniuse-lite/-/caniuse-lite-1.0.30001810.tgz", + "integrity": "sha512-TITQPUkaz+aVk5GL6NhOdwk1aEaNTSDPsGFWrTuhKGtjTF70jL/Oht2W4c6rXUe5fu7Ie19VIahAXHIIiWWNeg==", "dev": true, "funding": [ { @@ -2545,9 +2545,9 @@ "license": "MIT" }, "node_modules/electron-to-chromium": { - "version": "1.5.389", - "resolved": "https://registry.npmjs.org/electron-to-chromium/-/electron-to-chromium-1.5.389.tgz", - "integrity": "sha512-cEto7aeOqBfU1D+c5py5pE+ooscKE75JifxLBdFUZsqAxRS6y7kebtxAZvICszSl05gPjYHDTjY+lXpyGvpJbg==", + "version": "1.5.427", + "resolved": "https://registry.npmjs.org/electron-to-chromium/-/electron-to-chromium-1.5.427.tgz", + "integrity": "sha512-n14zb3FdsChZ2BNobqNHAJMcP3ifFv4paox2LvCrfVAQcqGiSURgbJl+PfMpHVCNFkStnNc+RRVtPBTVW5PDgw==", "dev": true, "license": "ISC" }, @@ -3395,9 +3395,9 @@ "license": "MIT" }, "node_modules/node-releases": { - "version": "2.0.50", - "resolved": "https://registry.npmjs.org/node-releases/-/node-releases-2.0.50.tgz", - "integrity": "sha512-J6l92tKHX6w8Jy5nO1Vuc01NoIiRGi/d6qBKVxh+IQ8Cr3b6HbVNfKiF8ZpFKufTwpwxMmce2W3iQZ861ZRyTg==", + "version": "2.0.55", + "resolved": "https://registry.npmjs.org/node-releases/-/node-releases-2.0.55.tgz", + "integrity": "sha512-mIrE/Cw9y+9Au6dS5vDKDhQza9YvG6w+ZrS6X+ZzA7yFW/soAeaups4Qzn1bL6g5FVy8WtP79+0j82oPIbqRjQ==", "dev": true, "license": "MIT", "engines": { @@ -4025,9 +4025,9 @@ "license": "MIT" }, "node_modules/update-browserslist-db": { - "version": "1.2.3", - "resolved": "https://registry.npmjs.org/update-browserslist-db/-/update-browserslist-db-1.2.3.tgz", - "integrity": "sha512-Js0m9cx+qOgDxo0eMiFGEueWztz+d4+M3rGlmKPT+T4IS/jP4ylw3Nwpu6cpTTP8R1MAC1kF4VbdLt3ARf209w==", + "version": "1.3.3", + "resolved": "https://registry.npmjs.org/update-browserslist-db/-/update-browserslist-db-1.3.3.tgz", + "integrity": "sha512-pJ2sYawQS0R/WI928Gj5GlPhTGzbMelq0+4INtSYNDV9ErKJcX6xjGWkoG/VnB3dpUm00zALaqkrUD77pO5TDQ==", "dev": true, "funding": [ { diff --git a/src/app/state.cpp b/src/app/state.cpp index 43aae48a..904bbc9a 100644 --- a/src/app/state.cpp +++ b/src/app/state.cpp @@ -553,10 +553,40 @@ void AppState::start_engine_impl(InputHook* hook) { try { capture_.start(hook); engine_thread_ = std::thread([this] { - while (engine_running_.load(std::memory_order_relaxed)) { + // True when the last tick stopped draining on a budget. The backlog is then worked + // off at full speed across short ticks rather than in one long critical section -- + // the sleep below is what makes "bounded drain" a pause for other threads instead + // of a throughput ceiling. + // + // It is also part of the exit condition, not just the pacing: an unbounded drain + // always left the ring empty, so shutdown persisted everything captured. A bounded + // one does not, and stopping the moment the flag flips would silently discard + // whatever the last slice did not reach. stop_engine() stops the producer before + // joining this thread, so continuing while a backlog remains terminates -- the + // queue is finite and nothing is refilling it. + // + // A do-while, not a while: stop_engine() can flip the flag before this thread is + // ever scheduled, and a plain loop would then exit having drained nothing at all + // -- losing whatever capture queued in between. One tick always runs. + // + // The exit condition asks the ring rather than trusting `backlog`, which describes + // the tick that has already happened. A thread asleep between ticks when the flag + // flips has a stale `false` from before the events arrived, and would exit over a + // full queue -- which is precisely what CI caught on macOS. + bool backlog = false; + // Ticks spent draining after a stop was requested. A tick that throws reports no + // backlog but also drains nothing, so without this a permanently failing tick over + // a non-empty ring would spin here and shutdown would never complete. The whole + // ring is kEngineDrainBudget * 32 events, so this cannot cut a healthy drain short. + int shutdown_ticks = 0; + constexpr int kMaxShutdownTicks = 64; + do { try { - engine_tick(); + backlog = engine_tick(); } catch (const std::exception& error) { + // A tick that threw tells us nothing about the queue; treat it as no + // backlog so a failing tick cannot keep shutdown spinning here. + backlog = false; try { std::ostringstream message; message << "engine tick failed: " << error.what(); @@ -566,14 +596,27 @@ void AppState::start_engine_impl(InputHook* hook) { // unhandled exception on this thread. } } catch (...) { + backlog = false; try { log().error("engine tick failed: unknown exception"); } catch (...) { // Keep the thread boundary intact even if the logger fails. } } - std::this_thread::sleep_for(std::chrono::milliseconds(100)); - } + // Checked before sleeping so a stop with an empty queue exits now rather than + // waiting out a tick interval nobody is waiting for. + const bool stopping = !engine_running_.load(std::memory_order_relaxed); + if (stopping && ++shutdown_ticks >= kMaxShutdownTicks) break; + // Only the shutdown path consults the ring. Pacing stays on `backlog` alone: + // an ordinary tick usually leaves a few events queued behind it, and treating + // that as urgent would run the loop at 1 ms forever for a handful of + // keystrokes. + if (stopping && !backlog && !capture_.has_pending_events()) break; + std::this_thread::sleep_for(std::chrono::milliseconds( + (backlog || stopping) ? kEngineBacklogTickIntervalMs + : kEngineTickIntervalMs)); + } while (engine_running_.load(std::memory_order_relaxed) || + backlog || capture_.has_pending_events()); }); } catch (...) { engine_running_.store(false, std::memory_order_release); @@ -589,8 +632,7 @@ void AppState::set_emit_hook(EmitHook hook) { void AppState::stop_engine() noexcept { engine_running_.store(false, std::memory_order_relaxed); - maintenance_stopping_.store(true, std::memory_order_release); - maintenance_ready_.notify_all(); + signal_maintenance([this] { maintenance_stopping_.store(true, std::memory_order_release); }); capture_.stop(); if (engine_thread_.joinable()) engine_thread_.join(); if (maintenance_thread_.joinable()) maintenance_thread_.join(); @@ -661,8 +703,9 @@ SessionRecord AppState::start_session(const std::string& goal, FocusMode mode) { storage_.begin_session_span_now(created.session_id); savepoint.release(); } catch (...) { - maintenance_paused_.store(replaced.has_value(), std::memory_order_release); - maintenance_ready_.notify_all(); + signal_maintenance([this, &replaced] { + maintenance_paused_.store(replaced.has_value(), std::memory_order_release); + }); throw; } @@ -732,8 +775,8 @@ void AppState::stop_session() { session_attended_ = false; pomodoro_.reset(); active_session_.reset(); - maintenance_paused_.store(false, std::memory_order_release); - maintenance_ready_.notify_all(); + signal_maintenance( + [this] { maintenance_paused_.store(false, std::memory_order_release); }); features_.reset_for_session(std::nullopt); context_tracker_.reset(); // Same reason as start_session: the payload names a window from the session being @@ -770,8 +813,8 @@ SessionRecord AppState::stop_session(const std::string& session_id) { pomodoro_.reset(); session_attended_ = false; active_session_.reset(); - maintenance_paused_.store(false, std::memory_order_release); - maintenance_ready_.notify_all(); + signal_maintenance( + [this] { maintenance_paused_.store(false, std::memory_order_release); }); features_.reset_for_session(std::nullopt); context_tracker_.reset(); // Inside the active-session branch on purpose, unlike the pending-span drop above. @@ -813,14 +856,18 @@ bool AppState::delete_session(const std::string& session_id) { // a missing foreign key, and the UI would keep rendering a session the user just // erased. Reset exactly what stop_session() resets, plus the derived prediction state. if (active_session_ && active_session_->session_id == session_id) { + // Queued events were captured *for the session being erased*, so they go with it. + // Only in this branch: deleting some other session leaves a queue that belongs to the + // session the user is still running, and that queue must survive. + capture_.discard_pending_events(); pomodoro_.reset(); // Its spans went with the row, so there is nothing left to close and nothing to // reconcile against. Leaving this true would make the next tick's level check see a // change that is not there. session_attended_ = false; active_session_.reset(); - maintenance_paused_.store(false, std::memory_order_release); - maintenance_ready_.notify_all(); + signal_maintenance( + [this] { maintenance_paused_.store(false, std::memory_order_release); }); features_.reset_for_session(std::nullopt); context_tracker_.reset(); context_tracker_.set_goal_categories(settings_.goal_categories); @@ -1009,6 +1056,12 @@ ActivityDeletionResult AppState::delete_all_activity_data() { std::lock_guard activity_lock(activity_boundary_mutex_); std::lock_guard store_lock(storage_mutex_); activity_epoch_.fetch_add(1, std::memory_order_release); + // The epoch fences rows this tick was about to write; it does not touch what capture has + // already queued. Those events were recorded before the user asked for deletion, so a + // later tick filing them -- into a session started after this point, no less -- would put + // pre-deletion window titles back on disk. Deleting activity means deleting the activity + // still in flight too. (Bounded drop: see CaptureThread::discard_pending_events.) + capture_.discard_pending_events(); ActivityDeletionResult result; @@ -1040,8 +1093,7 @@ ActivityDeletionResult AppState::delete_all_activity_data() { for (const char* retained : kRetainedArtifacts) result.retained.emplace_back(retained); active_session_.reset(); - maintenance_paused_.store(false, std::memory_order_release); - maintenance_ready_.notify_all(); + signal_maintenance([this] { maintenance_paused_.store(false, std::memory_order_release); }); session_attended_ = false; // every span was deleted with the rows above discard_pending_span_unlocked(); // and there is no session left for one to name latest_prediction_.reset(); @@ -1758,11 +1810,11 @@ void AppState::process_event_for_test(const CaptureEvent& event) { void AppState::request_retention_maintenance() { if (maintenance_stopping_.load(std::memory_order_acquire)) return; - bool expected = false; - if (maintenance_pending_.compare_exchange_strong(expected, true, - std::memory_order_acq_rel)) { - maintenance_ready_.notify_all(); - } + // Notifies even when the flag was already set: the wake is unconditional now, and a + // spurious one costs the worker a single predicate evaluation. + signal_maintenance([this] { + maintenance_pending_.store(true, std::memory_order_release); + }); } void AppState::run_retention_maintenance() noexcept { @@ -1846,7 +1898,7 @@ void AppState::run_retention_maintenance() noexcept { } } -void AppState::engine_tick() { +bool AppState::engine_tick() { // Three phases with different locks so a disk write never blocks an ordinary UI read: // 1) drain + classify under mutex_ (in-memory only), collecting persist jobs; // 2) flush them under storage_mutex_ in ONE transaction, after releasing mutex_; @@ -1878,13 +1930,61 @@ void AppState::engine_tick() { // The tick only schedules retention. An owned worker performs bounded storage batches. bool prune_due = false; std::uint64_t tick_activity_epoch = 0; + // Whether phase 1 gave up on a budget rather than on an empty ring. Returned to the engine + // loop, which then re-ticks immediately instead of sleeping out the tick interval. + bool drain_truncated = false; + // Set under mutex_ when the throttle allows a backlog line; written to the log after the + // lock is released. + std::optional backlog_to_log; { std::lock_guard lock(mutex_); tick_activity_epoch = activity_epoch_.load(std::memory_order_acquire); bool had_input = false; - while (auto ev = capture_.next_event()) { + // Bounded on purpose. `while (capture_.next_event())` only ends when the consumer + // outruns the producer, so under sustained input this loop -- and mutex_ with it -- + // was held for as long as the typing lasted, starving every command thread and + // deferring the idle poll, the persistence flush, and the UI emissions below. + // Leftover events stay in the ring (single consumer, so nobody else takes them) and + // are picked up by the next tick, which follows in kEngineBacklogTickIntervalMs. + std::size_t drained = 0; + const auto drain_started_ms = steady_now_ms(); + while (drained < kEngineDrainBudget) { + auto ev = capture_.next_event(); + if (!ev) break; + ++drained; if (is_input_event(ev->event_type)) had_input = true; if (auto job = compute_event(*ev)) jobs.push_back(std::move(*job)); + // The count budget is the primary bound and needs no clock, which keeps ticks + // driven by a ManualClock deterministic. This second bound covers the case where + // per-event work is heavy enough that 2,048 of them is already too long. + if (drained % kEngineDrainClockCheckStride == 0 && + steady_now_ms() - drain_started_ms >= kEngineDrainBudgetMs) { + drain_truncated = true; + break; + } + } + // Spending the budget is not the same as leaving work behind: a burst of exactly + // kEngineDrainBudget events is fully consumed, and reporting a backlog for it would + // buy an extra 1 ms tick and a log line describing a queue that is empty. Ask the ring + // instead of inferring it from the counter. + if (drained >= kEngineDrainBudget && capture_.has_pending_events()) { + drain_truncated = true; + } + if (drain_truncated) { + // Throttled: while a backlog lasts this tick runs every millisecond, and one line + // per tick would bury the log it is meant to explain. The decision is made here, + // under mutex_; the write happens after the lock is released, because the logger + // formats and writes to its sink synchronously and a slow disk would otherwise + // extend exactly the critical section this function exists to bound. + const auto log_now_ms = steady_now_ms(); + // Not a zero sentinel: steady_ms() is relative to an arbitrary epoch, so 0 is a + // legitimate reading (ManualClock is routinely set to it), and comparing against + // it would log on every tick for as long as the clock sat there. + if (!last_drain_backlog_log_ms_ || + log_now_ms - *last_drain_backlog_log_ms_ >= kEngineBacklogLogIntervalMs) { + last_drain_backlog_log_ms_ = log_now_ms; + backlog_to_log = drained; + } } // Idle timing runs off the tick's monotonic clock, not event timestamps: true AFK // means no events arrive at all, so we must measure wall time, not the last event. @@ -1942,13 +2042,20 @@ void AppState::engine_tick() { publish_live_read_unlocked(); } + if (backlog_to_log) { + log().info("engine: capture backlog, drained " + std::to_string(*backlog_to_log) + + " events this tick and yielded the state lock"); + } + if (prune_due) request_retention_maintenance(); { // If deletion won the boundary after phase 1, discard every buffered row and // event. If this tick won, deletion waits until persistence has completed. std::lock_guard activity_lock(activity_boundary_mutex_); - if (tick_activity_epoch != activity_epoch_.load(std::memory_order_acquire)) return; + if (tick_activity_epoch != activity_epoch_.load(std::memory_order_acquire)) { + return drain_truncated; + } if (!jobs.empty() || span_session_id) { std::lock_guard lock(storage_mutex_); Storage::Transaction txn(storage_); // one commit for the whole drain @@ -1964,7 +2071,7 @@ void AppState::engine_tick() { } } - if (!hook) return; + if (!hook) return drain_truncated; if (idle_edge == IdleTransition::WentIdle) { hook("idle", "{\"idle\":true}", tick_activity_epoch); } @@ -2014,6 +2121,7 @@ void AppState::engine_tick() { {"delivery", nlohmann::json(untracked_route)}}), tick_activity_epoch); } + return drain_truncated; } std::optional AppState::compute_event(const CaptureEvent& event) { diff --git a/src/app/state.hpp b/src/app/state.hpp index 51491daa..7901026c 100644 --- a/src/app/state.hpp +++ b/src/app/state.hpp @@ -46,6 +46,27 @@ inline constexpr std::int64_t kRetentionPruneIntervalMs = 24 * 60 * 60 * 1000; inline constexpr std::size_t kRetentionPruneBatchRows = 256; inline constexpr std::int64_t kRetentionPruneYieldMs = 10; +// How many capture events one tick may process while holding mutex_. The ring holds 65,536 +// events and the drain used to run until it observed an empty buffer, so a producer that kept +// up with the consumer made the critical section as long as the user kept typing -- and every +// session command, settings write, idle poll, persistence flush, and UI emission waited behind +// it. Hitting this ceiling drops nothing: the rest of the ring stays queued for the next tick, +// which the engine loop runs immediately instead of sleeping. +inline constexpr std::size_t kEngineDrainBudget = 2048; +// Wall-clock ceiling on the same drain, for when per-event cost (window extraction, ONNX +// inference) makes even the event budget too many. Checked every kEngineDrainClockCheckStride +// events rather than per event so the drain does not pay for a virtual clock call each time. +inline constexpr std::int64_t kEngineDrainBudgetMs = 20; +inline constexpr std::size_t kEngineDrainClockCheckStride = 128; +// Gap between ticks. The backlog value is deliberately not zero: it is what guarantees mutex_ +// is actually released long enough for a waiting command thread to take it between two +// bounded drains. +inline constexpr std::int64_t kEngineTickIntervalMs = 100; +inline constexpr std::int64_t kEngineBacklogTickIntervalMs = 1; +// A saturated drain is normal for a moment and a symptom if it persists, so it is logged -- +// but the loop above runs every millisecond while it lasts, hence the throttle. +inline constexpr std::int64_t kEngineBacklogLogIntervalMs = 30'000; + class AppState { public: // `logger` and `clock` are both optional (default null) so existing call sites keep @@ -336,9 +357,31 @@ class AppState { std::optional snapback_episode; }; - void engine_tick(); // features -> classifier -> tracker -> (emit) ; persist off-lock + // features -> classifier -> tracker -> (emit) ; persist off-lock. + // Returns true when the drain stopped on kEngineDrainBudget / kEngineDrainBudgetMs rather + // than on an empty ring — i.e. work is still queued and the caller should tick again now + // instead of sleeping out the usual interval. + bool engine_tick(); void request_retention_maintenance(); void run_retention_maintenance() noexcept; + + // Every change to the maintenance flags goes through here. + // + // The worker blocks on maintenance_ready_ with a predicate over those flags. Storing a + // flag and calling notify_all() *without* maintenance_mutex_ held lets the notification + // land in the window after the worker has evaluated the predicate and before it is + // actually blocked on the condition variable -- a lost wakeup. For `stopping` that is not + // a delay, it is a hang: the worker never wakes, so the join() in stop_engine() never + // returns and the process cannot exit. It reproduced as one random AppState test per CI + // job dying on a 120 s timeout, on whichever platform lost the race that run. + template + void signal_maintenance(Apply&& apply) { + { + std::lock_guard lock(maintenance_mutex_); + apply(); + } + maintenance_ready_.notify_all(); + } // Runs the event through features/classifier/tracker and updates in-memory state. // Requires mutex_. Does NO storage I/O — returns what to persist (nullopt if nothing). std::optional compute_event(const CaptureEvent& event); @@ -519,6 +562,12 @@ class AppState { // the process has been up, so a system clock jump cannot make a prune overdue or // unreachable. Seeded at construction because Storage::open just pruned. std::atomic last_prune_steady_ms_{0}; + // Uptime at the last "capture backlog" log line, or nullopt if none has been written yet. + // Guarded by mutex_ (decided inside the drain phase). Deliberately not an int with a 0 + // sentinel: steady_ms() counts from an arbitrary epoch, so 0 is a value the clock can + // legitimately hold, and a sentinel inside the clock's own domain would defeat the + // throttle for as long as it sat there. + std::optional last_drain_backlog_log_ms_; // Use the shared_ptr atomic free functions instead of atomic: the Apple // libc++ shipped with the supported command-line tools does not provide the C++20 class // specialization, while atomic_load/store(shared_ptr*) are available cross-platform. diff --git a/src/capture/capture_thread.hpp b/src/capture/capture_thread.hpp index de7fb90a..fbf9ab67 100644 --- a/src/capture/capture_thread.hpp +++ b/src/capture/capture_thread.hpp @@ -42,6 +42,26 @@ class CaptureThread { return event; } + // Engine side: whether a further next_event() would return an event right now. + bool has_pending_events() const { return buffer_.has_pending(); } + + // Engine side: throw away everything queued, returning how many events went. + // + // The ring is single-producer/single-consumer, so this must not run while the engine is + // draining. AppState's callers satisfy that by holding the same state lock the drain holds + // -- the two pop sites are mutually exclusive, never concurrent. + // + // "Delete my activity" has to erase what was captured before it, including what is still + // in this queue -- otherwise the tick after the deletion files pre-deletion window titles + // into whatever session exists by then. Bounded by kCapacity rather than by "until empty" + // so a producer that keeps pushing cannot hold the caller here indefinitely; anything it + // pushes after the boundary is, correctly, post-deletion activity. + std::size_t discard_pending_events() { + std::size_t discarded = 0; + while (discarded < kCapacity && buffer_.pop()) ++discarded; + return discarded; + } + std::uint64_t events_dropped() const { return dropped_.load(std::memory_order_relaxed); } bool running() const { return running_.load(std::memory_order_relaxed); } bool failed() const { return failed_.load(std::memory_order_acquire); } diff --git a/src/capture/ring_buffer.hpp b/src/capture/ring_buffer.hpp index a3b9209f..9f498518 100644 --- a/src/capture/ring_buffer.hpp +++ b/src/capture/ring_buffer.hpp @@ -48,6 +48,15 @@ class RingBuffer { return value; } + // Consumer side. Non-destructive "is there anything to pop", so a bounded drain can tell + // "I stopped because my budget ran out" from "I stopped because I emptied the ring" + // without consuming the event that would answer it. Only meaningful on the consumer + // thread: the producer can make a false reading true a moment later, which is harmless + // here (the next tick sees it) and is why this is not used for correctness decisions. + bool has_pending() const { + return tail_.load(std::memory_order_relaxed) != head_.load(std::memory_order_acquire); + } + private: static constexpr std::size_t kMask = Capacity - 1; // Heap, not std::array: 65,536 CaptureEvents is ~6 MB, which silently lived in diff --git a/tests/app_state_test_access.hpp b/tests/app_state_test_access.hpp index f0060597..f2569116 100644 --- a/tests/app_state_test_access.hpp +++ b/tests/app_state_test_access.hpp @@ -61,7 +61,24 @@ struct AppStateTestAccess { // driven by the tick's idle edges, so testing it through `update_idle_for_test` alone // would exercise the detector and skip everything that acts on it. Points the same way // as 14.2. - static void engine_tick(AppState& state) { state.engine_tick(); } + // Returns what the engine loop reads: true if the drain stopped on a budget with events + // still queued, false if it emptied the ring. + static bool engine_tick(AppState& state) { return state.engine_tick(); } + + // Starts the capture producer WITHOUT the engine thread. `start_engine_for_test` starts + // both, which makes "how much does one tick drain" a race against a thread already + // draining; this lets a test fill the ring and then drive engine_tick() by hand. + static void start_capture_only(AppState& state, InputHook* hook) { + state.capture_.start(hook); + } + + static void stop_capture(AppState& state) noexcept { state.capture_.stop(); } + + // Whether capture still has events the engine has not taken. Lets a test assert that + // shutdown or a deletion left nothing behind, which is otherwise invisible. + static bool capture_has_pending(const AppState& state) { + return state.capture_.has_pending_events(); + } static bool maintenance_pending(const AppState& state) { return state.maintenance_pending_.load(std::memory_order_acquire); diff --git a/tests/test_app_state.cpp b/tests/test_app_state.cpp index 196d7adc..1dcaff0d 100644 --- a/tests/test_app_state.cpp +++ b/tests/test_app_state.cpp @@ -88,6 +88,89 @@ class HyperfocusHook final : public InputHook { std::atomic running_{true}; }; +// Fills the capture ring with `count` key events as fast as the producer can push, then +// parks. Used to put more work in the buffer than one tick is allowed to consume. +class BurstHook final : public InputHook { +public: + explicit BurstHook(std::size_t count) : count_(count) {} + + void run(InputCallback on_event, const std::atomic&) override { + for (std::size_t i = 0; i < count_; ++i) { + CaptureEvent event; + event.event_type = EventType::KeyPress; + // Spread across seconds: the classifier throttles to one prediction per second of + // event time, so identical timestamps would make every event after the first do + // almost no work -- the opposite of the load this hook exists to create. + event.timestamp_secs = 1.0 + static_cast(i) * 0.01; + event.app_name = "Cursor"; + event.window_title = "state.cpp - Snapback"; + on_event(std::move(event)); + } + emitted_.store(true, std::memory_order_release); + + while (running_.load(std::memory_order_relaxed)) { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + } + + void stop() noexcept override { running_.store(false, std::memory_order_relaxed); } + + bool emitted() const { return emitted_.load(std::memory_order_acquire); } + +private: + std::size_t count_; + std::atomic running_{true}; + std::atomic emitted_{false}; +}; + +// Pushes events as fast as it can for `duration`, the way a user holding down a key or +// dragging the mouse does. The ring fills and starts dropping, which is the point: the engine +// then never observes an empty buffer, which is the condition an unbounded drain never exits. +class FloodHook final : public InputHook { +public: + explicit FloodHook(std::chrono::milliseconds duration) : duration_(duration) {} + + void run(InputCallback on_event, const std::atomic& stop_requested) override { + const auto until = std::chrono::steady_clock::now() + duration_; + double ts = 1.0; + while (std::chrono::steady_clock::now() < until && + !stop_requested.load(std::memory_order_acquire) && + running_.load(std::memory_order_relaxed)) { + CaptureEvent event; + event.event_type = EventType::KeyPress; + event.timestamp_secs = ts; + ts += 0.01; + event.app_name = "Cursor"; + event.window_title = "state.cpp - Snapback"; + on_event(std::move(event)); + } + flooding_.store(false, std::memory_order_release); + + while (running_.load(std::memory_order_relaxed)) { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + } + + void stop() noexcept override { running_.store(false, std::memory_order_relaxed); } + + bool flooding() const { return flooding_.load(std::memory_order_acquire); } + +private: + std::chrono::milliseconds duration_; + std::atomic running_{true}; + std::atomic flooding_{true}; +}; + +// Runs `hook` as the capture producer with no engine thread, waits until it has finished +// pushing, and hands the state back so the test can drive engine_tick() itself. +void fill_capture_ring(AppState& state, BurstHook& hook) { + AppStateTestAccess::start_capture_only(state, &hook); + for (int attempt = 0; attempt < 5000 && !hook.emitted(); ++attempt) { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + REQUIRE(hook.emitted()); +} + class ReturningHook final : public InputHook { public: void run(InputCallback, const std::atomic&) override { @@ -2504,6 +2587,182 @@ TEST_CASE("AppState destruction stops a running engine") { CHECK(hook.stopped()); } +TEST_CASE("one tick drains a bounded slice of the ring and reports the backlog") { + // The drain used to be `while (capture_.next_event())`, which ends only when the consumer + // outruns the producer -- so a user who keeps typing keeps mutex_ held, and every command, + // every persistence flush, and every UI emission waits for them to stop. + auto state = make_state(); + BurstHook hook(kEngineDrainBudget + 512); + fill_capture_ring(*state, hook); + + // First tick stops on the budget with events still queued. + CHECK(AppStateTestAccess::engine_tick(*state)); + + // Nothing was dropped: the remainder is still in the ring and the following ticks take it. + int ticks = 1; // the truncated one above + for (bool backlog = true; backlog;) { + backlog = AppStateTestAccess::engine_tick(*state); + ++ticks; + REQUIRE(ticks < 100); // a tick that never clears the backlog is the bug, not a pass + } + CHECK(ticks >= 2); // the burst could not have been consumed by one tick + // The ring is empty now, so another tick has nothing to truncate on. + CHECK_FALSE(AppStateTestAccess::engine_tick(*state)); + + AppStateTestAccess::stop_capture(*state); +} + +TEST_CASE("a truncated drain still runs the rest of the tick") { + // The point of bounding the drain: phases 2 and 3 -- persistence and emission -- used to be + // unreachable while the ring kept refilling. One tick that stops on the budget must still + // publish what it computed rather than deferring it until input stops. + auto state = make_state(); + const auto session = state->start_session("bounded drain", FocusMode::Normal); + + std::vector seen; + state->set_emit_hook([&seen](const std::string& name, const std::string&, std::uint64_t) { + seen.push_back(name); + }); + + BurstHook hook(kEngineDrainBudget + 512); + fill_capture_ring(*state, hook); + + REQUIRE(AppStateTestAccess::engine_tick(*state)); // truncated, backlog remains + + // Phase 3 ran: the prediction this tick computed went out on the hook. + CHECK(std::find(seen.begin(), seen.end(), "prediction") != seen.end()); + // Phase 2 ran: its rows are already committed, not waiting for the ring to empty. + const auto persisted_after_first_tick = + AppStateTestAccess::storage(*state).recent_predictions(1000).size(); + CHECK(persisted_after_first_tick > 0); + CHECK(state->latest_prediction().has_value()); + + // And the deferred events are not lost -- draining the rest produces more predictions, + // which is what proves the first tick stopped early rather than consuming everything. + while (AppStateTestAccess::engine_tick(*state)) { + } + CHECK(AppStateTestAccess::storage(*state).recent_predictions(1000).size() > + persisted_after_first_tick); + + state->set_emit_hook(nullptr); + AppStateTestAccess::stop_capture(*state); + state->stop_session(session.session_id); +} + +TEST_CASE("shutdown finishes the backlog instead of discarding it") { + // The unbounded drain always left the ring empty, so a stop persisted everything the user + // had done. A bounded drain must not quietly lose the tail: the engine loop keeps ticking + // while a backlog remains, which terminates because stop_engine stops the producer first. + auto state = make_state(); + const auto session = state->start_session("drain on the way out", FocusMode::Normal); + + BurstHook hook(kEngineDrainBudget + 512); // ~25 s of event time, > one tick's budget + state->start_engine_for_test(&hook); + for (int attempt = 0; attempt < 5000 && !hook.emitted(); ++attempt) { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + REQUIRE(hook.emitted()); + + state->stop_engine(); + + CHECK_FALSE(AppStateTestAccess::capture_has_pending(*state)); + // One tick's worth of those events spans ~20 s of event time and the throttle allows one + // prediction per second, so a shutdown that stopped after a single slice would leave about + // 20. Everything drained is ~25. + const auto persisted = AppStateTestAccess::storage(*state).recent_predictions(1000).size(); + CHECK(persisted >= 24); + + state->stop_session(session.session_id); +} + +TEST_CASE("deleting activity also erases what capture had queued") { + // The activity epoch fences rows a tick was about to write; it does not reach into the + // capture queue. Events recorded before the user asked for deletion were still sitting + // there, and the next tick filed them -- into whatever session existed by then. + auto state = make_state(); + const auto before = state->start_session("before the delete", FocusMode::Normal); + BurstHook hook(kEngineDrainBudget + 512); + fill_capture_ring(*state, hook); + + state->delete_all_activity_data(); + CHECK_FALSE(AppStateTestAccess::capture_has_pending(*state)); + + // A session started right after the deletion must not inherit that pre-deletion activity. + const auto after = state->start_session("after the delete", FocusMode::Normal); + for (int tick = 0; tick < 5; ++tick) AppStateTestAccess::engine_tick(*state); + CHECK(AppStateTestAccess::storage(*state).recent_predictions(1000).empty()); + + AppStateTestAccess::stop_capture(*state); + state->stop_session(after.session_id); + (void)before; +} + +TEST_CASE("deleting one session leaves another session's queued events alone") { + // The mirror of the case above: the drop is scoped to the session being erased. Deleting + // some *other* session must not throw away input the user is producing right now. + auto state = make_state(); + const auto keep = state->start_session("still running", FocusMode::Normal); + const auto other = state->start_session("a second session replaces it", FocusMode::Normal); + state->stop_session(other.session_id); + const auto live = state->start_session("the live one", FocusMode::Normal); + + BurstHook hook(64); + fill_capture_ring(*state, hook); + REQUIRE(AppStateTestAccess::capture_has_pending(*state)); + + state->delete_session(other.session_id); + CHECK(AppStateTestAccess::capture_has_pending(*state)); // not this session's events + + AppStateTestAccess::engine_tick(*state); + CHECK_FALSE(AppStateTestAccess::storage(*state).recent_predictions(1000).empty()); + + AppStateTestAccess::stop_capture(*state); + state->stop_session(live.session_id); + (void)keep; +} + +TEST_CASE("a command is not starved while capture floods the ring") { + // The end-to-end version of the bound: real engine thread, real producer, and a caller + // asking the question a UI asks constantly. Before the drain was bounded this loop blocked + // for as long as the flood lasted, because the tick held mutex_ until the ring ran dry and + // a fast producer never let it. With the bound the worst wait is one bounded slice. + auto state = make_state(); + FloodHook hook(std::chrono::milliseconds(3000)); + state->start_engine_for_test(&hook); + + std::chrono::steady_clock::duration worst{}; + int samples = 0; + while (hook.flooding()) { + const auto began = std::chrono::steady_clock::now(); + (void)state->settings(); // pure in-memory read, but it needs mutex_ + worst = std::max(worst, std::chrono::steady_clock::now() - began); + ++samples; + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } + state->stop_engine(); + + const auto worst_ms = + std::chrono::duration_cast(worst).count(); + INFO("worst settings() wait during the flood: " << worst_ms << " ms"); + CHECK(samples > 20); // the probe really did run throughout the flood + // Generous on purpose: a bounded slice costs single-digit milliseconds, while the + // unbounded drain this replaces held the lock for the whole 3-second flood. + CHECK(worst_ms < 1000); +} + +TEST_CASE("an under-budget drain reports no backlog") { + // Guards the off-by-one: if `drained >= budget` were `>=` against the wrong counter every + // tick would claim a backlog and the engine loop would spin at 1 ms forever. + auto state = make_state(); + BurstHook hook(16); + fill_capture_ring(*state, hook); + + CHECK_FALSE(AppStateTestAccess::engine_tick(*state)); + CHECK(state->latest_prediction().has_value()); // the 16 events were processed, not skipped + + AppStateTestAccess::stop_capture(*state); +} + TEST_CASE("AppState confirms capture only after the backend delivers an event") { auto state = make_state(); CHECK_FALSE(state->health().permissions.capture_probe_confirmed);