feat: add multi stream append support (#330)

Author: YoEight

.github/workflows/ci.yml

@@ -15,5 +15,5 @@ jobs:
     name: Tests (CI)
     uses: ./.github/workflows/tests.yml
     with:
-      image: ${{ fromJSON(vars.KURRENTDB_DOCKER_IMAGES).ci.fullname }}
+      runtime: ci
     secrets: inherit

.github/workflows/load-configuration.yml

@@ -0,0 +1,46 @@
+name: Load KurrentDB Runtime Configuration
+on:
+  workflow_call:
+    inputs:
+      runtime:
+        description: "The runtime's name. Current options are: `ci`, `previous-lts`, `latest`"
+        type: string
+
+    outputs:
+      runtime:
+        description: The runtime's name
+        value: ${{ inputs.runtime }}
+
+      registry:
+        description: The Docker registry
+        value: ${{ jobs.load.outputs.registry }}
+
+      image:
+        description: The Docker image
+        value: ${{ jobs.load.outputs.image }}
+
+      tag:
+        description: The Docker image tag
+        value: ${{ jobs.load.outputs.tag }}
+
+      full_image_name:
+        description: The full Docker image name (including registry, image, and tag)
+        value: ${{ jobs.load.outputs.full_image_name }}
+
+jobs:
+  load:
+    runs-on: ubuntu-latest
+    outputs:
+        registry: ${{ steps.set.outputs.registry }}
+        image: ${{ steps.set.outputs.image }}
+        tag: ${{ steps.set.outputs.tag }}
+        full_image_name: ${{ steps.set.outputs.full_image_name }}
+
+    steps:
+      - name: Set KurrentDB Runtime Configuration Properties
+        id: set
+        run: |
+          echo "registry=${{ fromJSON(vars.KURRENTDB_DOCKER_IMAGES)[inputs.runtime].registry }}" >> $GITHUB_OUTPUT
+          echo "tag=${{ fromJSON(vars.KURRENTDB_DOCKER_IMAGES)[inputs.runtime].tag }}" >> $GITHUB_OUTPUT
+          echo "image=${{ fromJSON(vars.KURRENTDB_DOCKER_IMAGES)[inputs.runtime].image }}" >> $GITHUB_OUTPUT
+          echo "full_image_name=${{ fromJSON(vars.KURRENTDB_DOCKER_IMAGES)[inputs.runtime].fullname }}" >> $GITHUB_OUTPUT

.github/workflows/lts.yml

@@ -15,7 +15,7 @@ jobs:
     name: Tests (LTS)
     uses: ./.github/workflows/tests.yml
     with:
-      image: ${{ fromJSON(vars.KURRENTDB_DOCKER_IMAGES).lts.fullname }}
+      runtime: lts
     secrets: inherit
 
   # Will be removed in the future
@@ -24,5 +24,7 @@ jobs:
 
     uses: ./.github/workflows/plugins-tests.yml
     with:
-      image: "docker.eventstore.com/eventstore-ee/eventstoredb-commercial:24.2.0-jammy"
+      registry: docker.eventstore.com/eventstore-ee
+      image: eventstoredb-commercial
+      tag: 24.2.0-jammy
     secrets: inherit

.github/workflows/plugins-tests.yml

@@ -3,9 +3,15 @@ name: enterprise plugins tests workflow
 on:
   workflow_call:
     inputs:
+      registry:
+        required: true
+        type: string
       image:
         required: true
         type: string
+      tag:
+        required: true
+        type: string
 
 jobs:
   single_node:
@@ -46,7 +52,7 @@ jobs:
       - name: Execute Gradle build
         run: ./gradlew ci --tests ${{ matrix.test }}Tests
         env:
-          KURRENTDB_IMAGE: ${{ inputs.image }}
+          KURRENTDB_IMAGE: ${{inputs.registry}}/${{ inputs.image }}:${{ inputs.tag }}
           SECURE: true
 
       - uses: actions/upload-artifact@v4
@@ -81,7 +87,9 @@ jobs:
       - name: Set up cluster with Docker Compose
         run: docker compose up -d
         env:
-          KURRENTDB_IMAGE: ${{ inputs.image }}
+          KURRENTDB_DOCKER_REGISTRY: ${{ inputs.registry }}
+          KURRENTDB_DOCKER_IMAGE: ${{ inputs.image }}
+          KURRENTDB_DOCKER_TAG: ${{ inputs.tag }}
 
       - name: Generate user certificates
         run: docker compose --file configure-user-certs-for-tests.yml up

.github/workflows/previous-lts.yml

@@ -15,5 +15,5 @@ jobs:
     name: Tests (Previous LTS)
     uses: ./.github/workflows/tests.yml
     with:
-      image: ${{ fromJSON(vars.KURRENTDB_DOCKER_IMAGES)['previous-lts'].fullname }}
+      runtime: previous-lts
     secrets: inherit

.github/workflows/tests.yml

@@ -3,18 +3,24 @@ name: tests workflow
 on:
   workflow_call:
     inputs:
-      image:
+      runtime:
         required: true
         type: string
 
 jobs:
+  load_configuration:
+    uses: ./.github/workflows/load-configuration.yml
+    with:
+      runtime: ${{ inputs.runtime }}
+
   single_node:
+    needs: load_configuration
     name: Single node
 
     strategy:
       fail-fast: false
       matrix:
-        test: [Streams, PersistentSubscriptions, Telemetry]
+        test: [Streams, PersistentSubscriptions, Telemetry, MultiStreamAppend]
 
     runs-on: ubuntu-latest
     steps:
@@ -41,16 +47,10 @@ jobs:
       - name: Execute Gradle build
         run: ./gradlew ci --tests ${{ matrix.test }}Tests
         env:
-          KURRENTDB_IMAGE: ${{ inputs.image }}
-
-      - uses: actions/upload-artifact@v4
-        if: failure()
-        with:
-          name: esdb_logs.tar.gz
-          path: /tmp/esdb_logs.tar.gz
-          if-no-files-found: error
+          KURRENTDB_IMAGE: ${{ needs.load_configuration.outputs.full_image_name }}
 
   secure:
+    needs: load_configuration
     name: Secure
 
     strategy:
@@ -86,7 +86,7 @@ jobs:
       - name: Execute Gradle build
         run: ./gradlew ci --tests ${{ matrix.test }}Tests
         env:
-          KURRENTDB_IMAGE: ${{ inputs.image }}
+          KURRENTDB_IMAGE: ${{ needs.load_configuration.outputs.full_image_name }}
           SECURE: true
 
       - uses: actions/upload-artifact@v4
@@ -96,12 +96,13 @@ jobs:
           path: /tmp/esdb_logs.tar.gz
 
   cluster:
+    needs: load_configuration
     name: Cluster
 
     strategy:
       fail-fast: false
       matrix:
-        test: [Streams, PersistentSubscriptions]
+        test: [Streams, PersistentSubscriptions, MultiStreamAppend]
 
     runs-on: ubuntu-latest
     steps:
@@ -117,7 +118,9 @@ jobs:
       - name: Set up cluster with Docker Compose
         run: docker compose up -d
         env:
-          KURRENTDB_IMAGE: ${{ inputs.image }}
+          KURRENTDB_DOCKER_REGISTRY: ${{ needs.load_configuration.outputs.registry }}
+          KURRENTDB_DOCKER_IMAGE: ${{ needs.load_configuration.outputs.image }}
+          KURRENTDB_DOCKER_TAG: ${{ needs.load_configuration.outputs.tag }}
 
       - name: Set up JDK 8
         uses: actions/setup-java@v3

docker-compose.yml

@@ -1,5 +1,3 @@
-version: "3.5"
-
 services:
   volumes-provisioner:
     image: hasnat/volumes-provisioner
@@ -26,71 +24,57 @@ services:
       - volumes-provisioner
 
   esdb-node1: &template
-    image: ${KURRENTDB_IMAGE:-docker.kurrent.io/eventstore/eventstoredb-ee:lts}
+    image: ${KURRENTDB_DOCKER_REGISTRY:-docker.kurrent.io/eventstore}/${KURRENTDB_DOCKER_IMAGE:-eventstoredb-ee}:${KURRENTDB_DOCKER_TAG:-lts}
     env_file:
       - vars.env
     environment:
       - EVENTSTORE_GOSSIP_SEED=172.30.240.12:2113,172.30.240.13:2113
-      - EVENTSTORE_INT_IP=172.30.240.11
-      - EVENTSTORE_CERTIFICATE_FILE=/etc/eventstore/certs/node1/node.crt
-      - EVENTSTORE_CERTIFICATE_PRIVATE_KEY_FILE=/etc/eventstore/certs/node1/node.key
-      - EVENTSTORE_ADVERTISE_HTTP_PORT_TO_CLIENT_AS=2111
+      - EVENTSTORE_REPLICATION_IP=172.30.240.11
+      - EVENTSTORE_CERTIFICATE_FILE=/etc/kurrentdb/certs/node1/node.crt
+      - EVENTSTORE_CERTIFICATE_PRIVATE_KEY_FILE=/etc/kurrentdb/certs/node1/node.key
+      - EVENTSTORE_ADVERTISE_NODE_PORT_TO_CLIENT_AS=2111
     ports:
       - 2111:2113
     networks:
       clusternetwork:
         ipv4_address: 172.30.240.11
     volumes:
-      - ./certs:/etc/eventstore/certs
+      - ./certs:/etc/kurrentdb/certs
     restart: unless-stopped
     depends_on:
       - cert-gen
 
   esdb-node2:
     <<: *template
-    env_file:
-      - vars.env
     environment:
       - EVENTSTORE_GOSSIP_SEED=172.30.240.11:2113,172.30.240.13:2113
-      - EVENTSTORE_INT_IP=172.30.240.12
-      - EVENTSTORE_CERTIFICATE_FILE=/etc/eventstore/certs/node2/node.crt
-      - EVENTSTORE_CERTIFICATE_PRIVATE_KEY_FILE=/etc/eventstore/certs/node2/node.key
-      - EVENTSTORE_ADVERTISE_HTTP_PORT_TO_CLIENT_AS=2112
+      - EVENTSTORE_REPLICATION_IP=172.30.240.12
+      - EVENTSTORE_CERTIFICATE_FILE=/etc/kurrentdb/certs/node2/node.crt
+      - EVENTSTORE_CERTIFICATE_PRIVATE_KEY_FILE=/etc/kurrentdb/certs/node2/node.key
+      - EVENTSTORE_ADVERTISE_NODE_PORT_TO_CLIENT_AS=2112
     ports:
       - 2112:2113
     networks:
       clusternetwork:
         ipv4_address: 172.30.240.12
-    volumes:
-      - ./certs:/etc/eventstore/certs
-    restart: unless-stopped
-    depends_on:
-      - cert-gen
 
   esdb-node3:
     <<: *template
-    env_file:
-      - vars.env
     environment:
       - EVENTSTORE_GOSSIP_SEED=172.30.240.11:2113,172.30.240.12:2113
-      - EVENTSTORE_INT_IP=172.30.240.13
-      - EVENTSTORE_CERTIFICATE_FILE=/etc/eventstore/certs/node3/node.crt
-      - EVENTSTORE_CERTIFICATE_PRIVATE_KEY_FILE=/etc/eventstore/certs/node3/node.key
-      - EVENTSTORE_ADVERTISE_HTTP_PORT_TO_CLIENT_AS=2113
+      - EVENTSTORE_REPLICATION_IP=172.30.240.13
+      - EVENTSTORE_CERTIFICATE_FILE=/etc/kurrentdb/certs/node3/node.crt
+      - EVENTSTORE_CERTIFICATE_PRIVATE_KEY_FILE=/etc/kurrentdb/certs/node3/node.key
+      - EVENTSTORE_ADVERTISE_NODE_PORT_TO_CLIENT_AS=2113
     ports:
       - 2113:2113
     networks:
       clusternetwork:
         ipv4_address: 172.30.240.13
-    volumes:
-      - ./certs:/etc/eventstore/certs
-    restart: unless-stopped
-    depends_on:
-      - cert-gen
 
 networks:
   clusternetwork:
-    name: eventstoredb.local
+    name: kurrentdb.local
     driver: bridge
     ipam:
       driver: default

src/main/java/io/kurrent/dbclient/AppendStreamFailure.java

@@ -0,0 +1,36 @@
+package io.kurrent.dbclient;
+
+public class AppendStreamFailure {
+    private final io.kurrentdb.protocol.streams.v2.AppendStreamFailure inner;
+
+    AppendStreamFailure(io.kurrentdb.protocol.streams.v2.AppendStreamFailure inner) {
+        this.inner = inner;
+    }
+
+    public String getStreamName() {
+        return this.inner.getStream();
+    }
+
+    public void visit(MultiAppendStreamErrorVisitor visitor) {
+        if (this.inner.getErrorCase() == io.kurrentdb.protocol.streams.v2.AppendStreamFailure.ErrorCase.STREAM_REVISION_CONFLICT) {
+            visitor.onWrongExpectedRevision(this.inner.getStreamRevisionConflict().getStreamRevision());
+            return;
+        }
+
+        if (this.inner.getErrorCase() == io.kurrentdb.protocol.streams.v2.AppendStreamFailure.ErrorCase.ACCESS_DENIED) {
+            visitor.onAccessDenied(this.inner.getAccessDenied());
+        }
+
+        if (this.inner.getErrorCase() == io.kurrentdb.protocol.streams.v2.AppendStreamFailure.ErrorCase.STREAM_DELETED) {
+            visitor.onStreamDeleted();
+            return;
+        }
+
+       if (this.inner.getErrorCase() == io.kurrentdb.protocol.streams.v2.AppendStreamFailure.ErrorCase.TRANSACTION_MAX_SIZE_EXCEEDED) {
+            visitor.onTransactionMaxSizeExceeded(this.inner.getTransactionMaxSizeExceeded().getMaxSize());
+            return;
+       }
+
+       throw new IllegalArgumentException("Append failure does not match any known error type");
+    }
+}

src/main/java/io/kurrent/dbclient/AppendStreamRequest.java

@@ -0,0 +1,27 @@
+package io.kurrent.dbclient;
+
+import java.util.Iterator;
+
+public class AppendStreamRequest {
+    private final String streamName;
+    private final Iterator<EventData> events;
+    private final StreamState expectedState;
+
+    public AppendStreamRequest(String streamName, Iterator<EventData> events, StreamState expectedState) {
+        this.streamName = streamName;
+        this.events = events;
+        this.expectedState = expectedState;
+    }
+
+    public String getStreamName() {
+        return streamName;
+    }
+
+    public Iterator<EventData> getEvents() {
+        return events;
+    }
+
+    public StreamState getExpectedState() {
+        return expectedState;
+    }
+}

src/main/java/io/kurrent/dbclient/AppendStreamSuccess.java

@@ -0,0 +1,21 @@
+package io.kurrent.dbclient;
+
+public class AppendStreamSuccess {
+    private final io.kurrentdb.protocol.streams.v2.AppendStreamSuccess inner;
+
+    AppendStreamSuccess(io.kurrentdb.protocol.streams.v2.AppendStreamSuccess inner) {
+        this.inner = inner;
+    }
+
+    public String getStreamName() {
+        return this.inner.getStream();
+    }
+
+    public long getStreamRevision() {
+        return this.inner.getStreamRevision();
+    }
+
+    public long getPosition() {
+        return this.inner.getPosition();
+    }
+}