fix(rpc): plug OOO engine stack context use-after-free

[jfeng18]

Problem

OooEngine::m_map stores raw pointers to callers' stack-allocated OutOfOrderContext. After the receiver takes a pointer out of the map, it may yield in do_collect(); a timed-out caller can meanwhile return and destroy the stack context, leaving the receiver dereferencing freed stack memory.

Fix

Add an m_collecting handshake: the receiver marks the context in flight; timed-out callers, the do_issue() failure path and the wait_completion() entry window wait for the hand-back instead of returning. The !th branch becomes a legal timed-out-caller state.

Tests

New test-ooo-timeout with 4 real-path cases covering all three windows plus a concurrency stress; all pass locally under ASan. Replaces the old white-box test error_thread_become_NULL (#define private public) whose injected state is now a legal real path.

Fixes #1291

[lihuiba]

After the receiver takes a pointer out of the map, it may yield in do_collect(); a timed-out caller can meanwhile return and destroy the stack context, leaving the receiver dereferencing freed stack memory.

Maybe we can detect such case and let the timed-out caller simply wait for the former one to finish?

[jfeng18]

Yes, that's exactly what this patch does: the receiver marks the context it has taken out of the map in m_collecting, and a timed-out caller that finds itself being collected waits in wait_if_collecting() until the hand-back completes, then returns the collected result instead of destroying its stack frame. The same handshake also covers the do_issue() failure path and the wait_completion() entry window.

[Coldwings]

LGTM

[lihuiba]

how about using a sleep loop if such case occurs:

int wait_completion(OutOfOrderContext& args) //recieving work
{
      ...
      auto ret = m_wait.wait(args.phaselock, args.timeout);
      // Check if collected
      if (args.phase == OooPhase::COLLECTED &&
          args.th == CURRENT) {
          return args.ret;
      }
      if (ret == -1) {
          // or just timed out
          {
              SCOPED_LOCK(m_mutex_map);
              m_map.erase(args.tag);
              m_cond_collected.notify_one();
          }
          // wait if args is in use
          while (args.in_use) thread_usleep(100);
          LOG_ERROR_RETURN(ETIMEDOUT, -1, "waiting for completion timeout");
      }