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

Author: guptapratykshh

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

@@ -164,8 +164,21 @@ object Topic {
             case State.Closed() =>
               Topic.closed.pure[F]
             case State.Active(subs, _) =>
-              foreach(subs)(_.send(a).void)
-                .as(Topic.rightUnit)
+              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(_)  =>
+                        // Channel send failed, check if topic was closed
+                        state.get.map {
+                          case State.Closed()     => Topic.closed
+                          case State.Active(_, _) => Topic.rightUnit
+                        }
+                    }
+                }
+              }
           }
 
         def subscribeAwait(maxQueued: Int): Resource[F, Stream[F, A]] =

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

@@ -218,7 +218,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]