teksilo-settings 0.9.1

Persistent reactive settings for Teksilo — key-value store, versioned setting files, MRU lists and window state.
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
// SPDX-License-Identifier: MPL-2.0
// SPDX-FileCopyrightText: 2026 FernTech

//! Live cross-process settings sync: a `notify`-based directory watcher
//! plus the registry that lets a changed path be dispatched to the
//! in-memory [`Reloadable`] handle that owns it.
//!
//! This is the read-side counterpart to the write-side cross-process
//! safety documented in `flush.rs` / `reload.rs`: every write in this
//! crate already merges safely against a peer's concurrent write, but a
//! process that never looks again will not notice a peer's write until
//! it happens to touch the same key itself. [`SettingsWatcher`] is what
//! makes it look again, automatically, the moment a peer's write lands
//! on disk.
//!
//! ## Shape, mirrored from `teksilo-i18n`'s `FtlFileWatcher`
//!
//! [`SettingsWatcher`] owns a `notify::RecommendedWatcher` background
//! thread and a type-erased sink `Arc<dyn Fn(PathBuf) + Send + Sync>`.
//! Exactly like `FtlFileWatcher`, it watches **directories**, not files:
//! atomic writers (this crate's own `write_atomic` included) write a
//! temp file and rename it over the target, which invalidates an
//! inode-level watch on the file itself. Unlike `FtlFileWatcher` — which
//! watches a fixed, already-existing set of `.ftl` files and derives
//! their parents — `SettingsWatcher` watches the settings *directories*
//! (`AppPaths::config_dir()` / `AppPaths::data_dir()`) directly, because
//! the set of settings files living there is open-ended and some of
//! them (e.g. `window_state.toml`) may not exist yet at watch-construction
//! time.
//!
//! The sink receives the changed path (not yet filtered against anything
//! this process cares about); [`SettingsRegistry::dispatch`] is what
//! decides whether the path names something registered and, if so,
//! calls its [`Reloadable::reload_from_disk`]. A path with no registered
//! owner (a `.lock` sidecar, a `.tmp` write-in-progress, an unrelated
//! file a peer dropped in the same directory) is a harmless no-op.
//!
//! ## The registry
//!
//! [`SettingsRegistry`] maps a canonical path to a `Weak<dyn Reloadable>`.
//! It never holds a strong reference itself: whoever opens a persisted
//! service (`SettingsBundle::open`, or application code opening its own
//! ad hoc `SettingsFile<T>` / `PersistedListModel<T>` / `MruList<T>`)
//! wraps it in an `Rc<dyn Reloadable>`, registers a weak clone via
//! [`SettingsRegistry::register`], and keeps the returned `Rc` alive for
//! as long as it wants peer writes to be picked up. When that `Rc` (and
//! every clone of it) is dropped, the registry's entry can no longer be
//! upgraded — [`SettingsRegistry::dispatch`] then quietly prunes it and
//! reports nothing happened. Nothing leaks and nothing is ever called on
//! a service that no longer exists.

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::rc::{Rc, Weak};
use std::sync::Arc;

use notify::{RecommendedWatcher, RecursiveMode, Watcher};

use crate::file::SettingsFileError;
use crate::reload::Reloadable;

/// Sink type invoked on the notify worker thread whenever a watched
/// settings directory reports a create/modify event. Implementations
/// must be thread-safe; `teksilo-app`'s implementation posts the path
/// through the winit `EventLoopProxy` as `AppEvent::SettingsReload`,
/// which hops back onto the UI thread where the (single-threaded,
/// `Rc`-based) [`SettingsRegistry`] actually lives.
pub type SettingsReloadSink = Arc<dyn Fn(PathBuf) + Send + Sync + 'static>;

/// Active directory watcher over one or more settings directories. One
/// per `TeksiloAppBuilder::run` invocation (when a settings bundle with
/// watching enabled is configured).
///
/// Owns the `notify::RecommendedWatcher` background thread for its whole
/// lifetime; dropping the `SettingsWatcher` stops the watcher and cleans
/// up. Kept alive by the caller for as long as live reload is wanted —
/// `teksilo-app` stores it on its window-loop handler, exactly like
/// `teksilo-i18n`'s `FtlFileWatcher`.
pub struct SettingsWatcher {
    _inner: RecommendedWatcher,
}

