Add Stream#counter(Stream<Long>> to count values until flush and feed the argument stream with each counter
Fine grained error/flush cascading, mapMany cascade flushes to join on, enhance action utils to display flushable actions.