feat: bring the MemWAL LSM surface to parity across the SDKs (#3962)

## Why Four of the eight LSM methods are **remote-only in the core**. `impl BaseTable for NativeTable` implements only `set`/`unset`/`get_lsm_write_spec` and `close_lsm_writers`; `flush_lsm`, `compact_lsm` and `get_lsm_stats` fall through to trait defaults returning `NotSupported` (`rust/lancedb/src/table.rs:679,687,696`), and `checkpoint_lsm` is built on all three. That explains the state of the bindings: Node had bound the four that work against a local table and stopped, so a Cloud user could install an LSM write spec but had no way to observe fresh-tier state or drive a checkpoint. Java had none of it at all. | SDK | set/unset/get spec | closeWriters | flush | compact | getStats | checkpoint | |---|---|---|---|---|---|---| | Rust core | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | | Python | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | | Node *(before)* | ✅ | ✅ | — | — | — | — | | **Node (after)** | ✅ | ✅ | **new** | **new** | **new** | **new** | | Java *(before)* | — | — | — | — | — | — | | **Java (after)** | **new** | n/a | **new** | **new** | **new** | **new** | Go and C are separate repos and are out of scope here. `closeLsmWriters` drains cached in-process shard writers, so it has no meaning for Java, which is a pure REST client. ## Node Adds napi bindings for `flushLsm`, `compactLsm`, `checkpointLsm` and `getLsmStats`, plus typed `LsmStats` / `BucketStats` / `GenerationStats` / `MemtableStats` objects — typed rather than a JSON blob, matching the existing `LsmWriteSpec` object in the same file, with `u64` cast to `i64` per that file's convention. Because these four are remote-only, the new tests assert each binding reaches the core and surfaces `NotSupported` against a local table. That covers the wiring; behavior against a real endpoint stays covered by the mocked-endpoint tests in `rust/lancedb/src/remote/table.rs`. ## Python No new methods. All eight are on `LanceTable`, `AsyncTable` and `RemoteTable` — the last four landed on the sync `RemoteTable` in #3961, which is merged into this branch. What was missing here was reachability. `LsmWriteSpec` was importable only from the private `lancedb._lancedb`, appearing in `table.py` solely under `if TYPE_CHECKING:`, and `docs/src/python/python.md` had no mention of it, which per the repo's docs guidance means it rendered nowhere in the API reference. It is now `lancedb.LsmWriteSpec`, in `__all__`, and documented. ## Java Java reaches LanceDB purely over REST through the generated Lance Namespace client, and these routes are not in that spec, so they are issued through a small dedicated client rather than added to the spec. That call is revisitable — LSM is one of four unspecified route families alongside `multipart_write`, `page_cache/prewarm` and `branches/diff|merge`. If those are ever regularized into the spec as a group, `LanceDbTableLsm` is one file that gets deleted. `LsmWriteSpec` here is deliberately **not** `org.lance.memwal.InitializeMemWalParams`. That type defaults to maintaining *no* indexes where a spec here defaults to maintaining *every* index, and it cannot express the `null` that asks the server to resolve the set: | Value | On the wire | Meaning | |---|---|---| | unset (null) | `null` | Server resolves **every** maintainable index | | `Collections.emptyList()` | `[]` | Maintain **none** | | `Arrays.asList("id_idx")` | `["id_idx"]` | Exactly those | A dedicated test pins null and `[]` as distinct on the wire, since collapsing them is the failure mode that motivated a LanceDB-owned type. `checkpointLsm` is ported from `rust/lancedb/src/table/checkpoint.rs` with its constants and status semantics intact: 429/503 retried in place against an 8-budget, 421 restarting from flush against a 3-budget, 5s poll, and a target watermark fixed after the seal so it terminates under write load. `getLsmStats` returns typed `LsmStats` / `BucketStats` / `GenerationStats` / `MemtableStats`, mirroring the Rust structs in `rust/lancedb/src/table/lsm_stats.rs` and the objects Node exposes. Decoding is strict — see below. ## Review feedback Both gatekeeper findings were real. Each was reproduced against the scripted test server first, and each fix ships with the reproducer as a regression test. **The transport was doubling every checkpoint retry budget.** `HttpClients.createDefault()` installs Apache's default response retry strategy, whose retryable-status list is exactly 429 and 503 — the two statuses `isRetryable` owns. A 429 held against `flush_lsm` issued **18** wire requests where the loop intends 9, and `compact_lsm` was retried in place despite the loop being built to fall through to a fresh stats poll instead. Timing confirmed the mechanism: that run took 25.4s ≈ 16.3s of the loop's own backoff plus 9 × the transport's 1s retry interval. Automatic retries are now disabled, so the checkpoint loop is the sole owner of the 421/429/503 transitions. A side effect worth noting: `testCheckpointRetriesRetryableStatusInPlace` was passing on a transport-absorbed 429 and never reaching `issue()`'s retry branch at all. It now exercises the real path. **Stats decoding failed open.** `getLsmStats` read the response with Jackson's `path()`, which yields a missing node that iterates as an empty array — making "malformed" indistinguishable from "no buckets", which is indistinguishable from "drained". Four separate payloads made `checkpointLsm()` report convergence for a checkpoint that never ran: | Response | Before | Now | |---|---|---| | `{"lsm_stats": null}` or absent key | disabled ✓ | disabled ✓ | | `{"lsm_stats": {}}` | **reported success** | `IllegalStateException` | | empty response body | **reported success** | `IllegalStateException` | | bucket missing required fields | **reported success** | `IllegalStateException` | The empty-body row is the one to weight: a proxy 200 with no body is a realistic production event, and it silently reported a checkpoint that never happened. Decoding is now strict and fails closed, matching the serde contract on the Rust side exactly. One deliberate deviation from the review comment, which asked that *only* explicit JSON `null` count as disabled: Rust has `#[serde(default)]` on `lsm_stats`, so an **absent key** decodes to `None` there too. Java now matches that. It is an absent-or-malformed **`buckets`** that fails closed, which is the case the comment was actually protecting. ## Testing - Java: **33 passing** (8 existing + 25 LSM) against a scripted `com.sun.net.httpserver.HttpServer` — no new test dependency. Wire assertions mirror `rust/lancedb/src/remote/table.rs:6581-6748`; checkpoint tests cover convergence, not piling onto a latched bucket, 421 restart-from-flush, 429 retry-in-place, terminal-status propagation, reissue exhaustion, the exact wire-request count against the retry budget, and five malformed stats payloads. - Node: **19 LSM tests passing**; `cargo check`, `npm run build`, `npm run tsc`, `npm run lint`, `npm run docs` all clean. - Python: `ruff format --check` and `ruff check` clean. - Java formatting: `./mvnw -pl lancedb-core spotless:apply` and `spotless:check` both clean under a JDK 11 toolchain. ## Note: spotless needs a pre-16 JDK `./mvnw spotless:apply` fails on JDK 16+ with `JCTree$JCImport.getQualifiedIdentifier()` — google-java-format 1.7, pinned at `java/pom.xml:34`, predates JDK 16's compiler API change. **This is pre-existing** and reproduces on a pristine `main` checkout. It is not a blocker, just a toolchain requirement. Spotless was run against these sources under JDK 11 and both `spotless:apply` and `spotless:check` pass on the whole module: ```shell JAVA_HOME=/path/to/jdk11 ./mvnw -pl lancedb-core spotless:apply ``` Bumping the plugin so it works on modern JDKs is still worth doing, but separately from this PR. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-23 14:38:35 +00:00 · 2026-08-19 11:44:46 -05:00
parent 11c1d81638
commit f1c4967eeb
25 changed files with 2581 additions and 16 deletions
@@ -0,0 +1,570 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.lancedb;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.sun.net.httpserver.HttpServer;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.UncheckedIOException;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * Unit tests for the MemWAL LSM routes, run against a scripted local HTTP server.
+ *
+ * <p>The wire assertions mirror the Rust mocked-endpoint tests in {@code
+ * rust/lancedb/src/remote/table.rs}, which are the contract these routes have to match.
+ */
+public class LanceDbTableLsmTest {
+  private static final ObjectMapper MAPPER = new ObjectMapper();
+
+  private HttpServer server;
+  private LanceDbRestClient client;
+  private LanceDbTableLsm lsm;
+
+  private final List<String> requestPaths = Collections.synchronizedList(new ArrayList<String>());
+  private final List<String> requestBodies = Collections.synchronizedList(new ArrayList<String>());
+  private final Map<String, Deque<Reply>> replies = new ConcurrentHashMap<String, Deque<Reply>>();
+
+  @BeforeEach
+  public void setUp() throws IOException {
+    start();
+  }
+
+  /** Tear down and restart the scripted server, for a test that scripts several exchanges. */
+  private void setUpFresh() {
+    try {
+      client.close();
+      server.stop(0);
+      requestPaths.clear();
+      requestBodies.clear();
+      replies.clear();
+      start();
+    } catch (IOException e) {
+      throw new UncheckedIOException(e);
+    }
+  }
+
+  private void start() throws IOException {
+    server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+    server.createContext(
+        "/",
+        exchange -> {
+          String path = exchange.getRequestURI().getPath();
+          requestPaths.add(path);
+          requestBodies.add(readAll(exchange.getRequestBody()));
+
+          Reply reply = nextReply(path);
+          byte[] out = reply.body.getBytes(StandardCharsets.UTF_8);
+          exchange.sendResponseHeaders(reply.status, out.length == 0 ? -1 : out.length);
+          if (out.length > 0) {
+            exchange.getResponseBody().write(out);
+          }
+          exchange.close();
+        });
+    server.start();
+
+    client =
+        LanceDbNamespaceClientBuilder.newBuilder()
+            .apiKey("test-key")
+            .database("test-db")
+            .endpoint("http://127.0.0.1:" + server.getAddress().getPort())
+            .buildRestClient();
+    lsm = new LanceDbTableLsm(client, "my_table");
+  }
+
+  @AfterEach
+  public void tearDown() throws IOException {
+    client.close();
+    server.stop(0);
+  }
+
+  // ===========================================================================
+  // set / unset / get spec
+  // ===========================================================================
+
+  @Test
+  public void testSetLsmWriteSpecUnsharded() throws Exception {
+    enqueue("set_lsm_write_spec", 200, "");
+
+    lsm.setLsmWriteSpec(LsmWriteSpec.unsharded());
+
+    assertEquals("/v1/table/my_table/set_lsm_write_spec/", requestPaths.get(0));
+    JsonNode body = MAPPER.readTree(requestBodies.get(0));
+    assertEquals("unsharded", body.get("sharding").get("mode").asText());
+    assertFalse(body.get("sharding").has("column"));
+    assertFalse(body.get("sharding").has("num_buckets"));
+  }
+
+  @Test
+  public void testSetLsmWriteSpecBucket() throws Exception {
+    enqueue("set_lsm_write_spec", 200, "");
+
+    lsm.setLsmWriteSpec(
+        LsmWriteSpec.bucket("id", 16).withMaintainedIndexes(Arrays.asList("id_idx")));
+
+    JsonNode body = MAPPER.readTree(requestBodies.get(0));
+    assertEquals("bucket", body.get("sharding").get("mode").asText());
+    assertEquals("id", body.get("sharding").get("column").asText());
+    assertEquals(16, body.get("sharding").get("num_buckets").asInt());
+    assertEquals(1, body.get("maintained_indexes").size());
+    assertEquals("id_idx", body.get("maintained_indexes").get(0).asText());
+  }
+
+  @Test
+  public void testSetLsmWriteSpecIdentity() throws Exception {
+    enqueue("set_lsm_write_spec", 200, "");
+
+    lsm.setLsmWriteSpec(LsmWriteSpec.identity("tenant"));
+
+    JsonNode body = MAPPER.readTree(requestBodies.get(0));
+    assertEquals("identity", body.get("sharding").get("mode").asText());
+    assertEquals("tenant", body.get("sharding").get("column").asText());
+    assertFalse(body.get("sharding").has("num_buckets"));
+  }
+
+  /**
+   * The tri-state that motivated a LanceDB-owned spec type: a null selection asks the server to
+   * resolve every maintainable index, while an empty list asks for none. They must not collapse.
+   */
+  @Test
+  public void testMaintainedIndexesNullAndEmptyAreDistinctOnTheWire() throws Exception {
+    enqueue("set_lsm_write_spec", 200, "");
+
+    lsm.setLsmWriteSpec(LsmWriteSpec.unsharded());
+    JsonNode fresh = MAPPER.readTree(requestBodies.get(0));
+    assertTrue(fresh.has("maintained_indexes"), "the key must be present");
+    assertTrue(fresh.get("maintained_indexes").isNull(), "a fresh spec sends null, not []");
+
+    lsm.setLsmWriteSpec(
+        LsmWriteSpec.unsharded().withMaintainedIndexes(Collections.<String>emptyList()));
+    JsonNode none = MAPPER.readTree(requestBodies.get(1));
+    assertTrue(none.get("maintained_indexes").isArray());
+    assertEquals(0, none.get("maintained_indexes").size());
+  }
+
+  @Test
+  public void testSetLsmWriteSpecWriterConfigDefaults() throws Exception {
+    enqueue("set_lsm_write_spec", 200, "");
+
+    Map<String, String> defaults = new HashMap<String, String>();
+    defaults.put("max_memtable_rows", "50000");
+    lsm.setLsmWriteSpec(LsmWriteSpec.unsharded().withWriterConfigDefaults(defaults));
+
+    JsonNode body = MAPPER.readTree(requestBodies.get(0));
+    assertEquals("50000", body.get("writer_config_defaults").get("max_memtable_rows").asText());
+  }
+
+  @Test
+  public void testUnsetLsmWriteSpec() {
+    enqueue("unset_lsm_write_spec", 200, "");
+
+    lsm.unsetLsmWriteSpec();
+
+    assertEquals("/v1/table/my_table/unset_lsm_write_spec/", requestPaths.get(0));
+    assertEquals("", requestBodies.get(0));
+  }
+
+  @Test
+  public void testGetLsmWriteSpec() {
+    enqueue(
+        "get_lsm_write_spec",
+        200,
+        "{\"lsm_write_spec\":{\"sharding\":{\"mode\":\"bucket\",\"column\":\"id\","
+            + "\"num_buckets\":16},\"maintained_indexes\":[\"id_idx\"],"
+            + "\"writer_config_defaults\":{\"durable_write\":\"true\"}}}");
+
+    Optional<LsmWriteSpec> spec = lsm.getLsmWriteSpec();
+
+    assertTrue(spec.isPresent());
+    assertEquals(LsmWriteSpec.Sharding.BUCKET, spec.get().sharding());
+    assertEquals("id", spec.get().column());
+    assertEquals(Integer.valueOf(16), spec.get().numBuckets());
+    assertEquals(Arrays.asList("id_idx"), spec.get().maintainedIndexes());
+    assertEquals("true", spec.get().writerConfigDefaults().get("durable_write"));
+  }
+
+  @Test
+  public void testGetLsmWriteSpecAbsent() {
+    enqueue("get_lsm_write_spec", 200, "{\"lsm_write_spec\":null}");
+
+    assertFalse(lsm.getLsmWriteSpec().isPresent());
+  }
+
+  // ===========================================================================
+  // stats
+  // ===========================================================================
+
+  @Test
+  public void testGetLsmStats() throws Exception {
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false, 7L, 8L)));
+
+    Optional<LsmStats> got = lsm.getLsmStats(true);
+
+    assertEquals("/v1/table/my_table/get_lsm_stats/", requestPaths.get(0));
+    assertTrue(MAPPER.readTree(requestBodies.get(0)).get("include_generation_rows").asBoolean());
+    assertTrue(got.isPresent());
+    BucketStats decoded = got.get().buckets().get(0);
+    assertEquals("shard-0", decoded.shardId());
+    assertEquals("Active", decoded.status());
+    assertEquals(1, decoded.writerEpoch());
+    assertEquals(2, decoded.manifestVersion());
+    assertEquals(9, decoded.currentGeneration());
+    assertFalse(decoded.compacting());
+    assertEquals(Arrays.asList(7L, 8L), generationNumbers(decoded));
+    assertEquals(1024, decoded.generations().get(0).bytes());
+    assertFalse(decoded.generations().get(0).rows().isPresent(), "rows absent unless requested");
+    assertFalse(decoded.memtables().isPresent(), "absent memtables stay absent");
+  }
+
+  /** The optional fields decode when the server does send them. */
+  @Test
+  public void testGetLsmStatsDecodesOptionalFields() {
+    enqueue(
+        "get_lsm_stats",
+        200,
+        "{\"lsm_stats\":{\"buckets\":[{\"shard_id\":\"shard-0\",\"status\":\"Active\","
+            + "\"writer_epoch\":1,\"manifest_version\":2,\"current_generation\":9,"
+            + "\"replay_after_wal_entry_position\":3,\"wal_entry_position_last_seen\":11,"
+            + "\"generations\":[{\"generation\":7,\"bytes\":1024,\"rows\":42}],"
+            + "\"compacting\":true,\"memtables\":[{\"generation\":8,\"rows\":5,"
+            + "\"bytes\":64,\"batches\":2,\"indexes\":[\"id_idx\"]}]}]}}");
+
+    BucketStats decoded = lsm.getLsmStats(true).get().buckets().get(0);
+
+    assertEquals(3, decoded.replayAfterWalEntryPosition());
+    assertEquals(11, decoded.walEntryPositionLastSeen());
+    assertTrue(decoded.compacting());
+    assertEquals(42, decoded.generations().get(0).rows().getAsLong());
+    assertTrue(decoded.memtables().isPresent());
+    MemtableStats memtable = decoded.memtables().get().get(0);
+    assertEquals(8, memtable.generation());
+    assertEquals(5, memtable.rows());
+    assertEquals(64, memtable.bytes());
+    assertEquals(2, memtable.batches());
+    assertEquals(Arrays.asList("id_idx"), memtable.indexes());
+  }
+
+  @Test
+  public void testGetLsmStatsAbsentWhenLsmDisabled() {
+    enqueue("get_lsm_stats", 200, "{\"lsm_stats\":null}");
+
+    assertFalse(lsm.getLsmStats().isPresent());
+  }
+
+  @Test
+  public void testGetLsmStatsDefaultsToExcludingGenerationRows() throws Exception {
+    enqueue("get_lsm_stats", 200, stats());
+
+    lsm.getLsmStats();
+
+    assertFalse(MAPPER.readTree(requestBodies.get(0)).get("include_generation_rows").asBoolean());
+  }
+
+  // ===========================================================================
+  // flush / compact
+  // ===========================================================================
+
+  @Test
+  public void testFlushAndCompactRoutes() {
+    enqueue("flush_lsm", 200, "");
+    enqueue("compact_lsm", 200, "");
+
+    lsm.flushLsm();
+    lsm.compactLsm();
+
+    assertEquals("/v1/table/my_table/flush_lsm/", requestPaths.get(0));
+    assertEquals("/v1/table/my_table/compact_lsm/", requestPaths.get(1));
+  }
+
+  @Test
+  public void testHttpErrorCarriesStatus() {
+    enqueue("flush_lsm", 404, "no such table");
+
+    LanceDbRestClient.HttpException e =
+        assertThrows(LanceDbRestClient.HttpException.class, () -> lsm.flushLsm());
+    assertEquals(404, e.statusCode());
+  }
+
+  // ===========================================================================
+  // checkpoint
+  // ===========================================================================
+
+  @Test
+  public void testCheckpointReturnsWhenLsmDisabled() {
+    enqueue("flush_lsm", 200, "");
+    enqueue("get_lsm_stats", 200, "{\"lsm_stats\":null}");
+
+    lsm.checkpointLsm();
+
+    assertEquals(0, countCalls("compact_lsm"), "nothing to compact when the LSM path is off");
+  }
+
+  @Test
+  public void testCheckpointReturnsWhenNoGenerationsOutstanding() {
+    enqueue("flush_lsm", 200, "");
+    // A bucket with no L0 generations yields no target, so the drain never starts.
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false)));
+
+    lsm.checkpointLsm();
+
+    assertEquals(0, countCalls("compact_lsm"));
+  }
+
+  @Test
+  public void testCheckpointConvergesOnceTargetGenerationsAreGone() {
+    enqueue("flush_lsm", 200, "");
+    // Watermark read: shard-0 holds generations 7 and 8, so target = 8.
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false, 7L, 8L)));
+    // First drain poll: both still outstanding, nothing compacting -> dispatch a pass.
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false, 7L, 8L)));
+    // Second drain poll: drained past the target -> done.
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false, 9L)));
+    enqueue("compact_lsm", 200, "");
+
+    lsm.checkpointLsm();
+
+    assertEquals(1, countCalls("compact_lsm"), "one pass dispatched");
+    assertEquals(3, countCalls("get_lsm_stats"), "watermark read plus two drain polls");
+  }
+
+  @Test
+  public void testCheckpointDoesNotPileOnWhileEveryTargetBucketIsCompacting() {
+    enqueue("flush_lsm", 200, "");
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", true, 4L)));
+    // Still compacting on the first poll, so no pass is dispatched; then it drains.
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", true, 4L)));
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false, 5L)));
+
+    lsm.checkpointLsm();
+
+    assertEquals(0, countCalls("compact_lsm"), "a latched bucket is left alone");
+  }
+
+  @Test
+  public void testCheckpointRetriesFromFlushAfterLostClaim() {
+    // 421 on the watermark read: the node lost its claim, so the whole thing restarts
+    // from flush rather than retrying the read in place.
+    enqueue("flush_lsm", 200, "");
+    enqueue("get_lsm_stats", 421, "no claim");
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false)));
+
+    lsm.checkpointLsm();
+
+    assertEquals(2, countCalls("flush_lsm"), "re-issued from flush");
+  }
+
+  @Test
+  public void testCheckpointRetriesRetryableStatusInPlace() {
+    enqueue("flush_lsm", 429, "latch held");
+    enqueue("flush_lsm", 200, "");
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false)));
+
+    lsm.checkpointLsm();
+
+    assertEquals(2, countCalls("flush_lsm"), "429 retried in place, not re-issued");
+  }
+
+  @Test
+  public void testCheckpointPropagatesTerminalStatus() {
+    enqueue("flush_lsm", 400, "bad request");
+
+    LanceDbRestClient.HttpException e =
+        assertThrows(LanceDbRestClient.HttpException.class, () -> lsm.checkpointLsm());
+    assertEquals(400, e.statusCode());
+    assertEquals(1, countCalls("flush_lsm"), "a terminal status is not retried");
+  }
+
+  @Test
+  public void testCheckpointGivesUpAfterRepeatedLostClaims() {
+    enqueue("flush_lsm", 421, "no claim");
+
+    IllegalStateException e = assertThrows(IllegalStateException.class, () -> lsm.checkpointLsm());
+    assertTrue(e.getMessage().contains("kept losing its claim"), e.getMessage());
+    assertEquals(4, countCalls("flush_lsm"), "the initial attempt plus MAX_REISSUES");
+  }
+
+  // ===========================================================================
+  // strict decoding
+  // ===========================================================================
+
+  /**
+   * A stats payload that does not decode must fail closed. Every one of these bodies used to be
+   * read as "no buckets", which is indistinguishable from a drained table, so {@code checkpointLsm}
+   * reported convergence for a checkpoint that never ran.
+   */
+  @Test
+  public void testCheckpointRejectsMalformedStats() {
+    Map<String, String> malformed = new LinkedHashMap<String, String>();
+    malformed.put("no response body at all", "");
+    malformed.put("stats object with no buckets", "{\"lsm_stats\":{}}");
+    malformed.put("bucket missing its required fields", "{\"lsm_stats\":{\"buckets\":[{}]}}");
+    malformed.put(
+        "bucket missing generations",
+        "{\"lsm_stats\":{\"buckets\":[{\"shard_id\":\"shard-0\",\"status\":\"Active\","
+            + "\"writer_epoch\":1,\"manifest_version\":2,\"current_generation\":9,"
+            + "\"replay_after_wal_entry_position\":0,\"wal_entry_position_last_seen\":0,"
+            + "\"compacting\":false}]}}");
+    malformed.put(
+        "generation with a non-numeric generation number",
+        "{\"lsm_stats\":{\"buckets\":[{\"shard_id\":\"shard-0\",\"status\":\"Active\","
+            + "\"writer_epoch\":1,\"manifest_version\":2,\"current_generation\":9,"
+            + "\"replay_after_wal_entry_position\":0,\"wal_entry_position_last_seen\":0,"
+            + "\"generations\":[{\"generation\":\"7\",\"bytes\":1024}],"
+            + "\"compacting\":false}]}}");
+
+    for (Map.Entry<String, String> each : malformed.entrySet()) {
+      setUpFresh();
+      enqueue("flush_lsm", 200, "");
+      enqueue("get_lsm_stats", 200, each.getValue());
+
+      assertThrows(
+          IllegalStateException.class,
+          () -> lsm.checkpointLsm(),
+          each.getKey() + " must not report convergence");
+    }
+  }
+
+  /** The one shape that legitimately means "this table has no LSM write path". */
+  @Test
+  public void testCheckpointTreatsNullStatsAsNotWalBacked() {
+    enqueue("flush_lsm", 200, "");
+    enqueue("get_lsm_stats", 200, "{\"lsm_stats\":null}");
+
+    lsm.checkpointLsm();
+
+    assertEquals(1, countCalls("get_lsm_stats"));
+  }
+
+  // ===========================================================================
+  // retry budget
+  // ===========================================================================
+
+  /**
+   * The transport must not retry on the checkpoint loop's behalf. Apache HttpClient's default
+   * strategy retries exactly 429 and 503 — the two statuses {@code isRetryable} owns — which
+   * doubled every budget here and also retried {@code compact_lsm} in place, where the loop is
+   * built to fall through to a fresh stats poll instead.
+   */
+  @Test
+  public void testCheckpointRetryBudgetIsNotDoubledByTheTransport() {
+    enqueue("flush_lsm", 429, "latch held");
+
+    LanceDbRestClient.HttpException e =
+        assertThrows(LanceDbRestClient.HttpException.class, () -> lsm.checkpointLsm());
+
+    assertEquals(429, e.statusCode(), "the exhausted budget propagates the last error as itself");
+    assertEquals(9, countCalls("flush_lsm"), "the initial request plus MAX_RETRIES, and no more");
+  }
+
+  // ===========================================================================
+  // harness
+  // ===========================================================================
+
+  private static List<Long> generationNumbers(BucketStats bucket) {
+    List<Long> numbers = new ArrayList<Long>();
+    for (GenerationStats generation : bucket.generations()) {
+      numbers.add(generation.generation());
+    }
+    return numbers;
+  }
+
+  /** Build an {@code lsm_stats} response body from bucket fragments. */
+  private static String stats(String... buckets) {
+    return "{\"lsm_stats\":{\"buckets\":[" + String.join(",", buckets) + "]}}";
+  }
+
+  private static String bucket(String shardId, boolean compacting, Long... generations) {
+    StringBuilder gens = new StringBuilder();
+    for (Long generation : generations) {
+      if (gens.length() > 0) {
+        gens.append(",");
+      }
+      gens.append("{\"generation\":").append(generation).append(",\"bytes\":1024}");
+    }
+    return "{\"shard_id\":\""
+        + shardId
+        + "\",\"status\":\"Active\",\"writer_epoch\":1,\"manifest_version\":2,"
+        + "\"current_generation\":9,\"replay_after_wal_entry_position\":0,"
+        + "\"wal_entry_position_last_seen\":0,\"generations\":["
+        + gens
+        + "],\"compacting\":"
+        + compacting
+        + "}";
+  }
+
+  /** Queue a reply for an operation. The last queued reply repeats once the queue drains. */
+  private void enqueue(String operation, int status, String body) {
+    replies.computeIfAbsent(operation, key -> new ArrayDeque<Reply>()).add(new Reply(status, body));
+  }
+
+  private Reply nextReply(String path) {
+    String operation = operationOf(path);
+    Deque<Reply> queued = replies.get(operation);
+    if (queued == null || queued.isEmpty()) {
+      return new Reply(200, "");
+    }
+    return queued.size() > 1 ? queued.poll() : queued.peek();
+  }
+
+  private long countCalls(String operation) {
+    return requestPaths.stream().filter(path -> operationOf(path).equals(operation)).count();
+  }
+
+  /** {@code /v1/table/my_table/flush_lsm/} -> {@code flush_lsm}. */
+  private static String operationOf(String path) {
+    String[] segments = path.split("/");
+    return segments.length == 0 ? "" : segments[segments.length - 1];
+  }
+
+  private static String readAll(InputStream in) throws IOException {
+    ByteArrayOutputStream out = new ByteArrayOutputStream();
+    byte[] buffer = new byte[4096];
+    int read;
+    while ((read = in.read(buffer)) != -1) {
+      out.write(buffer, 0, read);
+    }
+    return new String(out.toByteArray(), StandardCharsets.UTF_8);
+  }
+
+  private static final class Reply {
+    private final int status;
+    private final String body;
+
+    private Reply(int status, String body) {
+      this.status = status;
+      this.body = body;
+    }
+  }
+}