Initial PR (#1)

* Add lib & example app, with unit coverage

* Add CI files

* Cross-compile to Scala 3

* Initial integration support

* Extend integration support

* Add stress test

* Add timeout to actor system termination

* Actor system is already shutdown by endless

* Add scaladoc

* Add retry in self-messaging

* Add coverage for Akka runtime

* Add paradox site

---------

Co-authored-by: Jonas Chapuis <jonas.chapuis@zf.com>

.github/workflows/ci.yml

@@ -0,0 +1,30 @@
+name: CI
+on:
+  pull_request:
+jobs:
+  build:
+    runs-on: ubuntu-latest
+
+    services:
+      postgres:
+        image: postgres:11
+        env:
+          POSTGRES_USER: postgres
+          POSTGRES_PASSWORD: postgres
+          POSTGRES_DB: example
+        ports:
+          - 5432:5432
+        options: --health-cmd pg_isready --health-interval 10s --health-timeout 5s --health-retries 5
+
+    steps:
+      - uses: actions/checkout@v2
+        with:
+          fetch-depth: 0
+      - uses: coursier/setup-action@v1.3.3
+        with:
+          jvm: temurin:17
+          apps: sbt
+      - run: sbt scalafmtCheckAll compile coverage +test example/multi-jvm:test coverageReport coverageAggregate versionPolicyCheck documentation/makeSite
+      - uses: codecov/codecov-action@v2.1.0
+        env:
+          CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}

.github/workflows/release.yml

@@ -0,0 +1,67 @@
+name: Release
+on:
+  push:
+    tags:
+      - 'v*'
+  workflow_dispatch:
+
+jobs:
+  coverage:
+    runs-on: ubuntu-latest
+    services:
+      postgres:
+        image: postgres:11
+        env:
+          POSTGRES_USER: postgres
+          POSTGRES_PASSWORD: postgres
+          POSTGRES_DB: example
+        ports:
+          - 5432:5432
+        options: --health-cmd pg_isready --health-interval 10s --health-timeout 5s --health-retries 5
+    steps:
+      - uses: actions/checkout@v2.3.4
+      - uses: coursier/setup-action@v1.3.3
+        with:
+          jvm: temurin:17
+          apps: sbt
+      - name: Test and compute coverage
+        run: sbt scalafmtCheckAll coverage +test example/multi-jvm:test coverageReport coverageAggregate
+      - name: Codecov
+        uses: codecov/codecov-action@v2.1.0
+        env:
+          CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
+
+  publish:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+        with:
+          fetch-depth: 0
+      - uses: coursier/setup-action@v1.3.3
+        with:
+          jvm: temurin:17
+          apps: sbt
+      - run: sbt versionCheck ci-release
+        env:
+          PGP_PASSPHRASE: ${{ secrets.PGP_PASSPHRASE }}
+          PGP_SECRET: ${{ secrets.PGP_SECRET }}
+          SONATYPE_PASSWORD: ${{ secrets.SONATYPE_PASSWORD }}
+          SONATYPE_USERNAME: ${{ secrets.SONATYPE_USERNAME }}
+      - uses: codecov/codecov-action@v2
+
+  publishSite:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2.3.4
+      - uses: coursier/setup-action@v1.3.3
+        with:
+          jvm: temurin:17
+          apps: sbt
+      - name: Generate website
+        run: sbt documentation/makeSite
+      - uses: JamesIves/github-pages-deploy-action@4.1.5
+        with:
+          repository-name: endless4s/endless-transaction.github.io
+          branch: master
+          token: ${{ secrets.DOC_GITHUB_TOKEN }}
+          folder: documentation/target/site

.gitignore

@@ -1,5 +1,24 @@
 *.class
 *.log
 
