diff --git a/dashjoin-core/src/main/java/org/dashjoin/service/JSONDatabase.java b/dashjoin-core/src/main/java/org/dashjoin/service/JSONDatabase.java index 8d5db4ef1..1d65a3fff 100644 --- a/dashjoin-core/src/main/java/org/dashjoin/service/JSONDatabase.java +++ b/dashjoin-core/src/main/java/org/dashjoin/service/JSONDatabase.java @@ -6,6 +6,7 @@ import java.util.List; import java.util.Map; import java.util.Map.Entry; +import java.util.concurrent.Callable; import java.util.logging.Level; import org.apache.commons.lang3.NotImplementedException; import org.dashjoin.model.Property; @@ -17,6 +18,7 @@ import com.fasterxml.jackson.core.json.JsonReadFeature; import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.core.util.DefaultIndenter; +import com.fasterxml.jackson.core.util.Instantiatable; import com.fasterxml.jackson.databind.DeserializationFeature; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.SerializationFeature; @@ -58,83 +60,95 @@ public abstract class JSONDatabase implements Database { .enable(SerializationFeature.INDENT_OUTPUT) .disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES) - .setDefaultPrettyPrinter(new PrettyPrinter() { - - int indent = 0; - - void indent(JsonGenerator gen) throws IOException { - for (int i = 0; i < indent; i++) - gen.writeRaw(" "); - } - - void newline(JsonGenerator gen) throws IOException { - gen.writeRaw(DefaultIndenter.SYS_LF); - } - - @Override - public void writeStartObject(JsonGenerator gen) throws IOException { - gen.writeRaw('{'); - newline(gen); - indent++; - } - - @Override - public void writeStartArray(JsonGenerator gen) throws IOException { - gen.writeRaw('['); - newline(gen); - indent++; - } - - @Override - public void writeRootValueSeparator(JsonGenerator gen) throws IOException { - - } - - @Override - public void writeObjectFieldValueSeparator(JsonGenerator gen) throws IOException { - gen.writeRaw(": "); - } - - @Override - public void writeObjectEntrySeparator(JsonGenerator gen) throws IOException { - gen.writeRaw(','); - newline(gen); - indent(gen); - } - - @Override - public void writeEndObject(JsonGenerator gen, int nrOfEntries) throws IOException { - indent--; - newline(gen); - indent(gen); - gen.writeRaw('}'); - } - - @Override - public void writeEndArray(JsonGenerator gen, int nrOfValues) throws IOException { - indent--; - newline(gen); - indent(gen); - gen.writeRaw(']'); - } - - @Override - public void writeArrayValueSeparator(JsonGenerator gen) throws IOException { - gen.writeRaw(','); - newline(gen); - indent(gen); - } - - @Override - public void beforeObjectEntries(JsonGenerator gen) throws IOException { - indent(gen); - } - - @Override - public void beforeArrayValues(JsonGenerator gen) throws IOException { - indent(gen); - } - }); + .setDefaultPrettyPrinter(new JSONPrettyPrinter()); + + /** + * Pretty printer with 4 space indentation and LF line breaks. It keeps the current indent level + * in a field, so it implements Instantiatable: Jackson then creates a fresh copy for every write + * instead of sharing one instance (and one indent counter) between all threads. + */ + static class JSONPrettyPrinter implements PrettyPrinter, Instantiatable { + + int indent = 0; + + @Override + public JSONPrettyPrinter createInstance() { + return new JSONPrettyPrinter(); + } + + void indent(JsonGenerator gen) throws IOException { + for (int i = 0; i < indent; i++) + gen.writeRaw(" "); + } + + void newline(JsonGenerator gen) throws IOException { + gen.writeRaw(DefaultIndenter.SYS_LF); + } + + @Override + public void writeStartObject(JsonGenerator gen) throws IOException { + gen.writeRaw('{'); + newline(gen); + indent++; + } + + @Override + public void writeStartArray(JsonGenerator gen) throws IOException { + gen.writeRaw('['); + newline(gen); + indent++; + } + + @Override + public void writeRootValueSeparator(JsonGenerator gen) throws IOException { + + } + + @Override + public void writeObjectFieldValueSeparator(JsonGenerator gen) throws IOException { + gen.writeRaw(": "); + } + + @Override + public void writeObjectEntrySeparator(JsonGenerator gen) throws IOException { + gen.writeRaw(','); + newline(gen); + indent(gen); + } + + @Override + public void writeEndObject(JsonGenerator gen, int nrOfEntries) throws IOException { + indent--; + newline(gen); + indent(gen); + gen.writeRaw('}'); + } + + @Override + public void writeEndArray(JsonGenerator gen, int nrOfValues) throws IOException { + indent--; + newline(gen); + indent(gen); + gen.writeRaw(']'); + } + + @Override + public void writeArrayValueSeparator(JsonGenerator gen) throws IOException { + gen.writeRaw(','); + newline(gen); + indent(gen); + } + + @Override + public void beforeObjectEntries(JsonGenerator gen) throws IOException { + indent(gen); + } + + @Override + public void beforeArrayValues(JsonGenerator gen) throws IOException { + indent(gen); + } + } /** * like query, but for the "get all instances of table" queries, returns a map where the object ID @@ -159,21 +173,33 @@ public List> query(QueryMeta info, Map argum public boolean update(Table s, Map search, Map object) throws Exception { - Map r = read(s, search); - if (r == null) - return false; - - for (Entry e : object.entrySet()) { - if (e.getValue() == null) - r.remove(e.getKey()); - else { - // when nested objects are overwritten and the old value contains a pointer, - // make sure to re-use the pointer - copyPointers(r.get(e.getKey()), e.getValue()); - r.put(e.getKey(), e.getValue()); + // read, merge and write back as one step, so concurrent updates are not lost + return locked(s, search.get("ID"), () -> { + Map r = read(s, search); + if (r == null) + return false; + + for (Entry e : object.entrySet()) { + if (e.getValue() == null) + r.remove(e.getKey()); + else { + // when nested objects are overwritten and the old value contains a pointer, + // make sure to re-use the pointer + copyPointers(r.get(e.getKey()), e.getValue()); + r.put(e.getKey(), e.getValue()); + } } - } - return update(s, r); + return update(s, r); + }); + } + + /** + * Runs body while holding the write lock of record (table, id), so a read-modify-write sequence + * on that record cannot interleave with other writes to it. The lock is reentrant. The default + * does not lock; JSONFileDatabase overrides this. + */ + public T locked(Table s, Object id, Callable body) throws Exception { + return body.call(); } /** diff --git a/dashjoin-core/src/main/java/org/dashjoin/service/JSONFileDatabase.java b/dashjoin-core/src/main/java/org/dashjoin/service/JSONFileDatabase.java index 5239887ab..deb5ba022 100644 --- a/dashjoin-core/src/main/java/org/dashjoin/service/JSONFileDatabase.java +++ b/dashjoin-core/src/main/java/org/dashjoin/service/JSONFileDatabase.java @@ -2,15 +2,30 @@ import java.io.File; import java.io.IOException; +import java.io.OutputStream; import java.io.UnsupportedEncodingException; import java.nio.charset.Charset; +import java.nio.file.AccessDeniedException; +import java.nio.file.AtomicMoveNotSupportedException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardCopyOption; +import java.nio.file.StandardOpenOption; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.HashSet; import java.util.LinkedHashMap; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Map.Entry; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.Callable; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.logging.Logger; import java.util.stream.Collectors; import org.apache.commons.io.FileUtils; @@ -27,7 +42,13 @@ import jakarta.inject.Inject; /** - * file implementation of the config DB + * file implementation of the config DB. + * + * Thread safety: files are written to a temp file and then renamed over the target, so readers and + * a crash mid-write never see a partial file. Each record (its .json plus externalized string + * files) has a read/write lock; writes and read-modify-write sequences (see locked()) hold the + * write lock. The locks are JVM-wide but do not protect against other processes writing the same + * model folder. */ @ApplicationScoped @DJRuntime(RuntimeDefinitions.ONPREMISE) @@ -45,6 +66,90 @@ protected File getFile(String path) { return home.getFile(path); } + /** + * prefix of temp files written before the atomic rename. URLEncoder (Escape.filename) always + * encodes ~, so no record ID or externalized file can start with this + */ + static final String TMP_PREFIX = "~tmp-"; + + /** + * one lock per record file, shared by all instances (there can be more than one per JVM). Keys + * are lower case since macOS and Windows file systems are case insensitive + */ + private static final ConcurrentHashMap locks = + new ConcurrentHashMap<>(); + + static ReentrantReadWriteLock lock(File jsonFile) { + return locks.computeIfAbsent(jsonFile.getAbsolutePath().toLowerCase(Locale.ROOT), + k -> new ReentrantReadWriteLock()); + } + + @Override + public T locked(Table s, Object id, Callable body) throws Exception { + ReentrantReadWriteLock rw = lock(file(s, Collections.singletonMap("ID", id))); + if (!rw.isWriteLockedByCurrentThread() && services != null) + // create() calls getConfig(), which is synchronized and reads config files while setting up + // a tenant. Make sure that is done before we hold the lock, otherwise the two can deadlock + services.getConfig(); + Lock l = rw.writeLock(); + l.lock(); + try { + return body.call(); + } finally { + l.unlock(); + } + } + + interface StreamWriter { + void write(OutputStream out) throws IOException; + } + + /** + * write to a temp file in the same folder, then rename it over the target. Readers see either the + * old or the new content, never a mix or a truncated file + */ + static void atomicWrite(File target, StreamWriter writer) throws IOException { + Path dir = target.toPath().toAbsolutePath().getParent(); + // unlike mkdirs(), does not fail when another thread creates the folder at the same time + Files.createDirectories(dir); + // not Files.createTempFile, which would create the file with mode 600 on POSIX systems + Path tmp = dir.resolve(TMP_PREFIX + UUID.randomUUID() + ".part"); + try { + try (OutputStream out = + Files.newOutputStream(tmp, StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE)) { + writer.write(out); + } + move(tmp, target.toPath()); + } finally { + Files.deleteIfExists(tmp); + } + } + + static void move(Path tmp, Path target) throws IOException { + for (int attempt = 0;; attempt++) { + try { + try { + Files.move(tmp, target, StandardCopyOption.ATOMIC_MOVE, + StandardCopyOption.REPLACE_EXISTING); + } catch (AtomicMoveNotSupportedException e) { + Files.move(tmp, target, StandardCopyOption.REPLACE_EXISTING); + } + return; + } catch (AccessDeniedException e) { + // on Windows, the rename fails while another process (editor, virus scanner) has the + // target open. Retry for a short while + if (attempt >= 20) + throw e; + try { + Thread.sleep(25); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + throw e; + } + } + } + } + /** * get json file handle from table and id */ @@ -75,7 +180,7 @@ List secondaryFiles(File[] files, Object id) throws UnsupportedEncodingExc List res = new ArrayList<>(); if (files != null) { for (File f : files) { - if (f.getName().endsWith(".json")) + if (f.getName().endsWith(".json") || f.getName().startsWith(TMP_PREFIX)) continue; if (f.getName().startsWith(Escape.filename("" + id))) res.add(f); @@ -104,13 +209,20 @@ public Map> queryMap(QueryMeta info, Map object = objectMapper.readValue(f, tr); String id = "" + object.get("ID"); - readExternalizedStrings(id, secondaryFiles(files, id), object); + readExternalizedStrings(f.getParentFile(), id, object); res.put(id, object); } catch (Throwable t) { - log.severe("Fatal error: JSONDatabase object cannot be read: "+f); + log.severe("Fatal error: JSONDatabase object cannot be read: " + f); + } finally { + l.unlock(); } } } @@ -124,14 +236,20 @@ public Map> queryMap(QueryMeta info, Map object) throws Exception { + locked(s, object.get("ID"), () -> { + createLocked(s, object); + return null; + }); + } + + @SuppressWarnings("unchecked") + void createLocked(Table s, Map object) throws Exception { File file = file(s, object); - if (!file.getParentFile().exists()) - if (!file.getParentFile().mkdirs()) - throw new IOException("Error creating model folder"); + // unlike mkdirs(), does not fail when another thread creates the folder at the same time + Files.createDirectories(file.getParentFile().toPath()); // field names to externalize List externalizeFields = new ArrayList<>(); @@ -150,81 +268,114 @@ public void create(Table s, Map object) throws Exception { } } + // reject bad pointers before any file is written + checkPointers(file.getParentFile(), Escape.filename("" + object.get("ID")), object); writeExternalizedStrings(s, object.get("ID"), file, object, externalizeFields); - objectMapper.writeValue(file, object); + atomicWrite(file, out -> objectMapper.writeValue(out, object)); } @Override public Map read(Table s, Map search) throws Exception { File file = file(s, search); - if (file.exists()) { - Map res = objectMapper.readValue(file, tr); - if (res.get("ID") == null) - throw new IllegalArgumentException("Object must contain ID field: " + file); - readExternalizedStrings("" + res.get("ID"), secondaryFiles(s, search.get("ID")), res); - return res; + Lock l = lock(file).readLock(); + l.lock(); + try { + if (file.exists()) { + Map res = objectMapper.readValue(file, tr); + if (res.get("ID") == null) + throw new IllegalArgumentException("Object must contain ID field: " + file); + readExternalizedStrings(file.getParentFile(), "" + res.get("ID"), res); + return res; + } + return null; + } finally { + l.unlock(); } - return null; } @Override public boolean delete(Table s, Map search) throws Exception { - for (File d : secondaryFiles(s, search.get("ID"))) - d.delete(); - return file(s, search).delete(); + return locked(s, search.get("ID"), () -> { + File file = file(s, search); + if (file.exists()) { + // only delete the files named by the record's pointers. A prefix match on the ID would + // also hit other records (deleting "a" would delete the files of "a-b") + Map res = null; + try { + res = objectMapper.readValue(file, tr); + } catch (IOException e) { + // unreadable record: delete the JSON file only + } + Set pointerFiles = new HashSet<>(); + forEachPointerFile(file.getParentFile(), Escape.filename("" + search.get("ID")), res, + (map, key, f) -> pointerFiles.add(f)); + for (File d : pointerFiles) + d.delete(); + } + return file.delete(); + }); } /** - * traverse the JSON tree and search for keys called X-pointer, add key X with the file contents + * traverse the JSON tree and search for keys called X-pointer, add key X with the contents of + * file ID.pointer in folder dir. The pointers name the files, so the folder is not listed (which + * is slow for tables with thousands of records). + * + * A file may be referenced more than once if we use copy / paste or the duplicate plugin UI + * feature. The second time, the pointer field is removed to make sure the content gets its own ID. */ - void readExternalizedStrings(String id, List secondaryFiles, Object res) - throws IOException { - _readExternalizedStrings(id, - secondaryFiles.stream().map(f -> new FileUsed(f)).collect(Collectors.toList()), res); + void readExternalizedStrings(File dir, String id, Object res) throws IOException { + Set used = new HashSet<>(); + forEachPointerFile(dir, Escape.filename(id), res, (map, key, f) -> { + String field = key.substring(0, key.length() - "-pointer".length()); + map.put(field, FileUtils.readFileToString(f, Charset.defaultCharset())); + if (!used.add(f.getName())) + map.remove(key); + }); + } + + interface PointerVisitor { + /** + * called for a key X-pointer in map whose file f exists. May change map + */ + void visit(Map map, String key, File f) throws IOException; } + /** + * traverse the JSON tree depth first and call visitor for every X-pointer key whose file ID.pointer + * exists in dir. Used by read (load the contents) and delete (remove the files) + */ @SuppressWarnings("unchecked") - void _readExternalizedStrings(String id, List secondaryFiles, Object res) + static void forEachPointerFile(File dir, String escapedId, Object res, PointerVisitor visitor) throws IOException { if (res instanceof Map) { - List removeKeys = new ArrayList<>(); - for (Entry e : new ArrayList<>(((Map) res).entrySet())) { - _readExternalizedStrings(id, secondaryFiles, e.getValue()); + Map map = (Map) res; + // iterate over a copy, the visitor may change the map + for (Entry e : new ArrayList<>(map.entrySet())) { + forEachPointerFile(dir, escapedId, e.getValue(), visitor); if (e.getKey().endsWith("-pointer")) { - String field = e.getKey().substring(0, e.getKey().length() - "-pointer".length()); - for (FileUsed f : secondaryFiles) - if (f.file.getName().equals(Escape.filename(id) + "." + e.getValue())) { - ((Map) res).put(field, - FileUtils.readFileToString(f.file, Charset.defaultCharset())); - if (f.used) - removeKeys.add(e.getKey()); - else - f.used = true; - } + File f = secondaryFile(dir, escapedId, e.getValue()); + if (f != null && f.isFile()) + visitor.visit(map, e.getKey(), f); } - for (String key : removeKeys) - ((Map) res).remove(key); } } - if (res instanceof List) { + if (res instanceof List) for (Object i : (List) res) - _readExternalizedStrings(id, secondaryFiles, i); - } + forEachPointerFile(dir, escapedId, i, visitor); } /** - * make sure a file is only referenced once. Might be referenced multiple times if we use copy / - * paste or the duplicate plugin UI feature. The second time, the pointer field is removed to make - * sure the content gets its own ID. + * file ID.pointer in dir, or null if the pointer would leave dir (e.g. "../../etc/passwd"). + * Escape.filename encodes path separators in the ID, so only the pointer needs checking */ - static class FileUsed { - FileUsed(File file) { - this.file = file; - } - - boolean used; - File file; + static File secondaryFile(File dir, String escapedId, Object pointer) { + String name = escapedId + "." + pointer; + if (name.contains("/") || name.contains("\\") || name.contains(File.separator) + || name.indexOf('\0') >= 0) + return null; + return new File(dir, name); } /** @@ -263,28 +414,58 @@ void writeExternalizedStrings(Table s, Object id, File file, Object res, } } + /** + * file for the pointer next to the record's JSON file. Throws if the pointer is not a string or + * would leave the table folder (e.g. "../../x") + */ + static File pointerFile(Object id, File file, Object pointer) { + File res = pointer instanceof String + ? secondaryFile(file.getParentFile(), Escape.filename("" + id), pointer) + : null; + if (res == null) + throw new IllegalArgumentException("Invalid pointer: " + pointer); + return res; + } + + /** + * check all X-pointer values in the JSON tree, see pointerFile + */ + @SuppressWarnings("unchecked") + static void checkPointers(File dir, String escapedId, Object res) { + if (res instanceof Map) + for (Entry e : ((Map) res).entrySet()) { + checkPointers(dir, escapedId, e.getValue()); + if (e.getKey().endsWith("-pointer") && (!(e.getValue() instanceof String) + || secondaryFile(dir, escapedId, e.getValue()) == null)) + throw new IllegalArgumentException("Invalid pointer: " + e.getValue()); + } + if (res instanceof List) + for (Object i : (List) res) + checkPointers(dir, escapedId, i); + } + /** * remove field from map and write its value to the pointer file */ void writeString(Object id, String field, File file, Map map, String pointer) throws IOException { - String newName = Escape.filename("" + id) + "." + pointer; - File newFile = new File(file.getParentFile(), newName); + File newFile = pointerFile(id, file, pointer); if (!(map.get(field) instanceof String) && (map.get(field) != null)) throw new RuntimeException("Error externalizing data. The value of '" + field + "' must be a string. Please change the config."); map.put(field + "-pointer", pointer); - FileUtils.writeStringToFile(newFile, (String) map.remove(field), Charset.defaultCharset()); + String value = (String) map.remove(field); + atomicWrite(newFile, + out -> out.write(value == null ? new byte[0] : value.getBytes(Charset.defaultCharset()))); } /** * remove field from map and write its value to the pointer file */ void deleteString(Object id, String field, File file, Map map, String pointer) { - String newName = Escape.filename("" + id) + "." + pointer; - File newFile = new File(file.getParentFile(), newName); + File newFile = pointerFile(id, file, pointer); newFile.delete(); map.remove(field + "-pointer"); } diff --git a/dashjoin-core/src/main/java/org/dashjoin/service/PasswordDatabase.java b/dashjoin-core/src/main/java/org/dashjoin/service/PasswordDatabase.java index bb3adafc9..54deb5de7 100644 --- a/dashjoin-core/src/main/java/org/dashjoin/service/PasswordDatabase.java +++ b/dashjoin-core/src/main/java/org/dashjoin/service/PasswordDatabase.java @@ -1,6 +1,7 @@ package org.dashjoin.service; import java.util.Map; +import java.util.concurrent.Callable; import org.dashjoin.model.QueryMeta; import org.dashjoin.model.Table; import org.dashjoin.util.MapUtil; @@ -66,6 +67,11 @@ public boolean delete(Table s, Map search) throws Exception { return db.delete(s, search); } + @Override + public T locked(Table s, Object id, Callable body) throws Exception { + return db.locked(s, id, body); + } + @Override public Map> queryMap(QueryMeta info, Map arguments) throws Exception { diff --git a/dashjoin-core/src/main/java/org/dashjoin/service/UnionDatabase.java b/dashjoin-core/src/main/java/org/dashjoin/service/UnionDatabase.java index 148dd5651..8d1e2e297 100644 --- a/dashjoin-core/src/main/java/org/dashjoin/service/UnionDatabase.java +++ b/dashjoin-core/src/main/java/org/dashjoin/service/UnionDatabase.java @@ -156,11 +156,15 @@ else if (array && (e.getValue() instanceof List)) { public void create(Table m, Map object) throws Exception { if (object.get("ID") == null) throw new Exception("Object must contain an ID"); - if (read(m, object) != null) - throw new Exception("Object already exists. Please choose a different ID."); + // check and create as one step, so two concurrent creates of the same ID cannot both succeed + user().locked(m, object.get("ID"), () -> { + if (read(m, object) != null) + throw new Exception("Object already exists. Please choose a different ID."); - // delegate to user db - user().create(m, object); + // delegate to user db + user().create(m, object); + return null; + }); } @Override @@ -192,6 +196,12 @@ public Map readDBs(Table s, Map search) throws E @Override public boolean update(Table schema, Map search, Map object) throws Exception { + // the steps below read and write the user DB several times; hold the record lock throughout + return user().locked(schema, search.get("ID"), () -> updateLocked(schema, search, object)); + } + + boolean updateLocked(Table schema, Map search, Map object) + throws Exception { Map readOnly = readDBs(schema, search); Map user = user().read(schema, search); @@ -244,15 +254,17 @@ public boolean update(Table schema, Map search, Map search) throws Exception { - Map old = read(s, search); - if (old == null) - return false; + return user().locked(s, search.get("ID"), () -> { + Map old = read(s, search); + if (old == null) + return false; - // the old value might be in the read only part - // delete it from the user DB but ignore the result - user().delete(s, search); + // the old value might be in the read only part + // delete it from the user DB but ignore the result + user().delete(s, search); - return true; + return true; + }); } @Override diff --git a/dashjoin-core/src/test/java/org/dashjoin/service/JSONFileDatabaseConcurrencyTest.java b/dashjoin-core/src/test/java/org/dashjoin/service/JSONFileDatabaseConcurrencyTest.java new file mode 100644 index 000000000..d57db0df1 --- /dev/null +++ b/dashjoin-core/src/test/java/org/dashjoin/service/JSONFileDatabaseConcurrencyTest.java @@ -0,0 +1,383 @@ +package org.dashjoin.service; + +import java.io.File; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import org.dashjoin.model.Table; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import com.fasterxml.jackson.databind.DeserializationFeature; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.SerializationConfig; + +/** + * Stress tests for concurrent access to JSONFileDatabase. Each test asserts the correct behavior, + * so a failure means the race was observed. + */ +public class JSONFileDatabaseConcurrencyTest { + + static final int THREADS = 8; + static final long DURATION_MS = 3000; + + /** + * strict parser, so trailing garbage is detected. Note: JSONDatabase.objectMapper (and any default + * ObjectMapper) stops after the first JSON value, so Dashjoin itself reads a torn file + * "successfully" while git, editors and other tools see invalid JSON. + */ + static final ObjectMapper strict = + new ObjectMapper().enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS); + + @TempDir + Path tmp; + + JSONFileDatabase db; + ExecutorService pool; + final Table table = Table.ofName("stress"); + + @BeforeEach + void setup() { + db = new JSONFileDatabase() { + @Override + protected File getFile(String path) { + return tmp.resolve(path).toFile(); + } + }; + pool = Executors.newFixedThreadPool(THREADS * 2); + } + + @AfterEach + void teardown() throws Exception { + pool.shutdownNow(); + pool.awaitTermination(10, TimeUnit.SECONDS); + } + + static Map record(String id, int writer, int size) { + Map m = new LinkedHashMap<>(); + m.put("ID", id); + m.put("writer", writer); + m.put("payload", "x".repeat(size)); + return m; + } + + /** + * Writers overwrite the same record with short and long payloads while readers parse the file. + * Expect: every read sees a complete object, and the final file is valid JSON. + */ + @Test + public void concurrentWritesSameRecord() throws Exception { + File file = tmp.resolve("model/stress/rec.json").toFile(); + db.create(table, record("rec", -1, 10)); + + AtomicBoolean stop = new AtomicBoolean(); + AtomicInteger writes = new AtomicInteger(); + AtomicInteger reads = new AtomicInteger(); + ConcurrentLinkedQueue readErrors = new ConcurrentLinkedQueue<>(); + ConcurrentLinkedQueue writeErrors = new ConcurrentLinkedQueue<>(); + CountDownLatch done = new CountDownLatch(THREADS * 2); + + for (int t = 0; t < THREADS; t++) { + final int writer = t; + pool.submit(() -> { + try { + int i = 0; + while (!stop.get()) { + // alternate short and long content so a short write over a long one leaves a tail + int size = (i++ % 2 == 0) ? 10 : 20000 + writer * 1000; + db.create(table, record("rec", writer, size)); + writes.incrementAndGet(); + } + } catch (Throwable e) { + writeErrors.add(e.toString()); + } finally { + done.countDown(); + } + }); + pool.submit(() -> { + try { + while (!stop.get()) { + try { + Map m = db.read(table, Collections.singletonMap("ID", "rec")); + if (m == null) + readErrors.add("read returned null (file missing)"); + reads.incrementAndGet(); + } catch (Throwable e) { + readErrors.add(e.getClass().getSimpleName()); + } + } + } finally { + done.countDown(); + } + }); + } + + Thread.sleep(DURATION_MS); + stop.set(true); + done.await(30, TimeUnit.SECONDS); + + String finalError = null; + try { + strict.readTree(Files.readString(file.toPath())); + } catch (Exception e) { + finalError = e.getMessage(); + } + + System.out.println("concurrentWritesSameRecord: writes=" + writes + " successfulReads=" + reads + + " readErrors=" + readErrors.size() + " writeErrors=" + writeErrors.size() + + " finalFileValid=" + (finalError == null)); + if (!readErrors.isEmpty()) + System.out.println(" sample read errors: " + sample(readErrors)); + if (!writeErrors.isEmpty()) + System.out.println(" sample write errors: " + sample(writeErrors)); + + Assertions.assertTrue(writeErrors.isEmpty(), "write errors: " + sample(writeErrors)); + Assertions.assertTrue(readErrors.isEmpty(), + readErrors.size() + " reads failed (" + reads + " succeeded): " + sample(readErrors)); + Assertions.assertNull(finalError, "final file is invalid JSON: " + finalError); + } + + /** + * Repeat many short bursts and check the file after each burst, since the final state of one long + * run only gives a single sample of whether a torn write was left behind. + */ + @Test + public void fileValidAfterEachBurst() throws Exception { + File file = tmp.resolve("model/stress/burst.json").toFile(); + // create the folder up front; concurrent first writes to a new table hit a separate mkdirs race + // (see concurrentCreateNewTable) + db.create(table, record("burst", -1, 5)); + int bursts = 500; + int invalid = 0; + String firstError = null; + java.util.Random rnd = new java.util.Random(42); + for (int b = 0; b < bursts; b++) { + CountDownLatch start = new CountDownLatch(1); + List> futures = new ArrayList<>(); + for (int t = 0; t < THREADS; t++) { + // build the record before the start signal so all writers open the file at the same time; + // random sizes from tiny to ~500 KB so a shorter write can land over a longer one + Map rec = record("burst", t, rnd.nextBoolean() ? 5 : rnd.nextInt(500000)); + futures.add(pool.submit(() -> { + start.await(); + db.create(table, rec); + return null; + })); + } + start.countDown(); + for (var f : futures) + f.get(); + try { + strict.readTree(Files.readString(file.toPath())); + } catch (Exception e) { + invalid++; + if (firstError == null) + firstError = e.getMessage(); + } + } + System.out.println("fileValidAfterEachBurst: invalid files=" + invalid + " of " + bursts); + Assertions.assertEquals(0, invalid, "first error: " + firstError); + } + + /** + * Each thread updates its own field of the same record via update(search, delta). Expect: all + * fields are present at the end. + */ + @Test + public void lostUpdates() throws Exception { + Map init = new LinkedHashMap<>(); + init.put("ID", "lu"); + db.create(table, init); + + int perThread = 50; + CountDownLatch start = new CountDownLatch(1); + List> futures = new ArrayList<>(); + for (int t = 0; t < THREADS; t++) { + final int writer = t; + futures.add(pool.submit(() -> { + start.await(); + for (int i = 0; i < perThread; i++) { + Map delta = new LinkedHashMap<>(); + delta.put("f" + writer + "_" + i, i); + // a torn file makes update() throw; retry so we measure lost updates, not parse errors + for (int attempt = 0;; attempt++) + try { + db.update(table, Collections.singletonMap("ID", "lu"), delta); + break; + } catch (Exception e) { + if (attempt > 100) + throw e; + } + } + return null; + })); + } + start.countDown(); + for (var f : futures) + f.get(); + + Map res = db.read(table, Collections.singletonMap("ID", "lu")); + int expected = THREADS * perThread; + int actual = res.size() - 1; // minus ID + System.out.println("lostUpdates: fields expected=" + expected + " present=" + actual); + Assertions.assertEquals(expected, actual, (expected - actual) + " updates were lost"); + } + + /** + * Same as lostUpdates, but through UnionDatabase, which is the path the config DB uses. Its + * update() creates the record if missing, updates one key at a time and deletes records that only + * contain the ID, so it needs the record lock across all of these steps. + */ + @Test + public void lostUpdatesThroughUnion() throws Exception { + UnionDatabase union = new UnionDatabase(); + union._user = db; + union._dbs = new ArrayList<>(); + + int perThread = 50; + CountDownLatch start = new CountDownLatch(1); + ConcurrentLinkedQueue errors = new ConcurrentLinkedQueue<>(); + List> futures = new ArrayList<>(); + for (int t = 0; t < THREADS; t++) { + final int writer = t; + futures.add(pool.submit(() -> { + start.await(); + for (int i = 0; i < perThread; i++) { + Map delta = new LinkedHashMap<>(); + delta.put("f" + writer + "_" + i, i); + try { + union.update(table, new LinkedHashMap<>(Map.of("ID", "union")), delta); + } catch (Exception e) { + errors.add(e.toString()); + } + } + return null; + })); + } + start.countDown(); + for (var f : futures) + f.get(); + + Map res = db.read(table, Collections.singletonMap("ID", "union")); + int expected = THREADS * perThread; + int actual = res == null ? 0 : res.size() - 1; + System.out.println("lostUpdatesThroughUnion: fields expected=" + expected + " present=" + + actual + " errors=" + errors.size()); + Assertions.assertTrue(errors.isEmpty(), "errors: " + sample(errors)); + Assertions.assertEquals(expected, actual, (expected - actual) + " updates were lost"); + } + + /** + * The shared pretty printer keeps its indent level in a field. Serialize nested objects from + * many threads and compare to the single-threaded output. + */ + /** + * Jackson only gives each generator its own pretty printer if the printer implements + * Instantiatable. Otherwise all threads share one instance and its indent field. + */ + @Test + public void prettyPrinterNotShared() { + SerializationConfig cfg = JSONDatabase.objectMapper.getSerializationConfig(); + Assertions.assertNotSame(cfg.constructDefaultPrettyPrinter(), + cfg.constructDefaultPrettyPrinter(), + "all threads share one PrettyPrinter instance with a mutable indent field"); + } + + @Test + public void prettyPrinterThreadSafety() throws Exception { + Map nested = new LinkedHashMap<>(); + Map cur = nested; + for (int i = 0; i < 20; i++) { + Map child = new LinkedHashMap<>(); + List list = new ArrayList<>(); + for (int j = 0; j < 20; j++) + list.add(Map.of("a", List.of(j, Map.of("b", j)))); + cur.put("k" + i, list); + cur.put("c" + i, child); + cur = child; + } + String expected = JSONDatabase.toJsonString(nested); + + AtomicInteger mismatches = new AtomicInteger(); + AtomicInteger total = new AtomicInteger(); + CountDownLatch start = new CountDownLatch(1); + List> futures = new ArrayList<>(); + for (int t = 0; t < THREADS; t++) + futures.add(pool.submit(() -> { + start.await(); + for (int i = 0; i < 5000; i++) { + total.incrementAndGet(); + if (!expected.equals(JSONDatabase.toJsonString(nested))) + mismatches.incrementAndGet(); + } + return null; + })); + start.countDown(); + for (var f : futures) + f.get(); + + // after the run, a single-threaded call should still match (indent counter not drifted) + String after = JSONDatabase.toJsonString(nested); + System.out.println("prettyPrinterThreadSafety: mismatches=" + mismatches + " of " + total + + " driftAfterRun=" + !expected.equals(after)); + Assertions.assertEquals(0, mismatches.get(), "wrongly indented outputs"); + Assertions.assertEquals(expected, after, "indent counter drifted permanently"); + } + + /** + * Several threads create the first records of a table that has no folder yet. Expect: no errors. + * create() checks exists() and then mkdirs(); a thread that loses the race gets false from + * mkdirs() and throws "Error creating model folder". + */ + @Test + public void concurrentCreateNewTable() throws Exception { + int failures = 0; + String firstError = null; + int rounds = 200; + for (int r = 0; r < rounds; r++) { + Table t = Table.ofName("newtable" + r); + CountDownLatch start = new CountDownLatch(1); + List> futures = new ArrayList<>(); + for (int i = 0; i < THREADS; i++) { + final int id = i; + futures.add(pool.submit(() -> { + start.await(); + db.create(t, record("r" + id, id, 5)); + return null; + })); + } + start.countDown(); + for (var f : futures) + try { + f.get(); + } catch (java.util.concurrent.ExecutionException e) { + failures++; + if (firstError == null) + firstError = e.getCause().toString(); + } + } + System.out.println("concurrentCreateNewTable: failed creates=" + failures + " of " + + (rounds * THREADS)); + Assertions.assertEquals(0, failures, "first error: " + firstError); + } + + static String sample(ConcurrentLinkedQueue q) { + Map counts = new LinkedHashMap<>(); + for (String s : q) + counts.merge(s.length() > 150 ? s.substring(0, 150) : s, 1, Integer::sum); + return counts.toString(); + } +} diff --git a/dashjoin-core/src/test/java/org/dashjoin/service/JSONFileDatabasePointerTest.java b/dashjoin-core/src/test/java/org/dashjoin/service/JSONFileDatabasePointerTest.java new file mode 100644 index 000000000..dc42a7a62 --- /dev/null +++ b/dashjoin-core/src/test/java/org/dashjoin/service/JSONFileDatabasePointerTest.java @@ -0,0 +1,161 @@ +package org.dashjoin.service; + +import java.io.File; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import java.util.Map; +import org.dashjoin.model.Table; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +/** + * externalized strings are found via their -pointer fields, without listing the table folder + */ +public class JSONFileDatabasePointerTest { + + @TempDir + Path tmp; + + JSONFileDatabase db; + Path dir; + final Table table = Table.ofName("t"); + + @BeforeEach + void setup() throws Exception { + db = new JSONFileDatabase() { + @Override + protected File getFile(String path) { + return tmp.resolve(path).toFile(); + } + }; + dir = Files.createDirectories(tmp.resolve("model/t")); + } + + Map read(String id) throws Exception { + return db.read(table, Map.of("ID", id)); + } + + @Test + public void readsPointerFile() throws Exception { + Files.writeString(dir.resolve("a%40b.json"), "{\"ID\": \"a@b\", \"query-pointer\": \"query.sql\"}"); + Files.writeString(dir.resolve("a%40b.query.sql"), "select 1"); + Map res = read("a@b"); + Assertions.assertEquals("select 1", res.get("query")); + Assertions.assertEquals("query.sql", res.get("query-pointer")); + + // same via queryMap + org.dashjoin.model.QueryMeta q = new org.dashjoin.model.QueryMeta(); + q.query = "t"; + Assertions.assertEquals("select 1", db.queryMap(q, null).get("a@b").get("query")); + } + + @Test + public void nestedAndDuplicatePointers() throws Exception { + // the second reference to the same file gets the content but loses its pointer + Files.writeString(dir.resolve("x.json"), "{\"ID\": \"x\", \"children\": [" + + "{\"html-pointer\": \"html.html\"}, {\"html-pointer\": \"html.html\"}]}"); + Files.writeString(dir.resolve("x.html.html"), "hi"); + @SuppressWarnings("unchecked") + List> children = (List>) read("x").get("children"); + Assertions.assertEquals("hi", children.get(0).get("html")); + Assertions.assertEquals("html.html", children.get(0).get("html-pointer")); + Assertions.assertEquals("hi", children.get(1).get("html")); + Assertions.assertFalse(children.get(1).containsKey("html-pointer")); + } + + @Test + public void missingPointerFileIsIgnored() throws Exception { + Files.writeString(dir.resolve("m.json"), "{\"ID\": \"m\", \"query-pointer\": \"query.sql\"}"); + Map res = read("m"); + Assertions.assertFalse(res.containsKey("query")); + Assertions.assertEquals("query.sql", res.get("query-pointer")); + } + + @Test + public void pointerCannotLeaveFolder() throws Exception { + Files.writeString(tmp.resolve("secret.txt"), "secret"); + Files.writeString(dir.resolve("p.json"), + "{\"ID\": \"p\", \"query-pointer\": \"/../../../secret.txt\"}"); + Assertions.assertFalse(read("p").containsKey("query")); + } + + @Test + public void deleteRemovesOwnPointerFiles() throws Exception { + Files.writeString(dir.resolve("d.json"), "{\"ID\": \"d\", \"query-pointer\": \"query.sql\"," + + " \"children\": [{\"html-pointer\": \"html.html\"}]}"); + Files.writeString(dir.resolve("d.query.sql"), "select 1"); + Files.writeString(dir.resolve("d.html.html"), "hi"); + Assertions.assertTrue(db.delete(table, Map.of("ID", "d"))); + try (var files = Files.list(dir)) { + Assertions.assertEquals(0, files.count()); + } + } + + @Test + public void deleteKeepsFilesOfRecordsWithSamePrefix() throws Exception { + // deleting "fn" must not touch the files of "fn-all" or "fn.x" + for (String id : new String[] {"fn", "fn-all", "fn.x"}) { + Files.writeString(dir.resolve(id + ".json"), + "{\"ID\": \"" + id + "\", \"expression-pointer\": \"expression.jsonata\"}"); + Files.writeString(dir.resolve(id + ".expression.jsonata"), id); + } + Assertions.assertTrue(db.delete(table, Map.of("ID", "fn"))); + Assertions.assertFalse(Files.exists(dir.resolve("fn.json"))); + Assertions.assertFalse(Files.exists(dir.resolve("fn.expression.jsonata"))); + Assertions.assertEquals("fn-all", read("fn-all").get("expression")); + Assertions.assertEquals("fn.x", read("fn.x").get("expression")); + } + + @Test + public void deleteCannotLeaveFolder() throws Exception { + Files.writeString(tmp.resolve("secret.txt"), "secret"); + Files.writeString(dir.resolve("p.json"), + "{\"ID\": \"p\", \"query-pointer\": \"/../../../secret.txt\"}"); + Assertions.assertTrue(db.delete(table, Map.of("ID", "p"))); + Assertions.assertTrue(Files.exists(tmp.resolve("secret.txt"))); + } + + @Test + public void writeRejectsPointerLeavingFolder() throws Exception { + Map rec = new java.util.HashMap<>(); + rec.put("ID", "w"); + rec.put("query", "pwned"); + rec.put("query-pointer", "/../../../evil.txt"); + Assertions.assertThrows(IllegalArgumentException.class, () -> db.create(table, rec)); + Assertions.assertFalse(Files.exists(tmp.resolve("evil.txt"))); + Assertions.assertFalse(Files.exists(dir.resolve("w.json"))); + } + + @Test + public void writeWithNullValueCannotDeleteOutsideFolder() throws Exception { + Files.writeString(tmp.resolve("secret.txt"), "secret"); + Map rec = new java.util.HashMap<>(); + rec.put("ID", "w"); + rec.put("query-pointer", "/../../../secret.txt"); + Assertions.assertThrows(IllegalArgumentException.class, () -> db.create(table, rec)); + Assertions.assertTrue(Files.exists(tmp.resolve("secret.txt"))); + } + + @Test + public void writeRejectsNonStringPointer() throws Exception { + Map rec = new java.util.HashMap<>(); + rec.put("ID", "w"); + rec.put("query", "select 1"); + rec.put("query-pointer", 42); + Assertions.assertThrows(IllegalArgumentException.class, () -> db.create(table, rec)); + } + + @Test + public void writeValidPointer() throws Exception { + Map rec = new java.util.HashMap<>(); + rec.put("ID", "w"); + rec.put("query", "select 1"); + rec.put("query-pointer", "query.sql"); + db.create(table, rec); + Assertions.assertEquals("select 1", Files.readString(dir.resolve("w.query.sql"))); + Assertions.assertEquals("select 1", read("w").get("query")); + } +}