NestedEnumerator leaks state when an enumerator becomes empty between job runs

[liamstewart]

When NestedEnumerator reaches the end of a layer, it resets the cursor state for that layer so that items aren't skipped for the next nested iteration.

However, the reset only happens if enumeration happens. If something causes a nested iterator to become empty between job runs, say there is conditional logic in the enumerator builder the state of an item changes such that it no longer needs processing, the enumerator skips over nested resets. The next outer item is processed, but the inner state is stale so inner items are skipped.

test "empty enumerator resets later cursor state" do
  outer = [10, 20, 30]
  middle = [1, 2, 3]
  inner = [:a, :b, :c]

  # next- 20, 2, :b
  cursor = [0, 0, 0]
  short_circuit = false

  enums = [
    ->(cursor) { enumerator_builder.build_array_enumerator(outer, cursor: cursor) },
    ->(outer_item, cursor) {
      if outer_item == 20 && short_circuit
        enumerator_builder.build_array_enumerator([], cursor: cursor)
      else
        enumerator_builder.build_array_enumerator(middle, cursor: cursor)
      end
    },
    ->(outer_item, middle_item, cursor) {
      v = inner.map { |i| [outer_item, middle_item, i] }
      enumerator_builder.build_array_enumerator(v, cursor: cursor)
    },
  ]

  enum = NestedEnumerator.new(enums, cursor: cursor).each
  value_and_cursor = enum.take(1)[0]
  assert_equal([[20, 2, :b], [0, 0, 1]], value_and_cursor)

  # next would normally be [20, 2, :c] but we short-circuit 20 and don't finish iterating over it so next should be [30, 1, :a].
  # however, because the inner cursor state isn't reset, next actually is [30, 1, :c] - :a and :b are skipped
  short_circuit = true
  enum = NestedEnumerator.new(enums, cursor: value_and_cursor[1]).each
  value_and_cursor = enum.take(1)[0]
  assert_equal([[30, 1, :a], [1, nil, 2]], value_and_cursor)
end