Buffer QuestDB writes through an outage and replay them

CHANGELOG.md

@@ -6,6 +6,34 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## [Unreleased]
 
+## [1.6.0] - 2026-08-11
+
+### Added
+
+- The QuestDB sink holds readings in memory while the connection is down and
+  replays them when it comes back, so a QuestDB restart no longer leaves a gap
+  in the data. Rows are held until a flush confirms them: the ILP client keeps
+  written rows in its own buffer until the next flush and discards them when
+  that flush fails, which is where the readings were being lost.
+  `QuestDB.MaxBufferBytes` caps the buffer at 4 MiB by default. Past the cap
+  the oldest rows are evicted and the store call starts failing, which the
+  service escalates to a process exit, so the pod crash-loops instead of
+  growing until the kernel kills it. Set it to 0 for the previous
+  drop-on-disconnect behaviour.
+- `/healthz` no longer restarts a pod whose sink is recovering in place. A
+  health checker can implement the new `healthserver.Degrader` interface to
+  report that it is failing but holding data, and the liveness threshold skips
+  it. `/readyz` still goes red immediately. Without this the liveness probe
+  would restart the pod after 90s and throw away the buffer that is holding
+  the readings. Probe responses carry a `degraded` field.
+
+### Known limits
+
+- ILP over TCP has no server acknowledgement. The last flush before a socket
+  error is reported as successful even though QuestDB never stored those rows,
+  so they cannot be replayed. The buffer covers everything from the first
+  reported failure onwards.
+
 ## [1.5.3] - 2026-08-11
 
 ### Fixed

cmd/meterlogger/sinks.go

