@@ -1025,6 +1025,15 @@ final class LineSenderBuilder {
10251025 // runtime lands in a follow-up commit. For now we surface the
10261026 // count via logging so users can confirm orphans are being seen.
10271027 private boolean drainOrphans = false ;
1028+ // Orphan-scan exclusion for the connection pool. The pool co-manages
1029+ // exactly <orphanDrainBase>-<i> for i in [0, orphanDrainSlotCount) and
1030+ // recovers each of those on (re)creation, so pooled senders must never
1031+ // treat one another's live slots as drainable orphans. Anything else --
1032+ // a different base, a bare un-suffixed id, OR a same-base index at or
1033+ // above the count (a slot left behind by a larger pool before maxSize
1034+ // shrank) -- is still drained, so unacked data is never stranded.
1035+ private String orphanDrainBase ;
1036+ private int orphanDrainSlotCount ;
10281037 private long durableAckKeepaliveIntervalMillis = DURABLE_ACK_KEEPALIVE_NOT_SET ;
10291038 // Optional user-supplied async error handler. When null, the sender
10301039 // uses DefaultSenderErrorHandler.INSTANCE (loud-not-silent log).
@@ -1469,7 +1478,15 @@ public Sender build() {
14691478 } else {
14701479 if (!Files .exists (sfDir )) {
14711480 int rc = Files .mkdir (sfDir , Files .DIR_MODE_DEFAULT );
1472- if (rc != 0 ) {
1481+ // mkdir is non-zero on failure, but "already exists"
1482+ // is one such failure. Multiple SF senders sharing one
1483+ // sf_dir can be built concurrently (the pool calls
1484+ // build() outside its lock), so two threads can both
1485+ // pass the exists() check and race into mkdir; the
1486+ // loser gets EEXIST. Treat a benign creation race --
1487+ // the dir now exists -- as success and only fail when
1488+ // the directory is genuinely absent afterwards.
1489+ if (rc != 0 && !Files .exists (sfDir )) {
14731490 throw new LineSenderException (
14741491 "could not create sf_dir: " + sfDir + " rc=" + rc );
14751492 }
@@ -1544,7 +1561,7 @@ public Sender build() {
15441561 if (drainOrphans && sfDir != null ) {
15451562 io .questdb .client .std .ObjList <String > orphans =
15461563 io .questdb .client .cutlass .qwp .client .sf .cursor .OrphanScanner
1547- .scan (sfDir , senderId );
1564+ .scan (sfDir , senderId , orphanDrainBase , orphanDrainSlotCount );
15481565 if (orphans .size () > 0 ) {
15491566 org .slf4j .LoggerFactory .getLogger (LineSenderBuilder .class )
15501567 .info ("dispatching drainers for {} orphan slot(s) under {} "
@@ -2459,6 +2476,71 @@ public LineSenderBuilder senderId(String id) {
24592476 return this ;
24602477 }
24612478
2479+ /**
2480+ * The slot id ({@code sender_id}) currently configured on this
2481+ * builder, either parsed from the config string or left at its
2482+ * {@code "default"} default. Introspection hook for the connection
2483+ * pool, which derives a distinct per-slot id from this base so that
2484+ * multiple pooled senders sharing one {@code sf_dir} don't collide
2485+ * on the slot {@code flock}.
2486+ */
2487+ public String getConfiguredSenderId () {
2488+ return senderId ;
2489+ }
2490+
2491+ /**
2492+ * The store-and-forward group root ({@code sf_dir}) currently
2493+ * configured on this builder, or {@code null} when SF is disabled.
2494+ * Introspection hook for the connection pool, which needs the group
2495+ * root to locate its own managed slot dirs {@code <sf_dir>/<base>-<i>}
2496+ * when recovering unacked data a previous run left behind.
2497+ */
2498+ public String getConfiguredSfDir () {
2499+ return sfDir ;
2500+ }
2501+
2502+ /**
2503+ * Excludes the connection pool's <em>live</em> slot set from
2504+ * {@link #drainOrphans(boolean)} scanning: a sibling slot under
2505+ * {@code sf_dir} named {@code <base>-<index>} with
2506+ * {@code 0 <= index < slotCount} is never treated as a drainable orphan.
2507+ * <p>
2508+ * Internal introspection hook for the connection pool. The pool gives
2509+ * each pooled SF sender a distinct slot id {@code <base>-<index>} and
2510+ * recovers each slot's unacked data itself when it (re)creates that
2511+ * slot. Without this exclusion, one pooled sender's startup drainer
2512+ * could adopt a sibling pool slot's lock and dir, reintroducing the
2513+ * very "sf slot already in use" collision the per-slot ids were added
2514+ * to prevent.
2515+ * <p>
2516+ * Unlike a blanket {@code <base>-} prefix exclusion, the bound is the
2517+ * pool's {@code maxSize}: a same-base slot whose index is at or above
2518+ * {@code slotCount} (e.g. {@code <base>-3} left behind by a larger pool
2519+ * before {@code maxSize} shrank from 4 to 2) is NOT excluded and is
2520+ * drained like any foreign leftover, so its unacked data is recovered
2521+ * instead of being silently stranded. Foreign leftovers (a different
2522+ * base, or a bare un-suffixed id) are also still drained.
2523+ * <p>
2524+ * Pass a {@code null}/empty base or {@code slotCount <= 0} to disable
2525+ * the exclusion (the default).
2526+ */
2527+ public LineSenderBuilder orphanDrainExcludeManagedSlots (String base , int slotCount ) {
2528+ this .orphanDrainBase = base ;
2529+ this .orphanDrainSlotCount = slotCount ;
2530+ return this ;
2531+ }
2532+
2533+ /**
2534+ * True iff store-and-forward is enabled (an {@code sf_dir} was set).
2535+ * Introspection hook for the connection pool: SF senders own an
2536+ * exclusive on-disk slot, so each pooled sender needs its own slot
2537+ * id, whereas non-SF (memory-mode / HTTP / TCP) senders share no
2538+ * such resource and need no per-slot identity.
2539+ */
2540+ public boolean isStoreAndForwardEnabled () {
2541+ return sfDir != null ;
2542+ }
2543+
24622544 /**
24632545 * Per-call deadline for {@code Sender.flush()} spinning on a full
24642546 * cursor segment ring waiting for ACKs to drain space. Default
0 commit comments