-# virtual machine crash logs, see http://www.java.com/en/download/help/error_hotspot.xml
-hs_err_pid*
+# sbt specific
+.DS_Store/
+.cache/
+.history/
+.metals/
+.lib/
+.bloop/
+.vscode/
+dist/*
+target/
+lib_managed/
+src_managed/
+project/boot/
+project/plugins/project/
+.bsp/
+.DS_Store
+
+# Scala-IDE specific
+.scala_dependencies
+.worksheet
+.idea

.scalafmt.conf

@@ -0,0 +1,3 @@
+version = 3.7.17
+runner.dialect = Scala213Source3
+maxColumn = 100

CODE_OF_CONDUCT.md

@@ -0,0 +1,7 @@
+# Code of Conduct
+We are committed to providing a friendly, safe and welcoming environment for all, regardless of level of experience, gender, gender identity and expression, sexual orientation, disability, personal appearance, body size, race, ethnicity, age, religion, nationality, or other such characteristics.
+
+Everyone is expected to follow the Scala Code of Conduct when discussing the project on the available communication channels. If you are being harassed, please contact us immediately so that we can support you.
+
+## Moderation
+Any questions, concerns, or moderation requests please contact a member of the project.

CONTRIBUTING.md

@@ -0,0 +1,53 @@
+# Contributing
+
+## Requirements
+
+You will need the following tools:
+
+- [Git](https://git-scm.com/)
+- [Java 8](http://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html)
+- [SBT](http://www.scala-sbt.org/)
+
+## Workflow
+
+Create your own fork of the repository and work in a local branch based on `master`.
+
+## Code Style
+
+We use [scalafmt](https://scalameta.org/scalafmt/) to format the source code, and recommend you to setup
+your editor to “format on save”, as documented [here](https://scalameta.org/scalafmt/docs/installation.html).
+
+## Run Tests
+
+From the sbt shell:
+
+~~~
++test
+~~~
+
+## Preview Documentation
+
+From the sbt shell:
+
+~~~
+documentation/previewSite
+~~~
+
+It will open the generated documentation in your browser.
+
+## Publish a Release
+
+Push a Git tag:
+
+~~~ bash
+$ git tag v2.0.0
+$ git push origin v2.0.0
+~~~
+
+Then, reset the compatibility guarantees in `build.sbt`:
+
+~~~
+versionPolicyIntention := Compatibility.BinaryAndSourceCompatible
+~~~
+
+Commit and push the changes.

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2021 Jonas Chapuis & contributors
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1,3 @@
+# endless-transaction
+
+endless-transaction is a Scala library that provides a functional abstraction for distributed transactions based on cats-effect and the endless library. It is designed to simplify the process of coordinating transactions using the two-phase commit protocol. Transactions are implemented with persistent entities using event-sourcing, making the system resilient to failures.

akka/src/main/resources/reference.conf

@@ -0,0 +1,34 @@
+endless.transaction.akka {
+  # Parameters for retries in self-messaging i.e. the branch side-effect notifying the transaction entity
+  # about the protocol phase result. It could be that the transaction gets rebalanced to another
+  # node in the meantime, so we need to need to retry if the cluster doesn't respond in a timely
+  # manner.
+  retries {
+    initial-delay = 5s  # initial delay before the first retry
+    max-retries = 10    # maximum number of retries to avoid cascading failures
+  }
+
+  # Akka's ask timeout for the transaction entity
+  ask-timeout = 60s
+}
+
+akka {
+  actor {
+    serializers {
+      scalapb-serializer = "endless.runtime.akka.protobuf.ScalaPbSerializer"
+    }
+    serialization-identifiers {
+      "endless.runtime.akka.protobuf.ScalaPbSerializer" = 424242
+    }
+    serialization-bindings {
+      "endless.transaction.proto.events.TransactionEvent" = scalapb-serializer
+    }
+  }
+
+  cluster {
+    sharding {
+      # guaranteed pending transaction recovery
+      remember-entities = on
+    }
+  }
+}

akka/src/main/scala/endless/transaction/akka/AkkaTransactor.scala

@@ -0,0 +1,79 @@
+package endless.transaction.akka
+
+import akka.persistence.typed.{EventAdapter, EventSeq}
+import akka.util.Timeout
+import cats.Show
+import cats.effect.kernel.{Async, Resource}
+import endless.runtime.akka.deploy.AkkaCluster
+import endless.runtime.akka.deploy.AkkaDeployer.AkkaDeploymentParameters
+import endless.runtime.akka.syntax.deploy.DeploymentParameters
+import endless.transaction.impl.adapter.TransactionEventAdapter
+import endless.transaction.impl.data.{TransactionEvent, TransactionState}
+import endless.transaction.proto.events
+import endless.transaction.*
+import endless.transaction.impl.helpers.RetryHelpers
+import org.typelevel.log4cats.Logger
+
+import scala.concurrent.duration.*
+
+class AkkaTransactor[F[_]: Async: Logger](implicit akkaCluster: AkkaCluster[F])
+    extends Transactor[F] {
+
+  def coordinator[
+      TID: StringCodec: BinaryCodec,
+      BID: BinaryCodec: Show,
+      Q: BinaryCodec,
+      R: BinaryCodec
+  ](
+      transactionName: String,
+      branchForID: BID => Branch[F, TID, Q, R],
+      timeout: Option[FiniteDuration]
+  ): Resource[F, Coordinator[F, TID, BID, Q, R]] = {
+    type S = TransactionState[TID, BID, Q, R]
+    type E = TransactionEvent[TID, BID, Q, R]
+    val config = Config.load()
+    implicit val askTimeout: Timeout = Timeout(config.askTimeout)
+    val eventAdapter = new TransactionEventAdapter[TID, BID, Q, R]()
+    implicit val akkaDeploymentParameters: DeploymentParameters[F, TID, S, E] = {
+      AkkaDeploymentParameters[F, S, E](
+        customizeBehavior = (_, behavior) =>
+          behavior.eventAdapter(
+            new EventAdapter[TransactionEvent[TID, BID, Q, R], proto.events.TransactionEvent] {
+              override def toJournal(
+                  event: TransactionEvent[TID, BID, Q, R]
+              ): proto.events.TransactionEvent = eventAdapter.toJournal(event)
+
+              def manifest(event: TransactionEvent[TID, BID, Q, R]): String = event.getClass.getName
+
+              def fromJournal(
+                  p: events.TransactionEvent,
+                  manifest: String
+              ): EventSeq[TransactionEvent[TID, BID, Q, R]] =
+                EventSeq.single(eventAdapter.fromJournal(p))
+            }
+          )
+      )
+    }
+    implicit val retryParameters: RetryHelpers.RetryParameters = config.retries.parameters
+    deployEntityBasedCoordinator(
+      transactionName,
+      branchForID,
+      timeout,
+      endless.runtime.akka.syntax.deploy
+    ).map(_.repository)
+  }
+}
+
+object AkkaTransactor {
+
+  /** Create a new AkkaTransactor
+    * @param akkaCluster
+    *   the AkkaCluster
+    * @tparam F
+    *   the effect type
+    * @return
+    *   a new AkkaTransactor
+    */
+  def apply[F[_]: Async: Logger](implicit akkaCluster: AkkaCluster[F]): AkkaTransactor[F] =
+    new AkkaTransactor[F]
+}

akka/src/main/scala/endless/transaction/akka/Config.scala

@@ -0,0 +1,25 @@
+package endless.transaction.akka
+
+import com.typesafe.config.ConfigFactory
+import endless.transaction.impl.helpers.RetryHelpers.RetryParameters
+
+import scala.concurrent.duration.*
+
+final case class Config(retries: Config.RetryConfig, askTimeout: FiniteDuration)
+
+object Config {
+  final case class RetryConfig(initialDelay: FiniteDuration, maxRetries: Int) {
+    lazy val parameters: RetryParameters = RetryParameters(initialDelay, maxRetries)
+  }
+
+  def load(): Config = {
+    val config = ConfigFactory.load()
+    val retriesConfig = config.getConfig("endless.transaction.akka.retries")
+    val initialDelay = retriesConfig.getDuration("initial-delay", MILLISECONDS).millis
+    val maxRetries = retriesConfig.getInt("max-retries")
+
+    val askTimeout = config.getDuration("endless.transaction.akka.ask-timeout", SECONDS).seconds
+    Config(RetryConfig(initialDelay, maxRetries), askTimeout)
+  }
+
+}