Repository: looplab/eventhorizon
Pull request #340 (2021-09-27 09:16:14)

Fix / Prevent possible dead lock in memory event store
### Description

Fix to prevent possible dead lock in memory event store where publishing triggers consecutive actions on the same event store (which at that point is still locked).

Also fix a possible ordering error in the recorder when secondary events are saved as a result of the event store save action (sagas etc). Similar cause as the deadlock described above but with different results.

### Affected Components

- Event Store (memory)
- Event Store (recorder)

### Related Issues

### Solution and Design

### Steps to test and verify


Patch:
diff --git a/eventstore/memory/eventstore.go b/eventstore/memory/eventstore.go
index 283ce07..425e928 100644
--- a/eventstore/memory/eventstore.go
+++ b/eventstore/memory/eventstore.go
@@ -60,6 +60,31 @@ func WithEventHandler(h eh.EventHandler) Option {
 
 // Save implements the Save method of the eventhorizon.EventStore interface.
 func (s *EventStore) Save(ctx context.Context, events []eh.Event, originalVersion int) error {
+	if err := s.save(ctx, events, originalVersion); err != nil {
+		return err
+	}
+
+	// Let the optional event handler handle the events. Aborts the transaction
+	// in case of error.
+	if s.eventHandler != nil {
+		for _, e := range events {
+			if err := s.eventHandler.HandleEvent(ctx, e); err != nil {
+				return eh.CouldNotHandleEventError{
+					Err:   err,
+					Event: e,
+				}
+			}
+		}
+	}
+
+	return nil
+}
+
+// This method needs to be separate from the Save() method to not lock the mutex during publishing.
+func (s *EventStore) save(ctx context.Context, events []eh.Event, originalVersion int) error {
+	s.dbMu.Lock()
+	defer s.dbMu.Unlock()
+
 	if len(events) == 0 {
 		return eh.EventStoreError{
 			Err: eh.ErrNoEventsToAppend,
@@ -93,9 +118,6 @@ func (s *EventStore) Save(ctx context.Context, events []eh.Event, originalVersio
 		dbEvents[i] = e
 	}
 
-	s.dbMu.Lock()
-	defer s.dbMu.Unlock()
-
 	// Either insert a new aggregate or append to an existing.
 	if originalVersion == 0 {
 		aggregate := aggregateRecord{
@@ -124,19 +146,6 @@ func (s *EventStore) Save(ctx context.Context, events []eh.Event, originalVersio
 		}
 	}
 
-	// Let the optional event handler handle the events. Aborts the transaction
-	// in case of error.
-	if s.eventHandler != nil {
-		for _, e := range events {
-			if err := s.eventHandler.HandleEvent(ctx, e); err != nil {
-				return eh.CouldNotHandleEventError{
-					Err:   err,
-					Event: e,
-				}
-			}
-		}
-	}
-
 	return nil
 }
 
diff --git a/eventstore/recorder/eventstore.go b/eventstore/recorder/eventstore.go
index 28c1a59..7feefb2 100644
--- a/eventstore/recorder/eventstore.go
+++ b/eventstore/recorder/eventstore.go
@@ -25,8 +25,27 @@ import (
 type EventStore struct {
 	eh.EventStore
 	recording bool
-	record    []eh.Event
-	recordMu  sync.RWMutex
+	records   []*EventRecord
+	mu        sync.RWMutex
+}
+
+// Status is the status of the recorded event.
+type Status int
+
+const (
+	// Pending is for events that are being handled.
+	Pending Status = iota
+	// Succeeded is for events that was successfully saved (with potential side effects).
+	Succeeded
+	// Failed is for events that failed and where not saved.
+	Failed
+)
+
+// EventRecord is a record for an event with additional metadata for status.
+type EventRecord struct {
+	Event  eh.Event
+	Status Status
+	Err    error
 }
 
 // NewEventStore creates a new EventStore.
@@ -37,54 +56,108 @@ func NewEventStore(eventStore eh.EventStore) *EventStore {
 
 	return &EventStore{
 		EventStore: eventStore,
-		record:     make([]eh.Event, 0),
+		records:    make([]*EventRecord, 0),
 	}
 }
 
 // Save implements the Save method of the eventhorizon.EventStore interface.
 func (s *EventStore) Save(ctx context.Context, events []eh.Event, originalVersion int) error {
+	s.mu.RLock()
+	if !s.recording {
+		s.mu.RUnlock()
+		return s.EventStore.Save(ctx, events, originalVersion)
+	}
+	s.mu.RUnlock()
+
+	// Record before saving to keep the order of any potential
+	// secondary saved events from the underlying stores event handler(s).
+	s.mu.Lock()
+	records := make([]*EventRecord, len(events))
+	for i, e := range events {
+		records[i] = &EventRecord{Event: e, Status: Pending}
+	}
+	s.records = append(s.records, records...)
+	s.mu.Unlock()
+
 	if err := s.EventStore.Save(ctx, events, originalVersion); err != nil {
+		// Mark events as failed.
+		s.mu.Lock()
+		for _, r := range records {
+			r.Status = Failed
+			r.Err = err
+		}
+		s.mu.Unlock()
+
 		return err
 	}
 
-	// Only record events that are successfully saved.
-	s.recordMu.Lock()
-	defer s.recordMu.Unlock()
-	if s.recording {
-		s.record = append(s.record, events...)
+	// Mark events as succeeded.
+	s.mu.Lock()
+	for _, r := range records {
+		r.Status = Succeeded
 	}
+	s.mu.Unlock()
 
 	return nil
 }
 
 // StartRecording starts recording of handled events.
 func (s *EventStore) StartRecording() {
-	s.recordMu.Lock()
-	defer s.recordMu.Unlock()
+	s.mu.Lock()
+	defer s.mu.Unlock()
 
 	s.recording = true
 }
 
 // StopRecording stops recording of handled events.
 func (s *EventStore) StopRecording() {
-	s.recordMu.Lock()
-	defer s.recordMu.Unlock()
+	s.mu.Lock()
+	defer s.mu.Unlock()
 
 	s.recording = false
 }
 
-// GetRecord returns the events that happened during the recording.
-func (s *EventStore) GetRecord() []eh.Event {
-	s.recordMu.RLock()
-	defer s.recordMu.RUnlock()
+// FullRecording returns all events that happened during the recording, including status.
+func (s *EventStore) FullRecording() []*EventRecord {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	return s.records
+}
+
+// PendingEvents returns the events that where pending during the recording.
+func (s *EventStore) PendingEvents() []eh.Event {
+	return s.eventsByStatus(Pending)
+}
+
+// SuccessfulEvents returns the events that succeeded during the recording.
+func (s *EventStore) SuccessfulEvents() []eh.Event {
+	return s.eventsByStatus(Succeeded)
+}
+
+// FailedEvents returns the events that failed during the recording.
+func (s *EventStore) FailedEvents() []eh.Event {
+	return s.eventsByStatus(Failed)
+}
+
+func (s *EventStore) eventsByStatus(status Status) []eh.Event {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	var events []eh.Event
+	for _, r := range s.records {
+		if r.Status == status {
+			events = append(events, r.Event)
+		}
+	}
 
-	return s.record
+	return events
 }
 
 // ResetTrace resets the record.
 func (s *EventStore) ResetTrace() {
-	s.recordMu.Lock()
-	defer s.recordMu.Unlock()
+	s.mu.Lock()
+	defer s.mu.Unlock()
 
-	s.record = make([]eh.Event, 0)
+	s.records = make([]*EventRecord, 0)
 }


Test patch:
diff --git a/eventstore/recorder/eventstore_test.go b/eventstore/recorder/eventstore_test.go
index 10cc7b2..a860511 100644
--- a/eventstore/recorder/eventstore_test.go
+++ b/eventstore/recorder/eventstore_test.go
@@ -42,7 +42,7 @@ func TestEventStore(t *testing.T) {
 	savedEvents := eventstore.AcceptanceTest(t, store, context.Background())
 	store.StopRecording()
 
-	record := store.GetRecord()
+	record := store.SuccessfulEvents()
 	if !reflect.DeepEqual(record, savedEvents) {
 		t.Error("there should be events recorded:", record)
 	}
@@ -50,7 +50,7 @@ func TestEventStore(t *testing.T) {
 	// And then some more recording specific testing.
 
 	store.ResetTrace()
-	record = store.GetRecord()
+	record = store.SuccessfulEvents()
 	if len(record) != 0 {
 		t.Error("there should be no events recorded:", record)
 	}
@@ -74,7 +74,7 @@ func TestEventStore(t *testing.T) {
 		t.Error("there should be no error:", err)
 	}
 	aggregate1events = append(aggregate1events, event1)
-	record = store.GetRecord()
+	record = store.SuccessfulEvents()
 	if len(record) != 0 {
 		t.Error("there should be no events recorded:", record)
 	}