fix(influx): drain the spool forward and compact once per flush

Every batch re-read and rewrote the whole spool, which is quadratic on a
long outage and holds the file in heap. Draining now remembers a delivered
offset and cuts the head once per flush. The cap cuts to 90% so it does
not rewrite on every append. 413 now halves the batch (and the batch size
for later flushes) instead of dropping; a single line too big stays queued.
Drop warnings: the first of a flush in full, then one count.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
(cherry picked from commit 1c795a84b98175ebf89998ee52ef5c175017ac4b)

src/main/java/home/gateway/adapters/influx/InfluxWriter.java

@@ -20,6 +20,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -40,9 +41,11 @@ public final class InfluxWriter implements ReadingSink {
     private final HttpClient http;
     private final Map<String, Spool> spools = new ConcurrentHashMap<>();
     private volatile boolean reachable = true;
+    private volatile int batchSize;
 
     public InfluxWriter(InfluxSettings settings) {
         this.settings = settings;
+        this.batchSize = settings.batchSize();
         this.http = HttpClient.newBuilder().connectTimeout(settings.requestTimeout()).build();
         try {
             Files.createDirectories(settings.spoolDirectory());
@@ -77,27 +80,31 @@ public final class InfluxWriter implements ReadingSink {
     }
 
     private void flushDatabase(String database, Spool spool) {
+        AtomicInteger dropped = new AtomicInteger();
         try {
             Spool.Batch batch;
+            int settled;
             do {
-                batch = spool.peek(settings.batchSize());
-                int settled = deliver(database, batch.lines());
-                spool.remove(batch, settled);
-                if (settled < batch.lines().size()) {
-                    return;
-                }
-            } while (!batch.lines().isEmpty());
+                batch = spool.peek(batchSize);
+                settled = deliver(database, batch.lines(), dropped);
+                spool.settle(batch, settled);
+            } while (settled == batch.lines().size() && !batch.lines().isEmpty());
+            spool.compact();
         } catch (IOException | RuntimeException e) {
             log.error("Cannot flush {}", database, e);
         }
+        if (dropped.get() > 1) {
+            log.warn("Dropped {} more lines InfluxDB rejected for {}", dropped.get() - 1, database);
+        }
     }
 
     /**
      * Returns how many leading lines are settled, delivered or dropped. A rejected batch is
      * split until the bad line is alone, so one bad line cannot cost its neighbours or block
      * the queue; resending accepted points is harmless, as the same series and time overwrite.
+     * 413 only says the request was too big: the batch size shrinks and nothing is dropped.
      */
-    private int deliver(String database, List<String> lines) {
+    private int deliver(String database, List<String> lines, AtomicInteger dropped) {
         if (lines.isEmpty()) {
             return 0;
         }
@@ -109,17 +116,28 @@ public final class InfluxWriter implements ReadingSink {
             return 0;
         }
         if (lines.size() == 1) {
-            log.warn("Dropped a line InfluxDB rejected for {} ({}: {}): {}", database,
-                    answer.status(), answer.body().strip(), lines.get(0));
-            return 1;
+            return answer.status() == 413 ? 0 : drop(database, lines.get(0), answer, dropped);
         }
         int half = lines.size() / 2;
-        int first = deliver(database, lines.subList(0, half));
-        return first < half ? first : half + deliver(database, lines.subList(half, lines.size()));
+        if (answer.status() == 413) {
+            batchSize = Math.min(batchSize, half);
+        }
+        int first = deliver(database, lines.subList(0, half), dropped);
+        return first < half ? first
+                : half + deliver(database, lines.subList(half, lines.size()), dropped);
+    }
+
+    /** Only the first drop of a flush is logged in full, so a systematic conflict cannot flood the journal. */
+    private static int drop(String database, String line, Answer answer, AtomicInteger dropped) {
+        if (dropped.incrementAndGet() == 1) {
+            log.warn("Dropped a line InfluxDB rejected for {} ({}: {}): {}", database,
+                    answer.status(), answer.body().strip(), line);
+        }
+        return 1;
     }
 
     /**
-     * 400, 413 and 422 mean the data is bad. Other 4xx (401, 404, 429) are fixed outside, so
+     * 400, 413 and 422 mean the request is bad. Other 4xx (401, 404, 429) are fixed outside, so
      * their lines wait in the spool, bounded by its cap.
      */
     private static boolean isRejection(int status) {

src/main/java/home/gateway/adapters/influx/Spool.java

@@ -3,37 +3,52 @@ package home.gateway.adapters.influx;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.BufferedInputStream;
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.StandardCopyOption;
 import java.nio.file.StandardOpenOption;
-import java.util.Arrays;
+import java.util.ArrayList;
 import java.util.List;
 
 /**
- * The pending lines of one database, one per row of a file. Every access is serialized and
- * short, so the network is never touched under the lock.
+ * The pending lines of one database, one per row of an append-only file. Draining reads
+ * forward from the first undelivered byte and only remembers how far it got; the delivered
+ * head is cut off once per flush ({@link #compact}), so the cost of a flush is proportional
+ * to what was delivered, not to the size of the spool. A crash forgets the delivered mark and
+ * resends, which is harmless: the same series and time overwrite. Every access is serialized
+ * and short, so the network is never touched under the lock.
  */
 final class Spool {
 
     private static final Logger log = LoggerFactory.getLogger(Spool.class);
 
-    /** The first lines of the spool and the cap's drop count when they were read. */
-    record Batch(List<String> lines, long droppedMark) {
+    /** Cutting down to this share of the cap keeps the cap from rewriting the file on every append. */
+    private static final double CAP_TARGET = 0.9;
+
+    /** Lines read from byte {@code from} of the spool (counted from its first byte ever). */
+    record Batch(List<String> lines, long from) {
     }
 
     private final Path file;
     private final long capBytes;
-    private long droppedByCap;
+    /** Position, in bytes since the spool was opened, of the file's first byte. */
+    private long start;
+    /** Position up to which lines are settled; never behind {@code start}. */
+    private long delivered;
 
     Spool(Path file, long capBytes) throws IOException {
         this.file = file;
         this.capBytes = capBytes;
-        // A crash may leave a torn last line, which would glue onto the next append.
         if (Files.exists(file)) {
-            rewrite(readLines());
+            dropTornLine();
         }
     }
 
@@ -46,51 +61,111 @@ final class Spool {
     }
 
     synchronized Batch peek(int maxLines) throws IOException {
-        List<String> lines = readLines();
-        return new Batch(List.copyOf(lines.subList(0, Math.min(maxLines, lines.size()))),
-                droppedByCap);
+        List<String> lines = new ArrayList<>();
+        if (Files.exists(file)) {
+            try (InputStream in = new BufferedInputStream(Files.newInputStream(file))) {
+                in.skipNBytes(delivered - start);
+                ByteArrayOutputStream line = new ByteArrayOutputStream();
+                for (int b = in.read(); b >= 0 && lines.size() < maxLines; b = in.read()) {
+                    if (b == '\n') {
+                        lines.add(line.toString(StandardCharsets.UTF_8));
+                        line.reset();
+                    } else {
+                        line.write(b);
+                    }
+                }
+            }
+        }
+        return new Batch(List.copyOf(lines), delivered);
     }
 
-    /** Removes the batch's first {@code count} lines, minus those the cap dropped meanwhile. */
-    synchronized void remove(Batch batch, int count) throws IOException {
-        List<String> lines = readLines();
-        long stillThere = count - (droppedByCap - batch.droppedMark());
-        int toRemove = (int) Math.max(0, Math.min(lines.size(), stillThere));
-        rewrite(lines.subList(toRemove, lines.size()));
+    /** Marks the batch's first {@code count} lines as settled; the cap may have cut them already. */
+    synchronized void settle(Batch batch, int count) {
+        long end = batch.from();
+        for (String line : batch.lines().subList(0, count)) {
+            end += line.getBytes(StandardCharsets.UTF_8).length + 1;
+        }
+        delivered = Math.max(delivered, end);
+    }
+
+    /** Cuts the settled head off the file: one copy of the undelivered tail per flush. */
+    synchronized void compact() throws IOException {
+        if (delivered > start) {
+            cutHead(delivered - start);
+        }
     }
 
     private void dropOldest() throws IOException {
-        List<String> lines = readLines();
-        long bytes = Files.size(file);
-        int dropped = 0;
-        while (bytes > capBytes && dropped < lines.size()) {
-            bytes -= lines.get(dropped).getBytes(StandardCharsets.UTF_8).length + 1;
-            dropped++;
-        }
-        droppedByCap += dropped;
-        rewrite(lines.subList(dropped, lines.size()));
-        log.warn("Spool {} is over {} bytes: dropped {} oldest lines", file.getFileName(),
-                capBytes, dropped);
+        long size = Files.size(file);
+        long cut = findLineStart(size - (long) (capBytes * CAP_TARGET));
+        long lost = Math.max(0, start + cut - delivered);
+        cutHead(cut);
+        log.warn("Spool {} is over {} bytes: dropped its oldest {} undelivered bytes",
+                file.getFileName(), capBytes, lost);
     }
 
-    /** Complete lines only; a fragment after the last newline is a torn write. */
-    private List<String> readLines() throws IOException {
-        if (!Files.exists(file)) {
-            return List.of();
+    /** The first line start at or after {@code position}, or the end of the file. */
+    private long findLineStart(long position) throws IOException {
+        try (InputStream in = new BufferedInputStream(Files.newInputStream(file))) {
+            in.skipNBytes(Math.max(0, position - 1));
+            long offset = Math.max(0, position - 1);
+            for (int b = in.read(); b >= 0; b = in.read()) {
+                offset++;
+                if (b == '\n') {
+                    return offset;
+                }
+            }
+            return offset;
         }
-        String text = new String(Files.readAllBytes(file), StandardCharsets.UTF_8);
-        int end = text.lastIndexOf('\n');
-        return end < 0 ? List.of() : Arrays.asList(text.substring(0, end).split("\n"));
     }
 
-    private void rewrite(List<String> lines) throws IOException {
-        if (lines.isEmpty()) {
-            Files.deleteIfExists(file);
-            return;
+    private void cutHead(long bytes) throws IOException {
+        if (bytes >= Files.size(file)) {
+            Files.delete(file);
+        } else {
+            Path temp = file.resolveSibling(file.getFileName() + ".tmp");
+            try (InputStream in = Files.newInputStream(file);
+                    OutputStream out = Files.newOutputStream(temp)) {
+                in.skipNBytes(bytes);
+                in.transferTo(out);
+            }
+            Files.move(temp, file, StandardCopyOption.ATOMIC_MOVE,
+                    StandardCopyOption.REPLACE_EXISTING);
         }
-        Path temp = file.resolveSibling(file.getFileName() + ".tmp");
-        Files.writeString(temp, String.join("\n", lines) + "\n");
-        Files.move(temp, file, StandardCopyOption.ATOMIC_MOVE,
-                StandardCopyOption.REPLACE_EXISTING);
+        start += bytes;
+        delivered = Math.max(delivered, start);
+    }
+
+    /** A crash may leave a fragment after the last newline, which would glue onto the next append. */
+    private void dropTornLine() throws IOException {
+        try (FileChannel channel = FileChannel.open(file, StandardOpenOption.READ,
+                StandardOpenOption.WRITE)) {
+            long end = endOfLastLine(channel);
+            if (end < channel.size()) {
+                channel.truncate(end);
+            }
+        }
+        if (Files.size(file) == 0) {
+            Files.delete(file);
+        }
+    }
+
+    private static long endOfLastLine(FileChannel channel) throws IOException {
+        ByteBuffer chunk = ByteBuffer.allocate(8192);
+        long position = channel.size();
+        while (position > 0) {
+            int length = (int) Math.min(chunk.capacity(), position);
+            position -= length;
+            chunk.clear().limit(length);
+            while (chunk.hasRemaining() && channel.read(chunk, position + chunk.position()) >= 0) {
+                continue;
+            }
+            for (int i = length - 1; i >= 0; i--) {
+                if (chunk.get(i) == '\n') {
+                    return position + i + 1;
+                }
+            }
+        }
+        return 0;
     }
 }

src/test/java/home/gateway/adapters/influx/InfluxWriterTest.java

@@ -20,6 +20,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.function.Function;
+import java.util.stream.IntStream;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -198,15 +199,84 @@ class InfluxWriterTest {
     }
 
     @Test
-    void capDropsTheOldestLines() throws IOException {
-        int lineBytes = line(1, 1).length();
-        InfluxWriter writer = writer(lineBytes * 2L, 5000);
+    void capDropsTheOldestLinesAndKeepsHeadroom() throws IOException {
+        int lineBytes = line(10, 10).length();
+        InfluxWriter writer = writer(lineBytes * 10L, 5000);
         status = request -> 503;
-        writer.write(List.of(reading("db", 1, 1), reading("db", 2, 2), reading("db", 3, 3)));
+        writer.write(IntStream.rangeClosed(10, 20).mapToObj(i -> reading("db", i, i)).toList());
+        assertTrue(spooledBytes() <= lineBytes * 10L);
 
         status = request -> 204;
         writer.flush();
 
-        assertEquals(line(2, 2) + line(3, 3), requests.get(requests.size() - 1).body());
+        String body = requests.get(requests.size() - 1).body();
+        assertTrue(body.startsWith(line(12, 12)));
+        assertTrue(body.endsWith(line(20, 20)));
+    }
+
+    @Test
+    void drainingRewritesTheSpoolOnlyAfterTheLastBatch() throws IOException {
+        InfluxWriter writer = writer(1_000_000, 2);
+        writer.write(IntStream.rangeClosed(1, 5).mapToObj(i -> reading("db", i, i)).toList());
+        long before = spooledBytes();
+        List<Long> sizesDuringFlush = new CopyOnWriteArrayList<>();
+        status = request -> {
+            sizesDuringFlush.add(spoolSize());
+            return 204;
+        };
+
+        writer.flush();
+
+        assertEquals(List.of(before, before, before), sizesDuringFlush);
+        assertEquals(0, spooledBytes());
+    }
+
+    @Test
+    void batchesAfterAPartialDeliveryContinueInOrder() throws IOException {
+        InfluxWriter writer = writer(1_000_000, 2);
+        writer.write(IntStream.rangeClosed(1, 5).mapToObj(i -> reading("db", i, i)).toList());
+        status = request -> request.body().contains(" 3\n") ? 503 : 204;
+        writer.flush();
+        assertEquals(line(3, 3).length() * 3L, spooledBytes());
+
+        status = request -> 204;
+        writer.flush();
+
+        assertEquals(line(3, 3) + line(4, 4), requests.get(requests.size() - 2).body());
+    }
+
+    @Test
+    void tooBigRequestIsHalvedNotDropped() throws IOException {
+        InfluxWriter writer = writer(1_000_000, 8);
+        status = request -> request.body().lines().count() > 2 ? 413 : 204;
+        writer.write(IntStream.rangeClosed(1, 5).mapToObj(i -> reading("db", i, i)).toList());
+
+        writer.flush();
+
+        assertEquals(0, spooledBytes());
+        assertEquals(List.of(line(1, 1) + line(2, 2), line(3, 3), line(4, 4) + line(5, 5)),
+                requests.stream().map(Request::body).filter(body -> body.lines().count() <= 2)
+                        .toList());
+
+        requests.clear();
+        writer.write(IntStream.rangeClosed(6, 10).mapToObj(i -> reading("db", i, i)).toList());
+        writer.flush();
+
+        assertTrue(requests.stream().allMatch(request -> request.body().lines().count() <= 2));
+    }
+
+    @Test
+    void singleLineTooBigStaysInTheSpool() throws IOException {
+        InfluxWriter writer = writer();
+        status = request -> 413;
+        writer.write(List.of(reading("db", 1, 1)));
+
+        writer.flush();
+
+        assertEquals(line(1, 1).length(), spooledBytes());
+    }
+
+    private long spoolSize() {
+        return spool.resolve("db.lp").toFile().length();
     }
 }