diff --git a/core/src/main/scala/lightdb/Query.scala b/core/src/main/scala/lightdb/Query.scala
index 3f516bb1..421ce18c 100644
--- a/core/src/main/scala/lightdb/Query.scala
+++ b/core/src/main/scala/lightdb/Query.scala
@@ -14,8 +14,7 @@ import lightdb.materialized.{MaterializedAndDoc, MaterializedIndex}
 import lightdb.spatial.{DistanceAndDoc, Geo}
 import lightdb.store.{Conversion, Store, StoreMode}
 import lightdb.transaction.Transaction
-import lightdb.util.GroupedIterator
-import rapid.{Forge, Task}
+import rapid.{Forge, Grouped, Task}
 
 case class Query[Doc <: Document[Doc], Model <: DocumentModel[Doc]](model: Model,
                                                                     store: Store[Doc, Model],
@@ -283,18 +282,16 @@ case class Query[Doc <: Document[Doc], Model <: DocumentModel[Doc]](model: Model
   def aggregate(f: Model => List[AggregateFunction[_, _, Doc]]): AggregateQuery[Doc, Model] =
     AggregateQuery(this, f(model))
 
-  // TODO: Support this via stream
-  /*def grouped[F](f: Model => Field[Doc, F],
+  def grouped[F](f: Model => Field[Doc, F],
                  direction: SortDirection = SortDirection.Ascending)
-                (implicit transaction: Transaction[Doc]): GroupedIterator[Doc, F] = {
+                (implicit transaction: Transaction[Doc]): rapid.Stream[Grouped[F, Doc]] = {
     val field = f(model)
     val state = new IndexingState
-    val iterator = sort(Sort.ByField(field, direction))
-      .search
+    sort(Sort.ByField(field, direction))
+      .stream
       .docs
-      .iterator
-    GroupedIterator[Doc, F](iterator, doc => field.get(doc, field, state))
-  }*/
+      .groupSequential(doc => field.get(doc, field, state))
+  }
 }
 
 object Query {
diff --git a/core/src/main/scala/lightdb/doc/MaterializedBatchModel.scala b/core/src/main/scala/lightdb/doc/MaterializedBatchModel.scala
index ae6b10a3..7b38c000 100644
--- a/core/src/main/scala/lightdb/doc/MaterializedBatchModel.scala
+++ b/core/src/main/scala/lightdb/doc/MaterializedBatchModel.scala
@@ -2,6 +2,7 @@ package lightdb.doc
 
 import lightdb.Id
 import lightdb.transaction.Transaction
+import rapid.Task
 
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.atomic.AtomicInteger
@@ -12,7 +13,7 @@ trait MaterializedBatchModel[Doc <: Document[Doc], MaterialDoc <: Document[Mater
 
   private val map = new ConcurrentHashMap[Transaction[MaterialDoc], TransactionState]
 
-  private def changed(docState: DocState[MaterialDoc])(implicit transaction: Transaction[MaterialDoc]): Unit = {
+  private def changed(docState: DocState[MaterialDoc])(implicit transaction: Transaction[MaterialDoc]): Task[Unit] = Task {
     map.compute(transaction, (_, current) => {
       val state = Option(current).getOrElse(new TransactionState)
       state.changed(docState)
@@ -22,28 +23,30 @@ trait MaterializedBatchModel[Doc <: Document[Doc], MaterialDoc <: Document[Mater
     if (state.size > maxBatchSize) {
       val list = state.process()
       process(list)
+    } else {
+      Task.unit
     }
-  }
+  }.flatten
 
-  protected def process(list: List[List[DocState[MaterialDoc]]]): Unit
+  protected def process(list: List[List[DocState[MaterialDoc]]]): Task[Unit]
 
-  override protected def adding(doc: MaterialDoc)(implicit transaction: Transaction[MaterialDoc]): Unit =
+  override protected def adding(doc: MaterialDoc)(implicit transaction: Transaction[MaterialDoc]): Task[Unit] =
     changed(DocState.Added(doc))
 
   override protected def modifying(oldDoc: MaterialDoc, newDoc: MaterialDoc)
-                                  (implicit transaction: Transaction[MaterialDoc]): Unit =
+                                  (implicit transaction: Transaction[MaterialDoc]): Task[Unit] =
     changed(DocState.Modified(newDoc))
 
-  override protected def removing(doc: MaterialDoc)(implicit transaction: Transaction[MaterialDoc]): Unit =
+  override protected def removing(doc: MaterialDoc)(implicit transaction: Transaction[MaterialDoc]): Task[Unit] =
     changed(DocState.Removed(doc))
 
-  override protected def transactionStart(transaction: Transaction[MaterialDoc]): Unit = {}
+  override protected def transactionStart(transaction: Transaction[MaterialDoc]): Task[Unit] = Task.unit
 
-  override protected def transactionEnd(transaction: Transaction[MaterialDoc]): Unit = {
+  override protected def transactionEnd(transaction: Transaction[MaterialDoc]): Task[Unit] = Task {
     Option(map.get(transaction)).foreach { state =>
       if (state.size > 0) {
         val list = state.process()
-        process(list)
+        process(list).sync()
       }
       map.remove(transaction)
     }
diff --git a/core/src/main/scala/lightdb/doc/MaterializedModel.scala b/core/src/main/scala/lightdb/doc/MaterializedModel.scala
index 8737e02e..6213488a 100644
--- a/core/src/main/scala/lightdb/doc/MaterializedModel.scala
+++ b/core/src/main/scala/lightdb/doc/MaterializedModel.scala
@@ -6,29 +6,25 @@ import lightdb.transaction.Transaction
 import lightdb.trigger.BasicCollectionTrigger
 import rapid.Task
 
-import java.util.concurrent.ConcurrentHashMap
-import java.util.concurrent.atomic.AtomicInteger
-import scala.jdk.CollectionConverters._
-
 trait MaterializedModel[Doc <: Document[Doc], MaterialDoc <: Document[MaterialDoc], MaterialModel <: DocumentModel[MaterialDoc]] extends DocumentModel[Doc] { mm =>
   def materialCollection: Collection[MaterialDoc, MaterialModel]
 
-  protected def adding(doc: MaterialDoc)(implicit transaction: Transaction[MaterialDoc]): Unit
-  protected def modifying(oldDoc: MaterialDoc, newDoc: MaterialDoc)(implicit transaction: Transaction[MaterialDoc]): Unit
-  protected def removing(doc: MaterialDoc)(implicit transaction: Transaction[MaterialDoc]): Unit
-  protected def transactionStart(transaction: Transaction[MaterialDoc]): Unit
-  protected def transactionEnd(transaction: Transaction[MaterialDoc]): Unit
+  protected def adding(doc: MaterialDoc)(implicit transaction: Transaction[MaterialDoc]): Task[Unit]
+  protected def modifying(oldDoc: MaterialDoc, newDoc: MaterialDoc)(implicit transaction: Transaction[MaterialDoc]): Task[Unit]
+  protected def removing(doc: MaterialDoc)(implicit transaction: Transaction[MaterialDoc]): Task[Unit]
+  protected def transactionStart(transaction: Transaction[MaterialDoc]): Task[Unit]
+  protected def transactionEnd(transaction: Transaction[MaterialDoc]): Task[Unit]
 
   override def init[Model <: DocumentModel[Doc]](collection: Collection[Doc, Model]): Task[Unit] = {
     super.init(collection).map { _ =>
       materialCollection.trigger += new BasicCollectionTrigger[MaterialDoc, MaterialModel] {
         override def collection: Collection[MaterialDoc, MaterialModel] = materialCollection
 
-        override protected def adding(doc: MaterialDoc)(implicit transaction: Transaction[MaterialDoc]): Task[Unit] = Task(mm.adding(doc))
-        override protected def modifying(oldDoc: MaterialDoc, newDoc: MaterialDoc)(implicit transaction: Transaction[MaterialDoc]): Task[Unit] = Task(mm.modifying(oldDoc, newDoc))
-        override protected def removing(doc: MaterialDoc)(implicit transaction: Transaction[MaterialDoc]): Task[Unit] = Task(mm.removing(doc))
-        override def transactionStart(transaction: Transaction[MaterialDoc]): Task[Unit] = Task(mm.transactionStart(transaction))
-        override def transactionEnd(transaction: Transaction[MaterialDoc]): Task[Unit] = Task(mm.transactionEnd(transaction))
+        override protected def adding(doc: MaterialDoc)(implicit transaction: Transaction[MaterialDoc]): Task[Unit] = mm.adding(doc)
+        override protected def modifying(oldDoc: MaterialDoc, newDoc: MaterialDoc)(implicit transaction: Transaction[MaterialDoc]): Task[Unit] = mm.modifying(oldDoc, newDoc)
+        override protected def removing(doc: MaterialDoc)(implicit transaction: Transaction[MaterialDoc]): Task[Unit] = mm.removing(doc)
+        override def transactionStart(transaction: Transaction[MaterialDoc]): Task[Unit] = mm.transactionStart(transaction)
+        override def transactionEnd(transaction: Transaction[MaterialDoc]): Task[Unit] = mm.transactionEnd(transaction)
       }
     }
   }
diff --git a/core/src/main/scala/lightdb/transaction/Transaction.scala b/core/src/main/scala/lightdb/transaction/Transaction.scala
index d3657ec1..8e0f0b18 100644
--- a/core/src/main/scala/lightdb/transaction/Transaction.scala
+++ b/core/src/main/scala/lightdb/transaction/Transaction.scala
@@ -2,26 +2,21 @@ package lightdb.transaction
 
 import lightdb.doc.Document
 import lightdb.feature.FeatureSupport
+import rapid.Task
 
 final class Transaction[Doc <: Document[Doc]] extends FeatureSupport[TransactionKey] { transaction =>
-  def commit(): Unit = {
-    features.foreach {
-      case f: TransactionFeature => f.commit()
-      case _ => // Ignore
-    }
-  }
+  def commit(): Task[Unit] = features.map {
+    case f: TransactionFeature => f.commit()
+    case _ => Task.unit // Ignore
+  }.tasks.unit
 
-  def rollback(): Unit = {
-    features.foreach {
-      case f: TransactionFeature => f.rollback()
-      case _ => // Ignore
-    }
-  }
+  def rollback(): Task[Unit] = features.map {
+    case f: TransactionFeature => f.rollback()
+    case _ => Task.unit // Ignore
+  }.tasks.unit
 
-  def close(): Unit = {
-    features.foreach {
-      case f: TransactionFeature => f.close()
-      case _ => // Ignore
-    }
-  }
+  def close(): Task[Unit] = features.map {
+    case f: TransactionFeature => f.close()
+    case _ => Task.unit // Ignore
+  }.tasks.unit
 }
\ No newline at end of file
diff --git a/core/src/main/scala/lightdb/transaction/TransactionFeature.scala b/core/src/main/scala/lightdb/transaction/TransactionFeature.scala
index 11eae0d1..d67680bf 100644
--- a/core/src/main/scala/lightdb/transaction/TransactionFeature.scala
+++ b/core/src/main/scala/lightdb/transaction/TransactionFeature.scala
@@ -1,9 +1,11 @@
 package lightdb.transaction
 
+import rapid.Task
+
 trait TransactionFeature {
-  def commit(): Unit = {}
+  def commit(): Task[Unit] = Task.unit
 
-  def rollback(): Unit = {}
+  def rollback(): Task[Unit] = Task.unit
 
-  def close(): Unit = {}
+  def close(): Task[Unit] = Task.unit
 }
diff --git a/core/src/main/scala/lightdb/util/GroupedIterator.scala b/core/src/main/scala/lightdb/util/GroupedIterator.scala
deleted file mode 100644
index 6540c902..00000000
--- a/core/src/main/scala/lightdb/util/GroupedIterator.scala
+++ /dev/null
@@ -1,32 +0,0 @@
-package lightdb.util
-
-/**
- * Convenience Iterator that groups sorted elements together based on the grouper function.
- *
- * Note: this is only useful is the underlying iterator is properly sorted by G.
- */
-case class GroupedIterator[T, G](i: Iterator[T], grouper: T => G) extends Iterator[(G, List[T])] {
-  private var current: Option[(G, T)] = if (i.hasNext) {
-    val t = i.next()
-    Some(grouper(t), t)
-  } else {
-    None
-  }
-
-  override def hasNext: Boolean = current.isDefined
-
-  override def next(): (G, List[T]) = current match {
-    case Some((group, value)) =>
-      current = None
-      group -> (value :: i.takeWhile { t =>
-        val g = grouper(t)
-        if (g != group) {
-          current = Some((g, t))
-          false
-        } else {
-          true
-        }
-      }.toList)
-    case None => throw new NoSuchElementException("next on empty iterator")
-  }
-}
\ No newline at end of file
diff --git a/core/src/main/scala/lightdb/util/Initializable.scala b/core/src/main/scala/lightdb/util/Initializable.scala
index 9641843e..aeb73f26 100644
--- a/core/src/main/scala/lightdb/util/Initializable.scala
+++ b/core/src/main/scala/lightdb/util/Initializable.scala
@@ -2,30 +2,22 @@ package lightdb.util
 
 import rapid._
 
-import java.util.concurrent.atomic.AtomicInteger
-
 /**
  * Provides simple initialization support to avoid initialization being invoked more
  * than once. FlatMap on `init` to safely guarantee initialization was successful.
  */
 trait Initializable {
-  private val status = new AtomicInteger(0)
+  @volatile private var initialized = false
+  private lazy val singleton = initialize().map { _ =>
+    initialized = true
+  }.singleton
 
-  def isInitialized: Boolean = status.get() == 2
+  def isInitialized: Boolean = initialized
 
   /**
    * Calls initialize() exactly one time. Safe to call multiple times.
    */
-  final def init(): Task[Boolean] = Task {
-    if (status.compareAndSet(0, 1)) {
-      initialize().map { _ =>
-        status.set(2)
-        true
-      }
-    } else {
-      Task.pure(false)
-    }
-  }.flatten
+  final def init(): Task[Unit] = singleton
 
   /**
    * Define initialization functionality here, but never call directly.
diff --git a/core/src/test/scala/spec/AbstractBasicSpec.scala b/core/src/test/scala/spec/AbstractBasicSpec.scala
index 11238b94..28febe55 100644
--- a/core/src/test/scala/spec/AbstractBasicSpec.scala
+++ b/core/src/test/scala/spec/AbstractBasicSpec.scala
@@ -15,7 +15,7 @@ import lightdb.{Id, LightDB, Sort, StoredValue}
 import org.scalatest.matchers.should.Matchers
 import org.scalatest.wordspec.{AnyWordSpec, AsyncWordSpec}
 import perfolation.double2Implicits
-import rapid.AsyncTaskSpec
+import rapid.{AsyncTaskSpec, Task}
 
 import java.io.File
 import java.nio.file.Path
@@ -69,7 +69,7 @@ abstract class AbstractBasicSpec extends AsyncWordSpec with AsyncTaskSpec with M
 
   specName should {
     "initialize the database" in {
-      db.init().map(_ should be(true))
+      db.init().succeed
     }
     "verify the database is empty" in {
       db.people.transaction { implicit transaction =>
@@ -151,26 +151,29 @@ abstract class AbstractBasicSpec extends AsyncWordSpec with AsyncTaskSpec with M
     }
     "sort by age" in {
       db.people.transaction { implicit transaction =>
-        val people = db.people.query.sort(Sort.ByField(Person.age).descending).search.docs.list
-        people.map(_.name).take(3) should be(List("Ruth", "Zoey", "Quintin"))
+        db.people.query.sort(Sort.ByField(Person.age).descending).stream.docs.toList.map { people =>
+          people.map(_.name).take(3) should be(List("Ruth", "Zoey", "Quintin"))
+        }
       }
     }
     "group by age" in {
       db.people.transaction { implicit transaction =>
-        val list = db.people.query.grouped(_.age).toList
-        list.map(_._1) should be(List(2, 4, 11, 12, 13, 15, 21, 22, 23, 30, 33, 35, 42, 53, 62, 63, 72, 81, 89, 99, 101, 102))
-        list.map(_._2.map(_.name).toSet) should be(List(
-          Set("Penny"), Set("Jenna"), Set("Brenda"), Set("Greg"), Set("Veronica"), Set("Diana"),
-          Set("Adam", "Uba", "Oscar"), Set("Nancy"), Set("Fiona"), Set("Tori", "Yuri", "Wyatt"), Set("Kevin"),
-          Set("Charlie"), Set("Mike"), Set("Evan"), Set("Hanna"), Set("Xena"), Set("Linda"), Set("Sam"), Set("Ian"),
-          Set("Quintin"), Set("Zoey"), Set("Ruth")
-        ))
+        db.people.query.grouped(_.age).toList.map { list =>
+          list.map(_.group) should be(List(2, 4, 11, 12, 13, 15, 21, 22, 23, 30, 33, 35, 42, 53, 62, 63, 72, 81, 89, 99, 101, 102))
+          list.map(_.results.map(_.name).toSet) should be(List(
+            Set("Penny"), Set("Jenna"), Set("Brenda"), Set("Greg"), Set("Veronica"), Set("Diana"),
+            Set("Adam", "Uba", "Oscar"), Set("Nancy"), Set("Fiona"), Set("Tori", "Yuri", "Wyatt"), Set("Kevin"),
+            Set("Charlie"), Set("Mike"), Set("Evan"), Set("Hanna"), Set("Xena"), Set("Linda"), Set("Sam"), Set("Ian"),
+            Set("Quintin"), Set("Zoey"), Set("Ruth")
+          ))
+        }
       }
     }
     "delete some records" in {
       db.people.transaction { implicit transaction =>
-        db.people.delete(_._id -> linda._id) should be(true)
-        db.people.delete(_._id -> yuri._id) should be(true)
+        db.people.delete(_._id -> linda._id).and(db.people.delete(_._id -> yuri._id)).map { t =>
+          t should be(true -> true)
+        }
       }
     }
     "query with multiple nots" in {
@@ -180,143 +183,165 @@ abstract class AbstractBasicSpec extends AsyncWordSpec with AsyncTaskSpec with M
             .mustNot(_.age < 30)
             .mustNot(_.age > 35)
         }
-        val list = query.toList
-        list.map(_.name).toSet should be(Set("Charlie", "Kevin", "Tori", "Wyatt"))
+        query.toList.map { list =>
+          list.map(_.name).toSet should be(Set("Charlie", "Kevin", "Tori", "Wyatt"))
+        }
       }
     }
     "verify the records were deleted" in {
       db.people.transaction { implicit transaction =>
-        db.people.count should be(24)
+        db.people.count.map(_ should be(24))
       }
     }
     // TODO: Fix same transaction modifying the same record concurrently
     "modify a record" in {
       db.people.transaction { implicit transaction =>
         db.people.modify(adam._id) {
-          case Some(p) => Some(p.copy(name = "Allan"))
+          case Some(p) => Task.pure(Some(p.copy(name = "Allan")))
           case None => fail("Adam was not found!")
         }
-      } match {
+      }.map {
         case Some(p) => p.name should be("Allan")
         case None => fail("Allan was not returned!")
       }
     }
     "verify the record has been renamed" in {
       db.people.transaction { implicit transaction =>
-        db.people(_._id -> adam._id).name should be("Allan")
+        db.people(_._id -> adam._id).map(_.name should be("Allan"))
       }
     }
     "verify start time has been set" in {
-      db.startTime.get() should be > 0L
+      db.startTime.get().map(_ should be > 0L)
     }
     "dispose the database and prepare new instance" in {
       if (memoryOnly) {
         // Don't dispose
+        Task.unit.succeed
       } else {
-        db.dispose()
-        db = new DB
-        db.init()
+        db.dispose().flatMap { _ =>
+          db = new DB
+          db.init()
+        }.succeed
       }
     }
     "query the database to verify records were persisted properly" in {
       db.people.transaction { implicit transaction =>
-        db.people.iterator.toList.map(_.name).toSet should be(Set(
-          "Tori", "Ruth", "Nancy", "Jenna", "Hanna", "Wyatt", "Diana", "Ian", "Quintin", "Uba", "Oscar", "Kevin",
-          "Penny", "Charlie", "Evan", "Sam", "Mike", "Brenda", "Zoey", "Allan", "Xena", "Fiona", "Greg", "Veronica"
-        ))
+        db.people.stream.toList.map(_.map(_.name).toSet).map { set =>
+          set should be(Set(
+            "Tori", "Ruth", "Nancy", "Jenna", "Hanna", "Wyatt", "Diana", "Ian", "Quintin", "Uba", "Oscar", "Kevin",
+            "Penny", "Charlie", "Evan", "Sam", "Mike", "Brenda", "Zoey", "Allan", "Xena", "Fiona", "Greg", "Veronica"
+          ))
+        }
       }
     }
     "search using tokenized data and a parsed query" in {
       db.people.transaction { implicit transaction =>
-        val people = db.people.query.filter(_.search.words("nica 13", matchEndsWith = true)).toList
-        people.map(_.name) should be(List("Veronica"))
+        db.people.query.filter(_.search.words("nica 13", matchEndsWith = true)).toList.map { people =>
+          people.map(_.name) should be(List("Veronica"))
+        }
       }
     }
     "search using Filter.Builder and scoring" in {
       if (filterBuilderSupported) {
         db.people.transaction { implicit transaction =>
-          val results = db.people.query.scored.filter(_
+          db.people.query.scored.filter(_
             .builder
             .minShould(0)
             .should(_.search.words("nica 13", matchEndsWith = true), boost = Some(2.0))
             .should(_.age <=> (10, 15))
-          ).search.docs
-          val people = results.list
-          people.map(_.name) should be(List("Veronica", "Brenda", "Diana", "Greg", "Charlie", "Evan", "Fiona", "Hanna", "Ian", "Jenna", "Kevin", "Mike", "Nancy", "Oscar", "Penny", "Quintin", "Ruth", "Sam", "Tori", "Uba", "Wyatt", "Xena", "Zoey", "Allan"))
-          results.scores should be(List(6.0, 2.0, 2.0, 2.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0))
+          ).search.docs.flatMap { results =>
+            results.list.map { people =>
+              people.map(_.name) should be(List("Veronica", "Brenda", "Diana", "Greg", "Charlie", "Evan", "Fiona", "Hanna", "Ian", "Jenna", "Kevin", "Mike", "Nancy", "Oscar", "Penny", "Quintin", "Ruth", "Sam", "Tori", "Uba", "Wyatt", "Xena", "Zoey", "Allan"))
+              results.scores should be(List(6.0, 2.0, 2.0, 2.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0))
+            }
+          }
         }
+      } else {
+        Task.unit.succeed
       }
     }
     "search where city is not set" in {
       db.people.transaction { implicit transaction =>
-        val people = db.people.query.filter(_.city === None).toList
-        people.map(_.name).toSet should be(Set("Tori", "Ruth", "Sam", "Nancy", "Jenna", "Hanna", "Wyatt", "Diana", "Ian", "Quintin", "Uba", "Oscar", "Kevin", "Penny", "Charlie", "Mike", "Brenda", "Zoey", "Allan", "Xena", "Fiona", "Greg", "Veronica"))
+        db.people.query.filter(_.city === None).toList.map { people =>
+          people.map(_.name).toSet should be(Set("Tori", "Ruth", "Sam", "Nancy", "Jenna", "Hanna", "Wyatt", "Diana", "Ian", "Quintin", "Uba", "Oscar", "Kevin", "Penny", "Charlie", "Mike", "Brenda", "Zoey", "Allan", "Xena", "Fiona", "Greg", "Veronica"))
+        }
       }
     }
     "search where city is set" in {
       db.people.transaction { implicit transaction =>
-        val people = db.people.query.filter(_.builder.mustNot(_.city === None)).toList
-        people.map(_.name) should be(List("Evan"))
+        db.people.query.filter(_.builder.mustNot(_.city === None)).toList.map { people =>
+          people.map(_.name) should be(List("Evan"))
+        }
       }
     }
     "update the city for a user" in {
       db.people.transaction { implicit transaction =>
-        val p = db.people(zoey._id)
-        db.people.upsert(p.copy(city = Some(City("Los Angeles"))))
-      }
+        db.people(zoey._id).flatMap { p =>
+          db.people.upsert(p.copy(city = Some(City("Los Angeles"))))
+        }
+      }.succeed
     }
     "modify a record within a transaction and see it post-commit" in {
       db.people.transaction { implicit transaction =>
-        val original = db.people.query.filter(_.name === "Ruth").toList.head
-        db.people.upsert(original.copy(
-          name = "Not Ruth"
-        ))
-        transaction.commit()
-        val people = db.people.query.filter(_.name === "Not Ruth").toList
-        people.map(_.name) should be(List("Not Ruth"))
+        for {
+          original <- db.people.query.filter(_.name === "Ruth").first
+          _ <- db.people.upsert(original.copy(
+            name = "Not Ruth"
+          ))
+          _ <- transaction.commit()
+          people <- db.people.query.filter(_.name === "Not Ruth").toList
+        } yield people.map(_.name) should be(List("Not Ruth"))
       }
     }
     "query with single-value nicknames" in {
       db.people.transaction { implicit transaction =>
-        val people = db.people.query.filter(_.nicknames has "Grouchy").toList
-        people.map(_.name) should be(List("Oscar"))
+        db.people.query.filter(_.nicknames has "Grouchy").toList.map { people =>
+          people.map(_.name) should be(List("Oscar"))
+        }
       }
     }
     "query with indexes" in {
       db.people.transaction { implicit transaction =>
-        val results = db.people.query.filter(_.name IN List("Allan", "Brenda", "Charlie")).search.indexes().list
-        results.map(_(_.name)).toSet should be(Set("Allan", "Brenda", "Charlie"))
-        results.map(_(_.doc).name).toSet should be(Set("Allan", "Brenda", "Charlie"))
+        db.people.query.filter(_.name IN List("Allan", "Brenda", "Charlie")).search.indexes().flatMap(_.list).map { results =>
+          results.map(_(_.name)).toSet should be(Set("Allan", "Brenda", "Charlie"))
+          results.map(_(_.doc).name).toSet should be(Set("Allan", "Brenda", "Charlie"))
+        }
       }
     }
     "query with doc and indexes" in {
       db.people.transaction { implicit transaction =>
-        val results = db.people.query.filter(_.name IN List("Allan", "Brenda", "Charlie")).search.docAndIndexes().list
-        results.map(_(_.name)).toSet should be(Set("Allan", "Brenda", "Charlie"))
-        results.map(_.doc.name).toSet should be(Set("Allan", "Brenda", "Charlie"))
+        db.people.query.filter(_.name IN List("Allan", "Brenda", "Charlie")).stream.docAndIndexes().toList.map { results =>
+          results.map(_(_.name)).toSet should be(Set("Allan", "Brenda", "Charlie"))
+          results.map(_.doc.name).toSet should be(Set("Allan", "Brenda", "Charlie"))
+        }
       }
     }
     "query with multi-value nicknames" in {
       db.people.transaction { implicit transaction =>
-        val people = db.people.query
+        db.people.query
           .filter(_.nicknames has "Nica")
           .filter(_.nicknames has "Vera")
           .toList
-        people.map(_.name) should be(List("Veronica"))
+          .map { people =>
+            people.map(_.name) should be(List("Veronica"))
+          }
       }
     }
     "query name with regex match" in {
       db.people.transaction { implicit transaction =>
-        val people = db.people.query.filter(_.name ~* "Han.+").toList
-        people.map(_.name) should be(List("Hanna"))
+        db.people.query.filter(_.name ~* "Han.+").toList.map { people =>
+          people.map(_.name) should be(List("Hanna"))
+        }
       }
     }
     "query nicknames that contain ica" in {
       db.people.transaction { implicit transaction =>
-        val people = db.people.query
+        db.people.query
           .filter(_.nicknames.contains("ica"))
           .toList
-        people.map(_.name).toSet should be(Set("Tori", "Veronica"))
+          .map { people =>
+            people.map(_.name).toSet should be(Set("Tori", "Veronica"))
+          }
       }
     }
     // TODO: Fix support in SQL
@@ -328,32 +353,43 @@ abstract class AbstractBasicSpec extends AsyncWordSpec with AsyncTaskSpec with M
     }*/
     "query nicknames with regex match" in {
       db.people.transaction { implicit transaction =>
-        val people = db.people.query
+        db.people.query
           .filter(_.nicknames ~* ".+chy")
           .toList
-        people.map(_.name) should be(List("Oscar"))
+          .map { people =>
+            people.map(_.name) should be(List("Oscar"))
+          }
       }
     }
     "materialize empty nicknames" in {
       db.people.transaction { implicit transaction =>
-        val people = db.people.query.filter(_.name === "Ian").search.materialized(p => List(p.nicknames)).list
-        people.map(m => m(_.nicknames)) should be(List(Set.empty))
+        db.people.query.filter(_.name === "Ian").stream.materialized(p => List(p.nicknames)).toList.map { people =>
+          people.map(m => m(_.nicknames)) should be(List(Set.empty))
+        }
       }
     }
     "query with single-value, multiple nicknames" in {
       db.people.transaction { implicit transaction =>
-        val people = db.people.query
+        db.people.query
           .filter(_.nicknames has "Nica")
           .toList
-        people.map(_.name).toSet should be(Set("Veronica", "Tori"))
+          .map { people =>
+            people.map(_.name).toSet should be(Set("Veronica", "Tori"))
+          }
       }
     }
     "sort by name and page through results" in {
       db.people.transaction { implicit transaction =>
         val q = db.people.query.sort(Sort.ByField(Person.name)).limit(10)
-        q.offset(0).search.docs.list.map(_.name) should be(List("Allan", "Brenda", "Charlie", "Diana", "Evan", "Fiona", "Greg", "Hanna", "Ian", "Jenna"))
-        q.offset(10).search.docs.list.map(_.name) should be(List("Kevin", "Mike", "Nancy", "Not Ruth", "Oscar", "Penny", "Quintin", "Sam", "Tori", "Uba"))
-        q.offset(20).search.docs.list.map(_.name) should be(List("Veronica", "Wyatt", "Xena", "Zoey"))
+        for {
+          l1 <- q.offset(0).search.docs.flatMap(_.list).map(_.map(_.name))
+          l2 <- q.offset(10).search.docs.flatMap(_.list).map(_.map(_.name))
+          l3 <- q.offset(20).search.docs.flatMap(_.list).map(_.map(_.name))
+        } yield {
+          l1 should be(List("Allan", "Brenda", "Charlie", "Diana", "Evan", "Fiona", "Greg", "Hanna", "Ian", "Jenna"))
+          l2 should be(List("Kevin", "Mike", "Nancy", "Not Ruth", "Oscar", "Penny", "Quintin", "Sam", "Tori", "Uba"))
+          l3 should be(List("Veronica", "Wyatt", "Xena", "Zoey"))
+        }
       }
     }
     "filter by list of friend ids" in {
@@ -362,11 +398,13 @@ abstract class AbstractBasicSpec extends AsyncWordSpec with AsyncTaskSpec with M
           .builder
           .should(_.friends has fiona._id)
         )
-        q.toList.map(_.name).toSet should be(Set("Sam"))
+        q.toList.map { list =>
+          list.map(_.name).toSet should be(Set("Same"))
+        }
       }
     }
     "do a database backup" in {
-      DatabaseBackup.archive(db, new File(s"backups/$specName.zip")) should be(49)
+      DatabaseBackup.archive(db, new File(s"backups/$specName.zip")).map(_ should be(49))
     }
     "insert a lot more names" in {
       db.people.transaction { implicit transaction =>
@@ -378,65 +416,76 @@ abstract class AbstractBasicSpec extends AsyncWordSpec with AsyncTaskSpec with M
             nicknames = Set("robot", s"sf$index")
           )
         }
-        db.people.insert(p)
+        db.people.insert(p).succeed
       }
     }
     "verify the correct number of people exist in the database" in {
       db.people.transaction { implicit transaction =>
-        db.people.count should be(CreateRecords + 24)
+        db.people.count.map(_ should be(CreateRecords + 24))
       }
     }
     "verify id count matches total count" in {
       db.people.transaction { implicit transaction =>
-        val results = db.people.query.countTotal(true).search.id
-        results.total should be(Some(CreateRecords + 24))
-        results.list.length should be(CreateRecords + 24)
+        db.people.query.countTotal(true).search.id.flatMap { results =>
+          results.list.map { list =>
+            results.total should be(Some(CreateRecords + 24))
+            list.length should be(CreateRecords + 24)
+          }
+        }
       }
     }
     "verify the correct count in query total" in {
       db.people.transaction { implicit transaction =>
-        val results = db.people.query
+        db.people.query
           .filter(_.nicknames.has("robot"))
           .sort(Sort.ByField(Person.age).descending)
           .limit(100)
           .countTotal(true)
           .search
           .docs
-        results.list.length should be(100)
-        results.total should be(Some(CreateRecords))
-        results.remaining should be(Some(CreateRecords))
+          .flatMap { results =>
+            results.list.map { list =>
+              list.length should be(100)
+              results.total should be(Some(CreateRecords))
+              results.remaining should be(Some(CreateRecords))
+            }
+          }
       }
     }
     "verify the correct count in query total with offset" in {
       db.people.transaction { implicit transaction =>
-        val results = db.people.query
+        db.people.query
           .filter(_.nicknames has "robot")
           .limit(100)
           .offset(100)
           .countTotal(true)
           .search
           .docs
-        results.list.length should be(100)
-        results.total should be(Some(CreateRecords))
-        results.remaining should be(Some(CreateRecords - 100))
+          .flatMap { results =>
+            results.list.map { list =>
+              list.length should be(100)
+              results.total should be(Some(CreateRecords))
+              results.remaining should be(Some(CreateRecords - 100))
+            }
+          }
       }
     }
     "truncate the collection" in {
       db.people.transaction { implicit transaction =>
-        db.people.truncate() should be(CreateRecords + 24)
+        db.people.truncate().map(_ should be(CreateRecords + 24))
       }
     }
     "verify the collection is empty" in {
       db.people.transaction { implicit transaction =>
-        db.people.count should be(0)
+        db.people.count.map(_ should be(0))
       }
     }
     "restore from database backup" in {
-      DatabaseRestore.archive(db, new File(s"backups/$specName.zip")) should be(49)
+      DatabaseRestore.archive(db, new File(s"backups/$specName.zip")).map(_ should be(49))
     }
     "verify the correct number of records exist" in {
       db.people.transaction { implicit transaction =>
-        db.people.count should be(24)
+        db.people.count.map(_ should be(24))
       }
     }
     /*"insert an invalid record via JSON" in {
@@ -461,11 +510,11 @@ abstract class AbstractBasicSpec extends AsyncWordSpec with AsyncTaskSpec with M
     }*/
     "truncate the collection again" in {
       db.people.transaction { implicit transaction =>
-        db.people.truncate() should be(24)
+        db.people.truncate().map(_ should be(24))
       }
     }
     "dispose the database" in {
-      db.dispose()
+      db.dispose().succeed
     }
   }
 
@@ -530,28 +579,30 @@ abstract class AbstractBasicSpec extends AsyncWordSpec with AsyncTaskSpec with M
 
     def id(age: Int): Id[AgeLinks] = Id(age.toString)
 
-    override protected def process(list: List[List[DocState[Person]]]): Unit = db.ageLinks.transaction { implicit transaction =>
-      list.groupBy(_.head.doc.age).foreach {
-        case (age, states) =>
-          val firsts = states.map(_.head)
-          val add = firsts.collect {
-            case DocState.Added(doc) => doc._id
-            case DocState.Modified(doc) => doc._id
-          }
-          val remove = firsts.collect {
-            case DocState.Removed(doc) => doc._id
-          }.toSet
-          db.ageLinks.modify(AgeLinks.id(age)) { existing =>
-            val current = existing.getOrElse(AgeLinks(age, Nil))
-            val modified = current.copy(
-              people = (current.people ::: add).filterNot(remove.contains)
-            )
-            if (modified.people.isEmpty) {
-              None
-            } else {
-              Some(modified)
+    override protected def process(list: List[List[DocState[Person]]]): Task[Unit] = db.ageLinks.transaction { implicit transaction =>
+      Task {
+        list.groupBy(_.head.doc.age).foreach {
+          case (age, states) =>
+            val firsts = states.map(_.head)
+            val add = firsts.collect {
+              case DocState.Added(doc) => doc._id
+              case DocState.Modified(doc) => doc._id
             }
-          }
+            val remove = firsts.collect {
+              case DocState.Removed(doc) => doc._id
+            }.toSet
+            db.ageLinks.modify(AgeLinks.id(age)) { existing =>
+              val current = existing.getOrElse(AgeLinks(age, Nil))
+              val modified = current.copy(
+                people = (current.people ::: add).filterNot(remove.contains)
+              )
+              if (modified.people.isEmpty) {
+                Task.pure(None)
+              } else {
+                Task.pure(Some(modified))
+              }
+            }.sync()
+        }
       }
     }
   }
@@ -563,6 +614,6 @@ abstract class AbstractBasicSpec extends AsyncWordSpec with AsyncTaskSpec with M
 
     override def alwaysRun: Boolean = false
 
-    override def upgrade(ldb: LightDB): Unit = db.startTime.set(System.currentTimeMillis())
+    override def upgrade(ldb: LightDB): Task[Unit] = db.startTime.set(System.currentTimeMillis()).unit
   }
 }
diff --git a/core/src/test/scala/spec/AbstractFacetSpec.scala b/core/src/test/scala/spec/AbstractFacetSpec.scala
index 7c8c6d64..6feea7b3 100644
--- a/core/src/test/scala/spec/AbstractFacetSpec.scala
+++ b/core/src/test/scala/spec/AbstractFacetSpec.scala
@@ -9,11 +9,12 @@ import lightdb.filter._
 import lightdb.store.StoreManager
 import lightdb.upgrade.DatabaseUpgrade
 import org.scalatest.matchers.should.Matchers
-import org.scalatest.wordspec.AnyWordSpec
+import org.scalatest.wordspec.{AnyWordSpec, AsyncWordSpec}
+import rapid.AsyncTaskSpec
 
 import java.nio.file.Path
 
-abstract class AbstractFacetSpec extends AnyWordSpec with Matchers { spec =>
+abstract class AbstractFacetSpec extends AsyncWordSpec with AsyncTaskSpec with Matchers { spec =>
   protected lazy val specName: String = getClass.getSimpleName
 
   protected var db: DB = new DB
@@ -28,235 +29,262 @@ abstract class AbstractFacetSpec extends AnyWordSpec with Matchers { spec =>
 
   specName should {
     "initialize the database" in {
-      db.init() should be(true)
+      db.init().succeed
     }
     "verify the database is empty" in {
       db.entries.transaction { implicit transaction =>
-        db.entries.count should be(0)
+        db.entries.count.map(_ should be(0))
       }
     }
     "insert the records" in {
       db.entries.transaction { implicit transaction =>
-        db.entries.insert(List(one, two, three, four, five, six, seven)) should not be None
+        db.entries.insert(List(one, two, three, four, five, six, seven)).map(_ should not be None)
       }
     }
     "list author facets" in {
       db.entries.transaction { implicit transaction =>
-        val results = db.entries.query
+        db.entries.query
           .facet(_.authorsFacet)
           .search
           .docs
-        val authorsResult = results.facet(_.authorsFacet)
-        authorsResult.childCount should be(6)
-        authorsResult.totalCount should be(8)
-        authorsResult.values.map(_.value) should be(List("Bob", "Lisa", "James", "Susan", "Frank", "George"))
-        authorsResult.values.map(_.count) should be(List(2, 2, 1, 1, 1, 1))
+          .map { results =>
+            val authorsResult = results.facet(_.authorsFacet)
+            authorsResult.childCount should be(6)
+            authorsResult.totalCount should be(8)
+            authorsResult.values.map(_.value) should be(List("Bob", "Lisa", "James", "Susan", "Frank", "George"))
+            authorsResult.values.map(_.count) should be(List(2, 2, 1, 1, 1, 1))
+          }
       }
     }
     "list all publishDate facets" in {
       db.entries.transaction { implicit transaction =>
-        val results = db.entries.query
+        db.entries.query
           .facet(_.publishDateFacet)
           .search
           .docs
-        val publishDateResult = results.facet(_.publishDateFacet)
-        publishDateResult.values.map(_.value) should be(List("2010", "2012", "1999"))
-        publishDateResult.childCount should be(4)
-        publishDateResult.values.map(_.count) should be(List(2, 2, 2))
-        publishDateResult.totalCount should be(6)
+          .map { results =>
+            val publishDateResult = results.facet(_.publishDateFacet)
+            publishDateResult.values.map(_.value) should be(List("2010", "2012", "1999"))
+            publishDateResult.childCount should be(4)
+            publishDateResult.values.map(_.count) should be(List(2, 2, 2))
+            publishDateResult.totalCount should be(6)
+          }
       }
     }
     "list all support@one.com keyword facets" in {
       db.entries.transaction { implicit transaction =>
-        val results = db.entries.query
+        db.entries.query
           .filter(_.keywords has "support@one.com")
           .facet(_.keywordsFacet)
           .search
           .docs
-        val keywordsResult = results.facet(_.keywordsFacet)
-        keywordsResult.childCount should be(2)
-        keywordsResult.totalCount should be(3)
-        keywordsResult.values.map(_.value) should be(List("support@one.com", "support@two.com"))
-        keywordsResult.values.map(_.count) should be(List(2, 1))
+          .map { results =>
+            val keywordsResult = results.facet(_.keywordsFacet)
+            keywordsResult.childCount should be(2)
+            keywordsResult.totalCount should be(3)
+            keywordsResult.values.map(_.value) should be(List("support@one.com", "support@two.com"))
+            keywordsResult.values.map(_.count) should be(List(2, 1))
+          }
       }
     }
     "modify a record" in {
       db.entries.transaction { implicit transaction =>
-        db.entries.upsert(five.copy(name = "Cinco"))
+        db.entries.upsert(five.copy(name = "Cinco")).succeed
       }
     }
     "list all results for 2010" in {
       db.entries.transaction { implicit transaction =>
-        val results = db.entries.query
+        db.entries.query
           .filter(_.publishDateFacet.drillDown("2010"))
           .facet(_.authorsFacet)
           .facet(_.publishDateFacet, path = List("2010"))
           .search
           .docs
-        val authorResult = results.facet(_.authorsFacet)
-        authorResult.childCount should be(3)
-        authorResult.totalCount should be(3)
-        authorResult.values.map(_.value) should be(List("Bob", "James", "Lisa"))
-        authorResult.values.map(_.count) should be(List(1, 1, 1))
-        val publishResult = results.facet(_.publishDateFacet)
-        publishResult.childCount should be(1)
-        publishResult.totalCount should be(2)
-        publishResult.values.map(_.value) should be(List("10"))
-        publishResult.values.map(_.count) should be(List(2))
+          .map { results =>
+            val authorResult = results.facet(_.authorsFacet)
+            authorResult.childCount should be(3)
+            authorResult.totalCount should be(3)
+            authorResult.values.map(_.value) should be(List("Bob", "James", "Lisa"))
+            authorResult.values.map(_.count) should be(List(1, 1, 1))
+            val publishResult = results.facet(_.publishDateFacet)
+            publishResult.childCount should be(1)
+            publishResult.totalCount should be(2)
+            publishResult.values.map(_.value) should be(List("10"))
+            publishResult.values.map(_.count) should be(List(2))
+          }
       }
     }
     "exclude all results for 2010" in {
       db.entries.transaction { implicit transaction =>
-        val results = db.entries.query
+        db.entries.query
           .facet(_.authorsFacet)
           .facet(_.publishDateFacet)
           .filter(_.builder.mustNot(_.publishDateFacet.drillDown("2010")))
           .search
           .docs
-        val authorResult = results.facet(_.authorsFacet)
-        authorResult.childCount should be(5)
-        authorResult.totalCount should be(5)
-        authorResult.values.map(_.value) should be(List("Bob", "Lisa", "Susan", "Frank", "George"))
-        authorResult.values.map(_.count) should be(List(1, 1, 1, 1, 1))
-        val publishResult = results.facet(_.publishDateFacet)
-        publishResult.childCount should be(3)
-        publishResult.totalCount should be(4)
-        publishResult.values.map(_.value) should be(List("2012", "1999"))
-        publishResult.values.map(_.count) should be(List(2, 2))
+          .map { results =>
+            val authorResult = results.facet(_.authorsFacet)
+            authorResult.childCount should be(5)
+            authorResult.totalCount should be(5)
+            authorResult.values.map(_.value) should be(List("Bob", "Lisa", "Susan", "Frank", "George"))
+            authorResult.values.map(_.count) should be(List(1, 1, 1, 1, 1))
+            val publishResult = results.facet(_.publishDateFacet)
+            publishResult.childCount should be(3)
+            publishResult.totalCount should be(4)
+            publishResult.values.map(_.value) should be(List("2012", "1999"))
+            publishResult.values.map(_.count) should be(List(2, 2))
+          }
       }
     }
     "list all results for 2010/10" in {
       db.entries.transaction { implicit transaction =>
-        val results = db.entries.query
+        db.entries.query
           .facet(_.authorsFacet)
           .facet(_.publishDateFacet, path = List("2010", "10"))
           .filter(_.publishDateFacet.drillDown("2010", "10"))
           .search
           .docs
-        val authorResult = results.facet(_.authorsFacet)
-        authorResult.childCount should be(3)
-        authorResult.totalCount should be(3)
-        authorResult.values.map(_.value) should be(List("Bob", "James", "Lisa"))
-        authorResult.values.map(_.count) should be(List(1, 1, 1))
-        val publishResult = results.facet(_.publishDateFacet)
-        publishResult.childCount should be(2)
-        publishResult.totalCount should be(2)
-        publishResult.values.map(_.value) should be(List("15", "20"))
-        publishResult.values.map(_.count) should be(List(1, 1))
+          .map { results =>
+            val authorResult = results.facet(_.authorsFacet)
+            authorResult.childCount should be(3)
+            authorResult.totalCount should be(3)
+            authorResult.values.map(_.value) should be(List("Bob", "James", "Lisa"))
+            authorResult.values.map(_.count) should be(List(1, 1, 1))
+            val publishResult = results.facet(_.publishDateFacet)
+            publishResult.childCount should be(2)
+            publishResult.totalCount should be(2)
+            publishResult.values.map(_.value) should be(List("15", "20"))
+            publishResult.values.map(_.count) should be(List(1, 1))
+          }
       }
     }
     "list all results for 2010/10/20" in {
       db.entries.transaction { implicit transaction =>
-        val results = db.entries.query
+        db.entries.query
           .facet(_.authorsFacet)
           .facet(_.publishDateFacet, path = List("2010", "10", "20"))
           .filter(_.publishDateFacet.drillDown("2010", "10", "20"))
           .search
           .docs
-        val authorResult = results.facet(_.authorsFacet)
-        authorResult.childCount should be(1)
-        authorResult.totalCount should be(1)
-        authorResult.values.map(_.value) should be(List("Lisa"))
-        authorResult.values.map(_.count) should be(List(1))
-        val publishResult = results.facet(_.publishDateFacet)
-        publishResult.childCount should be(1)
-        publishResult.totalCount should be(0)
-        publishResult.values should be(Nil)
+          .map { results =>
+            val authorResult = results.facet(_.authorsFacet)
+            authorResult.childCount should be(1)
+            authorResult.totalCount should be(1)
+            authorResult.values.map(_.value) should be(List("Lisa"))
+            authorResult.values.map(_.count) should be(List(1))
+            val publishResult = results.facet(_.publishDateFacet)
+            publishResult.childCount should be(1)
+            publishResult.totalCount should be(0)
+            publishResult.values should be(Nil)
+          }
       }
     }
     "show only results for 1999" in {
       db.entries.transaction { implicit transaction =>
-        val results = db.entries.query
+        db.entries.query
           .facet(_.authorsFacet)
           .facet(_.publishDateFacet, path = List("1999"))
           .filter(_.publishDateFacet.drillDown("1999").onlyThisLevel)
           .search
           .docs
-        val authorResult = results.facet(_.authorsFacet)
-        authorResult.childCount should be(1)
-        authorResult.totalCount should be(1)
-        authorResult.values.map(_.value) should be(List("George"))
-        authorResult.values.map(_.count) should be(List(1))
-        val publishResult = results.facet(_.publishDateFacet)
-        publishResult.childCount should be(1)
-        publishResult.totalCount should be(0)
-        publishResult.values should be(Nil)
+          .map { results =>
+            val authorResult = results.facet(_.authorsFacet)
+            authorResult.childCount should be(1)
+            authorResult.totalCount should be(1)
+            authorResult.values.map(_.value) should be(List("George"))
+            authorResult.values.map(_.count) should be(List(1))
+            val publishResult = results.facet(_.publishDateFacet)
+            publishResult.childCount should be(1)
+            publishResult.totalCount should be(0)
+            publishResult.values should be(Nil)
+          }
       }
     }
     "show all results for support@two.com" in {
       db.entries.transaction { implicit transaction =>
-        val results = db.entries.query
+        db.entries.query
           .filter(_.keywordsFacet.drillDown("support@two.com"))
           .search
           .docs
-        results.list.map(_.name).toSet should be(Set("One", "Three"))
+          .flatMap { results =>
+            results.list.map(_.map(_.name).toSet should be(Set("One", "Three")))
+          }
       }
     }
     "show all results for support@three.com or support" in {
       db.entries.transaction { implicit transaction =>
-        val results = db.entries.query
+        db.entries.query
           .filter(_.builder
             .should(_.keywordsFacet.drillDown("support@three.com"))
             .should(_.keywordsFacet.drillDown("support"))
           )
-          .search
+          .stream
           .docs
-        results.list.map(_.name).toSet should be(Set("Four", "Cinco"))
+          .toList
+          .map { list =>
+            list.map(_.name).toSet should be(Set("Four", "Cinco"))
+          }
       }
     }
     "remove a keyword from One" in {
       db.entries.transaction { implicit transaction =>
-        db.entries.upsert(one.copy(keywords = List("support@one.com")))
+        db.entries.upsert(one.copy(keywords = List("support@one.com"))).succeed
       }
     }
     "show all results for support@two.com excluding updated" in {
       db.entries.transaction { implicit transaction =>
-        val results = db.entries.query
+        db.entries.query
           .filter(_.keywordsFacet.drillDown("support@two.com"))
-          .search
+          .stream
           .docs
-        results.list.map(_.name).toSet should be(Set("Three"))
+          .toList
+          .map { results =>
+            results.map(_.name).toSet should be(Set("Three"))
+          }
       }
     }
     "show only top-level results without a publish date" in {
       db.entries.transaction { implicit transaction =>
-        val results = db.entries.query
+        db.entries.query
           .facet(_.authorsFacet)
           .facet(_.publishDateFacet)
           .filter(_.publishDateFacet.drillDown().onlyThisLevel)
           .search
           .docs
-        val authorResult = results.facet(_.authorsFacet)
-        authorResult.totalCount should be(1)
-        authorResult.childCount should be(1)
-        authorResult.values.map(_.value) should be(List("Bob"))
-        authorResult.values.map(_.count) should be(List(1))
-        val publishResult = results.facet(_.publishDateFacet)
-        publishResult.childCount should be(1)
-        publishResult.totalCount should be(0)
+          .map { results =>
+            val authorResult = results.facet(_.authorsFacet)
+            authorResult.totalCount should be(1)
+            authorResult.childCount should be(1)
+            authorResult.values.map(_.value) should be(List("Bob"))
+            authorResult.values.map(_.count) should be(List(1))
+            val publishResult = results.facet(_.publishDateFacet)
+            publishResult.childCount should be(1)
+            publishResult.totalCount should be(0)
+          }
       }
     }
     "delete a facets document" in {
       db.entries.transaction { implicit transaction =>
-        db.entries.delete(four._id)
+        db.entries.delete(four._id).succeed
       }
     }
     "query all documents verifying deletion of Four" in {
       db.entries.transaction { implicit transaction =>
-        val results = db.entries.query.facet(_.authorsFacet).search.docs
-        results.getFacet(_.publishDateFacet) should be(None)
-        val authorResult = results.facet(_.authorsFacet)
-        authorResult.values.map(_.value) should be(List("Bob", "Lisa", "James", "Frank", "George"))
-        authorResult.values.map(_.count) should be(List(2, 2, 1, 1, 1))
+        db.entries.query.facet(_.authorsFacet).search.docs.map { results =>
+          results.getFacet(_.publishDateFacet) should be(None)
+          val authorResult = results.facet(_.authorsFacet)
+          authorResult.values.map(_.value) should be(List("Bob", "Lisa", "James", "Frank", "George"))
+          authorResult.values.map(_.count) should be(List(2, 2, 1, 1, 1))
+        }
       }
     }
     "truncate the collection" in {
       db.entries.transaction { implicit transaction =>
-        db.entries.truncate() should be(6)
+        db.entries.truncate().map(_ should be(6))
       }
     }
     "dispose the database" in {
-      db.dispose()
+      db.dispose().succeed
     }
   }
 
diff --git a/core/src/test/scala/spec/AbstractKeyValueSpec.scala b/core/src/test/scala/spec/AbstractKeyValueSpec.scala
index 3c8e496d..b10acd39 100644
--- a/core/src/test/scala/spec/AbstractKeyValueSpec.scala
+++ b/core/src/test/scala/spec/AbstractKeyValueSpec.scala
@@ -53,7 +53,7 @@ abstract class AbstractKeyValueSpec extends AsyncWordSpec with AsyncTaskSpec wit
 
   specName should {
     "initialize the database" in {
-      db.init().map(_ should be(true))
+      db.init().succeed
     }
     "verify the database is empty" in {
       db.users.transaction { implicit transaction =>
diff --git a/core/src/test/scala/spec/AbstractSpatialSpec.scala b/core/src/test/scala/spec/AbstractSpatialSpec.scala
index 02881a37..9cb4149e 100644
--- a/core/src/test/scala/spec/AbstractSpatialSpec.scala
+++ b/core/src/test/scala/spec/AbstractSpatialSpec.scala
@@ -11,11 +11,12 @@ import lightdb.store.StoreManager
 import lightdb.upgrade.DatabaseUpgrade
 import lightdb.{Id, LightDB}
 import org.scalatest.matchers.should.Matchers
-import org.scalatest.wordspec.AnyWordSpec
+import org.scalatest.wordspec.{AnyWordSpec, AsyncWordSpec}
+import rapid.AsyncTaskSpec
 
 import java.nio.file.Path
 
-abstract class AbstractSpatialSpec extends AnyWordSpec with Matchers { spec =>
+abstract class AbstractSpatialSpec extends AsyncWordSpec with AsyncTaskSpec with Matchers { spec =>
   private lazy val specName: String = getClass.getSimpleName
 
   private val id1 = Id[Person]("john")
@@ -66,50 +67,52 @@ abstract class AbstractSpatialSpec extends AnyWordSpec with Matchers { spec =>
 
   specName should {
     "initialize the database" in {
-      DB.init() should be(true)
+      DB.init().succeed
     }
     "store three people" in {
       DB.people.transaction { implicit transaction =>
-        DB.people.insert(List(p1, p2, p3)).length should be(3)
+        DB.people.insert(List(p1, p2, p3)).map(_.length should be(3))
       }
     }
     "verify exactly three people exist" in {
       DB.people.transaction { implicit transaction =>
-        DB.people.count should be(3)
+        DB.people.count.map(_ should be(3))
       }
     }
     "sort by distance from Oklahoma City" in {
       DB.people.transaction { implicit transaction =>
-        val list = DB.people.query.search.distance(
+        DB.people.query.stream.distance(
           _.point.list,
           from = oklahomaCity,
           radius = Some(1320.miles)
-        ).iterator.toList
-        val people = list.map(_.doc)
-        val distances = list.map(_.distance.map(_.mi.toInt))
-        people.zip(distances).map {
-          case (p, d) => p.name -> d
-        } should be(List(
-          "Jane Doe" -> List(28),
-          "John Doe" -> List(1316)
-        ))
+        ).toList.map { list =>
+          val people = list.map(_.doc)
+          val distances = list.map(_.distance.map(_.mi.toInt))
+          people.zip(distances).map {
+            case (p, d) => p.name -> d
+          } should be(List(
+            "Jane Doe" -> List(28),
+            "John Doe" -> List(1316)
+          ))
+        }
       }
     }
     "sort by distance from Noble using geo" in {
       DB.people.transaction { implicit transaction =>
-        val list = DB.people.query.search.distance(
+        DB.people.query.stream.distance(
           _.geo,
           from = noble,
           radius = Some(10_000.miles)
-        ).iterator.toList
-        val people = list.map(_.doc)
-        val distances = list.map(_.distance.map(_.mi))
-        people.zip(distances).map {
-          case (p, d) => p.name -> d
-        } should be(List(
-          "Jane Doe" -> List(16.01508397712445),
-          "Bob Dole" -> List(695.6419047674393, 1334.038796028706)
-        ))
+        ).toList.map { list =>
+          val people = list.map(_.doc)
+          val distances = list.map(_.distance.map(_.mi))
+          people.zip(distances).map {
+            case (p, d) => p.name -> d
+          } should be(List(
+            "Jane Doe" -> List(16.01508397712445),
+            "Bob Dole" -> List(695.6419047674393, 1334.038796028706)
+          ))
+        }
       }
     }
     "parse and insert from a GeometryCollection" in {
@@ -121,14 +124,14 @@ abstract class AbstractSpatialSpec extends AnyWordSpec with Matchers { spec =>
           age = 2,
           point = yonkers,
           geo = geo
-        ))
+        )).succeed
       }
     }
     "truncate the database" in {
-      DB.truncate()
+      DB.truncate().succeed
     }
     "dispose the database" in {
-      DB.dispose()
+      DB.dispose().succeed
     }
   }
 
diff --git a/core/src/test/scala/spec/AbstractSpecialCasesSpec.scala b/core/src/test/scala/spec/AbstractSpecialCasesSpec.scala
index 183cb9ae..49bfc0f1 100644
--- a/core/src/test/scala/spec/AbstractSpecialCasesSpec.scala
+++ b/core/src/test/scala/spec/AbstractSpecialCasesSpec.scala
@@ -8,55 +8,61 @@ import lightdb.store.StoreManager
 import lightdb.upgrade.DatabaseUpgrade
 import lightdb.{Id, LightDB, Sort, Timestamp}
 import org.scalatest.matchers.should.Matchers
-import org.scalatest.wordspec.AnyWordSpec
+import org.scalatest.wordspec.{AnyWordSpec, AsyncWordSpec}
+import rapid.AsyncTaskSpec
 
 import java.nio.file.Path
 
-trait AbstractSpecialCasesSpec extends AnyWordSpec with Matchers { spec =>
+trait AbstractSpecialCasesSpec extends AsyncWordSpec with AsyncTaskSpec with Matchers { spec =>
   private lazy val specName: String = getClass.getSimpleName
 
   specName should {
     "initialize the database" in {
-      DB.init()
+      DB.init().succeed
     }
     "insert a couple SpecialOne instances" in {
       DB.specialOne.t.insert(List(
         SpecialOne("First", WrappedString("Apple"), Person("Andrew", 1), _id = Id("first")),
         SpecialOne("Second", WrappedString("Banana"), Person("Bianca", 2), _id = Id("second"))
-      ))
+      )).succeed
     }
     "verify the SpecialOne instances were stored properly" in {
       DB.specialOne.transaction { implicit transaction =>
-        val list = DB.specialOne.iterator.toList
-        list.map(_.name).toSet should be(Set("First", "Second"))
-        list.map(_.wrappedString).toSet should be(Set(WrappedString("Apple"), WrappedString("Banana")))
-        list.map(_.person).toSet should be(Set(Person("Andrew", 1), Person("Bianca", 2)))
-        list.map(_._id).toSet should be(Set(SpecialOne.id("first"), SpecialOne.id("second")))
+        DB.specialOne.stream.toList.map { list =>
+          list.map(_.name).toSet should be(Set("First", "Second"))
+          list.map(_.wrappedString).toSet should be(Set(WrappedString("Apple"), WrappedString("Banana")))
+          list.map(_.person).toSet should be(Set(Person("Andrew", 1), Person("Bianca", 2)))
+          list.map(_._id).toSet should be(Set(SpecialOne.id("first"), SpecialOne.id("second")))
+        }
       }
     }
     "verify filtering by created works" in {
       DB.specialOne.transaction { implicit transaction =>
-        DB.specialOne.query.filter(_.created < Timestamp()).toList.map(_.name).toSet should be(Set("First", "Second"))
-        DB.specialOne.query.filter(_.created > Timestamp()).toList.map(_.name).toSet should be(Set.empty)
+        for {
+          _ <- DB.specialOne.query.filter(_.created < Timestamp()).toList.map(_.map(_.name).toSet should be(Set("First", "Second")))
+          _ <- DB.specialOne.query.filter(_.created > Timestamp()).toList.map(_.map(_.name).toSet should be(Set.empty))
+        } yield succeed
       }
     }
     "verify the storage of data is correct" in {
       DB.specialOne.transaction { implicit transaction =>
-        val list = DB.specialOne.query.sort(Sort.ByField(SpecialOne.name).asc).search.json(ref => List(ref._id)).list
-        list should be(List(obj("_id" -> "first"), obj("_id" -> "second")))
+        DB.specialOne.query.sort(Sort.ByField(SpecialOne.name).asc).stream.json(ref => List(ref._id)).toList.map { list =>
+          list should be(List(obj("_id" -> "first"), obj("_id" -> "second")))
+        }
       }
     }
     "group ids" in {
       DB.specialOne.transaction { implicit transaction =>
-        val list = DB.specialOne.query.aggregate(ref => List(ref._id.concat)).toList
-        list.map(_(_._id.concat)) should be(List(List(SpecialOne.id("first"), SpecialOne.id("second"))))
+        DB.specialOne.query.aggregate(ref => List(ref._id.concat)).toList.map { list =>
+          list.map(_(_._id.concat)) should be(List(List(SpecialOne.id("first"), SpecialOne.id("second"))))
+        }
       }
     }
     "truncate the database" in {
-      DB.truncate()
+      DB.truncate().succeed
     }
     "dispose the database" in {
-      DB.dispose()
+      DB.dispose().succeed
     }
   }