Subscription requests now take priority over notifications

[steveluscher]

Problem

When there are a ton of notifications ready to send out over the RPC subscriptions server, they can end up starving the request processor. This means that time-sensitive requests like PING and new subscription requests don't get through.

Summary of Changes

Always process waiting requests before sending out ready notifications.

Test plan

Add a sleep to let a ton of notifications pile up in the broadcast sender.

  select! {
      result = &mut receive_future => match result {
          Ok(_) => {
              break;
          },
          Err(soketto::connection::Error::Closed) => return Ok(()),
          Err(err) => return Err(err.into()),
      },
      result = broadcast_receiver.recv() => {
          // In both possible error cases (closed or lagged) we disconnect the client.
          if let Some(json) = broadcast_handler.handle(result?)? {
              sender.send_text(&*json).await?;
          }
+         sleep(Duration::from_secs(6)).await;
      },
      _ = &mut tripwire => {
          warn!("disconnecting websocket client: shutting down");
          return Ok(())
      },
  }

Send pings from the client.

import ws from "ws";

const w = new ws("ws://127.0.0.1:8900");

w.on("message", function (message) {
  console.log("<<<", new TextDecoder().decode(message));
});

// When connected, send a PING
w.on("open", function open() {
  console.log("Connected to WebSocket server.");
  w.send('{"id":1,"jsonrpc":"2.0","method":"blockSubscribe","params":["all"]}');
  console.log("Subscribed to blocks");
  setInterval(() => {
    w.ping();
    console.log("Sent PING.");
  }, 5000);
});

// Optional: Listen for PONG responses from the server
w.on("pong", function () {
  console.log("PONG received.");
});

Observe that they always get processed before the next notification gets sent.

Fixes #7022

[codecov-commenter]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 83.2%. Comparing base (55f4f8d) to head (eb1a15f).
Report is 2 commits behind head on master.

Additional details and impacted files

@@           Coverage Diff           @@
##           master    #7031   +/-   ##
=======================================
  Coverage    83.2%    83.2%           
=======================================
  Files         853      853           
  Lines      374866   374866           
=======================================
+ Hits       312054   312083   +29     
+ Misses      62812    62783   -29     

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[lijunwangs]

LGTM. Thanks for addressing this problem!

[lijunwangs]

@steveluscher @joncinque , The issue is important for Chainlink's customers, can we consider back port this to the next release branch?

[mergify]

Backports to the beta branch are to be avoided unless absolutely necessary for fixing bugs, security issues, and perf regressions. Changes intended for backport should be structured such that a minimum effective diff can be committed separately from any refactoring, plumbing, cleanup, etc that are not strictly necessary to achieve the goal. Any of the latter should go only into master and ride the normal stabilization schedule. Exceptions include CI/metrics changes, CLI improvements and documentation updates on a case by case basis.