ShardFollowNodeTask fetch operations twice

[dnhatn]

Since #31581, ShardFollowNodeTask may fetch some range twice. The following log indicates that we fetched the range [1680 to 2024] twice.

[ShardFollowNodeTask] fetch from=1620, to=1679, receive [1620 2024]
[ShardFollowNodeTask] fetch from=1680, to=1984, receive [1680 2084]

This bug and if the follower is not using the FollowingEngine (PR #32448) can explain why we have many deletes on the replicas of the follower (but not on the primaries of the follower).

[ShardFollowNodeTask] send shard request from=0, batch=405, max=-1, retry=0, gcp=-1, num_reads=1
[ShardFollowNodeTask] fetch from=0, to=-1, receive [0 404]
[ShardFollowNodeTask] send shard request from=405, batch=405, max=809, retry=0, gcp=1679, num_reads=1
[ShardFollowNodeTask] send shard request from=810, batch=405, max=1214, retry=0, gcp=1679, num_reads=2
[ShardFollowNodeTask] send shard request from=1215, batch=405, max=1619, retry=0, gcp=1679, num_reads=3
[ShardFollowNodeTask] send shard request from=1620, batch=405, max=1679, retry=0, gcp=1679, num_reads=4

[ShardFollowNodeTask] fetch from=810, to=1214, receive [810 1214]
[ShardFollowNodeTask] send shard request from=1680, batch=405, max=1984, retry=0, gcp=1984, num_reads=4
[ShardFollowNodeTask] fetch from=405, to=809, receive [405 809]
[ShardFollowNodeTask] fetch from=1215, to=1619, receive [1215 1619]
[ShardFollowNodeTask] fetch from=1620, to=1679, receive [1620 2024] (*)
[ShardFollowNodeTask] send shard request from=2025, batch=405, max=2322, retry=0, gcp=2322, num_reads=2
[ShardFollowNodeTask] fetch from=1680, to=1984, receive [1680 2084] (**)

[elasticmachine]

Pinging @elastic/es-distributed

[dnhatn]

/cc @martijnvg @jasontedor @bleskes

[bleskes]

Good catch @dnhatn. This feels like a bug to me? we should not get ops we didn't ask for?

[ShardFollowNodeTask] fetch from=1620, to=1679, receive [1620 2024] (*)

[bleskes]

Also, I'm not sure I follow how this explains the deletes, can you clarify? if the primary processes the same ops twice, but ignores the associated seq# and issues it's own (and replicates it) how does it explain that the primary has no deletes but the replica does?

[dnhatn]

@bleskes The primary uses version numbers to resolve the indexing plan and rejects the duplicate operations; Whereas the replica uses seq# to resolve the indexing plan, and indexes the duplicate operations as stale documents. Does it sound reasonable to you?

[bleskes]

Thx @dnhatn I forgot that we don't use the bulk shard request and that exceptions on the primary don't map to lack of replications. It does. Thanks!

[bleskes]

Thx @dnhatn I forgot that we don't use the bulk shard request and that exceptions on the primary don't map to lack of replications. It does. Thanks!

[dnhatn]

Fixed in #32455