From a69138d75b29db491cb207e48bfb4a512ac5ffbe Mon Sep 17 00:00:00 2001 From: Alex Vanderpot Date: Mon, 15 Jun 2026 22:25:49 -0400 Subject: [PATCH] ingest: add configurable MQTT write timeout (MQTT_WRITE_TIMEOUT_SECONDS) Bounds PINGREQ/SUBSCRIBE writes so a stalled write through the Cloudflare WebSocket proxy can't hang the client. Default 0 (paho's existing no-timeout behavior); wired through docker-compose. Recommended ~20s when behind a buffering reverse proxy. Co-Authored-By: Claude Opus 4.8 (1M context) --- docker-compose.yml | 1 + ingest/internal/ingestcommon/ingest.go | 4 ++++ 2 files changed, 5 insertions(+) diff --git a/docker-compose.yml b/docker-compose.yml index e11c70a..c58dada 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -64,6 +64,7 @@ services: - MQTT_STALE_AFTER_SECONDS=${MQTT_STALE_AFTER_SECONDS:-300} - MQTT_KEEPALIVE_SECONDS=${MQTT_KEEPALIVE_SECONDS:-30} - MQTT_PING_TIMEOUT_SECONDS=${MQTT_PING_TIMEOUT_SECONDS:-20} + - MQTT_WRITE_TIMEOUT_SECONDS=${MQTT_WRITE_TIMEOUT_SECONDS:-0} - MESHCORE_BATCH_FLUSH_SECONDS=${MESHCORE_BATCH_FLUSH_SECONDS:-10} - MESHCORE_BATCH_MAX_ROWS=${MESHCORE_BATCH_MAX_ROWS:-5000} - MESHCORE_BATCH_BUFFER=${MESHCORE_BATCH_BUFFER:-50000} diff --git a/ingest/internal/ingestcommon/ingest.go b/ingest/internal/ingestcommon/ingest.go index 8eb55e4..5c1e40b 100644 --- a/ingest/internal/ingestcommon/ingest.go +++ b/ingest/internal/ingestcommon/ingest.go @@ -154,6 +154,10 @@ func (d *Daemon) connectToBroker(broker MQTTBrokerConfig, idx int, maxRetries in // proxy's buffering/jitter; 10s was tight and produced false // "pingresp not received" disconnects. Configurable for tuning. opts.SetPingTimeout(time.Duration(GetEnvIntOrDefault("MQTT_PING_TIMEOUT_SECONDS", 20)) * time.Second) + // Bound writes (PINGREQ/SUBSCRIBE) so a stalled write through the + // Cloudflare WebSocket proxy can't hang the client. 0 (paho default) + // disables it; set MQTT_WRITE_TIMEOUT_SECONDS to enable. + opts.SetWriteTimeout(time.Duration(GetEnvIntOrDefault("MQTT_WRITE_TIMEOUT_SECONDS", 0)) * time.Second) // Start a fresh session on every (re)connect and resubscribe explicitly in // the OnConnect handler below. This avoids depending on broker-side session // persistence — which is exactly what left the client connected-but-