This is an automated email from the ASF dual-hosted git repository. spmallette pushed a commit to branch tinkergraph-storage in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit eb2b2a287ac34bb71e0716c5a642d6618ecb0094 Author: Stephen Mallette <[email protected]> AuthorDate: Mon Aug 3 20:03:06 2026 +0000 Make TinkerStorageGraph commits fsync-durable with a configurable sync mode The GraphBinary storage engine's flush() only pushed bytes into the OS page cache, so an acknowledged commit could be lost on OS crash or power loss. flush() now fsyncs on commit, and compaction is made crash-safe: the snapshot is fsync'd and atomically renamed into place with directory fsyncs before the log is truncated. A new gremlin.tinkergraph.storage.sync config key selects the durability mode: 'commit' (default, fsync every commit) or 'os' (flush to the OS only; survives process crash but not power loss). Assisted-by: Claude Code:claude-opus-4-8 --- CHANGELOG.asciidoc | 2 +- .../gremlin/tinkergraph/structure/TinkerGraph.java | 8 +++ .../structure/storage/GraphBinaryStorage.java | 80 +++++++++++++++++++--- .../tinkergraph/structure/storage/SyncMode.java | 70 +++++++++++++++++++ .../structure/storage/GraphBinaryStorageTest.java | 51 ++++++++++++++ .../structure/storage/SyncModeTest.java | 44 ++++++++++++ 6 files changed, 244 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc index 2571547e52..414863edeb 100644 --- a/CHANGELOG.asciidoc +++ b/CHANGELOG.asciidoc @@ -28,7 +28,7 @@ image::https://raw.githubusercontent.com/apache/tinkerpop/master/docs/static/ima * Fixed `gremlin-go` to report a malformed or truncated GraphBinary response as a deserialization error rather than a bare decoder message. * Made `TinkerGraph` an interface and renamed the in-memory implementation to `TinkerMemoryGraph`; `TinkerGraph.open()` and `gremlin.graph=...TinkerGraph` behave as before. *(breaking)* * Renamed `TinkerTransactionGraph` to `TinkerStorageGraph`. *(breaking)* -* Added a pluggable storage layer to `TinkerStorageGraph` that durably persists each committed transaction to disk, selected with the `gremlin.tinkergraph.storage` config key and shipping a GraphBinary engine. +* Added a pluggable storage layer to `TinkerStorageGraph` that durably persists each committed transaction to disk, selected with the `gremlin.tinkergraph.storage` config key and shipping a GraphBinary engine, with a `gremlin.tinkergraph.storage.sync` key to choose `commit` (fsync per commit) or `os` durability. * Removed automatic persistence from `TinkerMemoryGraph`, which is now purely in-memory and ignores `gremlin.tinkergraph.graphLocation`/`graphFormat`. Use `TinkerStorageGraph` for durability or `g.io()` for interchange. *(breaking)* [[release-4-0-0-beta-3]] diff --git a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/TinkerGraph.java b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/TinkerGraph.java index 0935c38fdb..3bb5200bef 100644 --- a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/TinkerGraph.java +++ b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/TinkerGraph.java @@ -59,6 +59,14 @@ public interface TinkerGraph extends Graph { * the graph holds data only in memory. Not valid on {@link TinkerMemoryGraph}. */ String GREMLIN_TINKERGRAPH_STORAGE = "gremlin.tinkergraph.storage"; + /** + * The durability mode a {@link TinkerStorageGraph} storage engine applies on commit. Either {@code commit} + * (default) to {@code fsync} every commit so acknowledged commits survive an OS crash or power loss, or {@code os} + * to only flush to the operating system so commits survive a JVM process crash but may be lost on OS crash or + * power loss. Only meaningful when {@link #GREMLIN_TINKERGRAPH_STORAGE} is set. See + * {@code org.apache.tinkerpop.gremlin.tinkergraph.structure.storage.SyncMode}. + */ + String GREMLIN_TINKERGRAPH_STORAGE_SYNC = "gremlin.tinkergraph.storage.sync"; String GREMLIN_TINKERGRAPH_ALLOW_NULL_PROPERTY_VALUES = "gremlin.tinkergraph.allowNullPropertyValues"; String GREMLIN_TINKERGRAPH_SERVICE = "gremlin.tinkergraph.service"; String GREMLIN_TINKERGRAPH_VERTEX_LABEL_CARDINALITY = "gremlin.tinkergraph.vertexLabelCardinality"; diff --git a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/GraphBinaryStorage.java b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/GraphBinaryStorage.java index b07891a93c..9b31349a2a 100644 --- a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/GraphBinaryStorage.java +++ b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/GraphBinaryStorage.java @@ -44,6 +44,11 @@ import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; import java.io.UncheckedIOException; +import java.nio.channels.FileChannel; +import java.nio.file.AtomicMoveNotSupportedException; +import java.nio.file.Files; +import java.nio.file.StandardCopyOption; +import java.nio.file.StandardOpenOption; import java.util.ArrayList; import java.util.Collection; import java.util.Iterator; @@ -58,6 +63,10 @@ import java.util.Map; * folded result re-applied to the in-memory graph. {@link #compact(AbstractTinkerGraph)} rewrites the snapshot from the * current committed state and truncates the log. * <p/> + * On commit the appended record is made durable according to the configured {@link SyncMode}: {@link SyncMode#COMMIT} + * (default) {@code fsync}s so the commit survives an OS crash or power loss, while {@link SyncMode#OS} only flushes to + * the operating system. + * <p/> * The in-memory graph remains authoritative (write-through). This engine does not support graphs larger than memory. */ public final class GraphBinaryStorage implements TinkerStorage { @@ -83,6 +92,8 @@ public final class GraphBinaryStorage implements TinkerStorage { private File logFile; private DataOutputStream logOut; + private FileOutputStream logFos; + private SyncMode syncMode = SyncMode.COMMIT; private boolean closed = false; @Override @@ -94,6 +105,7 @@ public final class GraphBinaryStorage implements TinkerStorage { this.directory = new File(location); this.snapshotFile = new File(directory, SNAPSHOT_FILE); this.logFile = new File(directory, LOG_FILE); + this.syncMode = SyncMode.fromConfigValue(config.getString(TinkerGraph.GREMLIN_TINKERGRAPH_STORAGE_SYNC, null)); ensureDirectory(); } @@ -242,7 +254,12 @@ public final class GraphBinaryStorage implements TinkerStorage { return; if (logOut != null) { try { + // flush the JVM buffer into the OS page cache; durable against a JVM process crash logOut.flush(); + // in COMMIT mode also force the OS page cache to the device, so an acknowledged commit is durable + // against an OS crash or power loss. OS mode stops at the flush above and accepts that weaker guarantee. + if (syncMode == SyncMode.COMMIT) + logFos.getFD().sync(); } catch (IOException ex) { throw new UncheckedIOException("Could not flush storage log", ex); } @@ -253,27 +270,67 @@ public final class GraphBinaryStorage implements TinkerStorage { public void compact(final AbstractTinkerGraph graph) { if (closed) return; - // Write a fresh snapshot of the current committed state, then truncate the log. + // Write a fresh snapshot of the current committed state, then truncate the log. This must be crash-safe: at + // no point may a crash leave the store without a readable snapshot-or-log covering the committed state. + // Ordering is write-tmp -> fsync tmp -> atomically rename tmp over the snapshot -> fsync dir (the rename is + // now durable) -> delete the log -> fsync dir. The old snapshot is only ever replaced by an atomic rename, so + // a crash at any step leaves either the old (snapshot + log) or the new (snapshot) intact — never neither. closeLog(); ensureDirectory(); final File tmp = new File(directory, SNAPSHOT_FILE + ".tmp"); - try (final DataOutputStream out = new DataOutputStream(new BufferedOutputStream(new FileOutputStream(tmp)))) { + try (final FileOutputStream fos = new FileOutputStream(tmp); + final DataOutputStream out = new DataOutputStream(new BufferedOutputStream(fos))) { final byte[] frame = encodeSnapshot(graph); if (frame.length > 0) writeFrame(out, frame); out.flush(); + // force the snapshot's bytes to the device before it is renamed into place + fos.getFD().sync(); } catch (IOException ex) { throw new UncheckedIOException("Could not write storage snapshot", ex); } - if (snapshotFile.exists() && !snapshotFile.delete()) - throw new UncheckedIOException(new IOException("Could not replace snapshot " + snapshotFile)); - if (!tmp.renameTo(snapshotFile)) - throw new UncheckedIOException(new IOException("Could not rename snapshot into place " + snapshotFile)); + try { + // atomically replace the snapshot; no delete-then-rename window where the snapshot is briefly absent + atomicMove(tmp, snapshotFile); + // fsync the directory so the rename survives a crash before we touch the log + syncDirectory(); + + // truncate the log now that the snapshot durably reflects the committed state + if (logFile.exists() && !logFile.delete()) + throw new IOException("Could not truncate storage log " + logFile); + // fsync the directory again so the log's removal is durable + syncDirectory(); + } catch (IOException ex) { + throw new UncheckedIOException("Could not finalize storage snapshot", ex); + } + } - // truncate the log - if (logFile.exists() && !logFile.delete()) - throw new UncheckedIOException(new IOException("Could not truncate storage log " + logFile)); + /** + * Atomically move {@code source} onto {@code target}, replacing any existing target. Falls back to a non-atomic + * replacing move on filesystems that do not support atomic moves. + */ + private static void atomicMove(final File source, final File target) throws IOException { + try { + Files.move(source.toPath(), target.toPath(), + StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING); + } catch (AtomicMoveNotSupportedException anse) { + Files.move(source.toPath(), target.toPath(), StandardCopyOption.REPLACE_EXISTING); + } + } + + /** + * fsync the storage directory so that recent namespace changes (a rename into place, a file deletion) are durable. + * A directory fsync is required because those operations only update the directory entry, which the earlier file + * fsync does not cover. + */ + private void syncDirectory() { + try (final FileChannel dirChannel = FileChannel.open(directory.toPath(), StandardOpenOption.READ)) { + dirChannel.force(true); + } catch (IOException ex) { + // some platforms (notably Windows) cannot open a directory as a channel; the atomic rename is the + // durability guarantee there, so treat inability to sync the directory as non-fatal + } } /** @@ -316,7 +373,9 @@ public final class GraphBinaryStorage implements TinkerStorage { private void ensureLogOpen() { if (logOut == null) { try { - logOut = new DataOutputStream(new BufferedOutputStream(new FileOutputStream(logFile, true))); + // retain the FileOutputStream so flush() can reach its FileDescriptor for fsync + logFos = new FileOutputStream(logFile, true); + logOut = new DataOutputStream(new BufferedOutputStream(logFos)); } catch (IOException ex) { throw new UncheckedIOException("Could not open storage log for append", ex); } @@ -332,6 +391,7 @@ public final class GraphBinaryStorage implements TinkerStorage { throw new UncheckedIOException("Could not close storage log", ex); } finally { logOut = null; + logFos = null; } } } diff --git a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/SyncMode.java b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/SyncMode.java new file mode 100644 index 0000000000..de049797fe --- /dev/null +++ b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/SyncMode.java @@ -0,0 +1,70 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.tinkerpop.gremlin.tinkergraph.structure.storage; + +/** + * The durability mode a {@link TinkerStorage} engine applies when flushing a committed transaction to disk, selected + * with the {@code gremlin.tinkergraph.storage.sync} configuration key. Each value is a complete, mutually-exclusive + * choice; they are ordered from strongest to weakest durability. + * <p/> + * The name of a mode describes <em>when</em> data is made durable, not <em>what</em> action is taken: {@link #COMMIT} + * performs an {@code fsync} so acknowledged commits survive an OS crash or power loss, whereas {@link #OS} only pushes + * bytes into the operating system's page cache, so commits survive a crash of the JVM process but not of the OS. + */ +public enum SyncMode { + + /** + * {@code fsync} on every commit. An acknowledged commit is durable against process crash, OS crash, and power + * loss. This is the default and the mode that honors the "each committed transaction is durably written to disk" + * contract. + */ + COMMIT, + + /** + * Flush to the operating system on every commit, but do not {@code fsync}. An acknowledged commit survives a crash + * of the JVM process but may be lost on an OS crash or power loss, since the data can still be sitting in the OS + * page cache. Faster than {@link #COMMIT}; use only when that weaker guarantee is acceptable. + */ + OS; + + // TODO: add an INTERVAL mode (group commit) — a peer value on this same key, encoded as "interval:<ms>", that + // fsyncs at most once per <ms> window rather than once per commit, bounding the crash-loss window by time while + // amortizing fsync cost across commits. It implies fsync (a batched COMMIT), so it slots in as a third + // mutually-exclusive mode without changing the meaning of COMMIT or OS. Deferred until concurrent commits are + // serialized, since group commit's payoff is amortizing one fsync across many concurrent committers. + + /** + * Resolve a configuration value to a {@link SyncMode}, matched case-insensitively, defaulting to {@link #COMMIT} + * when unset. + * + * @param configValue the raw configuration value, or {@code null} when unset + * @return the resolved mode + * @throws IllegalArgumentException if the value does not name a known mode + */ + public static SyncMode fromConfigValue(final String configValue) { + if (null == configValue) + return COMMIT; + try { + return SyncMode.valueOf(configValue.trim().toUpperCase()); + } catch (IllegalArgumentException iae) { + throw new IllegalArgumentException(String.format( + "Unknown storage sync mode '%s'; valid values are 'commit' and 'os'", configValue), iae); + } + } +} diff --git a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/GraphBinaryStorageTest.java b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/GraphBinaryStorageTest.java index 43c53d66f0..4bdf5b8888 100644 --- a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/GraphBinaryStorageTest.java +++ b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/GraphBinaryStorageTest.java @@ -18,6 +18,7 @@ */ package org.apache.tinkerpop.gremlin.tinkergraph.structure.storage; +import org.apache.commons.configuration2.Configuration; import org.apache.tinkerpop.gremlin.structure.T; import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerGraph; import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerStorageGraph; @@ -53,6 +54,56 @@ public class GraphBinaryStorageTest extends AbstractTinkerStorageConformanceTest assertTrue(new File(location, GraphBinaryStorage.SNAPSHOT_FILE).exists()); } + @Test + public void shouldPersistWithOsSyncMode() { + // 'os' is a weaker durability mode (no fsync); a graceful close/reopen must still round-trip the data. The + // OS-crash-loss window that distinguishes it from 'commit' cannot be exercised in a unit test. + final Configuration conf = config(); + conf.setProperty(TinkerGraph.GREMLIN_TINKERGRAPH_STORAGE_SYNC, "os"); + TinkerStorageGraph graph = TinkerStorageGraph.open(conf); + graph.addVertex(T.id, 1, "value", 42); + graph.tx().commit(); + graph.close(); + + graph = TinkerStorageGraph.open(conf); + assertEquals(1, countOf(graph.vertices())); + assertEquals(Integer.valueOf(42), graph.vertices(1).next().value("value")); + graph.close(); + } + + @Test + public void shouldPersistWithDefaultCommitSyncMode() { + // with no sync mode configured the engine defaults to 'commit' (fsync per commit); data must round-trip. + TinkerStorageGraph graph = open(); + graph.addVertex(T.id, 1, "value", 42); + graph.tx().commit(); + graph.close(); + + graph = open(); + assertEquals(Integer.valueOf(42), graph.vertices(1).next().value("value")); + graph.close(); + } + + @Test(expected = IllegalArgumentException.class) + public void shouldRejectUnknownSyncMode() { + final Configuration conf = config(); + conf.setProperty(TinkerGraph.GREMLIN_TINKERGRAPH_STORAGE_SYNC, "bogus"); + TinkerStorageGraph.open(conf); + } + + @Test + public void shouldLeaveNoTempSnapshotAfterCompaction() { + final TinkerStorageGraph graph = open(); + final String location = graph.configuration().getString(TinkerGraph.GREMLIN_TINKERGRAPH_GRAPH_LOCATION); + graph.addVertex(T.id, 1, "value", 1); + graph.tx().commit(); + graph.compact(); + // the atomic rename must consume the temp file, leaving a durable snapshot and no leftover .tmp + assertTrue(new File(location, GraphBinaryStorage.SNAPSHOT_FILE).exists()); + assertTrue(!new File(location, GraphBinaryStorage.SNAPSHOT_FILE + ".tmp").exists()); + graph.close(); + } + @Test public void shouldRecoverFromTruncatedTrailingFrame() throws Exception { TinkerStorageGraph graph = open(); diff --git a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/SyncModeTest.java b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/SyncModeTest.java new file mode 100644 index 0000000000..be1ddd2607 --- /dev/null +++ b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/SyncModeTest.java @@ -0,0 +1,44 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.tinkerpop.gremlin.tinkergraph.structure.storage; + +import org.junit.Test; + +import static org.junit.Assert.assertEquals; + +public class SyncModeTest { + + @Test + public void shouldDefaultToCommitWhenUnset() { + assertEquals(SyncMode.COMMIT, SyncMode.fromConfigValue(null)); + } + + @Test + public void shouldResolveCaseInsensitivelyAndTrim() { + assertEquals(SyncMode.COMMIT, SyncMode.fromConfigValue("commit")); + assertEquals(SyncMode.COMMIT, SyncMode.fromConfigValue("COMMIT")); + assertEquals(SyncMode.OS, SyncMode.fromConfigValue("os")); + assertEquals(SyncMode.OS, SyncMode.fromConfigValue(" Os ")); + } + + @Test(expected = IllegalArgumentException.class) + public void shouldRejectUnknownValue() { + SyncMode.fromConfigValue("interval:1000"); + } +}
