From e1f18bed6d9438672294c8c5969caf6ea4570abd Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Mon, 12 Oct 2026 03:11:49 +0530 Subject: [PATCH 1/5] feat(engine): add a file handle that survives interrupted callers A FileChannel closes itself for every thread when one thread is interrupted during I/O on it. FileHandle wraps a channel that several threads share: it clears the caller's interrupt flag for the call and sets it again afterwards, and if the channel was closed anyway, by an interrupt in the middle of the call or on another thread, it reopens the file and runs the call again. Calls are expected to be positional, so running one twice reads or writes the same bytes. --- .../io/hstore/engine/page/FileHandle.java | 114 +++++++++++++++++ .../io/hstore/engine/page/FileHandleTest.java | 118 ++++++++++++++++++ 2 files changed, 232 insertions(+) create mode 100644 engine/src/main/java/io/hstore/engine/page/FileHandle.java create mode 100644 engine/src/test/java/io/hstore/engine/page/FileHandleTest.java diff --git a/engine/src/main/java/io/hstore/engine/page/FileHandle.java b/engine/src/main/java/io/hstore/engine/page/FileHandle.java new file mode 100644 index 0000000..57961f7 --- /dev/null +++ b/engine/src/main/java/io/hstore/engine/page/FileHandle.java @@ -0,0 +1,114 @@ +// SPDX-FileCopyrightText: Metacog Labs +// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0 + +package io.hstore.engine.page; + +import java.io.IOException; +import java.nio.channels.ClosedByInterruptException; +import java.nio.channels.ClosedChannelException; +import java.nio.channels.FileChannel; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.util.concurrent.locks.ReentrantLock; +import java.util.function.Supplier; + +public final class FileHandle implements AutoCloseable { + + private static final int MAX_REOPENS = 64; + + @FunctionalInterface + public interface Io { + T apply(FileChannel channel) throws IOException; + } + + @FunctionalInterface + public interface Action { + void run(FileChannel channel) throws IOException; + } + + private final Path path; + private final ReentrantLock reopening = new ReentrantLock(); + private volatile FileChannel channel; + private volatile boolean closed; + + private FileHandle(Path path, FileChannel channel) { + this.path = path; + this.channel = channel; + } + + public static FileHandle open(Path path) throws IOException { + return new FileHandle(path, FileChannel.open(path, StandardOpenOption.CREATE, StandardOpenOption.READ, StandardOpenOption.WRITE)); + } + + public static T uninterrupted(Supplier work) { + boolean interrupted = Thread.interrupted(); + try { + return work.get(); + } finally { + if (interrupted) { + Thread.currentThread().interrupt(); + } + } + } + + public T call(Io io) throws IOException { + boolean interrupted = Thread.interrupted(); + try { + for (int attempt = 0; ; attempt++) { + FileChannel current = channel; + try { + return io.apply(current); + } catch (ClosedChannelException e) { + if (e instanceof ClosedByInterruptException) { + interrupted |= Thread.interrupted(); + } + if (closed || attempt >= MAX_REOPENS) { + throw e; + } + reopen(current); + } + } + } finally { + if (interrupted) { + Thread.currentThread().interrupt(); + } + } + } + + public void run(Action action) throws IOException { + call(channel -> { + action.run(channel); + return null; + }); + } + + public boolean isOpen() { + return !closed; + } + + public long size() throws IOException { + return call(FileChannel::size); + } + + private void reopen(FileChannel failed) throws IOException { + reopening.lock(); + try { + if (!closed && channel == failed) { + channel = FileChannel.open(path, StandardOpenOption.READ, StandardOpenOption.WRITE); + } + } finally { + reopening.unlock(); + } + } + + @Override + public void close() throws IOException { + reopening.lock(); + try { + closed = true; + channel.close(); + } finally { + reopening.unlock(); + } + } +} diff --git a/engine/src/test/java/io/hstore/engine/page/FileHandleTest.java b/engine/src/test/java/io/hstore/engine/page/FileHandleTest.java new file mode 100644 index 0000000..ed9c4dd --- /dev/null +++ b/engine/src/test/java/io/hstore/engine/page/FileHandleTest.java @@ -0,0 +1,118 @@ +// SPDX-FileCopyrightText: Metacog Labs +// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0 + +package io.hstore.engine.page; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.channels.ClosedChannelException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class FileHandleTest { + + @TempDir + Path directory; + + @Test + void aThreadInterruptedBeforehandWritesAndKeepsItsInterrupt() throws Exception { + Path path = directory.resolve("before"); + try (FileHandle file = FileHandle.open(path)) { + Thread.currentThread().interrupt(); + try { + write(file, 0, "first"); + assertTrue(Thread.currentThread().isInterrupted()); + } finally { + Thread.interrupted(); + } + write(file, 5, "second"); + assertEquals("firstsecond", read(file, 11)); + } + assertEquals("firstsecond", Files.readString(path)); + } + + @Test + void anInterruptDuringACallReopensTheFileAndRetries() throws Exception { + try (FileHandle file = FileHandle.open(directory.resolve("during"))) { + write(file, 0, "kept"); + AtomicInteger attempts = new AtomicInteger(); + try { + String text = file.call(channel -> { + if (attempts.getAndIncrement() == 0) { + Thread.currentThread().interrupt(); + } + ByteBuffer buffer = ByteBuffer.allocate(4); + channel.read(buffer, 0); + return new String(buffer.array(), StandardCharsets.UTF_8); + }); + assertEquals("kept", text); + assertEquals(2, attempts.get()); + assertTrue(Thread.currentThread().isInterrupted()); + } finally { + Thread.interrupted(); + } + write(file, 4, "!"); + assertEquals("kept!", read(file, 5)); + } + } + + @Test + void aChannelClosedUnderAnotherThreadIsReopenedForIt() throws Exception { + try (FileHandle file = FileHandle.open(directory.resolve("other"))) { + write(file, 0, "data"); + AtomicInteger attempts = new AtomicInteger(); + long size = file.call(channel -> { + if (attempts.getAndIncrement() == 0) { + Thread closer = Thread.ofPlatform().start(() -> { + try { + channel.close(); + } catch (IOException e) { + throw new IllegalStateException(e); + } + }); + try { + closer.join(); + } catch (InterruptedException e) { + throw new IllegalStateException(e); + } + } + return channel.size(); + }); + assertEquals(4, size); + assertEquals(2, attempts.get()); + assertFalse(Thread.currentThread().isInterrupted()); + } + } + + @Test + void aClosedHandleStaysClosed() throws Exception { + FileHandle file = FileHandle.open(directory.resolve("closed")); + file.close(); + assertThrows(ClosedChannelException.class, file::size); + } + + private static void write(FileHandle file, long position, String text) throws IOException { + ByteBuffer buffer = ByteBuffer.wrap(text.getBytes(StandardCharsets.UTF_8)); + file.run(channel -> { + while (buffer.hasRemaining()) { + channel.write(buffer, position + buffer.position()); + } + }); + } + + private static String read(FileHandle file, int length) throws IOException { + ByteBuffer buffer = ByteBuffer.allocate(length); + file.run(channel -> channel.read(buffer, 0)); + return new String(buffer.array(), StandardCharsets.UTF_8); + } +} From 891454d6f324344e7b8c2c83a3b8559229237df5 Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Mon, 12 Oct 2026 03:11:49 +0530 Subject: [PATCH 2/5] fix(engine): keep segment files and mapped tables open for interrupted callers A commit or read on an interrupted thread closed the segment file it touched, and every thread using that segment failed with ClosedChannelException from then on, including background compaction and the checkpoint at shutdown. Growing pages.dir or pages.ref on an interrupted thread did the same to those files. Do their I/O through FileHandle. --- .../io/hstore/engine/page/MappedLongs.java | 20 +++++----- .../io/hstore/engine/page/SegmentFile.java | 40 ++++++++++--------- .../hstore/engine/page/PageDirectoryTest.java | 19 +++++++++ .../io/hstore/engine/page/PageStoreTest.java | 27 +++++++++++++ 4 files changed, 77 insertions(+), 29 deletions(-) diff --git a/engine/src/main/java/io/hstore/engine/page/MappedLongs.java b/engine/src/main/java/io/hstore/engine/page/MappedLongs.java index ee86d5d..2f32c32 100644 --- a/engine/src/main/java/io/hstore/engine/page/MappedLongs.java +++ b/engine/src/main/java/io/hstore/engine/page/MappedLongs.java @@ -14,7 +14,6 @@ import java.nio.ByteOrder; import java.nio.channels.FileChannel; import java.nio.file.Path; -import java.nio.file.StandardOpenOption; import java.util.Arrays; final class MappedLongs implements AutoCloseable { @@ -28,20 +27,20 @@ final class MappedLongs implements AutoCloseable { private static final VarHandle ENTRY = LONG.varHandle(); private final Path path; - private final FileChannel channel; + private final FileHandle file; private boolean fresh; private volatile MemorySegment[] chunks = new MemorySegment[0]; - private MappedLongs(Path path, FileChannel channel) { + private MappedLongs(Path path, FileHandle file) { this.path = path; - this.channel = channel; + this.file = file; } static MappedLongs open(Path path, int magic, int format, String what) { try { - FileChannel channel = FileChannel.open(path, StandardOpenOption.CREATE, StandardOpenOption.READ, StandardOpenOption.WRITE); - long size = channel.size(); - MappedLongs longs = new MappedLongs(path, channel); + FileHandle file = FileHandle.open(path); + long size = file.size(); + MappedLongs longs = new MappedLongs(path, file); longs.mapThrough(size == 0 ? 0 : (int) ((size - 1) >>> CHUNK_SHIFT)); MemorySegment header = longs.chunks[0]; longs.fresh = header.get(INT, 0) == 0 && header.get(INT, 4) == 0 && header.get(LONG, 8) == 0; @@ -50,7 +49,7 @@ static MappedLongs open(Path path, int magic, int format, String what) { header.set(INT, 4, format); header.force(); } else if (header.get(INT, 0) != magic || header.get(INT, 4) != format) { - channel.close(); + file.close(); throw HStoreException.invalid(what + " " + path + " has an unknown format"); } return longs; @@ -101,7 +100,8 @@ private synchronized void mapThrough(int last) { MemorySegment[] grown = Arrays.copyOf(mapped, last + 1); try { for (int chunk = mapped.length; chunk <= last; chunk++) { - grown[chunk] = channel.map(FileChannel.MapMode.READ_WRITE, chunk * CHUNK_BYTES, CHUNK_BYTES, Arena.ofAuto()); + long offset = chunk * CHUNK_BYTES; + grown[chunk] = file.call(channel -> channel.map(FileChannel.MapMode.READ_WRITE, offset, CHUNK_BYTES, Arena.ofAuto())); } } catch (IOException e) { throw new UncheckedIOException("cannot grow " + path, e); @@ -112,7 +112,7 @@ private synchronized void mapThrough(int last) { @Override public void close() { try { - channel.close(); + file.close(); } catch (IOException e) { throw new UncheckedIOException("cannot close " + path, e); } diff --git a/engine/src/main/java/io/hstore/engine/page/SegmentFile.java b/engine/src/main/java/io/hstore/engine/page/SegmentFile.java index 1068065..815a640 100644 --- a/engine/src/main/java/io/hstore/engine/page/SegmentFile.java +++ b/engine/src/main/java/io/hstore/engine/page/SegmentFile.java @@ -14,7 +14,6 @@ import java.nio.channels.FileChannel; import java.nio.file.Files; import java.nio.file.Path; -import java.nio.file.StandardOpenOption; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.locks.ReentrantLock; @@ -25,23 +24,22 @@ final class SegmentFile implements AutoCloseable { private final int id; private final Path path; - private final FileChannel channel; + private final FileHandle file; private final AtomicBoolean dirty = new AtomicBoolean(); private final ReentrantLock remapping = new ReentrantLock(); private volatile MemorySegment mapped = MemorySegment.NULL; private Arena arena; private volatile boolean sealed; - private SegmentFile(int id, Path path, FileChannel channel) { + private SegmentFile(int id, Path path, FileHandle file) { this.id = id; this.path = path; - this.channel = channel; + this.file = file; } static SegmentFile open(int id, Path path) { try { - FileChannel channel = FileChannel.open(path, StandardOpenOption.CREATE, StandardOpenOption.READ, StandardOpenOption.WRITE); - return new SegmentFile(id, path, channel); + return new SegmentFile(id, path, FileHandle.open(path)); } catch (IOException e) { throw new UncheckedIOException("cannot open segment " + path, e); } @@ -62,7 +60,7 @@ int id() { long unitsOnDisk() { try { - return PageId.unitsFor(channel.size()); + return PageId.unitsFor(file.size()); } catch (IOException e) { throw HStoreException.io("cannot size segment " + path, e); } @@ -104,7 +102,7 @@ private MemorySegment read(MemorySegment view, long position, int maximum) { MemorySegment view() { try { - MemorySegment whole = covering(channel.size()); + MemorySegment whole = covering(file.size()); return whole == null ? MemorySegment.NULL : whole; } catch (IOException e) { throw HStoreException.io("cannot size segment " + path, e); @@ -135,13 +133,13 @@ private MemorySegment covering(long required) { if (required <= current.byteSize()) { return current; } - long size = channel.size(); + long size = file.size(); if (size < required) { return null; } Arena replacement = Arena.ofShared(); try { - current = channel.map(FileChannel.MapMode.READ_ONLY, 0, size, replacement); + current = file.call(channel -> channel.map(FileChannel.MapMode.READ_ONLY, 0, size, replacement)); } catch (IOException | RuntimeException e) { replacement.close(); throw e; @@ -177,11 +175,13 @@ private MemorySegment copy(long position, int maximum) { private void fill(ByteBuffer buffer, long position) { try { - while (buffer.hasRemaining()) { - if (channel.read(buffer, position + buffer.position()) < 0) { - return; + file.run(channel -> { + while (buffer.hasRemaining()) { + if (channel.read(buffer, position + buffer.position()) < 0) { + return; + } } - } + }); } catch (IOException e) { throw HStoreException.io("read failed in segment " + id + " at byte " + position, e); } @@ -190,9 +190,11 @@ private void fill(ByteBuffer buffer, long position) { void write(long position, MemorySegment image) { ByteBuffer buffer = image.asByteBuffer(); try { - while (buffer.hasRemaining()) { - channel.write(buffer, position + buffer.position()); - } + file.run(channel -> { + while (buffer.hasRemaining()) { + channel.write(buffer, position + buffer.position()); + } + }); dirty.set(true); } catch (IOException e) { throw HStoreException.io("write failed in segment " + id + " at byte " + position, e); @@ -202,7 +204,7 @@ void write(long position, MemorySegment image) { void force() { if (dirty.getAndSet(false)) { try { - channel.force(false); + file.run(channel -> channel.force(false)); } catch (IOException e) { dirty.set(true); throw HStoreException.io("sync failed for segment " + id, e); @@ -236,7 +238,7 @@ public void close() { remapping.unlock(); } try { - channel.close(); + file.close(); } catch (IOException e) { throw new UncheckedIOException(e); } diff --git a/engine/src/test/java/io/hstore/engine/page/PageDirectoryTest.java b/engine/src/test/java/io/hstore/engine/page/PageDirectoryTest.java index 975a58d..bfc90cb 100644 --- a/engine/src/test/java/io/hstore/engine/page/PageDirectoryTest.java +++ b/engine/src/test/java/io/hstore/engine/page/PageDirectoryTest.java @@ -13,12 +13,31 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; class PageDirectoryTest { @TempDir Path directory; + @Test + void anInterruptedThreadGrowsTheDirectoryWithoutClosingIt() throws Exception { + long far = 3L << 20; + boolean[] keptInterrupt = new boolean[1]; + try (PageDirectory pages = PageDirectory.open(directory.resolve(PageDirectory.FILE))) { + Thread grower = Thread.ofPlatform().start(() -> { + Thread.currentThread().interrupt(); + pages.put(far, 42); + keptInterrupt[0] = Thread.currentThread().isInterrupted(); + }); + grower.join(); + pages.put(far * 2, 43); + assertTrue(keptInterrupt[0]); + assertEquals(42, pages.get(far)); + assertEquals(43, pages.get(far * 2)); + } + } + @Test void addressesAndTheNextNumberSurviveReopenAcrossChunks() { Path file = directory.resolve(PageDirectory.FILE); diff --git a/engine/src/test/java/io/hstore/engine/page/PageStoreTest.java b/engine/src/test/java/io/hstore/engine/page/PageStoreTest.java index 071e682..15c5105 100644 --- a/engine/src/test/java/io/hstore/engine/page/PageStoreTest.java +++ b/engine/src/test/java/io/hstore/engine/page/PageStoreTest.java @@ -99,6 +99,33 @@ void finishedSegmentsAreMappedAndTheGrowingOneIsNotUntilItGrows() { } } + @Test + void anInterruptedThreadReadsAndWritesPagesWithoutClosingTheSegment() throws Exception { + try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, List.of())) { + long before = write(store, PAGE_SIZE); + byte[] expected = bytes(store.read(before)); + long[] written = new long[1]; + boolean[] stillInterrupted = new boolean[1]; + List seen = new ArrayList<>(); + Thread interrupted = Thread.ofPlatform().start(() -> { + Thread.currentThread().interrupt(); + written[0] = write(store, PAGE_SIZE / 2); + seen.add(bytes(store.read(before))); + store.sync(); + stillInterrupted[0] = Thread.currentThread().isInterrupted(); + }); + interrupted.join(); + assertTrue(stillInterrupted[0]); + assertArrayEquals(expected, seen.getFirst()); + long after = write(store, PAGE_SIZE); + assertEquals(store.segmentOf(before), store.segmentOf(after)); + assertEquals(0, store.mappedBytes(store.segmentOf(after))); + assertArrayEquals(expected, bytes(store.read(before))); + assertArrayEquals(image(new Random(PAGE_SIZE / 2), PAGE_SIZE / 2), bytes(store.read(written[0]))); + store.sync(); + } + } + @Test void pagesAreNumberedInOrderWhicheverSegmentHoldsThem() { try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, List.of())) { From b6500bb6d627f28b4253ddaf178b45c25b052908 Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Mon, 12 Oct 2026 03:11:49 +0530 Subject: [PATCH 3/5] fix(engine): keep the log and the change feed open for interrupted committers Committing on an interrupted thread closed the active write-ahead log segment, so that commit and every later one failed with "write-ahead log append failed" until the process restarted. The change feed's segments had the same problem for commits and for subscribers reading them. Do their I/O through FileHandle. --- .../io/hstore/engine/feed/ChangeFeed.java | 60 ++++++++++--------- .../io/hstore/engine/wal/WriteAheadLog.java | 28 +++++---- 2 files changed, 50 insertions(+), 38 deletions(-) diff --git a/engine/src/main/java/io/hstore/engine/feed/ChangeFeed.java b/engine/src/main/java/io/hstore/engine/feed/ChangeFeed.java index fd7f30c..0782bb2 100644 --- a/engine/src/main/java/io/hstore/engine/feed/ChangeFeed.java +++ b/engine/src/main/java/io/hstore/engine/feed/ChangeFeed.java @@ -6,6 +6,7 @@ import io.hstore.engine.HStoreException; import io.hstore.engine.page.ByteCursor; import io.hstore.engine.page.Checksums; +import io.hstore.engine.page.FileHandle; import java.io.IOException; import java.io.UncheckedIOException; @@ -62,13 +63,13 @@ public interface Gap { private static final class Segment { final long start; final Path path; - final FileChannel channel; + final FileHandle file; volatile long firstGeneration = EMPTY; - Segment(long start, Path path, FileChannel channel) { + Segment(long start, Path path, FileHandle file) { this.start = start; this.path = path; - this.channel = channel; + this.file = file; } } @@ -111,7 +112,7 @@ private void restore() throws IOException { try (Stream listing = Files.list(directory)) { for (Path path : listing.filter(file -> file.getFileName().toString().endsWith(".feed")).toList()) { long start = Long.parseUnsignedLong(path.getFileName().toString().replace(".feed", ""), 16); - segments.put(start, new Segment(start, path, FileChannel.open(path, StandardOpenOption.READ, StandardOpenOption.WRITE))); + segments.put(start, new Segment(start, path, FileHandle.open(path))); } } if (segments.isEmpty()) { @@ -119,7 +120,7 @@ private void restore() throws IOException { } Segment last = segments.lastEntry().getValue(); long position = segments.firstKey(); - for (Iterator frames = new FrameReader(position, last.start + last.channel.size()); frames.hasNext(); ) { + for (Iterator frames = new FrameReader(position, last.start + last.file.size()); frames.hasNext(); ) { Frame frame = frames.next(); remember(frame.generation(), position); position = frame.next(); @@ -146,7 +147,8 @@ private void cutAt(long position) throws IOException { segments.remove(later.start); discard(later); } - holding.channel.truncate(position - holding.start); + long length = position - holding.start; + holding.file.run(channel -> channel.truncate(length)); if (position == holding.start) { holding.firstGeneration = EMPTY; } @@ -185,12 +187,14 @@ public void append(CommitEvent event) { try { Segment active = segments.isEmpty() || end - segments.lastKey() >= segmentBytes ? roll() : segments.lastEntry().getValue(); ByteBuffer buffer = frame.segment().asByteBuffer(); - long position = end; - while (buffer.hasRemaining()) { - position += active.channel.write(buffer, position - active.start); - } + long local = end - active.start; + active.file.run(channel -> { + while (buffer.hasRemaining()) { + channel.write(buffer, local + buffer.position()); + } + }); remember(event.generation(), end); - end = position; + end += buffer.limit(); appended.signalAll(); } catch (IOException e) { throw HStoreException.io("change feed append failed", e); @@ -201,10 +205,10 @@ public void append(CommitEvent event) { private Segment roll() throws IOException { if (!segments.isEmpty()) { - segments.lastEntry().getValue().channel.force(false); + segments.lastEntry().getValue().file.run(channel -> channel.force(false)); } Path path = directory.resolve("%016x.feed".formatted(end)); - Segment segment = new Segment(end, path, FileChannel.open(path, StandardOpenOption.CREATE, StandardOpenOption.READ, StandardOpenOption.WRITE)); + Segment segment = new Segment(end, path, FileHandle.open(path)); segments.put(end, segment); try (FileChannel listing = FileChannel.open(directory, StandardOpenOption.READ)) { listing.force(true); @@ -219,7 +223,7 @@ public void sync() { return; } try { - segments.lastEntry().getValue().channel.force(false); + segments.lastEntry().getValue().file.run(channel -> channel.force(false)); } catch (IOException e) { throw HStoreException.io("change feed sync failed", e); } @@ -262,7 +266,7 @@ public long retain(long keepAfter) { private void discard(Segment segment) { files.writeLock().lock(); try { - segment.channel.close(); + segment.file.close(); Files.deleteIfExists(segment.path); } catch (IOException e) { throw new UncheckedIOException(e); @@ -321,8 +325,8 @@ public void close() { try { appended.signalAll(); for (Segment segment : segments.values()) { - segment.channel.force(true); - segment.channel.close(); + segment.file.run(channel -> channel.force(true)); + segment.file.close(); } } catch (IOException e) { throw new UncheckedIOException(e); @@ -374,14 +378,14 @@ private Frame read() { private Frame readLocked() { Map.Entry entry = segments.floorEntry(position); - if (entry == null || !entry.getValue().channel.isOpen()) { + if (entry == null || !entry.getValue().file.isOpen()) { return null; } - FileChannel channel = entry.getValue().channel; + FileHandle file = entry.getValue().file; long local = position - entry.getKey(); try { ByteBuffer header = ByteBuffer.allocate(FRAME_HEADER); - if (!readFully(channel, header, local)) { + if (!readFully(file, header, local)) { return null; } ByteCursor in = ByteCursor.over(MemorySegment.ofBuffer(header.flip())); @@ -392,7 +396,7 @@ private Frame readLocked() { return null; } ByteBuffer body = ByteBuffer.allocate(length); - if (!readFully(channel, body, local + FRAME_HEADER)) { + if (!readFully(file, body, local + FRAME_HEADER)) { return null; } byte[] payload = body.array(); @@ -404,13 +408,15 @@ private Frame readLocked() { } } - private boolean readFully(FileChannel channel, ByteBuffer buffer, long at) throws IOException { - while (buffer.hasRemaining()) { - if (channel.read(buffer, at + buffer.position()) < 0) { - return false; + private boolean readFully(FileHandle file, ByteBuffer buffer, long at) throws IOException { + return file.call(channel -> { + while (buffer.hasRemaining()) { + if (channel.read(buffer, at + buffer.position()) < 0) { + return false; + } } - } - return true; + return true; + }); } } diff --git a/engine/src/main/java/io/hstore/engine/wal/WriteAheadLog.java b/engine/src/main/java/io/hstore/engine/wal/WriteAheadLog.java index 4d1ad82..9df854a 100644 --- a/engine/src/main/java/io/hstore/engine/wal/WriteAheadLog.java +++ b/engine/src/main/java/io/hstore/engine/wal/WriteAheadLog.java @@ -6,6 +6,7 @@ import io.hstore.engine.HStoreException; import io.hstore.engine.page.ByteCursor; import io.hstore.engine.page.Checksums; +import io.hstore.engine.page.FileHandle; import java.io.IOException; import java.io.UncheckedIOException; @@ -44,7 +45,7 @@ public record Positioned(long lsn, long next, WalRecord record) { private final ReentrantLock lock = new ReentrantLock(); private final NavigableMap segments = new TreeMap<>(); private final AtomicLong bytesAppended = new AtomicLong(); - private FileChannel active; + private FileHandle active; private long activeStart; private long end; private boolean dirty; @@ -76,9 +77,12 @@ private void restore() { validEnd = positioned.next(); } activeStart = last.getKey(); - active = FileChannel.open(last.getValue(), StandardOpenOption.READ, StandardOpenOption.WRITE); - active.truncate(validEnd - activeStart); - active.force(true); + active = FileHandle.open(last.getValue()); + long length = validEnd - activeStart; + active.run(channel -> { + channel.truncate(length); + channel.force(true); + }); end = validEnd; } catch (IOException e) { throw new UncheckedIOException("cannot open write-ahead log " + directory, e); @@ -130,9 +134,11 @@ private void write(MemorySegment bytes) { } ByteBuffer buffer = bytes.asByteBuffer(); long position = end - activeStart; - while (buffer.hasRemaining()) { - position += active.write(buffer, position); - } + active.run(channel -> { + while (buffer.hasRemaining()) { + channel.write(buffer, position + buffer.position()); + } + }); dirty = true; } catch (IOException e) { throw HStoreException.io("write-ahead log append failed", e); @@ -141,12 +147,12 @@ private void write(MemorySegment bytes) { private void roll() throws IOException { if (active != null) { - active.force(true); + active.run(channel -> channel.force(true)); active.close(); } activeStart = end; Path path = pathFor(activeStart); - active = FileChannel.open(path, StandardOpenOption.CREATE, StandardOpenOption.READ, StandardOpenOption.WRITE); + active = FileHandle.open(path); segments.put(activeStart, path); syncDirectory(); } @@ -163,7 +169,7 @@ public void sync() { lock.lock(); try { if (dirty && active != null) { - active.force(false); + active.run(channel -> channel.force(false)); dirty = false; } } catch (IOException e) { @@ -225,7 +231,7 @@ public void close() { lock.lock(); try { if (active != null) { - active.force(true); + active.run(channel -> channel.force(true)); active.close(); active = null; } From 3f5f4c194563f1b674d603671de0c7c70237b4b8 Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Mon, 12 Oct 2026 03:11:49 +0530 Subject: [PATCH 4/5] fix(engine): checkpoint and shut down normally from an interrupted thread A checkpoint writes short-lived files, the catalog and the reference journal, that an interrupt would fail. Shutdown waited for the maintenance thread with join, which returns at once on an interrupted thread, so files could be closed while maintenance was still using them. Run both with the caller's interrupt flag cleared and set it again at the end. The new test commits, reads, replays the feed and checkpoints from an interrupted thread, keeps committing from another one, closes from an interrupted thread and checks everything is there after reopening. --- .../java/io/hstore/engine/StorageEngine.java | 41 +++++++----- .../engine/maintenance/Checkpointer.java | 5 ++ .../java/io/hstore/engine/InterruptTest.java | 65 +++++++++++++++++++ 3 files changed, 94 insertions(+), 17 deletions(-) create mode 100644 engine/src/test/java/io/hstore/engine/InterruptTest.java diff --git a/engine/src/main/java/io/hstore/engine/StorageEngine.java b/engine/src/main/java/io/hstore/engine/StorageEngine.java index c1f0f26..2d21364 100644 --- a/engine/src/main/java/io/hstore/engine/StorageEngine.java +++ b/engine/src/main/java/io/hstore/engine/StorageEngine.java @@ -380,24 +380,31 @@ public void halt() { } private void shutdown() { - closed = true; - transactions.shutdown(); - references.close(); - maintainer.interrupt(); + boolean interrupted = Thread.interrupted(); try { - maintainer.join(MAINTENANCE_INTERVAL.multipliedBy(4)); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - feed.close(); - wal.close(); - pages.close(); - cache.clear(); - try { - lock.release(); - lockChannel.close(); - } catch (IOException e) { - throw new UncheckedIOException(e); + closed = true; + transactions.shutdown(); + references.close(); + maintainer.interrupt(); + try { + maintainer.join(MAINTENANCE_INTERVAL.multipliedBy(4)); + } catch (InterruptedException e) { + interrupted = true; + } + feed.close(); + wal.close(); + pages.close(); + cache.clear(); + try { + lock.release(); + lockChannel.close(); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } finally { + if (interrupted) { + Thread.currentThread().interrupt(); + } } } } diff --git a/engine/src/main/java/io/hstore/engine/maintenance/Checkpointer.java b/engine/src/main/java/io/hstore/engine/maintenance/Checkpointer.java index ddd00e0..e97168f 100644 --- a/engine/src/main/java/io/hstore/engine/maintenance/Checkpointer.java +++ b/engine/src/main/java/io/hstore/engine/maintenance/Checkpointer.java @@ -7,6 +7,7 @@ import io.hstore.engine.catalog.CatalogStore; import io.hstore.engine.catalog.Generation; import io.hstore.engine.feed.ChangeFeed; +import io.hstore.engine.page.FileHandle; import io.hstore.engine.page.PageStore; import io.hstore.engine.txn.CrashPoint; import io.hstore.engine.txn.TransactionManager; @@ -55,6 +56,10 @@ public long completed() { } public Result checkpoint() { + return FileHandle.uninterrupted(this::take); + } + + private Result take() { pages.checkpoint(); Result result = transactions.exclusive(() -> { references.pause(); diff --git a/engine/src/test/java/io/hstore/engine/InterruptTest.java b/engine/src/test/java/io/hstore/engine/InterruptTest.java new file mode 100644 index 0000000..89bb417 --- /dev/null +++ b/engine/src/test/java/io/hstore/engine/InterruptTest.java @@ -0,0 +1,65 @@ +// SPDX-FileCopyrightText: Metacog Labs +// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0 + +package io.hstore.engine; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Path; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class InterruptTest { + + @TempDir + Path directory; + + @Test + void anInterruptedCallerLeavesTheEngineWorking() throws Exception { + StorageEngine engine = StorageEngine.open(directory, EngineTest.small()); + engine.write(txn -> txn.createNode(EngineTest.NODE, "before")); + onInterruptedThread(() -> { + engine.write(txn -> txn.createNode(EngineTest.NODE, "interrupted")); + engine.read(snapshot -> snapshot.resolve(0, EngineTest.NODE, "before")); + engine.feed().replay(0).count(); + engine.checkpoint(); + }); + long events = engine.feed().replay(0).count(); + for (int i = 0; i < 200; i++) { + String key = "after-" + i; + engine.write(txn -> txn.createNode(EngineTest.NODE, key)); + } + assertEquals(events + 200, engine.feed().replay(0).count()); + engine.verifyReferences(); + onInterruptedThread(engine::close); + try (StorageEngine reopened = StorageEngine.open(directory, EngineTest.small())) { + for (String key : List.of("before", "interrupted", "after-199")) { + assertTrue(reopened.read(snapshot -> snapshot.resolve(0, EngineTest.NODE, key)).isPresent(), key); + } + reopened.verifyReferences(); + } + } + + private static void onInterruptedThread(Runnable work) throws InterruptedException { + AtomicReference failure = new AtomicReference<>(); + AtomicBoolean keptInterrupt = new AtomicBoolean(); + Thread thread = Thread.ofPlatform().start(() -> { + Thread.currentThread().interrupt(); + try { + work.run(); + keptInterrupt.set(Thread.currentThread().isInterrupted()); + } catch (Throwable thrown) { + failure.set(thrown); + } + }); + thread.join(); + assertNull(failure.get()); + assertTrue(keptInterrupt.get()); + } +} From dcf94d497ee676e7cf062a875e087a5400e3ab8e Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Mon, 12 Oct 2026 03:11:50 +0530 Subject: [PATCH 5/5] docs(storage): explain how interrupted callers are handled --- docs/storage/pages.md | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/docs/storage/pages.md b/docs/storage/pages.md index 64925c6..990e8aa 100644 --- a/docs/storage/pages.md +++ b/docs/storage/pages.md @@ -301,6 +301,25 @@ ever appended to or deleted, never truncated, so a map can't end up pointing pas A short read at the end of the file returns what it got, and the header check rejects it. Both paths return read-only memory, so nothing can write into a stored image through a read. +### Interrupted threads + +A `FileChannel` closes itself, for every thread using it, when a thread is interrupted during a read, write, sync or +map on it. Commits, reads and feed subscribers run on the caller's threads, and callers do interrupt them, for +example when a request times out. Before this was handled, one interrupted commit closed the write-ahead log and +every later commit failed until the process restarted. + +Segment files, the active write-ahead log segment, the change feed's segments, `pages.dir` and `pages.ref` therefore +go through `page/FileHandle`: + +* It clears the caller's interrupt flag for the length of the call and sets it again afterwards, so the caller still + sees its interrupt. +* If the channel was closed anyway (an interrupt that arrived in the middle of the call, or another thread's + interrupt), it reopens the file and runs the call again. Every read and write is positional, so running it again + reads or writes the same bytes. + +A checkpoint and `close()` also run with the flag cleared, because they write short-lived files such as the catalog +and the reference journal. + ## Caching and I/O accounting `PagedNodeSource.load(pageId, schema)` (`tree/PagedNodeSource.java`):