feat: add durable state actor (#100)

Author: Tochemey

Earthfile

@@ -82,7 +82,7 @@ local-test:
     FROM +vendor
 
     WITH DOCKER --pull postgres:11
-        RUN go-acc ./... -o coverage.out --ignore egopb,test,example,mocks -- -mod=vendor -race -v
+        RUN go-acc ./... -o coverage.out --ignore egopb,test,example,mocks -- -mod=vendor -timeout 0 -race -v
     END
 
     SAVE ARTIFACT coverage.out AS LOCAL coverage.out
@@ -92,7 +92,8 @@ mock:
     FROM +code
 
     # generate the mocks
-    RUN mockery  --dir eventstore --name EventsStore --keeptree --exported=true --with-expecter=true --inpackage=true --disable-version-string=true --output ./mocks/eventstore --case snake
+    RUN mockery  --dir persistence --all --keeptree --exported=true --with-expecter=true --inpackage=true --disable-version-string=true --output ./mocks/persistence --case snake
     RUN mockery  --dir offsetstore --name OffsetStore --keeptree --exported=true --with-expecter=true --inpackage=true --disable-version-string=true --output ./mocks/offsetstore --case snake
 
+
     SAVE ARTIFACT ./mocks mocks AS LOCAL mocks

behavior.go

@@ -34,8 +34,8 @@ type Command proto.Message
 type Event proto.Message
 type State proto.Message
 
-// EntityBehavior defines an event sourced behavior when modeling a CQRS EntityBehavior.
-type EntityBehavior interface {
+// EventSourcedBehavior defines an event sourced behavior when modeling a CQRS EventSourcedBehavior.
+type EventSourcedBehavior interface {
 	// ID defines the id that will be used in the event journal.
 	// This helps track the entity in the events store.
 	ID() string
@@ -46,7 +46,7 @@ type EntityBehavior interface {
 	// which validations must be applied, and finally, which events will be persisted if any. When there is no event to be persisted a nil can
 	// be returned as a no-op. Command handlers are the meat of the event sourced actor.
 	// They encode the business rules of your event sourced actor and act as a guardian of the event sourced actor consistency.
-	// The command eventSourcedHandler must first validate that the incoming command can be applied to the current model state.
+	// The command must first validate that the incoming command can be applied to the current model state.
 	//  Any decision should be solely based on the data passed in the commands and the state of the Behavior.
 	// In case of successful validation, one or more events expressing the mutations are persisted.
 	// Once the events are persisted, they are applied to the state producing a new valid state.
@@ -58,3 +58,32 @@ type EntityBehavior interface {
 	// Event handlers must be pure functions as they will be used when instantiating the event sourced actor and replaying the event journal.
 	HandleEvent(ctx context.Context, event Event, priorState State) (state State, err error)
 }
+
+// DurableStateBehavior represents a type of Actor that persists its full state after processing each command instead of using event sourcing.
+// This type of Actor keeps its current state in memory during command handling and based upon the command response
+// persists its full state into a durable store. The store can be a SQL or NoSQL database.
+// The whole concept is given the current state of the actor and a command produce a new state with a higher version as shown in this diagram: (State, Command) => State
+// DurableStateBehavior reacts to commands which result in a new version of the actor state. Only the latest version of the actor state is
+// persisted to the durable store. There is no concept of history regarding the actor state since this is not an event sourced actor.
+// However, one can rely on the version number of the actor state and exactly know how the actor state has evolved overtime.
+// State actor version number are numerically incremented by the command handler which means it is imperative that the newer version of the state is greater than the current version by one.
+//
+// DurableStateBehavior will attempt to recover its state whenever available from the durable state.
+// During a normal shutdown process, it will persist its current state to the durable store prior to shutting down.
+// This behavior help maintain some consistency across the actor state evolution.
+type DurableStateBehavior interface {
+	// ID defines the id that will be used in the event journal.
+	// This helps track the entity in the events store.
+	ID() string
+	// InitialState returns the durable state actor initial state.
+	// This is set as the initial state when there are no snapshots found the entity
+	InitialState() State
+	// HandleCommand processes every command sent to the DurableStateBehavior. One needs to use the command, the priorVersion and the priorState sent to produce a newState and newVersion.
+	// This defines how to handle each incoming command, which validations must be applied, and finally, whether a resulting state will be persisted depending upon the response.
+	// They encode the business rules of your durable state actor and act as a guardian of the actor consistency.
+	// The command handler must first validate that the incoming command can be applied to the current model state.
+	// Any decision should be solely based on the data passed in the command, the priorVersion and the priorState.
+	// In case of successful validation and processing , the new state will be stored in the durable store depending upon response.
+	// The actor state will be updated with the newState only if the newVersion is 1 more than the already existing state.
+	HandleCommand(ctx context.Context, command Command, priorVersion uint64, priorState State) (newState State, newVersion uint64, err error)
+}

durable_state_actor.go

@@ -0,0 +1,233 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2022-2024 Tochemey
+ *
+ * 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.
+ */
+
+package ego
+
+import (
+	"context"
+	"fmt"
+	"math"
+	"time"
+
+	"github.com/tochemey/goakt/v2/actors"
+	"github.com/tochemey/goakt/v2/goaktpb"
+	"golang.org/x/sync/errgroup"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/anypb"
+	"google.golang.org/protobuf/types/known/timestamppb"
+
+	"github.com/tochemey/ego/v3/egopb"
+	"github.com/tochemey/ego/v3/eventstream"
+	"github.com/tochemey/ego/v3/internal/errorschain"
+	"github.com/tochemey/ego/v3/persistence"
+)
+
+var (
+	statesTopic = "topic.states.%d"
+)
+
+// durableStateActor is a durable state based actor
+type durableStateActor struct {
+	DurableStateBehavior
+	stateStore      persistence.StateStore
+	currentState    State
+	currentVersion  uint64
+	lastCommandTime time.Time
+	eventsStream    eventstream.Stream
+	actorSystem     actors.ActorSystem
+}
+
+// implements the actors.Actor interface
+var _ actors.Actor = (*durableStateActor)(nil)
+
+// newDurableStateActor creates an instance of actor provided the DurableStateBehavior
+func newDurableStateActor(behavior DurableStateBehavior, stateStore persistence.StateStore, eventsStream eventstream.Stream) *durableStateActor {
+	return &durableStateActor{
+		stateStore:           stateStore,
+		eventsStream:         eventsStream,
+		DurableStateBehavior: behavior,
+	}
+}
+
+// PreStart pre-starts the actor
+func (entity *durableStateActor) PreStart(ctx context.Context) error {
+	return errorschain.
+		New(errorschain.ReturnFirst()).
+		AddError(entity.durableStateRequired()).
+		AddError(entity.stateStore.Ping(ctx)).
+		AddError(entity.recoverFromStore(ctx)).
+		Error()
+}
+
+// Receive processes any message dropped into the actor mailbox.
+func (entity *durableStateActor) Receive(ctx *actors.ReceiveContext) {
+	switch command := ctx.Message().(type) {
+	case *goaktpb.PostStart:
+		entity.actorSystem = ctx.ActorSystem()
+	case *egopb.GetStateCommand:
+		entity.sendStateReply(ctx)
+	default:
+		entity.processCommand(ctx, command)
+	}
+}
+
+// PostStop prepares the actor to gracefully shutdown
+func (entity *durableStateActor) PostStop(ctx context.Context) error {
+	return errorschain.
+		New(errorschain.ReturnFirst()).
+		AddError(entity.stateStore.Ping(ctx)).
+		AddError(entity.persistStateAndPublish(ctx)).
+		Error()
+}
+
+// recoverFromStore reset the persistent actor to the latest state in case there is one
+// this is vital when the entity actor is restarting.
+func (entity *durableStateActor) recoverFromStore(ctx context.Context) error {
+	durableState, err := entity.stateStore.GetLatestState(ctx, entity.ID())
+	if err != nil {
+		return fmt.Errorf("failed unmarshal the latest state: %w", err)
+	}
+
+	if durableState != nil && proto.Equal(durableState, new(egopb.DurableState)) {
+		currentState := entity.InitialState()
+		if err := durableState.GetResultingState().UnmarshalTo(currentState); err != nil {
+			return fmt.Errorf("failed unmarshal the latest state: %w", err)
+		}
+
+		entity.currentState = currentState
+		entity.currentVersion = durableState.GetVersionNumber()
+		return nil
+	}
+
+	entity.currentState = entity.InitialState()
+	return nil
+}
+
+// processCommand processes the incoming command
+func (entity *durableStateActor) processCommand(receiveContext *actors.ReceiveContext, command Command) {
+	ctx := receiveContext.Context()
+	newState, newVersion, err := entity.HandleCommand(ctx, command, entity.currentVersion, entity.currentState)
+	if err != nil {
+		entity.sendErrorReply(receiveContext, err)
+		return
+	}
+
+	// check whether the pre-conditions have met
+	if err := entity.checkPreconditions(newState, newVersion); err != nil {
+		entity.sendErrorReply(receiveContext, err)
+		return
+	}
+
+	// set the current state with the newState
+	entity.currentState = newState
+	entity.lastCommandTime = timestamppb.Now().AsTime()
+	entity.currentVersion = newVersion
+
+	if err := entity.persistStateAndPublish(ctx); err != nil {
+		entity.sendErrorReply(receiveContext, err)
+		return
+	}
+
+	entity.sendStateReply(receiveContext)
+}
+
+// sendStateReply sends a state reply message
+func (entity *durableStateActor) sendStateReply(ctx *actors.ReceiveContext) {
+	state, _ := anypb.New(entity.currentState)
+	ctx.Response(&egopb.CommandReply{
+		Reply: &egopb.CommandReply_StateReply{
+			StateReply: &egopb.StateReply{
+				PersistenceId:  entity.ID(),
+				State:          state,
+				SequenceNumber: entity.currentVersion,
+				Timestamp:      entity.lastCommandTime.Unix(),
+			},
+		},
+	})
+}
+
+// sendErrorReply sends an error as a reply message
+func (entity *durableStateActor) sendErrorReply(ctx *actors.ReceiveContext, err error) {
+	ctx.Response(&egopb.CommandReply{
+		Reply: &egopb.CommandReply_ErrorReply{
+			ErrorReply: &egopb.ErrorReply{
+				Message: err.Error(),
+			},
+		},
+	})
+}
+
+// checkAndSetPreconditions validates the newState and the newVersion
+func (entity *durableStateActor) checkPreconditions(newState State, newVersion uint64) error {
+	currentState := entity.currentState
+	currentStateType := currentState.ProtoReflect().Descriptor().FullName()
+	latestStateType := newState.ProtoReflect().Descriptor().FullName()
+	if currentStateType != latestStateType {
+		return fmt.Errorf("mismatch state types: %s != %s", currentStateType, latestStateType)
+	}
+
+	proceed := int(math.Abs(float64(newVersion-entity.currentVersion))) == 1
+	if !proceed {
+		return fmt.Errorf("%s received version=(%d) while current version is (%d)",
+			entity.ID(),
+			newVersion,
+			entity.currentVersion)
+	}
+	return nil
+}
+
+// checks whether the durable state store is set or not
+func (entity *durableStateActor) durableStateRequired() error {
+	if entity.stateStore == nil {
+		return ErrDurableStateStoreRequired
+	}
+	return nil
+}
+
+// persistState persists the actor state
+func (entity *durableStateActor) persistStateAndPublish(ctx context.Context) error {
+	resultingState, _ := anypb.New(entity.currentState)
+	shardNumber := entity.actorSystem.GetPartition(entity.ID())
+	topic := fmt.Sprintf(statesTopic, shardNumber)
+
+	durableState := &egopb.DurableState{
+		PersistenceId:  entity.ID(),
+		VersionNumber:  entity.currentVersion,
+		ResultingState: resultingState,
+		Timestamp:      entity.lastCommandTime.Unix(),
+		Shard:          shardNumber,
+	}
+
+	eg, ctx := errgroup.WithContext(ctx)
+	eg.Go(func() error {
+		entity.eventsStream.Publish(topic, durableState)
+		return nil
+	})
+
+	eg.Go(func() error {
+		return entity.stateStore.WriteState(ctx, durableState)
+	})
+
+	return eg.Wait()
+}