Record realtime retransmits separate from drops

[lordnull]

If a source connection drops and reconnects, it re-registers with the realtime queue (rtq). The realtime queue records the difference between most recently sent sequence and last acked sequence numbers as dropped. Those will actually just be re-transmitted.

In addition, when an entry is actually dropped from the rtq (either due to max_bytes or overload protection), those are also logged as drops.

This can cause confusion when a connection is flakey: connections will be re-established and have the drops stat increase, but due to re-transmission, no objects are actually lost as would be indicated by the other types of drops. Therefore, on reconnect instead of increasing the drop count, a "re-transmitted" count should be used instead.

[jcapricebasho]

basho/riak_ee-issues#10

[cmeiklejohn]

Moving to 2.1.

[jonmeredith]

Investigating. Have isolated an issue that affects cascading realtime not accounting for drops correctly when there have been skips.

If cascading is not enabled (or pre-1.4), I believe everything works as designed. There is a small case where data is reported as dropped, when it is more accurately described as unknown if the realtime connection reconnects when the ACKed data has dropped off the RTQ, but the sending is inside it.

For skips, I can reproduce an issue where the skip count (which could be very high) is added to the drops every reconnect. Investigating a fix.

Details

Queue level stats
MinSeq - lowest queue sequence in RTQ
QSeq - highest queue sequence handed out

Per Consumer (remote site)
CSeq - highest queue sequence sent to consumer
ASeq - highest sequence number acked by consumer.

Two events cause drops to be recorded

1. Pushing to full RTQ (going over max_bytes)
2. Re-registering realtime connections

The consumer is in one of three states

1. CSeq and ASeq inside queue (MinSeq <= ASeq <= CSeq <= QSeq)
2. CSeq inside queue, aseq fallen off (ASeq < MinSeq <= CSeq <= QSeq)
3. CSeq and ASeq between MinSeq and QSet (ASeq <= CSeq < MinSeq <= QSeq)

Pushing to full RTQ

The queue is trimmed in trim_q_entries/6. TrimSeq is the queue
entry that is about to be trimmed and will no longer exist. MinSeq after
trimming is complete will be TrimSeq+1.

https://github.com/basho/riak_repl/blob/develop/src/riak_repl2_rtq.erl#L666

            Cs2 = [case CSeq < TrimSeq of
                       true ->
                           %% If the last sent qentry is before the trimseq
                           %% it will never be sent, so count it as a drop.
                           C#c{drops = C#c.drops + 1};
                       _ ->
                           C
                   end || C = #c{cseq = CSeq} <- Cs],

TrimSeq is the sequence number about to be removed. Once TrimSeq goes
beyond CSeq a drop is recorded against each consumer. After the queue
is brought back under the desired size, all of the CSeq/ASeqs are adjusted
as requred in trim_q

https://github.com/basho/riak_repl/blob/develop/src/riak_repl2_rtq.erl#L630

            NewCSeq = case ets:first(QTab) of
                          '$end_of_table' ->
                              QSeq; % if empty, make sure pull waits
                          MinSeq ->
                              MinSeq - 1
                      end,
            Cs3 = [case CSeq < NewCSeq of
                       true ->
                           C#c{cseq = NewCSeq, aseq = NewCSeq};
                       _ ->
                           C
                   end || C = #c{cseq = CSeq} <- Cs2],

Here's an example of each case

CSeq and ASeq inside trimmed queue (MinSeq <= ASeq <= CSeq <= QSeq)

|  99 | 100 | 101 | 102 | 103 | 104 | 105 |
   ^           ^           ^           ^
 MinSeq Trim  ASeq        CSeq         QSeq

CSeq > Trim, no drops.

CSeq inside trimmed queue, aseq fallen off (ASeq < MinSeq <= CSeq <= QSeq)

|x199x| 200 | 201 | 202 | 203 | 204 | 205 |
   ^           ^     ^     ^           ^
 MinSeq       ASeq  Trim CSeq         QSeq

CSeq > Trim, no drops while connected - if the connection is reregistered
RTQ will be unable to send all data on the wire and is at risk of a drop
on reconnection until the ACK comes in.

CSeq and ASeq between MinSeq and QSet (ASeq <= CSeq < MinSeq <= QSeq)

|x299x|x300x|x301x| 302 | 303 | 304 | 305 |
   ^           ^           ^     ^      ^
 ASeq        CSeq        MinSeq Qseq  Trim

CSeq < Trim, data dropped, unable to send.

Re-registering realtime connections.

When a realtime connection is dropped and re-established,
the consumer is re-registered.

https://github.com/basho/riak_repl/blob/develop/src/riak_repl2_rtq.erl#L309

            Drops = max(0, MinSeq - PrevASeq - 1),

            %% Re-registering, send from the last acked sequence
            CSeq = case C#c.aseq < MinSeq of % really max(PrevASeq, MinSeq)
                true -> MinSeq;
                false -> C#c.aseq
            end,

Here's an example of each case

CSeq and ASeq inside queue (MinSeq <= ASeq <= CSeq <= QSeq)

|  99 | 100 | 101 | 102 | 103 | 104 | 105 |
   ^           ^           ^           ^
 MinSeq      ASeq        CSeq         QSeq

Drops = max(0, 99 - 101 - 1 = -3) = 0
CSeq = max(99, 101) = 101

CSeq inside queue, aseq fallen off (ASeq < MinSeq <= CSeq <= QSeq)

|x199x| 200 | 201 | 202 | 203 | 204 | 205 |
   ^           ^           ^           ^
 ASeq        MinSeq        CSeq         QSeq

Drops = max(0, 201 - 199 - 1) = 1
CSeq = max(199, 201) = 201

The data sent between ASeq and MinSeq may have been delivered but
it is no longer in the queue to resend (1 entry).

CSeq and ASeq between MinSeq and QSet (ASeq <= CSeq < MinSeq <= QSeq)

|x299x|x300x|x301x| 302 | 303 | 304 | 305 |
   ^           ^           ^           ^
 ASeq        CSeq        MinSeq      QSeq

Drops = max(0, 303 - 299 - 1) = 3
CSeq = max(299, 303) = 303

The data between CSeq and MinSeq is definitely dropped (1 entry), it wasn't
ever sent over the wire.

The data between ASeq and CSeq may have been delivered or may not, there
is no way to know without requesting it from the remote node (2 entries).

cc: @jcapricebasho