Distributor: Skip re-fire of already pending tasks

On every cycle the distributor streams all tasks under waiting/ and
calls fireFromStorage() for each one, which allocates a fresh
ReplicationState and consolidates it into the pending PushOne for that
URI. PushOne deduplicates refs, since refBatchesToPush is a Set, but
its stateMap is a ListMultimap that appends unconditionally. A task
that stays in waiting/ across cycles therefore grows the pending
task's notification list by one entry per ref per cycle while adding
no work at all.

The distributor already snapshots the ref updates carried by pending
push tasks, in order to prune queue entries whose stored counterpart
is gone. Each streamed update is removed from that snapshot, leaving
only the prunable ones behind by the end. Reuse that removal as the
test, i.e a non-null result means a pending task on this node is already
holding the update, so re-firing it would append a redundant state. Skip
it, and let only updates with no pending task reach fireFromStorage().
Tasks written by another primary are absent from the snapshot and
continue to be fired, so cluster distribution is unaffected.

This change removes the repeat re-fires when Prune.TRUE rather than
every possible source of stateMap growth. The startup replay passes
Prune.FALSE and is deliberately left uncovered. It walks each waiting/
file exactly once, so no task is fired twice into the same pending push
and the growth is not reachable there today.

Change-Id: Ib9e77f9cf99cab3d5a6f110f0578ae02eafe9198