authorclover caruso <git@paperclover.net> 2026-10-10 20:48:39-07:00
committerclover caruso <git@paperclover.net> 2026-10-10 22:56:06-07:00
log5f50b6a9b5be1735c89716498f92a9c9b0134010
treed05a589680b3285790e7bb8dd35b65ff83def558
parent3f1c99c5f250c54828fc1ae415f4e8bd27db8a97
signature Signed by SSH key SHA256:52mNGHRsVFBDED9IAX5pe+LRWUefqTbxEReunq21QvU (~clover)

fix: report live sync status for held sections

Derive held section status from weak references to its worker, replica, and observer after releasing the background watch lock. Share status derivation with Section, counting queued edits without scanning recovery assets. Cover offline queue publication, live remote error recovery, and cache release while the background watch retains its pre-hold status. Fixes #117 Assisted-by: gpt-6.1-sol

3 files changed, 193 insertions(+), 28 deletions(-)

crates/notebook/src/background.rs+162-12
......@@ -6,7 +6,7 @@ use crate::fs;
66use crate::{
77 EditStatus, Error, Remote, Replica, Result,
88 discover::{Entry, Listed},
9 session::{Section, SyncStatus, reached},
9 session::{Observed, Section, SyncStatus, reached, sync_status},
1010 worker::Signal,
1111};
1212use onestore::Stamp;
......@@ -112,8 +112,14 @@ struct Watch {
112112 image: Option<Vec<u8>>,
113113 /// When the section is next checked.
114114 due: Instant,
115 /// The worker of the session that holds the section.
116 held: Option<Weak<Signal>>,
115 held: Option<Held>,
116}
117
118#[derive(Clone)]
119struct Held {
120 worker: Weak<Signal>,
121 replica: Weak<Replica>,
122 observed: Weak<Mutex<Observed>>,
117123}
118124
119125impl Background {
......@@ -429,7 +435,11 @@ impl Background {
429435 let reported = watched.reported;
430436 if let Some(watch) = watched.sections.iter_mut().find(|watch| watch.path == path) {
431437 worker.watched.store(reported, Ordering::Release);
432 watch.held = Some(Arc::downgrade(&worker));
438 watch.held = Some(Held {
439 worker: Arc::downgrade(&worker),
440 replica: Arc::downgrade(replica),
441 observed: Arc::downgrade(&section.observed),
442 });
433443 }
434444 }
435445 let (shared, path) = (Arc::downgrade(&self.0), path.to_owned());
......@@ -449,10 +459,9 @@ impl Background {
449459 self.0.touched(paths);
450460 }
451461
452 /// Each watched section's status as its last check left it, in watch order. A section a
453 /// session holds keeps the status it had before.
462 /// Each watched section's status, in watch order; held sections report their live state.
454463 pub fn status(&self) -> Vec<(String, SyncStatus)> {
455 self.0.watched.lock().map_or_else(
464 let sections: Vec<_> = self.0.watched.lock().map_or_else(
456465 |_| Vec::new(),
457466 |watched| {
458467 watched
......@@ -470,11 +479,34 @@ impl Background {
470479 .map(|error| io::Error::new(error.kind(), error.to_string())),
471480 queued: status.queued,
472481 },
482 watch.held.clone(),
473483 )
474484 })
475485 .collect()
476486 },
477 )
487 );
488 sections
489 .into_iter()
490 .map(|(path, fallback, held)| {
491 let live = held.and_then(|held| {
492 let worker = held.worker.upgrade()?;
493 let replica = held.replica.upgrade()?;
494 let observed = held.observed.upgrade()?;
495 let synced =
496 Some(worker.synced.load(Ordering::Acquire)).filter(|time| *time != 0);
497 Some(sync_status(&replica, synced, &observed))
498 });
499 let status = match live {
500 Some(Ok(status)) => status,
501 Some(Err(error)) => SyncStatus {
502 error: Some(reached(&error)),
503 ..fallback
504 },
505 None => fallback,
506 };
507 (path, status)
508 })
509 .collect()
478510 }
479511
480512 /// Sections whose file changed since the last call, by catalog path.
......@@ -678,7 +710,7 @@ impl Watched {
678710 for signal in self
679711 .sections
680712 .iter()
681 .filter_map(|watch| watch.held.as_ref()?.upgrade())
713 .filter_map(|watch| watch.held.as_ref()?.worker.upgrade())
682714 {
683715 signal.watched.store(reported, Ordering::Release);
684716 signal.wake();
......@@ -807,7 +839,7 @@ impl Watched {
807839 if !bound {
808840 return Next::Connect(self.connection);
809841 }
810 if let Some(worker) = watch.held.as_ref().and_then(Weak::upgrade) {
842 if let Some(worker) = watch.held.as_ref().and_then(|held| held.worker.upgrade()) {
811843 worker.wake();
812844 watch.due = now + interval;
813845 continue;
......@@ -1091,7 +1123,11 @@ mod tests {
10911123 let mut watched = Watched::default();
10921124 watched.watch(sections(1, true), now);
10931125 let (held, woken) = Signal::new();
1094 watched.sections[0].held = Some(Arc::downgrade(&held));
1126 watched.sections[0].held = Some(Held {
1127 worker: Arc::downgrade(&held),
1128 replica: Weak::new(),
1129 observed: Weak::new(),
1130 });
10951131 assert!(matches!(watched.next(now, false), Next::Connect(0)));
10961132 watched.watching(true);
10971133 assert!(held.watched.load(Ordering::Acquire));
......@@ -1430,6 +1466,116 @@ mod tests {
14301466 drop(rewatch);
14311467 }
14321468
1469 #[test]
1470 fn held_status_tracks_its_worker_without_retaining_the_replica() {
1471 use crate::session::SyncState;
1472 use onestore::{
1473 PageCreation,
1474 op::{Edit, Op, SectionOp},
1475 };
1476
1477 let directory = tempfile::tempdir().unwrap();
1478 let file = directory.path().join("Held.one");
1479 let source = onestore::create_section("Held.one", "", "Author").unwrap();
1480 fs::write(&file, &source).unwrap();
1481 let (changed, changes) = mpsc::channel();
1482 let section = Section::open(&file, directory.path().join("cache"), move || {
1483 let _ = changed.send(());
1484 })
1485 .unwrap();
1486 changes.recv_timeout(Duration::from_secs(3)).unwrap();
1487 section.set_offline(true);
1488
1489 let (stamps, _) = mpsc::channel();
1490 let files = Files {
1491 images: Arc::default(),
1492 stamps,
1493 };
1494 files.write("Held.one", source);
1495 let remote = files.clone();
1496 let background = Background::start(
1497 false,
1498 move |_| {
1499 let (files, listed) = (remote.clone(), remote.clone());
1500 let bind = move |path: &str| File(files.clone(), path.to_owned());
1501 let list = move |folder: String| std::future::ready(Ok(listed.list(&folder)));
1502 Ok(((bind, list), true))
1503 },
1504 || {},
1505 )
1506 .unwrap();
1507 background.set_offline(true);
1508 background.watch(vec![files.known("Held.one")]);
1509 background.hold("Held.one", &section);
1510 section
1511 .apply(
1512 "Author",
1513 Edit {
1514 at: crate::now(),
1515 ops: vec![Op::Section(SectionOp::Create(
1516 PageCreation::new(None, Some("Queued"), "Author").unwrap(),
1517 ))],
1518 },
1519 )
1520 .unwrap();
1521 let pending = section.pending().unwrap();
1522 assert_eq!(pending.len(), 1);
1523 let status = background.status().remove(0).1;
1524 assert_eq!(status.queued, 1);
1525 assert!(status.synced.is_some());
1526 assert_eq!(status.state(), SyncState::Syncing);
1527 {
1528 let watched = background.0.watched.lock().unwrap();
1529 assert!(watched.sections[0].status.synced.is_none());
1530 assert_eq!(watched.sections[0].status.queued, 0);
1531 }
1532
1533 let wait = |expected| {
1534 let deadline = Instant::now() + Duration::from_secs(3);
1535 loop {
1536 let status = background.status().remove(0).1;
1537 if status.state() == expected {
1538 return status;
1539 }
1540 changes
1541 .recv_timeout(deadline.saturating_duration_since(Instant::now()))
1542 .unwrap();
1543 }
1544 };
1545 section.set_offline(false);
1546 assert_eq!(wait(SyncState::UpToDate).queued, 0);
1547 assert!(matches!(
1548 section.status(pending[0].id).unwrap(),
1549 Some(EditStatus::Published { .. })
1550 ));
1551
1552 let published = fs::read_file(&file).unwrap();
1553 fs::remove_file(&file).unwrap();
1554 section.wake();
1555 assert_eq!(
1556 wait(SyncState::NotConnected).error.unwrap().kind(),
1557 io::ErrorKind::NotFound
1558 );
1559 fs::write(&file, published).unwrap();
1560 section.wake();
1561 assert!(wait(SyncState::UpToDate).error.is_none());
1562 assert!(
1563 background.0.watched.lock().unwrap().sections[0]
1564 .status
1565 .synced
1566 .is_none()
1567 );
1568
1569 let replica = Arc::downgrade(section.replica());
1570 section.close().unwrap();
1571 assert!(replica.upgrade().is_none());
1572 assert!(
1573 background.0.watched.lock().unwrap().sections[0]
1574 .held
1575 .is_none()
1576 );
1577 }
1578
14331579 #[test]
14341580 fn a_held_section_is_left_to_its_worker_which_the_watch_wakes() {
14351581 let (sender, stamps) = mpsc::channel();
......@@ -1462,7 +1608,11 @@ mod tests {
14621608 // As `hold` leaves it to a session's worker.
14631609 let (signal, woken) = Signal::new();
14641610 if let Ok(mut watched) = background.0.watched.lock() {
1465 watched.sections[0].held = Some(Arc::downgrade(&signal));
1611 watched.sections[0].held = Some(Held {
1612 worker: Arc::downgrade(&signal),
1613 replica: Weak::new(),
1614 observed: Weak::new(),
1615 });
14661616 signal.watched.store(watched.reported, Ordering::Release);
14671617 }
14681618 assert!(
crates/notebook/src/session.rs+30-15
......@@ -2084,7 +2084,30 @@ impl SyncStatus {
20842084}
20852085
20862086/// The last attempt's outcome; `None` before the first.
2087type Observed = Option<Option<io::Error>>;
2087pub(crate) type Observed = Option<Option<io::Error>>;
2088
2089pub(crate) fn sync_status(
2090 replica: &Replica,
2091 synced: Option<u64>,
2092 observed: &Mutex<Observed>,
2093) -> Result<SyncStatus> {
2094 let queued: u64 = crate::lock(&replica.section.connection)?.query_row(
2095 "SELECT count(*) FROM edits",
2096 [],
2097 |row| row.get(0),
2098 )?;
2099 let observed = observed
2100 .lock()
2101 .map_err(|_| io::Error::other("Synchronization observer panicked"))?;
2102 Ok(SyncStatus {
2103 synced,
2104 error: observed
2105 .as_ref()
2106 .and_then(Option::as_ref)
2107 .map(|error| io::Error::new(error.kind(), error.to_string())),
2108 queued,
2109 })
2110}
20882111
20892112/// A section file with its replica and background publication. `Send + Sync`: edits
20902113/// arrive from any thread without waiting, and `notify` wakes the host when events wait.
......@@ -2095,7 +2118,7 @@ pub struct Section {
20952118 events: Mutex<Receiver<Event>>,
20962119 sender: Sender<Event>,
20972120 notify: Notify,
2098 observed: Arc<Mutex<Observed>>,
2121 pub(crate) observed: Arc<Mutex<Observed>>,
20992122}
21002123
21012124impl Section {
......@@ -2554,19 +2577,11 @@ impl Section {
25542577 /// When the section file was last reached, why it could not be since, and what waits
25552578 /// for it. `notify` runs when it is first reached and when an error comes or goes.
25562579 pub fn sync_status(&self) -> Result<SyncStatus> {
2557 let queued = self.replica.recovery_summary()?.queued_edits;
2558 let observed = self
2559 .observed
2560 .lock()
2561 .map_err(|_| io::Error::other("Synchronization observer panicked"))?;
2562 Ok(SyncStatus {
2563 synced: self.worker.as_ref().and_then(SyncWorker::synced),
2564 error: observed
2565 .as_ref()
2566 .and_then(Option::as_ref)
2567 .map(|error| io::Error::new(error.kind(), error.to_string())),
2568 queued,
2569 })
2580 sync_status(
2581 &self.replica,
2582 self.worker.as_ref().and_then(SyncWorker::synced),
2583 &self.observed,
2584 )
25702585 }
25712586
25722587 /// The replica, which other threads may read pages from while the section is open.
crates/notebook/src/worker.rs+1-1
......@@ -14,7 +14,7 @@ pub(super) struct Signal {
1414 pub(crate) stopped: AtomicBool,
1515 pub(crate) offline: AtomicBool,
1616 /// FILETIME of the last step or poll that reached the remote; 0 before one has.
17 synced: AtomicU64,
17 pub(crate) synced: AtomicU64,
1818 /// Asked for by `wake`: steps run while offline until one leaves nothing to do.
1919 pub(crate) requested: AtomicBool,
2020 /// A watch on the file's folder wakes the worker when the file changes, so an idle worker