Merge 'mark completion as done only after callback will be executed' from Nikita Sivukhin

- otherwise, in multi-threading environment, other thread can think that
completion is finished and start execution
- this can lead to violated assertions (for example, page must be
loaded, but as callback is not executed yet assert will be fired)
Failing scenario:
1. main thread wants to execute pread - so it schedule IO and return
control to the caller
2. IO thread read data from the disk
3. IO thread executes complete(result)
4. complete func set result of the completion to Ok
5. main thread enter into the step loop again and check completion
status
6. completion marked as finished/is_completed - so main thread continue
execution
7. main thread check that page is loaded and fails with assertion -
because it's not loaded yet
8. IO thread executed the callback and finished the completion

Reviewed-by: Preston Thorpe <preston@turso.tech>

Closes #2922

core/io/mod.rs

@@ -226,27 +226,31 @@ impl Completion {
     }
 
     pub fn complete(&self, result: i32) {
-        if self.inner.result.set(None).is_ok() {
-            let result = Ok(result);
-            match &self.inner.completion_type {
-                CompletionType::Read(r) => r.callback(result),
-                CompletionType::Write(w) => w.callback(result),
-                CompletionType::Sync(s) => s.callback(result), // fix
-                CompletionType::Truncate(t) => t.callback(result),
-            };
-        }
+        let result = Ok(result);
+        match &self.inner.completion_type {
+            CompletionType::Read(r) => r.callback(result),
+            CompletionType::Write(w) => w.callback(result),
+            CompletionType::Sync(s) => s.callback(result), // fix
+            CompletionType::Truncate(t) => t.callback(result),
+        };
+        self.inner
+            .result
+            .set(None)
+            .expect("result must be set only once");
     }
 
     pub fn error(&self, err: CompletionError) {
-        if self.inner.result.set(Some(err)).is_ok() {
-            let result = Err(err);
-            match &self.inner.completion_type {
-                CompletionType::Read(r) => r.callback(result),
-                CompletionType::Write(w) => w.callback(result),
-                CompletionType::Sync(s) => s.callback(result), // fix
-                CompletionType::Truncate(t) => t.callback(result),
-            };
-        }
+        let result = Err(err);
+        match &self.inner.completion_type {
+            CompletionType::Read(r) => r.callback(result),
+            CompletionType::Write(w) => w.callback(result),
+            CompletionType::Sync(s) => s.callback(result), // fix
+            CompletionType::Truncate(t) => t.callback(result),
+        };
+        self.inner
+            .result
+            .set(Some(err))
+            .expect("result must be set only once");
     }
 
     pub fn abort(&self) {