TASK LOOPING (in both Core and Cloud)

[cicdw]

Thanks for contributing to Prefect!

Please describe your work and make sure your PR:

• adds new tests (if appropriate)
• updates CHANGELOG.md (if appropriate)
• updates docstrings for any new functions or function arguments, including docs/outline.toml for API reference docs (if appropriate)

Note that your PR will not be reviewed unless all three boxes are checked.

What does this PR change?

This PR introduces "Task Looping" (first described in PIN 11). The implementation here will actually work in Cloud almost immediately (once the Cloud serializers are updated).

In order to make Looping work off-the-shelf, this PR:

• introduces a new Looped state which carries the loop_count
• introduces a new LOOP signal for triggering a loop
• places the appropriate data in context (loop count + loop result)
• re-uses our cached_inputs mechanism for conveying relevant data to retries
• allows for loops to begin at arbitrary points by setting the appropriate task_loop_count in context

Why is this PR important?

This PR furthers the ability for Flows to dynamically spawn tasks. Mapping is to for loops as Looping is to while loops.

This of course will need extensive documentation to fully expose.

[codecov]

Codecov Report

Merging #1356 into master will increase coverage by 0.05%.
The diff coverage is 100%.

[jlowin]

Makes me wonder if both Looped and Mapped should become Looping and Mapping since they imply that more work is going to be done

[cicdw]

Yea I thought about this; I'm happy to change it but here's my reasoning for Looped: when a Task enters a Looped state it lets us know that a single iteration of the Loop was complete. The full "looping" mechanism actually involves Looped -> Running -> Looped -> Running.

This is in contrast to a Running state - when a Task is in a Running state, it actually means there is work currently being done.

[jlowin]

I'm in