feat: the mock can drop and invert packets, and carries a noisy channel

--drop-every/--drop-burst withhold datagrams the counter and stream clock
have already advanced over, and --reorder-every inverts a pair on the
wire; both are deterministic for the reason --glitch-at is, since a probe
asserts the spacing of what it finds and a random count is not a figure.
A `noise` signal gives the dense-trace and intensity-floor paths a input
that is not piecewise smooth.

verification.md records the measured readings, including the one that is
a finding rather than a baseline: a burst past kGlitchPackets (256) is
read as a producer clock glitch, each bridge advances the anchor by one
packet span against a timeline 300 ahead, and four consecutive failures
then throw the stream onto the arrival clock.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

tools/mock_udpstreamer.py

@@ -21,6 +21,8 @@ import socket
 import struct
 import time
 
+import numpy as np
+
 MAGIC = 0x53504455
 PKT_DATA, PKT_CONFIG, PKT_ACK, PKT_CONNECT, PKT_DISCONNECT = 0, 1, 2, 3, 4
 T_FLOAT64 = 9
@@ -61,6 +63,7 @@ def build_config(rate, time_mode=TIME_FULL_ARRAY, unit="V"):
         signal_desc("sine", rate=rate, time_mode=time_mode, time_idx=0, unit=unit),
         signal_desc("square", rate=rate, time_mode=time_mode, time_idx=0, unit=unit),
         signal_desc("ramp", rate=rate, time_mode=time_mode, time_idx=0, unit=unit),
+        signal_desc("noise", rate=rate, time_mode=time_mode, time_idx=0, unit=unit),
     ]
     payload = struct.pack("<I", len(sigs)) + b"".join(sigs)
     payload += bytes([PUBLISH_ACCUMULATE])
@@ -104,6 +107,9 @@ def build_data(
         1.0 if math.sin(2.0 * math.pi * freq * 0.5 * t) >= 0.0 else -1.0 for t in stamps
     ]
     ramp = [(freq * t) % 1.0 * 2.0 - 1.0 for t in stamps]
+    noisy = np.sin(100 * math.pi * np.array(stamps) + 0.5) + np.random.normal(
+        0, 0.5, batch
+    )
     # A glitch is an *inversion* of the square for `glitch_width` seconds, repeated once per
     # square period (2/freq -- the period of sin(2*pi*freq*0.5*t)). Inverting rather than
     # forcing a level makes it a narrow pulse whatever the square's polarity is, and repeating
@@ -137,7 +143,7 @@ def build_data(
         [wire_stamps[0]] * batch if time_mode == TIME_FIRST_SAMPLE else wire_stamps
     )
     payload = struct.pack("<QI", int(t0 * 1e9), batch)
-    for arr in (wire_time, sine, square, ramp):
+    for arr in (wire_time, sine, square, ramp, noisy):
         payload += struct.pack("<%dd" % batch, *arr)
     return header(PKT_DATA, counter, len(payload)) + payload
 
@@ -234,6 +240,44 @@ def main():
         "opposite side of zero, so --quiet-at is its last zero crossing: the shape a "
         "dropout trigger is read against. -1 = never",
     )
+    ap.add_argument(
+        "--drop-every",
+        type=int,
+        default=0,
+        help="withhold every Nth DATA packet from the wire -- packet loss. Deterministic "
+        "rather than a probability for the reason --glitch-at gives: a probe asserts the "
+        "spacing of the events it finds, and a random count is not a figure. The counter "
+        "and the stream clock advance over a withheld packet, which is the whole point: "
+        "the receiver orders the stream by the counter alone, so that is the only thing "
+        "that tells it a packet existed. 0 = never",
+    )
+    ap.add_argument(
+        "--drop-burst",
+        type=int,
+        default=1,
+        help="consecutive packets each --drop-every event withholds: one models a lost "
+        "datagram, a long run models a link that went away. The run length is what the "
+        "ingest's own glitch guard is sized against, since it bridges a packet whose "
+        "stamps jump more than 256 packet spans ahead. Measured at 1000 packets a second, "
+        "a burst of 200 is counted as loss and nothing else; one of 300 is read as a clock "
+        "glitch instead, and because each bridge advances the anchor by one packet span "
+        "while the true timeline is 300 ahead, the four following packets fail too and the "
+        "stream falls back to arrival timing. Keep a burst under the 2 s silence timeout "
+        "or the client reconnects and the mock restarts its clock, which is a different "
+        "test",
+    )
+    ap.add_argument(
+        "--reorder-every",
+        type=int,
+        default=0,
+        help="hold every Nth packet back and send it after the next one that goes out -- "
+        "an inversion on the wire. The receiver takes the later counter first, so the held "
+        "packet no longer advances the counter and is stale by definition: one inversion "
+        "costs one counter gap, one stale packet, and that packet's samples for good. A "
+        "--drop-every that swallows the packet after a held one deepens the inversion "
+        "rather than cancelling it, since the held packet waits for the next packet "
+        "actually sent. 0 = never",
+    )
     args = ap.parse_args()
 
     dt = args.stream_scale / args.rate  # stream seconds per sample
