A possible new architecture about atomic-store

[acjay]

Been working on further thoughts on how this would look in implementation. I'm interested in seeing whether it's possible to factor the functionality of Atomic Store into smaller, more generic & cohesive pieces. This could have a lot less protocol than today's Atomic Store.

Another thing that is codified here is the use of async method API facads (i.e. Future-returning methods) instead of presenting actor refs. I think this is more type-safe, more encapsulated, and less error-prone. But to take advantage of Actor supervision, all the pieces are still implemented with actors and set up so that the supervision chain is intact.

• It occurs to me that an async interface to a simple log could be shaped like:

// Interaction would be through an async interface,
// rather than explicit message passing.
trait Log[EventType] {
  def persistenceId: String
  def read: Future[Seq[EventType]]
  def append(events: Seq[EventType]): Future[Done]
}

object Log {
  def apply[EventType](persistenceId: String)(implicit actorRefFactory: ActorRefFactory) = Log[EventType] = ???
}

Then the whole machine would be:

• An extremely simple PersistentActor, wrapped with the above interface.
• A Flow for managing the atomic request queue

// General purpose utility for allowing actors to process 
// messages asynchronously in strict sequence, since the
// actor model only provides this guarantee synchronously.
def atomicFlow[Input, Output](
  bufferSize: Int, 
  bufferOverflowStrategy: OverflowStrategy
)(
  f: Input => Future[Output]
)(
  implicit 
  mat: ActorMaterializer
): ActorRef = {
  Source
    .actorRef[(Input, ActorRef)]
    .mapAsync(1) { case (input, replyTo) => (f(input), replyTo) }
    .to(Sink.foreach { case (output, replyTo) => replyTo ! output })
    .run()

• An actor for binding the log to the atomic flow and taking

class AtomicEventLog[CommandType, EventType, OutputType](persistenceId: String, validator: (Command, Seq[EventType], Seq[EventType] => Future[Done]) => Future[OutputType) {
  implicit val materializer = ActorMaterializer()
  val log = Log[EventType](persistenceId)
  val atomicFlowActorRef = atomicFlow(10, OverflowStrategy.DropLast) { command =>
      validator(command, log.read, log.append)
  }

  // Need to figure out how to apply timeout rules and 
  // to make `atomicFlow` failures are supervised by
  // this actor.

  def receive = {
    case command: CommandType => atomicFlowActorRef ! (command, sender())
    case EventsForId(persistenceId) => log.read pipeTo sender()
  }
}

• This would be instantiated within a Receptionist, as it is today. It would be responsible for the life-cycle of the system and supervision.

• The AtomicStore class would take the validator as an argument and export the Props for the receptionist.

• The receptionist would be started as a Cluster Singleton.

• The application would instantiate the atomic store, passing a validator. The dynamic aspects of the process are arguments to the validator, and it can return an arbitrary data structure. Below shows how this would for a validator that requires a user-submitted application-level command and configuration that is stored outside of the event log. To Atomic Store, the command is the pair of those pieces of data. So now we have an Atomic Store, prewired with its validation logic, rather than the "validation dance" required in the current version.

// TODO: How to make sure that timeouts cancel work that is
// mid-stream?

object MyAtomicStore extends AtomicStore[CommandAndConfig] { 
  case ((command, config), pastEvents, appendToLog) =>
    async {
      val calculator = Calculator(config, pastEvents)
      val (newEvents, additionalData) = await(calculator.validateCommand(command))
      await(appendToLog(newEvents))
      (newEvents, additionalData)
    }
  }
}

• In the service layer, this could be invoked much more simply:

  def processCommandForId(command: Command, scope: String) = async {
    val config = await(configForId(scope))
    await(atomicStore.processCommand((command, config), scope))
  }