Stream.stream_forward and Stream.stream_backward should handle gaps

[lachiemurray]

I didn't see an Issue for this, apologies if there is one.

Currently, if Stream.stream_forward or Stream.stream_backward is called on a stream containing "gaps" in then duplicate events may be streamed.

For example, imagine two streams each with three events:

stream 1:    1, 2, 3
stream 2:    4, 5, 6

Resulting in the $all stream:

$all:   1, 2, 3, 4, 5, 6

If the first stream is hard deleted, resulting in:

$all:   4, 5, 6

and stream_forward is called with a batch size of 1, then according to the logic here the following calls would be made:

read_storage_forward(conn, stream, 0, 1, opts) - returning {:ok, [%{event_number: 4}]}
read_storage_forward(conn, stream, 0, 2, opts) - returning {:ok, [%{event_number: 4}]}
read_storage_forward(conn, stream, 0, 3, opts) - returning {:ok, [%{event_number: 4}]}
read_storage_forward(conn, stream, 0, 4, opts) - returning {:ok, [%{event_number: 4}]}
read_storage_forward(conn, stream, 0, 5, opts) - returning {:ok, [%{event_number: 5}]}
read_storage_forward(conn, stream, 0, 6, opts) - returning {:ok, [%{event_number: 6}]}

The problem being that the next_version is set to the previous version + the number of events returned in the last batch - meaning that since there are no events 1-3 the first four calls all return event 4.

The reverse is true for stream_backward

Resolved by setting the next_version to the last event retrieved plus or minus 1