GenServer-based batch writer with configurable flush intervals, batch sizes, retry with exponential backoff, and backpressure.

Points or pre-encoded line protocol are buffered in memory and flushed either when the buffer reaches batch_size lines or when the flush_interval_ms timer fires — whichever comes first. Every flush, however triggered (size, timer, flush/2, write_sync/3), restarts the interval, so a write is never left buffered longer than one interval after the last flush.

A connection started with a batch_writer: option has one of these; reach it with InfluxElixir.write_batched/3, InfluxElixir.flush/2 and InfluxElixir.stats/2.

Options

Validated by start_link/1, which returns {:error, %NimbleOptions.ValidationError{}} for an unknown key or a value of the wrong type — a batch_size: 0 used to refuse every write as :buffer_full, and a misspelt key was silently ignored.

• :connection (term/0) - Required. connection term passed to InfluxElixir.Write.Writer, or the name of a registered connection, resolved on every flush

• :database (String.t/0) - database every flush writes to. A :database in :write_opts wins.

• :batch_size (pos_integer/0) - lines a flush is made of: the buffer is written when it holds this many lines (a payload of several lines counts them all; one write is never split, so a flush may hold more) The default value is 5000.

• :flush_interval_ms (pos_integer/0) - timer interval in milliseconds The default value is 1000.

• :jitter_ms (non_neg_integer/0) - random jitter added to the flush timer and retry backoff The default value is 0.

• :max_retries (non_neg_integer/0) - retry attempts for transport errors, 5xx, 408 and 429; every other error is permanent and not retried The default value is 3.

• :base_retry_delay_ms (non_neg_integer/0) - base of the exponential backoff: attempt N waits about base * 2^N The default value is 100.

• :max_retry_delay_ms (pos_integer/0) - cap on the exponential backoff, before jitter: no retry waits longer than this (plus :jitter_ms) The default value is 30000.

• :no_sync (boolean/0) - when true, write_sync/3 behaves like write/3. Not InfluxDB 3's no_sync write parameter: pass that, like accept_partial: false, in :write_opts. The default value is false.

• :write_opts (keyword/0) - options for InfluxElixir.Write.Writer.write/3 on every flush (:database, :timeout, :precision, ...). A Point's DateTime timestamp is encoded in its :precision; an integer timestamp must already be in that unit. The default value is [].

• :client (atom/0) - client module to write with instead of the configured one (InfluxElixir.Client.impl/0)

• :name (term/0) - a GenServer name to register the writer under

• :shutdown - how long a supervisor waits for the final flush when it stops the writer (OTP's default for workers); see "Shutdown" The default value is 5000.

What can be written

write/3 and write_sync/3 take a InfluxElixir.Write.Point, a list of them, or line protocol (a binary or iodata). The payload is encoded in the caller, so a point no server accepts is the caller's {:error, reason} (the one InfluxElixir.Write.LineProtocol.encode/2 gives) and never reaches, or crashes, the writer. Anything else is {:error, {:invalid_payload, payload}}, and the writer keeps running.

A Point is encoded for the connection's api_version and the flushes' :precision. One without a :timestamp is stamped with the time of the call (a DateTime in that precision), not left for the server to stamp: a retry re-sends the same bytes, and a server that committed a write which then timed out would otherwise store the retried points a second time, at a later time. Pre-encoded line protocol without timestamps is still stamped by the server, on every attempt.

Requests

The HTTP requests are made by a process of their own, one at a time and in the order the batches were taken, and the writer handles the answer as a message. A batch waiting out its retry delay is not in that order: an explicit flush/2 or write_sync/3 meanwhile is written first, so a point it rewrites (same series and time) can be overwritten by the older batch's retry. The writer therefore answers write/3, flush/2 and stats/2 while a request is in flight, whatever the server's speed, and a write/3 that completes a batch returns at once instead of waiting for the request.

Backpressure

The buffer is counted in lines. While any batch is in flight or being retried, the automatic flushes (batch size reached, timer fired) wait for every batch to finish instead of starting another against a server that is already failing; writes keep buffering meanwhile. Once the buffer holds 10 * batch_size lines, write/3 and write_sync/3 return {:error, :buffer_full} until the last batch ends, so the body of a flush is bounded by 10 * batch_size lines plus the one write that crossed the bound (a write is never split). An explicit flush/2, and write_sync/3 without :no_sync, always take the buffer immediately — their request waits only for the request in flight. When the last batch ends the deferred buffer is flushed if it has reached batch_size; otherwise the timer takes it.

Shutdown

The writer traps exits, so a supervisor stopping it — application shutdown, InfluxElixir.remove_connection/1 — runs terminate/2, which waits for the request in flight, then writes every batch that is still waiting to be retried and then the buffer, once each and in the order they were written, and answers any write_sync/3 caller waiting on them. A write that fails there is logged and dropped; it is not counted in stats/2, which nobody can read any more. The supervisor kills the writer if that takes longer than :shutdown.

Retry Policy

A write's outcome is :ok, retryable or permanent, by one rule:

• retryable: a transport error ({:connection_error, _}, :pool_timeout, :timeout), a 5xx status, 408 (request timeout) and 429 (too many requests);
• permanent: everything else — any other status, {:invalid_connection, _}, :no_database_specified, {:invalid_*, _}, {:unencodable_body, _}, {:invalid_options, _}. The same batch would fail again; it is discarded and logged at once, whatever max_retries is.

Retryable errors are retried with exponential backoff (base_retry_delay_ms * 2^attempt, capped at max_retry_delay_ms) and optional jitter. Retries are non-blocking: the writer keeps answering between attempts and during them. A write_sync/3 caller whose batch is being retried is answered with that chain's final result; a flush/2 caller is answered when its batch's first attempt has finished.

Stats

Call stats/2 to retrieve a map with :total_writes, :total_errors, and :total_bytes counters. A batch is counted once, when it has been written or has failed for good.