CON-649: Prevent unreadable database state when stopping Concourse Server in the middle of indexing

CHANGELOG.md

@@ -1,18 +1,28 @@
 ## Changelog
 
 #### Version 0.10.0 (TBD)
+
+##### New Features
 * Added an iterative connection builder that is accessible using the `Concourse.at()` static factory method.
-* Refactored the `concourse-import` framework to take advantage of version `1.1.0+` of the `data-transform-api` which has a more flexible notion of data transformations. As a result of this change, the `Importables` utility class has been removed. Custom importers that extend `DelimitedLineImporter` can leverage the protected `parseObject` and `importLines` methods to hook into the extraction and import logic in a manner similar to what was possible using the `Importables` functions.
 * Added the `com.cinchapi.concourse.valididate.Keys` utility class which contains the `#isWritable` method that determines if a proposed key can be written to Concourse.
-* Fixed a bug that caused data imported from STDIN to not have a `__datasource` tag, even if the `--annotate-data-source` flag was included with the CLI invocation.  
 * Added `Parsers#create` static factory methods that accept a `Criteria` object as a parameter. These new methods compliment existing ones which take a CCL `String` and `TCriteria` object respectively.
-* Upgrade the `ccl` dependency to the latest version, which adds support for local criteria evaluation using the `Parser#evaluate` method. The parsers returned from the `Parsers#create` factories all support local evaluation using the function defined in the newly created `Operators#evaluate` utility.
+* Upgraded the `ccl` dependency to the latest version, which adds support for local criteria evaluation using the `Parser#evaluate` method. The parsers returned from the `Parsers#create` factories all support local evaluation using the function defined in the newly created `Operators#evaluate` utility.
 * Added the `com.cinchapi.concourse.etl` package that contains data processing utilities:
 	*  A `Strainer` can be used to process a `Map<String, Object>` using Concourse's data model rules. In particular, the `Strainer` encapsulates logic to break down top-level sequence values and process their elements individually.
 	* The `Transform` class contains functions for common data transformations. 
+
+##### Improvements
+* Refactored the `concourse-import` framework to take advantage of version `1.1.0+` of the `data-transform-api` which has a more flexible notion of data transformations. As a result of this change, the `Importables` utility class has been removed. Custom importers that extend `DelimitedLineImporter` can leverage the protected `parseObject` and `importLines` methods to hook into the extraction and import logic in a manner similar to what was possible using the `Importables` functions.
+* Refactored the `Criteria` class into an interface that is implemented by any language symbols that can be immediately transformed to a well-built criteria (e.g. `ValueState` and `TimestampState`). The primary benefit of this change is that methods that took a generic Object parameter and checked whether that object could be built into a `Criteria` have now been removed from the `Concourse` driver since that logic is automatically captured within the new class hiearchy. Another positive side effect of this change is that it is no longer necessary to explicitly build a nested `Criteria` when using the `group` functionality of the `Criteria` builder.
+
+##### Bug Fixes
+* Fixed a bug that caused data imported from STDIN to not have a `__datasource` tag, even if the `--annotate-data-source` flag was included with the CLI invocation. 
+* Fixed a bug that allowed Concourse Server to start an environment's storage engine in a partially or wholly unreadable state if the Engine partially completed a block sync while Concourse Server was going through its shutdown routine. In this scenario, the partially written block is malformed and should not be processed by the Engine since the data contained in the malformed block is still contained in the Buffer. While the malformed block files can be safely deleted, the implemented fix causes the Engine to simply ignore them if they are encountered upon initialization. 
+* Added checks to ensure that a storage Engine cannot transport writes from the Buffer to the Database while Concourse Server is shutting down.
+
+##### Deprecations and Removed Features
 * Removed the `Strings` utility class in favor of `AnyStrings` from `accent4j`.
 * Removed the `StringSplitter` framework in favor of the same from `accent4j`.