@@ -104,10 +104,11 @@ func newQuestDBClient(
 	healthSrv *healthserver.Server,
 ) (*qdb.DBClient, error) {
 	client, err := qdb.NewDBClient(ctx, qdb.Config{
-		Host:     cfg.QuestDB.Host,
-		Port:     cfg.QuestDB.Port,
-		User:     cfg.QuestDB.User,
-		Password: cfg.QuestDB.Password,
+		Host:           cfg.QuestDB.Host,
+		Port:           cfg.QuestDB.Port,
+		User:           cfg.QuestDB.User,
+		Password:       cfg.QuestDB.Password,
+		MaxBufferBytes: cfg.QuestDB.MaxBufferBytes,
 	}, l)
 	if err != nil {
 		return nil, err

config.example.yaml

@@ -32,6 +32,9 @@ QuestDB:
   Port: 9009
   User: admin
   Password: quest
+  # Rows are held in memory while QuestDB is unreachable and replayed when it
+  # returns. Set to 0 to drop them instead.
+  MaxBufferBytes: 4194304
 
 # ── Stdout sink (debug) ──────────────────────────────────────
 # Logs readings instead of persisting them. Not for production.

documentation/configuration.md

@@ -120,6 +120,9 @@ QuestDB:
   Port: 9009                 # ILP (InfluxDB line protocol) TCP port
   User: admin
   Password: quest
+  MaxBufferBytes: 4194304    # hold up to 4 MiB of rows in memory while QuestDB
+                             # is unreachable and replay them on reconnect.
+                             # 0 drops rows during an outage instead.
 
 # ── PostgreSQL sink ──────────────────────────────────────────
 # Tables are created/migrated automatically on startup.
@@ -575,13 +578,18 @@ See [data-model.md](./data-model.md#water_meter-configurable-name) for the table
 
 ### QuestDB
 
-| Key                | Type   | Default | Notes                   |
-|--------------------|--------|---------|-------------------------|
-| `QuestDB.Enabled`  | bool   | `false` | Must be set explicitly  |
-| `QuestDB.Host`     | string |         | Hostname or IP          |
-| `QuestDB.Port`     | int    | 9009    | ILP TCP port            |
-| `QuestDB.User`     | string |         |                         |
-| `QuestDB.Password` | string |         |                         |
+| Key                      | Type   | Default   | Notes                                                  |
+|--------------------------|--------|-----------|--------------------------------------------------------|
+| `QuestDB.Enabled`        | bool   | `false`   | Must be set explicitly                                 |
+| `QuestDB.Host`           | string |           | Hostname or IP                                         |
+| `QuestDB.Port`           | int    | 9009      | ILP TCP port                                           |
+| `QuestDB.User`           | string |           |                                                        |
+| `QuestDB.Password`       | string |           |                                                        |
+| `QuestDB.MaxBufferBytes` | int    | `4194304` | Write buffer held during an outage; `0` disables it    |
+
+The write buffer holds rows that QuestDB has not confirmed, so an outage does not lose data.
+See [observability.md - QuestDB connection loss](./observability.md#questdb-connection-loss)
+for what happens when it fills up.
 
 ### PostgreSQL
 

documentation/observability.md

@@ -27,10 +27,29 @@ any sink fails, the endpoint returns `503 Service Unavailable` with details in t
 ### QuestDB connection loss
 
 QuestDB ILP runs over a single long-lived TCP connection. When the server closes it (restart, upgrade, host
-reboot) the sink closes the dead socket and redials with exponential backoff, from 1s up to 60s. Rows handed
-to the sink while the connection is down are dropped and counted; the count and the total downtime are
-logged on the successful reconnect. Because a flush with an empty buffer writes no bytes, it cannot detect a
-closed peer, so the loss surfaces on the first row written after the server went away.
+reboot) the sink closes the dead socket and redials with exponential backoff, from 1s up to 60s.
+
+Rows are held in memory until a flush confirms them, so an outage does not lose the readings taken during
+it. On reconnect the held rows are replayed in order and the downtime is logged. `QuestDB.MaxBufferBytes`
+caps the buffer, 4 MiB by default. Past the cap the oldest rows are evicted, the drop is logged, and the
+store call starts returning errors, which the service escalates to a process exit after five in a row. The
+pod then crash-loops until QuestDB is reachable again. Set `MaxBufferBytes` to 0 to drop rows during an
+outage instead of holding them.
+
+| State                         | `/readyz` | `/healthz` | Result                                    |
+|-------------------------------|-----------|------------|-------------------------------------------|
+| Connected                     | 200       | 200        | Normal                                    |
+| Disconnected, buffer has room | 503       | 200        | Pod keeps running, readings held in memory |
+| Disconnected, buffer full     | 503       | 503        | Pod restarts, readings dropped            |
+
+A buffering sink deliberately keeps `/healthz` green. Restarting the pod would throw away the buffer that is
+holding the data, so liveness leaves it alone until the buffer overflows.
+
+Two limits are worth knowing. A flush with an empty buffer writes no bytes, so it cannot detect a closed
+peer; the loss surfaces on the next flush that actually carries rows. And ILP over TCP has no server
+acknowledgement, so the last flush before the socket error is reported as successful even though QuestDB
+never stored it. Those rows are not recoverable. The buffer protects everything from the first reported
+failure onwards.
 
 ### Liveness detail
 

documentation/troubleshooting.md

@@ -156,10 +156,15 @@ QuestDB uses ILP/TCP and reports failures through `/readyz` and the write-error
 below.
 
 If QuestDB restarts while meterlogger is running, expect one `questdb: connection lost, will reconnect`
-error followed by `questdb: reconnecting` and `questdb: reconnected` with the downtime and the number of
-rows dropped in the gap. Repeated `questdb: reconnect failed` lines mean the server is still unreachable;
-the retry interval grows to a maximum of 60s. A continuous stream of `broken pipe` write errors on the same
-source port is the pre-1.5.3 behaviour and means the pod is running an old image.
+error followed by `questdb: reconnecting`, then `questdb: reconnected` with the downtime and
+`questdb: replayed buffered rows` with the number of readings recovered from the write buffer. Repeated
+`questdb: reconnect failed` lines mean the server is still unreachable; the retry interval grows to a
+maximum of 60s. A continuous stream of `broken pipe` write errors on the same source port is the pre-1.5.3
+behaviour and means the pod is running an old image.
+
+`questdb: write buffer full, dropping oldest rows` means the outage outlasted `QuestDB.MaxBufferBytes` of
+readings. From that point data is being lost and the pod will restart itself. Raise `MaxBufferBytes` if the
+deployment needs to ride out longer outages, keeping in mind that it is process memory.
 
 ---
 

internal/adapters/sink/qdb/buffer.go

@@ -0,0 +1,174 @@
+package qdb
+
+import (
+	"context"
+	"math/big"
+	"time"
+
+	qdbclient "github.com/questdb/go-questdb-client/v3"
+)
+
+// RowBuilder writes one ILP row into sender. It receives its own context
+// because a buffered row is replayed long after the store call that produced
+// it has returned and that call's timeout has expired.
+type RowBuilder func(ctx context.Context, sender qdbclient.LineSender) error
+
+// bufferedRow is one row waiting for the connection to come back, together
+// with the estimated size of its ILP encoding.
+type bufferedRow struct {
+	build RowBuilder
+	size  int
+}
+
+// rowBuffer holds rows written while the ILP connection is down, oldest first,
+// under a byte cap. Once the cap is reached the oldest rows are evicted: the
+// newest data is the most useful, and an unbounded buffer would trade a data
+// gap for an OOM kill.
+type rowBuffer struct {
+	maxBytes int
+
+	rows    []bufferedRow
+	bytes   int
+	dropped int64
+}
+
+func newRowBuffer(maxBytes int) *rowBuffer {
+	return &rowBuffer{maxBytes: maxBytes}
+}
+
+// enabled reports whether rows are buffered at all. A zero cap restores the
+// drop-on-disconnect behaviour.
+func (b *rowBuffer) enabled() bool { return b.maxBytes > 0 }
+
+// add appends a row, evicting the oldest rows if it does not fit. It reports
+// whether the row was stored without evicting anything.
+func (b *rowBuffer) add(row bufferedRow) bool {
+	if !b.enabled() || row.size > b.maxBytes {
+		b.dropped++
+		return false
+	}
+
+	evicted := false
+	for b.bytes+row.size > b.maxBytes && len(b.rows) > 0 {
+		b.bytes -= b.rows[0].size
+		b.rows = b.rows[1:]
+		b.dropped++
+		evicted = true
+	}
+
+	b.rows = append(b.rows, row)
+	b.bytes += row.size
+	return !evicted
+}
+
+// take removes and returns up to n rows from the front.
+func (b *rowBuffer) take(n int) []bufferedRow {
+	if n > len(b.rows) {
+		n = len(b.rows)
+	}
+	batch := b.rows[:n]
+	for _, row := range batch {
+		b.bytes -= row.size
+	}
+	// Copy so the retained tail does not keep the batch alive through the
+	// shared backing array.
+	b.rows = append([]bufferedRow(nil), b.rows[n:]...)
+	return batch
+}
+
+// pushFront puts an unsent batch back at the head, keeping insertion order.
+func (b *rowBuffer) pushFront(batch []bufferedRow) {
+	for _, row := range batch {
+		b.bytes += row.size
+	}
+	b.rows = append(batch, b.rows...)
+}
+
+func (b *rowBuffer) len() int { return len(b.rows) }
+
+// reset clears the buffer and the drop count after a successful drain.
+func (b *rowBuffer) reset() {
+	b.rows = nil
+	b.bytes = 0
+	b.dropped = 0
+}
+
+// Estimated ILP encoding sizes. Symbols and strings are measured exactly;
+// numeric values use the widest formatting they can produce, so the estimate
+// errs towards over-counting and the memory cap is never exceeded in practice.
+const (
+	sizeSeparators   = 3  // table/symbol-set/column-set/timestamp separators plus newline
+	sizeInt64Value   = 21 // -9223372036854775808 plus the 'i' suffix
+	sizeFloat64Value = 24 // widest strconv 'G' rendering
+	sizeBoolValue    = 1
+	sizeTimestampVal = 21 // microseconds since epoch plus the 't' suffix
+	sizeLong256Value = 66 // "0x" plus 64 hex digits
+	sizeStringQuotes = 2
+	sizeFieldName    = 2 // '=' and the ',' that separates fields
+)
+
+// sizingSender implements qdbclient.LineSender and measures the ILP encoding
+// of a row without sending anything. The client does not expose the encoded
+// length of a message, so the buffer estimates it here.
+type sizingSender struct {
+	size int
+}
+
+func (s *sizingSender) Table(name string) qdbclient.LineSender {
+	s.size += len(name) + sizeSeparators
+	return s
+}
+
+func (s *sizingSender) Symbol(name, val string) qdbclient.LineSender {
+	s.size += len(name) + len(val) + sizeFieldName
+	return s
+}
+
+func (s *sizingSender) Int64Column(name string, _ int64) qdbclient.LineSender {
+	s.size += len(name) + sizeInt64Value + sizeFieldName
+	return s
+}
+
+func (s *sizingSender) Long256Column(name string, _ *big.Int) qdbclient.LineSender {
+	s.size += len(name) + sizeLong256Value + sizeFieldName
+	return s
+}
+
+func (s *sizingSender) TimestampColumn(name string, _ time.Time) qdbclient.LineSender {
+	s.size += len(name) + sizeTimestampVal + sizeFieldName
+	return s
+}
+
+func (s *sizingSender) Float64Column(name string, _ float64) qdbclient.LineSender {
+	s.size += len(name) + sizeFloat64Value + sizeFieldName
+	return s
+}
+
+func (s *sizingSender) StringColumn(name, val string) qdbclient.LineSender {
+	s.size += len(name) + len(val) + sizeStringQuotes + sizeFieldName
+	return s
+}
+
+func (s *sizingSender) BoolColumn(name string, _ bool) qdbclient.LineSender {
+	s.size += len(name) + sizeBoolValue + sizeFieldName
+	return s
+}
+
+func (s *sizingSender) At(_ context.Context, _ time.Time) error {
+	s.size += sizeTimestampVal
+	return nil
+}
+
+func (s *sizingSender) AtNow(_ context.Context) error { return nil }
+func (s *sizingSender) Flush(_ context.Context) error { return nil }
+func (s *sizingSender) Close(_ context.Context) error { return nil }
+
+// measure runs build against a sizing sender and reports the estimated ILP
+// size of the row it writes.
+func measure(ctx context.Context, build RowBuilder) (int, error) {
+	s := &sizingSender{}
+	if err := build(ctx, s); err != nil {
+		return 0, err
+	}
+	return s.size, nil
+}