Fix cancelable to guarantee cancelation

kernel/shared/src/main/scala/cats/effect/kernel/GenSpawn.scala

@@ -264,7 +264,9 @@ trait GenSpawn[F[_], E] extends MonadCancel[F, E] with Unique[F] {
   def cancelable[A](fa: F[A], fin: F[Unit]): F[A] =
     uncancelable { poll =>
       start(fa) flatMap { fiber =>
-        poll(fiber.join).onCancel(fin *> fiber.cancel).flatMap(_.embed(poll(canceled *> never)))
+        poll(fiber.join)
+          .onCancel(fin.guarantee(fiber.cancel))
+          .flatMap(_.embed(poll(canceled *> never)))
       }
     }
 

tests/shared/src/test/scala/cats/effect/IOSpec.scala

@@ -1145,6 +1145,29 @@ class IOSpec extends BaseSpec with Discipline with IOPlatformSpecification {
         } must completeAs(())
       }
 
+      "cancelable waits for termination" in ticked { implicit ticker =>
+        def test(fin: IO[Unit]) = {
+          val go = IO.never.uncancelable.cancelable(fin)
+          go.start.flatMap(IO.sleep(1.second) *> _.cancel)
+        }
+
+        test(IO.unit) must nonTerminate
+        test(IO.raiseError(new Exception)) must nonTerminate
+        test(IO.canceled) must nonTerminate
+      }
+
+      "cancelable cancels task" in ticked { implicit ticker =>
+        def test(fin: IO[Unit]) =
+          IO.deferred[Unit].flatMap { latch =>
+            val go = IO.never[Unit].onCancel(latch.complete(()).void).cancelable(fin)
+            go.start.flatMap(IO.sleep(1.second) *> _.cancel) *> latch.get
+          }
+
+        test(IO.unit) must completeAs(())
+        test(IO.raiseError(new Exception)) must completeAs(())
+        test(IO.canceled) must completeAs(())
+      }
+
       "only unmask within current fiber" in ticked { implicit ticker =>
         var passed = false
         val test = IO uncancelable { poll =>