impl SettingsWatcher {
    /// Build a watcher over `dirs` (deduplicated by canonical path, so
    /// passing the same directory twice — e.g. `AppPaths::for_testing`,
    /// whose `config_dir()` and `data_dir()` are the same tempdir — never
    /// double-watches or double-fires) and a sink callback.
    ///
    /// A directory that does not exist (or can't be canonicalized for
    /// any other reason) is logged and skipped — not fatal — since a
    /// freshly-installed app may not have created its data directory yet
    /// when this is called. As long as at least the config directory
    /// exists (which `AppPaths` implies by the time `SettingsBundle` has
    /// successfully opened anything in it), watching still works for the
    /// files that matter.
    pub fn new(dirs: Vec<PathBuf>, sink: SettingsReloadSink) -> Result<Self, notify::Error> {
        let mut seen: std::collections::HashSet<PathBuf> = std::collections::HashSet::new();
        let mut targets: Vec<PathBuf> = Vec::new();
        for dir in dirs {
            let canonical = match dir.canonicalize() {
                Ok(p) => p,
                Err(e) => {
                    eprintln!(
                        "teksilo-settings: cannot watch `{}` ({e}); live settings reload \
                         disabled for that directory",
                        dir.display()
                    );
                    continue;
                }
            };
            if seen.insert(canonical.clone()) {
                targets.push(canonical);
            }
        }

        let sink_handle = sink.clone();
        let mut watcher = notify::recommended_watcher(
            move |res: Result<notify::Event, notify::Error>| match res {
                Ok(event) if should_reload(&event.kind) => {
                    for path in &event.paths {
                        (sink_handle)(path.clone());
                    }
                }
                Ok(_) => {}
                Err(e) => {
                    eprintln!("teksilo-settings: watcher error: {e}");
                }
            },
        )?;

        for target in &targets {
            watcher.watch(target, RecursiveMode::NonRecursive)?;
        }

        Ok(Self { _inner: watcher })
    }
}

/// Return `true` for event kinds that mean a file's content may have
/// changed. `notify` fires many events for other operations (access,
/// metadata-only, permissions) that never need a reload. Mirrors
/// `teksilo-i18n::file_watcher::should_reload` exactly.
fn should_reload(kind: &notify::EventKind) -> bool {
    use notify::EventKind::*;
    matches!(kind, Modify(_) | Create(_))
}

/// Best-effort canonical form of `path`, robust to the file itself not
/// existing yet on disk (unlike [`Path::canonicalize`], which requires
/// every component — including the leaf — to exist).
///
/// Canonicalizes the *parent* directory (which every registered settings
/// path has, and which is guaranteed to exist before anything is ever
/// written into it) and rejoins the file name. This is the same key a
/// [`SettingsWatcher`] event path resolves to, since the watcher is
/// always constructed over the canonicalized parent directory too — so
/// a registration made before a file's first write and an event fired
/// after it exists still land on the same map key.
fn canonical_settings_path(path: &Path) -> PathBuf {
    match (path.parent(), path.file_name()) {
        (Some(parent), Some(name)) => match parent.canonicalize() {
            Ok(canonical_parent) => canonical_parent.join(name),
            Err(_) => path.to_path_buf(),
        },
        _ => path.to_path_buf(),
    }
}

/// Registry mapping a canonical settings path to the live [`Reloadable`]
/// handle that owns it, so a file-watcher event naming that path can be
/// dispatched to the right in-memory state.
///
/// `Clone` is cheap (an `Rc` bump) — every clone shares the same
/// underlying map, matching the rest of this crate's handle types.
/// Holds only [`Weak`] references: see the module docs' "the registry"
/// section for the full ownership contract.
#[derive(Clone, Default)]
pub struct SettingsRegistry {
    entries: Rc<std::cell::RefCell<HashMap<PathBuf, Weak<dyn Reloadable>>>>,
}

