Don't send unexpired invocations across threads

[jackkleeman]

The cleaner suffers from a lot of contention due to sending all invocations, including unexpired ones, from the io thread to the worker thread. It's better if we can check for expiry inside the io thread, but we don't want to put all that logic inside the partition store crate. So we use a filter_map method to identify the invocations we care about and send them back to the worker.

Before:

  Executed completed invocations cleanup in 3.598163684s
  Executed completed invocations cleanup in 3.786564823s
  Executed completed invocations cleanup in 3.70789558s
  Executed completed invocations cleanup in 3.35952469s
  Executed completed invocations cleanup in 3.56261731s
  Executed completed invocations cleanup in 3.851260922s
  Executed completed invocations cleanup in 3.873757502s
  Executed completed invocations cleanup in 3.653827479s

After:

  Executed completed invocations cleanup in 274.850528ms
  Executed completed invocations cleanup in 257.149421ms
  Executed completed invocations cleanup in 399.182153ms
  Executed completed invocations cleanup in 377.276183ms
  Executed completed invocations cleanup in 375.380197ms
  Executed completed invocations cleanup in 382.006936ms
  Executed completed invocations cleanup in 440.970359ms
  Executed completed invocations cleanup in 444.923974ms

[github-actions]

Test Results

  7 files  ±0    7 suites  ±0   4m 49s ⏱️ + 1m 6s
 54 tests ±0   53 ✅ ±0  1 💤 ±0  0 ❌ ±0 
223 runs  ±0  220 ✅ ±0  3 💤 ±0  0 ❌ ±0 

Results for commit a6a8de5. ± Comparison against base commit 358a3e8.

♻️ This comment has been updated with latest results.