Integration robustness contracts

jvector-base/src/main/java/io/github/jbellis/jvector/disk/FileChannelSeekableSink.java

@@ -0,0 +1,70 @@
+/*
+ * 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 java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+
+/**
+ * {@link SeekableSink} over a {@link FileChannel}, translating region-relative positions by a fixed
+ * base offset. The channel is owned by the caller; {@link #close()} does not close it.
+ */
+final class FileChannelSeekableSink implements SeekableSink {
+    private final FileChannel channel;
+    private final long baseOffset;
+
+    FileChannelSeekableSink(FileChannel channel, long baseOffset) {
+        if (channel == null) {
+            throw new NullPointerException("channel");
+        }
+        if (baseOffset < 0) {
+            throw new IllegalArgumentException("baseOffset must be >= 0, got " + baseOffset);
+        }
+        this.channel = channel;
+        this.baseOffset = baseOffset;
+    }
+
+    @Override
+    public void writeAt(long position, ByteBuffer src) throws IOException {
+        if (position < 0) {
+            throw new IllegalArgumentException("position must be >= 0, got " + position);
+        }
+        long abs = baseOffset + position;
+        while (src.hasRemaining()) {
+            abs += channel.write(src, abs);
+        }
+    }
+
+    @Override
+    public int readAt(long position, ByteBuffer dst) throws IOException {
+        if (position < 0) {
+            throw new IllegalArgumentException("position must be >= 0, got " + position);
+        }
+        return channel.read(dst, baseOffset + position);
+    }
+
+    @Override
+    public void force() throws IOException {
+        channel.force(false);
+    }
+
+    @Override
+    public void close() {
+        // The channel is owned by the caller, per SeekableSink.over(...).
+    }
+}

jvector-base/src/main/java/io/github/jbellis/jvector/disk/ReaderSupplier.java

@@ -41,6 +41,23 @@ public interface ReaderSupplier extends AutoCloseable {
     default void prefetch(long offset, long length) {
     }
 
+    /**
+     * Releases the supplier's underlying resource. Two implementation families exist, with very
+     * different safety under concurrency:
+     * <ul>
+     * <li><b>Coordinated</b> (e.g. the jvector-native {@code MemorySegmentReader.Supplier}, whose
+     * shared-Arena close performs a liveness handshake): closing while vended readers are still
+     * in use degrades to {@code IllegalStateException} on those readers.</li>
+     * <li><b>Raw-release</b> (e.g. {@link SimpleMappedReader.Supplier}, which unmaps
+     * immediately): closing while any vended reader is mid-read invalidates the mapped pages
+     * underneath it, and the JVM fails with a native fault (SIGSEGV) rather than an
+     * exception.</li>
+     * </ul>
+     * Callers must not close a supplier until every reader vended by {@link #get()} is provably
+     * quiescent; implementations should document which family they belong to.
+     *
+     * @throws IOException if an I/O error occurs
+     */
     default void close() throws IOException {
     }
 }

jvector-base/src/main/java/io/github/jbellis/jvector/disk/SeekableSink.java

@@ -0,0 +1,65 @@
+/*
+ * 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 io.github.jbellis.jvector.annotations.Experimental;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+
+/**
+ * A seekable region that supports positional reads and writes, addressed in coordinates relative
+ * to the region's start (0-based). An embedder uses it to hand a compactor (or other writer) a
+ * bounded window inside a larger container file: positions are region-relative and the
+ * implementation adds the container's base offset, so the writer never needs to know the absolute
+ * offset.
+ *
+ * <p>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 <b>not</b> 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);
+    }
+}

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 <b>immediately</b> (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) {

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 <em>how</em> 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.
+ * <p>
+ * 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 <em>sequential</em> 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 <em>sequential</em> 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 <T>    the stream element type
+     */
+    <T> void forEach(Stream<T> source, Consumer<T> body);
+
+    /**
+     * Returns an executor backed by {@code pool}: each iteration is hosted as a parallel stream on
+     * that pool and the calling thread blocks on the result. This reproduces the behavior of the
+     * {@code ForkJoinPool}-based {@link GraphIndexBuilder} constructors.
+     *
+     * @param pool the pool that hosts the parallel iterations
+     * @return a pool-backed {@code ParallelExecutor}
+     */
+    static ParallelExecutor forkJoin(ForkJoinPool pool) {
+        return new ParallelExecutor() {
+            @Override
+            public void forEachInt(int upperBound, IntConsumer body) {
+                pool.submit(() -> IntStream.range(0, upperBound).parallel().forEach(body)).join();
+            }
+
+            @Override
+            public void forEach(IntStream source, IntConsumer body) {
+                pool.submit(() -> source.parallel().forEach(body)).join();
+            }
+
+            @Override
+            public <T> void forEach(Stream<T> source, Consumer<T> body) {
+                pool.submit(() -> source.parallel().forEach(body)).join();
+            }
+        };
+    }
+
+    /**
+     * Returns an executor that runs every iteration sequentially on the calling thread — no worker
+     * threads, no pool, and the common pool is left untouched. Graph structure and recall are
+     * equivalent to the {@link #forkJoin(ForkJoinPool)} path; only wall-clock and thread usage differ.
+     *
+     * @return a caller-runs {@code ParallelExecutor}
+     */
+    static ParallelExecutor callerRuns() {
+        return new ParallelExecutor() {
+            @Override
+            public void forEachInt(int upperBound, IntConsumer body) {
+                for (int i = 0; i < upperBound; i++) {
+                    body.accept(i);
+                }
+            }
+
+            @Override
+            public void forEach(IntStream source, IntConsumer body) {
+                source.forEach(body);
+            }
+
+            @Override
+            public <T> void forEach(Stream<T> source, Consumer<T> body) {
+                source.forEach(body);
+            }
+        };
+    }
+}

jvector-base/src/main/java/io/github/jbellis/jvector/graph/disk/CompactionDestination.java

@@ -0,0 +1,83 @@
+/*
+ * 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 io.github.jbellis.jvector.annotations.Experimental;
+import io.github.jbellis.jvector.disk.SeekableSink;
+
+import java.io.IOException;
+import java.nio.file.Path;
+
+/**
+ * Embedding extension point: tells {@link OnDiskGraphIndexCompactor} WHERE to write its compacted
+ * graph, so the body lands directly inside the embedder's container (after a header the embedder
+ * reserves) — eliminating the temp-file-and-copy. Resource-scoped: the compactor uses one
+ * {@link Target} per {@code compact(...)} call, commits on success, and always closes.
+ *
+ * <pre>{@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)
+ * }</pre>
+ *
+ * <p>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);
+    }
+}