impl SettingsRegistry {
    /// A fresh, empty registry.
    pub fn new() -> Self {
        Self::default()
    }

    /// Register `reloadable` under its canonical path and return it back
    /// unchanged, so a caller can register and retain in one expression:
    ///
    /// ```
    /// use teksilo_settings::{SettingsRegistry, SettingsFile, Migrator, Versioned};
    /// use serde::{Serialize, Deserialize};
    /// use std::rc::Rc;
    ///
    /// #[derive(Serialize, Deserialize, Default, Clone, PartialEq)]
    /// struct Prefs { version: u32 }
    /// impl Versioned for Prefs {
    ///     const CURRENT_VERSION: u32 = 1;
    ///     fn version(&self) -> u32 { self.version }
    ///     fn set_version(&mut self, v: u32) { self.version = v; }
    /// }
    ///
    /// let dir = tempfile::tempdir().unwrap();
    /// let file: SettingsFile<Prefs> =
    ///     SettingsFile::load(dir.path().join("prefs.toml"), Migrator::new()).unwrap();
    ///
    /// let registry = SettingsRegistry::new();
    /// // Keep `handle` alive for as long as reload should keep working.
    /// let handle = registry.register(Rc::new(file.clone()));
    /// drop(handle); // dropping it deregisters: no leak, no dangling call.
    /// ```
    ///
    /// The caller is responsible for keeping the returned `Rc` alive —
    /// only a `Weak` is retained internally, by design (see the module
    /// docs). Registering a second `Reloadable` under the same canonical
    /// path replaces the first entry.
    pub fn register(&self, reloadable: Rc<dyn Reloadable>) -> Rc<dyn Reloadable> {
        let key = canonical_settings_path(reloadable.path());
        self.entries
            .borrow_mut()
            .insert(key, Rc::downgrade(&reloadable));
        reloadable
    }

    /// Look up `changed_path`'s registered owner and call
    /// [`Reloadable::reload_from_disk`] on it.
    ///
    /// Returns `Ok(true)` if the owner's in-memory state actually
    /// changed, `Ok(false)` if nothing needed to change (including: the
    /// path names nothing registered, or its owner has been dropped —
    /// in the latter case the dead entry is pruned from the map so it
    /// doesn't accumulate forever).
    pub fn dispatch(&self, changed_path: &Path) -> Result<bool, SettingsFileError> {
        let key = canonical_settings_path(changed_path);
        let weak = { self.entries.borrow().get(&key).cloned() };
        let Some(weak) = weak else {
            return Ok(false);
        };
        match weak.upgrade() {
            Some(reloadable) => reloadable.reload_from_disk(),
            None => {
                self.entries.borrow_mut().remove(&key);
                Ok(false)
            }
        }
    }

    /// The canonical paths currently registered (including entries whose
    /// owner has since been dropped but not yet pruned by a `dispatch`
    /// call). Exposed for tests and diagnostics.
    pub fn registered_paths(&self) -> Vec<PathBuf> {
        self.entries.borrow().keys().cloned().collect()
    }

    /// Number of live (upgradeable) entries. Exposed for tests.
    pub fn live_count(&self) -> usize {
        self.entries
            .borrow()
            .values()
            .filter(|w| w.upgrade().is_some())
            .count()
    }
}

impl std::fmt::Debug for SettingsRegistry {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let entries = self.entries.borrow();
        f.debug_struct("SettingsRegistry")
            .field("registered_paths", &entries.len())
            .field(
                "live",
                &entries.values().filter(|w| w.upgrade().is_some()).count(),
            )
            .finish()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::file::SettingsFile;
    use crate::migration::{Migrator, Versioned};
    use serde::{Deserialize, Serialize};
    use std::time::{Duration, Instant};
    use tempfile::tempdir;

