cqrs: try to rewind to snapshot when first events is not a snapshot

Author: jkralik

Makefile

@@ -98,7 +98,7 @@ test: env
         -e TEST_ROOT_CA_KEY=/certs/root_ca.key \
 		-e ACME_DB_DIR=/home/certificate-authority \
 		cloud-test \
-		go test -race -p 1 -v ./... -covermode=atomic -coverprofile=/home/coverage.txt
+		go test -timeout=45m -race -p 1 -v ./... -covermode=atomic -coverprofile=/home/coverage.txt
 
 build: cloud-build $(SUBDIRS)
 

resource-aggregate/cqrs/aggregate/aggregateParralel_test.go

@@ -0,0 +1,136 @@
+package aggregate_test
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/kelseyhightower/envconfig"
+	"github.com/plgd-dev/cloud/pkg/net/grpc"
+	"github.com/plgd-dev/cloud/resource-aggregate/cqrs/aggregate"
+	"github.com/plgd-dev/cloud/resource-aggregate/cqrs/events"
+	"github.com/plgd-dev/cloud/resource-aggregate/cqrs/eventstore"
+	"github.com/plgd-dev/cloud/resource-aggregate/cqrs/eventstore/mongodb"
+	"github.com/plgd-dev/cloud/resource-aggregate/cqrs/utils"
+	"github.com/plgd-dev/cloud/resource-aggregate/pb"
+	"github.com/plgd-dev/kit/security/certManager"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/atomic"
+)
+
+func testNewEventstore(t *testing.T) *mongodb.EventStore {
+	var config certManager.Config
+	err := envconfig.Process("DIAL", &config)
+	assert.NoError(t, err)
+
+	dialCertManager, err := certManager.NewCertManager(config)
+	require.NoError(t, err)
+
+	tlsConfig := dialCertManager.GetClientTLSConfig()
+
+	store, err := mongodb.NewEventStore(
+		mongodb.Config{
+			URI: "mongodb://localhost:27017",
+		},
+		func(f func()) error { go f(); return nil },
+		mongodb.WithTLS(tlsConfig),
+	)
+	require.NoError(t, err)
+	require.NotNil(t, store)
+
+	return store
+}
+
+func cleanUpToSnapshot(ctx context.Context, t *testing.T, store *mongodb.EventStore, evs []eventstore.Event) {
+	for _, event := range evs {
+		if ru, ok := event.(*events.ResourceStateSnapshotTaken); ok {
+			if err := store.RemoveUpToVersion(ctx, []eventstore.VersionQuery{{GroupID: ru.GroupId(), AggregateID: ru.AggregateId(), Version: ru.Version()}}); err != nil {
+				require.NoError(t, err)
+			}
+			fmt.Printf("snapshot at version %v\n", event.Version())
+			break
+		}
+	}
+}
+
+func Test_parallelRequest(t *testing.T) {
+	store := testNewEventstore(t)
+	ctx := context.Background()
+	ctx = grpc.CtxWithIncomingUserID(ctx, "test")
+	defer store.Close(ctx)
+	defer func() {
+		err := store.Clear(ctx)
+		require.NoError(t, err)
+	}()
+
+	deviceID := "7397398d-3ae8-4d9a-62d6-511f7b736a60"
+	href := "/test/resource/1"
+	commandPub := pb.PublishResourceRequest{
+		ResourceId: &pb.ResourceId{
+			DeviceId: deviceID,
+			Href:     href,
+		},
+		Resource: &pb.Resource{
+			Id:       utils.MakeResourceId(deviceID, href),
+			DeviceId: deviceID,
+		},
+		CommandMetadata:      &pb.CommandMetadata{},
+		AuthorizationContext: &pb.AuthorizationContext{},
+	}
+
+	newAggragate := func(deviceID, href string) *aggregate.Aggregate {
+		a, err := aggregate.NewAggregate(deviceID, utils.MakeResourceId(deviceID, href), aggregate.NewDefaultRetryFunc(32), 16, store, func(context.Context) (aggregate.AggregateModel, error) {
+			ev := events.NewResourceStateSnapshotTaken()
+			ev.Id = utils.MakeResourceId(deviceID, href)
+			return ev, nil
+		}, nil)
+		require.NoError(t, err)
+		return a
+	}
+
+	concurrencyExcepTestA := newAggragate(commandPub.GetResourceId().GetDeviceId(), commandPub.GetResourceId().GetHref())
+	_, err := concurrencyExcepTestA.HandleCommand(ctx, &commandPub)
+	require.NoError(t, err)
+
+	numParallel := 3
+	var wg sync.WaitGroup
+	var anyError atomic.Bool
+	for i := 0; i < numParallel; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for j := 0; j < 100000; j++ {
+				if anyError.Load() {
+					return
+				}
+				commandContentChanged := pb.NotifyResourceChangedRequest{
+					ResourceId: &pb.ResourceId{
+						DeviceId: deviceID,
+						Href:     href,
+					},
+					Content: &pb.Content{
+						Data:        []byte("hello world"),
+						ContentType: "text",
+					},
+					CommandMetadata: &pb.CommandMetadata{
+						ConnectionId: uuid.New().String(),
+					},
+					Status:               pb.Status_OK,
+					AuthorizationContext: &pb.AuthorizationContext{},
+				}
+				aggr := newAggragate(commandPub.GetResourceId().GetDeviceId(), commandPub.GetResourceId().GetHref())
+				events, err := aggr.HandleCommand(ctx, &commandContentChanged)
+				if err != nil {
+					anyError.Store(true)
+					require.NoError(t, err)
+					return
+				}
+				cleanUpToSnapshot(ctx, t, store, events)
+			}
+		}()
+	}
+	wg.Wait()
+}