-* Refactored the `Criteria` class into an interface that is implemented by any language symbols that can be immediately transformed to a well-built criteria (e.g. `ValueState` and `TimestampState`). The primary benefit of this change is that methods that took a generic Object parameter and checked whether that object could be built into a `Criteria` have now been removed from the `Concourse` driver since that logic is automatically captured within the new class hiearchy. Another positive side effect of this change is that it is no longer necessary to explicitly build a nested `Criteria` when using the `group` functionality of the `Criteria` builder.
 * Deprecated `Criteria#getCclString` in favor of `Criteria#ccl`.
 
 #### Version 0.9.6 (February 16, 2019)

concourse-ete-tests/build.gradle

@@ -0,0 +1,24 @@
+/*
+ * Copyright (c) 2013-2019 Cinchapi Inc.
+ *
+ * Licensed 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.
+ */
+
+dependencies {
+    testCompile project(':concourse-driver-java')
+    testCompile project(':concourse-ete-test-core')
+}
+
+test {
+	exclude '**'
+}

concourse-ete-tests/src/test/java/com/cinchapi/concourse/bugrepro/CON649.java

@@ -0,0 +1,114 @@
+/*
+ * Copyright (c) 2013-2019 Cinchapi Inc.
+ *
+ * Licensed 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 com.cinchapi.concourse.bugrepro;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.thrift.transport.TTransportException;
+import org.junit.Assert;
+import org.junit.Test;
+
+import com.cinchapi.concourse.Concourse;
+import com.cinchapi.concourse.test.ClientServerTest;
+import com.cinchapi.concourse.util.Random;
+import com.google.common.collect.HashBasedTable;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import com.google.common.collect.Table;
+import com.google.common.collect.Table.Cell;
+
+/**
+ * Unit test to reproduce the issues described in CON-649.
+ *
+ * @author Jeff Nelson
+ */
+public class CON649 extends ClientServerTest {
+
+    @Test
+    public void repro()
+            throws IOException, TTransportException, InterruptedException {
+        List<Thread> clients = Lists.newArrayList();
+        Table<Long, String, String> expected = HashBasedTable.create();
+        for (int i = 0; i < Runtime.getRuntime().availableProcessors(); ++i) {
+            clients.add(new Thread(() -> {
+                Concourse $client = server.connect();
+                while (!Thread.currentThread().isInterrupted()) {
+                    try {
+                        long id = Random.getLong();
+                        String key = Random.getSimpleString();
+                        String value = Random.getSimpleString();
+                        $client.add(key, value, id);
+                        expected.put(id, key, value);
+                        Random.tinySleep(); // allow some transports to go
+                                            // through...
+                    }
+                    catch (Exception e) {
+                        Thread.currentThread().interrupt();
+                    }
+                }
+            }));
+        }
+        clients.forEach(Thread::start);
+        Thread.sleep(10000);
+        Random.microSleep();
+        server.stop();
+        Path db = server.getDatabaseDirectory().resolve("default");
+        List<Path> directories = ImmutableList.of(db.resolve("cpb"),
+                db.resolve("csb"), db.resolve("ctb"));
+        Map<String, AtomicInteger> counts = Maps.newLinkedHashMap();
+        directories.forEach(directory -> {
+            try {
+                Files.list(directory).forEach(file -> {
+                    System.out.println(file);
+                    String name = file.getFileName().toString().split("\\.")[0];
+                    counts.computeIfAbsent(name, key -> new AtomicInteger(0))
+                            .incrementAndGet();
+                });
+            }
+            catch (IOException e) {
+                e.printStackTrace();
+            }
+        });
+        Set<Integer> distinct = Sets.newHashSet();
+        counts.forEach((path, count) -> {
+            System.out.println(path + " = " + count);
+            distinct.add(count.get());
+        });
+        System.out.println(counts.size());
+        server.start();
+        client = server.connect();
+        for (Cell<Long, String, String> cell : expected.cellSet()) {
+            long record = cell.getRowKey();
+            String key = cell.getColumnKey();
+            Object value = cell.getValue();
+            Assert.assertEquals(value, client.get(key, record));
+        }
+    }
+
+    @Override
+    protected String getServerVersion() {
+        return ClientServerTest.LATEST_SNAPSHOT_VERSION;
+    }
+
+}

