## Description 

without this change, the unprocessed buffer will grow unboundedly until
OOM
it did not manifest on previous processor b/c it has sleep codes of 
```
_ = tokio::time::sleep(std::time::Duration::from_secs(config.sleep_duration))
```
and `sleep_duration` is 5 seconds.

## Test plan 

- correctness via the added objects_snapshot test
- oom via experiment result on benchmark env with mem usage
[link](https://app.datadoghq.com/metric/explorer?fromUser=false&start=1729280070529&end=1729283670529&paused=false#N4Ig7glgJg5gpgFxALlAGwIYE8D2BXJVEADxQEYAaELcqyKBAC1pEbghkcLIF8qo4AMwgA7CAgg4RKUAiwAHOChASAtnADOcAE4RNIKtrgBHPJoQaUAbVBGN8qVoD6gnNtUZCKiOq279VKY6epbINiAiGOrKQdpYZAYgUJ4YThr42gDGSsgg6gi6mZaBZnHKGABuMMiZUggYojoAdOqqblhNeBoY8MAA1ngARnBOkb7yGNnI2vKZALTDIpmMHtp9AAQU67Ui9Y3ao1FwyBp4EHOiAsQ6POuDWOvADlCH6jwgPAC6VK7ueJihcK-VT-DAxUrxD7fEAaORoHKgCbwhAIHJJHAwJyZAEaCCZRJoRpOOSKZTpQlQAlE+hMZQiNweNAffgQeyYLDEhRowkiJRfHh8GHyQkIADCUmEMBQIn+aB4QA)


---

## 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: