fix(compiler): Fix SO in Topology [fixes LNG-149] (#752)

* Skip `next` in `for` topology resolution

* Add tests

model/transform/src/main/scala/aqua/model/transform/topology/OpModelTreeCursor.scala

@@ -102,5 +102,5 @@ case class OpModelTreeCursor(
     }.getOrElse(Chain.empty).memoize
 
   override def toString: String =
-    s"[${tree.head.prev.length}]${op.show} /: ${moveUp.getOrElse("(|)")}"
+    s"[${tree.head.prev.length}] ${op.show} /: ${moveUp.getOrElse("(|)")}"
 }

model/transform/src/main/scala/aqua/model/transform/topology/Topology.scala

@@ -12,6 +12,7 @@ import cats.free.Cofree
 import cats.syntax.traverse.*
 import cats.syntax.show.*
 import cats.syntax.apply.*
+import cats.syntax.option.*
 import scribe.Logging
 
 /**
@@ -425,24 +426,37 @@ object Topology extends Logging {
     // Optimization: get all the path inside the For block out of the block, to avoid repeating
     // hops for every For iteration
     override def beginsOn(current: Topology): Eval[List[OnModel]] =
-      (current.forModel zip current.firstChild.map(_.beginsOn)).map { case (f, b) =>
-        // Take path until this for's iterator is used
-        b.map(
-          _.reverse
-            .foldLeft((true, List.empty[OnModel])) {
-              case ((true, acc), OnModel(_, r)) if r.exists(_.usesVarNames.contains(f.item)) =>
-                (false, acc)
-              case ((true, acc @ (OnModel(_, r @ (r0 ==: _)) :: _)), OnModel(p, _))
-                  if p.usesVarNames.contains(f.item) =>
-                // This is to take the outstanding relay and force moving there
-                (false, OnModel(r0, r) :: acc)
-              case ((true, acc), on) => (true, on :: acc)
-              case ((false, acc), _) => (false, acc)
-            }
-            ._2
-        )
+      // Skip `next` child because its `beginsOn` depends on `this.beginsOn`, see [bug LNG-149]
+      (current.forModel zip firstNotNextChild(current).map(_.beginsOn)).map {
+        case (model, childBeginsOn) =>
+          // Take path until this for's iterator is used
+          childBeginsOn.map(
+            _.reverse
+              .foldLeft((true, List.empty[OnModel])) {
+                case ((true, acc), OnModel(_, r))
+                    if r.exists(_.usesVarNames.contains(model.item)) =>
+                  (false, acc)
+                case ((true, acc @ (OnModel(_, r @ (r0 ==: _)) :: _)), OnModel(p, _))
+                    if p.usesVarNames.contains(model.item) =>
+                  // This is to take the outstanding relay and force moving there
+                  (false, OnModel(r0, r) :: acc)
+                case ((true, acc), on) => (true, on :: acc)
+                case ((false, acc), _) => (false, acc)
+              }
+              ._2
+          )
       } getOrElse super.beginsOn(current)
 
+    /**
+     * Find first child that is not `next`
+     */
+    private def firstNotNextChild(current: Topology): Option[Topology] = for {
+      first <- current.firstChild
+      notNext <- first.cursor.op match {
+        case _: NextModel => first.nextSibling
+        case _ => first.some
+      }
+    } yield notNext
   }
 
   object SeqNext extends Begins {

model/transform/src/test/scala/aqua/model/transform/topology/TopologySpec.scala

@@ -12,6 +12,10 @@ import cats.free.Cofree
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 import cats.syntax.show.*
+import cats.syntax.option.*
+import aqua.types.ArrayType
+import aqua.raw.ConstantRaw.initPeerId
+import aqua.model.ForModel.NullMode
 
 class TopologySpec extends AnyFlatSpec with Matchers {
 
@@ -412,7 +416,8 @@ class TopologySpec extends AnyFlatSpec with Matchers {
         through(relay),
         callRes(0, otherPeer),
         ParRes.wrap(
-          FoldRes("i", valueArray, Some(ForModel.NeverMode)).wrap(ParRes.wrap(callRes(2, otherPeer2), NextRes("i").leaf))
+          FoldRes("i", valueArray, Some(ForModel.NeverMode))
+            .wrap(ParRes.wrap(callRes(2, otherPeer2), NextRes("i").leaf))
         ),
         through(relay),
         callRes(3, initPeer)
@@ -879,4 +884,51 @@ class TopologySpec extends AnyFlatSpec with Matchers {
 
     proc.equalsOrShowDiff(expected) should be(true)
   }
+
+  "topology resolver" should "handle empty for correctly [bug LNG-149]" in {
+    val streamName = "array-inline"
+    val iterName = "a-0"
+    val stream = VarModel(streamName, StreamType(LiteralType.number))
+    val array = VarModel(s"$streamName-0", ArrayType(LiteralType.number))
+
+    val literal = (i: String) => LiteralModel(i, LiteralType.number)
+
+    val push = (i: String) =>
+      PushToStreamModel(
+        literal(i),
+        CallModel.Export(stream.name, stream.`type`)
+      ).leaf
+
+    val model = OnModel(initPeer, Chain.one(relay)).wrap(
+      SeqModel.wrap(
+        RestrictionModel(streamName, true).wrap(
+          push("1"),
+          push("2"),
+          CanonicalizeModel(stream, CallModel.Export(array.name, array.`type`)).leaf
+        ),
+        ForModel(iterName, array).wrap(
+          NextModel(iterName).leaf
+        )
+      )
+    )
+
+    val proc = Topology.resolve(model).value
+
+    val expected = SeqRes.wrap(
+      RestrictionRes(streamName, true).wrap(
+        ApRes(literal("1"), CallModel.Export(stream.name, stream.`type`)).leaf,
+        ApRes(literal("2"), CallModel.Export(stream.name, stream.`type`)).leaf,
+        CanonRes(
+          stream,
+          LiteralModel.fromRaw(initPeer),
+          CallModel.Export(array.name, array.`type`)
+        ).leaf
+      ),
+      FoldRes(iterName, array, NullMode.some).wrap(
+        NextRes(iterName).leaf
+      )
+    )
+
+    proc.equalsOrShowDiff(expected) shouldEqual true
+  }
 }

model/tree/src/main/scala/aqua/tree/TreeNode.scala

@@ -10,7 +10,11 @@ trait TreeNode[T <: TreeNode[T]] {
 
   lazy val leaf: Tree = Cofree(self, Eval.now(Chain.empty))
 
-  def wrap(children: Tree*): Tree = Cofree(self, Eval.now(Chain.fromSeq(children)))
+  def wrap(children: Tree*): Tree = wrap(Chain.fromSeq(children))
+
+  def wrap(children: List[Tree]): Tree = wrap(Chain.fromSeq(children))
+
+  def wrap(children: Chain[Tree]): Tree = Cofree(self, Eval.now(children))
 
   protected def wrapNonEmpty(children: List[Tree], empty: Tree): Tree = children match {
     case Nil => empty