test(Topic): Add failing tests for #3644 (lying publish1 result)

core/shared/src/test/scala/fs2/concurrent/TopicSuite.scala

@@ -185,4 +185,41 @@ class TopicSuite extends Fs2Suite {
 
     TestControl.executeEmbed(program) // will fail if program is deadlocked
   }
+
+  // https://github.com/typelevel/fs2/issues/3644
+  test(
+    "when publish1 returns success, subscribers must receive the event, even if the publish1 races with close".fail
+  ) {
+    val check: IO[Unit] =
+      Topic[IO, String]
+        .flatMap { t =>
+          t.subscribeAwaitUnbounded.replicateA(100).use { subs =>
+            IO.both(t.publish1("foo"), t.close) // racing publish1 and close
+              .flatMap {
+                case (_, Left(_)) =>
+                  fail("There's no reason for Topic closure to fail")
+                case (published, Right(())) =>
+                  // the topic is closed
+                  subs
+                    .traverse(sub =>
+                      sub.compile.toList // all subscriptions must terminate, since the Topic was closed
+                    )
+                    .map { eventss =>
+                      val expected: List[String] =
+                        published match {
+                          case Right(()) =>
+                            // publication succeeded, expecting singleton list with the event
+                            List("foo")
+                          case Left(Topic.Closed) =>
+                            // publication rejected, expecting empty list
+                            Nil
+                        }
+                      eventss.foreach(events => assertEquals(events, expected))
+                    }
+              }
+          }
+        }
+
+    check.replicateA_(1000)
+  }
 }