1use super::*;
2use crate::task::{self, JoinHandle};
3use std::{
4 collections::hash_map::RandomState,
5 hash::BuildHasher,
6 sync::{
7 Arc, Mutex,
8 atomic::{AtomicBool, AtomicU64, Ordering},
9 },
10};
11use web_time::Instant;
12
13pub(super) struct Signal {
14 pub(crate) stopped: AtomicBool,
15 pub(crate) offline: AtomicBool,
16 /// FILETIME of the last step or poll that reached the remote; 0 before one has.
17 pub(crate) synced: AtomicU64,
18 /// Asked for by `wake`: steps run while offline until one leaves nothing to do.
19 pub(crate) requested: AtomicBool,
20 /// A watch on the file's folder wakes the worker when the file changes, so an idle worker
21 /// waits for that instead of checking the file on its own (`Background::hold`).
22 pub(crate) watched: AtomicBool,
23 pause: Mutex<Pause>,
24 sender: task::Sender<()>,
25}
26
27/// How local edits wait before they publish (`SyncWorker::set_pause`).
28#[derive(Default)]
29struct Pause {
30 length: Duration,
31 /// When the first and the last edit waiting arrived.
32 waiting: Option<(Instant, Instant)>,
33}
34
35/// The longest local edits wait for a pause in typing.
36const LONGEST: Duration = Duration::from_secs(30);
37
38impl Signal {
39 pub(crate) fn new() -> (Arc<Self>, task::Receiver<()>) {
40 let (sender, receiver) = task::channel();
41 let signal = Self {
42 stopped: AtomicBool::new(false),
43 offline: AtomicBool::new(false),
44 synced: AtomicU64::new(0),
45 requested: AtomicBool::new(false),
46 watched: AtomicBool::new(false),
47 pause: Mutex::default(),
48 sender,
49 };
50 (Arc::new(signal), receiver)
51 }
52
53 pub(super) fn wake(&self) {
54 // One retained notification covers edits that arrive during network I/O.
55 let _ = self.sender.try_send(());
56 }
57
58 /// A burst of local edits is durable.
59 pub(super) fn edited(&self) {
60 if let Ok(mut pause) = self.pause.lock() {
61 let now = Instant::now();
62 pause.waiting = Some((pause.waiting.map_or(now, |(first, _)| first), now));
63 }
64 self.wake();
65 }
66
67 /// How much longer waiting edits wait for a pause in typing, if they do.
68 fn paused(&self) -> Option<Duration> {
69 let pause = self.pause.lock().ok()?;
70 let (first, last) = pause.waiting?;
71 let now = Instant::now();
72 Some(
73 (last + pause.length)
74 .min(first + LONGEST)
75 .checked_duration_since(now)?,
76 )
77 .filter(|wait| !wait.is_zero())
78 }
79}
80
81/// Owns automatic reconciliation. Dropping requests cancellation without blocking.
82/// The in-flight sync step retains cache ownership until it finishes.
83pub struct SyncWorker {
84 signal: Arc<Signal>,
85 thread: Option<JoinHandle<Result<()>>>,
86}
87
88impl SyncWorker {
89 pub(crate) fn cancel(&self) {
90 self.signal.stopped.store(true, Ordering::Release);
91 self.signal.wake();
92 }
93
94 /// Requests a retry, for example after a network reachability change. Working
95 /// offline, it synchronizes once, as OneNote's Sync Now does.
96 /// A pending contention backoff finishes before processing the notification.
97 pub fn wake(&self) {
98 self.signal.requested.store(true, Ordering::Release);
99 self.signal.wake();
100 }
101
102 /// When a step or poll last reached the remote.
103 pub fn synced(&self) -> Option<u64> {
104 Some(self.signal.synced.load(Ordering::Acquire)).filter(|time| *time != 0)
105 }
106
107 /// Local edits publish once `pause` passes without another, or once they have waited
108 /// half a minute, rather than each burst at once: a cloud drive uploads every publication
109 /// and turns each concurrent one into a conflict version. `wake` publishes them at once.
110 pub fn set_pause(&self, pause: Duration) {
111 if let Ok(mut current) = self.signal.pause.lock() {
112 current.length = pause;
113 }
114 self.signal.wake();
115 }
116
117 /// Working offline, the worker neither connects nor steps: local edits stay queued
118 /// until `wake`, or until working online again.
119 pub fn set_offline(&self, offline: bool) {
120 self.signal.offline.store(offline, Ordering::Release);
121 self.signal.wake();
122 }
123
124 /// Cancels future steps; native threads finish the current step and callback before returning.
125 /// A stopped worker leaves pending edits and uncertain attempts in the cache.
126 /// Call outside the worker's own callback, which cannot join its calling thread.
127 pub fn stop(mut self) -> Result<()> {
128 self.cancel();
129 let thread = self.thread.take().expect("Worker owns its thread");
130 #[cfg(target_arch = "wasm32")]
131 return thread.finished().unwrap_or(Ok(()));
132 #[cfg(not(target_arch = "wasm32"))]
133 thread
134 .join()
135 .map_err(|_| io::Error::other("Synchronization worker panicked"))?
136 }
137
138 pub(crate) async fn stop_async(self) -> Result<()> {
139 #[cfg(not(target_arch = "wasm32"))]
140 return self.stop();
141 #[cfg(target_arch = "wasm32")]
142 {
143 let mut worker = self;
144 worker.signal.stopped.store(true, Ordering::Release);
145 worker.signal.wake();
146 let thread = worker.thread.take().expect("Worker owns its thread");
147 thread
148 .join()
149 .await
150 .map_err(|_| io::Error::other("Synchronization worker panicked"))?
151 }
152 }
153}
154
155impl Drop for SyncWorker {
156 fn drop(&mut self) {
157 self.cancel();
158 }
159}
160
161impl Replica {
162 /// Starts one worker, reconnecting through `connect` after transport failures.
163 /// Local edits wake it; `interval` controls idle polling, unless a watch wakes it
164 /// (`Background::hold`), and transport retries. While
165 /// nothing is queued, or the queue waits on a remote that has not changed since, a
166 /// remote whose `stamp` holds is not read again.
167 /// Contended operations returning `NotCommitted` also back off by up to one second.
168 /// `observe` runs on the worker after each attempt, including connection errors.
169 /// Cache/document errors stop the worker; inspect them through `observe` or `stop`.
170 /// Remote calls and callbacks must be bounded for `stop` to have bounded latency.
171 pub fn start_sync<R, F, O>(
172 self: &Arc<Self>,
173 interval: Duration,
174 connect: F,
175 observe: O,
176 ) -> io::Result<SyncWorker>
177 where
178 R: Remote + 'static,
179 F: FnMut() -> io::Result<R> + Send + 'static,
180 O: FnMut(&Result<Synced>) + Send + 'static,
181 {
182 self.start_sync_replacing(interval, false, connect, observe, None)
183 }
184
185 pub(crate) fn start_sync_replacing<R, F, O>(
186 self: &Arc<Self>,
187 interval: Duration,
188 offline: bool,
189 mut connect: F,
190 mut observe: O,
191 replacing: Option<&SyncWorker>,
192 ) -> io::Result<SyncWorker>
193 where
194 R: Remote + 'static,
195 F: FnMut() -> io::Result<R> + Send + 'static,
196 O: FnMut(&Result<Synced>) + Send + 'static,
197 {
198 if interval.is_zero() || Instant::now().checked_add(interval).is_none() {
199 return Err(io::Error::new(
200 io::ErrorKind::InvalidInput,
201 "Synchronization interval must be positive and representable",
202 ));
203 }
204 let mut owner = self
205 .section
206 .worker
207 .lock()
208 .map_err(|_| io::Error::other("Synchronization worker registration panicked"))?;
209 let previous = owner.upgrade();
210 if let Some(previous) = &previous
211 && !replacing.is_some_and(|worker| {
212 Arc::ptr_eq(previous, &worker.signal) && previous.stopped.load(Ordering::Acquire)
213 })
214 {
215 return Err(io::ErrorKind::WouldBlock.into());
216 }
217 let (signal, receiver) = Signal::new();
218 signal.offline.store(offline, Ordering::Release);
219 if let Some(previous) = previous {
220 signal
221 .offline
222 .store(previous.offline.load(Ordering::Acquire), Ordering::Release);
223 if let (Ok(previous), Ok(mut next)) = (previous.pause.lock(), signal.pause.lock()) {
224 next.length = previous.length;
225 next.waiting = previous.waiting;
226 }
227 }
228 let replica = Arc::clone(self);
229 let worker_signal = Arc::clone(&signal);
230 let thread = task::spawn("onestore-sync", move || async move {
231 let mut remote: Option<R> = None;
232 let jitter = RandomState::new();
233 let mut contention = 0_u32;
234 // The first step, and the first after a failure, always runs, so `observe`
235 // hears that the remote is reachable and what state the queue is in.
236 let mut reported = false;
237 // With nothing to do: a watched worker waits for the watch, any other looks
238 // again after `interval`.
239 let rest = || {
240 task::wait(
241 &receiver,
242 (!worker_signal.watched.load(Ordering::Acquire)).then_some(interval),
243 )
244 };
245 while !worker_signal.stopped.load(Ordering::Acquire) {
246 if worker_signal.offline.load(Ordering::Acquire)
247 && !worker_signal.requested.load(Ordering::Acquire)
248 {
249 remote = None;
250 reported = false;
251 task::wait(&receiver, None).await;
252 continue;
253 }
254 if !worker_signal.requested.load(Ordering::Acquire)
255 && let Some(wait) = worker_signal.paused()
256 {
257 task::wait(&receiver, Some(wait)).await;
258 continue;
259 }
260 if let Ok(mut pause) = worker_signal.pause.lock() {
261 // The step publishes what waits; edits from now on wait anew.
262 pause.waiting = None;
263 }
264 let result = match remote.as_mut() {
265 Some(remote) => {
266 if reported && replica.settled(remote).await.unwrap_or(false) {
267 worker_signal.requested.store(false, Ordering::Release);
268 worker_signal.synced.store(crate::now(), Ordering::Release);
269 rest().await;
270 continue;
271 }
272 let result = replica.sync_once_async(remote).await;
273 reported = result.is_ok();
274 result
275 }
276 None => match connect() {
277 Ok(connected) => {
278 remote = Some(connected);
279 continue;
280 }
281 Err(error) => Err(Error::RemoteIo(error)),
282 },
283 };
284 if result.is_ok() {
285 worker_signal.synced.store(crate::now(), Ordering::Release);
286 }
287 observe(&result);
288 let idle = matches!(result, Ok(Synced { edit: None, .. }));
289 match result {
290 Ok(Synced {
291 edit: Some((_, EditStatus::Published { .. })),
292 ..
293 }) => {
294 contention = 0;
295 continue;
296 }
297 Ok(Synced { edit: None, .. }) => contention = 0,
298 Ok(_) => {}
299 Err(Error::Remote(onestore::CommitError {
300 state: onestore::CommitState::NotCommitted,
301 ref error,
302 })) if matches!(
303 error.kind(),
304 io::ErrorKind::WouldBlock | io::ErrorKind::ResourceBusy
305 ) =>
306 {
307 contention = contention.saturating_add(1);
308 let ceiling = (50_u64 << contention.min(5)).min(1000);
309 let until = Instant::now()
310 + Duration::from_millis(jitter.hash_one(contention) % ceiling);
311 // Local wakes must not keep competing writers in the same retry phase.
312 while !worker_signal.stopped.load(Ordering::Acquire) {
313 let Some(remaining) = until.checked_duration_since(Instant::now())
314 else {
315 break;
316 };
317 task::wait(&receiver, Some(remaining)).await;
318 }
319 continue;
320 }
321 Err(Error::RemoteIo(ref error))
322 if matches!(
323 error.kind(),
324 io::ErrorKind::WouldBlock | io::ErrorKind::ResourceBusy
325 ) => {}
326 Err(Error::RemoteIo(_) | Error::Remote(_)) => remote = None,
327 Err(Error::Io(ref error)) if error.kind() == io::ErrorKind::WouldBlock => {}
328 Err(error) => return Err(error),
329 }
330 worker_signal.requested.store(false, Ordering::Release);
331 if idle {
332 rest().await;
333 } else if !worker_signal.stopped.load(Ordering::Acquire) {
334 task::wait(&receiver, Some(interval)).await;
335 }
336 }
337 Ok(())
338 })?;
339 *owner = Arc::downgrade(&signal);
340 Ok(SyncWorker {
341 signal,
342 thread: Some(thread),
343 })
344 }
345}