    #[derive(Serialize, Deserialize, Debug, PartialEq, Clone, Default)]
    struct Prefs {
        version: u32,
        value: String,
    }
    impl Versioned for Prefs {
        const CURRENT_VERSION: u32 = 1;
        fn version(&self) -> u32 {
            self.version
        }
        fn set_version(&mut self, v: u32) {
            self.version = v;
        }
    }

    /// Poll `condition` for up to `timeout`, sleeping briefly between
    /// checks. Never a blind fixed sleep: returns as soon as the
    /// condition is true, and fails loudly (via the caller's assertion)
    /// if it never becomes true within the (generous) budget.
    fn poll_until(timeout: Duration, mut condition: impl FnMut() -> bool) -> bool {
        let deadline = Instant::now() + timeout;
        loop {
            if condition() {
                return true;
            }
            if Instant::now() >= deadline {
                return false;
            }
            std::thread::sleep(Duration::from_millis(20));
        }
    }

    const GENEROUS_TIMEOUT: Duration = Duration::from_secs(5);

    // -----------------------------------------------------------------
    // canonical_settings_path
    // -----------------------------------------------------------------

    #[test]
    fn canonical_settings_path_resolves_even_when_file_is_missing() {
        let dir = tempdir().unwrap();
        let missing = dir.path().join("not-yet-written.toml");
        let resolved = canonical_settings_path(&missing);
        // The parent must have been canonicalized even though the leaf
        // file does not exist.
        assert_eq!(
            resolved.parent().unwrap(),
            dir.path().canonicalize().unwrap()
        );
        assert_eq!(resolved.file_name().unwrap(), "not-yet-written.toml");
    }

    #[test]
    fn canonical_settings_path_is_stable_across_existence() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("general.toml");

        let before = canonical_settings_path(&path);
        std::fs::write(&path, "version = 1\n").unwrap();
        let after = canonical_settings_path(&path);

