More documentation comments

Author: simolus3

crates/core/src/lib.rs

@@ -44,7 +44,10 @@ pub extern "C" fn sqlite3_powersync_init(
     api: *mut sqlite::api_routines,
 ) -> c_int {
     sqlite::EXTENSION_INIT2(api);
-    unsafe { SQLITE3_API = api };
+    unsafe {
+        // SAFETY: This field is only assigned once, when the library is loaded.
+        SQLITE3_API = api
+    };
 
     let result = init_extension(db);
 

crates/core/src/sync/interface.rs

@@ -16,20 +16,39 @@ use crate::error::SQLiteError;
 use super::streaming_sync::SyncClient;
 use super::sync_status::DownloadSyncStatus;
 
+/// A request sent from a client SDK to the [SyncClient] with a `powersync_control` invocation.
 pub enum SyncControlRequest<'a> {
+    /// The client requests to start a sync iteration.
+    ///
+    /// Earlier iterations are implicitly dropped when receiving this request.
     StartSyncStream {
+        /// Bucket parameters to include in the request when opening a sync stream.
         parameters: Option<serde_json::Map<String, serde_json::Value>>,
     },
+    /// The client requests to stop the current sync iteration.
     StopSyncStream,
+    /// The client is forwading a sync event to the core extension.
     SyncEvent(SyncEvent<'a>),
 }
 
 pub enum SyncEvent<'a> {
+    /// A synthetic event forwarded to the [SyncClient] after being started.
     Initialize,
+    /// An event requesting the sync client to shut down.
     TearDown,
+    /// Notifies the sync client that a token has been refreshed.
+    ///
+    /// In response, we'll stop the current iteration to begin another one with the new token.
     DidRefreshToken,
+    /// Notifies the sync client that the current CRUD upload (for which the client SDK is
+    /// responsible) has finished.
+    ///
+    /// If pending CRUD entries have previously prevented a sync from completing, this even can be
+    /// used to try again.
     UploadFinished,
+    /// Forward a text line (JSON) received from the sync service.
     TextLine { data: &'a str },
+    /// Forward a binary line (BSON) received from the sync service.
     BinaryLine { data: &'a [u8] },
 }
 
@@ -40,19 +59,26 @@ pub enum Instruction {
         severity: LogSeverity,
         line: Cow<'static, str>,
     },
+    /// Update the download status for the ongoing sync iteration.
     UpdateSyncStatus {
         status: Rc<RefCell<DownloadSyncStatus>>,
     },
-    EstablishSyncStream {
-        request: StreamingSyncRequest,
-    },
+    /// Connect to the sync service using the [StreamingSyncRequest] created by the core extension,
+    /// and then forward received lines via [SyncEvent::TextLine] and [SyncEvent::BinaryLine].
+    EstablishSyncStream { request: StreamingSyncRequest },
     FetchCredentials {
+        /// Whether the credentials currently used have expired.
+        ///
+        /// If false, this is a pre-fetch.
         did_expire: bool,
     },
     // These are defined like this because deserializers in Kotlin can't support either an
     // object or a literal value
+    /// Close the websocket / HTTP stream to the sync service.
     CloseSyncStream {},
+    /// Flush the file-system if it's non-durable (only applicable to the Dart SDK).
     FlushFileSystem {},
+    /// Notify that a sync has been completed, prompting client SDKs to clear earlier errors.
     DidCompleteSync {},
 }
 
@@ -79,6 +105,10 @@ pub struct BucketRequest {
     pub after: String,
 }
 
+/// Wrapper around a [SyncClient].
+///
+/// We allocate one instance of this per database (in [register]) - the [SyncClient] has an initial
+/// empty state that doesn't consume any resources.
 struct SqlController {
     client: SyncClient,
 }

crates/core/src/sync/storage_adapter.rs

@@ -24,7 +24,8 @@ use super::{
 /// An adapter for storing sync state.
 ///
 /// This is used to encapsulate some SQL queries used for the sync implementation, making the code
-/// in `streaming_sync.rs` easier to read.
+/// in `streaming_sync.rs` easier to read. It also allows caching some prepared statements that are
+/// used frequently as an optimization, but we're not taking advantage of that yet.
 pub struct StorageAdapter {
     pub db: *mut sqlite::sqlite3,
     progress_stmt: ManagedStmt,

crates/core/src/sync/streaming_sync.rs

@@ -32,8 +32,19 @@ use super::{
     Checksum,
 };
 
+/// The sync client implementation, responsible for parsing lines received by the sync service and
+/// persisting them to the database.
+///
+/// The client consumes no resources and prepares no statements until a sync iteration is
+/// initialized.
 pub struct SyncClient {
     db: *mut sqlite::sqlite3,
+    /// The current [ClientState] (essentially an optional [StreamingSyncIteration]).
+    ///
+    /// This is guarded behind a mutex so that we can mutate the state without forcing callers to
+    /// obtain a mutable reference to the [SyncClient] itself. It doesn't mean much in practice
+    /// because it's impossible to run two `powersync_control` calls on the same database connection
+    /// concurrently.
     state: Mutex<ClientState>,
 }
 
@@ -91,7 +102,9 @@ impl SyncClient {
 }
 
 enum ClientState {
+    /// No sync iteration is currently active.
     Idle,
+    /// A sync iteration has begun on the database.
     IterationActive(SyncIterationHandle),
 }
 
@@ -108,11 +121,19 @@ impl ClientState {
     }
 }
 
+/// A handle that allows progressing a [StreamingSyncIteration].
+///
+/// The sync itertion itself is implemented as an `async` function, as this allows us to treat it
+/// as a coroutine that preserves internal state between multiple `powersync_control` invocations.
+/// At each invocation, the future is polled once (and gets access to context that allows it to
+/// render [Instruction]s to return from the function).
 struct SyncIterationHandle {
     future: Pin<Box<dyn Future<Output = Result<(), SQLiteError>>>>,
 }
 
 impl SyncIterationHandle {
+    /// Creates a new sync iteration in a pending state by preparing statements for
+    /// [StorageAdapter] and setting up the initial downloading state for [StorageAdapter] .
     fn new(
         db: *mut sqlite::sqlite3,
         parameters: Option<serde_json::Map<String, serde_json::Value>>,
@@ -128,6 +149,8 @@ impl SyncIterationHandle {
         Ok(Self { future })
     }
 
+    /// Forwards a [SyncEvent::Initialize] to the current sync iteration, returning the initial
+    /// instructions generated.
     fn initialize(&mut self) -> Result<Vec<Instruction>, SQLiteError> {
         let mut event = ActiveEvent::new(SyncEvent::Initialize);
         let result = self.run(&mut event)?;
@@ -160,9 +183,12 @@ impl SyncIterationHandle {
     }
 }
 
+/// A [SyncEvent] currently being handled by a [StreamingSyncIteration].
 struct ActiveEvent<'a> {
     handled: bool,
+    /// The event to handle
     event: SyncEvent<'a>,
+    /// Instructions to forward to the client when the `powersync_control` invocation completes.
     instructions: Vec<Instruction>,
 }
 
@@ -420,6 +446,10 @@ impl StreamingSyncIteration {
         )?)
     }
 
+    /// Prepares a sync iteration by handling the initial [SyncEvent::Initialize].
+    ///
+    /// This prepares a [StreamingSyncRequest] by fetching local sync state and the requested bucket
+    /// parameters.
     async fn prepare_request(&mut self) -> Result<Vec<String>, SQLiteError> {
         let event = Self::receive_event().await;
         let SyncEvent::Initialize = event.event else {
@@ -484,6 +514,10 @@ impl SyncTarget {
         }
     }
 
+    /// Starts tracking the received `Checkpoint`.
+    ///
+    /// This updates the internal state and returns a set of buckets to delete because they've been
+    /// tracked locally but not in the new checkpoint.
     fn track_checkpoint<'a>(&mut self, checkpoint: &Checkpoint<'a>) -> BTreeSet<String> {
         let mut to_delete: BTreeSet<String> = match &self {
             SyncTarget::Tracking(checkpoint) => checkpoint.buckets.keys().cloned().collect(),

crates/core/src/sync/sync_status.rs

@@ -10,13 +10,25 @@ use super::{
     storage_adapter::PersistedBucketProgress, streaming_sync::OwnedCheckpoint,
 };
 
+/// Information about a progressing download.
 #[derive(Serialize, Hash)]
 pub struct DownloadSyncStatus {
+    /// Whether the socket to the sync service is currently open and connected.
+    ///
+    /// This starts being true once we receive the first line, and is set to false as the iteration
+    /// ends.
     pub connected: bool,
+    /// Whether we've requested the client SDK to connect to the socket while not receiving sync
+    /// lines yet.
     pub connecting: bool,
+    /// Provides stats over which bucket priorities have already been synced (or when they've last
+    /// been changed).
+    ///
     /// Always sorted by descending [BucketPriority] in [SyncPriorityStatus] (or, in other words,
     /// increasing priority numbers).
     pub priority_status: Vec<SyncPriorityStatus>,
+    /// When a download is active (that is, a `checkpoint` or `checkpoint_diff` line has been
+    /// received), information about how far the download has progressed.
     pub downloading: Option<SyncDownloadProgress>,
 }
 

crates/core/src/util.rs

@@ -106,12 +106,19 @@ pub(crate) static mut SQLITE3_API: *mut api_routines = ptr::null_mut();
 
 impl SqliteMutex {
     pub fn new() -> Self {
-        let native_alloc = unsafe { (*SQLITE3_API).mutex_alloc };
+        let native_alloc = unsafe {
+            // SAFETY: SQLITE3_API is only set once when the library is loaded by SQLite.
+            (*SQLITE3_API).mutex_alloc
+        };
 
         Self {
             ptr: match native_alloc {
                 None => null_mut(),
-                Some(mutex_alloc) => unsafe { mutex_alloc(SQLITE_MUTEX_FAST as i32) },
+                Some(mutex_alloc) => unsafe {
+                    // SAFETY: We're allowed to call sqlite3_mutex_alloc with this bitmask:
+                    // https://sqlite.org/c3ref/mutex_alloc.html
+                    mutex_alloc(SQLITE_MUTEX_FAST as i32)
+                },
             },
         }
     }
@@ -126,7 +133,11 @@ unsafe impl RawMutex for SqliteMutex {
         if self.ptr.is_null() {
             // Disable mutex code
         } else {
-            unsafe { (*SQLITE3_API).mutex_enter.unwrap_unchecked()(self.ptr) }
+            unsafe {
+                // SAFETY: When we get here, we were able to allocate a mutex (so mutex methods
+                // must be present).
+                (*SQLITE3_API).mutex_enter.unwrap_unchecked()(self.ptr)
+            }
         }
     }
 
@@ -135,7 +146,11 @@ unsafe impl RawMutex for SqliteMutex {
             // Disable mutex code
             true
         } else {
-            let res = unsafe { (*SQLITE3_API).mutex_try.unwrap_unchecked()(self.ptr) };
+            let res = unsafe {
+                // SAFETY: When we get here, we were able to allocate a mutex (so mutex methods
+                // must be present).
+                (*SQLITE3_API).mutex_try.unwrap_unchecked()(self.ptr)
+            };
             res == 0
         }
     }
@@ -144,21 +159,33 @@ unsafe impl RawMutex for SqliteMutex {
         if self.ptr.is_null() {
             // Disable mutex code
         } else {
-            unsafe { (*SQLITE3_API).mutex_leave.unwrap_unchecked()(self.ptr) }
+            unsafe {
+                // SAFETY: When we get here, we were able to allocate a mutex (so mutex methods
+                // must be present). Also, this method is only allowed to be called after a caller
+                // has locked the mutex before.
+                (*SQLITE3_API).mutex_leave.unwrap_unchecked()(self.ptr)
+            }
         }
     }
 }
 
 impl Drop for SqliteMutex {
     fn drop(&mut self) {
         if !self.ptr.is_null() {
-            unsafe { (*SQLITE3_API).mutex_free.unwrap_unchecked()(self.ptr) };
+            unsafe {
+                // SAFETY: The pointer points to a valid mutex we own. This means that we have been
+                // able to allocate a mutex, so mutex methods must be present.
+                (*SQLITE3_API).mutex_free.unwrap_unchecked()(self.ptr)
+            };
         }
     }
 }
 
 pub type Mutex<T> = MutexApi<SqliteMutex, T>;
 
+/// Creates a [Mutex] implementation using `sqlite3_mutex_enter` and `sqlite3_mutex_free`.
+///
+/// When SQLite has been compiled without mutexes, the returned mutex doesn't do anything.
 pub fn sqlite3_mutex<T>(value: T) -> Mutex<T> {
     let raw = SqliteMutex::new();
     MutexApi::from_raw(raw, value)