[Networking] Do not spawn an RPC responder task if message_router dropped the message

[AhmedSoliman]

This also adds a new metric/counter to count the bytes we dropped after receiving due to a lossy service.

---

Stack created with Sapling. Best reviewed with ReviewStack.

• [RocksDB] Allow for WriteBatch buffer reuse after commit #4360
• -> [Networking] Do not spawn an RPC responder task if message_router dropped the message #4356

[github-actions]

Test Results

  7 files  +  2    7 suites  +2   2m 44s ⏱️ + 1m 44s
 47 tests + 21   47 ✅ + 21  0 💤 ±0  0 ❌ ±0 
200 runs  +156  200 ✅ +156  0 💤 ±0  0 ❌ ±0 

Results for commit daa27da. ± Comparison against base commit 42114b8.

This pull request removes 26 and adds 47 tests. Note that renamed tests count towards both.

dev.restate.sdktesting.tests.AwakeableIngressEndpointTest ‑ completeWithFailure(Client)
dev.restate.sdktesting.tests.AwakeableIngressEndpointTest ‑ completeWithSuccess(Client)
dev.restate.sdktesting.tests.BackwardCompatibilityTest$NewVersion ‑ completeAwakeable(Client)
dev.restate.sdktesting.tests.BackwardCompatibilityTest$NewVersion ‑ completeRetryableOperation(Client)
dev.restate.sdktesting.tests.BackwardCompatibilityTest$NewVersion ‑ proxyCallShouldBeDone(Client)
dev.restate.sdktesting.tests.BackwardCompatibilityTest$NewVersion ‑ proxyOneWayCallShouldBeDone(Client)
dev.restate.sdktesting.tests.BackwardCompatibilityTest$OldVersion ‑ createAwakeable(Client)
dev.restate.sdktesting.tests.BackwardCompatibilityTest$OldVersion ‑ startOneWayProxyCall(Client)
dev.restate.sdktesting.tests.BackwardCompatibilityTest$OldVersion ‑ startProxyCall(Client)
dev.restate.sdktesting.tests.BackwardCompatibilityTest$OldVersion ‑ startRetryableOperation(Client)
…

dev.restate.sdktesting.tests.CallOrdering ‑ ordering(boolean[], Client)[1]
dev.restate.sdktesting.tests.CallOrdering ‑ ordering(boolean[], Client)[2]
dev.restate.sdktesting.tests.CallOrdering ‑ ordering(boolean[], Client)[3]
dev.restate.sdktesting.tests.Cancellation ‑ cancelFromAdminAPI(BlockingOperation, Client, URI)[1]
dev.restate.sdktesting.tests.Cancellation ‑ cancelFromAdminAPI(BlockingOperation, Client, URI)[2]
dev.restate.sdktesting.tests.Cancellation ‑ cancelFromAdminAPI(BlockingOperation, Client, URI)[3]
dev.restate.sdktesting.tests.Cancellation ‑ cancelFromContext(BlockingOperation, Client)[1]
dev.restate.sdktesting.tests.Cancellation ‑ cancelFromContext(BlockingOperation, Client)[2]
dev.restate.sdktesting.tests.Cancellation ‑ cancelFromContext(BlockingOperation, Client)[3]
dev.restate.sdktesting.tests.Combinators ‑ awakeableOrTimeoutUsingAwakeableTimeoutCommand(Client)
…

♻️ This comment has been updated with latest results.