Merge pull request #146 from SpineEventEngine/catch-up

Implement catch-up storage and use transactions for `Delivery` routines

Author: armiol

build.gradle

@@ -99,6 +99,12 @@ subprojects {
 
     defaultRepositories(project)
 
+    // Required to fetch `androidx.annotation:annotation:1.1.0`,
+    // which is a transitive dependency of `com.google.cloud:google-cloud-datastore`.
+    repositories {
+        google()
+    }
+
     dependencies {
         errorprone deps.build.errorProneCore
         errorproneJavac deps.build.errorProneJavac

config

@@ -34,7 +34,7 @@ dependencies {
     }
 
     testImplementation project(path: ":testutil-gcloud")
-    testImplementation "io.spine:spine-server:"
+    testImplementation "io.spine:spine-server:$spineCoreVersion"
 }
 
 task startDatastore(type: com.github.psxpaul.task.ExecFork) {

datastore/build.gradle

@@ -1,7 +1,11 @@
-This folder contains a sample Datastore index configuration in `index.yaml` file.
+This folder contains sample configurations of Datastore indexes
 
-The Datastore index must be configured as follows:
- - Contain the Spine internal types config.
- - Contain your custom `Aggregate` types config.
- 
-For more details, please see `index.yaml`.
+Depending on the mode in which your Datastore runs one should choose either
+
+- `index-datastore-native.yaml` for the Datastore in native mode;
+- `index-firestore-in-datastore-mode.yaml` for the Firestore in Datastore mode.
+
+Each of the configurations lists the indexes for the system records (e.g. `Event`) and the samples 
+for the user-defined entities.
+
+Feel free to modify the samples according to your needs.

datastore/config/README.md

@@ -1,4 +1,4 @@
-# The GAE Datastore sample indexes.
+# The indexes for Datastore in native mode.
 
 indexes:
 
@@ -19,7 +19,7 @@ indexes:
       - name: when_received
       - name: version
 
-  # Index required for `DsInboxStorage.oldestMessageToDeliver` query.
+  # Index required for `DsInboxStorage.newestMessageToDeliver` query.
 
   - kind: spine.server.delivery.InboxMessage
     properties:

datastore/config/index.yaml …store/config/index-datastore-native.yamldatastore/config/index.yaml renamed to datastore/config/index-datastore-native.yaml datastore/config/index.yaml renamed to datastore/config/index-datastore-native.yaml

@@ -0,0 +1,61 @@
+# The indexes for Firestore in Datastore mode.
+
+indexes:
+
+  # Common index for all the applications.
+
+  - kind: spine.core.Event
+    ancestor: no
+    properties:
+      - name: type
+      - name: created
+
+  # Index required for `DsInboxStorage.readAll` query.
+
+  - kind: spine.server.delivery.InboxMessage
+    ancestor: yes
+    properties:
+      - name: inbox_shard
+      - name: of_total_inbox_shards
+      - name: received_at
+      - name: version
+
+  # Index required for `DsInboxStorage.newestMessageToDeliver` query.
+
+  - kind: spine.server.delivery.InboxMessage
+    ancestor: yes
+    properties:
+      - name: inbox_shard
+      - name: of_total_inbox_shards
+      - name: status
+
+  # Indexes for the Aggregates.
+  #
+  # Each custom Aggregate type must have such indexes in order to be retrieved properly.
+  #
+  # The Datastore kind represents the Protobuf type name of the Aggregate state. Both the property
+  # names and the index directions should be kept unchanged.
+
+  - kind: myapp.example.MyFirstAggregate # Replace with your Aggregate Protobuf type name.
+    # Copy as is:
+    # ------------------
+    properties:
+      - name: aggregate_id
+      - name: created
+        direction: desc
+      - name: version
+        direction: desc
+      - name: snapshot
+    # ------------------
+
+  - kind: myapp.example.MySecondAggregate # Replace with your Aggregate Protobuf type name.
+    # Copy as is:
+    # ------------------
+    properties:
+      - name: aggregate_id
+      - name: created
+        direction: desc
+      - name: version
+        direction: desc
+      - name: snapshot
+    # ------------------

datastore/config/index-firestore-in-datastore-mode.yaml

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2020, TeamDev. All rights reserved.
+ *
+ * Redistribution and use in source and/or binary forms, with or without
+ * modification, must retain the above copyright notice and the following
+ * disclaimer.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package io.spine.server.storage.datastore;
+
+/**
+ * A mode in which Google Datastore operates.
+ */
+public enum DatastoreMode {
+
+    /**
+     * Native Datastore mode.
+     *
+     * <p>In this mode, Datastore is mostly eventually consistent. Also it has significant
+     * <a href="https://cloud.google.com/datastore/docs/concepts/limits#Cloud_Datastore_limits">limitations</a>
+     * on accessing the records.
+     *
+     * <p>Since recently, the projects in Google Cloud cannot be created in this mode.
+     * So this one works primarily for the legacy applications.
+     */
+    NATIVE,
+
+    /**
+     * Firestore in Datastore mode.
+     *
+     * <p>In this mode, Datastore becomes strongly consistent in most cases. However, some limits
+     * still <a href="https://cloud.google.com/datastore/docs/concepts/limits#limits">apply.</a>
+     *
+     * <p>All new Cloud projects provide this mode as a default one.
+     */
+    FIRESTORE_AS_DATASTORE
+}

datastore/src/main/java/io/spine/server/storage/datastore/DatastoreMode.java

@@ -26,10 +26,12 @@
 import com.google.common.collect.Iterables;
 import com.google.errorprone.annotations.CanIgnoreReturnValue;
 import io.spine.annotation.Internal;
+import io.spine.annotation.SPI;
 import io.spine.server.BoundedContextBuilder;
 import io.spine.server.ContextSpec;
 import io.spine.server.aggregate.Aggregate;
 import io.spine.server.aggregate.AggregateStorage;
+import io.spine.server.delivery.CatchUpStorage;
 import io.spine.server.delivery.InboxStorage;
 import io.spine.server.entity.Entity;
 import io.spine.server.entity.storage.ColumnMapping;
@@ -140,21 +142,54 @@ public BoundedContextBuilder configureTenantIndex(BoundedContextBuilder builder)
 
         DsProjectionStorageDelegate<I> recordStorage =
                 configure(DsProjectionStorageDelegate.newDelegateBuilder(), cls, context);
-        DsPropertyStorage propertyStorage = createPropertyStorage(context);
         DsProjectionStorage<I> result =
                 new DsProjectionStorage<>(cls,
                                           recordStorage,
-                                          propertyStorage,
                                           context.isMultitenant());
         return result;
     }
 
+    /**
+     * {@inheritDoc}
+     *
+     * Creates a Datastore-specific {@link InboxStorage} taking into account the support
+     * of multi-tenant storage required.
+     *
+     * <p>By default, creates an instance of storage for Datastore in native mode.
+     *
+     * @apiNote In order to change this behavior and supply a custom implementation, an SPI
+     *         user should override {@link #inboxStorageWith(DatastoreWrapper, boolean)
+     *         inboxStorageWith(multitenant, DatastoreWrapper)} method.
+     * @see DsInboxStorage for more details on supported modes
+     */
     @Override
-    public InboxStorage createInboxStorage(boolean multitenant) {
+    public final InboxStorage createInboxStorage(boolean multitenant) {
         DatastoreWrapper wrapper = systemWrapperFor(InboxStorage.class, multitenant);
+        return inboxStorageWith(wrapper, multitenant);
+    }
+
+    /**
+     * Creates a Datastore-specific {@link InboxStorage}.
+     *
+     * <p>SPI users should override this method in order to supply a custom implementation.
+     *
+     * @param wrapper
+     *         a wrapper over Datastore
+     * @param multitenant
+     *         whether the created storage should support multi-tenancy
+     * @return a new instance of {@code InboxStorage}
+     */
+    @SPI
+    protected InboxStorage inboxStorageWith(DatastoreWrapper wrapper, boolean multitenant) {
         return new DsInboxStorage(wrapper, multitenant);
     }
 
+    @Override
+    public CatchUpStorage createCatchUpStorage(boolean multitenant) {
+        DatastoreWrapper wrapper = systemWrapperFor(CatchUpStorage.class, multitenant);
+        return new DsCatchUpStorage(wrapper, multitenant);
+    }
+
     public ColumnMapping<Value<?>> columnMapping() {
         return columnMapping;
     }
@@ -196,7 +231,8 @@ private NsConverterFactory converterFactory() {
     }
 
     private String namespaceFromOptions() {
-        return nullToEmpty(datastore.getOptions().getNamespace());
+        return nullToEmpty(datastore.getOptions()
+                                    .getNamespace());
     }
 
     /**

datastore/src/main/java/io/spine/server/storage/datastore/DatastoreStorageFactory.java

@@ -337,7 +337,7 @@ private static <R> StructuredQuery<R> limit(StructuredQuery<R> query,
      * Deletes all existing {@link Entity Entities} with the given keys.
      *
      * @param keys
-     *         {@link Key Keys} of the {@link Entity Entities} to delete. May be nonexistent
+     *         {@code Keys} of the {@code Entities} to delete. May be nonexistent
      */
     public void delete(Key... keys) {
         datastore.delete(keys);
@@ -418,7 +418,7 @@ public final TransactionWrapper newTransaction() {
     public KeyFactory keyFactory(Kind kind) {
         checkNotNull(kind);
         KeyFactory keyFactory = datastore.newKeyFactory()
-                                        .setKind(kind.value());
+                                         .setKind(kind.value());
         Namespace namespace = namespaceSupplier.get();
         _trace().log("Retrieving KeyFactory for kind `%s` in `%s` namespace.",
                      kind, namespace.value());

datastore/src/main/java/io/spine/server/storage/datastore/DatastoreWrapper.java

@@ -0,0 +1,123 @@
+/*
+ * Copyright 2020, TeamDev. All rights reserved.
+ *
+ * Redistribution and use in source and/or binary forms, with or without
+ * modification, must retain the above copyright notice and the following
+ * disclaimer.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package io.spine.server.storage.datastore;
+
+import com.google.cloud.datastore.EntityQuery;
+import com.google.cloud.datastore.LongValue;
+import com.google.cloud.datastore.Query;
+import com.google.cloud.datastore.StringValue;
+import com.google.common.collect.ImmutableList;
+import com.google.protobuf.Timestamp;
+import io.spine.server.delivery.CatchUp;
+import io.spine.server.delivery.CatchUpId;
+import io.spine.server.delivery.CatchUpReadRequest;
+import io.spine.server.delivery.CatchUpStorage;
+import io.spine.type.TypeUrl;
+
+import java.util.Iterator;
+
+import static com.google.cloud.datastore.StructuredQuery.PropertyFilter.eq;
+import static com.google.protobuf.util.Timestamps.toNanos;
+
+/**
+ * A Google Datastore-backed storage implementation of {@link CatchUpStorage}.
+ */
+public class DsCatchUpStorage extends DsMessageStorage<CatchUpId, CatchUp, CatchUpReadRequest>
+        implements CatchUpStorage {
+
+    protected DsCatchUpStorage(DatastoreWrapper datastore, boolean multitenant) {
+        super(datastore, multitenant);
+    }
+
+    @Override
+    public Iterable<CatchUp> readAll() {
+        EntityQuery.Builder builder = Query.newEntityQueryBuilder();
+        return readAsIterable(builder);
+    }
+
+    private Iterable<CatchUp> readAsIterable(EntityQuery.Builder builder) {
+        Iterator<CatchUp> iterator = read(builder);
+        return ImmutableList.copyOf(iterator);
+    }
+
+    @Override
+    public Iterable<CatchUp> readByType(TypeUrl url) {
+        EntityQuery.Builder builder = Query.newEntityQueryBuilder();
+        builder.setFilter(eq(Column.projectionType.columnName(), url.value()));
+        return readAsIterable(builder);
+    }
+
+    @Override
+    protected  CatchUpId idOf(CatchUp message) {
+        return message.getId();
+    }
+
+    @Override
+    protected  MessageColumn<CatchUp>[] columns() {
+        return Column.values();
+    }
+
+    /**
+     * The columns of the {@code InboxMessage} kind in Datastore.
+     */
+    private enum Column implements MessageColumn<CatchUp> {
+
+        status("catchup_status", (m) -> {
+            return StringValue.of(m.getStatus()
+                                   .toString());
+        }),
+
+        whenLastRead("when_last_read", (m) -> {
+            Timestamp timestamp = m.getWhenLastRead();
+            return LongValue.of(toNanos(timestamp));
+        }),
+
+        projectionType("projection_type", (m) -> {
+            return StringValue.of(m.getId()
+                                   .getProjectionType());
+        });
+
+        /**
+         * The column name.
+         */
+        private final String name;
+
+        /**
+         * Obtains the value of the column from the given message.
+         */
+        private final Getter<CatchUp> getter;
+
+        Column(String name, Getter<CatchUp> getter) {
+            this.name = name;
+            this.getter = getter;
+        }
+
+        @Override
+        public String columnName() {
+            return name;
+        }
+
+        @Override
+        public Getter<CatchUp> getter() {
+            return getter;
+        }
+    }
+}