fix(engine): keep at most a fixed number of segment files open

docs/architecture/overview.md

@@ -83,7 +83,7 @@ The engine has no dependencies outside `java.base`. All three modules build into
 | Package | Purpose |
 |---|---|
 | `io.hstore.engine` | `StorageEngine` (open/close, maintenance thread, checkpoint/compact entry points), `EngineOptions`, `EngineStats`, `Dictionary` (interned role names and role sets), `HStoreException` (error codes, retryable conflicts). |
-| `engine.page` | Byte-level storage. `PageStore` and `SegmentFile` hold segment files and allocation. `PageHeader` is the 80-byte header and crc32c seal. `PageId` is a page's permanent number, `PageAddress` its place in a segment, and `PageDirectory` the memory-mapped `pages.dir` that maps one to the other. `ReferenceCounts` is `pages.ref`, the count of references to each page, with its checkpoint journal. `ByteCursor` provides little-endian and varint codecs over `MemorySegment`. `IoTrace` counts page visits per query through a `ScopedValue`. |
+| `engine.page` | Byte-level storage. `PageStore` and `SegmentFile` hold segment files and allocation, and `FilePool` bounds how many segment files are open at once. `PageHeader` is the 80-byte header and crc32c seal. `PageId` is a page's permanent number, `PageAddress` its place in a segment, and `PageDirectory` the memory-mapped `pages.dir` that maps one to the other. `ReferenceCounts` is `pages.ref`, the count of references to each page, with its checkpoint journal. `ByteCursor` provides little-endian and varint codecs over `MemorySegment`. `IoTrace` counts page visits per query through a `ScopedValue`. |
 | `engine.tree` | The copy-on-write counted B+tree. `Tree` is the immutable handle and `TreeWriter` does path-copy updates. Nodes are `Leaf` and `Branch`, referenced by `Ref` (`Empty`, `Pending`, `Stored`, or `Inline` for a small leaf carried inside an edge record). Also here: `Summary` (monoid annotations), `TreeCursor`, `TreeAlgebra` (leapfrog, set algebra), `TreeDiff`, `BulkBuilder`, `NodeCodec`, `NodeCache`, `Materializer` (writes pending nodes to pages at checkpoints), `Epoch` (one checkpoint attempt, for the stamps on written nodes), `TreeWalker` and `TreeVerifier`. |
 | `engine.topology` | Hyperedge representation: `Hyperedge` (members tree plus order index), `Incidence` (one member's role set, weight, validity, payload and qualifier), `IncidenceCodec` (columnar leaf encoding), `MemberChange`, `Weight` (fixed point) and `Validity`. |
 | `engine.index` | `PostingIndex`, a two-level index from a key to a postings set. Small sets are inlined and large ones are promoted to their own tree. Used by the reverse incidence, type, canonical-key and symbol indexes. |

docs/operations/configuration.md

@@ -92,6 +92,7 @@ max_connections          = 200
 | `durability` | `HSTORE_DURABILITY` | `sync` | `sync`: a commit is acknowledged after the group-commit barrier has `fsync`ed it. `async`: acknowledged once appended, so a crash can lose a suffix of acknowledged commits but never corrupts the database. |
 | `wal_mode` | `HSTORE_WAL_MODE` | `references` | Applies to the pages a commit writes itself: spilled pages of bulk loads and trees that changed in more than 4,096 places. `references`: the WAL logs `PageRef(pageId, address, length, crc32c)` records and those pages are synced before the WAL. `images`: the WAL logs their full images. Ordinary commits log only the keys they changed, and pages written at checkpoints are not logged in either mode. See [WAL and recovery](../transactions/wal-and-recovery.md). |
 | `node_cache_mb` | `HSTORE_NODE_CACHE_MB` | `256` | Megabytes of Java heap for decoded tree nodes, each counted at an estimate of the memory it takes. |
+| `max_open_segment_files` | `HSTORE_MAX_OPEN_SEGMENT_FILES` | `256` | Segment files kept open at once. A segment file is opened when a page in it is written or read without its memory map, and once the limit is reached the least recently used file that no thread is using is closed to make room, after its unsynced writes are forced to disk. Must be between 1 and 2,147,483,647. |
 | `checkpoint_wal_mb` | `HSTORE_CHECKPOINT_WAL_MB` | `256` | WAL volume, or volume of pages written, since the last checkpoint that triggers a background checkpoint. |
 | `write_buffer_mb` | `HSTORE_WRITE_BUFFER_MB` | `64` | Megabytes of Java heap held by committed tree nodes that no checkpoint has written yet. A background checkpoint starts once more than half of it is in use; a commit that finds it full waits for a checkpoint, for at most 30 seconds. Must be between 1 and 16,777,216. |
 | `history_limit` | `HSTORE_HISTORY_LIMIT` | `64` | Committed generations kept addressable for `AT GENERATION`, `AS OF`, `HISTORY` and the Studio time slider. History is kept in memory only: after a restart it holds the checkpointed generation and the generations replayed from the WAL after it. |
@@ -157,6 +158,13 @@ The value must be between 1 and 16,777,216 (16 TiB). Two older names are refused
 `cache_mb`, the name this setting had before it was released, and `cache_nodes`, a node count. Both messages name
 `node_cache_mb`.
 
+**`max_open_segment_files`.** Each open segment file holds one file descriptor. Reads from sealed segments go
+through a read-only memory map, and a map stays valid after its file is closed, so in steady state only the
+segment being written and the ones written or remapped recently are open. Closing a file costs an `fsync` if it
+has unsynced writes, and opening it again costs one `open` call. The write-ahead log, the change feed,
+`pages.dir`, `pages.ref` and client connections need descriptors too, so keep the process limit (`ulimit -n`)
+well above this value.
+
 **`checkpoint_wal_mb`.** A checkpoint writes the tree nodes that commits left in memory, publishes the catalog
 image of the generation it wrote, and truncates WAL segments before the checkpoint LSN. Lower values shorten recovery, since recovery replays WAL after the last checkpoint,
 at the cost of more frequent catalog writes. Recovery time scales roughly linearly with WAL volume since the

docs/storage/pages.md

@@ -281,7 +281,7 @@ sequenceDiagram
 * Allocation is serialized by a `ReentrantLock`, and only the counter and directory updates run under it.
 * Rolling a segment marks the previous one `SEALED`, creates the next id, forces the `data/segments` directory so the new file's entry is durable before any page in it is, and resets `nextUnit` to 0.
 * `PageStore.write(pageId, image)` looks the number up in the directory and performs a positional `FileChannel.write` at `offset × 64`, then marks the file dirty. It refuses an image longer than the units allocated for it. A checkpoint writes each image as soon as it has allocated it. For the pages a commit writes, the write happens later, inside `TransactionManager.append`, after the WAL records have been appended (see [wal-and-recovery.md](../transactions/wal-and-recovery.md)).
-* `PageStore.sync()` calls `FileChannel.force(false)` on every dirty segment. The group committer calls it once per batch in `PAGE_REFERENCES` WAL mode, before the WAL sync, if a commit in the batch wrote pages. `PageStore.checkpoint()` does the same and then forces the page directory; only `Checkpointer` calls it.
+* `PageStore.sync()` calls `FileChannel.force(false)` on every segment written since the last sync, and leaves every other segment alone, open or not. The group committer calls it once per batch in `PAGE_REFERENCES` WAL mode, before the WAL sync, if a commit in the batch wrote pages. `PageStore.checkpoint()` does the same and then forces the page directory; only `Checkpointer` calls it.
 
 Compared with fixed page slots, packing removes almost all internal fragmentation for the many small nodes a hypergraph produces. A member tree stored as a page is still an image of at least 128 bytes (two units), even for two members. That is why small member trees are not stored as pages at all, but inline in their edge record ([persistent-tree.md](persistent-tree.md#inline-trees)).
 
@@ -326,6 +326,26 @@ 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.
 
+### Open files (`FilePool`)
+
+A store does not keep a file open per segment. `PageStore` owns a `FilePool` of at most `max_open_segment_files`
+(`EngineOptions.maxOpenSegmentFiles`, default 256) open `FileHandle`s, and every write, `pread`, map and sync on a
+segment borrows its handle from the pool for the length of that one call:
+
+* Opening the store only lists the segment files and reads their sizes with `stat`. A file is opened the first
+  time a call needs it, and rolling to a new segment opens (and creates) the new file straight away.
+* A handle in use by a call is never closed. When the pool is full, the handle that has gone unused the longest is
+  closed to make room. If every handle is in use, the caller waits until one is returned.
+* A file closed to make room is forced first if it has writes no sync has covered yet. A failed force puts the file
+  back in the unsynced set and fails the call that needed the room.
+* Each write marks its file unsynced. `sync()` forces only the unsynced files and takes them out of the set.
+* Deleting or closing a segment closes its handle once no call is using it. A deleted segment's entry refuses to
+  open again, so a late sync can't recreate the file.
+
+A memory map does not need its file to stay open, so a sealed segment that is mapped is read without a handle at
+all. Its handle is closed like any other once it is the least recently used. The map is only replaced when the
+segment grows past it, which takes a handle again for the `map` call.
+
 ### Interrupted threads
 
 A `FileChannel` closes itself, for every thread using it, when a thread is interrupted during a read, write, sync or

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

@@ -15,6 +15,7 @@
 public record EngineOptions(
         int pageSize,
         int pagesPerSegment,
+        int maxOpenSegmentFiles,
         long walSegmentBytes,
         long nodeCacheBytes,
         Durability durability,
@@ -33,6 +34,9 @@ public record EngineOptions(
         if (Integer.bitCount(pageSize) != 1 || pageSize < 1024 || pageSize > 1 << 20) {
             throw new IllegalArgumentException("page size must be a power of two between 1 KiB and 1 MiB: " + pageSize);
         }
+        if (maxOpenSegmentFiles < 1) {
+            throw new IllegalArgumentException("at least one segment file must be allowed open: " + maxOpenSegmentFiles);
+        }
         if (compactionLiveRatio <= 0 || compactionLiveRatio >= 1) {
             throw new IllegalArgumentException("compaction live ratio must be between 0 and 1: " + compactionLiveRatio);
         }
@@ -53,77 +57,82 @@ public record EngineOptions(
     }
 
     public static EngineOptions defaults() {
-        return new EngineOptions(16 * 1024, 16 * 1024, 64L << 20, 256L << 20, Durability.SYNC, WalMode.PAGE_REFERENCES, 64,
+        return new EngineOptions(16 * 1024, 16 * 1024, 256, 64L << 20, 256L << 20, Durability.SYNC, WalMode.PAGE_REFERENCES, 64,
                 256L << 20, 64L << 20, 0.5, 100_000, Duration.ofMillis(500), List.of(), List.of(), CrashPoint.Injector.NONE);
     }
 
     public EngineOptions withPageSize(int bytes) {
-        return new EngineOptions(bytes, pagesPerSegment, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit,
+        return new EngineOptions(bytes, pagesPerSegment, maxOpenSegmentFiles, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit,
                 checkpointWalBytes, writeBufferBytes, compactionLiveRatio, feedRetention, maintenanceInterval, slots, derivations, faults);
     }
 
     public EngineOptions withWalSegmentBytes(long bytes) {
-        return new EngineOptions(pageSize, pagesPerSegment, bytes, nodeCacheBytes, durability, walMode, historyLimit,
+        return new EngineOptions(pageSize, pagesPerSegment, maxOpenSegmentFiles, bytes, nodeCacheBytes, durability, walMode, historyLimit,
                 checkpointWalBytes, writeBufferBytes, compactionLiveRatio, feedRetention, maintenanceInterval, slots, derivations, faults);
     }
 
     public EngineOptions withPagesPerSegment(int pages) {
-        return new EngineOptions(pageSize, pages, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit,
+        return new EngineOptions(pageSize, pages, maxOpenSegmentFiles, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit,
+                checkpointWalBytes, writeBufferBytes, compactionLiveRatio, feedRetention, maintenanceInterval, slots, derivations, faults);
+    }
+
+    public EngineOptions withMaxOpenSegmentFiles(int files) {
+        return new EngineOptions(pageSize, pagesPerSegment, files, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit,
                 checkpointWalBytes, writeBufferBytes, compactionLiveRatio, feedRetention, maintenanceInterval, slots, derivations, faults);
     }
 
     public EngineOptions withDurability(Durability mode) {
-        return new EngineOptions(pageSize, pagesPerSegment, walSegmentBytes, nodeCacheBytes, mode, walMode, historyLimit,
+        return new EngineOptions(pageSize, pagesPerSegment, maxOpenSegmentFiles, walSegmentBytes, nodeCacheBytes, mode, walMode, historyLimit,
                 checkpointWalBytes, writeBufferBytes, compactionLiveRatio, feedRetention, maintenanceInterval, slots, derivations, faults);
     }
 
     public EngineOptions withWalMode(WalMode mode) {
-        return new EngineOptions(pageSize, pagesPerSegment, walSegmentBytes, nodeCacheBytes, durability, mode, historyLimit,
+        return new EngineOptions(pageSize, pagesPerSegment, maxOpenSegmentFiles, walSegmentBytes, nodeCacheBytes, durability, mode, historyLimit,
                 checkpointWalBytes, writeBufferBytes, compactionLiveRatio, feedRetention, maintenanceInterval, slots, derivations, faults);
     }
 
     public EngineOptions withHistoryLimit(int generations) {
-        return new EngineOptions(pageSize, pagesPerSegment, walSegmentBytes, nodeCacheBytes, durability, walMode, generations,
+        return new EngineOptions(pageSize, pagesPerSegment, maxOpenSegmentFiles, walSegmentBytes, nodeCacheBytes, durability, walMode, generations,
                 checkpointWalBytes, writeBufferBytes, compactionLiveRatio, feedRetention, maintenanceInterval, slots, derivations, faults);
     }
 
     public EngineOptions withCheckpointWalBytes(long bytes) {
-        return new EngineOptions(pageSize, pagesPerSegment, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit,
+        return new EngineOptions(pageSize, pagesPerSegment, maxOpenSegmentFiles, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit,
                 bytes, writeBufferBytes, compactionLiveRatio, feedRetention, maintenanceInterval, slots, derivations, faults);
     }
 
     public EngineOptions withWriteBufferBytes(long bytes) {
-        return new EngineOptions(pageSize, pagesPerSegment, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit,
+        return new EngineOptions(pageSize, pagesPerSegment, maxOpenSegmentFiles, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit,
                 checkpointWalBytes, bytes, compactionLiveRatio, feedRetention, maintenanceInterval, slots, derivations, faults);
     }
 
     public EngineOptions withMaintenanceInterval(Duration interval) {
-        return new EngineOptions(pageSize, pagesPerSegment, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit,
+        return new EngineOptions(pageSize, pagesPerSegment, maxOpenSegmentFiles, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit,
                 checkpointWalBytes, writeBufferBytes, compactionLiveRatio, feedRetention, interval, slots, derivations, faults);
     }
 
     public EngineOptions withNodeCacheBytes(long bytes) {
-        return new EngineOptions(pageSize, pagesPerSegment, walSegmentBytes, bytes, durability, walMode, historyLimit,
+        return new EngineOptions(pageSize, pagesPerSegment, maxOpenSegmentFiles, walSegmentBytes, bytes, durability, walMode, historyLimit,
                 checkpointWalBytes, writeBufferBytes, compactionLiveRatio, feedRetention, maintenanceInterval, slots, derivations, faults);
     }
 
     public EngineOptions withCompactionLiveRatio(double ratio) {
-        return new EngineOptions(pageSize, pagesPerSegment, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit,
+        return new EngineOptions(pageSize, pagesPerSegment, maxOpenSegmentFiles, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit,
                 checkpointWalBytes, writeBufferBytes, ratio, feedRetention, maintenanceInterval, slots, derivations, faults);
     }
 
     public EngineOptions withFeedRetention(long generations) {
-        return new EngineOptions(pageSize, pagesPerSegment, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit,
+        return new EngineOptions(pageSize, pagesPerSegment, maxOpenSegmentFiles, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit,
                 checkpointWalBytes, writeBufferBytes, compactionLiveRatio, generations, maintenanceInterval, slots, derivations, faults);
     }
 
     public EngineOptions withExtensions(List<Slot<?>> extensionSlots, List<Derivation> extensionDerivations) {
-        return new EngineOptions(pageSize, pagesPerSegment, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit,
+        return new EngineOptions(pageSize, pagesPerSegment, maxOpenSegmentFiles, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit,
                 checkpointWalBytes, writeBufferBytes, compactionLiveRatio, feedRetention, maintenanceInterval, extensionSlots, extensionDerivations, faults);
     }
 
     public EngineOptions withFaults(CrashPoint.Injector injector) {
-        return new EngineOptions(pageSize, pagesPerSegment, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit,
+        return new EngineOptions(pageSize, pagesPerSegment, maxOpenSegmentFiles, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit,
                 checkpointWalBytes, writeBufferBytes, compactionLiveRatio, feedRetention, maintenanceInterval, slots, derivations, injector);
     }
 }

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

@@ -82,7 +82,7 @@ private StorageEngine(Path directory, EngineOptions options, FileChannel lockCha
         CatalogStore catalog = CatalogStore.open(directory.resolve("catalog"));
         CatalogImage checkpoint = Recovery.discover(catalog);
         this.pages = PageStore.open(directory.resolve("data").resolve("segments"), options.pageSize(), options.pagesPerSegment(),
-                checkpoint.segments(), checkpoint.checkpointLsn());
+                options.maxOpenSegmentFiles(), checkpoint.segments(), checkpoint.checkpointLsn());
         this.wal = WriteAheadLog.open(directory.resolve("wal").resolve("segments"), options.walSegmentBytes());
         this.slots = new SlotRegistry(options.slots());
         FeedCodec codec = new FeedCodec(slots);