From 932617814a84ad003cc83643f34dc994e1291918 Mon Sep 17 00:00:00 2001 From: rob Date: Sun, 12 Jul 2026 02:40:58 +0000 Subject: [PATCH] accesslog-shipper: emit __shipstat__ sent-line checkpoints for splits reconciliation The splits completeness floor is TIME-based (trust hours complete since shipper connect) and blind to event-loss-by-rotation (2026-07-10: raw_events rotated every 1-2min, cruncher lost ~85% of us-32 events, floor still called the hours complete). Shipper now injects {"__shipstat__":1,"sess":..,"sent":N,"ts":..} every 1000 lines or 30s through the SAME tcp pipe. Cruncher diffs shipper-sent vs DB-ingested per gateway/hour -> a reconciliation-based floor that catches the loss a clock cannot. sess=connection-start epoch so resets on reconnect are detectable. Logic in a mounted .awk file (not inline) to dodge sh/awk quoting. Verified on busybox:1.36 (systime/fflush OK). NOT deployed until claude@'s cruncher filters __shipstat__ from event parsing - coordinating via SIP. Co-Authored-By: Claude Fable 5 --- accesslog-shipper.awk | 18 ++++++++++++++++++ drpc.yml | 7 ++++++- 2 files changed, 24 insertions(+), 1 deletion(-) create mode 100644 accesslog-shipper.awk diff --git a/accesslog-shipper.awk b/accesslog-shipper.awk new file mode 100644 index 00000000..f34ed95f --- /dev/null +++ b/accesslog-shipper.awk @@ -0,0 +1,18 @@ +# Streams dshackle access-log lines unchanged, and every 1000 lines OR 30s +# injects a __shipstat__ checkpoint carrying the cumulative sent-count for this +# connection. The splits cruncher (claude host) filters these OUT of event +# processing and uses them to reconcile shipped-vs-ingested per gateway per hour +# — the completeness signal the time-based floor cannot see (event-loss-by- +# rotation, 2026-07-10). `sess` = connection-start epoch so the cruncher detects +# counter resets on shipper reconnect (retry loop resumes at file END). +BEGIN { sess = systime(); last = sess; n = 0 } +{ + n++ + print + now = systime() + if (n % 1000 == 0 || now - last >= 30) { + printf "{\"__shipstat__\":1,\"sess\":%d,\"sent\":%d,\"ts\":%d}\n", sess, n, now + fflush() + last = now + } +} diff --git a/drpc.yml b/drpc.yml index 336b06fb..1cfeafc2 100644 --- a/drpc.yml +++ b/drpc.yml @@ -11,9 +11,14 @@ services: restart: unless-stopped volumes: - ./dshackle_logs:/logs:ro + # sent-line counter: injects __shipstat__ checkpoints into the stream so + # the listener cruncher can reconcile shipped-vs-ingested per gateway/hour + # (fixes the time-based completeness floor being blind to event-loss-by- + # rotation, 2026-07-10). Cruncher MUST filter __shipstat__ from events. + - ./accesslog-shipper.awk:/shipper.awk:ro command: >- sh -c 'while true; do - tail -F /logs/access_log.jsonl 2>/dev/null | nc ${SPLITS_LISTENER_HOST:-78.47.193.225} ${SPLITS_LISTENER_PORT:-9102}; + tail -F /logs/access_log.jsonl 2>/dev/null | awk -f /shipper.awk | nc ${SPLITS_LISTENER_HOST:-78.47.193.225} ${SPLITS_LISTENER_PORT:-9102}; echo shipper lost listener connection, retrying in 10s; sleep 10; done'