fix: ensure Topic#publish1 returns accurate result during concurrent closure

Author: guptapratykshh

core/shared/src/main/scala/fs2/concurrent/Topic.scala

@@ -164,9 +164,22 @@ object Topic {
           signalClosure.tryGet.flatMap {
             case Some(_) => Topic.closed.pure[F]
             case None    =>
-              state.get
-                .flatMap { case (subs, _) => foreach(subs)(_.send(a).void) }
-                .as(Topic.rightUnit)
+              state.get.flatMap { case (subs, _) =>
+                subs.foldLeft(F.pure(Topic.rightUnit)) { case (acc, (_, chan)) =>
+                  acc.flatMap {
+                    case Left(Topic.Closed) => Topic.closed.pure[F]
+                    case Right(_)           =>
+                      chan.send(a).flatMap {
+                        case Right(_) => Topic.rightUnit.pure[F]
+                        case Left(_)  =>
+                          signalClosure.tryGet.map {
+                            case Some(_) => Topic.closed
+                            case None    => Topic.rightUnit
+                          }
+                      }
+                  }
+                }
+              }
           }
 
         def subscribeAwait(maxQueued: Int): Resource[F, Stream[F, A]] =

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

@@ -188,7 +188,7 @@ class TopicSuite extends Fs2Suite {
 
   // 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
+    "when publish1 returns success, subscribers must receive the event, even if the publish1 races with close"
   ) {
     val check: IO[Unit] =
       Topic[IO, String]