test(qwp): pin the orphan drainer's inherited periodic sync interval end to end

The build -> startOrphanDrainers -> BackgroundDrainer -> CursorSendEngine
inheritance was verified-correct but mutation-invisible: passing 0 anywhere
along the chain would drain a PERIODIC slot without checkpoints and still
pass the suite (the only startOrphanDrainers test call sites used the
4-arg overload, which itself hardcodes 0L). The new integration test
adopts a fabricated orphan under sf_durability=periodic against a
never-acking server -- the drain can never finish, so the drainer and its
engine stay deterministically observable in the pool snapshot -- and
asserts the configured interval at the foreground engine, the drainer,
and the drainer's engine, failing on value rather than timeout. Seams:
@testonly drainer-pool accessor on the sender plus engine capture and
inherited-interval getter on the drainer.

Author: bluestreak01

core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java

@@ -1729,6 +1729,15 @@ public CursorSendEngine getCursorEngineForTesting() {
         return cursorEngine;
     }
 
+    /**
+     * Background orphan-drainer pool, or {@code null} when
+     * {@code drain_orphans} is off or no orphan slot was adopted.
+     */
+    @TestOnly
+    public BackgroundDrainerPool getDrainerPoolForTesting() {
+        return drainerPool;
+    }
+
     @TestOnly
     public SenderErrorDispatcher getErrorDispatcherForTesting() {
         return errorDispatcher;

core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/BackgroundDrainer.java

@@ -112,6 +112,12 @@ public final class BackgroundDrainer implements Runnable {
     private final long syncIntervalNanos;
     /** Latest known {@code engine.ackedFsn()}; published for visibility. */
     private volatile long ackedFsn = -1L;
+    /**
+     * Engine constructed by {@link #run()}, captured for test observation
+     * only (e.g. asserting the inherited periodic sync interval). May
+     * reference an already-closed engine once the drain ends.
+     */
+    private volatile CursorSendEngine engineForTesting;
     private volatile String lastErrorMessage;
     /**
      * Optional observer for durable-ack-unavailable transients and the
@@ -517,6 +523,25 @@ public boolean isStopRequested() {
         return stopRequested;
     }
 
+    /**
+     * Engine this drainer constructed, or {@code null} until {@link #run()}
+     * gets past engine construction. The reference outlives the drain, so
+     * tests can read construction-time state (it may be closed by then).
+     */
+    @TestOnly
+    public CursorSendEngine getEngineForTesting() {
+        return engineForTesting;
+    }
+
+    /**
+     * Periodic SF checkpoint interval this drainer inherited from the
+     * adopting sender at construction time.
+     */
+    @TestOnly
+    public long getSyncIntervalNanosForTesting() {
+        return syncIntervalNanos;
+    }
+
     /**
      * Stop check for the runner thread's park loops that also folds a
      * pending thread interrupt into the stop protocol. The pool delivers
@@ -615,6 +640,7 @@ public void run() {
                 outcome = DrainOutcome.FAILED;
                 return;
             }
+            engineForTesting = engine;
             long target = engine.publishedFsn();
             if (engine.ackedFsn() >= target) {
                 LOG.info("orphan slot already drained: {} (acked={} target={})",

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/OrphanScanIntegrationTest.java

@@ -25,6 +25,9 @@
 package io.questdb.client.test.cutlass.qwp.client.sf;
 
 import io.questdb.client.Sender;
+import io.questdb.client.cutlass.qwp.client.QwpWebSocketSender;
+import io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainer;
+import io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainerPool;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.OrphanScanner;
 import io.questdb.client.std.Files;
 import io.questdb.client.std.ObjList;
@@ -148,6 +151,107 @@ public void testScanFindsOrphanFromPriorSenderUnderSameGroupRoot() throws Except
         });
     }
 
+    /**
+     * The orphan drainer must inherit the adopting sender's PERIODIC
+     * store-and-forward checkpoint interval end to end:
+     * {@code Sender.build → startOrphanDrainers → BackgroundDrainer →
+     * CursorSendEngine}. A {@code 0} anywhere along that chain would make
+     * the drainer silently replay a PERIODIC slot without periodic
+     * checkpoints — no error, no log — which is exactly the mutation this
+     * test exists to catch.
+     */
+    @Test
+    public void testOrphanDrainerInheritsForegroundPeriodicSyncInterval() throws Exception {
+        TestUtils.assertMemoryLeak(() -> {
+            // Phase 1: fabricate an orphan slot holding unacked backlog
+            // (same recipe as
+            // testScanFindsOrphanFromPriorSenderUnderSameGroupRoot).
+            SilentHandler ghostSilent = new SilentHandler();
+            try (TestWebSocketServer ghostServer = new TestWebSocketServer(ghostSilent)) {
+                ghostServer.start();
+                Assert.assertTrue(ghostServer.awaitStart(5, TimeUnit.SECONDS));
+
+                String ghostCfg = "ws::addr=localhost:" + ghostServer.getPort()
+                        + ";sf_dir=" + sfDir + ";sender_id=ghost;close_flush_timeout_millis=0;";
+                try (Sender ghost = Sender.fromConfig(ghostCfg)) {
+                    ghost.table("foo").longColumn("v", 7L).atNow();
+                    ghost.flush();
+                    Assert.assertTrue("ghost frame must reach the wire before close",
+                            ghostSilent.awaitFrame(5, TimeUnit.SECONDS));
+                }
+            }
+            Assert.assertEquals("ghost slot must be a candidate orphan",
+                    1, OrphanScanner.scan(sfDir, "primary").size());
+
+            // Phase 2: adopt with a PERIODIC-durability sender. The server
+            // never acks, so the adopted slot can never finish draining —
+            // the drainer (and its engine) stay pinned in the pool snapshot
+            // while we assert the inheritance chain. The asserts below fail
+            // on the VALUE, not on a timeout, when inheritance breaks.
+            try (TestWebSocketServer primaryServer = new TestWebSocketServer(new SilentHandler())) {
+                primaryServer.start();
+                Assert.assertTrue(primaryServer.awaitStart(5, TimeUnit.SECONDS));
+
+                String primaryCfg = "ws::addr=localhost:" + primaryServer.getPort()
+                        + ";sf_dir=" + sfDir
+                        + ";sender_id=primary"
+                        + ";sf_durability=periodic"
+                        + ";sf_sync_interval_millis=123"
+                        + ";drain_orphans=true;";
+                try (Sender primary = Sender.fromConfig(primaryCfg)) {
+                    QwpWebSocketSender ws = (QwpWebSocketSender) primary;
+                    // Anchor: the foreground engine carries the configured
+                    // interval (already pinned by SfFromConfigTest; repeated
+                    // here so both ends of the inheritance are asserted on
+                    // the same sender instance).
+                    Assert.assertEquals(123_000_000L,
+                            ws.getCursorEngineForTesting().getSyncIntervalNanosForTesting());
+
+                    BackgroundDrainerPool pool = ws.getDrainerPoolForTesting();
+                    Assert.assertNotNull("a candidate orphan under drain_orphans=true "
+                            + "must have built the drainer pool", pool);
+
+                    // The drainer constructs its engine on its own thread —
+                    // await it. Once seen it cannot leave the snapshot: with
+                    // no acks the drain never completes.
+                    BackgroundDrainer drainer = null;
+                    long deadlineNanos = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+                    while (System.nanoTime() < deadlineNanos) {
+                        ObjList<BackgroundDrainer> snap = pool.snapshot();
+                        if (snap.size() > 0 && snap.get(0).getEngineForTesting() != null) {
+                            drainer = snap.get(0);
+                            break;
+                        }
+                        Thread.sleep(10);
+                    }
+                    Assert.assertNotNull(
+                            "orphan drainer must construct its engine within the deadline",
+                            drainer);
+                    Assert.assertEquals("exactly one orphan slot was fabricated",
+                            1, pool.snapshot().size());
+
+                    // The carried finding: sender → drainer leg.
+                    Assert.assertEquals(
+                            "drainer must inherit the adopting sender's "
+                                    + "sf_sync_interval_millis",
+                            123_000_000L, drainer.getSyncIntervalNanosForTesting());
+                    // Drainer → engine leg: the interval actually reached the
+                    // engine that performs the periodic checkpoints.
+                    Assert.assertEquals(
+                            "drainer's engine must inherit the adopting sender's "
+                                    + "sf_sync_interval_millis",
+                            123_000_000L,
+                            drainer.getEngineForTesting().getSyncIntervalNanosForTesting());
+
+                    // Fast teardown: the drainer can never finish against the
+                    // silent server — stop it now so close() spends the
+                    // graceful-drain window on unwinding, not waiting.
+                    drainer.requestStop();
+                }
+            }
+        });
+    }
+
     @Test
     public void testFailedSentinelHidesOrphanFromScan() throws Exception {
         TestUtils.assertMemoryLeak(() -> {