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