feat(compiler): for ... rec [LNG-307] (#1026)

* Add parser

* Add semantics

* Add inlining

* Add range test

* Rewrite to for ... rec

* Rewrite tests

* Fix import

* Add nested test

* Remove only

* Add yes|no test

* Add multi rec test

* Add pipeline test

* Unignore tests

* Change timeouts

* Add remote rec test

* Fix integration tests

* Add parser test

* Add semantics test

* Add inlining test

* Add comment

integration-tests/aqua/examples/recursiveStreams/multiRec.aqua

@@ -0,0 +1,22 @@
+aqua MultiRec
+
+export TestService, multiRecStream
+
+service TestService("test-srv"):
+  handle(i: i32) -> []i32
+
+func multiRecStream(init: i32, target: i32) -> []i32:
+  result: *string
+  loop: *i32
+
+  loop <<- init
+  for l <- loop rec:
+    news <- TestService.handle(l)
+    for n <- news:
+      loop <<- n
+    if l == target:
+      result <<- "done"
+
+  join result!
+
+  <- loop

integration-tests/aqua/examples/recursiveStreams/nested.aqua

@@ -0,0 +1,19 @@
+aqua Nested
+
+export nested
+
+func nested(n: u32) -> []u32:
+    result: *u32
+    iterator: *u32
+
+    iterator <<- 0
+    for i <- iterator rec:
+        if i < n:
+            for j <- iterator rec:
+                result <<- j
+            iterator <<- i + 1
+
+    if n > 0:
+        join result[n * (n + 1) / 2 - 1]
+
+    <- result

integration-tests/aqua/examples/recursiveStreams/pipeline.aqua

@@ -0,0 +1,29 @@
+aqua Pipeline
+
+export pipelineStream
+
+func pipelineStream(init: i32, target: i32) -> []i32:
+  result: *string
+
+  loop1: *i32
+  loop2: *i32
+  loop3: *i32
+
+  loop1 <<- init
+  for l <- loop1 rec:
+    if l < target:
+      loop1 <<- l + 1
+      loop2 <<- l * 3
+
+  for l <- loop2 rec:
+    loop3 <<- l 
+    loop3 <<- l + 1
+    loop3 <<- l + 2
+
+  for l <- loop3 rec:
+    if l == target:
+      result <<- "success"
+
+  join result!
+
+  <- loop3

integration-tests/aqua/examples/recursiveStreams/range.aqua

@@ -0,0 +1,18 @@
+aqua Range
+
+export range
+
+func range(a: i32, b: i32) -> []i32:
+    result: *i32
+    iterator: *i32
+
+    iterator <<- a
+    for i <- iterator rec:
+        if i < b:
+            result <<- i
+            iterator <<- i + 1
+
+    if b > a:
+        join result[b - a - 1]
+
+    <- result

integration-tests/aqua/examples/recursiveStreams/remoteRec.aqua

@@ -0,0 +1,19 @@
+aqua RemoteRec
+
+export RemoteSrv, remoteRecStream
+
+service RemoteSrv("remote-srv"):
+  handle(i: i32) -> i32
+
+func remoteRecStream(init: i32, target: i32, friend: string, friendRelay: string) -> []i32:
+  loop: *i32
+
+  loop <<- init
+  for l <- loop rec:
+    on friend via friendRelay:
+      if l < target:
+        loop <- RemoteSrv.handle(l)
+
+  join loop[target - init]
+
+  <- loop

integration-tests/aqua/examples/recursiveStreams/yesNo.aqua

@@ -0,0 +1,21 @@
+aqua YesNo
+
+export YesNoService, yesNoStream
+
+service YesNoService("yesno"):
+  get() -> string
+
+func yesNoStream() -> []string:
+  result: *string
+  loop: *string
+
+  loop <<- "yes"
+  for l <- loop rec:
+    if l == "yes":
+      loop <- YesNoService.get()
+    else:
+      result <<- "success"
+
+  join result!
+
+  <- loop

integration-tests/src/__test__/examples.spec.ts

@@ -90,7 +90,7 @@ import {
   streamArgsCall,
   modifyStreamCall,
   returnDerivedStreamCall,
-  lng280BugWithForEmptyStreamFuncCall
+  lng280BugWithForEmptyStreamFuncCall,
 } from "../examples/streamArgsCall.js";
 import { streamResultsCall } from "../examples/streamResultsCall.js";
 import { structuralTypingCall } from "../examples/structuralTypingCall.js";
@@ -135,7 +135,6 @@ import {
   joinIdxLocalCall,
   joinIdxRelayCall,
 } from "../examples/joinCall.js";
-import { recursiveStreamsCall } from "../examples/recursiveStreamsCall.js";
 import { renameVarsCall } from "../examples/renameVars.js";
 import {
   arraySugarCall,
@@ -161,6 +160,12 @@ import {
   returnArrowCall,
   returnArrowChainCall,
 } from "../examples/returnArrowCall.js";
+import { rangeCall } from "../examples/recursiveStreams/rangeCall.js";
+import { nestedCall } from "../examples/recursiveStreams/nestedCall.js";
+import { yesNoStreamCall } from "../examples/recursiveStreams/yesNoStreamCall.js";
+import { multiRecStreamCall } from "../examples/recursiveStreams/multiRecStreamCall.js";
+import { pipelineStreamCall } from "../examples/recursiveStreams/pipelineCall.js";
+import { remoteRecStreamCall } from "../examples/recursiveStreams/remoteRecCall.js";
 
 var selfPeerId: string;
 var peer1: IFluenceClient;
@@ -217,6 +222,77 @@ describe("Testing examples", () => {
     await stop();
   });
 
+  describe("for ... rec", () => {
+    const range = (start: number, end: number) =>
+      Array.from({ length: end - start }, (v, k) => k + start);
+
+    it("range", async () => {
+      for (const i of range(-5, 5)) {
+        for (const j of range(-5, 5)) {
+          const result = await rangeCall(i, j);
+          if (i < j) {
+            expect(result).toEqual(range(i, j));
+          } else {
+            expect(result).toEqual([]);
+          }
+        }
+      }
+    }, 15000);
+
+    /**
+     * This test does not work due to Aqua VM
+     */
+    it.skip("nested", async () => {
+      for (const i of range(0, 10)) {
+        const result = await nestedCall(i);
+        expect(result).toEqual(range(0, i).flatMap((x) => range(0, x + 1)));
+      }
+    }, 15000);
+
+    it("yes|no stream", async () => {
+      for (const i of range(1, 10)) {
+        const yesNo = await yesNoStreamCall(i);
+        expect(yesNo).toEqual(
+          range(0, i)
+            .map((_) => "yes")
+            .concat(["no"]),
+        );
+      }
+    }, 15000);
+
+    it("multi rec stream", async () => {
+      const handle = (i: number) => {
+        if (i % 3 === 0) return [i + 1];
+        if (i % 3 === 1) return [i + 1, i + 2];
+        return [];
+      };
+      for (const i of range(1, 10)) {
+        const loop = await multiRecStreamCall(0, i, handle);
+        range(0, i + 1).forEach((j) => {
+          expect(loop).toContain(j);
+        });
+      }
+    }, 15000);
+
+    it("pipeline", async () => {
+      for (const i of range(1, 10)) {
+        const result = await pipelineStreamCall(0, i);
+        expect(result.sort()).toEqual(range(0, i + 1));
+      }
+    }, 15000);
+
+    /**
+     * This test does not work due to `for ... rec`
+     * not taking topology into account
+     */
+    it.skip("remote rec", async () => {
+      for (const i of range(0, 10)) {
+        const result = await remoteRecStreamCall(0, i, peer2);
+        expect(result).toEqual(range(0, i + 1));
+      }
+    }, 15000);
+  });
+
   it("callArrow.aqua args bug 426", async () => {
     let argResult = await reproArgsBug426Call();
 
@@ -630,29 +706,41 @@ describe("Testing examples", () => {
   it.skip("streamArgs.aqua LNG-280 with for", async () => {
     let result = await lng280BugWithForCall();
     expect(result).toEqual([
-        "valueUseStream",
-        "valueReturnStream",
-        "valueUseStream",
-        "valueReturnStream",
-        "valueUseStream",
-        "valueReturnStream"
+      "valueUseStream",
+      "valueReturnStream",
+      "valueUseStream",
+      "valueReturnStream",
+      "valueUseStream",
+      "valueReturnStream",
     ]);
   });
 
   it("streamArgs.aqua LNG-280 with for and anonymous stream", async () => {
     let result = await lng280BugWithForAnonStreamCall();
-    expect(result).toEqual([[1, 1], [1, 2], [1, 3], [1, 4], [1, 5]]);
+    expect(result).toEqual([
+      [1, 1],
+      [1, 2],
+      [1, 3],
+      [1, 4],
+      [1, 5],
+    ]);
   });
 
   it("streamArgs.aqua LNG-280 with for and anonymous stream from function", async () => {
-      let result = await lng280BugWithForEmptyStreamFuncCall();
-      expect(result).toEqual([[1, 1], [1, 2], [1, 3], [1, 4], [1, 5]]);
-    });
+    let result = await lng280BugWithForEmptyStreamFuncCall();
+    expect(result).toEqual([
+      [1, 1],
+      [1, 2],
+      [1, 3],
+      [1, 4],
+      [1, 5],
+    ]);
+  });
 
   it.skip("streamArgs.aqua return derived stream", async () => {
-      let result = await returnDerivedStreamCall();
-      expect(result).toEqual([1]);
-    });
+    let result = await returnDerivedStreamCall();
+    expect(result).toEqual([1]);
+  });
 
   it("streamResults.aqua", async () => {
     let streamResultsResult = await streamResultsCall();
@@ -822,15 +910,6 @@ describe("Testing examples", () => {
   //     expect(res).toEqual("ok")
   // });
 
-  // TODO: uncomment
-  // it('recursiveStreams.aqua', async () => {
-  //     let [sucList, loopList] = await recursiveStreamsCall();
-  //     console.log(sucList);
-  //     console.log(loopList);
-  //     expect(loopList).toEqual(['yes', 'yes', 'yes', 'yes', 'no']);
-  //     expect(sucList.length).toEqual(5);
-  // });
-
   it("renameVars.aqua", async () => {
     let renameVarsResult = await renameVarsCall();
     expect(renameVarsResult).toEqual(["ok", "ok"]);

integration-tests/src/examples/recursiveStreams/multiRecStreamCall.ts

@@ -0,0 +1,14 @@
+import {
+  multiRecStream,
+  registerTestService,
+} from "../../compiled/examples/recursiveStreams/multiRec.js";
+
+export async function multiRecStreamCall(
+  init: number,
+  target: number,
+  handle: (i: number) => number[],
+): Promise<number[]> {
+  registerTestService({ handle });
+
+  return await multiRecStream(init, target);
+}

integration-tests/src/examples/recursiveStreams/nestedCall.ts

@@ -0,0 +1,5 @@
+import { nested } from "../../compiled/examples/recursiveStreams/nested.js";
+
+export async function nestedCall(n: number): Promise<number[]> {
+  return await nested(n);
+}

integration-tests/src/examples/recursiveStreams/pipelineCall.ts

@@ -0,0 +1,8 @@
+import { pipelineStream } from "../../compiled/examples/recursiveStreams/pipeline.js";
+
+export async function pipelineStreamCall(
+  init: number,
+  target: number,
+): Promise<number[]> {
+  return await pipelineStream(init, target);
+}

integration-tests/src/examples/recursiveStreams/rangeCall.ts

@@ -0,0 +1,5 @@
+import { range } from "../../compiled/examples/recursiveStreams/range.js";
+
+export async function rangeCall(a: number, b: number): Promise<number[]> {
+  return await range(a, b);
+}

integration-tests/src/examples/recursiveStreams/remoteRecCall.ts

@@ -0,0 +1,15 @@
+import { IFluenceClient } from "@fluencelabs/js-client";
+import { remoteRecStream } from "../../compiled/examples/recursiveStreams/remoteRec.js";
+
+export async function remoteRecStreamCall(
+  init: number,
+  target: number,
+  peer: IFluenceClient,
+): Promise<number[]> {
+  return await remoteRecStream(
+    init,
+    target,
+    peer.getPeerId(),
+    peer.getRelayPeerId(),
+  );
+}

integration-tests/src/examples/recursiveStreams/yesNoStreamCall.ts

@@ -0,0 +1,16 @@
+import {
+  yesNoStream,
+  registerYesNoService,
+} from "../../compiled/examples/recursiveStreams/yesNo.js";
+
+export async function yesNoStreamCall(limit: number): Promise<string[]> {
+  let i = 1;
+  registerYesNoService({
+    get: () => {
+      i += 1;
+      return i > limit ? "no" : "yes";
+    },
+  });
+
+  return await yesNoStream();
+}