[indexer] Objects Snapshot Processor continuously reads from stream (#…

…19232)

## Description 

In ci, we saw that the `objects_snapshot_processor` stalls if there are
unprocessed checkpoint data that we cannot commit unless we have a
continuous stream of checkpoints from `[start_cp, max_allowed_cp]`. To
address this, the processor should continue to read from stream whether
`unprocessed` is empty or not.

## Test plan 

Ran indexer manually to observe that `objects_snapshot_processor`
doesn't stall

---

## Release notes

Check each box that your changes affect. If none of the boxes relate to
your changes, release notes aren't required.

For each box you select, include information after the relevant heading
that describes the impact of your changes that a user might notice and
any actions they must take to implement updates.

- [ ] Protocol: 
- [ ] Nodes (Validators and Full nodes): 
- [ ] Indexer: 
- [ ] JSON-RPC: 
- [ ] GraphQL: 
- [ ] CLI: 
- [ ] Rust SDK:
- [ ] REST API:

crates/sui-indexer/src/handlers/objects_snapshot_processor.rs

@@ -190,13 +190,9 @@ impl ObjectsSnapshotProcessor {
                         // `objects_snapshot` table.
                         let max_allowed_cp = latest_indexer_cp - config.snapshot_min_lag as u64;
 
-                        // Fetch `batch_size` more data from the stream if the buffer is empty and
-                        // we still need to catch up.
-                        if unprocessed.is_empty() {
-                            if let Some(new_changes) = stream.next().await {
-                                for checkpoint in new_changes {
-                                    unprocessed.insert(checkpoint.checkpoint_sequence_number, checkpoint);
-                                }
+                        if let Some(new_changes) = stream.next().await {
+                            for checkpoint in new_changes {
+                                unprocessed.insert(checkpoint.checkpoint_sequence_number, checkpoint);
                             }
                         }