Implementations must support concurrent positional writes and reads to disjoint ranges (a + * {@link FileChannel} does). This is a generic IO primitive; the compaction extension point that + * hands one out is {@code io.github.jbellis.jvector.graph.disk.CompactionDestination}. + */ +@Experimental +public interface SeekableSink extends AutoCloseable { + + /** Write {@code src} fully at region-relative {@code position} (must be {@code >= 0}). */ + void writeAt(long position, ByteBuffer src) throws IOException; + + /** + * Read up to {@code dst.remaining()} bytes at region-relative {@code position} (must be + * {@code >= 0}); returns the number of bytes read, or {@code -1} at end of region. + */ + int readAt(long position, ByteBuffer dst) throws IOException; + + /** Force written bytes to durable storage. */ + void force() throws IOException; + + @Override + void close() throws IOException; + + /** + * Reference implementation over a {@link FileChannel} region. Every region-relative position is + * translated by {@code baseOffset}. The channel's lifecycle is owned by the caller — this + * {@link #close()} does not close the channel. + * + * @param channel the backing channel, opened for read and write + * @param baseOffset the absolute offset of the region's start within {@code channel} ({@code >= 0}) + */ + static SeekableSink over(FileChannel channel, long baseOffset) { + return new FileChannelSeekableSink(channel, baseOffset); + } +} diff --git a/jvector-base/src/main/java/io/github/jbellis/jvector/disk/SimpleMappedReader.java b/jvector-base/src/main/java/io/github/jbellis/jvector/disk/SimpleMappedReader.java index 46d91f8e3..142d6af6c 100644 --- a/jvector-base/src/main/java/io/github/jbellis/jvector/disk/SimpleMappedReader.java +++ b/jvector-base/src/main/java/io/github/jbellis/jvector/disk/SimpleMappedReader.java @@ -84,6 +84,15 @@ public SimpleMappedReader get() { return new SimpleMappedReader((MappedByteBuffer) buffer.duplicate()); } + /** + * Unmaps the shared mapping immediately (via {@code Unsafe.invokeCleaner}), with + * no coordination with outstanding readers — the raw-release family of + * {@link ReaderSupplier#close()}. Any reader vended by {@link #get()} that touches the + * mapping after this call faults natively (SIGSEGV) rather than throwing an exception, + * so close only once every vended reader is provably done. Where JDK 22+ is available, + * prefer the jvector-native {@code MemorySegmentReader}, whose close degrades to + * {@code IllegalStateException} instead. + */ @Override public void close() { if (unsafe != null) { diff --git a/jvector-base/src/main/java/io/github/jbellis/jvector/graph/ParallelExecutor.java b/jvector-base/src/main/java/io/github/jbellis/jvector/graph/ParallelExecutor.java new file mode 100644 index 000000000..e4ac29d6d --- /dev/null +++ b/jvector-base/src/main/java/io/github/jbellis/jvector/graph/ParallelExecutor.java @@ -0,0 +1,118 @@ +/* + * Copyright DataStax, Inc. + * + * 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 io.github.jbellis.jvector.graph; + +import java.util.concurrent.ForkJoinPool; +import java.util.function.Consumer; +import java.util.function.IntConsumer; +import java.util.stream.IntStream; +import java.util.stream.Stream; + +/** + * Runs a {@link GraphIndexBuilder}'s internal build/finalize iterations to completion, blocking the + * calling thread until every element has been processed. The implementation decides how the + * iteration is distributed: {@link #forkJoin(ForkJoinPool)} hosts a parallel stream on a dedicated + * pool (the historical behavior), while {@link #callerRuns()} runs everything sequentially on the + * calling thread with no worker threads and no pool. + *
+ * This is the seam that lets an embedder bound vector-graph construction to its own thread budget —
+ * e.g. one thread per compaction — instead of a jvector-owned all-core pool. It is the build/finalize
+ * counterpart to the caller-runs executor injection already available on the compaction merge path.
+ */
+public interface ParallelExecutor {
+ /**
+ * Runs {@code body} for each {@code i} in {@code [0, upperBound)}, blocking until all complete.
+ *
+ * @param upperBound the exclusive upper bound of the index range (may be {@code 0})
+ * @param body the action to apply to each index
+ */
+ void forEachInt(int upperBound, IntConsumer body);
+
+ /**
+ * Runs {@code body} for each element produced by {@code source}, blocking until all complete.
+ * Callers pass a sequential stream; the implementation decides whether to parallelize it.
+ *
+ * @param source the (sequential) stream of primitive ints to iterate
+ * @param body the action to apply to each element
+ */
+ void forEach(IntStream source, IntConsumer body);
+
+ /**
+ * Runs {@code body} for each element produced by {@code source}, blocking until all complete.
+ * Callers pass a sequential stream; the implementation decides whether to parallelize it.
+ *
+ * @param source the (sequential) stream to iterate
+ * @param body the action to apply to each element
+ * @param The compactor needs a real file (it uses a memory-mapped read-back during refinement and a
+ * random-access writer), so a {@link Target} is expressed as a container {@link Path} plus a base
+ * offset rather than an opaque stream. The generic {@link SeekableSink} primitive addresses the same
+ * window in region-relative coordinates and is what an embedder uses to read the committed body back
+ * for its checksum, e.g. {@code SeekableSink.over(channel, target.startOffset())}.
+ */
+@FunctionalInterface
+@Experimental
+public interface CompactionDestination {
+
+ /** Open a fresh target for one compaction. */
+ Target open() throws IOException;
+
+ /** One compaction's output region plus its commit/abort lifecycle. */
+ interface Target extends AutoCloseable {
+
+ /** The container file the graph body is written into. */
+ Path file();
+
+ /** The byte offset within {@link #file()} at which the graph body begins ({@code >= 0}). */
+ long startOffset();
+
+ /**
+ * Signalled exactly once, after the body has been fully written and forced, reporting its
+ * length ({@code file() size - startOffset()}). The embedder finalizes its container here
+ * (e.g. writes a footer/checksum). MUST be called before {@link #close()} on the success path.
+ */
+ void commit(long bodyLength) throws IOException;
+
+ /**
+ * Always runs (try-with-resources). If reached without a prior {@link #commit}, the
+ * compaction failed and the embedder discards the partial output; releases embedder resources.
+ */
+ @Override
+ void close() throws IOException;
+ }
+
+ /**
+ * Default standalone destination: writes to its own file at offset {@code 0} (today's
+ * {@code compact(Path)} behaviour). {@code commit} is a no-op marker; a {@code close} without a
+ * prior commit deletes the partial file.
+ */
+ static CompactionDestination toFile(Path path) {
+ return new FileCompactionDestination(path);
+ }
+}
diff --git a/jvector-base/src/main/java/io/github/jbellis/jvector/graph/disk/FileCompactionDestination.java b/jvector-base/src/main/java/io/github/jbellis/jvector/graph/disk/FileCompactionDestination.java
new file mode 100644
index 000000000..24d610d0e
--- /dev/null
+++ b/jvector-base/src/main/java/io/github/jbellis/jvector/graph/disk/FileCompactionDestination.java
@@ -0,0 +1,66 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * 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 io.github.jbellis.jvector.graph.disk;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+/**
+ * {@link CompactionDestination} that writes a standalone graph file at offset 0. Backs
+ * {@link CompactionDestination#toFile(Path)}.
+ */
+final class FileCompactionDestination implements CompactionDestination {
+ private final Path path;
+
+ FileCompactionDestination(Path path) {
+ if (path == null) {
+ throw new NullPointerException("path");
+ }
+ this.path = path;
+ }
+
+ @Override
+ public Target open() {
+ return new Target() {
+ private boolean committed;
+
+ @Override
+ public Path file() {
+ return path;
+ }
+
+ @Override
+ public long startOffset() {
+ return 0L;
+ }
+
+ @Override
+ public void commit(long bodyLength) {
+ // Standalone file: the graph IS the whole file; compact() already wrote and flushed it.
+ committed = true;
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (!committed) {
+ Files.deleteIfExists(path); // abort: discard the partial file
+ }
+ }
+ };
+ }
+}
diff --git a/jvector-base/src/main/java/io/github/jbellis/jvector/graph/disk/OnDiskGraphIndex.java b/jvector-base/src/main/java/io/github/jbellis/jvector/graph/disk/OnDiskGraphIndex.java
index ba34e2cb0..915eb08c1 100644
--- a/jvector-base/src/main/java/io/github/jbellis/jvector/graph/disk/OnDiskGraphIndex.java
+++ b/jvector-base/src/main/java/io/github/jbellis/jvector/graph/disk/OnDiskGraphIndex.java
@@ -227,7 +227,10 @@ private Int2ObjectHashMap
+ * Footer loading trusts the end of the supplier's range. With {@code useFooter=true}
+ * and a v5+ graph, metadata is located relative to the reader's {@code length()} — the file
+ * end, for whole-file suppliers. That is only correct when the graph is the last
+ * content in the supplier's range. Never footer-load a reused or embedder-owned container
+ * whose length extends past the graph body: stale bytes there either fail the load loudly
+ * or, if they end in a stale-but-still-valid footer, silently resurrect the old graph over
+ * the new bytes. For such containers, pass {@code useFooter=false} with the known
+ * {@code offset}, or use a region-bounded {@link ReaderSupplier} whose {@code length()} is
+ * the end of the graph's region.
*
* @param readerSupplier the reader supplier to use to read the graph and index.
* @param offset the offset in bytes from the start of the file where the index starts.
@@ -270,6 +283,9 @@ public static OnDiskGraphIndex load(ReaderSupplier readerSupplier, long offset,
/**
* Load an index from the given reader supplier where header and graph are located on the same file at offset 0.
+ * For v5+ graphs, metadata is located via the footer at the end of the supplier's range —
+ * the supplier must contain the graph and nothing after it; see the
+ * {@link #load(ReaderSupplier, long, boolean)} warning.
*
* @param readerSupplier the reader supplier to use to read the graph index.
*/
@@ -494,6 +510,21 @@ public View(RandomAccessReader reader) {
this.neighbors = new int[layerInfo.stream().mapToInt(li -> li.degree).max().orElse(0)];
}
+ /**
+ * Guards every on-disk record access. A node ordinal outside the L0 record space
+ * ({@code idUpperBound}, which exceeds {@code size(0)} for graphs renumbered with holes)
+ * would otherwise become a silent wild offset into the mapped file — reading garbage (or
+ * faulting) instead of failing diagnosably. Both package-private offset entry points call
+ * this, so each access is validated exactly once, with a single branch against a final
+ * bound.
+ */
+ private void requireValidNode(int node) {
+ if (node < 0 || node >= idUpperBound) {
+ throw new IllegalArgumentException(
+ "node ordinal " + node + " out of range [0, " + idUpperBound + ") for this graph");
+ }
+ }
+
@Override
public int dimension() {
return dimension;
@@ -512,6 +543,7 @@ public RandomAccessVectorValues copy() {
// package-private: OnDiskGraphIndexCompactor uses this for in-place neighbor refinement
long offsetFor(int node, FeatureId featureId) {
+ requireValidNode(node);
Feature feature = features.get(featureId);
// Separated features are just global offset + node offset
@@ -528,6 +560,7 @@ long offsetFor(int node, FeatureId featureId) {
// package-private: OnDiskGraphIndexCompactor uses this for in-place neighbor refinement
long neighborsOffsetFor(int level, int node) {
+ requireValidNode(node);
assert level == 0; // higher layers are in memory
// skip node ID + inline features
@@ -588,8 +621,14 @@ public NodesIterator getNeighborsIterator(int level, int node) {
// For layer 0, read from disk
reader.seek(neighborsOffsetFor(level, node));
nodeDegree = reader.readInt();
- assert nodeDegree <= neighbors.length
- : String.format("Node %d neighborCount %d > M %d", node, nodeDegree, neighbors.length);
+ if (nodeDegree < 0 || nodeDegree > neighbors.length) {
+ // A real check, not an assert: an out-of-range on-disk degree means the
+ // block is corrupt or the metadata is stale, and the garbage ints that a
+ // blind read would yield become out-of-range node ids downstream.
+ throw new IllegalStateException(String.format(
+ "Corrupt neighbor block: node %d at level 0 declares degree %d outside [0, %d] (block offset %d)",
+ node, nodeDegree, neighbors.length, neighborsOffsetFor(level, node)));
+ }
reader.read(neighbors, 0, nodeDegree);
stored = neighbors;
} else {
diff --git a/jvector-base/src/main/java/io/github/jbellis/jvector/util/RuntimeMode.java b/jvector-base/src/main/java/io/github/jbellis/jvector/util/RuntimeMode.java
new file mode 100644
index 000000000..50758893b
--- /dev/null
+++ b/jvector-base/src/main/java/io/github/jbellis/jvector/util/RuntimeMode.java
@@ -0,0 +1,89 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * 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 io.github.jbellis.jvector.util;
+
+import java.util.Locale;
+import java.util.logging.Logger;
+
+/**
+ * Library-wide runtime mode, from the {@code jvector.mode} system property.
+ *
+ * Two modes: {@code prod} (or {@code production}) and {@code dev} (or
+ * {@code development}), case-insensitive. The mode gates diagnostic
+ * walks — computations whose value is re-derived from first principles by
+ * traversing a structure, such as {@link
+ * io.github.jbellis.jvector.quantization.PQVectors#ramBytesUsed()} summing
+ * every compressed chunk. In production mode such walks are categorically
+ * replaced by incrementally-maintained values; in development mode they run in
+ * full, so a drifted cache or accounting bug is observable.
+ *
+ * Default is production. An embedding host once called the chunk
+ * walk once per inserted vector, turning index-build accounting into
+ * O(n²) — a single compaction burned two CPU-hours inside
+ * {@code ramBytesUsed} while build workers starved. A diagnostic full-walk is
+ * something a developer opts into, not something a production host should
+ * have to know to opt out of.
+ *
+ * Unrecognized values log a warning and resolve to production, not
+ * development: falling back to the diagnostic mode on a typo would silently
+ * reintroduce exactly the pathology above.
+ */
+public final class RuntimeMode {
+ private static final Logger LOG = Logger.getLogger(RuntimeMode.class.getName());
+
+ public static final String PROPERTY = "jvector.mode";
+
+ private static final boolean DEVELOPMENT =
+ parseIsDevelopment(System.getProperty(PROPERTY), LOG);
+
+ private RuntimeMode() {
+ }
+
+ /** True when diagnostic walks should run in full. */
+ public static boolean isDevelopment() {
+ return DEVELOPMENT;
+ }
+
+ /** True when diagnostic walks are replaced by maintained values. */
+ public static boolean isProduction() {
+ return !DEVELOPMENT;
+ }
+
+ /**
+ * Pure parse, exposed for tests (the static mode is fixed at class load).
+ */
+ static boolean parseIsDevelopment(String raw, Logger log) {
+ if (raw == null) {
+ return false;
+ }
+ switch (raw.trim().toLowerCase(Locale.ROOT)) {
+ case "dev":
+ case "development":
+ return true;
+ case "prod":
+ case "production":
+ case "":
+ return false;
+ default:
+ log.warning(() -> PROPERTY + "=" + raw
+ + " is not recognized (expected prod|production|dev|development); "
+ + "using production. Development mode re-enables full diagnostic "
+ + "walks and must be asked for exactly.");
+ return false;
+ }
+ }
+}
diff --git a/jvector-base/src/main/java/io/github/jbellis/jvector/util/work/LeakyBucketLimiter.java b/jvector-base/src/main/java/io/github/jbellis/jvector/util/work/LeakyBucketLimiter.java
new file mode 100644
index 000000000..0ebba8f3d
--- /dev/null
+++ b/jvector-base/src/main/java/io/github/jbellis/jvector/util/work/LeakyBucketLimiter.java
@@ -0,0 +1,67 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * 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 io.github.jbellis.jvector.util.work;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A leaky-bucket rate meter realizing the {@link WorkLimiter} facet: {@link #acquire} paces the
+ * aggregate admitted amount to a fixed {@code unitsPerSecond}, blocking the caller when the rate
+ * would be exceeded. The bucket drains during idle gaps (a burst after a quiet period is not
+ * charged for the idle time), and the first request after an idle period is admitted without
+ * delay — the cost of each request is paid by the next one, which is the standard smooth
+ * shaping behaviour. {@link #onProgress} is inherited as a no-op: this limiter only throttles.
+ *
+ * Thread-safe and reentrant: the emission clock is advanced under a short lock, then the caller
+ * sleeps outside the lock, so concurrent callers serialize their reservations but wait
+ * independently. The returned grant is a no-op — the cost is paid entirely at {@code acquire}.
+ *
+ * Obtain instances via {@link ProgressLimiter#rateLimited(double)}.
+ */
+final class LeakyBucketLimiter implements ProgressLimiter {
+ private final double nanosPerUnit;
+ private final Object lock = new Object();
+ // Earliest nanoTime at which the next reservation may start. Long.MIN_VALUE until the first
+ // acquire, so Math.max(now, nextFreeNanos) == now (a fully drained bucket) on the first call.
+ private long nextFreeNanos = Long.MIN_VALUE;
+
+ LeakyBucketLimiter(double unitsPerSecond) {
+ if (!(unitsPerSecond > 0) || Double.isInfinite(unitsPerSecond)) {
+ throw new IllegalArgumentException("unitsPerSecond must be finite and > 0, got " + unitsPerSecond);
+ }
+ this.nanosPerUnit = 1_000_000_000.0 / unitsPerSecond;
+ }
+
+ @Override
+ public Grant acquire(long amount) throws InterruptedException {
+ if (amount <= 0) {
+ return Grant.NOOP;
+ }
+ long startAt;
+ synchronized (lock) {
+ long now = System.nanoTime();
+ startAt = Math.max(now, nextFreeNanos); // drain if idle, else queue behind backlog
+ long cost = (long) Math.min((double) Long.MAX_VALUE, amount * nanosPerUnit);
+ nextFreeNanos = startAt + cost;
+ }
+ // Sleep (interruptibly, so cancellation aborts) until this request's slot opens.
+ for (long remaining = startAt - System.nanoTime(); remaining > 0; remaining = startAt - System.nanoTime()) {
+ TimeUnit.NANOSECONDS.sleep(remaining);
+ }
+ return Grant.NOOP;
+ }
+}
diff --git a/jvector-base/src/main/java/io/github/jbellis/jvector/util/work/ProgressLimiter.java b/jvector-base/src/main/java/io/github/jbellis/jvector/util/work/ProgressLimiter.java
new file mode 100644
index 000000000..1ae328ee6
--- /dev/null
+++ b/jvector-base/src/main/java/io/github/jbellis/jvector/util/work/ProgressLimiter.java
@@ -0,0 +1,110 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * 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 io.github.jbellis.jvector.util.work;
+
+import io.github.jbellis.jvector.annotations.Experimental;
+
+import java.util.Objects;
+import java.util.function.Consumer;
+
+/**
+ * The {@link ProgressTracker tracker} and the {@link WorkLimiter throttle} melded into one control
+ * surface. A long-running jvector operation accepts a single {@code ProgressLimiter} and uses both
+ * facets; an embedder may override only the facet it needs — the other defaults to a no-op, so
+ * {@link #UNLIMITED} behaves exactly as if no SPI were installed.
+ *
+ * Both methods default to no-ops here. A consumer that wants only one facet can still accept a
+ * lambda via the single-method parents ({@link ProgressTracker}, {@link WorkLimiter}); a consumer
+ * that wants both accepts a {@code ProgressLimiter}.
+ */
+@Experimental
+public interface ProgressLimiter extends ProgressTracker, WorkLimiter {
+
+ @Override
+ default void onProgress(WorkStage stage, long completed, long total) { }
+
+ @Override
+ default Grant acquire(long amount) throws InterruptedException { return Grant.NOOP; }
+
+ /** Observes nothing and limits nothing — behaviour identical to no SPI installed. */
+ ProgressLimiter UNLIMITED = new ProgressLimiter() { };
+
+ /**
+ * A leaky-bucket rate meter realizing the throttle facet: {@link #acquire} paces the aggregate
+ * admitted amount to {@code unitsPerSecond} (bytes/sec for the compaction consumer), blocking
+ * the caller when the rate would be exceeded and draining during idle gaps. {@link #onProgress}
+ * is a no-op and the returned grant is a no-op (cost is paid at {@code acquire}). Compose with
+ * {@link #logging(ProgressLimiter, Consumer)} to also log.
+ *
+ * @param unitsPerSecond the sustained admission rate; must be finite and {@code > 0}
+ * @throws IllegalArgumentException if {@code unitsPerSecond} is not finite and positive
+ */
+ static ProgressLimiter rateLimited(double unitsPerSecond) {
+ return new LeakyBucketLimiter(unitsPerSecond);
+ }
+
+ /**
+ * Wraps {@code delegate}, emitting a one-line message to {@code sink} on each
+ * {@link #onProgress} and on each {@link #acquire} that actually blocked, then delegating both
+ * facets to {@code delegate}. Composes over any limiter — e.g.
+ * {@code logging(rateLimited(bytesPerSecond), log::info)} logs a rate-limited operation. The
+ * delegate's grant is returned unchanged, so a semaphore delegate still releases on close.
+ *
+ * @param delegate the limiter to observe and delegate to; {@code null} means {@link #UNLIMITED}
+ * @param sink receives formatted log lines (e.g. {@code msg -> logger.info(msg)})
+ */
+ static ProgressLimiter logging(ProgressLimiter delegate, Consumer Best-effort and cheap: implementations must not throw (the caller invokes this on its
+ * orchestrating thread and treats it as fire-and-forget). See {@link ProgressLimiter} for the
+ * melded progress + throttle surface that most consumers accept.
+ */
+@Experimental
+@FunctionalInterface
+public interface ProgressTracker {
+ /**
+ * Reports progress for {@code stage}.
+ *
+ * @param stage the stage reporting progress
+ * @param completed work done so far in this stage, in stage-defined units; monotonically
+ * non-decreasing within a stage
+ * @param total total work for this stage, or {@code -1} if not yet known
+ */
+ void onProgress(WorkStage stage, long completed, long total);
+
+ /**
+ * Starts observing the lifetime of {@code stage}. The returned scope must be closed exactly
+ * once, normally with try-with-resources. Implementations may use this for a long-task timer;
+ * the default keeps instrumentation optional for embedders that only consume progress.
+ */
+ default PhaseScope startPhase(WorkStage stage) {
+ return PhaseScope.NOOP;
+ }
+
+ /** Closeable lifetime token returned by {@link #startPhase}. */
+ @FunctionalInterface
+ interface PhaseScope extends AutoCloseable {
+ @Override
+ void close();
+
+ PhaseScope NOOP = () -> { };
+ }
+
+ /** A tracker that discards every update. */
+ ProgressTracker NOOP = (stage, completed, total) -> { };
+}
diff --git a/jvector-base/src/main/java/io/github/jbellis/jvector/util/work/WorkLimiter.java b/jvector-base/src/main/java/io/github/jbellis/jvector/util/work/WorkLimiter.java
new file mode 100644
index 000000000..5355f0757
--- /dev/null
+++ b/jvector-base/src/main/java/io/github/jbellis/jvector/util/work/WorkLimiter.java
@@ -0,0 +1,59 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * 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 io.github.jbellis.jvector.util.work;
+
+import io.github.jbellis.jvector.annotations.Experimental;
+
+/**
+ * Admission contract: blocks until an amount of work may proceed, returning a {@link Grant} that
+ * the caller closes once the admitted work has completed.
+ *
+ * The unit of {@code amount} is defined by the consumer (e.g. bytes for IO, or rows,
+ * nodes, items); jvector fixes only the blocking-grant mechanism, never the meaning of the
+ * quantity. Implementations must be thread-safe and reentrant. {@code acquire} may block but must
+ * not throw for ordinary back-pressure.
+ */
+@Experimental
+@FunctionalInterface
+public interface WorkLimiter {
+ /**
+ * Blocks until {@code amount} units of work may proceed.
+ *
+ * @param amount the amount of work about to be performed, in consumer-defined units
+ * @return a non-null grant to {@link Grant#close() close} once that work has completed
+ * @throws InterruptedException if the calling thread is interrupted while blocked, which
+ * aborts the operation
+ */
+ Grant acquire(long amount) throws InterruptedException;
+
+ /**
+ * A handle released by the consumer once the admitted work has completed. For a rate-limiter
+ * realization (cost paid at {@link WorkLimiter#acquire}) {@link #close()} is a no-op; for a
+ * semaphore-style in-flight-amount realization it releases the permits taken by {@code acquire}.
+ */
+ interface Grant extends AutoCloseable {
+ /** Releases the grant. Never throws. */
+ @Override
+ void close();
+
+ /** A grant that holds nothing and releases nothing. */
+ Grant NOOP = () -> { };
+ }
+
+ /** A limiter that admits everything immediately. */
+ WorkLimiter UNLIMITED = amount -> Grant.NOOP;
+}
diff --git a/jvector-base/src/main/java/io/github/jbellis/jvector/util/work/WorkStage.java b/jvector-base/src/main/java/io/github/jbellis/jvector/util/work/WorkStage.java
new file mode 100644
index 000000000..001197a00
--- /dev/null
+++ b/jvector-base/src/main/java/io/github/jbellis/jvector/util/work/WorkStage.java
@@ -0,0 +1,33 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * 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 io.github.jbellis.jvector.util.work;
+
+import io.github.jbellis.jvector.annotations.Experimental;
+
+/**
+ * Identifies a stage of a long-running operation. The consumer defines its own stages; an
+ * {@code enum} satisfies this for free via {@link Enum#name()}.
+ *
+ * Part of the generic progress + work-admission SPI ({@link ProgressTracker},
+ * {@link WorkLimiter}, {@link ProgressLimiter}). Neither the stage identity nor the unit of work
+ * is fixed by jvector; both are supplied by the consumer.
+ */
+@Experimental
+public interface WorkStage {
+ /** The stage's name, stable within a single operation. */
+ String name();
+}
diff --git a/jvector-tests/src/test/java/io/github/jbellis/jvector/disk/TestSeekableSink.java b/jvector-tests/src/test/java/io/github/jbellis/jvector/disk/TestSeekableSink.java
new file mode 100644
index 000000000..9fde6b162
--- /dev/null
+++ b/jvector-tests/src/test/java/io/github/jbellis/jvector/disk/TestSeekableSink.java
@@ -0,0 +1,85 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * 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 io.github.jbellis.jvector.disk;
+
+import org.junit.Test;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class TestSeekableSink {
+
+ @Test
+ public void writesAndReadsInRegionRelativeCoordinates() throws IOException {
+ Path f = Files.createTempFile("sink", ".bin");
+ try (FileChannel ch = FileChannel.open(f, StandardOpenOption.WRITE, StandardOpenOption.READ)) {
+ long base = 100;
+ SeekableSink sink = SeekableSink.over(ch, base);
+ sink.writeAt(0, ByteBuffer.wrap("hello".getBytes(StandardCharsets.UTF_8)));
+ sink.writeAt(5, ByteBuffer.wrap("WORLD".getBytes(StandardCharsets.UTF_8)));
+ sink.force();
+
+ // Region-relative read returns what was written.
+ ByteBuffer dst = ByteBuffer.allocate(10);
+ assertEquals(10, sink.readAt(0, dst));
+ assertEquals("helloWORLD", new String(dst.array(), StandardCharsets.UTF_8));
+
+ // The bytes actually land at the absolute base offset (region-relative -> absolute).
+ ByteBuffer raw = ByteBuffer.allocate(10);
+ ch.read(raw, base);
+ assertEquals("helloWORLD", new String(raw.array(), StandardCharsets.UTF_8));
+
+ // Nothing was written before the region.
+ ByteBuffer before = ByteBuffer.allocate((int) base);
+ ch.read(before, 0);
+ for (byte b : before.array()) {
+ assertEquals("region must not write before its base", 0, b);
+ }
+
+ // close() must NOT close the caller-owned channel.
+ sink.close();
+ assertTrue("sink.close() must not close the caller's channel", ch.isOpen());
+ }
+ Files.deleteIfExists(f);
+ }
+
+ @Test
+ public void rejectsNegativeBaseAndPosition() throws IOException {
+ Path f = Files.createTempFile("sink", ".bin");
+ try (FileChannel ch = FileChannel.open(f, StandardOpenOption.WRITE, StandardOpenOption.READ)) {
+ try { SeekableSink.over(ch, -1); fail("negative base"); } catch (IllegalArgumentException expected) { }
+ SeekableSink sink = SeekableSink.over(ch, 0);
+ try { sink.writeAt(-1, ByteBuffer.allocate(1)); fail("negative write pos"); } catch (IllegalArgumentException expected) { }
+ try { sink.readAt(-1, ByteBuffer.allocate(1)); fail("negative read pos"); } catch (IllegalArgumentException expected) { }
+ }
+ Files.deleteIfExists(f);
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void rejectsNullChannel() {
+ SeekableSink.over(null, 0);
+ }
+}
diff --git a/jvector-tests/src/test/java/io/github/jbellis/jvector/util/TestRuntimeMode.java b/jvector-tests/src/test/java/io/github/jbellis/jvector/util/TestRuntimeMode.java
new file mode 100644
index 000000000..d05e6f50d
--- /dev/null
+++ b/jvector-tests/src/test/java/io/github/jbellis/jvector/util/TestRuntimeMode.java
@@ -0,0 +1,58 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * 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 io.github.jbellis.jvector.util;
+
+import java.util.logging.Logger;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class TestRuntimeMode {
+ private static final Logger LOG = Logger.getLogger(TestRuntimeMode.class.getName());
+
+ @Test
+ public void terseAndVerboseSpellingsBothParse() {
+ assertTrue(RuntimeMode.parseIsDevelopment("dev", LOG));
+ assertTrue(RuntimeMode.parseIsDevelopment("development", LOG));
+ assertTrue(RuntimeMode.parseIsDevelopment("DEVELOPMENT", LOG));
+ assertTrue(RuntimeMode.parseIsDevelopment(" Dev ", LOG));
+ assertFalse(RuntimeMode.parseIsDevelopment("prod", LOG));
+ assertFalse(RuntimeMode.parseIsDevelopment("production", LOG));
+ assertFalse(RuntimeMode.parseIsDevelopment("PROD", LOG));
+ }
+
+ @Test
+ public void defaultIsProduction() {
+ assertFalse("unset must be production — the diagnostic walk is opt-in",
+ RuntimeMode.parseIsDevelopment(null, LOG));
+ assertFalse(RuntimeMode.parseIsDevelopment("", LOG));
+ }
+
+ /**
+ * A typo must NOT fall back to development: that would silently
+ * reintroduce the per-insert full-walk pathology the gate exists to
+ * prevent.
+ */
+ @Test
+ public void unknownValuesResolveToProduction() {
+ assertFalse(RuntimeMode.parseIsDevelopment("porduction", LOG));
+ assertFalse(RuntimeMode.parseIsDevelopment("debug", LOG));
+ assertFalse(RuntimeMode.parseIsDevelopment("true", LOG));
+ }
+}
diff --git a/jvector-tests/src/test/java/io/github/jbellis/jvector/util/work/TestProgressLimiter.java b/jvector-tests/src/test/java/io/github/jbellis/jvector/util/work/TestProgressLimiter.java
new file mode 100644
index 000000000..e446ffc61
--- /dev/null
+++ b/jvector-tests/src/test/java/io/github/jbellis/jvector/util/work/TestProgressLimiter.java
@@ -0,0 +1,314 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * 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 io.github.jbellis.jvector.util.work;
+
+import io.github.jbellis.jvector.util.work.WorkLimiter.Grant;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class TestProgressLimiter {
+
+ private static final WorkStage STAGE = () -> "TEST";
+
+ private static long millisFor(ThrowingRunnable r) throws Exception {
+ long t0 = System.nanoTime();
+ r.run();
+ return (System.nanoTime() - t0) / 1_000_000L;
+ }
+
+ private interface ThrowingRunnable { void run() throws Exception; }
+
+ // ---- rateLimited (leaky bucket) ----
+
+ @Test
+ public void rateLimitedRejectsNonPositiveOrNonFiniteRate() {
+ for (double bad : new double[]{0.0, -1.0, -0.0, Double.NaN, Double.POSITIVE_INFINITY}) {
+ try {
+ ProgressLimiter.rateLimited(bad);
+ fail("expected IllegalArgumentException for rate " + bad);
+ } catch (IllegalArgumentException expected) {
+ // ok
+ }
+ }
+ }
+
+ @Test
+ public void rateLimitedAdmitsZeroOrNegativeAmountImmediately() throws Exception {
+ ProgressLimiter limiter = ProgressLimiter.rateLimited(1.0); // 1 unit/sec: any real wait would be seconds
+ long ms = millisFor(() -> {
+ try (Grant g = limiter.acquire(0)) { assertNotNull(g); }
+ try (Grant g = limiter.acquire(-100)) { assertNotNull(g); }
+ });
+ assertTrue("zero/negative amount must not block, waited " + ms + "ms", ms < 500);
+ }
+
+ @Test
+ public void rateLimitedPacesSubsequentAcquire() throws Exception {
+ ProgressLimiter limiter = ProgressLimiter.rateLimited(1000.0); // 1 unit/ms
+ limiter.acquire(200).close(); // warmup: drained bucket admits the first request immediately
+
+ long ms = millisFor(() -> limiter.acquire(200).close()); // must wait ~200ms behind the warmup reservation
+ assertTrue("expected pacing >= ~100ms at 1000 units/s after a 200-unit warmup, got " + ms + "ms", ms >= 100);
+ }
+
+ @Test
+ public void rateLimitedFirstAcquireIsNotDelayed() throws Exception {
+ ProgressLimiter limiter = ProgressLimiter.rateLimited(10.0); // slow: a delayed first call would be seconds
+ long ms = millisFor(() -> limiter.acquire(1000).close());
+ assertTrue("first acquire on a drained bucket must not block, waited " + ms + "ms", ms < 500);
+ }
+
+ @Test
+ public void rateLimitedIsInterruptible() throws Exception {
+ ProgressLimiter limiter = ProgressLimiter.rateLimited(100.0); // 100 units/sec
+ limiter.acquire(100).close(); // warmup reserves ~1s of future emission time
+
+ AtomicReference{@code
+ * try (CompactionDestination.Target t = destination.open()) {
+ * // ...compactor writes the graph into t.file() at t.startOffset()...
+ * t.commit(bodyLength); // success: body written & durable; embedder finalizes its footer
+ * } // close() always runs; no commit() => aborted (discard partial output)
+ * }
+ *
+ *