156 lines
6.3 KiB
Python
156 lines
6.3 KiB
Python
"""Two-sided clock evidence and recoverable owner-preserving admission."""
|
|
|
|
import threading
|
|
from dataclasses import replace
|
|
|
|
import pytest
|
|
from test_perception_streaming_grpc import identity
|
|
|
|
from k1link.perception.streaming_clock import (
|
|
ClockMappingError,
|
|
ClockProbe,
|
|
ClockReceipt,
|
|
ClockWindow,
|
|
)
|
|
from k1link.perception.streaming_continuity import StreamSuspended
|
|
from k1link.perception.streaming_lifecycle import StreamingLifecycle
|
|
from k1link.perception.streaming_queue import StreamMailbox
|
|
from k1link.perception.streaming_source_clock import SourceAnchor, SourceClockMonitor
|
|
|
|
|
|
def test_responder_warms_without_source_admission_and_rejects_wide_start():
|
|
monitor = SourceClockMonitor("worker", 10**12)
|
|
wide = sample(outbound=20_000_000, inbound=20_000_000)
|
|
now = wide.local_receive_ns + 10**12 + 1_000_000
|
|
anchor = SourceAnchor(10**12, wide.local_receive_ns + 1_000_000_000)
|
|
response = monitor.observe(wide, now, anchor, False)
|
|
assert response.anchor is monitor.anchor is None and response.bounds.samples == 1
|
|
with pytest.raises(StreamSuspended):
|
|
monitor.observed(now)
|
|
narrow = sample(2)
|
|
now = narrow.local_receive_ns + 10**12 + 1_000_000
|
|
response = monitor.observe(narrow, now, None, False)
|
|
assert response.anchor is None and response.bounds.samples == 2
|
|
response.bounds.require(now)
|
|
with pytest.raises(StreamSuspended):
|
|
monitor.observed(now) # Valid clock is not permission to start without an anchor.
|
|
third = sample(3)
|
|
now = third.local_receive_ns + 10**12 + 1_000_000
|
|
assert monitor.observe(third, now, anchor, False).anchor == anchor
|
|
assert monitor.observed(now)[1] < 5_000_000
|
|
|
|
|
|
def test_end_before_start_is_terminal_without_inventing_a_source_timeline():
|
|
monitor = SourceClockMonitor("worker", 10**12)
|
|
first = sample()
|
|
state = monitor.observe(first, first.local_receive_ns + 10**12 + 1_000_000, None, True)
|
|
assert state.ended and state.anchor is None
|
|
with pytest.raises(StreamSuspended):
|
|
monitor.observed(state.bounds.measured_at_ns)
|
|
|
|
|
|
def sample(number=1, outbound=1_000_000, inbound=2_000_000, offset=10**12):
|
|
t1 = 10**16 + number * 100_000_000
|
|
return ClockProbe(
|
|
"source",
|
|
"worker",
|
|
str(number),
|
|
t1,
|
|
t1 + outbound + offset,
|
|
t1 + outbound + offset + 10_000,
|
|
t1 + outbound + inbound + 10_000,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("outbound,inbound", [(1, 9_000_000), (9_000_000, 1), (1, 1)])
|
|
@pytest.mark.parametrize("offset", [-(10**12), 0, 10**12])
|
|
def test_responder_interval_contains_true_reverse_offset(outbound, inbound, offset):
|
|
probe = sample(outbound=outbound, inbound=inbound, offset=offset)
|
|
receipt = ClockReceipt(probe, probe.local_receive_ns + offset + 1_000_000)
|
|
window = ClockWindow("worker", "source", rate_ppm=500, timestamp_error_ns=50_000)
|
|
bounds = window.add(receipt)
|
|
lower, upper = bounds.offset_at(receipt.acknowledged_ns)
|
|
assert lower <= -offset <= upper
|
|
assert bounds.uncertainty_ns(receipt.acknowledged_ns + 1_000_000) > bounds.uncertainty_ns(
|
|
receipt.acknowledged_ns
|
|
)
|
|
|
|
|
|
def test_source_timeline_immutable_expiring_and_observation_bounds_conservative():
|
|
first = sample()
|
|
anchor = SourceAnchor(10**12, first.local_receive_ns + 2_000_000_000)
|
|
monitor = SourceClockMonitor("worker", anchor.source_zero_ns)
|
|
with pytest.raises(StreamSuspended):
|
|
monitor.observed(first.remote_send_ns)
|
|
now = first.local_receive_ns + 10**12 + 1_000_000
|
|
monitor.observe(first, now, anchor, False)
|
|
midpoint, uncertainty = monitor.observed(now)
|
|
actual = anchor.source_zero_ns + (now - 10**12) - anchor.local_zero_ns
|
|
assert midpoint - uncertainty <= actual <= monitor.source_now(now)
|
|
with pytest.raises(StreamSuspended):
|
|
monitor.observed(now + 2_000_000_000)
|
|
second = sample(30)
|
|
later = second.local_receive_ns + 10**12 + 1_000_000
|
|
with pytest.raises(ValueError, match="immutable"):
|
|
monitor.observe(
|
|
second, later, replace(anchor, local_zero_ns=anchor.local_zero_ns + 1), False
|
|
)
|
|
monitor.observe(second, later, anchor, True)
|
|
assert monitor.ended
|
|
with pytest.raises(ValueError, match="ended"):
|
|
monitor.observe(sample(31), later + 100_000_000, anchor, False)
|
|
|
|
|
|
def test_bad_ack_order_expired_jump_and_foreign_clock():
|
|
first = sample()
|
|
for stamp in (first.remote_send_ns - 1, first.remote_receive_ns + 500_000_001):
|
|
with pytest.raises(ClockMappingError):
|
|
ClockReceipt(first, stamp)
|
|
monitor = SourceClockMonitor("worker", 10**12)
|
|
anchor = SourceAnchor(10**12, first.local_receive_ns + 2_000_000_000)
|
|
monitor.observe(first, first.local_receive_ns + 10**12 + 1_000_000, anchor, False)
|
|
jumped = sample(2, offset=10**12 + 100_000_000)
|
|
with pytest.raises(ClockMappingError, match="envelope"):
|
|
monitor.observe(jumped, jumped.local_receive_ns + 10**12 + 101_000_000, anchor, False)
|
|
with pytest.raises(StreamSuspended):
|
|
monitor.observed(jumped.local_receive_ns + 10**12 + 101_000_000)
|
|
|
|
|
|
def test_clock_wait_does_not_stop_owner_or_reopen_mailbox_prematurely(tmp_path):
|
|
healthy = [True]
|
|
|
|
def source_now():
|
|
if not healthy[0]:
|
|
raise StreamSuspended("synthetic expired clock mapping")
|
|
return 1_000_000_000
|
|
|
|
run = StreamingLifecycle(
|
|
identity(),
|
|
tmp_path,
|
|
StreamMailbox(),
|
|
threading.Event(),
|
|
clock_ns=lambda: 1_000_000_000,
|
|
recover_input=True,
|
|
source_clock_ns=source_now,
|
|
)
|
|
run.ready()
|
|
try:
|
|
healthy[0] = False
|
|
with pytest.raises(StreamSuspended):
|
|
run.check_input(identity())
|
|
assert run.continuity.reason == "source-clock" and not run.stop_event.is_set()
|
|
run.renew(identity())
|
|
with pytest.raises(StreamSuspended):
|
|
run.begin_input(identity())
|
|
assert run.mailbox.epoch_drained and run.continuity.phase == "waiting"
|
|
healthy[0] = True
|
|
epoch = run.begin_input(identity())
|
|
assert epoch != identity() and epoch.lease_generation == 1
|
|
assert run.continuity.phase == "synchronizing"
|
|
healthy[0] = False
|
|
with pytest.raises(StreamSuspended, match="obsolete"):
|
|
run.check_input(identity())
|
|
assert run.continuity.phase == "synchronizing" # Stale peer cannot pause new epoch.
|
|
finally:
|
|
assert run.close()
|