resource-aggregate/cqrs/eventstore/mongodb/eventstore.go

@@ -316,6 +316,7 @@ type iterator struct {
 	dataUnmarshaler UnmarshalerFunc
 	LogDebugfFunc   LogDebugfFunc
 	groupID         string
+	loaded          bool
 }
 
 func (i *iterator) Next(ctx context.Context) (eventstore.EventUnmarshaler, bool) {
@@ -324,6 +325,7 @@ func (i *iterator) Next(ctx context.Context) (eventstore.EventUnmarshaler, bool)
 	if !i.iter.Next(ctx) {
 		return nil, false
 	}
+	i.loaded = true
 
 	err := i.iter.Decode(&event)
 	if err != nil {
@@ -345,7 +347,11 @@ func (i *iterator) Next(ctx context.Context) (eventstore.EventUnmarshaler, bool)
 }
 
 func (i *iterator) Err() error {
-	return i.iter.Err()
+	err := i.iter.Err()
+	if err == nil && !i.loaded {
+		err = fmt.Errorf("none event was loaded")
+	}
+	return err
 }
 
 func versionQueriesToMgoQuery(queries []eventstore.VersionQuery, op signOperator) (bson.M, error) {
@@ -417,7 +423,9 @@ func (l *loader) QueryHandlePool(ctx context.Context, iter *queryIterator) error
 	var errors []error
 	var errorsLock sync.Mutex
 
+	var loaded bool
 	for iter.Next(ctx, &query) {
+		loaded = true
 		queries = append(queries, query)
 		if len(queries) >= l.store.batchSize {
 			wg.Add(1)
@@ -449,6 +457,9 @@ func (l *loader) QueryHandlePool(ctx context.Context, iter *queryIterator) error
 	if len(errors) > 0 {
 		return fmt.Errorf("loader cannot load events: %v", errors)
 	}
+	if !loaded {
+		return fmt.Errorf("none snapshot event was loaded")
+	}
 
 	if iter.Err() != nil {
 		return iter.Err()

resource-aggregate/cqrs/eventstore/projection.go

@@ -97,15 +97,29 @@ type iterator struct {
 	reload             *VersionQuery
 }
 
-func (i *iterator) Rewind(ctx context.Context) {
+func (i *iterator) RewindToNextAggregateEvent(ctx context.Context) EventUnmarshaler {
+	for {
+		snapshot, nextAggregateEvent := i.RewindToSnapshot(ctx)
+		if nextAggregateEvent != nil {
+			return nextAggregateEvent
+		}
+		if snapshot == nil && nextAggregateEvent == nil {
+			return nil
+		}
+	}
+}
+
+func (i *iterator) RewindToSnapshot(ctx context.Context) (snapshot EventUnmarshaler, nextAggregateEvent EventUnmarshaler) {
 	for {
 		e, ok := i.iter.Next(ctx)
 		if !ok {
-			break
+			return nil, nil
+		}
+		if e.EventType() == i.model.SnapshotEventType() && e.GroupID() == i.model.groupId && e.AggregateID() == i.model.aggregateId {
+			return e, nil
 		}
 		if e.GroupID() != i.model.groupId || e.AggregateID() != i.model.aggregateId {
-			i.nextEventToProcess = e
-			return
+			return nil, e
 		}
 	}
 }
@@ -135,9 +149,19 @@ func (i *iterator) Next(ctx context.Context) (EventUnmarshaler, bool) {
 		ignore, reload := i.model.Update(tmp)
 		i.model.LogDebugfFunc("projection.iterator.next: GroupId %v: AggregateId %v: Version %v, EvenType %v, ignore %v reload %v", tmp.GroupID, tmp.AggregateID, tmp.Version, tmp.EventType, ignore, reload)
 		if reload {
-			i.reload = &VersionQuery{GroupID: tmp.GroupID(), AggregateID: tmp.AggregateID(), Version: i.model.version}
-			i.Rewind(ctx)
-			return nil, false
+			snapshot, nextAggregateEvent := i.RewindToSnapshot(ctx)
+			if snapshot == nil {
+				i.nextEventToProcess = nextAggregateEvent
+				i.reload = &VersionQuery{GroupID: tmp.GroupID(), AggregateID: tmp.AggregateID(), Version: i.model.version}
+				return nil, false
+			}
+			tmp = snapshot
+			ignore, reload = i.model.Update(tmp)
+			if reload {
+				i.nextEventToProcess = i.RewindToNextAggregateEvent(ctx)
+				i.reload = &VersionQuery{GroupID: tmp.GroupID(), AggregateID: tmp.AggregateID(), Version: i.model.version}
+				return nil, false
+			}
 		}
 		if ignore {
 			return i.RewindIgnore(ctx)
@@ -208,8 +232,8 @@ func (p *Projection) handle(ctx context.Context, iter Iter) (reloadQueries []Ver
 		if i.nextEventToProcess == nil {
 			_, ok := i.Next(ctx)
 			if ok {
-				//iterator need to mode to next
-				i.Rewind(ctx)
+				//iterator need to move to the next event
+				i.nextEventToProcess = i.RewindToNextAggregateEvent(ctx)
 			}
 		}