        assert_eq!(
            before, after,
            "registering before vs. after creation must key identically"
        );
    }

    // -----------------------------------------------------------------
    // SettingsRegistry
    // -----------------------------------------------------------------

    #[test]
    fn dispatch_on_unregistered_path_is_a_harmless_no_op() {
        let registry = SettingsRegistry::new();
        let dir = tempdir().unwrap();
        assert!(!registry.dispatch(&dir.path().join("unknown.toml")).unwrap());
    }

    #[test]
    fn register_then_dispatch_reloads_a_peers_write() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("prefs.toml");

        let a: SettingsFile<Prefs> = SettingsFile::load(path.clone(), Migrator::new()).unwrap();
        let b: SettingsFile<Prefs> = SettingsFile::load(path.clone(), Migrator::new()).unwrap();

        let registry = SettingsRegistry::new();
        let _handle = registry.register(Rc::new(b.clone()) as Rc<dyn Reloadable>);

        a.mutate(|p| p.value = "peer-write".into()).unwrap();

        assert!(registry.dispatch(&path).unwrap());
        assert_eq!(b.snapshot().value, "peer-write");
    }

    #[test]
    fn dropped_service_is_pruned_and_never_dispatched_to() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("prefs.toml");

        let a: SettingsFile<Prefs> = SettingsFile::load(path.clone(), Migrator::new()).unwrap();
        let b: SettingsFile<Prefs> = SettingsFile::load(path.clone(), Migrator::new()).unwrap();

        let registry = SettingsRegistry::new();
        let handle = registry.register(Rc::new(b.clone()) as Rc<dyn Reloadable>);
        assert_eq!(registry.live_count(), 1);

        drop(handle);
        // `b` itself is still alive (it's a separate `SettingsFile` clone
        // handle sharing the same on-disk file), but the *registration*
        // Rc was the only strong reference the registry's Weak could
        // upgrade through — dropping it must deregister.
        assert_eq!(registry.live_count(), 0);

        a.mutate(|p| p.value = "peer-write-after-drop".into())
            .unwrap();

        // No owner to dispatch to any more: reports nothing happened,
        // and prunes the dead entry.
        assert!(!registry.dispatch(&path).unwrap());
        assert!(registry.registered_paths().is_empty());
        // `b`'s own in-memory value is untouched — nothing was called on it.
        assert_eq!(b.snapshot().value, "");
    }

    #[test]
    fn registering_under_the_same_path_replaces_the_previous_owner() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("prefs.toml");

        let first: SettingsFile<Prefs> = SettingsFile::load(path.clone(), Migrator::new()).unwrap();
        let second: SettingsFile<Prefs> =
            SettingsFile::load(path.clone(), Migrator::new()).unwrap();

        let registry = SettingsRegistry::new();
        let _first_handle = registry.register(Rc::new(first.clone()) as Rc<dyn Reloadable>);
        let _second_handle = registry.register(Rc::new(second.clone()) as Rc<dyn Reloadable>);

        assert_eq!(registry.registered_paths().len(), 1);

        let writer: SettingsFile<Prefs> =
            SettingsFile::load(path.clone(), Migrator::new()).unwrap();
        writer.mutate(|p| p.value = "via-second".into()).unwrap();

        assert!(registry.dispatch(&path).unwrap());
        assert_eq!(second.snapshot().value, "via-second");
        // `first` was replaced in the registry and never touched.
        assert_eq!(first.snapshot().value, "");
    }

    // -----------------------------------------------------------------
    // SettingsWatcher — end-to-end, real notify + real filesystem.
    // -----------------------------------------------------------------

    #[test]
    fn construction_over_a_missing_directory_does_not_error() {
        let sink: SettingsReloadSink = Arc::new(|_path| {});
        let watcher = SettingsWatcher::new(
            vec![PathBuf::from("/definitely/does/not/exist/anywhere")],
            sink,
        );
        assert!(watcher.is_ok());
    }

    #[test]
    fn construction_dedupes_identical_directories() {
        // `AppPaths::for_testing` routes config_dir and data_dir to the
        // same tempdir; watching it twice must not error (and must not
        // register the OS-level watch twice).
        let dir = tempdir().unwrap();
        let sink: SettingsReloadSink = Arc::new(|_path| {});
        let watcher = SettingsWatcher::new(
            vec![dir.path().to_path_buf(), dir.path().to_path_buf()],
            sink,
        );
        assert!(watcher.is_ok());
    }

    /// `SettingsRegistry` is `Rc`-based (single-threaded) by design — it
    /// is only ever touched from the UI thread in production, reached
    /// via an `AppEvent` posted through the winit proxy (see
    /// `teksilo-app`'s wiring). A `SettingsWatcher`'s sink, in contrast,
    /// runs on the notify background thread and must be `Send + Sync`.
    /// These tests reproduce that exact split: the sink only pushes the
    /// changed path onto a thread-safe queue; the *test* thread (which
    /// is also where every `SettingsFile` / `SettingsRegistry` handle
    /// below was constructed) drains it and calls `dispatch` itself —
    /// exactly mirroring the real cross-thread handoff.
    type PathQueue = Arc<std::sync::Mutex<std::collections::VecDeque<PathBuf>>>;

    fn queueing_sink(queue: PathQueue) -> SettingsReloadSink {
        Arc::new(move |path| {
            queue.lock().unwrap().push_back(path);
        })
    }

    /// Drain every path currently queued and dispatch each through
    /// `registry` (on the calling thread), returning how many produced
    /// an *effective* reload (`Ok(true)`).
    fn drain_and_dispatch(registry: &SettingsRegistry, queue: &PathQueue) -> usize {
        let paths: Vec<PathBuf> = queue.lock().unwrap().drain(..).collect();
        paths
            .into_iter()
            .filter(|p| matches!(registry.dispatch(p), Ok(true)))
            .count()
    }

    /// An external write (a second handle, standing in for a peer
    /// process) to a watched, registered file drives exactly one
    /// *effective* reload — i.e. the registered `Reloadable`'s
    /// in-memory state flips exactly once, no matter how many raw
    /// filesystem events the OS coalesces the write into. This is the
    /// self-write-suppression / content-backstop contract in
    /// `reload_from_disk` doing its job on top of a possibly-noisy
    /// stream of notify events — exactly the property a watcher must
    /// preserve.
    #[test]
    fn external_write_triggers_exactly_one_effective_reload() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("prefs.toml");

        // Create the file up front so `dir` already contains something
        // watchable and the peer's first write is a real Modify.
        let peer: SettingsFile<Prefs> = SettingsFile::load(path.clone(), Migrator::new()).unwrap();
        let mine: SettingsFile<Prefs> = SettingsFile::load(path.clone(), Migrator::new()).unwrap();

        let registry = SettingsRegistry::new();
        let _handle = registry.register(Rc::new(mine.clone()) as Rc<dyn Reloadable>);

        let queue: PathQueue = Arc::new(std::sync::Mutex::new(std::collections::VecDeque::new()));
        let _watcher =
            SettingsWatcher::new(vec![dir.path().to_path_buf()], queueing_sink(queue.clone()))
                .unwrap();

        peer.mutate(|p| p.value = "external".into()).unwrap();

        let mut effective_reloads = 0usize;
        assert!(
            poll_until(GENEROUS_TIMEOUT, || {
                effective_reloads += drain_and_dispatch(&registry, &queue);
                effective_reloads >= 1
            }),
            "expected the external write to be picked up within the timeout"
        );

        // Give any duplicate/coalesced OS events a further bounded window
        // to arrive and prove they don't cause a second *effective*
        // reload (the stamp/content backstop must absorb them).
        let deadline = Instant::now() + Duration::from_millis(300);
        while Instant::now() < deadline {
            effective_reloads += drain_and_dispatch(&registry, &queue);
            std::thread::sleep(Duration::from_millis(20));
        }
        assert_eq!(
            effective_reloads, 1,
            "exactly one effective reload, however many raw fs events fired"
        );
        assert_eq!(mine.snapshot().value, "external");
    }

    /// Our own write must never be treated as a peer's: dispatching the
    /// path right after our own `mutate` must report `Ok(false)` and
    /// touch nothing, because `reload_from_disk`'s stamp check
    /// recognizes it as already-current.
    #[test]
    fn our_own_write_triggers_no_effective_reload() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("prefs.toml");

        let mine: SettingsFile<Prefs> = SettingsFile::load(path.clone(), Migrator::new()).unwrap();
        let registry = SettingsRegistry::new();
        let _handle = registry.register(Rc::new(mine.clone()) as Rc<dyn Reloadable>);

        mine.mutate(|p| p.value = "mine".into()).unwrap();

        // Simulate the watcher noticing our own write (it would: the OS
        // can't tell who wrote it) and dispatching it straight through.
        assert!(!registry.dispatch(&path).unwrap());
        assert_eq!(
            mine.snapshot().value,
            "mine",
            "our own value must be untouched"
        );
    }

    /// A dropped service's registration must never fire, even though its
    /// underlying file keeps being written to by a peer. Proven via a
    /// live sentinel registered on a *different* file in the same
    /// watched directory: once the sentinel's write is observed, the
    /// dropped service's counter is asserted to still be zero — bounding
    /// the wait deterministically instead of a blind sleep.
    #[test]
    fn dropped_service_is_never_called_by_a_live_watcher() {
        let dir = tempdir().unwrap();
        let dropped_path = dir.path().join("dropped.toml");
        let sentinel_path = dir.path().join("sentinel.toml");

        let dropped_peer: SettingsFile<Prefs> =
            SettingsFile::load(dropped_path.clone(), Migrator::new()).unwrap();
        let dropped_mine: SettingsFile<Prefs> =
            SettingsFile::load(dropped_path.clone(), Migrator::new()).unwrap();
        let sentinel_peer: SettingsFile<Prefs> =
            SettingsFile::load(sentinel_path.clone(), Migrator::new()).unwrap();
        let sentinel_mine: SettingsFile<Prefs> =
            SettingsFile::load(sentinel_path.clone(), Migrator::new()).unwrap();

        let registry = SettingsRegistry::new();
        let dropped_handle = registry.register(Rc::new(dropped_mine.clone()) as Rc<dyn Reloadable>);
        let _sentinel_handle =
            registry.register(Rc::new(sentinel_mine.clone()) as Rc<dyn Reloadable>);

        // Drop the "dropped" service's registration handle before any
        // write happens.
        drop(dropped_handle);
        assert_eq!(registry.live_count(), 1);

        let queue: PathQueue = Arc::new(std::sync::Mutex::new(std::collections::VecDeque::new()));
        let _watcher =
            SettingsWatcher::new(vec![dir.path().to_path_buf()], queueing_sink(queue.clone()))
                .unwrap();

        dropped_peer
            .mutate(|p| p.value = "should-never-land".into())
            .unwrap();
        sentinel_peer
            .mutate(|p| p.value = "sentinel-fired".into())
            .unwrap();

        let dropped_key = canonical_settings_path(&dropped_path);
        let sentinel_key = canonical_settings_path(&sentinel_path);
        let mut dropped_reloads = 0usize;
        let mut sentinel_reloads = 0usize;

        // Manual poll loop (not `poll_until`, which only tracks a single
        // `bool` condition) so both counters can be accumulated on every
        // iteration without a nested-closure double-borrow.
        let deadline = Instant::now() + GENEROUS_TIMEOUT;
        loop {
            let paths: Vec<PathBuf> = queue.lock().unwrap().drain(..).collect();
            for p in paths {
                let key = canonical_settings_path(&p);
                if let Ok(true) = registry.dispatch(&p) {
                    if key == dropped_key {
                        dropped_reloads += 1;
                    } else if key == sentinel_key {
                        sentinel_reloads += 1;
                    }
                }
            }
            if sentinel_reloads >= 1 || Instant::now() >= deadline {
                break;
            }
            std::thread::sleep(Duration::from_millis(20));
        }

        assert!(
            sentinel_reloads >= 1,
            "sentinel write should have been observed within the timeout"
        );
        assert_eq!(
            dropped_reloads, 0,
            "a dropped service's registration must never be dispatched to"
        );
        assert_eq!(
            dropped_mine.snapshot().value,
            "",
            "the dropped service's in-memory value must be untouched"
        );
    }

    /// The rename dance: deleting the underlying file and recreating it
    /// must not break the watch (which targets the parent directory, not
    /// the file's inode) — the recreation is picked up like any other
    /// write.
    #[test]
    fn survives_file_deletion_and_recreation() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("prefs.toml");

        // File must exist up front for the initial `load` to establish a
        // stamp baseline consistent with the rest of the suite.
        std::fs::write(&path, "version = 1\nvalue = \"\"\n").unwrap();
        let mine: SettingsFile<Prefs> = SettingsFile::load(path.clone(), Migrator::new()).unwrap();

        let registry = SettingsRegistry::new();
        let _handle = registry.register(Rc::new(mine.clone()) as Rc<dyn Reloadable>);

        let queue: PathQueue = Arc::new(std::sync::Mutex::new(std::collections::VecDeque::new()));
        let _watcher =
            SettingsWatcher::new(vec![dir.path().to_path_buf()], queueing_sink(queue.clone()))
                .unwrap();

        // Delete the file entirely (no peer handle involved — a plain
        // `remove_file`, closer to what an external "reset settings"
        // tool would do), then recreate it via a fresh locked write,
        // exactly like `write_atomic`'s temp-file-then-rename.
        std::fs::remove_file(&path).unwrap();

        let recreated: SettingsFile<Prefs> =
            SettingsFile::load(path.clone(), Migrator::new()).unwrap();
        recreated
            .mutate(|p| p.value = "recreated-after-delete".into())
            .unwrap();

        let mut effective_reloads = 0usize;
        assert!(
            poll_until(GENEROUS_TIMEOUT, || {
                effective_reloads += drain_and_dispatch(&registry, &queue);
                effective_reloads >= 1
            }),
            "the recreated file's write should still be observed after the watched \
             file was deleted and recreated"
        );
        assert_eq!(mine.snapshot().value, "recreated-after-delete");
    }
}