diff --git a/core/src/main/java/io/questdb/client/Sender.java b/core/src/main/java/io/questdb/client/Sender.java index 5106da96..bfef5189 100644 --- a/core/src/main/java/io/questdb/client/Sender.java +++ b/core/src/main/java/io/questdb/client/Sender.java @@ -38,6 +38,13 @@ import io.questdb.client.cutlass.qwp.client.QwpWebSocketSender; import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine; import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop; +import io.questdb.client.cutlass.qwp.client.sf.cursor.OrphanScanner; +import io.questdb.client.cutlass.qwp.client.sf.cursor.PersistedSymbolDict; +import io.questdb.client.cutlass.qwp.client.sf.cursor.SlotLock; +import io.questdb.client.cutlass.qwp.client.sf.cursor.MmapSegmentCorruptionException; +import io.questdb.client.cutlass.qwp.client.sf.cursor.SfRecoveryException; +import io.questdb.client.cutlass.qwp.client.sf.cursor.SfSanitizedResidueException; +import io.questdb.client.cutlass.qwp.client.sf.cursor.UnreplayableSlotException; import io.questdb.client.impl.ConfStringParser; import io.questdb.client.impl.ConfigString; import io.questdb.client.impl.ConfigView; @@ -48,6 +55,7 @@ import io.questdb.client.std.Decimal256; import io.questdb.client.std.Decimal64; import io.questdb.client.std.Files; +import io.questdb.client.std.FilesFacade; import io.questdb.client.std.IntList; import io.questdb.client.std.Numbers; import io.questdb.client.std.NumericException; @@ -790,12 +798,16 @@ default Sender uuidColumn(CharSequence name, long lo, long hi) { * unconnected sender; the I/O thread runs the same retry loop in * the background. The user thread can call {@code at()} / * {@code flush()} immediately; rows accumulate in the cursor SF - * engine until the wire is up. Connect failures are retried - * indefinitely in the background; a terminal upgrade failure - * (auth reject, capability mismatch) is delivered to the async - * error inbox as a {@link io.questdb.client.SenderError} (no - * synchronous throw on the user call site). Wire - * {@code error_handler=...} to observe these. + * engine until the wire is up. Transport failures (unreachable or + * dropped server) retry indefinitely and are never surfaced -- the + * buffered rows are safe in SF and the server may still appear. A + * terminal auth, upgrade or capability rejection on the initial + * connect (before the wire has ever come up) has no caller left to + * throw at, so it is delivered to the async error inbox as a + * {@link io.questdb.client.SenderError}; wire {@code error_handler=...} + * to observe it. Once the wire has come up even once, SF owns the + * buffered data and the same rejections (e.g. a credential rotation) + * become transients retried indefinitely. * *

* Default resolution when the caller does not pick a value: @@ -995,6 +1007,12 @@ final class LineSenderBuilder { // build() time. 0 or negative is a documented "disable" value, so // a Long.MIN_VALUE sentinel keeps it distinguishable from "unset". private static final long DURABLE_ACK_KEEPALIVE_NOT_SET = Long.MIN_VALUE; + private static final org.slf4j.Logger LOG = org.slf4j.LoggerFactory.getLogger(LineSenderBuilder.class); + // How many quarantined copies of one slot may pile up under sf_dir before build() + // refuses to set aside another. Each is an unreplayable slot a human still has to + // look at; accumulating them without bound would turn a disk-space problem into a + // second incident. + private static final int MAX_QUARANTINE_SLOT_ATTEMPTS = 64; private static final int MIN_BUFFER_SIZE = AuthUtils.CHALLENGE_LEN + 1; // challenge size + 1; // sf-client.md section 4.4: the inbox capacity must accommodate the // distinct error categories in a bursty error stream so that drop-oldest @@ -1010,6 +1028,16 @@ final class LineSenderBuilder { private static final int PROTOCOL_TCP = 0; private static final int PROTOCOL_UDP = 3; private static final int PROTOCOL_WEBSOCKET = 2; + @TestOnly + private static volatile Runnable quarantineAfterCloseHook; + @TestOnly + private static volatile FilesFacade quarantineFilesFacade = FilesFacade.INSTANCE; + // Suffix for a slot set aside by quarantineTornSlot. Deliberately NOT the + // sender's own slot name, so a restarted sender does not re-adopt it as its own; + // quarantineTornSlot then marks it .failed, so the orphan drainer skips it too and + // the bytes stay put for a human to inspect and resend. + private static final String QUARANTINE_SLOT_SUFFIX = + OrphanScanner.QUARANTINE_SLOT_INFIX; private final ObjList hosts = new ObjList<>(); private final IntList ports = new IntList(); private long authTimeoutMillis = QwpWebSocketSender.DEFAULT_AUTH_TIMEOUT_MS; @@ -1058,6 +1086,7 @@ final class LineSenderBuilder { private int errorInboxCapacity = PARAMETER_NOT_SET_EXPLICITLY; private int maxFrameRejections = PARAMETER_NOT_SET_EXPLICITLY; private long poisonMinEscalationWindowMillis = PARAMETER_NOT_SET_EXPLICITLY; + private long catchUpCapGapMinEscalationWindowMillis = PARAMETER_NOT_SET_EXPLICITLY; private String httpPath; private String httpSettingsPath; private int httpTimeout = PARAMETER_NOT_SET_EXPLICITLY; @@ -1066,8 +1095,8 @@ final class LineSenderBuilder { // explicitly", which build() resolves to SYNC when any reconnect_* // knob was tuned by the user, otherwise OFF. SYNC retries on the // user thread up to the reconnect cap. ASYNC returns immediately - // and lets the I/O thread retry in the background, surfacing - // terminal failures via the error inbox. + // and lets the I/O thread retry every endpoint/transport failure in + // the background without surfacing it to the producer. private InitialConnectMode initialConnectMode = null; private String keyId; private int maxBackgroundDrainers = DEFAULT_MAX_BACKGROUND_DRAINERS; @@ -1512,6 +1541,10 @@ public Sender build() { long actualPoisonMinEscalationWindowMillis = poisonMinEscalationWindowMillis != PARAMETER_NOT_SET_EXPLICITLY ? poisonMinEscalationWindowMillis : CursorWebSocketSendLoop.DEFAULT_POISON_MIN_ESCALATION_WINDOW_MILLIS; + long actualCatchUpCapGapMinEscalationWindowMillis = + catchUpCapGapMinEscalationWindowMillis != PARAMETER_NOT_SET_EXPLICITLY + ? catchUpCapGapMinEscalationWindowMillis + : CursorWebSocketSendLoop.DEFAULT_CATCHUP_CAP_GAP_MIN_ESCALATION_WINDOW_MILLIS; // sfDir is the parent (group root); the actual slot lives // under sfDir/senderId. This is what the engine sees — the @@ -1556,56 +1589,173 @@ public Sender build() { ? (sfSyncIntervalMillis == PARAMETER_NOT_SET_EXPLICITLY ? DEFAULT_SF_SYNC_INTERVAL_MILLIS : sfSyncIntervalMillis) * 1_000_000L : 0L; - CursorSendEngine cursorEngine = new CursorSendEngine( - slotPath, actualSfMaxSegmentBytes, - actualSfMaxTotalBytes, actualSfAppendDeadlineNanos, - actualSfSyncIntervalNanos); - int actualErrorInboxCapacity = errorInboxCapacity != PARAMETER_NOT_SET_EXPLICITLY - ? errorInboxCapacity - : io.questdb.client.cutlass.qwp.client.sf.cursor.SenderErrorDispatcher.DEFAULT_CAPACITY; - int actualConnectionListenerInboxCapacity = connectionListenerInboxCapacity != PARAMETER_NOT_SET_EXPLICITLY - ? connectionListenerInboxCapacity - : io.questdb.client.cutlass.qwp.client.sf.cursor.SenderConnectionDispatcher.DEFAULT_CAPACITY; - List wsEndpoints = - new ArrayList<>(hosts.size()); - for (int i = 0, n = hosts.size(); i < n; i++) { - wsEndpoints.add(new QwpWebSocketSender.Endpoint(hosts.getQuick(i), ports.getQuick(i))); - } - QwpWebSocketSender connected; - try { - connected = QwpWebSocketSender.connect( - wsEndpoints, - wsTlsConfig, - actualAutoFlushRows, - actualAutoFlushBytes, - actualAutoFlushIntervalNanos, - wsAuthHeader, - requestDurableAck, - cursorEngine, - actualCloseFlushTimeoutMillis, - actualReconnectMaxDurationMillis, - actualReconnectInitialBackoffMillis, - actualReconnectMaxBackoffMillis, - actualInitialConnectMode, - errorHandler, - actualErrorInboxCapacity, - actualDurableAckKeepaliveIntervalMillis, - authTimeoutMillis, - connectTimeoutMillis == PARAMETER_NOT_SET_EXPLICITLY ? 0 : connectTimeoutMillis, - connectionListener, - actualConnectionListenerInboxCapacity, - actualMaxFrameRejections, - actualPoisonMinEscalationWindowMillis - ); - } catch (Throwable t) { - // connect() failed before ownership of cursorEngine - // transferred — close it ourselves. + QwpWebSocketSender connected = null; + // The parent-anchored logical lock is stable across a slot rename. Keep it + // from before the directory-local lock is acquired until connect() has either + // adopted that engine or quarantine has closed, renamed and recreated it. + // This closes the inode-swap window in which an already-queued orphan drainer + // could otherwise acquire the renamed directory's old .lock and later operate + // on the fresh slot through the original pathname. + try (SlotLock logicalSlotLock = slotPath == null + ? null + : SlotLock.acquireLogical(slotPath)) { + // The constructor's own recovery seed can also fail terminally, and + // not only as UnreplayableSlotException: when SegmentRing.openExisting + // had to skip an unreadable segment it throws SfRecoveryException (it + // constructs UnreplayableSlotException nowhere), and where it cannot + // even prove the chain's identity -- no manifest -- it quarantines the + // corrupt files and returns an EMPTY recovery rather than refusing. + // Either way the frame range cannot be shown already-acked, so recovery + // sets the slot aside rather than risk seeding the ack cursor past + // frames that were never delivered. All three types below are load + // bearing; narrowing this catch to UnreplayableSlotException would + // restore the permanent build() brick for the segment-skip case. That verdict gets + // the exact same quarantine-and-continue treatment as the connect()-time + // verdict below -- constructing cursorEngine is not inside the loop below, + // so a throw here would otherwise escape build() entirely, uncaught. + // quarantineTornSlot(null, ...) renames the WHOLE slot directory aside + // (not just the unreadable segment file) before building the replacement + // at the original slotPath, so the replacement starts on a genuinely empty + // directory with nothing left to skip -- it cannot throw the same way + // twice, which is what makes looping unnecessary here. + boolean quarantined = false; + CursorSendEngine cursorEngine; try { - cursorEngine.close(); - } catch (Throwable ignored) { - // best-effort + try { + cursorEngine = new CursorSendEngine( + slotPath, actualSfMaxSegmentBytes, + actualSfMaxTotalBytes, actualSfAppendDeadlineNanos, + actualSfSyncIntervalNanos); + } catch (SfSanitizedResidueException first) { + // NOT terminal, and it must be intercepted ahead of its + // SfRecoveryException parent below. Recovery durably zeroed + // proven-dead sealed residue BEFORE failing closed, so the + // chain on disk is already healed: quarantining here would + // set aside a slot whose backlog replays perfectly. Retry + // once over the healed chain; a repeat is genuine and takes + // the terminal arm. + LOG.info("sf slot {}: sealed residue sanitized during recovery ({}); " + + "retrying over the healed chain", + slotPath, first.getMessage()); + cursorEngine = new CursorSendEngine( + slotPath, actualSfMaxSegmentBytes, + actualSfMaxTotalBytes, actualSfAppendDeadlineNanos, + actualSfSyncIntervalNanos); + } + } catch (UnreplayableSlotException | SfRecoveryException + | MmapSegmentCorruptionException e) { + // The terminal recovery verdicts, and the only ones build() + // sets a slot aside for. UnreplayableSlotException says the + // symbol dictionary cannot be rebuilt from any source; + // SfRecoveryException and MmapSegmentCorruptionException say + // the durable chain itself is proven corrupt or incomplete. + // None of the three clears on a retry, and senderId is stable + // with a not-fully-drained slot retained on close -- so + // without this arm every restart re-recovers the same slot and + // throws again, and the application cannot construct a Sender + // at all, not even to BUFFER new rows. + // + // Deliberately NOT catching plain MmapSegmentException or + // SfOperationalException: those are operational (EMFILE, + // ENOMEM, an unreadable-but-possibly-intact file). Aborting + // startup on them is correct; quarantining on them would + // convert a transient into the permanent loss of a healthy + // slot's durable frames. + if (slotPath == null) { + throw e; + } + quarantined = true; + cursorEngine = quarantineTornSlot( + null, e, sfDir, senderId, slotPath, actualSfMaxSegmentBytes, + actualSfMaxTotalBytes, actualSfAppendDeadlineNanos, + actualSfSyncIntervalNanos, errorHandler); + } + int actualErrorInboxCapacity = errorInboxCapacity != PARAMETER_NOT_SET_EXPLICITLY + ? errorInboxCapacity + : io.questdb.client.cutlass.qwp.client.sf.cursor.SenderErrorDispatcher.DEFAULT_CAPACITY; + int actualConnectionListenerInboxCapacity = connectionListenerInboxCapacity != PARAMETER_NOT_SET_EXPLICITLY + ? connectionListenerInboxCapacity + : io.questdb.client.cutlass.qwp.client.sf.cursor.SenderConnectionDispatcher.DEFAULT_CAPACITY; + List wsEndpoints = + new ArrayList<>(hosts.size()); + for (int i = 0, n = hosts.size(); i < n; i++) { + wsEndpoints.add(new QwpWebSocketSender.Endpoint(hosts.getQuick(i), ports.getQuick(i))); + } + // The recovery seed inside connect() is the authority on whether a recovered + // slot can be replayed: it rebuilds the dictionary from its intact prefix and + // then from the surviving frames' own delta sections, and throws + // UnreplayableSlotException only once neither source holds the missing ids. + // Quarantining on anything weaker would set aside slots that recovery can + // still rescue, so build() waits for that verdict rather than pre-judging it. + while (connected == null) { + try { + connected = QwpWebSocketSender.connect( + wsEndpoints, + wsTlsConfig, + actualAutoFlushRows, + actualAutoFlushBytes, + actualAutoFlushIntervalNanos, + wsAuthHeader, + requestDurableAck, + cursorEngine, + actualCloseFlushTimeoutMillis, + actualReconnectMaxDurationMillis, + actualReconnectInitialBackoffMillis, + actualReconnectMaxBackoffMillis, + actualInitialConnectMode, + errorHandler, + actualErrorInboxCapacity, + actualDurableAckKeepaliveIntervalMillis, + authTimeoutMillis, + connectTimeoutMillis == PARAMETER_NOT_SET_EXPLICITLY ? 0 : connectTimeoutMillis, + connectionListener, + actualConnectionListenerInboxCapacity, + actualMaxFrameRejections, + actualPoisonMinEscalationWindowMillis, + actualCatchUpCapGapMinEscalationWindowMillis + ); + } catch (UnreplayableSlotException e) { + // The one failure build() recovers from. The slot's frames reference ids + // that nothing still holds, so they can never go on the wire -- but that is + // no reason to take the producer down with them. Before this, the throw + // escaped build() and, because senderId is stable and a not-fully-drained + // slot is retained on close, every retry re-recovered the same slot and + // threw again: the application could not construct a Sender at all, so it + // could not even BUFFER new rows. An already-lost batch became an unbounded + // outage of everything after it. + // + // Set the slot aside instead, keep its bytes for forensics and resend, and + // start the producer on a clean one. Once only: a second such failure would + // mean the FRESH slot is unreplayable, which cannot happen, so let it out + // rather than loop. + if (quarantined || slotPath == null) { + try { + // close(false): we still hold the logical slot lock. + cursorEngine.close(false); + } catch (Throwable ignored) { + // best-effort + } + throw e; + } + quarantined = true; + cursorEngine = quarantineTornSlot( + cursorEngine, e, sfDir, senderId, slotPath, actualSfMaxSegmentBytes, + actualSfMaxTotalBytes, actualSfAppendDeadlineNanos, + actualSfSyncIntervalNanos, errorHandler); + } catch (Throwable t) { + // connect() failed before ownership of cursorEngine + // transferred — close it ourselves. close(false) + // because logicalSlotLock is still held here: a fresh + // slot is fully drained, so the default close would + // unlink the very lock file this scope holds. + try { + cursorEngine.close(false); + } catch (Throwable ignored) { + // best-effort + } + throw t; + } } - throw t; } // connect() succeeded — `connected` now owns cursorEngine // via setCursorEngine(engine, true). From here on, ANY @@ -1721,6 +1871,35 @@ public Sender build() { return sender; } + /** + * Minimum wall-clock time (millis) a symbol-dictionary catch-up CAP GAP must + * persist before an orphan store-and-forward drainer quarantines its slot. + *

+ * A cap gap means a symbol already accepted by one node is too large to + * re-register on the node the sender just failed over to, because that node + * advertises a smaller maximum batch size. On a homogeneous cluster this cannot + * happen; it takes a heterogeneous or mid-roll cluster, or an operator lowering + * the cap below existing data. + *

+ * A foreground sender retries such a gap indefinitely because the larger-cap node + * may simply be away. An orphan drainer may quarantine only once the gap has BOTH + * recurred many times AND persisted for this long. Raise it for a cluster whose + * rolling restarts take longer than the 5-minute default; set it to {@code 0} to + * quarantine an orphan slot as soon as the retry count is exhausted. + *

+ * WebSocket transport only. + */ + public LineSenderBuilder catchUpCapGapMinEscalationWindowMillis(long millis) { + if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) { + throw new LineSenderException("catch_up_cap_gap_min_escalation_window_millis is only supported for WebSocket transport"); + } + if (millis < 0) { + throw new LineSenderException("catch_up_cap_gap_min_escalation_window_millis must be >= 0: ").put(millis); + } + this.catchUpCapGapMinEscalationWindowMillis = millis; + return this; + } + /** * close() drain timeout in milliseconds. The sender's {@code close()} * method blocks up to this many millis waiting for the server to ACK @@ -2167,11 +2346,10 @@ public LineSenderBuilder initialConnectMode(InitialConnectMode mode) { } /** - * Opt in to retrying the initial connect with the same backoff / - * cap / auth-terminal policy as in-flight reconnect. Set true if - * your deployment expects the server to come up shortly after the - * sender. Auth failures (HTTP 401/403/non-101) stay terminal in - * either mode. + * Opt in to retrying the initial connect with backoff on the calling + * thread, bounded by the configured reconnect cap. Set true if your + * deployment expects the server to come up shortly after the sender. + * Auth failures (HTTP 401/403/non-101) stay terminal in this SYNC mode. *

* When this method is not called, the resolution rule documented * on {@link InitialConnectMode} applies: SYNC implicitly when any @@ -2476,7 +2654,7 @@ public LineSenderBuilder maxFrameRejections(int rejections) { * alone can turn a transient outage into a producer-fatal terminal. * This window guarantees a brief outage a chance to clear (an OK * at/beyond the suspect resets the detector) first. {@code 0} disables - * the dwell (legacy immediate escalation at the strike threshold). + * the dwell (immediate escalation at the strike threshold). * Default {@code 5_000} (5 s). WebSocket only. */ public LineSenderBuilder poisonMinEscalationWindowMillis(long millis) { @@ -2512,8 +2690,8 @@ public LineSenderBuilder reconnectMaxBackoffMillis(long millis) { * with exponential backoff until connect succeeds or this many * millis elapse, then throws. The background reconnect loop * (mid-stream outages and async initial connect) does NOT consult - * this value: it retries indefinitely and halts only on a terminal - * auth/upgrade error or {@code close()}. + * this value: endpoint and transport failures are retried indefinitely + * until {@code close()}. *

* Default {@code 300_000} (5 minutes). Lower for fail-fast startup; * higher for tolerating a slow server boot. Must be positive: a zero @@ -2954,6 +3132,143 @@ private static long parseSizeValue(@NotNull StringSink value, @NotNull String na } } + /** + * Sets a slot aside that either connect() (a symbol dictionary that cannot cover its + * surviving frames, {@code UnreplayableSlotException}) or the + * {@code CursorSendEngine} constructor itself (a corrupt or incomplete durable + * chain, {@code SfRecoveryException} / {@code MmapSegmentCorruptionException}) + * declared terminal, and returns a fresh engine on an empty slot so the producer + * can keep producing. + *

+ * {@code torn} is the live engine to release, or {@code null} when the verdict came + * from the constructor and no engine was ever built -- there is nothing to + * {@link CursorSendEngine#close(boolean)} in that case, only the directory to rename. + *

+ * Such a slot is unreplayable BY THIS PRODUCER: either its frames reference symbol ids + * the recovered dictionary lost (a host/power crash tore the unsynced side-file), so a + * producer seeded from the short dictionary would hand those ids to different symbols + * and silently misattribute values -- or recovery could not show a skipped segment's + * frames were already acked, so replaying would risk seeding the ack cursor past data + * that was never delivered. Detecting either is correct and load-bearing -- but simply + * THROWING is not a safe response. {@code senderId} defaults to a stable name, so a + * restarted process re-adopts the same slot; the engine's close retains a slot that is + * not fully drained; and so every subsequent {@code build()} would re-recover the same + * slot and throw again -- forever, until an operator deleted the directory by hand. The + * application could not construct a Sender at all, and so could not even BUFFER new + * rows. That trades a bounded, already-lost batch for an unbounded outage of everything + * after it, which inverts the one guarantee store-and-forward exists to give. + *

+ * So: rename the slot aside instead, and mark it {@code .failed}. The verdict is + * authoritative -- the recovery seed already tried every source of truth (the + * persisted prefix AND the surviving frames' own deltas, or the per-segment scan for + * a skipped segment), and the orphan drainer's own replay guard uses that same walk, so there + * is nothing a drainer could rebuild that the seed did not. {@code markFailed} (below) + * therefore quarantines the copy for a human rather than leaving a drainer to retry an + * unreplayable slot forever; a full-dictionary-fallback slot never reaches here, because + * its dictionary is discarded at recovery and it never throws. The bytes are preserved + * on disk for forensics and a manual resend, and the new name -- NOT the sender's own + * slot name -- keeps a restarted sender from re-adopting it (and, for the segment-skip case, also + * keeps a later recovery from ever re-scanning the individually-renamed {@code .corrupt} + * segment inside it). The producer, meanwhile, starts on a clean empty slot and never + * notices. + *

+ * If the rename fails (a Windows share lock, a read-only mount) there is no way to + * free the slot name without destroying data, so fall back to the old behaviour and + * throw -- loudly, and never silently dropping bytes. + */ + private static CursorSendEngine quarantineTornSlot( + CursorSendEngine torn, RuntimeException cause, String sfDir, + String senderId, String slotPath, + long sfMaxSegmentBytes, long sfMaxTotalBytes, long sfAppendDeadlineNanos, + long sfSyncIntervalNanos, + io.questdb.client.SenderErrorHandler errorHandler + ) { + // The verdict, and the reason, come from the recovery seed -- the only code that + // has tried every source of truth. Recomputing them here would mean a second, + // independently-drifting notion of "unreplayable". + final String detail = cause.getMessage(); + // Release the slot lock and the dictionary fd before renaming. connect()'s failure + // path already closed the engine; close() is idempotent, so make it explicit rather + // than depend on that. close(false): build() holds the logical slot lock across this + // whole transition -- that is precisely what serialises the rename against a queued + // orphan drainer -- so the engine must not unlink it. torn is null when the verdict + // came from the constructor itself: no engine was ever built, so there is nothing to + // close. + if (torn != null) { + try { + torn.close(false); + } catch (Throwable ignored) { + // Best-effort, like build()'s other rollback closes. A failed close + // here (fsync of a not-fully-drained slot, a torn dictionary fd) + // must not abort the quarantine: skipping the rename and markFailed + // below would restore the permanent build() brick this method + // exists to remove, and replace the recovery verdict in `cause` + // with a secondary close failure. + } + } + Runnable hook = quarantineAfterCloseHook; + if (hook != null) { + hook.run(); + } + + FilesFacade ff = quarantineFilesFacade; + String quarantinePath = null; + for (int i = 0; i < MAX_QUARANTINE_SLOT_ATTEMPTS; i++) { + String candidate = sfDir + "/" + senderId + QUARANTINE_SLOT_SUFFIX + i; + if (!ff.exists(candidate)) { + quarantinePath = candidate; + break; + } + } + if (quarantinePath == null || ff.rename(slotPath, quarantinePath) != 0) { + throw new LineSenderException( + detail + "; the affected data must be resent. The slot could not be set aside " + + "automatically (" + (quarantinePath == null + ? "too many quarantined slots already under " + sfDir + : "rename to " + quarantinePath + " failed") + + "), so this sender cannot start until " + + slotPath + " is moved or removed by hand"); + } + // Mark the quarantined copy so the orphan drainer treats it as a + // human-in-the-loop slot rather than silently retrying it forever. + OrphanScanner.markFailed(quarantinePath, detail); + LOG.error("{} -- the slot has been set aside at {} and the affected data must be resent; " + + "this sender continues on a fresh, empty slot at {}", + detail, quarantinePath, slotPath); + // build() no longer throws for this -- it starts the producer on a fresh slot so + // the outage stays bounded. But abandoning buffered rows is precisely the event a + // caller must be able to act on, and LOG.error alone cannot carry it: this client + // ships slf4j-api with no binding, so an embedding app with no provider gets a NOP + // logger and the loss is announced nowhere. Deliver it programmatically too, so an + // errorHandler can alert / page / record it. Dispatched synchronously here because + // the async SenderErrorDispatcher belongs to the connected sender, which does not + // exist yet at build time. A throwing handler must not turn a contained outage back + // into a failed build, so swallow anything it raises. + if (errorHandler != null) { + try { + errorHandler.onError(SenderError.dataLoss( + detail + " [slot set aside at " + quarantinePath + + "; sender continues on a fresh slot at " + slotPath + ']', + quarantinePath)); + } catch (Throwable handlerFailure) { + LOG.error("sender error handler threw while reporting a quarantined slot: {}", + String.valueOf(handlerFailure)); + } + } + return new CursorSendEngine(slotPath, sfMaxSegmentBytes, sfMaxTotalBytes, + sfAppendDeadlineNanos, sfSyncIntervalNanos); + } + + @TestOnly + public static void setQuarantineAfterCloseHookForTest(Runnable hook) { + quarantineAfterCloseHook = hook; + } + + @TestOnly + public static void setQuarantineFilesFacadeForTest(FilesFacade ff) { + quarantineFilesFacade = ff == null ? FilesFacade.INSTANCE : ff; + } + private static int resolveIPv4(String host) { try { byte[] addr = InetAddress.getByName(host).getAddress(); @@ -3457,6 +3772,12 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) { } pos = getValue(configurationString, pos, sink, "poison_min_escalation_window_millis"); poisonMinEscalationWindowMillis(parseLongValue(sink, "poison_min_escalation_window_millis")); + } else if (Chars.equals("catch_up_cap_gap_min_escalation_window_millis", sink)) { + if (protocol != PROTOCOL_WEBSOCKET) { + throw new LineSenderException("catch_up_cap_gap_min_escalation_window_millis is only supported for WebSocket transport"); + } + pos = getValue(configurationString, pos, sink, "catch_up_cap_gap_min_escalation_window_millis"); + catchUpCapGapMinEscalationWindowMillis(parseLongValue(sink, "catch_up_cap_gap_min_escalation_window_millis")); } else if (Chars.equals("initial_connect_retry", sink)) { if (protocol != PROTOCOL_WEBSOCKET) { throw new LineSenderException("initial_connect_retry is only supported for WebSocket transport"); @@ -3729,6 +4050,9 @@ private LineSenderBuilder fromConfigWebSocket(CharSequence configurationString) if (view.has("poison_min_escalation_window_millis")) { poisonMinEscalationWindowMillis(wsLong(view, v, "poison_min_escalation_window_millis")); } + if (view.has("catch_up_cap_gap_min_escalation_window_millis")) { + catchUpCapGapMinEscalationWindowMillis(wsLong(view, v, "catch_up_cap_gap_min_escalation_window_millis")); + } if (view.has("sf_append_deadline_millis")) { sfAppendDeadlineMillis(wsLong(view, v, "sf_append_deadline_millis")); } @@ -3912,6 +4236,7 @@ public java.util.Map wsConfigSnapshotForTest() { m.put("max_background_drainers", maxBackgroundDrainers); m.put("max_frame_rejections", maxFrameRejections); m.put("poison_min_escalation_window_millis", poisonMinEscalationWindowMillis); + m.put("catch_up_cap_gap_min_escalation_window_millis", catchUpCapGapMinEscalationWindowMillis); m.put("error_inbox_capacity", errorInboxCapacity); m.put("connection_listener_inbox_capacity", connectionListenerInboxCapacity); m.put("token", httpToken); diff --git a/core/src/main/java/io/questdb/client/SenderError.java b/core/src/main/java/io/questdb/client/SenderError.java index d8e78503..bcc6250d 100644 --- a/core/src/main/java/io/questdb/client/SenderError.java +++ b/core/src/main/java/io/questdb/client/SenderError.java @@ -28,7 +28,9 @@ import org.jetbrains.annotations.Nullable; /** - * Immutable description of a server-side rejection of an asynchronously published batch. + * Immutable description of a server-side rejection of an asynchronously published batch — + * or, for {@link Category#DATA_LOSS}, of a client-side verdict that buffered data has been + * permanently abandoned and must be re-ingested from its source. * *

Delivered to user code through two paths: *