Fuses written batches together and appends them to persist using one
compare_and_append call. Writing only happens for batch descriptions where
we know that no future batches will arrive, that is, for those batch
descriptions that are not beyond the frontier of both the
batch_descriptions and batches inputs.
Continuously writes the difference between persist_stream and
desired_stream into persist, such that the persist shard is made to
contain the same updates as desired_stream. This is done via a multi-stage
operator graph:
Whenever the frontier advances, this mints a new batch description (lower
and upper) that writers should use for writing the next set of batches to
persist.
Writes desired_stream - persist_stream to persist, but only for updates
that fall into batch a description that we get via batch_descriptions.
This forwards a HollowBatch for any batch of updates that was written.