From 600dca61482e4b2f98e0ac4cb1bc373ebc7e1ac7 Mon Sep 17 00:00:00 2001
From: Dmitriy Ryajov <dryajov@gmail.com>
Date: Thu, 14 Sep 2023 18:34:20 -0600
Subject: [PATCH] adding serialization to query iter

---
 datastore/fsds.nim                  | 13 +++++++++++++
 datastore/sql/sqliteds.nim          | 12 +++++++++++-
 datastore/threads/threadproxyds.nim |  3 +++
 3 files changed, 27 insertions(+), 1 deletion(-)

diff --git a/datastore/fsds.nim b/datastore/fsds.nim
index 163a52a..5526d5c 100644
--- a/datastore/fsds.nim
+++ b/datastore/fsds.nim
@@ -188,10 +188,23 @@ method query*(
   var
     iter = QueryIter.new()
 
+  let lock = newAsyncLock()
   proc next(): Future[?!QueryResponse] {.async.} =
+    defer:
+      if lock.locked:
+        lock.release()
+
+    if lock.locked:
+      return failure (ref DatastoreError)(msg: "Should always await query features")
+
     let
       path = walker()
 
+    if iter.finished:
+      return failure "iterator is finished"
+
+    await lock.acquire()
+
     if finished(walker):
       iter.finished = true
       return success (Key.none, EmptyBytes)
diff --git a/datastore/sql/sqliteds.nim b/datastore/sql/sqliteds.nim
index aa63274..e5fad22 100644
--- a/datastore/sql/sqliteds.nim
+++ b/datastore/sql/sqliteds.nim
@@ -151,9 +151,19 @@ method query*(
     if not (v == SQLITE_OK):
       return failure newException(DatastoreError, $sqlite3_errstr(v))
 
+  let lock = newAsyncLock()
   proc next(): Future[?!QueryResponse] {.async.} =
+    defer:
+      if lock.locked:
+        lock.release()
+
+    if lock.locked:
+      return failure (ref DatastoreError)(msg: "Should always await query features")
+
     if iter.finished:
-      return failure(newException(QueryEndedError, "Calling next on a finished query!"))
+      return failure((ref QueryEndedError)(msg: "Calling next on a finished query!"))
+
+    await lock.acquire()
 
     let
       v = sqlite3_step(s)
diff --git a/datastore/threads/threadproxyds.nim b/datastore/threads/threadproxyds.nim
index f497515..673e674 100644
--- a/datastore/threads/threadproxyds.nim
+++ b/datastore/threads/threadproxyds.nim
@@ -287,6 +287,9 @@ method query*(
       if lock.locked:
         lock.release()
 
+    if lock.locked:
+      return failure (ref DatastoreError)(msg: "Should always await query features")
+
     if iter.finished == true:
       return failure (ref QueryEndedError)(msg: "Calling next on a finished query!")