concourse-server/src/main/java/com/cinchapi/concourse/server/io/FileSystem.java

@@ -34,6 +34,7 @@
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.Set;
+import java.util.stream.Stream;
 
 import com.cinchapi.common.base.CheckedExceptions;
 import com.cinchapi.concourse.util.FileOps;
@@ -146,8 +147,8 @@ public static Iterator<String> fileOnlyIterator(final String directory) {
         return new ReadOnlyIterator<String>() {
 
             private final File[] files = new File(directory).listFiles();
-            private int position = 0;
             private File next = null;
+            private int position = 0;
             {
                 findNext();
             }
@@ -278,8 +279,8 @@ public static boolean hasDir(String dir) {
      * @param file
      * @return {@code true} if {@code file} exists
      */
-    public static boolean hasFile(String file) {
-        return hasFile(Paths.get(file));
+    public static boolean hasFile(Path file) {
+        return Files.exists(file) && !Files.isDirectory(file);
     }
 
     /**
@@ -289,8 +290,8 @@ public static boolean hasFile(String file) {
      * @param file
      * @return {@code true} if {@code file} exists
      */
-    public static boolean hasFile(Path file) {
-        return Files.exists(file) && !Files.isDirectory(file);
+    public static boolean hasFile(String file) {
+        return hasFile(Paths.get(file));
     }
 
     /**
@@ -320,6 +321,22 @@ public static void lock(String path) {
         }
     }
 
+    /**
+     * Return a {@link Stream} that contains a non-recursive list of all the
+     * files in the {@code directory}.
+     * 
+     * @param directory
+     * @return the files in the directory
+     */
+    public static Stream<Path> ls(Path directory) {
+        try {
+            return Files.list(directory);
+        }
+        catch (IOException e) {
+            throw CheckedExceptions.wrapAsRuntimeException(e);
+        }
+    }
+
     /**
      * Create a valid path that contains separators in the appropriate places
      * by joining all the {@link parts} together with the {@link File#separator}

concourse-server/src/main/java/com/cinchapi/concourse/server/storage/cache/BloomFilter.java

@@ -22,6 +22,7 @@
 import java.io.ObjectInputStream;
 import java.io.ObjectStreamClass;
 import java.nio.channels.FileChannel;
+import java.nio.file.Path;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.StampedLock;
 
@@ -63,6 +64,24 @@ public static BloomFilter create(int expectedInsertions) {
         return new BloomFilter(null, expectedInsertions);
     }
 
+    /**
+     * Create a new BloomFilter with enough capacity for
+     * {@code expectedInsertions}.
+     * <p>
+     * Note that overflowing a BloomFilter with significantly more elements than
+     * specified, will result in its saturation, and a sharp deterioration of
+     * its false positive probability (source:
+     * {@link BloomFilter#create(com.google.common.hash.Funnel, int)})
+     * <p>
+     * 
+     * @param file
+     * @param expectedInsertions
+     * @return the BloomFilter
+     */
+    public static BloomFilter create(Path file, int expectedInsertions) {
+        return create(file.toString(), expectedInsertions);
+    }
+
     /**
      * Create a new BloomFilter with enough capacity for
      * {@code expectedInsertions}.
@@ -81,6 +100,16 @@ public static BloomFilter create(String file, int expectedInsertions) {
         return new BloomFilter(file, expectedInsertions);
     }
 
+    /**
+     * Return the BloomFilter that is stored on disk in {@code file}.
+     * 
+     * @param file
+     * @return the BloomFilter
+     */
+    public static BloomFilter open(Path file) {
+        return open(file.toString());
+    }
+
     /**
      * Return the BloomFilter that is stored on disk in {@code file}.
      *