Commitf21772f6Recorded6 Jul 2026Repositorycourier

Fix send-path duplication: persist the send dedup across restarts

Message

The courier-spam storm was send-side duplication, not backlog re-delivery: David always saw many copies of the SAME message. courier sends exactly once per send-message tool call (no retry in telegram.sgl, sigil-telegram, or sigil-http; the MCP server guards handlers so an exception can't crash the process). The duplication is cross-process: the sendMessage ack read can block up to 25s, and if the leader's MCP client gives up sooner (short tool timeout, or a /mcp reconnect because the call "looks hung") it SIGKILLs courier and re-issues the same send. Each attempt reaches Telegram before the kill, so the message is delivered once per attempt. The v0.3.7/0.3.8 dedup couldn't stop it: the recent-sends cache was in-memory, wiped on every restart.

Fix (new module (courier dedup)): - Persist the (chat-id,text) dedup to disk (send-dedup.log beside the relay dir; override COURIERSENDDEDUPFILE) so it survives restarts. - Record a send BEFORE delivery, so a SIGKILL mid-send still dedups the leader's retry. Keys are FNV-1a hashes (no plaintext on disk); sliding window default 60s (COURIERSENDDEDUPWINDOW). - Handler never lets a transport error escape or re-fire a send: tg-ack-unconfirmed keeps the key and reports success; any other error means not-delivered (never-sent or rejected), so release the key for a genuine retry and return a clean "message not delivered" string.

Net: exactly-once on success, at-most-once on a mid-flight kill (the storm), at-least-once on definite non-delivery. Biases toward never spamming over never losing a notification.

Verified off-device against a mock (nothing reaches real Telegram): pre-fix control storms to 4 deliveries, fixed stays at 1. Adds repro/repro-send-duplication.sh + send_driver.py, mock send-failure injection, and test/test-send-dedup.sgl. Full suite 69 pass; check/lint clean.

Changed
 repro/.gitignore                              |   4 +++
 repro/evidence-2026-07-06-send-duplication.md |  75 ++++++++++++++++++++++++++++++++++++++++++++++
 repro/mock_telegram.py                        | 239 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 repro/repro-send-duplication.sh               |  89 ++++++++++++++++++++++++++++++++++++++++++++++++++++++
 repro/send_driver.py                          | 229 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 src/courier/dedup.sgl                         | 171 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 src/courier/telegram.sgl                      | 108 ++++++++++++++++++++++++++++++------------------------------------
 test/test-send-dedup.sgl                      | 180 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 8 files changed, 1035 insertions(+), 60 deletions(-)
Diff
repro/.gitignoreadded
@@ -0,0 +1,4 @@
+1
run/
+2
*.out
+3
*.err
+4
__pycache__/
repro/evidence-2026-07-06-send-duplication.mdadded
@@ -0,0 +1,75 @@
+1
# Send-path duplication: off-device evidence (2026-07-06)
+2
+3
All runs go through `repro/mock_telegram.py`; **nothing reached real Telegram**
+4
(courier's SEND path is forced to the mock via `COURIER_TELEGRAM_API_URL`).
+5
+6
## Setup
+7
+8
- `mock_telegram.py` — records every `sendMessage`; can inject send failure
+9
modes (`ok`, `reset`, `delay:N`, `status:N`, `notok`) via `POST /_mode`.
+10
- `send_driver.py` — a minimal MCP client (the "leader") that fires ONE
+11
logical `send-message` and models the harness reaction to a send that does
+12
not return a clean, timely success: a per-tool-call timeout, and on
+13
timeout/death a **kill + respawn + re-issue** of the same send (the
+14
`/mcp`-reconnect + agent-retry loop), up to `--retries` times. All restarts
+15
share one `COURIER_RELAY_DIR`, exactly as the real leader's restarts share
+16
`~/.courier`.
+17
+18
## Root cause
+19
+20
courier sends **exactly once per tool call** (traced: `telegram.sgl` →
+21
`tg-send-message` → `tg-api-call` → `http-post/json`; no retry at any layer,
+22
and the MCP server guards tool handlers so an exception cannot crash the
+23
process). The duplication is **cross-process**: the `sendMessage` ack read can
+24
block up to `*send-request-timeout*` (25s). If the leader's MCP client gives up
+25
sooner — a short tool timeout, or a human/harness `/mcp` because the call
+26
"looks hung" — it SIGKILLs courier and re-issues the send. Each attempt's POST
+27
reaches Telegram before the kill → the same message is delivered N times. The
+28
v0.3.7/0.3.8 `(chat-id,text)` dedup could not stop it: `recent-sends` was
+29
**in-memory**, so a restarted process started with an empty cache. The
+30
"`inv=1` then `--- courier restart ---`" in the live log is this
+31
SIGKILL-on-reconnect, not an internal segfault.
+32
+33
## Before (pre-fix master `4a5ae2b`, in-memory dedup)
+34
+35
```
+36
STORM (delay:8, tool-timeout 3, retries 3): 1 logical send -> 4 deliveries
+37
```
+38
+39
Each of the 4 attempts (gen0..gen3) re-delivered because the fresh process's
+40
in-memory dedup was empty.
+41
+42
## After (fix: persistent record-before-send dedup, `src/courier/dedup.sgl`)
+43
+44
```
+45
baseline ok 1 send -> 1 delivery "Message sent."
+46
reset (ack lost) 1 send -> 1 delivery "Message sent (ack unconfirmed)."
+47
STORM x3 1 send -> 1 delivery (gen0 killed mid-send; gen1 suppressed)
+48
STORM x5 1 send -> 1 delivery
+49
status:500 / notok 1 send, clean error "…(message not delivered)." NO crash, key released
+50
never-sent (dead port) 0 deliveries, clean error, key released (retry allowed)
+51
```
+52
+53
In the storm, gen0 records the dedup key **before** calling `tg-send-message`,
+54
delivers, then is SIGKILLed; gen1 (fresh process, empty memory) reads the same
+55
on-disk key and returns "Message sent." in ~0.1s without re-delivering.
+56
+57
## Reproduce
+58
+59
```
+60
sigil build
+61
# optional pre-fix control:
+62
git worktree add /tmp/courier-control master && (cd /tmp/courier-control && sigil deps install && sigil build)
+63
COURIER_CONTROL_BIN=/tmp/courier-control/build/dev/bin/courier repro/repro-send-duplication.sh
+64
# => ALL CHECKS PASSED (FIXED storms -> 1; CONTROL storm -> 4)
+65
```
+66
+67
## Note on exactly-once vs at-most-once
+68
+69
Telegram's `sendMessage` has no idempotency key, so exactly-once is only
+70
achievable by courier suppressing duplicate deliveries. The fix records a send
+71
**before** attempting delivery and treats a send killed mid-flight as
+72
delivered (at-most-once). A definite non-delivery (never-sent connect failure,
+73
or Telegram rejection) **releases** the key so a genuine retry can go through.
+74
This biases toward "never spam" over "never lose a notification" — the correct
+75
bias for this bot, and the whole point of the saga.
repro/mock_telegram.pyadded
@@ -0,0 +1,239 @@
+1
#!/usr/bin/env python3
+2
"""Mock Telegram Bot API for OFF-DEVICE courier repro.
+3
+4
Models the two semantics that matter for the getUpdates-backlog spam bug:
+5
+6
* getUpdates(offset=N) CONFIRMS (drops) every pending update with
+7
update_id < N, then returns the remaining pending updates. A call
+8
with no offset (or offset 0) confirms nothing and returns the whole
+9
backlog -- exactly how a fresh poller (offset reset to 0) re-fetches
+10
Telegram's ~24h backlog.
+11
* sendMessage records the outbound and returns ok. Nothing here ever
+12
reaches real Telegram; the point is to COUNT would-be sends.
+13
+14
Every request is appended to <logdir>/requests.log so the driver can
+15
assert on re-delivery. Control endpoints (not part of the Telegram API)
+16
let the driver seed and inject updates:
+17
+18
POST /_seed body: {"updates":[{...},...]} replace the backlog
+19
POST /_add body: {"text":"...","sender_id":"..","chat_id":".."} append one update
+20
GET /_stats {"delivered":[ids], "sends":N}
+21
+22
The sentinel-zero (2026-07-02) storm case is exercised by
+23
`repro-sentinel-zero.sh`: `_seed []` gives an EMPTY backlog so a cold start
+24
drains count=0, then `_add` accumulates a backlog before a restart -- the
+25
scenario where persisting offset=0 (pre-fix) causes re-delivery.
+26
+27
Usage: mock_telegram.py <port> <logdir>
+28
"""
+29
import json
+30
import sys
+31
import os
+32
import time
+33
import threading
+34
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+35
+36
PORT = int(sys.argv[1])
+37
LOGDIR = sys.argv[2]
+38
os.makedirs(LOGDIR, exist_ok=True)
+39
REQ_LOG = os.path.join(LOGDIR, "requests.log")
+40
SEND_LOG = os.path.join(LOGDIR, "sends.log")
+41
+42
# ------------------------------------------------------------------
+43
# sendMessage failure-injection (send-path duplication repro).
+44
#
+45
# The default "ok" mode returns a normal Telegram success. The other
+46
# modes model the transport conditions that make courier's send handler
+47
# NOT return a clean, timely success -- the situations under which the
+48
# leader's MCP client would re-issue the send and (pre-fix) re-deliver:
+49
#
+50
# ok normal 200 {ok:true} response
+51
# delay:<sec> sleep <sec> BEFORE responding (models a slow ack read;
+52
# with <sec> > courier's *send-request-timeout* the read
+53
# times out AFTER Telegram already recorded the send)
+54
# reset record the send, then drop the connection with NO HTTP
+55
# response (models a delivered-but-ack-lost read failure)
+56
# status:<code> record the send, respond with an HTTP error status
+57
# notok record the send, respond 200 {ok:false} (API-level error)
+58
#
+59
# Set the initial mode via env SEND_MODE; change at runtime via
+60
# POST /_mode {"mode":"..."}. Every mode still COUNTS the send, because
+61
# in all of them Telegram has actually received (and would deliver) the
+62
# message -- that is the whole point of counting deliveries at the mock.
+63
# ------------------------------------------------------------------
+64
SEND_MODE_DEFAULT = os.environ.get("SEND_MODE", "ok")
+65
+66
LOCK = threading.Lock()
+67
STATE = {
+68
"updates": [], # pending updates (list of dicts with update_id)
+69
"next_id": 1, # next update_id to assign via _add
+70
"getupdates": 0, # count of getUpdates calls
+71
"sends": 0, # count of sendMessage calls
+72
"delivered_ids": [], # update_ids the mock RETURNED to a poller (per call)
+73
"send_mode": SEND_MODE_DEFAULT,
+74
}
+75
+76
+77
def logline(path, msg):
+78
with open(path, "a") as f:
+79
f.write(msg + "\n")
+80
f.flush()
+81
+82
+83
def make_update(update_id, text, sender_id="42", chat_id="1001"):
+84
return {
+85
"update_id": update_id,
+86
"message": {
+87
"message_id": update_id,
+88
"date": 1000000 + update_id,
+89
"text": text,
+90
"from": {"id": int(sender_id), "is_bot": False,
+91
"first_name": "David", "username": "daviwil"},
+92
"chat": {"id": int(chat_id), "type": "private"},
+93
},
+94
}
+95
+96
+97
class Handler(BaseHTTPRequestHandler):
+98
def log_message(self, *a):
+99
pass # silence default stderr logging
+100
+101
def _body(self):
+102
n = int(self.headers.get("Content-Length", 0))
+103
raw = self.rfile.read(n) if n else b""
+104
try:
+105
return json.loads(raw) if raw else {}
+106
except Exception:
+107
return {}
+108
+109
def _reply(self, obj, code=200):
+110
data = json.dumps(obj).encode()
+111
try:
+112
self.send_response(code)
+113
self.send_header("Content-Type", "application/json")
+114
self.send_header("Content-Length", str(len(data)))
+115
self.end_headers()
+116
self.wfile.write(data)
+117
except (BrokenPipeError, ConnectionResetError):
+118
# courier was SIGKILLed mid-send (the storm's restart step) --
+119
# the delivery was still counted; the lost ack is expected.
+120
pass
+121
+122
def do_GET(self):
+123
if self.path == "/_stats":
+124
with LOCK:
+125
self._reply({"delivered": STATE["delivered_ids"],
+126
"sends": STATE["sends"],
+127
"getupdates": STATE["getupdates"],
+128
"pending": [u["update_id"] for u in STATE["updates"]]})
+129
return
+130
self._reply({"ok": False, "description": "not found"}, 404)
+131
+132
def do_POST(self):
+133
body = self._body()
+134
path = self.path
+135
+136
# ---- control endpoints ----
+137
if path == "/_seed":
+138
with LOCK:
+139
STATE["updates"] = list(body.get("updates", []))
+140
STATE["next_id"] = (max([u["update_id"] for u in STATE["updates"]],
+141
default=0) + 1)
+142
STATE["getupdates"] = 0
+143
STATE["sends"] = 0
+144
STATE["delivered_ids"] = []
+145
self._reply({"ok": True})
+146
return
+147
if path == "/_add":
+148
with LOCK:
+149
uid = STATE["next_id"]
+150
STATE["next_id"] += 1
+151
STATE["updates"].append(make_update(
+152
uid, body.get("text", "msg"),
+153
body.get("sender_id", "42"), body.get("chat_id", "1001")))
+154
self._reply({"ok": True, "update_id": uid})
+155
return
+156
if path == "/_mode":
+157
with LOCK:
+158
STATE["send_mode"] = body.get("mode", "ok")
+159
mode = STATE["send_mode"]
+160
logline(REQ_LOG, f"--- send_mode set to {mode!r} ---")
+161
self._reply({"ok": True, "mode": mode})
+162
return
+163
+164
# ---- Telegram Bot API ----
+165
# path looks like /bot<token>/<method>
+166
method = path.rsplit("/", 1)[-1]
+167
+168
if method == "getUpdates":
+169
offset = body.get("offset", 0) or 0
+170
with LOCK:
+171
STATE["getupdates"] += 1
+172
if offset > 0:
+173
# Confirm: drop everything below the offset.
+174
STATE["updates"] = [u for u in STATE["updates"]
+175
if u["update_id"] >= offset]
+176
result = list(STATE["updates"])
+177
ids = [u["update_id"] for u in result]
+178
STATE["delivered_ids"].append({"offset": offset, "returned": ids})
+179
logline(REQ_LOG,
+180
f"getUpdates offset={offset} -> returned={ids}")
+181
self._reply({"ok": True, "result": result})
+182
return
+183
+184
if method == "sendMessage":
+185
chat_id = body.get("chat_id")
+186
text = body.get("text", "")
+187
# Count the send FIRST: in every mode Telegram has received the
+188
# message and would deliver it to David's phone. That is exactly
+189
# the count that matters -- how many times the phone buzzes.
+190
with LOCK:
+191
STATE["sends"] += 1
+192
n = STATE["sends"]
+193
mode = STATE["send_mode"]
+194
logline(SEND_LOG, f"#{n} sendMessage chat_id={chat_id} text={text!r} mode={mode}")
+195
logline(REQ_LOG, f"sendMessage chat_id={chat_id} text={text!r} mode={mode}")
+196
+197
# ---- failure injection ----
+198
if mode == "reset":
+199
# Delivered, but ack read fails: drop the socket, no response.
+200
try:
+201
self.connection.close()
+202
except Exception:
+203
pass
+204
return
+205
if mode.startswith("delay:"):
+206
try:
+207
secs = float(mode.split(":", 1)[1])
+208
except ValueError:
+209
secs = 30.0
+210
time.sleep(secs)
+211
# fall through to a normal ok response (may arrive after the
+212
# client/courier read deadline already fired)
+213
if mode.startswith("status:"):
+214
try:
+215
code = int(mode.split(":", 1)[1])
+216
except ValueError:
+217
code = 500
+218
self._reply({"ok": False, "error_code": code,
+219
"description": "injected error"}, code=code)
+220
return
+221
if mode == "notok":
+222
self._reply({"ok": False, "error_code": 400,
+223
"description": "injected api error"})
+224
return
+225
+226
self._reply({"ok": True, "result": {
+227
"message_id": 9000 + n, "date": 1,
+228
"chat": {"id": chat_id, "type": "private"}, "text": text}})
+229
return
+230
+231
# Any other Telegram method (getMe, etc.) -> generic ok.
+232
logline(REQ_LOG, f"{method} (generic-ok)")
+233
self._reply({"ok": True, "result": {}})
+234
+235
+236
if __name__ == "__main__":
+237
logline(REQ_LOG, f"--- mock start on :{PORT} ---")
+238
srv = ThreadingHTTPServer(("127.0.0.1", PORT), Handler)
+239
srv.serve_forever()
repro/repro-send-duplication.shadded
@@ -0,0 +1,89 @@
+1
#!/usr/bin/env bash
+2
# Off-device regression harness for the courier SEND-PATH DUPLICATION storm.
+3
#
+4
# Proves, entirely against a local mock (NOTHING reaches real Telegram),
+5
# that one logical `send-message` is delivered EXACTLY ONCE even when the
+6
# leader's MCP client gives up on a slow send and SIGKILLs+restarts+re-issues
+7
# it -- the loop that spammed David with many copies of the same message.
+8
#
+9
# * FIXED binary (this worktree): storm -> 1 delivery.
+10
# * optional CONTROL binary (pre-fix master, set COURIER_CONTROL_BIN):
+11
# storm -> N>1 deliveries, demonstrating the bug the fix removes.
+12
#
+13
# Mechanism of the storm (see src/courier/dedup.sgl):
+14
# the sendMessage ack read can take up to 25s; if the client's tool-call
+15
# timeout is shorter (or a human /mcp's because it "looks hung"), it kills
+16
# courier and re-issues the send. Each attempt reaches Telegram before the
+17
# kill. The pre-fix in-memory dedup was wiped on every restart, so it could
+18
# not suppress the retry; the fix persists the dedup to disk and records a
+19
# send BEFORE delivery, so a restarted courier suppresses the re-issue.
+20
#
+21
# Usage:
+22
# ./repro-send-duplication.sh
+23
# COURIER_CONTROL_BIN=/path/to/prefix/courier ./repro-send-duplication.sh
+24
set -u
+25
+26
HERE="$(cd "$(dirname "$0")" && pwd)"
+27
BIN="${COURIER_BIN:-$HERE/../build/dev/bin/courier}"
+28
PORT="${PORT:-8611}"
+29
PY="${PYTHON:-python3}"
+30
RUNDIR="$HERE/run"
+31
FAILED=0
+32
+33
if [ ! -x "$BIN" ]; then
+34
echo "FATAL: courier binary not found at $BIN (run 'sigil build' first)"
+35
exit 2
+36
fi
+37
+38
start_mock() {
+39
rm -rf "$RUNDIR"; mkdir -p "$RUNDIR"
+40
"$PY" "$HERE/mock_telegram.py" "$PORT" "$RUNDIR/mocklog" 2>/dev/null &
+41
MOCK=$!
+42
sleep 1
+43
}
+44
stop_mock() { kill "$MOCK" 2>/dev/null; wait "$MOCK" 2>/dev/null; sleep 0.2; }
+45
+46
# run_case <label> <bin> <expected-deliveries> <driver-args...>
+47
run_case() {
+48
local label="$1" bin="$2" expect="$3"; shift 3
+49
start_mock
+50
"$PY" "$HERE/send_driver.py" --bin "$bin" \
+51
--api-url "http://127.0.0.1:$PORT" \
+52
--stats-url "http://127.0.0.1:$PORT/_stats" \
+53
--mode-endpoint "http://127.0.0.1:$PORT/_mode" \
+54
--logdir "$RUNDIR" "$@" >/dev/null 2>&1
+55
local got
+56
got="$(grep -c sendMessage "$RUNDIR/mocklog/sends.log" 2>/dev/null || echo 0)"
+57
stop_mock
+58
if [ "$got" = "$expect" ]; then
+59
echo " PASS $label: $got delivery(ies) (expected $expect)"
+60
else
+61
echo " FAIL $label: $got delivery(ies) (expected $expect)"
+62
FAILED=1
+63
fi
+64
}
+65
+66
echo "=== FIXED binary: $BIN ==="
+67
run_case "baseline ok -> 1" "$BIN" 1 --mode ok --tool-timeout 30 --retries 0 --text baseline
+68
run_case "reset (ack lost) -> 1" "$BIN" 1 --mode reset --tool-timeout 30 --retries 0 --text reset
+69
run_case "STORM x3 -> 1 (exactly-once)" "$BIN" 1 --mode delay:8 --tool-timeout 3 --retries 3 --text storm3
+70
run_case "STORM x5 -> 1 (exactly-once)" "$BIN" 1 --mode delay:8 --tool-timeout 2 --retries 5 --text storm5
+71
+72
if [ -n "${COURIER_CONTROL_BIN:-}" ] && [ -x "${COURIER_CONTROL_BIN}" ]; then
+73
echo "=== CONTROL (pre-fix) binary: $COURIER_CONTROL_BIN ==="
+74
# The pre-fix binary re-delivers once per attempt: 1 logical send + 3
+75
# retries = 4 deliveries. This is the storm the fix removes.
+76
run_case "STORM x3 -> 4 (bug present)" "$COURIER_CONTROL_BIN" 4 \
+77
--mode delay:8 --tool-timeout 3 --retries 3 --text storm3
+78
else
+79
echo "=== CONTROL skipped (set COURIER_CONTROL_BIN to a pre-fix courier) ==="
+80
fi
+81
+82
rm -rf "$RUNDIR"
+83
if [ "$FAILED" = 0 ]; then
+84
echo "ALL CHECKS PASSED"
+85
exit 0
+86
else
+87
echo "SOME CHECKS FAILED"
+88
exit 1
+89
fi
repro/send_driver.pyadded
@@ -0,0 +1,229 @@
+1
#!/usr/bin/env python3
+2
"""Off-device driver for the courier send-path duplication repro.
+3
+4
Acts as a minimal MCP client (the "leader") speaking JSON-RPC over
+5
courier's stdio. Fires ONE logical `send-message` and models how the
+6
real leader harness reacts when a send does not return a clean, timely
+7
success:
+8
+9
* a per-tool-call timeout (the client stops waiting for the response),
+10
* on timeout OR courier death, kill + respawn courier and RE-ISSUE the
+11
same send, up to --retries times (this is the harness /mcp-reconnect
+12
+ agent-retry loop that turned one logical send into N deliveries).
+13
+14
The bug is proven by pointing courier's SEND path at mock_telegram.py
+15
(COURIER_TELEGRAM_API_URL) and counting sendMessage hits: one logical
+16
send producing >1 delivery == reproduced.
+17
+18
NOTHING here can reach real Telegram: the api-url override forces every
+19
request to the local mock.
+20
+21
Usage:
+22
send_driver.py --bin <courier> --api-url <mock-base> --logdir <dir>
+23
[--mode ok|reset|delay:N|status:N|notok]
+24
[--tool-timeout SEC] [--retries N] [--text STR]
+25
[--mode-endpoint URL]
+26
"""
+27
import argparse
+28
import json
+29
import os
+30
import subprocess
+31
import sys
+32
import threading
+33
import time
+34
import urllib.request
+35
+36
CHAT_ID = "331005009" # David's real chat id, per the captured log. Never used
+37
# live: the api-url override sends only to the mock.
+38
+39
+40
def set_mock_mode(mode_endpoint, mode):
+41
data = json.dumps({"mode": mode}).encode()
+42
req = urllib.request.Request(mode_endpoint, data=data,
+43
headers={"Content-Type": "application/json"})
+44
urllib.request.urlopen(req, timeout=5).read()
+45
+46
+47
def mock_sends(stats_url):
+48
with urllib.request.urlopen(stats_url, timeout=5) as r:
+49
return json.load(r)["sends"]
+50
+51
+52
class Courier:
+53
"""A single courier subprocess speaking MCP over stdio."""
+54
+55
def __init__(self, bin_path, env, logdir, gen):
+56
self.gen = gen
+57
self.errf = open(os.path.join(logdir, f"courier-{gen}.stderr"), "wb")
+58
self.proc = subprocess.Popen(
+59
[bin_path, "serve"],
+60
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=self.errf,
+61
env=env, bufsize=0)
+62
self._id = 0
+63
self._lock = threading.Lock()
+64
+65
def _send(self, obj):
+66
line = (json.dumps(obj) + "\n").encode()
+67
self.proc.stdin.write(line)
+68
self.proc.stdin.flush()
+69
+70
def _rpc(self, method, params, timeout):
+71
"""Send a request, wait up to `timeout` for the matching response.
+72
Returns (result_or_error_dict, elapsed). Raises TimeoutError if no
+73
response arrives in time, or ProcessLookupError if courier dies."""
+74
with self._lock:
+75
self._id += 1
+76
rid = self._id
+77
self._send({"jsonrpc": "2.0", "id": rid, "method": method,
+78
"params": params})
+79
result = {}
+80
done = threading.Event()
+81
+82
def reader():
+83
nonlocal result
+84
for raw in self.proc.stdout:
+85
try:
+86
msg = json.loads(raw)
+87
except Exception:
+88
continue
+89
if msg.get("id") == rid:
+90
result = msg
+91
done.set()
+92
return
+93
+94
t = threading.Thread(target=reader, daemon=True)
+95
t.start()
+96
start = time.time()
+97
ok = done.wait(timeout)
+98
elapsed = time.time() - start
+99
if not ok:
+100
if self.proc.poll() is not None:
+101
raise ProcessLookupError(f"courier exited rc={self.proc.returncode}")
+102
raise TimeoutError(f"no response to {method} in {timeout}s")
+103
return result, elapsed
+104
+105
def notify(self, method, params):
+106
self._send({"jsonrpc": "2.0", "method": method, "params": params})
+107
+108
def initialize(self, timeout):
+109
res, _ = self._rpc("initialize", {
+110
"protocolVersion": "2024-11-05",
+111
"capabilities": {},
+112
"clientInfo": {"name": "send-driver", "version": "0"},
+113
}, timeout)
+114
self.notify("notifications/initialized", {})
+115
return res
+116
+117
def call_send(self, text, timeout):
+118
return self._rpc("tools/call", {
+119
"name": "send-message",
+120
"arguments": {"text": text, "to": CHAT_ID},
+121
}, timeout)
+122
+123
def alive(self):
+124
return self.proc.poll() is None
+125
+126
def kill(self):
+127
try:
+128
self.proc.kill()
+129
self.proc.wait(timeout=5)
+130
except Exception:
+131
pass
+132
try:
+133
self.errf.close()
+134
except Exception:
+135
pass
+136
+137
+138
def main():
+139
ap = argparse.ArgumentParser()
+140
ap.add_argument("--bin", required=True)
+141
ap.add_argument("--api-url", required=True, help="mock base, e.g. http://127.0.0.1:PORT")
+142
ap.add_argument("--logdir", required=True)
+143
ap.add_argument("--stats-url", required=True)
+144
ap.add_argument("--mode-endpoint", required=True)
+145
ap.add_argument("--mode", default="ok")
+146
ap.add_argument("--tool-timeout", type=float, default=30.0)
+147
ap.add_argument("--retries", type=int, default=0,
+148
help="max kill+respawn+re-send cycles after a timeout/death")
+149
ap.add_argument("--text", default="repro test message")
+150
args = ap.parse_args()
+151
+152
env = dict(os.environ)
+153
env["COURIER_TELEGRAM_TOKEN"] = "TESTTOKEN"
+154
env["COURIER_TELEGRAM_CHAT_ID"] = CHAT_ID
+155
env["COURIER_TELEGRAM_API_URL"] = args.api_url
+156
# Sends ENABLED against the mock (this is the whole point).
+157
env.pop("COURIER_DISABLE_TELEGRAM_SEND", None)
+158
env["CLAUDE_OPS_LOG_DIR"] = args.logdir
+159
# Isolate persistent state (relay dir + the send-dedup.log that the
+160
# fix writes beside it) to this run's logdir, so runs don't cross-
+161
# contaminate. Every restarted courier gen shares this dir -- exactly
+162
# how the real leader's restarts share ~/.courier -- so the persistent
+163
# dedup can do its job across process restarts.
+164
env["COURIER_RELAY_DIR"] = os.path.join(os.path.abspath(args.logdir), "relays")
+165
+166
set_mock_mode(args.mode_endpoint, args.mode)
+167
before = mock_sends(args.stats_url)
+168
+169
print(f"== driver: mode={args.mode} tool-timeout={args.tool_timeout}s "
+170
f"retries={args.retries} ==")
+171
+172
attempts = 0
+173
gen = 0
+174
outcome = "UNKNOWN"
+175
cur = Courier(args.bin, env, args.logdir, gen)
+176
try:
+177
cur.initialize(timeout=15)
+178
while True:
+179
attempts += 1
+180
print(f"-- attempt {attempts}: sending (courier gen {gen}) --")
+181
try:
+182
res, elapsed = cur.call_send(args.text, timeout=args.tool_timeout)
+183
txt = ""
+184
if "result" in res:
+185
try:
+186
txt = res["result"]["content"][0]["text"]
+187
except Exception:
+188
txt = json.dumps(res["result"])
+189
else:
+190
txt = json.dumps(res.get("error", res))
+191
print(f" tool returned in {elapsed:.1f}s: {txt!r}")
+192
outcome = "SEND_RETURNED"
+193
break
+194
except TimeoutError as e:
+195
print(f" TIMEOUT: {e}")
+196
except ProcessLookupError as e:
+197
print(f" COURIER DIED: {e}")
+198
# timeout or death: this is where the harness would restart+retry
+199
if attempts > args.retries:
+200
outcome = "GAVE_UP_AFTER_RETRIES"
+201
break
+202
cur.kill()
+203
gen += 1
+204
print(f" -> restart courier (gen {gen}) and re-issue send")
+205
cur = Courier(args.bin, env, args.logdir, gen)
+206
cur.initialize(timeout=15)
+207
finally:
+208
# Close stdin so courier shuts down cleanly, then reap.
+209
time.sleep(0.5)
+210
try:
+211
cur.proc.stdin.close()
+212
except Exception:
+213
pass
+214
time.sleep(0.5)
+215
alive = cur.alive()
+216
cur.kill()
+217
+218
after = mock_sends(args.stats_url)
+219
deliveries = after - before
+220
print(f"== RESULT: logical_sends=1 attempts={attempts} "
+221
f"deliveries_at_mock={deliveries} outcome={outcome} "
+222
f"last_courier_alive_before_kill={alive} ==")
+223
# Machine-readable summary line for scripts.
+224
print(f"SUMMARY mode={args.mode} deliveries={deliveries} "
+225
f"attempts={attempts} outcome={outcome}")
+226
+227
+228
if __name__ == "__main__":
+229
main()
src/courier/dedup.sgladded
@@ -0,0 +1,171 @@
+1
;;; (courier dedup) - Persistent send-idempotency backstop.
+2
;;;
+3
;;; courier delivers exactly one Telegram message per `send-message` tool
+4
;;; call. The danger is DUPLICATION ACROSS PROCESS RESTARTS: the ack read
+5
;;; for a sendMessage can take up to *send-request-timeout* (25s), and if
+6
;;; the leader's MCP client gives up sooner -- a short tool-call timeout,
+7
;;; or a human/harness `/mcp` reconnect because the call "looks hung" --
+8
;;; it SIGKILLs courier and re-issues the same send. Each attempt's POST
+9
;;; reaches Telegram before the kill, so the same message is delivered N
+10
;;; times (the courier-spam storm). The earlier in-memory (chat-id,text)
+11
;;; dedup could not stop this: a fresh process starts with an empty cache.
+12
;;;
+13
;;; The fix is to persist the dedup cache to disk and RECORD A SEND BEFORE
+14
;;; attempting delivery, so a restarted courier re-issuing the same send
+15
;;; recognises it as already-in-flight/delivered and suppresses it. This
+16
;;; trades exactly-once for at-most-once on genuinely ambiguous outcomes
+17
;;; (a send killed mid-flight is assumed delivered and not retried), which
+18
;;; is the correct bias for a notification bot: a rare missed notification
+19
;;; is vastly preferable to a spam storm.
+20
;;;
+21
;;; Keys are FNV-1a hashes of "<chat-id>:<text>" so no message plaintext
+22
;;; is written to disk and the file stays compact. Entries expire on a
+23
;;; sliding window (a repeat refreshes the timestamp), so a retry loop is
+24
;;; suppressed for as long as it keeps firing, then the key ages out.
+25
+26
(define-library (courier dedup)
+27
(import (sigil core)
+28
(sigil math)
+29
(sigil string)
+30
(sigil fs)
+31
(sigil path)
+32
(sigil process)
+33
(courier relay))
+34
(export *send-dedup-window*
+35
send-dedup-window
+36
send-dedup-path
+37
send-dedup-key
+38
dedup-check-and-record!
+39
dedup-unrecord!
+40
;; exported for tests
+41
dedup-load-entries
+42
dedup-prune)
+43
(begin
+44
+45
;; Default sliding-window size (seconds). Comfortably covers a
+46
;; /mcp-reconnect + agent-retry loop (each cycle is seconds) while
+47
;; rarely suppressing an intentional identical re-send. Overridable
+48
;; via COURIER_SEND_DEDUP_WINDOW for tests / tuning.
+49
(define *send-dedup-window* 60)
+50
+51
(define (send-dedup-window)
+52
(: -> number?)
+53
(let ((v (getenv "COURIER_SEND_DEDUP_WINDOW")))
+54
(or (and v (let ((n (string->number v)))
+55
(and n (> n 0) n)))
+56
*send-dedup-window*)))
+57
+58
;; Persist location. Lives beside the relay dir (same place the
+59
;; poller keeps its state), overridable for tests via
+60
;; COURIER_SEND_DEDUP_FILE.
+61
(define (send-dedup-path)
+62
(: -> string?)
+63
(let ((override (getenv "COURIER_SEND_DEDUP_FILE")))
+64
(if (and override (not (string=? override "")))
+65
override
+66
(path-join (path-dirname (default-relay-dir)) "send-dedup.log"))))
+67
+68
;; ---- FNV-1a 64-bit key hashing (no plaintext on disk) ----
+69
(define *fnv-offset* 14695981039346656037)
+70
(define *fnv-prime* 1099511628211)
+71
(define *fnv-mod* 18446744073709551616) ;; 2^64
+72
+73
(define (send-dedup-key chat-id text)
+74
(: (any-of integer? string?) string? -> string?)
+75
(let* ((s (string-append (if (string? chat-id)
+76
chat-id
+77
(number->string chat-id))
+78
":" text))
+79
(n (string-length s)))
+80
(let loop ((i 0) (h *fnv-offset*))
+81
(if (>= i n)
+82
(number->string h 16)
+83
(loop (+ i 1)
+84
(modulo (* (bitwise-xor h (char->integer (string-ref s i)))
+85
*fnv-prime*)
+86
*fnv-mod*))))))
+87
+88
;; ---- File-backed store: list of (key-string . epoch-seconds) ----
+89
+90
;; Parse one "epoch hash" line into (hash . epoch), or #f if malformed.
+91
(define (parse-entry line)
+92
(let ((line (string-trim line)))
+93
(if (string=? line "")
+94
#f
+95
(let ((sp (string-index line (lambda (c) (char=? c #\space)))))
+96
(if (not sp)
+97
#f
+98
(let ((epoch (string->number (substring line 0 sp)))
+99
(key (string-trim (substring line (+ sp 1)
+100
(string-length line)))))
+101
(if (and epoch (> (string-length key) 0))
+102
(cons key epoch)
+103
#f)))))))
+104
+105
(define (dedup-load-entries path)
+106
(: string? -> list?)
+107
(if (file-exists? path)
+108
(guard (e (else '()))
+109
(let loop ((lines (string-split (read-file-string path) "\n"))
+110
(acc '()))
+111
(if (null? lines)
+112
(reverse acc)
+113
(let ((entry (parse-entry (car lines))))
+114
(loop (cdr lines)
+115
(if entry (cons entry acc) acc))))))
+116
'()))
+117
+118
;; Drop entries older than the window.
+119
(define (dedup-prune entries now window)
+120
(: list? number? number? -> list?)
+121
(cond
+122
((null? entries) '())
+123
((< (- now (cdr (car entries))) window)
+124
(cons (car entries) (dedup-prune (cdr entries) now window)))
+125
(else (dedup-prune (cdr entries) now window))))
+126
+127
;; Remove any entry with the given key.
+128
(define (remove-key entries key)
+129
(cond
+130
((null? entries) '())
+131
((string=? (car (car entries)) key)
+132
(remove-key (cdr entries) key))
+133
(else (cons (car entries) (remove-key (cdr entries) key)))))
+134
+135
(define (has-key? entries key)
+136
(cond
+137
((null? entries) #f)
+138
((string=? (car (car entries)) key) #t)
+139
(else (has-key? (cdr entries) key))))
+140
+141
(define (save-entries! path entries)
+142
(ensure-directory (path-dirname path))
+143
(write-file-string path
+144
(apply string-append
+145
(map (lambda (e)
+146
(string-append (number->string (cdr e)) " " (car e) "\n"))
+147
entries))))
+148
+149
;;; Atomically (single-threaded MCP loop) check whether `key` was
+150
;;; recently sent and record it as sent NOW.
+151
;;;
+152
;;; Returns 'suppress if the key is already present within the window
+153
;;; (a duplicate — caller must NOT deliver), refreshing its timestamp
+154
;;; so an ongoing retry loop stays suppressed. Returns 'proceed if the
+155
;;; key is new, having recorded it BEFORE the caller attempts delivery
+156
;;; (so a crash/kill mid-delivery still dedups the next attempt).
+157
(define (dedup-check-and-record! path key now window)
+158
(: string? string? number? number? -> symbol?)
+159
(let* ((entries (dedup-prune (dedup-load-entries path) now window))
+160
(dup (has-key? entries key)))
+161
(save-entries! path (cons (cons key now) (remove-key entries key)))
+162
(if dup 'suppress 'proceed)))
+163
+164
;;; Remove a previously-recorded key. Called when the caller learns
+165
;;; the send definitely did NOT reach Telegram, so a genuine retry is
+166
;;; allowed to go through.
+167
(define (dedup-unrecord! path key)
+168
(: string? string? -> void?)
+169
(when (file-exists? path)
+170
(guard (e (else (values)))
+171
(save-entries! path (remove-key (dedup-load-entries path) key)))))))
src/courier/telegram.sglmodified
@@ -17,6 +17,7 @@
17
(sigil mcp server)
18
(sigil log)
19
(courier config)
+20
(courier dedup)
21
(courier relay))
22
(export register-send-message-tool!
23
register-send-media-tool!
@@ -40,33 +41,13 @@
41
;; read, so the request timeout alone can't bound it).
42
(define *send-connect-timeout* 10)
43
−43
;; Idempotency/dedup backstop for Telegram sends. courier delivers
−44
;; exactly one message per send-message tool call and returns a prompt
−45
;; result; but if the upstream MCP client re-issues an identical tool
−46
;; call (e.g. a timeout-driven retry), courier would deliver it AGAIN.
−47
;; We dedupe identical (chat-id, text) sends within this window: a
−48
;; repeat skips the actual delivery and returns the SAME success (an
−49
;; error would only make the client retry harder). The window must
−50
;; comfortably cover a client's retry interval; a rare intentional
−51
;; identical re-send being suppressed is acceptable for a bot.
−52
(define *send-dedup-window* 30) ;; seconds
−53
−54
;; Drop cache entries older than the dedup window. Self-contained
−55
;; (no list-lib dependency); the cache only ever holds the last few
−56
;; seconds of sends, so O(n) is fine.
−57
(define (dedup-prune sends now)
−58
(cond
−59
((null? sends) '())
−60
((< (- now (cdr (car sends))) *send-dedup-window*)
−61
(cons (car sends) (dedup-prune (cdr sends) now)))
−62
(else (dedup-prune (cdr sends) now))))
−63
−64
;; Is `key` present in the (already-pruned) recent-sends alist?
−65
(define (dedup-seen? sends key)
−66
(cond
−67
((null? sends) #f)
−68
((string=? (car (car sends)) key) #t)
−69
(else (dedup-seen? (cdr sends) key))))
+44
;; Idempotency/dedup backstop for Telegram sends lives in
+45
;; (courier dedup): a PERSISTENT, restart-surviving (chat-id,text)
+46
;; window. It is the load-bearing defence against the send-path
+47
;; duplication storm — see that module's header. The in-memory
+48
;; version this replaces could not survive the SIGKILL-and-retry
+49
;; loop that caused the spam, because a fresh process started with
+50
;; an empty cache.
51
52
;; ============================================================
53
;; Send Message Tool
@@ -84,9 +65,11 @@
65
"https://api.telegram.org"))
66
(send-disabled (courier-config-telegram-send-disabled config))
67
(send-delay (courier-config-telegram-send-delay config))
−87
;; Mutable recent-sends cache (alist of (key . send-second)),
−88
;; updated via set!. Single-threaded MCP loop → no race.
−89
(recent-sends '())
+68
;; Persistent dedup store (survives process restarts) + its
+69
;; sliding window. Resolved once at registration; both honour
+70
;; COURIER_SEND_DEDUP_FILE / COURIER_SEND_DEDUP_WINDOW.
+71
(dedup-path (send-dedup-path))
+72
(dedup-window (send-dedup-window))
73
;; Per-invocation counter, logged at DEBUG (off by default).
74
(send-invocation 0))
75
(mcp-server-register-tool! server
@@ -116,22 +99,22 @@
99
default-chat-id)))
100
(if (and token chat-id)
101
(let* ((now (current-second))
−119
(key (string-append (number->string chat-id)
−120
":" text))
+102
(key (send-dedup-key chat-id text))
103
(inv (begin (set! send-invocation
104
(+ send-invocation 1))
105
send-invocation)))
106
(log-debug "send-message handler" inv: inv
107
chat-id: chat-id)
−126
(set! recent-sends (dedup-prune recent-sends now))
−127
(cond
−128
;; Part 3 (dedup backstop): an identical
−129
;; (chat-id, text) send already DELIVERED within
−130
;; the window → suppress, return the same
−131
;; success. (Only delivered sends are recorded,
−132
;; below, so a never-sent failure is NOT deduped
−133
;; and a client retry correctly re-delivers.)
−134
((dedup-seen? recent-sends key)
+108
;; Persistent, restart-surviving dedup. RECORD
+109
;; the key BEFORE attempting delivery: if the
+110
;; leader SIGKILLs courier mid-send and re-issues
+111
;; the same send (the storm), the fresh process
+112
;; sees the recorded key and suppresses it. An
+113
;; in-memory cache could not do this — a restart
+114
;; wiped it, which is why the spam recurred.
+115
(case (dedup-check-and-record! dedup-path key
+116
now dedup-window)
+117
((suppress)
118
(log-info "Duplicate send-message suppressed"
119
chat-id: chat-id)
120
"Message sent.")
@@ -140,38 +123,43 @@
123
(when (and (number? send-delay) (> send-delay 0))
124
(sleep send-delay))
125
(if send-disabled
−143
;; Test hook: dry-run — no real delivery,
−144
;; but treat as sent for dedup purposes.
+126
;; Test hook: dry-run — no real delivery.
127
(begin
−146
(set! recent-sends
−147
(cons (cons key now) recent-sends))
128
(log-info "Telegram send disabled (dry-run)"
129
chat-id: chat-id)
130
"Message sent.")
−151
;; Part 1 (best-effort success): a send that
−152
;; reached Telegram but whose ack read failed
−153
;; (delivered, ack unconfirmed) returns SUCCESS
−154
;; so the client doesn't retry a delivered
−155
;; message; a genuine never-sent failure
−156
;; propagates as an error (client retry
−157
;; correctly re-delivers). Either delivered
−158
;; outcome records the key for the dedup
−159
;; backstop.
+131
;; Deliver exactly once. The key is already
+132
;; recorded. A transport error must NEVER
+133
;; escape this handler (an MCP error invites
+134
;; a client retry) and must never re-fire a
+135
;; send.
136
(guard (e ((tg-ack-unconfirmed? e)
−161
(set! recent-sends
−162
(cons (cons key now)
−163
recent-sends))
+137
;; Reached Telegram, ack read
+138
;; failed: treat as delivered,
+139
;; keep the recorded key.
140
(log-info "Telegram delivered, ack unconfirmed"
141
chat-id: chat-id)
−166
"Message sent (ack unconfirmed)."))
+142
"Message sent (ack unconfirmed).")
+143
(else
+144
;; ok:true is the ONLY delivery
+145
;; path; any other error means
+146
;; Telegram did not accept the
+147
;; message (never-sent or
+148
;; rejected). Release the key so
+149
;; a genuine retry can go through,
+150
;; and return cleanly (no re-raise,
+151
;; no crash).
+152
(dedup-unrecord! dedup-path key)
+153
(log-warn "Telegram send failed"
+154
chat-id: chat-id
+155
error: (format "~a" e))
+156
"Error: Telegram send failed (message not delivered)."))
157
(tg-send-message
158
(tg-client token: token
159
api-url: api-url
160
request-timeout: *send-request-timeout*
161
connect-timeout: *send-connect-timeout*)
162
chat-id text)
−173
(set! recent-sends
−174
(cons (cons key now) recent-sends))
163
(log-info "Telegram message sent" chat-id: chat-id)
164
"Message sent.")))))
165
"Error: Telegram not configured (missing token or chat ID)"))))))))))
test/test-send-dedup.sgladded
@@ -0,0 +1,180 @@
+1
;; Unit tests for (courier dedup) -- the persistent send-idempotency
+2
;; backstop that stops the send-path duplication storm.
+3
;;
+4
;; The storm is: a send whose ack read is slow gets the leader's MCP
+5
;; client to SIGKILL courier and re-issue the same send; each attempt
+6
;; reaches Telegram, so one logical send is delivered N times. The old
+7
;; in-memory dedup could not stop it because a restarted process starts
+8
;; with an empty cache. These tests pin the behaviour that fixes it:
+9
;; the dedup is file-backed (survives "restart" = a second call reading
+10
;; the same file) and a key is recorded BEFORE delivery.
+11
+12
(import (sigil test)
+13
(sigil core)
+14
(sigil string)
+15
(sigil fs)
+16
(sigil path)
+17
(sigil process)
+18
(courier dedup))
+19
+20
(define (fresh-path)
+21
(path-join (make-temp-directory) "send-dedup.log"))
+22
+23
;; ============================================================
+24
;; Key hashing
+25
;; ============================================================
+26
+27
(test-group "send-dedup-key"
+28
(test "is deterministic for the same (chat-id, text)"
+29
(assert-equal (send-dedup-key 331005009 "hello")
+30
(send-dedup-key 331005009 "hello")))
+31
+32
(test "differs for different text"
+33
(assert-true (not (string=? (send-dedup-key 331005009 "hello")
+34
(send-dedup-key 331005009 "hellO")))))
+35
+36
(test "differs for different chat-id"
+37
(assert-true (not (string=? (send-dedup-key 1 "hi")
+38
(send-dedup-key 2 "hi")))))
+39
+40
(test "accepts a string chat-id equivalently to a number"
+41
(assert-equal (send-dedup-key 42 "x")
+42
(send-dedup-key "42" "x")))
+43
+44
(test "is a single hex token safe for line-based storage"
+45
;; Text with spaces/newlines must not leak into the key (which is
+46
;; stored as `<epoch> <key>` on one line).
+47
(let ((k (send-dedup-key 7 "multi\nline text with spaces")))
+48
(assert-false (string-contains? k " "))
+49
(assert-false (string-contains? k "\n")))))
+50
+51
;; ============================================================
+52
;; check-and-record: the core suppress/proceed decision
+53
;; ============================================================
+54
+55
(test-group "dedup-check-and-record!"
+56
(test "a new key proceeds; an immediate repeat is suppressed"
+57
(let ((path (fresh-path))
+58
(k (send-dedup-key 1 "msg")))
+59
(assert-equal 'proceed (dedup-check-and-record! path k 1000 60))
+60
(assert-equal 'suppress (dedup-check-and-record! path k 1001 60))))
+61
+62
(test "distinct keys never suppress each other"
+63
(let ((path (fresh-path))
+64
(a (send-dedup-key 1 "aaa"))
+65
(b (send-dedup-key 1 "bbb")))
+66
(assert-equal 'proceed (dedup-check-and-record! path a 1000 60))
+67
(assert-equal 'proceed (dedup-check-and-record! path b 1000 60))))
+68
+69
(test "a repeat AFTER the window proceeds again (entry expired)"
+70
(let ((path (fresh-path))
+71
(k (send-dedup-key 1 "msg")))
+72
(assert-equal 'proceed (dedup-check-and-record! path k 1000 60))
+73
;; 61s later, outside the 60s window
+74
(assert-equal 'proceed (dedup-check-and-record! path k 1061 60))))
+75
+76
(test "suppression refreshes the timestamp (sliding window)"
+77
;; Record at t=1000; hit at t=1050 (within 60) suppresses AND
+78
;; refreshes to 1050; a hit at t=1100 is 50s after the refresh, so
+79
;; still within the window -> a retry loop stays suppressed as long
+80
;; as it keeps firing.
+81
(let ((path (fresh-path))
+82
(k (send-dedup-key 1 "msg")))
+83
(assert-equal 'proceed (dedup-check-and-record! path k 1000 60))
+84
(assert-equal 'suppress (dedup-check-and-record! path k 1050 60))
+85
(assert-equal 'suppress (dedup-check-and-record! path k 1100 60)))))
+86
+87
;; ============================================================
+88
;; Persistence across "restart" (a second reader of the same file)
+89
;; ============================================================
+90
+91
(test-group "persistence across restart"
+92
(test "a key recorded by one call is seen by a later call on the same file"
+93
;; This is the whole fix: process A records the key (before it is
+94
;; SIGKILLed mid-send), process B re-issues the same send and reads
+95
;; the SAME file -> suppress. Modeled here as two calls sharing path.
+96
(let ((path (fresh-path))
+97
(k (send-dedup-key 331005009 "storm")))
+98
(assert-equal 'proceed (dedup-check-and-record! path k 5000 60))
+99
;; "process B" -- brand new call, no in-memory state, same file
+100
(assert-equal 'suppress (dedup-check-and-record! path k 5002 60))))
+101
+102
(test "recorded entries are loadable from disk"
+103
(let ((path (fresh-path))
+104
(k (send-dedup-key 1 "persisted")))
+105
(dedup-check-and-record! path k 2000 60)
+106
(let ((entries (dedup-load-entries path)))
+107
(assert-equal 1 (length entries))
+108
(assert-equal k (car (car entries)))
+109
(assert-equal 2000 (cdr (car entries)))))))
+110
+111
;; ============================================================
+112
;; unrecord: release a key when delivery definitely did not happen
+113
;; ============================================================
+114
+115
(test-group "dedup-unrecord!"
+116
(test "unrecording a key lets the next send proceed"
+117
(let ((path (fresh-path))
+118
(k (send-dedup-key 1 "msg")))
+119
(assert-equal 'proceed (dedup-check-and-record! path k 1000 60))
+120
(dedup-unrecord! path k)
+121
(assert-equal 'proceed (dedup-check-and-record! path k 1001 60))))
+122
+123
(test "unrecording one key leaves others intact"
+124
(let ((path (fresh-path))
+125
(a (send-dedup-key 1 "a"))
+126
(b (send-dedup-key 1 "b")))
+127
(dedup-check-and-record! path a 1000 60)
+128
(dedup-check-and-record! path b 1000 60)
+129
(dedup-unrecord! path a)
+130
(assert-equal 'proceed (dedup-check-and-record! path a 1001 60))
+131
(assert-equal 'suppress (dedup-check-and-record! path b 1001 60))))
+132
+133
(test "unrecording on a missing file is a no-op (no crash)"
+134
(let ((path (fresh-path))
+135
(k (send-dedup-key 1 "x")))
+136
(dedup-unrecord! path k) ;; file does not exist yet
+137
(assert-equal 'proceed (dedup-check-and-record! path k 1000 60)))))
+138
+139
;; ============================================================
+140
;; prune + robustness
+141
;; ============================================================
+142
+143
(test-group "dedup-prune"
+144
(test "drops entries older than the window, keeps fresh ones"
+145
(let ((entries (list (cons "old" 1000) (cons "fresh" 1900))))
+146
(let ((kept (dedup-prune entries 1950 60)))
+147
(assert-equal 1 (length kept))
+148
(assert-equal "fresh" (car (car kept)))))))
+149
+150
(test-group "robustness"
+151
(test "a malformed dedup file loads as empty (guarded), send proceeds"
+152
(let ((path (fresh-path)))
+153
(ensure-directory (path-dirname path))
+154
(write-file-string path "garbage no epoch here\n\n \n")
+155
(assert-equal '() (dedup-load-entries path))
+156
(assert-equal 'proceed
+157
(dedup-check-and-record! path (send-dedup-key 1 "x") 1000 60)))))
+158
+159
;; ============================================================
+160
;; Config overrides
+161
;; ============================================================
+162
+163
(test-group "config overrides"
+164
(test "COURIER_SEND_DEDUP_WINDOW overrides the default window"
+165
(setenv! "COURIER_SEND_DEDUP_WINDOW" "120")
+166
(assert-equal 120 (send-dedup-window))
+167
(setenv! "COURIER_SEND_DEDUP_WINDOW" "")
+168
(assert-equal *send-dedup-window* (send-dedup-window)))
+169
+170
(test "a non-positive / non-numeric window falls back to the default"
+171
(setenv! "COURIER_SEND_DEDUP_WINDOW" "0")
+172
(assert-equal *send-dedup-window* (send-dedup-window))
+173
(setenv! "COURIER_SEND_DEDUP_WINDOW" "nonsense")
+174
(assert-equal *send-dedup-window* (send-dedup-window))
+175
(setenv! "COURIER_SEND_DEDUP_WINDOW" ""))
+176
+177
(test "COURIER_SEND_DEDUP_FILE overrides the path"
+178
(setenv! "COURIER_SEND_DEDUP_FILE" "/tmp/custom-dedup.log")
+179
(assert-equal "/tmp/custom-dedup.log" (send-dedup-path))
+180
(setenv! "COURIER_SEND_DEDUP_FILE" "")))