Harden StoppableQueueBlockingRunnable shutdown to not depend solely on the RUNNABLE_STOP wakeup

[Ma77Ball]

The three StoppableQueueBlockingRunnable threads (MainLoop, NetworkSender, PortStorageWriter) block in interruptible_get on a queue.get() with no timeout. By design this get() is indefinitely blocking so the engine only advances when a real message or the RUNNABLE_STOP marker arrives, and the Scala side mirrors this. That design is intentional and should be preserved: the loop must not return to receive() on a quiet queue.

This is a defensive-hardening request, not a report of a reproducible failure. As the code stands today, stop() reliably sets and delivers the RUNNABLE_STOP marker through correctly-locked queues, so no shutdown hang has been observed or reproduced in CI. The goal is to decouple "stop was requested" from "the marker wakeup was delivered," so that a future change (a new stop path that forgets the marker, or a change to the queue's notify logic) cannot silently reintroduce a shutdown hang.

Proposed approach (data path and blocking semantics preserved):

• Add a threading.Event stop flag; stop() sets it in addition to enqueueing the marker.
• interruptible_get polls with a short timeout and treats queue.Empty as "loop and wait again" (continue), so it never returns control to receive() / the handling loop on a timeout. It only returns on a real item, or raises InterruptRunnable when the flag is set. This keeps the indefinite-blocking semantics for the data path intact while ensuring a stop request is honored within one poll interval even if the single marker wakeup were ever missed.
• Thread an optional timeout through Getable.get -> InternalQueue.get -> LinkedBlockingMultiQueue.get (the last via Condition.wait_for). Default timeout=None keeps the existing blocking behavior unchanged; only the stoppable threads opt into polling.

Open question: whether this hardening is worth the added surface area (a timeout on a queue that is intentionally infinite-blocking) given no failure has been reproduced. See discussion in #5326.

[Yicong-Huang]

did this cause real failures? if so do we have CI run or reproducible in a workflow?

[Yicong-Huang]

The get() is designed to be indefinitely blocking, so that our engine state can be sychronized. python and scala side are the same. if we unblock it with a timeout, the engine may get into the next handling loop which may break other assumptions.

It is by design this get() can only be stopped through RUNNABLE_STOP.

[Ma77Ball]

Agreed on both points:

• No reproducer - confirmed, can't trigger a hang on current main.
• Indefinite blocking by design - preserved here: on timeout interruptible_get does continue, never returns to receive(), so the loop still only advances on a real message or RUNNABLE_STOP. The timeout just adds an idle stop-flag re-check.

So it's optional hardening, not a fix. Fine to close unless you think the safety net is worth it.

[Yicong-Huang]

I'd be conservative here: let's not change it unless there is real issue. The current model declares the only way to stop and we don't intend to stop the thread using another way in the near future. And the current model has faithfully served us for years.

I suggest we close this.