fix(engine): keep shared files open when a caller's thread is interrupted

docs/storage/pages.md

@@ -317,6 +317,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`):

engine/src/main/java/io/hstore/engine/StorageEngine.java

@@ -396,24 +396,31 @@ public void halt() {
     }
 
     private void shutdown() {
-        closed = true;
-        transactions.shutdown();
-        references.close();
-        maintainer.interrupt();
+        boolean interrupted = Thread.interrupted();
         try {
-            maintainer.join(Duration.ofSeconds(2));
-        } 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(Duration.ofSeconds(2));
+            } 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();
+            }
         }
     }
 }

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,15 +112,15 @@ private void restore() throws IOException {
         try (Stream<Path> 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()) {
             return;
         }
         Segment last = segments.lastEntry().getValue();
         long position = segments.firstKey();
-        for (Iterator<Frame> frames = new FrameReader(position, last.start + last.channel.size()); frames.hasNext(); ) {
+        for (Iterator<Frame> 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<Long, Segment> 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;
+            });
         }
     }
 

engine/src/main/java/io/hstore/engine/maintenance/Checkpointer.java

@@ -9,6 +9,7 @@
 import io.hstore.engine.catalog.Generation;
 import io.hstore.engine.catalog.SlotRegistry;
 import io.hstore.engine.feed.ChangeFeed;
+import io.hstore.engine.page.FileHandle;
 import io.hstore.engine.page.PageStore;
 import io.hstore.engine.tree.Epoch;
 import io.hstore.engine.tree.Materializer;
@@ -73,6 +74,10 @@ public Generation materialized() {
     }
 
     public Result checkpoint() {
+        return FileHandle.uninterrupted(this::take);
+    }
+
+    private Result take() {
         Generation target = transactions.current();
         Epoch epoch = new Epoch();
         LongStream.Builder written = LongStream.builder();

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> {
+        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> T uninterrupted(Supplier<T> work) {
+        boolean interrupted = Thread.interrupted();
+        try {
+            return work.get();
+        } finally {
+            if (interrupted) {
+                Thread.currentThread().interrupt();
+            }
+        }
+    }
+
+    public <T> T call(Io<T> 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();
+        }
+    }
+}