fix query cancellation with cancelable

typelevel · Aug 14, 2024 · 90c645c · 90c645c
1 parent d1f84e6
commit 90c645c
Show file tree

Hide file tree

Showing 29 changed files with 326 additions and 9 deletions.
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -56,7 +56,7 @@ jobs:
         run: sbt +update
 
       - name: Start up Postgres
-        run: docker-compose up -d
+        run: docker compose up -d
 
       - name: Check Headers
         run: sbt '++ ${{ matrix.scala }}' headerCheckAll

diff --git a/build.sbt b/build.sbt
@@ -43,7 +43,7 @@ ThisBuild / tlSonatypeUseLegacyHost := false
 ThisBuild / githubWorkflowJavaVersions := Seq(JavaSpec.temurin("11"))
 ThisBuild / githubWorkflowBuildPreamble ++= Seq(
   WorkflowStep.Run(
-    commands = List("docker-compose up -d"),
+    commands = List("docker compose up -d"),
     name = Some("Start up Postgres")
   ),
   WorkflowStep.Sbt(
@@ -426,6 +426,7 @@ lazy val hikari = project
     libraryDependencies ++= Seq(
       // needs to be excluded, otherwise coursier may resolve slf4j-api 2 if > Java 11
       "com.zaxxer" % "HikariCP" % hikariVersion exclude ("org.slf4j", "slf4j-api"),
+      "org.postgresql" % "postgresql" % postgresVersion % "test",
       "com.h2database" % "h2" % h2Version % "test",
       "org.slf4j" % "slf4j-api" % slf4jVersion,
       "org.slf4j" % "slf4j-nop" % slf4jVersion % "test"

diff --git a/modules/core/src/main/scala/doobie/hi/connection.scala b/modules/core/src/main/scala/doobie/hi/connection.scala
@@ -6,7 +6,8 @@ package doobie.hi
 
 import cats.Foldable
 import cats.data.Ior
-import cats.effect.Sync
+import cats.effect.{Resource, Sync}
+import cats.effect.kernel.Outcome
 import cats.effect.syntax.monadCancel.*
 import cats.syntax.all.*
 import doobie.enumerated.AutoGeneratedKeys
@@ -32,7 +33,6 @@ import doobie.free.{
   resultset as IFRS,
   statement as IFS
 }
-
 import scala.Predef.*
 import scala.concurrent.duration.{Duration, FiniteDuration, NANOSECONDS}
 import java.sql.{PreparedStatement, ResultSet, Savepoint}
@@ -173,9 +173,25 @@ object connection {
     val prepLogged = prep.onError { case e =>
       IFPS.performLogging(LogEvent.execFailure(loggingInfo, Duration.Zero, e))
     }
-
     createLogged
-      .bracket(ps => IFC.embed(ps, prepLogged *> execAndProcessLogged))(ps => IFC.embed(ps, IFPS.close))
+      .bracket(
+        ps => WeakAsyncConnectionIO.cancelable(
+          IFC.embed(ps, prepLogged *> execAndProcessLogged),
+          IFC.embed(ps, IFPS.close)
+        )
+      )(ps => IFC.embed(ps, IFPS.close))
+
+//    WeakAsyncConnectionIO.uncancelable { unmask =>
+//      createLogged
+//        .flatMap(
+//          ps => unmask {
+//            WeakAsyncConnectionIO.cancelable(
+//              IFC.embed(ps, prepLogged *> execAndProcessLogged),
+//              IFC.embed(ps, IFPS.close)
+//            )
+//          }.guarantee(IFC.embed(ps, IFPS.close))
+//        )
+//    }
   }
 
   /** Execute a PreparedStatement query and provide rows from the ResultSet in chunks
@@ -213,11 +229,24 @@ object connection {
 
     for {
       ps <- Stream.bracket(runPreExecWithLogging(create, loggingInfo))(IFC.embed(_, IFPS.close))
-      _ <- Stream.eval(runPreExecWithLogging(IFC.embed(ps, IFPS.setFetchSize(chunkSize) *> prep), loggingInfo))
+      _ <- Stream.eval(
+        WeakAsyncConnectionIO.cancelable(
+          runPreExecWithLogging(IFC.embed(ps, IFPS.setFetchSize(chunkSize) *> prep), loggingInfo),
+          IFC.embed(ps, IFPS.close)
+        )
+      )
       resultSet <- Stream.bracket(
-        IFC.embed(ps, execLogged)
+        WeakAsyncConnectionIO.cancelable(
+          IFC.embed(ps, execLogged),
+          IFC.embed(ps, IFPS.close)
+        )
       )(rs => IFC.embed(rs, IFRS.close))
-      ele <- repeatEvalChunks(IFC.embed(resultSet, resultset.getNextChunk[A](chunkSize)))
+      ele <- repeatEvalChunks(
+        WeakAsyncConnectionIO.cancelable(
+          IFC.embed(resultSet, resultset.getNextChunk[A](chunkSize)),
+          IFC.embed(ps, IFPS.close)
+        )
+      )
     } yield ele
   }
 

diff --git a/modules/core/src/test/scala/doobie/util/QueryCancellationSuite.scala b/modules/core/src/test/scala/doobie/util/QueryCancellationSuite.scala
@@ -0,0 +1,51 @@
+// Copyright (c) 2013-2020 Rob Norris and Contributors
+// This software is licensed under the MIT License (MIT).
+// For more information see LICENSE or https://opensource.org/licenses/MIT
+
+package doobie.util
+
+import cats.effect.IO
+import doobie.Transactor
+import doobie.*
+import doobie.implicits.*
+import cats.syntax.all.*
+
+import scala.concurrent.duration.DurationInt
+
+class QueryCancellationSuite extends munit.FunSuite {
+  import cats.effect.unsafe.implicits.global
+
+  val xa = Transactor.fromDriverManager[IO](
+    driver = "org.h2.Driver",
+    url = "jdbc:h2:mem:queryspec;DB_CLOSE_DELAY=-1",
+    user = "sa",
+    password = "",
+    logHandler = None
+  )
+
+  test("Query cancel") {
+    val scenario = WeakAsync.liftIO[ConnectionIO].use { elevator =>
+      for {
+        _ <- sql"CREATE TABLE IF NOT EXISTS example_table ( id INT)".update.run.transact(xa)
+        _ <- sql"TRUNCATE TABLE example_table".update.run.transact(xa)
+        _ <- sql"INSERT INTO example_table (id) VALUES (1)".update.run.transact(xa)
+        _ <- {
+          sql"select * from example_table for update".query[Int].unique >> elevator.liftIO(IO.never)
+        }.transact(xa).start
+
+        insertWithLockFiber <- {
+          for {
+            _ <- IO.sleep(100.milli)
+            insertFiber <- sql"UPDATE example_table SET id = 2".update.run.transact(xa).start
+            _ <- IO.sleep(100.milli)
+            _ <- insertFiber.cancel
+          } yield ()
+        }.start
+
+        _ <- IO.race(insertWithLockFiber.join, IO.sleep(5.seconds) >> IO(fail("Cancellation is blocked")))
+        result <- sql"SELECT * FROM example_table".query[Int].to[List].transact(xa)
+      } yield assertEquals(result, List(1))
+    }
+    scenario.unsafeRunSync()
+  }
+}
diff --git a/modules/free/src/main/scala/doobie/WeakAsync.scala b/modules/free/src/main/scala/doobie/WeakAsync.scala
@@ -15,6 +15,7 @@ import scala.concurrent.duration.FiniteDuration
 trait WeakAsync[F[_]] extends Sync[F] {
   def fromFuture[A](fut: F[Future[A]]): F[A]
   def fromFutureCancelable[A](fut: F[(Future[A], F[Unit])]): F[A]
+  def cancelable[A](fa: F[A], fin: F[Unit]): F[A]
 }
 
 object WeakAsync {
@@ -39,6 +40,8 @@ object WeakAsync {
       override def onCancel[A](fa: F[A], fin: F[Unit]): F[A] = F.onCancel(fa, fin)
       override def fromFuture[A](fut: F[Future[A]]): F[A] = F.fromFuture(fut)
       override def fromFutureCancelable[A](fut: F[(Future[A], F[Unit])]): F[A] = F.fromFutureCancelable(fut)
+
+      override def cancelable[A](fa: F[A], fin: F[Unit]): F[A] = F.cancelable(fa, fin)
     }
 
   /** Create a natural transformation for lifting an `Async` effect `F` into a `WeakAsync` effect `G`

diff --git a/modules/free/src/main/scala/doobie/free/blob.scala b/modules/free/src/main/scala/doobie/free/blob.scala
@@ -59,6 +59,7 @@ object blob { module =>
       def onCancel[A](fa: BlobIO[A], fin: BlobIO[Unit]): F[A]
       def fromFuture[A](fut: BlobIO[Future[A]]): F[A]
       def fromFutureCancelable[A](fut: BlobIO[(Future[A], BlobIO[Unit])]): F[A]
+      def cancelable[A](fa: BlobIO[A], fin: BlobIO[Unit]): F[A]
       def performLogging(event: LogEvent): F[Unit]
 
       // Blob
@@ -119,6 +120,9 @@ object blob { module =>
     case class FromFutureCancelable[A](fut: BlobIO[(Future[A], BlobIO[Unit])]) extends BlobOp[A] {
       def visit[F[_]](v: Visitor[F]) = v.fromFutureCancelable(fut)
     }
+    case class Cancelable[A](fa: BlobIO[A], fin: BlobIO[Unit]) extends BlobOp[A] {
+      def visit[F[_]](v: Visitor[F]) = v.cancelable(fa, fin)
+    }
     case class PerformLogging(event: LogEvent) extends BlobOp[Unit] {
       def visit[F[_]](v: Visitor[F]) = v.performLogging(event)
     }
@@ -181,6 +185,7 @@ object blob { module =>
   def onCancel[A](fa: BlobIO[A], fin: BlobIO[Unit]) = FF.liftF[BlobOp, A](OnCancel(fa, fin))
   def fromFuture[A](fut: BlobIO[Future[A]]) = FF.liftF[BlobOp, A](FromFuture(fut))
   def fromFutureCancelable[A](fut: BlobIO[(Future[A], BlobIO[Unit])]) = FF.liftF[BlobOp, A](FromFutureCancelable(fut))
+  def cancelable[A](fa: BlobIO[A], fin: BlobIO[Unit]) = FF.liftF[BlobOp, A](Cancelable(fa, fin))
   def performLogging(event: LogEvent) = FF.liftF[BlobOp, Unit](PerformLogging(event))
 
   // Smart constructors for Blob-specific operations.
@@ -216,6 +221,7 @@ object blob { module =>
       override def onCancel[A](fa: BlobIO[A], fin: BlobIO[Unit]): BlobIO[A] = module.onCancel(fa, fin)
       override def fromFuture[A](fut: BlobIO[Future[A]]): BlobIO[A] = module.fromFuture(fut)
       override def fromFutureCancelable[A](fut: BlobIO[(Future[A], BlobIO[Unit])]): BlobIO[A] = module.fromFutureCancelable(fut)
+      override def cancelable[A](fa: BlobIO[A], fin: BlobIO[Unit]): BlobIO[A] = module.cancelable(fa, fin)
     }
 
   implicit def MonoidBlobIO[A : Monoid]: Monoid[BlobIO[A]] = new Monoid[BlobIO[A]] {

diff --git a/modules/free/src/main/scala/doobie/free/callablestatement.scala b/modules/free/src/main/scala/doobie/free/callablestatement.scala
@@ -80,6 +80,7 @@ object callablestatement { module =>
       def onCancel[A](fa: CallableStatementIO[A], fin: CallableStatementIO[Unit]): F[A]
       def fromFuture[A](fut: CallableStatementIO[Future[A]]): F[A]
       def fromFutureCancelable[A](fut: CallableStatementIO[(Future[A], CallableStatementIO[Unit])]): F[A]
+      def cancelable[A](fa: CallableStatementIO[A], fin: CallableStatementIO[Unit]): F[A]
       def performLogging(event: LogEvent): F[Unit]
 
       // CallableStatement
@@ -362,6 +363,9 @@ object callablestatement { module =>
     case class FromFutureCancelable[A](fut: CallableStatementIO[(Future[A], CallableStatementIO[Unit])]) extends CallableStatementOp[A] {
       def visit[F[_]](v: Visitor[F]) = v.fromFutureCancelable(fut)
     }
+    case class Cancelable[A](fa: CallableStatementIO[A], fin: CallableStatementIO[Unit]) extends CallableStatementOp[A] {
+      def visit[F[_]](v: Visitor[F]) = v.cancelable(fa, fin)
+    }
     case class PerformLogging(event: LogEvent) extends CallableStatementOp[Unit] {
       def visit[F[_]](v: Visitor[F]) = v.performLogging(event)
     }
@@ -1090,6 +1094,7 @@ object callablestatement { module =>
   def onCancel[A](fa: CallableStatementIO[A], fin: CallableStatementIO[Unit]) = FF.liftF[CallableStatementOp, A](OnCancel(fa, fin))
   def fromFuture[A](fut: CallableStatementIO[Future[A]]) = FF.liftF[CallableStatementOp, A](FromFuture(fut))
   def fromFutureCancelable[A](fut: CallableStatementIO[(Future[A], CallableStatementIO[Unit])]) = FF.liftF[CallableStatementOp, A](FromFutureCancelable(fut))
+  def cancelable[A](fa: CallableStatementIO[A], fin: CallableStatementIO[Unit]) = FF.liftF[CallableStatementOp, A](Cancelable(fa, fin))
   def performLogging(event: LogEvent) = FF.liftF[CallableStatementOp, Unit](PerformLogging(event))
 
   // Smart constructors for CallableStatement-specific operations.
@@ -1347,6 +1352,7 @@ object callablestatement { module =>
       override def onCancel[A](fa: CallableStatementIO[A], fin: CallableStatementIO[Unit]): CallableStatementIO[A] = module.onCancel(fa, fin)
       override def fromFuture[A](fut: CallableStatementIO[Future[A]]): CallableStatementIO[A] = module.fromFuture(fut)
       override def fromFutureCancelable[A](fut: CallableStatementIO[(Future[A], CallableStatementIO[Unit])]): CallableStatementIO[A] = module.fromFutureCancelable(fut)
+      override def cancelable[A](fa: CallableStatementIO[A], fin: CallableStatementIO[Unit]): CallableStatementIO[A] = module.cancelable(fa, fin)
     }
 
   implicit def MonoidCallableStatementIO[A : Monoid]: Monoid[CallableStatementIO[A]] = new Monoid[CallableStatementIO[A]] {

diff --git a/modules/free/src/main/scala/doobie/free/clob.scala b/modules/free/src/main/scala/doobie/free/clob.scala
@@ -62,6 +62,7 @@ object clob { module =>
       def onCancel[A](fa: ClobIO[A], fin: ClobIO[Unit]): F[A]
       def fromFuture[A](fut: ClobIO[Future[A]]): F[A]
       def fromFutureCancelable[A](fut: ClobIO[(Future[A], ClobIO[Unit])]): F[A]
+      def cancelable[A](fa: ClobIO[A], fin: ClobIO[Unit]): F[A]
       def performLogging(event: LogEvent): F[Unit]
 
       // Clob
@@ -124,6 +125,9 @@ object clob { module =>
     case class FromFutureCancelable[A](fut: ClobIO[(Future[A], ClobIO[Unit])]) extends ClobOp[A] {
       def visit[F[_]](v: Visitor[F]) = v.fromFutureCancelable(fut)
     }
+    case class Cancelable[A](fa: ClobIO[A], fin: ClobIO[Unit]) extends ClobOp[A] {
+      def visit[F[_]](v: Visitor[F]) = v.cancelable(fa, fin)
+    }
     case class PerformLogging(event: LogEvent) extends ClobOp[Unit] {
       def visit[F[_]](v: Visitor[F]) = v.performLogging(event)
     }
@@ -192,6 +196,7 @@ object clob { module =>
   def onCancel[A](fa: ClobIO[A], fin: ClobIO[Unit]) = FF.liftF[ClobOp, A](OnCancel(fa, fin))
   def fromFuture[A](fut: ClobIO[Future[A]]) = FF.liftF[ClobOp, A](FromFuture(fut))
   def fromFutureCancelable[A](fut: ClobIO[(Future[A], ClobIO[Unit])]) = FF.liftF[ClobOp, A](FromFutureCancelable(fut))
+  def cancelable[A](fa: ClobIO[A], fin: ClobIO[Unit]) = FF.liftF[ClobOp, A](Cancelable(fa, fin))
   def performLogging(event: LogEvent) = FF.liftF[ClobOp, Unit](PerformLogging(event))
 
   // Smart constructors for Clob-specific operations.
@@ -229,6 +234,7 @@ object clob { module =>
       override def onCancel[A](fa: ClobIO[A], fin: ClobIO[Unit]): ClobIO[A] = module.onCancel(fa, fin)
       override def fromFuture[A](fut: ClobIO[Future[A]]): ClobIO[A] = module.fromFuture(fut)
       override def fromFutureCancelable[A](fut: ClobIO[(Future[A], ClobIO[Unit])]): ClobIO[A] = module.fromFutureCancelable(fut)
+      override def cancelable[A](fa: ClobIO[A], fin: ClobIO[Unit]): ClobIO[A] = module.cancelable(fa, fin)
     }
 
   implicit def MonoidClobIO[A : Monoid]: Monoid[ClobIO[A]] = new Monoid[ClobIO[A]] {

diff --git a/modules/free/src/main/scala/doobie/free/connection.scala b/modules/free/src/main/scala/doobie/free/connection.scala
@@ -74,6 +74,7 @@ object connection { module =>
       def onCancel[A](fa: ConnectionIO[A], fin: ConnectionIO[Unit]): F[A]
       def fromFuture[A](fut: ConnectionIO[Future[A]]): F[A]
       def fromFutureCancelable[A](fut: ConnectionIO[(Future[A], ConnectionIO[Unit])]): F[A]
+      def cancelable[A](fa: ConnectionIO[A], fin: ConnectionIO[Unit]): F[A]
       def performLogging(event: LogEvent): F[Unit]
 
       // Connection
@@ -183,6 +184,9 @@ object connection { module =>
     case class FromFutureCancelable[A](fut: ConnectionIO[(Future[A], ConnectionIO[Unit])]) extends ConnectionOp[A] {
       def visit[F[_]](v: Visitor[F]) = v.fromFutureCancelable(fut)
     }
+    case class Cancelable[A](fa: ConnectionIO[A], fin: ConnectionIO[Unit]) extends ConnectionOp[A] {
+      def visit[F[_]](v: Visitor[F]) = v.cancelable(fa, fin)
+    }
     case class PerformLogging(event: LogEvent) extends ConnectionOp[Unit] {
       def visit[F[_]](v: Visitor[F]) = v.performLogging(event)
     }
@@ -392,6 +396,7 @@ object connection { module =>
   def onCancel[A](fa: ConnectionIO[A], fin: ConnectionIO[Unit]) = FF.liftF[ConnectionOp, A](OnCancel(fa, fin))
   def fromFuture[A](fut: ConnectionIO[Future[A]]) = FF.liftF[ConnectionOp, A](FromFuture(fut))
   def fromFutureCancelable[A](fut: ConnectionIO[(Future[A], ConnectionIO[Unit])]) = FF.liftF[ConnectionOp, A](FromFutureCancelable(fut))
+  def cancelable[A](fa: ConnectionIO[A], fin: ConnectionIO[Unit]) = FF.liftF[ConnectionOp, A](Cancelable(fa, fin))
   def performLogging(event: LogEvent) = FF.liftF[ConnectionOp, Unit](PerformLogging(event))
 
   // Smart constructors for Connection-specific operations.
@@ -476,6 +481,7 @@ object connection { module =>
       override def onCancel[A](fa: ConnectionIO[A], fin: ConnectionIO[Unit]): ConnectionIO[A] = module.onCancel(fa, fin)
       override def fromFuture[A](fut: ConnectionIO[Future[A]]): ConnectionIO[A] = module.fromFuture(fut)
       override def fromFutureCancelable[A](fut: ConnectionIO[(Future[A], ConnectionIO[Unit])]): ConnectionIO[A] = module.fromFutureCancelable(fut)
+      override def cancelable[A](fa: ConnectionIO[A], fin: ConnectionIO[Unit]): ConnectionIO[A] = module.cancelable(fa, fin)
     }
 
   implicit def MonoidConnectionIO[A : Monoid]: Monoid[ConnectionIO[A]] = new Monoid[ConnectionIO[A]] {

diff --git a/modules/free/src/main/scala/doobie/free/databasemetadata.scala b/modules/free/src/main/scala/doobie/free/databasemetadata.scala
@@ -62,6 +62,7 @@ object databasemetadata { module =>
       def onCancel[A](fa: DatabaseMetaDataIO[A], fin: DatabaseMetaDataIO[Unit]): F[A]
       def fromFuture[A](fut: DatabaseMetaDataIO[Future[A]]): F[A]
       def fromFutureCancelable[A](fut: DatabaseMetaDataIO[(Future[A], DatabaseMetaDataIO[Unit])]): F[A]
+      def cancelable[A](fa: DatabaseMetaDataIO[A], fin: DatabaseMetaDataIO[Unit]): F[A]
       def performLogging(event: LogEvent): F[Unit]
 
       // DatabaseMetaData
@@ -290,6 +291,9 @@ object databasemetadata { module =>
     case class FromFutureCancelable[A](fut: DatabaseMetaDataIO[(Future[A], DatabaseMetaDataIO[Unit])]) extends DatabaseMetaDataOp[A] {
       def visit[F[_]](v: Visitor[F]) = v.fromFutureCancelable(fut)
     }
+    case class Cancelable[A](fa: DatabaseMetaDataIO[A], fin: DatabaseMetaDataIO[Unit]) extends DatabaseMetaDataOp[A] {
+      def visit[F[_]](v: Visitor[F]) = v.cancelable(fa, fin)
+    }
     case class PerformLogging(event: LogEvent) extends DatabaseMetaDataOp[Unit] {
       def visit[F[_]](v: Visitor[F]) = v.performLogging(event)
     }
@@ -856,6 +860,7 @@ object databasemetadata { module =>
   def onCancel[A](fa: DatabaseMetaDataIO[A], fin: DatabaseMetaDataIO[Unit]) = FF.liftF[DatabaseMetaDataOp, A](OnCancel(fa, fin))
   def fromFuture[A](fut: DatabaseMetaDataIO[Future[A]]) = FF.liftF[DatabaseMetaDataOp, A](FromFuture(fut))
   def fromFutureCancelable[A](fut: DatabaseMetaDataIO[(Future[A], DatabaseMetaDataIO[Unit])]) = FF.liftF[DatabaseMetaDataOp, A](FromFutureCancelable(fut))
+  def cancelable[A](fa: DatabaseMetaDataIO[A], fin: DatabaseMetaDataIO[Unit]) = FF.liftF[DatabaseMetaDataOp, A](Cancelable(fa, fin))
   def performLogging(event: LogEvent) = FF.liftF[DatabaseMetaDataOp, Unit](PerformLogging(event))
 
   // Smart constructors for DatabaseMetaData-specific operations.
@@ -1059,6 +1064,7 @@ object databasemetadata { module =>
       override def onCancel[A](fa: DatabaseMetaDataIO[A], fin: DatabaseMetaDataIO[Unit]): DatabaseMetaDataIO[A] = module.onCancel(fa, fin)
       override def fromFuture[A](fut: DatabaseMetaDataIO[Future[A]]): DatabaseMetaDataIO[A] = module.fromFuture(fut)
       override def fromFutureCancelable[A](fut: DatabaseMetaDataIO[(Future[A], DatabaseMetaDataIO[Unit])]): DatabaseMetaDataIO[A] = module.fromFutureCancelable(fut)
+      override def cancelable[A](fa: DatabaseMetaDataIO[A], fin: DatabaseMetaDataIO[Unit]): DatabaseMetaDataIO[A] = module.cancelable(fa, fin)
     }
 
   implicit def MonoidDatabaseMetaDataIO[A : Monoid]: Monoid[DatabaseMetaDataIO[A]] = new Monoid[DatabaseMetaDataIO[A]] {

diff --git a/modules/free/src/main/scala/doobie/free/driver.scala b/modules/free/src/main/scala/doobie/free/driver.scala
@@ -62,6 +62,7 @@ object driver { module =>
       def onCancel[A](fa: DriverIO[A], fin: DriverIO[Unit]): F[A]
       def fromFuture[A](fut: DriverIO[Future[A]]): F[A]
       def fromFutureCancelable[A](fut: DriverIO[(Future[A], DriverIO[Unit])]): F[A]
+      def cancelable[A](fa: DriverIO[A], fin: DriverIO[Unit]): F[A]
       def performLogging(event: LogEvent): F[Unit]
 
       // Driver
@@ -118,6 +119,9 @@ object driver { module =>
     case class FromFutureCancelable[A](fut: DriverIO[(Future[A], DriverIO[Unit])]) extends DriverOp[A] {
       def visit[F[_]](v: Visitor[F]) = v.fromFutureCancelable(fut)
     }
+    case class Cancelable[A](fa: DriverIO[A], fin: DriverIO[Unit]) extends DriverOp[A] {
+      def visit[F[_]](v: Visitor[F]) = v.cancelable(fa, fin)
+    }
     case class PerformLogging(event: LogEvent) extends DriverOp[Unit] {
       def visit[F[_]](v: Visitor[F]) = v.performLogging(event)
     }
@@ -168,6 +172,7 @@ object driver { module =>
   def onCancel[A](fa: DriverIO[A], fin: DriverIO[Unit]) = FF.liftF[DriverOp, A](OnCancel(fa, fin))
   def fromFuture[A](fut: DriverIO[Future[A]]) = FF.liftF[DriverOp, A](FromFuture(fut))
   def fromFutureCancelable[A](fut: DriverIO[(Future[A], DriverIO[Unit])]) = FF.liftF[DriverOp, A](FromFutureCancelable(fut))
+  def cancelable[A](fa: DriverIO[A], fin: DriverIO[Unit]) = FF.liftF[DriverOp, A](Cancelable(fa, fin))
   def performLogging(event: LogEvent) = FF.liftF[DriverOp, Unit](PerformLogging(event))
 
   // Smart constructors for Driver-specific operations.
@@ -199,6 +204,7 @@ object driver { module =>
       override def onCancel[A](fa: DriverIO[A], fin: DriverIO[Unit]): DriverIO[A] = module.onCancel(fa, fin)
       override def fromFuture[A](fut: DriverIO[Future[A]]): DriverIO[A] = module.fromFuture(fut)
       override def fromFutureCancelable[A](fut: DriverIO[(Future[A], DriverIO[Unit])]): DriverIO[A] = module.fromFutureCancelable(fut)
+      override def cancelable[A](fa: DriverIO[A], fin: DriverIO[Unit]): DriverIO[A] = module.cancelable(fa, fin)
     }
 
   implicit def MonoidDriverIO[A : Monoid]: Monoid[DriverIO[A]] = new Monoid[DriverIO[A]] {