fix: better (??) thread synchronization

tests/cbindings/storage.c

@@ -1,8 +1,11 @@
+#include <errno.h>
+#include <pthread.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
 #include <stdint.h>
 #include <stdbool.h>
+#include <time.h>
 #include <unistd.h>
 #include "../../library/libstorage.h"
 
@@ -51,6 +54,9 @@
 
 typedef struct
 {
+    pthread_mutex_t mutex;
+    pthread_cond_t cond;
+    bool done;
     int ret;
     char *msg;
     char *chunk;
@@ -60,6 +66,9 @@ typedef struct
 static Resp *alloc_resp(void)
 {
     Resp *r = (Resp *)calloc(1, sizeof(Resp));
+    pthread_mutex_init(&r->mutex, NULL);
+    pthread_cond_init(&r->cond, NULL);
+    r->done = false;
     r->msg = NULL;
     r->chunk = NULL;
     r->ret = -1;
@@ -83,6 +92,8 @@ static void free_resp(Resp *r)
         free(r->chunk);
     }
 
+    pthread_cond_destroy(&r->cond);
+    pthread_mutex_destroy(&r->mutex);
     free(r);
 }
 
@@ -93,7 +104,11 @@ static int get_ret(Resp *r)
         return RET_ERR;
     }
 
-    return r->ret;
+    pthread_mutex_lock(&r->mutex);
+    int ret = r->ret;
+    pthread_mutex_unlock(&r->mutex);
+
+    return ret;
 }
 
 // wait_resp waits until the async response is ready or max retries is reached.
@@ -101,13 +116,33 @@ static int get_ret(Resp *r)
 // indicate that the response is ready to be consumed.
 static void wait_resp(Resp *r)
 {
-    int retries = 0;
-
-    while (get_ret(r) == -1 && retries < MAX_RETRIES)
+    if (!r)
     {
-        usleep(1000 * 100); // 100 ms
-        retries++;
+        return;
     }
+
+    const long timeout_ms = MAX_RETRIES * 100;
+    struct timespec deadline;
+
+    clock_gettime(CLOCK_REALTIME, &deadline);
+    deadline.tv_sec += timeout_ms / 1000;
+    deadline.tv_nsec += (timeout_ms % 1000) * 1000000;
+    if (deadline.tv_nsec >= 1000000000)
+    {
+        deadline.tv_sec += 1;
+        deadline.tv_nsec -= 1000000000;
+    }
+
+    pthread_mutex_lock(&r->mutex);
+    while (!r->done)
+    {
+        int rc = pthread_cond_timedwait(&r->cond, &r->mutex, &deadline);
+        if (rc == ETIMEDOUT)
+        {
+            break;
+        }
+    }
+    pthread_mutex_unlock(&r->mutex);
 }
 
 // is_resp_ok checks if the async response indicates success.
@@ -122,6 +157,8 @@ static int is_resp_ok(Resp *r, char **res)
 
     wait_resp(r);
 
+    pthread_mutex_lock(&r->mutex);
+
     int ret = (r->ret == RET_OK) ? RET_OK : RET_ERR;
 
     // If a response pointer is provided, it’s safe to initialize it to NULL.
@@ -142,6 +179,8 @@ static int is_resp_ok(Resp *r, char **res)
         *res = strdup(r->msg);
     }
 
+    pthread_mutex_unlock(&r->mutex);
+
     free_resp(r);
 
     return ret;
@@ -168,6 +207,8 @@ static void callback(int ret, const char *msg, size_t len, void *userData)
         return;
     }
 
+    pthread_mutex_lock(&r->mutex);
+
     // If the reponse already has a message, just free it first.
     if (r->msg)
     {
@@ -184,8 +225,8 @@ static void callback(int ret, const char *msg, size_t len, void *userData)
         r->len = len;
     }
 
-    // For other cases, copy the message data.
-    if (msg && len > 0)
+    // For terminal responses, copy the message data.
+    if (ret != RET_PROGRESS && msg && len > 0)
     {
         // Allocate memory for the message plus null terminator.
         r->msg = (char *)malloc(len + 1);
@@ -194,6 +235,10 @@ static void callback(int ret, const char *msg, size_t len, void *userData)
         if (!r->msg)
         {
             r->len = 0;
+            r->ret = RET_ERR;
+            r->done = true;
+            pthread_cond_signal(&r->cond);
+            pthread_mutex_unlock(&r->mutex);
             return;
         }
 
@@ -205,15 +250,25 @@ static void callback(int ret, const char *msg, size_t len, void *userData)
 
         r->len = len;
     }
-    else
+    else if (ret != RET_PROGRESS)
     {
         r->msg = NULL;
         r->len = 0;
     }
 
-    // Publish the return code last. wait_resp uses this as the completion flag,
-    // so setting it earlier lets the caller read/free a partially written Resp.
+    // Progress updates are intermediate callbacks. Keep any copied chunk data,
+    // but wait for the final RET_OK/RET_ERR before completing the response.
+    if (ret == RET_PROGRESS)
+    {
+        pthread_mutex_unlock(&r->mutex);
+        return;
+    }
+
+    // Publish completion last so wait_resp can only observe a fully written Resp.
     r->ret = ret;
+    r->done = true;
+    pthread_cond_signal(&r->cond);
+    pthread_mutex_unlock(&r->mutex);
 }
 
 static int read_file(const char *filepath, char **res)
@@ -257,7 +312,7 @@ int setup(void **storage_ctx)
 
     wait_resp(r);
 
-    if (r->ret != RET_OK)
+    if (get_ret(r) != RET_OK)
     {
         free_resp(r);
         return RET_ERR;