@@ -265,6 +309,12 @@ def main():
     t_stream = 0.0
     started = None
     next_send = time.monotonic()
+    held = None  # (counter, datagram) built but withheld, for --reorder-every
+    drop_left = 0
+    dropped = 0
+    inverted = 0
+    noted_drop = False
+    noted_reorder = False
 
     while True:
         try:
@@ -278,6 +328,8 @@ def main():
                     t_stream = 0.0
                     started = time.monotonic()
                     next_send = started
+                    held = None
+                    drop_left = 0
                     sock.sendto(build_config(declared, time_mode, args.unit), peer)
                     print(
                         "mock udpstreamer: CONNECT from %s:%d, CONFIG sent" % peer,
@@ -306,7 +358,11 @@ def main():
         if args.clock_jitter > 0.0:
             pdt = dt * (1.0 + random.uniform(-args.clock_jitter, args.clock_jitter))
         spike = 0.0
-        if args.time_glitch_at >= 0.0 and not glitched and t_stream >= args.time_glitch_at:
+        if (
+            args.time_glitch_at >= 0.0
+            and not glitched
+            and t_stream >= args.time_glitch_at
+        ):
             spike = args.time_glitch_by
             glitched = True
             print(
@@ -314,22 +370,58 @@ def main():
                 % (t_stream, spike),
                 flush=True,
             )
-        sock.sendto(
-            build_data(
-                counter,
-                t_stream,
-                pdt,
-                args.batch,
-                args.freq,
-                time_mode,
-                args.noise,
-                spike,
-                args.glitch_at,
-                args.glitch_width,
-                args.quiet_at,
-            ),
-            peer,
+        pkt = build_data(
+            counter,
+            t_stream,
+            pdt,
+            args.batch,
+            args.freq,
+            time_mode,
+            args.noise,
+            spike,
+            args.glitch_at,
+            args.glitch_width,
+            args.quiet_at,
         )
+        burst = max(1, args.drop_burst)
+        if args.drop_every > 0 and drop_left == 0 and counter % args.drop_every == 0:
+            drop_left = burst
+        if drop_left > 0:
+            drop_left -= 1
+            dropped += 1
+            if not noted_drop:
+                noted_drop = True
+                print(
+                    "mock udpstreamer: dropping packet %d and the %d after it"
+                    % (counter, burst - 1),
+                    flush=True,
+                )
+        elif (
+            args.reorder_every > 0
+            and held is None
+            and counter % args.reorder_every == 0
+        ):
+            held = (counter, pkt)
+        else:
+            sock.sendto(pkt, peer)
+            if held is not None:
+                # Out of order on the wire now: the receiver has already taken this
+                # packet's counter, so the held one no longer advances it and is stale by
+                # definition -- one counter gap for the packet that overtook it, one stale
+                # packet for the one that lost, and the loser's samples are gone.
+                sock.sendto(held[1], peer)
+                inverted += 1
+                if not noted_reorder:
+                    noted_reorder = True
+                    print(
+                        "mock udpstreamer: sent packet %d before %d"
+                        % (counter, held[0]),
+                        flush=True,
+                    )
+                held = None
+        # The counter and the stream clock advance whatever became of the datagram. That
+        # is what makes a withheld packet a *lost* one and not a slower producer: nothing
+        # in a payload says which packet it is, only the header's counter does.
         counter += 1
         t_stream += args.batch * pdt
         next_send += period
@@ -338,7 +430,11 @@ def main():
             and started is not None
             and now - started > args.duration
         ):
-            print("mock udpstreamer: done (%d packets)" % counter, flush=True)
+            print(
+                "mock udpstreamer: done (%d packets, %d dropped, %d inverted)"
+                % (counter, dropped, inverted),
+                flush=True,
+            )
             return