When we throw with batchWithTimeout loop is not finalized

[oleg-slapdash]

When an async generator wrapped with batchWithTImeout is never "finalized", while batch and naked loop is.

Please see examples below, both "batch" and simple loop prints Loop Finally while batchWithTimeout does not print Loop Finally and continues reading from the stream.

Simple loop:

 async function* generate() {
    try {
      for (let i = 0; i < 20; i++) {
        console.log({ i });
        yield i;
        await new Promise((resolve) => setTimeout(resolve, 10));
      }

      console.log("Loop Completed");
    } catch (err) {
      console.log("Loop Caught error", err);
      throw err;
    } finally {
      console.log("Loop Finally");
    }
  }

  try {
    for await (const i of generate()) {
      console.log({ i });
      if (i === 10) {
        console.log("Thrown");
        throw new Error("Thrown");
      }
    }

    console.log("Global Completed");
  } catch (err) {
    console.log("Global Caught error", err);
    throw err;
  } finally {
    console.log("Global Finally");
  }

Output

{ i: 0 }
{ i: 0 }
{ i: 1 }
{ i: 1 }
{ i: 2 }
{ i: 2 }
{ i: 3 }
{ i: 3 }
{ i: 4 }
{ i: 4 }
{ i: 5 }
{ i: 5 }
{ i: 6 }
{ i: 6 }
{ i: 7 }
{ i: 7 }
{ i: 8 }
{ i: 8 }
{ i: 9 }
{ i: 9 }
{ i: 10 }
{ i: 10 }
Thrown
Loop Finally
Global Caught error Error: Thrown

Loop with a batch:

async function* generate() {
  try {
    for (let i = 0; i < 20; i++) {
      console.log({ i });
      yield i;
      await new Promise((resolve) => setTimeout(resolve, 10));
    }

    console.log("Loop Completed");
  } catch (err) {
    console.log("Loop Caught error", err);
    throw err;
  } finally {
    console.log("Loop Finally");
  }
}

try {
  for await (const chunk of batch(3, generate())) {
    console.log({ chunk });
    if (chunk.includes(10)) {
      console.log("Thrown");
      throw new Error("Thrown");
    }
  }

  console.log("Global Completed");
} catch (err) {
  console.log("Global Caught error", err);
  throw err;
} finally {
  console.log("Global Finally");
}

Output

{ i: 0 }
{ i: 1 }
{ i: 2 }
{ chunk: [ 0, 1, 2 ] }
{ i: 3 }
{ i: 4 }
{ i: 5 }
{ chunk: [ 3, 4, 5 ] }
{ i: 6 }
{ i: 7 }
{ i: 8 }
{ chunk: [ 6, 7, 8 ] }
{ i: 9 }
{ i: 10 }
{ i: 11 }
{ chunk: [ 9, 10, 11 ] }
Thrown
Loop Finally
Global Caught error Error: Thrown

async function* generate() {
  try {
    for (let i = 0; i < 20; i++) {
      console.log({ i });
      yield i;
      await new Promise((resolve) => setTimeout(resolve, 10));
    }

    console.log("Loop Completed");
  } catch (err) {
    console.log("Loop Caught error", err);
    throw err;
  } finally {
    console.log("Loop Finally");
  }
}

try {
  for await (const chunk of batchWithTimeout(3, 1000000, generate())) {
    console.log({ chunk });
    if (chunk.includes(10)) {
      console.log("Thrown");
      throw new Error("Thrown");
    }
  }

  console.log("Global Completed");
} catch (err) {
  console.log("Global Caught error", err);
  throw err;
} finally {
  console.log("Global Finally");
}

Output

{ i: 0 }
{ i: 1 }
{ i: 2 }
{ chunk: [ 0, 1, 2 ] }
{ i: 3 }
{ i: 4 }
{ i: 5 }
{ chunk: [ 3, 4, 5 ] }
{ i: 6 }
{ i: 7 }
{ i: 8 }
{ chunk: [ 6, 7, 8 ] }
{ i: 9 }
{ i: 10 }
{ i: 11 }
{ chunk: [ 9, 10, 11 ] }
Thrown
Global Caught error Error: Thrown
{ i: 12 }