Fix IB idle timer arithmetic for packet nanoseconds
Override `NonShittyWrapper._setTimer()` to compute packet silence in seconds from integer epoch nanoseconds, avoiding upstream's `datetime` subtraction error after `tcpDataArrived()`. Delegate pre-traffic datetime and sentinel handling upstream. Preserve the packet timestamp on expiry and clear the timer before `timeoutEvent` emission so listeners can safely rearm it. Add three offline regressions under `tests/providers/` for rescheduling, single expiry, datetime and sentinel states, disabling, and reentrant listener rearming. Document the wrapper overrides on the class. This fixes an explicitly armed packet-idle timer; heartbeat, default arming, and reconnect behavior are outside this change. Prompt-IO: ai/prompt-io/opencode/20260924_ib_idle_timer.md (this patch was generated in some part by `opencode` using `gpt-6-astra` (`openai`))ib_connectivity_harden
parent
efa72980f8
commit
d126302fb9
|
|
@ -0,0 +1,42 @@
|
||||||
|
---
|
||||||
|
model: openai/gpt-6-astra
|
||||||
|
service: opencode
|
||||||
|
scope: code
|
||||||
|
substantive: true
|
||||||
|
git_ref: efa72980
|
||||||
|
raw_file: 20260924_ib_idle_timer.raw.md
|
||||||
|
---
|
||||||
|
|
||||||
|
## Prompt
|
||||||
|
|
||||||
|
The human reported silent IB roaming and requested an immediate fix
|
||||||
|
after investigation found the custom nanosecond timestamps incompatible
|
||||||
|
with ib_async's datetime-based idle timeout arithmetic.
|
||||||
|
|
||||||
|
## Response summary
|
||||||
|
|
||||||
|
Override NonShittyWrapper._setTimer to accept integer nanosecond packet
|
||||||
|
timestamps while delegating datetime/sentinel states upstream. Keep the
|
||||||
|
packet timestamp unchanged on integer-path expiry, clear the timer
|
||||||
|
before timeoutEvent emission, and preserve reentrant listener rearming.
|
||||||
|
No default heartbeat, reconnect, or idle-timeout arming is introduced.
|
||||||
|
|
||||||
|
## Files changed
|
||||||
|
|
||||||
|
- piker/brokers/ib/api.py: compatible idle timer arithmetic/lifecycle.
|
||||||
|
- tests/test_ib_wrapper_timeout.py: three deterministic offline tests.
|
||||||
|
|
||||||
|
## Verification
|
||||||
|
|
||||||
|
Nineteen tests passed: new wrapper-timeout regressions and existing IB
|
||||||
|
history/method-proxy tests. Restoring the upstream method in memory
|
||||||
|
reproduced datetime-minus-int TypeError. No IB service or live network
|
||||||
|
was contacted; installed dependencies and original user work were
|
||||||
|
unchanged. The isolated worktree index remains unchanged.
|
||||||
|
|
||||||
|
## Human contributions
|
||||||
|
|
||||||
|
The human supplied rfkill/suspend/roam evidence, requested a session
|
||||||
|
health design, and prioritized fixing the discovered timestamp mismatch.
|
||||||
|
The implementation deliberately preserves their nanosecond feed schema.
|
||||||
|
Acceptance and live validation remain pending.
|
||||||
|
|
@ -0,0 +1,17 @@
|
||||||
|
---
|
||||||
|
model: openai/gpt-6-astra
|
||||||
|
service: opencode
|
||||||
|
git_ref: efa72980
|
||||||
|
diff_cmd: git diff HEAD -- piker/brokers/ib/api.py tests/test_ib_wrapper_timeout.py
|
||||||
|
---
|
||||||
|
|
||||||
|
> git diff HEAD -- piker/brokers/ib/api.py tests/test_ib_wrapper_timeout.py
|
||||||
|
|
||||||
|
NonShittyWrapper._setTimer accepts upstream datetime state and piker's
|
||||||
|
integer packet-nanosecond state. Expiry clears its handle and timeout
|
||||||
|
before notifying listeners, without replacing packet timestamps.
|
||||||
|
Tests cover traffic rescheduling, exactly-once expiry, pretraffic and
|
||||||
|
sentinel handling, disabling, and callback rearming.
|
||||||
|
|
||||||
|
Verification: 19 passed, including existing IB history/proxy tests.
|
||||||
|
Wall-clock packet silence remains distinct from session health.
|
||||||
|
|
@ -82,6 +82,7 @@ from ib_async import (
|
||||||
Wrapper,
|
Wrapper,
|
||||||
RequestError,
|
RequestError,
|
||||||
)
|
)
|
||||||
|
from ib_async.util import getLoop
|
||||||
import numpy as np
|
import numpy as np
|
||||||
|
|
||||||
# TODO: in hindsight, probably all imports should be
|
# TODO: in hindsight, probably all imports should be
|
||||||
|
|
@ -147,9 +148,45 @@ _bar_sizes = {
|
||||||
|
|
||||||
_show_wap_in_history: bool = False
|
_show_wap_in_history: bool = False
|
||||||
|
|
||||||
# overrides to sidestep pretty questionable design decisions in
|
|
||||||
# ``ib_async``:
|
|
||||||
class NonShittyWrapper(Wrapper):
|
class NonShittyWrapper(Wrapper):
|
||||||
|
'''
|
||||||
|
Overrides to sidestep pretty questionable design decisions in
|
||||||
|
``ib_async``.
|
||||||
|
|
||||||
|
'''
|
||||||
|
|
||||||
|
def _setTimer(self, delay: float = 0):
|
||||||
|
'''
|
||||||
|
Accept packet epoch nanoseconds alongside upstream datetimes.
|
||||||
|
|
||||||
|
This measures wall-clock packet silence, not session health.
|
||||||
|
|
||||||
|
'''
|
||||||
|
if isinstance(self.lastTime, datetime):
|
||||||
|
return super()._setTimer(delay)
|
||||||
|
|
||||||
|
if not self._timeout:
|
||||||
|
return
|
||||||
|
|
||||||
|
diff = (time.time_ns() - self.lastTime) / 1_000_000_000
|
||||||
|
if not delay:
|
||||||
|
delay = self._timeout - diff
|
||||||
|
|
||||||
|
if delay > 0:
|
||||||
|
self._timeoutHandle = getLoop().call_later(
|
||||||
|
delay,
|
||||||
|
self._setTimer,
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
self._logger.debug('Timeout')
|
||||||
|
# Match Wrapper.setTimeout(0)'s timer teardown, but keep
|
||||||
|
# lastTime in ns for quote/fill and timeoutEvent readers.
|
||||||
|
if self._timeoutHandle:
|
||||||
|
self._timeoutHandle.cancel()
|
||||||
|
self._timeoutHandle = None
|
||||||
|
self._timeout = 0
|
||||||
|
self.ib.timeoutEvent.emit(diff)
|
||||||
|
|
||||||
def tcpDataArrived(self):
|
def tcpDataArrived(self):
|
||||||
"""Override time stamps to be floats for now.
|
"""Override time stamps to be floats for now.
|
||||||
"""
|
"""
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,4 @@
|
||||||
|
'''
|
||||||
|
Provider-specific regression tests.
|
||||||
|
|
||||||
|
'''
|
||||||
|
|
@ -0,0 +1,183 @@
|
||||||
|
'''
|
||||||
|
Offline packet-idle timer regressions using real eventkit events.
|
||||||
|
|
||||||
|
'''
|
||||||
|
import asyncio
|
||||||
|
from datetime import datetime
|
||||||
|
from types import SimpleNamespace
|
||||||
|
|
||||||
|
from eventkit import Event
|
||||||
|
from ib_async import wrapper as upstream
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from piker.brokers.ib import api
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def timer(monkeypatch):
|
||||||
|
'''
|
||||||
|
Drive real asyncio TimerHandles with a deterministic wall clock.
|
||||||
|
|
||||||
|
'''
|
||||||
|
class ClockLoop:
|
||||||
|
now = 1_800_000_000.0
|
||||||
|
|
||||||
|
def __init__(self):
|
||||||
|
self.handles = []
|
||||||
|
|
||||||
|
def get_debug(self):
|
||||||
|
return False
|
||||||
|
|
||||||
|
def _timer_handle_cancelled(self, handle):
|
||||||
|
pass
|
||||||
|
|
||||||
|
def call_exception_handler(self, context):
|
||||||
|
raise context['exception']
|
||||||
|
|
||||||
|
def call_later(self, delay, callback):
|
||||||
|
handle = asyncio.TimerHandle(
|
||||||
|
self.now + delay, callback, (), self,
|
||||||
|
)
|
||||||
|
self.handles.append(handle)
|
||||||
|
return handle
|
||||||
|
|
||||||
|
def advance(self, seconds):
|
||||||
|
target = self.now + seconds
|
||||||
|
while self.handles:
|
||||||
|
handle = min(self.handles, key=lambda h: h.when())
|
||||||
|
if handle.when() > target:
|
||||||
|
break
|
||||||
|
self.handles.remove(handle)
|
||||||
|
self.now = handle.when()
|
||||||
|
if not handle.cancelled():
|
||||||
|
handle._run()
|
||||||
|
self.now = target
|
||||||
|
|
||||||
|
loop = ClockLoop()
|
||||||
|
|
||||||
|
class ClockDateTime(datetime):
|
||||||
|
@classmethod
|
||||||
|
def now(cls, tz=None):
|
||||||
|
return cls.fromtimestamp(loop.now, tz)
|
||||||
|
|
||||||
|
monkeypatch.setattr(upstream, 'datetime', ClockDateTime)
|
||||||
|
monkeypatch.setattr(upstream, 'getLoop', lambda: loop)
|
||||||
|
monkeypatch.setattr(api, 'getLoop', lambda: loop)
|
||||||
|
monkeypatch.setattr(
|
||||||
|
api.time, 'time_ns', lambda: int(loop.now * 1_000_000_000),
|
||||||
|
)
|
||||||
|
ib = SimpleNamespace(timeoutEvent=Event('timeoutEvent'))
|
||||||
|
wrapper = api.NonShittyWrapper(ib)
|
||||||
|
events = []
|
||||||
|
|
||||||
|
def record(diff):
|
||||||
|
events.append((diff, wrapper.lastTime))
|
||||||
|
|
||||||
|
ib.timeoutEvent.connect(record)
|
||||||
|
return loop, wrapper, events
|
||||||
|
|
||||||
|
|
||||||
|
def test_packet_timeout_reschedules_and_emits_once(timer):
|
||||||
|
'''
|
||||||
|
Upstream datetime subtraction raised TypeError after a packet
|
||||||
|
replaced NonShittyWrapper.lastTime with epoch nanoseconds.
|
||||||
|
Arm the real wrapper, deliver packets around the first deadline,
|
||||||
|
and drive its TimerHandles without sleeping or opening sockets.
|
||||||
|
New traffic must postpone expiry by the remaining idle interval;
|
||||||
|
the real eventkit listener must see seconds and the intact ns
|
||||||
|
timestamp exactly once, with the timer already disabled.
|
||||||
|
|
||||||
|
'''
|
||||||
|
loop, wrapper, events = timer
|
||||||
|
wrapper.setTimeout(5)
|
||||||
|
loop.advance(1)
|
||||||
|
wrapper.tcpDataArrived()
|
||||||
|
assert isinstance(wrapper.lastTime, int)
|
||||||
|
loop.advance(3)
|
||||||
|
wrapper.tcpDataArrived()
|
||||||
|
packet_time = wrapper.lastTime
|
||||||
|
loop.advance(1)
|
||||||
|
assert events == []
|
||||||
|
assert wrapper.lastTime == packet_time
|
||||||
|
assert wrapper._timeoutHandle.when() == loop.now + 4
|
||||||
|
loop.advance(4)
|
||||||
|
assert events == [(5.0, packet_time)]
|
||||||
|
assert wrapper._timeoutHandle is None
|
||||||
|
assert wrapper._timeout == 0
|
||||||
|
assert wrapper.lastTime == packet_time
|
||||||
|
wrapper._setTimer()
|
||||||
|
loop.advance(20)
|
||||||
|
assert events == [(5.0, packet_time)]
|
||||||
|
|
||||||
|
|
||||||
|
def test_datetime_timeout_and_sentinel(timer):
|
||||||
|
'''
|
||||||
|
Before any packet, Wrapper.setTimeout uses datetime timestamps.
|
||||||
|
The ns fix must preserve that upstream path and datetime.min's
|
||||||
|
sentinel short circuit. Run an armed pre-traffic timer to expiry,
|
||||||
|
then invoke the sentinel path and assert no callback is queued.
|
||||||
|
|
||||||
|
'''
|
||||||
|
loop, wrapper, events = timer
|
||||||
|
assert isinstance(wrapper.lastTime, datetime)
|
||||||
|
assert wrapper._timeoutHandle is None
|
||||||
|
wrapper.setTimeout(3)
|
||||||
|
loop.advance(3)
|
||||||
|
assert len(events) == 1
|
||||||
|
assert events[0][0] == 3.0
|
||||||
|
assert isinstance(events[0][1], datetime)
|
||||||
|
assert wrapper._timeoutHandle is None
|
||||||
|
wrapper.lastTime = datetime.min
|
||||||
|
wrapper._setTimer(2)
|
||||||
|
loop.advance(10)
|
||||||
|
assert len(events) == 1
|
||||||
|
assert wrapper.lastTime == datetime.min
|
||||||
|
assert wrapper._timeoutHandle is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_rearm_disable_and_reentrant_listener(timer):
|
||||||
|
'''
|
||||||
|
Replacing or disabling a packet-idle timer must cancel its old
|
||||||
|
handle. Expiry must also finish teardown before timeoutEvent
|
||||||
|
listeners rearm it, otherwise the new timer can be lost.
|
||||||
|
Advance cancelled deadlines deterministically, then rearm from
|
||||||
|
a real eventkit listener and deliver another packet. Both arms
|
||||||
|
must emit once, with packet timestamps intact during emission.
|
||||||
|
|
||||||
|
'''
|
||||||
|
loop, wrapper, events = timer
|
||||||
|
wrapper.setTimeout(2)
|
||||||
|
wrapper.tcpDataArrived()
|
||||||
|
old = wrapper._timeoutHandle
|
||||||
|
wrapper.setTimeout(4)
|
||||||
|
assert old.cancelled()
|
||||||
|
replacement = wrapper._timeoutHandle
|
||||||
|
wrapper.tcpDataArrived()
|
||||||
|
wrapper.setTimeout(0)
|
||||||
|
assert replacement.cancelled()
|
||||||
|
assert wrapper._timeoutHandle is None
|
||||||
|
wrapper.tcpDataArrived()
|
||||||
|
loop.advance(10)
|
||||||
|
assert events == []
|
||||||
|
|
||||||
|
observed = []
|
||||||
|
|
||||||
|
def rearm(diff):
|
||||||
|
observed.append((wrapper._timeout, wrapper._timeoutHandle))
|
||||||
|
if len(observed) == 1:
|
||||||
|
wrapper.setTimeout(3)
|
||||||
|
wrapper.tcpDataArrived()
|
||||||
|
|
||||||
|
wrapper.ib.timeoutEvent.connect(rearm)
|
||||||
|
wrapper.setTimeout(2)
|
||||||
|
wrapper.tcpDataArrived()
|
||||||
|
first_packet = wrapper.lastTime
|
||||||
|
loop.advance(2)
|
||||||
|
second_packet = wrapper.lastTime
|
||||||
|
assert wrapper._timeout == 3
|
||||||
|
assert wrapper._timeoutHandle is not None
|
||||||
|
loop.advance(3)
|
||||||
|
loop.advance(10)
|
||||||
|
assert events == [(2.0, first_packet), (3.0, second_packet)]
|
||||||
|
assert observed == [(0, None), (0, None)]
|
||||||
|
assert wrapper.lastTime == second_packet
|
||||||
Loading…
Reference in New Issue